openharness

import { AutonomousDeviceRelay } from './lib/autonomous-device/relay.js'
import type { AutonomousDeviceService, AutonomousDeviceFrame } from './lib/autonomous-device/service.js'
/**
 * BackendSocket — the CLI's dial-out connection to the backend's `/api/adapter-ws`.
 *
 * The adapter occupies the NODE side of its agentId on the backend hub:
 *   - `up`   { t:'up', frame }            → normalized claude events + `<x>_result` RPC replies
 *   - `down` { t:'down', connId, frame }  → web chat/control + data-plane RPC requests
 *
 * Mirrors the hosted runtime’s managerSocket: idempotent connect, exponential backoff (1s→30s),
 * WS liveness (`lib/wsLiveness.ts`: ping every 20s, 60s deadline on silence) + a 15s app-level
 * `{t:'ping'}` that refreshes the backend presence key, and a bounded FIFO queue for client-facing
 * outbound frames.
 *
 * Auth: the SSO access token rides as the first WS subprotocol.
 */

import { WebSocket } from 'ws'
import { watchSocketLiveness, type LivenessWatch } from './lib/wsLiveness.js'
import { stat, readFile } from 'fs/promises'
import { isAbsolute, join } from 'path'
import { hostname } from 'os'
import { env } from './config/env.js'
import { AuthSessionManager, AuthSessionError } from './lib/authSession.js'
import { VERSION } from './version.js'
import { registry, projectDisplayName, type RegisteredSession } from './lib/registry.js'
import { ENGINES, type AgentEngine } from './engines/types.js'
import { listDir } from './lib/fsBrowse.js'
import { linkCodexProfile, listCodexProfiles } from './lib/codexProfiles.js'
import { parseGridLaunchOverride, type GridLaunchOverride } from './lib/gridLaunch.js'
import { readAccountUsage, type AccountUsageReading } from './lib/accountUsage.js'
import { probeEngines } from './lib/engineProbe.js'
import { AgentCreationReceipts, AgentCreationReceiptError, creationFingerprint, validCreationId, type AgentCreationStatus } from './lib/agentCreationReceipt.js'
import { engineInstallRecipe } from './lib/engineInstall.js'
import { agentFrame, type AgentFrame } from './lib/agentFrame.js'
import { routeVoiceTask } from './lib/voiceRouter.js'
import { tailFile } from './lib/sessions.js'
import { messagesToEvents, windowRawLines, subagentStatsFromRawLines, type SessionEvent } from './lib/normalize.js'
import { listFileTree, readProjectFile } from './lib/files.js'
import { MediaPreviewError, readMediaPreviewChunk } from './lib/mediaPreview.js'
import { codexMessagesToEvents, windowCodexLines } from './engines/codex/normalizer.js'
import { codexSubagentResolverFor } from './engines/codex/subagent.js'
import { cursorMessagesToEvents, windowCursorLines } from './engines/cursor/normalizer.js'
import { loadCursorReplayTaskLinks } from './engines/cursor/subagent.js'
import { opencodeMessagesToEvents, windowOpencodeMessages } from './engines/opencode/normalizer.js'
import { kiloMessagesToEvents, windowKiloMessages } from './engines/kilo/normalizer.js'
import { museMessagesToEvents } from './engines/muse/normalizer.js'
import { ampMessagesToEvents } from './engines/amp/normalizer.js'
import { grokMessagesToEvents } from './engines/grok/normalizer.js'
import { agyMessagesToEvents } from './engines/agy/normalizer.js'
import { copilotMessagesToEvents } from './engines/copilot/normalizer.js'
import { ampThreadToEvents, readAmpThread } from './engines/amp/threadExport.js'
import { piMessagesToEvents, windowPiLines } from './engines/pi/normalizer.js'
import { commandcodeMessagesToEvents, windowCommandCodeLines } from './engines/commandcode/normalizer.js'
import { hermesMessagesToEvents, windowHermesMessages } from './engines/hermes/normalizer.js'
import { devinMessagesToEvents, windowDevinMessages } from './engines/devin/normalizer.js'
import { readHermesMessages } from './engines/hermes/reader.js'
import { readDevinMessages } from './engines/devin/reader.js'
import { readOpencodeMessages } from './engines/opencode/reader.js'
import { readKiloMessages } from './engines/kilo/reader.js'
import { E2eeManager, type PairResult } from './lib/e2ee/manager.js'
import type { TerminalStreamManager } from './lib/terminalStreamManager.js'
import {
  decodeTerminalHop,
  encodeTerminalLocal,
  encodeTerminalHop,
  TerminalHopDirection,
  type TerminalBinaryClear,
} from './lib/terminalBinary.js'
import { b64d, fingerprint, ENCRYPTED_RPC_RESULT_TYPES, isEncryptedDownType, isWrapped } from './lib/e2ee/core.js'
import { DEVICE_RECENT_SAFE_FRAME_BYTES, fitRecentReplyPayloadForDevice } from './lib/deviceRecentTrim.js'
import { shouldReplayCommander } from './lib/commanderReplay.js'
import { RuntimeProfileControlError, type RuntimeProfileErrorCode } from './lib/runtimeProfileController.js'
import { parseRuntimeProfile, type RuntimeModelOption } from './lib/runtimeProfile.js'
import { sid, preview, logFrame } from './lib/log.js'
import {
  TerminalP2pResponderPool,
  TERMINAL_P2P_DOWN_TYPES,
  TERMINAL_P2P_SIGNAL_TYPES,
  type TerminalP2pData,
  type TerminalP2pSignal,
} from './lib/terminalP2p.js'

// OpenCode has no per-session transcript file — its history is read from this SQLite store.
const OPENCODE_DB = join(env.OPENCODE_DATA_DIR, 'opencode.db')
// Kilo keeps history the same way opencode does, in its own store.
const KILO_DB = join(env.KILO_DATA_DIR, 'kilo.db')
const DEVIN_DB = join(env.DEVIN_HOME, 'sessions.db')
// Hermes history likewise comes from a SQLite store, not a per-session file.
const HERMES_DB = join(env.HERMES_HOME, 'state.db')

/** Answers the device `project_recent` RPC — set by cli.ts to the CommanderMirror's `recent`. */
export type RecentProvider = (sessionId: string, n: number) => Array<{ kind: string; text: string; recap?: string }>


const APP_PING_MS = 15_000
// How long the opening handshake may take before the attempt is abandoned and retried. `ws` waits
// forever by default, and the heartbeat below only starts on 'open' — so a TCP connection that came
// up while the network was flapping but never got its upgrade answered sat in CONNECTING for hours,
// `this.ws` set, every later connect() returning early, and the daemon reporting "cloud
// reconnecting…" until someone restarted it.
const HANDSHAKE_TIMEOUT_MS = 15_000
const BASE_DELAY_MS = 1_000
const MAX_DELAY_MS = 30_000
const QUEUE_MAX = 2_000
const DEVICE_AGENT_LIST_LIMIT = 100
const DEVICE_AGENT_NAME_MAX_CODEPOINTS = 15
const DEVICE_AGENT_NAME_MAX_BYTES = 39 // device project_t.name[40], including trailing NUL on-device.
const DEVICE_ELLIPSIS = '…'

export type Frame = Record<string, unknown>
type OutboundEnvelope = Record<string, unknown>

export interface LocalClientSink {
  sendFrame: (frame: Frame) => boolean
  sendBinary: (frame: Uint8Array) => boolean
}

function isLocalClientId(connId: string): boolean {
  return connId.startsWith('local:')
}

interface QueueItem {
  id: number
  data: string
  msg: OutboundEnvelope
  attempts: number
}

interface DownEnvelope {
  t: 'down'
  connId?: string
  frame?: Frame
}

function byteLen(s: string): number {
  return Buffer.byteLength(s, 'utf8')
}

function clipDeviceAgentName(input: string): string {
  const chars = [...input]
  if (chars.length <= DEVICE_AGENT_NAME_MAX_CODEPOINTS && byteLen(input) <= DEVICE_AGENT_NAME_MAX_BYTES) return input
  const ellipsisBytes = byteLen(DEVICE_ELLIPSIS)
  let out = ''
  for (const ch of chars.slice(0, DEVICE_AGENT_NAME_MAX_CODEPOINTS)) {
    if (byteLen(out) + byteLen(ch) + ellipsisBytes > DEVICE_AGENT_NAME_MAX_BYTES) break
    out += ch
  }
  return `${out}${DEVICE_ELLIPSIS}`
}

/**
 * Amp history comes from AMP's store, not from ours.
 *
 * The plugin's JSONL is a record of what the plugin saw; a thread that ran before the integration existed
 * — or in a pane still holding an older plugin — is simply not in it, and no local file can rebuild those
 * turns. `amp threads export` returns the thread complete, including the tools Amp runs server-side.
 *
 * The local file stays as the FALLBACK, because an export is a network call and a pane with no history at
 * all is worse than a partial one. Which source answered is logged either way: silently serving the lesser
 * record is exactly how the missing tool cards went unnoticed for a day.
 */
async function ampHistory(sessionId: string, lines: string[]): Promise<SessionEvent[]> {
  const messages = await readAmpThread(sessionId)
  if (messages) {
    console.log(`[history] ${sessionId.slice(0, 12)} amp · ${messages.length} message(s) from amp's own store`)
    return ampThreadToEvents(messages)
  }
  console.warn(`[history] ${sessionId.slice(0, 12)} amp · export unavailable — falling back to the local transcript`)
  return ampMessagesToEvents(lines)
}

/** Grok has no transcript windower yet. Keep BOTH `session_get` shapes on the same real-record replay:
 * web always sends a limit, while legacy callers omit it. Returning the whole small transcript for a
 * page is honest (`hasMore:false`) and cannot fall through to Claude's incompatible line cursor. */
export function grokHistoryPage(lines: string[], paginated: boolean):
  { events: SessionEvent[]; hasMore?: false; oldestCursor?: null } {
  const events = grokMessagesToEvents(lines)
  return paginated ? { events, hasMore: false, oldestCursor: null } : { events }
}

/** agy has no transcript windower either; same both-shapes replay as grok, for the same reason. */
export function agyHistoryPage(lines: string[], paginated: boolean):
  { events: SessionEvent[]; hasMore?: false; oldestCursor?: null } {
  const events = agyMessagesToEvents(lines)
  return paginated ? { events, hasMore: false, oldestCursor: null } : { events }
}

/** Copilot has no transcript windower either; same both-shapes replay as grok and agy. */
export function copilotHistoryPage(lines: string[], paginated: boolean):
  { events: SessionEvent[]; hasMore?: false; oldestCursor?: null } {
  const events = copilotMessagesToEvents(lines)
  return paginated ? { events, hasMore: false, oldestCursor: null } : { events }
}

export function deviceAgentListItem(
  raw: unknown,
): { id: unknown; name?: string; engine?: 'claude' | 'codex' | 'cursor' | 'opencode' | 'pi' | 'hermes' | 'commandcode' | 'devin' | 'muse' | 'amp' | 'kilo' | 'grok' | 'agy' | 'copilot'; selectedModel?: string | null } {
  const o = (raw ?? {}) as Record<string, unknown>
  const item: { id: unknown; name?: string; engine?: 'claude' | 'codex' | 'cursor' | 'opencode' | 'pi' | 'hermes' | 'commandcode' | 'devin' | 'muse' | 'amp' | 'kilo' | 'grok' | 'agy' | 'copilot'; selectedModel?: string | null } = { id: o.id }
  if (typeof o.name === 'string') item.name = clipDeviceAgentName(o.name)
  if (o.engine === 'claude' || o.engine === 'codex' || o.engine === 'cursor' || o.engine === 'opencode' || o.engine === 'pi' || o.engine === 'hermes' || o.engine === 'commandcode' || o.engine === 'devin' || o.engine === 'muse' || o.engine === 'amp' || o.engine === 'kilo' || o.engine === 'grok' || o.engine === 'agy' || o.engine === 'copilot') item.engine = o.engine
  // Runtime model/effort profile (opaque runtime-v1:...) — lets the device render + change model/effort.
  if (typeof o.selectedModel === 'string' || o.selectedModel === null) item.selectedModel = o.selectedModel
  return item
}

/**
 * How many models the DEVICE picker may receive. Matches its own PICK_MAX (ui_screens.c) so the wheel
 * never renders more rows than it was built for; the web picker is unbounded and still gets everything.
 */
const DEVICE_PICKER_MAX_MODELS = 24

export function compactRuntimePickerModels(
  models: RuntimeModelOption[],
  sessionId: string | undefined,
  pickerMode: unknown,
  selectedModel: unknown,
): Array<{ id: string }> {
  const compact = models.map(({ id }) => ({ id }))
  if ((pickerMode !== 'model' && pickerMode !== 'effort') || !sessionId) return compact

  const profiles = models.flatMap((item) => {
    const profile = parseRuntimeProfile(item.id)
    return profile?.sessionId === sessionId ? [{ item, profile }] : []
  })
  const selected = parseRuntimeProfile(selectedModel)
  const current = selected?.sessionId === sessionId ? selected : null

  if (pickerMode === 'effort') {
    if (!current) return []
    const seen = new Set<string>()
    return profiles.flatMap(({ item, profile }) => {
      if (profile.model !== current.model || profile.effort === 'auto' || seen.has(profile.effort)) return []
      seen.add(profile.effort)
      return [{ id: item.id }]
    })
  }

  const byModel = new Map<string, typeof profiles>()
  for (const entry of profiles) {
    const group = byModel.get(entry.profile.model) ?? []
    group.push(entry)
    byModel.set(entry.profile.model, group)
  }
  const rows = [...byModel.values()].map((group) => {
    const target = group.find(({ profile }) => current && profile.effort === current.effort)
      ?? group.find(({ profile }) => profile.effort === 'auto')
      ?? group[0]
    return { id: target.item.id, model: target.profile.model }
  })
  // Top N only. The picker is a scroll wheel on a 1.9" round screen, and a 49-row one was enough to stall
  // the device's the device UI task into a task-watchdog reset; devin alone publishes 72
  // models. The catalog arrives in the engine's own order — its curated/most-used first — so "top" is that
  // order, with the model the agent is RUNNING pinned in front so the list can never hide it.
  const ordered = current
    ? [...rows].sort((a, b) => Number(b.model === current.model) - Number(a.model === current.model))
    : rows
  return ordered.slice(0, DEVICE_PICKER_MAX_MODELS).map(({ id }) => ({ id }))
}

/** Fill in missing sub-agent aggregates on tool_end events by reading the sub-agent's own transcript
 *  (`<session>/subagents/agent-<id>.jsonl`). Async/background launchers only record
 *  `{status:'async_launched', agentId}` in the main transcript — without this join the delegation
 *  card shows "0 tools · worked for 0s" forever. Best-effort per agent; missing files are skipped. */
async function enrichSubagentStats(events: SessionEvent[], transcriptPath: string): Promise<void> {
  const subagentsDir = join(transcriptPath.replace(/\.jsonl$/, ''), 'subagents')
  for (const e of events) {
    if (e.type !== 'tool_end') continue
    const sub = e.payload.subagent
    if (!sub?.agentId || typeof sub.totalToolUseCount === 'number') continue
    try {
      const txt = await readFile(join(subagentsDir, `agent-${sub.agentId}.jsonl`), 'utf8')
      const stats = subagentStatsFromRawLines(txt.split('\n'))
      sub.totalToolUseCount = stats.totalToolUseCount
      if (sub.totalDurationMs === undefined) sub.totalDurationMs = stats.totalDurationMs
      if (sub.totalTokens === undefined) sub.totalTokens = stats.totalTokens
    } catch { /* subagent transcript absent (still spawning / pruned) — leave as-is */ }
  }
}

export class BackendSocket {
  private ws: WebSocket | null = null
  private connecting = false
  private retryingAuth = false
  private readonly auth: AuthSessionManager
  /** Constructor-without-auth is retained for isolated unit tests only. */
  private readonly testToken?: string
  private url: string
  private attempts = 0
  private closed = false
  private queue: QueueItem[] = []
  private draining = false
  private nextQueueId = 1
  private droppedSinceLog = 0
  private heartbeat: LivenessWatch | null = null
  private appPing: NodeJS.Timeout | null = null
  private readonly downChains = new Map<string, Promise<void>>()
  private readonly localClients = new Map<string, LocalClientSink>()
  private terminalStreams: TerminalStreamManager | null = null
  private readonly terminalP2p: TerminalP2pResponderPool
  private readonly p2pPendingOpens = new Map<string, Set<string>>()
  private readonly p2pStreams = new Map<string, Set<string>>()
  private onStatus: (connected: boolean) => void
  /** Cross-instance commander (device) client count, from backend `__clients` frames. */
  private commanderCount = 0
  /** Subset of commanderCount whose device is ACTIVELY rendering this machine (multi-attach). null = the
   *  backend doesn't send the signal (old build) → fall back to hasCommander so streaming isn't gated off. */
  private commanderActive: number | null = null
  private replayedCommanderGeneration: number | undefined
  private replayCommanderOnNextSnapshot = true
  /** Called when a commander attach is observed — cli.ts replays live state. */
  onCommanderJoin: (() => void) | null = null
  /** Called only when commander presence crosses zero; drives the disposable recap-worker grace. */
  onCommanderPresenceChanged: ((connected: boolean) => void) | null = null
  /** Called when the web cancels a turn (C-c) — cli.ts stops that session's turn heartbeat. */
  onCancel: ((sessionId: string) => void) | null = null
  /** Called when the web deletes an agent (`agent_delete`) — cli.ts signals only the validated engine
   *  process and forgets the session. Keeps recap + agent name. */
  onDeleteAgent: ((sessionId: string) => void) | null = null
  /** Called on `agent_create` — cli.ts spawns a fresh tmux session running the requested engine in the
   *  requested folder and returns its process-agent. Session metadata may bind later through hooks. */
  onCreateAgent: ((input: {
    engine: AgentEngine
    cwd: string
    bypassPermission: boolean
    /** A grid to point this agent at instead of the engine's own login; null when none was chosen. */
    grid: GridLaunchOverride | null
    /** A Codex CODEX_HOME folder to launch this agent against instead of `~/.codex`; codex only. */
    codexHome: string | null
  }) =>
    Promise<{ ok: true; session: RegisteredSession } | { ok: false; error: string; detail?: string }>) | null = null
  private readonly agentCreations = new AgentCreationReceipts(join(env.ADAPTER_DATA_DIR, 'agent-creations'))
  /** Injectable for queue-isolation tests; production uses the machine-local probe. */
  engineProbeProvider: typeof probeEngines = probeEngines
  /**
   * Called on `agent_retarget` — cli.ts re-execs an EXISTING agent's pane against a different grid,
   * or, when `grid` is null, back onto its own login.
   *
   * Separate from `onCreateAgent` because it is a different promise: the pane, its id and its
   * scrollback survive, and only the process is replaced. A running process's environment cannot be
   * edited, so there is no gentler way to move an agent that is already up.
   *
   * Separate from `onRestartAgent` because that one puts the agent back exactly as it was; this one
   * puts it back somewhere else. They share the swap underneath and differ in what they hand it.
   */
  onRetargetAgent: ((input: { agentId: string; grid: GridLaunchOverride | null }) =>
    Promise<{ ok: true } | { ok: false; error: string; detail?: string }>) | null = null
  /** Called on `agent_restart` — cli.ts stops the agent's live engine process and relaunches it in the
   *  SAME tmux pane, keeping the SAME agentId and (best-effort) resuming the same engine session.
   *  `resumed` tells the caller whether the relaunch actually resumed the prior conversation or had to
   *  fall back to a fresh one under the same agent/pane. */
  onRestartAgent: ((agentId: string) =>
    Promise<
      { ok: true; session: RegisteredSession; resumed: boolean }
      | { ok: false; error: string; detail?: string }
    >) | null = null
  /** Called when the web/device sends chat input to an agent terminal. */
  onMessage: ((sessionId: string, content: string) => void) | null = null
  /** Best-effort terminal-native title sync after a user renames an agent. */
  onAgentRename: ((session: RegisteredSession, name: string) => void) | null = null
  /** Called when a device answers an AskUserQuestion (`question_response`) — cli.ts drives the CLI's own
   *  terminal dialog (option digit / free text), since a remote machine has no
   *  programmatic answer channel the way the hosted runtime’s brain does. */
  onQuestionAnswer: ((payload: { requestId?: string; sessionId?: string; agentId?: string; answers?: Record<string, string> }) => void) | null = null
  /** Called when this machine was deleted/revoked (a `machine_revoked` down-frame, or a 401/403 on the
   *  upgrade) — CLI clears the saved SSO session and shuts down instead of retrying forever. */
  onRevoked: (() => void) | null = null
  /** Called with the machine's display name (`machine_meta` down-frame: seeded on connect, pushed on a
   *  web rename; null = unnamed) — cli mirrors it to a local file for `harness status`. */
  onMachineMeta: ((name: string | null) => void) | null = null
  /** Called when this machine is already connected from ANOTHER computer (HTTP 409 on the upgrade) — the
   *  SSO session is valid, so CLI keeps it and stops without a retry loop. */
  onBusy: (() => void) | null = null
  /** Answers the device `project_recent` RPC (cli.ts wires this to CommanderMirror.recent). */
  recentProvider: RecentProvider | null = null
  /** The person's own last questions for an agent, newest first. See the `agent_recent` case. */
  recentAsksProvider: ((agentId: string, n: number) => string[]) | null = null
  /** Runtime Model/Effort integration, wired by cli.ts for registered tmux sessions. */
  runtimeModelsProvider: ((sessionId?: string) => Promise<RuntimeModelOption[]>) | null = null
  /** Answers `usage_read` — this machine's own agent-account usage (lib/accountUsage.ts). A field
   *  rather than a direct call so a spec answers it without a real home, Keychain or network. */
  accountUsageReader: () => Promise<AccountUsageReading[]> = readAccountUsage
  runtimeProfileProvider: ((session: RegisteredSession) => string | null) | null = null
  onRuntimeProfileUpdate: ((sessionId: string, selectedModel: string) => Promise<void>) | null = null
  /** Web↔adapter E2EE: group-encrypts user events, runs the CPace pairing, holds per-conn sessions. */
  readonly e2ee: E2eeManager
  /** Backend-resolved machine id, persisted by the SSO login preflight. */
  readonly machineId: string

  /** The ONE place commander presence changes. Both callers — the `__clients` snapshot and `onGone` (our
   *  own backend link died) — mean the same thing when the count reaches zero: nobody is watching. They
   *  used to differ, and `onGone` forgot to drop the device's E2EE session, so `deviceE2eeConnected()`
   *  stayed true and the local dashboard kept a green "device connected" dot for a device long gone. */
  private setCommanderCount(commander: number, active: number | null): void {
    const hadCommander = this.commanderCount > 0
    this.commanderCount = commander
    this.commanderActive = active
    if (hadCommander !== (commander > 0)) this.onCommanderPresenceChanged?.(commander > 0)
    if (commander <= 0) {
      if (this.directDeviceSinks.size) this.e2ee.dropSessionsByRole('device', id => this.directDeviceSinks.has(id))
      else this.e2ee.dropSessionsByRole('device')
    }
  }

  /** True while at least one device (commander) client is connected — gates the LLM recap. */
  hasCommander(): boolean {
    return this.commanderCount > 0
  }

  /** True while at least one connected device is ACTIVELY rendering this machine — gates the full turn-card
   *  STREAM (processing/tool/todos). The recap still runs on hasCommander(), so a BACKGROUND machine gets its
   *  turn-done `summary` card (badge) without the live stream. Falls back to hasCommander() against a
   *  backend that doesn't emit the signal (commanderActive === null). */
  hasActiveCommander(): boolean {
    return this.commanderActive == null ? this.hasCommander() : this.commanderActive > 0
  }

  /** True while a desktop window on THIS computer is attached over loopback.
   *
   *  Separate from `hasCommander()` on purpose: a device and a window are different audiences that
   *  happen to want some of the same work done. The question watcher is the first thing to need it —
   *  polling a pane for a dialog is pointless with nobody rendering it, but "nobody" used to mean
   *  "no device", which left the window unable to learn that an agent was blocked. */
  hasLocalClient(): boolean {
    return this.localClients.size > 0
  }

  /** True after a paired device has completed the E2EE hello/welcome session. */
  deviceE2eeConnected(): boolean {
    return this.e2ee.deviceConnected()
  }

  private readonly directDeviceSinks = new Map<string, (frame: Record<string, unknown>) => void>()
  private readonly directDevicePins = new Map<string, string>()
  onDirectDeviceRevoked?: (fingerprint: string) => void
  attachDirectDevice(connId: string, send: (frame: Record<string, unknown>) => void): void { this.directDeviceSinks.set(connId, send) }
  detachDirectDevice(connId: string): void { this.directDeviceSinks.delete(connId); this.directDevicePins.delete(connId); this.e2ee.dropSession(connId); this.autonomousDeviceRelay?.drop(connId); this.onCommanderPresenceChanged?.(this.hasCommander()) }
  pairedDirectFingerprint(connId: string): string | null { const pub = this.directDevicePins.get(connId); return pub ? fingerprint(b64d(pub)) : null }
  async receiveDirectDevice(connId: string, frame: Record<string, unknown>, pairingAllowed: boolean): Promise<void> {
    if (!this.directDeviceSinks.has(connId)) return
    const type = frame.type
    if (type === 'machine_selected') return
    if (type === 'autonomous_device_request') { await this.autonomousDeviceRelay?.handle(connId, frame); return }
    const controls = pairingAllowed ? ['e2e_pair_intent', 'e2e_pair_cancel', 'e2e_pake', 'e2e_hello', 'e2e_status'] : ['e2e_hello', 'e2e_status']
    if ((type === 'e2e_pake' || type === 'e2e_pair_cancel') && this.e2ee.pendingConnection() !== connId) return
    if (typeof type === 'string' && controls.includes(type)) this.e2ee.handleFrame(connId, frame)
  }
  private autonomousDeviceRelay?: AutonomousDeviceRelay
  setAutonomousDeviceService(service: AutonomousDeviceService): void {
    this.autonomousDeviceRelay = new AutonomousDeviceRelay(
      this.e2ee,
      (connId, frame) => this.sendTo(connId, frame),
      service,
      this.machineId,
      () => this.onCommanderJoin?.(),
      identity => this.e2ee.revoke(fingerprint(b64d(identity))),
    )
  }
  directAutonomousDeviceSessions(): number { return this.autonomousDeviceRelay?.count(id => this.directDeviceSinks.has(id)) ?? 0 }
  autonomousDeviceConnected(): boolean { return this.autonomousDeviceRelay?.connected() ?? false }
  emitAutonomousDeviceEvent(frame: AutonomousDeviceFrame): void { this.autonomousDeviceRelay?.emit(frame) }

  /** Live backend link state (local dashboard + E2EE gating). */
  isConnected(): boolean {
    return this.ws?.readyState === WebSocket.OPEN
  }

  /** A browser waiting to pair (local dashboard), or null. */
  pendingPair(): ReturnType<E2eeManager['pendingPair']> {
    return this.e2ee.pendingPair()
  }

  /** Record the local dashboard port so it's surfaced to the web (in e2e_status) for approve-via-web. */
  setDashboardPort(port: number): void {
    this.e2ee.dashboardPort = port
  }

  constructor(machineId: string, auth?: AuthSessionManager, onStatus: (connected: boolean) => void = () => {}, computerId = '', autonomousEnv = 'prod') {
    this.auth = auth ?? new AuthSessionManager(env.BACKEND_WS_URL.replace(/\/$/, '').replace(/^wss:/, 'https:').replace(/^ws:/, 'http:'))
    this.testToken = auth ? undefined : machineId
    // `?label=<hostname>` lets the backend record which machine connected (shown on the machine card);
    // `?computer=<stable id>` enforces one-computer-per-computer (a 2nd computer is rejected with HTTP 409);
    // `?v=<VERSION>` is our own version, which the backend stores on the machine at every connect.
    const base = `${env.BACKEND_WS_URL.replace(/\/$/, '')}/api/adapter-ws?label=${encodeURIComponent(hostname())}&v=${encodeURIComponent(VERSION)}&autonomousEnv=${encodeURIComponent(autonomousEnv)}`
    // `?machine=<id>` is the machine this daemon still believes it is. The backend uses it to tell a
    // REVOKED daemon — one whose machine was deleted while it was offline — apart from a first-time
    // pairing, and answers 403 instead of quietly minting a replacement machine. Guarded on shape
    // because in test mode the first constructor argument carries a token, not a machine id.
    const claim = /^[a-f0-9]{32}$/.test(machineId) ? `&machine=${encodeURIComponent(machineId)}` : ''
    this.url = (computerId ? `${base}&computer=${encodeURIComponent(computerId)}` : base) + claim
    this.onStatus = onStatus
    this.machineId = machineId
    this.e2ee = new E2eeManager({
      machineId: this.machineId,
      sendTo: (connId, frame) => this.sendTo(connId, frame),
      sendUser: (frame) => this.sendUser(frame),
      isConnected: () => this.isConnected(),
      isConnectionAvailable: connId => this.directDeviceSinks.has(connId) || this.isConnected(),
      onIdentityPaired: (connId, pub) => { if (this.directDeviceSinks.has(connId)) this.directDevicePins.set(connId, pub) },
      onIdentityRevoked: identity => { this.autonomousDeviceRelay?.revoke(identity); this.onDirectDeviceRevoked?.(fingerprint(b64d(identity))) },
    })
    this.terminalP2p = new TerminalP2pResponderPool({
      sendSignal: (connId, type, payload) => this.sendP2pSignal(connId, type, payload),
      onData: (connId, data) => this.handleP2pData(connId, data),
      onUnavailable: (connId, reason) => this.demoteP2pConnection(connId, reason),
    })
  }

  setTerminalStreamManager(manager: TerminalStreamManager): void {
    this.terminalStreams = manager
  }

  /** Run CPace pairing for a code entered via `harness pair <code>` (delegated to the manager). */
  pair(code: string): Promise<PairResult> {
    return this.e2ee.onPair(code)
  }
  e2eeFingerprint(): string {
    return this.e2ee.fingerprint()
  }
  createSetupToken(): ReturnType<E2eeManager['createSetupToken']> {
    return this.e2ee.createSetupToken()
  }
  /** `harness pairings` — list paired clients. */
  listPairs(): ReturnType<E2eeManager['listPaired']> {
    return this.e2ee.listPaired()
  }
  /** `harness unpair <id>` — unpair one client (signals it to re-pair if online). */
  revoke(id: string): ReturnType<E2eeManager['revoke']> {
    return this.e2ee.revoke(id)
  }
  /** `harness unpair --all` — unpair every client. */
  revokeAll(): ReturnType<E2eeManager['revokeAll']> {
    return this.e2ee.revokeAll()
  }
  /** `harness remote-password set` — stretch + persist a new persistent remote password. */
  setRemotePassword(password: string): ReturnType<E2eeManager['setRemotePassword']> {
    return this.e2ee.setRemotePassword(password)
  }
  /** `harness remote-password clear` — remove the persistent remote password. */
  clearRemotePassword(): void {
    this.e2ee.clearRemotePassword()
  }
  /** `harness remote-password status` — whether one is set, and its fingerprint. */
  remotePasswordStatus(): ReturnType<E2eeManager['remotePasswordStatus']> {
    return this.e2ee.remotePasswordStatus()
  }

  connect(): void {
    if (this.closed || this.ws || this.connecting) return
    this.connecting = true
    void this.connectWithSession()
  }

  private async connectWithSession(): Promise<void> {
    let token: string
    try {
      token = this.testToken ?? await this.auth.accessToken()
    } catch (err) {
      this.connecting = false
      this.onStatus(false)
      const delay = err instanceof AuthSessionError && err.code === 'INVALID_REFRESH' ? MAX_DELAY_MS : BASE_DELAY_MS
      if (!this.closed) setTimeout(() => this.connect(), delay)
      return
    }
    if (this.closed) { this.connecting = false; return }
    // On timeout `ws` emits 'error' ("Opening handshake has timed out") then 'close', which lands in
    // onGone below and re-enters the ordinary backoff — the same path a refused connection takes.
    const ws = new WebSocket(this.url, [token], { handshakeTimeout: HANDSHAKE_TIMEOUT_MS })
    this.ws = ws
    this.connecting = false

    ws.on('open', () => {
      this.attempts = 0
      console.log(`[backend] connected → ${this.url}`)
      this.onStatus(true)
      this.drainQueue()

      this.heartbeat = watchSocketLiveness(ws, {
        onIdle: (idleMs) => console.log(`[backend] no traffic for ${Math.round(idleMs / 1000)}s — terminating the link`),
      })

      // App-level ping refreshes the backend's presence key (TTL 30s).
      this.appPing = setInterval(() => this.sendBestEffort({ t: 'ping' }), APP_PING_MS)
    })

    ws.on('message', (raw, isBinary) => {
      if (isBinary) {
        const hop = decodeTerminalHop(new Uint8Array(raw as Buffer))
        if (hop?.direction === TerminalHopDirection.down) this.enqueueTerminalBinary(hop.connId, hop.clientFrame)
        return
      }
      let env_: DownEnvelope
      try { env_ = JSON.parse(raw.toString()) as DownEnvelope } catch { return }
      if (env_.t === 'down' && env_.frame) {
        // A malformed/hostile down-frame (bad __e2e envelope, bad ephemeral key) can throw in the
        // pre-`try` part of dispatchDown; without this .catch that becomes an unhandledRejection and the
        // daemon exits. Contain it: log, drop the frame, keep the socket alive.
        this.enqueueDown(env_.frame, env_.connId ?? '')
      }
    })

    const onGone = (why: string): void => {
      if (this.ws !== ws) return
      this.ws = null
      if (this.heartbeat) { this.heartbeat.stop(); this.heartbeat = null }
      if (this.appPing) { clearInterval(this.appPing); this.appPing = null }
      this.draining = false
      // While the backend link is down we can neither observe device presence nor deliver a card, so
      // default the recap gate to OFF (safe value) instead of holding a stale count — otherwise a turn
      // completing during the gap burns a `claude -p` recap that goes nowhere. attachAdapter always
      // re-pushes the true count via recomputeAndSendClients on reconnect (and 0→N re-fires the replay).
      this.setCommanderCount(0, null) // active count is unknown until the next __clients snapshot
      void this.terminalStreams?.closeConnectionsWhere(
        (connId) => !isLocalClientId(connId),
        'backend disconnected',
        false,
      )
      void this.terminalP2p.stop()
      this.p2pPendingOpens.clear()
      this.p2pStreams.clear()
      this.replayCommanderOnNextSnapshot = true
      this.onStatus(false)
      if (this.closed) return
      const delay = Math.min(MAX_DELAY_MS, BASE_DELAY_MS * 2 ** Math.min(this.attempts++, 5))
      console.log(`[backend] disconnected (${why}) — retrying in ${Math.round(delay / 1000)}s (attempt ${this.attempts})`)
      setTimeout(() => this.connect(), delay)
    }
    ws.on('close', (code) => onGone(`close ${code}`))
    ws.on('error', (err) => {
      const e = err as Error & { code?: string }
      const msg = e.message || e.code || String(err)
      console.error('[backend] socket error:', msg)
      // 401/403 on the upgrade = this machine is gone/invalid (deleted while we were offline, or a bad
      // token). Retrying cannot help → stop for good and let CLI clear the saved SSO session.
      if (/Unexpected server response: 401\b/.test(msg) && !this.retryingAuth) {
        this.retryingAuth = true
        this.ws = null
        void this.auth.accessToken({ force: true, failedToken: token })
          .then(() => { this.retryingAuth = false; this.connect() })
          .catch(() => { this.retryingAuth = false; this.closed = true; this.onRevoked?.() })
      } else if (/Unexpected server response: 40[13]\b/.test(msg)) {
        this.closed = true
        this.onRevoked?.()
      }
      // 409 = another computer already holds this machine. Keep the SSO session; stop
      // retrying (the 40[13] regex above deliberately excludes 409, so without this it would loop).
      else if (/Unexpected server response: 409\b/.test(msg)) {
        this.closed = true
        this.onBusy?.()
      }
      try { ws.close() } catch { /* ignore */ }
    })
  }

  async stop(): Promise<void> {
    this.closed = true
    if (this.heartbeat) this.heartbeat.stop()
    if (this.appPing) clearInterval(this.appPing)
    await this.terminalStreams?.stop()
    await this.terminalP2p.stop()
    try { this.ws?.close() } catch { /* ignore */ }
    this.ws = null
  }

  /** Send an up-frame (event or RPC reply) to the WEB audience. Queued while disconnected.
   *  User-content events are group-encrypted (E2EE) here; system frames pass through as plaintext. */
  send(frame: Frame): void {
    if (env.LOG_FRAMES) logFrame('→', 'web', frame)
    for (const [connId, sink] of this.localClients) {
      if (!sink.sendFrame(frame)) void this.unregisterLocalClient(connId)
    }
    this.enqueue({ t: 'up', frame: this.e2ee.wrapUp(frame) })
  }

  /** Send an up-frame to the LOOPBACK clients only — never to the cloud.
   *
   *  For things that describe what is happening at THIS desk rather than what the machine is doing: the
   *  dial is a physical object on one table, and a finger moving on its glass is meaningful to the window
   *  in front of it and to nothing else. `send()` fans out to the web audience as well, which would scroll
   *  a window on a computer the user is not sitting at. */
  sendLocal(frame: Frame): void {
    if (env.LOG_FRAMES) logFrame('→', 'local', frame)
    for (const [connId, sink] of this.localClients) {
      if (!sink.sendFrame(frame)) void this.unregisterLocalClient(connId)
    }
  }

  /** Ask one local desktop to select focus, without opening panes in every window. */
  sendFirstLocal(frame: Frame): boolean {
    for (const [connId, sink] of this.localClients) {
      if (sink.sendFrame(frame)) return true
      void this.unregisterLocalClient(connId)
    }
    return false
  }

  /** Send an up-frame to exactly ONE web connection (E2EE pairing/welcome + targeted RPC replies). */
  sendTo(connId: string, frame: Frame): void {
    const direct = this.directDeviceSinks.get(connId)
    if (direct) { direct(frame); return }
    const local = this.localClients.get(connId)
    if (local) {
      if (!local.sendFrame(frame)) void this.unregisterLocalClient(connId)
      return
    }
    this.enqueue({ t: 'up', targetConnId: connId, frame })
  }

  /** Pairwise terminal output is never queued across reconnect: the stream/lease is closed on link loss. */
  sendTerminalTo(connId: string, type: string, payload: Record<string, unknown>): boolean {
    const local = this.localClients.get(connId)
    if (local) return local.sendFrame({ type, payload })
    const frame = this.e2ee.wrapTarget(connId, type, payload)
    if (!frame) return false
    if (this.routeTerminalOutputToP2p(connId, type, payload)) {
      if (this.terminalP2p.send(connId, JSON.stringify(frame))) return true
      this.demoteP2pConnection(connId, 'send_failed')
    }
    return this.sendBestEffort({
      t: 'up',
      targetConnId: connId,
      webEligible: true,
      commanderEligible: false,
      frame,
    })
  }

  /** Pairwise-encrypted binary terminal output/keyframe. The hop prefix exposes
   * only connId and direction to the opaque backend relay. */
  sendTerminalBinaryTo(connId: string, clear: TerminalBinaryClear): boolean {
    const local = this.localClients.get(connId)
    if (local) {
      const frame = encodeTerminalLocal(clear)
      return frame ? local.sendBinary(frame) : false
    }
    const clientFrame = this.e2ee.wrapTerminalBinary(connId, clear)
    if (!clientFrame) return false
    if (this.p2pStreams.get(connId)?.has(clear.streamId)) {
      if (this.terminalP2p.send(connId, Buffer.from(clientFrame))) return true
      this.demoteP2pConnection(connId, 'send_failed')
    }
    const packet = encodeTerminalHop(TerminalHopDirection.up, connId, clientFrame)
    if (!packet || !this.ws || this.ws.readyState !== WebSocket.OPEN) return false
    try { this.ws.send(packet); return true } catch { return false }
  }

  /** Send a user-level notification to every logged-in browser that owns this machine. */
  sendUser(frame: Frame): void {
    this.enqueue({ t: 'up', userEligible: true, webEligible: false, frame })
  }

  /** Send a DEVICE-audience frame (commanderEligible, not web). User/data frames are group-encrypted
   *  (E2EE) here so the backend relays only ciphertext; system/presence frames pass through. */
  /**
   * A tap on everything bound for a device, taken BEFORE E2EE wrapping.
   *
   * The dial on the USB cable is a second device audience, and it wants exactly what this one gets — the
   * same `commander_event` cards, in the same order, with the same recaps. Teeing here rather than adding
   * a parallel emit at each of the two dozen call sites is what keeps the two surfaces from drifting: a
   * new event kind reaches the cable the day it reaches the socket, without anyone remembering to add it.
   *
   * Plaintext on purpose: E2EE exists because the backend relays those frames. The cable relays nothing —
   * it is a wire the user physically owns, running to a process on their own computer.
   */
  onOutboundCommander?: (frame: Frame) => void

  sendCommander(frame: Frame): void {
    this.onOutboundCommander?.(frame)
    if (env.LOG_FRAMES) logFrame('→', 'device', frame)
    this.enqueue({ t: 'up', webEligible: false, commanderEligible: true, frame: this.e2ee.wrapCommander(frame) })
  }

  /** Attach one authenticated loopback desktop client to the same RPC and event plane as cloud web. */
  registerLocalClient(connId: string, sink: LocalClientSink): boolean {
    if (!isLocalClientId(connId) || this.localClients.has(connId)) return false
    this.localClients.set(connId, sink)
    return true
  }

  /** Release all connection-scoped state when the loopback WebSocket closes. */
  async unregisterLocalClient(connId: string): Promise<void> {
    if (!this.localClients.delete(connId)) return
    this.downChains.delete(connId)
    await this.terminalStreams?.closeConnection(connId, 'local client disconnected', false)
  }

  /** Route an authenticated local JSON frame through the existing per-client FIFO. */
  handleLocalFrame(connId: string, frame: Frame): void {
    if (!this.localClients.has(connId)) return
    this.enqueueDown(frame, connId)
  }

  /** Route an authenticated local terminal frame without applying cloud E2EE. */
  async handleLocalBinary(connId: string, frame: TerminalBinaryClear): Promise<void> {
    if (!this.localClients.has(connId)) return
    await this.terminalStreams?.handleBinary(connId, frame)
  }

  private enqueue(msg: OutboundEnvelope): void {
    const item: QueueItem = { id: this.nextQueueId++, data: JSON.stringify(msg), msg, attempts: 0 }
    if (this.queue.length >= QUEUE_MAX) this.dropOneQueued()
    if (this.queue.length >= QUEUE_MAX) {
      this.droppedSinceLog++
      this.logQueueDrops()
      return
    }
    this.queue.push(item)
    this.drainQueue()
  }

  private sendBestEffort(msg: OutboundEnvelope): boolean {
    const data = JSON.stringify(msg)
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      try { this.ws.send(data); return true } catch { /* ignore */ }
    }
    return false
  }

  private sendP2pSignal(connId: string, type: string, payload: TerminalP2pSignal): void {
    const frame = this.e2ee.wrapTarget(connId, type, { ...payload })
    if (frame) this.sendTo(connId, frame)
  }

  private handleP2pData(connId: string, data: TerminalP2pData): void {
    if (typeof data === 'string') {
      if (Buffer.byteLength(data, 'utf8') > 512 * 1024) return
      let frame: Frame
      try { frame = JSON.parse(data) as Frame } catch { return }
      if (typeof frame.type !== 'string' || !TERMINAL_P2P_DOWN_TYPES.has(frame.type)) return
      this.enqueueDown(frame, connId, 'p2p')
      return
    }
    if (data.length > 512 * 1024) return
    this.enqueueTerminalBinary(connId, data)
  }

  private noteTerminalInputRoute(
    connId: string,
    type: string,
    payload: Record<string, unknown>,
    transport: 'relay' | 'p2p',
  ): void {
    if (type === 'terminal_open' && typeof payload.requestId === 'string') {
      let pending = this.p2pPendingOpens.get(connId)
      if (!pending) { pending = new Set(); this.p2pPendingOpens.set(connId, pending) }
      if (transport === 'p2p') pending.add(payload.requestId)
      else pending.delete(payload.requestId)
      return
    }
    const streamId = typeof payload.streamId === 'string' ? payload.streamId : ''
    // Live-migration promotion: the client's remoteRelay.ts sends a SECOND terminal_resync over p2p
    // (after the first one, over relay, already drained/snapshotted the stream) once its own p2p
    // channel is ready — arriving here is our signal to start routing this stream's OUTPUT over p2p
    // too, mirroring what the client just did on its side. hasStream() guards against promoting a
    // streamId whose pane was closed in the same instant the migration was in flight.
    if (type === 'terminal_resync' && transport === 'p2p' && streamId
      && this.terminalStreams?.hasStream(connId, streamId)) {
      let streams = this.p2pStreams.get(connId)
      if (!streams) { streams = new Set(); this.p2pStreams.set(connId, streams) }
      streams.add(streamId)
      return
    }
    if (transport === 'relay' && streamId) this.p2pStreams.get(connId)?.delete(streamId)
  }

  private routeTerminalOutputToP2p(connId: string, type: string, payload: Record<string, unknown>): boolean {
    const requestId = typeof payload.requestId === 'string' ? payload.requestId : ''
    const streamId = typeof payload.streamId === 'string' ? payload.streamId : ''
    const pending = this.p2pPendingOpens.get(connId)
    if ((type === 'terminal_ready' || type === 'terminal_error') && requestId && pending?.has(requestId)) {
      pending.delete(requestId)
      if (pending.size === 0) this.p2pPendingOpens.delete(connId)
      if (type === 'terminal_ready' && streamId) {
        let streams = this.p2pStreams.get(connId)
        if (!streams) { streams = new Set(); this.p2pStreams.set(connId, streams) }
        streams.add(streamId)
      }
      return true
    }
    const streams = this.p2pStreams.get(connId)
    const selected = !!streamId && streams?.has(streamId) === true
    if (selected && type === 'terminal_closed') {
      streams!.delete(streamId)
      if (streams!.size === 0) this.p2pStreams.delete(connId)
    }
    return selected
  }

  private demoteP2pConnection(connId: string, reason: string): void {
    this.p2pPendingOpens.delete(connId)
    this.p2pStreams.delete(connId)
    console.warn(`[terminal-p2p] conn=${sid(connId)} fallback=relay reason=${reason}`)
  }

  private enqueueDown(frame: Frame, connId: string, transport: 'relay' | 'p2p' = 'relay'): void {
    const key = connId || '__backend__'
    const previous = this.downChains.get(key) ?? Promise.resolve()
    const next = previous
      .catch(() => { /* prior failure is already logged */ })
      .then(() => this.dispatchDown(frame, connId, transport))
      .catch((err) => {
        console.error('[backend] down-frame dispatch failed:', err instanceof Error ? err.message : err)
      })
      .finally(() => {
        if (this.downChains.get(key) === next) this.downChains.delete(key)
      })
    this.downChains.set(key, next)
  }

  private enqueueTerminalBinary(connId: string, raw: Uint8Array): void {
    const key = connId || '__backend__'
    const previous = this.downChains.get(key) ?? Promise.resolve()
    const next = previous
      .catch(() => { /* prior failure is already logged */ })
      .then(async () => {
        const clear = this.e2ee.unwrapTerminalBinary(connId, raw)
        if (clear) await this.terminalStreams?.handleBinary(connId, clear)
      })
      .catch((err) => {
        console.error('[backend] binary terminal dispatch failed:', err instanceof Error ? err.message : err)
      })
      .finally(() => {
        if (this.downChains.get(key) === next) this.downChains.delete(key)
      })
    this.downChains.set(key, next)
  }

  private drainQueue(): void {
    if (this.draining || !this.queue.length) return
    const ws = this.ws
    if (!ws || ws.readyState !== WebSocket.OPEN) return
    const item = this.queue[0]
    this.draining = true
    try {
      ws.send(item.data, (err?: Error) => {
        if (this.ws !== ws) return
        if (err) {
          item.attempts++
          this.draining = false
          console.error(`[backend] queued send failed (id=${item.id}, attempts=${item.attempts}):`, err.message)
          try { ws.close() } catch { /* ignore */ }
          return
        }
        if (this.queue[0] === item) this.queue.shift()
        this.draining = false
        this.drainQueue()
      })
    } catch (err) {
      item.attempts++
      this.draining = false
      console.error(`[backend] queued send threw (id=${item.id}, attempts=${item.attempts}):`, err instanceof Error ? err.message : err)
      try { ws.close() } catch { /* ignore */ }
    }
  }

  private dropOneQueued(): void {
    const idx = this.queue.findIndex((item) => !('targetConnId' in item.msg))
    const dropAt = idx >= 0 ? idx : 0
    if (this.queue.splice(dropAt, 1).length) {
      this.droppedSinceLog++
      this.logQueueDrops()
    }
  }

  private logQueueDrops(): void {
    if (this.droppedSinceLog === 1 || this.droppedSinceLog % 100 === 0) {
      console.warn(`[backend] outbound queue full; dropped ${this.droppedSinceLog} frame(s) so far`)
    }
  }

  // ── down-frame dispatch (the hosted runtime-role RPC switch) ────────────────────────────────────────────

  /** Emit an RPC reply. For an E2EE-session requester whose result carries user content, the reply is
   *  encrypted with that connection's session key and delivered ONLY to it. Content-bearing adapter data
   *  is never returned plaintext: even legacy backend nodeRequest (`connId === ''`) gets only an error. */
  private emitReply(connId: string, type: string, requestId: unknown, payload: Record<string, unknown>): void {
    const resultType = `${type}_result`
    // Before the E2EE wrap: an RPC reply is only readable here.
    if (env.LOG_FRAMES && type !== 'agent_read_file') logFrame('→', connId ? `conn:${sid(connId)}` : 'backend', { type: resultType, payload: { requestId, ...payload } })
    if (this.localClients.has(connId)) {
      this.sendTo(connId, { type: resultType, payload: { requestId, ...payload } })
      return
    }
    if (connId && this.e2ee.hasSession(connId) && ENCRYPTED_RPC_RESULT_TYPES.has(resultType)) {
      let replyPayload = payload
      if (resultType === 'agent_recent_result') {
        const trim = fitRecentReplyPayloadForDevice(
          payload,
          (candidate) => this.e2ee.rpcReplyFrameBytes(connId, resultType, requestId, candidate),
        )
        replyPayload = trim.payload
        if (trim.trimmed) {
          console.warn(
            `[recent-trim] agent=${String(payload.agentId ?? '')} originalFrame=${trim.originalBytes ?? 'unknown'} ` +
            `finalFrame=${trim.finalBytes ?? 'unknown'} target=${DEVICE_RECENT_SAFE_FRAME_BYTES} ` +
            `textBytes=${trim.textBytes} recapBytes=${trim.recapBytes}`,
          )
        }
      }
      const wrapped = this.e2ee.wrapRpcReply(connId, resultType, requestId, replyPayload)
      if (wrapped) { this.sendTo(connId, wrapped); return }
    }
    // Enforcement ("no E2EE ⇒ no adapter data"): content-bearing RPC replies must never leave the
    // adapter plaintext. A real client gets a targeted error; the legacy backend nodeRequest awaiter
    // (`connId === ''`) gets a broadcast error with the same requestId so it fails closed without data.
    // 
    if (ENCRYPTED_RPC_RESULT_TYPES.has(resultType)) {
      const errorFrame = { type: resultType, payload: { requestId, error: 'E2EE_REQUIRED' } }
      if (connId) this.sendTo(connId, errorFrame)
      else this.send(errorFrame)
      return
    }
    this.send({ type: resultType, payload: { requestId, ...payload } })
  }

  private async dispatchDown(frame: Frame, connId: string, transport: 'relay' | 'p2p' = 'relay'): Promise<void> {
    const type = frame.type as string | undefined
    if (!type) return
    const local = this.localClients.has(connId)
    // E2EE control frames (pairing/handshake) are handled by the manager, never as node RPCs.
    if (type.startsWith('e2e_')) {
      if (local) {
        this.sendTo(connId, { type: 'local_protocol_error', payload: { error: 'LOCAL_E2EE_UNSUPPORTED' } })
        return
      }
      this.e2ee.handleFrame(connId, frame)
      return
    }
    if (type === 'autonomous_device_request') {
      if (!local) await this.autonomousDeviceRelay?.handle(connId, frame)
      return
    }
    // Client→adapter encrypted frames: chat messages plus trusted web control actions. Plaintext
    // passes through for legacy/device transition paths; undecryptable ciphertext is dropped.
    if (!local && isEncryptedDownType(type)) {
      if (type !== 'message' && type !== 'question_response' && !isWrapped(frame.payload)) {
        const requestId = (frame.payload as { requestId?: unknown } | undefined)?.requestId
        if (requestId !== undefined) this.emitReply(connId, type, requestId, { error: 'E2EE_REQUIRED' })
        return
      }
      const dec = this.e2ee.unwrapDown(connId, frame)
      if (!dec) return
      frame = dec
    }
    if (!local && TERMINAL_P2P_SIGNAL_TYPES.has(type)) {
      await this.terminalP2p.handleSignal(connId, type, frame.payload)
      return
    }
    // Logged AFTER the unwrap above, so a down-frame reads as what the client actually asked for rather
    // than as an opaque __e2e envelope.
    // Terminal frames contain raw keystrokes, paste text and screen bytes after
    // unwrap. Never pass them to the frame logger, even in diagnostic mode.
    if (env.LOG_FRAMES && !type.startsWith('terminal_') && type !== 'agent_read_file') {
      logFrame('←', connId ? `conn:${sid(connId)}` : 'backend', frame)
    }
    const reply = (t: string, rid: unknown, p: Record<string, unknown>): void => this.emitReply(connId, t, rid, p)
    // Cross-instance client snapshot. Generation detects leave/join cycles that coalesce to the same
    // count; count rise remains the compatibility fallback for older backends.
    if (type === '__clients') {
      const payload = (frame.payload ?? {}) as { commander?: number; commanderActive?: number; commanderJoinGeneration?: number }
      const commander = Number(payload.commander ?? 0)
      const rawGeneration = payload.commanderJoinGeneration
      const generation = typeof rawGeneration === 'number' && Number.isSafeInteger(rawGeneration) && rawGeneration >= 0
        ? rawGeneration
        : undefined
      const replay = shouldReplayCommander(
        this.commanderCount,
        commander,
        this.replayedCommanderGeneration,
        generation,
        this.replayCommanderOnNextSnapshot,
      )
      this.setCommanderCount(commander, payload.commanderActive != null ? Number(payload.commanderActive) : null)
      if (replay) {
        this.replayCommanderOnNextSnapshot = false
        if (generation != null) this.replayedCommanderGeneration = generation
        this.onCommanderJoin?.()
      }
      return
    }
    if (type === '__client_disconnected') {
      // The backend is authoritative for the outer connId. Drop both kinds of
      // connection-scoped state immediately; otherwise a dead Desktop keeps a
      // terminal controller lease until the 30-second heartbeat timeout.
      this.autonomousDeviceRelay?.drop(connId)
      this.e2ee.dropSession(connId)
      await this.terminalP2p.closeConnection(connId, 'client_disconnected', false)
      this.p2pPendingOpens.delete(connId)
      this.p2pStreams.delete(connId)
      await this.terminalStreams?.closeConnection(
        connId,
        'client connection closed',
        false,
      )
      return
    }
    // Other backend-hub internal control frames (__clients_dirty) — not for us; drop silently.
    if (type.startsWith('__')) return

    // The machine was deleted/revoked from the web → stop for good (don't reconnect) and let the CLI
    // clear the saved token. `closed` blocks the reconnect that would otherwise fire on socket drop.
    if (type === 'machine_revoked') { this.closed = true; this.onRevoked?.(); return }

    // Machine display name (seed on connect + web renames) — mirrored locally for `harness status`.
    if (type === 'machine_meta') {
      const name = (frame.payload as { name?: unknown } | undefined)?.name
      this.onMachineMeta?.(typeof name === 'string' && name.trim() ? name.trim() : null)
      return
    }

    const payload = (frame.payload ?? {}) as Record<string, unknown>
    const requestId = payload.requestId

    if (type.startsWith('terminal_')) {
      this.noteTerminalInputRoute(connId, type, payload, transport)
      if (this.terminalStreams) await this.terminalStreams.handleFrame(connId, type, payload)
      return
    }

    try {
      switch (type) {
        case 'device_e2ee_pair':
          await this.e2ee.pairDeviceFromTrustedWeb(connId, payload)
          return

        case 'e2ee_pairings_list':
          reply(type, requestId, { pairs: this.e2ee.listPaired(connId) })
          return

        case 'e2ee_pairing_unpair':
          this.e2ee.revokeFromTrustedWeb(connId, payload)
          return

        case 'e2ee_pairings_unpair_all':
          this.e2ee.revokeAllFromTrustedWeb(connId, requestId)
          return

        case 'e2ee_browser_link_create': {
          const setup = this.e2ee.createSetupToken()
          reply(type, requestId, setup)
          return
        }

        case 'agents_list': {
          const projects = await Promise.all(registry.advertised().map((s) => this.toProject(s)))
          // Ordered by creation time, oldest → newest — a stable tab order that doesn't reshuffle as
          // sessions become active (createdAt = the session's registeredAt). The id breaks a tie so the
          // order is TOTAL: without it two agents registered in the same millisecond fall through to array
          // position, which is Map insertion order and differs between daemon runs — the web, the app and
          // the dial would each show a different order for the same registry. `cableHost.listAgents`
          // sorts by the same rule; the two must stay identical.
          projects.sort((a, b) => Date.parse(a.createdAt) - Date.parse(b.createdAt) || a.id.localeCompare(b.id))
          if (this.e2ee.sessionRole(connId) === 'device') {
            reply(type, requestId, { agents: projects.slice(0, DEVICE_AGENT_LIST_LIMIT).map(deviceAgentListItem) })
            return
          }
          reply(type, requestId, { agents: projects })
          return
        }

        case 'sessions_list': {
          const projectId = payload.agentId as string | undefined
          if (!projectId) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          const s = registry.resolve(projectId)
          // An agent whose engine has not reported a session yet has no transcript to list. Saying so
          // plainly beats inventing one: the web then shows the tab with an empty thread until the bind
          // lands, instead of pinning `currentSessionId` to an id no event will ever carry.
          if (!s || !s.sessionId) { reply(type, requestId, { sessions: [] }); return }
          const lines = s.transcriptPath ? await tailFile(s.transcriptPath, Infinity) : []
          const st = s.transcriptPath ? await stat(s.transcriptPath).catch(() => null) : null
          reply(type, requestId, {
            sessions: [{
              id: s.sessionId,
              title: projectDisplayName(s),
              timestamp: new Date(s.registeredAt).toISOString(),
              messageCount: lines.length,
              lastActivity: new Date(st?.mtimeMs ?? s.updatedAt).toISOString(),
              participants: [],
            }],
          })
          return
        }

        case 'session_get': {
          const sessionId = payload.sessionId as string | undefined
          if (!sessionId) { reply(type, requestId, { error: 'MISSING_SESSION_ID' }); return }
          // Only serve transcripts for a REGISTERED tmux session, read from its own trusted
          // transcriptPath — never resolve an arbitrary request-supplied id to a file (that let a
          // caller read any *.jsonl on the computer, incl. unshared claude history / traversal).
          const s = registry.resolve(sessionId)
          if (!s) { reply(type, requestId, { error: 'NOT_FOUND' }); return }
          if (s.engine === 'devin') {
            // Devin history comes from its SQLite store (no transcript file). Same replay/window shape.
            const rawLimit = payload.limit
            const limit = typeof rawLimit === 'number' && rawLimit > 0 ? Math.min(Math.floor(rawLimit), 500) : undefined
            const before = typeof payload.before === 'string' ? payload.before : undefined
            const messages = await readDevinMessages(DEVIN_DB, sessionId)
            const timestamp = new Date(s.updatedAt).toISOString()
            if (!limit) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: devinMessagesToEvents(messages), timestamp, engine: s.engine })
              return
            }
            const w = windowDevinMessages(messages, { limit, before })
            if (w.staleCursor) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: [], timestamp, engine: s.engine, hasMore: false, oldestCursor: null, staleCursor: true })
              return
            }
            const events = devinMessagesToEvents(w.window)
            if (before && events[events.length - 1]?.type === 'done') events.pop()
            reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events, timestamp, engine: s.engine, hasMore: w.hasMore, oldestCursor: w.oldestCursor })
            return
          }
          if (s.engine === 'hermes') {
            // Hermes history lives in its SQLite store (no transcript file). Same replay/window shape.
            const rawLimit = payload.limit
            const limit = typeof rawLimit === 'number' && rawLimit > 0 ? Math.min(Math.floor(rawLimit), 500) : undefined
            const before = typeof payload.before === 'string' ? payload.before : undefined
            const messages = await readHermesMessages(HERMES_DB, sessionId)
            const timestamp = new Date(s.updatedAt).toISOString()
            if (!limit) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: hermesMessagesToEvents(messages), timestamp, engine: s.engine })
              return
            }
            const w = windowHermesMessages(messages, { limit, before })
            if (w.staleCursor) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: [], timestamp, engine: s.engine, hasMore: false, oldestCursor: null, staleCursor: true })
              return
            }
            const events = hermesMessagesToEvents(w.window)
            if (before && events[events.length - 1]?.type === 'done') events.pop()
            reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events, timestamp, engine: s.engine, hasMore: w.hasMore, oldestCursor: w.oldestCursor })
            return
          }
          if (s.engine === 'opencode') {
            // OpenCode history comes from its SQLite DB (no transcript file). Same replay/window shape.
            const rawLimit = payload.limit
            const limit = typeof rawLimit === 'number' && rawLimit > 0 ? Math.min(Math.floor(rawLimit), 500) : undefined
            const before = typeof payload.before === 'string' ? payload.before : undefined
            const messages = await readOpencodeMessages(OPENCODE_DB, sessionId)
            const timestamp = new Date(s.updatedAt).toISOString()
            if (!limit) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: opencodeMessagesToEvents(messages), timestamp, engine: s.engine })
              return
            }
            const w = windowOpencodeMessages(messages, { limit, before })
            if (w.staleCursor) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: [], timestamp, engine: s.engine, hasMore: false, oldestCursor: null, staleCursor: true })
              return
            }
            const events = opencodeMessagesToEvents(w.window)
            if (before && events[events.length - 1]?.type === 'done') events.pop()
            reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events, timestamp, engine: s.engine, hasMore: w.hasMore, oldestCursor: w.oldestCursor })
            return
          }
          if (s.engine === 'kilo') {
            // Kilo history likewise comes from its SQLite DB. BOTH branches below are load-bearing: the
            // web client always sends a `limit`, so handling only the full-transcript one opens an empty
            // pane — the exact half-dispatch this file has been caught on before. Cursors are namespaced
            // `kilo:<index>` by `windowKiloMessages`, so a cursor from another engine reads as stale
            // rather than silently indexing into the wrong conversation.
            const rawLimit = payload.limit
            const limit = typeof rawLimit === 'number' && rawLimit > 0 ? Math.min(Math.floor(rawLimit), 500) : undefined
            const before = typeof payload.before === 'string' ? payload.before : undefined
            const messages = await readKiloMessages(KILO_DB, sessionId)
            const timestamp = new Date(s.updatedAt).toISOString()
            if (!limit) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: kiloMessagesToEvents(messages), timestamp, engine: s.engine })
              return
            }
            const w = windowKiloMessages(messages, { limit, before })
            if (w.staleCursor) {
              reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: [], timestamp, engine: s.engine, hasMore: false, oldestCursor: null, staleCursor: true })
              return
            }
            const events = kiloMessagesToEvents(w.window)
            if (before && events[events.length - 1]?.type === 'done') events.pop()
            reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events, timestamp, engine: s.engine, hasMore: w.hasMore, oldestCursor: w.oldestCursor })
            return
          }
          if (!s.transcriptPath) {
            reply(type, requestId, {
              id: sessionId,
              title: projectDisplayName(s),
              events: [],
              timestamp: new Date(s.updatedAt).toISOString(),
              engine: s.engine,
              hasMore: false,
              oldestCursor: null,
            })
            return
          }
          // Optional pagination: `limit` = window size; `before` = uuid cursor (oldest line the
          // client already holds). Absent → full transcript (legacy). Clamp limit defensively.
          const rawLimit = payload.limit
          const limit = typeof rawLimit === 'number' && rawLimit > 0 ? Math.min(Math.floor(rawLimit), 500) : undefined
          const before = typeof payload.before === 'string' ? payload.before : undefined
          const lines = await tailFile(s.transcriptPath, Infinity)
          const st = await stat(s.transcriptPath).catch(() => null)
          const timestamp = new Date(st?.mtimeMs ?? Date.now()).toISOString()

          if (!limit) {
            const fullEvents = s.engine === 'codex'
              ? codexMessagesToEvents(lines, codexSubagentResolverFor(s.codexHome))
              : s.engine === 'cursor'
                ? cursorMessagesToEvents(lines, sessionId, await loadCursorReplayTaskLinks(env.CURSOR_HOME, sessionId))
                : s.engine === 'muse'
                  ? museMessagesToEvents(lines)
                  : s.engine === 'amp'
                  ? await ampHistory(sessionId, lines)
                  : s.engine === 'grok'
                    ? grokHistoryPage(lines, false).events
                  : s.engine === 'agy'
                    ? agyHistoryPage(lines, false).events
                  : s.engine === 'copilot'
                    ? copilotHistoryPage(lines, false).events
                  : s.engine === 'pi'
                  ? piMessagesToEvents(lines)
                  : s.engine === 'commandcode'
                    ? commandcodeMessagesToEvents(lines)
                    : messagesToEvents(lines)
            if (s.engine !== 'cursor' && s.engine !== 'pi' && s.engine !== 'commandcode' && s.engine !== 'muse' && s.engine !== 'amp' && s.engine !== 'grok' && s.engine !== 'agy' && s.engine !== 'copilot') await enrichSubagentStats(fullEvents, s.transcriptPath)
            reply(type, requestId, {
              id: sessionId,
              title: projectDisplayName(s),
              events: fullEvents,
              timestamp,
              engine: s.engine,
            })
            return
          }

          // Muse has no windower, so it must not fall through to the raw one: that pairs claude's
          // line-uuid cursor with claude's normalizer, and a muse transcript comes back EMPTY — the web
          // pane opened blank with no error anywhere. Until a muse window exists, answer the page with
          // the whole transcript (`hasMore: false` ends the scroll honestly, and these sessions are
          // small: a real one measured 271 lines).
          // Amp is in the same position as muse and for the same reason: no windower, so falling
          // through would pair claude's line-uuid cursor with claude's normalizer and return nothing.
          if (s.engine === 'muse' || s.engine === 'amp' || s.engine === 'grok' || s.engine === 'agy' || s.engine === 'copilot') {
            const wholePage = s.engine === 'grok'
              ? grokHistoryPage(lines, true)
              : s.engine === 'agy'
                ? agyHistoryPage(lines, true)
                : s.engine === 'copilot'
                  ? copilotHistoryPage(lines, true)
                  : null
            reply(type, requestId, {
              id: sessionId,
              title: projectDisplayName(s),
              events: s.engine === 'amp'
                ? await ampHistory(sessionId, lines)
                : wholePage
                  ? wholePage.events
                  : museMessagesToEvents(lines),
              timestamp,
              engine: s.engine,
              hasMore: wholePage?.hasMore ?? false,
              oldestCursor: wholePage?.oldestCursor ?? null,
            })
            return
          }
          const w = s.engine === 'codex'
            ? windowCodexLines(lines, { limit, before })
            : s.engine === 'cursor'
              ? windowCursorLines(lines, { limit, before })
              : s.engine === 'pi'
                ? windowPiLines(lines, { limit, before })
                : s.engine === 'commandcode'
                  ? windowCommandCodeLines(lines, { limit, before })
                  : windowRawLines(lines, { limit, before })
          if (w.staleCursor) {
            reply(type, requestId, { id: sessionId, title: projectDisplayName(s), events: [], timestamp, engine: s.engine, hasMore: false, oldestCursor: null, staleCursor: true })
            return
          }
          const events = s.engine === 'codex'
            ? codexMessagesToEvents(w.window, codexSubagentResolverFor(s.codexHome))
            : s.engine === 'cursor'
              ? cursorMessagesToEvents(
                  w.window,
                  sessionId,
                  await loadCursorReplayTaskLinks(env.CURSOR_HOME, sessionId),
                  'startIndex' in w && typeof w.startIndex === 'number' ? w.startIndex : 0,
                  'initialTodos' in w && Array.isArray(w.initialTodos) ? w.initialTodos : [],
                )
              : s.engine === 'pi'
                ? piMessagesToEvents(w.window)
                : s.engine === 'commandcode'
                  ? commandcodeMessagesToEvents(w.window)
                  : messagesToEvents(w.window)
          // muse and amp are answered above and never reach here, so both are absent by design.
          if (s.engine !== 'cursor' && s.engine !== 'pi' && s.engine !== 'commandcode') await enrichSubagentStats(events, s.transcriptPath)
          // Older pages must not inject a spurious end-of-transcript marker mid-scroll.
          if (before && events[events.length - 1]?.type === 'done') events.pop()
          reply(type, requestId, {
            id: sessionId,
            title: projectDisplayName(s),
            events,
            timestamp,
            engine: s.engine,
            hasMore: w.hasMore,
            oldestCursor: w.oldestCursor,
          })
          return
        }

        case 'models_list': {
          const sessionId = typeof payload.agentId === 'string' && payload.agentId
            ? payload.agentId
            : undefined
          const models = this.runtimeModelsProvider
            ? await this.runtimeModelsProvider(sessionId)
            : [{ id: 'default', displayName: 'Remote CLI default' }]
          reply(type, requestId, {
            // The device derives labels from the opaque runtime-v1 id. Omitting the duplicate
            // displayName keeps the encrypted picker response below its 16 KiB decrypt cap.
            models: payload.compact === true
              ? compactRuntimePickerModels(models, sessionId, payload.pickerMode, payload.selectedModel)
              : models,
          })
          return
        }

        case 'engines_probe': {
          // Which engines this machine has, asked BEFORE a create rather than discovered by one
          // failing. Answered here — on the machine in question — because a Mac and the Docker rig
          // routinely hold different engines, and an availability list computed anywhere else is
          // wrong for exactly the remote case this request exists to serve.
          //
          // `engines` narrows the probe to what the caller is showing; an absent or malformed list
          // means "all of them", so an older app that sends nothing still gets a usable answer.
          const asked = Array.isArray(payload.engines)
            ? payload.engines.filter((id): id is AgentEngine =>
              typeof id === 'string' && (ENGINES as readonly string[]).includes(id))
            : undefined
          // A full probe starts interactive login shells and is intentionally detached from this
          // connection's ordered RPC chain. Request ids make its eventual reply safe to deliver out
          // of order; keeping it awaited here made a Create click sit behind an unrelated sweep.
          void this.engineProbeProvider(asked && asked.length > 0 ? asked : undefined)
            .then((availability) => reply(type, requestId, {
              engines: availability.map((entry) => ({
                engine: entry.engine,
                installed: entry.installed,
                command: entry.command,
                installable: entry.installable,
                installCommand: entry.installable ? engineInstallRecipe(entry.engine).command : null,
                // Static per-CLI-version capability, not a probe result: its mere presence is what
                // lets an older CLI (which never sends the field) keep reading as "unknown" rather
                // than "no", per the desktop app's `EngineAvailability.fromJson`.
                ...(entry.engine === 'codex' ? { supportsCodexHome: true } : {}),
              })),
            }))
            .catch(() => reply(type, requestId, { error: 'ENGINE_PROBE_FAILED' }))
          return
        }

        case 'agent_recent': {
          // Device tile restore at boot: the session's persisted LLM turn-summary (recap + body),
          // mirroring the hosted runtime’s SessionService.getRecentEvents. Empty until a turn was summarized
          // (device-gated) — never a resurrected full-text card.
          const projectId = payload.agentId as string | undefined
          if (!projectId) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          const n = Math.max(1, Math.min(5, Number(payload.n) || 2))
          const events = this.recentProvider ? this.recentProvider(projectId, n) : []
          // ASKS TRAVEL AS THEIR OWN LIST, beside the events rather than inside them. A question exists
          // the moment it is asked; a recap exists once the turn has been answered and summarised. They
          // are different lengths on any machine where a turn ended without one, so a reply that folds
          // the questions into the event rows loses exactly the newest ones — and a REMOTE agent then
          // reaches the router with nothing but its name.
          const asks = this.recentAsksProvider ? this.recentAsksProvider(projectId, n) : []
          reply(type, requestId, { agentId: projectId, events, asks })
          return
        }

        case 'voice_route': {
          // Voice router (REMOTE machine): pick the best-fit agent for a transcribed Overview voice task,
          // using each agent's name + recent-turn recap. Backend-originated (connId===''), so the transcript
          // is plaintext and the reply falls through emitReply's plaintext path (voice_route is not E2EE-gated).
          const transcript = typeof payload.transcript === 'string' ? payload.transcript : ''
          if (!transcript.trim()) {
            console.log('[voice-route] backend sent an empty transcript — nothing to route')
            reply(type, requestId, { error: 'MISSING_TRANSCRIPT' })
            return
          }
          const projects = (await Promise.all(registry.advertised().map((s) => this.toProject(s)))).slice(0, 24) // cap the router prompt (mirror the hosted runtime path)
          const agents = projects.map((p) => {
            // Use the last 3 turns' recaps (not just 1) — a single turn can misrepresent what the agent is
            // actually working on; 3 gives the router a more stable picture.
            const recents = this.recentProvider?.(p.id, 3) ?? []
            const recentSummary = recents.map((r) => r?.recap || r?.text || '').filter(Boolean).join(' · ')
            // engine: the router classifies with a CLI the machine actually runs (a live agent proves it is
            // installed and logged in), so it has to travel with the agent.
            const session = registry.resolve(p.id)
            return { id: p.id, name: p.name, recentSummary, engine: session?.engine, gateway: session?.gateway }
          })
          const decision = await routeVoiceTask(transcript, agents)   // logs the task, candidates and pick
          const chosen = projects.find((p) => p.id === decision.agentId)
          reply(type, requestId, { ...decision, agentName: chosen?.name ?? '' })
          return
        }

        case 'agent_update': {
          const projectId = payload.agentId as string | undefined
          if (!projectId) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          const hasName = Object.prototype.hasOwnProperty.call(payload, 'name')
          const hasProfile = Object.prototype.hasOwnProperty.call(payload, 'selectedModel')
          if (!hasName && !hasProfile) { reply(type, requestId, { error: 'MISSING_UPDATE' }); return }
          const name = typeof payload.name === 'string' ? payload.name.trim() : ''
          if (hasName && !name) { reply(type, requestId, { error: 'MISSING_NAME' }); return }
          let s = registry.resolve(projectId)
          if (!s) { reply(type, requestId, { error: 'AGENT_NOT_FOUND' }); return }
          if (hasProfile) {
            if (typeof payload.selectedModel !== 'string' || !this.onRuntimeProfileUpdate) {
              reply(type, requestId, { error: 'INVALID_RUNTIME_PROFILE' })
              return
            }
            try {
              await this.onRuntimeProfileUpdate(projectId, payload.selectedModel)
            } catch (error) {
              const code: RuntimeProfileErrorCode | 'INTERNAL' = error instanceof RuntimeProfileControlError ? error.code : 'INTERNAL'
              reply(type, requestId, { error: code })
              return
            }
          }
          if (hasName) {
            s = registry.rename(projectId, name) ?? s
            this.onAgentRename?.(s, name)
          }
          const agent = await this.toProject(s)
          reply(type, requestId, { agent })
          if (hasName) {
            const renamed = { type: 'agent_renamed', payload: { agentId: s.agentId, name, engine: s.engine } }
            this.send(renamed)          // every OTHER web client on this machine (group-encrypted)
            this.sendCommander(renamed) // and the device
            // The whole rename path was silent end to end, which is why "web1 renamed it, web2 never saw
            // it" had no evidence to work from: nothing said whether the request even arrived. One line
            // here splits the question in two — no line means it never reached the adapter, a line means
            // the fan-out is downstream.
            console.log(`[rename] ${sid(projectId)} → "${preview(name, 40)}" · broadcast to web + device`)
          }
          return
        }

        case 'agent_create_status': {
          const creationId = payload.creationId
          if (!validCreationId(creationId)) { reply(type, requestId, { error: 'INVALID_CREATION_ID' }); return }
          try {
            reply(type, requestId, { creationId, ...await this.creationStatusPayload(this.agentCreations.status(creationId)) })
          } catch (error) {
            reply(type, requestId, { error: error instanceof AgentCreationReceiptError ? error.code : 'INTERNAL' })
          }
          return
        }

        case 'agent_create': {
          const engine = payload.engine as AgentEngine | undefined
          const cwd = payload.cwd
          if (typeof engine !== 'string' || !ENGINES.includes(engine)) { reply(type, requestId, { error: 'INVALID_ENGINE' }); return }
          if (typeof cwd !== 'string' || !isAbsolute(cwd)) { reply(type, requestId, { error: 'INVALID_CWD' }); return }
          if (!this.onCreateAgent) { reply(type, requestId, { error: 'UNSUPPORTED_ON_REMOTE' }); return }
          const creationId = payload.creationId
          if (creationId !== undefined && !validCreationId(creationId)) {
            reply(type, requestId, { error: 'INVALID_CREATION_ID' }); return
          }
          // Absent is the ordinary case and stays indistinguishable from a client that predates grids;
          // present-but-malformed is refused here rather than half-applied at launch, because an agent
          // that quietly ran on the engine's own login would look like it worked.
          const grid = parseGridLaunchOverride(payload.grid)
          if (grid.state === 'invalid') { reply(type, requestId, { error: 'INVALID_GRID', detail: grid.reason }); return }
          // Same validation the desktop app already applies client-side (`Agent._safeCodexHome`) —
          // repeated here because a client's own check is not a guarantee about what actually
          // arrives on the wire.
          const rawCodexHome = typeof payload.codexHome === 'string' ? payload.codexHome : null
          const codexHome = rawCodexHome && rawCodexHome.startsWith('/') && rawCodexHome.length <= 4096
            && !/[\x00-\x1f\x7f]/.test(rawCodexHome)
            ? rawCodexHome
            : null
          if (codexHome && (engine !== 'codex' || grid.state === 'ok')) {
            reply(type, requestId, { error: 'INVALID_CODEX_HOME', detail: 'codexHome is only valid for codex, without a grid' })
            return
          }
          const input = {
            engine,
            cwd,
            bypassPermission: payload.bypassPermission === true,
            grid: grid.state === 'ok' ? grid.override : null,
            codexHome,
          }
          if (creationId !== undefined) {
            // Reserve before spawning. A transport retry carries the SAME creationId; a deliberate
            // New agent action carries a new one. Detach so a status check can pass a slow create
            // on this connection, just as engines_probe is detached above.
            const create = this.onCreateAgent
            try {
              void this.agentCreations.run(creationId, creationFingerprint(input), async () => {
                const result = await create(input)
                if (result.ok) return { state: 'created', agentId: result.session.agentId }
                // tmux may have executed before a timeout; registration cleanup is best-effort.
                // Neither can prove that no process started, so never encourage another launch.
                if (result.error === 'SPAWN_FAILED' || result.error === 'REGISTRATION_FAILED') return { state: 'unconfirmed' }
                return { state: 'failed', error: result.error, ...(result.detail ? { detail: result.detail.slice(0, 2000) } : {}) }
              }).then(async (status) => {
                reply(type, requestId, { creationId, ...await this.creationStatusPayload(status) })
              }).catch(() => reply(type, requestId, { error: 'INTERNAL' }))
            } catch (error) {
              reply(type, requestId, { error: error instanceof AgentCreationReceiptError ? error.code : 'INTERNAL' })
            }
            return
          }
          // Clients predating receipts retain their existing response shape.
          const result = await this.onCreateAgent(input)
          // `detail` carries the underlying cause (tmux's own message) so the person who clicked
          // Create can read it, rather than having to open a log on the machine that failed.
          if (!result.ok) {
            reply(type, requestId, result.detail ? { error: result.error, detail: result.detail } : { error: result.error })
            return
          }
          reply(type, requestId, { agent: await this.toProject(result.session) })
          return
        }

        // Move a RUNNING agent onto a grid. The pane survives; its process is re-exec'd with the
        // engine's grid environment and, when one is bound, `--resume <session>` so the conversation
        // comes back. Refused rather than half-applied: see cli.ts for what it checks first.
        case 'agent_retarget': {
          const agentId = (payload.agentId as string | undefined) || (payload.sessionId as string | undefined)
          if (!agentId) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          if (!this.onRetargetAgent) { reply(type, requestId, { error: 'UNSUPPORTED_ON_REMOTE' }); return }
          const clear = payload.clearGrid === true
          const target = parseGridLaunchOverride(payload.grid)
          // Exactly one, and `clearGrid` is a separate field rather than `grid: null` on purpose:
          // parseGridLaunchOverride already answers `absent` for both undefined and null, so
          // overloading null would make "I forgot the field" and "I mean own login" the same frame.
          if (clear && target.state !== 'absent') {
            reply(type, requestId, { error: 'INVALID_GRID', detail: 'clearGrid and grid are mutually exclusive' })
            return
          }
          if (!clear && target.state !== 'ok') {
            reply(type, requestId, {
              error: 'INVALID_GRID',
              detail: target.state === 'invalid' ? target.reason : 'grid is required',
            })
            return
          }
          const moved = await this.onRetargetAgent({ agentId, grid: target.state === 'ok' ? target.override : null })
          if (!moved.ok) {
            reply(type, requestId, moved.detail ? { error: moved.error, detail: moved.detail } : { error: moved.error })
            return
          }
          reply(type, requestId, { retargeted: true })
          return
        }

        // Delete an agent: signal its validated engine process and drop it from the list. Idempotent — an already
        // gone target still acks + re-emits agent_deleted so the web/device converge. E2EE-gated (the
        // frame arrived decrypted). Keeps the persisted recap + agent name for a later resume.
        case 'agent_delete': {
          const target = (payload.agentId as string | undefined) || (payload.sessionId as string | undefined)
          if (!target) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          this.onDeleteAgent?.(target)
          reply(type, requestId, { deleted: true })
          return
        }

        // Restart an agent: exit its live engine process and relaunch it in the SAME tmux pane, keeping
        // the SAME agentId. Follows agent_create/agent_delete's exact validate → delegate → reply shape.
        case 'agent_restart': {
          const target = payload.agentId as string | undefined
          if (!target) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          if (!this.onRestartAgent) { reply(type, requestId, { error: 'UNSUPPORTED_ON_REMOTE' }); return }
          const result = await this.onRestartAgent(target)
          if (!result.ok) {
            reply(type, requestId, result.detail ? { error: result.error, detail: result.detail } : { error: result.error })
            return
          }
          reply(type, requestId, { agent: await this.toProject(result.session), resumed: result.resumed })
          return
        }

        case 'agent_files': {
          // SourceTree list — rooted at the tmux session's working dir.
          const projectId = payload.agentId as string | undefined
          if (!projectId) { reply(type, requestId, { error: 'MISSING_AGENT_ID' }); return }
          const s = registry.resolve(projectId)
          if (!s?.cwd) { reply(type, requestId, { error: 'AGENT_NOT_FOUND' }); return }
          try { reply(type, requestId, { files: listFileTree(s.cwd) }) }
          catch (e) { reply(type, requestId, { error: e instanceof Error ? e.message : 'FILE_TREE_ERROR' }) }
          return
        }

        case 'fs_list_dir': {
          // One-level remote directory listing for the New Agent folder browser.
          const path = typeof payload.path === 'string' ? payload.path : ''
          const result = listDir(path)
          if ('error' in result) { reply(type, requestId, { error: result.error }); return }
          reply(type, requestId, { ...result })
          return
        }

        case 'codex_profiles_list': {
          // Which CODEX_HOME folders THIS machine can offer — answered here, on the machine in
          // question, for the same reason `engines_probe` is: a Codex profile is a folder on disk,
          // and a folder on a Mac means nothing on the Docker rig it was asked about instead.
          const observed = Array.isArray(payload.observedPaths)
            ? payload.observedPaths.filter((p): p is string => typeof p === 'string')
            : []
          try {
            reply(type, requestId, { profiles: listCodexProfiles(observed) })
          } catch {
            reply(type, requestId, { error: 'CODEX_PROFILES_FAILED' })
          }
          return
        }

        case 'codex_profile_link': {
          const path = typeof payload.path === 'string' ? payload.path : ''
          const result = linkCodexProfile(path)
          if ('error' in result) { reply(type, requestId, { error: result.error }); return }
          reply(type, requestId, { profile: result })
          return
        }

        case 'agent_read_file': {
          // Text reads keep their ≤5 MB guard; media uses bounded binary chunks.
          const projectId = payload.agentId as string | undefined
          const path = payload.path as string | undefined
          if (!projectId || !path) { reply(type, requestId, { error: 'MISSING_AGENT_OR_PATH' }); return }
          const s = registry.resolve(projectId)
          if (!s?.cwd) { reply(type, requestId, { error: 'AGENT_NOT_FOUND' }); return }
          if (payload.media === true) {
            if (typeof path !== 'string') { reply(type, requestId, { error: 'MEDIA_INVALID_REQUEST' }); return }
            try {
              reply(type, requestId, { ...await readMediaPreviewChunk(s.cwd, path, payload.offset, payload.revision) })
            } catch (error) {
              reply(type, requestId, { error: error instanceof MediaPreviewError ? error.message : 'MEDIA_READ_FAILED' })
            }
            return
          }
          try { reply(type, requestId, { path, content: readProjectFile(s.cwd, path) }) }
          catch (e) { reply(type, requestId, { error: e instanceof Error ? e.message : 'NOT_FOUND' }) }
          return
        }

        case 'claude_login_status':
          // Legacy RPC name; report the selected agent's actual engine when one was supplied.
          {
            const target = (payload.agentId as string | undefined) || (payload.sessionId as string | undefined)
            reply(type, requestId, { loggedIn: true, engine: (target ? registry.resolve(target)?.engine : undefined) ?? 'claude', account: hostname() })
          }
          return

        case 'message': {
          const content = payload.content as string | undefined
          const target = (payload.agentId as string | undefined) || (payload.sessionId as string | undefined)
          if (!content || !target) return
          // Inject into the pane; the resulting JSONL user/assistant lines drive the turn lifecycle back
          // to the web (mirror-all) — no synthetic events here. Prefer cli.ts's handler (inject + Enter
          // retry); fall back to a direct inject when unwired (isolation/tests).
          if (this.onMessage) this.onMessage(target, content)
          else console.warn('[backend] message handler is not wired; terminal input was not dispatched')
          return
        }

        case 'cancel': {
          const target = (payload.agentId as string | undefined) || (payload.sessionId as string | undefined)
          if (target) this.onCancel?.(target)
          return
        }

        case 'speaking': {
          // A device is capturing a voice instruction. Mirror the hosted runtime: re-broadcast the presence
          // signal to the web (indicator) + other devices. Ephemeral, no requestId.
          const p = payload as { speaking?: boolean; agentId?: string; sessionId?: string }
          const sessionId = p.sessionId ?? p.agentId
          const projectId = p.agentId ?? p.sessionId ?? null
          const on = p.speaking !== false
          // Echo the canonical agent id, whichever id the sender used to address it.
          const speaker = projectId ? registry.resolve(projectId) : undefined
          const speakerAgentId = speaker?.agentId ?? projectId
          const evt = { type: 'speaking', dbSessionId: speaker?.sessionId ?? sessionId, agentId: speakerAgentId, payload: { speaking: on, agentId: speakerAgentId, sessionId: speaker?.sessionId ?? sessionId } }
          this.send(evt) // web
          this.sendCommander(evt) // other devices (firmware ignores its own echo; matches node forwardToCommander)
          return
        }

        case 'question_response': {
          // A device answered an AskUserQuestion. There's no control channel into an interactive CLI, so
          // cli.ts keys the answer straight into that session's tmux dialog.
          const p = payload as { requestId?: string; sessionId?: string; agentId?: string; answers?: Record<string, string> }
          this.onQuestionAnswer?.(p)
          return
        }

        // This machine's Claude/Codex rate limits, read with ITS OWN credentials. The desktop reads the
        // account on the computer it runs on directly; this is how it reads one on a machine it does
        // not — which may be signed in to a different subscription entirely. The vendor's answer goes
        // back as it came: see lib/accountUsage.ts for why the parsing stays on the client.
        case 'usage_read': {
          reply(type, requestId, { providers: await this.accountUsageReader() })
          return
        }

        // v1 no-ops: no programmatic session control over an interactive tmux claude.
        case 'new_chat':
        case 'compact':
        case 'permission_response':
          return

        default:
          // Unknown RPC with a requestId: reject fast so the web promise doesn't wait out its 20s.
          if (requestId !== undefined) reply(type, requestId, { error: 'UNSUPPORTED' })
          return
      }
    } catch (err) {
      console.error(`[backend] dispatch ${type} failed:`, err)
      if (requestId !== undefined) reply(type, requestId, { error: 'INTERNAL' })
    }
  }

  /** Recover by stable runtime identity; a deleted agent must never become a fresh launch. */
  private async creationStatusPayload(status: AgentCreationStatus): Promise<Record<string, unknown>> {
    if (status.state === 'created') {
      const session = registry.byAgent(status.agentId)
      return session
        ? { state: 'created', agent: await this.toProject(session) }
        : { state: 'unavailable' }
    }
    // A recorded refusal is a completed outcome. Keep it separate from transport/dispatch errors
    // so clients can distinguish "safe to correct the choices" from "outcome still unknown".
    if (status.state === 'failed') {
      return { state: 'failed', failure: { code: status.error, ...(status.detail ? { detail: status.detail } : {}) } }
    }
    return status
  }

  /** Map a registered tmux session onto the web's Project shape (tabs in ProjectTabs). */
  private toProject(s: RegisteredSession): Promise<AgentFrame> {
    return agentFrame(s, {
      selectedModel: this.runtimeProfileProvider?.(s) ?? null,
      terminalAvailable: registry.terminalAvailable(s.agentId),
    })
  }
}