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 WsResponse,  wsRequestSchema,} from "@t3tools/contracts";import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger";import { ProjectRegistry } from "./projectRegistry";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 { createOrchestrationSystem, type OrchestrationSystem } from "./orchestration/runtime"; 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;  projectRegistry?: ProjectRegistry | undefined;  gitManager?: GitManager | undefined;  terminalManager?: TerminalManager | undefined;  authToken?: string | undefined;} 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,    projectRegistry: providedRegistry,    gitManager: providedGitManager,    terminalManager: providedTerminalManager,    authToken,  } = options;  const providerManager = new ProviderManager();  const terminalManager = providedTerminalManager ?? new TerminalManager();  const projectRegistry =    providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata"));  const gitManager = providedGitManager ?? new GitManager();  const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata");  let orchestrationSystem: OrchestrationSystem | 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;   function logOutgoingPush(push: WsPush, recipients: number) {    if (!logWebSocketEvents) return;    logger.event("outgoing push", {      channel: push.channel,      recipients,      payload: push.data,    });  }   function requireOrchestrationEngine() {    if (!orchestrationSystem) {      throw new Error("Orchestration engine is not started");    }    return orchestrationSystem.engine;  }   function attachOrchestrationSubscriptions() {    const orchestrationEngine = requireOrchestrationEngine();     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) => {      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);    });  }   const streamingTextBuffers = new Map<string, string>();   // Forward provider events to all connected WebSocket clients  providerManager.on("event", (event) => {    void (async () => {      const liveOrchestrationSystem = orchestrationSystem;      if (!liveOrchestrationSystem) {        return;      }      const orchestrationEngine = liveOrchestrationSystem.engine;      const snapshot = orchestrationEngine.getSnapshot();      const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId);      if (!thread) return;      const now = event.createdAt;      if (event.method === "turn/started" || event.method === "turn/completed") {        await orchestrationEngine.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: event.threadId ?? thread.session?.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}`;        const current = streamingTextBuffers.get(assistantMessageId) ?? "";        streamingTextBuffers.set(assistantMessageId, current + event.textDelta);        await orchestrationEngine.dispatchVolatile({          type: "message.send",          commandId: crypto.randomUUID(),          threadId: thread.id,          messageId: assistantMessageId,          role: "assistant",          text: event.textDelta,          streaming: true,          createdAt: now,        });      }      if (event.method === "turn/completed") {        const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;        const accumulatedText = streamingTextBuffers.get(assistantMessageId) ?? "";        streamingTextBuffers.delete(assistantMessageId);        await orchestrationEngine.dispatch({          type: "message.send",          commandId: crypto.randomUUID(),          threadId: thread.id,          messageId: assistantMessageId,          role: "assistant",          text: accumulatedText,          streaming: false,          createdAt: now,        });      }      if (event.method === "checkpoint/captured") {        const payload = asObject(event.payload);        const turnId = event.turnId ?? asString(payload?.turnId);        if (turnId) {          await orchestrationEngine.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 orchestrationEngine.dispatch({          type: "thread.session",          commandId: crypto.randomUUID(),          threadId: thread.id,          session: {            sessionId: event.sessionId,            provider: event.provider,            status: "error",            threadId: event.threadId ?? thread.session?.threadId ?? thread.id,            activeTurnId: event.turnId ?? null,            createdAt: thread.session?.createdAt ?? now,            updatedAt: now,            lastError: event.message,          },          createdAt: now,        });      }    })().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 });   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();  }   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 = requireOrchestrationEngine();     // 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);    });  });   async function handleMessage(ws: WebSocket, raw: unknown) {    let request: WsRequest;    try {      const parsed = JSON.parse(String(raw));      request = wsRequestSchema.parse(parsed);    } catch {      const errorResponse: WsResponse = {        id: "unknown",        error: { message: "Invalid request format" },      };      ws.send(JSON.stringify(errorResponse));      return;    }     try {      const result = await routeRequest(request);      const response: WsResponse = { id: request.id, result };      ws.send(JSON.stringify(response));    } catch (err) {      const message = err instanceof Error ? err.message : "Unknown server error";      const response: WsResponse = {        id: request.id,        error: { message },      };      ws.send(JSON.stringify(response));    }  }   async function routeRequest(request: WsRequest): Promise<unknown> {    const orchestrationEngine = requireOrchestrationEngine();     switch (request.method) {      case WS_METHODS.providersStartSession:        return providerManager.startSession(request.params as never);       case WS_METHODS.providersSendTurn:        return providerManager.sendTurn(request.params as never);       case WS_METHODS.providersInterruptTurn:        return providerManager.interruptTurn(request.params as never);       case WS_METHODS.providersRespondToRequest:        return providerManager.respondToRequest(request.params as never);       case WS_METHODS.providersStopSession: {        providerManager.stopSession(request.params as never);        return undefined;      }       case WS_METHODS.providersListCheckpoints:        return providerManager.listCheckpoints(request.params as never);       case WS_METHODS.providersGetCheckpointDiff:        return providerManager.getCheckpointDiff(request.params as never);       case WS_METHODS.providersRevertToCheckpoint:        {          const result = await providerManager.revertToCheckpoint(request.params as never);          const params = request.params as { sessionId?: string };          const sessionId = params.sessionId;          if (typeof sessionId === "string") {            const snapshot = orchestrationEngine.getSnapshot();            const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId);            if (thread) {              const now = new Date().toISOString();              await orchestrationEngine.dispatch({                type: "thread.revert",                commandId: crypto.randomUUID(),

Lines 1–500 of 797. Load more