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 <boudra.moha@gmail.com>

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<void> {
+  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<AgentStreamEvent, { type: "turn_completed" }> =>
+        event.type === "turn_completed",
+    );
+  }
+
   nextTurnCompletion(): Promise<Extract<AgentStreamEvent, { type: "turn_completed" }>> {
     return this.nextEvent(
       (event): event is Extract<AgentStreamEvent, { type: "turn_completed" }> =>
@@ -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", () => {

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<void> {
+    await new Promise<void>((resolve) => {
+      setImmediate(resolve);
+    });
+    if (this.activeTurnId !== turnId || this.pendingLocalPrompt?.turnId !== turnId) {
+      return;
+    }
+    this.emitPendingLocalPrompt();
+    this.completeTurn(turnId, []);
+  }
+
+  private async completePromptIfHandledWithoutTurn(turnId: string): Promise<void> {
+    await new Promise<void>((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) {

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<void> {
-    await this.request({ type: "prompt", message, ...(images?.length ? { images } : {}) });
+  ): Promise<PiPromptAck> {
+    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<string, unknown>;
+      if (typeof agentInvoked === "boolean") {
+        return { agentInvoked };
+      }
+    }
+    return {};
   }
 
   async compact(customInstructions?: string): Promise<void> {

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;

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<void>;
+  ): Promise<PiPromptAck>;
   compact(customInstructions?: string): Promise<void>;
   setAutoCompaction(enabled: boolean): Promise<void>;
   abort(): Promise<void>;

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<void> {
+  ): Promise<PiPromptAck> {
     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<PiSessionState> {
+    if (this.getStateError) {
+      throw this.getStateError;
+    }
     return this.state;
   }