From e4098b0edc11f6b2377444193dc48fbc2442badc Mon Sep 17 00:00:00 2001
From: ChrisPan <39005916+szupzj18@users.noreply.github.com>
Date: Mon, 24 Aug 2026 08:12:20 +0800
Subject: [PATCH 1/6] fix: use structural matching for serialized approval
 checkpoint boundary
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When an approval checkpoint with output guardrails is serialized and
deserialized, all RunItem objects are rebuilt, destroying object identity.
The _current_response_boundary function relied on object identity to locate
the processed response's items within _generated_items and _session_items,
so the boundary could not be proven for turns after the first, raising
'Cannot resume a serialized approval checkpoint with output guardrails'.

Add _structural_item_key and _structural_sequence_start helpers that match
items by (item_id, item_type, call_id) — the same identifiers used by
_merge_generated_items_with_processed. When identity matching fails, fall
back to structural matching to locate the processed items within the
generated and session item lists. The existing turn-1 type-based prefix
fallback is retained as a last resort for items without identifiers.

Fixes #4611
---
 src/agents/run_internal/blocked_output.py | 91 ++++++++++++++++++++---
 tests/test_agent_runner_streamed.py       | 67 +++++++++++++++++
 2 files changed, 146 insertions(+), 12 deletions(-)

diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py
index 667871f331..ad6eba315d 100644
--- a/src/agents/run_internal/blocked_output.py
+++ b/src/agents/run_internal/blocked_output.py
@@ -366,6 +366,45 @@ def _identity_sequence_start(
     return None
 
 
+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."""
+    item_id: str | None = None
+    item_type: str | None = None
+    call_id: str | None = None
+    raw = getattr(item, "raw_item", None)
+    if raw is not None:
+        if isinstance(raw, dict):
+            item_id = raw.get("id")
+            item_type = raw.get("type")
+            call_id = raw.get("call_id")
+        else:
+            item_id = getattr(raw, "id", None)
+            item_type = getattr(raw, "type", None)
+            call_id = getattr(raw, "call_id", None)
+    if item_id is None:
+        item_id = getattr(item, "id", None)
+    if item_type is None:
+        item_type = getattr(item, "type", None)
+    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 (call_id, item_id, type)."""
+    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,19 +432,47 @@ 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)
+                            ]
+                        )
+                    session_start = sess_structural
+                    suffixes.extend(run_state._session_items[session_start:])
                 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()
diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py
index a38424a8b3..d5bae71bf0 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,63 @@ 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.parametrize("serialized", [False, True], ids=["live", "serialized"])
 @pytest.mark.parametrize("attach_session", [False, True], ids=["without-session", "with-session"])

From 05555f0dd959be2d147ab9e17b59d20bab1b4abc Mon Sep 17 00:00:00 2001
From: ChrisPan <39005916+szupzj18@users.noreply.github.com>
Date: Mon, 24 Aug 2026 08:36:42 +0800
Subject: [PATCH 2/6] docs: align structural sequence docstring with key tuple
 order

---
 src/agents/run_internal/blocked_output.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py
index ad6eba315d..14900906f5 100644
--- a/src/agents/run_internal/blocked_output.py
+++ b/src/agents/run_internal/blocked_output.py
@@ -392,7 +392,7 @@ def _structural_sequence_start(
     container: Sequence[RunItem],
     sequence: Sequence[RunItem],
 ) -> int | None:
-    """Find a contiguous sequence by structural identity (call_id, item_id, type)."""
+    """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]

From 9d1c551c7121b31494831a1b55931491c660fa7d Mon Sep 17 00:00:00 2001
From: ChrisPan <39005916+szupzj18@users.noreply.github.com>
Date: Mon, 24 Aug 2026 09:02:33 +0800
Subject: [PATCH 3/6] Deduplicate structurally after deserialized boundary
 resolution

After serialization, _generated_items and _session_items may contain
independently deserialized copies of the same items. The id(item)
deduplication cannot recognize those copies as equivalent, allowing
duplicate items into the boundary when structural matching appends
both suffixes.

Track structural keys (item_id, item_type, call_id) alongside object
identity in the deduplication loop. Items without identifiers fall
back to identity-only deduplication.
---
 src/agents/run_internal/blocked_output.py | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py
index 14900906f5..dc8fc325ac 100644
--- a/src/agents/run_internal/blocked_output.py
+++ b/src/agents/run_internal/blocked_output.py
@@ -475,11 +475,15 @@ def _current_response_boundary(
                     proven = True
 
     current_items: list[RunItem] = []
-    seen: set[int] = set()
+    seen_ids: set[int] = set()
+    seen_keys: set[tuple[str | None, str | None, str | None]] = set()
     for item in (*processed_items, *suffixes, *response_items):
-        if id(item) in seen:
+        key = _structural_item_key(item)
+        if id(item) in seen_ids or (key != (None, None, None) and key in seen_keys):
             continue
-        seen.add(id(item))
+        seen_ids.add(id(item))
+        if key != (None, None, None):
+            seen_keys.add(key)
         current_items.append(item)
     return _CurrentResponseBoundary(
         items=tuple(current_items),

From 24122bfc377f8c71aaab3a0318519fe6dfb1b1dc Mon Sep 17 00:00:00 2001
From: ChrisPan <39005916+szupzj18@users.noreply.github.com>
Date: Mon, 24 Aug 2026 09:35:45 +0800
Subject: [PATCH 4/6] Read structural boundary keys without payload equality or
 hash hooks

---
 src/agents/run_internal/blocked_output.py | 49 ++++++++++++++++-------
 1 file changed, 34 insertions(+), 15 deletions(-)

diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py
index dc8fc325ac..f9d76b811d 100644
--- a/src/agents/run_internal/blocked_output.py
+++ b/src/agents/run_internal/blocked_output.py
@@ -366,25 +366,39 @@ 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."""
-    item_id: str | None = None
-    item_type: str | None = None
-    call_id: str | None = None
     raw = getattr(item, "raw_item", None)
-    if raw is not None:
-        if isinstance(raw, dict):
-            item_id = raw.get("id")
-            item_type = raw.get("type")
-            call_id = raw.get("call_id")
-        else:
-            item_id = getattr(raw, "id", None)
-            item_type = getattr(raw, "type", None)
-            call_id = getattr(raw, "call_id", 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:
-        item_id = getattr(item, "id", None)
+        fallback_id = getattr(item, "id", None)
+        if type(fallback_id) is str:
+            item_id = fallback_id
     if item_type is None:
-        item_type = getattr(item, "type", None)
+        fallback_type = getattr(item, "type", None)
+        if type(fallback_type) is str:
+            item_type = fallback_type
     return item_id, item_type, call_id
 
 
@@ -454,8 +468,13 @@ def _current_response_boundary(
                                 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
-                    suffixes.extend(run_state._session_items[session_start:])
                 proven = True
             elif run_state._current_turn == 1:
                 # Last resort: turn one has no accepted prefix, so a type-based

From 81100e81c029e8ae7ed092381404e9afa88c4a41 Mon Sep 17 00:00:00 2001
From: ChrisPan <39005916+szupzj18@users.noreply.github.com>
Date: Mon, 24 Aug 2026 09:36:04 +0800
Subject: [PATCH 5/6] Test blocked output retention without owner index map

---
 tests/test_agent_runner_streamed.py | 81 +++++++++++++++++++++++++++++
 1 file changed, 81 insertions(+)

diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py
index d5bae71bf0..663b219da0 100644
--- a/tests/test_agent_runner_streamed.py
+++ b/tests/test_agent_runner_streamed.py
@@ -2684,6 +2684,87 @@ def output_guardrail(
     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"])

From 024808854fcad016a00b4413634a8a8de5e93442 Mon Sep 17 00:00:00 2001
From: ChrisPan <39005916+szupzj18@users.noreply.github.com>
Date: Mon, 24 Aug 2026 09:56:27 +0800
Subject: [PATCH 6/6] Preserve distinct items with matching structural keys in
 boundary

---
 src/agents/run_internal/blocked_output.py | 22 +++++++++---
 tests/test_agent_runner.py                | 43 ++++++++++++++++++++++-
 2 files changed, 60 insertions(+), 5 deletions(-)

diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py
index f9d76b811d..dd306d4fc9 100644
--- a/src/agents/run_internal/blocked_output.py
+++ b/src/agents/run_internal/blocked_output.py
@@ -496,13 +496,27 @@ def _current_response_boundary(
     current_items: list[RunItem] = []
     seen_ids: set[int] = set()
     seen_keys: set[tuple[str | None, str | None, str | None]] = set()
-    for item in (*processed_items, *suffixes, *response_items):
-        key = _structural_item_key(item)
-        if id(item) in seen_ids or (key != (None, None, None) and key in seen_keys):
-            continue
+
+    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
+        _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