From a56b8e8613f7ce975c5edfb73dd28777d19bb532 Mon Sep 17 00:00:00 2001 From: Warren Date: Wed, 2 Sep 2026 22:59:16 -0700 Subject: [PATCH] fix(core): count replayed program items as parents with client-managed input A `program` item arriving in a run's input was only counted as a parent for a later `program_output` when the server manages the conversation. With client-managed history, a program that pauses across runs (a human-in-the-loop approval of a program-owned call, resumed with the app's own transcript as input) failed with "Model produced program_output item that does not match a parent program item" as soon as the model finished it. Pass the run's input items to `process_model_response` unconditionally and read `program` items from them in both modes, as `RunState` resume already does with durable input. The server-managed only branches, which also accept `program_output` and program caller ids as parents because the `program` itself may not be in the input there, are unchanged. Closes #4839 --- src/agents/run_internal/turn_resolution.py | 21 +++----- tests/test_programmatic_tool_calling.py | 62 +++++++++++++++++++++- 2 files changed, 69 insertions(+), 14 deletions(-) diff --git a/src/agents/run_internal/turn_resolution.py b/src/agents/run_internal/turn_resolution.py index ae9fec7619..8b3f1cfae1 100644 --- a/src/agents/run_internal/turn_resolution.py +++ b/src/agents/run_internal/turn_resolution.py @@ -1847,11 +1847,7 @@ def _coerce_approval_call( Sequence[RunItem], [deepcopy(getattr(item, "raw_item", item)) for item in original_pre_step_items], ) - classifier_server_managed_input_items = ( - deepcopy(ItemHelpers.input_to_new_input_list(original_input)) - if server_manages_conversation - else None - ) + classifier_input_items = deepcopy(ItemHelpers.input_to_new_input_list(original_input)) custom_calls_to_reconcile: list[ResponseCustomToolCall] = [] custom_call_identities: dict[str, tuple[str, str, str, str]] = {} @@ -2031,7 +2027,7 @@ def _append_reconciliation_call(call: ResponseFunctionToolCall) -> None: existing_items=classifier_existing_items, run_config=replace(run_config, tool_not_found_behavior="return_error_to_model"), server_manages_conversation=server_manages_conversation, - server_managed_input_items=classifier_server_managed_input_items, + input_items=classifier_input_items, allow_apply_patch_function_fallback=False, ) current_functions = {run.tool_call.call_id: run for run in classified.functions} @@ -2691,7 +2687,7 @@ def process_model_response( existing_items: Sequence[RunItem] | None = None, run_config: RunConfig | None = None, server_manages_conversation: bool = False, - server_managed_input_items: Sequence[Any] | None = None, + input_items: Sequence[Any] | None = None, allow_apply_patch_function_fallback: bool = True, ) -> ProcessedResponse: items: list[RunItem] = [] @@ -2730,8 +2726,11 @@ def process_model_response( hosted_mcp_tool_metadata = collect_mcp_list_tools_metadata(existing_items or ()) hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata(response.output)) + # A program replayed as input, for example when a run resumes after a + # human-in-the-loop pause with client-managed history, is a parent the + # model may still complete, so the input counts alongside this run's items. program_call_ids, completed_program_call_ids = _collect_program_parent_state( - [*(server_managed_input_items or ()), *(existing_items or ())], + [*(input_items or ()), *(existing_items or ())], server_manages_conversation=server_manages_conversation, ) _, response_completed_program_call_ids = _collect_program_parent_state(response.output) @@ -3565,11 +3564,7 @@ async def get_single_step_result_from_response( existing_items=pre_step_items, run_config=run_config, server_manages_conversation=server_manages_conversation, - server_managed_input_items=( - ItemHelpers.input_to_new_input_list(original_input) - if server_manages_conversation - else None - ), + input_items=ItemHelpers.input_to_new_input_list(original_input), ) except ModelBehaviorError: _preflight_response_invocations_after_processing_error( diff --git a/tests/test_programmatic_tool_calling.py b/tests/test_programmatic_tool_calling.py index 72c9b82c7f..1094f35e06 100644 --- a/tests/test_programmatic_tool_calling.py +++ b/tests/test_programmatic_tool_calling.py @@ -983,6 +983,26 @@ def test_process_model_response_accepts_program_output_for_retained_program() -> assert isinstance(processed.new_items[0], ToolCallOutputItem) +def test_process_model_response_accepts_program_output_for_program_in_input() -> None: + agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()]) + + processed = process_model_response( + agent=agent, + all_tools=agent.tools, + response=ModelResponse( + output=[_program_output()], + usage=Usage(), + response_id="response_1", + ), + output_schema=None, + handoffs=[], + input_items=[_program().model_dump(exclude_none=True)], + ) + + assert len(processed.new_items) == 1 + assert isinstance(processed.new_items[0], ToolCallOutputItem) + + @pytest.mark.parametrize( ("field", "value", "remove_field", "error_match"), [ @@ -1447,7 +1467,7 @@ def lookup_inventory(sku: str) -> str: output_schema=None, handoffs=[], server_manages_conversation=True, - server_managed_input_items=submitted_delta, + input_items=submitted_delta, ) assert child_type in {_raw_item_type(item.raw_item) for item in processed.new_items} @@ -1907,6 +1927,46 @@ def test_process_model_response_accepts_allowed_program_owned_shell_output( assert len(processed.new_items) == 2 +@pytest.mark.asyncio +async def test_runner_completes_a_program_replayed_as_client_managed_input() -> None: + """A run resumed from prior items finishes the program those items started. + + An application that pauses a run, for example for a human-in-the-loop + approval of a program-owned call, and resumes by passing its own history as + input has the parent program only in that input. + """ + model = ScriptedModel([[_program_output(), get_text_message("42 units are available")]]) + + @function_tool(allowed_callers=["programmatic"]) + def lookup_inventory(sku: str) -> InventoryOutput: + return InventoryOutput(sku=sku, available_units=42) + + agent = Agent( + name="inventory", + model=model, + tools=[ProgrammaticToolCallingTool(), lookup_inventory], + ) + prior_items: list[Any] = [ + {"role": "user", "content": "Check inventory"}, + _program().model_dump(exclude_none=True), + _function_call().model_dump(exclude_none=True), + { + "type": "function_call_output", + "call_id": FUNCTION_CALL_ID, + "output": '{"sku":"A-1","available_units":42}', + "caller": PROGRAM_CALLER, + }, + ] + + result = await Runner.run(agent, prior_items) + + assert result.final_output == "42 units are available" + assert [_raw_item_type(item.raw_item) for item in result.new_items] == [ + "program_output", + "message", + ] + + @pytest.mark.asyncio @pytest.mark.parametrize("streamed", [False, True]) async def test_runner_executes_and_replays_programmatic_function_calls(streamed: bool) -> None: