diff --git a/packages/server/src/server/agent/plugin-provider.test.ts b/packages/server/src/server/agent/plugin-provider.test.ts index fa274039a5..47d3d67cf9 100644 --- a/packages/server/src/server/agent/plugin-provider.test.ts +++ b/packages/server/src/server/agent/plugin-provider.test.ts @@ -4,9 +4,10 @@ import type { ProviderInput, ProviderRegistration, } from "@getpaseo/plugin/server/provider"; +import { setImmediate as nextTurn } from "node:timers/promises"; import { describe, expect, test } from "vitest"; import { createTestLogger } from "../../test-utils/test-logger.js"; -import type { AgentStreamEvent } from "./agent-sdk-types.js"; +import type { AgentClient, AgentStreamEvent } from "./agent-sdk-types.js"; import { PluginAgentClientRegistry } from "./plugin-provider.js"; import { isStaleProviderSessionError, @@ -25,6 +26,7 @@ interface ProviderHarnessOptions { capabilities?: ProviderConnection["capabilities"]; completeTurn?: boolean; openChildren?: (rootSessionId: string, emit: (event: ProviderEvent) => void) => void; + handleInput?: (input: ProviderInput, emit: (event: ProviderEvent) => void) => Promise; } function createProviderHarness(options: ProviderHarnessOptions = {}) { @@ -43,6 +45,7 @@ function createProviderHarness(options: ProviderHarnessOptions = {}) { capabilities, async send(input) { inputs.push(input); + if (await options.handleInput?.(input, emit)) return; if (input.type === "catalog") { emit({ type: "catalog", @@ -278,6 +281,59 @@ function expectNestedChildren(events: AgentStreamEvent[]) { } describe("PluginAgentClientRegistry", () => { + test.each([false, true])( + "contains a failed session open while send is pending: %s", + async (pendingSend) => { + let listener: ((event: ProviderEvent) => void) | undefined; + const unhandled: unknown[] = []; + const observeUnhandled = (reason: unknown) => unhandled.push(reason); + const registry = new PluginAgentClientRegistry(createTestLogger()); + registry.replace([ + { + id: "failing-provider", + label: "Failing provider", + async connect() { + return { + version: 1, + capabilities: ["session.persistence"], + async send(input) { + if (input.type !== "session.open") return; + listener?.({ + type: "request.failed", + requestId: input.requestId, + error: { message: "OMP persistent session registration is in progress" }, + }); + // Keep acceptance pending across a Node event-loop turn, as IPC can do. + if (pendingSend) await nextTurn(); + }, + onEvent(nextListener) { + listener = nextListener; + return () => { + listener = undefined; + }; + }, + async close() {}, + }; + }, + }, + ]); + process.on("unhandledRejection", observeUnhandled); + try { + await expect( + registry.clients()["failing-provider"]!.createSession({ + provider: "failing-provider", + cwd: "/workspace", + }), + ).rejects.toThrow("OMP persistent session registration is in progress"); + await nextTurn(); + expect(unhandled).toEqual([]); + } finally { + await registry.shutdown(); + process.off("unhandledRejection", observeUnhandled); + } + }, + ); + test("preserves nested provider child ownership during opening", async () => { const harness = createProviderHarness({ openChildren: openNestedChildren }); const registry = new PluginAgentClientRegistry(createTestLogger()); @@ -540,3 +596,167 @@ describe("PluginAgentClientRegistry", () => { expect(harness.inputs.map((input) => input.type)).toContain("session.close"); }); }); + +type RequestKind = "session.open" | "catalog" | "session.configure" | "session.prompt"; + +async function requestFromClient(client: AgentClient, kind: RequestKind): Promise { + if (kind === "catalog") return client.fetchCatalog({ scope: "global" }); + const session = await client.createSession({ provider: client.provider, cwd: "/workspace" }); + if (kind === "session.open") return session; + if (kind === "session.configure") return session.setMode("build"); + return session.startTurn("hello", { clientMessageId: "pending-message" }); +} + +async function withObservedProvider( + options: ProviderHarnessOptions, + run: (client: AgentClient, registry: PluginAgentClientRegistry) => Promise, +): Promise { + const harness = createProviderHarness(options); + const registry = new PluginAgentClientRegistry(createTestLogger()); + const unhandled: unknown[] = []; + const observe = (reason: unknown) => unhandled.push(reason); + process.on("unhandledRejection", observe); + registry.replace([harness.registration]); + try { + await run(registry.clients()[harness.registration.id]!, registry); + await registry.shutdown(); + await nextTurn(); + expect(unhandled).toEqual([]); + } finally { + await registry.shutdown(); + process.off("unhandledRejection", observe); + } +} + +describe("pending provider responses", () => { + test.each(["session.open", "catalog", "session.configure", "session.prompt"])( + "preserves the send error for %s", + async (kind) => { + const failure = new Error("Provider send failed"); + await withObservedProvider( + { + async handleInput(input) { + if (input.type === kind) throw failure; + return false; + }, + }, + async (client) => { + await expect(requestFromClient(client, kind)).rejects.toBe(failure); + }, + ); + }, + ); + + test.each([ + [ + "catalog", + expect.objectContaining({ + models: [expect.objectContaining({ id: "plugin-model", isDefault: true })], + defaultModeId: "build", + }), + ], + ["session.configure", undefined], + ] as const)( + "contains an early request.failed for %s and permits a successful retry", + async (kind, result) => { + let failed = false; + await withObservedProvider( + { + async handleInput(input, emit) { + if (input.type !== kind || !("requestId" in input) || failed) return false; + failed = true; + emit({ + type: "request.failed", + requestId: input.requestId, + error: { message: "Provider request failed", code: "busy", diagnostic: "retry" }, + }); + await nextTurn(); + return true; + }, + }, + async (client) => { + await expect(requestFromClient(client, kind)).rejects.toMatchObject({ + message: "Provider request failed", + code: "busy", + diagnostic: "retry", + }); + await expect(requestFromClient(client, kind)).resolves.toEqual(result); + }, + ); + }, + ); + + test.each([ + ["session.open", "Provider session closed"], + ["session.configure", "Provider session closed"], + ["session.prompt", StaleProviderSessionError], + ] as const)("contains session closure while accepting %s", async (kind, error) => { + await withObservedProvider( + { + async handleInput(input, emit) { + if (input.type !== kind || !("sessionId" in input)) return false; + emit({ + type: "session.closed", + sessionId: input.sessionId, + error: { message: "Provider session closed" }, + }); + await nextTurn(); + return true; + }, + }, + async (client) => { + await expect(requestFromClient(client, kind)).rejects.toThrow(error); + }, + ); + }); + + test.each([ + ["session.open", "Provider connection closed"], + ["catalog", "Provider closed"], + ["session.configure", "Provider closed"], + ["session.prompt", StaleProviderSessionError], + ] as const)("contains connection shutdown while accepting %s", async (kind, error) => { + let disconnect!: () => Promise; + await withObservedProvider( + { + async handleInput(input) { + if (input.type !== kind) return false; + await disconnect(); + await nextTurn(); + return true; + }, + }, + async (client, registry) => { + disconnect = () => registry.shutdown(); + await expect(requestFromClient(client, kind)).rejects.toThrow(error); + }, + ); + }); + + test.each([true, false])( + "preserves cancellation with acceptance pending: %s", + async (pending) => { + const controller = new AbortController(); + const reason = new Error("Catalog refresh cancelled"); + await withObservedProvider( + { + async handleInput(input) { + if (input.type !== "catalog") return false; + if (pending) { + controller.abort(reason); + await nextTurn(); + } else { + setImmediate(() => controller.abort(reason)); + } + return true; + }, + }, + async (client) => { + await expect( + client.fetchCatalog({ scope: "global" }, { signal: controller.signal }), + ).rejects.toBe(reason); + }, + ); + }, + ); +}); diff --git a/packages/server/src/server/agent/plugin-provider.ts b/packages/server/src/server/agent/plugin-provider.ts index 77ece5f667..1977c787e1 100644 --- a/packages/server/src/server/agent/plugin-provider.ts +++ b/packages/server/src/server/agent/plugin-provider.ts @@ -92,6 +92,9 @@ function deferred(): Deferred { resolve = onResolve; reject = onReject; }); + // Provider events can reject before send() settles and the caller awaits this promise. + // Observe that interval without replacing the rejecting promise returned to the caller. + void promise.catch(() => undefined); return { promise, resolve, reject }; }