diff --git a/CHANGELOG.md b/CHANGELOG.md index 1342b9339a..af339fd0e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5951,6 +5951,14 @@ This release closes that milestone with **621 merged pull requests**. 80 such requests failed. The same race in IM conversation binding, such as a first message arriving together with `/new`, no longer fails the message. The losing write now updates the row the other one created. ([#6852]) +- **mcp:** Cancelling a background MCP task on SQLite now reaches the remote + server on the next poll cycle instead of up to `mcp_tasks.lease_seconds` + (120 s by default) later. A cancel that committed while the poller was + claiming the task had its cleared poll lease written back, because + `FOR UPDATE` is a no-op on SQLite. The cancellation worker then waited for + that lease to expire, and the task was polled once more for a result that + was thrown away. Each claim now re-checks that the task is still claimable + when it writes the lease. Postgres was not affected. ([#6859]) - **persistence:** Startup schema upgrades no longer log a false `safe_add_column ... drifts from the model definition` warning for columns that already match the model. The check compared the model's generic type @@ -11744,6 +11752,7 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6843]: https://github.com/bytedance/deer-flow/pull/6843 [#6844]: https://github.com/bytedance/deer-flow/pull/6844 [#6852]: https://github.com/bytedance/deer-flow/pull/6852 +[#6859]: https://github.com/bytedance/deer-flow/pull/6859 [#6865]: https://github.com/bytedance/deer-flow/pull/6865 [#6876]: https://github.com/bytedance/deer-flow/pull/6876 [#6897]: https://github.com/bytedance/deer-flow/pull/6897 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index e8591f6537..0df9f8d59e 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -4516,6 +4516,11 @@ `(thread, run, user)` 唯一约束；在 Postgres 上，80 个这样的请求中有 26 个失败。 IM 会话绑定中的同类竞争（例如首条消息与 `/new` 同时到达）也不再导致消息处理失败。 落后的写入现在会更新另一方创建的记录。([#6852]) +- **MCP：** 在 SQLite 上取消后台 MCP 任务时，取消请求现在会在下一个轮询周期发往远程服务器， + 而不是最多延迟 `mcp_tasks.lease_seconds`（默认 120 秒）。由于 SQLite 上 `FOR UPDATE` + 不起作用，在轮询器认领任务期间提交的取消请求所清除的轮询租约会被写回；取消工作器 + 随后要等该租约过期，任务还会被多轮询一次，而结果会被丢弃。现在每次认领在写入租约时 + 都会重新确认任务仍可认领。Postgres 不受影响。([#6859]) - **持久化：** 启动时的 schema 升级不再为已与模型一致的列记录虚假的 `safe_add_column ... drifts from the model definition` 警告。该检查拿模型的通用类型名 和默认值写法（`STRING`、`BIGINTEGER`、`DefaultClause('0')`）去比较数据库自己的写法 @@ -9366,6 +9371,7 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6843]: https://github.com/bytedance/deer-flow/pull/6843 [#6844]: https://github.com/bytedance/deer-flow/pull/6844 [#6852]: https://github.com/bytedance/deer-flow/pull/6852 +[#6859]: https://github.com/bytedance/deer-flow/pull/6859 [#6865]: https://github.com/bytedance/deer-flow/pull/6865 [#6876]: https://github.com/bytedance/deer-flow/pull/6876 [#6897]: https://github.com/bytedance/deer-flow/pull/6897 diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 7eadaef115..aa443b69c6 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -70,7 +70,7 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis **This is defense in depth, not a trust boundary.** `npx`/`uvx` exist to fetch and execute remote packages, so an admin can still point one at a package they published; the boundary is admin authentication plus network reachability. Do not add a check here on the assumption that it makes MCP registration safe for untrusted admins — it does not, and the fix for that is not a bigger denylist. **Durable MCP task runtime** (`mcp_tasks`, `McpTaskService`; summarized in -[backend/AGENTS.md](../../../../AGENTS.md)): Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. Before creating an Agent notification run, one owner/token/version/expiry-fenced update reserves `notification_status=launching`; from that point, cancellation, ambiguous launch outcomes, and busy-thread conflicts preserve the reserved snapshot and idempotency key for recovery instead of coalescing to a newer event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools. +[backend/AGENTS.md](../../../../AGENTS.md)): Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease (claims re-check claimability per row: SQLite ignores `FOR UPDATE`), while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. Before creating an Agent notification run, one owner/token/version/expiry-fenced update reserves `notification_status=launching`; from that point, cancellation, ambiguous launch outcomes, and busy-thread conflicts preserve the reserved snapshot and idempotency key for recovery instead of coalescing to a newer event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools. **Task notification failure handling**: MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget for pre-reservation and dispatched-run failures before `dead_letter`. A permanently rejected launch, including a missing/mismatched target thread, is dead-lettered immediately instead of being recreated or reclaimed. An ambiguous launch result or busy-thread conflict preserves `launching`, the snapshot, and the same idempotency key without consuming that budget so a retry can reconcile a run that may already exist. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The HTTP cancel endpoint rejects requests with 503 when the loop is not running (`mcp_tasks_available` false, e.g. `mcp_tasks.enabled=false` with SQL persistence), so a cancellation is never acknowledged without a worker to perform it. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card. diff --git a/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py b/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py index 94a4c1b735..b43a552fa8 100644 --- a/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py +++ b/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py @@ -306,33 +306,49 @@ class McpTaskRepository: limit: int, ) -> list[dict[str, Any]]: lease_expires_at = now + timedelta(seconds=lease_seconds) - stmt = ( - select(McpTaskRow) - .where( - McpTaskRow.status.in_(_POLLABLE_STATUS_VALUES), - McpTaskRow.cancel_requested_at.is_(None), - McpTaskRow.next_poll_at.is_not(None), - McpTaskRow.next_poll_at <= now, - or_( - McpTaskRow.lease_expires_at.is_(None), - McpTaskRow.lease_expires_at < now, - ), - ) - .order_by(McpTaskRow.next_poll_at.asc(), McpTaskRow.id.asc()) - .limit(limit) - .with_for_update(skip_locked=True) + claimable = ( + McpTaskRow.status.in_(_POLLABLE_STATUS_VALUES), + McpTaskRow.cancel_requested_at.is_(None), + McpTaskRow.next_poll_at.is_not(None), + McpTaskRow.next_poll_at <= now, + or_( + McpTaskRow.lease_expires_at.is_(None), + McpTaskRow.lease_expires_at < now, + ), ) + stmt = select(McpTaskRow.id).where(*claimable).order_by(McpTaskRow.next_poll_at.asc(), McpTaskRow.id.asc()).limit(limit).with_for_update(skip_locked=True) async with self._sf() as session: - result = await session.execute(stmt) - rows = list(result.scalars()) - for row in rows: - row.lease_owner = lease_owner - row.lease_expires_at = lease_expires_at - row.lease_token = _new_claim_token() - row.poll_attempt_count += 1 - row.updated_at = now + candidate_ids = list((await session.execute(stmt)).scalars()) + # Atomic fence: ``with_for_update()`` is a no-op on SQLite, so a + # cancel request can commit between the SELECT and the claim. It + # clears the poll lease for the cancellation worker; writing the + # lease back would hold off the remote cancel until it expires. + # Each conditional UPDATE re-checks the claim predicates, and a + # row that stopped being claimable is skipped. It is not replaced + # from beyond the LIMIT: such a batch comes back short (SQLite + # only; Postgres row locks keep candidates claimable) and the + # remaining due rows are claimed on the next poll cycle. + claimed_ids: list[str] = [] + for task_id in candidate_ids: + update_result = await session.execute( + update(McpTaskRow) + .where(McpTaskRow.id == task_id, *claimable) + .values( + lease_owner=lease_owner, + lease_expires_at=lease_expires_at, + lease_token=_new_claim_token(), + poll_attempt_count=McpTaskRow.poll_attempt_count + 1, + updated_at=now, + ) + ) + if update_result.rowcount: + claimed_ids.append(task_id) + claimed: list[dict[str, Any]] = [] + if claimed_ids: + rows = {row.id: row for row in (await session.execute(select(McpTaskRow).where(McpTaskRow.id.in_(claimed_ids)))).scalars()} + claimed = [self._row_to_dict(rows[task_id], include_internal=True) for task_id in claimed_ids] await session.commit() - return [self._row_to_dict(row, include_internal=True) for row in rows] + return claimed async def apply_snapshot( self, diff --git a/backend/tests/test_mcp_task_postgres.py b/backend/tests/test_mcp_task_postgres.py index d319f47e23..b2c8a04b89 100644 --- a/backend/tests/test_mcp_task_postgres.py +++ b/backend/tests/test_mcp_task_postgres.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import os import uuid +from datetime import UTC, datetime, timedelta from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit import pytest @@ -236,3 +237,83 @@ async def test_postgres_repository_holds_share_lock_until_task_commit(postgres_r assert thread.user_id == "user-2" assert task is not None assert task.thread_incarnation == created["incarnation"] + + +@pytest.mark.asyncio +async def test_postgres_cancel_request_waits_for_poll_claim_and_then_releases_it(postgres_repositories) -> None: + """On Postgres the poll claim's row lock orders a concurrent cancel after it. + + The cancel then clears the fresh poll lease, so the cancel worker can claim + right away. This pins that the conditional claim UPDATE runs under the + ``FOR UPDATE SKIP LOCKED`` row lock rather than replacing it. + """ + thread_repo, task_repo, _session_factory = postgres_repositories + created = await thread_repo.create("thread-1", user_id="user-1") + now = datetime.now(UTC) + await task_repo.create( + task_id="task-claim-vs-cancel", + user_id="user-1", + thread_id="thread-1", + expected_thread_incarnation=created["incarnation"], + run_id=None, + tool_call_id=None, + server_name="reports", + driver_name="fake", + remote_task_id="remote-task-claim-vs-cancel", + task_name="Generate report", + status="working", + result=None, + result_preview=None, + result_truncated=False, + result_artifact=None, + error=None, + input_required=None, + next_poll_at=now - timedelta(seconds=1), + ) + engine = get_engine() + assert engine is not None + selected = asyncio.Event() + resume = asyncio.Event() + + class PauseAfterClaimSelectSession(AsyncSession): + paused = False + + async def execute(self, statement, *args, **kwargs): + result = await super().execute(statement, *args, **kwargs) + if not self.paused and getattr(statement, "is_select", False): + self.paused = True + selected.set() + await resume.wait() + return result + + claim_repo = McpTaskRepository(async_sessionmaker(engine, expire_on_commit=False, class_=PauseAfterClaimSelectSession)) + claim = asyncio.create_task(claim_repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=120, limit=1)) + cancel = None + try: + await asyncio.wait_for(selected.wait(), timeout=5) + cancel = asyncio.create_task( + task_repo.request_cancel( + "task-claim-vs-cancel", + user_id="user-1", + thread_id="thread-1", + thread_incarnation=created["incarnation"], + requested_at=now, + ) + ) + await asyncio.sleep(0.2) + assert not cancel.done() + resume.set() + claimed = await asyncio.wait_for(claim, timeout=5) + requested = await asyncio.wait_for(cancel, timeout=5) + finally: + resume.set() + for pending in (claim, cancel): + if pending is not None and not pending.done(): + pending.cancel() + await asyncio.gather(claim, *([cancel] if cancel is not None else []), return_exceptions=True) + + assert [(row["id"], row["lease_owner"]) for row in claimed] == [("task-claim-vs-cancel", "poller")] + assert requested is not None + assert requested["lease_owner"] is None + cancel_claims = await task_repo.claim_cancel_requests(now=now + timedelta(seconds=1), lease_owner="canceller", lease_seconds=60, limit=1) + assert [row["id"] for row in cancel_claims] == ["task-claim-vs-cancel"] diff --git a/backend/tests/test_mcp_task_repository.py b/backend/tests/test_mcp_task_repository.py index fda5f04579..cdfd9ae50b 100644 --- a/backend/tests/test_mcp_task_repository.py +++ b/backend/tests/test_mcp_task_repository.py @@ -171,6 +171,74 @@ async def test_interleaved_reclaim_fences_inflight_poll_and_cancel_mutations(tmp assert applied is False +@pytest.mark.asyncio +async def test_cancel_request_between_poll_claim_select_and_write_is_not_re_leased(tmp_path, monkeypatch): + """A cancel that commits inside a poll claim keeps its cleared lease, so the cancel worker claims at once.""" + repo = await _make_repo(tmp_path) + now = datetime(2026, 1, 1, tzinfo=UTC) + await _create_working_task(repo, task_id="task-cancelled", now=now) + await _create_working_task(repo, task_id="task-polled", now=now) + + async with _pause_claim_mutation(monkeypatch, repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=120, limit=2)) as (pending, resume): + requested = await repo.request_cancel("task-cancelled", user_id="user-1", thread_id="thread-1", thread_incarnation=None, requested_at=now) + assert requested is not None + assert requested["lease_owner"] is None + resume.set() + claimed = await asyncio.wait_for(pending, timeout=5) + + # The other candidate is still claimed; only the cancelled row is skipped. + assert [(row["id"], row["lease_owner"], row["poll_attempt_count"]) for row in claimed] == [("task-polled", "poller", 1)] + assert claimed[0]["lease_token"] + cancelled = await repo.get("task-cancelled", user_id="user-1", thread_id="thread-1", thread_incarnation=None) + assert cancelled is not None + assert cancelled["lease_owner"] is None + assert cancelled["poll_attempt_count"] == 0 + cancel_claims = await repo.claim_cancel_requests(now=now + timedelta(seconds=1), lease_owner="canceller", lease_seconds=60, limit=1) + assert [row["id"] for row in cancel_claims] == ["task-cancelled"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("interleaving", ["claimed_by_peer", "completed_by_previous_lease"]) +async def test_poll_claim_skips_a_row_that_stops_being_claimable_after_its_select(tmp_path, monkeypatch, interleaving): + """The claim write re-checks the lease, status and schedule its SELECT saw, not just the row id.""" + repo = await _make_repo(tmp_path) + now = datetime(2026, 1, 1, tzinfo=UTC) + task_id = "claim-interleaved" + await _create_working_task(repo, task_id=task_id, now=now) + claim_at = now + if interleaving == "completed_by_previous_lease": + previous = await repo.claim_due_tasks(now=now, lease_owner="previous", lease_seconds=60, limit=1) + # The previous lease has expired for the new claim, but its holder + # still finishes the poll it started inside the lease. + claim_at = now + timedelta(seconds=61) + + async with _pause_claim_mutation(monkeypatch, repo.claim_due_tasks(now=claim_at, lease_owner="poller", lease_seconds=120, limit=1)) as (pending, resume): + if interleaving == "claimed_by_peer": + peer = await repo.claim_due_tasks(now=claim_at, lease_owner="peer", lease_seconds=120, limit=1) + assert [row["id"] for row in peer] == [task_id] + else: + assert await repo.apply_snapshot( + task_id, + lease_owner="previous", + lease_token=previous[0]["lease_token"], + status="completed", + result={"done": True}, + result_preview=None, + result_truncated=False, + result_artifact=None, + error=None, + input_required=None, + next_poll_at=None, + polled_at=now + timedelta(seconds=30), + ) + before = await repo.get(task_id, user_id="user-1", thread_id="thread-1", thread_incarnation=None) + resume.set() + claimed = await asyncio.wait_for(pending, timeout=5) + + assert claimed == [] + assert await repo.get(task_id, user_id="user-1", thread_id="thread-1", thread_incarnation=None) == before + + @pytest.mark.asyncio @pytest.mark.parametrize("delivered", [True, False], ids=["success", "failure"]) async def test_interleaved_reclaim_fences_inflight_notification_completion(tmp_path, monkeypatch, delivered):