diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py b/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py index 4fdaef188d..10a734a51a 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py @@ -2483,7 +2483,16 @@ 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) + # Empty approval resumes prepend stored history; non-empty/replayed + # transcripts overlap-merge so clients do not double-write (#8140). + if raw_messages: + raw_messages = _reconstruct_messages_from_thread_snapshot( + stored_messages=stored_snapshot.messages, + incoming_messages=raw_messages, + stored_interrupt=stored_snapshot.interrupt, + ) + else: + raw_messages = snapshot_session.resume_seeded_messages(raw_messages) else: provider_suffix, snapshot_seed_messages = _split_service_session_input( stored_snapshot_messages=stored_snapshot.messages, @@ -2492,6 +2501,13 @@ async def run_agent_stream( ) raw_messages = provider_suffix elif not config.use_service_session: + if resume_payload is not None and raw_messages: + # Client-replayed transcript on predictive/generic resume: overlap-merge + # and mark seeded so save-time resume_seeded_messages does not prepend + # again (#8140). Empty interrupt-only resumes (e.g. confirm_changes) + # must stay empty so synthesized resume tool messages are the only + # turn input; history is restored at save when this flag stays false. + seeded_resume_from_snapshot = True raw_messages = _reconstruct_messages_from_thread_snapshot( stored_messages=stored_snapshot.messages, incoming_messages=raw_messages, diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py b/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py index b0694596e7..44cffa4cc7 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_snapshot_session.py @@ -134,11 +134,17 @@ def effective_state( def resume_seeded_messages(self, incoming: list[dict[str, Any]]) -> list[dict[str, Any]]: """Prepend copies of stored thread history to a resume request's messages. - Resume requests carry only the synthesized interrupt response; seeding + Resume requests often carry only the synthesized interrupt response; seeding with stored history keeps the persisted thread from being truncated. + + For non-empty client-replayed transcripts that already overlap stored + history, callers should use ``_reconstruct_messages_from_thread_snapshot`` + instead so messages are not double-persisted (#8140). """ if self._stored is None: return incoming + if not incoming: + return [copy.deepcopy(message) for message in self._stored.messages] return [copy.deepcopy(message) for message in self._stored.messages] + incoming async def save( diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py index 73b2a1d7ec..23b9b2341e 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py @@ -446,10 +446,16 @@ 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. Empty input prepends; + # non-empty/replayed transcripts overlap-merge (#8140). + if builder_seed_messages: + builder_seed_messages = _reconstruct_messages_from_thread_snapshot( + stored_messages=stored_snapshot.messages if stored_snapshot is not None else [], + incoming_messages=builder_seed_messages, + stored_interrupt=stored_snapshot.interrupt if stored_snapshot is not None else None, + ) + 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_snapshot_session.py b/python/packages/ag-ui/tests/ag_ui/test_snapshot_session.py index d0f7522e1a..4a00e653ca 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_snapshot_session.py +++ b/python/packages/ag-ui/tests/ag_ui/test_snapshot_session.py @@ -8,6 +8,8 @@ that interface only; runner integration is covered by the existing suite. """ +from typing import Any + import pytest from ag_ui.core import ( EventType, @@ -209,6 +211,38 @@ async def test_seeded_copies_do_not_alias_stored_snapshot(self) -> None: assert session.stored is not None assert session.stored.messages[0]["content"] == "hi" + async def test_replayed_transcript_uses_reconstructor_not_blind_prepend(self) -> None: + """Client-replayed history must not be naively prepended again (#8140).""" + from agent_framework_ag_ui._run_common import _reconstruct_messages_from_thread_snapshot + + stored: list[dict[str, Any]] = [ + {"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: list[dict[str, Any]] = [ + *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 [message["id"] for message in reconstructed] == ["u1", "a1", "u2"] + # resume_seeded_messages remains blind prepend for save-time / empty seeds. + snapshot = AGUIThreadSnapshot(messages=stored, interrupt=[{"interruptId": "int-1"}]) + store = await make_store_with("user-1", "t-replay", snapshot) + session = await ThreadSnapshotSession.open(store=store, scope="user-1", thread_id="t-replay") + seeded = session.resume_seeded_messages([{"id": "u2", "role": "user", "content": "approved"}]) + assert [message["id"] for message in seeded] == ["u1", "a1", "u2"] + async def test_without_stored_snapshot_returns_incoming_unchanged(self) -> None: session = await ThreadSnapshotSession.open(store=None, scope=None, thread_id="t1") incoming = [{"id": "m2", "role": "user", "content": "hello"}]