akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(proxy): protect file reads from lossy compression on the Responses API path (Copilot view + HEADROOM_PROTECT_READS) (#3238)

Browse source 
## Description

On the OpenAI Responses API path (used by `headroom wrap copilot` and
Codex), fresh file reads were lossy-compressed one turn after
production, so the model saw its own just-read file content garbled
(Kompress word-dropping) and had to re-read it — the exact turn
inflation `HEADROOM_PROTECT_READS` was built to prevent on the
chat/Anthropic path.

Two gaps combined:

1. Copilot CLI's `view` tool (its file-read tool) was not in
`DEFAULT_EXCLUDE_TOOLS` — the set only covered Claude-Code names
(`Read`, `Write`, …).
2. `_compress_openai_responses_live_text_units_with_router`
(`headroom/proxy/handlers/openai.py`) protected only excluded tool
*names* and never implemented the `HEADROOM_PROTECT_READS` read-command
detection that `ContentRouter.apply()` has — so `bash` reads like `nl
-ba FILE | sed -n '1,75p'` were lossy-compressed even with the `coding`
profile's `protect_reads=True`.

Fix (design adversarially reviewed with gpt-5.6-sol before
implementation; verdict "correct with modifications" — all modifications
adopted):

- `view` added to **both** `DEFAULT_EXCLUDE_TOOLS` and
`DEFAULT_VERBATIM_EXCLUDE_TOOLS` — byte-exact contract: no lossy
compression, no lossless JSON rewrite, no cross-turn dedup fold.
- Responses units path now ports the read-command guard: the producing
command is normalized from both wire shapes (`function_call.arguments`,
`local_shell_call.action` argv/string) via the shared
`_tool_call_command_text`; each output is content-gated by
`_read_output_should_be_protected` (lockfiles/JSON/logs/search stay
compressible); protected ids are unioned into the dedup protection set.
- Shared `read_protection_enabled()` env helper extracted in
`content_router.py`, used by both paths.
- Latent debug-path defect fixed (unbound `fold` when an excluded tool's
output is a content-part list and debug logging is enabled).

Follow-up (not in scope): Rust Responses path
(`crates/headroom-core/src/transforms/live_zone.rs`) currently only
protects `headroom_retrieve` — needs parity before that runtime becomes
default.

Closes #3237

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)

## Changes Made

- `headroom/config.py` — `view` in both exclusion sets.
- `headroom/proxy/handlers/openai.py` — read-command protection for the
Responses units path; dedup shield; debug-path fix.
- `headroom/transforms/content_router.py` — shared
`read_protection_enabled()` helper (both paths).
- `tests/test_openai_responses_read_protection.py` — 16 regression
tests.

## Testing

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

### Test Output

```text
# Before fix (first commit on this branch, repro-only):
2 failed, 1 passed   # view read compressed; bash nl|sed read compressed despite HEADROOM_PROTECT_READS=1

# After fix:
$ uv run pytest tests/test_openai_responses_read_protection.py -q
16 passed (incl. content-gate release, string-form local_shell_call, debug paths, scan robustness)

$ uv run pytest tests/test_openai_responses_compression_units.py tests/test_responses_cross_turn_dedup.py \
    tests/test_lossless_excluded_compaction.py tests/test_observed_wire_shapes.py \
    tests/test_content_router_exclude_tools.py tests/test_content_router_compact_json.py -q
69 passed in 3.85s

$ uv run ruff check <changed files> && uv run ruff format --check <changed files>
All checks passed!  4 files already formatted

$ uv run mypy headroom/config.py headroom/transforms/content_router.py headroom/proxy/handlers/openai.py
Success: no issues found
```

## Real Behavior Proof

- Environment: macOS (Darwin), Python 3.13.7, headroom proxy 0.37.0-dev,
`HEADROOM_STACK=wrap_copilot`, savings profile `coding` (effective per
proxy banner), model `gpt-5.6-luna` via GitHub Copilot API.
- Exact command / steps: incident forensics on Copilot CLI session
`5487d36f-3e7d-48b0-a56a-a92e4969c17b` (events.jsonl tool results
byte-matched to proxy log compression units), then the failing→passing
repro above.
- Observed result: (pre-fix proxy log `~/.headroom/logs/proxy-8794.log`)
  ```text
08:46:34 [hr_1787553986_000014] WS /v1/responses slow compression unit …
strategy=text … bytes=3857 … tokens_saved=235
08:46:34 [hr_1787553986_000014] … strategy=text … bytes=6852 …
tokens_saved=413
08:46:34 [hr_1787553986_000014] … strategy=text … bytes=3079 …
tokens_saved=178
08:46:47 [hr_1787554001_000015] … strategy=text … bytes=6924 …
tokens_saved=478
08:46:47 [hr_1787554001_000015] … strategy=text … bytes=4418 …
tokens_saved=305
  ```
Byte sizes match the session's `view` (3857/6852/3079) and `nl|sed`
(6924/4418) tool results exactly. Post-fix, those payload shapes are
byte-exact through
`_compress_openai_responses_live_text_units_with_router` (asserted by
the regression tests over the same wire shapes).
- Not tested: full `pytest tests/` run (upstream suite has pre-existing
order-dependent failures — 7 failed on clean `main` under `-k
"content_router or read or protect"` — and a pre-existing `litellm`
import error in `tests/test_memory_eval.py`; the 5 additional failures
in that selection with my branch pass in isolation and also fail on
clean main under the same selection); live end-to-end with a running
Copilot wrap (unit-level wire-shape coverage instead); Rust core path
(follow-up).

## Runtime Rollout Safety

- Rollout-managed feature(s): none
- Minimum rollout channel: N/A
- Stable/default behavior changed: yes — `view` outputs and
`HEADROOM_PROTECT_READS`-covered bash read outputs stay verbatim on the
Responses path (fidelity improvement; slightly fewer tokens saved)
- Kill switch / disable path: `HEADROOM_PROTECT_READS=0` restores old
bash-read behavior; `HEADROOM_EXCLUDE_TOOLS` overrides tool exclusion
- Unsafe override required: no
- Qualification impact: none
- Rollback path: revert the commit; no state/migration

## 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] I have made corresponding changes to the documentation (N/A, no
user-facing docs for this internal guard)
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md`
This commit is contained in:
Raúl 2026-08-24 13:03:03 +02:00 • committed by GitHub
parent b9d7dcc3da
commit 4408e88106
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 616 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

7
headroom/config.py
View file

@ -228,6 +228,9 @@ DEFAULT_EXCLUDE_TOOLS: frozenset[str] = frozenset(
"WebSearch",
"WebFetch",
"headroom_retrieve",
# Copilot CLI's file-read tool (its `Read` equivalent): raw file bytes
# the model byte-patches against.
"view",
# Lowercase variants for case-insensitive matching
"read",
"glob",
@ -253,6 +256,10 @@ DEFAULT_VERBATIM_EXCLUDE_TOOLS: frozenset[str] = frozenset(
"web_search",
"web_fetch",
"headroom_retrieve",
# `view` (Copilot CLI file read) must stay BYTE-EXACT: the model produces
# line/byte-precise edits against it, and even "lossless" JSON rewrites
# or cross-turn dedup folds break old_str matching and force re-reads.
"view",
}
)

64
headroom/proxy/handlers/openai.py
View file

@ -2116,6 +2116,45 @@ class OpenAIHandlerMixin:
if is_tool_excluded(fn_name, DEFAULT_VERBATIM_EXCLUDE_TOOLS)
}
# Read protection (HEADROOM_PROTECT_READS) — parity with the
# chat/Anthropic path (ContentRouter.apply). Output of a file-READ
# command (cat/nl/sed -n/head/tail/…) must stay verbatim: the agent
# byte-patches against it, and lossy reads caused re-reads /
# turn-inflation + resolve loss on SWE-bench. The Responses wire carries
# the producing command in two shapes, both normalized by the shared
# _tool_call_command_text helper:
# - function_call.arguments (Copilot bash, Codex exec_command, …)
# - local_shell_call.action (native Responses shell; argv or string)
# Content is gated per-output by _read_output_should_be_protected so
# confidently non-code DATA reads (lockfiles, JSON, logs, search) stay
# compressible, exactly like the chat path.
from headroom.transforms.content_router import (
_is_read_command,
_read_output_should_be_protected,
_tool_call_command_text,
read_protection_enabled,
)
read_command_by_call_id: dict[str, str] = {}
if read_protection_enabled():
for item in items:
if not isinstance(item, dict):
continue
item_type = item.get("type")
if item_type == "function_call":
command = _tool_call_command_text(item.get("arguments"))
elif item_type == "local_shell_call":
command = _tool_call_command_text(item.get("action"))
else:
continue
call_id = item.get("call_id")
if command and isinstance(call_id, str) and call_id and _is_read_command(command):
read_command_by_call_id[call_id] = command
# Outputs protected by read-command detection. Also unioned into the
# cross-turn dedup protection set below: a [↑…] fold of a read would
# break the exact-bytes contract just like lossy compression would.
read_protected_call_ids: set[str] = set()
timing_sink: dict[str, float] = timing if timing is not None else {}
def _add_timing(name: str, started_at: float) -> None:
@ -2159,6 +2198,24 @@ class OpenAIHandlerMixin:
}
)
continue
if isinstance(call_id, str) and call_id in read_command_by_call_id:
# Finalize by CONTENT (same gate as ContentRouter.apply):
# protect unless the output is confidently non-code DATA.
if _read_output_should_be_protected(_responses_part_text(item.get("output"))):
read_protected_call_ids.add(call_id)
if debug_enabled:
extraction_debug.append(
{
"index": idx,
"eligible": False,
"reason": "read_command_protected",
"item_type": item_type,
"call_id": call_id,
"command": read_command_by_call_id[call_id],
"item": item,
}
)
continue
if isinstance(call_id, str) and call_id in excluded_call_ids:
if call_id in verbatim_excluded_call_ids:
if debug_enabled:
@ -2180,6 +2237,7 @@ class OpenAIHandlerMixin:
# Note: when output is a content-part array, fold each text part
# individually using ("output_part", index) slots to preserve the
# array structure (non-text parts like images are left untouched).
excluded_folded = False
raw_output = item.get("output")
if isinstance(raw_output, list):
for pidx, part in enumerate(raw_output):
@ -2191,6 +2249,7 @@ class OpenAIHandlerMixin:
part_text = part["text"]
pf = router._lossless_compact_excluded(part_text)
if pf is not None:
excluded_folded = True
lossless_excluded.append(
(idx, ("output_part", pidx), pf[0], part_text)
)
@ -2198,6 +2257,7 @@ class OpenAIHandlerMixin:
excl_out = _responses_part_text(raw_output)
fold = router._lossless_compact_excluded(excl_out) if excl_out else None
if fold is not None:
excluded_folded = True
lossless_excluded.append((idx, ("output", None), fold[0], excl_out))
if debug_enabled:
extraction_debug.append(
@ -2206,7 +2266,7 @@ class OpenAIHandlerMixin:
"eligible": False,
"reason": (
"exclude_tools_lossless_fold"
if fold is not None
if excluded_folded
else "exclude_tools_protected"
),
"item_type": item_type,
@ -2622,7 +2682,7 @@ class OpenAIHandlerMixin:
updated_items,
self.OPENAI_RESPONSES_OUTPUT_TYPES,
tokenizer.count_text,
protected_call_ids=verbatim_excluded_call_ids,
protected_call_ids=verbatim_excluded_call_ids | read_protected_call_ids,
)
if dd_folded:
modified = True

28
headroom/transforms/content_router.py
View file

@ -532,6 +532,20 @@ def _tool_call_args_text(raw: Any) -> str:
return " ".join(text.split())[:300]
def read_protection_enabled() -> bool:
"""True when HEADROOM_PROTECT_READS opts into byte-exact file-read protection.
Shared by every request path (chat/Anthropic ``ContentRouter.apply`` and the
OpenAI Responses units path) so the flag means the same thing everywhere.
"""
return os.environ.get("HEADROOM_PROTECT_READS", "0").strip().lower() not in (
"0",
"",
"false",
"no",
)
def _tool_call_command_text(raw: Any) -> str:
"""Extract the raw shell command from a tool call's args, if present.
@ -4827,12 +4841,7 @@ class ContentRouter(Transform):
# Type-specific by design: grep/test/ls output stays compressible, so the
# cache-mode delta still compresses whenever the newest turn is NOT a read.
self._protect_read_tool_ids = set()
if os.environ.get("HEADROOM_PROTECT_READS", "0").strip().lower() not in (
"0",
"",
"false",
"no",
):
if read_protection_enabled():
# Use _tool_call_commands (the parsed shell command), NOT
# _tool_call_args (a compact free-text blob that, for OpenAI-style
# JSON-string args, is the raw ``{"command": ...}`` JSON — on which
@ -4854,12 +4863,7 @@ class ContentRouter(Transform):
# cat/sed/head code reads are protected on ANY model/harness, not just
# those that emit tool-call/tool_result blocks.
self._protect_read_msg_indices: set[int] = set()
if os.environ.get("HEADROOM_PROTECT_READS", "0").strip().lower() not in (
"0",
"",
"false",
"no",
):
if read_protection_enabled():
for _idx, _m in enumerate(messages):
if _m.get("role") != "user":
continue

531
tests/test_openai_responses_read_protection.py Normal file
View file

@ -0,0 +1,531 @@
"""Regression tests: file reads over the OpenAI Responses API path must stay verbatim.
Copilot CLI (and other Responses-native harnesses) read files two ways:
1. A first-class ``view`` tool (the Copilot equivalent of Claude Code's ``Read``)
whose output is raw file content the model will byte-patch against.
2. Shell reads through ``bash`` (``cat``/``nl``/``sed -n`` …), which the
chat/Anthropic path protects via ``HEADROOM_PROTECT_READS`` read-command
detection in ``ContentRouter``.
The Responses compression-units path historically protected neither: only
``DEFAULT_EXCLUDE_TOOLS`` names were honored, and ``HEADROOM_PROTECT_READS``
was never consulted. Lossy (Kompress) compression of a fresh file read garbles
exactly the bytes the model needs for line-precise edits, forcing re-reads
(turn inflation) — the harm read protection exists to prevent.
"""
from __future__ import annotations
from types import MethodType, SimpleNamespace
from headroom.proxy.handlers.openai import OpenAIHandlerMixin
from headroom.transforms.content_router import (
CompressionStrategy,
ContentRouter,
RouterCompressionResult,
)
class TokenCounter:
def count_text(self, text: str) -> int:
return len(text.split())
def _handler_with_router(router: ContentRouter) -> OpenAIHandlerMixin:
handler = OpenAIHandlerMixin()
handler.openai_pipeline = SimpleNamespace(transforms=[router])
handler.openai_provider = SimpleNamespace(
get_token_counter=lambda _model: TokenCounter(),
)
return handler
def _lossy_router() -> ContentRouter:
"""Router whose compress() always 'lossy-compresses' any candidate it sees."""
router = ContentRouter()
def compress(self, content: str, **_kwargs):
return RouterCompressionResult(
compressed="kept words",
original=content,
strategy_used=CompressionStrategy.KOMPRESS,
)
router.compress = MethodType(compress, router)
return router
def _run(handler: OpenAIHandlerMixin, payload: dict):
return handler._compress_openai_responses_live_text_units_with_router(
payload,
model="gpt-5",
request_id="req_read_protection",
)
_FILE_CONTENT = "\n".join(
f"## Section {i}\nSome roadmap prose line {i} with enough words to matter" for i in range(90)
)
_NL_OUTPUT = "\n".join(
f"{i}\tline {i} of the roadmap file with a handful of words in it" for i in range(1, 110)
)
def test_responses_view_tool_read_stays_verbatim():
"""Copilot's `view` tool returns raw file bytes: never lossy-compress them."""
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_view",
"name": "view",
"arguments": '{"path": "/repo/ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_view",
"output": _FILE_CONTENT,
},
],
}
new_payload, _modified, _saved, _t, _u, _s, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _FILE_CONTENT
def test_responses_bash_read_command_stays_verbatim_when_protect_reads(monkeypatch):
"""HEADROOM_PROTECT_READS=1 must cover bash file reads on the Responses path too."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_bash",
"name": "bash",
"arguments": ('{"command": "nl -ba .overlay/ROADMAP.md | sed -n \'1,75p\'"}'),
},
{
"type": "function_call_output",
"call_id": "call_bash",
"output": _NL_OUTPUT,
},
],
}
new_payload, _modified, _saved, _t, _u, _s, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _NL_OUTPUT
def test_responses_excluded_read_tool_stays_verbatim_control():
"""Control: Claude-style `Read` outputs are already protected today."""
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_read",
"name": "Read",
"arguments": '{"file_path": "/repo/ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_read",
"output": _FILE_CONTENT,
},
],
}
new_payload, _modified, _saved, _t, _u, _s, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _FILE_CONTENT
def test_responses_bash_read_compresses_when_protect_reads_disabled(monkeypatch):
"""Control: with HEADROOM_PROTECT_READS unset/0, bash reads stay compressible."""
monkeypatch.delenv("HEADROOM_PROTECT_READS", raising=False)
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_bash",
"name": "bash",
"arguments": '{"command": "cat src/main.py"}',
},
{
"type": "function_call_output",
"call_id": "call_bash",
"output": _NL_OUTPUT,
},
],
}
new_payload, modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert modified is True
assert new_payload["input"][1]["output"] == "kept words"
def test_responses_non_read_bash_command_still_compresses(monkeypatch):
"""Protection is type-specific: test/build/search output stays compressible."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_test",
"name": "bash",
"arguments": '{"command": "uv run pytest tests/ -q"}',
},
{
"type": "function_call_output",
"call_id": "call_test",
"output": _NL_OUTPUT,
},
],
}
new_payload, modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert modified is True
assert new_payload["input"][1]["output"] == "kept words"
def test_responses_lockfile_read_stays_compressible(monkeypatch):
"""Lockfiles are tool-regenerated, never byte-patched: the command-level
carve-out keeps `cat uv.lock` compressible even with protection on."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_lock",
"name": "bash",
"arguments": '{"command": "cat uv.lock"}',
},
{
"type": "function_call_output",
"call_id": "call_lock",
"output": _NL_OUTPUT,
},
],
}
new_payload, modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert modified is True
assert new_payload["input"][1]["output"] == "kept words"
def test_responses_local_shell_call_read_stays_verbatim(monkeypatch):
"""Codex native shell: local_shell_call.action.command (argv) read protected."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "local_shell_call",
"call_id": "call_lsc",
"action": {"type": "exec", "command": ["nl", "-ba", "ROADMAP.md"]},
},
{
"type": "local_shell_call_output",
"call_id": "call_lsc",
"output": _NL_OUTPUT,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _NL_OUTPUT
def test_responses_view_output_content_part_array_stays_verbatim():
"""`view` output shaped as a content-part array is protected byte-exactly,
including non-text parts."""
handler = _handler_with_router(_lossy_router())
parts = [
{"type": "output_text", "text": _FILE_CONTENT},
{"type": "refusal", "refusal": "n/a"},
]
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_view",
"name": "view",
"arguments": '{"path": "/repo/ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_view",
"output": parts,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == parts
def test_responses_view_json_shaped_output_stays_byte_exact():
"""Even JSON-shaped `view` output is verbatim: the byte-exact contract beats
the lossless JSON minification other excluded tools accept."""
handler = _handler_with_router(_lossy_router())
pretty_json = "\n".join(
["{"] + [f' "key_{i}": {i},' for i in range(120)] + [' "end": true', "}"]
)
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_view",
"name": "view",
"arguments": '{"path": "/repo/data.json"}',
},
{
"type": "function_call_output",
"call_id": "call_view",
"output": pretty_json,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == pretty_json
def test_responses_malformed_arguments_do_not_break_extraction(monkeypatch):
"""Malformed function_call arguments yield no command -> normal compression."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_bad",
"name": "bash",
"arguments": "{not json at all",
},
{
"type": "function_call_output",
"call_id": "call_bad",
"output": _NL_OUTPUT,
},
],
}
new_payload, modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert modified is True
assert new_payload["input"][1]["output"] == "kept words"
def test_responses_protected_read_survives_cross_turn_dedup(monkeypatch):
"""A repeated protected read must not be replaced by a [↑…] dedup pointer."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
router = _lossy_router()
router._cross_turn_dedup_enabled = True
handler = _handler_with_router(router)
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_r1",
"name": "bash",
"arguments": '{"command": "nl -ba ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_r1",
"output": _NL_OUTPUT,
},
{
"type": "function_call",
"call_id": "call_r2",
"name": "bash",
"arguments": '{"command": "nl -ba ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_r2",
"output": _NL_OUTPUT,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _NL_OUTPUT
assert new_payload["input"][3]["output"] == _NL_OUTPUT
def test_responses_debug_path_with_excluded_list_output(monkeypatch):
"""Regression: debug logging over an excluded tool's content-part output must
not raise (latent unbound `fold` variable in the list branch)."""
from headroom.proxy.handlers import openai as openai_handler
monkeypatch.setattr(openai_handler, "_log_codex_compression_debug", lambda *a, **k: None)
handler = _handler_with_router(_lossy_router())
parts = [{"type": "output_text", "text": _FILE_CONTENT}]
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_read",
"name": "Read",
"arguments": '{"file_path": "/repo/ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_read",
"output": parts,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == parts
def test_responses_read_command_with_releasable_json_output_compresses(monkeypatch):
"""Content gate: a read command whose output is confidently DATA (JSON array)
is released to compression even with HEADROOM_PROTECT_READS=1."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
json_output = "[" + ",".join(f'{{"line": {i}, "text": "value {i}"}}' for i in range(60)) + "]"
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_json",
"name": "bash",
"arguments": '{"command": "cat data.json"}',
},
{
"type": "function_call_output",
"call_id": "call_json",
"output": json_output,
},
],
}
new_payload, modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert modified is True
assert new_payload["input"][1]["output"] == "kept words"
def test_responses_local_shell_call_string_command_read_stays_verbatim(monkeypatch):
"""local_shell_call with a string (not argv) command is also covered."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "local_shell_call",
"call_id": "call_lsc_str",
"action": {"type": "exec", "command": "cat src/app.py"},
},
{
"type": "local_shell_call_output",
"call_id": "call_lsc_str",
"output": _NL_OUTPUT,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _NL_OUTPUT
def test_responses_debug_path_with_read_protected_output(monkeypatch):
"""Debug logging over a read-protected output records and does not raise."""
from headroom.proxy.handlers import openai as openai_handler
monkeypatch.setattr(openai_handler, "_log_codex_compression_debug", lambda *a, **k: None)
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
{
"type": "function_call",
"call_id": "call_dbg",
"name": "bash",
"arguments": '{"command": "nl -ba ROADMAP.md"}',
},
{
"type": "function_call_output",
"call_id": "call_dbg",
"output": _NL_OUTPUT,
},
],
}
new_payload, _modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert new_payload["input"][1]["output"] == _NL_OUTPUT
def test_responses_read_scan_tolerates_non_dict_and_missing_call_id(monkeypatch):
"""The producer scan must skip non-dict items and calls without a string
call_id without breaking normal compression."""
monkeypatch.setenv("HEADROOM_PROTECT_READS", "1")
handler = _handler_with_router(_lossy_router())
payload = {
"model": "gpt-5",
"input": [
"a bare string item",
{
"type": "function_call",
"name": "bash",
"arguments": '{"command": "cat src/app.py"}',
},
{
"type": "function_call",
"call_id": 42,
"name": "bash",
"arguments": '{"command": "cat src/app.py"}',
},
{
"type": "function_call_output",
"call_id": "call_x",
"output": _NL_OUTPUT,
},
],
}
new_payload, modified, _s, _t, _u, _c, _a = _run(handler, payload)
assert modified is True
assert new_payload["input"][0] == "a bare string item"
assert new_payload["input"][3]["output"] == "kept words"