/**
 * Multiplexed WebSocket server for terminal multiplexing.
 * Manages multiple terminal connections over a single persistent WebSocket.
 *
 * Session updates are delivered via polling of Next.js /api/sessions/patches
 * every 3s, then broadcast to all subscribed clients via WebSocket.
 */

import { WebSocketServer, WebSocket } from "ws";
import { spawn } from "node:child_process";
import * as fs from "node:fs";
import { createRequire } from "node:module";
import { type Socket, connect as netConnect } from "node:net";
import { dirname, join } from "node:path";
import {
  DEFAULT_DASHBOARD_NOTIFICATION_LIMIT,
  getEnvDefaults,
  getDashboardNotificationStorePath,
  getNodePtyPrebuildsSubdir,
  isWindows,
  loadConfig,
  normalizeDashboardNotificationLimit,
  recordActivityEvent,
  readDashboardNotificationsFromFile,
  type DashboardNotificationRecord,
} from "@aoagents/ao-core";
import {
  findTmux,
  resolveTmuxSession,
  resolvePipePath,
  tmuxHasSession,
  validateSessionId,
} from "./tmux-utils.js";

// These types mirror src/lib/mux-protocol.ts exactly.
// tsconfig.server.json constrains rootDir to "server/", so we cannot import
// across the boundary. Keep both in sync when updating the protocol.

// ── Client → Server ──
type ClientMessage =
  | { ch: "terminal"; id: string; type: "data"; data: string; projectId?: string }
  | { ch: "terminal"; id: string; type: "resize"; cols: number; rows: number; projectId?: string }
  | { ch: "terminal"; id: string; type: "open"; projectId?: string; tmuxName?: string }
  | { ch: "terminal"; id: string; type: "close"; projectId?: string }
  | { ch: "system"; type: "ping" }
  | { ch: "subscribe"; topics: Array<"sessions" | "notifications"> };

// ── Server → Client ──
type ServerMessage =
  | { ch: "terminal"; id: string; type: "data"; data: string; projectId?: string }
  | { ch: "terminal"; id: string; type: "exited"; code: number; projectId?: string }
  | { ch: "terminal"; id: string; type: "opened"; projectId?: string }
  | { ch: "terminal"; id: string; type: "error"; message: string; projectId?: string }
  | { ch: "sessions"; type: "snapshot"; sessions: SessionPatch[] }
  | { ch: "sessions"; type: "error"; error: string }
  | {
      ch: "notifications";
      type: "snapshot" | "append";
      notifications: DashboardNotificationRecord[];
      limit: number;
    }
  | { ch: "notifications"; type: "error"; error: string }
  | { ch: "system"; type: "pong" }
  | { ch: "system"; type: "error"; message: string };

// Mirrors AttentionLevel in src/lib/types.ts — keep in sync.
type AttentionLevel = "merge" | "action" | "respond" | "review" | "pending" | "working" | "done";

interface SessionPatch {
  id: string;
  status: string;
  activity: string | null;
  attentionLevel: AttentionLevel;
  lastActivityAt: string;
}

/**
 * Manages polling of session patches from Next.js /api/sessions/patches.
 * Broadcasts to all subscribed callbacks.
 * Lazily starts polling on first subscriber, stops when the last one leaves.
 */
export class SessionBroadcaster {
  private subscribers = new Set<(sessions: SessionPatch[]) => void>();
  private errorSubscribers = new Set<(error: string) => void>();
  private intervalId: ReturnType<typeof setInterval> | null = null;
  private polling = false;
  // Tracks the last fetch outcome so we only emit ui.session_broadcast_failed on
  // the healthy → failing transition (not every 3s during an outage).
  private lastFetchOk = true;
  private readonly baseUrl: string;
  private readonly authHeader: string | null;

  constructor(nextPort: string) {
    this.baseUrl = `http://localhost:${nextPort}`;
    const password = process.env.AO_REMOTE_AUTH_PASSWORD;
    this.authHeader = password
      ? `Basic ${Buffer.from(`${process.env.AO_REMOTE_AUTH_USER || "ao"}:${password}`).toString("base64")}`
      : null;
  }

  /**
   * Subscribe to session patches and errors. Returns an unsubscribe function.
   * Sends an immediate snapshot to the new subscriber, then polling updates.
   */
  subscribe(
    callback: (sessions: SessionPatch[]) => void,
    onError?: (error: string) => void,
  ): () => void {
    const wasEmpty = this.subscribers.size === 0;
    this.subscribers.add(callback);
    if (onError) this.errorSubscribers.add(onError);

    // Immediately send a one-off snapshot to just this new subscriber
    void this.fetchSnapshot().then((result) => {
      if (result.sessions && this.subscribers.has(callback)) {
        try {
          callback(result.sessions);
        } catch {
          // Isolate subscriber errors so one bad subscriber doesn't break others
        }
      } else if (result.error && onError && this.errorSubscribers.has(onError)) {
        try {
          onError(result.error);
        } catch {
          // Isolate subscriber errors
        }
      }
    });

    // Start polling if this is the first subscriber
    if (wasEmpty) {
      this.intervalId = setInterval(() => {
        if (this.polling) return;
        this.polling = true;
        void this.fetchSnapshot()
          .then((result) => {
            if (result.sessions && this.intervalId !== null) this.broadcast(result.sessions);
            else if (result.error && this.intervalId !== null) this.broadcastError(result.error);
          })
          .finally(() => {
            this.polling = false;
          });
      }, 3000);
    }

    return () => {
      this.subscribers.delete(callback);
      if (onError) this.errorSubscribers.delete(onError);
      if (this.subscribers.size === 0) {
        this.disconnect();
      }
    };
  }

  private broadcast(sessions: SessionPatch[]): void {
    for (const callback of this.subscribers) {
      try {
        callback(sessions);
      } catch (err) {
        console.error("[MuxServer] Session broadcast subscriber threw:", err);
      }
    }
  }

  private broadcastError(error: string): void {
    for (const callback of this.errorSubscribers) {
      try {
        callback(error);
      } catch (err) {
        console.error("[MuxServer] Session error subscriber threw:", err);
      }
    }
  }

  /** One-shot HTTP fetch of the current session list. */
  private async fetchSnapshot(): Promise<{
    sessions: SessionPatch[] | null;
    error: string | null;
  }> {
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), 4000);
    try {
      const res = await fetch(`${this.baseUrl}/api/sessions/patches`, {
        signal: controller.signal,
        ...(this.authHeader ? { headers: { Authorization: this.authHeader } } : {}),
      });
      clearTimeout(timeoutId);
      if (!res.ok) {
        const msg = `Session fetch failed: HTTP ${res.status}`;
        console.warn(`[SessionBroadcaster] ${msg}`);
        this.recordFetchFailure(msg, { httpStatus: res.status });
        return { sessions: null, error: msg };
      }
      const data = (await res.json()) as { sessions?: SessionPatch[] };
      this.lastFetchOk = true;
      return { sessions: data.sessions ?? null, error: null };
    } catch (err) {
      clearTimeout(timeoutId);
      const msg = err instanceof Error ? err.message : String(err);
      console.warn("[SessionBroadcaster] fetchSnapshot error:", msg);
      this.recordFetchFailure(msg);
      return { sessions: null, error: msg };
    }
  }

  /**
   * Emit ui.session_broadcast_failed once per healthy→failing transition.
   * The broadcaster polls every 3s; emitting on every failure during a long
   * outage would flood the events table (~20/min). Recovery resets the flag.
   */
  private recordFetchFailure(message: string, extra?: Record<string, unknown>): void {
    if (!this.lastFetchOk) return;
    this.lastFetchOk = false;
    recordActivityEvent({
      source: "ui",
      kind: "ui.session_broadcast_failed",
      level: "warn",
      summary: `session broadcaster fetch failed: ${message}`,
      data: {
        url: `${this.baseUrl}/api/sessions/patches`,
        errorMessage: message,
        ...extra,
      },
    });
  }

  private disconnect(): void {
    if (this.intervalId !== null) {
      clearInterval(this.intervalId);
      this.intervalId = null;
    }
  }
}

function notificationKey(record: DashboardNotificationRecord): string {
  return `${record.id}:${record.receivedAt}`;
}

function readDashboardLimit(configPath: string | undefined): number {
  if (!configPath) return DEFAULT_DASHBOARD_NOTIFICATION_LIMIT;
  try {
    const config = loadConfig(configPath);
    const dashboardConfig = config.notifiers?.["dashboard"];
    return normalizeDashboardNotificationLimit(dashboardConfig?.["limit"]);
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    console.warn("[NotificationBroadcaster] Could not read dashboard notifier limit:", message);
    return DEFAULT_DASHBOARD_NOTIFICATION_LIMIT;
  }
}

/**
 * Polls the dashboard notification JSONL store and broadcasts changes to mux
 * subscribers. The store is config-scoped and survives dashboard reloads.
 */
export class NotificationBroadcaster {
  private subscribers = new Set<
    (
      notifications: DashboardNotificationRecord[],
      type: "snapshot" | "append",
      limit: number,
    ) => void
  >();
  private errorSubscribers = new Set<(error: string) => void>();
  private intervalId: ReturnType<typeof setInterval> | null = null;
  private lastRecords: DashboardNotificationRecord[] = [];
  private readonly configPath: string | undefined;
  private readonly storePath: string | null;

  constructor(configPath = process.env["AO_CONFIG_PATH"]) {
    this.configPath = configPath;
    this.storePath = configPath ? getDashboardNotificationStorePath(configPath) : null;
  }

  subscribe(
    callback: (
      notifications: DashboardNotificationRecord[],
      type: "snapshot" | "append",
      limit: number,
    ) => void,
    onError?: (error: string) => void,
  ): () => void {
    const wasEmpty = this.subscribers.size === 0;
    this.subscribers.add(callback);
    if (onError) this.errorSubscribers.add(onError);

    const snapshot = this.fetchSnapshot();
    if (wasEmpty) {
      this.lastRecords = snapshot.notifications;
    }
    try {
      callback(snapshot.notifications, "snapshot", snapshot.limit);
    } catch {
      // Isolate subscriber errors so one bad socket does not break others.
    }

    if (snapshot.error && onError) {
      try {
        onError(snapshot.error);
      } catch {
        // Isolate subscriber errors.
      }
    }

    if (wasEmpty) {
      this.intervalId = setInterval(() => {
        const result = this.fetchSnapshot();
        if (result.error) {
          this.broadcastError(result.error);
          return;
        }

        const previousKeys = new Set(this.lastRecords.map(notificationKey));
        const appended = result.notifications.filter(
          (record) => !previousKeys.has(notificationKey(record)),
        );
        const trimmed = result.notifications.length < this.lastRecords.length;
        this.lastRecords = result.notifications;

        if (appended.length > 0 && !trimmed) {
          this.broadcast(appended, "append", result.limit);
        } else if (appended.length > 0 || trimmed) {
          this.broadcast(result.notifications, "snapshot", result.limit);
        }
      }, 1000);
    }

    return () => {
      this.subscribers.delete(callback);
      if (onError) this.errorSubscribers.delete(onError);
      if (this.subscribers.size === 0) {
        this.disconnect();
      }
    };
  }

  private fetchSnapshot(): {
    notifications: DashboardNotificationRecord[];
    error: string | null;
    limit: number;
  } {
    const limit = readDashboardLimit(this.configPath);
    if (!this.storePath) return { notifications: [], error: null, limit };

    try {
      return {
        notifications: readDashboardNotificationsFromFile(this.storePath, limit),
        error: null,
        limit,
      };
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      console.warn("[NotificationBroadcaster] fetchSnapshot error:", message);
      return { notifications: [], error: message, limit };
    }
  }

  private broadcast(
    notifications: DashboardNotificationRecord[],
    type: "snapshot" | "append",
    limit: number,
  ): void {
    for (const callback of this.subscribers) {
      try {
        callback(notifications, type, limit);
      } catch (err) {
        console.error("[MuxServer] Notification broadcast subscriber threw:", err);
      }
    }
  }

  private broadcastError(error: string): void {
    for (const callback of this.errorSubscribers) {
      try {
        callback(error);
      } catch (err) {
        console.error("[MuxServer] Notification error subscriber threw:", err);
      }
    }
  }

  private disconnect(): void {
    if (this.intervalId !== null) {
      clearInterval(this.intervalId);
      this.intervalId = null;
    }
  }
}

// node-pty is an optionalDependency — load dynamically
/* eslint-disable @typescript-eslint/consistent-type-imports -- node-pty is optional; static import would crash if missing */
type IPty = import("node-pty").IPty;
type PtySpawn = typeof import("node-pty").spawn;
type PtySpawnOptions = Parameters<PtySpawn>[2];
let ptySpawn: PtySpawn | undefined;
/* eslint-enable @typescript-eslint/consistent-type-imports */
const nodePtyRequire = createRequire(import.meta.url);

export function resolveNodePtySpawnHelperPath(): string | null {
  const override = process.env.AO_NODE_PTY_SPAWN_HELPER_PATH;
  if (override) return override;

  try {
    const packageJsonPath = nodePtyRequire.resolve("node-pty/package.json");
    return join(dirname(packageJsonPath), "prebuilds", getNodePtyPrebuildsSubdir(), "spawn-helper");
  } catch {
    return null;
  }
}

function isPosixSpawnpFailure(err: unknown): boolean {
  return err instanceof Error && err.message.includes("posix_spawnp");
}

function repairNodePtySpawnHelper(): string | null {
  if (isWindows()) return null;

  const spawnHelperPath = resolveNodePtySpawnHelperPath();
  if (!spawnHelperPath || !fs.existsSync(spawnHelperPath)) return null;

  fs.chmodSync(spawnHelperPath, 0o755);
  return spawnHelperPath;
}

try {
  const nodePty = await import("node-pty");
  ptySpawn = nodePty.spawn;
} catch (err) {
  console.warn("[MuxServer] node-pty not available — mux server will be disabled.", err);
}

interface ManagedTerminal {
  id: string;
  tmuxSessionId: string;
  pty: IPty | null;
  subscribers: Set<(data: string) => void>;
  exitCallbacks: Set<(exitCode: number) => void>;
  buffer: string[];
  bufferBytes: number;
  reattachAttempts: number;
  ptyLostEmitted: boolean;
  /**
   * Pending grace-period timer that resets reattachAttempts when the
   * currently-attached PTY survives REATTACH_RESET_GRACE_MS. Tracked so
   * cleanup paths (last-subscriber unsubscribe, subsequent re-attach) can
   * clear it and avoid keeping the dead PTY/terminal closure references
   * reachable for up to 5 s after teardown.
   */
  resetTimer?: ReturnType<typeof setTimeout>;
}

const RING_BUFFER_MAX = 50 * 1024; // 50KB max per terminal
const WS_BUFFER_HIGH_WATERMARK = 64 * 1024; // 64KB
const MAX_REATTACH_ATTEMPTS = 3;
/**
 * Grace period a freshly-attached PTY must survive before its successful
 * attach is allowed to reset the re-attach counter. Prevents tight crash
 * loops (e.g. attaching to a tmux session that no longer exists) from
 * gaming the MAX_REATTACH_ATTEMPTS cap by resetting the counter to 0
 * between every failed attempt.
 *
 * 5 s is comfortably longer than the ~40 ms a doomed `tmux attach-session`
 * takes to exit, while still being short enough that a healthy PTY which
 * crashes hours later gets a fresh retry budget.
 */
const REATTACH_RESET_GRACE_MS = 5_000;

/**
 * TerminalManager manages PTY processes independently of WebSocket connections.
 * A single manager instance is shared across all mux connections.
 */
export class TerminalManager {
  private terminals = new Map<string, ManagedTerminal>();
  private TMUX: string;
  private spawnHelperRepairAttempted = false;

  constructor(tmuxPath?: string) {
    const resolved = tmuxPath ?? findTmux();
    if (!resolved) {
      throw new Error("tmux not available on this platform");
    }
    this.TMUX = resolved;
  }

  private terminalKey(id: string, projectId?: string): string {
    return projectId ? `${projectId}:${id}` : id;
  }

  private spawnTmuxPty(args: string[], options: PtySpawnOptions): IPty {
    if (!ptySpawn) {
      throw new Error("node-pty not available");
    }

    try {
      return ptySpawn(this.TMUX, args, options);
    } catch (err) {
      if (this.spawnHelperRepairAttempted || !isPosixSpawnpFailure(err)) {
        throw err;
      }

      this.spawnHelperRepairAttempted = true;
      try {
        const repairedPath = repairNodePtySpawnHelper();
        if (repairedPath) {
          console.warn(
            `[MuxServer] node-pty posix_spawnp failed; set executable bit on ${repairedPath} and retrying once.`,
          );
        } else {
          console.warn(
            "[MuxServer] node-pty posix_spawnp failed; spawn-helper was not found, retrying once.",
          );
        }
      } catch (repairErr) {
        const message = repairErr instanceof Error ? repairErr.message : String(repairErr);
        console.warn(
          `[MuxServer] node-pty posix_spawnp failed; chmod spawn-helper failed (${message}), retrying once.`,
        );
      }

      return ptySpawn(this.TMUX, args, options);
    }
  }

  /**
   * Open/attach to a terminal. If already open, just return.
   * If has subscribers but PTY crashed, re-attach.
   */
  open(id: string, projectId?: string, tmuxName?: string): string {
    if (!validateSessionId(id)) {
      throw new Error(`Invalid session ID: ${id}`);
    }

    const key = this.terminalKey(id, projectId);
    const existing = this.terminals.get(key);
    const tmuxSessionId =
      tmuxName ??
      existing?.tmuxSessionId ??
      resolveTmuxSession(id, this.TMUX, undefined, undefined, projectId);
    if (!tmuxSessionId) {
      throw new Error(`Session not found: ${id}`);
    }

    // Get or create terminal entry
    let terminal = this.terminals.get(key);
    if (!terminal) {
      terminal = {
        id,
        tmuxSessionId,
        pty: null,
        subscribers: new Set(),
        exitCallbacks: new Set(),
        buffer: [],
        bufferBytes: 0,
        reattachAttempts: 0,
        ptyLostEmitted: false,
      };
      this.terminals.set(key, terminal);
    }

    // If PTY is already attached, we're done
    if (terminal.pty) {
      return tmuxSessionId;
    }

    // tmux 3.4 only honours the `=` exact-match prefix on has-session and
    // attach-session; set-option silently ignores it, so we use the bare id
    // here. The `=`-prefixed form is built below for attach-session.

    // Enable mouse mode
    const mouseProc = spawn(this.TMUX, ["set-option", "-t", tmuxSessionId, "mouse", "on"]);
    mouseProc.on("error", (err) => {
      console.error(`[MuxServer] Failed to set mouse mode for ${tmuxSessionId}:`, err.message);
    });

    // Hide the status bar
    const statusProc = spawn(this.TMUX, ["set-option", "-t", tmuxSessionId, "status", "off"]);
    statusProc.on("error", (err) => {
      console.error(`[MuxServer] Failed to hide status bar for ${tmuxSessionId}:`, err.message);
    });

    // Build environment
    const platformDefaults = getEnvDefaults();
    const homeDir = platformDefaults.HOME;
    const env = {
      HOME: platformDefaults.HOME,
      SHELL: platformDefaults.SHELL,
      USER: platformDefaults.USER,
      PATH: process.env.PATH || platformDefaults.PATH,
      TERM: "xterm-256color",
      LANG: process.env.LANG || "en_US.UTF-8",
      TMPDIR: platformDefaults.TMPDIR,
    };

    // Spawn PTY — use `=`-prefixed exact-match target so we never attach to
    // a session whose name happens to be a prefix of the requested id.
    const exactTmuxTarget = `=${tmuxSessionId}`;
    const pty = this.spawnTmuxPty(["attach-session", "-t", exactTmuxTarget], {
      name: "xterm-256color",
      cols: 80,
      rows: 24,
      cwd: homeDir,
      env,
    });

    terminal.pty = pty;

    // Schedule a grace-period reset of the re-attach counter. We only
    // consider an attach "really successful" if the PTY survives long
    // enough to suggest the underlying tmux session is actually usable.
    // The closure-captured `pty` reference is compared with terminal.pty
    // so a stale timer cannot reset the counter for a PTY that has
    // already exited or been replaced by re-attach. Any previously-
    // scheduled timer (from a now-replaced PTY) is cleared so we don't
    // keep its closure references reachable until the timer fires.
    if (terminal.resetTimer) {
      clearTimeout(terminal.resetTimer);
    }
    terminal.resetTimer = setTimeout(() => {
      terminal.resetTimer = undefined;
      if (terminal.pty === pty) {
        terminal.reattachAttempts = 0;
        terminal.ptyLostEmitted = false;
      }
    }, REATTACH_RESET_GRACE_MS);
    terminal.resetTimer.unref();

    // Wire up data events
    pty.onData((data: string) => {
      // Push to all subscribers — isolate each callback so a throw in one
      // (e.g. a closed ws.send) doesn't abort the loop or skip the buffer.
      for (const callback of terminal.subscribers) {
        try {
          callback(data);
        } catch (err) {
          console.error("[MuxServer] Subscriber callback threw:", err);
        }
      }

      // Append to ring buffer
      terminal.buffer.push(data);
      terminal.bufferBytes += Buffer.byteLength(data, "utf8");

      // Trim buffer if over limit
      while (terminal.bufferBytes > RING_BUFFER_MAX && terminal.buffer.length > 0) {
        const removed = terminal.buffer.shift() ?? "";
        terminal.bufferBytes -= Buffer.byteLength(removed, "utf8");
      }
    });

    // Handle PTY exit
    //
    // Async: the has-session probe shells out via promisified execFile and
    // must be awaited. node-pty fires onExit on the main thread; a sync
    // probe would freeze the entire web server (every WebSocket, HTTP
    // request, in-flight terminal) for up to the subprocess timeout when
    // tmux is slow to respond.
    pty.onExit(async ({ exitCode }) => {
      console.log(`[MuxServer] PTY exited for ${id} with code ${exitCode}`);
      terminal.pty = null;
      let reattachError: string | undefined;

      // Skip the re-attach loop entirely when the underlying tmux session is
      // gone (e.g. user pressed Ctrl-C in the pane and the launch command
      // exited, taking the only window with it). Without this guard we
      // burn three doomed attach-session spawns and emit a noisy
      // "Max re-attach attempts reached" log line for what is actually a
      // clean user-initiated termination — see issue #1756. The
      // MAX_REATTACH_ATTEMPTS bound from #1640 still covers tmux server
      // hiccups where the session does still exist.
      if (terminal.subscribers.size > 0 && !(await tmuxHasSession(this.TMUX, tmuxSessionId))) {
        console.log(`[MuxServer] tmux session ${tmuxSessionId} is gone, not re-attaching`);
        if (terminal.resetTimer) {
          clearTimeout(terminal.resetTimer);
          terminal.resetTimer = undefined;
        }
        if (!terminal.ptyLostEmitted) {
          terminal.ptyLostEmitted = true;
          recordActivityEvent({
            projectId,
            sessionId: id,
            source: "ui",
            kind: "ui.terminal_pty_lost",
            level: "warn",
            summary: `terminal PTY exited (code ${exitCode}) — tmux session gone`,
            data: {
              sessionId: id,
              exitCode,
              reattachAttempts: terminal.reattachAttempts,
              maxReattachAttempts: MAX_REATTACH_ATTEMPTS,
              reattachExhausted: false,
              reattachSkipped: true,
              tmuxSessionPresent: false,
              subscriberCount: terminal.subscribers.size,
            },
          });
        }
        for (const cb of terminal.exitCallbacks) {
          cb(exitCode);
        }
        return;
      }

      // Re-attach if subscribers are still present, up to MAX_REATTACH_ATTEMPTS.
      // The cap prevents an unbounded respawn loop when the PTY crashes immediately
      // after every attach (e.g. resource exhaustion or a broken tmux session).
      // The counter is reset by a delayed timer in open() once the new PTY has
      // survived REATTACH_RESET_GRACE_MS — see the comment on that constant.
      // Resetting here would defeat the cap: when ao stop kills the tmux session
      // out from under a still-subscribed dashboard, attach-session exits ~40 ms
      // after spawn and the loop runs at ~80 spawns/sec, exhausting the system
      // PTY pool in seconds (issue #1639).
      if (terminal.subscribers.size > 0 && terminal.reattachAttempts < MAX_REATTACH_ATTEMPTS) {
        terminal.reattachAttempts += 1;
        console.log(
          `[MuxServer] Re-attaching to ${id} (attempt ${terminal.reattachAttempts}/${MAX_REATTACH_ATTEMPTS})`,
        );
        try {
          this.open(id, projectId, tmuxSessionId);
          if (!terminal.ptyLostEmitted) {
            terminal.ptyLostEmitted = true;
            recordActivityEvent({
              projectId,
              sessionId: id,
              source: "ui",
              kind: "ui.terminal_pty_lost",
              level: "warn",
              summary: `terminal PTY exited (code ${exitCode}) — reattached`,
              data: {
                sessionId: id,
                exitCode,
                reattachAttempts: terminal.reattachAttempts,
                maxReattachAttempts: MAX_REATTACH_ATTEMPTS,
                reattachExhausted: false,
                reattachRecovered: true,
                subscriberCount: terminal.subscribers.size,
              },
            });
          }
          return; // re-attached — don't notify exit
        } catch (err) {
          reattachError = err instanceof Error ? err.message : String(err);
          console.error(`[MuxServer] Failed to re-attach ${id}:`, err);
        }
      } else if (terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS) {
        console.error(`[MuxServer] Max re-attach attempts reached for ${id}, giving up`);
      }

      // PTY actually died (vs user closed browser): only emit when subscribers
      // are still attached — otherwise the exit is just normal cleanup.
      // Keep this event one-shot for the terminal entry. Clients may re-open
      // the same terminal after a failed reattach; repeated PTY exits should
      // not flood the activity log for the same loss condition.
      if (terminal.subscribers.size > 0 && !terminal.ptyLostEmitted) {
        terminal.ptyLostEmitted = true;
        recordActivityEvent({
          projectId,
          sessionId: id,
          source: "ui",
          kind: "ui.terminal_pty_lost",
          level: "warn",
          summary: `terminal PTY exited (code ${exitCode})${
            terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS ? " — reattach exhausted" : ""
          }`,
          data: {
            sessionId: id,
            exitCode,
            reattachAttempts: terminal.reattachAttempts,
            maxReattachAttempts: MAX_REATTACH_ATTEMPTS,
            reattachExhausted: terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS,
            subscriberCount: terminal.subscribers.size,
            ...(reattachError ? { reattachError } : {}),
          },
        });
      }

      // Notify subscribers that the terminal has exited (re-attach failed or no subscribers)
      for (const cb of terminal.exitCallbacks) {
        cb(exitCode);
      }
    });

    console.log(`[MuxServer] Opened terminal ${id} (tmux: ${tmuxSessionId})`);
    return tmuxSessionId;
  }

  /**
   * Write data to the PTY if attached
   */
  write(id: string, data: string, projectId?: string): void {
    const terminal = this.terminals.get(this.terminalKey(id, projectId));
    if (terminal?.pty) {
      terminal.pty.write(data);
    }
  }

  /**
   * Resize the PTY if attached
   */
  resize(id: string, cols: number, rows: number, projectId?: string): void {
    const terminal = this.terminals.get(this.terminalKey(id, projectId));
    if (terminal?.pty) {
      terminal.pty.resize(cols, rows);
    }
  }

  /**
   * Subscribe to terminal data. Returns unsubscribe function.
   * Automatically opens the terminal if needed.
   * @param onExit - called when the PTY exits and cannot be re-attached
   */
  subscribe(
    id: string,
    projectId: string | undefined,
    callback: (data: string) => void,
    onExit?: (exitCode: number) => void,
  ): () => void {
    // Ensure terminal is open
    this.open(id, projectId);
    const key = this.terminalKey(id, projectId);
    const terminal = this.terminals.get(key);
    if (!terminal) {
      throw new Error(`Failed to open terminal: ${id}`);
    }

    // Add subscriber
    terminal.subscribers.add(callback);
    if (onExit) terminal.exitCallbacks.add(onExit);

    // Return unsubscribe function
    return () => {
      terminal.subscribers.delete(callback);
      if (onExit) terminal.exitCallbacks.delete(onExit);
      // Kill PTY and clean up when the last subscriber leaves
      if (terminal.subscribers.size === 0) {
        if (terminal.resetTimer) {
          clearTimeout(terminal.resetTimer);
          terminal.resetTimer = undefined;
        }
        if (terminal.pty) {
          terminal.pty.kill();
          terminal.pty = null;
        }
        this.terminals.delete(key);
      }
    };
  }

  /**
   * Get buffered data for a terminal
   */
  getBuffer(id: string, projectId?: string): string {
    const terminal = this.terminals.get(this.terminalKey(id, projectId));
    if (!terminal) return "";
    return terminal.buffer.join("");
  }
}

// ── Windows Pipe Relay (extracted for testability) ──

const intentionalWinPipeCloses = new WeakSet<Socket>();

/** Minimal WebSocket-like interface for the pipe relay handler */
export interface WsSink {
  send(data: string): void;
  readonly readyState: number;
}

/** Dependencies injected into the pipe relay handler */
export interface PipeRelayDeps {
  connect: (path: string) => Socket;
  resolvePipePath: (id: string, projectId?: string) => string | null;
}

/**
 * Handle a Windows terminal message by relaying through named pipes.
 * Extracted from the WebSocket connection handler for testability.
 */
export function handleWindowsPipeMessage(
  msg: {
    id: string;
    type: string;
    data?: string;
    cols?: number;
    rows?: number;
    projectId?: string;
  },
  ws: WsSink,
  winPipes: Map<string, Socket>,
  winPipeBuffers: Map<string, Buffer>,
  deps: PipeRelayDeps,
): void {
  const WS_OPEN = 1; // WebSocket.OPEN
  const { id, type, projectId } = msg;
  // MuxProvider keys subscribers under `${projectId}:${id}` when projectId is
  // provided, so every outbound terminal message must echo projectId back —
  // otherwise the client routes by id alone and the subscriber bucket
  // mismatches, leaving the xterm pane blank on /projects/[id]/sessions/[id].
  const echo = projectId ? { projectId } : {};
  // Project-scoped pipe-map key: matches the Unix `subscriptionKey` shape so
  // two projects sharing a sessionId on the same mux connection don't collide
  // on the same socket/buffer entry.
  const pipeKey = projectId ? `${projectId}:${id}` : id;

  // The Unix path validates inside TerminalManager.open(). The Windows pipe
  // relay bypasses TerminalManager entirely, so validate here too — `id`
  // becomes a map key and is constructed into a pipe path downstream.
  if (!validateSessionId(id)) {
    if (ws.readyState === WS_OPEN) {
      ws.send(
        JSON.stringify({
          ch: "terminal",
          id,
          type: "error",
          message: "invalid session id",
          ...echo,
        }),
      );
    }
    return;
  }

  if (type === "open") {
    if (winPipes.has(pipeKey)) {
      ws.send(JSON.stringify({ ch: "terminal", id, type: "opened", ...echo }));
    } else {
      const pipePath = deps.resolvePipePath(id, projectId);
      if (!pipePath) {
        throw new Error(`No PTY host pipe found for session ${id}`);
      }
      const pipeSocket = deps.connect(pipePath);
      winPipes.set(pipeKey, pipeSocket);
      winPipeBuffers.set(pipeKey, Buffer.alloc(0));
      let ptyLostEmitted = false;
      const recordWindowsPtyLost = (
        reason: "pipe_closed" | "host_not_alive" | "pipe_error",
        extra?: Record<string, unknown>,
      ): void => {
        if (ptyLostEmitted || ws.readyState !== WS_OPEN) return;
        ptyLostEmitted = true;
        recordActivityEvent({
          projectId,
          sessionId: id,
          source: "ui",
          kind: "ui.terminal_pty_lost",
          level: "warn",
          summary:
            reason === "host_not_alive"
              ? `terminal PTY host reported not alive for ${id}`
              : reason === "pipe_error"
                ? `terminal PTY host pipe errored for ${id}`
                : `terminal PTY host pipe closed for ${id}`,
          data: {
            sessionId: id,
            transport: "windows_pipe",
            reason,
            ...extra,
          },
        });
      };

      pipeSocket.on("error", (err) => {
        recordWindowsPtyLost("pipe_error", { errorMessage: err.message });
        winPipes.delete(pipeKey);
        winPipeBuffers.delete(pipeKey);
        pipeSocket.destroy();
        if (ws.readyState === WS_OPEN) {
          ws.send(
            JSON.stringify({
              ch: "terminal",
              id,
              type: "error",
              message: `PTY host not available: ${err.message}`,
              ...echo,
            }),
          );
        }
      });

      pipeSocket.on("connect", () => {
        if (ws.readyState === WS_OPEN) {
          ws.send(JSON.stringify({ ch: "terminal", id, type: "opened", ...echo }));
        }

        pipeSocket.on("data", (chunk: Buffer) => {
          const existing = winPipeBuffers.get(pipeKey) ?? Buffer.alloc(0);
          let buf = Buffer.concat([existing, chunk]);
          winPipeBuffers.set(pipeKey, buf);

          while (buf.length >= 5) {
            const msgType = buf.readUInt8(0);
            const length = buf.readUInt32BE(1);
            if (buf.length < 5 + length) break;
            const payload = buf.subarray(5, 5 + length);
            buf = buf.subarray(5 + length);
            winPipeBuffers.set(pipeKey, buf);

            if (msgType === 0x01 && ws.readyState === WS_OPEN) {
              ws.send(
                JSON.stringify({
                  ch: "terminal",
                  id,
                  type: "data",
                  data: payload.toString("utf-8"),
                  ...echo,
                }),
              );
            }
            if (msgType === 0x07) {
              try {
                const status = JSON.parse(payload.toString("utf-8")) as { alive: boolean };
                if (!status.alive && ws.readyState === WS_OPEN) {
                  recordWindowsPtyLost("host_not_alive");
                  ws.send(JSON.stringify({ ch: "terminal", id, type: "exited", code: 0, ...echo }));
                }
              } catch {
                /* ignore parse errors */
              }
            }
          }
        });

        pipeSocket.on("close", () => {
          winPipes.delete(pipeKey);
          winPipeBuffers.delete(pipeKey);
          const intentionalClose = intentionalWinPipeCloses.delete(pipeSocket);
          if (ws.readyState === WS_OPEN) {
            if (!intentionalClose) {
              recordWindowsPtyLost("pipe_closed");
            }
            ws.send(JSON.stringify({ ch: "terminal", id, type: "exited", code: 0, ...echo }));
          }
        });
      });
    }
  } else if (type === "data" && msg.data !== undefined) {
    const pipeSocket = winPipes.get(pipeKey);
    if (pipeSocket) {
      const inputBuf = Buffer.from(msg.data, "utf-8");
      const header = Buffer.alloc(5);
      header.writeUInt8(0x02, 0);
      header.writeUInt32BE(inputBuf.length, 1);
      pipeSocket.write(Buffer.concat([header, inputBuf]));
    }
  } else if (type === "resize" && msg.cols !== undefined && msg.rows !== undefined) {
    const pipeSocket = winPipes.get(pipeKey);
    if (pipeSocket) {
      const resizePayload = Buffer.from(JSON.stringify({ cols: msg.cols, rows: msg.rows }));
      const header = Buffer.alloc(5);
      header.writeUInt8(0x03, 0);
      header.writeUInt32BE(resizePayload.length, 1);
      pipeSocket.write(Buffer.concat([header, resizePayload]));
    }
  } else if (type === "close") {
    const pipeSocket = winPipes.get(pipeKey);
    if (pipeSocket) {
      intentionalWinPipeCloses.add(pipeSocket);
      pipeSocket.end();
      winPipes.delete(pipeKey);
      winPipeBuffers.delete(pipeKey);
    }
  }
}

/**
 * Create a mux WebSocket server (noServer mode).
 * Returns the WebSocketServer instance for manual upgrade routing.
 */
export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | null {
  // On Windows, we use named pipe relay instead of node-pty/tmux.
  // Allow the server to be created without ptySpawn on Windows.
  if (!ptySpawn && !isWindows()) {
    console.warn("[MuxServer] node-pty not available — mux WebSocket will be disabled");
    return null;
  }

  // On Windows, terminal I/O goes through named pipe relay — no TerminalManager needed.
  const terminalManager =
    ptySpawn && !isWindows() ? new TerminalManager(tmuxPath ?? undefined) : null;

  const nextPort = process.env.PORT || "3000";
  const broadcaster = new SessionBroadcaster(nextPort);
  const notificationBroadcaster = new NotificationBroadcaster();

  const wss = new WebSocketServer({ noServer: true });

  wss.on("connection", (ws, request) => {
    console.log("[MuxServer] New mux connection");

    const connectedAt = Date.now();
    // Best-effort remote addr — proxy headers if present, else socket peer.
    const xff = request?.headers["x-forwarded-for"];
    const xffStr = Array.isArray(xff) ? xff[0] : xff;
    const remoteAddr =
      (typeof xffStr === "string" ? xffStr.split(",")[0]?.trim() : undefined) ??
      request?.socket?.remoteAddress ??
      undefined;

    recordActivityEvent({
      source: "ui",
      kind: "ui.terminal_connected",
      level: "info",
      summary: "mux WebSocket connection opened",
      data: { remoteAddr },
    });

    const subscriptions = new Map<string, () => void>();
    // Windows: named pipe sockets keyed by session ID
    const winPipes = new Map<string, ReturnType<typeof netConnect>>();
    // Windows: framing buffers keyed by session ID
    const winPipeBuffers = new Map<string, Buffer>();
    let sessionUnsubscribe: (() => void) | null = null;
    let notificationUnsubscribe: (() => void) | null = null;
    let missedPongs = 0;
    let heartbeatLostEmitted = false;
    const MAX_MISSED_PONGS = 3;

    // Heartbeat: send native WebSocket ping every 15s.
    // Browsers automatically respond to native pings with pong frames —
    // no application-level code is needed on the client side.
    const heartbeatInterval = setInterval(() => {
      if (ws.readyState === WebSocket.OPEN) {
        // Send the ping first so it counts as a sent-but-unanswered probe
        ws.ping();
        missedPongs += 1;
        if (missedPongs >= MAX_MISSED_PONGS) {
          console.log("[MuxServer] Too many missed pongs, terminating connection");
          if (!heartbeatLostEmitted) {
            heartbeatLostEmitted = true;
            recordActivityEvent({
              source: "ui",
              kind: "ui.terminal_heartbeat_lost",
              level: "warn",
              summary: `mux WebSocket heartbeat lost (${missedPongs} missed pongs)`,
              data: {
                missedPongs,
                maxMissedPongs: MAX_MISSED_PONGS,
                connectionAgeMs: Date.now() - connectedAt,
                remoteAddr,
                subscriberCount: subscriptions.size,
              },
            });
          }
          ws.terminate();
        }
      }
    }, 15_000);

    // Native pong resets the missed counter
    ws.on("pong", () => {
      missedPongs = 0;
    });

    /**
     * Handle incoming messages
     */
    ws.on("message", (data) => {
      try {
        const msg = JSON.parse(data.toString("utf8")) as ClientMessage;

        if (msg.ch === "system") {
          if (msg.type === "ping") {
            const pong: ServerMessage = { ch: "system", type: "pong" };
            ws.send(JSON.stringify(pong));
          }
        } else if (msg.ch === "terminal") {
          const { id, type } = msg;
          const projectId = "projectId" in msg ? msg.projectId : undefined;
          const subscriptionKey = projectId ? `${projectId}:${id}` : id;

          try {
            if (type === "open") {
              if (isWindows()) {
                handleWindowsPipeMessage(
                  msg as {
                    id: string;
                    type: string;
                    projectId?: string;
                    data?: string;
                    cols?: number;
                    rows?: number;
                  },
                  ws,
                  winPipes,
                  winPipeBuffers,
                  { connect: netConnect, resolvePipePath },
                );
              } else {
                // --- Unix: tmux path with project scoping ---
                if (!terminalManager) throw new Error("Terminal manager not available");
                terminalManager.open(id, projectId, "tmuxName" in msg ? msg.tmuxName : undefined);

                // Send opened confirmation (idempotent — safe to send on re-open)
                const openedMsg: ServerMessage = {
                  ch: "terminal",
                  id,
                  type: "opened",
                  ...(projectId && { projectId }),
                };
                ws.send(JSON.stringify(openedMsg));

                // Subscribe and send history buffer only for new subscribers.
                // Skipping the buffer on re-open prevents duplicate output when
                // MuxProvider re-sends open for all terminals on reconnect.
                if (!subscriptions.has(subscriptionKey)) {
                  // Send buffered history to catch up the new subscriber
                  const buffer = terminalManager.getBuffer(id, projectId);
                  if (buffer) {
                    const bufferMsg: ServerMessage = {
                      ch: "terminal",
                      id,
                      type: "data",
                      data: buffer,
                      ...(projectId && { projectId }),
                    };
                    ws.send(JSON.stringify(bufferMsg));
                  }
                  const unsub = terminalManager.subscribe(
                    id,
                    projectId,
                    (data) => {
                      const dataMsg: ServerMessage = {
                        ch: "terminal",
                        id,
                        type: "data",
                        data,
                        ...(projectId && { projectId }),
                      };
                      if (ws.readyState === WebSocket.OPEN) {
                        ws.send(JSON.stringify(dataMsg));
                      }
                    },
                    (exitCode) => {
                      const exitedMsg: ServerMessage = {
                        ch: "terminal",
                        id,
                        type: "exited",
                        code: exitCode,
                        ...(projectId && { projectId }),
                      };
                      if (ws.readyState === WebSocket.OPEN) {
                        ws.send(JSON.stringify(exitedMsg));
                      }
                    },
                  );
                  subscriptions.set(subscriptionKey, unsub);
                }
              }
            } else if (type === "data" && "data" in msg) {
              if (isWindows()) {
                handleWindowsPipeMessage(
                  msg as { id: string; type: string; projectId?: string; data: string },
                  ws,
                  winPipes,
                  winPipeBuffers,
                  { connect: netConnect, resolvePipePath },
                );
              } else {
                terminalManager?.write(id, msg.data, projectId);
              }
            } else if (type === "resize" && "cols" in msg && "rows" in msg) {
              if (isWindows()) {
                handleWindowsPipeMessage(
                  msg as {
                    id: string;
                    type: string;
                    projectId?: string;
                    cols: number;
                    rows: number;
                  },
                  ws,
                  winPipes,
                  winPipeBuffers,
                  { connect: netConnect, resolvePipePath },
                );
              } else {
                terminalManager?.resize(id, msg.cols, msg.rows, projectId);
              }
            } else if (type === "close") {
              if (isWindows()) {
                handleWindowsPipeMessage(
                  msg as {
                    id: string;
                    type: string;
                    projectId?: string;
                    data?: string;
                    cols?: number;
                    rows?: number;
                  },
                  ws,
                  winPipes,
                  winPipeBuffers,
                  { connect: netConnect, resolvePipePath },
                );
              } else {
                // Unsubscribe this client only — TerminalManager is shared across
                // all mux connections so we must not kill the PTY here.
                const unsub = subscriptions.get(subscriptionKey);
                if (unsub) {
                  unsub();
                  subscriptions.delete(subscriptionKey);
                }
              }
            }
          } catch (err) {
            if (ws.readyState === WebSocket.OPEN) {
              const errorMsg: ServerMessage = {
                ch: "terminal",
                id,
                type: "error",
                message: err instanceof Error ? err.message : String(err),
                ...(projectId && { projectId }),
              };
              ws.send(JSON.stringify(errorMsg));
            }
          }
        } else if (msg.ch === "subscribe") {
          if (msg.topics.includes("sessions") && !sessionUnsubscribe) {
            sessionUnsubscribe = broadcaster.subscribe(
              (sessions) => {
                if (ws.readyState !== WebSocket.OPEN) return;
                if (ws.bufferedAmount > WS_BUFFER_HIGH_WATERMARK) {
                  console.warn("[MuxServer] Skipping session snapshot — socket backpressured");
                  return;
                }
                const snapMsg: ServerMessage = { ch: "sessions", type: "snapshot", sessions };
                ws.send(JSON.stringify(snapMsg));
              },
              (error) => {
                if (ws.readyState !== WebSocket.OPEN) return;
                const errMsg: ServerMessage = { ch: "sessions", type: "error", error };
                ws.send(JSON.stringify(errMsg));
              },
            );
          }
          if (msg.topics.includes("notifications") && !notificationUnsubscribe) {
            notificationUnsubscribe = notificationBroadcaster.subscribe(
              (notifications, type, limit) => {
                if (ws.readyState !== WebSocket.OPEN) return;
                if (ws.bufferedAmount > WS_BUFFER_HIGH_WATERMARK) {
                  console.warn("[MuxServer] Skipping notification update — socket backpressured");
                  return;
                }
                const msg: ServerMessage = {
                  ch: "notifications",
                  type,
                  notifications,
                  limit,
                };
                ws.send(JSON.stringify(msg));
              },
              (error) => {
                if (ws.readyState !== WebSocket.OPEN) return;
                const errMsg: ServerMessage = { ch: "notifications", type: "error", error };
                ws.send(JSON.stringify(errMsg));
              },
            );
          }
        }
      } catch (err) {
        console.error("[MuxServer] Failed to parse message:", err);
        recordActivityEvent({
          source: "ui",
          kind: "ui.terminal_protocol_error",
          level: "warn",
          summary: "invalid mux client message — parse failed",
          data: {
            errorMessage: err instanceof Error ? err.message : String(err),
            remoteAddr,
            subscriberCount: subscriptions.size,
          },
        });
        const errorMsg: ServerMessage = {
          ch: "system",
          type: "error",
          message: "Invalid message format",
        };
        if (ws.readyState === WebSocket.OPEN) {
          ws.send(JSON.stringify(errorMsg));
        }
      }
    });

    /**
     * Handle connection close
     */
    ws.on("close", (code, reason) => {
      console.log("[MuxServer] Mux connection closed");
      recordActivityEvent({
        source: "ui",
        kind: "ui.terminal_disconnected",
        level: "info",
        summary: "mux WebSocket connection closed",
        data: {
          code,
          reason: reason?.toString("utf8") || undefined,
          connectionAgeMs: Date.now() - connectedAt,
          subscriberCount: subscriptions.size,
          heartbeatLost: heartbeatLostEmitted,
          remoteAddr,
        },
      });
      clearInterval(heartbeatInterval);
      sessionUnsubscribe?.();
      sessionUnsubscribe = null;
      notificationUnsubscribe?.();
      notificationUnsubscribe = null;
      for (const unsub of subscriptions.values()) {
        unsub();
      }
      subscriptions.clear();
      // Windows: close all open pipe sockets
      for (const pipeSocket of winPipes.values()) {
        pipeSocket.destroy();
      }
      winPipes.clear();
      winPipeBuffers.clear();
    });

    // In the ws library, "error" is always followed by "close", so the close
    // handler below handles all cleanup.  Log the error here and nothing more.
    ws.on("error", (err) => {
      console.error("[MuxServer] WebSocket error:", err.message);
    });
  });

  console.log("[MuxServer] Mux WebSocket server created (noServer mode)");
  return wss;
}