From 6f5bd1c607565f41c175cb2549fa4653a84096e8 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Fri, 11 Sep 2026 06:58:37 +0800
Subject: [PATCH 1/8] fix(ag-ui): persist workflow HITL resume text in thread
 snapshots

When resume carries user replies only in interrupt.value with messages:[], seed the snapshot builder with those turns so hydrate keeps them (#8160).
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  | 61 +++++++++++++
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 90 +++++++++++++++++++
 2 files changed, 151 insertions(+)

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 4a20e171225..25df6733913 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -61,6 +61,61 @@
 _CHECKPOINT_REQUEST_OWNER_KEY = "ag_ui_workflow_request_owner"
 
 
+def _snapshot_messages_from_resume_value(value: Any) -> list[dict[str, Any]]:
+    """Convert a resolved workflow resume value into snapshot chat messages when user-visible.
+
+    Approval / structured tool payloads are skipped. Conversational HITL replies
+    (message lists or plain text) become replayable ``role: user`` turns for hydrate.
+    """
+    if isinstance(value, bool) or value is None:
+        return []
+    if isinstance(value, str):
+        text = value.strip()
+        return [{"role": "user", "content": text}] if text else []
+    if isinstance(value, dict):
+        # Function-approval style payloads are not chat turns.
+        if any(key in value for key in ("approved", "accepted", "functionCall", "function_call")):
+            return []
+        if value.get("role") in {"user", "assistant", "system", "tool"}:
+            return agui_messages_to_snapshot_format([_resume_message_to_agui_dict(value)])
+        return []
+    if isinstance(value, list):
+        message_like = [
+            _resume_message_to_agui_dict(item)
+            for item in value
+            if isinstance(item, dict) and item.get("role")
+        ]
+        if message_like:
+            return agui_messages_to_snapshot_format(message_like)
+    return []
+
+
+def _resume_message_to_agui_dict(message: dict[str, Any]) -> dict[str, Any]:
+    """Normalize resume message shapes (``contents`` or ``content``) for snapshot encoding."""
+    normalized = dict(message)
+    if normalized.get("content") not in (None, ""):
+        return normalized
+    contents = normalized.get("contents")
+    if isinstance(contents, list):
+        texts: list[str] = []
+        for part in contents:
+            if isinstance(part, dict) and part.get("type") in {"text", "input_text"}:
+                texts.append(str(part.get("text") or ""))
+        if texts:
+            normalized["content"] = "".join(texts)
+    return normalized
+
+
+def _snapshot_messages_from_workflow_resume(resume_payload: Any) -> list[dict[str, Any]]:
+    """Collect user-visible snapshot messages from a workflow resume payload."""
+    messages: list[dict[str, Any]] = []
+    for interrupt in _normalize_resume_interrupts(resume_payload):
+        if interrupt.get("status") not in {None, "resolved"}:
+            continue
+        messages.extend(_snapshot_messages_from_resume_value(interrupt.get("value")))
+    return messages
+
+
 def _checkpoint_id_from_input(input_data: dict[str, Any]) -> str | None:
     """Read an optional checkpoint id to resume from out of the AG-UI forwarded props."""
     forwarded_props = input_data.get("forwarded_props") or input_data.get("forwardedProps")
@@ -472,6 +527,12 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
             # 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)
+        if resume_payload is not None and snapshot_session.enabled:
+            # Conversational HITL resumes put the user reply in interrupt.value with
+            # messages:[]; fold that text into the snapshot so hydrate keeps it (#8160).
+            hitl_messages = _snapshot_messages_from_workflow_resume(resume_payload)
+            if hitl_messages:
+                builder_seed_messages = [*builder_seed_messages, *hitl_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_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index 4a4ec411605..455ddc7219e 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -15,6 +15,7 @@
     WorkflowContext,
     executor,
     handler,
+    response_handler,
 )
 
 from agent_framework_ag_ui import AgentFrameworkWorkflow
@@ -359,3 +360,92 @@ async def test_workflow_checkpoint_only_resume_preserves_thread_snapshot() -> No
     assert "Earlier reply" in contents
     # ...plus the newly produced output from the resumed run.
     assert any(isinstance(content, str) and "done" in content for content in contents)
+
+
+async def test_workflow_hitl_resume_persists_user_text_in_thread_snapshot() -> None:
+    """HITL resume with messages:[] must still record the user reply in the snapshot (#8160)."""
+    from agent_framework import Message
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
+
+    class MessageRequestExecutor(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="message_request_executor")
+
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+            del message
+            await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
+
+        @response_handler
+        async def handle_user_input(
+            self, original_request: dict, response: list[Message], ctx: WorkflowContext
+        ) -> None:
+            del original_request
+            user_text = response[0].text if response else ""
+            await ctx.yield_output(f"Captured response: {user_text}")  # type: ignore[arg-type]
+
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=MessageRequestExecutor()).build()
+    store = InMemoryAGUIThreadSnapshotStore()
+    agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store, checkpoint_storage=storage)
+
+    first_events = await _run(
+        agent,
+        {
+            "thread_id": "thread-hitl",
+            "run_id": "run-1",
+            "messages": [{"id": "user-1", "role": "user", "content": "start"}],
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in first_events]
+
+    await store.save(
+        scope="tenant-a",
+        thread_id="thread-hitl",
+        snapshot=AGUIThreadSnapshot(
+            messages=[
+                {"id": "user-1", "role": "user", "content": "start"},
+                {"id": "assistant-1", "role": "assistant", "content": "Need more detail"},
+            ],
+            state=None,
+            interrupt=None,
+        ),
+    )
+
+    resumed_events = await _run(
+        agent,
+        {
+            "thread_id": "thread-hitl",
+            "run_id": "run-2",
+            "messages": [],
+            "resume": {
+                "interrupts": [
+                    {
+                        "id": "handoff-user-input",
+                        "value": [
+                            {
+                                "role": "user",
+                                "contents": [{"type": "text", "text": "Please ship a replacement instead."}],
+                            }
+                        ],
+                    }
+                ]
+            },
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in resumed_events]
+
+    snapshot = await store.get(scope="tenant-a", thread_id="thread-hitl")
+    assert snapshot is not None
+    contents = [message.get("content") for message in snapshot.messages]
+    assert "start" in contents
+    assert any(isinstance(content, str) and "replacement" in content for content in contents)
+    assert any(
+        message.get("role") == "user"
+        and isinstance(message.get("content"), str)
+        and "replacement" in message["content"]
+        for message in snapshot.messages
+    )

From b04c2ea217b97655b3fe959fc5f13bb713f8ecc8 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Fri, 11 Sep 2026 13:10:33 +0800
Subject: [PATCH 2/8] fix(ag-ui): skip approval strings and dedupe HITL
 snapshot turns

Address Copilot review on #8273: do not hydrate bare approved/rejected resume strings as user messages, and avoid duplicating turns already present in request messages.
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  | 41 ++++++++++++++++++-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 24 +++++++++++
 2 files changed, 63 insertions(+), 2 deletions(-)

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 25df6733913..f1dccc26acd 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -59,6 +59,14 @@
 
 _REQUEST_OWNER_ATTRIBUTE = "_ag_ui_request_owner"
 _CHECKPOINT_REQUEST_OWNER_KEY = "ag_ui_workflow_request_owner"
+_APPROVAL_RESUME_STRINGS = frozenset({
+    "approved",
+    "rejected",
+    "accepted",
+    "denied",
+    "true",
+    "false",
+})
 
 
 def _snapshot_messages_from_resume_value(value: Any) -> list[dict[str, Any]]:
@@ -71,7 +79,11 @@ def _snapshot_messages_from_resume_value(value: Any) -> list[dict[str, Any]]:
         return []
     if isinstance(value, str):
         text = value.strip()
-        return [{"role": "user", "content": text}] if text else []
+        # Legacy request_info(str) resumes often send bare "approved"/"rejected"
+        # strings — those are not conversational HITL turns (#8160 / Copilot).
+        if not text or text.casefold() in _APPROVAL_RESUME_STRINGS:
+            return []
+        return [{"role": "user", "content": text}]
     if isinstance(value, dict):
         # Function-approval style payloads are not chat turns.
         if any(key in value for key in ("approved", "accepted", "functionCall", "function_call")):
@@ -116,6 +128,27 @@ def _snapshot_messages_from_workflow_resume(resume_payload: Any) -> list[dict[st
     return messages
 
 
+def _message_identity(message: dict[str, Any]) -> tuple[Any, ...]:
+    """Stable identity for deduping resume-synthesized turns against request messages."""
+    return (message.get("role"), message.get("id"), message.get("content"))
+
+
+def _append_unique_snapshot_messages(
+    existing: list[dict[str, Any]],
+    incoming: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Append resume-derived turns that are not already present in the seed."""
+    seen = {_message_identity(message) for message in existing}
+    merged = list(existing)
+    for message in incoming:
+        identity = _message_identity(message)
+        if identity in seen:
+            continue
+        seen.add(identity)
+        merged.append(message)
+    return merged
+
+
 def _checkpoint_id_from_input(input_data: dict[str, Any]) -> str | None:
     """Read an optional checkpoint id to resume from out of the AG-UI forwarded props."""
     forwarded_props = input_data.get("forwarded_props") or input_data.get("forwardedProps")
@@ -530,9 +563,13 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
         if resume_payload is not None and snapshot_session.enabled:
             # Conversational HITL resumes put the user reply in interrupt.value with
             # messages:[]; fold that text into the snapshot so hydrate keeps it (#8160).
+            # Skip when the client already included the same turn in `messages`.
             hitl_messages = _snapshot_messages_from_workflow_resume(resume_payload)
             if hitl_messages:
-                builder_seed_messages = [*builder_seed_messages, *hitl_messages]
+                builder_seed_messages = _append_unique_snapshot_messages(
+                    builder_seed_messages,
+                    hitl_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_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index 455ddc7219e..6fcf2d4e1db 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -449,3 +449,27 @@ async def handle_user_input(
         and "replacement" in message["content"]
         for message in snapshot.messages
     )
+
+
+def test_snapshot_messages_from_resume_skips_approval_strings() -> None:
+    from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
+
+    assert _snapshot_messages_from_resume_value("approved") == []
+    assert _snapshot_messages_from_resume_value("rejected") == []
+    assert _snapshot_messages_from_resume_value("Please refund me") == [
+        {"role": "user", "content": "Please refund me"}
+    ]
+
+
+def test_append_unique_snapshot_messages_dedupes_client_replay() -> None:
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
+
+    existing = [{"id": "u1", "role": "user", "content": "already present"}]
+    incoming = [
+        {"id": "u1", "role": "user", "content": "already present"},
+        {"id": "u2", "role": "user", "content": "new reply"},
+    ]
+    assert _append_unique_snapshot_messages(existing, incoming) == [
+        {"id": "u1", "role": "user", "content": "already present"},
+        {"id": "u2", "role": "user", "content": "new reply"},
+    ]

From 3953169b327a343cf41b3c98f926491ebcd82c69 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Mon, 14 Sep 2026 19:57:12 +0800
Subject: [PATCH 3/8] fix(ag-ui): harden HITL snapshot projection for resume

- Hashable multimodal content identity for resume dedupe
- Project only user turns from resume payloads
- Skip approvals via pending request type/data, not response text
- Count-aware role/content fallback when message IDs differ
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  | 108 +++++++++++++-----
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  |  47 +++++++-
 2 files changed, 126 insertions(+), 29 deletions(-)

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 f1dccc26acd..7bfe63b7708 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -4,9 +4,11 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import uuid
-from collections.abc import AsyncGenerator, Callable
+from collections import Counter
+from collections.abc import AsyncGenerator, Callable, Mapping
 from typing import Any, cast
 
 from ag_ui.core import (
@@ -59,43 +61,70 @@
 
 _REQUEST_OWNER_ATTRIBUTE = "_ag_ui_request_owner"
 _CHECKPOINT_REQUEST_OWNER_KEY = "ag_ui_workflow_request_owner"
-_APPROVAL_RESUME_STRINGS = frozenset({
-    "approved",
-    "rejected",
-    "accepted",
-    "denied",
-    "true",
-    "false",
-})
 
 
-def _snapshot_messages_from_resume_value(value: Any) -> list[dict[str, Any]]:
+def _hashable_message_content(content: Any) -> Any:
+    """Return a hashable, order-stable form of snapshot message content."""
+    if isinstance(content, (str, int, float, bool)) or content is None:
+        return content
+    try:
+        return json.dumps(content, sort_keys=True, default=str)
+    except TypeError:
+        return repr(content)
+
+
+def _pending_request_is_approval(pending_request: Any | None) -> bool:
+    """Whether a pending request_info event is an approval gate (not conversational HITL)."""
+    if pending_request is None:
+        return False
+    response_type: Any | None
+    try:
+        response_type = pending_request.response_type
+    except Exception:
+        response_type = getattr(pending_request, "_response_type", None)
+    if response_type is bool:
+        return True
+    type_name = getattr(response_type, "__name__", "") or str(response_type or "")
+    if "Approval" in type_name:
+        return True
+    data = getattr(pending_request, "data", None)
+    if isinstance(data, dict) and any(key in data for key in ("functionCall", "function_call")):
+        return True
+    return False
+
+
+def _snapshot_messages_from_resume_value(
+    value: Any,
+    *,
+    pending_request: Any | None = None,
+) -> list[dict[str, Any]]:
     """Convert a resolved workflow resume value into snapshot chat messages when user-visible.
 
-    Approval / structured tool payloads are skipped. Conversational HITL replies
-    (message lists or plain text) become replayable ``role: user`` turns for hydrate.
+    Approval / structured tool payloads are skipped based on the matched pending request's
+    type/data (not the response text). Only ``user`` turns are projected so resume cannot
+    forge assistant/system/tool history into the backend-owned snapshot.
     """
     if isinstance(value, bool) or value is None:
         return []
+    if _pending_request_is_approval(pending_request):
+        return []
     if isinstance(value, str):
         text = value.strip()
-        # Legacy request_info(str) resumes often send bare "approved"/"rejected"
-        # strings — those are not conversational HITL turns (#8160 / Copilot).
-        if not text or text.casefold() in _APPROVAL_RESUME_STRINGS:
+        if not text:
             return []
         return [{"role": "user", "content": text}]
     if isinstance(value, dict):
         # Function-approval style payloads are not chat turns.
         if any(key in value for key in ("approved", "accepted", "functionCall", "function_call")):
             return []
-        if value.get("role") in {"user", "assistant", "system", "tool"}:
+        if value.get("role") == "user":
             return agui_messages_to_snapshot_format([_resume_message_to_agui_dict(value)])
         return []
     if isinstance(value, list):
         message_like = [
             _resume_message_to_agui_dict(item)
             for item in value
-            if isinstance(item, dict) and item.get("role")
+            if isinstance(item, dict) and item.get("role") == "user"
         ]
         if message_like:
             return agui_messages_to_snapshot_format(message_like)
@@ -118,33 +147,57 @@ def _resume_message_to_agui_dict(message: dict[str, Any]) -> dict[str, Any]:
     return normalized
 
 
-def _snapshot_messages_from_workflow_resume(resume_payload: Any) -> list[dict[str, Any]]:
+def _snapshot_messages_from_workflow_resume(
+    resume_payload: Any,
+    pending_events: Mapping[str, Any] | None = None,
+) -> list[dict[str, Any]]:
     """Collect user-visible snapshot messages from a workflow resume payload."""
     messages: list[dict[str, Any]] = []
+    pending = pending_events or {}
     for interrupt in _normalize_resume_interrupts(resume_payload):
         if interrupt.get("status") not in {None, "resolved"}:
             continue
-        messages.extend(_snapshot_messages_from_resume_value(interrupt.get("value")))
+        interrupt_id = interrupt.get("id")
+        pending_request = pending.get(str(interrupt_id)) if interrupt_id is not None else None
+        messages.extend(
+            _snapshot_messages_from_resume_value(interrupt.get("value"), pending_request=pending_request)
+        )
     return messages
 
 
 def _message_identity(message: dict[str, Any]) -> tuple[Any, ...]:
     """Stable identity for deduping resume-synthesized turns against request messages."""
-    return (message.get("role"), message.get("id"), message.get("content"))
+    return (message.get("role"), message.get("id"), _hashable_message_content(message.get("content")))
+
+
+def _message_content_identity(message: dict[str, Any]) -> tuple[Any, ...]:
+    """Role+content identity used when message IDs differ across messages vs resume."""
+    return (message.get("role"), _hashable_message_content(message.get("content")))
 
 
 def _append_unique_snapshot_messages(
     existing: list[dict[str, Any]],
     incoming: list[dict[str, Any]],
 ) -> list[dict[str, Any]]:
-    """Append resume-derived turns that are not already present in the seed."""
-    seen = {_message_identity(message) for message in existing}
+    """Append resume-derived turns that are not already present in the seed.
+
+    Prefer id equality; when IDs differ (resume synthesizes a new id), fall back to a
+    count-aware role/content match so client-replayed turns are not persisted twice.
+    """
+    seen_ids = {message.get("id") for message in existing if message.get("id")}
+    remaining_content = Counter(_message_content_identity(message) for message in existing)
     merged = list(existing)
     for message in incoming:
-        identity = _message_identity(message)
-        if identity in seen:
+        message_id = message.get("id")
+        if message_id and message_id in seen_ids:
             continue
-        seen.add(identity)
+        content_key = _message_content_identity(message)
+        if remaining_content[content_key] > 0:
+            remaining_content[content_key] -= 1
+            continue
+        if message_id:
+            seen_ids.add(message_id)
+        remaining_content[content_key] += 1
         merged.append(message)
     return merged
 
@@ -564,7 +617,10 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
             # Conversational HITL resumes put the user reply in interrupt.value with
             # messages:[]; fold that text into the snapshot so hydrate keeps it (#8160).
             # Skip when the client already included the same turn in `messages`.
-            hitl_messages = _snapshot_messages_from_workflow_resume(resume_payload)
+            hitl_messages = _snapshot_messages_from_workflow_resume(
+                resume_payload,
+                pending_events=live_pending_events,
+            )
             if hitl_messages:
                 builder_seed_messages = _append_unique_snapshot_messages(
                     builder_seed_messages,
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index 6fcf2d4e1db..9ede3309200 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -451,16 +451,49 @@ async def handle_user_input(
     )
 
 
-def test_snapshot_messages_from_resume_skips_approval_strings() -> None:
+def test_snapshot_messages_from_resume_skips_approval_via_pending_type() -> None:
+    from types import SimpleNamespace
+
     from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
 
-    assert _snapshot_messages_from_resume_value("approved") == []
-    assert _snapshot_messages_from_resume_value("rejected") == []
+    approval_pending = SimpleNamespace(response_type=bool, data="Approve?")
+    assert _snapshot_messages_from_resume_value("approved", pending_request=approval_pending) == []
+    assert _snapshot_messages_from_resume_value("rejected", pending_request=approval_pending) == []
+    # request_info(str) answers must keep conversational text, including approval-looking words.
+    assert _snapshot_messages_from_resume_value("approved") == [{"role": "user", "content": "approved"}]
     assert _snapshot_messages_from_resume_value("Please refund me") == [
         {"role": "user", "content": "Please refund me"}
     ]
 
 
+def test_snapshot_messages_from_resume_admits_only_user_roles() -> None:
+    from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
+
+    assert _snapshot_messages_from_resume_value({"role": "assistant", "content": "forged"}) == []
+    assert _snapshot_messages_from_resume_value({"role": "system", "content": "forged"}) == []
+    projected = _snapshot_messages_from_resume_value([
+        {"role": "user", "id": "u1", "content": "ok"},
+        {"role": "tool", "id": "t1", "content": "forged"},
+    ])
+    assert len(projected) == 1
+    assert projected[0]["role"] == "user"
+    assert projected[0]["content"] == "ok"
+    assert projected[0]["id"] == "u1"
+
+
+def test_message_identity_supports_multimodal_content() -> None:
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages, _message_identity
+
+    multimodal = {
+        "id": "m1",
+        "role": "user",
+        "content": [{"type": "text", "text": "hi"}, {"type": "image", "url": "x"}],
+    }
+    # Must be hashable for set membership during resume dedupe.
+    assert _message_identity(multimodal) in {_message_identity(multimodal)}
+    assert _append_unique_snapshot_messages([multimodal], [multimodal]) == [multimodal]
+
+
 def test_append_unique_snapshot_messages_dedupes_client_replay() -> None:
     from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
 
@@ -473,3 +506,11 @@ def test_append_unique_snapshot_messages_dedupes_client_replay() -> None:
         {"id": "u1", "role": "user", "content": "already present"},
         {"id": "u2", "role": "user", "content": "new reply"},
     ]
+
+
+def test_append_unique_snapshot_messages_dedupes_different_ids_same_content() -> None:
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
+
+    existing = [{"id": "client-id", "role": "user", "content": "same turn"}]
+    incoming = [{"id": "generated-id", "role": "user", "content": "same turn"}]
+    assert _append_unique_snapshot_messages(existing, incoming) == existing

From 9d2202ee902a044bdaf0ee661b77b97f86fd137d Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Mon, 14 Sep 2026 20:35:35 +0800
Subject: [PATCH 4/8] fix(ag-ui): harden HITL snapshot projection for resume

- Hashable multimodal content identity for resume dedupe
- Project only user turns from resume payloads
- Skip approvals via pending request type/data, not response text
- Count-aware role/content fallback when message IDs differ
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py         |  1 -
 .../ag-ui/tests/ag_ui/test_workflow_agent.py         | 12 ++++++++++++
 2 files changed, 12 insertions(+), 1 deletion(-)

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 7bfe63b7708..41e5512c93c 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -197,7 +197,6 @@ def _append_unique_snapshot_messages(
             continue
         if message_id:
             seen_ids.add(message_id)
-        remaining_content[content_key] += 1
         merged.append(message)
     return merged
 
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index 9ede3309200..6fd3e168162 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -514,3 +514,15 @@ def test_append_unique_snapshot_messages_dedupes_different_ids_same_content() ->
     existing = [{"id": "client-id", "role": "user", "content": "same turn"}]
     incoming = [{"id": "generated-id", "role": "user", "content": "same turn"}]
     assert _append_unique_snapshot_messages(existing, incoming) == existing
+
+
+def test_append_unique_snapshot_messages_keeps_intentional_repeated_replies() -> None:
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
+
+    existing = [{"id": "u0", "role": "user", "content": "hello"}]
+    incoming = [
+        {"id": "r1", "role": "user", "content": "repeat"},
+        {"id": "r2", "role": "user", "content": "repeat"},
+    ]
+    merged = _append_unique_snapshot_messages(existing, incoming)
+    assert [m["id"] for m in merged] == ["u0", "r1", "r2"]

From 3dd94f25276610d32e0aede7844738d34fdd7731 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Wed, 16 Sep 2026 09:45:16 +0800
Subject: [PATCH 5/8] fix(ag-ui): limit HITL content dedupe to client-replay
 overlap

Role/content fallback for resume-synthesized turns only counts against the client-supplied messages list, so a second identical HITL reply on messages:[] is not collapsed against full thread history.
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  |  15 ++-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 106 +++++++++++++++++-
 2 files changed, 117 insertions(+), 4 deletions(-)

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 fa358669876..4c103f24750 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -184,14 +184,19 @@ def _message_content_identity(message: dict[str, Any]) -> tuple[Any, ...]:
 def _append_unique_snapshot_messages(
     existing: list[dict[str, Any]],
     incoming: list[dict[str, Any]],
+    *,
+    content_dedupe_against: list[dict[str, Any]] | None = None,
 ) -> list[dict[str, Any]]:
     """Append resume-derived turns that are not already present in the seed.
 
-    Prefer id equality; when IDs differ (resume synthesizes a new id), fall back to a
-    count-aware role/content match so client-replayed turns are not persisted twice.
+    Prefer id equality against ``existing``. Role/content fallback is limited to
+    ``content_dedupe_against`` (confirmed client-replay overlap). When that list is
+    empty or omitted, identical replies across separate HITL turns on ``messages: []``
+    resumes are kept rather than collapsed against full thread history.
     """
     seen_ids = {message.get("id") for message in existing if message.get("id")}
-    remaining_content = Counter(_message_content_identity(message) for message in existing)
+    content_source = content_dedupe_against if content_dedupe_against is not None else []
+    remaining_content = Counter(_message_content_identity(message) for message in content_source)
     merged = list(existing)
     for message in incoming:
         message_id = message.get("id")
@@ -677,6 +682,9 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
         run_id = str(input_data.get("run_id") or input_data.get("runId") or uuid.uuid4())
         snapshot_scope = cast(str | None, input_data.get(_SNAPSHOT_SCOPE_INPUT_KEY))
         raw_messages = list(cast(list[dict[str, Any]], input_data.get("messages", []) or []))
+        # Preserve the client-supplied transcript for content-only dedupe of HITL
+        # resume turns. Stored history is not a confirmed client-replay overlap.
+        client_request_messages = list(raw_messages)
         resume_payload = _extract_resume_payload(input_data)
         snapshot_session = await ThreadSnapshotSession.open(
             store=self.snapshot_store,
@@ -793,6 +801,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                 builder_seed_messages = _append_unique_snapshot_messages(
                     builder_seed_messages,
                     hitl_messages,
+                    content_dedupe_against=client_request_messages,
                 )
         snapshot_builder = _WorkflowSnapshotBuilder(builder_seed_messages) if snapshot_session.enabled else None
         if snapshot_builder is not None and effective_state:
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index 3058cd40b14..b63e9324ffb 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -511,6 +511,83 @@ async def handle_user_input(
     )
 
 
+async def test_workflow_hitl_resume_keeps_repeated_yes_on_empty_messages() -> None:
+    """A second HITL 'yes' with messages:[] must not be dropped as a content duplicate."""
+    from agent_framework import Message
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
+
+    class MessageRequestExecutor(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="message_request_executor")
+
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+            del message
+            await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
+
+        @response_handler
+        async def handle_user_input(
+            self, original_request: dict, response: list[Message], ctx: WorkflowContext
+        ) -> None:
+            del original_request
+            user_text = response[0].text if response else ""
+            await ctx.yield_output(f"Captured response: {user_text}")  # type: ignore[arg-type]
+
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=MessageRequestExecutor()).build()
+    store = InMemoryAGUIThreadSnapshotStore()
+    agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store, checkpoint_storage=storage)
+
+    await store.save(
+        scope="tenant-a",
+        thread_id="thread-hitl-yes",
+        snapshot=AGUIThreadSnapshot(
+            messages=[
+                {"id": "user-1", "role": "user", "content": "start"},
+                {"id": "assistant-1", "role": "assistant", "content": "confirm?"},
+                {"id": "user-yes-1", "role": "user", "content": "yes"},
+                {"id": "assistant-2", "role": "assistant", "content": "confirm again?"},
+            ],
+            state=None,
+            interrupt=None,
+        ),
+    )
+
+    resumed_events = await _run(
+        agent,
+        {
+            "thread_id": "thread-hitl-yes",
+            "run_id": "run-yes-2",
+            "messages": [],
+            "resume": {
+                "interrupts": [
+                    {
+                        "id": "handoff-user-input",
+                        "value": [
+                            {
+                                "role": "user",
+                                "contents": [{"type": "text", "text": "yes"}],
+                            }
+                        ],
+                    }
+                ]
+            },
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in resumed_events]
+
+    snapshot = await store.get(scope="tenant-a", thread_id="thread-hitl-yes")
+    assert snapshot is not None
+    yes_turns = [
+        message
+        for message in snapshot.messages
+        if message.get("role") == "user" and message.get("content") == "yes"
+    ]
+    assert len(yes_turns) >= 2
+
+
 def test_snapshot_messages_from_resume_skips_approval_via_pending_type() -> None:
     from types import SimpleNamespace
 
@@ -573,7 +650,15 @@ def test_append_unique_snapshot_messages_dedupes_different_ids_same_content() ->
 
     existing = [{"id": "client-id", "role": "user", "content": "same turn"}]
     incoming = [{"id": "generated-id", "role": "user", "content": "same turn"}]
-    assert _append_unique_snapshot_messages(existing, incoming) == existing
+    # Content fallback only applies against confirmed client-replay overlap.
+    assert (
+        _append_unique_snapshot_messages(
+            existing,
+            incoming,
+            content_dedupe_against=existing,
+        )
+        == existing
+    )
 
 
 def test_append_unique_snapshot_messages_keeps_intentional_repeated_replies() -> None:
@@ -586,3 +671,22 @@ def test_append_unique_snapshot_messages_keeps_intentional_repeated_replies() ->
     ]
     merged = _append_unique_snapshot_messages(existing, incoming)
     assert [m["id"] for m in merged] == ["u0", "r1", "r2"]
+
+
+def test_append_unique_snapshot_messages_keeps_second_hitl_yes_without_client_replay() -> None:
+    """messages:[] HITL resumes must not collapse a later identical reply against history."""
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
+
+    history = [
+        {"id": "u0", "role": "user", "content": "start"},
+        {"id": "a0", "role": "assistant", "content": "confirm?"},
+        {"id": "u1", "role": "user", "content": "yes"},
+        {"id": "a1", "role": "assistant", "content": "confirm again?"},
+    ]
+    second_yes = [{"id": "generated-yes-2", "role": "user", "content": "yes"}]
+    merged = _append_unique_snapshot_messages(
+        history,
+        second_yes,
+        content_dedupe_against=[],
+    )
+    assert [m["id"] for m in merged] == ["u0", "a0", "u1", "a1", "generated-yes-2"]

From 5867e5181e7421ef8186d101cdcbd4225c3fbbbd Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Thu, 17 Sep 2026 13:42:42 +0800
Subject: [PATCH 6/8] test(ag-ui): type HITL WorkflowContext and sort imports

Annotate response handlers as WorkflowContext[Any, str] so Python 3.14 validation and ty accept yield_output(str), and insert the ruff-required blank line between agent_framework and ag-ui imports.
---
 .../ag-ui/tests/ag_ui/test_workflow_agent.py       | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index b63e9324ffb..ad99e702178 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -425,6 +425,7 @@ async def finalizer(message: str, ctx: WorkflowContext[None, str]) -> None:
 async def test_workflow_hitl_resume_persists_user_text_in_thread_snapshot() -> None:
     """HITL resume with messages:[] must still record the user reply in the snapshot (#8160)."""
     from agent_framework import Message
+
     from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
     from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
 
@@ -433,17 +434,17 @@ def __init__(self) -> None:
             super().__init__(id="message_request_executor")
 
         @handler
-        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
             del message
             await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
 
         @response_handler
         async def handle_user_input(
-            self, original_request: dict, response: list[Message], ctx: WorkflowContext
+            self, original_request: dict, response: list[Message], ctx: WorkflowContext[Any, str]
         ) -> None:
             del original_request
             user_text = response[0].text if response else ""
-            await ctx.yield_output(f"Captured response: {user_text}")  # type: ignore[arg-type]
+            await ctx.yield_output(f"Captured response: {user_text}")
 
     storage = InMemoryCheckpointStorage()
     workflow = WorkflowBuilder(start_executor=MessageRequestExecutor()).build()
@@ -514,6 +515,7 @@ async def handle_user_input(
 async def test_workflow_hitl_resume_keeps_repeated_yes_on_empty_messages() -> None:
     """A second HITL 'yes' with messages:[] must not be dropped as a content duplicate."""
     from agent_framework import Message
+
     from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
     from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
 
@@ -522,17 +524,17 @@ def __init__(self) -> None:
             super().__init__(id="message_request_executor")
 
         @handler
-        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
             del message
             await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
 
         @response_handler
         async def handle_user_input(
-            self, original_request: dict, response: list[Message], ctx: WorkflowContext
+            self, original_request: dict, response: list[Message], ctx: WorkflowContext[Any, str]
         ) -> None:
             del original_request
             user_text = response[0].text if response else ""
-            await ctx.yield_output(f"Captured response: {user_text}")  # type: ignore[arg-type]
+            await ctx.yield_output(f"Captured response: {user_text}")
 
     storage = InMemoryCheckpointStorage()
     workflow = WorkflowBuilder(start_executor=MessageRequestExecutor()).build()

From b327659f33dd035dfbfa2b340a6e6fb062642d1d Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Thu, 17 Sep 2026 16:43:12 +0800
Subject: [PATCH 7/8] fix(ag-ui): content-dedupe only current-turn client
 messages

Filter content fallback to client rows whose ids are not already in the stored snapshot, so a replayed prior 'yes' cannot consume a later resume interrupt with the same text. Move HITL test executor to module scope for Python 3.14 annotation validation.
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  |  24 ++-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 147 +++++++++++++-----
 2 files changed, 127 insertions(+), 44 deletions(-)

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 62f0098ba44..6c8e1066141 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -190,9 +190,11 @@ def _append_unique_snapshot_messages(
     """Append resume-derived turns that are not already present in the seed.
 
     Prefer id equality against ``existing``. Role/content fallback is limited to
-    ``content_dedupe_against`` (confirmed client-replay overlap). When that list is
-    empty or omitted, identical replies across separate HITL turns on ``messages: []``
-    resumes are kept rather than collapsed against full thread history.
+    ``content_dedupe_against`` (current-turn client overlap / id remaps). Callers
+    must not pass replayed prior transcript rows here — those keep their ids and
+    would otherwise let an earlier user ``"yes"`` consume a later resume interrupt
+    with the same text. When the list is empty or omitted, identical replies across
+    separate HITL turns on ``messages: []`` resumes are kept.
     """
     seen_ids = {message.get("id") for message in existing if message.get("id")}
     content_source = content_dedupe_against if content_dedupe_against is not None else []
@@ -813,10 +815,24 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                 pending_events=live_pending_events,
             )
             if hitl_messages:
+                # Content fallback is only for the current request's newly supplied
+                # turns (e.g. client id remap of the resume reply). Rows already in
+                # the stored snapshot keep their ids when replayed in ``messages`` and
+                # must not starve a later identical resume interrupt.
+                stored_ids = {
+                    message.get("id")
+                    for message in (stored_snapshot.messages if stored_snapshot is not None else [])
+                    if message.get("id")
+                }
+                current_turn_client_messages = [
+                    message
+                    for message in client_request_messages
+                    if not (message.get("id") and message.get("id") in stored_ids)
+                ]
                 builder_seed_messages = _append_unique_snapshot_messages(
                     builder_seed_messages,
                     hitl_messages,
-                    content_dedupe_against=client_request_messages,
+                    content_dedupe_against=current_turn_client_messages,
                 )
         snapshot_builder = _WorkflowSnapshotBuilder(builder_seed_messages) if snapshot_session.enabled else None
         if snapshot_builder is not None and effective_state:
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index aaa94e9e254..fd8c7c2c94b 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -10,6 +10,7 @@
 from agent_framework import (
     Executor,
     InMemoryCheckpointStorage,
+    Message,
     Workflow,
     WorkflowBuilder,
     WorkflowContext,
@@ -21,6 +22,26 @@
 from agent_framework_ag_ui import AgentFrameworkWorkflow
 
 
+class _HitlMessageRequestExecutor(Executor):
+    """Minimal HITL executor shared by snapshot resume regression tests."""
+
+    def __init__(self) -> None:
+        super().__init__(id="message_request_executor")
+
+    @handler
+    async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+        del message
+        await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
+
+    @response_handler
+    async def handle_user_input(
+        self, original_request: dict[str, Any], response: list[Message], ctx: WorkflowContext[Any, str]
+    ) -> None:
+        del original_request
+        user_text = response[0].text if response else ""
+        await ctx.yield_output(f"Captured response: {user_text}")
+
+
 async def _run(agent: AgentFrameworkWorkflow, payload: dict[str, Any]) -> list[Any]:
     return [event async for event in agent.run(payload)]
 
@@ -457,30 +478,11 @@ async def finalizer(message: str, ctx: WorkflowContext[None, str]) -> None:
 
 async def test_workflow_hitl_resume_persists_user_text_in_thread_snapshot() -> None:
     """HITL resume with messages:[] must still record the user reply in the snapshot (#8160)."""
-    from agent_framework import Message
-
     from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
     from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
 
-    class MessageRequestExecutor(Executor):
-        def __init__(self) -> None:
-            super().__init__(id="message_request_executor")
-
-        @handler
-        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
-            del message
-            await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
-
-        @response_handler
-        async def handle_user_input(
-            self, original_request: dict, response: list[Message], ctx: WorkflowContext[Any, str]
-        ) -> None:
-            del original_request
-            user_text = response[0].text if response else ""
-            await ctx.yield_output(f"Captured response: {user_text}")
-
     storage = InMemoryCheckpointStorage()
-    workflow = WorkflowBuilder(start_executor=MessageRequestExecutor()).build()
+    workflow = WorkflowBuilder(start_executor=_HitlMessageRequestExecutor()).build()
     store = InMemoryAGUIThreadSnapshotStore()
     agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store, checkpoint_storage=storage)
 
@@ -547,30 +549,11 @@ async def handle_user_input(
 
 async def test_workflow_hitl_resume_keeps_repeated_yes_on_empty_messages() -> None:
     """A second HITL 'yes' with messages:[] must not be dropped as a content duplicate."""
-    from agent_framework import Message
-
     from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
     from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
 
-    class MessageRequestExecutor(Executor):
-        def __init__(self) -> None:
-            super().__init__(id="message_request_executor")
-
-        @handler
-        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
-            del message
-            await ctx.request_info({"prompt": "Need user follow-up"}, list[Message], request_id="handoff-user-input")
-
-        @response_handler
-        async def handle_user_input(
-            self, original_request: dict, response: list[Message], ctx: WorkflowContext[Any, str]
-        ) -> None:
-            del original_request
-            user_text = response[0].text if response else ""
-            await ctx.yield_output(f"Captured response: {user_text}")
-
     storage = InMemoryCheckpointStorage()
-    workflow = WorkflowBuilder(start_executor=MessageRequestExecutor()).build()
+    workflow = WorkflowBuilder(start_executor=_HitlMessageRequestExecutor()).build()
     store = InMemoryAGUIThreadSnapshotStore()
     agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store, checkpoint_storage=storage)
 
@@ -623,6 +606,62 @@ async def handle_user_input(
     assert len(yes_turns) >= 2
 
 
+async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -> None:
+    """Client-replayed prior 'yes' in messages must not drop a new resume interrupt yes."""
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
+
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=_HitlMessageRequestExecutor()).build()
+    store = InMemoryAGUIThreadSnapshotStore()
+    agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store, checkpoint_storage=storage)
+
+    prior = [
+        {"id": "user-1", "role": "user", "content": "start"},
+        {"id": "assistant-1", "role": "assistant", "content": "confirm?"},
+        {"id": "user-yes-1", "role": "user", "content": "yes"},
+        {"id": "assistant-2", "role": "assistant", "content": "confirm again?"},
+    ]
+    await store.save(
+        scope="tenant-a",
+        thread_id="thread-hitl-replay-yes",
+        snapshot=AGUIThreadSnapshot(messages=prior, state=None, interrupt=None),
+    )
+
+    resumed_events = await _run(
+        agent,
+        {
+            "thread_id": "thread-hitl-replay-yes",
+            "run_id": "run-yes-replay",
+            "messages": list(prior),
+            "resume": {
+                "interrupts": [
+                    {
+                        "id": "handoff-user-input",
+                        "value": [
+                            {
+                                "role": "user",
+                                "contents": [{"type": "text", "text": "yes"}],
+                            }
+                        ],
+                    }
+                ]
+            },
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in resumed_events]
+
+    snapshot = await store.get(scope="tenant-a", thread_id="thread-hitl-replay-yes")
+    assert snapshot is not None
+    yes_turns = [
+        message
+        for message in snapshot.messages
+        if message.get("role") == "user" and message.get("content") == "yes"
+    ]
+    assert len(yes_turns) >= 2
+
+
 def test_snapshot_messages_from_resume_skips_approval_via_pending_type() -> None:
     from types import SimpleNamespace
 
@@ -725,3 +764,31 @@ def test_append_unique_snapshot_messages_keeps_second_hitl_yes_without_client_re
         content_dedupe_against=[],
     )
     assert [m["id"] for m in merged] == ["u0", "a0", "u1", "a1", "generated-yes-2"]
+
+
+def test_append_unique_snapshot_messages_keeps_resume_yes_when_client_replays_prior_yes() -> None:
+    """Replayed prior-yes rows (same ids as stored) must not be passed as content fallback."""
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
+
+    history = [
+        {"id": "u0", "role": "user", "content": "start"},
+        {"id": "a0", "role": "assistant", "content": "confirm?"},
+        {"id": "u1", "role": "user", "content": "yes"},
+        {"id": "a1", "role": "assistant", "content": "confirm again?"},
+    ]
+    stored_ids = {message.get("id") for message in history if message.get("id")}
+    client_replay = list(history)
+    # Mirrors the run() call site: only count client turns not already stored by id.
+    current_turn_client_messages = [
+        message
+        for message in client_replay
+        if not (message.get("id") and message.get("id") in stored_ids)
+    ]
+    second_yes = [{"id": "generated-yes-2", "role": "user", "content": "yes"}]
+    merged = _append_unique_snapshot_messages(
+        history,
+        second_yes,
+        content_dedupe_against=current_turn_client_messages,
+    )
+    assert current_turn_client_messages == []
+    assert [m["id"] for m in merged] == ["u0", "a0", "u1", "a1", "generated-yes-2"]

From 6c9d65bab0ca2aab1a2cd4fb98851ab6f6cc7583 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Thu, 17 Sep 2026 16:56:51 +0800
Subject: [PATCH 8/8] test(ag-ui): seed pending interrupt before HITL yes
 resume cases

The repeated-yes / client-replay integration tests were resuming without a live request_info interrupt, which surfaces as RUN_ERROR. Mirror the working HITL snapshot test: first run to establish the pending interrupt, then overwrite the stored snapshot and resume.
---
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 25 +++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
index fd8c7c2c94b..6dfd3039405 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_agent.py
@@ -557,6 +557,19 @@ async def test_workflow_hitl_resume_keeps_repeated_yes_on_empty_messages() -> No
     store = InMemoryAGUIThreadSnapshotStore()
     agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store, checkpoint_storage=storage)
 
+    # Establish a live pending request_info interrupt on this workflow instance.
+    first_events = await _run(
+        agent,
+        {
+            "thread_id": "thread-hitl-yes",
+            "run_id": "run-yes-1",
+            "messages": [{"id": "user-1", "role": "user", "content": "start"}],
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in first_events]
+
+    # Simulate a thread that already persisted an earlier identical "yes".
     await store.save(
         scope="tenant-a",
         thread_id="thread-hitl-yes",
@@ -622,6 +635,18 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -
         {"id": "user-yes-1", "role": "user", "content": "yes"},
         {"id": "assistant-2", "role": "assistant", "content": "confirm again?"},
     ]
+
+    first_events = await _run(
+        agent,
+        {
+            "thread_id": "thread-hitl-replay-yes",
+            "run_id": "run-yes-setup",
+            "messages": [{"id": "user-1", "role": "user", "content": "start"}],
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in first_events]
+
     await store.save(
         scope="tenant-a",
         thread_id="thread-hitl-replay-yes",