From 60a7625ad9742dd2aa266cc71d4e4c9bf04b58d9 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 22:03:52 +0000 Subject: [PATCH] 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. --- .../server/agent/providers/pi/agent.test.ts | 28 +++++++++++++++++++ .../src/server/agent/providers/pi/agent.ts | 3 ++ .../server/agent/providers/pi/cli-runtime.ts | 4 +-- .../src/server/agent/providers/pi/runtime.ts | 10 ++++--- .../agent/providers/pi/test-utils/fake-pi.ts | 2 +- 5 files changed, 40 insertions(+), 7 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 8a1dc8db24..5c2f135652 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/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 ", ""); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 23cd206fdd..bb798f0d13 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/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", 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 4b4979d455..a32be667bf 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/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 { 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 { diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index 9155ee8081..1b52cfc62c 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/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; export interface PiRuntimeSession { readonly environment: Record; 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 b547ebe611..e704d65d5e 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 @@ -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); }