feat: add RunState parameter support to Runner.run() methods

This commit integrates RunState into the Runner API, allowing runs to be
resumed from a saved state. This is the final piece needed to make
human-in-the-loop (HITL) tool approval fully functional.

**Changes:**

1. **Import NextStepInterruption** (run.py:21-32)
   - Added NextStepInterruption to imports from _run_impl
   - Added RunState import

2. **Updated Method Signatures** (run.py:285-444)
   - Runner.run(): Added `RunState[TContext]` to input union type
   - Runner.run_sync(): Added `RunState[TContext]` to input union type
   - Runner.run_streamed(): Added `RunState[TContext]` to input union type
   - AgentRunner.run(): Added `RunState[TContext]` to input union type
   - AgentRunner.run_sync(): Added `RunState[TContext]` to input union type
   - AgentRunner.run_streamed(): Added `RunState[TContext]` to input union type

3. **RunState Resumption Logic** (run.py:524-584)
   - Check if input is RunState instance
   - Extract state fields when resuming: current_turn, original_input,
     generated_items, model_responses, context_wrapper
   - Prime server conversation tracker from model_responses if resuming
   - Cast context_wrapper to correct type after extraction

4. **Interruption Handling** (run.py:689-726)
   - Added `interruptions=[]` to successful RunResult creation
   - Added elif branch for NextStepInterruption
   - Return RunResult with interruptions when tool approval needed
   - Set final_output to None for interrupted runs

5. **RunResultStreaming Support** (run.py:879-918)
   - Handle RunState input for streaming runs
   - Added `interruptions=[]` field to RunResultStreaming creation
   - Extract original_input from RunState for result

**How It Works:**

When resuming from RunState:
```python
run_state.approve(approval_item)

result = await Runner.run(agent, run_state)
```

When a tool needs approval:
1. Run pauses at tool execution
2. Returns RunResult with interruptions=[ToolApprovalItem(...)]
3. User can inspect interruptions and approve/reject
4. User resumes by passing RunResult back to Runner.run()

**Remaining Work:**
- Add `state` property to RunResult for creating RunState from results
- Add comprehensive tests
- Add documentation/examples

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: mjschock

src/agents/run.py

@@ -22,6 +22,7 @@
     AgentToolUseTracker,
     NextStepFinalOutput,
     NextStepHandoff,
+    NextStepInterruption,
     NextStepRunAgain,
     QueueCompleteSentinel,
     RunImpl,
@@ -65,6 +66,7 @@
 from .models.multi_provider import MultiProvider
 from .result import RunResult, RunResultStreaming
 from .run_context import RunContextWrapper, TContext
+from .run_state import RunState
 from .stream_events import (
     AgentUpdatedStreamEvent,
     RawResponsesStreamEvent,
@@ -296,7 +298,7 @@ class Runner:
     async def run(
         cls,
         starting_agent: Agent[TContext],
-        input: str | list[TResponseInputItem],
+        input: str | list[TResponseInputItem] | RunState[TContext],
         *,
         context: TContext | None = None,
         max_turns: int = DEFAULT_MAX_TURNS,
@@ -371,7 +373,7 @@ async def run(
     def run_sync(
         cls,
         starting_agent: Agent[TContext],
-        input: str | list[TResponseInputItem],
+        input: str | list[TResponseInputItem] | RunState[TContext],
         *,
         context: TContext | None = None,
         max_turns: int = DEFAULT_MAX_TURNS,
@@ -444,7 +446,7 @@ def run_sync(
     def run_streamed(
         cls,
         starting_agent: Agent[TContext],
-        input: str | list[TResponseInputItem],
+        input: str | list[TResponseInputItem] | RunState[TContext],
         context: TContext | None = None,
         max_turns: int = DEFAULT_MAX_TURNS,
         hooks: RunHooks[TContext] | None = None,
@@ -519,7 +521,7 @@ class AgentRunner:
     async def run(
         self,
         starting_agent: Agent[TContext],
-        input: str | list[TResponseInputItem],
+        input: str | list[TResponseInputItem] | RunState[TContext],
         **kwargs: Unpack[RunOptions[TContext]],
     ) -> RunResult:
         context = kwargs.get("context")
@@ -532,19 +534,41 @@ async def run(
         if run_config is None:
             run_config = RunConfig()
 
+        # Check if we're resuming from a RunState
+        is_resumed_state = isinstance(input, RunState)
+        run_state: RunState[TContext] | None = None
+
+        if is_resumed_state:
+            # Resuming from a saved state
+            run_state = cast(RunState[TContext], input)
+            original_user_input = run_state._original_input
+            prepared_input = run_state._original_input
+
+            # Override context with the state's context if not provided
+            if context is None and run_state._context is not None:
+                context = run_state._context.context
+        else:
+            # Keep original user input separate from session-prepared input
+            raw_input = cast(str | list[TResponseInputItem], input)
+            original_user_input = raw_input
+            prepared_input = await self._prepare_input_with_session(
+                raw_input, session, run_config.session_input_callback
+            )
+
         if conversation_id is not None or previous_response_id is not None:
             server_conversation_tracker = _ServerConversationTracker(
                 conversation_id=conversation_id, previous_response_id=previous_response_id
             )
         else:
             server_conversation_tracker = None
 
-        # Keep original user input separate from session-prepared input
-        original_user_input = input
-        prepared_input = await self._prepare_input_with_session(
-            input, session, run_config.session_input_callback
-        )
+        # Prime the server conversation tracker from state if resuming
+        if server_conversation_tracker is not None and is_resumed_state and run_state is not None:
+            for response in run_state._model_responses:
+                server_conversation_tracker.track_server_items(response)
 
+        # Always create a fresh tool_use_tracker
+        # (it's rebuilt from the run state if needed during execution)
         tool_use_tracker = AgentToolUseTracker()
 
         with TraceCtxManager(
@@ -554,14 +578,23 @@ async def run(
             metadata=run_config.trace_metadata,
             disabled=run_config.tracing_disabled,
         ):
-            current_turn = 0
-            original_input: str | list[TResponseInputItem] = _copy_str_or_list(prepared_input)
-            generated_items: list[RunItem] = []
-            model_responses: list[ModelResponse] = []
-
-            context_wrapper: RunContextWrapper[TContext] = RunContextWrapper(
-                context=context,  # type: ignore
-            )
+            if is_resumed_state and run_state is not None:
+                # Restore state from RunState
+                current_turn = run_state._current_turn
+                original_input = run_state._original_input
+                generated_items = run_state._generated_items
+                model_responses = run_state._model_responses
+                # Cast to the correct type since we know this is TContext
+                context_wrapper = cast(RunContextWrapper[TContext], run_state._context)
+            else:
+                # Fresh run
+                current_turn = 0
+                original_input = _copy_str_or_list(prepared_input)
+                generated_items = []
+                model_responses = []
+                context_wrapper = RunContextWrapper(
+                    context=context,  # type: ignore
+                )
 
             input_guardrail_results: list[InputGuardrailResult] = []
             tool_input_guardrail_results: list[ToolInputGuardrailResult] = []
@@ -704,6 +737,7 @@ async def run(
                                 tool_input_guardrail_results=tool_input_guardrail_results,
                                 tool_output_guardrail_results=tool_output_guardrail_results,
                                 context_wrapper=context_wrapper,
+                                interruptions=[],
                             )
                             if not any(
                                 guardrail_result.output.tripwire_triggered
@@ -712,7 +746,22 @@ async def run(
                                 await self._save_result_to_session(
                                     session, [], turn_result.new_step_items
                                 )
-
+                            return result
+                        elif isinstance(turn_result.next_step, NextStepInterruption):
+                            # Tool approval is needed - return a result with interruptions
+                            result = RunResult(
+                                input=original_input,
+                                new_items=generated_items,
+                                raw_responses=model_responses,
+                                final_output=None,
+                                _last_agent=current_agent,
+                                input_guardrail_results=input_guardrail_results,
+                                output_guardrail_results=[],
+                                tool_input_guardrail_results=tool_input_guardrail_results,
+                                tool_output_guardrail_results=tool_output_guardrail_results,
+                                context_wrapper=context_wrapper,
+                                interruptions=turn_result.next_step.interruptions,
+                            )
                             return result
                         elif isinstance(turn_result.next_step, NextStepHandoff):
                             current_agent = cast(Agent[TContext], turn_result.next_step.new_agent)
@@ -756,7 +805,7 @@ async def run(
     def run_sync(
         self,
         starting_agent: Agent[TContext],
-        input: str | list[TResponseInputItem],
+        input: str | list[TResponseInputItem] | RunState[TContext],
         **kwargs: Unpack[RunOptions[TContext]],
     ) -> RunResult:
         context = kwargs.get("context")
@@ -835,7 +884,7 @@ def run_sync(
     def run_streamed(
         self,
         starting_agent: Agent[TContext],
-        input: str | list[TResponseInputItem],
+        input: str | list[TResponseInputItem] | RunState[TContext],
         **kwargs: Unpack[RunOptions[TContext]],
     ) -> RunResultStreaming:
         context = kwargs.get("context")
@@ -869,8 +918,14 @@ def run_streamed(
             context=context  # type: ignore
         )
 
+        # Handle RunState input
+        if isinstance(input, RunState):
+            input_for_result = input._original_input
+        else:
+            input_for_result = input
+
         streamed_result = RunResultStreaming(
-            input=_copy_str_or_list(input),
+            input=_copy_str_or_list(input_for_result),
             new_items=[],
             current_agent=starting_agent,
             raw_responses=[],
@@ -885,12 +940,13 @@ def run_streamed(
             _current_agent_output_schema=output_schema,
             trace=new_trace,
             context_wrapper=context_wrapper,
+            interruptions=[],
         )
 
         # Kick off the actual agent loop in the background and return the streamed result object.
         streamed_result._run_impl_task = asyncio.create_task(
             self._start_streaming(
-                starting_input=input,
+                starting_input=input_for_result,
                 streamed_result=streamed_result,
                 starting_agent=starting_agent,
                 max_turns=max_turns,