fix(compact): sanitize plaintext compaction replays

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,25 +906,77 @@ 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):
             skip_next_poison_compaction = True
             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

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

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

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"))

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
 

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]`

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.

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.

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

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.

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
-