---
name: sdk-test-hooks-deep
type: analysis
title: "Hook System Deep-Dive — Lifecycle Interception Architecture"
description: "Complete analysis of the Antigravity hook system: 9 hook types, Inspect/Decide/Transform categories, 3-level context hierarchy, HookRunner dispatch, and policy system with 9 priority levels"
tags: [hooks, lifecycle, policy, security, antigravity, tool-calls, dispatch]
timestamp: 2026-07-22
---

# Hook System Deep-Dive — Lifecycle Interception Architecture

## Overview

The hook system provides **9 lifecycle interception points** organized into three semantic categories. It is the primary mechanism for observability, policy enforcement, and data transformation in the Antigravity SDK.

The hook system spans three files:
- `hooks.py` (289 lines) — Hook type definitions, base classes, decorator factories
- `hook_runner.py` (284 lines) — Registration and dispatch engine
- `policy.py` (905 lines) — Declarative policy system layered on top of `PreToolCallDecideHook`

## Hook Category Architecture

```mermaid
graph TB
    subgraph "InspectHook[T] — Read-only, Non-blocking"
        direction TB
        OSS["OnSessionStartHook<br/>Session begins"]
        OSE["OnSessionEndHook<br/>Session ends"]
        PT["PostTurnHook<br/>data: str (response text)"]
        PTC["PostToolCallHook<br/>data: ToolResult"]
        OC["OnCompactionHook<br/>Context compacted"]
    end

    subgraph "DecideHook[T] — Read-only, Blocking"
        direction TB
        PreT["PreTurnHook<br/>data: Content (prompt)"]
        PreTC["PreToolCallDecideHook<br/>data: ToolCall"]
    end

    subgraph "TransformHook[T, R] — Modifying, Blocking"
        direction TB
        OTE["OnToolErrorHook<br/>Exception → recovery value"]
        OI["OnInteractionHook<br/>AskQuestionSpec → Result"]
    end

    style InspectHook fill:#2a4a6b,stroke:#4a8abf
    style DecideHook fill:#6b4a2a,stroke:#ab8a4a
    style TransformHook fill:#4a6b2a,stroke:#8abf4a
```

## Hook Type Reference

### Category 1: InspectHook (Read-only, Non-blocking)

Inspect hooks observe events without modifying them. They always return `None`. They never block the agent loop.

#### OnSessionStartHook

```python
class OnSessionStartHook(InspectHook[None]):
    """Called when the agent session begins.
    
    Data: None
    Return: None (always)
    Blocking: No
    """
```

**Dispatch point**: After `LocalConnectionStrategy.__aenter__()` completes the handshake and before the first message is sent.

**Use cases**:
- Initialize external services
- Log session start with metadata
- Set up monitoring dashboards
- Warm caches

#### OnSessionEndHook

```python
class OnSessionEndHook(InspectHook[None]):
    """Called when the agent session ends.
    
    Data: None
    Return: None (always)
    Blocking: No
    """
```

**Dispatch point**: First step of the 7-step disconnect sequence in `LocalConnection.disconnect()`.

**Use cases**:
- Flush logs
- Close external connections
- Report session telemetry
- Clean up temporary files

#### PostTurnHook

```python
class PostTurnHook(InspectHook[str]):
    """Called after a turn completes.
    
    Data: str (the model's response text)
    Return: None (always)
    Blocking: No
    """
```

**Dispatch point**: After `receive_steps()` yields a step with `is_complete_response=True`.

**Use cases**:
- Log responses
- Track response quality metrics
- Feed responses to analytics pipelines

#### PostToolCallHook

```python
class PostToolCallHook(InspectHook[types.ToolResult]):
    """Called after a tool call executes successfully.
    
    Data: ToolResult (the tool's output)
    Return: None (always)
    Blocking: No
    """
```

**Dispatch point**: After `ToolRunner.process_tool_calls()` returns, or after a built-in tool completes with `state=DONE`.

**Special case for subagents**: When a subagent trajectory completes, `PostToolCallHook` is dispatched with the subagent's last model response as the `ToolResult` content.

**Use cases**:
- Audit tool usage
- Track tool execution latency
- Log tool results for debugging
- Feed tool results to monitoring

#### OnCompactionHook

```python
class OnCompactionHook(InspectHook):
    """Called when the context window is compacted.
    
    Data: compaction metadata
    Return: None (always)
    Blocking: No
    """
```

**Dispatch point**: When the harness performs context compaction (triggered by `compaction_threshold` in `HarnessConfig`).

**Use cases**:
- Track compaction frequency
- Log what was lost in compaction
- Trigger external state persistence

### Category 2: DecideHook (Read-only, Blocking)

Decide hooks evaluate a decision and return a `HookResult`. They block the agent loop until a decision is made.

#### PreTurnHook

```python
class PreTurnHook(DecideHook[types.Content]):
    """Called before a turn starts.
    
    Data: Content (the user's prompt)
    Return: HookResult(allow=True/False, message=...)
    Blocking: Yes — short-circuits on first allow=False
    """
```

**Dispatch point**: Inside `Conversation.chat()` before `Connection.send()` is called.

**HookResult semantics**:
- `allow=True` — Turn proceeds normally
- `allow=False` — Turn is blocked; the `message` is returned as an error

**Use cases**:
- Rate limiting
- Input validation
- Content filtering
- Prompt injection detection

#### PreToolCallDecideHook

```python
class PreToolCallDecideHook(DecideHook[types.ToolCall]):
    """Called before a tool call executes.
    
    Data: ToolCall (name, args)
    Return: HookResult(allow=True/False, message=...)
    Blocking: Yes — short-circuits on first allow=False
    """
```

**Dispatch point**: Two locations:
1. **Built-in tools**: In `LocalConnection` when `StepUpdate` contains a `tool_confirmation_request`
2. **Python tools**: In `LocalConnection._handle_tool_call()` before `ToolRunner.process_tool_calls()`

**This is the most critical hook for security**. The entire policy system is built on top of it.

**Use cases**:
- Tool access control (allow/deny based on name, args, context)
- User confirmation prompts
- Audit logging before execution
- Argument sanitization

### Category 3: TransformHook (Modifying, Blocking)

Transform hooks can modify data or provide recovery values. They block until a non-None value is returned or all hooks are exhausted.

#### OnToolErrorHook

```python
class OnToolErrorHook(TransformHook[Exception, Any]):
    """Called when a tool call fails.
    
    Data: Exception (the error)
    Return: recovery value (replaces the error) or None (propagate error)
    Blocking: Yes — short-circuits on first non-None return
    """
```

**Dispatch point**: When `ToolRunner.process_tool_calls()` catches an exception, or when a built-in tool transitions to `state=ERROR`.

**Recovery semantics**:
- Return a value → the value is used as the tool result (error is swallowed)
- Return `None` → the error propagates to the model as a `ToolResult(error=...)`

**Use cases**:
- Retry logic (catch transient errors, retry)
- Fallback values (return cached data on network error)
- Error transformation (convert stack traces to user-friendly messages)

#### OnInteractionHook

```python
class OnInteractionHook(TransformHook[AskQuestionInteractionSpec, QuestionHookResult]):
    """Called when the agent needs user interaction.
    
    Data: AskQuestionInteractionSpec (questions from the model)
    Return: QuestionHookResult (user's answers) or None (fall through)
    Blocking: Yes — short-circuits on first non-None return
    """
```

**Dispatch point**: When `StepUpdate` contains a `questions_request` from the model.

**Use cases**:
- Automated testing (inject predefined answers)
- CLI REPL (prompt user for input)
- Headless mode (return default answers)

## Context Hierarchy

The hook system uses a 3-level context hierarchy for state management within hook chains.

```mermaid
graph TB
    subgraph "SessionContext — Per Session"
        SC["Key-value store<br/>Visible to ALL hooks in session"]
        SC1["get(key, default)"]
        SC2["set(key, value)"]
    end

    subgraph "TurnContext — Per Turn"
        TC["Key-value store<br/>Visible to hooks in this turn"]
        TC1["get(key, default)"]
        TC2["set(key, value)"]
    end

    subgraph "OperationContext — Per Tool Call"
        OC["Key-value store<br/>Visible to this tool call only"]
        OC1["get(key, default)"]
        OC2["set(key, value)"]
    end

    SC -->|"parent lookup"| TC
    TC -->|"parent lookup"| OC

    style SC fill:#2a4a6b
    style TC fill:#4a6b2a
    style OC fill:#6b4a2a
```

### Visibility Rules

| Context | Can See Own State | Can See Parent State | Can See Child State |
|---------|-------------------|---------------------|---------------------|
| SessionContext | Yes | N/A (root) | No |
| TurnContext | Yes | Yes (Session) | No |
| OperationContext | Yes | Yes (Turn + Session) | No |

State flows downward via parent chain lookup. Setting a key in `OperationContext` does NOT affect `TurnContext` or `SessionContext`.

### Context Lifecycle

| Context | Created | Destroyed |
|---------|---------|-----------|
| SessionContext | `dispatch_session_start()` | `dispatch_session_end()` |
| TurnContext | `dispatch_pre_turn()` | After `dispatch_post_turn()` |
| OperationContext | `dispatch_pre_tool_call()` | After `dispatch_post_tool_call()` or `dispatch_on_tool_error()` |

### Independence from ToolContext

**Important**: `HookContext` (Session/Turn/Operation) and `ToolContext` are completely independent. They do NOT share state. A tool cannot read hook context, and a hook cannot read tool context.

## Decorator API

Each hook type has a corresponding decorator for functional-style registration:

```python
from google.antigravity.hooks import hooks

# Inspect hooks — return None
@hooks.on_session_start
async def log_start():
    print("Session started")

@hooks.on_session_end
async def log_end():
    print("Session ended")

@hooks.post_turn
async def log_response(response_text: str):
    print(f"Response: {response_text[:100]}...")

@hooks.post_tool_call
async def log_tool(tool_result):
    print(f"Tool completed: {tool_result}")

@hooks.on_compaction
async def log_compaction(data):
    print("Context compacted")

# Decide hooks — return HookResult
@hooks.pre_turn
async def validate_input(prompt):
    if len(prompt) > 10000:
        return hooks.HookResult(allow=False, message="Prompt too long")
    return hooks.HookResult(allow=True)

@hooks.pre_tool_call_decide
async def block_dangerous(tool_call):
    if tool_call.name == "run_command" and "rm -rf" in tool_call.args.get("CommandLine", ""):
        return hooks.HookResult(allow=False, message="Blocked dangerous command")
    return hooks.HookResult(allow=True)

# Transform hooks — return transformed value or None
@hooks.on_tool_error
async def retry_on_timeout(error):
    if isinstance(error, TimeoutError):
        return "Request timed out, please try again"
    return None  # Propagate error

@hooks.on_interaction
async def auto_answer(spec):
    return hooks.QuestionHookResult(answers={"confirm": "yes"})
```

## HookRunner — Registration and Dispatch

### Registration

```python
class HookRunner:
    def __init__(self):
        self._session_start_hooks: list[OnSessionStartHook] = []
        self._session_end_hooks: list[OnSessionEndHook] = []
        self._pre_turn_hooks: list[PreTurnHook] = []
        self._post_turn_hooks: list[PostTurnHook] = []
        self._pre_tool_call_hooks: list[PreToolCallDecideHook] = []
        self._post_tool_call_hooks: list[PostToolCallHook] = []
        self._on_tool_error_hooks: list[OnToolErrorHook] = []
        self._on_interaction_hooks: list[OnInteractionHook] = []
        self._on_compaction_hooks: list[OnCompactionHook] = []

    def register_hook(self, hook: Any) -> None:
        """Infers hook type via isinstance checks and adds to the correct list."""
```

The `register_hook()` method uses `isinstance` checks against a registry of all 9 hook types. This allows registering hooks as class instances or decorated functions.

### Dispatch Flow

```mermaid
sequenceDiagram
    participant Conn as Connection
    participant HR as HookRunner
    participant H1 as Hook 1
    participant H2 as Hook 2
    participant H3 as Hook 3

    Note over HR: Dispatch order: registration order

    Conn->>HR: dispatch_pre_tool_call(tool_call)
    
    HR->>H1: execute(tool_call)
    H1-->>HR: HookResult(allow=True)
    
    HR->>H2: execute(tool_call)
    H2-->>HR: HookResult(allow=False, "blocked")
    
    Note over HR: Short-circuit: first allow=False
    HR-->>Conn: HookResult(allow=False, "blocked")
    Note over H3: H3 never executes
```

### Dispatch Semantics by Category

| Category | Iteration | Short-circuit Condition | Return |
|----------|-----------|------------------------|--------|
| Inspect | All hooks | Never (all execute) | None |
| Decide | In order | First `allow=False` | HookResult |
| Transform | In order | First non-None return | Transformed value |

### Dispatch Methods

```python
class HookRunner:
    # Session lifecycle
    async def dispatch_session_start(self) -> None:
        """Runs all OnSessionStartHook hooks in order."""
    
    async def dispatch_session_end(self) -> None:
        """Runs all OnSessionEndHook hooks in order."""
    
    # Turn lifecycle
    async def dispatch_pre_turn(self, prompt) -> tuple[HookResult, TurnContext]:
        """Runs PreTurnHook hooks. Returns result + new TurnContext."""
    
    async def dispatch_post_turn(self, turn_context, response) -> None:
        """Runs PostTurnHook hooks with the response text."""
    
    # Tool lifecycle
    async def dispatch_pre_tool_call(self, turn_context, tool_call) -> tuple[HookResult, ToolCall, OperationContext]:
        """Runs PreToolCallDecideHook hooks. Returns result + (possibly modified) ToolCall + OperationContext."""
    
    async def dispatch_post_tool_call(self, op_context, result) -> None:
        """Runs PostToolCallHook hooks with the tool result."""
    
    async def dispatch_on_tool_error(self, op_context, error) -> tuple[HookResult, Any]:
        """Runs OnToolErrorHook hooks. Returns result + recovery value."""
    
    # Interaction
    async def dispatch_interaction(self, turn_context, spec) -> tuple[HookResult, Any, OperationContext]:
        """Runs OnInteractionHook hooks with question spec."""
    
    # Compaction
    async def dispatch_compaction(self, turn_context, data) -> None:
        """Runs OnCompactionHook hooks."""
```

## Policy System

The policy system provides a **declarative API** layered on top of `PreToolCallDecideHook`. It is the recommended way to control tool access.

### Architecture

```mermaid
graph TB
    subgraph "Policy API (Declarative)"
        PB["policy.allow()"]
        PD["policy.deny()"]
        PAU["policy.ask_user()"]
        PSF["policy.safe_defaults()"]
        PW["policy.workspace_only()"]
        PCONF["policy.confirm_run_command()"]
    end

    subgraph "Policy Engine"
        PE["policy.enforce()<br/>Converts policies → PreToolCallDecideHook"]
        PH["_PolicyDecideHook<br/>Single hook instance"]
    end

    subgraph "Hook System"
        HR["HookRunner"]
    end

    PB --> PE
    PD --> PE
    PAU --> PE
    PSF --> PE
    PW --> PE
    PCONF --> PE
    PE --> PH
    PH --> HR
```

### 9 Priority Levels

Policies are evaluated using a priority-based model. Lower index = higher priority.

| Level | Specificity | Decision | Example | Builder |
|-------|------------|----------|---------|---------|
| 0 | Specific tool | DENY | `deny("run_command")` | `policy.deny(tool)` |
| 1 | Specific tool | ASK_USER | Ask before `run_command` | `policy.ask_user(tool, handler=fn)` |
| 2 | Specific tool | ALLOW | Allow `view_file` | `policy.allow(tool)` |
| 3 | Prefix (MCP server) | DENY | Deny all tools on server X | `policy.deny(mcp_server)` |
| 4 | Prefix (MCP server) | ASK_USER | Ask for all tools on server X | `policy.ask_user(mcp_server, handler=fn)` |
| 5 | Prefix (MCP server) | ALLOW | Allow all tools on server X | `policy.allow(mcp_server)` |
| 6 | Global wildcard | DENY | `deny("*")` — block everything | `policy.deny("*")` |
| 7 | Global wildcard | ASK_USER | Ask for everything | `policy.ask_user("*", handler=fn)` |
| 8 | Global wildcard | ALLOW | `allow("*")` — permit everything | `policy.allow("*")` |

### Evaluation Algorithm

```python
def evaluate(tool_call, policies):
    # Walk buckets from highest priority (0) to lowest (8)
    for priority in range(9):
        bucket = policies_by_priority[priority]
        for policy in bucket:
            if policy.matches(tool_call):
                # Check predicate (when clause)
                if policy.predicate and not policy.predicate(tool_call.args):
                    continue
                return policy.decision  # First match wins
    return ALLOW  # Default: open by default
```

**Key rules**:
1. Buckets are walked high-to-low (0 to 8)
2. Within each bucket, first match wins (short-circuit)
3. If no policy matches at all, the default is **ALLOW** (open by default)
4. Exceptions during evaluation result in **DENY** (fail-closed)

### Policy Builder Functions

#### Basic Policies

```python
from google.antigravity.hooks import policy

# Deny a specific tool
p = policy.deny("run_command")

# Allow a specific tool
p = policy.allow("view_file")

# Ask user before executing
p = policy.ask_user("run_command", handler=my_approval_handler)

# Global policies
p = policy.deny("*")     # Block everything
p = policy.allow("*")    # Allow everything
p = policy.deny_all()    # Alias for deny("*")
p = policy.allow_all()   # Alias for allow("*")
```

#### Conditional Policies (Predicates)

```python
# Only deny when the command contains "rm"
p = policy.deny("run_command",
    when=lambda args: "rm" in args.get("CommandLine", ""))

# Async predicates are supported
async def check_workspace(args):
    path = args.get("FilePath", "")
    return path.startswith("/safe/workspace")

p = policy.allow("view_file", when=check_workspace)
```

Predicates receive the tool call arguments as a dict. If the predicate function has a Pydantic model type annotation, the args are automatically deserialized into the model.

#### Composite Policies

```python
# Default for LocalAgentConfig: deny run_command, allow everything else
p = policy.confirm_run_command()

# Read-only + ask for writes
p = policy.safe_defaults(handler=my_approval_handler)

# Restrict file tools to specific directories
p = policy.workspace_only(["/path/to/project", "/path/to/config"])
```

#### MCP-Aware Policies

```python
# Allow specific tools on an MCP server
p = policy.allow(mcp_server_config, ["tool1", "tool2"])

# Deny all tools on an MCP server
p = policy.deny(mcp_server_config)
```

### Enforcement

```python
from google.antigravity.hooks import policy

policies = [
    policy.deny("*"),                              # Base: deny everything
    policy.allow("view_file"),                      # Exception: allow reads
    policy.allow("list_directory"),
    policy.ask_user("run_command", handler=my_fn),  # Ask for commands
]

# Convert to a single PreToolCallDecideHook
hook = policy.enforce(policies, mcp_servers=[mcp_config])
agent.register_hook(hook)
```

### Default Policy: confirm_run_command

The `LocalAgentConfig` ships with `confirm_run_command()` as the default policy:

```python
def confirm_run_command():
    return [
        policy.deny("run_command"),
        policy.allow("*"),
    ]
```

This means:
- `run_command` is denied (level 0: specific deny)
- Everything else is allowed (level 8: global allow)

In interactive mode (`run_interactive_loop`), the deny is automatically upgraded to `ask_user` so the user can approve commands.

## Safety Invariant

The `Agent.__aenter__()` method enforces a critical safety check:

```python
async def __aenter__(self):
    ...
    has_write_tools = any(t in enabled_tools for t in WRITE_TOOLS)
    has_mcp = len(self.config.mcp_servers) > 0
    has_policies = len(self.config.policies) > 0
    has_decide_hook = any(isinstance(h, PreToolCallDecideHook) for h in hooks)

    if (has_write_tools or has_mcp) and not has_policies and not has_decide_hook:
        raise ValueError(
            "Write tools or MCP servers are enabled but no policies "
            "or PreToolCallDecideHook are registered. "
            "This would allow unrestricted tool execution."
        )
```

This prevents accidental deployment of an unguarded agent with destructive capabilities.

## Disabling vs. Denying Tools

| Mechanism | Model Sees Tool? | Token Cost | When to Use |
|-----------|-----------------|------------|-------------|
| `CapabilitiesConfig.disabled_tools` | No | Zero | Tools irrelevant to agent purpose |
| `policy.deny()` | Yes | Wasted on failed calls | Conditional or argument-dependent restrictions |
| `CapabilitiesConfig.enabled_tools` | Only listed tools | Minimal | Strict allowlist |

**Best practice**: Use `disabled_tools` to remove irrelevant tools from the model's view (saves tokens). Use `policy.deny()` for tools that should be visible but restricted (the model learns from the denial).

## Interaction with LocalConnection

### Built-in Tool Confirmation Flow

```mermaid
sequenceDiagram
    participant H as Harness
    participant LC as LocalConnection
    participant HR as HookRunner
    participant User as User

    H-->>LC: StepUpdate{tool_confirmation_request, action}
    LC->>LC: Extract tool name + args from action
    LC->>HR: dispatch_pre_tool_call(tool_call)
    
    alt Policy: allow
        HR-->>LC: HookResult(allow=True)
        LC->>H: ToolConfirmation(accepted=true)
    else Policy: deny
        HR-->>LC: HookResult(allow=False)
        LC->>H: ToolConfirmation(accepted=false)
    else Policy: ask_user
        HR-->>LC: HookResult(ask_user=True)
        LC->>User: "Allow {tool_name}?"
        User-->>LC: yes/no
        LC->>H: ToolConfirmation(accepted=user_answer)
    end

    Note over LC: Track in _pending_builtin_tool_calls

    H-->>LC: StepUpdate{state: DONE}
    LC->>HR: dispatch_post_tool_call(result)
    
    H-->>LC: StepUpdate{state: ERROR}
    LC->>HR: dispatch_on_tool_error(error)
```

### Host-Side Tool Flow

```mermaid
sequenceDiagram
    participant H as Harness
    participant LC as LocalConnection
    participant HR as HookRunner
    participant TR as ToolRunner

    H-->>LC: OutputEvent{tool_call: {id, name, args}}
    LC->>HR: dispatch_pre_tool_call(tool_call)
    HR-->>LC: HookResult(allow=True)
    LC->>TR: process_tool_calls([tool_call])
    
    alt Success
        TR-->>LC: [ToolResult(success)]
        LC->>HR: dispatch_post_tool_call(result)
        LC->>H: InputEvent{tool_response: {id, response_json}}
    else Error
        TR-->>LC: Exception
        LC->>HR: dispatch_on_tool_error(error)
        
        alt Recovery
            HR-->>LC: recovery_value
            LC->>H: InputEvent{tool_response: {id, response_json: recovery}}
        else No recovery
            HR-->>LC: None
            LC->>H: InputEvent{tool_response: {id, response_json: error}}
        end
    end
```

## Complete Example: Logging + Security Hooks

```python
import asyncio
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.hooks import hooks, policy
from google.antigravity import types

async def main():
    config = LocalAgentConfig(
        system_instructions="You are a helpful assistant.",
        capabilities=types.CapabilitiesConfig(),
        policies=[
            policy.deny("run_command"),
            policy.allow("*"),
        ],
    )

    agent = Agent(config)

    # Inspect: Log session lifecycle
    @agent.register_hook
    @hooks.on_session_start
    async def on_start():
        print("[SESSION] Started")

    @agent.register_hook
    @hooks.on_session_end
    async def on_end():
        print("[SESSION] Ended")

    # Inspect: Log tool calls
    @agent.register_hook
    @hooks.post_tool_call
    async def log_tool(result: types.ToolResult):
        print(f"[TOOL] Completed: {result}")

    # Decide: Block dangerous patterns
    @agent.register_hook
    @hooks.pre_tool_call_decide
    async def security_check(tool_call: types.ToolCall):
        if tool_call.name == "run_command":
            cmd = tool_call.args.get("CommandLine", "")
            if any(danger in cmd for danger in ["rm -rf", "chmod 777", "curl | bash"]):
                return hooks.HookResult(allow=False, message=f"Blocked dangerous command: {cmd}")
        return hooks.HookResult(allow=True)

    # Transform: Retry on timeout
    @agent.register_hook
    @hooks.on_tool_error
    async def retry_timeout(error: Exception):
        if "timeout" in str(error).lower():
            return "Operation timed out. Please try again with a smaller scope."
        return None

    async with agent:
        response = await agent.chat("What files are in the current directory?")
        print(await response.text())

asyncio.run(main())
```

## Cross-References

- [[sdk-test-antigravity-sdk]] — Full SDK reference including hook registration via Agent
- [[sdk-test-protobuf]] — ToolConfirmationRequest and related protobuf messages
- [[sdk-test-localharness]] — How hooks integrate with the localharness lifecycle
- [[sdk-test-auth-chain]] — Auth hooks for token refresh
- [[sdk-test-decompiled-patterns]] — Go-side hook implementations in the decompiled binary

---

*Source: `hooks.py` (289 lines), `hook_runner.py` (284 lines), `policy.py` (905 lines). Total: 1,478 lines of hook infrastructure.*