fix(codex-websocket): normalize Responses-native tool events into canonical chunks

Preserve full response.output_item.done payloads in OpenAIWebSocketClient
and add normalization in ResponseExecutor so websocket-originated Codex
events flow through the Responses translator before retry/visibility
logic. Also normalize response.done into synthetic response.completed
so websocket streams end with consistent canonical stop chunks.

Delivers failing tests first proving websocket tool-call completion
and response.done normalization end-to-end.

Author: Mateusz

README.md

@@ -59,7 +59,7 @@ The proxy includes built-in resilience features for production use:
 - **Streaming protection** - Avoids retry after output has started, preventing corruption
 - **Health monitoring** - Tracks backend availability and performance
 
-Configure via the `resilience` section in `config.yaml` or see the [Failure Handling Guide](docs/user_guide/features/failure-handling.md). Request processing now runs through a single canonical manager path with no legacy split-handler fallback. `request_processing_unification` remains for operational controls such as per-backend `connector_stream_first` hints, optional `emit_path_selection_metadata`, and `promotion_requirements` reporting. When path diagnostics are enabled, `promotion_guardrails` uses strict missing-evidence semantics (absent measurements do not read as promotion-ready).
+Configure via the `resilience` section in `config.yaml` or see the [Failure Handling Guide](docs/user_guide/features/failure-handling.md). Request processing now runs through a single canonical manager path with no legacy split-handler fallback. `canonical_request_processing` provides the remaining runtime controls such as empty-stream recovery tuning.
 
 ## Quick Start
 

config/config.example.yaml

@@ -263,20 +263,10 @@ dynamic_compression:
   diff_max_lines_per_hunk: 100
   diff_max_total_lines: 500
 
-# Request-processing unification (canonical path is always used at runtime)
-request_processing_unification:
-  enable_core_canonical_path: true
-  enable_canonical_features: false
-  connector_stream_first: {}
-  retire_legacy_dual_path: false
-  emit_path_selection_metadata: false
-  promotion_requirements:
-    require_characterization_tests: true
-    require_equivalence_tests: true
-    max_non_stream_p95_latency_delta_pct: 10.0
-    max_stream_ttft_delta_pct: 10.0
-    max_memory_delta_pct: 10.0
-    require_cleanup_checks: true
+# Canonical request-processing runtime controls
+canonical_request_processing:
+  empty_stream_recovery_prompt: "The previous response was empty, please try again."
+  max_empty_stream_retries: 1
 
 # Logging
 logging:

config/schemas/app_config.schema.yaml

@@ -634,18 +634,10 @@ properties:
       capability_refresh_backoff_seconds:
         type: number
         minimum: 0
-  request_processing_unification:
+  canonical_request_processing:
     type: object
     additionalProperties: false
     properties:
-      enable_core_canonical_path: { type: boolean }
-      enable_canonical_features: { type: boolean }
-      connector_stream_first:
-        type: object
-        additionalProperties: { type: boolean }
-      retire_legacy_dual_path: { type: boolean }
-      emit_path_selection_metadata: { type: boolean }
-      legacy_streaming_client_blocking_envelope: { type: boolean }
       empty_stream_recovery_prompt:
         type: string
         description: "Recovery prompt appended to retry requests when stream produces no content"
@@ -654,9 +646,6 @@ properties:
         minimum: 0
         maximum: 5
         description: "Maximum number of empty stream retry attempts before failing"
-      promotion_requirements:
-        type: object
-        additionalProperties: false
         properties:
           require_characterization_tests: { type: boolean }
           require_equivalence_tests: { type: boolean }

src/connectors/openai_codex/executor.py

@@ -846,6 +846,9 @@ async def _streaming_iterator() -> AsyncIterator[ProcessedResponse]:
                     visible_output_emitted = False
                     with OverrideRenderer(renderer_key):
                         async for processed_chunk in stream_handle.iterator:
+                            processed_chunk = self._normalize_processed_stream_chunk(
+                                processed_chunk
+                            )
                             incompatible_tools = self._detect_incompatible_tool_calls(
                                 processed_chunk.content,
                                 context,
@@ -1164,6 +1167,53 @@ def _append_incompatible_tool_retry_steering(
         )
         return dict(adapted)
 
+    @staticmethod
+    def _coerce_stream_chunk_content(content: object) -> dict[str, Any] | None:
+        model_dump = getattr(content, "model_dump", None)
+        if callable(model_dump):
+            dumped = model_dump(exclude_none=True)
+            if isinstance(dumped, dict):
+                return cast(dict[str, Any], dumped)
+        if isinstance(content, Mapping):
+            return dict(content)
+        return None
+
+    def _normalize_processed_stream_chunk(
+        self, chunk: ProcessedResponse
+    ) -> ProcessedResponse:
+        metadata = chunk.metadata
+        event_type = metadata.get("event_type")
+        if not isinstance(event_type, str):
+            return chunk
+
+        content_dict = self._coerce_stream_chunk_content(chunk.content)
+        if not content_dict:
+            return chunk
+
+        if event_type == "response.done":
+            content_dict = {"type": "response.completed", "response": content_dict}
+        elif "choices" in content_dict or not str(
+            content_dict.get("type") or ""
+        ).startswith("response."):
+            return chunk
+
+        translation_service = getattr(self._base_connector, "translation_service", None)
+        if translation_service is None:
+            return chunk
+
+        translated = translation_service.to_domain_stream_chunk(
+            content_dict, "responses"
+        )
+        translated_content = self._coerce_stream_chunk_content(translated)
+        if translated_content is None:
+            return chunk
+
+        return ProcessedResponse(
+            content=translated_content,
+            usage=chunk.usage,
+            metadata=dict(metadata),
+        )
+
     @staticmethod
     def _extract_tool_calls(response_like: object) -> list[dict[str, object]]:
         if isinstance(response_like, Mapping):

src/connectors/openai_websocket_client.py

@@ -343,13 +343,14 @@ def _event_to_processed_response(
                     metadata={"event_type": event_type},
                 )
 
-        # Handle response.output_item.done
-        if event_type == "response.output_item.done":
-            output_item = event_data.get("item", {})
-            return ProcessedResponse(
-                content={"type": "output_item.done", "item": output_item},
-                metadata={"event_type": event_type},
-            )
+        # Preserve full Responses-native payloads for tool-call completion events.
+        # Downstream Codex translation needs fields like output_index and the exact
+        # top-level event type to reconstruct canonical tool-call chunks.
+        if event_type == "response.output_item.done":
+            return ProcessedResponse(
+                content=event_data,
+                metadata={"event_type": event_type},
+            )
 
         # Handle response.done
         if event_type == "response.done":

src/core/config/app_config.py

@@ -14,15 +14,14 @@
     BackendConfig,
     BackendSettings,
     BruteForceProtectionConfig,
+    CanonicalRequestProcessingConfig,
     CodebuffConfig,
     EditPrecisionConfig,
     EmptyResponseConfig,
     LoggingConfig,
     LogLevel,
     ModelAliasRule,
     PlanningPhaseConfig,
-    RequestProcessingPromotionRequirementsConfig,
-    RequestProcessingUnificationConfig,
     RewritingConfig,
     RoutingConfig,
     SessionConfig,
@@ -113,7 +112,7 @@ def save(self, path: str | Path) -> None:
             "failure_handling",
             "routing",
             "dynamic_compression",
-            "request_processing_unification",
+            "canonical_request_processing",
             "reasoning_model_token_floor",
             "memory",
             "database",
@@ -239,8 +238,7 @@ def load_config(
     "ParameterResolution",
     "ParameterSource",
     "PlanningPhaseConfig",
-    "RequestProcessingPromotionRequirementsConfig",
-    "RequestProcessingUnificationConfig",
+    "CanonicalRequestProcessingConfig",
     "RewritingConfig",
     "RoutingConfig",
     "SessionConfig",

src/core/config/models/__init__.py

@@ -8,6 +8,9 @@
     BackendSettings,
     get_openrouter_headers,
 )
+from src.core.config.models.canonical_request_processing import (
+    CanonicalRequestProcessingConfig,
+)
 from src.core.config.models.end_of_session import EndOfSessionConfig
 from src.core.config.models.logging import LoggingConfig, LogLevel
 from src.core.config.models.misc import (
@@ -19,10 +22,6 @@
 )
 from src.core.config.models.non_forwardable_config import NonForwardableTaggingConfig
 from src.core.config.models.notification import NotificationConfig
-from src.core.config.models.request_processing_unification import (
-    RequestProcessingPromotionRequirementsConfig,
-    RequestProcessingUnificationConfig,
-)
 from src.core.config.models.rewriting import (
     EditPrecisionConfig,
     ModelAliasRule,
@@ -56,8 +55,7 @@
     "ModelAliasRule",
     "NotificationConfig",
     "PlanningPhaseConfig",
-    "RequestProcessingPromotionRequirementsConfig",
-    "RequestProcessingUnificationConfig",
+    "CanonicalRequestProcessingConfig",
     "ReasoningModelTokenFloorConfig",
     "RewritingConfig",
     "ResilienceConfig",

src/core/config/models/app_config_model.py

@@ -11,6 +11,9 @@
 from src.core.config.models.auth import AuthConfig
 from src.core.config.models.auxiliary_routing import AuxiliaryRoutingConfig
 from src.core.config.models.backends import BackendSettings
+from src.core.config.models.canonical_request_processing import (
+    CanonicalRequestProcessingConfig,
+)
 from src.core.config.models.end_of_session import EndOfSessionConfig
 from src.core.config.models.logging import LoggingConfig
 from src.core.config.models.misc import (
@@ -24,9 +27,6 @@
 )
 from src.core.config.models.non_forwardable_config import NonForwardableTaggingConfig
 from src.core.config.models.notification import NotificationConfig
-from src.core.config.models.request_processing_unification import (
-    RequestProcessingUnificationConfig,
-)
 from src.core.config.models.rewriting import (
     EditPrecisionConfig,
     ModelAliasRule,
@@ -120,8 +120,8 @@ class AppConfigModel(DomainModel, IConfig):
     auxiliary_routing: AuxiliaryRoutingConfig = Field(
         default_factory=AuxiliaryRoutingConfig
     )
-    request_processing_unification: RequestProcessingUnificationConfig = Field(
-        default_factory=RequestProcessingUnificationConfig
+    canonical_request_processing: CanonicalRequestProcessingConfig = Field(
+        default_factory=CanonicalRequestProcessingConfig
     )
     compaction: CompactionConfig = Field(default_factory=CompactionConfig)
     dynamic_compression: DynamicCompressionConfig = Field(

src/core/config/models/canonical_request_processing.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from pydantic import Field
+
+from src.core.interfaces.model_bases import DomainModel
+
+
+class CanonicalRequestProcessingConfig(DomainModel):
+    """Runtime settings for the canonical request-processing pipeline."""
+
+    # Empty stream recovery tuning (operational flexibility)
+    empty_stream_recovery_prompt: str = Field(
+        default="The previous response was empty, please try again.",
+        description="Recovery prompt appended to retry requests when stream produces no content",
+    )
+    max_empty_stream_retries: int = Field(
+        default=1,
+        ge=0,
+        le=5,
+        description="Maximum number of empty stream retry attempts before failing",
+    )