diff --git a/.changeset/openapi-transport-unreachable.md b/.changeset/openapi-transport-unreachable.md new file mode 100644 index 000000000..8921b9b31 --- /dev/null +++ b/.changeset/openapi-transport-unreachable.md @@ -0,0 +1,6 @@ +--- +"executor": patch +"@executor-js/plugin-openapi": patch +--- + +OpenAPI tools that cannot reach the upstream server now return an `upstream_unreachable` error with an actionable network message instead of `Internal tool error [id]`. diff --git a/apps/cloud/src/mcp/session-build-semaphore.test.ts b/apps/cloud/src/mcp/session-build-semaphore.test.ts index 3d4ad7634..584b65ee0 100644 --- a/apps/cloud/src/mcp/session-build-semaphore.test.ts +++ b/apps/cloud/src/mcp/session-build-semaphore.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, beforeEach } from "@effect/vitest"; +import { describe, expect, it, beforeEach, afterEach, vi } from "@effect/vitest"; import { acquireBuildSlot, @@ -13,6 +13,10 @@ describe("session-build-semaphore", () => { resetBuildSlotsForTest(); }); + afterEach(() => { + vi.useRealTimers(); + }); + it("grants up to the cap immediately, with no wait", async () => { const results = await Promise.all([ acquireBuildSlot().promise, @@ -214,6 +218,7 @@ describe("session-build-semaphore", () => { }); it("proceeds without a slot when the queue wait exceeds the timeout, and does not count it as active", async () => { + vi.useFakeTimers(); await Promise.all([ acquireBuildSlot().promise, acquireBuildSlot().promise, @@ -223,6 +228,10 @@ describe("session-build-semaphore", () => { expect(currentActiveBuildsForTest()).toBe(4); const timedOutHandle = acquireBuildSlot(10); + await vi.advanceTimersByTimeAsync(9); + expect(currentQueueLengthForTest()).toBe(1); + expect(currentActiveBuildsForTest()).toBe(4); + await vi.advanceTimersByTimeAsync(1); const result = await timedOutHandle.promise; expect(result).toEqual({ acquired: false, waitMs: expect.any(Number), timedOut: true }); diff --git a/e2e/scenarios/openapi-unreachable-artifact.test.ts b/e2e/scenarios/openapi-unreachable-artifact.test.ts new file mode 100644 index 000000000..179ca5048 --- /dev/null +++ b/e2e/scenarios/openapi-unreachable-artifact.test.ts @@ -0,0 +1,226 @@ +// Cross-target: an artifact whose OpenAPI query cannot reach its upstream gets +// an actionable network error, not the opaque defect mask. This walks the real +// path from a saved artifact through the nested shell, execute-action, sandbox, +// OpenAPI transport, and back into ArtifactError. +import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import type { Page } from "playwright"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { ConnectionName, IntegrationSlug, type ArtifactId } from "@executor-js/sdk/shared"; + +import { scenario } from "../src/scenario"; +import { Api, Browser, Mcp, Target } from "../src/services"; +import { visit } from "../src/surfaces/browser"; +import type { McpSession } from "../src/surfaces/mcp"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +type DroppingUpstream = { + readonly url: string; + readonly requests: () => number; + readonly close: () => void; +}; + +// Accept the request, then drop the socket before sending response headers. +// This produces a real transport failure without relying on a hardcoded or +// temporarily-unused port. +const serveDroppingUpstream = () => + Effect.acquireRelease( + Effect.callback((resume) => { + let hits = 0; + const server = createServer((_request, response) => { + hits += 1; + response.destroy(); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}`, + requests: () => hits, + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), + ); + +const unreachableSpec = (baseUrl: string): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Unreachable API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/things": { + get: { + tags: ["things"], + operationId: "listThings", + summary: "List things", + responses: { + "200": { + description: "Things", + content: { + "application/json": { + schema: { type: "array", items: { type: "object" } }, + }, + }, + }, + }, + }, + }, + }, + }); + +const createConnectionCode = (slug: string) => ` +const created = await tools.executor.coreTools.connections.create({ + owner: "org", + name: "public", + integration: ${JSON.stringify(slug)}, + template: "none", +}); +return JSON.stringify(created.ok ? { ok: true } : { ok: false, error: created.error }); +`; + +const executeApproved = (session: McpSession, code: string) => + Effect.gen(function* () { + let result = yield* session.call("execute", { code }); + let guard = 0; + while (result.text.includes("executionId:") && guard < 10) { + result = yield* session.approvePaused(result.text); + guard += 1; + } + expect(result.ok, `execute completed (got: ${result.text.slice(0, 400)})`).toBe(true); + return result.text; + }); + +const artifactSource = (slug: string) => ` +function App() { + const query = useQuery(tools.${slug}.things.listThings.queryOptions({})); + const result = query.data; + return ( +
+

Upstream status

+
+ {query.isLoading ? ( + + ) : query.error ? ( + + ) : result?.ok === false ? ( + + ) : ( +

Unexpected upstream success

+ )} +
+
+ ); +} +`; + +const structuredOf = (result: { readonly raw: unknown }): Record => + ((result.raw as { structuredContent?: Record }).structuredContent ?? + {}) as Record; + +const artifactContent = (page: Page) => + page.frameLocator('[data-testid="artifact-shell-frame"]').frameLocator("iframe"); + +scenario( + "Artifacts · an unreachable OpenAPI host shows actionable retry guidance instead of an internal error", + { timeout: 180_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const browser = yield* Browser; + const mcp = yield* Mcp; + const { client: makeClient } = yield* Api; + + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const session = mcp.session(identity); + const upstream = yield* serveDroppingUpstream(); + const slug = unique("unreachable"); + const title = `Unreachable upstream ${randomBytes(4).toString("hex")}`; + let artifactId: ArtifactId | undefined; + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: unreachableSpec(upstream.url) }, + slug, + baseUrl: upstream.url, + }, + }); + + const created = yield* executeApproved(session, createConnectionCode(slug)); + expect(created, `the no-auth connection was created: ${created}`).toContain('"ok":true'); + + const rendered = yield* session.call("create-artifact", { + code: artifactSource(slug), + title, + description: "Shows whether the upstream API is reachable", + connections: { [slug]: `${slug}.org.public` }, + }); + expect(rendered.ok, `create-artifact succeeded: ${rendered.text}`).toBe(true); + + const structured = structuredOf(rendered); + artifactId = structured.artifactId as ArtifactId; + expect(artifactId, "the artifact was persisted").toBeTruthy(); + + yield* browser.session(identity, async ({ page, step }) => { + await step("Open the artifact that reads from the unreachable API", async () => { + await visit(page, String(structured.url)); + await page.getByRole("heading", { name: title }).waitFor({ timeout: 20_000 }); + }); + + await step( + "The artifact explains that the upstream host could not be reached", + async () => { + const state = artifactContent(page).getByTestId("upstream-state"); + await state.locator('[data-slot="artifact-error"]').waitFor({ timeout: 30_000 }); + const message = await state.innerText(); + + expect(message, "the user gets actionable network guidance").toContain( + "Could not reach the upstream server", + ); + expect(message, "the opaque defect mask never reaches the artifact").not.toContain( + "Internal tool error", + ); + expect(message, "the request path is not leaked").not.toContain("/things"); + }, + ); + }); + + expect(upstream.requests(), "the artifact made a real upstream request").toBeGreaterThan( + 0, + ); + }), + Effect.gen(function* () { + if (artifactId !== undefined) { + yield* client.artifacts.remove({ params: { artifactId } }).pipe(Effect.ignore); + } + yield* client.connections + .remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("public"), + }, + }) + .pipe(Effect.ignore); + yield* client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore); + }), + ); + }), + ), +); diff --git a/packages/plugins/openapi/src/sdk/backing.ts b/packages/plugins/openapi/src/sdk/backing.ts index b7a54d4f4..767753d30 100644 --- a/packages/plugins/openapi/src/sdk/backing.ts +++ b/packages/plugins/openapi/src/sdk/backing.ts @@ -727,7 +727,16 @@ export const invokeOpenApiBackedTool = (input: { details: error.cause ?? error, }), }) - : Effect.fail(error), + : error.reason === "transport_error" + ? Effect.succeed({ + ok: false as const, + failure: ToolResult.fail({ + code: "upstream_unreachable", + message: + "Could not reach the upstream server. Check your network and try again.", + }), + }) + : Effect.fail(error), ), ); diff --git a/packages/plugins/openapi/src/sdk/errors.ts b/packages/plugins/openapi/src/sdk/errors.ts index 6a5fc4a8c..b2042c5a4 100644 --- a/packages/plugins/openapi/src/sdk/errors.ts +++ b/packages/plugins/openapi/src/sdk/errors.ts @@ -39,7 +39,11 @@ export class OpenApiSpecOverrideError extends Schema.TaggedErrorClass; - readonly reason?: "response_headers_timeout" | "response_body_timeout" | "unknown_arguments"; + readonly reason?: + | "response_headers_timeout" + | "response_body_timeout" + | "unknown_arguments" + | "transport_error"; readonly cause?: unknown; }> {} diff --git a/packages/plugins/openapi/src/sdk/invoke.ts b/packages/plugins/openapi/src/sdk/invoke.ts index 2f9bd0f6f..4cf45a739 100644 --- a/packages/plugins/openapi/src/sdk/invoke.ts +++ b/packages/plugins/openapi/src/sdk/invoke.ts @@ -1,4 +1,4 @@ -import { Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect"; +import { Effect, Exit, Fiber, Layer, Option, Predicate, Schema, Stream } from "effect"; import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; import { isToolFile, type ToolFileValue } from "@executor-js/sdk/core"; @@ -1183,6 +1183,9 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( (err) => new OpenApiInvocationError({ message: "HTTP request failed", + ...(Predicate.isTagged(err.reason, "TransportError") + ? { reason: "transport_error" as const } + : {}), statusCode: Option.none(), cause: err, }), @@ -1197,7 +1200,6 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( }), ), ); - const fiber = runFork(responseEffect); const interrupt = () => { runFork(Fiber.interrupt(fiber)); }; @@ -1207,6 +1209,7 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( interrupt(); resume(Effect.succeed(Option.none())); }, responseHeadersTimeoutMs); + const fiber = runFork(responseEffect); signal.addEventListener("abort", interrupt, { once: true }); return Effect.sync(() => { clearTimeout(timer); @@ -1257,7 +1260,6 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( }), ), ); - const fiber = runFork(bodyEffect); const interrupt = () => { runFork(Fiber.interrupt(fiber)); }; @@ -1267,6 +1269,7 @@ export const invoke = Effect.fn("OpenApi.invoke")(function* ( interrupt(); resume(Effect.succeed(Option.none())); }, responseBodyTimeoutMs); + const fiber = runFork(bodyEffect); signal.addEventListener("abort", interrupt, { once: true }); return Effect.sync(() => { clearTimeout(timer); diff --git a/packages/plugins/openapi/src/sdk/upstream-failures.test.ts b/packages/plugins/openapi/src/sdk/upstream-failures.test.ts index 2089e2fdf..e16769422 100644 --- a/packages/plugins/openapi/src/sdk/upstream-failures.test.ts +++ b/packages/plugins/openapi/src/sdk/upstream-failures.test.ts @@ -11,8 +11,14 @@ // --------------------------------------------------------------------------- import { describe, expect, it } from "@effect/vitest"; -import { Effect, Exit, Schema } from "effect"; -import { FetchHttpClient, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"; +import { Cause, Data, Effect, Exit, Layer, Schema } from "effect"; +import { + FetchHttpClient, + HttpClient, + HttpClientError, + HttpServerRequest, + HttpServerResponse, +} from "effect/unstable/http"; import { HttpApi, HttpApiBuilder, @@ -43,8 +49,10 @@ import { import { openApiPlugin } from "./plugin"; -const testPlugins = () => - [openApiPlugin({ httpClientLayer: FetchHttpClient.layer }), memoryCredentialsPlugin()] as const; +class AdapterDefect extends Data.TaggedError("AdapterDefect") {} + +const testPlugins = (httpClientLayer = FetchHttpClient.layer) => + [openApiPlugin({ httpClientLayer }), memoryCredentialsPlugin()] as const; // `/things` GET op `listThings` under group "things" → tool path // `things.listThings`, used verbatim (dots and all) as the address tool segment. @@ -114,9 +122,11 @@ const FailureApi = HttpApi.make("failuresTest") // Build an executor + connection from the FailureApi HttpApi against an // arbitrary baseUrl (used for the Node-transport socket-drop / slow cases). -const buildExecutor = (baseUrl: string) => +const buildExecutor = (baseUrl: string, httpClientLayer = FetchHttpClient.layer) => Effect.gen(function* () { - const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() })); + const executor = yield* createExecutor( + makeTestConfig({ plugins: testPlugins(httpClientLayer) }), + ); yield* executor.openapi.addSpec( makeOpenApiHttpApiTestIntegrationConfig(FailureApi, { slug: "f", baseUrl }), ); @@ -394,9 +404,12 @@ describe("OpenAPI upstream failure modes", () => { const { baseUrl } = yield* startDroppingServer(); const { executor, address } = yield* buildExecutor(baseUrl); - const exit = yield* executor.execute(address, {}).pipe(Effect.exit); + const result = yield* executor.execute(address, {}); - expect(Exit.isFailure(exit)).toBe(true); + expect(result).toMatchObject({ + ok: false, + error: { code: "upstream_unreachable" }, + }); }), ); @@ -446,4 +459,82 @@ describe("OpenAPI upstream failure modes", () => { expect(result.data).toEqual([]); }), ); + + it.effect("request encoding failures remain invocation failures", () => + Effect.gen(function* () { + const httpClientLayer = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.EncodeError({ request, cause: new AdapterDefect() }), + }), + ), + ), + ); + const { executor, address } = yield* buildExecutor( + "https://upstream.example", + httpClientLayer, + ); + const exit = yield* executor.execute(address, {}).pipe(Effect.exit); + expect(Exit.isFailure(exit)).toBe(true); + }), + ); + + it.effect("transport defects remain defects", () => + Effect.gen(function* () { + const defect = new AdapterDefect(); + const httpClientLayer = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make(() => Effect.die(defect)), + ); + const { executor, address } = yield* buildExecutor( + "https://upstream.example", + httpClientLayer, + ); + const exit = yield* executor.execute(address, {}).pipe(Effect.exit); + expect(Exit.isFailure(exit)).toBe(true); + expect(Exit.match(exit, { onFailure: Cause.hasDies, onSuccess: () => false })).toBe(true); + }), + ); + + it.effect("interrupted transport remains interrupted", () => + Effect.gen(function* () { + const httpClientLayer = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make(() => Effect.interrupt), + ); + const { executor, address } = yield* buildExecutor( + "https://upstream.example", + httpClientLayer, + ); + const exit = yield* executor.execute(address, {}).pipe(Effect.exit); + expect(Exit.isFailure(exit)).toBe(true); + expect(Exit.match(exit, { onFailure: Cause.hasInterrupts, onSuccess: () => false })).toBe( + true, + ); + }), + ); + + // Port 1 refuses immediately. The same path used to throw `Internal tool + // error [hex]` because the raw HttpClientError carries the request URL. + it.effect("connection refused returns upstream_unreachable without leaking the path", () => + Effect.gen(function* () { + const { executor, address } = yield* buildExecutor("http://127.0.0.1:1"); + + const result = yield* executor.execute(address, {}); + + expect(result).toMatchObject({ + ok: false, + error: { code: "upstream_unreachable" }, + }); + const failure = result as { + readonly ok: false; + readonly error: { readonly message: string }; + }; + expect(failure.error.message).toContain("Could not reach"); + expect(failure.error.message).not.toContain("Internal tool error"); + expect(failure.error.message).not.toContain("/things"); + }), + ); });