fix(mcp): consume resolved notification snapshots instead of launching empty events (#6693)

* fix(mcp): consume resolved notification snapshots instead of launching empty events

A tracking-degraded snapshot recorded by a failed poll can resolve before any
notification worker claims it: the next poll succeeds, or an answered
input_required prompt returns the task to working. _record_event_if_changed
left that pending snapshot in place when the event disappeared, so
claim_notification_work rebuilt it as None and still claimed it, and the
service launched an empty notification run (dispatch_event or {}).

Consume such snapshots at claim time instead: advance notified_version, return
the status to none, clear the claim, and skip the row. The same guard consumes
rows left without a dispatch event by pre-fix claims, which would otherwise
relaunch empty notifications on every retry until dead-lettering.

* fix(mcp): clear consumed snapshot state for the review follow-ups

Review feedback on the claim-time consume guard:
- clear event_fingerprint when consuming, so an identical later attention
  event is recorded again instead of being suppressed by a snapshot that was
  never delivered;
- clear stale notification_error/attempt/dispatch state so a status
  projection cannot show a notification that no longer exists;
- cover the reclaimed expired-claim shape (claim, lease expiry, resolve,
  reclaim) and the identical-event sequence with regressions.

* docs(persistence): document the notification consume rule

mcp/AGENTS.md's effective chain has only 85 bytes of headroom on main (98,219 / 98,304), so the rule moves to persistence/AGENTS.md, which owns the mcp_tasks storage contract. No behavior change.

* fix(mcp): fence the notification consume against newer events (review)

Two P2 findings on the claim-time consume:
- the consume acknowledged the latest event version from a stale read, so on
  SQLite a completion recorded after the claim's SELECT was overwritten with
  status none and lost forever. The consume is now a conditional UPDATE fenced
  on the event version and phase that were read; a newer event makes it a
  no-op and stays claimable;
- a legacy launching/dispatched row whose dispatch payload is empty but whose
  event version moved on consumed that newer event. Such rows are now rebuilt
  and claimed; only a generation with nothing newer is consumed.
Regressions: a completion recorded while a claim is paused after its SELECT,
and legacy launching/dispatched rows with and without a newer completion.

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

@@ -30,3 +30,7 @@ Scheduled-task lifecycle rules live in one place each. `ScheduledTaskRepository.
 The per-owner run cap `scheduler.max_concurrent_runs_per_user` is effective `min(per_user, max_concurrent_runs)` (0 = off) and is checked in `claim_queued_run` right after the global count, under the same budget lock. `list_queued_runs` ranks same-thread FIFO heads per owner (`ROW_NUMBER() OVER (PARTITION BY owner)`) and orders the batch by that rank, so the drain is owner-fair even with the cap off; owners at their cap drop out before the batch limit, and orphan rows (task deleted while queued) are never held back by the cap. Scheduled-task repository boundaries coerce serialized timestamps before SQL `DateTime` binding.
 
 Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue.
+
+## MCP task notifications
+
+`mcp_tasks/sql.py::claim_notification_work` never returns a row without a dispatch event. A pending snapshot whose condition resolved before delivery (tracking recovered, or an answered `input_required` prompt returned the task to `working`) is consumed at claim time: `notified_version` advances, the content fingerprint clears so an identical later event is recorded again, stale error/attempt/dispatch state clears, status returns to `none`, and the claim lease is released instead of launching an empty event. A legacy row left without a dispatch event is consumed only when nothing newer exists; a newer recorded event is rebuilt and claimed instead. Consumption is a conditional UPDATE fenced on the event version and phase the claim read, so a concurrent newer event (SQLite has no row locks) is never acknowledged and stays claimable. Coverage: `tests/test_mcp_task_repository.py` (resolved snapshot, legacy retry, expired-claim reclaim, identical-event recurrence, interleaved newer completion, legacy launching/dispatched with and without a newer completion).

backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py

@@ -675,25 +675,78 @@ class McpTaskRepository:
         async with self._sf() as session:
             rows = list((await session.execute(stmt)).scalars())
             expires_at = now + timedelta(seconds=lease_seconds)
+            claimed: list[McpTaskRow] = []
             for row in rows:
+                rebuild_snapshot = row.notification_status in ("pending", "claimed") or (row.notification_status == "retry" and row.dispatch_version != row.event_version)
+                if not rebuild_snapshot and row.dispatch_event is None and row.dispatch_version != row.event_version:
+                    # A legacy in-flight phase whose dispatch payload is gone, with
+                    # a newer event already recorded: claim that newer event instead
+                    # of consuming it. Terminal tasks stop polling, so a consumed
+                    # completion would be lost permanently.
+                    rebuild_snapshot = True
+                event = (
+                    _notification_event(
+                        row,
+                        tracking_degraded=int(row.consecutive_poll_error_count or 0) >= tracking_degraded_after_errors,
+                    )
+                    if rebuild_snapshot
+                    else row.dispatch_event
+                )
+                if event is None:
+                    # The condition behind this snapshot resolved before delivery
+                    # (for example tracking recovered, or an input_required prompt
+                    # was answered), or the row carries an empty generation left by
+                    # a pre-fix claim. Consume only what was proven empty — the
+                    # current event version for a rebuild, the empty dispatch
+                    # generation otherwise — through a conditional UPDATE fenced on
+                    # the event version and phase this read saw. SQLite has no
+                    # ``SELECT FOR UPDATE``, so a newer event recorded after the
+                    # read must win: the fence then matches no row, this consume is
+                    # a no-op, and the newer event stays claimable. Clearing the
+                    # content fingerprint keeps an identical later event from being
+                    # suppressed by a snapshot that was never delivered.
+                    consumed_version = row.event_version if rebuild_snapshot else (row.dispatch_version if row.dispatch_version is not None else row.event_version)
+                    status_at_read = row.notification_status
+                    session.expunge(row)
+                    await session.execute(
+                        update(McpTaskRow)
+                        .where(
+                            McpTaskRow.id == row.id,
+                            McpTaskRow.event_version == consumed_version,
+                            McpTaskRow.notification_status == status_at_read,
+                        )
+                        .values(
+                            notified_version=consumed_version,
+                            event_fingerprint=None,
+                            notification_status="none",
+                            notification_error=None,
+                            notification_attempt_count=0,
+                            dispatch_version=None,
+                            dispatch_attempt=0,
+                            dispatch_event=None,
+                            notification_run_id=None,
+                            notification_lease_owner=None,
+                            notification_lease_expires_at=None,
+                            notification_lease_token=None,
+                            updated_at=now,
+                        )
+                    )
+                    continue
                 row.notification_lease_owner = lease_owner
                 row.notification_lease_expires_at = expires_at
                 row.notification_lease_token = _new_claim_token()
-                rebuild_snapshot = row.notification_status in ("pending", "claimed") or (row.notification_status == "retry" and row.dispatch_version != row.event_version)
                 if rebuild_snapshot:
                     if row.dispatch_version != row.event_version:
                         row.dispatch_attempt = 0
                         row.notification_attempt_count = 0
                     row.dispatch_version = row.event_version
-                    row.dispatch_event = _notification_event(
-                        row,
-                        tracking_degraded=int(row.consecutive_poll_error_count or 0) >= tracking_degraded_after_errors,
-                    )
+                    row.dispatch_event = event
                     row.notification_run_id = None
                     row.notification_status = "claimed"
                 row.updated_at = now
+                claimed.append(row)
             await session.commit()
-            return [self._row_to_dict(row) for row in rows]
+            return [self._row_to_dict(row) for row in claimed]
 
     async def begin_notification_launch(
         self,

backend/tests/test_mcp_task_repository.py

@@ -1644,6 +1644,395 @@ async def test_notification_retry_rebuilds_a_newer_event_and_resets_its_budget(t
     assert latest[0]["notification_attempt_count"] == 0
 
 
+@pytest.mark.asyncio
+async def test_resolved_notification_snapshot_is_consumed_instead_of_launched_empty(tmp_path):
+    """A degraded snapshot that recovers before delivery must not launch empty.
+
+    Three poll failures record the tracking-degraded snapshot; the next poll
+    succeeds before any notification worker claims it. The resolved snapshot
+    must be consumed (nothing to notify) instead of being rebuilt as an empty
+    event and launched.
+    """
+    repo = await _make_repo(tmp_path)
+    now = datetime.now(UTC)
+    await _create_working_task(repo, task_id="task-recovered", now=now)
+
+    for _ in range(3):
+        claimed = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=10)
+        await repo.release_claim(
+            "task-recovered",
+            lease_owner="poller",
+            lease_token=claimed[0]["lease_token"],
+            next_poll_at=now - timedelta(seconds=1),
+            error="temporary network failure",
+        )
+    degraded = await repo.get("task-recovered", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert degraded is not None
+    assert degraded["notification_status"] == "pending"
+    assert degraded["event_version"] == 1
+
+    claimed = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=10)
+    applied = await repo.apply_snapshot(
+        "task-recovered",
+        lease_owner="poller",
+        lease_token=claimed[0]["lease_token"],
+        status="working",
+        result=None,
+        result_preview=None,
+        result_truncated=False,
+        result_artifact=None,
+        error=None,
+        input_required=None,
+        next_poll_at=now + timedelta(seconds=5),
+        polled_at=now,
+    )
+    assert applied is True
+
+    notifications = await repo.claim_notification_work(
+        now=datetime.now(UTC),
+        lease_owner="notifier",
+        lease_seconds=60,
+        limit=10,
+        tracking_degraded_after_errors=3,
+    )
+    assert notifications == []
+
+    stored = await repo.get("task-recovered", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert stored is not None
+    assert stored["notification_status"] == "none"
+    assert stored["notified_version"] == stored["event_version"]
+
+
+@pytest.mark.asyncio
+async def test_legacy_empty_notification_retry_is_consumed_on_claim(tmp_path):
+    """Rows left by pre-fix claims can carry no event; claiming must consume them.
+
+    A pre-fix worker could claim a stale snapshot, fail delivery, and release it
+    as ``retry`` while ``dispatch_event`` stayed ``None`` and ``dispatch_version``
+    already matched ``event_version``. Re-claiming would launch another empty
+    notification on every retry until dead-lettering, so the claim must consume
+    it instead.
+    """
+    from sqlalchemy import update as sql_update
+
+    from deerflow.persistence.mcp_tasks.model import McpTaskRow
+
+    repo = await _make_repo(tmp_path)
+    now = datetime.now(UTC)
+    await _create_working_task(repo, task_id="task-legacy-empty", now=now)
+
+    async with repo._sf() as session:
+        await session.execute(
+            sql_update(McpTaskRow)
+            .where(McpTaskRow.id == "task-legacy-empty")
+            .values(
+                notification_status="retry",
+                event_version=1,
+                notified_version=0,
+                dispatch_version=1,
+                dispatch_event=None,
+                next_notification_at=now,
+            )
+        )
+        await session.commit()
+
+    notifications = await repo.claim_notification_work(
+        now=datetime.now(UTC),
+        lease_owner="notifier",
+        lease_seconds=60,
+        limit=10,
+        tracking_degraded_after_errors=3,
+    )
+    assert notifications == []
+
+    stored = await repo.get("task-legacy-empty", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert stored is not None
+    assert stored["notification_status"] == "none"
+    assert stored["notified_version"] == stored["event_version"]
+
+
+@pytest.mark.asyncio
+async def test_expired_notification_claim_is_consumed_when_the_event_resolved(tmp_path):
+    """Reclaiming an expired claim whose event resolved must not launch it.
+
+    Claim a real attention event, let the notification lease expire, resolve the
+    condition with a successful poll, then reclaim: the rebuilt snapshot is
+    empty, so the reclaim must consume it instead of launching an empty event.
+    """
+    repo = await _make_repo(tmp_path)
+    now = datetime.now(UTC)
+    await _create_working_task(repo, task_id="task-reclaim", now=now)
+
+    claimed_poll = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=1)
+    await repo.apply_snapshot(
+        "task-reclaim",
+        lease_owner="poller",
+        lease_token=claimed_poll[0]["lease_token"],
+        status="input_required",
+        result=None,
+        result_preview=None,
+        result_truncated=False,
+        result_artifact=None,
+        error=None,
+        input_required={"prompt": "Approve?"},
+        next_poll_at=now,
+        polled_at=now,
+    )
+    first = await repo.claim_notification_work(
+        now=now,
+        lease_owner="notifier",
+        lease_seconds=60,
+        limit=1,
+        tracking_degraded_after_errors=3,
+    )
+    assert first[0]["dispatch_event"]["status"] == "input_required"
+
+    claimed_poll = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=1)
+    await repo.apply_snapshot(
+        "task-reclaim",
+        lease_owner="poller",
+        lease_token=claimed_poll[0]["lease_token"],
+        status="working",
+        result=None,
+        result_preview=None,
+        result_truncated=False,
+        result_artifact=None,
+        error=None,
+        input_required=None,
+        next_poll_at=now,
+        polled_at=now,
+    )
+
+    reclaim = await repo.claim_notification_work(
+        now=now + timedelta(seconds=61),
+        lease_owner="notifier",
+        lease_seconds=60,
+        limit=1,
+        tracking_degraded_after_errors=3,
+    )
+    assert reclaim == []
+
+    stored = await repo.get("task-reclaim", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert stored is not None
+    assert stored["notification_status"] == "none"
+    assert stored["notified_version"] == stored["event_version"]
+    assert stored["dispatch_version"] is None
+    assert stored["dispatch_event"] is None
+    assert stored["notification_lease_owner"] is None
+
+
+@pytest.mark.asyncio
+async def test_identical_attention_event_after_consumed_snapshot_is_recorded(tmp_path):
+    """Consuming a resolved snapshot must not suppress an identical later event.
+
+    The degraded snapshot's content fingerprint must not survive consumption: if
+    the server degrades again with the same error string while no notification
+    was ever delivered, the event must be recorded and claimed again.
+    """
+    repo = await _make_repo(tmp_path)
+    now = datetime.now(UTC)
+    await _create_working_task(repo, task_id="task-repeat", now=now)
+
+    async def fail_three_times() -> None:
+        for _ in range(3):
+            claimed = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=10)
+            await repo.release_claim(
+                "task-repeat",
+                lease_owner="poller",
+                lease_token=claimed[0]["lease_token"],
+                next_poll_at=now - timedelta(seconds=1),
+                error="temporary network failure",
+            )
+
+    await fail_three_times()
+    degraded = await repo.get("task-repeat", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert degraded is not None
+    assert degraded["notification_status"] == "pending"
+    assert degraded["event_version"] == 1
+
+    # The task recovers before the notification is claimed; the snapshot is consumed.
+    claimed = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=10)
+    await repo.apply_snapshot(
+        "task-repeat",
+        lease_owner="poller",
+        lease_token=claimed[0]["lease_token"],
+        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),
+        polled_at=now,
+    )
+    consumed = await repo.claim_notification_work(
+        now=datetime.now(UTC),
+        lease_owner="notifier",
+        lease_seconds=60,
+        limit=10,
+        tracking_degraded_after_errors=3,
+    )
+    assert consumed == []
+
+    # The identical failure recurs; the event must be recorded again, not
+    # suppressed by the consumed snapshot's fingerprint.
+    await fail_three_times()
+    again = await repo.get("task-repeat", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert again is not None
+    assert again["notification_status"] == "pending"
+    assert again["event_version"] == 2
+
+    notifications = await repo.claim_notification_work(
+        now=datetime.now(UTC),
+        lease_owner="notifier",
+        lease_seconds=60,
+        limit=10,
+        tracking_degraded_after_errors=3,
+    )
+    assert len(notifications) == 1
+    assert notifications[0]["dispatch_event"]["tracking_degraded"] is True
+
+
+@pytest.mark.asyncio
+async def test_interleaved_consume_preserves_a_concurrently_recorded_completion(tmp_path, monkeypatch):
+    """A consume racing a newer event must not acknowledge that event (SQLite).
+
+    SQLite has no ``SELECT FOR UPDATE``, so a real completion recorded after the
+    claim's read but before its write must survive: the consume is fenced on the
+    event version it actually read, and the newer completion stays claimable.
+    """
+    repo = await _make_repo(tmp_path)
+    now = datetime.now(UTC)
+    await _create_working_task(repo, task_id="interleaved-consume", now=now)
+
+    # Record the degraded snapshot, then let the task recover so the snapshot is
+    # resolved but still pending when the claim tries to consume it.
+    for _ in range(3):
+        claimed = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=10)
+        await repo.release_claim(
+            "interleaved-consume",
+            lease_owner="poller",
+            lease_token=claimed[0]["lease_token"],
+            next_poll_at=now - timedelta(seconds=1),
+            error="temporary network failure",
+        )
+    claimed = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=10)
+    await repo.apply_snapshot(
+        "interleaved-consume",
+        lease_owner="poller",
+        lease_token=claimed[0]["lease_token"],
+        status="working",
+        result=None,
+        result_preview=None,
+        result_truncated=False,
+        result_artifact=None,
+        error=None,
+        input_required=None,
+        next_poll_at=now,
+        polled_at=now,
+    )
+
+    claim_kwargs = {"lease_owner": "notifier", "lease_seconds": 60, "limit": 1, "tracking_degraded_after_errors": 3}
+    async with _pause_claim_mutation(monkeypatch, repo.claim_notification_work(now=datetime.now(UTC), **claim_kwargs)) as (pending, resume):
+        # Record a real completion while the consume is paused after its SELECT.
+        poll = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=1)
+        assert await repo.apply_snapshot(
+            "interleaved-consume",
+            lease_owner="poller",
+            lease_token=poll[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,
+        )
+        resume.set()
+        consumed = await asyncio.wait_for(pending, timeout=5)
+
+    # The paused claim consumed nothing; the completion stays claimable.
+    assert consumed == []
+    stored = await repo.get("interleaved-consume", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert stored is not None
+    assert stored["notification_status"] == "pending"
+    assert stored["event_version"] == 2
+    assert stored["notified_version"] == 0
+
+    notifications = await repo.claim_notification_work(now=datetime.now(UTC), **claim_kwargs)
+    assert len(notifications) == 1
+    assert notifications[0]["dispatch_event"]["status"] == "completed"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("legacy_status", ["launching", "dispatched"])
+@pytest.mark.parametrize("newer_event", [True, False], ids=["newer-event", "no-newer-event"])
+async def test_legacy_empty_dispatch_respects_a_newer_event(tmp_path, legacy_status, newer_event):
+    """A legacy empty dispatch must not swallow a newer completion.
+
+    Pre-fix workers could leave a launching/dispatched row whose dispatch event
+    is gone. If a real completion was recorded afterwards, it must be claimed;
+    only an empty generation with nothing newer may be consumed.
+    """
+    from sqlalchemy import update as sql_update
+
+    from deerflow.persistence.mcp_tasks.model import McpTaskRow
+
+    repo = await _make_repo(tmp_path)
+    now = datetime.now(UTC)
+    await _create_working_task(repo, task_id="legacy-newer", now=now)
+
+    async with repo._sf() as session:
+        await session.execute(
+            sql_update(McpTaskRow)
+            .where(McpTaskRow.id == "legacy-newer")
+            .values(
+                notification_status=legacy_status,
+                event_version=1,
+                notified_version=0,
+                dispatch_version=1,
+                dispatch_event=None,
+                notification_run_id="legacy-empty-run",
+                next_notification_at=now,
+            )
+        )
+        await session.commit()
+
+    if newer_event:
+        poll = await repo.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=60, limit=1)
+        assert await repo.apply_snapshot(
+            "legacy-newer",
+            lease_owner="poller",
+            lease_token=poll[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,
+        )
+
+    claim_kwargs = {"lease_owner": "notifier", "lease_seconds": 60, "limit": 1, "tracking_degraded_after_errors": 3}
+    notifications = await repo.claim_notification_work(now=datetime.now(UTC), **claim_kwargs)
+
+    stored = await repo.get("legacy-newer", user_id="user-1", thread_id="thread-1", thread_incarnation=None)
+    assert stored is not None
+    if newer_event:
+        assert len(notifications) == 1
+        assert notifications[0]["dispatch_event"]["status"] == "completed"
+        assert notifications[0]["dispatch_version"] == 2
+    else:
+        assert notifications == []
+        assert stored["notification_status"] == "none"
+        assert stored["notified_version"] == stored["event_version"]
+
+
 @pytest.mark.asyncio
 async def test_unexpected_notification_failure_releases_lease_without_changing_phase(tmp_path):
     repo = await _make_repo(tmp_path)