Headroom Canonical Pipeline Stages: The Complete 11-Step Request Lifecycle

The canonical pipeline stages in Headroom are a fixed, ordered sequence of eleven lifecycle moments defined by the PipelineStage enum in headroom/pipeline.py, stored in the CANONICAL_PIPELINE_STAGES tuple, and validated by unit tests to ensure stable extension hooks.

The chopratejas/headroom repository provides a structured framework for processing LLM requests through a predictable lifecycle. These canonical pipeline stages form the backbone of the system, allowing extensions and transforms to hook into specific execution points. Every request processed by Headroom traverses these stages in strict order, from initial setup through final response handling.

What Are the Canonical Pipeline Stages?

The canonical pipeline stages represent immutable lifecycle moments that every Headroom request passes through. Defined in headroom/pipeline.py (lines 19-44), these stages are implemented as an enum and collected in a tuple that serves as the contract for the entire pipeline system. Extensions, probes, and transforms rely on this ordering to emit or react to events at precise moments.

The 11 Canonical Pipeline Stages in Order

Headroom processes requests through the following eleven stages, defined in CANONICAL_PIPELINE_STAGES:

1. SETUP — Early initialization runs before any request-specific work begins (lines 19-20).

2. PRE_START — Hook point just before the pipeline officially starts (lines 21-22).

3. POST_START — Hook point immediately after the pipeline has started (lines 22-23).

4. INPUT_RECEIVED — The raw request (messages, tools, headers) has been received (lines 24-25).

5. INPUT_CACHED — Cache-lookup or cache-write logic for the input executes here (lines 26-27).

6. INPUT_ROUTED — The request is examined and routed to the appropriate content-aware compressor (lines 28-29).

7. INPUT_COMPRESSED — The selected compressor(s) run, reducing token count (lines 30-31).

8. INPUT_REMEMBERED — State-remembering logic (e.g., prefix alignment) applies after compression (lines 32-33).

9. PRE_SEND — Final hook before the compressed request is sent to the LLM (lines 34-35).

10. POST_SEND — Hook immediately after the request has been handed to the model (lines 36-37).

11. RESPONSE_RECEIVED — The model's response is received and available for post-processing (lines 38-39).

How the Stages Are Defined in Code

In headroom/pipeline.py, the stages are declared as enum members and then assembled into the canonical tuple:

from enum import Enum, auto

class PipelineStage(Enum):
    SETUP = auto()
    PRE_START = auto()
    POST_START = auto()
    INPUT_RECEIVED = auto()
    INPUT_CACHED = auto()
    INPUT_ROUTED = auto()
    INPUT_COMPRESSED = auto()
    INPUT_REMEMBERED = auto()
    PRE_SEND = auto()
    POST_SEND = auto()
    RESPONSE_RECEIVED = auto()

CANONICAL_PIPELINE_STAGES: tuple[PipelineStage, ...] = (
    PipelineStage.SETUP,
    PipelineStage.PRE_START,
    PipelineStage.POST_START,
    PipelineStage.INPUT_RECEIVED,
    PipelineStage.INPUT_CACHED,
    PipelineStage.INPUT_ROUTED,
    PipelineStage.INPUT_COMPRESSED,
    PipelineStage.INPUT_REMEMBERED,
    PipelineStage.PRE_SEND,
    PipelineStage.POST_SEND,
    PipelineStage.RESPONSE_RECEIVED,
)

This tuple (lines 32-44 in headroom/pipeline.py) provides the stable contract that tests/test_canonical_pipeline.py validates to ensure enum order matches the canonical sequence.

Working with Pipeline Stages in Practice

Iterating Over Canonical Stages

Access the ordered stages to build stage-aware tooling:

from headroom.pipeline import CANONICAL_PIPELINE_STAGES

for stage in CANONICAL_PIPELINE_STAGES:
    print(stage.value)

This prints the stage names in exact execution order.

Emitting Pipeline Events from Extensions

Hook into specific stages using the PipelineExtensionManager:

from headroom.pipeline import PipelineExtensionManager, PipelineStage

def my_extension(event):
    if event.stage is PipelineStage.INPUT_COMPRESSED:
        # Log metrics after compression

        print("Compression finished, tokens saved:", event.metadata.get("tokens_saved"))
    return event

manager = PipelineExtensionManager(extensions=[my_extension])
manager.emit(
    stage=PipelineStage.INPUT_COMPRESSED,
    operation="compress",
    messages=[{"role": "user", "content": "Hello"}],
    metadata={"tokens_saved": 42},
)

Validating Stage Membership

Verify that a stage belongs to the canonical set before processing:

from headroom.pipeline import PipelineStage, CANONICAL_PIPELINE_STAGES

assert PipelineStage.INPUT_ROUTED in CANONICAL_PIPELINE_STAGES

Key Source Files

The canonical pipeline implementation spans several critical files:

• headroom/pipeline.py — Defines PipelineStage, CANONICAL_PIPELINE_STAGES, and the extension manager that routes events between stages.
• headroom/transforms/pipeline.py — Implements the actual transform flow (cache aligner → content router) invoked between INPUT_ROUTED and INPUT_COMPRESSED.
• headroom/proxy/server.py and headroom/client.py — Demonstrate real-world usage of the stages when processing requests through the Headroom SDK or Proxy.
• tests/test_canonical_pipeline.py — Unit tests that validate the canonical ordering and assert the existence of all eleven stages.

Summary

• The canonical pipeline stages in Headroom comprise eleven ordered lifecycle moments from SETUP to RESPONSE_RECEIVED.
• Stages are defined in the PipelineStage enum and collected in the CANONICAL_PIPELINE_STAGES tuple in headroom/pipeline.py.
• The INPUT_COMPRESSED stage handles token reduction, while INPUT_ROUTED determines which compressor to apply.
• Unit tests in tests/test_canonical_pipeline.py guarantee the stable ordering required for extension compatibility.
• Extensions hook into these stages via PipelineExtensionManager.emit() to execute logic at precise lifecycle points.

Frequently Asked Questions

What is the difference between PRE_START and POST_START in Headroom?

PRE_START executes immediately before the pipeline officially begins processing, making it ideal for initialization checks. POST_START runs right after the pipeline starts, allowing extensions to verify that startup sequence completed successfully. Both are defined in headroom/pipeline.py (lines 21-23) and represent adjacent but distinct lifecycle boundaries.

How do I create a custom extension that hooks into specific pipeline stages?

Import PipelineStage and PipelineExtensionManager from headroom.pipeline, then implement a function that checks event.stage against your target stage (e.g., PipelineStage.INPUT_COMPRESSED). Return the event to preserve the pipeline flow, and pass your function to PipelineExtensionManager(extensions=[your_function]).

Where are the canonical pipeline stages tested?

The tests/test_canonical_pipeline.py file contains unit tests that validate the enum order matches the CANONICAL_PIPELINE_STAGES tuple. These tests ensure that extensions can rely on the documented ordering across Headroom versions.

Can I add custom stages to the canonical pipeline?

No. The canonical pipeline stages form an immutable contract designed to maintain compatibility across the Headroom ecosystem. While you can create custom extensions that react to existing stages, you cannot modify the CANONICAL_PIPELINE_STAGES tuple or add new enum values to PipelineStage without breaking the stable interface.

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