T15 — Hook lifecycle system (6 events, 4 types, hot reload)

class HookExecutor:
    def __init__(self, hooks: list[Hook]):
        self._hooks = hooks
        self._reloader = HookReloader(hooks)

    async def execute(
        self, event: HookEvent, payload: dict
    ) -> HookResult:
        await self._reloader.reload_if_changed()
        relevant = [h for h in self._hooks if event in h.events]
        for hook in relevant:
            result = await self._dispatch(hook, payload)
            if result.blocked:           # only PRE_TOOL_USE can block
                return result
        return HookResult(blocked=False)

    async def _dispatch(self, hook: Hook, payload: dict) -> HookResult:
        match hook.type:
            case HookType.COMMAND:
                return await _run_command(hook, payload)
            case HookType.PROMPT:
                return await _run_prompt_hook(hook, payload)
            case HookType.HTTP:
                return await _run_http_hook(hook, payload)
            case HookType.AGENT:
                return await _run_agent_hook(hook, payload)

class HookReloader:
    def __init__(self, hooks_path: Path):
        self._path = hooks_path
        self._mtime: float | None = None

    async def reload_if_changed(self) -> bool:
        try:
            mtime = self._path.stat().st_mtime
        except FileNotFoundError:
            return False
        if mtime != self._mtime:
            self._mtime = mtime
            return True   # caller re-parses hooks file
        return False

# BUG: PostToolUse hook formatter thoát với error code khi file bị lock
# → HookExecutor nhận returncode != 0
# → Workflow bị block mặc dù POST_TOOL_USE không được phép block

# Claude Code docs fix:
# "End commands with ; exit 0 for PostToolUse hooks
#  so a hook failure doesn't block Claude's workflow unnecessarily"

# BAD hook command:
hooks:
  - events: [POST_TOOL_USE]
    command: "black {file} && isort {file}"  # fails if file locked

# GOOD hook command:
hooks:
  - events: [POST_TOOL_USE]
    command: "black {file} && isort {file}; exit 0"  # always exits 0

# Scenario: editor lưu hooks.yaml theo kiểu write-temp-then-rename
# 1. HookReloader.reload_if_changed() → mtime thay đổi → True
# 2. Caller bắt đầu re-parse hooks.yaml
# 3. Editor chưa rename xong → hooks.yaml bị truncated
# 4. json.loads / yaml.safe_load ném exception
# 5. HookExecutor falls back về empty hook list
#    → tất cả hooks bị disabled silently trong 1 execution

# Mitigation: parse với try/except, giữ lại hooks cũ nếu parse fail
try:
    new_hooks = parse_hooks_file(path)
    self._hooks = new_hooks
except Exception as e:
    logger.warning(f"Hot reload failed, keeping old hooks: {e}")
    # _hooks unchanged — degrade gracefully

# Prompt hook = gọi LLM để evaluate hook condition
# Nếu đặt vào PRE_TOOL_USE → mỗi tool call phải chờ LLM response

hooks:
  - events: [PRE_TOOL_USE]
    type: PROMPT
    prompt: "Is this tool call safe? {tool_name} with args {args}"
    # → Adds 500ms-2s latency mỗi tool call
    # → Nested: tool calls bên trong hook cũng trigger hooks

# Pattern tốt hơn: dùng COMMAND hook với simple rule-based check
# Reserve PROMPT hook cho PostToolUse audit (non-blocking, async)
hooks:
  - events: [PRE_TOOL_USE]
    type: COMMAND
    command: "check_tool_allowlist.sh {tool_name}; exit $?"
  - events: [POST_TOOL_USE]
    type: PROMPT
    prompt: "Summarize what {tool_name} did: {result}"

import asyncio, subprocess, json
from enum import Enum
from dataclasses import dataclass
from pathlib import Path

class HookEvent(str, Enum):
    SESSION_START = "SESSION_START"
    SESSION_END = "SESSION_END"
    PRE_COMPACT = "PRE_COMPACT"
    POST_COMPACT = "POST_COMPACT"
    PRE_TOOL_USE = "PRE_TOOL_USE"
    POST_TOOL_USE = "POST_TOOL_USE"

@dataclass
class Hook:
    events: list[HookEvent]
    command: str  # shell command for Command type
    can_block: bool = False  # only PRE_TOOL_USE hooks

@dataclass
class HookResult:
    blocked: bool
    reason: str = ""

async def run_command_hook(hook: Hook, payload: dict) -> HookResult:
    env_payload = json.dumps(payload)
    proc = await asyncio.create_subprocess_shell(
        hook.command,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, _ = await asyncio.wait_for(proc.communicate(env_payload.encode()), timeout=30)
    if proc.returncode != 0 and hook.can_block:
        return HookResult(blocked=True, reason=stdout.decode())
    return HookResult(blocked=False)