diff --git a/apps/server/src/terminal/Manager.test.ts b/apps/server/src/terminal/Manager.test.ts index ed25a0880b47..47d91e4516ec 100644 --- a/apps/server/src/terminal/Manager.test.ts +++ b/apps/server/src/terminal/Manager.test.ts @@ -24,6 +24,7 @@ import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Scope from "effect/Scope"; import * as TestClock from "effect/testing/TestClock"; +import { ChildProcessSpawner } from "effect/unstable/process"; import { expect } from "vite-plus/test"; import * as ProcessRunner from "../processRunner.ts"; @@ -953,6 +954,125 @@ it.layer( }), ); + it.effect("derives subprocess activity for every terminal from one shared process snapshot", () => + Effect.gen(function* () { + const runCalls: Array<{ command: string; args: ReadonlyArray }> = []; + // FakePtyAdapter assigns pids starting at 9000, so the two terminals + // opened below run as pids 9000 and 9001. + const psStdout = [" 100 9000 vim", " 101 100 git", " 200 9001 /usr/bin/python3"].join( + "\n", + ); + const processRunner: ProcessRunner.ProcessRunner["Service"] = { + run: (input) => + Effect.sync(() => { + runCalls.push({ command: input.command, args: input.args }); + return { + stdout: psStdout, + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }; + }), + }; + + const { manager, getEvents } = yield* createManager(5, { + subprocessPollIntervalMs: 20, + }).pipe( + Effect.provideService(ProcessRunner.ProcessRunner, processRunner), + Effect.provide(withHostPlatform("linux")), + ); + + yield* manager.open(openInput()); + yield* manager.open(openInput({ threadId: "thread-2" })); + + yield* waitFor( + Effect.map( + getEvents, + (events) => + events.some( + (event) => + event.type === "activity" && + event.hasRunningSubprocess === true && + event.label === "vim", + ) && + events.some( + (event) => + event.type === "activity" && + event.hasRunningSubprocess === true && + event.label === "python3", + ), + ), + "1200 millis", + ); + yield* waitFor( + Effect.sync(() => runCalls.length >= 3), + "1200 millis", + ); + + // Every spawn is the shared table snapshot — no per-terminal `pgrep` + // or per-child `ps -p` invocations. + expect(runCalls.every((call) => call.args.join(" ") === "-eo pid=,ppid=,comm=")).toBe(true); + }), + ); + + it.effect("keeps last known subprocess state when the process snapshot fails", () => + Effect.gen(function* () { + let failSnapshots = false; + let failedCalls = 0; + const processRunner: ProcessRunner.ProcessRunner["Service"] = { + run: () => + Effect.sync(() => { + if (failSnapshots) failedCalls += 1; + return { + stdout: failSnapshots ? "" : " 100 9000 vim", + stderr: "", + code: ChildProcessSpawner.ExitCode(failSnapshots ? 1 : 0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }; + }), + }; + + const { manager, getEvents } = yield* createManager(5, { + subprocessPollIntervalMs: 20, + }).pipe( + Effect.provideService(ProcessRunner.ProcessRunner, processRunner), + Effect.provide(withHostPlatform("linux")), + ); + + yield* manager.open(openInput()); + yield* waitFor( + Effect.map(getEvents, (events) => + events.some( + (event) => + event.type === "activity" && + event.hasRunningSubprocess === true && + event.label === "vim", + ), + ), + "1200 millis", + ); + + failSnapshots = true; + yield* waitFor( + Effect.sync(() => failedCalls >= 3), + "1200 millis", + ); + + // A failed snapshot is not authoritative: no terminal flips to idle. + const activityEvents = (yield* getEvents).filter((event) => event.type === "activity"); + expect(activityEvents.length).toBeGreaterThan(0); + expect(activityEvents.every((event) => event.hasRunningSubprocess === true)).toBe(true); + }), + ); + it.effect("caps persisted history to configured line limit", () => Effect.gen(function* () { const { manager, ptyAdapter } = yield* createManager(3); diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index 6dc9e1892b63..64c2dbb913fb 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -89,12 +89,21 @@ class TerminalSubprocessCheckError extends Schema.TaggedErrorClass detail !== null) + .join(", "); + return `Failed to inspect terminal subprocesses with ${this.command}${details.length > 0 ? ` (${details})` : ""}`; } } @@ -610,125 +619,102 @@ function isRetryableShellSpawnError(error: PtyAdapter.PtySpawnError): boolean { ); } -function parseFirstChildPidFromPgrep(stdout: string): number | null { +interface TerminalProcessTableSnapshot { + readonly childrenByParent: ReadonlyMap>; + readonly commandById: ReadonlyMap; +} + +function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot { + const childrenByParent = new Map(); + const commandById = new Map(); for (const line of stdout.split(/\r?\n/g)) { - const n = Number.parseInt(line.trim(), 10); - if (Number.isInteger(n) && n > 0) { - return n; - } + // `comm=` is the final column and may itself contain spaces, so only the + // first two tokens are structural. + const match = /^\s*(\d+)\s+(\d+)\s+(.+)$/.exec(line); + if (!match) continue; + const pid = Number(match[1]); + const ppid = Number(match[2]); + if (!Number.isInteger(pid) || !Number.isInteger(ppid)) continue; + commandById.set(pid, (match[3] ?? "").trim()); + const children = childrenByParent.get(ppid) ?? []; + children.push(pid); + childrenByParent.set(ppid, children); } - return null; + return { childrenByParent, commandById }; } -function windowsInspectSubprocess( - terminalPid: number, - platform: NodeJS.Platform, -): Effect.Effect< - TerminalSubprocessInspectResult, - TerminalSubprocessCheckError, - ProcessRunner.ProcessRunner -> { - const command = - 'Get-CimInstance Win32_Process -ErrorAction Stop | ForEach-Object { Write-Output "$($_.ProcessId)|$($_.ParentProcessId)|$($_.Name)" }'; - return Effect.gen(function* () { - const processRunner = yield* ProcessRunner.ProcessRunner; - return yield* processRunner.run({ - // powershell.exe is a real executable — never spawn it through cmd.exe - // shell mode, which would re-tokenize the `-Command` payload (pipes, - // semicolons) before PowerShell ever sees it. - command: "powershell.exe", - args: ["-NoProfile", "-NonInteractive", "-Command", command], - timeout: "1500 millis", - maxOutputBytes: 32_768, - outputMode: "truncate", - timeoutBehavior: "timedOutResult", - }); - }).pipe( - Effect.map((result) => { - if (result.code !== 0) { - return { hasRunningSubprocess: false, childCommand: null, processIds: [] } as const; - } - const processNameById = new Map(); - const childrenByParent = new Map(); - for (const line of result.stdout.split(/\r?\n/g)) { - const [pidRaw, parentPidRaw, nameRaw] = line.trim().split("|", 3); - const pid = Number(pidRaw); - const parentPid = Number(parentPidRaw); - if (!Number.isInteger(pid) || !Number.isInteger(parentPid)) continue; - processNameById.set(pid, nameRaw?.trim() ?? ""); - const children = childrenByParent.get(parentPid) ?? []; - children.push(pid); - childrenByParent.set(parentPid, children); - } - const directChildren = childrenByParent.get(terminalPid) ?? []; - const childPid = directChildren[0]; - if (childPid === undefined) { - return { hasRunningSubprocess: false, childCommand: null, processIds: [] } as const; - } - const processIds = new Set([terminalPid]); - const pending = [terminalPid]; - while (pending.length > 0) { - const parentPid = pending.pop(); - if (parentPid === undefined) continue; - for (const pid of childrenByParent.get(parentPid) ?? []) { - if (processIds.has(pid)) continue; - processIds.add(pid); - pending.push(pid); - } - } - const normalized = normalizeChildCommandName(processNameById.get(childPid) ?? "", platform); - return { - hasRunningSubprocess: true, - childCommand: normalized ? truncateTerminalWireLabel(normalized) : null, - processIds: [...processIds], - } as const; - }), - Effect.mapError( - (cause) => - new TerminalSubprocessCheckError({ - cause, - terminalPid, - command: "powershell", - }), - ), - ); +function parseWindowsProcessTable(stdout: string): TerminalProcessTableSnapshot { + const childrenByParent = new Map(); + const commandById = new Map(); + for (const line of stdout.split(/\r?\n/g)) { + const [pidRaw, parentPidRaw, nameRaw] = line.trim().split("|", 3); + const pid = Number(pidRaw); + const parentPid = Number(parentPidRaw); + if (!Number.isInteger(pid) || !Number.isInteger(parentPid)) continue; + commandById.set(pid, nameRaw?.trim() ?? ""); + const children = childrenByParent.get(parentPid) ?? []; + children.push(pid); + childrenByParent.set(parentPid, children); + } + return { childrenByParent, commandById }; } -const posixInspectSubprocess = Effect.fn("terminal.posixInspectSubprocess")(function* ( +function deriveSubprocessInspectResult( + snapshot: TerminalProcessTableSnapshot, terminalPid: number, platform: NodeJS.Platform, +): TerminalSubprocessInspectResult { + const childPid = (snapshot.childrenByParent.get(terminalPid) ?? [])[0]; + if (childPid === undefined) { + return { hasRunningSubprocess: false, childCommand: null, processIds: [] }; + } + const processIds = new Set([terminalPid]); + const pending = [terminalPid]; + while (pending.length > 0) { + const parentPid = pending.pop(); + if (parentPid === undefined) continue; + for (const pid of snapshot.childrenByParent.get(parentPid) ?? []) { + if (processIds.has(pid)) continue; + processIds.add(pid); + pending.push(pid); + } + } + const normalized = normalizeChildCommandName(snapshot.commandById.get(childPid) ?? "", platform); + return { + hasRunningSubprocess: true, + childCommand: normalized ? truncateTerminalWireLabel(normalized) : null, + processIds: [...processIds], + }; +} + +const POSIX_PS_ABSOLUTE_PATHS = ["/bin/ps", "/usr/bin/ps"] as const; + +// Resolve `ps` to an absolute path once at startup. Spawning by bare name +// walks every PATH entry per spawn (one failed posix_spawn per directory +// until the hit), which is measurable at a 1s poll cadence on long PATHs. +const resolvePosixPsCommand = Effect.fn("terminal.resolvePosixPsCommand")(function* () { + const fileSystem = yield* FileSystem.FileSystem; + for (const candidate of POSIX_PS_ABSOLUTE_PATHS) { + const exists = yield* fileSystem.exists(candidate).pipe(Effect.orElseSucceed(() => false)); + if (exists) return candidate; + } + return "ps"; +}); + +const posixProcessTableSnapshot = Effect.fn("terminal.posixProcessTableSnapshot")(function* ( + psCommand: string, ): Effect.fn.Return< - TerminalSubprocessInspectResult, + TerminalProcessTableSnapshot, TerminalSubprocessCheckError, ProcessRunner.ProcessRunner > { const processRunner = yield* ProcessRunner.ProcessRunner; - const runPgrep = processRunner - .run({ - command: "pgrep", - args: ["-P", String(terminalPid)], - timeout: "1 second", - maxOutputBytes: 32_768, - outputMode: "truncate", - timeoutBehavior: "timedOutResult", - }) - .pipe( - Effect.mapError( - (cause) => - new TerminalSubprocessCheckError({ - cause, - terminalPid, - command: "pgrep", - }), - ), - ); - - const runPs = processRunner + const result = yield* processRunner .run({ - command: "ps", - args: ["-eo", "pid=,ppid="], + command: psCommand, + args: ["-eo", "pid=,ppid=,comm="], timeout: "1 second", - maxOutputBytes: 262_144, + maxOutputBytes: 524_288, outputMode: "truncate", timeoutBehavior: "timedOutResult", }) @@ -737,120 +723,66 @@ const posixInspectSubprocess = Effect.fn("terminal.posixInspectSubprocess")(func (cause) => new TerminalSubprocessCheckError({ cause, - terminalPid, command: "ps", }), ), ); - - let childPid: number | null = null; - - const pgrepResult = yield* Effect.exit(runPgrep); - if (pgrepResult._tag === "Success") { - if (pgrepResult.value.code === 0) { - childPid = parseFirstChildPidFromPgrep(pgrepResult.value.stdout); - } else if (pgrepResult.value.code === 1) { - return { hasRunningSubprocess: false, childCommand: null, processIds: [] }; - } - } - - if (childPid === null) { - const psResult = yield* Effect.exit(runPs); - if (psResult._tag === "Failure" || psResult.value.code !== 0) { - return { hasRunningSubprocess: false, childCommand: null, processIds: [] }; - } - for (const line of psResult.value.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) { - childPid = pid; - break; - } - } - } - - if (childPid === null) { - return { hasRunningSubprocess: false, childCommand: null, processIds: [] }; - } - - const runComm = processRunner.run({ - command: "ps", - args: ["-p", String(childPid), "-o", "comm="], - timeout: "1 second", - maxOutputBytes: 8_192, - outputMode: "truncate", - timeoutBehavior: "timedOutResult", - }); - - const commResult = yield* Effect.exit(runComm); - let rawComm: string | null = null; - if (commResult._tag === "Success" && commResult.value && commResult.value.code === 0) { - rawComm = commResult.value.stdout.trim(); - } - - if (!rawComm || rawComm.length === 0) { - const runArgs = processRunner.run({ + if (result.code !== 0 || result.timedOut || result.stdoutTruncated) { + // Not authoritative: an empty or partial table would mark every terminal + // idle and clear its registered process ids. Failing skips the tick. + return yield* new TerminalSubprocessCheckError({ command: "ps", - args: ["-p", String(childPid), "-o", "args="], - timeout: "1 second", - maxOutputBytes: 16_384, - outputMode: "truncate", - timeoutBehavior: "timedOutResult", + exitCode: result.code, + timedOut: result.timedOut, + stdoutTruncated: result.stdoutTruncated, }); - const argsResult = yield* Effect.exit(runArgs); - if (argsResult._tag === "Success" && argsResult.value && argsResult.value.code === 0) { - const first = argsResult.value.stdout.trim().split(/\s+/)[0] ?? ""; - rawComm = first.length > 0 ? first : null; - } } - - const normalized = rawComm ? normalizeChildCommandName(rawComm, platform) : null; - const processIds = new Set([terminalPid]); - const psResult = yield* Effect.exit(runPs); - if (psResult._tag === "Success" && psResult.value.code === 0) { - const childrenByParent = new Map(); - for (const line of psResult.value.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; - const children = childrenByParent.get(ppid) ?? []; - children.push(pid); - childrenByParent.set(ppid, children); - } - const pending = [terminalPid]; - while (pending.length > 0) { - const parentPid = pending.pop(); - if (parentPid === undefined) continue; - for (const child of childrenByParent.get(parentPid) ?? []) { - if (processIds.has(child)) continue; - processIds.add(child); - pending.push(child); - } - } - } else { - processIds.add(childPid); - } - return { - hasRunningSubprocess: true, - childCommand: normalized ? truncateTerminalWireLabel(normalized) : null, - processIds: [...processIds], - }; + return parsePosixProcessTable(result.stdout); }); -function defaultSubprocessInspectorForPlatform(platform: NodeJS.Platform) { - return Effect.fn("terminal.defaultSubprocessInspector")(function* (terminalPid: number) { - if (!Number.isInteger(terminalPid) || terminalPid <= 0) { - return { hasRunningSubprocess: false, childCommand: null, processIds: [] }; - } - if (platform === "win32") { - return yield* windowsInspectSubprocess(terminalPid, platform); +const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnapshot")( + function* (): Effect.fn.Return< + TerminalProcessTableSnapshot, + TerminalSubprocessCheckError, + ProcessRunner.ProcessRunner + > { + const command = + 'Get-CimInstance Win32_Process -ErrorAction Stop | ForEach-Object { Write-Output "$($_.ProcessId)|$($_.ParentProcessId)|$($_.Name)" }'; + const processRunner = yield* ProcessRunner.ProcessRunner; + const result = yield* processRunner + .run({ + // powershell.exe is a real executable — never spawn it through cmd.exe + // shell mode, which would re-tokenize the `-Command` payload (pipes, + // semicolons) before PowerShell ever sees it. + command: "powershell.exe", + args: ["-NoProfile", "-NonInteractive", "-Command", command], + timeout: "1500 millis", + maxOutputBytes: 262_144, + outputMode: "truncate", + timeoutBehavior: "timedOutResult", + }) + .pipe( + Effect.mapError( + (cause) => + new TerminalSubprocessCheckError({ + cause, + command: "powershell", + }), + ), + ); + if (result.code !== 0 || result.timedOut || result.stdoutTruncated) { + // Not authoritative: an empty or partial table would mark every terminal + // idle and clear its registered process ids. Failing skips the tick. + return yield* new TerminalSubprocessCheckError({ + command: "powershell", + exitCode: result.code, + timedOut: result.timedOut, + stdoutTruncated: result.stdoutTruncated, + }); } - return yield* posixInspectSubprocess(terminalPid, platform); - }); -} + return parseWindowsProcessTable(result.stdout); + }, +); function capHistory(history: string, maxLines: number): string { if (history.length === 0) return history; @@ -1227,12 +1159,27 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func const baseEnv = options.env ?? process.env; const shellResolver = options.shellResolver ?? (() => defaultShellResolver(platform, baseEnv)); const processRunner = yield* ProcessRunner.ProcessRunner; - const subprocessInspector = - options.subprocessInspector ?? - ((terminalPid) => - defaultSubprocessInspectorForPlatform(platform)(terminalPid).pipe( - Effect.provideService(ProcessRunner.ProcessRunner, processRunner), - )); + // One process-table snapshot per poll tick, shared across every terminal. + // Per-terminal `pgrep`/`ps` calls multiply spawn load by terminal count and + // can exhaust the PID space on hosts with many sessions (#6332). + const fetchProcessTableSnapshot = ( + platform === "win32" + ? windowsProcessTableSnapshot() + : posixProcessTableSnapshot(yield* resolvePosixPsCommand()) + ).pipe(Effect.provideService(ProcessRunner.ProcessRunner, processRunner)); + const customSubprocessInspector = options.subprocessInspector; + const acquireSubprocessInspector: Effect.Effect< + TerminalSubprocessInspector, + TerminalSubprocessCheckError + > = + customSubprocessInspector !== undefined + ? Effect.succeed(customSubprocessInspector) + : Effect.map( + fetchProcessTableSnapshot, + (snapshot): TerminalSubprocessInspector => + (terminalPid) => + Effect.succeed(deriveSubprocessInspectResult(snapshot, terminalPid, platform)), + ); const subprocessPollIntervalMs = options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS; const processKillGraceMs = options.processKillGraceMs ?? DEFAULT_PROCESS_KILL_GRACE_MS; @@ -2064,6 +2011,21 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func return; } + const inspectorOption = yield* acquireSubprocessInspector.pipe( + Effect.map(Option.some), + Effect.catch((reason) => + Effect.logWarning("failed to snapshot processes for terminal subprocess polling", { + reason, + }).pipe(Effect.as(Option.none())), + ), + ); + + if (Option.isNone(inspectorOption)) { + return; + } + + const subprocessInspector = inspectorOption.value; + const checkSubprocessActivity = Effect.fn("terminal.checkSubprocessActivity")(function* ( session: TerminalSessionState & { pid: number }, ) {