Evan Mattson (moonbox3) · Sep 14, 2026 · Sep 8, 2026 · Sep 8, 2026 · Sep 9, 2026 · Sep 9, 2026
@@ -2483,7 +2483,13 @@ async def run_agent_stream(
             seeded_resume_from_snapshot = True
 
             if not config.use_service_session:
-                raw_messages = snapshot_session.resume_seeded_messages(raw_messages)
+                # Use the same overlap/reconstructor as non-resume turns so a client
+                # that replays its transcript on resume is not double-persisted (#8140).
+                raw_messages = _reconstruct_messages_from_thread_snapshot(
+                    stored_messages=stored_snapshot.messages,
+                    incoming_messages=raw_messages,
+                    stored_interrupt=stored_snapshot.interrupt,
+                )
             else:
                 provider_suffix, snapshot_seed_messages = _split_service_session_input(
                     stored_snapshot_messages=stored_snapshot.messages,

@@ -446,10 +446,17 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
             run_checkpoint_storage = _OwnedWorkflowCheckpointStorage(checkpoint_storage, request_owner)
         builder_seed_messages = raw_messages
         if resume_payload is not None or (checkpoint_id is not None and not raw_messages):
-            # Resume requests carry only the synthesized interrupt response, and a
-            # checkpoint-only resume carries no new messages at all; in both cases seed
-            # the builder with stored history to avoid persisting a truncated thread.
-            builder_seed_messages = snapshot_session.resume_seeded_messages(builder_seed_messages)
+            # Resume / checkpoint-only requests need stored history. Prefer the
+            # overlap reconstructor when a snapshot exists so client-replayed
+            # transcripts are not duplicated (#8140); otherwise prepend stored.
+            if stored_snapshot is not None:
+                builder_seed_messages = _reconstruct_messages_from_thread_snapshot(
+                    stored_messages=stored_snapshot.messages,
+                    incoming_messages=builder_seed_messages,
+                    stored_interrupt=stored_snapshot.interrupt,
+                )
+            else:
+                builder_seed_messages = snapshot_session.resume_seeded_messages(builder_seed_messages)
         snapshot_builder = _WorkflowSnapshotBuilder(builder_seed_messages) if snapshot_session.enabled else None
         if snapshot_builder is not None and effective_state:
             # Seed builder state so a run that emits no StateSnapshotEvent still

diff --git a/python/packages/ag-ui/tests/ag_ui/test_resume_transcript_dedupe.py b/python/packages/ag-ui/tests/ag_ui/test_resume_transcript_dedupe.py
@@ -0,0 +1,52 @@
+"""Regression for AG-UI resume with client-replayed transcript (#8140)."""
+
+from agent_framework_ag_ui._run_common import _reconstruct_messages_from_thread_snapshot
+
+
+def test_resume_with_replayed_transcript_does_not_duplicate_history():
+    """Clients that re-send the full transcript on resume must not double-write."""
+    stored = [
+        {"id": "u1", "role": "user", "content": "please run the tool"},
+        {
+            "id": "a1",
+            "role": "assistant",
+            "content": "",
+            "toolCalls": [{"id": "c1", "type": "function", "function": {"name": "needs_approval", "arguments": "{}"}}],
+        },
+    ]
+    # Reference AG-UI client replays the same transcript plus the resume turn.
+    incoming = [
+        *stored,
+        {"id": "u2", "role": "user", "content": "approved"},
+    ]
+
+    reconstructed = _reconstruct_messages_from_thread_snapshot(
+        stored_messages=stored,
+        incoming_messages=incoming,
+        stored_interrupt=[{"interruptId": "int-1"}],
+    )
+
+    assert [m.get("id") for m in reconstructed] == ["u1", "a1", "u2"]
+    assert len(reconstructed) == 3
+
+
+def test_resume_with_interrupt_only_still_seeds_history():
+    """Resume payloads that carry only the synthetic response still get history."""
+    stored = [
+        {"id": "u1", "role": "user", "content": "please run the tool"},
+        {
+            "id": "a1",
+            "role": "assistant",
+            "content": "",
+            "toolCalls": [{"id": "c1", "type": "function", "function": {"name": "needs_approval", "arguments": "{}"}}],
+        },
+    ]
+    incoming = [{"id": "u2", "role": "user", "content": "approved"}]
+
+    reconstructed = _reconstruct_messages_from_thread_snapshot(
+        stored_messages=stored,
+        incoming_messages=incoming,
+        stored_interrupt=[{"interruptId": "int-1"}],
+    )
+
+    assert [m.get("id") for m in reconstructed] == ["u1", "a1", "u2"]