test(openai_agents): cover real usage capture, span emission, and prompt_cache_key

Adds five tests in TestStreamingModelUsageAndCacheKey covering the three
changes in the prior commit:

- Usage is captured from ResponseCompletedEvent.response.usage.
- Usage falls back to zeros when the stream ends without a completed event.
- The streaming span's output_data carries the usage block.
- prompt_cache_key defaults to streaming_task_id contextvar.
- Callers can override prompt_cache_key via model_settings.extra_args.

The existing TestStreamingModel* tests in this file are unrelated and
remain broken for reasons predating this change (incorrect
_mock_adk_streaming fixture lookup and unset contextvars).

Author: x

src/agentex/lib/core/temporal/plugins/openai_agents/tests/test_streaming_model.py

@@ -845,4 +845,220 @@ async def test_missing_task_id_error(self, streaming_model):
                 handoffs=[],
                 tracing=None,
                 task_id=None  # Missing task_id
-            )
+            )
+
+
+class TestStreamingModelUsageAndCacheKey:
+    """Tests for real-Usage capture, span output_data["usage"], and prompt_cache_key routing."""
+
+    @staticmethod
+    def _async_iter(events):
+        async def _gen():
+            for event in events:
+                yield event
+        return _gen()
+
+    @staticmethod
+    def _make_response_completed_event(
+        *,
+        input_tokens: int = 0,
+        output_tokens: int = 0,
+        total_tokens: int = 0,
+        cached_tokens: int = 0,
+        reasoning_tokens: int = 0,
+        with_usage: bool = True,
+    ):
+        from openai.types.responses import ResponseCompletedEvent
+
+        usage = MagicMock()
+        usage.input_tokens = input_tokens
+        usage.output_tokens = output_tokens
+        usage.total_tokens = total_tokens
+        usage.input_tokens_details = MagicMock(cached_tokens=cached_tokens)
+        usage.output_tokens_details = MagicMock(reasoning_tokens=reasoning_tokens)
+
+        response = MagicMock()
+        response.output = []
+        response.usage = usage if with_usage else None
+
+        event = MagicMock(spec=ResponseCompletedEvent)
+        event.response = response
+        return event
+
+    @pytest.fixture
+    def context_set(self):
+        """Set/reset the streaming contextvars used by get_response."""
+        from agentex.lib.core.temporal.plugins.openai_agents.interceptors.context_interceptor import (
+            streaming_task_id,
+            streaming_trace_id,
+            streaming_parent_span_id,
+        )
+        task_token = streaming_task_id.set("test-task-abc")
+        trace_token = streaming_trace_id.set("test-trace-123")
+        parent_token = streaming_parent_span_id.set("test-parent-span")
+        try:
+            yield streaming_task_id, streaming_trace_id, streaming_parent_span_id
+        finally:
+            streaming_task_id.reset(task_token)
+            streaming_trace_id.reset(trace_token)
+            streaming_parent_span_id.reset(parent_token)
+
+    @pytest.fixture
+    def mock_span(self):
+        return MagicMock()
+
+    @pytest.fixture
+    def streaming_model_with_mock_tracer(self, streaming_model, mock_span):
+        """A streaming_model whose tracer.trace().span(...) yields a captured mock span."""
+        async_cm = MagicMock()
+        async_cm.__aenter__ = AsyncMock(return_value=mock_span)
+        async_cm.__aexit__ = AsyncMock(return_value=False)
+        trace_obj = MagicMock()
+        trace_obj.span = MagicMock(return_value=async_cm)
+        streaming_model.tracer = MagicMock()
+        streaming_model.tracer.trace = MagicMock(return_value=trace_obj)
+        return streaming_model
+
+    @pytest.mark.asyncio
+    async def test_get_response_captures_usage_from_completed_event(
+        self,
+        streaming_model_with_mock_tracer,
+        context_set,  # noqa: ARG002
+    ):
+        model = streaming_model_with_mock_tracer
+        completed = self._make_response_completed_event(
+            input_tokens=1234,
+            output_tokens=56,
+            total_tokens=1290,
+            cached_tokens=987,
+            reasoning_tokens=42,
+        )
+        model.client.responses.create = AsyncMock(return_value=self._async_iter([completed]))
+
+        response = await model.get_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=None,
+        )
+
+        assert response.usage.input_tokens == 1234
+        assert response.usage.output_tokens == 56
+        assert response.usage.total_tokens == 1290
+        assert response.usage.input_tokens_details.cached_tokens == 987
+        assert response.usage.output_tokens_details.reasoning_tokens == 42
+
+    @pytest.mark.asyncio
+    async def test_get_response_usage_falls_back_when_no_completed_event(
+        self,
+        streaming_model_with_mock_tracer,
+        context_set,  # noqa: ARG002
+    ):
+        model = streaming_model_with_mock_tracer
+        # Stream ends with no ResponseCompletedEvent
+        model.client.responses.create = AsyncMock(return_value=self._async_iter([]))
+
+        response = await model.get_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=None,
+        )
+
+        assert response.usage.input_tokens == 0
+        assert response.usage.output_tokens == 0
+        assert response.usage.total_tokens == 0
+        assert response.usage.input_tokens_details.cached_tokens == 0
+        assert response.usage.output_tokens_details.reasoning_tokens == 0
+
+    @pytest.mark.asyncio
+    async def test_get_response_emits_usage_in_span_output(
+        self,
+        streaming_model_with_mock_tracer,
+        context_set,  # noqa: ARG002
+        mock_span,
+    ):
+        model = streaming_model_with_mock_tracer
+        completed = self._make_response_completed_event(
+            input_tokens=100,
+            output_tokens=10,
+            total_tokens=110,
+            cached_tokens=80,
+            reasoning_tokens=5,
+        )
+        model.client.responses.create = AsyncMock(return_value=self._async_iter([completed]))
+
+        await model.get_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=None,
+        )
+
+        assert isinstance(mock_span.output, dict)
+        assert "usage" in mock_span.output
+        usage_block = mock_span.output["usage"]
+        assert usage_block["input_tokens"] == 100
+        assert usage_block["output_tokens"] == 10
+        assert usage_block["total_tokens"] == 110
+        assert usage_block["cached_input_tokens"] == 80
+        assert usage_block["reasoning_tokens"] == 5
+
+    @pytest.mark.asyncio
+    async def test_get_response_passes_prompt_cache_key_from_contextvar(
+        self,
+        streaming_model_with_mock_tracer,
+        context_set,  # noqa: ARG002
+    ):
+        model = streaming_model_with_mock_tracer
+        completed = self._make_response_completed_event()
+        model.client.responses.create = AsyncMock(return_value=self._async_iter([completed]))
+
+        await model.get_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=None,
+        )
+
+        kwargs = model.client.responses.create.call_args.kwargs
+        assert kwargs["prompt_cache_key"] == "test-task-abc"
+
+    @pytest.mark.asyncio
+    async def test_get_response_caller_override_for_prompt_cache_key(
+        self,
+        streaming_model_with_mock_tracer,
+        context_set,  # noqa: ARG002
+    ):
+        model = streaming_model_with_mock_tracer
+        completed = self._make_response_completed_event()
+        model.client.responses.create = AsyncMock(return_value=self._async_iter([completed]))
+
+        await model.get_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(extra_args={"prompt_cache_key": "my-key"}),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=None,
+        )
+
+        kwargs = model.client.responses.create.call_args.kwargs
+        assert kwargs["prompt_cache_key"] == "my-key"
+        # Make sure the override key was popped from extra_args and not double-passed.
+        assert "prompt_cache_key" not in {
+            k for k in kwargs if k != "prompt_cache_key"
+        } or list(kwargs).count("prompt_cache_key") == 1