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 { Effect, 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 { OrchestrationEngineLive } from "./orchestration/Layer";import { OrchestrationEngineService, type OrchestrationEngineShape } from "./orchestration/Service";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";import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents"; 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: OrchestrationEngineShape | 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,    });  }   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;  };   async function attachOrchestrationSubscriptions(    runtime: ManagedRuntime.ManagedRuntime<ProjectRepository | OrchestrationEngineService, unknown>,    orchestrationEngine: OrchestrationEngineShape,  ) {    unsubscribeReadModel = await runtime.runPromise(      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 = await runtime.runPromise(      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);      }),    );  }   // Forward provider events to all connected WebSocket clients  providerManager.on("event", (event) => {    void (async () => {      const liveOrchestrationEngine = orchestrationEngine;      const runtime = effectRuntime;      if (!liveOrchestrationEngine || !runtime) {        return;      }      const snapshot = await runtime.runPromise(liveOrchestrationEngine.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 runtime.runPromise(          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}`;        await runtime.runPromise(          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 latestSnapshot = await runtime.runPromise(liveOrchestrationEngine.getSnapshot());        const latestThread = latestSnapshot.threads.find(          (entry) => entry.session?.sessionId === event.sessionId,        );        const existingMessage = latestThread?.messages.find((m) => m.id === assistantMessageId);        const accumulatedText = existingMessage?.text ?? "";        await runtime.runPromise(          liveOrchestrationEngine.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 runtime.runPromise(            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 runtime.runPromise(          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,          }),        );      }    })().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();    const runtime = getEffectRuntime();     // 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));     void runtime      .runPromise(orchestrationEngine.getSnapshot())      .then((snapshot) => {        if (ws.readyState !== ws.OPEN) {          return;        }        const snapshotPush: WsPush = {          type: "push",          channel: ORCHESTRATION_WS_CHANNELS.readModel,          data: {            sequence: snapshot.sequence,            snapshot,          } satisfies OrchestrationReadModelPush,        };        logOutgoingPush(snapshotPush, 1);        ws.send(JSON.stringify(snapshotPush));      })      .catch(() => undefined);     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 };

Lines 1–500 of 849. Load more