import { EventEmitter } from "node:events";import fs from "node:fs";import path from "node:path"; import {  DEFAULT_TERMINAL_ID,  TerminalClearInput,  TerminalCloseInput,  TerminalOpenInput,  TerminalResizeInput,  TerminalWriteInput,  type TerminalEvent,  type TerminalSessionSnapshot,} from "@t3tools/contracts";import { Effect, Encoding, Layer, Path, Schema } from "effect"; import { createLogger } from "../../logger";import { PtyAdapter, PtyAdapterShape, type PtyExitEvent, type PtyProcess } from "../Services/PTY";import { runProcess } from "../../processRunner";import { ServerConfig } from "../../config";import {  ShellCandidate,  TerminalError,  TerminalManager,  TerminalManagerShape,  TerminalSessionState,  TerminalStartInput,} from "../Services/Manager"; const DEFAULT_HISTORY_LINE_LIMIT = 5_000;const DEFAULT_PERSIST_DEBOUNCE_MS = 40;const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000;const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128;const DEFAULT_OPEN_COLS = 120;const DEFAULT_OPEN_ROWS = 30;const TERMINAL_ENV_BLOCKLIST = new Set(["PORT", "ELECTRON_RENDERER_PORT", "ELECTRON_RUN_AS_NODE"]); const decodeTerminalOpenInput = Schema.decodeUnknownSync(TerminalOpenInput);const decodeTerminalWriteInput = Schema.decodeUnknownSync(TerminalWriteInput);const decodeTerminalResizeInput = Schema.decodeUnknownSync(TerminalResizeInput);const decodeTerminalClearInput = Schema.decodeUnknownSync(TerminalClearInput);const decodeTerminalCloseInput = Schema.decodeUnknownSync(TerminalCloseInput); type TerminalSubprocessChecker = (terminalPid: number) => Promise<boolean>; function defaultShellResolver(): string {  if (process.platform === "win32") {    return process.env.ComSpec ?? "cmd.exe";  }  return process.env.SHELL ?? "bash";} function normalizeShellCommand(value: string | undefined): string | null {  if (!value) return null;  const trimmed = value.trim();  if (trimmed.length === 0) return null;   if (process.platform === "win32") {    return trimmed;  }   const firstToken = trimmed.split(/\s+/g)[0]?.trim();  if (!firstToken) return null;  return firstToken.replace(/^['"]|['"]$/g, "");} function shellCandidateFromCommand(command: string | null): ShellCandidate | null {  if (!command || command.length === 0) return null;  const shellName = path.basename(command).toLowerCase();  if (process.platform !== "win32" && shellName === "zsh") {    return { shell: command, args: ["-o", "nopromptsp"] };  }  return { shell: command };} function formatShellCandidate(candidate: ShellCandidate): string {  if (!candidate.args || candidate.args.length === 0) return candidate.shell;  return `${candidate.shell} ${candidate.args.join(" ")}`;} function uniqueShellCandidates(candidates: Array<ShellCandidate | null>): ShellCandidate[] {  const seen = new Set<string>();  const ordered: ShellCandidate[] = [];  for (const candidate of candidates) {    if (!candidate) continue;    const key = formatShellCandidate(candidate);    if (seen.has(key)) continue;    seen.add(key);    ordered.push(candidate);  }  return ordered;} function resolveShellCandidates(shellResolver: () => string): ShellCandidate[] {  const requested = shellCandidateFromCommand(normalizeShellCommand(shellResolver()));   if (process.platform === "win32") {    return uniqueShellCandidates([      requested,      shellCandidateFromCommand(process.env.ComSpec ?? null),      shellCandidateFromCommand("powershell.exe"),      shellCandidateFromCommand("cmd.exe"),    ]);  }   return uniqueShellCandidates([    requested,    shellCandidateFromCommand(normalizeShellCommand(process.env.SHELL)),    shellCandidateFromCommand("/bin/zsh"),    shellCandidateFromCommand("/bin/bash"),    shellCandidateFromCommand("/bin/sh"),    shellCandidateFromCommand("zsh"),    shellCandidateFromCommand("bash"),    shellCandidateFromCommand("sh"),  ]);} function isRetryableShellSpawnError(error: unknown): boolean {  if (!(error instanceof Error)) return false;  const message = error.message.toLowerCase();  return (    message.includes("posix_spawnp failed") ||    message.includes("enoent") ||    message.includes("not found") ||    message.includes("file not found") ||    message.includes("no such file")  );} async function checkWindowsSubprocessActivity(terminalPid: number): Promise<boolean> {  const command = [    `$children = Get-CimInstance Win32_Process -Filter "ParentProcessId = ${terminalPid}" -ErrorAction SilentlyContinue`,    "if ($children) { exit 0 }",    "exit 1",  ].join("; ");  try {    const result = await runProcess(      "powershell.exe",      ["-NoProfile", "-NonInteractive", "-Command", command],      {        timeoutMs: 1_500,        allowNonZeroExit: true,        maxBufferBytes: 32_768,        outputMode: "truncate",      },    );    return result.code === 0;  } catch {    return false;  }} async function checkPosixSubprocessActivity(terminalPid: number): Promise<boolean> {  try {    const pgrepResult = await runProcess("pgrep", ["-P", String(terminalPid)], {      timeoutMs: 1_000,      allowNonZeroExit: true,      maxBufferBytes: 32_768,      outputMode: "truncate",    });    if (pgrepResult.code === 0) {      return pgrepResult.stdout.trim().length > 0;    }    if (pgrepResult.code === 1) {      return false;    }  } catch {    // Fall back to ps when pgrep is unavailable.  }   try {    const psResult = await runProcess("ps", ["-eo", "pid=,ppid="], {      timeoutMs: 1_000,      allowNonZeroExit: true,      maxBufferBytes: 262_144,      outputMode: "truncate",    });    if (psResult.code !== 0) {      return false;    }     for (const line of psResult.stdout.split(/\r?\n/g)) {      const [pidRaw, ppidRaw] = line.trim().split(/\s+/g);      const pid = Number(pidRaw);      const ppid = Number(ppidRaw);      if (!Number.isInteger(pid) || !Number.isInteger(ppid)) continue;      if (ppid === terminalPid) {        return true;      }    }    return false;  } catch {    return false;  }} async function defaultSubprocessChecker(terminalPid: number): Promise<boolean> {  if (!Number.isInteger(terminalPid) || terminalPid <= 0) {    return false;  }  if (process.platform === "win32") {    return checkWindowsSubprocessActivity(terminalPid);  }  return checkPosixSubprocessActivity(terminalPid);} function capHistory(history: string, maxLines: number): string {  if (history.length === 0) return history;  const hasTrailingNewline = history.endsWith("\n");  const lines = history.split("\n");  if (hasTrailingNewline) {    lines.pop();  }  if (lines.length <= maxLines) return history;  const capped = lines.slice(lines.length - maxLines).join("\n");  return hasTrailingNewline ? `${capped}\n` : capped;} function legacySafeThreadId(threadId: string): string {  return threadId.replace(/[^a-zA-Z0-9._-]/g, "_");} function toSafeThreadId(threadId: string): string {  return `terminal_${Encoding.encodeBase64Url(threadId)}`;} function toSafeTerminalId(terminalId: string): string {  return Encoding.encodeBase64Url(terminalId);} function toSessionKey(threadId: string, terminalId: string): string {  return `${threadId}\u0000${terminalId}`;} function shouldExcludeTerminalEnvKey(key: string): boolean {  const normalizedKey = key.toUpperCase();  if (normalizedKey.startsWith("T3CODE_")) {    return true;  }  if (normalizedKey.startsWith("VITE_")) {    return true;  }  return TERMINAL_ENV_BLOCKLIST.has(normalizedKey);} function createTerminalSpawnEnv(  baseEnv: NodeJS.ProcessEnv,  runtimeEnv?: Record<string, string> | null,): NodeJS.ProcessEnv {  const spawnEnv: NodeJS.ProcessEnv = {};  for (const [key, value] of Object.entries(baseEnv)) {    if (value === undefined) continue;    if (shouldExcludeTerminalEnvKey(key)) continue;    spawnEnv[key] = value;  }  if (runtimeEnv) {    for (const [key, value] of Object.entries(runtimeEnv)) {      spawnEnv[key] = value;    }  }  return spawnEnv;} function normalizedRuntimeEnv(  env: Record<string, string> | undefined,): Record<string, string> | null {  if (!env) return null;  const entries = Object.entries(env);  if (entries.length === 0) return null;  return Object.fromEntries(entries.toSorted(([left], [right]) => left.localeCompare(right)));} interface TerminalManagerEvents {  event: [event: TerminalEvent];} interface TerminalManagerOptions {  logsDir?: string;  historyLineLimit?: number;  ptyAdapter: PtyAdapterShape;  shellResolver?: () => string;  subprocessChecker?: TerminalSubprocessChecker;  subprocessPollIntervalMs?: number;  processKillGraceMs?: number;  maxRetainedInactiveSessions?: number;} export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents> {  private readonly sessions = new Map<string, TerminalSessionState>();  private readonly logsDir: string;  private readonly historyLineLimit: number;  private readonly ptyAdapter: PtyAdapterShape;  private readonly shellResolver: () => string;  private readonly persistQueues = new Map<string, Promise<void>>();  private readonly persistTimers = new Map<string, ReturnType<typeof setTimeout>>();  private readonly pendingPersistHistory = new Map<string, string>();  private readonly threadLocks = new Map<string, Promise<void>>();  private readonly persistDebounceMs: number;  private readonly subprocessChecker: TerminalSubprocessChecker;  private readonly subprocessPollIntervalMs: number;  private readonly processKillGraceMs: number;  private readonly maxRetainedInactiveSessions: number;  private subprocessPollTimer: ReturnType<typeof setInterval> | null = null;  private subprocessPollInFlight = false;  private readonly killEscalationTimers = new Map<PtyProcess, ReturnType<typeof setTimeout>>();  private readonly logger = createLogger("terminal");   constructor(options: TerminalManagerOptions) {    super();    this.logsDir = options.logsDir ?? path.resolve(process.cwd(), ".logs", "terminals");    this.historyLineLimit = options.historyLineLimit ?? DEFAULT_HISTORY_LINE_LIMIT;    this.ptyAdapter = options.ptyAdapter;    this.shellResolver = options.shellResolver ?? defaultShellResolver;    this.persistDebounceMs = DEFAULT_PERSIST_DEBOUNCE_MS;    this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker;    this.subprocessPollIntervalMs =      options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS;    this.processKillGraceMs = options.processKillGraceMs ?? DEFAULT_PROCESS_KILL_GRACE_MS;    this.maxRetainedInactiveSessions =      options.maxRetainedInactiveSessions ?? DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS;    fs.mkdirSync(this.logsDir, { recursive: true });  }   async open(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {    const input = decodeTerminalOpenInput(raw);    return this.runWithThreadLock(input.threadId, async () => {      await this.assertValidCwd(input.cwd);       const sessionKey = toSessionKey(input.threadId, input.terminalId);      const existing = this.sessions.get(sessionKey);      if (!existing) {        await this.flushPersistQueue(input.threadId, input.terminalId);        const history = await this.readHistory(input.threadId, input.terminalId);        const cols = input.cols ?? DEFAULT_OPEN_COLS;        const rows = input.rows ?? DEFAULT_OPEN_ROWS;        const session: TerminalSessionState = {          threadId: input.threadId,          terminalId: input.terminalId,          cwd: input.cwd,          status: "starting",          pid: null,          history,          exitCode: null,          exitSignal: null,          updatedAt: new Date().toISOString(),          cols,          rows,          process: null,          unsubscribeData: null,          unsubscribeExit: null,          hasRunningSubprocess: false,          runtimeEnv: normalizedRuntimeEnv(input.env),        };        this.sessions.set(sessionKey, session);        this.evictInactiveSessionsIfNeeded();        this.startSession(session, { ...input, cols, rows }, "started");        return this.snapshot(session);      }       const nextRuntimeEnv = normalizedRuntimeEnv(input.env);      const currentRuntimeEnv = existing.runtimeEnv;      const targetCols = input.cols ?? existing.cols;      const targetRows = input.rows ?? existing.rows;      const runtimeEnvChanged =        JSON.stringify(currentRuntimeEnv) !== JSON.stringify(nextRuntimeEnv);       if (existing.cwd !== input.cwd || runtimeEnvChanged) {        this.stopProcess(existing);        existing.cwd = input.cwd;        existing.runtimeEnv = nextRuntimeEnv;        existing.history = "";        await this.persistHistory(existing.threadId, existing.terminalId, existing.history);      } else if (existing.status === "exited" || existing.status === "error") {        existing.runtimeEnv = nextRuntimeEnv;        existing.history = "";        await this.persistHistory(existing.threadId, existing.terminalId, existing.history);      } else if (currentRuntimeEnv !== nextRuntimeEnv) {        existing.runtimeEnv = nextRuntimeEnv;      }       if (!existing.process) {        this.startSession(existing, { ...input, cols: targetCols, rows: targetRows }, "started");        return this.snapshot(existing);      }       if (existing.cols !== targetCols || existing.rows !== targetRows) {        existing.cols = targetCols;        existing.rows = targetRows;        existing.process.resize(targetCols, targetRows);        existing.updatedAt = new Date().toISOString();      }       return this.snapshot(existing);    });  }   async write(raw: TerminalWriteInput): Promise<void> {    const input = decodeTerminalWriteInput(raw);    const session = this.requireSession(input.threadId, input.terminalId);    if (!session.process || session.status !== "running") {      throw new Error(        `Terminal is not running for thread: ${input.threadId}, terminal: ${input.terminalId}`,      );    }    session.process.write(input.data);  }   async resize(raw: TerminalResizeInput): Promise<void> {    const input = decodeTerminalResizeInput(raw);    const session = this.requireSession(input.threadId, input.terminalId);    if (!session.process || session.status !== "running") {      throw new Error(        `Terminal is not running for thread: ${input.threadId}, terminal: ${input.terminalId}`,      );    }    session.cols = input.cols;    session.rows = input.rows;    session.updatedAt = new Date().toISOString();    session.process.resize(input.cols, input.rows);  }   async clear(raw: TerminalClearInput): Promise<void> {    const input = decodeTerminalClearInput(raw);    await this.runWithThreadLock(input.threadId, async () => {      const session = this.requireSession(input.threadId, input.terminalId);      session.history = "";      session.updatedAt = new Date().toISOString();      await this.persistHistory(input.threadId, input.terminalId, session.history);      this.emitEvent({        type: "cleared",        threadId: input.threadId,        terminalId: input.terminalId,        createdAt: new Date().toISOString(),      });    });  }   async restart(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {    const input = decodeTerminalOpenInput(raw);    return this.runWithThreadLock(input.threadId, async () => {      await this.assertValidCwd(input.cwd);       const sessionKey = toSessionKey(input.threadId, input.terminalId);      let session = this.sessions.get(sessionKey);      if (!session) {        const cols = input.cols ?? DEFAULT_OPEN_COLS;        const rows = input.rows ?? DEFAULT_OPEN_ROWS;        session = {          threadId: input.threadId,          terminalId: input.terminalId,          cwd: input.cwd,          status: "starting",          pid: null,          history: "",          exitCode: null,          exitSignal: null,          updatedAt: new Date().toISOString(),          cols,          rows,          process: null,          unsubscribeData: null,          unsubscribeExit: null,          hasRunningSubprocess: false,          runtimeEnv: normalizedRuntimeEnv(input.env),        };        this.sessions.set(sessionKey, session);        this.evictInactiveSessionsIfNeeded();      } else {        this.stopProcess(session);        session.cwd = input.cwd;        session.runtimeEnv = normalizedRuntimeEnv(input.env);      }       const cols = input.cols ?? session.cols;      const rows = input.rows ?? session.rows;       session.history = "";      await this.persistHistory(input.threadId, input.terminalId, session.history);      this.startSession(session, { ...input, cols, rows }, "restarted");      return this.snapshot(session);    });  }   async close(raw: TerminalCloseInput): Promise<void> {    const input = decodeTerminalCloseInput(raw);    await this.runWithThreadLock(input.threadId, async () => {      if (input.terminalId) {        await this.closeSession(input.threadId, input.terminalId, input.deleteHistory === true);        return;      }       const threadSessions = this.sessionsForThread(input.threadId);      for (const session of threadSessions) {        this.stopProcess(session);        this.sessions.delete(toSessionKey(session.threadId, session.terminalId));      }      await Promise.all(        threadSessions.map((session) =>          this.flushPersistQueue(session.threadId, session.terminalId),

Lines 1–500 of 1162. Load more