How Headroom's ToolResultInterceptorTransform Handles Tool-Result Interception

The ToolResultInterceptorTransform iterates through every tool_result message in a request, applies registered interceptors that reduce token counts, and prevents duplicate processing through progressive disclosure keys.

The ToolResultInterceptorTransform serves as the first stage in Headroom's compression pipeline, processing LLM request messages before they reach the token limit enforcer. Located in the chopratejas/headroom repository, this transform inspects every tool_result payload, delegates rewriting to specialized interceptors, and ensures only token-reducing transformations are accepted.

Core Architecture

The transform is implemented in headroom/proxy/interceptors/base.py and consists of several coordinated components that manage the interception lifecycle.

The apply_to_messages Method

At the heart of the system lies apply_to_messages (line 92 in base.py), the central routine that orchestrates tool-result interception. This method iterates over all messages containing tool_result content, builds a lookup mapping tool_use IDs to their corresponding tool names and arguments, and executes the interceptor chain. It returns a tuple containing the potentially rewritten messages and a list of TransformSpan objects that record token savings.

Pipeline Entry Point

The ToolResultInterceptorTransform.apply method (line 43 in base.py) acts as the pipeline-level wrapper. This entry point measures token counts before and after interception using the supplied Tokenizer, invokes apply_to_messages, and injects the names of firing interceptors into the transforms_applied list. This integration allows the pipeline to track which transformations modified the request.

Interceptor Registry

Interceptors are maintained in a global list named INTERCEPTORS, auto-populated when the module loads. Built-in interceptors like AstGrepReadOutline register themselves in headroom/proxy/interceptors/__init__.py (line 12). Users can extend the system by calling the register() function with a custom interceptor class, which adds the interceptor to the global registry in an idempotent manner.

Progressive Disclosure Mechanism

To prevent redundant processing of identical tool results, the transform implements progressive disclosure. Each interceptor may expose a progressive_disclosure_key(name, args) method that returns a unique identifier for the operation. The transform stores these keys in a fired dictionary; subsequent tool_result messages with the same key are skipped, ensuring a file outlined once is not repeatedly re-outlined (lines 124-140 in base.py).

Safety and Token-Reduction Guards

The transform enforces a strict safety contract: interceptors must reduce token count to be accepted. Before applying any rewrite, the transform counts tokens in both the original and transformed content using the configured Tokenizer. If after >= before, the rewrite is discarded (lines 106-112). Exceptions thrown by interceptors are caught, logged, and counted via _record_failure, ensuring that buggy interceptors cannot crash the pipeline.

Execution Flow

The interception process follows a precise sequence that respects Headroom's "frozen-message" contract, which protects prefix cache entries from modification.

Building the Tool-Use Index

First, _build_tool_use_index (line 71 in base.py) scans all messages to create a dictionary mapping every tool_use ID to its tool name and arguments. This index enables interceptors to understand which tool produced a given result—for example, distinguishing a Read file operation from a Grep search.

Pre-Seeding Frozen Messages

The transform respects the frozen prefix boundary at messages[:frozen_count]. For tool results within this immutable prefix, the transform pre-computes and records any progressive disclosure keys in the fired map. This ensures that tools referencing already-processed files in the frozen context are not re-processed in the mutable tail.

Processing Mutable Tool Results

For each mutable tool_result message, the transform executes the following steps:

1. Content Extraction: Calls _extract_tool_result_content to retrieve the raw tool output payload.
2. Context Lookup: Uses the tool-use index to find the originating tool_use name and input arguments.
3. Interceptor Iteration: For every registered interceptor:
   • Computes the progressive disclosure key and skips if already in fired.
   • Invokes interceptor.matches(name, args, current_content) to check applicability.
   • Calls interceptor.transform(name, args, current_content) to obtain rewritten content.
   • Validates token reduction; discards the rewrite if tokens increased or stayed equal.
   • Records a TransformSpan with the interceptor name and token delta.
   • Updates the current content and adds the key to fired.
4. Content Swap: Uses _swap_tool_result_content to replace the original payload with the rewritten version.

The method returns an InterceptionResult containing the final message list and the collected spans.

Pipeline Integration

The transform integrates into Headroom's compression pipeline through headroom/transforms/pipeline.py. The pipeline instantiates ToolResultInterceptorTransform when HeadroomConfig.intercept_tool_results is True or when the environment variable HEADROOM_INTERCEPT_TOOL_RESULTS is set. As the first stage in the pipeline, it ensures that tool results are optimized before subsequent transforms process the message list.

Practical Implementation Examples

Using the Transform in a Pipeline

The following example demonstrates how to instantiate the transform and process messages containing tool results:

from headroom.tokenizer import Tokenizer
from headroom.transforms.pipeline import HeadroomPipeline
from headroom.proxy.interceptors import ToolResultInterceptorTransform

# Build a deterministic tokenizer for the example

class SimpleCounter:
    def count_text(self, txt: str) -> int:
        return max(1, len(txt) // 4)
    
    def count_messages(self, msgs):
        return sum(self.count_text(m.get("content", "")) for m in msgs)

tokenizer = Tokenizer(SimpleCounter())

# Create pipeline with the interceptor transform as the first stage

pipeline = HeadroomPipeline(
    transforms=[ToolResultInterceptorTransform()],
    tokenizer=tokenizer,
)

# Sample messages with a Read tool_result

messages = [
    {"role": "assistant", "content": [
        {"type": "tool_use", "id": "1", "name": "Read",
         "input": {"file_path": "/repo/payments.py"}}
    ]},
    {"role": "user", "content": [
        {"type": "tool_result", "tool_use_id": "1", 
         "content": "def foo():\n    return 42\n"}
    ]},
]

# Run the pipeline - built-in ast-grep interceptor outlines the file

result = pipeline.apply(messages)

print("Transforms applied:", result.transforms_applied)
print("Rewritten content:", result.messages[1]["content"][0]["content"])

Adding a Custom Interceptor

Developers can extend the system by implementing the ToolResultInterceptor interface and registering the class:

from headroom.proxy.interceptors import register, ToolResultInterceptor

class UpperCaseInterceptor:
    """Simple interceptor that redacts Read results containing 'secret'."""
    name = "uppercase"

    def matches(self, tool_name, tool_input, tool_output):
        return tool_name == "Read" and "secret" in tool_output.lower()

    def transform(self, tool_name, tool_input, tool_output):
        # Replace large payload with short marker to save tokens

        return "<<<REDACTED>>>"

# Register the interceptor (idempotent on name)

register(UpperCaseInterceptor())

# Now automatically invoked when pipeline runs

Summary

• ToolResultInterceptorTransform is the first stage in Headroom's compression pipeline, processing tool_result messages before other transforms.
• The transform delegates rewriting to registered interceptors while enforcing a token-reduction-only policy (lines 106-112 in base.py).
• Progressive disclosure prevents duplicate processing of identical tool results through a fired key tracking system.
• The tool-use index provides interceptors with context about which tool generated each result.
• Failed interceptors are caught and logged via _record_failure, ensuring pipeline stability.
• Built-in interceptors like AstGrepReadOutline auto-register in headroom/proxy/interceptors/__init__.py, while custom interceptors integrate via the register() function.

Frequently Asked Questions

What happens if an interceptor throws an exception?

The ToolResultInterceptorTransform wraps each interceptor call in a try-except block. If an interceptor raises an exception during matches() or transform(), the error is caught and logged via _record_failure, and the transform skips that interceptor for the current tool result. This design ensures that buggy interceptors cannot crash the entire compression pipeline.

How does the transform prevent processing the same file multiple times?

The transform implements progressive disclosure by requiring interceptors to define a progressive_disclosure_key(name, args) method. When an interceptor fires successfully, its key is stored in a fired dictionary. Subsequent tool results with the same key are skipped automatically, preventing scenarios where repeated reads of the same file trigger redundant outlining operations.

Why must interceptors reduce token count?

Headroom enforces a strict safety guard (lines 106-112 in headroom/proxy/interceptors/base.py) that compares token counts before and after transformation using the supplied Tokenizer. If the transformed content does not strictly reduce the token count (after >= before), the rewrite is discarded. This ensures the compression pipeline never increases the context window size through interception.

Where are built-in interceptors registered?

Built-in interceptors such as AstGrepReadOutline register automatically when the module loads in headroom/proxy/interceptors/__init__.py (line 12). This file populates the global INTERCEPTORS list, making custom tool-result handlers available immediately without manual configuration. Users can add custom interceptors by calling register() from the same module.

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