Merge pull request 'Make live snapshot revisions safe across workers and restarts' (#384) from timmy/383-multi-worker-live-revisions into main

Closes #381

README.md

@@ -40,7 +40,11 @@ another worker replays a confirmed result instead of posting duplicate content.
 five-item Today plan syncs across the operator's devices. Server revisions prevent delayed
 responses from replacing a newer plan; same-account browser tabs exchange fresh snapshots,
 and reconnecting or returning to the dashboard refreshes server truth after replaying queued
-offline operations. After a healthy, fully paginated
+offline operations. Planning edits can remain offline for up to 30 days. After that, the
+expired edit is discarded visibly and the account plan is kept rather than replaying stale
+intent. The server retains no more than 4,096 operation receipts per account and removes
+receipts older than the same 30-day window; client base revisions keep a pruned replay from
+changing a newer Today or Later plan. After a healthy, fully paginated
 My Work refresh proves that an item is complete or otherwise no longer eligible, Stackchain
 queues an idempotent retirement before removing it locally; partial and degraded refreshes
 leave the plan unchanged, and offline retirements replay after reconnect. When
@@ -160,7 +164,10 @@ Every response also defines the browser execution boundary with a Content Securi
 Policy that allows scripts only from the dashboard origin, denies framing and
 plugins, and blocks unused browser capabilities. Keep these response headers when
 proxying; do not add inline scripts or broaden `script-src`. The dashboard bootstrap
-and stylesheet are same-origin static assets included in the offline PWA shell.
+is assembled in source order into one content-addressed JavaScript response. Dashboard
+HTML and the offline worker reference that exact fingerprint, while the runtime receives
+immutable caching and HTML/worker responses remain revalidated. The stylesheet and
+fingerprinted runtime are same-origin assets included atomically in the offline PWA shell.
 Use **Sign out & clear this device** on shared devices; it clears Stackchain's
 offline snapshots, drafts, outboxes, background IndexedDB, and PWA caches without
 removing unrelated forge preferences. Rotate either dashboard secret by replacing
@@ -202,10 +209,12 @@ below the route deadlines. Streaming diff reads share the same read capacity,
 while POST, PATCH, PUT, and DELETE requests are never coalesced.
 
 Live snapshots expose independent `context`, `events`, and `notifications`
-revisions. The browser sends its known revisions on later polls, so `/api/v1/live`
-can omit unchanged section bodies while still returning current freshness and
-retry metadata. The client retains omitted data and only rebuilds or persists the
-sections that changed.
+revision tokens. Each bounded opaque token includes a per-process generation, so a token
+from another worker or from before a restart cannot suppress different content. The browser
+sends its known tokens on later polls, so `/api/v1/live` can omit unchanged section bodies
+while still returning current freshness and retry metadata. The client retains omitted data
+and only rebuilds or persists the sections that changed. Malformed or oversized tokens are
+rejected before any upstream work.
 
 ## Offline mobile shell
 
@@ -224,8 +233,11 @@ or changing any Gitea issue or pull request. They automatically return to their
 existing priority position at the wake time, and **Bring back now** restores them
 early. Later wake times are scoped to the confirmed Gitea login and synchronize
 across signed-in tabs and devices. Offline changes apply immediately, survive reload,
-and replay after reconnect; `STACKCHAIN_LATER_DB` can override the default durable
-store at `.stackchain-state/later.sqlite3`.
+and replay after reconnect. Each queued edit carries the account revision it was based
+on; if another device has since changed the same item, Stackchain keeps the newer
+account plan and reports the conflict instead of silently overwriting it. Unrelated
+items continue syncing in the same batch. `STACKCHAIN_LATER_DB` can override the
+default durable store at `.stackchain-state/later.sqlite3`.
 
 After one successful online load, the installed dashboard precaches a versioned,
 subpath-scoped application shell. During a network outage or a dashboard HTTP

frontend/context-poller.js

@@ -1,3 +1,15 @@
+function buildLiveRevisionQuery(revisions = {}) {
+  const params = new URLSearchParams();
+  const tokenPattern = /^[0-9a-f]{16}\.[0-9]{1,20}$/;
+  ['context', 'events', 'notifications'].forEach((section) => {
+    const revision = revisions[section];
+    if (typeof revision === 'string' && tokenPattern.test(revision)) {
+      params.set(section + '_revision', revision);
+    }
+  });
+  return params.toString();
+}
+
 function createContextPoller({
   fetchContext,
   onSnapshot,
@@ -123,6 +135,8 @@ function createContextPoller({
   };
 }
 
+createContextPoller.buildRevisionQuery = buildLiveRevisionQuery;
+
 if (typeof module !== 'undefined' && module.exports) {
   module.exports = createContextPoller;
 }

frontend/dashboard.js

@@ -155,8 +155,9 @@
       status.textContent = state === 'saved' ? 'Today saved to account.' :
         (state === 'retrying' ? `Today saved on this device · retrying in ${Math.ceil(detail.delayMs / 1000)}s.` :
           (state === 'pending' ? 'Today saved on this device · sync pending.' :
-          (state === 'full' ? 'Another device filled Today · showing its saved plan.' :
-            'Today sync unavailable · changes stay on this device.')));
+            (state === 'full' ? 'Another device filled Today · showing its saved plan.' :
+              (state === 'expired' ? `Today edit expired after 30 days offline${detail.count > 1 ? 's' : ''} · account plan kept.` :
+                'Today sync unavailable · changes stay on this device.'))));
     },
   });
   todaySync.startLifecycle({ window, document });
@@ -186,9 +187,11 @@
     },
     onStatus: (state, detail = {}) => {
       qs('#later-sync-status').textContent = state === 'saved' ? 'Later saved to account.' :
-        (state === 'retrying' ? `Later saved on this device · retrying in ${Math.ceil(detail.delayMs / 1000)}s.` :
-          (state === 'pending' ? 'Later saved on this device · sync pending.' :
-            'Later sync unavailable · changes stay on this device.'));
+        (state === 'conflict' ? `Another device changed this Later item${detail.count > 1 ? 's' : ''} · account plan kept.` :
+          (state === 'retrying' ? `Later saved on this device · retrying in ${Math.ceil(detail.delayMs / 1000)}s.` :
+            (state === 'pending' ? 'Later saved on this device · sync pending.' :
+              (state === 'expired' ? `Later edit expired after 30 days offline${detail.count > 1 ? 's' : ''} · account plan kept.` :
+                'Later sync unavailable · changes stay on this device.'))));
     },
   });
   laterSync.startLifecycle({ window, document });
@@ -321,11 +324,7 @@
   setClock(); setInterval(setClock, 1000);
 
   async function fetchLiveSnapshot(revisions = {}, { signal } = {}) {
-    const params = new URLSearchParams();
-    Object.entries(revisions).forEach(([section, revision]) => {
-      if (Number.isInteger(revision) && revision >= 0) params.set(section + '_revision', revision);
-    });
-    const query = params.toString();
+    const query = createContextPoller.buildRevisionQuery(revisions);
     const res = await fetch('api/v1/live' + (query ? '?' + query : ''), {
       headers: { Accept: 'application/json' },
       signal,

frontend/later-sync.js

@@ -1,5 +1,6 @@
 function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStatus, createOperationId, createChannel, coordinator,
-  setTimer = globalThis.setTimeout, clearTimer = globalThis.clearTimeout, retryBaseMs = 1000, retryMaxMs = 30000 }) {
+  setTimer = globalThis.setTimeout, clearTimer = globalThis.clearTimeout, retryBaseMs = 1000, retryMaxMs = 30000,
+  now = Date.now, maxOfflineMs = 30 * 24 * 60 * 60 * 1000 }) {
   const prefix = 'stackchain.later-sync.v1.';
   const migrationPrefix = 'stackchain.later-sync-migrated.v1.';
   const snapshotPrefix = 'stackchain.later-sync-snapshot.v1.';
@@ -8,6 +9,7 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat
   let channelKey = '';
   let retryTimer = null;
   let retryAttempt = 0;
+  let expiredCount = 0;
   const knownOperationKeys = new Set();
 
   function cancelRetry() {
@@ -105,17 +107,32 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat
       const records = [...keys].map(recordKey => {
         const record = JSON.parse(storage.getItem(recordKey) || 'null');
         return record ? { ...record, recordKey } : null;
-      }).filter(record => record?.operation)
-        .sort((left, right) => Number(left.queued_at || 0) - Number(right.queued_at || 0) ||
-          left.operation.operation_id.localeCompare(right.operation.operation_id));
-      const latestByItem = new Map();
-      records.forEach(record => latestByItem.set(record.operation.item_id, record));
-      records.filter(record => latestByItem.get(record.operation.item_id) !== record).forEach(record => {
+      }).filter(record => record?.operation);
+      const expired = records.filter(record => Number(record.queued_at) >= 1_000_000_000_000 &&
+        now() - Number(record.queued_at) > maxOfflineMs);
+      expired.forEach(record => {
         storage.removeItem(record.recordKey);
         knownOperationKeys.delete(record.recordKey);
       });
-      return records.filter(record => latestByItem.get(record.operation.item_id) === record)
-        .map(record => record.operation).filter(operation =>
+      if (expired.length) {
+        expiredCount += expired.length;
+        onStatus?.('expired', { count: expiredCount });
+      }
+      const activeRecords = records.filter(record => !expired.includes(record))
+        .sort((left, right) => Number(left.queued_at || 0) - Number(right.queued_at || 0) ||
+          left.operation.operation_id.localeCompare(right.operation.operation_id));
+      const latestByItem = new Map();
+      activeRecords.forEach(record => latestByItem.set(record.operation.item_id, record));
+      activeRecords.filter(record => latestByItem.get(record.operation.item_id) !== record).forEach(record => {
+        storage.removeItem(record.recordKey);
+        knownOperationKeys.delete(record.recordKey);
+      });
+      return activeRecords.filter(record => latestByItem.get(record.operation.item_id) === record)
+        .map(record => ({
+          ...record.operation,
+          base_revision: Number.isInteger(record.operation.base_revision)
+            ? record.operation.base_revision : 0,
+        })).filter(operation =>
           operation && typeof operation.operation_id === 'string' &&
           ['defer', 'restore'].includes(operation.action) && typeof operation.item_id === 'string' &&
           (operation.action === 'restore' || typeof operation.wake_at === 'string'));
@@ -148,13 +165,16 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat
         (action === 'defer' && typeof wakeAt !== 'string')) return false;
     pending().filter(operation => operation.item_id === itemId)
       .forEach(operation => removeOperation(operation.operation_id));
-    const operation = { operation_id: operationId(), action, item_id: itemId, wake_at: wakeAt };
+    const operation = {
+      operation_id: operationId(), action, item_id: itemId, wake_at: wakeAt,
+      base_revision: Math.max(0, savedRevision()),
+    };
     const storageKey = key();
     if (!storageKey || !storage) return false;
     const recordKey = storageKey + '.operation.' + encodeURIComponent(operation.operation_id);
     let saved = false;
     try {
-      storage.setItem(recordKey, JSON.stringify({ operation, queued_at: Date.now() }));
+      storage.setItem(recordKey, JSON.stringify({ operation, queued_at: now() }));
       knownOperationKeys.add(recordKey);
       saved = true;
       coordinator?.notify('later');
@@ -183,9 +203,11 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat
     const ownerKey = key();
     if (!ownerKey) return false;
     ensureChannel();
+    expiredCount = 0;
     try {
       let operations = pending();
       let plan;
+      let conflictCount = 0;
       if (!operations.length) plan = await fetchJson('api/v1/later');
       while (operations.length) {
         if (key() !== ownerKey) return false;
@@ -202,6 +224,8 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat
           ...(plan.duplicate_operation_ids || []),
           ...(plan.rejected_operations || []).map(item => item.operation_id),
         ] : batch.map(item => item.operation_id);
+        conflictCount += (plan.rejected_operations || [])
+          .filter(item => item.reason === 'stale_intent').length;
         for (const operationId of received) {
           if (pending().some(candidate => candidate.operation_id === operationId) &&
               !removeOperation(operationId)) {
@@ -211,7 +235,9 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat
         operations = pending();
       }
       adopt(plan);
-      onStatus?.(pending().length ? 'pending' : 'saved');
+      onStatus?.(pending().length ? 'pending' :
+        (conflictCount ? 'conflict' : expiredCount ? 'expired' : 'saved'),
+        conflictCount ? { count: conflictCount } : expiredCount ? { count: expiredCount } : {});
       retryAttempt = 0;
       cancelRetry();
       return true;

frontend/service-worker.js

@@ -1,6 +1,6 @@
 const BASE = new URL('./', self.location.href).pathname;
 importScripts(BASE + 'static/background-issue-sync.js');
-const CACHE = 'stackchain-dashboard-shell-v54';
+const CACHE = 'stackchain-dashboard-shell-v56';
 const OUTAGE_STATUSES = new Set([500, 502, 503, 504]);
 const NAVIGATION_TIMEOUT_MS = self.__STACKCHAIN_NAVIGATION_TIMEOUT_MS || 4000;
 const SHELL = [

frontend/today-sync.js

@@ -1,5 +1,6 @@
 function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus, createOperationId, createChannel, coordinator,
-  setTimer = globalThis.setTimeout, clearTimer = globalThis.clearTimeout, retryBaseMs = 1000, retryMaxMs = 30000 }) {
+  setTimer = globalThis.setTimeout, clearTimer = globalThis.clearTimeout, retryBaseMs = 1000, retryMaxMs = 30000,
+  now = Date.now, maxOfflineMs = 30 * 24 * 60 * 60 * 1000 }) {
   const prefix = 'stackchain.today-sync.v1.';
   const migrationPrefix = 'stackchain.today-sync-migrated.v1.';
   const snapshotPrefix = 'stackchain.today-sync-snapshot.v1.';
@@ -8,6 +9,7 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
   let channelKey = '';
   let retryTimer = null;
   let retryAttempt = 0;
+  let expiredCount = 0;
   const knownOperationKeys = new Set();
 
   function cancelRetry() {
@@ -97,11 +99,28 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
         const candidate = storage.key?.(index);
         if (candidate?.startsWith(recordPrefix)) keys.add(candidate);
       }
-      return [...keys].map(recordKey => JSON.parse(storage.getItem(recordKey) || 'null'))
-        .filter(record => record?.operation)
+      const records = [...keys].map(recordKey => {
+        const record = JSON.parse(storage.getItem(recordKey) || 'null');
+        return record ? { ...record, recordKey } : null;
+      }).filter(record => record?.operation);
+      const expired = records.filter(record => Number(record.queued_at) >= 1_000_000_000_000 &&
+        now() - Number(record.queued_at) > maxOfflineMs);
+      expired.forEach(record => {
+        storage.removeItem(record.recordKey);
+        knownOperationKeys.delete(record.recordKey);
+      });
+      if (expired.length) {
+        expiredCount += expired.length;
+        onStatus?.('expired', { count: expiredCount });
+      }
+      return records.filter(record => !expired.includes(record))
         .sort((left, right) => Number(left.queued_at || 0) - Number(right.queued_at || 0) ||
           left.operation.operation_id.localeCompare(right.operation.operation_id))
-        .map(record => record.operation)
+        .map(record => ({
+          ...record.operation,
+          base_revision: Number.isInteger(record.operation.base_revision)
+            ? record.operation.base_revision : Math.max(0, savedRevision()),
+        }))
         .filter(operation => operation && typeof operation.operation_id === 'string' &&
           ['add', 'remove', 'move'].includes(operation.action) && typeof operation.item_id === 'string');
     } catch (_error) {
@@ -136,13 +155,16 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
       onStatus?.('pending');
       return true;
     }
-    const operation = { operation_id: operationId(), action, item_id: itemId, direction };
+    const operation = {
+      operation_id: operationId(), action, item_id: itemId, direction,
+      base_revision: Math.max(0, savedRevision()),
+    };
     const storageKey = key();
     if (!storageKey || !storage) return false;
     const recordKey = storageKey + '.operation.' + encodeURIComponent(operation.operation_id);
     let saved = false;
     try {
-      storage.setItem(recordKey, JSON.stringify({ operation, queued_at: Date.now() }));
+      storage.setItem(recordKey, JSON.stringify({ operation, queued_at: now() }));
       knownOperationKeys.add(recordKey);
       saved = true;
       coordinator?.notify('today');
@@ -169,6 +191,7 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
     const ownerKey = key();
     if (!ownerKey) return false;
     ensureChannel();
+    expiredCount = 0;
     try {
       let operations = pending();
       let plan;
@@ -199,7 +222,8 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
         operations = pending();
       }
       adopt(plan);
-      onStatus?.(pending().length ? 'pending' : hadConflict ? 'full' : 'saved');
+      onStatus?.(pending().length ? 'pending' : hadConflict ? 'full' :
+        expiredCount ? 'expired' : 'saved', expiredCount ? { count: expiredCount } : {});
       retryAttempt = 0;
       cancelRetry();
       return !hadConflict;

src/frontend_bundle.py

@@ -0,0 +1,70 @@
+"""Build the content-addressed browser shell served by the dashboard."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+
+SCRIPT_TAG = re.compile(r'^<script src="(static/[^"?]+\.js)"></script>$', re.MULTILINE)
+WORKER_RUNTIME_SOURCE = "static/background-issue-sync.js"
+
+
+@dataclass(frozen=True)
+class FrontendBuild:
+    dashboard_html: str
+    runtime_bytes: bytes
+    runtime_digest: str
+    runtime_name: str
+    page_sources: tuple[str, ...]
+    service_worker_source: str
+
+
+def _bundle(frontend_dir: Path, sources: tuple[str, ...]) -> bytes:
+    chunks = []
+    for source in sources:
+        path = frontend_dir / source.removeprefix("static/")
+        chunks.append(f"/* {source} */\n".encode() + path.read_bytes() + b"\n;\n")
+    return b"".join(chunks)
+
+
+def build_frontend(frontend_dir: Path) -> FrontendBuild:
+    """Return one internally consistent HTML, runtime, and offline-worker build."""
+    source_html = (frontend_dir / "index.html").read_text()
+    sources = tuple(SCRIPT_TAG.findall(source_html))
+    if not sources:
+        raise ValueError("dashboard entry document has no page scripts")
+
+    runtime = _bundle(frontend_dir, sources)
+    digest = hashlib.sha256(runtime).hexdigest()[:16]
+    runtime_name = f"runtime-{digest}.js"
+    dashboard_html = SCRIPT_TAG.sub("", source_html)
+    dashboard_html = dashboard_html.replace(
+        "</body>", f'<script src="{runtime_name}"></script>\n</body>'
+    )
+
+    worker = (frontend_dir / "service-worker.js").read_text()
+    for source in sources:
+        if source != WORKER_RUNTIME_SOURCE:
+            worker = worker.replace(f"  BASE + '{source}',\n", "")
+    worker = worker.replace(
+        "  BASE + 'static/dashboard.css',\n",
+        f"  BASE + 'static/dashboard.css',\n  BASE + '{runtime_name}',\n",
+    )
+    worker = re.sub(
+        r"const CACHE = 'stackchain-dashboard-shell-v\d+';",
+        f"const CACHE = 'stackchain-dashboard-shell-{digest}';",
+        worker,
+        count=1,
+    )
+
+    return FrontendBuild(
+        dashboard_html=dashboard_html,
+        runtime_bytes=runtime,
+        runtime_digest=digest,
+        runtime_name=runtime_name,
+        page_sources=sources,
+        service_worker_source=worker,
+    )

src/later_store.py

@@ -2,14 +2,26 @@
 
 import json
 import sqlite3
+import time
 from datetime import datetime
 from pathlib import Path
 
 
 class LaterStore:
-    def __init__(self, path: str | Path, *, timeout: float = 1.0):
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        timeout: float = 1.0,
+        operation_limit: int = 4096,
+        operation_retention_seconds: float = 30 * 24 * 60 * 60,
+        clock=time.time,
+    ):
         self.path = Path(path)
         self.timeout = timeout
+        self.operation_limit = operation_limit
+        self.operation_retention_seconds = operation_retention_seconds
+        self.clock = clock
 
     def _connect(self) -> sqlite3.Connection:
         self.path.parent.mkdir(parents=True, exist_ok=True)
@@ -29,12 +41,48 @@ class LaterStore:
             CREATE TABLE IF NOT EXISTS later_operations (
                 login TEXT NOT NULL,
                 operation_id TEXT NOT NULL,
+                created_at REAL NOT NULL,
                 PRIMARY KEY (login, operation_id)
             )
             """
         )
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS later_item_revisions (
+                login TEXT NOT NULL,
+                item_id TEXT NOT NULL,
+                revision INTEGER NOT NULL,
+                PRIMARY KEY (login, item_id)
+            )
+            """
+        )
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(later_operations)")}
+        if "created_at" not in columns:
+            connection.execute("ALTER TABLE later_operations ADD COLUMN created_at REAL")
+        connection.execute(
+            "UPDATE later_operations SET created_at = ? WHERE created_at IS NULL",
+            (self.clock(),),
+        )
+        connection.commit()
         return connection
 
+    def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
+        now = self.clock()
+        connection.execute(
+            "INSERT INTO later_operations(login, operation_id, created_at) VALUES (?, ?, ?)",
+            (login, operation_id, now),
+        )
+        connection.execute(
+            "DELETE FROM later_operations WHERE login = ? AND created_at < ?",
+            (login, now - self.operation_retention_seconds),
+        )
+        connection.execute(
+            "DELETE FROM later_operations WHERE login = ? AND rowid NOT IN "
+            "(SELECT rowid FROM later_operations WHERE login = ? "
+            "ORDER BY created_at DESC, rowid DESC LIMIT ?)",
+            (login, login, self.operation_limit),
+        )
+
     @staticmethod
     def _normalize_login(login: str) -> str:
         normalized = login.strip().lower()
@@ -74,54 +122,16 @@ class LaterStore:
         item_id: str,
         *,
         wake_at: str | None = None,
+        base_revision: int | None = None,
     ) -> dict:
-        login = self._normalize_login(login)
-        if not operation_id or not item_id:
-            raise ValueError("operation_id and item_id are required")
-        if action not in {"defer", "restore"}:
-            raise ValueError("unsupported Later action")
-        if action == "defer":
-            wake_at = self._validate_wake_at(wake_at)
-
-        with self._connect() as connection:
-            connection.execute("BEGIN IMMEDIATE")
-            row = connection.execute(
-                "SELECT revision, records FROM later_plans WHERE login = ?", (login,)
-            ).fetchone()
-            snapshot = self._snapshot(row)
-            duplicate = connection.execute(
-                "SELECT 1 FROM later_operations WHERE login = ? AND operation_id = ?",
-                (login, operation_id),
-            ).fetchone()
-            if duplicate:
-                return snapshot
-
-            records = dict(snapshot["records"])
-            before = records.get(item_id)
-            if action == "defer":
-                records[item_id] = wake_at
-                changed = before != wake_at
-            else:
-                changed = item_id in records
-                records.pop(item_id, None)
-
-            revision = snapshot["revision"] + (1 if changed else 0)
-            serialized = json.dumps(records, separators=(",", ":"), sort_keys=True)
-            if row is None:
-                connection.execute(
-                    "INSERT INTO later_plans(login, revision, records) VALUES (?, ?, ?)",
-                    (login, revision, serialized),
-                )
-            elif changed:
-                connection.execute(
-                    "UPDATE later_plans SET revision = ?, records = ? WHERE login = ?",
-                    (revision, serialized, login),
-                )
-            connection.execute(
-                "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)",
-                (login, operation_id),
-            )
-            return {"revision": revision, "records": records}
+        result = self.apply_batch(login, [{
+            "operation_id": operation_id,
+            "action": action,
+            "item_id": item_id,
+            "wake_at": wake_at,
+            "base_revision": base_revision,
+        }])
+        return {"revision": result["revision"], "records": result["records"]}
 
     def apply_batch(self, login: str, operations: list[dict]) -> dict:
         """Apply ordered deferrals with one SQLite write transaction."""
@@ -136,6 +146,20 @@ class LaterStore:
             revision = snapshot["revision"]
             accepted: list[str] = []
             duplicates: list[str] = []
+            rejected: list[dict[str, str]] = []
+
+            # Existing databases predate per-item revisions. Conservatively mark
+            # active deferrals as changed at the latest known plan revision.
+            for item_id in records:
+                connection.execute(
+                    "INSERT OR IGNORE INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)",
+                    (login, item_id, revision),
+                )
+            item_revisions = dict(connection.execute(
+                "SELECT item_id, revision FROM later_item_revisions WHERE login = ?",
+                (login,),
+            ).fetchall())
+            batch_start_item_revisions = dict(item_revisions)
 
             for operation in operations:
                 operation_id = operation.get("operation_id", "")
@@ -156,6 +180,19 @@ class LaterStore:
                     duplicates.append(operation_id)
                     continue
 
+                base_revision = operation.get("base_revision")
+                if base_revision is None:
+                    base_revision = snapshot["revision"]
+                if not isinstance(base_revision, int) or isinstance(base_revision, bool) or base_revision < 0:
+                    raise ValueError("base_revision must be a non-negative integer")
+                if batch_start_item_revisions.get(item_id, 0) > base_revision:
+                    self._record_operation(connection, login, operation_id)
+                    rejected.append({
+                        "operation_id": operation_id,
+                        "reason": "stale_intent",
+                    })
+                    continue
+
                 before = records.get(item_id)
                 if action == "defer":
                     records[item_id] = wake_at
@@ -164,10 +201,14 @@ class LaterStore:
                     changed = item_id in records
                     records.pop(item_id, None)
                 revision += 1 if changed else 0
-                connection.execute(
-                    "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)",
-                    (login, operation_id),
-                )
+                if changed:
+                    item_revisions[item_id] = revision
+                    connection.execute(
+                        "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?) "
+                        "ON CONFLICT(login, item_id) DO UPDATE SET revision = excluded.revision",
+                        (login, item_id, revision),
+                    )
+                self._record_operation(connection, login, operation_id)
                 accepted.append(operation_id)
 
             serialized = json.dumps(records, separators=(",", ":"), sort_keys=True)
@@ -186,5 +227,5 @@ class LaterStore:
                 "records": records,
                 "accepted_operation_ids": accepted,
                 "duplicate_operation_ids": duplicates,
-                "rejected_operations": [],
+                "rejected_operations": rejected,
             }

src/main.py

@@ -2,6 +2,7 @@ import asyncio
 import hmac
 import math
 import os
+import secrets
 import sqlite3
 import time
 from collections.abc import Awaitable, Coroutine
@@ -37,7 +38,7 @@ from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
 from src.suggestion_engine import compute
 from src.later_store import LaterStore
 from src.today_store import TodayPlanFull, TodayStore
-from src.views import router as frontend_router
+from src.views import FRONTEND_BUILD, router as frontend_router
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI):
@@ -108,6 +109,7 @@ _live_snapshot_refreshing_sections: set[str] = set()
 _live_section_revisions: dict[str, int] = {
     section: 0 for section in LIVE_SNAPSHOT_SECTIONS
 }
+_live_revision_generation = secrets.token_hex(8)
 _read_notification_ids: set[int] = set()
 _authored_action_operations: dict[
     str, tuple[tuple[Any, ...], asyncio.Task, float]
@@ -207,6 +209,7 @@ class TodayOperation(BaseModel):
     action: Literal["add", "remove", "move"]
     item_id: str = Field(min_length=1, max_length=500)
     direction: Literal["up", "down"] | None = None
+    base_revision: int | None = Field(default=None, ge=0)
 
     @model_validator(mode="after")
     def require_move_direction(self):
@@ -222,6 +225,7 @@ class LaterOperation(BaseModel):
     action: Literal["defer", "restore"]
     item_id: str = Field(min_length=1, max_length=500)
     wake_at: str | None = Field(default=None, max_length=100)
+    base_revision: int | None = Field(default=None, ge=0)
 
 
 class TodayOperationBatch(BaseModel):
@@ -587,7 +591,13 @@ async def require_operator_session(request: Request, call_next):
         return await call_next(request)
 
     public = (
-        path in {"/healthz", "/readyz", "/login", "/manifest.webmanifest"}
+        path in {
+            "/healthz",
+            "/readyz",
+            "/login",
+            "/manifest.webmanifest",
+            "/" + FRONTEND_BUILD.runtime_name,
+        }
         or path.startswith("/static/")
         or (path == "/api/v1/session" and request.method == "POST")
     )
@@ -969,6 +979,7 @@ async def update_later_plan(payload: LaterOperation | LaterOperationBatch):
             payload.action,
             payload.item_id,
             wake_at=payload.wake_at,
+            base_revision=payload.base_revision,
         )
     except ValueError as error:
         raise HTTPException(status_code=422, detail=str(error))
@@ -1575,7 +1586,7 @@ def _live_snapshot_payload(
     *,
     stale: bool,
     revalidating: bool,
-    known_revisions: dict[str, int | None] | None = None,
+    known_revisions: dict[str, str | None] | None = None,
 ) -> dict:
     payload = dict(value)
     now = time.monotonic()
@@ -1608,9 +1619,13 @@ def _live_snapshot_payload(
         "retry_in_seconds": min(retries, default=0),
         "sections": section_freshness,
     }
-    payload["revisions"] = dict(_live_section_revisions)
+    revision_tokens = {
+        section: f"{_live_revision_generation}.{revision}"
+        for section, revision in _live_section_revisions.items()
+    }
+    payload["revisions"] = revision_tokens
     for section, known_revision in (known_revisions or {}).items():
-        if known_revision is None or known_revision != _live_section_revisions[section]:
+        if known_revision is None or known_revision != revision_tokens[section]:
             continue
         payload.pop(section, None)
         if section == "notifications":
@@ -1660,9 +1675,15 @@ def _without_read_notifications(snapshot: dict) -> dict:
 
 @app.get("/api/v1/live")
 async def live_snapshot(
-    context_revision: int | None = Query(default=None, ge=0),
-    events_revision: int | None = Query(default=None, ge=0),
-    notifications_revision: int | None = Query(default=None, ge=0),
+    context_revision: str | None = Query(
+        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
+    ),
+    events_revision: str | None = Query(
+        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
+    ),
+    notifications_revision: str | None = Query(
+        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
+    ),
 ) -> JSONResponse:
     """Return a freshness-bounded snapshot and share identical upstream loads."""
     global _live_snapshot_task, _live_snapshot_value, _live_snapshot_created_at

src/today_store.py

@@ -2,6 +2,7 @@
 
 import json
 import sqlite3
+import time
 from pathlib import Path
 
 
@@ -10,10 +11,22 @@ class TodayPlanFull(ValueError):
 
 
 class TodayStore:
-    def __init__(self, path: str | Path, *, limit: int = 5, timeout: float = 1.0):
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        limit: int = 5,
+        timeout: float = 1.0,
+        operation_limit: int = 4096,
+        operation_retention_seconds: float = 30 * 24 * 60 * 60,
+        clock=time.time,
+    ):
         self.path = Path(path)
         self.limit = limit
         self.timeout = timeout
+        self.operation_limit = operation_limit
+        self.operation_retention_seconds = operation_retention_seconds
+        self.clock = clock
 
     def _connect(self) -> sqlite3.Connection:
         self.path.parent.mkdir(parents=True, exist_ok=True)
@@ -33,12 +46,38 @@ class TodayStore:
             CREATE TABLE IF NOT EXISTS today_operations (
                 login TEXT NOT NULL,
                 operation_id TEXT NOT NULL,
+                created_at REAL NOT NULL,
                 PRIMARY KEY (login, operation_id)
             )
             """
         )
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(today_operations)")}
+        if "created_at" not in columns:
+            connection.execute("ALTER TABLE today_operations ADD COLUMN created_at REAL")
+        connection.execute(
+            "UPDATE today_operations SET created_at = ? WHERE created_at IS NULL",
+            (self.clock(),),
+        )
+        connection.commit()
         return connection
 
+    def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
+        now = self.clock()
+        connection.execute(
+            "INSERT INTO today_operations(login, operation_id, created_at) VALUES (?, ?, ?)",
+            (login, operation_id, now),
+        )
+        connection.execute(
+            "DELETE FROM today_operations WHERE login = ? AND created_at < ?",
+            (login, now - self.operation_retention_seconds),
+        )
+        connection.execute(
+            "DELETE FROM today_operations WHERE login = ? AND rowid NOT IN "
+            "(SELECT rowid FROM today_operations WHERE login = ? "
+            "ORDER BY created_at DESC, rowid DESC LIMIT ?)",
+            (login, login, self.operation_limit),
+        )
+
     @staticmethod
     def _normalize_login(login: str) -> str:
         normalized = login.strip().lower()
@@ -123,10 +162,7 @@ class TodayStore:
                     "UPDATE today_plans SET revision = ?, ids = ? WHERE login = ?",
                     (revision, json.dumps(ids, separators=(",", ":")), login),
                 )
-            connection.execute(
-                "INSERT INTO today_operations(login, operation_id) VALUES (?, ?)",
-                (login, operation_id),
-            )
+            self._record_operation(connection, login, operation_id)
             return {"revision": revision, "ids": ids}
 
     def apply_batch(self, login: str, operations: list[dict]) -> dict:
@@ -163,6 +199,16 @@ class TodayStore:
                     duplicates.append(operation_id)
                     continue
 
+                base_revision = operation.get("base_revision")
+                if base_revision is None:
+                    base_revision = snapshot["revision"]
+                if not isinstance(base_revision, int) or isinstance(base_revision, bool) or base_revision < 0:
+                    raise ValueError("base_revision must be a non-negative integer")
+                if base_revision < snapshot["revision"]:
+                    self._record_operation(connection, login, operation_id)
+                    rejected.append({"operation_id": operation_id, "reason": "stale_intent"})
+                    continue
+
                 changed = False
                 if action == "add":
                     if item_id not in ids:
@@ -185,10 +231,7 @@ class TodayStore:
                         ids[index], ids[target] = ids[target], ids[index]
                         changed = True
                 revision += 1 if changed else 0
-                connection.execute(
-                    "INSERT INTO today_operations(login, operation_id) VALUES (?, ?)",
-                    (login, operation_id),
-                )
+                self._record_operation(connection, login, operation_id)
                 accepted.append(operation_id)
 
             serialized = json.dumps(ids, separators=(",", ":"))

src/views.py

@@ -1,12 +1,14 @@
 from pathlib import Path
 
-from fastapi import APIRouter
-from fastapi.responses import FileResponse, HTMLResponse
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import FileResponse, HTMLResponse, Response
+
+from src.frontend_bundle import build_frontend
 
 router = APIRouter()
 DASHBOARD_FILE = Path(__file__).resolve().parent.parent / "frontend" / "index.html"
 MANIFEST_FILE = DASHBOARD_FILE.parent / "manifest.webmanifest"
-SERVICE_WORKER_FILE = DASHBOARD_FILE.parent / "service-worker.js"
+FRONTEND_BUILD = build_frontend(DASHBOARD_FILE.parent)
 
 LOGIN_HTML = """<!doctype html>
 <html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
@@ -26,7 +28,7 @@ class RevalidatingHTMLResponse(HTMLResponse):
 
 @router.get("/", response_class=RevalidatingHTMLResponse)
 async def dashboard() -> str:
-    return DASHBOARD_FILE.read_text()
+    return FRONTEND_BUILD.dashboard_html
 
 
 @router.get("/login", response_class=RevalidatingHTMLResponse)
@@ -39,10 +41,21 @@ async def web_app_manifest() -> FileResponse:
     return FileResponse(MANIFEST_FILE, media_type="application/manifest+json")
 
 
-@router.get("/service-worker.js", response_class=FileResponse)
-async def service_worker() -> FileResponse:
-    return FileResponse(
-        SERVICE_WORKER_FILE,
+@router.get("/service-worker.js")
+async def service_worker() -> Response:
+    return Response(
+        FRONTEND_BUILD.service_worker_source,
         media_type="application/javascript",
         headers={"Service-Worker-Allowed": "/", "Cache-Control": "no-cache"},
     )
+
+
+@router.get("/runtime-{digest}.js")
+async def runtime_bundle(digest: str) -> Response:
+    if digest != FRONTEND_BUILD.runtime_digest:
+        raise HTTPException(status_code=404, detail="Runtime revision not found")
+    return Response(
+        FRONTEND_BUILD.runtime_bytes,
+        media_type="application/javascript",
+        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+    )

tests/dashboard_bundle.py

@@ -1,9 +1,17 @@
+import re
 from pathlib import Path
 
 from src.views import dashboard as dashboard_html
 
 
 FRONTEND = Path(__file__).resolve().parents[1] / "frontend"
+PAGE_SCRIPT_MANIFEST = "\n".join(
+    re.findall(
+        r'^<script src="static/[^\"]+\.js"></script>$',
+        (FRONTEND / "index.html").read_text(),
+        re.MULTILINE,
+    )
+)
 
 
 def dashboard_bundle_text() -> str:
@@ -18,11 +26,12 @@ def dashboard_bundle_text() -> str:
 
 
 async def dashboard() -> str:
-    # Exercise the real view lookup before adding its separately served assets.
+    # Exercise the real view lookup before adding its separately served bootstrap assets.
     html = await dashboard_html()
     return "\n".join(
         (
             html,
+            PAGE_SCRIPT_MANIFEST,
             (FRONTEND / "dashboard.css").read_text(),
             (FRONTEND / "dashboard.js").read_text(),
         )

tests/test_context_polling.py

@@ -116,6 +116,23 @@ const poller = createContextPoller({{
     }
 
 
+def test_live_revision_query_preserves_bounded_opaque_tokens():
+    script = f"""
+const createContextPoller = require({json.dumps(str(POLLER))});
+const query = createContextPoller.buildRevisionQuery({{
+  context: '0123456789abcdef.12',
+  events: 'fedcba9876543210.3',
+  notifications: 'not-a-token',
+  extra: '0123456789abcdef.1',
+}});
+process.stdout.write(JSON.stringify({{ query }}));
+"""
+
+    assert run_node(script) == {
+        "query": "context_revision=0123456789abcdef.12&events_revision=fedcba9876543210.3"
+    }
+
+
 def test_context_poller_aborts_a_stalled_request_and_recovers_on_schedule():
     script = f"""
 const createContextPoller = require({json.dumps(str(POLLER))});

tests/test_dashboard_auth.py

@@ -7,6 +7,7 @@ import pytest
 
 from src import main
 from src.session_store import SessionStoreError
+from src.views import FRONTEND_BUILD
 
 
 @pytest.fixture
@@ -956,9 +957,16 @@ async def test_public_routes_skip_session_registry_validation(access_control, mo
         login = await client.get("/login")
         manifest = await client.get("/manifest.webmanifest")
         static = await client.get("/static/session.js")
+        runtime = await client.get("/" + FRONTEND_BUILD.runtime_name)
 
     assert signed_in.status_code == 200
-    assert [login.status_code, manifest.status_code, static.status_code] == [200, 200, 200]
+    assert [login.status_code, manifest.status_code, static.status_code, runtime.status_code] == [
+        200,
+        200,
+        200,
+        200,
+    ]
+    assert runtime.headers["cache-control"] == "public, max-age=31536000, immutable"
 
 
 @pytest.mark.anyio

tests/test_frontend_bundle.py

@@ -0,0 +1,62 @@
+import gzip
+import re
+import shutil
+from pathlib import Path
+
+import pytest
+
+from src.frontend_bundle import build_frontend
+from src.views import dashboard, runtime_bundle, service_worker
+
+
+FRONTEND = Path(__file__).resolve().parents[1] / "frontend"
+PAGE_SCRIPT = re.compile(r'<script src="(static/[^"]+\.js)"></script>')
+
+
+def test_page_runtime_is_one_deterministic_content_addressed_bundle(tmp_path):
+    first = build_frontend(FRONTEND)
+    second = build_frontend(FRONTEND)
+
+    assert first.runtime_name == second.runtime_name
+    assert first.runtime_bytes == second.runtime_bytes
+    assert PAGE_SCRIPT.findall(first.dashboard_html) == []
+    assert first.dashboard_html.count("<script src=") == 1
+    assert f'<script src="{first.runtime_name}"></script>' in first.dashboard_html
+    assert first.runtime_name.startswith("runtime-")
+    assert first.runtime_name.endswith(".js")
+    assert len(gzip.compress(first.runtime_bytes, mtime=0)) <= 100 * 1024
+
+    changed_frontend = tmp_path / "frontend"
+    shutil.copytree(FRONTEND, changed_frontend)
+    (changed_frontend / "widgets.js").write_text(
+        (changed_frontend / "widgets.js").read_text() + "\n// changed\n"
+    )
+
+    changed = build_frontend(changed_frontend)
+    assert changed.runtime_name != first.runtime_name
+    assert changed.runtime_bytes != first.runtime_bytes
+
+
+def test_offline_shell_precaches_exact_runtime_without_superseded_page_modules():
+    build = build_frontend(FRONTEND)
+
+    assert f"BASE + '{build.runtime_name}'" in build.service_worker_source
+    for source in build.page_sources:
+        if source == "static/background-issue-sync.js":
+            continue
+        assert f"BASE + '{source}'" not in build.service_worker_source
+
+
+@pytest.mark.anyio
+async def test_runtime_asset_is_immutable_while_html_and_worker_revalidate():
+    build = build_frontend(FRONTEND)
+
+    html = await dashboard()
+    bundle = await runtime_bundle(build.runtime_digest)
+    worker = await service_worker()
+
+    assert build.runtime_name in html
+    assert bundle.body == build.runtime_bytes
+    assert bundle.headers["cache-control"] == "public, max-age=31536000, immutable"
+    assert worker.headers["cache-control"] == "no-cache"
+    assert build.runtime_name.encode() in worker.body

tests/test_later_store.py

@@ -86,6 +86,114 @@ def test_batch_applies_in_one_ordered_idempotent_unit(tmp_path):
     assert replay["duplicate_operation_ids"] == ["first", "second"]
 
 
+def test_batch_rejects_stale_same_item_intent_without_blocking_other_items(tmp_path):
+    store = LaterStore(tmp_path / "later.sqlite3")
+    first = store.apply_batch("timmy", [{
+        "operation_id": "newer-device",
+        "action": "defer",
+        "item_id": "issue:r:1:",
+        "wake_at": "2026-08-12T09:00:00.000Z",
+        "base_revision": 0,
+    }])
+
+    replay = store.apply_batch("timmy", [
+        {
+            "operation_id": "stale-device",
+            "action": "defer",
+            "item_id": "issue:r:1:",
+            "wake_at": "2026-08-10T09:00:00.000Z",
+            "base_revision": 0,
+        },
+        {
+            "operation_id": "unrelated-item",
+            "action": "defer",
+            "item_id": "issue:r:2:",
+            "wake_at": "2026-08-11T09:00:00.000Z",
+            "base_revision": 0,
+        },
+    ])
+
+    assert first["revision"] == 1
+    assert replay == {
+        "revision": 2,
+        "records": {
+            "issue:r:1:": "2026-08-12T09:00:00.000Z",
+            "issue:r:2:": "2026-08-11T09:00:00.000Z",
+        },
+        "accepted_operation_ids": ["unrelated-item"],
+        "duplicate_operation_ids": [],
+        "rejected_operations": [
+            {"operation_id": "stale-device", "reason": "stale_intent"}
+        ],
+    }
+    duplicate = store.apply_batch("timmy", [{
+        "operation_id": "stale-device",
+        "action": "restore",
+        "item_id": "issue:r:2:",
+        "base_revision": 2,
+    }])
+    assert duplicate["duplicate_operation_ids"] == ["stale-device"]
+    assert duplicate["revision"] == 2
+    assert duplicate["records"] == replay["records"]
+
+
+def test_later_receipts_are_bounded_per_account_and_existing_schema_migrates(tmp_path):
+    path = tmp_path / "later.sqlite3"
+    with sqlite3.connect(path) as connection:
+        connection.execute(
+            "CREATE TABLE later_operations (login TEXT NOT NULL, operation_id TEXT NOT NULL, "
+            "PRIMARY KEY (login, operation_id))"
+        )
+        connection.execute(
+            "INSERT INTO later_operations(login, operation_id) VALUES ('timmy', 'legacy')"
+        )
+
+    store = LaterStore(path, operation_limit=2, clock=lambda: 1_000)
+    for index in range(4):
+        store.apply(
+            "timmy", f"op-{index}", "defer", f"issue:r:{index}:",
+            wake_at="2026-08-10T09:00:00.000Z",
+        )
+    store.apply(
+        "alexander", "other", "defer", "issue:r:99:",
+        wake_at="2026-08-10T09:00:00.000Z",
+    )
+
+    with sqlite3.connect(path) as connection:
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(later_operations)")}
+        timmy = connection.execute(
+            "SELECT operation_id FROM later_operations WHERE login = 'timmy' ORDER BY rowid"
+        ).fetchall()
+        alexander = connection.execute(
+            "SELECT operation_id FROM later_operations WHERE login = 'alexander'"
+        ).fetchall()
+
+    assert "created_at" in columns
+    assert timmy == [("op-2",), ("op-3",)]
+    assert alexander == [("other",)]
+
+
+def test_later_receipt_age_pruning_is_account_scoped(tmp_path):
+    clock = [1_000.0]
+    path = tmp_path / "later.sqlite3"
+    store = LaterStore(path, operation_retention_seconds=60, clock=lambda: clock[0])
+    store.apply("timmy", "old", "restore", "issue:r:1:")
+    store.apply("alexander", "other-old", "restore", "issue:r:2:")
+    clock[0] += 61
+    store.apply("timmy", "fresh", "restore", "issue:r:3:")
+
+    with sqlite3.connect(path) as connection:
+        timmy = connection.execute(
+            "SELECT operation_id FROM later_operations WHERE login = 'timmy'"
+        ).fetchall()
+        alexander = connection.execute(
+            "SELECT operation_id FROM later_operations WHERE login = 'alexander'"
+        ).fetchall()
+
+    assert timmy == [("fresh",)]
+    assert alexander == [("other-old",)]
+
+
 @pytest.mark.anyio
 async def test_authenticated_later_api_uses_confirmed_account_and_csrf(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
@@ -129,6 +237,21 @@ async def test_authenticated_later_api_uses_confirmed_account_and_csrf(monkeypat
                 "X-CSRF-Token": client.cookies["stackchain_csrf"],
             },
         )
+        stale = await client.patch(
+            "/api/v1/later",
+            json={
+                "operations": [{
+                    "operation_id": "offline-stale",
+                    "action": "restore",
+                    "item_id": "issue:stackchain/dashboard:363:",
+                    "base_revision": 0,
+                }],
+            },
+            headers={
+                "Origin": "https://test",
+                "X-CSRF-Token": client.cookies["stackchain_csrf"],
+            },
+        )
         fetched = await client.get("/api/v1/later")
 
     assert forbidden.status_code == 403
@@ -143,6 +266,10 @@ async def test_authenticated_later_api_uses_confirmed_account_and_csrf(monkeypat
         "duplicate_operation_ids": [],
         "rejected_operations": [],
     }
+    assert stale.json()["rejected_operations"] == [
+        {"operation_id": "offline-stale", "reason": "stale_intent"}
+    ]
+    assert stale.json()["revision"] == 2
     assert fetched.json() == {"revision": 2, "records": {
         "issue:stackchain/dashboard:363:": "2026-08-10T09:00:00.000Z",
         "issue:stackchain/dashboard:365:": "2026-08-11T09:00:00.000Z",

tests/test_later_sync.py

@@ -90,12 +90,67 @@ sync.enqueue('defer','issue:r:2:','2026-08-10T09:00:00.000Z');
         "action": "defer",
         "item_id": "issue:r:2:",
         "wake_at": "2026-08-10T09:00:00.000Z",
+        "base_revision": 0,
     }]}
     assert result["records"]["issue:r:2:"] == "2026-08-10T09:00:00.000Z"
     assert result["status"] == "saved"
     assert result["pending"] == []
 
 
+def test_later_queue_expires_ancient_edits_before_replay():
+    script = f"""
+const createLaterSync=require({json.dumps(str(LATER_SYNC))});
+const values=new Map();const statuses=[];const requests=[];
+const storage={{get length(){{return values.size}},key:i=>[...values.keys()][i]||null,
+ getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}};
+const sync=createLaterSync({{storage,getLogin:()=> 'timmy',createOperationId:()=> 'old-later',
+ now:()=>1_800_000_000_000,maxOfflineMs:1000,
+ fetchJson:async(_url,options={{}})=>{{requests.push(options.method||'GET');return {{revision:4,records:{{}}}}}},
+ onRemoteRecords:()=>{{}},onStatus:(state,detail={{}})=>statuses.push([state,detail.count||0])}});
+(async()=>{{await sync.flush();sync.enqueue('restore','issue:r:2:');
+const key=[...values.keys()].find(value=>value.includes('.operation.'));
+const record=JSON.parse(values.get(key));record.queued_at=1_799_999_000_000;values.set(key,JSON.stringify(record));
+await sync.flush();process.stdout.write(JSON.stringify({{requests,statuses,pending:sync.pending()}}));}})();
+"""
+    result = run_node(script)
+
+    assert result["requests"] == ["GET", "GET"]
+    assert ["expired", 1] in result["statuses"]
+    assert result["statuses"][-1] == ["expired", 1]
+    assert result["pending"] == []
+
+
+def test_stale_receipt_adopts_server_truth_and_reports_another_device_conflict():
+    script = f"""
+const createLaterSync=require({json.dumps(str(LATER_SYNC))});
+const values=new Map();const requests=[];const statuses=[];const adopted=[];
+const storage={{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}};
+const sync=createLaterSync({{
+ storage,getLogin:()=> 'timmy',createOperationId:()=> 'offline-old',
+ fetchJson:async(_url,options={{}})=>{{
+   if(!options.method)return {{revision:3,records:{{'issue:r:1:':'2026-08-12T09:00:00.000Z'}}}};
+   const body=JSON.parse(options.body);requests.push(body);
+   return {{revision:4,records:{{'issue:r:1:':'2026-08-13T09:00:00.000Z'}},accepted_operation_ids:[],duplicate_operation_ids:[],rejected_operations:[{{operation_id:'offline-old',reason:'stale_intent'}}]}};
+ }},onRemoteRecords:records=>adopted.push(records),onStatus:(state,detail={{}})=>statuses.push([state,detail]),
+}});
+(async()=>{{await sync.flush();sync.enqueue('defer','issue:r:1:','2026-08-10T09:00:00.000Z');await sync.flush();
+process.stdout.write(JSON.stringify({{requests,statuses,adopted,pending:sync.pending()}}));}})();
+"""
+    result = run_node(script)
+    assert result["requests"] == [{"operations": [{
+        "operation_id": "offline-old",
+        "action": "defer",
+        "item_id": "issue:r:1:",
+        "wake_at": "2026-08-10T09:00:00.000Z",
+        "base_revision": 3,
+    }]}]
+    assert result["statuses"][-1] == ["conflict", {"count": 1}]
+    assert result["adopted"][-1] == {
+        "issue:r:1:": "2026-08-13T09:00:00.000Z"
+    }
+    assert result["pending"] == []
+
+
 def test_pending_later_edits_are_sent_as_one_batch_without_a_preflight_get():
     script = f"""
 const createLaterSync=require({json.dumps(str(LATER_SYNC))});
@@ -136,6 +191,7 @@ sync.enqueue('restore','issue:r:2:');
                 "action": "restore",
                 "item_id": "issue:r:2:",
                 "wake_at": None,
+                "base_revision": 0,
             }
         ],
         "status": "retrying",
@@ -283,10 +339,13 @@ async def test_dashboard_syncs_every_later_change_and_exposes_account_status():
     assert "laterSync.migrate(laterWork.read());" in html
     assert "laterSync.flush();" in html
     assert "Later saved to account." in html
+    assert "Another device changed this Later item" in html
+    assert "Today edit expired after 30 days offline" in html
+    assert "Later edit expired after 30 days offline" in html
 
 
 def test_later_sync_ships_atomically_in_the_offline_shell():
     source = (Path(__file__).parents[1] / "frontend" / "service-worker.js").read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/later-sync.js'" in source

tests/test_live_snapshot.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 
+import httpx
 import pytest
 
 from src import main
@@ -90,9 +91,9 @@ async def test_live_snapshot_fetches_user_once_and_updates_work_and_activity(mon
     assert result["sections"] == {
         "context": "fresh", "events": "fresh", "notifications": "fresh"
     }
-    assert result["revisions"] == {
-        "context": 1, "events": 1, "notifications": 1
-    }
+    assert set(result["revisions"]) == {"context", "events", "notifications"}
+    assert len(set(result["revisions"].values())) == 1
+    assert result["revisions"]["context"].endswith(".1")
 
 
 @pytest.mark.anyio
@@ -129,6 +130,56 @@ async def test_live_snapshot_omits_section_bodies_at_known_revisions(monkeypatch
     assert "freshness" in second
 
 
+@pytest.mark.anyio
+async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch):
+    async def user():
+        return {"id": 1, "login": "timmy"}
+
+    async def empty():
+        return []
+
+    monkeypatch.setattr(main, "current_user", user)
+    monkeypatch.setattr(main, "repos", empty)
+    monkeypatch.setattr(main, "issues", empty)
+    monkeypatch.setattr(main, "pull_requests", empty)
+    monkeypatch.setattr(main, "activity_events", lambda _user: empty())
+    monkeypatch.setattr(main, "notifications", empty)
+    monkeypatch.setattr(main, "_live_revision_generation", "worker-a")
+
+    worker_a = payload(await main.live_snapshot())
+    worker_a_token = worker_a["revisions"]["context"]
+
+    main._live_snapshot_value = None
+    main._live_snapshot_created_at = None
+    main._live_section_created_at = {
+        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_section_revisions = {
+        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    monkeypatch.setattr(main, "_live_revision_generation", "worker-b")
+
+    worker_b = payload(await main.live_snapshot(context_revision=worker_a_token))
+
+    assert worker_a_token == "worker-a.1"
+    assert worker_b["revisions"]["context"] == "worker-b.1"
+    assert worker_b["context"]["user"]["login"] == "timmy"
+
+
+@pytest.mark.anyio
+async def test_live_snapshot_rejects_malformed_revision_before_upstream_work(monkeypatch):
+    async def unexpected_user():
+        raise AssertionError("malformed revision reached upstream work")
+
+    monkeypatch.setattr(main, "current_user", unexpected_user)
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        response = await client.get("/api/v1/live?context_revision=not-a-token")
+
+    assert response.status_code == 422
+
+
 @pytest.mark.anyio
 async def test_live_snapshot_keeps_fresh_context_when_activity_fails(monkeypatch):
     async def user():

tests/test_markdown_renderer.py

@@ -137,4 +137,4 @@ def test_markdown_work_bodies_are_mobile_safe_block_containers():
     assert ".markdown-content { min-width:0; max-width:100%; overflow-wrap:anywhere;" in css
     assert ".markdown-content pre { max-width:100%; overflow-x:auto;" in css
     assert ".markdown-content a { min-height:44px;" in css
-    assert "stackchain-dashboard-shell-v54" in worker
+    assert "stackchain-dashboard-shell-v56" in worker

tests/test_mobile_composer_integration.py

@@ -35,4 +35,4 @@ def test_offline_shell_contains_every_local_dashboard_runtime_asset():
     shell_assets = set(re.findall(r"BASE \+ '([^']+)'", worker.split("async function sessionCsrf", 1)[0]))
 
     assert local_assets <= shell_assets, f"Offline shell is missing: {sorted(local_assets - shell_assets)}"
-    assert "stackchain-dashboard-shell-v54" in worker
+    assert "stackchain-dashboard-shell-v56" in worker

tests/test_readme.py

@@ -57,3 +57,12 @@ def test_readme_documents_offline_unread_update_conversations_and_boundaries():
     assert "previously opened unread update" in text
     assert "replies enter the account-bound durable outbox" in text
     assert "mark read, ownership, deferral, and older-message loading remain disabled" in text
+
+
+def test_readme_documents_planning_sync_retention_contract():
+    text = " ".join(README.read_text().split())
+
+    assert "30 days" in text
+    assert "4,096 operation receipts per account" in text
+    assert "expired edit" in text
+    assert "account plan is kept" in text

tests/test_service_worker.py

@@ -108,7 +108,7 @@ async function dispatchNotificationClick(route) {{
 def test_duplicate_aware_capture_ships_in_a_new_offline_shell():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/create-issue-sheet.js'" in source
     assert "BASE + 'static/dashboard.js'" in source
 
@@ -116,14 +116,14 @@ def test_duplicate_aware_capture_ships_in_a_new_offline_shell():
 def test_exact_later_picker_ships_atomically_in_a_new_offline_shell():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/later-picker.js'" in source
 
 
 def test_navigation_deadline_ships_in_a_new_shell_cache():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/dashboard.css'" in source
     assert "BASE + 'static/dashboard.js'" in source
     assert "BASE + 'static/install-app.js'" in source
@@ -132,21 +132,21 @@ def test_navigation_deadline_ships_in_a_new_shell_cache():
 def test_today_convergence_ships_in_a_new_shell_cache():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/today-sync.js'" in source
 
 
 def test_mobile_search_viewport_ships_in_a_new_offline_shell():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/mobile-search-viewport.js'" in source
 
 
 def test_update_ownership_flow_ships_atomically_in_a_new_offline_shell():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/update-ownership.js'" in source
 
 

tests/test_shared_context_snapshot.py

@@ -23,7 +23,7 @@ async def test_one_live_snapshot_updates_work_and_activity_on_one_timer():
 
     assert "fetch('api/v1/live'" in html
     assert "onSnapshot: renderLiveSnapshot" in html
-    assert "section + '_revision'" in html
+    assert "createContextPoller.buildRevisionQuery(revisions)" in html
     assert "workChanged = contextChanged || notificationsChanged" in html
     assert "renderContextSnapshot(snapshot.context)" in html
     assert "paintEventStream(snapshot.events)" in html

tests/test_today_store.py

@@ -107,6 +107,96 @@ def test_batch_applies_ordered_operations_once_and_rejects_only_capacity_conflic
     assert replay["duplicate_operation_ids"] == ["two", "three"]
 
 
+def test_operation_receipts_are_bounded_per_account_and_existing_schema_migrates(tmp_path):
+    path = tmp_path / "today.sqlite3"
+    with sqlite3.connect(path) as connection:
+        connection.execute(
+            "CREATE TABLE today_operations (login TEXT NOT NULL, operation_id TEXT NOT NULL, "
+            "PRIMARY KEY (login, operation_id))"
+        )
+        connection.execute(
+            "INSERT INTO today_operations(login, operation_id) VALUES ('timmy', 'legacy')"
+        )
+
+    store = TodayStore(path, limit=20, operation_limit=3, clock=lambda: 1_000)
+    for index in range(5):
+        store.apply("timmy", f"op-{index}", "add", f"issue:r:{index}:")
+    store.apply("alexander", "other", "add", "issue:r:99:")
+
+    with sqlite3.connect(path) as connection:
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(today_operations)")}
+        timmy = connection.execute(
+            "SELECT operation_id FROM today_operations WHERE login = 'timmy' ORDER BY rowid"
+        ).fetchall()
+        alexander = connection.execute(
+            "SELECT operation_id FROM today_operations WHERE login = 'alexander'"
+        ).fetchall()
+
+    assert "created_at" in columns
+    assert timmy == [("op-2",), ("op-3",), ("op-4",)]
+    assert alexander == [("other",)]
+
+
+def test_pruned_stale_today_operation_cannot_reorder_a_newer_plan(tmp_path):
+    store = TodayStore(tmp_path / "today.sqlite3", limit=5, operation_limit=1)
+    store.apply_batch("timmy", [
+        {"operation_id": "seed-1", "action": "add", "item_id": "issue:r:1:", "base_revision": 0},
+        {"operation_id": "seed-2", "action": "add", "item_id": "issue:r:2:", "base_revision": 0},
+    ])
+    store.apply_batch("timmy", [{
+        "operation_id": "offline-move", "action": "move", "item_id": "issue:r:2:",
+        "direction": "up", "base_revision": 2,
+    }])
+    newer = store.apply_batch("timmy", [{
+        "operation_id": "newer-move", "action": "move", "item_id": "issue:r:1:",
+        "direction": "up", "base_revision": 3,
+    }])
+
+    replay = store.apply_batch("timmy", [{
+        "operation_id": "offline-move", "action": "move", "item_id": "issue:r:2:",
+        "direction": "up", "base_revision": 2,
+    }])
+
+    assert replay["ids"] == newer["ids"] == ["issue:r:1:", "issue:r:2:"]
+    assert replay["revision"] == 4
+    assert replay["accepted_operation_ids"] == []
+    assert replay["rejected_operations"] == [
+        {"operation_id": "offline-move", "reason": "stale_intent"}
+    ]
+
+
+def test_today_receipt_age_pruning_is_account_scoped(tmp_path):
+    clock = [1_000.0]
+    path = tmp_path / "today.sqlite3"
+    store = TodayStore(
+        path, limit=5, operation_retention_seconds=60, clock=lambda: clock[0]
+    )
+    store.apply("timmy", "old", "add", "issue:r:1:")
+    store.apply("alexander", "other-old", "add", "issue:r:2:")
+    clock[0] += 61
+    store.apply("timmy", "fresh", "remove", "issue:r:1:")
+
+    with sqlite3.connect(path) as connection:
+        timmy = connection.execute(
+            "SELECT operation_id FROM today_operations WHERE login = 'timmy'"
+        ).fetchall()
+        alexander = connection.execute(
+            "SELECT operation_id FROM today_operations WHERE login = 'alexander'"
+        ).fetchall()
+
+    assert timmy == [("fresh",)]
+    assert alexander == [("other-old",)]
+
+
+def test_today_api_model_preserves_client_base_revision():
+    operation = main.TodayOperation(
+        operation_id="offline", action="move", item_id="issue:r:2:",
+        direction="up", base_revision=7,
+    )
+
+    assert operation.model_dump()["base_revision"] == 7
+
+
 @pytest.mark.anyio
 async def test_authenticated_today_api_uses_confirmed_account_and_csrf(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")

tests/test_today_sync.py

@@ -74,8 +74,8 @@ sync.enqueue('add', 'issue:r:1:');
     assert result == {
         "sharesInflight": True,
         "patches": [
-            {"operation_id": "op-1", "action": "add", "item_id": "issue:r:1:", "direction": None},
-            {"operation_id": "op-2", "action": "add", "item_id": "issue:r:2:", "direction": None},
+            {"operation_id": "op-1", "action": "add", "item_id": "issue:r:1:", "direction": None, "base_revision": 0},
+            {"operation_id": "op-2", "action": "add", "item_id": "issue:r:2:", "direction": None, "base_revision": 0},
         ],
         "pending": [],
         "ids": ["issue:r:1:", "issue:r:2:"],
@@ -86,7 +86,7 @@ sync.enqueue('add', 'issue:r:1:');
 def test_inflight_today_drain_ships_in_a_new_offline_shell():
     source = (Path(__file__).parents[1] / "frontend" / "service-worker.js").read_text()
 
-    assert "stackchain-dashboard-shell-v54" in source
+    assert "stackchain-dashboard-shell-v56" in source
     assert "BASE + 'static/today-sync.js'" in source
 
 
@@ -137,12 +137,41 @@ sync.enqueue('add', 'issue:r:2:');
         "action": "add",
         "item_id": "issue:r:2:",
         "direction": None,
+        "base_revision": 0,
     }]}
     assert result["adopted"] == ["issue:r:9:", "issue:r:2:"]
     assert result["status"] == "saved"
     assert result["pending"] == []
 
 
+def test_today_operations_capture_server_revision_and_expire_before_replay():
+    script = f"""
+const createTodaySync = require({json.dumps(str(TODAY_SYNC))});
+const values = new Map(); const statuses=[]; const requests=[];
+const storage={{get length(){{return values.size}},key:i=>[...values.keys()][i]||null,
+ getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}};
+const sync=createTodaySync({{storage,getLogin:()=> 'timmy',createOperationId:()=> 'old-op',
+ now:()=>1_800_000_000_000,maxOfflineMs:1000,
+ fetchJson:async(_url,options={{}})=>{{requests.push(options.method||'GET');return {{revision:7,ids:['server']}}}},
+ onRemoteIds:()=>{{}},onStatus:(state,detail={{}})=>statuses.push([state,detail.count||0])}});
+(async()=>{{await sync.flush();sync.enqueue('move','issue:r:2:','up');
+const pendingBefore=sync.pending();
+const key=[...values.keys()].find(value=>value.includes('.operation.'));
+const record=JSON.parse(values.get(key));record.queued_at=1_799_999_000_000;values.set(key,JSON.stringify(record));
+await sync.flush();process.stdout.write(JSON.stringify({{pendingBefore,requests,statuses,pendingAfter:sync.pending()}}));}})();
+"""
+    result = json.loads(subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True).stdout)
+
+    assert result["pendingBefore"] == [{
+        "operation_id": "old-op", "action": "move", "item_id": "issue:r:2:",
+        "direction": "up", "base_revision": 7,
+    }]
+    assert result["requests"] == ["GET", "GET"]
+    assert ["expired", 1] in result["statuses"]
+    assert result["statuses"][-1] == ["expired", 1]
+    assert result["pendingAfter"] == []
+
+
 def test_pending_today_edits_are_sent_as_one_batch_without_a_preflight_get():
     script = f"""
 const createTodaySync = require({json.dumps(str(TODAY_SYNC))});
@@ -184,6 +213,7 @@ sync.enqueue('move', 'issue:r:2:', 'up');
             "action": "move",
             "item_id": "issue:r:2:",
             "direction": "up",
+            "base_revision": 0,
         }],
         "status": "retrying",
     }
@@ -221,7 +251,7 @@ sync.enqueue('add','issue:r:2:');
     assert result["scheduled"] == [2000]
     assert result["attempts"] == 2
     assert result["patches"] == [{
-        "operation_id": "stable-op", "action": "add", "item_id": "issue:r:2:", "direction": None,
+        "operation_id": "stable-op", "action": "add", "item_id": "issue:r:2:", "direction": None, "base_revision": 0,
     }]
     assert result["pending"] == []
     assert ["retrying", 2000] in result["statuses"]
@@ -254,6 +284,7 @@ process.stdout.write(JSON.stringify({{first, duplicate, pending:sync.pending(),
             "action": "remove",
             "item_id": "issue:r:1:",
             "direction": None,
+            "base_revision": 0,
         }],
         "statuses": ["pending", "pending"],
     }
@@ -280,8 +311,8 @@ process.stdout.write(JSON.stringify({{first, second, pending:sync.pending()}}));
         "first": True,
         "second": False,
         "pending": [
-            {"operation_id": "migration-1", "action": "add", "item_id": "issue:r:1:", "direction": None},
-            {"operation_id": "migration-2", "action": "add", "item_id": "issue:r:2:", "direction": None},
+            {"operation_id": "migration-1", "action": "add", "item_id": "issue:r:1:", "direction": None, "base_revision": 0},
+            {"operation_id": "migration-2", "action": "add", "item_id": "issue:r:2:", "direction": None, "base_revision": 0},
         ],
     }
 
@@ -346,8 +377,8 @@ sync.enqueue('remove', '5');
     assert result == {
         "result": False,
         "patches": [
-            {"operation_id": "op-1", "action": "add", "item_id": "6", "direction": None},
-            {"operation_id": "op-2", "action": "remove", "item_id": "5", "direction": None},
+            {"operation_id": "op-1", "action": "add", "item_id": "6", "direction": None, "base_revision": 0},
+            {"operation_id": "op-2", "action": "remove", "item_id": "5", "direction": None, "base_revision": 0},
         ],
         "pending": [],
         "ids": ["1", "2", "3", "4"],

tests/test_views.py

@@ -2,7 +2,7 @@ from pathlib import Path
 
 import pytest
 
-from src.views import dashboard
+from src.views import FRONTEND_BUILD, dashboard
 
 
 @pytest.mark.anyio
@@ -28,7 +28,7 @@ async def test_dashboard_bootstrap_uses_csp_compatible_static_assets():
     html = await dashboard()
 
     assert '<link rel="stylesheet" href="static/dashboard.css" />' in html
-    assert '<script src="static/dashboard.js"></script>' in html
+    assert f'<script src="{FRONTEND_BUILD.runtime_name}"></script>' in html
     assert "<style>" not in html
     assert "<script>" not in html
 
@@ -42,7 +42,7 @@ async def test_global_search_preview_is_a_phone_safe_accessible_dialog():
     assert 'aria-labelledby="search-preview-title"' in html
     assert 'id="claim-search-result"' in html
     assert 'id="open-search-result-gitea"' in html
-    assert 'src="static/search-preview.js"' in html
+    assert "static/search-preview.js" in FRONTEND_BUILD.page_sources
     assert ".search-preview-panel" in css
     assert "height:100dvh" in css
     assert "env(safe-area-inset-bottom)" in css