diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index ec490197eb..f610b3c7e1 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/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). diff --git a/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py b/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py index b43a552fa8..5981f08b94 100644 --- a/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py +++ b/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, diff --git a/backend/tests/test_mcp_task_repository.py b/backend/tests/test_mcp_task_repository.py index cdfd9ae50b..da0ed659b5 100644 --- a/backend/tests/test_mcp_task_repository.py +++ b/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)