diff --git a/app/core/openai/requests.py b/app/core/openai/requests.py index b52d76fe55..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) @@ -864,6 +865,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 +906,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 +917,66 @@ 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): + return None + + text = _compaction_plaintext_summary(item) + if text is None: + return _assistant_compact_state_item("[unverified compact state omitted]") + return _assistant_compact_state_item(text) + + +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: + 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/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..0676ff084a 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()) @@ -535,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 @@ -1881,7 +1884,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 +1900,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..970c118190 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 False + + def _caller_is_self_contained(item: Mapping[str, JsonValue]) -> bool: caller = item.get("caller") return caller is None or caller == {"type": "direct"} @@ -677,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 @@ -1017,6 +1052,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/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/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/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index d4b3a0fda9..58eaf0f1f0 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -6873,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) diff --git a/tests/integration/test_proxy_compact.py b/tests/integration/test_proxy_compact.py index bb6b2fb151..f0b5229d8a 100644 --- a/tests/integration/test_proxy_compact.py +++ b/tests/integration/test_proxy_compact.py @@ -336,6 +336,69 @@ 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": [ + { + "type": "output_text", + "text": "[compact state] 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"}, + ] + + @pytest.mark.asyncio async def test_proxy_compact_preserves_historical_code_mode_side_effect_pair_before_ordinary_tail( async_client, monkeypatch @@ -828,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 8a883ad4be..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", @@ -1130,6 +1166,122 @@ def test_compact_strips_poisoned_local_compact_fallback_items(): assert request.to_payload()["input"] == [{"role": "user", "content": "continue"}] +def test_compact_preserves_opaque_encrypted_compaction_replay(): + 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"] == payload["input"] + + +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": [ + { + "type": "output_text", + "text": "[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_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 = { 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 9ed5b4d98f..bfb5fa662e 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -19415,6 +19415,106 @@ 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"}, + "execution": "client", + "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_client_anchor", + input_value=self_contained_tool_search_history, + continuity_state=None, + ) + is True + ) + + +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] = [ + { + "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"), diff --git a/tests/unit/test_replay_safety.py b/tests/unit/test_replay_safety.py index e17cf26c36..ad37ff46cb 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, @@ -715,6 +716,68 @@ def test_full_resend_suffix_accepts_only_self_contained_tool_loops( ) +def test_account_neutral_fresh_replay_rejects_encrypted_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 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", + "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 True + 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"},