fix(core): count replayed program items as parents with client-managed input

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(

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: