fix(runtime): reject delta resumes that discard pending tools (#7441)

* fix(runtime): reject delta resumes that discard pending tools

* fix(runtime): preserve pending head work on rejected delta replay

* fix(runtime): preserve non-Send checkpoint tasks on resume

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -551,6 +551,13 @@ The checkpoint storage settings `database.checkpoint_channel_mode` and
 both are frozen when the process first builds an agent (including through
 `DeerFlowClient`) and require a process restart to change safely.
 
+In delta mode, a run selecting an older checkpoint reports an error before rewriting
+state if either the selected checkpoint or the current head has pending tool
+execution. For `Command(resume=...)`, other pending graph nodes also prevent the
+rewrite, except for the synthetic input entry node. Resume pending work from the
+latest checkpoint; to regenerate completed work, submit new input there. Full-mode
+resumes and ordinary replay without pending execution keep their existing behavior.
+
 The optional `database.checkpoint_cache` section (delta channel mode only)
 caches materialized checkpoint histories: `type` is `memory` (default) or
 `redis`, and `max_entries: 0` disables the cache. The `redis` backend is

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -40,6 +40,10 @@ with real ASGI requests and LangGraph SDK decoding.
 
 ### Checkpoint Channel Modes (`full` / `delta`)
 
+Reject missing selected/head tuples or pending Sends before delta rewrites.
+`Command(resume=...)` also rejects pending nodes except `__start__` on both.
+Use `CheckpointResumeRejected` to skip pre-write rollback; resume at head.
+
 Checkpointer storage runs in one of two channel modes, selected by `checkpoint_channel_mode` in `config.yaml` (default `full`). `delta` mode adopts LangGraph 1.2's `DeltaChannel` for `messages`: checkpoints store a sentinel + per-step writes instead of the full message list, so storage/serde grows O(N) instead of O(N²) in turns. All checkpointer backends (memory/sqlite/postgres) serve both modes unchanged — the semantics live in the compiled graph's channel table, not in the saver.
 
 **Mode is process-frozen and restart-required.** `make_lead_agent` and the embedded `DeerFlowClient` freeze the resolved mode (`runtime/checkpoint_mode.py::freeze_checkpoint_channel_mode`) before compiling the graph with the mode-matched schema (`agents/thread_state.py::get_thread_state_schema`, plus `adapt_state_schema_for_mode` / `normalize_middleware_state_schemas` for middleware state). Adapted middleware schemas are cached by schema, mode, and resolved snapshot frequency so a pre-freeze ephemeral graph cannot leave a stale default-frequency schema behind. A second, different mode or frequency in the same process raises `CheckpointModeReconfigurationError`. To switch: edit config, restart.

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -36,7 +36,7 @@ from typing import Any, Final, Literal, cast
 from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY
 from langchain_core.messages import AIMessage, AIMessageChunk
 from langgraph.checkpoint.base import empty_checkpoint
-from langgraph.types import Overwrite
+from langgraph.types import Command, Overwrite
 
 from deerflow.agents.goal_state import GoalEvaluation, GoalState
 from deerflow.agents.interaction_policy import RunInteractionPolicy, resolve_run_interaction_policy
@@ -1033,6 +1033,7 @@ async def run_agent(
     # checkpoint lineage (see below).
     accessor: CheckpointStateAccessor | None = None
     rollback_point: RollbackPoint | None = None
+    checkpoint_resume_rejected = False
     journal = None
     partial_assistant_response: _PartialAssistantResponseAccumulator | None = None
     seq_stamper: _MessageSeqStamper | None = None
@@ -1430,6 +1431,7 @@ async def run_agent(
                     config=config,
                     thread_id=thread_id,
                     run_id=run_id,
+                    resume_command=_is_resume_command_input(graph_input),
                 )
             if resumed_messages is not None:
                 # The graph now starts from the selected state, so the
@@ -1750,6 +1752,7 @@ async def run_agent(
         await _finish_cancellation(record.abort_action)
 
     except Exception as exc:
+        checkpoint_resume_rejected = isinstance(exc, CheckpointResumeRejected)
         error_msg = f"{exc}"
         logger.exception("Run %s failed: %s", run_id, error_msg)
         await _ensure_stream_producer_lease()
@@ -1761,7 +1764,7 @@ async def run_agent(
             **terminal_status_kwargs,
         )
         if cancel_action is not None:
-            await _finish_cancellation(cancel_action)
+            await _finish_cancellation(cancel_action, restore_checkpoint=not checkpoint_resume_rejected)
         elif not record.ownership_lost:
             error_frame = {
                 "message": error_msg,
@@ -1801,7 +1804,9 @@ async def run_agent(
                         run_id,
                     )
 
-                if not record.ownership_lost and _is_edit_replay_run(record) and record.status != RunStatus.success:
+                # A rejected resume has not written state; compensating it through
+                # the state-only graph would itself discard the pending Sends.
+                if not record.ownership_lost and _is_edit_replay_run(record) and record.status != RunStatus.success and not checkpoint_resume_rejected:
                     if not record.finalizing:
                         await run_manager.set_finalizing(run_id, True)
                     try:
@@ -2928,6 +2933,15 @@ def _complete_state_replacement_values(
     return replacement_values
 
 
+def _is_resume_command_input(graph_input: Any) -> bool:
+    """Identify requests that need to preserve interrupted execution position."""
+    return isinstance(graph_input, Command) and graph_input.resume is not None
+
+
+class CheckpointResumeRejected(RuntimeError):
+    """Checkpoint preparation was rejected before any state write."""
+
+
 async def _linearize_delta_checkpoint_resume(
     *,
     accessor: CheckpointStateAccessor,
@@ -2935,6 +2949,7 @@ async def _linearize_delta_checkpoint_resume(
     config: dict[str, Any],
     thread_id: str,
     run_id: str,
+    resume_command: bool = False,
 ) -> list[Any] | None:
     """Replace a delta-mode checkpoint fork with an equivalent linear write.
 
@@ -2997,6 +3012,22 @@ async def _linearize_delta_checkpoint_resume(
     messages = values.get("messages") if isinstance(values, dict) else None
     if not isinstance(messages, list):
         raise RuntimeError(f"Run {run_id} could not materialize resume checkpoint {checkpoint_id}")
+    # Pending Sends cannot survive the state-only rewrite for any input shape.
+    # Resume commands also need branch-triggered execution positions preserved;
+    # fresh input may regenerate those nodes. Check both the selected state and
+    # the head that would be overwritten. The synthetic input entry is not work.
+    for location, checkpoint_config, inspected_snapshot in (("selected", source_config, snapshot), ("head", head.config, head)):
+        inspected_id = checkpoint_config["configurable"]["checkpoint_id"]
+        checkpoint_tuple = await accessor.aget_tuple(checkpoint_config)
+        if checkpoint_tuple is None:
+            raise CheckpointResumeRejected(f"Run {run_id} could not inspect pending execution for {location} checkpoint {inspected_id} while resuming checkpoint {checkpoint_id}")
+        pending_nodes = tuple(sorted(node for node in (getattr(inspected_snapshot, "next", None) or ()) if node != "__start__")) if resume_command else ()
+        if (checkpoint_tuple.checkpoint.get("channel_values") or {}).get("__pregel_tasks") or pending_nodes:
+            pending_detail = f" for {pending_nodes}" if pending_nodes else ""
+            raise CheckpointResumeRejected(
+                f"Run {run_id} cannot resume an older delta checkpoint {checkpoint_id}: the {location} checkpoint {inspected_id} has pending execution{pending_detail}. "
+                "Resume pending work from the latest checkpoint; to regenerate completed work, submit new input from the latest checkpoint."
+            )
     head_config["metadata"] = checkpoint_agent_binding_metadata(getattr(snapshot, "metadata", None))
 
     # Write through the thread's effective schema so every application and

backend/tests/test_delta_resume_pending_execution.py

@@ -0,0 +1,435 @@
+"""Historical delta resume must not erase a paused tool task (#7438)."""
+
+from __future__ import annotations
+
+import copy
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langchain_core.tools import tool
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+from langgraph.graph import END, StateGraph
+from langgraph.types import Command
+
+from deerflow.agents.thread_state import get_thread_state_schema
+from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+from deerflow.runtime.checkpoint_mode import inject_checkpoint_mode
+from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
+from deerflow.runtime.runs.manager import RunManager
+from deerflow.runtime.runs.schemas import RunStatus
+from deerflow.runtime.runs.worker import RunContext, _linearize_delta_checkpoint_resume, run_agent
+
+pytestmark = pytest.mark.anyio
+THREAD_ID = "pending-delta-resume"
+FINAL_ANSWER = "Probe completed."
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.fixture(autouse=True)
+def offline_config(monkeypatch, tmp_path):
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}}))
+    yield
+    reset_app_config()
+
+
+@pytest.fixture(params=["memory", "sqlite"])
+async def checkpointer(request, tmp_path):
+    if request.param == "memory":
+        yield InMemorySaver()
+    else:
+        async with AsyncSqliteSaver.from_conn_string(str(tmp_path / "checkpoints.sqlite")) as saver:
+            await saver.setup()
+            yield saver
+
+
+class _ProbeModel(FakeMessagesListChatModel):
+    def bind_tools(self, tools: Any, **kwargs: Any):
+        return self
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        if any(isinstance(message, ToolMessage) for message in messages):
+            answer = AIMessage(content=FINAL_ANSWER, id="final")
+        elif sum(isinstance(message, HumanMessage) for message in messages) >= 2:
+            answer = AIMessage(content="", id="call", tool_calls=[{"name": "probe", "args": {}, "id": "probe-call", "type": "tool_call"}])
+        else:
+            answer = AIMessage(content="First answer.", id="first-answer")
+        return ChatResult(generations=[ChatGeneration(message=answer)])
+
+
+def _config(mode, checkpoint_id=None):
+    config = {"configurable": {"thread_id": THREAD_ID, "checkpoint_ns": ""}}
+    if checkpoint_id:
+        config["configurable"]["checkpoint_id"] = checkpoint_id
+    inject_checkpoint_mode(config, mode)
+    return config
+
+
+async def _worker(graph, saver, manager, mode, config, graph_input, *, metadata=None, **kwargs):
+    record = await manager.create(THREAD_ID, metadata=metadata)
+    bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock())
+    await run_agent(
+        bridge,
+        manager,
+        record,
+        ctx=RunContext(checkpointer=saver, checkpoint_channel_mode=mode),
+        agent_factory=lambda **_kwargs: graph,
+        graph_input=graph_input,
+        config=config,
+        stream_modes=["values"],
+        **kwargs,
+    )
+    return record
+
+
+async def _paused_worker(saver, mode, *, branch_tasks=False):
+    calls = []
+
+    @tool
+    def probe() -> str:
+        """Record a harmless probe execution."""
+        calls.append("probe")
+        return "Probe result."
+
+    model = _ProbeModel(responses=[])
+    if branch_tasks:
+
+        async def model_node(state):
+            message = await model.ainvoke(state["messages"])
+            if message.tool_calls:
+                human = next(message for message in reversed(state["messages"]) if isinstance(message, HumanMessage))
+                message = message.model_copy(update={"id": f"call-{human.id}"})
+            return {"messages": [message]}
+
+        async def tools_node(state):
+            calls.append("probe")
+            return {"messages": [ToolMessage(content="Probe result.", tool_call_id="probe-call", id="probe-result")]}
+
+        builder = StateGraph(get_thread_state_schema(mode))
+        builder.add_node("model", model_node)
+        builder.add_node("tools", tools_node)
+        builder.set_entry_point("model")
+        builder.add_conditional_edges("model", lambda state: "tools" if state["messages"][-1].tool_calls else END)
+        builder.add_edge("tools", "model")
+        graph = builder.compile(checkpointer=saver)
+    else:
+        graph = create_agent(model, tools=[probe], state_schema=get_thread_state_schema(mode), checkpointer=saver)
+    config = _config(mode)
+    await graph.ainvoke({"messages": [HumanMessage(content="hello", id="h1")]}, config)
+    manager = RunManager()
+    record = await _worker(graph, saver, manager, mode, config, {"messages": [HumanMessage(content="please probe", id="h2")]}, interrupt_before=["tools"])
+    assert record.status == RunStatus.success
+    graph.interrupt_before_nodes = None
+
+    # Normal Worker finalization makes the paused checkpoint historical.
+    head_tuple = await saver.aget_tuple(config)
+    assert "runtime_run_duration" in head_tuple.metadata["writes"]
+    selected = await graph.aget_state(head_tuple.parent_config)
+    head = await graph.aget_state(config)
+    assert selected.config["configurable"]["checkpoint_id"] != head.config["configurable"]["checkpoint_id"]
+    assert selected.next == head.next == ("tools",)
+    assert calls == []
+    return graph, manager, calls, selected, head
+
+
+async def _earlier_checkpoint_without_sends(graph, saver, selector):
+    history = [snapshot async for snapshot in graph.aget_state_history(_config("delta"))]
+    selected = next(
+        snapshot
+        for snapshot in history
+        if any(message.id == "first-answer" for message in snapshot.values.get("messages", []))
+        and not any(message.id == "h2" for message in snapshot.values.get("messages", []))
+        and (snapshot.metadata.get("source") == "input" if selector == "input" else snapshot.next == ())
+    )
+    selected_tuple = await saver.aget_tuple(selected.config)
+    assert not (selected_tuple.checkpoint.get("channel_values") or {}).get("__pregel_tasks")
+    return await graph.aget_state(selected.config)
+
+
+@pytest.mark.parametrize("selector", ["input", "completed"])
+async def test_pending_head_rejects_earlier_selector_without_rewriting(checkpointer, selector):
+    graph, _, calls, _, head = await _paused_worker(checkpointer, "delta")
+    selected = await _earlier_checkpoint_without_sends(graph, checkpointer, selector)
+    head_tuple = await checkpointer.aget_tuple(head.config)
+    assert head_tuple.checkpoint["channel_values"]["__pregel_tasks"]
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+    before_config = copy.deepcopy(config)
+    before_checkpoints = [item.config async for item in checkpointer.alist(_config("delta"))]
+    accessor = CheckpointStateAccessor.bind(graph, checkpointer, mode="delta")
+
+    with pytest.raises(RuntimeError, match="pending execution.*latest checkpoint"):
+        await _linearize_delta_checkpoint_resume(accessor=accessor, checkpointer=checkpointer, config=config, thread_id=THREAD_ID, run_id="pending-head")
+
+    assert config == before_config
+    assert await graph.aget_state(_config("delta")) == head
+    assert await graph.aget_state(selected.config) == selected
+    assert [item.config async for item in checkpointer.alist(_config("delta"))] == before_checkpoints
+    result = await graph.ainvoke(Command(resume="continue"), _config("delta"))
+    assert calls == ["probe"]
+    assert result["messages"][-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("selector", ["input", "completed"])
+@pytest.mark.parametrize("input_kind", ["resume", "fresh"])
+@pytest.mark.parametrize("edit_replay", [False, True])
+async def test_worker_preserves_pending_head_with_earlier_selector(checkpointer, selector, input_kind, edit_replay):
+    graph, manager, calls, _, head = await _paused_worker(checkpointer, "delta")
+    selected = await _earlier_checkpoint_without_sends(graph, checkpointer, selector)
+    checkpoint_id = selected.config["configurable"]["checkpoint_id"]
+    graph_input = Command(resume="continue") if input_kind == "resume" else {"messages": [HumanMessage(content="new request", id="h3")]}
+
+    record = await _worker(graph, checkpointer, manager, "delta", _config("delta", checkpoint_id), graph_input, metadata={"replay_kind": "edit"} if edit_replay else None)
+
+    assert record.status == RunStatus.error
+    assert "pending execution" in record.error
+    assert record.run_id in record.error
+    assert checkpoint_id in record.error
+    assert head.config["configurable"]["checkpoint_id"] in record.error
+    assert calls == []
+    after = await graph.aget_state(_config("delta"))
+    assert after.values == head.values
+    assert after.next == head.next == ("tools",)
+    record = await _worker(graph, checkpointer, manager, "delta", _config("delta"), Command(resume="continue"))
+    assert record.status == RunStatus.success
+    assert calls == ["probe"]
+    messages = (await graph.aget_state(_config("delta"))).values["messages"]
+    assert any(isinstance(message, ToolMessage) and message.tool_call_id == "probe-call" for message in messages)
+    assert messages[-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("edit_replay", [False, True])
+async def test_rejected_resume_skips_rollback_when_cancellation_wins(checkpointer, monkeypatch, edit_replay):
+    graph, manager, calls, _, head = await _paused_worker(checkpointer, "delta")
+    selected = await _earlier_checkpoint_without_sends(graph, checkpointer, "completed")
+    original_set_status = manager.set_status_if_not_cancelled
+
+    async def cancellation_wins(run_id, status, **kwargs):
+        # Model the terminal manager decision after a durable rollback request
+        # wins admission; retain the real Worker, graph and checkpoint writes.
+        if status == RunStatus.error and "pending execution" in (kwargs.get("error") or ""):
+            return "rollback"
+        return await original_set_status(run_id, status, **kwargs)
+
+    monkeypatch.setattr(manager, "set_status_if_not_cancelled", cancellation_wins)
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+    record = await _worker(graph, checkpointer, manager, "delta", config, Command(resume="continue"), metadata={"replay_kind": "edit"} if edit_replay else None)
+
+    assert record.status == RunStatus.error
+    assert record.error == "Rolled back by user"
+    assert calls == []
+    after = await graph.aget_state(_config("delta"))
+    assert after.values == head.values
+    assert after.next == ("tools",)
+    record = await _worker(graph, checkpointer, manager, "delta", _config("delta"), Command(resume="continue"))
+    assert record.status == RunStatus.success
+    assert calls == ["probe"]
+    assert (await graph.aget_state(_config("delta"))).values["messages"][-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("head_completed", [False, True])
+async def test_historical_delta_resume_rejects_before_mutating_pending_work(checkpointer, head_completed):
+    graph, _, calls, selected, head = await _paused_worker(checkpointer, "delta")
+    if head_completed:
+        await graph.ainvoke(Command(resume="continue"), _config("delta"))
+        head = await graph.aget_state(_config("delta"))
+        selected = await graph.aget_state(selected.config)
+        assert head.next == ()
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+    before_config = copy.deepcopy(config)
+    before_checkpoints = [item.config async for item in checkpointer.alist(_config("delta"))]
+    accessor = CheckpointStateAccessor.bind(graph, checkpointer, mode="delta")
+
+    with pytest.raises(RuntimeError, match="pending execution.*latest checkpoint") as error:
+        await _linearize_delta_checkpoint_resume(accessor=accessor, checkpointer=checkpointer, config=config, thread_id=THREAD_ID, run_id="rejected-resume")
+
+    assert "rejected-resume" in str(error.value)
+    assert selected.config["configurable"]["checkpoint_id"] in str(error.value)
+    assert "new input from the latest checkpoint" in str(error.value)
+    assert config == before_config
+    assert await graph.aget_state(_config("delta")) == head
+    assert await graph.aget_state(selected.config) == selected
+    assert [item.config async for item in checkpointer.alist(_config("delta"))] == before_checkpoints
+    if not head_completed:
+        result = await graph.ainvoke(Command(resume="continue"), _config("delta"))
+        assert result["messages"][-1].content == FINAL_ANSWER
+    assert calls == ["probe"]
+
+
+@pytest.mark.parametrize("missing_checkpoint", ["selected", "head"])
+async def test_unavailable_checkpoint_rejects_without_rewriting(checkpointer, monkeypatch, missing_checkpoint):
+    graph, _, calls, _, head = await _paused_worker(checkpointer, "delta")
+    selected = await _earlier_checkpoint_without_sends(graph, checkpointer, "completed")
+    accessor = CheckpointStateAccessor.bind(graph, checkpointer, mode="delta")
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+    before_config = copy.deepcopy(config)
+    before_checkpoints = [item.config async for item in checkpointer.alist(_config("delta"))]
+    missing_id = (selected if missing_checkpoint == "selected" else head).config["configurable"]["checkpoint_id"]
+    original_get_tuple = accessor.aget_tuple
+
+    async def get_tuple(config):
+        if config["configurable"].get("checkpoint_id") == missing_id:
+            return None
+        return await original_get_tuple(config)
+
+    missing = AsyncMock(side_effect=get_tuple)
+    monkeypatch.setattr(accessor, "aget_tuple", missing)
+
+    with pytest.raises(RuntimeError, match="could not inspect pending execution"):
+        await _linearize_delta_checkpoint_resume(accessor=accessor, checkpointer=checkpointer, config=config, thread_id=THREAD_ID, run_id="missing-checkpoint")
+
+    assert missing.await_args.args[0]["configurable"]["checkpoint_id"] == missing_id
+    assert config == before_config
+    assert await graph.aget_state(_config("delta")) == head
+    assert [item.config async for item in checkpointer.alist(_config("delta"))] == before_checkpoints
+    result = await graph.ainvoke(Command(resume="continue"), _config("delta"))
+    assert calls == ["probe"]
+    assert result["messages"][-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("mode", ["full", "delta"])
+@pytest.mark.parametrize("selector", ["head", "latest-id", "paused-id"])
+@pytest.mark.parametrize("edit_replay", [False, True])
+async def test_worker_does_not_report_success_after_discarding_pending_tools(checkpointer, mode, selector, edit_replay):
+    graph, manager, calls, selected, head = await _paused_worker(checkpointer, mode)
+    checkpoint_id = None if selector == "head" else (head if selector == "latest-id" else selected).config["configurable"]["checkpoint_id"]
+    record = await _worker(graph, checkpointer, manager, mode, _config(mode, checkpoint_id), Command(resume="continue"), metadata={"replay_kind": "edit"} if edit_replay else None)
+
+    if mode == "delta" and selector == "paused-id":
+        assert record.status == RunStatus.error
+        assert "pending execution" in record.error
+        assert "latest checkpoint" in record.error
+        assert calls == []
+        after = await graph.aget_state(_config(mode))
+        assert after.values == head.values
+        assert after.next == ("tools",)
+        # The rejected resume's normal finalization may write duration metadata,
+        # but the latest execution position remains usable without the selector.
+        record = await _worker(graph, checkpointer, manager, mode, _config(mode), Command(resume="continue"))
+
+    assert record.status == RunStatus.success
+    assert calls == ["probe"]
+    messages = (await graph.aget_state(_config(mode))).values["messages"]
+    assert any(isinstance(message, ToolMessage) and message.tool_call_id == "probe-call" for message in messages)
+    assert messages[-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("selector", ["paused", "input", "completed"])
+@pytest.mark.parametrize("cleanup", ["ordinary", "edit", "rollback", "edit-rollback"])
+async def test_worker_preserves_non_send_pending_nodes(checkpointer, monkeypatch, selector, cleanup):
+    graph, manager, calls, paused, head = await _paused_worker(checkpointer, "delta", branch_tasks=True)
+    selected = paused if selector == "paused" else await _earlier_checkpoint_without_sends(graph, checkpointer, selector)
+    for snapshot in (selected, head):
+        checkpoint_tuple = await checkpointer.aget_tuple(snapshot.config)
+        assert not (checkpoint_tuple.checkpoint.get("channel_values") or {}).get("__pregel_tasks")
+    assert selected.next == (("tools",) if selector == "paused" else ("__start__",) if selector == "input" else ())
+    checkpoint_id = selected.config["configurable"]["checkpoint_id"]
+    config = _config("delta", checkpoint_id)
+    original_set_status = manager.set_status_if_not_cancelled
+    if "rollback" in cleanup:
+
+        async def cancellation_wins(run_id, status, **kwargs):
+            # Inject the terminal decision, preserving the actual Worker cleanup.
+            if status == RunStatus.error and "pending execution" in (kwargs.get("error") or ""):
+                return "rollback"
+            return await original_set_status(run_id, status, **kwargs)
+
+        monkeypatch.setattr(manager, "set_status_if_not_cancelled", cancellation_wins)
+
+    record = await _worker(graph, checkpointer, manager, "delta", config, Command(resume="continue"), metadata={"replay_kind": "edit"} if "edit" in cleanup else None)
+
+    assert record.status == RunStatus.error
+    if "rollback" in cleanup:
+        assert record.error == "Rolled back by user"
+    else:
+        location = "selected" if selector == "paused" else "head"
+        assert f"the {location} checkpoint" in record.error
+        assert "pending execution" in record.error
+        assert record.run_id in record.error
+        assert checkpoint_id in record.error
+        assert (selected if location == "selected" else head).config["configurable"]["checkpoint_id"] in record.error
+    assert config["configurable"]["checkpoint_id"] == checkpoint_id
+    assert calls == []
+    after = await graph.aget_state(_config("delta"))
+    assert after.values == head.values
+    assert after.next == ("tools",)
+
+    monkeypatch.setattr(manager, "set_status_if_not_cancelled", original_set_status)
+    recovered = await _worker(graph, checkpointer, manager, "delta", _config("delta"), Command(resume="continue"))
+    assert recovered.status == RunStatus.success
+    assert calls == ["probe"]
+    messages = (await graph.aget_state(_config("delta"))).values["messages"]
+    assert any(isinstance(message, ToolMessage) and message.tool_call_id == "probe-call" for message in messages)
+    assert messages[-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("selector", ["paused", "input", "completed"])
+async def test_non_send_fresh_input_replay_still_regenerates(checkpointer, selector):
+    graph, manager, calls, paused, head = await _paused_worker(checkpointer, "delta", branch_tasks=True)
+    selected = paused if selector == "paused" else await _earlier_checkpoint_without_sends(graph, checkpointer, selector)
+    for snapshot in (selected, head):
+        checkpoint_tuple = await checkpointer.aget_tuple(snapshot.config)
+        assert not (checkpoint_tuple.checkpoint.get("channel_values") or {}).get("__pregel_tasks")
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+
+    record = await _worker(graph, checkpointer, manager, "delta", config, {"messages": [HumanMessage(content="new request", id="h3")]})
+
+    assert record.status == RunStatus.success
+    assert "checkpoint_id" not in config["configurable"]
+    assert calls == ["probe"]
+    messages = (await graph.aget_state(_config("delta"))).values["messages"]
+    assert any(message.id == "h3" for message in messages)
+    assert messages[-1].content == FINAL_ANSWER
+
+
+@pytest.mark.parametrize("selector", ["input", "completed"])
+async def test_non_send_resume_without_pending_nodes_still_linearizes(checkpointer, selector):
+    graph, manager, calls, _, _ = await _paused_worker(checkpointer, "delta", branch_tasks=True)
+    await graph.ainvoke(Command(resume="continue"), _config("delta"))
+    head = await graph.aget_state(_config("delta"))
+    assert head.next == ()
+    selected = await _earlier_checkpoint_without_sends(graph, checkpointer, selector)
+    assert selected.next == (("__start__",) if selector == "input" else ())
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+
+    record = await _worker(graph, checkpointer, manager, "delta", config, Command(resume="continue"))
+
+    assert record.status == RunStatus.success
+    assert "checkpoint_id" not in config["configurable"]
+    assert calls == ["probe"]
+    after = await graph.aget_state(_config("delta"))
+    assert [message.id for message in after.values["messages"]] == [message.id for message in selected.values["messages"]]
+
+
+@pytest.mark.parametrize("edit_replay", [False, True])
+async def test_non_send_selected_nodes_reject_after_head_completed(checkpointer, edit_replay):
+    graph, manager, calls, selected, _ = await _paused_worker(checkpointer, "delta", branch_tasks=True)
+    await graph.ainvoke(Command(resume="continue"), _config("delta"))
+    head = await graph.aget_state(_config("delta"))
+    assert head.next == ()
+    selected = await graph.aget_state(selected.config)
+    assert selected.next == ("tools",)
+    selected_tuple = await checkpointer.aget_tuple(selected.config)
+    assert not (selected_tuple.checkpoint.get("channel_values") or {}).get("__pregel_tasks")
+    config = _config("delta", selected.config["configurable"]["checkpoint_id"])
+
+    record = await _worker(graph, checkpointer, manager, "delta", config, Command(resume="continue"), metadata={"replay_kind": "edit"} if edit_replay else None)
+
+    assert record.status == RunStatus.error
+    assert "the selected checkpoint" in record.error
+    assert "pending execution" in record.error
+    assert config["configurable"]["checkpoint_id"] == selected.config["configurable"]["checkpoint_id"]
+    assert calls == ["probe"]
+    after = await graph.aget_state(_config("delta"))
+    assert after.values == head.values
+    assert after.next == ()