How Headroom Handles Tool Result Interception: Architecture and Implementation

Headroom intercepts raw tool outputs before they reach the LLM by executing a registry of ToolResultInterceptor objects through a ToolResultInterceptorTransform that supports progressive disclosure and automatic failure recovery.

Headroom, an open-source LLM proxy from the chopratejas/headroom repository, implements tool result interception as a first-class compression stage. This mechanism allows you to rewrite, shorten, or annotate the text returned by tools (such as file reads) before the model consumes it, dramatically reducing token usage while preserving the ability to request full content later.

The Tool Result Interception Architecture

The interception system is built around three core concepts: a global interceptor registry, a transform adapter that runs the registry against message lists, and progressive disclosure to prevent redundant rewrites of the same file.

The Interceptor Protocol

All interceptors implement the ToolResultInterceptor protocol defined in headroom/proxy/interceptors/base.py. This interface requires three methods:

class ToolResultInterceptor(Protocol):
    """Rewrite a single tool_result's text."""
    name: str                                    # e.g. "ast-grep"

    def matches(self, tool_name, tool_input, tool_output) -> bool: ...
    def transform(self, tool_name, tool_input, tool_output) -> str | None: ...
    def progressive_disclosure_key(self, tool_name, tool_input) -> str | None: ...

• matches determines whether the interceptor should process this specific tool result.
• transform returns the rewritten text or None to pass through unchanged.
• progressive_disclosure_key (optional) returns a stable identifier—usually the file path—to ensure the same source is only outlined once per conversation.

Interceptors must be idempotent and must never raise exceptions. Errors are caught in the apply loop, logged at warning level, and recorded via interceptor_failure_counts().

The Registry and Apply Loop

The global registry is a simple list populated via register(interceptor). The heavy lifting occurs in apply_to_messages(messages, tokenizer, frozen_count) in headroom/proxy/interceptors/base.py:

1. Builds a tool-use index (_build_tool_use_index) linking each tool_result to its originating tool_use.
2. Pre-seeds the fired set from the frozen prefix (cached messages) to maintain progressive disclosure across the prefix/tail boundary.
3. Iterates mutable tool results:
   • Extracts original content via _extract_tool_result_content.
   • Looks up the associated tool_name and tool_input.
   • Walks the registry, skipping interceptors whose disclosure key already exists in the fired set.
   • Calls matches; if true, calls transform.
   • If the transformed text is shorter (token-wise), records a TransformSpan and updates fired with the disclosure key.
   • Swaps content back via _swap_tool_result_content.

The function returns an InterceptionResult containing the rewritten messages and a list of spans indicating token savings.

How Interceptors Work in the Pipeline

ToolResultInterceptorTransform

In headroom/proxy/interceptors/__init__.py, the ToolResultInterceptorTransform class wraps the apply loop as a formal compression stage:

class ToolResultInterceptorTransform(Transform):
    """Runs interceptors as the first compression stage."""
    name = "tool_result_interceptors"

    def apply(self, messages, tokenizer, **kwargs):
        tokens_before = tokenizer.count_messages(messages)
        frozen = int(kwargs.get("frozen_message_count") or 0)
        result = apply_to_messages(messages, tokenizer, frozen_count=frozen)
        tokens_after = tokenizer.count_messages(result.messages)
        transforms_applied = [f"interceptor:{s.tool}" for s in result.spans] if result.spans else []
        return TransformResult(
            messages=result.messages,
            tokens_before=tokens_before,
            tokens_after=tokens_after,
            transforms_applied=transforms_applied,
        )

This transform is automatically inserted at the head of the compression pipeline in headroom/transforms/pipeline.py when HeadroomConfig.intercept_tool_results (or the environment variable HEADROOM_INTERCEPT_TOOL_RESULTS) is truthy.

Progressive Disclosure and Deduplication

The progressive disclosure mechanism prevents the model from receiving redundant outlines of the same file. When an interceptor returns a key from progressive_disclosure_key(), the system tracks it in a fired set. Subsequent reads of the same file—identified by the same key—skip the interceptor and return the full content, allowing the model to request details only when needed.

Built-in Interceptor: AST-Grep Outline

The default interceptor, AstGrepReadOutline in headroom/proxy/interceptors/astgrep.py, shortens large code file reads by:

• Detecting Read-type calls where output exceeds HEADROOM_INTERCEPT_READ_MIN_CHARS (default 500).
• Ensuring no explicit line-range keys are present (the model requested the full file).
• Determining the source language from the file extension.
• Running the external ast-grep binary to collect top-level function and class signatures.
• Building a compact outline containing signatures, optional docstrings, and an elision marker (OUTLINE_MARKER).

The interceptor returns None if the binary is unavailable, the language is unsupported, or fewer than three definitions are found. It self-registers via base.register(AstGrepReadOutline()) at module import time.

Implementing Custom Tool Result Interceptors

You can add custom interceptors by implementing the protocol and registering them at runtime:

from headroom.proxy.interceptors.base import ToolResultInterceptor, register

class UppercaseInterceptor:
    name = "uppercase"

    def matches(self, tool_name, tool_input, tool_output):
        # Intercept any tool returning a string longer than 10 chars.

        return isinstance(tool_output, str) and len(tool_output) > 10

    def transform(self, tool_name, tool_input, tool_output):
        # Return the output in uppercase (demo only; may increase tokens).

        return tool_output.upper()

    # No progressive_disclosure_key needed (always runs).

# Register at import time or runtime.

register(UppercaseInterceptor())

Once registered, the interceptor automatically participates in the apply_to_messages loop for all subsequent requests.

Configuration and Failure Handling

Headroom provides multiple ways to enable tool result interception:

• Globally: Set HEADROOM_INTERCEPT_TOOL_RESULTS=1 or configure HeadroomConfig(intercept_tool_results=True).
• Per-run: Pass intercept_tool_results=True when constructing a Headroom instance.

Failure Handling and Observability

Every exception in matches, transform, or progressive_disclosure_key is caught and logged without breaking the pipeline. Failures are tracked per-interceptor:

from headroom.proxy.interceptors.base import (
    interceptor_failure_counts,
    reset_interceptor_failure_counts
)

reset_interceptor_failure_counts()

# ... run requests ...

print(interceptor_failure_counts())  # e.g., {"ast-grep": 0, "uppercase": 1}

The pipeline also exposes which interceptors actually rewrote content via the transforms_applied field in TransformResult, enabling dashboards to display per-interceptor token savings.

Summary

• Tool result interception in Headroom rewrites tool outputs before they reach the LLM, reducing token usage.
• The system uses a registry pattern with stateless ToolResultInterceptor objects defined in headroom/proxy/interceptors/base.py.
• Progressive disclosure keys prevent duplicate outlines of the same file within a conversation.
• The ToolResultInterceptorTransform in headroom/proxy/interceptors/__init__.py integrates interception as the first stage of the compression pipeline.
• The built-in ast-grep interceptor outlines large code files automatically when HEADROOM_INTERCEPT_TOOL_RESULTS is enabled.
• Custom interceptors implement the ToolResultInterceptor protocol and register via register(), with automatic failure isolation and counting.

Frequently Asked Questions

What is the purpose of tool result interception in Headroom?

Tool result interception allows Headroom to shorten or transform verbose tool outputs—such as large file reads—before they are passed to the LLM. This reduces token consumption and context window pressure while maintaining the semantic essence of the content, often by providing an outline that the model can expand upon request.

How does progressive disclosure prevent duplicate outlines?

Progressive disclosure uses a fired set to track which files have already been intercepted based on the progressive_disclosure_key (typically the file path). When an interceptor has already processed a specific file, subsequent tool results with the same key skip the interceptor and return the full content, ensuring the model only sees the outline once per conversation.

Can I disable the built-in ast-grep interceptor while keeping custom ones?

The built-in interceptor is registered globally at module import time in headroom/proxy/interceptors/astgrep.py. To disable it specifically while keeping the interception mechanism active, you would need to modify the registry by removing the AstGrepReadOutline instance from the internal interceptor list, or simply set HEADROOM_INTERCEPT_READ_MIN_CHARS to a very high value to prevent it from matching.

What happens if an interceptor throws an exception?

All interceptor methods are wrapped in exception handlers within apply_to_messages. If an exception occurs in matches, transform, or progressive_disclosure_key, the error is logged at warning level and the interceptor is skipped for that message. The failure is recorded in interceptor_failure_counts() for observability, and the pipeline continues processing remaining interceptors without中断.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}