Centaur Mental Model Wiki

Complete Markdown

# Centaur Mental Model Wiki

> Centaur is a self-hosted Kubernetes agent platform that lets teams share one AI agent accessed via Slack or API, running each conversation in an isolated sandbox with durable state, approved tool plugins, and credential-safe egress through iron-proxy.

## Context Links

- [Agent index](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/llms.txt)
- [Human interactive wiki](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2)
- [GitHub repository](https://github.com/paradigmxyz/centaur)

## Repository Metadata

- Repository: paradigmxyz/centaur

- Generated: 2026-05-21T23:46:21.175Z
- Updated: 2026-05-21T23:46:27.274Z
- Runtime: Claude Code
- Format: Mental Model
- Pages: 8

## Page Index

- 01. [How Centaur Works in Your Head](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/01-how-centaur-works-in-your-head.md) - The simplest mental model of Centaur: one shared agent, one sandbox per Slack thread, durable state so nothing is lost on restart, and iron-proxy so agents never touch raw credentials. Understand this page and every other page falls into place.
- 02. [Durability Invariants & Failure Modes](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/02-durability-invariants-failure-modes.md) - What Postgres owns, what lives only in-process, how reconnects and pod replacements are safe, and the exact failure scenarios where state can be lost or replayed incorrectly. The distinction between the event stream (client contract) and in-memory runtime state is the critical boundary.
- 03. [API Lifecycle: spawn → message → execute → events → release](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/03-api-lifecycle-spawn-message-execute-events-release.md) - The five-step durable API lifecycle, how each call maps to a Postgres write, what the SSE event stream guarantees, and what happens when a client reconnects mid-run with after_event_id. Also covers the Slackbot ingress path including HMAC signature verification.
- 04. [Sandbox Pods & the Warm Pool](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/04-sandbox-pods-the-warm-pool.md) - How sandbox Kubernetes pods are created, claimed, and recycled; why the warm pool exists (15-second startup cost); the POOL_EVICT_ON_STARTUP invariant that guarantees new pods run fresh code after a deploy; and the sandbox session state machine (idle → running → delivering → released).
- 05. [Harness Adapters: Amp, Claude Code, Codex, pi-mono](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/05-harness-adapters-amp-claude-code-codex-pi-mono.md) - How the harness_protocol layer normalizes four different agent CLIs into a single event stream. What each adapter does differently (Amp materializes attachments to files; Claude Code passes Anthropic content blocks directly; Codex/pi-mono extract plain text). The _VALID_STDOUT_EVENT_TYPES allowlist as a forward-compatibility boundary.
- 06. [Tool Plugin Model: Discovery, Secrets, & the centaur_sdk](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/06-tool-plugin-model-discovery-secrets-the-centaur_sdk.md) - How tools are discovered (Python files in tools/ or overlays), loaded by tool_manager.py, and exposed to sandboxes. The ToolContext / secret() resolution chain (tool context → pluggable backend → default). The SecretMode enum (replace vs inject). How tool authors import centaur_sdk and never see raw credentials.
- 07. [Durable Workflow Engine: Checkpoint/Replay](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/07-durable-workflow-engine-checkpoint-replay.md) - The checkpoint/replay model inspired by Cloudflare Workflows: the handler function IS the workflow, ctx.step() calls are discovered at runtime, and Postgres checkpoints each step result. On resume after a crash the handler re-executes top-to-bottom but skips already-checkpointed steps instantly. How this enables sleep, suspension, child agents, and idempotent re-runs.
- 08. [Secrets & Egress: iron-proxy as the Trust Boundary](https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/08-secrets-egress-iron-proxy-as-the-trust-boundary.md) - How iron-proxy is the single egress choke point for every sandbox, why it is per-sandbox rather than shared (a compromised pod cannot leak into another), the four secret transform types (replace, inject, gcp_auth, oauth_token, hmac_sign), the NetworkPolicy default-deny invariant, and what changes break the security model (shared proxy, raw key injection, relaxed NetworkPolicy).

## Source File Index

- `AGENTS.md`
- `centaur_sdk/backends/base.py`
- `centaur_sdk/backends/registry.py`
- `centaur_sdk/tests/test_tool_sdk.py`
- `centaur_sdk/tool_sdk.py`
- `docs/pages/architecture.mdx`
- `docs/pages/security.mdx`
- `docs/pages/what-is-centaur.mdx`
- `README.md`
- `services/api/api/agent.py`
- `services/api/api/app.py`
- `services/api/api/iron-proxy.base.yaml`
- `services/api/api/models.py`
- `services/api/api/proxy_config.py`
- `services/api/api/runtime_control.py`
- `services/api/api/sandbox/base.py`
- `services/api/api/sandbox/harness_protocol.py`
- `services/api/api/sandbox/kubernetes.py`
- `services/api/api/sandbox/normalize.py`
- `services/api/api/sandbox/prompt_assembly.py`
- `services/api/api/sandbox/registry.py`
- `services/api/api/slackbot_client.py`
- `services/api/api/tool_manager.py`
- `services/api/api/warm_pool.py`
- `services/api/api/workflow_engine.py`
- `services/api/tests/test_agent_resilience.py`
- `services/api/tests/test_amp_wrapper.py`
- `services/api/tests/test_harness_protocol.py`
- `services/api/tests/test_inflight_replay.py`
- `services/api/tests/test_proxy_config.py`
- `services/api/tests/test_warm_pool.py`
- `services/api/tests/test_workflow_engine_title.py`
- `services/api/tests/test_workflow_idempotency_unit.py`
- `workflows/paradigm_pulse_daily.py`
- `workflows/slack_sync.py`

---

## 01. How Centaur Works in Your Head

> The simplest mental model of Centaur: one shared agent, one sandbox per Slack thread, durable state so nothing is lost on restart, and iron-proxy so agents never touch raw credentials. Understand this page and every other page falls into place.

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/01-how-centaur-works-in-your-head.md
- Generated: 2026-05-21T23:42:47.796Z

### Source Files

- `README.md`
- `docs/pages/architecture.mdx`
- `docs/pages/what-is-centaur.mdx`
- `services/api/api/agent.py`
- `AGENTS.md`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [README.md](README.md)
- [AGENTS.md](AGENTS.md)
- [docs/pages/architecture.mdx](docs/pages/architecture.mdx)
- [docs/pages/what-is-centaur.mdx](docs/pages/what-is-centaur.mdx)
- [services/api/api/agent.py](services/api/api/agent.py)
- [services/api/api/sandbox/base.py](services/api/api/sandbox/base.py)
- [services/api/api/sandbox/harness_protocol.py](services/api/api/sandbox/harness_protocol.py)
- [services/sandbox/entrypoint.sh](services/sandbox/entrypoint.sh)
- [services/iron-proxy/iron-proxy.yaml](services/iron-proxy/iron-proxy.yaml)
</details>

# How Centaur Works in Your Head

Centaur has four invariants. Understand them and every subsystem — sandboxes, durable state, tools, credential injection — falls into a coherent picture. This page builds that picture deliberately: start with the invariants, trace a single Slack message through the full system, then inspect how each invariant holds under failure.

One shared agent for the whole team. One Kubernetes sandbox per Slack thread. Postgres as the single source of truth so nothing is lost on restart. And iron-proxy so agents never see raw credentials.

---

## The Four Invariants

| # | Invariant | What it prevents |
|---|-----------|------------------|
| 1 | **One shared agent** | Every team member talks to the same Centaur Slack bot rather than spinning up one-off local setups |
| 2 | **One sandbox per thread** | Conversations cannot bleed into each other; each thread gets its own Kubernetes pod with its own filesystem, shell, and process tree |
| 3 | **Durable state in Postgres** | A client disconnect, API restart, or pod replacement does not erase a running turn; every step is stored before any response is sent |
| 4 | **iron-proxy for credentials** | Sandbox pods receive only placeholder strings; real API keys are injected on the wire by a per-sandbox proxy, bound to specific upstream hosts |

Sources: [docs/pages/what-is-centaur.mdx:1-44](), [README.md:186-194]()

---

## The Anatomy of One Request

This is what happens when a user types `@centaur why are the billing tests failing?` in Slack.

```text
┌─────────────────────────────────────────────────────────────┐
│  Slack                                                      │
│  @centaur why are the billing tests failing?                │
└────────────────────┬────────────────────────────────────────┘
                     │ HMAC-verified webhook
                     ▼
┌─────────────────────────────────────────────────────────────┐
│  Slackbot  (Next.js + Slack Bolt)                           │
│  Verifies X-Slack-Signature, then calls Centaur API         │
└────────────────────┬────────────────────────────────────────┘
                     │ SLACKBOT_API_KEY
                     ▼
┌─────────────────────────────────────────────────────────────┐
│  Centaur API  (FastAPI + Postgres)                          │
│  POST /agent/spawn   → pin or create a sandbox              │
│  POST /agent/message → write user turn to Postgres          │
│  POST /agent/execute → create execution row, enqueue        │
│  GET  /agent/threads/{thread}/events → SSE stream           │
└────────────────────┬────────────────────────────────────────┘
                     │ kubectl exec attach (stdin/stdout NDJSON)
                     ▼
┌─────────────────────────────────────────────────────────────┐
│  Kubernetes Sandbox Pod  (centaur-agent:latest)             │
│  Harness CLI: amp | claude-code | codex | pi-mono           │
│  Reads workspace/AGENTS.md as system prompt                 │
│  Calls tools via: curl $CENTAUR_API_URL/tools/<tool>/<method│
│  LLM calls: HTTPS → iron-proxy → real API key injected      │
└────────────────────┬────────────────────────────────────────┘
                     │ SSE events back to Slackbot
                     ▼
              Slack thread reply
```

Sources: [AGENTS.md:70-80](), [services/api/api/agent.py:1-13]()

---

## Invariant 1: One Shared Agent

The Slack bot is a thin adapter that owns exactly two responsibilities: verifying incoming Slack requests with `SLACK_SIGNING_SECRET` (HMAC-SHA256), and translating Slack events into the three-step API protocol. It does not hold agent state, model context, or tool registrations.

Every team member's message reaches the same Centaur API. The API owns runtime assignment, execution serialization, cancellation, and delivery recovery. Clients are intentionally kept thin so there is no agent state to synchronize across team members.

Sources: [docs/pages/architecture.mdx:47-60](), [AGENTS.md:85-88]()

---

## Invariant 2: One Sandbox Per Thread

The `thread_key` is the fundamental unit of identity in Centaur. For Slack, it looks like `slack:C0AJ07U8Z1N:1773364194.179929` — channel plus message timestamp. The API maps exactly one active Kubernetes pod to each `thread_key` at a time.

The core logic is in `get_or_spawn()`:

```python
# services/api/api/agent.py:613-750 (condensed)
async def get_or_spawn(thread_key, harness, *, engine, persona) -> SandboxSession:
    """Tries (in order): DB session → warm pool → cold spawn."""
    session = await _db_get_session(thread_key)
    if session and session.db_state in _REUSABLE_DB_STATES:
        if await backend.status(session) == "running":
            return session          # existing pod → return immediately
    # pod is gone: save resume state, then try warm pool or cold spawn
    ...
```

A warm pool (`warm_pool.py`) pre-creates pods so common spawns take milliseconds instead of waiting for Kubernetes scheduling. If the warm pool is empty, the API cold-spawns a new pod. Either way, the `sandbox_sessions` table records the binding.

### What a sandbox pod contains

The entrypoint (`services/sandbox/entrypoint.sh`) runs at pod startup and:
1. Writes harness-specific config files (Amp settings, Codex config, Claude settings)
2. Clones the target repo into `~/workspace/` as a fresh branch
3. Assembles `workspace/AGENTS.md` from the base system prompt plus any org overlay and persona overlay
4. Copies skills into `workspace/.agents/skills/`
5. Touches `~/.ready` to signal readiness

The harness CLI (Amp, Claude Code, Codex, or pi-mono) then reads `workspace/AGENTS.md` as its system instructions and begins waiting for `turn.start` input on stdin.

Sources: [services/api/api/agent.py:613-750](), [services/sandbox/entrypoint.sh:1-208](), [AGENTS.md:436-470]()

---

## Invariant 3: Durable State in Postgres

This is the invariant that lets Centaur survive every common failure: client disconnect, API restart, pod death, workflow worker restart.

### The five-table contract

Every step of a turn writes to Postgres **before** any response is sent to the caller. The tables are:

| Table | What it stores |
|-------|----------------|
| `sandbox_sessions` | Thread-to-pod binding; inflight turn payload; last result cursor |
| `agent_runtime_assignments` | Thread-to-runtime pin and `assignment_generation` |
| `agent_message_requests` | Durable inbound transcript events |
| `agent_execution_requests` | Queued/running/terminal execution row |
| `agent_execution_events` | Replayable raw + projected execution events |
| `agent_final_delivery_outbox` | Final-result delivery obligation for retry paths |
| `chat_messages` | Persisted user/assistant messages for durable transcript surfaces |

Sources: [AGENTS.md:166-176](), [services/api/api/agent.py:268-298]()

### The three-step client protocol

```
POST /agent/spawn   → writes agent_runtime_assignments
POST /agent/message → writes agent_message_requests + chat_messages
POST /agent/execute → writes agent_execution_requests
GET  /agent/threads/{thread}/events → tails agent_execution_events (SSE)
```

Clients reconnect with `after_event_id` instead of restarting work. If the execution already finished and no more rows remain, the API emits the terminal `execution_state` snapshot so late joiners still get the answer.

### Inflight turn durability

Before writing to sandbox stdin, the API persists the full turn payload:

```python
# services/api/api/agent.py:1159-1165
durable_turn_id = f"turn-{uuid.uuid4().hex[:16]}"
await _db_set_inflight_turn(
    session.thread_key,
    durable_turn_id,
    turn_input,
    attempts=1,
)
```

If the pod dies mid-turn, `replay_inflight_turn()` re-sends the saved payload into the replacement pod. The attempt counter increments so operators can detect stuck retries.

### Reconciliation tick

Every 60 seconds, `reconcile_tick()` walks active `sandbox_sessions` rows and:
- Marks sessions `suspended` when the backing pod is gone (Step A)
- Enforces the idle TTL (`IDLE_TTL_S`, default 24 hours) by stopping pods that have been quiet (Step B)
- Reaps `running` rows with no live wire, turn, or execution (Step C)
- Reaps stuck inflight turns with no driving execution (Step D)

Sources: [services/api/api/agent.py:1516-1736]()

---

## Invariant 4: iron-proxy — No Raw Credentials in Sandboxes

This is the invariant that gives Centaur its security model. Sandbox pods are started with **stub** values for all third-party API keys. The real values live on a per-sandbox [iron-proxy](https://docs.iron.sh) pod.

### How it works

```text
Sandbox pod
  harness CLI
     │
     │ HTTPS (HTTPS_PROXY=http://firewall:8080)
     ▼
iron-proxy pod  (mitmproxy TLS MITM)
     │ matches outbound host and header
     │ replaces stub value with real credential from secrets service
     ▼
Upstream API  (api.anthropic.com, api.openai.com, api.github.com, …)
```

The proxy config (`iron-proxy.yaml`) sets `tls.mode: mitm` and issues a MITM CA cert to the sandbox at startup. The proxy's `transforms` block defines an allowlist of headers it will pass through — anything not on the list is stripped before the request leaves the cluster.

The credential injection map is managed by `firewall-manager` in the API. Each binding says: "for requests to `api.anthropic.com`, replace the `x-api-key` header with the real Anthropic key."

Agents and tool plugins refer to credentials by name (`secret("ANTHROPIC_API_KEY")`). The tool SDK returns the placeholder string. The proxy does the actual substitution at the network layer, so the plaintext key is never materialized in the pod's memory or filesystem.

Sources: [AGENTS.md:476-520](), [services/iron-proxy/iron-proxy.yaml:1-82](), [services/sandbox/entrypoint.sh:6-8](), [README.md:186-194]()

---

## How a Harness Talks to the API

Inside the pod, the agent harness calls tools via a bash helper (`/usr/local/bin/call`):

```bash
call slack get_channel_history '{"channel":"general"}'
# → POST http://$CENTAUR_API_URL/tools/slack/get_channel_history
```

The sandbox token (`sbx1.*` prefix, 2h TTL, HMAC-signed) is injected as `CENTAUR_API_KEY` at pod creation time and refreshed on each new turn. Tools never receive raw upstream secrets — they call `secret("NAME")` and the proxy handles the rest.

The wire between the API and the sandbox is **stdin/stdout NDJSON in Anthropic message format**:

```
→ stdin:  {"type":"turn.start","turn_id":1,"text":"why are billing tests failing?"}
← stdout: {"type":"assistant","message":{"role":"assistant","content":[...]}}
← stdout: {"type":"result","subtype":"success","result":"The tests fail because..."}
← stdout: {"type":"turn.done","turn_id":1,"result":"..."}
```

Harness-specific translation (materializing images to files for Amp, extracting text for Codex) happens inside the sandbox adapter (`harness_session.py`). The API and all clients always speak the canonical Anthropic format.

Sources: [AGENTS.md:178-207](), [services/api/api/sandbox/harness_protocol.py:1-60]()

---

## Failure Modes and Recovery

| Failure | What breaks | How it recovers |
|---------|-------------|-----------------|
| Client disconnects mid-stream | SSE connection drops | Reconnect with `after_event_id`; API replays from `agent_execution_events` |
| API restarts | In-memory `_runtime` dict is cleared | Rebuilt lazily from `sandbox_sessions` on next request |
| Sandbox pod dies mid-turn | Active turn is lost in-memory | `inflight_turn_input` in Postgres; `replay_inflight_turn()` re-sends to replacement pod |
| Workflow worker restarts | Running handler state is lost | Handler re-runs top-to-bottom; `ctx.step()` returns cached results for completed steps |
| iron-proxy restarts | Credential injection pauses | Key-injection map rebuilt from secrets-service cache |
| Pod capacity limit reached | New spawn fails | `_evict_idle_sessions_for_capacity()` stops oldest idle pods before cold-spawning |

Sources: [docs/pages/architecture.mdx:109-118](), [services/api/api/agent.py:334-395]()

---

## The Overlay Model

Centaur is designed to be extended without forking. The base repo (`paradigmxyz/centaur`) provides the control plane, workflow engine, and sandbox runtime. An org-level overlay repo sits alongside it:

```text
your-deployment/
├── centaur/              ← this repo (kernel)
└── centaur-overlay/      ← org tools, workflows, skills, personas
```

The Helm chart mounts the overlay at `/app/overlay/org`. The sandbox entrypoint merges overlays into `workspace/AGENTS.md` in order — base prompt, then org overlay. Later entries win on name collision.

This lets teams add tools, personas, workflows, and prompt customizations without touching the core repository.

Sources: [AGENTS.md:338-348](), [services/sandbox/entrypoint.sh:175-197]()

---

## Summary

Centaur is a durable control plane built on four hard invariants: one shared bot, one isolated sandbox pod per Slack thread, Postgres as the only source of truth, and iron-proxy to keep real credentials away from agent code. Once you see that every API call writes to Postgres before doing anything observable, and that the sandbox pod only ever sees placeholder strings for credentials, the rest of the system — tools, workflows, overlays, persona injection, harness adapters — becomes straightforward extensions of those four commitments. The architecture doc captures this precisely: "the event stream is the client contract; Slack and other clients should reconnect with `after_event_id` instead of trying to reconstruct state locally." Sources: [docs/pages/architecture.mdx:38-42]()

---

## 02. Durability Invariants & Failure Modes

> What Postgres owns, what lives only in-process, how reconnects and pod replacements are safe, and the exact failure scenarios where state can be lost or replayed incorrectly. The distinction between the event stream (client contract) and in-memory runtime state is the critical boundary.

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/02-durability-invariants-failure-modes.md
- Generated: 2026-05-21T23:43:45.898Z

### Source Files

- `services/api/api/agent.py`
- `services/api/api/runtime_control.py`
- `services/api/api/models.py`
- `services/api/tests/test_agent_resilience.py`
- `services/api/tests/test_inflight_replay.py`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/agent.py](services/api/api/agent.py)
- [services/api/api/sandbox/base.py](services/api/api/sandbox/base.py)
- [services/api/api/models.py](services/api/api/models.py)
- [services/api/api/db.py](services/api/api/db.py)
- [services/api/api/runtime_control.py](services/api/api/runtime_control.py)
- [services/api/tests/test_agent_resilience.py](services/api/tests/test_agent_resilience.py)
- [services/api/tests/test_inflight_replay.py](services/api/tests/test_inflight_replay.py)
- [services/api/db/migrations/001_initial.sql](services/api/db/migrations/001_initial.sql)
- [services/api/db/migrations/006_add_inflight_turn_state.sql](services/api/db/migrations/006_add_inflight_turn_state.sql)
- [services/api/db/migrations/008_agent_runtime_control_plane.sql](services/api/db/migrations/008_agent_runtime_control_plane.sql)
</details>

# Durability Invariants & Failure Modes

Centaur runs AI agents in ephemeral sandbox containers (Kubernetes pods) while serving clients over long-lived Server-Sent Event streams. The core durability challenge is that containers can crash, be replaced, or be evicted at any moment, yet clients and downstream callers must receive a consistent, deduplicated event stream. This page maps exactly what Postgres owns, what lives only in process memory, and the specific failure scenarios where state can be lost, replayed, or delivered more than once.

Understanding the boundary between the **event stream** (the SSE wire a client reads) and **in-memory runtime state** (stream handles, turn counters) is the central design question. Postgres is the source of truth for session identity and turn lifecycle; the process-local `_runtime` dict is a disposable cache of IO handles that gets rebuilt on reconnect.

---

## State Ownership Map

### What Postgres Owns (Durable)

All session and turn state that must survive a pod replacement lives in Postgres. The key tables and columns are:

**`sandbox_sessions`** — one row per active `thread_key`:

| Column | Purpose |
|---|---|
| `thread_key` (PK) | Stable identifier for a user's conversation thread |
| `sandbox_id` | The Kubernetes pod name of the current container |
| `state` | Lifecycle state: `creating → running → idle → suspended / gone / stopped` |
| `agent_thread_id` | The upstream agent's conversation ID (for resume after replacement) |
| `last_delivered_id` | Cursor into `chat_messages`; which messages have been flushed into stdin |
| `inflight_turn_id` | UUID of the active turn; non-null if a turn is in progress |
| `inflight_turn_input` | Full JSONB payload sent to the sandbox — the exact bytes used for replay |
| `inflight_attempts` | How many times this turn has been attempted (incremented on each replay) |
| `last_result` | Final turn result text; written atomically with turn completion |
| `wire_lease_id` | Non-null while a client SSE wire is connected; heartbeated every 30 s |

Sources: [services/api/db/migrations/001_initial.sql:2-13](), [services/api/db/migrations/006_add_inflight_turn_state.sql:4-11]()

**`chat_messages`** — append-only message log:

Every user message, system message, and assistant response is appended here. The `last_delivered_id` column in `sandbox_sessions` is a cursor into this table. On reconnect or replay, unflushed messages are re-read and re-delivered to the sandbox stdin.

Sources: [services/api/db/migrations/001_initial.sql:15-23]()

**Control-plane tables** (added in migration 008):

- `agent_runtime_assignments` — records which runtime container is assigned to a thread, along with prompt identity hash for change detection.
- `agent_execution_requests` — durable queue for the execution worker pool; each execution row tracks `status` (`queued → running → completed / failed_permanent / cancelled`).
- `agent_spawn_requests` / `agent_release_requests` — idempotency records keyed by caller-supplied `spawn_id` / `release_id`.
- `agent_message_requests` — message delivery records that bridge spawn and execution events.

Sources: [services/api/db/migrations/008_agent_runtime_control_plane.sql:3-85]()

### What Lives Only In-Process (Ephemeral)

```text
_runtime: dict[sandbox_id → RuntimeState]
```

`RuntimeState` (defined in `sandbox/base.py`) holds:

| Field | Type | Meaning |
|---|---|---|
| `turn_counter` | `int` | Monotonically increasing per-process turn number |
| `stdout_stream` | backend handle | Open async iterator for sandbox stdout |
| `stdin_stream` | backend handle | Open write handle for sandbox stdin |
| `attach_context` | backend-specific | Context manager for the current attach session |
| `prefetched_stdout` | `list[str] | None` | Lines buffered before live attach |
| `stdout_read_lock` | `asyncio.Lock` | Prevents concurrent stdout readers |
| `last_result` | `str | None` | Turn result; a best-effort bridge before DB write commits |

These values are populated by `_get_runtime()` and dropped by `_drop_runtime()`. They are not persisted. After a pod restart or API server restart, this dict is empty and every `sandbox_id` gets a fresh `RuntimeState()` on first access.

Sources: [services/api/api/sandbox/base.py:13-28](), [services/api/api/agent.py:97-112]()

---

## The Critical Boundary: Event Stream vs. Runtime State

```text
┌──────────────────────────────────────────────────────────────┐
│  CLIENT (SSE consumer)                                       │
│    ← receives: wire.ready, turn.started, content_block_*,   │
│                turn.done, keepalive                          │
└───────────────────┬──────────────────────────────────────────┘
                    │  SSE wire (EventSourceResponse)
┌───────────────────▼──────────────────────────────────────────┐
│  API SERVER PROCESS                                          │
│  stream_connect() / _stream_stdout()                         │
│  ┌────────────────┐  ┌───────────────────────────────────┐   │
│  │ _runtime       │  │ DB writes (before yield)          │   │
│  │ turn_counter   │  │ _db_set_inflight_turn()           │   │
│  │ stdout_stream  │  │ _db_complete_inflight_turn()      │   │
│  │ last_result    │  │ _persist_turn_messages()          │   │
│  └────────────────┘  └───────────────────────────────────┘   │
└───────────────────┬──────────────────────────────────────────┘
                    │  asyncpg pool
┌───────────────────▼──────────────────────────────────────────┐
│  POSTGRES                                                    │
│  sandbox_sessions  chat_messages  agent_execution_requests   │
└──────────────────────────────────────────────────────────────┘
```

The ordering guarantee: **Postgres is written before the `turn.done` event is yielded to the client**. The code in `_stream_stdout` calls `asyncio.gather(_persist_turn_messages(...), _db_complete_inflight_turn(...))` and only then emits the `turn.done` SSE payload. A client that disconnects just before receiving `turn.done` can reconnect and get the event from the newly-idle session.

Sources: [services/api/api/agent.py:933-953]()

---

## Lifecycle States and Valid Transitions

```mermaid
stateDiagram-v2
    [*] --> creating : get_or_spawn()
    creating --> running : stream_connect() called
    running --> idle : turn.done committed to DB
    running --> suspended : reconcile_tick() / idle TTL eviction
    idle --> running : inject_stdin() or stream_connect()
    idle --> suspended : idle TTL expires (default 24 h)
    running --> gone : container crash, no wire, no inflight
    suspended --> running : get_or_spawn() (resume path)
    gone --> [*] : cleaned up by reconcile Step E
    stopped --> [*] : cleaned up by reconcile Step E
```

The state machine is enforced by a `CHECK` constraint on `sandbox_sessions.state`. The `db.py` startup check verifies all eight required states are present before accepting traffic.

Sources: [services/api/api/db.py:19-30](), [services/api/api/agent.py:1516-1736]()

---

## Reconnect and Pod Replacement Safety

### Normal Client Reconnect (Same Container)

1. `stream_connect()` is called again on the same `sandbox_id`.
2. `_get_runtime()` returns the existing `RuntimeState` (turn counter preserved in process memory).
3. `backend.attach()` re-opens the stdout iterator from the container.
4. A `wire.ready` event is emitted with the current `turn_counter` so the client knows where it is.
5. If a turn is still running, the container continues emitting events; the client resumes reading from the live stream.

If the first stream connection dropped mid-turn without receiving `turn.done`, the container is still running and the client will receive the remaining events starting from the reattach point.

Sources: [services/api/api/agent.py:1024-1109]()

### Container Replacement (Pod Restart / Node Eviction)

`get_or_spawn()` handles this transparently:

1. The DB row for the `thread_key` exists with a stale `sandbox_id`.
2. `backend.status(session)` returns `"gone"` or similar.
3. The old row is deleted; `agent_thread_id`, `last_delivered_id`, `inflight_turn_id`, `inflight_turn_input`, and `inflight_attempts` are saved in local variables before the delete.
4. A new container is spawned, and `_db_insert_session()` inserts a new row carrying the preserved fields.
5. `inject_stdin()` or `replay_inflight_turn()` re-injects the in-flight turn payload at `attempts + 1`.
6. `_flush_pending()` re-reads any `chat_messages` newer than `last_delivered_id` to catch up messages that weren't delivered before the crash.

The `resume_thread_id=old_agent_thread_id` argument to `backend.create()` tells the new container to resume the upstream agent's conversation thread, preserving conversation context inside the harness.

Sources: [services/api/api/agent.py:613-750](), [services/api/api/agent.py:1287-1369]()

### Stdin Broken Pipe Recovery

Both `inject_stdin()` and `replay_inflight_turn()` catch `BrokenPipeError / OSError` on the first `write_stdin` attempt. If the container reports status `"running"`, a `reattach_stdin()` is performed and the write is retried once. If the container is gone, a `RuntimeError` is raised and the caller falls into the container-replacement path.

Sources: [services/api/api/agent.py:1184-1210](), [services/api/api/agent.py:1325-1353]()

### EOF Reattach (Transient Stream Drop)

`_stream_stdout` implements an EOF reattach loop: if the stdout iterator returns EOF but `backend.status()` still returns `"running"`, `close_streams()` is called and `backend.attach()` is retried up to `STREAM_EOF_REATTACH_MAX` times (default: 6) with `STREAM_EOF_REATTACH_BACKOFF_S` (default: 1 s) delay. This handles transient network interruptions to the container without triggering a full spawn cycle.

Sources: [services/api/api/agent.py:966-1007](), [services/api/tests/test_agent_resilience.py:72-107]()

---

## The In-Flight Turn Invariant

The in-flight turn is the most critical durability invariant. The lifecycle is:

```
inject_stdin() called
  → _db_set_inflight_turn() [inflight_turn_id, inflight_turn_input, attempts=1]
  → backend.write_stdin()
  → sandbox runs the turn

sandbox emits is_turn_done event
  → _db_complete_inflight_turn() [clears inflight_turn_id, writes last_result]
  → turn.done yielded to client
```

**Key guarantee**: `inflight_turn_input` is written to Postgres *before* the payload reaches the sandbox stdin. If the process dies between those two steps, the turn input is safe and can be replayed. If the process dies after the sandbox receives the input but before `_db_complete_inflight_turn()`, the row still has `inflight_turn_id` set and the replay path (`replay_inflight_turn()`) will re-send the same input to a new container.

Sources: [services/api/api/agent.py:1159-1165](), [services/api/api/agent.py:1287-1310]()

---

## Failure Modes and Exact Loss Scenarios

### Scenario 1: API Process Crash During a Turn (In-Flight Turn Preserved)

**What happens**: Process dies after `_db_set_inflight_turn()` but before `_db_complete_inflight_turn()`.

**State**: `inflight_turn_id` is non-null in Postgres; `_runtime` is empty in the new process.

**Recovery**: On next `get_or_spawn()`, the old row is found, `inflight_turn_id` and `inflight_turn_input` are preserved through the delete-and-reinsert cycle. `replay_inflight_turn()` re-sends the payload. **No user input is lost**; the turn may be retried with `inflight_attempts` incremented.

**Risk**: The sandbox may have already completed the turn but the result was not written. In that case, `_db_complete_inflight_turn()` was never called, and the replay will produce a second agent response to the same input. This is the primary **double-delivery risk**.

### Scenario 2: API Process Crash After Turn Completes (Last Result Preserved)

**What happens**: `_db_complete_inflight_turn()` wrote `last_result` and cleared `inflight_turn_id`, but `_persist_turn_messages()` for the assistant's `chat_messages` row failed (it is called concurrently and its failure is silently logged, not re-raised).

**State**: `last_result` is in Postgres but `chat_messages` may lack the assistant message.

**Impact**: The assistant turn result is visible via `get_status()`, but `_flush_pending()` on the next reconnect will not replay the missing assistant message (assistant messages are filtered out of flush, except `history_backfill` rows). The assistant response is **not replayed** to the sandbox on resume; only user messages are re-flushed.

Sources: [services/api/api/agent.py:452-483](), [services/api/api/agent.py:1767-1803]()

### Scenario 3: Container Crash Without Wire Connected (Turn Lost)

**What happens**: The sandbox pod is evicted while idle, and `reconcile_tick()` Step A detects the container is gone.

**State**: `reconcile_tick()` calls `_mark_inactive()`, which sets `state = 'suspended'` and **clears `inflight_turn_id` and `inflight_turn_input`**.

```python
# reconcile_tick Step D: stale inflight rows reaped after IDLE_TTL_S
"inflight_turn_id = NULL, inflight_turn_input = NULL, inflight_started_at = NULL"
```

**Impact**: Any in-flight turn that was active but had no wire (e.g., the client disconnected) and that times out past `IDLE_TTL_S` (default 24 h) will have its in-flight payload wiped. The user must resend their request.

Sources: [services/api/api/agent.py:1525-1549](), [services/api/api/agent.py:1666-1710]()

### Scenario 4: Wire Lease Staleness (Client Disconnect Not Detected Immediately)

**What happens**: The SSE client disconnects but the heartbeat task's 30-second poll is not yet overdue.

**State**: `wire_lease_id` remains set in Postgres; `supervise_wires()` or `reconcile_tick()` will detect it after 120 seconds of no heartbeat and clear the lease.

**Impact**: The session appears "delivering" for up to 120 seconds after the client disconnects. This does not cause data loss but may delay capacity eviction.

Sources: [services/api/api/agent.py:1375-1436]()

### Scenario 5: Spawn Race (Concurrent Spawn for Same Thread Key)

**What happens**: Two requests race to create a session for the same `thread_key`.

**Protection**: `_db_insert_session()` uses `INSERT ... ON CONFLICT (thread_key) DO NOTHING RETURNING thread_key`. The loser detects `row is None`, stops its container with `backend.stop_by_id()`, drops its runtime, and fetches the winning session from Postgres.

**Impact**: One container is wasted (stopped immediately). The client gets the winner's session. No state loss.

Sources: [services/api/api/agent.py:727-748]()

### Scenario 6: Double Completion (turn.done Emitted Twice)

**What happens**: A reconnect arrives during the window between `_db_complete_inflight_turn()` and the `turn.done` yield, and `stream_reconnect()` is called with `logs=True`.

**Protection**: `stream_reconnect()` accepts a `skip_done_count` parameter to skip already-seen `turn.done` events from the container's log buffer. The client is responsible for tracking which `turn.done` events it has already processed.

**Risk**: If the container has already emitted `turn.done` to its log buffer and the reconnecting client does not supply the correct `skip_done_count`, it may receive a duplicate `turn.done`.

Sources: [services/api/api/agent.py:1739-1764]()

---

## The Flush Pipeline: Message Cursor Semantics

`_flush_pending()` reads `chat_messages` rows that the sandbox has not yet received. The cursor (`last_delivered_id`) points to the `id` of the last message delivered to stdin.

```python
# Flush logic (simplified):
# If last_delivered_id is set: return rows created AFTER that message's created_at
# If NULL: return all replayable messages
role_filter = "(role <> 'assistant' OR metadata->>'history_backfill' = 'true')"
```

**What is re-flushed on reconnect**: User messages and system messages not yet seen by the sandbox. History-backfill assistant messages (imported Slack context) are also included.

**What is NOT re-flushed**: Regular assistant messages generated by the sandbox. The sandbox is assumed to hold its own context for these via the harness's internal memory (`agent_thread_id` resume).

**Cursor advancement**: `_advance_cursor()` writes `last_delivered_id` only after `backend.write_stdin()` succeeds. If stdin write fails fatally, the cursor is not advanced and the same messages will be re-flushed on the next attempt.

Sources: [services/api/api/agent.py:452-521]()

---

## Periodic Reconciliation

`reconcile_tick()` runs every 60 seconds and enforces five invariants:

| Step | Query target | Action |
|---|---|---|
| A | `state IN (running, idle, delivering, error)` | Check backend; mark `suspended` if container gone |
| B | `state = idle AND updated_at < NOW() - IDLE_TTL_S` | Stop container; mark `suspended` |
| C | `state IN (running, delivering, error)` with no wire, no inflight, no execution | Stop container; mark `suspended` |
| D | `state IN (running, delivering, error)` with stale `inflight_turn_id` and no execution | Reap the in-flight record; mark `suspended` |
| E | `state IN (gone, stopped)` older than 1 hour, or `suspended` older than `SUSPENDED_RETENTION_S` (7 days) | Delete the row |

The reconciler is conservative: transient backend errors cause `continue` (skip the row, don't destroy it). This prevents accidental reaping of healthy sessions during brief API hiccups.

Sources: [services/api/api/agent.py:1516-1736]()

---

## Schema Startup Validation

At every startup, `check_schema_compatibility()` verifies that the Postgres schema has:

- All eight required `sandbox_sessions` state values in the check constraint
- All eight required columns (`agent_thread_id`, `inflight_turn_id`, `inflight_turn_input`, `inflight_started_at`, `inflight_attempts`, `last_result`, `last_result_at`, `trace_id`)
- All required migration versions applied (`005`, `006`, `007`, `008`, `009`, `010`, `011`, `035`)

If any check fails, the health endpoint reflects incompatibility and the API should not accept traffic. This prevents a rolled-back migration from causing silent data loss where the code expects durable columns that are missing.

Sources: [services/api/api/db.py:177-240]()

---

## Summary

Centaur's durability model is built on a single rule: **Postgres is written before the event is yielded to the client**. The `_runtime` dict is a disposable cache; losing it on restart costs nothing except the stream handles. The two columns that make pod replacement transparent are `inflight_turn_input` (the exact bytes to replay) and `last_delivered_id` (which messages the sandbox already received). The primary failure mode where state can be silently lost is stale in-flight records reaped by `reconcile_tick()` Step D after `IDLE_TTL_S` elapses with no active wire and no execution — a scenario that occurs when both the client and the container die before the turn completes. All other failure paths either replay safely or surface an explicit error to the caller.

Sources: [services/api/api/agent.py:933-939](), [services/api/api/agent.py:1667-1710]()

---

## 03. API Lifecycle: spawn → message → execute → events → release

> The five-step durable API lifecycle, how each call maps to a Postgres write, what the SSE event stream guarantees, and what happens when a client reconnects mid-run with after_event_id. Also covers the Slackbot ingress path including HMAC signature verification.

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/03-api-lifecycle-spawn-message-execute-events-release.md
- Generated: 2026-05-21T23:45:58.876Z

### Source Files

- `services/api/api/app.py`
- `services/api/api/agent.py`
- `services/api/api/runtime_control.py`
- `services/api/api/slackbot_client.py`
- `docs/pages/architecture.mdx`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/app.py](services/api/api/app.py)
- [services/api/api/agent.py](services/api/api/agent.py)
- [services/api/api/runtime_control.py](services/api/api/runtime_control.py)
- [services/api/api/routers/agent.py](services/api/api/routers/agent.py)
- [services/api/api/slackbot_client.py](services/api/api/slackbot_client.py)
- [services/slackbot/src/slack/signature.ts](services/slackbot/src/slack/signature.ts)
- [services/slackbot/src/index.ts](services/slackbot/src/index.ts)
- [services/api/tests/test_direct_api_e2e.py](services/api/tests/test_direct_api_e2e.py)
</details>

# API Lifecycle: spawn → message → execute → events → release

Centaur's public-facing API is built around a five-step durable lifecycle that turns a natural-language request into a running sandbox, streams its output to connected clients, and then releases the runtime when the thread is done. Every step is a distinct HTTP call and every call writes durably to Postgres before the response is returned, so the system can survive process restarts, network blips, and mid-turn client disconnections without losing work.

This page covers each step in detail, explains the Postgres writes that back each transition, describes the SSE event stream and its reconnection semantics, and explains how the Slackbot ingress path feeds into the same lifecycle after verifying a Slack HMAC signature.

---

## Overview: the five calls

```text
POST /agent/spawn     → agent_runtime_assignments row (state=active)
POST /agent/message   → agent_message_requests + chat_messages rows
POST /agent/execute   → agent_execution_requests row (status=queued)
GET  /agent/threads/{key}/events (SSE) → streams agent_execution_events rows
POST /agent/threads/{key}/release → agent_runtime_assignments state=released
```

All five endpoints live under the `APIRouter` at prefix `/agent`, require an API-key via `verify_api_key`, and accept an optional scope guard `agent:execute`.

Sources: [services/api/api/routers/agent.py:58-62]()

---

## Step 1 — spawn

**Endpoint:** `POST /agent/spawn`

`spawn` creates or re-attaches a runtime assignment for a `thread_key`. A runtime assignment is the durable record binding a thread to a specific sandbox container and prompt identity for the lifetime of that conversation.

### What happens

1. The caller supplies `thread_key` and an optional `spawn_id` (idempotency key), `harness`, `engine`, `persona_id`, and `agents_md_override`.
2. `spawn_assignment` checks `agent_spawn_requests` for an existing row matching `(thread_key, spawn_id)` and returns the cached response if found.
3. A Postgres advisory lock `pg_advisory_xact_lock(hashtext(thread_key))` serializes concurrent spawns for the same thread.
4. `get_or_spawn` in `agent.py` resolves the container: it checks `sandbox_sessions` for a live session, falls back to the warm pool (`claim_container`), and finally does a cold spawn via the backend.
5. Inside the same transaction, `agent_runtime_assignments` is either re-used (if an active assignment with the same prompt identity already exists) or a new row is inserted with an incremented `assignment_generation`.
6. The idempotency response is recorded in `agent_spawn_requests`.

### Return value

```json
{
  "ok": true,
  "runtime_id": "<sandbox-id>",
  "thread_key": "...",
  "assignment_generation": 1,
  "persona_id": null,
  "prompt_ref": "harness:codex",
  "effective_agents_md_sha256": "<sha256>"
}
```

`assignment_generation` is the handle clients pass to all subsequent calls.

Sources: [services/api/api/runtime_control.py:417-636](), [services/api/api/agent.py:613-750]()

---

## Step 2 — message

**Endpoint:** `POST /agent/message`

`message` appends a user turn to durable storage without executing anything. The sandbox does not receive the input yet.

### Postgres writes

- `chat_messages` — one row per logical message. `id` is `msg:{thread_key}:{message_id}`. Inline `base64` image/document blobs are extracted into `attachments` and replaced with `attachment_ref` parts.
- `agent_message_requests` — idempotency log with `(thread_key, message_id, request_hash, event_json)`.

The advisory lock on `thread_key` runs inside the transaction so concurrent message inserts cannot interleave with a concurrent spawn.

Validation: the active assignment's `assignment_generation` must match the caller's value, otherwise the call fails with `ASSIGNMENT_GENERATION_STALE`.

Sources: [services/api/api/runtime_control.py:735-902]()

---

## Step 3 — execute

**Endpoint:** `POST /agent/execute`

`execute` enqueues a turn for the worker pool. The queue entry carries the `delivery` target (Slack channel, platform, etc.) and wakes an in-process worker.

### Postgres writes

- `agent_execution_requests` — one row inserted with `status='queued'`, `silence_deadline_at`, and `hard_deadline_at`. `execute_id` is the idempotency key.
- `agent_execution_events` — an initial `execution_state` event with `status='queued'` is appended immediately.
- `agent_final_delivery_outbox` — if the delivery platform is not `'dev'` and live-delivery is not enabled, an outbox row is inserted so the final answer can be sent to Slack even after SSE clients disconnect.

After the DB writes, `_worker_wake.set()` pings the in-process execution worker pool to pick up the new row without polling.

Sources: [services/api/api/runtime_control.py:1151-1317](), [services/api/api/routers/agent.py:251-295]()

### Auto-orchestration convenience path

When `assignment_generation` is omitted but `message` is set, `POST /agent/execute` auto-orchestrates all three steps internally. A single call runs `spawn_assignment → append_message → enqueue_execution` with generated idempotency IDs.

Sources: [services/api/api/routers/agent.py:298-358]()

---

## Step 4 — events (SSE stream)

**Endpoint:** `GET /agent/threads/{thread_key}/events`

This is the real-time output channel. It is a persistent Server-Sent Events stream backed by a polling loop over `agent_execution_events`.

### Query parameters

| Parameter | Default | Meaning |
|---|---|---|
| `after_event_id` | `0` | Resume cursor: only return rows with `event_id > after_event_id` |
| `execution_id` | none | Filter to events for one execution |
| `poll_ms` | `500` | Polling interval (clamped 50 ms – 5 s) |

### How the loop works

```python
cursor = max(0, after_event_id)
while True:
    rows = await pool.fetch(
        "SELECT event_id, event_kind, event_json FROM agent_execution_events "
        "WHERE thread_key = $1 AND event_id > $2 [AND execution_id = $3] "
        "ORDER BY event_id ASC LIMIT 200",
        thread_key, cursor, ...
    )
    if not rows:
        if execution_id:
            snapshot = await get_execution_terminal_snapshot(pool, execution_id)
            if snapshot:  # execution is already terminal — emit final state and return
                yield ServerSentEvent(...)
                return
        await asyncio.sleep(poll_s)
        continue
    for row in rows:
        cursor = int(row["event_id"])
        yield ServerSentEvent(id=str(cursor), event=row["event_kind"], data=...)
```

Sources: [services/api/api/routers/agent.py:884-957]()

### SSE event schema

Each event has:
- `id`: the `event_id` integer (monotonically increasing per Postgres insert order)
- `event`: the `event_kind` string (e.g. `execution_state`, `harness_event`)
- `data`: JSON payload

```text
id: 42
event: execution_state
data: {"type":"execution.state","execution_id":"exe_abc","thread_key":"...","status":"completed","result_text":"..."}
```

Keepalive comments are emitted every 15 s via `sse-starlette`'s `ping_message_factory`.

### Reconnecting with `after_event_id`

When a client disconnects and reconnects, it passes the highest `event_id` it received as `after_event_id`. The server resumes from `event_id > cursor` so no events are duplicated and none are missed.

If the execution is already terminal when the client reconnects (e.g., the job completed while the client was disconnected), and no new rows exist above the cursor, `get_execution_terminal_snapshot` synthesizes a final `execution_state` event from `agent_execution_requests` and returns it immediately rather than hanging the stream indefinitely.

The E2E test verifies this contract: the reconnect stream emits exactly the terminal event at the same `event_id` as the original stream's last event.

Sources: [services/api/api/routers/agent.py:921-934](), [services/api/api/runtime_control.py:1114-1148](), [services/api/tests/test_direct_api_e2e.py:169-188]()

### Sandbox stdout → SSE event path

The execution worker attaches to a sandbox's stdout via `_stream_stdout` in `agent.py`. Raw NDJSON lines from the container are:

1. Parsed as JSON.
2. Normalized by `normalize_harness_event(engine, evt)` into canonical event dicts.
3. Persisted into `agent_execution_events` via `append_execution_event`.
4. Yielded to `EventSourceResponse` as `{"data": "<json>"}` dicts.

The harness-level turn boundary is detected by `is_turn_done`. Before emitting `turn.done`, the worker writes `_db_complete_inflight_turn` (state=idle, clear inflight) and `_persist_turn_messages` (write assistant message to `chat_messages`) atomically via `asyncio.gather`. This ordering guarantees a reconnecting client cannot read a terminal SSE event before the durable state is committed.

Sources: [services/api/api/agent.py:839-953](), [services/api/api/agent.py:932-939]()

---

## Step 5 — release

**Endpoint:** `POST /agent/threads/{thread_key}/release`

`release` tears down the runtime assignment for a thread. It is idempotent via `agent_release_requests`.

### What happens

1. The advisory lock on `thread_key` is taken.
2. The active `agent_runtime_assignments` row is transitioned to `state='released'`.
3. If `cancel_inflight=true`, any `queued/running/cancel_requested/retry_wait` executions for the thread are cancelled in the same transaction.
4. If `stop_runtime=true` (default), the sandbox container is stopped asynchronously.
5. The response is written to `agent_release_requests` for idempotency.

Sources: [services/api/api/runtime_control.py:1643-1720]()

---

## Database state machine

```text
                   spawn
                     │
                     ▼
         ┌─── agent_runtime_assignments ───┐
         │   state: active                 │
         │   assignment_generation: N      │
         └─────────────────────────────────┘
                     │  message
                     ▼
         ┌─── agent_message_requests ──────┐
         │   chat_messages (user turn)     │
         └─────────────────────────────────┘
                     │  execute
                     ▼
         ┌─── agent_execution_requests ────┐
         │   status: queued → running      │
         │          → completed            │
         │          → failed_permanent     │
         │          → cancelled            │
         └─────────────────────────────────┘
                     │  worker streams
                     ▼
         ┌─── agent_execution_events ──────┐
         │   event_id (monotone)           │
         │   event_kind / event_json       │
         └─────────────────────────────────┘
                     │  release
                     ▼
         agent_runtime_assignments
           state: released
```

Every transition is a Postgres row write. There is no in-process queue between steps.

---

## Sandbox session durability

Independent of the control-plane tables above, `sandbox_sessions` tracks live container state. Key columns:

| Column | Role |
|---|---|
| `state` | `idle`, `running`, `delivering`, `error`, `suspended`, `gone`, `stopped` |
| `inflight_turn_id` | UUID of the active turn; set before stdin write, cleared on completion |
| `inflight_turn_input` | Full turn payload — enables replay if the container is replaced mid-turn |
| `last_delivered_id` | Cursor into `chat_messages`; prevents double-delivery on reconnect |
| `wire_lease_id` | UUID granted when `stream_connect` attaches to stdout |

If a sandbox dies mid-turn, `replay_inflight_turn` reads `inflight_turn_input` back from Postgres and re-sends it to a freshly spawned container with an incremented attempt counter.

Sources: [services/api/api/agent.py:221-266](), [services/api/api/agent.py:1287-1369]()

---

## Slackbot ingress path

The Slackbot is a separate Node.js/Hono service (`services/slackbot`). All Slack-originated events pass through HMAC signature verification before any application logic runs.

### HMAC verification

The `verifySlackSignature` function in `services/slackbot/src/slack/signature.ts` enforces Slack's v0 signing scheme:

```
base = "v0:{X-Slack-Request-Timestamp}:{raw_body}"
expected = "v0=" + HMAC-SHA256(signing_secret, base).hex()
```

Steps:
1. Read the raw request body (before JSON parsing).
2. Reject if `X-Slack-Signature` or `X-Slack-Request-Timestamp` headers are missing.
3. Reject if the timestamp is more than 5 minutes from `now` (replay protection).
4. Compute HMAC-SHA256 and compare with `timingSafeEqual` to prevent timing attacks.
5. Return `{ ok: false, status: 401 }` on any failure; `{ ok: true }` on success.

The middleware `slackSignatureMiddleware` applies this check to all inbound Slack routes (`/api/slack/events`, `/api/slack/actions`, `/api/slack/options`, `/api/slack/commands`).

Sources: [services/slackbot/src/slack/signature.ts:1-46](), [services/slackbot/src/index.ts:90-143]()

`SLACK_SIGNING_SECRET` is a required environment variable validated at container startup by `entrypoint.sh`.

Sources: [services/api/entrypoint.sh:9]()

### From Slack event to execution

Once signature verification passes, the Slackbot calls the Centaur API:
1. `POST /api/slack/agent-sessions` (via `slackbot_client.open_agent_session`) to open a UI session with a progress channel in Slack.
2. Calls into the `spawn → message → execute` lifecycle.
3. The execution worker streams harness events back through `slackbot_client.harness_event` and `slackbot_client.session_text` / `session_step`, progressively rendering the agent's output as Slack message blocks.
4. On completion, `slackbot_client.session_done` closes the Slack session.

The `slackbot_client.py` module is a thin async HTTP client for the Slackbot's internal REST API. It uses bearer-token auth (`SLACKBOT_API_KEY`) and retries on `{408, 429, 500-504}` with exponential back-off (base 0.25 s, 3 attempts).

Sources: [services/api/api/slackbot_client.py:32-85](), [services/api/api/slackbot_client.py:122-188]()

---

## Sequence diagram: full lifecycle

```mermaid
sequenceDiagram
    participant C as Client
    participant API as FastAPI /agent
    participant W as Execution Worker
    participant SB as Sandbox (container)
    participant DB as Postgres

    C->>API: POST /agent/spawn {thread_key, spawn_id}
    API->>DB: pg_advisory_lock; upsert sandbox_sessions; INSERT agent_runtime_assignments
    API-->>C: {assignment_generation: 1, runtime_id}

    C->>API: POST /agent/message {thread_key, assignment_generation: 1, message_id}
    API->>DB: INSERT chat_messages + agent_message_requests
    API-->>C: {ok: true, message_id}

    C->>API: POST /agent/execute {thread_key, assignment_generation: 1, execute_id}
    API->>DB: INSERT agent_execution_requests (status=queued) + execution_state event
    API-->>C: 202 {execution_id}

    C->>API: GET /agent/threads/{key}/events?after_event_id=0
    Note over API: poll loop begins

    W->>DB: claim execution row (status→running)
    W->>SB: flush pending messages → write stdin
    SB-->>W: stdout NDJSON stream
    W->>DB: INSERT agent_execution_events (harness events)
    W-->>API: SSE event stream

    API-->>C: id:1 event:harness_event data:{...}
    API-->>C: id:2 event:execution_state data:{status:running}
    SB-->>W: turn.done
    W->>DB: UPDATE sandbox_sessions (inflight cleared, state=idle)
    W->>DB: INSERT chat_messages (assistant reply)
    W->>DB: UPDATE agent_execution_requests (status=completed)
    W->>DB: INSERT agent_execution_events (execution_state completed)
    API-->>C: id:3 event:execution_state data:{status:completed,result_text:...}

    C->>API: POST /agent/threads/{key}/release
    API->>DB: UPDATE agent_runtime_assignments (state=released)
    API-->>C: {ok: true, released: true}
```

---

## Failure modes and invariants

| Failure | What breaks | Recovery |
|---|---|---|
| Client disconnects mid-stream | SSE loop exits; no data lost | Reconnect with `after_event_id` = last seen `event_id` |
| Execution is already terminal on reconnect | No new DB rows above cursor | `get_execution_terminal_snapshot` synthesizes a final event from the row |
| Sandbox dies mid-turn | `inflight_turn_input` is nil'd by reconciler | `replay_inflight_turn` re-sends payload to new container (attempt counter increments) |
| Slack replay attack | Timestamp too old | `verifySlackSignature` rejects with `stale_signature_timestamp` after 5-minute window |
| Concurrent spawns for same thread | Two API pods race | `pg_advisory_xact_lock` + `ON CONFLICT DO NOTHING` on `sandbox_sessions`; loser calls `backend.stop_by_id` on its own container |
| Repeated turn failures on same thread | Failure loop burns resources | `enqueue_execution` counts `failed_permanent` rows in a 5-minute window; rejects with `THREAD_FAILURE_LOOP` after threshold |
| Idle sandbox exceeds TTL | Container keeps consuming node resources | `reconcile_tick` evicts idle sessions older than `IDLE_TTL_S` (default 24 h) |

Sources: [services/api/api/agent.py:726-748](), [services/api/api/runtime_control.py:1250-1281](), [services/api/api/agent.py:1516-1736]()

---

## Summary

The five-step lifecycle (`spawn → message → execute → events → release`) is designed so that every mutation is a Postgres write that can be replayed or observed independently. The SSE stream is a polling view over `agent_execution_events` rows identified by a monotone `event_id`; clients reconnect by passing `after_event_id` and receive events from exactly the next undelivered row. The Slackbot ingress adds one pre-flight gate — HMAC-SHA256 signature verification with a 5-minute replay window — before feeding events into the same control-plane tables. The overall design has no in-process queues between steps: durability is Postgres, delivery is SSE, and recovery is row replay.

Sources: [services/api/api/runtime_control.py:1071-1088](), [services/slackbot/src/slack/signature.ts:33-45]()

---

## 04. Sandbox Pods & the Warm Pool

> How sandbox Kubernetes pods are created, claimed, and recycled; why the warm pool exists (15-second startup cost); the POOL_EVICT_ON_STARTUP invariant that guarantees new pods run fresh code after a deploy; and the sandbox session state machine (idle → running → delivering → released).

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/04-sandbox-pods-the-warm-pool.md
- Generated: 2026-05-21T23:42:19.280Z

### Source Files

- `services/api/api/warm_pool.py`
- `services/api/api/sandbox/kubernetes.py`
- `services/api/api/sandbox/base.py`
- `services/api/api/sandbox/registry.py`
- `services/api/tests/test_warm_pool.py`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/warm_pool.py](services/api/api/warm_pool.py)
- [services/api/api/sandbox/kubernetes.py](services/api/api/sandbox/kubernetes.py)
- [services/api/api/sandbox/base.py](services/api/api/sandbox/base.py)
- [services/api/api/sandbox/registry.py](services/api/api/sandbox/registry.py)
- [services/api/tests/test_warm_pool.py](services/api/tests/test_warm_pool.py)
- [services/api/api/agent.py](services/api/api/agent.py)
- [services/api/db/migrations/001_initial.sql](services/api/db/migrations/001_initial.sql)
- [services/api/db/migrations/003_add_delivering_state.sql](services/api/db/migrations/003_add_delivering_state.sql)
</details>

# Sandbox Pods & the Warm Pool

Centaur runs each agent conversation inside an isolated Kubernetes pod. Creating a pod from scratch takes roughly 15 seconds — enough latency to make every new conversation feel sluggish. The warm pool eliminates that wait by keeping a small set of pre-booted pods idle in the cluster, ready to be instantly claimed when a new thread arrives.

This page explains how pods are created and structured, how the warm pool maintains its target size, how the `POOL_EVICT_ON_STARTUP` invariant prevents stale code from surviving a deploy, and the full lifecycle of a sandbox session from creation through terminal release.

---

## Pod Structure

Each sandbox is a Kubernetes Pod with:

- A **sandbox container** running the agent harness (codex, amp, claude-code, etc.) as UID 1001 (`_AGENT_UID`), with all Linux capabilities dropped and a seccomp `RuntimeDefault` profile applied. `stdin: true` is set so the API can pipe NDJSON turns in.
- An **iron-proxy sidecar** (per-pod), which forwards egress through a managed firewall. Its health is polled via `/healthz` before the sandbox container starts.
- Optional **overlay init container** that copies org-specific files into a shared `emptyDir` volume before the main containers start.

The pod name is derived deterministically from the thread key via a SHA-1 digest: `centaur-centaur-sandbox-<normalized>-<sha1[:10]>`. Warm pods use a `warm-<timestamp>-<id>` placeholder key and carry the label `centaur.ai/warm=true` so they are distinguishable from assigned pods.

Sources: [services/api/api/sandbox/kubernetes.py:1028-1073](services/api/api/sandbox/kubernetes.py), [services/api/api/sandbox/kubernetes.py:1182-1210](services/api/api/sandbox/kubernetes.py)

---

## The Warm Pool

### Why It Exists

Starting a pod cold involves: creating a prompt Secret, a proxy ConfigMap, a Service, NetworkPolicies, an iron-proxy Pod (and waiting for it to become Ready), then the sandbox Pod itself (and waiting for `/home/agent/.ready`). This easily takes 15 seconds. The warm pool pre-runs all of that so the first user turn connects immediately.

### Configuration

| Env var | Default | Purpose |
|---|---|---|
| `WARM_POOL_SIZE` | `5` | Target number of idle pods to maintain |
| `WARM_POOL_HARNESS` | `codex` | Harness type for warm pods |
| `WARM_POOL_REPLENISH_INTERVAL` | `5.0` s | How often the background loop top-fills the pool |
| `WARM_POOL_BACKEND_TIMEOUT` | `30.0` s | Per-operation timeout for backend calls |
| `WARM_POOL_EVICT_ON_STARTUP` | `1` (enabled) | Whether to kill old warm pods on API start |

Sources: [services/api/api/warm_pool.py:29-42](services/api/api/warm_pool.py)

### Data Model

```python
@dataclass
class WarmContainer:
    sandbox_id: str   # Pod name
    harness:    str
    engine:     str
    created_at: float  # wall-clock epoch for age-based health checks
```

The pool is a plain Python list (`_pool: list[WarmContainer]`). All mutation is serialized through an `asyncio.Lock` (`_pool_lock`) so concurrent claims or replenishments do not race.

Sources: [services/api/api/warm_pool.py:45-53](services/api/api/warm_pool.py), [services/api/api/warm_pool.py:55-59](services/api/api/warm_pool.py)

### Backend Capability Gate

Only backends that declare `supports_warm_pool = True` participate. The `SandboxBackend` ABC defaults this property to `False`; `KubernetesExecutorBackend` overrides it to `True`. The registry always returns `KubernetesExecutorBackend` in production via `auto_configure()`.

Sources: [services/api/api/sandbox/base.py:60-62](services/api/api/sandbox/base.py), [services/api/api/sandbox/kubernetes.py:406-407](services/api/api/sandbox/kubernetes.py), [services/api/api/sandbox/registry.py:24-28](services/api/api/sandbox/registry.py)

---

## Warm Pool Lifecycle

```text
API process start
     │
     ├── POOL_EVICT_ON_STARTUP=true  → _evict_existing_warm()   (kill old pods from prior deploy)
     └── POOL_EVICT_ON_STARTUP=false → _recover_warm()          (adopt surviving pods)
           │
           ▼
     replenish()  ←─────────────────────────────────────────────────────────┐
     (spawn pods until len(_pool) == POOL_SIZE)                             │
           │                                                                 │
           └──► background loop: sleep(POOL_REPLENISH_INTERVAL) ───────────►┘
```

### Replenishment

`replenish()` first health-checks every existing pool entry by calling `backend.status_by_id()`. Entries not in `"running"` state or older than 3600 seconds are evicted and stopped. It then spawns new pods until the pool reaches `POOL_SIZE`. Each `_spawn_warm_container()` call goes through the full `backend.create(..., warm=True)` path with a 30-second timeout; failures are swallowed and logged — the loop tries again on the next tick.

Sources: [services/api/api/warm_pool.py:118-161](services/api/api/warm_pool.py), [services/api/api/warm_pool.py:91-115](services/api/api/warm_pool.py)

### The POOL_EVICT_ON_STARTUP Invariant

When the API restarts after a deploy (new container image, new overlay), any warm pods still running in the cluster were built with the **old** image and overlay refs. If those pods were adopted, the first claims after the deploy would run stale code.

`POOL_EVICT_ON_STARTUP` (enabled by default) guards against this. On startup, `_evict_existing_warm()` calls `backend.recover_warm(POOL_HARNESS)` to list all pods with label `centaur.ai/warm=true`, then stops every one that is not already assigned to a live thread. Only **assigned** sandbox IDs — pulled from `sandbox_sessions WHERE state IN ('running', 'idle', 'error')` — are spared.

```python
# warm_pool.py: start_replenish_loop
assigned = await _get_assigned_sandbox_ids()
if POOL_EVICT_ON_STARTUP:
    evicted = await _evict_existing_warm(assigned)
else:
    recovered = await _recover_warm(assigned)
count = await replenish()  # fill with fresh pods
```

After eviction, `replenish()` immediately fills the pool with freshly-spawned pods that use the current image. The cost is one full cold-start cycle on deploy, not on every user request.

Sources: [services/api/api/warm_pool.py:435-470](services/api/api/warm_pool.py), [services/api/api/warm_pool.py:484-513](services/api/api/warm_pool.py), [services/api/api/warm_pool.py:473-481](services/api/api/warm_pool.py)

---

## Claiming a Warm Pod

`claim_container(thread_key, harness, ...)` is called before any cold spawn is attempted:

1. **Harness match** — if `harness != POOL_HARNESS`, returns `None` immediately.
2. **Backend gate** — if the backend does not support warm pools, returns `None`.
3. **Kubernetes + persona/repo** — warm pods are generic; persona or repo injection requires cold-spawn for the Kubernetes backend, so `claim_container` returns `None` in that case. (Non-Kubernetes backends handle injection via exec after claim.)
4. **Pop** — the first entry in `_pool` is atomically popped under `_pool_lock`.
5. **Liveness check** — `backend.status_by_id()` is called; if the pod is not `"running"`, it is stopped and `None` is returned.
6. **Token refresh** — a fresh sandbox API token is minted for the thread and written into the pod via `exec_run`.
7. **Trace injection** — if a `trace_id` is provided, it is written to `/home/agent/.trace_id`.
8. **Persona/repo injection** — if applicable, `_inject_persona()` clones the repo, copies skills, and assembles the prompt inside the running pod via `exec_run`.
9. **Return** — a `SandboxSession` is constructed with the warm pod's `sandbox_id` bound to the new `thread_key`.

```python
# warm_pool.py:301-385 (condensed)
session = SandboxSession(
    sandbox_id=warm.sandbox_id,
    thread_key=thread_key,
    harness=harness,
    engine=warm.engine,
    started_at=time.time(),
    trace_id=trace_id or "",
)
```

The pool replenish loop will notice the deficit on its next tick and spawn a replacement pod.

Sources: [services/api/api/warm_pool.py:301-385](services/api/api/warm_pool.py)

---

## Sandbox Session State Machine

Once a pod is claimed (warm or cold), its lifecycle is tracked in the `sandbox_sessions` PostgreSQL table. The `state` column drives scheduling, reconciliation, and idle eviction decisions.

```mermaid
stateDiagram-v2
    [*] --> idle : fresh spawn, no in-flight turn
    [*] --> running : resumed with in-flight turn
    idle --> running : turn dispatched (_db_set_inflight_turn)
    running --> delivering : result delivery claimed (atomic UPDATE)
    running --> idle : SSE disconnect, no in-flight turn
    delivering --> idle : delivery complete
    idle --> suspended : idle TTL expired (reconcile_tick)
    running --> suspended : stale running TTL expired
    delivering --> suspended : stale running TTL expired
    idle --> gone : hard stop / pod deleted
    running --> gone : hard stop / pod deleted
    suspended --> [*]
    gone --> [*]
    idle --> released : runtime assignment GC
```

### State Descriptions

| State | Meaning |
|---|---|
| `idle` | Pod is alive and attached; no turn is active. The thread accepts new turns. |
| `running` | A turn is in-flight (`inflight_turn_id` set). Pod is processing. |
| `delivering` | The turn result is being atomically claimed for delivery to a client. |
| `suspended` | Pod has been stopped; session row retained for context continuity. |
| `released` | Runtime assignment (control-plane record) was GC'd; pod no longer active. |
| `gone` | Pod deleted; row may be retained briefly for diagnostics. |
| `error` | Turn ended abnormally; may be retried. |

### Transitions in Code

- **→ running**: `_db_set_inflight_turn()` sets `state = 'running'` atomically with the turn payload. Also set when SSE connects via `_db_update_state(thread_key, "running")`.
- **→ idle**: On SSE disconnect, if no in-flight turn remains, `_db_update_state(thread_key, "idle")` is called.
- **→ delivering**: Added by migration 003; claimed by an atomic `UPDATE ... SET state = 'delivering'` to prevent duplicate delivery.
- **→ released**: The `_release_stale_runtime_assignments()` GC marks `agent_runtime_assignments.state = 'released'` for assignments whose backend pod is no longer `running` or `created`.
- **→ suspended/gone**: `reconcile_tick()` (runs every 60 s) enforces `IDLE_TTL_S` (default 24 h) on idle rows and `inactive_running_ttl` on stuck running/delivering rows.

The constant `_REUSABLE_DB_STATES = {"running", "idle", "delivering", "error", "suspended"}` controls which sessions are considered eligible for reuse when a thread reconnects.

Sources: [services/api/api/agent.py:89](services/api/api/agent.py), [services/api/api/agent.py:238](services/api/api/agent.py), [services/api/api/agent.py:279](services/api/api/agent.py), [services/api/api/agent.py:1042](services/api/api/agent.py), [services/api/api/agent.py:1100-1101](services/api/api/agent.py), [services/api/api/agent.py:1492-1497](services/api/api/agent.py), [services/api/db/migrations/003_add_delivering_state.sql:3-6](services/api/db/migrations/003_add_delivering_state.sql)

---

## Warm vs Cold Spawn: Decision Flow

```text
New thread request for (thread_key, harness)
          │
          ▼
claim_container(thread_key, harness)
          │
    ┌─────┴──────┐
    │  harness   │ ──mismatch──► None
    │  matches?  │
    └─────┬──────┘
          │ match
          ▼
    pool non-empty?
          │ yes                   no
          ▼                       ▼
    status_by_id()          ─────────────────────
    running? ─no─► discard   Cold spawn path:
          │ yes              _evict_idle_for_capacity()
          ▼                  backend.create(thread_key, ...)
    token refresh             (15 s typical)
    trace inject
    persona inject (if any)
          │
          ▼
    SandboxSession returned
```

If `claim_container` returns `None`, the caller falls through to `backend.create()`, which goes through the full pod creation sequence in `KubernetesExecutorBackend.create()`.

Sources: [services/api/api/agent.py:681-750](services/api/api/agent.py), [services/api/api/warm_pool.py:301-385](services/api/api/warm_pool.py)

---

## Failure Modes

| Scenario | Behavior |
|---|---|
| Warm pod dies between replenish and claim | `status_by_id()` returns non-`running`; pod is stopped, `None` returned; caller falls back to cold spawn |
| Token refresh fails on claim | Warning logged; claim continues — stale token will expire on next use |
| `replenish()` spawn fails | Exception swallowed, `warm_container_spawn_failed` logged; next replenish tick retries |
| Pod older than 1 hour | Evicted by `replenish()` health check; replaced by a fresh pod |
| Deploy with `POOL_EVICT_ON_STARTUP=false` | Old pods are adopted — may run stale image if image was bumped. Default is enabled to prevent this. |
| Kubernetes backend + persona or repo requested | `claim_container` returns `None` by design; always cold-spawns to get the correct persona bundle baked in at pod creation |

Sources: [services/api/api/warm_pool.py:118-161](services/api/api/warm_pool.py), [services/api/api/warm_pool.py:311-345](services/api/api/warm_pool.py)

---

## Summary

The warm pool is a process-local in-memory list of pre-booted Kubernetes pods, maintained at `POOL_SIZE` (default 5) by a background asyncio task that runs every 5 seconds. Claiming a warm pod skips the ~15-second cold-start path and instead injects a fresh token and optionally a persona/repo in-place. The `POOL_EVICT_ON_STARTUP` invariant ensures that after any deploy, all pre-existing warm pods are replaced before the first claim, guaranteeing the first user-facing turn always runs the just-deployed code. Once claimed, a sandbox session progresses through the `idle → running → delivering → idle` cycle in PostgreSQL, with `reconcile_tick()` enforcing TTLs and cleaning up orphaned or stale sessions every 60 seconds.

Sources: [services/api/api/warm_pool.py:484-513](services/api/api/warm_pool.py)

---

## 05. Harness Adapters: Amp, Claude Code, Codex, pi-mono

> How the harness_protocol layer normalizes four different agent CLIs into a single event stream. What each adapter does differently (Amp materializes attachments to files; Claude Code passes Anthropic content blocks directly; Codex/pi-mono extract plain text). The _VALID_STDOUT_EVENT_TYPES allowlist as a forward-compatibility boundary.

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/05-harness-adapters-amp-claude-code-codex-pi-mono.md
- Generated: 2026-05-21T23:43:48.129Z

### Source Files

- `services/api/api/sandbox/harness_protocol.py`
- `services/api/api/sandbox/normalize.py`
- `services/api/api/sandbox/prompt_assembly.py`
- `services/api/tests/test_harness_protocol.py`
- `services/api/tests/test_amp_wrapper.py`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/sandbox/harness_protocol.py](services/api/api/sandbox/harness_protocol.py)
- [services/api/api/sandbox/normalize.py](services/api/api/sandbox/normalize.py)
- [services/api/api/sandbox/prompt_assembly.py](services/api/api/sandbox/prompt_assembly.py)
- [services/api/tests/test_harness_protocol.py](services/api/tests/test_harness_protocol.py)
- [services/api/tests/test_amp_wrapper.py](services/api/tests/test_amp_wrapper.py)
- [services/api/api/agent.py](services/api/api/agent.py)
- [services/sandbox/amp-wrapper.py](services/sandbox/amp-wrapper.py)
- [services/sandbox/claude-app-wrapper.py](services/sandbox/claude-app-wrapper.py)
- [services/sandbox/codex-app-wrapper.py](services/sandbox/codex-app-wrapper.py)
- [services/sandbox/harness_adapter.py](services/sandbox/harness_adapter.py)
</details>

# Harness Adapters: Amp, Claude Code, Codex, pi-mono

Centaur supports four distinct agent CLI backends — **Amp**, **Claude Code**, **Codex**, and **pi-mono** — each with its own wire protocol, event vocabulary, and subprocess lifecycle. A two-layer normalization pipeline (`harness_protocol.py` and `normalize.py`) converts all four into a single canonical NDJSON event stream that the rest of the API can consume without knowing which backend is running.

This page explains what each adapter does in isolation, where they share logic, and where they diverge. Understanding this boundary is essential when adding a new event type, debugging a missing turn-done signal, or extending a backend to pass richer content.

---

## Architecture Overview

```text
┌─────────────────────────────────────────────────────────────┐
│ Sandbox container                                           │
│                                                             │
│  ┌──────────────┐   ┌──────────────┐   ┌──────────────┐   │
│  │ amp-wrapper  │   │claude-app-   │   │codex-app-    │   │
│  │    .py       │   │ wrapper.py   │   │ wrapper.py   │   │
│  └──────┬───────┘   └──────┬───────┘   └──────┬───────┘   │
│         │                  │                   │            │
│         └──────────────────┴───────────────────┘           │
│                        NDJSON stdout                        │
└─────────────────────────────────────────────────────────────┘
                             │
                    (harness stdout pipe)
                             │
┌─────────────────────────────────────────────────────────────┐
│ API process  (services/api/)                                │
│                                                             │
│  agent.py  _stream_stdout()                                 │
│    ├── _VALID_STDOUT_EVENT_TYPES allowlist (warn unknown)   │
│    ├── extract_thread_id()   ← harness_protocol.py          │
│    ├── extract_result()      ← harness_protocol.py          │
│    ├── is_turn_done()        ← harness_protocol.py          │
│    └── normalize_harness_event()  ← normalize.py           │
│               ↓                                             │
│        canonical SSE stream → clients                       │
└─────────────────────────────────────────────────────────────┘
```

Sources: [services/api/api/agent.py:43-88](), [services/api/api/agent.py:870-909]()

---

## The harness_protocol Layer

`harness_protocol.py` contains pure functions — no I/O, no globals, no imports from other API modules — that implement the turn lifecycle protocol understood by every engine.

### is_turn_done

Determines when a main-agent turn has ended. The logic is engine-specific:

| Engine | Terminal event |
|--------|---------------|
| `amp`, `claude-code` | `type == "result"` OR `type == "assistant"` with `stop_reason == "end_turn"` and **no** `parent_tool_use_id` |
| `codex` | `type == "turn.completed"` or `"turn.failed"` |
| `pi-mono` | `type == "agent_end"` |
| any | `type == "error"` (except non-terminal amp-wrapper restart notices) |

The subagent carve-out is critical: when `parent_tool_use_id` is set on an `assistant` event, that event comes from a subagent, not the main agent, and must **not** close the turn. Similarly, an `assistant` event that contains only `tool_use` content blocks is mid-flight tooling, not a completed response.

Amp-wrapper restart notices (`"restarting (1/5)"`) are non-terminal errors. The wrapper emits them when it catches a crash and self-heals; the turn is still in progress. Only `"giving up"` in the message text promotes the error to a turn-ending event.

Sources: [services/api/api/sandbox/harness_protocol.py:24-63](), [services/api/tests/test_harness_protocol.py:12-98]()

### extract_result

Extracts the final answer text from whichever event type carries it:

- **amp/claude-code**: prefers `result` field on a `result` event; falls back to the last `text` content block inside an `assistant` event.
- **codex**: reads `item.text` when `item.type` is `agent_message` or `agentMessage` on `item.completed`.
- **pi-mono**: reads the last content block's `text` field from the `message` inside a `message_end` event.
- All engines: a top-level `turn.done` event (synthesized by the API) carries `result` as a string or `{"text": "..."}` dict.

Sources: [services/api/api/sandbox/harness_protocol.py:66-112]()

### extract_thread_id

Maps the engine-specific session identifier to a common `agent_thread_id`:

- **amp/claude-code**: `session_id` on `system/init` or `assistant` events.
- **codex**: `thread_id` on `thread.started`.
- **pi-mono**: `id` on `session`.

Sources: [services/api/api/sandbox/harness_protocol.py:115-128]()

### build_user_input and messages_to_content_blocks

These two functions build the harness-native user-input envelope sent to the subprocess. `build_user_input` wraps a list of content blocks into `{"type":"user","message":{"role":"user","content":[...]}}`, optionally attaching `steer`, `thread_key`, and `trace_id` fields.

`messages_to_content_blocks` flattens a multi-message conversation history into that block list:

- **attachment_ref** parts become `text` blocks containing a `curl` download instruction referencing the Centaur attachments API. This is how attachments are universally conveyed regardless of backend — the agent is told to download the file by ID.
- **assistant** messages are prefixed with `[Your previous response]:` or `[Previous Centaur response]:` (for history backfills from other sessions), so the model can distinguish its own prior output from the current user request.
- The first `text` part of each user message with a `user_id` is prefixed `<@user_id>:` for multi-user attribution.

Sources: [services/api/api/sandbox/harness_protocol.py:131-212](), [services/api/tests/test_harness_protocol.py:276-444]()

---

## The normalize Layer

`normalize.py` is a 1:1 Python port of `packages/harness-events/src/normalize.ts`. It converts raw NDJSON events into canonical dicts that the API's SSE stream can emit to clients.

### Main dispatcher

```python
# services/api/api/sandbox/normalize.py:850-896
def normalize_harness_event(engine: str, event: dict) -> list[dict]:
    normalized = (engine or "").strip().lower()
    if not normalized:
        # auto-detect from event type fingerprints
        ...
    if normalized == "codex":
        return _normalize_codex_event(event)
    if normalized == "pi-mono":
        return _normalize_pi_event(event)
    return _normalize_amp_like_event(event)   # amp, claude-code, and all personas
```

Persona names (`legal`, `eng`, etc.) fall through to the amp-like normalizer because personas are amp-based engines with custom prompts.

The auto-detection heuristic (when `engine` is empty) fingerprints event types: `item.*` and `turn.*` prefixes → codex; `session`, `agent_end`, `tool_execution_*` → pi-mono; everything else → amp.

Sources: [services/api/api/sandbox/normalize.py:847-896]()

---

## Per-Adapter Details

### Amp

**Wrapper:** `services/sandbox/amp-wrapper.py`

Amp runs as `amp --no-ide --no-notifications --dangerously-allow-all --execute --stream-json --stream-json-input --stream-json-thinking --mode <mode>`. The wrapper exists to handle three concerns the raw CLI cannot:

1. **Handoff chaining**: When `follow=true` appears in a `handoff` tool call, the wrapper detects the `newThreadID` in the subsequent tool result, kills the current process, and immediately spawns `amp threads continue <T-new>`. The intermediate events (between handoff and end-turn) are suppressed to avoid duplicate output.

2. **Crash recovery**: Up to 5 restarts (`MAX_CRASH_RESTARTS`). Each transient crash emits `{"type":"error","error":{"message":"amp exited with code N, restarting (K/5)"}}` (which `harness_protocol.is_turn_done` treats as non-terminal). On the sixth crash it emits `"giving up"` which **is** terminal.

3. **Heartbeats**: Before each run attempt the wrapper emits `{"type":"system","subtype":"wrapper_heartbeat","phase":"<startup|crash_restart|handoff_continue|interrupt_continue>"}`. These are lifecycle signals for observability; `_normalize_amp_like_event` filters `system` events that carry no `subagent_id` and no `init` subtype.

**Attachment handling**: Amp receives content blocks directly in the user envelope. The `attachment_ref` → `curl` instruction conversion happens upstream in `messages_to_content_blocks`, so Amp never sees a binary payload — it sees a plain-text download instruction.

**Successful result suppression**: The wrapper suppresses `result` events that are **not** errors before forwarding them to the API. The comment in the source explains the design: the API synthesizes `turn.done` from stream EOF + accumulated text, so forwarding a `result` event would duplicate the final answer. Error results (subtype `error_during_execution` or `is_error=True`) are forwarded so the API can persist a terminal error state rather than waiting indefinitely for EOF.

Sources: [services/sandbox/amp-wrapper.py:236-319](), [services/sandbox/amp-wrapper.py:333-392](), [services/api/tests/test_amp_wrapper.py:48-101]()

**Normalizer** (`_normalize_amp_like_event`): Amp events mostly pass through unchanged. The important transformations:

- `user` events containing `tool_result` content blocks are re-emitted as canonical `{"type":"tool","content":[...]}` events, extracting `tool_use_id` either from the block itself or from `parent_tool_use_id`.
- `system` events with a `subagent_id` (tasks) are translated into `subagent` lifecycle events (`started`, `working`, `completed`, `failed`).
- `stream_event` wrappers (carrying Anthropic streaming API events like `content_block_start`, `content_block_delta`) are unwrapped and emitted as `assistant`/`reasoning` events.
- Transient restart `error` events (containing `"restarting ("` but not `"giving up"`) are **dropped** from the canonical stream (return `[]`).

Sources: [services/api/api/sandbox/normalize.py:237-437]()

---

### Claude Code

**Wrapper:** `services/sandbox/claude-app-wrapper.py`

Claude Code runs as:

```
claude -p --input-format stream-json --output-format stream-json
       --verbose --include-partial-messages
       --dangerously-skip-permissions --permission-mode bypassPermissions
       [--append-system-prompt-file AGENTS.md]
       [--resume <session_id>]
```

The key distinction: the `--input-format stream-json` / `--output-format stream-json` flags mean Claude Code **natively speaks Anthropic content blocks**. The wrapper pipes Centaur's `{"type":"user","message":{...}}` envelopes **straight through** to the subprocess's stdin — no translation, no text extraction. Claude Code handles images, text blocks, and other content types at the model level.

Unlike the amp wrapper, there is **no crash-restart loop** and **no handoff chaining** — those are Amp-specific behaviors. The Claude Code wrapper is thinner: one subprocess, one thread reading stdout, one thread reading stdin. Interrupt handling works by SIGINT-ing the process group; if a turn was active, the wrapper emits a synthetic error result event so the API can transition state.

**Goal rewriting**: Because `claude -p` ignores slash commands, `/goal X` is intercepted by the wrapper and rewritten to `"Set this thread's working goal: X\n\nAcknowledge briefly..."`. This mirrors Codex's `thread/goal/set` RPC parity.

**OTel integration**: The wrapper reads `LMNR_BASE_URL` + `CENTAUR_TRACE_ID` and configures `OTEL_*` environment variables before spawning `claude`, directing telemetry to the same Laminar backend the codex wrapper uses.

**Normalizer**: Claude Code shares `_normalize_amp_like_event` with Amp. The event shapes are identical (both are Anthropic-protocol CLIs), so no separate normalizer path exists.

Sources: [services/sandbox/claude-app-wrapper.py:1-19](), [services/sandbox/claude-app-wrapper.py:262-283](), [services/sandbox/claude-app-wrapper.py:73-87]()

---

### Codex

**Wrapper:** `services/sandbox/codex-app-wrapper.py`

Codex runs as `codex app-server --listen stdio://` — a long-lived JSON-RPC server. The wrapper speaks a two-channel protocol:

- **Requests**: `{"id": N, "method": "...", "params": {...}}` → synchronous reply via matching `id`.
- **Notifications**: `{"method": "...", "params": {...}}` (no `id`) → asynchronous events forwarded to Centaur.

The wrapper calls `initialize`/`initialized` at startup, then translates each Centaur user input into `turn/start` (or `turn/steer` if a turn is active). Codex outputs events under the `turn.*` and `item.*` namespaces.

**Text extraction**: Codex does not accept Anthropic content blocks. The `text_from_blocks` function in the wrapper converts the incoming block list to a plain string:

```python
# services/sandbox/codex-app-wrapper.py:152-164
def text_from_blocks(blocks: list[dict[str, Any]]) -> str:
    for block in blocks:
        btype = block.get("type")
        if btype == "text":
            parts.append(str(block.get("text") or ""))
        elif btype == "image":
            parts.append("[User sent an image attachment; if needed, ask them to upload it as a file reference.]")
        else:
            parts.append(json.dumps(block, ensure_ascii=False))
    return "\n".join(p for p in parts if p).strip()
```

Images become a plain-text advisory; other block types are JSON-serialized. **This is the key divergence from Claude Code**: Codex cannot consume Anthropic image content blocks natively, so they are downgraded to text.

**Normalizer** (`_normalize_codex_event`): Codex events use dotted namespaces (`item.started`, `item.completed`, `turn.completed`). Most pass through unchanged. Key transformations:

- `thread.started` → `{"type":"system","subtype":"init","session_id":"<thread_id>"}`.
- `turn.completed` → a usage-metadata-only event (no text content; the API synthesizes `turn.done`).
- `turn.failed` → `{"type":"error","error":"..."}`.
- `item.completed` for `agent_message`/`agentMessage` items → **dropped** (to avoid duplicate output; the text was already streamed via `item.agentMessage.delta`).
- Tool calls (`mcp_tool_call`, `tool_call`, `function_call`, etc.) on `item.started` → canonical `assistant/tool_use`; on `item.completed` → canonical `tool/content`.
- `subagent` tool calls (where `tool_name == "subagent"`) are translated into `subagent` lifecycle events.

Sources: [services/sandbox/codex-app-wrapper.py:152-172](), [services/api/api/sandbox/normalize.py:613-665](), [services/api/api/sandbox/normalize.py:485-610]()

---

### pi-mono

There is no standalone pi-mono wrapper script in this repository. The `pi-mono` normalizer path is registered as a recognized engine in `normalize.py` and `harness_protocol.py`, and the event shapes it handles indicate a long-lived session protocol with its own message lifecycle.

**Event shapes**: pi-mono uses `session`, `agent_start`, `agent_end`, `message_start`, `message_update`, `message_end`, `tool_execution_start`, `tool_execution_update`, `tool_execution_end`.

**Text extraction**: Like Codex, pi-mono does not use Anthropic content blocks for input. `_normalize_pi_event` reconstructs tool-use and text events from pi-mono's own schema:

- `tool_execution_start` → `assistant/tool_use` (or `subagent/started` for `toolName == "subagent"`).
- `tool_execution_end` → `tool/result` (or `subagent/completed|failed`).
- `message_end` for assistant messages → the content blocks are normalized through `_normalize_pi_message_content`, which reads `block.type` in (`text`, `thinking`, `tool_call`/`toolcall`) and maps to canonical `assistant`, `reasoning`, or `assistant/tool_use` events.

Sources: [services/api/api/sandbox/normalize.py:698-840](), [services/api/api/sandbox/normalize.py:847-896]()

---

## The _VALID_STDOUT_EVENT_TYPES Allowlist

```python
# services/api/api/agent.py:43-86
_VALID_STDOUT_EVENT_TYPES = frozenset({
    "amp_raw_event", "assistant", "command_execution",
    "content_block_delta", "content_block_start", "content_block_stop",
    "error", "file_change",
    "message_delta", "message_start", "message_stop",
    "item.agentMessage.delta", "item.commandExecution.outputDelta",
    "item.completed", "item.fileChange.outputDelta",
    "item.fileChange.patchUpdated", "item.plan.delta",
    "item.reasoning.summaryPartAdded", "item.reasoning.summaryTextDelta",
    "item.reasoning.textDelta", "item.started", "item.updated",
    "reasoning", "result", "status", "subagent", "system",
    "thread.goal.cleared", "thread.goal.updated", "thread.started",
    "tool", "tool_result", "tool_use",
    "turn.done", "turn.completed", "turn.failed",
    "turn.plan.updated", "turn.started",
    "usage", "user",
})
```

This set is **not** a hard filter — unknown event types are not dropped. Instead, when `_stream_stdout` in `agent.py` encounters an event type absent from the set, it logs a `stdout_unknown_event_type` warning at the `warning` level and continues processing normally.

The design consequence is intentional: the allowlist acts as a **forward-compatibility boundary**. When any of the four backend CLIs ships a new event type, the API logs a warning so the operator knows normalization may be incomplete, but the raw event still flows through the `normalize_harness_event` → SSE path. This prevents silent breakage while surfacing gaps for incremental updates to the normalizer.

To add proper support for a new event type:
1. Add the type to `_VALID_STDOUT_EVENT_TYPES` to silence the warning.
2. Add handling in the appropriate `_normalize_*_event` function in `normalize.py`.
3. Add `is_turn_done`, `extract_result`, or `extract_thread_id` cases in `harness_protocol.py` if the new event carries lifecycle semantics.

Sources: [services/api/api/agent.py:43-86](), [services/api/api/agent.py:875-882]()

---

## Stable Tool Call IDs

When a backend does not supply a stable tool call ID (common in Codex where IDs may be missing or positional), the normalizer generates one deterministically:

```python
# services/api/api/sandbox/normalize.py:59-66
def _stable_tool_call_id(name: str, tool_input: Any, nonce: str = "") -> str:
    payload = {"input": tool_input or {}, "name": name or "tool", "nonce": nonce or ""}
    h = hashlib.sha1(_stable_sorted_json(payload).encode()).hexdigest()[:12]
    return f"tool-call-{h}"
```

The `nonce` is drawn from whatever positional or temporal identifier is available (`index`, `position`, `ordinal`, `event_seq`, `timestamp`, `created_at`). This ensures that `tool_result` events can be correlated to `tool_use` events even when the backend generates no IDs, at the cost of collisions if two calls to the same tool with the same input occur in the same position.

Sources: [services/api/api/sandbox/normalize.py:52-66](), [services/api/api/sandbox/normalize.py:463-482]()

---

## Summary

The `harness_protocol` + `normalize` tandem decouples Centaur's API from the idiosyncratic wire protocols of four different agent CLIs. **Amp** and **Claude Code** share a normalizer path because both are Anthropic-protocol CLIs; Claude Code passes content blocks directly while Amp adds crash recovery, handoff chaining, and result suppression. **Codex** and **pi-mono** each have their own normalizer path and both perform text extraction from content blocks at the wrapper boundary. The `_VALID_STDOUT_EVENT_TYPES` frozenset acts as a versioned changelog boundary: a type appearing in it signals that the normalizer fully handles it, while an unrecognized type generates a warning rather than a crash, keeping the system forward-compatible as CLI backends evolve.

Sources: [services/api/api/sandbox/normalize.py:847-896](), [services/api/api/agent.py:875-882]()

---

## 06. Tool Plugin Model: Discovery, Secrets, & the centaur_sdk

> How tools are discovered (Python files in tools/ or overlays), loaded by tool_manager.py, and exposed to sandboxes. The ToolContext / secret() resolution chain (tool context → pluggable backend → default). The SecretMode enum (replace vs inject). How tool authors import centaur_sdk and never see raw credentials.

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/06-tool-plugin-model-discovery-secrets-the-centaur_sdk.md
- Generated: 2026-05-21T23:45:21.248Z

### Source Files

- `services/api/api/tool_manager.py`
- `centaur_sdk/tool_sdk.py`
- `centaur_sdk/backends/base.py`
- `centaur_sdk/backends/registry.py`
- `centaur_sdk/tests/test_tool_sdk.py`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/tool_manager.py](services/api/api/tool_manager.py)
- [centaur_sdk/tool_sdk.py](centaur_sdk/tool_sdk.py)
- [centaur_sdk/__init__.py](centaur_sdk/__init__.py)
- [centaur_sdk/backends/base.py](centaur_sdk/backends/base.py)
- [centaur_sdk/backends/registry.py](centaur_sdk/backends/registry.py)
- [centaur_sdk/backends/stub.py](centaur_sdk/backends/stub.py)
- [centaur_sdk/backends/env.py](centaur_sdk/backends/env.py)
- [centaur_sdk/tests/test_tool_sdk.py](centaur_sdk/tests/test_tool_sdk.py)
- [tools/research/harmonic/client.py](tools/research/harmonic/client.py)
- [tools/research/harmonic/pyproject.toml](tools/research/harmonic/pyproject.toml)
</details>

# Tool Plugin Model: Discovery, Secrets, & the centaur_sdk

Centaur's tool plugin model lets external contributors add new capabilities by dropping a Python package into a `tools/` directory (or a configured overlay). The API service discovers, loads, and registers these tools at startup — and hot-reloads them on demand — without any manual wiring. The other half of the model is credential isolation: tool code never sees raw API keys. Instead it calls `secret()` from `centaur_sdk`, receives a placeholder token, and iron-proxy swaps that token for the real credential on the outbound wire. This page covers how those two halves fit together.

---

## Tool Discovery

### Directory Layout

`ToolManager` is initialized with one or more `tools_dirs`. Each base directory is scanned for Python packages at load time. The scanner supports one level of **category subdirectories**: if a child folder has no `pyproject.toml` it is treated as a category folder and its children are each treated as a tool candidate (e.g. `tools/research/harmonic/`).

```
tools/
  slack/                  ← top-level tool (has pyproject.toml)
  research/               ← category folder (no pyproject.toml)
    harmonic/             ← tool inside category
    crunchbase/
```

Hidden and underscore-prefixed directories (`.foo`, `_bar`) are skipped. When the same tool name appears in two different `tools_dirs`, the later one silently shadows the earlier one — this is the intended mechanism for private overlays.

Sources: [services/api/api/tool_manager.py:1376-1485]()

### pyproject.toml Manifest

Every tool package declares itself via a standard `pyproject.toml`. The `[project]` table supplies `name` (used as the tool's identifier) and `description`. The `[tool.centaur]` table carries runtime metadata:

| Key | Meaning |
|-----|---------|
| `module` | Python file to import (default: `client.py`) |
| `secrets` | Required credentials (tool unavailable if missing) |
| `optional_secrets` | Credentials used when present but not gating availability |
| `hosts` | Tool-level fallback host scope for secret entries |
| `timeout_s` / `timeout_env` | Per-tool call timeout override |
| `type = "persona"` | Marks the package as a persona, not a callable tool |

Example from `tools/research/harmonic/pyproject.toml`:

```toml
[tool.centaur]
module = "client.py"
secrets = [{type = "http", name = "HARMONIC_API_KEY", match_headers = ["apikey"], hosts = ["api.harmonic.ai"]}]
```

Sources: [tools/research/harmonic/pyproject.toml:16-18]()

### Module Loading

`ToolManager._load_tool()` registers the tool directory as a synthetic Python package under the `shared.tools_runtime.<name>` namespace, then imports the configured `module` file using `importlib.util`. This allows relative imports within a tool package to work correctly.

The key sequence during loading:

1. A bare `ToolContext(name=name, secrets={})` is created and pushed onto a `ContextVar` via `set_tool_context(ctx)`.
2. The module is executed via `spec.loader.exec_module(module)` while that context is live.
3. `_collect_methods(module)` calls the module's `_client()` factory once, then enumerates every public, non-lifecycle, non-property method on the returned instance.
4. `reset_tool_context(token)` pops the context whether or not loading succeeded.

The `_client()` pattern is the required convention: each tool module must expose this callable factory. Methods named `close`, `connect`, `disconnect`, or `shutdown` are excluded from the exposed method list (`_LIFECYCLE_METHODS`).

Sources: [services/api/api/tool_manager.py:1663-1763]()

---

## ToolContext and the secret() Resolution Chain

### ToolContext Dataclass

`ToolContext` is a lightweight dataclass defined in `centaur_sdk/tool_sdk.py`:

```python
@dataclass
class ToolContext:
    name: str
    secrets: dict[str, str] = field(default_factory=dict)
    thread_key: str | None = None
    container_id: str | None = None
```

It is stored in a `ContextVar[ToolContext]` named `_tool_ctx`, which makes it coroutine- and thread-safe: each invocation can carry its own context without interference.

Sources: [centaur_sdk/tool_sdk.py:19-27]()

### secret() Resolution Order

When tool code calls `secret("SOME_KEY")`, the function walks three sources in strict priority order:

```
1. ToolContext.secrets dict  (set by ToolManager at call time)
        ↓ miss
2. Pluggable backend          (SecretBackend from registry)
        ↓ miss
3. Caller-supplied default    (or KeyError with tool name included)
```

```python
def secret(key: str, default: str | None = None) -> str:
    try:
        ctx = _tool_ctx.get()
        val = ctx.secrets.get(key)
        if val is not None:
            return val
    except LookupError:
        pass

    from centaur_sdk.backends.registry import get_backend
    val = get_backend().get_sync(key)
    if val is not None:
        return val

    if default is not None:
        return default

    raise KeyError(f"Missing secret '{key}'{ctx_name}")
```

Sources: [centaur_sdk/tool_sdk.py:47-76]()

The test suite verifies each tier explicitly: tool context wins over backend, backend wins over default, and all-miss raises a `KeyError` that includes the tool name for easy diagnosis.

Sources: [centaur_sdk/tests/test_tool_sdk.py:27-64]()

### How ToolManager Populates ToolContext at Call Time

When a tool method is invoked, `call_tool()` (or `call_tool_raw()`) resolves placeholder values for all `replace`-mode `HttpSecret` entries declared by the tool, then constructs a fresh `ToolContext` that includes those placeholders plus any sandbox claims:

```python
resolved = await _resolve_secrets(all_secrets)
ctx = ToolContext(
    name=lt.name,
    secrets={**lt.ctx.secrets, **resolved},
    thread_key=sandbox_claims.get("thread_key") if sandbox_claims else None,
    container_id=sandbox_claims.get("container_id") if sandbox_claims else None,
)
token = set_tool_context(ctx)
```

`_resolve_secrets` only returns values for replace-mode `HttpSecret` entries (the placeholder token). Inject-mode, `GcpAuthSecret`, `OAuthTokenSecret`, and `PgDsnSecret` are invisible to tool code — those are handled entirely by iron-proxy.

Sources: [services/api/api/tool_manager.py:838-850](), [services/api/api/tool_manager.py:1885-1915]()

---

## The SecretMode Enum: replace vs inject

`SecretMode` is a two-value enum that controls how iron-proxy applies an HTTP credential:

| Mode | What the tool sees | What iron-proxy does |
|------|--------------------|----------------------|
| `REPLACE` | The placeholder token (e.g. `"HARMONIC_API_KEY"`) written into a header/path/query | Scans the outbound request for the token and swaps it for the real value from `secret_ref` |
| `INJECT` | Nothing — the credential is never in the tool's scope | Adds the resolved credential directly to the request (header or query param) before it leaves the network |

```python
class SecretMode(str, Enum):
    REPLACE = "replace"
    INJECT = "inject"
```

Sources: [services/api/api/tool_manager.py:65-73]()

The `HttpSecret` dataclass enforces that each mode declares the right fields: replace-mode requires at least one of `match_headers`, `match_path`, or `match_query`; inject-mode requires exactly one of `inject_header` or `inject_query_param`, and `inject_formatter` (a Go template like `Bearer {{ .Value }}`) is only valid alongside `inject_header`. Cross-mode fields are rejected at parse time.

Sources: [services/api/api/tool_manager.py:560-622]()

---

## Secret Types Beyond HTTP

`SecretDef` is a union of five types, each handled completely by iron-proxy with no raw credential reaching the sandbox:

| Type | `SecretDef` class | How it works |
|------|--------------------|--------------|
| HTTP header/query/path | `HttpSecret` | replace or inject mode as above |
| GCP service-account | `GcpAuthSecret` | iron-proxy mints OAuth2 tokens from a keyfile |
| OAuth2 token exchange | `OAuthTokenSecret` | iron-proxy runs the grant flow (refresh_token, client_credentials, password, or jwt_bearer) and injects Bearer tokens |
| Postgres DSN | `PgDsnSecret` | iron-proxy exposes a local TCP listener; sandbox connects to that, iron-proxy forwards to the upstream DSN |
| Per-request HMAC | `HmacSignSecret` | iron-proxy computes an HMAC signature and adds the resulting headers |

`_parse_secret()` in `tool_manager.py` maps each `type` key in `pyproject.toml` to the appropriate dataclass, validating required and optional fields with descriptive `ValueError` messages.

Sources: [services/api/api/tool_manager.py:647-820]()

---

## The Pluggable Backend System

### SecretBackend ABC

`centaur_sdk.backends.base.SecretBackend` is the abstract interface all backends implement:

```python
class SecretBackend(ABC):
    @abstractmethod
    async def get(self, key: str) -> str | None: ...

    @abstractmethod
    async def list_keys(self) -> list[str]: ...

    def get_sync(self, key: str) -> str | None:
        # runs in a background thread when called from an event loop
```

`get_sync` bridges sync tool code and async backends without blocking the event loop: it uses `asyncio.run()` outside a loop and `ThreadPoolExecutor` inside one.

Sources: [centaur_sdk/backends/base.py:9-34]()

### Built-in Backends

| Class | Module | Behavior |
|-------|--------|----------|
| `StubBackend` | `centaur_sdk.backends.stub` | Returns the key name as the value. Server-mode default. Falls back to env var if key is present (needed for Postgres DSNs that cannot go through firewall injection). |
| `EnvBackend` | `centaur_sdk.backends.env` | Returns `os.environ.get(key)`. CLI mode only — **banned in server mode**. |

Sources: [centaur_sdk/backends/stub.py:17-33](), [centaur_sdk/backends/env.py:10-17]()

### Registry

`centaur_sdk.backends.registry` holds a module-level singleton `_backend`. On first use, `get_backend()` calls `auto_configure()`, which installs `StubBackend`. This is the invariant that prevents real credentials from ever being resolvable inside the API process:

> **Do not use `EnvBackend` here.** Real secrets must never be resolvable inside the API process. See README.md § Security Architecture, invariant S1.

Sources: [centaur_sdk/backends/registry.py:24-46]()

CLI tools explicitly call `registry.configure(EnvBackend())` before using `secret()` so local development works without a running iron-proxy.

---

## Infra-Level Secrets

`ToolManager` maintains a hardcoded `_INFRA_SECRETS` class variable for first-party AI provider keys (Anthropic, OpenAI, xAI, Gemini, AMP, GitHub, Slack). These are always included in the secrets map handed to iron-proxy via `collect_secrets()`, ahead of any tool-declared secrets.

```python
_INFRA_SECRETS: ClassVar[list[HttpSecret]] = [
    HttpSecret(
        name="ANTHROPIC_API_KEY",
        secret_ref="ANTHROPIC_API_KEY",
        hosts=("api.anthropic.com",),
        match_headers=("X-Api-Key",),
    ),
    ...
]
```

Sources: [services/api/api/tool_manager.py:1594-1648]()

---

## End-to-End Flow

```
┌─────────────────────────────────────────────────────────────┐
│  tools/research/harmonic/                                   │
│    pyproject.toml  →  [tool.centaur] secrets = [...]        │
│    client.py       →  from centaur_sdk import secret        │
│                       def _client() -> HarmonicClient: ...  │
└────────────────────┬────────────────────────────────────────┘
                     │ ToolManager.discover()
                     ▼
┌─────────────────────────────────────────────────────────────┐
│  ToolManager._load_tool()                                   │
│    1. parse secrets from pyproject.toml                     │
│    2. set_tool_context(ToolContext(name, secrets={}))        │
│    3. importlib exec_module(client.py)                      │
│    4. _collect_methods(_client())   → LoadedTool            │
│    5. reset_tool_context(token)                             │
└────────────────────┬────────────────────────────────────────┘
                     │ HTTP POST /tools/harmonic/search
                     ▼
┌─────────────────────────────────────────────────────────────┐
│  ToolManager.call_tool()                                    │
│    1. _resolve_secrets() → {"HARMONIC_API_KEY": "HARMONIC_API_KEY"} │
│    2. set_tool_context(ToolContext(secrets=resolved, ...))  │
│    3. method.fn(**args) under asyncio.wait_for(timeout)     │
│    4. reset_tool_context(token)                             │
└────────────────────┬────────────────────────────────────────┘
                     │ outbound HTTPS to api.harmonic.ai
                     ▼
┌─────────────────────────────────────────────────────────────┐
│  iron-proxy (firewall)                                      │
│    sees "apikey: HARMONIC_API_KEY" in request header        │
│    resolves HARMONIC_API_KEY from env/1Password             │
│    replaces header with real credential before forwarding   │
└─────────────────────────────────────────────────────────────┘
```

The invariant is: tool code receives a string (`"HARMONIC_API_KEY"`) that is useless outside the iron-proxy perimeter. The real key is never present in the Python process.

---

## What Tool Authors Import

The `centaur_sdk` package surface is intentionally minimal. Everything a tool author needs is exported from `centaur_sdk/__init__.py`:

| Symbol | Purpose |
|--------|---------|
| `secret(key, default?)` | Resolve a credential via the three-tier chain |
| `current_thread_key()` | Get the active sandbox thread key |
| `save_attachment(...)` | Persist binary output scoped to the current thread |
| `save_attachment_from_path(...)` | Persist a local file as a thread attachment |
| `ToolContext` | Dataclass; authors read it via `get_tool_context()` but rarely construct it |
| `Table`, `render_text_table` | CLI formatting helpers |

Sources: [centaur_sdk/__init__.py:12-34]()

Tool code never imports `ToolManager`, never sets `ToolContext`, and never touches the backend registry. The call to `set_tool_context` / `reset_tool_context` is the exclusive responsibility of `ToolManager`.

---

## Failure Modes and Invariants

| Scenario | Behavior |
|----------|----------|
| Tool's `module` file missing | Logged as `tool_module_missing`, tool skipped; `ToolManager.load_failures` records it |
| `secret()` called before any context set | Falls through to backend (StubBackend in server mode), which returns the key name as the value — a safe stub, not a crash |
| Required secret unresolvable | `GET /tools` omits the tool from the listing; `GET /tools/<name>` returns HTTP 404 |
| Catch-all or IP host patterns in secrets | `tool_invalid_host` warning logged; tool still loads but the misconfigured secret is flagged |
| Tool shadows another in a later overlay dir | `tool_shadowed` logged; later entry wins silently |
| Tool method times out | `asyncio.TimeoutError` caught; returns `{"error": "Tool call timed out after Xs"}` |
| `EnvBackend` installed in server mode | Ruff per-file-ignore rule in pyproject.toml prevents the import; policy enforced at lint time, not runtime |

The overall design ensures that a misbehaving tool — including one that leaks its own `ToolContext` — can only ever leak a placeholder string that iron-proxy would reject from an unscoped origin.

Sources: [centaur_sdk/backends/registry.py:24-37](), [services/api/api/tool_manager.py:1540-1589]()

---

## 07. Durable Workflow Engine: Checkpoint/Replay

> The checkpoint/replay model inspired by Cloudflare Workflows: the handler function IS the workflow, ctx.step() calls are discovered at runtime, and Postgres checkpoints each step result. On resume after a crash the handler re-executes top-to-bottom but skips already-checkpointed steps instantly. How this enables sleep, suspension, child agents, and idempotent re-runs.

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/07-durable-workflow-engine-checkpoint-replay.md
- Generated: 2026-05-21T23:46:21.163Z

### Source Files

- `services/api/api/workflow_engine.py`
- `workflows/paradigm_pulse_daily.py`
- `workflows/slack_sync.py`
- `services/api/tests/test_workflow_engine_title.py`
- `services/api/tests/test_workflow_idempotency_unit.py`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/workflow_engine.py](services/api/api/workflow_engine.py)
- [workflows/paradigm_pulse_daily.py](workflows/paradigm_pulse_daily.py)
- [workflows/slack_sync.py](workflows/slack_sync.py)
- [services/api/tests/test_workflow_engine_title.py](services/api/tests/test_workflow_engine_title.py)
- [services/api/tests/test_workflow_idempotency_unit.py](services/api/tests/test_workflow_idempotency_unit.py)
</details>

# Durable Workflow Engine: Checkpoint/Replay

Centaur's workflow engine implements a **checkpoint/replay** execution model inspired by Cloudflare Workflows. The central idea is that a plain Python `async` function *is* the workflow: there is no separate DAG definition, no step registry, and no declarative schema. Steps are discovered at runtime when the handler calls `ctx.step()`. After each step completes, the engine persists its result to Postgres. If the process crashes or suspends, the handler is simply called again from the top — but every step that already has a Postgres checkpoint is returned instantly from the cache, making the re-execution functionally transparent.

This design means that dynamic branches, loops, and conditionals work naturally because they are just Python control flow. Sleeping for twelve hours, waiting for an external event, spawning a child agent, and retrying a flaky API call are all expressed as ordinary awaits inside the same handler function.

---

## The Handler-as-Workflow Contract

Handler discovery follows the same convention as Centaur tools. Python files placed in the `api/workflows/` built-in package or any directory listed in the `WORKFLOW_DIRS` environment variable are scanned at startup. A valid handler module must export:

- `WORKFLOW_NAME: str` — the canonical workflow name used as a lookup key.
- `async def handler(params, ctx: WorkflowContext)` — the entry point.

Optionally, a module may export:

- `Input` — a `@dataclass` that the raw `input_json` dict is auto-coerced into before the handler is called.
- `SCHEDULE` / `CRON` / `INTERVAL` — scheduling metadata picked up by the schedule engine.

A shorthand auto-handler is also synthesized when a module exports `PROMPT` and `SLACK_CHANNEL` but no explicit `handler`: the engine wraps `ctx.agent_turn(PROMPT)` followed by `ctx.post_to_slack(channel, result_text)` automatically.

Sources: [services/api/api/workflow_engine.py:1429-1486]()

The `paradigm_pulse_daily` workflow illustrates the full-handler pattern:

```python
# workflows/paradigm_pulse_daily.py
WORKFLOW_NAME = "paradigm_pulse_daily"
CRON = "45 7 * * *"
SLACK_CHANNEL = "paradigm-pulse"

async def handler(inp: dict[str, Any], ctx: WorkflowContext) -> dict[str, Any]:
    result = await ctx.agent_turn(PROMPT)
    text = str(result.get("result_text") or "").strip()
    ...
    await ctx.call_tool("slack", "send_message", args)
    return result
```

Sources: [workflows/paradigm_pulse_daily.py:15-228]()

---

## WorkflowContext: The Checkpoint Primitives

Every handler receives a `WorkflowContext` (`ctx`) that exposes three core primitives and several higher-level helpers. All of them share the same underlying invariant: *execute at most once, return the cached result on replay*.

### `ctx.step(name, fn)`

The fundamental primitive. On the first execution it calls `fn()`, awaits the result, writes it to `workflow_checkpoints`, and returns the value. On any subsequent execution (replay after crash, retry, or resume after suspension), the step name is looked up in the pre-loaded checkpoint dict and the cached value is returned instantly — `fn` is never called again.

```python
# services/api/api/workflow_engine.py (simplified)
async def step(self, name, fn, *, retry=None, timeout=None, ...):
    checkpoint_name = self._resolve_name(name)
    if checkpoint_name in self._checkpoints:
        return self._checkpoints[checkpoint_name]   # instant cache hit
    self._in_replay = False
    result = await self._call_step_fn(fn)           # only runs once
    await self._persist_checkpoint(checkpoint_name, result, ...)
    return result
```

Sources: [services/api/api/workflow_engine.py:330-398]()

Step names inside loops are automatically deduplicated using a counter: `name`, `name#2`, `name#3`, and so on, so a `for channel in channels` loop produces stable, ordered checkpoints without any manual naming.

Sources: [services/api/api/workflow_engine.py:251-262]()

### `ctx.sleep(name, duration)` and `ctx.sleep_until(name, when)`

Sleep writes a wake-time ISO timestamp as the checkpoint state, then raises `SuspendWorkflow` with `available_at` set to the future wake time. The worker sets the run's status to `sleeping` and stops. When the schedule reconciler notices `available_at <= now`, it re-queues the run. The handler re-executes top-to-bottom: when it reaches the sleep call again, the checkpoint already exists and the engine checks whether the wake time has passed. If it has, the call returns normally (no-op); if not, it re-raises `SuspendWorkflow`.

```python
async def sleep(self, name, duration):
    checkpoint_name = self._resolve_name(name)
    if checkpoint_name in self._checkpoints:
        wake_at = dt.datetime.fromisoformat(self._checkpoints[checkpoint_name])
        if dt.datetime.now(dt.timezone.utc) < wake_at:
            raise SuspendWorkflow(available_at=wake_at)
        return  # wake time passed — fall through
    wake_at = dt.datetime.now(dt.timezone.utc) + duration
    await self._persist_checkpoint(checkpoint_name, wake_at.isoformat(), step_kind="sleep")
    raise SuspendWorkflow(available_at=wake_at)
```

Sources: [services/api/api/workflow_engine.py:400-416]()

### `ctx.wait_for_event(name, event_type, correlation_id)`

When the event has not arrived yet, the engine writes a `_waiting` marker dict as the checkpoint (step_kind `event_wait`) and raises `SuspendWorkflow` with `status="waiting"`. On resume, the engine queries `workflow_events` for a matching `(event_type, correlation_id)` row. If found, the real payload replaces the wait-marker checkpoint and the handler continues. If not found, `SuspendWorkflow` is raised again. An optional `timeout` stores a deadline ISO string in the marker; when the deadline is exceeded, `TimeoutError` is raised.

Sources: [services/api/api/workflow_engine.py:452-543]()

---

## Checkpoint Persistence and Lease Fencing

`_persist_checkpoint` writes atomically inside a Postgres transaction that also extends the worker's lease. Before the write, it acquires a `FOR UPDATE` lock on the `workflow_runs` row and verifies that:

1. The run still exists.
2. The run is not `cancelled`.
3. The `worker_id` still matches the current worker.

If any check fails, `CancelledWorkflow` is raised, which causes the worker to abort cleanly without corrupting state. This fencing prevents two workers from racing on the same run after a lease expiry.

```python
# services/api/api/workflow_engine.py
async def _persist_checkpoint(self, checkpoint_name, value, ...):
    async with self._pool.acquire() as conn:
        async with conn.transaction():
            row = await conn.fetchrow(
                "SELECT status, worker_id FROM workflow_runs "
                "WHERE run_id = $1 FOR UPDATE", self.run_id,
            )
            if not row or row["status"] == "cancelled" or row["worker_id"] != self._worker_id:
                raise CancelledWorkflow()
            await conn.execute(
                "INSERT INTO workflow_checkpoints ... ON CONFLICT ... DO UPDATE SET state = EXCLUDED.state",
                ...
            )
            await conn.execute(
                "UPDATE workflow_runs SET worker_lease_expires_at = NOW() + ...", ...
            )
    self._checkpoints[checkpoint_name] = value
```

Sources: [services/api/api/workflow_engine.py:264-318]()

The in-memory `self._checkpoints` dict is updated after each successful write so that subsequent `step` calls during the same execution do not hit the database unnecessarily.

---

## Suspension and Resume Lifecycle

```text
┌─────────────────────────────────────────────────────────────┐
│  workflow_runs.status transitions                           │
│                                                             │
│  queued ──► running ──► completed                          │
│                │                                            │
│                ├──► sleeping  (ctx.sleep / ctx.sleep_until) │
│                │       │                                    │
│                │       └── available_at <= now ──► queued   │
│                │                                            │
│                ├──► waiting   (ctx.wait_for_event /         │
│                │               ctx.wait_for_workflow)       │
│                │       │                                    │
│                │       └── event/child arrives ──► queued   │
│                │                                            │
│                └──► failed_permanent / cancelled            │
└─────────────────────────────────────────────────────────────┘
```

The engine's reconcile loop (`_tick_once`) has two jobs:

1. **Re-queue expired leases**: any run with `status = 'running'` and an expired `worker_lease_expires_at` is reset to `queued`, making it available for another worker to claim.
2. **Wake sleeping/waiting runs**: runs whose `available_at <= NOW()` are promoted back to `queued`.

Expired-lease re-queueing ensures crash recovery: if a worker dies mid-step, the lease expires after `WORKFLOW_WORKER_LEASE_S` seconds (default 30s) and the run is automatically retried from its last checkpoint.

Sources: [services/api/api/workflow_engine.py:1068-1082](), [services/api/api/workflow_engine.py:153-184]()

---

## Idempotency on Run Creation

When `create_workflow_run` is called with a `trigger_key`, the engine takes a `pg_advisory_xact_lock` on `(workflow_name, trigger_key)` and checks for an existing row. If the row exists with a matching `request_hash`, the existing `run_id` is returned without creating a duplicate — the call is fully idempotent. If the row exists but the hash differs (different payload for the same key), a `ControlPlaneError(IDEMPOTENCY_PAYLOAD_MISMATCH)` is raised so callers know they have a logic error rather than silently ignoring the conflict.

Sources: [services/api/api/workflow_engine.py:1762-1841]()

The `test_workflow_idempotency_unit.py` suite explicitly verifies that `_insert_workflow_run` raises `IDEMPOTENCY_PAYLOAD_MISMATCH` when the `request_hash` of the new input does not match the stored hash, and that the warning log includes safe metadata (hash prefix, key list, run_id) without leaking the full payload.

Sources: [services/api/tests/test_workflow_idempotency_unit.py:24-69]()

---

## Child Workflows and Agent Turns

`ctx.start_workflow` / `ctx.wait_for_workflow` / `ctx.run_workflow` are the primitives for spawning sub-workflows. `start_workflow` wraps `create_workflow_run` inside a `ctx.step` (step_kind `child_workflow_start`) so the child `run_id` is checkpointed. `wait_for_workflow` stores a `_waiting` marker with the child `run_id` and suspends. When the child reaches a terminal state, the parent is re-queued, re-executes top-to-bottom, and at the `wait_for_workflow` call finds the checkpoint still in `_waiting` state but the child now terminal — it replaces the marker with the child's result and continues.

`ctx.run_agent` is a typed shorthand for `run_workflow("agent_turn", ...)`. It creates a child `agent_turn` workflow run, suspends until it completes, and returns `result_text`.

Sources: [services/api/api/workflow_engine.py:545-695](), [services/api/api/workflow_engine.py:738-779]()

---

## Per-Step Retry and Timeout

`ctx.step` accepts an optional `RetryPolicy` that controls how many times `fn` is re-attempted on failure within the same execution, along with fixed or exponential backoff. Raising `NonRetryableError` inside `fn` bypasses the retry loop and propagates immediately. An optional `timeout: dt.timedelta` wraps `fn` in `asyncio.wait_for`. A successful result — even on a later attempt — is checkpointed exactly once, making the step idempotent across external calls.

```python
@dataclass
class RetryPolicy:
    limit: int = 0
    delay: dt.timedelta = field(default_factory=dt.timedelta)
    backoff: str = "fixed"   # "fixed" | "exponential"
    max_delay: dt.timedelta | None = None
```

Sources: [services/api/api/workflow_engine.py:131-148](), [services/api/api/workflow_engine.py:358-398]()

---

## Replay-Safe Logging

`ctx.log(msg, **kwargs)` suppresses structured log lines during replay by checking the `_in_replay` flag. The flag is `True` when the context is first constructed with a non-empty checkpoints dict and is cleared to `False` the moment the first un-checkpointed step executes. This prevents duplicate log lines — an important ergonomic property since the handler may replay dozens of already-completed steps before reaching live work.

Sources: [services/api/api/workflow_engine.py:437-450]()

---

## Workflow Discovery and the External WORKFLOW_DIRS Mount

Built-in workflows live under `api/workflows/`. The `WORKFLOW_DIRS` env var accepts colon-separated filesystem paths; Python files found there are loaded into the `centaur.workflows.*` namespace. This lets operator-supplied workflows be bind-mounted into the container and hot-reloaded without rebuilding the image — the same pattern used for external tool plugins. Each handler is versioned by the SHA-256 of its source file, stored in `workflow_runs.workflow_version` for reproducibility.

Sources: [services/api/api/workflow_engine.py:1356-1532]()

---

## End-to-End Example: `paradigm_pulse_daily`

The daily digest workflow is the clearest real-world example. Its handler takes two steps:

1. `ctx.agent_turn(PROMPT)` — runs a child `agent_turn` workflow, suspending the parent until the AI completes.
2. `ctx.call_tool("slack", "send_message", args)` — posts the result to `#paradigm-pulse`, checkpointed as step_kind `tool_call` so the message is sent exactly once even if the workflow replays.

Because both operations are checkpointed, a crash between step 1 and step 2 causes the handler to replay, find the agent result in the checkpoint cache, and proceed directly to posting — without re-running the expensive AI turn.

Sources: [workflows/paradigm_pulse_daily.py:209-228]()

The `slack_sync` workflow demonstrates a different pattern: it iterates over Slack channels in a plain `for` loop, calling direct Postgres helpers rather than `ctx.step`, because its per-channel work is idempotent upserts that do not need step-level checkpointing. It uses `ctx.log` for structured observability that is suppressed on replay.

Sources: [workflows/slack_sync.py:324-470]()

---

## Summary

The checkpoint/replay model makes Centaur workflows crash-safe, resumable, and idempotent by construction. The handler function is ordinary Python; the engine transparently intercepts `ctx.step` calls to cache results in Postgres. Sleep, event suspension, and child-workflow waiting all use the same mechanism: write a marker checkpoint, raise `SuspendWorkflow`, and let the reconciler wake the run when the condition clears. Because each re-execution fast-forwards through all previously checkpointed steps before reaching the live frontier, complex multi-hour workflows can be interrupted at any point and resumed correctly without custom recovery logic in the handler.

Sources: [services/api/api/workflow_engine.py:1-14]()

---

## 08. Secrets & Egress: iron-proxy as the Trust Boundary

> How iron-proxy is the single egress choke point for every sandbox, why it is per-sandbox rather than shared (a compromised pod cannot leak into another), the four secret transform types (replace, inject, gcp_auth, oauth_token, hmac_sign), the NetworkPolicy default-deny invariant, and what changes break the security model (shared proxy, raw key injection, relaxed NetworkPolicy).

- Page Markdown: https://grok-wiki.com/public/wiki/paradigmxyz-centaur-57fc6b2755e2/pages/08-secrets-egress-iron-proxy-as-the-trust-boundary.md
- Generated: 2026-05-21T23:43:13.719Z

### Source Files

- `services/api/api/proxy_config.py`
- `services/api/api/iron-proxy.base.yaml`
- `services/api/api/tool_manager.py`
- `docs/pages/security.mdx`
- `services/api/tests/test_proxy_config.py`

<details>
<summary>Relevant source files</summary>
The following files were used as context for generating this wiki page:

- [services/api/api/proxy_config.py](services/api/api/proxy_config.py)
- [services/api/api/iron-proxy.base.yaml](services/api/api/iron-proxy.base.yaml)
- [services/api/api/tool_manager.py](services/api/api/tool_manager.py)
- [docs/pages/security.mdx](docs/pages/security.mdx)
- [services/api/tests/test_proxy_config.py](services/api/tests/test_proxy_config.py)
- [contrib/chart/templates/networkpolicy.yaml](contrib/chart/templates/networkpolicy.yaml)
</details>

# Secrets & Egress: iron-proxy as the Trust Boundary

Every Centaur sandbox runs untrusted, model-generated code. That code must be able to call external APIs — but it must never hold the real credentials to do so, and it must not be able to send traffic to arbitrary destinations. iron-proxy is the architectural mechanism that satisfies both requirements simultaneously: it is the single egress choke point through which all sandbox network traffic flows, and it is the only component that ever holds or resolves real secret values.

This page explains how iron-proxy is positioned in the network topology, why each sandbox gets its own dedicated proxy instance, how the four managed secret transform types work, what the NetworkPolicy default-deny invariant enforces, and which changes would silently break the model.

---

## The Egress Choke Point

Sandbox pods do not have a direct route to the internet. All outbound HTTP/HTTPS traffic is routed through iron-proxy, which acts as a TLS-terminating MITM (man-in-the-middle) proxy. The base config confirms the proxy operates in `mitm` mode with its own CA certificate:

```yaml
# services/api/api/iron-proxy.base.yaml:14-16
tls:
  mode: "mitm"
  ca_cert: "/etc/iron-proxy/ca.crt"
  ca_key: "/etc/iron-proxy/ca.key"
```

The proxy also runs an embedded DNS server (`:53`) alongside the HTTP tunnel listener (`:8080`) and a management API (`:9092`). All DNS lookups from the sandbox resolve through iron-proxy's DNS forwarder, meaning the proxy is in the path of every connection, not just those targeted at known hosts.

Sources: [services/api/api/iron-proxy.base.yaml:1-16]()

---

## Per-Sandbox Isolation

A single shared proxy would be a cross-contamination risk: a compromised sandbox could observe or inject traffic from other sandboxes sharing the same proxy. Centaur avoids this by giving every sandbox its own iron-proxy pod.

The security documentation makes this explicit:

> Because iron-proxy is per-sandbox rather than shared, a compromise of one sandbox's proxy cannot leak into another sandbox.

Sources: [docs/pages/security.mdx:47-51]()

Each iron-proxy pod carries two labels that uniquely identify its scope:

- `centaur.ai/iron-proxy: "true"` — marks it as a proxy pod
- `centaur.ai/sandbox-id: <id>` — ties it to exactly one sandbox

The NetworkPolicy for the API-side proxy selects pods with both labels, meaning only the API process and its associated proxy can communicate with each other on the proxy and Postgres listener ports.

Sources: [contrib/chart/templates/networkpolicy.yaml:300-346]()

---

## The NetworkPolicy Default-Deny Invariant

The Helm chart establishes a namespace-wide default-deny policy that blocks all ingress and egress for every pod in the namespace:

```yaml
# contrib/chart/templates/networkpolicy.yaml:9-12
spec:
  podSelector: {}
  policyTypes:
    - Ingress
    - Egress
```

`podSelector: {}` selects all pods. Without an additive allow rule, no pod can send or receive any traffic. Subsequent `NetworkPolicy` objects in the same file then grant the minimum set of permissions for each component:

| Component | Permitted Egress | Permitted Ingress |
|---|---|---|
| Sandbox pod (`centaur.ai/managed: "true"`) | API pod port 8000 only | (none defined) |
| API pod | Postgres, slackbot, its own iron-proxy, port 443 | slackbot, iron-proxy, sandbox pods, allowed CIDRs |
| iron-proxy (API-scoped) | TCP 80, 443, 5432 | API pod (proxy port + pg port range) |
| All pods | kube-system UDP/TCP 53 (DNS) | — |

Sandbox pods are labeled `centaur.ai/managed: "true"` and their NetworkPolicy permits egress only to the API pod on port 8000. The sandbox communicates with its iron-proxy through the API's own proxy listener — the sandbox does not hold the proxy's address directly.

Sources: [contrib/chart/templates/networkpolicy.yaml:1-14, 400-420]()

---

## How the API Owns iron-proxy Configuration

The API server is the single authority that produces iron-proxy's runtime config. `proxy_config.py` describes this clearly:

> Centralizes what was previously split between firewall-manager (rendering) and tool_manager (injection map). The API server owns iron-proxy's full config.

`render_proxy_yaml()` takes the list of all declared `SecretDef` objects, splices the managed transforms into the base YAML (inserted before `header_allowlist`), and returns the final iron-proxy config string. The base config provides the allowlist, header allowlist, TLS, DNS, management, and log settings; the API injects all secret-handling transforms at render time.

```python
# services/api/api/proxy_config.py:392-438 (abridged)
def render_proxy_yaml(secrets, base_config=None, *, pg_listen_ports=None):
    cfg = yaml.safe_load(base_config) or {}
    # Strip any previously managed transforms, then rebuild them:
    transforms = [t for t in (cfg.get("transforms") or [])
                  if (t or {}).get("name") not in _MANAGED_TRANSFORMS]
    new_transforms = [...]  # built from secrets
    # Insert before header_allowlist
    cfg["transforms"] = transforms
    ...
    return yaml.safe_dump(cfg, sort_keys=False)
```

The four managed transform names are tracked in `_MANAGED_TRANSFORMS`:

```python
# services/api/api/proxy_config.py:53-55
_MANAGED_TRANSFORMS: frozenset[str] = frozenset(
    {"secrets", "gcp_auth", "oauth_token", "hmac_sign"}
)
```

Sources: [services/api/api/proxy_config.py:1-17, 53-55, 392-438]()

---

## The Four Secret Transform Types

### 1. `secrets` — Replace and Inject Modes

`HttpSecret` is the general-purpose HTTP credential transform. It has two modes:

**Replace mode** (default): The tool writes a placeholder token (the `replacer`) into its request — in a header, query string, or path. iron-proxy scans the configured locations and swaps the placeholder for the resolved credential. The credential value never enters the sandbox.

```python
# services/api/api/tool_manager.py:65-73
class SecretMode(str, Enum):
    REPLACE = "replace"  # tool writes placeholder; proxy swaps it
    INJECT = "inject"    # proxy adds credential; tool never sees it
```

**Inject mode**: iron-proxy adds the credential to the request entirely by itself — the tool emits no placeholder and never observes the value. Supports `inject_header` (with an optional Go-template `inject_formatter` such as `Bearer {{ .Value }}`) or `inject_query_param`.

Each `HttpSecret` entry carries a `hosts` list that becomes `rules` in the rendered config. iron-proxy will only substitute or inject the credential for those specific hosts. A leaked placeholder cannot be redirected to an attacker-controlled host, and it cannot be smuggled out through a different field.

```yaml
# Rendered replace-mode entry (env source)
- source: {type: env, var: OPENAI_API_KEY}
  replace:
    proxy_value: OPENAI_API_KEY
    match_headers: [Authorization]
  rules:
    - host: api.openai.com
```

Sources: [services/api/api/proxy_config.py:101-156](), [services/api/api/tool_manager.py:65-122](), [services/api/tests/test_proxy_config.py:807-878]()

---

### 2. `gcp_auth` — GCP Service Account Token Injection

`GcpAuthSecret` feeds a Google service-account keyfile to iron-proxy. iron-proxy loads the keyfile, mints OAuth2 bearer tokens for the configured scopes, and injects them as `Authorization: Bearer` on matching upstreams. The sandbox never receives the keyfile or the minted token.

Multiple GCP service accounts can coexist. Each unique keyfile (`secret_ref`) gets its own `gcp_auth` transform; secrets sharing a `secret_ref` are merged into one transform with the union of their hosts and scopes.

Default values apply when left unset: `*.googleapis.com` for hosts, `https://www.googleapis.com/auth/cloud-platform` for scopes.

```python
# services/api/api/proxy_config.py:48-49
GCP_AUTH_SCOPES: tuple[str, ...] = ("https://www.googleapis.com/auth/cloud-platform",)
GCP_AUTH_HOSTS: tuple[str, ...] = ("*.googleapis.com",)
```

> Superseded by `oauth_token`; kept until tools migrate off the `gcp_auth` secret type.

Sources: [services/api/api/proxy_config.py:159-193](), [services/api/api/tool_manager.py:124-145]()

---

### 3. `oauth_token` — Generic OAuth2 Token Exchange

`OAuthTokenSecret` generalizes the GCP token-minting pattern to any OAuth2 provider. iron-proxy resolves each named credential field from its own secret source, runs the configured grant exchange, caches and refreshes the resulting access token, and injects it as `Authorization: Bearer` on matching hosts.

Four grant types are supported:

| Grant | RFC | Required fields |
|---|---|---|
| `refresh_token` | RFC 6749 | `refresh_token`, `client_id` (+ optional `client_secret`) |
| `client_credentials` | RFC 6749 §4.4 | `client_id`, `client_secret` |
| `password` | RFC 6749 §4.3 | `username`, `password`, `client_id` (+ optional `client_secret`) |
| `jwt_bearer` | RFC 7523 | `issuer`, `subject`, `private_key`, `audience` (+ optional `private_key_id`) |

Credential fields can be sourced from separate secrets or extracted from a single JSON-encoded secret via `json_key`. Optional `token_endpoint_headers` allows extra headers on the token POST itself — useful when the token endpoint requires an API key alongside form-body client auth.

Multiple `OAuthTokenSecret` entries that resolve to the same token (same grant, credential fields, and token endpoint) are merged, unioning their hosts and scopes into a single `tokens` entry.

Sources: [services/api/api/proxy_config.py:208-285](), [services/api/api/tool_manager.py:179-213](), [services/api/tests/test_proxy_config.py:970-1010]()

---

### 4. `hmac_sign` — Per-Request HMAC Signature

`HmacSignSecret` implements exchange-style HMAC authentication used by trading APIs such as FalconX. iron-proxy resolves all credentials, composes the canonical message template, computes the HMAC digest, and writes the configured request headers. The signing key and all other credentials never reach the sandbox.

Key parameters:

| Parameter | Allowed values |
|---|---|
| `algorithm` | `sha256`, `sha512`, `sha1` |
| `key_encoding` | `raw`, `base64`, `hex` |
| `output_encoding` | `base64`, `hex` |
| `timestamp_format` | `unix_seconds`, `unix_millis`, `unix_nanos`, `rfc3339` |

The `message` field is a Go template with access to `.Timestamp`, `.Method`, `.PathWithQuery`, and `.Body`. Headers are Go templates with access to `.Signature`, `.Timestamp`, and `.Credentials.<name>`. By default, chunked-body requests are refused (body cannot be deterministically hashed in flight); `allow_chunked_body: true` opts in.

```python
# services/api/api/tool_manager.py:289-297 (enum constants)
_HMAC_ALGORITHMS = frozenset({"sha256", "sha512", "sha1"})
_HMAC_KEY_ENCODINGS = frozenset({"raw", "base64", "hex"})
_HMAC_OUTPUT_ENCODINGS = frozenset({"base64", "hex"})
_HMAC_TIMESTAMP_FORMATS = frozenset({"unix_seconds", "unix_millis", "unix_nanos", "rfc3339"})
```

Sources: [services/api/api/proxy_config.py:288-356](), [services/api/api/tool_manager.py:229-257](), [services/api/tests/test_proxy_config.py:1337-1369]()

---

## Secret Source Backends

All four transform types share the same secret-source abstraction. The `FIREWALL_MANAGER_SECRET_SOURCE` environment variable controls where iron-proxy resolves credential values:

| Value | iron-proxy source type | Secret reference format |
|---|---|---|
| `env` (default) | `env` | Environment variable name |
| `onepassword` | `1password` | `op://<vault>/<ref>/credential` |
| `onepassword-connect` | `1password_connect` | `op://<vault>/<ref>/credential` |

The vault name is read from `OP_VAULT` (default: `ai-agents`). Secret TTL for cached values defaults to `10m` (configurable via `FIREWALL_MANAGER_SECRET_TTL`).

```python
# services/api/api/proxy_config.py:79-88
def _build_source(secret_ref: str) -> dict[str, str]:
    iron_proxy_type = _OP_REF_SOURCES.get(_secret_source_kind())
    if iron_proxy_type is not None:
        return {"type": iron_proxy_type,
                "secret_ref": f"op://{_op_vault()}/{secret_ref}/credential",
                "ttl": _secret_ttl()}
    return {"type": "env", "var": secret_ref}
```

Sources: [services/api/api/proxy_config.py:56-88]()

---

## Egress Domain Allowlist

The base config ships with a permissive `allowlist` transform that allows all domains:

```yaml
# services/api/api/iron-proxy.base.yaml:17-21
transforms:
  - name: allowlist
    config:
      domains:
        - "*"
```

This is a deliberate UX trade-off: operators can start without configuring an allowlist and tighten it later. To lock egress down, replace `"*"` with explicit hostname patterns (e.g., `*.anthropic.com`). iron-proxy rejects unlisted destinations with a 403.

The `header_allowlist` transform further constrains which request headers sandbox code can send. Headers not on the allowlist are stripped before forwarding. Notably, the header allowlist includes a regex pattern for common auth headers (`/^x-[a-z0-9-]*(api-key|apikey|secret|token|auth|key)$/`), reflecting the range of API credential header names tools may use.

Sources: [services/api/api/iron-proxy.base.yaml:17-78](), [docs/pages/security.mdx:58-74]()

---

## Infrastructure Secrets

Beyond tool-declared secrets, the API server registers a set of hardcoded infrastructure secrets covering the LLM API keys and platform credentials that the harness itself uses:

```python
# services/api/api/tool_manager.py:1594-1637 (abridged)
_INFRA_SECRETS: ClassVar[list[HttpSecret]] = [
    HttpSecret("ANTHROPIC_API_KEY", ..., hosts=("api.anthropic.com",), match_headers=("X-Api-Key",)),
    HttpSecret("OPENAI_API_KEY",    ..., hosts=("api.openai.com",),    match_headers=("Authorization",)),
    HttpSecret("XAI_API_KEY",       ..., hosts=("api.x.ai",),          match_headers=("Authorization",)),
    HttpSecret("GEMINI_API_KEY",    ..., hosts=("generativelanguage.googleapis.com",), match_headers=("X-Goog-Api-Key",)),
    HttpSecret("GITHUB_TOKEN",      ..., hosts=("github.com", "api.github.com"), ...),
    HttpSecret("SLACK_BOT_TOKEN",   ..., hosts=("*.slack.com",), ...),
    ...
]
```

`collect_secrets()` merges infrastructure secrets with tool-declared secrets before rendering:

```python
# services/api/api/tool_manager.py:1639-1648
def collect_secrets(self) -> list[SecretDef]:
    out: list[SecretDef] = list(self._INFRA_SECRETS)
    for lt in self.tools.values():
        out.extend(lt.all_secrets)
    return out
```

Sources: [services/api/api/tool_manager.py:1592-1648]()

---

## What Tools See vs. What iron-proxy Sees

The distinction between what is visible inside the sandbox and what is visible only to iron-proxy is a core invariant of the design:

| Secret type | What sandbox tool code sees | What iron-proxy holds |
|---|---|---|
| `http` (replace) | Placeholder token (e.g. `WAREHOUSE_API_KEY`) | Resolved credential value |
| `http` (inject) | Nothing | Resolved credential value |
| `gcp_auth` | Nothing | GCP keyfile; minted token |
| `oauth_token` | Nothing | Credential fields; minted access token |
| `hmac_sign` | Nothing | Signing key; computed HMAC digest |
| `pg_dsn` | Local DSN (`localhost:5432`) | Real upstream DSN |

The sandbox receives placeholder strings via `ToolContext.secrets` — only for replace-mode `HttpSecret`. All other secret types are applied entirely on the wire:

```python
# services/api/api/tool_manager.py:838-850
async def _resolve_secrets(secrets: list[SecretDef]) -> dict[str, str]:
    """Only replace-mode HttpSecret entries end up in the tool's ToolContext.
    Inject-mode HTTP secrets are applied entirely by iron-proxy.
    GcpAuthSecret, OAuthTokenSecret and PgDsnSecret are likewise not
    exposed via context."""
    return {s.name: s.replacer for s in secrets if _is_replace_secret(s)}
```

Sources: [services/api/api/tool_manager.py:838-850](), [docs/pages/security.mdx:82-114]()

---

## What Breaks the Security Model

The following changes each remove a different layer of the trust boundary:

### Shared proxy (one proxy for multiple sandboxes)
A single iron-proxy instance serving multiple sandboxes defeats per-sandbox isolation. If a sandbox can compromise its proxy (through a vulnerability in iron-proxy or in a transform), it gains visibility into every other sandbox's traffic and resolved credentials. The NetworkPolicy uses per-sandbox `centaur.ai/sandbox-id` labels specifically to prevent lateral movement between proxy pods.

### Raw key injection into sandbox environment
If real credential values are placed in sandbox environment variables, files, or the tool's `ToolContext.secrets`, they can be exfiltrated through logs, error messages, tool return values, or direct prompt injection. The entire point of the replace/inject model is that the sandbox holds only placeholder tokens that are meaningless outside the proxy.

### Relaxed NetworkPolicy
If the default-deny policy is removed or if sandbox pods receive egress beyond the API pod, they can bypass iron-proxy entirely by opening direct connections to external hosts. The default-deny-then-additive-allow structure means any misconfiguration defaults to blocked, not open. Removing the `centaur-default-deny` `NetworkPolicy` object would silently grant all pods full network access.

### Relaxed allowlist (`domains: ["*"]` left in production)
The default permissive allowlist means a prompt-injection attack can redirect a tool to an attacker-controlled host. The placeholder substitution is host-scoped — a credential will only be injected for matching hosts — but a separate secret exfiltration channel (sending arbitrary data via `fetch()` in the sandbox) would succeed against any reachable host. Locking the allowlist to the exact set of hosts each tool needs is the primary control against data exfiltration.

Sources: [docs/pages/security.mdx:47-51, 58-74, 130-156](), [contrib/chart/templates/networkpolicy.yaml:1-14]()

---

## Summary

iron-proxy is the singular outbound channel for every sandbox and the sole holder of resolved secret values. Its security properties depend on four reinforcing invariants: (1) per-sandbox deployment so a compromised proxy cannot bleed across sandbox boundaries, (2) NetworkPolicy default-deny enforced at the Kubernetes level so sandbox code cannot open connections that bypass the proxy, (3) the placeholder-not-value model so credentials cannot be read from the sandbox's environment or logs, and (4) host-scoped rules so every credential substitution is bound to the specific upstream that legitimately needs it. Audit and structured logging from iron-proxy and the agent execution record together provide the evidence trail needed to reconstruct what any agent did and which credentials it reached for. ([docs/pages/security.mdx:119-124]())

---