From da3d8e2ecb572e107b4ceec41d8966fe2f8f3db1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 24 Feb 2026 17:18:49 -0800 Subject: [PATCH] nice refactor --- apps/server/src/main.test.ts | 37 +- apps/server/src/main.ts | 50 +- apps/server/src/wsServer.test.ts | 218 ++++---- apps/server/src/wsServer.ts | 825 ++++++++++++++----------------- 4 files changed, 528 insertions(+), 602 deletions(-) diff --git a/apps/server/src/main.test.ts b/apps/server/src/main.test.ts index 0890be8addc9e3938b95235e196b49f7cc582a9a..a75dbb593769e36c9ac0c4087d7087aa6443bb32 100644 --- a/apps/server/src/main.test.ts +++ b/apps/server/src/main.test.ts @@ -1,3 +1,4 @@ +import * as Http from "node:http"; import { NodeServices } from "@effect/platform-node"; import { assert, it, vi } from "@effect/vitest"; import * as ConfigProvider from "effect/ConfigProvider"; @@ -10,13 +11,17 @@ import { CliConfig, makeCliCommand, type CliConfigShape } from "./main"; import { Open, type OpenShape } from "./open"; import { Server, type ServerShape, type ServerOptions } from "./wsServer"; -const start = vi.fn(async () => undefined); -const stop = vi.fn(async () => undefined); -const createServer = vi.fn((_: ServerOptions) => ({ - start, - stop, - httpServer: undefined as never, -})); +const start = vi.fn(() => undefined); +const stop = vi.fn(() => undefined); +const createServer = vi.fn((_: ServerOptions) => + Effect.acquireRelease( + Effect.sync(() => { + start(); + return {} as unknown as Http.Server; + }), + () => Effect.sync(() => stop()), + ), +); const findAvailablePort = vi.fn((preferred: number) => Effect.succeed(preferred)); const testLayer = Layer.mergeAll( @@ -47,13 +52,17 @@ const runCli = ( beforeEach(() => { vi.clearAllMocks(); - start.mockImplementation(async () => undefined); - stop.mockImplementation(async () => undefined); - createServer.mockImplementation((_: ServerOptions) => ({ - start, - stop, - httpServer: undefined as never, - })); + start.mockImplementation(() => undefined); + stop.mockImplementation(() => undefined); + createServer.mockImplementation((_: ServerOptions) => + Effect.acquireRelease( + Effect.sync(() => { + start(); + return {} as unknown as Http.Server; + }), + () => Effect.sync(() => stop()), + ), + ); findAvailablePort.mockImplementation((preferred: number) => Effect.succeed(preferred)); }); diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index 4e6e840bcce4351660d6157f2d4e86d00afdc271..26d1d70beb210f47f02f44ee6e3829691a406902 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -6,7 +6,12 @@ import { Command, Flag } from "effect/unstable/cli"; import { fixPath } from "./fixPath"; import { Open } from "./open"; -import { Server } from "./wsServer"; +import { + makeServerPersistenceLayer, + makeServerProviderLayer, + makeServerRuntimeServicesLayer, + Server, +} from "./wsServer"; const DEFAULT_PORT = 3773; @@ -168,29 +173,26 @@ const makeServerProgram = Effect.fn(function* (input: CliInput) { }); } - yield* Effect.acquireRelease( - Effect.tryPromise({ - try: async () => { - const server = createServer({ - port: config.port, - host: config.mode === "desktop" ? "127.0.0.1" : undefined, - cwd: config.cwd, - autoBootstrapProjectFromCwd: config.mode === "web", - stateDir: config.stateDir, - staticDir: config.staticDir, - devUrl: config.devUrl?.href, - authToken: config.authToken, - }); - await server.start(); - return server; - }, - catch: (cause) => new StartupError({ message: "Failed to start server", cause }), - }), - (server) => - Effect.tryPromise({ - try: () => server.stop(), - catch: (cause) => new StartupError({ message: "Failed to stop server", cause }), - }).pipe(Effect.catch((error) => Effect.logError("failed to stop server cleanly", { error }))), + const persistenceLayer = makeServerPersistenceLayer(config.stateDir); + const providerLayer = makeServerProviderLayer(config.stateDir); + const infrastructureLayer = providerLayer.pipe(Layer.provideMerge(persistenceLayer)); + const serverRuntimeLayer = Layer.merge( + makeServerRuntimeServicesLayer().pipe(Layer.provide(infrastructureLayer)), + infrastructureLayer, + ); + + yield* createServer({ + port: config.port, + host: config.mode === "desktop" ? "127.0.0.1" : undefined, + cwd: config.cwd, + autoBootstrapProjectFromCwd: config.mode === "web", + stateDir: config.stateDir, + staticDir: config.staticDir, + devUrl: config.devUrl?.href, + authToken: config.authToken, + }).pipe( + Effect.provide(serverRuntimeLayer), + Effect.mapError((cause) => new StartupError({ message: "Failed to start server", cause })), ); const url = `http://localhost:${config.port}`; diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index b2f2eba5b838efa6fd641fa59d40b3ccf53456da..ab92b4353255420db7f649288593c2d1573b0810 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -1,11 +1,12 @@ +import * as Http from "node:http"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { EventEmitter } from "node:events"; -import { Effect, Layer, PubSub, Stream } from "effect"; +import { Effect, Exit, Layer, PubSub, Scope, Stream } from "effect"; import { describe, expect, it, afterEach, vi } from "vitest"; -import { createServer } from "./wsServer"; +import { createServer, makeServerProviderLayer, makeServerRuntimeServicesLayer } from "./wsServer"; import WebSocket from "ws"; import { @@ -49,7 +50,8 @@ interface PendingMessages { const pendingBySocket = new WeakMap(); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); -const asProviderSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); +const asProviderSessionId = (value: string): ProviderSessionId => + ProviderSessionId.makeUnsafe(value); const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asProviderItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); @@ -286,7 +288,8 @@ function mergeWithDefaultsForTest(custom: KeybindingsConfig): ResolvedKeybinding } describe("WebSocket Server", () => { - let server: ReturnType | null = null; + let server: Http.Server | null = null; + let serverScope: Scope.Closeable | null = null; const connections: WebSocket[] = []; const tempDirs: string[] = []; @@ -296,7 +299,7 @@ describe("WebSocket Server", () => { return dir; } - function createTestServer( + async function createTestServer( options: { persistenceLayer?: Layer.Layer; cwd?: string; @@ -312,23 +315,53 @@ describe("WebSocket Server", () => { }; terminalManager?: TerminalManager; } = {}, - ): ReturnType { + ): Promise { + if (serverScope) { + throw new Error("Test server is already running"); + } + const stateDir = options.stateDir ?? makeTempDir("t3code-ws-state-"); - return createServer({ - port: 0, - cwd: options.cwd ?? "/test/project", - stateDir, - persistenceLayer: options.persistenceLayer ?? SqlitePersistenceMemory, - ...(options.autoBootstrapProjectFromCwd !== undefined - ? { autoBootstrapProjectFromCwd: options.autoBootstrapProjectFromCwd } - : {}), - ...(options.devUrl ? { devUrl: options.devUrl } : {}), - ...(options.authToken ? { authToken: options.authToken } : {}), - ...(options.providerLayer ? { providerLayer: options.providerLayer } : {}), - ...(options.open ? { open: options.open } : {}), - ...(options.gitManager ? { gitManager: options.gitManager as never } : {}), - ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}), - }); + const scope = await Effect.runPromise(Scope.make("sequential")); + const persistenceLayer = options.persistenceLayer ?? SqlitePersistenceMemory; + const providerLayer = options.providerLayer ?? makeServerProviderLayer(stateDir); + const infrastructureLayer = providerLayer.pipe(Layer.provideMerge(persistenceLayer)); + const runtimeLayer = Layer.merge( + makeServerRuntimeServicesLayer().pipe(Layer.provide(infrastructureLayer)), + infrastructureLayer, + ); + const runtimeServices = await Effect.runPromise( + Layer.build(runtimeLayer).pipe(Scope.provide(scope)), + ); + + try { + const runtime = await Effect.runPromise( + createServer({ + port: 0, + cwd: options.cwd ?? "/test/project", + stateDir, + ...(options.autoBootstrapProjectFromCwd !== undefined + ? { autoBootstrapProjectFromCwd: options.autoBootstrapProjectFromCwd } + : {}), + ...(options.devUrl ? { devUrl: options.devUrl } : {}), + ...(options.authToken ? { authToken: options.authToken } : {}), + ...(options.open ? { open: options.open } : {}), + ...(options.gitManager ? { gitManager: options.gitManager as never } : {}), + ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}), + }).pipe(Effect.provide(runtimeServices), Scope.provide(scope)), + ); + serverScope = scope; + return runtime; + } catch (error) { + await Effect.runPromise(Scope.close(scope, Exit.void)); + throw error; + } + } + + async function closeTestServer() { + if (!serverScope) return; + const scope = serverScope; + serverScope = null; + await Effect.runPromise(Scope.close(scope, Exit.void)); } afterEach(async () => { @@ -336,9 +369,7 @@ describe("WebSocket Server", () => { ws.close(); } connections.length = 0; - if (server) { - await server.stop(); - } + await closeTestServer(); server = null; for (const dir of tempDirs.splice(0, tempDirs.length)) { fs.rmSync(dir, { recursive: true, force: true }); @@ -347,10 +378,9 @@ describe("WebSocket Server", () => { }); it("sends welcome message on connect", async () => { - server = createTestServer({ cwd: "/test/project" }); + server = await createTestServer({ cwd: "/test/project" }); // Get the actual port after listen - await server.start(); - const addr = server.httpServer.address(); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); @@ -367,12 +397,11 @@ describe("WebSocket Server", () => { }); it("bootstraps the cwd project on startup when enabled", async () => { - server = createTestServer({ + server = await createTestServer({ cwd: "/test/bootstrap-workspace", autoBootstrapProjectFromCwd: true, }); - await server.start(); - const addr = server.httpServer.address(); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); @@ -402,12 +431,11 @@ describe("WebSocket Server", () => { // Keep test output clean while verifying websocket logs. }); - server = createTestServer({ + server = await createTestServer({ cwd: "/test/project", devUrl: "http://localhost:5173", }); - await server.start(); - const addr = server.httpServer.address(); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); @@ -430,9 +458,8 @@ describe("WebSocket Server", () => { it("responds to server.getConfig", async () => { const fakeHome = makeTempDir("t3code-home-"); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -459,9 +486,8 @@ describe("WebSocket Server", () => { }, }; - server = createTestServer({ cwd: "/my/workspace", open: openService }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace", open: openService }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -491,9 +517,8 @@ describe("WebSocket Server", () => { ); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -528,9 +553,8 @@ describe("WebSocket Server", () => { ); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -563,9 +587,8 @@ describe("WebSocket Server", () => { ); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -598,9 +621,8 @@ describe("WebSocket Server", () => { ); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -637,9 +659,8 @@ describe("WebSocket Server", () => { ); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -690,9 +711,8 @@ describe("WebSocket Server", () => { ); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -721,9 +741,8 @@ describe("WebSocket Server", () => { fs.writeFileSync(path.join(configDir, "keybindings.json"), "{not-json", "utf8"); vi.spyOn(os, "homedir").mockReturnValue(fakeHome); - server = createTestServer({ cwd: "/my/workspace" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/my/workspace" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -745,9 +764,8 @@ describe("WebSocket Server", () => { }); it("returns error for unknown methods", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -762,9 +780,8 @@ describe("WebSocket Server", () => { }); it("returns error when requesting turn diff for unknown thread", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -781,9 +798,8 @@ describe("WebSocket Server", () => { }); it("returns error when requesting full thread diff for unknown thread", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -799,9 +815,8 @@ describe("WebSocket Server", () => { }); it("returns retryable error when requested turn exceeds current checkpoint turn count", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -873,12 +888,11 @@ describe("WebSocket Server", () => { }; const providerLayer = Layer.succeed(ProviderService, providerService); - server = createTestServer({ + server = await createTestServer({ cwd: "/test", providerLayer, }); - await server.start(); - const addr = server.httpServer.address(); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -941,7 +955,9 @@ describe("WebSocket Server", () => { const domainPush = await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => { const event = push.data as { type?: string; payload?: { messageId?: string; text?: string } }; - return event.type === "thread.message-sent" && event.payload?.messageId === "assistant:item-1"; + return ( + event.type === "thread.message-sent" && event.payload?.messageId === "assistant:item-1" + ); }); const domainEvent = domainPush.data as { @@ -956,12 +972,11 @@ describe("WebSocket Server", () => { it("routes terminal RPC methods and broadcasts terminal events", async () => { const cwd = makeTempDir("t3code-ws-terminal-cwd-"); const terminalManager = new MockTerminalManager(); - server = createTestServer({ + server = await createTestServer({ cwd: "/test", terminalManager: terminalManager as unknown as TerminalManager, }); - await server.start(); - const addr = server.httpServer.address(); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1027,24 +1042,22 @@ describe("WebSocket Server", () => { it("detaches terminal event listener on stop for injected manager", async () => { const terminalManager = new MockTerminalManager(); - server = createTestServer({ + server = await createTestServer({ cwd: "/test", terminalManager: terminalManager as unknown as TerminalManager, }); - await server.start(); expect(terminalManager.listenerCount("event")).toBe(1); - await server.stop(); + await closeTestServer(); server = null; expect(terminalManager.listenerCount("event")).toBe(0); }); it("returns validation errors for invalid terminal open params", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1061,9 +1074,8 @@ describe("WebSocket Server", () => { }); it("handles invalid JSON gracefully", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1096,9 +1108,8 @@ describe("WebSocket Server", () => { }); it("returns errors for removed projects CRUD methods", async () => { - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1134,9 +1145,8 @@ describe("WebSocket Server", () => { fs.mkdirSync(path.join(workspace, ".git"), { recursive: true }); fs.writeFileSync(path.join(workspace, ".git", "HEAD"), "ref: refs/heads/main\n", "utf8"); - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1161,9 +1171,8 @@ describe("WebSocket Server", () => { it("supports git methods over websocket", async () => { const repoCwd = makeTempDir("t3code-ws-git-project-"); - server = createTestServer({ cwd: "/test" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1209,9 +1218,8 @@ describe("WebSocket Server", () => { runStackedAction: vi.fn(), }; - server = createTestServer({ cwd: "/test", gitManager }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test", gitManager }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1232,9 +1240,8 @@ describe("WebSocket Server", () => { runStackedAction: vi.fn().mockRejectedValue(new Error("Cannot push from detached HEAD.")), }; - server = createTestServer({ cwd: "/test", gitManager }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test", gitManager }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; const ws = await connectWs(port); @@ -1254,9 +1261,8 @@ describe("WebSocket Server", () => { }); it("rejects websocket connections without a valid auth token", async () => { - server = createTestServer({ cwd: "/test", authToken: "secret-token" }); - await server.start(); - const addr = server.httpServer.address(); + server = await createTestServer({ cwd: "/test", authToken: "secret-token" }); + const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; await expect(connectWs(port)).rejects.toThrow("WebSocket connection failed"); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index 5be76b1cd954716844685dd81ee3af5bcdd174dc..79deff0cf3a1b5f9ae9e88a98fb8fd1162b5445e 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -12,11 +12,12 @@ import { WS_CHANNELS, WS_METHODS, WebSocketRequest, + WebSocketResponse, WsPush, WsResponse, } from "@t3tools/contracts"; -import { NodeServices } from "@effect/platform-node"; -import { Effect, Exit, Layer, ManagedRuntime, Schema, Scope, ServiceMap, Stream } from "effect"; +import { NodeHttpServer, NodeServices } from "@effect/platform-node"; +import { Cause, Effect, Exit, Layer, Schema, Scope, ServiceMap, Stream } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; @@ -40,34 +41,21 @@ import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderComma import { OrchestrationProjectionPipelineLive } from "./orchestration/Layers/ProjectionPipeline"; import { OrchestrationProjectionSnapshotQueryLive } from "./orchestration/Layers/ProjectionSnapshotQuery"; import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRuntimeIngestion"; -import { - OrchestrationEngineService, - type OrchestrationEngineShape, -} from "./orchestration/Services/OrchestrationEngine"; -import { - ProjectionSnapshotQuery, - type ProjectionSnapshotQueryShape, -} from "./orchestration/Services/ProjectionSnapshotQuery"; +import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine"; +import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery"; import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor"; import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; -import assert from "node:assert"; import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore"; import { OrchestrationCommandReceiptRepositoryLive } from "./persistence/Layers/OrchestrationCommandReceipts"; import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry"; import { makeProviderServiceLive } from "./provider/Layers/ProviderService"; import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory"; -import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; +import { ProviderService } from "./provider/Services/ProviderService"; import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; import { ProviderSessionRuntimeRepositoryLive } from "./persistence/Layers/ProviderSessionRuntime"; -import { - CheckpointStore, - type CheckpointStoreShape, -} from "./checkpointing/Services/CheckpointStore"; -import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger"; +import { CheckpointStore } from "./checkpointing/Services/CheckpointStore"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlError from "effect/unstable/sql/SqlError"; -import * as Migrator from "effect/unstable/sql/Migrator"; import { clamp } from "effect/Number"; import { OpenDefaultService, type OpenShape } from "./open"; @@ -92,37 +80,24 @@ export interface ServerOptions { cwd: string; autoBootstrapProjectFromCwd?: boolean | undefined; stateDir?: string | undefined; - persistenceLayer?: - | Layer.Layer - | undefined; staticDir?: string | undefined; devUrl?: string | undefined; logWebSocketEvents?: boolean | undefined; gitManager?: GitManager | undefined; terminalManager?: TerminalManager | undefined; - providerLayer?: Layer.Layer | undefined; authToken?: string | undefined; open?: OpenShape | undefined; } -export interface ServerRuntime { - readonly start: () => Promise; - readonly stop: () => Promise; - readonly httpServer: http.Server; -} - export interface ServerShape { - readonly createServer: (options: ServerOptions) => ServerRuntime; + readonly createServer: ( + options: ServerOptions, + ) => Effect.Effect; readonly stopSignal: Effect.Effect; } export class Server extends ServiceMap.Service()("server/Server") {} -export const ServerLive = Layer.succeed(Server, { - createServer, - stopSignal: Effect.never, -} satisfies ServerShape); - const isServerNotRunningError = (error: unknown): boolean => { if (!(error instanceof Error)) return false; const maybeCode = (error as NodeJS.ErrnoException).code; @@ -195,47 +170,104 @@ function checkpointRefForThreadTurn(threadId: string, turnCount: number): Checkp return CheckpointRef.makeUnsafe(`${CHECKPOINT_REFS_PREFIX}/${encodedThreadId}/turn/${turnCount}`); } -type EffectRuntime = ManagedRuntime.ManagedRuntime< +export type ServerCoreRuntimeServices = | OrchestrationEngineService | ProjectionSnapshotQuery - | ProviderService | CheckpointStore - | OrchestrationReactor, - unknown ->; + | OrchestrationReactor; -export function createServer(options: ServerOptions): ServerRuntime { +export type ServerRuntimeServices = ServerCoreRuntimeServices | ProviderService; + +export function makeServerPersistenceLayer(stateDir: string) { + const dbPath = path.join(stateDir, "orchestration.sqlite"); + return makeSqlitePersistenceLive(dbPath); +} + +export function makeServerProviderLayer( + stateDir: string, +): Layer.Layer { + const providerLogsDir = path.join(stateDir, "logs", "providers"); + const codexAdapterLayer = makeCodexAdapterLive({ + nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), + }); + const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(codexAdapterLayer)); + const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRuntimeRepositoryLive), + ); + return makeProviderServiceLive({ + canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"), + }).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer)); +} + +export function makeServerRuntimeServicesLayer(): Layer.Layer< + ServerCoreRuntimeServices, + unknown, + SqlClient.SqlClient | ProviderService +> { + const orchestrationLayer = OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationProjectionPipelineLive), + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + ); + const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); + + const runtimeServicesLayer = Layer.mergeAll( + orchestrationLayer, + OrchestrationProjectionSnapshotQueryLive, + checkpointStoreLayer, + ); + const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( + Layer.provideMerge(runtimeServicesLayer), + ); + const providerCommandReactorLayer = ProviderCommandReactorLive.pipe( + Layer.provideMerge(runtimeServicesLayer), + ); + const checkpointReactorLayer = CheckpointReactorLive.pipe( + Layer.provideMerge(runtimeServicesLayer), + ); + const orchestrationReactorLayer = OrchestrationReactorLive.pipe( + Layer.provideMerge(runtimeIngestionLayer), + Layer.provideMerge(providerCommandReactorLayer), + Layer.provideMerge(checkpointReactorLayer), + ); + + return orchestrationReactorLayer; +} + +export class ServerLifecycleError extends Schema.TaggedErrorClass()( + "ServerLifecycleError", + { + operation: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) {} + +class RouteRequestError extends Schema.TaggedErrorClass()("RouteRequestError", { + message: Schema.String, +}) {} + +export const createServer = Effect.fn(function* ( + options: ServerOptions, +): Effect.fn.Return { const { port, host, cwd, autoBootstrapProjectFromCwd = false, - stateDir, - persistenceLayer: customPersistenceLayer, staticDir, devUrl, logWebSocketEvents: explicitLogWsEvents, gitManager = new GitManager(), terminalManager = new TerminalManager(), - providerLayer: customProviderLayer, authToken, open: openService = OpenDefaultService, } = options; - const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata"); - let effectRuntime: EffectRuntime | null = null; - let orchestrationEngine: OrchestrationEngineShape | null = null; - let projectionSnapshotQuery: ProjectionSnapshotQueryShape | null = null; - let providerService: ProviderServiceShape | null = null; - let checkpointStore: CheckpointStoreShape | null = null; const clients = new Set(); const logger = createLogger("ws"); const logWebSocketEvents = explicitLogWsEvents ?? parseBooleanEnv(process.env.T3CODE_LOG_WS_EVENTS) ?? Boolean(devUrl); let keybindingsConfig = loadResolvedKeybindingsConfig(logger); - let orchestrationDomainEventLogger: EventNdjsonLogger | undefined; - let orchestrationCommandLogger: EventNdjsonLogger | undefined; - let subscriptionsScope: Scope.Closeable | null = null; function logOutgoingPush(push: WsPush, recipients: number) { if (!logWebSocketEvents) return; @@ -246,86 +278,6 @@ export function createServer(options: ServerOptions): ServerRuntime { }); } - const getOrchestrationEngine = () => { - assert(orchestrationEngine, "Orchestration engine is not started"); - return orchestrationEngine; - }; - - const getEffectRuntime = () => { - assert(effectRuntime, "Effect runtime is not started"); - return effectRuntime; - }; - - const getProjectionSnapshotQuery = () => { - assert(projectionSnapshotQuery, "Projection snapshot query is not started"); - return projectionSnapshotQuery; - }; - - const getCheckpointStore = () => { - assert(checkpointStore, "Checkpoint store is not started"); - return checkpointStore; - }; - - async function bootstrapProjectForCwd() { - if (!autoBootstrapProjectFromCwd) { - return; - } - - const runtime = getEffectRuntime(); - const snapshotQuery = getProjectionSnapshotQuery(); - const engine = getOrchestrationEngine(); - const snapshot = await runtime.runPromise(snapshotQuery.getSnapshot()); - const existing = snapshot.projects.find((project) => project.workspaceRoot === cwd); - if (existing) { - return; - } - - const createdAt = new Date().toISOString(); - const projectName = path.basename(cwd) || "project"; - await runtime.runPromise( - engine.dispatchUnknownCommand({ - type: "project.create", - commandId: crypto.randomUUID(), - projectId: crypto.randomUUID(), - title: projectName, - workspaceRoot: cwd, - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - } - - const attachSubscriptionss = Effect.gen(function* () { - const orchestrationEngine = yield* OrchestrationEngineService; - const orchestrationReactor = yield* OrchestrationReactor; - subscriptionsScope = yield* Scope.make("sequential"); - - yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => - Effect.sync(() => { - orchestrationDomainEventLogger?.write({ - observedAt: new Date().toISOString(), - 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); - }), - ).pipe(Effect.forkIn(subscriptionsScope)); - - yield* orchestrationReactor.start.pipe(Scope.provide(subscriptionsScope)); - }); - const onTerminalEvent = (event: TerminalEvent) => { const push: WsPush = { type: "push", @@ -342,7 +294,6 @@ export function createServer(options: ServerOptions): ServerRuntime { } logOutgoingPush(push, recipients); }; - terminalManager.on("event", onTerminalEvent); // HTTP server — serves static files or redirects to Vite dev server const httpServer = http.createServer((req, res) => { @@ -403,104 +354,85 @@ export function createServer(options: ServerOptions): ServerRuntime { // 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; + const closeWebSocketServer = Effect.callback((resume) => { + wss.close((error) => { + if (error && !isServerNotRunningError(error)) { + resume( + Effect.fail( + new ServerLifecycleError({ operation: "closeWebSocketServer", cause: error }), + ), + ); + } else { + resume(Effect.void); } - - 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); - - // 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 closeAllClients = () => { + for (const client of clients) { + client.close(); + } + clients.clear(); + }; - ws.on("message", (raw) => { - void handleMessage(ws, raw); - }); + const listenOptions = host ? { host, port } : { port }; - ws.on("close", () => { - clients.delete(ws); - }); + const orchestrationEngine = yield* OrchestrationEngineService; + const projectionReadModelQuery = yield* ProjectionSnapshotQuery; + const liveCheckpointStore = yield* CheckpointStore; + const liveProviderService = yield* ProviderService; + const orchestrationReactor = yield* OrchestrationReactor; - ws.on("error", () => { - clients.delete(ws); - }); - }); + const subscriptionsScope = yield* Scope.make("sequential"); + yield* Effect.addFinalizer(() => + Scope.close(subscriptionsScope, Exit.void).pipe(Effect.orElseSucceed(() => undefined)), + ); - async function handleMessage(ws: WebSocket, raw: unknown) { - const messageText = websocketRawToString(raw); - if (messageText === null) { - const errorResponse: WsResponse = { - id: "unknown", - error: { message: "Invalid request format" }, + yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => + Effect.sync(() => { + const push: WsPush = { + type: "push", + channel: ORCHESTRATION_WS_CHANNELS.domainEvent, + data: event, }; - ws.send(JSON.stringify(errorResponse)); - return; - } + 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); + }), + ).pipe(Effect.forkIn(subscriptionsScope)); - let request: WebSocketRequest; - try { - request = Schema.decodeUnknownSync(Schema.fromJsonString(WebSocketRequest))(messageText); - } catch { - const errorResponse: WsResponse = { - id: "unknown", - error: { message: "Invalid request format" }, - }; - ws.send(JSON.stringify(errorResponse)); - return; - } + yield* orchestrationReactor.start.pipe(Scope.provide(subscriptionsScope)); - 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)); + if (autoBootstrapProjectFromCwd) { + const snapshot = yield* projectionReadModelQuery.getSnapshot(); + const existing = snapshot.projects.find((project) => project.workspaceRoot === cwd); + if (!existing) { + const createdAt = new Date().toISOString(); + const projectName = path.basename(cwd) || "project"; + yield* orchestrationEngine.dispatchUnknownCommand({ + type: "project.create", + commandId: crypto.randomUUID(), + projectId: crypto.randomUUID(), + title: projectName, + workspaceRoot: cwd, + defaultModel: "gpt-5-codex", + createdAt, + }); } } - async function routeRequest(request: WebSocketRequest): Promise { - const orchestrationEngine = getOrchestrationEngine(); - const projectionReadModelQuery = getProjectionSnapshotQuery(); - const effectRuntime = getEffectRuntime(); - const liveCheckpointStore = getCheckpointStore(); - - const computeTurnDiff = async (input: { + const routeRequest = Effect.fn(function* (request: WebSocketRequest) { + const computeTurnDiff = Effect.fn(function* (input: { readonly threadId: string; readonly fromTurnCount: number; readonly toTurnCount: number; - }) => { + }) { if (input.fromTurnCount > input.toTurnCount) { throw new Error( `Invalid turn diff range for thread '${input.threadId}': from ${input.fromTurnCount} exceeds to ${input.toTurnCount}.`, @@ -516,7 +448,7 @@ export function createServer(options: ServerOptions): ServerRuntime { }; } - const snapshot = await effectRuntime.runPromise(projectionReadModelQuery.getSnapshot()); + const snapshot = yield* projectionReadModelQuery.getSnapshot(); const thread = snapshot.threads.find((entry) => entry.id === input.threadId); if (!thread) { throw new Error(`Thread '${input.threadId}' not found.`); @@ -562,20 +494,19 @@ export function createServer(options: ServerOptions): ServerRuntime { ); } - const [fromExists, toExists] = await Promise.all([ - effectRuntime.runPromise( + const [fromExists, toExists] = yield* Effect.all( + [ liveCheckpointStore.hasCheckpointRef({ cwd: workspaceCwd, checkpointRef: fromCheckpointRef, }), - ), - effectRuntime.runPromise( liveCheckpointStore.hasCheckpointRef({ cwd: workspaceCwd, checkpointRef: toCheckpointRef, }), - ), - ]); + ], + { concurrency: "unbounded" }, + ); if (!fromExists) { throw new Error( `Filesystem checkpoint is unavailable for turn ${input.fromTurnCount} in thread ${input.threadId}.`, @@ -587,304 +518,282 @@ export function createServer(options: ServerOptions): ServerRuntime { ); } - const diff = await effectRuntime.runPromise( - liveCheckpointStore.diffCheckpoints({ - cwd: workspaceCwd, - fromCheckpointRef, - toCheckpointRef, - fallbackFromToHead: false, - }), - ); + const diff = yield* liveCheckpointStore.diffCheckpoints({ + cwd: workspaceCwd, + fromCheckpointRef, + toCheckpointRef, + fallbackFromToHead: false, + }); return { threadId: input.threadId, fromTurnCount: input.fromTurnCount, toTurnCount: input.toTurnCount, diff, }; - }; + }); switch (request.body._tag) { - case ORCHESTRATION_WS_METHODS.getSnapshot: { - return effectRuntime.runPromise(projectionReadModelQuery.getSnapshot()); - } + case ORCHESTRATION_WS_METHODS.getSnapshot: + return yield* projectionReadModelQuery.getSnapshot(); case ORCHESTRATION_WS_METHODS.dispatchCommand: { - orchestrationCommandLogger?.write({ - observedAt: new Date().toISOString(), - command: request.body.command, - }); - return effectRuntime.runPromise(orchestrationEngine.dispatch(request.body.command)); + const { command } = request.body; + return yield* orchestrationEngine.dispatch(command); } case ORCHESTRATION_WS_METHODS.getTurnDiff: { const body = stripRequestTag(request.body); - return computeTurnDiff(body); + return yield* computeTurnDiff(body); } case ORCHESTRATION_WS_METHODS.getFullThreadDiff: { const body = stripRequestTag(request.body); - return computeTurnDiff({ + return yield* computeTurnDiff({ threadId: body.threadId, fromTurnCount: 0, toTurnCount: body.toTurnCount, }); } - case ORCHESTRATION_WS_METHODS.replayEvents: - return effectRuntime.runPromise( - Stream.runCollect( - orchestrationEngine.readEvents( - clamp(request.body.fromSequenceExclusive, { - maximum: Number.MAX_SAFE_INTEGER, - minimum: 0, - }), - ), - ).pipe(Effect.map((events) => Array.from(events))), - ); + case ORCHESTRATION_WS_METHODS.replayEvents: { + const { fromSequenceExclusive } = request.body; + return yield* Stream.runCollect( + orchestrationEngine.readEvents( + clamp(fromSequenceExclusive, { + maximum: Number.MAX_SAFE_INTEGER, + minimum: 0, + }), + ), + ).pipe(Effect.map((events) => Array.from(events))); + } - case WS_METHODS.projectsSearchEntries: - return searchWorkspaceEntries(stripRequestTag(request.body)); + case WS_METHODS.projectsSearchEntries: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => searchWorkspaceEntries(body)); + } case WS_METHODS.shellOpenInEditor: { - await Effect.runPromise(openService.openInEditor(stripRequestTag(request.body))); - return undefined; + const body = stripRequestTag(request.body); + return yield* openService.openInEditor(body); } - case WS_METHODS.gitStatus: - return gitManager.status(stripRequestTag(request.body)); + case WS_METHODS.gitStatus: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => gitManager.status(body)); + } + + case WS_METHODS.gitPull: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => pullGitBranch(body)); + } - case WS_METHODS.gitPull: - return pullGitBranch(stripRequestTag(request.body)); + case WS_METHODS.gitRunStackedAction: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => gitManager.runStackedAction(body)); + } - case WS_METHODS.gitRunStackedAction: - return gitManager.runStackedAction(stripRequestTag(request.body)); - case WS_METHODS.gitListBranches: - return listGitBranches(stripRequestTag(request.body)); + case WS_METHODS.gitListBranches: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => listGitBranches(body)); + } - case WS_METHODS.gitCreateWorktree: - return createGitWorktree(stripRequestTag(request.body)); + case WS_METHODS.gitCreateWorktree: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => createGitWorktree(body)); + } - case WS_METHODS.gitRemoveWorktree: - return removeGitWorktree(stripRequestTag(request.body)); + case WS_METHODS.gitRemoveWorktree: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => removeGitWorktree(body)); + } - case WS_METHODS.gitCreateBranch: - return createGitBranch(stripRequestTag(request.body)); + case WS_METHODS.gitCreateBranch: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => createGitBranch(body)); + } - case WS_METHODS.gitCheckout: - return checkoutGitBranch(stripRequestTag(request.body)); + case WS_METHODS.gitCheckout: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => checkoutGitBranch(body)); + } - case WS_METHODS.gitInit: - return initGitRepo(stripRequestTag(request.body)); + case WS_METHODS.gitInit: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => initGitRepo(body)); + } - case WS_METHODS.terminalOpen: - return terminalManager.open(stripRequestTag(request.body)); + case WS_METHODS.terminalOpen: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => terminalManager.open(body)); + } - case WS_METHODS.terminalWrite: - await terminalManager.write(stripRequestTag(request.body)); - return undefined; + case WS_METHODS.terminalWrite: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => terminalManager.write(body)); + } - case WS_METHODS.terminalResize: - await terminalManager.resize(stripRequestTag(request.body)); - return undefined; + case WS_METHODS.terminalResize: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => terminalManager.resize(body)); + } - case WS_METHODS.terminalClear: - await terminalManager.clear(stripRequestTag(request.body)); - return undefined; + case WS_METHODS.terminalClear: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => terminalManager.clear(body)); + } - case WS_METHODS.terminalRestart: - return terminalManager.restart(stripRequestTag(request.body)); + case WS_METHODS.terminalRestart: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => terminalManager.restart(body)); + } - case WS_METHODS.terminalClose: - await terminalManager.close(stripRequestTag(request.body)); - return undefined; + case WS_METHODS.terminalClose: { + const body = stripRequestTag(request.body); + return yield* Effect.promise(() => terminalManager.close(body)); + } case WS_METHODS.serverGetConfig: - return { - cwd, - keybindings: keybindingsConfig, - }; + return { cwd, keybindings: keybindingsConfig }; - case WS_METHODS.serverUpsertKeybinding: - keybindingsConfig = upsertKeybindingRule(logger, stripRequestTag(request.body)); - return { - keybindings: keybindingsConfig, - }; + case WS_METHODS.serverUpsertKeybinding: { + const body = stripRequestTag(request.body); + keybindingsConfig = upsertKeybindingRule(logger, body); + return { keybindings: keybindingsConfig }; + } default: { const _exhaustiveCheck: never = request.body; - throw new Error(`Unknown method: ${String(_exhaustiveCheck)}`); - } - } - } - - async function createEffectRuntime() { - const dbPath = path.join(resolvedStateDir, "orchestration.sqlite"); - const providerLogsDir = path.join(resolvedStateDir, "logs", "providers"); - const orchestrationLogsDir = path.join(resolvedStateDir, "logs", "orchestration"); - orchestrationDomainEventLogger = makeEventNdjsonLogger( - path.join(orchestrationLogsDir, "orchestration-domain.ndjson"), - ); - orchestrationCommandLogger = makeEventNdjsonLogger( - path.join(orchestrationLogsDir, "orchestration-command.ndjson"), - ); - const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath); - const orchestrationLayer = OrchestrationEngineLive.pipe( - Layer.provide(OrchestrationProjectionPipelineLive), - Layer.provide(OrchestrationEventStoreLive), - Layer.provide(OrchestrationCommandReceiptRepositoryLive), - ); - const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); - const providerLayer = - customProviderLayer ?? - (() => { - const codexAdapterLayer = makeCodexAdapterLive({ - nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), + return yield* new RouteRequestError({ + message: `Unknown method: ${String(_exhaustiveCheck)}`, }); - const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe( - Layer.provide(codexAdapterLayer), - ); - const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( - Layer.provide(ProviderSessionRuntimeRepositoryLive), - ); - return makeProviderServiceLive({ - canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"), - }).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer)); - })(); - - const runtimeServicesLayer = Layer.mergeAll( - orchestrationLayer, - OrchestrationProjectionSnapshotQueryLive, - checkpointStoreLayer, - providerLayer, - ); - const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( - Layer.provideMerge(runtimeServicesLayer), - ); - const providerCommandReactorLayer = ProviderCommandReactorLive.pipe( - Layer.provideMerge(runtimeServicesLayer), - ); - const checkpointReactorLayer = CheckpointReactorLive.pipe( - Layer.provideMerge(runtimeServicesLayer), - ); - const orchestrationReactorLayer = OrchestrationReactorLive.pipe( - Layer.provideMerge(runtimeIngestionLayer), - Layer.provideMerge(providerCommandReactorLayer), - Layer.provideMerge(checkpointReactorLayer), - ); - const layer = orchestrationReactorLayer.pipe( - Layer.provide(persistenceLayer), - Layer.provideMerge(NodeServices.layer), - ); - const runtime = ManagedRuntime.make(layer); - - try { - const [ - nextOrchestrationEngine, - nextProjectionSnapshotQuery, - nextCheckpointStore, - nextProviderService, - ] = await Promise.all([ - runtime.runPromise(Effect.service(OrchestrationEngineService)), - runtime.runPromise(Effect.service(ProjectionSnapshotQuery)), - runtime.runPromise(Effect.service(CheckpointStore)), - runtime.runPromise(Effect.service(ProviderService)), - ]); - orchestrationEngine = nextOrchestrationEngine; - projectionSnapshotQuery = nextProjectionSnapshotQuery; - checkpointStore = nextCheckpointStore; - providerService = nextProviderService; - await runtime.runPromise(attachSubscriptionss); - } catch (error) { - await runtime.dispose().catch(() => undefined); - throw error; + } } - - return runtime; - } - - async function disposeEffectRuntime() { - if (subscriptionsScope && effectRuntime) { - await effectRuntime.runPromise(Scope.close(subscriptionsScope, Exit.void)); - subscriptionsScope = null; + }); + + const handleMessage = Effect.fnUntraced(function* (ws: WebSocket, raw: unknown) { + const encodeResponse = Schema.encodeEffect(Schema.fromJsonString(WsResponse)); + + const messageText = websocketRawToString(raw); + if (messageText === null) { + const errorResponse = yield* encodeResponse( + new WebSocketResponse({ + id: "unknown", + error: { message: "Invalid request format: Failed to read message" }, + }), + ); + ws.send(errorResponse); + return; } - orchestrationDomainEventLogger?.close(); - orchestrationDomainEventLogger = undefined; - orchestrationCommandLogger?.close(); - orchestrationCommandLogger = undefined; - - const runtime = effectRuntime; - const liveProviderService = providerService; - effectRuntime = null; - orchestrationEngine = null; - projectionSnapshotQuery = null; - checkpointStore = null; - providerService = null; - - if (!runtime) { + + const request = Schema.decodeExit(Schema.fromJsonString(WebSocketRequest))(messageText); + if (request._tag === "Failure") { + const errorResponse = yield* encodeResponse( + new WebSocketResponse({ + id: "unknown", + error: { message: `Invalid request format: ${Cause.pretty(request.cause)}` }, + }), + ); + ws.send(errorResponse); return; } - if (liveProviderService) { - await runtime.runPromise(liveProviderService.stopAll()).catch(() => undefined); + const result = yield* Effect.exit(routeRequest(request.value)); + if (result._tag === "Failure") { + const errorResponse = yield* encodeResponse( + new WebSocketResponse({ + id: request.value.id, + error: { message: Cause.pretty(result.cause) }, + }), + ); + ws.send(errorResponse); + return; } - await runtime.dispose(); - } - - async function start() { - effectRuntime = await createEffectRuntime(); - await bootstrapProjectForCwd(); - - return new Promise((resolve, reject) => { - const onError = (error: Error) => { - httpServer.off("error", onError); - void disposeEffectRuntime().finally(() => reject(error)); - }; - httpServer.once("error", onError); - const onListening = () => { - httpServer.off("error", onError); - resolve(); - }; - if (host) { - httpServer.listen(port, host, onListening); + const response = yield* encodeResponse( + new WebSocketResponse({ + id: request.value.id, + result: result.value, + }), + ); + ws.send(response); + }); + + 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; } - httpServer.listen(port, onListening); - }); - } - - async function stop(): Promise { - await disposeEffectRuntime(); - terminalManager.off("event", onTerminalEvent); - terminalManager.dispose(); - for (const client of clients) { - client.close(); + if (providedToken !== authToken) { + rejectUpgrade(socket, 401, "Unauthorized WebSocket connection"); + return; + } } - clients.clear(); - - const closeWebSocketServer = new Promise((resolve, reject) => { - wss.close((error) => { - if (error && !isServerNotRunningError(error)) { - reject(error); - return; - } - resolve(); - }); + + wss.handleUpgrade(request, socket, head, (ws) => { + wss.emit("connection", ws, request); }); + }); - const closeHttpServer = new Promise((resolve, reject) => { - httpServer.close((error) => { - if (error && !isServerNotRunningError(error)) { - reject(error); - return; - } - resolve(); - }); + wss.on("connection", (ws) => { + clients.add(ws); + + 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)); + + ws.on("message", (raw) => { + void Effect.runPromise(handleMessage(ws, raw)); }); - await Promise.all([closeWebSocketServer, closeHttpServer]); - } + ws.on("close", () => { + clients.delete(ws); + }); - return { start, stop, httpServer }; -} + ws.on("error", () => { + clients.delete(ws); + }); + }); + + yield* Effect.addFinalizer(() => + Effect.catch(liveProviderService.stopAll(), (cause) => + Effect.logWarning("failed to stop provider service", { cause }), + ), + ); + + terminalManager.on("event", onTerminalEvent); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + terminalManager.off("event", onTerminalEvent); + terminalManager.dispose(); + }), + ); + + yield* NodeHttpServer.make(() => httpServer, listenOptions); + + yield* Effect.addFinalizer(() => + Effect.all([ + Effect.sync(() => closeAllClients()), + closeWebSocketServer.pipe(Effect.orElseSucceed(() => undefined)), + ]), + ); + + return httpServer; +}); + +export const ServerLive = Layer.succeed(Server, { + createServer, + stopSignal: Effect.never, +} satisfies ServerShape); -- 2.54.0