diff --git a/README.md b/README.md index ce05d0865b..e5d7143013 100644 --- a/README.md +++ b/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`) diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 681562e1fa..db38b3a13c 100644 --- a/backend/app/gateway/AGENTS.md +++ b/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 diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index c2d6002612..435b500bc3 100644 --- a/backend/app/gateway/services.py +++ b/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 diff --git a/backend/docs/runtime-guidance-details.md b/backend/docs/runtime-guidance-details.md index aee3c238cd..86493105f7 100644 --- a/backend/docs/runtime-guidance-details.md +++ b/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. diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index e566392c38..69e3234971 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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). diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 8fb613b7dc..adecf29549 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/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 diff --git a/backend/tests/test_conversation_access.py b/backend/tests/test_conversation_access.py index 3d69215d30..4ec68e0bec 100644 --- a/backend/tests/test_conversation_access.py +++ b/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 diff --git a/backend/tests/test_delta_resume_pending_execution.py b/backend/tests/test_delta_resume_pending_execution.py index 712170dfaa..36895ff484 100644 --- a/backend/tests/test_delta_resume_pending_execution.py +++ b/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"] diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index fd696b3d04..ea07c9f55d 100644 --- a/backend/tests/test_gateway_services.py +++ b/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"), [ diff --git a/backend/tests/test_run_worker_mcp_tasks.py b/backend/tests/test_run_worker_mcp_tasks.py index 4375fd2287..98268fe09a 100644 --- a/backend/tests/test_run_worker_mcp_tasks.py +++ b/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(): diff --git a/backend/tests/test_runtime_lifecycle_e2e.py b/backend/tests/test_runtime_lifecycle_e2e.py index 8c551cda8d..a89cdc3ef3 100644 --- a/backend/tests/test_runtime_lifecycle_e2e.py +++ b/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 diff --git a/backend/tests/test_thread_run_idempotency.py b/backend/tests/test_thread_run_idempotency.py index fdc460c125..3c97a70a4b 100644 --- a/backend/tests/test_thread_run_idempotency.py +++ b/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