import { spawn } from "node:child_process";import fs from "node:fs";import http from "node:http";import os from "node:os";import path from "node:path";import type { Duplex } from "node:stream"; import {  EDITORS,  ORCHESTRATION_WS_CHANNELS,  ORCHESTRATION_WS_METHODS,  WS_CHANNELS,  WS_METHODS,  type OrchestrationReadModelPush,  type TerminalEvent,  type WsPush,  type WsRequest,  type ProviderEvent,  type WsResponse,  wsRequestSchema,} from "@t3tools/contracts";import { Layer, ManagedRuntime } from "effect";import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger";import { ProviderManager } from "./providerManager";import { GitManager } from "./gitManager";import {  checkoutGitBranch,  createGitBranch,  createGitWorktree,  initGitRepo,  listGitBranches,  pullGitBranch,  removeGitWorktree,} from "./git";import { TerminalManager } from "./terminalManager";import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";import { searchWorkspaceEntries } from "./workspaceEntries";import type { OrchestrationEngine } from "./orchestration/engine";import { OrchestrationLive } from "./orchestration/layers";import { OrchestrationEngineService } from "./orchestration/services";import { ProjectRepositoryLive } from "./persistence/Layers/Projects";import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";import assert from "node:assert"; const MIME_TYPES: Record<string, string> = {  ".html": "text/html; charset=utf-8",  ".js": "application/javascript; charset=utf-8",  ".css": "text/css; charset=utf-8",  ".json": "application/json; charset=utf-8",  ".png": "image/png",  ".jpg": "image/jpeg",  ".svg": "image/svg+xml",  ".ico": "image/x-icon",  ".woff": "font/woff",  ".woff2": "font/woff2",  ".ttf": "font/ttf",  ".map": "application/json",}; export interface ServerOptions {  port: number;  host?: string | undefined;  cwd: string;  stateDir?: string | undefined;  staticDir?: string | undefined;  devUrl?: string | undefined;  logWebSocketEvents?: boolean | undefined;  gitManager?: GitManager | undefined;  terminalManager?: TerminalManager | undefined;  authToken?: string | undefined;} const isServerNotRunningError = (error: unknown): boolean => {  if (!(error instanceof Error)) return false;  const maybeCode = (error as NodeJS.ErrnoException).code;  return (    maybeCode === "ERR_SERVER_NOT_RUNNING" || error.message.toLowerCase().includes("not running")  );}; function rejectUpgrade(socket: Duplex, statusCode: number, message: string): void {  socket.write(    `HTTP/1.1 ${statusCode} ${statusCode === 401 ? "Unauthorized" : "Bad Request"}\r\n` +      "Connection: close\r\n" +      "Content-Type: text/plain\r\n" +      `Content-Length: ${Buffer.byteLength(message)}\r\n` +      "\r\n" +      message,  );  socket.destroy();} function parseBooleanEnv(value: string | undefined): boolean | undefined {  if (value === undefined) return undefined;  const normalized = value.trim().toLowerCase();  if (["1", "true", "yes", "on"].includes(normalized)) return true;  if (["0", "false", "no", "off"].includes(normalized)) return false;  return undefined;} function asObject(value: unknown): Record<string, unknown> | null {  if (!value || typeof value !== "object") return null;  return value as Record<string, unknown>;} function asString(value: unknown): string | null {  return typeof value === "string" ? value : null;} function asNumber(value: unknown): number | null {  return typeof value === "number" && Number.isFinite(value) ? value : null;} const noop = () => {}; export function createServer(options: ServerOptions) {  const {    port,    host,    cwd,    stateDir,    staticDir,    devUrl,    logWebSocketEvents: explicitLogWsEvents,    gitManager = new GitManager(),    terminalManager = new TerminalManager(),    authToken,  } = options;  const providerManager = new ProviderManager();   const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata");  let effectRuntime: ManagedRuntime.ManagedRuntime<    ProjectRepository | OrchestrationEngineService,    unknown  > | null = null;  let projectRepository: ProjectRepositoryShape | null = null;  let orchestrationEngine: OrchestrationEngine | null = null;  const clients = new Set<WebSocket>();  const logger = createLogger("ws");  const logWebSocketEvents =    explicitLogWsEvents ?? parseBooleanEnv(process.env.T3CODE_LOG_WS_EVENTS) ?? Boolean(devUrl);  let keybindingsConfig = loadResolvedKeybindingsConfig(logger);  let unsubscribeReadModel = noop;  let unsubscribeDomainEvents = noop;  const pendingProviderEvents = new Map<string, ProviderEvent[]>();  const streamingAccumulators = new Map<string, Map<string, string>>();  const MAX_BUFFERED_PROVIDER_EVENTS = 200;   function logOutgoingPush(push: WsPush, recipients: number) {    if (!logWebSocketEvents) return;    logger.event("outgoing push", {      channel: push.channel,      recipients,      payload: push.data,    });  }   const getOrchestrationEngine = () => {    assert(orchestrationEngine, "Orchestration engine is not started");    return orchestrationEngine;  };   const getProjectRepository = () => {    assert(projectRepository, "Project repository is not started");    return projectRepository;  };   const getEffectRuntime = () => {    assert(effectRuntime, "Effect runtime is not started");    return effectRuntime;  };   function attachOrchestrationSubscriptions() {    const orchestrationEngine = getOrchestrationEngine();     unsubscribeReadModel = orchestrationEngine.subscribeToReadModel((snapshot) => {      const push: WsPush = {        type: "push",        channel: ORCHESTRATION_WS_CHANNELS.readModel,        data: {          sequence: snapshot.sequence,          snapshot,        } satisfies OrchestrationReadModelPush,      };      const message = JSON.stringify(push);      let recipients = 0;      for (const client of clients) {        if (client.readyState === client.OPEN) {          client.send(message);          recipients += 1;        }      }      logOutgoingPush(push, recipients);    });     unsubscribeDomainEvents = orchestrationEngine.subscribeToDomainEvents((event) => {      if (event.type === "thread.session-set") {        const payload = event.payload as Record<string, unknown> | null;        const session = payload?.session as Record<string, unknown> | undefined;        const sessionId = typeof session?.sessionId === "string" ? session.sessionId : null;        if (sessionId) {          const buffered = pendingProviderEvents.get(sessionId);          if (buffered && buffered.length > 0) {            pendingProviderEvents.delete(sessionId);            void (async () => {              for (const bufferedEvent of buffered) {                // eslint-disable-next-line no-await-in-loop -- sequential processing preserves event ordering                await processProviderEvent(bufferedEvent);              }            })().catch(() => undefined);          }        }      }       const push: WsPush = {        type: "push",        channel: ORCHESTRATION_WS_CHANNELS.domainEvent,        data: event,      };      const message = JSON.stringify(push);      let recipients = 0;      for (const client of clients) {        if (client.readyState === client.OPEN) {          client.send(message);          recipients += 1;        }      }      logOutgoingPush(push, recipients);    });  }   // Forward provider events to all connected WebSocket clients  async function processProviderEvent(event: ProviderEvent): Promise<void> {    const liveOrchestrationEngine = orchestrationEngine;    if (!liveOrchestrationEngine) {      return;    }    const snapshot = liveOrchestrationEngine.getSnapshot();    const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId);    if (!thread) {      const buffer = pendingProviderEvents.get(event.sessionId);      if (buffer) {        if (buffer.length < MAX_BUFFERED_PROVIDER_EVENTS) {          buffer.push(event);        }      } else {        pendingProviderEvents.set(event.sessionId, [event]);      }      return;    }    const now = event.createdAt;    if (event.method === "turn/started" || event.method === "turn/completed") {      await liveOrchestrationEngine.dispatch({        type: "thread.session",        commandId: crypto.randomUUID(),        threadId: thread.id,        session: {          sessionId: event.sessionId,          provider: event.provider,          status: event.method === "turn/started" ? "running" : "ready",          threadId: thread.id,          activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null,          createdAt: thread.session?.createdAt ?? now,          updatedAt: now,          lastError: null,        },        createdAt: now,      });    }    if (event.textDelta && event.textDelta.length > 0) {      const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;      let sessionAccumulators = streamingAccumulators.get(event.sessionId);      if (!sessionAccumulators) {        sessionAccumulators = new Map();        streamingAccumulators.set(event.sessionId, sessionAccumulators);      }      sessionAccumulators.set(        assistantMessageId,        (sessionAccumulators.get(assistantMessageId) ?? "") + event.textDelta,      );      await liveOrchestrationEngine.dispatch(        {          type: "message.send",          commandId: crypto.randomUUID(),          threadId: thread.id,          messageId: assistantMessageId,          role: "assistant",          text: event.textDelta,          streaming: true,          createdAt: now,        },        { transient: true },      );    }    if (event.method === "turn/completed") {      const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;      const sessionAccumulators = streamingAccumulators.get(event.sessionId);      const fullText = sessionAccumulators?.get(assistantMessageId) ?? "";      sessionAccumulators?.delete(assistantMessageId);      if (sessionAccumulators?.size === 0) {        streamingAccumulators.delete(event.sessionId);      }      await liveOrchestrationEngine.dispatch({        type: "message.send",        commandId: crypto.randomUUID(),        threadId: thread.id,        messageId: assistantMessageId,        role: "assistant",        text: fullText,        streaming: false,        createdAt: now,      });    }    if (event.method === "checkpoint/captured") {      const payload = asObject(event.payload);      const turnId = event.turnId ?? asString(payload?.turnId);      if (turnId) {        await liveOrchestrationEngine.dispatch({          type: "thread.turnDiff.complete",          commandId: crypto.randomUUID(),          threadId: thread.id,          turnId,          completedAt: now,          ...(asString(payload?.status) !== null            ? { status: asString(payload?.status) ?? undefined }            : {}),          files: [],          assistantMessageId: `assistant:${turnId}`,          ...(asNumber(payload?.turnCount) !== null            ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined }            : {}),          createdAt: now,        });      }    }    if (event.kind === "error" && event.message) {      await liveOrchestrationEngine.dispatch({        type: "thread.session",        commandId: crypto.randomUUID(),        threadId: thread.id,        session: {          sessionId: event.sessionId,          provider: event.provider,          status: "error",          threadId: thread.id,          activeTurnId: event.turnId ?? null,          createdAt: thread.session?.createdAt ?? now,          updatedAt: now,          lastError: event.message,        },        createdAt: now,      });    }  }   providerManager.on("event", (event) => {    void processProviderEvent(event).catch(() => undefined);  });   const onTerminalEvent = (event: TerminalEvent) => {    const push: WsPush = {      type: "push",      channel: WS_CHANNELS.terminalEvent,      data: event,    };    const message = JSON.stringify(push);    let recipients = 0;    for (const client of clients) {      if (client.readyState === client.OPEN) {        client.send(message);        recipients += 1;      }    }    logOutgoingPush(push, recipients);  };  terminalManager.on("event", onTerminalEvent);   // HTTP server — serves static files or redirects to Vite dev server  const httpServer = http.createServer((req, res) => {    // In dev mode, redirect to Vite dev server    if (devUrl) {      res.writeHead(302, { Location: devUrl });      res.end();      return;    }     // Serve static files from the web app build    if (!staticDir) {      res.writeHead(503, { "Content-Type": "text/plain" });      res.end("No static directory configured and no dev URL set.");      return;    }     const url = new URL(req.url ?? "/", `http://localhost:${port}`);    let filePath = path.join(staticDir, url.pathname);     // SPA fallback: if no file extension and not found, serve index.html    const ext = path.extname(filePath);    if (!ext) {      filePath = path.join(filePath, "index.html");    }     fs.stat(filePath, (err, stats) => {      if (err || !stats?.isFile()) {        // SPA fallback        const indexPath = path.join(staticDir, "index.html");        fs.readFile(indexPath, (readErr, data) => {          if (readErr) {            res.writeHead(404, { "Content-Type": "text/plain" });            res.end("Not Found");            return;          }          res.writeHead(200, { "Content-Type": "text/html; charset=utf-8" });          res.end(data);        });        return;      }       const fileExt = path.extname(filePath);      const contentType = MIME_TYPES[fileExt] ?? "application/octet-stream";       fs.readFile(filePath, (readErr, data) => {        if (readErr) {          res.writeHead(500, { "Content-Type": "text/plain" });          res.end("Internal Server Error");          return;        }        res.writeHead(200, { "Content-Type": contentType });        res.end(data);      });    });  });   // WebSocket server — upgrades from the HTTP server  const wss = new WebSocketServer({ noServer: true });   httpServer.on("upgrade", (request, socket, head) => {    if (authToken) {      let providedToken: string | null = null;      try {        const url = new URL(request.url ?? "/", `http://localhost:${port}`);        providedToken = url.searchParams.get("token");      } catch {        rejectUpgrade(socket, 400, "Invalid WebSocket URL");        return;      }       if (providedToken !== authToken) {        rejectUpgrade(socket, 401, "Unauthorized WebSocket connection");        return;      }    }     wss.handleUpgrade(request, socket, head, (ws) => {      wss.emit("connection", ws, request);    });  });   wss.on("connection", (ws) => {    clients.add(ws);    const orchestrationEngine = getOrchestrationEngine();     // Send welcome message with project info    const segments = cwd.split(/[/\\]/).filter(Boolean);    const projectName = segments[segments.length - 1] ?? "project";     const welcome: WsPush = {      type: "push",      channel: WS_CHANNELS.serverWelcome,      data: { cwd, projectName },    };    logOutgoingPush(welcome, 1);    ws.send(JSON.stringify(welcome));     const snapshotPush: WsPush = {      type: "push",      channel: ORCHESTRATION_WS_CHANNELS.readModel,      data: {        sequence: orchestrationEngine.getSnapshot().sequence,        snapshot: orchestrationEngine.getSnapshot(),      } satisfies OrchestrationReadModelPush,    };    logOutgoingPush(snapshotPush, 1);    ws.send(JSON.stringify(snapshotPush));     ws.on("message", (raw) => {      void handleMessage(ws, raw);    });     ws.on("close", () => {      clients.delete(ws);    });     ws.on("error", () => {      clients.delete(ws);    });  });

Lines 1–500 of 861. Load more