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/pi/agent.test.ts

@@ -2490,6 +2490,34 @@ 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 ", "");

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

@@ -1582,6 +1582,9 @@ export class PiRpcAgentSession implements AgentSession {
         (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",

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

@@ -3,7 +3,6 @@ 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 {
@@ -13,6 +12,7 @@ import {
 } from "../jsonl-rpc-process.js";
 import {
   buildPiLaunch,
+  PiClearedQueueSchema,
   type PiRuntime,
   type PiRuntimeLaunch,
   type PiRuntimeSession,
@@ -154,7 +154,7 @@ class PiCliRuntimeSession implements PiRuntimeSession {
   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);
+    return PiClearedQueueSchema.parse(result);
   }
 
   async compact(customInstructions?: string): 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,10 +40,11 @@ export interface PiStartSessionInput {
   extraArgs?: string[];
 }
 
-export interface PiClearedQueue {
-  steering: string[];
-  followUp: 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>;

packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts

@@ -443,7 +443,7 @@ export class FakePiSession implements PiRuntimeSession {
       const text = getUserMessageText(message.content);
       if (this.foregroundMessage === text) {
         this.foregroundMessage = null;
-      } else {
+      } else if (text) {
         const index = this.steeringMessages.indexOf(text);
         if (index >= 0) this.steeringMessages.splice(index, 1);
       }