fix: finish up human-in-the-loop port

Author: mjschock

src/agents/_run_impl.py

@@ -46,6 +46,35 @@
     from .run import RunConfig
     from .stream_events import StreamEvent
 
+# Per-process, ephemeral map linking a tool call ID to its nested
+# Agent run result within the same run; entry is removed after consumption.
+_agent_tool_run_results: dict[str, RunResult | RunResultStreaming] = {}
+
+
+def save_agent_tool_run_result(
+    tool_call: ResponseFunctionToolCall | None,
+    run_result: RunResult | RunResultStreaming,
+) -> None:
+    """Save the nested agent run result for later consumption.
+
+    This is used when an agent is used as a tool. The run result is stored
+    so that interruptions from the nested agent run can be collected.
+    """
+    if tool_call:
+        _agent_tool_run_results[tool_call.call_id] = run_result
+
+
+def consume_agent_tool_run_result(
+    tool_call: ResponseFunctionToolCall,
+) -> RunResult | RunResultStreaming | None:
+    """Consume and return the nested agent run result for a tool call.
+
+    This retrieves and removes the stored run result. Returns None if
+    no result was stored for this tool call.
+    """
+    run_result = _agent_tool_run_results.pop(tool_call.call_id, None)
+    return run_result
+
 
 @dataclass
 class ToolsToFinalOutputResult:
@@ -412,6 +441,8 @@ def as_tool(
         is_enabled: bool
         | Callable[[RunContextWrapper[Any], AgentBase[Any]], MaybeAwaitable[bool]] = True,
         on_stream: Callable[[AgentToolStreamEvent], MaybeAwaitable[None]] | None = None,
+        needs_approval: bool
+        | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]] = False,
         run_config: RunConfig | None = None,
         max_turns: int | None = None,
         hooks: RunHooks[TContext] | None = None,
@@ -441,6 +472,13 @@ def as_tool(
                 agent run. The callback receives an `AgentToolStreamEvent` containing the nested
                 agent, the originating tool call (when available), and each stream event. When
                 provided, the nested agent is executed in streaming mode.
+            needs_approval: Whether the tool needs approval before execution.
+                If True, the run will be interrupted and the tool call will need
+                to be approved using RunState.approve() or rejected using
+                RunState.reject() before continuing. Can be a bool
+                (always/never needs approval) or a function that takes
+                (run_context, tool_parameters, call_id) and returns whether this
+                specific call needs approval.
             failure_error_function: If provided, generate an error message when the tool (agent) run
                 fails. The message is sent to the LLM. If None, the exception is raised instead.
         """
@@ -449,10 +487,12 @@ def as_tool(
             name_override=tool_name or _transforms.transform_string_function_style(self.name),
             description_override=tool_description or "",
             is_enabled=is_enabled,
+            needs_approval=needs_approval,
             failure_error_function=failure_error_function,
         )
         async def run_agent(context: ToolContext, input: str) -> Any:
             from .run import DEFAULT_MAX_TURNS, Runner
+            from .tool_context import ToolContext
 
             resolved_max_turns = max_turns if max_turns is not None else DEFAULT_MAX_TURNS
             run_result: RunResult | RunResultStreaming
@@ -530,12 +570,24 @@ async def dispatch_stream_events() -> None:
                     conversation_id=conversation_id,
                     session=session,
                 )
+
+            # Store the run result keyed by tool_call_id so it can be retrieved later
+            # when the tool_call is available during result processing
+            # At runtime, context is actually a ToolContext which has tool_call_id
+            if isinstance(context, ToolContext):
+                _agent_tool_run_results[context.tool_call_id] = run_result
+
             if custom_output_extractor:
                 return await custom_output_extractor(run_result)
 
             return run_result.final_output
 
-        return run_agent
+        # Mark the function tool as an agent tool
+        run_agent_tool = run_agent
+        run_agent_tool._is_agent_tool = True
+        run_agent_tool._agent_instance = self
+
+        return run_agent_tool
 
     async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> str | None:
         if isinstance(self.instructions, str):

src/agents/agent.py

@@ -67,6 +67,9 @@ async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
 
     async def add_items(self, items: list[TResponseInputItem]) -> None:
         session_id = await self._get_session_id()
+        if not items:
+            return
+
         await self._openai_client.conversations.items.create(
             conversation_id=session_id,
             items=items,

src/agents/memory/openai_conversations_session.py

@@ -155,6 +155,13 @@ class RunResult(RunResultBase):
     )
     _last_processed_response: ProcessedResponse | None = field(default=None, repr=False)
     """The last processed model response. This is needed for resuming from interruptions."""
+    _tool_use_tracker_snapshot: dict[str, list[str]] = field(default_factory=dict, repr=False)
+    _current_turn_persisted_item_count: int = 0
+    """Number of items from new_items already persisted to session for the
+    current turn."""
+    _original_input: str | list[TResponseInputItem] | None = field(default=None, repr=False)
+    """The original input from the first turn. Unlike `input`, this is never updated during the run.
+    Used by to_state() to preserve the correct originalInput when serializing state."""
 
     def __post_init__(self) -> None:
         self._last_agent_ref = weakref.ref(self._last_agent)
@@ -204,9 +211,12 @@ def to_state(self) -> Any:
             ```
         """
         # Create a RunState from the current result
+        original_input_for_state = getattr(self, "_original_input", None)
         state = RunState(
             context=self.context_wrapper,
-            original_input=self.input,
+            original_input=original_input_for_state
+            if original_input_for_state is not None
+            else self.input,
             starting_agent=self.last_agent,
             max_turns=10,  # This will be overridden by the runner
         )
@@ -217,6 +227,8 @@ def to_state(self) -> Any:
         state._input_guardrail_results = self.input_guardrail_results
         state._output_guardrail_results = self.output_guardrail_results
         state._last_processed_response = self._last_processed_response
+        state._current_turn_persisted_item_count = self._current_turn_persisted_item_count
+        state.set_tool_use_tracker_snapshot(self._tool_use_tracker_snapshot)
 
         # If there are interruptions, set the current step
         if self.interruptions:
@@ -279,11 +291,32 @@ class RunResultStreaming(RunResultBase):
     _output_guardrails_task: asyncio.Task[Any] | None = field(default=None, repr=False)
     _stored_exception: Exception | None = field(default=None, repr=False)
 
+    _current_turn_persisted_item_count: int = 0
+    """Number of items from new_items already persisted to session for the
+    current turn."""
+
+    _stream_input_persisted: bool = False
+    """Whether the input has been persisted to the session. Prevents double-saving."""
+
+    _original_input_for_persistence: list[TResponseInputItem] = field(default_factory=list)
+    """Original turn input before session history was merged, used for
+    persistence (matches JS sessionInputOriginalSnapshot)."""
+
     # Soft cancel state
     _cancel_mode: Literal["none", "immediate", "after_turn"] = field(default="none", repr=False)
 
+    _original_input: str | list[TResponseInputItem] | None = field(default=None, repr=False)
+    """The original input from the first turn. Unlike `input`, this is never updated during the run.
+    Used by to_state() to preserve the correct originalInput when serializing state."""
+    _tool_use_tracker_snapshot: dict[str, list[str]] = field(default_factory=dict, repr=False)
+    _state: Any = field(default=None, repr=False)
+    """Internal reference to the RunState for streaming results."""
+
     def __post_init__(self) -> None:
         self._current_agent_ref = weakref.ref(self.current_agent)
+        # Store the original input at creation time (it will be set via input field)
+        if self._original_input is None:
+            self._original_input = self.input
 
     @property
     def last_agent(self) -> Agent[Any]:
@@ -508,9 +541,11 @@ def to_state(self) -> Any:
             ```
         """
         # Create a RunState from the current result
+        # Use _original_input (the input from the first turn) instead of input
+        # (which may have been updated during the run)
         state = RunState(
             context=self.context_wrapper,
-            original_input=self.input,
+            original_input=self._original_input if self._original_input is not None else self.input,
             starting_agent=self.last_agent,
             max_turns=self.max_turns,
         )
@@ -522,6 +557,8 @@ def to_state(self) -> Any:
         state._output_guardrail_results = self.output_guardrail_results
         state._current_turn = self.current_turn
         state._last_processed_response = self._last_processed_response
+        state._current_turn_persisted_item_count = self._current_turn_persisted_item_count
+        state.set_tool_use_tracker_snapshot(self._tool_use_tracker_snapshot)
 
         # If there are interruptions, set the current step
         if self.interruptions: