From 34746ceaf53e3344cf911e9be0ce8706fc73647d Mon Sep 17 00:00:00 2001 From: metaphorics <152830360+metaphorics@users.noreply.github.com> Date: Sun, 14 Jun 2026 07:37:14 +0900 Subject: [PATCH] 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. --- packages/coding-agent/CHANGELOG.md | 4 ++ packages/coding-agent/src/main.ts | 20 +++++-- .../src/modes/interactive-mode.ts | 9 +++ .../test/interactive-mode-plan-review.test.ts | 46 +++++++++++++++- .../test/main-interactive-input.test.ts | 55 +++++++++++++++++++ 5 files changed, 128 insertions(+), 6 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 7fc96cb5e..0e9c40fc0 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -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 diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index e22741244..8738fee2c 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -264,7 +264,7 @@ export async function submitInteractiveInput( InteractiveMode, "markPendingSubmissionStarted" | "finishPendingSubmission" | "showError" | "checkShutdownRequested" >, - session: Pick, + session: Pick, input: SubmittedUserInput, ): Promise { 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"; diff --git a/packages/coding-agent/src/modes/interactive-mode.ts b/packages/coding-agent/src/modes/interactive-mode.ts index 221787c27..854028e10 100644 --- a/packages/coding-agent/src/modes/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive-mode.ts @@ -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 }); } diff --git a/packages/coding-agent/test/interactive-mode-plan-review.test.ts b/packages/coding-agent/test/interactive-mode-plan-review.test.ts index 1f5db2e72..0c885a29f 100644 --- a/packages/coding-agent/test/interactive-mode-plan-review.test.ts +++ b/packages/coding-agent/test/interactive-mode-plan-review.test.ts @@ -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, { diff --git a/packages/coding-agent/test/main-interactive-input.test.ts b/packages/coding-agent/test/main-interactive-input.test.ts index 7f32d7eab..a0f0384d9 100644 --- a/packages/coding-agent/test/main-interactive-input.test.ts +++ b/packages/coding-agent/test/main-interactive-input.test.ts @@ -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(); + }); });