fix(omp): claim the oldest pending correlation and drop refused prompts

OMP echoes prompts in the order Paseo sent them, so the oldest unclaimed
correlation with the echoed text belongs to the echo. Claiming the newest
swapped client ids when a prompt timed out after OMP recorded it and the
same text was resubmitted: each echo reconciled to the other submission.

A prompt OMP answers with a failure response is never echoed, so its
correlation is dropped on dispatch. JsonlRpcProcess now rejects those
responses with JsonlRpcRequestRejectedError, which keeps a timeout
(where OMP may still carry the prompt out) distinguishable from a refusal.

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

@@ -4,7 +4,11 @@ import { PassThrough } from "node:stream";
 import pino from "pino";
 import { describe, expect, test } from "vitest";
 
-import { JsonlRpcProcess, type JsonlRpcExit } from "./jsonl-rpc-process.js";
+import {
+  JsonlRpcProcess,
+  JsonlRpcRequestRejectedError,
+  type JsonlRpcExit,
+} from "./jsonl-rpc-process.js";
 
 const CHILD_SOURCE = String.raw`
 const readline = require("node:readline");
@@ -154,13 +158,13 @@ describe("JsonlRpcProcess", () => {
     }
   });
 
-  test("rejects unsuccessful responses", async () => {
+  test("rejects unsuccessful responses as rejected requests", async () => {
     const transport = startProcess();
 
     try {
-      await expect(transport.request({ type: "fail" })).rejects.toThrow(
-        "child rejected the request",
-      );
+      const request = transport.request({ type: "fail" });
+      await expect(request).rejects.toThrow("child rejected the request");
+      await expect(request).rejects.toBeInstanceOf(JsonlRpcRequestRejectedError);
     } finally {
       await transport.close();
     }
@@ -173,9 +177,12 @@ describe("JsonlRpcProcess", () => {
     try {
       child.stderr.write("still waiting");
 
-      await expect(transport.request({ type: "hang" })).rejects.toThrow(
+      const request = transport.request({ type: "hang" });
+      await expect(request).rejects.toThrow(
         /JSONL RPC request timed out phase=hang elapsedMs=\d+ timeoutMs=50\nstill waiting/,
       );
+      // A timed-out request may still be carried out, so it is not a rejection.
+      await expect(request).rejects.not.toBeInstanceOf(JsonlRpcRequestRejectedError);
     } finally {
       await transport.close();
     }

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

@@ -40,6 +40,9 @@ interface PendingRequest {
   timer: NodeJS.Timeout | null;
 }
 
+/** The process answered the request with a failure, so it did not carry it out. */
+export class JsonlRpcRequestRejectedError extends Error {}
+
 export interface JsonlRpcExit {
   code: number | null;
   signal: NodeJS.Signals | null;
@@ -245,7 +248,7 @@ export class JsonlRpcProcess {
     this.pending.delete(response.id);
     if (!response.success) {
       pending.reject(
-        new Error(
+        new JsonlRpcRequestRejectedError(
           response.error ?? `${this.diagnosticName} ${response.command ?? "request"} failed`,
         ),
       );

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

@@ -6,6 +6,7 @@ import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js";
 import type { OmpUsagePollScheduler } from "./usage-poller.js";
 import { resolveOmpProviderParams } from "./provider-config.js";
 import { OmpHarness } from "./test-utils/omp-harness.js";
+import { JsonlRpcRequestRejectedError } from "../jsonl-rpc-process.js";
 
 test("OMP ready timeout defaults to 20 seconds and RPC timeout overrides both", () => {
   expect(resolveOmpProviderParams({}).runtimeProviderParams).toMatchObject({
@@ -774,12 +775,48 @@ describe("OMP agent client and session", () => {
     ]);
   });
 
-  test("a prompt that never reaches OMP leaves no correlation for a later submission", async () => {
+  test("a timed-out prompt OMP echoes late keeps its own correlation when the same text is resubmitted", async () => {
     const omp = new OmpHarness();
     await omp.start();
 
     const runtime = omp.runtime();
-    runtime.promptError = new Error("OMP prompt dispatch failed");
+    // OMP recorded the prompt, but Paseo's request timed out before it was answered.
+    runtime.promptError = new Error("OMP prompt request timed out");
+    await omp.requireStartTurnFromClient("retry me", "client-a");
+    await omp.waitForEvent("turn_failed");
+    runtime.promptError = null;
+
+    await omp.requireStartTurnFromClient("retry me", "client-b");
+    // OMP echoes both prompts in the order it received them.
+    runtime.beginTurn();
+    runtime.acceptPrompt("retry me", "omp-user-a");
+    runtime.acceptPrompt("retry me", "omp-user-b");
+    runtime.streamAssistantText("done", "omp-assistant-b");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(2);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      {
+        type: "user_message",
+        text: "retry me",
+        messageId: "omp-user-a",
+        clientMessageId: "client-a",
+      },
+      {
+        type: "user_message",
+        text: "retry me",
+        messageId: "omp-user-b",
+        clientMessageId: "client-b",
+      },
+    ]);
+  });
+
+  test("a prompt OMP rejects leaves no correlation for a later submission", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    const runtime = omp.runtime();
+    runtime.promptError = new JsonlRpcRequestRejectedError("OMP refused the prompt");
     await omp.requireStartTurnFromClient("retry me", "client-a");
     await omp.waitForEvent("turn_failed");
     runtime.promptError = null;

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

@@ -72,6 +72,7 @@ import { getUserMessageText } from "./message-history.js";
 import { mapOmpSystemNoticeToNotification } from "./system-notice.js";
 import { materializeProviderImage } from "../provider-image-output.js";
 import { OmpCliRuntime } from "./cli-runtime.js";
+import { JsonlRpcRequestRejectedError } from "../jsonl-rpc-process.js";
 import { listOmpImportableSessions, readOmpImportSessionConfig } from "./session-descriptor.js";
 import type { OmpRuntime, OmpRuntimeSession, OmpStartSessionInput } from "./runtime.js";
 import type {
@@ -999,6 +1000,11 @@ export class OmpAgentSession implements AgentSession {
           return;
         }
       } catch (error) {
+        if (error instanceof JsonlRpcRequestRejectedError) {
+          // OMP refused the prompt, so no echo will ever claim it. Any other
+          // failure (a timeout) leaves the prompt with OMP, which still echoes it.
+          this.forgetClientCorrelation(options?.clientMessageId);
+        }
         if (this.activeTurnId !== turnId) {
           return;
         }
@@ -1305,13 +1311,20 @@ export class OmpAgentSession implements AgentSession {
     }
   }
 
-  // Paseo runs one foreground turn at a time, so two unclaimed correlations with
-  // the same text mean the older prompt's turn ended without an echo (a failed
-  // or timed-out dispatch). The newest one belongs to the submission OMP is
-  // echoing now; the older one stays queued in case OMP echoes it late, and
-  // falls off the bounded queue otherwise.
+  private forgetClientCorrelation(clientMessageId: string | undefined): void {
+    const index = this.pendingClientCorrelations.findIndex(
+      (entry) => entry.clientMessageId === clientMessageId,
+    );
+    if (index !== -1) {
+      this.pendingClientCorrelations.splice(index, 1);
+    }
+  }
+
+  // OMP echoes prompts in the order Paseo sent them, and a prompt OMP refused
+  // is forgotten on dispatch, so the oldest unclaimed correlation with the
+  // echoed text belongs to this echo.
   private takeClientCorrelation(text: string): string | null {
-    const index = this.pendingClientCorrelations.findLastIndex((entry) => entry.text === text);
+    const index = this.pendingClientCorrelations.findIndex((entry) => entry.text === text);
     if (index === -1) {
       return null;
     }

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

@@ -44,6 +44,7 @@ export interface OmpStartSessionInput {
 
 export interface OmpRuntimeSession {
   onEvent(callback: (event: OmpRuntimeEvent) => void): () => void;
+  // Rejects with JsonlRpcRequestRejectedError when OMP refused the prompt.
   prompt(
     message: string,
     images?: Array<{ type: "image"; data: string; mimeType: string }>,