agenticraft_foundation.algebra.patterns¶

6 coordination patterns built from CSP primitives: request-response, pipeline, scatter-gather, barrier, mutex, producer-consumer.

Coordination patterns for multi-agent systems in CSP.

This module provides: - Request-Response: Client-server interaction - Pipeline: Sequential processing stages - Scatter-Gather: Parallel task distribution and collection - Barrier: Synchronization point for multiple agents - Mutex: Mutual exclusion pattern - Producer-Consumer: Buffered communication

`BarrierPattern` `dataclass` ¶

Bases: CoordinationPattern

Barrier pattern for multi-party synchronization.

All participants must reach the barrier before any can proceed.

CSP Model

PARTICIPANT_i = arrive → depart → SKIP BARRIER = arrive_1 → arrive_2 → ... → depart_1 → depart_2 → ... → SKIP

`MutexPattern` `dataclass` ¶

Bases: CoordinationPattern

Mutual exclusion pattern.

Only one process can be in the critical section at a time.

CSP Model

MUTEX = acquire → release → MUTEX PROCESS_i = acquire → critical_i → release → SKIP

`global_process()` ¶

The mutex resource process.

`system_process()` ¶

The full system with mutex and all processes.

`PipelinePattern` `dataclass` ¶

Bases: CoordinationPattern

Pipeline pattern for sequential multi-stage processing.

Interaction

Stage1 ─data→ Stage2 ─data→ Stage3 → ...

CSP Model

STAGE_1 = data_1_2 → SKIP STAGE_2 = data_1_2 → data_2_3 → SKIP STAGE_N = data_{n-1}_n → SKIP SYSTEM = STAGE_1 ||| STAGE_2 ||| ... ||| STAGE_N (synchronized on data events)

`ProducerConsumerPattern` `dataclass` ¶

Bases: CoordinationPattern

Producer-Consumer pattern with bounded buffer.

CSP Model

PRODUCER = produce → put → PRODUCER CONSUMER = get → consume → CONSUMER BUFFER = put → get → BUFFER (simplified unbounded)

`global_process()` ¶

Simplified: produce → put → get → consume → recurse.

`RequestResponsePattern` `dataclass` ¶

Bases: CoordinationPattern

Request-Response pattern for client-server interaction.

Interaction

Client ─request→ Server Server ─response→ Client

CSP Model

CLIENT = request → response → SKIP SERVER = request → response → SKIP SYSTEM = CLIENT |[{request, response}]| SERVER

`ScatterGatherPattern` `dataclass` ¶

Bases: CoordinationPattern

Scatter-Gather pattern for parallel task distribution.

Interaction

Coordinator ─task→ Worker1, Worker2, ... Worker1, Worker2, ... ─result→ Coordinator

CSP Model

COORD = task_w1 → task_w2 → result_w1 → result_w2 → SKIP WORKER_i = task_wi → result_wi → SKIP

`barrier(participants, arrive_event='arrive', depart_event='depart', repeatable=False)` ¶

Create a barrier synchronization pattern.

PARAMETER	DESCRIPTION
`participants`	List of participant names TYPE: `Sequence[str]`
`arrive_event`	Prefix for arrival events TYPE: `str` DEFAULT: `'arrive'`
`depart_event`	Prefix for departure events TYPE: `str` DEFAULT: `'depart'`
`repeatable`	Whether the pattern repeats TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Process`	The global process representing the pattern

`compose_agents(agents, sync_events=None)` ¶

Compose multiple agent processes in parallel.

PARAMETER	DESCRIPTION
`agents`	Dict mapping agent names to their processes TYPE: `dict[str, Process]`
`sync_events`	Events to synchronize on (all if None) TYPE: `set[str] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Process`	Parallel composition of all agents

`mutex(processes, acquire_event='acquire', release_event='release', critical_event_prefix='critical')` ¶

Create a mutual exclusion pattern.

PARAMETER	DESCRIPTION
`processes`	List of process names TYPE: `Sequence[str]`
`acquire_event`	Event to acquire mutex TYPE: `str` DEFAULT: `'acquire'`
`release_event`	Event to release mutex TYPE: `str` DEFAULT: `'release'`
`critical_event_prefix`	Prefix for critical section events TYPE: `str` DEFAULT: `'critical'`

RETURNS	DESCRIPTION
`Process`	The mutex resource process

`pipeline(stages, data_event_prefix='data', repeatable=False)` ¶

Create a pipeline pattern process.

PARAMETER	DESCRIPTION
`stages`	Ordered list of stage names TYPE: `Sequence[str]`
`data_event_prefix`	Prefix for data events TYPE: `str` DEFAULT: `'data'`
`repeatable`	Whether the pattern repeats TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Process`	The global process representing the pattern

`producer_consumer(producer='producer', consumer='consumer', produce_event='produce', consume_event='consume', put_event='put', get_event='get')` ¶

Create a producer-consumer pattern.

PARAMETER	DESCRIPTION
`producer`	Producer name TYPE: `str` DEFAULT: `'producer'`
`consumer`	Consumer name TYPE: `str` DEFAULT: `'consumer'`
`produce_event`	Event when producer creates item TYPE: `str` DEFAULT: `'produce'`
`consume_event`	Event when consumer uses item TYPE: `str` DEFAULT: `'consume'`
`put_event`	Event to put item in buffer TYPE: `str` DEFAULT: `'put'`
`get_event`	Event to get item from buffer TYPE: `str` DEFAULT: `'get'`

RETURNS	DESCRIPTION
`Process`	The global process representing the pattern

`request_response(client='client', server='server', request_event='request', response_event='response', repeatable=False)` ¶

Create a request-response pattern process.

PARAMETER	DESCRIPTION
`client`	Client participant name TYPE: `str` DEFAULT: `'client'`
`server`	Server participant name TYPE: `str` DEFAULT: `'server'`
`request_event`	Request event name TYPE: `str` DEFAULT: `'request'`
`response_event`	Response event name TYPE: `str` DEFAULT: `'response'`
`repeatable`	Whether the pattern repeats TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Process`	The global process representing the pattern

`scatter_gather(coordinator='coordinator', workers=(), task_event_prefix='task', result_event_prefix='result', repeatable=False)` ¶

Create a scatter-gather pattern process.

PARAMETER	DESCRIPTION
`coordinator`	Coordinator name TYPE: `str` DEFAULT: `'coordinator'`
`workers`	List of worker names TYPE: `Sequence[str]` DEFAULT: `()`
`task_event_prefix`	Prefix for task events TYPE: `str` DEFAULT: `'task'`
`result_event_prefix`	Prefix for result events TYPE: `str` DEFAULT: `'result'`
`repeatable`	Whether the pattern repeats TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Process`	The global process representing the pattern

`verify_pattern(pattern, spec=None)` ¶

Verify properties of a coordination pattern.

PARAMETER	DESCRIPTION
`pattern`	The pattern to verify TYPE: `CoordinationPattern`
`spec`	Optional specification to check refinement against TYPE: `Process \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`PatternVerification`	PatternVerification with results