authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-24 17:18:49-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logda3d8e2ecb572e107b4ceec41d8966fe2f8f3db1
tree47396eb2ba13b7690ff42eeb085f00b11dde3fe4
parent5cbb6ea728903f4a9aa25441f663fe65e456f2c9

nice refactor

4 files changed, 522 insertions(+), 596 deletions(-)

apps/server/src/main.test.ts+23-14
......@@ -1,3 +1,4 @@
1import * as Http from "node:http";
12import { NodeServices } from "@effect/platform-node";
23import { assert, it, vi } from "@effect/vitest";
34import * as ConfigProvider from "effect/ConfigProvider";
......@@ -10,13 +11,17 @@ import { CliConfig, makeCliCommand, type CliConfigShape } from "./main";
1011import { Open, type OpenShape } from "./open";
1112import { Server, type ServerShape, type ServerOptions } from "./wsServer";
1213
13const start = vi.fn(async () => undefined);
14const stop = vi.fn(async () => undefined);
15const createServer = vi.fn((_: ServerOptions) => ({
16 start,
17 stop,
18 httpServer: undefined as never,
19}));
14const start = vi.fn(() => undefined);
15const stop = vi.fn(() => undefined);
16const createServer = vi.fn((_: ServerOptions) =>
17 Effect.acquireRelease(
18 Effect.sync(() => {
19 start();
20 return {} as unknown as Http.Server;
21 }),
22 () => Effect.sync(() => stop()),
23 ),
24);
2025const findAvailablePort = vi.fn((preferred: number) => Effect.succeed(preferred));
2126
2227const testLayer = Layer.mergeAll(
......@@ -47,13 +52,17 @@ const runCli = (
4752
4853beforeEach(() => {
4954 vi.clearAllMocks();
50 start.mockImplementation(async () => undefined);
51 stop.mockImplementation(async () => undefined);
52 createServer.mockImplementation((_: ServerOptions) => ({
53 start,
54 stop,
55 httpServer: undefined as never,
56 }));
55 start.mockImplementation(() => undefined);
56 stop.mockImplementation(() => undefined);
57 createServer.mockImplementation((_: ServerOptions) =>
58 Effect.acquireRelease(
59 Effect.sync(() => {
60 start();
61 return {} as unknown as Http.Server;
62 }),
63 () => Effect.sync(() => stop()),
64 ),
65 );
5766 findAvailablePort.mockImplementation((preferred: number) => Effect.succeed(preferred));
5867});
5968
apps/server/src/main.ts+26-24
......@@ -6,7 +6,12 @@ import { Command, Flag } from "effect/unstable/cli";
66
77import { fixPath } from "./fixPath";
88import { Open } from "./open";
9import { Server } from "./wsServer";
9import {
10 makeServerPersistenceLayer,
11 makeServerProviderLayer,
12 makeServerRuntimeServicesLayer,
13 Server,
14} from "./wsServer";
1015
1116const DEFAULT_PORT = 3773;
1217
......@@ -168,29 +173,26 @@ const makeServerProgram = Effect.fn(function* (input: CliInput) {
168173 });
169174 }
170175
171 yield* Effect.acquireRelease(
172 Effect.tryPromise({
173 try: async () => {
174 const server = createServer({
175 port: config.port,
176 host: config.mode === "desktop" ? "127.0.0.1" : undefined,
177 cwd: config.cwd,
178 autoBootstrapProjectFromCwd: config.mode === "web",
179 stateDir: config.stateDir,
180 staticDir: config.staticDir,
181 devUrl: config.devUrl?.href,
182 authToken: config.authToken,
183 });
184 await server.start();
185 return server;
186 },
187 catch: (cause) => new StartupError({ message: "Failed to start server", cause }),
188 }),
189 (server) =>
190 Effect.tryPromise({
191 try: () => server.stop(),
192 catch: (cause) => new StartupError({ message: "Failed to stop server", cause }),
193 }).pipe(Effect.catch((error) => Effect.logError("failed to stop server cleanly", { error }))),
176 const persistenceLayer = makeServerPersistenceLayer(config.stateDir);
177 const providerLayer = makeServerProviderLayer(config.stateDir);
178 const infrastructureLayer = providerLayer.pipe(Layer.provideMerge(persistenceLayer));
179 const serverRuntimeLayer = Layer.merge(
180 makeServerRuntimeServicesLayer().pipe(Layer.provide(infrastructureLayer)),
181 infrastructureLayer,
182 );
183
184 yield* createServer({
185 port: config.port,
186 host: config.mode === "desktop" ? "127.0.0.1" : undefined,
187 cwd: config.cwd,
188 autoBootstrapProjectFromCwd: config.mode === "web",
189 stateDir: config.stateDir,
190 staticDir: config.staticDir,
191 devUrl: config.devUrl?.href,
192 authToken: config.authToken,
193 }).pipe(
194 Effect.provide(serverRuntimeLayer),
195 Effect.mapError((cause) => new StartupError({ message: "Failed to start server", cause })),
194196 );
195197
196198 const url = `http://localhost:${config.port}`;
apps/server/src/wsServer.test.ts+112-106
......@@ -1,11 +1,12 @@
1import * as Http from "node:http";
12import fs from "node:fs";
23import os from "node:os";
34import path from "node:path";
45import { EventEmitter } from "node:events";
56
6import { Effect, Layer, PubSub, Stream } from "effect";
7import { Effect, Exit, Layer, PubSub, Scope, Stream } from "effect";
78import { describe, expect, it, afterEach, vi } from "vitest";
8import { createServer } from "./wsServer";
9import { createServer, makeServerProviderLayer, makeServerRuntimeServicesLayer } from "./wsServer";
910import WebSocket from "ws";
1011
1112import {
......@@ -49,7 +50,8 @@ interface PendingMessages {
4950const pendingBySocket = new WeakMap<WebSocket, PendingMessages>();
5051
5152const asEventId = (value: string): EventId => EventId.makeUnsafe(value);
52const asProviderSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
53const asProviderSessionId = (value: string): ProviderSessionId =>
54 ProviderSessionId.makeUnsafe(value);
5355const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value);
5456const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
5557const asProviderItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);
......@@ -286,7 +288,8 @@ function mergeWithDefaultsForTest(custom: KeybindingsConfig): ResolvedKeybinding
286288}
287289
288290describe("WebSocket Server", () => {
289 let server: ReturnType<typeof createServer> | null = null;
291 let server: Http.Server | null = null;
292 let serverScope: Scope.Closeable | null = null;
290293 const connections: WebSocket[] = [];
291294 const tempDirs: string[] = [];
292295
......@@ -296,7 +299,7 @@ describe("WebSocket Server", () => {
296299 return dir;
297300 }
298301
299 function createTestServer(
302 async function createTestServer(
300303 options: {
301304 persistenceLayer?: Layer.Layer<SqlClient.SqlClient, any>;
302305 cwd?: string;
......@@ -312,23 +315,53 @@ describe("WebSocket Server", () => {
312315 };
313316 terminalManager?: TerminalManager;
314317 } = {},
315 ): ReturnType<typeof createServer> {
318 ): Promise<Http.Server> {
319 if (serverScope) {
320 throw new Error("Test server is already running");
321 }
322
316323 const stateDir = options.stateDir ?? makeTempDir("t3code-ws-state-");
317 return createServer({
318 port: 0,
319 cwd: options.cwd ?? "/test/project",
320 stateDir,
321 persistenceLayer: options.persistenceLayer ?? SqlitePersistenceMemory,
322 ...(options.autoBootstrapProjectFromCwd !== undefined
323 ? { autoBootstrapProjectFromCwd: options.autoBootstrapProjectFromCwd }
324 : {}),
325 ...(options.devUrl ? { devUrl: options.devUrl } : {}),
326 ...(options.authToken ? { authToken: options.authToken } : {}),
327 ...(options.providerLayer ? { providerLayer: options.providerLayer } : {}),
328 ...(options.open ? { open: options.open } : {}),
329 ...(options.gitManager ? { gitManager: options.gitManager as never } : {}),
330 ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}),
331 });
324 const scope = await Effect.runPromise(Scope.make("sequential"));
325 const persistenceLayer = options.persistenceLayer ?? SqlitePersistenceMemory;
326 const providerLayer = options.providerLayer ?? makeServerProviderLayer(stateDir);
327 const infrastructureLayer = providerLayer.pipe(Layer.provideMerge(persistenceLayer));
328 const runtimeLayer = Layer.merge(
329 makeServerRuntimeServicesLayer().pipe(Layer.provide(infrastructureLayer)),
330 infrastructureLayer,
331 );
332 const runtimeServices = await Effect.runPromise(
333 Layer.build(runtimeLayer).pipe(Scope.provide(scope)),
334 );
335
336 try {
337 const runtime = await Effect.runPromise(
338 createServer({
339 port: 0,
340 cwd: options.cwd ?? "/test/project",
341 stateDir,
342 ...(options.autoBootstrapProjectFromCwd !== undefined
343 ? { autoBootstrapProjectFromCwd: options.autoBootstrapProjectFromCwd }
344 : {}),
345 ...(options.devUrl ? { devUrl: options.devUrl } : {}),
346 ...(options.authToken ? { authToken: options.authToken } : {}),
347 ...(options.open ? { open: options.open } : {}),
348 ...(options.gitManager ? { gitManager: options.gitManager as never } : {}),
349 ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}),
350 }).pipe(Effect.provide(runtimeServices), Scope.provide(scope)),
351 );
352 serverScope = scope;
353 return runtime;
354 } catch (error) {
355 await Effect.runPromise(Scope.close(scope, Exit.void));
356 throw error;
357 }
358 }
359
360 async function closeTestServer() {
361 if (!serverScope) return;
362 const scope = serverScope;
363 serverScope = null;
364 await Effect.runPromise(Scope.close(scope, Exit.void));
332365 }
333366
334367 afterEach(async () => {
......@@ -336,9 +369,7 @@ describe("WebSocket Server", () => {
336369 ws.close();
337370 }
338371 connections.length = 0;
339 if (server) {
340 await server.stop();
341 }
372 await closeTestServer();
342373 server = null;
343374 for (const dir of tempDirs.splice(0, tempDirs.length)) {
344375 fs.rmSync(dir, { recursive: true, force: true });
......@@ -347,10 +378,9 @@ describe("WebSocket Server", () => {
347378 });
348379
349380 it("sends welcome message on connect", async () => {
350 server = createTestServer({ cwd: "/test/project" });
381 server = await createTestServer({ cwd: "/test/project" });
351382 // Get the actual port after listen
352 await server.start();
353 const addr = server.httpServer.address();
383 const addr = server.address();
354384 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
355385 expect(port).toBeGreaterThan(0);
356386
......@@ -367,12 +397,11 @@ describe("WebSocket Server", () => {
367397 });
368398
369399 it("bootstraps the cwd project on startup when enabled", async () => {
370 server = createTestServer({
400 server = await createTestServer({
371401 cwd: "/test/bootstrap-workspace",
372402 autoBootstrapProjectFromCwd: true,
373403 });
374 await server.start();
375 const addr = server.httpServer.address();
404 const addr = server.address();
376405 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
377406 expect(port).toBeGreaterThan(0);
378407
......@@ -402,12 +431,11 @@ describe("WebSocket Server", () => {
402431 // Keep test output clean while verifying websocket logs.
403432 });
404433
405 server = createTestServer({
434 server = await createTestServer({
406435 cwd: "/test/project",
407436 devUrl: "http://localhost:5173",
408437 });
409 await server.start();
410 const addr = server.httpServer.address();
438 const addr = server.address();
411439 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
412440 expect(port).toBeGreaterThan(0);
413441
......@@ -430,9 +458,8 @@ describe("WebSocket Server", () => {
430458 it("responds to server.getConfig", async () => {
431459 const fakeHome = makeTempDir("t3code-home-");
432460 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
433 server = createTestServer({ cwd: "/my/workspace" });
434 await server.start();
435 const addr = server.httpServer.address();
461 server = await createTestServer({ cwd: "/my/workspace" });
462 const addr = server.address();
436463 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
437464
438465 const ws = await connectWs(port);
......@@ -459,9 +486,8 @@ describe("WebSocket Server", () => {
459486 },
460487 };
461488
462 server = createTestServer({ cwd: "/my/workspace", open: openService });
463 await server.start();
464 const addr = server.httpServer.address();
489 server = await createTestServer({ cwd: "/my/workspace", open: openService });
490 const addr = server.address();
465491 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
466492
467493 const ws = await connectWs(port);
......@@ -491,9 +517,8 @@ describe("WebSocket Server", () => {
491517 );
492518 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
493519
494 server = createTestServer({ cwd: "/my/workspace" });
495 await server.start();
496 const addr = server.httpServer.address();
520 server = await createTestServer({ cwd: "/my/workspace" });
521 const addr = server.address();
497522 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
498523
499524 const ws = await connectWs(port);
......@@ -528,9 +553,8 @@ describe("WebSocket Server", () => {
528553 );
529554 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
530555
531 server = createTestServer({ cwd: "/my/workspace" });
532 await server.start();
533 const addr = server.httpServer.address();
556 server = await createTestServer({ cwd: "/my/workspace" });
557 const addr = server.address();
534558 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
535559
536560 const ws = await connectWs(port);
......@@ -563,9 +587,8 @@ describe("WebSocket Server", () => {
563587 );
564588 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
565589
566 server = createTestServer({ cwd: "/my/workspace" });
567 await server.start();
568 const addr = server.httpServer.address();
590 server = await createTestServer({ cwd: "/my/workspace" });
591 const addr = server.address();
569592 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
570593
571594 const ws = await connectWs(port);
......@@ -598,9 +621,8 @@ describe("WebSocket Server", () => {
598621 );
599622 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
600623
601 server = createTestServer({ cwd: "/my/workspace" });
602 await server.start();
603 const addr = server.httpServer.address();
624 server = await createTestServer({ cwd: "/my/workspace" });
625 const addr = server.address();
604626 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
605627
606628 const ws = await connectWs(port);
......@@ -637,9 +659,8 @@ describe("WebSocket Server", () => {
637659 );
638660 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
639661
640 server = createTestServer({ cwd: "/my/workspace" });
641 await server.start();
642 const addr = server.httpServer.address();
662 server = await createTestServer({ cwd: "/my/workspace" });
663 const addr = server.address();
643664 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
644665
645666 const ws = await connectWs(port);
......@@ -690,9 +711,8 @@ describe("WebSocket Server", () => {
690711 );
691712 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
692713
693 server = createTestServer({ cwd: "/my/workspace" });
694 await server.start();
695 const addr = server.httpServer.address();
714 server = await createTestServer({ cwd: "/my/workspace" });
715 const addr = server.address();
696716 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
697717
698718 const ws = await connectWs(port);
......@@ -721,9 +741,8 @@ describe("WebSocket Server", () => {
721741 fs.writeFileSync(path.join(configDir, "keybindings.json"), "{not-json", "utf8");
722742 vi.spyOn(os, "homedir").mockReturnValue(fakeHome);
723743
724 server = createTestServer({ cwd: "/my/workspace" });
725 await server.start();
726 const addr = server.httpServer.address();
744 server = await createTestServer({ cwd: "/my/workspace" });
745 const addr = server.address();
727746 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
728747
729748 const ws = await connectWs(port);
......@@ -745,9 +764,8 @@ describe("WebSocket Server", () => {
745764 });
746765
747766 it("returns error for unknown methods", async () => {
748 server = createTestServer({ cwd: "/test" });
749 await server.start();
750 const addr = server.httpServer.address();
767 server = await createTestServer({ cwd: "/test" });
768 const addr = server.address();
751769 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
752770
753771 const ws = await connectWs(port);
......@@ -762,9 +780,8 @@ describe("WebSocket Server", () => {
762780 });
763781
764782 it("returns error when requesting turn diff for unknown thread", async () => {
765 server = createTestServer({ cwd: "/test" });
766 await server.start();
767 const addr = server.httpServer.address();
783 server = await createTestServer({ cwd: "/test" });
784 const addr = server.address();
768785 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
769786
770787 const ws = await connectWs(port);
......@@ -781,9 +798,8 @@ describe("WebSocket Server", () => {
781798 });
782799
783800 it("returns error when requesting full thread diff for unknown thread", async () => {
784 server = createTestServer({ cwd: "/test" });
785 await server.start();
786 const addr = server.httpServer.address();
801 server = await createTestServer({ cwd: "/test" });
802 const addr = server.address();
787803 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
788804
789805 const ws = await connectWs(port);
......@@ -799,9 +815,8 @@ describe("WebSocket Server", () => {
799815 });
800816
801817 it("returns retryable error when requested turn exceeds current checkpoint turn count", async () => {
802 server = createTestServer({ cwd: "/test" });
803 await server.start();
804 const addr = server.httpServer.address();
818 server = await createTestServer({ cwd: "/test" });
819 const addr = server.address();
805820 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
806821
807822 const ws = await connectWs(port);
......@@ -873,12 +888,11 @@ describe("WebSocket Server", () => {
873888 };
874889 const providerLayer = Layer.succeed(ProviderService, providerService);
875890
876 server = createTestServer({
891 server = await createTestServer({
877892 cwd: "/test",
878893 providerLayer,
879894 });
880 await server.start();
881 const addr = server.httpServer.address();
895 const addr = server.address();
882896 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
883897
884898 const ws = await connectWs(port);
......@@ -941,7 +955,9 @@ describe("WebSocket Server", () => {
941955
942956 const domainPush = await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => {
943957 const event = push.data as { type?: string; payload?: { messageId?: string; text?: string } };
944 return event.type === "thread.message-sent" && event.payload?.messageId === "assistant:item-1";
958 return (
959 event.type === "thread.message-sent" && event.payload?.messageId === "assistant:item-1"
960 );
945961 });
946962
947963 const domainEvent = domainPush.data as {
......@@ -956,12 +972,11 @@ describe("WebSocket Server", () => {
956972 it("routes terminal RPC methods and broadcasts terminal events", async () => {
957973 const cwd = makeTempDir("t3code-ws-terminal-cwd-");
958974 const terminalManager = new MockTerminalManager();
959 server = createTestServer({
975 server = await createTestServer({
960976 cwd: "/test",
961977 terminalManager: terminalManager as unknown as TerminalManager,
962978 });
963 await server.start();
964 const addr = server.httpServer.address();
979 const addr = server.address();
965980 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
966981
967982 const ws = await connectWs(port);
......@@ -1027,24 +1042,22 @@ describe("WebSocket Server", () => {
10271042
10281043 it("detaches terminal event listener on stop for injected manager", async () => {
10291044 const terminalManager = new MockTerminalManager();
1030 server = createTestServer({
1045 server = await createTestServer({
10311046 cwd: "/test",
10321047 terminalManager: terminalManager as unknown as TerminalManager,
10331048 });
1034 await server.start();
10351049
10361050 expect(terminalManager.listenerCount("event")).toBe(1);
10371051
1038 await server.stop();
1052 await closeTestServer();
10391053 server = null;
10401054
10411055 expect(terminalManager.listenerCount("event")).toBe(0);
10421056 });
10431057
10441058 it("returns validation errors for invalid terminal open params", async () => {
1045 server = createTestServer({ cwd: "/test" });
1046 await server.start();
1047 const addr = server.httpServer.address();
1059 server = await createTestServer({ cwd: "/test" });
1060 const addr = server.address();
10481061 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
10491062
10501063 const ws = await connectWs(port);
......@@ -1061,9 +1074,8 @@ describe("WebSocket Server", () => {
10611074 });
10621075
10631076 it("handles invalid JSON gracefully", async () => {
1064 server = createTestServer({ cwd: "/test" });
1065 await server.start();
1066 const addr = server.httpServer.address();
1077 server = await createTestServer({ cwd: "/test" });
1078 const addr = server.address();
10671079 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
10681080
10691081 const ws = await connectWs(port);
......@@ -1096,9 +1108,8 @@ describe("WebSocket Server", () => {
10961108 });
10971109
10981110 it("returns errors for removed projects CRUD methods", async () => {
1099 server = createTestServer({ cwd: "/test" });
1100 await server.start();
1101 const addr = server.httpServer.address();
1111 server = await createTestServer({ cwd: "/test" });
1112 const addr = server.address();
11021113 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
11031114
11041115 const ws = await connectWs(port);
......@@ -1134,9 +1145,8 @@ describe("WebSocket Server", () => {
11341145 fs.mkdirSync(path.join(workspace, ".git"), { recursive: true });
11351146 fs.writeFileSync(path.join(workspace, ".git", "HEAD"), "ref: refs/heads/main\n", "utf8");
11361147
1137 server = createTestServer({ cwd: "/test" });
1138 await server.start();
1139 const addr = server.httpServer.address();
1148 server = await createTestServer({ cwd: "/test" });
1149 const addr = server.address();
11401150 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
11411151
11421152 const ws = await connectWs(port);
......@@ -1161,9 +1171,8 @@ describe("WebSocket Server", () => {
11611171 it("supports git methods over websocket", async () => {
11621172 const repoCwd = makeTempDir("t3code-ws-git-project-");
11631173
1164 server = createTestServer({ cwd: "/test" });
1165 await server.start();
1166 const addr = server.httpServer.address();
1174 server = await createTestServer({ cwd: "/test" });
1175 const addr = server.address();
11671176 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
11681177
11691178 const ws = await connectWs(port);
......@@ -1209,9 +1218,8 @@ describe("WebSocket Server", () => {
12091218 runStackedAction: vi.fn(),
12101219 };
12111220
1212 server = createTestServer({ cwd: "/test", gitManager });
1213 await server.start();
1214 const addr = server.httpServer.address();
1221 server = await createTestServer({ cwd: "/test", gitManager });
1222 const addr = server.address();
12151223 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
12161224
12171225 const ws = await connectWs(port);
......@@ -1232,9 +1240,8 @@ describe("WebSocket Server", () => {
12321240 runStackedAction: vi.fn().mockRejectedValue(new Error("Cannot push from detached HEAD.")),
12331241 };
12341242
1235 server = createTestServer({ cwd: "/test", gitManager });
1236 await server.start();
1237 const addr = server.httpServer.address();
1243 server = await createTestServer({ cwd: "/test", gitManager });
1244 const addr = server.address();
12381245 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
12391246
12401247 const ws = await connectWs(port);
......@@ -1254,9 +1261,8 @@ describe("WebSocket Server", () => {
12541261 });
12551262
12561263 it("rejects websocket connections without a valid auth token", async () => {
1257 server = createTestServer({ cwd: "/test", authToken: "secret-token" });
1258 await server.start();
1259 const addr = server.httpServer.address();
1264 server = await createTestServer({ cwd: "/test", authToken: "secret-token" });
1265 const addr = server.address();
12601266 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
12611267
12621268 await expect(connectWs(port)).rejects.toThrow("WebSocket connection failed");
apps/server/src/wsServer.ts+361-452
......@@ -12,11 +12,12 @@ import {
1212 WS_CHANNELS,
1313 WS_METHODS,
1414 WebSocketRequest,
15 WebSocketResponse,
1516 WsPush,
1617 WsResponse,
1718} from "@t3tools/contracts";
18import { NodeServices } from "@effect/platform-node";
19import { Effect, Exit, Layer, ManagedRuntime, Schema, Scope, ServiceMap, Stream } from "effect";
19import { NodeHttpServer, NodeServices } from "@effect/platform-node";
20import { Cause, Effect, Exit, Layer, Schema, Scope, ServiceMap, Stream } from "effect";
2021import { WebSocketServer, type WebSocket } from "ws";
2122
2223import { createLogger } from "./logger";
......@@ -40,34 +41,21 @@ import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderComma
4041import { OrchestrationProjectionPipelineLive } from "./orchestration/Layers/ProjectionPipeline";
4142import { OrchestrationProjectionSnapshotQueryLive } from "./orchestration/Layers/ProjectionSnapshotQuery";
4243import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRuntimeIngestion";
43import {
44 OrchestrationEngineService,
45 type OrchestrationEngineShape,
46} from "./orchestration/Services/OrchestrationEngine";
47import {
48 ProjectionSnapshotQuery,
49 type ProjectionSnapshotQueryShape,
50} from "./orchestration/Services/ProjectionSnapshotQuery";
44import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine";
45import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery";
5146import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor";
5247import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";
53import assert from "node:assert";
5448import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore";
5549import { OrchestrationCommandReceiptRepositoryLive } from "./persistence/Layers/OrchestrationCommandReceipts";
5650import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter";
5751import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry";
5852import { makeProviderServiceLive } from "./provider/Layers/ProviderService";
5953import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory";
60import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";
54import { ProviderService } from "./provider/Services/ProviderService";
6155import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore";
6256import { ProviderSessionRuntimeRepositoryLive } from "./persistence/Layers/ProviderSessionRuntime";
63import {
64 CheckpointStore,
65 type CheckpointStoreShape,
66} from "./checkpointing/Services/CheckpointStore";
67import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger";
57import { CheckpointStore } from "./checkpointing/Services/CheckpointStore";
6858import * as SqlClient from "effect/unstable/sql/SqlClient";
69import * as SqlError from "effect/unstable/sql/SqlError";
70import * as Migrator from "effect/unstable/sql/Migrator";
7159import { clamp } from "effect/Number";
7260import { OpenDefaultService, type OpenShape } from "./open";
7361
......@@ -92,37 +80,24 @@ export interface ServerOptions {
9280 cwd: string;
9381 autoBootstrapProjectFromCwd?: boolean | undefined;
9482 stateDir?: string | undefined;
95 persistenceLayer?:
96 | Layer.Layer<SqlClient.SqlClient, SqlError.SqlError | Migrator.MigrationError>
97 | undefined;
9883 staticDir?: string | undefined;
9984 devUrl?: string | undefined;
10085 logWebSocketEvents?: boolean | undefined;
10186 gitManager?: GitManager | undefined;
10287 terminalManager?: TerminalManager | undefined;
103 providerLayer?: Layer.Layer<ProviderService, unknown> | undefined;
10488 authToken?: string | undefined;
10589 open?: OpenShape | undefined;
10690}
10791
108export interface ServerRuntime {
109 readonly start: () => Promise<void>;
110 readonly stop: () => Promise<void>;
111 readonly httpServer: http.Server;
112}
113
11492export interface ServerShape {
115 readonly createServer: (options: ServerOptions) => ServerRuntime;
93 readonly createServer: (
94 options: ServerOptions,
95 ) => Effect.Effect<http.Server, unknown, Scope.Scope | ServerRuntimeServices>;
11696 readonly stopSignal: Effect.Effect<unknown, never>;
11797}
11898
11999export class Server extends ServiceMap.Service<Server, ServerShape>()("server/Server") {}
120100
121export const ServerLive = Layer.succeed(Server, {
122 createServer,
123 stopSignal: Effect.never,
124} satisfies ServerShape);
125
126101const isServerNotRunningError = (error: unknown): boolean => {
127102 if (!(error instanceof Error)) return false;
128103 const maybeCode = (error as NodeJS.ErrnoException).code;
......@@ -195,47 +170,104 @@ function checkpointRefForThreadTurn(threadId: string, turnCount: number): Checkp
195170 return CheckpointRef.makeUnsafe(`${CHECKPOINT_REFS_PREFIX}/${encodedThreadId}/turn/${turnCount}`);
196171}
197172
198type EffectRuntime = ManagedRuntime.ManagedRuntime<
173export type ServerCoreRuntimeServices =
199174 | OrchestrationEngineService
200175 | ProjectionSnapshotQuery
201 | ProviderService
202176 | CheckpointStore
203 | OrchestrationReactor,
204 unknown
205>;
177 | OrchestrationReactor;
178
179export type ServerRuntimeServices = ServerCoreRuntimeServices | ProviderService;
206180
207export function createServer(options: ServerOptions): ServerRuntime {
181export function makeServerPersistenceLayer(stateDir: string) {
182 const dbPath = path.join(stateDir, "orchestration.sqlite");
183 return makeSqlitePersistenceLive(dbPath);
184}
185
186export function makeServerProviderLayer(
187 stateDir: string,
188): Layer.Layer<ProviderService, unknown, SqlClient.SqlClient> {
189 const providerLogsDir = path.join(stateDir, "logs", "providers");
190 const codexAdapterLayer = makeCodexAdapterLive({
191 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),
192 });
193 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(codexAdapterLayer));
194 const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(
195 Layer.provide(ProviderSessionRuntimeRepositoryLive),
196 );
197 return makeProviderServiceLive({
198 canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"),
199 }).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer));
200}
201
202export function makeServerRuntimeServicesLayer(): Layer.Layer<
203 ServerCoreRuntimeServices,
204 unknown,
205 SqlClient.SqlClient | ProviderService
206> {
207 const orchestrationLayer = OrchestrationEngineLive.pipe(
208 Layer.provide(OrchestrationProjectionPipelineLive),
209 Layer.provide(OrchestrationEventStoreLive),
210 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
211 );
212 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));
213
214 const runtimeServicesLayer = Layer.mergeAll(
215 orchestrationLayer,
216 OrchestrationProjectionSnapshotQueryLive,
217 checkpointStoreLayer,
218 );
219 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
220 Layer.provideMerge(runtimeServicesLayer),
221 );
222 const providerCommandReactorLayer = ProviderCommandReactorLive.pipe(
223 Layer.provideMerge(runtimeServicesLayer),
224 );
225 const checkpointReactorLayer = CheckpointReactorLive.pipe(
226 Layer.provideMerge(runtimeServicesLayer),
227 );
228 const orchestrationReactorLayer = OrchestrationReactorLive.pipe(
229 Layer.provideMerge(runtimeIngestionLayer),
230 Layer.provideMerge(providerCommandReactorLayer),
231 Layer.provideMerge(checkpointReactorLayer),
232 );
233
234 return orchestrationReactorLayer;
235}
236
237export class ServerLifecycleError extends Schema.TaggedErrorClass<ServerLifecycleError>()(
238 "ServerLifecycleError",
239 {
240 operation: Schema.String,
241 cause: Schema.optional(Schema.Defect),
242 },
243) {}
244
245class RouteRequestError extends Schema.TaggedErrorClass<RouteRequestError>()("RouteRequestError", {
246 message: Schema.String,
247}) {}
248
249export const createServer = Effect.fn(function* (
250 options: ServerOptions,
251): Effect.fn.Return<http.Server, unknown, Scope.Scope | ServerRuntimeServices> {
208252 const {
209253 port,
210254 host,
211255 cwd,
212256 autoBootstrapProjectFromCwd = false,
213 stateDir,
214 persistenceLayer: customPersistenceLayer,
215257 staticDir,
216258 devUrl,
217259 logWebSocketEvents: explicitLogWsEvents,
218260 gitManager = new GitManager(),
219261 terminalManager = new TerminalManager(),
220 providerLayer: customProviderLayer,
221262 authToken,
222263 open: openService = OpenDefaultService,
223264 } = options;
224265
225 const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata");
226 let effectRuntime: EffectRuntime | null = null;
227 let orchestrationEngine: OrchestrationEngineShape | null = null;
228 let projectionSnapshotQuery: ProjectionSnapshotQueryShape | null = null;
229 let providerService: ProviderServiceShape | null = null;
230 let checkpointStore: CheckpointStoreShape | null = null;
231266 const clients = new Set<WebSocket>();
232267 const logger = createLogger("ws");
233268 const logWebSocketEvents =
234269 explicitLogWsEvents ?? parseBooleanEnv(process.env.T3CODE_LOG_WS_EVENTS) ?? Boolean(devUrl);
235270 let keybindingsConfig = loadResolvedKeybindingsConfig(logger);
236 let orchestrationDomainEventLogger: EventNdjsonLogger | undefined;
237 let orchestrationCommandLogger: EventNdjsonLogger | undefined;
238 let subscriptionsScope: Scope.Closeable | null = null;
239271
240272 function logOutgoingPush(push: WsPush, recipients: number) {
241273 if (!logWebSocketEvents) return;
......@@ -246,86 +278,6 @@ export function createServer(options: ServerOptions): ServerRuntime {
246278 });
247279 }
248280
249 const getOrchestrationEngine = () => {
250 assert(orchestrationEngine, "Orchestration engine is not started");
251 return orchestrationEngine;
252 };
253
254 const getEffectRuntime = () => {
255 assert(effectRuntime, "Effect runtime is not started");
256 return effectRuntime;
257 };
258
259 const getProjectionSnapshotQuery = () => {
260 assert(projectionSnapshotQuery, "Projection snapshot query is not started");
261 return projectionSnapshotQuery;
262 };
263
264 const getCheckpointStore = () => {
265 assert(checkpointStore, "Checkpoint store is not started");
266 return checkpointStore;
267 };
268
269 async function bootstrapProjectForCwd() {
270 if (!autoBootstrapProjectFromCwd) {
271 return;
272 }
273
274 const runtime = getEffectRuntime();
275 const snapshotQuery = getProjectionSnapshotQuery();
276 const engine = getOrchestrationEngine();
277 const snapshot = await runtime.runPromise(snapshotQuery.getSnapshot());
278 const existing = snapshot.projects.find((project) => project.workspaceRoot === cwd);
279 if (existing) {
280 return;
281 }
282
283 const createdAt = new Date().toISOString();
284 const projectName = path.basename(cwd) || "project";
285 await runtime.runPromise(
286 engine.dispatchUnknownCommand({
287 type: "project.create",
288 commandId: crypto.randomUUID(),
289 projectId: crypto.randomUUID(),
290 title: projectName,
291 workspaceRoot: cwd,
292 defaultModel: "gpt-5-codex",
293 createdAt,
294 }),
295 );
296 }
297
298 const attachSubscriptionss = Effect.gen(function* () {
299 const orchestrationEngine = yield* OrchestrationEngineService;
300 const orchestrationReactor = yield* OrchestrationReactor;
301 subscriptionsScope = yield* Scope.make("sequential");
302
303 yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>
304 Effect.sync(() => {
305 orchestrationDomainEventLogger?.write({
306 observedAt: new Date().toISOString(),
307 event,
308 });
309 const push: WsPush = {
310 type: "push",
311 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
312 data: event,
313 };
314 const message = JSON.stringify(push);
315 let recipients = 0;
316 for (const client of clients) {
317 if (client.readyState === client.OPEN) {
318 client.send(message);
319 recipients += 1;
320 }
321 }
322 logOutgoingPush(push, recipients);
323 }),
324 ).pipe(Effect.forkIn(subscriptionsScope));
325
326 yield* orchestrationReactor.start.pipe(Scope.provide(subscriptionsScope));
327 });
328
329281 const onTerminalEvent = (event: TerminalEvent) => {
330282 const push: WsPush = {
331283 type: "push",
......@@ -342,7 +294,6 @@ export function createServer(options: ServerOptions): ServerRuntime {
342294 }
343295 logOutgoingPush(push, recipients);
344296 };
345 terminalManager.on("event", onTerminalEvent);
346297
347298 // HTTP server — serves static files or redirects to Vite dev server
348299 const httpServer = http.createServer((req, res) => {
......@@ -403,104 +354,85 @@ export function createServer(options: ServerOptions): ServerRuntime {
403354 // WebSocket server — upgrades from the HTTP server
404355 const wss = new WebSocketServer({ noServer: true });
405356
406 httpServer.on("upgrade", (request, socket, head) => {
407 if (authToken) {
408 let providedToken: string | null = null;
409 try {
410 const url = new URL(request.url ?? "/", `http://localhost:${port}`);
411 providedToken = url.searchParams.get("token");
412 } catch {
413 rejectUpgrade(socket, 400, "Invalid WebSocket URL");
414 return;
415 }
416
417 if (providedToken !== authToken) {
418 rejectUpgrade(socket, 401, "Unauthorized WebSocket connection");
419 return;
357 const closeWebSocketServer = Effect.callback<void, ServerLifecycleError>((resume) => {
358 wss.close((error) => {
359 if (error && !isServerNotRunningError(error)) {
360 resume(
361 Effect.fail(
362 new ServerLifecycleError({ operation: "closeWebSocketServer", cause: error }),
363 ),
364 );
365 } else {
366 resume(Effect.void);
420367 }
421 }
422
423 wss.handleUpgrade(request, socket, head, (ws) => {
424 wss.emit("connection", ws, request);
425368 });
426369 });
427370
428 wss.on("connection", (ws) => {
429 clients.add(ws);
430
431 // Send welcome message with project info
432 const segments = cwd.split(/[/\\]/).filter(Boolean);
433 const projectName = segments[segments.length - 1] ?? "project";
434
435 const welcome: WsPush = {
436 type: "push",
437 channel: WS_CHANNELS.serverWelcome,
438 data: { cwd, projectName },
439 };
440 logOutgoingPush(welcome, 1);
441 ws.send(JSON.stringify(welcome));
371 const closeAllClients = () => {
372 for (const client of clients) {
373 client.close();
374 }
375 clients.clear();
376 };
442377
443 ws.on("message", (raw) => {
444 void handleMessage(ws, raw);
445 });
378 const listenOptions = host ? { host, port } : { port };
446379
447 ws.on("close", () => {
448 clients.delete(ws);
449 });
380 const orchestrationEngine = yield* OrchestrationEngineService;
381 const projectionReadModelQuery = yield* ProjectionSnapshotQuery;
382 const liveCheckpointStore = yield* CheckpointStore;
383 const liveProviderService = yield* ProviderService;
384 const orchestrationReactor = yield* OrchestrationReactor;
450385
451 ws.on("error", () => {
452 clients.delete(ws);
453 });
454 });
386 const subscriptionsScope = yield* Scope.make("sequential");
387 yield* Effect.addFinalizer(() =>
388 Scope.close(subscriptionsScope, Exit.void).pipe(Effect.orElseSucceed(() => undefined)),
389 );
455390
456 async function handleMessage(ws: WebSocket, raw: unknown) {
457 const messageText = websocketRawToString(raw);
458 if (messageText === null) {
459 const errorResponse: WsResponse = {
460 id: "unknown",
461 error: { message: "Invalid request format" },
391 yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>
392 Effect.sync(() => {
393 const push: WsPush = {
394 type: "push",
395 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
396 data: event,
462397 };
463 ws.send(JSON.stringify(errorResponse));
464 return;
465 }
398 const message = JSON.stringify(push);
399 let recipients = 0;
400 for (const client of clients) {
401 if (client.readyState === client.OPEN) {
402 client.send(message);
403 recipients += 1;
404 }
405 }
406 logOutgoingPush(push, recipients);
407 }),
408 ).pipe(Effect.forkIn(subscriptionsScope));
466409
467 let request: WebSocketRequest;
468 try {
469 request = Schema.decodeUnknownSync(Schema.fromJsonString(WebSocketRequest))(messageText);
470 } catch {
471 const errorResponse: WsResponse = {
472 id: "unknown",
473 error: { message: "Invalid request format" },
474 };
475 ws.send(JSON.stringify(errorResponse));
476 return;
477 }
410 yield* orchestrationReactor.start.pipe(Scope.provide(subscriptionsScope));
478411
479 try {
480 const result = await routeRequest(request);
481 const response: WsResponse = { id: request.id, result };
482 ws.send(JSON.stringify(response));
483 } catch (err) {
484 const message = err instanceof Error ? err.message : "Unknown server error";
485 const response: WsResponse = {
486 id: request.id,
487 error: { message },
488 };
489 ws.send(JSON.stringify(response));
412 if (autoBootstrapProjectFromCwd) {
413 const snapshot = yield* projectionReadModelQuery.getSnapshot();
414 const existing = snapshot.projects.find((project) => project.workspaceRoot === cwd);
415 if (!existing) {
416 const createdAt = new Date().toISOString();
417 const projectName = path.basename(cwd) || "project";
418 yield* orchestrationEngine.dispatchUnknownCommand({
419 type: "project.create",
420 commandId: crypto.randomUUID(),
421 projectId: crypto.randomUUID(),
422 title: projectName,
423 workspaceRoot: cwd,
424 defaultModel: "gpt-5-codex",
425 createdAt,
426 });
490427 }
491428 }
492429
493 async function routeRequest(request: WebSocketRequest): Promise<unknown> {
494 const orchestrationEngine = getOrchestrationEngine();
495 const projectionReadModelQuery = getProjectionSnapshotQuery();
496 const effectRuntime = getEffectRuntime();
497 const liveCheckpointStore = getCheckpointStore();
498
499 const computeTurnDiff = async (input: {
430 const routeRequest = Effect.fn(function* (request: WebSocketRequest) {
431 const computeTurnDiff = Effect.fn(function* (input: {
500432 readonly threadId: string;
501433 readonly fromTurnCount: number;
502434 readonly toTurnCount: number;
503 }) => {
435 }) {
504436 if (input.fromTurnCount > input.toTurnCount) {
505437 throw new Error(
506438 `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 {
516448 };
517449 }
518450
519 const snapshot = await effectRuntime.runPromise(projectionReadModelQuery.getSnapshot());
451 const snapshot = yield* projectionReadModelQuery.getSnapshot();
520452 const thread = snapshot.threads.find((entry) => entry.id === input.threadId);
521453 if (!thread) {
522454 throw new Error(`Thread '${input.threadId}' not found.`);
......@@ -562,20 +494,19 @@ export function createServer(options: ServerOptions): ServerRuntime {
562494 );
563495 }
564496
565 const [fromExists, toExists] = await Promise.all([
566 effectRuntime.runPromise(
497 const [fromExists, toExists] = yield* Effect.all(
498 [
567499 liveCheckpointStore.hasCheckpointRef({
568500 cwd: workspaceCwd,
569501 checkpointRef: fromCheckpointRef,
570502 }),
571 ),
572 effectRuntime.runPromise(
573503 liveCheckpointStore.hasCheckpointRef({
574504 cwd: workspaceCwd,
575505 checkpointRef: toCheckpointRef,
576506 }),
577 ),
578 ]);
507 ],
508 { concurrency: "unbounded" },
509 );
579510 if (!fromExists) {
580511 throw new Error(
581512 `Filesystem checkpoint is unavailable for turn ${input.fromTurnCount} in thread ${input.threadId}.`,
......@@ -587,304 +518,282 @@ export function createServer(options: ServerOptions): ServerRuntime {
587518 );
588519 }
589520
590 const diff = await effectRuntime.runPromise(
591 liveCheckpointStore.diffCheckpoints({
592 cwd: workspaceCwd,
593 fromCheckpointRef,
594 toCheckpointRef,
595 fallbackFromToHead: false,
596 }),
597 );
521 const diff = yield* liveCheckpointStore.diffCheckpoints({
522 cwd: workspaceCwd,
523 fromCheckpointRef,
524 toCheckpointRef,
525 fallbackFromToHead: false,
526 });
598527 return {
599528 threadId: input.threadId,
600529 fromTurnCount: input.fromTurnCount,
601530 toTurnCount: input.toTurnCount,
602531 diff,
603532 };
604 };
533 });
605534
606535 switch (request.body._tag) {
607 case ORCHESTRATION_WS_METHODS.getSnapshot: {
608 return effectRuntime.runPromise(projectionReadModelQuery.getSnapshot());
609 }
536 case ORCHESTRATION_WS_METHODS.getSnapshot:
537 return yield* projectionReadModelQuery.getSnapshot();
610538
611539 case ORCHESTRATION_WS_METHODS.dispatchCommand: {
612 orchestrationCommandLogger?.write({
613 observedAt: new Date().toISOString(),
614 command: request.body.command,
615 });
616 return effectRuntime.runPromise(orchestrationEngine.dispatch(request.body.command));
540 const { command } = request.body;
541 return yield* orchestrationEngine.dispatch(command);
617542 }
618543
619544 case ORCHESTRATION_WS_METHODS.getTurnDiff: {
620545 const body = stripRequestTag(request.body);
621 return computeTurnDiff(body);
546 return yield* computeTurnDiff(body);
622547 }
623548
624549 case ORCHESTRATION_WS_METHODS.getFullThreadDiff: {
625550 const body = stripRequestTag(request.body);
626 return computeTurnDiff({
551 return yield* computeTurnDiff({
627552 threadId: body.threadId,
628553 fromTurnCount: 0,
629554 toTurnCount: body.toTurnCount,
630555 });
631556 }
632557
633 case ORCHESTRATION_WS_METHODS.replayEvents:
634 return effectRuntime.runPromise(
635 Stream.runCollect(
636 orchestrationEngine.readEvents(
637 clamp(request.body.fromSequenceExclusive, {
638 maximum: Number.MAX_SAFE_INTEGER,
639 minimum: 0,
640 }),
641 ),
642 ).pipe(Effect.map((events) => Array.from(events))),
643 );
558 case ORCHESTRATION_WS_METHODS.replayEvents: {
559 const { fromSequenceExclusive } = request.body;
560 return yield* Stream.runCollect(
561 orchestrationEngine.readEvents(
562 clamp(fromSequenceExclusive, {
563 maximum: Number.MAX_SAFE_INTEGER,
564 minimum: 0,
565 }),
566 ),
567 ).pipe(Effect.map((events) => Array.from(events)));
568 }
644569
645 case WS_METHODS.projectsSearchEntries:
646 return searchWorkspaceEntries(stripRequestTag(request.body));
570 case WS_METHODS.projectsSearchEntries: {
571 const body = stripRequestTag(request.body);
572 return yield* Effect.promise(() => searchWorkspaceEntries(body));
573 }
647574
648575 case WS_METHODS.shellOpenInEditor: {
649 await Effect.runPromise(openService.openInEditor(stripRequestTag(request.body)));
650 return undefined;
576 const body = stripRequestTag(request.body);
577 return yield* openService.openInEditor(body);
651578 }
652579
653 case WS_METHODS.gitStatus:
654 return gitManager.status(stripRequestTag(request.body));
580 case WS_METHODS.gitStatus: {
581 const body = stripRequestTag(request.body);
582 return yield* Effect.promise(() => gitManager.status(body));
583 }
655584
656 case WS_METHODS.gitPull:
657 return pullGitBranch(stripRequestTag(request.body));
585 case WS_METHODS.gitPull: {
586 const body = stripRequestTag(request.body);
587 return yield* Effect.promise(() => pullGitBranch(body));
588 }
658589
659 case WS_METHODS.gitRunStackedAction:
660 return gitManager.runStackedAction(stripRequestTag(request.body));
661 case WS_METHODS.gitListBranches:
662 return listGitBranches(stripRequestTag(request.body));
590 case WS_METHODS.gitRunStackedAction: {
591 const body = stripRequestTag(request.body);
592 return yield* Effect.promise(() => gitManager.runStackedAction(body));
593 }
663594
664 case WS_METHODS.gitCreateWorktree:
665 return createGitWorktree(stripRequestTag(request.body));
595 case WS_METHODS.gitListBranches: {
596 const body = stripRequestTag(request.body);
597 return yield* Effect.promise(() => listGitBranches(body));
598 }
666599
667 case WS_METHODS.gitRemoveWorktree:
668 return removeGitWorktree(stripRequestTag(request.body));
600 case WS_METHODS.gitCreateWorktree: {
601 const body = stripRequestTag(request.body);
602 return yield* Effect.promise(() => createGitWorktree(body));
603 }
669604
670 case WS_METHODS.gitCreateBranch:
671 return createGitBranch(stripRequestTag(request.body));
605 case WS_METHODS.gitRemoveWorktree: {
606 const body = stripRequestTag(request.body);
607 return yield* Effect.promise(() => removeGitWorktree(body));
608 }
672609
673 case WS_METHODS.gitCheckout:
674 return checkoutGitBranch(stripRequestTag(request.body));
610 case WS_METHODS.gitCreateBranch: {
611 const body = stripRequestTag(request.body);
612 return yield* Effect.promise(() => createGitBranch(body));
613 }
675614
676 case WS_METHODS.gitInit:
677 return initGitRepo(stripRequestTag(request.body));
615 case WS_METHODS.gitCheckout: {
616 const body = stripRequestTag(request.body);
617 return yield* Effect.promise(() => checkoutGitBranch(body));
618 }
678619
679 case WS_METHODS.terminalOpen:
680 return terminalManager.open(stripRequestTag(request.body));
620 case WS_METHODS.gitInit: {
621 const body = stripRequestTag(request.body);
622 return yield* Effect.promise(() => initGitRepo(body));
623 }
681624
682 case WS_METHODS.terminalWrite:
683 await terminalManager.write(stripRequestTag(request.body));
684 return undefined;
625 case WS_METHODS.terminalOpen: {
626 const body = stripRequestTag(request.body);
627 return yield* Effect.promise(() => terminalManager.open(body));
628 }
685629
686 case WS_METHODS.terminalResize:
687 await terminalManager.resize(stripRequestTag(request.body));
688 return undefined;
630 case WS_METHODS.terminalWrite: {
631 const body = stripRequestTag(request.body);
632 return yield* Effect.promise(() => terminalManager.write(body));
633 }
689634
690 case WS_METHODS.terminalClear:
691 await terminalManager.clear(stripRequestTag(request.body));
692 return undefined;
635 case WS_METHODS.terminalResize: {
636 const body = stripRequestTag(request.body);
637 return yield* Effect.promise(() => terminalManager.resize(body));
638 }
639
640 case WS_METHODS.terminalClear: {
641 const body = stripRequestTag(request.body);
642 return yield* Effect.promise(() => terminalManager.clear(body));
643 }
693644
694 case WS_METHODS.terminalRestart:
695 return terminalManager.restart(stripRequestTag(request.body));
645 case WS_METHODS.terminalRestart: {
646 const body = stripRequestTag(request.body);
647 return yield* Effect.promise(() => terminalManager.restart(body));
648 }
696649
697 case WS_METHODS.terminalClose:
698 await terminalManager.close(stripRequestTag(request.body));
699 return undefined;
650 case WS_METHODS.terminalClose: {
651 const body = stripRequestTag(request.body);
652 return yield* Effect.promise(() => terminalManager.close(body));
653 }
700654
701655 case WS_METHODS.serverGetConfig:
702 return {
703 cwd,
704 keybindings: keybindingsConfig,
705 };
656 return { cwd, keybindings: keybindingsConfig };
706657
707 case WS_METHODS.serverUpsertKeybinding:
708 keybindingsConfig = upsertKeybindingRule(logger, stripRequestTag(request.body));
709 return {
710 keybindings: keybindingsConfig,
711 };
658 case WS_METHODS.serverUpsertKeybinding: {
659 const body = stripRequestTag(request.body);
660 keybindingsConfig = upsertKeybindingRule(logger, body);
661 return { keybindings: keybindingsConfig };
662 }
712663
713664 default: {
714665 const _exhaustiveCheck: never = request.body;
715 throw new Error(`Unknown method: ${String(_exhaustiveCheck)}`);
716 }
717 }
718 }
719
720 async function createEffectRuntime() {
721 const dbPath = path.join(resolvedStateDir, "orchestration.sqlite");
722 const providerLogsDir = path.join(resolvedStateDir, "logs", "providers");
723 const orchestrationLogsDir = path.join(resolvedStateDir, "logs", "orchestration");
724 orchestrationDomainEventLogger = makeEventNdjsonLogger(
725 path.join(orchestrationLogsDir, "orchestration-domain.ndjson"),
726 );
727 orchestrationCommandLogger = makeEventNdjsonLogger(
728 path.join(orchestrationLogsDir, "orchestration-command.ndjson"),
729 );
730 const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath);
731 const orchestrationLayer = OrchestrationEngineLive.pipe(
732 Layer.provide(OrchestrationProjectionPipelineLive),
733 Layer.provide(OrchestrationEventStoreLive),
734 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
735 );
736 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));
737 const providerLayer =
738 customProviderLayer ??
739 (() => {
740 const codexAdapterLayer = makeCodexAdapterLive({
741 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),
666 return yield* new RouteRequestError({
667 message: `Unknown method: ${String(_exhaustiveCheck)}`,
742668 });
743 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(
744 Layer.provide(codexAdapterLayer),
745 );
746 const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(
747 Layer.provide(ProviderSessionRuntimeRepositoryLive),
748 );
749 return makeProviderServiceLive({
750 canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"),
751 }).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer));
752 })();
753
754 const runtimeServicesLayer = Layer.mergeAll(
755 orchestrationLayer,
756 OrchestrationProjectionSnapshotQueryLive,
757 checkpointStoreLayer,
758 providerLayer,
759 );
760 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
761 Layer.provideMerge(runtimeServicesLayer),
762 );
763 const providerCommandReactorLayer = ProviderCommandReactorLive.pipe(
764 Layer.provideMerge(runtimeServicesLayer),
765 );
766 const checkpointReactorLayer = CheckpointReactorLive.pipe(
767 Layer.provideMerge(runtimeServicesLayer),
768 );
769 const orchestrationReactorLayer = OrchestrationReactorLive.pipe(
770 Layer.provideMerge(runtimeIngestionLayer),
771 Layer.provideMerge(providerCommandReactorLayer),
772 Layer.provideMerge(checkpointReactorLayer),
773 );
774 const layer = orchestrationReactorLayer.pipe(
775 Layer.provide(persistenceLayer),
776 Layer.provideMerge(NodeServices.layer),
777 );
778 const runtime = ManagedRuntime.make(layer);
779
780 try {
781 const [
782 nextOrchestrationEngine,
783 nextProjectionSnapshotQuery,
784 nextCheckpointStore,
785 nextProviderService,
786 ] = await Promise.all([
787 runtime.runPromise(Effect.service(OrchestrationEngineService)),
788 runtime.runPromise(Effect.service(ProjectionSnapshotQuery)),
789 runtime.runPromise(Effect.service(CheckpointStore)),
790 runtime.runPromise(Effect.service(ProviderService)),
791 ]);
792 orchestrationEngine = nextOrchestrationEngine;
793 projectionSnapshotQuery = nextProjectionSnapshotQuery;
794 checkpointStore = nextCheckpointStore;
795 providerService = nextProviderService;
796 await runtime.runPromise(attachSubscriptionss);
797 } catch (error) {
798 await runtime.dispose().catch(() => undefined);
799 throw error;
669 }
800670 }
671 });
801672
802 return runtime;
803 }
673 const handleMessage = Effect.fnUntraced(function* (ws: WebSocket, raw: unknown) {
674 const encodeResponse = Schema.encodeEffect(Schema.fromJsonString(WsResponse));
804675
805 async function disposeEffectRuntime() {
806 if (subscriptionsScope && effectRuntime) {
807 await effectRuntime.runPromise(Scope.close(subscriptionsScope, Exit.void));
808 subscriptionsScope = null;
676 const messageText = websocketRawToString(raw);
677 if (messageText === null) {
678 const errorResponse = yield* encodeResponse(
679 new WebSocketResponse({
680 id: "unknown",
681 error: { message: "Invalid request format: Failed to read message" },
682 }),
683 );
684 ws.send(errorResponse);
685 return;
809686 }
810 orchestrationDomainEventLogger?.close();
811 orchestrationDomainEventLogger = undefined;
812 orchestrationCommandLogger?.close();
813 orchestrationCommandLogger = undefined;
814
815 const runtime = effectRuntime;
816 const liveProviderService = providerService;
817 effectRuntime = null;
818 orchestrationEngine = null;
819 projectionSnapshotQuery = null;
820 checkpointStore = null;
821 providerService = null;
822
823 if (!runtime) {
687
688 const request = Schema.decodeExit(Schema.fromJsonString(WebSocketRequest))(messageText);
689 if (request._tag === "Failure") {
690 const errorResponse = yield* encodeResponse(
691 new WebSocketResponse({
692 id: "unknown",
693 error: { message: `Invalid request format: ${Cause.pretty(request.cause)}` },
694 }),
695 );
696 ws.send(errorResponse);
824697 return;
825698 }
826699
827 if (liveProviderService) {
828 await runtime.runPromise(liveProviderService.stopAll()).catch(() => undefined);
700 const result = yield* Effect.exit(routeRequest(request.value));
701 if (result._tag === "Failure") {
702 const errorResponse = yield* encodeResponse(
703 new WebSocketResponse({
704 id: request.value.id,
705 error: { message: Cause.pretty(result.cause) },
706 }),
707 );
708 ws.send(errorResponse);
709 return;
829710 }
830711
831 await runtime.dispose();
832 }
712 const response = yield* encodeResponse(
713 new WebSocketResponse({
714 id: request.value.id,
715 result: result.value,
716 }),
717 );
718 ws.send(response);
719 });
833720
834 async function start() {
835 effectRuntime = await createEffectRuntime();
836 await bootstrapProjectForCwd();
721 httpServer.on("upgrade", (request, socket, head) => {
722 if (authToken) {
723 let providedToken: string | null = null;
724 try {
725 const url = new URL(request.url ?? "/", `http://localhost:${port}`);
726 providedToken = url.searchParams.get("token");
727 } catch {
728 rejectUpgrade(socket, 400, "Invalid WebSocket URL");
729 return;
730 }
837731
838 return new Promise<void>((resolve, reject) => {
839 const onError = (error: Error) => {
840 httpServer.off("error", onError);
841 void disposeEffectRuntime().finally(() => reject(error));
842 };
843 httpServer.once("error", onError);
844 const onListening = () => {
845 httpServer.off("error", onError);
846 resolve();
847 };
848 if (host) {
849 httpServer.listen(port, host, onListening);
732 if (providedToken !== authToken) {
733 rejectUpgrade(socket, 401, "Unauthorized WebSocket connection");
850734 return;
851735 }
852 httpServer.listen(port, onListening);
736 }
737
738 wss.handleUpgrade(request, socket, head, (ws) => {
739 wss.emit("connection", ws, request);
853740 });
854 }
741 });
855742
856 async function stop(): Promise<void> {
857 await disposeEffectRuntime();
858 terminalManager.off("event", onTerminalEvent);
859 terminalManager.dispose();
743 wss.on("connection", (ws) => {
744 clients.add(ws);
860745
861 for (const client of clients) {
862 client.close();
863 }
864 clients.clear();
746 const segments = cwd.split(/[/\\]/).filter(Boolean);
747 const projectName = segments[segments.length - 1] ?? "project";
865748
866 const closeWebSocketServer = new Promise<void>((resolve, reject) => {
867 wss.close((error) => {
868 if (error && !isServerNotRunningError(error)) {
869 reject(error);
870 return;
871 }
872 resolve();
873 });
749 const welcome: WsPush = {
750 type: "push",
751 channel: WS_CHANNELS.serverWelcome,
752 data: { cwd, projectName },
753 };
754 logOutgoingPush(welcome, 1);
755 ws.send(JSON.stringify(welcome));
756
757 ws.on("message", (raw) => {
758 void Effect.runPromise(handleMessage(ws, raw));
874759 });
875760
876 const closeHttpServer = new Promise<void>((resolve, reject) => {
877 httpServer.close((error) => {
878 if (error && !isServerNotRunningError(error)) {
879 reject(error);
880 return;
881 }
882 resolve();
883 });
761 ws.on("close", () => {
762 clients.delete(ws);
884763 });
885764
886 await Promise.all([closeWebSocketServer, closeHttpServer]);
887 }
765 ws.on("error", () => {
766 clients.delete(ws);
767 });
768 });
888769
889 return { start, stop, httpServer };
890}
770 yield* Effect.addFinalizer(() =>
771 Effect.catch(liveProviderService.stopAll(), (cause) =>
772 Effect.logWarning("failed to stop provider service", { cause }),
773 ),
774 );
775
776 terminalManager.on("event", onTerminalEvent);
777 yield* Effect.addFinalizer(() =>
778 Effect.sync(() => {
779 terminalManager.off("event", onTerminalEvent);
780 terminalManager.dispose();
781 }),
782 );
783
784 yield* NodeHttpServer.make(() => httpServer, listenOptions);
785
786 yield* Effect.addFinalizer(() =>
787 Effect.all([
788 Effect.sync(() => closeAllClients()),
789 closeWebSocketServer.pipe(Effect.orElseSucceed(() => undefined)),
790 ]),
791 );
792
793 return httpServer;
794});
795
796export const ServerLive = Layer.succeed(Server, {
797 createServer,
798 stopSignal: Effect.never,
799} satisfies ServerShape);