diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts index ddf85eb316..f2093c6ba4 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/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 { + ): Promise { 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 diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index 5ae9d8f9c4..2296d25900 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/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 { - return this.process.requestStopWork(OmpRpcCommandSchema.parse(command)); + private async requestStopWork(command: OmpRpcCommand): Promise { + await this.process.requestStopWork(OmpRpcCommandSchema.parse(command)); } private emit(event: OmpRuntimeEvent): void { 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 cd6eff1ed8..8a1dc8db24 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,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", { diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index e7539ee02e..23cd206fdd 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/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 { 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). diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts index 8a838d74d9..f907189c34 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts +++ b/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" }); }); 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 78c1fc7cf6..4b4979d455 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/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 { - await this.requestStopWork({ type: "clear_queue" }); + 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); } async compact(customInstructions?: string): Promise { @@ -268,8 +272,8 @@ class PiCliRuntimeSession implements PiRuntimeSession { return this.options.process.request(command, timeoutMs); } - private requestStopWork(command: PiRpcCommand): Promise { - return this.options.process.requestStopWork(command); + private async requestStopWork(command: PiRpcCommand): Promise { + await this.options.process.requestStopWork(command); } private async waitForCompletion(command: PiRpcCommand): Promise { diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index 42399d3275..9155ee8081 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/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; onEvent(callback: (event: PiRuntimeEvent) => void): () => void; @@ -50,7 +55,7 @@ export interface PiRuntimeSession { message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, ): Promise; - clearQueue(): Promise; + clearQueue(): Promise; compact(customInstructions?: string): Promise; setAutoCompaction(enabled: boolean): Promise; abort(): Promise; 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 770f24625c..b547ebe611 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 @@ -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 | 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(); private nextHeldPrompt: { promise: Promise; reject: (error: Error) => void } | null = null; private activeHeldPrompt: { promise: Promise; reject: (error: Error) => void } | null = @@ -217,6 +222,7 @@ export class FakePiSession implements PiRuntimeSession { images?: Array<{ type: "image"; data: string; mimeType: string }>, ): Promise { 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 { + async clearQueue(): Promise { 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 { @@ -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); }