fix(gateway): preserve null input for static continuation (#7578)

* fix(gateway): preserve null input for static continuation

* fix(runtime): preserve continuation inputs during MCP projection

---------

Co-authored-by: haotianqi666 <250384991+haotianqi666@users.noreply.github.com>

README.md

@@ -718,6 +718,18 @@ version requires configuration changes, run `make config-upgrade` before restart
 See [Operations and Troubleshooting](frontend/src/content/en/application/operations-and-troubleshooting.mdx#upgrading-an-existing-checkout)
 for the commands for each mode.
 
+To continue a thread paused at a static `interrupt_before` or `interrupt_after`
+breakpoint, submit `input: null` (or omit `input`) to its `/runs`, `/runs/wait`,
+or `/runs/stream` endpoint. This continues pending graph work with the source
+turn's knowledge scope from the selected checkpoint (the head by default).
+An explicit empty object is new input, not a continuation; dynamic interrupts
+still require `command.resume`. In delta mode, historical continuation rejects
+rewrites that would discard pending execution; continue at the head instead.
+Continuations retain the checkpoint's MCP `background_tasks` projection; only
+dictionary input refreshes it from the task repository.
+Old null-input runs stored as `{}` cannot safely match a keyed null-input retry
+after upgrading and return 409. Check the original run before submitting a new key.
+
 When rolling out the resume-command idempotency fix across multiple Gateway
 workers or Pods, route **all keyed resume submissions and retries** (`command.resume`
 with `Idempotency-Key`, on thread-scoped `/runs`, `/runs/stream`, or `/runs/wait`)

backend/app/gateway/AGENTS.md

@@ -140,7 +140,7 @@ owner-scoped assistant version selection remains enabled.
 | **GitHub Event-Driven Agents** | Custom agents can declare a `github:` block in their `config.yaml` to bind to repos and event triggers. Webhook fan-out publishes one `InboundMessage` per matching binding to the channel bus; `GitHubChannel` routes those messages through `ChannelManager`. The response `dispatch` summarizes matched/fired/skipped agents. |
 
 Edit-replay and messages-input clarification may replace checkpoint knowledge
-scope with a validated client snapshot; regenerate and `Command` resume always
+scope with a validated client snapshot; regenerate, null-input and `Command` resume
 recover checkpoint scope from the originating genuine user turn, skipping
 server-generated hidden context, and ignore client values. Scope admission
 treats `input.messages: null` as an empty message collection so

backend/app/gateway/services.py

@@ -1971,11 +1971,19 @@ async def _recover_run_knowledge_scope(
     *,
     thread_id: str,
     target_message_id: str | None,
+    checkpoint_id: str | None = None,
+    checkpoint_ns: str = "",
 ) -> object | None:
-    """Resolve one replay/resume scope from the authoritative latest checkpoint."""
+    """Recover the source turn's scope from the selected checkpoint or head."""
+    checkpoint_kwargs: dict[str, Any] = {}
+    if checkpoint_id is not None:
+        checkpoint_kwargs["checkpoint_id"] = checkpoint_id
+    if checkpoint_ns:
+        checkpoint_kwargs["checkpoint_ns"] = checkpoint_ns
     accessor, config = await build_thread_checkpoint_state_accessor(
         request,
         thread_id=thread_id,
+        **checkpoint_kwargs,
     )
     try:
         snapshot = await accessor.aget(config)
@@ -2252,7 +2260,8 @@ async def start_run(
         if command and command.get("resume") is not None:
             graph_input = Command(resume=command["resume"])
         else:
-            graph_input = normalized_input
+            # LangGraph continues pending work with None; {} starts new input.
+            graph_input = normalized_input if body.input is not None else None
         # Keep the idempotency identity in the caller's pre-stamp shape. A
         # keyed retry can then reuse a run even though its stored run input has
         # Gateway-issued IDs, and old id-less run records remain comparable.
@@ -2307,7 +2316,7 @@ async def start_run(
         scope_graph_input = graph_input if isinstance(graph_input, dict) else {"messages": []}
         current_human_message = _current_human_message(graph_input)
         current_message_has_scope = current_human_message is not None and KNOWLEDGE_SCOPE_KEY in current_human_message.additional_kwargs
-        replay_requires_scope_recovery = isinstance(graph_input, Command) or (isinstance(target_message_id, str) and bool(target_message_id) and (replay_kind != "edit" or not current_message_has_scope))
+        replay_requires_scope_recovery = graph_input is None or isinstance(graph_input, Command) or (isinstance(target_message_id, str) and bool(target_message_id) and (replay_kind != "edit" or not current_message_has_scope))
         is_human_input_response = current_human_message is not None and "human_input_response" in current_human_message.additional_kwargs
         # Clarification and edit-replay messages may intentionally replace the
         # source scope. If either client omits its current selector snapshot,
@@ -2315,11 +2324,16 @@ async def start_run(
         # run to every operator-approved dataset. Other replay paths always use
         # server recovery regardless of client input.
         is_scope_recovery = replay_requires_scope_recovery or (is_human_input_response and not current_message_has_scope)
+        recovery_checkpoint: dict[str, Any] = {}
+        if graph_input is None:
+            configurable = config.get("configurable") or {}
+            recovery_checkpoint = {key: configurable[key] for key in ("checkpoint_id", "checkpoint_ns") if configurable.get(key)}
         recovery_scope = (
             await _recover_run_knowledge_scope(
                 request,
                 thread_id=thread_id,
                 target_message_id=(target_message_id if isinstance(target_message_id, str) else None),
+                **recovery_checkpoint,
             )
             if is_scope_recovery
             else None

backend/docs/runtime-guidance-details.md

@@ -295,7 +295,7 @@ PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py
 ## Checkpoint channel modes and replay
 
 Reject missing selected/head tuples or pending Sends before delta rewrites.
-`Command(resume=...)` also rejects pending nodes except `__start__` on both.
+Null input and `Command(resume=...)` also reject 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.

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

@@ -59,8 +59,10 @@ Use `CheckpointStateAccessor` with the thread's effective schema for state reads
 and mutations; preserve mode gates, parentage and pending writes. State replacement
 uses `Overwrite`. Delta resume/rollback rewrites the current head rather than
 forking; reject missing selected/head tuples and pending Sends before writing.
-`Command(resume=...)` also rejects pending nodes except `__start__` on both;
+Null input and `Command(resume=...)` also reject pending nodes except `__start__` on both;
 `CheckpointResumeRejected` skips pre-write rollback. Resume at head.
+Project MCP tasks only into dict input; `None`/`Command` retain checkpointed tasks
+without querying the repository or replacing the control input.
 
 Before changing checkpoint modes, caches, replay lookup, state replacement or
 rollback, read the [checkpoint contracts](../../../../docs/runtime-guidance-details.md#checkpoint-channel-modes-and-replay).

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

@@ -972,7 +972,7 @@ async def run_agent(
     *,
     ctx: RunContext,
     agent_factory: Any,
-    graph_input: dict,
+    graph_input: dict[str, Any] | Command | None,
     config: dict,
     thread_incarnation: str | None | object = _THREAD_INCARNATION_UNSET,
     stream_modes: list[str] | None = None,
@@ -1157,7 +1157,8 @@ async def run_agent(
 
         # Keep cancellable preflight work under the worker's terminal guard so
         # cancellation cannot strand a pending RunRecord or stream subscriber.
-        if ctx.mcp_task_repo is not None and record.user_id is not None:
+        # Continuations keep their checkpointed projection and control input.
+        if isinstance(graph_input, dict) and ctx.mcp_task_repo is not None and record.user_id is not None:
             try:
                 if thread_incarnation is _THREAD_INCARNATION_UNSET:
                     raise RuntimeError("MCP task projection requires a server-owned thread incarnation")
@@ -1431,7 +1432,7 @@ async def run_agent(
                     config=config,
                     thread_id=thread_id,
                     run_id=run_id,
-                    resume_command=_is_resume_command_input(graph_input),
+                    resume_command=graph_input is None or _is_resume_command_input(graph_input),
                 )
             if resumed_messages is not None:
                 # The graph now starts from the selected state, so the

backend/tests/test_conversation_access.py

@@ -226,6 +226,11 @@ def test_start_run_installs_fresh_capability_without_persisting_it(monkeypatch):
         empty = await services.start_run(RunCreateRequest(input={"messages": None}, conversation_references=["source"]), "empty-input", request)
         await empty.task
         assert callable(captured[3]["ctx"].conversation_reader)
+        static_resume = await services.start_run(RunCreateRequest(input=None, conversation_references=["source"]), "static-resume", request)
+        await static_resume.task
+        assert callable(captured[4]["ctx"].conversation_reader)
+        assert captured[4]["graph_input"] is None
+        assert static_resume.kwargs["input"] is None
         with pytest.raises(HTTPException) as invalid:
             await services.start_run(RunCreateRequest(input={"messages": "bad input"}, conversation_references=["source"]), "invalid-input", request)
         assert invalid.value.status_code == 422

backend/tests/test_delta_resume_pending_execution.py

@@ -326,7 +326,8 @@ async def test_worker_does_not_report_success_after_discarding_pending_tools(che
 
 @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):
+@pytest.mark.parametrize("resume_input", [Command(resume="continue"), None], ids=["command", "static"])
+async def test_worker_preserves_non_send_pending_nodes(checkpointer, monkeypatch, selector, cleanup, resume_input):
     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):
@@ -346,7 +347,7 @@ async def test_worker_preserves_non_send_pending_nodes(checkpointer, monkeypatch
 
         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)
+    record = await _worker(graph, checkpointer, manager, "delta", config, resume_input, metadata={"replay_kind": "edit"} if "edit" in cleanup else None)
 
     assert record.status == RunStatus.error
     if "rollback" in cleanup:
@@ -365,7 +366,7 @@ async def test_worker_preserves_non_send_pending_nodes(checkpointer, monkeypatch
     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"))
+    recovered = await _worker(graph, checkpointer, manager, "delta", _config("delta"), resume_input)
     assert recovered.status == RunStatus.success
     assert calls == ["probe"]
     messages = (await graph.aget_state(_config("delta"))).values["messages"]

backend/tests/test_gateway_services.py

@@ -3013,6 +3013,81 @@ async def test_recover_knowledge_scope_skips_hidden_conversation_reference_messa
     assert recovered == source_scope
 
 
+@pytest.mark.parametrize("input_fields", [{}, {"input": None}, {"input": {}}], ids=["omitted", "null", "empty-object"])
+@pytest.mark.parametrize("source_scope", [None, {"version": 1, "mode": "disabled"}, {"version": 1, "mode": "selected", "dataset_ids": ["source"]}])
+@pytest.mark.parametrize("checkpoint_fields", [{}, {"checkpoint_id": "selected"}, {"checkpoint_id": "selected", "checkpoint_ns": "child"}])
+@pytest.mark.asyncio
+async def test_static_resume_preserves_input_and_recovers_source_scope(_stub_app_config, input_fields, source_scope, checkpoint_fields):
+    from unittest.mock import AsyncMock, patch
+
+    from app.gateway.routers.thread_runs import RunCreateRequest
+    from app.gateway.services import start_run
+    from deerflow.runtime import RunManager
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    set_app_config(
+        AppConfig.model_validate(
+            {
+                "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
+                "knowledge_base": {"enabled": True, "scope_selection_enabled": True},
+                "tools": [{"name": "knowledge_search", "group": "knowledge", "use": "deerflow.community.ragflow.tools:knowledge_search_tool"}],
+            }
+        )
+    )
+    request = _make_start_run_request(RunManager(store=MemoryRunStore()))
+    recover_scope = AsyncMock(return_value=source_scope)
+    captured = {}
+
+    async def fake_run_agent(*_args, **kwargs):
+        captured.update(kwargs)
+
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
+        patch("app.gateway.services._recover_run_knowledge_scope", new=recover_scope),
+        patch("app.gateway.services._load_scope_agent_config", new=AsyncMock(return_value=SimpleNamespace(tool_groups=["knowledge"], knowledge_scope={"version": 1, "mode": "selected", "dataset_ids": ["new-default"]}))),
+    ):
+        record = await start_run(RunCreateRequest(**input_fields, config={"configurable": checkpoint_fields}), "thread-static-scope", request)
+        await record.task
+
+    if input_fields.get("input") is None:
+        assert captured["graph_input"] is None
+        assert captured["knowledge_scope"] == source_scope
+        recover_scope.assert_awaited_once_with(request, thread_id="thread-static-scope", target_message_id=None, **checkpoint_fields)
+    else:
+        assert captured["graph_input"] == {}
+        assert captured["knowledge_scope"] is None
+        recover_scope.assert_not_awaited()
+
+
+@pytest.mark.parametrize("mode", ["full", "delta"])
+@pytest.mark.asyncio
+async def test_static_resume_recovers_selected_checkpoint_scope(_stub_app_config, mode):
+    from unittest.mock import patch
+
+    from langchain_core.messages import HumanMessage
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    from app.gateway.services import _recover_run_knowledge_scope
+    from deerflow.runtime.checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph
+
+    accessor = CheckpointStateAccessor.bind(build_state_mutation_graph("scope", mode), InMemorySaver(), mode=mode)
+    config = {"configurable": {"thread_id": "scope-history", "checkpoint_ns": ""}}
+    source_scope = {"version": 1, "mode": "disabled"}
+    selected = await accessor.aupdate(config, {"messages": [HumanMessage(id="first", content="old turn", additional_kwargs={"knowledge_scope": source_scope})]}, as_node="scope")
+    await accessor.aupdate(config, {"messages": [HumanMessage(id="second", content="new turn", additional_kwargs={"knowledge_scope": {"version": 1, "mode": "selected", "dataset_ids": ["new"]}})]}, as_node="scope")
+
+    async def build_accessor(_request, *, thread_id, checkpoint_id=None):
+        read_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+        if checkpoint_id is not None:
+            read_config["configurable"]["checkpoint_id"] = checkpoint_id
+        return accessor, read_config
+
+    with patch("app.gateway.services.build_thread_checkpoint_state_accessor", side_effect=build_accessor):
+        scope = await _recover_run_knowledge_scope(SimpleNamespace(), thread_id="scope-history", target_message_id=None, checkpoint_id=selected["configurable"]["checkpoint_id"])
+    assert scope == source_scope
+
+
 @pytest.mark.parametrize(
     ("include_current_scope", "expected_scope", "recovery_calls"),
     [

backend/tests/test_run_worker_mcp_tasks.py

@@ -1,6 +1,82 @@
 """MCP task projection tests for the run worker."""
 
-from deerflow.runtime.runs.worker import _project_background_tasks
+from copy import deepcopy
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from langgraph.types import Command
+
+from deerflow.runtime.runs.manager import RunManager
+from deerflow.runtime.runs.schemas import RunStatus
+from deerflow.runtime.runs.worker import RunContext, _project_background_tasks, run_agent
+
+
+async def _run_with_task_repository(graph_input, task_repo):
+    manager = RunManager()
+    record = await manager.create("thread-mcp-projection", user_id="alice")
+    received = []
+
+    class RecordingAgent:
+        async def astream(self, graph_input, **_kwargs):
+            received.append(graph_input)
+            yield {"messages": []}
+
+    await run_agent(
+        SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock()),
+        manager,
+        record,
+        ctx=RunContext(checkpointer=None, mcp_task_repo=task_repo),
+        agent_factory=lambda **_kwargs: RecordingAgent(),
+        graph_input=graph_input,
+        config={},
+        thread_incarnation="incarnation-1",
+    )
+    assert record.status == RunStatus.success
+    assert len(received) == 1
+    return received[0]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("graph_input", [None, Command(resume="approved")], ids=["static", "dynamic"])
+@pytest.mark.parametrize("with_repository", [False, True], ids=["no-repository", "repository"])
+async def test_continuation_preserves_control_input_without_task_projection(graph_input, with_repository, caplog):
+    task_repo = SimpleNamespace(list_by_thread=AsyncMock(return_value=[])) if with_repository else None
+
+    received = await _run_with_task_repository(graph_input, task_repo)
+
+    assert received is graph_input
+    assert "failed to project MCP task state" not in caplog.text
+    if task_repo is not None:
+        task_repo.list_by_thread.assert_not_awaited()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("graph_input", [{}, {"messages": [], "background_tasks": [{"task_id": "stale"}]}], ids=["empty", "existing-projection"])
+async def test_new_input_refreshes_task_projection_without_mutating_input(graph_input, caplog):
+    task_rows = [{"id": "mcp-task-1", "task_name": "Research", "status": "working", "updated_at": "2026-08-15T08:00:00+00:00"}]
+    task_repo = SimpleNamespace(list_by_thread=AsyncMock(return_value=task_rows))
+    original_input = deepcopy(graph_input)
+
+    received = await _run_with_task_repository(graph_input, task_repo)
+
+    assert received == {**original_input, "background_tasks": _project_background_tasks(task_rows)}
+    assert received is not graph_input
+    assert graph_input == original_input
+    task_repo.list_by_thread.assert_awaited_once_with("thread-mcp-projection", user_id="alice", thread_incarnation="incarnation-1", limit=20)
+    assert "failed to project MCP task state" not in caplog.text
+
+
+@pytest.mark.anyio
+async def test_task_projection_failure_preserves_new_input_and_logs_warning(caplog):
+    task_repo = SimpleNamespace(list_by_thread=AsyncMock(side_effect=RuntimeError("task store unavailable")))
+    graph_input = {"messages": []}
+
+    received = await _run_with_task_repository(graph_input, task_repo)
+
+    assert received is graph_input
+    assert "failed to project MCP task state" in caplog.text
+    assert "task store unavailable" in caplog.text
 
 
 def test_project_background_tasks_neutralizes_task_names():

backend/tests/test_runtime_lifecycle_e2e.py

@@ -801,6 +801,101 @@ def test_stream_run_executes_real_lead_agent_setup_agent_business_path(isolated_
         assert not (isolated_deer_flow_home / "users" / "default" / "agents" / agent_name).exists()
 
 
+@pytest.mark.parametrize("checkpoint_mode", ["full", "delta"])
+@pytest.mark.parametrize("interrupt_at", ["interrupt_before", "interrupt_after"])
+@pytest.mark.parametrize("endpoint", ["", "/wait", "/stream"])
+@pytest.mark.parametrize("resume_input", ["null", "sdk-omitted"])
+def test_static_breakpoint_resumes_with_null_input(isolated_app, checkpoint_mode, interrupt_at, endpoint, resume_input, caplog):
+    from starlette.testclient import TestClient
+
+    from deerflow.config.app_config import get_app_config
+
+    get_app_config().database.checkpoint_channel_mode = checkpoint_mode
+    model = FakeToolCallingModel(responses=[AIMessage(content="One model call."), AIMessage(content="Repeated model call.")])
+    with (
+        patch("deerflow.agents.lead_agent.agent.create_chat_model", return_value=model),
+        TestClient(isolated_app) as client,
+    ):
+        assert isolated_app.state.mcp_task_repo is not None
+        csrf_token = _register_user(client)
+        thread_id = _create_thread(client, csrf_token)
+        headers = {"X-CSRF-Token": csrf_token}
+        path = f"/api/threads/{thread_id}/runs"
+        body = _run_body(
+            context={"thinking_enabled": False, "is_plan_mode": False, "subagent_enabled": False},
+            **{interrupt_at: ["model"]},
+        )
+        paused = client.post(f"{path}/wait", json=body, headers=headers)
+        assert paused.status_code == 200, paused.text
+        state = client.get(f"/api/threads/{thread_id}/state").json()
+        assert state["next"], state
+        if interrupt_at == "interrupt_before":
+            assert "model" in state["next"]
+            assert not any(message["type"] == "ai" for message in state["values"]["messages"])
+        else:
+            assert state["values"]["messages"][-1]["content"] == "One model call."
+
+        if resume_input == "null":
+            resumed = client.post(f"{path}{endpoint}", json={**body, "input": None}, headers=headers)
+            assert resumed.status_code == 200, resumed.text
+            if not endpoint:
+                _wait_for_status(client, thread_id, resumed.json()["run_id"], "success")
+            elif endpoint == "/stream":
+                assert not any(event["event"] == "error" for event in _parse_sse(resumed.text)), resumed.text
+        else:
+            import httpx
+            from langgraph_sdk.client import SyncLangGraphClient
+
+            def send_to_gateway(request):
+                assert "input" not in json.loads(request.content)
+                return client.request(request.method, f"/api{request.url.path}", content=request.content, headers=dict(request.headers))
+
+            with httpx.Client(base_url="http://testserver", transport=httpx.MockTransport(send_to_gateway)) as transport:
+                sdk = SyncLangGraphClient(transport)
+                options = {"context": body["context"], interrupt_at: ["model"], "headers": headers, "input": None}
+                if not endpoint:
+                    run = sdk.runs.create(thread_id, "lead_agent", **options)
+                    _wait_for_status(client, thread_id, run["run_id"], "success")
+                elif endpoint == "/wait":
+                    # Gateway does not accept the SDK's optional raise_error field.
+                    result = sdk.runs.wait(thread_id, "lead_agent", raise_error=None, **options)
+                    assert "__error__" not in result, result
+                else:
+                    events = list(sdk.runs.stream(thread_id, "lead_agent", stream_mode=["values"], **options))
+                    assert not any(event.event == "error" for event in events), events
+        state = client.get(f"/api/threads/{thread_id}/state").json()
+        assert state["next"] == [], state
+        replies = [message["content"] for message in state["values"]["messages"] if message["type"] == "ai"]
+        assert replies == ["One model call."]
+        assert "failed to project MCP task state" not in caplog.text
+
+
+@pytest.mark.parametrize("checkpoint_mode", ["full", "delta"])
+def test_null_input_does_not_start_a_new_turn(isolated_app, checkpoint_mode):
+    from starlette.testclient import TestClient
+
+    from deerflow.config.app_config import get_app_config
+
+    get_app_config().database.checkpoint_channel_mode = checkpoint_mode
+    model = FakeToolCallingModel(responses=[AIMessage(content="Original answer."), AIMessage(content="Unexpected repeat.")])
+    with patch("deerflow.agents.lead_agent.agent.create_chat_model", return_value=model), TestClient(isolated_app) as client:
+        headers = {"X-CSRF-Token": _register_user(client)}
+        thread_id = _create_thread(client, headers["X-CSRF-Token"])
+        path = f"/api/threads/{thread_id}/runs/wait"
+        body = _run_body(context={"thinking_enabled": False, "is_plan_mode": False, "subagent_enabled": False})
+        empty = client.post(path, json={**body, "input": None}, headers=headers)
+        assert empty.status_code == 200, empty.text
+        assert "__error__" in empty.json(), empty.text
+        assert not client.get(f"/api/threads/{thread_id}/state").json()["values"].get("messages")
+
+        completed = client.post(path, json=body, headers=headers)
+        assert completed.status_code == 200 and "__error__" not in completed.json(), completed.text
+        resumed = client.post(path, json={**body, "input": None}, headers=headers)
+        assert resumed.status_code == 200 and "__error__" not in resumed.json(), resumed.text
+        messages = client.get(f"/api/threads/{thread_id}/state").json()["values"]["messages"]
+        assert [m["content"] for m in messages if m["type"] == "ai"] == ["Original answer."]
+
+
 def test_cancel_interrupt_stops_running_background_run(isolated_app):
     """HTTP cancel?action=interrupt should stop the worker and persist interruption."""
     from starlette.testclient import TestClient

backend/tests/test_thread_run_idempotency.py

@@ -1001,6 +1001,39 @@ async def test_start_run_reuses_only_matching_resume_fingerprint(_stub_app_confi
     assert "different request" in str(excinfo.value.detail)
 
 
+@pytest.mark.parametrize("original_input", [None, {}], ids=["static-resume", "new-input"])
+@pytest.mark.anyio
+async def test_start_run_idempotency_distinguishes_static_resume_from_new_input(_stub_app_config, original_input):
+    from app.gateway.services import start_run
+
+    # {} also models a legacy null-input record, whose original intent is lost.
+    body = RunCreateRequest(input=original_input)
+    manager = RunManager(store=MemoryRunStore())
+    first = await manager.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="static-resume-key",
+        kwargs={"input": original_input, "config": None},
+    )
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=AssertionError("worker must not attach")),
+        patch("app.gateway.services._recover_run_knowledge_scope", new=AsyncMock(return_value=None)),
+    ):
+        reused = await start_run(body, "thread-1", _make_start_run_request(manager), idempotency_key="static-resume-key")
+        with pytest.raises(HTTPException) as excinfo:
+            await start_run(
+                RunCreateRequest(input={} if original_input is None else None),
+                "thread-1",
+                _make_start_run_request(manager),
+                idempotency_key="static-resume-key",
+            )
+    assert reused.run_id == first.run_id
+    assert reused.idempotency_reused is True
+    assert excinfo.value.status_code == 409
+    assert "different request" in str(excinfo.value.detail)
+
+
 @pytest.mark.anyio
 async def test_start_run_persists_resume_fingerprint_without_payload(_stub_app_config):
     from app.gateway.services import _run_idempotency_request, start_run