fix(push): close review findings in the desktop client (#8129)

- A gateway registration the registry cannot persist is enqueued for delete instead of leaking a live token. - Unregister outbox re-reads pending per pass, honours enqueues during a drain, and retries with backoff instead of waiting for the next launch. - Dispatcher batches registrations by 20 rather than starving the rest. - 401 compare-and-clear; a 401 after re-auth is unreachable; refused handshakes and 429s are cached briefly instead of re-handshaking per event. - Service is stopped on quit.
2026-09-29 08:03:20 +00:00 · 2026-09-06 15:19:20 -04:00
parent d12790c9b6
commit 74ca5f6953
11 changed files with 739 additions and 197 deletions
@@ -14,7 +14,14 @@ const REGISTER_INPUT = {
  filter: { sources: ['agent-task-complete'] as const, agentStates: ['finished'] as const }
 }

-function createService(options: { registerFails?: boolean; deleteFails?: boolean } = {}): {
+function createService(
+  options: {
+    registerFails?: boolean
+    deleteFails?: boolean
+    /** Runs before each delete resolves, so a suite can queue work mid-flush. */
+    onDelete?: (registrationId: string) => void
+  } = {}
+): {
  service: DesktopPushService
  registry: DeviceRegistry
  outbox: PushUnregisterOutbox
@@ -22,6 +29,7 @@ function createService(options: { registerFails?: boolean; deleteFails?: boolean
  deletes: string[]
  send: ReturnType<typeof vi.fn>
  dispatch: (event: MobileNotificationEvent) => void
+  retries: { run: () => void; delayMs: number }[]
 } {
  const userDataPath = mkdtempSync(join(tmpdir(), 'orca-push-service-'))
  const registry = new DeviceRegistry(userDataPath)
@@ -54,17 +62,22 @@ function createService(options: { registerFails?: boolean; deleteFails?: boolean
    ),
    deleteDevice: vi.fn(async (registrationId: string) => {
      deletes.push(registrationId)
+      options.onDelete?.(registrationId)
      return options.deleteFails
        ? { deleted: false, retryable: true }
        : { deleted: true, retryable: false }
    }),
    send: vi.fn(async () => ({ ok: true, results: [] }) as const)
  }
+  const retries: { run: () => void; delayMs: number }[] = []
  const service = DesktopPushService.create({
    runtime: runtime as never,
    runtimeRpc: runtimeRpc as never,
    gatewayUrl: 'https://push.onorca.dev',
-    client: client as never
+    client: client as never,
+    scheduleRetry: (run, delayMs) => {
+      retries.push({ run, delayMs })
+    }
  })!

  service.start()
@@ -75,7 +88,8 @@ function createService(options: { registerFails?: boolean; deleteFails?: boolean
    deviceId: device.deviceId,
    deletes,
    send: client.send,
-    dispatch: (event) => listener?.(event)
+    dispatch: (event) => listener?.(event),
+    retries
  }
 }

@@ -150,6 +164,83 @@ describe('DesktopPushService', () => {
    expect(harness.outbox.pending()).toEqual([])
  })

+  it('unregisters at the gateway when the device stopped being a phone mid-register', async () => {
+    const harness = createService()
+    vi.spyOn(harness.registry, 'setPushRegistration').mockReturnValue(false)
+
+    expect(
+      await harness.service.register({ deviceId: harness.deviceId, ...REGISTER_INPUT })
+    ).toEqual({ registered: false, reason: 'not_mobile' })
+    // register() kicks the flush off without awaiting it; join the same run.
+    await harness.service.flushUnregisterOutbox()
+    expect(harness.deletes).toEqual(['reg-1'])
+    expect(harness.outbox.pending()).toEqual([])
+  })
+
+  it('unregisters at the gateway when the registration cannot be written', async () => {
+    const harness = createService({ deleteFails: true })
+    vi.spyOn(harness.registry, 'setPushRegistration').mockImplementation(() => {
+      throw new Error('disk full')
+    })
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+    expect(
+      await harness.service.register({ deviceId: harness.deviceId, ...REGISTER_INPUT })
+    ).toEqual({ registered: false, reason: 'registration_storage_failed' })
+    // The gateway kept the token, so the delete stays queued until it lands.
+    expect(harness.outbox.pending()).toEqual([
+      expect.objectContaining({ registrationId: 'reg-1', deviceId: harness.deviceId })
+    ])
+    warn.mockRestore()
+  })
+
+  it('drains a delete queued while a flush is already running', async () => {
+    let queued = false
+    const harness = createService({
+      onDelete: () => {
+        if (queued) {
+          return
+        }
+        queued = true
+        harness.outbox.enqueue({ registrationId: 'reg-late', deviceId: 'device-late' })
+        // Mirrors unregister(): the trigger arrives while the flush is mid-await.
+        void harness.service.flushUnregisterOutbox()
+      }
+    })
+    harness.outbox.enqueue({ registrationId: 'reg-first', deviceId: 'device-first' })
+
+    await harness.service.flushUnregisterOutbox()
+
+    expect(harness.deletes).toEqual(['reg-first', 'reg-late'])
+    expect(harness.outbox.pending()).toEqual([])
+  })
+
+  it('retries a failed drain on a capped backoff instead of waiting for a relaunch', async () => {
+    const harness = createService({ deleteFails: true })
+    harness.outbox.enqueue({ registrationId: 'reg-stuck', deviceId: 'device-1' })
+
+    await harness.service.flushUnregisterOutbox()
+    expect(harness.retries.map((entry) => entry.delayMs)).toEqual([30_000])
+
+    harness.retries[0]?.run()
+    await new Promise((resolve) => setImmediate(resolve))
+    expect(harness.deletes).toEqual(['reg-stuck', 'reg-stuck'])
+    expect(harness.retries.map((entry) => entry.delayMs)).toEqual([30_000, 60_000])
+    expect(harness.outbox.pending()).toHaveLength(1)
+  })
+
+  it('stops re-arming the retry once the service is stopped', async () => {
+    const harness = createService({ deleteFails: true })
+    harness.outbox.enqueue({ registrationId: 'reg-stuck', deviceId: 'device-1' })
+    await harness.service.flushUnregisterOutbox()
+
+    harness.service.stop()
+    harness.retries[0]?.run()
+    await new Promise((resolve) => setImmediate(resolve))
+
+    expect(harness.retries).toHaveLength(1)
+  })
+
  it('pushes a dispatched notification through the subscribed dispatcher', async () => {
    const harness = createService()
    await harness.service.register({ deviceId: harness.deviceId, ...REGISTER_INPUT })
@@ -13,12 +13,19 @@ import { PushDispatcher } from './push-dispatcher'
 import { PushGatewayClient } from './push-gateway-client'
 import type { PushUnregisterOutbox } from './push-unregister-outbox'

+const OUTBOX_RETRY_BASE_MS = 30_000
+const OUTBOX_RETRY_MAX_MS = 10 * 60_000
+
+type RegisterStorageFailure = 'not_mobile' | 'registration_storage_failed'
+
 type DesktopPushServiceOptions = {
  runtime: OrcaRuntimeService
  runtimeRpc: OrcaRuntimeRpcServer
  gatewayUrl: string
  /** Test seam: lets a suite drive the service without a live gateway. */
  client?: PushGatewayClient
+  /** Test seam: lets a suite drive the outbox backoff without real timers. */
+  scheduleRetry?: (run: () => void, delayMs: number) => void
 }

 export class DesktopPushService {
@@ -28,8 +35,13 @@ export class DesktopPushService {
  private readonly outbox: PushUnregisterOutbox
  private readonly client: PushGatewayClient
  private readonly dispatcher: PushDispatcher
+  private readonly scheduleRetry: (run: () => void, delayMs: number) => void
  private unsubscribe: (() => void) | null = null
-  private flushing = false
+  private flushLoop: Promise<void> | null = null
+  private flushRequested = false
+  private retryArmed = false
+  private retryDelayMs = OUTBOX_RETRY_BASE_MS
+  private stopped = false

  private constructor(
    options: DesktopPushServiceOptions,
@@ -42,6 +54,12 @@ export class DesktopPushService {
    this.client = client
    this.outbox = options.runtimeRpc.getPushUnregisterOutbox()
    this.dispatcher = new PushDispatcher({ client, registry })
+    this.scheduleRetry =
+      options.scheduleRetry ??
+      ((run, delayMs) => {
+        // Why: a queued gateway delete must never hold the app open at quit.
+        setTimeout(run, delayMs).unref?.()
+      })
  }

  /** Returns null when the mobile runtime never came up, so there is nothing to push for. */
@@ -57,6 +75,7 @@ export class DesktopPushService {
  }

  start(): void {
+    this.stopped = false
    this.runtime.setMobilePushRegistrar(this)
    this.unsubscribe = this.runtime.onNotificationDispatched((event) => {
      this.dispatcher.enqueue(event)
@@ -70,6 +89,7 @@ export class DesktopPushService {
  }

  stop(): void {
+    this.stopped = true
    this.unsubscribe?.()
    this.unsubscribe = null
    this.runtimeRpc.setOnPushUnregisterQueued(null)
@@ -87,14 +107,16 @@ export class DesktopPushService {
        reason: result.reason === 'unreachable' ? 'gateway_unreachable' : 'gateway_rejected'
      }
    }
-    this.registry.setPushRegistration(input.deviceId, {
-      registrationId: result.registrationId,
-      platform: input.platform,
-      filter: input.filter,
-      registeredAt: Date.now()
-    })
+    const failure = this.storeRegistration(input, result.registrationId)
+    if (failure) {
+      // Why: the gateway now holds a token this host will never push to. Queue its
+      // delete instead of leaking it until the phone happens to register again.
+      this.outbox.enqueue({ registrationId: result.registrationId, deviceId: input.deviceId })
+    }
    void this.flushUnregisterOutbox()
-    return { registered: true, registrationId: result.registrationId }
+    return failure
+      ? { registered: false, reason: failure }
+      : { registered: true, registrationId: result.registrationId }
  }

  async unregister(deviceId: string): Promise<{ unregistered: boolean }> {
@@ -110,23 +132,85 @@ export class DesktopPushService {
    return { unregistered: true }
  }

+  /** Joining an in-flight drain still waits for the item this call queued. */
  async flushUnregisterOutbox(): Promise<void> {
-    if (this.flushing) {
-      return
+    this.flushRequested = true
+    this.flushLoop ??= this.runFlushLoop().finally(() => {
+      this.flushLoop = null
+    })
+    await this.flushLoop
+  }
+
+  private async runFlushLoop(): Promise<void> {
+    while (this.flushRequested && !this.stopped) {
+      // Cleared before the pass, so a delete queued mid-drain earns another one.
+      this.flushRequested = false
+      if (await this.drainPending()) {
+        this.scheduleFlushRetry()
+      } else {
+        this.retryDelayMs = OUTBOX_RETRY_BASE_MS
+      }
    }
-    this.flushing = true
+  }
+
+  /** Returns the refusal reason when a gateway-accepted registration cannot be stored. */
+  private storeRegistration(
+    input: MobilePushRegisterInput,
+    registrationId: string
+  ): RegisterStorageFailure | null {
    try {
-      // Safe to iterate while removing: the outbox swaps in a new array per write.
-      for (const item of this.outbox.pending()) {
+      const stored = this.registry.setPushRegistration(input.deviceId, {
+        registrationId,
+        platform: input.platform,
+        filter: input.filter,
+        registeredAt: Date.now()
+      })
+      // False means the device was removed or left mobile scope while the gateway
+      // call was in flight.
+      return stored ? null : 'not_mobile'
+    } catch (error) {
+      console.warn('[push] Failed to persist a push registration:', error)
+      return 'registration_storage_failed'
+    }
+  }
+
+  /** Returns true when the pass left behind an item the gateway may still accept. */
+  private async drainPending(): Promise<boolean> {
+    const attempted = new Set<string>()
+    let retryable = false
+    for (;;) {
+      // Re-read per item: a snapshot taken at loop entry misses anything queued
+      // while an await was in flight, and the outbox swaps arrays on every write.
+      const item = this.outbox.pending().find((candidate) => !attempted.has(candidate.reqId))
+      if (!item) {
+        return retryable
+      }
+      attempted.add(item.reqId)
+      try {
        const result = await this.client.deleteDevice(item.registrationId)
        if (result.deleted || !result.retryable) {
          this.outbox.remove(item.reqId)
+        } else {
+          retryable = true
        }
+      } catch (error) {
+        // One bad delete must not strand the rest of the queue.
+        console.warn('[push] Failed to drain the push unregister outbox:', error)
+        retryable = true
      }
-    } catch (error) {
-      console.warn('[push] Failed to drain the push unregister outbox:', error)
-    } finally {
-      this.flushing = false
    }
  }
+
+  private scheduleFlushRetry(): void {
+    if (this.retryArmed || this.stopped) {
+      return
+    }
+    this.retryArmed = true
+    const delayMs = this.retryDelayMs
+    this.retryDelayMs = Math.min(delayMs * 2, OUTBOX_RETRY_MAX_MS)
+    this.scheduleRetry(() => {
+      this.retryArmed = false
+      void this.flushUnregisterOutbox()
+    }, delayMs)
+  }
 }
@@ -132,6 +132,43 @@ describe('PushDispatcher', () => {
    })
  })

+  it('fans out past the per-request cap instead of starving the extra devices', async () => {
+    const devices = Array.from({ length: 25 }, (_, index) => ({
+      deviceId: `device-${index}`,
+      pushRegistration: registration({ registrationId: `reg-${index}` })
+    }))
+    const harness = createHarness({ devices })
+
+    harness.dispatcher.enqueue(notification())
+    await flush()
+
+    expect(harness.sends).toHaveLength(2)
+    expect(harness.sends[0]?.registrationIds).toHaveLength(20)
+    expect(harness.sends[1]?.registrationIds).toEqual([
+      'reg-20',
+      'reg-21',
+      'reg-22',
+      'reg-23',
+      'reg-24'
+    ])
+  })
+
+  it('drops a dead registration reported by a later chunk', async () => {
+    const devices = Array.from({ length: 25 }, (_, index) => ({
+      deviceId: `device-${index}`,
+      pushRegistration: registration({ registrationId: `reg-${index}` })
+    }))
+    const harness = createHarness({
+      devices,
+      results: [{ registrationId: 'reg-24', status: 'dead' }]
+    })
+
+    harness.dispatcher.enqueue(notification())
+    await flush()
+
+    expect(harness.cleared).toEqual(['device-24'])
+  })
+
  it('never pushes a dismissal', async () => {
    const harness = createHarness({
      devices: [{ deviceId: 'a', pushRegistration: registration() }]
@@ -11,8 +11,8 @@ import type { MobileNotificationEvent } from '../runtime-mobile-notification-con
 import type { PushGatewayClient, PushSendNotification } from './push-gateway-client'

 const PUSH_RETRY_DELAY_MS = 2_000
-// The gateway rejects a whole request above this; a host with more paired phones
-// than this still pushes to the first 20 rather than to none.
+// The gateway rejects a whole request above this, so a host with more paired
+// phones fans out across several sends rather than starving the extras.
 const MAX_REGISTRATIONS_PER_SEND = 20
 const PUSH_TITLE_MAX_LENGTH = 80
 const PUSH_BODY_MAX_LENGTH = 180
@@ -76,8 +76,13 @@ export class PushDispatcher {
  enqueue(event: MobileNotificationEvent): void {
    try {
      const plan = this.planSend(event)
-      if (plan) {
-        void this.deliver(plan.targets, plan.notification, 0)
+      if (!plan) {
+        return
+      }
+      // Each chunk is its own request, so its retry and dead-drop are independent.
+      for (let start = 0; start < plan.targets.length; start += MAX_REGISTRATIONS_PER_SEND) {
+        const chunk = plan.targets.slice(start, start + MAX_REGISTRATIONS_PER_SEND)
+        void this.deliver(chunk, plan.notification, 0)
      }
    } catch (error) {
      console.warn('[push] Failed to prepare a push notification:', error)
@@ -99,19 +104,16 @@ export class PushDispatcher {
    if (agentState === undefined) {
      return null
    }
-    const targets = this.registry
-      .listDevices()
-      .flatMap((device) => {
-        const registration = device.pushRegistration
-        if (!registration || !registration.filter.sources.includes(source)) {
-          return []
-        }
-        if (agentState !== null && !registration.filter.agentStates.includes(agentState)) {
-          return []
-        }
-        return [{ deviceId: device.deviceId, registrationId: registration.registrationId }]
-      })
-      .slice(0, MAX_REGISTRATIONS_PER_SEND)
+    const targets = this.registry.listDevices().flatMap((device) => {
+      const registration = device.pushRegistration
+      if (!registration || !registration.filter.sources.includes(source)) {
+        return []
+      }
+      if (agentState !== null && !registration.filter.agentStates.includes(agentState)) {
+        return []
+      }
+      return [{ deviceId: device.deviceId, registrationId: registration.registrationId }]
+    })
    if (targets.length === 0) {
      return null
    }
@@ -19,7 +19,9 @@ function jsonResponse(status: number, body: unknown): Response {
  })
 }

-function createFakeGateway(options: { sessionTtlMs?: number; devicesStatus?: number } = {}): {
+function createFakeGateway(
+  options: { sessionTtlMs?: number; devicesStatus?: number; rejectBearer?: boolean } = {}
+): {
  client: PushGatewayClient
  calls: Recorded[]
  expireSession: () => void
@@ -30,6 +32,7 @@ function createFakeGateway(options: { sessionTtlMs?: number; devicesStatus?: num
  const now = { value: NOW }
  const calls: Recorded[] = []
  const liveTokens = new Set<string>()
+  const knownRegistrations = new Set<string>()
  let issued = 0
  let pendingProof: string | null = null

@@ -68,18 +71,27 @@ function createFakeGateway(options: { sessionTtlMs?: number; devicesStatus?: num
      })
    }
    const bearer = headers.get('authorization')?.replace('Bearer ', '') ?? ''
-    if (!liveTokens.has(bearer)) {
+    if (options.rejectBearer || !liveTokens.has(bearer)) {
      return jsonResponse(401, { error: 'session_expired' })
    }
    if (url.endsWith('/v1/devices')) {
-      return options.devicesStatus
-        ? jsonResponse(options.devicesStatus, { error: 'nope' })
-        : jsonResponse(200, { registrationId: 'reg-1' })
+      if (options.devicesStatus) {
+        return jsonResponse(options.devicesStatus, { error: 'nope' })
+      }
+      knownRegistrations.add('reg-1')
+      return jsonResponse(200, { registrationId: 'reg-1' })
    }
    if (url.endsWith('/v1/send')) {
      return jsonResponse(200, { results: [{ registrationId: 'reg-1', status: 'queued' }] })
    }
-    return new Response(null, { status: 204 })
+    // Why explicit: a catch-all 204 would report every delete as accepted and
+    // leave the 404 branch of deleteDevice untested.
+    const deleted = /\/v1\/devices\/([^/]+)$/.exec(url)
+    if (deleted && init?.method === 'DELETE') {
+      const registrationId = decodeURIComponent(deleted[1] ?? '')
+      return new Response(null, { status: knownRegistrations.has(registrationId) ? 204 : 404 })
+    }
+    throw new Error(`unexpected request: ${init?.method ?? 'GET'} ${url}`)
  }) as unknown as typeof globalThis.fetch

  return {
@@ -193,8 +205,37 @@ describe('PushGatewayClient', () => {
    })
  })

+  it('treats a delete the gateway accepted as done', async () => {
+    const gateway = createFakeGateway()
+    await gateway.client.registerDevice(REGISTER_INPUT)
+
+    expect(await gateway.client.deleteDevice('reg-1')).toEqual({ deleted: true, retryable: false })
+    expect(gateway.calls.at(-1)).toMatchObject({ method: 'DELETE' })
+  })
+
  it('treats a delete of an unknown registration as done', async () => {
    const gateway = createFakeGateway()
-    expect(await gateway.client.deleteDevice('reg-1')).toEqual({ deleted: true, retryable: false })
+
+    expect(await gateway.client.deleteDevice('reg-gone')).toEqual({
+      deleted: true,
+      retryable: false
+    })
+  })
+
+  it('reports a 401 that survives the forced re-auth as unreachable', async () => {
+    const gateway = createFakeGateway({ rejectBearer: true })
+
+    expect(await gateway.client.registerDevice(REGISTER_INPUT)).toEqual({
+      ok: false,
+      reason: 'unreachable'
+    })
+    // Exactly one forced re-auth, not a handshake loop.
+    expect(gateway.calls.filter((call) => call.url.endsWith('/v1/host/challenge'))).toHaveLength(2)
+  })
+
+  it('keeps an unreachable-classified 401 retryable for a queued delete', async () => {
+    const gateway = createFakeGateway({ rejectBearer: true })
+
+    expect(await gateway.client.deleteDevice('reg-1')).toEqual({ deleted: false, retryable: true })
  })
 })
@@ -4,7 +4,6 @@
 import { z } from 'zod'
 import { cancelUnreadResponseBody } from '../../lib/unread-response-body'
 import type { E2EEKeypair } from '../e2ee-keypair'
-import { deriveRelayHostId } from '../relay/relay-http-client'
 import type {
  MobilePushAgentState,
  MobilePushApnsEnvironment,
@@ -12,33 +11,16 @@ import type {
  MobilePushPlatform,
  MobilePushSource
 } from '../../../shared/mobile-push-contract'
-import { answerPushHostChallenge, type PushHostChallenge } from './push-host-proof'
+import {
+  PUSH_REQUEST_DEADLINE_MS,
+  readPushGatewayJson,
+  type PushGatewayFailure,
+  type PushGatewayResponse,
+  type PushGatewayResult
+} from './push-gateway-response'
+import { PushGatewaySession } from './push-gateway-session'

-const PUSH_REQUEST_DEADLINE_MS = 15_000
-// Re-auth a little early so a send never spends its one retry on a token that
-// expired between the check and the request.
-const SESSION_RENEWAL_MARGIN_MS = 60_000
-
-const ChallengeResponseSchema = z
-  .object({
-    challengeId: z.string().min(1).max(512),
-    gatewayEphemeralPublicKeyB64: z.string().min(1).max(128),
-    nonceB64: z.string().min(1).max(128),
-    ciphertextB64: z
-      .string()
-      .min(1)
-      .max(8 * 1024),
-    expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER)
-  })
-  .strict()
-
-const SessionResponseSchema = z
-  .object({
-    sessionToken: z.string().min(1).max(1024),
-    expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER),
-    hostFingerprint: z.string().min(1).max(64)
-  })
-  .strict()
+export type { PushGatewayFailure, PushGatewayResult }

 const RegisterResponseSchema = z.object({ registrationId: z.string().min(1).max(512) })

@@ -55,9 +37,6 @@ const SendResponseSchema = z.object({

 export type PushSendResult = z.infer<typeof SendResponseSchema>['results'][number]

-export type PushGatewayFailure = { ok: false; reason: 'unreachable' | 'rejected' }
-export type PushGatewayResult<T> = ({ ok: true } & T) | PushGatewayFailure
-
 export type PushSendNotification = {
  notificationId?: string
  notificationSeq: number
@@ -76,23 +55,24 @@ type PushGatewayClientOptions = {
  now?: () => number
 }

-type CachedSession = { token: string; expiresAt: number }
+type AuthorizedResponse = { ok: true; response: Response; token: string } | PushGatewayFailure

 export class PushGatewayClient {
  private readonly origin: string
-  private readonly keypair: E2EEKeypair
  private readonly fetchImpl: typeof globalThis.fetch
-  private readonly now: () => number
+  private readonly session: PushGatewaySession
  readonly hostFingerprint: string
-  private session: CachedSession | null = null
-  private pendingSession: Promise<CachedSession | null> | null = null

  constructor(options: PushGatewayClientOptions) {
    this.origin = new URL(options.gatewayUrl).origin
-    this.keypair = options.keypair
    this.fetchImpl = options.fetch ?? globalThis.fetch
-    this.now = options.now ?? Date.now
-    this.hostFingerprint = deriveRelayHostId(options.keypair.publicKey)
+    this.session = new PushGatewaySession({
+      origin: this.origin,
+      keypair: options.keypair,
+      fetchImpl: this.fetchImpl,
+      now: options.now ?? Date.now
+    })
+    this.hostFingerprint = this.session.hostFingerprint
  }

  async registerDevice(input: {
@@ -113,7 +93,7 @@ export class PushGatewayClient {
        filter: { sources: [...input.filter.sources], agentStates: [...input.filter.agentStates] }
      }
    })
-    const parsed = await readJson(response, RegisterResponseSchema)
+    const parsed = await readPushGatewayJson(response, RegisterResponseSchema)
    return parsed.ok ? { ok: true, registrationId: parsed.value.registrationId } : parsed
  }

@@ -143,143 +123,53 @@ export class PushGatewayClient {
        notification: input.notification
      }
    })
-    const parsed = await readJson(response, SendResponseSchema)
+    const parsed = await readPushGatewayJson(response, SendResponseSchema)
    return parsed.ok ? { ok: true, results: parsed.value.results } : parsed
  }

  private async authorized(
    path: string,
    init: { method: string; body?: unknown }
-  ): Promise<{ ok: true; response: Response } | PushGatewayFailure> {
-    const first = await this.sendAuthorized(path, init, false)
-    // A 401 means the cached session died server-side; one forced re-auth, then give up.
-    if (first.ok && first.response.status === 401) {
-      await cancelUnreadResponseBody(first.response)
-      this.session = null
-      return await this.sendAuthorized(path, init, true)
+  ): Promise<PushGatewayResponse> {
+    const first = await this.sendAuthorized(path, init, null)
+    if (!first.ok || first.response.status !== 401) {
+      return first
    }
-    return first
+    // A 401 means that one session died server-side; one forced re-auth, then stop.
+    await cancelUnreadResponseBody(first.response)
+    const retried = await this.sendAuthorized(path, init, first.token)
+    if (retried.ok && retried.response.status === 401) {
+      await cancelUnreadResponseBody(retried.response)
+      // A 401 that survives a freshly minted session is the gateway being unusable
+      // right now, not this request being wrong: register should report it as
+      // unreachable, and send should still spend its one retry.
+      return { ok: false, reason: 'unreachable' }
+    }
+    return retried
  }

  private async sendAuthorized(
    path: string,
    init: { method: string; body?: unknown },
-    forceReauth: boolean
-  ): Promise<{ ok: true; response: Response } | PushGatewayFailure> {
-    const session = await this.ensureSession(forceReauth)
-    if (!session) {
-      return { ok: false, reason: 'unreachable' }
+    staleToken: string | null
+  ): Promise<AuthorizedResponse> {
+    const outcome = await this.session.ensure(staleToken)
+    if (!outcome.ok) {
+      return outcome
    }
    try {
      const response = await this.fetchImpl(`${this.origin}${path}`, {
        method: init.method,
        headers: {
-          authorization: `Bearer ${session.token}`,
+          authorization: `Bearer ${outcome.session.token}`,
          ...(init.body === undefined ? {} : { 'content-type': 'application/json' })
        },
        signal: AbortSignal.timeout(PUSH_REQUEST_DEADLINE_MS),
        ...(init.body === undefined ? {} : { body: JSON.stringify(init.body) })
      })
-      return { ok: true, response }
-    } catch {
-      return { ok: false, reason: 'unreachable' }
-    }
-  }
-
-  private async ensureSession(forceReauth: boolean): Promise<CachedSession | null> {
-    if (forceReauth) {
-      this.session = null
-    }
-    const cached = this.session
-    if (cached && cached.expiresAt - SESSION_RENEWAL_MARGIN_MS > this.now()) {
-      return cached
-    }
-    // Concurrent sends must not each burn a challenge; share one handshake.
-    this.pendingSession ??= this.openSession().finally(() => {
-      this.pendingSession = null
-    })
-    return await this.pendingSession
-  }
-
-  private async openSession(): Promise<CachedSession | null> {
-    const challenge = await this.requestChallenge()
-    if (!challenge) {
-      return null
-    }
-    const proofB64 = answerPushHostChallenge(challenge, {
-      gatewayOrigin: this.origin,
-      hostFingerprint: this.hostFingerprint,
-      hostPublicKey: this.keypair.publicKey,
-      hostSecretKey: this.keypair.secretKey,
-      now: this.now
-    })
-    if (!proofB64) {
-      return null
-    }
-    const response = await this.post('/v1/host/session', {
-      v: 1,
-      challengeId: challenge.challengeId,
-      proofB64
-    })
-    const parsed = await readJson(response, SessionResponseSchema)
-    if (!parsed.ok || parsed.value.hostFingerprint !== this.hostFingerprint) {
-      return null
-    }
-    this.session = { token: parsed.value.sessionToken, expiresAt: parsed.value.expiresAt }
-    return this.session
-  }
-
-  private async requestChallenge(): Promise<PushHostChallenge | null> {
-    const response = await this.post('/v1/host/challenge', {
-      v: 1,
-      hostPublicKeyB64: this.keypair.publicKeyB64
-    })
-    const parsed = await readJson(response, ChallengeResponseSchema)
-    return parsed.ok ? parsed.value : null
-  }
-
-  private async post(
-    path: string,
-    body: unknown
-  ): Promise<{ ok: true; response: Response } | PushGatewayFailure> {
-    try {
-      const response = await this.fetchImpl(`${this.origin}${path}`, {
-        method: 'POST',
-        headers: { 'content-type': 'application/json' },
-        signal: AbortSignal.timeout(PUSH_REQUEST_DEADLINE_MS),
-        body: JSON.stringify(body)
-      })
-      return { ok: true, response }
+      return { ok: true, response, token: outcome.session.token }
    } catch {
      return { ok: false, reason: 'unreachable' }
    }
  }
 }
-
-async function readJson<TSchema extends z.ZodType>(
-  result: { ok: true; response: Response } | PushGatewayFailure,
-  schema: TSchema
-): Promise<{ ok: true; value: z.infer<TSchema> } | PushGatewayFailure> {
-  if (!result.ok) {
-    return result
-  }
-  const { response } = result
-  if (!response.ok) {
-    await cancelUnreadResponseBody(response)
-    // 5xx and 429 are worth another attempt later; anything else is the gateway
-    // refusing this request as written.
-    return {
-      ok: false,
-      reason: response.status >= 500 || response.status === 429 ? 'unreachable' : 'rejected'
-    }
-  }
-  let payload: unknown
-  try {
-    payload = await response.json()
-  } catch {
-    await cancelUnreadResponseBody(response)
-    return { ok: false, reason: 'unreachable' }
-  }
-  const parsed = schema.safeParse(payload)
-  return parsed.success ? { ok: true, value: parsed.data } : { ok: false, reason: 'rejected' }
-}
@@ -0,0 +1,58 @@
+// Why: the authorized request path and the handshake that authorizes it must
+// classify a gateway response identically — otherwise the same 503 means "retry"
+// on one leg and "give up" on the other, and register/send disagree about why.
+import type { z } from 'zod'
+import { cancelUnreadResponseBody } from '../../lib/unread-response-body'
+
+export const PUSH_REQUEST_DEADLINE_MS = 15_000
+
+export type PushGatewayFailure = { ok: false; reason: 'unreachable' | 'rejected' }
+export type PushGatewayResult<T> = ({ ok: true } & T) | PushGatewayFailure
+export type PushGatewayResponse = { ok: true; response: Response } | PushGatewayFailure
+
+/** Unauthenticated POST; the handshake legs run before any session exists. */
+export async function postPushGatewayJson(
+  fetchImpl: typeof globalThis.fetch,
+  url: string,
+  body: unknown
+): Promise<PushGatewayResponse> {
+  try {
+    const response = await fetchImpl(url, {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      signal: AbortSignal.timeout(PUSH_REQUEST_DEADLINE_MS),
+      body: JSON.stringify(body)
+    })
+    return { ok: true, response }
+  } catch {
+    return { ok: false, reason: 'unreachable' }
+  }
+}
+
+export async function readPushGatewayJson<TSchema extends z.ZodType>(
+  result: PushGatewayResponse,
+  schema: TSchema
+): Promise<{ ok: true; value: z.infer<TSchema> } | PushGatewayFailure> {
+  if (!result.ok) {
+    return result
+  }
+  const { response } = result
+  if (!response.ok) {
+    await cancelUnreadResponseBody(response)
+    // 5xx and 429 are worth another attempt later; anything else is the gateway
+    // refusing this request as written.
+    return {
+      ok: false,
+      reason: response.status >= 500 || response.status === 429 ? 'unreachable' : 'rejected'
+    }
+  }
+  let payload: unknown
+  try {
+    payload = await response.json()
+  } catch {
+    await cancelUnreadResponseBody(response)
+    return { ok: false, reason: 'unreachable' }
+  }
+  const parsed = schema.safeParse(payload)
+  return parsed.success ? { ok: true, value: parsed.data } : { ok: false, reason: 'rejected' }
+}
@@ -0,0 +1,169 @@
+import { createHash } from 'node:crypto'
+import { describe, expect, it, vi } from 'vitest'
+import { buildPushChallengeFixture, createPushHostKeypair } from './push-host-challenge-fixtures'
+import { PushGatewaySession, type PushSessionOutcome } from './push-gateway-session'
+
+const GATEWAY_ORIGIN = 'https://push.onorca.dev'
+const NOW = 1_770_000_000_000
+
+function jsonResponse(status: number, body: unknown): Response {
+  return new Response(JSON.stringify(body), {
+    status,
+    headers: { 'content-type': 'application/json' }
+  })
+}
+
+function tokenOf(outcome: PushSessionOutcome): string | null {
+  return outcome.ok ? outcome.session.token : null
+}
+
+function createSessionHarness(
+  options: { sessionStatus?: number; challengeStatus?: number; wrongFingerprint?: boolean } = {}
+): {
+  session: PushGatewaySession
+  challenges: () => number
+  requests: () => number
+  now: { value: number }
+} {
+  const hostKeypair = createPushHostKeypair()
+  const hostFingerprint = createHash('sha256')
+    .update(hostKeypair.publicKey)
+    .digest('base64url')
+    .slice(0, 16)
+  const now = { value: NOW }
+  let issued = 0
+  let requests = 0
+  let pendingProof: string | null = null
+
+  const fetchImpl = (async (input: string, init?: RequestInit): Promise<Response> => {
+    const url = String(input)
+    requests += 1
+    if (url.endsWith('/v1/host/challenge')) {
+      if (options.challengeStatus) {
+        return jsonResponse(options.challengeStatus, { error: 'rate_limited' })
+      }
+      const built = buildPushChallengeFixture({
+        hostKeypair,
+        gatewayOrigin: GATEWAY_ORIGIN,
+        hostFingerprint,
+        issuedAt: now.value,
+        challengeId: `challenge-${++issued}`
+      })
+      pendingProof = built.proof
+      return jsonResponse(200, built.challenge)
+    }
+    if (options.sessionStatus) {
+      return jsonResponse(options.sessionStatus, { error: 'nope' })
+    }
+    const body = init?.body ? (JSON.parse(String(init.body)) as { proofB64: string }) : null
+    if (body?.proofB64 !== pendingProof) {
+      return jsonResponse(401, { error: 'bad_proof' })
+    }
+    return jsonResponse(200, {
+      sessionToken: `session-${issued}`,
+      expiresAt: now.value + 24 * 60 * 60_000,
+      hostFingerprint: options.wrongFingerprint ? 'someone-else' : hostFingerprint
+    })
+  }) as unknown as typeof globalThis.fetch
+
+  return {
+    session: new PushGatewaySession({
+      origin: GATEWAY_ORIGIN,
+      keypair: hostKeypair,
+      fetchImpl,
+      now: () => now.value
+    }),
+    challenges: () => issued,
+    requests: () => requests,
+    now
+  }
+}
+
+describe('PushGatewaySession', () => {
+  it('reuses the cached session until it nears expiry', async () => {
+    const harness = createSessionHarness()
+
+    expect(tokenOf(await harness.session.ensure(null))).toBe('session-1')
+    expect(tokenOf(await harness.session.ensure(null))).toBe('session-1')
+    expect(harness.challenges()).toBe(1)
+  })
+
+  it('drops only the exact session that received the 401', async () => {
+    const harness = createSessionHarness()
+    expect(tokenOf(await harness.session.ensure(null))).toBe('session-1')
+
+    // A request that 401ed on session-1 forces a fresh handshake.
+    expect(tokenOf(await harness.session.ensure('session-1'))).toBe('session-2')
+    // A second request whose 401 also named session-1 must keep the new token.
+    expect(tokenOf(await harness.session.ensure('session-1'))).toBe('session-2')
+    expect(harness.challenges()).toBe(2)
+  })
+
+  it('reports a refused handshake as rejected rather than unreachable', async () => {
+    const harness = createSessionHarness({ sessionStatus: 403 })
+
+    expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'rejected' })
+  })
+
+  it('reports a session minted for another host as rejected', async () => {
+    const harness = createSessionHarness({ wrongFingerprint: true })
+
+    expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'rejected' })
+  })
+
+  it('caches a refusal briefly instead of re-handshaking on every call', async () => {
+    const harness = createSessionHarness({ sessionStatus: 403 })
+
+    await harness.session.ensure(null)
+    await harness.session.ensure(null)
+    expect(harness.challenges()).toBe(1)
+
+    harness.now.value += 30_000
+    await harness.session.ensure(null)
+    expect(harness.challenges()).toBe(2)
+  })
+
+  it('never caches a transport failure, which may clear on the next try', async () => {
+    const fetchImpl = vi.fn(async () => {
+      throw new Error('network down')
+    }) as unknown as typeof globalThis.fetch
+    const session = new PushGatewaySession({
+      origin: GATEWAY_ORIGIN,
+      keypair: createPushHostKeypair(),
+      fetchImpl,
+      now: () => NOW
+    })
+
+    expect(await session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' })
+    expect(await session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' })
+    expect(fetchImpl).toHaveBeenCalledTimes(2)
+  })
+
+  it('reports a rate-limited challenge as unreachable and backs off', async () => {
+    const harness = createSessionHarness({ challengeStatus: 429 })
+
+    expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' })
+    expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' })
+    expect(harness.requests()).toBe(1)
+
+    harness.now.value += 60_000
+    await harness.session.ensure(null)
+    expect(harness.requests()).toBe(2)
+  })
+
+  it('reports a rate-limited session mint as unreachable, not refused', async () => {
+    const harness = createSessionHarness({ sessionStatus: 429 })
+
+    expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' })
+    // Cached for a minute, so the next dispatch does not spend more of the bucket.
+    expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' })
+    expect(harness.challenges()).toBe(1)
+  })
+
+  it('shares one handshake across concurrent callers', async () => {
+    const harness = createSessionHarness()
+
+    await Promise.all([harness.session.ensure(null), harness.session.ensure(null)])
+    expect(harness.challenges()).toBe(1)
+  })
+})
@@ -0,0 +1,157 @@
+// Why: the challenge/proof handshake every push request rides on, split out of
+// push-gateway-client.ts so the session cache and its refusal cache stay readable
+// next to the request methods rather than buried under them.
+import { z } from 'zod'
+import { cancelUnreadResponseBody } from '../../lib/unread-response-body'
+import type { E2EEKeypair } from '../e2ee-keypair'
+import { deriveRelayHostId } from '../relay/relay-http-client'
+import { answerPushHostChallenge } from './push-host-proof'
+import {
+  postPushGatewayJson,
+  readPushGatewayJson,
+  type PushGatewayFailure
+} from './push-gateway-response'
+
+// Re-auth a little early so a send never spends its one retry on a token that
+// expired between the check and the request.
+const SESSION_RENEWAL_MARGIN_MS = 60_000
+// Why: a gateway that refuses this host's proof refuses the identical next one,
+// so without this every dispatch pays two full handshake round trips to relearn it.
+const HANDSHAKE_REFUSAL_TTL_MS = 30_000
+// Why: the handshake routes sit behind a per-IP bucket. Backing off keeps this
+// host from spending the whole bucket on challenges it will never get to use.
+const HANDSHAKE_RATE_LIMIT_TTL_MS = 60_000
+
+const ChallengeResponseSchema = z
+  .object({
+    challengeId: z.string().min(1).max(512),
+    gatewayEphemeralPublicKeyB64: z.string().min(1).max(128),
+    nonceB64: z.string().min(1).max(128),
+    ciphertextB64: z
+      .string()
+      .min(1)
+      .max(8 * 1024),
+    expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER)
+  })
+  .strict()
+
+const SessionResponseSchema = z
+  .object({
+    sessionToken: z.string().min(1).max(1024),
+    expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER),
+    hostFingerprint: z.string().min(1).max(64)
+  })
+  .strict()
+
+export type PushSession = { token: string; expiresAt: number }
+export type PushSessionOutcome = { ok: true; session: PushSession } | PushGatewayFailure
+
+type PushGatewaySessionOptions = {
+  origin: string
+  keypair: E2EEKeypair
+  fetchImpl: typeof globalThis.fetch
+  now: () => number
+}
+
+export class PushGatewaySession {
+  private readonly origin: string
+  private readonly keypair: E2EEKeypair
+  private readonly fetchImpl: typeof globalThis.fetch
+  private readonly now: () => number
+  readonly hostFingerprint: string
+  private session: PushSession | null = null
+  private pending: Promise<PushSessionOutcome> | null = null
+  private negative: { until: number; reason: PushGatewayFailure['reason'] } | null = null
+
+  constructor(options: PushGatewaySessionOptions) {
+    this.origin = options.origin
+    this.keypair = options.keypair
+    this.fetchImpl = options.fetchImpl
+    this.now = options.now
+    this.hostFingerprint = deriveRelayHostId(options.keypair.publicKey)
+  }
+
+  /**
+   * `staleToken` is the token that just received a 401. Only that exact session is
+   * dropped: a concurrent request may already have installed a good one, and
+   * clearing unconditionally would throw it away and re-handshake for nothing.
+   */
+  async ensure(staleToken: string | null): Promise<PushSessionOutcome> {
+    if (staleToken !== null && this.session?.token === staleToken) {
+      this.session = null
+    }
+    const cached = this.session
+    if (cached && cached.expiresAt - SESSION_RENEWAL_MARGIN_MS > this.now()) {
+      return { ok: true, session: cached }
+    }
+    if (this.negative && this.negative.until > this.now()) {
+      return { ok: false, reason: this.negative.reason }
+    }
+    // Concurrent sends must not each burn a challenge; share one handshake.
+    this.pending ??= this.open().finally(() => {
+      this.pending = null
+    })
+    return await this.pending
+  }
+
+  private async open(): Promise<PushSessionOutcome> {
+    const challenge = await this.handshakePost(
+      '/v1/host/challenge',
+      { v: 1, hostPublicKeyB64: this.keypair.publicKeyB64 },
+      ChallengeResponseSchema
+    )
+    if (!challenge.ok) {
+      return this.remember(challenge)
+    }
+    const proofB64 = answerPushHostChallenge(challenge.value, {
+      gatewayOrigin: this.origin,
+      hostFingerprint: this.hostFingerprint,
+      hostPublicKey: this.keypair.publicKey,
+      hostSecretKey: this.keypair.secretKey,
+      now: this.now
+    })
+    if (!proofB64) {
+      // A challenge this host cannot answer is a refusal, not a dropped packet.
+      return this.remember({ ok: false, reason: 'rejected' })
+    }
+    const parsed = await this.handshakePost(
+      '/v1/host/session',
+      { v: 1, challengeId: challenge.value.challengeId, proofB64 },
+      SessionResponseSchema
+    )
+    if (!parsed.ok) {
+      return this.remember(parsed)
+    }
+    if (parsed.value.hostFingerprint !== this.hostFingerprint) {
+      // The gateway answered for some other host; that token is never usable here.
+      return this.remember({ ok: false, reason: 'rejected' })
+    }
+    this.session = { token: parsed.value.sessionToken, expiresAt: parsed.value.expiresAt }
+    this.negative = null
+    return { ok: true, session: this.session }
+  }
+
+  private async handshakePost<TSchema extends z.ZodType>(
+    path: string,
+    body: unknown,
+    schema: TSchema
+  ): Promise<{ ok: true; value: z.infer<TSchema> } | PushGatewayFailure> {
+    const response = await postPushGatewayJson(this.fetchImpl, `${this.origin}${path}`, body)
+    if (response.ok && response.response.status === 429) {
+      await cancelUnreadResponseBody(response.response)
+      // Rate limiting refuses the moment, not this host: back off, stay retryable
+      // so register reports gateway_unreachable and send keeps its one retry.
+      this.negative = { until: this.now() + HANDSHAKE_RATE_LIMIT_TTL_MS, reason: 'unreachable' }
+      return { ok: false, reason: 'unreachable' }
+    }
+    return await readPushGatewayJson(response, schema)
+  }
+
+  /** Caches refusals only: a transport failure may clear on the very next try. */
+  private remember(failure: PushGatewayFailure): PushGatewayFailure {
+    if (failure.reason === 'rejected') {
+      this.negative = { until: this.now() + HANDSHAKE_REFUSAL_TTL_MS, reason: 'rejected' }
+    }
+    return failure
+  }
+}
@@ -72,6 +72,9 @@ function installBeforeQuitHandler(): void {
    }
    state.isQuitting = true
    state.desktopRelayService?.fenceAndCloseNow()
+    // Why: drops the notification subscription so a late dispatch cannot start a
+    // push (and its unref'd outbox retry) on the way out.
+    state.desktopPushService?.stop()
    state.runtimeRpc?.setMobileRelayPairingProvider(null)
    state.unsubscribeAgentAwakeStatusChanges?.()
    state.unsubscribeAgentAwakeStatusChanges = null
@@ -38,7 +38,17 @@ export type MobilePushRegisterInput = {

 export type MobilePushRegisterResult =
  | { registered: true; registrationId: string }
-  | { registered: false; reason: 'gateway_unreachable' | 'gateway_rejected' | 'not_mobile' }
+  | {
+      registered: false
+      // `registration_storage_failed`: the gateway accepted the token but the host
+      // could not persist it, so the phone must register again rather than believe
+      // a push route that does not exist.
+      reason:
+        | 'gateway_unreachable'
+        | 'gateway_rejected'
+        | 'not_mobile'
+        | 'registration_storage_failed'
+    }

 function isStringMember<T extends string>(value: unknown, members: readonly T[]): value is T {
  return typeof value === 'string' && (members as readonly string[]).includes(value)