Merge remote-tracking branch 'origin/main' into brennanb2025/fix-stale-native-chat-queue

mobile/src/session/mobile-native-chat-controller-contract.ts

@@ -0,0 +1,64 @@
+import type { Dispatch, MutableRefObject, SetStateAction } from 'react'
+import type { detectAgentPermission } from './mobile-native-chat-permission'
+import type { parseAgentQuestion } from './mobile-native-chat-question'
+import type { AskAnswerSelection, AskPrompt, parseAskFromStatus } from './mobile-native-chat-ask'
+import type { MobileNativeChatSendOutcome } from './mobile-native-chat-send'
+import type { MobileNativeChatPendingMessage } from './use-mobile-native-chat-drafts'
+import type { useMobileNativeChatSession } from './use-mobile-native-chat-session'
+import type { MobileNativeChatSessionOptionPickersProps } from './MobileNativeChatSessionOptionPickers'
+
+export type MobileNativeChatController = {
+  /** Whether a tab's effective view is chat (per-tab override, else the default). */
+  isTabChatView: (tabId: string) => boolean
+  toggleTabChatView: (tabId: string) => void
+  showNativeChat: boolean
+  showNativeChatRef: MutableRefObject<boolean>
+  /** Resolved agent for the active chat tab (names the empty-state copy). */
+  nativeChatAgent: string | null
+  chatComposerText: string
+  setChatComposerText: Dispatch<SetStateAction<string>>
+  chatPending: MobileNativeChatPendingMessage[]
+  chatImagePreviewsByMessageId: Record<string, string[]>
+  nativeChatSession: ReturnType<typeof useMobileNativeChatSession>
+  nativeChatAgentWorking: boolean
+  nativeChatStreamingText?: string
+  /** Agent mid-turn, regardless of whether chat is the visible view. */
+  nativeChatStreamLive: boolean
+  /** Host/workspace/tab/session scope for stateful streaming suppression. */
+  nativeChatStreamScopeKey: string
+  nativeChatPermission: ReturnType<typeof detectAgentPermission>
+  nativeChatQuestion: ReturnType<typeof parseAgentQuestion>
+  /** The pending ask, already null while dismissed (dismissal lives here so it
+   *  survives the chat-view subtree unmounting on a view toggle). */
+  nativeChatAsk: ReturnType<typeof parseAskFromStatus>
+  /** Stable key for the current ask card (keys the card component). */
+  nativeChatAskKey: string | null
+  /** Hide the current ask until a genuinely different question arrives. */
+  dismissNativeChatAsk: () => void
+  handleNativeChatAnswerAsk: (
+    prompt: AskPrompt,
+    selections: AskAnswerSelection[]
+  ) => Promise<boolean>
+  handleNativeChatCancelAsk: () => Promise<boolean>
+  handleNativeChatRespondPermission: (text: string) => Promise<boolean>
+  handleNativeChatStop: () => void
+  nativeChatFilePaths: string[]
+  loadNativeChatFiles: (query: string) => void
+  handleNativeChatQuestionAnswer: (text: string) => Promise<boolean>
+  handleNativeChatSend: (text: string, images?: string[]) => Promise<boolean>
+  /** Outcome-preserving send: callers that pasted terminal input beforehand
+   *  (image sends) must see 'unknown' to heal a possibly-orphaned paste. Such a
+   *  caller passes its own `deadline` so the paste it already spent and this text
+   *  body share one budget instead of holding the composer for two. */
+  handleNativeChatSendWithOutcome: (
+    text: string,
+    images?: string[],
+    deadline?: number
+  ) => Promise<MobileNativeChatSendOutcome>
+  /** Launch-context text still parked on the agent's TUI input line, or null.
+   *  Image sends read it to size their leading clear (one Ctrl+U per line). */
+  readSeededLaunchDraft: () => string | null
+  /** Model/session-option pickers for the composer, or null when the active
+   *  agent has no session-option catalog. */
+  nativeChatSessionOptions: MobileNativeChatSessionOptionPickersProps | null
+}

mobile/src/session/use-mobile-native-chat-controller.ts

@@ -1,101 +1,29 @@
-import {
-  useCallback,
-  useLayoutEffect,
-  useRef,
-  type Dispatch,
-  type MutableRefObject,
-  type SetStateAction
-} from 'react'
+import { useCallback, useLayoutEffect, useRef, type MutableRefObject } from 'react'
 import { encodeNativeChatTranscriptIdentity } from '../../../src/shared/native-chat-transcript-retention'
 import { useMobileSessionViewMode } from './use-mobile-session-view-mode'
 import type { RpcClient } from '../transport/rpc-client'
 import type { ConnectionState } from '../transport/types'
-import {
-  parseAskFromStatus,
-  type AskAnswerSelection,
-  type AskPrompt
-} from './mobile-native-chat-ask'
 import { type MobileNativeChatTab, resolveMobileNativeChat } from './mobile-native-chat-eligibility'
-import { detectAgentPermission } from './mobile-native-chat-permission'
-import { parseAgentQuestion } from './mobile-native-chat-question'
 import { useMobileNativeChatPermissionSend } from './mobile-native-chat-permission-send'
-import type { MobileNativeChatSendOutcome } from './mobile-native-chat-send'
 import { useMobileNativeChatAnswerSend } from './use-mobile-native-chat-answer-send'
 import { useMobileNativeChatAskDismiss } from './use-mobile-native-chat-ask-dismiss'
 import { useMobileNativeChatCancelAsk } from './use-mobile-native-chat-cancel-ask'
-import {
-  useMobileNativeChatDrafts,
-  type MobileNativeChatPendingMessage
-} from './use-mobile-native-chat-drafts'
+import { useMobileNativeChatDrafts } from './use-mobile-native-chat-drafts'
 import { useMobileNativeChatFileSearch } from './use-mobile-native-chat-file-search'
 import { useMobileNativeChatMessageSend } from './use-mobile-native-chat-message-send'
 import { mobileNativeChatScopeKey } from './mobile-native-chat-scope-key'
 import { useMobileNativeChatSession } from './use-mobile-native-chat-session'
 import { useMobileNativeChatSessionOptions } from './use-mobile-native-chat-session-options'
-import type { MobileNativeChatSessionOptionPickersProps } from './MobileNativeChatSessionOptionPickers'
 import { useMobileNativeChatPrompts } from './use-mobile-native-chat-prompts'
 import { useMobileNativeChatStop } from './use-mobile-native-chat-stop'
 import { useNativeChatAcceptedAction } from './use-native-chat-action-outcomes'
 import { useThrottledLatestValue } from './use-throttled-latest-value'
+import type { MobileNativeChatController } from './mobile-native-chat-controller-contract'
+
+export type { MobileNativeChatController } from './mobile-native-chat-controller-contract'
 
 const NATIVE_CHAT_STREAM_THROTTLE_MS = 50
 
-export type MobileNativeChatController = {
-  /** Whether a tab's effective view is chat (per-tab override, else the default). */
-  isTabChatView: (tabId: string) => boolean
-  toggleTabChatView: (tabId: string) => void
-  showNativeChat: boolean
-  showNativeChatRef: MutableRefObject<boolean>
-  /** Resolved agent for the active chat tab (names the empty-state copy). */
-  nativeChatAgent: string | null
-  chatComposerText: string
-  setChatComposerText: Dispatch<SetStateAction<string>>
-  chatPending: MobileNativeChatPendingMessage[]
-  chatImagePreviewsByMessageId: Record<string, string[]>
-  nativeChatSession: ReturnType<typeof useMobileNativeChatSession>
-  nativeChatAgentWorking: boolean
-  nativeChatStreamingText?: string
-  /** Agent mid-turn, regardless of whether chat is the visible view. */
-  nativeChatStreamLive: boolean
-  /** Host/workspace/tab/session scope for stateful streaming suppression. */
-  nativeChatStreamScopeKey: string
-  nativeChatPermission: ReturnType<typeof detectAgentPermission>
-  nativeChatQuestion: ReturnType<typeof parseAgentQuestion>
-  /** The pending ask, already null while dismissed (dismissal lives here so it
-   *  survives the chat-view subtree unmounting on a view toggle). */
-  nativeChatAsk: ReturnType<typeof parseAskFromStatus>
-  /** Stable key for the current ask card (keys the card component). */
-  nativeChatAskKey: string | null
-  /** Hide the current ask until a genuinely different question arrives. */
-  dismissNativeChatAsk: () => void
-  handleNativeChatAnswerAsk: (
-    prompt: AskPrompt,
-    selections: AskAnswerSelection[]
-  ) => Promise<boolean>
-  handleNativeChatCancelAsk: () => Promise<boolean>
-  handleNativeChatRespondPermission: (text: string) => Promise<boolean>
-  handleNativeChatStop: () => void
-  nativeChatFilePaths: string[]
-  loadNativeChatFiles: (query: string) => void
-  handleNativeChatQuestionAnswer: (text: string) => Promise<boolean>
-  handleNativeChatSend: (text: string, images?: string[]) => Promise<boolean>
-  /** Outcome-preserving send: callers that pasted terminal input beforehand
-   *  (image sends) must see 'unknown' to heal a possibly-orphaned paste. Such a
-   *  caller passes its own `deadline` so the paste it already spent and this text
-   *  body share one budget instead of holding the composer for two. */
-  handleNativeChatSendWithOutcome: (
-    text: string,
-    images?: string[],
-    deadline?: number
-  ) => Promise<MobileNativeChatSendOutcome>
-  /** Launch-context text still parked on the agent's TUI input line, or null.
-   *  Image sends read it to size their leading clear (one Ctrl+U per line). */
-  readSeededLaunchDraft: () => string | null
-  /** Model/session-option pickers for the composer, or null when the active
-   *  agent has no session-option catalog. */
-  nativeChatSessionOptions: MobileNativeChatSessionOptionPickersProps | null
-}
-
 /** Owns mobile native-chat state and teardown outside the already dense session
  *  route. The route remains responsible only for choosing and rendering the view. */
 export function useMobileNativeChatController(args: {

src/main/runtime/rpc/remote-runtime-server-heartbeat-missed-probe-tolerance.test.ts

@@ -0,0 +1,139 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { WebSocket } from 'ws'
+import { RemoteRuntimeServerHeartbeat } from './remote-runtime-server-heartbeat'
+
+afterEach(() => {
+  vi.useRealTimers()
+})
+
+const INTERVAL_MS = 100
+
+function makeSocket(): WebSocket {
+  return { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
+}
+
+describe('RemoteRuntimeServerHeartbeat missed-probe tolerance', () => {
+  it('keeps a client whose pong is merely delayed past one probe window', async () => {
+    vi.useFakeTimers()
+    let now = 1_000
+    const socket = makeSocket()
+    const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now)
+    heartbeat.noteAlive(socket)
+    heartbeat.start(() => [socket]) // probe #1
+    heartbeat.noteAlive(socket) // pongs probe #1
+
+    // A transient blackhole: probe #2 goes out and its pong is stuck in the network.
+    now += INTERVAL_MS
+    await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+    expect(socket.ping).toHaveBeenCalledTimes(2)
+    expect(socket.terminate).not.toHaveBeenCalled()
+
+    // The path recovers and the delayed pong lands right after the next sweep would have run.
+    now += INTERVAL_MS
+    await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+    heartbeat.noteAlive(socket)
+
+    // One unanswered probe is not evidence the peer is gone: it must still be connected,
+    // and it must have been re-probed rather than reaped.
+    expect(socket.terminate).not.toHaveBeenCalled()
+    expect(socket.ping).toHaveBeenCalledTimes(3)
+    heartbeat.stop()
+  })
+
+  it('reaps only after consecutive probes go unanswered, counted not timed', async () => {
+    vi.useFakeTimers()
+    let now = 1_000
+    const socket = makeSocket()
+    const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now)
+    heartbeat.noteAlive(socket)
+    heartbeat.start(() => [socket])
+    heartbeat.noteAlive(socket)
+
+    const missesBeforeReap: number[] = []
+    for (let probe = 0; probe < 6; probe += 1) {
+      now += INTERVAL_MS
+      await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+      missesBeforeReap.push((socket.terminate as unknown as ReturnType<typeof vi.fn>).mock.calls.length)
+    }
+
+    // Silence must be tolerated for more than a single probe, and must eventually end the socket.
+    expect(missesBeforeReap[0]).toBe(0)
+    expect(missesBeforeReap[1]).toBe(0)
+    expect(missesBeforeReap.at(-1)).toBeGreaterThan(0)
+    heartbeat.stop()
+  })
+
+  it('lets any proof of life reset accumulated misses', async () => {
+    vi.useFakeTimers()
+    let now = 1_000
+    const socket = makeSocket()
+    const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now)
+    heartbeat.noteAlive(socket)
+    heartbeat.start(() => [socket])
+    heartbeat.noteAlive(socket)
+
+    // Two silent probes, then a single inbound frame, repeated well past any fixed budget.
+    for (let cycle = 0; cycle < 5; cycle += 1) {
+      now += INTERVAL_MS
+      await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+      now += INTERVAL_MS
+      await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+      heartbeat.noteAlive(socket)
+    }
+
+    expect(socket.terminate).not.toHaveBeenCalled()
+    heartbeat.stop()
+  })
+
+  it('clears accumulated misses when the server event loop resumes from a pause', async () => {
+    vi.useFakeTimers()
+    let now = 1_000
+    const socket = makeSocket()
+    const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now)
+    heartbeat.noteAlive(socket)
+    heartbeat.start(() => [socket])
+    heartbeat.noteAlive(socket)
+
+    const sweep = async (): Promise<void> => {
+      now += INTERVAL_MS
+      await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+    }
+    const reaped = (candidate: WebSocket): boolean =>
+      (candidate.terminate as unknown as ReturnType<typeof vi.fn>).mock.calls.length > 0
+
+    // Discover the budget rather than hardcoding it: the first sweep consumes the pong above, so
+    // every sweep after it is an unanswered probe.
+    await sweep()
+    let toleratedMisses = 0
+    // Bounded so a heartbeat that never reaps fails here instead of hanging out to the test timeout.
+    while (!reaped(socket) && toleratedMisses < 20) {
+      await sweep()
+      toleratedMisses += 1
+    }
+    expect(reaped(socket)).toBe(true)
+    expect(toleratedMisses).toBeGreaterThan(1)
+    heartbeat.stop()
+
+    // Replay on a fresh socket: bank misses to one short of the limit, suspend, then bank the same
+    // number again. The pause must write the earlier misses off, not top them up to the limit.
+    const resumed = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now)
+    const resumedSocket = makeSocket()
+    resumed.noteAlive(resumedSocket)
+    resumed.start(() => [resumedSocket])
+    resumed.noteAlive(resumedSocket)
+    await sweep()
+    for (let miss = 0; miss < toleratedMisses - 1; miss += 1) {
+      await sweep()
+    }
+    expect(reaped(resumedSocket)).toBe(false)
+
+    now += 3_600_000
+    await vi.advanceTimersByTimeAsync(INTERVAL_MS)
+    for (let miss = 0; miss < toleratedMisses - 1; miss += 1) {
+      await sweep()
+    }
+
+    expect(reaped(resumedSocket)).toBe(false)
+    resumed.stop()
+  })
+})

src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts

@@ -7,12 +7,13 @@ afterEach(() => {
 })
 
 describe('RemoteRuntimeServerHeartbeat', () => {
-  it('still reaps a client that misses a probe while another remains alive', async () => {
+  it('still reaps a persistently silent client while another remains alive', async () => {
     vi.useFakeTimers()
     let now = 1_000
     const responsiveSocket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
     const deadSocket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
-    const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now)
+    // Limit of 2 keeps the sweep count readable; production uses the module default.
+    const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now, 128, 2)
     heartbeat.noteAlive(responsiveSocket)
     heartbeat.noteAlive(deadSocket)
     // start() probes immediately: both are pinged now (probe #1) and cleared to await a pong.
@@ -20,12 +21,20 @@ describe('RemoteRuntimeServerHeartbeat', () => {
     // Only the responsive socket pongs the immediate probe.
     heartbeat.noteAlive(responsiveSocket)
 
+    // Miss #1: not yet evidence, so the silent socket is re-probed rather than reaped.
+    now += 100
+    await vi.advanceTimersByTimeAsync(100)
+    heartbeat.noteAlive(responsiveSocket)
+    expect(deadSocket.ping).toHaveBeenCalledTimes(2)
+    expect(deadSocket.terminate).not.toHaveBeenCalled()
+
+    // Miss #2 reaches the limit: consecutive silence is evidence.
     now += 100
     await vi.advanceTimersByTimeAsync(100)
 
-    expect(responsiveSocket.ping).toHaveBeenCalledTimes(2)
+    expect(responsiveSocket.ping).toHaveBeenCalledTimes(3)
     expect(responsiveSocket.terminate).not.toHaveBeenCalled()
-    expect(deadSocket.ping).toHaveBeenCalledTimes(1)
+    expect(deadSocket.ping).toHaveBeenCalledTimes(2)
     expect(deadSocket.terminate).toHaveBeenCalledTimes(1)
     heartbeat.stop()
   })
@@ -34,7 +43,8 @@ describe('RemoteRuntimeServerHeartbeat', () => {
     vi.useFakeTimers()
     let now = 1_000
     const socket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
-    const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now)
+    // Limit of 1 isolates the resume grant: without it the very next sweep would reap.
+    const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now, 128, 1)
     heartbeat.noteAlive(socket)
     // start() probes immediately (ping #1); the socket pongs it.
     heartbeat.start(() => [socket])

src/main/runtime/rpc/remote-runtime-server-heartbeat.ts

@@ -1,16 +1,27 @@
 import type { WebSocket } from 'ws'
 
+// Why: one unanswered probe is UNKNOWN, not death — a cellular/Tailscale blackhole or a stalled TCP
+// retransmit routinely swallows a single pong from a peer that is still there (STA-3320). Only a run of
+// consecutive unanswered probes is evidence. Three matches the web client's own 45s liveness budget
+// (25s idle + 20s probe grace), so both ends of a paired session give up on roughly the same evidence.
+const MISSED_PROBE_LIMIT = 3
+
 export class RemoteRuntimeServerHeartbeat {
   private timer: ReturnType<typeof setInterval> | null = null
   private lastTickAt: number | null = null
   private readonly alive = new WeakSet<WebSocket>()
+  // Why: consecutive probes a socket has left unanswered. Absent = zero, so live sockets cost nothing.
+  private readonly missedProbes = new WeakMap<WebSocket, number>()
 
   constructor(
     private readonly intervalMs: number,
     private readonly now: () => number = Date.now,
-    private readonly warningClientCount = 128
+    private readonly warningClientCount = 128,
+    private readonly missedProbeLimit = MISSED_PROBE_LIMIT
   ) {}
 
+  // Why: hot path — every inbound frame calls this. The sweep clears the miss count when it observes
+  // the flag, so proof of life stays a single WeakSet write.
   noteAlive(socket: WebSocket): void {
     this.alive.add(socket)
   }
@@ -47,16 +58,27 @@ export class RemoteRuntimeServerHeartbeat {
     for (const socket of clients) {
       clientCount += 1
       if (resumedFromPause) {
-        // Why: a delayed server tick cannot infer that clients missed a probe they had no chance to answer.
+        // Why: a delayed server tick cannot infer that clients missed a probe they had no chance to
+        // answer. Seeding alive also clears any misses banked before the pause, so a suspend can never
+        // top up a budget the client was never given a chance to defend.
         this.alive.add(socket)
       }
-      if (!this.alive.has(socket)) {
-        socket.terminate()
-        reaped += 1
-        continue
+      if (this.alive.has(socket)) {
+        this.alive.delete(socket)
+        this.missedProbes.delete(socket)
+      } else {
+        const missed = (this.missedProbes.get(socket) ?? 0) + 1
+        if (missed >= this.missedProbeLimit) {
+          this.missedProbes.delete(socket)
+          socket.terminate()
+          reaped += 1
+          continue
+        }
+        this.missedProbes.set(socket, missed)
       }
-      this.alive.delete(socket)
       try {
+        // Why: re-probe every sweep, including missed ones, so a path that just recovered can prove
+        // itself on the next tick instead of waiting out the rest of the budget.
         socket.ping()
       } catch {
         // Why: a mid-teardown socket is finalized by its close/error listener.

src/main/runtime/rpc/ws-transport-accept-order.test.ts

@@ -83,7 +83,7 @@ describe('WebSocketTransport accepted socket ordering', () => {
     ).toEqual([0, 0, 0, 0])
   })
 
-  it('reaps an unresponsive accepted socket by the first interval deadline', async () => {
+  it('reaps an unresponsive accepted socket once consecutive probes go unanswered', async () => {
     vi.useFakeTimers()
     const events: string[] = []
     let socket: WebSocket
@@ -99,19 +99,22 @@ describe('WebSocketTransport accepted socket ordering', () => {
     socket.once('open', () => events.push('open'))
     socket.emit('open')
     lifecycle.handleConnection(socket)
-    await vi.advanceTimersByTimeAsync(99)
+    // A single silent interval is not evidence: the socket is re-probed, not reaped.
+    await vi.advanceTimersByTimeAsync(100)
+    expect(socket.terminate).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(199)
     expect(socket.terminate).not.toHaveBeenCalled()
 
     await vi.advanceTimersByTimeAsync(1)
 
     expect(socket.terminate).toHaveBeenCalledTimes(1)
-    expect(events).toEqual(['open', 'ping', 'close'])
+    expect(events).toEqual(['open', 'ping', 'ping', 'ping', 'close'])
     expect(lifecycle.heartbeatConnections.size).toBe(0)
     expect(lifecycle.heartbeat.timer).toBeNull()
     expect(vi.getTimerCount()).toBe(0)
   })
 
-  it('keeps later sockets on the shared cadence and reaps them within two intervals', async () => {
+  it('keeps later sockets on the shared cadence and reaps them on the shared budget', async () => {
     vi.useFakeTimers()
     vi.setSystemTime(0)
     const firstPingTimes: number[] = []
@@ -150,13 +153,15 @@ describe('WebSocketTransport accepted socket ordering', () => {
     expect(laterPingTimes).toEqual([100])
     expect(laterSocket.terminate).not.toHaveBeenCalled()
 
-    await vi.advanceTimersByTimeAsync(99)
+    // Re-probed on every sweep while its misses bank, so a recovered path can answer immediately.
+    await vi.advanceTimersByTimeAsync(299)
+    expect(laterPingTimes).toEqual([100, 200, 300])
     expect(laterSocket.terminate).not.toHaveBeenCalled()
 
     await vi.advanceTimersByTimeAsync(1)
     expect(laterSocket.terminate).toHaveBeenCalledTimes(1)
-    expect(laterReapTimes).toEqual([200])
-    expect(firstPingTimes).toEqual([0, 100, 200])
+    expect(laterReapTimes).toEqual([400])
+    expect(firstPingTimes).toEqual([0, 100, 200, 300, 400])
     expect(
       ['pong', 'message', 'close', 'error'].map((event) => laterSocket.listenerCount(event))
     ).toEqual([0, 0, 0, 0])

src/main/runtime/rpc/ws-transport-transient-packet-loss.test.ts

@@ -0,0 +1,83 @@
+// Real ws server + real ws client over loopback: a live, well-behaved peer whose pong is swallowed by
+// a transient blackhole must not be terminated (STA-3320).
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import WebSocket from 'ws'
+import { WebSocketTransport } from './ws-transport'
+
+const PROBE_INTERVAL_MS = 150
+
+// Why: the heartbeat calls now() exactly once per sweep, so a per-call step makes elapsed exactly one
+// interval every sweep — real scheduler jitter can never be mistaken for a suspended loop, and no
+// assertion in this file depends on wall time.
+function createStepClock(stepMs: number): () => number {
+  let calls = 0
+  return () => 1_000_000 + calls++ * stepMs
+}
+
+describe('WebSocketTransport under transient packet loss', () => {
+  const transports: WebSocketTransport[] = []
+  const sockets: WebSocket[] = []
+
+  afterEach(async () => {
+    for (const socket of sockets) {
+      socket.removeAllListeners()
+      socket.terminate()
+    }
+    sockets.length = 0
+    await Promise.all(transports.map((t) => t.stop().catch(() => {})))
+    transports.length = 0
+  })
+
+  async function startTransport(): Promise<WebSocketTransport> {
+    const transport = new WebSocketTransport({
+      host: '127.0.0.1',
+      port: 0,
+      heartbeatIntervalMs: PROBE_INTERVAL_MS,
+      heartbeatNow: createStepClock(PROBE_INTERVAL_MS),
+      // Why: this test is about liveness, not the auth handshake; keep the pre-auth reaper out of it.
+      preAuthTimeoutMs: 600_000
+    })
+    transports.push(transport)
+    await transport.start()
+    return transport
+  }
+
+  it('keeps a live client whose pong is swallowed for a single probe', async () => {
+    const transport = await startTransport()
+    // autoPong:false hands us the protocol-level pong so we can drop exactly one, the way a
+    // cellular/Tailscale blackhole drops exactly the frames in flight during the stall.
+    const client = new WebSocket(`ws://127.0.0.1:${transport.resolvedPort}`, { autoPong: false })
+    sockets.push(client)
+
+    let probesReceived = 0
+    const pongedProbes: number[] = []
+    const swallowedProbe = 2
+    let closed = false
+    client.on('ping', () => {
+      probesReceived += 1
+      if (probesReceived === swallowedProbe) {
+        return
+      }
+      pongedProbes.push(probesReceived)
+      client.pong()
+    })
+    client.on('close', () => {
+      closed = true
+    })
+    await new Promise<void>((resolve, reject) => {
+      client.once('open', resolve)
+      client.once('error', reject)
+    })
+
+    // Bounded by counted probe events, never by elapsed time.
+    await vi.waitFor(() => expect(closed || probesReceived >= swallowedProbe + 2).toBe(true), {
+      timeout: 15_000,
+      interval: 10
+    })
+
+    expect({ closed, probesReceived }).toEqual({ closed: false, probesReceived: 4 })
+    // The peer was demonstrably alive across the blackhole: it answered the probes on both sides of it.
+    expect(pongedProbes).toEqual([1, 3, 4])
+    expect(client.readyState).toBe(WebSocket.OPEN)
+  })
+})

src/main/runtime/rpc/ws-transport.ts

@@ -21,7 +21,7 @@ type WebSocketMessageHandler = {
   ): void
 }['bivarianceHack']
 
-// Why: mobile clients background-suspend sockets with no TCP FIN, leaving half-opens that otherwise only the OS keepalive (~2h) reaps; a 15s ping/pong sweep bounds that to ~30s (clients auto-pong per RFC 6455).
+// Why: mobile clients background-suspend sockets with no TCP FIN, leaving half-opens that otherwise only the OS keepalive (~2h) reaps; a 15s ping/pong sweep bounds that to ~60s (clients auto-pong per RFC 6455), since a reap needs consecutive unanswered probes rather than one (STA-3320).
 const HEARTBEAT_INTERVAL_MS = 15_000
 
 export type WebSocketTransportOptions = {

src/renderer/src/web/web-runtime-client.test.ts

@@ -210,6 +210,8 @@ describe('WebRuntimeClient', () => {
     }
     timerWindow.setTimeout = setTimeout
     timerWindow.clearTimeout = clearTimeout
+    // Pin the one-sided reconnect jitter to zero so the redial lands on its exact backoff step.
+    const random = vi.spyOn(Math, 'random').mockReturnValue(0)
     const client = new WebRuntimeClient({
       v: 2,
       endpoint: 'ws://127.0.0.1:6768',
@@ -235,6 +237,7 @@ describe('WebRuntimeClient', () => {
       expect(replacementSocket.send).toHaveBeenCalledTimes(1)
     } finally {
       client.close()
+      random.mockRestore()
       vi.useRealTimers()
     }
   })

src/renderer/src/web/web-runtime-client.ts

@@ -16,6 +16,7 @@ import {
 } from './web-e2ee'
 import { SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY } from '../../../shared/protocol-version'
 import { createWebRuntimeUnauthorizedError } from './web-runtime-client-error'
+import { withReconnectJitter } from '../../../shared/reconnect-jitter'
 
 type WebRuntimeConnectionState =
   | 'disconnected'
@@ -641,8 +642,9 @@ export class WebRuntimeClient {
     if (this.reconnectTimer || this.intentionallyClosed) {
       return
     }
-    const delay =
+    const delay = withReconnectJitter(
       RECONNECT_DELAYS_MS[Math.min(this.reconnectAttempt, RECONNECT_DELAYS_MS.length - 1)]
+    )
     this.reconnectAttempt += 1
     this.reconnectTimer = window.setTimeout(() => {
       this.reconnectTimer = null

src/shared/reconnect-jitter.test.ts

@@ -0,0 +1,17 @@
+import { describe, expect, it } from 'vitest'
+import { withReconnectJitter } from './reconnect-jitter'
+
+describe('withReconnectJitter', () => {
+  it('never returns less than the backoff floor', () => {
+    expect(withReconnectJitter(500, () => 0)).toBe(500)
+  })
+
+  it('spreads a fleet that was dropped by one shared blip', () => {
+    const fleet = Array.from({ length: 32 }, (_, index) =>
+      withReconnectJitter(500, () => index / 32)
+    )
+    expect(new Set(fleet).size).toBeGreaterThan(1)
+    expect(Math.min(...fleet)).toBeGreaterThanOrEqual(500)
+    expect(Math.max(...fleet)).toBeLessThanOrEqual(600)
+  })
+})

src/shared/reconnect-jitter.ts

@@ -0,0 +1,7 @@
+// Why: a blip on a shared path (Tailscale, cellular, host restart) drops every socket on that path in
+// the same instant. A jitterless backoff ladder then re-dials all of them at the same millisecond,
+// which is exactly the thundering herd the host is least able to absorb while it is still recovering.
+// One-sided so a delay is never shortened below its backoff floor.
+export function withReconnectJitter(delayMs: number, random: () => number = Math.random): number {
+  return delayMs + Math.floor(delayMs * 0.2 * random())
+}

src/shared/remote-runtime-shared-control-state.ts

@@ -10,6 +10,7 @@ import type {
   SharedControlReadyWaiter
 } from './remote-runtime-shared-control-types'
 import { getSubscriptionId, isEndResult } from './remote-runtime-shared-control-protocol'
+import { withReconnectJitter } from './reconnect-jitter'
 import { tagRuntimeSubscriptionReplayResponse } from './runtime-subscription-replay'
 
 export function buildSharedControlDiagnostics(args: {
@@ -208,9 +209,3 @@ export function scheduleSharedControlReconnect(args: {
   return { timer, reconnectAttempt: args.reconnectAttempt + 1 }
 }
 
-function withReconnectJitter(delayMs: number): number {
-  // Why: when a remote host restarts, all passive subscriptions reconnect
-  // together. A small one-sided jitter avoids synchronized retry spikes.
-  const jitterMs = Math.floor(delayMs * 0.2 * Math.random())
-  return delayMs + jitterMs
-}