/**
 * Relay integration for joy-tmux.
 * Self-contained — no deps on joy-daemon internals.
 * External deps: socket.io-client, tweetnacl.
 */
import { setTimeout as sleep } from "timers/promises";
import { execSync } from 'node:child_process';
import { createCipheriv, createDecipheriv, createHmac, randomBytes } from 'node:crypto';
import { existsSync, readFileSync, writeFileSync, renameSync, mkdirSync, statfsSync } from 'node:fs';
import { join } from 'node:path';
import { hostname, platform, cpus, freemem, totalmem, loadavg, homedir } from 'node:os';
import { happyHomeDir, joyStateDir } from '../paths';
import { io, type Socket } from 'socket.io-client';
import tweetnacl from 'tweetnacl';

// ── Types ──────────────────────────────────────────────────────────────────────

export type EncryptionVariant = 'legacy' | 'dataKey';

export interface Credentials {
  token: string;
  serverUrl: string;
  machineId: string;
  encryption:
    | { type: 'dataKey'; publicKey: Uint8Array; machineKey: Uint8Array }
    | { type: 'legacy'; secret: Uint8Array };
}

export interface WireRecord {
  role: 'user' | 'agent' | 'session';
  content: { type: string; [k: string]: unknown };
  meta?: { sentFrom?: string; [k: string]: unknown };
}

// ── Crypto ─────────────────────────────────────────────────────────────────────

function b64encode(buf: Uint8Array): string {
  return Buffer.from(buf).toString('base64');
}

function b64decode(s: string): Uint8Array {
  return new Uint8Array(Buffer.from(s, 'base64'));
}

function randomBytesU8(n: number): Uint8Array {
  return new Uint8Array(randomBytes(n));
}

function encryptLegacy(data: unknown, key: Uint8Array): Uint8Array {
  const nonce = randomBytesU8(tweetnacl.secretbox.nonceLength);
  const pt = new TextEncoder().encode(JSON.stringify(data));
  const ct = tweetnacl.secretbox(pt, nonce, key);
  const out = new Uint8Array(nonce.length + ct.length);
  out.set(nonce);
  out.set(ct, nonce.length);
  return out;
}

function decryptLegacy(buf: Uint8Array, key: Uint8Array): unknown | null {
  const n = tweetnacl.secretbox.nonceLength;
  if (buf.length < n) return null;
  const pt = tweetnacl.secretbox.open(buf.slice(n), buf.slice(0, n), key);
  if (!pt) return null;
  try { return JSON.parse(new TextDecoder().decode(pt)); } catch { return null; }
}

function encryptDataKey(data: unknown, key: Uint8Array): Uint8Array {
  const nonce = randomBytesU8(12);
  const cipher = createCipheriv('aes-256-gcm', key, nonce);
  const pt = new TextEncoder().encode(JSON.stringify(data));
  const enc = Buffer.concat([cipher.update(pt), cipher.final()]);
  const tag = cipher.getAuthTag();
  const bundle = new Uint8Array(1 + 12 + enc.length + 16);
  bundle.set([0], 0);
  bundle.set(nonce, 1);
  bundle.set(new Uint8Array(enc), 13);
  bundle.set(new Uint8Array(tag), 13 + enc.length);
  return bundle;
}

function decryptDataKey(buf: Uint8Array, key: Uint8Array): unknown | null {
  if (buf.length < 1 + 12 + 16 || buf[0] !== 0) return null;
  const nonce = buf.slice(1, 13);
  const tag = buf.slice(buf.length - 16);
  const ct = buf.slice(13, buf.length - 16);
  try {
    const dec = createDecipheriv('aes-256-gcm', key, nonce);
    dec.setAuthTag(tag);
    return JSON.parse(new TextDecoder().decode(Buffer.concat([dec.update(ct), dec.final()])));
  } catch { return null; }
}

function encryptWire(variant: EncryptionVariant, key: Uint8Array, data: unknown): Uint8Array {
  return variant === 'legacy' ? encryptLegacy(data, key) : encryptDataKey(data, key);
}

function decryptWire(variant: EncryptionVariant, key: Uint8Array, buf: Uint8Array): unknown | null {
  return variant === 'legacy' ? decryptLegacy(buf, key) : decryptDataKey(buf, key);
}

// ── Blob crypto (image attachments) ────────────────────────────────────────────
//
// Mirrors happy-cli's deriveKey + decryptBlob: HMAC-SHA512 key tree derivation
// rooted at the encryption secret, then NaCl secretbox unwrap. Used to download
// + decrypt image attachments that the app uploads via /v1/sessions/{id}/attachments.

function hmacSha512(key: Uint8Array, data: Uint8Array): Uint8Array {
  return new Uint8Array(createHmac('sha512', key).update(data).digest());
}

function deriveKeyTreeRoot(seed: Uint8Array, usage: string): { key: Uint8Array; chainCode: Uint8Array } {
  const I = hmacSha512(new TextEncoder().encode(usage + ' Master Seed'), seed);
  return { key: I.slice(0, 32), chainCode: I.slice(32) };
}

function deriveKeyTreeChild(chainCode: Uint8Array, index: string): { key: Uint8Array; chainCode: Uint8Array } {
  const data = new Uint8Array([0, ...new TextEncoder().encode(index)]);
  const I = hmacSha512(chainCode, data);
  return { key: I.slice(0, 32), chainCode: I.slice(32) };
}

function deriveKey(master: Uint8Array, usage: string, path: string[]): Uint8Array {
  let state = deriveKeyTreeRoot(master, usage);
  for (const seg of path) {
    state = deriveKeyTreeChild(state.chainCode, seg);
  }
  return state.key;
}

/**
 * Decrypt a NaCl secretbox bundle: [24-byte nonce][ciphertext + 16-byte tag].
 * Returns null on tamper / wrong key. Matches happy-cli's decryptBlob.
 */
function decryptBlob(bundle: Uint8Array, key: Uint8Array): Uint8Array | null {
  const NONCE_LEN = tweetnacl.secretbox.nonceLength;
  if (bundle.length < NONCE_LEN + 16) return null;
  const nonce = bundle.slice(0, NONCE_LEN);
  const ciphertext = bundle.slice(NONCE_LEN);
  const plain = tweetnacl.secretbox.open(ciphertext, nonce, key);
  return plain ? new Uint8Array(plain) : null;
}

function libsodiumEncryptForPublicKey(data: Uint8Array, recipientPublicKey: Uint8Array): Uint8Array {
  const ephemeral = tweetnacl.box.keyPair();
  const nonce = randomBytesU8(tweetnacl.box.nonceLength);
  const ct = tweetnacl.box(data, nonce, recipientPublicKey, ephemeral.secretKey);
  const out = new Uint8Array(ephemeral.publicKey.length + nonce.length + ct.length);
  out.set(ephemeral.publicKey, 0);
  out.set(nonce, ephemeral.publicKey.length);
  out.set(ct, ephemeral.publicKey.length + nonce.length);
  return out;
}

// ── Credentials ────────────────────────────────────────────────────────────────

const DEFAULT_SERVER_URL = 'https://api.cluster-fluster.com';

export function loadCredentials(): Credentials | null {
  const happyHome = happyHomeDir();

  const accessKeyPath = join(happyHome, 'access.key');
  if (!existsSync(accessKeyPath)) return null;

  try {
    const ak = JSON.parse(readFileSync(accessKeyPath, 'utf8')) as {
      token?: string;
      encryption?: { publicKey?: string; machineKey?: string; secret?: string };
    };
    if (!ak.token) return null;

    let serverUrl = process.env.HAPPY_SERVER_URL ?? DEFAULT_SERVER_URL;
    let machineId: string | undefined;
    try {
      const s = JSON.parse(readFileSync(join(happyHome, 'settings.json'), 'utf8')) as { serverUrl?: string; machineId?: string };
      if (s.serverUrl && !process.env.HAPPY_SERVER_URL) serverUrl = s.serverUrl;
      if (s.machineId) machineId = s.machineId;
    } catch {}
    // M5 (machineId): a random fallback would silently break RPC on every restart
    if (!machineId) {
      process.stderr.write('[relay] WARNING: machineId missing from settings.json — RPC handlers will not be reachable. Run the happy-cli daemon once to populate it.\n');
      machineId = crypto.randomUUID();
    }

    let encryption: Credentials['encryption'] | null = null;
    if (ak.encryption?.publicKey) {
      encryption = {
        type: 'dataKey',
        publicKey: b64decode(ak.encryption.publicKey),
        machineKey: ak.encryption.machineKey ? b64decode(ak.encryption.machineKey) : new Uint8Array(),
      };
    } else if (ak.encryption?.secret) {
      encryption = { type: 'legacy', secret: b64decode(ak.encryption.secret) };
    }
    if (!encryption) return null;

    return { token: ak.token, serverUrl, machineId, encryption };
  } catch { return null; }
}

// ── RPC encryption (matches happy-cli encryptWithDataKey / decryptWithDataKey) ─

function encryptRpc(key: Uint8Array, data: unknown): string {
  const nonce = randomBytesU8(12);
  const cipher = createCipheriv('aes-256-gcm', key, nonce);
  const pt = new TextEncoder().encode(JSON.stringify(data));
  const ct = Buffer.concat([cipher.update(pt), cipher.final()]);
  const tag = cipher.getAuthTag();
  // version(1=0x00) + nonce(12) + ciphertext + tag(16)
  const bundle = new Uint8Array(1 + 12 + ct.length + 16);
  bundle[0] = 0;
  bundle.set(nonce, 1);
  bundle.set(new Uint8Array(ct), 13);
  bundle.set(new Uint8Array(tag), 13 + ct.length);
  return Buffer.from(bundle).toString('base64');
}

function decryptRpc(key: Uint8Array, b64: string): unknown {
  const bundle = Buffer.from(b64, 'base64');
  if (bundle.length < 1 + 12 + 16 || bundle[0] !== 0) return null;
  const nonce = bundle.slice(1, 13);
  const tag = bundle.slice(bundle.length - 16);
  const ct = bundle.slice(13, bundle.length - 16);
  try {
    const decipher = createDecipheriv('aes-256-gcm', key, nonce);
    decipher.setAuthTag(tag);
    const pt = Buffer.concat([decipher.update(ct), decipher.final()]);
    return JSON.parse(new TextDecoder().decode(pt));
  } catch { return null; }
}

// ── Relay HTTP + socket client ────────────────────────────────────────────────

interface RawMessage {
  id: string;
  seq: number;
  content: string | { c: string; t?: string };
  localId: string | null;
}

function rawContentB64(c: RawMessage['content']): string | null {
  if (typeof c === 'string') return c;
  if (c && typeof c === 'object' && typeof c.c === 'string') return c.c;
  return null;
}

export interface CreateSessionResult {
  sessionId: string;
  sessionKey: Uint8Array;
  variant: EncryptionVariant;
  // The server's CURRENT metadata + version. On tag-dedup the server returns the
  // EXISTING session (it ignores the POSTed metadata), so this is the source of
  // truth to merge onto — using it avoids clobbering a previously-set title.
  metadata: Record<string, unknown> | null;
  metadataVersion: number;
}

export class RelayClient {
  readonly serverUrl: string;
  readonly creds: Credentials;
  private socket: Socket | null = null;
  private listeners = new Map<string, Set<() => void>>();
  private activeSessions = new Set<RelaySession>();
  private machineAliveTimer: ReturnType<typeof setInterval> | null = null;
  // Server-side daemonState version (CAS) for machine-update-state; seeded from
  // getOrCreateMachine, re-synced from each ack. Plus the previous cpu-tick
  // snapshot so we can report CPU% as a busy-delta between heartbeats.
  private daemonStateVersion = 0;
  private prevCpuSample: { idle: number; total: number } | null = null;

  constructor(creds: Credentials) {
    this.creds = creds;
    this.serverUrl = creds.serverUrl;
  }

  trackSession(rs: RelaySession): void { this.activeSessions.add(rs); }
  untrackSession(rs: RelaySession): void { this.activeSessions.delete(rs); }

  onReconnect: (() => void) | null = null;

  private rpcHandlers = new Map<string, (params: unknown) => Promise<unknown>>();

  /** Register a session-scoped RPC handler that the app calls via apiSocket.sessionRPC. */
  registerSessionRpcHandler(relaySessionId: string, method: string, handler: (params: unknown) => Promise<unknown>): void {
    const prefixed = `${relaySessionId}:${method}`;
    this.rpcHandlers.set(prefixed, handler);
    log(`rpc: registering session ${prefixed}`);
    this.socket?.emit('rpc-register', { method: prefixed });
  }

  /** Drop every session-scoped RPC handler for a relay session (called on
   *  detach/stop) so dead handlers neither linger in the map nor get
   *  re-registered on the next socket reconnect. */
  unregisterSessionRpcHandlers(relaySessionId: string): void {
    const prefix = `${relaySessionId}:`;
    for (const key of this.rpcHandlers.keys()) {
      if (key.startsWith(prefix)) this.rpcHandlers.delete(key);
    }
  }

  registerRpcHandler(method: string, handler: (params: unknown) => Promise<unknown>): void {
    const prefixed = `${this.creds.machineId}:${method}`;
    this.rpcHandlers.set(prefixed, handler);
    log(`rpc: registering ${prefixed}`);
    this.socket?.emit('rpc-register', { method: prefixed }, (ack: unknown) => {
      log(`rpc: registered ${prefixed} ack=${JSON.stringify(ack)}`);
    });
  }

  /** Heartbeat machine presence so the app shows this machine online. The
   *  server marks the machine active on each machine-alive and lapses it to
   *  offline without one (this is what happy-cli's daemon did — joy now owns
   *  it). Beats immediately on every (re)connect, then every 20s. */
  private startMachineAlive(): void {
    const beat = () => {
      this.socket?.emit('machine-alive', { machineId: this.creds.machineId, time: Date.now() });
      this.pushDaemonState();
    };
    beat();
    if (!this.machineAliveTimer) this.machineAliveTimer = setInterval(beat, 20_000);
  }

  /** Machine encryption variant + key, matching getOrCreateMachine's metadata. */
  private machineCrypto(): { variant: EncryptionVariant; key: Uint8Array } {
    return this.creds.encryption.type === 'dataKey'
      ? { variant: 'dataKey', key: this.creds.encryption.machineKey }
      : { variant: 'legacy', key: this.creds.encryption.secret };
  }

  /** CPU busy % since the last sample (delta of idle vs total ticks across all
   *  cores). The first call has no previous sample, so it falls back to the
   *  1-min load average scaled by core count. */
  private sampleCpuPercent(): number {
    const list = cpus();
    let idle = 0, total = 0;
    for (const c of list) { idle += c.times.idle; for (const t of Object.values(c.times)) total += t; }
    const prev = this.prevCpuSample;
    this.prevCpuSample = { idle, total };
    if (!prev) return Math.max(0, Math.min(100, Math.round((loadavg()[0] / Math.max(1, list.length)) * 100)));
    const di = idle - prev.idle, dt = total - prev.total;
    if (dt <= 0) return 0;
    return Math.max(0, Math.min(100, Math.round((1 - di / dt) * 100)));
  }

  /** Push host CPU%/RAM% into the machine's encrypted daemonState (version-checked
   *  CAS; the app decrypts it and shows it on the machine header). Rides the
   *  machine-alive heartbeat — this op also bumps the machine's active/lastActiveAt
   *  server-side, so it doubles as presence. Best-effort: a 'Machine not found'
   *  (no session has registered the machine yet) or version drift just no-ops and
   *  the next beat retries with the re-synced version. */
  /**
   * Reclaimable-aware available memory. os.freemem() reports only TRULY free
   * pages — on macOS that excludes inactive/purgeable/file-cache the OS holds
   * but frees on demand, so `1 - free/total` read ~99% "used" while Activity
   * Monitor showed ~46% (2026-07-05); Linux freemem() ignores cache/buffers
   * the same way. Use MemAvailable (Linux) / vm_stat reclaimable pages (macOS),
   * falling back to freemem() on anything unexpected. Best-effort + cached 5s
   * so the heartbeat never blocks on vm_stat.
   */
  #availMemCache: { bytes: number; at: number } | null = null;
  private availableMemBytes(): number {
    const now = Date.now();
    if (this.#availMemCache && now - this.#availMemCache.at < 5000) return this.#availMemCache.bytes;
    let bytes = freemem();
    try {
      if (platform() === 'linux') {
        const m = /MemAvailable:\s+(\d+)\s+kB/.exec(readFileSync('/proc/meminfo', 'utf8'));
        if (m) bytes = Number(m[1]) * 1024;
      } else if (platform() === 'darwin') {
        const out = execSync('vm_stat', { encoding: 'utf8', timeout: 2000 });
        const pageSize = Number(/page size of (\d+) bytes/.exec(out)?.[1] ?? 4096);
        const pages = (label: string) => Number(new RegExp(label + ':\\s+(\\d+)\\.').exec(out)?.[1] ?? 0);
        // Match Activity Monitor's "Memory Used" = active + wired + compressed;
        // available = total - that. freemem()'s "free pages only" ignored the
        // large inactive/purgeable/cached pools macOS reclaims on demand, so it
        // read ~99% used. Verified against boite's live vm_stat (2026-07-05).
        const used = (pages('Pages active') + pages('Pages wired down') + pages('Pages occupied by compressor')) * pageSize;
        if (used > 0) bytes = Math.max(0, totalmem() - used);
      }
    } catch { /* keep freemem() fallback */ }
    this.#availMemCache = { bytes, at: now };
    return bytes;
  }

  private pushDaemonState(): void {
    if (!this.socket) return;
    const cpu = this.sampleCpuPercent();
    const memTotal = totalmem();
    const memFree = this.availableMemBytes(); // reclaimable-aware (see helper)
    const ram = Math.max(0, Math.min(100, Math.round((1 - memFree / memTotal) * 100)));
    const list = cpus();
    // Disk for the home filesystem (best-effort — statfs can fail on odd mounts).
    let diskFree = 0, diskTotal = 0;
    try {
      const s = statfsSync(homedir());
      diskFree = Number(s.bavail) * Number(s.bsize);
      diskTotal = Number(s.blocks) * Number(s.bsize);
    } catch { /* leave 0 — app shows cpu/ram regardless */ }
    const { variant, key } = this.machineCrypto();
    const daemonState = b64encode(encryptWire(variant, key, {
      cpu, ram, time: Date.now(),
      // Detail for the machine page (bytes + cpu info); the sidebar still uses cpu/ram %.
      cpuCount: list.length,
      cpuModel: list[0]?.model,
      load: loadavg()[0],
      memFree, memTotal,
      diskFree, diskTotal,
    }));
    this.socket.emit(
      'machine-update-state',
      { machineId: this.creds.machineId, daemonState, expectedVersion: this.daemonStateVersion },
      (ack: unknown) => {
        if (!isObj(ack)) return;
        const a = ack as { result?: string; version?: number };
        // success → server bumped to expectedVersion+1; version-mismatch → adopt
        // the server's current version so the next beat lands.
        if ((a.result === 'success' || a.result === 'version-mismatch') && typeof a.version === 'number') {
          this.daemonStateVersion = a.version;
        }
      },
    );
  }

  connect(): void {
    if (this.socket) return;
    this.socket = io(this.creds.serverUrl, {
      path: '/v1/updates',
      transports: ['websocket'],
      auth: { token: this.creds.token, clientType: 'machine-scoped', machineId: this.creds.machineId },
      reconnection: true,
      reconnectionDelay: 1_000,
      reconnectionDelayMax: 10_000,
    });
    let firstConnect = true;
    this.socket.on('connect', () => {
      log('socket connected');
      // Re-assert each session's CURRENT thinking value (not a blanket false —
      // that desynced from Session.#thinking, so the pane poll's change-gate
      // never re-asserted a true mid-turn and the app stuck on "not working").
      for (const rs of this.activeSessions) rs.reassertAlive();
      // Re-register all RPC handlers on (re)connect
      for (const method of this.rpcHandlers.keys()) {
        log(`rpc: re-registering ${method}`);
        this.socket?.emit('rpc-register', { method });
      }
      this.startMachineAlive();
      if (!firstConnect) this.onReconnect?.();
      firstConnect = false;
    });
    this.socket.on('disconnect', (r: string) => log(`socket disconnected: ${r}`));
    this.socket.on('connect_error', (e: Error) => log(`socket connect_error: ${e.message}`));
    this.socket.on('update', (p: unknown) => this.handlePoke(p));
    this.socket.on('rpc-request', async (req: unknown, callback: (res: string) => void) => {
      if (!isObj(req)) return;
      const method = String(req['method'] ?? '');
      log(`rpc: incoming request method=${method}`);
      const handler = this.rpcHandlers.get(method);
      const key = this.creds.encryption.type === 'dataKey'
        ? this.creds.encryption.machineKey
        : this.creds.encryption.secret;
      if (!handler) {
        callback(encryptRpc(key, { error: 'Method not found' }));
        return;
      }
      try {
        const params = decryptRpc(key, String(req['params'] ?? ''));
        const result = await handler(params);
        callback(encryptRpc(key, result));
      } catch (e) {
        callback(encryptRpc(key, { error: String(e) }));
      }
    });
  }

  close(): void { this.socket?.close(); this.socket = null; }

  /** Emit the server's version-checked session metadata update (used for summaries). */
  updateSessionMetadata(sid: string, expectedVersion: number, metadataB64: string): Promise<{ result: string; version?: number; metadata?: string } | null> {
    return new Promise((resolve) => {
      if (!this.socket) { resolve(null); return; }
      let done = false;
      const finish = (v: { result: string; version?: number; metadata?: string } | null) => { if (!done) { done = true; resolve(v); } };
      this.socket.emit('update-metadata', { sid, expectedVersion, metadata: metadataB64 }, (ack: unknown) => {
        finish(isObj(ack) ? (ack as { result: string; version?: number; metadata?: string }) : null);
      });
      setTimeout(() => finish(null), 5000);
    });
  }

  private handlePoke(payload: unknown): void {
    const sid = isObj(payload) ? String(payload['sessionId'] ?? '') : '';
    if (sid && this.listeners.has(sid)) {
      for (const cb of this.listeners.get(sid)!) try { cb(); } catch {}
      return;
    }
    for (const s of this.listeners.values()) for (const cb of s) try { cb(); } catch {}
  }

  subscribe(sessionId: string, onPoke: () => void): () => void {
    if (!this.listeners.has(sessionId)) this.listeners.set(sessionId, new Set());
    this.listeners.get(sessionId)!.add(onPoke);
    return () => {
      const s = this.listeners.get(sessionId);
      if (s) { s.delete(onPoke); if (!s.size) this.listeners.delete(sessionId); }
    };
  }

  private url(path: string): string {
    return `${this.creds.serverUrl.replace(/\/$/, '')}${path}`;
  }

  private headers(): Record<string, string> {
    return { Authorization: `Bearer ${this.creds.token}`, 'Content-Type': 'application/json' };
  }

  /**
   * Download + decrypt an attachment blob. Mirrors happy-cli's
   * downloadAndDecryptAttachment. Two-step flow:
   *   1. POST /v1/sessions/{id}/attachments/request-download → { downloadUrl }
   *   2. GET downloadUrl → encrypted bytes (NaCl secretbox bundle)
   * The blob key is derived from the session's encryption key with the
   * "Happy Blobs" usage and a variant-specific path. Returns null on any
   * failure (network, auth, decryption).
   */
  async downloadAndDecryptAttachment(
    relaySessionId: string,
    ref: string,
    sessionKey: Uint8Array,
    variant: EncryptionVariant,
  ): Promise<Uint8Array | null> {
    try {
      // Step 1: request a download URL (the server may presign an S3 URL or
      // hand back a self-served path requiring our bearer token).
      const reqRes = await fetch(this.url(`/v1/sessions/${relaySessionId}/attachments/request-download`), {
        method: 'POST',
        headers: this.headers(),
        body: JSON.stringify({ ref }),
      });
      if (!reqRes.ok) return null;
      const reqData = await reqRes.json() as { downloadUrl?: string };
      if (!reqData.downloadUrl) return null;

      // Step 2: fetch the encrypted bytes. Only send bearer when the URL
      // points back at our server — S3 presigned URLs reject extra headers.
      const isServerUrl = reqData.downloadUrl.startsWith(this.creds.serverUrl);
      const dlRes = await fetch(reqData.downloadUrl, {
        headers: isServerUrl ? { Authorization: `Bearer ${this.creds.token}` } : {},
      });
      if (!dlRes.ok) return null;
      const encrypted = new Uint8Array(await dlRes.arrayBuffer());

      // Step 3: decrypt with the per-session blob key.
      // Legacy sessions: deriveKey(secret, 'Happy Blobs', ['master']).
      // DataKey sessions: deriveKey(dataKey, 'Happy Blobs', ['session']).
      const path = variant === 'dataKey' ? ['session'] : ['master'];
      const blobKey = deriveKey(sessionKey, 'Happy Blobs', path);
      return decryptBlob(encrypted, blobKey);
    } catch (e) {
      log(`downloadAndDecryptAttachment failed for ${ref}: ${e}`);
      return null;
    }
  }

  /**
   * Upsert the machine row's metadata server-side. Mirrors happy-cli's
   * `getOrCreateMachine` (api.ts:144): POST /v1/machines with the
   * machineId and an encrypted MachineMetadata payload.
   *
   * The point of doing this from joy-tmux is purely a UX guarantee:
   * the app's path picker uses `selectedMachine.metadata.homeDir` to
   * format paths as `~/foo`. If happy-cli's daemon has never run on
   * this host, that field is undefined and the picker shows literal
   * `~/foo`. joy-tmux always knows its homedir, so we can guarantee
   * the field is set.
   *
   * Caveat: this REST POST is an unconditional upsert — the server
   * replaces the full metadata blob. If happy-cli's daemon had set
   * optional fields (cliAvailability, resumeSupport), our upsert
   * wipes them until happy-cli re-upserts on its next session spawn.
   * Acceptable trade-off for the picker reliability gain.
   */
  /** GET this machine's current (decrypted) metadata, so a re-upsert can carry
   *  forward app-owned fields instead of clobbering them. Best-effort → null. */
  // Last known app-set machine displayName (null = confirmed absent) — see
  // getOrCreateMachine for the TTL rationale.
  #displayNameCache: { name: string | null; at: number } | null = null;

  private async fetchOwnMachineMetadata(): Promise<Record<string, unknown> | null> {
    try {
      const res = await fetch(this.url('/v1/machines'), { headers: this.headers() });
      if (!res.ok) return null;
      const rows = await res.json() as Array<{ id: string; metadata?: string }>;
      const row = Array.isArray(rows) ? rows.find((m) => m.id === this.creds.machineId) : undefined;
      if (!row?.metadata) return null;
      const { variant, key } = this.machineCrypto();
      return decryptWire(variant, key, b64decode(row.metadata)) as Record<string, unknown> | null;
    } catch { return null; }
  }

  async getOrCreateMachine(metadata: Record<string, unknown>): Promise<boolean> {
    // Always report the LIVE hostname (an OS rename shouldn't need a daemon
    // restart), and never clobber an app-set displayName: this is a full-blob
    // upsert, so carry the current displayName forward when the caller didn't
    // supply one (the daemon's base blob never has it → it would otherwise wipe
    // a rename on every command-scan push).
    const blob: Record<string, unknown> = { ...metadata, host: hostname() };
    if (blob.displayName === undefined) {
      // Short-TTL cache: boot attaches many sessions and each push landed here,
      // re-downloading the FULL machine list per push just to carry one string
      // forward. 60s is long enough to dedup a burst, short enough that an
      // app-side rename (which happens on the server, invisibly to us) isn't
      // clobbered by a stale value for more than a minute.
      const now = Date.now();
      if (!this.#displayNameCache || now - this.#displayNameCache.at > 60_000) {
        const current = await this.fetchOwnMachineMetadata();
        const dn = current?.displayName;
        this.#displayNameCache = { name: typeof dn === 'string' && dn.length > 0 ? dn : null, at: now };
      }
      if (this.#displayNameCache.name) blob.displayName = this.#displayNameCache.name;
    }
    try {
      let encryptionKey: Uint8Array;
      let variant: EncryptionVariant;
      let dataEncryptionKeyB64: string | undefined;

      if (this.creds.encryption.type === 'dataKey') {
        variant = 'dataKey';
        encryptionKey = this.creds.encryption.machineKey;
        // Same envelope as createSession: [0x00][encrypted(machineKey, publicKey)]
        // so the server can hand the dataKey to authorized clients.
        const encryptedKey = libsodiumEncryptForPublicKey(encryptionKey, this.creds.encryption.publicKey);
        const bundle = new Uint8Array(1 + encryptedKey.length);
        bundle.set([0], 0);
        bundle.set(encryptedKey, 1);
        dataEncryptionKeyB64 = b64encode(bundle);
      } else {
        variant = 'legacy';
        encryptionKey = this.creds.encryption.secret;
      }

      const r = await fetch(this.url('/v1/machines'), {
        method: 'POST',
        headers: this.headers(),
        body: JSON.stringify({
          id: this.creds.machineId,
          metadata: b64encode(encryptWire(variant, encryptionKey, blob)),
          dataEncryptionKey: dataEncryptionKeyB64,
        }),
      });
      if (!r.ok) return false;
      // Seed the daemonState CAS version from the row so the first
      // machine-update-state beat lands without a version-mismatch round-trip.
      try {
        const body = await r.json() as { machine?: { daemonStateVersion?: number } };
        if (typeof body?.machine?.daemonStateVersion === 'number') {
          this.daemonStateVersion = body.machine.daemonStateVersion;
        }
      } catch { /* version self-syncs from the first ack */ }
      return true;
    } catch (e) {
      log(`getOrCreateMachine failed: ${e}`);
      return false;
    }
  }

  /**
   * Mark a session inactive on the API server (flips `active=false`).
   * Mirrors happy-cli's `deactivateSession`: POST /v1/sessions/{id}/archive.
   * Without this the session keeps showing as active in the app even after
   * the underlying tmux window has been killed, because killSession only
   * cleans up local state — it doesn't tell the server to archive.
   */
  async archiveSession(relaySessionId: string): Promise<boolean> {
    // Retry: a transient 5xx/network blip used to drop the archive entirely
    // (it was fire-and-forget), leaving a killed session stuck in the app's
    // active list. 404/410 = already gone = success; other 4xx = permanent.
    for (let attempt = 0; attempt < 4; attempt++) {
      try {
        const r = await fetch(this.url(`/v1/sessions/${relaySessionId}/archive`), {
          method: 'POST',
          headers: this.headers(),
          body: '{}',
        });
        if (r.ok || r.status === 404 || r.status === 410) return true;
        if (r.status >= 400 && r.status < 500) { log(`archiveSession ${relaySessionId}: HTTP ${r.status} (permanent)`); return false; }
        log(`archiveSession ${relaySessionId}: HTTP ${r.status} (attempt ${attempt + 1})`);
      } catch (e) {
        log(`archiveSession failed for ${relaySessionId} (attempt ${attempt + 1}): ${e}`);
      }
      if (attempt < 3) await sleep(500 * 2 ** attempt);
    }
    return false;
  }

  /**
   * Send a push notification to all the user's devices (mirrors happy-cli's
   * `notify`): fetch the account's Expo push tokens from the server (authed
   * with the daemon's bearer), then POST the messages straight to Expo. Returns
   * how many devices were targeted.
   */
  async sendPush(title: string, body: string): Promise<{ sent: number }> {
    const res = await fetch(this.url('/v1/push-tokens'), { headers: this.headers() });
    if (!res.ok) throw new Error(`push-tokens: HTTP ${res.status}`);
    const data = await res.json() as { tokens?: { token: string }[] };
    const tokens = (data.tokens ?? []).map(t => t.token).filter(Boolean);
    if (tokens.length === 0) return { sent: 0 };
    // One request PER TOKEN: Expo rejects a batch that mixes projects
    // (PUSH_TOO_MANY_EXPERIENCE_IDS), and an account that ever installed
    // another Expo app (the upstream happy app) holds mixed tokens — one
    // zombie token used to 400 the WHOLE batch and silently kill every
    // notification (found 2026-07-05: three stale @bulkacorp/happy tokens
    // blacked out all pushes).
    let sent = 0;
    for (const to of tokens) {
      try {
        const r = await fetch('https://exp.host/--/api/v2/push/send', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json', Accept: 'application/json' },
          body: JSON.stringify([{ to, title, body: body || undefined, sound: 'default', data: { source: 'joy-cli', timestamp: Date.now() } }]),
        });
        if (r.ok) {
          // HTTP 200 only means Expo ACCEPTED the request — per-ticket errors
          // (DeviceNotRegistered for a deleted app's token, etc.) ride in the
          // body and used to be counted as "sent".
          const j = await r.json().catch(() => null) as { data?: Array<{ status?: string; message?: string }> } | null;
          const ticket = j?.data?.[0];
          if (ticket?.status === "ok") sent++;
          else log(`push to ${to.slice(0, 28)}…: ticket ${ticket?.status ?? "?"} ${ticket?.message ?? ""}`);
        }
        else log(`push to ${to.slice(0, 28)}…: HTTP ${r.status}`);
      } catch (e) { log(`push to ${to.slice(0, 28)}…: ${e}`); }
    }
    return { sent };
  }

  /**
   * Auto-notification for a session (mirrors happy-cli's sendSessionNotification):
   * POST a push-event to the server, which decides whether to actually push —
   * suppressing it when the app is focused on this session. This is why we go
   * through the server instead of sendPush (which always blasts every device).
   */
  async sendSessionPushEvent(sessionId: string, kind: 'done' | 'permission' | 'question', title: string, body: string): Promise<void> {
    const body_ = JSON.stringify({ kind, title, body, data: { kind, sessionId, source: 'joy-tmux' } });
    // One retry: the very first outbound request after a daemon restart can hit a
    // transient "fetch failed" before the network/undici pool warms; a dropped
    // notification is user-visible, so give it a second shot.
    for (let attempt = 0; attempt < 2; attempt++) {
      try {
        const res = await fetch(this.url(`/v1/sessions/${encodeURIComponent(sessionId)}/push-event`), {
          method: 'POST', headers: this.headers(), body: body_,
        });
        // Success IS logged: "no push arrived" must be distinguishable between
        // never-sent, server-suppressed (app focused), and delivery failure.
        if (res.ok) { log(`push-event ${kind} sent for ${sessionId} (server decides suppression)`); return; }
        if (res.status >= 400 && res.status < 500) { log(`push-event ${kind} for ${sessionId}: HTTP ${res.status}`); return; }
        log(`push-event ${kind} for ${sessionId}: HTTP ${res.status} (attempt ${attempt + 1})`);
      } catch (e) {
        log(`push-event ${kind} failed for ${sessionId} (attempt ${attempt + 1}): ${e}`);
      }
      if (attempt === 0) await sleep(1000);
    }
  }

  async createSession(opts: { tag: string; metadata: unknown }): Promise<CreateSessionResult> {
    let sessionKey: Uint8Array;
    let variant: EncryptionVariant;
    let dataEncryptionKeyB64: string | null = null;

    if (this.creds.encryption.type === 'dataKey') {
      // machineKey is the stable per-machine symmetric key stored in access.key.
      // Using it as the sessionKey ensures messages can be decrypted across restarts,
      // even when the server deduplicates sessions by tag and returns an existing session ID.
      sessionKey = this.creds.encryption.machineKey;
      variant = 'dataKey';
      const encryptedKey = libsodiumEncryptForPublicKey(sessionKey, this.creds.encryption.publicKey);
      const bundle = new Uint8Array(1 + encryptedKey.length);
      bundle.set([0], 0); bundle.set(encryptedKey, 1);
      dataEncryptionKeyB64 = b64encode(bundle);
    } else {
      sessionKey = this.creds.encryption.secret;
      variant = 'legacy';
    }

    const res = await fetch(this.url('/v1/sessions'), {
      method: 'POST',
      headers: this.headers(),
      body: JSON.stringify({
        tag: opts.tag,
        metadata: b64encode(encryptWire(variant, sessionKey, opts.metadata)),
        agentState: null,
        dataEncryptionKey: dataEncryptionKeyB64,
      }),
    });
    if (!res.ok) throw new Error(`createSession: HTTP ${res.status}`);
    const data = await res.json() as { session: { id: string; metadata?: string | null; metadataVersion?: number } };
    let serverMeta: Record<string, unknown> | null = null;
    if (data.session.metadata) {
      try { serverMeta = decryptWire(variant, sessionKey, b64decode(data.session.metadata)) as Record<string, unknown> | null; } catch { serverMeta = null; }
    }
    return { sessionId: data.session.id, sessionKey, variant, metadata: serverMeta, metadataVersion: data.session.metadataVersion ?? 0 };
  }

  async append(sessionId: string, encrypted: Uint8Array, localId: string): Promise<{ seq: number }> {
    const res = await fetch(this.url(`/v3/sessions/${encodeURIComponent(sessionId)}/messages`), {
      method: 'POST',
      headers: this.headers(),
      body: JSON.stringify({ messages: [{ content: b64encode(encrypted), localId }] }),
    });
    if (!res.ok) {
      const err = new Error(`append: HTTP ${res.status}`) as Error & { status?: number };
      err.status = res.status; // lets drain() distinguish permanent (4xx) from transient
      throw err;
    }
    const data = await res.json() as { messages: Array<{ seq: number }> };
    if (!data.messages?.[0]) throw new Error('append: empty response');
    return data.messages[0];
  }

  async readSince(sessionId: string, afterSeq: number, limit = 100): Promise<{ messages: RawMessage[]; hasMore: boolean }> {
    const res = await fetch(
      this.url(`/v3/sessions/${encodeURIComponent(sessionId)}/messages?after_seq=${afterSeq}&limit=${limit}`),
      { headers: this.headers() },
    );
    if (!res.ok) throw new Error(`readSince: HTTP ${res.status}`);
    return res.json() as Promise<{ messages: RawMessage[]; hasMore: boolean }>;
  }

  async fetchLastSeq(sessionId: string): Promise<number> {
    let lastSeq = 0;
    let hasMore = true;
    while (hasMore) {
      const r = await this.readSince(sessionId, lastSeq, 100);
      for (const m of r.messages) if (m.seq > lastSeq) lastSeq = m.seq;
      hasMore = r.hasMore;
    }
    return lastSeq;
  }

  emitAlive(sessionId: string, thinking: boolean): void {
    this.socket?.volatile.emit('session-alive', {
      sid: sessionId, time: Date.now(), thinking, mode: 'remote',
    });
  }

  decryptMessage(msg: RawMessage, key: Uint8Array, variant: EncryptionVariant): unknown | null {
    const b64 = rawContentB64(msg.content);
    if (!b64) return null;
    try { return decryptWire(variant, key, b64decode(b64)); } catch { return null; }
  }
}

// ── Relay session (per tmux session) ─────────────────────────────────────────

/** Lifecycle state the app reads from metadata to colour a session's status. */
export type JoyLifecycleState = 'running' | 'detached' | 'archived';

/** Retry banner the app renders during 500-error auto-retry. */
export interface JoyRetryInfo {
  attempt: number;   // 1-based current attempt
  total: number;     // total attempts before giving up
  nextAt: number;    // epoch ms when the next re-send fires
  status: number;    // the HTTP status that triggered the retry (e.g. 500)
}

/** Compaction banner the app renders while Claude summarizes the conversation. */
export interface JoyCompactingInfo {
  trigger: 'auto' | 'manual'; // what kicked off the compaction
  since: number;              // epoch ms when compaction started
}

/**
 * Background-task progress ("N/M completed") the app shows while
 * run_in_background bash / background agents are in flight. Tracked from the
 * transcript so "working" stays continuous across turn-end (the foreground turn
 * ends the moment a background task launches) — independent of the brittle pane
 * footer poll.
 */
export interface JoyTasksInfo {
  done: number;   // finished in this batch
  total: number;  // launched in this batch
}

/**
 * The agent's active goal (Claude's `/goal`), surfaced from the transcript's
 * `goal_status` attachments. Present while a goal is in progress (met=false);
 * cleared to null when the goal is met/cleared. The app shows a goal bar.
 */
export interface JoyGoalInfo {
  condition: string; // the goal text the user set
  since: number;     // epoch ms when this goal became active
}

/**
 * An interactive auth/login URL the agent's CLI is showing in its pane (e.g.
 * Claude Code's `/login` OAuth flow). Detected by scanning the pane for an
 * auth-shaped URL; present while the prompt is up, cleared when it's gone. The
 * app shows a login bar with the URL + a field to submit the pasted code.
 */
export interface JoyLoginInfo {
  url: string;     // the reassembled auth URL
  since: number;   // epoch ms when it was first detected
  error?: string;  // a rejection/error message shown in the box (e.g. bad code)
}

/** Message-queue snapshot the app reads from metadata (replaces joy-queue-list polling). */
export interface JoyQueueInfo {
  queue: { id: string; text: string; createdAt: number }[];
  inFlight: string | null;
  paused: boolean;
}

export class RelaySession {
  private readonly client: RelayClient;
  readonly relaySessionId: string;
  private readonly sessionKey: Uint8Array;
  private readonly variant: EncryptionVariant;
  private lastSeq: number;
  private queue: Array<{ localId: string; wire: WireRecord; attempts: number }> = [];
  private draining = false;
  // Last-known session metadata blob + its server version, so we can merge in
  // a summary update (Claude's ai-title) without clobbering the other fields.
  private metadata: Record<string, unknown> | null;
  private metadataVersion = 0;

  onMessage: (text: string, seq: number) => void | Promise<void> = () => {};
  /**
   * Fires for each file/attachment event the app sends ahead of a user
   * message (envelope `role:'session'`, `ev.t:'file'`). The handler is
   * expected to download + decrypt the blob (via the parent RelayClient)
   * and stash it for the next user message.
   */
  onFileEvent: (ev: { ref: string; name: string; size: number; mimeType?: string }) => void = () => {};
  /** Exposed so handlers (e.g. attachment download) can derive blob keys. */
  get encryptionMaterial(): { sessionKey: Uint8Array; variant: EncryptionVariant } {
    return { sessionKey: this.sessionKey, variant: this.variant };
  }

  constructor(opts: {
    client: RelayClient;
    relaySessionId: string;
    sessionKey: Uint8Array;
    variant: EncryptionVariant;
    initialSeq?: number;
    metadata?: Record<string, unknown> | null;
    metadataVersion?: number;
  }) {
    this.client = opts.client;
    this.relaySessionId = opts.relaySessionId;
    this.sessionKey = opts.sessionKey;
    this.variant = opts.variant;
    this.lastSeq = opts.initialSeq ?? 0;
    this.metadata = opts.metadata ?? null;
    this.metadataVersion = opts.metadataVersion ?? 0;
  }

  // Serializes metadata writes for this session (see mergeMetadata).
  private metadataChain: Promise<void> = Promise.resolve();

  /**
   * Merge a patch into the session metadata and persist it (server is
   * version-checked; retries on mismatch). The single write path so the title,
   * lifecycle state, etc. don't clobber each other.
   *
   * Calls are SERIALIZED per session: concurrent patches (joy__state / joy__queue
   * / joy__retry / summary all fire near-simultaneously from the Session) used to
   * each read the same base metadata and race — the loser re-sent its stale blob
   * on the version-mismatch retry and erased the winner's field (e.g. a `detached`
   * state wiped by a queue update, leaving a dead session shown as live). The
   * daemon is the only writer of session metadata, so serializing here means each
   * merge reads an already-updated `this.metadata` and no field is lost.
   */
  private mergeMetadata(patch: Record<string, unknown>): Promise<void> {
    const run = this.metadataChain.then(() => this.doMergeMetadata(patch));
    this.metadataChain = run.catch(() => {}); // keep the chain alive past a failure
    return run;
  }

  private async doMergeMetadata(patch: Record<string, unknown>): Promise<void> {
    if (!this.metadata) return;
    for (let attempt = 0; attempt < 3; attempt++) {
      // Recompute merged + enc from CURRENT metadata each attempt, so a
      // version bump re-merges the patch onto fresh state instead of re-sending
      // a stale blob.
      const merged = { ...this.metadata, ...patch };
      const enc = b64encode(encryptWire(this.variant, this.sessionKey, merged));
      const ack = await this.client.updateSessionMetadata(this.relaySessionId, this.metadataVersion, enc);
      if (!ack) continue; // timeout/no-ack — retry rather than silently dropping the transition
      if (ack.result === 'success') {
        this.metadata = merged;
        if (typeof ack.version === 'number') this.metadataVersion = ack.version;
        return;
      }
      if (ack.result === 'version-mismatch' && typeof ack.version === 'number') {
        this.metadataVersion = ack.version;
        if (ack.metadata) {
          try {
            const dec = decryptWire(this.variant, this.sessionKey, b64decode(ack.metadata));
            if (dec && typeof dec === 'object') {
              this.metadata = dec as Record<string, unknown>;
            }
          } catch {}
        }
        continue; // retry with the server's current version
      }
      return;
    }
  }

  /**
   * Set the session's title (summary) — joy uses Claude's ai-title so the app
   * shows the real conversation title instead of "New Chat".
   */
  async updateSummary(title: string): Promise<void> {
    const current = this.metadata?.summary as { text?: string } | undefined;
    if (current?.text === title) return; // unchanged
    await this.mergeMetadata({ summary: { text: title, updatedAt: Date.now() } });
  }

  /**
   * Mirror the active model the app reads (session.metadata.currentModelCode) so
   * the composer's model selector reflects the real pane state — including model
   * switches the user makes interactively in the tmux pane (/model …).
   */
  async updateModelCode(code: string): Promise<void> {
    if ((this.metadata?.currentModelCode as string | undefined) === code) return;
    await this.mergeMetadata({ currentModelCode: code });
  }

  /**
   * Mirror the slash commands available to this session so the app's "/"
   * autocomplete (sync/suggestionCommands.ts reads metadata.slashCommands)
   * lists the real project + personal commands, not just the built-in
   * defaults. Caller passes a sorted list; skip the write when unchanged.
   */
  async updateSlashCommands(commands: string[]): Promise<void> {
    const current = this.metadata?.slashCommands as string[] | undefined;
    if (current && current.length === commands.length && current.every((c, i) => c === commands[i])) return;
    await this.mergeMetadata({ slashCommands: commands });
  }

  /**
   * Set the joy lifecycle state the app reads to colour the status:
   * 'running' (alive), 'detached' (Claude died, window still around — red), or
   * 'archived' (killed/cleaned up). Drives the red detached indicator.
   */
  async updateJoyState(state: JoyLifecycleState): Promise<void> {
    if ((this.metadata?.joy__state as string | undefined) === state) return;
    await this.mergeMetadata({ joy__state: state });
  }

  /**
   * Set (or clear, with null) the 500-error auto-retry banner the app shows
   * while a failed turn is being re-sent on a backoff schedule.
   */
  async updateRetry(info: JoyRetryInfo | null): Promise<void> {
    // Idempotent clear: skip the write when there's nothing set, so a recovered
    // session with no active retry can reconcile its banner without churn.
    if (info == null && this.metadata?.joy__retry == null) return;
    await this.mergeMetadata({ joy__retry: info });
  }

  /** Push the finishing-task N/M and the long-running-process count together in a
   *  SINGLE metadata patch, so the split can't be observed half-applied and a
   *  clear of one field can't be dropped by a still-pending set of the other.
   *  The caller (Session#reconcileBgTasks) dedups by DESIRED state, so we always
   *  write when called (no this.metadata skip that could race a pending write). */
  async updateBgTasks(tasks: JoyTasksInfo | null, agents: JoyTasksInfo | null, longRunning: number | null): Promise<void> {
    await this.mergeMetadata({ joy__tasks: tasks, joy__agents: agents, joy__longRunning: longRunning });
  }

  /** Context tokens used as of the latest turn (input + cache-read + cache-create
   *  from the transcript's cumulative usage). The app owns the window/threshold —
   *  we only report the raw count. null clears it. */
  async updateContext(tokens: number | null): Promise<void> {
    if (tokens == null && this.metadata?.joy__context == null) return;
    await this.mergeMetadata({ joy__context: tokens });
  }

  async updateCompacting(info: JoyCompactingInfo | null): Promise<void> {
    if (info == null && this.metadata?.joy__compacting == null) return;
    await this.mergeMetadata({ joy__compacting: info });
  }

  async updateGoal(info: JoyGoalInfo | null): Promise<void> {
    if (info == null && this.metadata?.joy__goal == null) return;
    await this.mergeMetadata({ joy__goal: info });
  }

  async updateLogin(info: JoyLoginInfo | null): Promise<void> {
    if (info == null && this.metadata?.joy__login == null) return;
    await this.mergeMetadata({ joy__login: info });
  }

  /**
   * Push the message-queue snapshot so the app can read it from metadata instead
   * of polling. Skips redundant writes: an empty queue when none is set, or a
   * snapshot identical to the current one (the queue broadcasts can repeat).
   */
  async updateQueue(info: JoyQueueInfo): Promise<void> {
    const empty = info.queue.length === 0 && !info.inFlight && !info.paused;
    const cur = this.metadata?.joy__queue as JoyQueueInfo | null | undefined;
    if (empty && cur == null) return;
    if (cur && JSON.stringify(cur) === JSON.stringify(info)) return;
    await this.mergeMetadata({ joy__queue: info });
  }

  private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
  private unsubscribe: (() => void) | null = null;

  start(): void {
    this.client.trackSession(this);
    this.unsubscribe = this.client.subscribe(this.relaySessionId, () => void this.pull());
    this.client.emitAlive(this.relaySessionId, false);
    // Poll for incoming messages every 3s (machine-scoped socket doesn't receive update pokes)
    // and send keepalive every 30s
    let ticks = 0;
    this.heartbeatTimer = setInterval(() => {
      void this.pull();
      // Keepalive every 30s — re-assert the CURRENT thinking state, not a
      // hardcoded false (which used to stomp "thinking" mid-turn).
      if (++ticks % 10 === 0) this.client.emitAlive(this.relaySessionId, this.lastThinking);
    }, 3_000);
  }

  /**
   * Downgrade a DETACHED session's heartbeat: drop the 3s message pull (a dead
   * pane ignores messages anyway — the poll is the daemon's only message path
   * because the machine-scoped socket never receives new-message pokes, but
   * that only matters for LIVE sessions) while keeping the 30s presence
   * keepalive, so the app still shows red "detached" instead of offline.
   * Without this, every detached session that ever accumulates polls the
   * server over HTTPS every 3s forever.
   */
  pausePull(): void {
    if (this.heartbeatTimer) clearInterval(this.heartbeatTimer);
    this.heartbeatTimer = setInterval(() => {
      this.client.emitAlive(this.relaySessionId, this.lastThinking);
    }, 30_000);
  }

  stop(): void {
    if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer = null; }
    this.unsubscribe?.();
    this.unsubscribe = null;
    this.client.unregisterSessionRpcHandlers(this.relaySessionId);
    this.client.untrackSession(this);
  }

  private pulling = false;

  // Consecutive decrypt-failure counts per seq (quarantine after 3 rounds).

  private decryptFailures = new Map<number, number>();

  private async pull(): Promise<void> {
    // Re-entrancy guard (like drain()): pull() fires from both the 3s heartbeat
    // and the poke subscription. Without this, a second pull could read the next
    // seq window and inject a later text message into the pane while an earlier
    // message is still parked awaiting its attachment download — reordering
    // app→Claude messages. Serialize so onMessage runs strictly in seq order.
    if (this.pulling) return;
    this.pulling = true;
    try {
      let { messages, hasMore } = await this.client.readSince(this.relaySessionId, this.lastSeq);
      if (messages.length > 0) log(`pull ${this.relaySessionId}: got ${messages.length} msgs, lastSeq=${this.lastSeq}`);
      let advanced = false;
      while (messages.length > 0) {
        const roundStart = this.lastSeq;
        for (const msg of messages) {
          // M2: decrypt before advancing seq so a failed decrypt doesn't silently consume the seq.
          // Quarantine backstop: a message that fails decrypt on several separate
          // pull rounds (key rotation, foreign-key history on a tag-dedup'd
          // session) is advanced past — with the old skip-forever behavior, a
          // whole page of undecryptable rows made readSince return the identical
          // page every round: a zero-delay hot loop that hammered the server and
          // blocked the session's inbound messages until a daemon restart.
          const dec = this.client.decryptMessage(msg, this.sessionKey, this.variant);
          if (dec === null) {
            const attempts = (this.decryptFailures.get(msg.seq) ?? 0) + 1;
            if (attempts >= 3) {
              this.decryptFailures.delete(msg.seq);
              log(`pull msg seq=${msg.seq} DECRYPT_FAILED x${attempts} — quarantining (advancing past it)`);
              if (msg.seq > this.lastSeq) { this.lastSeq = msg.seq; advanced = true; }
            } else {
              this.decryptFailures.set(msg.seq, attempts);
              log(`pull msg seq=${msg.seq} DECRYPT_FAILED (attempt ${attempts}) — skipping without advancing seq`);
            }
            continue;
          }
          this.decryptFailures.delete(msg.seq);
          if (msg.seq > this.lastSeq) { this.lastSeq = msg.seq; advanced = true; }
          log(`pull msg seq=${msg.seq} dec=${JSON.stringify(dec)?.slice(0, 120)}`);
          if (!isObj(dec)) continue;
          const role = dec['role'];

          // File attachment envelope: role='session', ev.t='file'. App sends
          // these ahead of the user-text message; the handler downloads and
          // stashes them, drained on the next text message.
          if (role === 'session') {
            const c = dec['content'] as { type?: string; data?: { ev?: { t?: string; ref?: string; name?: string; size?: number; mimeType?: string } } } | undefined;
            const ev = c?.data?.ev;
            if (c?.type === 'session' && ev?.t === 'file' && typeof ev.ref === 'string' && typeof ev.name === 'string' && typeof ev.size === 'number') {
              this.onFileEvent({ ref: ev.ref, name: ev.name, size: ev.size, mimeType: ev.mimeType });
            }
            continue;
          }

          if (role !== 'user') continue;
          // H1: skip messages sent by joy itself to avoid double-injecting into tmux.
          // Log the skip: this is the ONLY silent drop on the pull path, and an
          // app message swallowed here (mis-tagged meta) is otherwise undiagnosable
          // (the 2026-07-04 lost-message hunt had no way to see 6004's meta).
          const meta = dec['meta'] as { sentFrom?: string } | undefined;
          if (meta?.sentFrom === 'joy') {
            log(`pull: skip own-send seq=${msg.seq} sentFrom=joy`);
            continue;
          }
          if (meta?.sentFrom) log(`pull: user msg seq=${msg.seq} sentFrom=${meta.sentFrom}`);
          const c = dec['content'] as { type?: string; text?: string } | undefined;
          if (c?.type === 'text' && typeof c.text === 'string' && c.text.trim()) {
            // onMessage now hands the text to the session's in-memory verified
            // dispatch queue (it no longer types straight into tmux); the queue
            // does the actual ready+empty-box gated typing afterward. We still
            // AWAIT it so messages enter the queue in strict seq order (an earlier
            // one awaiting an attachment can't be overtaken by a later text-only
            // one). NOTE: lastSeq is advanced earlier, right after decrypt (see
            // above) — i.e. before this delivery is even attempted — so the
            // persisted cursor means "read + decrypted + handed off attempted",
            // NOT "delivered to Claude". Caveat (B1): a crash between that advance
            // and the enqueue (or items still queued-but-untyped) loses those
            // messages; a durable replay (B2) that re-pulls from a confirmed-seq
            // watermark and skips already-delivered ones is a separate project. A
            // failed hand-off is logged, not retried, to avoid a poison-pill loop.
            try {
              await this.onMessage(c.text.trim(), msg.seq);
            } catch (e) {
              log(`pull: onMessage failed for seq=${msg.seq}: ${e}`);
            }
          }
        }
        if (!hasMore) break;
        // Spin guard: if this whole page advanced the cursor by nothing (every
        // row undecryptable and not yet quarantined), refetching from the same
        // cursor returns the identical page — break and let the next 3s pull
        // tick retry (which also steps the quarantine counters) instead of
        // hot-looping inside one round.
        if (this.lastSeq === roundStart) break;
        ({ messages, hasMore } = await this.client.readSince(this.relaySessionId, this.lastSeq));
      }
      // Low: only write to disk when seq actually changed
      if (advanced) savePersistedSeq(this.relaySessionId, this.lastSeq);
    } catch (e) { log(`pull error for ${this.relaySessionId}: ${e}`); }
    finally { this.pulling = false; }
  }

  send(wire: WireRecord): void {
    this.queue.push({ localId: crypto.randomUUID(), wire, attempts: 0 });
    void this.drain();
  }

  private static readonly MAX_SEND_ATTEMPTS = 10;

  private async drain(): Promise<void> {
    if (this.draining) return;
    this.draining = true;
    while (this.queue.length > 0) {
      const item = this.queue[0];
      try {
        const enc = encryptWire(this.variant, this.sessionKey, item.wire);
        await this.client.append(this.relaySessionId, enc, item.localId);
        this.queue.shift();
      } catch (e) {
        item.attempts++;
        // A permanent (4xx) failure — e.g. the session was archived (404/410) or
        // auth rejected (401/403) — will never succeed, so drop it IMMEDIATELY
        // instead of backing off ~2.5min × 10 on it AND head-of-line-blocking every
        // queued message behind it. 408/429 are retryable. (H2: cap retries too so
        // a transient failure can't wedge the queue forever.)
        const status = (e as { status?: number }).status;
        const permanent = typeof status === "number" && status >= 400 && status < 500 && status !== 408 && status !== 429;
        if (permanent || item.attempts >= RelaySession.MAX_SEND_ATTEMPTS) {
          log(`send ${permanent ? `permanently failed (HTTP ${status})` : `failed after ${item.attempts} attempts`}, dropping: ${e}`);
          this.queue.shift();
          continue;
        }
        const delay = Math.min(500 * 2 ** item.attempts, 30_000);
        log(`send failed (attempt ${item.attempts}), retrying in ${delay}ms: ${e}`);
        await sleep(delay);
      }
    }
    this.draining = false;
  }

  /** Last thinking value we emitted — so the keepalive heartbeat re-asserts the
   *  REAL state instead of stomping it to false every 30s. */
  private lastThinking = false;

  setThinking(thinking: boolean): void {
    const changed = this.lastThinking !== thinking;
    this.lastThinking = thinking;
    this.client.emitAlive(this.relaySessionId, thinking);
    // ALSO persist on change: the ephemeral only reaches currently-connected
    // clients and the server never stores it, so a cold app start showed every
    // session idle until the next 30s keepalive happened to land ("close the
    // app and the status is lost"). The app treats joy__thinking + live
    // presence as thinking, so state survives restarts; presence gating keeps
    // a daemon death from freezing a stale blue.
    if (changed) {
      void this.mergeMetadata({ joy__thinking: thinking ? { since: Date.now() } : null }).catch(() => {});
    }
  }

  /** Clear a stale persisted joy__thinking (daemon restarted while the flag
   *  was set; the fresh Session starts not-thinking so the change-gate in
   *  setThinking would never write the false). Attach-time reconcile, same
   *  pattern as the retry/compacting banners. */
  async clearThinkingMeta(): Promise<void> {
    if (this.metadata?.joy__thinking == null) return;
    await this.mergeMetadata({ joy__thinking: null });
  }

  /** Re-emit presence with the CURRENT thinking value — used on socket
   *  reconnect to refresh the app without clobbering the real state. */
  reassertAlive(): void {
    this.client.emitAlive(this.relaySessionId, this.lastThinking);
  }

  /** Fire an auto push-notification for this session (done/permission/question).
   *  Title is the location "<host>/<folder>" (e.g. "faraz.vip/proj") so you see
   *  WHICH session at a glance; body is the AI summary (or the per-kind reason).
   *  The server suppresses it when the app is focused on this session. */
  /** Agent-authored notification (<joy-notify/> tag): free-form title + body,
   *  same server-side focus suppression as the automatic pushes. Title falls
   *  back to the session's host/folder so a push always identifies its source;
   *  when the agent titles it, the location rides in the body suffix instead. */
  notifyCustom(headline: string, detail: string | null): void {
    // Project-prefixed headline ("joy: Deploy finished") so every push reads
    // as <where>: <what> at a glance; detail (when given) is the body.
    const path = (this.metadata?.path as string | undefined)?.trim();
    const folder = path ? path.split(/[\\/]/).filter(Boolean).pop() : undefined;
    const finalTitle = folder ? `${folder}: ${headline}` : headline;
    void this.client.sendSessionPushEvent(this.relaySessionId, 'done', finalTitle, detail ?? headline);
  }

  notify(kind: 'done' | 'permission' | 'question', snippet?: string): void {
    const summary = (this.metadata?.summary as { text?: string } | undefined)?.text?.trim();
    // Prefer the caller's snippet (the reply's first line) — the session
    // summary alone read as project-name noise, telling the user nothing
    // about what actually happened.
    const body = kind === 'done' ? (snippet || summary || 'Finished')
      : kind === 'permission' ? (summary ? `Permission needed · ${summary}` : 'Permission needed')
      : (summary ? `Clarification needed · ${summary}` : 'Clarification needed');
    void this.client.sendSessionPushEvent(this.relaySessionId, kind, this.#notifyLocation(), body);
  }

  /** "<host>/<folder>" — e.g. "faraz.vip/proj". Metadata only (push title/body
   *  are NOT end-to-end encrypted, so no conversation content here). */
  #notifyLocation(): string {
    const host = (this.metadata?.host as string | undefined)?.trim();
    const path = (this.metadata?.path as string | undefined)?.trim();
    const folder = path ? path.split(/[\\/]/).filter(Boolean).pop() : undefined;
    return [host, folder].filter(Boolean).join('/') || 'Joy session';
  }

  /** Register a session-scoped RPC handler bound to this relay session. */
  registerRpc(method: string, handler: (params: unknown) => Promise<unknown>): void {
    this.client.registerSessionRpcHandler(this.relaySessionId, method, handler);
  }
}

// ── Wire encoding ──────────────────────────────────────────────────────────────

// opts.time is Claude's transcript timestamp (epoch ms) for this entry. The
// app sorts agent events by this embedded time, so stamping it from the
// transcript (not Date.now at mirror time) keeps a --resume replay in true
// chronological order. Falls back to now() when a caller omits it.
function sessionEnvelope(ev: Record<string, unknown>, opts: { turn: string; claudeUuid?: string; time?: number }): WireRecord {
  const data: Record<string, unknown> = {
    id: crypto.randomUUID(),
    time: opts.time ?? Date.now(),
    role: 'agent',
    turn: opts.turn,
    ev,
  };
  if (opts.claudeUuid) data.claudeUuid = opts.claudeUuid;
  return { role: 'session', content: { type: 'session', data }, meta: { sentFrom: 'joy' } };
}

export function encodeTurnStart(opts: { turn: string; claudeUuid?: string; time?: number }): WireRecord {
  return sessionEnvelope({ t: 'turn-start' }, opts);
}

export function encodeTextEvent(text: string, opts: { turn: string; claudeUuid?: string; time?: number }): WireRecord {
  return sessionEnvelope({ t: 'text', text }, opts);
}

export function encodeToolCallStart(opts: {
  call: string; name: string; input: unknown; turn: string; claudeUuid?: string; time?: number;
}): WireRecord {
  return sessionEnvelope({
    t: 'tool-call-start',
    call: opts.call,
    name: opts.name,
    title: opts.name,
    description: '',
    args: (opts.input && typeof opts.input === 'object') ? opts.input as Record<string, unknown> : {},
  }, opts);
}

export function encodeToolCallEnd(call: string, opts: { turn: string; claudeUuid?: string; time?: number }): WireRecord {
  return sessionEnvelope({ t: 'tool-call-end', call }, opts);
}

export function encodeTurnEnd(status: 'completed' | 'failed' | 'cancelled', opts: { turn: string; time?: number; usage?: unknown }): WireRecord {
  // Carry the turn's final token usage so the app can show real tokens/cost for
  // joy sessions (the raw transcript assistant entries report it as msg.usage).
  const ev: Record<string, unknown> = { t: 'turn-end', status };
  if (opts.usage) ev.usage = opts.usage;
  return sessionEnvelope(ev, opts);
}

// User messages are sorted by the relay's server-assigned createdAt, NOT an
// embedded time — so on a replay burst they'd diverge from agent events (a
// different clock). joyTime carries Claude's transcript timestamp; joy-app
// reads it (MessageMetaSchema.joyTime) and orders joy user messages by it,
// putting both sides on one clock.
export function encodeUserMessage(text: string, timeMs?: number): WireRecord {
  return { role: 'user', content: { type: 'text', text }, meta: { sentFrom: 'joy', joyTime: timeMs } };
}

// ── Bootstrap ─────────────────────────────────────────────────────────────────

export function initRelay(): RelayClient | null {
  const creds = loadCredentials();
  if (!creds) { log('no credentials found — relay disabled'); return null; }
  const client = new RelayClient(creds);
  client.connect();
  log(`initialized → ${creds.serverUrl}`);
  return client;
}

export async function createRelaySession(
  client: RelayClient,
  opts: { tag: string; cwd: string; id: string; state?: JoyLifecycleState },
): Promise<RelaySession> {
  const state = opts.state ?? 'running';
  const metadata = { path: opts.cwd, host: hostname(), version: '0.1.0', machineId: client.creds.machineId, joy__source: 'joy-tmux', joy__sessionId: opts.id, joy__state: state };
  const result = await client.createSession({ tag: opts.tag, metadata });
  // Resume from persisted seq so messages sent during downtime are delivered.
  // On first-ever start (no saved seq), fetch to end to avoid replaying history.
  let initialSeq = loadPersistedSeq(result.sessionId);
  if (initialSeq === 0) {
    initialSeq = await client.fetchLastSeq(result.sessionId);
    if (initialSeq > 0) savePersistedSeq(result.sessionId, initialSeq);
  }
  // On tag-dedup the server kept its EXISTING metadata (title, prior joy__state).
  // Use it as the base so we don't wipe the title; fall back to ours for a
  // brand-new session. Then push the intended state (a no-op if unchanged).
  const baseMeta = result.metadata ?? metadata;
  const rs = new RelaySession({
    client,
    relaySessionId: result.sessionId,
    sessionKey: result.sessionKey,
    variant: result.variant,
    initialSeq,
    metadata: baseMeta,
    metadataVersion: result.metadataVersion,
  });
  await rs.updateJoyState(state);
  return rs;
}

// ── lastSeq persistence ───────────────────────────────────────────────────────
// Stores the last processed seq per relay session so messages sent during
// joy-tmux downtime are delivered on next startup, not silently skipped.

function seqStatePath(relaySessionId: string): string {
  const dir = joyStateDir();
  if (!existsSync(dir)) mkdirSync(dir, { recursive: true });
  return join(dir, `${relaySessionId}.seq`);
}

function loadPersistedSeq(relaySessionId: string): number {
  try {
    const p = seqStatePath(relaySessionId);
    if (existsSync(p)) return parseInt(readFileSync(p, 'utf8').trim(), 10) || 0;
  } catch {}
  return 0;
}

function savePersistedSeq(relaySessionId: string, seq: number): void {
  // Low: write-temp-then-rename for atomic update — crash mid-write can't corrupt
  try {
    const p = seqStatePath(relaySessionId);
    const tmp = p + '.tmp';
    writeFileSync(tmp, String(seq));
    renameSync(tmp, p);
  } catch {}
}

// ── Util ──────────────────────────────────────────────────────────────────────

function log(msg: string): void { process.stderr.write(`[relay] ${msg}\n`); }

function isObj(v: unknown): v is Record<string, unknown> {
  return typeof v === 'object' && v !== null;
}