Merge remote-tracking branch 'origin/duncan/nip-rs-unread-manager' into duncan/nip-rs-unread-format

* origin/duncan/nip-rs-unread-manager:
  test(desktop): compress readStateManager test matrices to Biome-stable rows
  test(read-state): compress readStateManager tests to table-driven matrices
  test(nip-rs-unread-ui): compress mechanical test matrices to table-driven loops
  fix(nip-rs): close round-4 findings — projection-sole-authority, torn-register witness
  fix(nip-rs): close round-3 findings — production wiring, tombstone ordering, mark-all transition
  fix(nip-rs): close pass-3 findings — projection-based observer, loadComplete ready gate, evidence retention
  fix(nip-rs): apply max() for v2 frontier in hydration; add stale-ancillary witness
  fix(read-state): close timer race, hydration publishability, v2 durable leakage
  fix(nip-rs): close C1/I2/I3 residuals — fence timeout, commit point, real-primitive tests
  fix(nip-rs): add fence primitive, parsed-record loader, transactional storage (contracts 1-3)
  fix(nip-rs): close Thufir pass-3 findings (fence, parsed records, transactional state)
  fix(nip-rs): close Thufir pass-2 findings in slice-3 UI layer
  fix(nip-rs): add fencedLoader integration and manager/storage fixes
  fix(nip-rs): correct fencedLoader lapse detection and manager semantics
  test(read-state): add overlay and mark-all partial-refusal witnesses
  fix(read-state): close Thufir pass-1 findings in nip-rs unread UI (slice 3)
  fix(read-state): close Thufir pass-1 findings in readStateManager/Storage
  feat(read-state): wire right-click mark-read/unread to NIP-RS override layer (slice 3)
  feat(read-state): add override layer to ReadStateManager (slice 2)

Signed-off-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

desktop/src/app/AppShell.tsx

@@ -342,6 +342,7 @@ export function AppShell() {
     mutedRootIds,
     muteThread,
     unmuteThread,
+    getProjection,
   } = useUnreadChannels(sidebarChannels, activeChannel, {
     pubkey: identityQuery.data?.pubkey,
     relayClient,
@@ -404,8 +405,6 @@ export function AppShell() {
     mutedRootIds,
     channels,
   );
-
-  // Badge count consumes the shared NIP-RS read-state from useUnreadChannels.
   const { homeBadgeCount, homeBadgeCountExcludingHighPriority } =
     useHomeFeedNotificationState(
       homeFeedQuery.data,
@@ -733,6 +732,7 @@ export function AppShell() {
             threadActivityFeedItems,
             feedItemState,
             onOpenSettings: handleOpenSettings,
+            getProjection,
           }}
         >
           <HuddleProvider>

desktop/src/app/AppShellContext.tsx

@@ -1,5 +1,6 @@
 import * as React from "react";
 import type { ContextParentResolver } from "@/features/channels/readState/readStateManager";
+import type { ReadStateProjection } from "@/features/channels/readState/readStateManager";
 import type { ThreadActivityItem } from "@/features/channels/useUnreadChannels";
 import type { FeedItemState } from "@/features/home/useFeedItemState";
 import type { FeedItem } from "@/shared/api/types";
@@ -14,7 +15,7 @@ type AppShellContextValue = {
     readAt: string | null | undefined,
     options?: { topLevelOnly?: boolean },
   ) => void;
-  markChannelUnread: (channelId: string) => void;
+  markChannelUnread: (channelId: string) => boolean;
   openBrowseChannels: () => void;
   openCreateChannel: () => void;
   openChannelManagement: (channelId?: string) => void;
@@ -51,12 +52,15 @@ type AppShellContextValue = {
   // that render under AppShell (channel, home, projects, pulse, agents).
   // Used by config-nudge cards to deep-link to Settings → Agents.
   onOpenSettings: ((section: SettingsSection) => void) | null;
+  // Manager projection accessor for community rail polls — re-read per poll,
+  // never cached across notification boundaries.
+  getProjection: () => ReadStateProjection | null;
 };
 
 const AppShellContext = React.createContext<AppShellContextValue>({
   markAllChannelsRead: () => {},
   markChannelRead: () => {},
-  markChannelUnread: () => {},
+  markChannelUnread: () => false,
   openBrowseChannels: () => {},
   openCreateChannel: () => {},
   openChannelManagement: () => {},
@@ -83,6 +87,7 @@ const AppShellContext = React.createContext<AppShellContextValue>({
     unreadSet: EMPTY_SET,
   },
   onOpenSettings: null,
+  getProjection: () => null,
 });
 
 export function AppShellProvider({

desktop/src/features/channels/forcedUnreadStore.ts

@@ -1,5 +1,5 @@
 /**
- * Per-pubkey localStorage record store for channels manually marked unread via
+ * Per-pubkey localStorage cache of channels manually marked unread via
  * right-click → "mark unread". Keyed by channelId (not thread-root). Persisted
  * so the sidebar badge survives reload and the rail observer can read it for
  * inactive communities.
@@ -13,8 +13,12 @@
  * On identity change, the in-memory map is swapped to the current pubkey's
  * persisted data. Old pubkey data is NOT wiped from localStorage.
  *
- * NOT synced to the relay — NIP-RS markers are monotonic and cannot represent
- * a retrograde "unread" state. localStorage is best-effort (per-device).
+ * This store is the local-device cache of the NIP-RS manual-unread override
+ * layer (kind:30078 `ov_*` entries). Marking a channel unread publishes a
+ * NIP-RS override event to the relay via ReadStateManager so the override
+ * propagates to other devices. Marking a channel read clears the override
+ * both locally and on the relay. The `markerAtWhenForced` baseline still drives
+ * the "cross-device read already covered this" gate for the rail observer.
  */
 
 /** channelId → NIP-RS read marker (unix seconds) at force-time, or null */

desktop/src/features/channels/nip-rs-unread-ui.test.mjs

@@ -0,0 +1,622 @@
+/**
+ * NIP-RS manual-unread UI layer tests (slice 3).
+ *
+ * Tests exercise production code — applyOverrideUnread, applyOverrideRead,
+ * persistForcedUnread, overrideErrorMessage — by importing and invoking them
+ * directly with injected manager/store/toast dependencies.
+ *
+ * Behavioral requirements:
+ *  1. mark-unread calls the manager's NIP-RS override path and blocks the
+ *     local cache update when the manager refuses.
+ *  2. mark-read completes only when resulting override_active == false (spec
+ *     MUST at NIP-RS:537-539). null liveness ≠ inactive — pre-init fails
+ *     closed.
+ *  3. budget_exhausted, uint32_overflow, and load_incomplete refusals produce
+ *     distinct non-empty user-visible messages; already_inactive is silent.
+ *  4. forcedUnreadStore is the local cache of the NIP-RS override layer.
+ *  5. persistForcedUnread refreshes an existing entry (re-mark after override
+ *     died uses the fresh baseline).
+ */
+
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  applyMarkAllReadTransition,
+  applyOverrideRead,
+  applyOverrideUnread,
+  overrideErrorMessage,
+  persistForcedUnread,
+} from "./readStateOverride.ts";
+import { forcedUnreadStore } from "./forcedUnreadStore.ts";
+import { resolveChannelReadMarker } from "./useUnreadChannels.ts";
+import { applyMarkUnreadDividerOverlay } from "./ui/useChannelUnreadState.ts";
+
+// ── localStorage mock ────────────────────────────────────────────────────────
+
+if (typeof globalThis.window === "undefined") {
+  const store = new Map();
+  globalThis.window = {
+    localStorage: {
+      getItem: (key) => store.get(key) ?? null,
+      setItem: (key, value) => store.set(key, value),
+      removeItem: (key) => store.delete(key),
+    },
+  };
+}
+
+function makeIsolatedStorage() {
+  const store = new Map();
+  const prev = globalThis.window.localStorage;
+  globalThis.window.localStorage = {
+    getItem: (key) => store.get(key) ?? null,
+    setItem: (key, value) => store.set(key, value),
+    removeItem: (key) => store.delete(key),
+  };
+  return {
+    store,
+    restore: () => {
+      globalThis.window.localStorage = prev;
+    },
+  };
+}
+
+// ── Toast capture mock ───────────────────────────────────────────────────────
+// applyOverrideUnread / applyOverrideRead call toast.error() from sonner.
+// Capture calls by patching the module-level toast reference through a closure.
+// Since we can't monkey-patch ES module imports in Node, we drive the
+// production toast-policy logic directly via overrideErrorMessage instead.
+
+// ── Tests: overrideErrorMessage (toast policy) ───────────────────────────────
+
+test("overrideErrorMessage_budget_exhausted_unread_contains_budget_keyword", () => {
+  const msg = overrideErrorMessage("unread", "budget_exhausted");
+  assert.ok(msg?.includes("budget exhausted"), `got: ${msg}`);
+});
+
+test("overrideErrorMessage_uint32_overflow_unread_contains_counter_keyword", () => {
+  const msg = overrideErrorMessage("unread", "uint32_overflow");
+  assert.ok(msg?.includes("counter limit"), `got: ${msg}`);
+});
+
+test("overrideErrorMessage_uint32_overflow_read_contains_counter_keyword", () => {
+  const msg = overrideErrorMessage("read", "uint32_overflow");
+  assert.ok(msg?.includes("counter limit"), `got: ${msg}`);
+});
+
+test("overrideErrorMessage_load_incomplete_unread_contains_loading_keyword", () => {
+  const msg = overrideErrorMessage("unread", "load_incomplete");
+  assert.ok(msg?.includes("still loading"), `got: ${msg}`);
+});
+
+test("overrideErrorMessage_load_incomplete_read_contains_loading_keyword", () => {
+  const msg = overrideErrorMessage("read", "load_incomplete");
+  assert.ok(msg?.includes("still loading"), `got: ${msg}`);
+});
+
+test("overrideErrorMessage_already_inactive_returns_null_both_ops", () => {
+  assert.equal(overrideErrorMessage("unread", "already_inactive"), null);
+  assert.equal(overrideErrorMessage("read", "already_inactive"), null);
+});
+
+// ── Tests: applyOverrideUnread ───────────────────────────────────────────────
+
+test("applyOverrideUnread_success_returns_true", () => {
+  const toasts = [];
+  const result = applyOverrideUnread("ch-1", {
+    markChannelUnread: () => ({ success: true }),
+    markChannelRead: () => ({ success: true }),
+    getOverrideLiveness: () => null,
+  });
+  assert.equal(result, true);
+  assert.equal(toasts.length, 0);
+});
+
+for (const reason of [
+  "budget_exhausted",
+  "uint32_overflow",
+  "load_incomplete",
+]) {
+  test(`applyOverrideUnread_${reason}_returns_false`, () => {
+    const result = applyOverrideUnread("ch-1", {
+      markChannelUnread: () => ({ success: false, reason }),
+      markChannelRead: () => ({ success: true }),
+      getOverrideLiveness: () => null,
+    });
+    assert.equal(result, false);
+  });
+}
+
+// ── Tests: applyOverrideRead ─────────────────────────────────────────────────
+//
+// New semantics (NIP-RS:537-539):
+//   1. !isReadStateReady → fail closed (overrideStillActive)
+//   2. isReadStateReady + null liveness → known absence (overrideCleared, no C-bump)
+//   3. isReadStateReady + register exists → ALWAYS attempt C-bump, then re-check
+
+// Helper: builds a minimal OverrideAPIs object with sensible defaults.
+function makeApis(overrides = {}) {
+  return {
+    isReadStateReady: true,
+    markChannelUnread: () => ({ success: true }),
+    markChannelRead: () => ({ success: true }),
+    getOverrideLiveness: () => null,
+    ...overrides,
+  };
+}
+
+test("applyOverrideRead_manager_not_ready_fails_closed_overrideStillActive", () => {
+  // Step 1: !isReadStateReady → fail closed without calling manager.
+  let bumped = false;
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: false,
+      markChannelRead: () => {
+        bumped = true;
+        return { success: true };
+      },
+      getOverrideLiveness: () => ({ active: true, frontier: 50 }),
+    }),
+  );
+  assert.equal(result, "overrideStillActive");
+  assert.equal(bumped, false, "C-bump must not fire when manager not ready");
+});
+
+test("applyOverrideRead_ready_null_liveness_no_register_overrideCleared", () => {
+  // Step 2: ready + null liveness = known no-register → cleared, no C-bump.
+  let bumped = false;
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: true,
+      markChannelRead: () => {
+        bumped = true;
+        return { success: true };
+      },
+      getOverrideLiveness: () => null,
+    }),
+  );
+  assert.equal(result, "overrideCleared");
+  assert.equal(bumped, false, "no C-bump when no register exists");
+});
+
+test("applyOverrideRead_inactive_register_cbump_attempted_then_clears", () => {
+  // Step 3 with inactive register: C-bump must be attempted (NIP-RS:537-539).
+  // After the bump, post-call liveness is still inactive → overrideCleared.
+  let bumped = false;
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: true,
+      markChannelRead: () => {
+        bumped = true;
+        return { success: true };
+      },
+      getOverrideLiveness: () => ({ active: false, frontier: 200 }),
+    }),
+  );
+  assert.equal(result, "overrideCleared");
+  assert.equal(bumped, true, "C-bump attempted even when already inactive");
+});
+
+test("applyOverrideRead_active_liveness_success_then_inactive_overrideCleared", () => {
+  // Active register → C-bump succeeds → re-read shows inactive → cleared.
+  let callCount = 0;
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: true,
+      markChannelRead: () => ({ success: true }),
+      getOverrideLiveness: () => {
+        callCount++;
+        return callCount === 1
+          ? { active: true, frontier: 50 }
+          : { active: false, frontier: 50 };
+      },
+    }),
+  );
+  assert.equal(result, "overrideCleared");
+});
+
+test("applyOverrideRead_active_liveness_load_incomplete_overrideStillActive", () => {
+  // Active → C-bump refuses (load_incomplete) → liveness still active → retained.
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: true,
+      markChannelRead: () => ({ success: false, reason: "load_incomplete" }),
+      getOverrideLiveness: () => ({ active: true, frontier: 50 }),
+    }),
+  );
+  assert.equal(result, "overrideStillActive");
+});
+
+test("applyOverrideRead_already_inactive_race_overrideCleared", () => {
+  // already_inactive race: cleared by another device → silent success.
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: true,
+      markChannelRead: () => ({ success: false, reason: "already_inactive" }),
+      getOverrideLiveness: () => ({ active: true, frontier: 50 }),
+    }),
+  );
+  assert.equal(result, "overrideCleared");
+});
+
+test("applyOverrideRead_post_call_null_liveness_treated_as_cleared", () => {
+  // Post-bump liveness is null (register deleted remotely) → overrideCleared.
+  let callCount = 0;
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: true,
+      markChannelRead: () => ({ success: true }),
+      getOverrideLiveness: () => {
+        callCount++;
+        // Pre-call: active register. Post-call: register gone (null).
+        return callCount === 1 ? { active: true, frontier: 50 } : null;
+      },
+    }),
+  );
+  assert.equal(result, "overrideCleared");
+});
+
+// ── Tests: persistForcedUnread ───────────────────────────────────────────────
+
+test("persistForcedUnread_new_entry_adds_and_returns_true", () => {
+  const { restore } = makeIsolatedStorage();
+  try {
+    const forcedMap = {};
+    const result = persistForcedUnread("ch-1", forcedMap, () => 1000, "pk");
+    assert.equal(result, true);
+    assert.equal(Object.hasOwn(forcedMap, "ch-1"), true);
+    assert.equal(forcedMap["ch-1"], 1000);
+  } finally {
+    restore();
+  }
+});
+
+test("persistForcedUnread_existing_entry_with_same_ts_noop_returns_false", () => {
+  const { restore } = makeIsolatedStorage();
+  try {
+    const forcedMap = { "ch-1": 1000 };
+    const result = persistForcedUnread("ch-1", forcedMap, () => 1000, "pk");
+    assert.equal(result, false, "no change when ts identical");
+  } finally {
+    restore();
+  }
+});
+
+test("persistForcedUnread_existing_entry_with_new_ts_refreshes_baseline", () => {
+  // Re-mark after old override died: baseline must update to current frontier
+  const { restore } = makeIsolatedStorage();
+  try {
+    const forcedMap = { "ch-1": 500 };
+    const result = persistForcedUnread("ch-1", forcedMap, () => 1200, "pk");
+    assert.equal(result, true, "returns true when baseline updated");
+    assert.equal(forcedMap["ch-1"], 1200, "new baseline stored");
+  } finally {
+    restore();
+  }
+});
+
+test("persistForcedUnread_persists_to_store", () => {
+  const { restore } = makeIsolatedStorage();
+  try {
+    const forcedMap = {};
+    persistForcedUnread("ch-x", forcedMap, () => 999, "my-pk");
+    const stored = forcedUnreadStore.read("my-pk");
+    assert.equal(stored["ch-x"], 999, "entry visible from store");
+  } finally {
+    restore();
+  }
+});
+
+test("persistForcedUnread_null_timestamp_stores_null", () => {
+  const { restore } = makeIsolatedStorage();
+  try {
+    const forcedMap = {};
+    persistForcedUnread("ch-null", forcedMap, () => null, "pk");
+    assert.equal(forcedMap["ch-null"], null);
+  } finally {
+    restore();
+  }
+});
+
+// ── Tests: resolveChannelReadMarker (read-path gate) ─────────────────────────
+
+test("resolve_channel_read_marker_uses_max_of_caller_and_observed", () => {
+  const { markAt } = resolveChannelReadMarker(
+    "2024-01-01T00:00:00.000Z",
+    1704067999,
+  );
+  assert.equal(
+    markAt,
+    1704067999,
+    "observed latest wins when newer than caller",
+  );
+});
+
+test("resolve_channel_read_marker_uses_caller_when_newer", () => {
+  const future = new Date(Date.now() + 10_000_000).toISOString();
+  const { markAt } = resolveChannelReadMarker(future, 1);
+  assert.ok(markAt !== null && markAt > 1, "caller wins when newer");
+});
+
+test("resolve_channel_read_marker_returns_null_when_both_null", () => {
+  const { markAt } = resolveChannelReadMarker(null, undefined);
+  assert.equal(markAt, null);
+});
+
+test("resolve_channel_read_marker_clears_observed_when_covered", () => {
+  const { clearObserved } = resolveChannelReadMarker(
+    "2024-01-01T00:00:01.000Z",
+    1704067201,
+  );
+  assert.equal(
+    clearObserved,
+    true,
+    "observed cleared when read marker covers it",
+  );
+});
+
+// ── Tests: forcedUnreadStore persistence ─────────────────────────────────────
+
+test("forced_unread_store_read_returns_empty_for_unknown_pubkey", () => {
+  const { restore } = makeIsolatedStorage();
+  try {
+    assert.deepEqual(forcedUnreadStore.read("unknown-pk"), {});
+  } finally {
+    restore();
+  }
+});
+
+test("forced_unread_store_write_and_read_round_trips", () => {
+  const { restore } = makeIsolatedStorage();
+  try {
+    const pk = "test-pk";
+    forcedUnreadStore.write(pk, { "channel-a": 12345, "channel-b": null });
+    assert.deepEqual(forcedUnreadStore.read(pk), {
+      "channel-a": 12345,
+      "channel-b": null,
+    });
+  } finally {
+    restore();
+  }
+});
+
+// ── Tests: refusal-does-not-mutate-overlay (active-channel path) ─────────────
+//
+// Witnesses the requirement that a refused mark-unread does NOT update the
+// caller's local state. applyOverrideUnread returns false on any manager
+// refusal; callers gate cache writes on that return value. This covers the
+// active-channel session-overlay scenario (useChannelUnreadState.handleMarkUnread)
+// where the overlay must NOT be set before or on a refused call.
+
+test("applyOverrideUnread_refusal_caller_must_not_mutate_overlay", () => {
+  const _toasts = [];
+  const apis = {
+    markChannelUnread: () => ({ success: false, reason: "budget_exhausted" }),
+    markChannelRead: () => ({ success: true }),
+    getOverrideLiveness: () => null,
+  };
+  // Simulate caller logic: only update local state when applyOverrideUnread
+  // returns true. On false, the overlay map must remain unmodified.
+  const overlayMap = {};
+  const result = applyOverrideUnread("ch-overlay", apis);
+  if (result) {
+    overlayMap["ch-overlay"] = true; // would be set on success
+  }
+  assert.equal(result, false, "refusal returns false");
+  assert.equal(
+    Object.hasOwn(overlayMap, "ch-overlay"),
+    false,
+    "overlay not mutated on refusal",
+  );
+});
+
+// ── Tests: applyMarkUnreadDividerOverlay (active-channel handleMarkUnread) ────
+//
+// This is the extracted production transition from handleMarkUnread in
+// useChannelUnreadState. Tests verify: success adds to overlay, refusal does not.
+
+test("applyMarkUnreadDividerOverlay_success_adds_to_overlay_returns_true", () => {
+  const overlay = new Set();
+  const result = applyMarkUnreadDividerOverlay(
+    "ch-active",
+    () => true, // markFn returns true = accepted
+    overlay,
+  );
+  assert.equal(result, true, "returns true on acceptance");
+  assert.equal(overlay.has("ch-active"), true, "adds to overlay on success");
+});
+
+test("applyMarkUnreadDividerOverlay_refusal_does_not_add_to_overlay_returns_false", () => {
+  const overlay = new Set();
+  const result = applyMarkUnreadDividerOverlay(
+    "ch-refused",
+    () => false, // markFn returns false = refused
+    overlay,
+  );
+  assert.equal(result, false, "returns false on refusal");
+  assert.equal(
+    overlay.has("ch-refused"),
+    false,
+    "overlay not mutated on refusal",
+  );
+});
+
+// ── Tests: markAllChannelsRead partial refusal pattern ────────────────────────
+//
+// Witnesses that per-channel gating in markAllChannelsRead correctly clears
+// only channels whose override is confirmed inactive. Channels where the
+// C-bump refuses and liveness remains active must keep their local state.
+
+test("applyOverrideRead_partial_refusal_pattern_clears_only_inactive", () => {
+  // ch-cleared: liveness inactive after successful C-bump → cleared
+  // ch-stuck: liveness active and C-bump refuses → stays unread
+  const livenessMap = {
+    "ch-cleared": { active: true, frontier: 100 },
+    "ch-stuck": { active: true, frontier: 100 },
+  };
+  let clearCallCount = 0;
+  const apis = {
+    isReadStateReady: true,
+    markChannelUnread: () => ({ success: true }),
+    markChannelRead: (id) => {
+      clearCallCount++;
+      if (id === "ch-cleared") {
+        livenessMap["ch-cleared"] = { active: false, frontier: 101 };
+        return { success: true };
+      }
+      // ch-stuck: refuse with overflow; liveness stays active
+      return { success: false, reason: "uint32_overflow" };
+    },
+    getOverrideLiveness: (id) => livenessMap[id] ?? null,
+  };
+
+  // Simulate the per-channel loop in markAllChannelsRead
+  const forcedMap = { "ch-cleared": 99, "ch-stuck": 99 };
+  for (const channelId of ["ch-cleared", "ch-stuck"]) {
+    if (applyOverrideRead(channelId, apis) === "overrideCleared") {
+      delete forcedMap[channelId];
+    }
+  }
+
+  assert.equal(
+    Object.hasOwn(forcedMap, "ch-cleared"),
+    false,
+    "cleared channel removed from map",
+  );
+  assert.equal(
+    Object.hasOwn(forcedMap, "ch-stuck"),
+    true,
+    "stuck channel remains in map",
+  );
+  assert.equal(clearCallCount, 2, "C-bump attempted for both channels");
+});
+
+// ── Adversarial witnesses (pass-3 closures) ───────────────────────────────────
+
+test("adversarial-3: incomplete_load_null_liveness_not_known_absent_overrideStillActive", () => {
+  // IMPORTANT: isReadStateReady = false (load incomplete) + null getOverrideLiveness
+  // must NOT be treated as known register absence. The old bug: isReadStateReady was
+  // set on initialization (not load-complete), so a partially-loaded manager reported
+  // null liveness which was silently treated as "no register" and the local hint was
+  // cleared. With isReadStateReady now meaning loadComplete, this path is fail-closed.
+  //
+  // This test drives applyOverrideRead with isReadStateReady=false (semantically:
+  // load is not complete) and null liveness — must return overrideStillActive.
+  let markCalled = false;
+  const result = applyOverrideRead(
+    "ch-1",
+    makeApis({
+      isReadStateReady: false, // load not complete
+      markChannelRead: () => {
+        markCalled = true;
+        return { success: false, reason: "load_incomplete" };
+      },
+      getOverrideLiveness: () => null, // null from incomplete load ≠ known absent
+    }),
+  );
+  assert.equal(
+    result,
+    "overrideStillActive",
+    "incomplete load must fail closed",
+  );
+  assert.equal(
+    markCalled,
+    false,
+    "must not call manager when load not complete",
+  );
+});
+
+test("adversarial-4: applyMarkAllReadTransition_both_branches", () => {
+  // IMPORTANT: applyMarkAllReadTransition is the extracted production transition
+  // used inside markAllChannelsRead. Testing it directly (not a simulation)
+  // ensures production regressions to unconditional deletion are caught.
+  //
+  // ch-cleared: applyOverrideRead succeeds (liveness goes inactive) →
+  //   forcedUnread, latestByChannel, observedUnreadEvents all removed.
+  // ch-refused: applyOverrideRead refused (liveness stays active) →
+  //   all three maps preserved intact.
+  const livenessMap = {
+    "ch-cleared": { active: true, frontier: 100 },
+    "ch-refused": { active: true, frontier: 100 }, // uint32_overflow refusal
+  };
+
+  const apis = {
+    isReadStateReady: true,
+    markChannelUnread: () => ({ success: true }),
+    markChannelRead: (id) => {
+      if (id === "ch-cleared") {
+        livenessMap["ch-cleared"] = { active: false, frontier: 101 };
+        return { success: true };
+      }
+      return { success: false, reason: "uint32_overflow" };
+    },
+    getOverrideLiveness: (id) => livenessMap[id] ?? null,
+  };
+
+  const latestByChannel = new Map([
+    ["ch-cleared", 200],
+    ["ch-refused", 200],
+  ]);
+  const observedEvents = new Map([
+    ["ch-cleared", new Map([["ev1", { kind: 9, createdAt: 200 }]])],
+    ["ch-refused", new Map([["ev2", { kind: 9, createdAt: 200 }]])],
+  ]);
+  const forcedMap = { "ch-cleared": 99, "ch-refused": 99 };
+  const maps = {
+    forcedUnread: forcedMap,
+    latestByChannel,
+    observedUnreadEvents: observedEvents,
+  };
+
+  // Drive the production transition directly (not a reimplementation).
+  const clearedOutcome = applyMarkAllReadTransition("ch-cleared", apis, maps);
+  const refusedOutcome = applyMarkAllReadTransition("ch-refused", apis, maps);
+
+  assert.equal(clearedOutcome, "overrideCleared", "cleared channel: outcome");
+  assert.equal(
+    refusedOutcome,
+    "overrideStillActive",
+    "refused channel: outcome",
+  );
+
+  // ch-cleared: override confirmed inactive → all three maps purged.
+  assert.equal(
+    latestByChannel.has("ch-cleared"),
+    false,
+    "cleared: latestByChannel removed",
+  );
+  assert.equal(
+    observedEvents.has("ch-cleared"),
+    false,
+    "cleared: observedEvents removed",
+  );
+  assert.equal(
+    Object.hasOwn(forcedMap, "ch-cleared"),
+    false,
+    "cleared: forcedMap removed",
+  );
+
+  // ch-refused: override still active → all three maps preserved.
+  assert.equal(
+    latestByChannel.has("ch-refused"),
+    true,
+    "refused: latestByChannel preserved",
+  );
+  assert.equal(
+    observedEvents.has("ch-refused"),
+    true,
+    "refused: observedEvents preserved",
+  );
+  assert.equal(
+    Object.hasOwn(forcedMap, "ch-refused"),
+    true,
+    "refused: forcedMap preserved",
+  );
+});

desktop/src/features/channels/readState/readStateFencedLoader.ts

@@ -0,0 +1,191 @@
+/**
+ * NIP-RS full-state fenced loader.
+ *
+ * Implements the NIP-RS §Full-State Load procedure (NIP-RS.md:321-377):
+ * - EOSE-established fence via `relay.subscribeFenced()` — resolves only on
+ *   this subscription's own EOSE, never via the 250 ms fallback or CLOSED
+ *   (NIP-RS.md:341-345). Events are delivered synchronously; no timer drain.
+ * - ANY lapse (before EOSE, mid-enumeration, post-enumeration) forces
+ *   `complete: false` regardless of query results.
+ * - Descending `until` cursor with pinned-window check (NIP-RS.md:350-352).
+ * - `T === 0` terminates complete after the pinned window is discharged —
+ *   no lower standard timestamp exists, so history is fully exhausted.
+ * - Coordinate deduplicated by `d` tag (greatest `created_at`, lowest id).
+ * - Returns the deduped parsed records directly so the caller can process them
+ *   once for merge, metadata, conflict rotation, and `maxFetchedCreatedAt`.
+ */
+
+import type { RelayClient } from "@/shared/api/relayClientSession";
+import type { RelayEvent } from "@/shared/api/types";
+import { KIND_READ_STATE } from "@/shared/constants/kinds";
+import {
+  parseReadStateEvent,
+  type ParsedReadStateEvent,
+} from "@/features/channels/readState/readStateSnapshot";
+import { isValidReadStateDTag } from "@/features/channels/readState/readStateFormat";
+
+/** Number of events per enumeration band. MUST be ≥ L=2 (NIP-RS.md:347). */
+const BAND_LIMIT = 500;
+/** NIP-RS §Floor: relay MUST deliver ≥ L events when ≥ L exist (NIP-RS.md:360). */
+const L = 2;
+
+export type FencedLoadResult =
+  | { complete: true; events: ParsedReadStateEvent[] }
+  | { complete: false; events: ParsedReadStateEvent[] };
+
+/**
+ * Perform one full-state fenced enumeration for `pubkey`.
+ *
+ * Returns the coordinate-deduped parsed records and whether the load was proven
+ * complete per the NIP-RS §Full-State Load procedure. A `complete: false`
+ * result carries best-effort events for baseline display.
+ */
+export async function fencedEnumerationLoad(
+  relay: RelayClient,
+  pubkey: string,
+): Promise<FencedLoadResult> {
+  const baseFilter = {
+    kinds: [KIND_READ_STATE],
+    authors: [pubkey],
+    limit: BAND_LIMIT,
+  };
+
+  // ── Step 1: Establish the fence BEFORE the first query ──────────────────
+  // subscribeFenced() resolves `established` ONLY on EOSE — no fallback timer.
+  // Events are delivered synchronously to `fenceEvents`; no drain timer needed.
+  const fenceEvents: RelayEvent[] = [];
+  let fence: Awaited<ReturnType<typeof relay.subscribeFenced>>;
+  try {
+    fence = await relay.subscribeFenced(baseFilter, (ev) => {
+      fenceEvents.push(ev);
+    });
+  } catch {
+    return emptyIncomplete();
+  }
+
+  if (fence.lapsed) {
+    return emptyIncomplete();
+  }
+
+  // Wait for EOSE — resolves only when the relay confirms this subscription.
+  await fence.established;
+
+  if (fence.lapsed) {
+    await fence.unsubscribe();
+    return emptyIncomplete();
+  }
+
+  // ── Step 2: Descending enumeration ──────────────────────────────────────
+  let C = 0;
+  let until: number | undefined;
+  const bandEvents: RelayEvent[] = [];
+  let complete = false;
+
+  while (!fence.lapsed) {
+    const filter = { ...baseFilter, ...(until !== undefined ? { until } : {}) };
+    let band: RelayEvent[];
+    try {
+      band = await relay.fetchEvents(filter);
+    } catch {
+      break;
+    }
+    if (fence.lapsed) break;
+
+    if (band.length === 0) {
+      complete = true;
+      break;
+    }
+
+    if (band.length > C) C = band.length;
+    for (const ev of band) bandEvents.push(ev);
+
+    // T = lowest created_at in this band (NIP-RS.md:350).
+    let T = band[0].created_at;
+    for (const ev of band) if (ev.created_at < T) T = ev.created_at;
+
+    // ── Pinned window (NIP-RS.md:350-351) ───────────────────────────────
+    if (fence.lapsed) break;
+    let pinned: RelayEvent[];
+    try {
+      pinned = await relay.fetchEvents({ ...baseFilter, since: T, until: T });
+    } catch {
+      break;
+    }
+    if (fence.lapsed) break;
+
+    for (const ev of pinned) bandEvents.push(ev);
+    if (pinned.length > C) C = pinned.length;
+
+    if (pinned.length >= Math.max(C, L)) {
+      // Pinned window not discharged — potentially incomplete (spec :351).
+      break;
+    }
+
+    if (T === 0) {
+      // T=0: the pinned window was just discharged (pinned.length < max(C,L)).
+      // No event can have created_at < 0, so the history is fully exhausted —
+      // terminate complete directly.  An `until:0` continuation would be an
+      // inclusive re-fetch of the same second-zero events and cannot prove
+      // anything additional (spec :352).
+      complete = true;
+      break;
+    }
+
+    until = T - 1;
+  }
+
+  // ── Unsubscribe fence ────────────────────────────────────────────────────
+  // Events delivered synchronously — no timer drain needed.
+  await fence.unsubscribe();
+
+  // A lapse at ANY point (including after `complete` was tentatively set)
+  // forces complete:false.
+  if (fence.lapsed) {
+    const all = [...bandEvents, ...fenceEvents];
+    return buildResult(false, all, pubkey);
+  }
+
+  const all = [...bandEvents, ...fenceEvents];
+  return buildResult(complete, all, pubkey);
+}
+
+/**
+ * Deduplicate events by `d` tag: retain the event with the greatest
+ * `created_at`; break ties by the lexicographically lowest event id.
+ * Prevents superseded coordinate versions from being merged as concurrent
+ * replicas (NIP-RS.md:348).
+ */
+export function deduplicateByCoordinate(events: RelayEvent[]): RelayEvent[] {
+  const best = new Map<string, RelayEvent>();
+  for (const ev of events) {
+    const dTag = ev.tags.find((t) => t[0] === "d")?.[1];
+    if (!dTag || !isValidReadStateDTag(dTag)) continue;
+    const existing = best.get(dTag);
+    if (
+      !existing ||
+      ev.created_at > existing.created_at ||
+      (ev.created_at === existing.created_at && ev.id < existing.id)
+    ) {
+      best.set(dTag, ev);
+    }
+  }
+  return [...best.values()];
+}
+
+function emptyIncomplete(): FencedLoadResult {
+  return { complete: false, events: [] };
+}
+
+async function buildResult(
+  complete: boolean,
+  events: RelayEvent[],
+  pubkey: string,
+): Promise<FencedLoadResult> {
+  const deduped = deduplicateByCoordinate(events);
+  const parsed: ParsedReadStateEvent[] = [];
+  for (const ev of deduped) {
+    const p = await parseReadStateEvent(ev, pubkey);
+    if (p) parsed.push(p);
+  }
+  return { complete, events: parsed };
+}

desktop/src/features/channels/readState/readStateStorage.test.mjs

@@ -97,6 +97,7 @@ test("writeStoredReadState prunes all three keys consistently", () => {
       [staleThread, stale],
       [freshThread, nowSeconds],
     ]),
+    new Map(),
   );
 
   const state = JSON.parse(
@@ -128,15 +129,21 @@ test("writeStoredReadState round-trips through readStoredReadState", () => {
     new Map([["channel-9", nowSeconds]]),
     new Set(["channel-9"]),
     new Map([["channel-9", nowSeconds]]),
+    new Map([["channel-9", { s: 3, c: 1, b: nowSeconds, f: nowSeconds }]]),
   );
 
   const stored = readStoredReadState(pubkey);
   assert.equal(stored.contexts.get("channel-9"), nowSeconds);
   assert.equal(stored.publishableContextIds.has("channel-9"), true);
   assert.equal(stored.contextSourceCreatedAt.get("channel-9"), nowSeconds);
+  const reg = stored.overrideRegisters.get("channel-9");
+  assert.ok(reg, "overrideRegisters must round-trip");
+  assert.equal(reg.s, 3, "register S must round-trip");
+  assert.equal(reg.c, 1, "register C must round-trip");
+  assert.equal(reg.b, nowSeconds, "register B must round-trip");
 });
 
-test("writeStoredReadState survives a throwing localStorage.setItem", () => {
+test("writeStoredReadState_survives_throwing_localStorage_and_returns_false", () => {
   const ls = installLocalStorage();
   ls.setItem = () => {
     throw new Error("QuotaExceededError");
@@ -144,12 +151,19 @@ test("writeStoredReadState survives a throwing localStorage.setItem", () => {
   const pubkey = "d".repeat(64);
   const nowSeconds = Math.floor(Date.now() / 1_000);
 
+  let result;
   assert.doesNotThrow(() => {
-    writeStoredReadState(
+    result = writeStoredReadState(
       pubkey,
       new Map([["channel-1", nowSeconds]]),
       new Set(["channel-1"]),
       new Map([["channel-1", nowSeconds]]),
+      new Map(),
     );
   });
+  assert.equal(
+    result,
+    false,
+    "writeStoredReadState must return false on quota failure",
+  );
 });

desktop/src/features/channels/readState/readStateStorage.ts

@@ -8,6 +8,7 @@ import {
   MSG_PREFIX,
   READ_STATE_HORIZON_SECONDS,
   THREAD_PREFIX,
+  type OverrideRegister,
 } from "@/features/channels/readState/readStateFormat";
 import { setLocalStorageItemWithRecovery } from "@/shared/lib/localStorageQuota";
 
@@ -15,6 +16,7 @@ export type StoredReadState = {
   contexts: Map<string, number>;
   publishableContextIds: Set<string>;
   contextSourceCreatedAt: Map<string, number>;
+  overrideRegisters: Map<string, StoredOverrideEntry>;
 };
 
 function mergeLocalStorageKey(
@@ -100,6 +102,7 @@ export function readStoredReadState(pubkey: string): StoredReadState {
     contexts,
     publishableContextIds: readPublishableContextIds(pubkey),
     contextSourceCreatedAt: readContextSourceCreatedAt(pubkey),
+    overrideRegisters: readOverrideState(pubkey),
   };
 }
 
@@ -109,6 +112,87 @@ function isPrunableContextKey(contextId: string): boolean {
   );
 }
 
+// Key for persisted override state: registers + their frontier timestamps in one atomic write.
+// v2 stores {s, c, b, f} per context where f = the last known effective frontier.
+function localOverrideStateKey(pubkey: string): string {
+  return `buzz.nip-rs.override-state.v2:${pubkey}`;
+}
+
+// Legacy key retained for migration reads only — never written to after v2 migration.
+function localOverrideRegistersKeyLegacy(pubkey: string): string {
+  return `buzz.nip-rs.override-registers.v1:${pubkey}`;
+}
+
+export type StoredOverrideEntry = OverrideRegister & { f: number };
+
+function readOverrideState(pubkey: string): Map<string, StoredOverrideEntry> {
+  const result = new Map<string, StoredOverrideEntry>();
+  const raw = localStorage.getItem(localOverrideStateKey(pubkey));
+  if (raw) {
+    try {
+      const parsed = JSON.parse(raw);
+      if (isPlainRecord(parsed)) {
+        for (const [rawCtx, value] of Object.entries(parsed)) {
+          if (!isPlainRecord(value)) continue;
+          const { s, c, b, f } = value;
+          if (
+            typeof s === "number" &&
+            Number.isInteger(s) &&
+            s >= 0 &&
+            s <= 0xffffffff &&
+            typeof c === "number" &&
+            Number.isInteger(c) &&
+            c >= 0 &&
+            c <= 0xffffffff &&
+            typeof b === "number" &&
+            Number.isInteger(b) &&
+            b >= 0 &&
+            typeof f === "number" &&
+            Number.isInteger(f) &&
+            f >= 0
+          ) {
+            result.set(rawCtx, { s, c, b, f });
+          }
+        }
+      }
+    } catch {
+      // Corrupt storage — return empty; repopulated on next ingest.
+    }
+    return result;
+  }
+  // Migration: read from legacy v1 key if v2 key absent.
+  const legacyRaw = localStorage.getItem(
+    localOverrideRegistersKeyLegacy(pubkey),
+  );
+  if (!legacyRaw) return result;
+  try {
+    const parsed = JSON.parse(legacyRaw);
+    if (!isPlainRecord(parsed)) return result;
+    for (const [rawCtx, value] of Object.entries(parsed)) {
+      if (!isPlainRecord(value)) continue;
+      const { s, c, b } = value;
+      if (
+        typeof s === "number" &&
+        Number.isInteger(s) &&
+        s >= 0 &&
+        s <= 0xffffffff &&
+        typeof c === "number" &&
+        Number.isInteger(c) &&
+        c >= 0 &&
+        c <= 0xffffffff &&
+        typeof b === "number" &&
+        Number.isInteger(b) &&
+        b >= 0
+      ) {
+        result.set(rawCtx, { s, c, b, f: 0 }); // frontier unknown from legacy key
+      }
+    }
+  } catch {
+    // Corrupt legacy storage — ignore.
+  }
+  return result;
+}
+
 /**
  * Drops msg:/thread: markers older than the relay's 7-day horizon, then caps
  * the survivors at LOCAL_MAX_PRUNABLE_CONTEXTS (oldest first). Channel keys
@@ -147,7 +231,8 @@ export function writeStoredReadState(
   contexts: ReadonlyMap<string, number>,
   publishableContextIds: ReadonlySet<string>,
   contextSourceCreatedAt: ReadonlyMap<string, number>,
-): void {
+  overrideRegisters: ReadonlyMap<string, StoredOverrideEntry>,
+): boolean {
   const pruned = pruneStaleContexts(contexts, Math.floor(Date.now() / 1_000));
 
   const state: Record<string, string> = {};
@@ -155,6 +240,31 @@ export function writeStoredReadState(
     state[contextId] = new Date(timestamp * 1_000).toISOString();
   }
 
+  // Persist override registers atomically with their frontier timestamps (v2).
+  // Registers and frontiers in one JSON blob — a single write ensures they are
+  // never torn: a register cannot be present without its associated frontier.
+  // This is the ACTION COMMIT POINT: if ok4 is true the action is durably
+  // committed; ancillary frontier/cache write failures (ok1-ok3) do not fail
+  // the action.  If ok4 is false, the caller must roll back ALL state.
+  const overrideState: Record<
+    string,
+    { s: number; c: number; b: number; f: number }
+  > = {};
+  for (const [rawCtx, entry] of overrideRegisters) {
+    overrideState[rawCtx] = { s: entry.s, c: entry.c, b: entry.b, f: entry.f };
+  }
+  const ok4 = setLocalStorageItemWithRecovery(
+    localOverrideStateKey(pubkey),
+    JSON.stringify(overrideState),
+  );
+
+  // If the commit-point write failed, return immediately — do NOT write any
+  // ancillary keys.  The caller will roll back all in-memory and slot state.
+  // "v2 false ⇒ nothing durable anywhere" is the contract.
+  if (!ok4) return false;
+
+  // Ancillary writes: frontier cache, publishable set, source timestamps.
+  // Best-effort — failures do not fail the action.
   setLocalStorageItemWithRecovery(
     localReadStateKey(pubkey),
     JSON.stringify(state),
@@ -174,4 +284,7 @@ export function writeStoredReadState(
     localSourceCreatedAtKey(pubkey),
     JSON.stringify(sourceState),
   );
+
+  // The commit point: only the v2 override write determines action success.
+  return ok4;
 }

desktop/src/features/channels/readState/useReadState.ts

@@ -2,13 +2,22 @@ import * as React from "react";
 import {
   ReadStateManager,
   type ContextParentResolver,
+  type MarkResult,
+  type ReadStateProjection,
 } from "@/features/channels/readState/readStateManager";
+import type { OverrideLiveness } from "@/features/channels/readState/readStateFormat";
 import type { RelayClient } from "@/shared/api/relayClientSession";
 
 const noopGetTimestamp = () => null;
 const noopMarkRead = () => {};
 const noopDrainAdvances = (): ReadonlySet<string> => new Set<string>();
 const noopSetResolver = () => {};
+const noopMarkOverride = (): MarkResult => ({
+  success: false,
+  reason: "load_incomplete",
+});
+const noopGetOverrideLiveness = (): OverrideLiveness | null => null;
+const noopGetProjection = (): ReadStateProjection | null => null;
 
 /**
  * React hook that creates and manages a ReadStateManager instance.
@@ -93,31 +102,80 @@ export function useReadState(
     [],
   );
 
+  // Override APIs for the NIP-RS manual-unread layer (slice 3 seam).
+  const markChannelUnread = React.useCallback(
+    (channelId: string): MarkResult => {
+      return (
+        managerRef.current?.markChannelUnread(channelId) ?? {
+          success: false,
+          reason: "load_incomplete",
+        }
+      );
+    },
+    [],
+  );
+
+  const markChannelRead = React.useCallback((channelId: string): MarkResult => {
+    return (
+      managerRef.current?.markChannelRead(channelId) ?? {
+        success: false,
+        reason: "load_incomplete",
+      }
+    );
+  }, []);
+
+  const getOverrideLiveness = React.useCallback(
+    (channelId: string): OverrideLiveness | null => {
+      return managerRef.current?.getOverrideLiveness(channelId) ?? null;
+    },
+    [],
+  );
+
+  const getProjection = React.useCallback((): ReadStateProjection | null => {
+    return managerRef.current?.getProjection() ?? null;
+  }, []);
+
   const isReady = Boolean(
     pubkey && relayClient && initializedPubkey === pubkey,
   );
 
+  // loadComplete is true only when the manager's full-state enumeration has
+  // terminated with a complete verdict. Distinct from isReady (initialized):
+  // a manager can be initialized with an incomplete load.
+  const isLoadComplete =
+    isReady && (managerRef.current?.getProjection().loadComplete ?? false);
+
   if (!pubkey || !relayClient) {
     return {
       getEffectiveTimestamp: noopGetTimestamp,
       isReady: false,
+      isLoadComplete: false,
       markContextRead: noopMarkRead,
       seedContextRead: noopMarkRead,
       drainSyncedAdvances: noopDrainAdvances,
       setContextParentResolver: noopSetResolver,
       readStateVersion: 0,
       getOwnTimestamp: noopGetTimestamp,
+      markChannelUnread: noopMarkOverride,
+      markChannelRead: noopMarkOverride,
+      getOverrideLiveness: noopGetOverrideLiveness,
+      getProjection: noopGetProjection,
     };
   }
 
   return {
     getEffectiveTimestamp,
     isReady,
+    isLoadComplete,
     markContextRead,
     seedContextRead,
     drainSyncedAdvances,
     setContextParentResolver,
     readStateVersion,
     getOwnTimestamp,
+    markChannelUnread,
+    markChannelRead,
+    getOverrideLiveness,
+    getProjection,
   };
 }

desktop/src/features/channels/readStateOverride.ts

@@ -0,0 +1,194 @@
+/**
+ * UI-layer helpers for the NIP-RS manual-unread override layer (slice 3).
+ *
+ * Provides the seam between useUnreadChannels and the ReadStateManager's
+ * override APIs (markChannelUnread / markChannelRead / getOverrideLiveness),
+ * added by slice 2 (nip-rs-unread-manager).
+ */
+import { toast } from "sonner";
+
+import type { OverrideLiveness } from "@/features/channels/readState/readStateFormat";
+import type { MarkResult } from "@/features/channels/readState/readStateManager";
+import {
+  forcedUnreadStore,
+  type ForcedUnreadMap,
+} from "@/features/channels/forcedUnreadStore";
+import type { ObservedUnreadEvent } from "@/features/channels/unreadChannelCounts";
+
+export type { MarkResult, OverrideLiveness };
+
+/**
+ * User-visible error message for a failed NIP-RS override operation.
+ * Returns null for silent failure reasons (already_inactive race condition).
+ */
+export function overrideErrorMessage(
+  op: "unread" | "read",
+  reason: string,
+): string | null {
+  if (reason === "budget_exhausted")
+    return "Could not mark unread: override budget exhausted. Clear some channels first.";
+  if (reason === "uint32_overflow")
+    return op === "unread"
+      ? "Could not mark unread: counter limit reached for this channel."
+      : "Could not clear unread override: counter limit reached.";
+  if (reason === "load_incomplete")
+    return op === "unread"
+      ? "Could not mark unread: read state is still loading. Try again shortly."
+      : "Could not clear unread override: read state is still loading.";
+  if (reason === "already_inactive") return null;
+  return op === "unread"
+    ? "Could not mark unread."
+    : "Could not clear unread override.";
+}
+
+/** Slice-2 override APIs from ReadStateManager, proxied through useReadState. */
+export type OverrideAPIs = {
+  /** True only when the manager's full-state load is complete (loadComplete).
+   *  A manager that finished initializing with an incomplete load must NOT be
+   *  treated as ready — null liveness after an incomplete load means "not found
+   *  in a partial view", not "known absent register". */
+  isReadStateReady: boolean;
+  markChannelUnread: (channelId: string) => MarkResult;
+  markChannelRead: (channelId: string) => MarkResult;
+  getOverrideLiveness: (channelId: string) => OverrideLiveness | null;
+};
+
+/**
+ * Call the manager's override-unread path (S bump + B set to effective
+ * frontier). Returns true when the local cache should be updated; returns false
+ * and shows a toast when the manager refuses.
+ */
+export function applyOverrideUnread(
+  channelId: string,
+  apis: OverrideAPIs,
+): boolean {
+  const result = apis.markChannelUnread(channelId);
+  if (!result.success) {
+    const msg = overrideErrorMessage("unread", result.reason);
+    if (msg) toast.error(msg);
+    return false;
+  }
+  return true;
+}
+
+/**
+ * Outcome of the explicit mark-read transition for one channel.
+ *
+ * `overrideCleared` — the NIP-RS override is now known inactive (no register
+ *   exists, the frontier deactivated it, or the C-bump succeeded). Callers
+ *   should clear local presentation.
+ *
+ * `overrideStillActive` — the manager is unavailable (fail-closed) or the
+ *   C-bump was refused and liveness remains active. Callers MUST NOT clear
+ *   local unread presentation; NIP-RS:537-539 requires success only when
+ *   resulting override_active == false.
+ */
+export type OverrideReadOutcome = "overrideCleared" | "overrideStillActive";
+
+/**
+ * Attempt to clear the NIP-RS override for `channelId` as part of an explicit
+ * mark-read transition. Models the transition as a single outcome:
+ *
+ *   1. Load not complete (!isReadStateReady): fail-closed —
+ *      return overrideStillActive (null liveness after an incomplete load is
+ *      ambiguous, not known absence).
+ *   2. liveness === null (load complete, no register): return overrideCleared
+ *      (known absence after a complete load; no C-bump needed).
+ *   3. liveness exists (active or inactive): always call markChannelRead —
+ *      spec NIP-RS:537-539 requires the C-bump even when the frontier already
+ *      deactivated the override. Then re-read liveness.
+ *      a. Resulting liveness inactive (or already_inactive race): cleared.
+ *      b. uint32 refusal AND F > B (frontier already deactivated): cleared,
+ *         no error toast (representable C-bump was not possible, frontier
+ *         already provides the inactive verdict).
+ *      c. Other refusal keeping liveness active: show toast, overrideStillActive.
+ */
+export function applyOverrideRead(
+  channelId: string,
+  apis: OverrideAPIs,
+): OverrideReadOutcome {
+  // Step 1: manager unavailable → fail closed.
+  if (!apis.isReadStateReady) return "overrideStillActive";
+
+  const liveness = apis.getOverrideLiveness(channelId);
+
+  // Step 2: load complete but no register at all → known absence, cleared.
+  if (liveness === null) return "overrideCleared";
+
+  // Step 3: register exists — always attempt the C-bump (NIP-RS:537-539).
+  const result = apis.markChannelRead(channelId);
+
+  if (!result.success && result.reason === "already_inactive") {
+    // Race: cleared by another device between our check and the call.
+    return "overrideCleared";
+  }
+
+  // Re-read liveness after the attempt.
+  const afterLiveness = apis.getOverrideLiveness(channelId);
+  // Covers: successful C-bump (inactive after), F>B deactivation (inactive after),
+  // and uint32 refusal where F>B already deactivated the register.
+  if (afterLiveness === null || !afterLiveness.active) {
+    return "overrideCleared";
+  }
+
+  // Refused and still active.
+  if (!result.success) {
+    const msg = overrideErrorMessage("read", result.reason);
+    if (msg) toast.error(msg);
+  }
+  return "overrideStillActive";
+}
+
+/**
+ * Persist `channelId` into the forced-unread local cache with the current own
+ * timestamp as baseline. Always refreshes an existing entry (so a successful
+ * re-mark after the old override died uses the fresh baseline). Returns true
+ * when the map changed.
+ */
+export function persistForcedUnread(
+  channelId: string,
+  forcedMap: ForcedUnreadMap,
+  getOwnTimestamp: (id: string) => number | null,
+  pubkey: string | undefined,
+): boolean {
+  const newTs = getOwnTimestamp(channelId);
+  if (Object.hasOwn(forcedMap, channelId) && forcedMap[channelId] === newTs) {
+    return false;
+  }
+  forcedMap[channelId] = newTs;
+  if (pubkey) forcedUnreadStore.write(pubkey, forcedMap);
+  return true;
+}
+
+/**
+ * Per-channel evidence maps that `markAllChannelsRead` mutates.
+ * Extracted so the per-channel transition can be tested without the full hook.
+ */
+export type MarkAllEvidenceMaps = {
+  forcedUnread: ForcedUnreadMap;
+  latestByChannel: Map<string, number>;
+  observedUnreadEvents: Map<string, Map<string, ObservedUnreadEvent>>;
+};
+
+/**
+ * Apply the per-channel mark-all-read transition for a single channel ID.
+ *
+ * Calls `applyOverrideRead` to attempt the NIP-RS C-bump. On `overrideCleared`,
+ * removes the channel's entry from all three evidence maps. On
+ * `overrideStillActive` (refused or load incomplete), all maps are left
+ * intact — the caller must NOT wipe evidence for channels whose clear was
+ * refused. Returns the outcome for caller use.
+ */
+export function applyMarkAllReadTransition(
+  channelId: string,
+  apis: OverrideAPIs,
+  maps: MarkAllEvidenceMaps,
+): OverrideReadOutcome {
+  const outcome = applyOverrideRead(channelId, apis);
+  if (outcome === "overrideCleared") {
+    delete maps.forcedUnread[channelId];
+    maps.latestByChannel.delete(channelId);
+    maps.observedUnreadEvents.delete(channelId);
+  }
+  return outcome;
+}

desktop/src/features/channels/ui/useChannelUnreadState.ts

@@ -26,6 +26,26 @@ import { isConversationalUnreadKind } from "@/shared/constants/kinds";
 
 import { useWelcomeInitialUnreadSuppression } from "./useWelcomeInitialUnreadSuppression";
 
+/**
+ * Pure helper: apply the mark-unread divider overlay.
+ *
+ * Calls `markFn(channelId)` and, if it returns true (accepted), adds the
+ * channel to the `overlay` set. Returns the same boolean so callers can gate
+ * any additional side-effects on success.
+ *
+ * Extracted for testability: `handleMarkUnread` calls this instead of
+ * inlining the same two-step logic.
+ */
+export function applyMarkUnreadDividerOverlay(
+  channelId: string,
+  markFn: (id: string) => boolean,
+  overlay: Set<string>,
+): boolean {
+  const accepted = markFn(channelId);
+  if (accepted) overlay.add(channelId);
+  return accepted;
+}
+
 type UseChannelUnreadStateOptions = {
   activeChannelId: string | null;
   timelineMessages: TimelineMessage[];
@@ -36,7 +56,7 @@ type UseChannelUnreadStateOptions = {
   openThreadMessages?: MainTimelineEntry[];
   getChannelReadAt: (channelId: string) => number | null;
   getMessageReadAt: (messageId: string) => number | null;
-  markChannelUnread: (channelId: string) => void;
+  markChannelUnread: (channelId: string) => boolean;
   markMessageRead: (messageId: string, timestamp: number) => void;
   isThreadMuted: (rootId: string) => boolean;
   readStateVersion: number;
@@ -404,11 +424,12 @@ export function useChannelUnreadState({
 
   const handleMarkUnread = React.useCallback(() => {
     if (!activeChannelId) return;
-    // Mirror the deliberate mark-unread locally so the timeline marker is
-    // suppressed (see forcedUnreadRef above). Re-render so the memo re-runs.
-    forcedUnreadRef.current.add(activeChannelId);
+    applyMarkUnreadDividerOverlay(
+      activeChannelId,
+      markChannelUnread,
+      forcedUnreadRef.current,
+    );
     forceUnreadRender();
-    markChannelUnread(activeChannelId);
   }, [activeChannelId, markChannelUnread]);
 
   // Mark a message's directly-revealed children read (LP4 v3 open-at-level):

desktop/src/features/channels/useUnreadChannels.ts

@@ -21,6 +21,13 @@ import {
   forcedUnreadStore,
   type ForcedUnreadMap,
 } from "@/features/channels/forcedUnreadStore";
+import {
+  applyMarkAllReadTransition,
+  applyOverrideRead,
+  applyOverrideUnread,
+  persistForcedUnread,
+  type OverrideAPIs,
+} from "@/features/channels/readStateOverride";
 import {
   getThreadReference,
   isBroadcastReply,
@@ -98,14 +105,10 @@ function toUnixSeconds(isoOrMs: string | null | undefined): number | null {
 }
 
 // Resolve where the read marker should land when a channel is marked read.
-// Folds the caller's timeline position together with the newest event this
-// client has observed live (`observedLatest`), so an explicit "mark read" still
-// covers messages that arrived faster than channel metadata — this fold is
-// load-bearing for the Esc shortcut, sidebar mark-read, and empty-channel open,
-// all of which pass a null/stale caller value. `clearObserved` reports whether
-// the resulting marker covers the observed timestamp, signalling the caller to
-// drop its observed refs so the unread memo sees `latest === undefined` until a
-// genuinely newer event arrives.
+// Folds caller position + observed-live latest so explicit "mark read" covers
+// messages that arrived faster than channel metadata (load-bearing for Esc,
+// sidebar mark-read, and empty-channel open). Returns `clearObserved` when the
+// marker covers the observed timestamp so callers can drop stale refs.
 export function resolveChannelReadMarker(
   callerReadAt: string | null | undefined,
   observedLatest: number | undefined,
@@ -147,15 +150,12 @@ export function useUnreadChannels(
   } = options;
   const activeChannelId = activeChannel?.id ?? null;
   const normalizedPubkey = pubkey?.toLowerCase() ?? null;
-  // Scoped relay key for activity storage; empty string when relay not yet known
-  // so rows from an unknown relay never load into the wrong community.
+  // Scoped relay key; empty string when relay not yet known.
   const normalizedRelayUrl = relayUrlOption
     ? normalizeRelayUrl(relayUrlOption)
     : "";
-  // Single identity for the in-memory thread-activity buffer — computed once
-  // per render and used at reset, both writers, and the return fence. The
-  // helper returns "" when either value is absent, which never matches a valid
-  // loaded scope, so the fence returns [] until the buffer is seeded.
+  // Activity scope key: "" when pubkey/relay absent (never matches a valid
+  // loaded scope, so the fence returns [] until the buffer is seeded).
   const currentActivityScope = activityScopeKey(
     normalizedPubkey,
     normalizedRelayUrl,
@@ -164,20 +164,35 @@ export function useUnreadChannels(
   const {
     getEffectiveTimestamp,
     isReady: isReadStateReady,
+    isLoadComplete,
     markContextRead,
     drainSyncedAdvances,
     setContextParentResolver,
     readStateVersion,
     getOwnTimestamp,
+    markChannelUnread: markChannelOverrideUnread,
+    markChannelRead: markChannelOverrideRead,
+    getOverrideLiveness,
+    getProjection,
   } = useReadState(pubkey, relayClient);
+  const overrideApis = React.useMemo<OverrideAPIs>(
+    () => ({
+      isReadStateReady: isLoadComplete, // loadComplete, not initialized — null after incomplete load ≠ known absent
+      markChannelUnread: markChannelOverrideUnread,
+      markChannelRead: markChannelOverrideRead,
+      getOverrideLiveness,
+    }),
+    [
+      getOverrideLiveness,
+      isLoadComplete,
+      markChannelOverrideRead,
+      markChannelOverrideUnread,
+    ],
+  );
 
-  // Observed "latest external trigger event" per channel (unix seconds). This
-  // is *derived relay evidence*, not source-of-truth: it's populated from a
-  // one-shot catch-up REQ per channel (keyed on the NIP-RS read marker) plus
-  // ongoing live events. The only thing we ever do with it is compare against
-  // the NIP-RS read marker — see the unread memo below. Reset on identity
-  // change. Stale entries for channels the user has left are silently
-  // ignored by the memo (it iterates the current channels list, not the map).
+  // Observed "latest external trigger event" per channel — derived relay evidence
+  // used only to compare against the NIP-RS read marker. Reset on identity change;
+  // stale entries for left channels are ignored (memo iterates current channel list).
   const latestByChannelRef = React.useRef(new Map<string, number>());
   const observedUnreadEventsByChannelRef = React.useRef(
     new Map<string, Map<string, ObservedUnreadEvent>>(),
@@ -186,13 +201,10 @@ export function useUnreadChannels(
   const channelsRef = React.useRef(channels);
   channelsRef.current = channels;
 
-  // Channels manually marked unread this session (e.g., right-click → "mark
-  // unread"). Because NIP-RS read markers are monotonic, this flag is what
-  // makes the badge appear without lowering synced read state. Each entry
-  // stores the channel's NIP-RS read marker (unix seconds) at force-time, or
-  // null if no marker existed yet. Persisted to localStorage
-  // (buzz-forced-unread.v1) so it survives reload and is visible to the rail
-  // observer for inactive communities.
+  // Channels manually marked unread this session via right-click (NIP-RS local
+  // override cache). Monotonic read markers mean we can't lower them; this flag
+  // makes the badge appear without affecting synced state. Persisted so it
+  // survives reload and is visible to the rail observer for inactive communities.
   const forcedUnreadRef = React.useRef<ForcedUnreadMap>(
     pubkey ? forcedUnreadStore.read(pubkey) : {},
   );
@@ -217,39 +229,28 @@ export function useUnreadChannels(
     }
   }, [readStateVersion, drainSyncedAdvances]);
 
-  // Root event IDs of threads where the current user has replied at least once.
-  // Used to determine if thread replies should trigger unread notifications.
+  // Root event IDs of threads the current user has replied to (notification gating).
   const participatedRootIdsRef = React.useRef(new Set<string>());
 
   // Root event IDs of top-level messages authored by the current user.
-  // Used to notify the author when someone replies to their posts.
   const authoredRootIdsRef = React.useRef(new Set<string>());
 
-  // Root event IDs of threads where an external message @-mentioned the user.
-  // ORed into the badge gate so a mention recipient who never participated,
-  // authored, or followed the thread still gets the thread-unread badge.
+  // Thread roots with an external @-mention; ORed into badge gate for mention recipients.
   const mentionedRootIdsRef = React.useRef(new Set<string>());
 
-  // Root event IDs of threads the user has explicitly muted. Takes precedence
-  // over participation, follow, and authorship for notification suppression.
+  // Thread roots the user has explicitly muted (takes precedence over all gating).
   const mutedRootIdsRef = React.useRef(new Set<string>());
 
-  // Stable ref for the caller-supplied muted channel IDs. Updated every render
-  // so the catch-up loop always reads the latest set without being a dep.
+  // Caller-supplied muted channel IDs — updated each render, not a dep.
   const mutedChannelIdsRef = React.useRef<ReadonlySet<string>>(new Set());
   mutedChannelIdsRef.current = mutedChannelIdsOption ?? new Set();
 
-  // Thread reply events that triggered notifications — surfaced in the Home
-  // activity feed as synthetic FeedItems.
+  // Thread reply events that triggered notifications (surfaced in Home feed).
   const threadActivityRef = React.useRef<ThreadActivityItem[]>([]);
-  // Tracks the (pubkey:relayUrl) scope currently loaded into threadActivityRef.
-  // Writers guard against this before merging so in-flight writes from a prior
-  // scope cannot corrupt the new one; renders return [] until it matches.
+  // (pubkey:relayUrl) scope in threadActivityRef; guards in-flight scope drift.
   const threadActivityScopeRef = React.useRef<string>("");
 
-  // Tracks which channels we've already issued a catch-up REQ for this
-  // session. Prevents re-fetching on every channels-list refetch, while still
-  // letting newly-joined channels be caught up. Reset on identity change.
+  // Channels already caught-up this session; reset on identity change.
   const caughtUpChannelsRef = React.useRef(new Set<string>());
 
   const [latestVersion, bumpLatestVersion] = React.useReducer(
@@ -257,10 +258,8 @@ export function useUnreadChannels(
     0,
   );
 
-  // Version signal bumped only when the participated/authored/mentioned
-  // root-id sets change, so the gate snapshots (re-derived below) don't
-  // re-allocate on every observed external message the way reusing
-  // latestVersion would.
+  // Separate from latestVersion: bumped only when badge-gating set membership
+  // changes, so gate snapshots don't re-allocate on every external message.
   const [membershipVersion, bumpMembershipVersion] = React.useReducer(
     (x: number) => x + 1,
     0,
@@ -296,29 +295,19 @@ export function useUnreadChannels(
     bumpMembershipVersion();
   }, [pubkey, relayClient, normalizedRelayUrl]);
 
-  // `topLevelOnly` is the passive channel-open path (NIP-RS Option 1): the
-  // caller's `readAt` is already the newest TOP-LEVEL message, so the marker
-  // must land exactly there without folding in `observedLatest` (which counts
-  // thread replies) and without clearing observed refs. Leaving the refs intact
-  // keeps the sidebar dot lit for a channel whose only unread is an unopened
-  // thread reply — viewing the channel no longer absorbs the reply, so the dot
-  // persists until an explicit mark-read (Esc, sidebar, mark-all) or a newer
-  // top-level message advances the channel marker past it. Those explicit
-  // "mark read" actions omit the flag and keep the fold, since they mean
-  // "clear everything in this channel."
+  // `topLevelOnly` = passive channel-open path (NIP-RS Option 1): caller's
+  // `readAt` is already the newest top-level message; skip `observedLatest`
+  // (which counts thread replies) and keep refs intact so the sidebar dot
+  // persists until an explicit mark-read covers the reply or a newer top-level
+  // message advances past it. Explicit "mark read" actions omit the flag.
   const markChannelRead = React.useCallback(
     (
       channelId: string,
       readAt: string | null | undefined,
       { topLevelOnly = false }: { topLevelOnly?: boolean } = {},
     ) => {
-      if (Object.hasOwn(forcedUnreadRef.current, channelId)) {
-        delete forcedUnreadRef.current[channelId];
-        if (pubkey) {
-          forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
-        }
-        bumpLatestVersion();
-      }
+      // Frontier advance first (spec order: advance → C-bump → inspect liveness).
+      // The advance may make F > B, so applyOverrideRead sees an inactive override.
       const observedLatest = topLevelOnly
         ? undefined
         : latestByChannelRef.current.get(channelId);
@@ -326,37 +315,47 @@ export function useUnreadChannels(
         readAt,
         observedLatest,
       );
-      if (markAt === null) return;
-      markContextRead(channelId, markAt);
-      // Clear observed-latest refs when the read marker covers them so the
-      // unread memo sees `latest === undefined` until a genuinely new event
-      // arrives. Without this, `latest > readAt` resolves to `T > T` (false)
-      // but the channel lingers in the set when advanceContext's monotonic
-      // guard suppresses the readStateVersion bump.
-      if (clearObserved) {
+      if (markAt !== null) {
+        markContextRead(channelId, markAt);
+      }
+      // C-bump; clear local presentation only when liveness is confirmed inactive.
+      // null ≠ inactive — pre-init fails closed.
+      const outcome = applyOverrideRead(channelId, overrideApis);
+      if (outcome === "overrideCleared") {
+        if (Object.hasOwn(forcedUnreadRef.current, channelId)) {
+          delete forcedUnreadRef.current[channelId];
+          if (pubkey) {
+            forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
+          }
+          bumpLatestVersion();
+        }
+      }
+      if (markAt !== null && clearObserved) {
         latestByChannelRef.current.delete(channelId);
         observedUnreadEventsByChannelRef.current.delete(channelId);
         bumpLatestVersion();
       }
     },
-    [markContextRead, pubkey],
+    [markContextRead, overrideApis, pubkey],
   );
 
-  // Manually mark a channel unread (e.g., right-click → "mark unread"). Persists
-  // the current NIP-RS read marker as baseline to localStorage so the rail
-  // observer can detect when a cross-device read has since covered the force.
-  // NIP-RS markers are monotonic, so we do not publish a lower timestamp.
+  // Manually mark a channel unread (right-click). Returns true when the manager
+  // accepted (use to gate session-local UI overlays on success only).
   const markChannelUnread = React.useCallback(
-    (channelId: string) => {
-      if (!Object.hasOwn(forcedUnreadRef.current, channelId)) {
-        forcedUnreadRef.current[channelId] = getOwnTimestamp(channelId);
-        if (pubkey) {
-          forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
-        }
+    (channelId: string): boolean => {
+      if (!applyOverrideUnread(channelId, overrideApis)) return false;
+      if (
+        persistForcedUnread(
+          channelId,
+          forcedUnreadRef.current,
+          getOwnTimestamp,
+          pubkey,
+        )
+      )
         bumpLatestVersion();
-      }
+      return true;
     },
-    [getOwnTimestamp, pubkey],
+    [getOwnTimestamp, overrideApis, pubkey],
   );
 
   // Record the thread root of an EXTERNAL message that @-mentioned the user.
@@ -844,7 +843,9 @@ export function useUnreadChannels(
       for (const channel of channels) {
         if (channel.id === activeChannelId) continue;
 
-        if (Object.hasOwn(forcedUnreadRef.current, channel.id)) {
+        // Manager liveness is authoritative (isReadStateReady). Use
+        // getOverrideLiveness, not forcedUnreadRef which may be stale.
+        if (getOverrideLiveness(channel.id)?.active === true) {
           // Forced-unread is dot tier only — not high-priority.
           unread.add(channel.id);
           counts.set(channel.id, 1);
@@ -953,7 +954,6 @@ export function useUnreadChannels(
 
   const markAllChannelsRead = React.useCallback(() => {
     for (const channelId of unreadChannelIdsRef.current) {
-      delete forcedUnreadRef.current[channelId];
       const unixSeconds =
         latestByChannelRef.current.get(channelId) ??
         getEffectiveTimestamp(channelId) ??
@@ -961,19 +961,20 @@ export function useUnreadChannels(
       if (unixSeconds !== null) {
         markContextRead(channelId, unixSeconds);
       }
-      latestByChannelRef.current.delete(channelId);
-      observedUnreadEventsByChannelRef.current.delete(channelId);
+      applyMarkAllReadTransition(channelId, overrideApis, {
+        forcedUnread: forcedUnreadRef.current,
+        latestByChannel: latestByChannelRef.current,
+        observedUnreadEvents: observedUnreadEventsByChannelRef.current,
+      });
     }
     if (pubkey) {
       forcedUnreadStore.write(pubkey, forcedUnreadRef.current);
     }
     bumpLatestVersion();
-  }, [getEffectiveTimestamp, markContextRead, pubkey]);
+  }, [getEffectiveTimestamp, markContextRead, overrideApis, pubkey]);
 
-  // Identity-stable snapshots of the membership sets for the notify gate.
-  // Re-derived only when membershipVersion bumps (a set actually changed), so
-  // `isNotifiedForThread`'s useCallback deps invalidate on async discovery
-  // while live consumers keep reading the mutable refs directly.
+  // Stable snapshots of membership sets — re-derived only when membershipVersion
+  // bumps; live consumers read the mutable refs directly between bumps.
   // biome-ignore lint/correctness/useExhaustiveDependencies: membershipVersion is the intentional re-derivation signal
   const participatedRootIds = React.useMemo(
     () => new Set(participatedRootIdsRef.current) as ReadonlySet<string>,
@@ -998,10 +999,8 @@ export function useUnreadChannels(
     markAllChannelsRead,
     markChannelRead,
     markChannelUnread,
-    // Exposed so other surfaces (e.g. Home) can project per-item read state
-    // off the same NIP-RS read marker without instantiating a second
-    // ReadStateManager. readStateVersion is the invalidation signal callers
-    // should include in memo deps.
+    // Exposed for surfaces projecting per-item read state (Home) off the shared
+    // NIP-RS marker without a second ReadStateManager; readStateVersion = invalidation signal.
     getEffectiveTimestamp,
     getOwnTimestamp,
     readStateVersion,
@@ -1017,5 +1016,6 @@ export function useUnreadChannels(
     mutedRootIds: mutedRootIdsRef.current as ReadonlySet<string>,
     muteThread,
     unmuteThread,
+    getProjection,
   };
 }

desktop/src/features/communities/communityUnreadObserver.test.mjs

@@ -544,104 +544,70 @@ function baseRelay(unreadEvent, mutesPayload = null) {
   ]);
 }
 
-test("fetchCommunityUnread threaded reply in untracked root → hasUnread:false", async () => {
-  const relay = baseRelay(threadedReplyEvent());
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 100,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    // No root in any set → gate rejects the threaded reply
-    readThreadRelationships: readRelationships(),
-  });
-
-  assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
-});
-
-test("fetchCommunityUnread threaded reply in participatedRootIds → hasUnread:true", async () => {
-  const relay = baseRelay(threadedReplyEvent());
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 100,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    readThreadRelationships: readRelationships({
-      participatedRootIds: new Set([THREAD_ROOT_2]),
-    }),
-  });
-
-  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
-});
-
-test("fetchCommunityUnread #p-mention reply in untracked root → hasUnread:true (mention overrides)", async () => {
-  // A @mention of the user bypasses the follow/participation gate
-  const relay = baseRelay(
-    threadedReplyEvent({
+const THREAD_GATE_CASES = [
+  {
+    name: "threaded reply in untracked root → hasUnread:false",
+    unreadEvent: threadedReplyEvent(),
+    relationships: {},
+    expected: { hasUnread: false, mentionCount: 0 },
+  },
+  {
+    name: "threaded reply in participatedRootIds → hasUnread:true",
+    unreadEvent: threadedReplyEvent(),
+    relationships: { participatedRootIds: new Set([THREAD_ROOT_2]) },
+    expected: { hasUnread: true, mentionCount: 0 },
+  },
+  {
+    name: "#p-mention reply in untracked root → hasUnread:true (mention overrides)",
+    unreadEvent: threadedReplyEvent({
       id: "mention-reply".padEnd(64, "0"),
       extraTags: [["p", PUBKEY]],
     }),
-  );
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 100,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    readThreadRelationships: readRelationships(),
-  });
-
-  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
-});
-
-test("fetchCommunityUnread top-level post → hasUnread:true (no thread gate)", async () => {
-  // Top-level posts have no parentId — shouldNotifyForEvent returns true
-  const relay = baseRelay(
-    event({
+    relationships: {},
+    expected: { hasUnread: true, mentionCount: 0 },
+  },
+  {
+    name: "top-level post → hasUnread:true (no thread gate)",
+    unreadEvent: event({
       id: "toplevel".padEnd(64, "0"),
       created_at: 20,
       tags: [["h", CHANNEL_ID]],
     }),
-  );
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 100,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    readThreadRelationships: readRelationships(),
-  });
-
-  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
-});
-
-test("fetchCommunityUnread threaded reply whose root is in mutedRootIds → hasUnread:false", async () => {
-  const relay = baseRelay(
-    threadedReplyEvent({ id: "muted-reply".padEnd(64, "0") }),
-  );
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 100,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    // Root is participated but also muted — mute wins
-    readThreadRelationships: readRelationships({
+    relationships: {},
+    expected: { hasUnread: true, mentionCount: 0 },
+  },
+  {
+    name: "threaded reply whose root is in mutedRootIds → hasUnread:false",
+    unreadEvent: threadedReplyEvent({ id: "muted-reply".padEnd(64, "0") }),
+    relationships: {
       participatedRootIds: new Set([THREAD_ROOT_2]),
       mutedRootIds: new Set([THREAD_ROOT_2]),
-    }),
+    },
+    expected: { hasUnread: false, mentionCount: 0 },
+  },
+];
+
+for (const {
+  name,
+  unreadEvent,
+  relationships,
+  expected,
+} of THREAD_GATE_CASES) {
+  test(`fetchCommunityUnread ${name}`, async () => {
+    const relay = baseRelay(unreadEvent);
+    const result = await fetchCommunityUnread({
+      client: relay,
+      pubkey: PUBKEY,
+      nowSeconds: 100,
+      decryptReadState: async (v) => v,
+      decryptMutes: async (v) => v,
+      readThreadRelationships: readRelationships(relationships),
+    });
+    assert.deepEqual(result, expected);
   });
+}
 
-  assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
-});
-
-// ── Forced-unread persistence gate tests ─────────────────────────────────
+// ── Override register authoritative source tests ──────────────────────────
 
 // A relay that serves one channel (CHANNEL_ID) as a member channel,
 // with no read state and no incoming events.
@@ -727,8 +693,12 @@ function quietRelayWithReadState(readAtSeconds) {
   ]);
 }
 
-test("fetchCommunityUnread forced-unread channel lights rail dot (hasUnread:true)", async () => {
+test("fetchCommunityUnread stored active register lights rail (hasUnread:true)", async () => {
+  // Authoritative source: complete projection with active register (S=1,C=0,B=0 with F=0).
+  // The fetch returns no override events — projection is the authority.
   const relay = quietRelay();
+  // Active register: S=1, C=0, B=0 (baseline), frontier=0 → override_active = true
+  const overrides = new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]);
 
   const result = await fetchCommunityUnread({
     client: relay,
@@ -737,20 +707,22 @@ test("fetchCommunityUnread forced-unread channel lights rail dot (hasUnread:true
     decryptReadState: async (v) => v,
     decryptMutes: async (v) => v,
     readThreadRelationships: readRelationships(),
-    // Forced-unread map: CHANNEL_ID forced when marker was null (no prior read)
-    readForcedUnread: () => ({ [CHANNEL_ID]: null }),
+    getProjection: () => ({
+      loadComplete: true,
+      frontiers: new Map(),
+      overrides,
+    }),
   });
 
   assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
 });
 
-test("fetchCommunityUnread forced-unread channel not in member list → hasUnread:false", async () => {
-  // The channel is marked forced-unread but the user is not a member of ANY
-  // channel — forced-unread is not consulted when the channel set is empty.
-  const relay = relayFor([
-    // 1. member events — empty
-    () => [],
-  ]);
+test("fetchCommunityUnread stored tombstoned register does NOT light rail (hasUnread:false)", async () => {
+  // Authoritative source: complete projection with tombstoned register (S=1,C=1 → inactive).
+  // Must not light the dot.
+  const relay = quietRelay();
+  // Tombstoned: S=1, C=1, B=0 → override_active = false (S > C fails: 1 > 1 = false)
+  const overrides = new Map([[CHANNEL_ID, { s: 1, c: 1, b: 0 }]]);
 
   const result = await fetchCommunityUnread({
     client: relay,
@@ -759,13 +731,46 @@ test("fetchCommunityUnread forced-unread channel not in member list → hasUnrea
     decryptReadState: async (v) => v,
     decryptMutes: async (v) => v,
     readThreadRelationships: readRelationships(),
-    readForcedUnread: () => ({ [CHANNEL_ID]: null }),
+    getProjection: () => ({
+      loadComplete: true,
+      frontiers: new Map(),
+      overrides,
+    }),
   });
 
   assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
 });
 
-test("fetchCommunityUnread forced-unread channel that is also muted → hasUnread:false", async () => {
+test("fetchCommunityUnread old active register beyond horizon still lights rail", async () => {
+  // An active register that is older than the 7-day horizon must still light
+  // the rail. The fetch is tag-free and has no `since`, so it returns no
+  // events here, but the complete projection is the authority.
+  const relay = quietRelay();
+  const VERY_OLD_FRONTIER = 1; // far beyond any 7-day horizon
+  // S=1, C=0, B=VERY_OLD_FRONTIER, and frontier=VERY_OLD_FRONTIER → active
+  const overrides = new Map([
+    [CHANNEL_ID, { s: 1, c: 0, b: VERY_OLD_FRONTIER }],
+  ]);
+
+  const result = await fetchCommunityUnread({
+    client: relay,
+    pubkey: PUBKEY,
+    nowSeconds: 100 + 7 * 24 * 3600 + 1, // past the old 7-day cutoff
+    decryptReadState: async (v) => v,
+    decryptMutes: async (v) => v,
+    readThreadRelationships: readRelationships(),
+    getProjection: () => ({
+      loadComplete: true,
+      frontiers: new Map(),
+      overrides,
+    }),
+  });
+
+  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
+});
+
+test("fetchCommunityUnread no stored register and empty fetch → falls through to relay gate", async () => {
+  // No stored registers, no override event fetched, but a real unread event exists.
   const relay = relayFor([
     // 1. member events
     () => [
@@ -787,16 +792,20 @@ test("fetchCommunityUnread forced-unread channel that is also muted → hasUnrea
     ],
     // 3. visibility
     () => [],
-    // 4. read-state (parallel with mutes)
+    // 4. read-state — empty
     () => [],
-    // 5. mutes — CHANNEL_ID is muted
+    // 5. mutes
+    () => [],
+    // 6. unread events — one real unread
     () => [
       event({
-        pubkey: PUBKEY,
-        content: mutesContent([CHANNEL_ID]),
+        id: "real-unread".padEnd(64, "0"),
+        created_at: 20,
+        tags: [["h", CHANNEL_ID]],
       }),
     ],
-    // No per-channel fetches expected — muted channel is skipped
+    // 7. mention events
+    () => [],
   ]);
 
   const result = await fetchCommunityUnread({
@@ -806,15 +815,37 @@ test("fetchCommunityUnread forced-unread channel that is also muted → hasUnrea
     decryptReadState: async (v) => v,
     decryptMutes: async (v) => v,
     readThreadRelationships: readRelationships(),
-    // CHANNEL_ID is both forced-unread AND muted — mute wins
-    readForcedUnread: () => ({ [CHANNEL_ID]: null }),
+    // No projection — no override registers; falls through to relay evidence gate
+  });
+
+  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
+});
+
+test("fetchCommunityUnread channel not in member list → hasUnread:false", async () => {
+  // No members — register lookup is never reached.
+  const relay = relayFor([
+    () => [], // member events empty
+  ]);
+
+  const result = await fetchCommunityUnread({
+    client: relay,
+    pubkey: PUBKEY,
+    nowSeconds: 100,
+    decryptReadState: async (v) => v,
+    decryptMutes: async (v) => v,
+    readThreadRelationships: readRelationships(),
+    getProjection: () => ({
+      loadComplete: true,
+      frontiers: new Map(),
+      overrides: new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]),
+    }),
   });
 
   assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
 });
 
-test("fetchCommunityUnread readForcedUnread returns empty map → falls through to relay gate", async () => {
-  // No forced-unread, but there IS a real unread event → hasUnread:true via relay
+test("fetchCommunityUnread active register muted channel → hasUnread:false", async () => {
+  // Override is active, but the channel is muted — mute wins.
   const relay = relayFor([
     // 1. member events
     () => [
@@ -838,18 +869,14 @@ test("fetchCommunityUnread readForcedUnread returns empty map → falls through
     () => [],
     // 4. read-state
     () => [],
-    // 5. mutes
-    () => [],
-    // 6. unread events
+    // 5. mutes — CHANNEL_ID is muted
     () => [
       event({
-        id: "real-unread".padEnd(64, "0"),
-        created_at: 20,
-        tags: [["h", CHANNEL_ID]],
+        pubkey: PUBKEY,
+        content: mutesContent([CHANNEL_ID]),
       }),
     ],
-    // 7. mention events
-    () => [],
+    // No per-channel fetches expected — muted channel is skipped
   ]);
 
   const result = await fetchCommunityUnread({
@@ -859,33 +886,23 @@ test("fetchCommunityUnread readForcedUnread returns empty map → falls through
     decryptReadState: async (v) => v,
     decryptMutes: async (v) => v,
     readThreadRelationships: readRelationships(),
-    readForcedUnread: () => ({}), // empty — no forced-unread
-  });
-
-  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
-});
-
-test("fetchCommunityUnread forced-unread + synced marker advanced PAST baseline → hasUnread:false", async () => {
-  // markerAtWhenForced = 50, observed readAt = 100 (100 > 50 → cross-device read wins)
-  const relay = quietRelayWithReadState(100);
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 200,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    readThreadRelationships: readRelationships(),
-    // Forced at marker=50; synced marker is now 100 — covers the force
-    readForcedUnread: () => ({ [CHANNEL_ID]: 50 }),
+    // Active register, but channel is muted — mute wins
+    getProjection: () => ({
+      loadComplete: true,
+      frontiers: new Map(),
+      overrides: new Map([[CHANNEL_ID, { s: 1, c: 0, b: 0 }]]),
+    }),
   });
 
   assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
 });
 
-test("fetchCommunityUnread forced-unread + synced marker NOT advanced past baseline → hasUnread:true", async () => {
-  // markerAtWhenForced = 100, observed readAt = 100 (equal → force still stands)
+test("fetchCommunityUnread frontier advance deactivates stored register → hasUnread:false", async () => {
+  // Register: S=1, C=0, B=50 (active only when F<=50).
+  // Fetched frontier: 100 > 50 → override_active = false.
   const relay = quietRelayWithReadState(100);
+  // B=50 means active only when F<=50; fetched frontier is 100 → inactive
+  const overrides = new Map([[CHANNEL_ID, { s: 1, c: 0, b: 50 }]]);
 
   const result = await fetchCommunityUnread({
     client: relay,
@@ -894,28 +911,281 @@ test("fetchCommunityUnread forced-unread + synced marker NOT advanced past basel
     decryptReadState: async (v) => v,
     decryptMutes: async (v) => v,
     readThreadRelationships: readRelationships(),
-    // Forced at marker=100; synced marker is still 100 — no newer read
-    readForcedUnread: () => ({ [CHANNEL_ID]: 100 }),
-  });
-
-  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
-});
-
-test("fetchCommunityUnread forced-unread with null baseline + synced marker present → hasUnread:false", async () => {
-  // markerAtWhenForced = null (no marker at force-time), but readAt = 50 now
-  // A cross-device read appeared after the force → do NOT light the dot
-  const relay = quietRelayWithReadState(50);
-
-  const result = await fetchCommunityUnread({
-    client: relay,
-    pubkey: PUBKEY,
-    nowSeconds: 200,
-    decryptReadState: async (v) => v,
-    decryptMutes: async (v) => v,
-    readThreadRelationships: readRelationships(),
-    // Forced when no marker existed; a cross-device read has since appeared
-    readForcedUnread: () => ({ [CHANNEL_ID]: null }),
+    getProjection: () => ({
+      loadComplete: true,
+      frontiers: new Map(),
+      overrides,
+    }),
   });
 
   assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
 });
+
+// ── Adversarial witness tests ─────────────────────────────────────────────────
+//
+// These witnesses verify that the rail's override evaluation cannot resurrect
+// a superseded register or produce a false verdict from a torn/partial state.
+
+test("fetchCommunityUnread adversarial-1: stale live register + fetched tombstone → dark when no complete projection", async () => {
+  // The OLD per-field max join would: max(stale_live, tombstone) = live (resurrection).
+  // Without a complete projection, the fetched coordinate-deduped tombstone is
+  // authoritative; the register is inactive and the rail must remain dark.
+  //
+  // Setup: fetched event carries tombstone (S=0,C=4,B=0) for CHANNEL_ID.
+  const SLOT = "a".repeat(32); // valid 32-hex slot
+  const relay = relayFor([
+    // 1. member events
+    () => [
+      event({
+        tags: [
+          ["d", CHANNEL_ID],
+          ["p", PUBKEY],
+        ],
+      }),
+    ],
+    // 2. metadata
+    () => [
+      event({
+        tags: [
+          ["d", CHANNEL_ID],
+          ["t", "stream"],
+        ],
+      }),
+    ],
+    // 3. visibility
+    () => [],
+    // 4. read-state: tombstone event (S=0, C=4, B=0 → inactive, S not > C)
+    () => [
+      event({
+        pubkey: PUBKEY,
+        created_at: 300,
+        tags: [
+          ["d", `read-state:${SLOT}`],
+          ["t", "read-state"],
+        ],
+        // Contexts blob: channel has tombstone register (ov_s=0 absent means S=0, ov_c=4)
+        content: JSON.stringify({
+          v: 1,
+          client_id: "client",
+          contexts: {
+            [CHANNEL_ID]: 50,
+            [`ov_c:${CHANNEL_ID}`]: 4,
+          },
+        }),
+      }),
+    ],
+    // 5. mutes
+    () => [],
+    // 6. unread events — none (the override is inactive, no message events either)
+    () => [],
+    // 7. mention events
+    () => [],
+  ]);
+
+  // No complete projection — use fetched-only (no stored merge).
+  // Previously this would be called with readStoredRegisters returning a stale
+  // live register (S=5,C=0,B=100), which when max-joined with fetched tombstone
+  // (S=0,C=4,B=0) produced (S=5,C=4,B=100) → active (resurrection).
+  // New behavior: fetched state alone → (S=0,C=4,B=0) → inactive → dark.
+  const result = await fetchCommunityUnread({
+    client: relay,
+    pubkey: PUBKEY,
+    nowSeconds: 400,
+    decryptReadState: async (v) => v,
+    decryptMutes: async (v) => v,
+    readThreadRelationships: readRelationships(),
+    // No getProjection: fetched-deduped is the sole authority.
+  });
+
+  assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
+});
+
+test("fetchCommunityUnread adversarial-2: projection with frontier deactivates register → dark", async () => {
+  // CRITICAL 2 fix: the rail now reads projection.frontiers, not only fetched frontiers.
+  // A register (S=5,C=0,B=100) with projection frontier F=101 must evaluate as
+  // inactive (101 > 100) — previously the frontier was not in the stored projection,
+  // so it defaulted to F=0 and the register appeared active.
+  //
+  // The fetched relay state returns empty read-state (frontier not in fetched events).
+  // The projection has the committed frontier F=101 from a successful mark-read.
+  const relay = quietRelay(); // no read-state events fetched
+
+  const overrides = new Map([[CHANNEL_ID, { s: 5, c: 0, b: 100 }]]);
+  // projection.frontiers has F=101 for CHANNEL_ID → deactivates the register
+  const frontiers = new Map([[CHANNEL_ID, 101]]);
+
+  const result = await fetchCommunityUnread({
+    client: relay,
+    pubkey: PUBKEY,
+    nowSeconds: 200,
+    decryptReadState: async (v) => v,
+    decryptMutes: async (v) => v,
+    readThreadRelationships: readRelationships(),
+    getProjection: () => ({ loadComplete: true, frontiers, overrides }),
+  });
+
+  assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
+});
+
+test("fetchCommunityUnread adversarial-5: complete_projection_live_register_not_darkened_by_stale_fetched_tombstone", async () => {
+  // A fresh local mark-unread committed into the projection (S=5,C=0,B=100,F=50)
+  // must NOT be darkened by an independently fetched tombstone (S=0,C=4,B=0).
+  //
+  // The observer cannot adjudicate cross-source ordering: the manager owns
+  // coordinate dedupe and CRDT ingest.  A categorical tombstone-wins rule would
+  // hide the user's mark-unread until the relay publish catches up (the
+  // false-negative direction), which is the more harmful failure.
+  //
+  // When the projection is complete, projection.overrides is the SOLE liveness
+  // authority.  The fetched register shape is irrelevant.  Any transient
+  // false-positive resolves when the manager ingests the tombstone via its own
+  // CRDT path.
+  const SLOT = "a".repeat(32); // valid 32-hex slot
+  const relay = relayFor([
+    // 1. member events
+    () => [
+      event({
+        tags: [
+          ["d", CHANNEL_ID],
+          ["p", PUBKEY],
+        ],
+      }),
+    ],
+    // 2. metadata events
+    () => [
+      event({
+        tags: [
+          ["d", CHANNEL_ID],
+          ["t", "stream"],
+        ],
+      }),
+    ],
+    // 3. visibility events
+    () => [],
+    // 4. read-state: tombstone wire format (ov_c present, ov_s absent → s=0,c=4,b=0)
+    //    Frontier F=50 is also present so the projection frontier is not strictly
+    //    newer — only the live-register vs concurrent-tombstone ordering is under test.
+    () => [
+      event({
+        pubkey: PUBKEY,
+        created_at: 400,
+        tags: [
+          ["d", `read-state:${SLOT}`],
+          ["t", "read-state"],
+        ],
+        content: JSON.stringify({
+          v: 1,
+          client_id: "client",
+          contexts: {
+            [CHANNEL_ID]: 50,
+            [`ov_c:${CHANNEL_ID}`]: 4,
+            // ov_s: intentionally absent → tombstone (s=0)
+          },
+        }),
+      }),
+    ],
+    // 5. mutes
+    () => [],
+    // 6. unread events — none (liveness from override register, not messages)
+    () => [],
+    // 7. mention events — none
+    () => [],
+  ]);
+
+  // Complete projection: live register S=5,C=0,B=100 and frontier F=50.
+  // F=50 ≤ B=100, S=5 > C=0 → isOverrideActive returns true.
+  const overrides = new Map([[CHANNEL_ID, { s: 5, c: 0, b: 100 }]]);
+  const frontiers = new Map([[CHANNEL_ID, 50]]);
+
+  const result = await fetchCommunityUnread({
+    client: relay,
+    pubkey: PUBKEY,
+    nowSeconds: 500,
+    decryptReadState: async (v) => v,
+    decryptMutes: async (v) => v,
+    readThreadRelationships: readRelationships(),
+    getProjection: () => ({ loadComplete: true, frontiers, overrides }),
+  });
+
+  // Projection is the sole override authority when complete.  The fetched
+  // tombstone does not suppress the live projection register — the rail lights.
+  assert.deepEqual(result, { hasUnread: true, mentionCount: 0 });
+});
+
+test("fetchCommunityUnread adversarial-6: torn fetched register (ov_s+ov_c without ov_b) → rail stays dark", async () => {
+  // Integration witness: a fetched read-state event carrying an illegal partial
+  // override group — ov_s and ov_c present, ov_b ABSENT — must not light the
+  // rail.  The parser drops incomplete groups (any shape other than all-three or
+  // C-only tombstone), so the register never reaches liveness evaluation.
+  //
+  // This locks the parser → structured merge → rail path at the observer level.
+  // The readStateOverride parser-matrix tests confirm the drop at the unit level;
+  // this witness confirms the integration: a torn fetched group cannot influence
+  // fetchCommunityUnread() rail output.
+  const SLOT = "b".repeat(32); // valid 32-hex slot, distinct from other tests
+  const relay = relayFor([
+    // 1. member events
+    () => [
+      event({
+        tags: [
+          ["d", CHANNEL_ID],
+          ["p", PUBKEY],
+        ],
+      }),
+    ],
+    // 2. metadata events
+    () => [
+      event({
+        tags: [
+          ["d", CHANNEL_ID],
+          ["t", "stream"],
+        ],
+      }),
+    ],
+    // 3. visibility events
+    () => [],
+    // 4. read-state: torn group — ov_s + ov_c present, ov_b absent (S+C shape).
+    //    The parser treats any shape other than {S,C,B} or {C-only tombstone}
+    //    as invalid and drops the entire group.  No override register reaches
+    //    the authoritative map.  No message events follow either.
+    () => [
+      event({
+        pubkey: PUBKEY,
+        created_at: 300,
+        tags: [
+          ["d", `read-state:${SLOT}`],
+          ["t", "read-state"],
+        ],
+        content: JSON.stringify({
+          v: 1,
+          client_id: "client",
+          contexts: {
+            [`ov_s:${CHANNEL_ID}`]: 5,
+            [`ov_c:${CHANNEL_ID}`]: 0,
+            // ov_b: intentionally absent → illegal S+C partial group, dropped
+          },
+        }),
+      }),
+    ],
+    // 5. mutes
+    () => [],
+    // 6. unread events — none
+    () => [],
+    // 7. mention events — none
+    () => [],
+  ]);
+
+  const result = await fetchCommunityUnread({
+    client: relay,
+    pubkey: PUBKEY,
+    nowSeconds: 400,
+    decryptReadState: async (v) => v,
+    decryptMutes: async (v) => v,
+    readThreadRelationships: readRelationships(),
+    // No getProjection — fetched-deduped is sole authority, torn group is dropped.
+  });
+
+  // Torn group is silently dropped by the parser; no override register present;
+  // no message events — rail must remain dark.
+  assert.deepEqual(result, { hasUnread: false, mentionCount: 0 });
+});

desktop/src/features/communities/communityUnreadObserver.ts

@@ -1,14 +1,14 @@
 import { makeRootIdStore } from "@/features/channels/unreadRootIdStore";
-import {
-  forcedUnreadStore,
-  type ForcedUnreadMap,
-} from "@/features/channels/forcedUnreadStore";
 import { DM_NOTIFIABLE_EVENT_KINDS } from "@/features/channels/isDmNotifiableKind";
-import { mergeReadStateEvents } from "@/features/channels/readState/readStateSnapshot";
+import { mergeReadStateEventsStructured } from "@/features/channels/readState/readStateSnapshot";
 import {
+  isOverrideActive,
   maxReadAt,
   msgContextKey,
+  type OverrideRegister,
 } from "@/features/channels/readState/readStateFormat";
+import { deduplicateByCoordinate } from "@/features/channels/readState/readStateFencedLoader";
+import type { ReadStateProjection } from "@/features/channels/readState/readStateManager";
 import {
   getThreadReference,
   isBroadcastReply,
@@ -86,7 +86,6 @@ const METADATA_LIMIT = 1000;
 const UNREAD_EXISTENCE_LIMIT = 50;
 const MENTION_COUNT_LIMIT = 100;
 const READ_STATE_FETCH_LIMIT = 500;
-const READ_STATE_HORIZON_SECONDS = 7 * 24 * 60 * 60;
 
 export type CommunityUnreadObserverResult = {
   hasUnread: boolean;
@@ -146,9 +145,10 @@ export async function fetchObservedChannels(
 export async function pollCommunityUnread(
   community: Community,
   pubkey: string,
+  getProjection?: () => ReadStateProjection | null,
 ): Promise<CommunityUnreadObserverResult> {
   return withReadOnlyRelayClient(community.relayUrl, (client) =>
-    fetchCommunityUnread({ client, pubkey }),
+    fetchCommunityUnread({ client, pubkey, getProjection }),
   );
 }
 
@@ -159,16 +159,22 @@ export async function fetchCommunityUnread(args: {
   decryptReadState?: (ciphertext: string) => Promise<string>;
   decryptMutes?: (ciphertext: string) => Promise<string>;
   readThreadRelationships?: (pubkey: string) => ThreadRelationships;
-  readForcedUnread?: (pubkey: string) => ForcedUnreadMap;
+  /** Coherent manager projection for override evaluation. When provided and
+   *  `loadComplete` is true, the projection's `overrides` and `frontiers` are
+   *  the authoritative source — the fetched registers are still used for
+   *  frontier-only readAt, but override liveness is decided solely from the
+   *  projection (no per-field max join with fetched registers).
+   *
+   *  When null or `loadComplete` is false, use fetched coordinate-deduped
+   *  registers only — do not fall back to a partial stored view whose
+   *  per-field merge can resurrect a superseded register. */
+  getProjection?: () => ReadStateProjection | null;
 }): Promise<CommunityUnreadObserverResult> {
   const { client, pubkey } = args;
   const normalizedPubkey = pubkey.toLowerCase();
-  const nowSeconds = args.nowSeconds ?? Math.floor(Date.now() / 1_000);
   const decryptMutes = args.decryptMutes ?? nip44DecryptFromSelf;
   const readRelationships =
     args.readThreadRelationships ?? defaultReadThreadRelationships;
-  const readForcedUnread =
-    args.readForcedUnread ?? ((pk) => forcedUnreadStore.read(pk));
 
   const channels = await fetchObservedChannels(client, pubkey);
   if (channels.length === 0) {
@@ -176,11 +182,11 @@ export async function fetchCommunityUnread(args: {
   }
 
   const [readStateEvents, mutesEvents] = await Promise.all([
+    // Override state is exempt from finite-horizon fetching: a register may be
+    // older than seven days and must not be missed. Tag-free, no `since` filter.
     client.fetchEvents({
       kinds: [KIND_READ_STATE],
       authors: [pubkey],
-      "#t": ["read-state"],
-      since: nowSeconds - READ_STATE_HORIZON_SECONDS,
       limit: READ_STATE_FETCH_LIMIT,
     }),
     client.fetchEvents({
@@ -191,12 +197,69 @@ export async function fetchCommunityUnread(args: {
     }),
   ]);
 
-  const readState = await mergeReadStateEvents(
-    readStateEvents,
+  // Authoritative source (b): coordinate-deduped registers decoded from fetched
+  // events. deduplicateByCoordinate retains greatest created_at, lowest id.
+  const readState = await mergeReadStateEventsStructured(
+    deduplicateByCoordinate(readStateEvents),
     pubkey,
     args.decryptReadState,
   );
 
+  // Override liveness source: the manager projection when complete, otherwise
+  // the fetched state only. Never per-field max the two — a projection register
+  // and a fetched register for the same coordinate are NOT concurrent CRDT
+  // replicas; the fetched register may be a newer tombstone superseding a stale
+  // projection entry, and joining them per-field would resurrect the dead one.
+  //
+  // When the projection is complete (loadComplete=true) it is the most
+  // authoritative source because it includes committed local override actions.
+  // Use projection.overrides as the sole override verdict source. For frontiers,
+  // take max(projection, fetched) per channel so a locally-committed frontier
+  // advance is honoured even if the relay hasn't seen it yet.
+  //
+  // When no complete projection is available, use fetched-deduped state alone.
+  const projection = args.getProjection?.() ?? null;
+  const completeProjection = projection?.loadComplete ? projection : null;
+
+  // The observer must not adjudicate cross-source ordering it cannot see.
+  // The manager owns event-level coordinate dedupe and CRDT ingest; a fetched
+  // relay snapshot carries no coordinate/version provenance that would let the
+  // UI compare it against a locally-committed projection register.
+  //
+  // Therefore: when a complete projection is available, projection.overrides is
+  // the SOLE override-liveness authority — no exceptions for fetched tombstone
+  // shapes. A transient false-positive (e.g. projection live register vs stale
+  // fetched tombstone) resolves when the manager ingests the tombstone through
+  // its own CRDT path. A categorical tombstone-wins rule would produce the
+  // inverse false-negative: hiding a user's fresh local mark-unread until the
+  // relay publish catches up, which is the more harmful failure.
+  const authoritative: ReadonlyMap<string, OverrideRegister> =
+    completeProjection !== null
+      ? completeProjection.overrides
+      : readState.overrides;
+
+  // Effective frontier: max of fetched and (when available) projection frontier.
+  const effectiveFrontier = (channelId: string): number | null => {
+    const fetched = readState.frontiers.get(channelId) ?? null;
+    if (completeProjection === null) return fetched;
+    const projected = completeProjection.frontiers.get(channelId) ?? null;
+    if (fetched === null) return projected;
+    if (projected === null) return fetched;
+    return Math.max(fetched, projected);
+  };
+
+  // Build a unified frontier map (fetched + projection max) for per-event
+  // msg/thread context lookups inside isUnreadExternalEvent.
+  const effectiveFrontierMap: ReadonlyMap<string, number> = (() => {
+    if (completeProjection === null) return readState.frontiers;
+    const merged = new Map<string, number>(readState.frontiers);
+    for (const [ctx, pf] of completeProjection.frontiers) {
+      const existing = merged.get(ctx);
+      merged.set(ctx, existing !== undefined ? Math.max(existing, pf) : pf);
+    }
+    return merged;
+  })();
+
   let mutedIds = new Set<string>();
   if (mutesEvents.length > 0) {
     try {
@@ -217,32 +280,21 @@ export async function fetchCommunityUnread(args: {
     mutedRootIds,
   } = readRelationships(normalizedPubkey);
 
-  // Channels manually marked unread on this device. Stored as a record of
-  // { channelId: markerAtWhenForced } so the observer can gate the dot on
-  // whether a cross-device read has since advanced past the stored baseline.
-  const forcedUnreadMap = readForcedUnread(normalizedPubkey);
-
   let hasUnread = false;
   let mentionCount = 0;
 
   for (const channel of channels) {
     if (mutedIds.has(channel.id)) continue;
 
-    // Compute readAt first so the forced-unread gate can compare against it.
-    const readAt = readState.get(channel.id) ?? null;
+    // Compute readAt from effective frontiers (max of fetched + projection).
+    const readAt = effectiveFrontier(channel.id);
 
-    // Forced-unread lights the dot without a relay fetch, but only if the
-    // synced read marker has NOT advanced past the stored baseline. This
-    // prevents stale forced-unread from lighting the rail after a cross-device
-    // read has covered the channel (the drain path in useUnreadChannels only
-    // runs while the community is active, so the store may not be pruned for
-    // inactive communities).
-    if (!hasUnread && Object.hasOwn(forcedUnreadMap, channel.id)) {
-      const markerAtWhenForced = forcedUnreadMap[channel.id];
-      if (
-        readAt === null ||
-        (markerAtWhenForced !== null && readAt <= markerAtWhenForced)
-      ) {
+    // Override liveness: evaluate the authoritative register against the
+    // effective frontier. Uses projection.overrides when complete (no
+    // per-field max with fetched); fetched-deduped otherwise.
+    if (!hasUnread) {
+      const reg = authoritative.get(channel.id);
+      if (reg !== undefined && isOverrideActive(reg, readAt ?? 0)) {
         hasUnread = true;
       }
     }
@@ -274,7 +326,12 @@ export async function fetchCommunityUnread(args: {
     if (!hasUnread) {
       hasUnread = unreadEvents.some(
         (event) =>
-          isUnreadExternalEvent(event, readState, readAt, normalizedPubkey) &&
+          isUnreadExternalEvent(
+            event,
+            effectiveFrontierMap,
+            readAt,
+            normalizedPubkey,
+          ) &&
           shouldNotifyForEvent(event, normalizedPubkey, {
             participatedRootIds,
             followedRootIds,
@@ -287,7 +344,12 @@ export async function fetchCommunityUnread(args: {
     }
 
     mentionCount += mentionEvents.filter((event) =>
-      isUnreadExternalEvent(event, readState, readAt, normalizedPubkey),
+      isUnreadExternalEvent(
+        event,
+        effectiveFrontierMap,
+        readAt,
+        normalizedPubkey,
+      ),
     ).length;
   }
 

desktop/src/features/communities/useCommunityUnread.ts

@@ -3,6 +3,7 @@ import * as React from "react";
 import { getIdentity } from "@/shared/api/tauriIdentity";
 import { markCommunityRead } from "@/features/communities/communityMarkRead";
 import { pollCommunityUnread } from "@/features/communities/communityUnreadObserver";
+import { useAppShell } from "@/app/AppShellContext";
 
 import type { Community } from "./types";
 
@@ -54,6 +55,7 @@ export function useCommunityUnread(
   unreadByCommunity: Record<string, CommunityUnreadState>;
   markCommunityRead: (communityId: string) => Promise<void>;
 } {
+  const { getProjection } = useAppShell();
   const [unreadByCommunity, setUnreadByCommunity] = React.useState<
     Record<string, CommunityUnreadState>
   >(() => seedCommunityStates(communities, {}));
@@ -131,7 +133,11 @@ export function useCommunityUnread(
         if (cancelled) return;
         markLoading(community.id);
         try {
-          const result = await pollCommunityUnread(community, pubkey);
+          const result = await pollCommunityUnread(
+            community,
+            pubkey,
+            getProjection,
+          );
           if (cancelled) return;
           markReady(community.id, result);
         } catch (error) {
@@ -155,7 +161,7 @@ export function useCommunityUnread(
         window.clearTimeout(pollTimer);
       }
     };
-  }, [activeCommunityId, communities]);
+  }, [activeCommunityId, communities, getProjection]);
 
   const communitiesRef = React.useRef(communities);
   communitiesRef.current = communities;

desktop/src/shared/api/relayClientSession.ts

@@ -14,7 +14,9 @@ import {
 } from "@/shared/constants/kinds";
 import {
   getTextPayload,
+  createFencedSubscription,
   type ConnectionState,
+  type FenceHandle,
   type PendingEvent,
   type RelaySubscription,
   type RelaySubscriptionFilter,
@@ -145,7 +147,10 @@ export class RelayClient {
     }
 
     for (const [subId, sub] of this.subscriptions) {
-      if (sub.mode !== "live") {
+      if (sub.mode === "fenced") {
+        sub.lapsed = true;
+        sub.resolveEstablished();
+      } else if (sub.mode !== "live") {
         window.clearTimeout(sub.timeout);
         sub.reject(error);
       } else {
@@ -413,6 +418,25 @@ export class RelayClient {
     return this.subscribe(filter, onEvent);
   }
 
+  /** @see `createFencedSubscription` in relayClientShared for contract. */
+  async subscribeFenced(
+    filter: RelaySubscriptionFilter,
+    onEvent: (event: RelayEvent) => void,
+  ): Promise<FenceHandle> {
+    await this.ensureConnected();
+    const deps = {
+      connectionGeneration: () => this.connectionGeneration,
+      subscriptions: this.subscriptions,
+      sendReq: (id: string, f: RelaySubscriptionFilter) =>
+        this.sendRawWithReconnectRetry(
+          ["REQ", id, f],
+          "Failed to establish fenced subscription.",
+        ),
+      closeSub: (id: string) => this.closeSubscription(id),
+    };
+    return createFencedSubscription(deps, filter, onEvent);
+  }
+
   async subscribeToChannelMentionEvents(
     channelId: string,
     pubkey: string,
@@ -425,10 +449,7 @@ export class RelayClient {
   }
 
   async preconnect() {
-    // Explicit re-engagement. If the session went terminal (auth rejection)
-    // the caller is asking us to try again, so clear the latch. A manual
-    // reconnect also bypasses the current delay once; ordinary operations do
-    // not, so background traffic cannot continuously defeat backoff.
+    // Re-engage after terminal state; bypasses current delay once.
     this.terminal = false;
     this.keepAliveRequested = true;
     if (this.reconnectTimeout !== null) {
@@ -448,33 +469,25 @@ export class RelayClient {
 
   subscribeToReconnects(listener: () => void) {
     this.reconnectListeners.add(listener);
-
     return () => {
       this.reconnectListeners.delete(listener);
     };
   }
-
+  getConnectionGeneration(): number {
+    return this.connectionGeneration;
+  }
   /** Current connection state — synchronous read. */
   getConnectionState(): ConnectionState {
     return this.connectionStateEmitter.get();
   }
-
-  /**
-   * Subscribe to connection-state transitions. The listener is invoked
-   * immediately with the current state so callers don't need a separate
-   * `getConnectionState()` call to seed their UI.
-   */
+  /** Subscribe to connection-state transitions; listener fires immediately with current state. */
   subscribeToConnectionState(listener: (state: ConnectionState) => void) {
     return this.connectionStateEmitter.subscribe(listener);
   }
 
   private async ensureConnected() {
     if (shouldRefuseConnect({ terminal: this.terminal })) {
-      // Session is terminal (e.g. relay rejected auth). Refuse to connect
-      // until an explicit re-engagement (disconnect()/preconnect()) clears
-      // the flag. Without this, the reconnect timer's catch handler — and
-      // the retry wrappers in publishEvent / sendRawWithReconnectRetry —
-      // would race the terminal "disconnected" state back to "reconnecting".
+      // Terminal after auth rejection — refuse until preconnect() re-engages.
       throw new Error("Relay session is terminal; cannot reconnect.");
     }
 
@@ -491,9 +504,7 @@ export class RelayClient {
         hasPendingReconnect: this.reconnectTimeout !== null,
       })
     ) {
-      // The reconnect coordinator owns outage pacing. Query, publish, and
-      // subscription callers must wait for its scheduled attempt instead of
-      // clearing the timer and creating an immediate reconnect storm.
+      // Reconnect coordinator owns pacing; wait rather than creating a storm.
       return this.waitForScheduledReconnect();
     }
 
@@ -633,23 +644,16 @@ export class RelayClient {
   }
 
   private async sendRaw(payload: unknown[]) {
-    if (this.wsId === null) {
-      throw new Error("Relay socket is not connected.");
-    }
-
+    if (this.wsId === null) throw new Error("Relay socket is not connected.");
     await invoke("plugin:websocket|send", {
       id: this.wsId,
-      message: {
-        type: "Text",
-        data: JSON.stringify(payload),
-      },
+      message: { type: "Text", data: JSON.stringify(payload) },
     });
   }
 
   private normalizeRelayError(error: unknown, fallbackMessage: string) {
     return error instanceof Error ? error : new Error(fallbackMessage);
   }
-
   private recoverFromSocketFailure(
     error: unknown,
     fallbackMessage: string,
@@ -670,7 +674,6 @@ export class RelayClient {
         error,
         fallbackMessage,
       );
-
       try {
         await this.ensureConnected();
         await this.sendRaw(payload);
@@ -684,11 +687,7 @@ export class RelayClient {
   }
 
   private async closeSubscription(subId: string) {
-    if (this.wsId === null) {
-      return;
-    }
-
-    await this.sendRaw(["CLOSE", subId]);
+    if (this.wsId !== null) await this.sendRaw(["CLOSE", subId]);
   }
 
   async publishEvent(
@@ -819,8 +818,7 @@ export class RelayClient {
 
     if (type === "NOTICE" && typeof rest[0] === "string") {
       const notice: string = rest[0];
-      // Relay back-pressure signal — activate the gate so pending operations
-      // back off until the window expires.
+      // Relay rate-limit notice — activate backoff gate.
       if (notice.startsWith("rate-limited:")) {
         activateRateLimit(parseRateLimitHint(notice));
       }
@@ -968,9 +966,7 @@ export class RelayClient {
       return;
     }
 
-    // Apply ±25% jitter so a fleet of clients reconnecting simultaneously
-    // spreads their AUTH storms across a 50% window instead of all hitting
-    // the relay at the same instant.
+    // Apply ±25% jitter to spread reconnect storms.
     const jitter = this.reconnectDelayMs * (0.75 + Math.random() * 0.5);
     const delay = Math.min(jitter, RECONNECT_MAX_DELAY_MS);
     this.reconnectDelayMs = Math.min(
@@ -1011,12 +1007,7 @@ export class RelayClient {
     }
   }
 
-  private resetConnection(
-    error: Error,
-    options?: {
-      reconnect?: boolean;
-    },
-  ) {
+  private resetConnection(error: Error, options?: { reconnect?: boolean }) {
     this.onMessageChannel = null;
     this.stallWatchdog.stop();
     this.connectionGeneration++;
@@ -1062,6 +1053,12 @@ export class RelayClient {
     }
 
     for (const [subId, subscription] of this.subscriptions) {
+      if (subscription.mode === "fenced") {
+        subscription.lapsed = true;
+        subscription.resolveEstablished();
+        this.subscriptions.delete(subId);
+        continue;
+      }
       if (subscription.mode !== "live") {
         window.clearTimeout(subscription.timeout);
         subscription.reject(error);

desktop/src/shared/api/relayClientShared.ts

@@ -64,6 +64,173 @@ type LiveSubscription = {
   closedRetryTimeout?: number;
 };
 
+/**
+ * Fenced subscription for NIP-RS full-state loads.
+ *
+ * Unlike `live`, a fenced subscription:
+ * - Resolves `established` ONLY on this subscription's own EOSE — never via
+ *   a fallback timer or terminal CLOSED.
+ * - Delivers events synchronously (bypasses the EVENT_BATCH_MS buffer) so
+ *   the loader's synchronous drain barrier needs no timer.
+ * - On CLOSED or any connection lapse, sets `lapsed = true` and resolves
+ *   `established` (in case the loader is still awaiting it), then removes
+ *   itself without retrying.
+ */
+export type FencedSubscription = {
+  mode: "fenced";
+  filter: RelaySubscriptionFilter;
+  onEvent: (event: RelayEvent) => void;
+  /** Connection generation at subscribe time — mismatch = lapsed. */
+  generation: number;
+  /** Resolves on EOSE or lapse (check `lapsed` after awaiting). */
+  resolveEstablished: () => void;
+  /** True after CLOSED, reconnect, or generation change. */
+  lapsed: boolean;
+};
+
+/**
+ * Handle returned by `RelayClient.subscribeFenced()`.
+ *
+ * - `established` resolves ONLY on the subscription's own EOSE, or resolves
+ *   after setting `lapsed = true` if the connection lapses before EOSE.
+ * - `lapsed` is synchronously readable — check it after awaiting `established`
+ *   and at any point during enumeration.
+ * - `unsubscribe()` closes the relay subscription when the load is complete.
+ */
+export type FenceHandle = {
+  /** Resolves on EOSE (check `lapsed` after awaiting). */
+  established: Promise<void>;
+  /** True if the connection lapsed before or after EOSE. */
+  readonly lapsed: boolean;
+  /** Close the relay subscription. */
+  unsubscribe(): Promise<void>;
+};
+
+export function buildFenceHandle(
+  sub: FencedSubscription,
+  established: Promise<void>,
+  unsubscribe: () => Promise<void>,
+): FenceHandle {
+  return {
+    established,
+    get lapsed() {
+      return sub.lapsed;
+    },
+    unsubscribe,
+  };
+}
+
+/**
+ * Millis to wait for EOSE after the REQ is sent.  If the relay keeps the
+ * socket alive but never sends this subscription's EOSE, the fence lapses and
+ * the load concludes `complete: false`.  Reconnect / `retryLoad` owns recovery.
+ * Must NEVER count as establishment.
+ */
+export const FENCED_ESTABLISHMENT_TIMEOUT_MS = 30_000;
+
+/**
+ * Core of `RelayClient.subscribeFenced()` extracted for testability.
+ *
+ * Registers a fenced subscription, sends REQ, and returns a `FenceHandle`
+ * whose `established` resolves ONLY on EOSE — never on a fallback timer or
+ * CLOSED. Any lapse (generation mismatch, send failure, `resetConnection`,
+ * or establishment timeout) sets `lapsed = true` and resolves `established`
+ * so no caller suspends forever.
+ */
+export async function createFencedSubscription(
+  deps: {
+    connectionGeneration: () => number;
+    subscriptions: Map<string, RelaySubscription>;
+    sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise<void>;
+    closeSub: (subId: string) => Promise<void>;
+    /** Override the establishment timeout (ms). Defaults to FENCED_ESTABLISHMENT_TIMEOUT_MS. */
+    establishmentTimeoutMs?: number;
+  },
+  filter: RelaySubscriptionFilter,
+  onEvent: (event: RelayEvent) => void,
+): Promise<FenceHandle> {
+  const generation = deps.connectionGeneration();
+  let resolveEstablished = () => {};
+  const established = new Promise<void>((r) => {
+    resolveEstablished = r;
+  });
+  const subId = `fenced-${crypto.randomUUID()}`;
+  const fencedSub: FencedSubscription = {
+    mode: "fenced",
+    filter,
+    onEvent,
+    generation,
+    resolveEstablished,
+    lapsed: false,
+  };
+  const lapseAndReturn = (): FenceHandle => {
+    fencedSub.lapsed = true;
+    resolveEstablished();
+    deps.subscriptions.delete(subId);
+    return buildFenceHandle(fencedSub, established, async () => {});
+  };
+  if (deps.connectionGeneration() !== generation) return lapseAndReturn();
+  deps.subscriptions.set(subId, fencedSub);
+
+  // ── Establishment timeout ───────────────────────────────────────────────
+  // Install the timeout and wrapped resolver BEFORE sending REQ so that EOSE
+  // delivered synchronously during sendReq() (e.g. from a fake or fast relay)
+  // cancels the timer correctly and never lapses an already-established fence.
+  //
+  // A relay that stays alive but never delivers this subscription's EOSE must
+  // not hang initialize() forever.  The timeout lapses the fence exactly like
+  // a CLOSED or reconnect — it NEVER counts as establishment.
+  const timeoutMs =
+    deps.establishmentTimeoutMs ?? FENCED_ESTABLISHMENT_TIMEOUT_MS;
+  let establishmentTimer: ReturnType<typeof setTimeout> | null = null;
+  let alreadyEstablished = false;
+
+  // Wrap resolveEstablished so the timer is cancelled on normal EOSE.
+  const originalResolve = fencedSub.resolveEstablished;
+  fencedSub.resolveEstablished = () => {
+    alreadyEstablished = true;
+    if (establishmentTimer !== null) {
+      clearTimeout(establishmentTimer);
+      establishmentTimer = null;
+    }
+    originalResolve();
+  };
+  // Also patch the closed reference so the promise resolves via the wrapper.
+  resolveEstablished = fencedSub.resolveEstablished;
+
+  establishmentTimer = setTimeout(() => {
+    establishmentTimer = null;
+    // Only lapse if EOSE has not already established the fence.
+    if (
+      !alreadyEstablished &&
+      !fencedSub.lapsed &&
+      deps.subscriptions.get(subId) === fencedSub
+    ) {
+      fencedSub.lapsed = true;
+      fencedSub.resolveEstablished();
+      deps.subscriptions.delete(subId);
+    }
+  }, timeoutMs);
+
+  try {
+    await deps.sendReq(subId, filter);
+  } catch {
+    return lapseAndReturn();
+  }
+  if (deps.connectionGeneration() !== generation || fencedSub.lapsed)
+    return lapseAndReturn();
+
+  return buildFenceHandle(fencedSub, established, async () => {
+    if (establishmentTimer !== null) {
+      clearTimeout(establishmentTimer);
+      establishmentTimer = null;
+    }
+    if (deps.subscriptions.get(subId) !== fencedSub) return;
+    deps.subscriptions.delete(subId);
+    await deps.closeSub(subId);
+  });
+}
+
 export type PendingEvent = {
   event: RelayEvent;
   resolve: (event: RelayEvent) => void;
@@ -74,7 +241,8 @@ export type PendingEvent = {
 export type RelaySubscription =
   | HistorySubscription
   | FirstEventSubscription
-  | LiveSubscription;
+  | LiveSubscription
+  | FencedSubscription;
 
 export function sortEvents(events: RelayEvent[]) {
   return [...events].sort((left, right) => {

desktop/src/shared/api/relayClosedRecovery.ts

@@ -36,6 +36,13 @@ export function handleRelayClosed({
   const subscription = subscriptions.get(subId);
   if (!subscription) return;
   if (subscription.mode !== "live") {
+    if (subscription.mode === "fenced") {
+      // CLOSED lapses the fence — mark and resolve (unblocks any awaiter).
+      subscription.lapsed = true;
+      subscription.resolveEstablished();
+      subscriptions.delete(subId);
+      return;
+    }
     // Classify before rejecting so a `rate-limited:` history CLOSED arms the
     // gate for concurrent ops. A history sub can't be retried (the caller holds
     // the promise), so we still reject immediately after arming.
@@ -136,6 +143,11 @@ export function prepareSubscriptionEvent(
   if (subscription.mode === "first") {
     return false;
   }
+  if (subscription.mode === "fenced") {
+    // Fenced events are delivered synchronously — caller delivers directly.
+    subscription.onEvent(event);
+    return false; // handled here; do NOT enqueue in the batch buffer
+  }
   subscription.closedRetryAttempt = 0;
   clearClosedRetry(subscription);
   subscription.lastSeenCreatedAt = Math.max(
@@ -156,6 +168,11 @@ export function handleSubscriptionEose({
 }) {
   const subscription = subscriptions.get(subId);
   if (!subscription) return;
+  if (subscription.mode === "fenced") {
+    // EOSE proves this subscription's fence is established.
+    subscription.resolveEstablished();
+    return;
+  }
   if (subscription.mode === "live") {
     subscription.resolveReady?.();
     subscription.resolveReady = undefined;