Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 28 additions & 2 deletions src/agents/result.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,11 +209,37 @@ def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) -
return cast(T, self.final_output)

def to_input_list(self) -> list[TResponseInputItem]:
"""Creates a new input list, merging the original input with all the new items generated."""
"""Creates a new input list, merging the original input with all the new items generated.

Prefer ``_model_input_items`` only when it has already been reduced to safe
post-handoff items. If it still contains structured tool or reasoning items,
it may just be a partial model delta and ``new_items`` remains the complete
transcript.
"""
original_items: list[TResponseInputItem] = ItemHelpers.input_to_new_input_list(self.input)
new_items: list[TResponseInputItem] = []
reasoning_item_id_policy = getattr(self, "_reasoning_item_id_policy", None)
for item in self.new_items:

model_input_items = getattr(self, "_model_input_items", None)
filtered_model_items: list[RunItem] | None = None
if isinstance(model_input_items, list):
candidate_items = list(model_input_items)
contains_structured_items = False
for item in candidate_items:
converted = run_item_to_input_item(item, reasoning_item_id_policy)
if converted is not None and converted.get("type") in {
"function_call",
"function_call_output",
"reasoning",
}:
contains_structured_items = True
break
if candidate_items and not contains_structured_items:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Fall back to new_items when model items are server-side deltas

In server-managed conversation mode (conversation_id / previous_response_id), _model_input_items is not guaranteed to be the full transcript: the run loops pass only pending_server_items deltas on later turns (src/agents/run.py around items_for_model and pending_server_items, mirrored in src/agents/run_internal/run_loop.py). This branch now prefers _model_input_items whenever those deltas contain no function_call/function_call_output/reasoning items, so to_input_list() can return only the last delta instead of all accumulated new_items, breaking replay or switching back to client-managed history after a multi-turn server-managed run.

Useful? React with 👍 / 👎.

filtered_model_items = candidate_items

source_items = filtered_model_items if filtered_model_items is not None else self.new_items

for item in source_items:
converted = run_item_to_input_item(item, reasoning_item_id_policy)
if converted is None:
continue
Expand Down
106 changes: 106 additions & 0 deletions tests/test_handoff_history_duplication.py
Original file line number Diff line number Diff line change
Expand Up @@ -365,3 +365,109 @@ def test_full_handoff_scenario_no_duplication(self):
assert len(function_call_outputs) == 0, (
"No function_call_output items should be in model input"
)


class TestToInputListAfterHandoff:
"""Tests for Issue #2258: to_input_list() produces unparsable data after handoff.

When nest_handoff_history=True, to_input_list() should use _model_input_items
(filtered) instead of new_items (unfiltered), since self.input has been replaced
with the nested summary and the raw function_call/function_call_output items in
new_items would be orphaned (no parent assistant message to own them).
"""

def test_to_input_list_uses_model_input_items_when_available(self):
"""to_input_list() should prefer _model_input_items over new_items."""
from agents import RunContextWrapper, RunResult

agent = _create_mock_agent()
tool_call = _create_tool_call_item(agent)
tool_output = _create_tool_output_item(agent)
message = _create_message_item(agent)
handoff_call = _create_handoff_call_item(agent)
handoff_output = _create_handoff_output_item(agent)

# Simulate post-handoff state:
# - input is the nested summary (replaces original)
# - new_items has ALL items (session-oriented, unfiltered)
# - _model_input_items has only message items (model-oriented, filtered)
result = RunResult(
input=[
{
"role": "assistant",
"content": (
"For context, here is the conversation so far:\n"
"<CONVERSATION HISTORY>\n"
"1. user: What's the weather?\n"
"2. function_call: get_weather\n"
"3. function_call_output: Sunny, 22°C\n"
"</CONVERSATION HISTORY>"
),
}
],
new_items=[
tool_call, # would be orphaned in output
tool_output, # would be orphaned in output
message,
handoff_call, # would be orphaned in output
handoff_output, # would be orphaned in output
],
raw_responses=[],
final_output="Done",
input_guardrail_results=[],
output_guardrail_results=[],
tool_input_guardrail_results=[],
tool_output_guardrail_results=[],
_last_agent=agent,
context_wrapper=RunContextWrapper(context=None),
interruptions=[],
)

# Set _model_input_items to the filtered set (only the message)
result._model_input_items = [message]

input_list = result.to_input_list()

# Should have: 1 summary item (from input) + 1 message item (from _model_input_items)
assert len(input_list) == 2, (
f"Expected 2 items (summary + message), got {len(input_list)}: {input_list}"
)

# Verify no orphaned function_call items
types_in_output = [
item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
for item in input_list
]
assert "function_call" not in types_in_output, (
"function_call items should not appear as orphaned entries"
)
assert "function_call_output" not in types_in_output, (
"function_call_output items should not appear as orphaned entries"
)

def test_to_input_list_falls_back_to_new_items_without_handoff(self):
"""Without handoff (no _model_input_items), to_input_list() uses new_items."""
from agents import RunContextWrapper, RunResult

agent = _create_mock_agent()
message = _create_message_item(agent)

result = RunResult(
input="Hello",
new_items=[message],
raw_responses=[],
final_output="Done",
input_guardrail_results=[],
output_guardrail_results=[],
tool_input_guardrail_results=[],
tool_output_guardrail_results=[],
_last_agent=agent,
context_wrapper=RunContextWrapper(context=None),
interruptions=[],
)

# _model_input_items defaults to empty list, so should fall back to new_items
input_list = result.to_input_list()

# Should have: 1 item from input ("Hello") + 1 message item from new_items
assert len(input_list) == 2, f"Expected 2 items, got {len(input_list)}: {input_list}"