import fs from "node:fs";import os from "node:os";import path from "node:path";import { EventEmitter } from "node:events"; import { describe, expect, it, afterEach, vi } from "vitest";import { createServer } from "./wsServer";import WebSocket from "ws"; import {  DEFAULT_TERMINAL_ID,  WS_CHANNELS,  WS_METHODS,  type AppSettingsUpdateInput,  type ProjectAddInput,  type ProjectRemoveInput,  type ProjectUpdateScriptsInput,  type StateCatchUpInput,  type StateListMessagesInput,  type ThreadsCreateInput,  type ThreadsDeleteInput,  type ThreadsMarkVisitedInput,  type ThreadsUpdateBranchInput,  type ThreadsUpdateModelInput,  type ThreadsUpdateTerminalStateInput,  type ThreadsUpdateTitleInput,  type KeybindingsConfig,  type ResolvedKeybindingsConfig,  type StateEvent,  type WsPush,  type WsResponse,} from "@t3tools/contracts";import { compileResolvedKeybindingRule, DEFAULT_KEYBINDINGS } from "./keybindings";import type {  TerminalClearInput,  TerminalCloseInput,  TerminalEvent,  TerminalOpenInput,  TerminalResizeInput,  TerminalSessionSnapshot,  TerminalWriteInput,} from "@t3tools/contracts";import type { TerminalManager } from "./terminalManager";import { PersistenceService } from "./persistenceService";import { ProviderManager } from "./providerManager";import type { ApplyCheckpointRevertInput, StateSyncEngine } from "./stateSyncEngine";import { LegacyStateSyncEngine } from "./stateSyncEngineLegacy";import { LiveStoreReadPilotStateSyncEngine } from "./stateSyncEngineLiveStoreReadPilot";import { ShadowStateSyncEngine } from "./stateSyncEngineShadow";import { LiveStoreStateMirror } from "./livestore/liveStoreEngine";import { diffStateSnapshots } from "./livestore/parity"; interface PendingMessages {  queue: unknown[];  waiters: Array<(message: unknown) => void>;} const pendingBySocket = new WeakMap<WebSocket, PendingMessages>(); class MockTerminalManager extends EventEmitter<{ event: [event: TerminalEvent] }> {  private readonly sessions = new Map<string, TerminalSessionSnapshot>();   private key(threadId: string, terminalId: string): string {    return `${threadId}\u0000${terminalId}`;  }   async open(input: TerminalOpenInput): Promise<TerminalSessionSnapshot> {    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",        threadId: input.threadId,        terminalId,        createdAt: now,        snapshot,      });    });    return snapshot;  }   async write(input: TerminalWriteInput): Promise<void> {    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,      });    });  }   async resize(_input: TerminalResizeInput): Promise<void> {}   async clear(input: TerminalClearInput): Promise<void> {    const terminalId = input.terminalId ?? DEFAULT_TERMINAL_ID;    queueMicrotask(() => {      this.emit("event", {        type: "cleared",        threadId: input.threadId,        terminalId,        createdAt: new Date().toISOString(),      });    });  }   async restart(input: TerminalOpenInput): Promise<TerminalSessionSnapshot> {    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",        threadId: input.threadId,        terminalId,        createdAt: now,        snapshot,      });    });    return snapshot;  }   async close(input: TerminalCloseInput): Promise<void> {    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);      }    }  }   dispose(): void {}} class MockStateSyncEngine extends EventEmitter<{ stateEvent: [event: StateEvent] }>  implements StateSyncEngine{  readonly loadSnapshotMock = vi.fn<    () => {      projects: [];      threads: [];      lastStateSeq: number;    }  >(() => ({    projects: [],    threads: [],    lastStateSeq: 42,  }));  readonly listMessagesMock = vi.fn(() => ({    messages: [],    total: 0,    nextOffset: null,  }));  readonly catchUpMock = vi.fn(() => ({    events: [],    lastStateSeq: 42,  }));  readonly getAppSettingsMock = vi.fn(() => ({    codexBinaryPath: "",    codexHomePath: "",  }));  readonly updateAppSettingsMock = vi.fn(() => ({    codexBinaryPath: "",    codexHomePath: "",  }));  readonly createThreadMock = vi.fn(() => ({    thread: {      id: "thread-1",      codexThreadId: null,      projectId: "project-1",      title: "Thread",      model: "gpt-5.3-codex",      terminalOpen: false,      terminalHeight: 280,      terminalIds: ["default"],      runningTerminalIds: [],      activeTerminalId: "default",      terminalGroups: [{ id: "group-default", terminalIds: ["default"] }],      activeTerminalGroupId: "group-default",      createdAt: "2026-02-20T00:00:00.000Z",      updatedAt: "2026-02-20T00:00:00.000Z",      branch: null,      worktreePath: null,      turnDiffSummaries: [],    },  }));  readonly updateThreadTerminalStateMock = vi.fn(() => this.createThreadMock());  readonly updateThreadModelMock = vi.fn(() => this.createThreadMock());  readonly updateThreadTitleMock = vi.fn(() => this.createThreadMock());  readonly updateThreadBranchMock = vi.fn(() => this.createThreadMock());  readonly markThreadVisitedMock = vi.fn(() => this.createThreadMock());  readonly deleteThreadMock = vi.fn();  readonly listProjectsMock = vi.fn(() => []);  readonly addProjectMock = vi.fn(() => ({    project: {      id: "project-1",      cwd: "/tmp/project",      name: "project",      scripts: [],      createdAt: "2026-02-20T00:00:00.000Z",      updatedAt: "2026-02-20T00:00:00.000Z",    },    created: true,  }));  readonly removeProjectMock = vi.fn();  readonly updateProjectScriptsMock = vi.fn(() => ({    project: {      id: "project-1",      cwd: "/tmp/project",      name: "project",      scripts: [],      createdAt: "2026-02-20T00:00:00.000Z",      updatedAt: "2026-02-20T00:00:00.000Z",    },  }));  readonly applyCheckpointRevertMock = vi.fn();  readonly closeMock = vi.fn();   onStateEvent(listener: (event: StateEvent) => void): () => void {    this.on("stateEvent", listener);    return () => {      this.off("stateEvent", listener);    };  }   emitStateEvent(event: StateEvent): void {    this.emit("stateEvent", event);  }   loadSnapshot() {    return this.loadSnapshotMock();  }  listMessages(_raw: StateListMessagesInput) {    return this.listMessagesMock();  }  catchUp(_raw: StateCatchUpInput) {    return this.catchUpMock();  }  getAppSettings() {    return this.getAppSettingsMock();  }  updateAppSettings(_raw: AppSettingsUpdateInput) {    return this.updateAppSettingsMock();  }  createThread(_raw: ThreadsCreateInput) {    return this.createThreadMock();  }  updateThreadTerminalState(_raw: ThreadsUpdateTerminalStateInput) {    return this.updateThreadTerminalStateMock();  }  updateThreadModel(_raw: ThreadsUpdateModelInput) {    return this.updateThreadModelMock();  }  updateThreadTitle(_raw: ThreadsUpdateTitleInput) {    return this.updateThreadTitleMock();  }  updateThreadBranch(_raw: ThreadsUpdateBranchInput) {    return this.updateThreadBranchMock();  }  markThreadVisited(_raw: ThreadsMarkVisitedInput) {    return this.markThreadVisitedMock();  }  deleteThread(_raw: ThreadsDeleteInput): void {    this.deleteThreadMock();  }  listProjects() {    return this.listProjectsMock();  }  addProject(_raw: ProjectAddInput) {    return this.addProjectMock();  }  removeProject(_raw: ProjectRemoveInput): void {    this.removeProjectMock();  }  updateProjectScripts(_raw: ProjectUpdateScriptsInput) {    return this.updateProjectScriptsMock();  }  applyCheckpointRevert(input: ApplyCheckpointRevertInput): void {    this.applyCheckpointRevertMock(input);  }  close(): void {    this.closeMock();  }} 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);  });} async function sendRequest(ws: WebSocket, method: string, params?: unknown): Promise<WsResponse> {  const id = crypto.randomUUID();  const message = JSON.stringify({ id, method, ...(params !== undefined ? { params } : {}) });  ws.send(message);   // Wait for response with matching id  while (true) {    const parsed = (await waitForMessage(ws)) as Record<string, unknown>;    if (parsed.id === id) {      return parsed as WsResponse;    }  }} function compileKeybindings(bindings: KeybindingsConfig): ResolvedKeybindingsConfig {  const resolved: ResolvedKeybindingsConfig = [];  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;} async function waitForPush(ws: WebSocket, channel: string): Promise<WsPush> {  while (true) {    const message = (await waitForMessage(ws)) as WsPush;    if (message.type === "push" && message.channel === channel) {      return message;    }  }} async function waitForCondition(  predicate: () => boolean,  options: { timeoutMs?: number; intervalMs?: number } = {},): Promise<void> {  const timeoutMs = options.timeoutMs ?? 1500;  const intervalMs = options.intervalMs ?? 20;  const deadline = Date.now() + timeoutMs;  while (Date.now() < deadline) {    if (predicate()) {      return;    }    await new Promise((resolve) => setTimeout(resolve, intervalMs));  }  expect(predicate()).toBe(true);} const DEFAULT_RESOLVED_KEYBINDINGS = compileKeybindings([...DEFAULT_KEYBINDINGS]); function mergeWithDefaultsForTest(custom: KeybindingsConfig): ResolvedKeybindingsConfig {  if (custom.length === 0) {    return DEFAULT_RESOLVED_KEYBINDINGS;  }   const overriddenCommands = new Set(custom.map((binding) => binding.command));  const retainedDefaults = DEFAULT_KEYBINDINGS.filter(    (binding) => !overriddenCommands.has(binding.command),  );  return compileKeybindings([...retainedDefaults, ...custom].slice(-256));} describe("WebSocket Server", () => {  let server: ReturnType<typeof createServer> | null = null;  const connections: WebSocket[] = [];  const tempDirs: string[] = [];  const injectedPersistenceServices: PersistenceService[] = [];   function makeTempDir(prefix: string): string {    const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));    tempDirs.push(dir);    return dir;  }   function createTestServer(    options: {      cwd?: string;      devUrl?: string;      authToken?: string;      stateDir?: string;      gitManager?: {        status: (input: { cwd: string }) => Promise<unknown>;        runStackedAction: (input: { cwd: string; action: string }) => Promise<unknown>;      };      terminalManager?: TerminalManager;      persistenceService?: PersistenceService;      stateSyncEngine?: StateSyncEngine;      syncEngineMode?: "legacy" | "shadow" | "livestore-read-pilot" | "livestore";    } = {},  ): ReturnType<typeof createServer> {    const stateDir = options.stateDir ?? makeTempDir("t3code-ws-state-");    const persistenceService =      options.persistenceService ??      new PersistenceService({        dbPath: path.join(stateDir, "state.sqlite"),        legacyProjectsJsonPath: path.join(stateDir, "projects.json"),      });    injectedPersistenceServices.push(persistenceService);    return createServer({      port: 0,      cwd: options.cwd ?? "/test/project",      ...(options.devUrl ? { devUrl: options.devUrl } : {}),      ...(options.authToken ? { authToken: options.authToken } : {}),      persistenceService,      ...(options.stateSyncEngine ? { stateSyncEngine: options.stateSyncEngine } : {}),      ...(options.syncEngineMode ? { syncEngineMode: options.syncEngineMode } : {}),      ...(options.gitManager ? { gitManager: options.gitManager as never } : {}),      ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}),    });  }   afterEach(async () => {    for (const ws of connections) {      ws.close();    }    connections.length = 0;    if (server) {      await server.stop();    }    server = null;    for (const service of injectedPersistenceServices.splice(0, injectedPersistenceServices.length)) {      service.close();    }    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 = createTestServer({ cwd: "/test/project" });    // Get the actual port after listen    await server.start();    const addr = server.httpServer.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;    expect(message.type).toBe("push");

Lines 1–500 of 2049. Load more