fix(proxy): preserve content-part array structure in excluded-tool lossless fold write-back (#2261)

## Description

When a tool result is excluded from lossy compression (e.g. grep via
`HEADROOM_EXCLUDE_TOOLS`), the OpenAI Responses adapter performs a
byte-lossless fold on the output text. However, the excluded-tool fold
path joined all content-part text with `_responses_part_text()` and
recorded a `("output", None)` slot, which caused `_set_slot_text` to
replace the entire `output` with a plain string.

For content-part arrays (valid per OpenAI spec: `[{type: output_text,
text: "..."}, {type: input_image, ...}]`), this destroyed the array
structure — non-text parts like images and refusals were silently
dropped.

Closes #2235

## 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/proxy/handlers/openai.py` — In the excluded-tool lossless
fold path, detect list (content-part) outputs and fold each
`input_text`/`output_text` part individually using `("output_part",
index)` slots, matching the eligibility rule already used by
`_slot_texts()` in the normal compression path
- `tests/test_openai_responses_compression_units.py` — Strengthen
existing content-part test to assert output remains a list; add new test
with mixed parts (output_text + input_image + refusal) to verify
structure preservation and byte-identical non-text parts

## Testing

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

### Test Output

```text
$ uv run pytest tests/test_openai_responses_compression_units.py -q --no-header
26 passed in 1.04s

$ uv run pytest tests/test_openai_responses_compression_units.py tests/test_openai_responses_context_compaction.py tests/test_openai_responses_traffic_learner.py -q --no-header
39 passed in 6.43s
```

## Real Behavior Proof

- Environment: Linux 6.8.0-124-generic, Python 3.12.3, headroom main @
eac49656
- Exact command / steps: checkout branch, run `uv run pytest
tests/test_openai_responses_compression_units.py -x -q`, inspect output
structure of excluded-tool items with content-part arrays
- Observed result: All 26 tests pass. For content-part outputs with
mixed types, the compressed output remains a list with the same length
and part types — non-text parts are byte-identical, only
`output_text`/`input_text` parts are updated
- Not tested: Live Codex WS end-to-end (requires Codex Desktop with
content-part tool outputs)

## 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 fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes

Co-authored-by: lennney <lennney@users.noreply.github.com>
This commit is contained in:
GUOHAO LIU 2026-07-17 04:51:34 +08:00 committed by GitHub
parent 4e2bbfee3f
commit 8951a264a2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 88 additions and 6 deletions

View file

@ -1609,10 +1609,28 @@ class OpenAIHandlerMixin:
# Protected from lossy compression — but grep/log/json output
# can still be losslessly compacted. Reuse the router helper
# so the Responses path matches the chat/Anthropic behavior.
excl_out = _responses_part_text(item.get("output"))
fold = router._lossless_compact_excluded(excl_out) if excl_out else None
if fold is not None:
lossless_excluded.append((idx, ("output", None), fold[0], excl_out))
# 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).
raw_output = item.get("output")
if isinstance(raw_output, list):
for pidx, part in enumerate(raw_output):
if (
isinstance(part, dict)
and part.get("type") in {"input_text", "output_text"}
and isinstance(part.get("text"), str)
):
part_text = part["text"]
pf = router._lossless_compact_excluded(part_text)
if pf is not None:
lossless_excluded.append(
(idx, ("output_part", pidx), pf[0], part_text)
)
else:
excl_out = _responses_part_text(raw_output)
fold = router._lossless_compact_excluded(excl_out) if excl_out else None
if fold is not None:
lossless_excluded.append((idx, ("output", None), fold[0], excl_out))
if debug_enabled:
extraction_debug.append(
{

View file

@ -756,8 +756,72 @@ def test_openai_responses_adapter_losslessly_folds_excluded_output_content_parts
assert saved >= 0
assert "router:excluded:lossless" in transforms
folded = new_payload["input"][1]["output"]
assert len(folded) < len(grep_out)
assert search_unheading(folded) == grep_out
# Output must remain a list (content-part array) — not replaced with a string
assert isinstance(folded, list), f"expected list, got {type(folded).__name__}"
assert len(folded) == 1
assert isinstance(folded[0], dict) and folded[0].get("type") == "output_text"
assert len(folded[0]["text"]) < len(grep_out)
assert search_unheading(folded[0]["text"]) == grep_out
def test_openai_responses_adapter_losslessly_folds_excluded_grep_output_content_parts_with_non_text():
"""Excluded tool output with content-part array preserves non-text parts.
When output is a content-part array that includes non-text parts (images,
refusals), the lossless fold should only update output_text/input_text parts
and leave everything else intact.
"""
from headroom.transforms.lossless_compaction import search_unheading
router = ContentRouter()
router.config.exclude_tools = {"grep"}
handler = _handler_with_router(router)
grep_out = "".join(
f"src/part_{f}.py:{ln}:matching content in a content part\n"
for f in range(8)
for ln in range(6)
)
payload = {
"model": "gpt-5",
"input": [
{"type": "function_call", "call_id": "call_1", "name": "grep", "arguments": "{}"},
{
"type": "function_call_output",
"call_id": "call_1",
"output": [
{"type": "output_text", "text": grep_out},
{"type": "input_image", "image_url": "https://example.com/img.png"},
{"type": "refusal", "refusal": "I cannot process this request"},
],
},
],
}
new_payload, modified, saved, transforms, _units, _chain, _attempted = (
handler._compress_openai_responses_live_text_units_with_router(
payload,
model="gpt-5",
request_id="req_content_part_non_text",
)
)
assert modified is True
assert saved >= 0
assert "router:excluded:lossless" in transforms
folded_list = new_payload["input"][1]["output"]
# Structure preserved: list with same length and part types
assert isinstance(folded_list, list), f"expected list, got {type(folded_list).__name__}"
assert len(folded_list) == 3
# output_text part was compressed
assert folded_list[0]["type"] == "output_text"
assert len(folded_list[0]["text"]) < len(grep_out)
assert search_unheading(folded_list[0]["text"]) == grep_out
# Non-text parts are byte-identical
assert folded_list[1]["type"] == "input_image"
assert folded_list[1]["image_url"] == "https://example.com/img.png"
assert folded_list[2]["type"] == "refusal"
assert folded_list[2]["refusal"] == "I cannot process this request"
def test_openai_responses_adapter_excludes_tool_case_insensitively_with_debug(monkeypatch):