fix(mcp): skip tasks a cancel released while the poll claim was in flight (#6859)

* fix(mcp): skip tasks a cancel released while the poll claim was in flight

claim_due_tasks selected due rows with FOR UPDATE SKIP LOCKED, then set
the lease through an ORM flush that updates by id alone. FOR UPDATE is a
no-op on SQLite, so a request_cancel that committed between the SELECT
and the flush had its cleared poll lease written back. The cancellation
worker only claims rows without a live lease, so the remote cancel waited
for mcp_tasks.lease_seconds (120 s by default), and the task was polled
once more for a result apply_snapshot then discarded.

The claim now selects candidate ids and leases each one with a
conditional UPDATE that re-checks the claim predicates (pollable status,
no cancel request, due next_poll_at, no live lease), returning only the
rows it actually leased, the same atomic fence release_claim already
uses. On Postgres the row lock still orders the two writers.

* docs(changelog): reference #6859 for the MCP task claim fence

* docs(mcp): note that a short poll claim batch is not topped up

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

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

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.
 

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,

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"]

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):