fix: stop AgentBusyError on plan approval and loop/goal continuations

Plan approval dispatched the executor's first synthetic prompt without
checking whether the agent was still streaming the post-resolve
continuation (or a turn started by the approve-time compaction/clear),
surfacing "Failed to finalize approved plan: ... Agent is already
processing". Loop auto-submit and goal continuations hit the same throw
via submitInteractiveInput, which always called prompt/promptCustomMessage
without a streamingBehavior.

#approvePlan now aborts any in-flight turn before the synthetic prompt,
and submitInteractiveInput routes submissions through the steer/follow-up
queue (streamingBehavior: "followUp") when the session is streaming.
Non-streaming call shapes are unchanged. Extends the manual-/goal fix
(#2454) to the continuation and plan-approval paths.
This commit is contained in:
metaphorics
2026-06-14 07:37:14 +09:00
parent 11622e6190
commit 34746ceaf5
5 changed files with 128 additions and 6 deletions
+4
View File
@@ -2,6 +2,10 @@
## [Unreleased]
### Fixed
- Fixed `AgentBusyError` ("Agent is already processing. Use steer() or followUp()...") surfacing on mode transitions — as `Failed to finalize approved plan: ...` when a plan was approved while the agent was still streaming the post-`resolve` continuation (or a turn started by the approve-time compaction/clear), and as an error toast when a loop auto-submit or goal continuation fired during a streaming/compaction race. Plan approval now aborts any in-flight turn before dispatching the executor's first prompt, and `submitInteractiveInput` routes streaming-time loop, goal-continuation, and manual submissions through the follow-up queue (`streamingBehavior: "followUp"`) instead of throwing (synthetic continue-shortcuts stay developer-attributed and keep their prior behavior). Extends the manual-`/goal` fix in [#2454](https://github.com/can1357/oh-my-pi/issues/2454) to the continuation and plan-approval paths.
## [15.12.5] - 2026-06-13
### Changed
+15 -5
View File
@@ -264,7 +264,7 @@ export async function submitInteractiveInput(
InteractiveMode,
"markPendingSubmissionStarted" | "finishPendingSubmission" | "showError" | "checkShutdownRequested"
>,
session: Pick<AgentSession, "prompt" | "promptCustomMessage">,
session: Pick<AgentSession, "prompt" | "promptCustomMessage" | "isStreaming">,
input: SubmittedUserInput,
): Promise<void> {
if (input.cancelled) {
@@ -273,22 +273,32 @@ export async function submitInteractiveInput(
try {
using _keepalive = new EventLoopKeepalive();
const streamingBehavior = session.isStreaming ? ("followUp" as const) : undefined;
// Continue shortcuts submit an already-started synthetic developer prompt with
// no optimistic user message.
if (!input.started && !mode.markPendingSubmissionStarted(input)) {
return;
}
if (input.customType) {
await session.promptCustomMessage({
const message = {
customType: input.customType,
content: input.text,
display: input.display ?? false,
attribution: "agent",
});
attribution: "agent" as const,
};
await (streamingBehavior
? session.promptCustomMessage(message, { streamingBehavior })
: session.promptCustomMessage(message));
} else if (input.synthetic) {
// Synthetic continue shortcuts are hidden developer prompts. The streaming
// queue (#queueUserMessage) only carries user-attributed messages, so we do
// NOT pass streamingBehavior here: queueing would silently demote the
// developer directive to a visible user message. A synthetic submit while
// streaming keeps its prior behavior (rejected as busy) rather than changing
// its role.
await session.prompt(input.text, { synthetic: true, expandPromptTemplates: false });
} else {
await session.prompt(input.text, { images: input.images });
await session.prompt(input.text, { images: input.images, ...(streamingBehavior && { streamingBehavior }) });
}
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : "Unknown error occurred";
@@ -2217,6 +2217,15 @@ export class InteractiveMode implements InteractiveModeContext {
planFilePath: options.planFilePath,
contextPreserved: options.preserveContext === true,
});
// The executor's first turn must start on an idle session. The agent may still
// be streaming the post-`resolve` continuation (Agent.#emit is fire-and-forget)
// or a turn kicked off by the compaction/clear above; prompt() would then throw
// AgentBusyError ("Failed to finalize approved plan"). Abort the now-irrelevant
// in-flight turn first — abort() bumps the prompt generation and cancels pending
// continuations, so nothing re-streams in the synchronous gap before prompt().
if (this.session.isStreaming) {
await this.session.abort();
}
await this.session.prompt(planModePrompt, { synthetic: true });
}
@@ -1,7 +1,7 @@
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "bun:test";
import * as fs from "node:fs/promises";
import * as path from "node:path";
import { Agent } from "@oh-my-pi/pi-agent-core";
import { Agent, AgentBusyError } from "@oh-my-pi/pi-agent-core";
import type { AssistantMessage, Usage } from "@oh-my-pi/pi-ai";
import { KeybindingsManager } from "@oh-my-pi/pi-coding-agent/config/keybindings";
import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry";
@@ -577,6 +577,50 @@ describe("InteractiveMode plan review rendering", () => {
});
});
it("aborts an in-flight turn before dispatching the approved plan instead of surfacing AgentBusyError", async () => {
const planFilePath = "local://PLAN.md";
const resolvedPlanPath = resolveLocalUrlToPath(planFilePath, {
getArtifactsDir: () => session.sessionManager.getArtifactsDir(),
getSessionId: () => session.sessionManager.getSessionId(),
});
await Bun.write(resolvedPlanPath, "# Plan\n\nbody");
mode.planModeEnabled = true;
mode.planModePlanFilePath = planFilePath;
let streaming = false;
Object.defineProperty(session, "isStreaming", {
configurable: true,
get: () => streaming,
});
const abortSpy = vi.spyOn(session, "abort").mockImplementation(async () => {
// Clear the streaming flag only after an awaited tick, so the test fails
// if #approvePlan dispatches the prompt without awaiting abort() — the
// real abort() resolves only once the agent loop is idle.
await Promise.resolve();
streaming = false;
});
const promptSpy = vi.spyOn(session, "prompt").mockImplementation(async (_text, opts) => {
if (streaming && !(opts as { streamingBehavior?: string } | undefined)?.streamingBehavior)
throw new AgentBusyError();
return true;
});
// Simulate a re-stream landing during the overlay, then pick keep-context
// (options[2]) — that branch skips clear/compact so `this.session` stays the
// instance the spies are on.
vi.spyOn(mode, "showPlanReview").mockImplementation(async (_plan, _title, options) => {
streaming = true;
return options[2];
});
const errorSpy = vi.spyOn(mode, "showError");
await mode.handlePlanApproval({ planFilePath, planExists: true, title: "PLAN" });
expect(errorSpy).not.toHaveBeenCalledWith(expect.stringContaining("Failed to finalize approved plan"));
expect(promptSpy).toHaveBeenCalledTimes(1);
expect(isPlanApprovedCall(promptSpy.mock.calls[0] as unknown[])).toBe(true);
expect(abortSpy).toHaveBeenCalled();
});
it("keeps the existing approve-and-execute path clearing the session", async () => {
const planFilePath = "local://PLAN.md";
const resolvedPlanPath = resolveLocalUrlToPath(planFilePath, {
@@ -46,6 +46,7 @@ describe("submitInteractiveInput", () => {
const session = {
prompt: vi.fn(async () => true),
promptCustomMessage: vi.fn(async () => {}),
isStreaming: false,
};
const input = createInput({ text: "resume now", started: true, synthetic: true });
@@ -67,6 +68,7 @@ describe("submitInteractiveInput", () => {
const session = {
prompt: vi.fn(async () => true),
promptCustomMessage: vi.fn(async () => {}),
isStreaming: false,
};
const input = createInput();
@@ -88,6 +90,7 @@ describe("submitInteractiveInput", () => {
const session = {
prompt: vi.fn(async () => true),
promptCustomMessage: vi.fn(async () => {}),
isStreaming: false,
};
const input = createInput({ text: "continue goal", customType: "goal-continuation" });
@@ -103,4 +106,56 @@ describe("submitInteractiveInput", () => {
expect(mode.finishPendingSubmission).toHaveBeenCalledWith(input);
expect(mode.showError).not.toHaveBeenCalled();
});
it("queues goal-continuation as followUp when streaming", async () => {
const mode = {
markPendingSubmissionStarted: vi.fn(() => true),
finishPendingSubmission: vi.fn(),
showError: vi.fn(),
checkShutdownRequested: vi.fn(async () => {}),
};
const session = {
prompt: vi.fn(async () => true),
promptCustomMessage: vi.fn(async () => {}),
isStreaming: true,
};
const input = createInput({ text: "continue goal", customType: "goal-continuation" });
await submitInteractiveInput(mode, session, input);
expect(session.prompt).not.toHaveBeenCalled();
expect(session.promptCustomMessage).toHaveBeenCalledWith(
{
customType: "goal-continuation",
content: "continue goal",
display: false,
attribution: "agent",
},
{ streamingBehavior: "followUp" },
);
expect(mode.finishPendingSubmission).toHaveBeenCalledWith(input);
expect(mode.showError).not.toHaveBeenCalled();
});
it("queues a plain submission as followUp when streaming", async () => {
const mode = {
markPendingSubmissionStarted: vi.fn(() => true),
finishPendingSubmission: vi.fn(),
showError: vi.fn(),
checkShutdownRequested: vi.fn(async () => {}),
};
const session = {
prompt: vi.fn(async () => true),
promptCustomMessage: vi.fn(async () => {}),
isStreaming: true,
};
const input = createInput({ text: "loop prompt" });
await submitInteractiveInput(mode, session, input);
expect(session.prompt).toHaveBeenCalledWith("loop prompt", { images: undefined, streamingBehavior: "followUp" });
expect(session.promptCustomMessage).not.toHaveBeenCalled();
expect(mode.finishPendingSubmission).toHaveBeenCalledWith(input);
expect(mode.showError).not.toHaveBeenCalled();
});
});