From f77e0e4c6d14068cb249cbce3f31db7cc26ce968 Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Thu, 13 Aug 2026 16:48:45 +0400 Subject: [PATCH 1/7] fix(compact): sanitize plaintext compaction replays --- app/core/openai/requests.py | 63 ++++++++++++++-- tests/integration/test_proxy_compact.py | 56 ++++++++++++++ tests/unit/test_openai_requests.py | 98 +++++++++++++++++++++++++ 3 files changed, 211 insertions(+), 6 deletions(-) diff --git a/app/core/openai/requests.py b/app/core/openai/requests.py index b52d76fe55..c849d4581e 100644 --- a/app/core/openai/requests.py +++ b/app/core/openai/requests.py @@ -864,6 +864,7 @@ def strip_replayed_tool_call_namespaces_from_payload(payload: MutableJsonObject) _POISONED_LOCAL_COMPACT_FALLBACK_TEXT = "Local compact fallback preserved the latest encrypted reasoning state." +_COMPACT_STATE_TEXT_PREFIX = "[compact state] " _MAX_COMPACT_UPSTREAM_ESTIMATED_TOKENS = 100_000 _COMPACT_UPSTREAM_HEAD_ESTIMATED_TOKENS = 12_000 _ESTIMATED_CHARS_PER_TOKEN = 4 @@ -904,12 +905,10 @@ def _strip_poisoned_local_compact_fallback_items(payload: MutableJsonObject) -> skip_next_poison_compaction = False changed = False for item in input_items: - if skip_next_poison_compaction and is_json_mapping(item) and item.get("type") == "compaction": - encrypted_content = item.get("encrypted_content") - if isinstance(encrypted_content, str) and encrypted_content: - skip_next_poison_compaction = False - changed = True - continue + if skip_next_poison_compaction and _is_compaction_item(item): + skip_next_poison_compaction = False + changed = True + continue skip_next_poison_compaction = False if _is_poisoned_local_compact_fallback_message(item): @@ -917,12 +916,64 @@ def _strip_poisoned_local_compact_fallback_items(payload: MutableJsonObject) -> changed = True continue + replacement = _plaintext_compaction_replay_replacement(item) + if replacement is not None: + kept.append(replacement) + changed = True + continue + kept.append(item) if changed: payload["input"] = kept +def _is_compaction_item(item: JsonValue) -> bool: + if not is_json_mapping(item): + return False + return item.get("type") in {"compaction", "compaction_summary"} + + +def _plaintext_compaction_replay_replacement(item: JsonValue) -> JsonValue | None: + if not _is_compaction_item(item): + return None + assert is_json_mapping(item) + encrypted_content = item.get("encrypted_content") + if isinstance(encrypted_content, str) and _looks_like_provider_encrypted_content(encrypted_content): + return None + + text = _compaction_plaintext_summary(item) + if text is None: + return {"role": "assistant", "content": _COMPACT_STATE_TEXT_PREFIX + "[unverified compact state omitted]"} + return {"role": "assistant", "content": _COMPACT_STATE_TEXT_PREFIX + text} + + +def _looks_like_provider_encrypted_content(value: str) -> bool: + stripped = value.strip() + return stripped.startswith("gAAAA") and len(stripped) >= 80 + + +def _compaction_plaintext_summary(item: Mapping[str, JsonValue]) -> str | None: + for key in ("text", "summary"): + value = item.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + + content = item.get("content") + if is_json_list(content): + text_parts: list[str] = [] + for part in content: + if not is_json_mapping(part): + continue + part_text = part.get("text") + if isinstance(part_text, str) and part_text.strip(): + text_parts.append(part_text.strip()) + if text_parts: + return "\n".join(text_parts) + + return None + + def _is_poisoned_local_compact_fallback_message(item: JsonValue) -> bool: if not is_json_mapping(item): return False diff --git a/tests/integration/test_proxy_compact.py b/tests/integration/test_proxy_compact.py index bb6b2fb151..f4d9bdbeb4 100644 --- a/tests/integration/test_proxy_compact.py +++ b/tests/integration/test_proxy_compact.py @@ -336,6 +336,62 @@ async def fake_compact(payload, headers, access_token, account_id): assert "text" not in seen_payloads[0] +@pytest.mark.asyncio +async def test_proxy_compact_sanitizes_plaintext_replay_before_upstream(async_client, monkeypatch): + email = "compact-plaintext-replay@example.com" + raw_account_id = "acc_compact_plaintext_replay" + auth_json = _make_auth_json(raw_account_id, email) + files = {"auth_json": ("auth.json", json.dumps(auth_json), "application/json")} + response = await async_client.post("/api/accounts/import", files=files) + assert response.status_code == 200 + + seen_payloads: list[dict[str, object]] = [] + + async def fake_compact(payload, headers, access_token, account_id): + del headers, access_token, account_id + seen_payloads.append(cast(dict[str, object], payload.to_payload())) + return CompactResponsePayload.model_validate({"object": "response.compaction", "output": []}) + + monkeypatch.setattr(proxy_module, "core_compact_responses", fake_compact) + + payload = { + "model": "gpt-5.5", + "instructions": "compact", + "input": [ + {"role": "user", "content": "before"}, + { + "id": "cmp_local_summary", + "type": "compaction", + "status": "completed", + "summary": "I have the concrete code and evidence blockers in hand.", + }, + { + "id": "cmp_opaque_state", + "type": "compaction", + "status": "completed", + "encrypted_content": "opaque-provider-envelope-without-recognized-prefix", + }, + {"role": "user", "content": "continue"}, + ], + } + response = await async_client.post("/backend-api/codex/responses/compact", json=payload) + + assert response.status_code == 200 + assert len(seen_payloads) == 1 + assert seen_payloads[0]["input"] == [ + {"role": "user", "content": "before"}, + { + "role": "assistant", + "content": "[compact state] I have the concrete code and evidence blockers in hand.", + }, + { + "role": "assistant", + "content": "[compact state] [unverified compact state omitted]", + }, + {"role": "user", "content": "continue"}, + ] + + @pytest.mark.asyncio async def test_proxy_compact_preserves_historical_code_mode_side_effect_pair_before_ordinary_tail( async_client, monkeypatch diff --git a/tests/unit/test_openai_requests.py b/tests/unit/test_openai_requests.py index 8a883ad4be..483ba83c65 100644 --- a/tests/unit/test_openai_requests.py +++ b/tests/unit/test_openai_requests.py @@ -1130,6 +1130,104 @@ def test_compact_strips_poisoned_local_compact_fallback_items(): assert request.to_payload()["input"] == [{"role": "user", "content": "continue"}] +def test_compact_rewrites_plaintext_compaction_replay_without_encrypted_content(): + payload = { + "model": "gpt-5.5", + "instructions": "compact", + "input": [ + {"role": "user", "content": "before"}, + { + "id": "cmp_019ffa20-5e74-7b93-a86e-3c74bcf5521a", + "type": "compaction", + "status": "completed", + "encrypted_content": "I have the concrete code and evidence blockers in hand.", + }, + {"role": "user", "content": "continue"}, + ], + } + + request = ResponsesCompactRequest.model_validate(payload) + + assert request.to_payload()["input"] == [ + {"role": "user", "content": "before"}, + { + "role": "assistant", + "content": "[compact state] [unverified compact state omitted]", + }, + {"role": "user", "content": "continue"}, + ] + + +def test_compact_rewrites_plaintext_compaction_replay_summary(): + payload = { + "model": "gpt-5.5", + "instructions": "compact", + "input": [ + {"role": "user", "content": "before"}, + { + "id": "cmp_local_summary", + "type": "compaction", + "status": "completed", + "summary": "I have the concrete code and evidence blockers in hand.", + }, + {"role": "user", "content": "continue"}, + ], + } + + request = ResponsesCompactRequest.model_validate(payload) + + assert request.to_payload()["input"] == [ + {"role": "user", "content": "before"}, + { + "role": "assistant", + "content": "[compact state] I have the concrete code and evidence blockers in hand.", + }, + {"role": "user", "content": "continue"}, + ] + + +def test_compact_preserves_provider_encrypted_compaction_replay(): + encrypted_content = "gAAAA" + "A" * 80 + payload = { + "model": "gpt-5.5", + "instructions": "compact", + "input": [ + { + "id": "cmp_valid_provider_state", + "type": "compaction", + "status": "completed", + "encrypted_content": encrypted_content, + }, + {"role": "user", "content": "continue"}, + ], + } + + request = ResponsesCompactRequest.model_validate(payload) + + assert request.to_payload()["input"] == payload["input"] + + +def test_responses_preserves_provider_encrypted_compaction_replay(): + encrypted_content = "gAAAA" + "A" * 80 + payload = { + "model": "gpt-5.5", + "instructions": "continue", + "input": [ + { + "id": "cmp_valid_provider_state", + "type": "compaction", + "status": "completed", + "encrypted_content": encrypted_content, + }, + {"role": "user", "content": "continue"}, + ], + } + + request = ResponsesRequest.model_validate(payload) + + assert request.to_payload()["input"] == payload["input"] + + def test_compact_does_not_trim_many_small_input_items_for_upstream(): input_items = [{"role": "user", "content": f"item {idx}"} for idx in range(356)] payload = { From 4b47c752773c2d67313911b220c333bdf8f1b8ce Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Fri, 14 Aug 2026 21:00:53 +0400 Subject: [PATCH 2/7] fix(compact): preserve opaque compaction replays --- app/core/openai/requests.py | 14 ++++--- .../proposal.md | 17 +++++++++ .../specs/responses-api-compat/spec.md | 19 ++++++++++ .../tasks.md | 13 +++++++ tests/integration/test_proxy_compact.py | 13 +++++-- tests/unit/test_openai_requests.py | 38 ++++++++++++++----- 6 files changed, 95 insertions(+), 19 deletions(-) create mode 100644 openspec/changes/preserve-opaque-compaction-replays/proposal.md create mode 100644 openspec/changes/preserve-opaque-compaction-replays/specs/responses-api-compat/spec.md create mode 100644 openspec/changes/preserve-opaque-compaction-replays/tasks.md diff --git a/app/core/openai/requests.py b/app/core/openai/requests.py index c849d4581e..9b7f6a564a 100644 --- a/app/core/openai/requests.py +++ b/app/core/openai/requests.py @@ -939,18 +939,20 @@ def _plaintext_compaction_replay_replacement(item: JsonValue) -> JsonValue | Non return None assert is_json_mapping(item) encrypted_content = item.get("encrypted_content") - if isinstance(encrypted_content, str) and _looks_like_provider_encrypted_content(encrypted_content): + if isinstance(encrypted_content, str): return None text = _compaction_plaintext_summary(item) if text is None: - return {"role": "assistant", "content": _COMPACT_STATE_TEXT_PREFIX + "[unverified compact state omitted]"} - return {"role": "assistant", "content": _COMPACT_STATE_TEXT_PREFIX + text} + return _assistant_compact_state_item("[unverified compact state omitted]") + return _assistant_compact_state_item(text) -def _looks_like_provider_encrypted_content(value: str) -> bool: - stripped = value.strip() - return stripped.startswith("gAAAA") and len(stripped) >= 80 +def _assistant_compact_state_item(text: str) -> JsonValue: + return { + "role": "assistant", + "content": [{"type": "output_text", "text": _COMPACT_STATE_TEXT_PREFIX + text}], + } def _compaction_plaintext_summary(item: Mapping[str, JsonValue]) -> str | None: diff --git a/openspec/changes/preserve-opaque-compaction-replays/proposal.md b/openspec/changes/preserve-opaque-compaction-replays/proposal.md new file mode 100644 index 0000000000..542c2d9b58 --- /dev/null +++ b/openspec/changes/preserve-opaque-compaction-replays/proposal.md @@ -0,0 +1,17 @@ +## Why + +PR #1720 rewrites plaintext compaction replays before forwarding them upstream, but the current branch still treats only one ciphertext shape as provider-authored. Any other non-empty `encrypted_content` string is rewritten into synthetic assistant text, which drops the authoritative compact item identity and ciphertext even though the proxy's own compaction output normalizer preserves opaque encrypted strings unchanged. + +The branch also synthesizes assistant replacements after request validation, so compact replays forwarded as plaintext summaries can bypass the canonical assistant-content normalization path and still reach upstream with raw string `content`. + +## What Changes + +- Preserve any compaction item that already carries string `encrypted_content`; only rewrite plaintext summary/text/content shapes that lack encrypted payload. +- Emit rewritten plaintext compact-state summaries in canonical assistant `output_text` form so both standard and compact Responses payloads stay upstream-compatible. +- Document the replay-sanitization contract under `responses-api-compat`. + +## Impact + +- Prevents compact replay continuity loss for opaque provider envelopes. +- Keeps plaintext local summaries replay-safe without relying on post-validation string content. +- Aligns PR #1720's behavior change with the repository's OpenSpec-first merge gate. diff --git a/openspec/changes/preserve-opaque-compaction-replays/specs/responses-api-compat/spec.md b/openspec/changes/preserve-opaque-compaction-replays/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..97938b756c --- /dev/null +++ b/openspec/changes/preserve-opaque-compaction-replays/specs/responses-api-compat/spec.md @@ -0,0 +1,19 @@ +## ADDED Requirements + +### Requirement: Compaction replay sanitization preserves opaque encrypted state + +When a standard or compact Responses request replays a historical compaction item, the proxy MUST preserve that item unchanged whenever it already carries string `encrypted_content`. The proxy MUST NOT infer local plaintext provenance from ciphertext prefix or length heuristics, and MUST NOT replace such an item with synthetic assistant text solely because the encrypted payload shape is unfamiliar. + +When the replayed compaction item instead carries plaintext `summary`, `text`, or text-bearing `content` without string `encrypted_content`, the proxy MAY rewrite it into assistant context for upstream compatibility. Any assistant replacement synthesized during request serialization MUST use the canonical assistant content schema (`output_text` content parts), not a raw string `content` field. + +#### Scenario: compact replay keeps opaque encrypted content + +- **WHEN** `/backend-api/codex/responses/compact` receives a replayed `compaction` item whose `encrypted_content` is a non-empty opaque string +- **THEN** the forwarded upstream payload preserves that compaction item unchanged +- **AND** its `id`, `status`, and `encrypted_content` remain intact + +#### Scenario: plaintext compact summary rewrites to canonical assistant content + +- **WHEN** a replayed compact request contains a historical `compaction` item with plaintext `summary` and no string `encrypted_content` +- **THEN** the proxy rewrites it to an assistant item with `content` as an `output_text` array +- **AND** the synthesized text begins with the compact-state prefix used for local plaintext summaries diff --git a/openspec/changes/preserve-opaque-compaction-replays/tasks.md b/openspec/changes/preserve-opaque-compaction-replays/tasks.md new file mode 100644 index 0000000000..23e19d9686 --- /dev/null +++ b/openspec/changes/preserve-opaque-compaction-replays/tasks.md @@ -0,0 +1,13 @@ +## 1. Spec + +- [x] 1.1 Add a `responses-api-compat` delta for compact replay sanitization requirements. + +## 2. Implementation + +- [x] 2.1 Preserve opaque encrypted compaction items instead of inferring ciphertext provenance from prefix shape. +- [x] 2.2 Normalize synthesized plaintext compact-state assistant replacements to canonical `output_text` content arrays. + +## 3. Verification + +- [x] 3.1 Run focused unit/integration pytest coverage for compact replay handling. +- [x] 3.2 Run `ruff check` and `ruff format --check` on touched files. diff --git a/tests/integration/test_proxy_compact.py b/tests/integration/test_proxy_compact.py index f4d9bdbeb4..3f2a6920ae 100644 --- a/tests/integration/test_proxy_compact.py +++ b/tests/integration/test_proxy_compact.py @@ -382,11 +382,18 @@ async def fake_compact(payload, headers, access_token, account_id): {"role": "user", "content": "before"}, { "role": "assistant", - "content": "[compact state] I have the concrete code and evidence blockers in hand.", + "content": [ + { + "type": "output_text", + "text": "[compact state] I have the concrete code and evidence blockers in hand.", + } + ], }, { - "role": "assistant", - "content": "[compact state] [unverified compact state omitted]", + "id": "cmp_opaque_state", + "type": "compaction", + "status": "completed", + "encrypted_content": "opaque-provider-envelope-without-recognized-prefix", }, {"role": "user", "content": "continue"}, ] diff --git a/tests/unit/test_openai_requests.py b/tests/unit/test_openai_requests.py index 483ba83c65..22d731176d 100644 --- a/tests/unit/test_openai_requests.py +++ b/tests/unit/test_openai_requests.py @@ -1130,7 +1130,7 @@ def test_compact_strips_poisoned_local_compact_fallback_items(): assert request.to_payload()["input"] == [{"role": "user", "content": "continue"}] -def test_compact_rewrites_plaintext_compaction_replay_without_encrypted_content(): +def test_compact_preserves_opaque_encrypted_compaction_replay(): payload = { "model": "gpt-5.5", "instructions": "compact", @@ -1148,14 +1148,7 @@ def test_compact_rewrites_plaintext_compaction_replay_without_encrypted_content( request = ResponsesCompactRequest.model_validate(payload) - assert request.to_payload()["input"] == [ - {"role": "user", "content": "before"}, - { - "role": "assistant", - "content": "[compact state] [unverified compact state omitted]", - }, - {"role": "user", "content": "continue"}, - ] + assert request.to_payload()["input"] == payload["input"] def test_compact_rewrites_plaintext_compaction_replay_summary(): @@ -1180,7 +1173,12 @@ def test_compact_rewrites_plaintext_compaction_replay_summary(): {"role": "user", "content": "before"}, { "role": "assistant", - "content": "[compact state] I have the concrete code and evidence blockers in hand.", + "content": [ + { + "type": "output_text", + "text": "[compact state] I have the concrete code and evidence blockers in hand.", + } + ], }, {"role": "user", "content": "continue"}, ] @@ -1228,6 +1226,26 @@ def test_responses_preserves_provider_encrypted_compaction_replay(): assert request.to_payload()["input"] == payload["input"] +def test_responses_preserves_opaque_encrypted_compaction_replay(): + payload = { + "model": "gpt-5.5", + "instructions": "continue", + "input": [ + { + "id": "cmp_opaque_provider_state", + "type": "compaction", + "status": "completed", + "encrypted_content": "opaque-provider-envelope-without-recognized-prefix", + }, + {"role": "user", "content": "continue"}, + ], + } + + request = ResponsesRequest.model_validate(payload) + + assert request.to_payload()["input"] == payload["input"] + + def test_compact_does_not_trim_many_small_input_items_for_upstream(): input_items = [{"role": "user", "content": f"item {idx}"} for idx in range(356)] payload = { From 23513c59dbd2190dd0f30ea8d06bd95996e83ee6 Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Fri, 14 Aug 2026 03:57:51 +0400 Subject: [PATCH 3/7] fix(compact): preserve tool search pairs --- app/core/openai/requests.py | 1 + tests/integration/test_proxy_compact.py | 62 +++++++++++++++++++++++++ tests/unit/test_openai_requests.py | 36 ++++++++++++++ 3 files changed, 99 insertions(+) diff --git a/app/core/openai/requests.py b/app/core/openai/requests.py index 9b7f6a564a..c8c66e647d 100644 --- a/app/core/openai/requests.py +++ b/app/core/openai/requests.py @@ -48,6 +48,7 @@ "function_call_output": "function_call", "custom_tool_call_output": "custom_tool_call", "apply_patch_call_output": "apply_patch_call", + "tool_search_output": "tool_search_call", } _COMPACT_TOOL_CALL_ITEM_TYPES = frozenset(_COMPACT_TOOL_CALL_TYPE_BY_OUTPUT_TYPE.values()) _COMPACT_TOOL_CALL_OUTPUT_ITEM_TYPES = frozenset(_COMPACT_TOOL_CALL_TYPE_BY_OUTPUT_TYPE) diff --git a/tests/integration/test_proxy_compact.py b/tests/integration/test_proxy_compact.py index 3f2a6920ae..7e63380e22 100644 --- a/tests/integration/test_proxy_compact.py +++ b/tests/integration/test_proxy_compact.py @@ -460,6 +460,68 @@ async def fake_compact(payload, headers, access_token, account_id): ) +@pytest.mark.asyncio +async def test_proxy_compact_preserves_tool_search_pair_before_ordinary_tail(async_client, monkeypatch): + email = "compact-tool-search@example.com" + raw_account_id = "acc_compact_tool_search" + files = {"auth_json": ("auth.json", json.dumps(_make_auth_json(raw_account_id, email)), "application/json")} + response = await async_client.post("/api/accounts/import", files=files) + assert response.status_code == 200 + + seen_payloads: list[dict[str, object]] = [] + + async def fake_compact(payload, headers, access_token, account_id): + del headers, access_token, account_id + seen_payloads.append(cast(dict[str, object], payload.to_payload())) + return CompactResponsePayload.model_validate({"object": "response.compaction", "output": []}) + + monkeypatch.setattr(proxy_module, "core_compact_responses", fake_compact) + tool_call = { + "type": "tool_search_call", + "call_id": "call-search-tail", + "status": "completed", + "execution": "client", + "arguments": {"query": "codex-lb compaction tool search"}, + } + tool_output = { + "type": "tool_search_output", + "call_id": "call-search-tail", + "output": [{"title": "codex-lb compaction result"}], + } + ordinary_tail = {"role": "assistant", "content": "ordinary tail " + "x" * 500_000} + latest_request = {"role": "user", "content": "latest request"} + payload = { + "model": "gpt-5.6-sol", + "instructions": "hi", + "input": [ + {"role": "user", "content": "initial request"}, + {"role": "assistant", "content": "older answer " + "y" * 500_000}, + tool_call, + ordinary_tail, + tool_output, + latest_request, + ], + } + + response = await async_client.post("/backend-api/codex/responses/compact", json=payload) + + assert response.status_code == 200 + assert len(seen_payloads) == 1 + upstream_input = seen_payloads[0]["input"] + assert isinstance(upstream_input, list) + assert tool_call in upstream_input + assert tool_output in upstream_input + assert latest_request in upstream_input + assert all( + not ( + isinstance(item, dict) + and item.get("role") == "assistant" + and item.get("content") == [{"type": "output_text", "text": ordinary_tail["content"]}] + ) + for item in upstream_input + ) + + @pytest.mark.asyncio async def test_proxy_compact_omits_oversized_optional_tool_tail_before_upstream(async_client, monkeypatch): email = "compact-optional-tail@example.com" diff --git a/tests/unit/test_openai_requests.py b/tests/unit/test_openai_requests.py index 22d731176d..17761dbf9b 100644 --- a/tests/unit/test_openai_requests.py +++ b/tests/unit/test_openai_requests.py @@ -2769,6 +2769,42 @@ def test_compact_trimming_keeps_selected_tool_calls_with_matching_outputs(): assert tool_output in dumped_input +def test_compact_trimming_keeps_tool_search_outputs_with_matching_calls(): + tool_call = { + "type": "tool_search_call", + "call_id": "call_search_tail", + "status": "completed", + "execution": "client", + "arguments": {"query": "spawn_agent multi-agent schema", "limit": 8}, + } + tool_output = { + "type": "tool_search_output", + "call_id": "call_search_tail", + "output": "Found matching tools", + } + input_items = [ + {"role": "user", "content": "initial instructions"}, + {"role": "assistant", "content": "x" * 500_000}, + tool_call, + {"role": "assistant", "content": "y" * 500_000}, + tool_output, + {"role": "user", "content": "latest request"}, + ] + payload = { + "model": "gpt-5.1", + "instructions": "hi", + "input": input_items, + } + + request = ResponsesCompactRequest.model_validate(payload) + dumped = request.to_payload() + dumped_input = dumped["input"] + + assert isinstance(dumped_input, list) + assert tool_call in dumped_input + assert tool_output in dumped_input + + def test_compact_trimming_reconciles_duplicate_tool_call_ids_by_occurrence(): first_tool_call = { "type": "function_call", From edde0b142f153759043d5d769f69281b4db72b24 Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Fri, 14 Aug 2026 04:19:04 +0400 Subject: [PATCH 4/7] fix(proxy): accept compact tool-search replay state --- .../proxy/_service/http_bridge/helpers.py | 4 +- .../proxy/_service/websocket/helpers.py | 5 +- app/modules/proxy/replay_safety.py | 34 ++++++ .../proposal.md | 4 + .../specs/responses-api-compat/spec.md | 15 +++ .../tasks.md | 1 + .../integration/test_http_responses_bridge.py | 115 ++++++++++++++++++ tests/unit/test_proxy_utils.py | 70 +++++++++++ tests/unit/test_replay_safety.py | 67 ++++++++++ 9 files changed, 311 insertions(+), 4 deletions(-) diff --git a/app/modules/proxy/_service/http_bridge/helpers.py b/app/modules/proxy/_service/http_bridge/helpers.py index 27072ceaf1..971c9068ea 100644 --- a/app/modules/proxy/_service/http_bridge/helpers.py +++ b/app/modules/proxy/_service/http_bridge/helpers.py @@ -530,7 +530,7 @@ def _trim_http_bridge_previous_response_input_items(input_items: list[JsonValue] index for index, item in enumerate(input_items) if _http_bridge_input_item_type(item) - in {"function_call_output", "custom_tool_call_output", "apply_patch_call_output"} + in {"function_call_output", "custom_tool_call_output", "apply_patch_call_output", "tool_search_output"} ), None, ) @@ -544,7 +544,7 @@ def _trim_http_bridge_previous_response_input_items(input_items: list[JsonValue] def _is_http_bridge_previous_response_output_item(item: JsonValue) -> bool: item_type = _http_bridge_input_item_type(item) - if item_type in {"reasoning", "function_call", "custom_tool_call", "apply_patch_call"}: + if item_type in {"reasoning", "function_call", "custom_tool_call", "apply_patch_call", "tool_search_call"}: return _has_http_bridge_response_output_marker(item) if item_type != "message" or not isinstance(item, dict): return False diff --git a/app/modules/proxy/_service/websocket/helpers.py b/app/modules/proxy/_service/websocket/helpers.py index 3a0ee3d4c7..fa3669fb19 100644 --- a/app/modules/proxy/_service/websocket/helpers.py +++ b/app/modules/proxy/_service/websocket/helpers.py @@ -522,6 +522,7 @@ def _websocket_continuity_anchor_for_payload( "function_call_output": "function_call", "custom_tool_call_output": "custom_tool_call", "apply_patch_call_output": "apply_patch_call", + "tool_search_output": "tool_search_call", } _WEBSOCKET_TOOL_CALL_ITEM_TYPES = frozenset(_WEBSOCKET_TOOL_CALL_ITEM_TYPES_BY_OUTPUT_TYPE.values()) @@ -1881,7 +1882,7 @@ def _trim_websocket_previous_response_input_items(input_items: list[JsonValue]) index for index, item in enumerate(input_items) if _websocket_input_item_type(item) - in {"function_call_output", "custom_tool_call_output", "apply_patch_call_output"} + in {"function_call_output", "custom_tool_call_output", "apply_patch_call_output", "tool_search_output"} ), None, ) @@ -1897,7 +1898,7 @@ def _is_websocket_previous_response_output_item(item: JsonValue) -> bool: if isinstance(item, dict) and _websocket_input_item_type(item) is None and item.get("role") == "assistant": return True item_type = _websocket_input_item_type(item) - if item_type in {"reasoning", "function_call", "custom_tool_call", "apply_patch_call"}: + if item_type in {"reasoning", "function_call", "custom_tool_call", "apply_patch_call", "tool_search_call"}: return True if item_type != "message" or not isinstance(item, dict): return False diff --git a/app/modules/proxy/replay_safety.py b/app/modules/proxy/replay_safety.py index fd44be4fbb..e54c68b23f 100644 --- a/app/modules/proxy/replay_safety.py +++ b/app/modules/proxy/replay_safety.py @@ -15,6 +15,7 @@ "function_call_output": "function_call", "custom_tool_call_output": "custom_tool_call", "apply_patch_call_output": "apply_patch_call", + "tool_search_output": "tool_search_call", } _TOOL_CALL_TYPES = frozenset(_TOOL_CALL_TYPE_BY_OUTPUT_TYPE.values()) _ACCOUNT_NEUTRAL_REPLAY_OMITTED_ITEM_TYPES = frozenset( @@ -39,6 +40,7 @@ "additional_tools", "apply_patch_call", "apply_patch_call_output", + "compaction", "custom_tool_call", "custom_tool_call_output", "function_call", @@ -47,6 +49,8 @@ "input_image", "input_text", "message", + "tool_search_call", + "tool_search_output", } ) _ACCOUNT_NEUTRAL_MESSAGE_CONTENT_TYPES = frozenset( @@ -65,6 +69,7 @@ } _ACCOUNT_NEUTRAL_INPUT_ITEM_FIELDS = { "additional_tools": frozenset({"role", "tools", "type"}), + "compaction": frozenset({"encrypted_content", "id", "status", "type"}), "apply_patch_call": frozenset( { "call_id", @@ -93,6 +98,22 @@ "function_call_output": frozenset( {"call_id", "caller", "id", _INTERNAL_CHAT_MESSAGE_METADATA_FIELD, "output", "status", "type"} ), + "tool_search_call": frozenset( + {"arguments", "call_id", "caller", "execution", "id", _INTERNAL_CHAT_MESSAGE_METADATA_FIELD, "status", "type"} + ), + "tool_search_output": frozenset( + { + "call_id", + "caller", + "execution", + "id", + _INTERNAL_CHAT_MESSAGE_METADATA_FIELD, + "output", + "status", + "tools", + "type", + } + ), } _ACCOUNT_NEUTRAL_ITEM_STATUSES = frozenset({"completed", "failed"}) _ACCOUNT_NEUTRAL_APPLY_PATCH_OPERATION_FIELDS = { @@ -269,6 +290,10 @@ def responses_input_items_are_self_contained_fresh_replay(input_items: list[Json item_type = item_type_value if isinstance(item_type_value, str) else None if not _input_item_has_only_known_fields(item, item_type): return False + if item_type == "compaction": + if not _compaction_item_is_self_contained(item): + return False + continue call_id_value = item.get("call_id") call_id = call_id_value if isinstance(call_id_value, str) and call_id_value else None if item_type in _TOOL_CALL_TYPES: @@ -628,6 +653,9 @@ def _tool_call_is_self_contained(item_type: str, item: Mapping[str, JsonValue]) return _is_nonblank_string(item.get("name")) and isinstance(item.get("arguments"), str) if item_type == "custom_tool_call": return _is_nonblank_string(item.get("name")) and isinstance(item.get("input"), str) + if item_type == "tool_search_call": + arguments = item.get("arguments") + return isinstance(arguments, dict) and item.get("execution") in (None, "client") operation = item.get("operation") patch = item.get("patch") input_value = item.get("input") @@ -640,6 +668,10 @@ def _tool_call_is_self_contained(item_type: str, item: Mapping[str, JsonValue]) return _is_nonblank_string(input_value) +def _compaction_item_is_self_contained(item: Mapping[str, JsonValue]) -> bool: + return item.get("status") in (None, "completed") and _is_nonblank_string(item.get("encrypted_content")) + + def _caller_is_self_contained(item: Mapping[str, JsonValue]) -> bool: caller = item.get("caller") return caller is None or caller == {"type": "direct"} @@ -1017,6 +1049,8 @@ def _contains_account_scoped_input_state(value: JsonValue) -> bool: return True if item_type == "additional_tools" and not _tools_are_account_neutral(current.get("tools")): return True + if item_type == "compaction" and _compaction_item_is_self_contained(current): + continue if ( isinstance(item_type, str) and (item_type.endswith("_call") or item_type.endswith("_call_output")) diff --git a/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/proposal.md b/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/proposal.md index 947fc37ef1..fdf4aa4f8c 100644 --- a/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/proposal.md +++ b/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/proposal.md @@ -37,6 +37,10 @@ item type is real, only the classifier list is stale. there is nothing for the continuity recovery paths to do with it. - Add regression coverage for the classifier and for the HTTP-bridge masking surface. +- Treat `tool_search_call` / `tool_search_output` like the other client-side + tool pairs when trimming already-stored previous-response replay prefixes, so + compaction-preserved tool-search pairs do not get resent in full on top of a + `previous_response_id` anchor. ## Non-goals diff --git a/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/specs/responses-api-compat/spec.md b/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/specs/responses-api-compat/spec.md index d5290c93d9..3fb39de032 100644 --- a/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/specs/responses-api-compat/spec.md +++ b/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/specs/responses-api-compat/spec.md @@ -18,3 +18,18 @@ The service MUST classify an upstream `invalid_request_error` with `param=input` #### Scenario: hosted web search wording stays unclassified - **WHEN** upstream emits `invalid_request_error` with `param=input` and a message starting `No tool output found for web search call` - **THEN** the service does not treat it as a missing-tool-output continuity error + +### Requirement: Previous-response replay trimming handles tool-search output pairs +When a Responses HTTP bridge or WebSocket continuation carries `previous_response_id` and replays already-stored response output items before a fresh `tool_search_output`, the service MUST trim the replayed `tool_search_call` prefix and preserve the `tool_search_output` plus the fresh turn. The service MUST NOT forward both the replayed `tool_search_call` and its `tool_search_output` on top of the `previous_response_id` anchor. + +#### Scenario: HTTP bridge trims replayed tool-search call prefix +- **GIVEN** an HTTP bridge session has a completed previous response +- **WHEN** the next request carries `previous_response_id` and input `[tool_search_call, tool_search_output, user_message]` +- **THEN** the upstream request keeps the same `previous_response_id` +- **AND** its input is `[tool_search_output, user_message]` + +#### Scenario: WebSocket bridge trims replayed tool-search call prefix +- **GIVEN** a WebSocket Responses session has a completed previous response +- **WHEN** the next request carries `previous_response_id` and input `[tool_search_call, tool_search_output, user_message]` +- **THEN** the upstream request keeps the same `previous_response_id` +- **AND** its input is `[tool_search_output, user_message]` diff --git a/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/tasks.md b/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/tasks.md index 34446201c3..71ce7c8ee5 100644 --- a/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/tasks.md +++ b/openspec/changes/archive/2026-08-13-classify-tool-search-missing-tool-output/tasks.md @@ -3,4 +3,5 @@ - [x] Extend the missing-tool-output message classifier with the tool-search wording. - [x] Keep the hosted `web search call` wording unclassified. - [x] Add classifier and HTTP-bridge masking regression coverage. +- [x] Trim replayed tool-search call prefixes on previous-response HTTP bridge and WebSocket continuations. - [x] Run focused unit tests, lint/format, type check, architecture check, diff check, and strict OpenSpec validation. diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index d4b3a0fda9..350d6f106c 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -5661,6 +5661,121 @@ async def fake_connect_responses_websocket( assert second_upstream_payload["input"] == [replayed_apply_patch_output, next_user_message] +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_trims_replayed_tool_search_previous_response_prefix( + async_client, + monkeypatch, +): + _install_bridge_settings(monkeypatch, enabled=True) + account_id = await _import_account( + async_client, + "acc_http_bridge_tool_search_trim", + "http-bridge-tool-search-trim@example.com", + ) + account = await _get_account(account_id) + fake_upstream = _FakeBridgeUpstreamWebSocket() + + async def fake_select_account_with_budget( + self, + deadline, + *, + request_id, + kind, + request_stage="first_turn", + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids=None, + additional_limit_name=None, + api_key=None, + preferred_account_id=None, + ): + del preferred_account_id + del ( + self, + deadline, + request_id, + kind, + request_stage, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids, + additional_limit_name, + ) + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + del headers, access_token, account_id_header, base_url, session + return fake_upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + first = await async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Search the tools.", + "input": [{"role": "user", "content": [{"type": "input_text", "text": "find context"}]}], + "prompt_cache_key": "http-bridge-tool-search-trim-1", + }, + ) + assert first.status_code == 200 + first_body = first.json() + assert first_body["id"] == "resp_bridge_1" + + replayed_tool_search_call = { + "id": "tsc_replay", + "type": "tool_search_call", + "status": "completed", + "call_id": "call_search_1", + } + replayed_tool_search_output = { + "type": "tool_search_output", + "call_id": "call_search_1", + "output": [{"title": "context result"}], + } + next_user_message = {"role": "user", "content": [{"type": "input_text", "text": "continue"}]} + second = await async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Search the tools.", + "previous_response_id": first_body["id"], + "input": [replayed_tool_search_call, replayed_tool_search_output, next_user_message], + "prompt_cache_key": "http-bridge-tool-search-trim-1", + }, + ) + assert second.status_code == 200 + assert second.json()["id"] == "resp_bridge_2" + + assert len(fake_upstream.sent_text) == 2 + second_upstream_payload = json.loads(fake_upstream.sent_text[1]) + assert second_upstream_payload["previous_response_id"] == "resp_bridge_1" + assert second_upstream_payload["input"] == [replayed_tool_search_output, next_user_message] + + @pytest.mark.asyncio async def test_backend_responses_http_bridge_lite_request_omits_synthesized_tools( async_client, diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index 9ed5b4d98f..d9cfb0169c 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -21155,6 +21155,34 @@ def test_websocket_client_previous_response_full_resend_retry_allows_self_contai ) +def test_websocket_client_previous_response_full_resend_retry_allows_tool_search_history() -> None: + self_contained_tool_search_history: list[JsonValue] = [ + {"role": "user", "content": [{"type": "input_text", "text": "search tools"}]}, + { + "type": "tool_search_call", + "call_id": "call_search", + "arguments": {"query": "mcp tools"}, + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search", + "output": [{"title": "tool_search"}], + "status": "completed", + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + assert ( + proxy_service._websocket_client_previous_response_full_resend_is_retry_safe( + previous_response_id="resp_client_anchor", + input_value=self_contained_tool_search_history, + continuity_state=None, + ) + is True + ) + + @pytest.mark.asyncio async def test_prepare_websocket_response_create_request_fills_interrupted_pending_tool_outputs(monkeypatch): request_logs = _RequestLogsRecorder() @@ -40301,6 +40329,48 @@ def test_trim_websocket_previous_response_input_items_handles_apply_patch_replay assert trimmed == input_items[2:] +def test_trim_http_bridge_previous_response_input_items_handles_tool_search_replay(): + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "id": "tsc_replay", + "call_id": "call_search_1", + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search_1", + "output": [{"title": "result"}], + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + trimmed = proxy_service._trim_http_bridge_previous_response_input_items(input_items) + + assert trimmed == input_items[1:] + + +def test_trim_websocket_previous_response_input_items_handles_tool_search_replay(): + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "id": "tsc_replay", + "call_id": "call_search_1", + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search_1", + "output": [{"title": "result"}], + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + trimmed = proxy_service._trim_websocket_previous_response_input_items(input_items) + + assert trimmed == input_items[1:] + + def test_prepare_response_bridge_request_state_keeps_unconfirmed_missing_tool_output_history(): request_logs = _RequestLogsRecorder() service = proxy_service.ProxyService(_repo_factory(request_logs)) diff --git a/tests/unit/test_replay_safety.py b/tests/unit/test_replay_safety.py index e17cf26c36..8330eb3879 100644 --- a/tests/unit/test_replay_safety.py +++ b/tests/unit/test_replay_safety.py @@ -10,6 +10,7 @@ ) from app.modules.proxy.replay_safety import ( project_responses_input_for_account_neutral_fresh_replay, + responses_input_items_are_self_contained_fresh_replay, responses_input_suffix_matches_pending_tool_calls, responses_input_suffix_retains_prior_output, responses_payload_is_account_neutral_fresh_replay, @@ -151,6 +152,51 @@ def test_account_neutral_fresh_replay_accepts_self_contained_payloads( assert responses_payload_is_account_neutral_fresh_replay(payload) is True +def test_account_neutral_fresh_replay_accepts_compaction_context_item() -> None: + payload: dict[str, JsonValue] = { + "input": [ + { + "type": "compaction", + "status": "completed", + "encrypted_content": "encrypted-compact-context", + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "continue"}], + }, + ], + } + + assert responses_payload_is_account_neutral_fresh_replay(payload) is True + + +def test_account_neutral_replay_projection_preserves_compaction_without_response_id() -> None: + input_items: list[JsonValue] = [ + { + "type": "compaction", + "id": "cmp_owner_a", + "status": "completed", + "encrypted_content": "encrypted-compact-context", + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "continue"}], + }, + ] + + projection = project_responses_input_for_account_neutral_fresh_replay(input_items, stored_count=1) + + assert projection is not None + assert projection.input_items[0] == { + "type": "compaction", + "status": "completed", + "encrypted_content": "encrypted-compact-context", + } + assert responses_payload_is_account_neutral_fresh_replay({"input": projection.input_items}) is True + + def test_account_neutral_replay_projection_removes_response_owned_bookkeeping() -> None: metadata = {"turn_id": "turn_owner_a"} input_items: list[JsonValue] = [ @@ -328,6 +374,27 @@ def test_account_neutral_replay_projection_preserves_noncompleted_search_state_t assert responses_payload_is_account_neutral_fresh_replay({"input": projection.input_items}) is False +def test_account_neutral_fresh_replay_accepts_self_contained_tool_search_pair() -> None: + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "call_id": "call_search", + "arguments": {"query": "codex-lb"}, + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search", + "output": "Found codex-lb", + "status": "completed", + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + assert responses_input_items_are_self_contained_fresh_replay(input_items) is True + assert responses_payload_is_account_neutral_fresh_replay({"input": input_items}) is True + + @pytest.mark.parametrize( "suffix", [ From c5b5f9da47f6126a023a4ed527d86a51be9c19f2 Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Fri, 14 Aug 2026 04:33:45 +0400 Subject: [PATCH 5/7] fix(http-bridge): switch silent account-neutral retries --- .../_service/http_bridge/request_submit.py | 5 +- .../integration/test_http_responses_bridge.py | 80 +++++++++++++++++++ 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 23216a8c36..7f6b11c125 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -3020,12 +3020,13 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: # Account-scoped uploaded files cannot be replayed on a # different owner. Keep the preferred account mandatory for # both silent recovery and clean-close recovery. - require_preferred_reconnect = account_neutral_recovery or request_state.file_required_preferred_account + require_preferred_reconnect = request_state.file_required_preferred_account request_text = _prepare_websocket_request_state_for_visible_output_replay(request_state) if request_text is None: return False if account_neutral_recovery: - request_state.preferred_account_id = session.account.id + request_state.preferred_account_id = None + request_state.excluded_account_ids.add(session.account.id) elif not request_state.file_required_preferred_account: if hard_owner_bound and not model_fallback_replay and not fresh_hard_request_account_switch_allowed: request_state.preferred_account_id = session.account.id diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 350d6f106c..2cb7f358d8 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -13006,6 +13006,86 @@ async def fake_reconnect( assert replacement_upstream.sent_text == [retry_request.request_text] +@pytest.mark.asyncio +async def test_retry_account_neutral_precreated_request_switches_from_silent_account(app_instance, monkeypatch): + from app.modules.proxy.continuity import make_http_bridge_account_neutral_replay_key + + service = get_proxy_service_for_app(app_instance) + recovery_kind, recovery_key = make_http_bridge_account_neutral_replay_key("retry-silent-account") + first_account = cast(Account, SimpleNamespace(id="acct-silent", status=AccountStatus.ACTIVE, plan_type="plus")) + replacement_account = cast( + Account, + SimpleNamespace(id="acct-replacement", status=AccountStatus.ACTIVE, plan_type="plus"), + ) + replacement_upstream = _RecordingUpstreamWebSocket() + session = proxy_module._HTTPBridgeSession( + key=proxy_module._HTTPBridgeSessionKey(recovery_kind, recovery_key, None), + headers={"x-codex-turn-state": "stale-turn-state"}, + affinity=proxy_module._AffinityPolicy(), + request_model="gpt-5.5", + account=first_account, + upstream=cast(proxy_module.UpstreamWebSocket, _SilentUpstreamWebSocket()), + upstream_control=proxy_module._WebSocketUpstreamControl(), + pending_lock=anyio.Lock(), + pending_requests=deque(), + response_create_gate=asyncio.Semaphore(1), + queued_request_count=1, + last_used_at=time.monotonic(), + idle_ttl_seconds=120.0, + ) + request_state = proxy_module._WebSocketRequestState( + request_id="req-account-neutral-precreated-retry", + model="gpt-5.5", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + awaiting_response_created=True, + transport="http", + response_create_gate_acquired=True, + request_text=json.dumps({"type": "response.create", "model": "gpt-5.5", "input": []}), + ) + session.pending_requests.append(request_state) + reconnect_calls: list[dict[str, object]] = [] + + async def fake_reconnect( + self, + target_session, + *, + request_state, + restart_reader=False, + require_same_account=False, + require_preferred_account=False, + ): + del self, restart_reader + reconnect_calls.append( + { + "require_same_account": require_same_account, + "require_preferred_account": require_preferred_account, + "preferred_account_id": target_session.account.id, + "excluded_account_ids": set(request_state.excluded_account_ids), + } + ) + target_session.account = replacement_account + target_session.upstream = replacement_upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_reconnect_http_bridge_session", fake_reconnect) + + assert await service._retry_http_bridge_precreated_request(session) is True + + assert reconnect_calls == [ + { + "require_same_account": False, + "require_preferred_account": False, + "preferred_account_id": "acct-silent", + "excluded_account_ids": {"acct-silent"}, + } + ] + assert request_state.preferred_account_id is None + assert session.account.id == "acct-replacement" + assert replacement_upstream.sent_text == [request_state.request_text] + + @pytest.mark.asyncio async def test_v1_responses_http_bridge_send_failure_returns_upstream_unavailable( async_client, From 0ff7778a478cf9abab4f8e83d067d0ab53cb91b3 Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Fri, 14 Aug 2026 12:44:35 +0400 Subject: [PATCH 6/7] test(compact): avoid stack conflict for replay coverage --- .../integration/test_http_responses_bridge.py | 390 +++++++++--------- tests/integration/test_proxy_compact.py | 124 +++--- tests/unit/test_openai_requests.py | 72 ++-- tests/unit/test_proxy_utils.py | 140 +++---- tests/unit/test_replay_safety.py | 106 ++--- 5 files changed, 403 insertions(+), 429 deletions(-) diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 2cb7f358d8..58eaf0f1f0 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -5661,121 +5661,6 @@ async def fake_connect_responses_websocket( assert second_upstream_payload["input"] == [replayed_apply_patch_output, next_user_message] -@pytest.mark.asyncio -async def test_v1_responses_http_bridge_trims_replayed_tool_search_previous_response_prefix( - async_client, - monkeypatch, -): - _install_bridge_settings(monkeypatch, enabled=True) - account_id = await _import_account( - async_client, - "acc_http_bridge_tool_search_trim", - "http-bridge-tool-search-trim@example.com", - ) - account = await _get_account(account_id) - fake_upstream = _FakeBridgeUpstreamWebSocket() - - async def fake_select_account_with_budget( - self, - deadline, - *, - request_id, - kind, - request_stage="first_turn", - sticky_key, - sticky_kind, - reallocate_sticky, - sticky_max_age_seconds, - prefer_earlier_reset_accounts, - routing_strategy, - model, - exclude_account_ids=None, - additional_limit_name=None, - api_key=None, - preferred_account_id=None, - ): - del preferred_account_id - del ( - self, - deadline, - request_id, - kind, - request_stage, - sticky_key, - sticky_kind, - reallocate_sticky, - sticky_max_age_seconds, - prefer_earlier_reset_accounts, - routing_strategy, - model, - exclude_account_ids, - additional_limit_name, - ) - return AccountSelection(account=account, error_message=None, error_code=None) - - async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): - del self, force, timeout_seconds - return target - - async def fake_connect_responses_websocket( - headers, - access_token, - account_id_header, - *, - base_url=None, - session=None, - ): - del headers, access_token, account_id_header, base_url, session - return fake_upstream - - monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) - monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) - monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) - - first = await async_client.post( - "/v1/responses", - json={ - "model": "gpt-5.1", - "instructions": "Search the tools.", - "input": [{"role": "user", "content": [{"type": "input_text", "text": "find context"}]}], - "prompt_cache_key": "http-bridge-tool-search-trim-1", - }, - ) - assert first.status_code == 200 - first_body = first.json() - assert first_body["id"] == "resp_bridge_1" - - replayed_tool_search_call = { - "id": "tsc_replay", - "type": "tool_search_call", - "status": "completed", - "call_id": "call_search_1", - } - replayed_tool_search_output = { - "type": "tool_search_output", - "call_id": "call_search_1", - "output": [{"title": "context result"}], - } - next_user_message = {"role": "user", "content": [{"type": "input_text", "text": "continue"}]} - second = await async_client.post( - "/v1/responses", - json={ - "model": "gpt-5.1", - "instructions": "Search the tools.", - "previous_response_id": first_body["id"], - "input": [replayed_tool_search_call, replayed_tool_search_output, next_user_message], - "prompt_cache_key": "http-bridge-tool-search-trim-1", - }, - ) - assert second.status_code == 200 - assert second.json()["id"] == "resp_bridge_2" - - assert len(fake_upstream.sent_text) == 2 - second_upstream_payload = json.loads(fake_upstream.sent_text[1]) - assert second_upstream_payload["previous_response_id"] == "resp_bridge_1" - assert second_upstream_payload["input"] == [replayed_tool_search_output, next_user_message] - - @pytest.mark.asyncio async def test_backend_responses_http_bridge_lite_request_omits_synthesized_tools( async_client, @@ -6988,6 +6873,201 @@ async def fail_refresh(self, target, *, force=False, timeout_seconds): assert "x-codex-turn-state" not in response.headers +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_trims_replayed_tool_search_previous_response_prefix( + async_client, + monkeypatch, +): + _install_bridge_settings(monkeypatch, enabled=True) + account_id = await _import_account( + async_client, + "acc_http_bridge_tool_search_trim", + "http-bridge-tool-search-trim@example.com", + ) + account = await _get_account(account_id) + fake_upstream = _FakeBridgeUpstreamWebSocket() + + async def fake_select_account_with_budget( + self, + deadline, + *, + request_id, + kind, + request_stage="first_turn", + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids=None, + additional_limit_name=None, + api_key=None, + preferred_account_id=None, + ): + del preferred_account_id + del ( + self, + deadline, + request_id, + kind, + request_stage, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids, + additional_limit_name, + ) + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + del headers, access_token, account_id_header, base_url, session + return fake_upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + first = await async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Search the tools.", + "input": [{"role": "user", "content": [{"type": "input_text", "text": "find context"}]}], + "prompt_cache_key": "http-bridge-tool-search-trim-1", + }, + ) + assert first.status_code == 200 + first_body = first.json() + assert first_body["id"] == "resp_bridge_1" + + replayed_tool_search_call = { + "id": "tsc_replay", + "type": "tool_search_call", + "status": "completed", + "call_id": "call_search_1", + } + replayed_tool_search_output = { + "type": "tool_search_output", + "call_id": "call_search_1", + "output": [{"title": "context result"}], + } + next_user_message = {"role": "user", "content": [{"type": "input_text", "text": "continue"}]} + second = await async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Search the tools.", + "previous_response_id": first_body["id"], + "input": [replayed_tool_search_call, replayed_tool_search_output, next_user_message], + "prompt_cache_key": "http-bridge-tool-search-trim-1", + }, + ) + assert second.status_code == 200 + assert second.json()["id"] == "resp_bridge_2" + + assert len(fake_upstream.sent_text) == 2 + second_upstream_payload = json.loads(fake_upstream.sent_text[1]) + assert second_upstream_payload["previous_response_id"] == "resp_bridge_1" + assert second_upstream_payload["input"] == [replayed_tool_search_output, next_user_message] + + +@pytest.mark.asyncio +async def test_retry_account_neutral_precreated_request_switches_from_silent_account(app_instance, monkeypatch): + from app.modules.proxy.continuity import make_http_bridge_account_neutral_replay_key + + service = get_proxy_service_for_app(app_instance) + recovery_kind, recovery_key = make_http_bridge_account_neutral_replay_key("retry-silent-account") + first_account = cast(Account, SimpleNamespace(id="acct-silent", status=AccountStatus.ACTIVE, plan_type="plus")) + replacement_account = cast( + Account, + SimpleNamespace(id="acct-replacement", status=AccountStatus.ACTIVE, plan_type="plus"), + ) + replacement_upstream = _RecordingUpstreamWebSocket() + session = proxy_module._HTTPBridgeSession( + key=proxy_module._HTTPBridgeSessionKey(recovery_kind, recovery_key, None), + headers={"x-codex-turn-state": "stale-turn-state"}, + affinity=proxy_module._AffinityPolicy(), + request_model="gpt-5.5", + account=first_account, + upstream=cast(proxy_module.UpstreamWebSocket, _SilentUpstreamWebSocket()), + upstream_control=proxy_module._WebSocketUpstreamControl(), + pending_lock=anyio.Lock(), + pending_requests=deque(), + response_create_gate=asyncio.Semaphore(1), + queued_request_count=1, + last_used_at=time.monotonic(), + idle_ttl_seconds=120.0, + ) + request_state = proxy_module._WebSocketRequestState( + request_id="req-account-neutral-precreated-retry", + model="gpt-5.5", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + awaiting_response_created=True, + transport="http", + response_create_gate_acquired=True, + request_text=json.dumps({"type": "response.create", "model": "gpt-5.5", "input": []}), + ) + session.pending_requests.append(request_state) + reconnect_calls: list[dict[str, object]] = [] + + async def fake_reconnect( + self, + target_session, + *, + request_state, + restart_reader=False, + require_same_account=False, + require_preferred_account=False, + ): + del self, restart_reader + reconnect_calls.append( + { + "require_same_account": require_same_account, + "require_preferred_account": require_preferred_account, + "preferred_account_id": target_session.account.id, + "excluded_account_ids": set(request_state.excluded_account_ids), + } + ) + target_session.account = replacement_account + target_session.upstream = replacement_upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_reconnect_http_bridge_session", fake_reconnect) + + assert await service._retry_http_bridge_precreated_request(session) is True + + assert reconnect_calls == [ + { + "require_same_account": False, + "require_preferred_account": False, + "preferred_account_id": "acct-silent", + "excluded_account_ids": {"acct-silent"}, + } + ] + assert request_state.preferred_account_id is None + assert session.account.id == "acct-replacement" + assert replacement_upstream.sent_text == [request_state.request_text] + + @pytest.mark.asyncio async def test_v1_responses_http_bridge_transient_refresh_failure_returns_upstream_error(async_client, monkeypatch): _install_bridge_settings(monkeypatch, enabled=True) @@ -13006,86 +13086,6 @@ async def fake_reconnect( assert replacement_upstream.sent_text == [retry_request.request_text] -@pytest.mark.asyncio -async def test_retry_account_neutral_precreated_request_switches_from_silent_account(app_instance, monkeypatch): - from app.modules.proxy.continuity import make_http_bridge_account_neutral_replay_key - - service = get_proxy_service_for_app(app_instance) - recovery_kind, recovery_key = make_http_bridge_account_neutral_replay_key("retry-silent-account") - first_account = cast(Account, SimpleNamespace(id="acct-silent", status=AccountStatus.ACTIVE, plan_type="plus")) - replacement_account = cast( - Account, - SimpleNamespace(id="acct-replacement", status=AccountStatus.ACTIVE, plan_type="plus"), - ) - replacement_upstream = _RecordingUpstreamWebSocket() - session = proxy_module._HTTPBridgeSession( - key=proxy_module._HTTPBridgeSessionKey(recovery_kind, recovery_key, None), - headers={"x-codex-turn-state": "stale-turn-state"}, - affinity=proxy_module._AffinityPolicy(), - request_model="gpt-5.5", - account=first_account, - upstream=cast(proxy_module.UpstreamWebSocket, _SilentUpstreamWebSocket()), - upstream_control=proxy_module._WebSocketUpstreamControl(), - pending_lock=anyio.Lock(), - pending_requests=deque(), - response_create_gate=asyncio.Semaphore(1), - queued_request_count=1, - last_used_at=time.monotonic(), - idle_ttl_seconds=120.0, - ) - request_state = proxy_module._WebSocketRequestState( - request_id="req-account-neutral-precreated-retry", - model="gpt-5.5", - service_tier=None, - reasoning_effort=None, - api_key_reservation=None, - started_at=time.monotonic(), - awaiting_response_created=True, - transport="http", - response_create_gate_acquired=True, - request_text=json.dumps({"type": "response.create", "model": "gpt-5.5", "input": []}), - ) - session.pending_requests.append(request_state) - reconnect_calls: list[dict[str, object]] = [] - - async def fake_reconnect( - self, - target_session, - *, - request_state, - restart_reader=False, - require_same_account=False, - require_preferred_account=False, - ): - del self, restart_reader - reconnect_calls.append( - { - "require_same_account": require_same_account, - "require_preferred_account": require_preferred_account, - "preferred_account_id": target_session.account.id, - "excluded_account_ids": set(request_state.excluded_account_ids), - } - ) - target_session.account = replacement_account - target_session.upstream = replacement_upstream - - monkeypatch.setattr(proxy_module.ProxyService, "_reconnect_http_bridge_session", fake_reconnect) - - assert await service._retry_http_bridge_precreated_request(session) is True - - assert reconnect_calls == [ - { - "require_same_account": False, - "require_preferred_account": False, - "preferred_account_id": "acct-silent", - "excluded_account_ids": {"acct-silent"}, - } - ] - assert request_state.preferred_account_id is None - assert session.account.id == "acct-replacement" - assert replacement_upstream.sent_text == [request_state.request_text] - - @pytest.mark.asyncio async def test_v1_responses_http_bridge_send_failure_returns_upstream_unavailable( async_client, diff --git a/tests/integration/test_proxy_compact.py b/tests/integration/test_proxy_compact.py index 7e63380e22..f0b5229d8a 100644 --- a/tests/integration/test_proxy_compact.py +++ b/tests/integration/test_proxy_compact.py @@ -460,68 +460,6 @@ async def fake_compact(payload, headers, access_token, account_id): ) -@pytest.mark.asyncio -async def test_proxy_compact_preserves_tool_search_pair_before_ordinary_tail(async_client, monkeypatch): - email = "compact-tool-search@example.com" - raw_account_id = "acc_compact_tool_search" - files = {"auth_json": ("auth.json", json.dumps(_make_auth_json(raw_account_id, email)), "application/json")} - response = await async_client.post("/api/accounts/import", files=files) - assert response.status_code == 200 - - seen_payloads: list[dict[str, object]] = [] - - async def fake_compact(payload, headers, access_token, account_id): - del headers, access_token, account_id - seen_payloads.append(cast(dict[str, object], payload.to_payload())) - return CompactResponsePayload.model_validate({"object": "response.compaction", "output": []}) - - monkeypatch.setattr(proxy_module, "core_compact_responses", fake_compact) - tool_call = { - "type": "tool_search_call", - "call_id": "call-search-tail", - "status": "completed", - "execution": "client", - "arguments": {"query": "codex-lb compaction tool search"}, - } - tool_output = { - "type": "tool_search_output", - "call_id": "call-search-tail", - "output": [{"title": "codex-lb compaction result"}], - } - ordinary_tail = {"role": "assistant", "content": "ordinary tail " + "x" * 500_000} - latest_request = {"role": "user", "content": "latest request"} - payload = { - "model": "gpt-5.6-sol", - "instructions": "hi", - "input": [ - {"role": "user", "content": "initial request"}, - {"role": "assistant", "content": "older answer " + "y" * 500_000}, - tool_call, - ordinary_tail, - tool_output, - latest_request, - ], - } - - response = await async_client.post("/backend-api/codex/responses/compact", json=payload) - - assert response.status_code == 200 - assert len(seen_payloads) == 1 - upstream_input = seen_payloads[0]["input"] - assert isinstance(upstream_input, list) - assert tool_call in upstream_input - assert tool_output in upstream_input - assert latest_request in upstream_input - assert all( - not ( - isinstance(item, dict) - and item.get("role") == "assistant" - and item.get("content") == [{"type": "output_text", "text": ordinary_tail["content"]}] - ) - for item in upstream_input - ) - - @pytest.mark.asyncio async def test_proxy_compact_omits_oversized_optional_tool_tail_before_upstream(async_client, monkeypatch): email = "compact-optional-tail@example.com" @@ -953,6 +891,68 @@ async def fake_compact(payload, headers, access_token, account_id): assert "resp_compact_missing" not in response.text +@pytest.mark.asyncio +async def test_proxy_compact_preserves_tool_search_pair_before_ordinary_tail(async_client, monkeypatch): + email = "compact-tool-search@example.com" + raw_account_id = "acc_compact_tool_search" + files = {"auth_json": ("auth.json", json.dumps(_make_auth_json(raw_account_id, email)), "application/json")} + response = await async_client.post("/api/accounts/import", files=files) + assert response.status_code == 200 + + seen_payloads: list[dict[str, object]] = [] + + async def fake_compact(payload, headers, access_token, account_id): + del headers, access_token, account_id + seen_payloads.append(cast(dict[str, object], payload.to_payload())) + return CompactResponsePayload.model_validate({"object": "response.compaction", "output": []}) + + monkeypatch.setattr(proxy_module, "core_compact_responses", fake_compact) + tool_call = { + "type": "tool_search_call", + "call_id": "call-search-tail", + "status": "completed", + "execution": "client", + "arguments": {"query": "codex-lb compaction tool search"}, + } + tool_output = { + "type": "tool_search_output", + "call_id": "call-search-tail", + "output": [{"title": "codex-lb compaction result"}], + } + ordinary_tail = {"role": "assistant", "content": "ordinary tail " + "x" * 500_000} + latest_request = {"role": "user", "content": "latest request"} + payload = { + "model": "gpt-5.6-sol", + "instructions": "hi", + "input": [ + {"role": "user", "content": "initial request"}, + {"role": "assistant", "content": "older answer " + "y" * 500_000}, + tool_call, + ordinary_tail, + tool_output, + latest_request, + ], + } + + response = await async_client.post("/backend-api/codex/responses/compact", json=payload) + + assert response.status_code == 200 + assert len(seen_payloads) == 1 + upstream_input = seen_payloads[0]["input"] + assert isinstance(upstream_input, list) + assert tool_call in upstream_input + assert tool_output in upstream_input + assert latest_request in upstream_input + assert all( + not ( + isinstance(item, dict) + and item.get("role") == "assistant" + and item.get("content") == [{"type": "output_text", "text": ordinary_tail["content"]}] + ) + for item in upstream_input + ) + + @pytest.mark.asyncio async def test_proxy_compact_headers_normalize_weekly_only_with_stale_secondary(async_client, monkeypatch): email = "compact-weekly@example.com" diff --git a/tests/unit/test_openai_requests.py b/tests/unit/test_openai_requests.py index 17761dbf9b..9a64159150 100644 --- a/tests/unit/test_openai_requests.py +++ b/tests/unit/test_openai_requests.py @@ -1074,6 +1074,42 @@ def test_compact_strips_tool_fields(): assert "text" not in dumped +def test_compact_trimming_keeps_tool_search_outputs_with_matching_calls(): + tool_call = { + "type": "tool_search_call", + "call_id": "call_search_tail", + "status": "completed", + "execution": "client", + "arguments": {"query": "spawn_agent multi-agent schema", "limit": 8}, + } + tool_output = { + "type": "tool_search_output", + "call_id": "call_search_tail", + "output": "Found matching tools", + } + input_items = [ + {"role": "user", "content": "initial instructions"}, + {"role": "assistant", "content": "x" * 500_000}, + tool_call, + {"role": "assistant", "content": "y" * 500_000}, + tool_output, + {"role": "user", "content": "latest request"}, + ] + payload = { + "model": "gpt-5.1", + "instructions": "hi", + "input": input_items, + } + + request = ResponsesCompactRequest.model_validate(payload) + dumped = request.to_payload() + dumped_input = dumped["input"] + + assert isinstance(dumped_input, list) + assert tool_call in dumped_input + assert tool_output in dumped_input + + def test_responses_strips_poisoned_local_compact_fallback_items(): poisoned_message = { "type": "message", @@ -2769,42 +2805,6 @@ def test_compact_trimming_keeps_selected_tool_calls_with_matching_outputs(): assert tool_output in dumped_input -def test_compact_trimming_keeps_tool_search_outputs_with_matching_calls(): - tool_call = { - "type": "tool_search_call", - "call_id": "call_search_tail", - "status": "completed", - "execution": "client", - "arguments": {"query": "spawn_agent multi-agent schema", "limit": 8}, - } - tool_output = { - "type": "tool_search_output", - "call_id": "call_search_tail", - "output": "Found matching tools", - } - input_items = [ - {"role": "user", "content": "initial instructions"}, - {"role": "assistant", "content": "x" * 500_000}, - tool_call, - {"role": "assistant", "content": "y" * 500_000}, - tool_output, - {"role": "user", "content": "latest request"}, - ] - payload = { - "model": "gpt-5.1", - "instructions": "hi", - "input": input_items, - } - - request = ResponsesCompactRequest.model_validate(payload) - dumped = request.to_payload() - dumped_input = dumped["input"] - - assert isinstance(dumped_input, list) - assert tool_call in dumped_input - assert tool_output in dumped_input - - def test_compact_trimming_reconciles_duplicate_tool_call_ids_by_occurrence(): first_tool_call = { "type": "function_call", diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index d9cfb0169c..0a82ae951d 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -19415,6 +19415,76 @@ async def test_select_websocket_connect_account_stream_cap_is_local_overload(mon assert sent_payload["error"]["type"] == "rate_limit_error" +def test_websocket_client_previous_response_full_resend_retry_allows_tool_search_history() -> None: + self_contained_tool_search_history: list[JsonValue] = [ + {"role": "user", "content": [{"type": "input_text", "text": "search tools"}]}, + { + "type": "tool_search_call", + "call_id": "call_search", + "arguments": {"query": "mcp tools"}, + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search", + "output": [{"title": "tool_search"}], + "status": "completed", + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + assert ( + proxy_service._websocket_client_previous_response_full_resend_is_retry_safe( + previous_response_id="resp_client_anchor", + input_value=self_contained_tool_search_history, + continuity_state=None, + ) + is True + ) + + +def test_trim_http_bridge_previous_response_input_items_handles_tool_search_replay(): + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "id": "tsc_replay", + "call_id": "call_search_1", + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search_1", + "output": [{"title": "result"}], + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + trimmed = proxy_service._trim_http_bridge_previous_response_input_items(input_items) + + assert trimmed == input_items[1:] + + +def test_trim_websocket_previous_response_input_items_handles_tool_search_replay(): + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "id": "tsc_replay", + "call_id": "call_search_1", + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search_1", + "output": [{"title": "result"}], + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + trimmed = proxy_service._trim_websocket_previous_response_input_items(input_items) + + assert trimmed == input_items[1:] + + @pytest.mark.asyncio @pytest.mark.parametrize( ("error_code", "error_message"), @@ -21155,34 +21225,6 @@ def test_websocket_client_previous_response_full_resend_retry_allows_self_contai ) -def test_websocket_client_previous_response_full_resend_retry_allows_tool_search_history() -> None: - self_contained_tool_search_history: list[JsonValue] = [ - {"role": "user", "content": [{"type": "input_text", "text": "search tools"}]}, - { - "type": "tool_search_call", - "call_id": "call_search", - "arguments": {"query": "mcp tools"}, - "status": "completed", - }, - { - "type": "tool_search_output", - "call_id": "call_search", - "output": [{"title": "tool_search"}], - "status": "completed", - }, - {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, - ] - - assert ( - proxy_service._websocket_client_previous_response_full_resend_is_retry_safe( - previous_response_id="resp_client_anchor", - input_value=self_contained_tool_search_history, - continuity_state=None, - ) - is True - ) - - @pytest.mark.asyncio async def test_prepare_websocket_response_create_request_fills_interrupted_pending_tool_outputs(monkeypatch): request_logs = _RequestLogsRecorder() @@ -40329,48 +40371,6 @@ def test_trim_websocket_previous_response_input_items_handles_apply_patch_replay assert trimmed == input_items[2:] -def test_trim_http_bridge_previous_response_input_items_handles_tool_search_replay(): - input_items: list[JsonValue] = [ - { - "type": "tool_search_call", - "id": "tsc_replay", - "call_id": "call_search_1", - "status": "completed", - }, - { - "type": "tool_search_output", - "call_id": "call_search_1", - "output": [{"title": "result"}], - }, - {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, - ] - - trimmed = proxy_service._trim_http_bridge_previous_response_input_items(input_items) - - assert trimmed == input_items[1:] - - -def test_trim_websocket_previous_response_input_items_handles_tool_search_replay(): - input_items: list[JsonValue] = [ - { - "type": "tool_search_call", - "id": "tsc_replay", - "call_id": "call_search_1", - "status": "completed", - }, - { - "type": "tool_search_output", - "call_id": "call_search_1", - "output": [{"title": "result"}], - }, - {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, - ] - - trimmed = proxy_service._trim_websocket_previous_response_input_items(input_items) - - assert trimmed == input_items[1:] - - def test_prepare_response_bridge_request_state_keeps_unconfirmed_missing_tool_output_history(): request_logs = _RequestLogsRecorder() service = proxy_service.ProxyService(_repo_factory(request_logs)) diff --git a/tests/unit/test_replay_safety.py b/tests/unit/test_replay_safety.py index 8330eb3879..f815957c6e 100644 --- a/tests/unit/test_replay_safety.py +++ b/tests/unit/test_replay_safety.py @@ -152,51 +152,6 @@ def test_account_neutral_fresh_replay_accepts_self_contained_payloads( assert responses_payload_is_account_neutral_fresh_replay(payload) is True -def test_account_neutral_fresh_replay_accepts_compaction_context_item() -> None: - payload: dict[str, JsonValue] = { - "input": [ - { - "type": "compaction", - "status": "completed", - "encrypted_content": "encrypted-compact-context", - }, - { - "type": "message", - "role": "user", - "content": [{"type": "input_text", "text": "continue"}], - }, - ], - } - - assert responses_payload_is_account_neutral_fresh_replay(payload) is True - - -def test_account_neutral_replay_projection_preserves_compaction_without_response_id() -> None: - input_items: list[JsonValue] = [ - { - "type": "compaction", - "id": "cmp_owner_a", - "status": "completed", - "encrypted_content": "encrypted-compact-context", - }, - { - "type": "message", - "role": "user", - "content": [{"type": "input_text", "text": "continue"}], - }, - ] - - projection = project_responses_input_for_account_neutral_fresh_replay(input_items, stored_count=1) - - assert projection is not None - assert projection.input_items[0] == { - "type": "compaction", - "status": "completed", - "encrypted_content": "encrypted-compact-context", - } - assert responses_payload_is_account_neutral_fresh_replay({"input": projection.input_items}) is True - - def test_account_neutral_replay_projection_removes_response_owned_bookkeeping() -> None: metadata = {"turn_id": "turn_owner_a"} input_items: list[JsonValue] = [ @@ -374,27 +329,6 @@ def test_account_neutral_replay_projection_preserves_noncompleted_search_state_t assert responses_payload_is_account_neutral_fresh_replay({"input": projection.input_items}) is False -def test_account_neutral_fresh_replay_accepts_self_contained_tool_search_pair() -> None: - input_items: list[JsonValue] = [ - { - "type": "tool_search_call", - "call_id": "call_search", - "arguments": {"query": "codex-lb"}, - "status": "completed", - }, - { - "type": "tool_search_output", - "call_id": "call_search", - "output": "Found codex-lb", - "status": "completed", - }, - {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, - ] - - assert responses_input_items_are_self_contained_fresh_replay(input_items) is True - assert responses_payload_is_account_neutral_fresh_replay({"input": input_items}) is True - - @pytest.mark.parametrize( "suffix", [ @@ -782,6 +716,46 @@ def test_full_resend_suffix_accepts_only_self_contained_tool_loops( ) +def test_account_neutral_fresh_replay_accepts_compaction_context_item() -> None: + payload: dict[str, JsonValue] = { + "input": [ + { + "type": "compaction", + "status": "completed", + "encrypted_content": "encrypted-compact-context", + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "continue"}], + }, + ], + } + + assert responses_payload_is_account_neutral_fresh_replay(payload) is True + + +def test_account_neutral_fresh_replay_accepts_self_contained_tool_search_pair() -> None: + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "call_id": "call_search", + "arguments": {"query": "codex-lb"}, + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search", + "output": "Found codex-lb", + "status": "completed", + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + assert responses_input_items_are_self_contained_fresh_replay(input_items) is True + assert responses_payload_is_account_neutral_fresh_replay({"input": input_items}) is True + + def test_full_resend_tool_loop_manifest_tolerates_fresh_developer_interleave_after_historical_one() -> None: stored_input: list[JsonValue] = [ {"role": "user", "content": "first question"}, From db7ce90fb483f22b06149242bec3e53f07f2dfc1 Mon Sep 17 00:00:00 2001 From: Darafei Praliaskouski Date: Sat, 15 Aug 2026 01:36:18 +0400 Subject: [PATCH 7/7] fix(proxy): reject owner-bound replay state --- .../_service/http_bridge/request_submit.py | 5 ++- .../proxy/_service/websocket/helpers.py | 2 ++ app/modules/proxy/replay_safety.py | 5 ++- openspec/specs/responses-api-compat/spec.md | 8 ++++- tests/unit/test_proxy_http_bridge.py | 2 +- tests/unit/test_proxy_utils.py | 32 ++++++++++++++++++- tests/unit/test_replay_safety.py | 28 ++++++++++++++-- 7 files changed, 72 insertions(+), 10 deletions(-) diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 7f6b11c125..23216a8c36 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -3020,13 +3020,12 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool: # Account-scoped uploaded files cannot be replayed on a # different owner. Keep the preferred account mandatory for # both silent recovery and clean-close recovery. - require_preferred_reconnect = request_state.file_required_preferred_account + require_preferred_reconnect = account_neutral_recovery or request_state.file_required_preferred_account request_text = _prepare_websocket_request_state_for_visible_output_replay(request_state) if request_text is None: return False if account_neutral_recovery: - request_state.preferred_account_id = None - request_state.excluded_account_ids.add(session.account.id) + request_state.preferred_account_id = session.account.id elif not request_state.file_required_preferred_account: if hard_owner_bound and not model_fallback_replay and not fresh_hard_request_account_switch_allowed: request_state.preferred_account_id = session.account.id diff --git a/app/modules/proxy/_service/websocket/helpers.py b/app/modules/proxy/_service/websocket/helpers.py index fa3669fb19..0676ff084a 100644 --- a/app/modules/proxy/_service/websocket/helpers.py +++ b/app/modules/proxy/_service/websocket/helpers.py @@ -536,6 +536,8 @@ def _websocket_input_items_are_self_contained_fresh_replay(input_items: list[Jso call_id_value = item.get("call_id") call_id = call_id_value if isinstance(call_id_value, str) and call_id_value else None if item_type in _WEBSOCKET_TOOL_CALL_ITEM_TYPES: + if item_type == "tool_search_call" and item.get("execution") not in (None, "client"): + return False if call_id is not None: seen_call_ids_by_type[item_type].add(call_id) continue diff --git a/app/modules/proxy/replay_safety.py b/app/modules/proxy/replay_safety.py index e54c68b23f..970c118190 100644 --- a/app/modules/proxy/replay_safety.py +++ b/app/modules/proxy/replay_safety.py @@ -669,7 +669,7 @@ def _tool_call_is_self_contained(item_type: str, item: Mapping[str, JsonValue]) def _compaction_item_is_self_contained(item: Mapping[str, JsonValue]) -> bool: - return item.get("status") in (None, "completed") and _is_nonblank_string(item.get("encrypted_content")) + return False def _caller_is_self_contained(item: Mapping[str, JsonValue]) -> bool: @@ -709,6 +709,9 @@ def _apply_patch_operation_is_self_contained(operation: JsonValue | None) -> boo def _tool_output_is_self_contained(item_type: str, item: Mapping[str, JsonValue]) -> bool: if item.get("status") not in (None, "completed", "failed"): return False + if item_type == "tool_search_output": + tools = item.get("tools") + return isinstance(tools, list) and all(isinstance(tool, dict) for tool in tools) output = item.get("output") if isinstance(output, str): return True diff --git a/openspec/specs/responses-api-compat/spec.md b/openspec/specs/responses-api-compat/spec.md index 450c903864..e4cae4b417 100644 --- a/openspec/specs/responses-api-compat/spec.md +++ b/openspec/specs/responses-api-compat/spec.md @@ -426,6 +426,13 @@ When a direct WebSocket `response.create` request includes both `previous_respon - **THEN** the service MUST NOT replay that payload as a fresh turn without `previous_response_id` - **AND** the downstream client receives a retryable continuity failure rather than a fabricated fresh turn +#### Scenario: tool-search and compaction state stay owner-safe during fresh replay +- **WHEN** HTTP bridge or WebSocket recovery evaluates a full resend as account-neutral fresh replay +- **THEN** `tool_search_call` items are replayable only when their `execution` is absent or `client` +- **AND** matching `tool_search_output` items are validated using the canonical `tools` list +- **AND** encrypted `compaction` items or any remaining encrypted content keep the resend owner-bound and fail closed for cross-account replay +- **AND** reconnects for an established account-neutral recovery lane retain that lane's current account + ### Requirement: Public Responses errors mask previous-response misses Public Responses endpoints MUST NOT return an OpenAI-shaped `previous_response_not_found` error to clients. If a lower layer still raises or collects that error, the API layer MUST rewrite it to a retryable `stream_incomplete` continuity failure and remove the missing response id from the public payload. @@ -5112,4 +5119,3 @@ only an inactive `unknown` operation may enter a fresh recovery attempt. - **WHEN** a duplicate request finds a submitted operation still referenced by another pending request - **THEN** the proxy refuses a second dispatch and preserves the existing spool - diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 1ede9d331b..55d1f1ac31 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -21853,7 +21853,7 @@ async def fake_stream_events( assert third_call.kwargs["preferred_account_id"] is None assert third_call.kwargs["durable_lookup"] is None # When the fresh-replay session's own gate later times out (session.account - # is "acc-fallback"), the next replacement must also exclude it — a + # is "acc-fallback"), the next replacement must also exclude it -- a # "replacement" that could legally reselect the account that just proved # stuck isn't a replacement at all. assert third_call.kwargs["exclude_account_ids"] == ( diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index 0a82ae951d..bfb5fa662e 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -19422,12 +19422,13 @@ def test_websocket_client_previous_response_full_resend_retry_allows_tool_search "type": "tool_search_call", "call_id": "call_search", "arguments": {"query": "mcp tools"}, + "execution": "client", "status": "completed", }, { "type": "tool_search_output", "call_id": "call_search", - "output": [{"title": "tool_search"}], + "tools": [{"name": "tool_search"}], "status": "completed", }, {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, @@ -19443,6 +19444,35 @@ def test_websocket_client_previous_response_full_resend_retry_allows_tool_search ) +def test_websocket_client_previous_response_full_resend_retry_rejects_server_tool_search_history() -> None: + server_tool_search_history: list[JsonValue] = [ + {"role": "user", "content": [{"type": "input_text", "text": "search tools"}]}, + { + "type": "tool_search_call", + "call_id": "call_search", + "arguments": {"query": "mcp tools"}, + "execution": "server", + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search", + "tools": [{"name": "tool_search"}], + "status": "completed", + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + assert ( + proxy_service._websocket_client_previous_response_full_resend_is_retry_safe( + previous_response_id="resp_server_anchor", + input_value=server_tool_search_history, + continuity_state=None, + ) + is False + ) + + def test_trim_http_bridge_previous_response_input_items_handles_tool_search_replay(): input_items: list[JsonValue] = [ { diff --git a/tests/unit/test_replay_safety.py b/tests/unit/test_replay_safety.py index f815957c6e..ad37ff46cb 100644 --- a/tests/unit/test_replay_safety.py +++ b/tests/unit/test_replay_safety.py @@ -716,7 +716,7 @@ def test_full_resend_suffix_accepts_only_self_contained_tool_loops( ) -def test_account_neutral_fresh_replay_accepts_compaction_context_item() -> None: +def test_account_neutral_fresh_replay_rejects_encrypted_compaction_context_item() -> None: payload: dict[str, JsonValue] = { "input": [ { @@ -732,7 +732,7 @@ def test_account_neutral_fresh_replay_accepts_compaction_context_item() -> None: ], } - assert responses_payload_is_account_neutral_fresh_replay(payload) is True + assert responses_payload_is_account_neutral_fresh_replay(payload) is False def test_account_neutral_fresh_replay_accepts_self_contained_tool_search_pair() -> None: @@ -746,7 +746,7 @@ def test_account_neutral_fresh_replay_accepts_self_contained_tool_search_pair() { "type": "tool_search_output", "call_id": "call_search", - "output": "Found codex-lb", + "tools": [{"name": "read_docs", "description": "Read docs"}], "status": "completed", }, {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, @@ -756,6 +756,28 @@ def test_account_neutral_fresh_replay_accepts_self_contained_tool_search_pair() assert responses_payload_is_account_neutral_fresh_replay({"input": input_items}) is True +def test_account_neutral_fresh_replay_rejects_server_tool_search_pair() -> None: + input_items: list[JsonValue] = [ + { + "type": "tool_search_call", + "call_id": "call_search", + "arguments": {"query": "codex-lb"}, + "execution": "server", + "status": "completed", + }, + { + "type": "tool_search_output", + "call_id": "call_search", + "tools": [{"name": "read_docs", "description": "Read docs"}], + "status": "completed", + }, + {"role": "user", "content": [{"type": "input_text", "text": "continue"}]}, + ] + + assert responses_input_items_are_self_contained_fresh_replay(input_items) is False + assert responses_payload_is_account_neutral_fresh_replay({"input": input_items}) is False + + def test_full_resend_tool_loop_manifest_tolerates_fresh_developer_interleave_after_historical_one() -> None: stored_input: list[JsonValue] = [ {"role": "user", "content": "first question"},