diff --git a/packages/docker-workspace-runtime/src/index.ts b/packages/docker-workspace-runtime/src/index.ts index 8c2756463e..08d100b409 100644 --- a/packages/docker-workspace-runtime/src/index.ts +++ b/packages/docker-workspace-runtime/src/index.ts @@ -3,8 +3,11 @@ import { spawn } from "node:child_process"; import { createHash } from "node:crypto"; import { realpath } from "node:fs/promises"; +import { request as httpRequest } from "node:http"; import { Socket } from "node:net"; import path from "node:path"; +import { createInterface } from "node:readline"; +import type { Duplex } from "node:stream"; interface RuntimeState { workspaceId: string; @@ -29,6 +32,7 @@ interface LifecycleRequest { } interface ExecRequest { + type: "spawn"; protocolVersion: 1; argv: [string, ...string[]]; cwd?: string; @@ -44,14 +48,22 @@ interface ExecRequest { | { kind: "archive" }; options: { image?: unknown }; execId: string; + stdio: { kind: "pipes" } | { kind: "pty"; rows: number; cols: number; term?: string }; } +type PtyControl = + | ExecRequest + | { type: "resize"; protocolVersion: 1; id: number; rows: number; cols: number } + | { type: "signal"; protocolVersion: 1; signal: NodeJS.Signals }; + +const PTY_CLEANUP_TIMEOUT_MS = 1_000; + const [operation] = process.argv.slice(2); const requestedWorkspaceId = argument("--workspace-id"); try { if (operation === "describe") { - writeJson({ protocolVersion: 1, modes: ["pipes"] }); + writeJson({ protocolVersion: 1, modes: ["pipes", "pty"] }); } else if (!requestedWorkspaceId) { throw new Error("--workspace-id is required"); } else if (operation === "exec") { @@ -278,7 +290,12 @@ async function inspectVolume(name: string): Promise<{ owner: string | null } | n } async function execute(workspaceId: string): Promise { - const request = JSON.parse(await readOneShotEnvelope()) as ExecRequest; + const control = new Socket({ fd: 3, readable: true, writable: false }); + const lines = createInterface({ input: control, crlfDelay: Infinity }); + const iterator = lines[Symbol.asyncIterator](); + const first = await iterator.next(); + if (first.done) throw new Error("Runtime exec spawn control is required"); + const request = JSON.parse(first.value) as ExecRequest; assertProtocol(request.protocolVersion); if (!/^[a-f0-9]{32}$/.test(request.execId)) throw new Error("Invalid exec id"); const current = await inspect(workspaceId); @@ -296,11 +313,13 @@ async function execute(workspaceId: string): Promise { ]) ).trim(); assertContainedPath(current.state.root, cwd); - const encoded = Buffer.from( - JSON.stringify({ argv: request.argv, cwd, env: request.env }), - ).toString("base64url"); + const encoded = Buffer.from(JSON.stringify({ ...request, cwd })).toString("base64url"); const execId = request.execId; const container = resourceNames(workspaceId).container; + if (request.stdio.kind === "pty") { + await executeDockerPty({ workspaceId, request, controls: iterator, encoded, container }); + return; + } const child = spawn( "docker", [ @@ -350,6 +369,217 @@ async function execute(workspaceId: string): Promise { process.exitCode = exit.code ?? 1; } +async function executeDockerPty(input: { + workspaceId: string; + request: ExecRequest; + controls: AsyncIterator; + encoded: string; + container: string; +}): Promise { + if (input.request.stdio.kind !== "pty") throw new Error("Docker PTY execution requires PTY mode"); + const events = new Socket({ fd: 4, readable: false, writable: true }); + const created = await dockerApiJson<{ Id: string }>( + "POST", + `/containers/${encodeURIComponent(input.container)}/exec`, + { + AttachStdin: true, + AttachStdout: true, + AttachStderr: true, + Tty: true, + Env: [`PASEO_RUNTIME_EXEC=${input.encoded}`, `PASEO_RUNTIME_EXEC_ID=${input.request.execId}`], + Cmd: ["node", "/opt/paseo-workspace-runtime/workload.mjs"], + }, + ); + const stream = await startDockerExec(created.Id); + await resizeDockerExec(created.Id, input.request.stdio.cols, input.request.stdio.rows); + process.stdin.pipe(stream); + stream.on("data", (data: Buffer) => process.stdout.write(data)); + stream.resume(); + let requestedSignal: NodeJS.Signals | null = null; + let executionFinished = false; + let announceControlFailure!: (error: Error) => void; + const controlFailureStarted = new Promise((resolve) => { + announceControlFailure = resolve; + }); + const controlCleanup = handlePtyControls( + input.workspaceId, + input.request.execId, + created.Id, + input.controls, + (signal) => { + requestedSignal = signal; + }, + (id) => { + writePtyEvent(events, { type: "resized", protocolVersion: 1, id }); + }, + ).then( + () => new Promise(() => {}), + async (error) => { + if (executionFinished) return new Promise(() => {}); + const failure = error instanceof Error ? error : new Error(String(error)); + announceControlFailure(failure); + let cleanupError: Error | null = null; + try { + await withTimeout( + signalExec(input.workspaceId, input.request.execId, "SIGKILL"), + PTY_CLEANUP_TIMEOUT_MS, + "Docker PTY signal helper timed out", + ); + } catch (cleanupFailure) { + cleanupError = + cleanupFailure instanceof Error ? cleanupFailure : new Error(String(cleanupFailure)); + } + stream.destroy(); + return { failure, cleanupError }; + }, + ); + const streamEnded = new Promise((resolve, reject) => { + stream.once("end", resolve); + stream.once("error", reject); + }); + const firstOutcome = await Promise.race([ + streamEnded.then(() => "stream" as const), + controlFailureStarted.then(() => "control" as const), + ]); + const outcome = + firstOutcome === "control" ? await controlCleanup : { failure: null, cleanupError: null }; + executionFinished = true; + process.stdin.unpipe(stream); + writePtyEvent(events, { type: "eof", protocolVersion: 1 }); + if (outcome.failure) { + const cleanupDetail = outcome.cleanupError + ? `; cleanup failed: ${outcome.cleanupError.message}` + : ""; + writePtyEvent(events, { + type: "error", + protocolVersion: 1, + message: `${outcome.failure.message}${cleanupDetail}`, + }); + events.end(); + return; + } + const inspection = await dockerApiJson<{ ExitCode: number }>( + "GET", + `/exec/${encodeURIComponent(created.Id)}/json`, + ); + writePtyEvent(events, { + type: "exit", + protocolVersion: 1, + code: requestedSignal ? null : inspection.ExitCode, + signal: requestedSignal, + }); + events.end(); +} + +async function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { + let timeout: ReturnType | null = null; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new Error(message)), timeoutMs); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +} + +async function handlePtyControls( + workspaceId: string, + execId: string, + dockerExecId: string, + controls: AsyncIterator, + onSignal: (signal: NodeJS.Signals) => void, + onResize: (id: number) => void, +): Promise { + while (true) { + const next = await controls.next(); + if (next.done) return; + const control = JSON.parse(next.value) as PtyControl; + assertProtocol(control.protocolVersion); + if (control.type === "resize") { + await resizeDockerExec(dockerExecId, control.cols, control.rows); + onResize(control.id); + } else if (control.type === "signal") { + const signal = requireSignal(control.signal); + onSignal(signal); + await signalExec(workspaceId, execId, signal); + } else { + throw new Error(`Unexpected PTY control: ${control.type}`); + } + } +} + +async function resizeDockerExec(execId: string, cols: number, rows: number): Promise { + if (!Number.isInteger(cols) || cols < 1 || !Number.isInteger(rows) || rows < 1) { + throw new Error("Invalid PTY size"); + } + await dockerApiJson("POST", `/exec/${encodeURIComponent(execId)}/resize?h=${rows}&w=${cols}`); +} + +function writePtyEvent(stream: Socket, event: unknown): void { + stream.write(`${JSON.stringify(event)}\n`); +} + +async function dockerApiJson>( + method: "GET" | "POST", + apiPath: string, + body?: unknown, +): Promise { + const payload = body === undefined ? "" : JSON.stringify(body); + return new Promise((resolve, reject) => { + const request = httpRequest( + { + socketPath: "/var/run/docker.sock", + method, + path: apiPath, + headers: payload + ? { "Content-Type": "application/json", "Content-Length": Buffer.byteLength(payload) } + : undefined, + }, + (response) => { + void readStream(response).then((output) => { + if ((response.statusCode ?? 500) >= 300) { + return reject(new Error(`Docker API ${method} ${apiPath} failed: ${output.trim()}`)); + } + return resolve((output ? JSON.parse(output) : {}) as T); + }, reject); + }, + ); + request.once("error", reject); + request.end(payload); + }); +} + +async function startDockerExec(execId: string): Promise { + const payload = JSON.stringify({ Detach: false, Tty: true }); + return new Promise((resolve, reject) => { + const request = httpRequest({ + socketPath: "/var/run/docker.sock", + method: "POST", + path: `/exec/${encodeURIComponent(execId)}/start`, + headers: { + "Content-Type": "application/json", + "Content-Length": Buffer.byteLength(payload), + Connection: "Upgrade", + Upgrade: "tcp", + }, + }); + request.once("upgrade", (_response, socket, head) => { + if (head.length > 0) socket.unshift(head); + resolve(socket); + }); + request.once("response", (response) => { + void readStream(response).then((output) => { + return reject(new Error(`Docker exec attach failed: ${output.trim()}`)); + }, reject); + }); + request.once("error", reject); + request.end(payload); + }); +} + async function signalExec( workspaceId: string, execId: string, @@ -455,15 +685,6 @@ async function readStdin(): Promise { return readStream(process.stdin); } -async function readOneShotEnvelope(): Promise { - const stream = new Socket({ fd: 3, readable: true, writable: false }); - try { - return await readStream(stream); - } finally { - stream.destroy(); - } -} - async function readStream(stream: NodeJS.ReadableStream): Promise { const chunks: Buffer[] = []; for await (const chunk of stream) diff --git a/packages/docker-workspace-runtime/src/workload.mjs b/packages/docker-workspace-runtime/src/workload.mjs index 6694f29f21..9201d3099d 100644 --- a/packages/docker-workspace-runtime/src/workload.mjs +++ b/packages/docker-workspace-runtime/src/workload.mjs @@ -1,4 +1,4 @@ -import { spawn } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import { rmSync, writeFileSync } from "node:fs"; const encoded = process.env.PASEO_RUNTIME_EXEC; @@ -10,10 +10,19 @@ delete process.env.PASEO_RUNTIME_EXEC_ID; if (!execId || !/^[a-f0-9]+$/.test(execId)) throw new Error("PASEO_RUNTIME_EXEC_ID is required"); const pidFile = `/tmp/paseo-runtime-exec-${execId}.pid`; +if (request.stdio?.kind === "pty") { + const resized = spawnSync( + "stty", + ["cols", String(request.stdio.cols), "rows", String(request.stdio.rows)], + { stdio: ["inherit", "inherit", "inherit"] }, + ); + if (resized.status !== 0) throw new Error("Runtime PTY initial resize failed"); +} + const child = spawn(request.argv[0], request.argv.slice(1), { cwd: request.cwd, env: request.env, - detached: true, + detached: request.stdio?.kind !== "pty", stdio: ["inherit", "inherit", "inherit"], }); if (!child.pid) throw new Error("Runtime workload did not start"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index ea775bbf7c..ebc090dda7 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -277,6 +277,10 @@ export interface AgentManagerOptions { agentStreamCoalesceWindowMs?: number; rescueTimeouts?: AgentManagerRescueTimeouts; logger: Logger; + resolveWorkspaceExecution?: ( + workspaceId: string, + cwd: string, + ) => Promise; } export interface WaitForAgentOptions { @@ -642,6 +646,7 @@ export class AgentManager { private onWorkspaceStateMayHaveChanged?: (params: { cwd: string }) => void; private logger: Logger; private readonly rescueTimeouts: Required; + private readonly resolveWorkspaceExecution?: AgentManagerOptions["resolveWorkspaceExecution"]; private acceptingAgentRegistrations = true; constructor(options: AgentManagerOptions) { @@ -655,6 +660,7 @@ export class AgentManager { this.configurePaseoTools(options); this.appendSystemPrompt = options.appendSystemPrompt ?? ""; this.logger = options.logger.child({ module: "agent", component: "agent-manager" }); + this.resolveWorkspaceExecution = options.resolveWorkspaceExecution; this.rescueTimeouts = { reloadSessionCloseMs: options.rescueTimeouts?.reloadSessionCloseMs ?? RELOAD_SESSION_CLOSE_TIMEOUT_MS, @@ -1100,6 +1106,7 @@ export class AgentManager { client, storedConfig.cwd, options?.env, + options.workspaceId, ); const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext); const createOptions = this.buildCreateSessionOptions(options); @@ -1179,7 +1186,13 @@ export class AgentManager { `Provider '${handle.provider}' is not available. Please ensure the CLI is installed.`, ); } - const launchContext = await this.buildLaunchContext(resolvedAgentId, client, storedConfig.cwd); + const launchContext = await this.buildLaunchContext( + resolvedAgentId, + client, + storedConfig.cwd, + undefined, + options?.workspaceId, + ); const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext); const session = await client.resumeSession( handle, @@ -1227,7 +1240,13 @@ export class AgentManager { }, resolvedAgentId, ); - const launchContext = await this.buildLaunchContext(resolvedAgentId, client, storedConfig.cwd); + const launchContext = await this.buildLaunchContext( + resolvedAgentId, + client, + storedConfig.cwd, + undefined, + input.workspaceId, + ); const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext); const imported = await client.importSession( { @@ -1308,7 +1327,13 @@ export class AgentManager { provider, } as AgentSessionConfig; const { storedConfig, launchConfig } = await this.prepareSessionConfig(refreshConfig, agentId); - const launchContext = await this.buildLaunchContext(agentId, client, storedConfig.cwd); + const launchContext = await this.buildLaunchContext( + agentId, + client, + storedConfig.cwd, + undefined, + existing.workspaceId, + ); const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext); const session = handle @@ -4484,6 +4509,7 @@ export class AgentManager { client: AgentClient, cwd: string, env?: Record, + workspaceId?: string, ): Promise { const context: AgentLaunchContext = { agentId, @@ -4493,6 +4519,9 @@ export class AgentManager { PASEO_AGENT_CWD: cwd, }, }; + if (workspaceId && this.resolveWorkspaceExecution) { + context.workspaceExecution = await this.resolveWorkspaceExecution(workspaceId, cwd); + } if ( this.paseoToolsEnabled && client.capabilities.supportsNativePaseoTools && diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index c2c5855255..7add5cd02c 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -5,6 +5,7 @@ import type { } from "@getpaseo/protocol/agent-types"; import type { AgentAttachment } from "@getpaseo/protocol/messages"; import type { PaseoToolCatalog } from "./tools/types.js"; +import type { WorkspaceProcess, WorkspaceProcessInput } from "../workspace-runtime/index.js"; export type { AgentProviderNotice }; @@ -597,6 +598,9 @@ export interface AgentLaunchContext { * AgentSessionConfig; providers may adapt it to their native tool surface. */ paseoTools?: PaseoToolCatalog; + workspaceExecution?: { + run(input: Omit): Promise; + }; } export interface AgentCreateSessionOptions { diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 2296aa3960..09ca53ac28 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -96,6 +96,7 @@ import { type ToolCallDetail, type ToolCallTimelineItem, } from "../agent-sdk-types.js"; +import type { WorkspaceProcess } from "../../workspace-runtime/index.js"; import { isDefaultAgentCreateConfigUnattended, resolveDefaultAgentCreateConfig, @@ -103,6 +104,7 @@ import { import { importSessionFromPersistence } from "../provider-session-import.js"; import { checkProviderLaunchAvailable, + createProviderEnv, createProviderEnvSpec, resolveProviderLaunch, type ProviderRuntimeSettings, @@ -461,6 +463,7 @@ interface ACPAgentSessionOptions { handle?: AgentPersistenceHandle; agentId?: string; launchEnv?: Record; + workspaceExecution?: AgentLaunchContext["workspaceExecution"]; waitForInitialCommands?: boolean; initialCommandsWaitTimeoutMs?: number; terminateProcess?: ProcessTerminator; @@ -518,7 +521,7 @@ interface TerminalExit { interface TerminalEntry { id: string; - child: ChildProcess; + terminate(): Promise; output: string; truncated: boolean; outputByteLimit: number | null; @@ -528,6 +531,29 @@ interface TerminalEntry { rejectExit: (error: Error) => void; } +async function terminateWorkspaceProcess(process: WorkspaceProcess): Promise { + process.kill("SIGTERM"); + if (await exitsWithin(process.exited, 2_000)) return; + process.kill("SIGKILL"); + await process.exited; +} + +async function exitsWithin(exit: Promise, timeoutMs: number): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), timeoutMs); + }); + const result = await Promise.race([exit.then(() => true as const), timeout]); + if (timer) clearTimeout(timer); + return result; +} + +function toExactEnvironment(env: NodeJS.ProcessEnv): Record { + return Object.fromEntries( + Object.entries(env).filter((entry): entry is [string, string] => entry[1] !== undefined), + ); +} + export interface ConfigOptionSelector { id: string; label: string; @@ -875,6 +901,7 @@ export class ACPAgentClient implements AgentClient { capabilities: this.capabilities, agentId: launchContext?.agentId, launchEnv: launchContext?.env, + workspaceExecution: launchContext?.workspaceExecution, extensionCommandsParser: this.extensionCommandsParser, waitForInitialCommands: this.waitForInitialCommands, initialCommandsWaitTimeoutMs: this.initialCommandsWaitTimeoutMs, @@ -926,6 +953,7 @@ export class ACPAgentClient implements AgentClient { handle, agentId: launchContext?.agentId, launchEnv: launchContext?.env, + workspaceExecution: launchContext?.workspaceExecution, extensionCommandsParser: this.extensionCommandsParser, waitForInitialCommands: this.waitForInitialCommands, initialCommandsWaitTimeoutMs: this.initialCommandsWaitTimeoutMs, @@ -1405,6 +1433,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { ) => Promise; private readonly agentId?: string; private readonly launchEnv?: Record; + private readonly workspaceExecution?: AgentLaunchContext["workspaceExecution"]; private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); private readonly pendingPermissions = new Map(); private pendingUserMessage: PendingUserMessage | null = null; @@ -1464,6 +1493,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.availableModes = options.defaultModes; this.agentId = options.agentId; this.launchEnv = options.launchEnv; + this.workspaceExecution = options.workspaceExecution; this.initialHandle = options.handle; this.config = { ...config, provider: options.provider }; this.currentMode = config.modeId ?? null; @@ -2191,10 +2221,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { } const terminalTerminations = Array.from(this.terminalEntries.values(), (terminal) => - this.terminateProcess(terminal.child, { - gracefulTimeoutMs: 2_000, - forceTimeoutMs: 2_000, - }), + terminal.terminate(), ); await Promise.all(terminalTerminations); this.terminalEntries.clear(); @@ -2369,14 +2396,10 @@ export class ACPAgentSession implements AgentSession, ACPClient { const terminalCommand = resolveTerminalCommand(params.command, params.args); const commandEnvOverlays = terminalCommand.shell === false ? [env, createStringCommandShellEnvOverlay()] : [env]; - const child = spawnProcess(terminalCommand.command, terminalCommand.args, { - cwd: params.cwd ?? this.config.cwd, - ...createProviderEnvSpec({ - runtimeSettings: this.runtimeSettings, - overlays: commandEnvOverlays, - }), - shell: terminalCommand.shell, - stdio: ["ignore", "pipe", "pipe"], + const cwd = params.cwd ?? this.config.cwd; + const envSpec = createProviderEnvSpec({ + runtimeSettings: this.runtimeSettings, + overlays: commandEnvOverlays, }); let resolveExit!: (exit: TerminalExit) => void; @@ -2387,9 +2410,8 @@ export class ACPAgentSession implements AgentSession, ACPClient { }); waitForExit.catch(() => undefined); - const entry: TerminalEntry = { + const entryBase = { id: terminalId, - child, output: "", truncated: false, outputByteLimit: params.outputByteLimit ?? null, @@ -2399,6 +2421,59 @@ export class ACPAgentSession implements AgentSession, ACPClient { rejectExit, }; + if (this.workspaceExecution) { + const relativeCwd = path.relative(this.config.cwd, cwd); + if (relativeCwd.startsWith("..") || path.isAbsolute(relativeCwd)) { + throw new Error(`ACP terminal cwd escapes workspace: ${cwd}`); + } + const runtimeProcess = await this.workspaceExecution.run({ + cwd: relativeCwd || undefined, + argv: [terminalCommand.command, ...terminalCommand.args], + env: toExactEnvironment( + createProviderEnv({ + runtimeSettings: this.runtimeSettings, + overlays: commandEnvOverlays, + }), + ), + purpose: { kind: "terminal", terminalId }, + }); + const entry: TerminalEntry = { + ...entryBase, + terminate: () => terminateWorkspaceProcess(runtimeProcess), + }; + runtimeProcess.stdout.on("data", (chunk: Buffer | string) => + appendTerminalOutput(entry, chunk.toString()), + ); + runtimeProcess.stderr.on("data", (chunk: Buffer | string) => + appendTerminalOutput(entry, chunk.toString()), + ); + void runtimeProcess.exited.then(({ code, signal }) => { + const exit = { exitCode: code, signal }; + entry.exit = exit; + resolveExit(exit); + return undefined; + }, rejectExit); + runtimeProcess.stdin.end(); + this.terminalEntries.set(terminalId, entry); + return { terminalId }; + } + + const child = spawnProcess(terminalCommand.command, terminalCommand.args, { + cwd, + ...envSpec, + shell: terminalCommand.shell, + stdio: ["ignore", "pipe", "pipe"], + }); + const entry: TerminalEntry = { + ...entryBase, + terminate: async () => { + await this.terminateProcess(child, { + gracefulTimeoutMs: 2_000, + forceTimeoutMs: 2_000, + }); + }, + }; + child.stdout!.on("data", (chunk: Buffer | string) => appendTerminalOutput(entry, chunk.toString()), ); @@ -2437,7 +2512,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { async releaseTerminal(params: { sessionId: string; terminalId: string }): Promise { const entry = this.getTerminalEntry(params.terminalId); if (!entry.exit) { - await this.terminateProcess(entry.child, { gracefulTimeoutMs: 2_000, forceTimeoutMs: 2_000 }); + await entry.terminate(); } this.terminalEntries.delete(params.terminalId); } @@ -2445,7 +2520,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { async killTerminal(params: KillTerminalRequest): Promise> { const entry = this.getTerminalEntry(params.terminalId); if (!entry.exit) { - await this.terminateProcess(entry.child, { gracefulTimeoutMs: 2_000, forceTimeoutMs: 2_000 }); + await entry.terminate(); } return {}; } diff --git a/packages/server/src/server/agent/providers/acp-workspace-runtime.posix.test.ts b/packages/server/src/server/agent/providers/acp-workspace-runtime.posix.test.ts new file mode 100644 index 0000000000..41468e2f0d --- /dev/null +++ b/packages/server/src/server/agent/providers/acp-workspace-runtime.posix.test.ts @@ -0,0 +1,76 @@ +import { mkdir, mkdtemp, realpath, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { describe, expect, test } from "vitest"; + +import { createWorkspaceRuntimeService } from "../../workspace-runtime/index.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { ACPAgentSession } from "./acp-agent.js"; + +const posixDescribe = describe.runIf(process.platform !== "win32"); + +posixDescribe("ACP workspace terminal execution", () => { + test("uses the selected workspace runtime for ACP-created terminal commands", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-acp-runtime-")); + const cwd = path.join(root, "workspace"); + await mkdir(cwd); + const runtimeIds = new Map(); + const runtime = createWorkspaceRuntimeService({ + paseoHome: path.join(root, "home"), + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await runtime.create({ + workspaceId: "acp-workspace", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: cwd } }, + placement: { kind: "existing" }, + }); + const session = new ACPAgentSession( + { provider: "acp-test", cwd }, + { + provider: "acp-test", + logger: createTestLogger(), + defaultCommand: ["unused"], + defaultModes: [], + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: false, + supportsImages: false, + supportsFileAttachments: false, + supportsAudioAttachments: false, + supportsModes: false, + supportsModels: false, + supportsThinking: false, + supportsMcp: false, + supportsSlashCommands: false, + }, + workspaceExecution: { + run: (input) => runtime.run({ workspaceId: "acp-workspace", ...input }), + }, + }, + ); + + try { + const terminal = await session.createTerminal({ + sessionId: "session", + command: process.execPath, + args: ["-e", "process.stdout.write(`${process.cwd()}|λ`);process.exit(12)"], + cwd, + }); + await expect( + session.waitForTerminalExit({ sessionId: "session", terminalId: terminal.terminalId }), + ).resolves.toEqual({ exitCode: 12, signal: null }); + await expect( + session.terminalOutput({ sessionId: "session", terminalId: terminal.terminalId }), + ).resolves.toMatchObject({ output: `${await realpath(cwd)}|λ` }); + } finally { + await session.close(); + await runtime.destroy("acp-workspace"); + await rm(root, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 486a4a6c43..38cce0e84f 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -145,7 +145,10 @@ import { resolveSelectedWorkspaceRuntimeId, type WorkspaceArchiveContext, } from "./workspace-registry.js"; -import { createWorkspaceRuntimeService } from "./workspace-runtime/index.js"; +import { + createWorkspaceRuntimeService, + type WorkspaceRuntimeService, +} from "./workspace-runtime/index.js"; import { FileBackedChatService } from "./chat/chat-service.js"; import { CheckoutDiffManager } from "./checkout-diff-manager.js"; import { LoopService } from "./loop-service.js"; @@ -605,8 +608,29 @@ export async function createPaseoDaemon( app.set("trust proxy", resolveExpressTrustProxySetting(config)); let boundListenTarget: ListenTarget | null = null; let workspaceRegistry: FileBackedWorkspaceRegistry | null = null; + let workspaceRuntime: WorkspaceRuntimeService | null = null; const terminalManager = createConfiguredTerminalManager({ getTerminalActivityUrl: () => createTerminalActivityUrl(boundListenTarget), + launchPty: async (input) => { + const workspace = await workspaceRegistry?.get(input.workspaceId); + if (!workspace) throw new Error(`Workspace not found: ${input.workspaceId}`); + if (!resolveSelectedWorkspaceRuntimeId(workspace)) return null; + if (!workspaceRuntime) throw new Error("Workspace runtime is not available"); + const relativeCwd = path.relative(workspace.cwd, input.cwd); + if (relativeCwd.startsWith("..") || path.isAbsolute(relativeCwd)) { + throw new Error(`Terminal cwd escapes workspace: ${input.cwd}`); + } + return workspaceRuntime.openTerminal({ + workspaceId: input.workspaceId, + cwd: relativeCwd || undefined, + argv: input.argv, + env: input.env, + purpose: { kind: "terminal", terminalId: input.terminalId }, + rows: input.rows, + cols: input.cols, + term: input.term, + }); + }, }); applyTerminalAgentHookSetting({ store: daemonConfigStore, logger }); @@ -810,7 +834,7 @@ export async function createPaseoDaemon( path.join(config.paseoHome, "projects", "workspaces.json"), logger, ); - const workspaceRuntime = createWorkspaceRuntimeService({ + workspaceRuntime = createWorkspaceRuntimeService({ paseoHome: config.paseoHome, worktreesRoot: config.worktreesRoot, externalRuntimes: config.workspaceRuntimes, @@ -868,6 +892,15 @@ export async function createPaseoDaemon( workspaceGitService.onWorkspaceStateMayHaveChanged(cwd); }, mcpAuthToken: agentMcpAuthToken, + resolveWorkspaceExecution: async (workspaceId) => { + const workspace = await workspaceRegistry?.get(workspaceId); + if (!workspace || !resolveSelectedWorkspaceRuntimeId(workspace)) return undefined; + const runtime = workspaceRuntime; + if (!runtime) throw new Error("Workspace runtime is not available"); + return { + run: (input) => runtime.run({ workspaceId, ...input }), + }; + }, logger, }); @@ -1665,7 +1698,7 @@ export async function createPaseoDaemon( detachAgentStoragePersistence(); await agentStorage.flush().catch(() => undefined); await providerSnapshotManager.shutdown(); - terminalManager.killAll(); + await terminalManager.killAll(); speechService.stop(); await scheduleService.stop().catch(() => undefined); await relayRuntime?.stop().catch(() => undefined); diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 8f0fe6c04a..6f07b4131a 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -105,7 +105,7 @@ afterEach(async () => { while (terminalManagers.length > 0) { const manager = terminalManagers.pop(); if (manager) { - manager.killAll(); + await manager.killAll(); } } await flushTerminalContributionWork(); diff --git a/packages/server/src/server/terminal-activity-route.test.ts b/packages/server/src/server/terminal-activity-route.test.ts index afc97d77e8..3cf121017e 100644 --- a/packages/server/src/server/terminal-activity-route.test.ts +++ b/packages/server/src/server/terminal-activity-route.test.ts @@ -71,7 +71,7 @@ afterEach(async () => { for (const terminal of terminalsByCwd.flat()) { await manager.killTerminalAndWait(terminal.id); } - manager.killAll(); + await manager.killAll(); manager = null; } while (temporaryDirs.length > 0) { diff --git a/packages/server/src/server/test-utils/fixtures/workspace-runtime-command-fixture.mjs b/packages/server/src/server/test-utils/fixtures/workspace-runtime-command-fixture.mjs index 93c243f5df..5ecd2582e5 100644 --- a/packages/server/src/server/test-utils/fixtures/workspace-runtime-command-fixture.mjs +++ b/packages/server/src/server/test-utils/fixtures/workspace-runtime-command-fixture.mjs @@ -5,13 +5,24 @@ import { createHash } from "node:crypto"; import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; import path from "node:path"; import { Socket } from "node:net"; +import { createInterface } from "node:readline"; +import * as pty from "node-pty"; -const [operation] = process.argv.slice(2); +const operation = process.argv + .slice(2) + .find((value) => + ["describe", "create", "inspect", "exec", "signal", "pause", "resume", "destroy"].includes( + value, + ), + ); const workspaceId = argument("--workspace-id"); try { if (operation === "describe") { - writeJson({ protocolVersion: 1, modes: ["pipes"] }); + writeJson({ + protocolVersion: 1, + modes: argument("--modes") === "pipes" ? ["pipes"] : ["pipes", "pty"], + }); } else if (!workspaceId) { throw new Error("--workspace-id is required"); } else if (operation === "exec") { @@ -86,13 +97,23 @@ async function setLifecycle(id, options, lifecycle) { } async function execute(id) { - const envelope = JSON.parse(await readOneShotEnvelope()); + const control = new Socket({ fd: 3, readable: true, writable: false }); + const lines = createInterface({ input: control, crlfDelay: Infinity }); + const iterator = lines[Symbol.asyncIterator](); + const first = await iterator.next(); + if (first.done) throw new Error("spawn control is required"); + const envelope = JSON.parse(first.value); const state = await readState(id, envelope.options); if (!state) throw new Error(`Fixture workspace is missing: ${id}`); await writeFile( path.join(state.root, ".runtime-launch.json"), JSON.stringify({ argv: process.argv, purpose: envelope.purpose }), ); + if (envelope.stdio.kind === "pty") { + await executePty(id, envelope, iterator, control); + control.destroy(); + return; + } const child = spawn(envelope.argv[0], envelope.argv.slice(1), { cwd: path.resolve(state.root, envelope.cwd ?? "."), env: envelope.env, @@ -125,10 +146,88 @@ async function execute(id) { } } +async function executePty(id, envelope, controls, controlStream) { + const events = new Socket({ fd: 4, readable: false, writable: true }); + const child = pty.spawn(envelope.argv[0], envelope.argv.slice(1), { + cwd: path.resolve((await readState(id, envelope.options)).root, envelope.cwd ?? "."), + env: envelope.env, + name: envelope.stdio.term ?? "xterm-256color", + cols: envelope.stdio.cols, + rows: envelope.stdio.rows, + }); + await writeFile(execFile(id, envelope.execId, envelope.options), String(child.pid)); + if (envelope.options.closePtyControl) setTimeout(() => controlStream.destroy(), 100); + process.stdin.on("data", (data) => child.write(data.toString())); + child.onData((data) => process.stdout.write(data)); + if (envelope.options.invalidPtyEvent) { + setTimeout( + () => events.end(`${JSON.stringify({ protocolVersion: 1, type: "invalid" })}\n`), + 100, + ); + } + let requestedSignal = null; + void (async () => { + while (true) { + const next = await controls.next(); + if (next.done) return; + const control = JSON.parse(next.value); + if (control.type === "resize") { + child.resize(control.cols, control.rows); + events.write( + `${JSON.stringify({ protocolVersion: 1, type: "resized", id: control.id })}\n`, + ); + } else if (control.type === "signal") { + requestedSignal = control.signal; + child.kill(control.signal); + } else throw new Error(`Unexpected PTY control: ${control.type}`); + } + })().catch((error) => { + events.write( + `${JSON.stringify({ protocolVersion: 1, type: "error", message: error.message })}\n`, + ); + child.kill("SIGKILL"); + }); + const exit = await new Promise((resolve) => + child.onExit(({ exitCode, signal: signalNumber }) => + resolve({ exitCode, signal: signalNumber }), + ), + ); + await rm(execFile(id, envelope.execId, envelope.options), { force: true }); + process.stdin.destroy(); + if (envelope.options.invalidPtyEvent) return; + events.write(`${JSON.stringify({ protocolVersion: 1, type: "eof" })}\n`); + if (envelope.options.omitPtyExitEvent) { + events.end(); + return; + } + const exitEvent = `${JSON.stringify({ + protocolVersion: 1, + type: "exit", + code: requestedSignal || exit.signal ? null : exit.exitCode, + signal: requestedSignal, + })}\n`; + if (envelope.options.delayedPtyExitEvent) { + const writer = spawn( + process.execPath, + [ + "-e", + `setTimeout(()=>{require('node:fs').writeSync(4,${JSON.stringify(exitEvent)});},1000)`, + ], + { detached: true, stdio: ["ignore", "ignore", "ignore", "ignore", events] }, + ); + writer.unref(); + events.destroy(); + return; + } + events.end(exitEvent); +} + async function signal(id, execId, signalName) { const { options } = JSON.parse(await readStream(process.stdin)); const pid = Number(await readFile(execFile(id, execId, options), "utf8")); killGroup(pid, signalName); + if (options.signalHelperFailure === "error") throw new Error("fixture signal helper failed"); + if (options.signalHelperFailure === "hang") await new Promise(() => {}); } async function applyInspectBarrier(options) { @@ -153,15 +252,6 @@ async function applyInspectBarrier(options) { } } -async function readOneShotEnvelope() { - const stream = new Socket({ fd: 3, readable: true, writable: false }); - try { - return await readStream(stream); - } finally { - stream.destroy(); - } -} - async function readState(id, options) { try { return JSON.parse(await readFile(stateFile(id, options), "utf8")); diff --git a/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts b/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts index 841fed9ed9..d7e740bb2f 100644 --- a/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts +++ b/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts @@ -1,19 +1,25 @@ import { spawn } from "node:child_process"; import { randomBytes } from "node:crypto"; +import { constants } from "node:os"; import type { Readable, Writable } from "node:stream"; +import { StringDecoder } from "node:string_decoder"; import type { WorkspaceDriverCreateInput, WorkspaceDriverInspection, WorkspaceDriverSpawnInput, WorkspacePipeProcess, + WorkspacePtyProcess, WorkspaceRuntimeDriver, } from "../../drivers/index.js"; import { CommandRuntimeDescribeResponseSchema, CommandRuntimeLifecycleResponseSchema, + CommandRuntimePtyEventSchema, } from "./protocol.js"; +const COMMAND_RUNTIME_CLEANUP_TIMEOUT_MS = 1_000; + export interface CommandRuntimeConfig { command: readonly [string, ...string[]]; options?: Readonly>; @@ -23,15 +29,15 @@ export function createCommandRuntime( runtimeId: string, config: CommandRuntimeConfig, ): WorkspaceRuntimeDriver { - let described: Promise | null = null; + let described: Promise> | null = null; - function ensureCompatible(): Promise { + function describe(): Promise> { described ??= runCommand(["describe"], undefined).then((output) => { const description = CommandRuntimeDescribeResponseSchema.parse(JSON.parse(output)); if (!description.modes.includes("pipes")) { throw new Error(`Workspace runtime ${runtimeId} does not support pipes`); } - return undefined; + return new Set(description.modes); }); return described; } @@ -41,7 +47,7 @@ export function createCommandRuntime( workspaceId: string, input?: WorkspaceDriverCreateInput, ) { - await ensureCompatible(); + await describe(); const output = await runCommand( [operation, "--workspace-id", workspaceId], JSON.stringify({ protocolVersion: 1, input, options: config.options ?? {} }), @@ -64,7 +70,13 @@ export function createCommandRuntime( return response.inspection as WorkspaceDriverInspection; }, async spawn(input) { - await ensureCompatible(); + const modes = await describe(); + if (input.stdio.kind === "pty") { + if (!modes.has("pty")) { + throw new Error(`Workspace runtime ${runtimeId} does not support PTY mode`); + } + return spawnCommandPty(runtimeId, config.command, input, config.options ?? {}); + } return spawnCommandProcess(runtimeId, config.command, input, config.options ?? {}); }, async pause(workspaceId) { @@ -107,6 +119,316 @@ export function createCommandRuntime( } } +function spawnCommandPty( + runtimeId: string, + command: readonly [string, ...string[]], + input: WorkspaceDriverSpawnInput, + options: Readonly>, +): WorkspacePtyProcess { + if (input.stdio.kind !== "pty") throw new Error("PTY spawn requires PTY stdio"); + const execId = randomBytes(16).toString("hex"); + const child = spawn( + command[0], + [...command.slice(1), "exec", "--workspace-id", input.workspaceId], + { + env: commandEnvironment(), + shell: false, + stdio: ["pipe", "pipe", "pipe", "pipe", "pipe"], + }, + ); + const control = child.stdio[3] as Writable; + const events = child.stdio[4] as Readable; + const listeners = new Set<(data: string) => void>(); + let pendingData = ""; + const decoder = new StringDecoder("utf8"); + let stderr = ""; + let resizeId = 0; + let pendingResizeId: number | null = null; + let resizeTimeout: ReturnType | null = null; + let pendingWrites = ""; + let settled = false; + let sawEof = false; + let workloadExit: { code: number | null; signal: NodeJS.Signals | null } | null = null; + let eventsEnded = false; + let wrapperClosed = false; + let resolveWrapperClosed!: () => void; + const wrapperClosedPromise = new Promise((resolve) => { + resolveWrapperClosed = resolve; + }); + let wrapperExit: { code: number | null; signal: NodeJS.Signals | null } | null = null; + let failureCleanup: Promise | null = null; + let resolveExit!: (exit: { code: number | null; signal: NodeJS.Signals | null }) => void; + let rejectExit!: (error: Error) => void; + const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>( + (resolve, reject) => { + resolveExit = resolve; + rejectExit = reject; + }, + ); + exited.catch(() => undefined); + + child.stdout.on("data", (chunk: Buffer) => emitData(decoder.write(chunk))); + child.stdout.once("end", () => emitData(decoder.end())); + child.stderr.on("data", (chunk: Buffer | string) => { + stderr += chunk.toString(); + }); + void readNdjson(events, (value) => { + const event = CommandRuntimePtyEventSchema.parse(value); + if (event.type === "eof") { + sawEof = true; + return; + } + if (event.type === "resized") { + if (event.id !== pendingResizeId) return; + pendingResizeId = null; + if (resizeTimeout) clearTimeout(resizeTimeout); + resizeTimeout = null; + if (pendingWrites) { + child.stdin.write(pendingWrites); + pendingWrites = ""; + } + return; + } + if (event.type === "error") { + failPty(new Error(`Workspace runtime ${runtimeId} PTY failed: ${event.message}`)); + return; + } + if (workloadExit) { + throw new Error(`Workspace runtime ${runtimeId} returned duplicate PTY exit events`); + } + const signal = event.signal; + let parsedSignal: NodeJS.Signals | null = null; + if (signal !== null) { + if (!isNodeSignal(signal)) { + throw new Error(`Workspace runtime ${runtimeId} returned an invalid signal: ${signal}`); + } + parsedSignal = signal; + } + workloadExit = { code: event.code, signal: parsedSignal }; + child.stdin.end(); + control.end(); + }).then( + () => { + eventsEnded = true; + finishFromAuthoritativeExit(); + return undefined; + }, + (error) => { + failPty(error); + return undefined; + }, + ); + control.once("error", failPty); + child.stdin.once("error", failPty); + child.once("error", (error) => { + failPty(new Error(`Workspace runtime ${runtimeId} PTY failed: ${error.message}`)); + }); + child.once("close", (code, signal) => { + wrapperClosed = true; + wrapperExit = { code, signal }; + resolveWrapperClosed(); + finishFromAuthoritativeExit(); + }); + writeControl({ + type: "spawn", + protocolVersion: 1, + argv: input.argv, + cwd: input.cwd, + env: input.env, + purpose: input.purpose, + options, + execId, + stdio: input.stdio, + }); + + return { + kind: "pty", + onData(listener) { + listeners.add(listener); + if (pendingData) { + const data = pendingData; + pendingData = ""; + listener(data); + } + return () => listeners.delete(listener); + }, + write(data) { + if (pendingResizeId === null) child.stdin.write(data); + else pendingWrites += data; + }, + resize(cols, rows) { + resizeId += 1; + pendingResizeId = resizeId; + writeControl({ type: "resize", protocolVersion: 1, id: resizeId, cols, rows }); + if (resizeTimeout) clearTimeout(resizeTimeout); + resizeTimeout = setTimeout( + () => failPty(new Error("PTY resize acknowledgement timed out")), + 1000, + ); + }, + exited, + kill(signal = "SIGTERM") { + if (settled || failureCleanup) return; + writeControl({ type: "signal", protocolVersion: 1, signal }); + }, + }; + + function emitData(data: string): void { + if (!data) return; + if (listeners.size === 0) { + pendingData += data; + return; + } + for (const listener of listeners) listener(data); + } + + function writeControl(value: unknown): void { + try { + control.write(`${JSON.stringify(value)}\n`); + } catch (error) { + failPty(error); + } + } + + function finishFromAuthoritativeExit(): void { + if (settled || failureCleanup || !eventsEnded || !wrapperClosed) return; + if (!workloadExit || !sawEof) { + const detail = stderr.trim() || wrapperExit?.signal || wrapperExit?.code || "unknown"; + failPty( + new Error( + `Workspace runtime ${runtimeId} PTY wrapper ended without a valid fd4 exit event (${detail})`, + ), + ); + return; + } + settled = true; + resolveExit(workloadExit); + } + + function failPty(error: unknown): void { + if (settled || failureCleanup) return; + if (resizeTimeout) clearTimeout(resizeTimeout); + resizeTimeout = null; + control.destroy(); + events.destroy(); + child.stdin.destroy(); + const failure = error instanceof Error ? error : new Error(String(error)); + failureCleanup = (async () => { + let cleanupError: unknown; + if (!wrapperClosed) { + try { + await forceKillCommandRuntime( + command, + input.workspaceId, + execId, + options, + child, + wrapperClosedPromise, + ); + } catch (cleanupFailure) { + cleanupError = cleanupFailure; + } + } + settled = true; + const cleanupDetail = + cleanupError instanceof Error ? `; cleanup failed: ${cleanupError.message}` : ""; + rejectExit( + new Error(`Workspace runtime ${runtimeId} PTY failed: ${failure.message}${cleanupDetail}`), + ); + })(); + } +} + +async function forceKillCommandRuntime( + command: readonly [string, ...string[]], + workspaceId: string, + execId: string, + options: Readonly>, + wrapper: { kill(signal?: NodeJS.Signals): boolean }, + wrapperClosed: Promise, +): Promise { + const signalCommand = spawn( + command[0], + [ + ...command.slice(1), + "signal", + "--workspace-id", + workspaceId, + "--exec-id", + execId, + "--signal", + "SIGKILL", + ], + { env: commandEnvironment(), shell: false, stdio: ["pipe", "ignore", "ignore"] }, + ); + const signalResult = new Promise((resolve) => { + let finished = false; + const finish = (error: Error | null) => { + if (finished) return; + finished = true; + resolve(error); + }; + signalCommand.once("error", (error) => finish(error)); + signalCommand.stdin.once("error", (error) => finish(error)); + signalCommand.once("close", (code, signal) => + finish( + code === 0 ? null : new Error(`signal helper failed (${code ?? signal ?? "unknown"})`), + ), + ); + }); + signalCommand.stdin.end(JSON.stringify({ protocolVersion: 1, options })); + const helper = await waitBounded(signalResult, COMMAND_RUNTIME_CLEANUP_TIMEOUT_MS); + let cleanupError: Error | null = null; + if (helper.timedOut) { + cleanupError = new Error("signal helper timed out"); + signalCommand.kill("SIGKILL"); + } else { + cleanupError = helper.value; + if (cleanupError) signalCommand.kill("SIGKILL"); + } + wrapper.kill("SIGKILL"); + const wrapperResult = await waitBounded(wrapperClosed, COMMAND_RUNTIME_CLEANUP_TIMEOUT_MS); + if (wrapperResult.timedOut) { + const wrapperError = new Error("wrapper remained alive after SIGKILL"); + throw cleanupError ? new AggregateError([cleanupError, wrapperError]) : wrapperError; + } + if (cleanupError) throw cleanupError; +} + +async function waitBounded( + promise: Promise, + timeoutMs: number, +): Promise<{ timedOut: false; value: T } | { timedOut: true }> { + let timeout: ReturnType | null = null; + const result = await Promise.race([ + promise.then((value) => ({ timedOut: false as const, value })), + new Promise<{ timedOut: true }>((resolve) => { + timeout = setTimeout(() => resolve({ timedOut: true }), timeoutMs); + }), + ]); + if (timeout) clearTimeout(timeout); + return result; +} + +function isNodeSignal(value: string): value is NodeJS.Signals { + return value in constants.signals; +} + +async function readNdjson(stream: Readable, receive: (value: unknown) => void): Promise { + let pending = ""; + for await (const chunk of stream) { + pending += chunk.toString(); + let newline = pending.indexOf("\n"); + while (newline >= 0) { + const line = pending.slice(0, newline); + pending = pending.slice(newline + 1); + if (line) receive(JSON.parse(line)); + newline = pending.indexOf("\n"); + } + } + if (pending.trim()) receive(JSON.parse(pending)); +} + function spawnCommandProcess( runtimeId: string, command: readonly [string, ...string[]], @@ -128,7 +450,8 @@ function spawnCommandProcess( // The exec process owns the actionable spawn error through `exited`. }); metadata.end( - JSON.stringify({ + `${JSON.stringify({ + type: "spawn", protocolVersion: 1, argv: input.argv, cwd: input.cwd, @@ -136,7 +459,8 @@ function spawnCommandProcess( purpose: input.purpose, options, execId, - }), + stdio: input.stdio, + })}\n`, ); const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>( (resolve, reject) => { @@ -147,26 +471,6 @@ function spawnCommandProcess( }, ); - function forceKill(): void { - const signalCommand = spawn( - command[0], - [ - ...command.slice(1), - "signal", - "--workspace-id", - input.workspaceId, - "--exec-id", - execId, - "--signal", - "SIGKILL", - ], - { env: commandEnvironment(), shell: false, stdio: ["pipe", "ignore", "ignore"] }, - ); - signalCommand.stdin.end(JSON.stringify({ protocolVersion: 1, options })); - signalCommand.once("close", () => child.kill("SIGKILL")); - signalCommand.once("error", () => child.kill("SIGKILL")); - } - return { kind: "pipes", stdin: child.stdin, @@ -175,8 +479,19 @@ function spawnCommandProcess( exited, kill(signal = "SIGTERM") { if (child.exitCode !== null || child.signalCode !== null) return; - if (signal === "SIGKILL") forceKill(); - else child.kill(signal); + if (signal === "SIGKILL") { + void forceKillCommandRuntime( + command, + input.workspaceId, + execId, + options, + child, + exited.then( + () => undefined, + () => undefined, + ), + ).catch(() => undefined); + } else child.kill(signal); }, }; } diff --git a/packages/server/src/server/workspace-runtime/command/internal/protocol.ts b/packages/server/src/server/workspace-runtime/command/internal/protocol.ts index 4e07b6febc..cb6b508a6d 100644 --- a/packages/server/src/server/workspace-runtime/command/internal/protocol.ts +++ b/packages/server/src/server/workspace-runtime/command/internal/protocol.ts @@ -31,5 +31,25 @@ export const CommandRuntimeLifecycleResponseSchema = z.discriminatedUnion("type" export const CommandRuntimeDescribeResponseSchema = z.object({ protocolVersion: z.literal(1), - modes: z.array(z.enum(["pipes"])), + modes: z.array(z.enum(["pipes", "pty"])), }); + +export const CommandRuntimePtyEventSchema = z.discriminatedUnion("type", [ + z.object({ type: z.literal("eof"), protocolVersion: z.literal(1) }), + z.object({ + type: z.literal("resized"), + protocolVersion: z.literal(1), + id: z.number().int().nonnegative(), + }), + z.object({ + type: z.literal("exit"), + protocolVersion: z.literal(1), + code: z.number().int().nullable(), + signal: z.string().nullable(), + }), + z.object({ + type: z.literal("error"), + protocolVersion: z.literal(1), + message: z.string(), + }), +]); diff --git a/packages/server/src/server/workspace-runtime/drivers/index.ts b/packages/server/src/server/workspace-runtime/drivers/index.ts index 1292d4ee68..fe9f658bc6 100644 --- a/packages/server/src/server/workspace-runtime/drivers/index.ts +++ b/packages/server/src/server/workspace-runtime/drivers/index.ts @@ -52,7 +52,7 @@ export interface WorkspaceDriverSpawnInput { | { kind: "workspace-script"; script: string } | { kind: "setup" } | { kind: "archive" }; - stdio: { kind: "pipes" }; + stdio: { kind: "pipes" } | { kind: "pty"; rows: number; cols: number; term?: string }; } export interface WorkspaceProcessExit { @@ -69,11 +69,22 @@ export interface WorkspacePipeProcess { kill(signal?: NodeJS.Signals): void; } +export interface WorkspacePtyProcess { + kind: "pty"; + onData(listener: (data: string) => void): () => void; + write(data: string): void; + resize(cols: number, rows: number): void; + exited: Promise; + kill(signal?: NodeJS.Signals): void; +} + +export type WorkspaceDriverProcess = WorkspacePipeProcess | WorkspacePtyProcess; + export interface WorkspaceRuntimeDriver { readonly id: WorkspaceRuntimeId; create(input: WorkspaceDriverCreateInput): Promise; inspect(workspaceId: WorkspaceId): Promise; - spawn(input: WorkspaceDriverSpawnInput): Promise; + spawn(input: WorkspaceDriverSpawnInput): Promise; pause(workspaceId: WorkspaceId): Promise; resume(workspaceId: WorkspaceId): Promise; destroy(workspaceId: WorkspaceId): Promise; diff --git a/packages/server/src/server/workspace-runtime/index.ts b/packages/server/src/server/workspace-runtime/index.ts index 02e4147477..3882f5e4c7 100644 --- a/packages/server/src/server/workspace-runtime/index.ts +++ b/packages/server/src/server/workspace-runtime/index.ts @@ -49,6 +49,12 @@ export interface WorkspaceProcessInput extends WorkspaceSetupCommand { purpose: WorkspaceProcessPurpose; } +export interface WorkspaceTerminalInput extends WorkspaceProcessInput { + rows: number; + cols: number; + term?: string; +} + export interface WorkspaceProcessExit { code: number | null; signal: NodeJS.Signals | null; @@ -62,9 +68,18 @@ export interface WorkspaceProcess { kill(signal?: NodeJS.Signals): void; } +export interface WorkspaceTerminal { + onData(listener: (data: string) => void): () => void; + write(data: string): void; + resize(cols: number, rows: number): void; + readonly exited: Promise; + kill(signal?: NodeJS.Signals): void; +} + export interface WorkspaceRuntimeService { create(input: CreateWorkspaceInput): Promise<{ workspaceId: string; runtimeId: string }>; run(input: WorkspaceProcessInput): Promise; + openTerminal(input: WorkspaceTerminalInput): Promise; pause(workspaceId: string): Promise; resume(workspaceId: string): Promise; destroy(workspaceId: string): Promise; diff --git a/packages/server/src/server/workspace-runtime/internal/host-process.ts b/packages/server/src/server/workspace-runtime/internal/host-process.ts index 9a3a485a2c..ac74a2eced 100644 --- a/packages/server/src/server/workspace-runtime/internal/host-process.ts +++ b/packages/server/src/server/workspace-runtime/internal/host-process.ts @@ -1,10 +1,13 @@ import { spawn } from "node:child_process"; import { realpath } from "node:fs/promises"; +import { constants } from "node:os"; import path from "node:path"; +import * as pty from "node-pty"; import type { WorkspaceDriverSpawnInput, WorkspacePipeProcess, + WorkspacePtyProcess, WorkspaceProcessExit, } from "../drivers/index.js"; @@ -48,6 +51,74 @@ export async function spawnHostProcess( }; } +export async function spawnHostPty( + root: string, + input: WorkspaceDriverSpawnInput, +): Promise { + if (input.stdio.kind !== "pty") throw new Error("PTY spawn requires PTY stdio"); + const cwd = await resolveRuntimeCwd(root, input.cwd); + const process = pty.spawn(input.argv[0], input.argv.slice(1), { + cwd, + env: { ...input.env }, + name: input.stdio.term ?? "xterm-256color", + cols: input.stdio.cols, + rows: input.stdio.rows, + }); + const exited = new Promise((resolve) => { + process.onExit(({ exitCode, signal }) => { + const signalNumber = signal ?? 0; + resolve({ + code: signalNumber === 0 ? exitCode : null, + signal: signalName(signalNumber), + }); + }); + }); + const listeners = new Set<(data: string) => void>(); + let pendingData = ""; + process.onData((data) => { + if (listeners.size === 0) { + pendingData += data; + return; + } + for (const listener of listeners) listener(data); + }); + + return { + kind: "pty", + onData(listener) { + listeners.add(listener); + if (pendingData) { + const data = pendingData; + pendingData = ""; + listener(data); + } + return () => listeners.delete(listener); + }, + write(data) { + process.write(data); + }, + resize(cols, rows) { + process.resize(cols, rows); + }, + exited, + kill(signal) { + process.kill(signal); + }, + }; +} + +function signalName(signal: number): NodeJS.Signals | null { + if (signal === 0) return null; + for (const [name, number] of Object.entries(constants.signals)) { + if (number !== undefined && number === signal && isNodeSignal(name)) return name; + } + return null; +} + +function isNodeSignal(value: string): value is NodeJS.Signals { + return value in constants.signals; +} + export async function resolveRuntimeCwd(root: string, relativeCwd?: string): Promise { const normalizedRoot = await realpath(root); const cwd = await realpath(path.resolve(normalizedRoot, relativeCwd ?? ".")); diff --git a/packages/server/src/server/workspace-runtime/internal/local-runtime.ts b/packages/server/src/server/workspace-runtime/internal/local-runtime.ts index 78800b9849..5d7b8806d3 100644 --- a/packages/server/src/server/workspace-runtime/internal/local-runtime.ts +++ b/packages/server/src/server/workspace-runtime/internal/local-runtime.ts @@ -8,7 +8,7 @@ import type { WorkspaceDriverState, WorkspaceRuntimeDriver, } from "../drivers/index.js"; -import { resolveRuntimeCwd, spawnHostProcess } from "./host-process.js"; +import { resolveRuntimeCwd, spawnHostProcess, spawnHostPty } from "./host-process.js"; import { createRuntimeStateStore } from "./runtime-state.js"; export function createLocalRuntime(paseoHome: string): WorkspaceRuntimeDriver { @@ -59,7 +59,8 @@ export function createLocalRuntime(paseoHome: string): WorkspaceRuntimeDriver { }, inspect, async spawn(input: WorkspaceDriverSpawnInput) { - return spawnHostProcess((await requireReady(input.workspaceId)).root, input); + const root = (await requireReady(input.workspaceId)).root; + return input.stdio.kind === "pty" ? spawnHostPty(root, input) : spawnHostProcess(root, input); }, async pause(workspaceId) { const inspection = await inspect(workspaceId); diff --git a/packages/server/src/server/workspace-runtime/internal/service.ts b/packages/server/src/server/workspace-runtime/internal/service.ts index 2ed1514d2d..a1163713bd 100644 --- a/packages/server/src/server/workspace-runtime/internal/service.ts +++ b/packages/server/src/server/workspace-runtime/internal/service.ts @@ -4,10 +4,12 @@ import type { WorkspaceProcessInput, WorkspaceRuntimeRecordStore, WorkspaceRuntimeService, + WorkspaceTerminal, + WorkspaceTerminalInput, } from "../index.js"; import type { + WorkspaceDriverProcess, WorkspaceDriverCreateInput, - WorkspacePipeProcess, WorkspaceRuntimeDriver, } from "../drivers/index.js"; @@ -19,7 +21,7 @@ export function createService( records: WorkspaceRuntimeRecordStore, ): WorkspaceRuntimeService { const driversById = new Map(drivers.map((driver) => [driver.id, driver])); - const processesByWorkspaceId = new Map>(); + const processesByWorkspaceId = new Map>(); const workspaceTails = new Map>(); function requireRegistered(runtimeId: string): WorkspaceRuntimeDriver { @@ -44,13 +46,30 @@ export function createService( throw new Error(`Workspace runtime is ${inspection.status}: ${input.workspaceId}`); } const runtimeProcess = await driver.spawn({ ...input, stdio: { kind: "pipes" } }); - const processes = processesByWorkspaceId.get(input.workspaceId) ?? new Set(); - processes.add(runtimeProcess); - processesByWorkspaceId.set(input.workspaceId, processes); - void runtimeProcess.exited.then( - () => forgetProcess(input.workspaceId, runtimeProcess), - () => forgetProcess(input.workspaceId, runtimeProcess), - ); + if (runtimeProcess.kind !== "pipes") { + throw new Error(`Workspace runtime returned PTY mode for a pipe launch: ${driver.id}`); + } + trackProcess(input.workspaceId, runtimeProcess); + return runtimeProcess; + } + + async function openTerminalWithDriver( + driver: WorkspaceRuntimeDriver, + input: WorkspaceTerminalInput, + ): Promise { + validateRelativeCwd(input.cwd); + const inspection = await driver.inspect(input.workspaceId); + if (inspection.status !== "ready") { + throw new Error(`Workspace runtime is ${inspection.status}: ${input.workspaceId}`); + } + const runtimeProcess = await driver.spawn({ + ...input, + stdio: { kind: "pty", rows: input.rows, cols: input.cols, term: input.term }, + }); + if (runtimeProcess.kind !== "pty") { + throw new Error(`Workspace runtime does not support PTY mode: ${driver.id}`); + } + trackProcess(input.workspaceId, runtimeProcess); return runtimeProcess; } @@ -109,6 +128,11 @@ export function createService( runWithDriver(await resolve(input.workspaceId), input), ); }, + async openTerminal(input) { + return sequence(input.workspaceId, async () => + openTerminalWithDriver(await resolve(input.workspaceId), input), + ); + }, async pause(workspaceId) { await sequence(workspaceId, async () => { const driver = await resolve(workspaceId); @@ -147,7 +171,17 @@ export function createService( } } - function forgetProcess(workspaceId: string, runtimeProcess: WorkspacePipeProcess): void { + function trackProcess(workspaceId: string, runtimeProcess: WorkspaceDriverProcess): void { + const processes = processesByWorkspaceId.get(workspaceId) ?? new Set(); + processes.add(runtimeProcess); + processesByWorkspaceId.set(workspaceId, processes); + void runtimeProcess.exited.then( + () => forgetProcess(workspaceId, runtimeProcess), + () => forgetProcess(workspaceId, runtimeProcess), + ); + } + + function forgetProcess(workspaceId: string, runtimeProcess: WorkspaceDriverProcess): void { const processes = processesByWorkspaceId.get(workspaceId); processes?.delete(runtimeProcess); if (processes?.size === 0) processesByWorkspaceId.delete(workspaceId); @@ -191,7 +225,7 @@ function toDriverCreateInput(input: CreateWorkspaceInput): WorkspaceDriverCreate } async function waitForAll( - processes: readonly WorkspacePipeProcess[], + processes: readonly WorkspaceDriverProcess[], timeoutMilliseconds: number, ): Promise { if (processes.length === 0) return true; diff --git a/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts b/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts index 7edcc0e129..240ef1cda1 100644 --- a/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts +++ b/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts @@ -14,7 +14,7 @@ import type { WorkspaceDriverState, WorkspaceRuntimeDriver, } from "../drivers/index.js"; -import { resolveRuntimeCwd, spawnHostProcess } from "./host-process.js"; +import { resolveRuntimeCwd, spawnHostProcess, spawnHostPty } from "./host-process.js"; import { createRuntimeStateStore } from "./runtime-state.js"; interface WorktreeRuntimeState extends WorkspaceDriverState { @@ -97,7 +97,8 @@ export function createWorktreeRuntime(options: { }, inspect, async spawn(input: WorkspaceDriverSpawnInput) { - return spawnHostProcess((await requireReady(input.workspaceId)).root, input); + const root = (await requireReady(input.workspaceId)).root; + return input.stdio.kind === "pty" ? spawnHostPty(root, input) : spawnHostProcess(root, input); }, async pause(workspaceId) { const inspection = await inspect(workspaceId); diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts index 97f9aa8144..0f55a75d67 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts @@ -62,6 +62,200 @@ test("a trusted registered command is selected and receives secret launch data o await fixture.service.destroy(fixture.workspaceId); }); +test("the command runtime transports PTY input, Unicode output, resize, and signals", async () => { + const fixture = await createFixture("pty"); + await fixture.service.create(fixture.createInput); + const secret = "terminal-secret-value"; + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [ + processExecPath(), + "-e", + "process.stdin.setEncoding('utf8');process.stdout.write(`${process.stdout.isTTY}|${process.stdout.columns}x${process.stdout.rows}|λ|${process.env.SECRET}`);process.stdin.once('data',data=>{process.stdout.write(`|${data.trim()}|${process.stdout.columns}x${process.stdout.rows}`);process.exit(9)})", + ], + env: { SECRET: secret, PATH: process.env.PATH ?? "" }, + purpose: { kind: "terminal", terminalId: "command-pty" }, + rows: 24, + cols: 80, + }); + let output = ""; + terminal.onData((data) => { + output += data; + }); + await vi.waitFor(() => expect(output).toContain(`true|80x24|λ|${secret}`)); + terminal.resize(99, 41); + terminal.write("héllo\n"); + await expect(terminal.exited).resolves.toEqual({ code: 9, signal: null }); + expect(output).toContain("|héllo|99x41"); + const launch = JSON.parse( + await readFile(path.join(fixture.source, ".runtime-launch.json"), "utf8"), + ) as { argv: string[] }; + expect(JSON.stringify(launch.argv)).not.toContain(secret); + + const signaled = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: "command-signal" }, + rows: 24, + cols: 80, + }); + signaled.kill("SIGTERM"); + await expect(signaled.exited).resolves.toEqual({ code: null, signal: "SIGTERM" }); + const forced = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: "command-force" }, + rows: 24, + cols: 80, + }); + forced.kill("SIGKILL"); + await expect(forced.exited).resolves.toEqual({ code: null, signal: "SIGKILL" }); + await fixture.service.destroy(fixture.workspaceId); +}); + +test("a registered pipes-only command runtime fails closed when asked for a PTY", async () => { + const fixture = await createFixture("pipes-only", false, "pipes"); + await fixture.service.create(fixture.createInput); + + await expect( + fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: ["/bin/sh"], + env: {}, + purpose: { kind: "terminal", terminalId: "unsupported-pty" }, + rows: 24, + cols: 80, + }), + ).rejects.toThrow("Workspace runtime fixture does not support PTY mode"); + + await fixture.service.destroy(fixture.workspaceId); +}); + +test("the fd4 workload exit remains authoritative after the wrapper exits", async () => { + const fixture = await createFixture("delayed-pty-exit", false, "pty", { + delayedPtyExitEvent: true, + }); + await fixture.service.create(fixture.createInput); + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [processExecPath(), "-e", "process.exit(6)"], + env: {}, + purpose: { kind: "terminal", terminalId: "delayed-exit" }, + rows: 24, + cols: 80, + }); + + await expect(terminal.exited).resolves.toEqual({ code: 6, signal: null }); + await fixture.service.destroy(fixture.workspaceId); +}); + +test("a wrapper exit without an fd4 workload exit rejects", async () => { + const fixture = await createFixture("missing-pty-exit", false, "pty", { + omitPtyExitEvent: true, + }); + await fixture.service.create(fixture.createInput); + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [processExecPath(), "-e", "process.exit(6)"], + env: {}, + purpose: { kind: "terminal", terminalId: "missing-exit" }, + rows: 24, + cols: 80, + }); + + await expect(terminal.exited).rejects.toThrow("ended without a valid fd4 exit event"); + await fixture.service.destroy(fixture.workspaceId); +}); + +test("an invalid fd4 event rejects and terminates the wrapper workload", async () => { + const fixture = await createFixture("invalid-pty-event", false, "pty", { + invalidPtyEvent: true, + }); + const pidFile = path.join(fixture.source, "invalid-event.pid"); + await fixture.service.create(fixture.createInput); + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [ + processExecPath(), + "-e", + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)},String(process.pid));setInterval(()=>{},1000)`, + ], + env: {}, + purpose: { kind: "terminal", terminalId: "invalid-event" }, + rows: 24, + cols: 80, + }); + + await expect(terminal.exited).rejects.toThrow("Invalid discriminator value"); + const workloadPid = Number(await readFile(pidFile, "utf8")); + expect(processExists(workloadPid)).toBe(false); + await fixture.service.destroy(fixture.workspaceId); +}); + +test("a failed PTY control channel rejects and terminates the wrapper workload", async () => { + const fixture = await createFixture("failed-pty-control", false, "pty", { + closePtyControl: true, + }); + const pidFile = path.join(fixture.source, "failed-control.pid"); + await fixture.service.create(fixture.createInput); + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [ + processExecPath(), + "-e", + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)},String(process.pid));setInterval(()=>{},1000)`, + ], + env: {}, + purpose: { kind: "terminal", terminalId: "failed-control" }, + rows: 24, + cols: 80, + }); + const workloadPid = await vi.waitFor(async () => { + const pid = Number(await readFile(pidFile, "utf8")); + expect(processExists(pid)).toBe(true); + return pid; + }); + await new Promise((resolve) => setTimeout(resolve, 150)); + + terminal.resize(100, 40); + + await expect(terminal.exited).rejects.toThrow("PTY resize acknowledgement timed out"); + expect(processExists(workloadPid)).toBe(false); + await fixture.service.destroy(fixture.workspaceId); +}); + +test.each(["error", "hang"] as const)( + "PTY cleanup is bounded when the signal helper ends with %s", + async (signalHelperFailure) => { + const fixture = await createFixture(`pty-signal-helper-${signalHelperFailure}`, false, "pty", { + invalidPtyEvent: true, + signalHelperFailure, + }); + const pidFile = path.join(fixture.source, `signal-helper-${signalHelperFailure}.pid`); + await fixture.service.create(fixture.createInput); + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [ + processExecPath(), + "-e", + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)},String(process.pid));setInterval(()=>{},1000)`, + ], + env: {}, + purpose: { kind: "terminal", terminalId: `signal-helper-${signalHelperFailure}` }, + rows: 24, + cols: 80, + }); + + await expect(terminal.exited).rejects.toThrow("cleanup failed"); + const workloadPid = Number(await readFile(pidFile, "utf8")); + expect(processExists(workloadPid)).toBe(false); + await fixture.service.destroy(fixture.workspaceId); + }, + 5_000, +); + test("run admission racing pause cannot leave an unregistered workload running", async () => { const fixture = await createFixture("race", true); await fixture.service.create(fixture.createInput); @@ -107,7 +301,12 @@ test("an existing runtime selection cannot be switched before target driver disp await fixture.service.destroy(fixture.workspaceId); }); -async function createFixture(name: string, withBarrier = false) { +async function createFixture( + name: string, + withBarrier = false, + modes: "pipes" | "pty" = "pty", + runtimeOptions: Readonly> = {}, +) { const root = await mkdtemp(path.join(tmpdir(), `paseo-command-runtime-${name}-`)); cleanupRoots.push(root); const source = path.join(root, "source"); @@ -125,9 +324,14 @@ async function createFixture(name: string, withBarrier = false) { externalRuntimes: { fixture: { type: "command", - command: [processExecPath(), fixtureExecutable], + command: [ + processExecPath(), + fixtureExecutable, + ...(modes === "pipes" ? ["--modes", "pipes"] : []), + ], options: { stateDirectory, + ...runtimeOptions, ...(withBarrier ? { inspectBarrierDirectory: barrierDirectory } : {}), }, }, @@ -151,3 +355,12 @@ async function createFixture(name: string, withBarrier = false) { function processExecPath(): string { return process.execPath; } + +function processExists(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } +} diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts index e6b91bbb2d..108a0862cb 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts @@ -74,9 +74,9 @@ test("the command runtime owns Docker workspace materialization, pipes, and life purpose: { kind: "workspace-script", script: "write-contract" }, }); write.stdin.end("docker-state"); - const [stderr, exit] = await Promise.all([collect(write.stderr), write.exited]); + const [stderr, writeExit] = await Promise.all([collect(write.stderr), write.exited]); expect(stderr).toBe("runtime-stderr"); - expect(exit).toEqual({ code: 19, signal: null }); + expect(writeExit).toEqual({ code: 19, signal: null }); const env = await service.run({ workspaceId, @@ -118,6 +118,72 @@ test("the command runtime owns Docker workspace materialization, pipes, and life read.stdin.end(); await expect(collect(read.stdout)).resolves.toBe("docker-state"); await expect(read.exited).resolves.toEqual({ code: 0, signal: null }); + + const terminal = await recovered.openTerminal({ + workspaceId, + argv: [ + "/usr/local/bin/node", + "-e", + "process.stdin.setEncoding('utf8');process.stdout.write(`${process.cwd()}|${process.stdout.isTTY}|${process.stdout.columns}x${process.stdout.rows}|λ`);process.stdin.once('data',data=>{process.stdout.write(`|${data.trim()}|${process.stdout.columns}x${process.stdout.rows}`);process.exit(6)})", + ], + env: { PATH: "/usr/local/bin:/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: "docker-contract-terminal" }, + rows: 24, + cols: 80, + }); + let terminalOutput = ""; + terminal.onData((data) => { + terminalOutput += data; + }); + await Promise.race([ + vi.waitFor(() => expect(terminalOutput).toContain("/workspace|true|80x24|λ"), { + timeout: 10_000, + }), + terminal.exited.then((exit) => { + throw new Error(`Docker PTY exited before initial output: ${JSON.stringify(exit)}`); + }), + ]); + terminal.resize(103, 39); + terminal.write("héllo\n"); + await expect(terminal.exited).resolves.toEqual({ code: 6, signal: null }); + expect(terminalOutput).toContain("|héllo|103x39"); + + const terminalSignal = await recovered.openTerminal({ + workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: "docker-signal-terminal" }, + rows: 24, + cols: 80, + }); + terminalSignal.kill("SIGTERM"); + await expect(terminalSignal.exited).resolves.toEqual({ code: null, signal: "SIGTERM" }); + await vi.waitFor(() => expect(dockerProcesses(workspaceId)).not.toContain("sleep 30")); + + const terminalForced = await recovered.openTerminal({ + workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: "docker-force-terminal" }, + rows: 24, + cols: 80, + }); + terminalForced.kill("SIGKILL"); + await expect(terminalForced.exited).resolves.toEqual({ code: null, signal: "SIGKILL" }); + await vi.waitFor(() => expect(dockerProcesses(workspaceId)).not.toContain("sleep 30")); + + const terminalControlFailure = await recovered.openTerminal({ + workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: "docker-control-failure-terminal" }, + rows: 24, + cols: 80, + }); + terminalControlFailure.resize(0, 0); + await expect(terminalControlFailure.exited).rejects.toThrow("Invalid PTY size"); + expect(dockerProcesses(workspaceId)).not.toContain("sleep 30"); + const committedOnly = await recovered.run({ workspaceId, argv: ["/bin/sh", "-c", "test ! -e .env"], diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts index 8f8458f97e..651cf62318 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts @@ -23,6 +23,55 @@ afterEach(async () => { }); posixDescribe.each(["local", "worktree"] as const)("%s runtime public contract", (runtimeId) => { + test("opens an interactive PTY with input, Unicode output, resize, EOF, and signals", async () => { + const fixture = await createFixture(runtimeId); + await fixture.service.create(fixture.createInput); + + const terminal = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: [ + process.execPath, + "-e", + "process.stdin.setEncoding('utf8');process.stdout.write(`${process.cwd()}|${process.stdout.isTTY}|${process.stdout.columns}x${process.stdout.rows}|λ`);process.stdin.once('data',data=>{process.stdout.write(`|${data.trim()}|${process.stdout.columns}x${process.stdout.rows}`);process.exit(7)})", + ], + env: { PATH: process.env.PATH ?? "" }, + purpose: { kind: "terminal", terminalId: `${runtimeId}-terminal` }, + rows: 24, + cols: 80, + term: "xterm-256color", + }); + const output = collectTerminal(terminal); + await waitForTerminalOutput(output, "|true|80x24|λ"); + terminal.resize(101, 37); + terminal.write("héllo\n"); + await expect(terminal.exited).resolves.toEqual({ code: 7, signal: null }); + expect(output.value()).toContain("|héllo|101x37"); + + const signaled = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: `${runtimeId}-signal-terminal` }, + rows: 24, + cols: 80, + }); + signaled.kill("SIGTERM"); + await expect(signaled.exited).resolves.toEqual({ code: null, signal: "SIGTERM" }); + + const forced = await fixture.service.openTerminal({ + workspaceId: fixture.workspaceId, + argv: ["/bin/sleep", "30"], + env: { PATH: "/usr/bin:/bin" }, + purpose: { kind: "terminal", terminalId: `${runtimeId}-force-terminal` }, + rows: 24, + cols: 80, + }); + forced.kill("SIGKILL"); + await expect(forced.exited).resolves.toEqual({ code: null, signal: "SIGKILL" }); + + await fixture.service.destroy(fixture.workspaceId); + }); + test("creates, pipes an exact environment, preserves status, and owns only its resources", async () => { const fixture = await createFixture(runtimeId); const created = await fixture.service.create(fixture.createInput); @@ -362,6 +411,20 @@ async function collect(stream: NodeJS.ReadableStream): Promise { return Buffer.concat(chunks).toString("utf8"); } +function collectTerminal(terminal: { onData(listener: (data: string) => void): () => void }): { + value(): string; +} { + let output = ""; + terminal.onData((data) => { + output += data; + }); + return { value: () => output }; +} + +async function waitForTerminalOutput(output: { value(): string }, marker: string): Promise { + await vi.waitFor(() => expect(output.value()).toContain(marker)); +} + function isProcessAlive(pid: number): boolean { try { process.kill(pid, 0); diff --git a/packages/server/src/server/worktree-bootstrap.posix.test.ts b/packages/server/src/server/worktree-bootstrap.posix.test.ts index a0a796c83d..39a008c5e9 100644 --- a/packages/server/src/server/worktree-bootstrap.posix.test.ts +++ b/packages/server/src/server/worktree-bootstrap.posix.test.ts @@ -44,7 +44,7 @@ async function cleanupTerminalManager(terminalManager: TerminalManager): Promise ); const terminals = terminalsByCwd.flat(); await Promise.all(terminals.map((terminal) => killTerminal(terminalManager, terminal))); - terminalManager.killAll(); + await terminalManager.killAll(); } function killTerminal(terminalManager: TerminalManager, terminal: TerminalSession): Promise { diff --git a/packages/server/src/server/worktree-bootstrap.test.ts b/packages/server/src/server/worktree-bootstrap.test.ts index c878ef1387..e14d81b3a1 100644 --- a/packages/server/src/server/worktree-bootstrap.test.ts +++ b/packages/server/src/server/worktree-bootstrap.test.ts @@ -36,7 +36,7 @@ async function cleanupTerminalManager(terminalManager: TerminalManager): Promise ); const terminals = terminalsByCwd.flat(); await Promise.all(terminals.map((terminal) => killTerminal(terminalManager, terminal))); - terminalManager.killAll(); + await terminalManager.killAll(); } function killTerminal(terminalManager: TerminalManager, terminal: TerminalSession): Promise { diff --git a/packages/server/src/terminal/terminal-manager-factory.ts b/packages/server/src/terminal/terminal-manager-factory.ts index 88dcc2d692..acab080dfe 100644 --- a/packages/server/src/terminal/terminal-manager-factory.ts +++ b/packages/server/src/terminal/terminal-manager-factory.ts @@ -1,8 +1,11 @@ import type { TerminalManager } from "./terminal-manager.js"; +import type { TerminalPtyLaunchInput } from "./terminal.js"; +import type { WorkspaceTerminal } from "../server/workspace-runtime/index.js"; import { createWorkerTerminalManager } from "./worker-terminal-manager.js"; export interface ConfiguredTerminalManagerOptions { getTerminalActivityUrl?: () => string | null; + launchPty?: (input: TerminalPtyLaunchInput) => Promise; } export function createConfiguredTerminalManager( diff --git a/packages/server/src/terminal/terminal-manager.test.ts b/packages/server/src/terminal/terminal-manager.test.ts index 47905d1424..8fb9bb2eb7 100644 --- a/packages/server/src/terminal/terminal-manager.test.ts +++ b/packages/server/src/terminal/terminal-manager.test.ts @@ -36,7 +36,7 @@ afterEach(async () => { for (const terminal of terminalsByCwd.flat()) { await manager.killTerminalAndWait(terminal.id); } - manager.killAll(); + await manager.killAll(); } await new Promise((resolve) => setTimeout(resolve, 50)); while (temporaryDirs.length > 0) { @@ -344,7 +344,7 @@ it("kills all terminals and clears state", async () => { const tmpId = tmpSession.id; const homeId = homeSession.id; - manager.killAll(); + await manager.killAll(); expect(manager.listDirectories()).toEqual([]); expect(manager.getTerminal(tmpId)).toBeUndefined(); diff --git a/packages/server/src/terminal/terminal-manager.ts b/packages/server/src/terminal/terminal-manager.ts index df4ced389b..f8f5640e5c 100644 --- a/packages/server/src/terminal/terminal-manager.ts +++ b/packages/server/src/terminal/terminal-manager.ts @@ -4,6 +4,7 @@ import { type TerminalSession, type TerminalStateSnapshot, type TerminalStateSnapshotOptions, + type TerminalPtyLauncher, } from "./terminal.js"; import { captureTerminalLines, type CaptureTerminalLinesResult } from "./terminal-capture.js"; import { randomBytes, randomUUID } from "node:crypto"; @@ -85,7 +86,7 @@ export interface TerminalManager { options?: { start?: number; end?: number; stripAnsi?: boolean }, ): Promise; listDirectories(): string[]; - killAll(): void; + killAll(): Promise; subscribeTerminalsChanged(listener: TerminalsChangedListener): () => void; subscribeTerminalActivity(listener: TerminalActivityListener): () => void; subscribeTerminalWorkspaceContributionChanged( @@ -95,6 +96,7 @@ export interface TerminalManager { export interface TerminalManagerOptions { getTerminalActivityUrl?: () => string | null; + resolvePtyLauncher?: (terminalId: string) => TerminalPtyLauncher | undefined; } function createActivityToken(): string { @@ -341,6 +343,7 @@ export function createTerminalManager( ...(terminalActivityUrl ? { PASEO_TERMINAL_ACTIVITY_URL: terminalActivityUrl } : {}), }; terminalActivityTokenById.set(terminalId, activityToken); + const launchPty = managerOptions.resolvePtyLauncher?.(terminalId); let session: TerminalSession; try { session = registerSession( @@ -356,6 +359,7 @@ export function createTerminalManager( ...(options.cols !== undefined ? { cols: options.cols } : {}), ...(mergedEnv ? { env: mergedEnv } : {}), activityEnv, + ...(launchPty ? { launchPty } : {}), }), ); } catch (error) { @@ -463,10 +467,14 @@ export function createTerminalManager( return Array.from(terminalsByCwd.keys()); }, - killAll(): void { - for (const id of Array.from(terminalsById.keys())) { - removeSessionById(id, { kill: true }); - } + async killAll(): Promise { + const sessions = Array.from(terminalsById.values()); + for (const session of sessions) removeSessionById(session.id, { kill: false }); + await Promise.all( + sessions.map((session) => + session.killAndWait({ gracefulTimeoutMs: 500, forceTimeoutMs: 500 }), + ), + ); }, subscribeTerminalsChanged(listener: TerminalsChangedListener): () => void { diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 4e16f8de04..1449ee2f65 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -1,17 +1,33 @@ import { createTerminalManager } from "./terminal-manager.js"; +import { randomUUID } from "node:crypto"; +import { constants } from "node:os"; import { captureTerminalLines } from "./terminal-capture.js"; import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js"; -import type { TerminalSession, TerminalStateSnapshotOptions } from "./terminal.js"; +import type { + TerminalPtyProcess, + TerminalPtyLaunchInput, + TerminalSession, + TerminalStateSnapshotOptions, +} from "./terminal.js"; import type { TerminalWorkerRequest, + TerminalWorkerParentMessage, + TerminalWorkerPtyEvent, + TerminalWorkerMessage, TerminalWorkerStateResult, - TerminalWorkerToParentMessage, WorkerTerminalInfo, } from "./terminal-worker-protocol.js"; type TerminalCreateRequest = Extract; -const manager = createTerminalManager(); +const ptyProcesses = new Map(); +const pendingPtySpawns = new Map< + string, + { resolve: (process: TerminalPtyProcess | null) => void; reject: (error: Error) => void } +>(); +const manager = createTerminalManager({ + resolvePtyLauncher: () => launchRuntimePty, +}); const unsubscribeByTerminalId = new Map void>>(); const outputCoalescerByTerminalId = new Map(); let ipcClosing = false; @@ -36,21 +52,133 @@ process.on("uncaughtException", (error) => { reportInFlightTerminalCreateFailure(error); }); -function sendToParent(message: TerminalWorkerToParentMessage): void { +function sendToParent(message: TerminalWorkerMessage): void { if (ipcClosing || !process.connected || !process.send) { + closePtyBridge(new Error("Terminal parent IPC is not connected")); return; } try { process.send(message, (error) => { - if (error) { - ipcClosing = true; - } + if (error) closePtyBridge(error); }); - } catch { - ipcClosing = true; + } catch (error) { + closePtyBridge(error); } } +function closePtyBridge(error: unknown): void { + ipcClosing = true; + const failure = error instanceof Error ? error : new Error(String(error)); + for (const pending of pendingPtySpawns.values()) pending.reject(failure); + pendingPtySpawns.clear(); + for (const proxy of ptyProcesses.values()) { + proxy.emitExit({ code: 1, signal: null }); + } + ptyProcesses.clear(); +} + +interface RuntimePtyProxy extends TerminalPtyProcess { + emitData(data: string): void; + emitExit(exit: { code: number | null; signal: NodeJS.Signals | null }): void; +} + +function launchRuntimePty( + input: TerminalPtyLaunchInput, +): TerminalPtyProcess | null | Promise { + return new Promise((resolve, reject) => { + const processId = randomUUID(); + pendingPtySpawns.set(processId, { resolve, reject }); + sendToParent({ type: "ptySpawn", processId, input }); + }); +} + +function createRuntimePtyProxy(processId: string): RuntimePtyProxy { + const dataListeners = new Set<(data: string) => void>(); + const exitListeners = new Set<(event: { exitCode: number; signal?: number }) => void>(); + let pendingData = ""; + let pendingExit: { exitCode: number; signal?: number } | null = null; + return { + pid: -1, + write(data) { + sendToParent({ type: "ptyWrite", processId, data }); + }, + resize(cols, rows) { + sendToParent({ type: "ptyResize", processId, cols, rows }); + }, + kill(signal) { + sendToParent({ type: "ptyKill", processId, signal }); + }, + onData(listener) { + dataListeners.add(listener); + if (pendingData) { + const data = pendingData; + pendingData = ""; + listener(data); + } + return { dispose: () => dataListeners.delete(listener) }; + }, + onExit(listener) { + exitListeners.add(listener); + const exit = pendingExit; + if (exit) queueMicrotask(() => listener(exit)); + return { dispose: () => exitListeners.delete(listener) }; + }, + emitData(data) { + if (dataListeners.size === 0) { + pendingData += data; + return; + } + for (const listener of dataListeners) listener(data); + }, + emitExit(exit) { + const signal = exit.signal ? constants.signals[exit.signal] : undefined; + const event = { exitCode: exit.code ?? 0, signal }; + if (exitListeners.size === 0) pendingExit = event; + else for (const listener of exitListeners) listener(event); + dataListeners.clear(); + if (exitListeners.size > 0) exitListeners.clear(); + }, + }; +} + +function handlePtyEvent(message: TerminalWorkerPtyEvent): void { + if (message.type === "ptySpawnResult") { + const pending = pendingPtySpawns.get(message.processId); + if (!pending) return; + pendingPtySpawns.delete(message.processId); + if (message.mode === "legacy") { + pending.resolve(null); + return; + } + const proxy = createRuntimePtyProxy(message.processId); + ptyProcesses.set(message.processId, proxy); + pending.resolve(proxy); + return; + } + if (message.type === "ptySpawnError") { + const pending = pendingPtySpawns.get(message.processId); + pendingPtySpawns.delete(message.processId); + pending?.reject(new Error(message.error)); + return; + } + const proxy = ptyProcesses.get(message.processId); + if (!proxy) return; + if (message.type === "ptyData") proxy.emitData(message.data); + else { + ptyProcesses.delete(message.processId); + setImmediate(() => proxy.emitExit(message.exit)); + } +} + +function isPtyEvent(message: TerminalWorkerParentMessage): message is TerminalWorkerPtyEvent { + return ( + message.type === "ptySpawnResult" || + message.type === "ptySpawnError" || + message.type === "ptyData" || + message.type === "ptyExit" + ); +} + function buildTerminalStateResult( session: TerminalSession | undefined, options?: TerminalStateSnapshotOptions, @@ -316,7 +444,7 @@ async function handleRequest(message: TerminalWorkerRequest): Promise { } case "killAll": { - manager.killAll(); + await manager.killAll(); for (const terminalId of Array.from(unsubscribeByTerminalId.keys())) { clearTerminalSubscriptions(terminalId); } @@ -333,7 +461,11 @@ async function handleRequest(message: TerminalWorkerRequest): Promise { } } -process.on("message", (message: TerminalWorkerRequest) => { +process.on("message", (message: TerminalWorkerParentMessage) => { + if (isPtyEvent(message)) { + handlePtyEvent(message); + return; + } void handleRequest(message).catch((error: unknown) => { sendToParent({ type: "response", @@ -345,6 +477,6 @@ process.on("message", (message: TerminalWorkerRequest) => { }); process.once("disconnect", () => { - ipcClosing = true; - manager.killAll(); + closePtyBridge(new Error("Terminal parent IPC disconnected")); + void manager.killAll(); }); diff --git a/packages/server/src/terminal/terminal-worker-protocol.ts b/packages/server/src/terminal/terminal-worker-protocol.ts index 6b722c978a..97310159d9 100644 --- a/packages/server/src/terminal/terminal-worker-protocol.ts +++ b/packages/server/src/terminal/terminal-worker-protocol.ts @@ -8,6 +8,7 @@ import type { import type { TerminalState } from "@getpaseo/protocol/messages"; import type { TerminalActivity, TerminalActivityState } from "@getpaseo/protocol/terminal-activity"; import type { CaptureTerminalLinesResult } from "./terminal-capture.js"; +import type { TerminalPtyLaunchInput } from "./terminal.js"; export interface WorkerTerminalInfo { id: string; @@ -97,6 +98,18 @@ export type TerminalWorkerRequest = message: ClientMessage; }; +export type TerminalWorkerPtyEvent = + | { type: "ptySpawnResult"; processId: string; mode: "runtime" | "legacy" } + | { type: "ptySpawnError"; processId: string; error: string } + | { type: "ptyData"; processId: string; data: string } + | { + type: "ptyExit"; + processId: string; + exit: { code: number | null; signal: NodeJS.Signals | null }; + }; + +export type TerminalWorkerParentMessage = TerminalWorkerRequest | TerminalWorkerPtyEvent; + export type TerminalWorkerResponse = | { type: "response"; @@ -148,6 +161,14 @@ export type TerminalWorkerEvent = export type TerminalWorkerToParentMessage = TerminalWorkerResponse | TerminalWorkerEvent; +export type TerminalWorkerPtyCommand = + | { type: "ptySpawn"; processId: string; input: TerminalPtyLaunchInput } + | { type: "ptyWrite"; processId: string; data: string } + | { type: "ptyResize"; processId: string; cols: number; rows: number } + | { type: "ptyKill"; processId: string; signal?: NodeJS.Signals }; + +export type TerminalWorkerMessage = TerminalWorkerToParentMessage | TerminalWorkerPtyCommand; + export type TerminalWorkerCaptureResult = CaptureTerminalLinesResult; // The worker fills TerminalStateSnapshot.replayPreamble on getTerminalState so // the parent can cache the input-mode preamble instead of re-deriving it. diff --git a/packages/server/src/terminal/terminal-workspace-runtime.posix.test.ts b/packages/server/src/terminal/terminal-workspace-runtime.posix.test.ts new file mode 100644 index 0000000000..4f53a5a422 --- /dev/null +++ b/packages/server/src/terminal/terminal-workspace-runtime.posix.test.ts @@ -0,0 +1,378 @@ +import { chmod, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { afterEach, describe, expect, test, vi } from "vitest"; + +import { createWorkspaceRuntimeService } from "../server/workspace-runtime/index.js"; +import { createConfiguredTerminalManager } from "./terminal-manager-factory.js"; +import type { TerminalManager } from "./terminal-manager.js"; + +const posixDescribe = describe.runIf(process.platform !== "win32"); +const cleanup: Array<{ root: string; manager: TerminalManager }> = []; + +afterEach(async () => { + await Promise.all( + cleanup.splice(0).map(async ({ root, manager }) => { + await manager.killAll(); + await rm(root, { recursive: true, force: true }); + }), + ); +}); + +posixDescribe("workspace-bound terminal manager", () => { + test("keeps terminal state in Paseo while the selected runtime owns the PTY", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-runtime-")); + const cwd = path.join(root, "workspace"); + await mkdir(cwd); + const runtimeIds = new Map(); + const runtime = createWorkspaceRuntimeService({ + paseoHome: path.join(root, "home"), + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await runtime.create({ + workspaceId: "selected-workspace", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: cwd } }, + placement: { kind: "existing" }, + }); + const silentShell = path.join(root, "silent-shell"); + const silentShellReady = path.join(root, "silent-shell-ready"); + await writeFile( + silentShell, + `#!/usr/bin/env node\nprocess.stdin.setRawMode(true);process.stdin.setEncoding('utf8');require('node:fs').writeFileSync(${JSON.stringify(silentShellReady)},'ready');process.stdin.once('data',data=>{process.stdout.write(\`ready|λ|\${data.trim()}|\${process.stdout.columns}x\${process.stdout.rows}\`);process.exit(4)});\n`, + ); + await chmod(silentShell, 0o755); + const originalShell = process.env.SHELL; + process.env.SHELL = silentShell; + const manager = createConfiguredTerminalManager({ + launchPty: async (input) => + runtime.openTerminal({ + workspaceId: input.workspaceId, + argv: input.argv, + env: input.env, + purpose: { kind: "terminal", terminalId: input.terminalId }, + rows: input.rows, + cols: input.cols, + term: input.term, + }), + }); + if (originalShell === undefined) delete process.env.SHELL; + else process.env.SHELL = originalShell; + cleanup.push({ root, manager }); + + const terminal = await manager.createTerminal({ + workspaceId: "selected-workspace", + cwd, + rows: 24, + cols: 80, + }); + let output = ""; + terminal.subscribe((message) => { + if (message.type === "output") output += message.data; + }); + await vi.waitFor(async () => expect(await readFile(silentShellReady, "utf8")).toBe("ready")); + terminal.send({ type: "resize", rows: 35, cols: 97 }); + terminal.send({ type: "input", data: "héllo\n" }); + await vi.waitFor(() => expect(output).toContain("ready|λ|héllo|97x35")); + await vi.waitFor(() => expect(manager.getTerminal(terminal.id)).toBeUndefined()); + }); + + test("a selected runtime failure never falls back to a host PTY", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-runtime-failure-")); + const manager = createConfiguredTerminalManager({ + launchPty: async () => { + throw new Error("selected runtime PTY unavailable"); + }, + }); + cleanup.push({ root, manager }); + await expect(manager.createTerminal({ workspaceId: "selected", cwd: root })).rejects.toThrow( + "selected runtime PTY unavailable", + ); + }); + + test("forwards initial runtime input before a following resize without a readiness timer", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-immediate-input-")); + let input = ""; + let observeResize!: () => void; + const resized = new Promise((resolve) => { + observeResize = resolve; + }); + let finish!: () => void; + const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>( + (resolve) => { + finish = () => resolve({ code: null, signal: "SIGTERM" }); + }, + ); + const manager = createConfiguredTerminalManager({ + launchPty: async () => ({ + onData: () => () => {}, + write: (data) => { + input += data; + }, + resize: observeResize, + exited, + kill: finish, + }), + }); + cleanup.push({ root, manager }); + const terminal = await manager.createTerminal({ + workspaceId: "selected-workspace", + cwd: root, + command: process.execPath, + args: ["-e", "setInterval(()=>{},1000)"], + }); + + terminal.send({ type: "input", data: "first-input" }); + terminal.send({ type: "resize", rows: 35, cols: 97 }); + await resized; + + expect(input).toBe("first-input"); + }); + + test("an unexpected terminal-worker exit terminates its parent-owned runtime PTY", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-worker-exit-")); + const cwd = path.join(root, "workspace"); + const pidFile = path.join(root, "runtime-pty.pid"); + await mkdir(cwd); + const runtimeIds = new Map(); + const runtime = createWorkspaceRuntimeService({ + paseoHome: path.join(root, "home"), + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await runtime.create({ + workspaceId: "selected-workspace", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: cwd } }, + placement: { kind: "existing" }, + }); + const manager = createConfiguredTerminalManager({ + launchPty: async (input) => { + if (input.workspaceId === "legacy-crasher") return null; + return runtime.openTerminal({ + workspaceId: input.workspaceId, + argv: input.argv, + env: input.env, + purpose: { kind: "terminal", terminalId: input.terminalId }, + rows: input.rows, + cols: input.cols, + term: input.term, + }); + }, + }); + cleanup.push({ root, manager }); + + try { + await manager.createTerminal({ + workspaceId: "selected-workspace", + cwd, + command: process.execPath, + args: [ + "-e", + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)},String(process.pid));process.on('SIGTERM',()=>{});setInterval(()=>{},1000)`, + ], + }); + await vi.waitFor(async () => + expect(processExists(Number(await readFile(pidFile, "utf8")))).toBe(true), + ); + + await expect( + manager.createTerminal({ + workspaceId: "legacy-crasher", + cwd, + command: process.execPath, + args: ["-e", "process.kill(process.ppid, 'SIGKILL')"], + }), + ).rejects.toThrow("Terminal worker exited"); + + await vi.waitFor( + async () => expect(processExists(Number(await readFile(pidFile, "utf8")))).toBe(false), + { timeout: 5_000 }, + ); + } finally { + await runtime.destroy("selected-workspace"); + } + }, 15_000); + + test("killAll waits for parent-owned runtime PTYs to terminate", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-kill-all-")); + const cwd = path.join(root, "workspace"); + const pidFile = path.join(root, "runtime-pty.pid"); + await mkdir(cwd); + const runtimeIds = new Map(); + const runtime = createWorkspaceRuntimeService({ + paseoHome: path.join(root, "home"), + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await runtime.create({ + workspaceId: "selected-workspace", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: cwd } }, + placement: { kind: "existing" }, + }); + const manager = createConfiguredTerminalManager({ + launchPty: async (input) => + runtime.openTerminal({ + workspaceId: input.workspaceId, + argv: input.argv, + env: input.env, + purpose: { kind: "terminal", terminalId: input.terminalId }, + rows: input.rows, + cols: input.cols, + term: input.term, + }), + }); + + try { + await manager.createTerminal({ + workspaceId: "selected-workspace", + cwd, + command: process.execPath, + args: [ + "-e", + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)},String(process.pid));process.on('SIGTERM',()=>{});setInterval(()=>{},1000)`, + ], + }); + await vi.waitFor(async () => + expect(processExists(Number(await readFile(pidFile, "utf8")))).toBe(true), + ); + + await manager.killAll(); + + expect(processExists(Number(await readFile(pidFile, "utf8")))).toBe(false); + } finally { + await runtime.destroy("selected-workspace"); + await rm(root, { recursive: true, force: true }); + } + }, 15_000); + + test("a hung runtime PTY launch cannot block cleanup of an already-owned PTY", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-hung-launch-")); + const cwd = path.join(root, "workspace"); + const pidFile = path.join(root, "runtime-pty.pid"); + await mkdir(cwd); + const runtimeIds = new Map(); + const runtime = createWorkspaceRuntimeService({ + paseoHome: path.join(root, "home"), + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await runtime.create({ + workspaceId: "selected-workspace", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: cwd } }, + placement: { kind: "existing" }, + }); + let markHungLaunchStarted!: () => void; + const hungLaunchStarted = new Promise((resolve) => { + markHungLaunchStarted = resolve; + }); + const manager = createConfiguredTerminalManager({ + launchPty: async (input) => { + if (input.workspaceId === "hung-workspace") { + markHungLaunchStarted(); + return new Promise(() => {}); + } + if (input.workspaceId === "legacy-crasher") return null; + return runtime.openTerminal({ + workspaceId: input.workspaceId, + argv: input.argv, + env: input.env, + purpose: { kind: "terminal", terminalId: input.terminalId }, + rows: input.rows, + cols: input.cols, + term: input.term, + }); + }, + }); + + try { + await manager.createTerminal({ + workspaceId: "selected-workspace", + cwd, + command: process.execPath, + args: [ + "-e", + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)},String(process.pid));process.on('SIGTERM',()=>{});setInterval(()=>{},1000)`, + ], + }); + await vi.waitFor(async () => + expect(processExists(Number(await readFile(pidFile, "utf8")))).toBe(true), + ); + + await manager.createTerminal({ + workspaceId: "legacy-crasher", + cwd, + command: process.execPath, + args: ["-e", "setTimeout(()=>process.kill(process.ppid, 'SIGKILL'),500)"], + }); + + const hungCreate = manager.createTerminal({ + workspaceId: "hung-workspace", + cwd, + command: process.execPath, + args: ["-e", "setInterval(()=>{},1000)"], + }); + const hungCreateResult = hungCreate.then( + () => null, + (error: unknown) => error, + ); + await hungLaunchStarted; + const hungCreateError = await hungCreateResult; + expect(hungCreateError).toBeInstanceOf(Error); + expect((hungCreateError as Error).message).toContain("Terminal worker exited"); + await manager.killAll(); + + expect(processExists(Number(await readFile(pidFile, "utf8")))).toBe(false); + } finally { + await runtime.destroy("selected-workspace"); + await rm(root, { recursive: true, force: true }); + } + }, 15_000); + + test("killAll reports a runtime PTY that remains alive after SIGKILL", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-terminal-stuck-cleanup-")); + const manager = createConfiguredTerminalManager({ + launchPty: async () => ({ + onData: () => () => {}, + write: () => {}, + resize: () => {}, + exited: new Promise(() => {}), + kill: () => {}, + }), + }); + + try { + await manager.createTerminal({ + workspaceId: "selected-workspace", + cwd: root, + command: process.execPath, + args: ["-e", "setInterval(()=>{},1000)"], + }); + + await expect(manager.killAll()).rejects.toThrow("Runtime PTY remained alive after SIGKILL"); + } finally { + await rm(root, { recursive: true, force: true }); + } + }, 10_000); +}); + +function processExists(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } +} diff --git a/packages/server/src/terminal/terminal.ts b/packages/server/src/terminal/terminal.ts index b9d1b74017..9e38ce91b7 100644 --- a/packages/server/src/terminal/terminal.ts +++ b/packages/server/src/terminal/terminal.ts @@ -128,6 +128,74 @@ export interface CreateTerminalOptions { title?: string; command?: string; args?: string[]; + launchPty?: TerminalPtyLauncher; +} + +export interface TerminalPtyProcess { + pid: number; + write(data: string): void; + resize(cols: number, rows: number): void; + kill(signal?: NodeJS.Signals): void; + onData(listener: (data: string) => void): { dispose(): void }; + onExit(listener: (event: { exitCode: number; signal?: number }) => void): { dispose(): void }; +} + +export interface TerminalPtyLaunchInput { + terminalId: string; + workspaceId: string; + cwd: string; + argv: readonly [string, ...string[]]; + env: Record; + rows: number; + cols: number; + term: string; +} + +export type TerminalPtyLauncher = ( + input: TerminalPtyLaunchInput, +) => TerminalPtyProcess | null | Promise; + +async function launchTerminalPty(input: { + launchPty: TerminalPtyLauncher; + terminalId: string; + workspaceId: string; + cwd: string; + command: string; + args: string[] | string; + env: Record; + rows: number; + cols: number; +}): Promise { + const runtimePty = await input.launchPty({ + terminalId: input.terminalId, + workspaceId: input.workspaceId, + cwd: input.cwd, + argv: [input.command, ...(Array.isArray(input.args) ? input.args : [input.args])], + env: input.env, + rows: input.rows, + cols: input.cols, + term: "xterm-256color", + }); + if (runtimePty) return runtimePty; + return launchLegacyTerminalPty(input); +} + +function launchLegacyTerminalPty(input: { + cwd: string; + command: string; + args: string[] | string; + env: Record; + rows: number; + cols: number; +}): TerminalPtyProcess { + ensureNodePtySpawnHelperExecutableForCurrentPlatform(); + return pty.spawn(input.command, input.args, { + name: "xterm-256color", + cols: input.cols, + rows: input.rows, + cwd: input.cwd, + env: input.env, + }); } function toTerminalActivity(snapshot: { @@ -935,26 +1003,38 @@ export async function createTerminal(options: CreateTerminalOptions): Promise void): boolean { this.sentMessages.push(message); callback(null); + if (message.type === "killAll") { + queueMicrotask(() => + this.emitWorkerMessage({ type: "response", requestId: message.requestId, ok: true }), + ); + } return true; } @@ -150,7 +155,7 @@ afterEach(async () => { .catch(() => {}), ), ); - manager?.killAll(); + await manager?.killAll(); manager = null; while (temporaryDirs.length > 0) { const dir = temporaryDirs.pop(); @@ -497,7 +502,11 @@ it("injects parent-minted terminal activity env through the worker", async () => }); it("starts the default shell through the worker and accepts quoted commands", async () => { + const originalShell = process.env.SHELL; + if (!isPlatform("win32")) process.env.SHELL = "/bin/sh"; manager = createWorkerTerminalManager(); + if (originalShell === undefined) delete process.env.SHELL; + else process.env.SHELL = originalShell; const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-shell-")); temporaryDirs.push(cwd); const markerPath = join(cwd, "shell quoted marker.txt"); diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index 0f0243d7a8..7d9ed3a5da 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -12,6 +12,7 @@ import type { TerminalCommandFinishedInfo, TerminalExitInfo, TerminalSession, + TerminalPtyLaunchInput, TerminalStateSnapshot, } from "./terminal.js"; import type { CaptureTerminalLinesResult } from "./terminal-capture.js"; @@ -27,14 +28,19 @@ import type { } from "./terminal-manager.js"; import type { TerminalWorkerRequest, + TerminalWorkerParentMessage, TerminalWorkerResponse, TerminalWorkerToParentMessage, + TerminalWorkerMessage, + TerminalWorkerPtyCommand, WorkerCreateTerminalOptions, TerminalWorkerStateResult, WorkerTerminalInfo, } from "./terminal-worker-protocol.js"; const REQUEST_TIMEOUT_MS = 10000; +const RUNTIME_PTY_TERM_TIMEOUT_MS = 500; +const RUNTIME_PTY_KILL_TIMEOUT_MS = 500; type RequiredWorkerTerminalInfo = WorkerTerminalInfo & { workspaceId: string }; @@ -82,10 +88,10 @@ interface WorkerTerminalRecord { interface TerminalWorkerProcess { connected: boolean; killed: boolean; - send(message: TerminalWorkerRequest, callback: (error: Error | null) => void): boolean; + send(message: TerminalWorkerParentMessage, callback: (error: Error | null) => void): boolean; disconnect(): void; kill(): boolean; - on(event: "message", listener: (message: TerminalWorkerToParentMessage) => void): this; + on(event: "message", listener: (message: TerminalWorkerMessage) => void): this; on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; } @@ -93,8 +99,21 @@ interface WorkerTerminalManagerOptions { requestTimeoutMs?: number; forkWorker?: () => TerminalWorkerProcess; getTerminalActivityUrl?: () => string | null; + launchPty?: RuntimeTerminalLauncher; } +interface RuntimeTerminalProcess { + onData(listener: (data: string) => void): () => void; + write(data: string): void; + resize(cols: number, rows: number): void; + exited: Promise<{ code: number | null; signal: NodeJS.Signals | null }>; + kill(signal?: NodeJS.Signals): void; +} + +type RuntimeTerminalLauncher = ( + input: TerminalPtyLaunchInput, +) => Promise; + function createActivityToken(): string { return randomBytes(32).toString("base64url"); } @@ -128,6 +147,22 @@ function isResponse(message: TerminalWorkerToParentMessage): message is Terminal return message.type === "response"; } +function isPtyCommand(message: TerminalWorkerMessage): message is TerminalWorkerPtyCommand { + return ( + message.type === "ptySpawn" || + message.type === "ptyWrite" || + message.type === "ptyResize" || + message.type === "ptyKill" + ); +} + +function capturePromiseError(promise: Promise): Promise { + return promise.then( + () => null, + (error: unknown) => error, + ); +} + function cloneTerminalInfo(info: RequiredWorkerTerminalInfo): RequiredWorkerTerminalInfo { return { id: info.id, @@ -153,6 +188,7 @@ export function createWorkerTerminalManager( const worker = managerOptions.forkWorker ? managerOptions.forkWorker() : forkTerminalWorker(); const requestTimeoutMs = managerOptions.requestTimeoutMs ?? REQUEST_TIMEOUT_MS; const pendingRequests = new Map(); + const runtimePtys = new Map(); const recordsById = new Map(); const terminalIdsByCwd = new Map>(); const terminalActivityTokenById = new Map(); @@ -161,7 +197,9 @@ export function createWorkerTerminalManager( const terminalWorkspaceContributionChangedListeners = new Set(); let workerExited = false; + let shuttingDown = false; let workerShutdownTimer: ReturnType | null = null; + let workerFailureCleanup: Promise | null = null; function emitTerminalsChanged(event: TerminalsChangedEvent): void { for (const listener of Array.from(terminalsChangedListeners)) { @@ -588,7 +626,11 @@ export function createWorkerTerminalManager( } } - worker.on("message", (message: TerminalWorkerToParentMessage) => { + worker.on("message", (message: TerminalWorkerMessage) => { + if (isPtyCommand(message)) { + handlePtyCommand(message); + return; + } if (isResponse(message)) { const pending = pendingRequests.get(message.requestId); if (!pending) { @@ -606,13 +648,180 @@ export function createWorkerTerminalManager( handleWorkerEvent(message); }); - worker.on("exit", (code, signal) => { + function handlePtyCommand(message: TerminalWorkerPtyCommand): void { + if (message.type === "ptySpawn") { + void launchRuntimePty(message); + return; + } + const process = runtimePtys.get(message.processId); + if (!process) return; + if (message.type === "ptyWrite") process.write(message.data); + else if (message.type === "ptyResize") process.resize(message.cols, message.rows); + else process.kill(message.signal); + } + + async function launchRuntimePty( + message: Extract, + ): Promise { + try { + const process = (await managerOptions.launchPty?.(message.input)) ?? null; + if (!process) { + await sendWorkerMessage({ + type: "ptySpawnResult", + processId: message.processId, + mode: "legacy", + }); + return; + } + if (workerExited || shuttingDown) { + await terminateRuntimePty(process); + return; + } + runtimePtys.set(message.processId, process); + try { + await sendWorkerMessage({ + type: "ptySpawnResult", + processId: message.processId, + mode: "runtime", + }); + } catch (error) { + runtimePtys.delete(message.processId); + await terminateRuntimePty(process); + beginWorkerFailure(error); + return; + } + process.onData((data) => { + void sendWorkerMessage({ type: "ptyData", processId: message.processId, data }).catch( + beginWorkerFailure, + ); + }); + void observeRuntimePtyExit(message.processId, process); + } catch (error) { + void sendWorkerMessage({ + type: "ptySpawnError", + processId: message.processId, + error: error instanceof Error ? error.message : String(error), + }).catch(beginWorkerFailure); + } + } + + async function observeRuntimePtyExit( + processId: string, + process: RuntimeTerminalProcess, + ): Promise { + try { + const exit = await process.exited; + runtimePtys.delete(processId); + await sendWorkerMessage({ type: "ptyExit", processId, exit }); + } catch (error) { + runtimePtys.delete(processId); + try { + await sendWorkerMessage({ + type: "ptyData", + processId, + data: `${error instanceof Error ? error.message : String(error)}\r\n`, + }); + await sendWorkerMessage({ + type: "ptyExit", + processId, + exit: { code: 1, signal: null }, + }); + } catch (sendError) { + beginWorkerFailure(sendError); + } + } + } + + function sendWorkerMessage(message: TerminalWorkerParentMessage): Promise { + if (workerExited || !worker.connected) { + return Promise.reject(new Error("Terminal worker is not running")); + } + return new Promise((resolve, reject) => { + try { + worker.send(message, (error) => { + if (error) reject(error); + else resolve(); + }); + } catch (error) { + reject(error); + } + }); + } + + function beginWorkerFailure(error: unknown): void { + if (workerFailureCleanup) return; workerExited = true; + const failure = error instanceof Error ? error : new Error(String(error)); + rejectPendingRequests(failure); + for (const terminalId of Array.from(recordsById.keys())) removeRecord(terminalId); + workerFailureCleanup = capturePromiseError(cleanupAfterWorkerFailure()); + } + + async function cleanupAfterWorkerFailure(): Promise { + try { + await terminateRuntimePtys(); + } finally { + if (worker.connected) { + try { + worker.disconnect(); + } catch { + // The worker can exit between the connected check and disconnect. + } + } + } + } + + async function terminateRuntimePtys(): Promise { + const processes = Array.from(runtimePtys.values()); + runtimePtys.clear(); + const results = await Promise.allSettled(processes.map(terminateRuntimePty)); + const failures = results.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length > 0) { + const details = failures + .map((failure) => (failure instanceof Error ? failure.message : String(failure))) + .join("; "); + throw new AggregateError(failures, `Failed to terminate runtime PTYs: ${details}`); + } + } + + async function terminateRuntimePty(process: RuntimeTerminalProcess): Promise { + try { + process.kill("SIGTERM"); + } catch { + // The process can exit between the state check and the signal. + } + if (await waitForRuntimePtyExit(process, RUNTIME_PTY_TERM_TIMEOUT_MS)) return; + try { + process.kill("SIGKILL"); + } catch { + // The process can exit between the timeout and the signal. + } + if (!(await waitForRuntimePtyExit(process, RUNTIME_PTY_KILL_TIMEOUT_MS))) { + throw new Error("Runtime PTY remained alive after SIGKILL"); + } + } + + async function waitForRuntimePtyExit( + process: RuntimeTerminalProcess, + timeoutMs: number, + ): Promise { + return Promise.race([ + process.exited.then( + () => true, + () => true, + ), + new Promise((resolve) => setTimeout(() => resolve(false), timeoutMs)), + ]); + } + + worker.on("exit", (code, signal) => { if (workerShutdownTimer) { clearTimeout(workerShutdownTimer); workerShutdownTimer = null; } - rejectPendingRequests(new Error(`Terminal worker exited (${signal ?? code ?? "unknown"})`)); + beginWorkerFailure(new Error(`Terminal worker exited (${signal ?? code ?? "unknown"})`)); }); function sendRequest(input: TerminalWorkerRequestInput): Promise { @@ -813,23 +1022,45 @@ export function createWorkerTerminalManager( return Array.from(terminalIdsByCwd.keys()); }, - killAll(): void { - void sendRequest({ type: "killAll" }) - .catch(() => { - // no-op - }) - .finally(() => { - if (worker.connected) { - worker.disconnect(); - } - if (!worker.killed && !workerShutdownTimer) { - workerShutdownTimer = setTimeout(() => { - worker.kill(); - }, 1000); - } - }); - for (const terminalId of Array.from(recordsById.keys())) { - removeRecord(terminalId); + async killAll(): Promise { + shuttingDown = true; + let requestError: unknown; + let cleanupError: unknown; + try { + if (!workerExited) { + await sendRequest({ type: "killAll" }); + } + } catch (error) { + requestError = error; + beginWorkerFailure(error); + } + try { + if (workerFailureCleanup) { + cleanupError = await workerFailureCleanup; + } else { + await terminateRuntimePtys(); + } + } catch (error) { + cleanupError = error; + } finally { + for (const terminalId of Array.from(recordsById.keys())) { + removeRecord(terminalId); + } + if (worker.connected) worker.disconnect(); + if (!worker.killed && !workerShutdownTimer) { + workerShutdownTimer = setTimeout(() => { + worker.kill(); + }, 1000); + } + } + if (cleanupError) { + if (requestError) { + throw new AggregateError([requestError, cleanupError], "Terminal cleanup failed"); + } + throw cleanupError; + } + if (requestError) { + throw requestError; } }, @@ -858,6 +1089,6 @@ export function createWorkerTerminalManager( }; } -export function terminateWorkerTerminalManager(manager: TerminalManager): void { - manager.killAll(); +export function terminateWorkerTerminalManager(manager: TerminalManager): Promise { + return manager.killAll(); }