/** * Server - HTTP/WebSocket server service interface. * * Owns startup and shutdown lifecycle of the HTTP server, static asset serving, * and WebSocket request routing. * * @module Server */import http from "node:http";import type { Duplex } from "node:stream"; import Mime from "@effect/platform-node/Mime";import {  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  type ClientOrchestrationCommand,  type OrchestrationCommand,  ORCHESTRATION_WS_CHANNELS,  ORCHESTRATION_WS_METHODS,  PROVIDER_SEND_TURN_MAX_IMAGE_BYTES,  ProjectId,  ThreadId,  TerminalEvent,  WS_CHANNELS,  WS_METHODS,  WebSocketRequest,  WsPush,  WsResponse,} from "@t3tools/contracts";import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";import {  Cause,  Effect,  Exit,  FileSystem,  Layer,  Path,  Ref,  Schema,  Scope,  ServiceMap,  Stream,  Struct,} from "effect";import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger";import { GitManager } from "./git/Services/GitManager.ts";import { TerminalManager } from "./terminal/Services/Manager.ts";import { Keybindings } from "./keybindings";import { searchWorkspaceEntries } from "./workspaceEntries";import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine";import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery";import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor";import { ProviderService } from "./provider/Services/ProviderService";import { ProviderHealth } from "./provider/Services/ProviderHealth";import { CheckpointDiffQuery } from "./checkpointing/Services/CheckpointDiffQuery";import { clamp } from "effect/Number";import { Open, resolveAvailableEditors } from "./open";import { ServerConfig } from "./config";import { GitCore } from "./git/Services/GitCore.ts";import { tryHandleProjectFaviconRequest } from "./projectFaviconRoute";import {  ATTACHMENTS_ROUTE_PREFIX,  normalizeAttachmentRelativePath,  resolveAttachmentRelativePath,} from "./attachmentPaths";import {  createAttachmentId,  resolveAttachmentPath,  resolveAttachmentPathById,} from "./attachmentStore.ts";import { parseBase64DataUrl } from "./imageMime.ts";import { AnalyticsService } from "./telemetry/Services/AnalyticsService.ts";import { expandHomePath } from "./os-jank.ts"; /** * ServerShape - Service API for server lifecycle control. */export interface ServerShape {  /**   * Start HTTP and WebSocket listeners.   */  readonly start: Effect.Effect<    http.Server,    ServerLifecycleError,    Scope.Scope | ServerRuntimeServices | ServerConfig | FileSystem.FileSystem | Path.Path  >;   /**   * Wait for process shutdown signals.   */  readonly stopSignal: Effect.Effect<void, never>;} /** * Server - Service tag for HTTP/WebSocket lifecycle management. */export class Server extends ServiceMap.Service<Server, ServerShape>()("t3/wsServer/Server") {} 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.end(    `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,  );} function websocketRawToString(raw: unknown): string | null {  if (typeof raw === "string") {    return raw;  }  if (raw instanceof Uint8Array) {    return Buffer.from(raw).toString("utf8");  }  if (raw instanceof ArrayBuffer) {    return Buffer.from(new Uint8Array(raw)).toString("utf8");  }  if (Array.isArray(raw)) {    const chunks: string[] = [];    for (const chunk of raw) {      if (typeof chunk === "string") {        chunks.push(chunk);        continue;      }      if (chunk instanceof Uint8Array) {        chunks.push(Buffer.from(chunk).toString("utf8"));        continue;      }      if (chunk instanceof ArrayBuffer) {        chunks.push(Buffer.from(new Uint8Array(chunk)).toString("utf8"));        continue;      }      return null;    }    return chunks.join("");  }  return null;} function toPosixRelativePath(input: string): string {  return input.replaceAll("\\", "/");} function resolveWorkspaceWritePath(params: {  workspaceRoot: string;  relativePath: string;  path: Path.Path;}): Effect.Effect<{ absolutePath: string; relativePath: string }, RouteRequestError> {  const normalizedInputPath = params.relativePath.trim();  if (params.path.isAbsolute(normalizedInputPath)) {    return Effect.fail(      new RouteRequestError({        message: "Workspace file path must be relative to the project root.",      }),    );  }   const absolutePath = params.path.resolve(params.workspaceRoot, normalizedInputPath);  const relativeToRoot = toPosixRelativePath(    params.path.relative(params.workspaceRoot, absolutePath),  );  if (    relativeToRoot.length === 0 ||    relativeToRoot === "." ||    relativeToRoot.startsWith("../") ||    relativeToRoot === ".." ||    params.path.isAbsolute(relativeToRoot)  ) {    return Effect.fail(      new RouteRequestError({        message: "Workspace file path must stay within the project root.",      }),    );  }   return Effect.succeed({    absolutePath,    relativePath: relativeToRoot,  });} function stripRequestTag<T extends { _tag: string }>(body: T) {  return Struct.omit(body, ["_tag"]);} function messageFromCause(cause: Cause.Cause<unknown>): string {  const squashed = Cause.squash(cause);  const message =    squashed instanceof Error ? squashed.message.trim() : String(squashed).trim();  return message.length > 0 ? message : Cause.pretty(cause);} export type ServerCoreRuntimeServices =  | OrchestrationEngineService  | ProjectionSnapshotQuery  | CheckpointDiffQuery  | OrchestrationReactor  | ProviderService  | ProviderHealth; export type ServerRuntimeServices =  | ServerCoreRuntimeServices  | GitManager  | GitCore  | TerminalManager  | Keybindings  | Open  | AnalyticsService; export class ServerLifecycleError extends Schema.TaggedErrorClass<ServerLifecycleError>()(  "ServerLifecycleError",  {    operation: Schema.String,    cause: Schema.optional(Schema.Defect),  },) {} class RouteRequestError extends Schema.TaggedErrorClass<RouteRequestError>()("RouteRequestError", {  message: Schema.String,}) {} export const createServer = Effect.fn(function* (): Effect.fn.Return<  http.Server,  ServerLifecycleError,  Scope.Scope | ServerRuntimeServices | ServerConfig | FileSystem.FileSystem | Path.Path> {  const serverConfig = yield* ServerConfig;  const {    port,    cwd,    keybindingsConfigPath,    staticDir,    devUrl,    authToken,    host,    logWebSocketEvents,    autoBootstrapProjectFromCwd,  } = serverConfig;  const availableEditors = resolveAvailableEditors();   const gitManager = yield* GitManager;  const terminalManager = yield* TerminalManager;  const keybindingsManager = yield* Keybindings;  const providerHealth = yield* ProviderHealth;  const git = yield* GitCore;  const fileSystem = yield* FileSystem.FileSystem;  const path = yield* Path.Path;   yield* keybindingsManager.syncDefaultKeybindingsOnStartup.pipe(    Effect.catch((error) =>      Effect.logWarning("failed to sync keybindings defaults on startup", {        path: error.configPath,        detail: error.detail,        cause: error.cause,      }),    ),  );   const providerStatuses = yield* providerHealth.getStatuses;   const clients = yield* Ref.make(new Set<WebSocket>());  const logger = createLogger("ws");   function logOutgoingPush(push: WsPush, recipients: number) {    if (!logWebSocketEvents) return;    logger.event("outgoing push", {      channel: push.channel,      recipients,      payload: push.data,    });  }   const encodePush = Schema.encodeEffect(Schema.fromJsonString(WsPush));  const broadcastPush = Effect.fnUntraced(function* (push: WsPush) {    const message = yield* encodePush(push);    let recipients = 0;    for (const client of yield* Ref.get(clients)) {      if (client.readyState === client.OPEN) {        client.send(message);        recipients += 1;      }    }    logOutgoingPush(push, recipients);  });   const onTerminalEvent = Effect.fnUntraced(function* (event: TerminalEvent) {    yield* broadcastPush({      type: "push",      channel: WS_CHANNELS.terminalEvent,      data: event,    });  });   const normalizeDispatchCommand = Effect.fnUntraced(function* (input: {    readonly command: ClientOrchestrationCommand;  }) {    const normalizeProjectWorkspaceRoot = Effect.fnUntraced(function* (workspaceRoot: string) {      const normalizedWorkspaceRoot = path.resolve(yield* expandHomePath(workspaceRoot.trim()));      const workspaceStat = yield* fileSystem        .stat(normalizedWorkspaceRoot)        .pipe(Effect.catch(() => Effect.succeed(null)));      if (!workspaceStat) {        return yield* new RouteRequestError({          message: `Project directory does not exist: ${normalizedWorkspaceRoot}`,        });      }      if (workspaceStat.type !== "Directory") {        return yield* new RouteRequestError({          message: `Project path is not a directory: ${normalizedWorkspaceRoot}`,        });      }      return normalizedWorkspaceRoot;    });     if (input.command.type === "project.create") {      return {        ...input.command,        workspaceRoot: yield* normalizeProjectWorkspaceRoot(input.command.workspaceRoot),      } satisfies OrchestrationCommand;    }     if (      input.command.type === "project.meta.update" &&      input.command.workspaceRoot !== undefined    ) {      return {        ...input.command,        workspaceRoot: yield* normalizeProjectWorkspaceRoot(input.command.workspaceRoot),      } satisfies OrchestrationCommand;    }     if (input.command.type !== "thread.turn.start") {      return input.command as OrchestrationCommand;    }    const turnStartCommand = input.command;     const normalizedAttachments = yield* Effect.forEach(      turnStartCommand.message.attachments,      (attachment) =>        Effect.gen(function* () {          const parsed = parseBase64DataUrl(attachment.dataUrl);          if (!parsed || !parsed.mimeType.startsWith("image/")) {            return yield* new RouteRequestError({              message: `Invalid image attachment payload for '${attachment.name}'.`,            });          }           const bytes = Buffer.from(parsed.base64, "base64");          if (bytes.byteLength === 0 || bytes.byteLength > PROVIDER_SEND_TURN_MAX_IMAGE_BYTES) {            return yield* new RouteRequestError({              message: `Image attachment '${attachment.name}' is empty or too large.`,            });          }           const attachmentId = createAttachmentId(turnStartCommand.threadId);          if (!attachmentId) {            return yield* new RouteRequestError({              message: "Failed to create a safe attachment id.",            });          }           const persistedAttachment = {            type: "image" as const,            id: attachmentId,            name: attachment.name,            mimeType: parsed.mimeType.toLowerCase(),            sizeBytes: bytes.byteLength,          };           const attachmentPath = resolveAttachmentPath({            stateDir: serverConfig.stateDir,            attachment: persistedAttachment,          });          if (!attachmentPath) {            return yield* new RouteRequestError({              message: `Failed to resolve persisted path for '${attachment.name}'.`,            });          }           yield* fileSystem.makeDirectory(path.dirname(attachmentPath), { recursive: true }).pipe(            Effect.mapError(              () =>                new RouteRequestError({                  message: `Failed to create attachment directory for '${attachment.name}'.`,                }),            ),          );          yield* fileSystem.writeFile(attachmentPath, bytes).pipe(            Effect.mapError(              () =>                new RouteRequestError({                  message: `Failed to persist attachment '${attachment.name}'.`,                }),            ),          );           return persistedAttachment;        }),      { concurrency: 1 },    );     return {      ...turnStartCommand,      message: {        ...turnStartCommand.message,        attachments: normalizedAttachments,      },    } satisfies OrchestrationCommand;  });   // HTTP server — serves static files or redirects to Vite dev server  const httpServer = http.createServer((req, res) => {    const respond = (      statusCode: number,      headers: Record<string, string>,      body?: string | Uint8Array,    ) => {      res.writeHead(statusCode, headers);      res.end(body);    };     void Effect.runPromise(      Effect.gen(function* () {        const url = new URL(req.url ?? "/", `http://localhost:${port}`);        if (tryHandleProjectFaviconRequest(url, res)) {          return;        }         if (url.pathname.startsWith(ATTACHMENTS_ROUTE_PREFIX)) {          const rawRelativePath = url.pathname.slice(ATTACHMENTS_ROUTE_PREFIX.length);          const normalizedRelativePath = normalizeAttachmentRelativePath(rawRelativePath);          if (!normalizedRelativePath) {            respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path");            return;          }           const isIdLookup =            !normalizedRelativePath.includes("/") && !normalizedRelativePath.includes(".");          const filePath = isIdLookup            ? resolveAttachmentPathById({                stateDir: serverConfig.stateDir,                attachmentId: normalizedRelativePath,              })            : resolveAttachmentRelativePath({                stateDir: serverConfig.stateDir,                relativePath: normalizedRelativePath,              });          if (!filePath) {            respond(              isIdLookup ? 404 : 400,              { "Content-Type": "text/plain" },              isIdLookup ? "Not Found" : "Invalid attachment path",            );            return;          }           const fileInfo = yield* fileSystem            .stat(filePath)            .pipe(Effect.catch(() => Effect.succeed(null)));          if (!fileInfo || fileInfo.type !== "File") {            respond(404, { "Content-Type": "text/plain" }, "Not Found");            return;          }           const contentType = Mime.getType(filePath) ?? "application/octet-stream";          res.writeHead(200, {            "Content-Type": contentType,            "Cache-Control": "public, max-age=31536000, immutable",          });          const streamExit = yield* Stream.runForEach(fileSystem.stream(filePath), (chunk) =>            Effect.sync(() => {              if (!res.destroyed) {                res.write(chunk);              }            }),          ).pipe(Effect.exit);          if (streamExit._tag === "Failure") {            if (!res.destroyed) {              res.destroy();            }            return;          }          if (!res.writableEnded) {            res.end();          }          return;        } 

Lines 1–500 of 1022. Load more