Add hooks and tests

Author: danielmillerp

src/agentex/lib/core/temporal/plugins/openai_agents/README.md

@@ -700,6 +700,52 @@ The most important innovation is **runtime task_id extraction**. Unlike passing
 
 ---
 
+## Testing
+
+### Running Tests
+
+The streaming model implementation has comprehensive tests in `tests/test_streaming_model.py` that verify all configurations, tool types, and edge cases.
+
+#### From Repository Root
+
+```bash
+# Run all tests
+rye run pytest src/agentex/lib/core/temporal/plugins/openai_agents/tests/test_streaming_model.py -v
+
+# Run without parallel execution (more stable)
+rye run pytest src/agentex/lib/core/temporal/plugins/openai_agents/tests/test_streaming_model.py -v -n0
+
+# Run specific test
+rye run pytest src/agentex/lib/core/temporal/plugins/openai_agents/tests/test_streaming_model.py::TestStreamingModelSettings::test_temperature_setting -v
+```
+
+#### From Test Directory
+
+```bash
+cd src/agentex/lib/core/temporal/plugins/openai_agents/tests
+
+# Run all tests
+rye run pytest test_streaming_model.py -v
+
+# Run without parallel execution (recommended)
+rye run pytest test_streaming_model.py -v -n0
+
+# Run specific test class
+rye run pytest test_streaming_model.py::TestStreamingModelSettings -v
+```
+
+#### Test Coverage
+
+The test suite covers:
+- **ModelSettings**: All configuration parameters (temperature, reasoning, truncation, etc.)
+- **Tool Types**: Function tools, web search, file search, computer tools, MCP tools, etc.
+- **Streaming**: Redis context creation, task ID threading, error handling
+- **Edge Cases**: Missing task IDs, multiple computer tools, handoffs
+
+**Note**: Tests run faster without parallel execution (`-n0` flag) and avoid potential state pollution between test workers. All 29 tests pass individually; parallel execution may show 4-6 intermittent failures due to shared mock state.
+
+---
+
 ## Conclusion
 
 This implementation threads task_id through Temporal's OpenAI plugin architecture to enable real-time streaming while maintaining workflow determinism. The key innovation is extracting task_id at runtime rather than at plugin initialization, allowing dynamic streaming channels per workflow execution.

src/agentex/lib/core/temporal/plugins/openai_agents/__init__.py

@@ -6,30 +6,43 @@
 The streaming implementation works by:
 1. Threading a task_id through the workflow execution
 2. Streaming LLM responses to Redis in real-time from activities
-3. Returning complete responses to maintain Temporal determinism
+3. Streaming lifecycle events (tool calls, handoffs) via hooks and activities
+4. Returning complete responses to maintain Temporal determinism
 
-Example:
+Example - Complete Setup:
     >>> from agentex.lib.core.temporal.plugins.openai_agents import (
     ...     CustomStreamingOpenAIAgentsPlugin,
     ...     StreamingModelProvider,
+    ...     TemporalStreamingHooks,
     ... )
     >>> from temporalio.contrib.openai_agents import ModelActivityParameters
     >>> from datetime import timedelta
+    >>> from agents import Agent, Runner
     >>>
-    >>> # Create streaming model provider
+    >>> # 1. Create streaming model provider
     >>> model_provider = StreamingModelProvider()
     >>>
-    >>> # Create plugin with streaming support
+    >>> # 2. Create plugin with streaming support
     >>> plugin = CustomStreamingOpenAIAgentsPlugin(
     ...     model_params=ModelActivityParameters(
     ...         start_to_close_timeout=timedelta(seconds=120),
     ...     ),
     ...     model_provider=model_provider,
     ... )
     >>>
-    >>> # In workflow, pass task_id through context
+    >>> # 3. In workflow, create hooks for streaming lifecycle events
+    >>> # The hooks automatically use the built-in streaming activity
+    >>> hooks = TemporalStreamingHooks(task_id="your-task-id")
+    >>>
+    >>> # 4. Run agent with streaming context and hooks
     >>> context = {"task_id": "your-task-id"}
-    >>> result = await runner.run(agent, input, context=context)
+    >>> result = await Runner.run(agent, input, context=context, hooks=hooks)
+
+This gives you:
+- Real-time streaming of LLM responses (via StreamingModel)
+- Real-time streaming of tool calls (via TemporalStreamingHooks)
+- Real-time streaming of agent handoffs (via TemporalStreamingHooks)
+- Full Temporal durability and observability
 """
 
 from agentex.lib.core.temporal.plugins.openai_agents.streaming_model import (
@@ -45,6 +58,12 @@
     StreamingTemporalRunner,
     StreamingTemporalModelStub,
 )
+from agentex.lib.core.temporal.plugins.openai_agents.hooks import (
+    TemporalStreamingHooks,
+)
+from agentex.lib.core.temporal.plugins.openai_agents.activities import (
+    stream_lifecycle_content,
+)
 
 __all__ = [
     "CustomStreamingOpenAIAgentsPlugin",
@@ -54,4 +73,6 @@
     "StreamingActivityModelInput",
     "StreamingTemporalRunner",
     "StreamingTemporalModelStub",
+    "TemporalStreamingHooks",
+    "stream_lifecycle_content",
 ]

src/agentex/lib/core/temporal/plugins/openai_agents/activities.py

@@ -0,0 +1,78 @@
+"""Temporal activities for streaming agent lifecycle events.
+
+This module provides reusable Temporal activities for streaming content
+to the AgentEx UI, designed to work with TemporalStreamingHooks.
+"""
+
+from typing import Union
+
+from temporalio import activity
+
+from agentex.lib import adk
+from agentex.types.text_content import TextContent
+from agentex.types.task_message_update import StreamTaskMessageFull
+from agentex.types.task_message_content import (
+    TaskMessageContent,
+    ToolRequestContent,
+    ToolResponseContent,
+)
+
+
+@activity.defn(name="stream_lifecycle_content")
+async def stream_lifecycle_content(
+    task_id: str,
+    content: Union[TextContent, ToolRequestContent, ToolResponseContent, TaskMessageContent],
+) -> None:
+    """Stream agent lifecycle content to the AgentEx UI.
+
+    This is a universal streaming activity that can handle any type of agent
+    lifecycle content (text messages, tool requests, tool responses, etc.).
+    It uses the AgentEx streaming context to send updates to the UI in real-time.
+
+    Designed to work seamlessly with TemporalStreamingHooks. The hooks class
+    will call this activity automatically when lifecycle events occur.
+
+    Args:
+        task_id: The AgentEx task ID for routing the content to the correct UI session
+        content: The content to stream - can be any of:
+            - TextContent: Plain text messages (e.g., handoff notifications)
+            - ToolRequestContent: Tool invocation requests with call_id and name
+            - ToolResponseContent: Tool execution results with call_id and output
+            - TaskMessageContent: Generic task message content
+
+    Example:
+        Register this activity with your Temporal worker::
+
+            from agentex.lib.core.temporal.plugins.openai_agents import (
+                TemporalStreamingHooks,
+                stream_lifecycle_content,
+            )
+
+            # In your workflow
+            hooks = TemporalStreamingHooks(
+                task_id=params.task.id,
+                stream_activity=stream_lifecycle_content
+            )
+            result = await Runner.run(agent, input, hooks=hooks)
+
+    Note:
+        This activity is non-blocking and will not throw exceptions to the workflow.
+        Any streaming errors are logged but do not fail the activity. This ensures
+        that streaming failures don't break the agent execution.
+    """
+    try:
+        async with adk.streaming.streaming_task_message_context(
+            task_id=task_id,
+            initial_content=content,
+        ) as streaming_context:
+            # Send the content as a full message update
+            await streaming_context.stream_update(
+                StreamTaskMessageFull(
+                    parent_task_message=streaming_context.task_message,
+                    content=content,
+                    type="full",
+                )
+            )
+    except Exception as e:
+        # Log error but don't fail the activity - streaming failures shouldn't break execution
+        activity.logger.warning(f"Failed to stream content to task {task_id}: {e}")

src/agentex/lib/core/temporal/plugins/openai_agents/hooks.py

@@ -0,0 +1,193 @@
+"""Temporal streaming hooks for OpenAI Agents SDK lifecycle events.
+
+This module provides a convenience class for streaming agent lifecycle events
+to the AgentEx UI via Temporal activities.
+"""
+
+from typing import Any
+from datetime import timedelta
+
+from agents import Agent, RunContextWrapper, RunHooks, Tool
+from agents.tool_context import ToolContext
+from temporalio import workflow
+
+from agentex.types.text_content import TextContent
+from agentex.types.task_message_content import ToolRequestContent, ToolResponseContent
+from agentex.lib.core.temporal.plugins.openai_agents.activities import stream_lifecycle_content
+
+
+class TemporalStreamingHooks(RunHooks):
+    """Convenience hooks class for streaming OpenAI Agent lifecycle events to the AgentEx UI.
+
+    This class automatically streams agent lifecycle events (tool calls, handoffs) to the
+    AgentEx UI via Temporal activities. It subclasses the OpenAI Agents SDK's RunHooks
+    to intercept lifecycle events and forward them for real-time UI updates.
+
+    Lifecycle events streamed:
+        - Tool requests (on_tool_start): Streams when a tool is about to be invoked
+        - Tool responses (on_tool_end): Streams the tool's execution result
+        - Agent handoffs (on_handoff): Streams when control transfers between agents
+
+    Usage:
+        Basic usage - streams all lifecycle events::
+
+            from agentex.lib.core.temporal.plugins.openai_agents import TemporalStreamingHooks
+
+            hooks = TemporalStreamingHooks(task_id="abc123")
+            result = await Runner.run(agent, input, hooks=hooks)
+
+        Advanced - subclass for custom behavior::
+
+            class MyCustomHooks(TemporalStreamingHooks):
+                async def on_tool_start(self, context, agent, tool):
+                    # Add custom logic before streaming
+                    await self.my_custom_logging(tool)
+                    # Call parent to stream to UI
+                    await super().on_tool_start(context, agent, tool)
+
+                async def on_agent_start(self, context, agent):
+                    # Override empty methods for additional tracking
+                    print(f"Agent {agent.name} started")
+
+    Power users can ignore this class and subclass agents.RunHooks directly for full control.
+
+    Note:
+        Tool arguments are not available in hooks due to OpenAI SDK architecture.
+        The SDK's hook signature doesn't include tool arguments - they're only passed
+        to the actual tool function. This is why arguments={} in ToolRequestContent.
+
+    Attributes:
+        task_id: The AgentEx task ID for routing streamed events
+        timeout: Timeout for streaming activity calls (default: 10 seconds)
+    """
+
+    def __init__(
+        self,
+        task_id: str,
+        timeout: timedelta = timedelta(seconds=10),
+    ):
+        """Initialize the streaming hooks.
+
+        Args:
+            task_id: AgentEx task ID for routing streamed events to the correct UI session
+            timeout: Timeout for streaming activity invocations (default: 10 seconds)
+        """
+        super().__init__()
+        self.task_id = task_id
+        self.timeout = timeout
+
+    async def on_agent_start(self, context: RunContextWrapper, agent: Agent) -> None:
+        """Called when an agent starts execution.
+
+        Default implementation does nothing. Override to add custom behavior.
+
+        Args:
+            context: The run context wrapper
+            agent: The agent that is starting
+        """
+        pass
+
+    async def on_agent_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None:
+        """Called when an agent completes execution.
+
+        Default implementation does nothing. Override to add custom behavior.
+
+        Args:
+            context: The run context wrapper
+            agent: The agent that completed
+            output: The agent's output
+        """
+        pass
+
+    async def on_tool_start(self, context: RunContextWrapper, agent: Agent, tool: Tool) -> None:
+        """Stream tool request when a tool starts execution.
+
+        Extracts the tool_call_id from the context and streams a ToolRequestContent
+        message to the UI showing that the tool is about to execute.
+
+        Note: Tool arguments are not available in the hook context due to OpenAI SDK
+        design. The hook signature doesn't include tool arguments - they're passed
+        directly to the tool function instead. We send an empty dict as a placeholder.
+
+        Args:
+            context: The run context wrapper (will be a ToolContext with tool_call_id)
+            agent: The agent executing the tool
+            tool: The tool being executed
+        """
+        tool_context = context if isinstance(context, ToolContext) else None
+        tool_call_id = tool_context.tool_call_id if tool_context else f"call_{id(tool)}"
+
+        await workflow.execute_activity_method(
+            stream_lifecycle_content,
+            args=[
+                self.task_id,
+                ToolRequestContent(
+                    author="agent",
+                    tool_call_id=tool_call_id,
+                    name=tool.name,
+                    arguments={},  # Not available in hook context - SDK limitation
+                ),
+            ],
+            start_to_close_timeout=self.timeout,
+        )
+
+    async def on_tool_end(
+        self, context: RunContextWrapper, agent: Agent, tool: Tool, result: str
+    ) -> None:
+        """Stream tool response when a tool completes execution.
+
+        Extracts the tool_call_id and streams a ToolResponseContent message to the UI
+        showing the tool's execution result.
+
+        Args:
+            context: The run context wrapper (will be a ToolContext with tool_call_id)
+            agent: The agent that executed the tool
+            tool: The tool that was executed
+            result: The tool's execution result
+        """
+        tool_context = context if isinstance(context, ToolContext) else None
+        tool_call_id = (
+            getattr(tool_context, "tool_call_id", f"call_{id(tool)}")
+            if tool_context
+            else f"call_{id(tool)}"
+        )
+
+        await workflow.execute_activity_method(
+            stream_lifecycle_content,
+            args=[
+                self.task_id,
+                ToolResponseContent(
+                    author="agent",
+                    tool_call_id=tool_call_id,
+                    name=tool.name,
+                    content=result,
+                ),
+            ],
+            start_to_close_timeout=self.timeout,
+        )
+
+    async def on_handoff(
+        self, context: RunContextWrapper, from_agent: Agent, to_agent: Agent
+    ) -> None:
+        """Stream handoff message when control transfers between agents.
+
+        Sends a text message to the UI indicating that one agent is handing off
+        to another agent.
+
+        Args:
+            context: The run context wrapper
+            from_agent: The agent transferring control
+            to_agent: The agent receiving control
+        """
+        await workflow.execute_activity_method(
+            stream_lifecycle_content,
+            args=[
+                self.task_id,
+                TextContent(
+                    author="agent",
+                    content=f"Handoff from {from_agent.name} to {to_agent.name}",
+                    type="text",
+                ),
+            ],
+            start_to_close_timeout=self.timeout,
+        )