fix(pi): warn only for messages removed by Stop

Pi input extensions can rewrite a queued steer before it is consumed. Preserve the native clear_queue result instead of inferring removal from unmatched original text, so delivered rewrites never receive a removal warning.

packages/server/src/server/agent/providers/jsonl-rpc-process.ts

@@ -193,7 +193,8 @@ export class JsonlRpcProcess {
    * Send a command whose goal an exited process has already met — stopping a turn,
    * clearing a queue. Resolves once the child has exited instead of rejecting,
    * because nothing is queued and nothing is running, which is what the caller
-   * asked for. Use `request` when the caller needs an answer from a live process.
+   * asked for. Returns the native reply when available, or undefined after exit.
+   * Use `request` when the caller requires an answer from a live process.
    *
    * Keyed on an observed exit rather than on `disposed`: `close()` disposes the
    * transport before termination completes, and a child that outlives SIGKILL can
@@ -202,12 +203,12 @@ export class JsonlRpcProcess {
   async requestStopWork(
     command: { type: string; [key: string]: unknown },
     timeoutMs?: number | null,
-  ): Promise<void> {
+  ): Promise<unknown> {
     if (this.exited) {
       return;
     }
     try {
-      await this.request(command, timeoutMs);
+      return await this.request(command, timeoutMs);
     } catch (error) {
       // A dying child breaks its stdin pipe a fraction before it emits `exit`, which
       // fails this request and starts a termination. Wait for that termination to reach

packages/server/src/server/agent/providers/omp/cli-runtime.ts

@@ -307,8 +307,8 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
     return this.process.request(OmpRpcCommandSchema.parse(command), timeoutMs);
   }
 
-  private requestStopWork(command: OmpRpcCommand): Promise<void> {
-    return this.process.requestStopWork(OmpRpcCommandSchema.parse(command));
+  private async requestStopWork(command: OmpRpcCommand): Promise<void> {
+    await this.process.requestStopWork(OmpRpcCommandSchema.parse(command));
   }
 
   private emit(event: OmpRuntimeEvent): void {

packages/server/src/server/agent/providers/pi/agent.test.ts

@@ -2490,6 +2490,40 @@ describe("PiRpcAgentSession", () => {
 });
 
 describe("PiRpcAgentSession steering", () => {
+  test("does not report a consumed steer rewritten by a Pi input extension", async () => {
+    const { pi, session, events } = await createSession();
+    pi.latestSession().steerTextTransform = (text) => text.replace("please ", "");
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("please continue", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    pi.latestSession().finishSubmittedUserMessage({
+      id: "native-transformed-steer",
+      parentId: null,
+      text: "continue",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems().filter((item) => item.type === "notification")).toEqual([]);
+  });
+
+  test("names the rewritten steer that Pi actually removed", async () => {
+    const { pi, session, events } = await createSession();
+    pi.latestSession().steerTextTransform = (text) => text.replace("please ", "");
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("please continue", { expectedTurnId: turnId });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toContainEqual({
+      type: "notification",
+      level: "warning",
+      message: "Not delivered (stopped): continue",
+    });
+  });
+
   test("does not consume a queued steer when the outstanding foreground prompt has the same text", async () => {
     const { pi, session, events } = await createSession();
     const { turnId } = await session.startTurn("continue", {

packages/server/src/server/agent/providers/pi/agent.ts

@@ -83,7 +83,12 @@ import {
   listPiImportableSessions,
   readPiImportSessionConfig,
 } from "./session-descriptor.js";
-import type { PiRuntime, PiRuntimeSession, PiStartSessionInput } from "./runtime.js";
+import type {
+  PiClearedQueue,
+  PiRuntime,
+  PiRuntimeSession,
+  PiStartSessionInput,
+} from "./runtime.js";
 import type {
   PiAgentSessionEvent,
   PiAgentMessage,
@@ -1566,6 +1571,31 @@ export class PiRpcAgentSession implements AgentSession {
     };
   }
 
+  private reportClearedSteers(
+    cleared: PiClearedQueue | undefined,
+    turnId: string | undefined,
+  ): void {
+    if (!cleared) return;
+    // Pi returns what it actually removed, after any input-extension rewrites.
+    for (const text of cleared.steering) {
+      const index = this.pendingSteerSubmissions.findIndex(
+        (submission) => !submission.consumed && submission.text === text,
+      );
+      const submission = index >= 0 ? this.pendingSteerSubmissions.splice(index, 1)[0] : undefined;
+      const label = submission?.label ?? (text || "queued message");
+      this.emit({
+        type: "timeline",
+        provider: this.provider,
+        turnId,
+        item: {
+          type: "notification",
+          level: "warning",
+          message: `Not delivered (stopped): ${label}`,
+        },
+      });
+    }
+  }
+
   async interrupt(): Promise<void> {
     const turnId = this.activeTurnId ?? undefined;
     const interruption: typeof this.interruptingTurn =
@@ -1573,24 +1603,8 @@ export class PiRpcAgentSession implements AgentSession {
     this.interruptingTurn = interruption;
     try {
       try {
-        await this.runtimeSession.clearQueue();
-        // Only an acknowledged clear_queue proves queued messages were dropped.
-        for (const submission of this.pendingSteerSubmissions.splice(0)) {
-          if (submission.consumed) {
-            this.pendingSteerSubmissions.push(submission);
-            continue;
-          }
-          this.emit({
-            type: "timeline",
-            provider: this.provider,
-            turnId,
-            item: {
-              type: "notification",
-              level: "warning",
-              message: `Not delivered (stopped): ${submission.label}`,
-            },
-          });
-        }
+        const cleared = await this.runtimeSession.clearQueue();
+        this.reportClearedSteers(cleared, turnId);
       } catch (error) {
         // COMPAT(piClearQueueFallback): added in v0.5.0, remove after 2027-03-01 once
         // the pi floor supports clear_queue (added in pi 0.84.4).

packages/server/src/server/agent/providers/pi/cli-runtime.test.ts

@@ -419,11 +419,12 @@ describe("PiCliRuntime", () => {
 
   test("sends the clear_queue RPC frame", async () => {
     const child = createPiChild();
-    replyToCommands(child, () => ({}));
+    const removed = { steering: ["rewritten message"], followUp: [] };
+    replyToCommands(child, () => removed);
     const session = await createRuntime(child).startSession({ cwd: "/workspace/project" });
 
     const clearQueueCommand = capturePendingCommand(child, "clear_queue");
-    await session.clearQueue();
+    await expect(session.clearQueue()).resolves.toEqual(removed);
 
     expect(await clearQueueCommand).toMatchObject({ type: "clear_queue" });
   });

packages/server/src/server/agent/providers/pi/cli-runtime.ts

@@ -3,6 +3,7 @@ import { readPiTranscript, type PiTranscript } from "./transcript.js";
 import { createExternalProcessEnv } from "../../../paseo-env.js";
 import type { ChildProcessWithoutNullStreams } from "node:child_process";
 import type { Logger } from "pino";
+import { z } from "zod";
 
 import type { ProviderRuntimeSettings } from "../../provider-launch-config.js";
 import {
@@ -16,6 +17,7 @@ import {
   type PiRuntimeLaunch,
   type PiRuntimeSession,
   type PiStartSessionInput,
+  type PiClearedQueue,
 } from "./runtime.js";
 import type {
   PiModel,
@@ -149,8 +151,10 @@ class PiCliRuntimeSession implements PiRuntimeSession {
     });
   }
 
-  async clearQueue(): Promise<void> {
-    await this.requestStopWork({ type: "clear_queue" });
+  async clearQueue(): Promise<PiClearedQueue | undefined> {
+    const result = await this.options.process.requestStopWork({ type: "clear_queue" });
+    if (result === undefined) return undefined;
+    return z.object({ steering: z.array(z.string()), followUp: z.array(z.string()) }).parse(result);
   }
 
   async compact(customInstructions?: string): Promise<void> {
@@ -268,8 +272,8 @@ class PiCliRuntimeSession implements PiRuntimeSession {
     return this.options.process.request(command, timeoutMs);
   }
 
-  private requestStopWork(command: PiRpcCommand): Promise<void> {
-    return this.options.process.requestStopWork(command);
+  private async requestStopWork(command: PiRpcCommand): Promise<void> {
+    await this.options.process.requestStopWork(command);
   }
 
   private async waitForCompletion(command: PiRpcCommand): Promise<void> {

packages/server/src/server/agent/providers/pi/runtime.ts

@@ -39,6 +39,11 @@ export interface PiStartSessionInput {
   extraArgs?: string[];
 }
 
+export interface PiClearedQueue {
+  steering: string[];
+  followUp: string[];
+}
+
 export interface PiRuntimeSession {
   readonly environment: Record<string, string>;
   onEvent(callback: (event: PiRuntimeEvent) => void): () => void;
@@ -50,7 +55,7 @@ export interface PiRuntimeSession {
     message: string,
     images?: Array<{ type: "image"; data: string; mimeType: string }>,
   ): Promise<void>;
-  clearQueue(): Promise<void>;
+  clearQueue(): Promise<PiClearedQueue | undefined>;
   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

@@ -5,6 +5,7 @@ import type {
   PiRuntimeLaunch,
   PiRuntimeSession,
   PiStartSessionInput,
+  PiClearedQueue,
 } from "../runtime.js";
 import type {
   PiAgentMessage,
@@ -16,6 +17,7 @@ import type {
   PiSessionStats,
 } from "../rpc-types.js";
 import { buildPiLaunch } from "../runtime.js";
+import { getUserMessageText } from "../history-mapper.js";
 
 type FakePiSubagentSubscriptionLevel = "off" | "progress" | "events";
 type FakePiSubagentStatus = "pending" | "running" | "completed" | "failed" | "aborted";
@@ -135,6 +137,7 @@ export class FakePiSession implements PiRuntimeSession {
   readonly prompts: Array<{ message: string; imageCount: number }> = [];
   readonly steerCalls: Array<{ message: string; imageCount: number }> = [];
   steerError: Error | null = null;
+  steerTextTransform: (text: string) => string = (text) => text;
   readonly controlRequests: string[] = [];
   clearQueueError: Error | null = null;
   clearQueueCompletion: Promise<void> | null = null;
@@ -182,6 +185,8 @@ export class FakePiSession implements PiRuntimeSession {
   state: PiSessionState;
 
   private readonly subscribers = new Set<(event: PiRuntimeEvent) => void>();
+  private readonly steeringMessages: string[] = [];
+  private foregroundMessage: string | null = null;
   private readonly subagentMessageResults = new Map<string, FakePiSubagentMessagesResult[]>();
   private nextHeldPrompt: { promise: Promise<void>; reject: (error: Error) => void } | null = null;
   private activeHeldPrompt: { promise: Promise<void>; reject: (error: Error) => void } | null =
@@ -217,6 +222,7 @@ export class FakePiSession implements PiRuntimeSession {
     images?: Array<{ type: "image"; data: string; mimeType: string }>,
   ): Promise<PiPromptAck> {
     this.prompts.push({ message, imageCount: images?.length ?? 0 });
+    if (!message.startsWith("/")) this.foregroundMessage = message;
     const heldPrompt = this.nextHeldPrompt;
     if (heldPrompt) {
       this.nextHeldPrompt = null;
@@ -262,14 +268,16 @@ export class FakePiSession implements PiRuntimeSession {
     if (this.steerError) {
       throw this.steerError;
     }
+    this.steeringMessages.push(this.steerTextTransform(message));
   }
 
-  async clearQueue(): Promise<void> {
+  async clearQueue(): Promise<PiClearedQueue> {
     this.controlRequests.push("clear_queue");
     await this.clearQueueCompletion;
     if (this.clearQueueError) {
       throw this.clearQueueError;
     }
+    return { steering: this.steeringMessages.splice(0), followUp: [] };
   }
 
   async compact(customInstructions?: string): Promise<void> {
@@ -430,6 +438,16 @@ export class FakePiSession implements PiRuntimeSession {
   }
 
   emit(event: PiRuntimeEvent): void {
+    const message = event.type === "message_end" ? (event.message as PiAgentMessage) : null;
+    if (message?.role === "user") {
+      const text = getUserMessageText(message.content);
+      if (this.foregroundMessage === text) {
+        this.foregroundMessage = null;
+      } else {
+        const index = this.steeringMessages.indexOf(text);
+        if (index >= 0) this.steeringMessages.splice(index, 1);
+      }
+    }
     for (const subscriber of this.subscribers) {
       subscriber(event);
     }