fix: summarize programmatic transcripts during nested handoffs

src/agents/handoffs/history.py

@@ -2,6 +2,7 @@
 
 import json
 from collections import deque
+from collections.abc import Mapping
 from copy import deepcopy
 from dataclasses import replace
 from typing import TYPE_CHECKING, Any, cast
@@ -602,6 +603,8 @@ def _split_role_and_name(role_text: str) -> tuple[str, str | None]:
 
 def _should_forward_pre_item(input_item: TResponseInputItem) -> bool:
     """Return False when the previous transcript item is represented in the summary."""
+    if _is_programmatic_transcript_item(input_item):
+        return False
     role_candidate = input_item.get("role")
     if isinstance(role_candidate, str) and role_candidate == "assistant":
         return False
@@ -611,9 +614,22 @@ def _should_forward_pre_item(input_item: TResponseInputItem) -> bool:
 
 def _should_forward_new_item(input_item: TResponseInputItem) -> bool:
     """Return False for tool or side-effect items that the summary already covers."""
+    if _is_programmatic_transcript_item(input_item):
+        return False
     # Items with a role should always be forwarded.
     role_candidate = input_item.get("role")
     if isinstance(role_candidate, str) and role_candidate:
         return True
     type_candidate = input_item.get("type")
     return not (isinstance(type_candidate, str) and type_candidate in _SUMMARY_ONLY_INPUT_TYPES)
+
+
+def _is_programmatic_transcript_item(input_item: TResponseInputItem) -> bool:
+    """Return whether an item belongs to an indivisible hosted-program transcript."""
+    if input_item.get("type") in {"program", "program_output"}:
+        return True
+
+    caller = input_item.get("caller")
+    if isinstance(caller, Mapping):
+        return caller.get("type") == "program"
+    return getattr(caller, "type", None) == "program"

tests/test_programmatic_tool_calling.py

@@ -71,7 +71,7 @@
 from agents.tool_context import ToolContext
 
 from .fake_model import FakeModel
-from .test_responses import get_text_message
+from .test_responses import get_handoff_tool_call, get_text_message
 
 PROGRAM_CALL_ID = "call_program"
 FUNCTION_CALL_ID = "call_lookup"
@@ -2527,6 +2527,81 @@ def lookup_inventory(sku: str) -> InventoryOutput:
         session.close()
 
 
+@pytest.mark.asyncio
+async def test_nested_handoff_summarizes_complete_programmatic_transcript() -> None:
+    model = FakeModel()
+    delegate = Agent(name="delegate", model=model)
+    model.add_multiple_turn_outputs(
+        [
+            [_program(), _function_call()],
+            [_program_output(), get_handoff_tool_call(delegate)],
+            [get_text_message("done")],
+        ]
+    )
+
+    @function_tool(allowed_callers=["programmatic"])
+    def lookup_inventory(sku: str) -> InventoryOutput:
+        return InventoryOutput(sku=sku, available_units=42)
+
+    triage = Agent(
+        name="triage",
+        model=model,
+        handoffs=[delegate],
+        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
+    )
+    captured_inputs: list[list[Any]] = []
+
+    def capture_model_input(data: Any) -> Any:
+        captured_inputs.append(list(data.model_data.input))
+        return data.model_data
+
+    session = SQLiteSession("programmatic-tool-calling-nested-handoff")
+    try:
+        result = await Runner.run(
+            triage,
+            "Check inventory and delegate the final response.",
+            run_config=RunConfig(
+                nest_handoff_history=True,
+                call_model_input_filter=capture_model_input,
+            ),
+            session=session,
+        )
+
+        assert result.final_output == "done"
+        handoff_input = captured_inputs[-1]
+        handoff_types = [_raw_item_type(item) for item in handoff_input]
+        assert not {
+            "program",
+            "function_call",
+            "function_call_output",
+            "program_output",
+        }.intersection(handoff_types)
+
+        summary_text = "\n".join(
+            cast(str, item.get("content"))
+            for item in handoff_input
+            if isinstance(item, dict) and isinstance(item.get("content"), str)
+        )
+        assert '"type": "program"' in summary_text
+        assert '"type": "function_call"' in summary_text
+        assert '"type": "function_call_output"' in summary_text
+        assert '"type": "program_output"' in summary_text
+
+        session_items = await session.get_items()
+        assert [_raw_item_type(item) for item in session_items] == [
+            None,
+            "program",
+            "function_call",
+            "function_call_output",
+            "program_output",
+            "function_call",
+            "function_call_output",
+            "message",
+        ]
+    finally:
+        session.close()
+
+
 @pytest.mark.asyncio
 async def test_non_function_programmatic_outputs_preserve_caller() -> None:
     caller = cast(Any, PROGRAM_CALLER)