import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import {  DEFAULT_TERMINAL_ID,  type TerminalAttachStreamEvent,  type TerminalEvent,  type TerminalMetadataStreamEvent,  type TerminalOpenInput,  type TerminalRestartInput,} from "@t3tools/contracts";import { HostProcessPlatform } from "@t3tools/shared/hostProcess";import * as Data from "effect/Data";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Encoding from "effect/Encoding";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 PlatformError from "effect/PlatformError";import * as Path from "effect/Path";import * as Ref from "effect/Ref";import * as Schedule from "effect/Schedule";import * as Scope from "effect/Scope";import { TestClock } from "effect/testing";import { expect } from "vite-plus/test"; import * as ProcessRunner from "../../processRunner.ts";import type { TerminalManagerShape } from "../Services/Manager.ts";import {  type PtyAdapterShape,  type PtyExitEvent,  type PtyProcess,  type PtySpawnInput,  PtySpawnError,} from "../Services/PTY.ts";import { makeTerminalManagerWithOptions } from "./Manager.ts"; class WaitForConditionError extends Data.TaggedError("WaitForConditionError")<{  readonly message: string;}> {} class FakePtyProcess implements PtyProcess {  readonly writes: string[] = [];  readonly resizeCalls: Array<{ cols: number; rows: number }> = [];  readonly killSignals: Array<string | undefined> = [];  readonly pid: number;  private readonly dataListeners = new Set<(data: string) => void>();  private readonly exitListeners = new Set<(event: PtyExitEvent) => void>();  killed = false;   constructor(pid: number) {    this.pid = pid;  }   write(data: string): void {    this.writes.push(data);  }   resize(cols: number, rows: number): void {    this.resizeCalls.push({ cols, rows });  }   kill(signal?: string): void {    this.killed = true;    this.killSignals.push(signal);  }   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: string): void {    for (const listener of this.dataListeners) {      listener(data);    }  }   emitExit(event: PtyExitEvent): void {    for (const listener of this.exitListeners) {      listener(event);    }  }} class FakePtyAdapter implements PtyAdapterShape {  readonly spawnInputs: PtySpawnInput[] = [];  readonly processes: FakePtyProcess[] = [];  readonly spawnFailures: Error[] = [];  private readonly mode: "sync" | "async";  private nextPid = 9000;   constructor(mode: "sync" | "async" = "sync") {    this.mode = mode;  }   spawn(input: PtySpawnInput): Effect.Effect<PtyProcess, PtySpawnError> {    this.spawnInputs.push(input);    const failure = this.spawnFailures.shift();    if (failure) {      return Effect.fail(        new PtySpawnError({          adapter: "fake",          message: "Failed to spawn PTY process",          cause: failure,        }),      );    }    const process = new FakePtyProcess(this.nextPid++);    this.processes.push(process);    if (this.mode === "async") {      return Effect.tryPromise({        try: async () => process,        catch: (cause) =>          new PtySpawnError({            adapter: "fake",            message: "Failed to spawn PTY process",            cause,          }),      });    }    return Effect.succeed(process);  }} const waitFor = <E, R>(  predicate: Effect.Effect<boolean, E, R>,  timeout: Duration.Input = 800,): Effect.Effect<void, WaitForConditionError | E, R> =>  predicate.pipe(    Effect.filterOrFail(      (done) => done,      () => new WaitForConditionError({ message: "Condition not met" }),    ),    Effect.retry(Schedule.spaced("15 millis")),    Effect.timeoutOption(timeout),    Effect.flatMap((result) =>      Option.match(result, {        onNone: () =>          Effect.fail(new WaitForConditionError({ message: "Timed out waiting for condition" })),        onSome: () => Effect.void,      }),    ),  ); function openInput(overrides: Partial<TerminalOpenInput> = {}): TerminalOpenInput {  return {    threadId: "thread-1",    terminalId: DEFAULT_TERMINAL_ID,    cwd: process.cwd(),    cols: 100,    rows: 24,    ...overrides,  };} function restartInput(overrides: Partial<TerminalRestartInput> = {}): TerminalRestartInput {  return {    threadId: "thread-1",    terminalId: DEFAULT_TERMINAL_ID,    cwd: process.cwd(),    cols: 100,    rows: 24,    ...overrides,  };} const historyLogPath = (logsDir: string, threadId = "thread-1") =>  Effect.service(Path.Path).pipe(    Effect.map(({ join }) => join(logsDir, `terminal_${Encoding.encodeBase64Url(threadId)}.log`)),  ); const multiTerminalHistoryLogPath = (  logsDir: string,  threadId = "thread-1",  terminalId = DEFAULT_TERMINAL_ID,) =>  Effect.service(Path.Path).pipe(    Effect.map(({ join }) => {      const threadPart = `terminal_${Encoding.encodeBase64Url(threadId)}`;      return join(        logsDir,        terminalId === DEFAULT_TERMINAL_ID          ? `${threadPart}.log`          : `${threadPart}_${Encoding.encodeBase64Url(terminalId)}.log`,      );    }),  ); interface CreateManagerOptions {  shellResolver?: () => string;  env?: NodeJS.ProcessEnv;  subprocessInspector?: (terminalPid: number) => Effect.Effect<{    readonly hasRunningSubprocess: boolean;    readonly childCommand: string | null;    readonly processIds: ReadonlyArray<number>;  }>;  subprocessPollIntervalMs?: number;  processKillGraceMs?: number;  maxRetainedInactiveSessions?: number;  ptyAdapter?: FakePtyAdapter;} interface ManagerFixture {  readonly baseDir: string;  readonly logsDir: string;  readonly ptyAdapter: FakePtyAdapter;  readonly manager: TerminalManagerShape;  readonly getEvents: Effect.Effect<ReadonlyArray<TerminalEvent>>;} const createManager = (  historyLineLimit = 5,  options: CreateManagerOptions = {},): Effect.Effect<  ManagerFixture,  PlatformError.PlatformError,  FileSystem.FileSystem | Path.Path | Scope.Scope | ProcessRunner.ProcessRunner> =>  Effect.flatMap(Effect.service(FileSystem.FileSystem), (fs) =>    Effect.gen(function* () {      const { join } = yield* Path.Path;      const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-terminal-" });      const logsDir = join(baseDir, "userdata", "logs", "terminals");      const ptyAdapter = options.ptyAdapter ?? new FakePtyAdapter();       const manager = yield* makeTerminalManagerWithOptions({        logsDir,        historyLineLimit,        ptyAdapter,        ...(options.shellResolver !== undefined ? { shellResolver: options.shellResolver } : {}),        ...(options.env !== undefined ? { env: options.env } : {}),        ...(options.subprocessInspector !== undefined          ? { subprocessInspector: options.subprocessInspector }          : {}),        ...(options.subprocessPollIntervalMs !== undefined          ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs }          : {}),        processKillGraceMs: options.processKillGraceMs ?? 1,        ...(options.maxRetainedInactiveSessions !== undefined          ? { maxRetainedInactiveSessions: options.maxRetainedInactiveSessions }          : {}),      });      const eventsRef = yield* Ref.make<ReadonlyArray<TerminalEvent>>([]);      const unsubscribe = yield* manager.subscribe((event) =>        Ref.update(eventsRef, (events) => [...events, event]),      );      yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));       return {        baseDir,        logsDir,        join,        ptyAdapter,        manager,        getEvents: Ref.get(eventsRef),      };    }),  ); const withHostPlatform = (platform: NodeJS.Platform) =>  Layer.succeed(HostProcessPlatform, platform); it.layer(  Layer.merge(NodeServices.layer, ProcessRunner.layer.pipe(Layer.provide(NodeServices.layer))),  { excludeTestServices: true },)("TerminalManager", (it) => {  it.effect("spawns lazily and reuses running terminal per thread", () =>    Effect.gen(function* () {      const { manager, ptyAdapter } = yield* createManager();      const [first, second] = yield* Effect.all(        [manager.open(openInput()), manager.open(openInput())],        { concurrency: "unbounded" },      );      const third = yield* manager.open(openInput());       assert.equal(first.threadId, "thread-1");      assert.equal(first.terminalId, DEFAULT_TERMINAL_ID);      assert.equal(second.threadId, "thread-1");      assert.equal(third.threadId, "thread-1");      expect(ptyAdapter.spawnInputs).toHaveLength(1);    }),  );   it.effect("attaches to running sessions without restarting them", () =>    Effect.gen(function* () {      const { manager, ptyAdapter } = yield* createManager();       yield* manager.open(openInput());      const attachEvents = yield* Ref.make<ReadonlyArray<TerminalAttachStreamEvent>>([]);      const unsubscribe = yield* manager.attachStream(        {          threadId: "thread-1",          terminalId: DEFAULT_TERMINAL_ID,          cols: 100,          rows: 40,        },        (event) => Ref.update(attachEvents, (events) => [...events, event]),      );      yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));       const snapshot = (yield* Ref.get(attachEvents)).find((event) => event.type === "snapshot");      expect(snapshot).toBeDefined();      if (!snapshot || snapshot.type !== "snapshot") return;      assert.equal(snapshot.snapshot.threadId, "thread-1");      assert.equal(snapshot.snapshot.terminalId, DEFAULT_TERMINAL_ID);      expect(ptyAdapter.spawnInputs).toHaveLength(1);    }),  );   it.effect("keeps attach streams live when a terminal id is closed and reopened", () =>    Effect.gen(function* () {      const { manager, ptyAdapter } = yield* createManager();      const attachEvents = yield* Ref.make<ReadonlyArray<TerminalAttachStreamEvent>>([]);      const unsubscribe = yield* manager.attachStream(openInput(), (event) =>        Ref.update(attachEvents, (events) => [...events, event]),      );      yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));       yield* manager.close({        threadId: "thread-1",        terminalId: DEFAULT_TERMINAL_ID,        deleteHistory: true,      });      yield* manager.open(openInput());       const events = yield* Ref.get(attachEvents);      expect(events.map((event) => event.type)).toEqual(["snapshot", "closed", "snapshot"]);      expect(        events.filter((event) => event.type === "snapshot").map((event) => event.snapshot.status),      ).toEqual(["running", "running"]);      expect(ptyAdapter.spawnInputs).toHaveLength(2);    }),  );   it.effect("attaches to exited sessions without restarting them", () =>    Effect.gen(function* () {      const { manager, ptyAdapter, getEvents } = yield* createManager();       yield* manager.open(openInput());      const process = ptyAdapter.processes[0];      expect(process).toBeDefined();      if (!process) return;       process.emitExit({ exitCode: 0, signal: 0 });       yield* waitFor(        Effect.map(getEvents, (events) => events.some((event) => event.type === "exited")),        "1200 millis",      );       const attachEvents = yield* Ref.make<ReadonlyArray<TerminalAttachStreamEvent>>([]);      const unsubscribe = yield* manager.attachStream(        openInput({          env: {            T3CODE_WORKTREE_PATH: "/tmp/should-not-restart",          },          worktreePath: "/tmp/should-not-restart",        }),        (event) => Ref.update(attachEvents, (events) => [...events, event]),      );      yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));       const snapshot = (yield* Ref.get(attachEvents)).find((event) => event.type === "snapshot");      expect(snapshot).toBeDefined();      if (!snapshot || snapshot.type !== "snapshot") return;      assert.equal(snapshot.snapshot.status, "exited");      assert.equal(snapshot.snapshot.worktreePath, null);      expect(ptyAdapter.spawnInputs).toHaveLength(1);    }),  );   it.effect("restarts inactive sessions from attach only when requested", () =>    Effect.gen(function* () {      const { manager, ptyAdapter, getEvents } = yield* createManager();       yield* manager.open(openInput());      const process = ptyAdapter.processes[0];      expect(process).toBeDefined();      if (!process) return;       process.emitExit({ exitCode: 0, signal: 0 });       yield* waitFor(        Effect.map(getEvents, (events) => events.some((event) => event.type === "exited")),        "1200 millis",      );       const attachEvents = yield* Ref.make<ReadonlyArray<TerminalAttachStreamEvent>>([]);      const unsubscribe = yield* manager.attachStream(        {          ...openInput({            env: {              T3CODE_WORKTREE_PATH: "/tmp/restart-requested",            },            worktreePath: "/tmp/restart-requested",          }),          restartIfNotRunning: true,        },        (event) => Ref.update(attachEvents, (events) => [...events, event]),      );      yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));       const snapshot = (yield* Ref.get(attachEvents)).find((event) => event.type === "snapshot");      expect(snapshot).toBeDefined();      if (!snapshot || snapshot.type !== "snapshot") return;      assert.equal(snapshot.snapshot.status, "running");      assert.equal(snapshot.snapshot.worktreePath, "/tmp/restart-requested");      expect(ptyAdapter.spawnInputs).toHaveLength(2);    }),  );   const makeDirectory = (filePath: string) =>    Effect.flatMap(Effect.service(FileSystem.FileSystem), (fs) =>      fs.makeDirectory(filePath, { recursive: true }),    );   const chmod = (filePath: string, mode: number) =>    Effect.flatMap(Effect.service(FileSystem.FileSystem), (fs) => fs.chmod(filePath, mode));   const pathExists = (filePath: string) =>    Effect.flatMap(Effect.service(FileSystem.FileSystem), (fs) => fs.exists(filePath));   const readFileString = (filePath: string) =>    Effect.flatMap(Effect.service(FileSystem.FileSystem), (fs) => fs.readFileString(filePath));   const writeFileString = (filePath: string, contents: string) =>    Effect.flatMap(Effect.service(FileSystem.FileSystem), (fs) =>      fs.writeFileString(filePath, contents),    );   it.effect("preserves non-notFound cwd stat failures", () =>    Effect.gen(function* () {      if ((yield* HostProcessPlatform) === "win32") return;       const path = yield* Path.Path;       const { manager, baseDir } = yield* createManager();      const blockedRoot = path.join(baseDir, "blocked-root");      const blockedCwd = path.join(blockedRoot, "cwd");      yield* makeDirectory(blockedCwd);      yield* chmod(blockedRoot, 0o000);       const error = yield* Effect.flip(manager.open(openInput({ cwd: blockedCwd }))).pipe(        Effect.ensuring(chmod(blockedRoot, 0o755).pipe(Effect.ignore)),      );       expect(error).toMatchObject({        _tag: "TerminalCwdError",        cwd: blockedCwd,        reason: "statFailed",      });    }),  );   it.effect("supports asynchronous PTY spawn effects", () =>    Effect.gen(function* () {      const { manager, ptyAdapter } = yield* createManager(5, {        ptyAdapter: new FakePtyAdapter("async"),      });       const snapshot = yield* manager.open(openInput());       assert.equal(snapshot.status, "running");      expect(ptyAdapter.spawnInputs).toHaveLength(1);      expect(ptyAdapter.processes).toHaveLength(1);    }),  );   it.effect("forwards write and resize to active pty process", () =>    Effect.gen(function* () {      const { manager, ptyAdapter } = yield* createManager();      yield* manager.open(openInput());      const process = ptyAdapter.processes[0];      expect(process).toBeDefined();      if (!process) return;       yield* manager.write({        threadId: "thread-1",        terminalId: DEFAULT_TERMINAL_ID,        data: "ls\n",      });      yield* manager.resize({        threadId: "thread-1",        terminalId: DEFAULT_TERMINAL_ID,        cols: 120,        rows: 30,      }); 

Lines 1–500 of 1555. Load more