p1_goal_loop.py

"""Pattern 1: a goal-based loop. A Stop hook refuses to let the agent stop until a
machine-checkable condition holds (the test suite passes), with a hard iteration cap."""

import asyncio
import subprocess

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

MAX_ITERATIONS = 4
state = {"iterations": 0, "history": []}


def run_tests() -> tuple[bool, str]:
    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):
    """Stop hook: the verifier. Deterministic, independent of the model's own judgement."""
    state["iterations"] += 1
    passed, output = run_tests()
    state["history"].append(passed)
    print(f"  [Stop hook] iteration {state['iterations']}: tests {'PASS' if passed else 'FAIL'}")
    if passed:
        return {}  # goal met: let the agent stop
    if state["iterations"] >= MAX_ITERATIONS:
        print("  [Stop hook] iteration cap reached, allowing stop")
        return {}  # bounded execution: never loop forever
    # Not met: block the stop and feed the evidence back as the next instruction
    return {
        "decision": "block",
        "reason": f"The goal is not met. `python3 -m unittest -q tests.test_utils` still fails:\n{output}\n"
                  f"Fix src/utils.py (not the tests) and re-run the suite. Iteration {state['iterations']} of {MAX_ITERATIONS}.",
    }


async def main():
    options = ClaudeAgentOptions(
        cwd=PROJECT,
        permission_mode="acceptEdits",
        allowed_tools=["Read", "Edit", "Bash(python3 *)"],
        hooks={"Stop": [HookMatcher(hooks=[enforce_goal])]},
        max_turns=20,
    )
    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`.",
        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()
    print(f"\nFinal verification outside the agent: tests pass = {passed}; "
          f"Stop-hook verdicts = {state['history']}")


asyncio.run(main())