Multi-Agent Orchestration Platform

A deterministic workflow engine coordinating specialized AI agents (Supervisor → Researcher → Coder → Reviewer) via a LangGraph state machine, with Postgres checkpointing after every node transition and a reliable Redis task queue feeding horizontally scaled workers.

FastAPI ──▶ Redis queue ──▶ worker(s) ──▶ LangGraph
   (auth,      (reserve/ack,               │
   validate,    reclaim on           supervisor ──(route)──▶ researcher ─▶ coder ─▶ reviewer
   rate limit)  crash)                                                 ▲          │
                                                                       └─ reject ─┘
                                                        approve │ budget spent │ human gate
                                                                        ▼
                                                                       END
             every super-step checkpointed to Postgres (pooled AsyncPostgresSaver)

Infinite loops are impossible by construction: the reviewer's reject edge is bounded by max_revisions, and recursion_limit (25) is the hard backstop that raises GraphRecursionError instead of burning tokens. Routing and review verdicts use structured output (typed enums), so odd model output can't misroute or crash the graph, and untrusted task text is fenced so it can't hijack the review gate.

Run it

echo "OPENAI_API_KEY=sk-..." > .env
docker compose up -d --build          # postgres, redis, api (:8000), worker, retention
docker compose up -d --scale worker=4 # scale workers horizontally

Or locally without containers for the app:

docker compose up -d postgres redis   # deps only
uv sync
uv run uvicorn orchestrator.api:app
uv run python -m orchestrator.worker

API

Tenant comes from the X-API-Key header (or X-Tenant-Id in key-less dev mode) — never from the path or body, so no caller can touch another tenant's tasks. IDs are charset-validated ([A-Za-z0-9_-]{1,64}).

H='-H content-type:application/json -H X-Tenant-Id:acme'
curl -X POST localhost:8000/tasks $H -d '{"task_id":"t1","task":"write fizzbuzz"}'
curl localhost:8000/tasks/t1 -H X-Tenant-Id:acme        # lifecycle status + curated state
curl localhost:8000/tasks -H X-Tenant-Id:acme           # list this tenant's tasks
curl localhost:8000/tasks/t1/checkpoints -H X-Tenant-Id:acme   # time-travel history
curl -N localhost:8000/tasks/t1/events -H X-Tenant-Id:acme     # SSE progress stream
curl -X DELETE localhost:8000/tasks/t1 -H X-Tenant-Id:acme     # cancel
curl -X POST localhost:8000/tasks/t1/approve -H X-Tenant-Id:acme  # resume a HITL gate

GET /healthz, GET /readyz (pings Redis + Postgres), and GET /stats (queue + dead-letter depth) support probes and monitoring.

Submit with "require_approval": true to park the graph before the reviewer for human sign-off; "callback_url" to get a POST on completion; "token_budget" and "max_revisions" to cap cost per task.

Tests

uv run pytest        # integration tests skip if Postgres/Redis are down; CI makes skips fatal

29 tests: routing (typed, total), reject→revise halting, recursion_limit backstop, schema validation, structured-output parsing and content-block normalization, the full API contract (auth/validation/status/404/duplicate), worker three-state dispatch (fresh / resume / completed-skip) and cancellation, crash-resume through real Postgres, reliable-queue reserve/ack/reclaim, and the retention purge round-trip.

Reliability

• No lost tasks. reserve() moves a task to a per-worker processing list; it's removed only on a terminal outcome. A worker killed mid-task leaves the entry to be reclaimed on restart. Enqueue is Lua-atomic (no SETNX/LPUSH gap).
• Idempotent re-delivery. A completed thread re-delivered is a no-op; a crashed one resumes from its last checkpoint without re-running done nodes.
• Bounded failures. Tasks retry up to max_attempts, then dead-letter; the dedupe key is cleared on any terminal outcome so re-submit works.
• Graceful shutdown. SIGTERM drains the in-flight task before exit.
• Reconnecting pool. The checkpointer runs over a health-checked psycopg pool, with advisory-locked migration and startup retry for DB outages.
• Concurrency-safe. A per-thread lock stops two workers running the same thread at once.

Schema migrations: every AgentState field except task keeps a default, so checkpoints written before a field existed still deserialize on resume.

Deployment readiness

• Async writes — pooled AsyncPostgresSaver; checkpoint saves never block the event loop.
• Connection pooling — psycopg AsyncConnectionPool per process (prepare_threshold=0, PgBouncer-session-mode compatible); size via POOL_SIZE.
• Multi-tenancy — tenant derived from the API key server-side; reads and writes are scoped to it; IDs validated so : can't forge a thread.
• State retention — the retention compose service runs scripts/purge_checkpoints.sql daily (version-aware blob pruning included).
• Container + CI — Dockerfile (one image, api/worker entrypoints); GitHub Actions runs the suite against service containers and fails on skips.
• Redis durability — AOF enabled with a volume; queued tasks survive restart.
• PgBouncer — in production, point DATABASE_URL at PgBouncer in session-pooling mode, sized for workers × POOL_SIZE connections.
• Managed Postgres & TLS — RDS or equivalent; require sslmode=require, rediss://, and HTTPS at the edge. Workers are stateless — scale freely.

What's deliberately deferred

See ROADMAP.md. Not built (YAGNI until demand): checkpoint replay/fork endpoint (read-only time-travel is shipped; replay needs per-node idempotency guarantees first), parallel fan-out (Send API), declarative graph DSL, object-storage artifacts, priority queues, and LangSmith tracing (enable via env when needed — zero code).