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

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

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

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

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

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

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

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

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

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

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.

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:

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

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

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

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

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

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

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)

# 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

# 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

# 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

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:

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:

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

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

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

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.