diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py index 667871f331..dd306d4fc9 100644 --- a/src/agents/run_internal/blocked_output.py +++ b/src/agents/run_internal/blocked_output.py @@ -366,6 +366,59 @@ def _identity_sequence_start( return None +def _structural_field(raw_item: Any, field: str) -> str | None: + """Read an exact string field without invoking payload equality or hashing hooks.""" + if type(raw_item) is dict: + values = raw_item + else: + try: + values = object.__getattribute__(raw_item, "__dict__") + except AttributeError: + return None + if type(values) is not dict: + return None + value = _exact_dict_field(values, field) + if type(value) is not str: + return None + return value + + +def _structural_item_key(item: RunItem) -> tuple[str | None, str | None, str | None]: + """Return (item_id, item_type, call_id) for structural matching after deserialization.""" + raw = getattr(item, "raw_item", None) + if raw is None: + return None, None, None + item_id = _structural_field(raw, "id") + item_type = _structural_field(raw, "type") + call_id = _structural_field(raw, "call_id") + if item_id is None: + fallback_id = getattr(item, "id", None) + if type(fallback_id) is str: + item_id = fallback_id + if item_type is None: + fallback_type = getattr(item, "type", None) + if type(fallback_type) is str: + item_type = fallback_type + return item_id, item_type, call_id + + +def _structural_sequence_start( + container: Sequence[RunItem], + sequence: Sequence[RunItem], +) -> int | None: + """Find a contiguous sequence by structural identity (item_id, item_type, call_id).""" + if not sequence or len(sequence) > len(container): + return None + sequence_keys = [_structural_item_key(item) for item in sequence] + for start in range(len(container) - len(sequence) + 1): + if all( + _structural_item_key(container[start + offset]) == key + for offset, key in enumerate(sequence_keys) + ): + return start + return None + + def _current_response_boundary( new_items: Sequence[RunItem], processed_response: ProcessedResponse | None, @@ -393,26 +446,77 @@ def _current_response_boundary( if session_start is not None: suffixes.extend(run_state._session_items[session_start:]) proven = True - if generated_start is None and session_start is None and run_state._current_turn == 1: - current_response_prefix = tuple(run_state._generated_items[: len(processed_items)]) - if len(current_response_prefix) == len(processed_items) and all( - type(actual) is type(expected) - for actual, expected in zip(current_response_prefix, processed_items, strict=False) - ): - # Serialization rebuilds item identities, but turn one has no accepted prefix. - processed_items = current_response_prefix - generated_start = 0 - session_start = 0 - suffixes.extend(run_state._generated_items) - suffixes.extend(run_state._session_items) + if generated_start is None and session_start is None and processed_items: + # Serialization rebuilds item identities. Fall back to structural + # matching (call_id, item_id, type) to locate the processed response's + # items within the generated and session item lists. + gen_structural = _structural_sequence_start(run_state._generated_items, processed_items) + sess_structural = _structural_sequence_start(run_state._session_items, processed_items) + if gen_structural is not None or sess_structural is not None: + if gen_structural is not None: + processed_items = tuple( + run_state._generated_items[ + gen_structural : gen_structural + len(processed_items) + ] + ) + generated_start = gen_structural + suffixes.extend(run_state._generated_items[generated_start:]) + if sess_structural is not None: + if gen_structural is None: + processed_items = tuple( + run_state._session_items[ + sess_structural : sess_structural + len(processed_items) + ] + ) + suffixes.extend(run_state._session_items[sess_structural:]) + # When both owners match, they hold copies of the same + # current response; the generated suffix extended above + # is canonical so independently deserialized copies + # cannot survive deduplication. The session start index + # is still retained for owner cleanup. + session_start = sess_structural proven = True + elif run_state._current_turn == 1: + # Last resort: turn one has no accepted prefix, so a type-based + # prefix match is safe even without call_id/item_id identifiers. + current_response_prefix = tuple(run_state._generated_items[: len(processed_items)]) + if len(current_response_prefix) == len(processed_items) and all( + type(actual) is type(expected) + for actual, expected in zip( + current_response_prefix, processed_items, strict=False + ) + ): + processed_items = current_response_prefix + generated_start = 0 + session_start = 0 + suffixes.extend(run_state._generated_items) + suffixes.extend(run_state._session_items) + proven = True current_items: list[RunItem] = [] - seen: set[int] = set() - for item in (*processed_items, *suffixes, *response_items): - if id(item) in seen: + seen_ids: set[int] = set() + seen_keys: set[tuple[str | None, str | None, str | None]] = set() + + def _remember(item: RunItem) -> None: + seen_ids.add(id(item)) + key = _structural_item_key(item) + if key != (None, None, None): + seen_keys.add(key) + + # Preserve distinct occurrences within the processed and supplied + # sequences; only collapse independently deserialized owner copies when + # merging the suffixes. + for item in (*processed_items, *response_items): + if id(item) not in seen_ids: + _remember(item) + current_items.append(item) + for item in suffixes: + if id(item) in seen_ids: + continue + key = _structural_item_key(item) + if key != (None, None, None) and key in seen_keys: continue - seen.add(id(item)) + _remember(item) current_items.append(item) return _CurrentResponseBoundary( items=tuple(current_items), diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index f822fefea0..594bdb3757 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -91,7 +91,11 @@ ) from agents.run_internal.oai_conversation import OpenAIServerConversationTracker from agents.run_internal.run_loop import get_new_response -from agents.run_internal.run_steps import NextStepFinalOutput, SingleStepResult +from agents.run_internal.run_steps import ( + NextStepFinalOutput, + ProcessedResponse, + SingleStepResult, +) from agents.run_internal.session_persistence import ( _collect_retry_owned_tail_serializations, persist_session_items_for_guardrail_trip, @@ -290,6 +294,43 @@ def __eq__(self, other: object) -> bool: assert equality_calls == [] +def test_boundary_preserves_distinct_items_with_same_structural_key() -> None: + """Two distinct items reusing the same id/call_id must both be retained. + + Structural deduplication must only collapse independently deserialized + owner copies, not distinct occurrences within the processed response + itself (a custom model may legitimately reuse an identifier). + """ + agent = Agent(name="test") + raw_call = { + "type": "function_call", + "name": "echo", + "arguments": "{}", + "call_id": "call-reused", + "id": "call-reused", + } + first = ToolCallItem(agent=agent, raw_item=dict(raw_call)) + second = ToolCallItem(agent=agent, raw_item=dict(raw_call)) + + processed = ProcessedResponse( + new_items=[first, second], + handoffs=[], + functions=[], + computer_actions=[], + local_shell_calls=[], + shell_calls=[], + apply_patch_calls=[], + tools_used=["echo"], + mcp_approval_requests=[], + interruptions=[], + ) + + boundary = blocked_output._current_response_boundary((), processed, None) + assert len(boundary.items) == 2 + assert first in boundary.items + assert second in boundary.items + + def test_blocked_function_batch_rejects_non_direct_typed_callers() -> None: class GenericCaller(BaseModel): type: str diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index a38424a8b3..663b219da0 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -2605,6 +2605,16 @@ def output_guardrail( state._current_turn = 2 state._current_turn_persisted_item_count = 1 restored = await RunState.from_json(agent, state.to_json()) + # Corrupt the processed response's item identifiers so structural matching + # cannot locate them within _generated_items, making the boundary unprovable. + if restored._last_processed_response: + for item in restored._last_processed_response.new_items: + raw = getattr(item, "raw_item", None) + if raw is not None and not isinstance(raw, dict): + if hasattr(raw, "call_id"): + raw.call_id = "corrupted" + if hasattr(raw, "id"): + raw.id = "corrupted" restored.approve(restored.get_interruptions()[0]) with pytest.raises(UserError, match="current response boundary cannot be proven"): @@ -2617,6 +2627,144 @@ def output_guardrail( assert tool_calls == 0 +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_serialized_approval_checkpoint_with_output_guardrail_resumes_after_earlier_turn( + mode: str, +) -> None: + """A serialized approval checkpoint with output guardrails must resume when the + interruption follows an earlier model response (turn > 1).""" + + @function_tool(name_override="normal_tool") + def normal_tool() -> str: + return "normal result" + + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + return "approval result" + + def output_guardrail( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=False) + + model = ScriptedModel( + [ + [get_function_tool_call("normal_tool", "{}", call_id="call-normal")], + [get_function_tool_call("approval_tool", "{}", call_id="call-approval")], + [get_text_message("done")], + ] + ) + agent = Agent( + name="test", + model=model, + tools=[normal_tool, approval_tool], + output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)], + ) + + first = await Runner.run(agent, "Use normal_tool then approval_tool") + assert len(first.interruptions) == 1 + + state = first.to_state() + state.approve(first.interruptions[0]) + + # Serialize and deserialize — this rebuilds all item objects, destroying + # object identity between processed_response.new_items and _generated_items. + restored = await RunState.from_json(agent, state.to_json()) + + if mode == "non_streamed": + result = await Runner.run(agent, restored, session=None) + else: + streamed = Runner.run_streamed(agent, restored, session=None) + await consume_stream(streamed) + result = streamed + + assert result.final_output == "done" + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_serialized_approval_without_owner_index_map_retains_blocked_output( + mode: str, +) -> None: + """A resumed approval checkpoint whose output guardrail trips must retain the + blocked call/output pair even when the serialized state predates the + generated/session shared-reference index map, leaving the two owners as + independently deserialized copies of the same response.""" + + @function_tool(name_override="normal_tool") + def normal_tool() -> str: + return "normal result" + + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + return "approval-secret" + + def output_guardrail( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + def stop_on_approval(_context: RunContextWrapper[Any], results: list[Any]) -> Any: + return ToolsToFinalOutputResult( + is_final_output=any(result.tool.name == "approval_tool" for result in results) + ) + + model = ScriptedModel( + [ + [get_function_tool_call("normal_tool", "{}", call_id="call-normal")], + [get_function_tool_call("approval_tool", "{}", call_id="call-approval")], + ] + ) + agent = Agent( + name="test", + model=model, + tools=[normal_tool, approval_tool], + output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)], + tool_use_behavior=stop_on_approval, + ) + + first = await Runner.run(agent, "Use normal_tool then approval_tool") + assert len(first.interruptions) == 1 + + state = first.to_state() + state.approve(first.interruptions[0]) + + # Serialize without the generated/session shared-reference index map so the + # two owners deserialize as independent copies of the same response. + state_json = state.to_json() + del state_json["generated_session_item_indexes"] + restored = await RunState.from_json(agent, state_json) + + session = SimpleListSession() + with pytest.raises(OutputGuardrailTripwireTriggered): + if mode == "non_streamed": + await Runner.run(agent, restored, session=session) + else: + streamed = Runner.run_streamed(agent, restored, session=session) + await consume_stream(streamed) + + saved_items = await session.get_items() + saved = [ + (item.get("type"), item.get("call_id")) for item in saved_items if isinstance(item, dict) + ] + assert saved.count(("function_call", "call-approval")) == 1 + assert saved.count(("function_call_output", "call-approval")) == 1 + assert "approval-secret" not in json.dumps(saved_items) + blocked_output = next( + item + for item in saved_items + if isinstance(item, dict) + and item.get("type") == "function_call_output" + and item.get("call_id") == "call-approval" + ) + assert blocked_output.get("output") == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + + @pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) @pytest.mark.parametrize("serialized", [False, True], ids=["live", "serialized"]) @pytest.mark.parametrize("attach_session", [False, True], ids=["without-session", "with-session"])