fix(compression): reject lossy unmarked tool output in unit router path (#1479)

## Description

Closes #1342

Codex shell output currently goes through the unit-router compression
path as a plain `local_shell_call_output` string. When that path picks a
lossy strategy and the compressed text carries no CCR retrieval marker,
the agent gets a summary that can't be reversed back to the original
shell log. That breaks the point of showing command output at all.

This change keeps structured shell output verbatim unless the
replacement stays recoverable. Other tool-output paths stay unchanged.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `headroom/transforms/compression_units.py`: add a lossy-strategy set
and a structured-shell heuristic, then reject lossy unmarked
replacements for `role="tool"` plus
`item_type="local_shell_call_output"` by returning the original text
with `reason="lossy_unrecoverable_tool_output"`.
- `tests/test_compression_units.py`: add regression coverage for the
failing case, the recoverable-marker case, non-shell tool output, and
assistant text so the guard stays scoped.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
GitHub Actions on head 6790f03486:
- PR Governance: green
- CI: green, including lint, build-wheel, docker-native-e2e, test shards 1-4, test-agno, test-dashboard-ui, and test-extras
- Init E2E: green
- Wrap E2E: green
- Evaluation Suite smoke-test: green
```

## Real Behavior Proof

- Environment: current PR head
`6790f03486` in GitHub Actions.
- Exact command / steps: exercise `compress_unit_with_router` with
structured multi-line `local_shell_call_output`, return a lossy unmarked
replacement, and assert the original shell text is kept with
`reason="lossy_unrecoverable_tool_output"`. Paired tests prove that
CCR-marked replacements still compress, non-shell tool output still
compresses, and assistant text still compresses when explicitly allowed.
- Observed result: the new regression coverage passes on the PR head and
the full PR check set is green.
- Not tested: end-to-end live shell sessions through the Responses API;
intentionally unstructured shell output below this heuristic remains
compressible.

## 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
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A, backend compression-path change.

## Additional Notes

`mypy headroom` was not run for this PR body refresh, so the type-check
box stays unchecked here. The changelog and docs boxes are N/A for this
targeted bug fix.
This commit is contained in:
Rod Boev 2026-06-30 17:30:12 -04:00 committed by GitHub
parent 312129a8e7
commit de24cd5fc0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 115 additions and 5 deletions

View file

@ -110,6 +110,17 @@ _CCR_MARKER_RE = re.compile(
r"(?m)^.*(?:Retrieve more: hash=|Retrieve original: hash=|<<ccr:[^>]+>>).*$"
)
_LOSSY_UNMARKED_STRATEGIES = {
CompressionStrategy.KOMPRESS.value,
CompressionStrategy.TEXT.value,
CompressionStrategy.CODE_AWARE.value,
}
def _is_structured_shell_output(text: str) -> bool:
nonempty_lines = [line for line in text.splitlines() if line.strip()]
return len(nonempty_lines) >= 3
def find_content_router(transforms: object) -> ContentRouter | None:
"""Return the first ContentRouter in a pipeline or iterable."""
@ -325,6 +336,19 @@ def compress_unit_with_router(
reason="rejected_not_smaller",
)
if (
unit.role == "tool"
and unit.item_type == "local_shell_call_output"
and _is_structured_shell_output(unit.text)
and strategy in _LOSSY_UNMARKED_STRATEGIES
):
if not _CCR_MARKER_RE.search(replacement):
return _with_reason(
strategy=strategy,
router_result=router_result,
reason="lossy_unrecoverable_tool_output",
)
return UnitCompressionResult(
original=unit.text,
compressed=replacement,

View file

@ -35,8 +35,9 @@ def test_compression_unit_accepts_token_shrinking_replacement():
text="alpha beta gamma delta epsilon",
provider="openai",
endpoint="responses",
role="tool",
item_type="local_shell_call_output",
role="assistant",
item_type="message",
metadata={"compress_assistant": "true"},
min_bytes=1,
),
router=Router("alpha beta"),
@ -46,7 +47,91 @@ def test_compression_unit_accepts_token_shrinking_replacement():
assert result.modified is True
assert result.tokens_saved == 3
assert result.compressed == "alpha beta"
assert "router:openai:responses:local_shell_call_output:kompress" in result.transforms_applied
assert "router:openai:responses:message:kompress" in result.transforms_applied
def test_compression_unit_keeps_lossy_unmarked_tool_output_verbatim():
original = (
"src/app.py:12 render shell status panel\n"
"src/ui.py:44 draw health badge\n"
"src/theme.py:9 set accent color"
)
result = compress_unit_with_router(
CompressionUnit(
text=original,
provider="openai",
endpoint="responses",
role="tool",
item_type="local_shell_call_output",
min_bytes=1,
),
router=Router("shell output looks organized and green"),
tokenizer=TokenCounter(),
)
assert result.modified is False
assert result.reason == "lossy_unrecoverable_tool_output"
assert result.original == original
assert result.compressed == original
def test_compression_unit_accepts_lossy_tool_output_when_recoverable():
original = "alpha beta gamma delta epsilon zeta eta theta"
result = compress_unit_with_router(
CompressionUnit(
text=original,
provider="openai",
endpoint="responses",
role="tool",
item_type="local_shell_call_output",
min_bytes=1,
),
router=Router("summary <<ccr:abc123>>"),
tokenizer=TokenCounter(),
)
assert result.modified is True
assert result.reason is None
assert result.compressed == "summary <<ccr:abc123>>"
def test_compression_unit_still_compresses_non_shell_tool_output():
result = compress_unit_with_router(
CompressionUnit(
text="alpha beta gamma delta epsilon zeta eta theta",
provider="openai",
endpoint="responses",
role="tool",
item_type="function_call_output",
min_bytes=1,
),
router=Router("summary for tool=0"),
tokenizer=TokenCounter(),
)
assert result.modified is True
assert result.reason is None
assert result.compressed == "summary for tool=0"
def test_compression_unit_still_compresses_assistant_text():
result = compress_unit_with_router(
CompressionUnit(
text="alpha beta gamma delta epsilon",
provider="openai",
endpoint="responses",
role="assistant",
item_type="message",
min_bytes=1,
metadata={"compress_assistant": "true"},
),
router=Router("alpha beta"),
tokenizer=TokenCounter(),
)
assert result.modified is True
assert result.reason is None
assert result.compressed == "alpha beta"
def test_compression_unit_rejects_non_shrinking_replacement():
@ -108,8 +193,9 @@ def test_batch_compression_preserves_provider_slot_references():
text="alpha beta gamma",
provider="openai",
endpoint="responses",
role="tool",
item_type="function_call_output",
role="assistant",
item_type="message",
metadata={"compress_assistant": "true"},
min_bytes=1,
),
slot=("input", 3, "output"),