diff --git a/src/agents/handoffs/history.py b/src/agents/handoffs/history.py index 594db69f9a..fad8555ac5 100644 --- a/src/agents/handoffs/history.py +++ b/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" diff --git a/tests/test_programmatic_tool_calling.py b/tests/test_programmatic_tool_calling.py index 4d98781fca..13beeacc82 100644 --- a/tests/test_programmatic_tool_calling.py +++ b/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)