From cd6c608c9e25a0a1c1332c70bb68209d4c612094 Mon Sep 17 00:00:00 2001 From: Ethan Greenfeld <39423472+ebg1223@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:29:42 -0700 Subject: [PATCH] Fix Pi slash commands hanging after local completion (#2066) * fix(pi): complete locally handled prompts Slash commands the Pi CLI handles without starting a model turn previously left the Paseo turn hanging until interrupt. Parse the optional agentInvoked correlation from prompt acknowledgements, surface command_output/notify output as timeline content, and complete the turn locally when no agent turn will start. Older Pi binaries without the correlation are detected via a getState compatibility probe. * review: guard command output without an active turn, simplify delegates - Drop late-arriving command_output when no turn is active instead of emitting an orphaned timeline item with turnId undefined - Remove redundant 'text' in event check (type is already narrowed) - Inline recordNoTurnNotification delegate * refactor(pi): consolidate local prompt state --------- Co-authored-by: Mohamed Boudra --- .../server/agent/providers/pi/agent.test.ts | 142 ++++++++++++++++ .../src/server/agent/providers/pi/agent.ts | 160 ++++++++++++++++-- .../server/agent/providers/pi/cli-runtime.ts | 16 +- .../server/agent/providers/pi/rpc-types.ts | 7 + .../src/server/agent/providers/pi/runtime.ts | 3 +- .../agent/providers/pi/test-utils/fake-pi.ts | 9 +- 6 files changed, 317 insertions(+), 20 deletions(-) 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 9ebbe00fe..467e68bc4 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -12,6 +12,7 @@ import { import { tmpdir } from "node:os"; import path from "node:path"; import pino from "pino"; +import { setImmediate as waitForImmediate } from "node:timers/promises"; import { describe, expect, onTestFinished, test } from "vitest"; import type { AgentSession, AgentSessionConfig, AgentStreamEvent } from "../../agent-sdk-types.js"; @@ -54,6 +55,9 @@ function readUtf8File(pathname: string): string { closeSync(fd); } } +async function flushTurnScheduling(): Promise { + await waitForImmediate(); +} async function createSession(pi = new FakePi()): Promise<{ pi: FakePi; @@ -147,6 +151,13 @@ class SessionEvents { }); } + turnCompletedEvents() { + return this.events.filter( + (event): event is Extract => + event.type === "turn_completed", + ); + } + nextTurnCompletion(): Promise> { return this.nextEvent( (event): event is Extract => @@ -885,6 +896,137 @@ describe("PiRpcAgentSession", () => { error: "Pi exited", }); }); + test("completes locally handled slash commands when agentInvoked is false", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.promptAck = { agentInvoked: false }; + + const { turnId: usageTurnId } = await session.startTurn("/usage"); + fakeSession.emit({ + type: "command_output", + text: "\u001b[38;2;138;138;138mUsage 12%\u001b[39m", + }); + + await flushTurnScheduling(); + const usageCompletion = await events.nextTurnCompletion(); + expect(usageCompletion).toMatchObject({ type: "turn_completed", turnId: usageTurnId }); + expect(events.timelineAndCompletionEvents()).toEqual([ + { type: "timeline", item: { type: "user_message", text: "/usage" } }, + { type: "timeline", item: { type: "assistant_message", text: "Usage 12%" } }, + { type: "turn_completed" }, + ]); + + const { turnId: helloTurnId } = await session.startTurn("hello"); + fakeSession.finishTurn(); + await flushTurnScheduling(); + expect(events.turnCompletedEvents()).toHaveLength(2); + expect(events.turnCompletedEvents()[1]).toMatchObject({ + type: "turn_completed", + turnId: helloTurnId, + }); + }); + + test("does not synthesize completion when agentInvoked is true for slash prompts", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.promptAck = { agentInvoked: true }; + + const { turnId } = await session.startTurn("/usage"); + await flushTurnScheduling(); + expect(events.turnCompletedEvents()).toHaveLength(0); + + fakeSession.emit({ type: "agent_start" }); + fakeSession.finishTurn(); + await flushTurnScheduling(); + + const completion = await events.nextTurnCompletion(); + expect(completion).toMatchObject({ type: "turn_completed", turnId }); + expect(events.turnCompletedEvents()).toHaveLength(1); + }); + + test("probes slash prompts without agentInvoked and surfaces buffered notify output", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + + const { turnId } = await session.startTurn("/plan on"); + fakeSession.emit({ + type: "extension_ui_request", + id: "notify-plan", + method: "notify", + message: "Plan mode enabled", + }); + + await flushTurnScheduling(); + const completion = await events.nextTurnCompletion(); + expect(completion).toMatchObject({ type: "turn_completed", turnId }); + expect(events.timelineAndCompletionEvents()).toEqual([ + { type: "timeline", item: { type: "user_message", text: "/plan on" } }, + { type: "timeline", item: { type: "assistant_message", text: "Plan mode enabled" } }, + { type: "turn_completed" }, + ]); + }); + + test("does not synthesize completion when lifecycle starts before the no-turn probe", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + + const { turnId } = await session.startTurn("/custom-template-cmd"); + fakeSession.emit({ + type: "extension_ui_request", + id: "notify-buffered", + method: "notify", + message: "Should not appear after turn starts", + }); + fakeSession.emit({ type: "agent_start" }); + + await flushTurnScheduling(); + expect(events.turnCompletedEvents()).toHaveLength(0); + expect(events.timelineItems()).toEqual([]); + + fakeSession.finishTurn(); + await flushTurnScheduling(); + const completion = await events.nextTurnCompletion(); + expect(completion).toMatchObject({ type: "turn_completed", turnId }); + expect(events.turnCompletedEvents()).toHaveLength(1); + }); + + test("fails slash turns when the no-turn getState barrier errors", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.getStateError = new Error("get_state timed out"); + + const { turnId } = await session.startTurn("/local-command on"); + await flushTurnScheduling(); + + await expect(events.nextTurnFailure()).resolves.toMatchObject({ + turnId, + error: "get_state timed out", + }); + + fakeSession.getStateError = null; + const { turnId: recoveryTurnId } = await session.startTurn("hello"); + fakeSession.finishTurn(); + await flushTurnScheduling(); + await expect(events.nextTurnCompletion()).resolves.toMatchObject({ + type: "turn_completed", + turnId: recoveryTurnId, + }); + }); + + test("does not probe non-slash prompts when agentInvoked is missing", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + + const { turnId } = await session.startTurn("hello"); + await flushTurnScheduling(); + expect(events.turnCompletedEvents()).toHaveLength(0); + + fakeSession.finishTurn(); + await flushTurnScheduling(); + const completion = await events.nextTurnCompletion(); + expect(completion).toMatchObject({ type: "turn_completed", turnId }); + expect(events.turnCompletedEvents()).toHaveLength(1); + }); }); describe("PiRpcAgentClient", () => { diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index e57126e7f..8cbdfa420 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -3,6 +3,7 @@ import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "no import { homedir, tmpdir } from "node:os"; import { join, resolve as resolvePath } from "node:path"; import type { Logger } from "pino"; +import stripAnsi from "strip-ansi"; import { z } from "zod"; import { @@ -228,6 +229,12 @@ interface PendingPiUserMessage { turnId: string | undefined; } +interface PendingLocalPrompt { + turnId: string; + text: string; + outputs: string[]; +} + interface PendingExtensionResult { resolve: (value: unknown) => void; reject: (error: Error) => void; @@ -1064,6 +1071,7 @@ export class PiRpcAgentSession implements AgentSession { private activeAskUserDialog: ActiveAskUserDialog | null = null; private pendingCombinedAskUserResponse: PendingCombinedAskUserResponse | null = null; private activeTurnId: string | null = null; + private pendingLocalPrompt: PendingLocalPrompt | null = null; private activeAssistantMessageId: string | null = null; private lastKnownThinkingOptionId: string | null; currentLeafOverrideId: string | null | undefined; @@ -1124,29 +1132,43 @@ export class PiRpcAgentSession implements AgentSession { const payload = convertPromptInput(prompt, { model: this.state.model }); const turnId = randomUUID(); this.activeTurnId = turnId; + this.pendingLocalPrompt = { turnId, text: payload.text, outputs: [] }; this.activeAssistantMessageId = null; + const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null; - void this.runtimeSession.prompt(payload.text, payload.images).catch((error) => { - if (this.activeTurnId !== turnId) { - return; - } - this.activeTurnId = null; - if (isPiRequestAbortError(error)) { + void (async () => { + try { + const ack = await this.runtimeSession.prompt(payload.text, payload.images); + if (ack.agentInvoked === false) { + await this.completeNoTurnPrompt(turnId); + return; + } + if (ack.agentInvoked === undefined && shouldProbeForNoTurnPrompt) { + await this.completePromptIfHandledWithoutTurn(turnId); + } + } catch (error) { + if (this.activeTurnId !== turnId) { + return; + } + this.activeTurnId = null; + this.pendingLocalPrompt = null; + if (isPiRequestAbortError(error)) { + this.emit({ + type: "turn_canceled", + provider: PI_PROVIDER, + turnId, + reason: toDiagnosticErrorMessage(error), + }); + return; + } this.emit({ - type: "turn_canceled", + type: "turn_failed", provider: PI_PROVIDER, turnId, - reason: toDiagnosticErrorMessage(error), + error: toDiagnosticErrorMessage(error), }); - return; } - this.emit({ - type: "turn_failed", - provider: PI_PROVIDER, - turnId, - error: toDiagnosticErrorMessage(error), - }); - }); + })(); return { turnId }; } @@ -1242,6 +1264,7 @@ export class PiRpcAgentSession implements AgentSession { await this.runtimeSession.abort(); if (turnId && this.activeTurnId === turnId) { this.activeTurnId = null; + this.pendingLocalPrompt = null; this.emit({ type: "turn_canceled", provider: PI_PROVIDER, reason: "interrupted", turnId }); } } @@ -1371,6 +1394,82 @@ export class PiRpcAgentSession implements AgentSession { return this.activeTurnId ?? undefined; } + private async completeNoTurnPrompt(turnId: string): Promise { + await new Promise((resolve) => { + setImmediate(resolve); + }); + if (this.activeTurnId !== turnId || this.pendingLocalPrompt?.turnId !== turnId) { + return; + } + this.emitPendingLocalPrompt(); + this.completeTurn(turnId, []); + } + + private async completePromptIfHandledWithoutTurn(turnId: string): Promise { + await new Promise((resolve) => { + setImmediate(resolve); + }); + + let runtimeState: PiSessionState; + try { + runtimeState = await this.runtimeSession.getState(); + } catch (error) { + if (this.activeTurnId === turnId && this.pendingLocalPrompt?.turnId === turnId) { + throw error; + } + return; + } + + if ( + this.activeTurnId !== turnId || + this.pendingLocalPrompt?.turnId !== turnId || + runtimeState.isStreaming + ) { + return; + } + this.state = runtimeState; + + this.emitPendingLocalPrompt(); + this.completeTurn(turnId, []); + } + + private emitPendingLocalPrompt(): void { + const prompt = this.pendingLocalPrompt; + this.pendingLocalPrompt = null; + if (!prompt) { + return; + } + if (prompt.text) { + this.emit({ + type: "timeline", + provider: PI_PROVIDER, + turnId: prompt.turnId, + item: { + type: "user_message", + text: prompt.text, + }, + }); + } + for (const output of prompt.outputs) { + this.emit({ + type: "timeline", + provider: PI_PROVIDER, + turnId: prompt.turnId, + item: { + type: "assistant_message", + text: output, + }, + }); + } + } + + private bufferLocalPromptOutput(message: string): void { + if (!this.pendingLocalPrompt) { + return; + } + this.pendingLocalPrompt.outputs.push(message); + } + private parseSlashCommandInput(text: string): PiSlashCommandInvocation | null { const trimmed = text.trim(); if (!trimmed.startsWith("/") || trimmed.length <= 1) { @@ -1613,6 +1712,7 @@ export class PiRpcAgentSession implements AgentSession { if (this.handleEntryCaptureMarker(message) || this.handleCommandResultMarker(message)) { return; } + this.bufferLocalPromptOutput(message); } if (this.respondToCombinedAskUserFollowUp(event)) { @@ -1667,6 +1767,26 @@ export class PiRpcAgentSession implements AgentSession { return false; } + private handleCommandOutput(textValue: unknown): void { + if (!this.activeTurnId) { + return; + } + const text = stripAnsi(optionalString(textValue) ?? "").trim(); + if (!text) { + return; + } + if (this.pendingLocalPrompt) { + this.bufferLocalPromptOutput(text); + return; + } + this.emit({ + type: "timeline", + provider: PI_PROVIDER, + turnId: this.currentTurnIdForEvent(), + item: { type: "assistant_message", text }, + }); + } + private handleRuntimeEvent(event: PiRuntimeEvent): void { if (event.type === "extension_ui_request") { this.handleExtensionUiRequest(event); @@ -1676,6 +1796,10 @@ export class PiRpcAgentSession implements AgentSession { this.handleProcessExit(event.error); return; } + if (event.type === "command_output") { + this.handleCommandOutput(event.text); + return; + } this.handleSessionEvent(event); } @@ -1686,6 +1810,7 @@ export class PiRpcAgentSession implements AgentSession { } const turnId = this.activeTurnId; this.activeTurnId = null; + this.pendingLocalPrompt = null; this.emit({ type: "turn_failed", provider: PI_PROVIDER, @@ -1699,6 +1824,7 @@ export class PiRpcAgentSession implements AgentSession { switch (event.type) { case "agent_start": + this.pendingLocalPrompt = null; this.emit({ type: "thread_started", provider: PI_PROVIDER, @@ -1706,6 +1832,7 @@ export class PiRpcAgentSession implements AgentSession { }); return; case "turn_start": + this.pendingLocalPrompt = null; this.emit({ type: "turn_started", provider: PI_PROVIDER, @@ -1922,6 +2049,7 @@ export class PiRpcAgentSession implements AgentSession { private completeTurn(turnId: string | undefined, messages: PiAgentMessage[]): void { this.activeTurnId = null; + this.pendingLocalPrompt = null; this.activeAssistantMessageId = null; const errorMessage = latestPiErrorMessage(messages); if (typeof errorMessage === "string" && errorMessage.length > 0) { 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 001c6339c..4b7418695 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.ts @@ -15,6 +15,7 @@ import type { PiAgentMessage, PiCommandsRpcType, PiModel, + PiPromptAck, PiRpcCommand, PiRpcResponse, PiRpcSlashCommand, @@ -136,8 +137,19 @@ class PiCliRuntimeSession implements PiRuntimeSession { async prompt( message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): Promise { - await this.request({ type: "prompt", message, ...(images?.length ? { images } : {}) }); + ): Promise { + const data = await this.request({ + type: "prompt", + message, + ...(images?.length ? { images } : {}), + }); + if (typeof data === "object" && data !== null && !Array.isArray(data)) { + const { agentInvoked } = data as Record; + if (typeof agentInvoked === "boolean") { + return { agentInvoked }; + } + } + return {}; } async compact(customInstructions?: string): Promise { 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 a13fa9708..6b0cc6c76 100644 --- a/packages/server/src/server/agent/providers/pi/rpc-types.ts +++ b/packages/server/src/server/agent/providers/pi/rpc-types.ts @@ -5,6 +5,9 @@ export interface PiImageContent { data: string; mimeType: string; } +export interface PiPromptAck { + agentInvoked?: boolean; +} export interface PiTextContent { type: "text"; @@ -181,6 +184,10 @@ export type PiRuntimeEvent = method: string; [key: string]: unknown; } + | { + type: "command_output"; + text?: string; + } | { type: "process_exit"; error: string; diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index e3d4ca72b..a4e4be58a 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/packages/server/src/server/agent/providers/pi/runtime.ts @@ -1,6 +1,7 @@ import type { PiAgentMessage, PiModel, + PiPromptAck, PiRpcSlashCommand, PiRuntimeEvent, PiSessionState, @@ -38,7 +39,7 @@ export interface PiRuntimeSession { prompt( message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): Promise; + ): Promise; compact(customInstructions?: string): Promise; setAutoCompaction(enabled: boolean): Promise; abort(): 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 2fc4072e1..d27b44f66 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 @@ -7,6 +7,7 @@ import type { import type { PiAgentMessage, PiModel, + PiPromptAck, PiRpcSlashCommand, PiRuntimeEvent, PiSessionState, @@ -73,6 +74,8 @@ export class FakePiSession implements PiRuntimeSession { commands: PiRpcSlashCommand[] = []; compactError: Error | null = null; emitCompactEnd = true; + getStateError: Error | null = null; + promptAck: PiPromptAck = {}; state: PiSessionState; private readonly subscribers = new Set<(event: PiRuntimeEvent) => void>(); @@ -104,7 +107,7 @@ export class FakePiSession implements PiRuntimeSession { async prompt( message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): Promise { + ): Promise { this.prompts.push({ message, imageCount: images?.length ?? 0 }); const heldPrompt = this.nextHeldPrompt; if (heldPrompt) { @@ -120,6 +123,7 @@ export class FakePiSession implements PiRuntimeSession { } this.handleTreeNavigationCommand(message); this.handleEntryCaptureCommand(message); + return this.promptAck; } holdNextPrompt(): void { @@ -166,6 +170,9 @@ export class FakePiSession implements PiRuntimeSession { } async getState(): Promise { + if (this.getStateError) { + throw this.getStateError; + } return this.state; }