From a79b3733421ead0dcea3cbe32247ea2464400dcb Mon Sep 17 00:00:00 2001 From: Vegard Stikbakke Date: Tue, 25 Aug 2026 15:19:49 +0200 Subject: [PATCH] feat(coding-agent): expose RPC queue clearing closes #8432 --- packages/coding-agent/CHANGELOG.md | 4 ++ packages/coding-agent/docs/rpc.md | 23 ++++++++ .../coding-agent/src/modes/rpc/rpc-client.ts | 8 +++ .../coding-agent/src/modes/rpc/rpc-mode.ts | 4 ++ .../coding-agent/src/modes/rpc/rpc-types.ts | 8 +++ .../test/rpc-client-clear-queue.test.ts | 29 ++++++++++ .../rpc-prompt-response-semantics.test.ts | 54 +++++++++++++++++++ 7 files changed, 130 insertions(+) create mode 100644 packages/coding-agent/test/rpc-client-clear-queue.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 491aa86b4..381fb7e8a 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Added RPC `clear_queue` to retrieve and remove queued steering and follow-up messages ([#8432](https://github.com/earendil-works/pi/issues/8432)). + ### Fixed - Fixed extension messages sent with `triggerTurn: false` while the agent is running being inserted between a tool call and its result, which made providers that validate message order reject the replayed history. They are now appended once the turn's tool results are in ([#8537](https://github.com/earendil-works/pi/issues/8537)). diff --git a/packages/coding-agent/docs/rpc.md b/packages/coding-agent/docs/rpc.md index f92c12e8e..52dbf884f 100644 --- a/packages/coding-agent/docs/rpc.md +++ b/packages/coding-agent/docs/rpc.md @@ -134,6 +134,29 @@ Response: {"type": "response", "command": "abort", "success": true} ``` +#### clear_queue + +Remove queued steering and follow-up messages and return their text. + +```json +{"type": "clear_queue"} +``` + +Response: +```json +{ + "type": "response", + "command": "clear_queue", + "success": true, + "data": { + "steering": ["Change direction"], + "followUp": ["Summarize when finished"] + } +} +``` + +To implement interactive Esc behavior, send `clear_queue` before `abort`, then restore the returned text in the client editor. `abort` continues queued messages when they remain in the session. + #### new_session Start a fresh session. Can be cancelled by a `session_before_switch` extension event handler. diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index a4cc4b208..8b8b8cc38 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -220,6 +220,14 @@ export class RpcClient { await this.send({ type: "abort" }); } + /** + * Clear queued steering and follow-up messages, returning their text. + */ + async clearQueue(): Promise<{ steering: string[]; followUp: string[] }> { + const response = await this.send({ type: "clear_queue" }); + return this.getData(response); + } + /** * Start a new session, optionally with parent tracking. * @param parentSession - Optional parent session path for lineage tracking diff --git a/packages/coding-agent/src/modes/rpc/rpc-mode.ts b/packages/coding-agent/src/modes/rpc/rpc-mode.ts index 1b5d8a656..fc8083bed 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-mode.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-mode.ts @@ -430,6 +430,10 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise Promise; + getData: (response: unknown) => T; +}; + +describe("RpcClient clearQueue", () => { + it("sends the clear_queue RPC command", async () => { + const client = new RpcClient(); + const privateClient = client as unknown as RpcClientPrivate; + const send = vi.fn(async () => ({ + type: "response", + command: "clear_queue", + success: true, + data: { steering: ["Change direction"], followUp: ["Summarize when finished"] }, + })); + privateClient.send = send; + privateClient.getData = (response: unknown): T => { + return (response as { data: T }).data; + }; + + const result = await client.clearQueue(); + + expect(send).toHaveBeenCalledWith({ type: "clear_queue" }); + expect(result).toEqual({ steering: ["Change direction"], followUp: ["Summarize when finished"] }); + }); +}); diff --git a/packages/coding-agent/test/rpc-prompt-response-semantics.test.ts b/packages/coding-agent/test/rpc-prompt-response-semantics.test.ts index 698773d0d..bdc47b4a5 100644 --- a/packages/coding-agent/test/rpc-prompt-response-semantics.test.ts +++ b/packages/coding-agent/test/rpc-prompt-response-semantics.test.ts @@ -285,4 +285,58 @@ describe("RPC prompt response semantics", () => { await cleanup(); } }); + + it("returns and clears queued steering and follow-up messages", async () => { + const { lineHandler, cleanup } = await startRpcMode({ withAuth: true, responseDelayMs: 500 }); + + try { + lineHandler(JSON.stringify({ id: "clear-start", type: "prompt", message: "Start" })); + await vi.waitFor(() => { + expect(getPromptResponses(rpcIo.outputLines, "clear-start")).toHaveLength(1); + }); + + lineHandler( + JSON.stringify({ + id: "clear-steering", + type: "prompt", + message: "Change direction", + streamingBehavior: "steer", + }), + ); + await vi.waitFor(() => { + expect(getPromptResponses(rpcIo.outputLines, "clear-steering")).toHaveLength(1); + }); + + lineHandler( + JSON.stringify({ + id: "clear-follow-up", + type: "prompt", + message: "Summarize when finished", + streamingBehavior: "followUp", + }), + ); + await vi.waitFor(() => { + expect(getPromptResponses(rpcIo.outputLines, "clear-follow-up")).toHaveLength(1); + }); + + lineHandler(JSON.stringify({ id: "clear", type: "clear_queue" })); + await vi.waitFor(() => { + expect(parseOutputLines(rpcIo.outputLines)).toContainEqual({ + id: "clear", + type: "response", + command: "clear_queue", + success: true, + data: { + steering: ["Change direction"], + followUp: ["Summarize when finished"], + }, + }); + }); + + await sleep(600); + expect(parseOutputLines(rpcIo.outputLines).filter((record) => record.type === "agent_start")).toHaveLength(1); + } finally { + await cleanup(); + } + }); });