From e473b5cd8b6f2e82f1295aeee81cb697ffde2ed5 Mon Sep 17 00:00:00 2001 From: David Brailovsky Date: Fri, 25 Sep 2026 14:51:52 +0200 Subject: [PATCH] feat(coding-agent): report prompt disposition in RPC responses 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 --- packages/coding-agent/CHANGELOG.md | 1 + packages/coding-agent/docs/rpc-commands.md | 12 +- packages/coding-agent/docs/rpc.md | 4 +- packages/coding-agent/docs/sdk.md | 2 +- .../coding-agent/src/core/agent-session.ts | 296 +++++++++--------- .../coding-agent/src/modes/rpc/rpc-client.ts | 25 +- .../coding-agent/src/modes/rpc/rpc-mode.ts | 16 +- .../coding-agent/src/modes/rpc/rpc-types.ts | 16 +- .../test/agent-session-concurrent.test.ts | 4 +- .../rpc-prompt-response-semantics.test.ts | 138 +++++++- .../7150-rpc-prompt-during-compaction.test.ts | 9 +- 11 files changed, 335 insertions(+), 188 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index ca85bcf72..04810c1c8 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -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. diff --git a/packages/coding-agent/docs/rpc-commands.md b/packages/coding-agent/docs/rpc-commands.md index bd67f8492..24f603265 100644 --- a/packages/coding-agent/docs/rpc-commands.md +++ b/packages/coding-agent/docs/rpc-commands.md @@ -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 diff --git a/packages/coding-agent/docs/rpc.md b/packages/coding-agent/docs/rpc.md index ad6ff90e8..a39cc8da2 100644 --- a/packages/coding-agent/docs/rpc.md +++ b/packages/coding-agent/docs/rpc.md @@ -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. diff --git a/packages/coding-agent/docs/sdk.md b/packages/coding-agent/docs/sdk.md index 596caccec..be1428d2d 100644 --- a/packages/coding-agent/docs/sdk.md +++ b/packages/coding-agent/docs/sdk.md @@ -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. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 338d0e0d0..b3c8a3aff 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -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 { + ): Promise { 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 { - await this._queueUserInput(text, images, "steer", options?.source ?? "interactive"); + async steer( + text: string, + images?: ImageContent[], + options?: { source?: InputSource }, + ): Promise { + 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 { - await this._queueUserInput(text, images, "followUp", options?.source ?? "interactive"); + async followUp( + text: string, + images?: ImageContent[], + options?: { source?: InputSource }, + ): Promise { + return this._queueUserInput(text, images, "followUp", options?.source ?? "interactive"); } /** diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index af357ee27..94ee29fbf 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -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 { - await this.send({ type: "prompt", message, images }); + async prompt( + message: string, + images?: ImageContent[], + streamingBehavior?: "steer" | "followUp", + ): Promise { + 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 { - await this.send({ type: "steer", message, images }); + async steer(message: string, images?: ImageContent[]): Promise { + 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 { - await this.send({ type: "follow_up", message, images }); + async followUp(message: string, images?: ImageContent[]): Promise { + const response = await this.send({ type: "follow_up", message, images }); + return this.getData<{ disposition: QueuedInputDisposition }>(response).disposition; } /** diff --git a/packages/coding-agent/src/modes/rpc/rpc-mode.ts b/packages/coding-agent/src/modes/rpc/rpc-mode.ts index f4857ffb2..1c0995d36 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-mode.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-mode.ts @@ -400,11 +400,9 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise { - 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 { 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 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 5d7c147ff..98325b19d 100644 --- a/packages/coding-agent/test/rpc-prompt-response-semantics.test.ts +++ b/packages/coding-agent/test/rpc-prompt-response-semantics.test.ts @@ -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 { return new Promise((resolve) => setTimeout(resolve, ms)); } -async function createRuntimeHost(options: { withAuth: boolean; responseDelayMs: number; model?: Model }): Promise<{ +async function createRuntimeHost(options: { + withAuth: boolean; + responseDelayMs: number; + model?: Model; + extensionsResult?: LoadExtensionsResult; +}): Promise<{ runtimeHost: AgentSessionRuntime; cleanup: () => Promise; }> { @@ -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 }): Promise<{ +async function startRpcMode(options: Parameters[0]): Promise<{ lineHandler: (line: string) => void; cleanup: () => Promise; }> { @@ -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( diff --git a/packages/coding-agent/test/suite/regressions/7150-rpc-prompt-during-compaction.test.ts b/packages/coding-agent/test/suite/regressions/7150-rpc-prompt-during-compaction.test.ts index e1588d1ce..558084ecd 100644 --- a/packages/coding-agent/test/suite/regressions/7150-rpc-prompt-during-compaction.test.ts +++ b/packages/coding-agent/test/suite/regressions/7150-rpc-prompt-during-compaction.test.ts @@ -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") }), );