feat(lifecycle): add on_turn_start and on_turn_end hooks to RunHooksBase (#2671)

Both RunHooksBase and AgentHooksBase get two new hook methods:

- on_turn_start(context, agent, turn_number): fires before each LLM call
- on_turn_end(context, agent, turn_number): fires after all tool calls for
  the turn complete (i.e. just before the next-step decision)

Turn numbers are 1-indexed and increment each time through the agent
loop, regardless of handoffs.  The hooks are called in both the sync
and streaming code paths.  Agent-level hooks on agent.hooks are also
called, matching the existing on_tool_start/on_tool_end pattern.

Closes #2671

Author: Ubuntu

src/agents/lifecycle.py

@@ -86,6 +86,36 @@ async def on_tool_end(
         """Called immediately after a local tool is invoked."""
         pass
 
+    async def on_turn_start(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: TAgent,
+        turn_number: int,
+    ) -> None:
+        """Called at the start of each agent turn, before the LLM is invoked.
+
+        Args:
+            context: The run context wrapper.
+            agent: The current agent.
+            turn_number: The 1-indexed turn number (increments each time through the agent loop).
+        """
+        pass
+
+    async def on_turn_end(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: TAgent,
+        turn_number: int,
+    ) -> None:
+        """Called at the end of each agent turn, after all tool calls for that turn complete.
+
+        Args:
+            context: The run context wrapper.
+            agent: The current agent.
+            turn_number: The 1-indexed turn number.
+        """
+        pass
+
 
 class AgentHooksBase(Generic[TContext, TAgent]):
     """A class that receives callbacks on various lifecycle events for a specific agent. You can
@@ -148,6 +178,36 @@ async def on_tool_end(
         """Called immediately after a local tool is invoked."""
         pass
 
+    async def on_turn_start(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: TAgent,
+        turn_number: int,
+    ) -> None:
+        """Called at the start of each agent turn, before the LLM is invoked.
+
+        Args:
+            context: The run context wrapper.
+            agent: The current agent.
+            turn_number: The 1-indexed turn number (increments each time through the agent loop).
+        """
+        pass
+
+    async def on_turn_end(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: TAgent,
+        turn_number: int,
+    ) -> None:
+        """Called at the end of each agent turn, after all tool calls for that turn complete.
+
+        Args:
+            context: The run context wrapper.
+            agent: The current agent.
+            turn_number: The 1-indexed turn number.
+        """
+        pass
+
     async def on_llm_start(
         self,
         context: RunContextWrapper[TContext],

src/agents/run.py

@@ -110,8 +110,8 @@
 from .tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult
 from .tracing import Span, SpanError, agent_span, get_current_trace
 from .tracing.context import TraceCtxManager, create_trace_for_run
-from .tracing.span_data import AgentSpanData
-from .util import _error_tracing
+from .tracing.span_data import AgentSpanData, TaskSpanData
+from .util import _coro, _error_tracing
 
 DEFAULT_AGENT_RUNNER: AgentRunner = None  # type: ignore
 # the value is set at the end of the module
@@ -968,6 +968,17 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult:
 
                     logger.debug("Running agent %s (turn %s)", current_agent.name, current_turn)
 
+                    await asyncio.gather(
+                        hooks.on_turn_start(context_wrapper, current_agent, current_turn),
+                        (
+                            current_agent.hooks.on_turn_start(
+                                context_wrapper, current_agent, current_turn
+                            )
+                            if current_agent.hooks
+                            else _coro.noop_coroutine()
+                        ),
+                    )
+
                     if session_persistence_enabled:
                         try:
                             last_saved_input_snapshot_for_rewind = (
@@ -1093,6 +1104,17 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult:
                     last_saved_input_snapshot_for_rewind = None
                     should_run_agent_start_hooks = False
 
+                    await asyncio.gather(
+                        hooks.on_turn_end(context_wrapper, current_agent, current_turn),
+                        (
+                            current_agent.hooks.on_turn_end(
+                                context_wrapper, current_agent, current_turn
+                            )
+                            if current_agent.hooks
+                            else _coro.noop_coroutine()
+                        ),
+                    )
+
                     model_responses.append(turn_result.model_response)
                     original_input = turn_result.original_input
                     # For model input, use new_step_items (filtered on handoffs).

src/agents/run_internal/run_loop.py

@@ -820,6 +820,17 @@ async def _save_stream_items_without_count(
                 streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                 break
 
+            await asyncio.gather(
+                hooks.on_turn_start(context_wrapper, current_agent, current_turn),
+                (
+                    current_agent.hooks.on_turn_start(
+                        context_wrapper, current_agent, current_turn
+                    )
+                    if current_agent.hooks
+                    else _coro.noop_coroutine()
+                ),
+            )
+
             if current_turn == 1:
                 all_input_guardrails = starting_agent.input_guardrails + (
                     run_config.input_guardrails or []
@@ -909,6 +920,17 @@ async def _save_stream_items_without_count(
                     tool_use_tracker
                 )
 
+                await asyncio.gather(
+                    hooks.on_turn_end(context_wrapper, current_agent, current_turn),
+                    (
+                        current_agent.hooks.on_turn_end(
+                            context_wrapper, current_agent, current_turn
+                        )
+                        if current_agent.hooks
+                        else _coro.noop_coroutine()
+                    ),
+                )
+
                 streamed_result.raw_responses = streamed_result.raw_responses + [
                     turn_result.model_response
                 ]

tests/test_turn_lifecycle_hooks.py

@@ -0,0 +1,189 @@
+"""Tests for on_turn_start / on_turn_end lifecycle hooks (issue #2671)."""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any, Optional
+
+import pytest
+
+from agents import Agent, Runner
+from agents.items import ModelResponse, TResponseInputItem
+from agents.lifecycle import AgentHooks, RunHooks
+from agents.run_context import RunContextWrapper, TContext
+from agents.tool import FunctionTool, Tool
+
+from .fake_model import FakeModel
+from .test_responses import get_function_tool, get_function_tool_call, get_text_message
+
+
+class TurnTrackingRunHooks(RunHooks):
+    """Records turn numbers seen by on_turn_start and on_turn_end."""
+
+    def __init__(self) -> None:
+        self.turn_starts: list[int] = []
+        self.turn_ends: list[int] = []
+        self.events: dict[str, int] = defaultdict(int)
+
+    async def on_turn_start(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> None:
+        self.turn_starts.append(turn_number)
+        self.events["on_turn_start"] += 1
+
+    async def on_turn_end(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> None:
+        self.turn_ends.append(turn_number)
+        self.events["on_turn_end"] += 1
+
+
+class TurnTrackingAgentHooks(AgentHooks):
+    """Records turn numbers seen on agent-level hooks."""
+
+    def __init__(self) -> None:
+        self.turn_starts: list[int] = []
+        self.turn_ends: list[int] = []
+
+    async def on_turn_start(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> None:
+        self.turn_starts.append(turn_number)
+
+    async def on_turn_end(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> None:
+        self.turn_ends.append(turn_number)
+
+
+@pytest.mark.asyncio
+async def test_on_turn_start_and_end_single_turn() -> None:
+    """on_turn_start and on_turn_end are both called once for a single-turn run."""
+    model = FakeModel()
+    model.set_next_output([get_text_message("hello")])
+
+    hooks = TurnTrackingRunHooks()
+    agent = Agent(name="A", model=model)
+
+    await Runner.run(agent, input="hi", hooks=hooks)
+
+    assert hooks.turn_starts == [1]
+    assert hooks.turn_ends == [1]
+
+
+@pytest.mark.asyncio
+async def test_on_turn_numbers_multi_turn() -> None:
+    """Turn numbers increment correctly across multiple turns."""
+    model = FakeModel()
+    # Turn 1: model calls a tool; turn 2: model produces final output.
+    tool = get_function_tool("my_tool", "tool_result")
+    model.add_multiple_turn_outputs([
+        [get_function_tool_call("my_tool", "{}")],
+        [get_text_message("done")],
+    ])
+
+    hooks = TurnTrackingRunHooks()
+    agent = Agent(name="A", model=model, tools=[tool])
+
+    await Runner.run(agent, input="hi", hooks=hooks)
+
+    assert hooks.turn_starts == [1, 2]
+    assert hooks.turn_ends == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_on_turn_start_fires_before_llm() -> None:
+    """on_turn_start fires before the LLM call each turn."""
+    call_order: list[str] = []
+
+    class OrderTrackingHooks(RunHooks):
+        async def on_turn_start(self, context: Any, agent: Any, turn_number: int) -> None:
+            call_order.append(f"turn_start:{turn_number}")
+
+        async def on_llm_start(self, context: Any, agent: Any, system_prompt: Any, input_items: Any) -> None:
+            call_order.append(f"llm_start")
+
+        async def on_llm_end(self, context: Any, agent: Any, response: Any) -> None:
+            call_order.append(f"llm_end")
+
+        async def on_turn_end(self, context: Any, agent: Any, turn_number: int) -> None:
+            call_order.append(f"turn_end:{turn_number}")
+
+    model = FakeModel()
+    model.set_next_output([get_text_message("hello")])
+    hooks = OrderTrackingHooks()
+    agent = Agent(name="A", model=model)
+
+    await Runner.run(agent, input="hi", hooks=hooks)
+
+    # turn_start must come before llm_start, llm_end before turn_end
+    ts_idx = call_order.index("turn_start:1")
+    ls_idx = call_order.index("llm_start")
+    le_idx = call_order.index("llm_end")
+    te_idx = call_order.index("turn_end:1")
+
+    assert ts_idx < ls_idx
+    assert ls_idx < le_idx
+    assert le_idx < te_idx
+
+
+@pytest.mark.asyncio
+async def test_agent_level_on_turn_start_and_end() -> None:
+    """Agent-level on_turn_start / on_turn_end hooks are also called."""
+    model = FakeModel()
+    model.set_next_output([get_text_message("hello")])
+
+    agent_hooks = TurnTrackingAgentHooks()
+    agent = Agent(name="A", model=model, hooks=agent_hooks)
+
+    await Runner.run(agent, input="hi")
+
+    assert agent_hooks.turn_starts == [1]
+    assert agent_hooks.turn_ends == [1]
+
+
+@pytest.mark.asyncio
+async def test_run_and_agent_hooks_both_called() -> None:
+    """Both run-level and agent-level hooks fire for the same turn."""
+    model = FakeModel()
+    model.set_next_output([get_text_message("hi")])
+
+    run_hooks = TurnTrackingRunHooks()
+    agent_hooks = TurnTrackingAgentHooks()
+    agent = Agent(name="A", model=model, hooks=agent_hooks)
+
+    await Runner.run(agent, input="hi", hooks=run_hooks)
+
+    assert run_hooks.turn_starts == [1]
+    assert run_hooks.turn_ends == [1]
+    assert agent_hooks.turn_starts == [1]
+    assert agent_hooks.turn_ends == [1]
+
+
+@pytest.mark.asyncio
+async def test_on_turn_hooks_with_streaming() -> None:
+    """on_turn_start and on_turn_end are called when using the streaming runner."""
+    model = FakeModel()
+    model.set_next_output([get_text_message("streamed")])
+
+    hooks = TurnTrackingRunHooks()
+    agent = Agent(name="A", model=model)
+
+    result = Runner.run_streamed(agent, input="hi", hooks=hooks)
+    async for _ in result.stream_events():
+        pass
+
+    assert hooks.turn_starts == [1]
+    assert hooks.turn_ends == [1]