Coding OS
Sign in
Browse docs▾

Dispatcher Contract

Purpose: Define the provider-neutral request/result protocol and the adapter-owned execution responsibilities for formula agents. Read when: changing DispatchRequest, DispatchResult, an adapter dispatcher, or cross-adapter routing. Skip when: changing an interactive provider UI that never invokes formula dispatch. Read next: Codex adapter, Claude SDK, adapter parity.

Nav: Section Index | Docs Index

Source of truth. Update this contract before extending dispatcher behavior. A disagreement is drift to repair, not permission for the code to redefine the contract silently.

Purpose

Formula agents (F1..F11) need to run somewhere. The runtime is different per agent CLI:

Agent Spawn channel SDK kind
Claude claude-agent-sdk.query() required Python library
Codex default codex exec --json; optional openai-codex app-server stable CLI plus opt-in beta Python SDK with a pinned runtime
any DefaultDispatcher (DB-only fallback) n/a - inline only

The contract here is the agent-agnostic shape every dispatcher must satisfy so the supervisor (cos_supervise, cos_dispatch_formula) does not need to know which runtime it is talking to.

Contract Surface

Defined in src/core/thinking_os/dispatcher.py.

IO models

class DispatchRequest(BaseModel):
    formula_id: str            # safe slug — embeddable in filenames
    agent_file: str            # absolute or thinking_os-relative path
    prompt: str                # composed system+user prompt
    input_slice: dict          # upstream-only EvidenceBundle view
    persona_id: str | None
    intensity: Literal["light", "standard", "full"]
    allowed_tools: list[str]
    timeout_s: float
    session_id: str | None
    cwd: str | None
    model: str | None          # forwarded to the adapter; None = adapter default
    effort: str | None         # adapter-declared reasoning effort
    complexity: str            # Cynefin level; "" is below every adaptive gate
    max_budget_usd: float | None
    long_context: bool
    adapter: str | None        # target-runtime HINT (e.g. "codex"); see below
    max_turns: int | None      # adapter-owned cap when the runtime supports it

class DispatchResult(BaseModel):
    formula_id: str
    status: Literal["ok", "timeout", "error", "skipped"]
    output_json: dict          # validated against formula's output_schema
    latency_ms: int
    error: str | None
    dispatcher_name: str       # "claude-sdk" | "codex-sdk" | "default" | ...
    raw_transcript: str | None
    error_category: str | None # capacity | auth | unavailable | timeout | provider | invalid
    retryable: bool
    retry_after_s: int | None
    outcome: str               # known_failed | unknown

Protocol

@runtime_checkable
class AgentDispatcher(Protocol):
    name: str
    async def dispatch(self, request: DispatchRequest) -> DispatchResult: ...
    def available(self) -> bool: ...

Status semantics

status Meaning Caller action
ok Sub-agent ran, returned valid output_json Validate against formula schema, persist
timeout Sub-agent exceeded timeout_s Surface as transient; main agent may retry
error SDK failure, parse failure, missing agent file, subprocess rc≠0 Surface as internal; do not silently downgrade
skipped Dispatcher cannot spawn (no SDK, no binary, stub adapter) Main agent inlines the formula and records output

Architecture

┌──────────────────────────────────────────┐
│ src/core/thinking_os/dispatcher.py       │   agent-agnostic
│   • DispatchRequest / DispatchResult     │
│   • AgentDispatcher Protocol             │
│   • get_dispatcher() factory             │
│ src/core/thinking_os/dispatcher_helpers.py   │
│   • load_agent_prompt()                  │
│   • extract_json_block()                 │
└─────────────────┬────────────────────────┘
                  │ importlib (path-based; no static link)
   ┌──────────────┼───────────────────────────┐
   ▼              ▼                            ▼
src/adapters/claude/  src/adapters/codex/  src/core/thinking_os/dispatchers/
sdk_dispatcher.py     sdk_dispatcher.py    default.py
   │                     │                   │
   ▼                     ▼                   ▼
claude-agent-sdk     codex CLI default     DB-only fallback
                     Python SDK opt-in     (skipped)

Why three implementations and not one: the SDKs are different runtimes, not different views of the same runtime. A unified body would either have to re-implement what each SDK does (more code, more surface to break) or abstract over differences that do not exist as a real abstraction (in-proc async generator vs. subprocess vs. nothing). Hexagonal here gives us:

  • src/core/ agent-agnostic (Rule 1)
  • adapters self-contained (Principle P8)
  • new agents add a folder, not a switch statement

Factory rules (get_dispatcher)

  1. If COS_FORCE_DEFAULT_DISPATCHER=1 → DefaultDispatcher. Tests use this.
  2. Detect agent from COS_AGENT env, then COS_AGENT_DIR folder name.
  3. Try to load src/adapters/<agent>/sdk_dispatcher.py::build_dispatcher().
  4. Call available(); if False, fall through to DefaultDispatcher.
  5. DefaultDispatcher.available() is always True — last-resort path.

The loader is importlib.util.spec_from_file_location so src/core/ never has a static import on src/adapters/.

  1. Supervision-gated adapter switch. DispatchRequest.adapter remains an advisory hint while supervision is disabled. When project settings enable supervision, the dispatcher resolves that id through the manifest registry, applies runtime health policy, and invokes that adapter for the request. A run never changes adapter after execution starts. Per-call cost ceilings ride on max_budget_usd; adapter-specific budget carriers are not added.

Parity rules

Every adapter dispatcher MUST:

  1. Import DispatchRequest/DispatchResult from thinking_os.dispatcher (not by relative path or sys.path injection).
  2. Use load_agent_prompt and extract_json_block from thinking_os.dispatcher_helpers rather than re-implementing them.
  3. Expose build_dispatcher() -> AgentDispatcher as the factory entry-point.
  4. Return status="error" (not raise) on FileNotFoundError, SDK import failure, subprocess rc≠0, parse failure, etc.
  5. Set dispatcher_name to the same string as self.name.
  6. Forward declared model and effort; surface unsupported budget, context, tool, or turn controls instead of silently dropping them.
  7. Normalize native capacity failures into error_category, retryable, and retry_after_s without leaking credentials or full provider payloads.
  8. Never retry on another backend after a provider turn may have started; duplicate execution is worse than a visible error.

The Codex CLI backend additionally MUST use the current non-interactive surface (codex exec), write the prompt to stdin, parse JSONL events, and run formula output in a read-only sandbox with approvals disabled. Formula dispatch ignores user configuration, disables hooks, and clears MCP servers so host customizations cannot recurse into or mutate a supervised sub-run. The optional Python SDK is beta and selected explicitly. Its availability is independent of a global CLI installation because published SDK builds include a pinned runtime; normal SDK dispatch must not replace that runtime with a newer system binary implicitly.

Adding a new adapter dispatcher

  1. Create src/adapters/<agent>/sdk_dispatcher.py.
  2. Import the contract + helpers from thinking_os.*.
  3. Implement a class with name, available(), and async dispatch().
  4. Add build_dispatcher() -> YourDispatcher.
  5. Declare runtime_entrypoints.dispatch, capabilities, models, and efforts in src/adapters/<agent>/adapter.yaml.
  6. Add a parity test in tests/test_adapter_parity.py.
  7. Core routing needs no provider-specific edit.

Tests

File What it covers
src/core/thinking_os/tests/test_dispatcher.py Protocol shape, factory, default path
src/core/thinking_os/tests/test_dispatcher.py Codex CLI/SDK paths and request parity
tests/test_adapter_parity.py Hook + dispatcher parity across agents