Content Detection and Content Routing
Learning Headroom
From Boundaries to Decisions
**Builds on ← 3.1 Headroom Pipeline and Integration Boundaries:** content crosses the boundary as raw blocks, but a boundary only decides WHEN Headroom looks at data — not WHAT to do with it.
**The problem:** a JSON tool output, a pytest log, a grep result, and a paragraph of prose all need *different* compression strategies. One-size-fits-all compression wastes tokens or breaks structure.
**The fix:** a detection step (`content_type_detection`) followed by a dispatch step (`router_dispatch`).
```mermaid
flowchart LR
A["Raw content block crosses boundary"] --> B{"ContentRouter detect_content_type"}
B -->|JSON array| C[SmartCrusher]
B -->|Search or grep| D[SearchCompressor]
B -->|Build or test log| E[LogCompressor]
B -->|Diff| F[DiffCompressor]
B -->|Plain text| G[TextCrusher]
B -->|Unrecognized| H["Kompress fallback"]
```
**Where this leads →** 4.1 SmartCrusher Eligibility Gates and 5.1 Specialized Compressors both assume content already arrived at the *correct* compressor — that correctness is decided here.
The Detector: Turning Content Into a Type
**Definition.** Content type detection is a function
\[
\text{detect\_content\_type}(content) \rightarrow \text{Detection}(content\_type,\ signal)
\]
where \(content\_type \in \text{ContentType}\) (e.g. `SEARCH_RESULTS`, `BUILD_OUTPUT`, `PLAIN_TEXT`, ...).
**Why a function, not a guess:** detection looks for *structural signals* in the text — fixed patterns that reliably identify a format — not keywords or vibes.
| Content pattern | Signal Headroom looks for |
|---|---|
| `file:line:content` repeated lines | grep/ripgrep output format |
| `pytest`, `npm`, `cargo` markers, PASS/FAIL banners | build tool output patterns |
| `---`/`+++` file headers and `@@` hunks | unified diff format |
| `[ { ... }, { ... } ]` bracket/brace structure | JSON array |
| none of the above | plain text (fallback) |
```python
from headroom.transforms import detect_content_type, ContentType
content = "src/main.py:42:def process():"
detection = detect_content_type(content)
# detection.content_type == ContentType.SEARCH_RESULTS
```
**Why it works:** each format has a near-unique syntactic fingerprint (colon-separated file:line, `@@` hunk markers, brace/bracket nesting) — cheap to check, hard to confuse with another type.
The Router: One Type, One Compressor
**Definition (router_dispatch).** Given a detected type \(t\), the ContentRouter maps it to exactly one compressor:
\[
\text{route}(t) = c, \quad c \in \{\text{SmartCrusher, SearchCompressor, LogCompressor, DiffCompressor, TextCrusher, Kompress}, \ldots\}
\]
Exactly one — never zero, never two. No block leaves un-routed.
| Detected content | Compressor | Typical savings |
|---|---|---|
| JSON arrays (tool outputs) | SmartCrusher | 70–90% |
| Search / grep results | SearchCompressor | 80–95% |
| Build / test logs | LogCompressor | 85–95% |
| Diffs | DiffCompressor | 40–80% |
| Source code | CodeAwareCompressor (opt-in) | 40–70% |
| Plain text | TextCrusher | 30–60% |
| Anything else | Kompress (ML fallback) | varies |
**Why route at all instead of one universal compressor?** Each compressor exploits structure specific to its format — SmartCrusher prunes/dedupes JSON fields, LogCompressor collapses repetitive log noise, SearchCompressor strips redundant path/context — a generic compressor can't safely do any of that without knowing the shape.
```mermaid
flowchart TD
T[content_type] --> R[router_dispatch]
R --> S1["Right compressor: max safe savings"]
```
Worked Example, Part 1 — JSON Array and Build Log
**Block A:**
```
[
{"id": 1, "name": "alice", "status": "active"},
{"id": 2, "name": "bob", "status": "active"},
{"id": 3, "name": "carol", "status": "inactive"}
]
```
Step 1 — scan for structural signal: outermost characters are `[` ... `]`, containing comma-separated `{ }` objects → matches JSON array pattern.
Step 2 — `content_type = JSON_ARRAY` (json_array_detection ✓)
Step 3 — `route(JSON_ARRAY) = SmartCrusher`
Step 4 — expected savings: 70–90% (repeated keys `id`, `name`, `status` across objects are exactly what SmartCrusher prunes/dedupes)
**Block B:**
```
collected 42 items
test_login.py::test_valid_user PASSED
test_login.py::test_bad_password FAILED
===== 1 failed, 41 passed in 3.21s =====
```
Step 1 — scan for structural signal: `PASSED`/`FAILED` markers, `collected N items`, pytest summary banner → matches build/test tool output pattern.
Step 2 — `content_type = BUILD_OUTPUT` (log_detection ✓)
Step 3 — `route(BUILD_OUTPUT) = LogCompressor`
Step 4 — expected savings: 85–95% (repetitive per-test lines and banner formatting are exactly what LogCompressor collapses)
Worked Example, Part 2 — Search Results and the Fallback Case
**Block C:**
```
src/auth/login.py:42:def check_password(user, pw):
src/auth/login.py:58: if not verify_hash(pw):
src/utils/hash.py:12:def verify_hash(pw):
```
Step 1 — signal: repeated `path:line_number:code_snippet` lines → ripgrep/grep output format.
Step 2 — `content_type = SEARCH_RESULTS` (search_result_detection ✓)
Step 3 — `route(SEARCH_RESULTS) = SearchCompressor`
Step 4 — savings 80–95% (repeated file paths + surrounding context are prunable)
**Block D (the fallback case):**
```
The quarterly review meeting has been rescheduled to next
Thursday at 2pm. Please update your calendars accordingly
and notify any external attendees of the change.
```
Step 1 — signal check: no `file:line:` pattern, no PASS/FAIL banner, no `---`/`@@` diff markers, no `[{...}]` brackets → none of the specific fingerprints match.
Step 2 — `content_type = PLAIN_TEXT` (text_fallback_detection ✓) — this is a *deliberate* fallback, not a failure.
Step 3 — `route(PLAIN_TEXT) = TextCrusher`
Step 4 — savings 30–60% — lower, because prose lacks the heavy structural redundancy the specialized compressors exploit.
**Edge case:** if TextCrusher's own signals also don't fire strongly, the ContentRouter's final catch-all is Kompress (ML-based) — every block is guaranteed *some* route, never zero.
Recap: Detect, Then Dispatch
**The two-step skill (this is the pass condition):**
1. `content_type_detection` — scan for structural signal, assign one `ContentType`
2. `router_dispatch` — map that type to exactly one compressor via the routing table
```mermaid
flowchart LR
A["Content block"] --> B["Detect signal"]
B --> C["Assign ContentType"]
C --> D["Route to ONE compressor"]
D --> E["Compressed output"]
```
**Efficiency note:** ContentRouter also keeps a two-tier TTL-bounded cache (skip set + result cache, default 30 min) so the *same* content isn't re-detected/re-compressed repeatedly.
**Where this leads:**
- → **4.1 SmartCrusher Eligibility Gates:** assumes a block already arrived correctly routed as JSON — gates decide if it's *safe* to compress, not what type it is.
- → **5.1 Specialized Compressors and Preservation Rules:** extends this same routing table with rules for what must be preserved per compressor.
Back to course