fix(openai-agents): restore created_at deterministic timestamp (AGX1-378)

Thread the workflow-supplied created_at through
UnifiedEmitter.auto_send_turn(turn, created_at=created_at) so the first
agent message of the turn is stamped with the deterministic timestamp
(e.g. workflow.now()) just as the original inline loop did before the
unified-harness migration.

The foundation (b4b8b33) wired auto_send_turn to accept and forward
created_at to every streaming_task_message_context call. This commit
connects the call site in run_agent_streamed_auto_send to that new
parameter, restoring the behaviour that the migration comment documented
as a known trade-off.

Update the stale limitation comment to reflect the fix. Add
test_run_agent_streamed_auto_send_forwards_created_at, which drives
the activity through a fake stream with a pinned datetime and asserts
every streaming context receives that datetime.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: declan-scale

src/agentex/lib/core/services/adk/providers/openai.py

@@ -742,18 +742,13 @@ async def run_agent_streamed_auto_send(
         ) as span:
             heartbeat_if_in_workflow("run agent streamed auto send")
 
-            # created_at limitation: the unified auto_send path
-            # (UnifiedEmitter.auto_send_turn -> auto_send) does not thread the
-            # workflow-supplied created_at through to the per-message streaming
-            # contexts it opens for the agent's turn. The previous inline loop
-            # stamped the first message of the turn with workflow.now() to win
-            # the race against the workflow's user-echo at the server; under the
-            # unified surface that first agent message instead falls back to the
-            # server's wall clock. This is an accepted trade-off for migrating
-            # onto the shared harness; if strict first-message ordering becomes
-            # necessary, auto_send must accept and dispense a created_at.
-            # The dispenser is still used below for the guardrail-rejection
-            # messages, which open their own streaming contexts directly.
+            # AGX1-378 restored: created_at is now threaded through
+            # UnifiedEmitter.auto_send_turn -> auto_send -> every
+            # streaming_task_message_context call, so the first agent message of
+            # the turn is stamped with the workflow-supplied timestamp (e.g.
+            # workflow.now()) just as the original inline loop did.
+            # The dispenser is still used below for guardrail-rejection messages,
+            # which open their own streaming contexts directly.
             _take_created_at = _make_created_at_dispenser(created_at)
 
             async with mcp_server_context(mcp_server_params, mcp_timeout_seconds) as servers:
@@ -817,7 +812,7 @@ async def run_agent_streamed_auto_send(
                 )
 
                 try:
-                    await emitter.auto_send_turn(turn)
+                    await emitter.auto_send_turn(turn, created_at=created_at)
 
                 except InputGuardrailTripwireTriggered as e:
                     # Handle guardrail trigger by sending a rejection message

tests/lib/adk/providers/test_openai_activities.py

@@ -618,6 +618,48 @@ def _assert_tools_conversion(self, starting_agent, tools_case, _original_tools):
         else:
             raise ValueError(f"Unknown tools_case: {tools_case}")
 
+    @patch("agents.Runner.run_streamed")
+    async def test_run_agent_streamed_auto_send_forwards_created_at(self, mock_runner_run_streamed):
+        """created_at is forwarded to every streaming context opened by auto_send_turn (AGX1-378)."""
+        from datetime import datetime, timezone
+
+        from agentex.lib.core.temporal.activities.adk.providers.openai_activities import (
+            RunAgentStreamedAutoSendParams,
+        )
+
+        deterministic_ts = datetime(2025, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+
+        mock_streaming_result = self._create_streaming_result_mock()
+
+        async def _no_events():
+            return
+            yield  # make it an async generator
+
+        mock_streaming_result.stream_events = _no_events
+        mock_runner_run_streamed.return_value = mock_streaming_result
+
+        mock_tracer = self._create_mock_tracer()
+        openai_service, openai_activities, env = self._create_test_setup(mock_tracer)
+        mock_ctx, recorded_created_ats = self._setup_streaming_service_mocks_with_created_at(openai_service)
+
+        params = RunAgentStreamedAutoSendParams(
+            input_list=[{"role": "user", "content": "hello"}],
+            mcp_server_params=[],
+            agent_name="test_agent",
+            agent_instructions="You are a helpful assistant",
+            trace_id="test-trace-id",
+            parent_span_id="test-span-id",
+            task_id="test-task-id",
+            created_at=deterministic_ts,
+        )
+
+        await env.run(openai_activities.run_agent_streamed_auto_send, params)
+
+        assert all(ts == deterministic_ts for ts in recorded_created_ats), (
+            f"Expected all streaming contexts to receive created_at={deterministic_ts!r}, "
+            f"got: {recorded_created_ats!r}"
+        )
+
     def _setup_streaming_service_mocks(self, openai_service):
         """Helper method to setup streaming service mocks for run_agent_auto_send."""
         from unittest.mock import AsyncMock
@@ -665,6 +707,39 @@ async def mock_streaming_context_manager(*_args, **kwargs):
 
         return mock_streaming_context
 
+    def _setup_streaming_service_mocks_with_created_at(self, openai_service):
+        """Like _setup_streaming_service_mocks but also records every created_at kwarg."""
+        from contextlib import asynccontextmanager
+        from unittest.mock import AsyncMock
+
+        from agentex.types.task_message import TaskMessage
+
+        mock_streaming_service = AsyncMock()
+        mock_agentex_client = AsyncMock()
+
+        mock_streaming_context = AsyncMock()
+        mock_task_message = Mock(spec=TaskMessage)
+        mock_task_message.id = "test-task-message-id"
+        mock_task_message.task_id = "test-task-id"
+        mock_task_message.content = {"type": "text", "content": "test"}
+        mock_streaming_context.task_message = mock_task_message
+        mock_streaming_context.stream_update = AsyncMock()
+
+        recorded_created_ats: list = []
+
+        @asynccontextmanager
+        async def mock_ctx_manager(*_args, **kwargs):
+            recorded_created_ats.append(kwargs.get("created_at"))
+            yield mock_streaming_context
+
+        mock_streaming_service.streaming_task_message_context = mock_ctx_manager
+        mock_streaming_context.opened_contents = []
+
+        openai_service.streaming_service = mock_streaming_service
+        openai_service.agentex_client = mock_agentex_client
+
+        return mock_streaming_context, recorded_created_ats
+
     def _create_code_interpreter_tool_call_mock(self, call_id="code_interpreter_call_123"):
         """Helper to create ResponseCodeInterpreterToolCall mock objects."""
         return ResponseCodeInterpreterToolCall(