fix: use structural matching for serialized approval checkpoint boundary

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),

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

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"])