From 1ceb9ea488ab9db046b428e448fdf7ca8de7117f Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Wed, 29 Jul 2026 14:59:56 +0300 Subject: [PATCH] fix(coding-agent): harden experimental client-server CLI Make connection, protocol, storage, and lock failures fail fast. Align the remote TUI's spacing, activity indicator, and exit behavior with legacy mode. --- packages/agent/CHANGELOG.md | 2 +- packages/agent/src/harness/agent-harness.ts | 18 +++ packages/client/src/index.ts | 9 +- packages/client/test/client.test.ts | 31 +++++ packages/coding-agent/CHANGELOG.md | 2 +- .../src/experimental/client-server/backend.ts | 107 +++++++++++---- .../src/experimental/client-server/cli.ts | 15 +- .../client-server/client-controller.ts | 37 ++++- .../src/experimental/client-server/tui.ts | 128 +++++++++++++----- packages/coding-agent/src/main.ts | 3 + ...experimental-client-server-backend.test.ts | 126 ++++++++++++++++- .../test/experimental-client-tui.test.ts | 19 +++ .../experimental-client-view-model.test.ts | 45 ++++++ packages/protocol/README.md | 2 +- packages/protocol/src/index.ts | 18 ++- packages/protocol/test/protocol.test.ts | 16 +++ packages/server/README.md | 2 +- packages/server/src/connection.ts | 2 +- packages/server/src/core.ts | 52 +++++-- packages/server/src/server.ts | 27 ++-- packages/server/src/types.ts | 5 +- packages/server/src/unix-listener.ts | 21 ++- packages/server/test/conformance.test.ts | 18 ++- packages/server/test/server.test.ts | 27 ++++ packages/server/test/support.ts | 4 + packages/server/test/transport-smoke.test.ts | 52 ++++++- 26 files changed, 677 insertions(+), 111 deletions(-) create mode 100644 packages/coding-agent/test/experimental-client-tui.test.ts diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index a9a2bc8e5..cdae3d36e 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -5,7 +5,7 @@ ### Added -- Added `AgentHarness.getPhase()` for observing the current lifecycle phase. +- Added `AgentHarness.getPhase()` for observing the current lifecycle phase and `terminate()` for stopping a compromised run without further persistence. ## [0.83.0] - 2026-07-29 diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index c42c51275..e460f25f5 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -180,6 +180,7 @@ export class AgentHarness< private runAbortController?: AbortController; private runPromise?: Promise; private pendingSessionWrites: PendingSessionWrite[] = []; + private terminalError?: Error; private model: Model; private thinkingLevel: ThinkingLevel; private systemPrompt: AgentHarnessSystemPrompt | undefined; @@ -352,6 +353,7 @@ export class AgentHarness< } private async createTurnState(): Promise> { + if (this.terminalError) throw this.terminalError; const context = await this.session.buildContext(); const resources = this.getResources(); const sessionMetadata = await this.session.getMetadata(); @@ -510,6 +512,10 @@ export class AgentHarness< } private async flushPendingSessionWrites(): Promise { + if (this.terminalError) { + this.pendingSessionWrites = []; + throw this.terminalError; + } while (this.pendingSessionWrites.length > 0) { const write = this.pendingSessionWrites[0]!; if (write.type === "message") { @@ -536,6 +542,7 @@ export class AgentHarness< } private async handleAgentEvent(event: AgentEvent, signal?: AbortSignal): Promise { + if (this.terminalError) throw this.terminalError; if (event.type === "message_end") { await this.session.appendMessage(event.message); await this.emitAny(event, signal); @@ -1026,6 +1033,17 @@ export class AgentHarness< this.streamOptions = cloneStreamOptions(streamOptions); } + /** Permanently stop the harness without emitting or persisting failure events. */ + terminate(error: Error): void { + if (this.terminalError) return; + this.terminalError = error; + this.pendingSessionWrites = []; + this.steerQueue = []; + this.followUpQueue = []; + this.nextTurnQueue = []; + this.runAbortController?.abort(); + } + async abort(): Promise { const clearedSteer = [...this.steerQueue]; const clearedFollowUp = [...this.followUpQueue]; diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index e62395b88..62eab1e76 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -485,16 +485,17 @@ export class PiClient { private failConnection(error: Error): void { if (this.stateValue === "disconnected") return; - this.setConnectionState("disconnected", error); + const reject = this.connectReject; + const pending = [...this.pendingRequests.values()]; this.transport = undefined; this.decoder = undefined; - const reject = this.connectReject; this.connectResolve = undefined; this.connectReject = undefined; - reject?.(error); - for (const pending of this.pendingRequests.values()) pending.reject(error); this.pendingRequests.clear(); this.attachedSessionIds.clear(); + reject?.(error); + for (const request of pending) request.reject(error); + this.setConnectionState("disconnected", error); } private isCurrentConnection(connectionId: number): boolean { diff --git a/packages/client/test/client.test.ts b/packages/client/test/client.test.ts index def2175f1..7c2a11a8b 100644 --- a/packages/client/test/client.test.ts +++ b/packages/client/test/client.test.ts @@ -360,6 +360,37 @@ describe("PiClient", () => { expect(states).toEqual(["connecting", "connected", "disconnected", "connecting", "connected"]); }); + test("supports synchronous reconnect from a disconnection listener", async () => { + const first = new MemoryByteServer(); + const second = new MemoryByteServer(); + let connection = 0; + for (const server of [first, second]) { + server.onMessage((message) => { + if (message.type !== "hello") return; + server.send({ + type: "hello", + version: PROTOCOL_VERSION, + connectionId: `connection-${connection}`, + snapshot: { ...baseServerSnapshot, revision: connection }, + }); + }); + } + const client = new PiClient({ + token: "bearer-secret", + transportFactory: (handlers) => (connection++ === 0 ? first : second).connect(handlers), + }); + await client.connect(); + let reconnect: Promise | undefined; + client.onConnectionStateChange(({ state }) => { + if (state === "disconnected") reconnect = client.reconnect(); + }); + + first.close(); + expect(reconnect).toBeDefined(); + await expect(reconnect).resolves.toMatchObject({ revision: 2 }); + expect(client.connectionState).toBe("connected"); + }); + test("rejects pending requests on transport errors", async () => { const server = new MemoryByteServer(); const client = await connectClient(server); diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 8acef6609..4f087988a 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -5,7 +5,7 @@ ### Added - Added `--alt` to run interactive mode with the alternate-screen TUI ([#7304](https://github.com/earendil-works/pi/issues/7304)). -- Added an experimental `PI_EXPERIMENTAL=1` client/server CLI with durable server-owned sessions, authenticated Unix socket transport, and a remote-session TUI. +- Added an experimental `PI_EXPERIMENTAL=1` client/server CLI with durable server-owned sessions, authenticated Unix socket transport, and a remote-session TUI with legacy message spacing, working activity, and double-Ctrl+C exit behavior. ### Fixed diff --git a/packages/coding-agent/src/experimental/client-server/backend.ts b/packages/coding-agent/src/experimental/client-server/backend.ts index 95c4430a4..72db8936b 100644 --- a/packages/coding-agent/src/experimental/client-server/backend.ts +++ b/packages/coding-agent/src/experimental/client-server/backend.ts @@ -92,6 +92,7 @@ export interface ExperimentalPiSessionBackendOptions { interface AcquiredLock { compromisedError(): Error | undefined; + onCompromised(listener: (error: Error) => void): () => void; release(): Promise; } @@ -107,20 +108,19 @@ function errorCode(error: unknown): string | undefined { : undefined; } -function toPiServerError(error: unknown): PiServerError { +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); } - return new PiServerError("invalid_request", error.message); } - if (error instanceof SessionError) { - return new PiServerError(error.code === "not_found" ? "not_found" : "invalid_request", error.message); + if (error instanceof SessionError && error.code === "not_found") { + return new PiServerError("not_found", error.message); } - const message = error instanceof Error ? error.message : String(error); - return new PiServerError("invalid_request", message); + return error instanceof Error ? error : new Error(String(error)); } function parseCreatedAt(value: string): number { @@ -223,14 +223,17 @@ export class ExperimentalPiSessionBackend implements PiSessionBackend { async listSessions(): Promise { const metadata = await this.repo.list(); - const summaries = await Promise.all( - metadata.map(async (entry): Promise => { + return Promise.all( + metadata.map(async (entry): Promise => { try { const session = await this.repo.open(entry); const branch = await getFullActiveBranch(session); const createdAt = parseCreatedAt(entry.createdAt); const state = readStoredSessionState(branch, createdAt); - if (!state.model || state.invalidThinkingLevel !== undefined) return undefined; + 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: entry.id, name: state.name, @@ -243,12 +246,11 @@ export class ExperimentalPiSessionBackend implements PiSessionBackend { attached: false, locked: await this.isLocked(entry.id), }; - } catch { - return undefined; + } catch (error) { + throw new Error(`Failed to read experimental session ${entry.id}`, { cause: error }); } }), ); - return summaries.filter((summary): summary is SessionSummary => summary !== undefined); } async createSession(options: CreateSessionOptions): Promise { @@ -405,16 +407,13 @@ export class ExperimentalPiSessionBackend implements PiSessionBackend { } private async isLocked(sessionId: string): Promise { - try { - return await lockfile.check(await this.ensureLockTarget(sessionId), { realpath: false, stale: 30_000 }); - } catch { - return false; - } + return lockfile.check(await this.ensureLockTarget(sessionId), { realpath: false, stale: 30_000 }); } private async acquireLock(sessionId: string): Promise { const target = await this.ensureLockTarget(sessionId); let compromised: Error | undefined; + const compromiseListeners = new Set<(error: Error) => void>(); let release: (() => Promise) | undefined; try { release = await lockfile.lock(target, { @@ -424,6 +423,7 @@ export class ExperimentalPiSessionBackend implements PiSessionBackend { retries: 0, onCompromised: (error) => { compromised = error; + for (const listener of compromiseListeners) listener(error); }, }); } catch (error) { @@ -433,12 +433,33 @@ export class ExperimentalPiSessionBackend implements PiSessionBackend { throw error; } let released = false; + let releasing: Promise | undefined; return { compromisedError: () => compromised, + onCompromised: (listener) => { + compromiseListeners.add(listener); + if (compromised) listener(compromised); + return () => compromiseListeners.delete(listener); + }, release: async () => { if (released) return; - released = true; - await release?.().catch(() => {}); + if (compromised) { + released = true; + compromiseListeners.clear(); + return; + } + if (releasing) return releasing; + const current = (async () => { + await release?.(); + released = true; + compromiseListeners.clear(); + })(); + releasing = current; + try { + await current; + } finally { + if (!released && releasing === current) releasing = undefined; + } }, }; } @@ -471,13 +492,14 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { private readonly env: NodeExecutionEnv; private readonly acquired: AcquiredLock; private readonly harness: PhasedAgentHarness; + private readonly unsubscribeLockCompromise: () => void; private readonly listeners = new Set<(event: PiSessionRuntimeEvent) => void>(); private readonly liveItems = new Map(); private readonly liveOrder: string[] = []; private readonly toolInputs = new Map(); private unsubscribeHarness: () => void; private revision = 0; - private updatedAt = Date.now(); + private lockCompromise?: Error; private queuedSteerCount = 0; private queuedSteer: ReturnType[] = []; private mutationInFlight = false; @@ -495,6 +517,7 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { createBashTool({ commandPrefix: this.settingsManager.getShellCommandPrefix(), prepare: async (execution) => { + this.assertUsable(); const metadata = await this.session.getMetadata(); execution.env.PI_SESSION_ID = metadata.id; if ("path" in metadata && typeof metadata.path === "string") @@ -520,6 +543,7 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { followUpMode: this.settingsManager.getFollowUpMode(), streamOptions: options.streamOptions, systemPrompt: async () => { + this.assertUsable(); const metadata = await this.session.getMetadata(); const cwd = "cwd" in metadata && typeof metadata.cwd === "string" ? metadata.cwd : this.env.cwd; return buildSystemPrompt({ @@ -546,6 +570,7 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { // Protocol progress is best-effort and must never fail the harness run. } }); + this.unsubscribeLockCompromise = this.acquired.onCompromised((error) => this.handleLockCompromise(error)); } getPhase(): SessionPhase { @@ -567,7 +592,7 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { name: stored.name, cwd: metadata.cwd, createdAt, - updatedAt: Math.max(stored.updatedAt, this.updatedAt), + updatedAt: stored.updatedAt, phase: this.getPhase(), model: { provider: this.harness.getModel().provider, id: this.harness.getModel().id }, thinkingLevel: this.harness.getThinkingLevel(), @@ -584,7 +609,9 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { this.assertIdle("prompt"); try { await this.harness.prompt(input.text); + this.throwIfLockCompromised(); } catch (error) { + this.throwIfLockCompromised(); throw toPiServerError(error); } } @@ -594,7 +621,9 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { if (this.getPhase() !== "turn") throw new PiServerError("busy", "Session is not accepting steering input"); try { await this.harness.steer(input.text); + this.throwIfLockCompromised(); } catch (error) { + this.throwIfLockCompromised(); throw toPiServerError(error); } } @@ -603,7 +632,9 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { this.assertUsable(); try { await this.harness.abort(); + this.throwIfLockCompromised(); } catch (error) { + this.throwIfLockCompromised(); throw toPiServerError(error); } } @@ -648,9 +679,10 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { async dispose(): Promise { if (this.disposePromise) return this.disposePromise; this.disposed = true; - this.disposePromise = (async () => { + const current = (async () => { try { this.unsubscribeHarness(); + this.unsubscribeLockCompromise(); this.listeners.clear(); if (this.harness.getPhase() !== "idle") { await this.harness.abort().catch(() => {}); @@ -661,14 +693,36 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { await this.acquired.release(); } })(); - return this.disposePromise; + this.disposePromise = current; + try { + await current; + } catch (error) { + if (this.disposePromise === current) this.disposePromise = undefined; + throw error; + } } private assertUsable(): void { + this.throwIfLockCompromised(); if (this.disposed) throw new PiServerError("invalid_request", "Session runtime is disposed"); - const compromised = this.acquired.compromisedError(); - if (compromised) + } + + private throwIfLockCompromised(): void { + const compromised = this.lockCompromise ?? this.acquired.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.mutationInFlight = true; + this.harness.terminate(error); + this.emit({ + type: "error", + error: new PiServerError("session_locked", `Session lock was compromised: ${error.message}`), + }); } private assertIdle(operation: string): void { @@ -683,7 +737,9 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { this.mutationInFlight = true; try { await mutation(); + this.throwIfLockCompromised(); } catch (error) { + this.throwIfLockCompromised(); throw toPiServerError(error); } finally { this.mutationInFlight = false; @@ -703,7 +759,6 @@ export class ExperimentalPiSessionRuntime implements PiSessionRuntime { private touch(): void { this.revision += 1; - this.updatedAt = Date.now(); } private emit(event: PiSessionRuntimeEvent): void { diff --git a/packages/coding-agent/src/experimental/client-server/cli.ts b/packages/coding-agent/src/experimental/client-server/cli.ts index 5e12a3e07..75910f295 100644 --- a/packages/coding-agent/src/experimental/client-server/cli.ts +++ b/packages/coding-agent/src/experimental/client-server/cli.ts @@ -107,22 +107,22 @@ async function chooseClientOnlySession( args: ExperimentalCliArgs, settingsManager: SettingsManager, model: ModelRef | undefined, -): Promise { +): Promise<{ session?: PiSessionClient; created: boolean }> { const sessions = await client.listSessions(); const labels = sessions.map(sessionLabel); const selected = await showStartupSelector(settingsManager, "Select remote session", [ { label: NEW_SESSION_LABEL, value: -1 }, ...labels.map((label, index) => ({ label, value: index })), ]); - if (selected === undefined) return undefined; + if (selected === undefined) return { created: false }; if (selected >= 0) { const summary = sessions[selected]; if (!summary) throw new Error("Selected remote session no longer exists"); - return client.attachSession(summary.id); + return { session: await client.attachSession(summary.id), created: false }; } const cwd = args.cwd ?? (await promptForRemoteCwd(settingsManager)); - if (!cwd) return undefined; - return client.createSession({ cwd, model, thinkingLevel: args.thinking }); + if (!cwd) return { created: false }; + return { session: await client.createSession({ cwd, model, thinkingLevel: args.thinking }), created: true }; } async function createController( @@ -145,8 +145,9 @@ async function createController( }); created = true; } else { - session = await chooseClientOnlySession(client, args, settingsManager, requestedModel); - created = session !== undefined && !snapshot.sessions.some((candidate) => candidate.id === session?.id); + const selection = await chooseClientOnlySession(client, args, settingsManager, requestedModel); + session = selection.session; + created = selection.created; } if (!session) { client.disconnect(); diff --git a/packages/coding-agent/src/experimental/client-server/client-controller.ts b/packages/coding-agent/src/experimental/client-server/client-controller.ts index 35463b8ae..cb8b68b8c 100644 --- a/packages/coding-agent/src/experimental/client-server/client-controller.ts +++ b/packages/coding-agent/src/experimental/client-server/client-controller.ts @@ -1,5 +1,6 @@ import type { PiClient, PiSessionClient, Unsubscribe } from "@earendil-works/pi-client"; import type { + JsonValue, ModelMetadata, ModelRef, ServerEvent, @@ -16,10 +17,29 @@ export interface ExperimentalClientView { transcript: readonly TranscriptItem[]; } +function isJsonValue(value: unknown): value is JsonValue { + if (value === null || typeof value === "boolean" || typeof value === "string") return true; + if (typeof value === "number") return Number.isFinite(value); + if (Array.isArray(value)) return value.every(isJsonValue); + if (typeof value !== "object" || Object.getPrototypeOf(value) !== Object.prototype) return false; + return Object.values(value).every(isJsonValue); +} + +function parsePartialToolInput(value: string): JsonValue { + try { + const parsed: unknown = JSON.parse(value); + if (isJsonValue(parsed)) return parsed; + } catch { + // Incomplete JSON is expected while tool arguments stream; preserve the raw prefix until it becomes valid. + } + return value; +} + export class ExperimentalSessionViewModel { private snapshotValue: SessionSnapshot | undefined; private readonly progressItems = new Map(); private readonly progressOrder: string[] = []; + private readonly toolCallBuffers = new Map(); get snapshot(): SessionSnapshot | undefined { return this.snapshotValue; @@ -29,6 +49,7 @@ export class ExperimentalSessionViewModel { this.snapshotValue = undefined; this.progressItems.clear(); this.progressOrder.length = 0; + this.toolCallBuffers.clear(); this.applySnapshot(snapshot); } @@ -39,6 +60,7 @@ export class ExperimentalSessionViewModel { this.snapshotValue = structuredClone(snapshot); this.progressItems.clear(); this.progressOrder.length = 0; + this.toolCallBuffers.clear(); } applyProgress(progress: TranscriptProgress): void { @@ -46,7 +68,14 @@ export class ExperimentalSessionViewModel { this.setProgressItem(progress.item); return; } - if (progress.type === "item_updated" || progress.type === "item_finished") { + if (progress.type === "item_updated") { + this.setProgressItem(progress.item); + return; + } + if (progress.type === "item_finished") { + for (const key of this.toolCallBuffers.keys()) { + if (key.startsWith(`${progress.item.id}:`)) this.toolCallBuffers.delete(key); + } this.setProgressItem(progress.item); return; } @@ -58,6 +87,12 @@ export class ExperimentalSessionViewModel { if (progress.kind === "thinking" && part.type === "thinking") { return { ...part, thinking: part.thinking + progress.delta }; } + if (progress.kind === "tool_call" && part.type === "tool_call") { + const key = `${progress.messageId}:${progress.contentIndex}`; + const buffer = (this.toolCallBuffers.get(key) ?? "") + progress.delta; + this.toolCallBuffers.set(key, buffer); + return { ...part, input: parsePartialToolInput(buffer) }; + } return structuredClone(part); }); this.setProgressItem({ ...item, content }); diff --git a/packages/coding-agent/src/experimental/client-server/tui.ts b/packages/coding-agent/src/experimental/client-server/tui.ts index bfb876f18..5e62f01f0 100644 --- a/packages/coding-agent/src/experimental/client-server/tui.ts +++ b/packages/coding-agent/src/experimental/client-server/tui.ts @@ -8,18 +8,32 @@ import type { ToolTranscriptItem, TranscriptItem, } from "@earendil-works/pi-protocol"; -import { Container, ProcessTerminal, setKeybindings, Text, TUI } from "@earendil-works/pi-tui"; +import { Container, ProcessTerminal, Spacer, setKeybindings, Text, TUI } from "@earendil-works/pi-tui"; import { getAgentDir } from "../../config.ts"; import { KeybindingsManager } from "../../core/keybindings.ts"; import type { SettingsManager } from "../../core/settings-manager.ts"; import { AssistantMessageComponent } from "../../modes/interactive/components/assistant-message.ts"; import { CustomEditor } from "../../modes/interactive/components/custom-editor.ts"; import { ExtensionSelectorComponent } from "../../modes/interactive/components/extension-selector.ts"; +import { IdleStatus, WorkingStatusIndicator } from "../../modes/interactive/components/status-indicator.ts"; import { ToolExecutionComponent } from "../../modes/interactive/components/tool-execution.ts"; import { UserMessageComponent } from "../../modes/interactive/components/user-message.ts"; import { getEditorTheme, initTheme, stopThemeWatcher, theme } from "../../modes/interactive/theme/theme.ts"; import type { ExperimentalClientController, ExperimentalClientView } from "./client-controller.ts"; +export class DoubleClearAction { + private lastTriggeredAt?: number; + + trigger(now = Date.now()): "clear" | "exit" { + if (this.lastTriggeredAt !== undefined && now - this.lastTriggeredAt < 500) { + this.lastTriggeredAt = undefined; + return "exit"; + } + this.lastTriggeredAt = now; + return "clear"; + } +} + const ZERO_USAGE: Usage = { input: 0, output: 0, @@ -93,12 +107,16 @@ export class ExperimentalClientTui { private readonly transcript = new Container(); private readonly editor: CustomEditor; private readonly footer = new Text("", 1, 0); - private readonly status = new Text("", 1, 0); + private readonly status = new Container(); private unsubscribeController?: () => void; private unsubscribeConnection?: () => void; private latestView?: ExperimentalClientView; private toolsExpanded = false; private hideThinking: boolean; + private readonly doubleClearAction = new DoubleClearAction(); + private readonly idleStatus = new IdleStatus(); + private workingStatus?: WorkingStatusIndicator; + private statusMessage?: Text; private shuttingDown = false; private finishRun?: () => void; @@ -107,35 +125,41 @@ export class ExperimentalClientTui { this.settingsManager = settingsManager; this.hideThinking = settingsManager.getHideThinkingBlock(); initTheme(settingsManager.getTheme(), true); - const keybindings = KeybindingsManager.create(); - setKeybindings(keybindings); - this.ui = new TUI(new ProcessTerminal(), settingsManager.getShowHardwareCursor(), getAgentDir()); - this.ui.setClearOnShrink(settingsManager.getClearOnShrink()); - this.editor = new CustomEditor(this.ui, getEditorTheme(), keybindings, { - paddingX: settingsManager.getEditorPaddingX(), - autocompleteMaxVisible: settingsManager.getAutocompleteMaxVisible(), - }); - this.editor.onSubmit = (text) => void this.handleSubmit(text); - this.editor.onAction("app.interrupt", () => void this.handleInterrupt()); - this.editor.onAction("app.clear", () => this.editor.setText("")); - this.editor.onAction("app.exit", () => void this.shutdown()); - this.editor.onAction("app.model.select", () => this.showModelSelector()); - this.editor.onAction("app.thinking.cycle", () => void this.cycleThinking()); - this.editor.onAction("app.thinking.toggle", () => { - this.hideThinking = !this.hideThinking; - this.rebuildTranscript(); - }); - this.editor.onAction("app.tools.expand", () => { - this.toolsExpanded = !this.toolsExpanded; - this.rebuildTranscript(); - }); - this.editor.onAction("app.session.new", () => void this.newSession()); - this.editor.onAction("app.session.resume", () => void this.showResumeSelector()); - this.ui.addChild(this.transcript); - this.ui.addChild(this.status); - this.ui.addChild(this.editor); - this.ui.addChild(this.footer); - this.ui.setFocus(this.editor); + try { + const keybindings = KeybindingsManager.create(); + setKeybindings(keybindings); + this.ui = new TUI(new ProcessTerminal(), settingsManager.getShowHardwareCursor(), getAgentDir()); + this.ui.setClearOnShrink(settingsManager.getClearOnShrink()); + this.editor = new CustomEditor(this.ui, getEditorTheme(), keybindings, { + paddingX: settingsManager.getEditorPaddingX(), + autocompleteMaxVisible: settingsManager.getAutocompleteMaxVisible(), + }); + this.editor.onSubmit = (text) => void this.handleSubmit(text); + this.editor.onAction("app.interrupt", () => void this.handleInterrupt()); + this.editor.onAction("app.clear", () => this.handleClear()); + this.editor.onAction("app.exit", () => void this.shutdown()); + this.editor.onAction("app.model.select", () => this.showModelSelector()); + this.editor.onAction("app.thinking.cycle", () => void this.cycleThinking()); + this.editor.onAction("app.thinking.toggle", () => { + this.hideThinking = !this.hideThinking; + this.rebuildTranscript(); + }); + this.editor.onAction("app.tools.expand", () => { + this.toolsExpanded = !this.toolsExpanded; + this.rebuildTranscript(); + }); + this.editor.onAction("app.session.new", () => void this.newSession()); + this.editor.onAction("app.session.resume", () => void this.showResumeSelector()); + this.ui.addChild(this.transcript); + this.ui.addChild(this.status); + this.ui.addChild(new Spacer(1)); + this.ui.addChild(this.editor); + this.ui.addChild(this.footer); + this.ui.setFocus(this.editor); + } catch (error) { + stopThemeWatcher(); + throw error; + } } async run(initialPrompt?: string): Promise { @@ -163,6 +187,7 @@ export class ExperimentalClientTui { this.unsubscribeController = undefined; this.unsubscribeConnection?.(); this.unsubscribeConnection = undefined; + this.clearWorkingStatus(); await this.ui.terminal.drainInput(1000).catch(() => {}); this.ui.stop(); stopThemeWatcher(); @@ -175,6 +200,7 @@ export class ExperimentalClientTui { this.latestView = view; this.rebuildTranscript(); const snapshot = view.snapshot; + this.syncWorkingStatus(); this.footer.setText( theme.fg( "dim", @@ -196,6 +222,7 @@ export class ExperimentalClientTui { const renderedToolIds = new Set(); for (const item of view.transcript) { if (item.role === "user") { + if (this.transcript.children.length > 0) this.transcript.addChild(new Spacer(1)); this.transcript.addChild( new UserMessageComponent(userText(item), undefined, this.settingsManager.getOutputPad()), ); @@ -267,6 +294,7 @@ export class ExperimentalClientTui { private async submitPrompt(text: string): Promise { try { this.setStatus(""); + this.showWorkingStatus(); await this.controller.submit(text); } catch (error) { this.showError(error); @@ -320,6 +348,14 @@ export class ExperimentalClientTui { } } + private handleClear(): void { + if (this.doubleClearAction.trigger() === "exit") { + void this.shutdown(); + return; + } + this.editor.setText(""); + } + private async handleInterrupt(): Promise { try { if (this.controller.phase === "idle") { @@ -470,10 +506,38 @@ export class ExperimentalClientTui { } private setStatus(message: string, error = false): void { - this.status.setText(message ? theme.fg(error ? "error" : "muted", message) : ""); + this.clearWorkingStatus(); + this.status.clear(); + this.statusMessage = message ? new Text(theme.fg(error ? "error" : "muted", message), 1, 0) : undefined; + if (this.statusMessage) this.status.addChild(this.statusMessage); + this.syncWorkingStatus(); this.ui.requestRender(); } + private syncWorkingStatus(): void { + const shouldShow = this.latestView?.snapshot.phase === "turn" && !this.statusMessage; + if (!shouldShow) { + this.clearWorkingStatus(true); + return; + } + this.showWorkingStatus(); + } + + private showWorkingStatus(): void { + if (this.workingStatus) return; + this.status.clear(); + this.workingStatus = new WorkingStatusIndicator(this.ui, "Working..."); + this.status.addChild(this.workingStatus); + } + + private clearWorkingStatus(preserveSpace = false): void { + if (!this.workingStatus) return; + this.workingStatus.dispose(); + this.status.removeChild(this.workingStatus); + this.workingStatus = undefined; + if (preserveSpace && this.ui.getClearOnShrink()) this.status.addChild(this.idleStatus); + } + private showError(error: unknown): void { this.setStatus(error instanceof Error ? error.message : String(error), true); } diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index bf5e19cfb..f2e65ad02 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -560,6 +560,9 @@ export async function main(args: string[], options?: MainOptions) { } if (options?.experimentalRunner) { + // Authentication and settings migrations must complete before the server runtime owns credentials. + runMigrations(cwd); + time("runMigrations"); await options.experimentalRunner(args); return; } diff --git a/packages/coding-agent/test/experimental-client-server-backend.test.ts b/packages/coding-agent/test/experimental-client-server-backend.test.ts index 77d6905a8..c54f42dd7 100644 --- a/packages/coding-agent/test/experimental-client-server-backend.test.ts +++ b/packages/coding-agent/test/experimental-client-server-backend.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { appendFile, mkdir, mkdtemp, readdir, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { fauxAssistantMessage, fauxProvider, fauxToolCall } from "@earendil-works/pi-ai"; @@ -7,11 +7,12 @@ import { PiClient } from "@earendil-works/pi-client"; import { createUnixTransportFactory } from "@earendil-works/pi-client/unix"; import type { TranscriptProgress } from "@earendil-works/pi-protocol"; import { PiServer, PiServerError } from "@earendil-works/pi-server"; -import { describe, expect, test } from "vitest"; +import lockfile, { type LockOptions } from "proper-lockfile"; +import { describe, expect, test, vi } from "vitest"; 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 { ExperimentalPiSessionBackend } from "../src/experimental/client-server/backend.ts"; +import { ExperimentalPiSessionBackend, toPiServerError } from "../src/experimental/client-server/backend.ts"; import { ExperimentalClientController } from "../src/experimental/client-server/client-controller.ts"; async function createFixture() { @@ -129,8 +130,13 @@ describe("experimental AgentHarness server backend", () => { 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); @@ -140,6 +146,43 @@ describe("experimental AgentHarness server backend", () => { } }); + test("fails session listing when persisted JSONL is unreadable", async () => { + const fixture = await createFixture(); + 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 experimental session/); + } finally { + await rm(fixture.root, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 }); + } + }); + + test("propagates lock status check failures", async () => { + const fixture = await createFixture(); + 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 experimental session/); + } finally { + checkSpy.mockRestore(); + await rm(fixture.root, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 }); + } + }); + + 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 createFixture(); fixture.backend.setDefaultSessionOptions({ @@ -158,6 +201,83 @@ describe("experimental AgentHarness server backend", () => { } }); + test("terminates an active runtime without persisting after its session lock is compromised", async () => { + const fixture = await createFixture(); + const realLock = lockfile.lock.bind(lockfile); + let compromise: LockOptions["onCompromised"]; + let forceRelease: (() => Promise) | undefined; + const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async (file, options) => { + compromise = options?.onCompromised; + forceRelease = await realLock(file, options); + return forceRelease; + }); + let runtime: Awaited> | undefined; + try { + runtime = await fixture.backend.createSession({ + id: "server-compromised-lock", + cwd: fixture.cwd, + model: { provider: fixture.faux.provider.id, id: "faux-reasoning" }, + }); + fixture.faux.setResponses([fauxAssistantMessage("long response ".repeat(1_000))]); + let startedResolve: (() => void) | undefined; + const started = new Promise((resolvePromise) => { + startedResolve = resolvePromise; + }); + runtime.subscribe(() => { + if (runtime?.getPhase() === "turn") startedResolve?.(); + }); + const prompt = runtime.prompt({ text: "start" }); + await 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 ExperimentalPiSessionBackend.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 rm(fixture.root, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 }); + } + }); + + test("propagates lock release failure and allows disposal retry", async () => { + const fixture = await createFixture(); + const realLock = lockfile.lock.bind(lockfile); + let releaseAttempts = 0; + const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async (file, options) => { + const release = await realLock(file, options); + return async () => { + releaseAttempts += 1; + if (releaseAttempts === 1) throw new Error("release failed"); + await release(); + }; + }); + let runtime: Awaited> | undefined; + try { + runtime = await fixture.backend.createSession({ id: "server-release-retry", cwd: fixture.cwd }); + await expect(runtime.dispose()).rejects.toThrow("release failed"); + await expect(runtime.dispose()).resolves.toBeUndefined(); + expect(releaseAttempts).toBe(2); + } finally { + lockSpy.mockRestore(); + await runtime?.dispose(); + await rm(fixture.root, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 }); + } + }); + test("exposes accepted steering text to every session snapshot", async () => { const fixture = await createFixture(); const runtime = await fixture.backend.createSession({ diff --git a/packages/coding-agent/test/experimental-client-tui.test.ts b/packages/coding-agent/test/experimental-client-tui.test.ts new file mode 100644 index 000000000..0258916f9 --- /dev/null +++ b/packages/coding-agent/test/experimental-client-tui.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, test } from "vitest"; +import { DoubleClearAction } from "../src/experimental/client-server/tui.ts"; + +describe("experimental client TUI", () => { + test("exits only when app.clear is triggered twice within 500 ms", () => { + const action = new DoubleClearAction(); + + expect(action.trigger(1_000)).toBe("clear"); + expect(action.trigger(1_499)).toBe("exit"); + }); + + test("starts a new double-clear window after the timeout", () => { + const action = new DoubleClearAction(); + + expect(action.trigger(1_000)).toBe("clear"); + expect(action.trigger(1_500)).toBe("clear"); + expect(action.trigger(1_999)).toBe("exit"); + }); +}); diff --git a/packages/coding-agent/test/experimental-client-view-model.test.ts b/packages/coding-agent/test/experimental-client-view-model.test.ts index a658a0213..da35e4bfb 100644 --- a/packages/coding-agent/test/experimental-client-view-model.test.ts +++ b/packages/coding-agent/test/experimental-client-view-model.test.ts @@ -45,6 +45,51 @@ describe("experimental client view model", () => { expect(model.view()?.transcript[0]).toMatchObject({ content: [{ type: "text", text: "saved response" }] }); }); + test("applies streamed tool-call argument deltas", () => { + const model = new ExperimentalSessionViewModel(); + model.applySnapshot({ + ...snapshot(1), + transcript: [ + { + id: "assistant-1", + role: "assistant", + content: [{ type: "tool_call", toolCallId: "call-1", toolName: "bash", input: null }], + status: "streaming", + model: { provider: "faux", id: "faux-1" }, + timestamp: 1, + }, + ], + }); + model.applyProgress({ + type: "assistant_delta", + messageId: "assistant-1", + contentIndex: 0, + kind: "tool_call", + delta: '{"command":', + }); + expect(model.view()?.transcript[0]).toMatchObject({ content: [{ input: '{"command":' }] }); + + model.applyProgress({ + type: "item_updated", + item: { + id: "assistant-1", + role: "assistant", + content: [{ type: "tool_call", toolCallId: "call-1", toolName: "bash", input: null }], + status: "streaming", + model: { provider: "faux", id: "faux-1" }, + timestamp: 1, + }, + }); + model.applyProgress({ + type: "assistant_delta", + messageId: "assistant-1", + contentIndex: 0, + kind: "tool_call", + delta: '"pwd"}', + }); + expect(model.view()?.transcript[0]).toMatchObject({ content: [{ input: { command: "pwd" } }] }); + }); + test("appends transient tool progress and replaces it by id", () => { const model = new ExperimentalSessionViewModel(); model.applySnapshot(snapshot(1)); diff --git a/packages/protocol/README.md b/packages/protocol/README.md index 7a33b7dc4..4b437ba78 100644 --- a/packages/protocol/README.md +++ b/packages/protocol/README.md @@ -81,7 +81,7 @@ The CBOR subset supports: - definite-length arrays - definite-length maps represented by objects with unique string keys -Undefined object properties are omitted. Top-level undefined, undefined array entries, sparse arrays, non-finite or unsafe numbers, tags, indefinite-length items, malformed UTF-8, trailing data, excessive nesting, and oversized values are rejected. +Undefined object properties are omitted. JSON-valued protocol fields reject CBOR byte strings and non-plain objects. Top-level undefined, undefined array entries, sparse arrays, non-finite or unsafe numbers, tags, indefinite-length items, malformed UTF-8, trailing data, excessive nesting, and oversized values are rejected. Default limits are 16 MiB per CBOR payload/frame, 1,000,000 array elements or map entries, and 64 nested item levels. Options can configure these limits. A frame decoder validates the declared length before buffering payload bytes. diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 18b85167e..f9f107bcd 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -453,13 +453,27 @@ export class ProtocolValidationError extends Error { } } +function isProtocolValue(value: unknown, optionalProperty = false): boolean { + if (value === undefined) return optionalProperty; + if (value === null || typeof value === "boolean" || typeof value === "number" || typeof value === "string") { + return true; + } + if (Array.isArray(value)) return value.every((item) => isProtocolValue(item)); + if (typeof value !== "object" || Object.getPrototypeOf(value) !== Object.prototype) return false; + return Object.values(value).every((item) => isProtocolValue(item, true)); +} + export function parseClientMessage(value: unknown): ClientMessage { - if (!Check(ClientMessageSchema, value)) throw new ProtocolValidationError("Invalid client protocol message"); + if (!Check(ClientMessageSchema, value) || !isProtocolValue(value)) { + throw new ProtocolValidationError("Invalid client protocol message"); + } return value; } export function parseServerMessage(value: unknown): ServerMessage { - if (!Check(ServerMessageSchema, value)) throw new ProtocolValidationError("Invalid server protocol message"); + if (!Check(ServerMessageSchema, value) || !isProtocolValue(value)) { + throw new ProtocolValidationError("Invalid server protocol message"); + } return value; } diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index 0766ebc7e..40346cb01 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -207,6 +207,22 @@ describe("validated framed protocol APIs", () => { expect(() => decoder.push(encodeClientMessage(clientHello))).toThrow(/failed/i); }); + test("rejects CBOR byte strings nested in JSON-valued fields", () => { + const wire = encodeFrame( + encodeCbor({ + type: "response", + id: "request-1", + ok: false, + error: { + code: "invalid_request", + message: "invalid", + details: { nested: new Uint8Array([1, 2, 3]) }, + }, + }), + ); + expect(() => new ServerMessageDecoder().push(wire)).toThrow(ProtocolValidationError); + }); + test("rejects truncated and oversized framing through the validated decoder", () => { const truncated = new ServerMessageDecoder(); expect(truncated.push(new Uint8Array([0, 0, 0, 2, 1]))).toEqual([]); diff --git a/packages/server/README.md b/packages/server/README.md index 401c862e9..38dca08ef 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -73,6 +73,6 @@ The listener rejects paths exceeding the platform's `sockaddr_un` limit, creates To keep Node's automatic pipe cleanup from unlinking a replacement path, the listener binds an owned private socket inode and exposes the configured path as a hard link. Normal shutdown and stale-start recovery clean both owned links after inode verification. -Top-level options include `maxFrameLength`, `maxPendingBytes`, `handshakeTimeoutMs`, `gracefulCloseTimeoutMs`, `serverId`, and `onError`. Call `await server.close()` for idempotent graceful listener shutdown and runtime disposal. +Top-level options include `maxFrameLength`, `maxPendingBytes`, `handshakeTimeoutMs`, `gracefulCloseTimeoutMs`, `serverId`, and `onError`. Timeout values must fit Node's timer range of 1 through 2,147,483,647 milliseconds. Call `await server.close()` for idempotent graceful listener shutdown and runtime disposal. The current server is same-host Unix only. It does not expose an HTTP health endpoint. Supervisors should use socket readiness plus the protocol handshake. There is no standalone server package CLI and no session close/delete protocol operation. diff --git a/packages/server/src/connection.ts b/packages/server/src/connection.ts index 8380c4da3..8044ada28 100644 --- a/packages/server/src/connection.ts +++ b/packages/server/src/connection.ts @@ -4,7 +4,7 @@ import type { MaybePromise } from "./types.ts"; export interface ByteConnection { readonly closed: boolean; send(chunk: Uint8Array): Promise; - close(): MaybePromise; + close(finalChunk?: Uint8Array): MaybePromise; } export interface ByteConnectionHandler { diff --git a/packages/server/src/core.ts b/packages/server/src/core.ts index b2c745d0b..145e8ddc1 100644 --- a/packages/server/src/core.ts +++ b/packages/server/src/core.ts @@ -279,7 +279,7 @@ export class PiServerCore { }; const live = await this.acquireSession(id, () => this.backend.createSession(options)); await this.attach(state, live); - const session = await this.broadcastSessionSnapshot(live); + const session = this.snapshotForConnection(await this.broadcastSessionSnapshot(live), state); void this.broadcastServerSnapshot(); return { command: "create" as const, session }; } @@ -288,7 +288,7 @@ export class PiServerCore { this.backend.openSession(command.sessionId), ); await this.attach(state, live); - const session = await this.broadcastSessionSnapshot(live); + const session = this.snapshotForConnection(await this.broadcastSessionSnapshot(live), state); void this.broadcastServerSnapshot(); return { command: "attach" as const, session }; } @@ -302,37 +302,41 @@ export class PiServerCore { } case "prompt": { const live = this.requireAttached(state, command.sessionId); - const session = await this.runOperation(live, () => live.runtime.prompt({ text: command.text })); + const session = await this.runOperation(state, live, () => live.runtime.prompt({ text: command.text })); return { command: "prompt" as const, session }; } case "steer": { const live = this.requireAttached(state, command.sessionId); - const session = await this.runOperation(live, () => live.runtime.steer({ text: command.text })); + const session = await this.runOperation(state, live, () => live.runtime.steer({ text: command.text })); return { command: "steer" as const, session }; } case "abort": { const live = this.requireAttached(state, command.sessionId); - const session = await this.runOperation(live, () => live.runtime.abort()); + const session = await this.runOperation(state, live, () => live.runtime.abort()); return { command: "abort" as const, session }; } case "set_model": { const live = this.requireAttached(state, command.sessionId); - const session = await this.runOperation(live, () => live.runtime.setModel(command.model)); + const session = await this.runOperation(state, live, () => live.runtime.setModel(command.model)); return { command: "set_model" as const, session }; } case "set_thinking": { const live = this.requireAttached(state, command.sessionId); - const session = await this.runOperation(live, () => live.runtime.setThinking(command.thinkingLevel)); + const session = await this.runOperation(state, live, () => live.runtime.setThinking(command.thinkingLevel)); return { command: "set_thinking" as const, session }; } } } - private async runOperation(live: LiveSession, operation: () => Promise): Promise { + private async runOperation( + connection: ConnectionState, + live: LiveSession, + operation: () => Promise, + ): Promise { live.operationCount += 1; try { await operation(); - return await this.broadcastSessionSnapshot(live); + return this.snapshotForConnection(await this.broadcastSessionSnapshot(live), connection); } finally { live.operationCount -= 1; this.scheduleMaybeDispose(live); @@ -402,6 +406,10 @@ export class PiServerCore { } private handleRuntimeEvent(live: LiveSession, event: PiSessionRuntimeEvent): void { + if (event.type === "error") { + void this.terminateLiveSession(live, event.error).catch((error: unknown) => this.reportError(error)); + return; + } if (event.type === "progress") { const envelope: EventEnvelope = { type: "event", @@ -414,6 +422,15 @@ export class PiServerCore { this.scheduleMaybeDispose(live); } + private async terminateLiveSession(live: LiveSession, error: PiServerError): Promise { + this.reportError(error); + live.unsubscribe(); + const connections = [...live.connections]; + await Promise.all(connections.map((connection) => this.closeConnection(connection.connection))); + await Promise.all(connections.map((connection) => this.disconnect(connection))); + await this.maybeDispose(live); + } + private async normalizedSnapshot(live: LiveSession): Promise { const snapshot = await live.runtime.snapshot(); if (snapshot.id !== live.id) { @@ -427,6 +444,10 @@ export class PiServerCore { }; } + private snapshotForConnection(snapshot: SessionSnapshot, connection: ConnectionState): SessionSnapshot { + return { ...snapshot, attached: connection.sessionIds.has(snapshot.id) }; + } + private async broadcastSessionSnapshot(live: LiveSession): Promise { const snapshot = await this.normalizedSnapshot(live); const envelope: EventEnvelope = { type: "event", event: { type: "session_snapshot", snapshot } }; @@ -592,8 +613,13 @@ export class PiServerCore { connection.stage = "closing"; clearTimeout(connection.handshakeTimeout); const message: ServerHelloError = { type: "hello_error", error }; - await this.sendMessage(connection, message); - await this.closeConnection(connection.connection); + let finalFrame: Uint8Array | undefined; + try { + finalFrame = encodeServerMessage(message, { maxFrameLength: this.maxFrameLength }); + } catch (encodeError) { + this.reportError(encodeError); + } + await this.closeConnection(connection.connection, finalFrame); await this.disconnect(connection); } @@ -625,9 +651,9 @@ export class PiServerCore { this.connections.clear(); } - private async closeConnection(connection: ByteConnection): Promise { + private async closeConnection(connection: ByteConnection, finalChunk?: Uint8Array): Promise { try { - await connection.close(); + await connection.close(finalChunk); } catch (error) { this.reportError(error); } diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 57518522b..b9708a8a4 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -1,12 +1,12 @@ import { DEFAULT_MAX_FRAME_LENGTH } from "@earendil-works/pi-protocol"; import { PiServerCore } from "./core.ts"; import type { PiServerOptions, PiSessionBackend, UnixListenerOptions } from "./types.ts"; -import { UnixListener } from "./unix-listener.ts"; +import { UnixListener, validateUnixSocketPath } from "./unix-listener.ts"; const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000; const DEFAULT_GRACEFUL_CLOSE_TIMEOUT_MS = 5_000; const MAX_UINT32 = 0xffff_ffff; -const MAX_UNIX_SOCKET_PATH_BYTES = process.platform === "linux" ? 107 : 103; +const MAX_TIMER_DELAY_MS = 2_147_483_647; interface ResolvedOptions { unix: UnixListenerOptions; @@ -111,12 +111,20 @@ function resolveOptions(options: PiServerOptions): ResolvedOptions { throw new TypeError("PiServer maxPendingBytes must be a safe integer at least maxFrameLength + 4"); } const handshakeTimeoutMs = options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS; - if (!Number.isSafeInteger(handshakeTimeoutMs) || handshakeTimeoutMs <= 0) { - throw new TypeError("PiServer handshakeTimeoutMs must be a positive integer"); + if ( + !Number.isSafeInteger(handshakeTimeoutMs) || + handshakeTimeoutMs <= 0 || + handshakeTimeoutMs > MAX_TIMER_DELAY_MS + ) { + throw new TypeError(`PiServer handshakeTimeoutMs must be an integer between 1 and ${MAX_TIMER_DELAY_MS}`); } const gracefulCloseTimeoutMs = options.gracefulCloseTimeoutMs ?? DEFAULT_GRACEFUL_CLOSE_TIMEOUT_MS; - if (!Number.isSafeInteger(gracefulCloseTimeoutMs) || gracefulCloseTimeoutMs <= 0) { - throw new TypeError("PiServer gracefulCloseTimeoutMs must be a positive integer"); + if ( + !Number.isSafeInteger(gracefulCloseTimeoutMs) || + gracefulCloseTimeoutMs <= 0 || + gracefulCloseTimeoutMs > MAX_TIMER_DELAY_MS + ) { + throw new TypeError(`PiServer gracefulCloseTimeoutMs must be an integer between 1 and ${MAX_TIMER_DELAY_MS}`); } validateUnixOptions(options.unix); return { @@ -130,12 +138,7 @@ function resolveOptions(options: PiServerOptions): ResolvedOptions { } function validateUnixOptions(options: UnixListenerOptions): void { - if (!options.path) throw new TypeError("PiServer Unix socket path must not be empty"); - if (Buffer.byteLength(options.path) > MAX_UNIX_SOCKET_PATH_BYTES) { - throw new TypeError( - `PiServer Unix socket path is too long; maximum is ${MAX_UNIX_SOCKET_PATH_BYTES} UTF-8 bytes`, - ); - } + validateUnixSocketPath(options.path, "PiServer Unix socket path"); if (options.mode !== undefined && (!Number.isInteger(options.mode) || options.mode < 0 || options.mode > 0o777)) { throw new TypeError("PiServer Unix socket mode must be an integer between 0 and 0o777"); } diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts index 091c79d76..575953f04 100644 --- a/packages/server/src/types.ts +++ b/packages/server/src/types.ts @@ -25,7 +25,10 @@ export interface CreateSessionOptions { thinkingLevel?: ThinkingLevel; } -export type PiSessionRuntimeEvent = { type: "snapshot" } | { type: "progress"; progress: TranscriptProgress }; +export type PiSessionRuntimeEvent = + | { type: "snapshot" } + | { type: "progress"; progress: TranscriptProgress } + | { type: "error"; error: PiServerError }; /** * One acquired durable session. Implementations must reject conflicting prompt diff --git a/packages/server/src/unix-listener.ts b/packages/server/src/unix-listener.ts index 866a77474..b4dda245e 100644 --- a/packages/server/src/unix-listener.ts +++ b/packages/server/src/unix-listener.ts @@ -8,6 +8,14 @@ import type { UnixListenerOptions } from "./types.ts"; const DEFAULT_SOCKET_MODE = 0o600; const SOCKET_PROBE_TIMEOUT_MS = 1_000; +const MAX_UNIX_SOCKET_PATH_BYTES = process.platform === "linux" ? 107 : 103; + +export function validateUnixSocketPath(path: string, description = "Unix socket path"): void { + if (!path) throw new TypeError(`${description} must not be empty`); + if (Buffer.byteLength(path) > MAX_UNIX_SOCKET_PATH_BYTES) { + throw new TypeError(`${description} is too long; maximum is ${MAX_UNIX_SOCKET_PATH_BYTES} UTF-8 bytes`); + } +} interface ListenerOptions { listener: UnixListenerOptions; @@ -47,9 +55,10 @@ export class UnixListener { if (this.server) throw new Error("Unix listener is already started"); if (this.closing) throw new Error("Unix listener is closing or closed"); + const ownedBindPath = getOwnedBindPath(this.path); + validateUnixSocketPath(ownedBindPath, "PiServer private Unix bind path"); await mkdir(dirname(this.path), { recursive: true, mode: 0o700 }); await removeStaleSocket(this.path); - const ownedBindPath = getOwnedBindPath(this.path); await removeStaleSocket(ownedBindPath); this.ownedBindPath = ownedBindPath; const server = createServer((socket) => this.acceptSocket(socket)); @@ -218,13 +227,18 @@ class UnixByteConnection implements ByteConnection { return tracked; } - close(): Promise { + close(finalChunk?: Uint8Array): Promise { if (this.closedValue || this.socket.destroyed) { this.markClosed(); return Promise.resolve(); } if (this.closePromise) return this.closePromise; this.closing = true; + if (finalChunk && (this.pendingBytes > 0 || this.socket.writableLength > 0)) { + this.socket.destroy(); + this.markClosed(); + return Promise.resolve(); + } this.closePromise = new Promise((resolve) => { this.resolveClose = resolve; const timer = setTimeout(() => { @@ -234,7 +248,8 @@ class UnixByteConnection implements ByteConnection { timer.unref(); this.socket.once("close", () => clearTimeout(timer)); try { - this.socket.end(); + if (finalChunk) this.socket.end(finalChunk); + else this.socket.end(); } catch { this.socket.destroy(); } diff --git a/packages/server/test/conformance.test.ts b/packages/server/test/conformance.test.ts index c46623a23..7e458e258 100644 --- a/packages/server/test/conformance.test.ts +++ b/packages/server/test/conformance.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { encodeClientMessage, encodeFrame, PROTOCOL_VERSION } from "@earendil-works/pi-protocol"; import { afterEach, describe, expect, test } from "vitest"; -import { PiServer, type PiServerOptions } from "../src/index.ts"; +import { PiServer, PiServerError, type PiServerOptions } from "../src/index.ts"; import { ConformanceBackend, connectUnix, Deferred, TEST_TOKEN, type WireClient } from "./support.ts"; const servers = new Set(); @@ -225,6 +225,22 @@ describe("Unix transport conformance", () => { expect(secondRuntime.disposeCount).toBe(1); }); + test("disconnects attached clients when a runtime reports a terminal error", async () => { + const backend = new ConformanceBackend(); + backend.seed("terminal"); + const errors: Error[] = []; + const { server } = await startServer(backend, { onError: (error) => errors.push(error) }); + const client = await connect(server); + await client.hello(); + await client.request({ command: "attach", sessionId: "terminal" }); + const runtime = backend.latestRuntime("terminal"); + + runtime.emitError(new PiServerError("session_locked", "lock ownership lost")); + await client.waitForClose(); + await runtime.disposed.promise; + expect(errors).toContainEqual(expect.objectContaining({ code: "session_locked" })); + }); + test("does not expose unexpected backend errors to clients", async () => { class FailingBackend extends ConformanceBackend { private listCount = 0; diff --git a/packages/server/test/server.test.ts b/packages/server/test/server.test.ts index eb410c4e7..4baf9fb84 100644 --- a/packages/server/test/server.test.ts +++ b/packages/server/test/server.test.ts @@ -589,6 +589,33 @@ describe("PiServer Unix integration", () => { expect(await prompt).toMatchObject({ ok: true, result: { command: "prompt", session: { phase: "idle" } } }); }); + test("returns operation attachment state relative to the requesting connection", async () => { + const backend = new MemoryBackend(); + backend.seed(); + const { server } = await startServer(backend); + const first = await connect(server); + const second = await connect(server); + await first.hello(); + await second.hello(); + await attach(first, "session-1"); + await attach(second, "session-1"); + + const prompt = first.request({ command: "prompt", sessionId: "session-1", text: "hello" }); + await first.next( + (message) => + message.type === "event" && + message.event.type === "session_snapshot" && + message.event.snapshot.phase === "turn", + ); + await first.request({ command: "detach", sessionId: "session-1" }); + backend.latestRuntime("session-1").finishPrompt(); + + expect(await prompt).toMatchObject({ + ok: true, + result: { command: "prompt", session: { id: "session-1", attached: false } }, + }); + }); + test("keeps busy work alive after disconnect and disposes when it next becomes idle", async () => { const backend = new MemoryBackend(); backend.seed(); diff --git a/packages/server/test/support.ts b/packages/server/test/support.ts index 434fbd7b3..da2673472 100644 --- a/packages/server/test/support.ts +++ b/packages/server/test/support.ts @@ -125,6 +125,10 @@ export class ConformanceRuntime implements PiSessionRuntime { for (const listener of this.listeners) listener({ type: "progress", progress }); } + emitError(error: PiServerError): void { + for (const listener of this.listeners) listener({ type: "error", error }); + } + private update(updates: Partial): void { this.stored.snapshot = { ...this.stored.snapshot, diff --git a/packages/server/test/transport-smoke.test.ts b/packages/server/test/transport-smoke.test.ts index c108c81ed..412cb6bb2 100644 --- a/packages/server/test/transport-smoke.test.ts +++ b/packages/server/test/transport-smoke.test.ts @@ -9,7 +9,9 @@ import { type ServerMessage, ServerMessageDecoder, } from "@earendil-works/pi-protocol"; -import { afterEach, expect, test } from "vitest"; +import { afterEach, expect, test, vi } from "vitest"; +import type { ByteConnection } from "../src/connection.ts"; +import { PiServerCore } from "../src/core.ts"; import { PiServer, type PiSessionBackend } from "../src/index.ts"; const TOKEN = "transport-smoke-token"; @@ -55,6 +57,15 @@ test("rejects Unix socket paths that cannot fit in sockaddr_un", () => { expect(() => new PiServer(backend, { token: TOKEN, unix: { path: `/tmp/${"x".repeat(512)}` } })).toThrow(/too long/); }); +test("rejects an overlong derived private Unix bind path", async () => { + const maxLength = process.platform === "linux" ? 107 : 103; + const suffixLength = Buffer.byteLength("/tmp//s"); + const path = `/tmp/${"x".repeat(maxLength - suffixLength)}/s`; + server = new PiServer(backend, { token: TOKEN, unix: { path } }); + + await expect(server.start()).rejects.toThrow(/private Unix bind path.*too long/); +}); + test("Unix socket accepts a fragmented framed-CBOR hello", async () => { const path = await makeSocketPath(); server = new PiServer(backend, { token: TOKEN, unix: { path } }); @@ -81,6 +92,45 @@ test("rejects concurrent start calls without leaking the Unix listener", async ( await expect(lstat(path)).rejects.toMatchObject({ code: "ENOENT" }); }); +test("handshake timeout cleanup does not wait for a blocked output queue", async () => { + class BlockedConnection implements ByteConnection { + closed = false; + finalChunk?: Uint8Array; + + send(): Promise { + return new Promise(() => {}); + } + + close(finalChunk?: Uint8Array): void { + this.finalChunk = finalChunk; + this.closed = true; + } + } + const core = new PiServerCore(backend, { + token: TOKEN, + maxFrameLength: 1024, + handshakeTimeoutMs: 10, + }); + const connection = new BlockedConnection(); + core.accept(connection); + + await vi.waitFor(() => expect(connection.closed).toBe(true)); + expect(connection.finalChunk).toBeInstanceOf(Uint8Array); + const messages = new ServerMessageDecoder().push(connection.finalChunk!); + expect(messages).toMatchObject([{ type: "hello_error", error: { code: "invalid_request" } }]); + await core.close(); +}); + +test("rejects timeout values above Node's maximum timer delay", () => { + const unix = { path: "/tmp/pi-server-timeout-test.sock" }; + expect(() => new PiServer(backend, { token: TOKEN, unix, handshakeTimeoutMs: 2_147_483_648 })).toThrow( + /handshakeTimeoutMs/, + ); + expect(() => new PiServer(backend, { token: TOKEN, unix, gracefulCloseTimeoutMs: 2_147_483_648 })).toThrow( + /gracefulCloseTimeoutMs/, + ); +}); + test("rejects pending-byte limits smaller than one maximum frame", async () => { const path = await makeSocketPath(); expect(