feat(lifecycle): add TurnControl return value to on_turn_start hooks

Address seratch's review feedback on openai#2911: hooks that only observe
cannot affect agent loop orchestration. This commit adds a TurnControl
return type ('continue' | 'stop') so on_turn_start can now halt the
run before the LLM is called for that turn.

Changes:
- lifecycle.py: on_turn_start now returns Union[TurnControl, None]
  (None and 'continue' are equivalent; 'stop' halts the loop)
- run.py (non-streaming path): checks return value; raises
  MaxTurnsExceeded with descriptive message on 'stop'
- run_internal/run_loop.py (streaming path): checks return value;
  signals QueueCompleteSentinel on 'stop'
- __init__.py: exports TurnControl, RunHooksBase, AgentHooksBase
- tests: 4 new test cases covering stop-on-turn-N, stop-on-turn-1,
  explicit 'continue', and agent-level stop

The MaxTurnsExceeded raise on 'stop' keeps behaviour consistent with
the existing max_turns limit: callers can catch and inspect
.run_data if needed.

Author: Aditya Singh

src/agents/__init__.py

@@ -65,7 +65,7 @@
     ToolCallOutputItem,
     TResponseInputItem,
 )
-from .lifecycle import AgentHooks, RunHooks
+from .lifecycle import AgentHooks, AgentHooksBase, RunHooks, RunHooksBase, TurnControl
 from .memory import (
     OpenAIConversationsSession,
     OpenAIResponsesCompactionArgs,
@@ -361,7 +361,10 @@ def enable_verbose_stdout_logging():
     "ReasoningItem",
     "ItemHelpers",
     "RunHooks",
+    "RunHooksBase",
     "AgentHooks",
+    "AgentHooksBase",
+    "TurnControl",
     "Session",
     "SessionABC",
     "SessionSettings",

src/agents/lifecycle.py

@@ -1,4 +1,6 @@
-from typing import Any, Generic, Optional
+from __future__ import annotations
+
+from typing import Any, Generic, Literal, Optional, Union
 
 from typing_extensions import TypeVar
 
@@ -9,10 +11,26 @@
 
 TAgent = TypeVar("TAgent", bound=AgentBase, default=AgentBase)
 
+TurnControl = Literal["continue", "stop"]
+"""Return value for :meth:`RunHooksBase.on_turn_start` / :meth:`AgentHooksBase.on_turn_start`.
+
+* ``"continue"`` (default / ``None``) – proceed with the turn as normal.
+* ``"stop"`` – abort the run gracefully after this hook returns, exactly as if
+  ``max_turns`` had been reached.  The model is **not** called for this turn and
+  :meth:`on_turn_end` is **not** fired.
+"""
+
 
 class RunHooksBase(Generic[TContext, TAgent]):
     """A class that receives callbacks on various lifecycle events in an agent run. Subclass and
     override the methods you need.
+
+    Turn-lifecycle hooks
+    --------------------
+    :meth:`on_turn_start` and :meth:`on_turn_end` fire once per iteration of the
+    agent loop.  :meth:`on_turn_start` may return ``"stop"`` to halt the run
+    gracefully before the LLM is called for that turn (useful for implementing
+    custom turn-budget logic, external kill-switches, etc.).
     """
 
     async def on_llm_start(
@@ -91,15 +109,24 @@ async def on_turn_start(
         context: RunContextWrapper[TContext],
         agent: TAgent,
         turn_number: int,
-    ) -> None:
+    ) -> Union[TurnControl, None]:
         """Called at the start of each agent turn, before the LLM is invoked.
 
+        Returning ``"stop"`` (or raising :class:`StopAgentRun`) will halt the run
+        gracefully — the model is **not** called for this turn and
+        :meth:`on_turn_end` is **not** fired.  Returning ``None`` or ``"continue"``
+        proceeds normally.
+
         Args:
             context: The run context wrapper.
             agent: The current agent.
-            turn_number: The 1-indexed turn number (increments each time through the agent loop).
+            turn_number: The 1-indexed turn number (increments each time through the
+                agent loop).
+
+        Returns:
+            ``None`` / ``"continue"`` to proceed, or ``"stop"`` to halt the run.
         """
-        pass
+        return None
 
     async def on_turn_end(
         self,
@@ -122,6 +149,12 @@ class AgentHooksBase(Generic[TContext, TAgent]):
     set this on `agent.hooks` to receive events for that specific agent.
 
     Subclass and override the methods you need.
+
+    Turn-lifecycle hooks
+    --------------------
+    :meth:`on_turn_start` and :meth:`on_turn_end` fire once per iteration of the
+    agent loop.  :meth:`on_turn_start` may return ``"stop"`` to halt the run
+    gracefully before the LLM is called for that turn.
     """
 
     async def on_start(self, context: AgentHookContext[TContext], agent: TAgent) -> None:
@@ -183,15 +216,22 @@ async def on_turn_start(
         context: RunContextWrapper[TContext],
         agent: TAgent,
         turn_number: int,
-    ) -> None:
+    ) -> Union[TurnControl, None]:
         """Called at the start of each agent turn, before the LLM is invoked.
 
+        Returning ``"stop"`` halts the run gracefully before the model is called.
+        Returning ``None`` or ``"continue"`` proceeds normally.
+
         Args:
             context: The run context wrapper.
             agent: The current agent.
-            turn_number: The 1-indexed turn number (increments each time through the agent loop).
+            turn_number: The 1-indexed turn number (increments each time through the
+                agent loop).
+
+        Returns:
+            ``None`` / ``"continue"`` to proceed, or ``"stop"`` to halt the run.
         """
-        pass
+        return None
 
     async def on_turn_end(
         self,

src/agents/run.py

@@ -968,7 +968,7 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult:
 
                     logger.debug("Running agent %s (turn %s)", current_agent.name, current_turn)
 
-                    await asyncio.gather(
+                    run_hook_control, agent_hook_control = await asyncio.gather(
                         hooks.on_turn_start(context_wrapper, current_agent, current_turn),
                         (
                             current_agent.hooks.on_turn_start(
@@ -978,6 +978,14 @@ def _with_reasoning_item_id_policy(result: RunResult) -> RunResult:
                             else _coro.noop_coroutine()
                         ),
                     )
+                    if run_hook_control == "stop" or agent_hook_control == "stop":
+                        logger.debug(
+                            "Turn %s: on_turn_start hook requested stop; halting run.",
+                            current_turn,
+                        )
+                        raise MaxTurnsExceeded(
+                            f"Run halted by on_turn_start hook at turn {current_turn}"
+                        )
 
                     if session_persistence_enabled:
                         try:

src/agents/run_internal/run_loop.py

@@ -820,7 +820,7 @@ async def _save_stream_items_without_count(
                 streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                 break
 
-            await asyncio.gather(
+            run_hook_control, agent_hook_control = await asyncio.gather(
                 hooks.on_turn_start(context_wrapper, current_agent, current_turn),
                 (
                     current_agent.hooks.on_turn_start(
@@ -830,6 +830,18 @@ async def _save_stream_items_without_count(
                     else _coro.noop_coroutine()
                 ),
             )
+            if run_hook_control == "stop" or agent_hook_control == "stop":
+                logger.debug(
+                    "Turn %s: on_turn_start hook requested stop; halting run.",
+                    current_turn,
+                )
+                streamed_result._max_turns_handled = True
+                streamed_result.current_turn = current_turn
+                if run_state is not None:
+                    run_state._current_turn = current_turn
+                    run_state._current_step = None
+                streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
+                break
 
             if current_turn == 1:
                 all_input_guardrails = starting_agent.input_guardrails + (

tests/test_turn_lifecycle_hooks.py

@@ -199,3 +199,142 @@ async def test_on_turn_hooks_with_streaming() -> None:
 
     assert hooks.turn_starts == [1]
     assert hooks.turn_ends == [1]
+
+
+# ---------------------------------------------------------------------------
+# TurnControl tests: on_turn_start returning "stop" halts the loop
+# ---------------------------------------------------------------------------
+
+class StopAfterTurnRunHooks(RunHooks):
+    """Stops the run when on_turn_start is called for a turn > stop_after."""
+
+    def __init__(self, stop_after: int = 1) -> None:
+        self.stop_after = stop_after
+        self.turn_starts: list[int] = []
+        self.turn_ends: list[int] = []
+
+    async def on_turn_start(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> Optional[str]:
+        self.turn_starts.append(turn_number)
+        if turn_number > self.stop_after:
+            return "stop"
+        return None
+
+    async def on_turn_end(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> None:
+        self.turn_ends.append(turn_number)
+
+
+class StopAfterTurnAgentHooks(AgentHooks):
+    """Agent-level hooks that return 'stop' after a configurable turn."""
+
+    def __init__(self, stop_after: int = 1) -> None:
+        self.stop_after = stop_after
+        self.turn_starts: list[int] = []
+
+    async def on_turn_start(
+        self,
+        context: RunContextWrapper[TContext],
+        agent: Any,
+        turn_number: int,
+    ) -> Optional[str]:
+        self.turn_starts.append(turn_number)
+        if turn_number > self.stop_after:
+            return "stop"
+        return None
+
+
+@pytest.mark.asyncio
+async def test_run_hook_stop_halts_loop() -> None:
+    """Returning 'stop' from RunHooks.on_turn_start raises MaxTurnsExceeded before the LLM is called.
+
+    Turn 1: hook returns None → LLM executes, returns a tool call.
+    Turn 2: hook returns "stop" → MaxTurnsExceeded is raised before the LLM is called.
+    """
+    from agents.exceptions import MaxTurnsExceeded
+
+    tool = get_function_tool("my_tool", "tool_result")
+    model = FakeModel()
+    model.add_multiple_turn_outputs(
+        [
+            [get_function_tool_call("my_tool", "{}")],  # turn 1: tool call
+            [get_text_message("turn2")],                # turn 2: would be final — never reached
+        ]
+    )
+
+    hooks = StopAfterTurnRunHooks(stop_after=1)
+    agent = Agent(name="A", model=model, tools=[tool])
+
+    with pytest.raises(MaxTurnsExceeded, match="halted by on_turn_start hook"):
+        await Runner.run(agent, input="hi", hooks=hooks, max_turns=10)
+
+    # on_turn_start fires for turn 1 (None → continue) AND turn 2 (returns "stop")
+    assert hooks.turn_starts == [1, 2]
+    # on_turn_end fires for turn 1 (completed), NOT turn 2 (never ran)
+    assert hooks.turn_ends == [1]
+
+
+@pytest.mark.asyncio
+async def test_agent_hook_stop_halts_loop() -> None:
+    """Returning 'stop' from AgentHooksBase.on_turn_start also raises MaxTurnsExceeded."""
+    from agents.exceptions import MaxTurnsExceeded
+
+    tool = get_function_tool("my_tool", "tool_result")
+    model = FakeModel()
+    model.add_multiple_turn_outputs(
+        [
+            [get_function_tool_call("my_tool", "{}")],
+            [get_text_message("turn2")],
+        ]
+    )
+
+    agent_hooks = StopAfterTurnAgentHooks(stop_after=1)
+    agent = Agent(name="A", model=model, tools=[tool], hooks=agent_hooks)
+
+    with pytest.raises(MaxTurnsExceeded, match="halted by on_turn_start hook"):
+        await Runner.run(agent, input="hi", max_turns=10)
+
+    assert agent_hooks.turn_starts == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_continue_return_value_is_valid() -> None:
+    """Returning the literal 'continue' from on_turn_start is treated as proceed."""
+    model = FakeModel()
+    model.set_next_output([get_text_message("hello")])
+
+    class ExplicitContinueHooks(RunHooks):
+        async def on_turn_start(self, context: Any, agent: Any, turn_number: int) -> str:
+            return "continue"
+
+    hooks = ExplicitContinueHooks()
+    agent = Agent(name="A", model=model)
+    result = await Runner.run(agent, input="hi", hooks=hooks)
+    assert result.final_output == "hello"
+
+
+@pytest.mark.asyncio
+async def test_stop_on_first_turn_raises_max_turns() -> None:
+    """If on_turn_start returns 'stop' on turn 1, MaxTurnsExceeded is raised immediately."""
+    from agents.exceptions import MaxTurnsExceeded
+
+    model = FakeModel()
+    model.set_next_output([get_text_message("should not appear")])
+
+    class ImmediateStopHooks(RunHooks):
+        async def on_turn_start(self, context: Any, agent: Any, turn_number: int) -> str:
+            return "stop"
+
+    hooks = ImmediateStopHooks()
+    agent = Agent(name="A", model=model)
+
+    with pytest.raises(MaxTurnsExceeded, match="halted by on_turn_start hook"):
+        await Runner.run(agent, input="hi", hooks=hooks)