From e699f07a17dad19b8e24ae6956588089ffd2edef Mon Sep 17 00:00:00 2001 From: Byeonghoon Yoo Date: Thu, 23 Jul 2026 05:33:04 +0900 Subject: [PATCH] fix(omp): complete delayed model turns after local-only results (#2282) Wait briefly for extension-queued model turns before completing local-only OMP prompts, and preserve autonomous turn completion. --- .../server/agent/providers/omp/agent.test.ts | 112 ++++++++++++++- .../src/server/agent/providers/omp/agent.ts | 92 ++++++++++-- .../providers/omp/test-utils/omp-harness.ts | 132 +++++++++++++++++- 3 files changed, 320 insertions(+), 16 deletions(-) diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index 69b2d8ff3..fe2fc3749 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "vitest"; import type { PaseoToolCatalog } from "../../tools/types.js"; -import type { OmpProviderIdleScheduler } from "./agent.js"; +import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js"; import { OmpHarness } from "./test-utils/omp-harness.js"; class ManualIdleScheduler implements OmpProviderIdleScheduler { @@ -30,6 +30,41 @@ class ManualIdleScheduler implements OmpProviderIdleScheduler { } } +class ManualNoTurnScheduler implements OmpNoTurnScheduler { + private settleResolve: (() => void) | null = null; + private aborted = false; + + waitForSettle(signal: AbortSignal): Promise { + if (signal.aborted) { + this.aborted = true; + return Promise.resolve(); + } + return new Promise((resolve) => { + this.settleResolve = resolve; + signal.addEventListener( + "abort", + () => { + this.aborted = true; + this.settleResolve = null; + resolve(); + }, + { once: true }, + ); + }); + } + + settle(): void { + const resolve = this.settleResolve; + if (!resolve) throw new Error("OMP has not requested a no-turn settle wait"); + this.settleResolve = null; + resolve(); + } + + wasAborted(): boolean { + return this.aborted; + } +} + function createToolCatalog(): PaseoToolCatalog { return { tools: new Map([ @@ -235,6 +270,81 @@ describe("OMP agent client and session", () => { expect(omp.completedTurnCount()).toBe(1); }); + test("completes a local-only prompt when no OMP turn begins", async () => { + const omp = new OmpHarness(); + await omp.start(); + + await expect(omp.runPromptWithoutTurn("/model")).resolves.toMatchObject({ finalText: "" }); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("waits for a delayed queued model turn after OMP's local-only result", async () => { + const omp = new OmpHarness(); + await omp.start(); + + const completion = await omp.runPromptAfterDelayedFalseLocalOnlyResult( + "hello OMP", + "delayed queued model turn completed", + ); + + expect(completion.completedBeforeTurn).toBe(false); + expect(completion.result).toMatchObject({ finalText: "delayed queued model turn completed" }); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("completes an async local-only result after the settle window", async () => { + const scheduler = new ManualNoTurnScheduler(); + const omp = new OmpHarness({ noTurnScheduler: scheduler }); + await omp.start(); + const prompt = await omp.startPromptWithFalseLocalOnlyResult("local-only"); + + expect(prompt.completed()).toBe(false); + scheduler.settle(); + await expect(prompt.completion).resolves.toMatchObject({ finalText: "" }); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("cancels an async local-only settle when the OMP session closes", async () => { + const scheduler = new ManualNoTurnScheduler(); + const omp = new OmpHarness({ noTurnScheduler: scheduler }); + await omp.start(); + const prompt = await omp.startPromptWithFalseLocalOnlyResult("local-only"); + + await omp.close(); + + expect(scheduler.wasAborted()).toBe(true); + expect(prompt.completed()).toBe(false); + expect(omp.completedTurnCount()).toBe(0); + }); + + test("preserves a correlated invoked result over a local-only prompt ack", async () => { + const omp = new OmpHarness(); + await omp.start(); + + const completion = await omp.runPromptAfterCorrelatedTrueResult( + "hello OMP", + "correlated model turn completed", + ); + + expect(completion.completedBeforeTurn).toBe(false); + expect(completion.result).toMatchObject({ finalText: "correlated model turn completed" }); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("completes an autonomous OMP turn without a foreground turn ID", async () => { + const omp = new OmpHarness(); + await omp.start(); + + await omp.runAutonomousTurn("autonomous turn completed"); + + expect(omp.completedTurnCount()).toBe(1); + expect(omp.timeline()).toContainEqual({ + type: "assistant_message", + text: "autonomous turn completed", + messageId: "omp-assistant-1", + }); + }); + test("resumes an OMP session and replays its history", async () => { const omp = new OmpHarness(); await omp.resume( diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 63e9f23d5..eec64cee4 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; import { homedir } from "node:os"; +import { setImmediate as waitForImmediate, setTimeout as delay } from "node:timers/promises"; import type { Logger } from "pino"; import stripAnsi from "strip-ansi"; @@ -148,12 +149,23 @@ export interface OmpAgentClientOptions { runtime?: OmpRuntime; subagentCardScheduler?: OmpSubagentCardScheduler; providerIdleScheduler?: OmpProviderIdleScheduler; + noTurnScheduler?: OmpNoTurnScheduler; } export interface OmpProviderIdleScheduler { waitForRetry(): Promise; } +export interface OmpNoTurnScheduler { + waitForSettle(signal: AbortSignal): Promise; +} + +// COMPAT(ompDelayedLocalOnlyResult): OMP 17.0.5 can report a regular prompt as +// local-only shortly before an extension-queued model turn starts. Added in +// v0.2.0-beta.1; remove after January 20, 2027 once the minimum OMP version +// guarantees prompt_result waits for queued extension work. +const OMP_NO_TURN_SETTLE_MS = 5_000; + interface OmpPromptPayload { text: string; images?: OmpImageContent[]; @@ -184,6 +196,7 @@ interface OmpAgentSessionOptions { logger: Logger; subagentCardScheduler?: OmpSubagentCardScheduler; providerIdleScheduler?: OmpProviderIdleScheduler; + noTurnScheduler?: OmpNoTurnScheduler; paseoTools?: PaseoToolCatalog; /** * When false (resumed sessions), replayed session events are dropped until @@ -201,6 +214,14 @@ function createOmpProviderIdleScheduler(): OmpProviderIdleScheduler { }; } +function createOmpNoTurnScheduler(): OmpNoTurnScheduler { + return { + waitForSettle: async (signal) => { + await delay(OMP_NO_TURN_SETTLE_MS, undefined, { signal }); + }, + }; +} + interface OmpResumeConfig { cwd: string; model?: string; @@ -905,7 +926,9 @@ export class OmpAgentSession implements AgentSession { private activeNoTurnPromptText: string | null = null; private readonly pendingNoTurnOutputs: Array<{ turnId: string; message: string }> = []; private activePromptRequestId: string | null = null; + private activePromptAgentInvoked: boolean | null = null; private readonly pendingPromptResults = new Map(); + private pendingNoTurnCompletionAbort: AbortController | null = null; private lastKnownThinkingOptionId: string | null; private outOfBandCompactionEmit: ((event: AgentStreamEvent) => void) | null = null; private outOfBandCompactionStarted = false; @@ -917,6 +940,7 @@ export class OmpAgentSession implements AgentSession { private state: OmpSessionState; private readonly currentModeId: string | null; private readonly providerIdleScheduler: OmpProviderIdleScheduler; + private readonly noTurnScheduler: OmpNoTurnScheduler; private closed = false; private live: boolean; private readonly emittedUserMessageIds = new Set(); @@ -930,6 +954,7 @@ export class OmpAgentSession implements AgentSession { this.paseoTools = options.paseoTools; this.live = options.live ?? true; this.providerIdleScheduler = options.providerIdleScheduler ?? createOmpProviderIdleScheduler(); + this.noTurnScheduler = options.noTurnScheduler ?? createOmpNoTurnScheduler(); this.subagentCardTracker = new OmpSubagentCardTracker({ scheduler: options.subagentCardScheduler, }); @@ -1005,8 +1030,12 @@ export class OmpAgentSession implements AgentSession { if (ack.requestId) { this.pendingPromptResults.delete(ack.requestId); } - const agentInvoked = correlatedResult ?? ack.agentInvoked; - if (agentInvoked === false) { + this.activePromptAgentInvoked = correlatedResult ?? ack.agentInvoked ?? null; + if (correlatedResult === false) { + this.scheduleNoTurnPromptCompletion(turnId); + return; + } + if (correlatedResult !== true && ack.agentInvoked === false) { await this.completeNoTurnPrompt(turnId); return; } @@ -1183,6 +1212,7 @@ export class OmpAgentSession implements AgentSession { return; } this.closed = true; + this.cancelNoTurnPromptCompletion(); try { await this.runtimeSession.close(); } finally { @@ -1309,11 +1339,40 @@ export class OmpAgentSession implements AgentSession { return this.activeTurnId ?? undefined; } + private scheduleNoTurnPromptCompletion(turnId: string): void { + this.cancelNoTurnPromptCompletion(); + const abort = new AbortController(); + this.pendingNoTurnCompletionAbort = abort; + void this.noTurnScheduler + .waitForSettle(abort.signal) + .then(async () => { + if (this.pendingNoTurnCompletionAbort !== abort) { + return undefined; + } + this.pendingNoTurnCompletionAbort = null; + return await this.completeNoTurnPrompt(turnId); + }) + .catch((error: unknown) => { + if (!abort.signal.aborted) { + this.logger.debug({ err: error }, "OMP local-only settle wait failed"); + } + }); + } + + private cancelNoTurnPromptCompletion(): void { + this.pendingNoTurnCompletionAbort?.abort(); + this.pendingNoTurnCompletionAbort = null; + } + private async completeNoTurnPrompt(turnId: string): Promise { - await new Promise((resolve) => { - setImmediate(resolve); - }); - if (this.activeTurnId !== turnId || this.activeTurnStarted || this.activeTurnHasUserMessage) { + await waitForImmediate(); + if ( + this.closed || + this.activeTurnId !== turnId || + this.activeTurnStarted || + this.activePromptAgentInvoked === true || + this.activeTurnHasUserMessage + ) { return; } this.emitBufferedNoTurnOutputs(turnId); @@ -1321,8 +1380,10 @@ export class OmpAgentSession implements AgentSession { } private clearNoTurnBuffers(): void { + this.cancelNoTurnPromptCompletion(); this.activeNoTurnPromptText = null; this.activePromptRequestId = null; + this.activePromptAgentInvoked = null; this.pendingNoTurnOutputs.splice(0, this.pendingNoTurnOutputs.length); } @@ -1736,12 +1797,13 @@ export class OmpAgentSession implements AgentSession { ? event.agentInvoked : undefined; if (requestId && agentInvoked !== undefined) { - if ( - requestId === this.activePromptRequestId && - agentInvoked === false && - this.activeTurnId - ) { - void this.completeNoTurnPrompt(this.activeTurnId); + if (requestId === this.activePromptRequestId && this.activeTurnId) { + this.activePromptAgentInvoked = agentInvoked; + if (agentInvoked === false) { + this.scheduleNoTurnPromptCompletion(this.activeTurnId); + } else { + this.cancelNoTurnPromptCompletion(); + } } else if (this.activePromptRequestId === null) { this.pendingPromptResults.set(requestId, agentInvoked); } @@ -2139,7 +2201,7 @@ export class OmpAgentSession implements AgentSession { turnId: string | undefined, messages: OmpAgentMessage[], ): Promise { - while (!this.closed && this.activeTurnId === turnId) { + while (!this.closed && this.activeTurnStarted && this.currentTurnIdForEvent() === turnId) { try { const state = await this.runtimeSession.getState(); this.state = state; @@ -2188,6 +2250,7 @@ export class OmpAgentClient implements AgentClient { private readonly modelRoleParams: OmpModelRoleParams; private readonly subagentCardScheduler?: OmpSubagentCardScheduler; private readonly providerIdleScheduler?: OmpProviderIdleScheduler; + private readonly noTurnScheduler?: OmpNoTurnScheduler; private readonly runtime: OmpRuntime; constructor(options: OmpAgentClientOptions) { @@ -2209,6 +2272,7 @@ export class OmpAgentClient implements AgentClient { this.modelRoleParams = modelRoleParams; this.subagentCardScheduler = options.subagentCardScheduler; this.providerIdleScheduler = options.providerIdleScheduler; + this.noTurnScheduler = options.noTurnScheduler; this.runtime = options.runtime ?? createRuntime(options.logger, runtimeSettings); } @@ -2249,6 +2313,7 @@ export class OmpAgentClient implements AgentClient { logger: this.logger, subagentCardScheduler: this.subagentCardScheduler, providerIdleScheduler: this.providerIdleScheduler, + noTurnScheduler: this.noTurnScheduler, paseoTools: launchContext?.paseoTools, }); } catch (error) { @@ -2289,6 +2354,7 @@ export class OmpAgentClient implements AgentClient { logger: this.logger, subagentCardScheduler: this.subagentCardScheduler, providerIdleScheduler: this.providerIdleScheduler, + noTurnScheduler: this.noTurnScheduler, paseoTools: launchContext?.paseoTools, live: false, }); diff --git a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts index a2fda872e..c7e3c905e 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts @@ -1,6 +1,7 @@ import { mkdtemp, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { setImmediate as waitForImmediate } from "node:timers/promises"; import pino from "pino"; import type { @@ -11,7 +12,12 @@ import type { AgentTimelineItem, } from "../../../agent-sdk-types.js"; import type { PaseoToolCatalog } from "../../../tools/types.js"; -import { OmpAgentClient, OmpAgentSession, type OmpProviderIdleScheduler } from "../agent.js"; +import { + OmpAgentClient, + OmpAgentSession, + type OmpNoTurnScheduler, + type OmpProviderIdleScheduler, +} from "../agent.js"; import type { OmpAgentMessage, OmpRpcSlashCommand } from "../rpc-types.js"; import { FakeOmp } from "./fake-omp.js"; @@ -59,11 +65,17 @@ export class OmpHarness { private readonly events: AgentStreamEvent[] = []; private session: OmpAgentSession | null = null; - constructor(options: { providerIdleScheduler?: OmpProviderIdleScheduler } = {}) { + constructor( + options: { + providerIdleScheduler?: OmpProviderIdleScheduler; + noTurnScheduler?: OmpNoTurnScheduler; + } = {}, + ) { this.client = new OmpAgentClient({ logger: pino({ level: "silent" }), runtime: this.omp, providerIdleScheduler: options.providerIdleScheduler, + noTurnScheduler: options.noTurnScheduler, }); } @@ -251,6 +263,121 @@ export class OmpHarness { return await run; } + async runPromptWithoutTurn(input: string): Promise { + const session = this.requireSession(); + this.omp.latestSession().promptAck = { agentInvoked: false }; + return await session.run(input); + } + + async startPromptWithFalseLocalOnlyResult( + input: string, + ): Promise<{ completed: () => boolean; completion: Promise }> { + const session = this.requireSession(); + const runtime = this.omp.latestSession(); + runtime.promptAck = { requestId: "prompt-local-only" }; + const promptStarted = runtime.nextPrompt(); + const completion = session.run(input); + let isCompleted = false; + void completion.then( + () => { + isCompleted = true; + return undefined; + }, + () => { + isCompleted = true; + return undefined; + }, + ); + await promptStarted; + await waitForImmediate(); + runtime.emit({ + type: "prompt_result", + id: "prompt-local-only", + agentInvoked: false, + }); + return { completed: () => isCompleted, completion }; + } + + async runPromptAfterCorrelatedTrueResult( + input: string, + output: string, + ): Promise<{ completedBeforeTurn: boolean; result: unknown }> { + const session = this.requireSession(); + const runtime = this.omp.latestSession(); + runtime.promptAck = { requestId: "prompt-invoked", agentInvoked: false }; + const promptStarted = runtime.nextPrompt(); + const run = session.run(input); + let completed = false; + void run.then( + () => { + completed = true; + return undefined; + }, + () => { + completed = true; + return undefined; + }, + ); + await promptStarted; + runtime.emit({ + type: "prompt_result", + id: "prompt-invoked", + agentInvoked: true, + }); + await waitForImmediate(); + await waitForImmediate(); + const completedBeforeTurn = completed; + runtime.acceptPrompt(input, "user-1"); + runtime.beginTurn(); + runtime.streamAssistantText(output); + runtime.finishTurn(); + return { completedBeforeTurn, result: await run }; + } + + async runPromptAfterDelayedFalseLocalOnlyResult( + input: string, + output: string, + ): Promise<{ completedBeforeTurn: boolean; result: unknown }> { + const session = this.requireSession(); + const runtime = this.omp.latestSession(); + runtime.promptAck = { requestId: "prompt-1" }; + const promptStarted = runtime.nextPrompt(); + const run = session.run(input); + let completed = false; + void run.then( + () => { + completed = true; + return undefined; + }, + () => { + completed = true; + return undefined; + }, + ); + await promptStarted; + await waitForImmediate(); + runtime.emit({ + type: "prompt_result", + id: "prompt-1", + agentInvoked: false, + }); + await waitForImmediate(); + const completedBeforeTurn = completed; + runtime.acceptPrompt(input, "user-1"); + runtime.beginTurn(); + runtime.streamAssistantText(output); + runtime.finishTurn(); + return { completedBeforeTurn, result: await run }; + } + + async runAutonomousTurn(output: string): Promise { + const runtime = this.omp.latestSession(); + runtime.beginTurn(); + runtime.streamAssistantText(output); + runtime.finishTurn(); + await waitForImmediate(); + } + timeline(): AgentTimelineItem[] { return this.events.flatMap((event) => (event.type === "timeline" ? [event.item] : [])); } @@ -357,6 +484,7 @@ export class OmpHarness { async close(): Promise { await this.requireSession().close(); + await waitForImmediate(); } isClosed(): boolean {