From 24b5fd5010a1fbea68276c3fa3e3c345bc6a759e Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 21:44:08 +0200 Subject: [PATCH] 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. --- .../agent/providers/jsonl-rpc-process.test.ts | 19 ++++++--- .../agent/providers/jsonl-rpc-process.ts | 5 ++- .../server/agent/providers/omp/agent.test.ts | 41 ++++++++++++++++++- .../src/server/agent/providers/omp/agent.ts | 25 ++++++++--- .../src/server/agent/providers/omp/runtime.ts | 1 + 5 files changed, 76 insertions(+), 15 deletions(-) diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts index 2b3e88659a..6f5a6aa4e5 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts +++ b/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(); } diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts index 791df2500a..ab9ef0a866 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/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`, ), ); 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 d2b5494b70..48ddffbc15 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/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; diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 6661bfc7a0..8504c2d022 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/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; } diff --git a/packages/server/src/server/agent/providers/omp/runtime.ts b/packages/server/src/server/agent/providers/omp/runtime.ts index dc8001e3ec..a96854493d 100644 --- a/packages/server/src/server/agent/providers/omp/runtime.ts +++ b/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 }>,