sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(coding-agent): add server session backend

Browse source 
Add durable coding-agent session runtimes with locking, protocol projection, and JSONL persistence. Remove the obsolete legacy server implementation and dependency cycle.
This commit is contained in:
Christian Klotz 2026-07-31 21:02:44 +03:00
parent 3d264e85b4
commit 4abc08a8b3
No known key found for this signature in database
37 changed files with 2037 additions and 2014 deletions
Download patch file Download diff file Expand all files Collapse all files

5
package-lock.json generated
View file

@ -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"

4
package.json
View file

@ -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",

12
packages/agent/src/harness/agent-harness.ts
View file

@ -183,6 +183,7 @@ export class AgentHarness<
private readonly activeTasks = new Map<Promise<void>, TrackedTaskKind>();
private shutdownPromise?: Promise<void>;
private isShutdown = false;
private discardPendingWrites = false;
private pendingSessionWrites: PendingSessionWrite[] = [];
private model: Model<any>;
private thinkingLevel: ThinkingLevel;
@ -552,6 +553,10 @@ export class AgentHarness<
}
private async flushPendingSessionWrites(): Promise<void> {
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<void> {
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 = [];

27
packages/agent/test/harness/agent-harness.test.ts
View file

@ -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;

13
packages/coding-agent/install-lock/package-lock.json generated
View file

@ -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",

13
packages/coding-agent/npm-shrinkwrap.json generated
View file

@ -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",

7
packages/coding-agent/package.json
View file

@ -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",

211
packages/coding-agent/src/server/backend.ts Normal file
View file

@ -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<CodingAgentServerBackend> {
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<ModelMetadata[]> {
return this.models.list();
}
async listSessions(): Promise<SessionSummary[]> {
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<PiSessionRuntime> {
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<PiSessionRuntime> {
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<Api>,
thinkingLevel: ThinkingLevel,
lease: SessionLease,
): Promise<CodingAgentSessionRuntime> {
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<void> {
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";

101
packages/coding-agent/src/server/create-harness.ts Normal file
View file

@ -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<Api>;
thinkingLevel: ThinkingLevel;
env: NodeExecutionEnv;
streamOptions: AgentHarnessStreamOptions;
assertUsable(): void;
}
export function createServerHarness(options: CreateServerHarnessOptions): AgentHarness<ExecutionToolContext> {
let harness: AgentHarness<ExecutionToolContext>;
const tools: AgentHarnessTool<ExecutionToolContext>[] = [
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<ExecutionToolContext>({
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;
}

17
packages/coding-agent/src/server/errors.ts Normal file
View file

@ -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));
}

4
packages/coding-agent/src/server/index.ts Normal file
View file

@ -0,0 +1,4 @@
export {
CodingAgentServerBackend,
type CodingAgentServerBackendOptions,
} from "./backend.ts";

69
packages/coding-agent/src/server/model-catalog.ts Normal file
View file

@ -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<ModelMetadata[]> {
await this.modelRuntime.getAvailable();
return this.modelRuntime
.getModels()
.map((model) => toProtocolModelMetadata(model, this.modelRuntime.hasConfiguredAuth(model.provider)));
}
async resolve(reference?: ModelRef): Promise<Model<Api>> {
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<Api>, persisted?: ThinkingLevel): ThinkingLevel {
return persisted === undefined ? this.resolveThinkingLevel(model) : clampThinkingLevel(model, persisted);
}
resolveThinkingLevel(model: Model<Api>, 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;
}
}

111
packages/coding-agent/src/server/session-lock.ts Normal file
View file

@ -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<void>;
}
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<void> {
await mkdir(this.root, { recursive: true, mode: 0o700 });
}
async isLocked(sessionId: string): Promise<boolean> {
return lockfile.check(await this.ensureTarget(sessionId), { realpath: false, stale: STALE_MS });
}
async acquire(sessionId: string): Promise<SessionLease> {
const target = await this.ensureTarget(sessionId);
let compromised: Error | undefined;
const listeners = new Set<(error: Error) => void>();
let releaseLock: (() => Promise<void>) | 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<string> {
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<void> | undefined,
listeners: Set<(error: Error) => void>,
getCompromised: () => Error | undefined,
): SessionLease {
let released = false;
let releasing: Promise<void> | 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;
}
},
};
}
}

140
packages/coding-agent/src/server/session-persistence.ts Normal file
View file

@ -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<ThinkingLevel>(["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<StoredSession[]> {
return Promise.all((await this.repository.list()).map((metadata) => this.load(metadata)));
}
async find(sessionId: string): Promise<JsonlSessionMetadata | undefined> {
return (await this.repository.list()).find((metadata) => metadata.id === sessionId);
}
async create(options: { id: string; cwd: string }): Promise<Session> {
return this.repository.create(options);
}
async open(metadata: JsonlSessionMetadata): Promise<Session> {
return this.repository.open(metadata);
}
async delete(session: Session): Promise<void> {
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<StoredSession> {
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<SessionTreeEntry[]> {
const branch: SessionTreeEntry[] = [];
const visited = new Set<string>();
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;
}

275
packages/coding-agent/src/server/session-runtime.ts Normal file
View file

@ -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<Api>;
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<ExecutionToolContext>;
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<void>;
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<SessionSnapshot> {
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<void> {
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<void> {
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<void> {
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<void> {
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<void> {
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<void> {
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<void>): Promise<void> {
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;
}
}
}

304
packages/coding-agent/src/server/transcript/live.ts Normal file
View file

@ -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<ToolTranscriptItem, { status: "running" }>;
type FinishedToolTranscriptItem = Exclude<ToolTranscriptItem, { status: "running" }>;
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<typeof toProtocolUsage>[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<string, TranscriptItem>();
private readonly liveOrder: string[] = [];
private readonly toolInputs = new Map<string, { call: ToolCall; input: JsonValue; timestamp: number }>();
private queuedSteerItems: ReturnType<typeof projectUserMessage>[] = [];
revision = 0;
constructor(onPhase: (phase: "idle" | "turn") => void) {
this.onPhase = onPhase;
}
get queuedSteer(): ReturnType<typeof projectUserMessage>[] {
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<typeof transcriptItemId>[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 };
}
}

102
packages/coding-agent/src/server/transcript/projection.ts Normal file
View file

@ -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<AssistantTranscriptItem, { status: "streaming" }>;
type FinishedToolTranscriptItem = Exclude<ToolTranscriptItem, { status: "running" }>;
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<AssistantTranscriptItem, { status: "streaming" }>;
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<string, ToolCall>();
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;
}

153
packages/coding-agent/test/server/backend.test.ts Normal file
View file

@ -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);
}
});
});

70
packages/coding-agent/test/server/fixture.ts Normal file
View file

@ -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<ReturnType<typeof createServerBackendFixture>>,
): Promise<void> {
await rm(fixture.root, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 });
}
export function abortableResponse(): { response: FauxResponseFactory; started: Promise<void> } {
let startedResolve: (() => void) | undefined;
const started = new Promise<void>((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<void>((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }));
}
return fauxAssistantMessage("aborted");
},
};
}

158
packages/coding-agent/test/server/persistence.test.ts Normal file
View file

@ -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<string, unknown>;
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<string, unknown>;
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);
}
});
});

244
packages/coding-agent/test/server/runtime.test.ts Normal file
View file

@ -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<void>) | undefined;
const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async (file, options) => {
compromise = options?.onCompromised;
forceRelease = await realLock(file, options);
return forceRelease;
});
let runtime: Awaited<ReturnType<typeof fixture.backend.createSession>> | 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<ReturnType<typeof fixture.backend.createSession>> | 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<TranscriptProgress, { type: "item_updated" }> =>
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);
}
});
});

16
packages/server/README.md
View file

@ -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.

11
packages/server/package.json
View file

@ -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": {

1
packages/server/src/index.ts
View file

@ -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";

161
packages/server/src/legacy/cli.ts
View file

@ -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 <path>] [--label <label>]\n server status <instance-id>\n server stop <instance-id>\n server rpc <instance-id> <json-command>\n server rpc-stream <instance-id>\n server --help\n server --version\n\nRPC stream stdin expects JSONL RpcCommand or extension_ui_response messages.`,
);
}
function printResponse(response: unknown): void {
console.log(JSON.stringify(response, null, 2));
}
function getFlagValue(args: string[], flag: string): string | undefined {
const index = args.indexOf(flag);
if (index === -1 || index + 1 >= args.length) {
return undefined;
}
return args[index + 1];
}
async function rpcStream(instanceId: string): Promise<void> {
const socket = createConnection(getSocketPath());
let stdinBuffer = "";
process.stdin.setEncoding("utf8");
await new Promise<void>((resolve, reject) => {
socket.once("connect", () => {
socket.write(encodeMessage({ type: "rpc_stream", instanceId }));
resolve();
});
socket.once("error", reject);
});
socket.on("data", (chunk: Buffer | string) => {
process.stdout.write(chunk.toString());
});
console.error(`connected to rpc stream ${instanceId}; send JSONL RpcCommand or extension_ui_response on stdin`);
socket.on("error", (error) => {
console.error(error instanceof Error ? error.message : String(error));
process.exit(1);
});
socket.on("end", () => {
process.exit(0);
});
process.stdin.on("data", (chunk: string) => {
stdinBuffer += chunk;
while (true) {
const newlineIndex = stdinBuffer.indexOf("\n");
if (newlineIndex === -1) {
return;
}
const line = stdinBuffer.slice(0, newlineIndex).trim();
stdinBuffer = stdinBuffer.slice(newlineIndex + 1);
if (!line) {
continue;
}
const parsed = JSON.parse(line) as RpcCommand | RpcExtensionUIResponse;
socket.write(encodeMessage(parsed));
}
});
}
async function main(): Promise<void> {
const args = process.argv.slice(2);
if (args.length === 0 || args[0] === "--help" || args[0] === "-h") {
printHelp();
process.exit(0);
}
if (args[0] === "--version" || args[0] === "-v") {
console.log(packageJson.version);
process.exit(0);
}
if (args[0] === "serve") {
await serve();
return;
}
if (args[0] === "list") {
printResponse(await sendIpcRequest({ type: "list" }));
return;
}
if (args[0] === "spawn") {
const spawnCwd = getFlagValue(args, "--cwd") ?? cwd();
const label = getFlagValue(args, "--label");
printResponse(await sendIpcRequest({ type: "spawn", cwd: spawnCwd, label }));
return;
}
if (args[0] === "status") {
const instanceId = args[1];
if (!instanceId) {
console.error("Usage: server status <instance-id>");
process.exit(1);
}
printResponse(await sendIpcRequest({ type: "status", instanceId }));
return;
}
if (args[0] === "stop") {
const instanceId = args[1];
if (!instanceId) {
console.error("Usage: server stop <instance-id>");
process.exit(1);
}
printResponse(await sendIpcRequest({ type: "stop", instanceId }));
return;
}
if (args[0] === "rpc") {
const instanceId = args[1];
const commandJson = args[2];
if (!instanceId || !commandJson) {
console.error("Usage: server rpc <instance-id> <json-command>");
process.exit(1);
}
printResponse(
await sendIpcRequest({
type: "rpc",
instanceId,
command: JSON.parse(commandJson),
}),
);
return;
}
if (args[0] === "rpc-stream") {
const instanceId = args[1];
if (!instanceId) {
console.error("Usage: server rpc-stream <instance-id>");
process.exit(1);
}
await rpcStream(instanceId);
return;
}
console.error(`Unknown command: ${args[0]}`);
printHelp();
process.exit(1);
}
await main();

69
packages/server/src/legacy/config.ts
View file

@ -1,69 +0,0 @@
import { existsSync, readFileSync } from "node:fs";
import { homedir } from "node:os";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
const CONFIG_DIR_NAME = ".pi";
const ENV_SERVER_DIR = "PI_SERVER_DIR";
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
/**
* Detect if we're running as a Bun compiled binary.
* Bun binaries have import.meta.url containing "$bunfs", "~BUN", or "%7EBUN" (Bun's virtual filesystem path)
*/
export const isBunBinary =
import.meta.url.includes("$bunfs") || import.meta.url.includes("~BUN") || import.meta.url.includes("%7EBUN");
interface PackageJson {
version?: string;
}
function getPackageJsonPath(): string {
let dir = __dirname;
while (dir !== dirname(dir)) {
const packageJsonPath = join(dir, "package.json");
if (existsSync(packageJsonPath)) {
return packageJsonPath;
}
dir = dirname(dir);
}
return join(__dirname, "package.json");
}
let pkg: PackageJson = {};
try {
pkg = JSON.parse(readFileSync(getPackageJsonPath(), "utf-8")) as PackageJson;
} catch (e: unknown) {
const err = e as NodeJS.ErrnoException;
if (err.code !== "ENOENT") throw e;
}
export const VERSION: string = pkg.version || "0.0.0";
export function getServerDir(): string {
const envDir = process.env[ENV_SERVER_DIR];
if (envDir) {
return envDir;
}
const piDir = process.env.PI_CONFIG_DIR || join(homedir(), CONFIG_DIR_NAME);
return join(piDir, "server");
}
export function getAuthPath(): string {
return join(getServerDir(), "auth.json");
}
export function getMachinePath(): string {
return join(getServerDir(), "machine.json");
}
export function getInstancesPath(): string {
return join(getServerDir(), "instances.json");
}
export function getSocketPath(): string {
return join(getServerDir(), "server.sock");
}

161
packages/server/src/legacy/handler.ts
View file

@ -1,161 +0,0 @@
import type {
AgentSessionEvent,
RpcCommand,
RpcExtensionUIRequest,
RpcExtensionUIResponse,
RpcResponse,
} from "@earendil-works/pi-coding-agent";
import type {
ErrorResponse,
InstanceSummary,
ListRequest,
ListResponse,
RpcBridgeResponse,
RpcReadyResponse,
RpcRequest,
RpcStreamRequest,
ServerRequest,
ServerResponse,
SpawnRequest,
SpawnResponse,
StatusRequest,
StatusResponse,
StopRequest,
StopResponse,
} from "./ipc/protocol.ts";
import { supervisor } from "./supervisor.ts";
import type { InstanceRecord } from "./types.ts";
function toInstanceSummary(instance: InstanceRecord): InstanceSummary {
return {
id: instance.id,
status: instance.status,
cwd: instance.cwd,
label: instance.label,
sessionId: instance.sessionId,
sessionFile: instance.sessionFile,
radiusPiId: instance.radiusPiId,
};
}
function unknownInstanceError(instanceId: string): ErrorResponse {
return {
type: "error",
ok: false,
error: `Unknown instance: ${instanceId}`,
};
}
// Overhead types
export async function handleIpcRequest(request: SpawnRequest): Promise<SpawnResponse | ErrorResponse>;
export async function handleIpcRequest(request: ListRequest): Promise<ListResponse | ErrorResponse>;
export async function handleIpcRequest(request: StopRequest): Promise<StopResponse | ErrorResponse>;
export async function handleIpcRequest(request: StatusRequest): Promise<StatusResponse | ErrorResponse>;
export async function handleIpcRequest(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse>;
export async function handleIpcRequest(request: RpcStreamRequest): Promise<RpcReadyResponse | ErrorResponse>;
export async function handleIpcRequest(request: ServerRequest): Promise<ServerResponse>;
export async function handleIpcRequest(request: ServerRequest): Promise<ServerResponse> {
switch (request.type) {
case "spawn": {
const instance = await supervisor.spawnInstance({
cwd: request.cwd,
label: request.label,
});
return {
type: "spawn_result",
ok: true,
instance: toInstanceSummary(instance),
};
}
case "list": {
return {
type: "list_result",
ok: true,
instances: supervisor.listInstances().map(toInstanceSummary),
};
}
case "status": {
const instance = supervisor.getInstance(request.instanceId);
if (!instance) {
return unknownInstanceError(request.instanceId);
}
return {
type: "status_result",
ok: true,
instance: toInstanceSummary(instance),
};
}
case "stop": {
const instance = await supervisor.stopInstance(request.instanceId);
if (!instance) {
return unknownInstanceError(request.instanceId);
}
return {
type: "stop_result",
ok: true,
instanceId: request.instanceId,
};
}
case "rpc": {
const response = await supervisor.handleRpc(request.instanceId, request.command);
if (!response) {
return unknownInstanceError(request.instanceId);
}
return {
type: "rpc_result",
ok: true,
response,
};
}
case "rpc_stream": {
const instance = supervisor.getInstance(request.instanceId);
if (!instance) {
return unknownInstanceError(request.instanceId);
}
return {
type: "rpc_ready",
ok: true,
instance: toInstanceSummary(instance),
};
}
}
}
export function openRpcStream(
instanceId: string,
onResponse: (response: RpcResponse) => void,
onSessionEvent: (event: AgentSessionEvent) => void,
onUiRequest: (request: RpcExtensionUIRequest) => void,
):
| {
handleRequest(request: RpcCommand | RpcExtensionUIResponse): Promise<void>;
close(): void;
}
| undefined {
const handle = supervisor.openRpcStream(instanceId, onSessionEvent, onUiRequest);
if (!handle) {
return undefined;
}
return {
async handleRequest(request): Promise<void> {
if (request.type === "extension_ui_response") {
handle.handleUiResponse(request);
return;
}
const response = await handle.handleRpc(request);
onResponse(response);
},
close(): void {
handle.close();
},
};
}

11
packages/server/src/legacy/index.ts
View file

@ -1,11 +0,0 @@
export * from "./config.ts";
export * from "./handler.ts";
export * from "./ipc/client.ts";
export * from "./ipc/protocol.ts";
export * from "./ipc/server.ts";
export * from "./radius.ts";
export * from "./rpc-process.ts";
export * from "./serve.ts";
export * from "./storage.ts";
export * from "./supervisor.ts";
export * from "./types.ts";

63
packages/server/src/legacy/ipc/client.ts
View file

@ -1,63 +0,0 @@
import { createConnection } from "node:net";
import { getSocketPath } from "../config.ts";
import { encodeMessage, parseResponseLine, type ServerRequest, type ServerResponse } from "./protocol.ts";
export async function sendIpcRequest(request: ServerRequest): Promise<ServerResponse> {
const socketPath = getSocketPath();
return new Promise<ServerResponse>((resolve, reject) => {
const socket = createConnection(socketPath);
let buffer = "";
let settled = false;
const cleanup = () => {
socket.removeAllListeners();
socket.end();
};
socket.on("connect", () => {
socket.write(encodeMessage(request));
});
socket.on("data", (chunk: Buffer | string) => {
buffer += chunk.toString();
const newlineIndex = buffer.indexOf("\n");
if (newlineIndex === -1) {
return;
}
const line = buffer.slice(0, newlineIndex).trim();
if (!line) {
return;
}
try {
settled = true;
resolve(parseResponseLine(line));
cleanup();
} catch (error) {
settled = true;
reject(error);
cleanup();
}
});
socket.on("error", (error) => {
if (settled) {
return;
}
settled = true;
reject(error);
cleanup();
});
socket.on("end", () => {
if (settled) {
return;
}
settled = true;
reject(new Error(`Server socket closed before a response was received: ${socketPath}`));
cleanup();
});
});
}

142
packages/server/src/legacy/ipc/protocol.ts
View file

@ -1,142 +0,0 @@
import type {
AgentSessionEvent,
RpcCommand,
RpcExtensionUIRequest,
RpcExtensionUIResponse,
RpcResponse,
} from "@earendil-works/pi-coding-agent";
import type { InstanceStatus } from "../types.ts";
export interface SpawnRequest {
type: "spawn";
cwd: string;
label?: string;
provider?: string;
model?: string;
}
export interface ListRequest {
type: "list";
}
export interface StopRequest {
type: "stop";
instanceId: string;
}
export interface StatusRequest {
type: "status";
instanceId: string;
}
export interface RpcRequest {
type: "rpc";
instanceId: string;
command: RpcCommand;
}
export interface RpcStreamRequest {
type: "rpc_stream";
instanceId: string;
}
export interface RequestMap {
spawn: SpawnRequest;
list: ListRequest;
stop: StopRequest;
status: StatusRequest;
rpc: RpcRequest;
rpc_stream: RpcStreamRequest;
}
export type ServerRequest = RequestMap[keyof RequestMap];
export interface InstanceSummary {
id: string;
status: InstanceStatus;
cwd: string;
label?: string;
sessionId?: string;
sessionFile?: string;
radiusPiId?: string;
}
export interface ResponseBase {
ok: boolean;
error?: string;
}
export interface SpawnResponse extends ResponseBase {
type: "spawn_result";
instance?: InstanceSummary;
}
export interface ListResponse extends ResponseBase {
type: "list_result";
instances?: InstanceSummary[];
}
export interface StopResponse extends ResponseBase {
type: "stop_result";
instanceId?: string;
}
export interface StatusResponse extends ResponseBase {
type: "status_result";
instance?: InstanceSummary;
}
export interface RpcBridgeResponse extends ResponseBase {
type: "rpc_result";
response: RpcResponse;
}
export interface RpcReadyResponse extends ResponseBase {
type: "rpc_ready";
instance?: InstanceSummary;
}
export interface ErrorResponse extends ResponseBase {
type: "error";
ok: false;
error: string;
}
export interface ResponseMap {
spawn: SpawnResponse;
list: ListResponse;
stop: StopResponse;
status: StatusResponse;
rpc: RpcBridgeResponse;
rpc_stream: RpcReadyResponse;
}
export type ServerResponse = ResponseMap[keyof ResponseMap] | ErrorResponse;
export type RpcClientMessage = RpcCommand | RpcExtensionUIResponse;
export type RpcServerMessage =
| RpcReadyResponse
| RpcResponse
| AgentSessionEvent
| RpcExtensionUIRequest
| ErrorResponse;
export type ProtocolMessage = ServerRequest | ServerResponse | RpcClientMessage | RpcServerMessage;
export type ResponseFor<T extends ServerRequest> = T extends { type: infer K }
? K extends keyof ResponseMap
? ResponseMap[K] | ErrorResponse
: ErrorResponse
: ErrorResponse;
export function encodeMessage(message: ProtocolMessage): string {
return `${JSON.stringify(message)}\n`;
}
export function parseRequestLine(line: string): ServerRequest {
const value = JSON.parse(line) as ServerRequest;
return value;
}
export function parseResponseLine(line: string): ServerResponse {
const value = JSON.parse(line) as ServerResponse;
return value;
}

209
packages/server/src/legacy/ipc/server.ts
View file

@ -1,209 +0,0 @@
import { existsSync, unlinkSync } from "node:fs";
import { createConnection, createServer, type Server } from "node:net";
import type { AgentSessionEvent, RpcExtensionUIRequest, RpcResponse } from "@earendil-works/pi-coding-agent";
import { getSocketPath } from "../config.ts";
import {
type ErrorResponse,
encodeMessage,
type ListRequest,
type ListResponse,
parseRequestLine,
type RpcBridgeResponse,
type RpcReadyResponse,
type RpcRequest,
type RpcStreamRequest,
type ServerRequest,
type ServerResponse,
type SpawnRequest,
type SpawnResponse,
type StatusRequest,
type StatusResponse,
type StopRequest,
type StopResponse,
} from "./protocol.ts";
export interface IpcRequestHandler {
(request: SpawnRequest): Promise<SpawnResponse | ErrorResponse> | SpawnResponse | ErrorResponse;
(request: ListRequest): Promise<ListResponse | ErrorResponse> | ListResponse | ErrorResponse;
(request: StopRequest): Promise<StopResponse | ErrorResponse> | StopResponse | ErrorResponse;
(request: StatusRequest): Promise<StatusResponse | ErrorResponse> | StatusResponse | ErrorResponse;
(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse> | RpcBridgeResponse | ErrorResponse;
(request: RpcStreamRequest): Promise<RpcReadyResponse | ErrorResponse> | RpcReadyResponse | ErrorResponse;
(request: ServerRequest): Promise<ServerResponse> | ServerResponse;
openRpcStream(
instanceId: string,
onResponse: (response: RpcResponse) => void,
onSessionEvent: (event: AgentSessionEvent) => void,
onUiRequest: (request: RpcExtensionUIRequest) => void,
):
| {
handleRequest(request: RpcRequest["command"] | { type: "extension_ui_response" }): Promise<void>;
close(): void;
}
| undefined;
}
export async function startIpcServer(handler: IpcRequestHandler): Promise<Server> {
const socketPath = getSocketPath();
await removeStaleSocketIfNeeded(socketPath);
const server = createServer((socket) => {
let buffer = "";
socket.on("data", async (chunk: Buffer | string) => {
buffer += chunk.toString();
const newlineIndex = buffer.indexOf("\n");
if (newlineIndex === -1) {
return;
}
const line = buffer.slice(0, newlineIndex).trim();
buffer = buffer.slice(newlineIndex + 1);
if (!line) {
return;
}
try {
const request = parseRequestLine(line);
if (request.type === "rpc_stream") {
const response = await handler(request);
if (!response.ok || response.type !== "rpc_ready" || !response.instance) {
socket.end(encodeMessage(response));
return;
}
socket.removeAllListeners("data");
const rpcStream = handler.openRpcStream(
request.instanceId,
(response) => {
socket.write(encodeMessage(response));
},
(event) => {
socket.write(encodeMessage(event));
},
(request) => {
socket.write(encodeMessage(request));
},
);
if (!rpcStream) {
socket.end(
encodeMessage({ type: "error", ok: false, error: `Unknown instance: ${request.instanceId}` }),
);
return;
}
socket.write(encodeMessage(response));
let rpcRequestQueue = Promise.resolve();
socket.on("data", (rpcChunk: Buffer | string) => {
buffer += rpcChunk.toString();
for (;;) {
const rpcNewlineIndex = buffer.indexOf("\n");
if (rpcNewlineIndex === -1) {
break;
}
const rpcLine = buffer.slice(0, rpcNewlineIndex).trim();
buffer = buffer.slice(rpcNewlineIndex + 1);
if (!rpcLine) {
continue;
}
rpcRequestQueue = rpcRequestQueue
.then(async () => {
try {
await rpcStream.handleRequest(JSON.parse(rpcLine));
} catch (rpcError: unknown) {
socket.write(
encodeMessage({
type: "error",
ok: false,
error: rpcError instanceof Error ? rpcError.message : String(rpcError),
}),
);
}
})
.catch((rpcError: Error) => {
socket.write(
encodeMessage({
type: "error",
ok: false,
error: rpcError.message,
}),
);
});
}
});
socket.once("close", () => rpcStream.close());
return;
}
const response = await handler(request);
socket.end(encodeMessage(response));
} catch (error: unknown) {
const response: ErrorResponse = {
type: "error",
ok: false,
error: error instanceof Error ? error.message : String(error),
};
socket.end(encodeMessage(response));
}
});
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(socketPath, () => {
server.off("error", reject);
resolve();
});
});
return server;
}
async function removeStaleSocketIfNeeded(socketPath: string): Promise<void> {
if (!existsSync(socketPath)) {
return;
}
const isLive = await isSocketLive(socketPath);
if (isLive) {
throw new Error(`server is already running: ${socketPath}`);
}
unlinkSync(socketPath);
}
async function isSocketLive(socketPath: string): Promise<boolean> {
return new Promise<boolean>((resolve, reject) => {
const socket = createConnection(socketPath);
let settled = false;
const finish = (result: boolean) => {
if (settled) {
return;
}
settled = true;
socket.removeAllListeners();
socket.destroy();
resolve(result);
};
socket.on("connect", () => finish(true));
socket.on("error", (error: NodeJS.ErrnoException) => {
if (error.code === "ECONNREFUSED" || error.code === "ENOENT") {
finish(false);
return;
}
if (error.code === "EPIPE" || error.code === "ECONNRESET") {
finish(false);
return;
}
if (settled) {
return;
}
settled = true;
socket.removeAllListeners();
socket.destroy();
reject(error);
});
});
}

440
packages/server/src/legacy/radius.ts
View file

@ -1,440 +0,0 @@
import { hostname, platform } from "node:os";
import type { OAuthCredential } from "@earendil-works/pi-ai";
import { readStoredCredential } from "@earendil-works/pi-coding-agent";
import { getServerDir, getSocketPath, VERSION } from "./config.ts";
import { loadMachine, saveMachine } from "./storage.ts";
import type { InstanceRecord, MachineRecord, RadiusRegistration } from "./types.ts";
const DEFAULT_RADIUS_URL = "https://radius.pi.dev/";
const DEFAULT_SERVER_BASE_PATH = "/v1/";
const NOT_FOUND_RETRY_THRESHOLD = 3;
const HEARTBEAT_BACKOFF_BASE_MS = 1_000;
const HEARTBEAT_BACKOFF_MAX_MS = 30_000;
const RADIUS_PROVIDER = "radius";
interface RegisterMachineResponse extends RadiusRegistration {
id: string;
}
interface RegisterPiResponse extends RadiusRegistration {
id: string;
}
interface RadiusPresenceCoordinator {
getLiveInstance(instanceId: string): InstanceRecord | undefined;
listLiveInstances(): InstanceRecord[];
updateInstance(instance: InstanceRecord): void;
}
interface PiHeartbeatState {
timer?: NodeJS.Timeout;
intervalMs: number;
radiusPiId: string;
consecutiveNotFoundCount: number;
transientFailureCount: number;
}
class RadiusHttpError extends Error {
readonly status: number;
constructor(status: number, message: string) {
super(message);
this.name = "RadiusHttpError";
this.status = status;
}
}
async function post<T>(path: string, body: unknown): Promise<T> {
const response = await fetch(new URL(path, getRadiusServerBaseUrl()), {
method: "POST",
headers: {
Authorization: `Bearer ${getRadiusAccessToken()}`,
"Content-Type": "application/json",
},
body: JSON.stringify(body),
});
if (!response.ok) {
throw new RadiusHttpError(response.status, `Radius request failed: ${response.status} ${await response.text()}`);
}
return (await response.json()) as T;
}
async function maybePost(path: string, body: unknown): Promise<void> {
const response = await fetch(new URL(path, getRadiusServerBaseUrl()), {
method: "POST",
headers: {
Authorization: `Bearer ${getRadiusAccessToken()}`,
"Content-Type": "application/json",
},
body: JSON.stringify(body),
});
if (!response.ok) {
throw new RadiusHttpError(response.status, `Radius request failed: ${response.status} ${await response.text()}`);
}
}
function isNotFoundError(error: unknown): error is RadiusHttpError {
return error instanceof RadiusHttpError && error.status === 404;
}
function computeBackoffDelayMs(failureCount: number): number {
const exponentialDelay = Math.min(
HEARTBEAT_BACKOFF_MAX_MS,
HEARTBEAT_BACKOFF_BASE_MS * 2 ** Math.max(0, failureCount - 1),
);
const jitterMs = Math.floor(Math.random() * Math.max(250, exponentialDelay / 4));
return Math.min(HEARTBEAT_BACKOFF_MAX_MS, exponentialDelay + jitterMs);
}
function formatRadiusError(error: unknown): string {
if (error instanceof RadiusHttpError) {
return `HTTP ${error.status}: ${error.message}`;
}
if (error instanceof Error) {
return error.message;
}
return String(error);
}
function logRadiusRetry(scope: string, action: string, delayMs: number, failureCount: number, error: unknown): void {
console.error(
`${scope} ${action} failed (attempt ${failureCount}); retrying in ${delayMs}ms: ${formatRadiusError(error)}`,
);
}
export function getRadiusUrl(): string {
return process.env.PI_RADIUS_URL || DEFAULT_RADIUS_URL;
}
export function getRadiusServerBaseUrl(): string {
const explicitUrl = process.env.PI_RADIUS_SERVER_URL;
if (explicitUrl) {
return explicitUrl;
}
return new URL(DEFAULT_SERVER_BASE_PATH, getRadiusUrl()).toString();
}
function getStoredRadiusCredential(): OAuthCredential | undefined {
const credential = readStoredCredential(RADIUS_PROVIDER);
return credential?.type === "oauth" ? credential : undefined;
}
export function getRadiusAccessToken(): string {
const storedCredential = getStoredRadiusCredential();
if (typeof storedCredential?.access === "string" && storedCredential.access) {
return storedCredential.access;
}
const apiKey = process.env.RADIUS_API_KEY;
if (apiKey) {
return apiKey;
}
throw new Error("Radius credentials are required in ~/.pi/agent/auth.json or RADIUS_API_KEY");
}
export function isRadiusEnabled(): boolean {
return !!getStoredRadiusCredential()?.access || !!process.env.RADIUS_API_KEY;
}
export class RadiusPresence {
private machineHeartbeatTimer?: NodeJS.Timeout;
private machineHeartbeatIntervalMs = 0;
private machineConsecutiveNotFoundCount = 0;
private machineTransientFailureCount = 0;
private readonly piHeartbeatStates = new Map<string, PiHeartbeatState>();
private machine?: MachineRecord;
private coordinator?: RadiusPresenceCoordinator;
setCoordinator(coordinator: RadiusPresenceCoordinator): void {
this.coordinator = coordinator;
}
async start(label?: string): Promise<MachineRecord | undefined> {
if (!isRadiusEnabled()) {
return undefined;
}
const registered = await this.registerMachine(label);
this.startMachineHeartbeat(registered.heartbeatIntervalMs);
return this.machine;
}
async stop(): Promise<void> {
if (this.machineHeartbeatTimer) {
clearTimeout(this.machineHeartbeatTimer);
this.machineHeartbeatTimer = undefined;
}
for (const [instanceId, state] of this.piHeartbeatStates) {
if (state.timer) {
clearTimeout(state.timer);
}
this.piHeartbeatStates.delete(instanceId);
}
if (!this.machine || !isRadiusEnabled()) {
return;
}
try {
await maybePost(`machines/${this.machine.id}/disconnect`, {});
} catch (error) {
if (!isNotFoundError(error)) {
throw error;
}
}
}
async registerPi(instance: InstanceRecord): Promise<InstanceRecord> {
if (!isRadiusEnabled()) {
return instance;
}
const machine = this.machine ?? loadMachine();
if (!machine) {
throw new Error("No registered machine available for Pi registration");
}
const registered = await post<RegisterPiResponse>("pis/register", {
machineId: machine.id,
label: instance.label,
cwd: instance.cwd,
hostname: hostname(),
pid: process.pid,
transport: "local-rpc",
capabilities: { rpc: true, relay: false, iroh: false },
sessionId: instance.sessionId,
});
const registeredInstance = { ...instance, radiusPiId: registered.id };
this.startPiHeartbeat(instance.id, registered.heartbeatIntervalMs, registered.id);
return registeredInstance;
}
async disconnectPi(instance: InstanceRecord): Promise<void> {
const state = this.piHeartbeatStates.get(instance.id);
if (state) {
if (state.timer) {
clearTimeout(state.timer);
}
this.piHeartbeatStates.delete(instance.id);
}
if (!isRadiusEnabled() || !instance.radiusPiId) {
return;
}
try {
await maybePost(`pis/${instance.radiusPiId}/disconnect`, {});
} catch (error) {
if (!isNotFoundError(error)) {
throw error;
}
}
}
private async registerMachine(label?: string): Promise<RegisterMachineResponse> {
const existingMachine = this.machine ?? loadMachine();
const registered = await post<RegisterMachineResponse>("machines/register", {
machineId: existingMachine?.id,
label,
hostname: hostname(),
platform: platform(),
arch: process.arch,
version: VERSION,
capabilities: { spawn: true, relay: false, iroh: false },
});
const timestamp = new Date().toISOString();
this.machine = {
id: registered.id,
createdAt: existingMachine?.createdAt ?? timestamp,
lastSeenAt: timestamp,
label,
};
saveMachine(this.machine);
this.machineConsecutiveNotFoundCount = 0;
this.machineTransientFailureCount = 0;
return registered;
}
private startMachineHeartbeat(intervalMs: number): void {
this.machineHeartbeatIntervalMs = intervalMs;
this.scheduleMachineHeartbeat(intervalMs);
}
private scheduleMachineHeartbeat(delayMs: number): void {
if (this.machineHeartbeatTimer) {
clearTimeout(this.machineHeartbeatTimer);
}
this.machineHeartbeatTimer = setTimeout(() => {
void this.heartbeatMachine();
}, delayMs);
}
private startPiHeartbeat(instanceId: string, intervalMs: number, radiusPiId: string): void {
const existingState = this.piHeartbeatStates.get(instanceId);
if (existingState?.timer) {
clearTimeout(existingState.timer);
}
const state: PiHeartbeatState = existingState ?? {
intervalMs,
radiusPiId,
consecutiveNotFoundCount: 0,
transientFailureCount: 0,
};
state.intervalMs = intervalMs;
state.radiusPiId = radiusPiId;
state.consecutiveNotFoundCount = 0;
state.transientFailureCount = 0;
this.piHeartbeatStates.set(instanceId, state);
this.schedulePiHeartbeat(instanceId, intervalMs);
}
private schedulePiHeartbeat(instanceId: string, delayMs: number): void {
const state = this.piHeartbeatStates.get(instanceId);
if (!state) {
return;
}
if (state.timer) {
clearTimeout(state.timer);
}
state.timer = setTimeout(() => {
void this.heartbeatPi(instanceId);
}, delayMs);
}
private async heartbeatMachine(): Promise<void> {
if (!this.machine || !isRadiusEnabled()) {
return;
}
try {
await maybePost(`machines/${this.machine.id}/heartbeat`, {
cwd: getServerDir(),
socketPath: getSocketPath(),
});
this.machineConsecutiveNotFoundCount = 0;
this.machineTransientFailureCount = 0;
this.scheduleMachineHeartbeat(this.machineHeartbeatIntervalMs);
} catch (error) {
if (!isNotFoundError(error)) {
this.machineTransientFailureCount += 1;
const delayMs = computeBackoffDelayMs(this.machineTransientFailureCount);
logRadiusRetry("Radius machine", "heartbeat", delayMs, this.machineTransientFailureCount, error);
this.scheduleMachineHeartbeat(delayMs);
return;
}
this.machineTransientFailureCount = 0;
this.machineConsecutiveNotFoundCount += 1;
if (this.machineConsecutiveNotFoundCount < NOT_FOUND_RETRY_THRESHOLD) {
this.scheduleMachineHeartbeat(this.machineHeartbeatIntervalMs);
return;
}
try {
await this.reRegisterMachineAndPis();
} catch (recoveryError) {
this.machineTransientFailureCount += 1;
const delayMs = computeBackoffDelayMs(this.machineTransientFailureCount);
logRadiusRetry(
"Radius machine",
"re-registration",
delayMs,
this.machineTransientFailureCount,
recoveryError,
);
this.scheduleMachineHeartbeat(delayMs);
}
}
}
private async heartbeatPi(instanceId: string): Promise<void> {
if (!isRadiusEnabled()) {
return;
}
const state = this.piHeartbeatStates.get(instanceId);
if (!state) {
return;
}
try {
await maybePost(`pis/${state.radiusPiId}/heartbeat`, {});
state.consecutiveNotFoundCount = 0;
state.transientFailureCount = 0;
this.schedulePiHeartbeat(instanceId, state.intervalMs);
} catch (error) {
if (!isNotFoundError(error)) {
state.transientFailureCount += 1;
const delayMs = computeBackoffDelayMs(state.transientFailureCount);
logRadiusRetry(`Radius Pi ${instanceId}`, "heartbeat", delayMs, state.transientFailureCount, error);
this.schedulePiHeartbeat(instanceId, delayMs);
return;
}
state.transientFailureCount = 0;
state.consecutiveNotFoundCount += 1;
if (state.consecutiveNotFoundCount < NOT_FOUND_RETRY_THRESHOLD) {
this.schedulePiHeartbeat(instanceId, state.intervalMs);
return;
}
try {
const recovered = await this.reRegisterPi(instanceId);
if (!recovered) {
const delayMs = computeBackoffDelayMs(1);
console.error(`Radius Pi ${instanceId} re-registration skipped; retrying in ${delayMs}ms`);
this.schedulePiHeartbeat(instanceId, delayMs);
}
} catch (recoveryError) {
state.transientFailureCount += 1;
const delayMs = computeBackoffDelayMs(state.transientFailureCount);
logRadiusRetry(
`Radius Pi ${instanceId}`,
"re-registration",
delayMs,
state.transientFailureCount,
recoveryError,
);
this.schedulePiHeartbeat(instanceId, delayMs);
}
}
}
private async reRegisterMachineAndPis(): Promise<void> {
const registered = await this.registerMachine(this.machine?.label);
this.startMachineHeartbeat(registered.heartbeatIntervalMs);
const instances = this.coordinator?.listLiveInstances() ?? [];
for (const instance of instances) {
try {
await this.reRegisterPi(instance.id);
} catch (error) {
console.error(`Radius Pi ${instance.id} re-registration failed: ${formatRadiusError(error)}`);
}
}
}
private async reRegisterPi(instanceId: string): Promise<boolean> {
const instance = this.coordinator?.getLiveInstance(instanceId);
if (!instance) {
const state = this.piHeartbeatStates.get(instanceId);
if (state) {
if (state.timer) {
clearTimeout(state.timer);
}
this.piHeartbeatStates.delete(instanceId);
}
return false;
}
if (!this.machine) {
await this.reRegisterMachineAndPis();
return true;
}
const registeredInstance = await this.registerPi(instance);
this.coordinator?.updateInstance(registeredInstance);
return true;
}
}
export const radiusPresence = new RadiusPresence();

201
packages/server/src/legacy/rpc-process.ts
View file

@ -1,201 +0,0 @@
import { type ChildProcess, spawn } from "node:child_process";
import { randomUUID } from "node:crypto";
import { createRequire } from "node:module";
import { dirname, join } from "node:path";
import type {
AgentSessionEvent,
RpcCommand,
RpcExtensionUIRequest,
RpcExtensionUIResponse,
RpcResponse,
} from "@earendil-works/pi-coding-agent";
import { isBunBinary } from "./config.ts";
interface PendingRequest {
resolve(response: RpcResponse): void;
reject(error: Error): void;
}
const require = createRequire(import.meta.url);
function toError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}
export class RpcProcessInstance {
readonly process: ChildProcess;
private exited = false;
private nextRequestId = 0;
private stdoutBuffer = "";
private stderrBuffer = "";
private readonly pendingRequests = new Map<string, PendingRequest>();
private readonly eventListeners = new Set<(event: AgentSessionEvent) => void>();
private readonly exitListeners = new Set<(error?: Error) => void>();
private uiRequestHandler: ((request: RpcExtensionUIRequest) => void) | undefined;
constructor(options: { cwd: string }) {
const rpcCommand = this.getSpawnCommand();
this.process = spawn(rpcCommand.command, rpcCommand.args, {
cwd: options.cwd,
env: process.env,
stdio: ["pipe", "pipe", "pipe"],
});
if (!this.process.stdin || !this.process.stdout) {
throw new Error("Failed to create RPC process stdio");
}
this.attachListeners();
}
private getSpawnCommand(): { command: string; args: string[] } {
if (isBunBinary) {
return {
command: join(dirname(process.execPath), process.platform === "win32" ? "pi.exe" : "pi"),
args: ["--mode", "rpc"],
};
}
return {
command: process.execPath,
args: [require.resolve("@earendil-works/pi-coding-agent/rpc-entry")],
};
}
private attachListeners(): void {
this.process.stdout?.setEncoding("utf8");
this.process.stdout?.on("data", (chunk: string) => {
this.stdoutBuffer += chunk;
while (true) {
const newlineIndex = this.stdoutBuffer.indexOf("\n");
if (newlineIndex === -1) {
break;
}
const line = this.stdoutBuffer.slice(0, newlineIndex).trim();
this.stdoutBuffer = this.stdoutBuffer.slice(newlineIndex + 1);
if (!line) {
continue;
}
this.handleLine(line);
}
});
this.process.stderr?.setEncoding("utf8");
this.process.stderr?.on("data", (chunk: string) => {
this.stderrBuffer += chunk;
});
this.process.once("error", (error) => {
this.exited = true;
const wrapped = new Error(`RPC process error: ${error.message}. Stderr: ${this.stderrBuffer}`);
this.rejectAllPending(wrapped);
this.notifyExit(wrapped);
});
this.process.once("exit", (code, signal) => {
this.exited = true;
const error = new Error(`RPC process exited (code=${code} signal=${signal}). Stderr: ${this.stderrBuffer}`);
this.rejectAllPending(error);
this.notifyExit(error);
});
}
private handleLine(line: string): void {
const parsed = JSON.parse(line) as { type?: string; id?: string };
switch (parsed.type) {
case "response": {
if (!parsed.id) {
return;
}
const pending = this.pendingRequests.get(parsed.id);
if (!pending) {
return;
}
this.pendingRequests.delete(parsed.id);
pending.resolve(parsed as RpcResponse);
return;
}
case "extension_ui_request": {
this.uiRequestHandler?.(parsed as RpcExtensionUIRequest);
return;
}
default: {
for (const listener of this.eventListeners) {
listener(parsed as AgentSessionEvent);
}
}
}
}
private rejectAllPending(error: Error): void {
for (const [id, pending] of this.pendingRequests) {
this.pendingRequests.delete(id);
pending.reject(error);
}
}
private notifyExit(error?: Error): void {
for (const listener of this.exitListeners) {
listener(error);
}
}
send(command: RpcCommand): Promise<RpcResponse> {
if (this.exited) {
throw new Error(`RPC process is not running. Stderr: ${this.stderrBuffer}`);
}
const id = command.id ?? `server_${++this.nextRequestId}_${randomUUID()}`;
const fullCommand = { ...command, id };
return new Promise<RpcResponse>((resolve, reject) => {
this.pendingRequests.set(id, { resolve, reject });
this.process.stdin?.write(`${JSON.stringify(fullCommand)}\n`, (error) => {
if (!error) {
return;
}
this.pendingRequests.delete(id);
reject(toError(error));
});
});
}
handleUiResponse(response: RpcExtensionUIResponse): void {
if (this.exited) {
return;
}
this.process.stdin?.write(`${JSON.stringify(response)}\n`);
}
setUiRequestHandler(handler?: (request: RpcExtensionUIRequest) => void): void {
this.uiRequestHandler = handler;
}
onEvent(listener: (event: AgentSessionEvent) => void): () => void {
this.eventListeners.add(listener);
return () => {
this.eventListeners.delete(listener);
};
}
onExit(listener: (error?: Error) => void): () => void {
this.exitListeners.add(listener);
return () => {
this.exitListeners.delete(listener);
};
}
async dispose(): Promise<void> {
this.uiRequestHandler = undefined;
this.rejectAllPending(new Error("RPC process disposed"));
if (this.exited) {
return;
}
this.process.kill("SIGTERM");
await new Promise<void>((resolve) => {
this.process.once("exit", () => resolve());
});
}
}
export function createRpcProcessInstance(options: { cwd: string }): RpcProcessInstance {
return new RpcProcessInstance(options);
}

77
packages/server/src/legacy/serve.ts
View file

@ -1,77 +0,0 @@
import { existsSync, mkdirSync, unlinkSync } from "node:fs";
import { dirname } from "node:path";
import { getSocketPath } from "./config.ts";
import { handleIpcRequest, openRpcStream } from "./handler.ts";
import { startIpcServer } from "./ipc/server.ts";
import { getRadiusServerBaseUrl, isRadiusEnabled, radiusPresence } from "./radius.ts";
import { supervisor } from "./supervisor.ts";
export async function serve(): Promise<void> {
const socketPath = getSocketPath();
mkdirSync(dirname(socketPath), { recursive: true });
const server = await startIpcServer(
Object.assign(handleIpcRequest, {
openRpcStream,
}),
);
try {
await supervisor.recoverAfterRestart();
if (isRadiusEnabled()) {
const machine = await radiusPresence.start();
console.log(`radius integration enabled: ${socketPath} -> ${getRadiusServerBaseUrl()}`);
if (machine) {
console.log(`radius machine id: ${machine.id}`);
}
} else {
console.log("radius integration disabled: login radius in ~/.pi/agent/auth.json or set RADIUS_API_KEY");
}
} catch (error) {
server.close();
if (existsSync(socketPath)) {
unlinkSync(socketPath);
}
throw error;
}
console.log(`server listening on ${socketPath}`);
let shutdownPromise: Promise<void> | undefined;
const shutdown = async (exitCode: number) => {
if (shutdownPromise) {
await shutdownPromise;
process.exit(exitCode);
}
shutdownPromise = (async () => {
server.close();
await supervisor.shutdown();
await radiusPresence.stop();
if (existsSync(socketPath)) {
unlinkSync(socketPath);
}
})();
await shutdownPromise;
process.exit(exitCode);
};
process.on("SIGINT", () => {
void shutdown(0);
});
process.on("SIGTERM", () => {
void shutdown(0);
});
process.on("uncaughtException", (error) => {
console.error(error);
void shutdown(1);
});
process.on("unhandledRejection", (reason) => {
console.error(reason);
void shutdown(1);
});
await new Promise<void>(() => {
// Keep the process alive until a signal or fatal error triggers shutdown.
});
}

70
packages/server/src/legacy/storage.ts
View file

@ -1,70 +0,0 @@
import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { getInstancesPath, getMachinePath, getServerDir } from "./config.ts";
import type { InstanceRecord, MachineRecord } from "./types.ts";
function ensureServerDir(): void {
const serverDir = getServerDir();
if (!existsSync(serverDir)) {
mkdirSync(serverDir, { recursive: true });
}
}
export function loadMachine(): MachineRecord | undefined {
const machinePath = getMachinePath();
if (!existsSync(machinePath)) {
return undefined;
}
const data = readFileSync(machinePath, "utf-8");
return JSON.parse(data) as MachineRecord;
}
export function saveMachine(machine: MachineRecord): void {
ensureServerDir();
writeFileSync(getMachinePath(), JSON.stringify(machine, null, 2));
}
export function deleteMachine(): void {
const machinePath = getMachinePath();
if (!existsSync(machinePath)) {
return;
}
rmSync(machinePath);
}
export function loadInstances(): InstanceRecord[] {
const instancesPath = getInstancesPath();
if (!existsSync(instancesPath)) {
return [];
}
const data = readFileSync(instancesPath, "utf-8");
return JSON.parse(data) as InstanceRecord[];
}
export function saveInstances(instances: InstanceRecord[]): void {
ensureServerDir();
writeFileSync(getInstancesPath(), JSON.stringify(instances, null, 2));
}
export function getInstance(instanceId: string): InstanceRecord | undefined {
return loadInstances().find((instance) => instance.id === instanceId);
}
export function upsertInstance(instance: InstanceRecord): void {
const instances = loadInstances();
const index = instances.findIndex((existing) => existing.id === instance.id);
if (index === -1) {
instances.push(instance);
saveInstances(instances);
return;
}
instances[index] = instance;
saveInstances(instances);
}
export function removeInstance(instanceId: string): void {
const instances = loadInstances().filter((instance) => instance.id !== instanceId);
saveInstances(instances);
}

354
packages/server/src/legacy/supervisor.ts
View file

@ -1,354 +0,0 @@
import { randomUUID } from "node:crypto";
import type {
AgentSessionEvent,
AgentSessionEventListener,
RpcCommand,
RpcExtensionUIRequest,
RpcExtensionUIResponse,
RpcResponse,
} from "@earendil-works/pi-coding-agent";
import { radiusPresence } from "./radius.ts";
import { createRpcProcessInstance, type RpcProcessInstance } from "./rpc-process.ts";
import { getInstance, loadInstances, removeInstance, saveInstances, upsertInstance } from "./storage.ts";
import type { InstanceRecord, InstanceStatus } from "./types.ts";
interface LiveInstanceResources {
rpcProcess?: RpcProcessInstance;
radiusPiId?: string;
sessionId?: string;
}
interface LiveInstance {
record: InstanceRecord;
resources: LiveInstanceResources;
subscribers: Set<AgentSessionEventListener>;
onUiRequest?: (request: RpcExtensionUIRequest) => void;
unsubscribeEvents?: () => void;
unsubscribeExit?: () => void;
}
function cloneInstance(record: InstanceRecord): InstanceRecord {
return { ...record };
}
// Only refresh persisted session metadata after commands that can plausibly change
// the instance identity/details we store in instances.json. Most RPCs mutate transient
// runtime state only, so forcing a follow-up get_state after every command is wasted IO.
//
// - new_session / switch_session / fork / clone can change sessionId/sessionFile
// - set_session_name changes a persisted session detail we may want reflected externally
// - prompt can materialize or advance persisted session state after the child processes it
const SESSION_METADATA_COMMANDS: ReadonlySet<RpcCommand["type"]> = new Set([
"new_session",
"switch_session",
"fork",
"clone",
"set_session_name",
"prompt",
]);
function shouldRefreshSessionMetadata(command: RpcCommand): boolean {
return SESSION_METADATA_COMMANDS.has(command.type);
}
function isGetStateSuccess(
response: RpcResponse,
): response is Extract<
RpcResponse,
{ success: true; command: "get_state"; data: { sessionId: string; sessionFile?: string } }
> {
return response.success === true && response.command === "get_state" && "data" in response;
}
export class ServerSupervisor {
private readonly liveInstances = new Map<string, LiveInstance>();
private setStatus(live: LiveInstance, status: InstanceStatus): void {
live.record = {
...live.record,
status,
lastSeenAt: new Date().toISOString(),
};
upsertInstance(live.record);
}
private updateRecord(live: LiveInstance, updates: Partial<InstanceRecord>): void {
live.record = {
...live.record,
...updates,
lastSeenAt: new Date().toISOString(),
};
if (updates.radiusPiId !== undefined) {
live.resources.radiusPiId = updates.radiusPiId;
}
if (updates.sessionId !== undefined) {
live.resources.sessionId = updates.sessionId;
}
upsertInstance(live.record);
}
private clearBindings(live: LiveInstance): void {
live.unsubscribeEvents?.();
live.unsubscribeExit?.();
live.unsubscribeEvents = undefined;
live.unsubscribeExit = undefined;
live.onUiRequest = undefined;
live.resources.rpcProcess?.setUiRequestHandler(undefined);
}
private bindRpcProcess(live: LiveInstance, rpcProcess: RpcProcessInstance): void {
this.clearBindings(live);
live.resources.rpcProcess = rpcProcess;
live.unsubscribeEvents = rpcProcess.onEvent((event) => {
for (const subscriber of live.subscribers) {
subscriber(event);
}
});
live.unsubscribeExit = rpcProcess.onExit((error) => {
void this.handleUnexpectedRpcExit(live, error);
});
rpcProcess.setUiRequestHandler((request) => {
live.onUiRequest?.(request);
});
}
private async handleUnexpectedRpcExit(live: LiveInstance, _error?: Error): Promise<void> {
if (this.liveInstances.get(live.record.id) !== live) {
return;
}
if (live.record.status === "stopping" || live.record.status === "stopped") {
return;
}
this.setStatus(live, "error");
this.clearBindings(live);
live.resources.rpcProcess = undefined;
if (live.resources.radiusPiId) {
try {
await radiusPresence.disconnectPi(live.record);
this.updateRecord(live, { radiusPiId: undefined });
} catch (error) {
console.error(`Failed to disconnect Radius Pi ${live.record.id}: ${String(error)}`);
}
}
this.liveInstances.delete(live.record.id);
}
private getRpcProcess(live: LiveInstance): RpcProcessInstance | undefined {
return live.resources.rpcProcess;
}
private async syncInstanceRecord(live: LiveInstance): Promise<void> {
const rpcProcess = this.getRpcProcess(live);
if (!rpcProcess) {
this.updateRecord(live, {});
return;
}
const response = await rpcProcess.send({ type: "get_state" });
if (!isGetStateSuccess(response)) {
this.updateRecord(live, {});
return;
}
this.updateRecord(live, {
sessionId: response.data.sessionId,
sessionFile: response.data.sessionFile,
});
}
private async cleanupAcquiredResources(live: LiveInstance): Promise<void> {
const rpcProcess = live.resources.rpcProcess;
this.clearBindings(live);
if (live.resources.radiusPiId) {
await radiusPresence.disconnectPi(live.record);
live.resources.radiusPiId = undefined;
live.record = {
...live.record,
radiusPiId: undefined,
lastSeenAt: new Date().toISOString(),
};
}
live.resources.sessionId = undefined;
if (rpcProcess) {
live.resources.rpcProcess = undefined;
await rpcProcess.dispose();
}
}
private async failSpawn(live: LiveInstance, error: unknown): Promise<never> {
this.setStatus(live, "error");
try {
await this.cleanupAcquiredResources(live);
} finally {
this.setStatus(live, "stopped");
this.liveInstances.delete(live.record.id);
}
throw error;
}
updateInstance(instance: InstanceRecord): void {
const live = this.liveInstances.get(instance.id);
if (live) {
live.record = instance;
live.resources.radiusPiId = instance.radiusPiId;
live.resources.sessionId = instance.sessionId;
}
upsertInstance(instance);
}
openRpcStream(
instanceId: string,
onEvent: (event: AgentSessionEvent) => void,
onUiRequest: (request: RpcExtensionUIRequest) => void,
):
| {
handleRpc(command: RpcCommand): Promise<RpcResponse>;
handleUiResponse(response: RpcExtensionUIResponse): void;
close(): void;
}
| undefined {
const live = this.liveInstances.get(instanceId);
const rpcProcess = live ? this.getRpcProcess(live) : undefined;
if (!live || !rpcProcess) {
return undefined;
}
live.subscribers.add(onEvent);
live.onUiRequest = onUiRequest;
return {
handleRpc: async (command) => {
const response = await rpcProcess.send(command);
if (shouldRefreshSessionMetadata(command)) {
await this.syncInstanceRecord(live);
}
return response;
},
handleUiResponse: (response) => {
rpcProcess.handleUiResponse(response);
},
close: () => {
if (live.onUiRequest === onUiRequest) {
live.onUiRequest = undefined;
}
live.subscribers.delete(onEvent);
},
};
}
getLiveInstance(instanceId: string): InstanceRecord | undefined {
const live = this.liveInstances.get(instanceId);
return live ? cloneInstance(live.record) : undefined;
}
listLiveInstances(): InstanceRecord[] {
return [...this.liveInstances.values()].map((live) => cloneInstance(live.record));
}
async recoverAfterRestart(): Promise<void> {
const recoveredAt = new Date().toISOString();
const instances = loadInstances().map((instance) => ({
...instance,
status: instance.status === "online" || instance.status === "starting" ? "stopped" : instance.status,
lastSeenAt: recoveredAt,
}));
for (const instance of instances) {
await radiusPresence.disconnectPi(instance);
}
saveInstances(instances);
}
listInstances(): InstanceRecord[] {
return loadInstances().map(cloneInstance);
}
getInstance(instanceId: string): InstanceRecord | undefined {
const live = this.liveInstances.get(instanceId);
if (live) {
return cloneInstance(live.record);
}
const stored = getInstance(instanceId);
return stored ? cloneInstance(stored) : undefined;
}
async spawnInstance(options: { cwd: string; label?: string }): Promise<InstanceRecord> {
const now = new Date().toISOString();
const live: LiveInstance = {
record: {
id: randomUUID(),
status: "starting",
cwd: options.cwd,
createdAt: now,
lastSeenAt: now,
label: options.label,
},
resources: {},
subscribers: new Set(),
};
this.liveInstances.set(live.record.id, live);
upsertInstance(live.record);
try {
const rpcProcess = createRpcProcessInstance({ cwd: options.cwd });
this.bindRpcProcess(live, rpcProcess);
await this.syncInstanceRecord(live);
const registeredRecord = await radiusPresence.registerPi(live.record);
this.updateRecord(live, { radiusPiId: registeredRecord.radiusPiId });
this.setStatus(live, "online");
return cloneInstance(live.record);
} catch (error) {
return await this.failSpawn(live, error);
}
}
async stopInstance(instanceId: string): Promise<InstanceRecord | undefined> {
const live = this.liveInstances.get(instanceId);
if (!live) {
return undefined;
}
this.setStatus(live, "stopping");
try {
await this.cleanupAcquiredResources(live);
} finally {
live.record = {
...live.record,
status: "stopped",
lastSeenAt: new Date().toISOString(),
};
this.liveInstances.delete(instanceId);
removeInstance(instanceId);
}
return cloneInstance(live.record);
}
async handleRpc(instanceId: string, command: RpcCommand): Promise<RpcResponse | undefined> {
const live = this.liveInstances.get(instanceId);
const rpcProcess = live ? this.getRpcProcess(live) : undefined;
if (!live || !rpcProcess) {
return undefined;
}
const response = await rpcProcess.send(command);
if (shouldRefreshSessionMetadata(command)) {
await this.syncInstanceRecord(live);
}
return response;
}
async shutdown(): Promise<void> {
for (const instanceId of [...this.liveInstances.keys()]) {
await this.stopInstance(instanceId);
}
}
}
export const supervisor = new ServerSupervisor();
radiusPresence.setCoordinator({
getLiveInstance(instanceId) {
return supervisor.getLiveInstance(instanceId);
},
listLiveInstances() {
return supervisor.listLiveInstances();
},
updateInstance(instance) {
supervisor.updateInstance(instance);
},
});

25
packages/server/src/legacy/types.ts
View file

@ -1,25 +0,0 @@
export type InstanceStatus = "starting" | "online" | "stopping" | "stopped" | "error";
export interface MachineRecord {
id: string;
createdAt: string;
lastSeenAt?: string;
label?: string;
}
export interface RadiusRegistration {
heartbeatIntervalMs: number;
expiresInMs: number;
}
export interface InstanceRecord {
id: string;
status: InstanceStatus;
cwd: string;
createdAt: string;
lastSeenAt?: string;
label?: string;
sessionId?: string;
sessionFile?: string;
radiusPiId?: string;
}