sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(coding-agent): report prompt disposition in RPC responses

Browse source 
Successful RPC prompt, steer, and follow_up responses include
data.disposition ("handled", "queued", or "started" for prompt), so clients
know whether to wait for agent_settled.

fixes #9098
This commit is contained in:
David Brailovsky 2026-09-25 14:51:52 +02:00
parent b3487650f6
commit e473b5cd8b
11 changed files with 335 additions and 188 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/coding-agent/CHANGELOG.md
View file

@ -4,6 +4,7 @@
### Added
- Added per-input disposition to successful RPC `prompt`, `steer`, and `follow_up` responses, `AgentSession.steer()`/`followUp()`, and `RpcClient.prompt()`/`steer()`/`followUp()`; `RpcClient.prompt()` also accepts `streamingBehavior` ([#9098](https://github.com/earendil-works/pi/issues/9098), [#9803](https://github.com/earendil-works/pi/issues/9803)).
- Added image generation to `ModelRuntime`: `generateImages()` with runtime-resolved auth (stored credentials, OAuth, runtime API keys, `models.json` headers), plus `getModelsOfType()`, `getModelOfType()`, `getAvailableOfType()`, `getAllModels()`, and `getAllAvailable()`. OpenRouter image models are listed under the `openrouter` provider and share its credential; an upstream ID can have separate chat and image entries. `models.json` providers and extension registrations without a model list keep built-in image generation. Extension model lists can include discriminated chat, image, and classifier entries with operation implementations; when supplied, they replace the provider catalog across every operation. Chat-facing reads (`getModels()`, `getAvailableSnapshot()`, the model picker) are unchanged.
- Added classifier support to `ModelRuntime`, including `classify()`, classifier model accessors, runtime-resolved authentication, and the built-in TypeSafe `jev-latest` model.
- Added `types=chat,image,classifier` to pi.dev model catalog requests so remote refreshes overlay every supported model type; entries of unknown model types are ignored.

12
packages/coding-agent/docs/rpc-commands.md
View file

@ -34,9 +34,11 @@ If the agent is streaming and no `streamingBehavior` is specified, the command r
Response:
```json
{"id": "req-1", "type": "response", "command": "prompt", "success": true}
{"id": "req-1", "type": "response", "command": "prompt", "success": true, "data": {"disposition": "started"}}
```
`data.disposition` is `"handled"` if an extension command or input handler consumed the prompt, `"queued"` if Pi queued it during a run, or `"started"` if Pi accepted it to start a run. This describes the submitted prompt, not independent work started by an extension or a guarantee of completion.
`success: true` means the prompt was accepted, queued, or handled immediately. `success: false` means the prompt was rejected before acceptance. Failures after acceptance are reported through the normal event and message stream, not as a second `response` for the same request id.
The `images` field is optional. Each image uses `ImageContent` format: `{"type": "image", "data": "base64-encoded-data", "mimeType": "image/png"}`.
@ -58,9 +60,11 @@ The `images` field is optional. Each image uses `ImageContent` format (same as `
Response:
```json
{"type": "response", "command": "steer", "success": true}
{"type": "response", "command": "steer", "success": true, "data": {"disposition": "queued"}}
```
`data.disposition` is `"handled"` if an input handler consumed this steer, or `"queued"` if Pi queued it (including after a handler transformed it). It does not guarantee this message remains queued.
See [set_steering_mode](#set_steering_mode) for controlling how steering messages are processed.
### follow_up
@ -80,9 +84,11 @@ The `images` field is optional. Each image uses `ImageContent` format (same as `
Response:
```json
{"type": "response", "command": "follow_up", "success": true}
{"type": "response", "command": "follow_up", "success": true, "data": {"disposition": "queued"}}
```
`data.disposition` has the same `"handled"` or `"queued"` meaning as for `steer`, applied to this follow-up.
See [set_follow_up_mode](#set_follow_up_mode) for controlling how follow-up messages are processed.
### abort

4
packages/coding-agent/docs/rpc.md
View file

@ -61,9 +61,11 @@ A successful `prompt` response means the prompt was accepted, queued, or handled
```json
{"id":"req-2","type":"prompt","message":"Review this repository"}
{"id":"req-2","type":"response","command":"prompt","success":true}
{"id":"req-2","type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}
```
`data.disposition` reports what happened to the prompt. If it is `"handled"`, no run started for this prompt, so don't wait for `agent_settled`. See [RPC Commands](rpc-commands.md#prompt) for all values.
Continue consuming [events](json.md) after that response. `agent_end` marks the end of one low-level agent run, but retries, overflow recovery, compaction, steering, or follow-up work can still follow. Wait for `agent_settled` when the client needs to know Pi will not continue automatically.
Subscribe before sending a prompt to avoid missing a fast completion. `RpcClient.promptAndWait()` does this internally. If using separate `RpcClient` calls, install the event listener before `prompt()` and call `waitForIdle()` only while a run is active.

2
packages/coding-agent/docs/sdk.md
View file

@ -65,7 +65,7 @@ After a runtime replacement, subscriptions belong to the old `AgentSession` and
A prompt sent while the session is already streaming must specify whether it should steer the current run or follow it. Calling `prompt()` without that choice rejects rather than guessing.
A steering message enters after the current assistant turn and its tool calls. A follow-up enters after the current run finishes its pending work. `steer()` and `followUp()` expose those behaviors directly.
A steering message enters after the current assistant turn and its tool calls. A follow-up enters after the current run finishes its pending work. `steer()` and `followUp()` expose those behaviors directly and return `"queued"` if the input was queued (including after an extension transformed it), or `"handled"` if an extension consumed it.
`abort()` stops the active operation and waits for the session to become idle. `waitForIdle()` waits without aborting it.

296
packages/coding-agent/src/core/agent-session.ts
View file

@ -260,6 +260,9 @@ export interface ExtensionBindings {
onError?: ExtensionErrorListener;
}
export type QueuedInputDisposition = "handled" | "queued";
export type PromptDisposition = QueuedInputDisposition | "started";
/** Options for AgentSession.prompt() */
export interface PromptOptions {
/** Whether to dispatch extension commands and expand skill commands and prompt templates (default: true) */
@ -270,8 +273,8 @@ export interface PromptOptions {
streamingBehavior?: "steer" | "followUp";
/** Source of input for extension input event handlers. Defaults to "interactive". */
source?: InputSource;
/** Internal hook used by RPC mode to observe prompt preflight acceptance or rejection. */
preflightResult?: (success: boolean) => void;
/** Internal hook used by RPC mode to observe how an accepted prompt was dispatched. Not called if the prompt is rejected. */
preflightResult?: (disposition: PromptDisposition) => void;
}
/** Options for model/thinking mutations. */
@ -1610,153 +1613,141 @@ export class AgentSession {
}
const expandPromptTemplates = options?.expandPromptTemplates ?? true;
const preflightResult = options?.preflightResult;
let messages: AgentMessage[] | undefined;
try {
// Handle extension commands first (execute immediately, even during streaming)
// Extension commands manage their own LLM interaction via pi.sendMessage()
if (expandPromptTemplates && text.startsWith("/")) {
const handled = await this._tryExecuteExtensionCommand(text);
if (handled) {
// Extension command executed, no prompt to send
preflightResult?.(true);
return;
}
}
if (this._compactionAbortController !== undefined) {
throw new Error(
"Cannot submit a prompt while compaction is in progress. Wait for compaction to finish and retry.",
);
}
// Emit input event for extension interception (before skill/template expansion)
const processedInput = await this._runInputHandlers(
text,
options?.images,
options?.source ?? "interactive",
this.isStreaming ? options?.streamingBehavior : undefined,
);
if (!processedInput) {
preflightResult?.(true);
// Handle extension commands first (execute immediately, even during streaming)
// Extension commands manage their own LLM interaction via pi.sendMessage()
if (expandPromptTemplates && text.startsWith("/")) {
const handled = await this._tryExecuteExtensionCommand(text);
if (handled) {
// Extension command executed, no prompt to send
preflightResult?.("handled");
return;
}
const { text: currentText, images: currentImages } = processedInput;
// Expand skill commands (/skill:name args) and prompt templates (/template args)
let expandedText = currentText;
if (expandPromptTemplates) {
expandedText = this._expandSkillCommand(expandedText);
expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]);
}
// If streaming, queue via steer() or followUp() based on option
if (this.isStreaming) {
if (!options?.streamingBehavior) {
throw new Error(
"Agent is already processing. Specify streamingBehavior ('steer' or 'followUp') to queue the message.",
);
}
if (options.streamingBehavior === "followUp") {
await this._queueFollowUp(expandedText, currentImages);
} else {
await this._queueSteer(expandedText, currentImages);
}
preflightResult?.(true);
return;
}
// Flush any pending bash and custom messages before the new prompt
this._flushPendingBashMessages();
this._flushPendingCustomMessages();
// Validate model
if (!this.model) {
throw new Error(formatNoModelSelectedMessage());
}
const hasConfiguredAuth =
this._modelRuntime.hasConfiguredAuth(this.model.provider) ||
(await this._modelRuntime.checkAuth(this.model.provider)) !== undefined;
if (!hasConfiguredAuth) {
const isOAuth = this._modelRuntime.isUsingOAuth(this.model.provider);
if (isOAuth) {
throw new Error(
`Authentication failed for "${this.model.provider}". ` +
`Credentials may have expired or network is unavailable. ` +
`Run '/login ${this.model.provider}' to re-authenticate.`,
);
}
throw new Error(formatNoApiKeyFoundMessage(this.model.provider));
}
// Check if we need to compact before sending (catches aborted responses).
// The user's new prompt is sent below, so do not call agent.continue() here.
const lastAssistant = this._findLastAssistantMessage();
if (lastAssistant) {
await this._checkCompaction(lastAssistant, false);
}
// Emit before_agent_start before normalizing images so extension-driven model
// selection determines the resize profile used for the request and history.
const selectedToolsBefore = this._baseSystemPromptOptions.selectedTools;
const result = await this._extensionRunner.emitBeforeAgentStart(
expandedText,
currentImages,
this._baseSystemPromptOptions,
);
// Handlers may edit event.systemPromptOptions.selectedTools or call setActiveTools(),
// which updates the live loadout instead. An explicit edit wins; otherwise the live
// loadout is authoritative, so a setActiveTools() call is not undone here.
const handlerEditedTools =
result.systemPromptOptions.selectedTools.length !== selectedToolsBefore.length ||
result.systemPromptOptions.selectedTools.some((name, index) => name !== selectedToolsBefore[index]);
if (!handlerEditedTools) result.systemPromptOptions.selectedTools = this.getActiveToolNames();
const normalized = await this._normalizePromptImages(currentImages);
const userText =
normalized.hints.length > 0 ? `${expandedText}\n\n${normalized.hints.join("\n")}` : expandedText;
// Build messages only after hooks and image normalization have completed.
messages = [];
const userContent: (TextContent | ImageContent)[] = [{ type: "text", text: userText }];
userContent.push(...normalized.images);
messages.push({
role: "user",
content: userContent,
timestamp: Date.now(),
});
// Inject any pending "nextTurn" messages as context alongside the user message
for (const msg of this._pendingNextTurnMessages) {
messages.push(msg);
}
this._pendingNextTurnMessages = [];
for (const msg of result.messages) {
messages.push({
role: "custom",
customType: msg.customType,
// Untyped extensions can pass null/missing content; normalize at ingestion.
content: msg.content ?? [],
display: msg.display,
details: msg.details,
timestamp: Date.now(),
});
}
const updateMessage = this._preparePromptAndToolLoadout(result.systemPromptOptions);
this._runSystemPromptOptions = result.systemPromptOptions;
if (updateMessage) messages.unshift(updateMessage);
} catch (error) {
preflightResult?.(false);
throw error;
}
if (!messages) {
if (this._compactionAbortController !== undefined) {
throw new Error(
"Cannot submit a prompt while compaction is in progress. Wait for compaction to finish and retry.",
);
}
// Emit input event for extension interception (before skill/template expansion)
const processedInput = await this._runInputHandlers(
text,
options?.images,
options?.source ?? "interactive",
this.isStreaming ? options?.streamingBehavior : undefined,
);
if (!processedInput) {
preflightResult?.("handled");
return;
}
const { text: currentText, images: currentImages } = processedInput;
// Expand skill commands (/skill:name args) and prompt templates (/template args)
let expandedText = currentText;
if (expandPromptTemplates) {
expandedText = this._expandSkillCommand(expandedText);
expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]);
}
// If streaming, queue via steer() or followUp() based on option
if (this.isStreaming) {
if (!options?.streamingBehavior) {
throw new Error(
"Agent is already processing. Specify streamingBehavior ('steer' or 'followUp') to queue the message.",
);
}
if (options.streamingBehavior === "followUp") {
await this._queueFollowUp(expandedText, currentImages);
} else {
await this._queueSteer(expandedText, currentImages);
}
preflightResult?.("queued");
return;
}
preflightResult?.(true);
// Flush any pending bash and custom messages before the new prompt
this._flushPendingBashMessages();
this._flushPendingCustomMessages();
// Validate model
if (!this.model) {
throw new Error(formatNoModelSelectedMessage());
}
const hasConfiguredAuth =
this._modelRuntime.hasConfiguredAuth(this.model.provider) ||
(await this._modelRuntime.checkAuth(this.model.provider)) !== undefined;
if (!hasConfiguredAuth) {
const isOAuth = this._modelRuntime.isUsingOAuth(this.model.provider);
if (isOAuth) {
throw new Error(
`Authentication failed for "${this.model.provider}". ` +
`Credentials may have expired or network is unavailable. ` +
`Run '/login ${this.model.provider}' to re-authenticate.`,
);
}
throw new Error(formatNoApiKeyFoundMessage(this.model.provider));
}
// Check if we need to compact before sending (catches aborted responses).
// The user's new prompt is sent below, so do not call agent.continue() here.
const lastAssistant = this._findLastAssistantMessage();
if (lastAssistant) {
await this._checkCompaction(lastAssistant, false);
}
// Emit before_agent_start before normalizing images so extension-driven model
// selection determines the resize profile used for the request and history.
const selectedToolsBefore = this._baseSystemPromptOptions.selectedTools;
const result = await this._extensionRunner.emitBeforeAgentStart(
expandedText,
currentImages,
this._baseSystemPromptOptions,
);
// Handlers may edit event.systemPromptOptions.selectedTools or call setActiveTools(),
// which updates the live loadout instead. An explicit edit wins; otherwise the live
// loadout is authoritative, so a setActiveTools() call is not undone here.
const handlerEditedTools =
result.systemPromptOptions.selectedTools.length !== selectedToolsBefore.length ||
result.systemPromptOptions.selectedTools.some((name, index) => name !== selectedToolsBefore[index]);
if (!handlerEditedTools) result.systemPromptOptions.selectedTools = this.getActiveToolNames();
const normalized = await this._normalizePromptImages(currentImages);
const userText = normalized.hints.length > 0 ? `${expandedText}\n\n${normalized.hints.join("\n")}` : expandedText;
// Build messages only after hooks and image normalization have completed.
const messages: AgentMessage[] = [];
const userContent: (TextContent | ImageContent)[] = [{ type: "text", text: userText }];
userContent.push(...normalized.images);
messages.push({
role: "user",
content: userContent,
timestamp: Date.now(),
});
// Inject any pending "nextTurn" messages as context alongside the user message
for (const msg of this._pendingNextTurnMessages) {
messages.push(msg);
}
this._pendingNextTurnMessages = [];
for (const msg of result.messages) {
messages.push({
role: "custom",
customType: msg.customType,
// Untyped extensions can pass null/missing content; normalize at ingestion.
content: msg.content ?? [],
display: msg.display,
details: msg.details,
timestamp: Date.now(),
});
}
const updateMessage = this._preparePromptAndToolLoadout(result.systemPromptOptions);
this._runSystemPromptOptions = result.systemPromptOptions;
if (updateMessage) messages.unshift(updateMessage);
preflightResult?.("started");
await this._runAgentPrompt(messages);
}
@ -1825,7 +1816,7 @@ export class AgentSession {
images: ImageContent[] | undefined,
behavior: "steer" | "followUp",
source: InputSource,
): Promise<void> {
): Promise<QueuedInputDisposition> {
if (text.startsWith("/")) {
this._throwIfExtensionCommand(text);
}
@ -1836,7 +1827,7 @@ export class AgentSession {
source,
this.isStreaming ? behavior : undefined,
);
if (!processedInput) return;
if (!processedInput) return "handled";
let expandedText = this._expandSkillCommand(processedInput.text);
expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]);
@ -1846,6 +1837,7 @@ export class AgentSession {
} else {
await this._queueFollowUp(expandedText, processedInput.images);
}
return "queued";
}
/**
@ -1857,8 +1849,12 @@ export class AgentSession {
* @param options Input source; defaults to interactive
* @throws Error if text is an extension command
*/
async steer(text: string, images?: ImageContent[], options?: { source?: InputSource }): Promise<void> {
await this._queueUserInput(text, images, "steer", options?.source ?? "interactive");
async steer(
text: string,
images?: ImageContent[],
options?: { source?: InputSource },
): Promise<QueuedInputDisposition> {
return this._queueUserInput(text, images, "steer", options?.source ?? "interactive");
}
/**
@ -1869,8 +1865,12 @@ export class AgentSession {
* @param options Input source; defaults to interactive
* @throws Error if text is an extension command
*/
async followUp(text: string, images?: ImageContent[], options?: { source?: InputSource }): Promise<void> {
await this._queueUserInput(text, images, "followUp", options?.source ?? "interactive");
async followUp(
text: string,
images?: ImageContent[],
options?: { source?: InputSource },
): Promise<QueuedInputDisposition> {
return this._queueUserInput(text, images, "followUp", options?.source ?? "interactive");
}
/**

25
packages/coding-agent/src/modes/rpc/rpc-client.ts
View file

@ -7,7 +7,7 @@
import { type ChildProcess, spawn } from "node:child_process";
import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core";
import type { ImageContent } from "@earendil-works/pi-ai";
import type { SessionStats } from "../../core/agent-session.ts";
import type { PromptDisposition, QueuedInputDisposition, SessionStats } from "../../core/agent-session.ts";
import type { BashResult } from "../../core/bash-executor.ts";
import type { CompactionResult } from "../../core/compaction/index.ts";
import type { SessionEntry, SessionTreeNode } from "../../core/session-manager.ts";
@ -192,25 +192,32 @@ export class RpcClient {
/**
* Send a prompt to the agent.
* Returns immediately after sending; use onEvent() to receive streaming events.
* Use waitForIdle() to wait for completion.
* Returns the prompt's disposition after acceptance; use onEvent() to receive streaming events.
* If the disposition is "handled", no run started for this prompt, so don't wait for agent_settled.
*/
async prompt(message: string, images?: ImageContent[]): Promise<void> {
await this.send({ type: "prompt", message, images });
async prompt(
message: string,
images?: ImageContent[],
streamingBehavior?: "steer" | "followUp",
): Promise<PromptDisposition> {
const response = await this.send({ type: "prompt", message, images, streamingBehavior });
return this.getData<{ disposition: PromptDisposition }>(response).disposition;
}
/**
* Queue a steering message to interrupt the agent mid-run.
*/
async steer(message: string, images?: ImageContent[]): Promise<void> {
await this.send({ type: "steer", message, images });
async steer(message: string, images?: ImageContent[]): Promise<QueuedInputDisposition> {
const response = await this.send({ type: "steer", message, images });
return this.getData<{ disposition: QueuedInputDisposition }>(response).disposition;
}
/**
* Queue a follow-up message to be processed after the agent finishes.
*/
async followUp(message: string, images?: ImageContent[]): Promise<void> {
await this.send({ type: "follow_up", message, images });
async followUp(message: string, images?: ImageContent[]): Promise<QueuedInputDisposition> {
const response = await this.send({ type: "follow_up", message, images });
return this.getData<{ disposition: QueuedInputDisposition }>(response).disposition;
}
/**

16
packages/coding-agent/src/modes/rpc/rpc-mode.ts
View file

@ -400,11 +400,9 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
images: command.images,
streamingBehavior: command.streamingBehavior,
source: "rpc",
preflightResult: (didSucceed) => {
if (didSucceed) {
preflightSucceeded = true;
output(success(id, "prompt"));
}
preflightResult: (disposition) => {
preflightSucceeded = true;
output(success(id, "prompt", { disposition }));
},
})
.catch((e) => {
@ -416,13 +414,13 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
}
case "steer": {
await session.steer(command.message, command.images, { source: "rpc" });
return success(id, "steer");
const disposition = await session.steer(command.message, command.images, { source: "rpc" });
return success(id, "steer", { disposition });
}
case "follow_up": {
await session.followUp(command.message, command.images, { source: "rpc" });
return success(id, "follow_up");
const disposition = await session.followUp(command.message, command.images, { source: "rpc" });
return success(id, "follow_up", { disposition });
}
case "abort": {

16
packages/coding-agent/src/modes/rpc/rpc-types.ts
View file

@ -7,7 +7,7 @@
import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core";
import type { ImageContent, Model } from "@earendil-works/pi-ai";
import type { SessionStats } from "../../core/agent-session.ts";
import type { PromptDisposition, QueuedInputDisposition, SessionStats } from "../../core/agent-session.ts";
import type { BashResult } from "../../core/bash-executor.ts";
import type { CompactionResult } from "../../core/compaction/index.ts";
import type { SessionEntry, SessionTreeNode } from "../../core/session-manager.ts";
@ -114,10 +114,16 @@ export interface RpcSessionState {
// Success responses with data
export type RpcResponse =
// Prompting (async - events follow)
| { id?: string; type: "response"; command: "prompt"; success: true }
| { id?: string; type: "response"; command: "steer"; success: true }
| { id?: string; type: "response"; command: "follow_up"; success: true }
// Prompting
| { id?: string; type: "response"; command: "prompt"; success: true; data: { disposition: PromptDisposition } }
| { id?: string; type: "response"; command: "steer"; success: true; data: { disposition: QueuedInputDisposition } }
| {
id?: string;
type: "response";
command: "follow_up";
success: true;
data: { disposition: QueuedInputDisposition };
}
| { id?: string; type: "response"; command: "abort"; success: true }
| {
id?: string;

4
packages/coding-agent/test/agent-session-concurrent.test.ts
View file

@ -161,7 +161,7 @@ describe("AgentSession concurrent prompt guard", () => {
await new Promise((resolve) => setTimeout(resolve, 10));
// steer should work while streaming
await expect(session.steer("Steering message")).resolves.toBeUndefined();
await expect(session.steer("Steering message")).resolves.toBe("queued");
expect(session.pendingMessageCount).toBe(1);
// Cleanup
@ -177,7 +177,7 @@ describe("AgentSession concurrent prompt guard", () => {
await new Promise((resolve) => setTimeout(resolve, 10));
// followUp should work while streaming
await expect(session.followUp("Follow-up message")).resolves.toBeUndefined();
await expect(session.followUp("Follow-up message")).resolves.toBe("queued");
expect(session.pendingMessageCount).toBe(1);
// Cleanup

138
packages/coding-agent/test/rpc-prompt-response-semantics.test.ts
View file

@ -13,11 +13,12 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { AgentSession } from "../src/core/agent-session.ts";
import type { AgentSessionRuntime } from "../src/core/agent-session-runtime.ts";
import { AuthStorage } from "../src/core/auth-storage.ts";
import type { LoadExtensionsResult } from "../src/core/extensions/index.ts";
import { SessionManager } from "../src/core/session-manager.ts";
import { SettingsManager } from "../src/core/settings-manager.ts";
import { runRpcMode } from "../src/modes/rpc/rpc-mode.ts";
import { createInMemoryModelRegistry, getModelRuntime } from "./model-runtime-test-utils.ts";
import { createTestResourceLoader } from "./utilities.ts";
import { createTestExtensionsResult, createTestResourceLoader } from "./utilities.ts";
const rpcIo = vi.hoisted(() => ({
outputLines: [] as string[],
@ -95,7 +96,12 @@ function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function createRuntimeHost(options: { withAuth: boolean; responseDelayMs: number; model?: Model<any> }): Promise<{
async function createRuntimeHost(options: {
withAuth: boolean;
responseDelayMs: number;
model?: Model<any>;
extensionsResult?: LoadExtensionsResult;
}): Promise<{
runtimeHost: AgentSessionRuntime;
cleanup: () => Promise<void>;
}> {
@ -140,7 +146,7 @@ async function createRuntimeHost(options: { withAuth: boolean; responseDelayMs:
settingsManager,
cwd: tempDir,
modelRuntime: getModelRuntime(modelRegistry),
resourceLoader: createTestResourceLoader(),
resourceLoader: createTestResourceLoader({ extensionsResult: options.extensionsResult }),
});
const runtimeHost = {
@ -168,7 +174,7 @@ async function createRuntimeHost(options: { withAuth: boolean; responseDelayMs:
};
}
async function startRpcMode(options: { withAuth: boolean; responseDelayMs: number; model?: Model<any> }): Promise<{
async function startRpcMode(options: Parameters<typeof createRuntimeHost>[0]): Promise<{
lineHandler: (line: string) => void;
cleanup: () => Promise<void>;
}> {
@ -227,7 +233,8 @@ describe("RPC prompt response semantics", () => {
}
});
it("emits one success response when prompt preflight succeeds", async () => {
// #9098: a successful prompt may start an agent run or be consumed by an extension.
it("emits one started response when prompt preflight succeeds", async () => {
const { lineHandler, cleanup } = await startRpcMode({ withAuth: true, responseDelayMs: 0 });
try {
@ -241,6 +248,7 @@ describe("RPC prompt response semantics", () => {
type: "response",
command: "prompt",
success: true,
data: { disposition: "started" },
});
});
} finally {
@ -248,6 +256,38 @@ describe("RPC prompt response semantics", () => {
}
});
it("reports extension commands and intercepted input as handled without starting a run", async () => {
const { lineHandler, cleanup } = await startRpcMode({
withAuth: false,
responseDelayMs: 0,
extensionsResult: await createTestExtensionsResult([
(pi) => {
pi.registerCommand("handled", { handler: async () => {} });
pi.on("input", (event) => {
if (event.text === "handled input") return { action: "handled" };
});
},
]),
});
try {
for (const [id, message] of [
["command", "/handled"],
["input", "handled input"],
]) {
lineHandler(JSON.stringify({ id, type: "prompt", message }));
await vi.waitFor(() => {
expect(getPromptResponses(rpcIo.outputLines, id)).toEqual([
{ id, type: "response", command: "prompt", success: true, data: { disposition: "handled" } },
]);
});
}
expect(parseOutputLines(rpcIo.outputLines).filter((line) => line.type === "agent_start")).toHaveLength(0);
} finally {
await cleanup();
}
});
it("emits one success response when prompt is queued during streaming", async () => {
const { lineHandler, cleanup } = await startRpcMode({ withAuth: true, responseDelayMs: 100 });
@ -275,6 +315,7 @@ describe("RPC prompt response semantics", () => {
type: "response",
command: "prompt",
success: true,
data: { disposition: "queued" },
});
});
@ -284,6 +325,89 @@ describe("RPC prompt response semantics", () => {
}
});
// #9803: a handler can consume A while independently queueing B; report A's outcome.
it.each(["steer", "follow_up"] as const)(
"reports %s as handled even when an extension queues another message",
async (type) => {
const { lineHandler, cleanup } = await startRpcMode({
withAuth: true,
responseDelayMs: 500,
extensionsResult: await createTestExtensionsResult([
(pi) => {
pi.on("input", (event) => {
if (event.text === "A" && event.source === "rpc") {
pi.sendUserMessage("B", { deliverAs: type === "steer" ? "steer" : "followUp" });
return { action: "handled" };
}
});
},
]),
});
try {
lineHandler(JSON.stringify({ id: "start", type: "prompt", message: "Start" }));
await vi.waitFor(() => expect(getPromptResponses(rpcIo.outputLines, "start")).toHaveLength(1));
lineHandler(JSON.stringify({ id: "A", type, message: "A" }));
await vi.waitFor(() => {
expect(parseOutputLines(rpcIo.outputLines)).toContainEqual({
id: "A",
type: "response",
command: type,
success: true,
data: { disposition: "handled" },
});
expect(parseOutputLines(rpcIo.outputLines)).toContainEqual({
type: "queue_update",
steering: type === "steer" ? ["B"] : [],
followUp: type === "follow_up" ? ["B"] : [],
});
});
await vi.waitFor(() => {
expect(
parseOutputLines(rpcIo.outputLines).filter((line) => line.type === "response" && line.id === "A"),
).toHaveLength(1);
});
} finally {
await cleanup();
}
},
);
it.each(["steer", "follow_up"] as const)("reports %s as queued after input transformation", async (type) => {
const { lineHandler, cleanup } = await startRpcMode({
withAuth: false,
responseDelayMs: 0,
extensionsResult: await createTestExtensionsResult([
(pi) => {
pi.on("input", (event) => {
if (event.text === "A") return { action: "transform", text: "B" };
});
},
]),
});
try {
lineHandler(JSON.stringify({ id: "A", type, message: "A" }));
await vi.waitFor(() => {
expect(parseOutputLines(rpcIo.outputLines)).toContainEqual({
id: "A",
type: "response",
command: type,
success: true,
data: { disposition: "queued" },
});
expect(parseOutputLines(rpcIo.outputLines)).toContainEqual({
type: "queue_update",
steering: type === "steer" ? ["B"] : [],
followUp: type === "follow_up" ? ["B"] : [],
});
});
} finally {
await cleanup();
}
});
it("returns and clears queued steering and follow-up messages", async () => {
const { lineHandler, cleanup } = await startRpcMode({ withAuth: true, responseDelayMs: 500 });
@ -302,7 +426,9 @@ describe("RPC prompt response semantics", () => {
}),
);
await vi.waitFor(() => {
expect(getPromptResponses(rpcIo.outputLines, "clear-steering")).toHaveLength(1);
expect(getPromptResponses(rpcIo.outputLines, "clear-steering")).toMatchObject([
{ data: { disposition: "queued" } },
]);
});
lineHandler(

9
packages/coding-agent/test/suite/regressions/7150-rpc-prompt-during-compaction.test.ts
View file

@ -1,5 +1,6 @@
import { fauxAssistantMessage } from "@earendil-works/pi-ai";
import { afterEach, describe, expect, it } from "vitest";
import type { PromptDisposition } from "../../../src/core/agent-session.ts";
import { createHarness, getMessageText, getUserTexts, type Harness } from "../harness.ts";
describe("issue #7150: RPC prompt during manual compaction", () => {
@ -57,13 +58,13 @@ describe("issue #7150: RPC prompt during manual compaction", () => {
const compactPromise = harness.session.compact();
await compactionStarted;
let preflightResult: boolean | undefined;
let preflightResult: PromptDisposition | undefined;
let promptError: unknown;
try {
await harness.session.prompt("PROBE-7150", {
source: "rpc",
preflightResult: (success) => {
preflightResult = success;
preflightResult: (result) => {
preflightResult = result;
},
});
} catch (error) {
@ -79,7 +80,7 @@ describe("issue #7150: RPC prompt during manual compaction", () => {
entry.type === "message" && entry.message.role === "user" ? [getMessageText(entry.message)] : [],
);
expect(preflightResult).toBe(false);
expect(preflightResult).toBeUndefined();
expect(promptError).toEqual(
expect.objectContaining({ message: expect.stringContaining("compaction is in progress") }),
);