import { closeSync, lstatSync, mkdirSync, openSync, readdirSync, readSync, statSync } from "node:fs";
import { errMessage, stringValue } from "agent-relay-sdk";
import { basename, dirname, extname, isAbsolute, join, relative, resolve } from "node:path";
import type { ServerWebSocket } from "bun";
import { proxyArtifactRequest } from "./artifact-proxy";
import { fireAndForget } from "./async-guard";
import type { OrchestratorConfig } from "./config";
import type { CommandLoopHealth } from "./command-poller";
import type { ProviderProbeCache } from "./provider-probe";
import type { RegistrationDriver } from "./registration";
import type { RelayClient } from "./relay";
import { captureSession, captureSessionMirror, captureTerminal, createTerminalGuest, listSessions, sendTerminalInput, resizeTerminal, stopTerminalGuest, validateTerminalInputData, validateTerminalResize } from "./spawn";
import { acquireTerminalStream, type TerminalStreamHandle, type TerminalStreamSubscriber } from "./terminal-stream";
import { VERSION, runtimeMetadata } from "./version";
import { branchMergePreviewResponse, mergePreviewResponse, recoveryBranchesResponse, probeWorkspace, workspaceDiff, workspaceGitState } from "./workspace-probe";

interface DirectoryEntry {
  name: string;
  path: string;
}

interface DirectoryListing {
  path: string;
  parent?: string;
  baseDir: string;
  entries: DirectoryEntry[];
}

export type GitFileStatus = "staged" | "unstaged" | "staged-unstaged" | "untracked";

export interface FileEntry {
  name: string;
  path: string;
  type: "file" | "directory" | "symlink";
  /** True when the directory entry is itself a symlink (resolved type is reported in `type`). */
  symlink?: boolean;
  /** Working-tree git status for this entry (or aggregate, for directories). */
  gitStatus?: GitFileStatus;
  size?: number;
  modifiedAt?: number;
}

export interface FileListing {
  path: string;
  parent?: string;
  baseDir: string;
  entries: FileEntry[];
}

export interface FileReadResult {
  path: string;
  name: string;
  mediaType: string;
  encoding: "utf8" | "binary" | "base64";
  size: number;
  modifiedAt?: number;
  truncated: boolean;
  content?: string;
}

export interface FileStatResult {
  path: string;
  name: string;
  type: "file" | "directory";
  baseDir: string;
  size?: number;
  modifiedAt?: number;
}

const MAX_FILE_PREVIEW_BYTES = 1024 * 1024;
const MAX_IMAGE_PREVIEW_BYTES = 8 * 1024 * 1024;

interface TerminalSocketData {
  kind: "terminal";
  config: OrchestratorConfig;
  session: string;
  stream?: TerminalStreamHandle;
  paused?: boolean;
  // Backfill is deferred until the client reports its size, so the captured pane
  // wraps identically to the viewer's xterm. Until then, live bytes queue.
  synced?: boolean;
  ready?: boolean;
  queue?: Uint8Array[];
  syncTimer?: ReturnType<typeof setTimeout>;
  // Bytes dropped (not sent) while this viewer was paused. On resume, 0 means the
  // screen is unchanged → resume the live stream without a reset+backfill (which
  // otherwise wipes scrollback/scroll position on every focus flick — #272).
  droppedWhilePaused?: number;
}

type TerminalSocket = ServerWebSocket<TerminalSocketData>;

function listDirectories(requestedPath: string | undefined, baseDir: string): DirectoryListing {
  const base = resolve(baseDir);
  const target = resolve(requestedPath || base);
  const rel = relative(base, target);

  if (rel && (rel.startsWith("..") || rel.startsWith("/"))) {
    throw new Error(`Path must be within baseDir: ${baseDir}`);
  }

  let stat;
  try { stat = statSync(target); } catch { throw new Error(`Path does not exist: ${target}`); }
  if (!stat.isDirectory()) throw new Error(`Not a directory: ${target}`);

  const entries = readdirSync(target, { withFileTypes: true })
    .filter((e) => e.isDirectory() && !e.name.startsWith("."))
    .map((e) => ({ name: e.name, path: join(target, e.name) }))
    .sort((a, b) => a.name.localeCompare(b.name));

  const parent = dirname(target);
  const parentRel = relative(base, parent);
  return {
    path: target,
    parent: parentRel && !parentRel.startsWith("..") && !parentRel.startsWith("/") && parent !== target ? parent : undefined,
    baseDir: base,
    entries,
  };
}

function createDirectory(parentPath: string, name: string, baseDir: string): DirectoryListing {
  const base = resolve(baseDir);
  const parent = resolve(parentPath);
  const parentRel = relative(base, parent);
  if (parentRel && (parentRel.startsWith("..") || parentRel.startsWith("/"))) {
    throw new Error(`Path must be within baseDir: ${baseDir}`);
  }
  if (!name || name.includes("/") || name.includes("\\") || name.startsWith(".")) {
    throw new Error(`Invalid directory name: ${name}`);
  }
  const target = join(parent, name);
  mkdirSync(target, { recursive: true });
  return listDirectories(target, baseDir);
}

// Containment is lexical (resolve + relative), per the project's path-containment
// rule. This intentionally follows symlinks wherever they point: the `path` query
// param itself can't escape baseDir with `../`, but symlinks placed inside baseDir
// by the operator are treated as first-class entries (same trust level as the
// authenticated dashboard, which can already open a host terminal).
function resolveInsideBase(requestedPath: string | undefined, baseDir: string): { base: string; target: string } {
  const base = resolve(baseDir);
  const target = resolve(requestedPath || base);
  const rel = relative(base, target);
  if (rel && (rel.startsWith("..") || isAbsolute(rel))) {
    throw new Error(`Path must be within baseDir: ${baseDir}`);
  }
  try {
    statSync(target);
  } catch {
    throw new Error(`Path does not exist: ${target}`);
  }
  return { base, target };
}

function parentInsideBase(target: string, base: string): string | undefined {
  const parent = dirname(target);
  const parentRel = relative(base, parent);
  return parentRel && !parentRel.startsWith("..") && !parentRel.startsWith("/") && parent !== target ? parent : undefined;
}

export function listFiles(requestedPath: string | undefined, baseDir: string, options: { git?: boolean } = {}): FileListing {
  const { base, target } = resolveInsideBase(requestedPath, baseDir);
  const stat = statSync(target);
  if (!stat.isDirectory()) throw new Error(`Not a directory: ${target}`);

  // Perf: directories need no stat at all (Dirent already tells us the kind).
  // Files need one statSync for size/mtime; symlinks need one statSync (which
  // follows the link) to resolve their target kind. This replaces the previous
  // lstat + realpath + stat triple per entry.
  const entries = readdirSync(target, { withFileTypes: true })
    .map((entry): FileEntry => {
      const entryPath = join(target, entry.name);
      const isLink = entry.isSymbolicLink();
      const result: FileEntry = { name: entry.name, path: entryPath, type: "file" };
      if (isLink) result.symlink = true;

      if (isLink) {
        try {
          const st = statSync(entryPath); // follows the link
          result.type = st.isDirectory() ? "directory" : "file";
          if (st.isFile()) result.size = st.size;
          result.modifiedAt = st.mtimeMs;
        } catch {
          result.type = "symlink"; // broken link — surface it rather than hide it
          try { result.modifiedAt = lstatSync(entryPath).mtimeMs; } catch { /* ignore */ }
        }
      } else if (entry.isDirectory()) {
        result.type = "directory";
      } else {
        try {
          const st = statSync(entryPath);
          result.size = st.size;
          result.modifiedAt = st.mtimeMs;
        } catch { /* entry vanished mid-listing */ }
      }
      return result;
    })
    .sort((a, b) => {
      if (a.type === "directory" && b.type !== "directory") return -1;
      if (a.type !== "directory" && b.type === "directory") return 1;
      return a.name.localeCompare(b.name);
    });

  if (options.git) applyGitStatus(target, entries);

  return {
    path: target,
    parent: parentInsideBase(target, base),
    baseDir: base,
    entries,
  };
}

/** Overlay working-tree git status onto directory entries (best-effort, never throws). */
function applyGitStatus(target: string, entries: FileEntry[]): void {
  const root = runGitIn(["rev-parse", "--show-toplevel"], target);
  if (root === null) return; // not a git repo

  // `--no-renames` reports renames as delete+add, sidestepping the two-path -z
  // parse. The `.` pathspec scopes output to this directory's subtree.
  const raw = runGitIn(["status", "--porcelain", "-z", "--no-renames", "--", "."], target);
  if (!raw) return;

  // name -> { staged, unstaged, untracked }, aggregated across descendants so a
  // directory inherits the status of files changed beneath it.
  const agg = new Map<string, { staged: boolean; unstaged: boolean; untracked: boolean }>();
  for (const token of raw.split("\0")) {
    if (token.length < 4) continue;
    const x = token[0];
    const y = token[1];
    const p = token.slice(3);
    const abs = resolve(root, p);
    const rel = relative(target, abs);
    if (!rel || rel.startsWith("..") || isAbsolute(rel)) continue;
    const name = (rel.split("/")[0] ?? "").replace(/\/$/, "");
    if (!name) continue;
    const cur = agg.get(name) ?? { staged: false, unstaged: false, untracked: false };
    if (x === "?" && y === "?") cur.untracked = true;
    else {
      if (x !== " " && x !== "?") cur.staged = true;
      if (y !== " " && y !== "?") cur.unstaged = true;
    }
    agg.set(name, cur);
  }

  for (const entry of entries) {
    const a = agg.get(entry.name);
    if (!a) continue;
    if (a.staged && a.unstaged) entry.gitStatus = "staged-unstaged";
    else if (a.staged) entry.gitStatus = "staged";
    else if (a.unstaged) entry.gitStatus = "unstaged";
    else if (a.untracked) entry.gitStatus = "untracked";
  }
}

/** Run git in `cwd`, returning trimmed stdout, or null on failure. */
function runGitIn(args: string[], cwd: string): string | null {
  try {
    const proc = Bun.spawnSync(["git", "-C", cwd, ...args], { stdin: "ignore", stdout: "pipe", stderr: "ignore" });
    if (proc.exitCode !== 0) return null;
    return proc.stdout.toString().replace(/\n+$/, "");
  } catch {
    return null;
  }
}

export function statFilePath(requestedPath: string | undefined, baseDir: string): FileStatResult {
  if (!requestedPath) throw new Error("path is required");
  const { base, target } = resolveInsideBase(requestedPath, baseDir);
  const stat = statSync(target);
  if (!stat.isFile() && !stat.isDirectory()) throw new Error(`Not a file or directory: ${target}`);
  return {
    path: target,
    name: basename(target),
    type: stat.isDirectory() ? "directory" : "file",
    baseDir: base,
    ...(stat.isFile() ? { size: stat.size } : {}),
    modifiedAt: stat.mtimeMs,
  };
}

function readBytes(target: string, count: number): Buffer {
  const buffer = Buffer.alloc(count);
  const fd = openSync(target, "r");
  try {
    readSync(fd, buffer, 0, count, 0);
  } finally {
    closeSync(fd);
  }
  return buffer;
}

export function readFilePreview(requestedPath: string | undefined, baseDir: string): FileReadResult {
  if (!requestedPath) throw new Error("path is required");
  const { target } = resolveInsideBase(requestedPath, baseDir);
  const stat = statSync(target);
  if (!stat.isFile()) throw new Error(`Not a file: ${target}`);

  const mediaType = mediaTypeForPath(target);
  const resultBase = {
    path: target,
    name: basename(target),
    mediaType,
    size: stat.size,
    modifiedAt: stat.mtimeMs,
  };

  // Images are returned as base64 so the dashboard can render them inline. They
  // must be read whole — a truncated image would be corrupt — so they get a
  // higher cap and fall back to a plain binary marker when too large.
  if (mediaType.startsWith("image/")) {
    if (stat.size > MAX_IMAGE_PREVIEW_BYTES) {
      return { ...resultBase, truncated: true, encoding: "binary" };
    }
    return { ...resultBase, truncated: false, encoding: "base64", content: readBytes(target, stat.size).toString("base64") };
  }

  const buffer = readBytes(target, Math.min(stat.size, MAX_FILE_PREVIEW_BYTES));
  const truncated = stat.size > MAX_FILE_PREVIEW_BYTES;

  if (isBinaryBuffer(buffer)) {
    return { ...resultBase, truncated, encoding: "binary" };
  }

  return {
    ...resultBase,
    truncated,
    encoding: "utf8",
    content: new TextDecoder("utf-8", { fatal: true }).decode(buffer),
  };
}

function isBinaryBuffer(buffer: Buffer): boolean {
  if (buffer.includes(0)) return true;
  try {
    new TextDecoder("utf-8", { fatal: true }).decode(buffer);
    return false;
  } catch {
    return true;
  }
}

function mediaTypeForPath(path: string): string {
  const name = basename(path).toLowerCase();
  const ext = extname(path).toLowerCase();
  if (ext === ".md" || ext === ".markdown") return "text/markdown";
  if (ext === ".json" || ext === ".jsonl") return "application/json";
  if (ext === ".yaml" || ext === ".yml") return "application/yaml";
  if (ext === ".toml") return "application/toml";
  if (ext === ".html") return "text/html";
  if (ext === ".css") return "text/css";
  if (ext === ".js" || ext === ".mjs" || ext === ".cjs") return "text/javascript";
  if (ext === ".ts" || ext === ".tsx" || ext === ".jsx") return "text/typescript";
  if (ext === ".txt" || ext === ".log" || ext === ".sh" || name.startsWith(".env")) return "text/plain";
  if (ext === ".png") return "image/png";
  if (ext === ".jpg" || ext === ".jpeg") return "image/jpeg";
  if (ext === ".gif") return "image/gif";
  if (ext === ".webp") return "image/webp";
  if (ext === ".svg") return "image/svg+xml";
  if (ext === ".bmp") return "image/bmp";
  if (ext === ".ico") return "image/x-icon";
  if (ext === ".avif") return "image/avif";
  return "application/octet-stream";
}

function json(data: unknown, status = 200): Response {
  return new Response(JSON.stringify(data), {
    status,
    headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" },
  });
}

function error(message: string, status = 400): Response {
  return json({ error: message }, status);
}

function authorized(req: Request, config: OrchestratorConfig): boolean {
  if (!config.token) return true;
  const url = new URL(req.url);
  if (url.searchParams.get("token") === config.token) return true;
  return req.headers.get("x-agent-relay-token") === config.token;
}

// #1425 — the manual codex reset-consume entry point, satisfied by OrchestratorQuotaPoller.
// Kept as a narrow interface so the API server does not depend on the whole poller.
export interface CodexResetConsumer {
  manualConsumeResetCredit(input: { provider?: string; accountKey?: string; idempotencyKey: string }): Promise<{
    outcome: string;
    availableCount?: number;
    accountKey: string;
    provider: string;
  }>;
}

export function startApiServer(config: OrchestratorConfig, probeCache: ProviderProbeCache, relay?: RelayClient, codexResetConsumer?: CodexResetConsumer, registration?: Pick<RegistrationDriver, "getHealth">, commandLoop?: { getHealth(): CommandLoopHealth }): { stop(): void; url: string } {
  const server = Bun.serve<TerminalSocketData>({
    port: config.apiPort,
    hostname: "0.0.0.0",
    async fetch(req, server) {
      const url = new URL(req.url);

      if (req.method === "OPTIONS") {
        return new Response(null, {
          headers: {
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "GET, POST, HEAD, DELETE, OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type, X-Agent-Relay-Token, X-Artifact-Filename, X-Artifact-Digest, X-Artifact-Kind, X-Artifact-Sensitivity, X-Artifact-Expires-At",
          },
        });
      }

      // Runner self-heal: a runner whose runtime token expired proxies it here to
      // get a fresh one. No orchestrator-token auth (the runner doesn't hold it) —
      // the relay is the gate: it only re-mints a genuine, non-revoked runner token
      // owned by this orchestrator. The signed (even expired) token is the identity.
      if (req.method === "POST" && url.pathname === "/api/runtime-tokens/runner-renew") {
        if (!relay) return error("re-mint unavailable", 503);
        const body = await req.json().catch(() => null);
        const token = body && typeof body === "object" && typeof (body as { token?: unknown }).token === "string"
          ? (body as { token: string }).token
          : "";
        if (!token) return error("token required", 400);
        const reminted = await relay.remintRunnerToken(token);
        if (!reminted) return error("runner token re-mint failed", 502);
        return json(reminted);
      }

      // #1425 — manual codex banked-reset trigger. The relay (MCP tool / dashboard button) calls
      // this on the orchestrator that owns the codex app-server websocket. Orchestrator-token
      // gated like the other relay→orchestrator endpoints. The caller supplies the idempotency
      // key so a retried request can't double-consume.
      if (req.method === "POST" && url.pathname === "/api/codex/reset-consume") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        if (!codexResetConsumer) return error("codex reset-consume unavailable", 503);
        try {
          const body = await req.json().catch(() => null) as { idempotencyKey?: unknown; accountKey?: unknown; provider?: unknown } | null;
          const idempotencyKey = body && typeof body.idempotencyKey === "string" && body.idempotencyKey.trim()
            ? body.idempotencyKey.trim()
            : "";
          if (!idempotencyKey) return error("idempotencyKey required", 400);
          const accountKey = body && typeof body.accountKey === "string" && body.accountKey.trim() ? body.accountKey.trim() : undefined;
          // #1425 finding 4 — honor the relay's declared reset-credit provider so the consume can
          // only ever target that provider's account on this host, never a different provider's.
          const provider = body && typeof body.provider === "string" && body.provider.trim() ? body.provider.trim() : undefined;
          const result = await codexResetConsumer.manualConsumeResetCredit({ idempotencyKey, ...(accountKey ? { accountKey } : {}), ...(provider ? { provider } : {}) });
          return json(result);
        } catch (e) {
          return error((e as Error).message, 502);
        }
      }

      if (url.pathname === "/api/artifacts" || url.pathname.startsWith("/api/artifacts/")) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        return proxyArtifactRequest(req, config).catch((e) => error((e as Error).message, 502));
      }

      if (req.method === "GET" && url.pathname === "/api/directories") {
        try {
          const listing = listDirectories(url.searchParams.get("path") || undefined, config.baseDir);
          return json(listing);
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "POST" && url.pathname === "/api/directories") {
        try {
          const body = await req.json() as { path: string; name: string };
          if (!body.path || !body.name) return error("path and name are required");
          const listing = createDirectory(body.path, body.name, config.baseDir);
          return json(listing, 201);
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/files/list") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const git = url.searchParams.get("git") === "1";
          return json(listFiles(url.searchParams.get("path") || undefined, config.baseDir, { git }));
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/files/read") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          return json(readFilePreview(url.searchParams.get("path") || undefined, config.baseDir));
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/files/stat") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          return json(statFilePath(url.searchParams.get("path") || undefined, config.baseDir));
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/workspace/probe") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const { target } = resolveInsideBase(url.searchParams.get("path") || undefined, config.baseDir);
          return json(await probeWorkspace(target));
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/workspace/state") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const { target } = resolveInsideBase(url.searchParams.get("path") || undefined, config.baseDir);
          return json(await workspaceGitState({
            worktreePath: target,
            baseRef: url.searchParams.get("baseRef") || undefined,
            baseSha: url.searchParams.get("baseSha") || undefined,
          }));
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/workspace/diff") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const { target } = resolveInsideBase(url.searchParams.get("path") || undefined, config.baseDir);
          return json(await workspaceDiff({
            worktreePath: target,
            baseRef: url.searchParams.get("baseRef") || undefined,
            baseSha: url.searchParams.get("baseSha") || undefined,
            includePatch: url.searchParams.get("patch") !== "0",
          }));
        } catch (e) {
          return error((e as Error).message);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/workspace/merge-preview") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        return await mergePreviewResponse(url, config.baseDir);
      }

      if (req.method === "GET" && url.pathname === "/api/workspace/branch-merge-preview") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        return await branchMergePreviewResponse(url, config.baseDir);
      }

      if (req.method === "GET" && url.pathname === "/api/workspace/recovery-branches") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        return await recoveryBranchesResponse(url, config.baseDir);
      }

      if (req.method === "GET" && url.pathname === "/api/providers") {
        return (async () => {
          const snapshot = await probeCache.getSnapshot(url.searchParams.get("refresh") === "1");
          return json(snapshot);
        })();
      }

      if (req.method === "GET" && url.pathname === "/api/sessions") {
        return json({ sessions: listSessions(config.tmuxPrefix) });
      }

      if (req.method === "GET" && url.pathname === "/api/version") {
        const runtime = runtimeMetadata();
        return json({
          package: runtime.package,
          contracts: runtime.contracts,
          capabilities: runtime.capabilities,
          orchestrator: VERSION,
          runner: VERSION,
          adapters: Object.fromEntries(config.providers.map((provider) => [provider, VERSION])),
        });
      }

      const logMatch = url.pathname.match(/^\/api\/logs\/([^/]+)$/);
      if (req.method === "GET" && logMatch) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const session = decodeURIComponent(logMatch[1]!);
          const lines = Number(url.searchParams.get("lines") || "100");
          // ?stream=mirror returns the clean session-mirror diagnostics log instead
          // of the provider's ANSI TUI capture.
          if (url.searchParams.get("stream") === "mirror") {
            return json(captureSessionMirror(session, config, Number.isFinite(lines) ? lines : 200));
          }
          return json(captureSession(session, config, Number.isFinite(lines) ? lines : 100, {
            raw: url.searchParams.get("raw") === "1",
          }));
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      const terminalMatch = url.pathname.match(/^\/api\/terminal\/([^/]+)$/);
      if (req.method === "GET" && terminalMatch) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const session = decodeURIComponent(terminalMatch[1]!);
          return json(captureTerminal(session, config));
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      const terminalStreamMatch = url.pathname.match(/^\/api\/terminal\/([^/]+)\/stream$/);
      if (req.method === "GET" && terminalStreamMatch) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const session = decodeURIComponent(terminalStreamMatch[1]!);
          captureTerminal(session, config);
          const upgraded = server.upgrade(req, {
            data: { kind: "terminal", config, session },
          });
          if (!upgraded) return new Response("WebSocket upgrade failed", { status: 400 });
          return undefined;
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      const terminalInputMatch = url.pathname.match(/^\/api\/terminal\/([^/]+)\/input$/);
      if (req.method === "POST" && terminalInputMatch) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const session = decodeURIComponent(terminalInputMatch[1]!);
          const body = await req.json();
          return json(sendTerminalInput(session, config, body));
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      const terminalResizeMatch = url.pathname.match(/^\/api\/terminal\/([^/]+)\/resize$/);
      if (req.method === "POST" && terminalResizeMatch) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const session = decodeURIComponent(terminalResizeMatch[1]!);
          const body = await req.json();
          return json(resizeTerminal(session, config, body));
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      if (req.method === "POST" && url.pathname === "/api/terminal-guests") {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const body = await req.json();
          return json(await createTerminalGuest(cleanTerminalGuestInput(body), config), 201);
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      const terminalGuestMatch = url.pathname.match(/^\/api\/terminal-guests\/([^/]+)$/);
      if (req.method === "DELETE" && terminalGuestMatch) {
        if (!authorized(req, config)) return error("unauthorized", 401);
        try {
          const session = decodeURIComponent(terminalGuestMatch[1]!);
          return json(stopTerminalGuest(session, config));
        } catch (e) {
          return error((e as Error).message, 400);
        }
      }

      if (req.method === "GET" && url.pathname === "/api/health") {
        // #1676 — surface relay-peer and registration degradation here so a host stuck
        // re-registering is distinguishable from a healthy one WITHOUT log archaeology
        // (previously it looked identical, and spawns routed to it burned their full
        // timeout). `status`/`ok` deliberately stay "ok": this probe is the
        // Relay-INDEPENDENT local readiness signal the self-upgrade guard gates on, and a
        // degraded peer must not make this process look un-upgradable. Read `degraded`.
        const peer = relay?.getHealth();
        const reg = registration?.getHealth();
        // #1760 — surface the command loop's own health so a CRAWLING loop (a stuck inline handler
        // head-of-line-blocking the queue) is distinguishable from a healthy one. Kept OUT of the
        // top-level `degraded` flag on purpose: `degraded`/`ok` gate the self-upgrade guard's
        // Relay-independent readiness probe, and a transiently-stalled command loop must not make
        // this process look un-upgradable. Consumers read `commandLoop.stalled` directly.
        return json({
          status: "ok", ok: true, id: config.id, hostname: config.hostname,
          degraded: Boolean(peer?.degraded || reg?.degraded),
          ...(peer ? { relay: peer } : {}), ...(reg ? { registration: reg } : {}),
          ...(commandLoop ? { commandLoop: commandLoop.getHealth() } : {}),
        });
      }

      return error("Not found", 404);
    },
    websocket: {
      open(ws) {
        const socket = ws as unknown as TerminalSocket;
        if (socket.data?.kind === "terminal") startTerminalSocket(socket);
      },
      message(ws, data) {
        const socket = ws as unknown as TerminalSocket;
        if (socket.data?.kind === "terminal") handleTerminalSocketMessage(socket, data);
      },
      close(ws) {
        const socket = ws as unknown as TerminalSocket;
        if (socket.data?.kind === "terminal") stopTerminalSocket(socket);
      },
    },
  });

  const url = `http://${config.hostname}:${config.apiPort}`;
  console.error(`[orchestrator] API server listening on :${config.apiPort}`);
  return { stop: () => server.stop(), url };
}

function startTerminalSocket(ws: TerminalSocket): void {
  ws.data.queue = [];
  const subscriber: TerminalStreamSubscriber = {
    onData: (bytes) => {
      if (ws.data.paused) {
        // Count what we drop so resume can tell whether a re-backfill is needed.
        ws.data.droppedWhilePaused = (ws.data.droppedWhilePaused ?? 0) + bytes.length;
        return;
      }
      // Queue live bytes until the client has reported its size and the backfill is
      // flushed, so the viewer never sees live output before (or mis-sized against) history.
      if (!ws.data.ready) {
        ws.data.queue?.push(bytes);
        return;
      }
      try {
        ws.send(bytes);
      } catch {}
    },
    onClose: (reason) => {
      // Tell the client why, THEN actually close the socket. Sending only the frame
      // (the old behaviour) left a live-looking socket the client never reconnected
      // on — a backpressure drop made the terminal silently dead to input and output
      // (#271). The client reconnects + re-backfills on a transient reason.
      try {
        ws.send(JSON.stringify({ type: "closed", reason: reason ?? null }));
      } catch {}
      try {
        ws.close();
      } catch {}
    },
    bufferedAmount: () => {
      try {
        return ws.getBufferedAmount();
      } catch {
        return 0;
      }
    },
  };
  try {
    ws.data.stream = acquireTerminalStream(ws.data.session, ws.data.config, subscriber);
    // Wait for the client's resize to size the pane before backfilling. Fall back to
    // the pane's current size if no resize arrives (e.g. a non-fitting client).
    ws.data.syncTimer = setTimeout(() => {
      if (!ws.data.synced) void fireAndForget("Terminal socket sync/backfill", () => syncAndBackfill(ws));
    }, 700);
  } catch (e) {
    ws.send(JSON.stringify({ type: "error", error: errMessage(e) }));
    ws.close();
  }
}

// Backfill comes straight from tmux's own grid (see terminal-stream.ts): the snapshot's
// `content` is a `capture-pane -e` repaint of the current screen (styled, cursor parked).
// tmux is the real emulator and never drifts, so the repaint is byte-faithful and live
// relative deltas then apply identically on the client. Live bytes that arrive before the
// snapshot are already on tmux's grid (and thus in the capture), so we discard the queue
// and only forward bytes that arrive after.
async function syncAndBackfill(ws: TerminalSocket, cols?: number, rows?: number): Promise<void> {
  if (ws.data.syncTimer) {
    clearTimeout(ws.data.syncTimer);
    ws.data.syncTimer = undefined;
  }
  if (ws.data.synced) return;
  ws.data.synced = true;
  // Initial connect: live bytes are still queued (ready=false), so the reset+content can't
  // split a live sequence — emit immediately (no ground gate, no first-paint latency).
  await sendBackfill(ws, cols, rows);
  // But only START forwarding live output at a sequence boundary, so the first forwarded
  // flush doesn't begin mid-escape-sequence (an orphan tail on the client — #276).
  const stream = ws.data.stream;
  const flip = () => {
    ws.data.queue = [];
    ws.data.ready = true;
  };
  if (stream) stream.whenAtGround(flip);
  else flip();
}

// Send a full reset: a control frame with current geometry/status, then the serialized
// emulator state as a raw byte frame. Used on connect, resume, and refresh. On resume /
// refresh the live stream is flowing to this socket, so `gateGround` defers the
// reset+content until a sequence boundary — otherwise the reset would splice between the
// halves of a live escape sequence and orphan its tail on the client (#276).
async function sendBackfill(ws: TerminalSocket, cols?: number, rows?: number, gateGround = false): Promise<void> {
  const stream = ws.data.stream;
  if (!stream) return;
  const snapshot = await stream.backfill(cols, rows);
  if (ws.data.stream !== stream) return; // socket torn down mid-backfill
  const emit = () => {
    if (ws.data.stream !== stream) return;
    ws.send(JSON.stringify({
      type: "reset",
      session: snapshot.session,
      running: snapshot.running,
      agentAlive: snapshot.agentAlive,
      cols: snapshot.cols,
      rows: snapshot.rows,
      capturedAt: snapshot.capturedAt,
    }));
    if (snapshot.content) {
      ws.send(new TextEncoder().encode(snapshot.content));
    }
  };
  if (gateGround) await new Promise<void>((resolve) => stream.whenAtGround(() => { emit(); resolve(); }));
  else emit();
}

function handleTerminalSocketMessage(ws: TerminalSocket, data: string | Buffer): void {
  let payload: unknown;
  try {
    payload = JSON.parse(typeof data === "string" ? data : data.toString("utf8"));
  } catch {
    ws.send(JSON.stringify({ type: "error", error: "invalid terminal socket frame" }));
    return;
  }

  if (!payload || typeof payload !== "object" || Array.isArray(payload)) return;
  const frame = payload as Record<string, unknown>;
  try {
    if (frame.type === "input") {
      // Same envelope as the HTTP input route (#143): type + 4096-char cap. Invalid
      // frames throw → caught below → terminal error frame, tmux untouched.
      const text = validateTerminalInputData(frame);
      if (text) ws.data.stream?.write(new TextEncoder().encode(text));
    } else if (frame.type === "resize") {
      // Same bounds as the HTTP resize route (#143): cols 10-500, rows 5-200.
      const { cols, rows } = validateTerminalResize(frame);
      // First resize sizes the pane and triggers the (size-matched) backfill;
      // later ones just reflow the live stream.
      if (!ws.data.synced) {
        void fireAndForget("Terminal socket sync/backfill", () => syncAndBackfill(ws, cols, rows));
      } else {
        ws.data.stream?.resize(cols, rows);
      }
    } else if (frame.type === "pause") {
      const wasPaused = ws.data.paused === true;
      ws.data.paused = frame.paused === true;
      // On resume: only re-backfill if bytes were actually dropped while paused. A
      // focus flick (alt-tab) pauses+resumes with zero output in between — re-backfill
      // there needlessly wipes scrollback and scroll position (#272). When nothing was
      // dropped the client's grid already matches tmux, so just resume the live stream.
      if (wasPaused && !ws.data.paused && ws.data.synced) {
        const dropped = ws.data.droppedWhilePaused ?? 0;
        ws.data.droppedWhilePaused = 0;
        if (dropped > 0) void fireAndForget("Terminal socket backfill", () => sendBackfill(ws, undefined, undefined, true));
      }
    } else if (frame.type === "refresh") {
      if (ws.data.synced) void fireAndForget("Terminal socket backfill", () => sendBackfill(ws, undefined, undefined, true));
    } else if (frame.type === "interactive") {
      // Read-only watchers must not reflow the shared tmux window (#273); the typist owns
      // sizing. The client reports its interactivity so the stream knows who may resize.
      ws.data.stream?.setInteractive(frame.interactive === true);
    }
  } catch (e) {
    ws.send(JSON.stringify({ type: "error", error: errMessage(e) }));
  }
}

function stopTerminalSocket(ws: TerminalSocket): void {
  if (ws.data.syncTimer) {
    clearTimeout(ws.data.syncTimer);
    ws.data.syncTimer = undefined;
  }
  ws.data.stream?.release();
  ws.data.stream = undefined;
}

function cleanTerminalGuestInput(value: unknown): { agentId?: string; policyName?: string; spawnRequestId?: string; tmuxSession?: string } {
  if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error("terminal guest body must be an object");
  const body = value as Record<string, unknown>;
  const result = {
    agentId: stringValue(body.agentId),
    policyName: stringValue(body.policyName),
    spawnRequestId: stringValue(body.spawnRequestId),
    tmuxSession: stringValue(body.tmuxSession),
  };
  if (!result.agentId && !result.policyName && !result.spawnRequestId && !result.tmuxSession) {
    throw new Error("agentId, policyName, spawnRequestId, or tmuxSession required");
  }
  return result;
}