From 5c440739896b65b1eab8cfe12c88d0828e8a0c9a Mon Sep 17 00:00:00 2001
From: minelhi <3417378192@qq.com>
Date: Wed, 9 Sep 2026 01:35:09 +0800
Subject: [PATCH 1/6] Python: include checkpoint_id on AG-UI interrupt metadata
 (#8150)

Attach the pause workflow checkpoint id to RUN_FINISHED interrupt
metadata.agent_framework so multi-worker clients can resume via
forwardedProps.checkpoint_id without a side channel.
---
 .../agent_framework_ag_ui/_workflow_run.py    | 120 +++++++++++++++++-
 .../ag-ui/tests/ag_ui/test_workflow_run.py    |  20 +++
 2 files changed, 135 insertions(+), 5 deletions(-)

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 f14aa0e206c..d7c764e4d8f 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
@@ -127,7 +127,12 @@ def _workflow_interrupt_value(request_data: Any) -> Any:
     return {"data": safe_request_data}
 
 
-def _workflow_interrupt_metadata(request_payload: dict[str, Any], value: Any) -> dict[str, Any]:
+def _workflow_interrupt_metadata(
+    request_payload: dict[str, Any],
+    value: Any,
+    *,
+    checkpoint_id: str | None = None,
+) -> dict[str, Any]:
     """Build Agent Framework metadata for workflow request_info interrupts."""
     agent_framework_metadata = {
         key: make_json_safe(value)
@@ -139,12 +144,89 @@ def _workflow_interrupt_metadata(request_payload: dict[str, Any], value: Any) ->
             "response_type": request_payload.get("response_type"),
             "data": request_payload.get("data"),
             "value": value,
+            "checkpoint_id": checkpoint_id,
         }.items()
         if value is not None
     }
     return {"agent_framework": agent_framework_metadata}
 
 
+def _attach_checkpoint_id_to_interrupts(
+    interrupts: list[dict[str, Any]],
+    checkpoint_id: str | None,
+) -> list[dict[str, Any]]:
+    """Attach ``checkpoint_id`` to each interrupt's ``metadata.agent_framework`` (issue #8150).
+
+    Multi-worker hosts need the pause checkpoint on the wire so the next resume can pass
+    ``forwardedProps.checkpoint_id`` without a side-channel lookup. No-op when checkpointing
+    is inactive or the id is already present.
+    """
+    if not checkpoint_id or not interrupts:
+        return interrupts
+
+    attached: list[dict[str, Any]] = []
+    for interrupt in interrupts:
+        entry = dict(interrupt)
+        metadata = entry.get("metadata")
+        if isinstance(metadata, dict):
+            metadata = dict(metadata)
+        else:
+            metadata = {}
+        agent_framework = metadata.get("agent_framework")
+        if isinstance(agent_framework, dict):
+            agent_framework = dict(agent_framework)
+        else:
+            agent_framework = {}
+        agent_framework.setdefault("checkpoint_id", checkpoint_id)
+        metadata["agent_framework"] = agent_framework
+        entry["metadata"] = metadata
+        attached.append(entry)
+    return attached
+
+
+async def _latest_checkpoint_id_for_workflow(
+    workflow: Workflow,
+    checkpoint_storage: CheckpointStorage | None,
+) -> str | None:
+    """Best-effort latest checkpoint id for the workflow when storage is configured."""
+    if checkpoint_storage is None:
+        return None
+    get_latest = getattr(checkpoint_storage, "get_latest", None)
+    if get_latest is None:
+        return None
+    try:
+        latest = await get_latest(workflow_name=workflow.name)
+    except Exception:  # pragma: no cover - defensive for storage drift
+        logger.warning("Could not resolve latest workflow checkpoint id for interrupt metadata", exc_info=True)
+        return None
+    if latest is None:
+        return None
+    checkpoint_id = getattr(latest, "checkpoint_id", None)
+    return str(checkpoint_id) if checkpoint_id is not None else None
+
+
+async def _build_run_finished_with_checkpointed_interrupts(
+    *,
+    run_id: str,
+    thread_id: str,
+    interrupts: list[dict[str, Any]],
+    workflow: Workflow,
+    checkpoint_storage: CheckpointStorage | None,
+    known_checkpoint_id: str | None = None,
+) -> Any:
+    """Build RUN_FINISHED, attaching the pause checkpoint id to interrupt metadata when available."""
+    # Prefer the latest persisted checkpoint (the pause boundary). Fall back to an
+    # explicitly known id (e.g. cold replay short-circuit) when storage has nothing newer.
+    pause_checkpoint_id = await _latest_checkpoint_id_for_workflow(workflow, checkpoint_storage)
+    if pause_checkpoint_id is None:
+        pause_checkpoint_id = known_checkpoint_id
+    return _build_run_finished_event(
+        run_id=run_id,
+        thread_id=thread_id,
+        interrupts=_attach_checkpoint_id_to_interrupts(interrupts, pause_checkpoint_id),
+    )
+
+
 async def _pending_request_events(workflow: Workflow) -> dict[str, Any]:
     """Best-effort retrieval of pending request_info events from workflow context."""
     runner_context = getattr(workflow, "_runner_context", None)
@@ -1133,12 +1215,26 @@ async def run_workflow_stream(
             interrupt_event_value = _workflow_interrupt_event_value(request_payload)
             if interrupt_event_value is not None:
                 yield CustomEvent(name=_INTERRUPT_CARD_EVENT_NAME, value=interrupt_event_value)
-        yield _build_run_finished_event(run_id=run_id, thread_id=thread_id, interrupts=pending_interrupts)
+        yield await _build_run_finished_with_checkpointed_interrupts(
+            run_id=run_id,
+            thread_id=thread_id,
+            interrupts=pending_interrupts,
+            workflow=workflow,
+            checkpoint_storage=checkpoint_storage,
+            known_checkpoint_id=checkpoint_id,
+        )
         return
 
     if checkpoint_id is None and not responses and not messages:
         yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
-        yield _build_run_finished_event(run_id=run_id, thread_id=thread_id, interrupts=pending_interrupts)
+        yield await _build_run_finished_with_checkpointed_interrupts(
+            run_id=run_id,
+            thread_id=thread_id,
+            interrupts=pending_interrupts,
+            workflow=workflow,
+            checkpoint_storage=checkpoint_storage,
+            known_checkpoint_id=checkpoint_id,
+        )
         return
 
     def _drain_open_message() -> list[TextMessageEndEvent]:
@@ -1244,7 +1340,14 @@ def _drain_open_blocks() -> list[BaseEvent]:
                         yield end_event
                     if not interrupts:
                         interrupts.extend(_interrupts_from_pending_requests(await _pending_request_events(workflow)))
-                    yield _build_run_finished_event(run_id=run_id, thread_id=thread_id, interrupts=interrupts)
+                    yield await _build_run_finished_with_checkpointed_interrupts(
+                        run_id=run_id,
+                        thread_id=thread_id,
+                        interrupts=interrupts,
+                        workflow=workflow,
+                        checkpoint_storage=checkpoint_storage,
+                        known_checkpoint_id=checkpoint_id,
+                    )
                     terminal_emitted = True
                 elif state_value not in _TERMINAL_STATES:
                     yield CustomEvent(name="status", value={"state": state_value})
@@ -1391,4 +1494,11 @@ def _drain_open_blocks() -> list[BaseEvent]:
     if not terminal_emitted and not run_error_emitted:
         if not interrupts:
             interrupts.extend(_interrupts_from_pending_requests(await _pending_request_events(workflow)))
-        yield _build_run_finished_event(run_id=run_id, thread_id=thread_id, interrupts=interrupts)
+        yield await _build_run_finished_with_checkpointed_interrupts(
+            run_id=run_id,
+            thread_id=thread_id,
+            interrupts=interrupts,
+            workflow=workflow,
+            checkpoint_storage=checkpoint_storage,
+            known_checkpoint_id=checkpoint_id,
+        )
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
index f1cd16c9593..a13733b52ef 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
@@ -95,6 +95,24 @@ def _interrupt_metadata_value(interrupt: dict[str, Any]) -> dict[str, Any]:
     return cast(dict[str, Any], value)
 
 
+def test_attach_checkpoint_id_to_interrupts_setdefault() -> None:
+    """Issue #8150: attach pause checkpoint_id without overwriting an existing one."""
+    from agent_framework_ag_ui._workflow_run import _attach_checkpoint_id_to_interrupts
+
+    empty = _attach_checkpoint_id_to_interrupts([{"id": "r1"}], None)
+    assert empty == [{"id": "r1"}]
+
+    attached = _attach_checkpoint_id_to_interrupts(
+        [{"id": "r1", "metadata": {"agent_framework": {"type": "workflow_request_info"}}}],
+        "cp-123",
+    )
+    assert attached[0]["metadata"]["agent_framework"]["checkpoint_id"] == "cp-123"
+    assert attached[0]["metadata"]["agent_framework"]["type"] == "workflow_request_info"
+
+    preserved = _attach_checkpoint_id_to_interrupts(attached, "cp-other")
+    assert preserved[0]["metadata"]["agent_framework"]["checkpoint_id"] == "cp-123"
+
+
 async def test_workflow_run_maps_custom_and_text_events():
     """Custom workflow events and yielded text are mapped to AG-UI events."""
 
@@ -682,6 +700,8 @@ async def handle_approval(self, original_request: Content, response: Content, ct
     )
     assert checkpoints, "expected the interrupted run to create a checkpoint"
     resume_checkpoint_id = checkpoints[-1].checkpoint_id
+    # Issue #8150: interrupt metadata must carry the pause checkpoint for multi-worker resume.
+    assert interrupt_payload[0]["metadata"]["agent_framework"]["checkpoint_id"] == resume_checkpoint_id
 
     # Resume on a FRESH workflow instance so no pending requests exist in memory until
     # the checkpoint is restored -- a cold restore, as after a process restart.

From 8e36bd4c2c93cc4c29959dae17549dd3849d0240 Mon Sep 17 00:00:00 2001
From: minelhi <3417378192@qq.com>
Date: Wed, 9 Sep 2026 10:26:18 +0800
Subject: [PATCH 2/6] Python: Prefer runner pause checkpoint over shared
 get_latest (#8150)

---
 .../agent_framework_ag_ui/_workflow_run.py    | 106 +++++++++++++----
 .../ag-ui/tests/ag_ui/test_workflow_run.py    | 110 ++++++++++++++++++
 2 files changed, 195 insertions(+), 21 deletions(-)

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 d7c764e4d8f..063a06289f0 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
@@ -130,8 +130,6 @@ def _workflow_interrupt_value(request_data: Any) -> Any:
 def _workflow_interrupt_metadata(
     request_payload: dict[str, Any],
     value: Any,
-    *,
-    checkpoint_id: str | None = None,
 ) -> dict[str, Any]:
     """Build Agent Framework metadata for workflow request_info interrupts."""
     agent_framework_metadata = {
@@ -144,7 +142,6 @@ def _workflow_interrupt_metadata(
             "response_type": request_payload.get("response_type"),
             "data": request_payload.get("data"),
             "value": value,
-            "checkpoint_id": checkpoint_id,
         }.items()
         if value is not None
     }
@@ -184,25 +181,88 @@ def _attach_checkpoint_id_to_interrupts(
     return attached
 
 
-async def _latest_checkpoint_id_for_workflow(
+def _resolve_effective_checkpoint_storage(
     workflow: Workflow,
     checkpoint_storage: CheckpointStorage | None,
-) -> str | None:
-    """Best-effort latest checkpoint id for the workflow when storage is configured."""
-    if checkpoint_storage is None:
+) -> CheckpointStorage | None:
+    """Prefer the run argument, else the storage already bound on the workflow runner."""
+    if checkpoint_storage is not None:
+        return checkpoint_storage
+    runner = getattr(workflow, "_runner", None)
+    ctx = getattr(runner, "context", None) if runner is not None else None
+    if ctx is None:
         return None
-    get_latest = getattr(checkpoint_storage, "get_latest", None)
-    if get_latest is None:
+    getter = getattr(ctx, "_get_effective_checkpoint_storage", None)
+    if callable(getter):
+        try:
+            return cast(CheckpointStorage | None, getter())
+        except Exception:  # pragma: no cover - defensive for duck-typed runners
+            return None
+    return None
+
+
+def _runner_previous_checkpoint_id(workflow: Workflow) -> str | None:
+    """Return the checkpoint id last persisted by *this* workflow runner instance."""
+    runner = getattr(workflow, "_runner", None)
+    if runner is None:
         return None
+    checkpoint_id = getattr(runner, "_previous_checkpoint_id", None)
+    return str(checkpoint_id) if checkpoint_id is not None else None
+
+
+def _interrupt_request_ids(interrupts: list[dict[str, Any]]) -> set[str]:
+    return {str(item["id"]) for item in interrupts if item.get("id") is not None}
+
+
+async def _checkpoint_covers_interrupt_ids(
+    storage: CheckpointStorage,
+    checkpoint_id: str,
+    request_ids: set[str],
+) -> bool:
+    """True when the checkpoint's pending request_info set covers ``request_ids``."""
+    if not request_ids:
+        return False
     try:
-        latest = await get_latest(workflow_name=workflow.name)
-    except Exception:  # pragma: no cover - defensive for storage drift
-        logger.warning("Could not resolve latest workflow checkpoint id for interrupt metadata", exc_info=True)
-        return None
-    if latest is None:
+        checkpoint = await storage.load(checkpoint_id)
+    except Exception:  # pragma: no cover - storage/type drift
+        return False
+    pending = getattr(checkpoint, "pending_request_info_events", None) or {}
+    return request_ids.issubset({str(key) for key in dict(pending)})
+
+
+async def _pause_checkpoint_id_for_interrupts(
+    *,
+    workflow: Workflow,
+    checkpoint_storage: CheckpointStorage | None,
+    interrupts: list[dict[str, Any]],
+    known_checkpoint_id: str | None = None,
+) -> str | None:
+    """Resolve the pause checkpoint for *this* run's interrupts.
+
+    Prefer the runner's last-saved id (scoped to this instance) over shared
+    ``get_latest(workflow_name=...)``, which can race across owners. Validate that
+    the candidate actually records the pending interrupt ids before advertising it.
+    """
+    if not interrupts:
         return None
-    checkpoint_id = getattr(latest, "checkpoint_id", None)
-    return str(checkpoint_id) if checkpoint_id is not None else None
+
+    storage = _resolve_effective_checkpoint_storage(workflow, checkpoint_storage)
+    request_ids = _interrupt_request_ids(interrupts)
+    candidates: list[str] = []
+    runner_id = _runner_previous_checkpoint_id(workflow)
+    if runner_id is not None:
+        candidates.append(runner_id)
+    if known_checkpoint_id is not None and known_checkpoint_id not in candidates:
+        candidates.append(known_checkpoint_id)
+
+    if storage is None:
+        # Without storage we cannot prove coverage; only advertise the id this runner saved.
+        return runner_id
+
+    for candidate in candidates:
+        if await _checkpoint_covers_interrupt_ids(storage, candidate, request_ids):
+            return candidate
+    return None
 
 
 async def _build_run_finished_with_checkpointed_interrupts(
@@ -215,11 +275,15 @@ async def _build_run_finished_with_checkpointed_interrupts(
     known_checkpoint_id: str | None = None,
 ) -> Any:
     """Build RUN_FINISHED, attaching the pause checkpoint id to interrupt metadata when available."""
-    # Prefer the latest persisted checkpoint (the pause boundary). Fall back to an
-    # explicitly known id (e.g. cold replay short-circuit) when storage has nothing newer.
-    pause_checkpoint_id = await _latest_checkpoint_id_for_workflow(workflow, checkpoint_storage)
-    if pause_checkpoint_id is None:
-        pause_checkpoint_id = known_checkpoint_id
+    if not interrupts:
+        return _build_run_finished_event(run_id=run_id, thread_id=thread_id, interrupts=interrupts)
+
+    pause_checkpoint_id = await _pause_checkpoint_id_for_interrupts(
+        workflow=workflow,
+        checkpoint_storage=checkpoint_storage,
+        interrupts=interrupts,
+        known_checkpoint_id=known_checkpoint_id,
+    )
     return _build_run_finished_event(
         run_id=run_id,
         thread_id=thread_id,
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
index a13733b52ef..b2c8e538f24 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
@@ -113,6 +113,116 @@ def test_attach_checkpoint_id_to_interrupts_setdefault() -> None:
     assert preserved[0]["metadata"]["agent_framework"]["checkpoint_id"] == "cp-123"
 
 
+@pytest.mark.asyncio
+async def test_pause_checkpoint_id_ignores_competing_shared_latest() -> None:
+    """Prefer this runner's pause checkpoint over a newer shared get_latest() winner."""
+    from agent_framework import WorkflowCheckpoint
+    from agent_framework_ag_ui._workflow_run import (
+        _build_run_finished_with_checkpointed_interrupts,
+        _pause_checkpoint_id_for_interrupts,
+    )
+
+    class ApprovalExecutor(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="approval_executor")
+
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+            del message
+            function_call = Content.from_function_call(
+                call_id="refund-call",
+                name="submit_refund",
+                arguments={"order_id": "12345"},
+            )
+            approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call)
+            await ctx.request_info(approval_request, Content, request_id="approval-1")
+
+        @response_handler
+        async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
+            del original_request, response
+            await ctx.yield_output("done")  # type: ignore[arg-type]
+
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=ApprovalExecutor(), checkpoint_storage=storage).build()
+    first_events = [
+        event
+        async for event in run_workflow_stream(
+            {"messages": [{"role": "user", "content": "go"}]},
+            workflow,
+            checkpoint_storage=storage,
+        )
+    ]
+    first_finished = [event for event in first_events if event.type == "RUN_FINISHED"][0]
+    interrupt_payload = _interrupts_from_run_finished(first_finished)
+    pause_id = interrupt_payload[0]["metadata"]["agent_framework"]["checkpoint_id"]
+
+    # Inject a newer shared checkpoint that does not cover this interrupt (other owner / stale).
+    competing = WorkflowCheckpoint(
+        workflow_name=workflow.name,
+        graph_signature_hash="competing",
+        pending_request_info_events={},
+        timestamp="9999-01-01T00:00:00+00:00",
+    )
+    await storage.save(competing)
+    latest = await storage.get_latest(workflow_name=workflow.name)
+    assert latest is not None
+    assert latest.checkpoint_id == competing.checkpoint_id
+
+    resolved = await _pause_checkpoint_id_for_interrupts(
+        workflow=workflow,
+        checkpoint_storage=storage,
+        interrupts=interrupt_payload,
+    )
+    assert resolved == pause_id
+
+    rebuilt = await _build_run_finished_with_checkpointed_interrupts(
+        run_id="run-1",
+        thread_id="thread-1",
+        interrupts=interrupt_payload,
+        workflow=workflow,
+        checkpoint_storage=storage,
+    )
+    rebuilt_interrupts = _interrupts_from_run_finished(rebuilt)
+    assert rebuilt_interrupts[0]["metadata"]["agent_framework"]["checkpoint_id"] == pause_id
+
+
+@pytest.mark.asyncio
+async def test_builder_checkpoint_storage_attaches_id_without_run_arg() -> None:
+    """WorkflowBuilder(checkpoint_storage=...) alone must still advertise pause checkpoint_id."""
+
+    class ApprovalExecutor(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="approval_executor")
+
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+            del message
+            function_call = Content.from_function_call(
+                call_id="refund-call",
+                name="submit_refund",
+                arguments={"order_id": "12345"},
+            )
+            approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call)
+            await ctx.request_info(approval_request, Content, request_id="approval-1")
+
+        @response_handler
+        async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
+            del original_request, response
+            await ctx.yield_output("done")  # type: ignore[arg-type]
+
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=ApprovalExecutor(), checkpoint_storage=storage).build()
+    # Deliberately omit checkpoint_storage= on the AG-UI entrypoint (builder path only).
+    events = [
+        event async for event in run_workflow_stream({"messages": [{"role": "user", "content": "go"}]}, workflow)
+    ]
+    finished = [event for event in events if event.type == "RUN_FINISHED"][0]
+    interrupt_payload = _interrupts_from_run_finished(finished)
+    checkpoints = await storage.list_checkpoints(workflow_name=workflow.name)
+    assert checkpoints
+    assert interrupt_payload[0]["metadata"]["agent_framework"]["checkpoint_id"] == checkpoints[-1].checkpoint_id
+
+
 async def test_workflow_run_maps_custom_and_text_events():
     """Custom workflow events and yielded text are mapped to AG-UI events."""
 

From c6f10d8f15fad7d3d28c846d27c8d52337ea421f Mon Sep 17 00:00:00 2001
From: lsmlhi_25 <lhi25wy@163.com>
Date: Wed, 9 Sep 2026 15:15:02 +0800
Subject: [PATCH 3/6] Python: Fix ag-ui lint and ty ignores for #8150 tests

---
 python/packages/ag-ui/tests/ag_ui/test_workflow_run.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
index b2c8e538f24..b3d9f612072 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
@@ -117,6 +117,7 @@ def test_attach_checkpoint_id_to_interrupts_setdefault() -> None:
 async def test_pause_checkpoint_id_ignores_competing_shared_latest() -> None:
     """Prefer this runner's pause checkpoint over a newer shared get_latest() winner."""
     from agent_framework import WorkflowCheckpoint
+
     from agent_framework_ag_ui._workflow_run import (
         _build_run_finished_with_checkpointed_interrupts,
         _pause_checkpoint_id_for_interrupts,
@@ -140,7 +141,7 @@ async def start(self, message: Any, ctx: WorkflowContext) -> None:
         @response_handler
         async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
             del original_request, response
-            await ctx.yield_output("done")  # type: ignore[arg-type]
+            await ctx.yield_output("done")  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
 
     storage = InMemoryCheckpointStorage()
     workflow = WorkflowBuilder(start_executor=ApprovalExecutor(), checkpoint_storage=storage).build()
@@ -208,7 +209,7 @@ async def start(self, message: Any, ctx: WorkflowContext) -> None:
         @response_handler
         async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
             del original_request, response
-            await ctx.yield_output("done")  # type: ignore[arg-type]
+            await ctx.yield_output("done")  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
 
     storage = InMemoryCheckpointStorage()
     workflow = WorkflowBuilder(start_executor=ApprovalExecutor(), checkpoint_storage=storage).build()

From 68f5abe845db05d357ea85f4b34ddb7b6ab8dbb5 Mon Sep 17 00:00:00 2001
From: lsmlhi_25 <lhi25wy@163.com>
Date: Wed, 9 Sep 2026 17:23:08 +0800
Subject: [PATCH 4/6] Python: Core resolve_pause_checkpoint_id for AG-UI
 (#8150)

Move pause-checkpoint selection into Workflow and require a run-scoped
baseline so leftover runner ids are not advertised across runs.
---
 .../agent_framework_ag_ui/_workflow_run.py    |  94 +++++-----------
 .../ag-ui/tests/ag_ui/test_workflow_run.py    |  48 ++++++++
 .../agent_framework/_workflows/_workflow.py   |  71 ++++++++++++
 .../workflow/test_pause_checkpoint_resolve.py | 106 ++++++++++++++++++
 4 files changed, 253 insertions(+), 66 deletions(-)
 create mode 100644 python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py

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 063a06289f0..7d078465ab9 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
@@ -181,88 +181,36 @@ def _attach_checkpoint_id_to_interrupts(
     return attached
 
 
-def _resolve_effective_checkpoint_storage(
-    workflow: Workflow,
-    checkpoint_storage: CheckpointStorage | None,
-) -> CheckpointStorage | None:
-    """Prefer the run argument, else the storage already bound on the workflow runner."""
-    if checkpoint_storage is not None:
-        return checkpoint_storage
-    runner = getattr(workflow, "_runner", None)
-    ctx = getattr(runner, "context", None) if runner is not None else None
-    if ctx is None:
-        return None
-    getter = getattr(ctx, "_get_effective_checkpoint_storage", None)
-    if callable(getter):
-        try:
-            return cast(CheckpointStorage | None, getter())
-        except Exception:  # pragma: no cover - defensive for duck-typed runners
-            return None
-    return None
-
-
-def _runner_previous_checkpoint_id(workflow: Workflow) -> str | None:
-    """Return the checkpoint id last persisted by *this* workflow runner instance."""
-    runner = getattr(workflow, "_runner", None)
-    if runner is None:
-        return None
-    checkpoint_id = getattr(runner, "_previous_checkpoint_id", None)
-    return str(checkpoint_id) if checkpoint_id is not None else None
-
-
 def _interrupt_request_ids(interrupts: list[dict[str, Any]]) -> set[str]:
     return {str(item["id"]) for item in interrupts if item.get("id") is not None}
 
 
-async def _checkpoint_covers_interrupt_ids(
-    storage: CheckpointStorage,
-    checkpoint_id: str,
-    request_ids: set[str],
-) -> bool:
-    """True when the checkpoint's pending request_info set covers ``request_ids``."""
-    if not request_ids:
-        return False
-    try:
-        checkpoint = await storage.load(checkpoint_id)
-    except Exception:  # pragma: no cover - storage/type drift
-        return False
-    pending = getattr(checkpoint, "pending_request_info_events", None) or {}
-    return request_ids.issubset({str(key) for key in dict(pending)})
-
-
 async def _pause_checkpoint_id_for_interrupts(
     *,
     workflow: Workflow,
     checkpoint_storage: CheckpointStorage | None,
     interrupts: list[dict[str, Any]],
     known_checkpoint_id: str | None = None,
+    baseline_checkpoint_id: str | None = None,
 ) -> str | None:
-    """Resolve the pause checkpoint for *this* run's interrupts.
+    """Resolve the pause checkpoint for *this* run's interrupts via core.
 
-    Prefer the runner's last-saved id (scoped to this instance) over shared
-    ``get_latest(workflow_name=...)``, which can race across owners. Validate that
-    the candidate actually records the pending interrupt ids before advertising it.
+    ``baseline_checkpoint_id`` must be the runner id captured before ``workflow.run()`` so
+    a leftover id from a prior run is not advertised when this run did not persist.
     """
     if not interrupts:
         return None
 
-    storage = _resolve_effective_checkpoint_storage(workflow, checkpoint_storage)
-    request_ids = _interrupt_request_ids(interrupts)
-    candidates: list[str] = []
-    runner_id = _runner_previous_checkpoint_id(workflow)
-    if runner_id is not None:
-        candidates.append(runner_id)
-    if known_checkpoint_id is not None and known_checkpoint_id not in candidates:
-        candidates.append(known_checkpoint_id)
-
-    if storage is None:
-        # Without storage we cannot prove coverage; only advertise the id this runner saved.
-        return runner_id
-
-    for candidate in candidates:
-        if await _checkpoint_covers_interrupt_ids(storage, candidate, request_ids):
-            return candidate
-    return None
+    resolve = getattr(workflow, "resolve_pause_checkpoint_id", None)
+    if not callable(resolve):
+        return None
+
+    return await resolve(
+        _interrupt_request_ids(interrupts),
+        checkpoint_storage=checkpoint_storage,
+        known_checkpoint_id=known_checkpoint_id,
+        baseline_checkpoint_id=baseline_checkpoint_id,
+    )
 
 
 async def _build_run_finished_with_checkpointed_interrupts(
@@ -273,6 +221,7 @@ async def _build_run_finished_with_checkpointed_interrupts(
     workflow: Workflow,
     checkpoint_storage: CheckpointStorage | None,
     known_checkpoint_id: str | None = None,
+    baseline_checkpoint_id: str | None = None,
 ) -> Any:
     """Build RUN_FINISHED, attaching the pause checkpoint id to interrupt metadata when available."""
     if not interrupts:
@@ -283,6 +232,7 @@ async def _build_run_finished_with_checkpointed_interrupts(
         checkpoint_storage=checkpoint_storage,
         interrupts=interrupts,
         known_checkpoint_id=known_checkpoint_id,
+        baseline_checkpoint_id=baseline_checkpoint_id,
     )
     return _build_run_finished_event(
         run_id=run_id,
@@ -1279,6 +1229,9 @@ async def run_workflow_stream(
             interrupt_event_value = _workflow_interrupt_event_value(request_payload)
             if interrupt_event_value is not None:
                 yield CustomEvent(name=_INTERRUPT_CARD_EVENT_NAME, value=interrupt_event_value)
+        baseline_checkpoint_id = (
+            workflow.get_last_checkpoint_id() if hasattr(workflow, "get_last_checkpoint_id") else None
+        )
         yield await _build_run_finished_with_checkpointed_interrupts(
             run_id=run_id,
             thread_id=thread_id,
@@ -1286,11 +1239,15 @@ async def run_workflow_stream(
             workflow=workflow,
             checkpoint_storage=checkpoint_storage,
             known_checkpoint_id=checkpoint_id,
+            baseline_checkpoint_id=baseline_checkpoint_id,
         )
         return
 
     if checkpoint_id is None and not responses and not messages:
         yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
+        baseline_checkpoint_id = (
+            workflow.get_last_checkpoint_id() if hasattr(workflow, "get_last_checkpoint_id") else None
+        )
         yield await _build_run_finished_with_checkpointed_interrupts(
             run_id=run_id,
             thread_id=thread_id,
@@ -1298,6 +1255,7 @@ async def run_workflow_stream(
             workflow=workflow,
             checkpoint_storage=checkpoint_storage,
             known_checkpoint_id=checkpoint_id,
+            baseline_checkpoint_id=baseline_checkpoint_id,
         )
         return
 
@@ -1355,6 +1313,8 @@ def _drain_open_blocks() -> list[BaseEvent]:
     if checkpoint_storage is not None or checkpoint_id is not None:
         checkpoint_kwargs = {"checkpoint_storage": checkpoint_storage, "checkpoint_id": checkpoint_id}
 
+    baseline_checkpoint_id = workflow.get_last_checkpoint_id() if hasattr(workflow, "get_last_checkpoint_id") else None
+
     try:
         telemetry_conversation_id = str(supplied_thread_id) if supplied_thread_id is not None else None
         telemetry_context = partial(_use_telemetry_conversation_id, telemetry_conversation_id)
@@ -1411,6 +1371,7 @@ def _drain_open_blocks() -> list[BaseEvent]:
                         workflow=workflow,
                         checkpoint_storage=checkpoint_storage,
                         known_checkpoint_id=checkpoint_id,
+                        baseline_checkpoint_id=baseline_checkpoint_id,
                     )
                     terminal_emitted = True
                 elif state_value not in _TERMINAL_STATES:
@@ -1565,4 +1526,5 @@ def _drain_open_blocks() -> list[BaseEvent]:
             workflow=workflow,
             checkpoint_storage=checkpoint_storage,
             known_checkpoint_id=checkpoint_id,
+            baseline_checkpoint_id=baseline_checkpoint_id,
         )
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
index b3d9f612072..4253289919c 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
@@ -169,10 +169,12 @@ async def handle_approval(self, original_request: Content, response: Content, ct
     assert latest is not None
     assert latest.checkpoint_id == competing.checkpoint_id
 
+    # baseline=None: treat as first advertisement after the pause run already advanced the runner id.
     resolved = await _pause_checkpoint_id_for_interrupts(
         workflow=workflow,
         checkpoint_storage=storage,
         interrupts=interrupt_payload,
+        baseline_checkpoint_id=None,
     )
     assert resolved == pause_id
 
@@ -182,6 +184,7 @@ async def handle_approval(self, original_request: Content, response: Content, ct
         interrupts=interrupt_payload,
         workflow=workflow,
         checkpoint_storage=storage,
+        baseline_checkpoint_id=None,
     )
     rebuilt_interrupts = _interrupts_from_run_finished(rebuilt)
     assert rebuilt_interrupts[0]["metadata"]["agent_framework"]["checkpoint_id"] == pause_id
@@ -224,6 +227,51 @@ async def handle_approval(self, original_request: Content, response: Content, ct
     assert interrupt_payload[0]["metadata"]["agent_framework"]["checkpoint_id"] == checkpoints[-1].checkpoint_id
 
 
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_is_run_scoped_without_storage() -> None:
+    """Stale runner ids must not be advertised when baseline shows this run did not persist."""
+
+    class ApprovalExecutor(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="approval_executor")
+
+        @handler
+        async def start(self, message: Any, ctx: WorkflowContext) -> None:
+            del message
+            await ctx.request_info("need-input", str, request_id="req-1")
+
+        @response_handler
+        async def handle(self, original_request: str, response: str, ctx: WorkflowContext) -> None:
+            del original_request, response
+            await ctx.yield_output("done")  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
+
+    workflow = WorkflowBuilder(start_executor=ApprovalExecutor()).build()
+    # Simulate a leftover id from a prior run on a workflow with no checkpoint storage.
+    workflow._runner._previous_checkpoint_id = "stale-from-prior-run"  # pyright: ignore[reportPrivateUsage]
+
+    from agent_framework_ag_ui._workflow_run import _pause_checkpoint_id_for_interrupts
+
+    interrupts = [{"id": "req-1", "value": "need-input"}]
+    assert (
+        await _pause_checkpoint_id_for_interrupts(
+            workflow=workflow,
+            checkpoint_storage=None,
+            interrupts=interrupts,
+            baseline_checkpoint_id="stale-from-prior-run",
+        )
+        is None
+    )
+    assert (
+        await _pause_checkpoint_id_for_interrupts(
+            workflow=workflow,
+            checkpoint_storage=None,
+            interrupts=interrupts,
+            baseline_checkpoint_id=None,
+        )
+        == "stale-from-prior-run"
+    )
+
+
 async def test_workflow_run_maps_custom_and_text_events():
     """Custom workflow events and yielded text are mapped to AG-UI events."""
 
diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py
index 6e9f256b9bd..74c4b6c1080 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow.py
@@ -1269,6 +1269,77 @@ def output_types(self) -> list[type[Any] | types.UnionType]:
 
         return list(output_types)
 
+    def get_last_checkpoint_id(self) -> str | None:
+        """Return the checkpoint id last persisted or restored by this workflow runner.
+
+        Capture this value before ``run()`` when a host needs a run-scoped pause id:
+        after the run, only a different id indicates *this* run advanced the chain.
+        """
+        checkpoint_id = self._runner._previous_checkpoint_id  # pyright: ignore[reportPrivateUsage]
+        return str(checkpoint_id) if checkpoint_id is not None else None
+
+    async def resolve_pause_checkpoint_id(
+        self,
+        request_ids: Collection[str],
+        *,
+        checkpoint_storage: CheckpointStorage | None = None,
+        known_checkpoint_id: str | None = None,
+        baseline_checkpoint_id: str | None = None,
+    ) -> str | None:
+        """Resolve the persisted pause checkpoint that covers ``request_ids``.
+
+        Prefers this runner's last-saved id when it advanced past ``baseline_checkpoint_id``
+        (the id captured before ``run()``), over shared ``get_latest(workflow_name=...)``.
+        Storage precedence is the run argument, else the runner's effective
+        (runtime / builder) storage. When storage is available, candidates are accepted
+        only if their ``pending_request_info_events`` cover ``request_ids``.
+
+        Args:
+            request_ids: Pending request_info ids that must be present on the checkpoint.
+            checkpoint_storage: Optional storage override for this lookup.
+            known_checkpoint_id: Fallback id (for example a cold-resume short-circuit).
+            baseline_checkpoint_id: Runner checkpoint id captured before the run that
+                produced these requests. When set (including ``None`` as “no prior id”),
+                the runner candidate is used only if ``get_last_checkpoint_id()`` differs.
+
+        Returns:
+            A checkpoint id suitable for durable resume, or ``None`` when none is safe.
+        """
+        ids = {str(request_id) for request_id in request_ids if request_id is not None and str(request_id)}
+        if not ids:
+            return None
+
+        storage = checkpoint_storage
+        if storage is None:
+            storage = self._runner.context._get_effective_checkpoint_storage()  # pyright: ignore[reportPrivateUsage]
+
+        current = self.get_last_checkpoint_id()
+        # Run-scoped: do not advertise a pre-run leftover when this run did not persist.
+        runner_candidate: str | None = None
+        if current is not None and current != baseline_checkpoint_id:
+            runner_candidate = current
+
+        candidates: list[str] = []
+        if runner_candidate is not None:
+            candidates.append(runner_candidate)
+        if known_checkpoint_id is not None and known_checkpoint_id not in candidates:
+            candidates.append(str(known_checkpoint_id))
+
+        if storage is None:
+            # Without storage we cannot prove coverage; only advertise a run-scoped runner id.
+            return runner_candidate
+
+        for candidate in candidates:
+            try:
+                checkpoint = await storage.load(candidate)
+            except Exception:  # pragma: no cover - storage/type drift
+                logger.debug("Could not load pause checkpoint candidate %s", candidate, exc_info=True)
+                continue
+            pending = getattr(checkpoint, "pending_request_info_events", None) or {}
+            if ids.issubset({str(key) for key in dict(pending)}):
+                return candidate
+        return None
+
     async def cancel_pending_requests(
         self,
         request_ids: Collection[str],
diff --git a/python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py b/python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py
new file mode 100644
index 00000000000..5273bad577e
--- /dev/null
+++ b/python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py
@@ -0,0 +1,106 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for Workflow.resolve_pause_checkpoint_id / get_last_checkpoint_id."""
+
+from typing import Any
+
+import pytest
+
+from agent_framework import (
+    Content,
+    Executor,
+    InMemoryCheckpointStorage,
+    WorkflowBuilder,
+    WorkflowCheckpoint,
+    WorkflowContext,
+    handler,
+    response_handler,
+)
+
+
+class _ApprovalExecutor(Executor):
+    def __init__(self) -> None:
+        super().__init__(id="approval_executor")
+
+    @handler
+    async def start(self, message: Any, ctx: WorkflowContext) -> None:
+        del message
+        function_call = Content.from_function_call(
+            call_id="refund-call",
+            name="submit_refund",
+            arguments={"order_id": "12345"},
+        )
+        approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call)
+        await ctx.request_info(approval_request, Content, request_id="approval-1")
+
+    @response_handler
+    async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
+        del original_request, response
+        await ctx.yield_output("done")  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_prefers_runner_over_shared_latest() -> None:
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
+    baseline = workflow.get_last_checkpoint_id()
+
+    async for _ in workflow.run("go", stream=True):
+        pass
+
+    pause_id = workflow.get_last_checkpoint_id()
+    assert pause_id is not None
+    assert pause_id != baseline
+
+    competing = WorkflowCheckpoint(
+        workflow_name=workflow.name,
+        graph_signature_hash="competing",
+        pending_request_info_events={},
+        timestamp="9999-01-01T00:00:00+00:00",
+    )
+    await storage.save(competing)
+
+    resolved = await workflow.resolve_pause_checkpoint_id(
+        {"approval-1"},
+        checkpoint_storage=storage,
+        baseline_checkpoint_id=baseline,
+    )
+    assert resolved == pause_id
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_requires_run_scoped_change_without_storage() -> None:
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor()).build()
+    workflow._runner._previous_checkpoint_id = "leftover"  # pyright: ignore[reportPrivateUsage]
+
+    assert (
+        await workflow.resolve_pause_checkpoint_id(
+            {"approval-1"},
+            baseline_checkpoint_id="leftover",
+        )
+        is None
+    )
+    assert (
+        await workflow.resolve_pause_checkpoint_id(
+            {"approval-1"},
+            baseline_checkpoint_id=None,
+        )
+        == "leftover"
+    )
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_uses_builder_storage() -> None:
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
+    baseline = workflow.get_last_checkpoint_id()
+
+    async for _ in workflow.run("go", stream=True):
+        pass
+
+    pause_id = workflow.get_last_checkpoint_id()
+    resolved = await workflow.resolve_pause_checkpoint_id(
+        {"approval-1"},
+        baseline_checkpoint_id=baseline,
+    )
+    assert resolved == pause_id

From d4b60ebffcf0a385cf98d937065e09cd539e54d0 Mon Sep 17 00:00:00 2001
From: lsmlhi_25 <lhi25wy@163.com>
Date: Thu, 10 Sep 2026 01:11:34 +0800
Subject: [PATCH 5/6] fix(python): resolve pyright failures in pause checkpoint
 lookup

Unblocks Package Checks / merge-gatekeeper on PR #8163.
---
 .../agent_framework_ag_ui/_workflow_run.py     |  9 +++++++--
 .../agent_framework/_workflows/_workflow.py    | 18 ++++++++++++------
 2 files changed, 19 insertions(+), 8 deletions(-)

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 7d078465ab9..4f2668efeed 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
@@ -8,7 +8,7 @@
 import json
 import logging
 import uuid
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from functools import partial
 from types import UnionType
 from typing import Any, Union, cast, get_args, get_origin, get_type_hints
@@ -205,7 +205,12 @@ async def _pause_checkpoint_id_for_interrupts(
     if not callable(resolve):
         return None
 
-    return await resolve(
+    # getattr returns a plain object to the type checker; cast to an awaitable callable.
+    resolve_fn = cast(
+        Callable[..., Awaitable[str | None]],
+        resolve,
+    )
+    return await resolve_fn(
         _interrupt_request_ids(interrupts),
         checkpoint_storage=checkpoint_storage,
         known_checkpoint_id=known_checkpoint_id,
diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py
index 74c4b6c1080..47385df13ad 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow.py
@@ -1305,13 +1305,14 @@ async def resolve_pause_checkpoint_id(
         Returns:
             A checkpoint id suitable for durable resume, or ``None`` when none is safe.
         """
-        ids = {str(request_id) for request_id in request_ids if request_id is not None and str(request_id)}
+        ids = {str(request_id) for request_id in request_ids if request_id}
         if not ids:
             return None
 
+        # Prefer explicit storage; otherwise load via public RunnerContext APIs
+        # (Protocol has no private `_get_effective_checkpoint_storage`).
         storage = checkpoint_storage
-        if storage is None:
-            storage = self._runner.context._get_effective_checkpoint_storage()  # pyright: ignore[reportPrivateUsage]
+        use_context_storage = storage is None and self._runner.context.has_checkpointing()
 
         current = self.get_last_checkpoint_id()
         # Run-scoped: do not advertise a pre-run leftover when this run did not persist.
@@ -1325,17 +1326,22 @@ async def resolve_pause_checkpoint_id(
         if known_checkpoint_id is not None and known_checkpoint_id not in candidates:
             candidates.append(str(known_checkpoint_id))
 
-        if storage is None:
+        if storage is None and not use_context_storage:
             # Without storage we cannot prove coverage; only advertise a run-scoped runner id.
             return runner_candidate
 
         for candidate in candidates:
             try:
-                checkpoint = await storage.load(candidate)
+                if storage is not None:
+                    checkpoint = await storage.load(candidate)
+                else:
+                    checkpoint = await self._runner.context.load_checkpoint(candidate)
             except Exception:  # pragma: no cover - storage/type drift
                 logger.debug("Could not load pause checkpoint candidate %s", candidate, exc_info=True)
                 continue
-            pending = getattr(checkpoint, "pending_request_info_events", None) or {}
+            if checkpoint is None:
+                continue
+            pending = checkpoint.pending_request_info_events or {}
             if ids.issubset({str(key) for key in dict(pending)}):
                 return candidate
         return None

From d42a8b3da76f60d196a2a331118854515e474d28 Mon Sep 17 00:00:00 2001
From: LI <2484593937@qq.com>
Date: Fri, 11 Sep 2026 21:36:09 +0800
Subject: [PATCH 6/6] fix(ag-ui): address #8163 review on pause checkpoint
 ownership (#2)

- Workflow captures run baseline / restored id inside run(); resolve uses them
  so callers need not thread baseline_checkpoint_id.
- Exclude restored checkpoint from pause candidates after resume.
- Drop issue reference in docstring; emit RUN_FINISHED via existing
  _build_run_finished_event after attaching pause id.
- Move core pause-resolve coverage into test_workflow.py (no new test file).
---
 .../agent_framework_ag_ui/_workflow_run.py    | 101 ++++++++---------
 .../ag-ui/tests/ag_ui/test_workflow_run.py    |  23 ++--
 .../agent_framework/_workflows/_workflow.py   |  48 +++++---
 .../workflow/test_pause_checkpoint_resolve.py | 106 ------------------
 .../core/tests/workflow/test_workflow.py      | 106 ++++++++++++++++++
 5 files changed, 199 insertions(+), 185 deletions(-)
 delete mode 100644 python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py

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 4f2668efeed..6eeca4b3bbc 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
@@ -60,6 +60,8 @@
 
 logger = logging.getLogger(__name__)
 
+_BASELINE_OMITTED = object()
+
 
 _TERMINAL_STATES: set[str] = {
     WorkflowRunState.IDLE.value,
@@ -152,7 +154,7 @@ def _attach_checkpoint_id_to_interrupts(
     interrupts: list[dict[str, Any]],
     checkpoint_id: str | None,
 ) -> list[dict[str, Any]]:
-    """Attach ``checkpoint_id`` to each interrupt's ``metadata.agent_framework`` (issue #8150).
+    """Attach ``checkpoint_id`` to each interrupt's ``metadata.agent_framework``.
 
     Multi-worker hosts need the pause checkpoint on the wire so the next resume can pass
     ``forwardedProps.checkpoint_id`` without a side-channel lookup. No-op when checkpointing
@@ -191,12 +193,13 @@ async def _pause_checkpoint_id_for_interrupts(
     checkpoint_storage: CheckpointStorage | None,
     interrupts: list[dict[str, Any]],
     known_checkpoint_id: str | None = None,
-    baseline_checkpoint_id: str | None = None,
+    baseline_checkpoint_id: Any = _BASELINE_OMITTED,
 ) -> str | None:
     """Resolve the pause checkpoint for *this* run's interrupts via core.
 
-    ``baseline_checkpoint_id`` must be the runner id captured before ``workflow.run()`` so
-    a leftover id from a prior run is not advertised when this run did not persist.
+    When ``baseline_checkpoint_id`` is omitted, core uses the baseline captured at
+    ``workflow.run()`` start. Pass ``None`` explicitly for short-circuit paths that
+    did not call ``run()`` and should advertise the current runner id when present.
     """
     if not interrupts:
         return None
@@ -210,28 +213,26 @@ async def _pause_checkpoint_id_for_interrupts(
         Callable[..., Awaitable[str | None]],
         resolve,
     )
-    return await resolve_fn(
-        _interrupt_request_ids(interrupts),
-        checkpoint_storage=checkpoint_storage,
-        known_checkpoint_id=known_checkpoint_id,
-        baseline_checkpoint_id=baseline_checkpoint_id,
-    )
+    kwargs: dict[str, Any] = {
+        "checkpoint_storage": checkpoint_storage,
+        "known_checkpoint_id": known_checkpoint_id,
+    }
+    if baseline_checkpoint_id is not _BASELINE_OMITTED:
+        kwargs["baseline_checkpoint_id"] = baseline_checkpoint_id
+    return await resolve_fn(_interrupt_request_ids(interrupts), **kwargs)
 
 
-async def _build_run_finished_with_checkpointed_interrupts(
+async def _interrupts_with_pause_checkpoint(
     *,
-    run_id: str,
-    thread_id: str,
     interrupts: list[dict[str, Any]],
     workflow: Workflow,
     checkpoint_storage: CheckpointStorage | None,
     known_checkpoint_id: str | None = None,
-    baseline_checkpoint_id: str | None = None,
-) -> Any:
-    """Build RUN_FINISHED, attaching the pause checkpoint id to interrupt metadata when available."""
+    baseline_checkpoint_id: Any = _BASELINE_OMITTED,
+) -> list[dict[str, Any]]:
+    """Attach a run-scoped pause checkpoint id to interrupts when available."""
     if not interrupts:
-        return _build_run_finished_event(run_id=run_id, thread_id=thread_id, interrupts=interrupts)
-
+        return interrupts
     pause_checkpoint_id = await _pause_checkpoint_id_for_interrupts(
         workflow=workflow,
         checkpoint_storage=checkpoint_storage,
@@ -239,11 +240,7 @@ async def _build_run_finished_with_checkpointed_interrupts(
         known_checkpoint_id=known_checkpoint_id,
         baseline_checkpoint_id=baseline_checkpoint_id,
     )
-    return _build_run_finished_event(
-        run_id=run_id,
-        thread_id=thread_id,
-        interrupts=_attach_checkpoint_id_to_interrupts(interrupts, pause_checkpoint_id),
-    )
+    return _attach_checkpoint_id_to_interrupts(interrupts, pause_checkpoint_id)
 
 
 async def _pending_request_events(workflow: Workflow) -> dict[str, Any]:
@@ -1234,33 +1231,29 @@ async def run_workflow_stream(
             interrupt_event_value = _workflow_interrupt_event_value(request_payload)
             if interrupt_event_value is not None:
                 yield CustomEvent(name=_INTERRUPT_CARD_EVENT_NAME, value=interrupt_event_value)
-        baseline_checkpoint_id = (
-            workflow.get_last_checkpoint_id() if hasattr(workflow, "get_last_checkpoint_id") else None
-        )
-        yield await _build_run_finished_with_checkpointed_interrupts(
+        yield _build_run_finished_event(
             run_id=run_id,
             thread_id=thread_id,
-            interrupts=pending_interrupts,
-            workflow=workflow,
-            checkpoint_storage=checkpoint_storage,
-            known_checkpoint_id=checkpoint_id,
-            baseline_checkpoint_id=baseline_checkpoint_id,
+            interrupts=await _interrupts_with_pause_checkpoint(
+                interrupts=pending_interrupts,
+                workflow=workflow,
+                checkpoint_storage=checkpoint_storage,
+                baseline_checkpoint_id=None,
+            ),
         )
         return
 
     if checkpoint_id is None and not responses and not messages:
         yield RunStartedEvent(run_id=run_id, thread_id=thread_id)
-        baseline_checkpoint_id = (
-            workflow.get_last_checkpoint_id() if hasattr(workflow, "get_last_checkpoint_id") else None
-        )
-        yield await _build_run_finished_with_checkpointed_interrupts(
+        yield _build_run_finished_event(
             run_id=run_id,
             thread_id=thread_id,
-            interrupts=pending_interrupts,
-            workflow=workflow,
-            checkpoint_storage=checkpoint_storage,
-            known_checkpoint_id=checkpoint_id,
-            baseline_checkpoint_id=baseline_checkpoint_id,
+            interrupts=await _interrupts_with_pause_checkpoint(
+                interrupts=pending_interrupts,
+                workflow=workflow,
+                checkpoint_storage=checkpoint_storage,
+                baseline_checkpoint_id=None,
+            ),
         )
         return
 
@@ -1318,8 +1311,6 @@ def _drain_open_blocks() -> list[BaseEvent]:
     if checkpoint_storage is not None or checkpoint_id is not None:
         checkpoint_kwargs = {"checkpoint_storage": checkpoint_storage, "checkpoint_id": checkpoint_id}
 
-    baseline_checkpoint_id = workflow.get_last_checkpoint_id() if hasattr(workflow, "get_last_checkpoint_id") else None
-
     try:
         telemetry_conversation_id = str(supplied_thread_id) if supplied_thread_id is not None else None
         telemetry_context = partial(_use_telemetry_conversation_id, telemetry_conversation_id)
@@ -1369,14 +1360,14 @@ def _drain_open_blocks() -> list[BaseEvent]:
                         yield end_event
                     if not interrupts:
                         interrupts.extend(_interrupts_from_pending_requests(await _pending_request_events(workflow)))
-                    yield await _build_run_finished_with_checkpointed_interrupts(
+                    yield _build_run_finished_event(
                         run_id=run_id,
                         thread_id=thread_id,
-                        interrupts=interrupts,
-                        workflow=workflow,
-                        checkpoint_storage=checkpoint_storage,
-                        known_checkpoint_id=checkpoint_id,
-                        baseline_checkpoint_id=baseline_checkpoint_id,
+                        interrupts=await _interrupts_with_pause_checkpoint(
+                            interrupts=interrupts,
+                            workflow=workflow,
+                            checkpoint_storage=checkpoint_storage,
+                        ),
                     )
                     terminal_emitted = True
                 elif state_value not in _TERMINAL_STATES:
@@ -1524,12 +1515,12 @@ def _drain_open_blocks() -> list[BaseEvent]:
     if not terminal_emitted and not run_error_emitted:
         if not interrupts:
             interrupts.extend(_interrupts_from_pending_requests(await _pending_request_events(workflow)))
-        yield await _build_run_finished_with_checkpointed_interrupts(
+        yield _build_run_finished_event(
             run_id=run_id,
             thread_id=thread_id,
-            interrupts=interrupts,
-            workflow=workflow,
-            checkpoint_storage=checkpoint_storage,
-            known_checkpoint_id=checkpoint_id,
-            baseline_checkpoint_id=baseline_checkpoint_id,
+            interrupts=await _interrupts_with_pause_checkpoint(
+                interrupts=interrupts,
+                workflow=workflow,
+                checkpoint_storage=checkpoint_storage,
+            ),
         )
diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
index 4253289919c..3f163cda361 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py
@@ -118,8 +118,9 @@ async def test_pause_checkpoint_id_ignores_competing_shared_latest() -> None:
     """Prefer this runner's pause checkpoint over a newer shared get_latest() winner."""
     from agent_framework import WorkflowCheckpoint
 
+    from agent_framework_ag_ui._run_common import _build_run_finished_event
     from agent_framework_ag_ui._workflow_run import (
-        _build_run_finished_with_checkpointed_interrupts,
+        _interrupts_with_pause_checkpoint,
         _pause_checkpoint_id_for_interrupts,
     )
 
@@ -169,22 +170,22 @@ async def handle_approval(self, original_request: Content, response: Content, ct
     assert latest is not None
     assert latest.checkpoint_id == competing.checkpoint_id
 
-    # baseline=None: treat as first advertisement after the pause run already advanced the runner id.
+    # Workflow-owned baseline from the pause run should already allow advertising pause_id.
     resolved = await _pause_checkpoint_id_for_interrupts(
         workflow=workflow,
         checkpoint_storage=storage,
         interrupts=interrupt_payload,
-        baseline_checkpoint_id=None,
     )
     assert resolved == pause_id
 
-    rebuilt = await _build_run_finished_with_checkpointed_interrupts(
-        run_id="run-1",
-        thread_id="thread-1",
-        interrupts=interrupt_payload,
-        workflow=workflow,
-        checkpoint_storage=storage,
-        baseline_checkpoint_id=None,
+    rebuilt = _build_run_finished_event(
+        "run-1",
+        "thread-1",
+        interrupts=await _interrupts_with_pause_checkpoint(
+            interrupts=interrupt_payload,
+            workflow=workflow,
+            checkpoint_storage=storage,
+        ),
     )
     rebuilt_interrupts = _interrupts_from_run_finished(rebuilt)
     assert rebuilt_interrupts[0]["metadata"]["agent_framework"]["checkpoint_id"] == pause_id
@@ -252,12 +253,12 @@ async def handle(self, original_request: str, response: str, ctx: WorkflowContex
     from agent_framework_ag_ui._workflow_run import _pause_checkpoint_id_for_interrupts
 
     interrupts = [{"id": "req-1", "value": "need-input"}]
+    workflow._run_baseline_checkpoint_id = "stale-from-prior-run"  # pyright: ignore[reportPrivateUsage]
     assert (
         await _pause_checkpoint_id_for_interrupts(
             workflow=workflow,
             checkpoint_storage=None,
             interrupts=interrupts,
-            baseline_checkpoint_id="stale-from-prior-run",
         )
         is None
     )
diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py
index 47385df13ad..106807a3450 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow.py
@@ -394,6 +394,12 @@ def __init__(
         # so a subsequent ``run()`` is allowed.
         self._active_run: weakref.ref[ResponseStream[WorkflowEvent, WorkflowRunResult]] | None = None
 
+        # Run-scoped pause checkpoint bookkeeping (owned by Workflow, not callers).
+        # Captured at the start of each ``_run_core`` so ``resolve_pause_checkpoint_id``
+        # can tell a newly persisted pause from a leftover / restored id.
+        self._run_baseline_checkpoint_id: str | None = None
+        self._restored_checkpoint_id: str | None = None
+
     @property
     def status(self) -> WorkflowRunState:
         """Return the current run-level status of this workflow instance.
@@ -882,6 +888,12 @@ async def _run_core(
         if checkpoint_storage is not None:
             self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
 
+        # Own the run boundary for pause-checkpoint resolution: capture the runner id
+        # before this run advances it, and remember an incoming restore so it is not
+        # re-advertised if a later pause save fails.
+        self._run_baseline_checkpoint_id = self.get_last_checkpoint_id()
+        self._restored_checkpoint_id = str(checkpoint_id) if checkpoint_id is not None else None
+
         try:
             # Async validation: a fresh-message run is only allowed when the
             # runner context has fully drained from any prior run. If it still
@@ -1270,11 +1282,7 @@ def output_types(self) -> list[type[Any] | types.UnionType]:
         return list(output_types)
 
     def get_last_checkpoint_id(self) -> str | None:
-        """Return the checkpoint id last persisted or restored by this workflow runner.
-
-        Capture this value before ``run()`` when a host needs a run-scoped pause id:
-        after the run, only a different id indicates *this* run advanced the chain.
-        """
+        """Return the checkpoint id last persisted or restored by this workflow runner."""
         checkpoint_id = self._runner._previous_checkpoint_id  # pyright: ignore[reportPrivateUsage]
         return str(checkpoint_id) if checkpoint_id is not None else None
 
@@ -1284,23 +1292,26 @@ async def resolve_pause_checkpoint_id(
         *,
         checkpoint_storage: CheckpointStorage | None = None,
         known_checkpoint_id: str | None = None,
-        baseline_checkpoint_id: str | None = None,
+        baseline_checkpoint_id: str | None | object = _MISSING,
     ) -> str | None:
         """Resolve the persisted pause checkpoint that covers ``request_ids``.
 
-        Prefers this runner's last-saved id when it advanced past ``baseline_checkpoint_id``
-        (the id captured before ``run()``), over shared ``get_latest(workflow_name=...)``.
+        Prefers this runner's last-saved id when it advanced past the run baseline
+        (captured automatically at ``run()`` start), over shared ``get_latest(workflow_name=...)``.
         Storage precedence is the run argument, else the runner's effective
         (runtime / builder) storage. When storage is available, candidates are accepted
         only if their ``pending_request_info_events`` cover ``request_ids``.
 
+        An incoming restored checkpoint id is excluded so a failed pause save after
+        resume cannot re-advertise pre-response state.
+
         Args:
             request_ids: Pending request_info ids that must be present on the checkpoint.
             checkpoint_storage: Optional storage override for this lookup.
             known_checkpoint_id: Fallback id (for example a cold-resume short-circuit).
-            baseline_checkpoint_id: Runner checkpoint id captured before the run that
-                produced these requests. When set (including ``None`` as “no prior id”),
-                the runner candidate is used only if ``get_last_checkpoint_id()`` differs.
+            baseline_checkpoint_id: Optional override for the pre-run runner id. When omitted,
+                uses the baseline captured by the most recent ``run()``. Pass ``None``
+                explicitly to treat any current runner id as newly advanced.
 
         Returns:
             A checkpoint id suitable for durable resume, or ``None`` when none is safe.
@@ -1309,21 +1320,32 @@ async def resolve_pause_checkpoint_id(
         if not ids:
             return None
 
+        if baseline_checkpoint_id is _MISSING:
+            baseline_checkpoint_id = self._run_baseline_checkpoint_id
+
         # Prefer explicit storage; otherwise load via public RunnerContext APIs
         # (Protocol has no private `_get_effective_checkpoint_storage`).
         storage = checkpoint_storage
         use_context_storage = storage is None and self._runner.context.has_checkpointing()
 
         current = self.get_last_checkpoint_id()
+        excluded: set[str] = set()
+        if self._restored_checkpoint_id is not None:
+            excluded.add(self._restored_checkpoint_id)
+
         # Run-scoped: do not advertise a pre-run leftover when this run did not persist.
         runner_candidate: str | None = None
-        if current is not None and current != baseline_checkpoint_id:
+        if current is not None and current != baseline_checkpoint_id and current not in excluded:
             runner_candidate = current
 
         candidates: list[str] = []
         if runner_candidate is not None:
             candidates.append(runner_candidate)
-        if known_checkpoint_id is not None and known_checkpoint_id not in candidates:
+        if (
+            known_checkpoint_id is not None
+            and known_checkpoint_id not in candidates
+            and known_checkpoint_id not in excluded
+        ):
             candidates.append(str(known_checkpoint_id))
 
         if storage is None and not use_context_storage:
diff --git a/python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py b/python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py
deleted file mode 100644
index 5273bad577e..00000000000
--- a/python/packages/core/tests/workflow/test_pause_checkpoint_resolve.py
+++ /dev/null
@@ -1,106 +0,0 @@
-# Copyright (c) Microsoft. All rights reserved.
-
-"""Tests for Workflow.resolve_pause_checkpoint_id / get_last_checkpoint_id."""
-
-from typing import Any
-
-import pytest
-
-from agent_framework import (
-    Content,
-    Executor,
-    InMemoryCheckpointStorage,
-    WorkflowBuilder,
-    WorkflowCheckpoint,
-    WorkflowContext,
-    handler,
-    response_handler,
-)
-
-
-class _ApprovalExecutor(Executor):
-    def __init__(self) -> None:
-        super().__init__(id="approval_executor")
-
-    @handler
-    async def start(self, message: Any, ctx: WorkflowContext) -> None:
-        del message
-        function_call = Content.from_function_call(
-            call_id="refund-call",
-            name="submit_refund",
-            arguments={"order_id": "12345"},
-        )
-        approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call)
-        await ctx.request_info(approval_request, Content, request_id="approval-1")
-
-    @response_handler
-    async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
-        del original_request, response
-        await ctx.yield_output("done")  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
-
-
-@pytest.mark.asyncio
-async def test_resolve_pause_checkpoint_id_prefers_runner_over_shared_latest() -> None:
-    storage = InMemoryCheckpointStorage()
-    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
-    baseline = workflow.get_last_checkpoint_id()
-
-    async for _ in workflow.run("go", stream=True):
-        pass
-
-    pause_id = workflow.get_last_checkpoint_id()
-    assert pause_id is not None
-    assert pause_id != baseline
-
-    competing = WorkflowCheckpoint(
-        workflow_name=workflow.name,
-        graph_signature_hash="competing",
-        pending_request_info_events={},
-        timestamp="9999-01-01T00:00:00+00:00",
-    )
-    await storage.save(competing)
-
-    resolved = await workflow.resolve_pause_checkpoint_id(
-        {"approval-1"},
-        checkpoint_storage=storage,
-        baseline_checkpoint_id=baseline,
-    )
-    assert resolved == pause_id
-
-
-@pytest.mark.asyncio
-async def test_resolve_pause_checkpoint_id_requires_run_scoped_change_without_storage() -> None:
-    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor()).build()
-    workflow._runner._previous_checkpoint_id = "leftover"  # pyright: ignore[reportPrivateUsage]
-
-    assert (
-        await workflow.resolve_pause_checkpoint_id(
-            {"approval-1"},
-            baseline_checkpoint_id="leftover",
-        )
-        is None
-    )
-    assert (
-        await workflow.resolve_pause_checkpoint_id(
-            {"approval-1"},
-            baseline_checkpoint_id=None,
-        )
-        == "leftover"
-    )
-
-
-@pytest.mark.asyncio
-async def test_resolve_pause_checkpoint_id_uses_builder_storage() -> None:
-    storage = InMemoryCheckpointStorage()
-    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
-    baseline = workflow.get_last_checkpoint_id()
-
-    async for _ in workflow.run("go", stream=True):
-        pass
-
-    pause_id = workflow.get_last_checkpoint_id()
-    resolved = await workflow.resolve_pause_checkpoint_id(
-        {"approval-1"},
-        baseline_checkpoint_id=baseline,
-    )
-    assert resolved == pause_id
diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py
index 014694b9b80..64b310f3742 100644
--- a/python/packages/core/tests/workflow/test_workflow.py
+++ b/python/packages/core/tests/workflow/test_workflow.py
@@ -26,6 +26,7 @@
     Message,
     ResponseStream,
     WorkflowBuilder,
+    WorkflowCheckpoint,
     WorkflowCheckpointException,
     WorkflowContext,
     WorkflowConvergenceException,
@@ -1699,3 +1700,108 @@ async def test_output_executors_filtering_with_run_responses_streaming() -> None
 
 
 # endregion
+
+
+# ---------------------------------------------------------------------------
+# Pause checkpoint resolution (AG-UI interrupt metadata)
+# ---------------------------------------------------------------------------
+
+class _ApprovalExecutor(Executor):
+    def __init__(self) -> None:
+        super().__init__(id="approval_executor")
+
+    @handler
+    async def start(self, message: Any, ctx: WorkflowContext) -> None:
+        del message
+        function_call = Content.from_function_call(
+            call_id="refund-call",
+            name="submit_refund",
+            arguments={"order_id": "12345"},
+        )
+        approval_request = Content.from_function_approval_request(id="approval-1", function_call=function_call)
+        await ctx.request_info(approval_request, Content, request_id="approval-1")
+
+    @response_handler
+    async def handle_approval(self, original_request: Content, response: Content, ctx: WorkflowContext) -> None:
+        del original_request, response
+        await ctx.yield_output("done")  # type: ignore[arg-type]  # pyrefly: ignore[bad-argument-type]  # ty: ignore[invalid-argument-type]
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_prefers_runner_over_shared_latest() -> None:
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
+    async for _ in workflow.run("go", stream=True):
+        pass
+
+    pause_id = workflow.get_last_checkpoint_id()
+    assert pause_id is not None
+
+    competing = WorkflowCheckpoint(
+        workflow_name=workflow.name,
+        graph_signature_hash="competing",
+        pending_request_info_events={},
+        timestamp="9999-01-01T00:00:00+00:00",
+    )
+    await storage.save(competing)
+
+    # Workflow owns the run baseline captured at run() start.
+    resolved = await workflow.resolve_pause_checkpoint_id(
+        {"approval-1"},
+        checkpoint_storage=storage,
+    )
+    assert resolved == pause_id
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_requires_run_scoped_change_without_storage() -> None:
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor()).build()
+    workflow._runner._previous_checkpoint_id = "leftover"  # pyright: ignore[reportPrivateUsage]
+    workflow._run_baseline_checkpoint_id = "leftover"  # pyright: ignore[reportPrivateUsage]
+
+    assert await workflow.resolve_pause_checkpoint_id({"approval-1"}) is None
+    assert (
+        await workflow.resolve_pause_checkpoint_id(
+            {"approval-1"},
+            baseline_checkpoint_id=None,
+        )
+        == "leftover"
+    )
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_excludes_restored_checkpoint() -> None:
+    """After resume, the restored id must not be re-advertised if a new pause save fails."""
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
+
+    async for _ in workflow.run("go", stream=True):
+        pass
+    restored = workflow.get_last_checkpoint_id()
+    assert restored is not None
+
+    # Simulate a resume that restored `restored` but did not persist a newer pause.
+    workflow._run_baseline_checkpoint_id = restored  # pyright: ignore[reportPrivateUsage]
+    workflow._restored_checkpoint_id = restored  # pyright: ignore[reportPrivateUsage]
+    workflow._runner._previous_checkpoint_id = restored  # pyright: ignore[reportPrivateUsage]
+
+    assert (
+        await workflow.resolve_pause_checkpoint_id(
+            {"approval-1"},
+            checkpoint_storage=storage,
+            known_checkpoint_id=restored,
+        )
+        is None
+    )
+
+
+@pytest.mark.asyncio
+async def test_resolve_pause_checkpoint_id_uses_builder_storage() -> None:
+    storage = InMemoryCheckpointStorage()
+    workflow = WorkflowBuilder(start_executor=_ApprovalExecutor(), checkpoint_storage=storage).build()
+    async for _ in workflow.run("go", stream=True):
+        pass
+
+    pause_id = workflow.get_last_checkpoint_id()
+    resolved = await workflow.resolve_pause_checkpoint_id({"approval-1"})
+    assert resolved == pause_id