/**
* TerminalManager - Terminal session orchestration service interface.
*
* Owns terminal lifecycle operations, output fanout, and session state
* transitions for thread-scoped terminals.
*
* @module TerminalManager
*/
import {
DEFAULT_TERMINAL_ID,
TerminalCwdError,
TerminalCwdNotDirectoryError,
TerminalCwdNotFoundError,
TerminalCwdStatError,
TerminalError,
TerminalHistoryError,
TerminalNotRunningError,
TerminalProviderInstanceNotFoundError,
TerminalProviderEnvironmentError,
TerminalResizeError,
TerminalSessionLookupError,
TerminalWriteError,
type TerminalAttachInput,
type TerminalAttachStreamEvent,
type TerminalClearInput,
type TerminalCloseInput,
type TerminalEvent,
type TerminalMetadataStreamEvent,
type TerminalOpenInput,
type TerminalResizeInput,
type TerminalRestartInput,
type TerminalSessionSnapshot,
type TerminalSessionStatus,
type TerminalSummary,
type TerminalWriteInput,
ClaudeSettings,
CodexSettings,
ProviderInstanceId,
} from "@t3tools/contracts";
import { makeKeyedCoalescingWorker } from "@t3tools/shared/KeyedCoalescingWorker";
import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
import { getTerminalLabel } from "@t3tools/shared/terminalLabels";
import * as DateTime from "effect/DateTime";
import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Encoding from "effect/Encoding";
import * as Equal from "effect/Equal";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as Schema from "effect/Schema";
import * as Scope from "effect/Scope";
import * as Semaphore from "effect/Semaphore";
import * as SynchronizedRef from "effect/SynchronizedRef";
import * as ServerConfig from "../config.ts";
import { mergeProviderInstanceEnvironment } from "../provider/ProviderInstanceEnvironment.ts";
import { resolveCodexHomeLayout } from "../provider/Drivers/CodexHomeLayout.ts";
import { makeClaudeEnvironment } from "../provider/Drivers/ClaudeHome.ts";
import { deriveProviderInstanceConfigMap } from "../provider/Layers/ProviderInstanceRegistryHydration.ts";
import * as ServerSettings from "../serverSettings.ts";
import {
increment,
terminalRestartsTotal,
terminalSessionsTotal,
} from "../observability/Metrics.ts";
import { expandHomePath } from "../pathExpansion.ts";
import * as ProcessRunner from "../processRunner.ts";
import * as PortScanner from "../preview/PortScanner.ts";
import * as PtyAdapter from "./PtyAdapter.ts";
export {
TerminalCwdError,
TerminalCwdNotDirectoryError,
TerminalCwdNotFoundError,
TerminalCwdStatError,
TerminalError,
TerminalHistoryError,
TerminalNotRunningError,
TerminalProviderInstanceNotFoundError,
TerminalProviderEnvironmentError,
TerminalResizeError,
TerminalSessionLookupError,
TerminalWriteError,
};
const DEFAULT_HISTORY_LINE_LIMIT = 5_000;
const DEFAULT_HISTORY_BYTE_LIMIT = 8 * 1024 * 1024;
const MAX_HISTORY_CHUNK_LENGTH = 16 * 1024;
const DEFAULT_PERSIST_DEBOUNCE_MS = 40;
const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;
const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000;
const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128;
const DEFAULT_OPEN_COLS = 120;
const DEFAULT_OPEN_ROWS = 30;
const TERMINAL_ENV_BLOCKLIST = new Set(["PORT", "ELECTRON_RENDERER_PORT", "ELECTRON_RUN_AS_NODE"]);
const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
const MAX_TERMINAL_LABEL_LENGTH = 128;
const decodeClaudeSettings = Schema.decodeUnknownOption(ClaudeSettings);
const decodeCodexSettings = Schema.decodeUnknownOption(CodexSettings);
class TerminalSubprocessCheckError extends Schema.TaggedErrorClass<TerminalSubprocessCheckError>()(
"TerminalSubprocessCheckError",
{
cause: Schema.optional(Schema.Defect()),
command: Schema.Literals(["powershell", "ps"]),
exitCode: Schema.optional(Schema.NullOr(Schema.Number)),
timedOut: Schema.optional(Schema.Boolean),
stdoutTruncated: Schema.optional(Schema.Boolean),
},
) {
override get message(): string {
const details = [
this.exitCode !== undefined && this.exitCode !== null ? `exit code ${this.exitCode}` : null,
this.timedOut ? "timed out" : null,
this.stdoutTruncated ? "output truncated" : null,
]
.filter((detail) => detail !== null)
.join(", ");
return `Failed to inspect terminal subprocesses with ${this.command}${details.length > 0 ? ` (${details})` : ""}`;
}
}
class TerminalProcessSignalError extends Schema.TaggedErrorClass<TerminalProcessSignalError>()(
"TerminalProcessSignalError",
{
cause: Schema.optional(Schema.Defect()),
signal: Schema.Literals(["SIGTERM", "SIGKILL"]),
terminalPid: Schema.Number,
},
) {
override get message(): string {
return `Failed to send ${this.signal} to terminal process ${this.terminalPid}`;
}
}
/**
* TerminalManager - Service tag for terminal session orchestration.
*/
export class TerminalManager extends Context.Service<
TerminalManager,
{
/**
* Open or attach to a terminal session.
*
* Reuses an existing session for the same thread/terminal id and restores
* persisted history on first open.
*/
readonly open: (
input: TerminalOpenInput,
) => Effect.Effect<TerminalSessionSnapshot, TerminalError>;
/**
* Attach to a terminal and stream its initial snapshot followed by live events.
*
* Returns an unsubscribe function.
*/
readonly attachStream: (
input: TerminalAttachInput,
listener: (event: TerminalAttachStreamEvent) => Effect.Effect<void>,
) => Effect.Effect<() => void, TerminalError>;
/**
* Write input bytes to a terminal session.
*/
readonly write: (input: TerminalWriteInput) => Effect.Effect<void, TerminalError>;
/**
* Resize the PTY backing a terminal session.
*/
readonly resize: (input: TerminalResizeInput) => Effect.Effect<void, TerminalError>;
/**
* Clear terminal output history.
*/
readonly clear: (input: TerminalClearInput) => Effect.Effect<void, TerminalError>;
/**
* Restart a terminal session in place.
*
* Always resets history before spawning the new process.
*/
readonly restart: (
input: TerminalRestartInput,
) => Effect.Effect<TerminalSessionSnapshot, TerminalError>;
/**
* Close an active terminal session.
*
* When `terminalId` is omitted, closes all sessions for the thread.
*/
readonly close: (input: TerminalCloseInput) => Effect.Effect<void, TerminalError>;
/**
* Subscribe to terminal runtime events with a direct callback.
*
* Returns an unsubscribe function.
*/
readonly subscribe: (
listener: (event: TerminalEvent) => Effect.Effect<void>,
) => Effect.Effect<() => void>;
/**
* Subscribe to lightweight terminal metadata with an initial full snapshot.
*
* Returns an unsubscribe function.
*/
readonly subscribeMetadata: (
listener: (event: TerminalMetadataStreamEvent) => Effect.Effect<void>,
) => Effect.Effect<() => void>;
}
>()("t3/terminal/Manager/TerminalManager") {}
interface TerminalSubprocessInspectResult {
readonly hasRunningSubprocess: boolean;
readonly childCommand: string | null;
readonly processIds: ReadonlyArray<number>;
}
interface TerminalSubprocessInspector {
(
terminalPid: number,
): Effect.Effect<TerminalSubprocessInspectResult, TerminalSubprocessCheckError>;
}
const resizePtyProcess = (
session: TerminalSessionState,
process: PtyAdapter.PtyProcess,
cols: number,
rows: number,
) =>
Effect.try({
try: () => process.resize(cols, rows),
catch: (cause) =>
new TerminalResizeError({
threadId: session.threadId,
terminalId: session.terminalId,
terminalPid: process.pid,
cols,
rows,
cause,
}),
});
export interface ShellCandidate {
shell: string;
args?: string[];
}
export interface TerminalStartInput extends TerminalOpenInput {
cols: number;
rows: number;
}
interface TerminalSessionState {
threadId: string;
terminalId: string;
cwd: string;
worktreePath: string | null;
status: TerminalSessionStatus;
pid: number | null;
history: BoundedTerminalHistory;
pendingHistoryControlSequence: string;
pendingProcessEvents: Array<PendingProcessEvent>;
pendingProcessEventIndex: number;
processEventDrainRunning: boolean;
exitCode: number | null;
exitSignal: number | null;
updatedAt: string;
eventSequence: number;
cols: number;
rows: number;
process: PtyAdapter.PtyProcess | null;
unsubscribeData: (() => void) | null;
unsubscribeExit: (() => void) | null;
hasRunningSubprocess: boolean;
/** Normalized child command name when `hasRunningSubprocess`; cleared when idle. */
childCommandLabel: string | null;
runtimeEnv: Record<string, string> | null;
}
interface PersistHistoryRequest {
history: BoundedTerminalHistory;
immediate: boolean;
}
type PendingProcessEvent =
| { type: "output"; data: string }
| { type: "exit"; event: PtyAdapter.PtyExitEvent };
type DrainProcessEventAction =
| { type: "idle" }
| {
type: "output";
threadId: string;
terminalId: string;
sequence: number;
history: BoundedTerminalHistory | null;
data: string;
}
| {
type: "exit";
process: PtyAdapter.PtyProcess | null;
threadId: string;
terminalId: string;
sequence: number;
exitCode: number | null;
exitSignal: number | null;
};
interface TerminalManagerState {
sessions: Map<string, TerminalSessionState>;
killFibers: Map<PtyAdapter.PtyProcess, Fiber.Fiber<void, never>>;
}
function truncateTerminalWireLabel(value: string): string {
if (value.length <= MAX_TERMINAL_LABEL_LENGTH) return value;
return value.slice(0, MAX_TERMINAL_LABEL_LENGTH);
}
function normalizeChildCommandName(raw: string, platform: NodeJS.Platform): string | null {
let trimmed = raw.trim();
if (trimmed.length === 0) return null;
if (
(trimmed.startsWith("[") && trimmed.endsWith("]")) ||
(trimmed.startsWith("(") && trimmed.endsWith(")"))
) {
trimmed = trimmed.slice(1, -1).trim();
}
const firstToken = (trimmed.split(/\s+/)[0] ?? trimmed).trim();
if (firstToken.length === 0) return null;
const separators = platform === "win32" ? /[\\/]/ : /\//;
const base = firstToken.split(separators).at(-1) ?? firstToken;
const withoutExe =
platform === "win32" && base.toLowerCase().endsWith(".exe") ? base.slice(0, -4) : base;
return withoutExe.length > 0 ? withoutExe : null;
}
function terminalWireLabel(session: TerminalSessionState): string {
if (session.hasRunningSubprocess && session.childCommandLabel) {
const trimmed = session.childCommandLabel.trim();
if (trimmed.length > 0) {
return truncateTerminalWireLabel(trimmed);
}
}
return truncateTerminalWireLabel(getTerminalLabel(session.terminalId));
}
function snapshot(session: TerminalSessionState): TerminalSessionSnapshot {
return {
threadId: session.threadId,
terminalId: session.terminalId,
cwd: session.cwd,
worktreePath: session.worktreePath,
status: session.status,
pid: session.pid,
history: session.history.value(),
exitCode: session.exitCode,
exitSignal: session.exitSignal,
label: terminalWireLabel(session),
updatedAt: session.updatedAt,
sequence: session.eventSequence,
};
}
function summary(session: TerminalSessionState): TerminalSummary {
return {
threadId: session.threadId,
terminalId: session.terminalId,
cwd: session.cwd,
worktreePath: session.worktreePath,
status: session.status,
pid: session.pid,
exitCode: session.exitCode,
exitSignal: session.exitSignal,
hasRunningSubprocess: session.hasRunningSubprocess,
label: terminalWireLabel(session),
updatedAt: session.updatedAt,
};
}
function shouldPublishTerminalMetadataEvent(event: TerminalEvent): boolean {
switch (event.type) {
case "started":
case "restarted":
case "exited":
case "closed":
case "error":
case "activity":
return true;
case "output":
case "cleared":
return false;
}
}
function terminalEventToAttachEvent(event: TerminalEvent): TerminalAttachStreamEvent | null {
switch (event.type) {
case "started":
return {
type: "snapshot",
snapshot: event.snapshot,
};
case "output":
case "exited":
case "closed":
case "error":
case "cleared":
case "restarted":
case "activity":
return event;
}
}
function isDuplicateAttachSnapshotEvent(
event: TerminalEvent,
initialSnapshot: TerminalSessionSnapshot,
) {
return typeof event.sequence === "number" && typeof initialSnapshot.sequence === "number"
? event.sequence <= initialSnapshot.sequence
: event.type === "started" &&
event.snapshot.threadId === initialSnapshot.threadId &&
event.snapshot.terminalId === initialSnapshot.terminalId &&
event.snapshot.updatedAt <= initialSnapshot.updatedAt;
}
function advanceEventSequence(session: TerminalSessionState): {
readonly updatedAt: string;
readonly sequence: number;
} {
const updatedAt = DateTime.formatIso(DateTime.nowUnsafe());
session.eventSequence += 1;
session.updatedAt = updatedAt;
return { updatedAt, sequence: session.eventSequence };
}
function cleanupProcessHandles(session: TerminalSessionState): void {
session.unsubscribeData?.();
session.unsubscribeData = null;
session.unsubscribeExit?.();
session.unsubscribeExit = null;
}
function enqueueProcessEvent(
session: TerminalSessionState,
expectedPid: number,
event: PendingProcessEvent,
): boolean {
if (!session.process || session.status !== "running" || session.pid !== expectedPid) {
return false;
}
session.pendingProcessEvents.push(event);
if (session.processEventDrainRunning) {
return false;
}
session.processEventDrainRunning = true;
return true;
}
function defaultShellResolver(platform: NodeJS.Platform, env: NodeJS.ProcessEnv): string {
if (platform === "win32") {
return "pwsh.exe";
}
return env.SHELL ?? "bash";
}
function normalizeShellCommand(
value: string | undefined,
platform: NodeJS.Platform,
): string | null {
if (!value) return null;
const trimmed = value.trim();
if (trimmed.length === 0) return null;
if (platform === "win32") {
return trimmed;
}
const firstToken = trimmed.split(/\s+/g)[0]?.trim();
if (!firstToken) return null;
return firstToken.replace(/^['"]|['"]$/g, "");
}
function basenameForPlatform(command: string, platform: NodeJS.Platform): string {
const normalized =
platform === "win32" ? command.replaceAll("/", "\\") : command.replaceAll("\\", "/");
const parts = normalized
.split(platform === "win32" ? /\\+/ : /\/+/)
.filter((part) => part.length > 0);
return parts.at(-1) ?? normalized;
}
function joinWindowsPath(...parts: ReadonlyArray<string>): string {
return parts
.map((part, index) => {
if (index === 0) return part.replace(/[\\/]+$/g, "");
return part.replace(/^[\\/]+|[\\/]+$/g, "");
})
.filter((part) => part.length > 0)
.join("\\");
}
function shellCandidateFromCommand(
command: string | null,
platform: NodeJS.Platform,
): ShellCandidate | null {
if (!command || command.length === 0) return null;
const shellName = basenameForPlatform(command, platform).toLowerCase();
if (platform === "win32" && (shellName === "pwsh.exe" || shellName === "powershell.exe")) {
return { shell: command, args: ["-NoLogo"] };
}
if (platform !== "win32" && shellName === "zsh") {
return { shell: command, args: ["-o", "nopromptsp"] };
}
return { shell: command };
}
function windowsSystemRoot(env: NodeJS.ProcessEnv): string {
return env.SystemRoot?.trim() || env.windir?.trim() || "C:\\Windows";
}
function windowsPowerShellPath(env: NodeJS.ProcessEnv): string {
return joinWindowsPath(
windowsSystemRoot(env),
"System32",
"WindowsPowerShell",
"v1.0",
"powershell.exe",
);
}
function windowsCmdPath(env: NodeJS.ProcessEnv): string {
return joinWindowsPath(windowsSystemRoot(env), "System32", "cmd.exe");
}
function formatShellCandidate(candidate: ShellCandidate): string {
if (!candidate.args || candidate.args.length === 0) return candidate.shell;
return `${candidate.shell} ${candidate.args.join(" ")}`;
}
function uniqueShellCandidates(candidates: Array<ShellCandidate | null>): ShellCandidate[] {
const seen = new Set<string>();
const ordered: ShellCandidate[] = [];
for (const candidate of candidates) {
if (!candidate) continue;
const key = formatShellCandidate(candidate);
if (seen.has(key)) continue;
seen.add(key);
ordered.push(candidate);
}
return ordered;
}
function resolveShellCandidates(
shellResolver: () => string,
platform: NodeJS.Platform,
env: NodeJS.ProcessEnv,
): ShellCandidate[] {
const requested = shellCandidateFromCommand(
normalizeShellCommand(shellResolver(), platform),
platform,
);
if (platform === "win32") {
return uniqueShellCandidates([
requested,
shellCandidateFromCommand("pwsh.exe", platform),
shellCandidateFromCommand(windowsPowerShellPath(env), platform),
shellCandidateFromCommand("powershell.exe", platform),
shellCandidateFromCommand(env.ComSpec ?? null, platform),
shellCandidateFromCommand(windowsCmdPath(env), platform),
shellCandidateFromCommand("cmd.exe", platform),
]);
}
return uniqueShellCandidates([
requested,
shellCandidateFromCommand(normalizeShellCommand(env.SHELL, platform), platform),
shellCandidateFromCommand("/bin/zsh", platform),
shellCandidateFromCommand("/bin/bash", platform),
shellCandidateFromCommand("/bin/sh", platform),
shellCandidateFromCommand("zsh", platform),
shellCandidateFromCommand("bash", platform),
shellCandidateFromCommand("sh", platform),
]);
}
function isRetryableShellSpawnError(error: PtyAdapter.PtySpawnError): boolean {
const queue: unknown[] = [error];
const seen = new Set<unknown>();
const messages: string[] = [];
while (queue.length > 0) {
const current = queue.shift();
if (!current || seen.has(current)) {
continue;
}
seen.add(current);
if (typeof current === "string") {
messages.push(current);
continue;
}
if (current instanceof Error) {
messages.push(current.message);
if (current.cause) {
queue.push(current.cause);
}
continue;
}
if (typeof current === "object") {
const value = current as { message?: unknown; cause?: unknown };
if (typeof value.message === "string") {
messages.push(value.message);
}
if (value.cause) {
queue.push(value.cause);
}
}
}
const message = messages.join(" ").toLowerCase();
return (
message.includes("posix_spawnp failed") ||
message.includes("enoent") ||
message.includes("not found") ||
message.includes("file not found") ||
message.includes("no such file")
);
}
interface TerminalProcessTableSnapshot {
readonly childrenByParent: ReadonlyMap<number, ReadonlyArray<number>>;
readonly commandById: ReadonlyMap<number, string>;
}
function parsePosixProcessTable(stdout: string): TerminalProcessTableSnapshot {
const childrenByParent = new Map<number, number[]>();
const commandById = new Map<number, string>();
for (const line of stdout.split(/\r?\n/g)) {
// `comm=` is the final column and may itself contain spaces, so only the
// first two tokens are structural.
const match = /^\s*(\d+)\s+(\d+)\s+(.+)$/.exec(line);
if (!match) continue;
const pid = Number(match[1]);
const ppid = Number(match[2]);
if (!Number.isInteger(pid) || !Number.isInteger(ppid)) continue;
commandById.set(pid, (match[3] ?? "").trim());
const children = childrenByParent.get(ppid) ?? [];
children.push(pid);
childrenByParent.set(ppid, children);
}
return { childrenByParent, commandById };
}
function parseWindowsProcessTable(stdout: string): TerminalProcessTableSnapshot {
const childrenByParent = new Map<number, number[]>();
const commandById = new Map<number, string>();
for (const line of stdout.split(/\r?\n/g)) {
const [pidRaw, parentPidRaw, nameRaw] = line.trim().split("|", 3);
const pid = Number(pidRaw);
const parentPid = Number(parentPidRaw);
if (!Number.isInteger(pid) || !Number.isInteger(parentPid)) continue;
commandById.set(pid, nameRaw?.trim() ?? "");
const children = childrenByParent.get(parentPid) ?? [];
children.push(pid);
childrenByParent.set(parentPid, children);
}
return { childrenByParent, commandById };
}
function deriveSubprocessInspectResult(
snapshot: TerminalProcessTableSnapshot,
terminalPid: number,
platform: NodeJS.Platform,
): TerminalSubprocessInspectResult {
const childPid = (snapshot.childrenByParent.get(terminalPid) ?? [])[0];
if (childPid === undefined) {
return { hasRunningSubprocess: false, childCommand: null, processIds: [] };
}
const processIds = new Set<number>([terminalPid]);
const pending = [terminalPid];
while (pending.length > 0) {
const parentPid = pending.pop();
if (parentPid === undefined) continue;
for (const pid of snapshot.childrenByParent.get(parentPid) ?? []) {
if (processIds.has(pid)) continue;
processIds.add(pid);
pending.push(pid);
}
}
const normalized = normalizeChildCommandName(snapshot.commandById.get(childPid) ?? "", platform);
return {
hasRunningSubprocess: true,
childCommand: normalized ? truncateTerminalWireLabel(normalized) : null,
processIds: [...processIds],
};
}
const POSIX_PS_ABSOLUTE_PATHS = ["/bin/ps", "/usr/bin/ps"] as const;
// Resolve `ps` to an absolute path once at startup. Spawning by bare name
// walks every PATH entry per spawn (one failed posix_spawn per directory
// until the hit), which is measurable at a 1s poll cadence on long PATHs.
const resolvePosixPsCommand = Effect.fn("terminal.resolvePosixPsCommand")(function* () {
const fileSystem = yield* FileSystem.FileSystem;
for (const candidate of POSIX_PS_ABSOLUTE_PATHS) {
const exists = yield* fileSystem.exists(candidate).pipe(Effect.orElseSucceed(() => false));
if (exists) return candidate;
}
return "ps";
});
const posixProcessTableSnapshot = Effect.fn("terminal.posixProcessTableSnapshot")(function* (
psCommand: string,
): Effect.fn.Return<
TerminalProcessTableSnapshot,
TerminalSubprocessCheckError,
ProcessRunner.ProcessRunner
> {
const processRunner = yield* ProcessRunner.ProcessRunner;
const result = yield* processRunner
.run({
command: psCommand,
args: ["-eo", "pid=,ppid=,comm="],
timeout: "1 second",
maxOutputBytes: 524_288,
outputMode: "truncate",
timeoutBehavior: "timedOutResult",
})
.pipe(
Effect.mapError(
(cause) =>
new TerminalSubprocessCheckError({
cause,
command: "ps",
}),
),
);
if (result.code !== 0 || result.timedOut || result.stdoutTruncated) {
// Not authoritative: an empty or partial table would mark every terminal
// idle and clear its registered process ids. Failing skips the tick.
return yield* new TerminalSubprocessCheckError({
command: "ps",
exitCode: result.code,
timedOut: result.timedOut,
stdoutTruncated: result.stdoutTruncated,
});
}
return parsePosixProcessTable(result.stdout);
});
const windowsProcessTableSnapshot = Effect.fn("terminal.windowsProcessTableSnapshot")(
function* (): Effect.fn.Return<
TerminalProcessTableSnapshot,
TerminalSubprocessCheckError,
ProcessRunner.ProcessRunner
> {
const command =
'Get-CimInstance Win32_Process -ErrorAction Stop | ForEach-Object { Write-Output "$($_.ProcessId)|$($_.ParentProcessId)|$($_.Name)" }';
const processRunner = yield* ProcessRunner.ProcessRunner;
const result = yield* processRunner
.run({
// powershell.exe is a real executable — never spawn it through cmd.exe
// shell mode, which would re-tokenize the `-Command` payload (pipes,
// semicolons) before PowerShell ever sees it.
command: "powershell.exe",
args: ["-NoProfile", "-NonInteractive", "-Command", command],
timeout: "1500 millis",
maxOutputBytes: 262_144,
outputMode: "truncate",
timeoutBehavior: "timedOutResult",
})
.pipe(
Effect.mapError(
(cause) =>
new TerminalSubprocessCheckError({
cause,
command: "powershell",
}),
),
);
if (result.code !== 0 || result.timedOut || result.stdoutTruncated) {
// Not authoritative: an empty or partial table would mark every terminal
// idle and clear its registered process ids. Failing skips the tick.
return yield* new TerminalSubprocessCheckError({
command: "powershell",
exitCode: result.code,
timedOut: result.timedOut,
stdoutTruncated: result.stdoutTruncated,
});
}
return parseWindowsProcessTable(result.stdout);
},
);
interface TerminalHistoryChunk {
data: string;
byteLength: number;
lineBreaks: number;
}
export class BoundedTerminalHistory {
private readonly maxLines: number;
private readonly maxBytes: number;
private chunks: Array<TerminalHistoryChunk | undefined> = [];
private start = 0;
private byteLength = 0;
private lineBreaks = 0;
// Reading the old string's tail on each append can force chunk concatenation.
private lastCodeUnit: number | undefined;
private cachedValue: string | null = "";
constructor(maxLines: number, initial: string, maxBytes = DEFAULT_HISTORY_BYTE_LIMIT) {
this.maxLines = maxLines;
this.maxBytes = maxBytes;
this.append(initial);
}
append(text: string): void {
if (text.length === 0) return;
this.cachedValue = null;
if (this.maxBytes <= 0 || this.maxLines <= 0) {
this.clear();
// Preserve the existing zero-line limit's trailing newline behavior.
if (this.maxBytes > 0 && text.endsWith("\n")) this.appendChunk("\n");
return;
}
let offset = 0;
const previous = this.chunks.at(-1);
const lastCode = this.lastCodeUnit;
const firstCode = text.charCodeAt(0);
if (
previous &&
lastCode !== undefined &&
lastCode >= 0xd800 &&
lastCode <= 0xdbff &&
firstCode >= 0xdc00 &&
firstCode <= 0xdfff
) {
// Joining a split surrogate changes its UTF-8 size from 3 to 4 bytes.
previous.data += text[0];
previous.byteLength += 1;
this.byteLength += 1;
this.lastCodeUnit = firstCode;
offset = 1;
this.trim();
}
while (offset < text.length) {
let end = Math.min(offset + MAX_HISTORY_CHUNK_LENGTH, text.length);
const before = text.charCodeAt(end - 1);
const after = text.charCodeAt(end);
if (before >= 0xd800 && before <= 0xdbff && after >= 0xdc00 && after <= 0xdfff) {
end -= 1;
}
const data = text.slice(offset, end);
// Detach small chunks from large input strings so evicted prefixes can be collected.
this.appendChunk(
text.length > MAX_HISTORY_CHUNK_LENGTH
? Buffer.from(data, "utf16le").toString("utf16le")
: data,
);
this.trim();
offset = end;
}
}
private appendChunk(data: string): void {
const byteLength = Buffer.byteLength(data);
let lineBreaks = 0;
for (let index = data.indexOf("\n"); index !== -1; index = data.indexOf("\n", index + 1)) {
lineBreaks += 1;
}
const previous = this.chunks.at(-1);
if (previous && previous.data.length + data.length <= MAX_HISTORY_CHUNK_LENGTH) {
previous.data += data;
previous.byteLength += byteLength;
previous.lineBreaks += lineBreaks;
} else {
this.chunks.push({ data, byteLength, lineBreaks });
}
this.byteLength += byteLength;
this.lineBreaks += lineBreaks;
this.lastCodeUnit = data.charCodeAt(data.length - 1);
this.cachedValue = null;
}
private discardChunk(): void {
const first = this.chunks[this.start]!;
this.byteLength -= first.byteLength;
this.lineBreaks -= first.lineBreaks;
this.chunks[this.start++] = undefined;
}
private trimChunk(offset: number, byteLength: number, lineBreaks: number): void {
const first = this.chunks[this.start]!;
if (offset === first.data.length) {
this.discardChunk();
return;
}
first.data = first.data.slice(offset);
first.byteLength -= byteLength;
first.lineBreaks -= lineBreaks;
this.byteLength -= byteLength;
this.lineBreaks -= lineBreaks;
}
private trim(): void {
const trailingNewline = this.lastCodeUnit === 10;
let linesToDrop = this.lineBreaks + (trailingNewline ? 0 : 1) - this.maxLines;
while (linesToDrop > 0) {
const first = this.chunks[this.start]!;
if (first.lineBreaks < linesToDrop) {
linesToDrop -= first.lineBreaks;
this.discardChunk();
continue;
}
let offset = 0;
for (let line = 0; line < linesToDrop; line += 1) {
offset = first.data.indexOf("\n", offset) + 1;
}
this.trimChunk(offset, Buffer.byteLength(first.data.slice(0, offset)), linesToDrop);
linesToDrop = 0;
}
while (this.byteLength > this.maxBytes) {
const first = this.chunks[this.start]!;
const bytesToDrop = this.byteLength - this.maxBytes;
if (first.byteLength <= bytesToDrop) {
this.discardChunk();
continue;
}
if (first.byteLength === first.data.length && first.lineBreaks === 0) {
// ASCII without newlines needs no scan to find the byte cutoff.
this.trimChunk(bytesToDrop, bytesToDrop, 0);
continue;
}
let offset = 0;
let bytes = 0;
let lineBreaks = 0;
// Scan only the discarded prefix of one small chunk, never all history.
while (bytes < bytesToDrop) {
const codePoint = first.data.codePointAt(offset)!;
bytes += codePoint <= 0x7f ? 1 : codePoint <= 0x7ff ? 2 : codePoint <= 0xffff ? 3 : 4;
offset += codePoint <= 0xffff ? 1 : 2;
if (codePoint === 10) lineBreaks += 1;
}
this.trimChunk(offset, bytes, lineBreaks);
}
if (
this.start === this.chunks.length ||
(this.start > 2_048 && this.start * 2 >= this.chunks.length)
) {
this.chunks = this.chunks.slice(this.start);
this.start = 0;
if (this.chunks.length === 0) this.lastCodeUnit = undefined;
}
}
clear(): void {
this.chunks = [];
this.start = 0;
this.byteLength = 0;
this.lineBreaks = 0;
this.lastCodeUnit = undefined;
this.cachedValue = "";
}
value(): string {
if (this.cachedValue !== null) return this.cachedValue;
this.cachedValue = this.chunks
.slice(this.start)
.map((chunk) => chunk!.data)
.join("");
return this.cachedValue;
}
}
function isCsiFinalByte(codePoint: number): boolean {
return codePoint >= 0x40 && codePoint <= 0x7e;
}
function shouldStripCsiSequence(body: string, finalByte: string): boolean {
if (finalByte === "n") {
return true;
}
if (finalByte === "R" && /^[0-9;?]*$/.test(body)) {
return true;
}
if (finalByte === "c" && /^[>0-9;?]*$/.test(body)) {
return true;
}
// DECRQM mode queries (…$p) and DECRPM replies (…$y): replaying a stored
// query makes the terminal answer again, and the shell echoes the answer as
// junk at the prompt. The `$` guard keeps setters like DECSTR (!p) and
// DECSCL ("p) intact.
if ((finalByte === "p" || finalByte === "y") && /^[0-9;?]*\$$/.test(body)) {
return true;
}
// XTVERSION query (>q). DECSCUSR (space-intermediate q) stays.
if (finalByte === "q" && /^>[0-9;]*$/.test(body)) {
return true;
}
// Kitty keyboard protocol query/reply (?u). Restore-cursor (bare u) stays.
if (finalByte === "u" && body.startsWith("?")) {
return true;
}
return false;
}
// DECRQSS ($q) and XTGETTCAP (+q) queries plus their replies ([01]$r / [01]+r):
// pure request/response traffic with no visual value, and replaying a stored
// query triggers a fresh reply.
function shouldStripDcsSequence(content: string): boolean {
return /^[01]?[$+][qr]/.test(content);
}
function shouldStripOscSequence(content: string): boolean {
return /^(10|11|12);(?:\?|rgb:)/.test(content);
}
function stripStringTerminator(value: string): string {
if (value.endsWith("\u001b\\")) {
return value.slice(0, -2);
}
const lastCharacter = value.at(-1);
if (lastCharacter === "\u0007" || lastCharacter === "\u009c") {
return value.slice(0, -1);
}
return value;
}
function findStringTerminatorIndex(input: string, start: number): number | null {
for (let index = start; index < input.length; index += 1) {
const codePoint = input.charCodeAt(index);
if (codePoint === 0x07 || codePoint === 0x9c) {
return index + 1;
}
if (codePoint === 0x1b && input.charCodeAt(index + 1) === 0x5c) {
return index + 2;
}
}
return null;
}
function isEscapeIntermediateByte(codePoint: number): boolean {
return codePoint >= 0x20 && codePoint <= 0x2f;
}
function isEscapeFinalByte(codePoint: number): boolean {
return codePoint >= 0x30 && codePoint <= 0x7e;
}
function findEscapeSequenceEndIndex(input: string, start: number): number | null {
let cursor = start;
while (cursor < input.length && isEscapeIntermediateByte(input.charCodeAt(cursor))) {
cursor += 1;
}
if (cursor >= input.length) {
return null;
}
return isEscapeFinalByte(input.charCodeAt(cursor)) ? cursor + 1 : start + 1;
}
function sanitizeTerminalHistoryChunk(
pendingControlSequence: string,
data: string,
): { visibleText: string; pendingControlSequence: string } {
const input = `${pendingControlSequence}${data}`;
let visibleText = "";
let index = 0;
const append = (value: string) => {
visibleText += value;
};
while (index < input.length) {
const codePoint = input.charCodeAt(index);
if (codePoint === 0x1b) {
const nextCodePoint = input.charCodeAt(index + 1);
if (Number.isNaN(nextCodePoint)) {
return { visibleText, pendingControlSequence: input.slice(index) };
}
if (nextCodePoint === 0x5b) {
let cursor = index + 2;
while (cursor < input.length) {
if (isCsiFinalByte(input.charCodeAt(cursor))) {
const sequence = input.slice(index, cursor + 1);
const body = input.slice(index + 2, cursor);
if (!shouldStripCsiSequence(body, input[cursor] ?? "")) {
append(sequence);
}
index = cursor + 1;
break;
}
cursor += 1;
}
if (cursor >= input.length) {
return { visibleText, pendingControlSequence: input.slice(index) };
}
continue;
}
if (
nextCodePoint === 0x5d ||
nextCodePoint === 0x50 ||
nextCodePoint === 0x5e ||
nextCodePoint === 0x5f
) {
const terminatorIndex = findStringTerminatorIndex(input, index + 2);
if (terminatorIndex === null) {
return { visibleText, pendingControlSequence: input.slice(index) };
}
const sequence = input.slice(index, terminatorIndex);
const content = stripStringTerminator(input.slice(index + 2, terminatorIndex));
const strip =
(nextCodePoint === 0x5d && shouldStripOscSequence(content)) ||
(nextCodePoint === 0x50 && shouldStripDcsSequence(content));
if (!strip) {
append(sequence);
}
index = terminatorIndex;
continue;
}
const escapeSequenceEndIndex = findEscapeSequenceEndIndex(input, index + 1);
if (escapeSequenceEndIndex === null) {
return { visibleText, pendingControlSequence: input.slice(index) };
}
append(input.slice(index, escapeSequenceEndIndex));
index = escapeSequenceEndIndex;
continue;
}
if (codePoint === 0x9b) {
let cursor = index + 1;
while (cursor < input.length) {
if (isCsiFinalByte(input.charCodeAt(cursor))) {
const sequence = input.slice(index, cursor + 1);
const body = input.slice(index + 1, cursor);
if (!shouldStripCsiSequence(body, input[cursor] ?? "")) {
append(sequence);
}
index = cursor + 1;
break;
}
cursor += 1;
}
if (cursor >= input.length) {
return { visibleText, pendingControlSequence: input.slice(index) };
}
continue;
}
if (codePoint === 0x9d || codePoint === 0x90 || codePoint === 0x9e || codePoint === 0x9f) {
const terminatorIndex = findStringTerminatorIndex(input, index + 1);
if (terminatorIndex === null) {
return { visibleText, pendingControlSequence: input.slice(index) };
}
const sequence = input.slice(index, terminatorIndex);
const content = stripStringTerminator(input.slice(index + 1, terminatorIndex));
const strip =
(codePoint === 0x9d && shouldStripOscSequence(content)) ||
(codePoint === 0x90 && shouldStripDcsSequence(content));
if (!strip) {
append(sequence);
}
index = terminatorIndex;
continue;
}
append(input[index] ?? "");
index += 1;
}
return { visibleText, pendingControlSequence: "" };
}
function legacySafeThreadId(threadId: string): string {
return threadId.replace(/[^a-zA-Z0-9._-]/g, "_");
}
function toSafeThreadId(threadId: string): string {
return `terminal_${Encoding.encodeBase64Url(threadId)}`;
}
function toSafeTerminalId(terminalId: string): string {
return Encoding.encodeBase64Url(terminalId);
}
function toSessionKey(threadId: string, terminalId: string): string {
return `${threadId}\u0000${terminalId}`;
}
function shouldExcludeTerminalEnvKey(key: string): boolean {
const normalizedKey = key.toUpperCase();
if (normalizedKey.startsWith("T3CODE_")) {
return true;
}
if (normalizedKey.startsWith("VITE_")) {
return true;
}
return TERMINAL_ENV_BLOCKLIST.has(normalizedKey);
}
// Marker variables the AppImage runtime injects into the process it launches.
// They describe the AppImage itself, not the user's session, so terminals must
// not inherit them.
const APPIMAGE_RUNTIME_ENV_KEYS = ["APPIMAGE", "APPDIR", "ARGV0", "OWD"] as const;
// Colon-separated search-path variables the AppImage runtime points at its
// temporary mount (e.g. /tmp/.mount_T3-XXXX/usr/bin, the bundled glib schemas,
// and an $APPDIR/usr/share XDG data entry). Only the mount segments are
// dropped; the user's real entries are preserved. When nothing but mount
// segments remain the variable is removed entirely so consumers fall back to
// their platform default (e.g. gsettings finds the host schemas instead of
// reporting "No schemas installed"). See issues #1699 and #5059.
const APPIMAGE_PATH_LIKE_ENV_KEYS = [
"PATH",
"LD_LIBRARY_PATH",
"XDG_DATA_DIRS",
"GSETTINGS_SCHEMA_DIR",
] as const;
function isPathSegmentUnderAppDir(segment: string, appDir: string): boolean {
return segment === appDir || segment.startsWith(`${appDir}/`);
}
// On Linux AppImage builds the runtime mounts the app under a temporary dir and
// injects APPIMAGE/APPDIR/ARGV0/OWD plus mount entries on PATH/LD_LIBRARY_PATH.
// The integrated terminal inherits the server process environment, so without
// this scrub those leak into the PTY and tools resolve against the AppImage
// mount instead of the user's real environment (e.g. `php` reporting
// PHP_BINARY as the AppImage path). See issue #1699. The scrub is gated on an
// actual AppImage launch so non-AppImage environments are left untouched.
function stripAppImageRuntimeEnv(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv {
if (env.APPIMAGE === undefined && env.APPDIR === undefined) return env;
const scrubbed: NodeJS.ProcessEnv = { ...env };
for (const key of APPIMAGE_RUNTIME_ENV_KEYS) {
delete scrubbed[key];
}
const appDir = env.APPDIR?.replace(/\/+$/, "");
if (appDir) {
for (const key of APPIMAGE_PATH_LIKE_ENV_KEYS) {
const value = scrubbed[key];
if (value === undefined) continue;
const kept = value
.split(":")
.filter((segment) => segment.length > 0 && !isPathSegmentUnderAppDir(segment, appDir));
if (kept.length > 0) {
scrubbed[key] = kept.join(":");
} else {
delete scrubbed[key];
}
}
}
return scrubbed;
}
function createTerminalSpawnEnv(
baseEnv: NodeJS.ProcessEnv,
runtimeEnv?: Record<string, string> | null,
): NodeJS.ProcessEnv {
const spawnEnv: NodeJS.ProcessEnv = {};
for (const [key, value] of Object.entries(baseEnv)) {
if (value === undefined) continue;
if (shouldExcludeTerminalEnvKey(key)) continue;
spawnEnv[key] = value;
}
if (runtimeEnv) {
for (const [key, value] of Object.entries(runtimeEnv)) {
spawnEnv[key] =
key === "CODEX_HOME" || key === "CLAUDE_CONFIG_DIR" ? expandHomePath(value) : value;
}
}
// Both PTY backends feed truecolor-capable terminal clients.
if (spawnEnv.COLORTERM === undefined || spawnEnv.COLORTERM === "") {
spawnEnv.COLORTERM = "truecolor";
}
return stripAppImageRuntimeEnv(spawnEnv);
}
function normalizedRuntimeEnv(
env: Record<string, string> | undefined,
): Record<string, string> | null {
if (!env) return null;
const entries = Object.entries(env);
if (entries.length === 0) return null;
return Object.fromEntries(entries.toSorted(([left], [right]) => left.localeCompare(right)));
}
interface TerminalManagerOptions {
logsDir: string;
historyLineLimit?: number;
historyByteLimit?: number;
ptyAdapter: PtyAdapter.PtyAdapter["Service"];
shellResolver?: () => string;
env?: NodeJS.ProcessEnv;
subprocessInspector?: TerminalSubprocessInspector;
subprocessPollIntervalMs?: number;
processKillGraceMs?: number;
maxRetainedInactiveSessions?: number;
registerTerminalProcesses?: (input: {
readonly threadId: string;
readonly terminalId: string;
readonly processIds: ReadonlyArray<number>;
}) => Effect.Effect<void>;
unregisterTerminal?: (input: {
readonly threadId: string;
readonly terminalId: string;
}) => Effect.Effect<void>;
resolveProviderInstanceEnvironment?: (
providerInstanceId: string,
env: Record<string, string> | undefined,
) => Effect.Effect<
Record<string, string>,
TerminalProviderInstanceNotFoundError | TerminalProviderEnvironmentError
>;
}
export const resolveProviderInstanceTerminalEnvironment = Effect.fn(
"terminal.resolveProviderInstanceTerminalEnvironment",
)(function* (input: {
readonly serverSettings: ServerSettings.ServerSettingsService["Service"];
readonly path: Path.Path;
readonly rawProviderInstanceId: string;
readonly env: Record<string, string> | undefined;
}) {
const providerInstanceId = ProviderInstanceId.make(input.rawProviderInstanceId);
const settings = yield* input.serverSettings.getSettings.pipe(
Effect.mapError((cause) => new TerminalProviderEnvironmentError({ providerInstanceId, cause })),
);
const instance = deriveProviderInstanceConfigMap(settings)[providerInstanceId];
if (instance === undefined) {
return yield* new TerminalProviderInstanceNotFoundError({ providerInstanceId });
}
let resolved = mergeProviderInstanceEnvironment(instance.environment, input.env ?? {});
if (instance.driver === "codex") {
const config = decodeCodexSettings(instance.config ?? {});
if (Option.isSome(config)) {
const layout = yield* resolveCodexHomeLayout(config.value).pipe(
Effect.provideService(Path.Path, input.path),
);
if (layout.effectiveHomePath)
resolved = { ...resolved, CODEX_HOME: layout.effectiveHomePath };
}
} else if (instance.driver === "claudeAgent") {
const config = decodeClaudeSettings(instance.config ?? {});
if (Option.isSome(config)) {
resolved = yield* makeClaudeEnvironment(config.value, resolved).pipe(
Effect.provideService(Path.Path, input.path),
);
}
}
return Object.fromEntries(
Object.entries(resolved).filter((entry): entry is [string, string] => entry[1] !== undefined),
);
});
export const make = Effect.fn("TerminalManager.make")(function* () {
const { terminalLogsDir } = yield* ServerConfig.ServerConfig;
const ptyAdapter = yield* PtyAdapter.PtyAdapter;
const portDiscovery = yield* PortScanner.PortDiscovery;
const serverSettings = yield* ServerSettings.ServerSettingsService;
const path = yield* Path.Path;
const resolveProviderInstanceEnvironment = Effect.fn(
"terminal.resolveProviderInstanceEnvironment",
)((rawProviderInstanceId: string, env: Record<string, string> | undefined) =>
resolveProviderInstanceTerminalEnvironment({
serverSettings,
path,
rawProviderInstanceId,
env,
}),
);
return yield* makeWithOptions({
logsDir: terminalLogsDir,
ptyAdapter,
registerTerminalProcesses: portDiscovery.registerTerminalProcesses,
unregisterTerminal: portDiscovery.unregisterTerminal,
resolveProviderInstanceEnvironment,
});
});
export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(function* (
options: TerminalManagerOptions,
) {
const fileSystem = yield* FileSystem.FileSystem;
const path = yield* Path.Path;
const context = yield* Effect.context<never>();
const runFork = Effect.runForkWith(context);
const logsDir = options.logsDir;
const historyLineLimit = options.historyLineLimit ?? DEFAULT_HISTORY_LINE_LIMIT;
const historyByteLimit = options.historyByteLimit ?? DEFAULT_HISTORY_BYTE_LIMIT;
const platform = yield* HostProcessPlatform;
// Terminals must inherit the user's full environment (minus the blocklist
// applied in createTerminalSpawnEnv) — an allowlist here silently strips
// things like PSModulePath, DISPLAY, proxies, and toolchain variables.
// `options.env` is the test seam.
const baseEnv = options.env ?? process.env;
const shellResolver = options.shellResolver ?? (() => defaultShellResolver(platform, baseEnv));
const processRunner = yield* ProcessRunner.ProcessRunner;
const resolveLaunchInputEnvironment = Effect.fn("terminal.resolveLaunchInputEnvironment")(
function* <Input extends TerminalOpenInput | TerminalAttachInput | TerminalRestartInput>(
input: Input,
): Effect.fn.Return<
Input,
TerminalProviderInstanceNotFoundError | TerminalProviderEnvironmentError
> {
if (input.providerInstanceId === undefined) return input;
const resolver = options.resolveProviderInstanceEnvironment;
if (resolver === undefined) {
return yield* new TerminalProviderInstanceNotFoundError({
providerInstanceId: ProviderInstanceId.make(input.providerInstanceId),
});
}
const env = yield* resolver(input.providerInstanceId, input.env);
return { ...input, env };
},
);
// One process-table snapshot per poll tick, shared across every terminal.
// Per-terminal `pgrep`/`ps` calls multiply spawn load by terminal count and
// can exhaust the PID space on hosts with many sessions (#6332).
const fetchProcessTableSnapshot = (
platform === "win32"
? windowsProcessTableSnapshot()
: posixProcessTableSnapshot(yield* resolvePosixPsCommand())
).pipe(Effect.provideService(ProcessRunner.ProcessRunner, processRunner));
const customSubprocessInspector = options.subprocessInspector;
const acquireSubprocessInspector: Effect.Effect<
TerminalSubprocessInspector,
TerminalSubprocessCheckError
> =
customSubprocessInspector !== undefined
? Effect.succeed(customSubprocessInspector)
: Effect.map(
fetchProcessTableSnapshot,
(snapshot): TerminalSubprocessInspector =>
(terminalPid) =>
Effect.succeed(deriveSubprocessInspectResult(snapshot, terminalPid, platform)),
);
const subprocessPollIntervalMs =
options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS;
const processKillGraceMs = options.processKillGraceMs ?? DEFAULT_PROCESS_KILL_GRACE_MS;
const maxRetainedInactiveSessions =
options.maxRetainedInactiveSessions ?? DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS;
const registerTerminalProcesses = options.registerTerminalProcesses ?? (() => Effect.void);
const unregisterTerminal = options.unregisterTerminal ?? (() => Effect.void);
yield* fileSystem.makeDirectory(logsDir, { recursive: true }).pipe(Effect.orDie);
const managerStateRef = yield* SynchronizedRef.make<TerminalManagerState>({
sessions: new Map(),
killFibers: new Map(),
});
const threadLocksRef = yield* SynchronizedRef.make(new Map<string, Semaphore.Semaphore>());
const terminalEventListeners = new Set<(event: TerminalEvent) => Effect.Effect<void>>();
const workerScope = yield* Scope.make("sequential");
yield* Effect.addFinalizer(() => Scope.close(workerScope, Exit.void));
const publishEvent = (event: TerminalEvent) =>
Effect.gen(function* () {
for (const listener of terminalEventListeners) {
yield* listener(event).pipe(Effect.ignoreCause({ log: true }));
}
});
const historyPath = (threadId: string, terminalId: string) => {
const threadPart = toSafeThreadId(threadId);
if (terminalId === DEFAULT_TERMINAL_ID) {
return path.join(logsDir, `${threadPart}.log`);
}
return path.join(logsDir, `${threadPart}_${toSafeTerminalId(terminalId)}.log`);
};
const legacyHistoryPath = (threadId: string) =>
path.join(logsDir, `${legacySafeThreadId(threadId)}.log`);
const readManagerState = SynchronizedRef.get(managerStateRef);
const modifyManagerState = <A>(
f: (state: TerminalManagerState) => readonly [A, TerminalManagerState],
) => SynchronizedRef.modify(managerStateRef, f);
const getThreadSemaphore = (threadId: string) =>
SynchronizedRef.modifyEffect(threadLocksRef, (current) => {
const existing: Option.Option<Semaphore.Semaphore> = Option.fromNullishOr(
current.get(threadId),
);
return Option.match(existing, {
onNone: () =>
Semaphore.make(1).pipe(
Effect.map((semaphore) => {
const next = new Map(current);
next.set(threadId, semaphore);
return [semaphore, next] as const;
}),
),
onSome: (semaphore) => Effect.succeed([semaphore, current] as const),
});
});
const withThreadLock = <A, E, R>(
threadId: string,
effect: Effect.Effect<A, E, R>,
): Effect.Effect<A, E, R> =>
Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect));
const clearKillFiber = Effect.fn("terminal.clearKillFiber")(function* (
process: PtyAdapter.PtyProcess | null,
) {
if (!process) return;
const fiber: Option.Option<Fiber.Fiber<void, never>> = yield* modifyManagerState<
Option.Option<Fiber.Fiber<void, never>>
>((state) => {
const existing: Option.Option<Fiber.Fiber<void, never>> = Option.fromNullishOr(
state.killFibers.get(process),
);
if (Option.isNone(existing)) {
return [Option.none<Fiber.Fiber<void, never>>(), state] as const;
}
const killFibers = new Map(state.killFibers);
killFibers.delete(process);
return [existing, { ...state, killFibers }] as const;
});
if (Option.isSome(fiber)) {
yield* Fiber.interrupt(fiber.value).pipe(Effect.ignore);
}
});
const registerKillFiber = Effect.fn("terminal.registerKillFiber")(function* (
process: PtyAdapter.PtyProcess,
fiber: Fiber.Fiber<void, never>,
) {
yield* modifyManagerState((state) => {
const killFibers = new Map(state.killFibers);
killFibers.set(process, fiber);
return [undefined, { ...state, killFibers }] as const;
});
});
const runKillEscalation = Effect.fn("terminal.runKillEscalation")(function* (
process: PtyAdapter.PtyProcess,
threadId: string,
terminalId: string,
) {
const terminated = yield* Effect.try({
try: () => process.kill("SIGTERM"),
catch: (cause) =>
new TerminalProcessSignalError({
cause,
signal: "SIGTERM",
terminalPid: process.pid,
}),
}).pipe(
Effect.as(true),
Effect.catch((error) =>
Effect.logWarning("failed to kill terminal process", {
threadId,
terminalId,
signal: "SIGTERM",
cause: error,
}).pipe(Effect.as(false)),
),
);
if (!terminated) {
return;
}
yield* Effect.sleep(processKillGraceMs);
yield* Effect.try({
try: () => process.kill("SIGKILL"),
catch: (cause) =>
new TerminalProcessSignalError({
cause,
signal: "SIGKILL",
terminalPid: process.pid,
}),
}).pipe(
Effect.catch((error) =>
Effect.logWarning("failed to force-kill terminal process", {
threadId,
terminalId,
signal: "SIGKILL",
cause: error,
}),
),
);
});
const startKillEscalation = Effect.fn("terminal.startKillEscalation")(function* (
process: PtyAdapter.PtyProcess,
threadId: string,
terminalId: string,
) {
const fiber = yield* runKillEscalation(process, threadId, terminalId).pipe(
Effect.ensuring(
modifyManagerState((state) => {
if (!state.killFibers.has(process)) {
return [undefined, state] as const;
}
const killFibers = new Map(state.killFibers);
killFibers.delete(process);
return [undefined, { ...state, killFibers }] as const;
}),
),
Effect.forkIn(workerScope),
);
yield* registerKillFiber(process, fiber);
});
const persistWorker = yield* makeKeyedCoalescingWorker<
string,
PersistHistoryRequest,
never,
never
>({
merge: (current, next) => ({
history: next.history,
immediate: current.immediate || next.immediate,
}),
process: Effect.fn("terminal.persistHistoryWorker")(function* (sessionKey, request) {
if (!request.immediate) {
yield* Effect.sleep(DEFAULT_PERSIST_DEBOUNCE_MS);
}
const [threadId, terminalId] = sessionKey.split("\u0000");
if (!threadId || !terminalId) {
return;
}
yield* fileSystem
.writeFileString(historyPath(threadId, terminalId), request.history.value())
.pipe(
Effect.catch((error) =>
Effect.logWarning("failed to persist terminal history", {
threadId,
terminalId,
error,
}),
),
);
}),
});
const queuePersist = Effect.fn("terminal.queuePersist")(function* (
threadId: string,
terminalId: string,
history: BoundedTerminalHistory,
) {
yield* persistWorker.enqueue(toSessionKey(threadId, terminalId), {
history,
immediate: false,
});
});
const flushPersist = Effect.fn("terminal.flushPersist")(function* (
threadId: string,
terminalId: string,
) {
yield* persistWorker.drainKey(toSessionKey(threadId, terminalId));
});
const persistHistory = Effect.fn("terminal.persistHistory")(function* (
threadId: string,
terminalId: string,
history: BoundedTerminalHistory,
) {
yield* persistWorker.enqueue(toSessionKey(threadId, terminalId), {
history,
immediate: true,
});
yield* flushPersist(threadId, terminalId);
});
const readHistoryTail = Effect.fn("terminal.readHistoryTail")(function* (filePath: string) {
const file = yield* fileSystem.open(filePath, { flag: "r" });
const info = yield* file.stat;
const limit = BigInt(historyByteLimit);
const offset = info.size > limit ? info.size - limit : 0n;
yield* file.seek(offset, "start");
const bytes = new Uint8Array(Number(info.size - offset));
let length = 0;
while (length < bytes.length) {
const read = Number(yield* file.read(bytes.subarray(length)));
if (read === 0) break;
length += read;
}
let start = 0;
if (offset > 0n) {
// A tail read can start inside a UTF-8 code point. Skip its remaining bytes.
while (start < length && ((bytes[start] ?? 0) & 0xc0) === 0x80) start += 1;
}
return {
history: new TextDecoder("utf-8", { ignoreBOM: true }).decode(bytes.subarray(start, length)),
truncated: offset > 0n,
};
});
const readHistory = Effect.fn("terminal.readHistory")(function* (
threadId: string,
terminalId: string,
) {
const nextPath = historyPath(threadId, terminalId);
if (
yield* fileSystem
.exists(nextPath)
.pipe(
Effect.mapError(
(cause) => new TerminalHistoryError({ operation: "read", threadId, terminalId, cause }),
),
)
) {
const { history: raw, truncated } = yield* readHistoryTail(nextPath).pipe(
Effect.scoped,
Effect.mapError(
(cause) => new TerminalHistoryError({ operation: "read", threadId, terminalId, cause }),
),
);
const history = new BoundedTerminalHistory(historyLineLimit, raw, historyByteLimit);
const capped = history.value();
if (truncated || capped !== raw) {
yield* fileSystem
.writeFileString(nextPath, capped)
.pipe(
Effect.mapError(
(cause) =>
new TerminalHistoryError({ operation: "truncate", threadId, terminalId, cause }),
),
);
}
return history;
}
if (terminalId !== DEFAULT_TERMINAL_ID) {
return new BoundedTerminalHistory(historyLineLimit, "", historyByteLimit);
}
const legacyPath = legacyHistoryPath(threadId);
if (
!(yield* fileSystem
.exists(legacyPath)
.pipe(
Effect.mapError(
(cause) =>
new TerminalHistoryError({ operation: "migrate", threadId, terminalId, cause }),
),
))
) {
return new BoundedTerminalHistory(historyLineLimit, "", historyByteLimit);
}
const { history: raw } = yield* readHistoryTail(legacyPath).pipe(
Effect.scoped,
Effect.mapError(
(cause) => new TerminalHistoryError({ operation: "migrate", threadId, terminalId, cause }),
),
);
const history = new BoundedTerminalHistory(historyLineLimit, raw, historyByteLimit);
const capped = history.value();
yield* fileSystem
.writeFileString(nextPath, capped)
.pipe(
Effect.mapError(
(cause) =>
new TerminalHistoryError({ operation: "migrate", threadId, terminalId, cause }),
),
);
yield* fileSystem.remove(legacyPath, { force: true }).pipe(
Effect.catch((cleanupError) =>
Effect.logWarning("failed to remove legacy terminal history", {
threadId,
error: cleanupError,
}),
),
);
return history;
});
const deleteHistory = Effect.fn("terminal.deleteHistory")(function* (
threadId: string,
terminalId: string,
) {
yield* fileSystem.remove(historyPath(threadId, terminalId), { force: true }).pipe(
Effect.catch((error) =>
Effect.logWarning("failed to delete terminal history", {
threadId,
terminalId,
error,
}),
),
);
if (terminalId === DEFAULT_TERMINAL_ID) {
yield* fileSystem.remove(legacyHistoryPath(threadId), { force: true }).pipe(
Effect.catch((error) =>
Effect.logWarning("failed to delete terminal history", {
threadId,
terminalId,
error,
}),
),
);
}
});
const deleteAllHistoryForThread = Effect.fn("terminal.deleteAllHistoryForThread")(function* (
threadId: string,
) {
const threadPrefix = `${toSafeThreadId(threadId)}_`;
const entries = yield* fileSystem
.readDirectory(logsDir, { recursive: false })
.pipe(Effect.orElseSucceed(() => [] as Array<string>));
yield* Effect.forEach(
entries.filter(
(name) =>
name === `${toSafeThreadId(threadId)}.log` ||
name === `${legacySafeThreadId(threadId)}.log` ||
name.startsWith(threadPrefix),
),
(name) =>
fileSystem.remove(path.join(logsDir, name), { force: true }).pipe(
Effect.catch((error) =>
Effect.logWarning("failed to delete terminal histories for thread", {
threadId,
error,
}),
),
),
{ discard: true },
);
});
const assertValidCwd = Effect.fn("terminal.assertValidCwd")(function* (cwd: string) {
const stats = yield* fileSystem.stat(cwd).pipe(
Effect.catchTags({
PlatformError: (cause) =>
cause.reason._tag === "NotFound"
? new TerminalCwdNotFoundError({ cwd })
: new TerminalCwdStatError({ cwd, cause }),
}),
);
if (stats.type !== "Directory") {
return yield* new TerminalCwdNotDirectoryError({ cwd });
}
});
const getSession = Effect.fn("terminal.getSession")(function* (
threadId: string,
terminalId: string,
): Effect.fn.Return<Option.Option<TerminalSessionState>> {
return yield* Effect.map(readManagerState, (state) =>
Option.fromNullishOr(state.sessions.get(toSessionKey(threadId, terminalId))),
);
});
const requireSession = Effect.fn("terminal.requireSession")(function* (
threadId: string,
terminalId: string,
): Effect.fn.Return<TerminalSessionState, TerminalSessionLookupError> {
return yield* Effect.flatMap(getSession(threadId, terminalId), (session) =>
Option.match(session, {
onNone: () =>
Effect.fail(
new TerminalSessionLookupError({
threadId,
terminalId,
}),
),
onSome: Effect.succeed,
}),
);
});
const sessionsForThread = Effect.fn("terminal.sessionsForThread")(function* (threadId: string) {
return yield* readManagerState.pipe(
Effect.map((state) =>
[...state.sessions.values()].filter((session) => session.threadId === threadId),
),
);
});
const evictInactiveSessionsIfNeeded = Effect.fn("terminal.evictInactiveSessionsIfNeeded")(
function* () {
yield* modifyManagerState((state) => {
const inactiveSessions = [...state.sessions.values()].filter(
(session) => session.status !== "running",
);
if (inactiveSessions.length <= maxRetainedInactiveSessions) {
return [undefined, state] as const;
}
inactiveSessions.sort(
(left, right) =>
left.updatedAt.localeCompare(right.updatedAt) ||
left.threadId.localeCompare(right.threadId) ||
left.terminalId.localeCompare(right.terminalId),
);
const sessions = new Map(state.sessions);
const toEvict = inactiveSessions.length - maxRetainedInactiveSessions;
for (const session of inactiveSessions.slice(0, toEvict)) {
const key = toSessionKey(session.threadId, session.terminalId);
sessions.delete(key);
}
return [undefined, { ...state, sessions }] as const;
});
},
);
const drainProcessEvents = Effect.fn("terminal.drainProcessEvents")(function* (
session: TerminalSessionState,
expectedPid: number,
) {
while (true) {
const action: DrainProcessEventAction = yield* Effect.sync(() => {
if (session.pid !== expectedPid || !session.process || session.status !== "running") {
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
return { type: "idle" } as const;
}
const nextEvent = session.pendingProcessEvents[session.pendingProcessEventIndex];
if (!nextEvent) {
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
return { type: "idle" } as const;
}
session.pendingProcessEventIndex += 1;
if (session.pendingProcessEventIndex >= session.pendingProcessEvents.length) {
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
}
if (nextEvent.type === "output") {
const sanitized = sanitizeTerminalHistoryChunk(
session.pendingHistoryControlSequence,
nextEvent.data,
);
session.pendingHistoryControlSequence = sanitized.pendingControlSequence;
if (sanitized.visibleText.length > 0) {
session.history.append(sanitized.visibleText);
}
const eventStamp = advanceEventSequence(session);
return {
type: "output",
threadId: session.threadId,
terminalId: session.terminalId,
sequence: eventStamp.sequence,
history: sanitized.visibleText.length > 0 ? session.history : null,
data: nextEvent.data,
} as const;
}
const process = session.process;
cleanupProcessHandles(session);
session.process = null;
session.pid = null;
session.hasRunningSubprocess = false;
session.childCommandLabel = null;
session.status = "exited";
session.pendingHistoryControlSequence = "";
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
session.exitCode = Number.isInteger(nextEvent.event.exitCode)
? nextEvent.event.exitCode
: null;
session.exitSignal = Number.isInteger(nextEvent.event.signal)
? nextEvent.event.signal
: null;
const eventStamp = advanceEventSequence(session);
return {
type: "exit",
process,
threadId: session.threadId,
terminalId: session.terminalId,
sequence: eventStamp.sequence,
exitCode: session.exitCode,
exitSignal: session.exitSignal,
} as const;
});
if (action.type === "idle") {
return;
}
if (action.type === "output") {
if (action.history !== null) {
yield* queuePersist(action.threadId, action.terminalId, action.history);
}
yield* publishEvent({
type: "output",
threadId: action.threadId,
terminalId: action.terminalId,
sequence: action.sequence,
data: action.data,
});
continue;
}
yield* clearKillFiber(action.process);
yield* unregisterTerminal({
threadId: action.threadId,
terminalId: action.terminalId,
});
yield* publishEvent({
type: "exited",
threadId: action.threadId,
terminalId: action.terminalId,
sequence: action.sequence,
exitCode: action.exitCode,
exitSignal: action.exitSignal,
});
yield* evictInactiveSessionsIfNeeded();
return;
}
});
const stopProcess = Effect.fn("terminal.stopProcess")(function* (session: TerminalSessionState) {
const process = session.process;
if (!process) return;
const updatedAt = yield* nowIso;
yield* modifyManagerState((state) => {
cleanupProcessHandles(session);
session.process = null;
session.pid = null;
session.hasRunningSubprocess = false;
session.childCommandLabel = null;
session.status = "exited";
session.pendingHistoryControlSequence = "";
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
session.updatedAt = updatedAt;
return [undefined, state] as const;
});
yield* clearKillFiber(process);
yield* unregisterTerminal({
threadId: session.threadId,
terminalId: session.terminalId,
});
yield* startKillEscalation(process, session.threadId, session.terminalId);
yield* evictInactiveSessionsIfNeeded();
});
const trySpawn = Effect.fn("terminal.trySpawn")(function* (
shellCandidates: ReadonlyArray<ShellCandidate>,
spawnEnv: NodeJS.ProcessEnv,
session: TerminalSessionState,
index = 0,
lastError: PtyAdapter.PtySpawnError | null = null,
): Effect.fn.Return<
{ process: PtyAdapter.PtyProcess; shellLabel: string },
PtyAdapter.PtySpawnError
> {
if (index >= shellCandidates.length) {
return yield* new PtyAdapter.PtySpawnError({
adapter: "terminal-manager",
attemptedShells: shellCandidates.map((candidate) => formatShellCandidate(candidate)),
...(lastError ? { cause: lastError } : {}),
});
}
const candidate = shellCandidates[index];
if (!candidate) {
return yield* (
lastError ??
new PtyAdapter.PtySpawnError({
adapter: "terminal-manager",
attemptedShells: [],
})
);
}
const attempt = yield* Effect.result(
options.ptyAdapter.spawn({
shell: candidate.shell,
...(candidate.args ? { args: candidate.args } : {}),
cwd: session.cwd,
cols: session.cols,
rows: session.rows,
env: spawnEnv,
}),
);
if (attempt._tag === "Success") {
return {
process: attempt.success,
shellLabel: formatShellCandidate(candidate),
};
}
const spawnError = attempt.failure;
if (!isRetryableShellSpawnError(spawnError)) {
return yield* spawnError;
}
return yield* trySpawn(shellCandidates, spawnEnv, session, index + 1, spawnError);
});
const startSession = Effect.fn("terminal.startSession")(function* (
session: TerminalSessionState,
input: TerminalStartInput,
eventType: "started" | "restarted",
) {
yield* stopProcess(session);
yield* Effect.annotateCurrentSpan({
"terminal.thread_id": session.threadId,
"terminal.id": session.terminalId,
"terminal.event_type": eventType,
"terminal.cwd": input.cwd,
});
const startingAt = yield* nowIso;
yield* modifyManagerState((state) => {
session.status = "starting";
session.cwd = input.cwd;
session.worktreePath = input.worktreePath ?? null;
session.cols = input.cols;
session.rows = input.rows;
session.exitCode = null;
session.exitSignal = null;
session.hasRunningSubprocess = false;
session.childCommandLabel = null;
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
session.updatedAt = startingAt;
return [undefined, state] as const;
});
let ptyProcess: PtyAdapter.PtyProcess | null = null;
let startedShell: string | null = null;
const startResult = yield* Effect.result(
increment(terminalSessionsTotal, { lifecycle: eventType }).pipe(
Effect.andThen(
Effect.gen(function* () {
const shellCandidates = resolveShellCandidates(shellResolver, platform, baseEnv);
const terminalEnv = createTerminalSpawnEnv(baseEnv, session.runtimeEnv);
const spawnResult = yield* trySpawn(shellCandidates, terminalEnv, session);
ptyProcess = spawnResult.process;
startedShell = spawnResult.shellLabel;
const processPid = ptyProcess.pid;
const unsubscribeData = ptyProcess.onData((data) => {
if (!enqueueProcessEvent(session, processPid, { type: "output", data })) {
return;
}
runFork(drainProcessEvents(session, processPid));
});
const unsubscribeExit = ptyProcess.onExit((event) => {
if (!enqueueProcessEvent(session, processPid, { type: "exit", event })) {
return;
}
runFork(drainProcessEvents(session, processPid));
});
let eventStamp: ReturnType<typeof advanceEventSequence> = {
updatedAt: session.updatedAt,
sequence: session.eventSequence,
};
yield* modifyManagerState((state) => {
session.process = ptyProcess;
session.pid = processPid;
session.status = "running";
session.unsubscribeData = unsubscribeData;
session.unsubscribeExit = unsubscribeExit;
eventStamp = advanceEventSequence(session);
return [undefined, state] as const;
});
yield* publishEvent({
type: eventType,
threadId: session.threadId,
terminalId: session.terminalId,
sequence: eventStamp.sequence,
snapshot: snapshot(session),
});
}),
),
),
);
if (startResult._tag === "Success") {
return;
}
{
const error = startResult.failure;
if (ptyProcess) {
yield* startKillEscalation(ptyProcess, session.threadId, session.terminalId);
}
yield* modifyManagerState((state) => {
cleanupProcessHandles(session);
session.status = "error";
session.pid = null;
session.process = null;
session.hasRunningSubprocess = false;
session.childCommandLabel = null;
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
advanceEventSequence(session);
return [undefined, state] as const;
});
yield* unregisterTerminal({
threadId: session.threadId,
terminalId: session.terminalId,
});
yield* evictInactiveSessionsIfNeeded();
const message = error.message;
yield* publishEvent({
type: "error",
threadId: session.threadId,
terminalId: session.terminalId,
sequence: session.eventSequence,
message,
});
yield* Effect.logError("failed to start terminal", {
threadId: session.threadId,
terminalId: session.terminalId,
cause: error,
...(startedShell ? { shell: startedShell } : {}),
});
}
});
const closeSession = Effect.fn("terminal.closeSession")(function* (
threadId: string,
terminalId: string,
deleteHistoryOnClose: boolean,
) {
const key = toSessionKey(threadId, terminalId);
const session = yield* getSession(threadId, terminalId);
const closedEventSequence = Option.isSome(session) ? session.value.eventSequence + 1 : 0;
if (Option.isSome(session)) {
yield* stopProcess(session.value);
yield* unregisterTerminal({ threadId, terminalId });
yield* persistHistory(threadId, terminalId, session.value.history);
}
yield* flushPersist(threadId, terminalId);
const removed = yield* modifyManagerState((state) => {
if (!state.sessions.has(key)) {
return [false, state] as const;
}
const sessions = new Map(state.sessions);
sessions.delete(key);
return [true, { ...state, sessions }] as const;
});
if (removed) {
yield* publishEvent({
type: "closed",
threadId,
terminalId,
sequence: closedEventSequence,
});
}
if (deleteHistoryOnClose) {
yield* deleteHistory(threadId, terminalId);
}
});
const pollSubprocessActivity = Effect.fn("terminal.pollSubprocessActivity")(function* () {
const state = yield* readManagerState;
const runningSessions = [...state.sessions.values()].filter(
(session): session is TerminalSessionState & { pid: number } =>
session.status === "running" && Number.isInteger(session.pid),
);
if (runningSessions.length === 0) {
return;
}
const inspectorOption = yield* acquireSubprocessInspector.pipe(
Effect.map(Option.some),
Effect.catch((reason) =>
Effect.logWarning("failed to snapshot processes for terminal subprocess polling", {
reason,
}).pipe(Effect.as(Option.none<TerminalSubprocessInspector>())),
),
);
if (Option.isNone(inspectorOption)) {
return;
}
const subprocessInspector = inspectorOption.value;
const checkSubprocessActivity = Effect.fn("terminal.checkSubprocessActivity")(function* (
session: TerminalSessionState & { pid: number },
) {
const terminalPid = session.pid;
const inspectResult = yield* subprocessInspector(terminalPid).pipe(
Effect.map(Option.some),
Effect.catch((reason) =>
Effect.logWarning("failed to check terminal subprocess activity", {
threadId: session.threadId,
terminalId: session.terminalId,
terminalPid,
reason,
}).pipe(Effect.as(Option.none<TerminalSubprocessInspectResult>())),
),
);
if (Option.isNone(inspectResult)) {
return;
}
const next = inspectResult.value;
yield* registerTerminalProcesses({
threadId: session.threadId,
terminalId: session.terminalId,
processIds: next.processIds,
});
const nextChildLabel = next.hasRunningSubprocess ? next.childCommand : null;
const event = yield* modifyManagerState((state) => {
const liveSession: Option.Option<TerminalSessionState> = Option.fromNullishOr(
state.sessions.get(toSessionKey(session.threadId, session.terminalId)),
);
if (
Option.isNone(liveSession) ||
liveSession.value.status !== "running" ||
liveSession.value.pid !== terminalPid ||
(liveSession.value.hasRunningSubprocess === next.hasRunningSubprocess &&
liveSession.value.childCommandLabel === nextChildLabel)
) {
return [Option.none(), state] as const;
}
liveSession.value.hasRunningSubprocess = next.hasRunningSubprocess;
liveSession.value.childCommandLabel = nextChildLabel;
const eventStamp = advanceEventSequence(liveSession.value);
return [
Option.some({
type: "activity" as const,
threadId: liveSession.value.threadId,
terminalId: liveSession.value.terminalId,
sequence: eventStamp.sequence,
hasRunningSubprocess: next.hasRunningSubprocess,
label: terminalWireLabel(liveSession.value),
}),
state,
] as const;
});
if (Option.isSome(event)) {
yield* publishEvent(event.value);
}
});
yield* Effect.forEach(runningSessions, checkSubprocessActivity, {
concurrency: "unbounded",
discard: true,
});
});
const hasRunningSessions = readManagerState.pipe(
Effect.map((state) =>
[...state.sessions.values()].some((session) => session.status === "running"),
),
);
yield* Effect.forever(
hasRunningSessions.pipe(
Effect.flatMap((active) =>
active
? pollSubprocessActivity().pipe(
Effect.flatMap(() => Effect.sleep(subprocessPollIntervalMs)),
)
: Effect.sleep(subprocessPollIntervalMs),
),
),
).pipe(Effect.forkIn(workerScope));
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
const sessions = yield* modifyManagerState(
(state) =>
[
[...state.sessions.values()],
{
...state,
sessions: new Map(),
},
] as const,
);
const cleanupSession = Effect.fn("terminal.cleanupSession")(function* (
session: TerminalSessionState,
) {
cleanupProcessHandles(session);
if (!session.process) return;
yield* clearKillFiber(session.process);
yield* runKillEscalation(session.process, session.threadId, session.terminalId);
});
yield* Effect.forEach(sessions, cleanupSession, {
concurrency: "unbounded",
discard: true,
});
}).pipe(Effect.ignoreCause({ log: true })),
);
const openLocked = Effect.fn("terminal.openLocked")(function* (input: TerminalOpenInput) {
const terminalId = input.terminalId;
yield* assertValidCwd(input.cwd);
const sessionKey = toSessionKey(input.threadId, terminalId);
const existing = yield* getSession(input.threadId, terminalId);
if (Option.isNone(existing)) {
yield* flushPersist(input.threadId, terminalId);
const history = yield* readHistory(input.threadId, terminalId);
const cols = input.cols ?? DEFAULT_OPEN_COLS;
const rows = input.rows ?? DEFAULT_OPEN_ROWS;
const session: TerminalSessionState = {
threadId: input.threadId,
terminalId,
cwd: input.cwd,
worktreePath: input.worktreePath ?? null,
status: "starting",
pid: null,
history,
pendingHistoryControlSequence: "",
pendingProcessEvents: [],
pendingProcessEventIndex: 0,
processEventDrainRunning: false,
exitCode: null,
exitSignal: null,
updatedAt: yield* nowIso,
eventSequence: 0,
cols,
rows,
process: null,
unsubscribeData: null,
unsubscribeExit: null,
hasRunningSubprocess: false,
childCommandLabel: null,
runtimeEnv: normalizedRuntimeEnv(input.env),
};
const createdSession = session;
yield* modifyManagerState((state) => {
const sessions = new Map(state.sessions);
sessions.set(sessionKey, createdSession);
return [undefined, { ...state, sessions }] as const;
});
yield* evictInactiveSessionsIfNeeded();
yield* startSession(
session,
{
threadId: input.threadId,
terminalId,
cwd: input.cwd,
...(input.worktreePath !== undefined ? { worktreePath: input.worktreePath } : {}),
cols,
rows,
...(input.env ? { env: input.env } : {}),
},
"started",
);
return snapshot(session);
}
const liveSession = existing.value;
const nextRuntimeEnv = normalizedRuntimeEnv(input.env);
const currentRuntimeEnv = liveSession.runtimeEnv;
const targetCols = input.cols ?? liveSession.cols;
const targetRows = input.rows ?? liveSession.rows;
const runtimeEnvChanged = !Equal.equals(currentRuntimeEnv, nextRuntimeEnv);
const nextWorktreePath =
input.worktreePath !== undefined ? (input.worktreePath ?? null) : liveSession.worktreePath;
const launchContextChanged =
liveSession.cwd !== input.cwd ||
runtimeEnvChanged ||
liveSession.worktreePath !== nextWorktreePath;
if (launchContextChanged) {
yield* stopProcess(liveSession);
liveSession.cwd = input.cwd;
liveSession.worktreePath = nextWorktreePath;
liveSession.runtimeEnv = nextRuntimeEnv;
liveSession.history.clear();
liveSession.pendingHistoryControlSequence = "";
liveSession.pendingProcessEvents = [];
liveSession.pendingProcessEventIndex = 0;
liveSession.processEventDrainRunning = false;
yield* persistHistory(liveSession.threadId, liveSession.terminalId, liveSession.history);
} else if (liveSession.status === "exited" || liveSession.status === "error") {
liveSession.runtimeEnv = nextRuntimeEnv;
liveSession.worktreePath = nextWorktreePath;
liveSession.history.clear();
liveSession.pendingHistoryControlSequence = "";
liveSession.pendingProcessEvents = [];
liveSession.pendingProcessEventIndex = 0;
liveSession.processEventDrainRunning = false;
yield* persistHistory(liveSession.threadId, liveSession.terminalId, liveSession.history);
}
if (!liveSession.process) {
yield* startSession(
liveSession,
{
threadId: input.threadId,
terminalId,
cwd: input.cwd,
worktreePath: liveSession.worktreePath,
cols: targetCols,
rows: targetRows,
...(input.env ? { env: input.env } : {}),
},
"started",
);
return snapshot(liveSession);
}
if (liveSession.cols !== targetCols || liveSession.rows !== targetRows) {
yield* resizePtyProcess(liveSession, liveSession.process, targetCols, targetRows);
liveSession.cols = targetCols;
liveSession.rows = targetRows;
liveSession.updatedAt = yield* nowIso;
}
return snapshot(liveSession);
});
const open: TerminalManager["Service"]["open"] = (input) =>
withThreadLock(
input.threadId,
resolveLaunchInputEnvironment(input).pipe(Effect.flatMap(openLocked)),
);
const openOrAttachForStream = (input: TerminalAttachInput) =>
withThreadLock(
input.threadId,
Effect.gen(function* () {
const terminalId = input.terminalId;
const existing = yield* getSession(input.threadId, terminalId);
if (Option.isNone(existing)) {
if (!input.cwd) {
return yield* new TerminalSessionLookupError({
threadId: input.threadId,
terminalId,
});
}
const resolvedInput = yield* resolveLaunchInputEnvironment({
...input,
terminalId,
cwd: input.cwd,
});
return yield* openLocked(resolvedInput);
}
const session = existing.value;
const targetCols = input.cols ?? session.cols;
const targetRows = input.rows ?? session.rows;
if (!session.process && input.cwd && input.restartIfNotRunning === true) {
const resolvedInput = yield* resolveLaunchInputEnvironment({
...input,
terminalId,
cwd: input.cwd,
});
return yield* openLocked(resolvedInput);
}
if (
session.process &&
session.status === "running" &&
(session.cols !== targetCols || session.rows !== targetRows)
) {
const process = session.process;
yield* resizePtyProcess(session, process, targetCols, targetRows);
session.cols = targetCols;
session.rows = targetRows;
session.updatedAt = yield* nowIso;
}
return snapshot(session);
}),
);
const readAllTerminalMetadata = () =>
readManagerState.pipe(
Effect.map((state) =>
[...state.sessions.values()]
.map(summary)
.sort(
(left, right) =>
right.updatedAt.localeCompare(left.updatedAt) ||
left.threadId.localeCompare(right.threadId) ||
left.terminalId.localeCompare(right.terminalId),
),
),
);
const readTerminalMetadata = (input: {
readonly threadId: string;
readonly terminalId: string;
}) =>
getSession(input.threadId, input.terminalId).pipe(
Effect.map((session) => (Option.isSome(session) ? summary(session.value) : null)),
);
const subscribe: TerminalManager["Service"]["subscribe"] = (listener) =>
Effect.sync(() => {
terminalEventListeners.add(listener);
return () => {
terminalEventListeners.delete(listener);
};
});
const attachStream: TerminalManager["Service"]["attachStream"] = (input, listener) => {
let unsubscribe: (() => void) | null = null;
return Effect.gen(function* () {
const bufferedEvents: TerminalEvent[] = [];
let deliverLive = false;
unsubscribe = yield* subscribe((event) => {
if (event.threadId !== input.threadId || event.terminalId !== input.terminalId) {
return Effect.void;
}
if (!deliverLive) {
bufferedEvents.push(event);
return Effect.void;
}
const attachEvent = terminalEventToAttachEvent(event);
return attachEvent ? listener(attachEvent) : Effect.void;
});
const initialSnapshot = yield* openOrAttachForStream(input);
yield* listener({
type: "snapshot",
snapshot: initialSnapshot,
});
for (const event of bufferedEvents) {
if (isDuplicateAttachSnapshotEvent(event, initialSnapshot)) {
continue;
}
const attachEvent = terminalEventToAttachEvent(event);
if (attachEvent) {
yield* listener(attachEvent);
}
}
deliverLive = true;
return () => {
unsubscribe?.();
unsubscribe = null;
};
}).pipe(
Effect.catchCause((cause) =>
Effect.flatMap(
Effect.sync(() => {
unsubscribe?.();
unsubscribe = null;
}),
() => Effect.failCause(cause),
),
),
);
};
const metadataEventFromTerminalEvent = (
event: TerminalEvent,
): Effect.Effect<TerminalMetadataStreamEvent | null> => {
if (!shouldPublishTerminalMetadataEvent(event)) {
return Effect.succeed(null);
}
if (event.type === "closed") {
return Effect.succeed({
type: "remove" as const,
threadId: event.threadId,
terminalId: event.terminalId,
});
}
return readTerminalMetadata({
threadId: event.threadId,
terminalId: event.terminalId,
}).pipe(
Effect.map((terminal) =>
terminal
? {
type: "upsert" as const,
terminal,
}
: null,
),
);
};
const offerMetadataEvent = (
listener: (event: TerminalMetadataStreamEvent) => Effect.Effect<void>,
event: TerminalEvent,
) =>
metadataEventFromTerminalEvent(event).pipe(
Effect.flatMap((metadataEvent) => (metadataEvent ? listener(metadataEvent) : Effect.void)),
);
const subscribeMetadata: TerminalManager["Service"]["subscribeMetadata"] = (listener) => {
let unsubscribe: (() => void) | null = null;
return Effect.gen(function* () {
const bufferedEvents: TerminalEvent[] = [];
let deliverLive = false;
unsubscribe = yield* subscribe((event) => {
if (!deliverLive) {
bufferedEvents.push(event);
return Effect.void;
}
return offerMetadataEvent(listener, event);
});
const terminals = yield* readAllTerminalMetadata();
yield* listener({
type: "snapshot",
terminals,
});
for (const event of bufferedEvents) {
yield* offerMetadataEvent(listener, event);
}
deliverLive = true;
return () => {
unsubscribe?.();
unsubscribe = null;
};
}).pipe(
Effect.catchCause((cause) =>
Effect.flatMap(
Effect.sync(() => {
unsubscribe?.();
unsubscribe = null;
}),
() => Effect.failCause(cause),
),
),
);
};
const write: TerminalManager["Service"]["write"] = Effect.fn("terminal.write")(function* (input) {
const terminalId = input.terminalId;
const session = yield* requireSession(input.threadId, terminalId);
const process = session.process;
if (!process || session.status !== "running") {
if (session.status === "exited") return;
return yield* new TerminalNotRunningError({
threadId: input.threadId,
terminalId,
});
}
yield* Effect.try({
try: () => process.write(input.data),
catch: (cause) =>
new TerminalWriteError({
threadId: input.threadId,
terminalId,
terminalPid: process.pid,
cause,
}),
});
});
const resizeLocked = Effect.fn("terminal.resize")(function* (input: TerminalResizeInput) {
const session = yield* getSession(input.threadId, input.terminalId);
// ResizeObserver traffic can already be in flight when the UI closes the session.
if (Option.isNone(session)) {
return;
}
const process = session.value.process;
if (!process || session.value.status !== "running") {
return;
}
yield* resizePtyProcess(session.value, process, input.cols, input.rows);
session.value.cols = input.cols;
session.value.rows = input.rows;
session.value.updatedAt = yield* nowIso;
});
const resize: TerminalManager["Service"]["resize"] = (input) =>
withThreadLock(input.threadId, resizeLocked(input));
const clear: TerminalManager["Service"]["clear"] = (input) =>
withThreadLock(
input.threadId,
Effect.gen(function* () {
const terminalId = input.terminalId;
const session = yield* requireSession(input.threadId, terminalId);
session.history.clear();
session.pendingHistoryControlSequence = "";
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
const eventStamp = advanceEventSequence(session);
yield* persistHistory(input.threadId, terminalId, session.history);
yield* publishEvent({
type: "cleared",
threadId: input.threadId,
terminalId,
sequence: eventStamp.sequence,
});
}),
);
const restartResolved = (input: TerminalRestartInput) =>
Effect.gen(function* () {
yield* increment(terminalRestartsTotal, { scope: "thread" });
const terminalId = input.terminalId;
yield* assertValidCwd(input.cwd);
const sessionKey = toSessionKey(input.threadId, terminalId);
const existingSession = yield* getSession(input.threadId, terminalId);
let session: TerminalSessionState;
if (Option.isNone(existingSession)) {
const cols = input.cols ?? DEFAULT_OPEN_COLS;
const rows = input.rows ?? DEFAULT_OPEN_ROWS;
session = {
threadId: input.threadId,
terminalId,
cwd: input.cwd,
worktreePath: input.worktreePath ?? null,
status: "starting",
pid: null,
history: new BoundedTerminalHistory(historyLineLimit, "", historyByteLimit),
pendingHistoryControlSequence: "",
pendingProcessEvents: [],
pendingProcessEventIndex: 0,
processEventDrainRunning: false,
exitCode: null,
exitSignal: null,
updatedAt: yield* nowIso,
eventSequence: 0,
cols,
rows,
process: null,
unsubscribeData: null,
unsubscribeExit: null,
hasRunningSubprocess: false,
childCommandLabel: null,
runtimeEnv: normalizedRuntimeEnv(input.env),
};
const createdSession = session;
yield* modifyManagerState((state) => {
const sessions = new Map(state.sessions);
sessions.set(sessionKey, createdSession);
return [undefined, { ...state, sessions }] as const;
});
yield* evictInactiveSessionsIfNeeded();
} else {
session = existingSession.value;
yield* stopProcess(session);
session.cwd = input.cwd;
session.worktreePath = input.worktreePath ?? null;
session.runtimeEnv = normalizedRuntimeEnv(input.env);
}
const cols = input.cols ?? session.cols;
const rows = input.rows ?? session.rows;
session.history.clear();
session.pendingHistoryControlSequence = "";
session.pendingProcessEvents = [];
session.pendingProcessEventIndex = 0;
session.processEventDrainRunning = false;
yield* persistHistory(input.threadId, terminalId, session.history);
yield* startSession(
session,
{
threadId: input.threadId,
terminalId,
cwd: input.cwd,
...(input.worktreePath !== undefined ? { worktreePath: input.worktreePath } : {}),
cols,
rows,
...(input.env ? { env: input.env } : {}),
},
"restarted",
);
return snapshot(session);
});
const restart: TerminalManager["Service"]["restart"] = (input) =>
withThreadLock(
input.threadId,
resolveLaunchInputEnvironment(input).pipe(Effect.flatMap(restartResolved)),
);
const close: TerminalManager["Service"]["close"] = (input) =>
withThreadLock(
input.threadId,
Effect.gen(function* () {
if (input.terminalId) {
yield* closeSession(input.threadId, input.terminalId, input.deleteHistory === true);
return;
}
const threadSessions = yield* sessionsForThread(input.threadId);
yield* Effect.forEach(
threadSessions,
(session) => closeSession(input.threadId, session.terminalId, false),
{ discard: true },
);
if (input.deleteHistory) {
yield* deleteAllHistoryForThread(input.threadId);
}
}),
);
return TerminalManager.of({
open,
attachStream,
write,
resize,
clear,
restart,
close,
subscribe,
subscribeMetadata,
});
});
export const layer = Layer.effect(TerminalManager, make()).pipe(Layer.provide(ProcessRunner.layer));
