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 f26568c0c3..cd6eff1ed8 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,107 @@ describe("PiRpcAgentSession", () => { }); describe("PiRpcAgentSession steering", () => { + 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"); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index b02ccf783c..6502088a2d 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -232,10 +232,16 @@ 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 // supports the steer RPC. Older binaries answer with "Unknown command: steer". function isPiDefinitiveSteerRejection(error: unknown): boolean { @@ -1211,6 +1217,7 @@ export class PiRpcAgentSession implements AgentSession { private activePromptRequestId: string | null = null; private readonly pendingPromptResults = new Map(); private readonly pendingSteerSubmissions: PiPendingSteerSubmission[] = []; + private pendingForegroundSubmission: PiPendingForegroundSubmission | null = null; currentLeafOverrideId: string | null | undefined; private readonly treeUserEntriesById = new Map(); private readonly pendingExtensionResults = new Map(); @@ -1320,6 +1327,7 @@ export class PiRpcAgentSession implements AgentSession { this.pendingSettledMessages = null; this.activePromptRequestId = null; this.pendingSteerSubmissions.length = 0; + this.pendingForegroundSubmission = { text: payload.text, consumed: false }; this.clearNoTurnBuffers(); const payload = await convertPromptInput(prompt, { model: this.state.model }); if (this.closed) throw new Error("Pi session is closed"); @@ -1366,6 +1374,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({ @@ -1404,8 +1413,14 @@ export class PiRpcAgentSession implements AgentSession { return { status: "unavailable" }; } + const imageCount = payload.images?.length ?? 0; + const label = + payload.text.trim().length === 0 && imageCount > 0 + ? `${imageCount} ${imageCount === 1 ? "image" : "images"}` + : payload.text; const submission: PiPendingSteerSubmission = { text: payload.text, + label, clientMessageId: options.clientMessageId ?? null, consumed: false, }; @@ -1444,6 +1459,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; @@ -1561,7 +1581,7 @@ export class PiRpcAgentSession implements AgentSession { item: { type: "notification", level: "warning", - message: `Not delivered (stopped): ${submission.text}`, + message: `Not delivered (stopped): ${submission.label}`, }, }); } @@ -1587,6 +1607,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", @@ -1611,6 +1632,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", @@ -2226,6 +2248,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", @@ -2507,6 +2530,12 @@ export class PiRpcAgentSession implements AgentSession { 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, ); @@ -2595,6 +2624,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();