# agent-mesh

Synchronous agent chains stall when one specialist fails, so `agent-mesh` coordinates role-specific workers through an event contract with retries, dead letters, causal traces, and partial-result fan-in.

## Run the offline proof

From a clone of this repository, the deterministic example needs no credentials
or network access:

```bash
python -m pip install -e ".[dev]"
python examples/research_brief.py
```

It runs two jobs: one completes with researcher and analyst results, and one
finishes as `degraded` with the research result retained after a simulated
permanent analyst failure.

## What the tests prove

```bash
python -m pytest -q
ruff check .
```

The full local suite produced `74 passed, 4 skipped, 1 warning in 34.02s`; Ruff
reported `All checks passed!`. The passing tests exercise fan-out and exclusive
consumer-group delivery, acknowledgements, retries, dead-letter exhaustion,
visibility-timeout reclaim in memory, correlation/causation history, normal and
degraded fan-in, HTTP endpoints, the common in-memory/Redis bus contract using
`fakeredis`, and Redis-backed fan-in recovery across reconstructed store and
aggregator objects.

They also cover the model-driven planner: a nacked and redelivered job re-plans
to byte-identical task ids instead of doubling the fan-out, a role with no
worker is refused where the error is legible rather than dead-lettering later,
and a plan that fails to parse dead-letters the job instead of running it.

## Not implemented / not proven

- The default suite uses `FakeProvider`; four live OpenAI/xAI tests were skipped,
  so the result above does not prove current provider credentials, model IDs,
  latency, or availability.
- Redis behavior is tested with `fakeredis`, not a real Redis server under
  network partitions, process crashes, or production load.
- Delivery is intentionally at-least-once, not exactly-once; consumers must keep
  their own side effects idempotent.
- There is no published throughput, latency, or scale benchmark.
- The MCP adapter exists, but the default suite has no end-to-end MCP transport
  test.

## Why It Exists

Most multi-agent demos are synchronous chains: one call waits for the next call,
and one slow or failing agent stalls or voids the whole run. They often have no
retry boundary, no isolation between specialist roles, and no reliable way to
reconstruct what happened after a bad run.

agent-mesh is an event-driven backend for coordinating specialist agents through
a small, explicit event contract. Work is submitted as a job, decomposed into
role-specific tasks, processed by independent workers, and joined by an
aggregator that can complete with partial results when a specialist fails
permanently.

The default provider is fake, so the full test suite and demo run with no API
keys.

## Architecture

```text
submit
  |
  v
JOB_SUBMITTED
  |
  v
supervisor
  |
  v
TASK_REQUESTED fan-out
  |                         |
  v                         v
researcher worker       analyst worker
  |                         |
  v                         v
TASK_SUCCEEDED/FAILED   TASK_SUCCEEDED/FAILED
  \                         /
   \                       /
    v                     v
        aggregator fan-in
              |
              v
      JOB_COMPLETED or JOB_FAILED

TASK_REQUESTED retry budget exhausted
              |
              v
        DEAD_LETTER
              |
              v
      dead-letter sweeper
              |
              v
         TASK_FAILED
```

## Event Flow

`Supervisor.submit(goal)` publishes `JOB_SUBMITTED` with the job id as the
correlation id. The supervisor consumes that event, stores the job state, and
publishes one `TASK_REQUESTED` event per role. Each worker is in its own consumer
group, so every role receives a copy of the task stream while work remains
exclusive within that role. Workers publish `TASK_SUCCEEDED` or `TASK_FAILED`.
The aggregator consumes both, joins by job id, and publishes `JOB_COMPLETED` or
`JOB_FAILED` once every expected role has settled.

Every child event preserves the same correlation id and records the causation id
of the event that produced it. `GET /jobs/{job_id}/events` returns the causal
trace for debugging.

## Planning the fan-out with a model

`default_decomposer` is a fixed three-role split. `LLMDecomposer` lets a model
plan it instead — and the interesting part is not the prompt, it is that
delivery here is at-least-once.

`Supervisor.run_once` decomposes on every delivery of `JOB_SUBMITTED`, so the
same goal gets planned more than once in normal operation: after a nack, after a
crash between publish and ack, after a consumer-group rebalance. `TaskSpec`
generates a random `task_id` by default, so even the built-in decomposer
republishes tasks under fresh identities on redelivery. Point that path at a
model whose wording varies between calls and one job becomes two different
fan-outs, billed twice, against an `expected` map the workers no longer match.

So the planner makes the plan stable rather than trusting the model to be:

- **Task identity is derived, not generated** — `task_id` is a digest of the goal
  and the role. A test re-plans the same goal with a model that rewords every
  instruction and asserts the ids are identical.
- **Roles come from an allow-list** — a model naming a specialist nobody runs
  produces work that can only dead-letter, so it is refused at planning time.
- **Fan-out is capped**, and a duplicate role is refused because `expected` is
  keyed by role and the twin would settle the job early.
- **It fails closed** — unparseable output raises, the delivery is nacked, and
  the job dead-letters where an operator can see it.

```python
from agent_mesh.planner import LLMDecomposer

supervisor = Supervisor(
    bus, store, decomposer=LLMDecomposer(model, roles=("researcher", "analyst"))
)
```

## Reliability Semantics

The bus is a `Protocol` with two implementations. `InMemoryBus` is the default
used by the examples and the ordinary test suite. `RedisStreamsBus` provides the
same contract on Redis Streams for real deployments and is available with:

```bash
pip install -e .[redis]
```

`tests/test_bus_contract.py` runs one conformance suite against both
implementations, using `fakeredis` for the Redis-backed case, so the abstraction
is verified rather than assumed.

Fan-in state has the same split. `JobStore` is the in-memory reference and
`RedisJobStore` persists expected roles, partial results, failures, and the final
status. Redis updates use optimistic transactions, so only the process that
atomically changes `running` to a terminal state emits the completion event.
Restarting an aggregator between two role results does not forget the first.

| Guarantee | Mechanism |
| --- | --- |
| Fan-out by role | One topic copied to each subscribed consumer group |
| Exclusive work within a role | A pending set per topic and group |
| At-least-once delivery | Messages remain pending until acked |
| Crash recovery | Visibility-timeout reclaim returns unacked work to the queue |
| Idempotency | Bus-level seen set skips already-acked event ids |
| Retry budget | `nack` redelivers until `max_deliveries` is reached |
| Poison-message isolation | Exhausted messages move to `DEAD_LETTER` |
| Joins do not hang | Dead-letter sweeper converts poison tasks into `TASK_FAILED` |
| Partial failure | Aggregator marks jobs `degraded` when some roles fail permanently |
| Durable fan-in | Redis optimistic updates retain partial joins across restarts |
| Traceability | Correlation and causation ids on every event |

## Providers

`FakeProvider` is deterministic and needs no credentials. It can cycle through
fixed responses, fail transiently for a configured number of calls, or fail
permanently for degradation-path tests.

SDK-backed providers read credentials from the process environment, or from a
gitignored `.env` file loaded with `agent_mesh.config.load_dotenv()`. Values
already present in the real environment always win over `.env`, so
platform-injected secrets are not overridden by a stale file baked into an
image. `describe_credentials()` reports whether credentials are present without
ever revealing a value.

`OpenAIProvider` uses the official OpenAI SDK and reads `OPENAI_API_KEY` when an
API key is not passed directly. `OpenAIProvider.DEFAULT_MODEL` is
`gpt-4o-mini`, overridable with `OPENAI_MODEL` or a `model=` argument.

`GrokProvider` uses the same SDK against xAI's OpenAI-compatible endpoint,
`https://api.x.ai/v1`, so both vendors share one client path. It reads
`XAI_API_KEY`, falling back to `GROK_API_KEY`. `GrokProvider.DEFAULT_MODEL` is
`grok-4.5`, overridable with `XAI_MODEL` or a `model=` argument.

xAI retires model ids aggressively and does not keep bare-major aliases alive.
A plausible-looking id such as `grok-3` returns `404` at the first live call
rather than failing at import. The live test suite therefore asserts that each
provider default model id appears in that provider's `GET /v1/models` listing.

Real keys must never be committed. `.env` is gitignored.

## MCP Integration

`agent_mesh.mcp_server.create_mcp_server(supervisor, store, mesh)` exposes three
MCP tools:

| Tool | Result |
| --- | --- |
| `submit_job(goal)` | Submits a goal, drains the mesh, and returns the job id |
| `get_job(job_id)` | Returns job status and output |
| `list_jobs()` | Returns known jobs |

Install MCP support with:

```bash
pip install 'agent-mesh[mcp]'
```

## Deployment quickstart

```bash
pip install -e .[dev]
python examples/research_brief.py
```

Run the deployable Redis-backed API and workers (fake providers by default):

```bash
docker compose up --build
curl -X POST http://localhost:8000/jobs \
  -H "content-type: application/json" \
  -d '{"goal":"Compare two implementation options"}'
```

Set `AGENT_MESH_PROVIDER_MODE=mixed` with `OPENAI_API_KEY` and `XAI_API_KEY`
to run Grok as researcher and OpenAI as analyst. Redis uses append-only
persistence and a named Compose volume.

To use SDK-backed providers:

```bash
pip install -e .[llm]
export OPENAI_API_KEY=...
export XAI_API_KEY=...
```

## Running against live models

`examples/live_brief.py` runs the full mesh with Grok serving the researcher
role and GPT serving the analyst role in the same job, which is the point of
the provider abstraction.

```bash
python examples/live_brief.py
```

An optional goal can be passed as command line arguments. For the no-credentials
equivalent, run `examples/research_brief.py`, which uses the deterministic fake
provider.

## HTTP API

Create an app with `agent_mesh.api.create_app(bus, store, supervisor, mesh)`.

| Method | Path | Description |
| --- | --- | --- |
| `GET` | `/healthz` | Health check |
| `POST` | `/jobs` | Submit `{ "goal": "..." }`; returns `202` and `job_id` |
| `GET` | `/jobs` | List all jobs |
| `GET` | `/jobs/{job_id}` | Get status, output, results, and failures |
| `GET` | `/jobs/{job_id}/events` | Get the full causal event trace |
| `GET` | `/dead-letters` | Inspect dead-lettered messages |

## Testing

```bash
ruff check .
pytest -q
```

The default suite uses `FakeProvider`, so it runs with no credentials. See the
measured result and its limits at the top of this README.

Live integration tests are opt-in and skipped otherwise:

```bash
AGENT_MESH_LIVE=1 pytest tests/test_live_providers.py
```

The live tier can catch failures the fake provider structurally cannot, including
a retired model id, a changed base URL, or a moved auth scheme.

## Design Notes

The in-memory bus and job store are reference implementations. The supplied
server and Compose stack use Redis Streams plus `RedisJobStore`; transport and
fan-in state therefore survive process replacement instead of mixing a durable
queue with an in-memory join that forgets half-completed jobs.

The worker treats permanent provider failures as settled task failures and lets
transient failures propagate to the bus retry path. That split keeps policy out
of provider adapters and prevents a single bad request from blocking a fan-in.