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,  ProviderDriverKind,  ProviderInstanceId,  ServerSettingsError,  TerminalProviderInstanceNotFoundError,} from "@t3tools/contracts";import { HostProcessPlatform } from "@t3tools/shared/hostProcess";import * as Data from "effect/Data";import * as Clock from "effect/Clock";import * as Deferred from "effect/Deferred";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 * as Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock";import { ChildProcessSpawner } from "effect/unstable/process";import { expect } from "vite-plus/test"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts";import * as ServerConfig from "../config.ts";import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";import * as ProcessRunner from "../processRunner.ts";import * as ServerSettings from "../serverSettings.ts";import * as TerminalManager from "./Manager.ts";import * as PtyAdapter from "./PtyAdapter.ts"; class WaitForConditionError extends Data.TaggedError("WaitForConditionError")<{  readonly message: string;}> {} class FakePtyProcess implements PtyAdapter.PtyProcess {  readonly writes: string[] = [];  readonly resizeCalls: Array<{ cols: number; rows: number }> = [];  readonly killSignals: Array<string | undefined> = [];  readonly pid: number;  writeFailure: unknown | undefined;  resizeFailure: unknown | undefined;  private readonly dataListeners = new Set<(data: string) => void>();  private readonly exitListeners = new Set<(event: PtyAdapter.PtyExitEvent) => void>();  killed = false;   constructor(pid: number) {    this.pid = pid;  }   write(data: string): void {    if (this.writeFailure !== undefined) {      throw this.writeFailure;    }    this.writes.push(data);  }   resize(cols: number, rows: number): void {    if (this.resizeFailure !== undefined) {      throw this.resizeFailure;    }    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: PtyAdapter.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: PtyAdapter.PtyExitEvent): void {    for (const listener of this.exitListeners) {      listener(event);    }  }} class FakePtyAdapter {  readonly spawnInputs: PtyAdapter.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: PtyAdapter.PtySpawnInput,  ): Effect.Effect<PtyAdapter.PtyProcess, PtyAdapter.PtySpawnError> {    this.spawnInputs.push(input);    const failure = this.spawnFailures.shift();    if (failure) {      return Effect.fail(        new PtyAdapter.PtySpawnError({          adapter: "fake",          shell: input.shell,          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 PtyAdapter.PtySpawnError({            adapter: "fake",            shell: input.shell,            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>;  }>;  processTable?: Effect.Effect<    ReadonlyArray<{ readonly pid: number; readonly ppid: number; readonly name: string }>,    never  >;  subprocessPollIntervalMs?: number;  processKillGraceMs?: number;  maxRetainedInactiveSessions?: number;  historyByteLimit?: number;  ptyAdapter?: FakePtyAdapter;  resolveProviderInstanceEnvironment?: Parameters<    typeof TerminalManager.makeWithOptions  >[0]["resolveProviderInstanceEnvironment"];} interface ManagerFixture {  readonly baseDir: string;  readonly logsDir: string;  readonly ptyAdapter: FakePtyAdapter;  readonly manager: TerminalManager.TerminalManager["Service"];  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* TerminalManager.makeWithOptions({        logsDir,        historyLineLimit,        ptyAdapter,        ...(options.historyByteLimit !== undefined          ? { historyByteLimit: options.historyByteLimit }          : {}),        ...(options.shellResolver !== undefined ? { shellResolver: options.shellResolver } : {}),        ...(options.env !== undefined ? { env: options.env } : {}),        ...(options.subprocessInspector !== undefined          ? { subprocessInspector: options.subprocessInspector }          : {}),        ...(options.processTable !== undefined ? { processTable: options.processTable } : {}),        ...(options.subprocessPollIntervalMs !== undefined          ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs }          : {}),        processKillGraceMs: options.processKillGraceMs ?? 1,        ...(options.maxRetainedInactiveSessions !== undefined          ? { maxRetainedInactiveSessions: options.maxRetainedInactiveSessions }          : {}),        ...(options.resolveProviderInstanceEnvironment !== undefined          ? { resolveProviderInstanceEnvironment: options.resolveProviderInstanceEnvironment }          : {}),      });      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); // Apply the existing line policy, then find the longest code-point-aligned byte tail.function retainedHistory(text: string, maxLines: number, maxBytes = Infinity): string {  const terminated = text.endsWith("\n");  const lines = text.split("\n");  if (terminated) lines.pop();  const retained = lines.slice(Math.max(0, lines.length - maxLines)).join("\n");  const capped = terminated ? `${retained}\n` : retained;  if (Buffer.byteLength(capped) <= maxBytes) return capped;  const points = Array.from(capped);  let start = points.length;  let bytes = 0;  while (start > 0) {    const next = Buffer.byteLength(points[start - 1]!);    if (bytes + next > maxBytes) break;    bytes += next;    start -= 1;  }  return points.slice(start).join("");} it("preserves line and byte limits across arbitrary chunks, Unicode, ANSI sequences, and clear", () => {  let randomSeed = 0x20260904;  const fragments = [    "",    "a",    "\n",    "\n\n",    "\r",    "\r\n",    "café",    "名",    "🚀",    "\u001b[31m",    "\u001b[0m",    "\u001b]8;;url\u0007",    "\ud83d",    "\ude80",  ];  const nextFragment = () => {    randomSeed = (Math.imul(randomSeed, 1_664_525) + 1_013_904_223) >>> 0;    return fragments[randomSeed % fragments.length]!;  };   for (const maxBytes of [0, 3, 8, 64, Infinity]) {    for (const maxLines of [0, 1, 3, 5, 5_000]) {      let expected = retainedHistory("before\ninitial\n", maxLines, maxBytes);      const history = new TerminalManager.BoundedTerminalHistory(        maxLines,        "before\ninitial\n",        maxBytes,      );      expect(history.value()).toBe(expected);       for (let step = 0; step < 300; step += 1) {        if (step % 73 === 0) {          history.clear();          expected = "";          expect(history.value()).toBe(expected);        }        const chunk = nextFragment() + nextFragment();        history.append(chunk);        expected = retainedHistory(expected + chunk, maxLines, maxBytes);        expect(history.value()).toBe(expected);      }    }  }}); it("bounds long partial lines and joins surrogate pairs across chunk boundaries", () => {  const maxBytes = 65_539;  let expected = "";  const history = new TerminalManager.BoundedTerminalHistory(5_000, "", maxBytes);  const writes = [    "a".repeat(16_383) + "😀" + "b".repeat(70_000),    "\r" + "c".repeat(70_000) + "\ud83d",    "\ude80" + "d".repeat(100),    "\uFEFF" + "名".repeat(30_000),  ];  for (const text of writes) {    history.append(text);    expected = retainedHistory(expected + text, 5_000, maxBytes);    expect(history.value()).toBe(expected);    expect(Buffer.byteLength(history.value())).toBeLessThanOrEqual(maxBytes);  }}); it("preserves retained lines as older storage is compacted", () => {  for (const maxLines of [3, 5_000]) {    let expected = "";    const history = new TerminalManager.BoundedTerminalHistory(maxLines, expected);    for (let batch = 0; batch < 40; batch += 1) {      const chunk = Array.from({ length: 300 }, (_, line) => `${batch}:${line}\n`).join("");      history.append(chunk);      expected = retainedHistory(expected + chunk, maxLines);      expect(history.value()).toBe(expected);    }  }}); 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]),

Lines 1–500 of 2621. Load more