From c8f1972c8ca3485be32ce1b52f6d77d361442e36 Mon Sep 17 00:00:00 2001 From: Christian Stewart Date: Sat, 18 Jul 2026 00:31:11 -0700 Subject: [PATCH] fix(coding-agent): drain advisor reviews in print mode --- docs/advisor-watchdog.md | 11 ++ packages/coding-agent/CHANGELOG.md | 4 + .../src/advisor/__tests__/advisor.test.ts | 97 +++++++++++++ .../coding-agent/src/advisor/advise-tool.ts | 4 + packages/coding-agent/src/advisor/runtime.ts | 36 +++-- .../src/modes/noninteractive-dispose.test.ts | 12 +- packages/coding-agent/src/modes/print-mode.ts | 17 ++- .../coding-agent/src/session/agent-session.ts | 59 +++++++- .../agent-session-advisor-suppression.test.ts | 76 +++++++++++ .../test/print-mode-working-indicator.test.ts | 128 +++++++++++++++++- .../test/silent-abort-print-mode.test.ts | 2 + 11 files changed, 424 insertions(+), 22 deletions(-) diff --git a/docs/advisor-watchdog.md b/docs/advisor-watchdog.md index 65870bf11..431deb796 100644 --- a/docs/advisor-watchdog.md +++ b/docs/advisor-watchdog.md @@ -38,6 +38,17 @@ advisor: The advisor role uses normal model-role resolution, including provider-prefixed ids, canonical ids, and optional thinking suffixes. +### Headless runs + +Use `--advisor` to enable the advisor for one print-mode process without +persisting `advisor.enabled`: + +```sh +omp -p --advisor "Review this task." +``` + +While a primary prompt is running, advisor concerns and blockers continue to steer that live turn. After the final prompt settles, print mode preserves late advisor notes without starting hidden primary turns, then waits up to ten minutes for final reviews before disposing the session. Error exits use a 30-second drain budget so failed automation can terminate. If either deadline expires, OMP logs the reviews that disposal will abandon; completed reviews retain their transcript and token/cost usage. + Slash commands: | Command | Effect | diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 4522715df..9a360dc20 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed headless print mode disposing the session before a final advisor review completed, which could drop the advisor transcript and usage ([#5942](https://github.com/can1357/oh-my-pi/pull/5942)). + ## [17.0.4] - 2026-07-18 ### Fixed diff --git a/packages/coding-agent/src/advisor/__tests__/advisor.test.ts b/packages/coding-agent/src/advisor/__tests__/advisor.test.ts index 9e5eec54e..dd9689bc9 100644 --- a/packages/coding-agent/src/advisor/__tests__/advisor.test.ts +++ b/packages/coding-agent/src/advisor/__tests__/advisor.test.ts @@ -865,6 +865,65 @@ describe("advisor", () => { expect(promptInputs[1]).toContain("second"); }); + it("waits for an in-flight review within the catch-up deadline", async () => { + const promptStarted = Promise.withResolvers(); + const releasePrompt = Promise.withResolvers(); + const messages: AgentMessage[] = [{ role: "user", content: "first", timestamp: 1 } as AgentMessage]; + const agent: AdvisorAgent = { + prompt: async () => { + promptStarted.resolve(); + await releasePrompt.promise; + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const runtime = new AdvisorRuntime(agent, { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }); + + runtime.onTurnEnd(); + await promptStarted.promise; + let settled = false; + const catchup = runtime.waitForCatchup(1000, 1).then(caughtUp => { + settled = true; + return caughtUp; + }); + await Promise.resolve(); + expect(settled).toBe(false); + + releasePrompt.resolve(); + expect(await catchup).toBe(true); + }); + + it("reports an in-flight review that exceeds the catch-up deadline", async () => { + const promptStarted = Promise.withResolvers(); + const releasePrompt = Promise.withResolvers(); + const messages: AgentMessage[] = [{ role: "user", content: "first", timestamp: 1 } as AgentMessage]; + const agent: AdvisorAgent = { + prompt: async () => { + promptStarted.resolve(); + await releasePrompt.promise; + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const runtime = new AdvisorRuntime(agent, { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }); + + runtime.onTurnEnd(); + await promptStarted.promise; + expect(await runtime.waitForCatchup(20, 1)).toBe(false); + expect(runtime.backlog).toBe(1); + + releasePrompt.resolve(); + await settleUntil(() => runtime.backlog === 0); + }); + it("preserves the next user turn when an accepted empty stop is pruned", async () => { const promptInputs: string[] = []; const agent = makeAgent(promptInputs); @@ -3779,6 +3838,44 @@ describe("advisor", () => { // or it strands and #drainStrandedQueuedMessages auto-resumes it. Do not swap // the call site back to session `isStreaming`. describe("resolveAdvisorDeliveryChannel", () => { + it("preserves every severity when a headless drain forbids primary turns", () => { + for (const severity of [undefined, "nit", "concern", "blocker"] as const) { + expect( + resolveAdvisorDeliveryChannel({ + severity, + autoResumeSuppressed: false, + streaming: false, + aborting: false, + terminalAnswerNoQueuedWork: true, + preserveOnly: true, + }), + ).toBe("preserve"); + } + }); + + it("keeps live headless advice on normal delivery channels until the primary finishes", () => { + expect( + resolveAdvisorDeliveryChannel({ + severity: "nit", + autoResumeSuppressed: false, + streaming: true, + aborting: false, + preserveOnly: true, + }), + ).toBe("aside"); + for (const severity of ["concern", "blocker"] as const) { + expect( + resolveAdvisorDeliveryChannel({ + severity, + autoResumeSuppressed: false, + streaming: true, + aborting: false, + preserveOnly: true, + }), + ).toBe("steer"); + } + }); + it("routes a non-interrupting nit to the aside queue regardless of state", () => { expect( resolveAdvisorDeliveryChannel({ diff --git a/packages/coding-agent/src/advisor/advise-tool.ts b/packages/coding-agent/src/advisor/advise-tool.ts index 73f9cde4f..df46b8398 100644 --- a/packages/coding-agent/src/advisor/advise-tool.ts +++ b/packages/coding-agent/src/advisor/advise-tool.ts @@ -101,6 +101,8 @@ export function isAdvisorInterruptImmuneTurnActive(opts: { /** * Decide how one advisor note reaches the primary agent. * + * - A `preserveOnly` caller records every note that arrives while the primary + * is idle as a visible card and never starts a new primary turn. * - A non-interrupting `nit` always rides the non-interrupting aside queue. * - An interrupting `concern`/`blocker` is normally steered into the agent: into * the live turn while one is streaming, or (when idle) a triggered turn so the @@ -129,7 +131,9 @@ export function resolveAdvisorDeliveryChannel(opts: { aborting: boolean; terminalAnswerNoQueuedWork?: boolean; interruptImmuneTurnActive?: boolean; + preserveOnly?: boolean; }): AdvisorDeliveryChannel { + if (opts.preserveOnly && !opts.streaming) return "preserve"; if (!isInterruptingSeverity(opts.severity)) return "aside"; if (opts.autoResumeSuppressed && (opts.aborting || !opts.streaming)) return "preserve"; if (opts.terminalAnswerNoQueuedWork && opts.severity !== "blocker" && !opts.streaming && !opts.aborting) diff --git a/packages/coding-agent/src/advisor/runtime.ts b/packages/coding-agent/src/advisor/runtime.ts index 547bc0248..b0772ab3f 100644 --- a/packages/coding-agent/src/advisor/runtime.ts +++ b/packages/coding-agent/src/advisor/runtime.ts @@ -219,8 +219,7 @@ interface PendingDelta { interface CatchupWaiter { threshold: number; - resolve: () => void; - finish: () => void; + finish: (caughtUp: boolean) => void; timer?: NodeJS.Timeout; } @@ -366,7 +365,13 @@ export class AdvisorRuntime { } } - waitForCatchup(maxMs: number, threshold: number, signal?: AbortSignal): Promise { + /** + * Wait until the advisor backlog falls below `threshold`. + * + * Returns `false` when the deadline, abort signal, or a runtime failure releases + * the waiter before the requested backlog was drained. + */ + waitForCatchup(maxMs: number, threshold: number, signal?: AbortSignal): Promise { if ( this.disposed || signal?.aborted || @@ -378,21 +383,26 @@ export class AdvisorRuntime { // primary would otherwise park for the full catch-up budget. this.#failing ) - return Promise.resolve(); - const { promise, resolve } = Promise.withResolvers(); + return Promise.resolve(this.#backlog < threshold); + const { promise, resolve } = Promise.withResolvers(); let waiter!: CatchupWaiter; - const finish = (): void => { + const finish = (caughtUp: boolean): void => { const idx = this.#waiters.indexOf(waiter); if (idx >= 0) this.#waiters.splice(idx, 1); clearTimeout(waiter.timer); - signal?.removeEventListener("abort", finish); - resolve(); + signal?.removeEventListener("abort", abort); + resolve(caughtUp); + }; + const abort = (): void => finish(false); + waiter = { + threshold, + finish, + timer: setTimeout(abort, maxMs), }; - waiter = { threshold, resolve, finish, timer: setTimeout(finish, maxMs) }; this.#waiters.push(waiter); - signal?.addEventListener("abort", finish, { once: true }); + signal?.addEventListener("abort", abort, { once: true }); if (signal?.aborted) { - finish(); + abort(); } return promise; } @@ -571,14 +581,14 @@ export class AdvisorRuntime { for (let i = this.#waiters.length - 1; i >= 0; i--) { const w = this.#waiters[i]; if (this.#backlog < w.threshold) { - w.finish(); + w.finish(true); } } } #wakeAllWaiters(): void { for (const w of [...this.#waiters]) { - w.finish(); + w.finish(false); } } diff --git a/packages/coding-agent/src/modes/noninteractive-dispose.test.ts b/packages/coding-agent/src/modes/noninteractive-dispose.test.ts index 59e3117e8..c38306e29 100644 --- a/packages/coding-agent/src/modes/noninteractive-dispose.test.ts +++ b/packages/coding-agent/src/modes/noninteractive-dispose.test.ts @@ -8,6 +8,7 @@ import { describe, expect, it, spyOn } from "bun:test"; import type { AssistantMessage } from "@oh-my-pi/pi-ai"; import type { AgentSession } from "../session/agent-session"; +import * as telemetryExport from "../telemetry-export"; import { runPrintMode } from "./print-mode"; /** Stand-in for `process.exit`: it terminates, so nothing after it should run. */ @@ -35,11 +36,19 @@ describe("print-mode error exit disposes the session before exit", () => { extensionRunner: undefined, subscribe: () => {}, state: { messages: [errorMsg] }, + prepareForHeadlessAdvisorDrain: () => {}, + waitForAdvisorCatchup: async () => { + order.push("catchup"); + return true; + }, dispose: async () => { order.push("dispose"); }, } as unknown as AgentSession; + const flushSpy = spyOn(telemetryExport, "flushTelemetryExport").mockImplementation(async () => { + order.push("flush"); + }); const exitSpy = spyOn(process, "exit").mockImplementation(((code: number) => { order.push("exit"); throw new ProcessExit(code); @@ -53,8 +62,9 @@ describe("print-mode error exit disposes the session before exit", () => { } finally { exitSpy.mockRestore(); stderrSpy.mockRestore(); + flushSpy.mockRestore(); } - expect(order).toEqual(["dispose", "exit"]); + expect(order).toEqual(["catchup", "flush", "dispose", "exit"]); }); }); diff --git a/packages/coding-agent/src/modes/print-mode.ts b/packages/coding-agent/src/modes/print-mode.ts index 158dc4cba..a5c30a409 100644 --- a/packages/coding-agent/src/modes/print-mode.ts +++ b/packages/coding-agent/src/modes/print-mode.ts @@ -29,6 +29,11 @@ export interface PrintModeOptions { printThoughts?: boolean; } +/** Matches the longest built-in provider request deadline while bounding tool-loop stalls. */ +export const PRINT_MODE_ADVISOR_DRAIN_TIMEOUT_MS = 10 * 60_000; +/** Error exits cannot hold automation for the full normal drain budget. */ +export const PRINT_MODE_ERROR_ADVISOR_DRAIN_TIMEOUT_MS = 30_000; + /** Drop the provider-opaque replay payload (e.g. encrypted reasoning items) before printing. */ function stripProviderPayload(message: T): T { if (!("providerPayload" in message) || message.providerPayload === undefined) return message; @@ -130,6 +135,10 @@ export async function runPrintMode(session: AgentSession, options: PrintModeOpti await logger.time("print:prompt:next", () => session.prompt(message)); } + // From this point onward a late blocker must be recorded without starting a + // primary turn whose response print mode would never emit. + session.prepareForHeadlessAdvisorDrain(); + // In text mode, output final response if (mode === "text") { const state = session.state; @@ -151,6 +160,7 @@ export async function runPrintMode(session: AgentSession, options: PrintModeOpti // `dispose()` (releaseTabsForOwner) actually runs — otherwise an // OMP-owned Chromium survives this exit (issue #5643). `dispose()` // is idempotent, so the unreachable call below is a harmless no-op. + await session.waitForAdvisorCatchup(PRINT_MODE_ERROR_ADVISOR_DRAIN_TIMEOUT_MS); await flushTelemetryExport(); await session.dispose({ mnemopiConsolidateTimeoutMs: SHUTDOWN_CONSOLIDATE_BUDGET_MS }); const flushed = process.stderr.write(`${errorLine}\n`); @@ -180,14 +190,15 @@ export async function runPrintMode(session: AgentSession, options: PrintModeOpti } } - // Ensure stdout is fully flushed before returning - // This prevents race conditions where the process exits before all output is written + await session.waitForAdvisorCatchup(PRINT_MODE_ADVISOR_DRAIN_TIMEOUT_MS); + + // Ensure stdout, including late JSON advisor events, is fully flushed before returning. + // This prevents race conditions where the process exits before all output is written. await new Promise((resolve, reject) => { process.stdout.write("", err => { if (err) reject(err); else resolve(); }); }); - await session.dispose({ mnemopiConsolidateTimeoutMs: SHUTDOWN_CONSOLIDATE_BUDGET_MS }); } diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index aeb4a2e49..7f09e83a8 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1896,6 +1896,8 @@ export class AgentSession { * suppresses advisor concern/blocker auto-resume until the user next resumes. * Advisor advice is still recorded into the transcript, just not auto-run. */ #advisorAutoResumeSuppressed = false; + /** Print-mode sessions preserve advisor notes without starting hidden primary turns. */ + #preserveAdvisorAdvice = false; #advisorPrimaryTurnsCompleted = 0; #advisorInterruptImmuneTurnStart: number | undefined; #planModeState: PlanModeState | undefined; @@ -2034,6 +2036,8 @@ export class AgentSession { #turnIndex = 0; #messageEndPersistenceTail: Promise = Promise.resolve(); #pendingMessageEndPersistence = new Map>(); + /** Async lifecycle handlers for visible advisor cards emitted outside the primary loop. */ + #pendingAdvisorCardEvents = new Set>(); #persistedMessageKeys: { anchor: string; keys: Set } | undefined; #skills: Skill[]; @@ -3364,6 +3368,7 @@ export class AgentSession { const channel = resolveAdvisorDeliveryChannel({ severity, autoResumeSuppressed: this.#advisorAutoResumeSuppressed, + preserveOnly: this.#preserveAdvisorAdvice, // Key on the live agent-core loop, not session `isStreaming` (which also // counts `#promptInFlightCount` during post-turn unwind). Only a running // loop consumes a steer at its next boundary. @@ -4204,7 +4209,12 @@ export class AgentSession { * everything it schedules — settles. */ #handleAgentEvent = async (event: AgentEvent): Promise => { if (event.type !== "agent_end") { - return this.#processAgentEvent(event); + const processing = this.#processAgentEvent(event); + if ((event.type === "message_start" || event.type === "message_end") && isAdvisorCard(event.message)) { + this.#pendingAdvisorCardEvents.add(processing); + void processing.finally(() => this.#pendingAdvisorCardEvents.delete(processing)).catch(() => {}); + } + return processing; } const { promise, resolve } = Promise.withResolvers(); this.#trackPostPromptTask(promise); @@ -6944,6 +6954,53 @@ export class AgentSession { await this.agent.waitForIdle(); await this.#waitForPostPromptRecovery(); } + /** + * Prevent advisor notes from starting hidden primary turns while a headless + * caller prints and drains the final primary response. + */ + prepareForHeadlessAdvisorDrain(): void { + this.#preserveAdvisorAdvice = true; + } + + async #waitForPendingAdvisorCardEvents(timeoutMs: number): Promise { + const deadline = Date.now() + Math.max(0, timeoutMs); + while (this.#pendingAdvisorCardEvents.size > 0) { + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) return false; + const settled = Promise.allSettled([...this.#pendingAdvisorCardEvents]).then(() => true as const); + const { promise: timedOut, resolve } = Promise.withResolvers(); + const timer = setTimeout(() => resolve(false), remainingMs); + try { + if (!(await Promise.race([settled, timedOut]))) return false; + } finally { + clearTimeout(timer); + } + } + return true; + } + + /** + * Wait for active advisor reviews and their emitted card events before a + * headless caller disposes the session. Returns `false` and logs work disposal + * will abandon when the shared deadline expires or an advisor fails. + */ + async waitForAdvisorCatchup(timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + const results = await Promise.all(this.#advisors.map(advisor => advisor.runtime.waitForCatchup(timeoutMs, 1))); + const cardEventsCaughtUp = await this.#waitForPendingAdvisorCardEvents(Math.max(0, deadline - Date.now())); + const abandoned = this.#advisors.filter( + (advisor, index) => results[index] === false && advisor.runtime.backlog > 0, + ); + if (abandoned.length > 0 || !cardEventsCaughtUp) { + logger.warn("advisor shutdown drain incomplete; disposal will abandon reviews or cards", { + timeoutMs, + advisors: abandoned.map(advisor => ({ name: advisor.name, backlog: advisor.runtime.backlog })), + pendingAdvisorCards: this.#pendingAdvisorCardEvents.size, + }); + return false; + } + return true; + } async drainAsyncJobDeliveriesForAcp(options?: { timeoutMs?: number }): Promise { const manager = this.#asyncJobManager; diff --git a/packages/coding-agent/test/agent-session-advisor-suppression.test.ts b/packages/coding-agent/test/agent-session-advisor-suppression.test.ts index d8cc5dfb6..ad6540722 100644 --- a/packages/coding-agent/test/agent-session-advisor-suppression.test.ts +++ b/packages/coding-agent/test/agent-session-advisor-suppression.test.ts @@ -61,6 +61,12 @@ interface CompletedAdvisorHarness { advisorMock: MockModel; } +interface AdvisorTestExtensionRunner { + hasHandlers(eventType: string): boolean; + emitBeforeAgentStart(): Promise; + emit(event: { type: string; message?: AgentMessage }): Promise; +} + describe("AgentSession advisor auto-resume suppression", () => { let tempDir: TempDir; let session: AgentSession; @@ -166,6 +172,7 @@ describe("AgentSession advisor auto-resume suppression", () => { async function createCompletedAdvisorSession( severity: "concern" | "blocker" = "concern", + extensionRunner?: AdvisorTestExtensionRunner, ): Promise { const model = getBundledModel("anthropic", "claude-sonnet-4-5")!; const mock = createMockModel({ @@ -206,6 +213,7 @@ describe("AgentSession advisor auto-resume suppression", () => { modelRegistry, advisorTools: [], advisorStreamFn: advisorMock.stream, + extensionRunner: extensionRunner as never, }); return { session, sessionManager, mock, advisorMock }; } @@ -270,6 +278,74 @@ describe("AgentSession advisor auto-resume suppression", () => { expect(mock.calls.length).toBe(1); }); + it("waits for preserved advisor card hooks and persistence before reporting catch-up", async () => { + const hookStarted = Promise.withResolvers(); + const releaseHook = Promise.withResolvers(); + const extensionRunner: AdvisorTestExtensionRunner = { + hasHandlers: eventType => eventType === "message_end", + emitBeforeAgentStart: async () => undefined, + emit: async event => { + if (event.type !== "message_end" || !event.message || !isAdvisorCard(event.message)) return; + hookStarted.resolve(); + await releaseHook.promise; + }, + }; + const { session, sessionManager, mock } = await createCompletedAdvisorSession("concern", extensionRunner); + const persisted = capturePersistedAdvice(sessionManager); + + expect(session.setAdvisorEnabled(true)).toBe(true); + await session.prompt("answer with exactly one line"); + await hookStarted.promise; + + expect(await session.waitForAdvisorCatchup(0)).toBe(false); + expect(persisted).toEqual([]); + + let catchupSettled = false; + const catchup = session.waitForAdvisorCatchup(1000).then(caughtUp => { + catchupSettled = true; + return caughtUp; + }); + await Promise.resolve(); + expect(catchupSettled).toBe(false); + expect(persisted).toEqual([]); + + releaseHook.resolve(); + expect(await catchup).toBe(true); + expect(persisted.at(-1)).toContain("Fixture verdict confirmed"); + expect(mock.calls).toHaveLength(1); + }); + + it("waits for preserved advisor card start hooks before reporting catch-up", async () => { + const hookStarted = Promise.withResolvers(); + const releaseHook = Promise.withResolvers(); + const extensionRunner: AdvisorTestExtensionRunner = { + hasHandlers: eventType => eventType === "message_start", + emitBeforeAgentStart: async () => undefined, + emit: async event => { + if (event.type !== "message_start" || !event.message || !isAdvisorCard(event.message)) return; + hookStarted.resolve(); + await releaseHook.promise; + }, + }; + const { session, mock } = await createCompletedAdvisorSession("concern", extensionRunner); + + expect(session.setAdvisorEnabled(true)).toBe(true); + await session.prompt("answer with exactly one line"); + await hookStarted.promise; + + let catchupSettled = false; + const catchup = session.waitForAdvisorCatchup(1000).then(caughtUp => { + catchupSettled = true; + return caughtUp; + }); + await Promise.resolve(); + expect(catchupSettled).toBe(false); + + releaseHook.resolve(); + expect(await catchup).toBe(true); + expect(mock.calls).toHaveLength(1); + }); + it("steers a late advisor blocker after a terminal answer so the primary corrects it", async () => { const { session, mock } = await createCompletedAdvisorSession("blocker"); diff --git a/packages/coding-agent/test/print-mode-working-indicator.test.ts b/packages/coding-agent/test/print-mode-working-indicator.test.ts index 8a824bf28..9d8f61915 100644 --- a/packages/coding-agent/test/print-mode-working-indicator.test.ts +++ b/packages/coding-agent/test/print-mode-working-indicator.test.ts @@ -1,7 +1,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test"; import type { AssistantMessage } from "@oh-my-pi/pi-ai"; -import { runPrintMode } from "@oh-my-pi/pi-coding-agent/modes/print-mode"; -import type { AgentSession } from "@oh-my-pi/pi-coding-agent/session/agent-session"; +import { + PRINT_MODE_ADVISOR_DRAIN_TIMEOUT_MS, + PRINT_MODE_ERROR_ADVISOR_DRAIN_TIMEOUT_MS, + runPrintMode, +} from "@oh-my-pi/pi-coding-agent/modes/print-mode"; +import type { AgentSession, AgentSessionEvent } from "@oh-my-pi/pi-coding-agent/session/agent-session"; function makeAssistantMessage(text: string): AssistantMessage { const timestamp = Date.now(); @@ -35,6 +39,7 @@ function createDelayedSession(finalMessage: AssistantMessage): DelayedSession { const messages: AssistantMessage[] = []; const { promise: promptStarted, resolve: markPromptStarted } = Promise.withResolvers(); const { promise: promptReleased, resolve: resolvePrompt } = Promise.withResolvers(); + let advisorDrainPrepared = false; const session = { state: { messages }, @@ -44,11 +49,18 @@ function createDelayedSession(finalMessage: AssistantMessage): DelayedSession { extensionRunner: undefined, subscribe: () => () => {}, prompt: async () => { + if (advisorDrainPrepared) throw new Error("headless advisor delivery armed before prompt completion"); markPromptStarted(); await promptReleased; messages.push(finalMessage); return true; }, + prepareForHeadlessAdvisorDrain: () => { + advisorDrainPrepared = true; + }, + waitForAdvisorCatchup: async () => { + if (!advisorDrainPrepared) throw new Error("advisor catch-up started before headless delivery was armed"); + }, dispose: async () => {}, } as unknown as AgentSession; @@ -58,19 +70,27 @@ function createDelayedSession(finalMessage: AssistantMessage): DelayedSession { describe("print mode working indicator", () => { let stderrOutput: string[]; let stdoutOutput: string[]; + let stdoutEvents: Array<"write" | "flush">; beforeEach(() => { stderrOutput = []; stdoutOutput = []; + stdoutEvents = []; vi.spyOn(process.stderr, "write").mockImplementation((chunk: unknown) => { stderrOutput.push(String(chunk)); return true; }); vi.spyOn(process.stdout, "write").mockImplementation((...args: unknown[]) => { const chunk = args[0]; - if (typeof chunk === "string") stdoutOutput.push(chunk); + if (typeof chunk === "string") { + stdoutOutput.push(chunk); + if (chunk.length > 0) stdoutEvents.push("write"); + } const last = args[args.length - 1]; - if (typeof last === "function") last(); + if (typeof last === "function") { + stdoutEvents.push("flush"); + last(); + } return true; }); }); @@ -122,4 +142,104 @@ describe("print mode working indicator", () => { expect(stderrOutput.join("")).toBe("Working...\n"); }); + + it("flushes late JSON advisor events after catch-up before disposing", async () => { + const message = makeAssistantMessage("advisor-aware answer"); + const messages: AssistantMessage[] = []; + const { promise: catchup, resolve: resolveCatchup } = Promise.withResolvers(); + const { promise: catchupStarted, resolve: markCatchupStarted } = Promise.withResolvers(); + let disposed = false; + let catchupTimeoutMs: number | undefined; + let subscriber: ((event: AgentSessionEvent) => void) | undefined; + const session = { + state: { messages }, + sessionManager: { getHeader: () => undefined }, + extensionRunner: undefined, + subscribe: (listener: (event: AgentSessionEvent) => void) => { + subscriber = listener; + return () => {}; + }, + prompt: async () => { + messages.push(message); + return true; + }, + prepareForHeadlessAdvisorDrain: () => {}, + waitForAdvisorCatchup: async (timeoutMs: number) => { + catchupTimeoutMs = timeoutMs; + markCatchupStarted(); + await catchup; + subscriber?.({ + type: "message_end", + message: { + role: "custom", + customType: "advisor", + content: "late advisor review", + display: true, + attribution: "agent", + timestamp: Date.now(), + }, + }); + }, + dispose: async () => { + disposed = true; + }, + } as unknown as AgentSession; + + const run = runPrintMode(session, { mode: "json", initialMessage: "hello" }); + await catchupStarted; + expect(disposed).toBe(false); + resolveCatchup(); + await run; + + expect(disposed).toBe(true); + expect(catchupTimeoutMs).toBe(PRINT_MODE_ADVISOR_DRAIN_TIMEOUT_MS); + expect(stdoutOutput.join("")).toContain("late advisor review"); + expect(stdoutEvents.at(-1)).toBe("flush"); + }); + + it("waits for advisor catch-up before hard-exit disposal", async () => { + const message = makeAssistantMessage(""); + message.stopReason = "error"; + message.errorMessage = "primary request failed"; + const messages: AssistantMessage[] = []; + const { promise: catchup, resolve: resolveCatchup } = Promise.withResolvers(); + const { promise: catchupStarted, resolve: markCatchupStarted } = Promise.withResolvers(); + let disposed = false; + let exitCode: number | undefined; + let catchupTimeoutMs: number | undefined; + vi.spyOn(process, "exit").mockImplementation(code => { + exitCode = code as number; + throw new Error("process exit"); + }); + const session = { + state: { messages }, + sessionManager: { getHeader: () => undefined }, + extensionRunner: undefined, + subscribe: () => () => {}, + prompt: async () => { + messages.push(message); + return true; + }, + prepareForHeadlessAdvisorDrain: () => {}, + waitForAdvisorCatchup: async (timeoutMs: number) => { + catchupTimeoutMs = timeoutMs; + markCatchupStarted(); + await catchup; + }, + dispose: async () => { + disposed = true; + }, + } as unknown as AgentSession; + + const run = runPrintMode(session, { mode: "text", initialMessage: "hello" }); + await catchupStarted; + expect(disposed).toBe(false); + resolveCatchup(); + + await expect(run).rejects.toThrow("process exit"); + expect(disposed).toBe(true); + expect(exitCode).toBe(1); + expect(catchupTimeoutMs).toBe(PRINT_MODE_ERROR_ADVISOR_DRAIN_TIMEOUT_MS); + expect(stderrOutput.join("")).toContain("primary request failed"); + }); }); diff --git a/packages/coding-agent/test/silent-abort-print-mode.test.ts b/packages/coding-agent/test/silent-abort-print-mode.test.ts index 2a876872f..9ef5b18df 100644 --- a/packages/coding-agent/test/silent-abort-print-mode.test.ts +++ b/packages/coding-agent/test/silent-abort-print-mode.test.ts @@ -50,6 +50,8 @@ function createMockSession( extensionRunner: undefined, subscribe: () => () => {}, prompt: async () => {}, + prepareForHeadlessAdvisorDrain: () => {}, + waitForAdvisorCatchup: async () => true, dispose, } as unknown as AgentSession; }