feat(coding-agent): expose RPC queue clearing

closes #8432
This commit is contained in:
Vegard Stikbakke
2026-08-25 15:19:49 +02:00
parent 587be985a0
commit a79b373342
7 changed files with 130 additions and 0 deletions
+4
View File
@@ -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)).
+23
View File
@@ -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.
@@ -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
@@ -430,6 +430,10 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
return success(id, "abort");
}
case "clear_queue": {
return success(id, "clear_queue", session.clearQueue());
}
case "new_session": {
const options = command.parentSession ? { parentSession: command.parentSession } : undefined;
const result = await runtimeHost.newSession(options);
@@ -23,6 +23,7 @@ export type RpcCommand =
| { id?: string; type: "steer"; message: string; images?: ImageContent[] }
| { id?: string; type: "follow_up"; message: string; images?: ImageContent[] }
| { id?: string; type: "abort" }
| { id?: string; type: "clear_queue" }
| { id?: string; type: "new_session"; parentSession?: string }
// State
@@ -118,6 +119,13 @@ export type RpcResponse =
| { id?: string; type: "response"; command: "steer"; success: true }
| { id?: string; type: "response"; command: "follow_up"; success: true }
| { id?: string; type: "response"; command: "abort"; success: true }
| {
id?: string;
type: "response";
command: "clear_queue";
success: true;
data: { steering: string[]; followUp: string[] };
}
| { id?: string; type: "response"; command: "new_session"; success: true; data: { cancelled: boolean } }
// State
@@ -0,0 +1,29 @@
import { describe, expect, it, vi } from "vitest";
import { RpcClient } from "../src/modes/rpc/rpc-client.ts";
type RpcClientPrivate = {
send: (command: { type: string }) => Promise<unknown>;
getData: <T>(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 = <T>(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"] });
});
});
@@ -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();
}
});
});