From 62f79e88bb6130fb4215b00ad8581e63f58b489e Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 15:27:26 +0100 Subject: [PATCH] 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 --- .../agent/providers/jsonl-rpc-process.test.ts | 22 ++++++ .../agent/providers/jsonl-rpc-process.ts | 24 ++++--- .../server/agent/providers/omp/agent.test.ts | 19 +++++ .../src/server/agent/providers/omp/agent.ts | 3 + .../agent/providers/omp/cli-runtime.test.ts | 25 +++++++ .../server/agent/providers/omp/cli-runtime.ts | 8 +-- .../server/agent/providers/omp/rpc-types.ts | 30 ++++++++ .../providers/omp/test-utils/omp-harness.ts | 8 +++ .../server/daemon-e2e/omp.real.e2e.test.ts | 69 ++++++++++++++++++- 9 files changed, 190 insertions(+), 18 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 62b6c9ead9..7dfbb6abb8 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 @@ -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[] = []; + 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(); 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 d58e2235ca..9d8b9af623 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/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): 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); } 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 11eaa63fb7..537c94f2a6 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/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(); diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index b4e7b40bee..2dc6c58dea 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/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; diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts index bca1396ca6..2deae45850 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts +++ b/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[] = []; diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index 51d340ab2c..75bd79f5f0 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/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, ) { 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 }) => { diff --git a/packages/server/src/server/agent/providers/omp/rpc-types.ts b/packages/server/src/server/agent/providers/omp/rpc-types.ts index 4af44845b2..b0514c7863 100644 --- a/packages/server/src/server/agent/providers/omp/rpc-types.ts +++ b/packages/server/src/server/agent/providers/omp/rpc-types.ts @@ -635,6 +635,36 @@ export type OmpSubagentEventPayload = z.infer; export type OmpAgentSessionEvent = z.infer; export type OmpRuntimeEvent = z.infer; + +/** + * 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; export type OmpTodoPhase = z.infer; export type OmpTodoReminderEvent = z.infer; diff --git a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts index 233a10984e..1820c187c9 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts +++ b/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 { + const session = this.requireSession(); + this.omp.queueSessionSetup((runtime) => { + runtime.promptAck = { requestId, agentInvoked: false }; + }); + return await session.run(input); + } + async runPromptWithoutTurn(input: string): Promise { const session = this.requireSession(); this.omp.latestSession().promptAck = { agentInvoked: false }; diff --git a/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts index f336f325bf..510e8b384d 100644 --- a/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts +++ b/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 { } } -async function createHarness(): Promise { +// 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 { 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 () => {