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 () => {