diff --git a/.plans/terminal-runtime-state-plan.md b/.plans/terminal-runtime-state-plan.md deleted file mode 100644 index 0b90488b06920394ecdaa3a6f35ac02e4a713bf1..0000000000000000000000000000000000000000 --- a/.plans/terminal-runtime-state-plan.md +++ /dev/null @@ -1,154 +0,0 @@ -# Terminal Runtime And State Plan - -## Summary - -Terminals appear to "hang" because PTY processes are exiting immediately at startup in the current runtime. - -Repro in this branch: - -- `node` + `node-pty`: interactive shell stays alive and accepts writes. -- `bun` + `node-pty`: shell exits immediately (`exitCode: 0`, `signal: 1`), so the UI opens a terminal pane but no live process remains. - -This means the current breakage is primarily a runtime compatibility issue, plus missing reconciliation between desired terminal state and observed process state. - -## Root Cause (Current) - -1. Server dev runtime uses Bun (`apps/server/package.json` `dev` script). -2. Terminal PTY implementation depends on `node-pty`. -3. Under Bun, spawned PTY shells exit immediately. -4. Frontend still mounts terminals, so UX looks like startup hang/blank terminal. - -## Design Goals - -1. Do not store full terminal output in SQLite. -2. Preserve terminal topology and intent across reconnects/restarts: - - open/split/new/close - - active terminal/group - - cwd/env profile -3. Keep runtime truth accurate even when processes change outside the app. -4. Support deterministic recovery and idempotent command handling. - -## Proposed Architecture - -### 1) Runtime Boundary - -- Run terminal PTY host on Node runtime only. -- If main server remains Bun, isolate PTY behind a Node sidecar process (JSON-RPC/IPC). -- Treat PTY host as volatile runtime, not source of durable truth. - -### 2) Event-Sourced In SQLite (Control Plane Only) - -Persist only control/state events, not output payloads. - -Recommended event families: - -- `terminal.layout.updated` - - terminal ids, split groups, active terminal id/group id -- `terminal.open.requested` - - threadId, terminalId, cwd, envHash/envProfileId, cols, rows -- `terminal.close.requested` - - terminalId, deleteHistory -- `terminal.resize.requested` - - cols, rows -- `terminal.clear.requested` -- `terminal.input.sent` - - input metadata only (`byteLength`, optional hash, optional commandBoundary flag), no raw payload -- `terminal.runtime.started` - - pid, startedAt, runtime instance id -- `terminal.runtime.exited` - - pid, exitCode, exitSignal, observedAt, reason (`normal|killed|external|crash|unknown`) -- `terminal.runtime.error` - - normalized error code/message -- `terminal.output.segment.recorded` - - segment file id + byte offsets only - -Use idempotency keys/command ids for each client command so reconnect retries are safe. - -### 3) Output Storage In Log Files (Data Plane) - -- Store raw output only in append-only files (per thread+terminal). -- Rotate into segments (size-based and/or time-based). -- Keep lightweight SQLite index rows: - - `segment_id`, `thread_id`, `terminal_id`, `start_offset`, `end_offset`, `created_at`, checksum. -- Optional retention policy (max bytes / days / segment count per terminal). - -This keeps SQLite small while preserving replay capability. - -### 4) Reconciliation Loop (Runtime Truth) - -Do not treat client dispatch as sole truth. - -Maintain dual state: - -- Desired state: from event stream/projection. -- Observed runtime state: from PTY host watchers (process exit, health checks, poll). - -On mismatch, emit runtime observation events: - -- process missing but desired running -> `terminal.runtime.exited` with reason `external|unknown` -- process exists but projection says closed -> terminate process and emit correction event - -This makes out-of-band kills/restarts explicit and recoverable. - -### 5) Startup/Recovery Flow - -1. Rebuild terminal projection from SQLite control events. -2. Load terminal layout/topology immediately for UI. -3. Reconcile each desired running terminal with runtime: - - if process alive: attach stream + tail recent output from logs - - if process dead: mark exited; auto-restart only if policy says so -4. Resume incremental output from last persisted offset. - -## Data Model Additions - -Add dedicated tables (or equivalent repos): - -- `terminal_control_events` (or reuse orchestration_events with terminal aggregate kind) -- `terminal_output_segments` (index only, no raw output) -- `terminal_runtime_snapshot` (latest observed runtime state cache) - -Avoid embedding output blobs in projection snapshots. - -## Implementation Plan - -### Phase 0: Stop the bleeding - -1. Run PTY on Node runtime in all dev/prod paths (or ship Node sidecar now). -2. Add startup diagnostics: runtime, PTY spawn result, immediate-exit telemetry. -3. Surface clear UI error when PTY host unavailable. - -### Phase 1: Durable control events - -1. Introduce terminal aggregate + commands/events. -2. Persist layout and terminal lifecycle intent events. -3. Build projection for terminal UI state from events. - -### Phase 2: File-backed output + index - -1. Implement segmented append-only output logger. -2. Persist only segment metadata/offsets in SQLite. -3. Update terminal history restore to tail from segments (bounded bytes/lines). - -### Phase 3: Runtime reconciliation - -1. Add process observer + periodic verifier. -2. Emit runtime observation events on drift. -3. Add recovery tests for out-of-band process termination. - -### Phase 4: Hardening - -1. Backpressure and truncation policies for high-volume output. -2. Crash-safe fsync/flush strategy for segment writes. -3. Retention and compaction jobs. - -## Testing Strategy - -Backend tests should cover: - -1. Rebuild projection from control events (open/split/close/resize/active terminal). -2. Runtime reconciliation when process is killed externally. -3. Recovery after restart with existing segment files. -4. Idempotent command handling on duplicate dispatches. -5. Segment index consistency and retention behavior. - -Use real runtime-layer tests for terminal adapter behavior where possible; only mock external process/service boundaries. diff --git a/apps/server/package.json b/apps/server/package.json index 89d53aa3ccc88035745590789127f69c544c2674..4978cbcd075a34de986db7a09d5683f08382b8a6 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -31,6 +31,7 @@ "@effect/vitest": "4.0.0-beta.10", "@t3tools/contracts": "workspace:*", "@t3tools/web": "workspace:*", + "@types/bun": "^1.3.9", "@types/node": "^22.10.2", "@types/ws": "^8.5.13", "tsdown": "^0.20.3", diff --git a/apps/server/src/ptyAdapter.test.ts b/apps/server/src/ptyAdapter.test.ts index 5fd37292fe2fadcea31f1ed4dfa70842dd423aab..25f30d6fe3abb274ef131c2af03f4ec29801d4b1 100644 --- a/apps/server/src/ptyAdapter.test.ts +++ b/apps/server/src/ptyAdapter.test.ts @@ -2,7 +2,9 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { assert, it, vi } from "@effect/vitest"; +import { Effect, Layer } from "effect"; +import { afterEach } from "vitest"; import { ensureNodePtySpawnHelperExecutable } from "./ptyAdapter"; @@ -10,45 +12,54 @@ function fileMode(filePath: string): number { return fs.statSync(filePath).mode & 0o777; } -describe("ensureNodePtySpawnHelperExecutable", () => { - const tempDirs: string[] = []; +const tempDirs: string[] = []; - afterEach(() => { - for (const dir of tempDirs.splice(0, tempDirs.length)) { - fs.rmSync(dir, { recursive: true, force: true }); - } - }); +afterEach(() => { + vi.unstubAllGlobals(); + for (const dir of tempDirs.splice(0, tempDirs.length)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +const helperLayer = it.layer(Layer.empty); - it("adds executable bits when helper exists but is not executable", () => { - if (process.platform === "win32") return; +helperLayer("ensureNodePtySpawnHelperExecutable", (it) => { + it.effect("adds executable bits when helper exists but is not executable", () => + Effect.sync(() => { + if (process.platform === "win32") return; - const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pty-helper-test-")); - tempDirs.push(dir); - const helperPath = path.join(dir, "spawn-helper"); - fs.writeFileSync(helperPath, "#!/bin/sh\nexit 0\n", "utf8"); - fs.chmodSync(helperPath, 0o644); + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pty-helper-test-")); + tempDirs.push(dir); + const helperPath = path.join(dir, "spawn-helper"); + fs.writeFileSync(helperPath, "#!/bin/sh\nexit 0\n", "utf8"); + fs.chmodSync(helperPath, 0o644); - ensureNodePtySpawnHelperExecutable(helperPath); + ensureNodePtySpawnHelperExecutable(helperPath); - expect(fileMode(helperPath) & 0o111).toBe(0o111); - }); + assert.equal(fileMode(helperPath) & 0o111, 0o111); + }), + ); - it("keeps executable helper as executable", () => { - if (process.platform === "win32") return; + it.effect("keeps executable helper as executable", () => + Effect.sync(() => { + if (process.platform === "win32") return; - const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pty-helper-test-")); - tempDirs.push(dir); - const helperPath = path.join(dir, "spawn-helper"); - fs.writeFileSync(helperPath, "#!/bin/sh\nexit 0\n", "utf8"); - fs.chmodSync(helperPath, 0o755); + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pty-helper-test-")); + tempDirs.push(dir); + const helperPath = path.join(dir, "spawn-helper"); + fs.writeFileSync(helperPath, "#!/bin/sh\nexit 0\n", "utf8"); + fs.chmodSync(helperPath, 0o755); - ensureNodePtySpawnHelperExecutable(helperPath); + ensureNodePtySpawnHelperExecutable(helperPath); - expect(fileMode(helperPath) & 0o111).toBe(0o111); - }); + assert.equal(fileMode(helperPath) & 0o111, 0o111); + }), + ); - it("does not throw when helper path is missing", () => { - const missing = path.join(os.tmpdir(), `does-not-exist-${Date.now()}`); - expect(() => ensureNodePtySpawnHelperExecutable(missing)).not.toThrow(); - }); + it.effect("does not throw when helper path is missing", () => + Effect.sync(() => { + const missing = path.join(os.tmpdir(), `does-not-exist-${Date.now()}`); + ensureNodePtySpawnHelperExecutable(missing); + }), + ); }); diff --git a/apps/server/src/ptyAdapter.ts b/apps/server/src/ptyAdapter.ts index 60061e04c88d3ddb13a4b151b082faf4b4c2d83f..0e655f484e5de389f034146e6090513755d7e574 100644 --- a/apps/server/src/ptyAdapter.ts +++ b/apps/server/src/ptyAdapter.ts @@ -2,7 +2,7 @@ import fs from "node:fs"; import { createRequire } from "node:module"; import path from "node:path"; -import * as nodePty from "node-pty"; +import { Effect, Layer, ServiceMap } from "effect"; export interface PtyExitEvent { exitCode: number; @@ -27,10 +27,17 @@ export interface PtySpawnInput { env: NodeJS.ProcessEnv; } -export interface PtyAdapter { +export interface PtyAdapterShape { spawn(input: PtySpawnInput): PtyProcess; } +export class PtyAdapter extends ServiceMap.Service()( + "terminal/PtyAdapter", +) { + static readonly layer = () => + typeof Bun !== "undefined" ? BunPtyAdapterLive : NodePtyAdapterLive; +} + const requireForNodePty = createRequire(import.meta.url); let didEnsureSpawnHelperExecutable = false; @@ -78,7 +85,7 @@ export function ensureNodePtySpawnHelperExecutable(explicitPath?: string): void } class NodePtyProcess implements PtyProcess { - constructor(private readonly process: nodePty.IPty) {} + constructor(private readonly process: import("node-pty").IPty) {} get pid(): number { return this.process.pid; @@ -116,16 +123,134 @@ class NodePtyProcess implements PtyProcess { } } -export class NodePtyAdapter implements PtyAdapter { - spawn(input: PtySpawnInput): PtyProcess { - ensureNodePtySpawnHelperExecutable(); - const ptyProcess = nodePty.spawn(input.shell, input.args ?? [], { - cwd: input.cwd, - cols: input.cols, - rows: input.rows, - env: input.env, - name: globalThis.process.platform === "win32" ? "xterm-color" : "xterm-256color", - }); - return new NodePtyProcess(ptyProcess); +class BunPtyProcess implements PtyProcess { + private readonly dataListeners = new Set<(data: string) => void>(); + private readonly exitListeners = new Set<(event: PtyExitEvent) => void>(); + private readonly decoder = new TextDecoder(); + private didExit = false; + + constructor(private readonly process: Bun.Subprocess) { + void this.process.exited + .then((exitCode) => { + this.emitExit({ + exitCode: Number.isInteger(exitCode) ? exitCode : 0, + signal: typeof this.process.signalCode === "number" ? this.process.signalCode : null, + }); + }) + .catch(() => { + this.emitExit({ exitCode: 1, signal: null }); + }); + } + + get pid(): number { + return this.process.pid; + } + + write(data: string): void { + if (!this.process.terminal) { + throw new Error("Bun PTY terminal handle is unavailable"); + } + this.process.terminal.write(data); + } + + resize(cols: number, rows: number): void { + if (!this.process.terminal?.resize) { + throw new Error("Bun PTY resize is unavailable"); + } + this.process.terminal.resize(cols, rows); + } + + kill(signal?: string): void { + if (!signal) { + this.process.kill(); + return; + } + this.process.kill(signal as NodeJS.Signals); + } + + onData(callback: (data: string) => void): () => void { + this.dataListeners.add(callback); + return () => { + this.dataListeners.delete(callback); + }; + } + + onExit(callback: (event: PtyExitEvent) => void): () => void { + this.exitListeners.add(callback); + return () => { + this.exitListeners.delete(callback); + }; + } + + emitData(data: Uint8Array): void { + if (this.didExit) return; + const text = this.decoder.decode(data, { stream: true }); + if (text.length === 0) return; + for (const listener of this.dataListeners) { + listener(text); + } + } + + private emitExit(event: PtyExitEvent): void { + if (this.didExit) return; + this.didExit = true; + + const remainder = this.decoder.decode(); + if (remainder.length > 0) { + for (const listener of this.dataListeners) { + listener(remainder); + } + } + + for (const listener of this.exitListeners) { + listener(event); + } } } +export const NodePtyAdapterLive = Layer.effect( + PtyAdapter, + Effect.gen(function* () { + const nodePty = yield* Effect.promise(() => import("node-pty")); + return { + spawn(input: PtySpawnInput): PtyProcess { + ensureNodePtySpawnHelperExecutable(); + const ptyProcess = nodePty.spawn(input.shell, input.args ?? [], { + cwd: input.cwd, + cols: input.cols, + rows: input.rows, + env: input.env, + name: globalThis.process.platform === "win32" ? "xterm-color" : "xterm-256color", + }); + return new NodePtyProcess(ptyProcess); + }, + }; + }), +); + +export const BunPtyAdapterLive = Layer.effect( + PtyAdapter, + Effect.gen(function* () { + if (process.platform === "win32") { + return yield* Effect.die("Bun PTY terminal support is unavailable on Windows."); + } + return { + spawn(input: PtySpawnInput): PtyProcess { + let processHandle: BunPtyProcess | null = null; + const command = [input.shell, ...(input.args ?? [])]; + const subprocess = Bun.spawn(command, { + cwd: input.cwd, + env: input.env, + terminal: { + cols: input.cols, + rows: input.rows, + data: (_terminal, data) => { + processHandle?.emitData(data); + }, + }, + }); + processHandle = new BunPtyProcess(subprocess); + return processHandle; + }, + }; + }), +); diff --git a/apps/server/src/serverLayers.ts b/apps/server/src/serverLayers.ts index 189dea33a1577aa6be3849e1fd810839c36e7ae4..e3e5e60b8543133d76910bfcfc6e09e1934b1d6c 100644 --- a/apps/server/src/serverLayers.ts +++ b/apps/server/src/serverLayers.ts @@ -26,6 +26,9 @@ import { makeProviderServiceLive } from "./provider/Layers/ProviderService"; import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory"; import { ProviderService } from "./provider/Services/ProviderService"; +import { makeTerminalManagerLive, TerminalManager } from "./terminalManager"; +import { PtyAdapter } from "./ptyAdapter"; + export function makeServerProviderLayer(): Layer.Layer< ProviderService, unknown, @@ -48,9 +51,13 @@ export function makeServerProviderLayer(): Layer.Layer< } export function makeServerRuntimeServicesLayer(): Layer.Layer< - OrchestrationEngineService | ProjectionSnapshotQuery | CheckpointStore | OrchestrationReactor, + | OrchestrationEngineService + | ProjectionSnapshotQuery + | CheckpointStore + | OrchestrationReactor + | TerminalManager, unknown, - SqlClient.SqlClient | ProviderService + SqlClient.SqlClient | ProviderService | ServerConfig > { const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionPipelineLive), @@ -79,5 +86,12 @@ export function makeServerRuntimeServicesLayer(): Layer.Layer< Layer.provideMerge(checkpointReactorLayer), ); - return orchestrationReactorLayer; + const terminalLayer = Effect.gen(function* () { + const { stateDir } = yield* ServerConfig; + return makeTerminalManagerLive({ + logsDir: path.join(stateDir, "logs", "terminals"), + }).pipe(Layer.provide(PtyAdapter.layer())); + }).pipe(Layer.unwrap); + + return Layer.merge(orchestrationReactorLayer, terminalLayer); } diff --git a/apps/server/src/terminalManager.test.ts b/apps/server/src/terminalManager.test.ts index c98b0b3a782b4811bad75e55b158e7d018adf5dc..861215d922005fb483cce65bf1a0932eaa7bcf62 100644 --- a/apps/server/src/terminalManager.test.ts +++ b/apps/server/src/terminalManager.test.ts @@ -9,8 +9,8 @@ import { } from "@t3tools/contracts"; import { afterEach, describe, expect, it } from "vitest"; -import type { PtyAdapter, PtyExitEvent, PtyProcess, PtySpawnInput } from "./ptyAdapter"; -import { TerminalManager } from "./terminalManager"; +import type { PtyAdapterShape, PtyExitEvent, PtyProcess, PtySpawnInput } from "./ptyAdapter"; +import { TerminalManagerRuntime } from "./terminalManager"; class FakePtyProcess implements PtyProcess { readonly writes: string[] = []; @@ -60,7 +60,7 @@ class FakePtyProcess implements PtyProcess { } } -class FakePtyAdapter implements PtyAdapter { +class FakePtyAdapter implements PtyAdapterShape { readonly spawnInputs: PtySpawnInput[] = []; readonly processes: FakePtyProcess[] = []; readonly spawnFailures: Error[] = []; @@ -150,7 +150,7 @@ describe("TerminalManager", () => { const logsDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-terminal-")); tempDirs.push(logsDir); const ptyAdapter = new FakePtyAdapter(); - const manager = new TerminalManager({ + const manager = new TerminalManagerRuntime({ logsDir, ptyAdapter, historyLineLimit, diff --git a/apps/server/src/terminalManager.ts b/apps/server/src/terminalManager.ts index a9e6533141a9b28e58441a67ad38640266f12268..a79722cc7cc347c16aeeb0ca67ddea609ade57de 100644 --- a/apps/server/src/terminalManager.ts +++ b/apps/server/src/terminalManager.ts @@ -13,10 +13,10 @@ import { type TerminalSessionSnapshot, type TerminalSessionStatus, } from "@t3tools/contracts"; -import { Schema } from "effect"; +import { Effect, Layer, Schema, ServiceMap } from "effect"; import { createLogger } from "./logger"; -import { NodePtyAdapter, type PtyAdapter, type PtyExitEvent, type PtyProcess } from "./ptyAdapter"; +import { PtyAdapter, PtyAdapterShape, type PtyExitEvent, type PtyProcess } from "./ptyAdapter"; import { runProcess } from "./processRunner"; const DEFAULT_HISTORY_LINE_LIMIT = 5_000; @@ -41,12 +41,27 @@ export interface TerminalManagerEvents { export interface TerminalManagerOptions { logsDir?: string; historyLineLimit?: number; - ptyAdapter?: PtyAdapter; + ptyAdapter: PtyAdapterShape; shellResolver?: () => string; subprocessChecker?: TerminalSubprocessChecker; subprocessPollIntervalMs?: number; } +export interface TerminalManagerShape { + readonly open: (input: TerminalOpenInput) => Effect.Effect; + readonly write: (input: TerminalWriteInput) => Effect.Effect; + readonly resize: (input: TerminalResizeInput) => Effect.Effect; + readonly clear: (input: TerminalClearInput) => Effect.Effect; + readonly restart: (input: TerminalOpenInput) => Effect.Effect; + readonly close: (input: TerminalCloseInput) => Effect.Effect; + readonly subscribe: (listener: (event: TerminalEvent) => void) => Effect.Effect<() => void>; + readonly dispose: () => Effect.Effect; +} + +export class TerminalManager extends ServiceMap.Service()( + "terminal/TerminalManager", +) {} + interface TerminalSessionState { threadId: string; terminalId: string; @@ -73,8 +88,8 @@ interface ShellCandidate { interface TerminalStartInput extends TerminalOpenInput { cols: number; - rows: number -}; + rows: number; +} function defaultShellResolver(): string { if (process.platform === "win32") { @@ -300,14 +315,14 @@ function normalizedRuntimeEnv( if (!env) return null; const entries = Object.entries(env); if (entries.length === 0) return null; - return Object.fromEntries(entries.sort(([left], [right]) => left.localeCompare(right))); + return Object.fromEntries(entries.toSorted(([left], [right]) => left.localeCompare(right))); } -export class TerminalManager extends EventEmitter { +export class TerminalManagerRuntime extends EventEmitter { private readonly sessions = new Map(); private readonly logsDir: string; private readonly historyLineLimit: number; - private readonly ptyAdapter: PtyAdapter; + private readonly ptyAdapter: PtyAdapterShape; private readonly shellResolver: () => string; private readonly persistQueues = new Map>(); private readonly persistTimers = new Map>(); @@ -320,11 +335,11 @@ export class TerminalManager extends EventEmitter { private subprocessPollInFlight = false; private readonly logger = createLogger("terminal"); - constructor(options: TerminalManagerOptions = {}) { + constructor(options: TerminalManagerOptions) { super(); this.logsDir = options.logsDir ?? path.resolve(process.cwd(), ".logs", "terminals"); this.historyLineLimit = options.historyLineLimit ?? DEFAULT_HISTORY_LINE_LIMIT; - this.ptyAdapter = options.ptyAdapter ?? new NodePtyAdapter(); + this.ptyAdapter = options.ptyAdapter; this.shellResolver = options.shellResolver ?? defaultShellResolver; this.persistDebounceMs = DEFAULT_PERSIST_DEBOUNCE_MS; this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker; @@ -1046,3 +1061,35 @@ export class TerminalManager extends EventEmitter { } } } + +export function makeTerminalManagerLive( + options: Omit = {}, +): Layer.Layer { + return Layer.effect( + TerminalManager, + Effect.gen(function* () { + const ptyAdapter = yield* PtyAdapter; + const runtime = new TerminalManagerRuntime({ ...options, ptyAdapter }); + + return { + open: (input) => Effect.promise(() => runtime.open(input)), + write: (input) => Effect.promise(() => runtime.write(input)), + resize: (input) => Effect.promise(() => runtime.resize(input)), + clear: (input) => Effect.promise(() => runtime.clear(input)), + restart: (input) => Effect.promise(() => runtime.restart(input)), + close: (input) => Effect.promise(() => runtime.close(input)), + subscribe: (listener) => + Effect.sync(() => { + runtime.on("event", listener); + return () => { + runtime.off("event", listener); + }; + }), + dispose: () => + Effect.sync(() => { + runtime.dispose(); + }), + } satisfies TerminalManagerShape; + }), + ); +} diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index eeaf8b1bf80f67dd63a164adcb2b90bd0756840b..4a4a2bee1e42694006bcae2bc24765b3ac7abde1 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -2,7 +2,6 @@ import * as Http from "node:http"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { EventEmitter } from "node:events"; import { Effect, Exit, Layer, PubSub, Scope, Stream } from "effect"; import { describe, expect, it, afterEach, vi } from "vitest"; @@ -38,7 +37,7 @@ import type { TerminalSessionSnapshot, TerminalWriteInput, } from "@t3tools/contracts"; -import type { TerminalManager } from "./terminalManager"; +import type { TerminalManagerShape } from "./terminalManager"; import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite"; import { SqlClient } from "effect/unstable/sql"; import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; @@ -63,111 +62,135 @@ const defaultOpenService: OpenShape = { openInEditor: () => Effect.void, }; -class MockTerminalManager extends EventEmitter<{ event: [event: TerminalEvent] }> { +class MockTerminalManager implements TerminalManagerShape { private readonly sessions = new Map(); + private readonly listeners = new Set<(event: TerminalEvent) => void>(); private key(threadId: string, terminalId: string): string { return `${threadId}\u0000${terminalId}`; } - async open(input: TerminalOpenInput): Promise { - const now = new Date().toISOString(); - const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; - const snapshot: TerminalSessionSnapshot = { - threadId: input.threadId, - terminalId, - cwd: input.cwd, - status: "running", - pid: 4242, - history: "", - exitCode: null, - exitSignal: null, - updatedAt: now, - }; - this.sessions.set(this.key(input.threadId, terminalId), snapshot); - queueMicrotask(() => { - this.emit("event", { - type: "started", + emitEvent(event: TerminalEvent): void { + for (const listener of this.listeners) { + listener(event); + } + } + + subscriptionCount(): number { + return this.listeners.size; + } + + readonly open: TerminalManagerShape["open"] = (input: TerminalOpenInput) => + Effect.sync(() => { + const now = new Date().toISOString(); + const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; + const snapshot: TerminalSessionSnapshot = { threadId: input.threadId, terminalId, - createdAt: now, - snapshot, + cwd: input.cwd, + status: "running", + pid: 4242, + history: "", + exitCode: null, + exitSignal: null, + updatedAt: now, + }; + this.sessions.set(this.key(input.threadId, terminalId), snapshot); + queueMicrotask(() => { + this.emitEvent({ + type: "started", + threadId: input.threadId, + terminalId, + createdAt: now, + snapshot, + }); }); + return snapshot; }); - return snapshot; - } - async write(input: TerminalWriteInput): Promise { - const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; - const existing = this.sessions.get(this.key(input.threadId, terminalId)); - if (!existing) { - throw new Error(`Unknown terminal thread: ${input.threadId}`); - } - queueMicrotask(() => { - this.emit("event", { - type: "output", - threadId: input.threadId, - terminalId, - createdAt: new Date().toISOString(), - data: input.data, + readonly write: TerminalManagerShape["write"] = (input: TerminalWriteInput) => + Effect.sync(() => { + const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; + const existing = this.sessions.get(this.key(input.threadId, terminalId)); + if (!existing) { + throw new Error(`Unknown terminal thread: ${input.threadId}`); + } + queueMicrotask(() => { + this.emitEvent({ + type: "output", + threadId: input.threadId, + terminalId, + createdAt: new Date().toISOString(), + data: input.data, + }); }); }); - } - async resize(_input: TerminalResizeInput): Promise {} + readonly resize: TerminalManagerShape["resize"] = (_input: TerminalResizeInput) => Effect.void; - async clear(input: TerminalClearInput): Promise { - const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; - queueMicrotask(() => { - this.emit("event", { - type: "cleared", - threadId: input.threadId, - terminalId, - createdAt: new Date().toISOString(), + readonly clear: TerminalManagerShape["clear"] = (input: TerminalClearInput) => + Effect.sync(() => { + const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; + queueMicrotask(() => { + this.emitEvent({ + type: "cleared", + threadId: input.threadId, + terminalId, + createdAt: new Date().toISOString(), + }); }); }); - } - async restart(input: TerminalOpenInput): Promise { - const now = new Date().toISOString(); - const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; - const snapshot: TerminalSessionSnapshot = { - threadId: input.threadId, - terminalId, - cwd: input.cwd, - status: "running", - pid: 5252, - history: "", - exitCode: null, - exitSignal: null, - updatedAt: now, - }; - this.sessions.set(this.key(input.threadId, terminalId), snapshot); - queueMicrotask(() => { - this.emit("event", { - type: "restarted", + readonly restart: TerminalManagerShape["restart"] = (input: TerminalOpenInput) => + Effect.sync(() => { + const now = new Date().toISOString(); + const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID; + const snapshot: TerminalSessionSnapshot = { threadId: input.threadId, terminalId, - createdAt: now, - snapshot, + cwd: input.cwd, + status: "running", + pid: 5252, + history: "", + exitCode: null, + exitSignal: null, + updatedAt: now, + }; + this.sessions.set(this.key(input.threadId, terminalId), snapshot); + queueMicrotask(() => { + this.emitEvent({ + type: "restarted", + threadId: input.threadId, + terminalId, + createdAt: now, + snapshot, + }); }); + return snapshot; }); - return snapshot; - } - async close(input: TerminalCloseInput): Promise { - if (input.terminalId) { - this.sessions.delete(this.key(input.threadId, input.terminalId)); - return; - } - for (const key of [...this.sessions.keys()]) { - if (key.startsWith(`${input.threadId}\u0000`)) { - this.sessions.delete(key); + readonly close: TerminalManagerShape["close"] = (input: TerminalCloseInput) => + Effect.sync(() => { + if (input.terminalId) { + this.sessions.delete(this.key(input.threadId, input.terminalId)); + return; } - } - } + for (const key of [...this.sessions.keys()]) { + if (key.startsWith(`${input.threadId}\u0000`)) { + this.sessions.delete(key); + } + } + }); + + readonly subscribe: TerminalManagerShape["subscribe"] = (listener) => + Effect.sync(() => { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + }); - dispose(): void {} + readonly dispose: TerminalManagerShape["dispose"] = () => Effect.void; } function connectWs(port: number, token?: string): Promise { @@ -320,7 +343,7 @@ describe("WebSocket Server", () => { status: (input: { cwd: string }) => Promise; runStackedAction: (input: { cwd: string; action: string }) => Promise; }; - terminalManager?: TerminalManager; + terminalManager?: TerminalManagerShape; } = {}, ): Promise { if (serverScope) { @@ -990,7 +1013,7 @@ describe("WebSocket Server", () => { const terminalManager = new MockTerminalManager(); server = await createTestServer({ cwd: "/test", - terminalManager: terminalManager as unknown as TerminalManager, + terminalManager, }); const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; @@ -1048,7 +1071,7 @@ describe("WebSocket Server", () => { createdAt: new Date().toISOString(), data: "manual test output\n", }; - terminalManager.emit("event", manualEvent); + terminalManager.emitEvent(manualEvent); const push = (await waitForMessage(ws)) as WsPush; expect(push.type).toBe("push"); @@ -1060,15 +1083,15 @@ describe("WebSocket Server", () => { const terminalManager = new MockTerminalManager(); server = await createTestServer({ cwd: "/test", - terminalManager: terminalManager as unknown as TerminalManager, + terminalManager, }); - expect(terminalManager.listenerCount("event")).toBe(1); + expect(terminalManager.subscriptionCount()).toBe(1); await closeTestServer(); server = null; - expect(terminalManager.listenerCount("event")).toBe(0); + expect(terminalManager.subscriptionCount()).toBe(0); }); it("returns validation errors for invalid terminal open params", async () => { diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index d4afd8e714432fb89255a4955f26c32ff7c9d64d..9ff5d0c84a7e73f8b4f54c367c82cb8e6c17754c 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -30,7 +30,7 @@ import { pullGitBranch, removeGitWorktree, } from "./git"; -import { TerminalManager } from "./terminalManager"; +import { TerminalManager, type TerminalManagerShape } from "./terminalManager"; import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings"; import { searchWorkspaceEntries } from "./workspaceEntries"; import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine"; @@ -61,7 +61,7 @@ export interface ServerOptions { autoBootstrapProjectFromCwd?: boolean | undefined; logWebSocketEvents?: boolean | undefined; gitManager?: GitManager | undefined; - terminalManager?: TerminalManager | undefined; + terminalManager?: TerminalManagerShape | undefined; } export interface ServerShape { @@ -152,7 +152,7 @@ export type ServerCoreRuntimeServices = | OrchestrationReactor | Open; -export type ServerRuntimeServices = ServerCoreRuntimeServices | ProviderService; +export type ServerRuntimeServices = ServerCoreRuntimeServices | ProviderService | TerminalManager; export class ServerLifecycleError extends Schema.TaggedErrorClass()( "ServerLifecycleError", @@ -172,10 +172,11 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E Scope.Scope | ServerRuntimeServices | ServerConfig > { const { port, cwd, staticDir, devUrl, authToken, mode, host } = yield* ServerConfig; + const defaultTerminalManager = yield* TerminalManager; const { logWebSocketEvents: explicitLogWsEvents, gitManager = new GitManager(), - terminalManager = new TerminalManager(), + terminalManager = defaultTerminalManager, } = options; const autoBootstrapProjectFromCwd = options.autoBootstrapProjectFromCwd ?? mode === "web"; @@ -541,32 +542,32 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E case WS_METHODS.terminalOpen: { const body = stripRequestTag(request.body); - return yield* Effect.promise(() => terminalManager.open(body)); + return yield* terminalManager.open(body); } case WS_METHODS.terminalWrite: { const body = stripRequestTag(request.body); - return yield* Effect.promise(() => terminalManager.write(body)); + return yield* terminalManager.write(body); } case WS_METHODS.terminalResize: { const body = stripRequestTag(request.body); - return yield* Effect.promise(() => terminalManager.resize(body)); + return yield* terminalManager.resize(body); } case WS_METHODS.terminalClear: { const body = stripRequestTag(request.body); - return yield* Effect.promise(() => terminalManager.clear(body)); + return yield* terminalManager.clear(body); } case WS_METHODS.terminalRestart: { const body = stripRequestTag(request.body); - return yield* Effect.promise(() => terminalManager.restart(body)); + return yield* terminalManager.restart(body); } case WS_METHODS.terminalClose: { const body = stripRequestTag(request.body); - return yield* Effect.promise(() => terminalManager.close(body)); + return yield* terminalManager.close(body); } case WS_METHODS.serverGetConfig: @@ -690,13 +691,15 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E ), ); - terminalManager.on("event", onTerminalEvent); + const unsubscribeTerminalEvents = yield* terminalManager.subscribe(onTerminalEvent); yield* Effect.addFinalizer(() => Effect.sync(() => { - terminalManager.off("event", onTerminalEvent); - terminalManager.dispose(); + unsubscribeTerminalEvents(); }), ); + yield* Effect.addFinalizer(() => + terminalManager.dispose().pipe(Effect.orElseSucceed(() => undefined)), + ); yield* NodeHttpServer.make(() => httpServer, listenOptions); diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index b99ca864e7780a800a53ac25fff4077b3cc0d399..db897c75cadad619856b4dd0a49ea48536cea82e 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -2,7 +2,7 @@ "extends": "../../tsconfig.base.json", "compilerOptions": { "composite": true, - "types": ["node"], + "types": ["node", "bun"], "lib": ["ES2023", "esnext.disposable"], "noEmit": true, "allowImportingTsExtensions": true, diff --git a/bun.lock b/bun.lock index 6e316d6e4ee10c659038f01aef0c4c14424fa60b..6c77bf45e5917f560fab90d7e4b0c41dccdb2bed 100644 --- a/bun.lock +++ b/bun.lock @@ -48,6 +48,7 @@ "@effect/vitest": "4.0.0-beta.10", "@t3tools/contracts": "workspace:*", "@t3tools/web": "workspace:*", + "@types/bun": "^1.3.9", "@types/node": "^22.10.2", "@types/ws": "^8.5.13", "tsdown": "^0.20.3", @@ -527,6 +528,8 @@ "@types/babel__traverse": ["@types/babel__traverse@7.28.0", "", { "dependencies": { "@babel/types": "^7.28.2" } }, "sha512-8PvcXf70gTDZBgt9ptxJ8elBeBjcLOAcOtoO/mPJjtji1+CdGbHgm77om1GrsPxsiE+uXIpNSK64UYaIwQXd4Q=="], + "@types/bun": ["@types/bun@1.3.9", "", { "dependencies": { "bun-types": "1.3.9" } }, "sha512-KQ571yULOdWJiMH+RIWIOZ7B2RXQGpL1YQrBtLIV3FqDcCu6FsbFUBwhdKUlCKUpS3PJDsHlJ1QKlpxoVR+xtw=="], + "@types/cacheable-request": ["@types/cacheable-request@6.0.3", "", { "dependencies": { "@types/http-cache-semantics": "*", "@types/keyv": "^3.1.4", "@types/node": "*", "@types/responselike": "^1.0.0" } }, "sha512-IQ3EbTzGxIigb1I3qPZc1rWJnH0BmSKv5QYTalEwweFvyBDLSAe24zP0le/hyi7ecGfZVlIVAg4BZqb8WBwKqw=="], "@types/chai": ["@types/chai@5.2.3", "", { "dependencies": { "@types/deep-eql": "*", "assertion-error": "^2.0.1" } }, "sha512-Mw558oeA9fFbv65/y4mHtXDs9bPnFMZAL/jxdPFUpOHHIXX91mcgEHbS5Lahr+pwZFR8A7GQleRWeI6cGFC2UA=="], @@ -629,6 +632,8 @@ "buffer-crc32": ["buffer-crc32@0.2.13", "", {}, "sha512-VO9Ht/+p3SN7SKWqcrgEzjGbRSJYTx+Q1pTQC0wrWqHx0vpJraQ6GtHx8tvcg1rlK1byhU5gccxgOgj7B0TDkQ=="], + "bun-types": ["bun-types@1.3.9", "", { "dependencies": { "@types/node": "*" } }, "sha512-+UBWWOakIP4Tswh0Bt0QD0alpTY8cb5hvgiYeWCMet9YukHbzuruIEeXC2D7nMJPB12kbh8C7XJykSexEqGKJg=="], + "bundle-name": ["bundle-name@4.1.0", "", { "dependencies": { "run-applescript": "^7.0.0" } }, "sha512-tjwM5exMg6BGRI+kNmTntNsvdZS1X8BFYS6tnJ2hdH0kVxM6/eVZ2xy+FqStSWvYmtfFMDLIxurorHwDKfDz5Q=="], "cac": ["cac@6.7.14", "", {}, "sha512-b6Ilus+c3RrdDk+JhLKUAQfzzgLEPy6wcXqS7f/xe1EETvsDP6GORG7SFuOs6cID5YkqchW/LXZbX5bc8j7ZcQ=="], @@ -1451,6 +1456,8 @@ "babel-dead-code-elimination/@babel/parser": ["@babel/parser@7.29.0", "", { "dependencies": { "@babel/types": "^7.29.0" }, "bin": "./bin/babel-parser.js" }, "sha512-IyDgFV5GeDUVX4YdF/3CPULtVGSXXMLh1xVIgdCgxApktqnQV0r7/8Nqthg+8YLGaAtdyIlo2qIdZrbCv4+7ww=="], + "bun-types/@types/node": ["@types/node@24.10.13", "", { "dependencies": { "undici-types": "~7.16.0" } }, "sha512-oH72nZRfDv9lADUBSo104Aq7gPHpQZc4BTx38r9xf9pg5LfP6EzSyH2n7qFmmxRQXh7YlUXODcYsg6PuTDSxGg=="], + "cliui/wrap-ansi": ["wrap-ansi@7.0.0", "", { "dependencies": { "ansi-styles": "^4.0.0", "string-width": "^4.1.0", "strip-ansi": "^6.0.0" } }, "sha512-YVGIj2kamLSTxw6NsZjoBxfSwsn0ycdesmc4p+Q21c5zPuZ1pl+NfxVdxPtdHvmNVOQ6XSYG4AUtyt/Fi7D16Q=="], "clone-response/mimic-response": ["mimic-response@1.0.1", "", {}, "sha512-j5EctnkH7amfV/q5Hgmoal1g2QHFJRraOtmx0JpIqkxhBhI/lJSl1nMpQ45hVarwNETOoWEimndZ4QK0RHxuxQ=="], @@ -1501,6 +1508,8 @@ "@tailwindcss/node/lightningcss/lightningcss-win32-x64-msvc": ["lightningcss-win32-x64-msvc@1.30.2", "", { "os": "win32", "cpu": "x64" }, "sha512-5g1yc73p+iAkid5phb4oVFMB45417DkRevRbt/El/gKXJk4jid+vPFF/AXbxn05Aky8PapwzZrdJShv5C0avjw=="], + "bun-types/@types/node/undici-types": ["undici-types@7.16.0", "", {}, "sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw=="], + "electron/@types/node/undici-types": ["undici-types@7.16.0", "", {}, "sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw=="], "rolldown-plugin-dts/@babel/types/@babel/helper-string-parser": ["@babel/helper-string-parser@8.0.0-rc.1", "", {}, "sha512-vi/pfmbrOtQmqgfboaBhaCU50G7mcySVu69VU8z+lYoPPB6WzI9VgV7WQfL908M4oeSH5fDkmoupIqoE0SdApw=="],