fix: fence stale notification delivery claims (#6211)

* fix: fence stale notification delivery claims

* test: cover notification claim migration upgrades

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/scheduler/notification_delivery.py

@@ -191,7 +191,7 @@ class NotificationDeliveryWorker:
                 # reclaims the row on a later poll.
                 logger.exception("Notification delivery %s crashed; isolating from the rest of the batch", row.get("id"))
                 try:
-                    await self._delivery_repo.mark_failed(row["id"], error="delivery crashed before completion")
+                    await self._delivery_repo.mark_failed(row["id"], claim_token=row.get("claim_token"), error="delivery crashed before completion")
                 except Exception:
                     logger.warning("Could not mark crashed delivery %s as failed; stale reset will recover it", row.get("id"), exc_info=True)
 
@@ -293,12 +293,13 @@ class NotificationDeliveryWorker:
             if still_connected is False:
                 await self._delivery_repo.mark_failed(
                     delivery_id,
+                    claim_token=delivery.get("claim_token"),
                     error=f"target is no longer a connected {provider} identity of its owner",
                     terminal=True,
                 )
                 return
             if still_connected is None:
-                await self._delivery_repo.mark_failed(delivery_id, error="could not verify the target's channel connection", count_attempt=False)
+                await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error="could not verify the target's channel connection", count_attempt=False)
                 return
         # Re-check channel liveness at delivery time, not enqueue time: the
         # channel may have been disabled or disconnected after the outbox row
@@ -310,7 +311,7 @@ class NotificationDeliveryWorker:
             # Channel outage is not the delivery's fault: park the row
             # without consuming its retry budget so it survives an
             # hours-long outage and delivers once the channel returns.
-            await self._delivery_repo.mark_failed(delivery_id, error=f"channel '{provider}' is not running", count_attempt=False)
+            await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error=f"channel '{provider}' is not running", count_attempt=False)
             return
         enriched = delivery
         if delivery.get("event") == "run_completed":
@@ -326,12 +327,12 @@ class NotificationDeliveryWorker:
                 text_markdown=render_notification_text(enriched),
             )
         except ChannelUnavailable as exc:
-            await self._delivery_repo.mark_failed(delivery_id, error=str(exc), count_attempt=False)
+            await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error=str(exc), count_attempt=False)
             return
         except Exception as exc:
-            await self._delivery_repo.mark_failed(delivery_id, error=str(exc))
+            await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error=str(exc))
             return
-        await self._delivery_repo.mark_sent(delivery_id)
+        await self._delivery_repo.mark_sent(delivery_id, claim_token=delivery.get("claim_token"))
 
     async def _run_loop(self) -> None:
         while not self._stop.is_set():

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

@@ -26,7 +26,8 @@ The empty-DB path keeps using `create_all` because `Base.metadata` is the only a
 `0023_run_change_seq` → `0023_user_preferences` →
 `0024_project_documents` → `0025_repair_run_change_seq` →
 `0026_mcp_task_lease_tokens` → `0027_notification_deliveries` →
-`0028_parked_attempts` (current head). The preference
+`0028_parked_attempts` → `0029_scheduler_agent_tasks` →
+`0030_notification_claim_tokens` (current head). The preference
 revision adds a separate owner/key table with a cascading users foreign key and
 does not alter users; the project-documents revision adds a new owner-scoped
 shelf table, and the MCP lease-token revision adds two nullable token columns to

backend/packages/harness/deerflow/persistence/migrations/versions/0030_notification_claim_tokens.py

@@ -0,0 +1,28 @@
+"""Add fencing tokens to notification delivery claims.
+
+Revision ID: 0030_notification_claim_tokens
+Revises: 0029_scheduler_agent_tasks
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+revision: str = "0030_notification_claim_tokens"
+down_revision: str | Sequence[str] | None = "0029_scheduler_agent_tasks"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_add_column
+
+    safe_add_column("notification_deliveries", sa.Column("claim_token", sa.String(length=64), nullable=True))
+
+
+def downgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_drop_column
+
+    safe_drop_column("notification_deliveries", "claim_token")

backend/packages/harness/deerflow/persistence/notification_deliveries/model.py

@@ -37,6 +37,9 @@ class NotificationDeliveryRow(Base):
     # pending -> sending -> sent | failed. A failure with retries remaining
     # returns the row to pending with a backoff ``available_at``.
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
+    # Unique fencing token for the current sending claim. Completion writes
+    # must present the token they received when claiming the row.
+    claim_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
     attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     # Incremented on channel-down / transport parking (``count_attempt=False``)
     # so permanently absent channels eventually reach a terminal ``failed``

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

@@ -3,8 +3,10 @@
 The completion hook calls :meth:`NotificationDeliveryRepository.enqueue`
 (idempotent), and the delivery worker uses ``claim_due_deliveries`` /
 ``mark_sent`` / ``mark_failed``. Claim flips rows ``pending -> sending``
-inside one transaction guarded by ``status = 'pending'``, so concurrent
-workers (multi-pod deployments) cannot double-send the same row.
+inside one transaction guarded by ``status = 'pending'`` and stamps a fresh
+``claim_token``. Completion writes that carry the token are conditionally
+updated only while that claim owns the row, fencing stale workers after
+reconciliation.
 """
 
 from __future__ import annotations
@@ -191,6 +193,8 @@ class NotificationDeliveryRepository:
         re-SELECT would also match rows a concurrent worker flipped in the
         same window and hand them out twice -- the multi-pod double-send the
         docstring above rules out.
+
+        Each claim stamps a token that completion methods use for fencing.
         """
         if limit <= 0:
             return []
@@ -220,7 +224,7 @@ class NotificationDeliveryRepository:
                             NotificationDeliveryRow.id.in_(due_ids),
                             NotificationDeliveryRow.status == "pending",
                         )
-                        .values(status="sending", updated_at=now)
+                        .values(status="sending", claim_token=uuid.uuid4().hex, updated_at=now)
                         .returning(NotificationDeliveryRow)
                     )
                 )
@@ -249,12 +253,12 @@ class NotificationDeliveryRepository:
                     NotificationDeliveryRow.status == "sending",
                     NotificationDeliveryRow.updated_at <= cutoff,
                 )
-                .values(status="pending", updated_at=now)
+                .values(status="pending", claim_token=None, updated_at=now)
             )
             await session.commit()
             return result.rowcount
 
-    async def mark_sent(self, delivery_id: str) -> dict[str, Any]:
+    async def mark_sent(self, delivery_id: str, *, claim_token: str | None = None) -> dict[str, Any]:
         async with self.session_factory() as session:
             row = await session.get(NotificationDeliveryRow, delivery_id)
             if row is None:
@@ -265,12 +269,100 @@ class NotificationDeliveryRepository:
                 # must not reopen or relabel it.
                 logger.info("notification delivery %s is already %s; ignoring late mark_sent", delivery_id, row.status)
                 return self._to_dict(row)
+            if claim_token is not None:
+                finalized = await self._fenced_completion(
+                    session,
+                    delivery_id,
+                    claim_token,
+                    values={"status": "sent", "sent_at": datetime.now(UTC), "claim_token": None},
+                )
+                if finalized is None:
+                    return await self._fenced_out_row(session, delivery_id, claim_token, "mark_sent")
+                await session.commit()
+                return self._to_dict(finalized)
             row.status = "sent"
+            row.claim_token = None
             row.sent_at = datetime.now(UTC)
             await session.commit()
             await session.refresh(row)
             return self._to_dict(row)
 
+    @staticmethod
+    async def _fenced_completion(
+        session: AsyncSession,
+        delivery_id: str,
+        claim_token: str,
+        *,
+        values: dict[str, Any],
+    ) -> NotificationDeliveryRow | None:
+        """Apply a completion only while the token still owns ``sending``."""
+        values = {**values, "updated_at": datetime.now(UTC)}
+        result = await session.execute(
+            update(NotificationDeliveryRow)
+            .where(
+                NotificationDeliveryRow.id == delivery_id,
+                NotificationDeliveryRow.status == "sending",
+                NotificationDeliveryRow.claim_token == claim_token,
+            )
+            .values(**values)
+            .returning(NotificationDeliveryRow)
+        )
+        return result.scalars().first()
+
+    async def _fenced_out_row(
+        self,
+        session: AsyncSession,
+        delivery_id: str,
+        claim_token: str,
+        operation: str,
+    ) -> dict[str, Any]:
+        logger.info(
+            "notification delivery %s is no longer owned by claim %s; ignoring late %s",
+            delivery_id,
+            claim_token[:8],
+            operation,
+        )
+        await session.rollback()
+        current = await session.get(NotificationDeliveryRow, delivery_id)
+        if current is None:
+            raise LookupError(f"notification delivery {delivery_id} not found")
+        return self._to_dict(current)
+
+    @staticmethod
+    def _failure_outcome(
+        row: NotificationDeliveryRow,
+        *,
+        error: str | None,
+        count_attempt: bool,
+        terminal: bool,
+        now: datetime,
+    ) -> dict[str, Any]:
+        values: dict[str, Any] = {"last_error": error, "claim_token": None, "updated_at": now}
+        if terminal:
+            values["status"] = "failed"
+        elif not count_attempt:
+            created_at = row.created_at
+            if created_at.tzinfo is None:
+                created_at = created_at.replace(tzinfo=UTC)
+            park_expired = (now - created_at) >= _CHANNEL_PARK_MAX_AGE or row.parked_attempts >= _CHANNEL_PARK_MAX_ATTEMPTS
+            if park_expired:
+                values["status"] = "failed"
+                values["last_error"] = error or "channel did not return before parking limit"
+            else:
+                values["parked_attempts"] = row.parked_attempts + 1
+                values["status"] = "pending"
+                values["available_at"] = now + timedelta(seconds=_CHANNEL_DOWN_RETRY_SECONDS)
+        else:
+            attempts = row.attempts + 1
+            values["attempts"] = attempts
+            if attempts >= row.max_attempts:
+                values["status"] = "failed"
+            else:
+                delay = min(_RETRY_BASE_SECONDS * 2 ** (attempts - 1), _RETRY_MAX_SECONDS)
+                values["status"] = "pending"
+                values["available_at"] = now + timedelta(seconds=delay)
+        return values
+
     async def mark_failed(
         self,
         delivery_id: str,
@@ -278,6 +370,7 @@ class NotificationDeliveryRepository:
         error: str | None = None,
         count_attempt: bool = True,
         terminal: bool = False,
+        claim_token: str | None = None,
     ) -> dict[str, Any]:
         """Record a failed attempt; reschedule with backoff while retries
         remain, otherwise finalize the row as ``failed``.
@@ -305,30 +398,16 @@ class NotificationDeliveryRepository:
                 # pending and send it later after all.
                 logger.info("notification delivery %s is already %s; ignoring late mark_failed", delivery_id, row.status)
                 return self._to_dict(row)
-            row.last_error = error
-            if terminal:
-                row.status = "failed"
-            elif not count_attempt:
-                now = datetime.now(UTC)
-                created_at = row.created_at
-                if created_at.tzinfo is None:
-                    created_at = created_at.replace(tzinfo=UTC)
-                park_expired = (now - created_at) >= _CHANNEL_PARK_MAX_AGE or row.parked_attempts >= _CHANNEL_PARK_MAX_ATTEMPTS
-                if park_expired:
-                    row.status = "failed"
-                    row.last_error = error or "channel did not return before parking limit"
-                else:
-                    row.parked_attempts += 1
-                    row.status = "pending"
-                    row.available_at = now + timedelta(seconds=_CHANNEL_DOWN_RETRY_SECONDS)
-            else:
-                row.attempts += 1
-                if row.attempts >= row.max_attempts:
-                    row.status = "failed"
-                else:
-                    delay = min(_RETRY_BASE_SECONDS * 2 ** (row.attempts - 1), _RETRY_MAX_SECONDS)
-                    row.status = "pending"
-                    row.available_at = datetime.now(UTC) + timedelta(seconds=delay)
+            now = datetime.now(UTC)
+            values = self._failure_outcome(row, error=error, count_attempt=count_attempt, terminal=terminal, now=now)
+            if claim_token is not None:
+                updated = await self._fenced_completion(session, delivery_id, claim_token, values=values)
+                if updated is None:
+                    return await self._fenced_out_row(session, delivery_id, claim_token, "mark_failed")
+                await session.commit()
+                return self._to_dict(updated)
+            for name, value in values.items():
+                setattr(row, name, value)
             await session.commit()
             await session.refresh(row)
             return self._to_dict(row)

backend/tests/test_migration_0027_0028_notification_outbox.py

@@ -1,8 +1,8 @@
 """Migration tests for the scheduled-task notification outbox (issue #4254).
 
 ``0027_notification_deliveries`` creates the outbox table and
-``0028_parked_attempts`` adds its parking counter. The latest scheduler-agent
-migration test now owns the chain-head pin.
+``0028_parked_attempts`` adds its parking counter. The scheduler-agent and
+notification claim-token migrations extend that chain in order.
 """
 
 from __future__ import annotations
@@ -21,6 +21,8 @@ pytestmark = pytest.mark.asyncio
 
 OUTBOX = "0027_notification_deliveries"
 PARKED = "0028_parked_attempts"
+SCHEDULER = "0029_scheduler_agent_tasks"
+CLAIM_TOKENS = "0030_notification_claim_tokens"
 PREVIOUS = "0026_mcp_task_lease_tokens"
 TABLE = "notification_deliveries"
 
@@ -30,8 +32,10 @@ async def test_outbox_revisions_chain_after_0026():
     assert len(script.get_heads()) == 1
     assert script.get_revision(OUTBOX).down_revision == PREVIOUS
     assert script.get_revision(PARKED).down_revision == OUTBOX
+    assert script.get_revision(SCHEDULER).down_revision == PARKED
+    assert script.get_revision(CLAIM_TOKENS).down_revision == SCHEDULER
     # alembic_version.version_num is VARCHAR(32); a longer id fails on Postgres.
-    assert max(len(OUTBOX), len(PARKED)) <= 32
+    assert max(len(OUTBOX), len(PARKED), len(SCHEDULER), len(CLAIM_TOKENS)) <= 32
 
 
 async def test_outbox_revisions_upgrade_and_downgrade(tmp_path):
@@ -56,14 +60,61 @@ async def test_outbox_revisions_upgrade_and_downgrade(tmp_path):
         await asyncio.to_thread(bootstrap._upgrade, cfg, OUTBOX)
         columns = await outbox_columns()
         assert columns is not None and "parked_attempts" not in columns
+        assert "claim_token" not in columns
 
         await asyncio.to_thread(bootstrap._upgrade, cfg, "head")
         columns = await outbox_columns()
         assert columns["parked_attempts"]["nullable"] is False
         # The backfill default is dropped so the schema matches create_all.
         assert columns["parked_attempts"]["default"] is None
+        assert columns["claim_token"]["nullable"] is True
 
         await asyncio.to_thread(command.downgrade, cfg, PREVIOUS)
         assert await outbox_columns() is None
     finally:
         await engine.dispose()
+
+
+async def test_bootstrap_claim_tokens_preserves_existing_scheduler_and_outbox(tmp_path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'scheduler.db'}")
+    cfg = bootstrap._get_alembic_config(engine)
+    try:
+        await asyncio.to_thread(bootstrap._upgrade, cfg, SCHEDULER)
+        async with engine.begin() as conn:
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO scheduled_tasks "
+                    "(id,user_id,context_mode,title,prompt,schedule_type,schedule_spec,timezone,status,overlap_policy,run_count,goal_objective,created_at,updated_at) "
+                    "VALUES ('task','owner','fresh_thread_per_run','Goal','Prompt','cron','{}','UTC','enabled','enqueue',7,'deliver',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
+                )
+            )
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO notification_deliveries "
+                    "(id,task_id,task_run_id,event,provider,target,owner_user_id,status,attempts,parked_attempts,max_attempts,payload_json,available_at,created_at,updated_at) "
+                    "VALUES ('delivery','task','occurrence','success','telegram','target','owner','pending',2,1,5,'{}',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
+                )
+            )
+
+        async def assert_preserved():
+            async with engine.connect() as conn:
+                assert (await conn.execute(sa.text("SELECT run_count,goal_objective FROM scheduled_tasks WHERE id='task'"))).one() == (7, "deliver")
+                assert (await conn.execute(sa.text("SELECT status,attempts,parked_attempts FROM notification_deliveries WHERE id='delivery'"))).one() == ("pending", 2, 1)
+
+        for _ in range(2):
+            await bootstrap.bootstrap_schema(engine, backend="sqlite")
+            await assert_preserved()
+            async with engine.connect() as conn:
+                assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == CLAIM_TOKENS
+                assert (await conn.execute(sa.text("SELECT claim_token FROM notification_deliveries WHERE id='delivery'"))).scalar_one() is None
+
+        await asyncio.to_thread(command.downgrade, cfg, SCHEDULER)
+        await assert_preserved()
+        async with engine.connect() as conn:
+            columns = await conn.run_sync(lambda sync: {col["name"] for col in sa.inspect(sync).get_columns(TABLE)})
+            assert "claim_token" not in columns
+            assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == SCHEDULER
+        await bootstrap.bootstrap_schema(engine, backend="sqlite")
+        await assert_preserved()
+    finally:
+        await engine.dispose()

backend/tests/test_migration_0029_scheduler_agent_tasks.py

@@ -23,16 +23,18 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
 
 REVISION = "0029_scheduler_agent_tasks"
+CURRENT_HEAD = "0030_notification_claim_tokens"
 PREVIOUS = "0028_parked_attempts"
 TASK_FIELDS = {"origin_thread_id", "goal_objective", "max_runs", "end_at", "standing_notes"}
 OCCURRENCE_FIELDS = {"goal_objective", "goal_verdict", "stop_requested_run_id"}
 pytestmark = pytest.mark.asyncio
 
 
-async def test_0029_owns_single_chain_head():
+async def test_0029_remains_in_the_single_migration_chain():
     script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
-    assert script.get_heads() == [REVISION]
+    assert script.get_heads() == [CURRENT_HEAD]
     assert script.get_revision(REVISION).down_revision == PREVIOUS
+    assert script.get_revision(CURRENT_HEAD).down_revision == REVISION
     assert len(REVISION) <= 32
 
 

backend/tests/test_notification_deliveries_repository.py

@@ -188,6 +188,11 @@ class TestNotificationDeliveryRepository:
 
         assert await repo.reset_stale_sending_rows(now=datetime.now(UTC) + timedelta(minutes=11), timeout=timedelta(minutes=10)) == 1
 
+        async with repo.session_factory() as session:
+            recovered = await session.get(NotificationDeliveryRow, stuck["id"])
+            assert recovered is not None
+            assert recovered.claim_token is None
+
         reclaimed = await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(minutes=11), limit=1)
         assert [row["id"] for row in reclaimed] == [stuck["id"]]
 
@@ -196,7 +201,7 @@ class TestNotificationDeliveryRepository:
         await repo.enqueue(**_enqueue_kwargs(task_run_id="run-sent"))
         await repo.enqueue(**_enqueue_kwargs(task_run_id="run-pending"))
         claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1)
-        await repo.mark_sent(claimed[0]["id"])
+        await repo.mark_sent(claimed[0]["id"], claim_token=claimed[0]["claim_token"])
 
         reset = await repo.reset_stale_sending_rows(now=datetime.now(UTC) + timedelta(hours=1), timeout=timedelta(minutes=10))
 
@@ -206,7 +211,8 @@ class TestNotificationDeliveryRepository:
     async def test_mark_sent_finalises_delivery(self, repo):
         delivery = await repo.enqueue(**_enqueue_kwargs())
 
-        updated = await repo.mark_sent(delivery["id"])
+        claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        updated = await repo.mark_sent(delivery["id"], claim_token=claimed["claim_token"])
 
         assert updated["status"] == "sent"
         assert updated["sent_at"] is not None
@@ -216,7 +222,8 @@ class TestNotificationDeliveryRepository:
         delivery = await repo.enqueue(**_enqueue_kwargs())
         before = datetime.now(UTC)
 
-        updated = await repo.mark_failed(delivery["id"], error="wecom: rate limited")
+        claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="wecom: rate limited")
 
         assert updated["status"] == "pending"
         assert updated["attempts"] == 1
@@ -232,7 +239,8 @@ class TestNotificationDeliveryRepository:
         # disconnected the target): no backoff, no parking, budget untouched.
         delivery = await repo.enqueue(**_enqueue_kwargs())
 
-        updated = await repo.mark_failed(delivery["id"], error="target is no longer connected", terminal=True)
+        claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="target is no longer connected", terminal=True)
 
         assert updated["status"] == "failed"
         assert updated["attempts"] == 0
@@ -245,19 +253,69 @@ class TestNotificationDeliveryRepository:
         # A late write from a crash handler or a stale claimant must neither
         # reopen a finished row nor relabel it.
         dropped = await repo.enqueue(**_enqueue_kwargs())
-        await repo.mark_failed(dropped["id"], error="target is no longer connected", terminal=True)
-        after_late_failure = await repo.mark_failed(dropped["id"], error="delivery crashed before completion")
-        after_late_sent = await repo.mark_sent(dropped["id"])
+        dropped_claim = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        await repo.mark_failed(dropped["id"], claim_token=dropped_claim["claim_token"], error="target is no longer connected", terminal=True)
+        after_late_failure = await repo.mark_failed(dropped["id"], claim_token=dropped_claim["claim_token"], error="delivery crashed before completion")
+        after_late_sent = await repo.mark_sent(dropped["id"], claim_token=dropped_claim["claim_token"])
         assert after_late_failure["status"] == "failed"
         assert after_late_sent["status"] == "failed"
         assert after_late_sent["last_error"] == "target is no longer connected"
         assert after_late_sent["attempts"] == 0
 
         sent = await repo.enqueue(**_enqueue_kwargs(target="other-target"))
-        await repo.mark_sent(sent["id"])
-        after_late_failure = await repo.mark_failed(sent["id"], error="late failure", count_attempt=False)
+        sent_claim = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        await repo.mark_sent(sent["id"], claim_token=sent_claim["claim_token"])
+        after_late_failure = await repo.mark_failed(sent["id"], claim_token=sent_claim["claim_token"], error="late failure", count_attempt=False)
         assert after_late_failure["status"] == "sent"
         assert after_late_failure["last_error"] is None
+
+    @pytest.mark.anyio
+    async def test_stale_worker_cannot_finalize_delivery_after_reclaim(self, repo):
+        """A recovered delivery must reject writes from its previous claim.
+
+        Worker A claims the row, then its sending lease is recovered after a
+        timeout. Worker B claims the same row. A late failure from Worker A
+        must not rewrite B's active claim or consume another retry attempt.
+        """
+        claimed_a = await repo.enqueue(**_enqueue_kwargs())
+        first_claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1)
+        assert first_claimed[0]["id"] == claimed_a["id"]
+
+        recovery_time = datetime.now(UTC) + timedelta(minutes=11)
+        assert (
+            await repo.reset_stale_sending_rows(
+                now=recovery_time,
+                timeout=timedelta(minutes=10),
+            )
+            == 1
+        )
+
+        second_claimed = await repo.claim_due_deliveries(now=recovery_time, limit=1)
+        assert second_claimed[0]["id"] == claimed_a["id"]
+
+        late_result = await repo.mark_failed(
+            first_claimed[0]["id"],
+            claim_token=first_claimed[0]["claim_token"],
+            error="late result from stale worker",
+        )
+
+        assert late_result["status"] == "sending"
+        assert late_result["attempts"] == 0
+        assert late_result["last_error"] is None
+        late_terminal = await repo.mark_failed(
+            first_claimed[0]["id"],
+            claim_token=first_claimed[0]["claim_token"],
+            error="late terminal result from stale worker",
+            terminal=True,
+        )
+        assert late_terminal["status"] == "sending"
+        assert late_terminal["last_error"] is None
+        late_sent = await repo.mark_sent(
+            first_claimed[0]["id"],
+            claim_token=first_claimed[0]["claim_token"],
+        )
+        assert late_sent["status"] == "sending"
+        assert late_sent["sent_at"] is None
         assert await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=2), limit=10) == []
 
     @pytest.mark.anyio
@@ -268,7 +326,7 @@ class TestNotificationDeliveryRepository:
         for _ in range(delivery["max_attempts"]):
             claimed = await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1)
             assert len(claimed) == 1
-            updated = await repo.mark_failed(claimed[0]["id"], error="boom")
+            updated = await repo.mark_failed(claimed[0]["id"], claim_token=claimed[0]["claim_token"], error="boom")
 
         assert updated["status"] == "failed"
         # A failed (exhausted) row is never claimed again.
@@ -281,7 +339,8 @@ class TestNotificationDeliveryRepository:
         delivery = await repo.enqueue(**_enqueue_kwargs())
         before = datetime.now(UTC)
 
-        updated = await repo.mark_failed(delivery["id"], error="channel 'wecom' is not running", count_attempt=False)
+        claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="channel 'wecom' is not running", count_attempt=False)
 
         assert updated["status"] == "pending"
         assert updated["attempts"] == 0
@@ -292,11 +351,13 @@ class TestNotificationDeliveryRepository:
         assert retry_at >= before + timedelta(minutes=10)
 
         for _ in range(_CHANNEL_PARK_MAX_ATTEMPTS - 1):
-            updated = await repo.mark_failed(updated["id"], error="channel still down", count_attempt=False)
+            claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1))[0]
+            updated = await repo.mark_failed(claimed["id"], claim_token=claimed["claim_token"], error="channel still down", count_attempt=False)
             assert updated["status"] == "pending"
             assert updated["attempts"] == 0
 
-        updated = await repo.mark_failed(updated["id"], error="channel still down", count_attempt=False)
+        claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1))[0]
+        updated = await repo.mark_failed(claimed["id"], claim_token=claimed["claim_token"], error="channel still down", count_attempt=False)
         assert updated["status"] == "failed"
         assert updated["attempts"] == 0
 
@@ -308,7 +369,8 @@ class TestNotificationDeliveryRepository:
             row.created_at = datetime.now(UTC) - _CHANNEL_PARK_MAX_AGE - timedelta(minutes=1)
             await session.commit()
 
-        updated = await repo.mark_failed(delivery["id"], error="channel 'wecom' is not running", count_attempt=False)
+        claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0]
+        updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="channel 'wecom' is not running", count_attempt=False)
 
         assert updated["status"] == "failed"
         assert updated["attempts"] == 0

backend/tests/test_notification_delivery_worker.py

@@ -39,10 +39,10 @@ class FakeDeliveryRepo:
         self.events.append("claim")
         return claimed
 
-    async def mark_sent(self, delivery_id):
+    async def mark_sent(self, delivery_id, *, claim_token=None):
         self.sent.append(delivery_id)
 
-    async def mark_failed(self, delivery_id, *, error=None, count_attempt=True, terminal=False):
+    async def mark_failed(self, delivery_id, *, claim_token=None, error=None, count_attempt=True, terminal=False):
         self.failed.append((delivery_id, error, count_attempt))
         if terminal:
             self.terminal_failures.append((delivery_id, error))