From 4abc08a8b3db341b8a5b7fd78f2f4aee0780ad2e Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Fri, 31 Jul 2026 21:02:44 +0300 Subject: [PATCH] feat(coding-agent): add server session backend Add durable coding-agent session runtimes with locking, protocol projection, and JSONL persistence. Remove the obsolete legacy server implementation and dependency cycle. --- package-lock.json | 5 +- package.json | 4 +- packages/agent/src/harness/agent-harness.ts | 12 +- .../agent/test/harness/agent-harness.test.ts | 27 ++ .../install-lock/package-lock.json | 13 + packages/coding-agent/npm-shrinkwrap.json | 13 + packages/coding-agent/package.json | 7 +- packages/coding-agent/src/server/backend.ts | 211 +++++++++ .../coding-agent/src/server/create-harness.ts | 101 ++++ packages/coding-agent/src/server/errors.ts | 17 + packages/coding-agent/src/server/index.ts | 4 + .../coding-agent/src/server/model-catalog.ts | 69 +++ .../coding-agent/src/server/session-lock.ts | 111 +++++ .../src/server/session-persistence.ts | 140 ++++++ .../src/server/session-runtime.ts | 275 +++++++++++ .../src/server/transcript/live.ts | 304 ++++++++++++ .../src/server/transcript/projection.ts | 102 ++++ .../coding-agent/test/server/backend.test.ts | 153 ++++++ packages/coding-agent/test/server/fixture.ts | 70 +++ .../test/server/persistence.test.ts | 158 +++++++ .../coding-agent/test/server/runtime.test.ts | 244 ++++++++++ packages/server/README.md | 16 +- packages/server/package.json | 11 +- packages/server/src/index.ts | 1 - packages/server/src/legacy/cli.ts | 161 ------- packages/server/src/legacy/config.ts | 69 --- packages/server/src/legacy/handler.ts | 161 ------- packages/server/src/legacy/index.ts | 11 - packages/server/src/legacy/ipc/client.ts | 63 --- packages/server/src/legacy/ipc/protocol.ts | 142 ------ packages/server/src/legacy/ipc/server.ts | 209 --------- packages/server/src/legacy/radius.ts | 440 ------------------ packages/server/src/legacy/rpc-process.ts | 201 -------- packages/server/src/legacy/serve.ts | 77 --- packages/server/src/legacy/storage.ts | 70 --- packages/server/src/legacy/supervisor.ts | 354 -------------- packages/server/src/legacy/types.ts | 25 - 37 files changed, 2037 insertions(+), 2014 deletions(-) create mode 100644 packages/coding-agent/src/server/backend.ts create mode 100644 packages/coding-agent/src/server/create-harness.ts create mode 100644 packages/coding-agent/src/server/errors.ts create mode 100644 packages/coding-agent/src/server/index.ts create mode 100644 packages/coding-agent/src/server/model-catalog.ts create mode 100644 packages/coding-agent/src/server/session-lock.ts create mode 100644 packages/coding-agent/src/server/session-persistence.ts create mode 100644 packages/coding-agent/src/server/session-runtime.ts create mode 100644 packages/coding-agent/src/server/transcript/live.ts create mode 100644 packages/coding-agent/src/server/transcript/projection.ts create mode 100644 packages/coding-agent/test/server/backend.test.ts create mode 100644 packages/coding-agent/test/server/fixture.ts create mode 100644 packages/coding-agent/test/server/persistence.test.ts create mode 100644 packages/coding-agent/test/server/runtime.test.ts delete mode 100644 packages/server/src/legacy/cli.ts delete mode 100644 packages/server/src/legacy/config.ts delete mode 100644 packages/server/src/legacy/handler.ts delete mode 100644 packages/server/src/legacy/index.ts delete mode 100644 packages/server/src/legacy/ipc/client.ts delete mode 100644 packages/server/src/legacy/ipc/protocol.ts delete mode 100644 packages/server/src/legacy/ipc/server.ts delete mode 100644 packages/server/src/legacy/radius.ts delete mode 100644 packages/server/src/legacy/rpc-process.ts delete mode 100644 packages/server/src/legacy/serve.ts delete mode 100644 packages/server/src/legacy/storage.ts delete mode 100644 packages/server/src/legacy/supervisor.ts delete mode 100644 packages/server/src/legacy/types.ts diff --git a/package-lock.json b/package-lock.json index 442d4d74d..b8dda5cc6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -5529,6 +5529,7 @@ "@earendil-works/pi-ai": "^0.83.0", "@earendil-works/pi-client": "^0.83.0", "@earendil-works/pi-protocol": "^0.83.0", + "@earendil-works/pi-server": "^0.83.0", "@earendil-works/pi-tui": "^0.83.0", "@silvia-odwyer/photon-node": "0.3.4", "chalk": "5.6.2", @@ -5715,12 +5716,8 @@ "license": "MIT", "dependencies": { "@earendil-works/pi-ai": "^0.83.0", - "@earendil-works/pi-coding-agent": "^0.83.0", "@earendil-works/pi-protocol": "^0.83.0" }, - "bin": { - "server": "dist/legacy/cli.js" - }, "devDependencies": { "shx": "0.4.0", "vitest": "4.1.9" diff --git a/package.json b/package.json index d63205796..53958623c 100644 --- a/package.json +++ b/package.json @@ -13,8 +13,8 @@ ], "scripts": { "clean": "npm run clean --workspaces", - "build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../coding-agent && npm run build && cd ../server && npm run build", - "build:offline": "cd packages/tui && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../coding-agent && npm run build && cd ../server && npm run build", + "build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../storage/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/tui && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../storage/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", "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", diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 53b1f3347..98833d080 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -183,6 +183,7 @@ export class AgentHarness< private readonly activeTasks = new Map, TrackedTaskKind>(); private shutdownPromise?: Promise; private isShutdown = false; + private discardPendingWrites = false; private pendingSessionWrites: PendingSessionWrite[] = []; private model: Model; private thinkingLevel: ThinkingLevel; @@ -552,6 +553,10 @@ export class AgentHarness< } private async flushPendingSessionWrites(): Promise { + if (this.discardPendingWrites) { + this.pendingSessionWrites = []; + return; + } while (this.pendingSessionWrites.length > 0) { const write = this.pendingSessionWrites[0]!; if (write.type === "message") { @@ -578,6 +583,7 @@ export class AgentHarness< } private async handleAgentEvent(event: AgentEvent, signal?: AbortSignal): Promise { + if (this.discardPendingWrites) return; if (event.type === "message_end") { await this.session.appendMessage(event.message); await this.emitAny(event, signal); @@ -1101,7 +1107,11 @@ export class AgentHarness< } /** Permanently stop this harness instance without deleting its durable session. */ - requestShutdown(): void { + requestShutdown(options: { discardPendingWrites?: boolean } = {}): void { + if (options.discardPendingWrites) { + this.discardPendingWrites = true; + this.pendingSessionWrites = []; + } if (this.isShutdown) return; this.isShutdown = true; this.pendingSessionWrites = []; diff --git a/packages/agent/test/harness/agent-harness.test.ts b/packages/agent/test/harness/agent-harness.test.ts index ed7b55206..f78053475 100644 --- a/packages/agent/test/harness/agent-harness.test.ts +++ b/packages/agent/test/harness/agent-harness.test.ts @@ -213,6 +213,33 @@ describe("AgentHarness", () => { }); }); + it("can discard active work when shutting down after ownership is lost", async () => { + const registration = newFaux(); + const entered = deferred(); + const release = deferred(); + registration.setResponses([ + async () => { + entered.resolve(); + await release.promise; + return fauxAssistantMessage("must not persist"); + }, + ]); + const session = await createInMemorySession(); + const harness = new AgentHarness({ models, session, model: registration.getModel() }); + const prompt = harness.prompt("hello"); + await entered.promise; + + harness.requestShutdown({ discardPendingWrites: true }); + release.resolve(); + + await expect(prompt).resolves.toMatchObject({ role: "assistant" }); + await expect(harness.waitForShutdown()).resolves.toBeUndefined(); + const messages = (await session.getEntries()).flatMap((entry) => + entry.type === "message" ? [entry.message] : [], + ); + expect(messages.map((message) => message.role)).toEqual(["user"]); + }); + it("allows a hook to request shutdown without deadlocking its operation", async () => { const registration = newFaux(); let providerCalls = 0; diff --git a/packages/coding-agent/install-lock/package-lock.json b/packages/coding-agent/install-lock/package-lock.json index 5766db1ec..5e3cb66a6 100644 --- a/packages/coding-agent/install-lock/package-lock.json +++ b/packages/coding-agent/install-lock/package-lock.json @@ -508,6 +508,7 @@ "@earendil-works/pi-ai": "^0.83.0", "@earendil-works/pi-client": "^0.83.0", "@earendil-works/pi-protocol": "^0.83.0", + "@earendil-works/pi-server": "^0.83.0", "@earendil-works/pi-tui": "^0.83.0", "@silvia-odwyer/photon-node": "0.3.4", "chalk": "5.6.2", @@ -546,6 +547,18 @@ "node": ">=22.19.0" } }, + "node_modules/@earendil-works/pi-server": { + "version": "0.83.0", + "resolved": "https://registry.npmjs.org/@earendil-works/pi-server/-/pi-server-0.83.0.tgz", + "license": "MIT", + "dependencies": { + "@earendil-works/pi-ai": "^0.83.0", + "@earendil-works/pi-protocol": "^0.83.0" + }, + "engines": { + "node": ">=22.19.0" + } + }, "node_modules/@earendil-works/pi-tui": { "version": "0.83.0", "resolved": "https://registry.npmjs.org/@earendil-works/pi-tui/-/pi-tui-0.83.0.tgz", diff --git a/packages/coding-agent/npm-shrinkwrap.json b/packages/coding-agent/npm-shrinkwrap.json index 80e7e71db..1bd12e31f 100644 --- a/packages/coding-agent/npm-shrinkwrap.json +++ b/packages/coding-agent/npm-shrinkwrap.json @@ -13,6 +13,7 @@ "@earendil-works/pi-ai": "^0.83.0", "@earendil-works/pi-client": "^0.83.0", "@earendil-works/pi-protocol": "^0.83.0", + "@earendil-works/pi-server": "^0.83.0", "@earendil-works/pi-tui": "^0.83.0", "@silvia-odwyer/photon-node": "0.3.4", "chalk": "5.6.2", @@ -536,6 +537,18 @@ "node": ">=22.19.0" } }, + "node_modules/@earendil-works/pi-server": { + "version": "0.83.0", + "resolved": "https://registry.npmjs.org/@earendil-works/pi-server/-/pi-server-0.83.0.tgz", + "license": "MIT", + "dependencies": { + "@earendil-works/pi-ai": "^0.83.0", + "@earendil-works/pi-protocol": "^0.83.0" + }, + "engines": { + "node": ">=22.19.0" + } + }, "node_modules/@earendil-works/pi-tui": { "version": "0.83.0", "resolved": "https://registry.npmjs.org/@earendil-works/pi-tui/-/pi-tui-0.83.0.tgz", diff --git a/packages/coding-agent/package.json b/packages/coding-agent/package.json index 5c1186325..e146a7faa 100644 --- a/packages/coding-agent/package.json +++ b/packages/coding-agent/package.json @@ -22,6 +22,10 @@ "./client": { "types": "./dist/client/index.d.ts", "import": "./dist/client/index.js" + }, + "./server": { + "types": "./dist/server/index.d.ts", + "import": "./dist/server/index.js" } }, "files": [ @@ -35,7 +39,7 @@ "scripts": { "clean": "shx rm -rf dist", "build": "tsgo -p tsconfig.build.json && shx chmod +x dist/cli.js dist/rpc-entry.js && npm run copy-assets", - "build:binary": "npm --prefix ../tui run build && npm --prefix ../ai run build && npm --prefix ../agent run build && npm --prefix ../protocol run build && npm --prefix ../client run build && npm run build && bun build --compile ./dist/bun/cli.js ./src/utils/image-resize-worker.ts --outfile dist/pi && npm run copy-binary-assets", + "build:binary": "npm --prefix ../tui run build && npm --prefix ../ai run build && npm --prefix ../agent run build && npm --prefix ../protocol run build && npm --prefix ../client run build && npm --prefix ../server run build && npm run build && bun build --compile ./dist/bun/cli.js ./src/utils/image-resize-worker.ts --outfile dist/pi && npm run copy-binary-assets", "copy-assets": "shx mkdir -p dist/modes/interactive/theme && shx cp src/modes/interactive/theme/*.json dist/modes/interactive/theme/ && shx mkdir -p dist/modes/interactive/assets && shx cp src/modes/interactive/assets/*.png dist/modes/interactive/assets/ && shx mkdir -p dist/core/export-html/vendor && shx cp src/core/export-html/template.html src/core/export-html/template.css src/core/export-html/template.js dist/core/export-html/ && shx cp src/core/export-html/vendor/*.js dist/core/export-html/vendor/", "copy-binary-assets": "shx cp package.json dist/ && shx cp README.md dist/ && shx cp CHANGELOG.md dist/ && shx mkdir -p dist/theme && shx cp src/modes/interactive/theme/*.json dist/theme/ && shx mkdir -p dist/assets && shx cp src/modes/interactive/assets/*.png dist/assets/ && shx mkdir -p dist/export-html/vendor && shx cp src/core/export-html/template.html dist/export-html/ && shx cp src/core/export-html/vendor/*.js dist/export-html/vendor/ && shx cp -r docs dist/ && shx cp -r examples dist/ && shx cp ../../node_modules/@silvia-odwyer/photon-node/photon_rs_bg.wasm dist/", "test": "vitest --run", @@ -47,6 +51,7 @@ "@earendil-works/pi-ai": "^0.83.0", "@earendil-works/pi-client": "^0.83.0", "@earendil-works/pi-protocol": "^0.83.0", + "@earendil-works/pi-server": "^0.83.0", "@earendil-works/pi-tui": "^0.83.0", "@silvia-odwyer/photon-node": "0.3.4", "chalk": "5.6.2", diff --git a/packages/coding-agent/src/server/backend.ts b/packages/coding-agent/src/server/backend.ts new file mode 100644 index 000000000..72b77946f --- /dev/null +++ b/packages/coding-agent/src/server/backend.ts @@ -0,0 +1,211 @@ +import { isAbsolute, join, resolve } from "node:path"; +import type { Session } from "@earendil-works/pi-agent-core"; +import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import type { Api, Model } from "@earendil-works/pi-ai"; +import type { ModelMetadata, ModelRef, SessionSummary, ThinkingLevel } from "@earendil-works/pi-protocol"; +import { + type CreateSessionOptions, + PiServerError, + type PiSessionBackend, + type PiSessionRuntime, +} from "@earendil-works/pi-server"; +import { getAgentDir } from "../config.ts"; +import { ModelRuntime } from "../core/model-runtime.ts"; +import { SettingsManager } from "../core/settings-manager.ts"; +import { toPiServerError } from "./errors.ts"; +import { CodingAgentModelCatalog } from "./model-catalog.ts"; +import { type SessionLease, SessionLockManager } from "./session-lock.ts"; +import { CodingAgentSessionPersistence } from "./session-persistence.ts"; +import { CodingAgentSessionRuntime } from "./session-runtime.ts"; + +const DEFAULT_SESSION_ROOT_NAME = "server-sessions"; + +export interface CodingAgentServerBackendOptions { + agentDir?: string; + sessionRoot?: string; + defaultCwd?: string; + modelRuntime?: ModelRuntime; + settingsManager?: SettingsManager; +} + +export class CodingAgentServerBackend implements PiSessionBackend { + private readonly modelRuntime: ModelRuntime; + private readonly settingsManager: SettingsManager; + private readonly defaultCwd: string; + private readonly models: CodingAgentModelCatalog; + private readonly persistence: CodingAgentSessionPersistence; + private readonly locks: SessionLockManager; + + private constructor(options: { + modelRuntime: ModelRuntime; + settingsManager: SettingsManager; + sessionRoot: string; + defaultCwd: string; + }) { + this.modelRuntime = options.modelRuntime; + this.settingsManager = options.settingsManager; + this.defaultCwd = options.defaultCwd; + this.models = new CodingAgentModelCatalog(this.modelRuntime, this.settingsManager); + this.persistence = new CodingAgentSessionPersistence( + new NodeExecutionEnv({ cwd: this.defaultCwd }), + options.sessionRoot, + ); + this.locks = new SessionLockManager(options.sessionRoot); + } + + static async create(options: CodingAgentServerBackendOptions = {}): Promise { + const agentDir = resolve(options.agentDir ?? getAgentDir()); + const defaultCwd = resolve(options.defaultCwd ?? process.cwd()); + const settingsSource = + options.settingsManager ?? SettingsManager.create(defaultCwd, agentDir, { projectTrusted: false }); + const settingsManager = SettingsManager.inMemory(settingsSource.getGlobalSettings(), { projectTrusted: false }); + const modelRuntime = + options.modelRuntime ?? + (await ModelRuntime.create({ + authPath: join(agentDir, "auth.json"), + modelsPath: join(agentDir, "models.json"), + allowModelNetwork: false, + })); + await modelRuntime.getAvailable(); + const sessionRoot = resolve(options.sessionRoot ?? join(agentDir, DEFAULT_SESSION_ROOT_NAME)); + const backend = new CodingAgentServerBackend({ modelRuntime, settingsManager, sessionRoot, defaultCwd }); + await validateCwd(defaultCwd); + await backend.locks.initialize(); + return backend; + } + + setDefaultSessionOptions(options: { model?: ModelRef; thinkingLevel?: ThinkingLevel }): void { + this.models.setDefaults(options); + } + + listModels(): Promise { + return this.models.list(); + } + + async listSessions(): Promise { + const stored = await this.persistence.list(); + return Promise.all( + stored.map(async ({ metadata, state, createdAt }) => { + try { + if (!state.model) throw new Error("stored session has no model"); + if (state.invalidThinkingLevel !== undefined) { + throw new Error(`stored session has invalid thinking level: ${state.invalidThinkingLevel}`); + } + return { + id: metadata.id, + name: state.name, + cwd: metadata.cwd, + createdAt, + updatedAt: state.updatedAt, + phase: "idle" as const, + model: state.model, + thinkingLevel: state.thinkingLevel ?? "off", + attached: false, + locked: await this.locks.isLocked(metadata.id), + }; + } catch (error) { + throw new Error(`Failed to read coding-agent session ${metadata.id}`, { cause: error }); + } + }), + ); + } + + async createSession(options: CreateSessionOptions): Promise { + const cwd = options.cwd ?? this.defaultCwd; + await validateCwd(cwd); + const model = await this.models.resolve(options.model); + const thinkingLevel = this.models.resolveThinkingLevel(model, options.thinkingLevel); + const lease = await this.locks.acquire(options.id); + let session: Session | undefined; + try { + if (await this.persistence.find(options.id)) { + throw new PiServerError("invalid_request", `Session already exists: ${options.id}`); + } + session = await this.persistence.create({ cwd, id: options.id }); + await session.appendModelChange(model.provider, model.id); + await session.appendThinkingLevelChange(thinkingLevel); + if (options.name !== undefined) await session.appendSessionName(options.name); + return await this.createRuntime(session, model, thinkingLevel, lease); + } catch (error) { + const cleanupErrors: unknown[] = []; + if (session) { + try { + await this.persistence.delete(session); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + } + try { + await lease.release(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + if (cleanupErrors.length > 0) { + throw new AggregateError([error, ...cleanupErrors], `Failed to create and clean up session ${options.id}`); + } + throw toPiServerError(error); + } + } + + async openSession(sessionId: string): Promise { + const metadata = await this.persistence.find(sessionId); + if (!metadata) throw new PiServerError("not_found", `Session was not found: ${sessionId}`); + const lease = await this.locks.acquire(sessionId); + try { + const session = await this.persistence.open(metadata); + const { state } = await this.persistence.inspect(session); + if (!state.model) throw new PiServerError("invalid_request", `Session ${sessionId} has no saved model`); + if (state.invalidThinkingLevel !== undefined) { + throw new PiServerError( + "invalid_request", + `Session ${sessionId} has invalid thinking level: ${state.invalidThinkingLevel}`, + ); + } + const model = await this.models.resolve(state.model); + const thinkingLevel = this.models.recoverThinkingLevel(model, state.thinkingLevel); + if (state.thinkingLevel !== thinkingLevel) await session.appendThinkingLevelChange(thinkingLevel); + return await this.createRuntime(session, model, thinkingLevel, lease); + } catch (error) { + await lease.release(); + throw toPiServerError(error); + } + } + + private async createRuntime( + session: Session, + model: Model, + thinkingLevel: ThinkingLevel, + lease: SessionLease, + ): Promise { + const metadata = await session.getMetadata(); + if (!("cwd" in metadata) || typeof metadata.cwd !== "string") { + throw new PiServerError("invalid_request", "Session metadata is missing cwd"); + } + return CodingAgentSessionRuntime.create({ + session, + persistence: this.persistence, + modelRuntime: this.modelRuntime, + settingsManager: this.settingsManager, + model, + thinkingLevel, + lease, + cwd: metadata.cwd, + }); + } +} + +async function validateCwd(cwd: string): Promise { + if (!isAbsolute(cwd)) throw new PiServerError("invalid_request", `Session cwd must be absolute: ${cwd}`); + const env = new NodeExecutionEnv({ cwd }); + try { + const info = await env.fileInfo(cwd); + if (!info.ok || info.value.kind !== "directory") { + const suffix = info.ok ? "is not a directory" : info.error.message; + throw new PiServerError("invalid_request", `Invalid session cwd ${cwd}: ${suffix}`); + } + } finally { + await env.cleanup(); + } +} + +export { toPiServerError } from "./errors.ts"; diff --git a/packages/coding-agent/src/server/create-harness.ts b/packages/coding-agent/src/server/create-harness.ts new file mode 100644 index 000000000..f4c3b2ef2 --- /dev/null +++ b/packages/coding-agent/src/server/create-harness.ts @@ -0,0 +1,101 @@ +import { + AgentHarness, + type AgentHarnessStreamOptions, + type AgentHarnessTool, + createBashTool, + createEditTool, + createReadTool, + createWriteTool, + type ExecutionToolContext, + type Session, +} from "@earendil-works/pi-agent-core"; +import type { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import type { Api, Model } from "@earendil-works/pi-ai"; +import type { ThinkingLevel } from "@earendil-works/pi-protocol"; +import type { ModelRuntime } from "../core/model-runtime.ts"; +import { mergeProviderAttributionHeaders } from "../core/provider-attribution.ts"; +import type { SettingsManager } from "../core/settings-manager.ts"; +import { buildSystemPrompt } from "../core/system-prompt.ts"; + +const TOOL_NAMES = ["read", "bash", "edit", "write"] as const; +const TOOL_SNIPPETS: Record<(typeof TOOL_NAMES)[number], string> = { + read: "Read file contents", + bash: "Execute bash commands (ls, grep, find, etc.)", + edit: "Make precise file edits with exact text replacement, including multiple disjoint edits in one call", + write: "Create or overwrite files", +}; +const TOOL_GUIDELINES = [ + "Inspect PI_* environment variables for current model and session details.", + "Use read to examine files instead of cat or sed.", + "Use edit for precise changes (edits[].oldText must match exactly)", + "When changing multiple separate locations in one file, use one edit call with multiple entries in edits[] instead of multiple edit calls", + "Each edits[].oldText is matched against the original file, not after earlier edits are applied. Do not emit overlapping or nested edits. Merge nearby changes into one edit.", + "Keep edits[].oldText as small as possible while still being unique in the file. Do not pad with large unchanged regions.", + "Use write only for new files or complete rewrites.", +]; + +export interface CreateServerHarnessOptions { + session: Session; + modelRuntime: ModelRuntime; + settingsManager: SettingsManager; + model: Model; + thinkingLevel: ThinkingLevel; + env: NodeExecutionEnv; + streamOptions: AgentHarnessStreamOptions; + assertUsable(): void; +} + +export function createServerHarness(options: CreateServerHarnessOptions): AgentHarness { + let harness: AgentHarness; + const tools: AgentHarnessTool[] = [ + createReadTool(), + createBashTool({ + commandPrefix: options.settingsManager.getShellCommandPrefix(), + prepare: async (execution) => { + options.assertUsable(); + const metadata = await options.session.getMetadata(); + execution.env.PI_SESSION_ID = metadata.id; + if ("path" in metadata && typeof metadata.path === "string") execution.env.PI_SESSION_FILE = metadata.path; + execution.env.PI_PROVIDER = harness.getModel().provider; + execution.env.PI_MODEL = harness.getModel().id; + execution.env.PI_REASONING_LEVEL = harness.getThinkingLevel(); + }, + }), + createEditTool(), + createWriteTool(), + ]; + harness = new AgentHarness({ + session: options.session, + models: options.modelRuntime, + model: options.model, + thinkingLevel: options.thinkingLevel, + tools, + activeToolNames: [...TOOL_NAMES], + toolContext: { env: options.env }, + resources: {}, + steeringMode: options.settingsManager.getSteeringMode(), + followUpMode: options.settingsManager.getFollowUpMode(), + streamOptions: options.streamOptions, + systemPrompt: async () => { + options.assertUsable(); + const metadata = await options.session.getMetadata(); + const cwd = "cwd" in metadata && typeof metadata.cwd === "string" ? metadata.cwd : options.env.cwd; + return buildSystemPrompt({ + cwd, + selectedTools: [...TOOL_NAMES], + toolSnippets: TOOL_SNIPPETS, + promptGuidelines: TOOL_GUIDELINES, + contextFiles: [], + skills: [], + }); + }, + }); + harness.on("before_provider_request", (event) => { + const merged = mergeProviderAttributionHeaders(event.model, options.settingsManager, event.sessionId); + const headers = merged + ? Object.fromEntries(Object.entries(merged).filter((entry): entry is [string, string] => entry[1] !== null)) + : undefined; + return { streamOptions: { headers } }; + }); + return harness; +} diff --git a/packages/coding-agent/src/server/errors.ts b/packages/coding-agent/src/server/errors.ts new file mode 100644 index 000000000..f30fd8164 --- /dev/null +++ b/packages/coding-agent/src/server/errors.ts @@ -0,0 +1,17 @@ +import { AgentHarnessError, SessionError } from "@earendil-works/pi-agent-core"; +import { PiServerError } from "@earendil-works/pi-server"; + +export function toPiServerError(error: unknown): Error { + if (error instanceof PiServerError) return error; + if (error instanceof AgentHarnessError) { + if (error.code === "busy") return new PiServerError("busy", error.message); + if (error.code === "invalid_argument") return new PiServerError("invalid_request", error.message); + if (error.code === "session" && error.cause instanceof SessionError && error.cause.code === "not_found") { + return new PiServerError("not_found", error.message); + } + } + if (error instanceof SessionError && error.code === "not_found") { + return new PiServerError("not_found", error.message); + } + return error instanceof Error ? error : new Error(String(error)); +} diff --git a/packages/coding-agent/src/server/index.ts b/packages/coding-agent/src/server/index.ts new file mode 100644 index 000000000..4b99f8da4 --- /dev/null +++ b/packages/coding-agent/src/server/index.ts @@ -0,0 +1,4 @@ +export { + CodingAgentServerBackend, + type CodingAgentServerBackendOptions, +} from "./backend.ts"; diff --git a/packages/coding-agent/src/server/model-catalog.ts b/packages/coding-agent/src/server/model-catalog.ts new file mode 100644 index 000000000..0869f869c --- /dev/null +++ b/packages/coding-agent/src/server/model-catalog.ts @@ -0,0 +1,69 @@ +import { type Api, clampThinkingLevel, type Model } from "@earendil-works/pi-ai"; +import type { ModelMetadata, ModelRef, ThinkingLevel } from "@earendil-works/pi-protocol"; +import { PiServerError, toProtocolModelMetadata } from "@earendil-works/pi-server"; +import { DEFAULT_THINKING_LEVEL } from "../core/defaults.ts"; +import type { ModelRuntime } from "../core/model-runtime.ts"; +import type { SettingsManager } from "../core/settings-manager.ts"; + +export class CodingAgentModelCatalog { + private readonly modelRuntime: ModelRuntime; + private readonly settingsManager: SettingsManager; + private defaultModel: ModelRef | undefined; + private defaultThinkingLevel: ThinkingLevel | undefined; + + constructor(modelRuntime: ModelRuntime, settingsManager: SettingsManager) { + this.modelRuntime = modelRuntime; + this.settingsManager = settingsManager; + } + + setDefaults(options: { model?: ModelRef; thinkingLevel?: ThinkingLevel }): void { + this.defaultModel = options.model; + this.defaultThinkingLevel = options.thinkingLevel; + } + + async list(): Promise { + await this.modelRuntime.getAvailable(); + return this.modelRuntime + .getModels() + .map((model) => toProtocolModelMetadata(model, this.modelRuntime.hasConfiguredAuth(model.provider))); + } + + async resolve(reference?: ModelRef): Promise> { + await this.modelRuntime.getAvailable(); + const requested = reference ?? this.defaultModel; + let model = requested ? this.modelRuntime.getModel(requested.provider, requested.id) : undefined; + if (!model && !requested) { + const provider = this.settingsManager.getDefaultProvider(); + const id = this.settingsManager.getDefaultModel(); + const configured = provider && id ? this.modelRuntime.getModel(provider, id) : undefined; + if (configured && this.modelRuntime.hasConfiguredAuth(configured.provider)) model = configured; + } + if (!model && !requested) model = this.modelRuntime.getAvailableSnapshot()[0]; + if (!model) { + const label = requested ? `${requested.provider}/${requested.id}` : "a default model"; + throw new PiServerError("invalid_request", `Could not resolve ${label}`); + } + if (!this.modelRuntime.hasConfiguredAuth(model.provider)) { + throw new PiServerError("invalid_request", `Model is not authenticated: ${model.provider}/${model.id}`); + } + return model; + } + + recoverThinkingLevel(model: Model, persisted?: ThinkingLevel): ThinkingLevel { + return persisted === undefined ? this.resolveThinkingLevel(model) : clampThinkingLevel(model, persisted); + } + + resolveThinkingLevel(model: Model, requested?: ThinkingLevel): ThinkingLevel { + const fallback = + this.defaultThinkingLevel ?? this.settingsManager.getDefaultThinkingLevel() ?? DEFAULT_THINKING_LEVEL; + const level = requested ?? fallback; + const clamped = clampThinkingLevel(model, level); + if (requested !== undefined && clamped !== requested) { + throw new PiServerError( + "invalid_request", + `Thinking level ${requested} is not supported by ${model.provider}/${model.id}`, + ); + } + return clamped; + } +} diff --git a/packages/coding-agent/src/server/session-lock.ts b/packages/coding-agent/src/server/session-lock.ts new file mode 100644 index 000000000..cc591c469 --- /dev/null +++ b/packages/coding-agent/src/server/session-lock.ts @@ -0,0 +1,111 @@ +import { createHash } from "node:crypto"; +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { PiServerError } from "@earendil-works/pi-server"; +import lockfile from "proper-lockfile"; + +const STALE_MS = 30_000; +const UPDATE_MS = 10_000; + +export interface SessionLease { + compromisedError(): Error | undefined; + onCompromised(listener: (error: Error) => void): () => void; + release(): Promise; +} + +function errorCode(error: unknown): string | undefined { + return typeof error === "object" && error !== null && "code" in error + ? String((error as { code?: unknown }).code) + : undefined; +} + +export class SessionLockManager { + private readonly root: string; + + constructor(sessionRoot: string) { + this.root = join(sessionRoot, ".locks"); + } + + async initialize(): Promise { + await mkdir(this.root, { recursive: true, mode: 0o700 }); + } + + async isLocked(sessionId: string): Promise { + return lockfile.check(await this.ensureTarget(sessionId), { realpath: false, stale: STALE_MS }); + } + + async acquire(sessionId: string): Promise { + const target = await this.ensureTarget(sessionId); + let compromised: Error | undefined; + const listeners = new Set<(error: Error) => void>(); + let releaseLock: (() => Promise) | undefined; + try { + releaseLock = await lockfile.lock(target, { + realpath: false, + stale: STALE_MS, + update: UPDATE_MS, + retries: 0, + onCompromised: (error) => { + compromised = error; + for (const listener of listeners) listener(error); + }, + }); + } catch (error) { + if (errorCode(error) === "ELOCKED") { + throw new PiServerError("session_locked", `Session is locked: ${sessionId}`); + } + throw error; + } + return this.createLease( + () => releaseLock?.(), + listeners, + () => compromised, + ); + } + + private async ensureTarget(sessionId: string): Promise { + await this.initialize(); + const digest = createHash("sha256").update(sessionId, "utf8").digest("hex"); + const target = join(this.root, digest); + await writeFile(target, "", { flag: "a", mode: 0o600 }); + return target; + } + + private createLease( + releaseLock: () => Promise | undefined, + listeners: Set<(error: Error) => void>, + getCompromised: () => Error | undefined, + ): SessionLease { + let released = false; + let releasing: Promise | undefined; + return { + compromisedError: getCompromised, + onCompromised: (listener) => { + listeners.add(listener); + const compromised = getCompromised(); + if (compromised) listener(compromised); + return () => listeners.delete(listener); + }, + release: async () => { + if (released) return; + if (getCompromised()) { + released = true; + listeners.clear(); + return; + } + if (releasing) return releasing; + const current = (async () => { + await releaseLock(); + released = true; + listeners.clear(); + })(); + releasing = current; + try { + await current; + } finally { + if (!released && releasing === current) releasing = undefined; + } + }, + }; + } +} diff --git a/packages/coding-agent/src/server/session-persistence.ts b/packages/coding-agent/src/server/session-persistence.ts new file mode 100644 index 000000000..27078065d --- /dev/null +++ b/packages/coding-agent/src/server/session-persistence.ts @@ -0,0 +1,140 @@ +import { + createJsonlSessionStore, + createSessionRepository, + type JsonlSessionCreateOptions, + type JsonlSessionListOptions, + type JsonlSessionMetadata, + type Session, + type SessionRepository, + type SessionTreeEntry, +} from "@earendil-works/pi-agent-core"; +import type { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import type { ThinkingLevel } from "@earendil-works/pi-protocol"; +import { isAssistantMessage } from "./transcript/projection.ts"; + +export interface StoredSessionState { + model: { provider: string; id: string } | undefined; + thinkingLevel: ThinkingLevel | undefined; + invalidThinkingLevel: string | undefined; + name: string | undefined; + updatedAt: number; +} + +export interface StoredSession { + metadata: JsonlSessionMetadata; + state: StoredSessionState; + createdAt: number; +} + +const THINKING_LEVELS = new Set(["off", "minimal", "low", "medium", "high", "xhigh", "max"]); + +export class CodingAgentSessionPersistence { + private readonly repository: SessionRepository< + JsonlSessionMetadata, + JsonlSessionCreateOptions, + JsonlSessionListOptions + >; + + constructor(fs: NodeExecutionEnv, sessionsRoot: string) { + this.repository = createSessionRepository({ store: createJsonlSessionStore({ fs, sessionsRoot }) }); + } + + async list(): Promise { + return Promise.all((await this.repository.list()).map((metadata) => this.load(metadata))); + } + + async find(sessionId: string): Promise { + return (await this.repository.list()).find((metadata) => metadata.id === sessionId); + } + + async create(options: { id: string; cwd: string }): Promise { + return this.repository.create(options); + } + + async open(metadata: JsonlSessionMetadata): Promise { + return this.repository.open(metadata); + } + + async delete(session: Session): Promise { + const metadata = await session.getMetadata(); + if ( + !("cwd" in metadata) || + typeof metadata.cwd !== "string" || + !("path" in metadata) || + typeof metadata.path !== "string" + ) { + throw new Error(`Cannot delete session ${metadata.id}: JSONL metadata is incomplete`); + } + await this.repository.delete({ ...metadata, cwd: metadata.cwd, path: metadata.path }); + } + + async inspect( + session: Session, + ): Promise<{ branch: SessionTreeEntry[]; state: StoredSessionState; createdAt: number }> { + const metadata = await session.getMetadata(); + const branch = await getFullActiveBranch(session); + const createdAt = parseCreatedAt(metadata.createdAt); + return { branch, state: readStoredSessionState(branch, createdAt), createdAt }; + } + + private async load(metadata: JsonlSessionMetadata): Promise { + try { + const session = await this.open(metadata); + const { state, createdAt } = await this.inspect(session); + return { metadata, state, createdAt }; + } catch (error) { + throw new Error(`Failed to read coding-agent session ${metadata.id}`, { cause: error }); + } + } +} + +export async function getFullActiveBranch(session: Session): Promise { + const branch: SessionTreeEntry[] = []; + const visited = new Set(); + let id = await session.getLeafId(); + while (id !== null) { + if (visited.has(id)) throw new Error(`Session branch contains a cycle at ${id}`); + visited.add(id); + const entry = await session.getEntry(id); + if (!entry) throw new Error(`Session branch entry ${id} was not found`); + branch.unshift(entry); + id = entry.parentId; + } + return branch; +} + +function readStoredSessionState(entries: readonly SessionTreeEntry[], createdAt: number): StoredSessionState { + let model: StoredSessionState["model"]; + let thinkingLevel: ThinkingLevel | undefined; + let invalidThinkingLevel: string | undefined; + let name: string | undefined; + let updatedAt = createdAt; + for (const entry of entries) { + const entryTime = Date.parse(entry.timestamp); + if (!Number.isFinite(entryTime) || entryTime < 0) { + throw new Error(`Invalid timestamp on session entry ${entry.id}: ${entry.timestamp}`); + } + updatedAt = Math.max(updatedAt, entryTime); + if (entry.type === "model_change") model = { provider: entry.provider, id: entry.modelId }; + if (entry.type === "thinking_level_change") { + if (THINKING_LEVELS.has(entry.thinkingLevel as ThinkingLevel)) { + thinkingLevel = entry.thinkingLevel as ThinkingLevel; + invalidThinkingLevel = undefined; + } else { + thinkingLevel = undefined; + invalidThinkingLevel = entry.thinkingLevel; + } + } + if (entry.type === "message" && isAssistantMessage(entry.message)) { + model = { provider: entry.message.provider, id: entry.message.model }; + } + if (entry.type === "session_info") name = entry.name?.trim() || undefined; + } + return { model, thinkingLevel, invalidThinkingLevel, name, updatedAt }; +} + +function parseCreatedAt(value: string): number { + const parsed = Date.parse(value); + if (!Number.isFinite(parsed) || parsed < 0) throw new Error(`Invalid session creation timestamp: ${value}`); + return parsed; +} diff --git a/packages/coding-agent/src/server/session-runtime.ts b/packages/coding-agent/src/server/session-runtime.ts new file mode 100644 index 000000000..d40b0a50d --- /dev/null +++ b/packages/coding-agent/src/server/session-runtime.ts @@ -0,0 +1,275 @@ +import type { AgentHarness, ExecutionToolContext, Session } from "@earendil-works/pi-agent-core"; +import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import { type Api, clampThinkingLevel, type Model } from "@earendil-works/pi-ai"; +import type { ModelRef, SessionPhase, SessionSnapshot, ThinkingLevel } from "@earendil-works/pi-protocol"; +import { + PiServerError, + type PiSessionRuntime, + type PiSessionRuntimeEvent, + type PromptInput, + type SteerInput, +} from "@earendil-works/pi-server"; +import type { ModelRuntime } from "../core/model-runtime.ts"; +import type { SettingsManager } from "../core/settings-manager.ts"; +import { createServerHarness } from "./create-harness.ts"; +import { toPiServerError } from "./errors.ts"; +import type { SessionLease } from "./session-lock.ts"; +import type { CodingAgentSessionPersistence } from "./session-persistence.ts"; +import { LiveTranscript } from "./transcript/live.ts"; +import { projectBranchTranscript } from "./transcript/projection.ts"; + +export interface CreateCodingAgentSessionRuntimeOptions { + session: Session; + persistence: CodingAgentSessionPersistence; + modelRuntime: ModelRuntime; + settingsManager: SettingsManager; + model: Model; + thinkingLevel: ThinkingLevel; + lease: SessionLease; + cwd: string; +} + +export class CodingAgentSessionRuntime implements PiSessionRuntime { + private readonly session: Session; + private readonly persistence: CodingAgentSessionPersistence; + private readonly modelRuntime: ModelRuntime; + private readonly env: NodeExecutionEnv; + private readonly lease: SessionLease; + private readonly harness: AgentHarness; + private readonly liveTranscript = new LiveTranscript((phase) => { + this.phase = phase; + }); + private readonly unsubscribeHarness: () => void; + private readonly unsubscribeLockCompromise: () => void; + private lockCompromise?: Error; + private terminalError?: PiServerError; + private mutationInFlight = false; + private phase: SessionPhase = "idle"; + private disposed = false; + private disposePromise?: Promise; + + private constructor(options: CreateCodingAgentSessionRuntimeOptions, env: NodeExecutionEnv) { + this.session = options.session; + this.persistence = options.persistence; + this.modelRuntime = options.modelRuntime; + this.env = env; + this.lease = options.lease; + const retry = options.settingsManager.getProviderRetrySettings(); + const idleTimeoutMs = options.settingsManager.getHttpIdleTimeoutMs(); + this.harness = createServerHarness({ + session: this.session, + modelRuntime: this.modelRuntime, + settingsManager: options.settingsManager, + model: options.model, + thinkingLevel: options.thinkingLevel, + env: this.env, + streamOptions: { + transport: options.settingsManager.getTransport(), + timeoutMs: retry.timeoutMs ?? (idleTimeoutMs === 0 ? 2_147_483_647 : idleTimeoutMs), + maxRetries: retry.maxRetries, + maxRetryDelayMs: retry.maxRetryDelayMs, + }, + assertUsable: () => this.assertUsable(), + }); + this.unsubscribeHarness = this.harness.subscribe((event) => { + try { + this.liveTranscript.handle(event); + } catch (error) { + this.handleProjectionFailure(error); + } + }); + this.unsubscribeLockCompromise = this.lease.onCompromised((error) => this.handleLockCompromise(error)); + } + + static create(options: CreateCodingAgentSessionRuntimeOptions): CodingAgentSessionRuntime { + const env = new NodeExecutionEnv({ cwd: options.cwd, shellPath: options.settingsManager.getShellPath() }); + return new CodingAgentSessionRuntime(options, env); + } + + getPhase(): SessionPhase { + return this.phase; + } + + async snapshot(): Promise { + this.assertUsable(); + const metadata = await this.session.getMetadata(); + if (!("cwd" in metadata) || typeof metadata.cwd !== "string") { + throw new PiServerError("invalid_request", "Session metadata is missing cwd"); + } + const { branch, state, createdAt } = await this.persistence.inspect(this.session); + const queuedSteer = this.liveTranscript.queuedSteer; + return { + id: metadata.id, + name: state.name, + cwd: metadata.cwd, + createdAt, + updatedAt: state.updatedAt, + phase: this.getPhase(), + model: { provider: this.harness.getModel().provider, id: this.harness.getModel().id }, + thinkingLevel: this.harness.getThinkingLevel(), + attached: false, + locked: true, + revision: this.liveTranscript.revision, + transcript: this.liveTranscript.mergeTranscript(projectBranchTranscript(branch)), + queuedSteer, + queuedSteerCount: queuedSteer.length, + }; + } + + async prompt(input: PromptInput): Promise { + this.assertIdle("prompt"); + this.phase = "turn"; + try { + await this.harness.prompt(input.text); + this.assertUsable(); + } catch (error) { + this.assertUsable(); + throw toPiServerError(error); + } finally { + if (!this.lockCompromise) this.phase = "idle"; + } + } + + async steer(input: SteerInput): Promise { + this.assertUsable(); + if (this.getPhase() !== "turn") throw new PiServerError("busy", "Session is not accepting steering input"); + try { + await this.harness.steer(input.text); + this.assertUsable(); + } catch (error) { + this.assertUsable(); + throw toPiServerError(error); + } + } + + async abort(): Promise { + this.assertUsable(); + if (this.mutationInFlight) throw new PiServerError("busy", "Cannot abort while session is being updated"); + try { + await this.harness.abort(); + this.assertUsable(); + } catch (error) { + this.assertUsable(); + throw toPiServerError(error); + } + } + + async setModel(reference: ModelRef): Promise { + await this.runExclusiveMutation("set model", () => { + const model = this.modelRuntime.getModel(reference.provider, reference.id); + if (!model) throw new PiServerError("invalid_request", `Unknown model: ${reference.provider}/${reference.id}`); + if (!this.modelRuntime.hasConfiguredAuth(model.provider)) { + throw new PiServerError( + "invalid_request", + `Model is not authenticated: ${reference.provider}/${reference.id}`, + ); + } + return async () => { + await this.harness.setModel(model); + const current = this.harness.getThinkingLevel(); + const clamped = clampThinkingLevel(model, current); + if (clamped !== current) await this.harness.setThinkingLevel(clamped); + }; + }); + } + + async setThinking(thinkingLevel: ThinkingLevel): Promise { + await this.runExclusiveMutation("set thinking", () => { + const model = this.harness.getModel(); + if (clampThinkingLevel(model, thinkingLevel) !== thinkingLevel) { + throw new PiServerError( + "invalid_request", + `Thinking level ${thinkingLevel} is not supported by ${model.provider}/${model.id}`, + ); + } + return () => this.harness.setThinkingLevel(thinkingLevel); + }); + } + + subscribe(listener: (event: PiSessionRuntimeEvent) => void): () => void { + return this.liveTranscript.subscribe(listener); + } + + async dispose(): Promise { + if (this.disposePromise) return this.disposePromise; + this.disposed = true; + const current = (async () => { + try { + this.unsubscribeHarness(); + this.unsubscribeLockCompromise(); + this.liveTranscript.clear(); + if (this.getPhase() !== "idle") await this.harness.abort().catch(() => {}); + await this.harness.waitForIdle().catch(() => {}); + await this.env.cleanup(); + } finally { + await this.lease.release(); + } + })(); + this.disposePromise = current; + try { + await current; + } catch (error) { + if (this.disposePromise === current) this.disposePromise = undefined; + throw error; + } + } + + private assertUsable(): void { + if (this.terminalError) throw this.terminalError; + this.throwIfLockCompromised(); + if (this.disposed) throw new PiServerError("invalid_request", "Session runtime is disposed"); + } + + private throwIfLockCompromised(): void { + const compromised = this.lockCompromise ?? this.lease.compromisedError(); + if (compromised) { + throw new PiServerError("session_locked", `Session lock was compromised: ${compromised.message}`); + } + } + + private handleLockCompromise(error: Error): void { + if (this.lockCompromise) return; + this.lockCompromise = error; + this.terminateHarness(new PiServerError("session_locked", `Session lock was compromised: ${error.message}`)); + } + + private handleProjectionFailure(error: unknown): void { + if (this.terminalError) return; + this.terminalError = new PiServerError("invalid_request", "Session protocol projection failed"); + this.terminalError.cause = error instanceof Error ? error : new Error(String(error)); + this.terminateHarness(this.terminalError); + } + + private terminateHarness(error: PiServerError): void { + this.unsubscribeHarness(); + this.harness.requestShutdown({ discardPendingWrites: true }); + this.liveTranscript.emit({ type: "error", error }); + } + + private assertIdle(operation: string): void { + this.assertUsable(); + if (this.mutationInFlight || this.getPhase() !== "idle") { + throw new PiServerError("busy", `Cannot ${operation} while session is busy`); + } + } + + private async runExclusiveMutation(operation: string, prepare: () => () => Promise): Promise { + this.assertIdle(operation); + const mutation = prepare(); + this.mutationInFlight = true; + try { + await mutation(); + this.assertUsable(); + } catch (error) { + if (this.terminalError || this.lockCompromise || this.lease.compromisedError() || this.disposed) { + this.assertUsable(); + } + this.terminalError = new PiServerError("invalid_request", `Session failed to ${operation}`); + this.terminalError.cause = toPiServerError(error); + this.terminateHarness(this.terminalError); + throw this.terminalError; + } finally { + this.mutationInFlight = false; + } + } +} diff --git a/packages/coding-agent/src/server/transcript/live.ts b/packages/coding-agent/src/server/transcript/live.ts new file mode 100644 index 000000000..a8d744ee9 --- /dev/null +++ b/packages/coding-agent/src/server/transcript/live.ts @@ -0,0 +1,304 @@ +import type { AgentHarnessEvent, AgentMessage } from "@earendil-works/pi-agent-core"; +import type { ToolCall } from "@earendil-works/pi-ai"; +import type { JsonValue, ToolTranscriptItem, TranscriptItem, TranscriptProgress } from "@earendil-works/pi-protocol"; +import { + type PiSessionRuntimeEvent, + sanitizeProtocolDetails, + toProtocolJsonValue, + toProtocolUsage, +} from "@earendil-works/pi-server"; +import { + isAssistantMessage, + isToolResultMessage, + isUserMessage, + projectAssistantMessage, + projectToolResult, + projectUserMessage, + transcriptItemId, +} from "./projection.ts"; + +type RunningToolTranscriptItem = Extract; +type FinishedToolTranscriptItem = Exclude; + +interface ProjectedToolResult { + content: ToolTranscriptItem["content"]; + details?: JsonValue; + usage?: ToolTranscriptItem["usage"]; +} + +function projectToolProgressResult(value: unknown): ProjectedToolResult { + if (typeof value !== "object" || value === null) return { content: [] }; + const source = value as { content?: unknown; details?: unknown; usage?: unknown }; + const content: ToolTranscriptItem["content"] = []; + if (Array.isArray(source.content)) { + for (const part of source.content) { + if (typeof part !== "object" || part === null || !("type" in part)) continue; + const candidate = part as { type?: unknown; text?: unknown; data?: unknown; mimeType?: unknown }; + if (candidate.type === "text" && typeof candidate.text === "string") { + content.push({ type: "text", text: candidate.text }); + } else if ( + candidate.type === "image" && + typeof candidate.data === "string" && + typeof candidate.mimeType === "string" && + candidate.mimeType + ) { + content.push({ type: "image", data: candidate.data, mimeType: candidate.mimeType }); + } + } + } + const details = sanitizeProtocolDetails(source.details); + const usage = toProtocolUsage(source.usage as Parameters[0]); + return { + content, + ...(details === undefined ? {} : { details }), + ...(usage === undefined ? {} : { usage }), + }; +} + +export class LiveTranscript { + private readonly onPhase: (phase: "idle" | "turn") => void; + private readonly listeners = new Set<(event: PiSessionRuntimeEvent) => void>(); + private readonly liveItems = new Map(); + private readonly liveOrder: string[] = []; + private readonly toolInputs = new Map(); + private queuedSteerItems: ReturnType[] = []; + revision = 0; + + constructor(onPhase: (phase: "idle" | "turn") => void) { + this.onPhase = onPhase; + } + + get queuedSteer(): ReturnType[] { + return structuredClone(this.queuedSteerItems); + } + + mergeTranscript(persisted: readonly TranscriptItem[]): TranscriptItem[] { + const result = persisted.map((item) => this.liveItems.get(item.id) ?? item); + const persistedIds = new Set(persisted.map((item) => item.id)); + for (const id of this.liveOrder) { + if (persistedIds.has(id)) continue; + const item = this.liveItems.get(id); + if (item) result.push(item); + } + return result; + } + + subscribe(listener: (event: PiSessionRuntimeEvent) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + clear(): void { + this.listeners.clear(); + } + + emit(event: PiSessionRuntimeEvent): void { + for (const listener of this.listeners) { + try { + listener(event); + } catch { + // A disconnected transport subscriber cannot affect the harness. + } + } + } + + handle(event: AgentHarnessEvent): void { + if (event.type === "agent_start") this.onPhase("turn"); + if (event.type === "agent_end" || event.type === "settled") this.onPhase("idle"); + if (event.type === "message_start") { + this.handleMessageStart(event.message); + return; + } + if (event.type === "message_update" && isAssistantMessage(event.message)) { + const item = projectAssistantMessage(event.message, true); + this.remember(item); + const update = event.assistantMessageEvent; + if (update.type === "text_delta" || update.type === "thinking_delta" || update.type === "toolcall_delta") { + this.emitProgress({ + type: "assistant_delta", + messageId: item.id, + contentIndex: update.contentIndex, + kind: update.type === "text_delta" ? "text" : update.type === "thinking_delta" ? "thinking" : "toolCall", + delta: update.delta, + }); + } else this.emitProgress({ type: "item_updated", item }); + return; + } + if (event.type === "message_end") { + this.handleMessageEnd(event.message); + return; + } + if (event.type === "tool_execution_start") { + const input = toProtocolJsonValue(event.args); + const startedAt = Date.now(); + this.toolInputs.set(event.toolCallId, { + call: { type: "toolCall", id: event.toolCallId, name: event.toolName, arguments: event.args }, + input, + timestamp: startedAt, + }); + const item = this.createToolItem( + event.toolCallId, + event.toolName, + input, + { content: [] }, + "running", + startedAt, + ); + this.remember(item); + this.emitProgress({ type: "item_started", item }); + return; + } + if (event.type === "tool_execution_update") { + const call = this.toolInputs.get(event.toolCallId) ?? { + call: { type: "toolCall", id: event.toolCallId, name: event.toolName, arguments: event.args }, + input: toProtocolJsonValue(event.args), + timestamp: Date.now(), + }; + this.toolInputs.set(event.toolCallId, call); + const item = this.createToolItem( + event.toolCallId, + event.toolName, + call.input, + projectToolProgressResult(event.partialResult), + "running", + call.timestamp, + ); + this.remember(item); + this.emitProgress({ type: "item_updated", item }); + return; + } + if (event.type === "tool_execution_end") { + const call = this.toolInputs.get(event.toolCallId) ?? { + call: { type: "toolCall", id: event.toolCallId, name: event.toolName, arguments: {} }, + input: null, + timestamp: Date.now(), + }; + const item = this.createToolItem( + event.toolCallId, + event.toolName, + call.input, + projectToolProgressResult(event.result), + event.isError ? "error" : "complete", + call.timestamp, + ); + this.remember(item); + this.emitProgress({ type: "item_finished", item }); + return; + } + if (event.type === "queue_update") { + this.queuedSteerItems = event.steer + .filter(isUserMessage) + .map((message, index) => projectUserMessage(message, `queued-steer-${message.timestamp}-${index}`)); + this.emitChange(); + return; + } + if ( + [ + "agent_start", + "agent_end", + "model_update", + "thinking_level_update", + "abort", + "settled", + "save_point", + ].includes(event.type) + ) + this.emitChange(); + } + + private handleMessageStart(message: AgentMessage): void { + if (isAssistantMessage(message)) { + const item = projectAssistantMessage(message, true); + this.remember(item); + this.emitChange({ type: "item_started", item }); + } else if (isUserMessage(message)) { + const item = projectUserMessage(message); + this.remember(item); + this.emitChange({ type: "item_started", item }); + } else this.emitChange(); + } + + private handleMessageEnd(message: Parameters[0]): void { + const id = transcriptItemId(message); + if (isAssistantMessage(message)) { + const item = projectAssistantMessage(message); + this.remember(item); + this.revision += 1; + this.emit({ type: "progress", progress: { type: "item_finished", item } }); + this.forget(id); + this.emit({ type: "snapshot" }); + } else if (isToolResultMessage(message)) { + const call = this.toolInputs.get(message.toolCallId); + const item = projectToolResult(message, call?.call); + this.remember(item); + this.forget(item.id); + this.toolInputs.delete(message.toolCallId); + this.emitChange(); + } else { + this.forget(id); + this.emitChange(); + } + } + + private remember(item: TranscriptItem): void { + if (!this.liveItems.has(item.id)) this.liveOrder.push(item.id); + this.liveItems.set(item.id, item); + } + + private forget(id: string): void { + this.liveItems.delete(id); + const index = this.liveOrder.indexOf(id); + if (index >= 0) this.liveOrder.splice(index, 1); + } + + private emitProgress(progress: TranscriptProgress): void { + this.revision += 1; + this.emit({ type: "progress", progress }); + } + + private emitChange(progress?: TranscriptProgress): void { + this.revision += 1; + if (progress) this.emit({ type: "progress", progress }); + this.emit({ type: "snapshot" }); + } + + private createToolItem( + toolCallId: string, + toolName: string, + input: JsonValue, + result: ProjectedToolResult, + status: "running", + timestamp: number, + ): RunningToolTranscriptItem; + private createToolItem( + toolCallId: string, + toolName: string, + input: JsonValue, + result: ProjectedToolResult, + status: "complete" | "error", + timestamp: number, + ): FinishedToolTranscriptItem; + private createToolItem( + toolCallId: string, + toolName: string, + input: JsonValue, + result: ProjectedToolResult, + status: ToolTranscriptItem["status"], + timestamp: number, + ): ToolTranscriptItem { + const common = { + id: `tool-${toolCallId}`, + role: "tool", + toolCallId, + toolName: toolName || "unknown", + input, + content: result.content, + ...(result.details === undefined ? {} : { details: result.details }), + ...(result.usage === undefined ? {} : { usage: result.usage }), + timestamp, + } as const; + if (status === "running") return { ...common, status, isError: false }; + if (status === "error") return { ...common, status, isError: true }; + return { ...common, status, isError: false }; + } +} diff --git a/packages/coding-agent/src/server/transcript/projection.ts b/packages/coding-agent/src/server/transcript/projection.ts new file mode 100644 index 000000000..d08e3291e --- /dev/null +++ b/packages/coding-agent/src/server/transcript/projection.ts @@ -0,0 +1,102 @@ +import type { AgentMessage, SessionTreeEntry } from "@earendil-works/pi-agent-core"; +import type { AssistantMessage, ToolCall, ToolResultMessage, UserMessage } from "@earendil-works/pi-ai"; +import type { AssistantTranscriptItem, ToolTranscriptItem, TranscriptItem } from "@earendil-works/pi-protocol"; +import { + toProtocolAssistantMessage, + toProtocolToolResultMessage, + toProtocolUserMessage, +} from "@earendil-works/pi-server"; + +type FinishedAssistantTranscriptItem = Exclude; +type FinishedToolTranscriptItem = Exclude; + +function timestamp(value: number): number { + if (!Number.isSafeInteger(value) || value < 0) throw new Error(`Invalid message timestamp: ${value}`); + return value; +} + +export function transcriptItemId(message: AgentMessage): string { + if (message.role === "toolResult") return `tool-${message.toolCallId || timestamp(message.timestamp)}`; + return `${message.role}-${timestamp(message.timestamp)}`; +} + +export function projectAssistantMessage( + message: AssistantMessage, + streaming: true, + id?: string, +): Extract; +export function projectAssistantMessage( + message: AssistantMessage, + streaming?: false, + id?: string, +): FinishedAssistantTranscriptItem; +export function projectAssistantMessage( + message: AssistantMessage, + streaming = false, + id = transcriptItemId(message), +): AssistantTranscriptItem { + const item = toProtocolAssistantMessage(streaming ? { ...message, stopReason: "pending" } : message, { id }); + if (streaming && item.status !== "streaming") throw new Error("Expected a streaming assistant transcript item"); + if (!streaming && item.status === "streaming") throw new Error("Expected a finished assistant transcript item"); + return item; +} + +export function projectUserMessage(message: UserMessage, id = transcriptItemId(message)) { + return toProtocolUserMessage(message, { id }); +} + +export function projectToolResult( + message: ToolResultMessage, + call?: ToolCall, + id = transcriptItemId(message), +): FinishedToolTranscriptItem { + const item = toProtocolToolResultMessage(message, { + id, + call: call ?? { + type: "toolCall", + id: message.toolCallId, + name: message.toolName || "unknown", + arguments: {}, + }, + }); + if (item.status === "running") throw new Error("Expected a finished tool transcript item"); + return item; +} + +export function isAssistantMessage(message: AgentMessage): message is AssistantMessage { + return message.role === "assistant"; +} + +export function isUserMessage(message: AgentMessage): message is UserMessage { + return message.role === "user"; +} + +export function isToolResultMessage(message: AgentMessage): message is ToolResultMessage { + return message.role === "toolResult"; +} + +export function projectBranchTranscript(entries: readonly SessionTreeEntry[]): TranscriptItem[] { + const calls = new Map(); + const transcript: TranscriptItem[] = []; + for (const entry of entries) { + if (entry.type !== "message") continue; + const message = entry.message; + if (isUserMessage(message)) { + transcript.push(projectUserMessage(message, entry.id)); + continue; + } + if (isAssistantMessage(message)) { + for (const part of message.content) { + if (part.type === "toolCall") { + calls.set(part.id, part); + } + } + transcript.push(projectAssistantMessage(message, false, entry.id)); + continue; + } + if (isToolResultMessage(message)) { + transcript.push(projectToolResult(message, calls.get(message.toolCallId), entry.id)); + } + } + return transcript; +} diff --git a/packages/coding-agent/test/server/backend.test.ts b/packages/coding-agent/test/server/backend.test.ts new file mode 100644 index 000000000..ac40f3f15 --- /dev/null +++ b/packages/coding-agent/test/server/backend.test.ts @@ -0,0 +1,153 @@ +import { join } from "node:path"; +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import type { TranscriptProgress } from "@earendil-works/pi-protocol"; +import { describe, expect, test } from "vitest"; +import { CodingAgentServerBackend, toPiServerError } from "../../src/server/backend.ts"; +import { createServerBackendFixture, removeServerBackendFixture } from "./fixture.ts"; + +describe("coding-agent server backend", () => { + test("normalizes snapshots, restores model/thinking, and holds an exclusive session lock", async () => { + const fixture = await createServerBackendFixture(); + let runtime = await fixture.backend.createSession({ + id: "server-session-1", + cwd: fixture.cwd, + name: "Backend test", + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + thinkingLevel: "high", + }); + try { + const initial = await runtime.snapshot(); + expect(initial).toMatchObject({ + id: "server-session-1", + name: "Backend test", + cwd: fixture.cwd, + phase: "idle", + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + thinkingLevel: "high", + revision: 0, + transcript: [], + locked: true, + }); + + const models = await fixture.backend.listModels(); + expect(models).toContainEqual( + expect.objectContaining({ + provider: fixture.faux.provider.id, + id: "faux-reasoning", + authenticated: true, + supportedThinkingLevels: expect.arrayContaining(["off", "high"]), + }), + ); + const summaries = await fixture.backend.listSessions(); + expect(summaries).toContainEqual( + expect.objectContaining({ id: "server-session-1", name: "Backend test", locked: true }), + ); + + const secondBackend = await CodingAgentServerBackend.create({ + defaultCwd: fixture.cwd, + sessionRoot: join(fixture.root, "sessions"), + modelRuntime: fixture.modelRuntime, + settingsManager: fixture.settingsManager, + }); + await expect(secondBackend.openSession("server-session-1")).rejects.toMatchObject({ + code: "session_locked", + }); + + fixture.faux.setResponses([fauxAssistantMessage("normalized response")]); + const progress: TranscriptProgress[] = []; + runtime.subscribe(() => { + throw new Error("network subscriber failed"); + }); + runtime.subscribe((event) => { + if (event.type === "progress") progress.push(event.progress); + }); + await runtime.prompt({ text: "hello" }); + const completed = await runtime.snapshot(); + expect(completed.revision).toBeGreaterThan(initial.revision); + expect(completed.phase).toBe("idle"); + expect(completed.transcript.map((item) => item.role)).toEqual(["user", "assistant"]); + expect(completed.transcript[1]).toMatchObject({ + role: "assistant", + content: [{ type: "text", text: "normalized response" }], + status: "complete", + stopReason: "stop", + }); + expect(completed.transcript.some((item) => "parentId" in item)).toBe(false); + expect(progress.some((event) => event.type === "assistant_delta")).toBe(true); + + await runtime.setModel({ provider: fixture.faux.provider.id, id: "faux-plain" }); + expect(await runtime.snapshot()).toMatchObject({ + model: { provider: fixture.faux.provider.id, id: "faux-plain" }, + thinkingLevel: "off", + }); + await runtime.setModel({ provider: fixture.faux.provider.id, id: "faux-reasoning" }); + await runtime.setThinking("high"); + + await runtime.dispose(); + const persistedSummary = (await secondBackend.listSessions()).find( + (summary) => summary.id === "server-session-1", + ); + expect(persistedSummary).toBeDefined(); + runtime = await secondBackend.openSession("server-session-1"); + const restored = await runtime.snapshot(); + expect(restored.updatedAt).toBe(persistedSummary?.updatedAt); + expect(restored.model).toEqual(initial.model); + expect(restored.thinkingLevel).toBe("high"); + expect(restored.transcript).toEqual(completed.transcript); + } finally { + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("preserves unexpected operational errors for boundary-safe handling", () => { + const operational = new Error("private filesystem detail"); + expect(toPiServerError(operational)).toBe(operational); + }); + test("applies server-only defaults to newly created sessions", async () => { + const fixture = await createServerBackendFixture(); + fixture.backend.setDefaultSessionOptions({ + model: { provider: fixture.faux.provider.id, id: "faux-plain" }, + thinkingLevel: "off", + }); + const runtime = await fixture.backend.createSession({ id: "server-defaults", cwd: fixture.cwd }); + try { + expect(await runtime.snapshot()).toMatchObject({ + model: { provider: fixture.faux.provider.id, id: "faux-plain" }, + thinkingLevel: "off", + }); + } finally { + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("maps unsupported model and thinking requests to protocol-safe errors", async () => { + const fixture = await createServerBackendFixture(); + try { + await expect( + fixture.backend.createSession({ + id: "bad-model", + cwd: fixture.cwd, + model: { provider: "missing", id: "missing" }, + }), + ).rejects.toMatchObject({ + name: "PiServerError", + code: "invalid_request", + message: expect.stringContaining("Could not resolve missing/missing"), + }); + await expect( + fixture.backend.createSession({ + id: "bad-thinking", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + thinkingLevel: "max", + }), + ).rejects.toMatchObject({ + name: "PiServerError", + code: "invalid_request", + message: expect.stringContaining("not supported"), + }); + } finally { + await removeServerBackendFixture(fixture); + } + }); +}); diff --git a/packages/coding-agent/test/server/fixture.ts b/packages/coding-agent/test/server/fixture.ts new file mode 100644 index 000000000..adea8f449 --- /dev/null +++ b/packages/coding-agent/test/server/fixture.ts @@ -0,0 +1,70 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { type FauxResponseFactory, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai"; +import { AuthStorage } from "../../src/core/auth-storage.ts"; +import { ModelRuntime } from "../../src/core/model-runtime.ts"; +import { SettingsManager } from "../../src/core/settings-manager.ts"; +import { CodingAgentServerBackend } from "../../src/server/backend.ts"; + +export async function createServerBackendFixture() { + const root = await mkdtemp(join(tmpdir(), "pi-experimental-backend-")); + const cwd = join(root, "workspace"); + await mkdir(cwd); + const faux = fauxProvider({ + provider: `faux-${randomUUID()}`, + models: [ + { id: "faux-reasoning", name: "Faux Reasoning", reasoning: true }, + { id: "faux-plain", name: "Faux Plain", reasoning: false }, + ], + tokensPerSecond: 100, + tokenSize: { min: 2, max: 3 }, + }); + const modelRuntime = await ModelRuntime.create({ + credentials: AuthStorage.inMemory(), + modelsPath: null, + allowModelNetwork: false, + }); + modelRuntime.registerNativeProvider(faux.provider); + await modelRuntime.refresh({ allowNetwork: false }); + const settingsManager = SettingsManager.inMemory({ + defaultProvider: faux.provider.id, + defaultModel: "faux-reasoning", + defaultThinkingLevel: "high", + transport: "sse", + retry: { provider: { timeoutMs: 5_000, maxRetries: 0, maxRetryDelayMs: 100 } }, + }); + const backend = await CodingAgentServerBackend.create({ + defaultCwd: cwd, + sessionRoot: join(root, "sessions"), + modelRuntime, + settingsManager, + }); + return { root, cwd, faux, modelRuntime, settingsManager, backend }; +} + +export async function removeServerBackendFixture( + fixture: Awaited>, +): Promise { + await rm(fixture.root, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 }); +} + +export function abortableResponse(): { response: FauxResponseFactory; started: Promise } { + let startedResolve: (() => void) | undefined; + const started = new Promise((resolve) => { + startedResolve = resolve; + }); + return { + started, + response: async (_context, options) => { + startedResolve?.(); + const signal = options?.signal; + if (!signal) throw new Error("Expected an abort signal"); + if (!signal.aborted) { + await new Promise((resolve) => signal.addEventListener("abort", () => resolve(), { once: true })); + } + return fauxAssistantMessage("aborted"); + }, + }; +} diff --git a/packages/coding-agent/test/server/persistence.test.ts b/packages/coding-agent/test/server/persistence.test.ts new file mode 100644 index 000000000..c4dad1284 --- /dev/null +++ b/packages/coding-agent/test/server/persistence.test.ts @@ -0,0 +1,158 @@ +import { randomUUID } from "node:crypto"; +import { appendFile, readdir, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import { describe, expect, test, vi } from "vitest"; +import { createServerBackendFixture, removeServerBackendFixture } from "./fixture.ts"; + +describe("coding-agent server persistence", () => { + test("fails session listing when persisted JSONL is unreadable", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-corrupt-session", + cwd: fixture.cwd, + }); + await runtime.dispose(); + const files = await readdir(join(fixture.root, "sessions"), { recursive: true }); + const sessionFile = files.find((file) => file.endsWith(".jsonl")); + if (!sessionFile) throw new Error("Expected persisted JSONL session"); + await appendFile(join(fixture.root, "sessions", sessionFile), "{invalid json\n"); + + try { + await expect(fixture.backend.listSessions()).rejects.toThrow(/Failed to read coding-agent session/); + } finally { + await removeServerBackendFixture(fixture); + } + }); + test("surfaces corrupt session headers instead of omitting them", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ id: "server-corrupt-header", cwd: fixture.cwd }); + await runtime.dispose(); + const files = await readdir(join(fixture.root, "sessions"), { recursive: true }); + const sessionFile = files.find((file) => file.endsWith(".jsonl")); + if (!sessionFile) throw new Error("Expected persisted JSONL session"); + await writeFile(join(fixture.root, "sessions", sessionFile), "{invalid header\n"); + try { + await expect(fixture.backend.listSessions()).rejects.toThrow(/Invalid JSONL session file/); + await expect(fixture.backend.openSession("server-corrupt-header")).rejects.toThrow( + /Invalid JSONL session file/, + ); + } finally { + await removeServerBackendFixture(fixture); + } + }); + test("rejects invalid persisted creation timestamps", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ id: "server-invalid-created-at", cwd: fixture.cwd }); + await runtime.dispose(); + const files = await readdir(join(fixture.root, "sessions"), { recursive: true }); + const relative = files.find((file) => file.endsWith(".jsonl")); + if (!relative) throw new Error("Expected persisted JSONL session"); + const sessionFile = join(fixture.root, "sessions", relative); + const lines = (await readFile(sessionFile, "utf8")).split("\n"); + const header = JSON.parse(lines[0]!) as Record; + header.timestamp = "not-a-date"; + lines[0] = JSON.stringify(header); + await writeFile(sessionFile, lines.join("\n")); + try { + await expect(fixture.backend.listSessions()).rejects.toMatchObject({ + cause: expect.objectContaining({ message: expect.stringMatching(/Invalid session creation timestamp/) }), + }); + await expect(fixture.backend.openSession("server-invalid-created-at")).rejects.toThrow( + /Invalid session creation timestamp/, + ); + } finally { + await removeServerBackendFixture(fixture); + } + }); + test("rejects invalid persisted entry timestamps", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ id: "server-invalid-entry-time", cwd: fixture.cwd }); + await runtime.dispose(); + const files = await readdir(join(fixture.root, "sessions"), { recursive: true }); + const relative = files.find((file) => file.endsWith(".jsonl")); + if (!relative) throw new Error("Expected persisted JSONL session"); + const sessionFile = join(fixture.root, "sessions", relative); + const lines = (await readFile(sessionFile, "utf8")).split("\n"); + const entry = JSON.parse(lines[1]!) as Record; + entry.timestamp = "not-a-date"; + lines[1] = JSON.stringify(entry); + await writeFile(sessionFile, lines.join("\n")); + try { + await expect(fixture.backend.listSessions()).rejects.toMatchObject({ + cause: expect.objectContaining({ + message: expect.stringMatching(/Invalid timestamp on session entry/), + }), + }); + await expect(fixture.backend.openSession("server-invalid-entry-time")).rejects.toThrow( + /Invalid timestamp on session entry/, + ); + } finally { + await removeServerBackendFixture(fixture); + } + }); + test("removes a partially persisted session when creation fails", async () => { + const fixture = await createServerBackendFixture(); + const appendSpy = vi + .spyOn(NodeExecutionEnv.prototype, "appendFile") + .mockRejectedValueOnce(new Error("write failed")); + try { + await expect(fixture.backend.createSession({ id: "server-partial-create", cwd: fixture.cwd })).rejects.toThrow( + "write failed", + ); + expect(await fixture.backend.listSessions()).toEqual([]); + await expect(fixture.backend.openSession("server-partial-create")).rejects.toMatchObject({ + code: "not_found", + }); + } finally { + appendSpy.mockRestore(); + await removeServerBackendFixture(fixture); + } + }); + test("repairs a persisted thinking level that is unsupported by its model", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-interrupted-model-change", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + thinkingLevel: "high", + }); + await runtime.dispose(); + const files = await readdir(join(fixture.root, "sessions"), { recursive: true }); + const relative = files.find((file) => file.endsWith(".jsonl")); + if (!relative) throw new Error("Expected persisted JSONL session"); + const sessionFile = join(fixture.root, "sessions", relative); + const lines = (await readFile(sessionFile, "utf8")).trimEnd().split("\n"); + const previous = JSON.parse(lines.at(-1)!) as { id: string }; + await appendFile( + sessionFile, + `${JSON.stringify({ + type: "model_change", + id: randomUUID(), + parentId: previous.id, + timestamp: new Date().toISOString(), + provider: fixture.faux.provider.id, + modelId: "faux-plain", + })}\n`, + ); + try { + const reopened = await fixture.backend.openSession("server-interrupted-model-change"); + try { + expect(await reopened.snapshot()).toMatchObject({ + model: { provider: fixture.faux.provider.id, id: "faux-plain" }, + thinkingLevel: "off", + }); + } finally { + await reopened.dispose(); + } + const repaired = await fixture.backend.openSession("server-interrupted-model-change"); + try { + expect((await repaired.snapshot()).thinkingLevel).toBe("off"); + } finally { + await repaired.dispose(); + } + } finally { + await removeServerBackendFixture(fixture); + } + }); +}); diff --git a/packages/coding-agent/test/server/runtime.test.ts b/packages/coding-agent/test/server/runtime.test.ts new file mode 100644 index 000000000..51a30b2ce --- /dev/null +++ b/packages/coding-agent/test/server/runtime.test.ts @@ -0,0 +1,244 @@ +import { join } from "node:path"; +import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import { fauxAssistantMessage, fauxToolCall } from "@earendil-works/pi-ai"; +import type { TranscriptProgress } from "@earendil-works/pi-protocol"; +import lockfile, { type LockOptions } from "proper-lockfile"; +import { describe, expect, test, vi } from "vitest"; +import { CodingAgentServerBackend } from "../../src/server/backend.ts"; +import { abortableResponse, createServerBackendFixture, removeServerBackendFixture } from "./fixture.ts"; + +describe("coding-agent session runtime", () => { + test("propagates lock status check failures", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ id: "server-lock-check", cwd: fixture.cwd }); + await runtime.dispose(); + const checkSpy = vi.spyOn(lockfile, "check").mockRejectedValue(new Error("lock check failed")); + try { + await expect(fixture.backend.listSessions()).rejects.toThrow(/Failed to read coding-agent session/); + } finally { + checkSpy.mockRestore(); + await removeServerBackendFixture(fixture); + } + }); + test("terminates an active runtime without persisting after its session lock is compromised", async () => { + const fixture = await createServerBackendFixture(); + const realLock = lockfile.lock.bind(lockfile); + let compromise: LockOptions["onCompromised"]; + let forceRelease: (() => Promise) | undefined; + const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async (file, options) => { + compromise = options?.onCompromised; + forceRelease = await realLock(file, options); + return forceRelease; + }); + let runtime: Awaited> | undefined; + try { + runtime = await fixture.backend.createSession({ + id: "server-compromised-lock", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + }); + const pendingResponse = abortableResponse(); + fixture.faux.setResponses([pendingResponse.response]); + const prompt = runtime.prompt({ text: "start" }); + expect(runtime.getPhase()).toBe("turn"); + await pendingResponse.started; + await forceRelease?.(); + compromise?.(new Error("lock ownership lost")); + + await expect(prompt).rejects.toMatchObject({ code: "session_locked" }); + await expect(runtime.snapshot()).rejects.toMatchObject({ code: "session_locked" }); + await runtime.dispose(); + const secondBackend = await CodingAgentServerBackend.create({ + defaultCwd: fixture.cwd, + sessionRoot: join(fixture.root, "sessions"), + modelRuntime: fixture.modelRuntime, + settingsManager: fixture.settingsManager, + }); + const reopened = await secondBackend.openSession("server-compromised-lock"); + try { + expect((await reopened.snapshot()).transcript.some((item) => item.role === "assistant")).toBe(false); + } finally { + await reopened.dispose(); + } + } finally { + lockSpy.mockRestore(); + await runtime?.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("propagates lock release failure and allows disposal retry", async () => { + const fixture = await createServerBackendFixture(); + const realLock = lockfile.lock.bind(lockfile); + let releaseAttempts = 0; + const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async (file, options) => { + const release = await realLock(file, options); + return async () => { + releaseAttempts += 1; + if (releaseAttempts === 1) throw new Error("release failed"); + await release(); + }; + }); + let runtime: Awaited> | undefined; + try { + runtime = await fixture.backend.createSession({ id: "server-release-retry", cwd: fixture.cwd }); + await expect(runtime.dispose()).rejects.toThrow("release failed"); + await expect(runtime.dispose()).resolves.toBeUndefined(); + expect(releaseAttempts).toBe(2); + } finally { + lockSpy.mockRestore(); + await runtime?.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("exposes accepted steering text to every session snapshot", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-visible-steer", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + }); + try { + const pendingResponse = abortableResponse(); + fixture.faux.setResponses([pendingResponse.response]); + const prompt = runtime.prompt({ text: "start" }); + expect(runtime.getPhase()).toBe("turn"); + await pendingResponse.started; + await runtime.steer({ text: "adjust the approach" }); + expect(await runtime.snapshot()).toMatchObject({ + queuedSteerCount: 1, + queuedSteer: [{ role: "user", content: [{ type: "text", text: "adjust the approach" }] }], + }); + await runtime.abort(); + await prompt; + } finally { + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("rejects concurrent structural mutations instead of queueing them", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-concurrent-mutation", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + }); + try { + const changingModel = runtime.setModel({ provider: fixture.faux.provider.id, id: "faux-plain" }); + await expect(runtime.setThinking("medium")).rejects.toMatchObject({ code: "busy" }); + await expect(runtime.prompt({ text: "must not queue" })).rejects.toMatchObject({ code: "busy" }); + await expect(runtime.abort()).rejects.toMatchObject({ code: "busy" }); + await changingModel; + } finally { + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("terminates after a model change is only partially persisted", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-partial-model-change", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + thinkingLevel: "high", + }); + const originalAppendFile = NodeExecutionEnv.prototype.appendFile; + const appendSpy = vi.spyOn(NodeExecutionEnv.prototype, "appendFile"); + appendSpy.mockImplementationOnce(function (this: NodeExecutionEnv, ...args) { + return originalAppendFile.apply(this, args); + }); + appendSpy.mockRejectedValueOnce(new Error("thinking write failed")); + try { + await expect(runtime.setModel({ provider: fixture.faux.provider.id, id: "faux-plain" })).rejects.toMatchObject( + { + code: "invalid_request", + message: "Session failed to set model", + cause: expect.objectContaining({ message: expect.stringContaining("thinking write failed") }), + }, + ); + await expect(runtime.snapshot()).rejects.toMatchObject({ + code: "invalid_request", + message: "Session failed to set model", + }); + } finally { + appendSpy.mockRestore(); + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("streams live bash output into progress and the final snapshot", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-session-bash", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + thinkingLevel: "low", + }); + try { + fixture.faux.setResponses([ + fauxAssistantMessage( + fauxToolCall("bash", { command: "printf first; sleep 0.2; printf second" }, { id: "bash-call" }), + { stopReason: "toolUse" }, + ), + fauxAssistantMessage("done"), + ]); + const progress: TranscriptProgress[] = []; + runtime.subscribe((event) => { + if (event.type === "progress") progress.push(event.progress); + }); + await runtime.prompt({ text: "run it" }); + + const toolUpdates = progress.filter( + (event): event is Extract => + event.type === "item_updated" && event.item.role === "tool", + ); + expect( + toolUpdates.some((event) => + event.item.content.some((part) => part.type === "text" && part.text.includes("first")), + ), + ).toBe(true); + const final = await runtime.snapshot(); + expect(final.transcript.map((item) => item.role)).toEqual(["user", "assistant", "tool", "assistant"]); + const tool = final.transcript.find((item) => item.role === "tool"); + expect(tool).toMatchObject({ + toolCallId: "bash-call", + toolName: "bash", + input: { command: "printf first; sleep 0.2; printf second" }, + status: "complete", + isError: false, + }); + if (!tool || tool.role !== "tool") throw new Error("Expected tool transcript item"); + const output = tool.content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join(""); + expect(output).toContain("first"); + expect(output).toContain("second"); + } finally { + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); + test("terminates when executed tool input cannot be represented by the protocol", async () => { + const fixture = await createServerBackendFixture(); + const runtime = await fixture.backend.createSession({ + id: "server-invalid-tool-input", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + }); + try { + let projectionError: Error | undefined; + runtime.subscribe((event) => { + if (event.type === "error") projectionError = event.error; + }); + fixture.faux.setResponses([ + fauxAssistantMessage(fauxToolCall("read", { path: Number.NaN }), { stopReason: "toolUse" }), + ]); + await expect(runtime.prompt({ text: "trigger invalid input" })).rejects.toMatchObject({ + code: "invalid_request", + message: "Session protocol projection failed", + }); + expect(projectionError?.cause).toBeInstanceOf(TypeError); + await expect(runtime.snapshot()).rejects.toMatchObject({ code: "invalid_request" }); + } finally { + await runtime.dispose(); + await removeServerBackendFixture(fixture); + } + }); +}); diff --git a/packages/server/README.md b/packages/server/README.md index 3fa91e395..6d24fffee 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -1,18 +1,12 @@ # @earendil-works/pi-server -Experimental. This package is under active development and may change or be removed without notice. Its CLI, APIs, and behavior are not yet stable. +Experimental. This package is under active development and may change or be removed without notice. Its APIs and behavior are not yet stable. -Server package for pi. - -## CLI - -```bash -server --help -``` +Transport-neutral session server for pi. ## Session server core -The package also exports the new `PiServer` session server. This API is additive while the legacy child-process supervisor and `server` CLI are migrated. +The package exports the `PiServer` session server and listener composition APIs. ```ts import type { PiSessionBackend } from "@earendil-works/pi-server"; @@ -50,7 +44,3 @@ Custom transports can use `@earendil-works/pi-server/testing` for deterministic `@earendil-works/pi-ai` domain objects and `@earendil-works/pi-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 `pi-ai` unions are mapped exhaustively, and compile-time field manifests enumerate current `pi-ai` properties so additions require an explicit review. The protocol mirrors `pi-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. - -## Legacy server migration - -The existing IPC, supervisor, process management, persistence, and Radius modules remain available during migration. The new Unix session protocol supersedes the legacy socket framing and RPC proxy only after the coding-agent backend and CLI replacement have landed. Radius is presence and registration infrastructure, not a transport, and requires a separate integration with the new server lifecycle before the legacy supervisor can be removed. diff --git a/packages/server/package.json b/packages/server/package.json index a619eec3c..5d68ef386 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -18,13 +18,7 @@ "types": "./dist/transports/unix/index.d.ts", "import": "./dist/transports/unix/index.js" }, - "./legacy": { - "types": "./dist/legacy/index.d.ts", - "import": "./dist/legacy/index.js" - } - }, - "bin": { - "server": "./dist/legacy/cli.js" + "./package.json": "./package.json" }, "files": [ "dist", @@ -34,7 +28,7 @@ "scripts": { "clean": "shx rm -rf dist", "dev": "tsgo -p tsconfig.build.json --watch --preserveWatchOutput", - "build": "tsgo -p tsconfig.build.json && shx chmod +x dist/legacy/cli.js", + "build": "tsgo -p tsconfig.build.json", "test": "vitest --run", "typecheck": "tsgo -p tsconfig.test.json", "prepublishOnly": "npm run clean && npm run build" @@ -55,7 +49,6 @@ }, "dependencies": { "@earendil-works/pi-ai": "^0.83.0", - "@earendil-works/pi-coding-agent": "^0.83.0", "@earendil-works/pi-protocol": "^0.83.0" }, "devDependencies": { diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 46625268a..de4b0d895 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -1,5 +1,4 @@ export * from "./errors.ts"; -export * from "./legacy/index.ts"; export * from "./listener.ts"; export * from "./protocol.ts"; export * from "./server.ts"; diff --git a/packages/server/src/legacy/cli.ts b/packages/server/src/legacy/cli.ts deleted file mode 100644 index 738d416ed..000000000 --- a/packages/server/src/legacy/cli.ts +++ /dev/null @@ -1,161 +0,0 @@ -#!/usr/bin/env node -import { readFileSync } from "node:fs"; -import { createConnection } from "node:net"; -import { dirname, join } from "node:path"; -import { cwd } from "node:process"; -import { fileURLToPath } from "node:url"; -import type { RpcCommand, RpcExtensionUIResponse } from "@earendil-works/pi-coding-agent"; -import { getSocketPath } from "./config.ts"; -import { sendIpcRequest } from "./ipc/client.ts"; -import { encodeMessage } from "./ipc/protocol.ts"; -import { serve } from "./serve.ts"; - -const __filename = fileURLToPath(import.meta.url); -const __dirname = dirname(__filename); -const packageJson = JSON.parse(readFileSync(join(__dirname, "../package.json"), "utf-8")) as { - version: string; -}; - -function printHelp(): void { - console.log( - `server v${packageJson.version}\n\nUsage:\n server serve\n server list\n server spawn [--cwd ] [--label