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 01/16] 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 02/16] 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 03/16] 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 04/16] 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 05/16] 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 06/16] 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 07/16] 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 08/16] 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",

From 971767ae94b441be72b0567bd7187cc8a51fb69e Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Sat, 26 Sep 2026 17:04:42 +0800
Subject: [PATCH 09/16] fix(ag-ui): sanitize resume user turns before
 snapshotting

Strip control fields (tool_calls/toolCalls, actionExecutionId,
function_approvals) from resume-derived user messages so hydrate/replay
cannot reinterpret them as tool-control history.
---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  | 16 +++++++++++--
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 24 +++++++++++++++++++
 2 files changed, 38 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 6c8e1066141..f5be5ca35ae 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -137,10 +137,21 @@ def _snapshot_messages_from_resume_value(
     return []
 
 
+_RESUME_USER_ALLOWED_KEYS = frozenset({"id", "role", "content", "contents", "name"})
+
+
 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)
+    """Normalize resume user turns for snapshot encoding.
+
+    Only chat-safe user fields are retained. Control fields such as ``tool_calls`` /
+    ``toolCalls``, ``actionExecutionId``, or ``function_approvals`` must not enter the
+    backend-owned thread snapshot — later hydrate/replay can otherwise reinterpret a
+    user turn as assistant/tool-control history.
+    """
+    normalized = {key: value for key, value in message.items() if key in _RESUME_USER_ALLOWED_KEYS}
+    normalized["role"] = "user"
     if normalized.get("content") not in (None, ""):
+        normalized.pop("contents", None)
         return normalized
     contents = normalized.get("contents")
     if isinstance(contents, list):
@@ -150,6 +161,7 @@ def _resume_message_to_agui_dict(message: dict[str, Any]) -> dict[str, Any]:
                 texts.append(str(part.get("text") or ""))
         if texts:
             normalized["content"] = "".join(texts)
+    normalized.pop("contents", None)
     return normalized
 
 
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 6dfd3039405..08d5d468b76 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
@@ -717,6 +717,30 @@ def test_snapshot_messages_from_resume_admits_only_user_roles() -> None:
     assert projected[0]["id"] == "u1"
 
 
+def test_snapshot_messages_from_resume_strips_user_control_fields() -> None:
+    """Resume-derived user turns must not carry tool/control fields into the snapshot."""
+    from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
+
+    projected = _snapshot_messages_from_resume_value(
+        {
+            "role": "user",
+            "id": "u-control",
+            "content": "hello",
+            "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "x", "arguments": "{}"}}],
+            "toolCalls": [{"id": "tc2"}],
+            "actionExecutionId": "ae-1",
+            "function_approvals": [{"id": "fa-1"}],
+        }
+    )
+    assert len(projected) == 1
+    msg = projected[0]
+    assert msg["role"] == "user"
+    assert msg["content"] == "hello"
+    assert msg["id"] == "u-control"
+    for key in ("tool_calls", "toolCalls", "actionExecutionId", "function_approvals", "contents"):
+        assert key not in msg
+
+
 def test_message_identity_supports_multimodal_content() -> None:
     from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages, _message_identity
 

From 9314dfa9401a68eb5ade195cbe312743ff082803 Mon Sep 17 00:00:00 2001
From: ktz03 <2484593937@qq.com>
Date: Wed, 30 Sep 2026 19:46:56 +0800
Subject: [PATCH 10/16] fix(ag-ui): preserve multimodal workflow resume
 snapshots

---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  |  12 +-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 167 ++++++++++++++++--
 2 files changed, 154 insertions(+), 25 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 f5be5ca35ae..75cabaae404 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -155,12 +155,8 @@ def _resume_message_to_agui_dict(message: dict[str, Any]) -> dict[str, Any]:
         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)
+        # Let the snapshot adapter normalize text and media without losing their order.
+        normalized["content"] = contents
     normalized.pop("contents", None)
     return normalized
 
@@ -177,9 +173,7 @@ def _snapshot_messages_from_workflow_resume(
             continue
         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)
-        )
+        messages.extend(_snapshot_messages_from_resume_value(interrupt.get("value"), pending_request=pending_request))
     return 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 08d5d468b76..79a2e0f1f0a 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
@@ -476,6 +476,7 @@ async def finalizer(message: str, ctx: WorkflowContext[None, str]) -> None:
     # The final assistant text is still preserved alongside it.
     assert any(message.get("content") == "Final answer." for message in snapshot.messages)
 
+
 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_ag_ui import InMemoryAGUIThreadSnapshotStore
@@ -612,9 +613,7 @@ async def test_workflow_hitl_resume_keeps_repeated_yes_on_empty_messages() -> No
     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"
+        message for message in snapshot.messages if message.get("role") == "user" and message.get("content") == "yes"
     ]
     assert len(yes_turns) >= 2
 
@@ -680,9 +679,7 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -
     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"
+        message for message in snapshot.messages if message.get("role") == "user" and message.get("content") == "yes"
     ]
     assert len(yes_turns) >= 2
 
@@ -697,9 +694,7 @@ def test_snapshot_messages_from_resume_skips_approval_via_pending_type() -> None
     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"}
-    ]
+    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:
@@ -707,10 +702,12 @@ def test_snapshot_messages_from_resume_admits_only_user_roles() -> None:
 
     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"},
-    ])
+    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"
@@ -741,6 +738,146 @@ def test_snapshot_messages_from_resume_strips_user_control_fields() -> None:
         assert key not in msg
 
 
+@pytest.mark.parametrize("message_list", [False, True], ids=["message", "message-list"])
+@pytest.mark.parametrize(
+    ("parts", "expected_content"),
+    [
+        pytest.param(
+            [
+                {"type": "input_text", "text": "Before"},
+                {
+                    "type": "image",
+                    "source": {"type": "url", "value": "https://example.com/image.png", "mimeType": "image/png"},
+                },
+                {"type": "text", "text": "After"},
+            ],
+            [
+                {"type": "text", "text": "Before"},
+                {"type": "binary", "url": "https://example.com/image.png", "mimeType": "image/png"},
+                {"type": "text", "text": "After"},
+            ],
+            id="mixed-image",
+        ),
+        pytest.param(
+            [{"type": "audio", "source": {"type": "base64", "value": "YWJj", "mimeType": "audio/wav"}}],
+            [{"type": "binary", "data": "YWJj", "mimeType": "audio/wav"}],
+            id="media-only-audio",
+        ),
+        pytest.param(
+            [{"type": "binary", "url": "https://example.com/video.mp4", "mimeType": "video/mp4"}],
+            [{"type": "binary", "url": "https://example.com/video.mp4", "mimeType": "video/mp4"}],
+            id="legacy-binary",
+        ),
+        pytest.param(
+            [{"type": "input_text", "text": "First"}, {"type": "text", "text": "Second"}],
+            "FirstSecond",
+            id="text-only",
+        ),
+    ],
+)
+def test_snapshot_messages_from_resume_preserves_contents(
+    message_list: bool, parts: list[dict[str, Any]], expected_content: Any
+) -> None:
+    """Resume contents retain media and ordering without admitting user control fields."""
+    from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
+
+    message = {
+        "id": "u-media",
+        "role": "user",
+        "name": "user",
+        "contents": parts,
+        "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "x", "arguments": "{}"}}],
+        "toolCalls": [{"id": "tc2"}],
+        "actionExecutionId": "ae-1",
+        "function_approvals": [{"id": "fa-1"}],
+    }
+
+    projected = _snapshot_messages_from_resume_value([message] if message_list else message)
+
+    assert projected == [{"id": "u-media", "role": "user", "name": "user", "content": expected_content}]
+
+
+@pytest.mark.parametrize("content", [None, "", "Preferred content"])
+def test_snapshot_messages_from_resume_contents_respects_content_precedence(content: str | None) -> None:
+    """Contents supply missing/empty content while explicit content retains precedence."""
+    from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
+
+    projected = _snapshot_messages_from_resume_value(
+        {"id": "u1", "role": "user", "content": content, "contents": [{"type": "text", "text": "Fallback"}]}
+    )
+
+    assert projected == [{"id": "u1", "role": "user", "content": content or "Fallback"}]
+
+
+async def test_workflow_hitl_resume_persists_media_only_contents_for_replay() -> None:
+    """A media-only HITL reply survives persistence and conversion from stored history."""
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._message_adapters import agui_messages_to_agent_framework
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY
+
+    class MediaRequestExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("Send media", dict[str, Any], request_id="media-input")
+
+        @response_handler
+        async def capture(
+            self, original_request: str, response: dict[str, Any], ctx: WorkflowContext[Any, str]
+        ) -> None:
+            del original_request, response
+            await ctx.yield_output("Captured media")
+
+    store = InMemoryAGUIThreadSnapshotStore()
+    workflow = WorkflowBuilder(start_executor=MediaRequestExecutor(id="media-request")).build()
+    agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store)
+    first_events = await _run(
+        agent,
+        {
+            "thread_id": "media-thread",
+            "messages": [{"role": "user", "content": "start"}],
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in first_events]
+
+    resumed_events = await _run(
+        agent,
+        {
+            "thread_id": "media-thread",
+            "messages": [],
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+            "resume": {
+                "interrupts": [
+                    {
+                        "id": "media-input",
+                        "value": {
+                            "id": "media-reply",
+                            "role": "user",
+                            "contents": [
+                                {
+                                    "type": "audio",
+                                    "source": {"type": "base64", "value": "YWJj", "mimeType": "audio/wav"},
+                                }
+                            ],
+                        },
+                    }
+                ]
+            },
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in resumed_events]
+
+    snapshot = await store.get(scope="tenant-a", thread_id="media-thread")
+    assert snapshot is not None
+    reply = next(message for message in snapshot.messages if message.get("id") == "media-reply")
+    assert reply["content"] == [{"type": "binary", "data": "YWJj", "mimeType": "audio/wav"}]
+    replayed = agui_messages_to_agent_framework([reply])
+    assert replayed[0].contents[0].type == "data"
+    assert replayed[0].contents[0].uri == "data:audio/wav;base64,YWJj"
+    assert replayed[0].contents[0].media_type == "audio/wav"
+
+
 def test_message_identity_supports_multimodal_content() -> None:
     from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages, _message_identity
 
@@ -829,9 +966,7 @@ def test_append_unique_snapshot_messages_keeps_resume_yes_when_client_replays_pr
     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)
+        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(

From f46597cc44d5e2b522504964361bd1b3fdd8439a Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Fri, 2 Oct 2026 11:10:39 +0800
Subject: [PATCH 11/16] fix(ag-ui): project validated workflow resume responses
 into snapshots

---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  | 123 +++++++---------
 .../agent_framework_ag_ui/_workflow_run.py    |  23 ++-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 137 +++++++++++++++++-
 3 files changed, 201 insertions(+), 82 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 75cabaae404..83e99961add 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -8,7 +8,7 @@
 import logging
 import uuid
 from collections import Counter
-from collections.abc import AsyncGenerator, Callable, Mapping
+from collections.abc import AsyncGenerator, Callable
 from typing import Any, cast
 
 from ag_ui.core import (
@@ -79,41 +79,17 @@ def _hashable_message_content(content: Any) -> Any:
         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 based on the matched pending request's
-    type/data (not the response text). Only ``user`` turns are projected so resume cannot
+    Values have already been coerced and validated by the execution path. 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()
         if not text:
@@ -161,22 +137,6 @@ def _resume_message_to_agui_dict(message: dict[str, Any]) -> dict[str, Any]:
     return normalized
 
 
-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
-        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"), _hashable_message_content(message.get("content")))
@@ -299,6 +259,19 @@ def __init__(self, raw_messages: list[dict[str, Any]]) -> None:
         self.state: dict[str, Any] | None = None
         self.interrupt: list[dict[str, Any]] | None = None
 
+    def append_resume_messages(
+        self, messages: list[dict[str, Any]], *, content_dedupe_against: list[dict[str, Any]]
+    ) -> None:
+        """Seed validated user replies before any workflow output is observed."""
+        self._synthesized_messages = _append_unique_snapshot_messages(
+            self._synthesized_messages,
+            agui_messages_to_snapshot_format(messages),
+            content_dedupe_against=content_dedupe_against,
+        )
+        self._synthesized_message_ids = {
+            message_id for message in self._synthesized_messages if (message_id := message.get("id"))
+        }
+
     def observe(self, event: BaseEvent) -> None:
         """Fold one replayable AG-UI event into the latest snapshot state."""
         if isinstance(event, StateSnapshotEvent):
@@ -727,7 +700,7 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                     "AgentFrameworkWorkflow (or the AG-UI endpoint), or WorkflowBuilder "
                     "checkpoint storage on the workflow instance."
                 )
-        live_pending_events = await _pending_request_events(self.workflow) if self.workflow is not None else {}
+        live_pending_events = await _pending_request_events(workflow)
         if self.workflow is not None and checkpoint_id is None:
             for request_event in live_pending_events.values():
                 owner = getattr(request_event, _REQUEST_OWNER_ATTRIBUTE, None)
@@ -761,6 +734,9 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                     code="WORKFLOW_RESUME_NOT_FOUND",
                 )
                 return
+        effective_pending_events = live_pending_events
+        if checkpoint_id is not None:
+            effective_pending_events = dict(checkpoint.pending_request_info_events or {})
         if self.workflow is not None and checkpoint_id is None:
             for interrupt_id in resume_interrupt_ids:
                 request_event = live_pending_events.get(interrupt_id)
@@ -812,35 +788,31 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                 )
             else:
                 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).
-            # Skip when the client already included the same turn in `messages`.
-            hitl_messages = _snapshot_messages_from_workflow_resume(
-                resume_payload,
-                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=current_turn_client_messages,
-                )
         snapshot_builder = _WorkflowSnapshotBuilder(builder_seed_messages) if snapshot_session.enabled else None
+
+        def append_conversational_responses(values: list[Any]) -> None:
+            # Execution owns response validation and approval classification. Project only
+            # validated conversational values from its effective live/checkpoint requests.
+            if snapshot_builder is None or resume_payload is None:
+                return
+            hitl_messages = [
+                message for value in values for message in _snapshot_messages_from_resume_value(make_json_safe(value))
+            ]
+            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)
+            ]
+            snapshot_builder.append_resume_messages(
+                hitl_messages,
+                content_dedupe_against=current_turn_client_messages,
+            )
+
         if snapshot_builder is not None and effective_state:
             # Seed builder state so a run that emits no StateSnapshotEvent still
             # persists the latest known Shared State instead of dropping it.
@@ -849,7 +821,12 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                 snapshot_builder.state = cast(dict[str, Any], state_snapshot)
         run_error_emitted = False
         async for event in run_workflow_stream(
-            input_data, workflow, checkpoint_storage=run_checkpoint_storage, checkpoint_id=checkpoint_id
+            input_data,
+            workflow,
+            checkpoint_storage=run_checkpoint_storage,
+            checkpoint_id=checkpoint_id,
+            pending_request_events=effective_pending_events,
+            on_conversational_responses=append_conversational_responses,
         ):
             if snapshot_builder is not None:
                 snapshot_builder.observe(event)
diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py b/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py
index 1f01c63c9f1..7017931580b 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow_run.py
@@ -874,12 +874,15 @@ def _coerce_responses_for_pending_requests(
 def _coerce_responses_for_pending_requests_strict(
     responses: dict[str, Any],
     pending_events: dict[str, Any],
+    *,
+    on_conversational_responses: Callable[[list[Any]], None] | None = None,
 ) -> tuple[dict[str, Any], RunErrorEvent | None]:
     """Coerce resume responses or return RUN_ERROR for invalid pending response payloads."""
     if not responses or not pending_events:
         return responses, None
 
     normalized: dict[str, Any] = {}
+    conversational_values: list[Any] = []
     pending_by_id: dict[str, Any] = {}
     for request_id, event in pending_events.items():
         pending_by_id[str(request_id)] = event
@@ -920,6 +923,14 @@ def _coerce_responses_for_pending_requests_strict(
                 ),
             )
         normalized[request_key] = coerced_value
+        # Typed execution controls must never become conversational snapshot history.
+        # Use the response execution actually accepts, including JSON-string Content.
+        if not isinstance(coerced_value, bool) and not (
+            isinstance(coerced_value, Content) and coerced_value.type == "function_approval_response"
+        ):
+            conversational_values.append(coerced_value)
+    if on_conversational_responses is not None:
+        on_conversational_responses(conversational_values)
     return normalized, None
 
 
@@ -1136,6 +1147,8 @@ async def run_workflow_stream(
     *,
     checkpoint_storage: CheckpointStorage | None = None,
     checkpoint_id: str | None = None,
+    pending_request_events: dict[str, Any] | None = None,
+    on_conversational_responses: Callable[[list[Any]], None] | None = None,
 ) -> AsyncGenerator[BaseEvent]:
     """Run a Workflow and emit AG-UI protocol events.
 
@@ -1153,6 +1166,8 @@ async def run_workflow_stream(
             mirroring ``Workflow.run(checkpoint_id=...)``. Any incoming messages are
             treated as request-info responses (or ignored) rather than a new
             start-executor message, so resume stays consistent with the core API.
+        pending_request_events: Optional effective pending requests already resolved by the wrapper.
+        on_conversational_responses: Optional callback for validated non-control responses.
     """
     supplied_thread_id = input_data.get("thread_id") or input_data.get("threadId")
     thread_id = supplied_thread_id or str(uuid.uuid4())
@@ -1179,7 +1194,9 @@ async def run_workflow_stream(
     # persisted pending set instead. Only do so when a resume payload is present, so a
     # pure checkpoint restore still surfaces its pending interrupts instead of tripping
     # the "resume required" contract.
-    if checkpoint_id is not None and resume_payload is not None:
+    if pending_request_events is not None:
+        pending_before_run = pending_request_events
+    elif checkpoint_id is not None and resume_payload is not None:
         # Prefer the explicit AG-UI storage argument; otherwise allow the workflow's
         # builder/runtime storage so builder-emitted pause IDs remain round-trippable.
         if checkpoint_storage is None and not workflow._runner.context.has_checkpointing():  # pyright: ignore[reportPrivateUsage]
@@ -1227,7 +1244,9 @@ async def run_workflow_stream(
         if request_id in pending_interrupt_ids
     }
     responses = _merge_workflow_response_sources(resume_responses, message_responses)
-    responses, response_error = _coerce_responses_for_pending_requests_strict(responses, pending_before_run)
+    responses, response_error = _coerce_responses_for_pending_requests_strict(
+        responses, pending_before_run, on_conversational_responses=on_conversational_responses
+    )
     if response_error is not None:
         yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
         yield response_error
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 79a2e0f1f0a..c79cc095d7d 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
@@ -8,6 +8,7 @@
 
 import pytest
 from agent_framework import (
+    Content,
     Executor,
     InMemoryCheckpointStorage,
     Message,
@@ -684,15 +685,9 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -
     assert len(yes_turns) >= 2
 
 
-def test_snapshot_messages_from_resume_skips_approval_via_pending_type() -> None:
-    from types import SimpleNamespace
-
+def test_snapshot_messages_from_resume_keeps_conversational_approval_text() -> None:
     from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
 
-    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"}]
 
@@ -976,3 +971,131 @@ def test_append_unique_snapshot_messages_keeps_resume_yes_when_client_replays_pr
     )
     assert current_turn_client_messages == []
     assert [m["id"] for m in merged] == ["u0", "a0", "u1", "a1", "generated-yes-2"]
+
+
+@pytest.mark.parametrize("workflow_mode", ["live", "factory", "cold-checkpoint"])
+@pytest.mark.parametrize("response_kind", ["content-approval", "bool-approval", "text"])
+async def test_workflow_snapshot_uses_validated_resume_responses(workflow_mode: str, response_kind: str) -> None:
+    """Approval controls stay out of snapshots for live, factory and cold resumes."""
+    import json
+
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY
+
+    handled: list[Any] = []
+
+    class ContentApprovalExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            call = Content.from_function_call(call_id="call-1", name="refund", arguments={"amount": 5})
+            request = Content.from_function_approval_request(id="reply-1", function_call=call)
+            await ctx.request_info(request, Content, request_id="reply-1")
+
+        @response_handler
+        async def respond(self, original_request: Content, response: Content, ctx: WorkflowContext[Any, str]) -> None:
+            del original_request
+            handled.append(response)
+            assert response.type == "function_approval_response"
+            assert response.approved is True
+            await ctx.yield_output("Decision processed")
+
+    class BoolApprovalExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("Approve?", bool, request_id="reply-1")
+
+        @response_handler
+        async def respond(self, original_request: str, response: bool, ctx: WorkflowContext[Any, str]) -> None:
+            del original_request
+            handled.append(response)
+            assert response is True
+            await ctx.yield_output("Decision processed")
+
+    class TextExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("Say something", str, request_id="reply-1")
+
+        @response_handler
+        async def respond(self, original_request: str, response: str, ctx: WorkflowContext[Any, str]) -> None:
+            del original_request
+            handled.append(response)
+            assert response == "approved"
+            await ctx.yield_output("Reply processed")
+
+    executor_type: type[Executor] = {
+        "content-approval": ContentApprovalExecutor,
+        "bool-approval": BoolApprovalExecutor,
+        "text": TextExecutor,
+    }[response_kind]
+
+    def build_workflow() -> Workflow:
+        return WorkflowBuilder(start_executor=executor_type(id="requester")).build()
+
+    storage = InMemoryCheckpointStorage()
+    store = InMemoryAGUIThreadSnapshotStore()
+    workflow = build_workflow()
+    agent = AgentFrameworkWorkflow(
+        workflow=workflow if workflow_mode != "factory" else None,
+        workflow_factory=(lambda _thread_id: build_workflow()) if workflow_mode == "factory" else None,
+        snapshot_store=store,
+        checkpoint_storage=storage,
+    )
+    first_events = await _run(
+        agent,
+        {
+            "thread_id": "approval-thread",
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+            "messages": [{"id": "start", "role": "user", "content": "start"}],
+        },
+    )
+    assert "RUN_ERROR" not in [event.type for event in first_events]
+    first_finished = next(event for event in first_events if event.type == "RUN_FINISHED")
+    interrupt = _interrupts_from_finished(first_finished)[0]
+    request = interrupt["metadata"]["agent_framework"]["value"]
+    if isinstance(request, str):
+        request = json.loads(request)
+    if response_kind == "content-approval":
+        # The execution path accepts a JSON-string Content response; snapshotting
+        # must consume its validated Content instead of persisting that raw string.
+        value = json.dumps(
+            {
+                "type": "function_approval_response",
+                "id": "reply-1",
+                "approved": True,
+                "function_call": request["function_call"],
+            }
+        )
+    elif response_kind == "bool-approval":
+        value = "true"
+    else:
+        value = "approved"
+    payload: dict[str, Any] = {
+        "thread_id": "approval-thread",
+        _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        "messages": [],
+        "resume": {"interrupts": [{"id": "reply-1", "value": value}]},
+    }
+    if workflow_mode == "cold-checkpoint":
+        checkpoints = await storage.list_checkpoints(workflow_name=workflow.name)
+        latest = max(checkpoints, key=lambda checkpoint: checkpoint.timestamp)
+        payload["forwarded_props"] = {"checkpoint_id": latest.checkpoint_id}
+        agent = AgentFrameworkWorkflow(workflow=build_workflow(), snapshot_store=store, checkpoint_storage=storage)
+    resumed = await _run(agent, payload)
+    assert "RUN_ERROR" not in [event.type for event in resumed]
+    assert len(handled) == 1
+    snapshot = await store.get(scope="tenant-a", thread_id="approval-thread")
+    assert snapshot is not None
+    user_contents = [message["content"] for message in snapshot.messages if message.get("role") == "user"]
+    assert user_contents == (["start", "approved"] if response_kind == "text" else ["start"])
+    # Hydration must replay the same safe history without executing the response again.
+    hydrated = await _run(
+        agent, {"thread_id": "approval-thread", _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a", "messages": []}
+    )
+    replay = next(event for event in hydrated if event.type == "MESSAGES_SNAPSHOT")
+    replay_users = [message.content for message in replay.messages if message.role == "user"]
+    assert replay_users == user_contents
+    assert len(handled) == 1

From 02ceeb7e34554dcee8d85ccb0891c417565e6279 Mon Sep 17 00:00:00 2001
From: ktz03 <2484593937@qq.com>
Date: Fri, 2 Oct 2026 13:25:59 +0800
Subject: [PATCH 12/16] fix(ag-ui): preserve cold checkpoint-only restore

---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  |  4 ++-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 31 +++++++++++++------
 2 files changed, 25 insertions(+), 10 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 83e99961add..19a3e3a04d2 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -735,7 +735,9 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]:
                 )
                 return
         effective_pending_events = live_pending_events
-        if checkpoint_id is not None:
+        # Resolve persisted requests for response validation, but let a pure cold
+        # restore reach the core so it can re-emit its pending interrupt cards.
+        if checkpoint_id is not None and resume_payload is not None:
             effective_pending_events = dict(checkpoint.pending_request_info_events or {})
         if self.workflow is not None and checkpoint_id is None:
             for interrupt_id in resume_interrupt_ids:
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 c79cc095d7d..6bbedafe428 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
@@ -973,10 +973,12 @@ def test_append_unique_snapshot_messages_keeps_resume_yes_when_client_replays_pr
     assert [m["id"] for m in merged] == ["u0", "a0", "u1", "a1", "generated-yes-2"]
 
 
-@pytest.mark.parametrize("workflow_mode", ["live", "factory", "cold-checkpoint"])
+@pytest.mark.parametrize(
+    "workflow_mode", ["live", "factory", "cold-checkpoint", "cold-restore", "cold-factory-restore"]
+)
 @pytest.mark.parametrize("response_kind", ["content-approval", "bool-approval", "text"])
 async def test_workflow_snapshot_uses_validated_resume_responses(workflow_mode: str, response_kind: str) -> None:
-    """Approval controls stay out of snapshots for live, factory and cold resumes."""
+    """Snapshots preserve conversational replies and pending cards across resume and cold restore."""
     import json
 
     from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
@@ -1079,23 +1081,34 @@ def build_workflow() -> Workflow:
         "messages": [],
         "resume": {"interrupts": [{"id": "reply-1", "value": value}]},
     }
-    if workflow_mode == "cold-checkpoint":
+    restore_only = workflow_mode in {"cold-restore", "cold-factory-restore"}
+    if workflow_mode == "cold-checkpoint" or restore_only:
         checkpoints = await storage.list_checkpoints(workflow_name=workflow.name)
         latest = max(checkpoints, key=lambda checkpoint: checkpoint.timestamp)
         payload["forwarded_props"] = {"checkpoint_id": latest.checkpoint_id}
-        agent = AgentFrameworkWorkflow(workflow=build_workflow(), snapshot_store=store, checkpoint_storage=storage)
+        agent = AgentFrameworkWorkflow(
+            workflow=build_workflow() if workflow_mode != "cold-factory-restore" else None,
+            workflow_factory=(lambda _thread_id: build_workflow()) if workflow_mode == "cold-factory-restore" else None,
+            snapshot_store=store,
+            checkpoint_storage=storage,
+        )
+    if restore_only:
+        del payload["resume"]
     resumed = await _run(agent, payload)
-    assert "RUN_ERROR" not in [event.type for event in resumed]
-    assert len(handled) == 1
+    assert not [event.model_dump() for event in resumed if event.type == "RUN_ERROR"]
+    if restore_only:
+        restored_finished = next(event for event in resumed if event.type == "RUN_FINISHED")
+        assert [pending["id"] for pending in _interrupts_from_finished(restored_finished)] == ["reply-1"]
+    assert len(handled) == (0 if restore_only else 1)
     snapshot = await store.get(scope="tenant-a", thread_id="approval-thread")
     assert snapshot is not None
     user_contents = [message["content"] for message in snapshot.messages if message.get("role") == "user"]
-    assert user_contents == (["start", "approved"] if response_kind == "text" else ["start"])
-    # Hydration must replay the same safe history without executing the response again.
+    assert user_contents == (["start", "approved"] if response_kind == "text" and not restore_only else ["start"])
+    # Hydration must replay the same safe history without executing a pending response.
     hydrated = await _run(
         agent, {"thread_id": "approval-thread", _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a", "messages": []}
     )
     replay = next(event for event in hydrated if event.type == "MESSAGES_SNAPSHOT")
     replay_users = [message.content for message in replay.messages if message.role == "user"]
     assert replay_users == user_contents
-    assert len(handled) == 1
+    assert len(handled) == (0 if restore_only else 1)

From 429cae23227d0ea9fffc46e1ae9440a59dc9340c Mon Sep 17 00:00:00 2001
From: ktz03 <2484593937@qq.com>
Date: Fri, 2 Oct 2026 15:49:10 +0800
Subject: [PATCH 13/16] test(ag-ui): select interrupted checkpoints
 deterministically

---
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 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 6bbedafe428..15ae9e9f27d 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
@@ -4,7 +4,9 @@
 
 from __future__ import annotations
 
+from datetime import datetime, timezone
 from typing import Any, cast
+from unittest.mock import Mock
 
 import pytest
 from agent_framework import (
@@ -977,13 +979,20 @@ def test_append_unique_snapshot_messages_keeps_resume_yes_when_client_replays_pr
     "workflow_mode", ["live", "factory", "cold-checkpoint", "cold-restore", "cold-factory-restore"]
 )
 @pytest.mark.parametrize("response_kind", ["content-approval", "bool-approval", "text"])
-async def test_workflow_snapshot_uses_validated_resume_responses(workflow_mode: str, response_kind: str) -> None:
+async def test_workflow_snapshot_uses_validated_resume_responses(
+    workflow_mode: str, response_kind: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
     """Snapshots preserve conversational replies and pending cards across resume and cold restore."""
     import json
 
     from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
     from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY
 
+    # A coarse clock can give the initial and interrupted checkpoints the same timestamp.
+    checkpoint_clock = Mock(wraps=datetime)
+    checkpoint_clock.now.return_value = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    monkeypatch.setattr("agent_framework._workflows._checkpoint.datetime", checkpoint_clock)
+
     handled: list[Any] = []
 
     class ContentApprovalExecutor(Executor):
@@ -1084,8 +1093,12 @@ def build_workflow() -> Workflow:
     restore_only = workflow_mode in {"cold-restore", "cold-factory-restore"}
     if workflow_mode == "cold-checkpoint" or restore_only:
         checkpoints = await storage.list_checkpoints(workflow_name=workflow.name)
-        latest = max(checkpoints, key=lambda checkpoint: checkpoint.timestamp)
-        payload["forwarded_props"] = {"checkpoint_id": latest.checkpoint_id}
+        assert len(checkpoints) > 1
+        assert len({checkpoint.timestamp for checkpoint in checkpoints}) == 1
+        pending_checkpoint = next(
+            checkpoint for checkpoint in checkpoints if "reply-1" in checkpoint.pending_request_info_events
+        )
+        payload["forwarded_props"] = {"checkpoint_id": pending_checkpoint.checkpoint_id}
         agent = AgentFrameworkWorkflow(
             workflow=build_workflow() if workflow_mode != "cold-factory-restore" else None,
             workflow_factory=(lambda _thread_id: build_workflow()) if workflow_mode == "cold-factory-restore" else None,

From ca59e87f380e6d7ca90f7717e799d5bd13b86701 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Wed, 7 Oct 2026 05:36:16 -0700
Subject: [PATCH 14/16] fix(ag-ui): normalize current-turn snapshot
 deduplication

---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  |   9 +-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 141 ++++++++++++++++++
 2 files changed, 148 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 19a3e3a04d2..441c2f13103 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -164,7 +164,12 @@ def _append_unique_snapshot_messages(
     """
     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 []
-    remaining_content = Counter(_message_content_identity(message) for message in content_source)
+    # Reserve exact-id overlap before content matching, regardless of incoming order.
+    # A client turn matched by id must not also consume a distinct identical reply.
+    matched_ids = seen_ids.intersection(message.get("id") for message in incoming if message.get("id"))
+    remaining_content = Counter(
+        _message_content_identity(message) for message in content_source if message.get("id") not in matched_ids
+    )
     merged = list(existing)
     for message in incoming:
         message_id = message.get("id")
@@ -812,7 +817,7 @@ def append_conversational_responses(values: list[Any]) -> None:
             ]
             snapshot_builder.append_resume_messages(
                 hitl_messages,
-                content_dedupe_against=current_turn_client_messages,
+                content_dedupe_against=agui_messages_to_snapshot_format(current_turn_client_messages),
             )
 
         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 15ae9e9f27d..c8b7142b54f 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
@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
 from datetime import datetime, timezone
 from typing import Any, cast
 from unittest.mock import Mock
@@ -687,6 +688,130 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -
     assert len(yes_turns) >= 2
 
 
+@pytest.mark.parametrize(
+    "client_id",
+    [None, "client-reply", "resume-1", "resume-2"],
+    ids=["without-id", "remapped-id", "matching-first-id", "matching-last-id"],
+)
+@pytest.mark.parametrize(
+    ("parts", "expected_content"),
+    [
+        pytest.param([{"type": "input_text", "text": "yes"}], "yes", id="text-parts"),
+        pytest.param(
+            [
+                {"type": "input_text", "text": "Before"},
+                {
+                    "type": "image",
+                    "source": {"type": "url", "value": "https://example.com/image.png", "mimeType": "image/png"},
+                },
+                {"type": "text", "text": "After"},
+            ],
+            [
+                {"type": "text", "text": "Before"},
+                {"type": "binary", "url": "https://example.com/image.png", "mimeType": "image/png"},
+                {"type": "text", "text": "After"},
+            ],
+            id="mixed-image",
+        ),
+        pytest.param(
+            [{"type": "audio", "source": {"type": "base64", "value": "YWJj", "mimeType": "audio/wav"}}],
+            [{"type": "binary", "data": "YWJj", "mimeType": "audio/wav"}],
+            id="media-only-audio",
+        ),
+    ],
+)
+async def test_workflow_hitl_resume_dedupes_normalized_current_turn_content(
+    client_id: str | None, parts: list[dict[str, Any]], expected_content: Any
+) -> None:
+    """Only current-turn overlap consumes one matching reply after content normalization."""
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY, AGUIThreadSnapshot
+
+    handled: list[list[dict[str, Any]]] = []
+
+    class MessageRequestExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("Send replies", list[dict[str, Any]], request_id="user-input")
+
+        @response_handler
+        async def capture(
+            self, original_request: str, response: list[dict[str, Any]], ctx: WorkflowContext[Any, str]
+        ) -> None:
+            del original_request
+            handled.append(response)
+            await ctx.yield_output("Captured replies")
+
+    store = InMemoryAGUIThreadSnapshotStore()
+    workflow = WorkflowBuilder(start_executor=MessageRequestExecutor(id="message-request")).build()
+    agent = AgentFrameworkWorkflow(workflow=workflow, snapshot_store=store)
+    start_message = {"id": "start", "role": "user", "content": "start"}
+    first_events = await _run(
+        agent,
+        {
+            "thread_id": "overlap-thread",
+            "messages": [start_message],
+            _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+        },
+    )
+    assert not [event.model_dump() for event in first_events if event.type == "RUN_ERROR"]
+    first_finished = next(event for event in first_events if event.type == "RUN_FINISHED")
+    assert _interrupts_from_finished(first_finished)[0]["id"] == "user-input"
+
+    # A prior identical reply must not consume either of this interrupt's replies.
+    await store.save(
+        scope="tenant-a",
+        thread_id="overlap-thread",
+        snapshot=AGUIThreadSnapshot(
+            messages=[start_message, {"id": "prior-reply", "role": "user", "content": expected_content}],
+            state=None,
+            interrupt=None,
+        ),
+    )
+    client_reply: dict[str, Any] = {"role": "user", "content": deepcopy(parts)}
+    if client_id is not None:
+        client_reply["id"] = client_id
+    resume_replies = [
+        {"id": "resume-1", "role": "user", "contents": deepcopy(parts)},
+        {"id": "resume-2", "role": "user", "contents": deepcopy(parts)},
+    ]
+    payload = {
+        "thread_id": "overlap-thread",
+        "messages": [
+            start_message,
+            {"id": "prior-reply", "role": "user", "content": deepcopy(parts)},
+            client_reply,
+        ],
+        "resume": {"interrupts": [{"id": "user-input", "value": resume_replies}]},
+        _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a",
+    }
+    original_payload = deepcopy(payload)
+    resumed_events = await _run(agent, payload)
+    assert not [event.model_dump() for event in resumed_events if event.type == "RUN_ERROR"]
+    assert handled == [resume_replies]
+    assert payload == original_payload
+
+    snapshot = await store.get(scope="tenant-a", thread_id="overlap-thread")
+    assert snapshot is not None
+    user_messages = [message for message in snapshot.messages if message.get("role") == "user"]
+    assert [message["content"] for message in user_messages] == ["start", *[expected_content] * 3]
+    assert user_messages[1]["id"] == "prior-reply"
+    if client_id is not None:
+        assert user_messages[2]["id"] == client_id
+    assert user_messages[3]["id"] == ("resume-1" if client_id == "resume-2" else "resume-2")
+
+    hydrated = await _run(agent, {"thread_id": "overlap-thread", "messages": [], _SNAPSHOT_SCOPE_INPUT_KEY: "tenant-a"})
+    replay = next(event for event in hydrated if event.type == "MESSAGES_SNAPSHOT")
+    replay_users = [
+        message.model_dump(by_alias=True, exclude_none=True)["content"]
+        for message in replay.messages
+        if message.role == "user"
+    ]
+    assert replay_users == [message["content"] for message in user_messages]
+    assert handled == [resume_replies]
+
+
 def test_snapshot_messages_from_resume_keeps_conversational_approval_text() -> None:
     from agent_framework_ag_ui._workflow import _snapshot_messages_from_resume_value
 
@@ -930,6 +1055,22 @@ def test_append_unique_snapshot_messages_keeps_intentional_repeated_replies() ->
     assert [m["id"] for m in merged] == ["u0", "r1", "r2"]
 
 
+def test_append_unique_snapshot_messages_keeps_current_overlap_after_historical_id_match() -> None:
+    """Matching historical ids must not consume a current client turn's content allowance."""
+    from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
+
+    history = [{"id": "prior-reply", "role": "user", "content": "yes"}]
+    current = [{"id": "client-reply", "role": "user", "content": "yes"}]
+    incoming = [
+        {"id": "prior-reply", "role": "user", "content": "yes"},
+        {"id": "resume-reply", "role": "user", "content": "yes"},
+    ]
+
+    assert _append_unique_snapshot_messages(history + current, incoming, content_dedupe_against=current) == (
+        history + current
+    )
+
+
 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

From eb423fae61e77b94692735c809e4cfb97f086c62 Mon Sep 17 00:00:00 2001
From: ktz03 <2484593937@qq.com>
Date: Fri, 9 Oct 2026 18:16:17 +0800
Subject: [PATCH 15/16] fix(ag-ui): preserve replayed HITL replies and typed
 message IDs

---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  | 46 ++++++----
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 87 ++++++++++++++++++-
 2 files changed, 115 insertions(+), 18 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 441c2f13103..175f7e48d63 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -4,13 +4,13 @@
 
 from __future__ import annotations
 
-import json
 import logging
 import uuid
 from collections import Counter
 from collections.abc import AsyncGenerator, Callable
 from typing import Any, cast
 
+import msgspec
 from ag_ui.core import (
     BaseEvent,
     MessagesSnapshotEvent,
@@ -41,6 +41,8 @@
     _extract_resume_payload,
     _normalize_resume_interrupts,
     _reconstruct_messages_from_thread_snapshot,
+    _snapshot_messages_match,
+    _snapshot_overlap_length,
 )
 from ._snapshot_session import ThreadSnapshotSession, _event_messages_to_snapshot_dicts
 from ._snapshots import (
@@ -73,10 +75,7 @@ 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)
+    return msgspec.json.encode(make_json_safe(content), order="deterministic")
 
 
 def _snapshot_messages_from_resume_value(
@@ -125,6 +124,8 @@ def _resume_message_to_agui_dict(message: dict[str, Any]) -> dict[str, Any]:
     user turn as assistant/tool-control history.
     """
     normalized = {key: value for key, value in message.items() if key in _RESUME_USER_ALLOWED_KEYS}
+    if not normalized.get("id") and message.get("message_id"):
+        normalized["id"] = message["message_id"]
     normalized["role"] = "user"
     if normalized.get("content") not in (None, ""):
         normalized.pop("contents", None)
@@ -147,6 +148,28 @@ def _message_content_identity(message: dict[str, Any]) -> tuple[Any, ...]:
     return (message.get("role"), _hashable_message_content(message.get("content")))
 
 
+def _current_turn_client_messages(
+    stored_messages: list[dict[str, Any]], client_messages: list[dict[str, Any]]
+) -> list[dict[str, Any]]:
+    """Exclude confirmed prior transcript overlap before resume content matching."""
+    if not stored_messages or not client_messages:
+        return client_messages
+    matched_count = max(
+        (
+            overlap
+            for start, stored_message in enumerate(stored_messages)
+            if _snapshot_messages_match(stored_message, client_messages[0])
+            and (
+                (overlap := _snapshot_overlap_length(stored_messages, client_messages, start=start)) >= 2
+                or (stored_message.get("id") and stored_message.get("id") == client_messages[0].get("id"))
+            )
+        ),
+        default=0,
+    )
+    stored_ids = {message.get("id") for message in stored_messages if message.get("id")}
+    return [message for message in client_messages[matched_count:] if message.get("id") not in stored_ids]
+
+
 def _append_unique_snapshot_messages(
     existing: list[dict[str, Any]],
     incoming: list[dict[str, Any]],
@@ -805,16 +828,9 @@ def append_conversational_responses(values: list[Any]) -> None:
             hitl_messages = [
                 message for value in values for message in _snapshot_messages_from_resume_value(make_json_safe(value))
             ]
-            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)
-            ]
+            current_turn_client_messages = _current_turn_client_messages(
+                stored_snapshot.messages if stored_snapshot is not None else [], client_request_messages
+            )
             snapshot_builder.append_resume_messages(
                 hitl_messages,
                 content_dedupe_against=agui_messages_to_snapshot_format(current_turn_client_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 c8b7142b54f..f3bb744db82 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
@@ -622,7 +622,8 @@ async def test_workflow_hitl_resume_keeps_repeated_yes_on_empty_messages() -> No
     assert len(yes_turns) >= 2
 
 
-async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -> None:
+@pytest.mark.parametrize("replay_ids", [True, False], ids=["with-ids", "without-ids"])
+async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes(replay_ids: bool) -> 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
@@ -661,7 +662,9 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -
         {
             "thread_id": "thread-hitl-replay-yes",
             "run_id": "run-yes-replay",
-            "messages": list(prior),
+            "messages": [
+                {key: value for key, value in message.items() if replay_ids or key != "id"} for message in prior
+            ],
             "resume": {
                 "interrupts": [
                     {
@@ -685,7 +688,7 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes() -
     yes_turns = [
         message for message in snapshot.messages if message.get("role") == "user" and message.get("content") == "yes"
     ]
-    assert len(yes_turns) >= 2
+    assert len(yes_turns) == 2
 
 
 @pytest.mark.parametrize(
@@ -1013,6 +1016,84 @@ def test_message_identity_supports_multimodal_content() -> None:
     assert _append_unique_snapshot_messages([multimodal], [multimodal]) == [multimodal]
 
 
+def test_message_identity_reuses_model_serialization() -> None:
+    from pydantic import BaseModel
+
+    from agent_framework_ag_ui._workflow import _message_content_identity
+
+    class TextPart(BaseModel):
+        type: str = "text"
+        text: str
+
+    modeled = {"role": "user", "content": [TextPart(text="yes")]}
+    wire = {"role": "user", "content": [{"text": "yes", "type": "text"}]}
+    assert _message_content_identity(modeled) == _message_content_identity(wire)
+
+
+@pytest.mark.parametrize("with_client_overlap", [False, True])
+async def test_workflow_hitl_resume_preserves_typed_message_id(with_client_overlap: bool) -> None:
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY
+
+    class TypedMessageRequestExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("first", list[Message], request_id="first-input")
+
+        @response_handler
+        async def reply(self, original_request: str, response: list[Message], ctx: WorkflowContext[Any, str]) -> None:
+            assert response
+            if original_request == "first":
+                await ctx.request_info("second", list[Message], request_id="second-input")
+            else:
+                await ctx.yield_output("Done")
+
+    store = InMemoryAGUIThreadSnapshotStore()
+    agent = AgentFrameworkWorkflow(
+        workflow=WorkflowBuilder(start_executor=TypedMessageRequestExecutor(id="typed-request")).build(),
+        snapshot_store=store,
+    )
+    start = {"id": "start", "role": "user", "content": "start"}
+    reply = Message(role="user", contents=["yes"], message_id="typed-reply")
+    setup = await _run(agent, {"thread_id": "typed-thread", "messages": [start], _SNAPSHOT_SCOPE_INPUT_KEY: "scope"})
+    assert not [event for event in setup if event.type == "RUN_ERROR"]
+    current = {"id": "typed-reply", "role": "user", "content": "yes"}
+    payload = {
+        "thread_id": "typed-thread",
+        "messages": [start, current] if with_client_overlap else [],
+        "resume": {"interrupts": [{"id": "first-input", "value": [reply.to_dict()]}]},
+        _SNAPSHOT_SCOPE_INPUT_KEY: "scope",
+    }
+    original_payload = deepcopy(payload)
+    resumed = await _run(agent, payload)
+    assert not [event for event in resumed if event.type == "RUN_ERROR"]
+    assert payload == original_payload
+    snapshot = await store.get(scope="scope", thread_id="typed-thread")
+    assert snapshot is not None
+    user_replies = [message for message in snapshot.messages if message.get("content") == "yes"]
+    assert user_replies == [current]
+    prior_transcript = deepcopy(snapshot.messages)
+
+    replay = await _run(
+        agent,
+        {
+            "thread_id": "typed-thread",
+            "messages": prior_transcript,
+            "resume": {
+                "interrupts": [
+                    {"id": "second-input", "value": [Message("user", ["next"], message_id="next").to_dict()]}
+                ]
+            },
+            _SNAPSHOT_SCOPE_INPUT_KEY: "scope",
+        },
+    )
+    assert not [event for event in replay if event.type == "RUN_ERROR"]
+    snapshot = await store.get(scope="scope", thread_id="typed-thread")
+    assert snapshot is not None
+    assert [message for message in snapshot.messages if message.get("content") == "yes"] == [current]
+
+
 def test_append_unique_snapshot_messages_dedupes_client_replay() -> None:
     from agent_framework_ag_ui._workflow import _append_unique_snapshot_messages
 

From ef8801b734da7707937253e01e0fc38d6ac84127 Mon Sep 17 00:00:00 2001
From: ktz03 <2484593937@qq.com>
Date: Mon, 12 Oct 2026 17:35:35 +0800
Subject: [PATCH 16/16] fix(ag-ui): preserve Content HITL replies in thread
 snapshots

---
 .../ag-ui/agent_framework_ag_ui/_workflow.py  |  20 ++-
 .../ag-ui/tests/ag_ui/test_workflow_agent.py  | 120 ++++++++++++++++++
 2 files changed, 135 insertions(+), 5 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 175f7e48d63..3b07d9770b8 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_workflow.py
@@ -31,11 +31,11 @@
     ToolCallResultEvent,
     ToolCallStartEvent,
 )
-from agent_framework import CheckpointID, CheckpointStorage, Workflow, WorkflowCheckpoint
+from agent_framework import CheckpointID, CheckpointStorage, Content, Message, Workflow, WorkflowCheckpoint
 from agent_framework._telemetry import mark_feature_used
 
 from ._feature_usage import FeatureIndex
-from ._message_adapters import agui_messages_to_snapshot_format
+from ._message_adapters import agent_framework_messages_to_agui, agui_messages_to_snapshot_format
 from ._run_common import (
     _cancelled_resume_interrupt_ids,
     _extract_resume_payload,
@@ -87,6 +87,18 @@ def _snapshot_messages_from_resume_value(
     ``user`` turns are projected so resume cannot
     forge assistant/system/tool history into the backend-owned snapshot.
     """
+    if isinstance(value, Content):
+        value = [value]
+    if isinstance(value, list) and value and all(isinstance(item, Content) for item in value):
+        # Reuse the user-content encoder for ordered text/media, excluding tool and
+        # approval controls before they can be reinterpreted as snapshot history.
+        contents = [item for item in value if item.type in {"text", "uri", "data"}]
+        if not contents:
+            return []
+        return agui_messages_to_snapshot_format(
+            agent_framework_messages_to_agui([Message(role="user", contents=contents)])
+        )
+    value = make_json_safe(value)
     if isinstance(value, bool) or value is None:
         return []
     if isinstance(value, str):
@@ -825,9 +837,7 @@ def append_conversational_responses(values: list[Any]) -> None:
             # validated conversational values from its effective live/checkpoint requests.
             if snapshot_builder is None or resume_payload is None:
                 return
-            hitl_messages = [
-                message for value in values for message in _snapshot_messages_from_resume_value(make_json_safe(value))
-            ]
+            hitl_messages = [message for value in values for message in _snapshot_messages_from_resume_value(value)]
             current_turn_client_messages = _current_turn_client_messages(
                 stored_snapshot.messages if stored_snapshot is not None else [], client_request_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 f3bb744db82..6f5df70d923 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
@@ -691,6 +691,126 @@ async def test_workflow_hitl_resume_keeps_yes_when_messages_replay_prior_yes(rep
     assert len(yes_turns) == 2
 
 
+@pytest.mark.parametrize("workflow_mode", ["live", "factory", "cold-checkpoint"])
+@pytest.mark.parametrize("reply_kind", ["text", "text-list", "media-list", "audio", "control", "mixed-control"])
+async def test_workflow_hitl_content_reply_survives_snapshot_hydration(workflow_mode: str, reply_kind: str) -> None:
+    """Accepted Content replies survive hydration as one ordered user turn, without control parts."""
+    import msgspec
+
+    from agent_framework_ag_ui import InMemoryAGUIThreadSnapshotStore
+    from agent_framework_ag_ui._snapshots import _SNAPSHOT_SCOPE_INPUT_KEY
+
+    handled: list[list[Content]] = []
+
+    class ContentRequestExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("Send a reply", Content, request_id="content-input")
+
+        @response_handler
+        async def respond(self, original_request: str, response: Content, ctx: WorkflowContext[Any, str]) -> None:
+            del original_request
+            handled.append([response])
+            await ctx.yield_output("Reply received")
+
+    class ContentListRequestExecutor(Executor):
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext[Any, str]) -> None:
+            del message
+            await ctx.request_info("Send a reply", list[Content], request_id="content-input")
+
+        @response_handler
+        async def respond(self, original_request: str, response: list[Content], ctx: WorkflowContext[Any, str]) -> None:
+            del original_request
+            handled.append(response)
+            await ctx.yield_output("Reply received")
+
+    text = Content.from_text("approved", raw_representation={"private": "not-chat-history"})
+    audio = Content.from_data(b"audio", "audio/wav")
+    image = Content.from_uri("https://example.com/reply.png", media_type="image/png")
+    control = Content.from_function_call(call_id="control-call", name="refund", arguments={"amount": 5})
+    replies = {
+        "text": [text],
+        "text-list": [text, Content.from_text(" again")],
+        "media-list": [text, image, Content.from_text(" after")],
+        "audio": [audio],
+        "control": [control],
+        "mixed-control": [control, text],
+    }[reply_kind]
+    is_list = reply_kind in {"text-list", "media-list", "mixed-control"}
+    executor_type = ContentListRequestExecutor if is_list else ContentRequestExecutor
+
+    def build_workflow() -> Workflow:
+        return WorkflowBuilder(start_executor=executor_type(id="content-requester")).build()
+
+    workflow = build_workflow()
+    store = InMemoryAGUIThreadSnapshotStore()
+    storage = InMemoryCheckpointStorage()
+    agent = AgentFrameworkWorkflow(
+        workflow=workflow if workflow_mode != "factory" else None,
+        workflow_factory=(lambda _thread_id: build_workflow()) if workflow_mode == "factory" else None,
+        snapshot_store=store,
+        checkpoint_storage=storage,
+    )
+    start = {"id": "start", "role": "user", "content": "start"}
+    setup = await _run(agent, {"thread_id": "content-thread", "messages": [start], _SNAPSHOT_SCOPE_INPUT_KEY: "scope"})
+    assert not [event.model_dump() for event in setup if event.type == "RUN_ERROR"]
+    finished = next(event for event in setup if event.type == "RUN_FINISHED")
+    assert [interrupt["id"] for interrupt in _interrupts_from_finished(finished)] == ["content-input"]
+    # Exercise the wire format the public resume path coerces into Content objects.
+    wire_parts = [part.to_dict() for part in replies]
+    wire_with_raw = deepcopy(wire_parts)
+    raw_representation = {"private": "not-chat-history"}
+    wire_with_raw[0]["raw_representation"] = raw_representation
+    wire_value = msgspec.json.encode(wire_with_raw if is_list else wire_with_raw[0]).decode()
+    payload: dict[str, Any] = {
+        "thread_id": "content-thread",
+        "messages": [],
+        _SNAPSHOT_SCOPE_INPUT_KEY: "scope",
+        "resume": {"interrupts": [{"id": "content-input", "value": wire_value}]},
+    }
+    original_payload = deepcopy(payload)
+    if workflow_mode == "cold-checkpoint":
+        checkpoints = await storage.list_checkpoints(workflow_name=workflow.name)
+        pending = next(
+            checkpoint for checkpoint in checkpoints if "content-input" in checkpoint.pending_request_info_events
+        )
+        payload["forwarded_props"] = {"checkpoint_id": pending.checkpoint_id}
+        original_payload = deepcopy(payload)
+        agent = AgentFrameworkWorkflow(workflow=build_workflow(), snapshot_store=store, checkpoint_storage=storage)
+    resumed = await _run(agent, payload)
+    assert not [event.model_dump() for event in resumed if event.type == "RUN_ERROR"]
+    assert [part.to_dict() for part in handled[0]] == wire_parts
+    assert handled[0][0].raw_representation == raw_representation
+    assert payload == original_payload
+
+    expected: Any = {
+        "text": "approved",
+        "text-list": "approved again",
+        "media-list": [
+            {"type": "text", "text": "approved"},
+            {"type": "binary", "mimeType": "image/png", "url": "https://example.com/reply.png"},
+            {"type": "text", "text": " after"},
+        ],
+        "audio": [{"type": "binary", "mimeType": "audio/wav", "data": "YXVkaW8="}],
+        "control": None,
+        "mixed-control": "approved",
+    }[reply_kind]
+    snapshot = await store.get(scope="scope", thread_id="content-thread")
+    assert snapshot is not None
+    user_replies = [
+        message for message in snapshot.messages if message.get("role") == "user" and message["id"] != "start"
+    ]
+    assert [message["content"] for message in user_replies] == ([] if expected is None else [expected])
+    assert all(set(message) == {"id", "role", "content"} for message in user_replies)
+    hydrated = await _run(agent, {"thread_id": "content-thread", "messages": [], _SNAPSHOT_SCOPE_INPUT_KEY: "scope"})
+    replay = next(event for event in hydrated if event.type == "MESSAGES_SNAPSHOT")
+    replay_replies = [message for message in replay.messages if message.role == "user" and message.id != "start"]
+    assert [message.model_dump(by_alias=True, exclude_none=True) for message in replay_replies] == user_replies
+    assert len(handled) == 1
+
+
 @pytest.mark.parametrize(
     "client_id",
     [None, "client-reply", "resume-1", "resume-2"],