// Live terminal streaming via tmux control mode (`tmux -C`).
//
// One control-mode client is attached per tmux *session* and shared (ref-counted)
// across every connected viewer, so the cost is flat in viewer count: the control
// protocol is parsed once per session, then raw bytes fan out to all subscribers.
//
// Output: tmux emits `%output %<pane> <octal-escaped-bytes>` lines; we decode them
// to raw bytes, coalesce on a short window, and broadcast. No screen-scraping, no
// full-buffer repaint — the client just writes the byte stream to xterm.
//
// Backfill: tmux control mode does not replay the pre-attach screen, so on viewer
// attach we repaint from tmux's *own* grid via `capture-pane -e` (styled, current
// screen only — no scrollback, exactly what `tmux attach` shows) and park the cursor
// where tmux has it. tmux is the real terminal emulator: its grid never drifts, so the
// repaint is byte-faithful by definition, and subsequent live relative-cursor deltas
// apply identically on the client (which now mirrors tmux exactly). We deliberately do
// NOT keep a server-side @xterm/headless mirror: seeding such an emulator mid-stream
// from a snapshot and then feeding it Claude's purely-relative deltas accumulates error
// (the seed is never byte-perfect), which ghosted the screen into a staircase. tmux's
// grid is the only authoritative source, so we read it directly.
//
// Input/resize: written as plain command lines to the control client's stdin
// (`send-keys -H <hex>`, `resize-window`), so a keystroke costs no process spawn.
// The relay's own injection path (message delivery, /compact, initial prompts) uses
// its separate `tmux send-keys` calls and is unaffected — control mode is just
// another observing client.

import { sessionLiveness, tmuxCommand, tmuxSocketForSession, type TerminalSnapshot } from "./spawn";
import { TERMINAL_BACKFILL_SCROLLBACK_LINES, TERMINAL_BACKPRESSURE_MAX_BYTES, TERMINAL_COMMAND_TIMEOUT_MS, TERMINAL_DEBUG, TERMINAL_FLUSH_MAX_BYTES, TERMINAL_FLUSH_MS, TERMINAL_GROUND_WAIT_MAX_MS, TERMINAL_RESIZE_SETTLE_MS, TERMINAL_RESYNC_DEBOUNCE_MS, TERMINAL_RESYNC_GROUND_DEFER_MAX_MS, TERMINAL_RESYNC_GROUND_RETRY_MS, TERMINAL_RESYNC_MAX_INTERVAL_MS, type OrchestratorConfig } from "./config";
import { errMessage } from "agent-relay-sdk";
import { fireAndForget, guardedTimeout } from "./async-guard";

const FLUSH_MS = TERMINAL_FLUSH_MS;
const FLUSH_MAX_BYTES = TERMINAL_FLUSH_MAX_BYTES;
const BACKPRESSURE_MAX_BYTES = TERMINAL_BACKPRESSURE_MAX_BYTES;
// After a resize the TUI repaints asynchronously; let tmux's grid settle before we
// capture, so the backfill is the post-resize frame, not a half-reflowed one.
const RESIZE_SETTLE_MS = TERMINAL_RESIZE_SETTLE_MS;
// Live deltas are relative-cursor moves; replaying them onto a client seeded mid-stream
// from a capture-pane snapshot can drift (lost scroll-region/SGR/wrap state → doubled
// statusline, faded suggestion rendered solid, cursor off by a row). We can't transfer
// full emulator state, so we periodically re-stamp tmux's authoritative grid in place to
// snap the client back. Debounce after output settles; cap so continuous "thinking"
// streams still correct. 0 disables the corrector.
const RESYNC_DEBOUNCE_MS = TERMINAL_RESYNC_DEBOUNCE_MS;
const RESYNC_MAX_INTERVAL_MS = TERMINAL_RESYNC_MAX_INTERVAL_MS;
// When a resync falls due mid-escape-sequence we defer it (ground-state gate, #276) and
// re-check after this short interval until the stream reaches a sequence boundary.
const RESYNC_GROUND_RETRY_MS = TERMINAL_RESYNC_GROUND_RETRY_MS;
// Hard cap on ground-state deferral: if the stream sits mid-sequence this long (a stalled
// or dead pane — rare), inject the repaint anyway, CAN-prefixed to abort the client's
// half-parsed sequence. The orphan-tail risk is accepted in that degenerate case (#276).
const RESYNC_GROUND_DEFER_MAX_MS = TERMINAL_RESYNC_GROUND_DEFER_MAX_MS;
// Per-command reply timeout for the in-band control protocol. A reply that never lands
// means a desync; we reject and reset the reply queue so the next command starts clean.
const COMMAND_TIMEOUT_MS = TERMINAL_COMMAND_TIMEOUT_MS;
// Upper bound a backfill/ready-flip waits for the live stream to reach a sequence boundary
// before proceeding anyway (whenAtGround fallback for a stalled stream).
const GROUND_WAIT_MAX_MS = TERMINAL_GROUND_WAIT_MAX_MS;
// CAN (cancel) aborts a half-parsed sequence on the client.
const CAN_BYTE = 0x18;

interface PendingCommand {
  wantReply: boolean;
  lines: string[];
  resolve: (lines: string[]) => void;
  reject: (err: Error) => void;
  timer: ReturnType<typeof setTimeout> | null;
  // Fired SYNCHRONOUSLY when this command's reply block closes — before readLoop hands any
  // later same-chunk %output to enqueue(). The resync screen-capture uses it to snapshot the
  // pre-capture delta boundary (#275 dedup); see snapshotPreCaptureDeltas.
  onFinishSync?: () => void;
}
// On attach we include this many lines of tmux scrollback ABOVE the current screen so the
// viewer can scroll back through pre-attach history (the client lands on the live screen;
// the history sits in its scroll buffer). This is a one-time per-attach paint, so it's
// fine to be large; the live resync corrector only ever repaints the current screen, so it
// never disturbs this history. 0 = current screen only.
const BACKFILL_SCROLLBACK_LINES = TERMINAL_BACKFILL_SCROLLBACK_LINES;

export interface TerminalStreamSubscriber {
  onData(bytes: Uint8Array): void;
  onClose(reason?: string): void;
  // Bytes currently buffered on the subscriber's transport (e.g. WS bufferedAmount).
  // When it exceeds the backpressure cap the subscriber is dropped and expected to
  // reconnect and re-backfill — append streams can't drop bytes, so we shed the slow
  // client instead of ballooning orchestrator memory.
  bufferedAmount?(): number;
}

export interface TerminalStreamHandle {
  backfill(cols?: number, rows?: number): Promise<TerminalSnapshot>;
  write(bytes: Uint8Array): void;
  resize(cols: number, rows: number): void;
  // Declare whether this viewer is interactive (typing). Only the interactive viewer
  // sizes the shared tmux window, so read-only watchers can't reflow a working
  // terminal by attaching/refreshing at their own width (#273).
  setInteractive(interactive: boolean): void;
  // Run `cb` when the outbound stream is at an ANSI sequence boundary (or after a short
  // fallback). Used to gate reset/backfill injection so it can't split a live escape
  // sequence on the client (#276).
  whenAtGround(cb: () => void): void;
  release(): void;
}

const DEFAULT_COLS = 80;
const DEFAULT_ROWS = 24;
function tdbg(...args: unknown[]): void {
  if (TERMINAL_DEBUG) console.error("[term-debug]", ...args);
}

// --- Pure protocol helpers (unit-tested) ---

// Decode a tmux control-mode `%output` payload: printable ASCII is literal, every
// other byte is `\ooo` octal, and backslash is `\\`.
export function decodeControlOutput(data: string): Uint8Array {
  const out: number[] = [];
  for (let i = 0; i < data.length; ) {
    const ch = data[i]!;
    if (ch === "\\") {
      const next = data[i + 1];
      if (next === "\\") {
        out.push(0x5c);
        i += 2;
        continue;
      }
      let oct = "";
      let j = i + 1;
      while (j < data.length && oct.length < 3 && data[j]! >= "0" && data[j]! <= "7") {
        oct += data[j];
        j += 1;
      }
      if (oct.length > 0) {
        out.push(parseInt(oct, 8) & 0xff);
        i = j;
        continue;
      }
      out.push(0x5c);
      i += 1;
      continue;
    }
    out.push(ch.charCodeAt(0) & 0xff);
    i += 1;
  }
  return Uint8Array.from(out);
}

// Byte-faithful single-byte decode (true ISO-8859-1: byte N → U+00NN). We CANNOT use
// `new TextDecoder("latin1")` — per the WHATWG encoding standard "latin1" is an alias for
// windows-1252, which remaps 0x80–0x9F to printable code points (e.g. 0x96 → U+2013). Those
// bytes are exactly the UTF-8 continuation/lead octets of box-drawing & powerline glyphs
// (▐ = U+2590 = E2 96 90), so a windows-1252 read is NOT reversible via `charCodeAt & 0xff`
// and corrupts every multi-byte glyph (#270 regression). fromCharCode maps each byte 1:1, so
// the round-trip below is exact. Chunked to keep fromCharCode.apply off the arg-count limit.
export function decodeLatin1(bytes: Uint8Array): string {
  let out = "";
  const CHUNK = 0x8000;
  for (let i = 0; i < bytes.length; i += CHUNK) {
    out += String.fromCharCode.apply(null, bytes.subarray(i, i + CHUNK) as unknown as number[]);
  }
  return out;
}

// The control stream is read byte-faithfully (see decodeLatin1) so the `%output` octal path
// above sees raw bytes. But command-reply blocks (capture-pane grid rows) arrive as raw
// UTF-8, so their lines reach us as the byte-faithful latin1 representation. Re-decode them
// to real UTF-8 here, or multi-byte glyphs (box-drawing, powerline) double-encode on the
// next repaint and render as mojibake (#270). Safe to do per line: no UTF-8 continuation
// byte is 0x0A/0x0D, so the newline split never lands mid-sequence.
const REPLY_UTF8_DECODER = new TextDecoder("utf-8");
export function latin1LineToUtf8(line: string): string {
  const bytes = new Uint8Array(line.length);
  for (let i = 0; i < line.length; i++) bytes[i] = line.charCodeAt(i) & 0xff;
  return REPLY_UTF8_DECODER.decode(bytes);
}

type ControlLine =
  | { type: "output"; pane: string; bytes: Uint8Array }
  | { type: "exit"; reason?: string }
  // Command-reply block framing. Every command written to the control client's stdin is
  // answered, in order, with a `%begin … %end` (or `%error`) block carrying the same
  // command number on both ends. The content lines between them are the command's output
  // (e.g. capture-pane grid rows). Correlating these in-band serializes captures with
  // `%output` deltas — the ordering guarantee that kills the duplicate-text race (#275).
  | { type: "begin"; num: number }
  | { type: "end"; num: number }
  | { type: "error"; num: number }
  | { type: "other" };

// `%begin <timestamp> <cmd-number> <flags>` → the command number is the 2nd field.
function parseBlockNum(line: string, prefixLen: number): number {
  const n = Number(line.slice(prefixLen).trim().split(/\s+/)[1]);
  return Number.isFinite(n) ? n : -1;
}

export function parseControlLine(line: string): ControlLine {
  if (line.startsWith("%output ")) {
    const rest = line.slice(8);
    const sp = rest.indexOf(" ");
    const pane = sp === -1 ? rest : rest.slice(0, sp);
    const data = sp === -1 ? "" : rest.slice(sp + 1);
    return { type: "output", pane, bytes: decodeControlOutput(data) };
  }
  if (line.startsWith("%begin ")) return { type: "begin", num: parseBlockNum(line, 7) };
  if (line.startsWith("%end ")) return { type: "end", num: parseBlockNum(line, 5) };
  if (line.startsWith("%error ")) return { type: "error", num: parseBlockNum(line, 7) };
  if (line === "%exit" || line.startsWith("%exit ")) {
    const reason = line.slice(5).trim();
    return { type: "exit", ...(reason ? { reason } : {}) };
  }
  return { type: "other" };
}

// --- ANSI ground-state tracker (unit-tested) ---
//
// tmux chunks `%output` at pty-read / flush boundaries that can land mid-escape-sequence.
// That's fine for xterm (its parser is stateful across writes), but if we splice an
// out-of-band repaint (resync) BETWEEN the two halves of a split sequence, the injected
// ESC aborts the half-parsed CSI and the sequence's tail bytes then arrive in ground state
// and print as literal text — the stray `S` (final byte of `CSI Ps S`, scroll-up, which
// scroll-region TUIs like Claude Code emit constantly). So we track whether the outbound
// stream is at a sequence boundary (ground) and only inject there. This is a deliberately
// minimal VT state machine: enough to know "are we mid-sequence?", not a full parser.
type AnsiState = "ground" | "esc" | "esc-charset" | "csi" | "string" | "string-esc";

export function advanceAnsiState(state: AnsiState, byte: number): AnsiState {
  // CAN (0x18) / SUB (0x1a) abort any in-progress sequence from any state → ground.
  if (byte === 0x18 || byte === 0x1a) return "ground";
  switch (state) {
    case "ground":
      if (byte === 0x1b) return "esc";
      if (byte === 0x9b) return "csi"; // C1 CSI
      if (byte === 0x9d || byte === 0x90) return "string"; // C1 OSC / DCS
      return "ground"; // text, UTF-8 continuation bytes, lone C1 ST, etc.
    case "esc":
      if (byte === 0x5b) return "csi"; // '[' → CSI
      if (byte === 0x5d || byte === 0x50 || byte === 0x58 || byte === 0x5e || byte === 0x5f)
        return "string"; // ']' OSC, 'P' DCS, 'X' SOS, '^' PM, '_' APC → string until ST/BEL
      if (byte >= 0x28 && byte <= 0x2b) return "esc-charset"; // ( ) * + → next byte designates a charset
      if (byte >= 0x20 && byte <= 0x2f) return "esc"; // other intermediate, keep collecting
      return "ground"; // final byte of a 2-byte escape (ESC M, ESC 7, ESC =, …)
    case "esc-charset":
      return "ground"; // the single charset-designator byte
    case "csi":
      if (byte === 0x1b) return "esc"; // ESC cancels and restarts a sequence
      if (byte >= 0x40 && byte <= 0x7e) return "ground"; // final byte ends the CSI
      return "csi"; // parameter (0x30–0x3f) / intermediate (0x20–0x2f) / executed C0
    case "string":
      if (byte === 0x07 || byte === 0x9c) return "ground"; // BEL or C1 ST terminates
      if (byte === 0x1b) return "string-esc"; // possible 7-bit ST (ESC \)
      return "string";
    case "string-esc":
      if (byte === 0x5c) return "ground"; // ST: ESC \
      if (byte === 0x1b) return "string-esc";
      return "string"; // stray ESC inside the string — stay in string mode
  }
}

// Fold a chunk of outbound bytes into the running ANSI state. Returns the state AFTER the
// chunk, so the caller knows whether the stream currently sits at a sequence boundary.
export function scanAnsiState(bytes: Uint8Array, state: AnsiState = "ground"): AnsiState {
  for (let i = 0; i < bytes.length; i++) state = advanceAnsiState(state, bytes[i]!);
  return state;
}

// Parse a `#{pane_width} #{pane_height}` reply line into positive dims (omits non-finite).
export function parsePaneDims(line: string): { cols?: number; rows?: number } {
  const [w, h] = line.trim().split(/\s+/).map(Number);
  return {
    ...(w !== undefined && Number.isFinite(w) && w > 0 ? { cols: w } : {}),
    ...(h !== undefined && Number.isFinite(h) && h > 0 ? { rows: h } : {}),
  };
}

// Encode raw input bytes for `send-keys -H` (space-separated hex octets).
export function encodeSendKeysHex(bytes: Uint8Array): string {
  return Array.from(bytes)
    .map((b) => b.toString(16).padStart(2, "0"))
    .join(" ");
}

// Build a client repaint from tmux's current-screen grid: lay the rows out top-down
// (explicit CRLF so it's independent of the client's convertEol), then park the cursor
// where tmux has it so the next live relative delta continues from the right cell.
export function buildScreenRepaint(content: string, cursorX?: number, cursorY?: number): string {
  let out = content.replace(/\n$/, "").replace(/\n/g, "\r\n");
  if (cursorX != null && cursorY != null && Number.isFinite(cursorX) && Number.isFinite(cursorY)) {
    out += `\x1b[${cursorY + 1};${cursorX + 1}H`;
  }
  return out;
}

// Build an in-place authoritative repaint that overwrites the client grid with tmux's
// current screen WITHOUT a full-screen clear (so it can run mid-stream as a drift
// corrector without flicker). Each row is absolute-positioned, SGR-reset, and erased
// (`\x1b[2K`) before its styled cells are written, so any drift — a doubled statusline, a
// suggestion stuck solid instead of faded, leftover cells — is stamped out. The cursor is
// re-parked at tmux's true position and visibility restored. Wrapped in cursor-hide/show
// so the multi-row paint doesn't visibly skitter the cursor across the screen.
export function buildInPlaceRepaint(
  content: string,
  rows: number,
  cursor: { cursorX?: number; cursorY?: number; visible?: boolean } = {},
): string {
  const lines = content.replace(/\n$/, "").split("\n");
  const height = Number.isFinite(rows) && rows > 0 ? Math.trunc(rows) : lines.length;
  let out = "\x1b[?25l\x1b[m";
  for (let i = 0; i < height; i++) {
    out += `\x1b[${i + 1};1H\x1b[m\x1b[2K`;
    if (lines[i]) out += lines[i];
  }
  out += "\x1b[m";
  if (cursor.cursorX != null && cursor.cursorY != null && Number.isFinite(cursor.cursorX) && Number.isFinite(cursor.cursorY)) {
    out += `\x1b[${cursor.cursorY + 1};${cursor.cursorX + 1}H`;
  }
  out += cursor.visible === false ? "\x1b[?25l" : "\x1b[?25h";
  return out;
}

// --- Shared session stream ---

class SessionStream {
  private readonly subscribers = new Set<TerminalStreamSubscriber>();
  private proc: ReturnType<typeof Bun.spawn> | null = null;
  private pending: Uint8Array[] = [];
  private pendingBytes = 0;
  private flushTimer: ReturnType<typeof setTimeout> | null = null;
  private lineBuf = "";
  private closed = false;
  // Last size we resized the pane to / reported to viewers (tmux is the source of truth;
  // these are just for redundant-resize avoidance and debug logging).
  private termCols = DEFAULT_COLS;
  private termRows = DEFAULT_ROWS;
  // Drift corrector: re-stamp tmux's authoritative grid after live deltas settle.
  private resyncTimer: ReturnType<typeof setTimeout> | null = null;
  private resyncCapTimer: ReturnType<typeof setTimeout> | null = null;
  private resyncDirty = false;
  // Running ANSI parse state of the outbound (live) stream. A resync repaint or backfill
  // reset is only injected when this is "ground" (a sequence boundary), so it can't split
  // an escape sequence. Tracked over live flush bytes only — injected repaints are balanced.
  private broadcastState: AnsiState = "ground";
  private groundWaiters: Array<() => void> = [];
  // The viewer that owns window sizing (the interactive typist). While set, only it may
  // resize the shared tmux window; read-only watchers render at the current pane size.
  private sizingOwner: TerminalStreamSubscriber | null = null;
  // In-band command-reply correlation (#275): one FIFO entry per command written to the
  // control client's stdin; tmux answers each with a %begin…%end (or %error) block, in
  // order. The first block on attach is unsolicited (discarded via attachBlockSeen).
  private pendingCommands: PendingCommand[] = [];
  private currentBlock: { num: number; lines: string[] } | null = null;
  private attachBlockSeen = false;
  // Resync runs an async in-band capture; guard against overlapping captures and track how
  // long the ground gate has been deferring (for the CAN fallback).
  private resyncInFlight = false;
  private groundDeferStart = 0;
  // #275 dedup: deltas that were in `pending` at the resync capture's %end — the PRE-capture
  // %output. doResync drains exactly these before the repaint; %output that arrives after the
  // capture (even later in the same readLoop chunk) accumulates in a fresh `pending` and
  // applies ON TOP of the repaint on the trailing flush, never re-scrolled beneath it.
  private resyncPreCapture: Uint8Array[] | null = null;
  private resyncPreCaptureBytes = 0;

  constructor(
    private readonly session: string,
    private readonly config: OrchestratorConfig,
    private readonly onEmpty: () => void,
  ) {}

  private socket: string | undefined;

  start(): void {
    const socket = this.config.env.AGENT_RELAY_TMUX_SOCKET || tmuxSocketForSession(this.session);
    this.socket = socket;
    // Pin the window size before attaching so the control client can't reflow the
    // pane (default window-size would shrink it to the new client's 80x24). These are
    // one-time, pre-attach spawnSyncs (the control client isn't up yet) — fine to block.
    const dims = this.paneDimsSync();
    if (dims.cols) this.termCols = dims.cols;
    if (dims.rows) this.termRows = dims.rows;
    Bun.spawnSync(tmuxCommand(socket, "set-window-option", "-t", this.session, "window-size", "manual"), {
      stdin: "ignore",
      stdout: "ignore",
      stderr: "ignore",
    });
    if (dims.cols && dims.rows) {
      Bun.spawnSync(
        tmuxCommand(socket, "resize-window", "-t", this.session, "-x", String(dims.cols), "-y", String(dims.rows)),
        { stdin: "ignore", stdout: "ignore", stderr: "ignore" },
      );
    }
    try {
      this.proc = Bun.spawn(tmuxCommand(socket, "-C", "attach-session", "-t", this.session), {
        stdin: "pipe",
        stdout: "pipe",
        stderr: "ignore",
      });
    } catch (e) {
      this.fail(errMessage(e));
      return;
    }
    // #1676 — guarded: an unconsumed rejection from either exits the Bun process.
    void fireAndForget("Terminal read loop", () => this.readLoop());
    void fireAndForget("Terminal session exit watch", this.proc.exited.then(() => this.fail("terminal session ended")));
  }

  private async readLoop(): Promise<void> {
    const proc = this.proc;
    if (!proc?.stdout || typeof proc.stdout === "number") return;
    const reader = (proc.stdout as ReadableStream<Uint8Array>).getReader();
    try {
      for (;;) {
        const { done, value } = await reader.read();
        if (done) break;
        // True ISO-8859-1, NOT TextDecoder("latin1") (= windows-1252, lossy in 0x80–0x9F).
        this.lineBuf += decodeLatin1(value);
        let nl: number;
        while ((nl = this.lineBuf.indexOf("\n")) !== -1) {
          const line = this.lineBuf.slice(0, nl).replace(/\r$/, "");
          this.lineBuf = this.lineBuf.slice(nl + 1);
          this.handleLine(line);
        }
      }
    } catch {
      // reader aborted on teardown
    }
    this.fail("terminal stream closed");
  }

  private handleLine(line: string): void {
    // Inside a reply block, every line is the command's output until its OWN %end/%error
    // (matched by command number — a captured grid row could otherwise masquerade as one).
    // `%output` notifications never appear inside a block, so this can't swallow live deltas.
    if (this.currentBlock) {
      const parsed = parseControlLine(line);
      if ((parsed.type === "end" || parsed.type === "error") && parsed.num === this.currentBlock.num) {
        this.finishBlock(parsed.type === "error");
      } else {
        this.currentBlock.lines.push(line);
      }
      return;
    }
    const parsed = parseControlLine(line);
    if (parsed.type === "begin") {
      this.currentBlock = { num: parsed.num, lines: [] };
    } else if (parsed.type === "output") {
      this.enqueue(parsed.bytes);
    } else if (parsed.type === "exit") {
      this.fail(parsed.reason ? `tmux exit: ${parsed.reason}` : "tmux exit");
    }
    // A stray %end/%error with no open block, or any %other notification — ignore.
  }

  // Resolve the just-closed reply block against the FIFO head. The very first block on
  // control-mode attach is unsolicited (tmux's empty initial command) — discard it so the
  // correlation never drifts off by one.
  private finishBlock(isError: boolean): void {
    const block = this.currentBlock;
    this.currentBlock = null;
    if (!block) return;
    if (!this.attachBlockSeen) {
      this.attachBlockSeen = true;
      return;
    }
    const entry = this.pendingCommands.shift();
    if (!entry) return; // unexpected extra block — drop it rather than mis-correlate
    if (entry.timer) clearTimeout(entry.timer);
    // Snapshot the delta boundary NOW, synchronously, before resolve() and before readLoop
    // processes any further %output in this chunk — so the boundary is exactly this block's %end.
    if (entry.onFinishSync) {
      try { entry.onFinishSync(); } catch {}
    }
    if (isError) entry.reject(new Error(block.lines.join(" ").trim() || "tmux command error"));
    else entry.resolve(block.lines.map(latin1LineToUtf8));
  }

  private enqueue(bytes: Uint8Array): void {
    if (bytes.length === 0) return;
    this.pending.push(bytes);
    this.pendingBytes += bytes.length;
    if (this.pendingBytes >= FLUSH_MAX_BYTES) {
      this.flush();
      return;
    }
    if (this.flushTimer === null) {
      this.flushTimer = setTimeout(() => this.flush(), FLUSH_MS);
    }
  }

  private flush(): void {
    if (this.flushTimer !== null) {
      clearTimeout(this.flushTimer);
      this.flushTimer = null;
    }
    if (this.pendingBytes === 0) return;
    const merged = new Uint8Array(this.pendingBytes);
    let offset = 0;
    for (const chunk of this.pending) {
      merged.set(chunk, offset);
      offset += chunk.length;
    }
    this.pending = [];
    this.pendingBytes = 0;
    // Track ANSI ground state over LIVE bytes only (injected repaints are balanced by
    // construction). The resync gate and whenAtGround read this to know it's safe to inject.
    this.broadcastState = scanAnsiState(merged, this.broadcastState);
    this.broadcast(merged);
    if (this.broadcastState === "ground") this.fireGroundWaiters();
    // Live deltas just went out; schedule an authoritative resync to correct any drift.
    this.scheduleResync();
  }

  private broadcast(bytes: Uint8Array): void {
    for (const sub of [...this.subscribers]) {
      if (sub.bufferedAmount && sub.bufferedAmount() > BACKPRESSURE_MAX_BYTES) {
        this.removeSubscriber(sub);
        try {
          sub.onClose("backpressure");
        } catch {}
        continue;
      }
      try {
        sub.onData(bytes);
      } catch {}
    }
  }

  // Run `cb` at the next ANSI sequence boundary on the live stream (or now, if already at
  // one), so an injected reset/backfill can't splice between the halves of a live escape
  // sequence (#276). Falls back to firing anyway after GROUND_WAIT_MAX_MS if the stream
  // stalls mid-sequence (rare — a dead pane); the reset path does a full clear regardless.
  whenAtGround(cb: () => void): void {
    if (this.closed || this.broadcastState === "ground") {
      cb();
      return;
    }
    let fired = false;
    const fire = () => {
      if (fired) return;
      fired = true;
      clearTimeout(timer);
      const i = this.groundWaiters.indexOf(waiter);
      if (i !== -1) this.groundWaiters.splice(i, 1);
      try {
        cb();
      } catch {}
    };
    const waiter = fire;
    const timer = setTimeout(fire, GROUND_WAIT_MAX_MS);
    this.groundWaiters.push(waiter);
  }

  private fireGroundWaiters(): void {
    if (this.groundWaiters.length === 0) return;
    const waiters = this.groundWaiters;
    this.groundWaiters = [];
    for (const w of waiters) {
      try {
        w();
      } catch {}
    }
  }

  // Schedule a drift-correcting resync: a trailing debounce fires once output settles,
  // and a capped interval guarantees correction during continuous output.
  private scheduleResync(): void {
    if (RESYNC_DEBOUNCE_MS <= 0 || this.subscribers.size === 0) return;
    this.resyncDirty = true;
    if (this.resyncTimer) clearTimeout(this.resyncTimer);
    this.resyncTimer = guardedTimeout("Terminal resync", RESYNC_DEBOUNCE_MS, () => this.doResync());
    if (!this.resyncCapTimer && RESYNC_MAX_INTERVAL_MS > 0) {
      this.resyncCapTimer = guardedTimeout("Terminal resync", RESYNC_MAX_INTERVAL_MS, () => this.doResync());
    }
  }

  private async doResync(): Promise<void> {
    this.clearResyncTimers();
    if (!this.resyncDirty || this.closed || this.subscribers.size === 0) return;
    // One in-band capture at a time; the next flush re-arms us.
    if (this.resyncInFlight) {
      this.resyncTimer = guardedTimeout("Terminal resync", RESYNC_GROUND_RETRY_MS, () => this.doResync());
      return;
    }
    // Ground-state gate (#276): never splice a repaint between the two halves of a split
    // escape sequence (the injected ESC aborts the half-parsed CSI and its tail prints as
    // literal text — the stray "S"). If we're mid-sequence, keep the work pending and
    // re-check shortly. If the stream stays mid-sequence past the hard cap (stalled pane),
    // force the repaint with a CAN prefix to abort the client's half-sequence.
    let forceAbort = false;
    if (this.broadcastState !== "ground") {
      const now = Date.now();
      if (this.groundDeferStart === 0) this.groundDeferStart = now;
      if (now - this.groundDeferStart < RESYNC_GROUND_DEFER_MAX_MS) {
        this.resyncTimer = guardedTimeout("Terminal resync", RESYNC_GROUND_RETRY_MS, () => this.doResync());
        return;
      }
      forceAbort = true;
    }
    this.groundDeferStart = 0;
    this.resyncDirty = false;
    this.resyncInFlight = true;
    try {
      const repaint = await this.resyncRepaint();
      // Ordering (#275): the %output snapshotted at the capture's %end (resyncPreCapture) is
      // the PRE-capture stream — drain it to subscribers BEFORE the repaint so scrolled lines
      // can't re-apply on top of it (the duplicate-text race). %output that arrived AFTER the
      // capture stayed in `pending` and applies on top of the repaint via the trailing flush.
      this.drainPreCaptureDeltas();
      if (!forceAbort && this.broadcastState !== "ground") {
        this.resyncDirty = true;
        this.flush(); // let the sequence tail complete normally; retry with a fresh capture
        if (!this.resyncTimer) this.resyncTimer = guardedTimeout("Terminal resync", RESYNC_GROUND_RETRY_MS, () => this.doResync());
        return;
      }
      if (!repaint || this.closed || this.subscribers.size === 0) {
        this.flush(); // release any post-capture deltas normally (no repaint to inject)
        return;
      }
      const out = forceAbort ? this.prependCan(repaint) : repaint;
      this.broadcast(out);
      this.flush(); // post-capture deltas, on top of the fresh repaint
      tdbg(`resync ${this.session} bytes=${out.length} viewers=${this.subscribers.size}${forceAbort ? " (forced)" : ""}`);
    } catch (e) {
      this.drainPreCaptureDeltas();
      this.flush();
      tdbg(`resync ${this.session} failed: ${errMessage(e)}`);
    } finally {
      this.resyncInFlight = false;
    }
  }

  private prependCan(bytes: Uint8Array): Uint8Array {
    const out = new Uint8Array(bytes.length + 1);
    out[0] = CAN_BYTE;
    out.set(bytes, 1);
    return out;
  }

  private clearResyncTimers(): void {
    if (this.resyncTimer) {
      clearTimeout(this.resyncTimer);
      this.resyncTimer = null;
    }
    if (this.resyncCapTimer) {
      clearTimeout(this.resyncCapTimer);
      this.resyncCapTimer = null;
    }
  }

  // Build an in-place authoritative repaint from tmux's grid: absolute-position each row,
  // reset SGR + clear it, then write tmux's styled cells. This overwrites any drift (a
  // doubled statusline, a solid-instead-of-faded suggestion, stale cells) without a
  // full-screen clear flash, and re-parks the cursor at tmux's true position/visibility.
  // The grid + cursor are read in-band through the control client (#275), so the capture
  // is serialized with %output deltas and costs no process spawn.
  private async resyncRepaint(): Promise<Uint8Array | null> {
    // ORDER IS LOAD-BEARING for the #275 dedup. The screen capture is the serialization
    // point: doResync drains `pending` (pre-capture %output) and emits this repaint right
    // after. So the capture MUST be the LAST in-band read here — any awaited round-trip
    // after it (the old readCursorState/paneDims) lets live %output land in `pending` that
    // doResync then flushes BEFORE this now-stale repaint, re-scrolling lines the repaint
    // still shows on-screen into the client's scrollback → duplicated text under load.
    // Read the cursor first; reuse the cached row count (resize drives its own backfill, so
    // termRows is authoritative without a round-trip); capture the grid last.
    const cursor = await this.readCursorState();
    // Capture LAST, and tag it so its %end snapshots the pre-capture delta boundary (the
    // capture-pane reply IS the serialization point against the %output stream).
    const lines = await this.command(
      `capture-pane -p -e -t "${this.session}"`,
      true,
      () => this.snapshotPreCaptureDeltas(),
    ).catch(() => null);
    const body = lines ? lines.join("\n") : "";
    if (!body) return null;
    return new TextEncoder().encode(buildInPlaceRepaint(body, this.termRows, cursor));
  }

  // Swap the live `pending` buffer aside at the resync capture's %end: these chunks are the
  // PRE-capture %output (drained before the repaint), and a fresh `pending` collects anything
  // after (applied on top, next flush). Runs synchronously inside finishBlock (#275).
  private snapshotPreCaptureDeltas(): void {
    this.resyncPreCapture = this.pending;
    this.resyncPreCaptureBytes = this.pendingBytes;
    this.pending = [];
    this.pendingBytes = 0;
    if (this.flushTimer !== null) {
      clearTimeout(this.flushTimer);
      this.flushTimer = null;
    }
  }

  // Broadcast the snapshotted pre-capture deltas (mirrors flush()'s ANSI-state bookkeeping)
  // ahead of the repaint. No-op when nothing was snapshotted (capture timed out / errored).
  private drainPreCaptureDeltas(): void {
    const chunks = this.resyncPreCapture;
    const total = this.resyncPreCaptureBytes;
    this.resyncPreCapture = null;
    this.resyncPreCaptureBytes = 0;
    if (!chunks || total === 0) return;
    const merged = new Uint8Array(total);
    let offset = 0;
    for (const chunk of chunks) {
      merged.set(chunk, offset);
      offset += chunk.length;
    }
    this.broadcastState = scanAnsiState(merged, this.broadcastState);
    this.broadcast(merged);
    if (this.broadcastState === "ground") this.fireGroundWaiters();
  }

  // Read tmux's authoritative grid (styled) plus a consistent cursor, and turn it into a
  // client repaint. capture-pane reads tmux's real emulator grid, so it's internally
  // coherent; we bracket content/cursor/content (in-band, cheap now) to guard the read
  // against a mid-render change.
  private async captureScreen(): Promise<{ content: string; cursorX?: number; cursorY?: number }> {
    let content = await this.readBackfill();
    let cursor = await this.readCursor();
    for (let attempt = 0; attempt < 4; attempt++) {
      const recheck = await this.readBackfill();
      if (recheck === content) break;
      content = recheck;
      cursor = await this.readCursor();
    }
    return { content: buildScreenRepaint(content, cursor.cursorX, cursor.cursorY), ...cursor };
  }

  // Backfill capture: current screen plus scrollback history above it (cursor is
  // viewport-relative, so it still parks on the live screen). Falls back to screen-only.
  private async readBackfill(): Promise<string> {
    if (BACKFILL_SCROLLBACK_LINES <= 0) return this.readScreen();
    const lines = await this.runCommand(
      `capture-pane -p -e -S -${BACKFILL_SCROLLBACK_LINES} -t "${this.session}"`,
    ).catch(() => null);
    return lines ? lines.join("\n") : this.readScreen();
  }

  // Current-screen-only capture (no scrollback) — used by the live resync corrector so it
  // overwrites just the visible grid and leaves the client's scroll-back history intact.
  private async readScreen(): Promise<string> {
    const lines = await this.runCommand(`capture-pane -p -e -t "${this.session}"`).catch(() => null);
    return lines ? lines.join("\n") : "";
  }

  private async readCursor(): Promise<{ cursorX?: number; cursorY?: number }> {
    const lines = await this.runCommand(
      `display-message -p -t "${this.session}" "#{cursor_x} #{cursor_y}"`,
    ).catch(() => [] as string[]);
    const [x, y] = (lines[0] ?? "").trim().split(/\s+/).map(Number);
    return {
      ...(Number.isFinite(x) ? { cursorX: x } : {}),
      ...(Number.isFinite(y) ? { cursorY: y } : {}),
    };
  }

  // Cursor position plus visibility (cursor_flag), so a resync repaint restores whether
  // the TUI had the cursor shown or hidden rather than guessing.
  private async readCursorState(): Promise<{ cursorX?: number; cursorY?: number; visible?: boolean }> {
    const lines = await this.runCommand(
      `display-message -p -t "${this.session}" "#{cursor_x} #{cursor_y} #{cursor_flag}"`,
    ).catch(() => [] as string[]);
    const [x, y, flag] = (lines[0] ?? "").trim().split(/\s+/);
    const cx = Number(x);
    const cy = Number(y);
    return {
      ...(Number.isFinite(cx) ? { cursorX: cx } : {}),
      ...(Number.isFinite(cy) ? { cursorY: cy } : {}),
      ...(flag === "0" || flag === "1" ? { visible: flag === "1" } : {}),
    };
  }

  // Repaint a freshly-attached (or resumed/refreshed) viewer from tmux's current grid.
  // When the viewer's dimensions are given we resize the tmux pane first so the TUI
  // re-renders at the viewer's width, let it settle, then capture the post-resize frame.
  async backfill(sub: TerminalStreamSubscriber, cols?: number, rows?: number): Promise<TerminalSnapshot> {
    let resized = false;
    if (this.maySize(sub) && Number.isFinite(cols) && Number.isFinite(rows) && (cols as number) >= 10 && (rows as number) >= 5) {
      const c = Math.trunc(cols as number);
      const r = Math.trunc(rows as number);
      if (c !== this.termCols || r !== this.termRows) {
        void fireAndForget("Terminal resize", this.command(`resize-window -t "${this.session}" -x ${c} -y ${r}`));
        this.termCols = c;
        this.termRows = r;
        resized = true;
      }
    }
    if (resized && RESIZE_SETTLE_MS > 0) await Bun.sleep(RESIZE_SETTLE_MS);
    const { content } = await this.captureScreen();
    // Report tmux's actual pane size (authoritative) back to the viewer.
    const dims = await this.paneDims();
    if (dims.cols) this.termCols = dims.cols;
    if (dims.rows) this.termRows = dims.rows;
    const live = sessionLiveness(this.session, this.config.env.AGENT_RELAY_TMUX_SOCKET || tmuxSocketForSession(this.session));
    tdbg(`backfill ${this.session} req=${cols}x${rows} term=${this.termCols}x${this.termRows} contentLen=${content.length} viewers=${this.subscribers.size}`);
    return {
      session: this.session,
      content,
      running: live.running,
      agentAlive: live.agentAlive,
      cols: this.termCols,
      rows: this.termRows,
      capturedAt: Date.now(),
    };
  }

  write(sub: TerminalStreamSubscriber, bytes: Uint8Array): void {
    if (this.closed || !this.proc || bytes.length === 0) return;
    // Typing is the strongest interactivity signal — the typist owns window sizing.
    this.sizingOwner = sub;
    void fireAndForget("Terminal send-keys", this.command(`send-keys -t "${this.session}" -H ${encodeSendKeysHex(bytes)}`));
  }

  setInteractive(sub: TerminalStreamSubscriber, interactive: boolean): void {
    if (interactive) {
      this.sizingOwner = sub;
    } else if (this.sizingOwner === sub) {
      this.sizingOwner = null;
    }
  }

  // A viewer may size the shared window only if it's the sizing owner, or nobody owns
  // sizing yet (first viewer, before anyone has declared interactivity — it still needs
  // a sensible size). Once an interactive viewer exists, read-only watchers never resize.
  private maySize(sub: TerminalStreamSubscriber): boolean {
    return this.sizingOwner === null || this.sizingOwner === sub;
  }

  resize(sub: TerminalStreamSubscriber, cols: number, rows: number): void {
    if (this.closed || !this.proc) return;
    if (!this.maySize(sub)) return; // read-only watcher: don't reflow the shared window
    if (!Number.isFinite(cols) || !Number.isFinite(rows) || cols < 10 || rows < 5) return;
    const c = Math.trunc(cols);
    const r = Math.trunc(rows);
    void fireAndForget("Terminal resize", this.command(`resize-window -t "${this.session}" -x ${c} -y ${r}`));
    this.termCols = c;
    this.termRows = r;
    tdbg(`resize ${this.session} -> ${c}x${r} viewers=${this.subscribers.size}`);
  }

  // Pane dimensions, in-band through the control client (no spawn).
  private async paneDims(): Promise<{ cols?: number; rows?: number }> {
    const lines = await this.runCommand(
      `display-message -p -t "${this.session}" "#{pane_width} #{pane_height}"`,
    ).catch(() => [] as string[]);
    return parsePaneDims(lines[0] ?? "");
  }

  // Synchronous pane-dims read for start(), which runs before the control client attaches.
  private paneDimsSync(): { cols?: number; rows?: number } {
    try {
      const out = Bun.spawnSync(
        tmuxCommand(this.socket, "display-message", "-p", "-t", this.session, "#{pane_width} #{pane_height}"),
        { stdin: "ignore", stdout: "pipe", stderr: "ignore" },
      ).stdout.toString().trim();
      return parsePaneDims(out);
    } catch {
      return {};
    }
  }

  // Run a control command expecting its reply lines (the FIFO correlates them in order).
  private runCommand(line: string): Promise<string[]> {
    return this.command(line, true);
  }

  // Write a command to the control client's stdin and register a FIFO entry for its reply
  // block. `wantReply` commands resolve with the block's lines (or reject on %error /
  // timeout); fire-and-forget commands resolve with [] once their empty block closes.
  private command(line: string, wantReply = false, onFinishSync?: () => void): Promise<string[]> {
    if (this.closed) return wantReply ? Promise.reject(new Error("terminal stream closed")) : Promise.resolve([]);
    const stdin = this.proc?.stdin;
    if (!stdin || typeof stdin === "number") {
      return wantReply ? Promise.reject(new Error("control client unavailable")) : Promise.resolve([]);
    }
    return new Promise<string[]>((resolve, reject) => {
      const entry: PendingCommand = { wantReply, lines: [], resolve, reject, timer: null, onFinishSync };
      if (wantReply) entry.timer = setTimeout(() => this.commandTimeout(entry), COMMAND_TIMEOUT_MS);
      this.pendingCommands.push(entry);
      try {
        (stdin as { write(data: string): void; flush?(): void }).write(`${line}\n`);
        (stdin as { flush?(): void }).flush?.();
      } catch (e) {
        const idx = this.pendingCommands.indexOf(entry);
        if (idx !== -1) this.pendingCommands.splice(idx, 1);
        if (entry.timer) clearTimeout(entry.timer);
        if (wantReply) reject(e instanceof Error ? e : new Error(String(e)));
        else resolve([]);
      }
    });
  }

  // A reply that never arrived means a protocol desync — reject the awaited command and
  // reset the whole reply queue + block state so the next command starts clean. The byte
  // stream itself keeps flowing; only in-flight captures fail (the resync just skips a beat).
  private commandTimeout(entry: PendingCommand): void {
    if (!this.pendingCommands.includes(entry)) return;
    tdbg(`command timeout ${this.session}; resetting reply queue (${this.pendingCommands.length} pending)`);
    const pend = this.pendingCommands;
    this.pendingCommands = [];
    this.currentBlock = null;
    for (const e of pend) {
      if (e.timer) clearTimeout(e.timer);
      if (e.wantReply) e.reject(new Error("control command timed out"));
      else e.resolve([]);
    }
  }

  addSubscriber(sub: TerminalStreamSubscriber): void {
    this.subscribers.add(sub);
    tdbg(`attach ${this.session} viewers=${this.subscribers.size} term=${this.termCols}x${this.termRows}`);
  }

  removeSubscriber(sub: TerminalStreamSubscriber): void {
    if (!this.subscribers.delete(sub)) return;
    if (this.sizingOwner === sub) this.sizingOwner = null;
    tdbg(`detach ${this.session} viewers=${this.subscribers.size}`);
    if (this.subscribers.size === 0) this.destroy();
  }

  private fail(reason: string): void {
    if (this.closed) return;
    const subs = [...this.subscribers];
    this.subscribers.clear();
    this.destroy();
    for (const sub of subs) {
      try {
        sub.onClose(reason);
      } catch {}
    }
  }

  private destroy(): void {
    if (this.closed) return;
    this.closed = true;
    if (this.flushTimer !== null) {
      clearTimeout(this.flushTimer);
      this.flushTimer = null;
    }
    this.clearResyncTimers();
    // Reject any in-flight in-band commands so awaiters (captures) unwind instead of hanging.
    const pend = this.pendingCommands;
    this.pendingCommands = [];
    this.currentBlock = null;
    for (const e of pend) {
      if (e.timer) clearTimeout(e.timer);
      if (e.wantReply) e.reject(new Error("terminal stream closed"));
      else e.resolve([]);
    }
    this.fireGroundWaiters();
    try {
      this.proc?.kill();
    } catch {}
    this.proc = null;
    this.onEmpty();
  }
}

const streams = new Map<string, SessionStream>();

export function acquireTerminalStream(
  session: string,
  config: OrchestratorConfig,
  subscriber: TerminalStreamSubscriber,
): TerminalStreamHandle {
  let stream = streams.get(session);
  if (!stream) {
    stream = new SessionStream(session, config, () => {
      if (streams.get(session) === stream) streams.delete(session);
    });
    streams.set(session, stream);
    stream.start();
  }
  const active = stream;
  active.addSubscriber(subscriber);
  return {
    backfill: (cols, rows) => active.backfill(subscriber, cols, rows),
    write: (bytes) => active.write(subscriber, bytes),
    resize: (cols, rows) => active.resize(subscriber, cols, rows),
    setInteractive: (interactive) => active.setInteractive(subscriber, interactive),
    whenAtGround: (cb) => active.whenAtGround(cb),
    release: () => active.removeSubscriber(subscriber),
  };
}

export function activeTerminalStreamCount(): number {
  return streams.size;
}