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.

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<void> {
+    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(

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<void>;
 }
 
+export interface OmpNoTurnScheduler {
+  waitForSettle(signal: AbortSignal): Promise<void>;
+}
+
+// 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<string, boolean>();
+  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<string>();
@@ -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<void> {
-    await new Promise<void>((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<void> {
-    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,
       });

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<unknown> {
+    const session = this.requireSession();
+    this.omp.latestSession().promptAck = { agentInvoked: false };
+    return await session.run(input);
+  }
+
+  async startPromptWithFalseLocalOnlyResult(
+    input: string,
+  ): Promise<{ completed: () => boolean; completion: Promise<unknown> }> {
+    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<void> {
+    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<void> {
     await this.requireSession().close();
+    await waitForImmediate();
   }
 
   isClosed(): boolean {