import * as Http from "node:http";import fs from "node:fs";import os from "node:os";import path from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices";import { Effect, Exit, Layer, PlatformError, PubSub, Scope, Stream } from "effect";import { describe, expect, it, afterEach, vi } from "vitest";import { createServer } from "./wsServer";import WebSocket from "ws";import { ServerConfig, type ServerConfigShape } from "./config";import { makeServerProviderLayer, makeServerRuntimeServicesLayer } from "./serverLayers"; import {  DEFAULT_TERMINAL_ID,  EDITORS,  EventId,  ORCHESTRATION_WS_CHANNELS,  ORCHESTRATION_WS_METHODS,  ProviderItemId,  ThreadId,  TurnId,  WS_CHANNELS,  WS_METHODS,  type WebSocketResponse,  type ProviderRuntimeEvent,  type ServerProviderStatus,  type KeybindingsConfig,  type ResolvedKeybindingsConfig,  type WsPush,} from "@t3tools/contracts";import { compileResolvedKeybindingRule, DEFAULT_KEYBINDINGS } from "./keybindings";import type {  TerminalClearInput,  TerminalCloseInput,  TerminalEvent,  TerminalOpenInput,  TerminalResizeInput,  TerminalSessionSnapshot,  TerminalWriteInput,} from "@t3tools/contracts";import { TerminalManager, type TerminalManagerShape } from "./terminal/Services/Manager";import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./persistence/Layers/Sqlite";import { SqlClient, SqlError } from "effect/unstable/sql";import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";import { ProviderHealth, type ProviderHealthShape } from "./provider/Services/ProviderHealth";import { Open, type OpenShape } from "./open";import { GitManager, type GitManagerShape } from "./git/Services/GitManager.ts";import type { GitCoreShape } from "./git/Services/GitCore.ts";import { GitCore } from "./git/Services/GitCore.ts";import { GitCommandError, GitManagerError } from "./git/Errors.ts";import { MigrationError } from "@effect/sql-sqlite-bun/SqliteMigrator";import { AnalyticsService } from "./telemetry/Services/AnalyticsService.ts"; interface PendingMessages {  queue: unknown[];  waiters: Array<(message: unknown) => void>;} const pendingBySocket = new WeakMap<WebSocket, PendingMessages>(); const asEventId = (value: string): EventId => EventId.makeUnsafe(value);const asProviderItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); const defaultOpenService: OpenShape = {  openBrowser: () => Effect.void,  openInEditor: () => Effect.void,}; const defaultProviderStatuses: ReadonlyArray<ServerProviderStatus> = [  {    provider: "codex",    status: "ready",    available: true,    authStatus: "authenticated",    checkedAt: "2026-01-01T00:00:00.000Z",  },]; const defaultProviderHealthService: ProviderHealthShape = {  getStatuses: Effect.succeed(defaultProviderStatuses),}; class MockTerminalManager implements TerminalManagerShape {  private readonly sessions = new Map<string, TerminalSessionSnapshot>();  private readonly listeners = new Set<(event: TerminalEvent) => void>();   private key(threadId: string, terminalId: string): string {    return `${threadId}\u0000${terminalId}`;  }   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,        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;    });   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,        });      });    });   readonly resize: TerminalManagerShape["resize"] = (_input: TerminalResizeInput) => Effect.void;   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(),        });      });    });   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,        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;    });   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);      };    });   readonly dispose: TerminalManagerShape["dispose"] = Effect.void;} function connectWs(port: number, token?: string): Promise<WebSocket> {  return new Promise((resolve, reject) => {    const query = token ? `?token=${encodeURIComponent(token)}` : "";    const ws = new WebSocket(`ws://127.0.0.1:${port}/${query}`);    const pending: PendingMessages = { queue: [], waiters: [] };    pendingBySocket.set(ws, pending);     ws.on("message", (raw) => {      const parsed = JSON.parse(String(raw));      const waiter = pending.waiters.shift();      if (waiter) {        waiter(parsed);        return;      }      pending.queue.push(parsed);    });     ws.once("open", () => resolve(ws));    ws.once("error", () => reject(new Error("WebSocket connection failed")));  });} function waitForMessage(ws: WebSocket): Promise<unknown> {  const pending = pendingBySocket.get(ws);  if (!pending) {    return Promise.reject(new Error("WebSocket not initialized"));  }   const queued = pending.queue.shift();  if (queued !== undefined) {    return Promise.resolve(queued);  }   return new Promise((resolve) => {    pending.waiters.push(resolve);  });} function asWebSocketResponse(message: unknown): WebSocketResponse | null {  if (typeof message !== "object" || message === null) return null;  if (!("id" in message)) return null;  const id = (message as { id?: unknown }).id;  if (typeof id !== "string") return null;  return message as WebSocketResponse;} async function sendRequest(  ws: WebSocket,  method: string,  params?: unknown,): Promise<WebSocketResponse> {  const id = crypto.randomUUID();  const body =    method === ORCHESTRATION_WS_METHODS.dispatchCommand      ? { _tag: method, command: params }      : params && typeof params === "object" && !Array.isArray(params)        ? { _tag: method, ...(params as Record<string, unknown>) }        : { _tag: method };  const message = JSON.stringify({ id, body });  ws.send(message);   // Wait for response with matching id  while (true) {    const parsed = asWebSocketResponse(await waitForMessage(ws));    if (!parsed) {      continue;    }    if (parsed.id === id) {      return parsed;    }    if (parsed.id === "unknown") {      return parsed;    }  }} async function waitForPush(  ws: WebSocket,  channel: string,  predicate?: (push: WsPush) => boolean,  maxMessages = 120,): Promise<WsPush> {  const take = async (remaining: number): Promise<WsPush> => {    if (remaining <= 0) {      throw new Error(`Timed out waiting for push on ${channel}`);    }    const message = (await waitForMessage(ws)) as WsPush;    if (message.type !== "push" || message.channel !== channel) {      return take(remaining - 1);    }    if (!predicate || predicate(message)) {      return message;    }    return take(remaining - 1);  };  return take(maxMessages);} async function requestPath(  port: number,  requestPath: string,): Promise<{ statusCode: number; body: string }> {  return new Promise((resolve, reject) => {    const req = Http.request(      {        hostname: "127.0.0.1",        port,        path: requestPath,        method: "GET",      },      (res) => {        const chunks: Buffer[] = [];        res.on("data", (chunk) => {          chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));        });        res.on("end", () => {          resolve({            statusCode: res.statusCode ?? 0,            body: Buffer.concat(chunks).toString("utf8"),          });        });      },    );    req.once("error", reject);    req.end();  });} function compileKeybindings(bindings: KeybindingsConfig): ResolvedKeybindingsConfig {  const resolved: Array<ResolvedKeybindingsConfig[number]> = [];  for (const binding of bindings) {    const compiled = compileResolvedKeybindingRule(binding);    if (!compiled) {      throw new Error(`Unexpected invalid keybinding in test setup: ${binding.command}`);    }    resolved.push(compiled);  }  return resolved;} const DEFAULT_RESOLVED_KEYBINDINGS = compileKeybindings([...DEFAULT_KEYBINDINGS]);const VALID_EDITOR_IDS = new Set(EDITORS.map((editor) => editor.id)); function expectAvailableEditors(value: unknown): void {  expect(Array.isArray(value)).toBe(true);  for (const editorId of value as unknown[]) {    expect(typeof editorId).toBe("string");    expect(VALID_EDITOR_IDS.has(editorId as (typeof EDITORS)[number]["id"])).toBe(true);  }} describe("WebSocket Server", () => {  let server: Http.Server | null = null;  let serverScope: Scope.Closeable | null = null;  const connections: WebSocket[] = [];  const tempDirs: string[] = [];   function makeTempDir(prefix: string): string {    const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));    tempDirs.push(dir);    return dir;  }   async function createTestServer(    options: {      persistenceLayer?: Layer.Layer<        SqlClient.SqlClient,        SqlError.SqlError | MigrationError | PlatformError.PlatformError      >;      cwd?: string;      autoBootstrapProjectFromCwd?: boolean;      logWebSocketEvents?: boolean;      devUrl?: string;      authToken?: string;      stateDir?: string;      staticDir?: string;      providerLayer?: Layer.Layer<ProviderService, never>;      providerHealth?: ProviderHealthShape;      open?: OpenShape;      gitManager?: GitManagerShape;      gitCore?: Pick<GitCoreShape, "listBranches" | "initRepo" | "pullCurrentBranch">;      terminalManager?: TerminalManagerShape;    } = {},  ): Promise<Http.Server> {    if (serverScope) {      throw new Error("Test server is already running");    }     const stateDir = options.stateDir ?? makeTempDir("t3code-ws-state-");    const scope = await Effect.runPromise(Scope.make("sequential"));    const persistenceLayer = options.persistenceLayer ?? SqlitePersistenceMemory;    const providerLayer = options.providerLayer ?? makeServerProviderLayer();    const providerHealthLayer = Layer.succeed(      ProviderHealth,      options.providerHealth ?? defaultProviderHealthService,    );    const openLayer = Layer.succeed(Open, options.open ?? defaultOpenService);    const serverConfigLayer = Layer.succeed(ServerConfig, {      mode: "web",      port: 0,      host: undefined,      cwd: options.cwd ?? "/test/project",      keybindingsConfigPath: path.join(stateDir, "keybindings.json"),      stateDir,      staticDir: options.staticDir,      devUrl: options.devUrl ? new URL(options.devUrl) : undefined,      noBrowser: true,      authToken: options.authToken,      autoBootstrapProjectFromCwd: options.autoBootstrapProjectFromCwd ?? false,      logWebSocketEvents: options.logWebSocketEvents ?? Boolean(options.devUrl),    } satisfies ServerConfigShape);    const infrastructureLayer = providerLayer.pipe(Layer.provideMerge(persistenceLayer));    const runtimeOverrides = Layer.mergeAll(      options.gitManager ? Layer.succeed(GitManager, options.gitManager) : Layer.empty,      options.gitCore        ? Layer.succeed(GitCore, options.gitCore as unknown as GitCoreShape)        : Layer.empty,      options.terminalManager        ? Layer.succeed(TerminalManager, options.terminalManager)        : Layer.empty,    );     const runtimeLayer = Layer.merge(      Layer.merge(        makeServerRuntimeServicesLayer().pipe(Layer.provide(infrastructureLayer)),        infrastructureLayer,      ),      runtimeOverrides,    );    const dependenciesLayer = Layer.empty.pipe(      Layer.provideMerge(runtimeLayer),      Layer.provideMerge(providerHealthLayer),      Layer.provideMerge(openLayer),      Layer.provideMerge(serverConfigLayer),      Layer.provideMerge(AnalyticsService.layerTest),      Layer.provideMerge(NodeServices.layer),    );    const runtimeServices = await Effect.runPromise(      Layer.build(dependenciesLayer).pipe(Scope.provide(scope)),    );     try {      const runtime = await Effect.runPromise(        createServer().pipe(Effect.provide(runtimeServices), Scope.provide(scope)),      );      serverScope = scope;      return runtime;    } catch (error) {      await Effect.runPromise(Scope.close(scope, Exit.void));      throw error;    }  }   async function closeTestServer() {    if (!serverScope) return;    const scope = serverScope;    serverScope = null;    await Effect.runPromise(Scope.close(scope, Exit.void));  }   afterEach(async () => {    for (const ws of connections) {      ws.close();    }    connections.length = 0;    await closeTestServer();    server = null;    for (const dir of tempDirs.splice(0, tempDirs.length)) {      fs.rmSync(dir, { recursive: true, force: true });    }    vi.restoreAllMocks();  });   it("sends welcome message on connect", async () => {    server = await createTestServer({ cwd: "/test/project" });    // Get the actual port after listen    const addr = server.address();    const port = typeof addr === "object" && addr !== null ? addr.port : 0;    expect(port).toBeGreaterThan(0);     const ws = await connectWs(port);    connections.push(ws);     const message = (await waitForMessage(ws)) as WsPush;

Lines 1–500 of 1705. Load more