p1b_goal_loop_blocked.py

"""Pattern 1b: same loop, but the goal has a criterion the prompt never mentioned, so the first stop is blocked. 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
import os
CHANGELOG = os.path.join(PROJECT, "CHANGELOG.md")
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()
    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 {}  # 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": "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 *)"],
        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())