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 }>,