用户请求 → 你的后端 → SDK query(resume=session_id, sessionStore=store) → Claude Code 子进程

# SDK 产出的 session_id，直接用 CLI 继续
claude -p "继续之前的工作" --resume "$session_id"

# 交互模式接管
claude --resume "$session_id"

# 第一步：执行分析，捕获 session_id
session_id=$(claude -p "Review this PR for issues" \
  --output-format json --bare \
  --allowedTools "Read,Glob,Grep" | jq -r '.session_id')

# 第二步：基于分析结果继续
claude -p "Fix the issues you found" \
  --resume "$session_id" --bare \
  --allowedTools "Read,Edit,Write,Bash"

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Glob", "Grep"])

    async with ClaudeSDKClient(options=options) as client:
        # 第一轮
        await client.query("分析 auth 模块")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

        # 第二轮：自动延续同一会话
        await client.query("重构为 JWT 方案")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

asyncio.run(main())

class SessionStore(Protocol):
    async def append(self, key: SessionKey, entries: list[SessionStoreEntry]) -> None: ...
    async def load(self, key: SessionKey) -> list[SessionStoreEntry] | None: ...

hook 触发: tool_use_id=846qh, is_resumed=False → defer
hook 触发: tool_use_id=846qh, is_resumed=True  → allow (同一个 id!)
session JSONL: tool_result for_id=37J9QLbN427H (与 defer 时的 tool_use 一致)

# === executor.py 完整 HITL 实现 ===

from claude_agent_sdk import (
    query, ClaudeAgentOptions, ResultMessage, AssistantMessage, SystemMessage,
    create_sdk_mcp_server, tool, HookMatcher,
)

# --- Tool 定义 ---
@tool("request_approval", "提交审批请求，暂停执行等待人工决策。", {
    "action": str, "title": str, "context": str, "next_steps": str,
})
async def _request_approval_tool(args: dict) -> dict:
    decision = args.get("_decision", "approved")
    feedback = args.get("_feedback", "")
    if decision == "approved":
        return {"content": [{"type": "text", "text": f"✅ Approved. Proceed with: {args.get('next_steps', '')}"}]}
    elif decision == "pending":
        # 关键：用 ✅ 开头 + 明确 "任务完成" 措辞，防止 agent 误判为拒绝而重试
        return {"content": [{"type": "text", "text": "✅ 审批已成功提交，正在等待人工审批。你的任务到此完成，无需任何后续操作。请立即结束当前对话。"}]}
    else:
        return {"content": [{"type": "text", "text": f"❌ Rejected: {feedback}"}]}

_hitl_server = create_sdk_mcp_server(name="hitl", tools=[_request_approval_tool])


async def execute_task(run: TaskRun, workspaces_dir: str) -> TaskResult:
    workspace_path = str(Path(workspaces_dir) / run.workspace)
    is_resumed = bool(run.resume_session)

    # --- PreToolUse Hook ---
    _resume_state = {"first_done": False, "pending_defer": None}

    async def _hitl_hook(input_data, tool_use_id, context):
        tool_name = input_data.get("tool_name", "")
        if "request_approval" not in tool_name:
            return {}

        # Resume 后第一次调用（SDK 重放的 deferred tool）：注入 decision
        if is_resumed and not _resume_state["first_done"]:
            _resume_state["first_done"] = True
            original_input = input_data.get("tool_input", {})
            updated = {**original_input, "_decision": run.hitl_decision, "_feedback": run.hitl_feedback}
            return {"hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "updatedInput": updated,
            }}

        # Reject resume 后续调用（agent 修改方案后重新提交）：
        # allow 但标记为 pending，捕获新 tool_use_id
        if is_resumed and run.hitl_decision == "rejected":
            original_input = input_data.get("tool_input", {})
            _resume_state["pending_defer"] = {
                "tool_use_id": tool_use_id,
                "input": original_input,
            }
            updated = {**original_input, "_decision": "pending"}
            return {"hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "updatedInput": updated,
            }}

        # Fresh session 或 approve resume 后续调用：正常 defer
        return {"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "defer",
        }}

    # --- Prompt 构造 ---
    # re-defer approve 时，用 prompt 文本传递决定（SDK 不会重放 tool call）
    prompt = run.prompt
    if is_resumed and not run.prompt and run.hitl_decision == "approved":
        prompt = "✅ Approved. Proceed with the action you submitted for approval."
    elif is_resumed and not run.prompt and run.hitl_decision == "rejected":
        prompt = f"❌ Rejected: {run.hitl_feedback}" if run.hitl_feedback else "❌ Rejected. Stop."

    # --- SDK Options ---
    # system_prompt 只加最小 HITL 提示，不限制工具并发（效率优先）
    options = ClaudeAgentOptions(
        cwd=workspace_path,
        system_prompt="When request_approval returns a message containing '审批已成功提交', your task is DONE. End immediately without any further tool calls or retries.",
        permission_mode="bypassPermissions",
        allowed_tools=["Bash(*)", "Read(*)", "Write(*)", "mcp__hitl__request_approval"],
        max_turns=run.max_turns,
        mcp_servers={"hitl": _hitl_server},
        hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[_hitl_hook])]},
    )
    if run.resume_session:
        options.resume = run.resume_session

    # --- Execute & Result ---
    result_msg = None
    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, ResultMessage):
            result_msg = msg

    if result_msg:
        deferred = None
        if hasattr(result_msg, "deferred_tool_use") and result_msg.deferred_tool_use:
            dtu = result_msg.deferred_tool_use
            deferred = {"id": dtu.id, "name": dtu.name, "input": dtu.input}

        # 关键：检测 re-defer（reject 后 agent 重新提交了 request_approval）
        if not deferred and _resume_state.get("pending_defer"):
            pd = _resume_state["pending_defer"]
            deferred = {"id": pd["tool_use_id"], "name": "mcp__hitl__request_approval", "input": pd["input"]}
            return TaskResult(
                task_id=run.task_id, success=True,
                result=result_msg.result, cost_usd=result_msg.total_cost_usd,
                duration_ms=result_msg.duration_ms, num_turns=result_msg.num_turns,
                session_id=result_msg.session_id,
                stop_reason="tool_deferred",  # 合成 deferred → 上层创建新 approval
                deferred_tool_use=deferred,
            )

        return TaskResult(
            task_id=run.task_id, success=(result_msg.subtype == "success"),
            result=result_msg.result, cost_usd=result_msg.total_cost_usd,
            duration_ms=result_msg.duration_ms, num_turns=result_msg.num_turns,
            session_id=result_msg.session_id, stop_reason=result_msg.stop_reason,
            deferred_tool_use=deferred,
        )

[Fresh Session]
  User prompt → Agent calls request_approval
  Hook: defer → SDK stops → ResultMessage(stop_reason="tool_deferred")
  → 创建 Approval 记录

[Resume 1: Reject]
  resume(session, decision=rejected, feedback="改成xxx")
  Hook (first call): allow + inject {_decision: "rejected", _feedback: "改成xxx"}
  Tool returns: "❌ Rejected: 改成xxx"
  Agent 修改方案 → 再次调用 request_approval
  Hook (subsequent, only for rejected): allow + inject {_decision: "pending"} + 捕获 tool_use_id
  Tool returns: "✅ 审批已成功提交...任务到此完成"
  Agent 停止（一句话结束）→ ResultMessage(stop_reason="end_turn")
  检测 pending_defer → 合成 stop_reason="tool_deferred"
  → 创建新的 Approval 记录（title/context 是 agent 修改后的版本）

[Resume 2: Approve]
  resume(session, decision=approved) + prompt="✅ Approved. Proceed."
  Agent 看到 prompt → 执行操作 → 完成

result_msg = None
async for msg in query(...):
    if isinstance(msg, ResultMessage):
        result_msg = msg
# 处理 result_msg（最后一条才是最终结果）