From c8e47042b4e66ea4a2b2b29f34b6709b4ad30d10 Mon Sep 17 00:00:00 2001 From: Brandon Werner Date: Thu, 2 Jul 2026 09:14:15 -0700 Subject: [PATCH 1/4] fix(poll): fail closed on ambiguous cursor read; rehydrate stale cursors (F1+F4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Split the cursor-read miss cases so the background Teams poll never re-pushes a chat's newest message on an ambiguous read: - Add chat_cursors.resolve_cursor() → PRESENT / ABSENT / UNRESOLVED. A backend read error or corrupt payload is UNRESOLVED (fail closed), distinct from a successful empty read (ABSENT). - _register_watched_chat classifies via resolve_cursor: PRESENT rehydrates (fresh OR stale — F4, catch-up via the timestamp gate), ABSENT registers a genuinely-new chat, UNRESOLVED flags needs_resolution and pushes nothing. - Extract _poll_watched_chat + _resolve_pending_chat: a needs_resolution chat re-reads the cursor each cycle and delivers nothing until it resolves. Closes F1 and F4 from the multi-instance cursor design. Stops the fleet replay flood + stale-imperative injection surface from transient reads and the 24h staleness cap. --- src/entrabot/mcp_server.py | 224 +++++++++++++++++--------- src/entrabot/tools/chat_cursors.py | 88 ++++++++++ tests/test_mcp_server_chat_cursors.py | 159 ++++++++++++++++-- tests/tools/test_chat_cursors.py | 83 +++++++++- 4 files changed, 469 insertions(+), 85 deletions(-) diff --git a/src/entrabot/mcp_server.py b/src/entrabot/mcp_server.py index 5d8057f..6c9895d 100644 --- a/src/entrabot/mcp_server.py +++ b/src/entrabot/mcp_server.py @@ -1208,45 +1208,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 +1339,111 @@ 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 + + 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) + + async def _background_poll() -> None: """Background polling loop — pushes inbound Teams messages to Claude Code. @@ -1344,8 +1463,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 +1489,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/tools/chat_cursors.py b/src/entrabot/tools/chat_cursors.py index 887bacf..7dff1a9 100644 --- a/src/entrabot/tools/chat_cursors.py +++ b/src/entrabot/tools/chat_cursors.py @@ -34,7 +34,9 @@ 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 @@ -115,6 +117,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. diff --git a/tests/test_mcp_server_chat_cursors.py b/tests/test_mcp_server_chat_cursors.py index 6d78fd3..0c2725e 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,131 @@ 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 # --------------------------------------------------------------------------- diff --git a/tests/tools/test_chat_cursors.py b/tests/tools/test_chat_cursors.py index 652810f..e07e648 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,87 @@ 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 + + + class TestOnDiskLayout: def test_cursor_written_under_chat_cursors_prefix(self, tmp_data_dir) -> None: save_cursor( From 34ac4b98f6fd0fbff928660e4e67549342e5e298 Mon Sep 17 00:00:00 2001 From: Brandon Werner Date: Thu, 2 Jul 2026 09:24:52 -0700 Subject: [PATCH 2/4] fix(poll): fleet idempotency ledger + optimistic-concurrency cursor writes (F5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make cross-instance Teams delivery exactly-once and cursor writes conflict-safe: - BlobStore.get_with_etag(); MemoryBackend gains read_text_with_etag() and write_text(if_match=) — Local uses a content-hash ETag, Blob uses the real ETag; a stale If-Match raises ConcurrencyError. - chat_cursors.claim_delivery(): the shared cursor seen_ids_tail is the fleet delivery ledger. CAS-merges (union seen, max last_ts) with If-Match and returns only the message ids THIS instance won; losers re-read and claim nothing. Read error / corrupt cursor / exhausted retries all fail closed. - Steady-state poll pushes only claimed ids and persists via the CAS write, so a slow writer can't regress the watermark. New-chat first delivery now also flows through claim_delivery (bootstrap defers the push a cycle). Closes the fleet idempotency + concurrency-merge items from the design. --- src/entrabot/mcp_server.py | 27 ++++-- src/entrabot/storage/backend.py | 66 +++++++++++-- src/entrabot/storage/blob.py | 16 ++++ src/entrabot/tools/chat_cursors.py | 128 ++++++++++++++++++++++++++ tests/storage/test_backend.py | 110 +++++++++++++++++++++- tests/storage/test_blob.py | 39 ++++++++ tests/test_mcp_server_chat_cursors.py | 87 +++++++++++++++++ tests/tools/test_chat_cursors.py | 108 ++++++++++++++++++++++ 8 files changed, 564 insertions(+), 17 deletions(-) diff --git a/src/entrabot/mcp_server.py b/src/entrabot/mcp_server.py index 6c9895d..cbb6cfd 100644 --- a/src/entrabot/mcp_server.py +++ b/src/entrabot/mcp_server.py @@ -1425,23 +1425,32 @@ async def _poll_watched_chat( 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 m in new_msgs: - chat_state["seen_ids"].add(m["message_id"]) + 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", "")): - 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) + if m["message_id"] in claimed: + await _push_channel_notification(m, chat_id=chat_id) async def _background_poll() -> None: diff --git a/src/entrabot/storage/backend.py b/src/entrabot/storage/backend.py index 4c0ac38..10e81d5 100644 --- a/src/entrabot/storage/backend.py +++ b/src/entrabot/storage/backend.py @@ -20,18 +20,30 @@ 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.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 +56,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 +106,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 +187,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 "" 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 7dff1a9..5c50674 100644 --- a/src/entrabot/tools/chat_cursors.py +++ b/src/entrabot/tools/chat_cursors.py @@ -40,6 +40,7 @@ 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") @@ -225,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..ee4e24c 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 # --------------------------------------------------------------------------- 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 0c2725e..237ae3a 100644 --- a/tests/test_mcp_server_chat_cursors.py +++ b/tests/test_mcp_server_chat_cursors.py @@ -280,6 +280,93 @@ async def _rec(*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 == [] + + # --------------------------------------------------------------------------- # Marking a chat dirty + flushing # --------------------------------------------------------------------------- diff --git a/tests/tools/test_chat_cursors.py b/tests/tools/test_chat_cursors.py index e07e648..bf7e544 100644 --- a/tests/tools/test_chat_cursors.py +++ b/tests/tools/test_chat_cursors.py @@ -317,6 +317,114 @@ def read_text(self, key: str) -> str | 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( From f61c4804c344e2dd1351cc69a6fe91ce32af0e11 Mon Sep 17 00:00:00 2001 From: Brandon Werner Date: Thu, 2 Jul 2026 09:29:47 -0700 Subject: [PATCH 3/4] fix(storage): fail loud on half-configured blob backend at boot (F2) A silent Local fallback when only one of ENTRABOT_BLOB_ENDPOINT / ENTRABOT_BLOB_CONTAINER is set is how a mis-enved fleet instance lands on an empty local store and re-bootstraps every chat (replay-flood root cause). - Add BackendMisconfiguredError; get_backend() raises it on a half-configured blob env instead of returning LocalBackend. ENTRABOT_KEEP_MEMORY_LOCAL=true still forces Local. - Add assert_backend_config(): resolves + validates the backend once at boot, logs the resolved backend + container so operators can confirm fleet-wide agreement, and raises on half-config. Wired into _initialize. - Isolate test_one_chat_403_does_not_starve_others to a tmp data dir (its cursor writes were leaking to the real data dir). Closes F2. --- src/entrabot/errors.py | 25 +++++++++ src/entrabot/mcp_server.py | 14 +++++ src/entrabot/storage/backend.py | 55 +++++++++++++++++-- tests/storage/test_backend.py | 82 +++++++++++++++++++++++++++- tests/test_mcp_server_integration.py | 11 +++- 5 files changed, 179 insertions(+), 8 deletions(-) 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 cbb6cfd..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: diff --git a/src/entrabot/storage/backend.py b/src/entrabot/storage/backend.py index 10e81d5..5785b4a 100644 --- a/src/entrabot/storage/backend.py +++ b/src/entrabot/storage/backend.py @@ -26,6 +26,7 @@ from typing import TYPE_CHECKING, Protocol, runtime_checkable from entrabot.config import get_config +from entrabot.errors import BackendMisconfiguredError from entrabot.storage.blob import BlobStore, ConcurrencyError from entrabot.tools.teams import acquire_agent_user_storage_token @@ -221,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: @@ -237,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/tests/storage/test_backend.py b/tests/storage/test_backend.py index ee4e24c..0fbe78d 100644 --- a/tests/storage/test_backend.py +++ b/tests/storage/test_backend.py @@ -311,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( @@ -332,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/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" From 7743b5055f722bef6a8154305409b57187579d78 Mon Sep 17 00:00:00 2001 From: Brandon Werner Date: Thu, 2 Jul 2026 09:31:38 -0700 Subject: [PATCH 4/4] docs: record fleet-safe cursor fix status (F1/F2/F4/F5/idempotency shipped, F3 deferred) Update the design doc status + Shipped/Deferred sections, TODOS.md item, and engineering-status.md to reflect what landed on this branch. --- TODOS.md | 14 +++-- ...ESIGN-multi-instance-cursor-consistency.md | 51 ++++++++++++++++++- docs/engineering-status.md | 2 +- 3 files changed, 61 insertions(+), 6 deletions(-) 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
", "run