fix(t099): notification & transport reliability

.gitignore

@@ -23,6 +23,7 @@ playwright-report/
 # Visual-review scratch (screenshots)
 .scratch/
 
-# Web build runtime state (Slack workspace registry + push subscriptions)
+# Web build runtime state (Slack workspace registry + push subscriptions + sweep watermark)
 slack-workspaces.json
+slack-sweep-state.json
 web-push-subs.json

CLAUDE.md

@@ -65,12 +65,16 @@ cdp-browser/
 ├── preload.js           # IPC bridge (contextBridge)
 ├── core/                # Backend-agnostic shared CommonJS modules (consumed by main.js + web/server.mjs)
 │   ├── notifications.js           # Pure notification logic (dedup, cap, OS-toast gating, Slack workspace key: parseSlackContext/slackGroupKey)
-│   ├── notifications-sidechain.js # Notification Side-Channel state machine (createNotificationCenter, DI); Slack-only cred extraction (xoxc+d-cookie, carries enterpriseId for Grid grouping t092) via onCreds dep + listCreds/getCreds/markCredsStale/setSelfUserId accessors (t069). Side-channel cdpCall has a timeout + rejects pending on close, and reconcile reaps a hung non-OPEN socket on a still-live target (t096)
+│   ├── notifications-sidechain.js # Notification Side-Channel state machine (createNotificationCenter, DI); Slack-only cred extraction (xoxc+d-cookie, carries enterpriseId for Grid grouping t092) via onCreds dep + listCreds/getCreds/markCredsStale/setSelfUserId accessors (t069). markCredsStale re-extracts over the live socket (refreshCreds) so a rotated token self-heals; recordCreds fires onCredsStuck when the re-extract reads the same stale token so the server reloads the keeper-owned parked tab, never a pin (t099). Side-channel cdpCall has a timeout + rejects pending on close, and reconcile reaps a hung non-OPEN socket on a still-live target (t096)
 │   ├── remote-page-connector.js   # Remote Page connect choreography (createRemotePageConnector, DI) + screencast rate ceiling (SCREENCAST_TARGET_FPS/EVERY_NTH_FRAME)
 │   ├── theme-emulation.js         # Pure theme-sync logic (emulatedMediaParams)
 │   ├── cdp-endpoints.js           # Pure /json URL builders
 │   ├── settings-store.js          # Pure settings/pins/ui-state store (device-suffixed ui-state keys pass by prefix allowlist, t093)
 │   ├── line-splitter.js           # Pure NDJSON frame reassembly for the streaming input channel
+│   ├── atomic-write.js            # Atomic write-temp-then-rename (crash-safe JSON persistence); shared by server + main (t099)
+│   ├── ws-backpressure.js         # Pure WS fan-out predicates: shouldSkipClient (over-buffer drop) + isClientDead (heartbeat reap) — half-open sleeping client can't buffer frames unbounded (t099)
+│   ├── request-guards.js          # Pure mutation-payload shape guards: isValidConfig/isValidPinsArray — a malformed body can't wipe config/pins (t099)
+│   ├── slack-sweep-state.js       # Persisted Slack sweep {watermark, seeded} — serialize/deserialize + DI debounced persister; restart resumes from watermark instead of re-seeding (t099, ADR-0016)
 │   ├── frame-throttle.js          # Pure screencast rate throttle (createFrameThrottle, DI clock; fresh-frame-wins)
 │   ├── frame-ack-gate.js          # Pure one-in-flight gate + watchdog for WS paint-ack backpressure (t056)
 │   ├── paint-ack-pacer.js         # Pure adaptive paint-ack watchdog window: EWMA of observed paint latency sizes the stranded-paint timeout (floor 1s, cap 5s) so a slow device isn't tripped early (t096, P2)

core/atomic-write.js

@@ -0,0 +1,15 @@
+// Atomic file write: write to a temp sibling then rename (atomic on the same filesystem), so
+// a crash mid-write can't truncate/reset the target JSON (settings, push subs, notifications,
+// Slack registry + sweep state). Shared by web/server.mjs and main.js. `fs` is injectable for
+// tests. A leftover `.tmp` from a prior crash is simply overwritten on the next write.
+const fs = require("fs")
+
+function atomicWriteFileSync(path, data, deps = {}) {
+  const writeFileSync = deps.writeFileSync || fs.writeFileSync
+  const renameSync = deps.renameSync || fs.renameSync
+  const tmp = `${path}.tmp`
+  writeFileSync(tmp, data)
+  renameSync(tmp, path)
+}
+
+module.exports = { atomicWriteFileSync }

core/atomic-write.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it, vi } from "vitest"
+// @ts-expect-error — CJS module, no types
+import { atomicWriteFileSync } from "./atomic-write.js"
+
+describe("atomicWriteFileSync", () => {
+  it("writes to a temp sibling then renames onto the target", () => {
+    const calls: string[] = []
+    const writeFileSync = vi.fn((p: string) => calls.push(`write:${p}`))
+    const renameSync = vi.fn((from: string, to: string) => calls.push(`rename:${from}->${to}`))
+
+    atomicWriteFileSync("/data/settings.json", "{}", { writeFileSync, renameSync })
+
+    expect(writeFileSync).toHaveBeenCalledWith("/data/settings.json.tmp", "{}")
+    expect(renameSync).toHaveBeenCalledWith("/data/settings.json.tmp", "/data/settings.json")
+    // write must happen before rename
+    expect(calls).toEqual([
+      "write:/data/settings.json.tmp",
+      "rename:/data/settings.json.tmp->/data/settings.json",
+    ])
+  })
+
+  it("does not rename (leaving the original intact) when the write throws", () => {
+    const writeFileSync = vi.fn(() => {
+      throw new Error("disk full")
+    })
+    const renameSync = vi.fn()
+
+    expect(() => atomicWriteFileSync("/data/x.json", "{}", { writeFileSync, renameSync })).toThrow(
+      "disk full",
+    )
+    expect(renameSync).not.toHaveBeenCalled()
+  })
+})

core/notifications-sidechain.js

@@ -95,6 +95,9 @@ function createNotificationCenter(deps) {
   const persist = () => save(notifications)
 
   const sideChannels = new Map() // targetId -> ws
+  // Live cred-capable (Slack) sockets: targetId -> { cdpCall, target }. Lets markCredsStale
+  // re-run extraction over an already-open socket without waiting for a reconnect (t099).
+  const credSockets = new Map()
   // Slack credential records keyed by teamId (t069). Populated by side-channel extraction;
   // read by the server-side content sweep (t071). Each: { teamId, token, cookie, name, url,
   // enterpriseId, fresh, lastError, selfUserId? }. Creds are the user's own session secrets
@@ -109,8 +112,17 @@ function createNotificationCenter(deps) {
   const WS_OPEN = WebSocketCtor && WebSocketCtor.OPEN != null ? WebSocketCtor.OPEN : 1
 
   function recordCreds(team, cookie) {
-    const prev = credsByTeam.get(team.teamId) || {}
-    const rec = markFresh(prev, {
+    const prev = credsByTeam.get(team.teamId)
+    // If a re-extraction (after a 401) produced the SAME token that was already stale, the
+    // live tab's localConfig itself is stale — re-extract can't fix it. Signal the server to
+    // reload the keeper-owned parked tab so Slack regenerates a token (t099); keep the record
+    // stale (don't mark a known-bad token fresh). A changed token gets a fresh chance below.
+    if (prev && prev.fresh === false && prev.token === team.token) {
+      log(`[slack-creds] ${team.teamId} still stale after re-extract — signalling reload`)
+      if (deps.onCredsStuck) deps.onCredsStuck(team.teamId)
+      return
+    }
+    const rec = markFresh(prev || {}, {
       teamId: team.teamId,
       token: team.token,
       cookie,
@@ -233,8 +245,10 @@ function createNotificationCenter(deps) {
       // document-start for future loads + the already-loaded document.
       cdp("Page.addScriptToEvaluateOnNewDocument", { source: sourceFor(adapter) })
       cdp("Runtime.evaluate", { expression: sourceFor(adapter) })
-      // Slack only: pull the workspace creds for the server-side sweep (ADR-0011).
+      // Slack only: pull the workspace creds for the server-side sweep (ADR-0011). Keep the
+      // live cdpCall handle so a later 401 can re-extract over this same socket (t099).
       if (adapter.extractCreds && deps.onCreds) {
+        credSockets.set(target.id, { cdpCall, target })
         extractSlackCreds(cdpCall, target).catch(() => {})
       }
     })
@@ -253,6 +267,7 @@ function createNotificationCenter(deps) {
     })
     const drop = () => {
       if (sideChannels.get(target.id) === ws) sideChannels.delete(target.id)
+      credSockets.delete(target.id)
       // Free any in-flight awaitable call so a stalled socket can't leak its promise.
       for (const p of pending.values()) p.reject(new Error("side-channel closed"))
       pending.clear()
@@ -287,6 +302,21 @@ function createNotificationCenter(deps) {
     }
   }
 
+  // Re-extract creds over any already-open Slack side-channel socket (t099). One live tab's
+  // localConfig_v2 + shared `d` cookie refresh EVERY workspace, so the first live socket wins.
+  // Called on a 401 (markCredsStale) so a rotated xoxc token self-heals within a sweep cycle
+  // without waiting for a reconnect. Best-effort; resolves false when no live socket exists.
+  async function refreshCreds() {
+    for (const [id, ws] of sideChannels) {
+      if (ws.readyState !== WS_OPEN) continue
+      const handle = credSockets.get(id)
+      if (!handle) continue
+      await extractSlackCreds(handle.cdpCall, handle.target)
+      return true
+    }
+    return false
+  }
+
   // Attach to newly-seen adapter-matching Tab targets, drop vanished/changed ones.
   // `targets` is optional; when omitted, the injected `listTargets()` is called.
   async function reconcile(targets) {
@@ -356,12 +386,17 @@ function createNotificationCenter(deps) {
     // Slack cred accessors (t069) — the server-side sweep (t071) reads these.
     listCreds: () => [...credsByTeam.values()],
     getCreds: (teamId) => credsByTeam.get(teamId) || null,
-    // Mark a workspace's creds stale (e.g. the sweep got a 401) so the health surface
-    // (t074) can flag it and the parked-tab keeper (t070) re-extracts. Keeps the last creds.
+    // Mark a workspace's creds stale (e.g. the sweep got a 401) so the health surface (t074)
+    // can flag it, then immediately re-extract over a live Slack socket so a rotated token
+    // self-heals within a sweep cycle (t099). Keeps the last creds until a fresh read replaces
+    // them. If the re-extract reads the same stale token, recordCreds fires deps.onCredsStuck.
     markCredsStale: (teamId, reason) => {
       const prev = credsByTeam.get(teamId)
       if (prev) credsByTeam.set(teamId, markStale(prev, reason))
+      refreshCreds().catch(() => {})
     },
+    // Force a re-extraction over any live Slack socket (t099) — exposed for the server + tests.
+    refreshCreds,
     // The sweep caches the resolved self user id (for channel @-mention parity) here.
     setSelfUserId: (teamId, selfUserId) => {
       const prev = credsByTeam.get(teamId)

core/notifications-sidechain.test.ts

@@ -446,6 +446,50 @@ describe("slack cred extraction (t069)", () => {
     center.setSelfUserId("T0EXAMPLE02", "U_ME")
     expect(center.getCreds("T0EXAMPLE02")?.selfUserId).toBe("U_ME")
   })
+
+  it("markCredsStale re-extracts over the live socket; a rotated token clears the stale state (t099)", async () => {
+    const onCreds = vi.fn()
+    const onCredsStuck = vi.fn()
+    const { center } = makeCenter({ onCreds, onCredsStuck, onSlackSignal: vi.fn() })
+    await center.reconcile([slackTarget()])
+    const ws = FakeWs.instances[0]
+    ws.open()
+    await answerCredExtraction(ws)
+
+    center.markCredsStale("T0EXAMPLE02", "invalid_auth") // fires refreshCreds()
+    await Promise.resolve()
+    // Slack rotated the token in localConfig — the re-extract reads the fresh value.
+    const rotated = JSON.stringify({
+      teams: { T0EXAMPLE02: { token: "xoxc-NEW", name: "Acme", url: "https://acme.slack.com/" } },
+    })
+    await answerCredExtraction(ws, { config: rotated })
+
+    expect(center.getCreds("T0EXAMPLE02")).toMatchObject({ token: "xoxc-NEW", fresh: true })
+    expect(onCredsStuck).not.toHaveBeenCalled()
+  })
+
+  it("signals onCredsStuck when the re-extract reads the SAME stale token (t099)", async () => {
+    const onCreds = vi.fn()
+    const onCredsStuck = vi.fn()
+    const { center } = makeCenter({ onCreds, onCredsStuck, onSlackSignal: vi.fn() })
+    await center.reconcile([slackTarget()])
+    const ws = FakeWs.instances[0]
+    ws.open()
+    await answerCredExtraction(ws)
+
+    center.markCredsStale("T0EXAMPLE02", "invalid_auth")
+    await Promise.resolve()
+    await answerCredExtraction(ws) // same LOCAL_CONFIG → same token, still stale
+
+    expect(onCredsStuck).toHaveBeenCalledWith("T0EXAMPLE02")
+    // A known-bad token is not marked fresh — the health surface stays degraded.
+    expect(center.getCreds("T0EXAMPLE02")?.fresh).toBe(false)
+  })
+
+  it("refreshCreds resolves false when no live Slack socket exists", async () => {
+    const { center } = makeCenter({ onCreds: vi.fn() })
+    expect(await center.refreshCreds()).toBe(false)
+  })
 })
 
 describe("slack hijack ↔ sweep handoff (t071)", () => {

core/push-subscriptions.js

@@ -7,9 +7,20 @@ export function reconcileDeviceId(existingSubs, incoming) {
   const existing = existingSubs.find((sub) => sub.endpoint === incoming.endpoint)
 
   if (existing && existing.deviceId) {
-    // Endpoint match wins; ignore any cached deviceId from the client
+    // Endpoint match wins — recovers a storage-wiped device that kept the same push endpoint.
+    // A cached client id that conflicts with the stored binding is ignored.
     return { deviceId: existing.deviceId, isNew: false }
   }
 
+  // New endpoint. Adopt the client's self-asserted deviceId when it has one: after a
+  // revocation/rotation the endpoint changes but the device keeps its id, and the per-device
+  // prefs live in ui-state keyed by that id (not in the subs file), so re-binding the id to
+  // the new endpoint restores them instead of orphaning. Single-user tailnet tool — the client
+  // is trusted to name its own device. Only mint when the client sends no id at all (legacy).
+  if (incoming.deviceId) {
+    const known = existingSubs.some((sub) => sub.deviceId === incoming.deviceId)
+    return { deviceId: incoming.deviceId, isNew: !known }
+  }
+
   return { deviceId: randomUUID(), isNew: true }
 }

core/push-subscriptions.test.mjs

@@ -105,6 +105,37 @@ describe("reconcileDeviceId", () => {
     expect(result2.isNew).toBe(false)
   })
 
+  it("adopts a client-asserted deviceId on a new endpoint (endpoint rotation / revocation recovery)", () => {
+    // A revoked subscription re-subscribes with a NEW endpoint but the same device keeps its
+    // known deviceId (localStorage intact). The per-device prefs live in ui-state keyed by that
+    // id, so re-binding it to the new endpoint recovers them instead of orphaning.
+    const existingSubs = [
+      { endpoint: "https://push.example.com/api/v1/old", deviceId: "device-keep" },
+    ]
+    const incoming = {
+      endpoint: "https://push.example.com/api/v1/rotated",
+      deviceId: "device-keep",
+    }
+
+    const result = reconcileDeviceId(existingSubs, incoming)
+
+    expect(result.deviceId).toBe("device-keep")
+    expect(result.isNew).toBe(false)
+  })
+
+  it("adopts a client-asserted deviceId even when unknown to the store (first subscribe with a client-minted id)", () => {
+    const existingSubs = []
+    const incoming = {
+      endpoint: "https://push.example.com/api/v1/fresh",
+      deviceId: "device_local_abc",
+    }
+
+    const result = reconcileDeviceId(existingSubs, incoming)
+
+    expect(result.deviceId).toBe("device_local_abc")
+    expect(result.isNew).toBe(true)
+  })
+
   it("does not modify the input arrays", () => {
     const existingSubs = [
       { endpoint: "https://push.example.com/api/v1/sub1", deviceId: "device-1" },

core/request-guards.js

@@ -0,0 +1,25 @@
+// Pure request-payload shape guards (t099). A malformed/undecryptable body is rejected upstream
+// (400); these guard the SHAPE of the dangerous mutation payloads so a syntactically-valid but
+// wrong body (e.g. an empty {} from a masked decrypt) can't persist and wipe config/pins.
+
+function isValidConfig(v) {
+  return (
+    !!v &&
+    typeof v === "object" &&
+    typeof v.host === "string" &&
+    v.host.trim().length > 0 &&
+    v.port != null &&
+    v.port !== "" &&
+    Number.isFinite(Number(v.port))
+  )
+}
+
+function isPinObject(v) {
+  return !!v && typeof v === "object" && typeof v.id === "string" && v.id.length > 0
+}
+
+function isValidPinsArray(v) {
+  return Array.isArray(v) && v.every(isPinObject)
+}
+
+module.exports = { isValidConfig, isPinObject, isValidPinsArray }

core/request-guards.test.ts

@@ -0,0 +1,35 @@
+import { describe, expect, it } from "vitest"
+// @ts-expect-error — CJS module, no types
+import { isValidConfig, isValidPinsArray } from "./request-guards.js"
+
+describe("isValidConfig", () => {
+  it("accepts a well-shaped host/port config", () => {
+    expect(isValidConfig({ host: "10.0.0.1", port: 9222 })).toBe(true)
+    expect(isValidConfig({ host: "example.test", port: "9222" })).toBe(true)
+  })
+
+  it("rejects the empty / masked-body object that would wipe the CDP address", () => {
+    expect(isValidConfig({})).toBe(false)
+  })
+
+  it("rejects a missing/blank host or non-numeric port", () => {
+    expect(isValidConfig({ host: "", port: 9222 })).toBe(false)
+    expect(isValidConfig({ host: "h", port: "nope" })).toBe(false)
+    expect(isValidConfig({ port: 9222 })).toBe(false)
+    expect(isValidConfig(null)).toBe(false)
+  })
+})
+
+describe("isValidPinsArray", () => {
+  it("accepts an array of pin objects with string ids", () => {
+    expect(isValidPinsArray([{ id: "p1" }, { id: "p2", url: "https://x" }])).toBe(true)
+    expect(isValidPinsArray([])).toBe(true)
+  })
+
+  it("rejects non-arrays and arrays with malformed members", () => {
+    expect(isValidPinsArray({})).toBe(false)
+    expect(isValidPinsArray(null)).toBe(false)
+    expect(isValidPinsArray([{ id: "p1" }, { url: "no-id" }])).toBe(false)
+    expect(isValidPinsArray([{ id: 5 }])).toBe(false)
+  })
+})