/**
 * Relay client for the sub-client half (T23a): the `client`-role side of the
 * 2.0.0 relay protocol. A DSH process on another machine pairs with a relay
 * server's gateway (Bearer device token, T16) and reaches the server's
 * shared sessions through `POST relay/v1/handshake|invoke|stream` — this
 * module speaks that wire from the far end, with no dependencies beyond the
 * Node 18+ standard library and the global `fetch`.
 *
 * Every call re-reads the credentials through the injected getters, so a
 * re-pair in the settings page applies to the NEXT request without touching
 * this module (the same volatile-row discipline src/client-routes.ts keeps).
 * Missing credentials never touch the network: the client reports
 * `unpaired` and every entry point throws immediately.
 *
 * Two wire details are load-bearing. `accept: application/json` — the
 * gateway answers its pairing wall as a JSON 401 `{reason:'unpaired'}` only
 * to requests that accept JSON; without the header a revoked token would
 * draw the HTML pairing page and be unclassifiable. `redirect: 'manual'` —
 * a wrong or hijacked server must not walk the pairing token through a
 * redirect chain (the same rule src/client-routes.ts probes with).
 *
 * State moves only where the contract says so: `offline` on transport
 * failure, `revoked` on the gateway's unpaired wall, `incompatible` on the
 * relay-only wall or a handshake protocol mismatch, and back to `online` on
 * any success envelope. Everything else — the per-call refusals
 * (`not-shared` / `no-session` / `forbidden-method`), a 200 `{ok:false}`
 * envelope, the gateway's own `relay-unauthorized` — is an ANSWER about the
 * call, not about the link, and leaves the state alone.
 *
 * Since T43 the client also reconnects on its own: an `offline` client walks
 * a 1s → 2s → 5s → 10s → 30s ladder (0–20% jitter per wait) until a success
 * lifts it back to `online`; `unpaired`, `revoked` and `incompatible` never
 * reconnect on their own — they need a user action, which arrives as
 * `credentialsChanged()` (or a fresh `connect()`). The wait is observable as
 * `nextRetryAt`, the last failure code as `lastError`; both carry no
 * credential material.
 *
 * Since T42 the client also judges INTERFACE compatibility: when the wiring
 * injects `computeOwnFingerprints`, every completed handshake is followed by
 * a group-by-group comparison of the server's `fingerprints` map against the
 * locally computed one, stored as `compat` ({@link RelayCompatVerdict}).
 * Groups either side could not compute land in `unavailable` — never in
 * `different` — so a partial view stays silent. The verdict is read live by
 * the status route; listeners additionally hear about it through the same
 * notification channel the state changes use.
 */

import { createHash } from 'node:crypto'

import { compareFingerprints } from './fingerprint.js'
import type { RelayCompatVerdict } from './fingerprint.js'
import type { ServerPluginEntry } from './plugin-bridge.js'

/** The one protocol version this client speaks; the handshake verifies it. */
const RELAY_PROTOCOL = 1

const HANDSHAKE_PATH = '/_dsh/zen-remote/relay/v1/handshake'
const INVOKE_PATH = '/_dsh/zen-remote/relay/v1/invoke'
const STREAM_PATH = '/_dsh/zen-remote/relay/v1/stream'
/** the relay's existing authenticated health route
 * (`GET …/relay/ping`, answers `{ok:true}`). Used by probeHealth() only. */
const PING_PATH = '/_dsh/zen-remote/relay/ping'
const UNSHARE_PATH = '/_dsh/zen-remote/relay/v1/unshare'
const EVENT_RESULT_PATH = '/_dsh/zen-remote/relay/v1/event-result'
const HTTP_PATH = '/_dsh/zen-remote/relay/v1/http'
const UPLOAD_PATH = '/_dsh/zen-remote/relay/v1/upload'
const PLUGIN_PATH = '/_dsh/zen-remote/relay/v1/plugin'
const PLUGIN_TUNNEL_OPEN_PATH = '/_dsh/zen-remote/relay/v1/plugin-tunnel/open'
const PLUGIN_TUNNEL_SEND_PATH = '/_dsh/zen-remote/relay/v1/plugin-tunnel/send'
const PLUGIN_TUNNEL_CLOSE_PATH = '/_dsh/zen-remote/relay/v1/plugin-tunnel/close'
/** T59: a desktop-client device renaming ITSELF in the gateway's table. */
const DEVICE_NAME_PATH = '/_dsh/zen-remote/relay/v1/device/name'

/** How long a stream may stay line-silent before it is judged dead. */
const DEFAULT_IDLE_TIMEOUT_MS = 45_000

/** How long one request/response round-trip may take in full. */
const DEFAULT_REQUEST_TIMEOUT_MS = 15_000

/**
 * The invoke round-trip budget (T31-fix): the configured request timeout
 * plus 2 s per MiB of request body, capped at fifteen minutes. The large
 * bodies are the prompt routes' INLINE images — `session/prompt` /
 * `subagents/prompt` content blocks carry base64 `image` data, one image up
 * to 20 MiB by default (`maxImageBytes`, dsh-attachment-local) or ~27 MiB
 * encoded — so a full-size picture over a slow link gets minutes, not the
 * base seconds. The cap never lands below the configured base timeout.
 */
const INVOKE_TIMEOUT_PER_MIB_MS = 2_000
const INVOKE_TIMEOUT_CAP_MS = 900_000

/**
 * How often an ONLINE client re-runs the handshake purely to refresh the two
 * names (T59): the server's display name and this device's name in its
 * table. The stream heartbeat covers the server name mid-stream, but a
 * device-side rename (an admin edit in the gateway's list) only reaches the
 * server→client direction when THIS side sends a request — the handshake's
 * marking header is always the gateway's freshest record. Deliberately quiet:
 * a failed refresh changes nothing (the heartbeat/idle clocks judge the
 * link), and the state never leaves `online`.
 */
const INFO_REFRESH_MS = 30_000

/**
 * The budget one size-backed exchange gets (invoke since T31-fix, upload
 * since T51): whole MiB only (floored), so a small call keeps the plain
 * base budget. The upload's size is the DECLARED `Content-Length` of the
 * local request when it carried one; an unknown size (a chunked body has no
 * length header, and this process's undici forbids hand-set length headers)
 * gets the full cap — the honest budget for an upload whose end is not in
 * sight.
 */
function sizeBackedTimeoutMs(requestTimeoutMs: number, bytes: number | undefined): number {
  const mib = bytes === undefined ? Number.POSITIVE_INFINITY : Math.floor(bytes / (1024 * 1024))
  return Math.min(requestTimeoutMs + INVOKE_TIMEOUT_PER_MIB_MS * mib, Math.max(requestTimeoutMs, INVOKE_TIMEOUT_CAP_MS))
}

/** The budget one invoke exchange gets, from its serialized body size. */
function invokeTimeoutMs(requestTimeoutMs: number, bodyJson: string): number {
  return sizeBackedTimeoutMs(requestTimeoutMs, Buffer.byteLength(bodyJson, 'utf8'))
}

/** 0 or less disables the periodic refresh entirely (tests that count
 * handshakes inject 0; the T59 refresh test injects the real cadence). */
function infoRefreshDelay(options: CreateRelayClientOptions): number {
  return options.infoRefreshMs ?? INFO_REFRESH_MS
}

/**
 * The automatic reconnect ladder (T43): after the client lands `offline` it
 * retries after 1s, 2s, 5s, 10s, 30s — and then every 30s until the link
 * comes back. Each wait gains 0–20% of random jitter so a fleet of clients
 * that lost the same server does not retry in lockstep. A success back to
 * `online` resets the ladder to the first step.
 */
const RETRY_STEPS_MS = [1_000, 2_000, 5_000, 10_000, 30_000] as const

/** Upper bound of the jitter fraction added to one retry delay (0–20%). */
const RETRY_JITTER = 0.2

/**
 * The clock face the reconnect machinery runs on: wall time, timers and the
 * jitter source. Injectable so tests drive the whole backoff sequence
 * deterministically. The default timers are `unref()`ed — a pending retry
 * must never keep the host process alive on its own.
 */
export interface RelayClock {
  /** Epoch milliseconds. */
  now(): number
  setTimeout(fn: () => void, ms: number): unknown
  clearTimeout(handle: unknown): void
  /** One random sample in [0, 1] — the jitter source. */
  random(): number
}

/**
 * The default clock: real time, `unref()`ed timers, `Math.random`. Exported
 * so the interceptor's own waits (the merged-stream reopen delays) run on
 * the SAME clock face — one injectable seam for tests instead of two.
 */
export const defaultClock: RelayClock = {
  now: () => Date.now(),
  setTimeout: (fn, ms) => {
    const handle = setTimeout(fn, ms)
    if (typeof handle.unref === 'function') handle.unref()
    return handle
  },
  clearTimeout: (handle) => clearTimeout(handle as ReturnType<typeof setTimeout>),
  random: () => Math.random(),
}

/** The connection states the settings surface renders (and `subscribe`
 * listeners react to). */
export type RelayState = 'unpaired' | 'connecting' | 'online' | 'offline' | 'revoked' | 'incompatible'

/** The handshake reply, verbatim from `POST relay/v1/handshake`. */
export interface RelayHandshake {
  relayProtocol: number
  serverId: string
  serverName: string
  dshVersion: string
  fingerprints: Record<string, string>
  /**
   * The SERVER's record of THIS device's name (T59) — what the pairing
   * registered and renames have since made of it. `undefined` from an older
   * server that does not send the field; the client then keeps whatever it
   * has and pushes nothing.
   */
  deviceName?: string
  /**
   * The server's per-plugin bridge table (PB-2), verbatim from the
   * handshake. `undefined` from an OLDER server whose handshake carries no
   * `plugins` field at all — and that distinction is load-bearing:
   * {@link compareBridges} reads `undefined` as `server-too-old`, while an
   * empty table means "this server speaks the field, nothing registered"
   * (`not-on-server`). Never normalized into `{}` here.
   */
  plugins?: Record<string, ServerPluginEntry>
}

/**
 * One relay failure. `code` is the mapped reason (the contract's state
 * vocabulary, a server refusal like `not-shared`, or a DSH error code from
 * a 200 `{ok:false}` envelope); `status` carries the HTTP status when a
 * response existed. `reason` is the OPTIONAL structured detail some error
 * frames carry beside the message — today only the server's `unshared`
 * stream-closure frame, whose `reason` ('manual' | 'client' | 'idle', T34)
 * the interceptor's closed-session display keys on. Fields are assigned in
 * the constructor body rather than declared as parameter properties: Node's
 * strip-only type mode rejects that syntax (the same rule as UploadError in
 * index.ts).
 */
export class RelayError extends Error {
  code: string
  status?: number
  reason?: string
  /** The host error's sanitized `details` (invoke path): the UI maps
   *  `details.reason` codes like `MODEL_DOES_NOT_SUPPORT_IMAGES` onto a
   *  user-facing line — absent details stay undefined. */
  details?: Record<string, unknown>

  constructor(code: string, message?: string, httpStatus?: number, reason?: string, details?: Record<string, unknown>) {
    super(message === undefined ? code : message)
    this.name = 'RelayError'
    this.code = code
    if (httpStatus !== undefined) this.status = httpStatus
    if (reason !== undefined) this.reason = reason
    if (details !== undefined) this.details = details
  }
}

/** The error body's structured `details`, when it is a record of scalars —
 *  the shape the server's own sanitizer emits. */
function detailsOf(error: Record<string, unknown>): Record<string, unknown> | undefined {
  const raw = error.details
  if (!isRecord(raw)) return undefined
  const out: Record<string, unknown> = {}
  for (const [key, value] of Object.entries(raw)) {
    if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean' || value === null) out[key] = value
  }
  return Object.keys(out).length > 0 ? out : undefined
}

export interface CreateRelayClientOptions {
  /** The normalized server address (no trailing slash), read live per
   * request; `undefined` counts as unpaired. */
  getServerUrl: () => string | undefined
  /** The pairing token, read live per request; `undefined` counts as
   * unpaired. */
  getToken: () => string | undefined
  /** Outbound fetch, defaulting to the global one. */
  fetchImpl?: typeof fetch
  /** How long a stream may go without receiving ANY line (frames and
   * `ping`s alike) before it is judged dead and aborted; default 45000. */
  idleTimeoutMs?: number
  /** How long one request/response round-trip (handshake, invoke — and the
   * headers of a stream open) may take; default 15000. */
  requestTimeoutMs?: number
  /** Clock/timers/jitter for the reconnect backoff; defaults to the real
   * ones (timers `unref()`ed). Tests inject a manual clock. */
  clock?: RelayClock
  /**
   * How often an ONLINE client re-runs the handshake to refresh the two
   * names (T59, {@link INFO_REFRESH_MS}); `0` disables the periodic refresh
   * — handshake-counting tests inject 0, the refresh test injects a cadence
   * its fake clock can drive.
   */
  infoRefreshMs?: number
  /** This side's own interface fingerprints (T42), computed after each
   * completed handshake and compared group by group against the handshake's
   * map. May be sync or async; a throw counts as "nothing computed" (an
   * empty map — every group lands `unavailable`). Absent: `compat` stays
   * undefined and no comparison ever runs (the host role's client, tests). */
  computeOwnFingerprints?: () => Promise<Record<string, string>> | Record<string, string>
}

export interface RelayClient {
  readonly state: RelayState
  /** The last completed handshake, undefined until one succeeds. */
  readonly handshakeInfo: RelayHandshake | undefined
  /** Digest ({@link relayCredentialsDigest}) of the address+token the last
   * completed handshake actually used, undefined until one succeeds. The
   * status route compares it against the CURRENT row credentials to keep a
   * cached "online" verdict from outliving the credentials it was earned
   * with. */
  readonly lastHandshakeDigest: string | undefined
  /** Epoch ms of the next automatic reconnect, or null while none is
   * scheduled (connected, attempting, or in a state that never reconnects).
   * Rendered by the settings page as "离线，将于 N 秒后重试". */
  readonly nextRetryAt: number | null
  /** The error CODE of the most recent failure ('offline',
   * 'relay-unauthorized', a DSH code, …) — never a message, never a token or
   * URL. Cleared when a request succeeds again. */
  readonly lastError: string | undefined
  /** The interface-compatibility verdict of the most recent handshake (T42):
   * group names that matched, differed, or could not be compared. Undefined
   * until a first handshake ran WITH `computeOwnFingerprints` wired, and
   * cleared again whenever the link or the credentials move (unpaired,
   * revoked, credentialsChanged) — it describes the credentials it was
   * earned with, never the current ones. */
  readonly compat: RelayCompatVerdict | undefined
  /** Observe state changes; a throwing listener never blocks the others.
   * The CURRENT state is also broadcast when only the handshake identity
   * moved (the T59 heartbeat names) — listeners re-read the getters. */
  subscribe(listener: (state: RelayState) => void): () => void
  /** The row's `serverName` was just committed (T59): forward it to the
   * gateway's device table when it differs from the server's record —
   * immediately while online, queued for the next `online` otherwise. A
   * name equal to the server's record is a no-op BY CONTRACT: that is the
   * settings page's server-driven write echoing back, and pushing it would
   * set the two ends overwriting each other. */
  queueDeviceName(name: string): void
  /** Whether a locally-pushed rename is queued or in flight (T59-fix). The
   * status route answers an EMPTY `deviceName` while this is true — an
   * in-flight answer could still carry the pre-push record, and the page
   * following it would bounce the fresh save back to the old name. */
  readonly deviceNameSyncing: boolean
  /** Run the handshake; success resolves with it and leaves `online`. */
  connect(): Promise<RelayHandshake>
  /** One immediate connection attempt from `offline`, resetting the backoff
   * ladder (the settings page's 立即重连). Answers whether an attempt was
   * started — the reconnect route turns a false into a 409. */
  reconnect(): boolean

  /** ONE lightweight authenticated health probe
   * (`GET /ping`). Never changes `state`; de-duplicated while in flight;
   * resolves false on any refusal. NOT a poller — call it on a real signal. */
  probeHealth(): Promise<boolean>

  /** a CONTROLLED fast-path for a reliable
   * "network is back" signal. Fires only from `offline`, refuses within
   * NETWORK_HINT_MIN_GAP_MS of the previous hint (no storm), and otherwise
   * behaves exactly like reconnect(). connect() de-duplicates concurrent
   * attempts via connectInFlight. Answers whether an attempt started. */
  networkRestored(): boolean
  /** Notify that the credentials source may have changed (the loader's
   * volatile-update for the row): a pending reconnect wait is cancelled and,
   * with credentials present, one immediate attempt runs with the new
   * values; without them the client lands `unpaired`. */
  credentialsChanged(): void
  /** Stop the automatic reconnect machinery (plugin row teardown). Explicit
   * connect() calls still work, but no timer is armed again. */
  stop(): void
  /** One invoke round-trip; resolves with the unwrapped `value`, throws
   * RelayError otherwise. A caller abort surfaces as `RelayError('aborted')`. */
  invoke(namespace: string, method: string, args: unknown, signal?: AbortSignal): Promise<unknown>
  /** Close one session's remote access on the SERVER (T34): the original
   * (non-virtual) session id rides `POST relay/v1/unshare`, the server
   * unshares it with reason `'client'`. Resolves on the success envelope;
   * every refusal (`not-shared`, a wall, transport death) throws RelayError.
   * A success proves the link and lifts a stale `offline` back to `online`,
   * exactly like invoke. */
  unshare(sessionId: string, signal?: AbortSignal): Promise<void>
  /** Answer one forwarded Remote event (T32): `eventId` is the ORIGINAL id
   * (the interceptor swapped the virtual one back), `result` the Remote
   * event OUTCOME, forwarded verbatim — the gateway validates it. The error
   * mapping is invoke's: a success envelope resolves (with the value, in
   * practice undefined), everything else throws RelayError. */
  postEventResult(eventId: string, result: unknown, signal?: AbortSignal): Promise<unknown>
  /** Open the NDJSON stream route. `frame` lines are yielded, `ping` lines
   * only refresh the idle clock, `end` finishes the iteration, an `error`
   * line throws its RelayError. A caller abort ENDS the iteration normally;
   * `break` aborts the underlying request. */
  openStream(namespace: string, method: string, args: unknown, signal?: AbortSignal): AsyncIterable<unknown>
  /** One plain-HTTP round-trip (T41b): `route` / `query` name a registered
   * `/api` GET (relay-access's RELAY_HTTP_ROUTES; `query` is the URL-encoded
   * query string with the session id already restored). Resolves with the
   * UPSTREAM answer — its status, content type and body text ride inside the
   * success envelope, so a 404 from the underlying route is a RESOLVED
   * result here, never a RelayError. */
  http(route: string, query: string, signal?: AbortSignal): Promise<RelayHttpResult>
  /** One binary upload round-trip (T51): `sessionId` is the ORIGINAL
   * session id, `name` the optional display filename, `body` the raw byte
   * stream forwarded verbatim. The upstream answer (its status, content
   * type and body text — the host route answers its business failures as
   * 200-with-envelope, so those ride RESOLVED like `http`'s) comes back in
   * the success envelope; every refusal or link death throws RelayError.
   * The round-trip budget scales with `bytes` (the local request's declared
   * `Content-Length`, when it had one) and a timeout there never moves the
   * connection state — the invoke route's rules (T31-fix). */
  upload(options: RelayUploadOptions, signal?: AbortSignal): Promise<RelayHttpResult>
  /** One bridged-plugin round-trip (PB-4): the coordinates travel in the
   * x-zr-* headers (PB-3's protocol), no body. Resolves with the UNREAD
   * fetch Response the moment the headers arrive — the caller streams the
   * body itself — and EVERY HTTP status is carried verbatim (a refusal's
   * `x-zen-remote-error` header is the caller's classifier, never an
   * exception here). The header wait gets the plain request timeout; the
   * BODY carries no deadline of our own (a stream takes as long as it
   * takes) — the caller's signal aborts it. Transport deaths judge the
   * link like every exchange (offline); `requireCredentials` throws the
   * usual RelayError when nothing is configured. */
  pluginRequest(request: RelayPluginRequest, signal?: AbortSignal): Promise<Response>
  /** Open one WebSocket byte tunnel (PB-10): the coordinates travel like a
   * pluginRequest's, plus the browser handshake's `sec-websocket-*` headers
   * and any early head bytes (base64). Resolves once the NDJSON stream's
   * FIRST line named the tunnel id; `frames` then yields the upstream's
   * raw bytes (each `data` line decoded) and ENDS when the server writes
   * the `end` line (upstream closed or errored). A refusal (any non-200 or
   * a missing open line) throws RelayError with the wire's
   * `x-zen-remote-error` code. */
  pluginTunnelOpen(
    request: RelayPluginRequest & {
      wsHeaders: { 'sec-websocket-key': string; 'sec-websocket-version': string; 'sec-websocket-protocol'?: string; 'sec-websocket-extensions'?: string }
      head?: string
    },
    signal?: AbortSignal,
  ): Promise<{ tunnel: string; frames: AsyncIterable<Uint8Array> }>
  /** One ordered upstream chunk (PB-10): `seq` must count from 0 without
   * gaps — the server answers 409 `tunnel-out-of-order` and closes the
   * tunnel otherwise. Resolves after the server took the bytes (drain
   * included). */
  pluginTunnelSend(tunnel: string, seq: number, bytes: Uint8Array, signal?: AbortSignal): Promise<void>
  /** Idempotent teardown (PB-10). */
  pluginTunnelClose(tunnel: string, signal?: AbortSignal): Promise<void>
}

/** The inputs of one {@link RelayClient.pluginRequest} round-trip (PB-4) —
 * the PB-3 wire protocol's coordinates, already validated and virtual-id-
 * restored by the caller. */
export interface RelayPluginRequest {
  plugin: string
  version: string
  digest: string
  /** The declared route path, verbatim (never decoded). */
  path: string
  /** The re-serialized query string, session id already restored to its
   * ORIGINAL form. */
  query: string
  /** The request method (PB-9); defaults to GET. Must be one the route
   * declared — the server answers 405 `method-undeclared` otherwise. */
  method?: string
  /** The request BODY (PB-9), forwarded verbatim as a stream. `null` means
   * no body. A streaming body rides chunked (undici forbids hand-set length
   * headers), so the server's streaming byte count is the size gate. */
  body?: ReadableStream<Uint8Array> | null
  /** The body's content type, passed through to the plugin when set. */
  contentType?: string
  /** The declared body size, when the local request carried one — sizes the
   * header-wait budget and nothing else (a streaming body goes chunked). */
  contentLength?: number
  /** The one request header passed through to the plugin beside `range`. */
  accept?: string
  range?: string
}

/** The inputs of one {@link RelayClient.upload} round-trip. */
export interface RelayUploadOptions {
  /** The ORIGINAL (non-virtual) session id, already restored by the caller. */
  sessionId: string
  /** The file's display name, when the local request carried one. */
  name?: string
  /** The raw request body. `null` (a bodyless upload — the host route would
   * answer its own empty-stream business failure) forwards as an empty
   * stream. */
  body: ReadableStream<Uint8Array> | null
  /** The local request's declared byte count, when its `Content-Length` was
   * readable — sizes the round-trip budget and nothing else. */
  bytes?: number
}

/** The upstream answer one {@link RelayClient.http} round-trip carries. */
export interface RelayHttpResult {
  /** The underlying `/api` route's HTTP status. */
  status: number
  /** The underlying response's content type, when it sent one. */
  contentType: string | undefined
  /** The underlying response body, decoded as text (both registered routes
   * answer buffered JSON). */
  body: string
}

/**
 * The credential digest recorded beside a completed handshake (T23a-fix):
 * sha256 over `url + "\n" + token`, first 16 hex chars. Enough to detect a
 * changed address or a re-pair, without keeping — or ever exposing — a
 * plaintext copy of either value.
 */
export function relayCredentialsDigest(serverUrl: string, token: string): string {
  return createHash('sha256').update(`${serverUrl}\n${token}`).digest('hex').slice(0, 16)
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === 'object' && value !== null && !Array.isArray(value)
}

/**
 * Whether the loader's `loader/volatile-update` announcement names the
 * device's own field (T59-fix): the announcement rides EVERY volatile
 * commit — an unrelated knob, or the settings page's own follow write — and
 * only a commit that actually moved `serverName` may queue a push. The
 * paths are the loader's changed-field lists (`[['serverName'], …]`).
 */
export function volatileUpdateTouchesServerName(paths: unknown): boolean {
  return Array.isArray(paths) && paths.some((entry) => Array.isArray(entry) && entry[0] === 'serverName')
}

function messageOf(error: unknown): string {
  return error instanceof Error ? error.message : String(error)
}

/**
 * One response body, parsed leniently. Reading the text is allowed to
 * REJECT (a connection cut mid-body is a transport failure the caller maps
 * to `offline`); only the JSON step is lenient — an HTML wall or any other
 * non-JSON body becomes `undefined`, which the classifier then treats as
 * "no usable body".
 */
async function readPayload(response: Response): Promise<unknown> {
  const text = await response.text()
  if (text === '') return undefined
  try {
    return JSON.parse(text)
  } catch {
    return undefined
  }
}

/** One parsed NDJSON line's verdict. */
type LineVerdict =
  | { kind: 'none' }
  | { kind: 'frame'; frame: unknown }
  | { kind: 'end' }
  | { kind: 'error'; error: RelayError }
  | { kind: 'info'; serverName?: string; deviceName?: string }

/** A string field of an info line, or undefined — absent, blank or not a string. */
function infoString(value: unknown): string | undefined {
  return typeof value === 'string' && value !== '' ? value : undefined
}

function classifyLine(line: string): LineVerdict {
  let parsed: unknown
  try {
    parsed = JSON.parse(line)
  } catch {
    return { kind: 'none' } // a damaged line is skipped, never fatal
  }
  if (!isRecord(parsed)) return { kind: 'none' }
  if (parsed.type === 'frame' && 'frame' in parsed) return { kind: 'frame', frame: parsed.frame }
  if (parsed.type === 'end') return { kind: 'end' }
  if (parsed.type === 'error') {
    const detail = isRecord(parsed.error) ? parsed.error : {}
    const code = typeof detail.code === 'string' && detail.code !== '' ? detail.code : 'internal'
    const message = typeof detail.message === 'string' ? detail.message : undefined
    // The structured detail beside the message (T34): today only the
    // `unshared` frame carries one. An absent or non-string reason stays
    // undefined — the consumer degrades to its own default.
    const reason = typeof detail.reason === 'string' ? detail.reason : undefined
    return { kind: 'error', error: new RelayError(code, message, undefined, reason) }
  }
  // T59: a ping may carry the names as of NOW (`serverName` — the server's
  // display name; `deviceName` — this device's name in its table). An old
  // server's bare `{"type":"ping"}` carries neither and stays `none`.
  if (parsed.type === 'ping') {
    const serverName = infoString(parsed.serverName)
    const deviceName = infoString(parsed.deviceName)
    if (serverName !== undefined || deviceName !== undefined) return { kind: 'info', serverName, deviceName }
  }
  // Any shape the protocol does not define — ignored.
  return { kind: 'none' }
}

/**
 * Build one relay client. Pure state machine + fetch plumbing; the row is
 * only ever seen through the two getters.
 */
export function createRelayClient(options: CreateRelayClientOptions): RelayClient {
  const idleTimeoutMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS
  const requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS
  const fetchImpl: typeof fetch = options.fetchImpl ?? ((url, init) => fetch(url, init))
  const clock: RelayClock = options.clock ?? defaultClock

  let state: RelayState = 'unpaired'
  let handshakeInfo: RelayHandshake | undefined
  let compat: RelayCompatVerdict | undefined
  let lastHandshakeDigest: string | undefined
  let connectInFlight: Promise<RelayHandshake> | undefined
  // T43 reconnect machinery: the armed timer handle, how many consecutive
  // attempts have failed (the ladder index), when the armed timer fires, the
  // code of the most recent failure, and whether the row teardown stopped
  // the machinery.
  let retryTimer: unknown | undefined
  let retryAttempt = 0
  let nextRetryAt: number | null = null
  let lastError: string | undefined
  let stopped = false
  // Digest of the credentials the machinery last SAW — the change detector
  // behind credentialsChanged().
  const initialCredentials = currentCredentials()
  let credentialsDigest =
    initialCredentials === undefined ? undefined : relayCredentialsDigest(initialCredentials.url, initialCredentials.token)
  const listeners = new Set<(state: RelayState) => void>()

  /** One broadcast of the CURRENT state to every subscriber; a throwing
   * listener never blocks the rest. State transitions are the only trigger —
   * including the handshake's connecting→online pair, which is how the T42
   * compat verdict (stored just before `online`) reaches the settings
   * surface: woken listeners read the getter fresh. */
  function notify(): void {
    for (const listener of [...listeners]) {
      try {
        listener(state)
      } catch {
        // One broken listener never blocks the rest.
      }
    }
  }

  function setState(next: RelayState): void {
    if (state === next) return
    state = next
    // The reconnect hooks ride the transitions (T43): landing offline arms
    // the next wait; online resets the ladder; the three user-action states
    // never reconnect and drop any pending wait; an attempt start leaves the
    // ladder alone but the wait display is fireRetry's business.
    if (next === 'offline') scheduleRetry()
    else if (next === 'online') {
      resetRetry()
      // T59: a server confirmation retires the local push authority (the
      // handshake/stream/invoke that just landed saw the table as it is),
      // the periodic name refresh arms, and a rename queued while offline
      // goes out now.
      deviceNameAuthoritative = false
      armInfoRefresh()
      flushPendingDeviceName()
    }
    else if (next === 'unpaired' || next === 'revoked' || next === 'incompatible') cancelRetry()
    // A compat verdict describes the server the CURRENT credentials pointed
    // at. Unpairing (or a revocation — the same "this server is gone" wall)
    // invalidates that answer, so the verdict goes with the link (T42-fix).
    if (next === 'unpaired' || next === 'revoked') compat = undefined
    notify()
  }

  /**
   * The live credentials, or undefined — the requireCredentials logic
   * without the announcement, shared with credentialsChanged's change
   * detection.
   */
  function currentCredentials(): { url: string; token: string } | undefined {
    const url = options.getServerUrl()
    const token = options.getToken()
    if (url === undefined || url.trim() === '' || token === undefined || token === '') return undefined
    // The getter contract delivers a normalized address; stripping stray
    // trailing slashes anyway keeps a hand-edited row from silently producing
    // `…/relay//v1/…`, which no route would ever match.
    return { url: url.trim().replace(/\/+$/u, ''), token }
  }

  /**
   * The live credentials, or the unpaired failure. Never a request without
   * both halves — and the unpaired state is (re)announced, so a row whose
   * address or token was cleared later shows up as unpaired too.
   */
  function requireCredentials(): { url: string; token: string } {
    const creds = currentCredentials()
    if (creds === undefined) {
      setState('unpaired')
      throw new RelayError('unpaired', 'no server address or pairing token is configured')
    }
    credentialsDigest = relayCredentialsDigest(creds.url, creds.token)
    return creds
  }

  // ---- the reconnect ladder (T43) ------------------------------------------

  function cancelRetry(): void {
    if (retryTimer !== undefined) {
      clock.clearTimeout(retryTimer)
      retryTimer = undefined
    }
    retryAttempt = 0
    nextRetryAt = null
  }

  function resetRetry(): void {
    cancelRetry()
    lastError = undefined
  }

  function scheduleRetry(): void {
    // One wait at a time — a second offline entry while a timer is armed
    // (e.g. a status-route connect failing between two ticks) leaves the
    // armed timer in charge.
    if (stopped || retryTimer !== undefined) return
    const base = RETRY_STEPS_MS[Math.min(retryAttempt, RETRY_STEPS_MS.length - 1)]
    const delay = Math.round(base * (1 + clock.random() * RETRY_JITTER))
    retryAttempt += 1
    nextRetryAt = clock.now() + delay
    retryTimer = clock.setTimeout(fireRetry, delay)
  }

  function fireRetry(): void {
    retryTimer = undefined
    nextRetryAt = null
    if (stopped || state !== 'offline') return
    // The attempt's own outcome drives the machine: success → online (ladder
    // reset), a still-offline failure → scheduleRetry armed the next step,
    // a wall (revoked/incompatible/unpaired) → cancelRetry ran.
    void connect().catch(() => {})
  }

  /**
   * The error-code vocabulary the diagnostics readout accepts (T43-fix):
   * short, alphanumeric with / _ - separators. A server that answers a
   * "code" quoting URLs or anything stranger does not get to write it into
   * the client's state — it degrades to `unexpected`.
   */
  function sanitizeCode(code: string): string {
    return /^[a-z0-9/_-]{1,64}$/i.test(code) ? code : 'unexpected'
  }

  /** Record the failing call's code for the diagnostics surface (the CODE
   * only — messages quote URLs, and codes never carry credentials). */
  function noteFailure(code: string): void {
    lastError = sanitizeCode(code)
  }

  function subscribe(listener: (state: RelayState) => void): () => void {
    listeners.add(listener)
    return () => {
      listeners.delete(listener)
    }
  }

  // ---- the T59 device-name sync ----------------------------------------------

  /**
   * True while THIS side's `deviceName` is the authority: a rename this
   * client pushed that the gateway just answered OK to. In-flight stream
   * heartbeats still carry the OLD name (their marking header was written
   * when the stream opened), and without this flag those pings would pull
   * `handshakeInfo` back to the stale name — and the settings page would
   * then "follow" it straight over the user's fresh save. The next server
   * confirmation (handshake or refresh) retires the flag: the answer there
   * IS the table's current record.
   */
  let deviceNameAuthoritative = false

  /** A rename waiting for the link: queued while offline (or before the
   * first handshake), flushed once `online` lands. */
  let pendingDeviceName: string | undefined

  // T59-fix: a push that is IN FLIGHT right now, plus a sequence counter
  // bumped at every push start AND settle. An info-refresh answer that
  // overlapped a push (seq moved, or a push is still queued) carries the
  // PRE-push record and must not win — see {@link refreshInfo}.
  let deviceNamePushBusy = false
  let deviceNamePushSeq = 0

  /**
   * Whether a locally-pushed rename is queued or in flight (T59-fix). The
   * status route answers an EMPTY `deviceName` while this is true: an
   * in-flight status answer could still carry the pre-push record, and a
   * page that followed it would bounce the user's fresh save straight back
   * to the old name.
   */
  function deviceNameSyncing(): boolean {
    return deviceNamePushBusy || pendingDeviceName !== undefined
  }

  /** The periodic info-refresh's armed timer (see {@link INFO_REFRESH_MS}). */
  let infoTimer: unknown | undefined

  /** Fold one heartbeat's names into the stored handshake (T59). */
  function applyInfoDelta(serverName: string | undefined, deviceName: string | undefined): void {
    if (handshakeInfo === undefined) return
    const next = { ...handshakeInfo }
    let changed = false
    if (serverName !== undefined && serverName !== next.serverName) {
      next.serverName = serverName
      changed = true
    }
    // A locally-pushed rename outranks the stream heartbeat until the server
    // confirms it — see {@link deviceNameAuthoritative}; a push IN FLIGHT is
    // equally authoritative for the same reason (T59-fix): the heartbeat's
    // name was stamped before our rename reached the table.
    if (deviceName !== undefined && deviceName !== next.deviceName && !deviceNameAuthoritative && !deviceNamePushBusy) {
      next.deviceName = deviceName
      changed = true
    }
    if (!changed) return
    handshakeInfo = next
    // The state itself did not move, but the identity did: subscribers (the
    // interceptor's rename path above all) must see the new names. The
    // current-state broadcast is exactly that — no reconnect can come from
    // it, because only setState moves the machine.
    notify()
  }

  /**
   * Push one name to the gateway's `device/name` endpoint (T59): the gateway
   * renames the CALLING device's own record. Resolves `true` when the table
   * now carries `name` (the push answered OK) or when no push was needed;
   * a failed push queues the name for the next `online` and resolves false.
   * Never touches the connection state — a refused rename is not a dead
   * link.
   */
  async function pushDeviceName(name: string): Promise<boolean> {
    const creds = currentCredentials()
    if (creds === undefined) {
      pendingDeviceName = name
      return false
    }
    // T59-fix: the busy window covers the whole round-trip (the status route
    // answers an empty deviceName while it lasts), and the sequence counter
    // lets an overlapping info refresh know its answer is stale.
    const digestAtStart = credentialsDigest
    deviceNamePushBusy = true
    deviceNamePushSeq += 1
    let payload: unknown
    try {
      const controller = new AbortController()
      const timer = setTimeout(() => controller.abort(), requestTimeoutMs)
      if (typeof timer.unref === 'function') timer.unref()
      try {
        const response = await fetchImpl(`${creds.url}${DEVICE_NAME_PATH}`, {
          ...requestInit(creds.token, JSON.stringify({ name })),
          signal: controller.signal,
        })
        if (response.ok) payload = await readPayload(response)
      } finally {
        clearTimeout(timer)
      }
    } catch {
      payload = undefined
    }
    deviceNamePushBusy = false
    deviceNamePushSeq += 1
    // T59-fix2: TWO different endings hide behind the old combined guard.
    // A stopped client has nothing to come back for, and an answer about
    // DIFFERENT credentials describes a table we no longer talk to — those
    // give up outright (a later row commit re-queues a fresh name). A
    // merely dropped link is NOT the end: the name re-queues and the next
    // `online` transition flushes it, exactly like a push that was refused
    // before it ever left.
    if (stopped || credentialsDigest !== digestAtStart) return false
    if (state !== 'online') {
      pendingDeviceName = name
      return false
    }
    const confirmed = isRecord(payload) && payload.ok === true && infoString(payload.name) === name
    if (!confirmed) {
      pendingDeviceName = name
      return false
    }
    // The table took the new name: remember it locally (without a state
    // announcement — nothing but the name moved) and mark it authoritative
    // until the server's own answer says the same.
    if (handshakeInfo !== undefined && handshakeInfo.deviceName !== name) {
      handshakeInfo = { ...handshakeInfo, deviceName: name }
    }
    deviceNameAuthoritative = true
    pendingDeviceName = undefined
    return true
  }

  /** Flush a queued rename against the fresh online state (T59). */
  function flushPendingDeviceName(): void {
    if (pendingDeviceName === undefined) return
    const name = pendingDeviceName
    pendingDeviceName = undefined
    if (handshakeInfo === undefined) {
      pendingDeviceName = name
      return
    }
    if (name === (handshakeInfo.deviceName ?? '')) return
    void pushDeviceName(name)
  }

  /**
   * The backend's one entry point (T59): the row's `serverName` was just
   * committed. Online with a handshake, the name travels to the gateway only
   * when it DIFFERS from the server's record — the settings page's
   * server-driven write lands here with both sides equal, which is exactly
   * what keeps the two ends from overwriting each other. Offline (or not yet
   * handshaken), the name queues for the next `online`.
   */
  function queueDeviceName(name: string): void {
    const trimmed = name.trim()
    if (trimmed === '') return
    if (state !== 'online' || handshakeInfo === undefined) {
      pendingDeviceName = trimmed
      return
    }
    if (trimmed === (handshakeInfo.deviceName ?? '')) {
      pendingDeviceName = undefined
      return
    }
    void pushDeviceName(trimmed)
  }

  /**
   * Canonical JSON for CHANGE DETECTION (PB-2-fix): every object's keys are
   * sorted, recursively, so two tables that differ only in key order — the
   * server's map insertion order AND each entry's field order — compare
   * equal. Anything else (values, entry sets) differs, and JSON.stringify
   * answers scalars verbatim.
   */
  function canonicalJson(value: unknown): string {
    if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]`
    if (value !== null && typeof value === 'object') {
      const record = value as Record<string, unknown>
      return `{${Object.keys(record)
        .sort()
        .map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`)
        .join(',')}}`
    }
    return JSON.stringify(value)
  }

  /**
   * One QUIET handshake re-run (T59): refreshes the two names while online —
   * no state announcement, no ladder movement, failures silent (the next
   * tick retries; the idle and heartbeat clocks own the link's verdict).
   * The reply carries the gateway's freshest record of BOTH names (its
   * marking headers are written per forward), so this is also how a rename
   * an admin made on the server reaches an otherwise-idle client.
   */
  async function refreshInfo(): Promise<void> {
    if (stopped || state !== 'online' || connectInFlight !== undefined) return
    const creds = currentCredentials()
    if (creds === undefined) return
    // T59-fix: remember the world at request time — the answer only counts
    // if we are still the same online client (the credential digest catches
    // a re-pair to another server), and a name push that overlapped the
    // round-trip makes the answer's deviceName a PRE-push record.
    const digestAtStart = credentialsDigest
    const pushSeqAtStart = deviceNamePushSeq
    let payload: unknown
    try {
      const controller = new AbortController()
      const timer = setTimeout(() => controller.abort(), requestTimeoutMs)
      if (typeof timer.unref === 'function') timer.unref()
      try {
        const response = await fetchImpl(`${creds.url}${HANDSHAKE_PATH}`, {
          ...requestInit(creds.token, '{}'),
          signal: controller.signal,
        })
        if (!response.ok) return
        payload = await readPayload(response)
      } finally {
        clearTimeout(timer)
      }
    } catch {
      return
    }
    if (stopped || state !== 'online' || credentialsDigest !== digestAtStart) return
    if (!isRecord(payload) || payload.ok !== true || payload.relayProtocol !== RELAY_PROTOCOL) return
    // T59-fix: NAMES ONLY. serverId / dshVersion / fingerprints belong to the
    // real handshake (the compat verdict is computed there, on both maps
    // fresh); a quiet refresh never disturbs them. PB-2 is the one addition:
    // the plugin bridge table rides along, because a consent change on the
    // server must reach an otherwise-idle client without a reconnect — and
    // unlike the compat verdict it is never cached as a conclusion, the
    // comparison runs live off this map. A refresh answer WITHOUT the field
    // keeps whatever the real handshake stored.
    const previous = handshakeInfo
    if (previous === undefined) return
    const serverName = typeof payload.serverName === 'string' ? payload.serverName : previous.serverName
    const serverDeviceName = infoString(payload.deviceName)
    // A push that overlapped this round-trip — STARTED before it (the seq
    // moved by the settle), still IN FLIGHT, or still QUEUED — makes the
    // answer's deviceName a pre-push record: dropped, and the local
    // authority stays; the next clean refresh confirms instead.
    const pushedDuringRefresh = deviceNamePushSeq !== pushSeqAtStart || deviceNamePushBusy || pendingDeviceName !== undefined
    const deviceName = pushedDuringRefresh ? undefined : serverDeviceName
    if (deviceName !== undefined) deviceNameAuthoritative = false
    // PB-2-fix: compare the incoming table by CONTENT, not by reference — the
    // payload was parsed fresh, so an unchanged table would otherwise read as
    // changed on every tick and wake every subscriber (the interceptor's
    // catalog listener re-pulls on each wake) for nothing. An unchanged table
    // keeps the PREVIOUS object: identity is the "nothing moved" proof the
    // changed flag reads.
    let plugins = previous.plugins
    if (isRecord(payload.plugins)) {
      const incoming = payload.plugins as Record<string, ServerPluginEntry>
      if (canonicalJson(incoming) !== canonicalJson(previous.plugins ?? {})) plugins = incoming
    }
    let changed = previous.serverName !== serverName
    if (deviceName !== undefined && deviceName !== previous.deviceName) changed = true
    if (plugins !== previous.plugins) changed = true
    if (!changed) return
    handshakeInfo = {
      ...previous,
      serverName,
      ...(deviceName !== undefined ? { deviceName } : {}),
      ...(plugins !== previous.plugins ? { plugins } : {}),
    }
    notify()
  }

  /** Arm one refresh tick; re-arms itself while the client stays online. */
  function armInfoRefresh(): void {
    const delay = infoRefreshDelay(options)
    if (delay <= 0 || infoTimer !== undefined || stopped) return
    infoTimer = clock.setTimeout(() => {
      infoTimer = undefined
      if (stopped || state !== 'online') return
      // T59-fix2: a rename whose push FAILED while the link stayed up waits
      // in the queue with no state transition coming to flush it — it
      // retries on this cadence, or it would hang forever: syncing (and
      // the empty `deviceName` the status route answers) would never clear,
      // and the next reconnect would push the stale queued name over
      // whatever an administrator renamed the device to meanwhile.
      flushPendingDeviceName()
      void refreshInfo().then(() => {
        if (state === 'online' && !stopped) armInfoRefresh()
      })
    }, delay)
  }

  /** The shared request face of every route. ONLY endpoint headers live
   * here — content-length, host, connection and transfer-encoding are the
   * dispatcher's business and must never be set by hand: DSH's process
   * swaps the global fetch dispatcher for its own undici, which rejects a
   * manual content-length outright (fetch failed / UND_ERR_INVALID_ARG).
   * The JSON body arrives PRE-SERIALIZED: the invoke caller needs the JSON
   * once for its size-based timeout, so every route stringifies exactly
   * one. */
  function requestInit(token: string, bodyJson: string): RequestInit {
    return {
      method: 'POST',
      headers: {
        authorization: `Bearer ${token}`,
        'content-type': 'application/json',
        // Load-bearing (module comment): without it the gateway's pairing
        // wall arrives as HTML and a revocation cannot be classified.
        accept: 'application/json',
      },
      body: bodyJson,
      // A wrong server must not walk the pairing token through a redirect.
      redirect: 'manual',
    }
  }

  /**
   * The binary upload's init (T51): the same endpoint-headers-only
   * discipline as {@link requestInit} — the body rides as the raw byte
   * stream with `content-type: application/octet-stream` (the ONE header
   * the host's upload route mandates, RT dsh-client-file-upload
   * lib/index.js:18) and no content-length: DSH's undici refuses a hand-set
   * length outright, so a stream body goes chunked and the server's
   * streaming byte count is the size gate. `duplex: 'half'` is what a
   * streaming request body requires.
   */
  function uploadInit(token: string, body: ReadableStream<Uint8Array>): RequestInit {
    const init: RequestInit = {
      method: 'POST',
      headers: {
        authorization: `Bearer ${token}`,
        'content-type': 'application/octet-stream',
        accept: 'application/json',
      },
      body,
      redirect: 'manual',
    }
    // `duplex` is what a streaming request body requires — the resolved
    // RequestInit type here hides it (the DOM-vs-undici conditional in
    // @types/node collapses to the DOM face, which has no such field), so
    // the assignment carries its own cast.
    ;(init as { duplex?: string }).duplex = 'half'
    return init
  }

  /**
   * Map one non-success response onto the RelayError (and state) contract.
   * Only the two walls move the state — the gateway's unpaired 401 means
   * the token died (`revoked`), the relay-only 403 means the far end is not
   * a 2.0.0 relay server (`incompatible`). The 401 `relay-unauthorized`
   * deliberately does NOT become `revoked`: the gateway ACCEPTED the token
   * and the server's internal secret disagrees — a server-side fault, and
   * unpairing a valid device would be the wrong lesson.
   */
  function failureOf(status: number, payload: unknown): RelayError {
    const body = isRecord(payload) ? payload : {}
    const error = isRecord(body.error) ? body.error : {}
    const code = typeof error.code === 'string' && error.code !== '' ? error.code : undefined
    const message = typeof error.message === 'string' ? error.message : undefined
    const reason = typeof body.reason === 'string' ? body.reason : undefined
    if (status === 401) {
      if (reason === 'unpaired') {
        setState('revoked')
        return new RelayError('revoked', 'the pairing token was rejected by the server gateway', 401)
      }
      if (code === 'relay-unauthorized') return new RelayError('relay-unauthorized', message, 401)
    }
    if (status === 403) {
      if (reason === 'relay-only') {
        setState('incompatible')
        return new RelayError('incompatible', 'the server refused the relay prefix — it is not running the 2.0.0 relay', 403)
      }
      if (code === 'not-shared' || code === 'no-session' || code === 'forbidden-method') {
        return new RelayError(code, message, 403)
      }
    }
    // 200 {ok:false} envelopes, 400/404/429 answers and anything else
    // unheard-of: the body's error code is the answer, the state is not
    // touched.
    if (code !== undefined) return new RelayError(code, message, status, undefined, detailsOf(error))
    // The bad-gateway band carries no relay body of its own — it is what a
    // reverse proxy answers when the gateway behind it is down (nginx: an
    // HTML 502). That IS the offline case, and the e2e contract expects it.
    if (status === 502 || status === 503 || status === 504) {
      noteFailure('offline')
      setState('offline')
      return new RelayError('offline', `the relay chain answered ${status} — the gateway is unreachable`, status)
    }
    return new RelayError(`http-${status}`, `unexpected relay response with status ${status}`, status)
  }

  /**
   * The server-restart stream line (the server App exits cleanly or reloads
   * its plugin row: relay-server.ts's closeAll ends every NDJSON stream with
   * `error{code:'server-restart'}`) is a LINK fact, not an answer about one
   * call — the whole server is going down. Mark the client offline exactly
   * like a transport death would: the reconnect ladder arms, the merged
   * streams' offline annotations apply, and the eventual reconnect runs a
   * fresh handshake that picks up the (possibly changed) server name. The
   * thrown RelayError keeps the specific code for the diagnostics ring.
   */
  function noteServerRestart(error: RelayError): void {
    if (error.code !== 'server-restart') return
    noteFailure('offline')
    setState('offline')
  }

  /** `aborted` — a caller's signal, not a transport fault. */
  function abortedError(): RelayError {
    return new RelayError('aborted', 'the caller aborted the request')
  }

  /**
   * One request/response exchange (handshake, invoke, upload) with the
   * shared timeout wiring. The URL and the RequestInit arrive PREBUILT —
   * the JSON routes compose theirs from {@link requestInit}, the upload
   * from {@link uploadInit} — so the body shape (serialized string or byte
   * stream) stays the caller's business. Resolves only on a success
   * envelope — 2xx plus `ok:true` — and then lifts a stale `offline` back
   * to `online`, because a success proves the link. EVERY other outcome
   * throws the mapped RelayError; transport failures (fetch rejection,
   * mid-body cut, timeout) set `offline` first — EXCEPT on a route that
   * opted out (`timeoutSetsOffline: false`, the invoke and upload routes):
   * a slow call — a 28 MiB inline-image prompt or a 100 MiB attachment
   * crawling up a slow link — and, since T51-fix, an upload CUT MID-STREAM
   * (a reset under a huge body is the size gate and the drain doing their
   * job, not a dead link) both fail THIS call and leave the connection
   * state exactly where it was; the diagnostics still record the failure
   * code. Only the handshake / stream-header legs and real network-layer
   * failures on the opted-in routes judge the link.
   */
  async function exchangeRequest(
    url: string,
    init: RequestInit,
    signal: AbortSignal | undefined,
    applySuccessState: boolean,
    opts: { timeoutMs?: number; timeoutSetsOffline?: boolean; transportSetsOffline?: boolean } = {},
  ): Promise<Record<string, unknown>> {
    const timeoutMs = opts.timeoutMs ?? requestTimeoutMs
    const timeoutSetsOffline = opts.timeoutSetsOffline ?? true
    // The transport-failure opt-out (T51-fix) is SEPARATE from the timeout
    // one: the invoke route opts out of the timeout judging the link but
    // keeps real transport deaths offline (the T43 reconnect ladder hangs
    // off them); the upload route opts out of BOTH — a reset under a huge
    // body is the size gate doing its job, not a dead link.
    const transportSetsOffline = opts.transportSetsOffline ?? true
    const controller = new AbortController()
    const onExternalAbort = (): void => {
      controller.abort()
    }
    let timedOut = false
    if (signal !== undefined) {
      if (signal.aborted) controller.abort()
      else signal.addEventListener('abort', onExternalAbort)
    }
    const timer = setTimeout(() => {
      timedOut = true
      controller.abort()
    }, timeoutMs)
    if (typeof timer.unref === 'function') timer.unref()
    try {
      let response: Response
      try {
        response = await fetchImpl(url, { ...init, signal: controller.signal })
      } catch (error) {
        if (signal?.aborted) throw abortedError()
        if (timedOut && !timeoutSetsOffline) {
          noteFailure('request-timeout')
          throw new RelayError('request-timeout', `no response within ${timeoutMs} ms`)
        }
        noteFailure('offline')
        if (transportSetsOffline) setState('offline')
        throw new RelayError('offline', timedOut ? `no response within ${timeoutMs} ms` : messageOf(error))
      }
      let payload: unknown
      try {
        payload = await readPayload(response)
      } catch (error) {
        if (signal?.aborted) throw abortedError()
        noteFailure('offline')
        if (transportSetsOffline) setState('offline')
        throw new RelayError('offline', messageOf(error))
      }
      if (response.status >= 200 && response.status < 300 && isRecord(payload) && payload.ok === true) {
        // A success answers the link question: the last failure is superseded.
        lastError = undefined
        if (applySuccessState) setState('online')
        return payload
      }
      throw failureOf(response.status, payload)
    } finally {
      clearTimeout(timer)
      if (signal !== undefined) signal.removeEventListener('abort', onExternalAbort)
    }
  }

  /** The JSON routes' one exchange: path + pre-serialized body over
   * {@link exchangeRequest}. */
  async function exchange(
    pathName: string,
    bodyJson: string,
    signal: AbortSignal | undefined,
    applySuccessState: boolean,
    creds: { url: string; token: string },
    opts: { timeoutMs?: number; timeoutSetsOffline?: boolean } = {},
  ): Promise<Record<string, unknown>> {
    return exchangeRequest(`${creds.url}${pathName}`, requestInit(creds.token, bodyJson), signal, applySuccessState, opts)
  }

  function connect(): Promise<RelayHandshake> {
    // A connect already running IS the answer (T23a-fix2): a caller that
    // raced one — the status route abandons its wait after 5s while the
    // attempt keeps going — joins the SAME promise instead of stacking a
    // second handshake on the wire. The slot clears the moment the attempt
    // settles, so a later connect is a genuinely fresh one.
    if (connectInFlight !== undefined) return connectInFlight
    const attempt = (async (): Promise<RelayHandshake> => {
      // Fail before 'connecting' is announced when nothing is configured —
      // an unpaired client must never look like it is trying.
      const creds = requireCredentials()
      const previous = state
      setState('connecting')
      let payload: Record<string, unknown>
      try {
        payload = await exchange(HANDSHAKE_PATH, '{}', undefined, false, creds)
      } catch (error) {
        // The two walls keep the state the mapping gave them (revoked /
        // incompatible). Every OTHER failure is an answer the contract says
        // must not move the state — but the entry into this attempt already
        // announced 'connecting', and a refusal would leave it stuck there
        // forever, so the entry state is restored instead (T23a-fix). One
        // correction (T23a-fix2): `unpaired` is reserved for "no credentials
        // configured" — this attempt HAD them, so a restored first attempt
        // lands on `offline`, the honest "configured but not reached".
        if (state === 'connecting') setState(previous === 'unpaired' ? 'offline' : previous)
        // Whatever the failure was, its code is the diagnostics answer — even
        // when the restore left the state untouched (e.g. an online verdict
        // surviving an unmapped handshake failure). Server-supplied codes go
        // through the sanitizer first (T43-fix).
        if (error instanceof RelayError) lastError = sanitizeCode(error.code)
        throw error
      }
      // The protocol gate runs BEFORE the online transition: an incompatible
      // server must end `incompatible`, never flicker through online.
      if (payload.relayProtocol !== RELAY_PROTOCOL) {
        setState('incompatible')
        lastError = 'incompatible'
        throw new RelayError(
          'incompatible',
          `the server speaks relay protocol ${String(payload.relayProtocol)}, this client speaks ${RELAY_PROTOCOL}`,
        )
      }
      handshakeInfo = {
        relayProtocol: RELAY_PROTOCOL,
        serverId: typeof payload.serverId === 'string' ? payload.serverId : '',
        serverName: typeof payload.serverName === 'string' ? payload.serverName : '',
        dshVersion: typeof payload.dshVersion === 'string' ? payload.dshVersion : '',
        fingerprints: isRecord(payload.fingerprints) ? (payload.fingerprints as Record<string, string>) : {},
        // T59: the gateway's record of THIS device's name, when the server
        // speaks the field at all (an older server omits it and the client
        // keeps whatever it had).
        ...(infoString(payload.deviceName) !== undefined ? { deviceName: infoString(payload.deviceName) } : {}),
        // PB-2: the plugin bridge table, kept VERBATIM — an absent or
        // malformed field stays `undefined` so the comparison reads
        // server-too-old, never a silently empty table.
        ...(isRecord(payload.plugins) ? { plugins: payload.plugins as Record<string, ServerPluginEntry> } : {}),
      }
      // Interface compatibility (T42): compare the handshake's group map
      // against this side's own, right here where both are fresh. A compute
      // failure degrades to "nothing computed" — every group lands in
      // `unavailable`, which the comparison contract keeps away from
      // `different`. The verdict is stored BEFORE the online transition, so
      // every subscriber woken by it reads the getter fresh.
      if (options.computeOwnFingerprints !== undefined) {
        let own: Record<string, string> = {}
        try {
          own = (await options.computeOwnFingerprints()) ?? {}
        } catch {
          // Own fingerprints unavailable — the verdict says so, honestly.
        }
        compat = compareFingerprints(handshakeInfo.fingerprints, own)
      }
      // Digest of exactly the credentials this handshake used — never the
      // plaintext values (T23a-fix).
      lastHandshakeDigest = relayCredentialsDigest(creds.url, creds.token)
      setState('online')
      return handshakeInfo
    })()
    connectInFlight = attempt
    const clear = (): void => {
      if (connectInFlight === attempt) connectInFlight = undefined
    }
    attempt.then(clear, clear)
    return attempt
  }

  /** One immediate attempt, ladder reset to the first step (T43). Only an
   * `offline` client has anything to reconnect. */
  function reconnect(): boolean {
    if (stopped || state !== 'offline') return false
    cancelRetry()
    void connect().catch(() => {})
    return true
  }

  /**
   * one authenticated `GET /ping` — the independent
   * health confirmation a restored link needs when no business traffic is
   * flowing. Never throws, never moves `state`, de-duplicated in flight.
   */
  let healthProbeInFlight: Promise<boolean> | undefined
  async function probeHealth(): Promise<boolean> {
    if (healthProbeInFlight !== undefined) return healthProbeInFlight
    const attempt = (async (): Promise<boolean> => {
      let creds: { url: string, token: string }
      try {
        creds = requireCredentials()
      } catch {
        return false
      }
      const ac = new AbortController()
      const timer = setTimeout(() => ac.abort(), Math.min(requestTimeoutMs, 8_000))
      if (typeof timer.unref === 'function') timer.unref()
      try {
        const response = await fetchImpl(`${creds.url}${PING_PATH}`, {
          method: 'GET',
          headers: {
            authorization: `Bearer ${creds.token}`,
            accept: 'application/json',
          },
          redirect: 'manual',
          signal: ac.signal,
        })
        return response.ok
      } catch {
        return false
      } finally {
        clearTimeout(timer)
      }
    })()
    healthProbeInFlight = attempt
    try {
      return await attempt
    } finally {
      if (healthProbeInFlight === attempt) healthProbeInFlight = undefined
    }
  }

  /** controlled fast-path; see the interface note. */
  const NETWORK_HINT_MIN_GAP_MS = 3_000
  let lastNetworkHintAt = 0
  function networkRestored(): boolean {
    if (stopped || state !== 'offline') return false
    const now = clock.now()
    if (now - lastNetworkHintAt < NETWORK_HINT_MIN_GAP_MS) return false
    lastNetworkHintAt = now
    cancelRetry()
    void connect().catch(() => {})
    return true
  }

  /**
   * The credentials source moved (a pairing write, an unpair, a hand edit
   * committed by the loader): drop any pending wait — the ladder belongs to
   * the OLD credentials — clear the stored compat verdict (it describes the
   * server the OLD credentials pointed at, T23b2-fix3) and dial the new
   * values at once. Nothing configured lands `unpaired`, exactly like a
   * request would. A connect already running joined in with the old values
   * (its getters were read at its start); when it settles, ONE follow-up
   * goes out with the new values — skipped if a still-newer change
   * superseded this one.
   */
  function credentialsChanged(): void {
    const creds = currentCredentials()
    const digest = creds === undefined ? undefined : relayCredentialsDigest(creds.url, creds.token)
    if (digest === credentialsDigest) return
    credentialsDigest = digest
    // Same invalidation rule as setState's unpaired/revoked branch: a
    // verdict the current credentials never earned must not stay visible —
    // even when the very next handshake lands online, there is a window
    // between the change and that handshake.
    compat = undefined
    cancelRetry()
    if (stopped) return
    if (creds === undefined) {
      setState('unpaired')
      return
    }
    const inFlight = connectInFlight
    if (inFlight === undefined) {
      void connect().catch(() => {})
      return
    }
    // The running attempt is the old credentials' dial. Joining it would be
    // a no-op (T23a-fix2 dedupe) AND answer nothing about the new values —
    // the follow-up below is the new dial (T43-fix).
    void inFlight.catch(() => {}).then(() => {
      if (stopped) return
      const latest = currentCredentials()
      if (latest === undefined || relayCredentialsDigest(latest.url, latest.token) !== digest) return
      void connect().catch(() => {})
    })
  }

  function stop(): void {
    stopped = true
    cancelRetry()
    if (infoTimer !== undefined) {
      clock.clearTimeout(infoTimer)
      infoTimer = undefined
    }
  }

  async function invoke(namespace: string, method: string, args: unknown, signal?: AbortSignal): Promise<unknown> {
    const creds = requireCredentials()
    // One serialization, two uses: the body on the wire and the size the
    // round-trip budget scales with. The timeout deliberately does NOT move
    // the connection state — a slow prompt is a slow call, not a dead link.
    const bodyJson = JSON.stringify({ namespace, method, args })
    const payload = await exchange(INVOKE_PATH, bodyJson, signal, true, creds, {
      timeoutMs: invokeTimeoutMs(requestTimeoutMs, bodyJson),
      timeoutSetsOffline: false,
    })
    return payload.value
  }

  async function postEventResult(eventId: string, result: unknown, signal?: AbortSignal): Promise<unknown> {
    const creds = requireCredentials()
    const payload = await exchange(EVENT_RESULT_PATH, JSON.stringify({ eventId, result }), signal, true, creds)
    return payload.value
  }

  async function unshare(sessionId: string, signal?: AbortSignal): Promise<void> {
    const creds = requireCredentials()
    await exchange(UNSHARE_PATH, JSON.stringify({ sessionId }), signal, true, creds)
  }

  async function http(route: string, query: string, signal?: AbortSignal): Promise<RelayHttpResult> {
    const creds = requireCredentials()
    const payload = await exchange(HTTP_PATH, JSON.stringify({ route, query }), signal, true, creds)
    const value = payload.value
    if (
      !isRecord(value) ||
      typeof value.status !== 'number' ||
      !Number.isSafeInteger(value.status) ||
      typeof value.body !== 'string'
    ) {
      throw new RelayError('internal', 'the relay http answer carries no upstream response')
    }
    return {
      status: value.status,
      contentType: typeof value.contentType === 'string' ? value.contentType : undefined,
      body: value.body,
    }
  }

  async function upload(options: RelayUploadOptions, signal?: AbortSignal): Promise<RelayHttpResult> {
    const creds = requireCredentials()
    const params = new URLSearchParams()
    params.set('sessionId', options.sessionId)
    if (options.name !== undefined) params.set('name', options.name)
    // The empty-body upload still rides a (closed) stream: the wire route
    // always speaks a request body, and a missing one would be a transport
    // error, not the host's own empty-upload business failure.
    const body =
      options.body ??
      new ReadableStream<Uint8Array>({
        start: (controller) => {
          controller.close()
        },
      })
    // The budget scales with the DECLARED size (T31-fix's rule); a timeout
    // deliberately does not move the connection state — a slow upload is a
    // slow call, not a dead link.
    const payload = await exchangeRequest(
      `${creds.url}${UPLOAD_PATH}?${params.toString()}`,
      uploadInit(creds.token, body),
      signal,
      true,
      { timeoutMs: sizeBackedTimeoutMs(requestTimeoutMs, options.bytes), timeoutSetsOffline: false, transportSetsOffline: false },
    )
    const value = payload.value
    if (
      !isRecord(value) ||
      typeof value.status !== 'number' ||
      !Number.isSafeInteger(value.status) ||
      typeof value.body !== 'string'
    ) {
      throw new RelayError('internal', 'the relay upload answer carries no upstream response')
    }
    return {
      status: value.status,
      contentType: typeof value.contentType === 'string' ? value.contentType : undefined,
      body: value.body,
    }
  }

  /**
   * The bridged-plugin forward (PB-4): headers in, an UNREAD Response out.
   * Deliberately NOT the shared exchange (it buffers the body): the fetch
   * resolves on headers, the plain request timeout guards exactly that
   * wait, and the body streams under the CALLER's signal alone. The
   * caller's abort listener survives past the header arrival (registered
   * `once` on the caller's own request-scoped signal, so it dies with that
   * request) — a browser hang-up mid-stream must tear the relay round-trip
   * down, and the server's own route then aborts its loopback in turn.
   */
  async function pluginRequest(request: RelayPluginRequest, signal?: AbortSignal): Promise<Response> {
    const creds = requireCredentials()
    const controller = new AbortController()
    const onExternalAbort = (): void => {
      controller.abort()
    }
    if (signal !== undefined) {
      if (signal.aborted) controller.abort()
      else signal.addEventListener('abort', onExternalAbort, { once: true })
    }
    // The header wait gets the plain request timeout for a bodyless call;
    // with a BODY (PB-9) the budget scales with its declared size like the
    // upload's (a slow body is a slow call, never a dead link), and the
    // timeout never moves the connection state.
    const hasBody = request.body !== undefined && request.body !== null
    const headerTimeoutMs = hasBody
      ? sizeBackedTimeoutMs(requestTimeoutMs, request.contentLength)
      : requestTimeoutMs
    const timer = setTimeout(() => controller.abort(), headerTimeoutMs)
    if (typeof timer.unref === 'function') timer.unref()
    try {
      let response: Response
      const init: RequestInit = {
        // The WIRE method is always POST — the relay route is a POST endpoint
        // whatever the bridged call is; the bridged method travels in
        // `x-zr-method` (the server never reads this request line's verb).
        method: 'POST',
        headers: {
          authorization: `Bearer ${creds.token}`,
          // No accept of our own unless the caller carries one: this
          // header FORWARDS to the plugin (PB-3's protocol), and the
          // default below exists only so a revoked-token wall still
          // answers JSON instead of the pairing HTML page.
          accept: request.accept ?? 'application/json',
          ...(request.range !== undefined ? { range: request.range } : {}),
          // PB-9: the body's content type is the OTHER passthrough header.
          ...(hasBody && request.contentType !== undefined ? { 'content-type': request.contentType } : {}),
          'x-zr-plugin': request.plugin,
          'x-zr-plugin-version': request.version,
          'x-zr-plugin-digest': request.digest,
          'x-zr-path': request.path,
          'x-zr-query': request.query,
          // PB-9: the method travels in the HEADER (the wire protocol's
          // coordinate), not just the request line — the server reads it
          // from here, and an omitted header means GET there no matter what
          // this request line says.
          'x-zr-method': request.method ?? 'GET',
        },
        ...(hasBody ? { body: request.body } : {}),
        // A wrong server must not walk the pairing token through a redirect.
        redirect: 'manual',
        signal: controller.signal,
      }
      if (hasBody) {
        // `duplex: 'half'` is mandatory for a streaming request body; the
        // resolved RequestInit type hides the field (the DOM-vs-undici
        // conditional in @types/node), so the assignment carries its own cast.
        ;(init as { duplex?: string }).duplex = 'half'
      }
      try {
        response = await fetchImpl(`${creds.url}${PLUGIN_PATH}`, init)
      } catch (error) {
        // A caller abort is NOT a transport death (PB-9-fix): the caller
        // aborts exactly when its own request-body gate tripped or the
        // browser hung up — both are this call's verdicts to answer, never
        // the link's. Same discipline as the invoke/upload timeouts
        // (T31-fix/T51-fix): no offline, no failure recorded on the link.
        if (signal?.aborted) throw abortedError()
        // A body-bearing call keeps the upload route's rule (T51-fix): the
        // timeout judges THIS call, not the link — but a real transport
        // death still reads offline (the reconnect ladder hangs off it).
        if (hasBody && controller.signal.aborted && !signal?.aborted) {
          noteFailure('request-timeout')
          throw new RelayError('request-timeout', `no response within ${headerTimeoutMs} ms`)
        }
        noteFailure('offline')
        setState('offline')
        throw new RelayError('offline', messageOf(error))
      }
      // Headers are here: every status — bridged answer or x-zen-remote-error
      // refusal — travels to the caller verbatim, body untouched.
      return response
    } finally {
      // The header wait is over (or never started): no deadline on the body.
      clearTimeout(timer)
    }
  }

  /**
   * The WebSocket byte tunnel's open half (PB-10). The wire: one POST whose
   * answer is an NDJSON river — first line names the tunnel, `data` lines
   * carry base64 chunks of the upstream socket's bytes, `ping` lines keep
   * proxies warm, an `end` line closes the river. Everything after the
   * headers is line-framed HERE: the caller gets an async iterable of
   * decoded byte chunks and never sees the framing.
   */
  async function pluginTunnelOpen(
    request: RelayPluginRequest & {
      wsHeaders: { 'sec-websocket-key': string; 'sec-websocket-version': string; 'sec-websocket-protocol'?: string; 'sec-websocket-extensions'?: string }
      head?: string
    },
    signal?: AbortSignal,
  ): Promise<{ tunnel: string; frames: AsyncIterable<Uint8Array> }> {
    const creds = requireCredentials()
    const payload = {
      headers: request.wsHeaders,
      ...(request.head !== undefined && request.head !== '' ? { head: request.head } : {}),
    }
    const response = await fetch(`${creds.url}${PLUGIN_TUNNEL_OPEN_PATH}`, {
      method: 'POST',
      headers: {
        authorization: `Bearer ${creds.token}`,
        accept: 'application/x-ndjson',
        'content-type': 'application/json',
        'x-zr-plugin': request.plugin,
        'x-zr-plugin-version': request.version,
        'x-zr-plugin-digest': request.digest,
        'x-zr-path': request.path,
        'x-zr-query': request.query,
      },
      body: JSON.stringify(payload),
      redirect: 'manual',
      signal,
    }).catch((error: unknown) => {
      noteFailure('offline')
      setState('offline')
      throw new RelayError('offline', messageOf(error))
    })
    if (response.status < 200 || response.status >= 300 || response.body === null) {
      // The refusal carries its code in the marker header (every
      // plugin-bridge wall does) — surface it verbatim.
      const code = response.headers.get('x-zen-remote-error') ?? `http-${response.status}`
      const bodyText = await response.text().catch(() => '')
      throw new RelayError(code, bodyText === '' ? code : bodyText, response.status)
    }
    let tunnelId: string | undefined
    let upstreamClosed = false
    let pendingError: RelayError | undefined
    const reader = response.body.getReader()
    const decoder = new TextDecoder()
    let lineBuffer = ''
    const nextLine = async (): Promise<string | null> => {
      while (true) {
        const idx = lineBuffer.indexOf('\n')
        if (idx >= 0) {
          const line = lineBuffer.slice(0, idx)
          lineBuffer = lineBuffer.slice(idx + 1)
          return line
        }
        const { done, value } = await reader.read()
        if (done) {
          const tail = lineBuffer
          lineBuffer = ''
          return tail === '' ? null : tail
        }
        lineBuffer += decoder.decode(value, { stream: true })
      }
    }
    // The FIRST line must name the tunnel; anything else up front is a
    // protocol violation the caller sees as a refused open.
    const first = await nextLine()
    if (first !== null) {
      try {
        const parsed: unknown = JSON.parse(first)
        if (typeof parsed === 'object' && parsed !== null && (parsed as { type?: unknown }).type === 'open' &&
            typeof (parsed as { tunnel?: unknown }).tunnel === 'string') {
          tunnelId = (parsed as { tunnel: string }).tunnel
        }
      } catch {
        // Falls through to the refusal below.
      }
    }
    if (tunnelId === undefined) {
      const code = response.headers.get('x-zen-remote-error') ?? 'internal'
      throw new RelayError(code, `the tunnel open stream never named a tunnel`, response.status)
    }
    const frames: AsyncIterable<Uint8Array> = {
      [Symbol.asyncIterator]() {
        return {
          async next(): Promise<IteratorResult<Uint8Array>> {
            if (upstreamClosed) return { done: true, value: undefined }
            while (true) {
              const line = await nextLine()
              if (line === null) {
                upstreamClosed = true
                return { done: true, value: undefined }
              }
              let parsed: unknown
              try {
                parsed = JSON.parse(line)
              } catch {
                continue // a damaged line is skipped, like every NDJSON reader here
              }
              if (parsed === null || typeof parsed !== 'object') continue
              const frame = parsed as { type?: unknown; b64?: unknown; reason?: unknown }
              if (frame.type === 'data' && typeof frame.b64 === 'string' && frame.b64 !== '') {
                return { done: false, value: new Uint8Array(Buffer.from(frame.b64, 'base64')) }
              }
              if (frame.type === 'end') {
                upstreamClosed = true
                if (typeof frame.reason === 'string' && frame.reason === 'upstream-error') {
                  pendingError = new RelayError('internal', 'the upstream tunnel socket failed')
                }
                return { done: true, value: undefined }
              }
              // ping and anything unrecognized: skipped.
            }
          },
        }
      },
    }
    void pendingError
    return { tunnel: tunnelId, frames }
  }

  async function pluginTunnelSend(tunnel: string, seq: number, bytes: Uint8Array, signal?: AbortSignal): Promise<void> {
    const creds = requireCredentials()
    // The body IS the payload — raw bytes, not JSON. The requestInit shape
    // lives in exchangeRequest's url+init contract: headers hand-built here.
    await exchangeRequest(
      `${creds.url}${PLUGIN_TUNNEL_SEND_PATH}`,
      {
        method: 'POST',
        headers: {
          authorization: `Bearer ${creds.token}`,
          accept: 'application/json',
          'content-type': 'application/octet-stream',
          'x-zr-tunnel': tunnel,
          'x-zr-seq': String(seq),
        },
        // undici's BodyInit types reject ArrayBufferView here (the DOM face
        // wins in @types/node) — the cast carries the wire reality.
        body: new Uint8Array(bytes) as unknown as BodyInit,
        redirect: 'manual',
      },
      signal,
      true,
      { timeoutMs: sizeBackedTimeoutMs(requestTimeoutMs, bytes.byteLength), timeoutSetsOffline: false },
    )
  }

  async function pluginTunnelClose(tunnel: string, signal?: AbortSignal): Promise<void> {
    const creds = requireCredentials()
    await fetch(`${creds.url}${PLUGIN_TUNNEL_CLOSE_PATH}`, {
      method: 'POST',
      headers: {
        authorization: `Bearer ${creds.token}`,
        accept: 'application/json',
        'x-zr-tunnel': tunnel,
      },
      redirect: 'manual',
      signal,
    }).catch((error: unknown) => {
      throw new RelayError('offline', messageOf(error))
    })
  }

  function openStream(namespace: string, method: string, args: unknown, signal?: AbortSignal): AsyncIterable<unknown> {
    // Eager, not generator-lazy: an unconfigured client fails at CALL time,
    // exactly like connect/invoke, instead of hiding the error inside the
    // first next().
    requireCredentials()
    return streamLines(namespace, method, args, signal)
  }

  /**
   * The NDJSON body pump. The request is owned by `controller`, which three
   * things may abort: the caller's signal (→ the iteration ends normally),
   * the idle clock (no line of any kind within `idleTimeoutMs` → offline),
   * and the generator's own exit — `break`, a thrown error line, normal
   * end — which tears the request down so the server's upstream does not
   * keep pumping behind a consumer that left.
   */
  async function* streamLines(
    namespace: string,
    method: string,
    args: unknown,
    signal: AbortSignal | undefined,
  ): AsyncGenerator<unknown> {
    if (signal?.aborted) return
    // Re-checked: the generator body runs at the first next(), so the
    // credentials may have vanished since openStream validated them.
    const { url, token } = requireCredentials()
    const controller = new AbortController()
    const onExternalAbort = (): void => {
      controller.abort()
    }
    if (signal !== undefined) signal.addEventListener('abort', onExternalAbort)

    let idleFired = false
    let headersTimedOut = false
    let idleTimer: ReturnType<typeof setTimeout> | undefined
    const armIdle = (): void => {
      if (idleTimer !== undefined) clearTimeout(idleTimer)
      idleTimer = setTimeout(() => {
        idleFired = true
        controller.abort()
      }, idleTimeoutMs)
      if (typeof idleTimer.unref === 'function') idleTimer.unref()
    }
    const disarmIdle = (): void => {
      if (idleTimer !== undefined) {
        clearTimeout(idleTimer)
        idleTimer = undefined
      }
    }
    // Headers get the request timeout; the BODY is governed by the idle
    // clock alone — a healthy stream stays silent for hours. The timer is
    // disarmed the moment the response HEADERS arrive (it is cleared again
    // in the finally for the pre-header failure paths), or it would abort
    // every stream still alive past requestTimeoutMs (T23a-fix).
    const headerTimer = setTimeout(() => {
      headersTimedOut = true
      controller.abort()
    }, requestTimeoutMs)
    if (typeof headerTimer.unref === 'function') headerTimer.unref()
    // Hoisted for the teardown below: even a fetchImpl that ignores the
    // signal must have its response body released (T23a-fix).
    let reader: ReadableStreamDefaultReader<Uint8Array> | undefined

    try {
      let response: Response
      try {
        response = await fetchImpl(`${url}${STREAM_PATH}`, {
          ...requestInit(token, JSON.stringify({ namespace, method, args })),
          signal: controller.signal,
        })
      } catch (error) {
        if (signal?.aborted) return
        noteFailure('offline')
        setState('offline')
        throw new RelayError('offline', headersTimedOut ? `no response within ${requestTimeoutMs} ms` : messageOf(error))
      }
      if (!response.ok) {
        // The NDJSON protocol only starts once the stream is allowed — a
        // refusal is the same JSON the invoke route answers, mapped the
        // same way. The header clock keeps guarding this leg: a server that
        // answers 4xx headers but never sends the body must not hang the
        // stream forever (T23a-fix2) — the abort lands here as offline.
        let payload: unknown
        try {
          payload = await readPayload(response)
        } catch (error) {
          if (signal?.aborted) return
          noteFailure('offline')
          setState('offline')
          throw new RelayError('offline', messageOf(error))
        }
        throw failureOf(response.status, payload)
      }
      // The 200 itself is a successful response: the link works. Only NOW
      // is the header phase truly over and its clock disarmed.
      clearTimeout(headerTimer)
      setState('online')
      if (response.body === null) return
      armIdle()
      const decoder = new TextDecoder()
      let buffer = ''
      reader = response.body.getReader()
      while (true) {
        const { done, value } = await reader.read()
        armIdle() // bytes arrived — any line content refreshes the clock
        if (done) break
        buffer += decoder.decode(value, { stream: true })
        let idx: number
        while ((idx = buffer.indexOf('\n')) >= 0) {
          const line = buffer.slice(0, idx)
          buffer = buffer.slice(idx + 1)
          const verdict = classifyLine(line)
          if (verdict.kind === 'frame') {
            // The idle clock measures TRANSPORT silence, not the consumer's
            // processing time: it is paused across the yield and re-armed
            // when the value is pulled.
            disarmIdle()
            yield verdict.frame
            armIdle()
          } else if (verdict.kind === 'end') {
            return
          } else if (verdict.kind === 'error') {
            noteServerRestart(verdict.error)
            throw verdict.error
          } else if (verdict.kind === 'info') {
            // T59: the heartbeat's names, folded in quietly — a state-sparing
            // notify wakes the interceptor's rename path without any reconnect.
            applyInfoDelta(verdict.serverName, verdict.deviceName)
          }
        }
      }
      // Tolerant tail: a final line without its newline still counts.
      buffer += decoder.decode()
      if (buffer.trim() !== '') {
        const verdict = classifyLine(buffer)
        if (verdict.kind === 'frame') {
          disarmIdle()
          yield verdict.frame
        } else if (verdict.kind === 'end') {
          return
        } else if (verdict.kind === 'error') {
          noteServerRestart(verdict.error)
          throw verdict.error
        } else if (verdict.kind === 'info') {
          applyInfoDelta(verdict.serverName, verdict.deviceName)
        }
      }
    } catch (error) {
      if (signal?.aborted) return
      if (error instanceof RelayError) throw error
      // Anything else escaping the body read is the transport dying.
      noteFailure('offline')
      setState('offline')
      throw new RelayError('offline', idleFired ? `no line received for ${idleTimeoutMs} ms` : messageOf(error))
    } finally {
      disarmIdle()
      clearTimeout(headerTimer)
      if (signal !== undefined) signal.removeEventListener('abort', onExternalAbort)
      // Release the body even if a custom fetchImpl ignored the signal —
      // an in-flight reader keeps the connection's buffers alive. Cancel
      // may reject on an already-dead stream; that is fine.
      if (reader !== undefined) reader.cancel().catch(() => {})
      // However we left the body — break, error line, normal end, caller
      // abort — the request itself must not linger: the server aborts its
      // upstream when the socket closes. Aborting a finished request is a
      // no-op.
      controller.abort()
    }
  }

  return {
    get state(): RelayState {
      return state
    },
    get handshakeInfo(): RelayHandshake | undefined {
      return handshakeInfo
    },
    get lastHandshakeDigest(): string | undefined {
      return lastHandshakeDigest
    },
    get nextRetryAt(): number | null {
      return nextRetryAt
    },
    get lastError(): string | undefined {
      return lastError
    },
    get compat(): RelayCompatVerdict | undefined {
      return compat
    },
    get deviceNameSyncing(): boolean {
      return deviceNameSyncing()
    },
    subscribe,
    queueDeviceName,
    connect,
    reconnect,
    // the two new surfaces. `probeHealth` is the
    // authenticated GET /ping health check (never moves `state`);
    // `networkRestored` is the controlled fast-path (offline-only + 3s gap).
    probeHealth,
    networkRestored,
    credentialsChanged,
    stop,
    invoke,
    unshare,
    postEventResult,
    openStream,
    http,
    upload,
    pluginRequest,
    pluginTunnelOpen,
    pluginTunnelSend,
    pluginTunnelClose,
  }
}