diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 019f08302..653194039 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -99,6 +99,21 @@ function todoTimeline(items: { text: string; completed: boolean }[]): AgentStrea }; } +function compactionTimeline( + status: "loading" | "completed", + trigger?: "auto" | "manual", +): AgentStreamEventPayload { + return { + type: "timeline", + provider: "pi", + item: { + type: "compaction", + status, + ...(trigger ? { trigger } : {}), + }, + }; +} + function findToolByCallId(state: StreamItem[], callId: string): AgentToolCallItem | undefined { return state.find( (item): item is AgentToolCallItem => @@ -693,6 +708,27 @@ describe("stream reducer canonical tool calls", () => { assert.strictEqual(todos.items[1]?.completed, true); }); + it("preserves compaction trigger when completed update replaces loading marker", () => { + const state = hydrateStreamState([ + { + event: compactionTimeline("loading", "auto"), + timestamp: new Date("2025-01-01T10:50:00Z"), + }, + { + event: compactionTimeline("completed"), + timestamp: new Date("2025-01-01T10:50:01Z"), + }, + ]); + + const compactions = state.filter( + (item): item is Extract => item.kind === "compaction", + ); + + assert.strictEqual(compactions.length, 1); + assert.strictEqual(compactions[0].status, "completed"); + assert.strictEqual(compactions[0].trigger, "auto"); + }); + it("renders Claude TodoWrite as todo_list and suppresses tool call badge", () => { const state = hydrateStreamState([ { diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 44bfa2e35..c67e0d0ed 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -726,8 +726,8 @@ function reduceTimelineCompaction( const updated: CompactionItem = { ...existing, status: "completed", - trigger: item.trigger, - preTokens: item.preTokens, + trigger: item.trigger ?? existing.trigger, + preTokens: item.preTokens ?? existing.preTokens, }; return [...state.slice(0, loadingIdx), updated, ...state.slice(loadingIdx + 1)]; } diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 6f2b84006..51d6a6825 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -534,6 +534,37 @@ describe("Codex app-server provider", () => { await session.close(); }); + test("initializes Codex app-server without making Paseo the request originator", async () => { + let initializeParams: unknown; + const appServer = createFakeCodexAppServer({ + initialize: (params) => { + initializeParams = params; + return {}; + }, + "collaborationMode/list": () => ({ data: [] }), + "skills/list": () => ({ data: [] }), + }); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + await session.connect(); + + expect(initializeParams).toEqual({ + clientInfo: { + name: "codex_app_server_daemon", + title: "Codex App Server Daemon", + version: "0.0.0", + }, + capabilities: { experimentalApi: true }, + }); + appServer.assertNoErrors(); + await session.close(); + }); + test("rewinds the conversation to a freshly emitted Codex user message id", async () => { const appServer = createFakeCodexAppServer(); const session = new CodexAppServerAgentSession( @@ -2703,7 +2734,17 @@ describe("Codex persisted sessions", () => { }), ); expect(calls).toEqual([ - { method: "initialize", params: expect.any(Object) }, + { + method: "initialize", + params: { + clientInfo: { + name: "codex_app_server_daemon", + title: "Codex App Server Daemon", + version: "0.0.0", + }, + capabilities: { experimentalApi: true }, + }, + }, { method: "thread/list", params: { limit: 50, cwd: "/workspace/project-a" } }, ]); }); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index d00fd6411..9f1dc6aa4 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -103,6 +103,14 @@ const TURN_START_TIMEOUT_MS = 90 * 1000; const INTERRUPT_TIMEOUT_MS = 2_000; const CODEX_PROVIDER = "codex" as const; const CODEX_IMAGE_ATTACHMENT_DIR = "paseo-attachments"; +// Codex treats most app-server client names as the model-request originator. +// This reserved Codex name is non-originating, so requests keep Codex's default +// CLI identity instead of showing up as Paseo in provider usage logs. +const CODEX_NON_ORIGINATING_APP_SERVER_CLIENT_INFO = { + name: "codex_app_server_daemon", + title: "Codex App Server Daemon", + version: "0.0.0", +} as const; const ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN = "\n\n---\n\n"; const CODEX_TOOL_THREAD_ITEM_TYPES = new Set([ "commandExecution", @@ -2818,11 +2826,7 @@ function buildCodexAppServerInitializeParams(): { capabilities: { experimentalApi: true }; } { return { - clientInfo: { - name: "paseo", - title: "Paseo", - version: "0.0.0", - }, + clientInfo: CODEX_NON_ORIGINATING_APP_SERVER_CLIENT_INFO, capabilities: { experimentalApi: true, }, diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index 5b4eb5838..6866df671 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -2,7 +2,7 @@ import { closeSync, existsSync, fstatSync, openSync, readSync } from "node:fs"; import pino from "pino"; import { describe, expect, test } from "vitest"; -import type { AgentSessionConfig, AgentStreamEvent } from "../../agent-sdk-types.js"; +import type { AgentSession, AgentSessionConfig, AgentStreamEvent } from "../../agent-sdk-types.js"; import { PiRpcAgentClient, PiRpcAgentSession, transformPiModels } from "./agent.js"; import { FakePi } from "./test-utils/fake-pi.js"; @@ -693,12 +693,202 @@ describe("PiRpcAgentClient", () => { ]; await expect(session.listCommands()).resolves.toEqual([ + { + name: "compact", + description: "Manually compact the session context", + argumentHint: "[instructions]", + }, + { + name: "autocompact", + description: "Toggle automatic context compaction", + argumentHint: "[on|off|toggle]", + }, { name: "review", description: "Review changes", argumentHint: "" }, { name: "fix-tests", description: "Fix tests", argumentHint: "" }, { name: "skill:docs", description: "Read docs", argumentHint: "" }, ]); }); + test("lists Pi compact even when RPC get_commands omits built-in slash commands", async () => { + const { pi, session } = await createSession(); + pi.latestSession().commands = [ + { name: "review", description: "Review changes", source: "extension" }, + ]; + + await expect(session.listCommands()).resolves.toContainEqual({ + name: "compact", + description: "Manually compact the session context", + argumentHint: "[instructions]", + }); + await expect(session.listCommands()).resolves.toContainEqual({ + name: "autocompact", + description: "Toggle automatic context compaction", + argumentHint: "[on|off|toggle]", + }); + }); + + test("preserves known argument hints when RPC get_commands returns built-in slash commands", async () => { + const { pi, session } = await createSession(); + pi.latestSession().commands = [ + { name: "compact", description: "Compact from RPC", source: "extension" }, + { name: "autocompact", description: "Auto compact from RPC", source: "extension" }, + ]; + + await expect(session.listCommands()).resolves.toEqual([ + { + name: "compact", + description: "Compact from RPC", + argumentHint: "[instructions]", + }, + { + name: "autocompact", + description: "Auto compact from RPC", + argumentHint: "[on|off|toggle]", + }, + ]); + }); + + test("executes Pi compact through RPC instead of prompt text", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const handler = (session as AgentSession).tryHandleOutOfBand?.("/compact focus on tests"); + const events: AgentStreamEvent[] = []; + + expect(handler).not.toBeNull(); + await handler?.run({ emit: (event) => events.push(event) }); + + expect(fakeSession.compactRequests).toEqual([{ customInstructions: "focus on tests" }]); + expect(fakeSession.prompts).toEqual([]); + expect(events).toEqual([ + { + type: "timeline", + provider: "pi", + item: { type: "compaction", status: "loading", trigger: "manual" }, + }, + { + type: "timeline", + provider: "pi", + item: { type: "compaction", status: "completed", trigger: "manual" }, + }, + ]); + }); + + test("closes Pi compact loading marker when RPC rejects after compaction starts", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.emitCompactEnd = false; + fakeSession.compactError = new Error("summarizer failed"); + const handler = (session as AgentSession).tryHandleOutOfBand?.("/compact"); + const events: AgentStreamEvent[] = []; + + expect(handler).not.toBeNull(); + await handler?.run({ emit: (event) => events.push(event) }); + + expect(events).toEqual([ + { + type: "timeline", + provider: "pi", + item: { type: "compaction", status: "loading", trigger: "manual" }, + }, + { + type: "timeline", + provider: "pi", + item: { type: "compaction", status: "completed", trigger: "manual" }, + }, + { + type: "timeline", + provider: "pi", + item: { + type: "assistant_message", + text: "[Error] Failed to compact context: summarizer failed", + }, + }, + ]); + }); + + test("executes Pi autocompact through RPC instead of prompt text", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const handler = (session as AgentSession).tryHandleOutOfBand?.("/autocompact off"); + const events: AgentStreamEvent[] = []; + + expect(handler).not.toBeNull(); + await handler?.run({ emit: (event) => events.push(event) }); + + expect(fakeSession.setAutoCompactionRequests).toEqual([false]); + expect(fakeSession.prompts).toEqual([]); + expect(events).toEqual([ + { + type: "timeline", + provider: "pi", + item: { type: "assistant_message", text: "Auto-compaction disabled." }, + }, + ]); + }); + + test("rejects unknown Pi autocompact mode instead of toggling", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const handler = (session as AgentSession).tryHandleOutOfBand?.("/autocompact banana"); + const events: AgentStreamEvent[] = []; + + expect(handler).not.toBeNull(); + await handler?.run({ emit: (event) => events.push(event) }); + + expect(fakeSession.setAutoCompactionRequests).toEqual([]); + expect(events).toEqual([ + { + type: "timeline", + provider: "pi", + item: { + type: "assistant_message", + text: "[Error] Usage: /autocompact [on|off|toggle]", + }, + }, + ]); + }); + + test("toggles Pi autocompact through current RPC state", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.state.autoCompactionEnabled = false; + const handler = (session as AgentSession).tryHandleOutOfBand?.("/autocompact"); + const events: AgentStreamEvent[] = []; + + expect(handler).not.toBeNull(); + await handler?.run({ emit: (event) => events.push(event) }); + + expect(fakeSession.setAutoCompactionRequests).toEqual([true]); + expect(events).toContainEqual({ + type: "timeline", + provider: "pi", + item: { type: "assistant_message", text: "Auto-compaction enabled." }, + }); + }); + + test("rejects Pi autocompact toggle when current RPC state is unavailable", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + delete fakeSession.state.autoCompactionEnabled; + const handler = (session as AgentSession).tryHandleOutOfBand?.("/autocompact"); + const events: AgentStreamEvent[] = []; + + expect(handler).not.toBeNull(); + await handler?.run({ emit: (event) => events.push(event) }); + + expect(fakeSession.setAutoCompactionRequests).toEqual([]); + expect(events).toEqual([ + { + type: "timeline", + provider: "pi", + item: { + type: "assistant_message", + text: "[Error] Auto-compaction state is unavailable. Use /autocompact on or /autocompact off.", + }, + }, + ]); + }); + test("rewinds conversation through the Pi tree navigation bridge", async () => { const { pi, session, events } = await createSession(); pi.latestSession().capturedUserEntries = [ diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index b76ab5c7b..e07176a33 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -86,6 +86,19 @@ const QUESTION_COMMENT_HEADER = "Comment"; const PI_ASK_USER_FREEFORM_SENTINEL = "✏️ Type custom response..."; const COMBINED_ASK_USER_METADATA = "ask_user_select_optional_comment"; +const PI_HANDLED_BUILTIN_SLASH_COMMANDS: AgentSlashCommand[] = [ + { + name: "compact", + description: "Manually compact the session context", + argumentHint: "[instructions]", + }, + { + name: "autocompact", + description: "Toggle automatic context compaction", + argumentHint: "[on|off|toggle]", + }, +]; + const PI_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, supportsSessionPersistence: true, @@ -205,6 +218,13 @@ interface ExtensionUiMappingOptions { allowFreeform?: boolean; } +interface PiSlashCommandInvocation { + commandName: string; + args?: string; +} + +type AutoCompactMode = boolean | "toggle" | "unknown"; + function normalizePiModelLabel(label: string): string { return label.trim().replace(/[_\s]+/g, " "); } @@ -247,6 +267,20 @@ function normalizePiThinkingOption(value: string | null | undefined): PiThinking return isPiThinkingLevel(value) ? value : null; } +function parseAutoCompactMode(value: string | undefined): AutoCompactMode { + const mode = (value ?? "toggle").trim().toLowerCase(); + if (mode === "on" || mode === "true" || mode === "enable" || mode === "enabled") { + return true; + } + if (mode === "off" || mode === "false" || mode === "disable" || mode === "disabled") { + return false; + } + if (mode === "toggle") { + return "toggle"; + } + return "unknown"; +} + function mapThinkingOption(option: (typeof PI_THINKING_OPTIONS)[number]) { const mappedOption = { id: option.id, @@ -922,6 +956,9 @@ export class PiRpcAgentSession implements AgentSession { private readonly seenUserEntryIds = new Set(); private readonly pendingUserMessages: PendingPiUserMessage[] = []; private readonly pendingExtensionResults = new Map(); + private outOfBandCompactionEmit: ((event: AgentStreamEvent) => void) | null = null; + private outOfBandCompactionStarted = false; + private outOfBandCompactionCompleted = false; private state: PiSessionState; private closed = false; @@ -1127,11 +1164,46 @@ export class PiRpcAgentSession implements AgentSession { async listCommands(): Promise { const commands = await this.runtimeSession.getCommands(); - return commands.map((command) => ({ - name: command.name, - description: command.description ?? command.source, - argumentHint: "", - })); + const mappedCommands = new Map( + PI_HANDLED_BUILTIN_SLASH_COMMANDS.map((command) => [command.name, { ...command }]), + ); + for (const command of commands) { + const knownCommand = mappedCommands.get(command.name); + mappedCommands.set(command.name, { + name: command.name, + description: command.description ?? command.source, + argumentHint: knownCommand?.argumentHint ?? "", + }); + } + return [...mappedCommands.values()]; + } + + tryHandleOutOfBand( + prompt: AgentPromptInput, + ): { run(ctx: { emit: (event: AgentStreamEvent) => void }): Promise } | null { + if (typeof prompt !== "string") { + return null; + } + const parsed = this.parseSlashCommandInput(prompt); + if (!parsed) { + return null; + } + const commandName = parsed.commandName.toLowerCase(); + if (commandName === "compact") { + return { + run: async ({ emit }) => { + await this.executeCompactCommand(parsed.args, emit); + }, + }; + } + if (commandName === "autocompact") { + return { + run: async ({ emit }) => { + await this.executeAutoCompactCommand(parsed.args, emit); + }, + }; + } + return null; } async setModel(modelId: string | null): Promise { @@ -1172,6 +1244,128 @@ export class PiRpcAgentSession implements AgentSession { return this.activeTurnId ?? undefined; } + private parseSlashCommandInput(text: string): PiSlashCommandInvocation | null { + const trimmed = text.trim(); + if (!trimmed.startsWith("/") || trimmed.length <= 1) { + return null; + } + const withoutPrefix = trimmed.slice(1); + const firstWhitespaceIdx = withoutPrefix.search(/\s/); + const commandName = + firstWhitespaceIdx === -1 ? withoutPrefix : withoutPrefix.slice(0, firstWhitespaceIdx); + if (!commandName || commandName.includes("/")) { + return null; + } + const rawArgs = + firstWhitespaceIdx === -1 ? "" : withoutPrefix.slice(firstWhitespaceIdx + 1).trim(); + return rawArgs.length > 0 ? { commandName, args: rawArgs } : { commandName }; + } + + private async executeCompactCommand( + customInstructions: string | undefined, + emit: (event: AgentStreamEvent) => void, + ): Promise { + if (this.outOfBandCompactionEmit) { + throw new Error("A Pi compact command is already running"); + } + this.outOfBandCompactionEmit = emit; + this.outOfBandCompactionStarted = false; + this.outOfBandCompactionCompleted = false; + try { + await this.runtimeSession.compact(customInstructions); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if ( + this.outOfBandCompactionEmit === emit && + this.outOfBandCompactionStarted && + !this.outOfBandCompactionCompleted + ) { + this.emitCompactionTimeline({ + turnId: undefined, + item: { + type: "compaction", + status: "completed", + trigger: "manual", + }, + }); + } + emit({ + type: "timeline", + provider: PI_PROVIDER, + item: { + type: "assistant_message", + text: `[Error] Failed to compact context: ${message}`, + }, + }); + } finally { + if (this.outOfBandCompactionEmit === emit && !this.outOfBandCompactionStarted) { + this.outOfBandCompactionEmit = null; + this.outOfBandCompactionStarted = false; + this.outOfBandCompactionCompleted = false; + } + } + } + + private async executeAutoCompactCommand( + mode: string | undefined, + emit: (event: AgentStreamEvent) => void, + ): Promise { + let enabled = parseAutoCompactMode(mode); + if (enabled === "unknown") { + emit({ + type: "timeline", + provider: PI_PROVIDER, + item: { + type: "assistant_message", + text: "[Error] Usage: /autocompact [on|off|toggle]", + }, + }); + return; + } + if (enabled === "toggle") { + const state = await this.runtimeSession.getState(); + if (typeof state.autoCompactionEnabled !== "boolean") { + emit({ + type: "timeline", + provider: PI_PROVIDER, + item: { + type: "assistant_message", + text: "[Error] Auto-compaction state is unavailable. Use /autocompact on or /autocompact off.", + }, + }); + return; + } + enabled = !state.autoCompactionEnabled; + } + + try { + await this.runtimeSession.setAutoCompaction(enabled); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + emit({ + type: "timeline", + provider: PI_PROVIDER, + item: { + type: "assistant_message", + text: `[Error] Failed to set auto-compaction: ${message}`, + }, + }); + return; + } + this.state = { + ...this.state, + autoCompactionEnabled: enabled, + }; + emit({ + type: "timeline", + provider: PI_PROVIDER, + item: { + type: "assistant_message", + text: `Auto-compaction ${enabled ? "enabled" : "disabled"}.`, + }, + }); + } + private async requestEntryCapture(reason: string): Promise { const requestId = randomUUID(); const resultPromise = this.waitForExtensionResult(requestId); @@ -1433,9 +1627,7 @@ export class PiRpcAgentSession implements AgentSession { return; } case "compaction_start": - this.emit({ - type: "timeline", - provider: PI_PROVIDER, + this.emitCompactionTimeline({ turnId, item: { type: "compaction", @@ -1445,13 +1637,12 @@ export class PiRpcAgentSession implements AgentSession { }); return; case "compaction_end": - this.emit({ - type: "timeline", - provider: PI_PROVIDER, + this.emitCompactionTimeline({ turnId, item: { type: "compaction", status: "completed", + trigger: event.reason === "manual" ? "manual" : "auto", }, }); return; @@ -1463,6 +1654,37 @@ export class PiRpcAgentSession implements AgentSession { } } + private emitCompactionTimeline(input: { + turnId: string | undefined; + item: Extract["item"]; + }): void { + const emitOutOfBand = this.outOfBandCompactionEmit; + if (emitOutOfBand && input.item.type === "compaction") { + if (input.item.status === "loading") { + this.outOfBandCompactionStarted = true; + } + if (input.item.status === "completed") { + this.outOfBandCompactionCompleted = true; + } + } + const event: AgentStreamEvent = { + type: "timeline", + provider: PI_PROVIDER, + ...(emitOutOfBand ? {} : { turnId: input.turnId }), + item: input.item, + }; + if (emitOutOfBand) { + emitOutOfBand(event); + if (input.item.type === "compaction" && input.item.status === "completed") { + this.outOfBandCompactionEmit = null; + this.outOfBandCompactionStarted = false; + this.outOfBandCompactionCompleted = false; + } + return; + } + this.emit(event); + } + private handleMessageUpdate( event: Extract, turnId: string | undefined, diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.ts index 84f8cbaf9..07af803a0 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.ts @@ -129,6 +129,17 @@ class PiCliRuntimeSession implements PiRuntimeSession { await this.request({ type: "prompt", message, ...(images?.length ? { images } : {}) }); } + async compact(customInstructions?: string): Promise { + await this.request({ + type: "compact", + ...(customInstructions ? { customInstructions } : {}), + }); + } + + async setAutoCompaction(enabled: boolean): Promise { + await this.request({ type: "set_auto_compaction", enabled }); + } + async abort(): Promise { await this.request({ type: "abort" }); } diff --git a/packages/server/src/server/agent/providers/pi/rpc-types.ts b/packages/server/src/server/agent/providers/pi/rpc-types.ts index 90f9aa4e1..b25a61659 100644 --- a/packages/server/src/server/agent/providers/pi/rpc-types.ts +++ b/packages/server/src/server/agent/providers/pi/rpc-types.ts @@ -79,6 +79,7 @@ export interface PiSessionState { thinkingLevel: PiThinkingLevel; isStreaming: boolean; isCompacting: boolean; + autoCompactionEnabled?: boolean; sessionFile?: string; sessionId: string; sessionName?: string; @@ -111,6 +112,8 @@ export interface PiRpcSlashCommand { export type PiRpcCommand = | { id?: string; type: "prompt"; message: string; images?: PiImageContent[] } + | { id?: string; type: "compact"; customInstructions?: string } + | { id?: string; type: "set_auto_compaction"; enabled: boolean } | { id?: string; type: "abort" } | { id?: string; type: "get_state" } | { id?: string; type: "get_messages" } diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index e479c5894..0d813c12f 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/packages/server/src/server/agent/providers/pi/runtime.ts @@ -37,6 +37,8 @@ export interface PiRuntimeSession { message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, ): Promise; + compact(customInstructions?: string): Promise; + setAutoCompaction(enabled: boolean): Promise; abort(): Promise; getState(): Promise; getMessages(): Promise; diff --git a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts index 08d92f201..53121c23a 100644 --- a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts +++ b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts @@ -51,6 +51,8 @@ export class FakePi implements PiRuntime { export class FakePiSession implements PiRuntimeSession { readonly prompts: Array<{ message: string; imageCount: number }> = []; + readonly compactRequests: Array<{ customInstructions?: string }> = []; + readonly setAutoCompactionRequests: boolean[] = []; readonly setModelRequests: Array<{ provider: string; modelId: string }> = []; readonly setThinkingLevelRequests: string[] = []; readonly treeNavigationRequests: string[] = []; @@ -69,6 +71,8 @@ export class FakePiSession implements PiRuntimeSession { cost: 0, }; commands: PiRpcSlashCommand[] = []; + compactError: Error | null = null; + emitCompactEnd = true; state: PiSessionState; private readonly subscribers = new Set<(event: PiRuntimeEvent) => void>(); @@ -79,6 +83,7 @@ export class FakePiSession implements PiRuntimeSession { thinkingLevel: "medium", isStreaming: false, isCompacting: false, + autoCompactionEnabled: true, sessionFile: launch.session ?? "/tmp/pi-session", sessionId: "pi-session-1", messageCount: 0, @@ -102,6 +107,25 @@ export class FakePiSession implements PiRuntimeSession { this.handleEntryCaptureCommand(message); } + async compact(customInstructions?: string): Promise { + this.compactRequests.push(customInstructions === undefined ? {} : { customInstructions }); + this.emit({ type: "compaction_start", reason: "manual" }); + if (this.emitCompactEnd) { + this.emit({ type: "compaction_end", reason: "manual" }); + } + if (this.compactError) { + throw this.compactError; + } + } + + async setAutoCompaction(enabled: boolean): Promise { + this.setAutoCompactionRequests.push(enabled); + this.state = { + ...this.state, + autoCompactionEnabled: enabled, + }; + } + async abort(): Promise { this.abortRequested = true; } diff --git a/packages/server/src/server/auto-archive-on-merge/archive-if-safe.test.ts b/packages/server/src/server/auto-archive-on-merge/archive-if-safe.test.ts index 789326195..6531cde4a 100644 --- a/packages/server/src/server/auto-archive-on-merge/archive-if-safe.test.ts +++ b/packages/server/src/server/auto-archive-on-merge/archive-if-safe.test.ts @@ -233,15 +233,15 @@ describe("archiveIfSafe", () => { expect(harness.deps.archivePaseoWorktree).not.toHaveBeenCalled(); }); - test("does nothing when the upstream status is unknown", async () => { + test("archives when the PR is merged and the upstream branch was deleted", async () => { const harness = createHarness({ - getSnapshot: async () => createSnapshot({ git: { aheadOfOrigin: null } }), + getSnapshot: async () => + createSnapshot({ git: { aheadOfOrigin: null, behindOfOrigin: null } }), }); await runArchiveIfSafe(harness); - expect(harness.deps.isPaseoOwnedWorktreeCwd).not.toHaveBeenCalled(); - expect(harness.deps.archivePaseoWorktree).not.toHaveBeenCalled(); + expect(harness.deps.archivePaseoWorktree).toHaveBeenCalledTimes(1); }); test("does nothing when the cwd is not a Paseo-owned worktree", async () => { diff --git a/packages/server/src/server/auto-archive-on-merge/archive-if-safe.ts b/packages/server/src/server/auto-archive-on-merge/archive-if-safe.ts index d98ceaa86..7831a733e 100644 --- a/packages/server/src/server/auto-archive-on-merge/archive-if-safe.ts +++ b/packages/server/src/server/auto-archive-on-merge/archive-if-safe.ts @@ -78,10 +78,10 @@ export async function archiveIfSafe(input: { return; } - if (snapshot.git.isDirty === true || snapshot.git.aheadOfOrigin === null) { + if (snapshot.git.isDirty === true) { return; } - if (snapshot.git.aheadOfOrigin > 0) { + if (typeof snapshot.git.aheadOfOrigin === "number" && snapshot.git.aheadOfOrigin > 0) { return; } diff --git a/packages/server/src/server/daemon-e2e/pi.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/pi.real.e2e.test.ts index 8b2a9cd40..8140a6517 100644 --- a/packages/server/src/server/daemon-e2e/pi.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/pi.real.e2e.test.ts @@ -1,4 +1,4 @@ -import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; import { randomUUID } from "node:crypto"; @@ -24,6 +24,7 @@ process.env.PASEO_SUPERVISED = "0"; const PI_TEST_TIMEOUT_MS = 240_000; const PI_REAL_TEST_MODEL = getRealProviderConfig("pi").model; +const PI_FREE_COMPACTION_TEST_MODEL = "openrouter/openai/gpt-oss-20b:free"; type ToolCallItem = Extract; @@ -31,6 +32,18 @@ function tmpCwd(prefix = "daemon-real-pi-"): string { return mkdtempSync(path.join(tmpdir(), prefix)); } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function writePiCompactionSettings( + cwd: string, + compaction: { enabled: boolean; reserveTokens?: number; keepRecentTokens?: number }, +): void { + mkdirSync(path.join(cwd, ".pi"), { recursive: true }); + writeFileSync(path.join(cwd, ".pi/settings.json"), JSON.stringify({ compaction }, null, 2)); +} + function createPiClient(): AgentClient { return createRealProviderClient("pi", pino({ level: "silent" })); } @@ -68,6 +81,27 @@ async function fetchCanonicalTimeline( return timeline.entries.map((entry) => entry.item); } +async function waitForTimelineItem( + client: DaemonClient, + agentId: string, + predicate: (item: AgentTimelineItem) => boolean, + timeoutMs = PI_TEST_TIMEOUT_MS, +): Promise { + const deadline = Date.now() + timeoutMs; + let lastItems: AgentTimelineItem[] = []; + while (Date.now() < deadline) { + lastItems = await fetchCanonicalTimeline(client, agentId); + const item = lastItems.find(predicate); + if (item) { + return item; + } + await sleep(250); + } + throw new Error( + `Timed out waiting for Pi timeline item. Last timeline: ${JSON.stringify(lastItems)}`, + ); +} + async function withConnectedPiDaemon( run: (context: { client: DaemonClient }) => Promise, ): Promise { @@ -101,6 +135,200 @@ beforeEach((context) => { } }); +test( + "real Pi daemon lists Paseo-handled compact slash commands", + async () => { + const cwd = tmpCwd("pi-compact-commands-"); + + try { + await withConnectedPiDaemon(async ({ client }) => { + const agent = await client.createAgent({ + cwd, + title: "pi-compact-commands", + provider: "pi", + model: PI_FREE_COMPACTION_TEST_MODEL, + }); + + const result = await client.listCommands(agent.id); + expect(result.commands).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + name: "compact", + description: "Manually compact the session context", + }), + expect.objectContaining({ + name: "autocompact", + description: "Toggle automatic context compaction", + }), + ]), + ); + }); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, + PI_TEST_TIMEOUT_MS, +); + +test( + "real Pi daemon executes manual compact out-of-band instead of prompt text", + async () => { + const cwd = tmpCwd("pi-manual-compact-"); + + try { + writePiCompactionSettings(cwd, { + enabled: true, + keepRecentTokens: 1, + }); + + await withConnectedPiDaemon(async ({ client }) => { + const agent = await client.createAgent({ + cwd, + title: "pi-manual-compact", + provider: "pi", + model: PI_FREE_COMPACTION_TEST_MODEL, + }); + + await client.sendMessage(agent.id, "Reply exactly: compact-ready"); + const finish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS); + expect(finish.status).toBe("idle"); + + await client.sendMessage(agent.id, "/compact summarize this e2e run"); + + await waitForTimelineItem( + client, + agent.id, + (item) => item.type === "compaction" && item.status === "completed", + ); + + const items = await fetchCanonicalTimeline(client, agent.id); + expect(items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "compaction", + status: "loading", + trigger: "manual", + }), + expect.objectContaining({ + type: "compaction", + status: "completed", + }), + ]), + ); + expect( + items.some((item) => item.type === "user_message" && item.text.includes("/compact")), + ).toBe(false); + expect( + items.some( + (item) => item.type === "assistant_message" && item.text.includes("Failed to compact"), + ), + ).toBe(false); + }); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, + PI_TEST_TIMEOUT_MS, +); + +test( + "real Pi daemon toggles auto-compaction out-of-band instead of prompt text", + async () => { + const cwd = tmpCwd("pi-autocompact-toggle-"); + + try { + await withConnectedPiDaemon(async ({ client }) => { + const agent = await client.createAgent({ + cwd, + title: "pi-autocompact-toggle", + provider: "pi", + model: PI_FREE_COMPACTION_TEST_MODEL, + }); + + await client.sendMessage(agent.id, "/autocompact off"); + await waitForTimelineItem( + client, + agent.id, + (item) => item.type === "assistant_message" && item.text === "Auto-compaction disabled.", + ); + + await client.sendMessage(agent.id, "/autocompact"); + await waitForTimelineItem( + client, + agent.id, + (item) => item.type === "assistant_message" && item.text === "Auto-compaction enabled.", + ); + + const items = await fetchCanonicalTimeline(client, agent.id); + expect( + items.some((item) => item.type === "user_message" && item.text.includes("/autocompact")), + ).toBe(false); + }); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, + PI_TEST_TIMEOUT_MS, +); + +test( + "real Pi daemon surfaces automatic threshold compaction events", + async () => { + const cwd = tmpCwd("pi-auto-compact-"); + + try { + writePiCompactionSettings(cwd, { + enabled: true, + reserveTokens: 126_000, + keepRecentTokens: 1, + }); + + await withConnectedPiDaemon(async ({ client }) => { + const agent = await client.createAgent({ + cwd, + title: "pi-auto-compact", + provider: "pi", + model: PI_FREE_COMPACTION_TEST_MODEL, + }); + + await client.sendMessage(agent.id, "Reply exactly: auto-compact-ready"); + const finish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS); + expect(finish.status, JSON.stringify(finish)).toBe("idle"); + + await waitForTimelineItem( + client, + agent.id, + (item) => item.type === "compaction" && item.status === "completed", + ); + + const items = await fetchCanonicalTimeline(client, agent.id); + expect(items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "compaction", + status: "loading", + trigger: "auto", + }), + expect.objectContaining({ + type: "compaction", + status: "completed", + }), + ]), + ); + expect( + items.some( + (item) => + item.type === "assistant_message" && item.text.includes("Auto-compaction failed"), + ), + ).toBe(false); + }); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, + PI_TEST_TIMEOUT_MS, +); + test( "bash tool call records completed shell detail and output", async () => { diff --git a/packages/server/src/server/speech/providers/local/runtime.ts b/packages/server/src/server/speech/providers/local/runtime.ts index 0653a75e5..ba9ec9aac 100644 --- a/packages/server/src/server/speech/providers/local/runtime.ts +++ b/packages/server/src/server/speech/providers/local/runtime.ts @@ -156,6 +156,7 @@ export async function initializeLocalSpeechServices(params: { const workerClient = localConfig ? new LocalSpeechWorkerClient({ + logger, config: { modelsDir: localConfig.modelsDir, voiceSttModel: localModels.voiceLocalSttModel, diff --git a/packages/server/src/server/speech/providers/local/worker-client.test.ts b/packages/server/src/server/speech/providers/local/worker-client.test.ts index b42ff4adc..3c028be13 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.test.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.test.ts @@ -20,6 +20,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; class FakeLocalSpeechWorker extends EventEmitter { public connected = true; public killed = false; + public pid = 12345; + public readonly stderr = new EventEmitter() as NodeJS.ReadableStream; public readonly sent: LocalSpeechWorkerRequest[] = []; public disconnects = 0; public kills = 0; @@ -88,9 +90,9 @@ class PausedIpcWorker { } on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this; - on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; + on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; on( - event: "message" | "exit", + event: "message" | "close", listener: | ((message: LocalSpeechWorkerToParentMessage) => void) | ((code: number | null, signal: NodeJS.Signals | null) => void), @@ -103,6 +105,7 @@ class PausedIpcWorker { function createClient(options?: { idleTtlMs?: number }) { const workers: FakeLocalSpeechWorker[] = []; const client = new LocalSpeechWorkerClient({ + logger: pino({ level: "silent" }), config: { modelsDir: "/tmp/models", voiceSttModel: "parakeet-tdt-0.6b-v2-int8", @@ -120,6 +123,19 @@ function createClient(options?: { idleTtlMs?: number }) { return { client, workers }; } +function createCapturingLogger(): { logger: pino.Logger; records: Array> } { + const records: Array> = []; + const logger = pino( + { level: "trace" }, + { + write(line: string) { + records.push(JSON.parse(line) as Record); + }, + }, + ); + return { logger, records }; +} + async function waitForMicrotasks(): Promise { await new Promise((resolve) => setTimeout(resolve, 0)); } @@ -217,6 +233,7 @@ describe("LocalSpeechWorkerClient", () => { it("does not surface real IPC backpressure when replaying native-sized dictation frames", async () => { const workers: PausedIpcWorker[] = []; const client = new LocalSpeechWorkerClient({ + logger: pino({ level: "silent" }), config: { modelsDir: "/tmp/models", voiceSttModel: "parakeet-tdt-0.6b-v2-int8", @@ -257,6 +274,80 @@ describe("LocalSpeechWorkerClient", () => { } }); + it("logs worker exit details and includes actionable context in the surfaced error", async () => { + const { logger, records } = createCapturingLogger(); + const workers: FakeLocalSpeechWorker[] = []; + const client = new LocalSpeechWorkerClient({ + logger, + config: { + modelsDir: "/tmp/models", + voiceSttModel: "parakeet-tdt-0.6b-v2-int8", + dictationSttModel: "parakeet-tdt-0.6b-v2-int8", + voiceTtsModel: "kokoro-en-v0_19", + }, + forkWorker: () => { + const worker = new FakeLocalSpeechWorker(); + workers.push(worker); + return worker; + }, + }); + const provider = new WorkerBackedSpeechToTextProvider(client, "dictationStt"); + const session = provider.createSession({ logger: pino({ level: "silent" }) }); + + const connect = session.connect(); + workers[0].emit("exit", null, "SIGABRT"); + workers[0].stderr.emit("data", "dyld: Library not loaded: libsherpa-onnx-c-api.dylib"); + workers[0].emit("close", null, "SIGABRT"); + + await expect(connect).rejects.toThrow( + "Local speech worker exited (signal SIGABRT) while handling session.create (dictationStt). Last stderr: dyld: Library not loaded: libsherpa-onnx-c-api.dylib", + ); + const exitRecord = records.find((record) => record.msg === "Local speech worker exited"); + expect(exitRecord).toMatchObject({ + workerPid: 12345, + signal: "SIGABRT", + stderrTail: "dyld: Library not loaded: libsherpa-onnx-c-api.dylib", + pendingRequests: [expect.objectContaining({ type: "session.create", kind: "dictationStt" })], + }); + }); + + it("does not log intentional shutdowns as worker crashes", async () => { + const { logger, records } = createCapturingLogger(); + const workers: FakeLocalSpeechWorker[] = []; + const client = new LocalSpeechWorkerClient({ + logger, + config: { + modelsDir: "/tmp/models", + voiceSttModel: "parakeet-tdt-0.6b-v2-int8", + dictationSttModel: "parakeet-tdt-0.6b-v2-int8", + voiceTtsModel: "kokoro-en-v0_19", + }, + forkWorker: () => { + const worker = new FakeLocalSpeechWorker(); + workers.push(worker); + return worker; + }, + }); + + const synthesize = client.synthesizeSpeech("hello"); + workers[0].respond(workers[0].sent[0], { + audio: bufferToWorkerBytes(Buffer.from([1])), + format: "pcm;rate=24000", + }); + await synthesize; + + client.shutdown(); + workers[0].emit("close", null, "SIGTERM"); + + expect(records.find((record) => record.msg === "Local speech worker exited")).toBeUndefined(); + expect( + records.find((record) => record.msg === "Local speech worker closed after shutdown"), + ).toMatchObject({ + workerPid: 12345, + signal: "SIGTERM", + }); + }); + it("forwards VAD session events through the shared worker", async () => { const { client, workers } = createClient(); const provider = new WorkerBackedTurnDetectionProvider(client); diff --git a/packages/server/src/server/speech/providers/local/worker-client.ts b/packages/server/src/server/speech/providers/local/worker-client.ts index 88c0d46fb..1f0fa96b9 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.ts @@ -28,6 +28,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; const DEFAULT_REQUEST_TIMEOUT_MS = 30000; const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000; const DEFAULT_LOCAL_SAMPLE_RATE = 16000; +const STDERR_TAIL_MAX_CHARS = 8000; +const USER_ERROR_STDERR_MAX_CHARS = 1000; type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Request ? Request extends LocalSpeechWorkerRequest @@ -38,21 +40,27 @@ type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Requ interface LocalSpeechWorkerProcess { connected: boolean; killed: boolean; + pid?: number; + stderr?: Readable | null; send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean; disconnect(): void; kill(): boolean; on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this; - on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; + on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; } interface PendingRequest { resolve: (value: unknown) => void; reject: (error: Error) => void; timeout: ReturnType; + type: LocalSpeechWorkerRequest["type"]; + summary: Record; + startedAt: number; } interface LocalSpeechWorkerClientOptions { config: LocalSpeechWorkerConfig; + logger: pino.Logger; requestTimeoutMs?: number; idleTtlMs?: number; forkWorker?: () => LocalSpeechWorkerProcess; @@ -90,7 +98,7 @@ function forkLocalSpeechWorker(): LocalSpeechWorkerProcess { env, execArgv: resolveWorkerExecArgv(), serialization: "advanced", - stdio: ["ignore", "ignore", "inherit", "ipc"], + stdio: ["ignore", "ignore", "pipe", "ipc"], }) as LocalSpeechWorkerProcess; } @@ -100,8 +108,75 @@ function isResponse( return message.type === "response"; } +function truncateStart(value: string, maxChars: number): string { + if (value.length <= maxChars) { + return value; + } + return value.slice(value.length - maxChars); +} + +function summarizeWorkerRequest(message: LocalSpeechWorkerRequest): Record { + switch (message.type) { + case "session.create": + return { kind: message.kind, sessionId: message.sessionId }; + case "session.append": + return { sessionId: message.sessionId, audioBytes: message.audio.byteLength }; + case "session.commit": + case "session.clear": + case "session.flush": + case "session.reset": + case "session.close": + return { sessionId: message.sessionId }; + case "stt.transcribe": + return { + model: message.model, + audioBytes: message.audio.byteLength, + format: message.format, + }; + case "tts.synthesize": + return { textLength: message.text.length }; + } +} + +function formatExitStatus(code: number | null, signal: NodeJS.Signals | null): string { + const parts: string[] = []; + if (code !== null) { + parts.push(`code ${code}`); + } + if (signal) { + parts.push(`signal ${signal}`); + } + return parts.length > 0 ? parts.join(", ") : "unknown exit status"; +} + +function formatPendingRequestForMessage(request: Record): string { + const type = typeof request.type === "string" ? request.type : "unknown request"; + const kind = typeof request.kind === "string" ? ` (${request.kind})` : ""; + return `${type}${kind}`; +} + +function buildWorkerExitMessage(params: { + code: number | null; + signal: NodeJS.Signals | null; + pendingRequests: Array>; + stderrTail: string; +}): string { + const pending = + params.pendingRequests.length > 0 + ? ` while handling ${params.pendingRequests + .slice(0, 3) + .map(formatPendingRequestForMessage) + .join(", ")}` + : ""; + const stderr = params.stderrTail + ? ` Last stderr: ${truncateStart(params.stderrTail, USER_ERROR_STDERR_MAX_CHARS)}` + : " Check daemon.log and macOS DiagnosticReports for Paseo Voice crash details."; + return `Local speech worker exited (${formatExitStatus(params.code, params.signal)})${pending}.${stderr}`; +} + export class LocalSpeechWorkerClient { private readonly config: LocalSpeechWorkerConfig; + private readonly logger: pino.Logger; private readonly requestTimeoutMs: number; private readonly idleTtlMs: number; private readonly forkWorker: () => LocalSpeechWorkerProcess; @@ -109,11 +184,16 @@ export class LocalSpeechWorkerClient { private readonly activeSessionIds = new Set(); private readonly sessionEmitters = new Map(); private worker: LocalSpeechWorkerProcess | null = null; + private workerPid: number | null = null; + private stderrTail = ""; + private stderrLineBuffer = ""; private inFlightRequests = 0; private idleTimer: ReturnType | null = null; + private readonly intentionalWorkerCloses = new WeakSet(); constructor(options: LocalSpeechWorkerClientOptions) { this.config = options.config; + this.logger = options.logger.child({ component: "local-speech-worker-client" }); this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS; this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker; @@ -214,7 +294,9 @@ export class LocalSpeechWorkerClient { this.sessionEmitters.clear(); const worker = this.worker; this.worker = null; + this.workerPid = null; if (worker && !worker.killed) { + this.intentionalWorkerCloses.add(worker); try { worker.disconnect(); } catch { @@ -232,6 +314,7 @@ export class LocalSpeechWorkerClient { const worker = this.ensureWorker(); const requestId = randomUUID(); const message = { ...input, requestId } as LocalSpeechWorkerRequest; + const requestSummary = summarizeWorkerRequest(message); this.inFlightRequests++; this.clearIdleTimer(); @@ -246,6 +329,9 @@ export class LocalSpeechWorkerClient { resolve: (value) => resolve(value as T), reject, timeout, + type: message.type, + summary: requestSummary, + startedAt: Date.now(), }); worker.send(message, (error) => { @@ -271,11 +357,44 @@ export class LocalSpeechWorkerClient { } const worker = this.forkWorker(); this.worker = worker; + this.workerPid = worker.pid ?? null; + this.stderrTail = ""; + this.stderrLineBuffer = ""; + this.logger.info( + { + workerPid: this.workerPid, + modelsDir: this.config.modelsDir, + voiceSttModel: this.config.voiceSttModel, + dictationSttModel: this.config.dictationSttModel, + voiceTtsModel: this.config.voiceTtsModel, + }, + "Local speech worker spawned", + ); + worker.stderr?.on("data", (chunk: Buffer | string) => this.handleWorkerStderr(chunk)); worker.on("message", (message) => this.handleWorkerMessage(message)); - worker.on("exit", () => this.handleWorkerExit()); + worker.on("close", (code, signal) => this.handleWorkerExit(worker, code, signal)); return worker; } + private handleWorkerStderr(chunk: Buffer | string): void { + const text = Buffer.isBuffer(chunk) ? chunk.toString("utf8") : chunk; + this.stderrTail = truncateStart(this.stderrTail + text, STDERR_TAIL_MAX_CHARS); + + this.stderrLineBuffer += text; + const lines = this.stderrLineBuffer.split(/\r?\n/); + this.stderrLineBuffer = lines.pop() ?? ""; + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) { + continue; + } + this.logger.warn( + { workerPid: this.workerPid, stderr: trimmed }, + "Local speech worker stderr", + ); + } + } + private handleWorkerMessage(message: LocalSpeechWorkerToParentMessage): void { if (isResponse(message)) { const pending = this.pendingRequests.get(message.requestId); @@ -317,13 +436,84 @@ export class LocalSpeechWorkerClient { } } - private handleWorkerExit(): void { - this.worker = null; + private handleWorkerExit( + worker: LocalSpeechWorkerProcess, + code: number | null, + signal: NodeJS.Signals | null, + ): void { + const wasCurrentWorker = this.worker === worker; + const wasIntentionalClose = this.intentionalWorkerCloses.has(worker); + this.intentionalWorkerCloses.delete(worker); + const workerPid = worker.pid ?? (wasCurrentWorker ? this.workerPid : null); + const pendingRequests = this.describePendingRequests(); + const activeSessionCount = this.activeSessionIds.size; + const stderrTail = this.getStderrTail(); + + if (wasIntentionalClose) { + this.logger.info( + { + workerPid, + code, + signal, + pendingRequests, + activeSessionCount, + stderrTail: stderrTail || null, + }, + "Local speech worker closed after shutdown", + ); + if (wasCurrentWorker) { + this.worker = null; + this.workerPid = null; + } + return; + } + + if (!wasCurrentWorker) { + this.logger.warn( + { + workerPid, + code, + signal, + pendingRequests, + activeSessionCount, + stderrTail: stderrTail || null, + }, + "Stale local speech worker closed", + ); + return; + } + + const error = new Error( + buildWorkerExitMessage({ + code, + signal, + pendingRequests, + stderrTail, + }), + ); + + this.logger.error( + { + err: error, + workerPid, + code, + signal, + pendingRequests, + activeSessionCount, + stderrTail: stderrTail || null, + }, + "Local speech worker exited", + ); + + if (wasCurrentWorker) { + this.worker = null; + this.workerPid = null; + } this.clearIdleTimer(); - this.rejectAllPending(new Error("Local speech worker exited")); + this.rejectAllPending(error); for (const [sessionId, emitter] of this.sessionEmitters) { if (this.activeSessionIds.has(sessionId)) { - emitter.emit("error", new Error("Local speech worker exited")); + this.emitErrorIfObserved(sessionId, emitter, error, workerPid); } } this.activeSessionIds.clear(); @@ -331,6 +521,24 @@ export class LocalSpeechWorkerClient { this.inFlightRequests = 0; } + private describePendingRequests(): Array> { + const now = Date.now(); + return Array.from(this.pendingRequests.entries()).map(([requestId, request]) => + Object.assign( + { + requestId, + type: request.type, + ageMs: Math.max(0, now - request.startedAt), + }, + request.summary, + ), + ); + } + + private getStderrTail(): string { + return truncateStart(this.stderrTail.trim(), STDERR_TAIL_MAX_CHARS); + } + private rejectAllPending(error: Error): void { for (const [requestId, pending] of this.pendingRequests) { clearTimeout(pending.timeout); @@ -344,7 +552,27 @@ export class LocalSpeechWorkerClient { if (!emitter) { return; } - emitter.emit("error", error instanceof Error ? error : new Error(String(error))); + this.emitErrorIfObserved( + sessionId, + emitter, + error instanceof Error ? error : new Error(String(error)), + ); + } + + private emitErrorIfObserved( + sessionId: string, + emitter: EventEmitter, + error: Error, + workerPid: number | null = this.workerPid, + ): void { + if (emitter.listenerCount("error") > 0) { + emitter.emit("error", error); + return; + } + this.logger.warn( + { err: error, workerPid, sessionId }, + "Local speech worker session error had no listener", + ); } private scheduleIdleShutdownIfReady(): void {