Julius Marmingelint fix47d5dc28 months agoHistory
975 lines · 31.4 KB · TypeScript
RawDownload
import { EventEmitter } from "node:events";import fs from "node:fs";import path from "node:path"; import {  DEFAULT_TERMINAL_ID,  type TerminalClearInput,  type TerminalCloseInput,  type TerminalEvent,  type TerminalOpenInput,  type TerminalResizeInput,  type TerminalSessionSnapshot,  type TerminalSessionStatus,  type TerminalWriteInput,  terminalClearInputSchema,  terminalCloseInputSchema,  terminalOpenInputSchema,  terminalResizeInputSchema,  terminalWriteInputSchema,} from "@t3tools/contracts"; import { createLogger } from "./logger";import { NodePtyAdapter, type PtyAdapter, type PtyExitEvent, type PtyProcess } from "./ptyAdapter";import { runProcess } from "./processRunner"; const DEFAULT_HISTORY_LINE_LIMIT = 5_000;const DEFAULT_PERSIST_DEBOUNCE_MS = 40;const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;const TERMINAL_ENV_BLOCKLIST = new Set(["PORT", "ELECTRON_RENDERER_PORT", "ELECTRON_RUN_AS_NODE"]); type TerminalSubprocessChecker = (terminalPid: number) => Promise<boolean>; export interface TerminalManagerEvents {  event: [event: TerminalEvent];} export interface TerminalManagerOptions {  logsDir?: string;  historyLineLimit?: number;  ptyAdapter?: PtyAdapter;  shellResolver?: () => string;  subprocessChecker?: TerminalSubprocessChecker;  subprocessPollIntervalMs?: number;} interface TerminalSessionState {  threadId: string;  terminalId: string;  cwd: string;  status: TerminalSessionStatus;  pid: number | null;  history: string;  exitCode: number | null;  exitSignal: number | null;  updatedAt: string;  cols: number;  rows: number;  process: PtyProcess | null;  unsubscribeData: (() => void) | null;  unsubscribeExit: (() => void) | null;  hasRunningSubprocess: 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 uniqueShells(shells: Array<string | null>): string[] {  const seen = new Set<string>();  const ordered: string[] = [];  for (const shell of shells) {    if (!shell || shell.length === 0) continue;    if (seen.has(shell)) continue;    seen.add(shell);    ordered.push(shell);  }  return ordered;} function resolveShellCandidates(shellResolver: () => string): string[] {  const requested = normalizeShellCommand(shellResolver());   if (process.platform === "win32") {    return uniqueShells([requested, process.env.ComSpec ?? null, "powershell.exe", "cmd.exe"]);  }   return uniqueShells([    requested,    normalizeShellCommand(process.env.SHELL),    "/bin/zsh",    "/bin/bash",    "/bin/sh",    "zsh",    "bash",    "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_${Buffer.from(threadId, "utf8").toString("base64url")}`;} function toSafeTerminalId(terminalId: string): string {  return Buffer.from(terminalId, "utf8").toString("base64url");} 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): 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;  }  return spawnEnv;} export class TerminalManager extends EventEmitter<TerminalManagerEvents> {  private readonly sessions = new Map<string, TerminalSessionState>();  private readonly logsDir: string;  private readonly historyLineLimit: number;  private readonly ptyAdapter: PtyAdapter;  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 subprocessPollTimer: ReturnType<typeof setInterval> | null = null;  private subprocessPollInFlight = false;  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 ?? new NodePtyAdapter();    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;    fs.mkdirSync(this.logsDir, { recursive: true });  }   async open(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {    const input = terminalOpenInputSchema.parse(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 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: input.cols,          rows: input.rows,          process: null,          unsubscribeData: null,          unsubscribeExit: null,          hasRunningSubprocess: false,        };        this.sessions.set(sessionKey, session);        this.startSession(session, input, "started");        return this.snapshot(session);      }       if (existing.cwd !== input.cwd) {        this.stopProcess(existing);        existing.cwd = input.cwd;        existing.history = "";        await this.persistHistory(existing.threadId, existing.terminalId, existing.history);      } else if (existing.status === "exited" || existing.status === "error") {        existing.history = "";        await this.persistHistory(existing.threadId, existing.terminalId, existing.history);      }       if (!existing.process) {        this.startSession(existing, input, "started");        return this.snapshot(existing);      }       if (existing.cols !== input.cols || existing.rows !== input.rows) {        existing.cols = input.cols;        existing.rows = input.rows;        existing.process.resize(input.cols, input.rows);        existing.updatedAt = new Date().toISOString();      }       return this.snapshot(existing);    });  }   async write(raw: TerminalWriteInput): Promise<void> {    const input = terminalWriteInputSchema.parse(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 = terminalResizeInputSchema.parse(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 = terminalClearInputSchema.parse(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 = terminalOpenInputSchema.parse(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) {        session = {          threadId: input.threadId,          terminalId: input.terminalId,          cwd: input.cwd,          status: "starting",          pid: null,          history: "",          exitCode: null,          exitSignal: null,          updatedAt: new Date().toISOString(),          cols: input.cols,          rows: input.rows,          process: null,          unsubscribeData: null,          unsubscribeExit: null,          hasRunningSubprocess: false,        };        this.sessions.set(sessionKey, session);      } else {        this.stopProcess(session);        session.cwd = input.cwd;      }       session.history = "";      await this.persistHistory(input.threadId, input.terminalId, session.history);      this.startSession(session, input, "restarted");      return this.snapshot(session);    });  }   async close(raw: TerminalCloseInput): Promise<void> {    const input = terminalCloseInputSchema.parse(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),        ),      );       if (input.deleteHistory) {        await this.deleteAllHistoryForThread(input.threadId);      }      this.updateSubprocessPollingState();    });  }   dispose(): void {    this.stopSubprocessPolling();    const sessions = [...this.sessions.values()];    this.sessions.clear();    for (const session of sessions) {      this.stopProcess(session);    }    for (const timer of this.persistTimers.values()) {      clearTimeout(timer);    }    this.persistTimers.clear();    this.pendingPersistHistory.clear();    this.threadLocks.clear();    this.persistQueues.clear();  }   private startSession(    session: TerminalSessionState,    input: TerminalOpenInput,    eventType: "started" | "restarted",  ): void {    this.stopProcess(session);     session.status = "starting";    session.cwd = input.cwd;    session.cols = input.cols;    session.rows = input.rows;    session.exitCode = null;    session.exitSignal = null;    session.hasRunningSubprocess = false;    session.updatedAt = new Date().toISOString();     let ptyProcess: PtyProcess | null = null;    let startedShell: string | null = null;    try {      const shellCandidates = resolveShellCandidates(this.shellResolver);      const terminalEnv = createTerminalSpawnEnv(process.env);      let lastSpawnError: unknown = null;       for (const shell of shellCandidates) {        try {          ptyProcess = this.ptyAdapter.spawn({            shell,            cwd: session.cwd,            cols: session.cols,            rows: session.rows,            env: terminalEnv,          });          startedShell = shell;          break;        } catch (error) {          lastSpawnError = error;          if (!isRetryableShellSpawnError(error)) {            throw error;

Lines 1–500 of 975.