diff --git a/TODOS.md b/TODOS.md
index 7ecbcb8..051effb 100644
--- a/TODOS.md
+++ b/TODOS.md
@@ -84,10 +84,16 @@ Observed twice: when the parent Claude process exits, the `entrabot-mcp` child k
 - **Source:** Live observation 2026-04-17 (second occurrence in one day)
 
 ### Multi-instance cursor consistency: bootstrap fails open → fleet replay flood
-The background Teams poll re-pushes a chat's newest message on *any* failure to read a fresh cloud cursor — absent, stale, corrupt, and read-exception all fall through to `_bootstrap_chat`, which pushes. In a fleet (N instances → one blob container) these misses are routine: silent `LocalBackend` fallback when blob env is half-configured, transient blob read failures, the 24h `is_stale` cap re-bootstrapping a cold store, and last-writer-wins cursor writes with no ETag. Result: idle chats replay their newest (weeks-old) message. Also a security surface — the replay re-injects stale imperative messages ("read special data in my Documents", "ship to <address>", "run <script>") into the agent's channel. The steady-state gate (`_filter_new_messages`) already supports N-instances-one-store; only the bootstrap decision and write coordination break it. Confirmed 2026-07-02 (a base instance pointed at a shared blob container replayed ~5-week-old DMs while the blob cursors were correct + fresh). Fix: fail closed on read-miss + per-message cloud idempotency ledger + co-locate `watched_chats` in cloud + catch-up-read instead of bootstrap-on-stale + `If-Match` ETag concurrency on cursor writes + assert uniform backend at boot.
-- **Effort:** M
-- **Depends on:** "MCP server orphans when Claude Code exits" (enforce the singleton so duplicate pollers can't start); ADR-005 blob backend.
-- **See:** `docs/architecture/DESIGN-multi-instance-cursor-consistency.md`
+**Status: mostly shipped on `fix/fleet-safe-cursor`** (F1, F2, F4, F5, per-message cloud idempotency). F3 (cloud-authoritative `watched_chats`) deferred; F6 served by the existing process singleton.
+
+The background Teams poll re-pushed a chat's newest message on *any* failure to read a fresh cloud cursor — absent, stale, corrupt, and read-exception all fell through to `_bootstrap_chat`, which pushes. In a fleet (N instances → one blob container) these misses are routine: silent `LocalBackend` fallback when blob env is half-configured, transient blob read failures, the 24h `is_stale` cap re-bootstrapping a cold store, and last-writer-wins cursor writes with no ETag. Result: idle chats replay their newest (weeks-old) message. Also a security surface — the replay re-injects stale imperative messages ("read special data in my Documents", "ship to <address>", "run <script>") into the agent's channel. Confirmed 2026-07-02 (a base instance pointed at a shared blob container replayed ~5-week-old DMs while the blob cursors were correct + fresh).
+
+**Shipped:** fail-closed cursor classification (`resolve_cursor` → PRESENT/ABSENT/UNRESOLVED); rehydrate-and-catch-up on stale instead of re-bootstrap; per-message cloud idempotency via `claim_delivery` (CAS on the shared `seen_ids_tail` ledger); `If-Match` ETag concurrency on cursor writes (`get_with_etag` + `read_text_with_etag` + `write_text(if_match=)`); boot-time backend assertion (`assert_backend_config`, half-config → `BackendMisconfiguredError`).
+
+**Remaining:** F3 — co-locate the `watched_chats` list in the cloud store (local file becomes a cache). Deferred as the largest structural change and not required by the fail-closed + idempotency guarantees.
+- **Effort:** S (F3 only)
+- **Depends on:** "MCP server orphans when Claude Code exits" (enforce the singleton so duplicate pollers can't start — the process singleton exists; orphan cleanup is the remaining gap); ADR-005 blob backend.
+- **See:** `docs/architecture/DESIGN-multi-instance-cursor-consistency.md` (§Shipped / §Deferred)
 - **Source:** Live diagnosis 2026-07-02.
 
 ### Daily summary scheduler: wrong day + double-fire
diff --git a/docs/architecture/DESIGN-multi-instance-cursor-consistency.md b/docs/architecture/DESIGN-multi-instance-cursor-consistency.md
index e860db5..16a720d 100644
--- a/docs/architecture/DESIGN-multi-instance-cursor-consistency.md
+++ b/docs/architecture/DESIGN-multi-instance-cursor-consistency.md
@@ -1,6 +1,10 @@
 # DESIGN — Multi-Instance Cursor Consistency (fleet-safe Teams channel poll)
 
-> **Status:** Proposed. Problem confirmed in production 2026-07-02.
+> **Status:** Partially shipped (branch `fix/fleet-safe-cursor`). F1, F2, F4,
+> F5, and per-message cloud idempotency are implemented + tested. F6 is served
+> by the existing process singleton (`src/entrabot/singleton.py`); orphan
+> cleanup is tracked separately. F3 (cloud-authoritative `watched_chats`) is
+> deferred — see "Deferred" below. Problem confirmed in production 2026-07-02.
 > **Owner:** unassigned (handed off for implementation).
 > **Related:** ADR-005 (cloud-hosted memory), `docs/decisions/005-cloud-hosted-memory.md`;
 > TODOS.md → "MCP server orphans when Claude Code exits" (P1) and
@@ -227,3 +231,48 @@ surface entirely. Treat F1 as security-relevant, not cosmetic.
   `seen_ids_tail`). A separate `delivered/` object is optional and additive.
 - Land F1 + F6 first (stops the flood and the injection surface), then F2/F3,
   then F4/F5 (correctness under long downtime and high write concurrency).
+
+## Shipped (branch `fix/fleet-safe-cursor`)
+
+Landed with full TDD (`pytest -v && ruff check .` green):
+
+- **F1 — fail closed on read miss.** `chat_cursors.resolve_cursor()` classifies
+  a cursor read as `PRESENT` / `ABSENT` / `UNRESOLVED`. A backend read error or
+  a corrupt payload is `UNRESOLVED` (distinct from a successful empty read,
+  `ABSENT`). `_register_watched_chat` flags `UNRESOLVED` chats
+  `needs_resolution`; `_poll_watched_chat` re-reads them each cycle and pushes
+  nothing until they resolve. Only a provably-`ABSENT` chat may surface its
+  newest message once.
+- **F4 — catch-up read on stale.** A present-but-stale cursor now rehydrates
+  (never re-bootstraps); the steady-state timestamp gate surfaces only messages
+  newer than `last_ts`. Idle chats emit nothing.
+- **Per-message cloud idempotency.** `chat_cursors.claim_delivery()` treats the
+  shared cursor's `seen_ids_tail` as the fleet delivery ledger: an ETag
+  compare-and-swap records the candidate batch and returns only the ids THIS
+  instance won. Across N instances exactly one wins each id; losers claim
+  nothing. New-chat first delivery also flows through it (bootstrap defers the
+  push one cycle). All ambiguity (read error / corrupt / exhausted retries)
+  fails closed.
+- **F5 — optimistic concurrency.** `BlobStore.get_with_etag()` +
+  `MemoryBackend.read_text_with_etag()` / `write_text(if_match=)` (Local uses a
+  content-hash ETag, Blob the real ETag). `claim_delivery` merges `max(last_ts)`
+  + `union(seen_ids)` and retries on `ConcurrencyError`, so a slow writer can't
+  regress the watermark.
+- **F2 — uniform backend at boot.** `get_backend()` raises
+  `BackendMisconfiguredError` on a half-configured blob env instead of silently
+  using Local. `assert_backend_config()` (called from `_initialize`) logs the
+  resolved backend + container and fails loud on half-config.
+- **F6 — singleton.** Served by the existing `src/entrabot/singleton.py` flock
+  acquired at the top of `main()`. Orphan cleanup (background tasks outside
+  FastMCP's lifespan cancel scope) remains tracked in `TODOS.md` ("MCP server
+  orphans when Claude Code exits").
+
+### Deferred
+
+- **F3 — cloud-authoritative `watched_chats`.** Not shipped in this branch. It
+  is the largest structural change (moving the watched-chats list into the
+  cloud store with the local file as a cache) and is not required by the
+  fail-closed + idempotency guarantees above: with F1/F2 + the idempotency
+  ledger, an instance that registers a chat late or on a divergent local list
+  still cannot replay — it either fails closed (unresolved) or loses the
+  delivery claim. Tracked as a follow-up.
diff --git a/docs/engineering-status.md b/docs/engineering-status.md
index a4ebc6c..500b3d6 100644
--- a/docs/engineering-status.md
+++ b/docs/engineering-status.md
@@ -14,7 +14,7 @@ Source of truth for detail: `TODOS.md` in the repository root. One line each bel
 - **Script-toolkit docs closeout** — `./status.sh` is the canonical entry; finish the remaining script-reference polish and smoke verification. See `TODOS.md` P1.
 - **Test isolation: blob env leakage** — `tmp_data_dir` fixture in `tests/tools/test_interaction_log.py` doesn't clear `ENTRABOT_BLOB_ENDPOINT`; 10 tests fail on any machine with blob env configured. Partially addressed: `test_interaction_log.py`, `test_daily_summary.py`, and `test_email_poll.py` fixtures now unset blob env; session-scoped autouse fixture still open.
 - **MCP server orphans on Claude Code exit** — background poll tasks sit outside FastMCP's lifespan cancel scope; new sessions spawn a second server, both poll Graph independently.
-- **Multi-instance cursor consistency (fleet-safe channel poll)** — issue #17 fixed single-instance restart replay, but the poll's bootstrap path still fails *open* on any cloud-cursor read miss, so a mis-enved or transiently-failing instance in an N-instance fleet replays every idle chat's newest (weeks-old) message — and re-injects stale imperative messages into the channel. Fix plan: fail-closed miss path + per-message cloud idempotency + cloud-authoritative `watched_chats` + ETag cursor writes + boot-time backend assertion. See `docs/architecture/DESIGN-multi-instance-cursor-consistency.md`. Confirmed 2026-07-02.
+- **Multi-instance cursor consistency (fleet-safe channel poll)** — mostly shipped on `fix/fleet-safe-cursor`: the poll's bootstrap path now fails *closed* on any cloud-cursor read miss (`resolve_cursor` → PRESENT/ABSENT/UNRESOLVED), stale cursors catch up instead of re-bootstrapping, delivery is idempotent across the fleet via an ETag compare-and-swap on the shared cursor ledger (`claim_delivery`), cursor writes use `If-Match` concurrency, and a half-configured blob backend fails loud at boot. Remaining: F3 (cloud-authoritative `watched_chats`) deferred. See `docs/architecture/DESIGN-multi-instance-cursor-consistency.md` §Shipped. Confirmed 2026-07-02.
 - **Daily summary scheduler — wrong day + double-fire** — UTC-based `target_day` summarizes the brand-new UTC day at 5pm PDT; scheduler fired twice at the same second on 2026-04-17.
 
 ## Recently Shipped
diff --git a/src/entrabot/errors.py b/src/entrabot/errors.py
index 920478e..30a0dad 100644
--- a/src/entrabot/errors.py
+++ b/src/entrabot/errors.py
@@ -33,6 +33,31 @@ class ConfigError(EntraBotError):
     """Configuration errors (invalid or removed settings)."""
 
 
+class BackendMisconfiguredError(ConfigError):
+    """Storage backend env is half-configured (design F2).
+
+    Exactly one of ``ENTRABOT_BLOB_ENDPOINT`` / ``ENTRABOT_BLOB_CONTAINER``
+    is set. Silently falling back to Local here is how a mis-enved fleet
+    instance ends up on an empty local store and re-bootstraps every chat
+    (the replay-flood root cause). Fail loud so the operator fixes the env
+    instead of the fleet diverging. Set ``ENTRABOT_KEEP_MEMORY_LOCAL=true``
+    to deliberately force Local.
+    """
+
+    def __init__(self, *, endpoint: str | None, container: str | None) -> None:
+        self.endpoint = endpoint
+        self.container = container
+        missing = "container" if endpoint else "endpoint"
+        super().__init__(
+            "storage backend is half-configured: "
+            f"blob {missing} is unset "
+            f"(endpoint={'set' if endpoint else 'unset'}, "
+            f"container={'set' if container else 'unset'}). "
+            "Set BOTH ENTRABOT_BLOB_ENDPOINT and ENTRABOT_BLOB_CONTAINER, or set "
+            "ENTRABOT_KEEP_MEMORY_LOCAL=true to force local storage."
+        )
+
+
 class InsecureKeyringBackendError(EntraBotError):
     """The selected keyring backend is not an approved OS keystore."""
 
diff --git a/src/entrabot/mcp_server.py b/src/entrabot/mcp_server.py
index 5d8057f..662b6ed 100644
--- a/src/entrabot/mcp_server.py
+++ b/src/entrabot/mcp_server.py
@@ -1022,6 +1022,20 @@ async def _initialize() -> None:
     await _init_auth()
     await _init_poll()
 
+    # F2: resolve + validate the storage backend once at boot. A half-configured
+    # blob env raises here (fail loud) instead of silently using an empty Local
+    # store and re-bootstrapping every chat (the fleet replay-flood root cause).
+    # Logs the resolved backend + container so operators can confirm fleet-wide
+    # agreement in production logs.
+    try:
+        from entrabot.storage.backend import assert_backend_config
+
+        assert_backend_config(logger=logger)
+    except Exception as exc:
+        if logger:
+            logger.error("Storage backend misconfigured: %s", exc)
+        raise
+
     # authorization fix: log the active-sponsor-channel binding TTL so operators
     # can verify the security configuration in production logs.
     if logger:
@@ -1208,45 +1222,59 @@ def _register_watched_chat(chat_id: str, *, persist: bool = True) -> None:
     """
     watched = _state.get("watched_chats", {})
     if chat_id not in watched:
-        # Issue #17: try to rehydrate a persisted cursor first. If it exists
-        # and ``last_ts`` is within the staleness cap, we keep the prior
-        # process's seen-set + watermark and skip ``_bootstrap_chat`` (which
-        # would otherwise re-fire the "newest at boot" message even when that
-        # message is days old). If absent, stale, or corrupt, fall through to
-        # the existing fresh-state path and let the bootstrap path baseline.
-        from entrabot.tools.chat_cursors import is_stale, load_cursor
-
-        rehydrated: dict | None = None
-        try:
-            cursor = load_cursor(chat_id)
-        except Exception as exc:  # noqa: BLE001
-            # Backend read failure (transient disk/blob error) → bootstrap.
-            cursor = None
-            if logger:
-                logger.warning(
-                    "Cursor load failed for %s (treating as absent): %s: %s",
-                    chat_id,
-                    type(exc).__name__,
-                    exc,
-                )
-        if cursor and not is_stale(cursor.get("last_ts")):
-            rehydrated = cursor
-
-        if rehydrated is not None:
+        # Issue #17 / fleet-safety F1+F4: classify the persisted cursor read
+        # into PRESENT / ABSENT / UNRESOLVED and fail CLOSED on ambiguity.
+        #
+        #   PRESENT (fresh OR stale) → rehydrate seen-set + watermark and mark
+        #     bootstrapped. The steady-state timestamp gate does catch-up (F4);
+        #     a present cursor is NEVER re-bootstrapped, so idle chats can't
+        #     replay their newest message.
+        #   ABSENT (read succeeded, no cursor) → genuinely new chat; leave
+        #     un-bootstrapped so the first poll surfaces the newest message once.
+        #   UNRESOLVED (read failed / corrupt / ambiguous) → do NOT push. Flag
+        #     ``needs_resolution`` so the poll loop re-reads before delivering
+        #     anything. This is the core fix for the fleet replay flood: a
+        #     transient blob 401/timeout/throttle must never be mistaken for a
+        #     new chat and trigger a bootstrap push of a weeks-old message.
+        from entrabot.tools.chat_cursors import CursorOutcome, resolve_cursor
+
+        resolution = resolve_cursor(chat_id)
+
+        if resolution.outcome is CursorOutcome.PRESENT:
+            cursor = resolution.cursor or {}
             watched[chat_id] = {
-                "seen_ids": set(rehydrated.get("seen_ids_tail") or []),
-                "last_ts": rehydrated.get("last_ts"),
+                "seen_ids": set(cursor.get("seen_ids_tail") or []),
+                "last_ts": cursor.get("last_ts"),
                 "bootstrapped": True,
+                "needs_resolution": False,
             }
             if logger:
                 logger.info(
                     "Rehydrated chat cursor (last_ts=%s, seen_ids=%d): %s",
-                    rehydrated.get("last_ts"),
+                    cursor.get("last_ts"),
                     len(watched[chat_id]["seen_ids"]),
                     chat_id,
                 )
-        else:
-            watched[chat_id] = {"seen_ids": set(), "last_ts": None, "bootstrapped": False}
+        elif resolution.outcome is CursorOutcome.UNRESOLVED:
+            watched[chat_id] = {
+                "seen_ids": set(),
+                "last_ts": None,
+                "bootstrapped": False,
+                "needs_resolution": True,
+            }
+            if logger:
+                logger.warning(
+                    "Cursor UNRESOLVED at registration for %s — failing closed, "
+                    "will re-resolve before delivering (no bootstrap push)",
+                    chat_id,
+                )
+        else:  # ABSENT — genuinely new chat.
+            watched[chat_id] = {
+                "seen_ids": set(),
+                "last_ts": None,
+                "bootstrapped": False,
+                "needs_resolution": False,
+            }
             if logger:
                 logger.info("Registered chat for background polling: %s", chat_id)
         _state["watched_chats"] = watched
@@ -1325,6 +1353,120 @@ async def _bootstrap_chat(chat_id: str) -> None:
     _schedule_cursor_save(chat_id)
 
 
+def _apply_present_cursor(chat_state: dict, cursor: dict | None) -> None:
+    """Rehydrate *chat_state* in place from a PRESENT cursor read.
+
+    Carries the persisted seen-set + watermark and marks the chat bootstrapped
+    so the steady-state gate (not ``_bootstrap_chat``) drives delivery. Clears
+    ``needs_resolution``.
+    """
+    cursor = cursor or {}
+    chat_state["seen_ids"] = set(cursor.get("seen_ids_tail") or [])
+    chat_state["last_ts"] = cursor.get("last_ts")
+    chat_state["bootstrapped"] = True
+    chat_state["needs_resolution"] = False
+
+
+def _resolve_pending_chat(chat_id: str, chat_state: dict) -> None:
+    """Re-attempt cursor resolution for a chat flagged ``needs_resolution``.
+
+    Fail-closed retry (design F1): the registration read was ambiguous
+    (transient failure / corrupt), so we re-read here on the poll cadence.
+
+    * PRESENT → rehydrate in place; the NEXT cycle delivers via the
+      steady-state gate. This cycle pushes nothing.
+    * ABSENT → clear the flag; the chat is genuinely new, so the next cycle's
+      bootstrap surfaces its newest message once.
+    * UNRESOLVED → leave the flag set; retry again next cycle. Still no push.
+
+    Never pushes and never bootstraps within this call — the resolution cycle
+    is always delivery-free.
+    """
+    from entrabot.tools.chat_cursors import CursorOutcome, resolve_cursor
+
+    resolution = resolve_cursor(chat_id)
+    if resolution.outcome is CursorOutcome.UNRESOLVED:
+        return  # still ambiguous — fail closed, retry next cycle
+    if resolution.outcome is CursorOutcome.PRESENT:
+        _apply_present_cursor(chat_state, resolution.cursor)
+        if logger:
+            logger.info(
+                "Re-resolved chat cursor to PRESENT (last_ts=%s): %s",
+                chat_state.get("last_ts"),
+                chat_id,
+            )
+    else:  # ABSENT — genuinely new; allow the next cycle to bootstrap.
+        chat_state["needs_resolution"] = False
+        if logger:
+            logger.info("Re-resolved chat cursor to ABSENT (new chat): %s", chat_id)
+
+
+async def _poll_watched_chat(
+    chat_id: str,
+    chat_state: dict,
+    agent_display_name: str,
+) -> None:
+    """Run one poll cycle for a single watched chat.
+
+    Three mutually-exclusive branches, in priority order:
+
+    1. ``needs_resolution`` → re-read the cursor (fail-closed). Never delivers.
+    2. not ``bootstrapped`` → ``_bootstrap_chat`` (surface newest once). Only
+       reached for a chat that resolved to ABSENT — a genuinely new chat.
+    3. bootstrapped → steady-state: read, filter, push genuinely-new messages,
+       advance + persist the cursor.
+    """
+    from entrabot.tools.teams import filter_human_messages, read
+
+    # (1) Fail-closed resolution retry — pushes nothing this cycle.
+    if chat_state.get("needs_resolution"):
+        _resolve_pending_chat(chat_id, chat_state)
+        return
+
+    # (2) Genuinely-new chat → bootstrap once.
+    if not chat_state.get("bootstrapped"):
+        await _bootstrap_chat(chat_id)
+        return
+
+    # (3) Steady state.
+    raw_messages = await _with_token_retry(read, chat_id=chat_id, count=10)
+    human_msgs = filter_human_messages(raw_messages, agent_display_name)
+    new_msgs = _filter_new_messages(
+        human_msgs,
+        chat_state["last_ts"],
+        chat_state["seen_ids"],
+    )
+    if not new_msgs:
+        return
+
+    from entrabot.tools.chat_cursors import claim_delivery
+
+    newest = max(new_msgs, key=lambda m: m.get("sent_at", ""))
+    candidate_ids = [m["message_id"] for m in new_msgs]
+
+    # Fleet idempotency + F5: claim delivery in the SHARED cloud ledger via an
+    # ETag compare-and-swap. Only ids this instance wins are pushed; the CAS
+    # write persists the advanced cursor (union seen-set, max watermark) so a
+    # sibling can't clobber it with a stale one. On any ambiguity claim_delivery
+    # fails closed (returns []), so we push nothing rather than risk a replay.
+    claimed = set(claim_delivery(chat_id, candidate_ids, newest["sent_at"]))
+
+    # Advance the in-memory cursor for ALL candidates — they were handled this
+    # cycle regardless of which instance pushed — so the next local cycle won't
+    # re-evaluate them.
+    chat_state["last_ts"] = newest["sent_at"]
+    for mid in candidate_ids:
+        chat_state["seen_ids"].add(mid)
+
+    # Bounded cleanup (keep last 500)
+    if len(chat_state["seen_ids"]) > SEEN_SET_MAX:
+        chat_state["seen_ids"] = set(sorted(chat_state["seen_ids"])[-100:])
+
+    for m in sorted(new_msgs, key=lambda m: m.get("sent_at", "")):
+        if m["message_id"] in claimed:
+            await _push_channel_notification(m, chat_id=chat_id)
+
+
 async def _background_poll() -> None:
     """Background polling loop — pushes inbound Teams messages to Claude Code.
 
@@ -1344,8 +1486,6 @@ async def _background_poll() -> None:
     """
     import asyncio
 
-    from entrabot.tools.teams import filter_human_messages, read
-
     if logger:
         logger.info("Starting background Teams poll (interval=%ds)", BACKGROUND_POLL_INTERVAL)
 
@@ -1372,48 +1512,7 @@ async def _background_poll() -> None:
 
             for chat_id, chat_state in watched.items():
                 try:
-                    # Bootstrap on first encounter
-                    if not chat_state.get("bootstrapped"):
-                        await _bootstrap_chat(chat_id)
-                        continue
-
-                    raw_messages = await _with_token_retry(
-                        read,
-                        chat_id=chat_id,
-                        count=10,
-                    )
-                    human_msgs = filter_human_messages(
-                        raw_messages,
-                        agent_display_name,
-                    )
-                    new_msgs = _filter_new_messages(
-                        human_msgs,
-                        chat_state["last_ts"],
-                        chat_state["seen_ids"],
-                    )
-
-                    if new_msgs:
-                        newest = max(new_msgs, key=lambda m: m.get("sent_at", ""))
-                        chat_state["last_ts"] = newest["sent_at"]
-                        for m in new_msgs:
-                            chat_state["seen_ids"].add(m["message_id"])
-
-                        # Bounded cleanup (keep last 500)
-                        if len(chat_state["seen_ids"]) > SEEN_SET_MAX:
-                            chat_state["seen_ids"] = set(sorted(chat_state["seen_ids"])[-100:])
-
-                        for m in sorted(
-                            new_msgs,
-                            key=lambda m: m.get("sent_at", ""),
-                        ):
-                            await _push_channel_notification(m, chat_id=chat_id)
-
-                        # Issue #17: persist the advanced cursor through the
-                        # MemoryBackend so the next process picks up where we
-                        # left off instead of replaying the newest-at-boot
-                        # message as fresh. Debounced ~1s so a chatty group
-                        # chat doesn't trigger a write per message.
-                        _schedule_cursor_save(chat_id)
+                    await _poll_watched_chat(chat_id, chat_state, agent_display_name)
                 except Exception as chat_exc:
                     # One chat's failure must not starve the others in this
                     # cycle. Log and move on; the next cycle will retry.
diff --git a/src/entrabot/storage/backend.py b/src/entrabot/storage/backend.py
index 4c0ac38..5785b4a 100644
--- a/src/entrabot/storage/backend.py
+++ b/src/entrabot/storage/backend.py
@@ -20,18 +20,31 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from typing import TYPE_CHECKING, Protocol, runtime_checkable
 
 from entrabot.config import get_config
-from entrabot.storage.blob import BlobStore
+from entrabot.errors import BackendMisconfiguredError
+from entrabot.storage.blob import BlobStore, ConcurrencyError
 from entrabot.tools.teams import acquire_agent_user_storage_token
 
 if TYPE_CHECKING:
     from collections.abc import Coroutine
 
 
+def _content_etag(content: str) -> str:
+    """Deterministic content-hash ETag for local files.
+
+    Mirrors Azure Blob ETag semantics closely enough for optimistic
+    concurrency: identical content → identical ETag; any change → different
+    ETag. Used by :class:`LocalBackend` so the same CAS code path works for
+    both local and blob storage.
+    """
+    return hashlib.sha256(content.encode("utf-8")).hexdigest()
+
+
 @runtime_checkable
 class MemoryBackend(Protocol):
     """Sync key→text store for agent state.
@@ -44,8 +57,25 @@ def read_text(self, key: str) -> str | None:
         """Return text at *key*, or ``None`` if it doesn't exist."""
         ...
 
-    def write_text(self, key: str, content: str) -> None:
-        """Replace *key*'s content with *content*. Creates parents as needed."""
+    def read_text_with_etag(self, key: str) -> tuple[str | None, str | None]:
+        """Return ``(text, etag)`` for *key*, or ``(None, None)`` if absent.
+
+        The ETag is an opaque concurrency token to pass back to
+        :meth:`write_text` as ``if_match`` for an optimistic-concurrency
+        write (design F5). Backends without native ETags synthesize one from
+        the content hash.
+        """
+        ...
+
+    def write_text(self, key: str, content: str, *, if_match: str | None = None) -> str | None:
+        """Replace *key*'s content with *content*. Creates parents as needed.
+
+        When ``if_match`` is provided, the write is conditional: it succeeds
+        only if *key*'s current ETag equals ``if_match``, otherwise it raises
+        :class:`entrabot.storage.blob.ConcurrencyError`. ``if_match=None``
+        (default) is an unconditional overwrite. Returns the new ETag (or
+        ``None`` for backends that don't track one).
+        """
         ...
 
     def append_text(self, key: str, content: str) -> None:
@@ -77,10 +107,26 @@ def read_text(self, key: str) -> str | None:
             return None
         return p.read_text()
 
-    def write_text(self, key: str, content: str) -> None:
+    def read_text_with_etag(self, key: str) -> tuple[str | None, str | None]:
+        p = self._path(key)
+        if not p.exists():
+            return None, None
+        content = p.read_text()
+        return content, _content_etag(content)
+
+    def write_text(self, key: str, content: str, *, if_match: str | None = None) -> str | None:
         p = self._path(key)
+        if if_match is not None:
+            # Conditional write: current content hash must equal if_match.
+            current = _content_etag(p.read_text()) if p.exists() else None
+            if current != if_match:
+                raise ConcurrencyError(
+                    f"write_text({key!r}) refused: If-Match={if_match!r} is stale "
+                    f"(current={current!r})"
+                )
         p.parent.mkdir(parents=True, exist_ok=True)
         p.write_text(content)
+        return _content_etag(content)
 
     def append_text(self, key: str, content: str) -> None:
         p = self._path(key)
@@ -142,8 +188,17 @@ def read_text(self, key: str) -> str | None:
             return None
         return data.decode("utf-8")
 
-    def write_text(self, key: str, content: str) -> None:
-        _run_sync(self._store.put(key, content.encode("utf-8")))
+    def read_text_with_etag(self, key: str) -> tuple[str | None, str | None]:
+        try:
+            data, etag = _run_sync(self._store.get_with_etag(key))
+        except KeyError:
+            return None, None
+        return data.decode("utf-8"), etag
+
+    def write_text(self, key: str, content: str, *, if_match: str | None = None) -> str | None:
+        return _run_sync(
+            self._store.put(key, content.encode("utf-8"), if_match=if_match or None)
+        )
 
     def append_text(self, key: str, content: str) -> None:
         existing = self.read_text(key) or ""
@@ -167,11 +222,13 @@ def get_backend() -> MemoryBackend:
       2. ``blob_endpoint`` AND ``blob_container`` set → :class:`BlobBackend`
          wrapping a :class:`BlobStore` whose token provider is the Agent
          User's storage-scope three-hop token.
-      3. Otherwise → :class:`LocalBackend` rooted at ``cfg.data_dir``.
+      3. Neither set → :class:`LocalBackend` rooted at ``cfg.data_dir``.
 
-    Half-configured cloud (endpoint without container, or vice versa) is
-    treated as not-configured rather than raising — the local fallback is
-    safer for the hot path.
+    Half-configured cloud (exactly one of endpoint/container set) is a HARD
+    error (:class:`BackendMisconfiguredError`), not a silent Local fallback
+    (design F2). A silent fallback is how a mis-enved fleet instance lands on
+    an empty local store and re-bootstraps every chat → replay flood; failing
+    loud surfaces the misconfiguration instead of letting the fleet diverge.
     """
     cfg = get_config()
     if cfg.keep_memory_local:
@@ -183,4 +240,48 @@ def get_backend() -> MemoryBackend:
             token_provider=lambda: acquire_agent_user_storage_token(get_config()),
         )
         return BlobBackend(store)
+    if cfg.blob_endpoint or cfg.blob_container:
+        raise BackendMisconfiguredError(
+            endpoint=cfg.blob_endpoint,
+            container=cfg.blob_container,
+        )
     return LocalBackend(cfg.data_dir)
+
+
+def assert_backend_config(logger=None) -> dict[str, str | None]:
+    """Resolve + validate the storage backend once at boot (design F2).
+
+    Raises :class:`BackendMisconfiguredError` on a half-configured blob env
+    (so a mis-enved instance refuses to start instead of silently using an
+    empty Local store), and logs the resolved backend + container so operators
+    can confirm fleet-wide agreement in production logs.
+
+    Returns a small summary ``{"backend": ..., "container": ..., "root": ...}``
+    for callers/tests. Never resolves the storage token — construction is
+    lazy, so this is safe to call before auth.
+    """
+    cfg = get_config()
+    backend = get_backend()  # raises on half-config
+    if isinstance(backend, BlobBackend):
+        summary: dict[str, str | None] = {
+            "backend": "BlobBackend",
+            "endpoint": cfg.blob_endpoint,
+            "container": cfg.blob_container,
+            "root": None,
+        }
+        if logger:
+            logger.info(
+                "Resolved MemoryBackend=BlobBackend endpoint=%s container=%s",
+                cfg.blob_endpoint,
+                cfg.blob_container,
+            )
+    else:
+        summary = {
+            "backend": "LocalBackend",
+            "endpoint": None,
+            "container": None,
+            "root": str(cfg.data_dir),
+        }
+        if logger:
+            logger.info("Resolved MemoryBackend=LocalBackend root=%s", cfg.data_dir)
+    return summary
diff --git a/src/entrabot/storage/blob.py b/src/entrabot/storage/blob.py
index fa6911e..786976e 100644
--- a/src/entrabot/storage/blob.py
+++ b/src/entrabot/storage/blob.py
@@ -104,6 +104,22 @@ async def get(self, path: str) -> bytes:
             resp.raise_for_status()
             return resp.content
 
+    async def get_with_etag(self, path: str) -> tuple[bytes, str]:
+        """Download *path* and its current ETag.
+
+        Returns ``(content, etag)``. Raises ``KeyError`` on 404 so callers can
+        distinguish "absent" from a transport failure. The ETag lets a caller
+        do a later ``put(..., if_match=etag)`` for optimistic concurrency
+        (fleet-safe cursor writes — design F5).
+        """
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(self._url(path), headers=self._headers())
+            _check_auth(resp)
+            if resp.status_code == 404:
+                raise KeyError(path)
+            resp.raise_for_status()
+            return resp.content, resp.headers.get("ETag", "")
+
     async def exists(self, path: str) -> bool:
         """Probe whether *path* exists. HEAD request — doesn't pull the body."""
         async with httpx.AsyncClient() as client:
diff --git a/src/entrabot/tools/chat_cursors.py b/src/entrabot/tools/chat_cursors.py
index 887bacf..5c50674 100644
--- a/src/entrabot/tools/chat_cursors.py
+++ b/src/entrabot/tools/chat_cursors.py
@@ -34,10 +34,13 @@
 import json
 import logging
 from collections.abc import Iterable
+from dataclasses import dataclass
 from datetime import UTC, datetime
+from enum import Enum
 from urllib.parse import quote
 
 from entrabot.storage.backend import get_backend
+from entrabot.storage.blob import ConcurrencyError
 
 logger = logging.getLogger("entrabot.tools.chat_cursors")
 
@@ -115,6 +118,92 @@ def load_cursor(chat_id: str) -> dict | None:
     return parsed
 
 
+class CursorOutcome(Enum):
+    """Classification of a cursor read, used to decide whether the poll may push.
+
+    The distinction is load-bearing for fleet safety (design doc F1):
+
+    * ``ABSENT`` — the read SUCCEEDED and no cursor exists. This is the only
+      case where "surface the newest message once" is allowed: a genuinely new
+      chat that no instance has ever cursor-ed.
+    * ``PRESENT`` — a cursor exists and parsed (fresh OR stale). Always
+      rehydrate; the steady-state timestamp gate does catch-up (F4). Never
+      re-bootstrap a present cursor.
+    * ``UNRESOLVED`` — the read FAILED, or the payload is corrupt / the wrong
+      shape. Ambiguous: a transient blob 401/timeout/throttle or a partial
+      write could be hiding a live cursor. **Fail closed — never push.** The
+      caller must retry the read on a later cycle before delivering anything.
+    """
+
+    ABSENT = "absent"
+    PRESENT = "present"
+    UNRESOLVED = "unresolved"
+
+
+@dataclass(frozen=True)
+class CursorResolution:
+    """Result of :func:`resolve_cursor` — an outcome plus the parsed cursor.
+
+    ``cursor`` is populated only for :attr:`CursorOutcome.PRESENT`; it is
+    ``None`` for ``ABSENT`` and ``UNRESOLVED``.
+    """
+
+    outcome: CursorOutcome
+    cursor: dict | None
+
+
+def resolve_cursor(chat_id: str) -> CursorResolution:
+    """Classify *chat_id*'s cursor read into ABSENT / PRESENT / UNRESOLVED.
+
+    Unlike :func:`load_cursor` (which collapses every miss to ``None``), this
+    keeps the three cases apart so the poll can fail closed. The rules:
+
+    * Backend read raises (transient blob/disk error, 401 refresh race,
+      throttle) → ``UNRESOLVED``. The whole point of the fleet-safety fix: an
+      ambiguous read must never be mistaken for "new chat, push newest".
+    * Read returns ``None`` (the backend positively determined the key is
+      absent) → ``ABSENT``.
+    * Content present but not valid JSON, or valid JSON that isn't an object →
+      ``UNRESOLVED``. A partial/corrupt write is ambiguous; do not treat it as
+      a clean slate and re-push.
+    * Content present and a JSON object (fresh OR stale) → ``PRESENT``.
+    """
+    backend = get_backend()
+    try:
+        raw = backend.read_text(cursor_key(chat_id))
+    except Exception as exc:  # noqa: BLE001 — any read failure is ambiguous.
+        logger.warning(
+            "Cursor read failed for %s (UNRESOLVED, failing closed): %s: %s",
+            chat_id,
+            type(exc).__name__,
+            exc,
+        )
+        return CursorResolution(CursorOutcome.UNRESOLVED, None)
+
+    if raw is None:
+        return CursorResolution(CursorOutcome.ABSENT, None)
+
+    try:
+        parsed = json.loads(raw)
+    except (ValueError, json.JSONDecodeError) as exc:
+        logger.warning(
+            "Corrupt chat cursor for %s (UNRESOLVED, failing closed): %s",
+            chat_id,
+            exc,
+        )
+        return CursorResolution(CursorOutcome.UNRESOLVED, None)
+
+    if not isinstance(parsed, dict):
+        logger.warning(
+            "Unexpected chat cursor shape for %s (UNRESOLVED, failing closed): %r",
+            chat_id,
+            type(parsed),
+        )
+        return CursorResolution(CursorOutcome.UNRESOLVED, None)
+
+    return CursorResolution(CursorOutcome.PRESENT, parsed)
+
+
 def save_cursor(chat_id: str, state: dict) -> None:
     """Persist *state* as *chat_id*'s cursor through the configured backend.
 
@@ -137,6 +226,133 @@ def save_cursor(chat_id: str, state: dict) -> None:
     backend.write_text(cursor_key(chat_id), json.dumps(payload))
 
 
+def _later_ts(a: str | None, b: str | None) -> str | None:
+    """Return the later of two ISO-8601 timestamps, ignoring ``None``.
+
+    Parses to timezone-aware datetimes so mixed sub-second precision compares
+    correctly (``"...15Z"`` vs ``"...15.261Z"``, where a naive string ``max``
+    would pick the wrong one). Falls back to string ``max`` if either value is
+    unparseable — defensive, never raises.
+    """
+    if a is None:
+        return b
+    if b is None:
+        return a
+
+    def _parse(ts: str) -> datetime | None:
+        try:
+            dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+        except (ValueError, AttributeError):
+            return None
+        return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt
+
+    da, db = _parse(a), _parse(b)
+    if da is None or db is None:
+        return max(a, b)
+    return a if da >= db else b
+
+
+# Bounded retries for the ETag compare-and-swap in :func:`claim_delivery`.
+# Each retry re-reads the shared cursor, so a handful covers realistic fleet
+# write contention; exhausting them fails closed (claim nothing → push
+# nothing) rather than risk a double delivery.
+CLAIM_MAX_ATTEMPTS = 4
+
+
+def claim_delivery(
+    chat_id: str,
+    candidate_ids: Iterable[str],
+    last_ts: str | None = None,
+) -> list[str]:
+    """Atomically claim delivery of *candidate_ids* in the shared cloud cursor.
+
+    The persisted cursor's ``seen_ids_tail`` doubles as the fleet delivery
+    ledger. This reads the shared cursor, computes which candidates are NOT
+    already recorded as delivered, writes the merged cursor
+    (``seen ∪ candidates``, ``max(last_ts)``) back with an ``If-Match``
+    precondition, and returns the newly-claimed ids — the ones THIS instance
+    should push.
+
+    Fleet guarantee (design "per-message cloud idempotency" + F5): across N
+    instances polling the same message, exactly one wins the compare-and-swap
+    for a given id; the losers hit ``ConcurrencyError``, re-read, see the id
+    already delivered, and claim nothing for it. Delivery is idempotent across
+    the whole fleet.
+
+    Fail-closed everywhere: a read failure, a corrupt shared cursor, or
+    exhausted CAS retries all return ``[]`` (claim nothing → push nothing)
+    rather than risk re-injecting a message.
+    """
+    candidates = list(candidate_ids)
+    if not candidates:
+        return []
+
+    backend = get_backend()
+    key = cursor_key(chat_id)
+
+    for _attempt in range(CLAIM_MAX_ATTEMPTS):
+        try:
+            raw, etag = backend.read_text_with_etag(key)
+        except Exception as exc:  # noqa: BLE001 — ambiguous read → fail closed.
+            logger.warning(
+                "claim_delivery read failed for %s (fail closed, no push): %s: %s",
+                chat_id,
+                type(exc).__name__,
+                exc,
+            )
+            return []
+
+        existing: dict = {}
+        if raw is not None:
+            try:
+                parsed = json.loads(raw)
+            except (ValueError, json.JSONDecodeError):
+                logger.warning(
+                    "claim_delivery: corrupt shared cursor for %s (fail closed)",
+                    chat_id,
+                )
+                return []
+            if isinstance(parsed, dict):
+                existing = parsed
+
+        seen = set(existing.get("seen_ids_tail") or [])
+        newly = [mid for mid in candidates if mid not in seen]
+        if not newly:
+            return []  # every candidate already delivered by a sibling
+
+        merged_seen = seen | set(candidates)
+        payload = {
+            "last_ts": _later_ts(existing.get("last_ts"), last_ts),
+            "seen_ids_tail": bound_seen_ids(sorted(merged_seen)),
+            "bootstrapped": True,
+            "last_written_at": _now_iso(),
+        }
+
+        try:
+            backend.write_text(key, json.dumps(payload), if_match=etag or None)
+        except ConcurrencyError:
+            logger.info(
+                "claim_delivery: cursor ETag conflict for %s; re-reading and retrying",
+                chat_id,
+            )
+            continue
+        except Exception as exc:  # noqa: BLE001 — write failure → fail closed.
+            logger.warning(
+                "claim_delivery write failed for %s (fail closed, no push): %s: %s",
+                chat_id,
+                type(exc).__name__,
+                exc,
+            )
+            return []
+        return newly
+
+    logger.warning(
+        "claim_delivery: CAS retries exhausted for %s (fail closed, no push)",
+        chat_id,
+    )
+    return []
+
+
 def is_stale(last_ts: str | None) -> bool:
     """Return True if *last_ts* is too old to safely rehydrate from.
 
diff --git a/tests/storage/test_backend.py b/tests/storage/test_backend.py
index 71f2395..0fbe78d 100644
--- a/tests/storage/test_backend.py
+++ b/tests/storage/test_backend.py
@@ -99,20 +99,35 @@ class _FakeBlobStore:
     """In-memory async BlobStore stand-in.
 
     Mirrors the real ``BlobStore`` API surface used by ``BlobBackend``:
-    async ``get`` (raises KeyError on miss), ``put``, ``exists``, ``list``.
+    async ``get`` (raises KeyError on miss), ``get_with_etag``, ``put``
+    (honors ``if_match`` for optimistic concurrency), ``exists``, ``list``.
     """
 
     def __init__(self) -> None:
         self.data: dict[str, bytes] = {}
+        self.etags: dict[str, str] = {}
+        self._counter = 0
 
     async def get(self, path: str) -> bytes:
         if path not in self.data:
             raise KeyError(path)
         return self.data[path]
 
+    async def get_with_etag(self, path: str) -> tuple[bytes, str]:
+        if path not in self.data:
+            raise KeyError(path)
+        return self.data[path], self.etags[path]
+
     async def put(self, path: str, data: bytes, *, if_match: str | None = None) -> str:
+        from entrabot.storage.blob import ConcurrencyError
+
+        if if_match is not None and self.etags.get(path) != if_match:
+            raise ConcurrencyError(f"put({path!r}) refused: If-Match={if_match!r} stale")
+        self._counter += 1
+        etag = f'"etag-{self._counter}"'
         self.data[path] = data
-        return f'"etag-{len(self.data)}"'
+        self.etags[path] = etag
+        return etag
 
     async def exists(self, path: str) -> bool:
         return path in self.data
@@ -164,6 +179,97 @@ def test_list(self) -> None:
         ]
 
 
+# ---------------------------------------------------------------------------
+# ETag optimistic concurrency (design F5) — read_text_with_etag + if_match
+# ---------------------------------------------------------------------------
+class TestLocalBackendEtag:
+    def test_read_with_etag_absent_returns_none_none(self, tmp_path: Path) -> None:
+        backend = LocalBackend(tmp_path)
+        content, etag = backend.read_text_with_etag("absent.json")
+        assert content is None
+        assert etag is None
+
+    def test_read_with_etag_present_returns_content_and_stable_etag(
+        self, tmp_path: Path
+    ) -> None:
+        backend = LocalBackend(tmp_path)
+        backend.write_text("c.json", "hello")
+        content, etag = backend.read_text_with_etag("c.json")
+        assert content == "hello"
+        assert etag is not None
+        # ETag is stable for unchanged content.
+        _, etag2 = backend.read_text_with_etag("c.json")
+        assert etag2 == etag
+
+    def test_etag_changes_when_content_changes(self, tmp_path: Path) -> None:
+        backend = LocalBackend(tmp_path)
+        backend.write_text("c.json", "v1")
+        _, e1 = backend.read_text_with_etag("c.json")
+        backend.write_text("c.json", "v2")
+        _, e2 = backend.read_text_with_etag("c.json")
+        assert e1 != e2
+
+    def test_conditional_write_succeeds_when_etag_matches(self, tmp_path: Path) -> None:
+        backend = LocalBackend(tmp_path)
+        backend.write_text("c.json", "v1")
+        _, etag = backend.read_text_with_etag("c.json")
+        new_etag = backend.write_text("c.json", "v2", if_match=etag)
+        assert backend.read_text("c.json") == "v2"
+        assert new_etag is not None and new_etag != etag
+
+    def test_conditional_write_raises_on_stale_etag(self, tmp_path: Path) -> None:
+        from entrabot.storage.blob import ConcurrencyError
+
+        backend = LocalBackend(tmp_path)
+        backend.write_text("c.json", "v1")
+        _, etag = backend.read_text_with_etag("c.json")
+        # A concurrent writer bumps the content out from under us.
+        backend.write_text("c.json", "raced")
+        with pytest.raises(ConcurrencyError):
+            backend.write_text("c.json", "v2", if_match=etag)
+        # The racing write survives; our stale write did not land.
+        assert backend.read_text("c.json") == "raced"
+
+    def test_conditional_write_on_absent_with_etag_raises(self, tmp_path: Path) -> None:
+        from entrabot.storage.blob import ConcurrencyError
+
+        backend = LocalBackend(tmp_path)
+        with pytest.raises(ConcurrencyError):
+            backend.write_text("c.json", "v1", if_match="some-etag")
+
+    def test_unconditional_write_ignores_etag(self, tmp_path: Path) -> None:
+        backend = LocalBackend(tmp_path)
+        backend.write_text("c.json", "v1")
+        # if_match=None → unconditional, always writes.
+        backend.write_text("c.json", "v2", if_match=None)
+        assert backend.read_text("c.json") == "v2"
+
+
+class TestBlobBackendEtag:
+    def test_read_with_etag_absent_returns_none_none(self) -> None:
+        backend = BlobBackend(_FakeBlobStore())
+        assert backend.read_text_with_etag("absent") == (None, None)
+
+    def test_read_with_etag_present_returns_content_and_etag(self) -> None:
+        store = _FakeBlobStore()
+        backend = BlobBackend(store)
+        backend.write_text("c.json", "hello")
+        content, etag = backend.read_text_with_etag("c.json")
+        assert content == "hello"
+        assert etag
+
+    def test_conditional_write_raises_concurrency_on_stale(self) -> None:
+        from entrabot.storage.blob import ConcurrencyError
+
+        store = _FakeBlobStore()
+        backend = BlobBackend(store)
+        backend.write_text("c.json", "v1")
+        _, etag = backend.read_text_with_etag("c.json")
+        backend.write_text("c.json", "raced")  # bumps etag
+        with pytest.raises(ConcurrencyError):
+            backend.write_text("c.json", "v2", if_match=etag)
+
+
 # ---------------------------------------------------------------------------
 # get_backend factory
 # ---------------------------------------------------------------------------
@@ -205,16 +311,46 @@ def test_blob_endpoint_set_returns_blob_backend(
         backend = get_backend()
         assert isinstance(backend, BlobBackend)
 
-    def test_blob_endpoint_without_container_falls_back_local(
+    def test_half_configured_blob_endpoint_without_container_raises(
         self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        """Half-configured cloud (endpoint without container) falls back
-        to Local — better safe than panicking inside the hot path.
+        """F2: half-configured cloud (endpoint without container) is a HARD
+        misconfiguration, not a silent Local fallback.
+
+        A silent fallback is exactly how a mis-enved fleet instance ends up on
+        an empty Local store and re-bootstraps every chat → replay flood. Fail
+        loud so the operator sees it instead of the fleet diverging.
         """
+        from entrabot.errors import BackendMisconfiguredError
+
         monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
         monkeypatch.setenv("ENTRABOT_BLOB_ENDPOINT", "https://entclaw.blob.core.windows.net")
         monkeypatch.delenv("ENTRABOT_BLOB_CONTAINER", raising=False)
         monkeypatch.delenv("ENTRABOT_KEEP_MEMORY_LOCAL", raising=False)
+        with pytest.raises(BackendMisconfiguredError):
+            get_backend()
+
+    def test_half_configured_blob_container_without_endpoint_raises(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from entrabot.errors import BackendMisconfiguredError
+
+        monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
+        monkeypatch.delenv("ENTRABOT_BLOB_ENDPOINT", raising=False)
+        monkeypatch.setenv("ENTRABOT_BLOB_CONTAINER", "agent-abc-123")
+        monkeypatch.delenv("ENTRABOT_KEEP_MEMORY_LOCAL", raising=False)
+        with pytest.raises(BackendMisconfiguredError):
+            get_backend()
+
+    def test_keep_memory_local_suppresses_half_config_error(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """The explicit local escape hatch wins even over a half-config, so an
+        operator can always force Local without untangling blob env first."""
+        monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
+        monkeypatch.setenv("ENTRABOT_BLOB_ENDPOINT", "https://entclaw.blob.core.windows.net")
+        monkeypatch.delenv("ENTRABOT_BLOB_CONTAINER", raising=False)
+        monkeypatch.setenv("ENTRABOT_KEEP_MEMORY_LOCAL", "true")
         assert isinstance(get_backend(), LocalBackend)
 
     def test_keep_memory_local_overrides_blob_endpoint(
@@ -226,3 +362,49 @@ def test_keep_memory_local_overrides_blob_endpoint(
         monkeypatch.setenv("ENTRABOT_BLOB_CONTAINER", "agent-abc-123")
         monkeypatch.setenv("ENTRABOT_KEEP_MEMORY_LOCAL", "true")
         assert isinstance(get_backend(), LocalBackend)
+
+
+# ---------------------------------------------------------------------------
+# assert_backend_config — boot-time uniform-backend assertion (F2)
+# ---------------------------------------------------------------------------
+class TestAssertBackendConfig:
+    def test_returns_local_summary(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from entrabot.storage.backend import assert_backend_config
+
+        monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
+        monkeypatch.delenv("ENTRABOT_BLOB_ENDPOINT", raising=False)
+        monkeypatch.delenv("ENTRABOT_BLOB_CONTAINER", raising=False)
+        summary = assert_backend_config()
+        assert summary["backend"] == "LocalBackend"
+
+    def test_returns_blob_summary_with_container(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from entrabot.storage.backend import assert_backend_config
+
+        monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
+        monkeypatch.setenv("ENTRABOT_BLOB_ENDPOINT", "https://entclaw.blob.core.windows.net")
+        monkeypatch.setenv("ENTRABOT_BLOB_CONTAINER", "agent-abc-123")
+        monkeypatch.delenv("ENTRABOT_KEEP_MEMORY_LOCAL", raising=False)
+        monkeypatch.setattr(
+            "entrabot.storage.backend.acquire_agent_user_storage_token",
+            lambda cfg: "fake-storage-token",
+        )
+        summary = assert_backend_config()
+        assert summary["backend"] == "BlobBackend"
+        assert summary["container"] == "agent-abc-123"
+
+    def test_raises_on_half_configured_blob(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from entrabot.errors import BackendMisconfiguredError
+        from entrabot.storage.backend import assert_backend_config
+
+        monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
+        monkeypatch.setenv("ENTRABOT_BLOB_ENDPOINT", "https://entclaw.blob.core.windows.net")
+        monkeypatch.delenv("ENTRABOT_BLOB_CONTAINER", raising=False)
+        monkeypatch.delenv("ENTRABOT_KEEP_MEMORY_LOCAL", raising=False)
+        with pytest.raises(BackendMisconfiguredError):
+            assert_backend_config()
diff --git a/tests/storage/test_blob.py b/tests/storage/test_blob.py
index 6bdbd1b..4ba7506 100644
--- a/tests/storage/test_blob.py
+++ b/tests/storage/test_blob.py
@@ -90,6 +90,45 @@ async def test_get_sends_bearer_token(self) -> None:
             assert route.calls.last.request.headers["authorization"] == "Bearer stub-token"
 
 
+class TestBlobGetWithEtag:
+    @pytest.mark.asyncio
+    async def test_get_with_etag_returns_bytes_and_etag(self) -> None:
+        """get_with_etag returns (content, ETag) so the caller can do a later
+        If-Match conditional write (optimistic concurrency for cursors)."""
+        with respx.mock:
+            respx.get(f"{BLOB_URL}/chat_cursors/c.json").mock(
+                return_value=httpx.Response(
+                    200,
+                    content=b'{"last_ts": "t"}',
+                    headers={"ETag": '"0xETAG1"'},
+                ),
+            )
+            store = _make_store()
+            data, etag = await store.get_with_etag("chat_cursors/c.json")
+            assert data == b'{"last_ts": "t"}'
+            assert etag == '"0xETAG1"'
+
+    @pytest.mark.asyncio
+    async def test_get_with_etag_raises_keyerror_on_404(self) -> None:
+        with respx.mock:
+            respx.get(f"{BLOB_URL}/missing.json").mock(
+                return_value=httpx.Response(404)
+            )
+            store = _make_store()
+            with pytest.raises(KeyError):
+                await store.get_with_etag("missing.json")
+
+    @pytest.mark.asyncio
+    async def test_get_with_etag_401_raises_token_expired(self) -> None:
+        from entrabot.errors import TokenExpiredError
+
+        with respx.mock:
+            respx.get(f"{BLOB_URL}/x.json").mock(return_value=httpx.Response(401))
+            store = _make_store()
+            with pytest.raises(TokenExpiredError):
+                await store.get_with_etag("x.json")
+
+
 class TestBlobExists:
     @pytest.mark.asyncio
     async def test_exists_true_on_200(self) -> None:
diff --git a/tests/test_mcp_server_chat_cursors.py b/tests/test_mcp_server_chat_cursors.py
index 6d78fd3..237ae3a 100644
--- a/tests/test_mcp_server_chat_cursors.py
+++ b/tests/test_mcp_server_chat_cursors.py
@@ -71,13 +71,15 @@ def _isolate_state(tmp_path, monkeypatch):
 # ---------------------------------------------------------------------------
 class TestRehydrateOnRegister:
     def test_no_cursor_present_leaves_state_unbootstrapped(self, tmp_path) -> None:
-        """First-time registration → no cursor file → falls through to bootstrap."""
+        """ABSENT (successful empty read) → new chat, may bootstrap-push once."""
         mcp_server._register_watched_chat("19:fresh@thread.v2", persist=False)
         state = mcp_server._state["watched_chats"]["19:fresh@thread.v2"]
-        # Default fresh state: not bootstrapped, empty seen, no last_ts.
+        # Default fresh state: not bootstrapped, empty seen, no last_ts, and NOT
+        # flagged for re-resolution (it resolved cleanly to ABSENT).
         assert state["bootstrapped"] is False
         assert state["last_ts"] is None
         assert state["seen_ids"] == set()
+        assert state.get("needs_resolution") is False
 
     def test_fresh_cursor_present_rehydrates_and_skips_bootstrap(
         self, tmp_path
@@ -102,8 +104,14 @@ def test_fresh_cursor_present_rehydrates_and_skips_bootstrap(
         assert state["last_ts"] == recent
         assert state["seen_ids"] == {"msg-a", "msg-b"}
 
-    def test_stale_cursor_falls_through_to_bootstrap(self, tmp_path) -> None:
-        """A cursor older than the staleness cap → ignore and bootstrap."""
+    def test_stale_cursor_rehydrates_for_catch_up(self, tmp_path) -> None:
+        """F4: a stale-but-present cursor rehydrates (NOT re-bootstrap).
+
+        The steady-state timestamp gate does catch-up (surfaces only messages
+        newer than ``last_ts``); an idle chat with no new messages emits
+        nothing. Re-bootstrapping a stale cursor is exactly the replay-flood
+        bug this fix removes.
+        """
         old = (
             datetime.now(UTC)
             - timedelta(seconds=chat_cursors.CURSOR_STALENESS_SECONDS + 3600)
@@ -120,13 +128,19 @@ def test_stale_cursor_falls_through_to_bootstrap(self, tmp_path) -> None:
         mcp_server._register_watched_chat("19:stale@thread.v2", persist=False)
 
         state = mcp_server._state["watched_chats"]["19:stale@thread.v2"]
-        # Stale → treat as fresh registration; bootstrap will re-baseline.
-        assert state["bootstrapped"] is False
-        assert state["last_ts"] is None
-        assert state["seen_ids"] == set()
+        # Present cursor → rehydrate + catch up; never re-baseline-and-push.
+        assert state["bootstrapped"] is True
+        assert state["last_ts"] == old
+        assert state["seen_ids"] == {"msg-a"}
+        assert state.get("needs_resolution") is False
+
+    def test_corrupt_cursor_marks_needs_resolution(self, tmp_path) -> None:
+        """F1: corrupt JSON is ambiguous → fail closed, do NOT bootstrap-push.
 
-    def test_corrupt_cursor_falls_through_to_bootstrap(self, tmp_path) -> None:
-        """Corrupt JSON → treat as absent (defensive: boot must not die)."""
+        A partial write could be hiding a live cursor. The chat is flagged
+        ``needs_resolution`` so the poll loop re-reads before delivering
+        anything — it must never surface the newest message on a corrupt read.
+        """
         from entrabot.storage.backend import get_backend
 
         get_backend().write_text(
@@ -139,6 +153,218 @@ def test_corrupt_cursor_falls_through_to_bootstrap(self, tmp_path) -> None:
         state = mcp_server._state["watched_chats"]["19:corrupt@thread.v2"]
         assert state["bootstrapped"] is False
         assert state["last_ts"] is None
+        assert state["needs_resolution"] is True
+
+    def test_read_error_marks_needs_resolution(self, tmp_path, monkeypatch) -> None:
+        """F1 core: a transient read failure must fail closed, never bootstrap.
+
+        This is the exact fleet-replay trigger — a blob 401/timeout/throttle at
+        registration used to fall through to ``_bootstrap_chat`` and push the
+        newest (weeks-old) message. Now it flags ``needs_resolution``.
+        """
+        import entrabot.tools.chat_cursors as cc
+
+        class ExplodingBackend:
+            def read_text(self, key: str) -> str | None:
+                raise OSError("simulated transient blob read failure")
+
+        monkeypatch.setattr(cc, "get_backend", lambda: ExplodingBackend())
+
+        mcp_server._register_watched_chat("19:boom@thread.v2", persist=False)
+
+        state = mcp_server._state["watched_chats"]["19:boom@thread.v2"]
+        assert state["bootstrapped"] is False
+        assert state["needs_resolution"] is True
+
+
+# ---------------------------------------------------------------------------
+# Poll loop — fail-closed handling of a needs_resolution chat (F1)
+# ---------------------------------------------------------------------------
+class TestPollFailClosed:
+    """The per-chat poll step must never push while a chat is unresolved.
+
+    A chat flagged ``needs_resolution`` (transient read failure or corrupt
+    cursor at registration) must re-read the cursor and, until it resolves,
+    push NOTHING and NOT bootstrap. This is the security-relevant fail-closed
+    guarantee: an ambiguous read never re-injects stale messages.
+    """
+
+    @pytest.mark.asyncio
+    async def test_unresolved_chat_does_not_push_or_bootstrap(
+        self, monkeypatch
+    ) -> None:
+        chat_id = "19:unresolved@thread.v2"
+        chat_state = {
+            "seen_ids": set(),
+            "last_ts": None,
+            "bootstrapped": False,
+            "needs_resolution": True,
+        }
+        mcp_server._state["watched_chats"] = {chat_id: chat_state}
+
+        # Cursor read still failing → resolve_cursor returns UNRESOLVED.
+        import entrabot.tools.chat_cursors as cc
+
+        monkeypatch.setattr(
+            cc,
+            "resolve_cursor",
+            lambda cid: cc.CursorResolution(cc.CursorOutcome.UNRESOLVED, None),
+        )
+
+        pushed: list = []
+        bootstrapped: list = []
+        monkeypatch.setattr(
+            mcp_server,
+            "_push_channel_notification",
+            _async_recorder(pushed),
+        )
+        monkeypatch.setattr(mcp_server, "_bootstrap_chat", _async_recorder(bootstrapped))
+
+        await mcp_server._poll_watched_chat(chat_id, chat_state, "EntraBot Agent")
+
+        assert pushed == []
+        assert bootstrapped == []
+        # Still unresolved → stays flagged for retry next cycle.
+        assert chat_state["needs_resolution"] is True
+        assert chat_state["bootstrapped"] is False
+
+    @pytest.mark.asyncio
+    async def test_unresolved_chat_rehydrates_when_read_recovers(
+        self, monkeypatch
+    ) -> None:
+        chat_id = "19:recovers@thread.v2"
+        chat_state = {
+            "seen_ids": set(),
+            "last_ts": None,
+            "bootstrapped": False,
+            "needs_resolution": True,
+        }
+        mcp_server._state["watched_chats"] = {chat_id: chat_state}
+
+        recovered = {
+            "last_ts": "2026-06-09T18:00:00Z",
+            "seen_ids_tail": ["m1", "m2"],
+            "bootstrapped": True,
+        }
+        import entrabot.tools.chat_cursors as cc
+
+        monkeypatch.setattr(
+            cc,
+            "resolve_cursor",
+            lambda cid: cc.CursorResolution(cc.CursorOutcome.PRESENT, recovered),
+        )
+
+        pushed: list = []
+        bootstrapped: list = []
+        monkeypatch.setattr(
+            mcp_server, "_push_channel_notification", _async_recorder(pushed)
+        )
+        monkeypatch.setattr(mcp_server, "_bootstrap_chat", _async_recorder(bootstrapped))
+
+        await mcp_server._poll_watched_chat(chat_id, chat_state, "EntraBot Agent")
+
+        # Resolution cycle rehydrates but does NOT push (fail-closed): the
+        # steady-state gate handles delivery on the NEXT cycle.
+        assert pushed == []
+        assert bootstrapped == []
+        assert chat_state["needs_resolution"] is False
+        assert chat_state["bootstrapped"] is True
+        assert chat_state["last_ts"] == "2026-06-09T18:00:00Z"
+        assert chat_state["seen_ids"] == {"m1", "m2"}
+
+
+def _async_recorder(sink: list):
+    async def _rec(*args, **kwargs):
+        sink.append((args, kwargs))
+
+    return _rec
+
+
+# ---------------------------------------------------------------------------
+# Poll loop — steady-state pushes only fleet-claimed messages (idempotency)
+# ---------------------------------------------------------------------------
+class TestPollSteadyStateIdempotent:
+    """Steady-state delivery routes through ``claim_delivery`` so only the
+    messages THIS instance wins in the shared ledger are pushed."""
+
+    @pytest.mark.asyncio
+    async def test_pushes_only_claimed_messages(self, monkeypatch) -> None:
+        chat_id = "19:steady@thread.v2"
+        chat_state = {
+            "seen_ids": set(),
+            "last_ts": "2026-06-09T18:00:00Z",
+            "bootstrapped": True,
+            "needs_resolution": False,
+        }
+        mcp_server._state["watched_chats"] = {chat_id: chat_state}
+
+        msgs = [
+            {"message_id": "m1", "sent_at": "2026-06-09T18:05:00Z", "text": "one"},
+            {"message_id": "m2", "sent_at": "2026-06-09T18:06:00Z", "text": "two"},
+        ]
+
+        async def fake_token_retry(fn, **kwargs):
+            return msgs
+
+        monkeypatch.setattr(mcp_server, "_with_token_retry", fake_token_retry)
+        monkeypatch.setattr(
+            "entrabot.tools.teams.filter_human_messages",
+            lambda messages, name: messages,
+        )
+
+        # Sibling already delivered m2 → only m1 is ours to push.
+        import entrabot.tools.chat_cursors as cc
+
+        monkeypatch.setattr(cc, "claim_delivery", lambda cid, ids, ts: ["m1"])
+
+        pushed: list = []
+        monkeypatch.setattr(
+            mcp_server, "_push_channel_notification", _async_recorder(pushed)
+        )
+
+        await mcp_server._poll_watched_chat(chat_id, chat_state, "EntraBot Agent")
+
+        pushed_ids = [call[0][0]["message_id"] for call in pushed]
+        assert pushed_ids == ["m1"]
+        # Both candidates are marked seen in-memory (handled this cycle),
+        # regardless of who pushed.
+        assert chat_state["seen_ids"] == {"m1", "m2"}
+        assert chat_state["last_ts"] == "2026-06-09T18:06:00Z"
+
+    @pytest.mark.asyncio
+    async def test_claim_returns_empty_pushes_nothing(self, monkeypatch) -> None:
+        chat_id = "19:steady2@thread.v2"
+        chat_state = {
+            "seen_ids": set(),
+            "last_ts": "2026-06-09T18:00:00Z",
+            "bootstrapped": True,
+            "needs_resolution": False,
+        }
+        mcp_server._state["watched_chats"] = {chat_id: chat_state}
+
+        msgs = [{"message_id": "m1", "sent_at": "2026-06-09T18:05:00Z"}]
+
+        async def fake_token_retry(fn, **kwargs):
+            return msgs
+
+        monkeypatch.setattr(mcp_server, "_with_token_retry", fake_token_retry)
+        monkeypatch.setattr(
+            "entrabot.tools.teams.filter_human_messages",
+            lambda messages, name: messages,
+        )
+        # Everything already claimed by a sibling (or CAS failed closed).
+        import entrabot.tools.chat_cursors as cc
+
+        monkeypatch.setattr(cc, "claim_delivery", lambda cid, ids, ts: [])
+
+        pushed: list = []
+        monkeypatch.setattr(
+            mcp_server, "_push_channel_notification", _async_recorder(pushed)
+        )
+
+        await mcp_server._poll_watched_chat(chat_id, chat_state, "EntraBot Agent")
+
+        assert pushed == []
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/test_mcp_server_integration.py b/tests/test_mcp_server_integration.py
index 143cf6e..e067bac 100644
--- a/tests/test_mcp_server_integration.py
+++ b/tests/test_mcp_server_integration.py
@@ -2017,11 +2017,20 @@ class TestBackgroundPollPerChatResilience:
     from being polled and pushing notifications."""
 
     @pytest.mark.asyncio
-    async def test_one_chat_403_does_not_starve_others(self, monkeypatch) -> None:
+    async def test_one_chat_403_does_not_starve_others(
+        self, monkeypatch, tmp_path
+    ) -> None:
         import asyncio as _asyncio
 
         from entrabot import mcp_server
 
+        # Isolate the storage backend to a tmp dir so claim_delivery's cursor
+        # persistence doesn't leak into (or read from) the real data dir.
+        monkeypatch.setenv("ENTRABOT_DATA_DIR", str(tmp_path))
+        monkeypatch.delenv("ENTRABOT_BLOB_ENDPOINT", raising=False)
+        monkeypatch.delenv("ENTRABOT_BLOB_CONTAINER", raising=False)
+        monkeypatch.delenv("ENTRABOT_KEEP_MEMORY_LOCAL", raising=False)
+
         bad_chat = "19:bad@thread.v2"
         good_chat = "19:good@thread.v2"
 
diff --git a/tests/tools/test_chat_cursors.py b/tests/tools/test_chat_cursors.py
index 652810f..bf7e544 100644
--- a/tests/tools/test_chat_cursors.py
+++ b/tests/tools/test_chat_cursors.py
@@ -26,10 +26,12 @@
 from entrabot.tools.chat_cursors import (
     CURSOR_STALENESS_SECONDS,
     MAX_SEEN_IDS_TAIL,
+    CursorOutcome,
     bound_seen_ids,
     cursor_key,
     is_stale,
     load_cursor,
+    resolve_cursor,
     save_cursor,
 )
 
@@ -234,8 +236,195 @@ def test_staleness_cap_is_24_hours(self) -> None:
 
 
 # ---------------------------------------------------------------------------
-# Storage key shape — confirm the on-disk layout matches the issue spec
+# resolve_cursor — fail-closed classification (F1)
 # ---------------------------------------------------------------------------
+class TestResolveCursor:
+    """resolve_cursor must distinguish the three read outcomes that decide
+    whether the poll may push:
+
+    * ABSENT      — read SUCCEEDED and no cursor exists → genuinely new chat,
+                    may surface newest message once.
+    * PRESENT     — cursor exists & parsed (fresh OR stale) → rehydrate; the
+                    steady-state timestamp gate does catch-up (F4).
+    * UNRESOLVED  — read FAILED / corrupt / ambiguous → fail closed, NEVER push.
+
+    Collapsing UNRESOLVED into ABSENT is exactly the fleet-replay bug: a
+    transient blob read failure must not be read as "new chat, push newest".
+    """
+
+    def test_absent_cursor_is_absent_outcome(self, tmp_data_dir) -> None:
+        res = resolve_cursor("19:absent@thread.v2")
+        assert res.outcome is CursorOutcome.ABSENT
+        assert res.cursor is None
+
+    def test_fresh_cursor_is_present_outcome(self, tmp_data_dir) -> None:
+        recent = (datetime.now(UTC) - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
+        save_cursor(
+            "19:fresh@thread.v2",
+            {"last_ts": recent, "seen_ids_tail": ["m1"], "bootstrapped": True},
+        )
+        res = resolve_cursor("19:fresh@thread.v2")
+        assert res.outcome is CursorOutcome.PRESENT
+        assert res.cursor is not None
+        assert res.cursor["last_ts"] == recent
+
+    def test_stale_cursor_is_present_not_absent(self, tmp_data_dir) -> None:
+        # F4: a stale-but-present cursor is still PRESENT. It must rehydrate and
+        # let the timestamp gate catch up — NOT re-bootstrap and re-push.
+        old = (
+            datetime.now(UTC) - timedelta(seconds=CURSOR_STALENESS_SECONDS + 3600)
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
+        save_cursor(
+            "19:stale@thread.v2",
+            {"last_ts": old, "seen_ids_tail": ["m1"], "bootstrapped": True},
+        )
+        res = resolve_cursor("19:stale@thread.v2")
+        assert res.outcome is CursorOutcome.PRESENT
+        assert res.cursor is not None
+        assert res.cursor["last_ts"] == old
+
+    def test_corrupt_json_is_unresolved_not_absent(self, tmp_data_dir) -> None:
+        from entrabot.storage.backend import get_backend
+
+        get_backend().write_text(cursor_key("19:corrupt@thread.v2"), "{not-json")
+        res = resolve_cursor("19:corrupt@thread.v2")
+        # Corrupt is ambiguous — a partial write could hide a live cursor.
+        # Fail closed: UNRESOLVED, never ABSENT.
+        assert res.outcome is CursorOutcome.UNRESOLVED
+        assert res.cursor is None
+
+    def test_non_dict_json_is_unresolved(self, tmp_data_dir) -> None:
+        from entrabot.storage.backend import get_backend
+
+        get_backend().write_text(cursor_key("19:list@thread.v2"), "[1, 2, 3]")
+        res = resolve_cursor("19:list@thread.v2")
+        assert res.outcome is CursorOutcome.UNRESOLVED
+        assert res.cursor is None
+
+    def test_backend_read_error_is_unresolved(self, tmp_data_dir, monkeypatch) -> None:
+        # A transient backend read failure (401 refresh race, timeout, throttle)
+        # must be UNRESOLVED so the caller fails closed — this is the core F1 fix.
+        import entrabot.tools.chat_cursors as cc
+
+        class ExplodingBackend:
+            def read_text(self, key: str) -> str | None:
+                raise OSError("simulated transient blob read failure")
+
+        monkeypatch.setattr(cc, "get_backend", lambda: ExplodingBackend())
+        res = resolve_cursor("19:boom@thread.v2")
+        assert res.outcome is CursorOutcome.UNRESOLVED
+        assert res.cursor is None
+
+
+
+# ---------------------------------------------------------------------------
+# claim_delivery — fleet idempotency + optimistic-concurrency merge
+# ---------------------------------------------------------------------------
+class TestClaimDelivery:
+    """The shared cursor's ``seen_ids_tail`` is the fleet delivery ledger.
+
+    ``claim_delivery`` atomically records a batch of candidate message ids in
+    the shared cloud cursor and returns only the ids THIS instance won — the
+    ones it should push. Across N instances one wins each id; the rest see it
+    already delivered and push nothing.
+    """
+
+    def test_claims_all_ids_on_empty_store(self, tmp_data_dir) -> None:
+        claimed = resolve_claim("19:c@thread.v2", ["m1", "m2"], "2026-06-09T18:00:00Z")
+        assert set(claimed) == {"m1", "m2"}
+        cur = load_cursor("19:c@thread.v2")
+        assert cur is not None
+        assert set(cur["seen_ids_tail"]) == {"m1", "m2"}
+        assert cur["last_ts"] == "2026-06-09T18:00:00Z"
+
+    def test_already_delivered_ids_are_not_reclaimed(self, tmp_data_dir) -> None:
+        resolve_claim("19:c@thread.v2", ["m1"], "2026-06-09T18:00:00Z")
+        claimed = resolve_claim("19:c@thread.v2", ["m1", "m2"], "2026-06-09T18:05:00Z")
+        assert claimed == ["m2"]
+
+    def test_empty_candidates_returns_empty_without_write(self, tmp_data_dir) -> None:
+        assert resolve_claim("19:c@thread.v2", [], None) == []
+        assert load_cursor("19:c@thread.v2") is None
+
+    def test_fleet_two_instances_one_push(self, tmp_data_dir, monkeypatch) -> None:
+        """Two instances sharing one store poll the same new message → exactly
+        one of them is told to push it."""
+        import entrabot.tools.chat_cursors as cc
+        from entrabot.storage.backend import BlobBackend
+        from tests.storage.test_backend import _FakeBlobStore
+
+        shared = _FakeBlobStore()
+        monkeypatch.setattr(cc, "get_backend", lambda: BlobBackend(shared))
+
+        a = cc.claim_delivery("19:c@thread.v2", ["m1"], "2026-06-09T18:00:00Z")
+        b = cc.claim_delivery("19:c@thread.v2", ["m1"], "2026-06-09T18:00:00Z")
+
+        assert sorted([*a, *b]) == ["m1"]
+
+    def test_concurrency_merge_keeps_max_ts_and_union(
+        self, tmp_data_dir, monkeypatch
+    ) -> None:
+        """A racing writer trips one 412; the retry re-reads and merges: the
+        final cursor keeps the max watermark and the union of delivered ids."""
+        import entrabot.tools.chat_cursors as cc
+        from entrabot.storage.blob import ConcurrencyError
+
+        class RacingBackend:
+            def __init__(self) -> None:
+                self.tripped = False
+                self.store: dict[str, str] = {}
+                self.etags: dict[str, str] = {}
+                self._n = 0
+
+            def read_text_with_etag(self, key: str):
+                return self.store.get(key), self.etags.get(key)
+
+            def write_text(self, key, content, *, if_match=None):
+                if not self.tripped:
+                    self.tripped = True
+                    self.store[key] = (
+                        '{"last_ts": "2026-06-09T20:00:00Z", '
+                        '"seen_ids_tail": ["sibling"], "bootstrapped": true}'
+                    )
+                    self.etags[key] = "sib-etag"
+                    raise ConcurrencyError("simulated 412")
+                if if_match is not None and self.etags.get(key) != if_match:
+                    raise ConcurrencyError("stale")
+                self._n += 1
+                self.store[key] = content
+                self.etags[key] = f"etag-{self._n}"
+                return self.etags[key]
+
+        backend = RacingBackend()
+        monkeypatch.setattr(cc, "get_backend", lambda: backend)
+
+        claimed = cc.claim_delivery("19:c@thread.v2", ["mine"], "2026-06-09T18:00:00Z")
+        assert claimed == ["mine"]
+
+        import json as _json
+
+        final = _json.loads(backend.store["chat_cursors/19%3Ac%40thread.v2.json"])
+        assert final["last_ts"] == "2026-06-09T20:00:00Z"
+        assert set(final["seen_ids_tail"]) == {"sibling", "mine"}
+
+    def test_read_error_fails_closed(self, tmp_data_dir, monkeypatch) -> None:
+        """A backend read failure must claim nothing (fail closed — no push)."""
+        import entrabot.tools.chat_cursors as cc
+
+        class ExplodingBackend:
+            def read_text_with_etag(self, key: str):
+                raise OSError("transient read failure")
+
+        monkeypatch.setattr(cc, "get_backend", lambda: ExplodingBackend())
+        assert cc.claim_delivery("19:c@thread.v2", ["m1"], "t") == []
+
+
+def resolve_claim(chat_id, ids, last_ts):
+    from entrabot.tools.chat_cursors import claim_delivery
+
+    return claim_delivery(chat_id, ids, last_ts)
+
+
 class TestOnDiskLayout:
     def test_cursor_written_under_chat_cursors_prefix(self, tmp_data_dir) -> None:
         save_cursor(