p7_deferred_approval.py

"""Pattern 7: deferred approval. A PreToolUse hook pauses the run at a risky command; the process ends with
stop_reason=tool_deferred; later, with approval recorded, the session is resumed and the same hook allows the call."""

import asyncio
import re
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="deploy-")) / "project"
shutil.copytree(PRISTINE, WORK)
(WORK / "deploy.sh").write_text("#!/bin/sh\necho 'deploying version 1.2.3 to production'\n")
(WORK / "deploy.sh").chmod(0o755)

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


async def main():
    result = await run_until_not_deferred("Deploy the app by running ./deploy.sh, then report what it printed.")
    print(f"\nfinal: {result.subtype} -> {result.result!r}")


asyncio.run(main())