p9_circuit_breaker.py

"""Pattern 9: a circuit breaker on a goal loop. Part of the verifier's condition depends on something
OUTSIDE the agent's control (a staging health check that is down), so retrying can never satisfy it.
The Stop hook must notice that the verifier's state stopped changing and stop, instead of looping."""

import asyncio
import os
import shutil
import tempfile
from pathlib import Path

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher, ResultMessage, query
from demo_util import PRISTINE, PROJECT

WORK = Path(tempfile.mkdtemp(prefix="breaker-")) / "project"
shutil.copytree(PRISTINE, WORK)
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
        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 def main():
    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),
    ):
        if isinstance(m, ResultMessage):
            print(f"\n[result] {m.subtype} after {m.num_turns} turns, ${m.total_cost_usd:.4f}")
            print(f"[result] {(m.result or '')[:400]}")
    print("\nverifier log (iteration, signature, stalls):", state["log"])


asyncio.run(main())