fix(omp): a prompt OMP 18.3 rejects after acknowledging it now fails with OMP's error (#6115)

* fix(omp): fail the turn when OMP 18.3 rejects a prompt after acknowledging it

OMP 18.3 acknowledges a prompt, then reports a rejection such as a missing
API key only as a second failed response frame for the same request id. The
JSONL RPC transport dropped any response with no pending request, so the
turn stayed running with no error.

The transport now publishes a response nothing is waiting for like any
other frame, and the OMP runtime reads a failed prompt response as the
prompt_result error later OMP releases send.

* fix(omp): drop prompt results held for an OMP process that was replaced

Request ids restart with each OMP process, so a rejection the old process
reported after its prompt timed out could fail a prompt on the new process
with the same id. Move the real-OMP rejection case into the existing OMP
real E2E suite.

* test(omp): verify late-response routing at the transport boundary

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>

packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts

@@ -154,6 +154,28 @@ describe("JsonlRpcProcess", () => {
     }
   });
 
+  test("publishes unmatched responses without publishing the response that settles a request", async () => {
+    const child = createInMemoryChildProcess();
+    const transport = startProcess({ child });
+    const messages: Record<string, unknown>[] = [];
+    transport.onMessage((message) => messages.push(message));
+
+    try {
+      const { id, promise } = transport.startRequest({ type: "prompt" });
+      const response = { id, type: "response", command: "prompt" };
+      const rejection = { ...response, success: false, error: "rejected after acknowledgement" };
+      const unknownResponse = { ...response, id: "unknown-request", success: true };
+      child.stdout.write(`${JSON.stringify({ ...response, success: true })}\n`);
+      child.stdout.write(`${JSON.stringify(rejection)}\n`);
+      child.stdout.write(`${JSON.stringify(unknownResponse)}\n`);
+
+      await expect(promise).resolves.toBeUndefined();
+      expect(messages).toEqual([rejection, unknownResponse]);
+    } finally {
+      await transport.close();
+    }
+  });
+
   test("rejects unsuccessful responses", async () => {
     const transport = startProcess();
 

packages/server/src/server/agent/providers/jsonl-rpc-process.ts

@@ -258,25 +258,27 @@ export class JsonlRpcProcess {
     this.frameDecoder.write(chunk);
   }
 
-  /** Route a complete logical frame to its consumer (response or subscriber). */
+  /** Route a complete logical frame to its consumer (pending request or subscriber). */
   private dispatchFrame(message: Record<string, unknown>): void {
-    if (message.type === "response") {
-      this.handleResponse(message as unknown as JsonlRpcResponse);
+    const pending =
+      message.type === "response" && typeof message.id === "string"
+        ? this.pending.get(message.id)
+        : undefined;
+    if (pending) {
+      this.settleRequest(pending, message as unknown as JsonlRpcResponse & { id: string });
       return;
     }
+    // A response nothing is waiting for is published like any other frame, so a provider
+    // can read it as an event, such as OMP 18.3 rejecting a prompt after acknowledging it.
     for (const subscriber of this.messageSubscribers) {
       subscriber(message);
     }
   }
 
-  private handleResponse(response: JsonlRpcResponse): void {
-    if (!response.id) {
-      return;
-    }
-    const pending = this.pending.get(response.id);
-    if (!pending) {
-      return;
-    }
+  private settleRequest(
+    pending: PendingRequest,
+    response: JsonlRpcResponse & { id: string },
+  ): void {
     if (pending.timer) {
       clearTimeout(pending.timer);
     }

packages/server/src/server/agent/providers/omp/agent.test.ts

@@ -1236,6 +1236,25 @@ describe("OMP agent client and session", () => {
     expect(omp.runtime().prompts).toEqual([{ message: "continue", imageCount: 0 }]);
   });
 
+  test("ignores a prompt rejection a previous OMP process left for a reused request id", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+    omp.emit({
+      type: "prompt_result",
+      id: "req_1",
+      agentInvoked: false,
+      status: "error",
+      error: { message: "No API key found for anthropic." },
+    });
+    omp.processExit("OMP RPC process exited with code 1 and signal null");
+
+    await omp.runPromptWithoutTurnOnNextRuntime("/session", "req_1");
+
+    expect(omp.runtimeLaunches()).toHaveLength(2);
+    expect(omp.turnFailures()).toEqual(["OMP RPC process exited with code 1 and signal null"]);
+    expect(omp.completedTurnCount()).toBe(1);
+  });
+
   test("reports an immediate relaunch failure without retrying in a loop", async () => {
     const omp = new OmpHarness();
     await omp.start();

packages/server/src/server/agent/providers/omp/agent.ts

@@ -1045,6 +1045,9 @@ export class OmpAgentSession implements AgentSession {
         const state = await next.getState();
         if (this.closed) throw new Error("OMP session is closed");
         this.unsubscribeRuntime?.();
+        // Request ids restart with each OMP process, so results held for the old one cannot
+        // correlate with prompts sent to the new one.
+        this.pendingPromptResults.clear();
         this.runtimeSession = next;
         this.hostTools = restarted.hostTools;
         this.state = state;

packages/server/src/server/agent/providers/omp/cli-runtime.test.ts

@@ -441,6 +441,31 @@ describe("OMP CLI runtime", () => {
     await expect(session.prompt("hello")).resolves.toEqual({ requestId: "req_1" });
   });
 
+  test("reports OMP 18.3's late prompt rejection as a failed prompt result", async () => {
+    const child = createOmpChild();
+    onOmpCommand(child, (command) => {
+      if (command.type !== "prompt") return;
+      const response = { id: command.id, type: "response", command: "prompt" };
+      const rejection = { ...response, success: false, error: "No API key found for anthropic." };
+      child.stdout.write(
+        `${JSON.stringify({ ...response, success: true })}\n${JSON.stringify(rejection)}\n`,
+      );
+    });
+    const session = await createRuntime(child).startSession({ cwd: "/workspace/project" });
+    const events: unknown[] = [];
+    session.onEvent((event) => events.push(event));
+
+    await expect(session.prompt("hello")).resolves.toEqual({ requestId: "req_1" });
+
+    expect(events).toContainEqual({
+      type: "prompt_result",
+      id: "req_1",
+      agentInvoked: false,
+      status: "error",
+      error: { message: "No API key found for anthropic." },
+    });
+  });
+
   test("negotiates RPC protocol v2 when OMP advertises it", async () => {
     const child = createOmpChild({ supportedProtocolVersions: [1, 2] });
     const commands: Record<string, unknown>[] = [];

packages/server/src/server/agent/providers/omp/cli-runtime.ts

@@ -27,7 +27,7 @@ import {
   OmpModelsResultSchema,
   OmpPromptAckSchema,
   OmpRpcCommandSchema,
-  OmpRuntimeEventSchema,
+  parseOmpRuntimeEvent,
   OmpSessionStateSchema,
   OmpSessionStatsSchema,
   type OmpThinkingLevel,
@@ -121,9 +121,9 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
     private readonly launchEnvironment: Record<string, string>,
   ) {
     process.onMessage((message) => {
-      const event = OmpRuntimeEventSchema.safeParse(message);
-      if (event.success) {
-        this.emit(event.data);
+      const event = parseOmpRuntimeEvent(message);
+      if (event) {
+        this.emit(event);
       }
     });
     process.onExit(({ error }) => {

packages/server/src/server/agent/providers/omp/rpc-types.ts

@@ -635,6 +635,36 @@ export type OmpSubagentEventPayload = z.infer<typeof OmpSubagentEventPayloadSche
 export type OmpAssistantMessageEvent = z.infer<typeof OmpAssistantMessageEventSchema>;
 export type OmpAgentSessionEvent = z.infer<typeof OmpAgentSessionEventSchema>;
 export type OmpRuntimeEvent = z.infer<typeof OmpRuntimeEventSchema>;
+
+/**
+ * COMPAT(ompLatePromptRejection): added in v0.11.0, remove after 2027-04-05 once the
+ * minimum supported OMP version emits prompt_result errors.
+ * OMP 18.3 reports a prompt it rejects after acknowledging it, such as one whose model has
+ * no API key, only as a second failed `response` frame for the prompt id. Later releases
+ * report the same outcome as a `prompt_result` with `status: "error"`.
+ */
+const OmpLatePromptRejectionSchema = z.object({
+  type: z.literal("response"),
+  command: z.literal("prompt"),
+  id: z.string(),
+  success: z.literal(false),
+  error: z.string(),
+});
+
+export function parseOmpRuntimeEvent(message: unknown): OmpRuntimeEvent | null {
+  const rejection = OmpLatePromptRejectionSchema.safeParse(message);
+  if (rejection.success) {
+    return {
+      type: "prompt_result",
+      id: rejection.data.id,
+      agentInvoked: false,
+      status: "error",
+      error: { message: rejection.data.error },
+    };
+  }
+  const event = OmpRuntimeEventSchema.safeParse(message);
+  return event.success ? event.data : null;
+}
 export type OmpTodoItem = z.infer<typeof OmpTodoItemSchema>;
 export type OmpTodoPhase = z.infer<typeof OmpTodoPhaseSchema>;
 export type OmpTodoReminderEvent = z.infer<typeof OmpTodoReminderEventSchema>;

packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts

@@ -294,6 +294,14 @@ export class OmpHarness {
     return await run;
   }
 
+  async runPromptWithoutTurnOnNextRuntime(input: string, requestId: string): Promise<unknown> {
+    const session = this.requireSession();
+    this.omp.queueSessionSetup((runtime) => {
+      runtime.promptAck = { requestId, agentInvoked: false };
+    });
+    return await session.run(input);
+  }
+
   async runPromptWithoutTurn(input: string): Promise<unknown> {
     const session = this.requireSession();
     this.omp.latestSession().promptAck = { agentInvoked: false };

packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts

@@ -1,5 +1,5 @@
 import { execFile } from "node:child_process";
-import { mkdtempSync, rmSync } from "node:fs";
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import { promisify } from "node:util";
@@ -10,6 +10,7 @@ import { afterEach, beforeAll, describe, expect, test } from "vitest";
 import type { AgentTimelineItem } from "../agent/agent-sdk-types.js";
 import {
   MIN_SUPPORTED_OMP_VERSION,
+  OmpAgentClient,
   formatOmpVersionSupport,
 } from "../agent/providers/omp/agent.js";
 import { DaemonClient } from "../test-utils/daemon-client.js";
@@ -48,14 +49,53 @@ async function preflight(): Promise<void> {
   }
 }
 
-async function createHarness(): Promise<Harness> {
+// OMP exits at startup when no model is usable at all, so the home offers one offline
+// model, and an agent on any other provider's model has no credentials.
+function createOfflineOmpHome(): string {
+  const home = mkdtempSync(path.join(tmpdir(), "paseo-real-omp-offline-"));
+  roots.add(home);
+  const agentDir = path.join(home, ".omp", "agent");
+  mkdirSync(agentDir, { recursive: true });
+  writeFileSync(
+    path.join(agentDir, "models.yml"),
+    [
+      "providers:",
+      "  offline:",
+      "    baseUrl: http://127.0.0.1:9/v1",
+      "    api: openai-completions",
+      "    auth: none",
+      "    models:",
+      "      - id: offline-1",
+      "        name: Offline",
+      "        api: openai-completions",
+      "        reasoning: false",
+      "        input: [text]",
+      "        contextWindow: 128000",
+      "        maxTokens: 4096",
+      "        cost: {input: 0, output: 0, cacheRead: 0, cacheWrite: 0}",
+      "",
+    ].join("\n"),
+  );
+  return home;
+}
+
+async function createHarness(options: { ompHome?: string } = {}): Promise<Harness> {
   const cwd = mkdtempSync(path.join(tmpdir(), "paseo-real-omp-"));
   const paseoHomeRoot = mkdtempSync(path.join(tmpdir(), "paseo-real-omp-home-"));
   const staticDir = mkdtempSync(path.join(tmpdir(), "paseo-real-omp-static-"));
   for (const root of [cwd, paseoHomeRoot, staticDir]) roots.add(root);
   const logger = pino({ level: process.env.OMP_E2E_LOG_LEVEL ?? "silent" });
   const daemon = await createTestPaseoDaemon({
-    agentClients: createRealProviderClients(["omp"], logger),
+    agentClients: options.ompHome
+      ? {
+          omp: new OmpAgentClient({
+            logger,
+            runtimeSettings: {
+              env: { HOME: options.ompHome, ANTHROPIC_API_KEY: "", ANTHROPIC_OAUTH_TOKEN: "" },
+            },
+          }),
+        }
+      : createRealProviderClients(["omp"], logger),
     providerOverrides: { omp: { enabled: true } },
     logger,
     paseoHomeRoot,
@@ -186,6 +226,29 @@ afterEach(() => {
 });
 
 describe("daemon E2E (real OMP)", () => {
+  test(
+    "a prompt OMP rejects before its agent runs fails the turn with OMP's error",
+    async () => {
+      const harness = await createHarness({ ompHome: createOfflineOmpHome() });
+      try {
+        const agent = await harness.client.createAgent({
+          cwd: harness.cwd,
+          title: "no-credentials",
+          provider: "omp",
+          model: "anthropic/claude-haiku-4-5",
+          modeId: "full",
+        });
+        await harness.client.sendMessage(agent.id, "Reply with ok.");
+        const finish = await harness.client.waitForFinish(agent.id, 60_000);
+        expect(finish.status).toBe("error");
+        expect(finish.error).toContain("No API key found for anthropic");
+      } finally {
+        await closeHarness(harness);
+      }
+    },
+    TIMEOUT_MS,
+  );
+
   test(
     "prompt, native tool, and resumed follow-up",
     async () => {