mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
57bf720d5c
## Description
`ContentRouter` only compressed JSON when the **whole** `tool_result`
block was a single JSON value. JSON embedded inside larger output (`gh
api` dumps, MCP tool results, `curl | jq` tails, log lines ending in a
JSON blob) was invisible to the JSON compressors — and in practice that
embedded shape is the large majority of JSON an agent actually sees.
This adds a structural routing step: find balanced JSON spans at **any
offset** in a block and route each one through the router's **existing,
unchanged** `_apply_strategy_to_content`, splicing the result back with
the surrounding bytes kept exact.
Because each span takes the same dispatch path a whole-block JSON
already takes, SmartCrusher/CodeCompressor register their `<<ccr:…>>`
retrieval markers exactly as before — CCR is hash-keyed, so it is
location-agnostic and unaffected by nesting.
Closes #
## Type of Change
- [x] New feature (non-breaking change that adds functionality)
## Changes Made
- New `headroom/transforms/recursive_json.py` — `route_embedded_json()`:
deterministic balanced-span scan + splice; skips spans already carrying
a `<<ccr:` marker (never re-compresses); token-gated.
- One guarded call at the top of `_apply_strategy_to_content` plus an
`_allow_embedded` **one-shot re-entrancy guard** (not a depth cap).
- **No size/min or depth thresholds** — the only gates are correctness
(round-trip) and benefit (token reduction). Strict no-op when a block
has no embedded JSON, so the 97%+ of non-JSON blocks are byte-identical
to today.
- Deterministic + per-block → prefix-cache- and CCR-store-stable.
- `tests/test_recursive_json.py`.
## Testing
- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`)
- [x] Type checking passes (`mypy`)
- [x] New tests added for new functionality
- [x] Manual testing performed
### Test Output
```text
$ python -m pytest tests/test_recursive_json.py tests/test_content_router_compact_json.py tests/test_content_router_tool_role_reversibility.py -q
19 passed in 5.90s
$ ruff check headroom/transforms/recursive_json.py headroom/transforms/content_router.py tests/test_recursive_json.py
All checks passed!
$ mypy headroom/transforms/recursive_json.py headroom/transforms/content_router.py
Success: no issues found in 2 source files
```
## Real Behavior Proof
- **Environment:** local,
`ContentRouter(ContentRouterConfig(lossless=False))`, pure-Python
content detector.
- **Exact steps:** `router.compress(block)` where `block` = prose with a
120-row JSON array embedded mid-text (`"I queried the ECS API
...\n[{...}]\nAll services healthy."`).
- **Observed result:** `strategy_used=MIXED`; block **11,986 → 3,798
chars**; leading/trailing prose preserved byte-exact; the embedded JSON
folded to a columnar table. Previously this block's embedded array was
not routed to the JSON compressor at all.
- **CCR:** a span already containing `<<ccr:` is passed through
untouched (unit-tested); folded spans go through the unchanged dispatch,
so markers register and resolve identically to whole-block JSON.
- **Not tested:** live proxy end-to-end with markers force-enabled
(covered by the unchanged dispatch path + unit tests); non-CC transcript
shapes beyond the local corpus.
## Review Readiness
- [x] I have performed a self-review
- [x] This PR is ready for human review
## Checklist
- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] My changes generate no new warnings
- [x] I have added tests that prove my feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md`
## Additional Notes
Scoped to the OSS structural-routing step only. The lossless *fold
kinds* it routes into (JSON columnar / log template) are maintained in
the `headroom-lossless-guard` extension. N/A: no docs/screenshots;
`Closes #` left blank (no tracking issue).
🤖 Generated with [Claude Code](https://claude.com/claude-code)
67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
"""Unit tests for headroom.transforms.recursive_json — the structural (embedded)
|
|
JSON routing step. Uses a fake dispatch so the mechanism is tested in isolation
|
|
from the real compressors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from headroom.transforms.recursive_json import route_embedded_json
|
|
|
|
|
|
def _upper_dispatch(span: str) -> str | None:
|
|
"""Fake compressor: returns a shorter deterministic stand-in for any span."""
|
|
try:
|
|
v = json.loads(span)
|
|
except ValueError:
|
|
return None
|
|
return f"<TABLE n={len(v)}>" if isinstance(v, list) else "<OBJ>"
|
|
|
|
|
|
def test_embedded_json_routed_and_surroundings_exact() -> None:
|
|
payload = json.dumps([{"id": i, "ok": True} for i in range(6)], separators=(",", ":"))
|
|
content = f"Fetched rows from API:\n{payload}\nDone (200 OK)."
|
|
out = route_embedded_json(content, _upper_dispatch)
|
|
assert out is not None
|
|
assert out.startswith("Fetched rows from API:\n")
|
|
assert out.endswith("\nDone (200 OK).")
|
|
assert "<TABLE n=6>" in out
|
|
|
|
|
|
def test_ccr_marker_span_passed_through() -> None:
|
|
# A span already carrying a CCR marker must never be re-routed (R1).
|
|
content = 'prefix [{"a":1,"b":2},{"a":3,"b":"<<ccr:deadbeef,json,900>>"}] suffix'
|
|
out = route_embedded_json(content, _upper_dispatch)
|
|
assert out is None # only span contains a marker → skipped → nothing to do
|
|
|
|
|
|
def test_no_json_is_noop() -> None:
|
|
assert route_embedded_json("just prose, nothing structured here", _upper_dispatch) is None
|
|
|
|
|
|
def test_whole_block_json_is_callers_job() -> None:
|
|
# A block that IS a single JSON value is skipped (routed by the caller).
|
|
content = json.dumps([{"a": i} for i in range(5)], separators=(",", ":"))
|
|
assert route_embedded_json(content, _upper_dispatch) is None
|
|
|
|
|
|
def test_benefit_gate_declines_when_not_smaller() -> None:
|
|
payload = json.dumps([{"a": i} for i in range(5)], separators=(",", ":"))
|
|
content = f"x {payload} y"
|
|
# Dispatch that returns something LARGER → must be declined (outcome gate).
|
|
assert route_embedded_json(content, lambda s: s + " " * 999) is None
|
|
|
|
|
|
def test_deterministic() -> None:
|
|
payload = json.dumps([{"k": i} for i in range(8)], separators=(",", ":"))
|
|
content = f"a {payload} b {payload} c"
|
|
r1 = route_embedded_json(content, _upper_dispatch)
|
|
r2 = route_embedded_json(content, _upper_dispatch)
|
|
assert r1 == r2 and r1 is not None
|
|
assert r1.count("<TABLE n=8>") == 2 # both embedded spans routed
|
|
|
|
|
|
def test_scalar_array_not_routed() -> None:
|
|
# array of scalars is not a "routable" JSON shape (no dict rows)
|
|
content = "nums: [1,2,3,4,5,6,7,8] done"
|
|
assert route_embedded_json(content, _upper_dispatch) is None
|