From a561e0bf8f54f5315e968ecb3eba80fb709c13c6 Mon Sep 17 00:00:00 2001 From: Jeffrey Tseng Date: Sun, 23 Aug 2026 23:43:20 -0400 Subject: [PATCH] feat(runtime): add hosted Pi server --- package-lock.json | 46 ++- package.json | 5 +- packages/client/package.json | 4 + packages/client/src/websocket.ts | 183 +++++++++ packages/coding-agent-server/README.md | 18 + packages/coding-agent-server/package.json | 31 ++ packages/coding-agent-server/src/cli.ts | 103 +++++ packages/coding-agent-server/src/index.ts | 9 + packages/coding-agent-server/src/service.ts | 374 ++++++++++++++++++ .../coding-agent-server/src/usage-reporter.ts | 164 ++++++++ .../test/usage-reporter.test.ts | 78 ++++ .../test/websocket-flow.test.ts | 155 ++++++++ .../coding-agent-server/tsconfig.build.json | 15 + .../coding-agent-server/tsconfig.test.json | 5 + packages/coding-agent-server/vitest.config.ts | 3 + .../coding-agent/src/core/session-manager.ts | 6 + packages/server/README.md | 6 + packages/server/package.json | 8 +- .../server/src/transports/websocket/index.ts | 7 + .../src/transports/websocket/listener.ts | 254 ++++++++++++ .../server/src/transports/websocket/preset.ts | 15 + .../server/src/transports/websocket/types.ts | 19 + 22 files changed, 1501 insertions(+), 7 deletions(-) create mode 100644 packages/client/src/websocket.ts create mode 100644 packages/coding-agent-server/README.md create mode 100644 packages/coding-agent-server/package.json create mode 100644 packages/coding-agent-server/src/cli.ts create mode 100644 packages/coding-agent-server/src/index.ts create mode 100644 packages/coding-agent-server/src/service.ts create mode 100644 packages/coding-agent-server/src/usage-reporter.ts create mode 100644 packages/coding-agent-server/test/usage-reporter.test.ts create mode 100644 packages/coding-agent-server/test/websocket-flow.test.ts create mode 100644 packages/coding-agent-server/tsconfig.build.json create mode 100644 packages/coding-agent-server/tsconfig.test.json create mode 100644 packages/coding-agent-server/vitest.config.ts create mode 100644 packages/server/src/transports/websocket/index.ts create mode 100644 packages/server/src/transports/websocket/listener.ts create mode 100644 packages/server/src/transports/websocket/preset.ts create mode 100644 packages/server/src/transports/websocket/types.ts diff --git a/package-lock.json b/package-lock.json index 31f5b53b8..6b4540941 100644 --- a/package-lock.json +++ b/package-lock.json @@ -735,6 +735,10 @@ "resolved": "packages/coding-agent", "link": true }, + "node_modules/@earendil-works/diffex-coding-agent-server": { + "resolved": "packages/coding-agent-server", + "link": true + }, "node_modules/@earendil-works/diffex-evals": { "resolved": "packages/evals", "link": true @@ -2199,6 +2203,16 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@typescript/native-preview": { "version": "7.0.0-dev.20260120.1", "resolved": "https://registry.npmjs.org/@typescript/native-preview/-/native-preview-7.0.0-dev.20260120.1.tgz", @@ -4076,9 +4090,9 @@ "optional": true }, "node_modules/nanoid": { - "version": "3.3.16", - "resolved": "https://registry.npmjs.org/nanoid/-/nanoid-3.3.16.tgz", - "integrity": "sha512-bzlKTyNJ7+LdGIIwy8ijFpIqEQIvafahV7eYykJ8Cvh42EdJeODoJ6gUJXpQJvej1BddH8OqTXZNE/KfbWAu8Q==", + "version": "3.3.18", + "resolved": "https://registry.npmjs.org/nanoid/-/nanoid-3.3.18.tgz", + "integrity": "sha512-DTg4MJbGMWkfi6VZFdNt2/caMbQy4Ou+Op/hJQvGEWcnVfoA1QA+xzRKAzw9jD6+GVOOeYr/mIcuDSdug6F6+w==", "dev": true, "funding": [ { @@ -5597,6 +5611,28 @@ "@mariozechner/clipboard": "0.3.9" } }, + "packages/coding-agent-server": { + "name": "@earendil-works/diffex-coding-agent-server", + "version": "0.2.0", + "license": "Apache-2.0", + "dependencies": { + "@earendil-works/diffex-coding-agent": "^0.2.0", + "@earendil-works/diffex-protocol": "^0.2.0", + "@earendil-works/diffex-server": "^0.2.0" + }, + "bin": { + "diffex-runtime-server": "dist/cli.js" + }, + "devDependencies": { + "@earendil-works/diffex-client": "^0.2.0", + "shx": "0.4.0", + "vitest": "4.1.9", + "ws": "8.21.0" + }, + "engines": { + "node": ">=22.19.0" + } + }, "packages/coding-agent/examples/extensions/custom-provider-anthropic": { "name": "diffex-extension-custom-provider-anthropic", "version": "0.2.0", @@ -5830,9 +5866,11 @@ "license": "MIT", "dependencies": { "@earendil-works/diffex-ai": "^0.2.0", - "@earendil-works/diffex-protocol": "^0.2.0" + "@earendil-works/diffex-protocol": "^0.2.0", + "ws": "8.21.0" }, "devDependencies": { + "@types/ws": "8.18.1", "shx": "0.4.0", "vitest": "4.1.9" }, diff --git a/package.json b/package.json index 31a6af94f..a9552c6ab 100644 --- a/package.json +++ b/package.json @@ -13,8 +13,8 @@ ], "scripts": { "clean": "npm run clean --workspaces", - "build": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../harness-trace && npm run build && cd ../evolution && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build", - "build:offline": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../harness-trace && npm run build && cd ../evolution && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build", + "build": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../harness-trace && npm run build && cd ../evolution && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build && cd ../coding-agent-server && npm run build", + "build:offline": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../harness-trace && npm run build && cd ../evolution && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build && cd ../coding-agent-server && npm run build", "check": "biome check --write --error-on-warnings . && npm run check:pinned-deps && npm run check:ts-imports && npm run check:shrinkwrap && npm run check:install-lock:coding-agent && tsgo --noEmit && npm run check:browser-smoke", "check:browser-smoke": "node scripts/check-browser-smoke.mjs", "check:pinned-deps": "node scripts/check-pinned-deps.mjs", @@ -65,6 +65,7 @@ }, "version": "0.0.3", "overrides": { + "nanoid": "3.3.18", "protobufjs": "7.6.5", "rimraf": "6.1.2", "gaxios": { diff --git a/packages/client/package.json b/packages/client/package.json index 371b36037..3c229f284 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -14,6 +14,10 @@ "types": "./dist/unix.d.ts", "import": "./dist/unix.js" }, + "./websocket": { + "types": "./dist/websocket.d.ts", + "import": "./dist/websocket.js" + }, "./package.json": "./package.json" }, "sideEffects": false, diff --git a/packages/client/src/websocket.ts b/packages/client/src/websocket.ts new file mode 100644 index 000000000..384ddaddc --- /dev/null +++ b/packages/client/src/websocket.ts @@ -0,0 +1,183 @@ +import { DEFAULT_MAX_FRAME_LENGTH } from "@earendil-works/diffex-protocol"; +import type { ByteTransport, ByteTransportFactory, ByteTransportHandlers } from "./transport.ts"; + +type WebSocketData = ArrayBuffer | ArrayBufferView; + +interface WebSocketMessageEvent { + readonly data: unknown; +} + +interface WebSocketCloseEvent { + readonly code: number; + readonly reason: string; +} + +interface WebSocketErrorEvent { + readonly error?: unknown; +} + +interface WebSocketLike { + binaryType: string; + readonly bufferedAmount: number; + readonly readyState: number; + send(data: WebSocketData): void; + close(code?: number, reason?: string): void; + addEventListener(type: "open", listener: () => void, options?: { once?: boolean }): void; + addEventListener(type: "message", listener: (event: WebSocketMessageEvent) => void): void; + addEventListener(type: "close", listener: (event: WebSocketCloseEvent) => void, options?: { once?: boolean }): void; + addEventListener(type: "error", listener: (event: WebSocketErrorEvent) => void, options?: { once?: boolean }): void; +} + +export interface WebSocketConstructor { + new (url: string | URL, protocols?: string | string[]): WebSocketLike; + readonly OPEN: number; +} + +export interface WebSocketTransportOptions { + url: string | URL | (() => string | URL | Promise); + protocols?: string | string[]; + maxPendingBytes?: number; + webSocket?: WebSocketConstructor; +} + +/** Creates authenticated browser-compatible WebSocket transports for DiffexClient connection attempts. */ +export function createWebSocketTransportFactory(options: WebSocketTransportOptions): ByteTransportFactory { + const maxPendingBytes = options.maxPendingBytes ?? DEFAULT_MAX_FRAME_LENGTH * 4; + if (!Number.isSafeInteger(maxPendingBytes) || maxPendingBytes <= 0) { + throw new TypeError("WebSocket transport maxPendingBytes must be a positive safe integer"); + } + + return async (handlers) => { + const url = typeof options.url === "function" ? await options.url() : options.url; + const WebSocketImpl = options.webSocket ?? resolveGlobalWebSocket(); + return connectWebSocket(WebSocketImpl, url, options.protocols, maxPendingBytes, handlers); + }; +} + +function resolveGlobalWebSocket(): WebSocketConstructor { + const candidate = (globalThis as { WebSocket?: WebSocketConstructor }).WebSocket; + if (!candidate) throw new Error("WebSocket is not available in this runtime"); + return candidate; +} + +function connectWebSocket( + WebSocketImpl: WebSocketConstructor, + url: string | URL, + protocols: string | string[] | undefined, + maxPendingBytes: number, + handlers: ByteTransportHandlers, +): Promise { + return new Promise((resolve, reject) => { + const socket = protocols === undefined ? new WebSocketImpl(url) : new WebSocketImpl(url, protocols); + socket.binaryType = "arraybuffer"; + let connected = false; + let terminal = false; + + socket.addEventListener( + "open", + () => { + if (terminal) return; + connected = true; + resolve( + new WebSocketByteTransport(socket, WebSocketImpl.OPEN, maxPendingBytes, () => { + terminal = true; + }), + ); + }, + { once: true }, + ); + socket.addEventListener("message", (event) => { + if (terminal) return; + const bytes = binaryMessage(event.data); + if (!bytes) { + terminal = true; + socket.close(1003, "Binary protocol frames required"); + handlers.onError(new Error("WebSocket transport received a non-binary message")); + return; + } + handlers.onData(bytes); + }); + socket.addEventListener( + "close", + (event) => { + if (terminal) return; + terminal = true; + if (connected) handlers.onClose(); + else reject(new Error(`WebSocket transport closed before connecting (${event.code}: ${event.reason})`)); + }, + { once: true }, + ); + socket.addEventListener( + "error", + (event) => { + if (terminal) return; + terminal = true; + const error = event.error instanceof Error ? event.error : new Error("WebSocket transport failed"); + socket.close(); + if (connected) handlers.onError(error); + else reject(error); + }, + { once: true }, + ); + }); +} + +function binaryMessage(data: unknown): Uint8Array | undefined { + if (data instanceof ArrayBuffer) return new Uint8Array(data); + if (ArrayBuffer.isView(data)) return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + return undefined; +} + +class WebSocketByteTransport implements ByteTransport { + readonly #socket: WebSocketLike; + readonly #openState: number; + readonly #maxPendingBytes: number; + readonly #markLocalClose: () => void; + #closed = false; + #pendingBytes = 0; + #sendTail: Promise = Promise.resolve(); + + constructor(socket: WebSocketLike, openState: number, maxPendingBytes: number, markLocalClose: () => void) { + this.#socket = socket; + this.#openState = openState; + this.#maxPendingBytes = maxPendingBytes; + this.#markLocalClose = markLocalClose; + } + + send(chunk: Uint8Array): Promise { + if (!(chunk instanceof Uint8Array)) { + return Promise.reject(new TypeError("WebSocket transport chunks must be Uint8Array")); + } + if (this.#closed || this.#socket.readyState !== this.#openState) { + return Promise.reject(new Error("WebSocket transport is closed")); + } + if (this.#pendingBytes + chunk.byteLength > this.#maxPendingBytes) { + return Promise.reject(new Error("WebSocket transport exceeded its pending byte limit")); + } + + const bytes = chunk.slice(); + this.#pendingBytes += bytes.byteLength; + const send = this.#sendTail.then(() => { + if (this.#closed || this.#socket.readyState !== this.#openState) { + throw new Error("WebSocket transport is closed"); + } + this.#socket.send(bytes); + if (this.#socket.bufferedAmount > this.#maxPendingBytes) { + this.close(); + throw new Error("WebSocket transport exceeded its buffered byte limit"); + } + }); + const tracked = send.finally(() => { + this.#pendingBytes -= bytes.byteLength; + }); + this.#sendTail = tracked.catch(() => {}); + return tracked; + } + + close(): void { + if (this.#closed) return; + this.#closed = true; + this.#markLocalClose(); + this.#socket.close(1000, "Client closed"); + } +} diff --git a/packages/coding-agent-server/README.md b/packages/coding-agent-server/README.md new file mode 100644 index 000000000..bbfe6b946 --- /dev/null +++ b/packages/coding-agent-server/README.md @@ -0,0 +1,18 @@ +# @earendil-works/diffex-coding-agent-server + +Production service adapter between the Pi-derived Diffex coding-agent runtime and `@earendil-works/diffex-server`. + +The hosted control plane allocates one cloud session per sandbox and passes its stable session ID to this service. +Protocol clients attach to that existing session rather than creating an unrelated runtime identity. + +This package owns coding-agent snapshot projection and runtime lifecycle integration. +It does not own WorkOS authentication, billing, cloud policy, sandbox allocation, or public attachment credentials. + +## Runtime process + +The `diffex-runtime-server` executable starts the service with an authenticated WebSocket listener. +The cloud bootstrap supplies `DIFFEX_SESSION_ID`, `DIFFEX_WORKSPACE`, and an internal `DIFFEX_RUNTIME_SHARED_SECRET` of at least 32 bytes. +It also supplies `DIFFEX_MODEL_PROVIDER`, `DIFFEX_MODEL_ID`, and `DIFFEX_MODEL_API_KEY` through the sandbox provider's secret-injection mechanism. +The environment value is an opaque Daytona placeholder, the real provider key never enters the sandbox, and the placeholder is removed from the runtime process environment after the in-memory model client is configured. +Completed model usage is written to a durable sandbox outbox before it is reported to `DIFFEX_USAGE_ENDPOINT`. +Public WorkOS tokens and cloud attachment tickets are consumed by the control plane and are never passed to this process. diff --git a/packages/coding-agent-server/package.json b/packages/coding-agent-server/package.json new file mode 100644 index 000000000..a0dbf1e33 --- /dev/null +++ b/packages/coding-agent-server/package.json @@ -0,0 +1,31 @@ +{ + "name": "@earendil-works/diffex-coding-agent-server", + "version": "0.2.0", + "description": "Hosted Diffex protocol service backed by the Pi-derived coding-agent runtime", + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "bin": { "diffex-runtime-server": "./dist/cli.js" }, + "exports": { ".": { "types": "./dist/index.d.ts", "import": "./dist/index.js" } }, + "files": ["dist", "README.md"], + "scripts": { + "clean": "shx rm -rf dist", + "build": "tsgo -p tsconfig.build.json && shx chmod +x dist/cli.js", + "test": "vitest --run", + "typecheck": "tsgo -p tsconfig.test.json", + "prepublishOnly": "npm run clean && npm run build" + }, + "engines": { "node": ">=22.19.0" }, + "license": "Apache-2.0", + "dependencies": { + "@earendil-works/diffex-coding-agent": "^0.2.0", + "@earendil-works/diffex-protocol": "^0.2.0", + "@earendil-works/diffex-server": "^0.2.0" + }, + "devDependencies": { + "@earendil-works/diffex-client": "^0.2.0", + "shx": "0.4.0", + "vitest": "4.1.9", + "ws": "8.21.0" + } +} diff --git a/packages/coding-agent-server/src/cli.ts b/packages/coding-agent-server/src/cli.ts new file mode 100644 index 000000000..873225b18 --- /dev/null +++ b/packages/coding-agent-server/src/cli.ts @@ -0,0 +1,103 @@ +#!/usr/bin/env node + +import { timingSafeEqual } from "node:crypto"; +import { join } from "node:path"; +import { ModelRuntime } from "@earendil-works/diffex-coding-agent"; +import { createWebSocketServer } from "@earendil-works/diffex-server/websocket"; +import { createHostedCodingAgentService } from "./service.ts"; +import { createDurableUsageReporter } from "./usage-reporter.ts"; + +const sessionId = requiredEnvironment("DIFFEX_SESSION_ID"); +const workspace = requiredEnvironment("DIFFEX_WORKSPACE"); +const sharedSecret = requiredEnvironment("DIFFEX_RUNTIME_SHARED_SECRET"); +if (Buffer.byteLength(sharedSecret) < 32) { + throw new Error("DIFFEX_RUNTIME_SHARED_SECRET must contain at least 32 bytes"); +} +const port = integerEnvironment("PORT", 8787); +const host = process.env.HOST ?? "0.0.0.0"; +const path = process.env.DIFFEX_RUNTIME_PATH ?? "/runtime"; +const agentDir = process.env.DIFFEX_AGENT_DIR; +const modelProvider = requiredEnvironment("DIFFEX_MODEL_PROVIDER"); +const modelId = requiredEnvironment("DIFFEX_MODEL_ID"); +const modelApiKey = requiredEnvironment("DIFFEX_MODEL_API_KEY"); +const modelRuntime = await ModelRuntime.create({ + ...(agentDir + ? { + authPath: join(agentDir, "auth.json"), + modelsPath: join(agentDir, "models.json"), + } + : {}), + refreshOnCreate: false, +}); +await modelRuntime.setRuntimeApiKey(modelProvider, modelApiKey); +delete process.env.DIFFEX_MODEL_API_KEY; +delete process.env.DIFFEX_RUNTIME_SHARED_SECRET; +const runtimeStateDirectory = requiredEnvironment("DIFFEX_RUNTIME_STATE_DIR"); +const usageReporter = createDurableUsageReporter({ + endpoint: requiredEnvironment("DIFFEX_USAGE_ENDPOINT"), + sharedSecret, + outboxDirectory: join(runtimeStateDirectory, "usage-outbox"), + onError(error) { + process.stderr.write(`UsageReporterError: ${error.message}\n`); + }, +}); + +const service = await createHostedCodingAgentService({ + sessionId, + cwd: workspace, + agentDir, + sessionDir: process.env.DIFFEX_SESSION_DIR, + modelRuntime, + defaultModel: { provider: modelProvider, id: modelId }, + ownerUserId: requiredEnvironment("DIFFEX_OWNER_USER_ID"), + orgId: requiredEnvironment("DIFFEX_ORG_ID"), + usageReporter, +}); +const server = createWebSocketServer(service, { + host, + port, + path, + authorize(request) { + const authorization = request.headers.authorization; + if (!authorization?.startsWith("Bearer ")) return { authorized: false }; + return { authorized: secretsEqual(authorization.slice(7), sharedSecret) }; + }, + onError(error) { + process.stderr.write(`${error.name}: ${error.message}\n`); + }, +}); + +await server.start(); +process.stdout.write(`Diffex runtime server listening for session ${sessionId}\n`); + +let closing = false; +const close = async (): Promise => { + if (closing) return; + closing = true; + await server.close(); + await usageReporter.close(); +}; +process.once("SIGINT", () => void close().finally(() => process.exit(0))); +process.once("SIGTERM", () => void close().finally(() => process.exit(0))); + +function requiredEnvironment(name: string): string { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`${name} is required`); + return value; +} + +function integerEnvironment(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw === undefined) return fallback; + const value = Number(raw); + if (!Number.isSafeInteger(value) || value <= 0 || value > 65_535) { + throw new Error(`${name} must be an integer from 1 through 65535`); + } + return value; +} + +function secretsEqual(candidate: string, expected: string): boolean { + const candidateBytes = Buffer.from(candidate); + const expectedBytes = Buffer.from(expected); + return candidateBytes.byteLength === expectedBytes.byteLength && timingSafeEqual(candidateBytes, expectedBytes); +} diff --git a/packages/coding-agent-server/src/index.ts b/packages/coding-agent-server/src/index.ts new file mode 100644 index 000000000..7f3034aef --- /dev/null +++ b/packages/coding-agent-server/src/index.ts @@ -0,0 +1,9 @@ +export { + createHostedCodingAgentService, + type HostedCodingAgentServiceOptions, +} from "./service.ts"; +export { + createDurableUsageReporter, + type RuntimeUsageRecord, + type RuntimeUsageReporter, +} from "./usage-reporter.ts"; diff --git a/packages/coding-agent-server/src/service.ts b/packages/coding-agent-server/src/service.ts new file mode 100644 index 000000000..8f817f147 --- /dev/null +++ b/packages/coding-agent-server/src/service.ts @@ -0,0 +1,374 @@ +import { createHash, randomUUID } from "node:crypto"; +import { join } from "node:path"; +import { + type AgentSession, + type AgentSessionEvent, + createAgentSession, + getAgentDir, + ModelRuntime, + SessionManager, +} from "@earendil-works/diffex-coding-agent"; +import type { + ModelMetadata, + ModelRef, + SessionMetadata, + SessionPhase, + SessionSnapshot, + ThinkingLevel, + TranscriptItem, + UserTranscriptItem, +} from "@earendil-works/diffex-protocol"; +import { + type CreateSessionOptions, + DiffexServerError, + type DiffexServerService, + type DiffexSessionRuntime, + type DiffexSessionRuntimeEvent, + type PromptInput, + type SteerInput, + toProtocolAssistantMessage, + toProtocolModelMetadata, + toProtocolToolResultMessage, + toProtocolUserMessage, +} from "@earendil-works/diffex-server"; +import type { RuntimeUsageReporter } from "./usage-reporter.ts"; +export interface HostedCodingAgentServiceOptions { + sessionId: string; + cwd: string; + agentDir?: string; + sessionDir?: string; + modelRuntime?: ModelRuntime; + defaultModel?: ModelRef; + modelBaseUrl?: string; + ownerUserId?: string; + orgId?: string; + usageReporter?: Pick; +} + +/** Creates the one-cloud-session-per-sandbox service used by the hosted runtime. */ +export async function createHostedCodingAgentService( + options: HostedCodingAgentServiceOptions, +): Promise { + const agentDir = options.agentDir ?? getAgentDir(); + const sessionDir = options.sessionDir ?? join(agentDir, "hosted-sessions"); + const modelRuntime = + options.modelRuntime ?? + (await ModelRuntime.create({ + authPath: join(agentDir, "auth.json"), + modelsPath: join(agentDir, "models.json"), + })); + const existingSession = (await SessionManager.list(options.cwd, sessionDir)).find( + (session) => session.id === options.sessionId, + ); + const initialManager = existingSession + ? SessionManager.open(existingSession.path, sessionDir, options.cwd) + : SessionManager.create(options.cwd, sessionDir, { + id: options.sessionId, + persistImmediately: true, + }); + const sessionFile = initialManager.getSessionFile(); + const sessionHeader = initialManager.getHeader(); + if (!sessionFile || !sessionHeader) { + throw new Error("Hosted session persistence was not initialized"); + } + let acquired = false; + const createdAt = Date.parse(sessionHeader.timestamp); + + const openRuntime = async (): Promise => { + if (acquired) throw new DiffexServerError("session_locked", "Hosted session is already acquired"); + acquired = true; + try { + const manager = SessionManager.open(sessionFile, sessionDir, options.cwd); + const { session } = await createAgentSession({ + cwd: options.cwd, + agentDir, + modelRuntime, + sessionManager: manager, + ...(options.defaultModel + ? { + model: requiredModel(modelRuntime, options.defaultModel, options.modelBaseUrl), + } + : {}), + }); + return new CodingAgentProtocolRuntime( + session, + options.defaultModel, + options.modelBaseUrl, + options.usageReporter, + options.ownerUserId, + options.orgId, + () => { + acquired = false; + }, + ); + } catch (error) { + acquired = false; + throw error; + } + }; + + return { + listSessions(): Promise { + return Promise.resolve([ + { + id: options.sessionId, + createdAt, + updatedAt: Date.now(), + cwd: options.cwd, + }, + ]); + }, + listModels(): Promise { + const configured = options.defaultModel; + return Promise.resolve( + modelRuntime + .getAvailableSnapshot() + .filter((model) => !configured || (model.provider === configured.provider && model.id === configured.id)) + .map((model) => toProtocolModelMetadata(model, modelRuntime.hasConfiguredAuth(model.provider))), + ); + }, + createSession(_input: CreateSessionOptions): Promise { + return Promise.reject( + new DiffexServerError("not_implemented", "Hosted sessions are allocated by the Diffex cloud control plane"), + ); + }, + openSession(sessionId: string): Promise { + if (sessionId !== options.sessionId) { + return Promise.reject(new DiffexServerError("not_found", "Hosted session not found")); + } + return openRuntime(); + }, + }; +} + +function requiredModel(modelRuntime: ModelRuntime, model: ModelRef, baseUrl?: string) { + const resolved = modelRuntime.getModel(model.provider, model.id); + if (!resolved) { + throw new DiffexServerError("not_found", `Configured runtime model was not found: ${model.provider}/${model.id}`); + } + if (!modelRuntime.hasConfiguredAuth(model.provider)) { + throw new DiffexServerError("invalid_request", `Configured runtime model has no credentials: ${model.provider}`); + } + return baseUrl ? { ...resolved, baseUrl } : resolved; +} + +class CodingAgentProtocolRuntime implements DiffexSessionRuntime { + readonly #session: AgentSession; + readonly #configuredModel: ModelRef | undefined; + readonly #modelBaseUrl: string | undefined; + readonly #usageReporter: Pick | undefined; + readonly #ownerUserId: string | undefined; + readonly #orgId: string | undefined; + readonly #release: () => void; + readonly #listeners = new Set<(event: DiffexSessionRuntimeEvent) => void>(); + readonly #unsubscribe: () => void; + #revision = 0; + #disposed = false; + #phase: SessionPhase; + #queuedSteer: UserTranscriptItem[] = []; + + constructor( + session: AgentSession, + configuredModel: ModelRef | undefined, + modelBaseUrl: string | undefined, + usageReporter: Pick | undefined, + ownerUserId: string | undefined, + orgId: string | undefined, + release: () => void, + ) { + this.#session = session; + this.#configuredModel = configuredModel; + this.#modelBaseUrl = modelBaseUrl; + this.#usageReporter = usageReporter; + this.#ownerUserId = ownerUserId; + this.#orgId = orgId; + this.#release = release; + this.#phase = session.isStreaming ? "turn" : "idle"; + this.#queuedSteer = reconcileQueuedSteer([], session.getSteeringMessages()); + this.#unsubscribe = session.subscribe((event) => this.#handleEvent(event)); + } + + snapshot(): SessionSnapshot { + const header = this.#session.sessionManager.getHeader(); + const model = this.#session.model; + if (!header || !model) { + throw new Error("Hosted session is not initialized"); + } + const entries = this.#session.sessionManager.getBranch(); + const updatedAt = entries.at(-1)?.timestamp ?? header.timestamp; + const name = this.#session.sessionManager.getSessionName(); + return { + id: header.id, + ...(name ? { name } : {}), + cwd: this.#session.sessionManager.getCwd(), + createdAt: Date.parse(header.timestamp), + updatedAt: Date.parse(updatedAt), + phase: this.getPhase(), + model: { provider: model.provider, id: model.id }, + thinkingLevel: this.#session.thinkingLevel, + attached: true, + locked: true, + revision: this.#revision, + transcript: projectTranscript(this.#session), + queuedSteer: this.#queuedSteer, + queuedSteerCount: this.#queuedSteer.length, + }; + } + + getPhase(): SessionPhase { + return this.#phase; + } + + async prompt(input: PromptInput): Promise { + await this.#session.prompt(input.text); + } + + async steer(input: SteerInput): Promise { + await this.#session.steer(input.text); + } + + async abort(): Promise { + await this.#session.abort(); + } + + async setModel(model: ModelRef): Promise { + if ( + this.#configuredModel && + (model.provider !== this.#configuredModel.provider || model.id !== this.#configuredModel.id) + ) { + throw new DiffexServerError("not_found", "Model is not available in this hosted session"); + } + const resolved = requiredModel(this.#session.modelRuntime, model, this.#modelBaseUrl); + await this.#session.setModel(resolved); + } + + async setThinking(thinkingLevel: ThinkingLevel): Promise { + this.#session.setThinkingLevel(thinkingLevel); + } + + subscribe(listener: (event: DiffexSessionRuntimeEvent) => void): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + async dispose(): Promise { + if (this.#disposed) return; + this.#disposed = true; + this.#unsubscribe(); + this.#listeners.clear(); + try { + await this.#session.dispose(); + } finally { + this.#release(); + } + } + + #handleEvent(event: AgentSessionEvent): void { + if (event.type === "message_end" && event.message.role === "assistant" && this.#usageReporter) { + const message = event.message; + this.#usageReporter.record({ + eventId: assistantUsageEventId(this.#session.sessionManager.getSessionId(), message), + sessionId: this.#session.sessionManager.getSessionId(), + ownerUserId: requiredUsageIdentity("ownerUserId", this.#ownerUserId), + orgId: requiredUsageIdentity("orgId", this.#orgId), + provider: message.provider, + model: message.model, + usage: { + input: message.usage.input, + output: message.usage.output, + cacheRead: message.usage.cacheRead, + cacheWrite: message.usage.cacheWrite, + }, + occurredAt: message.timestamp, + }); + } + this.#phase = phaseAfterEvent(this.#phase, event, this.#session.isStreaming); + if (event.type === "queue_update") { + this.#queuedSteer = reconcileQueuedSteer(this.#queuedSteer, event.steering); + } + this.#revision += 1; + for (const listener of this.#listeners) listener({ type: "snapshot" }); + } +} + +function requiredUsageIdentity(name: string, value: string | undefined): string { + if (!value) throw new Error(`Hosted runtime ${name} is not configured`); + return value; +} + +function assistantUsageEventId( + sessionId: string, + message: Extract["message"], +): string { + return createHash("sha256").update(sessionId).update("\0").update(JSON.stringify(message)).digest("hex"); +} + +function projectTranscript(session: AgentSession): TranscriptItem[] { + const result: TranscriptItem[] = []; + const toolCalls = new Map(); + for (const entry of session.sessionManager.getBranch()) { + if (entry.type !== "message") continue; + const message = entry.message; + if (message.role === "user") { + result.push(toProtocolUserMessage(message, { id: entry.id })); + } else if (message.role === "assistant") { + for (const content of message.content) { + if (content.type === "toolCall") toolCalls.set(content.id, content); + } + result.push(toProtocolAssistantMessage(message, { id: entry.id })); + } else if (message.role === "toolResult") { + const call = toolCalls.get(message.toolCallId); + if (!call) { + throw new Error(`Tool result ${message.toolCallId} has no matching tool call in the active branch`); + } + result.push(toProtocolToolResultMessage(message, { id: entry.id, call })); + } + } + return result; +} + +type ToolCall = Parameters[1]["call"]; + +function reconcileQueuedSteer( + previous: readonly UserTranscriptItem[], + messages: readonly string[], +): UserTranscriptItem[] { + const remaining = [...previous]; + return messages.map((text) => { + const index = remaining.findIndex( + (item) => item.content.length === 1 && item.content[0]?.type === "text" && item.content[0].text === text, + ); + if (index >= 0) { + return remaining.splice(index, 1)[0]!; + } + return { + id: `queued-${randomUUID()}`, + role: "user", + content: [{ type: "text", text }], + timestamp: Date.now(), + }; + }); +} + +function phaseAfterEvent(current: SessionPhase, event: AgentSessionEvent, isStreaming: boolean): SessionPhase { + switch (event.type) { + case "compaction_start": + return "compaction"; + case "summarization_retry_attempt_start": + return event.source === "branchSummary" ? "branch_summary" : "compaction"; + case "auto_retry_start": + case "summarization_retry_scheduled": + return "retry"; + case "agent_start": + case "turn_start": + return "turn"; + case "agent_settled": + case "summarization_retry_finished": + return isStreaming ? "turn" : "idle"; + case "compaction_end": + case "auto_retry_end": + return isStreaming ? "turn" : "idle"; + default: + return current; + } +} diff --git a/packages/coding-agent-server/src/usage-reporter.ts b/packages/coding-agent-server/src/usage-reporter.ts new file mode 100644 index 000000000..8bb2c7a5c --- /dev/null +++ b/packages/coding-agent-server/src/usage-reporter.ts @@ -0,0 +1,164 @@ +import { existsSync, mkdirSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; + +export interface RuntimeUsageRecord { + eventId: string; + sessionId: string; + ownerUserId: string; + orgId: string; + provider: string; + model: string; + usage: { + input: number; + output: number; + cacheRead: number; + cacheWrite: number; + }; + occurredAt: number; +} + +export interface RuntimeUsageReporter { + record(event: RuntimeUsageRecord): void; + flush(): Promise; + close(): Promise; +} + +export function createDurableUsageReporter(input: { + endpoint: string; + sharedSecret: string; + outboxDirectory: string; + retryIntervalMs?: number; + fetch?: typeof fetch; + onError?: (error: Error) => void; +}): RuntimeUsageReporter { + const endpoint = new URL(input.endpoint); + if (endpoint.protocol !== "https:" && endpoint.hostname !== "localhost" && endpoint.hostname !== "127.0.0.1") { + throw new TypeError("Runtime usage endpoint must use HTTPS"); + } + if (Buffer.byteLength(input.sharedSecret) < 32) { + throw new TypeError("Runtime usage shared secret must contain at least 32 bytes"); + } + const retryIntervalMs = input.retryIntervalMs ?? 10_000; + if (!Number.isSafeInteger(retryIntervalMs) || retryIntervalMs <= 0) { + throw new TypeError("Runtime usage retry interval must be positive"); + } + mkdirSync(input.outboxDirectory, { recursive: true, mode: 0o700 }); + const fetchImpl = input.fetch ?? fetch; + let closed = false; + let flushPromise: Promise | undefined; + let flushRequested = false; + + const flush = (): Promise => { + flushRequested = true; + if (flushPromise) return flushPromise; + flushPromise = (async () => { + do { + flushRequested = false; + await flushOutbox({ + endpoint, + sharedSecret: input.sharedSecret, + outboxDirectory: input.outboxDirectory, + fetchImpl, + onError: input.onError, + }); + } while (flushRequested); + })().finally(() => { + flushPromise = undefined; + }); + return flushPromise; + }; + const timer = setInterval(() => { + if (!closed) void flush(); + }, retryIntervalMs); + timer.unref(); + void flush(); + + return { + record(event) { + if (closed) throw new Error("Runtime usage reporter is closed"); + validateUsageRecord(event); + const file = usageFile(input.outboxDirectory, event.eventId); + if (!existsSync(file)) { + writeFileSync(file, `${JSON.stringify(event)}\n`, { + encoding: "utf8", + mode: 0o600, + flag: "wx", + }); + } + void flush(); + }, + flush, + async close() { + if (closed) return; + closed = true; + clearInterval(timer); + await flush(); + }, + }; +} + +async function flushOutbox(input: { + endpoint: URL; + sharedSecret: string; + outboxDirectory: string; + fetchImpl: typeof fetch; + onError?: (error: Error) => void; +}): Promise { + const files = readdirSync(input.outboxDirectory) + .filter((file) => file.endsWith(".json")) + .sort(); + for (const file of files) { + const path = join(input.outboxDirectory, file); + try { + const event = JSON.parse(readFileSync(path, "utf8")) as RuntimeUsageRecord; + validateUsageRecord(event); + const response = await input.fetchImpl(input.endpoint, { + method: "POST", + headers: { + authorization: `Bearer ${input.sharedSecret}`, + "content-type": "application/json", + }, + body: JSON.stringify(event), + signal: AbortSignal.timeout(5_000), + }); + if (!response.ok) { + throw new Error(`Runtime usage endpoint returned ${String(response.status)}`); + } + rmSync(path); + } catch (error) { + reportError(input.onError, error); + return; + } + } +} + +function usageFile(directory: string, eventId: string): string { + if (!/^[A-Za-z0-9][A-Za-z0-9._-]*$/u.test(eventId)) { + throw new TypeError("Runtime usage eventId is not filesystem-safe"); + } + return join(directory, `${eventId}.json`); +} + +function validateUsageRecord(event: RuntimeUsageRecord): void { + for (const value of [event.eventId, event.sessionId, event.ownerUserId, event.orgId, event.provider, event.model]) { + if (typeof value !== "string" || value.length === 0) { + throw new TypeError("Runtime usage identity fields must not be empty"); + } + } + for (const value of Object.values(event.usage)) { + if (!Number.isSafeInteger(value) || value < 0) { + throw new TypeError("Runtime usage values must be non-negative safe integers"); + } + } + if (!Number.isSafeInteger(event.occurredAt) || event.occurredAt < 0) { + throw new TypeError("Runtime usage occurredAt must be a timestamp"); + } +} + +function reportError(observer: ((error: Error) => void) | undefined, error: unknown): void { + try { + observer?.(error instanceof Error ? error : new Error(String(error))); + } catch { + // Usage diagnostics never alter the durable retry state. + } +} diff --git a/packages/coding-agent-server/test/usage-reporter.test.ts b/packages/coding-agent-server/test/usage-reporter.test.ts new file mode 100644 index 000000000..c4cee2cb3 --- /dev/null +++ b/packages/coding-agent-server/test/usage-reporter.test.ts @@ -0,0 +1,78 @@ +import { existsSync, mkdtempSync, readdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createDurableUsageReporter, type RuntimeUsageRecord } from "../src/usage-reporter.ts"; + +const directories: string[] = []; +const EVENT: RuntimeUsageRecord = { + eventId: "assistant-entry-1", + sessionId: "session-1", + ownerUserId: "user-1", + orgId: "org-1", + provider: "anthropic", + model: "claude-opus-4-8", + usage: { input: 100, output: 20, cacheRead: 10, cacheWrite: 5 }, + occurredAt: 1_000, +}; + +afterEach(() => { + for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }); +}); + +describe("durable runtime usage reporter", () => { + it("persists before delivery and removes the event only after acknowledgement", async () => { + const outboxDirectory = temporaryDirectory(); + let resolveFetch!: (response: Response) => void; + const fetch = vi.fn( + () => + new Promise((resolve) => { + resolveFetch = resolve; + }), + ); + const reporter = createDurableUsageReporter({ + endpoint: "https://api.diffex.test/v1/runtime/sessions/session-1/usage", + sharedSecret: "runtime-usage-shared-secret-for-tests", + outboxDirectory, + fetch, + }); + + reporter.record(EVENT); + expect(readdirSync(outboxDirectory)).toEqual(["assistant-entry-1.json"]); + await vi.waitFor(() => expect(fetch).toHaveBeenCalledOnce()); + resolveFetch(new Response(null, { status: 204 })); + await reporter.flush(); + expect(readdirSync(outboxDirectory)).toEqual([]); + await reporter.close(); + }); + + it("keeps failed events and retries them after process restart", async () => { + const outboxDirectory = temporaryDirectory(); + const failed = createDurableUsageReporter({ + endpoint: "https://api.diffex.test/v1/runtime/sessions/session-1/usage", + sharedSecret: "runtime-usage-shared-secret-for-tests", + outboxDirectory, + fetch: vi.fn().mockResolvedValue(new Response(null, { status: 503 })), + }); + failed.record(EVENT); + await failed.flush(); + expect(existsSync(join(outboxDirectory, "assistant-entry-1.json"))).toBe(true); + await failed.close(); + + const recovered = createDurableUsageReporter({ + endpoint: "https://api.diffex.test/v1/runtime/sessions/session-1/usage", + sharedSecret: "runtime-usage-shared-secret-for-tests", + outboxDirectory, + fetch: vi.fn().mockResolvedValue(new Response(null, { status: 204 })), + }); + await recovered.flush(); + expect(readdirSync(outboxDirectory)).toEqual([]); + await recovered.close(); + }); +}); + +function temporaryDirectory(): string { + const directory = mkdtempSync(join(tmpdir(), "diffex-usage-outbox-")); + directories.push(directory); + return directory; +} diff --git a/packages/coding-agent-server/test/websocket-flow.test.ts b/packages/coding-agent-server/test/websocket-flow.test.ts new file mode 100644 index 000000000..54ee4bd6d --- /dev/null +++ b/packages/coding-agent-server/test/websocket-flow.test.ts @@ -0,0 +1,155 @@ +import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { DiffexClient } from "@earendil-works/diffex-client"; +import { createWebSocketTransportFactory } from "@earendil-works/diffex-client/websocket"; +import { ModelRuntime } from "@earendil-works/diffex-coding-agent"; +import { DiffexServer } from "@earendil-works/diffex-server"; +import { TestServerService } from "@earendil-works/diffex-server/testing"; +import { createWebSocketListener } from "@earendil-works/diffex-server/websocket"; +import { afterEach, describe, expect, it } from "vitest"; +import { WebSocket } from "ws"; +import { createHostedCodingAgentService } from "../src/service.ts"; + +const temporaryDirectories: string[] = []; + +afterEach(() => { + for (const directory of temporaryDirectories.splice(0)) { + rmSync(directory, { recursive: true, force: true }); + } +}); + +describe("hosted Diffex runtime", () => { + it("carries the native protocol over an authorized WebSocket", async () => { + const service = new TestServerService(); + service.seed("cloud-session-1", "Cloud session"); + const listener = createWebSocketListener({ + port: 0, + path: "/runtime", + authorize: (request) => ({ authorized: request.headers.authorization === "Bearer internal-secret" }), + }); + const server = new DiffexServer(service, { listeners: [listener] }); + await server.start(); + + try { + const address = server.addresses[0]; + if (!address) throw new Error("Missing WebSocket server address"); + await expect( + DiffexClient.connect({ + transportFactory: authorizedTransport(address, "wrong-secret"), + }), + ).rejects.toThrow(); + const client = await DiffexClient.connect({ + transportFactory: authorizedTransport(address, "internal-secret"), + }); + + try { + expect((await client.listSessions()).map((session) => session.id)).toEqual(["cloud-session-1"]); + const session = await client.attachSession("cloud-session-1"); + expect(session.snapshot?.id).toBe("cloud-session-1"); + const prompt = session.prompt("hello cloud"); + const runtime = service.latestRuntime("cloud-session-1"); + await waitFor(() => runtime.getPhase() === "turn"); + runtime.emitProgress({ + type: "assistant_delta", + messageId: "assistant-streaming", + contentIndex: 0, + kind: "text", + delta: "reply:", + }); + runtime.finishPrompt(); + await expect(prompt).resolves.toMatchObject({ + phase: "idle", + transcript: [ + { role: "user", content: [{ type: "text", text: "hello cloud" }] }, + { role: "assistant", content: [{ type: "text", text: "reply:hello cloud" }] }, + ], + }); + await session.dispose(); + } finally { + await client.dispose(); + } + + const reconnected = await DiffexClient.connect({ + transportFactory: authorizedTransport(address, "internal-secret"), + }); + try { + const session = await reconnected.attachSession("cloud-session-1"); + expect(session.snapshot?.transcript).toHaveLength(2); + await session.dispose(); + } finally { + await reconnected.dispose(); + } + } finally { + await server.close(); + } + }); + + it("persists an empty cloud session before the first model response", async () => { + const root = temporaryDirectory(); + const cwd = join(root, "workspace"); + const agentDir = join(root, "agent"); + const sessionDir = join(root, "sessions"); + mkdirSync(cwd, { recursive: true }); + mkdirSync(agentDir, { recursive: true }); + const modelRuntime = await ModelRuntime.create({ + authPath: join(agentDir, "auth.json"), + modelsPath: join(agentDir, "models.json"), + refreshOnCreate: false, + }); + await modelRuntime.setRuntimeApiKey("anthropic", "test-api-key"); + + const first = await createHostedCodingAgentService({ + sessionId: "cloud-session-2", + cwd, + agentDir, + sessionDir, + modelRuntime, + }); + const firstMetadata = (await first.listSessions())[0]; + expect(firstMetadata?.id).toBe("cloud-session-2"); + + const reopened = await createHostedCodingAgentService({ + sessionId: "cloud-session-2", + cwd, + agentDir, + sessionDir, + modelRuntime, + }); + const reopenedMetadata = (await reopened.listSessions())[0]; + expect(reopenedMetadata?.createdAt).toBe(firstMetadata?.createdAt); + const runtime = await reopened.openSession("cloud-session-2"); + expect(runtime.snapshot()).toMatchObject({ + id: "cloud-session-2", + cwd, + phase: "idle", + transcript: [], + }); + await runtime.dispose(); + }); +}); + +function temporaryDirectory(): string { + const directory = mkdtempSync(join(tmpdir(), "diffex-hosted-runtime-")); + temporaryDirectories.push(directory); + return directory; +} + +function authorizedTransport(address: string, secret: string) { + return createWebSocketTransportFactory({ + url: address, + webSocket: class extends WebSocket { + constructor(url: string | URL, protocols?: string | string[]) { + super(url, protocols, { headers: { authorization: `Bearer ${secret}` } }); + } + } as never, + }); +} + +async function waitFor(predicate: () => boolean): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 1)); + } + throw new Error("Timed out waiting for runtime state"); +} diff --git a/packages/coding-agent-server/tsconfig.build.json b/packages/coding-agent-server/tsconfig.build.json new file mode 100644 index 000000000..eac5a85dc --- /dev/null +++ b/packages/coding-agent-server/tsconfig.build.json @@ -0,0 +1,15 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src", + "paths": { + "@earendil-works/diffex-coding-agent": ["../coding-agent/dist/index.d.ts"], + "@earendil-works/diffex-protocol": ["../protocol/dist/index.d.ts"], + "@earendil-works/diffex-server": ["../server/dist/index.d.ts"], + "@earendil-works/diffex-server/websocket": ["../server/dist/transports/websocket/index.d.ts"] + } + }, + "include": ["src/**/*.ts"], + "exclude": ["node_modules", "dist", "test"] +} diff --git a/packages/coding-agent-server/tsconfig.test.json b/packages/coding-agent-server/tsconfig.test.json new file mode 100644 index 000000000..ccf6137dd --- /dev/null +++ b/packages/coding-agent-server/tsconfig.test.json @@ -0,0 +1,5 @@ +{ + "extends": "./tsconfig.build.json", + "compilerOptions": { "noEmit": true, "rootDir": "." }, + "include": ["src/**/*.ts", "test/**/*.ts"] +} diff --git a/packages/coding-agent-server/vitest.config.ts b/packages/coding-agent-server/vitest.config.ts new file mode 100644 index 000000000..5539adc2a --- /dev/null +++ b/packages/coding-agent-server/vitest.config.ts @@ -0,0 +1,3 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ test: { include: ["test/**/*.test.ts"] } }); diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index 8f4a86ff3..e75f1f957 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -41,6 +41,8 @@ export interface SessionHeader { export interface NewSessionOptions { id?: string; parentSession?: string; + /** Persist the session header immediately, even before the first assistant response. */ + persistImmediately?: boolean; } export interface SessionEntryBase { @@ -967,6 +969,10 @@ export class SessionManager { if (this.persist) { const fileTimestamp = timestamp.replace(/[:.]/g, "-"); this.sessionFile = join(this.getSessionDir(), `${fileTimestamp}_${this.sessionId}.jsonl`); + if (options?.persistImmediately) { + writeFileSync(this.sessionFile, `${JSON.stringify(header)}\n`, { flag: "wx" }); + this.flushed = true; + } } return this.sessionFile; } diff --git a/packages/server/README.md b/packages/server/README.md index e2f56f501..921a5e620 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -48,3 +48,9 @@ Custom transports can use `@earendil-works/diffex-server/testing` for determinis `@earendil-works/diffex-ai` domain objects and `@earendil-works/diffex-protocol` wire DTOs remain independent. This package owns their boundary and exports `toProtocolModelMetadata()`, `toProtocolAssistantMessage()`, `toProtocolUserMessage()`, and `toProtocolToolResultMessage()`. The adapters reject invalid tool inputs, identifiers, timestamps, and mismatched tool results; `toProtocolToolResultMessage()` requires the original `ToolCall` so it can verify the association and convert its arguments itself. Diagnostic details are explicitly sanitized. Closed `diffex-ai` unions are mapped exhaustively, and compile-time field manifests enumerate current `diffex-ai` properties so additions require an explicit review. The protocol mirrors `diffex-ai` vocabulary such as `toolCall` and `toolUse` where the semantics are identical. Protocol schemas enforce consistent lifecycle states, and tests encode adapter output through the runtime schemas so incompatible changes fail in the bridging package. + +## WebSocket listener + +Hosted runtimes can compose an authenticated WebSocket listener from the explicit `@earendil-works/diffex-server/websocket` subpath. +The authorization callback runs before the protocol connection is accepted and must validate the cloud-issued, session-scoped credential. +Text messages are rejected, compression is disabled, and inbound and outbound byte limits are enforced independently from protocol frame validation. diff --git a/packages/server/package.json b/packages/server/package.json index 046964448..921988448 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -17,6 +17,10 @@ "./unix": { "types": "./dist/transports/unix/index.d.ts", "import": "./dist/transports/unix/index.js" + }, + "./websocket": { + "types": "./dist/transports/websocket/index.d.ts", + "import": "./dist/transports/websocket/index.js" } }, "files": [ @@ -48,9 +52,11 @@ }, "dependencies": { "@earendil-works/diffex-ai": "^0.2.0", - "@earendil-works/diffex-protocol": "^0.2.0" + "@earendil-works/diffex-protocol": "^0.2.0", + "ws": "8.21.0" }, "devDependencies": { + "@types/ws": "8.18.1", "shx": "0.4.0", "vitest": "4.1.9" } diff --git a/packages/server/src/transports/websocket/index.ts b/packages/server/src/transports/websocket/index.ts new file mode 100644 index 000000000..e6a95d6fa --- /dev/null +++ b/packages/server/src/transports/websocket/index.ts @@ -0,0 +1,7 @@ +export { createWebSocketListener, WebSocketByteConnection } from "./listener.ts"; +export { createWebSocketServer } from "./preset.ts"; +export type { + WebSocketAuthorization, + WebSocketListenerOptions, + WebSocketServerOptions, +} from "./types.ts"; diff --git a/packages/server/src/transports/websocket/listener.ts b/packages/server/src/transports/websocket/listener.ts new file mode 100644 index 000000000..b5a52c611 --- /dev/null +++ b/packages/server/src/transports/websocket/listener.ts @@ -0,0 +1,254 @@ +import { createServer, type IncomingMessage, type Server } from "node:http"; +import type { Duplex } from "node:stream"; +import { DEFAULT_MAX_FRAME_LENGTH } from "@earendil-works/diffex-protocol"; +import { type RawData, WebSocket, WebSocketServer } from "ws"; +import type { ByteConnection, ByteConnectionAcceptor } from "../../connection.ts"; +import type { DiffexServerListener } from "../../listener.ts"; +import type { WebSocketListenerOptions } from "./types.ts"; + +const DEFAULT_GRACEFUL_CLOSE_TIMEOUT_MS = 5_000; + +class DiffexWebSocketListener implements DiffexServerListener { + readonly #options: Required> & WebSocketListenerOptions; + readonly #connections = new Set(); + #server?: Server; + #webSockets?: WebSocketServer; + #accept?: ByteConnectionAcceptor; + #address?: string; + #closing = false; + + constructor(options: WebSocketListenerOptions) { + this.#options = resolveOptions(options); + } + + get address(): string | undefined { + return this.#address; + } + + async start(accept: ByteConnectionAcceptor): Promise { + if (this.#server) throw new Error("WebSocket listener is already started"); + if (this.#closing) throw new Error("WebSocket listener is closing or closed"); + this.#accept = accept; + + const webSockets = new WebSocketServer({ + noServer: true, + maxPayload: this.#options.maxFrameLength ?? DEFAULT_MAX_FRAME_LENGTH, + perMessageDeflate: false, + }); + const server = createServer((_request, response) => { + response.writeHead(404, { "content-type": "text/plain", "cache-control": "no-store" }); + response.end("Not found"); + }); + server.on("upgrade", (request, socket, head) => { + void this.#upgrade(webSockets, request, socket, head); + }); + server.on("error", (error) => this.#reportError(error)); + webSockets.on("connection", (socket) => this.#acceptSocket(socket)); + webSockets.on("error", (error) => this.#reportError(error)); + + this.#server = server; + this.#webSockets = webSockets; + try { + await new Promise((resolve, reject) => { + const onError = (error: Error): void => { + server.off("listening", onListening); + reject(error); + }; + const onListening = (): void => { + server.off("error", onError); + resolve(); + }; + server.once("error", onError); + server.once("listening", onListening); + server.listen(this.#options.port, this.#options.host); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("WebSocket listener has no TCP address"); + this.#address = `ws://${formatHost(address.address)}:${address.port}${this.#options.path}`; + } catch (error) { + await this.close(); + throw error; + } + } + + async close(): Promise { + if (this.#closing) return; + this.#closing = true; + this.#address = undefined; + await Promise.all([...this.#connections].map((connection) => connection.close())); + await Promise.all([closeWebSocketServer(this.#webSockets), closeHttpServer(this.#server)]); + this.#connections.clear(); + this.#webSockets = undefined; + this.#server = undefined; + } + + async #upgrade(webSockets: WebSocketServer, request: IncomingMessage, socket: Duplex, head: Buffer): Promise { + try { + const requestUrl = new URL(request.url ?? "/", "http://runtime.internal"); + if (requestUrl.pathname !== this.#options.path || this.#closing) { + rejectUpgrade(socket, 404, "Not Found"); + return; + } + const authorization = await this.#options.authorize(request); + if (!authorization.authorized || this.#closing) { + rejectUpgrade(socket, 401, "Unauthorized"); + return; + } + webSockets.handleUpgrade(request, socket, head, (client) => { + webSockets.emit("connection", client, request); + }); + } catch (error) { + this.#reportError(error); + rejectUpgrade(socket, 401, "Unauthorized"); + } + } + + #acceptSocket(socket: WebSocket): void { + if (this.#closing || !this.#accept) { + socket.close(1012, "Server shutting down"); + return; + } + const connection = new WebSocketByteConnection( + socket, + this.#options.maxPendingBytes ?? DEFAULT_MAX_FRAME_LENGTH * 4, + this.#options.gracefulCloseTimeoutMs ?? DEFAULT_GRACEFUL_CLOSE_TIMEOUT_MS, + ); + this.#connections.add(connection); + const handler = this.#accept(connection); + socket.on("message", (data, isBinary) => { + if (!isBinary) { + handler.onError(new Error("Diffex WebSocket protocol requires binary messages")); + socket.close(1003, "Binary messages required"); + return; + } + handler.onData(rawDataBytes(data)); + }); + socket.once("error", (error) => handler.onError(error)); + socket.once("close", () => { + connection.markClosed(); + this.#connections.delete(connection); + handler.onClose(); + }); + } + + #reportError(error: unknown): void { + try { + this.#options.onError?.(error instanceof Error ? error : new Error(String(error))); + } catch { + // Error observers cannot affect listener state. + } + } +} + +export function createWebSocketListener(options: WebSocketListenerOptions): DiffexServerListener { + return new DiffexWebSocketListener(options); +} + +export class WebSocketByteConnection implements ByteConnection { + readonly #socket: WebSocket; + readonly #maxPendingBytes: number; + readonly #gracefulCloseTimeoutMs: number; + #closed = false; + #pendingBytes = 0; + #writeTail: Promise = Promise.resolve(); + + constructor(socket: WebSocket, maxPendingBytes: number, gracefulCloseTimeoutMs: number) { + this.#socket = socket; + this.#maxPendingBytes = maxPendingBytes; + this.#gracefulCloseTimeoutMs = gracefulCloseTimeoutMs; + } + + get closed(): boolean { + return this.#closed; + } + + send(chunk: Uint8Array): Promise { + if (!(chunk instanceof Uint8Array)) return Promise.reject(new TypeError("WebSocket chunks must be Uint8Array")); + if (this.#closed || this.#socket.readyState !== WebSocket.OPEN) { + return Promise.reject(new Error("WebSocket connection is closed")); + } + if (this.#pendingBytes + chunk.byteLength > this.#maxPendingBytes) { + return Promise.reject(new Error("WebSocket connection exceeded its pending byte limit")); + } + const bytes = chunk.slice(); + this.#pendingBytes += bytes.byteLength; + const send = this.#writeTail.then( + () => + new Promise((resolve, reject) => { + if (this.#closed || this.#socket.readyState !== WebSocket.OPEN) { + reject(new Error("WebSocket connection is closed")); + return; + } + this.#socket.send(bytes, { binary: true }, (error) => (error ? reject(error) : resolve())); + }), + ); + const tracked = send.finally(() => { + this.#pendingBytes -= bytes.byteLength; + }); + this.#writeTail = tracked.catch(() => {}); + return tracked; + } + + async close(finalChunk?: Uint8Array): Promise { + if (this.#closed) return; + if (finalChunk) await this.send(finalChunk); + this.#closed = true; + if (this.#socket.readyState === WebSocket.CLOSED) return; + await new Promise((resolve) => { + const timer = setTimeout(() => { + this.#socket.terminate(); + resolve(); + }, this.#gracefulCloseTimeoutMs); + timer.unref(); + this.#socket.once("close", () => { + clearTimeout(timer); + resolve(); + }); + this.#socket.close(1000, "Server closed"); + }); + } + + markClosed(): void { + this.#closed = true; + } +} + +function resolveOptions(options: WebSocketListenerOptions) { + if (!Number.isSafeInteger(options.port) || options.port < 0 || options.port > 65_535) { + throw new TypeError("WebSocket listener port must be an integer from 0 through 65535"); + } + const path = options.path ?? "/runtime"; + if (!path.startsWith("/") || path.includes("?") || path.includes("#")) { + throw new TypeError("WebSocket listener path must be an absolute path without query or fragment"); + } + const maxPendingBytes = options.maxPendingBytes ?? DEFAULT_MAX_FRAME_LENGTH * 4; + if (!Number.isSafeInteger(maxPendingBytes) || maxPendingBytes <= 0) { + throw new TypeError("WebSocket listener maxPendingBytes must be a positive safe integer"); + } + return { ...options, host: options.host ?? "127.0.0.1", path, maxPendingBytes }; +} + +function rawDataBytes(data: RawData): Uint8Array { + if (data instanceof ArrayBuffer) return new Uint8Array(data); + if (Array.isArray(data)) return new Uint8Array(Buffer.concat(data)); + return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); +} + +function rejectUpgrade(socket: Duplex, status: number, reason: string): void { + if (socket.destroyed) return; + socket.end(`HTTP/1.1 ${status} ${reason}\r\nConnection: close\r\nCache-Control: no-store\r\n\r\n`); +} + +function formatHost(host: string): string { + return host.includes(":") ? `[${host}]` : host; +} + +async function closeWebSocketServer(server: WebSocketServer | undefined): Promise { + if (!server) return; + await new Promise((resolve) => server.close(() => resolve())); +} + +async function closeHttpServer(server: Server | undefined): Promise { + if (!server || !server.listening) return; + await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))); +} diff --git a/packages/server/src/transports/websocket/preset.ts b/packages/server/src/transports/websocket/preset.ts new file mode 100644 index 000000000..13bc40ac3 --- /dev/null +++ b/packages/server/src/transports/websocket/preset.ts @@ -0,0 +1,15 @@ +import { DiffexServer } from "../../server.ts"; +import type { DiffexServerService } from "../../types.ts"; +import { createWebSocketListener } from "./listener.ts"; +import type { WebSocketServerOptions } from "./types.ts"; + +export function createWebSocketServer(service: DiffexServerService, options: WebSocketServerOptions): DiffexServer { + const listener = createWebSocketListener(options); + return new DiffexServer(service, { + listeners: [listener], + maxFrameLength: options.maxFrameLength, + handshakeTimeoutMs: options.handshakeTimeoutMs, + serverId: options.serverId, + onError: options.onError, + }); +} diff --git a/packages/server/src/transports/websocket/types.ts b/packages/server/src/transports/websocket/types.ts new file mode 100644 index 000000000..98a613afa --- /dev/null +++ b/packages/server/src/transports/websocket/types.ts @@ -0,0 +1,19 @@ +import type { IncomingMessage } from "node:http"; +import type { DiffexServerOptions } from "../../types.ts"; + +export interface WebSocketAuthorization { + authorized: boolean; +} + +export interface WebSocketListenerOptions { + host?: string; + port: number; + path?: string; + authorize(request: IncomingMessage): WebSocketAuthorization | Promise; + maxPendingBytes?: number; + gracefulCloseTimeoutMs?: number; + maxFrameLength?: number; + onError?: (error: Error) => void; +} + +export interface WebSocketServerOptions extends Omit, WebSocketListenerOptions {}