fix(pi): avoid delivery warnings for consumed image steers

Pi retains empty text entries in its steering queue after consuming image-only input. Require an unconsumed submission before reporting those entries as removed, and model that native behavior in the fake.

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,322 @@ describe("PiRpcAgentSession", () => {
 });
 
 describe("PiRpcAgentSession steering", () => {
+  test.each([false, true])(
+    "does not warn for Pi's stale consumed image queue entry (native echo: %s)",
+    async (echoEntry) => {
+      const { pi, session, events } = await createSession();
+      const native = pi.latestSession();
+      native.setModelResult = {
+        provider: "openai",
+        id: "gpt-4o",
+        input: ["text", "image"],
+      };
+      await session.setModel("openai/gpt-4o");
+      const { turnId } = await session.startTurn("work");
+      await session.steerActiveTurn(
+        [{ type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" }],
+        { expectedTurnId: turnId },
+      );
+      if (echoEntry) {
+        native.finishSubmittedUserMessage({ id: "image-steer", parentId: null, text: "" });
+      } else {
+        native.emit({ type: "message_end", message: { role: "user", content: [] } });
+      }
+
+      await session.interrupt();
+
+      expect(events.timelineItems().filter((item) => item.type === "notification")).toEqual([]);
+    },
+  );
+
+  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", {
+      clientMessageId: "client-prompt-1",
+    });
+    await session.steerActiveTurn("continue", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    pi.latestSession().emit({ type: "agent_start" });
+    pi.latestSession().emit({ type: "turn_start" });
+    pi.latestSession().finishSubmittedUserMessage({
+      id: "native-prompt-1",
+      parentId: null,
+      text: "continue",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "continue",
+        messageId: "native-prompt-1",
+        clientMessageId: "client-prompt-1",
+      },
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): continue",
+      },
+    ]);
+  });
+
+  test("keeps a consumed steer distinct from an identical foreground prompt", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("continue", { clientMessageId: "client-prompt-1" });
+    await session.steerActiveTurn("continue", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    const native = pi.latestSession();
+    native.finishSubmittedUserMessage({ id: "native-prompt-1", parentId: null, text: "continue" });
+    native.finishSubmittedUserMessage({
+      id: "native-steer-1",
+      parentId: "native-prompt-1",
+      text: "continue",
+    });
+    await session.interrupt();
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "continue",
+        messageId: "native-prompt-1",
+        clientMessageId: "client-prompt-1",
+      },
+      {
+        type: "user_message",
+        text: "continue",
+        messageId: "native-steer-1",
+        clientMessageId: "client-steer-1",
+      },
+    ]);
+  });
+
+  test.each([1, 2])(
+    "labels a stopped image-only steer containing %i image(s)",
+    async (imageCount) => {
+      const { pi, session, events } = await createSession();
+      const native = pi.latestSession();
+      native.setModelResult = {
+        provider: "openai",
+        id: "gpt-4o",
+        name: "GPT-4o",
+        input: ["text", "image"],
+      };
+      await session.setModel("openai/gpt-4o");
+      const { turnId } = await session.startTurn([
+        { type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" },
+      ]);
+      await session.steerActiveTurn(
+        Array.from({ length: imageCount }, () => ({
+          type: "image" as const,
+          data: ONE_BY_ONE_PNG_BASE64,
+          mimeType: "image/png",
+        })),
+        { expectedTurnId: turnId, clientMessageId: "client-steer-1" },
+      );
+      native.emit({ type: "message_end", message: { role: "user", content: [] } });
+      await session.interrupt();
+      expect(native.steerCalls).toEqual([{ message: "", imageCount }]);
+      expect(events.timelineItems()).toEqual([
+        {
+          type: "notification",
+          level: "warning",
+          message: `Not delivered (stopped): ${imageCount} ${imageCount === 1 ? "image" : "images"}`,
+        },
+      ]);
+    },
+  );
+
+  test("reports an unconsumed steer as not delivered when Stop clears the queue", async () => {
+    const { session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("Reply STEER_CHECK_7391", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toContainEqual({
+      type: "notification",
+      level: "warning",
+      message: "Not delivered (stopped): Reply STEER_CHECK_7391",
+    });
+  });
+
+  test("does not report a consumed steer before its native entry ID arrives", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("already consumed", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    pi.latestSession().emit({
+      type: "message_end",
+      message: { role: "user", content: "already consumed" },
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([]);
+  });
+
+  test("tracks native consumption before the steer RPC is acknowledged", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    const native = pi.latestSession();
+    native.steer = async (message) => {
+      native.finishSubmittedUserMessage({ id: "native-entry-1", parentId: null, text: message });
+    };
+
+    await expect(
+      session.steerActiveTurn("already consumed", {
+        expectedTurnId: turnId,
+        clientMessageId: "client-steer-1",
+      }),
+    ).resolves.toEqual({ status: "accepted" });
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "already consumed",
+        messageId: "native-entry-1",
+        clientMessageId: "client-steer-1",
+      },
+    ]);
+  });
+
+  test("keeps identical steers distinct from native message IDs when one is consumed", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("same text", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    await session.steerActiveTurn("same text", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-2",
+    });
+    pi.latestSession().finishSubmittedUserMessage({
+      id: "native-entry-1",
+      parentId: null,
+      text: "same text",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "same text",
+        messageId: "native-entry-1",
+        clientMessageId: "client-steer-1",
+      },
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): same text",
+      },
+    ]);
+  });
+
+  test("retains steer tracking when clear_queue fails so Stop can be retried", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("queued", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    const native = pi.latestSession();
+    native.clearQueueError = new Error("Queue clear failed");
+
+    await expect(session.interrupt()).rejects.toThrow("Queue clear failed");
+    expect(native.controlRequests).toEqual(["clear_queue"]);
+    expect(events.timelineItems()).toEqual([]);
+
+    native.clearQueueError = null;
+    await session.interrupt();
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): queued",
+      },
+    ]);
+  });
+
+  test("reports an acknowledged queue removal even when abort subsequently fails", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("queued", { expectedTurnId: turnId });
+    const native = pi.latestSession();
+    native.abort = async () => {
+      throw new Error("Abort failed");
+    };
+
+    await expect(session.interrupt()).rejects.toThrow("Abort failed");
+    native.abort = async () => {};
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): queued",
+      },
+    ]);
+  });
+
+  test("does not claim queue removal when an older Pi lacks clear_queue", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("queued", { expectedTurnId: turnId });
+    pi.latestSession().clearQueueError = new Error("Unknown command: clear_queue");
+
+    await session.interrupt();
+
+    expect(pi.latestSession().controlRequests).toEqual(["clear_queue", "abort"]);
+    expect(events.timelineItems()).toEqual([]);
+  });
+
   test("steers the active Pi turn and correlates the echoed user message", async () => {
     const { pi, session, events } = await createSession();
     const fakeSession = pi.latestSession();

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,
@@ -232,7 +237,14 @@ interface StartTurnResult {
 
 interface PiPendingSteerSubmission {
   text: string;
+  label: string;
   clientMessageId: string | null;
+  consumed: boolean;
+}
+
+interface PiPendingForegroundSubmission {
+  text: string;
+  consumed: boolean;
 }
 
 // COMPAT(piSteerFallback): added in v0.5.0, remove after 2027-03-01 once the pi floor
@@ -1210,6 +1222,7 @@ export class PiRpcAgentSession implements AgentSession {
   private activePromptRequestId: string | null = null;
   private readonly pendingPromptResults = new Map<string, boolean>();
   private readonly pendingSteerSubmissions: PiPendingSteerSubmission[] = [];
+  private pendingForegroundSubmission: PiPendingForegroundSubmission | null = null;
   currentLeafOverrideId: string | null | undefined;
   private readonly treeUserEntriesById = new Map<string, PiCapturedEntry>();
   private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
@@ -1319,10 +1332,12 @@ export class PiRpcAgentSession implements AgentSession {
     this.pendingSettledMessages = null;
     this.activePromptRequestId = null;
     this.pendingSteerSubmissions.length = 0;
+    this.pendingForegroundSubmission = null;
     this.clearNoTurnBuffers();
     const payload = await convertPromptInput(prompt, { model: this.state.model });
     if (this.closed) throw new Error("Pi session is closed");
     if (this.activeTurnId !== turnId) return { turnId };
+    this.pendingForegroundSubmission = { text: payload.text, consumed: false };
     this.activeNoTurnPromptText = payload.text;
     const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null;
 
@@ -1365,6 +1380,7 @@ export class PiRpcAgentSession implements AgentSession {
         this.pendingSettledMessages = null;
         this.activeAssistantMessageId = null;
         this.pendingSteerSubmissions.length = 0;
+        this.pendingForegroundSubmission = null;
         this.clearNoTurnBuffers();
         if (isPiRequestAbortError(error)) {
           this.emit({
@@ -1403,9 +1419,29 @@ export class PiRpcAgentSession implements AgentSession {
       return { status: "unavailable" };
     }
 
+    const imageCount = payload.images?.length ?? 0;
+    const isImageOnly = payload.text.trim().length === 0 && imageCount > 0;
+    let label = payload.text;
+    if (isImageOnly) {
+      if (imageCount === 1) {
+        label = "1 image";
+      } else {
+        label = `${imageCount} images`;
+      }
+    }
+    const submission: PiPendingSteerSubmission = {
+      text: payload.text,
+      label,
+      clientMessageId: options.clientMessageId ?? null,
+      consumed: false,
+    };
+    // Pi can consume the message before acknowledging the steer RPC.
+    this.pendingSteerSubmissions.push(submission);
     try {
       await this.runtimeSession.steer(payload.text, payload.images);
     } catch (error) {
+      const index = this.pendingSteerSubmissions.indexOf(submission);
+      if (index >= 0) this.pendingSteerSubmissions.splice(index, 1);
       if (isPiDefinitiveSteerRejection(error)) {
         return { status: "unavailable" };
       }
@@ -1416,10 +1452,6 @@ export class PiRpcAgentSession implements AgentSession {
     if (this.closed || this.activeTurnId !== options.expectedTurnId) {
       return { status: "unavailable" };
     }
-    this.pendingSteerSubmissions.push({
-      text: payload.text,
-      clientMessageId: options.clientMessageId ?? null,
-    });
     if (options.clearPendingPermissions) {
       await this.clearPendingPermissionsForSteer();
     }
@@ -1438,6 +1470,11 @@ export class PiRpcAgentSession implements AgentSession {
   }
 
   private takePendingSteerSubmission(text: string): PiPendingSteerSubmission | undefined {
+    // The original prompt is also echoed, possibly after a matching steer is queued.
+    if (this.pendingForegroundSubmission?.text === text) {
+      this.pendingForegroundSubmission = null;
+      return undefined;
+    }
     const index = this.pendingSteerSubmissions.findIndex((submission) => submission.text === text);
     if (index < 0) {
       return undefined;
@@ -1534,6 +1571,34 @@ 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;
+      // Pi retains consumed image-only entries in its text queue. An empty native
+      // entry proves removal only when we still have an unconsumed submission.
+      if (!text && !submission) continue;
+      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 =
@@ -1541,7 +1606,8 @@ export class PiRpcAgentSession implements AgentSession {
     this.interruptingTurn = interruption;
     try {
       try {
-        await this.runtimeSession.clearQueue();
+        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).
@@ -1564,6 +1630,7 @@ export class PiRpcAgentSession implements AgentSession {
         this.pendingSettledMessages = null;
         this.activeAssistantMessageId = null;
         this.pendingSteerSubmissions.length = 0;
+        this.pendingForegroundSubmission = null;
         this.clearNoTurnBuffers();
         this.emit({
           type: "turn_failed",
@@ -1588,6 +1655,7 @@ export class PiRpcAgentSession implements AgentSession {
       this.pendingSettledMessages = null;
       this.activeAssistantMessageId = null;
       this.pendingSteerSubmissions.length = 0;
+      this.pendingForegroundSubmission = null;
       this.clearNoTurnBuffers();
       this.emit({
         type: "turn_canceled",
@@ -2203,6 +2271,7 @@ export class PiRpcAgentSession implements AgentSession {
     this.activeTurnStartedEmitted = false;
     this.pendingSettledMessages = null;
     this.pendingSteerSubmissions.length = 0;
+    this.pendingForegroundSubmission = null;
     this.clearNoTurnBuffers();
     this.emit({
       type: "turn_failed",
@@ -2481,6 +2550,21 @@ export class PiRpcAgentSession implements AgentSession {
     event: Extract<PiAgentSessionEvent, { type: "message_end" }>,
     turnId: string | undefined,
   ): void {
+    if (event.message.role === "user") {
+      // Entry IDs arrive through the extension later; consumption is already known.
+      const text = getUserMessageText(event.message.content);
+      const foreground = this.pendingForegroundSubmission;
+      // prompt() runs asynchronously, so its first user event may follow steer().
+      if (foreground && !foreground.consumed && foreground.text === text) {
+        foreground.consumed = true;
+        return;
+      }
+      const submission = this.pendingSteerSubmissions.find(
+        (pending) => !pending.consumed && pending.text === text,
+      );
+      if (submission) submission.consumed = true;
+      return;
+    }
     if (event.message.role === "assistant") {
       this.activeAssistantMessageId = null;
       return;
@@ -2563,6 +2647,7 @@ export class PiRpcAgentSession implements AgentSession {
     this.activeTurnStartedEmitted = false;
     this.pendingSettledMessages = null;
     this.pendingSteerSubmissions.length = 0;
+    this.pendingForegroundSubmission = null;
     this.clearNoTurnBuffers();
     if (typeof errorMessage === "string" && errorMessage.length > 0) {
       this.usagePoller.stopTurn();

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

@@ -12,10 +12,12 @@ import {
 } from "../jsonl-rpc-process.js";
 import {
   buildPiLaunch,
+  PiClearedQueueSchema,
   type PiRuntime,
   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 PiClearedQueueSchema.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

@@ -1,4 +1,5 @@
 import type { PiTranscript } from "./transcript.js";
+import { z } from "zod";
 import type {
   PiModel,
   PiPromptAck,
@@ -39,6 +40,12 @@ export interface PiStartSessionInput {
   extraArgs?: string[];
 }
 
+export const PiClearedQueueSchema = z.object({
+  steering: z.array(z.string()),
+  followUp: z.array(z.string()),
+});
+export type PiClearedQueue = z.infer<typeof PiClearedQueueSchema>;
+
 export interface PiRuntimeSession {
   readonly environment: Record<string, string>;
   onEvent(callback: (event: PiRuntimeEvent) => void): () => void;
@@ -50,7 +57,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 if (text) {
+        const index = this.steeringMessages.indexOf(text);
+        if (index >= 0) this.steeringMessages.splice(index, 1);
+      }
+    }
     for (const subscriber of this.subscribers) {
       subscriber(event);
     }