Function run_single_hook

tutorials/19_process_and_pipe_communication.py:615–663

对应 hooks.rs:152-200 的 run_command() 完整流程: 启动子进程 → stdin 传 JSON → 读 stdout → 解析退出码

(
        script: str,
        event: str,
        tool_name: str,
        tool_input: str,
        payload: str,
    )

            return self.denied

    def run_single_hook(
        script: str,
        event: str,
        tool_name: str,
        tool_input: str,
        payload: str,
    ) -> HookCommandOutcome:
        """
        对应 hooks.rs:152-200 的 run_command()
        完整流程: 启动子进程 → stdin 传 JSON → 读 stdout → 解析退出码
        """
        # 构造子进程环境变量 (hooks.rs:166-172)
        env = os.environ.copy()
        env["HOOK_EVENT"] = event
        env["HOOK_TOOL_NAME"] = tool_name
        env["HOOK_TOOL_INPUT"] = tool_input
        env["HOOK_TOOL_IS_ERROR"] = "0"

        try:
            result = subprocess.run(
                ["python3", "-c", script],
                input=payload,          # stdin 传 JSON (hooks.rs:174)
                capture_output=True,    # stdout + stderr (hooks.rs:163-165)
                text=True,
                env=env,                # 环境变量 (hooks.rs:166-172)
                timeout=5,             # 防止 hook 挂死
            )

            stdout = result.stdout.strip()
            message = stdout if stdout else None

            # 退出码协议 (hooks.rs:179-196)
            if result.returncode == 0:
                return HookCommandOutcome("Allow", message)
            elif result.returncode == 2:
                return HookCommandOutcome(
                    "Deny",
                    message or f"Hook denied tool `{tool_name}`",
                )
            else:
                return HookCommandOutcome(
                    "Warn",
                    f"Hook exited with code {result.returncode}: {stdout or result.stderr.strip()}",
                )

        except subprocess.TimeoutExpired:
            return HookCommandOutcome("Warn", "Hook 超时被杀死")
        except Exception as e:
            return HookCommandOutcome("Warn", f"Hook 执行失败: {e}")

    def run_hooks(
        hooks: list[str],