fix(codex): use Responses API format for non-streaming translation so usage is preserved

- Change executor to_domain_response call from 'openai' to 'openai-responses'
  so Responses API fields (input_tokens/output_tokens) are properly normalized
- Add input_tokens/output_tokens fallback mapping in UsageSummary.from_dict
- Compute total_tokens when missing from input+output
- Update test assertion for new format parameter
- Update demo script to validate both streaming and legacy frontend paths

Author: Mateusz

dev/scripts/demo_codex_usage_reporting_fix.py

@@ -69,7 +69,7 @@ async def _cancel() -> None:
         )
 
 
-class _FakeTransportWithoutProviderUsage:
+class _FakeTransportWithStreamingUsage:
     async def initiate_streaming_request(
         self,
         url: str,
@@ -82,12 +82,27 @@ async def _iterator() -> AsyncIterator[ProcessedResponse]:
                 content={
                     "choices": [
                         {
-                            "delta": {"content": "This should carry token usage."},
-                            "finish_reason": "stop",
+                            "delta": {"content": "Hello from Codex streaming. "},
+                            "finish_reason": None,
                         }
                     ]
                 }
             )
+            yield ProcessedResponse(
+                content={
+                    "choices": [
+                        {
+                            "delta": {"content": "This carries token usage."},
+                            "finish_reason": "stop",
+                        }
+                    ],
+                    "usage": {
+                        "input_tokens": 17,
+                        "output_tokens": 9,
+                        "total_tokens": 26,
+                    },
+                }
+            )
 
         async def _cancel() -> None:
             return None
@@ -193,7 +208,7 @@ async def _run_demo() -> None:
                         stream=True,
                     )
 
-                    response_executor._transport = _FakeTransportWithoutProviderUsage()  # type: ignore[attr-defined]
+                    response_executor._transport = _FakeTransportWithStreamingUsage()  # type: ignore[attr-defined]
 
                     stream_result = await backend.chat_completions(
                         request_data=streaming_request,
@@ -218,15 +233,24 @@ async def _run_demo() -> None:
 
                     usage_payloads: list[dict[str, Any]] = []
                     async for chunk in stream_content:
+                        # Check explicit usage field
                         if chunk.usage is not None:
                             usage_payloads.append(chunk.usage.model_dump())
+                        # Check metadata
                         usage_metadata = chunk.metadata.get("usage")
                         if isinstance(usage_metadata, dict):
                             usage_payloads.append(dict(usage_metadata))
+                        # Check content dict (where Codex SSE would embed usage)
+                        if isinstance(chunk.content, dict):
+                            content_usage = chunk.content.get("usage")
+                            if isinstance(content_usage, dict):
+                                usage_payloads.append(dict(content_usage))
 
                     print("[stream] usage payloads:", usage_payloads)
                     if not usage_payloads:
-                        raise RuntimeError("Streaming usage is missing")
+                        raise RuntimeError(
+                            "Streaming usage is missing; no chunks carried usage data"
+                        )
 
                     max_total = max(
                         int(p.get("total_tokens", 0)) for p in usage_payloads

src/connectors/openai_codex/executor.py

@@ -583,10 +583,11 @@ async def _execute_non_streaming(
                     )
 
             # Parse response using translation service with renderer override
+            # Codex uses OpenAI Responses API format which has 'output' not 'choices'
             with OverrideRenderer(renderer_key):
                 domain_response = (
                     self._base_connector.translation_service.to_domain_response(
-                        response_json, "openai"
+                        response_json, "openai-responses"
                     )
                 )
 

src/core/domain/usage_summary.py

@@ -49,8 +49,15 @@ def from_dict(cls, data: dict[str, Any]) -> UsageSummary:
             UsageSummary instance
         """
         prompt_tokens = data.get("prompt_tokens")
+        if not isinstance(prompt_tokens, int):
+            prompt_tokens = data.get("input_tokens")
         completion_tokens = data.get("completion_tokens")
+        if not isinstance(completion_tokens, int):
+            completion_tokens = data.get("output_tokens")
         total_tokens = data.get("total_tokens")
+        if not isinstance(total_tokens, int):
+            computed = (prompt_tokens or 0) + (completion_tokens or 0)
+            total_tokens = computed if computed > 0 else None
 
         # Extract extensions
         # If "extensions" key exists, use it directly; otherwise extract all non-standard fields

tests/unit/connectors/openai_codex/test_executor.py

@@ -391,7 +391,7 @@ async def post_side_effect(*args, **kwargs):
         compatibility_layer.detect_incompatible_tool_calls.assert_called()
         compatibility_layer.append_incompatible_tool_steering.assert_called_once()
         mock_base_connector.translation_service.to_domain_response.assert_called_once_with(
-            second_response.json.return_value, "openai"
+            second_response.json.return_value, "openai-responses"
         )
 
     @pytest.mark.asyncio