refactor: address PR review comments on created_at plumbing

- Drop redundant in_temporal_workflow() guard in MessagesModule.create /
  create_batch; workflow_now_if_in_workflow() already returns None outside
  a workflow.
- Extract the inline _take_created_at closure in providers/openai.py into
  a module-level _make_created_at_dispenser helper, eliminating the
  duplicated block between run_agent_auto_send and run_agent_streamed_auto_send.
- Drop @pytest.mark.asyncio from tests/lib/adk/test_messages_module.py to
  match the project's asyncio_mode="auto" convention used in the sibling
  test_messages_service.py.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: declan-scale

src/agentex/lib/adk/_modules/messages.py

@@ -85,7 +85,7 @@ async def create(
         """
         # Default created_at to workflow.now() so two awaited adk.messages.create
         # calls from the same workflow are guaranteed monotonic at the server.
-        if created_at is None and in_temporal_workflow():
+        if created_at is None:
             created_at = workflow_now_if_in_workflow()
         params = CreateMessageParams(
             trace_id=trace_id,
@@ -185,7 +185,7 @@ async def create_batch(
         Returns:
             List[TaskMessageEntity]: The created messages.
         """
-        if created_at is None and in_temporal_workflow():
+        if created_at is None:
             created_at = workflow_now_if_in_workflow()
         params = CreateMessagesBatchParams(
             task_id=task_id,

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

@@ -4,6 +4,7 @@
 from typing import Any, Literal
 from datetime import datetime
 from contextlib import AsyncExitStack, asynccontextmanager
+from collections.abc import Callable
 
 from mcp import StdioServerParameters
 from agents import Agent, Runner, RunResult, RunResultStreaming
@@ -74,6 +75,22 @@ async def mcp_server_context(
         yield servers
 
 
+def _make_created_at_dispenser(initial: datetime | None) -> Callable[[], datetime | None]:
+    # Returns a closure that yields the workflow-supplied created_at exactly
+    # once (on the first call), then None forever after. Used to stamp the
+    # first agent message of a turn with workflow.now() while letting
+    # subsequent messages fall back to server wall-clock — see the call sites
+    # in run_agent_auto_send / run_agent_streamed_auto_send for context.
+    pending: list[datetime | None] = [initial]
+
+    def take() -> datetime | None:
+        value = pending[0]
+        pending[0] = None
+        return value
+
+    return take
+
+
 class OpenAIService:
     """Service for OpenAI agent operations using the agents library."""
 
@@ -377,15 +394,7 @@ async def run_agent_auto_send(
         ) as span:
             heartbeat_if_in_workflow("run agent auto send")
 
-            # See run_agent_streamed_auto_send for the rationale: only the
-            # first message opened in this turn carries the workflow-supplied
-            # created_at; the rest fall back to server wall clock.
-            _pending_created_at: list[datetime | None] = [created_at]
-
-            def _take_created_at() -> datetime | None:
-                value = _pending_created_at[0]
-                _pending_created_at[0] = None
-                return value
+            _take_created_at = _make_created_at_dispenser(created_at)
 
             async with mcp_server_context(mcp_server_params, mcp_timeout_seconds) as servers:
                 tools = (
@@ -758,12 +767,7 @@ async def run_agent_streamed_auto_send(
             # user-echo at the server. Subsequent messages in the same turn are
             # separated by network/processing latency and rely on the server's
             # wall clock.
-            _pending_created_at: list[datetime | None] = [created_at]
-
-            def _take_created_at() -> datetime | None:
-                value = _pending_created_at[0]
-                _pending_created_at[0] = None
-                return value
+            _take_created_at = _make_created_at_dispenser(created_at)
 
             async with mcp_server_context(mcp_server_params, mcp_timeout_seconds) as servers:
                 tools = (

tests/lib/adk/test_messages_module.py

@@ -11,8 +11,6 @@
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock, patch
 
-import pytest
-
 import agentex.lib.adk._modules.messages as _messages_mod
 from agentex.types.task_message import TaskMessage
 from agentex.types.text_content import TextContent
@@ -38,7 +36,6 @@ def _make_module() -> tuple[AsyncMock, MessagesModule]:
 
 
 class TestMessagesModuleCreate:
-    @pytest.mark.asyncio
     async def test_outside_workflow_does_not_inject_created_at(self) -> None:
         mock_service, module = _make_module()
         mock_service.create_message.return_value = _make_task_message()
@@ -52,7 +49,6 @@ async def test_outside_workflow_does_not_inject_created_at(self) -> None:
         kwargs = mock_service.create_message.call_args.kwargs
         assert kwargs["created_at"] is None
 
-    @pytest.mark.asyncio
     async def test_inside_workflow_auto_injects_workflow_now(self) -> None:
         mock_service, module = _make_module()
         mock_service.create_message.return_value = _make_task_message()
@@ -80,7 +76,6 @@ async def fake_execute_activity(**call_kwargs):
         params = captured["request"]
         assert params.created_at == _FIXED_NOW
 
-    @pytest.mark.asyncio
     async def test_caller_supplied_created_at_is_respected(self) -> None:
         mock_service, module = _make_module()
         mock_service.create_message.return_value = _make_task_message()
@@ -99,7 +94,6 @@ async def test_caller_supplied_created_at_is_respected(self) -> None:
 
 
 class TestMessagesModuleCreateBatch:
-    @pytest.mark.asyncio
     async def test_inside_workflow_auto_injects_workflow_now(self) -> None:
         mock_service, module = _make_module()
         mock_service.create_messages_batch.return_value = [_make_task_message()]