{"id":"f1485590-bee2-4c05-9876-c24b389c318d","slug":"ramannanda9-agent-harness","kind":"mcp","name":"agent-harness","summary":"BYO-LLM multi-agent harness: hybrid DAG orchestration, router, two-tier memory, streaming-primary event model, sandboxed tool execution (native + Docker)","owner":"ramannanda9","repoUrl":"https://github.com/ramannanda9/agent-harness","license":"MIT","licenseTier":"permissive","trust":"community","stars":5,"match":null,"fewStars":false,"format":"mcp","createdAt":"2026-05-11T19:47:42.000Z","addedAt":"2026-10-21T17:21:00.857Z","logoUrl":"https://avatars.githubusercontent.com/u/883240?u=ee7d1f37cdc514e6c4495c6b82786033f412b25b&v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["cloud-infra","memory-knowledge"],"tags":["ai-agents","llm","multi-agent","orchestration","python","rust","streaming"],"runtime":"python","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-26T04:52:09.009Z","sample":false,"url":"https://conduits.dev/tools/ramannanda9-agent-harness","language":"Python","transports":["stdio"],"install":{"pypi":"react-agent-harness"},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"# react-agent-harness\n\nBring-your-own-LLM multi-agent harness: hybrid DAG planning with replan-on-failure,\ntwo-tier memory (semantic KV + episodic vector), a streaming-primary event model,\nand cost/token budgets with per-call-site attribution (classifier, router,\nplanner, synthesizer, agent).\n\nConfig-driven — register tools and agents, run any goal. No subclassing.\n\n## Install\n\n```bash\npip install -e \".[dev]\" # core + tests + lint\npip install -e \".[openai,http,dev]\" # OpenAI adapter + HTTPFetch tool\npip install -e \".[lance,redis,dev]\" # LanceDB + Redis durable memory\npip install -e \".[mcp]\" # MCP tool adapter\npip install -e \".[otel]\" # OpenTelemetry tracing\n```\n\nThe core package has no runtime dependencies. The `openai`, `http`, `lance`,\n`redis`, `anthropic`, `mcp`, and `otel` extras pull in heavier libs only if\nyou opt in.\n\n## Quickstart\n\n```bash\npython examples/quickstart.py # mock LLM, no keys, no network\nOPENAI_API_KEY=sk-... python examples/openai_demo.py # real OpenAI + HTTPFetch\n```\n\n`quickstart.py` exercises the full wiring against a deterministic `MockLLM`.\n`openai_demo.py` runs the same shape against a real model — streams live\nevents to stdout and prints elapsed time + cost at the end.\n\n## Architecture\n\n```\nharness/runtime.py AgentRuntime — single entry point; BudgetGuard with cost/token caps + per-call-site breakdown\nharness/events.py BusEvent + EventType — canonical event vocabulary\nharness/llm/openai.py OpenAILLM — OpenAI API-key adapter with usage + cost tracking\nharness/llm/anthropic.py AnthropicLLM — direct Anthropic API-key adapter with prompt-caching support\nharness/llm/claude_code.py ClaudeCodeLLM — Claude subscription OAuth adapter (experimental, ToS caveats)\nharness/llm/openai_codex.py OpenAICodexLLM — ChatGPT subscription OAuth adapter (experimental, ToS caveats)\nharness/llm/auth.py Shared OAuth + auth-file primitives for the subscription adapters\nharness/llm/fallback.py FallbackLLM — transparent retry on transient upstream errors\nharness/llm/routing.py RoutingLLM — dispatch calls to different adapters by a selector\nharness/trace.py JSONL trace recorder + replay — durable, per-event flush\nharness/trace_viewer.py Local web timeline viewer for recorded JSONL traces\nharness/annotation.py Annotation store + AnnotationHook — RLHF trajectory capture\nharness/hitl.py HITL approval gate — interactive CLI, session-allow list\nharness/tool_policy.py Persistent tool policy — user-scoped allow rules, CLI management\nharness/console.py ConsoleRenderer — centralised BusEvent formatting + render_budget helper\nharness/steering.py Async steering — agent.steer(text), StdinRouter pub/sub, FileSteer, factory helpers\nharness/checkpoint.py CheckpointStore + _ResumeHint + maybe_resume_key — pluggable run-state persistence (file + Redis); auto-resume built into dispatch_stream / run_stream\nharness/runstate.py RunState / OrchestratorState — the explicit, serializable position of a run: step + phase, per-action and per-task status, budget, failure reason\nharness/otel.py OTELHook — OpenTelemetry span exporter (opt-in)\nharness/executor_bridge.py ExecutorBridge + ExecutorTool — controlled subprocess launcher with optional Docker sandboxing\nharness/oauth_browser.py Localhost OAuth callback server + open_or_print_url — shared by MCP browser-OAuth and LLM login flows\nharness/persistent.py PersistentAgent + SQLiteSessionStore — durable chat sessions around user-built agents\norchestrator/planner.py Hybrid DAG orchestrator — plan, replan, synthesize\nagents/base.py Generic BaseAgent — ReAct loop, no subclassing needed\nmemory/manager.py MemoryManager — semantic KV + episodic vector\nmemory/working.py WorkingMemory — LLM summarization eviction, checkpoint/restore\nmemory/episodic_lance.py LanceDB episodic store — IVF_PQ ANN, batch writes\nmemory/redis_store.py Redis semantic store — durable KV with TTL\nmemory/stores.py InMemory stores — local dev default, no deps\ntools/builtin/http_fetch.py HTTPFetch — minimal read-only GET tool\ntools/builtin/fetch_image.py FetchImage — fetch URL and return OpenAI image_url block\ntools/builtin/subagent.py SubAgentTool — expose a BaseAgent as a parent-callable streaming tool\ntools/mcp/adapter.py MCP tool adapter — stdio, SSE, streamable-HTTP transports\ntools/mcp/auth.py ApiKeyMCPAuth + BrowserOAuthMCPAuth — auth primitives for remote MCP servers\nharness/streaming.py Multi-producer fan-in for parallel streaming tools (sub-agents)\n```\n\nExecution is **streaming-primary**: every path yields `BusEvent`s for\ndispatch, routing, plan, thoughts, tool calls, observations, task completions,\nreplans, and synthesis. The blocking variants drain the same stream.\n\n`dispatch_stream(goal)` is the recommended entry point — it classifies\ncomplexity with one cheap LLM call and delegates automatically to the routed\nor orchestrated path. Use the lower-level paths directly only when you need\nexplicit control.\n\n## Examples\n\n| Script | What it shows | Requires |\n|---|---|---|\n| `examples/quickstart.py` | End-to-end against `MockLLM` + `EchoTool` — reference wiring. | nothing |\n| `examples/openai_demo.py` | Real OpenAI + `HTTPFetch` + `shell` (via `ah-executor`), routed single-agent run, live event stream, cost reporting. | `OPENAI_API_KEY`, `[openai,http]` |\n| `examples/vision_demo.py` | Multimodal agent: fetches two images in parallel via `FetchImage`, describes each using the LLM's vision capability, synthesises a report. | `OPENAI_API_KEY`, `[openai,http]` |\n| `examples/complex_sysaudit_demo.py` | Three heterogeneous agents in parallel: `shell_agent` (ah-executor), `filesystem_agent` (MCP), `web_agent` (HTTPFetch) — orchestrated path, DAG plan, synthesis. | `OPENAI_API_KEY`, `[openai,http,mcp]`, `ah-executor`, `npx` |\n| `examples/executor_bridge_demo.py` | `ExecutorBridge` backends side-by-side: allowlist, env scrubbing, Docker network/fs isolation, timeout, positional-arg tools. | `ah-executor` and/or Docker |\n| `examples/durable_memory_demo.py` | Redis (semantic) + LanceDB (episodic) memory persistence across two related goals. | `OPENAI_API_KEY`, `[openai,redis,lance]`, Redis reachable |\n| `examples/mcp_demo.py` | Connects to an MCP filesystem server and gives the agent its tools. | `OPENAI_API_KEY`, `[openai,mcp]`, `npx` |\n| `examples/mcp_auth_demo.py` | Connects to an authenticated remote MCP server using bearer or auth-file credentials. | `OPENAI_API_KEY`, `[openai,mcp]`, `MCP_URL`, `MCP_BEARER_TOKEN` or `MCP_AUTH_PROVIDER` |\n| `examples/subscription_auth_demo.py` | Runs an agent through subscription-backed providers: direct `openai-codex` OAuth or direct `claude-code` OAuth. | `agent-harness login openai-codex` or `agent-harness login claude-code` |\n| `examples/coordinator_demo.py` | Sub-agent-as-tool pattern: a `coordinator` ReAct agent delegates dynamically to `researcher` / `analyst` / `reporter` via `SubAgentTool`. Demonstrates parallel delegation through `actions: [...]`. | `OPENAI_API_KEY`, `[openai,http]` |\n| `examples/persistent_agent_demo.py` | Persistent local assistant: SQLite session + semantic memory, Lance episodic memory, shell tool, and a browser researcher via `@playwright/mcp`. Supports `--provider openai`, `--provider openai-codex`, or `--provider claude-code`. | `[openai,mcp,lance]`, `OPENAI_API_KEY` or `python -m harness.cli login openai-codex` / `claude-code`, `ah-executor`, `npx` (Node 18+) |\n\n## Adding a new domain (3 steps)\n\n```python\n# 1. Register tools\ntools.register(MyTool())\n\n# 2. Register agent config — no subclassing\nagents.register(AgentConfig(\n agent_id=\"my_agent\",\n role=\"does X using tools Y and Z\",\n system_prompt=\"You are an expert at X...\",\n allowed_tools=[\"my_tool\"],\n))\n\n# 3. Run\nresult = await runtime.dispatch(\"my goal\")\n```\n\nAgents can also attach reusable prompt skills from Agent Skills-style\ndirectories:\n\n```text\nskills/\n web-research/\n SKILL.md\n references/ # optional\n scripts/ # optional\n```\n\n```md\n---\nname: web-research\ndescription: Research current information from primary sources.\nallowed-tools:\n - browser_navigate\n - browser_snapshot\n---\n\nPrefer primary sources, capture dates, and cite URLs.\n```\n\n```python\nfrom harness.skills import load_skill, load_skills\n\nweb_research = load_skill(\"skills/web-research\")\n# Or load every immediate child directory with a SKILL.md:\n# skills = load_skills(\"skills\")\n# Calling load_skills() with no path loads ~/.agent-harness/skills\n# and returns [] when that default directory does not exist.\n\nagents.register(AgentConfig(\n agent_id=\"researcher\",\n role=\"researches current information\",\n system_prompt=\"You are a careful research agent.\",\n allowed_tools=[\"browser_navigate\", \"browser_snapshot\"],\n skills=[web_research],\n))\n```\n\nSkills add reusable instructions and tool hints to that agent's system prompt.\nThey do not grant tools; `allowed-tools` in `SKILL.md` is treated as a hint\nonly. Tool access still comes from the wired tool map and the agent's\nconfigured tool surface. You can also construct `Skill(...)` directly when a\nprogrammatic skill is more convenient.\n\n## LLM clients\n\nThe harness is **BYO-LLM.** Any object with\n`async def complete(system, messages, **kwargs) -> dict` works. Optional\n`async def stream_complete(system, messages) -> AsyncGenerator[str, None]`\nenables `TOKEN` events.\n\nThe OpenAI adapter is shipped because it's the easiest spin:\n\n```python\nfrom harness.llm.openai import OpenAILLM\n\nllm = OpenAILLM(model=\"gpt-4o-mini\") # reads OPENAI_API_KEY from env\nruntime = AgentRuntime(..., llm=llm)\n```\n\nReasoning effort can be set once on an adapter, or overridden for one agent:\n\n```python\nllm = OpenAILLM(model=\"gpt-5\", reasoning_effort=\"high\")\n\nresearcher = AgentConfig(\n agent_id=\"researcher\",\n role=\"deep research\",\n system_prompt=\"Research carefully.\",\n allowed_tools=[\"web_search\"],\n reasoning_effort=\"medium\", # overrides the adapter default for this agent\n)\n```\n\nThe adapter includes the configured value on every provider request. OpenAI\nsupports `none`, `minimal`, `low`, `medium`, `high`, and `xhigh`; Anthropic\nsupports `low`, `medium`, `high`, `xhigh`, and `max` on compatible models.\nUnsupported provider/value combinations fail before the request is sent.\n\nCredential-backed adapters can also plug into the same contract. This is the\nshape used for provider-specific subscription or OAuth flows without teaching\nagents about auth:\n\n```bash\nagent-harness login openai-codex\nagent-harness auth status openai-codex\nagent-harness login claude-code\nagent-harness auth status claude-code\n```\n\n> **⚠️ Subscription adapters are experimental — use the metered API in production.**\n>\n> `OpenAICodexLLM` and `ClaudeCodeLLM` bridge **ChatGPT / Claude\n> subscription OAuth credentials** into the harness by talking to\n> internal CLI endpoints with CLI-shaped User-Agent and billing headers.\n> This route:\n>\n> - **May violate OpenAI's and Anthropic's Terms of Service.** Both\n> providers prohibit using subscription accounts (ChatGPT Plus/Pro,\n> Claude Pro/Max) for arbitrary programmatic access — subscriptions\n> price for the official CLI's intended use only.\n> - **May result in account suspension** if abuse detection classifies\n> harness traffic as misuse.\n> - **Depends on undocumented internal endpoints**\n> (`/backend-api/codex/responses`, the Anthropic Messages API with\n> `claude-code-*` beta flags) that providers can change or revoke at\n> any time.\n>\n> **Use these adapters only for personal research on accounts you own.**\n> Do not use them to serve other users. For anything else, prefer the\n> metered API path:\n>\n> - `OpenAILLM` with `OPENAI_API_KEY` (optionally routed through a\n> gateway like LiteLLM/Helicone for cost headers).\n> - The standard Anthropic Messages API with an Anthropic API key.\n\nDirect `openai-codex` OAuth follows the Codex/Pi-style ChatGPT\nsubscription route rather than the stable OpenAI Platform API. The\nCodex OAuth client id can be overridden with\n`AGENT_HARNESS_OPENAI_CODEX_CLIENT_ID`.\n\n```python\nfrom harness.llm.openai_codex import OpenAICodexLLM\n\nllm = OpenAICodexLLM(\n model=\"gpt-5.5\",\n auth_file=\"~/.agent-harness/auth/auth.json\", # Pi-shaped openai-codex OAuth entry\n)\nruntime = AgentRuntime(..., llm=llm)\n```\n\n`OpenAICodexLLM` calls the Codex backend directly\n(`https://chatgpt.com/backend-api/codex/responses`) with OAuth credentials.\nThe stable fallback remains `OpenAILLM` with `OPENAI_API_KEY`.\n\nFor Claude Code-style setups, use `ClaudeCodeLLM` with Claude Pro/Max OAuth\ncredentials stored in the same auth file. It calls the Anthropic Messages API\ndirectly with Claude-Code-compatible OAuth headers:\n\n```bash\nagent-harness login claude-code\npython examples/subscription_auth_demo.py claude-code\n```\n\n```python\nfrom harness.llm.claude_code import ClaudeCodeLLM\n\nllm = ClaudeCodeLLM(\n model=\"claude-sonnet-4-6\",\n auth_file=\"~/.agent-harness/auth/auth.json\",\n)\n```\n\n### Cost shaping + reliability\n\nTwo patterns, ordered by how production teams actually solve this:\n\n**1. Per-call-site LLM injection (the recommended pattern)**\n\n`AgentRuntime` exposes one slot per orchestrator call site. Each defaults to\n`llm` when unset, so existing code keeps working. The classifier and router\nboth see only the goal + agent descriptions (~300 tokens) and emit a\none-token decision — natural candidates for a cheaper model. The planner\nand synthesiser produce structured DAGs and final answers and usually want\nto stay on the main model.\n\n```python\nruntime = AgentRuntime(\n agent_registry=agents,\n tool_registry=tools,\n memory=memory,\n llm=premium, # default — agent ReAct loops use this\n classifier_llm=cheap, # simple vs complex dispatch decision\n router_llm=cheap, # single-agent picker\n # planner_llm=... # defaults to llm; override only if you want\n # synthesizer_llm=... # defaults to llm\n)\n```\n\nNo guessing, no keyword matching, no fragility — you read the runtime\nconstruction and you know exactly which model serves which purpose. The\nbudget guard is wired into every distinct LLM instance automatically\n(deduped by object identity, so injecting the same wrapper into multiple\nslots costs no extra calls).\n\n**2. `FallbackLLM` for resilience**\n\nTry each adapter in order; transparently switch to the next on rate\nlimits, timeouts, or 5xx errors:\n\n```python\nfrom harness.llm.fallback import FallbackLLM\n\nllm = FallbackLLM([\n AnthropicLLM(model=\"claude-sonnet-4-6\"), # primary\n OpenAILLM(model=\"gpt-4o-mini\"), # backup\n])\nruntime = AgentRuntime(..., llm=llm)\nprint(llm.last_route) # 0 if primary worked, 1 if backup did\n```\n\nPermanent errors (auth, bad request) propagate immediately — only transient\nupstream errors trigger fallback. Customise with `transient_errors=...`.\nStreaming retries only fire before the first token; mid-stream failures\npropagate to preserve response integrity.\n\n**3. `RoutingLLM` for bring-your-own-selector cases**\n\nWhen you need runtime routing — capability gating (`vision` vs\n`long_context`), learned classifiers (RouteLLM-style), cascade\nrouting (cheap-then-escalate-on-low-confidence) — wrap a routes dict\nwith your own selector callable:\n\n```python\nfrom harness.llm.routing import RoutingLLM\n\ndef by_capability(system, messages):\n if _needs_vision(messages):\n return \"vision\"\n if _estimated_tokens(system, messages) > 100_000:\n return \"long_context\"\n return \"default\"\n\nllm = RoutingLLM(\n routes={\n \"default\": OpenAILLM(model=\"gpt-4o-mini\"),\n \"vision\": OpenAILLM(model=\"gpt-4o\"),\n \"long_context\": AnthropicLLM(model=\"claude-sonnet-4-6\"),\n },\n selector=by_capability,\n default_route=\"default\",\n)\n```\n\nThe harness intentionally does not ship default selectors. Naive selectors\n(keyword matching, fixed token thresholds) misroute in subtle ways and\nencourage the wrong mental model — if you're reaching for one, you almost\ncertainly want per-call-site injection instead.\n\nCompose freely: `FallbackLLM([premium, backup])` injected into the\n`llm=` slot gives the agent loops resilience, with `classifier_llm=cheap`\nand `router_llm=cheap` shaping the cheap-call cost — all without a custom\nselector.\n\n---\n\n`ClaudeCodeLLM` reads a `claude-code` OAuth entry, refreshes it automatically\nwhen expired, and retries once after `401`/`403`. This mirrors Pi's Claude\nPro/Max extension approach rather than shelling out to the Claude CLI. The\ndefault model is the current canonical Sonnet release ID, `claude-sonnet-4-6`;\nset `CLAUDE_CODE_MODEL` or pass `model=\"claude-opus-4-7\"` to choose another\nmodel.\n\nBoth adapters stream incrementally — `stream_complete()` yields each\nSSE delta token as it arrives, and `complete()` consumes the same\nstream and returns the concatenated text once finished. Cost / token\nusage is captured from the final stream event into `last_usage`.\n\nThe Claude billing header's `cc_version` is read from\n`CLAUDE_CODE_VERSION` (env) or from `claude --version` if the CLI is\ninstalled; falls back to `unknown` otherwise. Pinning a specific\nversion with `CLAUDE_CODE_VERSION=2.1.150` is recommended if you want\nstable behavior across CLI upgrades.\n\nDo not copy browser/app refresh tokens into repo files. Store OAuth auth files\nunder `~/.agent-harness/auth` or reuse an existing Pi auth file with private\nfile permissions (`0600`).\n\nTo use Anthropic / Gemini / Ollama / a local SGLang or vLLM server / anything\nelse — write a 30-line adapter implementing those two methods. See\n`harness/llm/openai.py` for the reference shape; the harness never imports a\nprovider SDK directly.\n\n## Built-in tools\n\nOne tool ships out of the box — `HTTPFetch`, intentionally boring. Anything\nheavier (auth, retries, connection pooling, scraping) belongs in a tool you\nwrite yourself. Tools just need `.name` and `async def execute(**kwargs)`.\n\n```python\nfrom tools.builtin.http_fetch import HTTPFetch\n\ntools.register(HTTPFetch(max_bytes=64 * 1024)) # body capped at 64 KiB\n```\n\nReturns `{status, content_type, body, truncated, url}`. Errors come back as\n`{\"error\": \"...\"}` so the agent treats them as observations rather than\ncrashing the loop.\n\n## Memory write timing\n\n- **During run**: `write_working_fact()` — lightweight KV, namespaced, short TTL\n- **End of run**: `write_run_end()` — LLM extraction → global semantic + episodic vector\n\n### Memory reconciliation (default-on)\n\n`write_run_end` runs the LLM-arbitrated reconciler by default: instead of\nextract-and-overwrite, the LLM sees existing relevant memory + new evidence\nand emits a plan of per-fact actions (`ADD` / `UPDATE` / `MERGE` / `DELETE`\n/ `NOOP`). Same call count as the legacy extraction step; the prompt is\nlarger only when there's existing context to reconcile against.\n\n```python\nmanager = MemoryManager(\n semantic_store=…,\n episodic_store=…,\n llm=…,\n reconcile_on_write=True, # default — set False for legacy extract path\n allow_destructive_reconcile=False, # default — DELETE actions demoted to NOOP\n auto_compact_threshold={\"agent_task\": 20}, # optional — fire compact()\n # when an agent accumulates this\n # many task episodes\n)\n```\n\n`allow_destructive_reconcile=False` keeps the LLM from removing data unless\nyou've vetted that DELETE actions are sensible for your workload — demoted\ndecisions land in `manager.get_conflict_log()` so you can audit.\n\n`manager.compact(goal=\"…\", agent_id=\"…\")` is the same primitive with no\nnew evidence — a pure cleanup pass that consolidates accumulated episodes\nand prunes redundant facts. Triggered automatically by\n`auto_compact_threshold`, or call it explicitly.\n\nEpisodic supersede is now **hard-delete** (no `active=False` tombstones\naccumulating per run): `memory_policy=\"latest\"` writes and reconciler\n`DELETE` actions both remove rows.\n\nIf the LLM returns a response that doesn't parse as a reconcile plan\n(older / smaller models that don't follow the multi-action schema),\n`write_run_end` silently falls back to the legacy extract-and-overwrite\npath — no crash, no missed run-end write.\n\n### Tool-result caching (opt-in, per run)\n\n`AgentConfig.cache_tool_results = True` memoizes tool calls within a single\nrun, keyed by `(tool_name, args)`. Useful for multi-agent runs where agents\nredo each other's idempotent reads (`HTTPFetch` on stable URLs,\n`kubectl get ...` discovery, MCP filesystem reads).\n\nA tool can veto caching for itself with `cacheable = False` on the\ninstance — required for anything with side effects or time-dependent\noutput. Errors are never cached (a transient failure shouldn't poison the\nrest of the run).\n\n```python\nclass HTTPFetch:\n name = \"http_fetch\"\n cacheable = True # default; explicit for clarity\n\nagents.register(AgentConfig(\n agent_id=\"web\",\n ...,\n cache_tool_results=True,\n))\n```\n\n### Tool observation budget\n\nTool results emitted on the event stream stay intact, but the observation\ntext appended back into the agent's working memory is capped by\n`AgentConfig.max_observation_chars` (default `20_000`). This prevents a\nsingle browser snapshot, MCP response, or shell output from dominating the\nnext LLM call. When capped, the model sees a structured JSON envelope with\n`truncated`, `original_chars`, `shown_chars`, `omitted_chars`, and `content`.\n\nSet `max_observation_chars=0` to disable this cap for agents that must feed\nfull raw tool output back into the next step.\n\nDefaults are in-memory (`InMemorySemanticStore`, `InMemoryEpisodicStore`).\nFor durable storage:\n\n```python\n# Semantic: Redis\nimport redis.asyncio as redis\nfrom memory.redis_store import RedisSemanticStore\n\nclient = redis.Redis(host=\"localhost\", decode_responses=True)\nsemantic = RedisSemanticStore(client, key_prefix=\"agent-harness:\")\n\n# Episodic: LanceDB\nfrom memory.episodic_lance import LanceDBEpisodicStore\nepisodic = LanceDBEpisodicStore(db_path=\"./lance_episodic\")\n\nmemory = MemoryManager(semantic_store=semantic, episodic_store=episodic, llm=llm)\n```\n\nOr write your own backend conforming to the `SemanticStore` / `EpisodicStore`\nprotocols in `memory/manager.py`.\n\n## Execution paths\n\n### Recommended: `dispatch` / `dispatch_stream`\n\nOne call. The harness classifies the goal, picks the right path, and runs it.\n\n```python\nfrom harness.events import EventType\n\nasync for event in runtime.dispatch_stream(\"investigate the GPU spike on worker-07\"):\n if event.type == EventType.DISPATCH:\n print(f\"complexity={event.payload['complexity']} path={event.payload['path']}\")\n elif event.type == EventType.ROUTE:\n print(f\"→ {event.payload['agent_id']}: {event.payload['rationale']}\")\n elif event.type == EventType.ACTION:\n print(f\" tool: {event.payload['tool']}\")\n elif event.type == EventType.DONE:\n print(event.payload[\"answer\"])\n elif event.type == EventType.TASK_DONE:\n print(event.payload[\"answer\"]) # routed path ends here, no DONE\n\nresult = await runtime.dispatch(\"what is the capital of France?\") # blocking\n```\n\n**Classification logic:**\n- 1 agent registered → always `\"simple\"`, no LLM call made.\n- Multiple agents → one cheap LLM call classifies `\"simple\"` (one agent handles\n it end-to-end) or `\"complex\"` (benefits from task decomposition and specialist\n routing). Unknown response defaults to `\"simple\"`.\n- `\"simple\"` path: LLM router picks the best agent → `run_routed_stream`.\n- `\"complex\"` path: planner decomposes into a DAG → `run_stream`.\n\n---\n\nFour lower-level paths are available when you need explicit control:\n\n### 1. Routed — `run_routed` / `run_routed_stream`\n\nOne lightweight LLM call picks the best agent, then that agent runs its\nReAct loop directly. No task decomposition, no synthesis step. Use this\nfor single-turn goals where one agent handles everything end-to-end.\n\n```python\nfrom harness.events import EventType\n\nasync for event in runtime.run_routed_stream(\"what is the current disk usage?\"):\n if event.type == EventType.ROUTE:\n print(f\"→ {event.payload['agent_id']}: {event.payload['rationale']}\")\n elif event.type == EventType.ACTION:\n print(f\" tool: {event.payload['tool']}\")\n elif event.type == EventType.TASK_DONE:\n print(event.payload[\"answer\"])\n```\n\nFast-path: if only one agent is registered, `route()` returns it immediately\nwithout an LLM call.\n\n### 2. Direct — `run_agent` / `run_agent_stream`\n\nYou name the agent; it runs its ReAct loop directly. Use when you already\nknow which agent to call and want to skip routing entirely.\n\n```python\nresult = await runtime.run_agent(\"researcher\", \"summarise this document\")\n```\n\n### 3. Orchestrated — `run` / `run_stream`\n\nThe planner decomposes the goal into a task DAG, assigns tasks to specialist\nagents, runs them (in parallel where dependencies allow), and synthesises a\nfinal answer. Use for multi-agent goals where different specialists handle\ndifferent parts of the work.\n\n```python\nasync for event in runtime.run_stream(\"investigate GPU spike on worker-07\"):\n if event.type == EventType.PLAN:\n for t in event.payload[\"plan\"][\"tasks\"]:\n print(f\" {t['id']}@{t['agent_id']}: {t['instruction']}\")\n elif event.type == EventType.REPLAN:\n print(f\"[replan #{event.payload['replan_count']}]\")\n elif event.type == EventType.DONE:\n print(event.payload[\"answer\"])\n```\n\n### 4. Pre-built — `run_with_plan` / `run_with_plan_stream`\n\nSupply a hand-written `Plan` and bypass the LLM planner entirely. Use\nthis for deterministic, repeatable workflows where the decomposition is\nknown upfront — CI pipelines, ETL jobs, scheduled tasks. The plan is\nvalidated against registered agents before execution; everything\ndownstream (parallel batches, replan-on-failure, synthesis, memory\nwrites, steering) is identical to `run_stream`.\n\n```python\nfrom orchestrator.planner import Plan, Task\n\nplan = Plan([\n Task(\"t1\", \"analyst\", \"Analyse error logs from the last hour\"),\n Task(\"t2\", \"reporter\", \"Write an incident summary\", depends_on=[\"t1\"]),\n])\n\n# streaming\nasync for event in runtime.run_with_plan_stream(plan, goal=\"Incident report\"):\n if event.type == EventType.DONE:\n print(event.payload[\"answer\"])\n\n# blocking\nresult = await runtime.run_with_plan(plan, goal=\"Incident report\")\n```\n\nThe `goal` string is passed to the synthesiser and used for memory\ncontext injection into agents — even though the plan shape is fixed, the\nagents themselves still read from memory.\n\nIf a task fails mid-run and `on_failure=\"replan\"`, the replan call does\ngo to the LLM — the bypass is for the *initial* plan only.\n\n### 5. Sub-agents as tools — `SubAgentTool`\n\nA different decomposition model from the orchestrator. Instead of a\nseparate planner LLM deciding the DAG upfront, **one main agent's ReAct\nloop decides per step whether to delegate** and to whom. Use when the\npath is exploratory — you don't know which specialist you'll need until\nyou've seen partial results.\n\n```python\nfrom tools.builtin.subagent import SubAgentTool\n\nresearcher = BaseAgent(config=researcher_config, ...)\nanalyst = BaseAgent(config=analyst_config, ...)\n\nmain_agent = BaseAgent(\n config=AgentConfig(\n agent_id=\"coordinator\",\n role=\"decides what to research and analyses results\",\n system_prompt=\"...\",\n allowed_tools=[\"delegate_research\", \"delegate_analyse\"],\n max_subagent_depth=3, # bounds the delegation chain\n ),\n tools={\n \"delegate_research\": SubAgentTool(researcher),\n \"delegate_analyse\": SubAgentTool(analyst),\n },\n ...\n)\n\nasync for event in main_agent.run_stream(goal):\n # Sub-agent events bubble up tagged with event.parent_agent_id so\n # renderers can indent or group them.\n # SUBAGENT_START fires before the sub's first event (payload:\n # task, invocation_id).\n # SUBAGENT_DONE fires after the sub's TASK_DONE (payload: success,\n # steps, confidence, answer, error, invocation_id).\n ...\n```\n\nWhen the main agent's LLM emits `{\"action\": \"delegate_research\", \"args\":\n{\"task\": \"...\"}}`, the wrapped sub-agent runs its own ReAct loop with a\n**fresh `WorkingMemory`**; the sub's final answer becomes the main agent's\nnext observation. Each delegation = a fresh sub run; cross-delegation\ncontinuity flows through long-term memory (`MemoryManager.build_context(\nagent_id=…)`), not through WM carry-over — same model as the\norchestrator's per-task agents.\n\n**Crash-resume.** A delegation mints its own `run_id`, which reaches the\nparent on the `subagent_start` event and is recorded on the action. If the\nprocess dies mid-delegation, resuming the parent continues the sub-agent from\nits own checkpoint instead of re-running it. Resuming is best-effort: with no\ncheckpoint store, no stored state, or a pre-0.13 checkpoint, the delegation\nsimply runs again. Execution is at-least-once either way — the step in flight\nwhen the checkpoint was written is replayed, which `ActionState.attempts`\nrecords.\n\nParallel delegation works via the existing `actions: [...]` shape:\n\n```json\n{\n \"thought\": \"research and analyse can run in parallel\",\n \"actions\": [\n {\"tool\": \"delegate_research\", \"args\": {\"task\": \"find baseline metrics\"}},\n {\"tool\": \"delegate_analyse\", \"args\": {\"task\": \"score recent incidents\"}}\n ]\n}\n```\n\nBoth sub-agent streams interleave through a fan-in helper\n(`harness/streaming.py:fan_in`) so the parent's event stream stays a\nsingle sequence even when multiple sub-agents are working concurrently.\n\n**Sub-agents as tools vs. Orchestrator — which to pick?**\n\n| | Sub-agents as tools | Orchestrator |\n|---|---|---|\n| Plan timing | Per step (dynamic) | Upfront DAG |\n| Who plans | The main agent's LLM | A separate planner LLM |\n| Best for | Exploratory work, \"I don't know what I need until I see partial results\" | Known workflows (audits, ETL, scheduled jobs) |\n| Replan-on-failure | Implicit — main agent reacts to sub's failure | Explicit `on_failure=\"replan\"` |\n| Recursion guard | `max_subagent_depth` on `AgentConfig` | N/A — DAG is flat |\n\nBoth are first-class. Most real systems combine them — the orchestrator\nplans a high-level DAG, individual tasks within it use sub-agent tools\nfor finer dynamic decomposition.\n\n### 6. Persistent chat sessions — `PersistentAgent`\n\n`PersistentAgent` is a wrapper around a coordinator `BaseAgent`, not a new\nagent constructor. Build agents, sub-agents, MCP tools, and auth exactly as\nusual; then wrap the top-level coordinator to add durable chat/session state.\n\n```python\nfrom harness.persistent import PersistentAgent, SQLiteSessionStore\n\napp = PersistentAgent(\n coordinator=coordinator_agent,\n session_store=SQLiteSessionStore(\"~/.agent-harness/sessions.sqlite\"),\n memory=memory_manager,\n llm=llm,\n llm_registry={\n \"fast\": lambda: OpenAILLM(model=\"gpt-5.4-mini\"),\n \"deep\": lambda: OpenAILLM(model=\"gpt-5\"),\n },\n default_model=\"fast\",\n)\n\nasync for event in app.chat(\"I like the above; can you do X?\", session_id=\"thread-1\"):\n renderer.render(event)\n```\n\nUse `app.capabilities()` to inspect the already-wired coordinator,\nsub-agents, and MCP tools. The demo exposes this with\n`--show-capabilities`.\n\nModel switching is session-scoped when an `llm_registry` is configured:\n\n```python\nawait app.switch_model(\"thread-1\", \"coordinator\", \"deep\")\nawait app.clear_model_override(\"thread-1\", \"coordinator\")\n```\n\nRegistry values are zero-argument factories. Model names `default`,\n`reset`, and `clear` are reserved for clearing overrides in interactive\ncontrols.\n\nThe durable session stores only `{agent_id: model_name}` overrides; the\ntranscript is unchanged. At the start of each turn, `PersistentAgent`\nresets agents to their construction-time LLMs, then applies that session's\noverrides. The facade assumes one active chat turn at a time for a given\n`PersistentAgent` instance.\n\nPersistent sessions also expose a small control surface for user interfaces:\n\n```python\nstate = await app.session_state(\"thread-1\")\nprint(state.tokens_in_total, state.tokens_out_total)\nsessions = await app.list_sessions()\nmatches = await app.list_sessions(query=\"research\")\ncached = app.cached_memory_context(\"thread-1\")\nawait app.save_to_memory(\"thread-1\") # reconcile pending turns, keep cache warm\nawait app.force_compact(\"thread-1\") # structural reorg (summary + trim + reconcile + evict)\napp.forget_memory_cache(\"thread-1\")\nawait app.clear_session(\"thread-1\") # clear transcript only; memory retained\nawait app.delete_session(\"thread-1\") # delete transcript only; memory retained\n```\n\nThere is deliberately no `close()` method — the SQLite transcript is the\ndurable record, and any of `chat`/`save_to_memory`/`force_compact` on the\nsame `session_id` later will resume from it. To exit, just stop calling\n`chat` (and let the process end).\n\nFor simple REPLs, `PersistentCommandHandler` maps slash commands to those\nprimitives and returns structured results:\n\n```python\nfrom harness.persistent_controls import PersistentCommandHandler\n\ncontrols = PersistentCommandHandler(app)\nresult = await controls.handle(\"/sessions research\", session_id=session_id)\nif result.session_id:\n session_id = result.session_id\nif result.text:\n print(result.text)\n```\n\nThe command surface is defined once in `SLASH_COMMAND_SPECS` and exposed via\n`slash_command_specs()`. Tab-completion in the demo binds those specs to\n`prompt_toolkit` via `SlashCommandCompleter`:\n\n```python\nfrom prompt_toolkit import PromptSession\nfrom prompt_toolkit.history import FileHistory\nfrom harness.persistent_completion import SlashCommandCompleter\n\nsession = PromptSession(\n history=FileHistory(\"~/.agent-harness/demo_history\"),\n completer=SlashCommandCompleter(app, session_id_provider=lambda: session_id),\n complete_while_typing=False, # Tab-triggered; doesn't query the store every keystroke\n)\nmessage = (await session.prompt_async(\"> \")).strip()\n```\n\nName completion comes from the spec registry; session-id arguments call\n`app.list_sessions(query=...)` (SQLite pushes the filter down to\n`lower(session_id) LIKE ?` so per-keystroke cost stays bounded). Other UIs\n(web autocomplete, fzf, IDE plugins) should consume `slash_command_specs()`\ndirectly without importing the prompt_toolkit binding.\n\nThe demo uses that utility for:\n\n- `/capabilities`, `/agents`, `/mcp` inspect wired agents and tools\n- `/session` shows turns, context-pressure estimate, reconcile cadence, and summary\n- `/usage` shows persisted total and last-run provider-reported token usage\n- `/models` lists switchable model names when an LLM registry is configured\n- `/model [agent_id] [model|default]` shows or changes a session-scoped model override\n- `/background ` launches a sub-agent task in the background for the current process\n- `/tasks [collect|cancel] [id|all]` lists background tasks, appends completed results to the transcript, or cancels running work\n- `/sessions [query]` lists known session ids, optionally filtered by id text\n- `/memory` shows the cached per-session memory context\n- `/save` flushes turns after the last reconcile checkpoint into long-term memory **without** evicting the cached prior (foreground prefix stays warm)\n- `/compact` forces a structural reorg — summary + trim + reconcile + cache evict — used when the transcript is bloating, not for routine \"save\"\n- `/forget` evicts cached memory context so the next turn refetches it\n- `/plan [on|off]` toggles plan-before-execute mode — see below\n- `/switch ` resumes or creates a logical session id\n- `/new [id]` starts a fresh session id and refuses to collide with an existing id\n- `/clear` clears the current transcript/summary/counters; long-term memory is retained\n- `/delete [id] confirm` deletes a transcript row; long-term memory is retained\n- `/end` exits the demo — no auto-flush; call `/save` first if you want pending facts persisted\n\nBackground sub-agent tasks are process-local. They keep running while the\nREPL/server process is alive; completed results become durable only when\ncollected into the session transcript with `/tasks collect `.\n\nOverlapping delegations to one sub-agent are safe as of 0.13: each invocation\nruns on its own copy of the tool and its nested agent, so concurrent runs no\nlonger share working memory. What remains your responsibility is whether that\nagent's **tools and LLM adapter** tolerate concurrent use — a tool holding a\nsingle connection or a non-reentrant client still needs its own guarding.\n\nWhen a `PersistentAgent` coordinator is wired with `SubAgentTool`s, it also\ngets LLM-visible background tools automatically:\n\n- `background_delegate_(task)` starts that sub-agent and returns a `task_id`\n- `check_background_task(task_id)` reports status and any available result preview\n- `collect_background_task(task_id)` returns the completed answer to the coordinator and appends it to the transcript\n\nThe regular `delegate_` tools remain blocking; background delegation\nuses separate tool names so the model's intent is explicit in traces.\n\n**Plan mode** (`/plan on`, or **Shift-Tab** at the prompt) gates each turn behind an approval step. The\ncoordinator's LLM produces a structured plan (`{summary, steps[]}`) before\nany tools run; `PersistentAgent.chat` yields a `PLAN_PROPOSED` event so\nthe renderer can display it, then routes approval through\n`harness.hitl.request_plan_approval` — the plan-specific sibling of the\nprimitive that already gates individual tool calls.\n\n**Plan mode approves intent, not arguments.** Most realistic agent tasks\nhave args that can only be known at runtime — a URL discovered by a\nsearch, a file path extracted from a directory listing, a row id from a\nquery. The planner is told to use concrete `args` only when the value\nwas user-supplied or otherwise knowable upfront; for everything else it\nreturns `args: null` (or omits the field), and the renderer shows\n`args: (resolved at runtime)` instead of fabricating placeholder JSON.\nThe approval banner surfaces `dynamic_steps: N` alongside the summary so\nthe reviewer sees how many step args will be filled in during execution.\nThe approved-plan prior tells the executor explicitly that\n`(resolved at runtime)` means *\"derive the argument from your\nobservations of the prior step,\"* not *\"a value you must invent.\"*\n\nPlan approval uses a focused HITL surface:\n\n- `Enter` or `y` approves the plan and runs the ReAct loop with the plan injected as a pinned prior.\n- `n` rejects; an `ERROR` event is yielded and **nothing is written to the session store** (cancelled turn = never happened, same as `Esc`-cancel).\n- Free text becomes a correction — the planner re-runs with the feedback folded into the system prompt. Up to `_PLAN_REVISION_LIMIT` revision cycles before the gate yields a clean \"revision limit reached\" error.\n\nPlan mode is off by default and persists per-session in SQLite, so the\npreference survives process restart (and `/clear`, which is a transcript\nreset, not a settings reset). Esc during plan generation or during the\nHITL wait unwinds `asyncio.CancelledError` through the chat generator\ncleanly — no partial plan written. The setting is on `SessionState.plan_mode_enabled`; flip it programmatically via `PersistentAgent.set_plan_mode(session_id, enabled)`.\n\n**Press `Esc` during a turn to cancel it.** Every example that streams a\n`BusEvent` loop wraps it in `ConsoleRenderer.render_stream(events, *,\nterminal_event_type=...)` — one call that renders each event, captures\nthe top-level terminal (TASK_DONE / DONE), and listens for Esc on stdin.\nThe cancel raises `asyncio.CancelledError` through the agent's `await`\npoints, which propagates out of `PersistentAgent.chat` (and every other\nstream) before `_finalize_turn` runs — so a cancelled turn writes\nnothing to the session store, matching the standard chat-UX \"cancelled\nturn = never happened\" semantic.\n\nThe underlying primitives in `harness/cancellation.py` are reusable for\nany \"run X but stop early on Y\" pattern: `run_until_cancelled(coro, *,\ntrigger)` races a coroutine against an `asyncio.Event`,\n`escape_listener(trigger)` is the prompt_toolkit Esc binding,\n`consume_with_cancel(events, *, on_event)` is the renderer-agnostic\nlower-level form for demos that need bespoke per-event handling (e.g.\n`complex_sysaudit_demo.py` capturing a custom DONE payload). The Esc\nbinding is a no-op on non-TTY stdin, so the same code paths work for\npipe / file / CI input without conditional wiring.\n\nEach chat turn gets a fresh `WorkingMemory`. Continuity comes from the\nSQLite session state (rolling summary + recent messages) and normal\n`MemoryManager` recall, not from carrying old ReAct scratchpads forever.\n\n**Prefix-cache aware.** The full prompt stays byte-identical across plain\nchat turns within a compaction window. Four things make that work:\n\n1. The accumulated session transcript is sent to the coordinator as real\n `user`/`assistant` role messages — not folded into one inline-rendered\n text blob. Each turn extends the message list by exactly the previous\n turn's user+assistant pair plus the new user task.\n2. Memory context (`MemoryManager.build_context` result) lives in a\n pinned user-message prior, **not in the system prompt**. The system\n prompt is now pure agent identity + tool list + ReAct format — purely\n static. Memory context is fetched once per session, cached on the\n `PersistentAgent`, and refreshed only at compaction or explicit\n \"remember\" signals.\n3. **Compaction triggers on context-size pressure, not turn counts.**\n `compact_at_context_fraction=0.5` (default) reads the coordinator\n LLM's `input_token_budget` and fires when the transcript crosses\n ~50% of it. After compaction, `retain_context_fraction=0.15`\n keeps the newest messages that fit in ~15% of the input budget\n verbatim and folds older messages into the rolling summary.\n Chat-only sessions go thousands of turns between\n compactions; browser-heavy sessions compact when budget pressure\n actually warrants it.\n4. **Background memory accumulation, foreground cache stability.**\n Every `async_reconcile_every_turns` turns (default 10), a\n non-blocking `write_run_end` samples the last N turn-pairs from the\n durable transcript and updates long-term memory — but does **not**\n evict the per-session memory cache. New facts are immediately\n visible to OTHER sessions; THIS session sees them at the next\n compaction (where the cache is already breaking for the summary\n refresh).\n\nOpenAI's automatic prefix cache and Anthropic's `cache_control` markers\nboth match on longest-identical prefix. Together these four changes let\na typical session pay one cold compaction when the transcript actually\nfills the context window, with warm cache hits on every turn in\nbetween, while still accumulating memory in the background.\n\nThe legacy `reconcile_every_turns`, `compact_every_turns`, and\n`compact_message_threshold` knobs have been **removed** — passing them\nto `PersistentAgentConfig` is a clear `TypeError` rather than a silent\nno-op so existing user code can be updated explicitly. Reconciliation\nfires on (a) user-explicit `remember` / `prefer` / `always` / etc.\nsignals (immediate, evicts cache), (b) periodic background async path\n(non-blocking, no eviction), or (c) compaction (with the summary LLM\ncall, evicts cache).\n\nThe demo stores local state under `~/.agent-harness` by default:\n\n- `sessions.sqlite` for chat/session state\n- `memory/semantic.sqlite` for semantic facts/preferences\n- `memory/lance_episodic` for searchable episodic summaries\n\nBy default the demo uses `OpenAILLM` and requires `OPENAI_API_KEY`.\nTo run it with stored OpenAI subscription credentials instead:\n\n```bash\npython -m harness.cli login openai-codex\npython examples/persistent_agent_demo.py --provider openai-codex\n```\n\nOr with stored Claude Code credentials:\n\n```bash\npython -m harness.cli login claude-code\npython examples/persistent_agent_demo.py --provider claude-code\n```\n\nThe wrapper owns cadence:\n\n- session transcript is written every turn without an LLM;\n- `MemoryManager.write_run_end(...)` runs synchronously only for explicit\n durable user signals such as \"remember\" / \"prefer\" / \"always\";\n- background async reconciliation samples the durable transcript every\n `async_reconcile_every_turns` turns without evicting the current\n session's memory-context cache;\n- session compaction runs under context pressure and updates the stored\n summary while trimming older transcript messages.\n\nFor MCP, put MCP tools on the coordinator or sub-agents before wrapping.\n`PersistentAgent` does not special-case MCP auth; it preserves the existing\ntool wiring model.\n\n---\n\nEvent types by path:\n\n| Event | Dispatch | Routed | Direct | Orchestrated | Pre-built |\n|---|---|---|---|---|---|\n| `DISPATCH` | ✓ | — | — | — | — |\n| `ROUTE` | ✓ (simple) | ✓ | — | — | — |\n| `THOUGHT` / `TOKEN` / `ACTION` / `OBSERVATION` | ✓ | ✓ | ✓ | ✓ | ✓ |\n| `SUBAGENT_START` / `SUBAGENT_DONE` | ✓ | ✓ | ✓ | ✓ | ✓ |\n| `TASK_DONE` | ✓ | ✓ | ✓ | ✓ | ✓ |\n| `PLAN` / `REPLAN` / `SYNTHESIS` / `DONE` | ✓ (complex) | — | — | ✓ | ✓ |\n| `ERROR` | ✓ | ✓ | ✓ | ✓ | ✓ |\n\n`TOKEN` events fire only when your LLM client exposes\n`async def stream_complete(system, messages) -> AsyncGenerator[str, None]`.\nNon-streaming clients still work — they emit the full response in one\n`THOUGHT` event per step.\n\n## Console rendering\n\n`ConsoleRenderer` handles all `BusEvent` types with consistent label\nand truncation formatting so event-loop boilerplate stays out of your\nscripts.\n\n```python\nfrom harness.console import ConsoleRenderer, trunc\n\nrenderer = ConsoleRenderer(\n truncate=140, # max chars for long text fields\n sep_char=\"─\", # separator character\n sep_width=72, # separator width\n agent_label_width=16, # width of [agent_id] column\n show_tokens=False, # True to print TOKEN events inline\n)\n\nasync for event in runtime.dispatch_stream(goal):\n renderer.render(event) # handles every EventType\n```\n\nFor events with custom section headers (e.g. a \"PROJECT HEALTH REPORT\"\nblock), handle that event yourself and skip `render` for it — the\nrenderer is additive:\n\n```python\nasync for event in runtime.run_stream(goal):\n if event.type == EventType.DONE:\n renderer.sep(\"═\")\n print(\"MY CUSTOM HEADER\")\n renderer.sep(\"═\")\n print(event.payload[\"answer\"])\n else:\n renderer.render(event)\n```\n\n`trunc(s, n)` is exported for standalone use when you need to truncate\na string to `n` characters with a trailing `…`.\n\n## Working memory budget\n\n`AgentConfig.working_memory_max_tokens` controls per-agent eviction. **Default\nis `None` — auto-derived from the LLM's context window at runtime** so the\nthreshold adapts when you swap models (a 128K `gpt-5.4-mini` gets ~99K of WM\nheadroom; a 200K `claude-sonnet-4-6` gets ~159K; a tiny 8K fallback gets ~6K).\nConcretely the WM compacts at `0.8 × llm.input_token_budget`, leaving 20%\nheadroom for system prompt, memory context, tool schemas, and tokeniser\nvariance.\n\nEach shipped adapter (`OpenAILLM`, `AnthropicLLM`, `ClaudeCodeLLM`,\n`OpenAICodexLLM`) exposes `context_window` and `input_token_budget`\nproperties driven by a per-provider lookup table. For models the table\ndoesn't know (new releases, fine-tunes), pass `context_window=N` explicitly:\n\n```python\nllm = OpenAILLM(model=\"gpt-6-preview\", context_window=256_000)\n# or hard-cap WM independent of the LLM:\nAgentConfig(..., working_memory_max_tokens=16_000)\n```\n\nCounting defaults to a `chars/4` heuristic (stable for code/JSON/text within\n~10–20% of real BPE counts, zero deps). For exact counts plug your own\ncounter into `WorkingMemory` directly:\n\n```python\nimport tiktoken\nenc = tiktoken.get_encoding(\"cl100k_base\")\nwm = WorkingMemory(llm=..., token_counter=lambda s: len(enc.encode(s)))\n```\n\n## Cost tracking\n\nThe harness deliberately **does not maintain a pricing table** — tables go\nstale, and per-org rollups don't belong in an SDK. Token counts come from the\nprovider authoritatively (free). Dollars are optional and follow this\nprecedence, per call:\n\n**1. Gateway header (recommended).** Route the OpenAI client through LiteLLM\nproxy, Helicone, OpenRouter via `base_url=...` and `OpenAILLM` reads\n`x-litellm-response-cost` / `x-cost-usd` / `x-helicone-cost-usd` from the\nresponse automatically. Real cost, gateway-maintained pricing.\n\n```python\nllm = OpenAILLM(model=\"gpt-4o-mini\", base_url=\"https://my-litellm/v1\")\n```\n\n**2. `cost_fn(usage) -> float`** — caller-supplied. For when you hit a\nprovider directly and want a local pricing function:\n\n```python\ndef my_pricing(usage):\n rates = {\"gpt-4o-mini\": (0.15e-6, 0.60e-6)} # (input, output) USD/token\n served = usage.get(\"model\", \"\")\n for prefix, (in_rate, out_rate) in rates.items():\n if served.startswith(prefix):\n return usage[\"tokens_in\"] * in_rate + usage[\"tokens_out\"] * out_rate\n return 0.0\n\nllm = OpenAILLM(model=\"gpt-4o-mini\", cost_fn=my_pricing)\n```\n\n**3. Neither.** Tokens still flow on every call (visible via `last_usage` and\nin OTEL span attributes when OTEL is enabled). `cost_usd` is just omitted. No\ncrash, no surprise charges from a stale rate table.\n\n`AgentRuntime` wires a fresh `BudgetGuard` per run and calls\n`llm.set_budget(guard)` automatically (duck-typed; safe if the adapter\ndoesn't support it). When `max_total_cost_usd` is exceeded, the next\n`BudgetGuard.check()` raises and aborts the run. The final `DONE` event\npayload carries `cost_usd` and `elapsed_seconds` at the top level:\n\n```python\nasync for event in runtime.dispatch_stream(goal):\n if event.type == EventType.DONE:\n print(f\"${event.payload['cost_usd']:.4f} in {event.payload['elapsed_seconds']:.1f}s\")\n```\n\nCost ceiling fires on the *next* `check()` (start of next ReAct step or\norchestrator batch), not synchronously mid-call — accept this for 0.0.1, the\nguard's job is preventing runaway loops, not bounding individual calls.\n\n### Token limits + per-call-site breakdown\n\n`GuardrailConfig.max_input_tokens` / `max_output_tokens` cap raw token\nusage independently of dollar cost. This is the only enforcement available\nto subscription-auth runs (`ClaudeCodeLLM`, `OpenAICodexLLM`) — those tiers\ndon't expose pricing, so cost stays 0 and only token caps can fire.\n\n```python\nruntime = AgentRuntime(\n ...,\n guardrail_config=GuardrailConfig(\n max_total_cost_usd=2.0,\n max_input_tokens=100_000,\n max_output_tokens=20_000,\n ),\n)\n```\n\nPer-call-site attribution lives on the terminal event's `budget` payload\n— a snapshot of spending bucketed by the LLM slot that ran each call.\nThe runtime tags classifier / router / planner / synthesizer calls\nautomatically; ReAct agent calls go into the totals but don't get a\nbucket. So `cheap` (used for both `classifier_llm` and `router_llm`) and\n`premium` (used for `planner_llm`) report separately even though one is\nthe same physical LLM instance shared across slots:\n\n```python\nasync for event in runtime.dispatch_stream(goal):\n # Routed (simple) goals terminate with TASK_DONE; orchestrated goals\n # with DONE. Both carry the same ``budget`` shape.\n if event.type in (EventType.TASK_DONE, EventType.DONE):\n budget = event.payload[\"budget\"]\n print(f\"total: in={budget['tokens_in']} out={budget['tokens_out']} \"\n f\"${budget['cost_usd']:.4f}\")\n for slot, stats in budget[\"breakdown\"].items():\n print(f\" {slot}: in={stats['tokens_in']} out={stats['tokens_out']}\")\n```\n\nThe same `budget` dict is attached to `runtime.run(...)` and\n`runtime.dispatch(...)` return values under the `budget` key, so blocking\ncallers don't need to read events.\n\nAnthropic / Claude Code adapters count input tokens as the *total* that\nhit the wire (non-cached + cache-creation + cache-read), so token caps\nreflect actual consumption regardless of cache hit rate. Cost calculation\nvia `cost_fn` still respects cache pricing.\n\n### Evals via the trace recorder\n\nThere's no shipped evals framework — opinions on scorers, judge models,\nand golden-set management belong outside the orchestration core. The\n[trace recorder](#trace-recorder--replay--local-viewer) already writes\nper-event token/cost/latency to JSONL, so a few lines of glue cover most\nin-house eval setups:\n\n```python\nimport json\nfrom harness.trace import record_trace\n\n# 1. Record traces while running a fixture set.\nfor fixture in fixtures:\n async for _event in record_trace(\n runtime.dispatch_stream(fixture[\"input\"]),\n path=f\"runs/{fixture['id']}.jsonl\",\n ):\n pass\n\n# 2. Score offline by replaying.\ndef score_run(path: str, expected: str) -> dict:\n answer = \"\"\n budget = {\"tokens_in\": 0, \"tokens_out\": 0, \"cost_usd\": 0.0, \"breakdown\": {}}\n for line in open(path):\n event = json.loads(line)\n if event[\"type\"] in (\"done\", \"task_done\"):\n answer = event[\"payload\"].get(\"answer\", \"\")\n budget = event[\"payload\"].get(\"budget\", budget)\n return {\n \"success\": expected.lower() in answer.lower(),\n **budget, # tokens_in, tokens_out, cost_usd, breakdown\n }\n```\n\nPlug in your own scorer (exact-match, LLM-judge, semantic similarity) on\ntop. External tools like Braintrust, LangSmith, and Weave are\npurpose-built for this and ingest the same JSONL shape directly.\n\n## Tool execution\n\nTools that shell out (`kubectl`, `curl`, `sh -c …`) should not run inside the\nagent process. `ExecutorBridge` provides a controlled subprocess launcher with\ntwo backends.\n\n### What every backend enforces\n\n- **Tool allowlist** — set at startup; the LLM cannot extend it.\n- **Wall-clock timeout** per call.\n- **Output size cap** (default 1 MiB, configurable).\n\n### `backend=\"none\"` (default) — Rust executor\n\nRoutes each call through the compiled Rust binary at `executor/`. Adds\nprocess-level isolation (a tool crash cannot reach the agent) and a scrubbed\nenvironment (only `PATH` is forwarded). Does **not** provide syscall filtering,\nfilesystem namespacing, or network isolation.\n\n```bash\ncargo install --path executor # installs ah-executor to ~/.cargo/bin\n```\n\n```python\nfrom harness.executor_bridge import ExecutorBridge, ExecutorConfig, ExecutorTool\n\n# binary_path auto-discovered from PATH via shutil.which(\"ah-executor\")\nbridge = ExecutorBridge(ExecutorConfig(\n allowed_tools=(\"kubectl\", \"curl\"), # \"shell\" is opt-in only\n))\n\nkubectl_tool = ExecutorTool(\"kubectl\", \"kubectl\", bridge, arg_key=\"args\")\n```\n\n### `backend=\"docker\"` — real OS-level isolation\n\nEach tool call runs in a fresh Docker container. Provides network isolation,\nread-only filesystem, and memory/CPU limits. The Rust binary is not used in\nthis mode. Requires Docker daemon on the host.\n\n```python\nfrom harness.executor_bridge import ExecutorBridge, ExecutorConfig, ExecutorTool\n\nbridge = ExecutorBridge(ExecutorConfig(\n allowed_tools=(\"kubectl\",),\n backend=\"docker\",\n docker_image=\"bitnami/kubectl:latest\",\n docker_network=\"none\", # no outbound network\n docker_memory=\"256m\",\n docker_cpus=\"1.0\",\n docker_read_only=True,\n))\n\nkubectl_tool = ExecutorTool(\"kubectl\", \"kubectl\", bridge, arg_key=\"args\")\n```\n\nThe `shell` tool (dict-style args) works with both backends:\n\n```python\nshell_tool = ExecutorTool(\"shell\", \"shell\", bridge)\n# LLM calls: {\"action\": \"shell\", \"args\": {\"cmd\": \"jq '.name' data.json\"}}\n```\n\n## Tests\n\n```bash\npytest\n```\n\n## MCP Tools\n\nConnect any [MCP](https://modelcontextprotocol.io)-compatible server and its\ntools become available to agents — no wrapper code needed.\n\n```bash\npip install -e \".[mcp]\"\n```\n\n```python\nfrom mcp import StdioServerParameters\nfrom tools.mcp import MCPServerConnection\n\nparams = StdioServerParameters(\n command=\"npx\",\n args=[\"-y\", \"@modelcontextprotocol/server-filesystem\", \"/tmp\"],\n)\n\nasync with MCPServerConnection(params, server_name=\"filesystem\") as conn:\n conn.register_tools(tool_registry) # bulk-register all discovered tools\n agents.register(AgentConfig(\n agent_id=\"explorer\",\n role=\"explores the filesystem\",\n system_prompt=\"...\",\n allowed_tools=conn.tool_names, # auto-populated from MCP server\n ))\n result = await runtime.run(\"list files in /tmp\")\n```\n\nSupports **stdio**, **SSE**, and **streamable-HTTP** transports. The\n`MCPServerConnection` context manager handles the full lifecycle —\nconnect, discover, and cleanup.\n\n### Auth options\n\nPick the provider that matches how your MCP server authenticates:\n\n| Provider | When to use |\n|---|---|\n| `StaticMCPAuth` | Literal header/env values you have in hand |\n| `BearerMCPAuth` | A single bearer token string |\n| `ApiKeyMCPAuth` | API-key headers backed by environment variables |\n| `OAuthMCPAuth` | Bearer token cached in the shared `auth.json` file |\n| `BrowserOAuthMCPAuth` | Full OAuth 2.0 + PKCE flow with browser login |\n\n**API keys backed by env vars** — generic, no vendor coupling:\n\n```python\nimport os\nfrom tools.mcp.auth import ApiKeyMCPAuth, StreamableHttpServerParams\nfrom tools.mcp import MCPServerConnection\n\nauth = ApiKeyMCPAuth({\n \"DD-Api-Key\": \"DD_API_KEY\",\n \"DD-Application-Key\": \"DD_APP_KEY\",\n})\nparams = StreamableHttpServerParams(url=\"https://mcp.datadoghq.com/\")\n\nasync with MCPServerConnection(params, server_name=\"datadog\", auth=auth) as conn:\n conn.register_tools(tool_registry)\n```\n\n**Browser-based OAuth (PKCE) for hosted MCP servers**:\n\n```python\nfrom tools.mcp.auth import BrowserOAuthMCPAuth, StreamableHttpServerParams\nfrom tools.mcp import MCPServerConnection\n\nauth = BrowserOAuthMCPAuth(\n server_url=\"https://mcp.example.com/\",\n provider_name=\"mcp:example\",\n client_id=\"abc123\", # from the provider's developer console\n client_secret=\"shh\", # optional (PKCE-only flows omit)\n scopes=[\"read\", \"write\"],\n)\nparams = StreamableHttpServerParams(url=\"https://mcp.example.com/\")\n\nasync with MCPServerConnection(params, auth=auth) as conn:\n conn.register_tools(tool_registry)\n```\n\nFirst connect opens the browser, captures the redirect on\n`http://127.0.0.1:8765/callback`, persists tokens to\n`~/.agent-harness/auth/auth.json` (chmod 0600), and refreshes them\ntransparently on every subsequent run. Register your OAuth app with that\nredirect URI.\n\nServers that support RFC 7591 dynamic client registration work without\nsupplying `client_id` — the MCP SDK registers a fresh client on first\nconnect.\n\n**Cached OAuth from the auth.json file** (for tokens you already minted\nelsewhere):\n\n```python\nfrom tools.mcp import OAuthMCPAuth, MCPServerConnection\n\nauth = OAuthMCPAuth.from_auth_file(\n \"~/.agent-harness/auth/auth.json\",\n provider=\"my-service\",\n)\n```\n\nSee `examples/mcp_demo.py` for local stdio MCP and `examples/mcp_auth_demo.py`\nfor authenticated remote MCP.\n\n## OpenTelemetry Tracing\n\nVisualize agent runs in Jaeger, Datadog, or any OTEL-compatible backend.\n\n```bash\npip install -e \".[otel]\"\n```\n\nOne flag enables tracing:\n\n```python\nruntime = AgentRuntime(\n agent_registry=agents,\n tool_registry=tools,\n memory=memory,\n llm=llm,\n enable_otel=True, # ← that's it\n)\n```\n\nSpan hierarchy:\n\n```\n[run] goal=\"Fetch httpbin.org/json...\"\n ├── [plan] task_count=2\n ├── [task] agent=researcher, task_id=t1\n │ ├── thought \"I need to fetch...\"\n │ ├── action tool=http_fetch\n │ └── thought \"Got the response...\"\n ├── [task] agent=researcher, task_id=t2\n │ └── thought \"Extracting title...\"\n └── [synthesis] confidence=0.95\n```\n\nLocal Jaeger setup:\n\n```bash\n# Start Jaeger (OTLP on :4318, UI on :16686)\ndocker run -d --name jaeger \\\n -p 16686:16686 -p 4318:4318 \\\n jaegertracing/all-in-one:latest\n\n# Run your agent\nOPENAI_API_KEY=sk-... python examples/openai_demo.py\n\n# View traces at http://localhost:16686\n```\n\nThe OTEL hook is a side-channel on the existing `Tracer` — the in-memory trace\nis always available via `result[\"trace\"]` regardless of whether OTEL is enabled.\nZero overhead and zero imports when `enable_otel=False`.\n\n## Trace recorder + replay + local viewer\n\nFor local debug and post-mortem inspection without an OTEL backend, the\nharness ships a JSONL trace recorder and a stdlib-only HTML viewer. Wrap\nany streaming call:\n\n```python\nfrom harness.trace import record_trace, replay\n\nasync for event in record_trace(runtime.dispatch_stream(goal), \"run.jsonl\"):\n ... # your normal handling\n```\n\nEach `BusEvent` is flushed per-line, so a partial trace survives a crash.\nView the trace in your browser:\n\n```bash\nagent-harness trace view run.jsonl # opens http://127.0.0.1:8765/\n```\n\nThe viewer is a single embedded HTML page — vertical timeline, filter by\nagent / event type / text, expandable per-event JSON. No build step, no\nexternal services.\n\nReplay a trace through `ConsoleRenderer` (great for grepping or piping\ninto another script):\n\n```bash\nagent-harness trace replay run.jsonl\nagent-harness trace replay run.jsonl --realtime --speed 2.0\n```\n\nProgrammatic replay yields reconstructed `BusEvent` objects:\n\n```python\nasync for event in replay(\"run.jsonl\", realtime=False):\n ... # reuse the same loops you write for live streams\n```\n\nThis is complementary to OTEL — OTEL is for production observability and\nlong-term storage in Jaeger/Datadog; the JSONL recorder is for local\ndebugging, sharing reproductions, and replaying past runs.\n\n## Vision / multimodal agents\n\n`WorkingMemory` accepts `str | list` content so image blocks pass through to\nvision-capable LLMs without modification.\n\n```bash\npip install -e \".[openai,http]\"\n```\n\n```python\nfrom tools.builtin.fetch_image import FetchImage\n\ntools.register(FetchImage())\n\nagents.register(AgentConfig(\n agent_id=\"vision_agent\",\n role=\"fetches images and describes their visual content\",\n system_prompt=\"Use fetch_image to retrieve images — you can see them directly.\",\n allowed_tools=[\"fetch_image\"],\n working_memory_max_tokens=16_000, # images use ~500 tokens each in budget\n))\n\nresult = await runtime.run_agent(\"vision_agent\", \"describe https://example.com/photo.jpg\")\n```\n\n`FetchImage` downloads the URL, base64-encodes the body, and returns an OpenAI\n`image_url` content block. The agent appends it to `WorkingMemory` as a content\nlist; the LLM receives the actual image. `OBSERVATION` events and the\nsummarization LLM see `[image]` as a placeholder so text-only paths are never\nhanded raw base64.\n\nImage token budget: a fixed `500` token estimate per image block (conservative\nmid-point of GPT-4o `auto` detail range). Override with a real counter if you\nneed exact figures.\n\n## Trajectory capture and RLHF\n\nEvery agent run automatically logs its full `WorkingMemory` message history as\na `\"trajectory\"` tracer event. Wire `InMemoryAnnotationStore` to collect and\nrate trajectories:\n\n```python\nfrom harness.annotation import InMemoryAnnotationStore\n\nstore = InMemoryAnnotationStore()\nruntime = AgentRuntime(..., annotation_store=store)\n\nawait runtime.run_agent(\"my_agent\", \"task\")\n\n# drain unrated trajectories\nfor annotation in store.list_unrated():\n print(annotation.agent_id, annotation.answer, annotation.confidence)\n store.rate(annotation.annotation_id, rating=0.9) # 1.0 = ideal\n\n# export to training pipeline\ntraining_data = store.list_all()\n```\n\n`Annotation` fields:\n\n| Field | Type | Description |\n|---|---|---|\n| `messages` | `list[dict]` | Full `WorkingMemory` trajectory — system prompt, every thought, tool call, observation, and final answer |\n| `answer` | `str` | Agent's final answer (`\"\"` on failure) |\n| `confidence` | `float` | Agent's self-reported confidence `[0, 1]` |\n| `steps` | `int` | ReAct steps taken |\n| `error` | `str` | `\"\"` on success; failure reason otherwise |\n| `summarization_count` | `int` | Number of `WorkingMemory` compression passes |\n| `rating` | `float \\| None` | Human rating `[0, 1]` — `None` until rated |\n| `correction` | `str \\| None` | Human-supplied correct answer when `rating < 1` |\n\nTrajectory capture fires in a `finally` block — it records on success, on\nmax-steps exhaustion, and on crash. Swap `InMemoryAnnotationStore` for any\nbackend that implements the same interface (`write`, `get`, `list_all`,\n`list_run`, `list_unrated`, `rate`, `count`).\n\n## Human-in-the-Loop (HITL)\n\nGate specific tool calls behind an interactive CLI prompt. Opt-in per agent via\n`hitl_tools`; zero overhead when unused. No extra dependencies — checkpoints\nare stored as JSON files by default.\n\n```python\nagents.register(AgentConfig(\n agent_id=\"file_agent\",\n role=\"manages files\",\n system_prompt=\"...\",\n allowed_tools=[\"read_file\", \"write_file\", \"delete_file\"],\n hitl_tools=[\"write_file\", \"delete_file\"], # these two require human approval\n))\n\n# AgentRuntime auto-creates a FileCheckpointStore when hitl_tools are present.\nruntime = AgentRuntime(...)\nawait runtime.run_agent(\"file_agent\", \"clean up the logs directory\")\n```\n\nCheckpoints are written to `~/.agent-harness/checkpoints/` by default.\nOverride the directory:\n\n```python\nfrom harness.checkpoint import FileCheckpointStore\n\nruntime = AgentRuntime(..., checkpoint_store=FileCheckpointStore(\"/var/lib/myapp/ckp\"))\n```\n\nFor Redis-backed storage (shared across processes or machines):\n\n```python\nimport redis.asyncio as aioredis\nfrom harness.checkpoint import RedisCheckpointStore\n\nclient = aioredis.from_url(\"redis://localhost:6379\", decode_responses=True)\nruntime = AgentRuntime(..., checkpoint_store=RedisCheckpointStore(client))\n```\n\nWhen the agent calls `write_file` or `delete_file` a prompt appears:\n\n```\n────────────────────────────────────────────────────────────\n HITL Approval Required\n────────────────────────────────────────────────────────────\n Tool: delete_file\n Args: {\"path\": \"/var/log/app.log\"}\n Agent: file_agent step=2\n Run: 3f7a1b2c-...:file_agent\n ID: a1b2-c3d4\n────────────────────────────────────────────────────────────\n y = approve once | a = allow 'delete_file' for session | A = always allow 'delete_file' | n = reject | = steer\n Ctrl-C to pause. Resume: python my_script.py --resume 3f7a1b2c-...:file_agent\n────────────────────────────────────────────────────────────\n Approve? [y/n/a/A/correction]:\n```\n\n**Prompt semantics:**\n\n| Input | Effect |\n|---|---|\n| `y` / `yes` | Tool runs once |\n| `n` / `no` | Tool skipped; agent sees a rejection observation |\n| `a` / `allow` | Tool runs **and** added to session allow-list; no further prompts for this tool (or command prefix for shell-like tools) |\n| `A` / `always` | Tool runs **and** a user-scoped allow rule is stored in `~/.agent-harness/policies/tool_policy.json` |\n| any other text | Correction: tool skipped, text injected into `WorkingMemory` as a user message; LLM self-corrects on the next step |\n\nFor shell-like tools (`shell`, `bash`, `run`, `exec`), `a` and `A` allow the\n**first word** of the command — e.g. approving `shell git commit ...` allows\nall `git` commands in that scope but still prompts for `shell rm ...`.\nPersistent rules are user-local, not repo files. Manage them with:\n\n```bash\nagent-harness policy list\nagent-harness policy revoke \nagent-harness policy clear\n```\n\n**Wall-time budget** is suspended while waiting for input — human think-time\ndoes not count against `max_wall_time_seconds`.\n\n**Gating does not depend on durability.** A tool listed in `hitl_tools` is\nalways gated; with no checkpoint store configured you still get the prompt,\nyou just don't get crash-resume.\n\n### Step-level checkpointing\n\nEnable periodic crash-resume independent of HITL:\n\n```python\nAgentConfig(\n agent_id=\"long_runner\",\n ...\n checkpoint_every=3, # checkpoint before every 3rd step (0 = disabled)\n)\n```\n\nThe same `CheckpointStore` is used for both HITL and step checkpoints. Resume\nworks with `runtime.resume(key)` regardless of how the checkpoint was created.\n\n### Checkpoint namespacing\n\nEach agent writes to its own key so orchestrated runs never overwrite each other:\n\n| Path | Checkpoint key | Stored at |\n|---|---|---|\n| Single-agent (`run_agent`, `run_routed`) | `:` | `~/.agent-harness/checkpoints/:.json` |\n| Orchestrated (`run`, `run_stream`) | `` (orchestrator) + `::` (each task) | one file per task, one file for the orchestrator |\n\nOrchestrated agent keys carry the **task id** because a plan may name the same\nagent for two tasks; keyed only by agent id, both would write to one file and\noverwrite each other.\n\nThe orchestrator checkpoint stores the goal, the *current* plan (replanning\nreplaces it), a `TaskState` per planned task, and the replan count. Per-task\nstatus — `pending / running / done / failed / skipped` plus a durable attempt\ncounter — is what lets a task interrupted mid-flight resume from its agent's\nown position rather than restarting. It is written at each task transition and\ndeleted on clean `DONE`.\n\n### Crash / Ctrl-C resume\n\nA checkpoint is a `RunState` (`harness/runstate.py`): where the run is (`step`\nplus a `think / approve / act / observe` phase), the actions of the current\nstep with per-action status, the full `WorkingMemory`, the budget so far, and —\nif it failed — why. It is written before every HITL prompt and (if\n`checkpoint_every > 0`) at each periodic step.\n\nResuming is not a separate code path: the loop loads the state and keeps\nadvancing it, so a run that died partway through a batch of approvals comes\nback knowing which calls were already approved and which tools already ran.\n\nRecorded spending is restored, so a resumed run cannot start its budget over.\nThe exception is a run that failed *because* of the budget — restoring the\nguard that just tripped would fail again at the same point, so the raised\n`GuardrailConfig` is allowed to take effect instead.\n\n> **Checkpoints from before 0.13 cannot be resumed.** They predate per-action\n> and per-task status, so there is no honest way to reconstruct where such a\n> run actually was; `runtime.resume()` raises `LegacyCheckpointError` rather\n> than guessing. Checkpoints are short-lived by design —\n> `FileCheckpointStore.purge_old` defaults to 7 days, `RedisCheckpointStore`\n> expires after 24h.\n\n**What the banner prints:**\n\n- **Single-agent run**: `--resume :` — restores just that agent.\n- **Orchestrated run**: `--resume ` — restores the full orchestration.\n\n```\n Run interrupted — checkpoint saved.\n Resume: python my_script.py --resume 3f7a1b2c-...\n```\n\n**Auto-resume — no script changes required.** When `checkpoint_store` is\nconfigured, `dispatch_stream` and `run_stream` detect `--resume ` in\n`sys.argv` automatically. Your existing script resumes transparently:\n\n```bash\npython my_script.py --resume 3f7a1b2c-...\n```\n\nThe runtime detects the flag, loads the checkpoint, and streams events\nidentically to a fresh run. Scripts need zero resume-specific code.\n\nFor **explicit control** — streaming resume or blocking resume:\n\n```python\n# streaming (same event sequence as the original run)\nasync for event in runtime.resume_stream(\"3f7a1b2c-...\"):\n ...\n\n# blocking\nresult = await runtime.resume(\"3f7a1b2c-...:file_agent\") # single-agent\nresult = await runtime.resume(\"3f7a1b2c-...\") # orchestrated\n```\n\nBoth `resume_stream` and `resume` auto-detect the checkpoint type (agent vs\norchestrator) from the stored data and call the right path.\n\nIf you need the resume key from `sys.argv` directly:\n\n```python\nfrom harness.checkpoint import maybe_resume_key\n\nkey = maybe_resume_key() # returns None if --resume is absent\n```\n\n**Orchestrated resume** skips completed tasks (injects their stored results\ndirectly into the synthesis step) and re-runs only the tasks that had not yet\nfinished. If an individual agent's HITL checkpoint is still on disk, that agent\nis resumed at its saved step rather than re-run from scratch.\n\n### Correction steering and replanning\n\nWhen the human types a correction instead of y/n:\n\n- **Single-agent run**: correction is injected as a `user` message in\n `WorkingMemory`. The LLM sees it on the next think step and self-corrects\n without replanning. Suitable for redirecting tool choice or adjusting\n parameters.\n- **Orchestrated run**: the correction steers only the current agent. Because\n the orchestrator checkpoint records task results as they complete, a full\n `runtime.resume(run_id)` after the agent finishes will continue the remaining\n tasks with correct upstream context.\n\nThe `annotation_store` and `checkpoint_store` are independent — both can be\nwired simultaneously for RLHF data collection with HITL review.\n\n## Async steering\n\nHITL is synchronous — it only fires when a gated tool is about to run. For\nout-of-band course-correction (HTTP handler, supervisor agent, file watcher,\nor a human typing in the terminal), each `BaseAgent` exposes a\nnon-blocking `steer(text)` method. Items are drained at the **top of each\nReAct iteration**, before the per-step checkpoint write and before the\nnext think, then appended to `WorkingMemory` as a `Human guidance: `\nuser message. The LLM sees them on the next think and adjusts. One\n`HUMAN_GUIDANCE` `BusEvent` fires per drained item.\n\nWhy a queue instead of writing straight to `WorkingMemory`: `steer()` is\nsynchronous and callable from any coroutine; `WorkingMemory.append` is\nasync (eviction can call the LLM). The queue is the producer/consumer\nboundary, enforces step-boundary delivery, and keeps WM single-writer.\n\n### Programmatic API (always available)\n\n```python\nagent.steer(\"skip the legal database, use academic sources only\")\n```\n\nFires immediately; the agent picks it up at the next step boundary.\nWorst-case latency = remaining tool time + next-think time.\n\n### Sources via factory (so orchestrated agents are reachable)\n\n`BaseAgent` and `AgentRuntime` both accept `steering_source_factory` — a\ncallable `(agent) -> async ctx mgr`. The agent enters the source on\n`run_stream`, exits on completion. No live-agent registry; agents the\nruntime constructs internally still get steering.\n\nTwo built-in factories:\n\n```python\nfrom harness.steering import file_steering_factory, stdin_steering_factory\n\n# 1. File-based — one file per agent, polled for appends (no shared resource)\nruntime = AgentRuntime(\n ...,\n steering_source_factory=file_steering_factory(\n \"/tmp/ah-{run_id}-{agent_id}.steer\"\n ),\n)\n# Steer from any other terminal:\n# echo \"wrap up and synthesise\" >> /tmp/ah--researcher.steer\n\n# 2. Stdin-based — single shared StdinRouter with prefix routing\nruntime = AgentRuntime(\n ...,\n steering_source_factory=stdin_steering_factory(),\n)\n# At the terminal:\n# researcher: skip the legal db, focus on academic\n# writer: keep the report under 500 words\n# *: stop after this step\n```\n\nSingle-agent stdin runs accept lines with no prefix. Multi-agent runs\nrequire `agent_id: text` (or `*: text` for broadcast); unknown or\nunprefixed lines print a stderr hint and are discarded.\n\nThe stdin factory's underlying `StdinRouter` is started/stopped\nautomatically — the runtime detects the factory's async-context-manager\nshape and wraps `dispatch_stream` / `run_stream` / `run_routed_stream`\naround it. Ref-counted so nested calls (`dispatch_stream → run_stream`)\ndon't double-start the router.\n\n### HITL coordination\n\nWhen a `StdinRouter` is active, HITL calls `router.claim_next_line()`\n**before** printing its approval banner — the next stdin line resolves\nHITL's pending Future and bypasses pub/sub. After resolution, subsequent\nlines route to steering subscribers normally. When no router is active,\nHITL falls back to a standalone `prompt_toolkit` session, ensuring consistent\nkey-bindings (like Enter-submits and Alt-Enter/Ctrl-J-newline) across both paths.\n\n### Constraints\n\n- Steering arrives **between steps**, never mid-tool, never mid-think.\n Tools that are already running complete; the LLM stream that's\n already producing completes; guidance lands at the next safe boundary.\n- Guidance queued **after** the LLM emits `action: \"finish\"` is lost —\n the agent already decided it's done.\n- Crash between drain and next checkpoint write → the queued items are\n in the persisted WM. Crash between checkpoint write and next drain →\n lost; re-steer after `--resume`.\n\nSee `examples/complex_sysaudit_demo.py` for stdin steering across three\nagents alongside HITL on the shell tool.\n\n## AgentConfig reference\n\n| Field | Default | Description |\n|---|---|---|\n| `agent_id` | required | Unique identifier for the agent |\n| `role` | required | Plain-English description used by the planner for agent selection |\n| `system_prompt` | required | Base system prompt for the agent |\n| `allowed_tools` | required | Tool names the agent may call |\n| `skills` | `[]` | Reusable prompt/context bundles attached to this agent; skills do not grant tools |\n| `max_steps` | `10` | Maximum ReAct iterations before the run is terminated |\n| `max_wall_time_seconds` | (guardrail) | See `GuardrailConfig` |\n| `memory_context_enabled` | `True` | Prepend relevant long-term memory to the system prompt |\n| `confidence_from_llm` | `True` | Use the `confidence` field from the LLM response; set `False` to always return `1.0` |\n| `working_memory_max_tokens` | `None` (auto-derive from `llm.input_token_budget × 0.8`; pass an int to hard-cap) | Token budget for in-context working memory before rolling summarisation kicks in |\n| `max_observation_chars` | `20_000` | Character cap for model-facing tool observations appended to working memory; event payloads keep the original result. Set `0` to disable |\n| `hitl_tools` | `[]` | Tool names that require human approval before execution |\n| `checkpoint_every` | `0` | Write a crash-resumable checkpoint every N steps; `0` disables periodic checkpoints |\n| `stream_tokens` | `False` | Emit `TOKEN` events as the LLM streams. Disabled by default — enable if you want to render partial output in real time: `AgentConfig(..., stream_tokens=True)` |\n| `reasoning_effort` | `None` (adapter/provider default) | Per-agent reasoning depth override. Supported values depend on the provider; set it once on the LLM adapter to apply across all calls instead. |\n"}