fix(relay): stop reviving a session outside the cell's map, keep the seat feed on live sessions, and log demote outcomes (#32149)

* fix(relay): repair cell seat-log drift, and keep a new session's generation above a listed seat

Directors mirror each cell's seat feed, so the feed must list exactly the cell's live sessions.
The cell now sweeps its seat log against its sessions on a director's poll: a seat no session
backs is left (4409) and the live session listed, and a live session the feed lost is listed
again, each once two sweeps in a row find it. A demotion naming an unbacked seat repairs it at
once. A host's new session now starts above any generation the feed still lists, which closes
the drained-session reconnect race that removed a live seat.

Claude-Session: asia-drain-pulse

* fix(relay): refuse a rebind whose session left the map during activation, and keep a per-host generation floor

A rebind whose database activation outlived the orphan grace (or a drain) revived the removed
session outside the map: connected, still in the seat feed, and unreachable by phones or
demotion for as long as the host stayed up. Activation now rechecks the map after its awaits and
refuses with PEER_DROPPED so the host redials. Generations come from a per-host floor that
outlives the session, and an activation at a generation already used here is refused.

Claude-Session: asia-drain-pulse

* fix(relay): log each demote outcome on the director by host digest

The director now reads the cell's existing {v:1, outcome} demote reply and logs it; the cell's
feed stays the only source of its map. The supersede line now reaches the log as well.

Claude-Session: asia-drain-pulse

cloud/apps/relay/src/cell-reserve-admission.blackbox.test.ts

@@ -8,10 +8,13 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import WebSocket from 'ws'
 import type { RawData } from 'ws'
 import { CELL_FLAG_DEFAULTS, type CellFlags } from './cell-flags.js'
+import { CELL_SEAT_DRIFT_SWEEP_MS } from './cell-seat-drift.js'
+import type { CellSeatLog } from './cell-seat-log.js'
 import type { RelayConfig } from './config.js'
 import { PostgresDatabase } from './database.js'
 import type { AppliedControlFlags } from './relay-control-flag-channel.js'
 import { createRelayServer, HOST_HELLO_SHED_OLDEST_WAIT_MS } from './relay-server.js'
+import { RELAY_WEBSOCKET_FORCE_CLOSE_MS } from './relay-websocket-close.js'
 
 // A step-5 cell whose database pool is wedged: every query waits forever and the hello shed
 // is armed. Booked and recently seated hosts must still connect; everything else must not.
@@ -71,6 +74,20 @@ function cellConfig(port: number, issuer: string): RelayConfig {
   }
 }
 
+// The registry's private feed log, to plant the drift the cell must repair.
+function seatLogOf(sessions: object): CellSeatLog {
+  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: HostSessionRegistry holds its feed log as `seatLog`.
+  return (sessions as { seatLog: CellSeatLog }).seatLog
+}
+
+// The JSON lines a test's console.warn spy saw for one event.
+function warned(event: string): Array<Record<string, unknown>> {
+  return vi
+    .mocked(console.warn)
+    .mock.calls.map(([line]) => (typeof line === 'string' && line.startsWith('{') ? JSON.parse(line) : null))
+    .filter((entry) => entry?.event === event)
+}
+
 function nextMessage(socket: WebSocket): Promise<Record<string, unknown>> {
   return new Promise((resolve, reject) => {
     socket.once('message', (data: RawData) => {
@@ -249,6 +266,9 @@ describe('a reserve-mode cell admits from memory with its database wedged', () =
       book,
       setFlags: (next: Partial<CellFlags>) => {
         applied = { generation: applied.generation + 1, flags: { ...CELL_FLAG_DEFAULTS, ...next } }
+      },
+      advance: (ms: number) => {
+        elapsedMs += ms
       }
     }
   }
@@ -330,6 +350,150 @@ describe('a reserve-mode cell admits from memory with its database wedged', () =
     expect(cell.book(identity, 9)).toEqual([{ outcome: 'ok' }])
   })
 
+  describe('keeps its seat feed on its live sessions', () => {
+    const seatEpochs = (cell: { relay: { sessions: { seatFeed(since: number | null): unknown } } }) => {
+      const page = cell.relay.sessions.seatFeed(null)
+      // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a null cursor always answers a full page.
+      return (page as { full: Array<{ relayHostId: string; epoch: number }> }).full.map((entry) => entry.epoch)
+    }
+
+    it('repairs at once the seat a director names when no session backs it', async () => {
+      const cell = await startCell({ admitMode: 'reserve' })
+      const identity = await cell.host()
+      cell.book(identity, 9)
+      await cell.connect(identity, 9)
+      // A stale seat at a generation above the live session's: its changes could never replace it.
+      const log = seatLogOf(cell.relay.sessions)
+      log.append({ kind: 'join', userId: 'user-1', relayHostId: identity.hostId, epoch: 8, generation: 5, at: 4_242 })
+      const before = cell.relay.sessions.seatFeed(null)
+      const seq = 'full' in before ? before.seq : 0
+      const seat = { v: 1 as const, userId: 'user-1', relayHostId: identity.hostId, epoch: 8, joinedAt: 4_242 }
+      expect(cell.relay.sessions.demote(seat)).toBe('not-seated')
+      expect(cell.relay.sessions.seatFeed(seq)).toMatchObject({
+        changes: [
+          { kind: 'leave', epoch: 8, generation: 5, closeCode: 4409 },
+          { kind: 'join', epoch: 9, generation: 1 }
+        ]
+      })
+      expect(seatEpochs(cell)).toEqual([9])
+      expect(warned('orca_relay_cell_seat_log_drift')).toMatchObject([
+        { kind: 'unbacked-seat', via: 'demote', reason: 'epoch', seatEpoch: 8, sessionEpoch: 9, relisted: true }
+      ])
+      expect(JSON.stringify(warned('orca_relay_cell_seat_log_drift'))).not.toContain(identity.hostId)
+      // Asked again, there is nothing left to repair.
+      expect(cell.relay.sessions.demote(seat)).toBe('not-seated')
+      expect(cell.relay.sessions.seatFeed(seq + 2)).toMatchObject({ changes: [] })
+    })
+
+    it('leaves a seat no session backs once two sweeps in a row find it, with no director asking', async () => {
+      const cell = await startCell({ admitMode: 'db' })
+      const ghost = await cell.host()
+      seatLogOf(cell.relay.sessions).append({
+        kind: 'join',
+        userId: 'user-1',
+        relayHostId: ghost.hostId,
+        epoch: 10,
+        generation: 3,
+        at: 1_000
+      })
+      expect(seatEpochs(cell)).toEqual([10])
+      cell.advance(CELL_SEAT_DRIFT_SWEEP_MS)
+      expect(seatEpochs(cell)).toEqual([])
+      expect(warned('orca_relay_cell_seat_log_drift')).toMatchObject([
+        { kind: 'unbacked-seat', via: 'sweep', reason: 'no-session', seatEpoch: 10, seatGeneration: 3, relisted: false }
+      ])
+      // Left as moved: the cell will not seat it again at 10 from memory.
+      expect(cell.relay.sessions.seatFeed(0)).toMatchObject({ changes: [{}, { kind: 'leave', closeCode: 4409 }] })
+    })
+
+    it('lists again a live session the feed lost, once two sweeps in a row find it', async () => {
+      const cell = await startCell({ admitMode: 'reserve' })
+      const identity = await cell.host()
+      cell.book(identity, 9)
+      await cell.connect(identity, 9)
+      seatLogOf(cell.relay.sessions).append({
+        kind: 'leave',
+        userId: 'user-1',
+        relayHostId: identity.hostId,
+        epoch: 9,
+        generation: 1,
+        at: 5_000
+      })
+      expect(seatEpochs(cell)).toEqual([])
+      cell.advance(CELL_SEAT_DRIFT_SWEEP_MS)
+      expect(seatEpochs(cell)).toEqual([9])
+      expect(warned('orca_relay_cell_seat_log_drift')).toMatchObject([{ kind: 'unlisted-session', sessionEpoch: 9 }])
+    })
+
+    // A drained session's seat stays listed, with no session, until its socket's close lands.
+    it('repairs nothing for a seat that one sweep finds unbacked and is gone by the next', async () => {
+      const cell = await startCell({ admitMode: 'reserve' })
+      const identity = await cell.host()
+      cell.book(identity, 9)
+      const live = await cell.connect(identity, 9)
+      // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: ws keeps its net.Socket as `_socket`.
+      ;(live.socket as unknown as { _socket: { pause(): void } })._socket.pause()
+      cell.relay.sessions.drainHost({
+        attemptId: 'attempt-1',
+        userId: 'user-1',
+        relayHostId: identity.hostId,
+        sourceAssignmentEpoch: 9,
+        graceMs: 0
+      })
+      await new Promise((resolve) => setTimeout(resolve, 20))
+      expect(seatEpochs(cell)).toEqual([9])
+      await new Promise((resolve) => setTimeout(resolve, RELAY_WEBSOCKET_FORCE_CLOSE_MS + 300))
+      cell.advance(CELL_SEAT_DRIFT_SWEEP_MS)
+      expect(seatEpochs(cell)).toEqual([])
+      expect(warned('orca_relay_cell_seat_log_drift')).toEqual([])
+    }, 10_000)
+
+    // A drained session leaves the map at once, but its socket closes only after its release (up
+    // to 3 s) and the peer's close reply (up to the 1 s force-close). A desktop of the same
+    // identity dialling back in that window restarted at generation 1, so the old socket's late
+    // leave, at generation 1 too, removed the live seat from the feed.
+    it('keeps a session that reconnected while its drained predecessor was still closing', async () => {
+      const cell = await startCell({ admitMode: 'reserve' })
+      const identity = await cell.host()
+      cell.book(identity, 9)
+      const first = await cell.connect(identity, 9)
+      expect(first.ack).toMatchObject({ generation: 1 })
+      // A peer slow to answer the close frame: the cell's side closes only at its force-close.
+      // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: ws keeps its net.Socket as `_socket`.
+      ;(first.socket as unknown as { _socket: { pause(): void } })._socket.pause()
+      expect(
+        cell.relay.sessions.drainHost({
+          attemptId: 'attempt-1',
+          userId: 'user-1',
+          relayHostId: identity.hostId,
+          sourceAssignmentEpoch: 9,
+          graceMs: 0
+        })
+      ).toBe('accepted')
+      await new Promise((resolve) => setTimeout(resolve, 20))
+      cell.book(identity, 10)
+      expect((await cell.connect(identity, 10)).ack).toMatchObject({ type: 'host-hello-ack', generation: 2 })
+      // Past the force-close, so the old socket's leave has landed.
+      await new Promise((resolve) => setTimeout(resolve, RELAY_WEBSOCKET_FORCE_CLOSE_MS + 300))
+      expect(seatEpochs(cell)).toEqual([10])
+    }, 10_000)
+
+    // The same identity connecting afresh at each turn, as desktops sharing one identity do: each
+    // turn replaces the session and its seat together, so no sweep finds anything to repair.
+    it('stays on the live session when the same identity connects afresh each turn', async () => {
+      const cell = await startCell({ admitMode: 'reserve' })
+      const identity = await cell.host()
+      for (const epoch of [9, 10, 11]) {
+        cell.book(identity, epoch)
+        expect((await cell.connect(identity, epoch)).ack).toMatchObject({ type: 'host-hello-ack' })
+        expect(seatEpochs(cell)).toEqual([epoch])
+        cell.advance(CELL_SEAT_DRIFT_SWEEP_MS)
+      }
+      expect(seatEpochs(cell)).toEqual([11])
+      expect(warned('orca_relay_cell_seat_log_drift')).toEqual([])
+    })
+  })
+
   it('leaves a drained seat to its drain grace instead of the shorter demotion close', async () => {
     const cell = await startCell({ admitMode: 'reserve' })
     const identity = await cell.host()
@@ -341,6 +505,8 @@ describe('a reserve-mode cell admits from memory with its database wedged', () =
     const seat = { v: 1 as const, userId: 'user-1', relayHostId: identity.hostId, epoch: 9, joinedAt }
     expect(cell.relay.sessions.demote(seat)).toBe('draining')
     expect(cell.relay.sessions.demote(seat)).toBe('draining')
+    const after = cell.relay.sessions.seatFeed(null)
+    expect('full' in after && after.full.map((entry) => entry.epoch)).toEqual([9])
   })
 
   it('gives every booking unit back when the cell starts to drain', async () => {

cloud/apps/relay/src/cell-reserve-client.ts

@@ -2,6 +2,7 @@ import {
   CELL_DEMOTE_PATH,
   CELL_RESERVE_MAX_ITEMS,
   CELL_RESERVE_PATH,
+  DemoteResponseSchema,
   ReserveResponseSchema,
   type DemoteRequest,
   type ReserveItem
@@ -46,14 +47,18 @@ export class CellReserveClient {
     })
   }
 
-  // The cell treats repeats as one; a failure is only logged, and the next poll retries.
-  async demote(target: ReserveTarget, request: DemoteRequest): Promise<boolean> {
+  // The cell treats repeats as one. Null: no answer the director can read; the next poll retries.
+  async demote(target: ReserveTarget, request: DemoteRequest): Promise<string | null> {
     try {
       const response = await this.post(target, CELL_DEMOTE_PATH, request)
-      await response.body?.cancel().catch(() => undefined)
-      return response.ok
+      if (!response.ok) {
+        await response.body?.cancel().catch(() => undefined)
+        return null
+      }
+      const body = DemoteResponseSchema.safeParse(await response.json().catch(() => null))
+      return body.success ? body.data.outcome : null
     } catch {
-      return false
+      return null
     }
   }
 

cloud/apps/relay/src/cell-reserve-contract.ts

@@ -56,3 +56,8 @@ export const DemoteRequestSchema = z.object({
   joinedAt: z.number().int().nonnegative()
 })
 export type DemoteRequest = z.infer<typeof DemoteRequestSchema>
+
+// `not-seated`: the cell holds no live session for that seat. Lenient for the same reason as
+// ReserveResponseSchema; every image since the route shipped answers `outcome`.
+export const DemoteResponseSchema = z.object({ v: z.literal(1), outcome: z.string() }).passthrough()
+export type DemoteOutcome = 'demoted' | 'already-demoted' | 'draining' | 'not-seated' | 'off'

cloud/apps/relay/src/cell-seat-drift.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, it } from 'vitest'
+import { CELL_SEAT_DRIFT_SWEEP_MS, SeatDriftConfirmation, unbackedSeatReason } from './cell-seat-drift.js'
+
+describe('cell seat drift', () => {
+  const seat = { epoch: 10, generation: 3 }
+  const live = { state: 'active', socket: {}, assignmentEpoch: 10, generation: 3 }
+
+  it('names why no live session backs a seat, and nothing when one does', () => {
+    expect(unbackedSeatReason(seat, live)).toBeNull()
+    expect(unbackedSeatReason(seat, { ...live, state: 'drain-only' })).toBeNull()
+    expect(unbackedSeatReason(seat, null)).toBe('no-session')
+    expect(unbackedSeatReason(seat, { ...live, state: 'closed' })).toBe('closed')
+    expect(unbackedSeatReason(seat, { ...live, socket: null })).toBe('no-socket')
+    expect(unbackedSeatReason(seat, { ...live, assignmentEpoch: 11 })).toBe('epoch')
+    expect(unbackedSeatReason(seat, { ...live, generation: 1 })).toBe('generation')
+  })
+
+  it('confirms only what two sweeps in a row found, at most once per sweep interval', () => {
+    const drift = new SeatDriftConfirmation()
+    expect(drift.due(0)).toBe(true)
+    expect(drift.confirm(['a', 'b'])).toEqual(new Set())
+    expect(drift.due(CELL_SEAT_DRIFT_SWEEP_MS - 1)).toBe(false)
+    expect(drift.due(CELL_SEAT_DRIFT_SWEEP_MS)).toBe(true)
+    expect(drift.confirm(['b', 'c'])).toEqual(new Set(['b']))
+    expect(drift.confirm(['c'])).toEqual(new Set(['c']))
+  })
+})

cloud/apps/relay/src/cell-seat-drift.ts

@@ -0,0 +1,47 @@
+import type { CellSeat } from './cell-seat-log.js'
+
+// How often the cell checks its seat log against its live sessions, on a director's feed poll.
+export const CELL_SEAT_DRIFT_SWEEP_MS = 30_000
+
+export type SeatBackingSession = {
+  state: string
+  socket: unknown
+  assignmentEpoch: number
+  generation: number
+}
+
+export type UnbackedSeatReason = 'no-session' | 'closed' | 'no-socket' | 'epoch' | 'generation'
+
+// Why no live session backs this seat, or null when one does. A session backs its seat only at
+// the epoch and generation it joined with; anything else is a seat the feed should not list.
+export function unbackedSeatReason(
+  seat: Pick<CellSeat, 'epoch' | 'generation'>,
+  session: SeatBackingSession | null
+): UnbackedSeatReason | null {
+  if (!session) return 'no-session'
+  if (session.state === 'closed') return 'closed'
+  if (!session.socket) return 'no-socket'
+  if (session.assignmentEpoch !== seat.epoch) return 'epoch'
+  if (session.generation !== seat.generation) return 'generation'
+  return null
+}
+
+// Drift found on two sweeps in a row is repaired: a drained session's seat stays listed for the
+// few seconds until its socket's close lands, and that is not drift.
+export class SeatDriftConfirmation {
+  private seen = new Set<string>()
+  private sweptAt = Number.NEGATIVE_INFINITY
+
+  due(now: number): boolean {
+    if (now - this.sweptAt < CELL_SEAT_DRIFT_SWEEP_MS) return false
+    this.sweptAt = now
+    return true
+  }
+
+  // Returns the ids seen on the previous sweep too, and remembers this sweep's for the next.
+  confirm(ids: readonly string[]): Set<string> {
+    const confirmed = new Set(ids.filter((id) => this.seen.has(id)))
+    this.seen = new Set(ids)
+    return confirmed
+  }
+}

cloud/apps/relay/src/cell-seat-log.ts

@@ -101,6 +101,10 @@ export class CellSeatLog {
     return { seq: lastSeq, changes, more: lastSeq < this.headSeq }
   }
 
+  seatList(): CellSeat[] {
+    return [...this.seats.values()]
+  }
+
   seatOf(userId: string, relayHostId: string): CellSeat | undefined {
     return this.seats.get(JSON.stringify([userId, relayHostId]))
   }

cloud/apps/relay/src/host-session-registry.test.ts

@@ -584,6 +584,63 @@ describe('host session cleanup races', () => {
     )
   })
 
+  // The root of a seat c34's feed listed for hours: a rebind whose database activation outlived
+  // the orphan grace revived the removed session outside the map, still listed and unreachable.
+  it('refuses a rebind whose session left the map while its activation was in flight', async () => {
+    const blocked = deferred<string>()
+    const activateControl = vi
+      .fn<RelayAssignmentStore['activateControl']>()
+      .mockResolvedValueOnce('control:production-gce-c3:1')
+      .mockReturnValueOnce(blocked.promise)
+    const { registry, activate, releaseActivity } = createRegistry(activateControl)
+    const key = { userId: identity.sub, relayHostId: identity.relayHostId }
+    const originalSocket = new FakeSocket()
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface.
+    await activate(originalSocket as unknown as WebSocket, identity, null, 1, false, 1)
+    const original = registry.get(key)
+    originalSocket.close(1006, 'network reset')
+    expect(original?.state).toBe('orphaned')
+
+    const rebindSocket = new FakeSocket()
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface.
+    const rebinding = activate(rebindSocket as unknown as WebSocket, identity, original, 1, true, 1)
+    await vi.advanceTimersByTimeAsync(0)
+    expect(activateControl).toHaveBeenCalledTimes(2)
+    // The grace runs out while the database answers slowly.
+    await vi.advanceTimersByTimeAsync(CONTROL_CONTINUITY_LIMITS.orphanGraceMs)
+    expect(registry.get(key)).toBeNull()
+    blocked.resolve('control:production-gce-c3:1')
+    await rebinding
+
+    expect(rebindSocket.close).toHaveBeenCalledWith(RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded')
+    expect(registry.get(key)).toBeNull()
+    expect(original?.state).toBe('closed')
+    const page = registry.seatFeed(null)
+    expect('full' in page && page.full).toEqual([])
+    expect(releaseActivity).toHaveBeenCalledWith(key, 'control:production-gce-c3:1')
+  })
+
+  // Two hellos proven against the same empty map could take the same generation; an activation at
+  // a generation this host already had here is refused, so a late leave cannot share it.
+  it('refuses a fresh activation at a generation this host already had here', async () => {
+    const activateControl = vi
+      .fn<RelayAssignmentStore['activateControl']>()
+      .mockResolvedValue('control:production-gce-c3:2')
+    const { registry, activate } = createRegistry(activateControl)
+    const key = { userId: identity.sub, relayHostId: identity.relayHostId }
+    const first = new FakeSocket()
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface.
+    await activate(first as unknown as WebSocket, identity, null, 2, false, 1)
+    first.close(1006, 'network reset')
+    await vi.advanceTimersByTimeAsync(CONTROL_CONTINUITY_LIMITS.orphanGraceMs)
+    expect(registry.get(key)).toBeNull()
+    const reused = new FakeSocket()
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface.
+    await activate(reused as unknown as WebSocket, identity, null, 2, false, 1)
+    expect(reused.close).toHaveBeenCalledWith(RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded')
+    expect(registry.get(key)).toBeNull()
+  })
+
   it('skips a closed queued control so its live retry avoids abandoned database work', async () => {
     const stalled = deferred<string>()
     const activateControl = vi

cloud/apps/relay/src/host-session-registry.ts

@@ -35,7 +35,8 @@ import type { AssignmentLeaseShadow } from './assignment/assignment-lease-shadow
 import type { RelayAssignmentStore } from './assignment-store.js'
 import type { CellAdmitMode, CellReserveBook } from './cell-reserve-book.js'
 import type { DemoteRequest, ReserveOutcome, ReserveRequest } from './cell-reserve-contract.js'
-import { CellSeatLog, type CellSeatChange, type CellSeatFeedPage } from './cell-seat-log.js'
+import { SeatDriftConfirmation, unbackedSeatReason, type UnbackedSeatReason } from './cell-seat-drift.js'
+import { CellSeatLog, type CellSeat, type CellSeatChange, type CellSeatFeedPage } from './cell-seat-log.js'
 import { isRelayDatabaseTransientError } from './database.js'
 import { printableCloseReason } from './control-close-reason-text.js'
 import { ControlRenewalBatch } from './control-renewal-batch.js'
@@ -296,6 +297,9 @@ const NO_REJOIN_CLOSE_CODES: ReadonlySet<number> = new Set([
 ])
 // A demoted duplicate keeps its splices this long, then closes so the desktop re-assigns.
 export const DEMOTION_CLOSE_MS = 60_000
+// Hosts whose highest generation is remembered; insertion order is age order. A host evicted
+// here is long gone, so its successor restarting low meets no late leave.
+const GENERATION_FLOOR_MAX_HOSTS = 100_000
 // A flip back registers memory-admitted controls with at most this share of the pool in
 // flight (US 10 -> 3, Asia 16 -> 5), so DB-path hellos and renewals keep the rest.
 export const REREGISTER_POOL_SHARE = 1 / 3
@@ -334,6 +338,9 @@ export class HostSessionRegistry {
   // Seat key -> the demoted control socket. Its own set, not drainSentHosts: it ends with
   // that socket (close, rebind or a new join), so a later booking here is a fresh seat.
   private readonly demotedSeats = new Map<string, WebSocket>()
+  private readonly seatDrift = new SeatDriftConfirmation()
+  // The highest generation each host has had on this cell, kept after its session leaves the map.
+  private readonly generationFloors = new Map<string, number>()
   private reregistration: {
     queue: ReregistrationEntry[]
     // Sessions already queued, given up on, or done in this pass: the rescan skips them.
@@ -776,6 +783,13 @@ export class HostSessionRegistry {
       session.assignmentEpoch !== request.epoch ||
       seat?.joinedAt !== request.joinedAt
     ) {
+      // The director named exactly the seat this feed lists: if no live session backs it, the
+      // feed is what is wrong, and every director mirrors it, so correct the feed now.
+      const unbacked =
+        seat?.epoch === request.epoch && seat.joinedAt === request.joinedAt
+          ? unbackedSeatReason(seat, session)
+          : null
+      if (seat && unbacked) this.repairSeatDrift(seat, session, unbacked, 'demote')
       return 'not-seated'
     }
     // A drain already told this host to go: its own grace stands, never cut to the demotion's.
@@ -1682,7 +1696,11 @@ export class HostSessionRegistry {
       (existing.state === 'orphaned' || existing.state === 'active') &&
       sameHostIdentity(existing.identity, identity)
     )
-    const generation = rebind ? existing!.generation : (existing?.generation ?? 0) + 1
+    // Above every generation this host has had here, not only the one in the map: a session
+    // dropped from the map while its socket is still closing (drain, regional deadline, idle
+    // rehome) leaves later, and a successor reusing its generation had its seat removed by that leave.
+    const floor = this.generationFloors.get(key) ?? 0
+    const generation = rebind ? existing!.generation : Math.max(existing?.generation ?? 0, floor) + 1
     const ephemeral = nacl.box.keyPair()
     const challengeNonce = randomBytes(nacl.box.nonceLength)
     const challengeSecret = randomBytes(32)
@@ -1912,6 +1930,24 @@ export class HostSessionRegistry {
       }
       return
     }
+    // The awaits above can outlive what this hello was proven against: an orphan grace expiring
+    // or a drain removes `existing` from the map, and another activation may take this generation.
+    // Rebinding a removed session revived it outside the map: connected, still in the feed, and
+    // unreachable by phones or demotion for as long as the host stayed up. Redial instead.
+    const key = this.key(identity.sub, identity.relayHostId)
+    if (
+      (this.sessions.get(key) ?? null) !== existing ||
+      (!rebind && generation <= (this.generationFloors.get(key) ?? 0))
+    ) {
+      if (controlActivityId) {
+        this.releaseActivityBestEffort(
+          { userId: identity.sub, relayHostId: identity.relayHostId },
+          controlActivityId
+        )
+      }
+      this.refuseProvenControl(socket, refused, RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded')
+      return
+    }
     if (existing) this.observer.recordReconnect()
     if (rebind && existing) {
       const previousSocket = existing.socket
@@ -2005,6 +2041,7 @@ export class HostSessionRegistry {
     // A host that proved itself again is not signed out, whatever it said last.
     this.hostCloseReasons.forget(sessionKey)
     this.sessions.set(sessionKey, session)
+    this.recordGenerationFloor(sessionKey, generation)
     this.demotedSeats.delete(sessionKey)
     this.wireActiveControl(session)
     this.sendHelloAck(session)
@@ -2028,9 +2065,87 @@ export class HostSessionRegistry {
   }
 
   seatFeed(sinceSeq: number | null): CellSeatFeedPage {
+    this.sweepSeatDrift()
     return { ...this.seatLog.read(sinceSeq), seats: this.seatLog.seatCount() }
   }
 
+  // Every director mirrors this feed, so it must list exactly the live sessions. Swept on a
+  // director's poll, since only directors read it; drift seen twice in a row is repaired.
+  private sweepSeatDrift(): void {
+    if (!this.seatDrift.due(this.now())) return
+    const found = new Map<string, () => void>()
+    for (const seat of this.seatLog.seatList()) {
+      const session = this.get(seat)
+      const reason = unbackedSeatReason(seat, session)
+      if (!reason) continue
+      const id = JSON.stringify(['seat', seat.userId, seat.relayHostId, seat.epoch, seat.generation, seat.joinedAt])
+      found.set(id, () => this.repairSeatDrift(seat, session, reason, 'sweep'))
+    }
+    for (const session of this.sessions.values()) {
+      if (!liveSeatSession(session) || this.seatLog.seatOf(session.identity.sub, session.relayHostId)) continue
+      const id = JSON.stringify(['session', session.identity.sub, session.relayHostId, session.generation])
+      found.set(id, () => {
+        this.appendSeatChange(session, 'join')
+        this.logSeatDrift(session.relayHostId, { kind: 'unlisted-session', via: 'sweep', sessionEpoch: session.assignmentEpoch })
+      })
+    }
+    for (const id of this.seatDrift.confirm([...found.keys()])) found.get(id)!()
+  }
+
+  // Leaves the stale seat as moved (no rejoin from memory at its epoch), then lists the live
+  // session, if any, as it really is. The leave carries the stale seat's own generation, so the
+  // generation rule cannot drop it.
+  private repairSeatDrift(
+    seat: CellSeat,
+    session: HostSession | null,
+    reason: UnbackedSeatReason,
+    via: 'sweep' | 'demote'
+  ): void {
+    this.seatLog.append({
+      kind: 'leave',
+      userId: seat.userId,
+      relayHostId: seat.relayHostId,
+      epoch: seat.epoch,
+      generation: seat.generation,
+      closeCode: RELAY_CLOSE_CODE.WRONG_CELL,
+      at: this.now()
+    })
+    const relisted = session !== null && liveSeatSession(session)
+    if (relisted) this.appendSeatChange(session, 'join')
+    this.logSeatDrift(seat.relayHostId, {
+      kind: 'unbacked-seat',
+      via,
+      reason,
+      seatEpoch: seat.epoch,
+      seatGeneration: seat.generation,
+      seatAgeMs: this.now() - seat.joinedAt,
+      ...(session
+        ? { sessionEpoch: session.assignmentEpoch, sessionGeneration: session.generation, sessionState: session.state }
+        : {}),
+      relisted
+    })
+  }
+
+  private recordGenerationFloor(key: string, generation: number): void {
+    this.generationFloors.delete(key)
+    this.generationFloors.set(key, generation)
+    if (this.generationFloors.size > GENERATION_FLOOR_MAX_HOSTS) {
+      const oldest = this.generationFloors.keys().next().value
+      if (oldest !== undefined) this.generationFloors.delete(oldest)
+    }
+  }
+
+  private logSeatDrift(relayHostId: string, detail: Record<string, unknown>): void {
+    console.warn(
+      JSON.stringify({
+        event: 'orca_relay_cell_seat_log_drift',
+        ...this.logIdentity(),
+        relayHostIdDigest: relayHostLogDigest(relayHostId),
+        ...detail
+      })
+    )
+  }
+
   private appendSeatChange(
     session: HostSession,
     kind: CellSeatChange['kind'],
@@ -2719,3 +2834,8 @@ function controlErrorCode(type: string, error: unknown): string {
 function databaseCause(error: unknown): PhoneDatabaseCause {
   return isRelayDatabaseTransientError(error) ? 'database_transient' : 'error'
 }
+
+// A session the feed must list: it holds its socket and is seated, active or draining.
+function liveSeatSession(session: HostSession): boolean {
+  return session.socket !== null && (session.state === 'active' || session.state === 'drain-only')
+}

cloud/apps/relay/src/relay-server.ts

@@ -275,7 +275,7 @@ export function createRelayServer(
               }))
             ),
         demote: (seat, row) => {
-          reservePlacement?.demoteSeat(seat)
+          reservePlacement?.demoteSeat(seat, 'reconcile')
           // The re-assign mints above the row, so the mirror can take it.
           reservePlacement?.raiseEpochFloor(seat, row.epoch)
           console.warn(

cloud/apps/relay/src/relay-websocket-close.ts

@@ -1,6 +1,6 @@
 import type WebSocket from 'ws'
 
-const RELAY_WEBSOCKET_FORCE_CLOSE_MS = 1_000
+export const RELAY_WEBSOCKET_FORCE_CLOSE_MS = 1_000
 const forceCloseTimers = new WeakMap<WebSocket, ReturnType<typeof setTimeout>>()
 
 // Delivering the rejection is best effort, and deliberately so. The close frame carrying the

cloud/apps/relay/src/reserve-assignment.test.ts

@@ -54,6 +54,7 @@ function setup(
     mode?: 'off' | 'dry-run' | 'on'
     startedAt?: number
     log?: (line: string) => void
+    demote?: (cellId: string) => string | null
   } = {}
 ) {
   const now = { value: 100_000 }
@@ -72,7 +73,7 @@ function setup(
     }),
     demote: vi.fn(async (target: { cellId: string }, request: { epoch: number; joinedAt: number }) => {
       demoted.push({ cellId: target.cellId, epoch: request.epoch, joinedAt: request.joinedAt })
-      return true
+      return options.demote ? options.demote(target.cellId) : 'demoted'
     })
   } satisfies Pick<CellReserveClient, 'reserve' | 'demote'>
   // Each director's estimate starts empty; a second of refill gives it its share.
@@ -503,4 +504,42 @@ describe('reserve assignment on a director', () => {
       { outcomes: { ok: 2 }, cells: { c1: { outcomes: { ok: 2 }, fullWithRoom: 0, minFullFreeShare: null } } }
     ])
   })
+
+  describe('when a cell answers a demotion', () => {
+    const settle = () => new Promise((resolve) => setTimeout(resolve, 0))
+    const outcomes = (lines: string[]) =>
+      lines
+        .map((line) => JSON.parse(line))
+        .filter((entry) => entry.event === 'orca_relay_reserve_demote_outcome')
+
+    // The map follows the cell's feed: a refusing cell repairs that feed itself.
+    it('logs the answer by host digest and leaves the seat to the feed', async () => {
+      const lines: string[] = []
+      const { assignment, directory } = setup({
+        feeds: [feed('c1', { full: [seated(10, 1_000)] }), feed('c2')],
+        demote: () => 'not-seated',
+        log: (line) => lines.push(line)
+      })
+      assignment.demoteSeat({ ...HOST, cellId: 'c1', epoch: 10, joinedAt: 1_000 }, 'reconcile')
+      await settle()
+      expect(outcomes(lines)).toMatchObject([
+        { via: 'reconcile', cellId: 'c1', epoch: 10, outcome: 'not-seated' }
+      ])
+      expect(outcomes(lines)[0].relayHostIdDigest).toMatch(/^[0-9a-f]{12}$/)
+      expect(lines.join('\n')).not.toContain(HOST.relayHostId)
+      expect(directory.seatsOnCell('c1')).toHaveLength(1)
+    })
+
+    it('records a cell it could not reach as unreachable', async () => {
+      const lines: string[] = []
+      const { assignment } = setup({
+        feeds: [feed('c1', { full: [seated(10, 1_000)] }), feed('c2')],
+        demote: () => null,
+        log: (line) => lines.push(line)
+      })
+      assignment.demoteSeat({ ...HOST, cellId: 'c1', epoch: 10, joinedAt: 1_000 }, 'reconcile')
+      await settle()
+      expect(outcomes(lines)).toMatchObject([{ outcome: 'unreachable' }])
+    })
+  })
 })

cloud/apps/relay/src/reserve-assignment.ts

@@ -422,19 +422,36 @@ export class ReserveAssignment {
   }
 
   // Fire-and-forget; the cell checks the seat's epoch and join itself. False: no such cell listed.
-  demoteSeat(seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }): boolean {
+  demoteSeat(seat: DemotedSeat, via: DemoteVia): boolean {
     const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId)
     if (!cell) return false
-    void this.input.client.demote(cell, {
-      v: 1,
-      userId: seat.userId,
-      relayHostId: seat.relayHostId,
-      epoch: seat.epoch,
-      joinedAt: seat.joinedAt
-    })
+    void this.input.client
+      .demote(cell, {
+        v: 1,
+        userId: seat.userId,
+        relayHostId: seat.relayHostId,
+        epoch: seat.epoch,
+        joinedAt: seat.joinedAt
+      })
+      .then((outcome) => this.settleDemote(seat, via, outcome))
+      .catch(() => console.warn('[orca-relay] reserve demote outcome not recorded'))
     return true
   }
 
+  // The cell repairs its own feed when it refuses a seat it still lists; this only records the answer.
+  private settleDemote(seat: DemotedSeat, via: DemoteVia, outcome: string | null): void {
+    this.log(
+      JSON.stringify({
+        event: 'orca_relay_reserve_demote_outcome',
+        via,
+        cellId: seat.cellId,
+        epoch: seat.epoch,
+        relayHostIdDigest: relayHostLogDigest(seat.relayHostId),
+        outcome: outcome ?? 'unreachable'
+      })
+    )
+  }
+
   // Supersede at placement: the old seats this director's map holds on reserve cells are told
   // to go. Each cell checks the seat's epoch and join itself, so a stale map only misses.
   private supersede(identity: Identity, seats: readonly ShadowSeat[], cellId: string, epoch: number): void {
@@ -456,9 +473,9 @@ export class ReserveAssignment {
     toCellId: string,
     via: 'placement' | 'feed'
   ): void {
-    if (!this.demoteSeat(seat)) return
+    if (!this.demoteSeat(seat, via)) return
     this.summary.supersedes += 1
-    this.input.log?.(
+    this.log(
       JSON.stringify({
         event: 'orca_relay_reserve_supersede',
         relayHostIdDigest: relayHostLogDigest(seat.relayHostId),
@@ -479,7 +496,7 @@ export class ReserveAssignment {
       latenciesMs.length === 0
         ? null
         : Math.round(latenciesMs[Math.min(latenciesMs.length - 1, Math.floor(share * latenciesMs.length))]!)
-    ;(this.input.log ?? ((line: string) => console.log(line)))(
+    this.log(
       JSON.stringify({
         event: 'orca_relay_reserve_placement_summary',
         mode: this.input.mode,
@@ -489,8 +506,15 @@ export class ReserveAssignment {
     )
     this.summary = emptySummary()
   }
+
+  private log(line: string): void {
+    ;(this.input.log ?? ((entry: string) => console.log(entry)))(line)
+  }
 }
 
+type DemotedSeat = { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }
+type DemoteVia = 'reconcile' | 'placement' | 'feed'
+
 type PlacementSummary = {
   sticky: number
   stickyReserves: number