Merge branch 'main' of github.com:getpaseo/paseo
This commit is contained in:
@@ -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 {
|
function findToolByCallId(state: StreamItem[], callId: string): AgentToolCallItem | undefined {
|
||||||
return state.find(
|
return state.find(
|
||||||
(item): item is AgentToolCallItem =>
|
(item): item is AgentToolCallItem =>
|
||||||
@@ -693,6 +708,27 @@ describe("stream reducer canonical tool calls", () => {
|
|||||||
assert.strictEqual(todos.items[1]?.completed, true);
|
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<StreamItem, { kind: "compaction" }> => 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", () => {
|
it("renders Claude TodoWrite as todo_list and suppresses tool call badge", () => {
|
||||||
const state = hydrateStreamState([
|
const state = hydrateStreamState([
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -726,8 +726,8 @@ function reduceTimelineCompaction(
|
|||||||
const updated: CompactionItem = {
|
const updated: CompactionItem = {
|
||||||
...existing,
|
...existing,
|
||||||
status: "completed",
|
status: "completed",
|
||||||
trigger: item.trigger,
|
trigger: item.trigger ?? existing.trigger,
|
||||||
preTokens: item.preTokens,
|
preTokens: item.preTokens ?? existing.preTokens,
|
||||||
};
|
};
|
||||||
return [...state.slice(0, loadingIdx), updated, ...state.slice(loadingIdx + 1)];
|
return [...state.slice(0, loadingIdx), updated, ...state.slice(loadingIdx + 1)];
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -534,6 +534,37 @@ describe("Codex app-server provider", () => {
|
|||||||
await session.close();
|
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 () => {
|
test("rewinds the conversation to a freshly emitted Codex user message id", async () => {
|
||||||
const appServer = createFakeCodexAppServer();
|
const appServer = createFakeCodexAppServer();
|
||||||
const session = new CodexAppServerAgentSession(
|
const session = new CodexAppServerAgentSession(
|
||||||
@@ -2703,7 +2734,17 @@ describe("Codex persisted sessions", () => {
|
|||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
expect(calls).toEqual([
|
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" } },
|
{ method: "thread/list", params: { limit: 50, cwd: "/workspace/project-a" } },
|
||||||
]);
|
]);
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -103,6 +103,14 @@ const TURN_START_TIMEOUT_MS = 90 * 1000;
|
|||||||
const INTERRUPT_TIMEOUT_MS = 2_000;
|
const INTERRUPT_TIMEOUT_MS = 2_000;
|
||||||
const CODEX_PROVIDER = "codex" as const;
|
const CODEX_PROVIDER = "codex" as const;
|
||||||
const CODEX_IMAGE_ATTACHMENT_DIR = "paseo-attachments";
|
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 ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN = "\n\n---\n\n";
|
||||||
const CODEX_TOOL_THREAD_ITEM_TYPES = new Set([
|
const CODEX_TOOL_THREAD_ITEM_TYPES = new Set([
|
||||||
"commandExecution",
|
"commandExecution",
|
||||||
@@ -2818,11 +2826,7 @@ function buildCodexAppServerInitializeParams(): {
|
|||||||
capabilities: { experimentalApi: true };
|
capabilities: { experimentalApi: true };
|
||||||
} {
|
} {
|
||||||
return {
|
return {
|
||||||
clientInfo: {
|
clientInfo: CODEX_NON_ORIGINATING_APP_SERVER_CLIENT_INFO,
|
||||||
name: "paseo",
|
|
||||||
title: "Paseo",
|
|
||||||
version: "0.0.0",
|
|
||||||
},
|
|
||||||
capabilities: {
|
capabilities: {
|
||||||
experimentalApi: true,
|
experimentalApi: true,
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import { closeSync, existsSync, fstatSync, openSync, readSync } from "node:fs";
|
|||||||
import pino from "pino";
|
import pino from "pino";
|
||||||
import { describe, expect, test } from "vitest";
|
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 { PiRpcAgentClient, PiRpcAgentSession, transformPiModels } from "./agent.js";
|
||||||
import { FakePi } from "./test-utils/fake-pi.js";
|
import { FakePi } from "./test-utils/fake-pi.js";
|
||||||
|
|
||||||
@@ -693,12 +693,202 @@ describe("PiRpcAgentClient", () => {
|
|||||||
];
|
];
|
||||||
|
|
||||||
await expect(session.listCommands()).resolves.toEqual([
|
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: "review", description: "Review changes", argumentHint: "" },
|
||||||
{ name: "fix-tests", description: "Fix tests", argumentHint: "" },
|
{ name: "fix-tests", description: "Fix tests", argumentHint: "" },
|
||||||
{ name: "skill:docs", description: "Read docs", 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 () => {
|
test("rewinds conversation through the Pi tree navigation bridge", async () => {
|
||||||
const { pi, session, events } = await createSession();
|
const { pi, session, events } = await createSession();
|
||||||
pi.latestSession().capturedUserEntries = [
|
pi.latestSession().capturedUserEntries = [
|
||||||
|
|||||||
@@ -86,6 +86,19 @@ const QUESTION_COMMENT_HEADER = "Comment";
|
|||||||
const PI_ASK_USER_FREEFORM_SENTINEL = "✏️ Type custom response...";
|
const PI_ASK_USER_FREEFORM_SENTINEL = "✏️ Type custom response...";
|
||||||
const COMBINED_ASK_USER_METADATA = "ask_user_select_optional_comment";
|
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 = {
|
const PI_CAPABILITIES: AgentCapabilityFlags = {
|
||||||
supportsStreaming: true,
|
supportsStreaming: true,
|
||||||
supportsSessionPersistence: true,
|
supportsSessionPersistence: true,
|
||||||
@@ -205,6 +218,13 @@ interface ExtensionUiMappingOptions {
|
|||||||
allowFreeform?: boolean;
|
allowFreeform?: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
interface PiSlashCommandInvocation {
|
||||||
|
commandName: string;
|
||||||
|
args?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
type AutoCompactMode = boolean | "toggle" | "unknown";
|
||||||
|
|
||||||
function normalizePiModelLabel(label: string): string {
|
function normalizePiModelLabel(label: string): string {
|
||||||
return label.trim().replace(/[_\s]+/g, " ");
|
return label.trim().replace(/[_\s]+/g, " ");
|
||||||
}
|
}
|
||||||
@@ -247,6 +267,20 @@ function normalizePiThinkingOption(value: string | null | undefined): PiThinking
|
|||||||
return isPiThinkingLevel(value) ? value : null;
|
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]) {
|
function mapThinkingOption(option: (typeof PI_THINKING_OPTIONS)[number]) {
|
||||||
const mappedOption = {
|
const mappedOption = {
|
||||||
id: option.id,
|
id: option.id,
|
||||||
@@ -922,6 +956,9 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
private readonly seenUserEntryIds = new Set<string>();
|
private readonly seenUserEntryIds = new Set<string>();
|
||||||
private readonly pendingUserMessages: PendingPiUserMessage[] = [];
|
private readonly pendingUserMessages: PendingPiUserMessage[] = [];
|
||||||
private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
|
private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
|
||||||
|
private outOfBandCompactionEmit: ((event: AgentStreamEvent) => void) | null = null;
|
||||||
|
private outOfBandCompactionStarted = false;
|
||||||
|
private outOfBandCompactionCompleted = false;
|
||||||
private state: PiSessionState;
|
private state: PiSessionState;
|
||||||
private closed = false;
|
private closed = false;
|
||||||
|
|
||||||
@@ -1127,11 +1164,46 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
|
|
||||||
async listCommands(): Promise<AgentSlashCommand[]> {
|
async listCommands(): Promise<AgentSlashCommand[]> {
|
||||||
const commands = await this.runtimeSession.getCommands();
|
const commands = await this.runtimeSession.getCommands();
|
||||||
return commands.map((command) => ({
|
const mappedCommands = new Map<string, AgentSlashCommand>(
|
||||||
name: command.name,
|
PI_HANDLED_BUILTIN_SLASH_COMMANDS.map((command) => [command.name, { ...command }]),
|
||||||
description: command.description ?? command.source,
|
);
|
||||||
argumentHint: "",
|
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<void> } | 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<void> {
|
async setModel(modelId: string | null): Promise<void> {
|
||||||
@@ -1172,6 +1244,128 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
return this.activeTurnId ?? undefined;
|
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<void> {
|
||||||
|
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<void> {
|
||||||
|
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<void> {
|
private async requestEntryCapture(reason: string): Promise<void> {
|
||||||
const requestId = randomUUID();
|
const requestId = randomUUID();
|
||||||
const resultPromise = this.waitForExtensionResult(requestId);
|
const resultPromise = this.waitForExtensionResult(requestId);
|
||||||
@@ -1433,9 +1627,7 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
case "compaction_start":
|
case "compaction_start":
|
||||||
this.emit({
|
this.emitCompactionTimeline({
|
||||||
type: "timeline",
|
|
||||||
provider: PI_PROVIDER,
|
|
||||||
turnId,
|
turnId,
|
||||||
item: {
|
item: {
|
||||||
type: "compaction",
|
type: "compaction",
|
||||||
@@ -1445,13 +1637,12 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
});
|
});
|
||||||
return;
|
return;
|
||||||
case "compaction_end":
|
case "compaction_end":
|
||||||
this.emit({
|
this.emitCompactionTimeline({
|
||||||
type: "timeline",
|
|
||||||
provider: PI_PROVIDER,
|
|
||||||
turnId,
|
turnId,
|
||||||
item: {
|
item: {
|
||||||
type: "compaction",
|
type: "compaction",
|
||||||
status: "completed",
|
status: "completed",
|
||||||
|
trigger: event.reason === "manual" ? "manual" : "auto",
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
return;
|
return;
|
||||||
@@ -1463,6 +1654,37 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private emitCompactionTimeline(input: {
|
||||||
|
turnId: string | undefined;
|
||||||
|
item: Extract<AgentStreamEvent, { type: "timeline" }>["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(
|
private handleMessageUpdate(
|
||||||
event: Extract<PiAgentSessionEvent, { type: "message_update" }>,
|
event: Extract<PiAgentSessionEvent, { type: "message_update" }>,
|
||||||
turnId: string | undefined,
|
turnId: string | undefined,
|
||||||
|
|||||||
@@ -129,6 +129,17 @@ class PiCliRuntimeSession implements PiRuntimeSession {
|
|||||||
await this.request({ type: "prompt", message, ...(images?.length ? { images } : {}) });
|
await this.request({ type: "prompt", message, ...(images?.length ? { images } : {}) });
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async compact(customInstructions?: string): Promise<void> {
|
||||||
|
await this.request({
|
||||||
|
type: "compact",
|
||||||
|
...(customInstructions ? { customInstructions } : {}),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
async setAutoCompaction(enabled: boolean): Promise<void> {
|
||||||
|
await this.request({ type: "set_auto_compaction", enabled });
|
||||||
|
}
|
||||||
|
|
||||||
async abort(): Promise<void> {
|
async abort(): Promise<void> {
|
||||||
await this.request({ type: "abort" });
|
await this.request({ type: "abort" });
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -79,6 +79,7 @@ export interface PiSessionState {
|
|||||||
thinkingLevel: PiThinkingLevel;
|
thinkingLevel: PiThinkingLevel;
|
||||||
isStreaming: boolean;
|
isStreaming: boolean;
|
||||||
isCompacting: boolean;
|
isCompacting: boolean;
|
||||||
|
autoCompactionEnabled?: boolean;
|
||||||
sessionFile?: string;
|
sessionFile?: string;
|
||||||
sessionId: string;
|
sessionId: string;
|
||||||
sessionName?: string;
|
sessionName?: string;
|
||||||
@@ -111,6 +112,8 @@ export interface PiRpcSlashCommand {
|
|||||||
|
|
||||||
export type PiRpcCommand =
|
export type PiRpcCommand =
|
||||||
| { id?: string; type: "prompt"; message: string; images?: PiImageContent[] }
|
| { 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: "abort" }
|
||||||
| { id?: string; type: "get_state" }
|
| { id?: string; type: "get_state" }
|
||||||
| { id?: string; type: "get_messages" }
|
| { id?: string; type: "get_messages" }
|
||||||
|
|||||||
@@ -37,6 +37,8 @@ export interface PiRuntimeSession {
|
|||||||
message: string,
|
message: string,
|
||||||
images?: Array<{ type: "image"; data: string; mimeType: string }>,
|
images?: Array<{ type: "image"; data: string; mimeType: string }>,
|
||||||
): Promise<void>;
|
): Promise<void>;
|
||||||
|
compact(customInstructions?: string): Promise<void>;
|
||||||
|
setAutoCompaction(enabled: boolean): Promise<void>;
|
||||||
abort(): Promise<void>;
|
abort(): Promise<void>;
|
||||||
getState(): Promise<PiSessionState>;
|
getState(): Promise<PiSessionState>;
|
||||||
getMessages(): Promise<PiAgentMessage[]>;
|
getMessages(): Promise<PiAgentMessage[]>;
|
||||||
|
|||||||
@@ -51,6 +51,8 @@ export class FakePi implements PiRuntime {
|
|||||||
|
|
||||||
export class FakePiSession implements PiRuntimeSession {
|
export class FakePiSession implements PiRuntimeSession {
|
||||||
readonly prompts: Array<{ message: string; imageCount: number }> = [];
|
readonly prompts: Array<{ message: string; imageCount: number }> = [];
|
||||||
|
readonly compactRequests: Array<{ customInstructions?: string }> = [];
|
||||||
|
readonly setAutoCompactionRequests: boolean[] = [];
|
||||||
readonly setModelRequests: Array<{ provider: string; modelId: string }> = [];
|
readonly setModelRequests: Array<{ provider: string; modelId: string }> = [];
|
||||||
readonly setThinkingLevelRequests: string[] = [];
|
readonly setThinkingLevelRequests: string[] = [];
|
||||||
readonly treeNavigationRequests: string[] = [];
|
readonly treeNavigationRequests: string[] = [];
|
||||||
@@ -69,6 +71,8 @@ export class FakePiSession implements PiRuntimeSession {
|
|||||||
cost: 0,
|
cost: 0,
|
||||||
};
|
};
|
||||||
commands: PiRpcSlashCommand[] = [];
|
commands: PiRpcSlashCommand[] = [];
|
||||||
|
compactError: Error | null = null;
|
||||||
|
emitCompactEnd = true;
|
||||||
state: PiSessionState;
|
state: PiSessionState;
|
||||||
|
|
||||||
private readonly subscribers = new Set<(event: PiRuntimeEvent) => void>();
|
private readonly subscribers = new Set<(event: PiRuntimeEvent) => void>();
|
||||||
@@ -79,6 +83,7 @@ export class FakePiSession implements PiRuntimeSession {
|
|||||||
thinkingLevel: "medium",
|
thinkingLevel: "medium",
|
||||||
isStreaming: false,
|
isStreaming: false,
|
||||||
isCompacting: false,
|
isCompacting: false,
|
||||||
|
autoCompactionEnabled: true,
|
||||||
sessionFile: launch.session ?? "/tmp/pi-session",
|
sessionFile: launch.session ?? "/tmp/pi-session",
|
||||||
sessionId: "pi-session-1",
|
sessionId: "pi-session-1",
|
||||||
messageCount: 0,
|
messageCount: 0,
|
||||||
@@ -102,6 +107,25 @@ export class FakePiSession implements PiRuntimeSession {
|
|||||||
this.handleEntryCaptureCommand(message);
|
this.handleEntryCaptureCommand(message);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async compact(customInstructions?: string): Promise<void> {
|
||||||
|
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<void> {
|
||||||
|
this.setAutoCompactionRequests.push(enabled);
|
||||||
|
this.state = {
|
||||||
|
...this.state,
|
||||||
|
autoCompactionEnabled: enabled,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
async abort(): Promise<void> {
|
async abort(): Promise<void> {
|
||||||
this.abortRequested = true;
|
this.abortRequested = true;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -233,15 +233,15 @@ describe("archiveIfSafe", () => {
|
|||||||
expect(harness.deps.archivePaseoWorktree).not.toHaveBeenCalled();
|
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({
|
const harness = createHarness({
|
||||||
getSnapshot: async () => createSnapshot({ git: { aheadOfOrigin: null } }),
|
getSnapshot: async () =>
|
||||||
|
createSnapshot({ git: { aheadOfOrigin: null, behindOfOrigin: null } }),
|
||||||
});
|
});
|
||||||
|
|
||||||
await runArchiveIfSafe(harness);
|
await runArchiveIfSafe(harness);
|
||||||
|
|
||||||
expect(harness.deps.isPaseoOwnedWorktreeCwd).not.toHaveBeenCalled();
|
expect(harness.deps.archivePaseoWorktree).toHaveBeenCalledTimes(1);
|
||||||
expect(harness.deps.archivePaseoWorktree).not.toHaveBeenCalled();
|
|
||||||
});
|
});
|
||||||
|
|
||||||
test("does nothing when the cwd is not a Paseo-owned worktree", async () => {
|
test("does nothing when the cwd is not a Paseo-owned worktree", async () => {
|
||||||
|
|||||||
@@ -78,10 +78,10 @@ export async function archiveIfSafe(input: {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (snapshot.git.isDirty === true || snapshot.git.aheadOfOrigin === null) {
|
if (snapshot.git.isDirty === true) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (snapshot.git.aheadOfOrigin > 0) {
|
if (typeof snapshot.git.aheadOfOrigin === "number" && snapshot.git.aheadOfOrigin > 0) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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 { tmpdir } from "node:os";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
@@ -24,6 +24,7 @@ process.env.PASEO_SUPERVISED = "0";
|
|||||||
|
|
||||||
const PI_TEST_TIMEOUT_MS = 240_000;
|
const PI_TEST_TIMEOUT_MS = 240_000;
|
||||||
const PI_REAL_TEST_MODEL = getRealProviderConfig("pi").model;
|
const PI_REAL_TEST_MODEL = getRealProviderConfig("pi").model;
|
||||||
|
const PI_FREE_COMPACTION_TEST_MODEL = "openrouter/openai/gpt-oss-20b:free";
|
||||||
|
|
||||||
type ToolCallItem = Extract<AgentTimelineItem, { type: "tool_call" }>;
|
type ToolCallItem = Extract<AgentTimelineItem, { type: "tool_call" }>;
|
||||||
|
|
||||||
@@ -31,6 +32,18 @@ function tmpCwd(prefix = "daemon-real-pi-"): string {
|
|||||||
return mkdtempSync(path.join(tmpdir(), prefix));
|
return mkdtempSync(path.join(tmpdir(), prefix));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function sleep(ms: number): Promise<void> {
|
||||||
|
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 {
|
function createPiClient(): AgentClient {
|
||||||
return createRealProviderClient("pi", pino({ level: "silent" }));
|
return createRealProviderClient("pi", pino({ level: "silent" }));
|
||||||
}
|
}
|
||||||
@@ -68,6 +81,27 @@ async function fetchCanonicalTimeline(
|
|||||||
return timeline.entries.map((entry) => entry.item);
|
return timeline.entries.map((entry) => entry.item);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function waitForTimelineItem(
|
||||||
|
client: DaemonClient,
|
||||||
|
agentId: string,
|
||||||
|
predicate: (item: AgentTimelineItem) => boolean,
|
||||||
|
timeoutMs = PI_TEST_TIMEOUT_MS,
|
||||||
|
): Promise<AgentTimelineItem> {
|
||||||
|
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(
|
async function withConnectedPiDaemon(
|
||||||
run: (context: { client: DaemonClient }) => Promise<void>,
|
run: (context: { client: DaemonClient }) => Promise<void>,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
@@ -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(
|
test(
|
||||||
"bash tool call records completed shell detail and output",
|
"bash tool call records completed shell detail and output",
|
||||||
async () => {
|
async () => {
|
||||||
|
|||||||
@@ -156,6 +156,7 @@ export async function initializeLocalSpeechServices(params: {
|
|||||||
|
|
||||||
const workerClient = localConfig
|
const workerClient = localConfig
|
||||||
? new LocalSpeechWorkerClient({
|
? new LocalSpeechWorkerClient({
|
||||||
|
logger,
|
||||||
config: {
|
config: {
|
||||||
modelsDir: localConfig.modelsDir,
|
modelsDir: localConfig.modelsDir,
|
||||||
voiceSttModel: localModels.voiceLocalSttModel,
|
voiceSttModel: localModels.voiceLocalSttModel,
|
||||||
|
|||||||
@@ -20,6 +20,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js";
|
|||||||
class FakeLocalSpeechWorker extends EventEmitter {
|
class FakeLocalSpeechWorker extends EventEmitter {
|
||||||
public connected = true;
|
public connected = true;
|
||||||
public killed = false;
|
public killed = false;
|
||||||
|
public pid = 12345;
|
||||||
|
public readonly stderr = new EventEmitter() as NodeJS.ReadableStream;
|
||||||
public readonly sent: LocalSpeechWorkerRequest[] = [];
|
public readonly sent: LocalSpeechWorkerRequest[] = [];
|
||||||
public disconnects = 0;
|
public disconnects = 0;
|
||||||
public kills = 0;
|
public kills = 0;
|
||||||
@@ -88,9 +90,9 @@ class PausedIpcWorker {
|
|||||||
}
|
}
|
||||||
|
|
||||||
on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this;
|
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(
|
on(
|
||||||
event: "message" | "exit",
|
event: "message" | "close",
|
||||||
listener:
|
listener:
|
||||||
| ((message: LocalSpeechWorkerToParentMessage) => void)
|
| ((message: LocalSpeechWorkerToParentMessage) => void)
|
||||||
| ((code: number | null, signal: NodeJS.Signals | null) => void),
|
| ((code: number | null, signal: NodeJS.Signals | null) => void),
|
||||||
@@ -103,6 +105,7 @@ class PausedIpcWorker {
|
|||||||
function createClient(options?: { idleTtlMs?: number }) {
|
function createClient(options?: { idleTtlMs?: number }) {
|
||||||
const workers: FakeLocalSpeechWorker[] = [];
|
const workers: FakeLocalSpeechWorker[] = [];
|
||||||
const client = new LocalSpeechWorkerClient({
|
const client = new LocalSpeechWorkerClient({
|
||||||
|
logger: pino({ level: "silent" }),
|
||||||
config: {
|
config: {
|
||||||
modelsDir: "/tmp/models",
|
modelsDir: "/tmp/models",
|
||||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||||
@@ -120,6 +123,19 @@ function createClient(options?: { idleTtlMs?: number }) {
|
|||||||
return { client, workers };
|
return { client, workers };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function createCapturingLogger(): { logger: pino.Logger; records: Array<Record<string, unknown>> } {
|
||||||
|
const records: Array<Record<string, unknown>> = [];
|
||||||
|
const logger = pino(
|
||||||
|
{ level: "trace" },
|
||||||
|
{
|
||||||
|
write(line: string) {
|
||||||
|
records.push(JSON.parse(line) as Record<string, unknown>);
|
||||||
|
},
|
||||||
|
},
|
||||||
|
);
|
||||||
|
return { logger, records };
|
||||||
|
}
|
||||||
|
|
||||||
async function waitForMicrotasks(): Promise<void> {
|
async function waitForMicrotasks(): Promise<void> {
|
||||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
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 () => {
|
it("does not surface real IPC backpressure when replaying native-sized dictation frames", async () => {
|
||||||
const workers: PausedIpcWorker[] = [];
|
const workers: PausedIpcWorker[] = [];
|
||||||
const client = new LocalSpeechWorkerClient({
|
const client = new LocalSpeechWorkerClient({
|
||||||
|
logger: pino({ level: "silent" }),
|
||||||
config: {
|
config: {
|
||||||
modelsDir: "/tmp/models",
|
modelsDir: "/tmp/models",
|
||||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
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 () => {
|
it("forwards VAD session events through the shared worker", async () => {
|
||||||
const { client, workers } = createClient();
|
const { client, workers } = createClient();
|
||||||
const provider = new WorkerBackedTurnDetectionProvider(client);
|
const provider = new WorkerBackedTurnDetectionProvider(client);
|
||||||
|
|||||||
@@ -28,6 +28,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js";
|
|||||||
const DEFAULT_REQUEST_TIMEOUT_MS = 30000;
|
const DEFAULT_REQUEST_TIMEOUT_MS = 30000;
|
||||||
const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000;
|
const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000;
|
||||||
const DEFAULT_LOCAL_SAMPLE_RATE = 16000;
|
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
|
type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Request
|
||||||
? Request extends LocalSpeechWorkerRequest
|
? Request extends LocalSpeechWorkerRequest
|
||||||
@@ -38,21 +40,27 @@ type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Requ
|
|||||||
interface LocalSpeechWorkerProcess {
|
interface LocalSpeechWorkerProcess {
|
||||||
connected: boolean;
|
connected: boolean;
|
||||||
killed: boolean;
|
killed: boolean;
|
||||||
|
pid?: number;
|
||||||
|
stderr?: Readable | null;
|
||||||
send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean;
|
send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean;
|
||||||
disconnect(): void;
|
disconnect(): void;
|
||||||
kill(): boolean;
|
kill(): boolean;
|
||||||
on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this;
|
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 {
|
interface PendingRequest {
|
||||||
resolve: (value: unknown) => void;
|
resolve: (value: unknown) => void;
|
||||||
reject: (error: Error) => void;
|
reject: (error: Error) => void;
|
||||||
timeout: ReturnType<typeof setTimeout>;
|
timeout: ReturnType<typeof setTimeout>;
|
||||||
|
type: LocalSpeechWorkerRequest["type"];
|
||||||
|
summary: Record<string, unknown>;
|
||||||
|
startedAt: number;
|
||||||
}
|
}
|
||||||
|
|
||||||
interface LocalSpeechWorkerClientOptions {
|
interface LocalSpeechWorkerClientOptions {
|
||||||
config: LocalSpeechWorkerConfig;
|
config: LocalSpeechWorkerConfig;
|
||||||
|
logger: pino.Logger;
|
||||||
requestTimeoutMs?: number;
|
requestTimeoutMs?: number;
|
||||||
idleTtlMs?: number;
|
idleTtlMs?: number;
|
||||||
forkWorker?: () => LocalSpeechWorkerProcess;
|
forkWorker?: () => LocalSpeechWorkerProcess;
|
||||||
@@ -90,7 +98,7 @@ function forkLocalSpeechWorker(): LocalSpeechWorkerProcess {
|
|||||||
env,
|
env,
|
||||||
execArgv: resolveWorkerExecArgv(),
|
execArgv: resolveWorkerExecArgv(),
|
||||||
serialization: "advanced",
|
serialization: "advanced",
|
||||||
stdio: ["ignore", "ignore", "inherit", "ipc"],
|
stdio: ["ignore", "ignore", "pipe", "ipc"],
|
||||||
}) as LocalSpeechWorkerProcess;
|
}) as LocalSpeechWorkerProcess;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -100,8 +108,75 @@ function isResponse(
|
|||||||
return message.type === "response";
|
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<string, unknown> {
|
||||||
|
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, unknown>): 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<Record<string, unknown>>;
|
||||||
|
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 {
|
export class LocalSpeechWorkerClient {
|
||||||
private readonly config: LocalSpeechWorkerConfig;
|
private readonly config: LocalSpeechWorkerConfig;
|
||||||
|
private readonly logger: pino.Logger;
|
||||||
private readonly requestTimeoutMs: number;
|
private readonly requestTimeoutMs: number;
|
||||||
private readonly idleTtlMs: number;
|
private readonly idleTtlMs: number;
|
||||||
private readonly forkWorker: () => LocalSpeechWorkerProcess;
|
private readonly forkWorker: () => LocalSpeechWorkerProcess;
|
||||||
@@ -109,11 +184,16 @@ export class LocalSpeechWorkerClient {
|
|||||||
private readonly activeSessionIds = new Set<string>();
|
private readonly activeSessionIds = new Set<string>();
|
||||||
private readonly sessionEmitters = new Map<string, EventEmitter>();
|
private readonly sessionEmitters = new Map<string, EventEmitter>();
|
||||||
private worker: LocalSpeechWorkerProcess | null = null;
|
private worker: LocalSpeechWorkerProcess | null = null;
|
||||||
|
private workerPid: number | null = null;
|
||||||
|
private stderrTail = "";
|
||||||
|
private stderrLineBuffer = "";
|
||||||
private inFlightRequests = 0;
|
private inFlightRequests = 0;
|
||||||
private idleTimer: ReturnType<typeof setTimeout> | null = null;
|
private idleTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
|
private readonly intentionalWorkerCloses = new WeakSet<LocalSpeechWorkerProcess>();
|
||||||
|
|
||||||
constructor(options: LocalSpeechWorkerClientOptions) {
|
constructor(options: LocalSpeechWorkerClientOptions) {
|
||||||
this.config = options.config;
|
this.config = options.config;
|
||||||
|
this.logger = options.logger.child({ component: "local-speech-worker-client" });
|
||||||
this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
|
this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
|
||||||
this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
|
this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
|
||||||
this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker;
|
this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker;
|
||||||
@@ -214,7 +294,9 @@ export class LocalSpeechWorkerClient {
|
|||||||
this.sessionEmitters.clear();
|
this.sessionEmitters.clear();
|
||||||
const worker = this.worker;
|
const worker = this.worker;
|
||||||
this.worker = null;
|
this.worker = null;
|
||||||
|
this.workerPid = null;
|
||||||
if (worker && !worker.killed) {
|
if (worker && !worker.killed) {
|
||||||
|
this.intentionalWorkerCloses.add(worker);
|
||||||
try {
|
try {
|
||||||
worker.disconnect();
|
worker.disconnect();
|
||||||
} catch {
|
} catch {
|
||||||
@@ -232,6 +314,7 @@ export class LocalSpeechWorkerClient {
|
|||||||
const worker = this.ensureWorker();
|
const worker = this.ensureWorker();
|
||||||
const requestId = randomUUID();
|
const requestId = randomUUID();
|
||||||
const message = { ...input, requestId } as LocalSpeechWorkerRequest;
|
const message = { ...input, requestId } as LocalSpeechWorkerRequest;
|
||||||
|
const requestSummary = summarizeWorkerRequest(message);
|
||||||
this.inFlightRequests++;
|
this.inFlightRequests++;
|
||||||
this.clearIdleTimer();
|
this.clearIdleTimer();
|
||||||
|
|
||||||
@@ -246,6 +329,9 @@ export class LocalSpeechWorkerClient {
|
|||||||
resolve: (value) => resolve(value as T),
|
resolve: (value) => resolve(value as T),
|
||||||
reject,
|
reject,
|
||||||
timeout,
|
timeout,
|
||||||
|
type: message.type,
|
||||||
|
summary: requestSummary,
|
||||||
|
startedAt: Date.now(),
|
||||||
});
|
});
|
||||||
|
|
||||||
worker.send(message, (error) => {
|
worker.send(message, (error) => {
|
||||||
@@ -271,11 +357,44 @@ export class LocalSpeechWorkerClient {
|
|||||||
}
|
}
|
||||||
const worker = this.forkWorker();
|
const worker = this.forkWorker();
|
||||||
this.worker = worker;
|
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("message", (message) => this.handleWorkerMessage(message));
|
||||||
worker.on("exit", () => this.handleWorkerExit());
|
worker.on("close", (code, signal) => this.handleWorkerExit(worker, code, signal));
|
||||||
return worker;
|
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 {
|
private handleWorkerMessage(message: LocalSpeechWorkerToParentMessage): void {
|
||||||
if (isResponse(message)) {
|
if (isResponse(message)) {
|
||||||
const pending = this.pendingRequests.get(message.requestId);
|
const pending = this.pendingRequests.get(message.requestId);
|
||||||
@@ -317,13 +436,84 @@ export class LocalSpeechWorkerClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private handleWorkerExit(): void {
|
private handleWorkerExit(
|
||||||
this.worker = null;
|
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.clearIdleTimer();
|
||||||
this.rejectAllPending(new Error("Local speech worker exited"));
|
this.rejectAllPending(error);
|
||||||
for (const [sessionId, emitter] of this.sessionEmitters) {
|
for (const [sessionId, emitter] of this.sessionEmitters) {
|
||||||
if (this.activeSessionIds.has(sessionId)) {
|
if (this.activeSessionIds.has(sessionId)) {
|
||||||
emitter.emit("error", new Error("Local speech worker exited"));
|
this.emitErrorIfObserved(sessionId, emitter, error, workerPid);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
this.activeSessionIds.clear();
|
this.activeSessionIds.clear();
|
||||||
@@ -331,6 +521,24 @@ export class LocalSpeechWorkerClient {
|
|||||||
this.inFlightRequests = 0;
|
this.inFlightRequests = 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private describePendingRequests(): Array<Record<string, unknown>> {
|
||||||
|
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 {
|
private rejectAllPending(error: Error): void {
|
||||||
for (const [requestId, pending] of this.pendingRequests) {
|
for (const [requestId, pending] of this.pendingRequests) {
|
||||||
clearTimeout(pending.timeout);
|
clearTimeout(pending.timeout);
|
||||||
@@ -344,7 +552,27 @@ export class LocalSpeechWorkerClient {
|
|||||||
if (!emitter) {
|
if (!emitter) {
|
||||||
return;
|
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 {
|
private scheduleIdleShutdownIfReady(): void {
|
||||||
|
|||||||
Reference in New Issue
Block a user