From e63d5b240f908ec3cf69b49ac7488b33dc5231c0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 24 Feb 2026 13:46:30 -0800 Subject: [PATCH] effect cli --- apps/server/src/codexAppServerManager.test.ts | 15 +- apps/server/src/index.ts | 194 ++---------- apps/server/src/keybindings.ts | 2 +- apps/server/src/main.test.ts | 131 ++++++++ apps/server/src/main.ts | 284 ++++++++++++++++++ apps/server/src/open.ts | 93 ++++++ apps/server/src/orchestration/Layer.test.ts | 131 ++++---- .../Layers/CheckpointReactor.test.ts | 4 +- .../Layers/ProjectionSnapshotQuery.test.ts | 49 ++- .../Layers/ProviderCommandReactor.test.ts | 10 +- .../Layers/ProviderRuntimeIngestion.test.ts | 2 +- .../orchestration/commandInvariants.test.ts | 67 +++-- .../decider.projectScripts.test.ts | 59 ++-- .../src/orchestration/projector.test.ts | 55 ++-- .../src/provider/Layers/CodexAdapter.test.ts | 80 ++--- .../provider/Layers/ProviderService.test.ts | 58 ++-- apps/server/src/wsServer.test.ts | 97 ++++-- apps/server/src/wsServer.ts | 76 ++--- apps/server/tsconfig.json | 3 +- packages/contracts/src/git.ts | 2 +- packages/contracts/src/keybindings.test.ts | 4 +- packages/contracts/src/keybindings.ts | 13 +- scripts/dev-runner.mjs | 9 + 23 files changed, 952 insertions(+), 486 deletions(-) create mode 100644 apps/server/src/main.test.ts create mode 100644 apps/server/src/main.ts create mode 100644 apps/server/src/open.ts diff --git a/apps/server/src/codexAppServerManager.test.ts b/apps/server/src/codexAppServerManager.test.ts index afb23d48a768f7c1a52b386659e1e919625682f1..f0af874337229dfe373ece3efd84e5702cd3a2f4 100644 --- a/apps/server/src/codexAppServerManager.test.ts +++ b/apps/server/src/codexAppServerManager.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from "vitest"; +import { ProviderSessionId } from "@t3tools/contracts"; import { CodexAppServerManager, @@ -7,6 +8,8 @@ import { normalizeCodexModelSlug, } from "./codexAppServerManager"; +const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); + function createSendTurnHarness() { const manager = new CodexAppServerManager(); const context = { @@ -183,7 +186,7 @@ describe("sendTurn", () => { createSendTurnHarness(); const result = await manager.sendTurn({ - sessionId: "sess_1", + sessionId: asSessionId("sess_1"), input: "Inspect this image", attachments: [ { @@ -229,7 +232,7 @@ describe("sendTurn", () => { const { manager, context, sendRequest } = createSendTurnHarness(); await manager.sendTurn({ - sessionId: "sess_1", + sessionId: asSessionId("sess_1"), attachments: [ { type: "image", @@ -257,7 +260,7 @@ describe("sendTurn", () => { await expect( manager.sendTurn({ - sessionId: "sess_1", + sessionId: asSessionId("sess_1"), }), ).rejects.toThrow("Turn input must include text or attachments."); }); @@ -278,7 +281,7 @@ describe("thread checkpoint control", () => { }, }); - const result = await manager.readThread("sess_1"); + const result = await manager.readThread(asSessionId("sess_1")); expect(requireSession).toHaveBeenCalledWith("sess_1"); expect(sendRequest).toHaveBeenCalledWith(context, "thread/read", { @@ -308,7 +311,7 @@ describe("thread checkpoint control", () => { ], }); - const result = await manager.readThread("sess_1"); + const result = await manager.readThread(asSessionId("sess_1")); expect(sendRequest).toHaveBeenCalledWith(context, "thread/read", { threadId: "thread_1", @@ -334,7 +337,7 @@ describe("thread checkpoint control", () => { }, }); - const result = await manager.rollbackThread("sess_1", 2); + const result = await manager.rollbackThread(asSessionId("sess_1"), 2); expect(sendRequest).toHaveBeenCalledWith(context, "thread/rollback", { threadId: "thread_1", diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index c2ce1b3449454edaeefcdb80b0995d6494a068cd..f23154e7235e2f8fa2743cbcf3391122a7babb7f 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -1,177 +1,17 @@ -import fs from "node:fs"; -import net from "node:net"; -import os from "node:os"; -import path from "node:path"; -import { fileURLToPath } from "node:url"; - -import { fixPath } from "./fixPath"; -import { createLogger } from "./logger"; -import { createServer } from "./wsServer"; - -fixPath(); - -const DEFAULT_PORT = 3773; -// Dummy comment for a no-op change request. -// Additional dummy comment requested in chat. -const cwd = process.cwd(); -const logger = createLogger("server"); - -type RuntimeMode = "web" | "desktop"; - -function parseBooleanEnv(value: string | undefined): boolean | undefined { - if (value === undefined) return undefined; - const normalized = value.trim().toLowerCase(); - if (["1", "true", "yes", "on"].includes(normalized)) return true; - if (["0", "false", "no", "off"].includes(normalized)) return false; - return undefined; -} - -function parsePort(value: string | undefined): number | undefined { - if (!value) return undefined; - const parsed = Number(value); - if (!Number.isInteger(parsed) || parsed < 1 || parsed > 65535) { - throw new Error(`Invalid T3CODE_PORT: ${value}`); - } - return parsed; -} - -function expandHomePath(input: string): string { - if (input === "~") return os.homedir(); - if (input.startsWith("~/")) { - return path.join(os.homedir(), input.slice(2)); - } - return input; -} - -function resolveStateDir(raw: string | undefined): string { - if (!raw || raw.trim().length === 0) { - return path.join(os.homedir(), ".t3", "userdata"); - } - return path.resolve(expandHomePath(raw.trim())); -} - -async function findAvailablePort(preferred: number): Promise { - return new Promise((resolve, reject) => { - const server = net.createServer(); - server.listen(preferred, () => { - server.close(() => resolve(preferred)); - }); - server.on("error", () => { - // Preferred port busy — let the OS pick one - const fallback = net.createServer(); - fallback.listen(0, () => { - const addr = fallback.address(); - const port = typeof addr === "object" && addr !== null ? addr.port : 0; - fallback.close(() => { - if (port > 0) resolve(port); - else reject(new Error("Could not find an available port.")); - }); - }); - fallback.on("error", reject); - }); - }); -} - -function resolveStaticDir(): string | undefined { - const __dirname = path.dirname(fileURLToPath(import.meta.url)); - - // Check for bundled client (npm publish / npx scenario): - // dist/client/ lives alongside dist/index.js - const bundledClient = path.resolve(__dirname, "client"); - try { - const stat = fs.statSync(path.join(bundledClient, "index.html")); - if (stat.isFile()) return bundledClient; - } catch { - // Not bundled — check monorepo layout - } - - // Monorepo layout: apps/server/dist/index.js → apps/web/dist/ - const monorepoClient = path.resolve(__dirname, "../../web/dist"); - try { - const stat = fs.statSync(path.join(monorepoClient, "index.html")); - if (stat.isFile()) return monorepoClient; - } catch { - // Not found — probably dev mode, Vite will serve - } - - return undefined; -} - -async function main() { - const mode: RuntimeMode = process.env.T3CODE_MODE === "desktop" ? "desktop" : "web"; - const requestedPort = parsePort(process.env.T3CODE_PORT); - const port = - requestedPort ?? (mode === "desktop" ? DEFAULT_PORT : await findAvailablePort(DEFAULT_PORT)); - const stateDir = resolveStateDir(process.env.T3CODE_STATE_DIR); - const devUrl = process.env.VITE_DEV_SERVER_URL; - const noBrowser = parseBooleanEnv(process.env.T3CODE_NO_BROWSER) ?? mode === "desktop"; - const authToken = process.env.T3CODE_AUTH_TOKEN; - const staticDir = devUrl ? undefined : resolveStaticDir(); - - if (!devUrl && !staticDir) { - logger.warn("web bundle missing and no VITE_DEV_SERVER_URL; web UI unavailable", { - hint: "Run `bun run --cwd apps/web build` or set VITE_DEV_SERVER_URL for dev mode.", - }); - } - - const server = createServer({ - port, - host: mode === "desktop" ? "127.0.0.1" : undefined, - cwd, - autoBootstrapProjectFromCwd: mode === "web", - stateDir, - staticDir, - devUrl, - authToken, - }); - await server.start(); - - const url = `http://localhost:${port}`; - logger.info("T3 Code running", { - url, - cwd, - mode, - stateDir, - authEnabled: Boolean(authToken), - }); - - // Open browser (dynamic import because `open` is ESM-only) - if (!noBrowser) { - try { - const open = await import("open"); - const target = devUrl ?? url; - await open.default(target); - } catch { - logger.info("browser auto-open unavailable", { - hint: `Open ${devUrl ?? url} in your browser.`, - }); - } - } - - // Graceful shutdown - let shuttingDown = false; - async function shutdown() { - if (shuttingDown) return; - shuttingDown = true; - logger.info("shutting down"); - try { - await server.stop(); - process.exit(0); - } catch (error) { - logger.error("failed to stop server cleanly", { error }); - process.exit(1); - } - } - - process.on("SIGINT", () => { - void shutdown(); - }); - process.on("SIGTERM", () => { - void shutdown(); - }); -} - -main().catch((err) => { - logger.error("failed to start T3 Code", { error: err }); - process.exit(1); -}); +import * as NodeRuntime from "@effect/platform-node/NodeRuntime"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { CliConfigLive, makeCliCommand } from "./main"; +import { OpenLive } from "./open"; +import { Command } from "effect/unstable/cli"; +import { version } from "../package.json" with { type: "json" }; +import { ServerLive } from "./wsServer"; + +const RuntimeLayer = Layer.mergeAll(CliConfigLive, ServerLive, OpenLive, NodeServices.layer); + +Command.run(makeCliCommand(), { version }).pipe( + Effect.provide(RuntimeLayer), + NodeRuntime.runMain, +); diff --git a/apps/server/src/keybindings.ts b/apps/server/src/keybindings.ts index 428662a7498cd719520d73a82d5515ae78c82eda..70f43dfeb277703a1bff3dc17e49e6c854f05939 100644 --- a/apps/server/src/keybindings.ts +++ b/apps/server/src/keybindings.ts @@ -326,7 +326,7 @@ function loadCustomKeybindingsConfig( const sanitized: Mutable = []; let invalidEntries = 0; for (const entry of parsed) { - const result = Schema.decodeUnknownExit(KeybindingRule)(entry); + const result = Schema.decodeExit(KeybindingRule)(entry); if (result._tag === "Success") { if (!compileResolvedKeybindingRule(result.value)) { invalidEntries += 1; diff --git a/apps/server/src/main.test.ts b/apps/server/src/main.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..e98612e8489951cbadc8ed5d0c6923e426280f8b --- /dev/null +++ b/apps/server/src/main.test.ts @@ -0,0 +1,131 @@ +import { NodeServices } from "@effect/platform-node"; +import { assert, it, vi } from "@effect/vitest"; +import * as ConfigProvider from "effect/ConfigProvider"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Command from "effect/unstable/cli/Command"; +import { beforeEach } from "vitest"; + +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 findAvailablePort = vi.fn((preferred: number) => Effect.succeed(preferred)); + +const testLayer = Layer.mergeAll( + Layer.succeed(CliConfig, { + cwd: "/tmp/t3-test-workspace", + fixPath: Effect.void, + findAvailablePort, + resolveStaticDir: () => undefined, + } satisfies CliConfigShape), + Layer.succeed(Server, { + createServer, + } satisfies ServerShape), + Layer.succeed(Open, { + openBrowser: (_target: string) => Effect.void, + openInEditor: () => Effect.void, + } satisfies OpenShape), + NodeServices.layer, +); + +const runCli = ( + args: ReadonlyArray, + env: Record = { T3CODE_NO_BROWSER: "true" }, +) => + Command.runWith(makeCliCommand(), { version: "0.0.0-test" })(args).pipe( + Effect.provide(ConfigProvider.layer(ConfigProvider.fromEnv({ env }))), + ); + +beforeEach(() => { + vi.clearAllMocks(); + start.mockImplementation(async () => undefined); + stop.mockImplementation(async () => undefined); + createServer.mockImplementation((_: ServerOptions) => ({ + start, + stop, + httpServer: undefined as never, + })); + findAvailablePort.mockImplementation((preferred: number) => Effect.succeed(preferred)); +}); + +it.layer(testLayer)("server cli", (it) => { + it.effect("parses --port and --token and wires scoped start/stop", () => + Effect.gen(function* () { + yield* runCli(["--port", "4010", "--token", "secret"]); + + assert.equal(createServer.mock.calls.length, 1); + const options = createServer.mock.calls[0]?.[0]; + assert.equal(options?.port, 4010); + assert.equal(options?.authToken, "secret"); + assert.equal(start.mock.calls.length, 1); + assert.equal(stop.mock.calls.length, 1); + }), + ); + + it.effect("uses env fallbacks when flags are not provided", () => + Effect.gen(function* () { + yield* runCli([], { + T3CODE_NO_BROWSER: "true", + T3CODE_PORT: "4999", + T3CODE_AUTH_TOKEN: "env-token", + }); + + assert.equal(createServer.mock.calls.length, 1); + const options = createServer.mock.calls[0]?.[0]; + assert.equal(options?.port, 4999); + assert.equal(options?.authToken, "env-token"); + assert.equal(findAvailablePort.mock.calls.length, 0); + }), + ); + + it.effect("uses dynamic port discovery in web mode when port is omitted", () => + Effect.gen(function* () { + findAvailablePort.mockImplementation((_preferred: number) => Effect.succeed(5444)); + yield* runCli([]); + + assert.deepStrictEqual(findAvailablePort.mock.calls, [[3773]]); + assert.equal(createServer.mock.calls.length, 1); + const options = createServer.mock.calls[0]?.[0]; + assert.equal(options?.port, 5444); + assert.equal(options?.host, undefined); + }), + ); + + it.effect("uses fixed localhost defaults in desktop mode", () => + Effect.gen(function* () { + yield* runCli([], { + T3CODE_MODE: "desktop", + T3CODE_NO_BROWSER: "true", + }); + + assert.equal(findAvailablePort.mock.calls.length, 0); + assert.equal(createServer.mock.calls.length, 1); + const options = createServer.mock.calls[0]?.[0]; + assert.equal(options?.port, 3773); + assert.equal(options?.host, "127.0.0.1"); + }), + ); + + it.effect("fails for out-of-range --port values", () => + Effect.gen(function* () { + const result = yield* Effect.result(runCli(["--port", "70000"])); + + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal( + result.failure.message, + "Invalid --port: 70000. Expected an integer between 1 and 65535.", + ); + } + assert.equal(createServer.mock.calls.length, 0); + }), + ); +}); diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts new file mode 100644 index 0000000000000000000000000000000000000000..cb74206a1be15b92e2f4bd8f8a84bbaf0e49c727 --- /dev/null +++ b/apps/server/src/main.ts @@ -0,0 +1,284 @@ +import fs from "node:fs"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +import * as Data from "effect/Data"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Config from "effect/Config"; +import { ServiceMap } from "effect"; +import * as Command from "effect/unstable/cli/Command"; +import * as Flag from "effect/unstable/cli/Flag"; + +import { fixPath } from "./fixPath"; +import { Open } from "./open"; +import { Server } from "./wsServer"; + +const DEFAULT_PORT = 3773; + +type RuntimeMode = "web" | "desktop"; + +export class StartupError extends Data.TaggedError("StartupError")<{ + readonly message: string; + readonly cause?: unknown; +}> {} + +interface CliInput { + readonly port: Option.Option; + readonly token: Option.Option; +} + +interface ResolvedCliConfig { + readonly port: number; + readonly mode: RuntimeMode; + readonly cwd: string; + readonly stateDir: string; + readonly devUrl: URL | undefined; + readonly noBrowser: boolean; + readonly authToken: string | undefined; + readonly staticDir: string | undefined; +} + +export interface CliConfigShape { + readonly cwd: string; + readonly fixPath: Effect.Effect; + readonly findAvailablePort: (preferred: number) => Effect.Effect; + readonly resolveStaticDir: () => string | undefined; +} + +export class CliConfig extends ServiceMap.Service()( + "server/CliConfig", +) {} + +const CliEnvConfig = Config.all({ + mode: Config.string("T3CODE_MODE").pipe( + Config.option, + Config.map( + Option.match({ + onNone: () => "web" as const, + onSome: (value) => (value === "desktop" ? "desktop" : "web"), + }), + ), + ), + port: Config.port("T3CODE_PORT").pipe(Config.option, Config.map(Option.getOrUndefined)), + stateDir: Config.string("T3CODE_STATE_DIR").pipe( + Config.option, + Config.map(Option.getOrUndefined), + ), + devUrl: Config.url("VITE_DEV_SERVER_URL").pipe(Config.option, Config.map(Option.getOrUndefined)), + noBrowser: Config.boolean("T3CODE_NO_BROWSER").pipe( + Config.option, + Config.map(Option.getOrUndefined), + ), + authToken: Config.string("T3CODE_AUTH_TOKEN").pipe( + Config.option, + Config.map(Option.getOrUndefined), + ), +}); + +function assertValidPort(value: number, source: string): Effect.Effect { + if (Number.isInteger(value) && value >= 1 && value <= 65535) { + return Effect.succeed(value); + } + return Effect.fail( + new StartupError({ + message: `Invalid ${source}: ${value}. Expected an integer between 1 and 65535.`, + }), + ); +} + +function expandHomePath(input: string): string { + if (input === "~") return os.homedir(); + if (input.startsWith("~/")) { + return path.join(os.homedir(), input.slice(2)); + } + return input; +} + +function resolveStateDir(raw: string | undefined): string { + if (!raw || raw.trim().length === 0) { + return path.join(os.homedir(), ".t3", "userdata"); + } + return path.resolve(expandHomePath(raw.trim())); +} + +async function findAvailablePort(preferred: number): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.listen(preferred, () => { + server.close(() => resolve(preferred)); + }); + server.on("error", () => { + const fallback = net.createServer(); + fallback.listen(0, () => { + const addr = fallback.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + fallback.close(() => { + if (port > 0) resolve(port); + else reject(new Error("Could not find an available port.")); + }); + }); + fallback.on("error", reject); + }); + }); +} + +function resolveStaticDir(): string | undefined { + const __dirname = path.dirname(fileURLToPath(import.meta.url)); + + const bundledClient = path.resolve(__dirname, "client"); + try { + const stat = fs.statSync(path.join(bundledClient, "index.html")); + if (stat.isFile()) return bundledClient; + } catch { + // Not bundled — check monorepo layout. + } + + const monorepoClient = path.resolve(__dirname, "../../web/dist"); + try { + const stat = fs.statSync(path.join(monorepoClient, "index.html")); + if (stat.isFile()) return monorepoClient; + } catch { + // Not found — probably dev mode. + } + + return undefined; +} + +function resolveCliConfig( + input: CliInput, +): Effect.Effect { + return Effect.gen(function* () { + const cliConfig = yield* CliConfig; + const env = yield* CliEnvConfig.asEffect().pipe( + Effect.mapError( + (cause) => + new StartupError({ + message: "Failed to read environment configuration", + cause, + }), + ), + ); + const mode: RuntimeMode = env.mode === "desktop" ? "desktop" : "web"; + + const cliPort = Option.getOrUndefined(input.port); + const requestedPort = + cliPort === undefined ? env.port : yield* assertValidPort(cliPort, "--port"); + + const port = + requestedPort ?? + (mode === "desktop" ? DEFAULT_PORT : yield* cliConfig.findAvailablePort(DEFAULT_PORT)); + const stateDir = resolveStateDir(env.stateDir); + const devUrl = env.devUrl; + const noBrowser = env.noBrowser ?? mode === "desktop"; + const authToken = Option.getOrUndefined(input.token) ?? env.authToken; + const staticDir = devUrl ? undefined : cliConfig.resolveStaticDir(); + + return { + mode, + port, + stateDir, + devUrl, + noBrowser, + authToken, + staticDir, + cwd: cliConfig.cwd, + }; + }); +} + +const makeServerProgram = Effect.fn(function* (input: CliInput) { + const cliConfig = yield* CliConfig; + const serverDeps = yield* Server; + const openDeps = yield* Open; + yield* cliConfig.fixPath; + const config = yield* resolveCliConfig(input); + + if (!config.devUrl && !config.staticDir) { + yield* Effect.logWarning("web bundle missing and no VITE_DEV_SERVER_URL; web UI unavailable", { + hint: "Run `bun run --cwd apps/web build` or set VITE_DEV_SERVER_URL for dev mode.", + }); + } + + yield* Effect.acquireRelease( + Effect.tryPromise({ + try: async () => { + const server = serverDeps.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 url = `http://localhost:${config.port}`; + yield* Effect.logInfo("T3 Code running", { + url, + cwd: config.cwd, + mode: config.mode, + stateDir: config.stateDir, + authEnabled: Boolean(config.authToken), + }); + + if (!config.noBrowser) { + const target = config.devUrl ?? url; + yield* openDeps.openBrowser(target.toString()).pipe( + Effect.catch(() => + Effect.logInfo("browser auto-open unavailable", { + hint: `Open ${target} in your browser.`, + }), + ), + ); + } +}, Effect.scoped); + +export function makeCliCommand() { + const portFlag = Flag.integer("port").pipe( + Flag.withDescription("Port for the HTTP/WebSocket server."), + Flag.optional, + ); + const tokenFlag = Flag.string("token").pipe( + Flag.withDescription("Auth token required for WebSocket connections."), + Flag.optional, + ); + + return Command.make("t3", { + port: portFlag, + token: tokenFlag, + }).pipe( + Command.withDescription("Run the T3 Code server."), + Command.withHandler((input) => makeServerProgram(input)), + ); +} + +export const CliConfigLive = Layer.succeed(CliConfig, { + cwd: process.cwd(), + fixPath: Effect.sync(fixPath), + findAvailablePort: (preferred) => + Effect.tryPromise({ + try: () => findAvailablePort(preferred), + catch: (cause) => + new StartupError({ + message: "Failed to discover an available port", + cause, + }), + }), + resolveStaticDir, +} satisfies CliConfigShape); diff --git a/apps/server/src/open.ts b/apps/server/src/open.ts new file mode 100644 index 0000000000000000000000000000000000000000..723d656eb6217af4122afeb08d9e0b23acca84ee --- /dev/null +++ b/apps/server/src/open.ts @@ -0,0 +1,93 @@ +import { spawn } from "node:child_process"; + +import { EDITORS, type EditorId } from "@t3tools/contracts"; +import { ServiceMap, Schema, Effect, Layer } from "effect"; + +export class OpenError extends Schema.TaggedErrorClass()("OpenError", { + message: Schema.String, + cause: Schema.optional(Schema.Defect), +}) {} + +export interface OpenInEditorInput { + readonly cwd: string; + readonly editor: EditorId; +} + +interface EditorLaunch { + readonly command: string; + readonly args: ReadonlyArray; +} + +export interface OpenShape { + readonly openBrowser: (target: string) => Effect.Effect; + readonly openInEditor: (input: OpenInEditorInput) => Effect.Effect; +} + +export class Open extends ServiceMap.Service()("server/Open") {} + +export function resolveEditorLaunch( + input: OpenInEditorInput, + platform: NodeJS.Platform = process.platform, +): EditorLaunch { + const editorDef = EDITORS.find((editor) => editor.id === input.editor); + if (!editorDef) { + throw new OpenError({ message: `Unknown editor: ${input.editor}` }); + } + + if (editorDef.command) { + return { command: editorDef.command, args: [input.cwd] }; + } + + if (editorDef.id !== "file-manager") { + throw new OpenError({ message: `Unsupported editor: ${input.editor}` }); + } + + switch (platform) { + case "darwin": + return { command: "open", args: [input.cwd] }; + case "win32": + return { command: "explorer", args: [input.cwd] }; + default: + return { command: "xdg-open", args: [input.cwd] }; + } +} + +const OpenDefault: OpenShape = { + openBrowser: (target) => + Effect.tryPromise({ + try: async () => { + const open = await import("open"); + await open.default(target); + }, + catch: (cause) => + new OpenError({ + message: "Browser auto-open failed", + cause, + }), + }), + openInEditor: (input) => + Effect.try({ + try: () => { + const launch = resolveEditorLaunch(input); + const child = spawn(launch.command, [...launch.args], { + detached: true, + stdio: "ignore", + }); + child.on("error", () => { + // Best-effort behavior for detached editor launches. + }); + child.unref(); + }, + catch: (cause) => + cause instanceof OpenError + ? cause + : new OpenError({ + message: "Failed to open editor", + cause, + }), + }), +}; + +export const OpenLive = Layer.succeed(Open, OpenDefault); + +export const OpenDefaultService = OpenDefault; diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts index ac9df75cce21b0411ef6e353c55b6c8a000096ba..242f2c8661eae6e8282f5a110fb8bd4402bacfc8 100644 --- a/apps/server/src/orchestration/Layer.test.ts +++ b/apps/server/src/orchestration/Layer.test.ts @@ -1,4 +1,12 @@ -import type { OrchestrationEvent } from "@t3tools/contracts"; +import { + CheckpointRef, + CommandId, + MessageId, + ProjectId, + ThreadId, + TurnId, + type OrchestrationEvent, +} from "@t3tools/contracts"; import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect"; import { describe, expect, it } from "vitest"; @@ -14,6 +22,13 @@ import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts"; import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); +const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); +const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); +const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); +const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); +const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); + async function createOrchestrationSystem() { const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionPipelineLive), @@ -43,8 +58,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: "cmd-project-1-create", - projectId: "project-1", + commandId: asCommandId("cmd-project-1-create"), + projectId: asProjectId("project-1"), title: "Project 1", workspaceRoot: "/tmp/project-1", defaultModel: "gpt-5-codex", @@ -54,9 +69,9 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: "cmd-thread-1-create", - threadId: "thread-1", - projectId: "project-1", + commandId: asCommandId("cmd-thread-1-create"), + threadId: asThreadId("thread-1"), + projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", branch: null, @@ -67,10 +82,10 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.turn.start", - commandId: "cmd-turn-start-1", - threadId: "thread-1", + commandId: asCommandId("cmd-turn-start-1"), + threadId: asThreadId("thread-1"), message: { - messageId: "msg-1", + messageId: asMessageId("msg-1"), role: "user", text: "hello", attachments: [], @@ -93,8 +108,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: "cmd-project-replay-create", - projectId: "project-replay", + commandId: asCommandId("cmd-project-replay-create"), + projectId: asProjectId("project-replay"), title: "Replay Project", workspaceRoot: "/tmp/project-replay", defaultModel: "gpt-5-codex", @@ -104,9 +119,9 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: "cmd-thread-replay-create", - threadId: "thread-replay", - projectId: "project-replay", + commandId: asCommandId("cmd-thread-replay-create"), + threadId: asThreadId("thread-replay"), + projectId: asProjectId("project-replay"), title: "replay", model: "gpt-5-codex", branch: null, @@ -117,8 +132,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.delete", - commandId: "cmd-thread-replay-delete", - threadId: "thread-replay", + commandId: asCommandId("cmd-thread-replay-delete"), + threadId: asThreadId("thread-replay"), }), ); @@ -143,8 +158,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: "cmd-project-stream-create", - projectId: "project-stream", + commandId: asCommandId("cmd-project-stream-create"), + projectId: asProjectId("project-stream"), title: "Stream Project", workspaceRoot: "/tmp/project-stream", defaultModel: "gpt-5-codex", @@ -164,9 +179,9 @@ describe("OrchestrationEngine", () => { yield* Effect.sleep("10 millis"); yield* engine.dispatch({ type: "thread.create", - commandId: "cmd-stream-thread-create", - threadId: "thread-stream", - projectId: "project-stream", + commandId: asCommandId("cmd-stream-thread-create"), + threadId: asThreadId("thread-stream"), + projectId: asProjectId("project-stream"), title: "domain-stream", model: "gpt-5-codex", branch: null, @@ -175,8 +190,8 @@ describe("OrchestrationEngine", () => { }); yield* engine.dispatch({ type: "thread.meta.update", - commandId: "cmd-stream-thread-update", - threadId: "thread-stream", + commandId: asCommandId("cmd-stream-thread-update"), + threadId: asThreadId("thread-stream"), title: "domain-stream-updated", }); eventTypes.push((yield* Queue.take(eventQueue)).type); @@ -196,8 +211,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: "cmd-project-turn-diff-create", - projectId: "project-turn-diff", + commandId: asCommandId("cmd-project-turn-diff-create"), + projectId: asProjectId("project-turn-diff"), title: "Turn Diff Project", workspaceRoot: "/tmp/project-turn-diff", defaultModel: "gpt-5-codex", @@ -207,9 +222,9 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: "cmd-thread-turn-diff-create", - threadId: "thread-turn-diff", - projectId: "project-turn-diff", + commandId: asCommandId("cmd-thread-turn-diff-create"), + threadId: asThreadId("thread-turn-diff"), + projectId: asProjectId("project-turn-diff"), title: "Turn diff thread", model: "gpt-5-codex", branch: null, @@ -220,11 +235,11 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.turn.diff.complete", - commandId: "cmd-turn-diff-complete", - threadId: "thread-turn-diff", - turnId: "turn-1", + commandId: asCommandId("cmd-turn-diff-complete"), + threadId: asThreadId("thread-turn-diff"), + turnId: asTurnId("turn-1"), completedAt: createdAt, - checkpointRef: "refs/t3/checkpoints/thread-turn-diff/turn/1", + checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), status: "ready", files: [], checkpointTurnCount: 1, @@ -237,9 +252,9 @@ describe("OrchestrationEngine", () => { ); expect(thread?.checkpoints).toEqual([ { - turnId: "turn-1", + turnId: asTurnId("turn-1"), checkpointTurnCount: 1, - checkpointRef: "refs/t3/checkpoints/thread-turn-diff/turn/1", + checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), status: "ready", files: [], assistantMessageId: null, @@ -250,13 +265,17 @@ describe("OrchestrationEngine", () => { }); it("keeps processing queued commands after a storage failure", async () => { - const events: OrchestrationEvent[] = []; + type StoredEvent = + ReturnType extends Effect.Effect + ? A + : never; + const events: StoredEvent[] = []; let nextSequence = 1; let shouldFailFirstAppend = true; const flakyStore: OrchestrationEventStoreShape = { append(event) { - if (shouldFailFirstAppend && event.commandId === "cmd-flaky-1") { + if (shouldFailFirstAppend && event.commandId === asCommandId("cmd-flaky-1")) { shouldFailFirstAppend = false; return Effect.fail( new PersistenceSqlError({ @@ -268,7 +287,7 @@ describe("OrchestrationEngine", () => { const savedEvent = { ...event, sequence: nextSequence, - } satisfies OrchestrationEvent; + } as StoredEvent; nextSequence += 1; events.push(savedEvent); return Effect.succeed(savedEvent); @@ -295,8 +314,8 @@ describe("OrchestrationEngine", () => { await runtime.runPromise( engine.dispatch({ type: "project.create", - commandId: "cmd-project-flaky-create", - projectId: "project-flaky", + commandId: asCommandId("cmd-project-flaky-create"), + projectId: asProjectId("project-flaky"), title: "Flaky Project", workspaceRoot: "/tmp/project-flaky", defaultModel: "gpt-5-codex", @@ -308,9 +327,9 @@ describe("OrchestrationEngine", () => { runtime.runPromise( engine.dispatch({ type: "thread.create", - commandId: "cmd-flaky-1", - threadId: "thread-flaky-fail", - projectId: "project-flaky", + commandId: asCommandId("cmd-flaky-1"), + threadId: asThreadId("thread-flaky-fail"), + projectId: asProjectId("project-flaky"), title: "flaky-fail", model: "gpt-5-codex", branch: null, @@ -323,9 +342,9 @@ describe("OrchestrationEngine", () => { const result = await runtime.runPromise( engine.dispatch({ type: "thread.create", - commandId: "cmd-flaky-2", - threadId: "thread-flaky-ok", - projectId: "project-flaky", + commandId: asCommandId("cmd-flaky-2"), + threadId: asThreadId("thread-flaky-ok"), + projectId: asProjectId("project-flaky"), title: "flaky-ok", model: "gpt-5-codex", branch: null, @@ -347,10 +366,10 @@ describe("OrchestrationEngine", () => { system.run( engine.dispatch({ type: "thread.turn.start", - commandId: "cmd-invariant-missing-thread", - threadId: "thread-missing", + commandId: asCommandId("cmd-invariant-missing-thread"), + threadId: asThreadId("thread-missing"), message: { - messageId: "msg-missing", + messageId: asMessageId("msg-missing"), role: "user", text: "hello", attachments: [], @@ -371,8 +390,8 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "project.create", - commandId: "cmd-project-duplicate-create", - projectId: "project-duplicate", + commandId: asCommandId("cmd-project-duplicate-create"), + projectId: asProjectId("project-duplicate"), title: "Duplicate Project", workspaceRoot: "/tmp/project-duplicate", defaultModel: "gpt-5-codex", @@ -383,9 +402,9 @@ describe("OrchestrationEngine", () => { await system.run( engine.dispatch({ type: "thread.create", - commandId: "cmd-thread-duplicate-1", - threadId: "thread-duplicate", - projectId: "project-duplicate", + commandId: asCommandId("cmd-thread-duplicate-1"), + threadId: asThreadId("thread-duplicate"), + projectId: asProjectId("project-duplicate"), title: "duplicate", model: "gpt-5-codex", branch: null, @@ -398,9 +417,9 @@ describe("OrchestrationEngine", () => { system.run( engine.dispatch({ type: "thread.create", - commandId: "cmd-thread-duplicate-2", - threadId: "thread-duplicate", - projectId: "project-duplicate", + commandId: asCommandId("cmd-thread-duplicate-2"), + threadId: asThreadId("thread-duplicate"), + projectId: asProjectId("project-duplicate"), title: "duplicate", model: "gpt-5-codex", branch: null, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 338a64e874c56b38048f6aeb30d42b336ef21066..da8fc58c39ea43699a88aee8424a808036d10e17 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -41,6 +41,8 @@ const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); +const asProviderThreadId = (value: string): ProviderThreadId => + ProviderThreadId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); function checkpointRefForTurn(threadId: ThreadId, turnCount: number): CheckpointRef { @@ -62,7 +64,7 @@ function createProviderServiceHarness(cwd: string, hasSession = true) { sessionId: asSessionId("sess-1"), provider: "codex", status: "ready", - threadId: asThreadId("thread-1"), + threadId: asProviderThreadId("provider-thread-1"), cwd, createdAt: now, updatedAt: now, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 3f57264c3187945a3ca89aae9bfba49e9b39e4f8..867d932e3507241353dde6bcc2f03b86a16766d8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -1,3 +1,13 @@ +import { + CheckpointRef, + EventId, + MessageId, + ProjectId, + ProviderSessionId, + ProviderThreadId, + ThreadId, + TurnId, +} from "@t3tools/contracts"; import { assert, it } from "@effect/vitest"; import { Effect, Layer } from "effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -7,6 +17,15 @@ import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); +const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); +const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); +const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); +const asEventId = (value: string): EventId => EventId.makeUnsafe(value); +const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); +const asProviderSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); +const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); + const projectionSnapshotLayer = it.layer( OrchestrationProjectionSnapshotQueryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)), ); @@ -185,7 +204,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { assert.equal(snapshot.updatedAt, "2026-02-24T00:00:09.000Z"); assert.deepEqual(snapshot.projects, [ { - id: "project-1", + id: asProjectId("project-1"), title: "Project 1", workspaceRoot: "/tmp/project-1", defaultModel: "gpt-5-codex", @@ -205,22 +224,22 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ]); assert.deepEqual(snapshot.threads, [ { - id: "thread-1", - projectId: "project-1", + id: asThreadId("thread-1"), + projectId: asProjectId("project-1"), title: "Thread 1", model: "gpt-5-codex", branch: null, worktreePath: null, - latestTurnId: "turn-1", + latestTurnId: asTurnId("turn-1"), createdAt: "2026-02-24T00:00:02.000Z", updatedAt: "2026-02-24T00:00:03.000Z", deletedAt: null, messages: [ { - id: "message-1", + id: asMessageId("message-1"), role: "assistant", text: "hello from projection", - turnId: "turn-1", + turnId: asTurnId("turn-1"), streaming: false, createdAt: "2026-02-24T00:00:04.000Z", updatedAt: "2026-02-24T00:00:05.000Z", @@ -228,33 +247,33 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ], activities: [ { - id: "activity-1", + id: asEventId("activity-1"), tone: "info", kind: "runtime.note", summary: "provider started", payload: { stage: "start" }, - turnId: "turn-1", + turnId: asTurnId("turn-1"), createdAt: "2026-02-24T00:00:06.000Z", }, ], checkpoints: [ { - turnId: "turn-1", + turnId: asTurnId("turn-1"), checkpointTurnCount: 1, - checkpointRef: "checkpoint-1", + checkpointRef: asCheckpointRef("checkpoint-1"), status: "ready", files: [{ path: "README.md", kind: "modified", additions: 2, deletions: 1 }], - assistantMessageId: "message-1", + assistantMessageId: asMessageId("message-1"), completedAt: "2026-02-24T00:00:08.000Z", }, ], session: { - threadId: "thread-1", + threadId: asThreadId("thread-1"), status: "running", providerName: "codex", - providerSessionId: "provider-session-1", - providerThreadId: "provider-thread-1", - activeTurnId: "turn-1", + providerSessionId: asProviderSessionId("provider-session-1"), + providerThreadId: asProviderThreadId("provider-thread-1"), + activeTurnId: asTurnId("turn-1"), lastError: null, updatedAt: "2026-02-24T00:00:07.000Z", }, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index b35cc22af7790b3f6ef62548f5838417a6c3f245..95265cfe5a575dc6b667a31d0940f78db285be94 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -2,11 +2,13 @@ import type { ProviderRuntimeEvent } from "@t3tools/contracts"; import { ApprovalRequestId, CommandId, + MessageId, ProjectId, ProviderSessionId, ProviderThreadId, ProviderTurnId, ThreadId, + TurnId, } from "@t3tools/contracts"; import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; @@ -30,6 +32,8 @@ const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); +const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); +const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); async function waitFor( predicate: () => boolean, @@ -166,7 +170,7 @@ describe("ProviderCommandReactor", () => { commandId: asCommandId("cmd-turn-start-1"), threadId: asThreadId("thread-1"), message: { - messageId: "user-message-1", + messageId: asMessageId("user-message-1"), role: "user", text: "hello reactor", attachments: [], @@ -198,7 +202,7 @@ describe("ProviderCommandReactor", () => { providerName: "codex", providerSessionId: asSessionId("sess-1"), providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), - activeTurnId: "turn-1", + activeTurnId: asTurnId("turn-1"), lastError: null, updatedAt: now, }, @@ -211,7 +215,7 @@ describe("ProviderCommandReactor", () => { type: "thread.turn.interrupt", commandId: asCommandId("cmd-turn-interrupt"), threadId: asThreadId("thread-1"), - turnId: "turn-1", + turnId: asTurnId("turn-1"), createdAt: now, }), ); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index dcf2c18e256e0e32a4fa46e69b30264199eaaa5c..9d2f3bf80eb0c816fe80c9819e5e6a79cc94b8da 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -313,7 +313,7 @@ describe("ProviderRuntimeIngestion", () => { sessionId: asSessionId("sess-1"), createdAt: now, turnId: asProviderTurnId("turn-9"), - toolKind: "read", + toolKind: "other", title: "Read file", detail: "/tmp/file.ts", }); diff --git a/apps/server/src/orchestration/commandInvariants.test.ts b/apps/server/src/orchestration/commandInvariants.test.ts index 87fbe951813ee25e6fa59cec0b46aa85e11a0232..d02f80e300745f20db13c21d455814d264688e9a 100644 --- a/apps/server/src/orchestration/commandInvariants.test.ts +++ b/apps/server/src/orchestration/commandInvariants.test.ts @@ -1,5 +1,12 @@ import { describe, expect, it } from "vitest"; -import type { OrchestrationCommand, OrchestrationReadModel } from "@t3tools/contracts"; +import { + MessageId, + CommandId, + ProjectId, + ThreadId, + type OrchestrationCommand, + type OrchestrationReadModel, +} from "@t3tools/contracts"; import { Effect } from "effect"; import { @@ -17,7 +24,7 @@ const readModel: OrchestrationReadModel = { updatedAt: now, projects: [ { - id: "project-a", + id: ProjectId.makeUnsafe("project-a"), title: "Project A", workspaceRoot: "/tmp/project-a", defaultModel: "gpt-5-codex", @@ -27,7 +34,7 @@ const readModel: OrchestrationReadModel = { deletedAt: null, }, { - id: "project-b", + id: ProjectId.makeUnsafe("project-b"), title: "Project B", workspaceRoot: "/tmp/project-b", defaultModel: "gpt-5-codex", @@ -39,8 +46,8 @@ const readModel: OrchestrationReadModel = { ], threads: [ { - id: "thread-1", - projectId: "project-a", + id: ThreadId.makeUnsafe("thread-1"), + projectId: ProjectId.makeUnsafe("project-a"), title: "Thread A", model: "gpt-5-codex", branch: null, @@ -52,10 +59,11 @@ const readModel: OrchestrationReadModel = { session: null, activities: [], checkpoints: [], + deletedAt: null, }, { - id: "thread-2", - projectId: "project-b", + id: ThreadId.makeUnsafe("thread-2"), + projectId: ProjectId.makeUnsafe("project-b"), title: "Thread B", model: "gpt-5-codex", branch: null, @@ -67,16 +75,17 @@ const readModel: OrchestrationReadModel = { session: null, activities: [], checkpoints: [], + deletedAt: null, }, ], }; const messageSendCommand: OrchestrationCommand = { type: "thread.turn.start", - commandId: "cmd-1", - threadId: "thread-1", + commandId: CommandId.makeUnsafe("cmd-1"), + threadId: ThreadId.makeUnsafe("thread-1"), message: { - messageId: "msg-1", + messageId: MessageId.makeUnsafe("msg-1"), role: "user", text: "hello", attachments: [], @@ -86,11 +95,13 @@ const messageSendCommand: OrchestrationCommand = { describe("commandInvariants", () => { it("finds threads by id and project", () => { - expect(findThreadById(readModel, "thread-1")?.projectId).toBe("project-a"); - expect(findThreadById(readModel, "missing")).toBeUndefined(); - expect(listThreadsByProjectId(readModel, "project-b").map((thread) => thread.id)).toEqual([ - "thread-2", - ]); + expect(findThreadById(readModel, ThreadId.makeUnsafe("thread-1"))?.projectId).toBe("project-a"); + expect(findThreadById(readModel, ThreadId.makeUnsafe("missing"))).toBeUndefined(); + expect( + listThreadsByProjectId(readModel, ProjectId.makeUnsafe("project-b")).map( + (thread) => thread.id, + ), + ).toEqual([ThreadId.makeUnsafe("thread-2")]); }); it("requires existing thread", async () => { @@ -98,17 +109,17 @@ describe("commandInvariants", () => { requireThread({ readModel, command: messageSendCommand, - threadId: "thread-1", + threadId: ThreadId.makeUnsafe("thread-1"), }), ); - expect(thread.id).toBe("thread-1"); + expect(thread.id).toBe(ThreadId.makeUnsafe("thread-1")); await expect( Effect.runPromise( requireThread({ readModel, command: messageSendCommand, - threadId: "missing", + threadId: ThreadId.makeUnsafe("missing"), }), ), ).rejects.toThrow("does not exist"); @@ -120,16 +131,16 @@ describe("commandInvariants", () => { readModel, command: { type: "thread.create", - commandId: "cmd-2", - threadId: "thread-3", - projectId: "project-a", + commandId: CommandId.makeUnsafe("cmd-2"), + threadId: ThreadId.makeUnsafe("thread-3"), + projectId: ProjectId.makeUnsafe("project-a"), title: "new", model: "gpt-5-codex", branch: null, worktreePath: null, createdAt: now, }, - threadId: "thread-3", + threadId: ThreadId.makeUnsafe("thread-3"), }), ); @@ -139,16 +150,16 @@ describe("commandInvariants", () => { readModel, command: { type: "thread.create", - commandId: "cmd-3", - threadId: "thread-1", - projectId: "project-a", + commandId: CommandId.makeUnsafe("cmd-3"), + threadId: ThreadId.makeUnsafe("thread-1"), + projectId: ProjectId.makeUnsafe("project-a"), title: "dup", model: "gpt-5-codex", branch: null, worktreePath: null, createdAt: now, }, - threadId: "thread-1", + threadId: ThreadId.makeUnsafe("thread-1"), }), ), ).rejects.toThrow("already exists"); @@ -157,7 +168,7 @@ describe("commandInvariants", () => { it("requires non-negative integers", async () => { await Effect.runPromise( requireNonNegativeInteger({ - commandType: "thread.revert", + commandType: "thread.checkpoint.revert", field: "turnCount", value: 0, }), @@ -166,7 +177,7 @@ describe("commandInvariants", () => { await expect( Effect.runPromise( requireNonNegativeInteger({ - commandType: "thread.revert", + commandType: "thread.checkpoint.revert", field: "turnCount", value: -1, }), diff --git a/apps/server/src/orchestration/decider.projectScripts.test.ts b/apps/server/src/orchestration/decider.projectScripts.test.ts index 65207dab7aaa5fa24d33c300b06d4ca6a26be157..c56d427fb126e7553723c986d775d23938d03f48 100644 --- a/apps/server/src/orchestration/decider.projectScripts.test.ts +++ b/apps/server/src/orchestration/decider.projectScripts.test.ts @@ -1,9 +1,16 @@ +import { CommandId, EventId, MessageId, ProjectId, ThreadId } from "@t3tools/contracts"; import { describe, expect, it } from "vitest"; import { Effect } from "effect"; import { decideOrchestrationCommand } from "./decider.ts"; import { createEmptyReadModel, projectEvent } from "./projector.ts"; +const asCommandId = (value: string): CommandId => CommandId.makeUnsafe(value); +const asEventId = (value: string): EventId => EventId.makeUnsafe(value); +const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); +const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); +const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); + describe("decider project scripts", () => { it("emits empty scripts on project.create", async () => { const now = new Date().toISOString(); @@ -13,8 +20,8 @@ describe("decider project scripts", () => { decideOrchestrationCommand({ command: { type: "project.create", - commandId: "cmd-project-create-scripts", - projectId: "project-scripts", + commandId: asCommandId("cmd-project-create-scripts"), + projectId: asProjectId("project-scripts"), title: "Scripts", workspaceRoot: "/tmp/scripts", createdAt: now, @@ -34,17 +41,17 @@ describe("decider project scripts", () => { const readModel = await Effect.runPromise( projectEvent(initial, { sequence: 1, - eventId: "evt-project-create-scripts", - aggregateType: "project", - aggregateId: "project-scripts", + eventId: asEventId("evt-project-create-scripts"), + aggregateKind: "project", + aggregateId: asProjectId("project-scripts"), type: "project.created", occurredAt: now, - commandId: "cmd-project-create-scripts", + commandId: asCommandId("cmd-project-create-scripts"), causationEventId: null, - correlationId: "cmd-project-create-scripts", + correlationId: asCommandId("cmd-project-create-scripts"), metadata: {}, payload: { - projectId: "project-scripts", + projectId: asProjectId("project-scripts"), title: "Scripts", workspaceRoot: "/tmp/scripts", defaultModel: null, @@ -69,8 +76,8 @@ describe("decider project scripts", () => { decideOrchestrationCommand({ command: { type: "project.meta.update", - commandId: "cmd-project-update-scripts", - projectId: "project-scripts", + commandId: asCommandId("cmd-project-update-scripts"), + projectId: asProjectId("project-scripts"), scripts: Array.from(scripts), }, readModel, @@ -88,17 +95,17 @@ describe("decider project scripts", () => { const withProject = await Effect.runPromise( projectEvent(initial, { sequence: 1, - eventId: "evt-project-create", - aggregateType: "project", - aggregateId: "project-1", + eventId: asEventId("evt-project-create"), + aggregateKind: "project", + aggregateId: asProjectId("project-1"), type: "project.created", occurredAt: now, - commandId: "cmd-project-create", + commandId: asCommandId("cmd-project-create"), causationEventId: null, - correlationId: "cmd-project-create", + correlationId: asCommandId("cmd-project-create"), metadata: {}, payload: { - projectId: "project-1", + projectId: asProjectId("project-1"), title: "Project", workspaceRoot: "/tmp/project", defaultModel: null, @@ -111,18 +118,18 @@ describe("decider project scripts", () => { const readModel = await Effect.runPromise( projectEvent(withProject, { sequence: 2, - eventId: "evt-thread-create", - aggregateType: "thread", - aggregateId: "thread-1", + eventId: asEventId("evt-thread-create"), + aggregateKind: "thread", + aggregateId: asThreadId("thread-1"), type: "thread.created", occurredAt: now, - commandId: "cmd-thread-create", + commandId: asCommandId("cmd-thread-create"), causationEventId: null, - correlationId: "cmd-thread-create", + correlationId: asCommandId("cmd-thread-create"), metadata: {}, payload: { - threadId: "thread-1", - projectId: "project-1", + threadId: asThreadId("thread-1"), + projectId: asProjectId("project-1"), title: "Thread", model: "gpt-5-codex", branch: null, @@ -137,10 +144,10 @@ describe("decider project scripts", () => { decideOrchestrationCommand({ command: { type: "thread.turn.start", - commandId: "cmd-turn-start", - threadId: "thread-1", + commandId: asCommandId("cmd-turn-start"), + threadId: asThreadId("thread-1"), message: { - messageId: "message-user-1", + messageId: asMessageId("message-user-1"), role: "user", text: "hello", attachments: [], diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 45802c2b98d3cf050890723fdceeb153b5862d00..798aa710024042ed1c74d3dbfc1a89c872f09742 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -1,4 +1,4 @@ -import type { OrchestrationEvent } from "@t3tools/contracts"; +import { CommandId, EventId, ProjectId, ThreadId, type OrchestrationEvent } from "@t3tools/contracts"; import { Effect } from "effect"; import { describe, expect, it } from "vitest"; @@ -8,24 +8,27 @@ function makeEvent(input: { sequence: number; type: OrchestrationEvent["type"]; occurredAt: string; - aggregateType: OrchestrationEvent["aggregateType"]; + aggregateKind: OrchestrationEvent["aggregateKind"]; aggregateId: string; commandId: string | null; payload: unknown; }): OrchestrationEvent { return { sequence: input.sequence, - eventId: `event-${input.sequence}`, + eventId: EventId.makeUnsafe(`event-${input.sequence}`), type: input.type, - aggregateType: input.aggregateType, - aggregateId: input.aggregateId as OrchestrationEvent["aggregateId"], + aggregateKind: input.aggregateKind, + aggregateId: + input.aggregateKind === "project" + ? ProjectId.makeUnsafe(input.aggregateId) + : ThreadId.makeUnsafe(input.aggregateId), occurredAt: input.occurredAt, - commandId: input.commandId as OrchestrationEvent["commandId"], + commandId: input.commandId === null ? null : CommandId.makeUnsafe(input.commandId), causationEventId: null, correlationId: null, metadata: {}, - payload: input.payload, - }; + payload: input.payload as never, + } as OrchestrationEvent; } describe("orchestration projector", () => { @@ -39,7 +42,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 1, type: "thread.created", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: now, commandId: "cmd-thread-create", @@ -89,7 +92,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 1, type: "thread.created", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: now, commandId: "cmd-invalid", @@ -119,7 +122,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 7, type: "thread.turn-start-requested", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-01-01T00:00:00.000Z", commandId: "cmd-unhandled", @@ -148,7 +151,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 1, type: "thread.created", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", @@ -172,7 +175,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 2, type: "thread.session-set", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: startedAt, commandId: "cmd-running", @@ -210,7 +213,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 1, type: "thread.created", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", @@ -234,7 +237,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 2, type: "thread.message-sent", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: deltaAt, commandId: "cmd-delta", @@ -258,7 +261,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 3, type: "thread.message-sent", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: completeAt, commandId: "cmd-complete", @@ -293,7 +296,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 1, type: "thread.created", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: createdAt, commandId: "cmd-create", @@ -315,7 +318,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 2, type: "thread.message-sent", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:01.000Z", commandId: "cmd-user-1", @@ -333,7 +336,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 3, type: "thread.message-sent", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.000Z", commandId: "cmd-assistant-1", @@ -351,7 +354,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 4, type: "thread.turn-diff-completed", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.500Z", commandId: "cmd-turn-1-complete", @@ -369,7 +372,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 5, type: "thread.activity-appended", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:02.750Z", commandId: "cmd-activity-1", @@ -389,7 +392,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 6, type: "thread.message-sent", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:03.000Z", commandId: "cmd-user-2", @@ -407,7 +410,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 7, type: "thread.message-sent", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.000Z", commandId: "cmd-assistant-2", @@ -425,7 +428,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 8, type: "thread.turn-diff-completed", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.500Z", commandId: "cmd-turn-2-complete", @@ -443,7 +446,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 9, type: "thread.activity-appended", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:04.750Z", commandId: "cmd-activity-2", @@ -463,7 +466,7 @@ describe("orchestration projector", () => { makeEvent({ sequence: 10, type: "thread.reverted", - aggregateType: "thread", + aggregateKind: "thread", aggregateId: "thread-1", occurredAt: "2026-02-23T10:00:05.000Z", commandId: "cmd-revert", diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index c8c6bb334e5029e05382a89dd1986abb4aacfcbe..f6a9a108a2528525a8e844ee1b399ce28673bc67 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -1,10 +1,16 @@ -import type { - ProviderApprovalDecision, - ProviderEvent, - ProviderSendTurnInput, - ProviderSession, - ProviderSessionStartInput, - ProviderTurnStartResult, +import { + ApprovalRequestId, + EventId, + ProviderItemId, + type ProviderApprovalDecision, + type ProviderEvent, + type ProviderSendTurnInput, + ProviderSessionId, + type ProviderSession, + type ProviderSessionStartInput, + ProviderThreadId, + ProviderTurnId, + type ProviderTurnStartResult, } from "@t3tools/contracts"; import { afterAll, assert, it, vi } from "@effect/vitest"; import { assertFailure } from "@effect/vitest/utils"; @@ -16,15 +22,21 @@ import { ProviderAdapterValidationError } from "../Errors.ts"; import { CodexAdapter } from "../Services/CodexAdapter.ts"; import { makeCodexAdapterLive } from "./CodexAdapter.ts"; +const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); +const asThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); +const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); +const asEventId = (value: string): EventId => EventId.makeUnsafe(value); +const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); + class FakeCodexManager extends CodexAppServerManager { public startSessionImpl = vi.fn( async (input: ProviderSessionStartInput): Promise => { const now = new Date().toISOString(); return { - sessionId: "sess-1", + sessionId: asSessionId("sess-1"), provider: "codex", status: "ready", - threadId: "thread-1", + threadId: asThreadId("thread-1"), cwd: input.cwd, createdAt: now, updatedAt: now, @@ -34,29 +46,29 @@ class FakeCodexManager extends CodexAppServerManager { public sendTurnImpl = vi.fn( async (_input: ProviderSendTurnInput): Promise => ({ - threadId: "thread-1", - turnId: "turn-1", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), }), ); public interruptTurnImpl = vi.fn( - async (_sessionId: string, _turnId?: string): Promise => undefined, + async (_sessionId: ProviderSessionId, _turnId?: ProviderTurnId): Promise => undefined, ); - public readThreadImpl = vi.fn(async (_sessionId: string) => ({ - threadId: "thread-1", + public readThreadImpl = vi.fn(async (_sessionId: ProviderSessionId) => ({ + threadId: asThreadId("thread-1"), turns: [], })); - public rollbackThreadImpl = vi.fn(async (_sessionId: string, _numTurns: number) => ({ - threadId: "thread-1", + public rollbackThreadImpl = vi.fn(async (_sessionId: ProviderSessionId, _numTurns: number) => ({ + threadId: asThreadId("thread-1"), turns: [], })); public respondToRequestImpl = vi.fn( async ( - _sessionId: string, - _requestId: string, + _sessionId: ProviderSessionId, + _requestId: ApprovalRequestId, _decision: ProviderApprovalDecision, ): Promise => undefined, ); @@ -71,33 +83,33 @@ class FakeCodexManager extends CodexAppServerManager { return this.sendTurnImpl(input); } - override interruptTurn(sessionId: string, turnId?: string): Promise { + override interruptTurn(sessionId: ProviderSessionId, turnId?: ProviderTurnId): Promise { return this.interruptTurnImpl(sessionId, turnId); } - override readThread(sessionId: string) { + override readThread(sessionId: ProviderSessionId) { return this.readThreadImpl(sessionId); } - override rollbackThread(sessionId: string, numTurns: number) { + override rollbackThread(sessionId: ProviderSessionId, numTurns: number) { return this.rollbackThreadImpl(sessionId, numTurns); } override respondToRequest( - sessionId: string, - requestId: string, + sessionId: ProviderSessionId, + requestId: ApprovalRequestId, decision: ProviderApprovalDecision, ): Promise { return this.respondToRequestImpl(sessionId, requestId, decision); } - override stopSession(_sessionId: string): void {} + override stopSession(_sessionId: ProviderSessionId): void {} override listSessions(): ProviderSession[] { return []; } - override hasSession(_sessionId: string): boolean { + override hasSession(_sessionId: ProviderSessionId): boolean { return false; } @@ -144,7 +156,7 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { const adapter = yield* CodexAdapter; const result = yield* adapter .sendTurn({ - sessionId: "sess-missing", + sessionId: asSessionId("sess-missing"), input: "hello", attachments: [], }) @@ -176,15 +188,15 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); const event: ProviderEvent = { - id: "evt-msg-complete", + id: asEventId("evt-msg-complete"), kind: "notification", provider: "codex", - sessionId: "sess-1", + sessionId: asSessionId("sess-1"), createdAt: new Date().toISOString(), method: "item/completed", - threadId: "thread-1", - turnId: "turn-1", - itemId: "msg_1", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + itemId: asItemId("msg_1"), payload: { item: { type: "agentMessage", @@ -220,13 +232,13 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { ).pipe(Effect.forkChild); const event: ProviderEvent = { - id: "evt-1", + id: asEventId("evt-1"), kind: "notification", provider: "codex", - sessionId: "sess-1", + sessionId: asSessionId("sess-1"), createdAt: new Date().toISOString(), method: "turn/started", - turnId: "turn-1", + turnId: asTurnId("turn-1"), }; lifecycleManager.emit("event", event); diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 5e97f3538f8ec19baf9a267c14f29d52790ac970..8b63284967544e299a6f34073a13d8e6e8ba4859 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -45,17 +45,6 @@ import { SqlitePersistenceMemory, } from "../../persistence/Layers/Sqlite.ts"; -function decodeIssueMessage(cause: unknown): string { - if (cause instanceof Error && cause.message.length > 0) { - return cause.message; - } - try { - return JSON.stringify(cause); - } catch { - return "Invalid provider input."; - } -} - const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); @@ -526,11 +515,18 @@ routing.layer("ProviderServiceLive routing", (it) => { const payload = runningRuntime.value.runtimePayload; assert.equal(payload !== null && typeof payload === "object", true); if (payload !== null && typeof payload === "object" && !Array.isArray(payload)) { - assert.equal(payload.cwd, process.cwd()); - assert.equal(payload.model, null); - assert.equal(payload.activeTurnId, "turn-1"); - assert.equal(payload.lastError, null); - assert.equal(payload.lastRuntimeEvent, "provider.sendTurn"); + const runtimePayload = payload as { + cwd: string; + model: string | null; + activeTurnId: string | null; + lastError: string | null; + lastRuntimeEvent: string | null; + }; + assert.equal(runtimePayload.cwd, process.cwd()); + assert.equal(runtimePayload.model, null); + assert.equal(runtimePayload.activeTurnId, "turn-1"); + assert.equal(runtimePayload.lastError, null); + assert.equal(runtimePayload.lastRuntimeEvent, "provider.sendTurn"); } } @@ -669,32 +665,22 @@ validation.layer("ProviderServiceLive validation", (it) => { Effect.gen(function* () { const provider = yield* ProviderService; - let parseCause: unknown; - let decodeFailed = false; - try { - Schema.decodeUnknownSync(ProviderSessionStartInput)({ - provider: "invalid-provider", - }); - } catch (cause) { - parseCause = cause; - decodeFailed = true; - } - assert.equal(decodeFailed, true); - const failure = yield* Effect.result( provider.startSession({ provider: "invalid-provider", } as never), ); - assertFailure( - failure, - new ProviderValidationError({ - operation: "ProviderService.startSession", - issue: decodeIssueMessage(parseCause), - cause: parseCause, - }), - ); + assert.equal(failure._tag, "Failure"); + if (failure._tag !== "Failure") { + return; + } + assert.equal(failure.failure._tag, "ProviderValidationError"); + if (failure.failure._tag !== "ProviderValidationError") { + return; + } + assert.equal(failure.failure.operation, "ProviderService.startSession"); + assert.equal(failure.failure.issue.includes("invalid-provider"), true); }), ); diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 75c02ca4079de6fd5aa5384ac4b860f381edfd62..b2f2eba5b838efa6fd641fa59d40b3ccf53456da 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -10,15 +10,20 @@ import WebSocket from "ws"; import { DEFAULT_TERMINAL_ID, + EventId, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, + ProviderItemId, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, WS_CHANNELS, WS_METHODS, + type WebSocketResponse, type ProviderRuntimeEvent, type KeybindingsConfig, type ResolvedKeybindingsConfig, type WsPush, - type WsResponse, } from "@t3tools/contracts"; import { compileResolvedKeybindingRule, DEFAULT_KEYBINDINGS } from "./keybindings"; import type { @@ -34,6 +39,7 @@ import type { TerminalManager } from "./terminalManager"; import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite"; import { SqlClient } from "effect/unstable/sql"; import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; +import type { OpenShape } from "./open"; interface PendingMessages { queue: unknown[]; @@ -42,6 +48,12 @@ interface PendingMessages { const pendingBySocket = new WeakMap(); +const asEventId = (value: string): EventId => EventId.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); + class MockTerminalManager extends EventEmitter<{ event: [event: TerminalEvent] }> { private readonly sessions = new Map(); @@ -187,7 +199,19 @@ function waitForMessage(ws: WebSocket): Promise { }); } -async function sendRequest(ws: WebSocket, method: string, params?: unknown): Promise { +function asWebSocketResponse(message: unknown): WebSocketResponse | null { + if (typeof message !== "object" || message === null) return null; + if (!("id" in message)) return null; + const id = (message as { id?: unknown }).id; + if (typeof id !== "string") return null; + return message as WebSocketResponse; +} + +async function sendRequest( + ws: WebSocket, + method: string, + params?: unknown, +): Promise { const id = crypto.randomUUID(); const body = method === ORCHESTRATION_WS_METHODS.dispatchCommand @@ -200,12 +224,15 @@ async function sendRequest(ws: WebSocket, method: string, params?: unknown): Pro // Wait for response with matching id while (true) { - const parsed = (await waitForMessage(ws)) as Record; + const parsed = asWebSocketResponse(await waitForMessage(ws)); + if (!parsed) { + continue; + } if (parsed.id === id) { - return parsed as WsResponse; + return parsed; } if (parsed.id === "unknown") { - return parsed as WsResponse; + return parsed; } } } @@ -233,7 +260,7 @@ async function waitForPush( } function compileKeybindings(bindings: KeybindingsConfig): ResolvedKeybindingsConfig { - const resolved: ResolvedKeybindingsConfig = []; + const resolved: Array = []; for (const binding of bindings) { const compiled = compileResolvedKeybindingRule(binding); if (!compiled) { @@ -278,6 +305,7 @@ describe("WebSocket Server", () => { authToken?: string; stateDir?: string; providerLayer?: Layer.Layer; + open?: OpenShape; gitManager?: { status: (input: { cwd: string }) => Promise; runStackedAction: (input: { cwd: string; action: string }) => Promise; @@ -297,6 +325,7 @@ describe("WebSocket Server", () => { ...(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 } : {}), }); @@ -420,6 +449,33 @@ describe("WebSocket Server", () => { }); }); + it("routes shell.openInEditor through the injected open service", async () => { + const openCalls: Array<{ cwd: string; editor: string }> = []; + const openService: OpenShape = { + openBrowser: () => Effect.void, + openInEditor: (input) => { + openCalls.push({ cwd: input.cwd, editor: input.editor }); + return Effect.void; + }, + }; + + server = createTestServer({ cwd: "/my/workspace", open: openService }); + await server.start(); + const addr = server.httpServer.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + + const ws = await connectWs(port); + connections.push(ws); + await waitForMessage(ws); + + const response = await sendRequest(ws, WS_METHODS.shellOpenInEditor, { + cwd: "/my/workspace", + editor: "cursor", + }); + expect(response.error).toBeUndefined(); + expect(openCalls).toEqual([{ cwd: "/my/workspace", editor: "cursor" }]); + }); + it("reads keybindings from ~/.t3/keybindings.json", async () => { const fakeHome = makeTempDir("t3code-home-"); const configDir = path.join(fakeHome, ".t3"); @@ -787,7 +843,7 @@ describe("WebSocket Server", () => { it("keeps orchestration domain push behavior for provider runtime events", async () => { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); - const sessionId = "sess-test"; + const sessionId = asProviderSessionId("sess-test"); const emitRuntimeEvent = (event: ProviderRuntimeEvent) => { Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); }; @@ -798,11 +854,15 @@ describe("WebSocket Server", () => { sessionId, provider: "codex", status: "ready", - threadId: "provider-thread-1", + threadId: asProviderThreadId("provider-thread-1"), createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), }), - sendTurn: () => Effect.succeed({ threadId: "provider-thread-1", turnId: "provider-turn-1" }), + sendTurn: () => + Effect.succeed({ + threadId: asProviderThreadId("provider-thread-1"), + turnId: asProviderTurnId("provider-turn-1"), + }), interruptTurn: () => unsupported(), respondToRequest: () => unsupported(), stopSession: () => unsupported(), @@ -870,12 +930,12 @@ describe("WebSocket Server", () => { emitRuntimeEvent({ type: "message.delta", - eventId: "evt-ws-runtime-message-delta", + eventId: asEventId("evt-ws-runtime-message-delta"), provider: "codex", sessionId, createdAt: new Date().toISOString(), - turnId: "turn-1", - itemId: "item-1", + turnId: asProviderTurnId("turn-1"), + itemId: asProviderItemId("item-1"), delta: "hello from runtime", }); @@ -1015,15 +1075,18 @@ describe("WebSocket Server", () => { // Send garbage ws.send("not json at all"); - let response: WsResponse | null = null; + let response: WebSocketResponse | null = null; for (let attempt = 0; attempt < 5; attempt += 1) { - const message = (await waitForMessage(ws)) as Record; - if (typeof message.id === "string" && message.id === "unknown") { - response = message as WsResponse; + const message = asWebSocketResponse(await waitForMessage(ws)); + if (!message) { + continue; + } + if (message.id === "unknown") { + response = message; break; } if (message.error) { - response = message as WsResponse; + response = message; break; } } diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index bb2bb8aeb1ad8cee46e762caa7940976a25c4449..b45b266714c45088e330d6c98e6ce1cb42f509b6 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -1,4 +1,3 @@ -import { spawn } from "node:child_process"; import fs from "node:fs"; import http from "node:http"; import os from "node:os"; @@ -7,7 +6,6 @@ import type { Duplex } from "node:stream"; import { CheckpointRef, - EDITORS, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, TerminalEvent, @@ -18,7 +16,7 @@ import { WsResponse, } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; -import { Effect, Exit, Layer, ManagedRuntime, Schema, Scope, Stream } from "effect"; +import { Effect, Exit, Layer, ManagedRuntime, Schema, Scope, ServiceMap, Stream } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; @@ -62,12 +60,16 @@ import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionD import { ProviderService, type ProviderServiceShape } 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 { + CheckpointStore, + type CheckpointStoreShape, +} from "./checkpointing/Services/CheckpointStore"; import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger"; 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"; const MIME_TYPES: Record = { ".html": "text/html; charset=utf-8", @@ -100,8 +102,25 @@ export interface ServerOptions { 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; +} + +export class Server extends ServiceMap.Service()("server/Server") {} + +export const ServerLive = Layer.succeed(Server, { + createServer, +} satisfies ServerShape); + const isServerNotRunningError = (error: unknown): boolean => { if (!(error instanceof Error)) return false; const maybeCode = (error as NodeJS.ErrnoException).code; @@ -162,9 +181,7 @@ function websocketRawToString(raw: unknown): string | null { return null; } -function stripRequestTag( - body: T, -): Omit { +function stripRequestTag(body: T): Omit { const { _tag: _ignoredTag, ...rest } = body; return rest; } @@ -185,7 +202,7 @@ type EffectRuntime = ManagedRuntime.ManagedRuntime< unknown >; -export function createServer(options: ServerOptions) { +export function createServer(options: ServerOptions): ServerRuntime { const { port, host, @@ -200,6 +217,7 @@ export function createServer(options: ServerOptions) { terminalManager = new TerminalManager(), providerLayer: customProviderLayer, authToken, + open: openService = OpenDefaultService, } = options; const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata"); @@ -626,42 +644,7 @@ export function createServer(options: ServerOptions) { return searchWorkspaceEntries(stripRequestTag(request.body)); case WS_METHODS.shellOpenInEditor: { - const body = request.body; - const editorDef = EDITORS.find((e) => e.id === body.editor); - if (!editorDef) throw new Error(`Unknown editor: ${body.editor}`); - - let command: string; - let args: string[]; - - if (editorDef.command) { - command = editorDef.command; - args = [body.cwd]; - } else if (editorDef.id === "file-manager") { - // Use platform-specific file manager command - switch (process.platform) { - case "darwin": - command = "open"; - break; - case "win32": - command = "explorer"; - break; - default: - command = "xdg-open"; - break; - } - args = [body.cwd]; - } else { - return undefined; - } - - const child = spawn(command, args, { - detached: true, - stdio: "ignore", - }); - child.on("error", () => { - /* ignore spawn failures for detached editors */ - }); - child.unref(); + await Effect.runPromise(openService.openInEditor(stripRequestTag(request.body))); return undefined; } @@ -763,10 +746,7 @@ export function createServer(options: ServerOptions) { ); return makeProviderServiceLive({ canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"), - }).pipe( - Layer.provide(adapterRegistryLayer), - Layer.provide(providerSessionDirectoryLayer), - ); + }).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer)); })(); const runtimeServicesLayer = Layer.mergeAll( diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index 23300dbb118c26b6ee0b591cbc7e337918d81b92..b99ca864e7780a800a53ac25fff4077b3cc0d399 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -12,6 +12,5 @@ } ] }, - "include": ["src", "tsdown.config.ts", "scripts", "integration"], - "exclude": ["src/**/*.test.ts"] + "include": ["src", "tsdown.config.ts", "scripts", "integration"] } diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 503d928a065272fa011ffba95665244344a8beaf..3d36967bdd5a1424ae3a65f161f13f7cb9e511de 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -68,7 +68,7 @@ export class GitRemoveWorktreeInput extends Schema.Class )({ cwd: TrimmedNonEmptyString, path: TrimmedNonEmptyString, - force: Schema.Boolean.pipe(Schema.NullOr), + force: Schema.optional(Schema.Boolean), }) {} export class GitCreateBranchInput extends Schema.Class( diff --git a/packages/contracts/src/keybindings.test.ts b/packages/contracts/src/keybindings.test.ts index 212514b1975649d86c78f8b277da8dfe7133c2c2..50c1075c1623d1895c024c4707d7aedc15150f97 100644 --- a/packages/contracts/src/keybindings.test.ts +++ b/packages/contracts/src/keybindings.test.ts @@ -12,9 +12,7 @@ const decode = (schema: S, input: unknown): Schema.Schema. Schema.decodeUnknownSync(schema as never)(input) as Schema.Schema.Type; const decodeResolvedRuleStrict = (input: unknown) => - Schema.decodeUnknownSync(ResolvedKeybindingRule as never)(input, { - onExcessProperty: "error", - }); + Schema.decodeUnknownSync(ResolvedKeybindingRule as never)(input); describe("keybindings contracts", () => { it("parses keybinding rules", () => { diff --git a/packages/contracts/src/keybindings.ts b/packages/contracts/src/keybindings.ts index fa71cfbd7469eebb69b8194c49cae45dc5ab2b64..9768e4f90aced5ac0e759cfd4b5193386323429f 100644 --- a/packages/contracts/src/keybindings.ts +++ b/packages/contracts/src/keybindings.ts @@ -70,11 +70,14 @@ export type KeybindingWhenNode = export class ResolvedKeybindingRule extends Schema.Class( "ResolvedKeybindingRule", -)({ - command: KeybindingCommand, - shortcut: KeybindingShortcut, - whenAst: Schema.optional(KeybindingWhenNode), -}) {} +)( + { + command: KeybindingCommand, + shortcut: KeybindingShortcut, + whenAst: Schema.optional(KeybindingWhenNode), + }, + { parseOptions: { onExcessProperty: "error" } }, +) {} export const ResolvedKeybindingsConfig = Schema.Array(ResolvedKeybindingRule).check( Schema.isMaxLength(256), diff --git a/scripts/dev-runner.mjs b/scripts/dev-runner.mjs index 89b9ef3c26cec93e40f5a199a10c6122046355e7..dce79715c9740632a8c7e822912f24513ca44ae6 100644 --- a/scripts/dev-runner.mjs +++ b/scripts/dev-runner.mjs @@ -84,6 +84,15 @@ function main() { VITE_WS_URL: `ws://localhost:${serverPort}`, VITE_DEV_SERVER_URL: `http://localhost:${webPort}`, }; + + if (mode === "dev" || mode === "dev:server" || mode === "dev:web") { + // Running server/web in browser mode should not inherit desktop launcher state. + env.T3CODE_MODE = "web"; + delete env.T3CODE_NO_BROWSER; + delete env.T3CODE_AUTH_TOKEN; + delete env.T3CODE_DESKTOP_WS_URL; + } + if (mode === "dev" && !env.T3CODE_LOG_WS_EVENTS) { env.T3CODE_LOG_WS_EVENTS = "1"; } -- 2.54.0