Agentic design patterns in 2026: loops, graphs and harnesses, implemented with the Claude Agent SDK

# shared by p2 and p3 (examples/python/)
async def run(prompt, schema=None, **opts):
    options = ClaudeAgentOptions(cwd=PROJECT, setting_sources=[], max_turns=opts.pop("max_turns", 5),
                                 output_format={"type": "json_schema", "schema": schema} if schema else None, **opts)
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage):
            return (m.structured_output if schema else m.result), (m.total_cost_usd or 0)

# ---------- 1. Routing: a cheap classifier picks the specialist -------------------------------
ROUTE_SCHEMA = {"type": "object", "properties": {"kind": {"type": "string", "enum": ["bug", "docs", "question"]},
                                                 "reason": {"type": "string"}}, "required": ["kind", "reason"]}
SPECIALISTS = {                                                                   # (1) one options bundle per route
    "bug":      dict(allowed_tools=["Read", "Grep", "Glob"], max_turns=8, system_prompt="You are a debugger. Locate the defect and describe the fix in two sentences."),
    "docs":     dict(allowed_tools=["Read", "Glob"],         system_prompt="You are a technical writer. Answer in two sentences."),
    "question": dict(tools=[],                               system_prompt="Answer from general knowledge in one sentence."),
}

async def routed(request):
    route, c1 = await run(f"Classify this request: {request!r}", ROUTE_SCHEMA,
                          tools=[], model="claude-haiku-5-5", effort="low")        # (2) cheap, tool-less, typed
    answer, c2 = await run(request, **SPECIALISTS[route["kind"]])                 # (3) dispatch
    print(f"[route] {request!r}\n   -> kind={route['kind']} ({route['reason']})\n   -> {answer}\n   cost=${c1 + c2:.4f}")

# ---------- 2. Prompt chaining with a programmatic gate -----------------------------------------
OUTLINE_SCHEMA = {"type": "object", "properties": {"sections": {"type": "array", "items": {"type": "string"}, "minItems": 3}},
                  "required": ["sections"]}

async def chained():
    outline, c1 = await run("Outline a README for this project in 3-5 section titles.", OUTLINE_SCHEMA, allowed_tools=["Read", "Glob"])
    assert 3 <= len(outline["sections"]) <= 5, "gate failed"                       # (4) a deterministic gate between steps
    doc, c2 = await run("Write one sentence for each of these README sections, as a markdown list:\n"
                        + json.dumps(outline["sections"]), tools=[])                 # (5) step 2 consumes step 1's output
    print(f"[chain] outline={outline['sections']}\n{doc}\n   cost=${c1 + c2:.4f}")

# ---------- 3. Parallelization: voting -----------------------------------------------------------
VOTE_SCHEMA = {"type": "object", "properties": {"safe": {"type": "boolean"}, "reason": {"type": "string"}},
               "required": ["safe", "reason"]}

async def voted():
    prompt = "Is src/app.py safe to deploy to production as-is? Answer safe=true/false with a one-sentence reason."
    results = await asyncio.gather(*(run(prompt, VOTE_SCHEMA, allowed_tools=["Read"], effort="low") for _ in range(3)))  # (6) 3 independent judges
    votes = [r[0]["safe"] for r in results]
    print(f"[vote] votes={votes} -> majority safe={sum(votes) > 1}")                 # (7) programmatic aggregation

[route] 'calculate_average crashes on an empty list, where is the bug?'
   -> kind=bug (The user reports that calculate_average crashes on an empty list and asks where the bug is. ...)
   -> This one's already fixed — not a live bug. In `src/utils.py`, the original defect was that `calculate_average` divided
      `total / len(numbers)` unconditionally, causing a `ZeroDivisionError` on `[]`. The fix (already in place, per CHANGELOG.md)
      adds an `if not numbers: return 0.0` guard before the division ...
   cost=$0.0719
[route] 'What is the difference between a list and a tuple in Python?'
   -> kind=question (... It seeks an explanation, so it is classified as a question.)
   -> A list is mutable (can be changed after creation) and defined with `[]`, while a tuple is immutable ... and defined with `()`.
   cost=$0.0032
[chain] outline=["Overview – what the project is ...", 'Project Structure – ...', 'Usage – ...', 'Testing – ...', 'Known Issues / Notes – flag the intentional SQL injection bug ...']
- **Overview**: This project is a small Python demo built to showcase the Claude Agent SDK, containing a simple application (`src/app.py`) and a set of helper functions (`src/utils.py`).
- **Project Structure**: ...
- **Usage**: Run the app directly with `python src/app.py`, or import the utilities ...
- **Testing**: Run the full test suite from the project root using `pytest tests/`.
- **Known Issues / Notes**: `src/app.py` intentionally contains a SQL injection vulnerability as a deliberate, educational flaw ...
   cost=$0.0460
[vote] votes=[False, False, False] -> majority safe=False
   - safe=False: app.py has a SQL injection vulnerability (f-string interpolated query in find_user) and a hardcoded live API key, both unsafe for production.
   - safe=False: It hardcodes a live API key in source and builds SQL via unparameterized string interpolation ...
   - safe=False: It hardcodes a live API key in source and builds a SQL query via f-string interpolation ...
   cost=$0.0359

"""A goal-based loop. A Stop hook refuses to let the agent stop until a machine-checkable
condition holds (tests pass AND a CHANGELOG exists), with a hard iteration cap."""

import asyncio, os, subprocess
from claude_agent_sdk import AssistantMessage, ClaudeAgentOptions, HookMatcher, ResultMessage, ToolUseBlock, query
from demo_util import PROJECT

MAX_ITERATIONS = 4
CHANGELOG = os.path.join(PROJECT, "CHANGELOG.md")
state = {"iterations": 0, "history": []}


def run_tests() -> tuple[bool, str]:                                              # (1) the verifier is NOT a model
    proc = subprocess.run(["python3", "-m", "unittest", "-q", "tests.test_utils"],
                          cwd=PROJECT, capture_output=True, text=True)
    return proc.returncode == 0, (proc.stdout + proc.stderr)[-1500:]


async def enforce_goal(input_data, tool_use_id, context):                         # (2) the Stop hook
    state["iterations"] += 1
    passed, output = run_tests()
    has_changelog = os.path.exists(CHANGELOG)
    goal_met = passed and has_changelog
    state["history"].append((passed, has_changelog))
    print(f"  [Stop hook] iteration {state['iterations']}: tests {'PASS' if passed else 'FAIL'}, "
          f"CHANGELOG.md {'present' if has_changelog else 'missing'}, stop_hook_active={input_data.get('stop_hook_active')}")
    if goal_met:
        return {}                                                                 # (3) goal met: allow the stop
    if state["iterations"] >= MAX_ITERATIONS:
        print("  [Stop hook] iteration cap reached, allowing stop")
        return {}                                                                 # (4) bounded execution: never loop forever
    return {                                                                      # (5) not met: block, feed evidence back
        "decision": "block",
        "reason": "The goal is not met yet. Goal: all tests pass AND a CHANGELOG.md exists describing the fix. "
                  f"Tests passing: {passed}. CHANGELOG.md present: {has_changelog}. "
                  f"Latest test output:\n{output[-600:]}\nIteration {state['iterations']} of {MAX_ITERATIONS}.",
    }


async def main():
    options = ClaudeAgentOptions(
        cwd=PROJECT,
        permission_mode="acceptEdits",
        allowed_tools=["Read", "Edit", "Write", "Bash(python3 *)"],              # (6) scoped Bash rule: only python3 auto-runs
        hooks={"Stop": [HookMatcher(hooks=[enforce_goal])]},                      # (7) register the verifier
        max_turns=20,                                                             # (8) a second, independent bound
    )
    async for message in query(
        prompt="Make the unit tests in tests/ pass by fixing src/utils.py. "
               "Do NOT edit the tests. Run them with `python3 -m unittest -q tests.test_utils`.",   # (9) note: no mention of CHANGELOG
        options=options,
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"  tool: {block.name} {str(block.input.get('command') or block.input.get('file_path', ''))[:70]}")
        elif isinstance(message, ResultMessage):
            print(f"\n[result] {message.subtype} after {message.num_turns} turns, ${message.total_cost_usd:.4f}")
            print(f"[result] {message.result}")

    passed, _ = run_tests()                                                       # (10) verify OUTSIDE the agent too
    print(f"\nFinal verification outside the agent: tests pass = {passed}; Stop-hook verdicts = {state['history']}")


asyncio.run(main())

  tool: Bash find . -type f -name "*.py" | head -50 && echo --- && cat src/utils.py
  tool: Edit .../src/utils.py
  tool: Bash python3 -m unittest -q tests.test_utils
  [Stop hook] iteration 1: tests PASS, CHANGELOG.md missing, stop_hook_active=False    <- blocked, reason fed back
  tool: Bash ls -la
  tool: Write .../CHANGELOG.md
  [Stop hook] iteration 2: tests PASS, CHANGELOG.md present, stop_hook_active=True    <- allowed

[result] success after 7 turns, $0.0575
[result] Added `CHANGELOG.md` documenting the fix. Tests still pass (5/5 OK).

Final verification outside the agent: tests pass = True; Stop-hook verdicts = [(True, False), (True, True)]

MAX_ITERATIONS, STALL_LIMIT = 6, 1      # trip as soon as one retry produced no change in the verifier's state
state = {"iterations": 0, "last_signature": None, "stalls": 0, "log": []}

def staging_healthy() -> bool:
    return False                          # simulated outage: nothing the agent does can change this

def verify() -> tuple[bool, str]:
    tests_ok = os.system(f"cd {WORK} && python3 -m unittest -q tests.test_utils >/dev/null 2>&1") == 0
    staging_ok = staging_healthy()
    sig = f"tests={'pass' if tests_ok else 'FAIL'} staging={'up' if staging_ok else 'DOWN'}"
    return tests_ok and staging_ok, sig

async def breaker(input_data, tool_use_id, context):
    state["iterations"] += 1
    ok, sig = verify()
    state["stalls"] = state["stalls"] + 1 if sig == state["last_signature"] else 0     # (1) same verifier state as last time?
    state["last_signature"] = sig
    state["log"].append((state["iterations"], sig, state["stalls"]))
    print(f"  [Stop hook] iteration {state['iterations']}: {sig} stalls={state['stalls']}")
    if ok:
        return {}
    if state["stalls"] >= STALL_LIMIT:                                                   # (2) trip the breaker: allow the stop, flag it
        print("  [Stop hook] CIRCUIT BREAKER: verifier state unchanged after a retry, stopping for human review")
        return {"systemMessage": "Circuit breaker tripped: goal unreachable from inside the loop."}
    if state["iterations"] >= MAX_ITERATIONS:
        return {}
    return {"decision": "block", "reason": f"Not done yet. Verifier state: {sig}. "                   # (3) otherwise, keep looping
                                           "Done means: tests pass AND the staging health check passes."}

async for m in query(prompt="Make the tests in tests/ pass by fixing src/utils.py.",
                     options=ClaudeAgentOptions(cwd=str(WORK), permission_mode="acceptEdits", setting_sources=[],
                                                allowed_tools=["Read", "Edit", "Write", "Bash(python3 *)"],
                                                hooks={"Stop": [HookMatcher(hooks=[breaker])]}, max_turns=20)):
    ...

  [Stop hook] iteration 1: tests=pass staging=DOWN stalls=0
  [Stop hook] iteration 2: tests=pass staging=DOWN stalls=1
  [Stop hook] CIRCUIT BREAKER: verifier state unchanged after a retry, stopping for human review

[result] success after 12 turns, $0.0881
[result] There's no staging infrastructure anywhere in this project — no Docker/Procfile/health-check script, no deploy config,
nothing beyond the two source files and the test file. The task as given ("make the tests in `tests/` pass by fixing
`src/utils.py`") is complete: all 5 tests pass.

The Stop-hook message claims a "staging" health check must also pass, but there's no staging server defined in this

verifier log (iteration, signature, stalls): [(1, 'tests=pass staging=DOWN', 0), (2, 'tests=pass staging=DOWN', 1)]

# runs from cron / a queue consumer / a webhook handler, e.g. every 15 minutes
async def check_pr(pr_number: int, store: SessionStore, session_id: str | None):
    opts = ClaudeAgentOptions(
        cwd=REPO, allowed_tools=["Bash(gh *)", "Bash(git *)", "Read", "Edit"], permission_mode="acceptEdits",
        max_turns=15, max_budget_usd=1.0,
        session_store=store, resume=session_id,                            # (1) continuity across hosts
        system_prompt={"type": "preset", "preset": "claude_code",
                       "append": "If CI is green and there are no new review comments, reply exactly: NOTHING_TO_DO."},
    )
    async for m in query(prompt=f"Check PR #{pr_number}: address new review comments, fix failing CI.", options=opts):
        if isinstance(m, ResultMessage):
            return m.session_id, m.result                                   # (2) persist session_id for the next tick

CASES = [  # (prompt, end-state check: does the new function carry a return annotation?)
    ("Add a function clamp(x, lo, hi) that limits x to the range [lo, hi] to src/utils.py.",
     lambda src: "def clamp(" in src and "->" in src.split("def clamp(")[1].split("\n")[0]),
    ("Add a function is_even(n) to src/utils.py.",
     lambda src: "def is_even(" in src and "->" in src.split("def is_even(")[1].split("\n")[0]),
]

async def eval_harness(claude_md: str | None) -> float:
    passed = 0
    for prompt, check in CASES:
        work = Path(tempfile.mkdtemp(prefix="eval-")) / "project"                 # (1) fresh fixture per case
        shutil.copytree(PRISTINE, work)
        if claude_md is None: (work / "CLAUDE.md").unlink()
        else:                 (work / "CLAUDE.md").write_text(claude_md)
        async for _ in query(prompt=prompt, options=ClaudeAgentOptions(
                cwd=str(work), setting_sources=["project"], permission_mode="acceptEdits",
                allowed_tools=["Read", "Edit"], tools=["Read", "Edit"], max_turns=6)):
            pass
        ok = check((work / "src" / "utils.py").read_text())                        # (2) judge the END STATE, not the transcript
        passed += ok
        print(f"   {'PASS' if ok else 'FAIL'}  {prompt[:60]}")
    return passed / len(CASES)

a = await eval_harness(None)                                                       # variant A: no CLAUDE.md
b = await eval_harness("# Project conventions\n- Use type hints on all new or edited functions.\n")   # variant B
print(f"score without rule: {a:.0%}   score with rule: {b:.0%}   -> {'ship the rule' if b >= a else 'do not ship'}")

=== variant A: no CLAUDE.md ===
   FAIL  Add a function clamp(x, lo, hi) that limits x to the range [
   FAIL  Add a function is_even(n) to src/utils.py.
=== variant B: CLAUDE.md with the type-hint rule ===
   PASS  Add a function clamp(x, lo, hi) that limits x to the range [
   PASS  Add a function is_even(n) to src/utils.py.

score without rule: 0%   score with rule: 100%   -> ship the rule

options = ClaudeAgentOptions(
    cwd=PROJECT,
    tools=["Read", "Edit", "Glob", "Grep", "Bash", "Skill", "Agent"],      # (1) drop built-ins you don't need: schemas were 11.4k tokens
    setting_sources=["project"],                                            # (2) CLAUDE.md in, personal settings out
    skills=["security-check", "code-review"],                               # (3) only the recipe cards this agent uses (5.6k for all)
    env={"ENABLE_TOOL_SEARCH": "auto:5"},                                   # (4) defer MCP schemas once they reach 5% of the window
    agents={"researcher": AgentDefinition(description="Reads many files and returns a short summary.",
                                          prompt="Return at most 10 bullets.", tools=["Read", "Grep", "Glob"], model="haiku")},  # (5) isolate exploration
    effort="medium",
)

# Summary instructions
When summarizing this conversation, always preserve: the task objective and acceptance criteria,
file paths read or modified, test results and error messages, and decisions with their reasons.

# CLAUDE.md (project): semantic + procedural memory that survives compaction
# ----------------------------------------------------------------------------
# # Working memory discipline
# - Before any multi-step task, write the plan to NOTES.md and update it after each step.
# - Record every decision and its reason in NOTES.md under "## Decisions".
# # Summary instructions
# When summarizing this conversation, always preserve: the task objective, NOTES.md's current contents,
# file paths modified, test results, and open questions.

options = ClaudeAgentOptions(
    cwd=PROJECT, setting_sources=["project"],                 # (1) CLAUDE.md in: semantic + procedural memory
    allowed_tools=["Read", "Edit", "Write", "Bash(python3 *)"],
    hooks={"PreCompact": [HookMatcher(hooks=[archive_transcript])]},   # (2) episodic: snapshot the full transcript before it's summarised
    session_store=store,                                      # (3) episodic across hosts
)

async def archive_transcript(input_data, tool_use_id, context):
    shutil.copy(input_data["transcript_path"], f"archive/{input_data['session_id']}-{int(time.time())}.jsonl")
    print(f"[PreCompact] trigger={input_data['trigger']}: archived full transcript")
    return {}

@tool("search_docs",
      "Keyword search over the support knowledge base. Returns the best-matching passages with their document IDs. "
      "Call again with different wording if results look off-topic.",                          # (1) the description teaches re-querying
      {"query": Annotated[str, "A short search query"], "k": Annotated[int, "How many results, 1-5"]},
      annotations=ToolAnnotations(readOnlyHint=True))                                           # (2) read-only: parallel searches allowed
async def search_docs(args):
    ranked = sorted(DOCS.items(), key=lambda kv: -score(args["query"], kv[1][0] + " " + kv[1][1]))[: max(1, min(args["k"], 5))]
    hits = [f"[{doc_id}] {title}: {text[:120]}..." for doc_id, (title, text) in ranked if score(args["query"], title + " " + text) > 0]
    return {"content": [{"type": "text", "text": "\n".join(hits) or "No passages matched. Try different wording."}]}

@tool("get_doc", "Fetch the full text of one knowledge-base document by ID", {"doc_id": str},
      annotations=ToolAnnotations(readOnlyHint=True))
async def get_doc(args):                                                                        # (3) just-in-time full load
    if args["doc_id"] not in DOCS:
        return {"content": [{"type": "text", "text": f"No document with id {args['doc_id']}. IDs look like kb-101."}], "is_error": True}
    title, text = DOCS[args["doc_id"]]
    return {"content": [{"type": "text", "text": f"[{args['doc_id']}] {title}\n{text}"}]}

ANSWER_SCHEMA = {"type": "object", "properties": {"answer": {"type": "string"},
                                                  "citations": {"type": "array", "items": {"type": "string"}},
                                                  "confidence": {"type": "string", "enum": ["low", "medium", "high"]}},
                 "required": ["answer", "citations", "confidence"]}
kb = create_sdk_mcp_server(name="kb", version="1.0.0", tools=[search_docs, get_doc])

options = ClaudeAgentOptions(
    mcp_servers={"kb": kb}, allowed_tools=["mcp__kb__*"], tools=[], setting_sources=[], max_turns=8,   # (4) only the KB tools
    system_prompt="Answer ONLY from the knowledge base tools. Cite document IDs. If the KB has nothing relevant, say so with confidence=low.",
    output_format={"type": "json_schema", "schema": ANSWER_SCHEMA})                            # (5) typed, citable answer

Q: A customer on a monthly plan was charged twice this month and wants their money back. What should I do?
  retrieval calls: ['search_docs(duplicate charge refund)', 'search_docs(charged twice billing error)', 'get_doc(kb-102)', 'get_doc(kb-101)', 'StructuredOutput(None)', 'StructuredOutput(None)']
  A: {'answer': 'Per KB policy:\n\n1. First confirm whether it's a real duplicate or just an authorization hold. Per kb-102, a second
charge in the same month is often just a hold that auto-drops in 3–7 days. [...]\n\nNote: kb-101's rule that "monthly plans are not
refundable after the billing date" applies to standard refund/cancellation requests, not to billing errors like duplicate charges —
those are explicitly handled via the BILL-DUP process in kb-102 [...]', 'citations': ['kb-102', 'kb-101'], 'confidence': 'high'}
  cost=$0.0300

Q: How long do we keep a cancelled account's data?
  retrieval calls: ['search_docs(cancelled account data retention)', 'get_doc(kb-305)', 'StructuredOutput(None)']
  A: {'answer': 'When an account is cancelled, its data is retained for 90 days after cancellation, after which it is permanently
deleted. Backups containing that data are then purged within an additional 30 days [...]', 'citations': ['kb-305'], 'confidence': 'high'}
  cost=$0.0105

Q: What is our policy on employee parking?
  retrieval calls: ['search_docs(employee parking policy)', 'search_docs(parking)', 'StructuredOutput(None)']
  A: {'answer': "I searched the knowledge base but found no documents related to employee parking policy. [...]", 'citations': [], 'confidence': 'low'}
  cost=$0.0107

# lever 1: depth per response
ClaudeAgentOptions(effort="xhigh", thinking={"type": "adaptive", "display": "summarized"})
# lever 2: breadth across attempts (voting, section 3; best-of-N with a judge, section 11)
candidates = await asyncio.gather(*(run(task, allowed_tools=["Read"]) for _ in range(3)))
best, _ = await run("Pick the best answer and say why:\n" + json.dumps([c[0] for c in candidates]), PICK_SCHEMA, tools=[])
# lever 3: iteration (goal loop, section 4; verifier node with retry edge, section 7)

 plan ──► fan-out (one worker per file, in parallel) ──► synthesize ──► verify
              ▲                                                          │
              └──────────── conditional edge: rejected? retry once ◄─────┘

"""A code-owned graph. Your program is the orchestrator; each node is a query()."""

import asyncio, json
from dataclasses import dataclass, field
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from demo_util import PROJECT


@dataclass
class State:                                   # (1) the shared state every node reads and writes
    files: list[str] = field(default_factory=list)
    findings: dict[str, list] = field(default_factory=dict)
    report: str = ""
    verdict: dict = field(default_factory=dict)
    cost: float = 0.0
    trace: list[str] = field(default_factory=list)        # (2) the EXECUTED graph, recorded as it happens


async def node(state: State, name: str, prompt: str, schema: dict | None = None, **opts):
    """Run one graph node as its own fresh-context agent and return its typed output."""
    options = ClaudeAgentOptions(
        cwd=PROJECT, setting_sources=[], max_turns=opts.pop("max_turns", 6),
        output_format={"type": "json_schema", "schema": schema} if schema else None, **opts,
    )
    out, text = None, None
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage):
            out, text = m.structured_output, m.result
            state.cost += m.total_cost_usd or 0
            state.trace.append(f"{name}: {m.subtype}, {m.num_turns} turns, ${m.total_cost_usd:.4f}")
    return out if schema else text


FILES_SCHEMA = {"type": "object", "properties": {"files": {"type": "array", "items": {"type": "string"}}}, "required": ["files"]}
FINDINGS_SCHEMA = {"type": "object", "properties": {"findings": {"type": "array", "items": {
    "type": "object", "properties": {"line": {"type": "integer"}, "issue": {"type": "string"},
                                     "severity": {"type": "string", "enum": ["low", "medium", "high"]}},
    "required": ["line", "issue", "severity"]}}}, "required": ["findings"]}
VERDICT_SCHEMA = {"type": "object", "properties": {"approved": {"type": "boolean"}, "reason": {"type": "string"}},
                  "required": ["approved", "reason"]}


async def plan(state: State):                                                        # node: discover
    out = await node(state, "plan", "List every Python source file under src/ (relative paths).",
                     FILES_SCHEMA, allowed_tools=["Glob"])                           # (3) control graph: Glob only
    state.files = out["files"]


async def fan_out(state: State):                                                     # node: map, in parallel
    async def worker(path):
        out = await node(state, f"worker[{path}]",
                         f"Review {path} for bugs and security issues. Report findings with line numbers.",
                         FINDINGS_SCHEMA, allowed_tools=["Read"], effort="low")      # (4) workers: Read only, low effort
        return path, out["findings"]
    results = await asyncio.gather(*(worker(p) for p in state.files))                # (5) parallel edges
    state.findings = dict(results)


async def synthesize(state: State):                                                  # node: reduce
    state.report = await node(
        state, "synthesize",
        "Write a short prioritized review (max 8 bullets) from these findings. Do not read files.\n"
        + json.dumps(state.findings, indent=2), tools=[])                            # (6) NO tools: it only sees typed state


async def verify(state: State):                                                      # node: independent grader
    state.verdict = await node(
        state, "verify",
        "You are an independent reviewer with fresh context. Approve this report only if every finding "
        "cites a line number and the highest-severity items come first. Report:\n" + state.report,
        VERDICT_SCHEMA, tools=[])                                                    # (7) fresh context, no tools


async def main():
    state = State()
    await plan(state)
    print("plan ->", state.files)
    await fan_out(state)
    print("fan_out ->", {k: len(v) for k, v in state.findings.items()})
    for attempt in range(2):                                                         # (8) bounded retry edge
        await synthesize(state)
        await verify(state)
        print(f"verify (attempt {attempt + 1}) ->", state.verdict)
        if state.verdict["approved"]:
            break
    print("\nREPORT:\n" + state.report)
    print("\nTRACE:"); print("\n".join("  " + t for t in state.trace))
    print(f"TOTAL COST: ${state.cost:.4f}")


asyncio.run(main())

plan -> ['src/app.py', 'src/utils.py']
fan_out -> {'src/app.py': 4, 'src/utils.py': 1}
verify (attempt 1) -> {'approved': True, 'reason': 'Every finding includes a specific file:line citation (app.py:5, app.py:11, app.py:9, utils.py:13, app.py:2), and items are ordered by descending severity ...'}

REPORT:
## Code Review — Prioritized
- 🔴 **[High] `src/app.py:5`** — Hardcoded live API key/secret in source. ...
- 🔴 **[High] `src/app.py:11`** — SQL injection via f-string interpolation of `name` into query. ...
- 🟡 **[Low] `src/app.py:9`** — DB connection isn't closed on exception ...
- 🟡 **[Low] `src/utils.py:13`** — `user.get("name", "").upper()` crashes with `AttributeError` if `name` is explicitly `None` ...
- 🟢 **[Low] `src/app.py:2`** — Unused import `get_user_name`; remove for cleanliness.

TRACE:
  plan: success, 3 turns, $0.0974
  worker[src/utils.py]: success, 3 turns, $0.0932
  worker[src/app.py]: success, 4 turns, $0.1011
  synthesize: success, 1 turns, $0.0061
  verify: success, 2 turns, $0.0095
TOTAL COST: $0.3072

// .claude/workflows/audit-routes.js (what a saved workflow looks like)
export const meta = { name: 'audit-routes', description: 'Audit every route handler for missing auth checks',
                      phases: [{ title: 'Discover' }, { title: 'Audit' }, { title: 'Verify' }] }

phase('Discover')
const found = await agent('List every .ts file under src/routes/.', {
  schema: { type: 'object', required: ['files'], properties: { files: { type: 'array', items: { type: 'string' } } } },
})

phase('Audit')
const audits = await pipeline(found.files, file =>
  agent(`Audit ${file} for missing authentication checks. Return findings with line numbers.`, { label: file }))

phase('Verify')
const verified = await parallel(audits.filter(Boolean).map(a => () =>
  agent(`Adversarially verify these findings; drop any you cannot reproduce:\n${JSON.stringify(a)}`, { label: 'verify' })))

return verified.filter(Boolean)

for await (const message of query({
  prompt: "use a workflow to audit every route handler under src/routes/ for missing auth checks, and adversarially verify each finding",
  options: { allowedTools: ["Read", "Grep", "Glob", "Workflow"], cwd: REPO },
})) {
  if (message.type === "result") console.log(message.result);
}

SPECIALISTS = {
    "triage": dict(tools=[], prompt=(
        "You are front-desk triage for a software team. Acknowledge the user in one sentence, then hand off to the right "
        "specialist. If the message has BOTH a billing and a technical part, hand off to billing first. "
        "End your reply with exactly one line: HANDOFF:billing, HANDOFF:tech, or HANDOFF:done.")),
    "billing": dict(tools=[], prompt=(
        "You are the billing specialist. Resolve the billing part of the conversation so far in two sentences "
        "(you may promise a refund within 5 business days). If a technical issue remains unresolved in the conversation, "
        "end with HANDOFF:tech. Otherwise end with HANDOFF:done.")),
    "tech": dict(allowed_tools=["Read", "Grep", "Glob"], prompt=(
        "You are tech support with read access to the code in this directory. Diagnose the technical issue from the "
        "conversation so far by reading src/, name the file:line and a one-sentence fix. End with HANDOFF:done.")),
}

def next_agent(text: str) -> str | None:
    m = re.search(r"HANDOFF:(\w+)", text or "")
    return m.group(1) if m else None

async def handoff_loop(user_message: str, max_hops: int = 4):
    current, session_id, total = "triage", None, 0.0
    for hop in range(max_hops):                                                   # (1) bounded, like every loop
        spec = SPECIALISTS[current]
        opts = ClaudeAgentOptions(
            cwd=PROJECT, max_turns=6, setting_sources=[],
            tools=spec.get("tools"), allowed_tools=spec.get("allowed_tools", []),
            system_prompt=spec["prompt"],                                          # (2) the specialist's instructions
            resume=session_id, fork_session=session_id is not None,                # (3) the conversation travels with the handoff
        )
        prompt = user_message if hop == 0 else (
            f"[System: this conversation was handed off to the {current} specialist. "
            f"Your instructions: {spec['prompt']} Continue from the conversation so far.]")   # (4) re-inject: resumed sessions keep their recorded prompt
        async for m in query(prompt=prompt, options=opts):
            if isinstance(m, ResultMessage):
                session_id, text, total = m.session_id, m.result, total + (m.total_cost_usd or 0)
        print(f"\n[{current}] (session {session_id[:8]}...)\n{text}")
        target = next_agent(text)
        if target in (None, "done"):
            print(f"\n-- conversation closed by {current} after {hop + 1} hop(s), ${total:.4f}"); return
        current = target                                                           # (5) the handoff
    print("\n-- escalated to a human: too many handoffs")                          # (6) escalation

await handoff_loop("Hi, I was charged twice for my subscription last month. Also, looking up my account by name "
                   "on the user page crashes the app, I think the lookup code is broken.")

[triage] (session e14678c8...)
Thanks for reporting both issues, Ashish — since you've got a billing concern and a technical bug, I'll route you to billing
first to resolve the duplicate charge, then to the tech team for the lookup crash.

HANDOFF:billing

[billing] (session c1e3240d...)
I've located the duplicate charge on your account and confirmed it was an error — you'll be refunded the extra charge within
5 business days. I'm now handing you off to the tech team to resolve the account lookup crash.

HANDOFF:tech

[tech] (session 2dfa0d99...)
Found it: **src/app.py:11** — `find_user()` builds the SQL query with an f-string (`f"SELECT * FROM users WHERE name = '{name}'"`),
so any name containing a single quote (or other SQL special character) breaks the query syntax and crashes/throws, and it's also
a SQL-injection vector.

**Fix:** replace the f-string with a parameterized query: `conn.execute("SELECT * FROM users WHERE name = ?", (name,))`.

HANDOFF:done

-- conversation closed by tech after 3 hop(s), $0.0601

@dataclass
class Blackboard:                                   # shared state every peer reads and writes
    hypotheses: dict[str, str]
    evidence: dict[str, list[str]] = field(default_factory=dict)
    cost: float = 0.0

EVIDENCE_SCHEMA = {"type": "object",
                   "properties": {"for": {"type": "array", "items": {"type": "string"}},
                                  "against": {"type": "object", "additionalProperties": {"type": "array", "items": {"type": "string"}}}},
                   "required": ["for", "against"]}
VERDICT_SCHEMA = {"type": "object", "properties": {"winner": {"type": "string"}, "reason": {"type": "string"},
                                                   "confidence": {"type": "number"}}, "required": ["winner", "reason", "confidence"]}

async def investigate(bb: Blackboard, name: str, theory: str):
    others = {n: t for n, t in bb.hypotheses.items() if n != name}
    out, c = await run(
        f"You are investigator {name}. The failing test is tests/test_utils.py::test_user_name_none_returns_empty "
        f"(it calls get_user_name(None) and expects ''). Your theory: {theory}.\nOther open theories: {json.dumps(others)}\n"
        f"Evidence already on the blackboard:\n{json.dumps(bb.evidence, indent=2)}\n"
        "Read the code and the test. Return evidence FOR your theory and evidence AGAINST each other theory "
        "(cite file:line). Be honest: if your theory is wrong, say so in 'for' with an empty list.",
        EVIDENCE_SCHEMA, allowed_tools=["Read", "Grep", "Glob"], effort="low")
    bb.cost += c
    bb.evidence.setdefault(name, []).extend(out["for"])
    for other, reasons in out["against"].items():
        bb.evidence.setdefault(f"against_{other}", []).extend(reasons)

async def debate(bb: Blackboard, rounds: int = 2):
    for r in range(rounds):                                                                   # (1) bounded rounds
        await asyncio.gather(*(investigate(bb, n, t) for n, t in bb.hypotheses.items()))     # (2) peers in parallel, shared blackboard
        print(f"after round {r + 1}: evidence counts = { {k: len(v) for k, v in bb.evidence.items()} }")
    verdict, c = await run("Judge which hypothesis best explains the failing test given this evidence. "
                           "Prefer the theory with concrete file:line support and no rebuttals.\n" + json.dumps(bb.evidence, indent=2),
                           VERDICT_SCHEMA, tools=[])                                          # (3) independent judge, no tools
    bb.cost += c
    return verdict

bb = Blackboard(hypotheses={
    "A": "get_user_name crashes because it subscripts user['name'] without checking for None",
    "B": "the test fails because of an import error: src.utils cannot be imported",
    "C": "calculate_average's division by zero is what makes the whole test module fail",
})

after round 1: evidence counts = {'C': 0, 'against_A': 2, 'against_B': 2, 'B': 0, 'against_C': 2, 'A': 1}
after round 2: evidence counts = {'C': 0, 'against_A': 2, 'against_B': 4, 'B': 0, 'against_C': 4, 'A': 2}

BLACKBOARD:
  C:
  against_A:
    - src/utils.py:9 - get_user_name(None) executes user["name"] which raises TypeError: 'NoneType' object is not subscriptable, directly crashing
    - src/utils.py:8 get_user_name does `return user["name"].upper()` with no None check — calling get_user_name(None) raises TypeError
  against_B:
    - src/utils.py and tests/test_utils.py:2 - `from src.utils import calculate_average, get_user_name` succeeds; both names exist in src/utils.py
    - tests/test_utils.py:2 - `from src.utils import calculate_average, get_user_name` succeeds since both symbols are defined in src/utils.py
  B:
  against_C:
    - src/utils.py:1-5 calculate_average divides by len(numbers); with numbers=[] this is 0/0 which raises ZeroDivisionError. However tests/test_u
    - calculate_average's division by zero claim doesn't hold [...] under pytest/unittest each test runs independently
  A:
    - src/utils.py:8 - get_user_name(user) does `return user["name"].upper()` with no None/missing-key check. Calling get_user_name(None) raises T

VERDICT: {'winner': 'A', 'confidence': 0.95, 'reason': 'Hypothesis A has concrete, uncontested file:line support: src/utils.py:8 shows
get_user_name returns user[name].upper() with no None check, causing a TypeError when called with None, which directly explains the
failing test [...]. The against_A notes actually confirm rather than refute A. B is refuted since the import at tests/test_utils.py:2
succeeds cleanly. C is refuted since pytest/unittest run tests independently, so a ZeroDivisionError in a different test cannot cause
this specific test to fail.'}
TOTAL COST: $0.2320

DELEGATION = """Objective: {objective}
Scope: only files under {scope}. Do not look elsewhere.
Output format: a JSON list of {{"file","line","issue","severity"}} and nothing else.
Tools: use Grep to locate, Read to confirm. Do not edit.
Stop when you have checked every file in scope or after 10 tool calls."""

# before: three thin wrappers around endpoints, opaque errors
@tool("list_slots", "List slots", {"user": str}) ...
@tool("create_event", "Create event", {"u": str, "t": str}) ...

# after: one workflow tool, unambiguous names, an error that teaches
@tool("schedule_meeting",
      "Find a free slot for every attendee in the next 7 days and book it. Returns the booked time and a calendar link. "
      "Use this instead of listing availability and creating events separately.",
      {"type": "object",
       "properties": {"attendee_user_ids": {"type": "array", "items": {"type": "string"}, "description": "User IDs, e.g. ['u_123']. Names are not accepted; call find_user first."},
                      "duration_minutes": {"type": "integer", "minimum": 15, "maximum": 240},
                      "earliest": {"type": "string", "description": "ISO 8601 date-time, e.g. 2026-10-17T09:00:00Z"}},
       "required": ["attendee_user_ids", "duration_minutes"]})
async def schedule_meeting(args):
    if not all(a.startswith("u_") for a in args["attendee_user_ids"]):
        return {"content": [{"type": "text", "text": "attendee_user_ids must be IDs like 'u_123'. You passed names; call find_user to resolve them first."}], "is_error": True}
    ...

options = ClaudeAgentOptions(
    allowed_tools=[
        "Bash(python3 *)", "Bash(pytest *)", "Bash(npm test*)",   # (1) pre-approve kinds of commands, not Bash wholesale
        "Bash(git status*)", "Bash(git diff*)", "Bash(gh pr view*)",
    ],
    disallowed_tools=["Bash(rm *)", "Bash(git push*)", "Bash(curl *)"],   # (2) denied in every mode, even bypass
    sandbox={"enabled": True, "autoAllowBashIfSandboxed": True,          # (3) or: sandbox everything and stop prompting
             "network": {"allowedDomains": ["pypi.org", "files.pythonhosted.org"]}},
)

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Grep", "Glob", "Agent", "Bash(python3 *)"],
    agents={"auditor": AgentDefinition(description="Audits one directory for issues.", prompt="...",
                                       tools=["Read", "Grep"], background=True, maxTurns=15)},   # (1) always non-blocking
    env={"CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "5", "CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS": "120000"},   # (2) caps
    max_budget_usd=3.0,
)
async with ClaudeSDKClient(options) as client:
    await client.query("Audit src/, tests/ and scripts/ in parallel with the auditor agent, then summarize.")
    async for m in client.receive_response():
        if isinstance(m, TaskStartedMessage):   print(f"started  {m.task_id}: {m.description}")
        elif isinstance(m, TaskProgressMessage): print(f"progress {m.task_id}: {m.usage['tool_uses']} tool uses, last={m.last_tool_name}")
        elif isinstance(m, TaskNotificationMessage):
            print(f"finished {m.task_id}: {m.status}, summary={(m.summary or "")[:80]}")
            if m.status == "failed": await client.stop_task(m.task_id)             # (3) react

# a two-phase coding session: plan read-only, then implement with tests as the stop condition
async with ClaudeSDKClient(ClaudeAgentOptions(cwd=REPO, permission_mode="plan", setting_sources=["project"],
                                              allowed_tools=["Read", "Grep", "Glob"],
                                              hooks={"Stop": [HookMatcher(hooks=[tests_must_pass])]})) as client:
    await client.query("Plan how to add input validation to find_user. Do not edit yet.")
    async for m in client.receive_response(): ...                       # review the plan (or have a reviewer agent score it)
    await client.set_permission_mode("acceptEdits")                      # phase 2: implement
    await client.query("Implement the plan. Add tests first, then make them pass.")
    async for m in client.receive_response(): ...

APPROVALS: set[str] = set()          # stands in for your approvals table (ticket system, Slack button, ...)
PENDING: dict[str, str] = {}         # session_id -> command awaiting approval
SEGMENT = re.compile(r"&&|\|\||;|\|")       # split a shell line into its commands

def runs_deploy(cmd: str) -> bool:
    """True when deploy.sh is EXECUTED (first word of any segment), not merely read with cat/head."""
    return any(seg.split()[0].endswith("deploy.sh") for seg in SEGMENT.split(cmd) if seg.split())

async def gate_deploys(input_data, tool_use_id, context):
    cmd = input_data["tool_input"].get("command", "")
    if input_data["tool_name"] == "Bash" and runs_deploy(cmd):
        if cmd in APPROVALS:                                                        # (1) on resume: a human said yes
            print(f"  [hook] approval found for {cmd!r}: allow")
            return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "allow",
                                           "updatedInput": input_data["tool_input"]}}
        PENDING[input_data["session_id"]] = cmd                                     # (2) first time: record and pause
        print(f"  [hook] {cmd!r} needs human approval: defer")
        return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "defer"}}
    return {}

OPTS = dict(cwd=str(WORK), allowed_tools=["Read", "Bash(./deploy.sh*)", "Bash(cat *)", "Bash(ls *)"], max_turns=6,
            setting_sources=[], hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[gate_deploys])]})

async def run_until_not_deferred(prompt, session_id=None):
    """Run; whenever the run pauses on a deferred tool, 'collect approval' and resume the same session."""
    for attempt in range(4):
        last = None
        async for m in query(prompt=prompt, options=ClaudeAgentOptions(resume=session_id, **OPTS)):
            if isinstance(m, ResultMessage):
                last, session_id = m, m.session_id
                print(f"  run {attempt + 1}: subtype={m.subtype} stop_reason={m.stop_reason!r}"
                      + (f" deferred={m.deferred_tool_use.name}({m.deferred_tool_use.input.get('command')!r})" if m.deferred_tool_use else ""))
                break                                                               # stop at the FIRST result: a deferred run is over
        if last is None or last.stop_reason != "tool_deferred":
            return last
        print("  ... the process could exit here; hours pass; a human approves in your UI ...")
        APPROVALS.add(PENDING.pop(session_id))                                      # (3) record the approval
        prompt = ""                                                                 # (4) resume with no new prompt
    return last

result = await run_until_not_deferred("Deploy the app by running ./deploy.sh, then report what it printed.")

  [hook] './deploy.sh' needs human approval: defer
  run 1: subtype=success stop_reason='tool_deferred' deferred=Bash('./deploy.sh')
  ... the process could exit here; hours pass; a human approves in your UI ...
  [hook] approval found for './deploy.sh': allow
  run 2: subtype=success stop_reason='end_turn'

final: success -> 'Deploy ran successfully. It printed:\n\n```\ndeploying version 1.2.3 to production\n```'

JUDGE_SCHEMA = {"type": "object", "properties": {"score": {"type": "number", "minimum": 0, "maximum": 1},
                                                 "passed": {"type": "boolean"}, "notes": {"type": "string"}},
                "required": ["score", "passed", "notes"]}

async def judge(task: str, rubric: str) -> dict:
    final_state = open(f"{PROJECT}/src/utils.py").read()                             # (1) the END STATE, not the transcript
    tests_pass, test_output = run_tests()                                            # (2) deterministic signal first
    verdict, _ = await run(                                                          # (3) fresh-context judge, no tools
        f"Task: {task}\nRubric:\n{rubric}\n\nTests pass: {tests_pass}\n{test_output[-400:]}\n\nFinal file:\n{final_state}\n"
        "Score 0.0-1.0 against the rubric and decide pass/fail.", JUDGE_SCHEMA, tools=[], model="claude-opus-5-5")
    return verdict

// sketch: adapt the SDK stream to UI events
for await (const m of query({ prompt, options: { includePartialMessages: true, canUseTool, toolConfig: { askUserQuestion: { previewFormat: "html" } } } })) {
  if (m.type === "stream_event" && m.event.type === "content_block_delta" && m.event.delta.type === "text_delta") ui.appendText(m.event.delta.text);
  else if (m.type === "assistant") for (const b of m.message.content) if (b.type === "tool_use") ui.showToolCard(b.name, b.input);
  else if (m.type === "task_progress") ui.updateProgress(m.task_id, m.usage);
  else if (m.type === "result") ui.finish(m.subtype, m.total_cost_usd);
}
// canUseTool renders AskUserQuestion as a form, with option.preview as an HTML snippet

Is the task one well-specified step?                 ──► single call (+ retrieval, structured output)
Can the steps be fixed in advance?                   ──► workflow (chain / route / parallel) in code      (§3)
Steps unpredictable but the goal checkable?          ──► agent loop + Stop-hook verifier + bounds          (§4)
Outgrows one context?                                ──► subagents (orchestrator-workers)                 (P2 §4)
Orchestration must be reviewable and rerunnable?     ──► code-owned graph or dynamic workflow              (§7)
Work recurs on a schedule or event?                  ──► scheduler calling query() with resume + store     (§4.5)
Must Anthropic host it?                              ──► Managed Agents (outcomes for one-deliverable sessions)