diff --git a/.changeset/toolcall-parallel-reads.md b/.changeset/toolcall-parallel-reads.md new file mode 100644 index 000000000..b4769f2ce --- /dev/null +++ b/.changeset/toolcall-parallel-reads.md @@ -0,0 +1,7 @@ +--- +"@executor-js/sdk": patch +--- + +**Faster dynamic tool calls: independent storage reads run concurrently** + +Every dynamic tool call paid for its bookkeeping reads one at a time: first the tool row, then the active policy rules, then the connection row, and later the credential resolution followed by the integration row. Each read is a separate storage round-trip, so the serial chain added tens of milliseconds per call locally and more against a remote database. The reads are mutually independent, so they now run concurrently: the tool, policy, and connection reads overlap before approval, and credential resolution overlaps the integration read after approval. Approval enforcement still completes before credential resolution starts — a declined call never triggers a token refresh — and each read's failure still surfaces at the same point with the same error as before. diff --git a/e2e/scenarios/tool-call-contract.test.ts b/e2e/scenarios/tool-call-contract.test.ts new file mode 100644 index 000000000..7cb5394e3 --- /dev/null +++ b/e2e/scenarios/tool-call-contract.test.ts @@ -0,0 +1,541 @@ +// Cross-target: the dynamic tool-call contract over MCP — the exact envelope +// an agent sees when it calls a tool, misaddresses one, or has its approval +// declined. The invoke path resolves several storage reads per call (tool row, +// policy rules, connection row, credentials, integration row); this pins the +// externally observable guarantees that must hold no matter how those reads +// are scheduled internally: +// +// 1. A well-addressed call on a live connection reaches the upstream and +// returns its payload. +// 2. A wrong tool name fails with `tool_not_found` and suggests the +// connection's real tools; a wrong connection name and a call after the +// connection was removed fail with the same shape — never the opaque +// "Internal tool error" defect mask. +// 3. An approval-gated call that the user DECLINES is never executed: the +// upstream sees no request, and credential resolution never starts — the +// authorization server records no refresh grant, even when the stored +// token is expired and an executed call would have had to refresh. The +// counterfactual (the same call approved) proves the refresh was real +// work the decline suppressed, not an assertion that would pass vacuously. +import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + OAuthClientSlug, +} from "@executor-js/sdk/shared"; +import { serveOAuthTestServer } from "@executor-js/sdk/testing"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; +import type { McpSession } from "../src/surfaces/mcp"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +type UpstreamHandle = { + readonly url: string; + /** How many times the route was actually served — the ground truth for + * "this call executed" / "this call never executed". */ + readonly requests: () => number; +}; + +/** Upstream on 127.0.0.1 that answers `GET ` with `payload` for any + * caller and records every hit. */ +const serveUpstream = (route: string, payload: unknown) => + Effect.acquireRelease( + Effect.callback void }>((resume) => { + let hits = 0; + const server = createServer((request, response) => { + if (request.method === "GET" && (request.url ?? "").startsWith(route)) { + hits += 1; + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify(payload)); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not_found" })); + }); + 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 invokeByAddressCode = (address: string, args: unknown) => ` +const segments = ${JSON.stringify(address)}.split(".").slice(1); +let node = tools; +for (const segment of segments) node = node[segment]; +const result = await node(${JSON.stringify(args)}); +return JSON.stringify(result); +`; + +/** The ToolResult envelope every sandbox tool call resolves to. */ +type ToolEnvelope = { + readonly ok: boolean; + readonly data?: unknown; + readonly error?: { + readonly code?: string; + readonly message?: string; + readonly details?: { + readonly path?: string; + readonly suggestions?: readonly string[]; + }; + }; +}; + +/** Run `execute`, auto-approving any paused execution, and require the MCP + * call itself to complete (the tool call inside may still be `ok: false`). */ +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; + }); + +/** Invoke a dynamic tool by full address and parse the envelope it returns. */ +const invokeEnvelope = (session: McpSession, address: string, args: unknown = {}) => + Effect.map( + executeApproved(session, invokeByAddressCode(address, args)), + (text) => JSON.parse(text) as ToolEnvelope, + ); + +// --------------------------------------------------------------------------- +// 1 + 2: success and the not-found error identity. +// --------------------------------------------------------------------------- + +/** OpenAPI 3 spec with one no-auth operation (no securitySchemes ⇒ the + * integration is no-auth, so a `template: "none"` connection can call it). */ +const widgetsSpec = (baseUrl: string): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Widgets API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/widgets": { + get: { + operationId: "listWidgets", + summary: "List widgets", + responses: { "200": { description: "widgets" } }, + }, + }, + }, + }); + +/** Create the no-auth connection through the gateway core tool — the same + * programmatic wire-up an agent performs. */ +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 }); +`; + +scenario( + "Tool calls · a live tool answers, and every misaddressed call fails with tool_not_found, never an internal error", + { timeout: 180_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeClient } = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveUpstream("/widgets", { + widgets: [{ id: 1, name: "anvil" }], + }); + const slug = unique("toolcall"); + const session = mcp.session(identity); + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: widgetsSpec(upstream.url) }, + slug, + baseUrl: upstream.url, + }, + }); + const created = JSON.parse( + yield* executeApproved(session, createConnectionCode(slug)), + ) as ToolEnvelope; + expect(created.ok, `the no-auth connection was created: ${JSON.stringify(created)}`).toBe( + true, + ); + + const tools = yield* client.tools.list({ query: {} }); + const address = tools + .filter((tool) => String(tool.integration) === slug) + .map((tool) => String(tool.address)) + .find((addr) => addr.endsWith("listWidgets")); + expect(address, "the listWidgets tool is in the catalog").toBeDefined(); + const path = address!.replace(/^tools\./, ""); + + // 1. A well-addressed call executes and carries the upstream's payload. + const success = yield* invokeEnvelope(session, address!); + expect( + success.ok, + `the call succeeded (got: ${JSON.stringify(success.error ?? {}).slice(0, 400)})`, + ).toBe(true); + expect(JSON.stringify(success.data), "the upstream's payload comes back").toContain( + "anvil", + ); + expect(upstream.requests(), "the upstream served exactly one call").toBe(1); + + // 2a. A wrong TOOL name on a live connection: tool_not_found, and the + // suggestions name the connection's real tools so the agent can + // self-correct. + const wrongTool = address!.replace(/listWidgets$/, "makeWidget"); + const notFound = yield* invokeEnvelope(session, wrongTool); + expect(notFound.ok, "a wrong tool name fails").toBe(false); + expect(notFound.error?.code, "the failure is identified as tool_not_found").toBe( + "tool_not_found", + ); + expect(notFound.error?.message ?? "", "the message names the problem").toContain( + "Tool not found", + ); + expect(notFound.error?.details?.path, "the failing path is a structured field").toBe( + wrongTool.replace(/^tools\./, ""), + ); + expect( + notFound.error?.details?.suggestions ?? [], + "the connection's real tool is suggested", + ).toContain(path); + + // 2b. A wrong CONNECTION name: same identity — tool_not_found. + const ghost = yield* invokeEnvelope(session, address!.replace(".public.", ".ghost.")); + expect(ghost.ok, "a wrong connection name fails").toBe(false); + expect(ghost.error?.code, "an unknown connection reports tool_not_found").toBe( + "tool_not_found", + ); + + // 2c. The connection is REMOVED: the previously working address now + // fails with the same identifiable shape — never the scrubbed + // "Internal tool error [id]" defect mask. + yield* client.connections.remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("public"), + }, + }); + const gone = yield* invokeEnvelope(session, address!); + expect(gone.ok, "a call on a removed connection fails").toBe(false); + expect( + gone.error?.code, + `a missing connection reports tool_not_found (got: ${JSON.stringify(gone.error ?? {}).slice(0, 400)})`, + ).toBe("tool_not_found"); + expect( + gone.error?.message ?? "", + "the defect mask never surfaces for a missing connection", + ).not.toContain("Internal tool error"); + + expect(upstream.requests(), "no misaddressed call ever reached the upstream").toBe(1); + }), + // Selfhost shares one workspace identity — leaked resources fail other + // scenarios' zero-state assertions. + Effect.gen(function* () { + 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); + }), + ); + }), + ), +); + +// --------------------------------------------------------------------------- +// 3: a declined approval leaves the call unexecuted — and never even starts +// credential resolution. +// --------------------------------------------------------------------------- + +const issuesSpec = ( + baseUrl: string, + oauth: { + readonly authorizationEndpoint: string; + readonly tokenEndpoint: string; + }, +): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Issues API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/issues": { + get: { + operationId: "listIssues", + summary: "List issues", + security: [{ oauth: ["issues.read"] }], + responses: { "200": { description: "issues" } }, + }, + }, + }, + components: { + securitySchemes: { + oauth: { + type: "oauth2", + flows: { + authorizationCode: { + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: { "issues.read": "Read issues" }, + }, + }, + }, + }, + }, + }); + +scenario( + "Tool calls · a declined approval leaves the call unexecuted and never touches the credential", + { timeout: 180_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeClient } = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveUpstream("/issues", { + issues: [{ id: 1, title: "first" }], + }); + // Instantly-expiring access tokens: any EXECUTED call must first redeem + // a refresh grant at the AS. That makes credential resolution itself + // observable from the AS's request ledger — the signal that must stay + // at zero for a declined call. + const oauth = yield* serveOAuthTestServer({ + scopes: ["issues.read"], + tokenExpiresInSeconds: 0, + }); + const slug = unique("declined"); + const clientSlug = OAuthClientSlug.make(unique("declinedc")); + const pattern = `${slug}.*`; + + const refreshGrants = Effect.map( + oauth.requests, + (requests) => + requests.filter( + (request) => + request.path === "/token" && + request.method === "POST" && + request.body.includes("grant_type=refresh_token"), + ).length, + ); + + const cleanup = Effect.gen(function* () { + const policies = yield* client.policies.list(); + yield* Effect.forEach( + policies.filter((policy) => policy.pattern === pattern), + (policy) => + client.policies + .remove({ + params: { policyId: policy.id }, + payload: { owner: "org" }, + }) + .pipe(Effect.ignore), + ); + yield* client.connections + .remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("main"), + }, + }) + .pipe(Effect.ignore); + yield* client.oauth + .removeClient({ + params: { slug: clientSlug }, + payload: { owner: "org" }, + }) + .pipe(Effect.ignore); + yield* client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore); + }).pipe(Effect.ignore); + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: issuesSpec(upstream.url, oauth) }, + slug, + baseUrl: upstream.url, + authenticationTemplate: [ + { + slug: "oauth", + kind: "oauth2", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: ["issues.read"], + }, + ], + }, + }); + yield* client.oauth.createClient({ + payload: { + owner: "org", + slug: clientSlug, + grant: "authorization_code", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + clientId: "test-client", + clientSecret: "test-secret", + originIntegration: IntegrationSlug.make(slug), + }, + }); + + const started = yield* client.oauth.start({ + payload: { + client: clientSlug, + clientOwner: "org", + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make(slug), + template: AuthTemplateSlug.make("oauth"), + }, + }); + expect(started.status, "oauth.start redirects to the authorization server").toBe( + "redirect", + ); + if (started.status !== "redirect") return yield* Effect.die("no redirect"); + + // Drive the test IdP's consent by hand (authorize → login → code). + const code = yield* Effect.promise(async () => { + const authorize = await fetch(started.authorizationUrl, { + redirect: "manual", + }); + const loginUrl = authorize.headers.get("location"); + if (!loginUrl) throw new Error(`authorize did not redirect: ${authorize.status}`); + const login = await fetch(loginUrl, { + method: "POST", + headers: { + authorization: `Basic ${Buffer.from("alice:password").toString("base64")}`, + }, + redirect: "manual", + }); + const callbackUrl = login.headers.get("location"); + if (!callbackUrl) throw new Error(`login did not redirect: ${login.status}`); + const minted = new URL(callbackUrl).searchParams.get("code"); + if (!minted) throw new Error("callback carried no authorization code"); + return minted; + }); + yield* client.oauth.complete({ + payload: { state: started.state, code }, + }); + + const tools = yield* client.tools.list({ query: {} }); + const address = tools + .filter((tool) => String(tool.integration) === slug) + .map((tool) => String(tool.address)) + .find((addr) => addr.endsWith("listIssues")); + expect(address, "the listIssues tool is in the catalog").toBeDefined(); + + // Gate every tool of this integration behind human approval. The + // pattern is unique per run, so a leak cannot gate another scenario. + yield* client.policies.create({ + payload: { owner: "org", pattern, action: "require_approval" }, + }); + + const session = mcp.session(identity); + const grantsBefore = yield* refreshGrants; + + const paused = yield* session.call("execute", { + code: invokeByAddressCode(address!, {}), + }); + expect( + paused.text, + `the gated call paused for approval (got: ${paused.text.slice(0, 400)})`, + ).toContain("Execution paused"); + const match = /\bexecutionId:\s*(\S+)/.exec(paused.text); + expect(match, "the paused result carries an executionId").not.toBeNull(); + + // While paused, NOTHING has run: the upstream saw no request and — + // although the stored token is expired, so an executing call would + // have to refresh first — the AS recorded no refresh grant. The + // approval gate sits before credential resolution. + expect(upstream.requests(), "the upstream saw nothing while the call was paused").toBe(0); + expect( + yield* refreshGrants, + "credential resolution did not start while the call was paused", + ).toBe(grantsBefore); + + const declined = yield* session.call("resume", { + executionId: match![1], + action: "decline", + }); + expect( + declined.text.toLowerCase(), + `the declined resume tells the agent the approval was refused (got: ${declined.text.slice(0, 400)})`, + ).toContain("declined"); + + expect(upstream.requests(), "a declined call never reached the upstream").toBe(0); + expect( + yield* refreshGrants, + "a declined call never redeemed a refresh grant — credential resolution never started", + ).toBe(grantsBefore); + + // Counterfactual: the SAME call, approved, refreshes and executes — + // proving the zeros above measured real work the decline suppressed. + let approved = yield* session.call("execute", { + code: invokeByAddressCode(address!, {}), + }); + let guard = 0; + while (approved.text.includes("executionId:") && guard < 10) { + approved = yield* session.approvePaused(approved.text); + guard += 1; + } + expect( + approved.ok, + `the approved execute completed (got: ${approved.text.slice(0, 400)})`, + ).toBe(true); + const envelope = JSON.parse(approved.text) as ToolEnvelope; + expect( + envelope.ok, + `the approved call succeeded (got: ${JSON.stringify(envelope.error ?? {}).slice(0, 400)})`, + ).toBe(true); + expect(upstream.requests(), "the approved call reached the upstream exactly once").toBe( + 1, + ); + expect( + yield* refreshGrants, + "the approved call is the one that redeemed a refresh grant", + ).toBe(grantsBefore + 1); + }), + cleanup, + ); + }), + ), +); diff --git a/packages/core/sdk/src/executor.test.ts b/packages/core/sdk/src/executor.test.ts index 377973a7f..2dc9bc959 100644 --- a/packages/core/sdk/src/executor.test.ts +++ b/packages/core/sdk/src/executor.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it } from "@effect/vitest"; -import { Data, Effect, Predicate, Result } from "effect"; +import { Data, Effect, Predicate, Result, Scheduler } from "effect"; +import { ElicitationResponse, type ElicitationHandler } from "./elicitation"; import { ToolNotFoundError } from "./errors"; +import { createExecutor } from "./executor"; +import { StorageError, type FumaDb } from "./fuma-runtime"; import { AuthTemplateSlug, ConnectionName, @@ -15,7 +18,7 @@ import { import { definePlugin } from "./plugin"; import type { CredentialProvider } from "./provider"; import { IntegrationDetectionResult } from "./types"; -import { makeTestExecutor, memoryCredentialsPlugin } from "./testing"; +import { makeTestConfig, makeTestExecutor, memoryCredentialsPlugin } from "./testing"; import { serveOAuthTestServer } from "./testing/oauth-test-server"; // removed: v1 secret browser-handoff, source.configure, case-insensitive tool-id @@ -783,3 +786,609 @@ describe("muscle memory (observed output shapes)", () => { }), ); }); + +// --------------------------------------------------------------------------- +// Dynamic tool-call read concurrency. The invoke path runs its independent +// storage reads concurrently: tool row + policy rules + connection row before +// approval, then credential resolution + integration row after approval. +// These tests pin BOTH halves of that contract — the dominant read (tool row; +// credential resolution) launches first, exactly as the sequential code +// ordered it, and the speculative reads launch before the dominant one +// completes on an asynchronous backend — AND the invariants the overlap must +// not disturb, most importantly that a declined approval never starts +// credential resolution (which can trigger an upstream token refresh). +// --------------------------------------------------------------------------- + +/** Records the lifecycle of every armed read as an ordered event log: + * `start:` is pushed SYNCHRONOUSLY at the instant the real read is + * invoked — never after an artificial pre-read suspension, which would + * itself manufacture the overlap the probe claims to observe — and + * `end:` is pushed when the read settles. An armed read's COMPLETION + * is deferred by one `setImmediate` tick (its launch is never delayed), so + * the log shows the code's own launch order against a backend that, like + * any real one, does not answer synchronously. The tick is `setImmediate` + * on purpose: the effect scheduler starts plainly-forked children from a + * `setImmediate` queued at fork time, so a completion deferred to the SAME + * FIFO queue is guaranteed to land after those children have launched — + * a timer would race them across event-loop phases. */ +const makeReadOrderRecorder = () => { + let targets: ReadonlySet = new Set(); + const events: string[] = []; + const record = (key: string, run: () => Promise): Promise => { + if (!targets.has(key)) return run(); + events.push(`start:${key}`); + return run() + .then( + (result) => + new Promise((resolve) => { + setImmediate(() => resolve(result)); + }), + ) + .finally(() => { + events.push(`end:${key}`); + }); + }; + return { + arm: (keys: readonly string[]) => { + targets = new Set(keys); + }, + record, + /** First occurrence — repeat reads of the same key log later entries. */ + at: (event: string) => events.indexOf(event), + /** Snapshot of the full event log, for failure diagnostics. */ + log: () => events.slice(), + }; +}; + +/** Wrap a test `FumaDb` so every read runs through the recorder as + * `.`. `withContext` re-wraps so the executor's + * context-bound handles stay recorded. */ +const withRecordedReads = ( + db: FumaDb, + record: (key: string, run: () => Promise) => Promise, +): FumaDb => { + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "findFirst" || prop === "findMany") { + return (table: unknown, query: unknown) => + record(`${String(table)}.${prop}`, () => + (Reflect.get(target, prop) as (t: unknown, q: unknown) => Promise).call( + target, + table, + query, + ), + ); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + +const countingProvider = ( + calls: { count: number }, + wrapGet?: (run: () => Promise) => Promise, +) => { + const store = new Map(); + const provider: CredentialProvider = { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => + Effect.promise(() => { + calls.count++; + const run = async () => store.get(String(id)) ?? null; + return wrapGet ? wrapGet(run) : run(); + }), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + }; + return provider; +}; + +const invokeConcurrencyPlugin = (provider: CredentialProvider) => + definePlugin(() => ({ + id: "demo" as const, + credentialProviders: [provider], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ tools: [{ name: ToolName.make("run"), description: "run" }] }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => ctx.core.integrations.register({ slug: INTEG, description: "Demo", config: {} }), + }), + }))(); + +const seedRunConnection = < + E extends { + readonly demo: { readonly seed: () => Effect.Effect }; + readonly connections: { + readonly create: (input: { + owner: "org"; + name: ConnectionName; + integration: IntegrationSlug; + template: AuthTemplateSlug; + from: { provider: ProviderKey; id: ProviderItemId }; + }) => Effect.Effect; + }; + }, +>( + executor: E, +) => + Effect.gen(function* () { + yield* executor.demo.seed(); + yield* executor.connections.create({ + owner: "org", + name: CONN, + integration: INTEG, + template: TEMPLATE, + from: { + provider: ProviderKey.make("memory"), + id: ProviderItemId.make("v"), + }, + }); + }); + +/** Run the recorded tool-row invoke once and return the recorder. When + * `maxOps` is set, the execute call runs under that `MaxOpsBeforeYield` + * budget so the run loop's cooperative yield lands at an adversarial + * position instead of the default one. */ +const recordToolRowLaunch = (maxOps?: number) => + Effect.gen(function* () { + const recorder = makeReadOrderRecorder(); + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: withRecordedReads(config.db, recorder.record), + }); + yield* seedRunConnection(executor); + + recorder.arm(["tool.findFirst", "tool_policy.findMany", "connection.findFirst"]); + const execute = executor.execute(addr("run"), {}); + const out = yield* maxOps === undefined + ? execute + : execute.pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, maxOps)); + expect(out).toEqual({ ran: "run" }); + return recorder; + }); + +/** Same, for the post-approval half: credential resolution vs the + * integration-row read. */ +const recordCredentialLaunch = (maxOps?: number) => + Effect.gen(function* () { + const recorder = makeReadOrderRecorder(); + const calls = { count: 0 }; + const provider = countingProvider(calls, (run) => recorder.record("credential.get", run)); + const config = makeTestConfig({ + plugins: [invokeConcurrencyPlugin(provider)] as const, + }); + const executor = yield* createExecutor({ + ...config, + db: withRecordedReads(config.db, recorder.record), + }); + yield* seedRunConnection(executor); + + recorder.arm(["credential.get", "integration.findFirst"]); + const execute = executor.execute(addr("run"), {}); + const out = yield* maxOps === undefined + ? execute + : execute.pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, maxOps)); + expect(out).toEqual({ ran: "run" }); + return recorder; + }); + +/** Assert every armed read genuinely ran to completion — `start:` and + * `end:` present for each key — carrying the budget and full event log into + * any failure. Launch ORDER is deliberately not asserted here: under an + * adversarial op budget the child's own cooperative yield can park the + * dominant read's inline launch, letting a speculative read start first. + * That reversal is accepted (see the launch-order comment in executor.ts — + * it costs no wall time on any backend, and the only way to suppress it, + * a fiber-lifetime `PreventSchedulerYield`, provably disables effect + * timeouts across provider code). What must hold at EVERY budget is that + * the call completes correctly and no read is lost or stranded. */ +const expectAllReadsRan = ( + recorder: ReturnType, + budget: number | undefined, + keys: readonly string[], +) => { + const ran = keys.every( + (key) => recorder.at(`start:${key}`) >= 0 && recorder.at(`end:${key}`) >= 0, + ); + expect({ budget, ran, log: recorder.log() }).toMatchObject({ budget, ran: true }); +}; + +// Sweeping low budgets lands the run loop's cooperative yield at every +// position in the launch window (6 and 8 are where runtime probes reproduced +// launch-order reversals). Budgets 1 and 2 are excluded because they deadlock +// the effect run loop itself (a resumed fiber re-yields before evaluating a +// single operation), independent of this code. +const ADVERSARIAL_BUDGETS: readonly number[] = [3, 4, 5, 6, 7, 8, 9, 10, 11, 12]; + +describe("execute read concurrency", () => { + it.effect( + "launches the tool-row read first, with the policy and connection reads overlapping it", + () => + Effect.gen(function* () { + const recorder = yield* recordToolRowLaunch(); + // The dominant tool-row read launches BEFORE either speculative read + // starts — the launch order the sequential code had. An immediate + // fork regresses this by running a speculative read inline, ahead of + // the tool row, on any backend that answers without suspending. + expect(recorder.at("start:tool.findFirst")).toBeGreaterThanOrEqual(0); + expect(recorder.at("start:tool.findFirst")).toBeLessThan( + recorder.at("start:tool_policy.findMany"), + ); + expect(recorder.at("start:tool.findFirst")).toBeLessThan( + recorder.at("start:connection.findFirst"), + ); + // And both speculative reads are in flight before the tool row + // completes — genuine overlap, not a serial chain: on this deferred + // (asynchronous) backend a sequential ordering would log the tool + // row's `end` before either speculative `start`. + expect(recorder.at("start:tool_policy.findMany")).toBeLessThan( + recorder.at("end:tool.findFirst"), + ); + expect(recorder.at("start:connection.findFirst")).toBeLessThan( + recorder.at("end:tool.findFirst"), + ); + }), + ); + + it.effect( + "starts credential resolution first, with the integration-row read overlapping it", + () => + Effect.gen(function* () { + const recorder = yield* recordCredentialLaunch(); + // Credential resolution — the dominant work, and the read whose + // failure must keep dominating — launches before the speculative + // integration-row read starts, which in turn starts before the + // credential read completes on this deferred (asynchronous) backend. + expect(recorder.at("start:credential.get")).toBeGreaterThanOrEqual(0); + expect(recorder.at("start:credential.get")).toBeLessThan( + recorder.at("start:integration.findFirst"), + ); + expect(recorder.at("start:integration.findFirst")).toBeLessThan( + recorder.at("end:credential.get"), + ); + }), + ); + + it.effect( + "completes with every pre-approval read run at every adversarial scheduler budget", + () => + Effect.gen(function* () { + for (const budget of ADVERSARIAL_BUDGETS) { + const recorder = yield* recordToolRowLaunch(budget); + expectAllReadsRan(recorder, budget, [ + "tool.findFirst", + "tool_policy.findMany", + "connection.findFirst", + ]); + } + }), + { timeout: 60_000 }, + ); + + it.effect( + "completes with the credential and integration reads run at every adversarial budget", + () => + Effect.gen(function* () { + for (const budget of ADVERSARIAL_BUDGETS) { + const recorder = yield* recordCredentialLaunch(budget); + expectAllReadsRan(recorder, budget, ["credential.get", "integration.findFirst"]); + } + }), + { timeout: 60_000 }, + ); + + it.effect("a declined approval never resolves credentials", () => + Effect.gen(function* () { + const calls = { count: 0 }; + const executor = yield* makeTestExecutor({ + plugins: [invokeConcurrencyPlugin(countingProvider(calls))] as const, + }); + yield* seedRunConnection(executor); + yield* executor.policies.create({ + owner: "org", + pattern: "demo.*", + action: "require_approval", + }); + + // Setup (connection create / tool sync) may read the credential; only + // reads issued by the declined call itself are the regression signal. + calls.count = 0; + const decliningHandler: ElicitationHandler = () => + Effect.succeed(ElicitationResponse.make({ action: "decline" })); + const result = yield* Effect.result( + executor.execute(addr("run"), {}, { onElicitation: decliningHandler }), + ); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ElicitationDeclinedError")(result.failure)).toBe(true); + // The decline happened BEFORE credential resolution started: a token + // refresh (a network side effect) must never fire for a declined call. + expect(calls.count).toBe(0); + }), + ); + + it.effect("fails with ConnectionNotFoundError when the tool row outlives its connection", () => + Effect.gen(function* () { + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor(config); + yield* seedRunConnection(executor); + + // Remove ONLY the connection row, leaving the tool rows behind — the + // inconsistent state the ConnectionNotFoundError branch reports. The + // concurrent connection read must still surface this error, not a + // policy or tool-row failure. + yield* Effect.promise(() => config.db.deleteMany("connection", {})); + + const result = yield* Effect.result(executor.execute(addr("run"), {})); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ConnectionNotFoundError")(result.failure)).toBe(true); + }), + ); +}); + +// --------------------------------------------------------------------------- +// Speculative read abandonment. The concurrent reads above are forked, and a +// branch that returns without needing one must interrupt it instead of +// consuming it — so a read that hangs cannot gate an error that never needed +// it, and a read that fails cannot mask that error or leak as an unhandled +// rejection. Reads a branch DOES need keep failing exactly where the +// sequential code failed. +// --------------------------------------------------------------------------- + +/** Deterministic read faults keyed by `
.`: nothing is armed + * until the test says so (setup reads pass through untouched), and the test + * can check that an armed read really started before it was abandoned. */ +const makeReadFaults = () => { + let hung: ReadonlySet = new Set(); + let failed: ReadonlyMap = new Map(); + const started = new Set(); + return { + hangReads: (keys: readonly string[]) => { + hung = new Set(keys); + }, + failReads: (byCode: Readonly>) => { + failed = new Map(Object.entries(byCode)); + }, + started, + fault: (key: string): Promise | undefined => { + if (hung.has(key)) { + started.add(key); + // Never settles. The driver promise takes no abort signal, so an + // interrupted read is abandoned exactly like this in production. + return new Promise(() => {}); + } + const code = failed.get(key); + if (code === undefined) return undefined; + started.add(key); + // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: simulate the raw driver-promise rejection that fumaEffect normalizes into a StorageFailure + return Promise.reject(Object.assign(new Error(code), { code })); + }, + }; +}; + +/** Wrap a test `FumaDb` so armed reads hang or reject. Every other read + * passes through, deferred by one `setImmediate` tick — the wrapper is an + * ASYNCHRONOUS backend, because that is where these invariants bite: on a + * backend that answers in microtasks an early-exit branch can finish before + * the plainly-forked speculative reads' scheduler tick, abandoning them + * before they ever issue a read, and the hang the test armed would go + * unexercised. `withContext` re-wraps so context-bound handles stay + * faulted. */ +const withFaultedReads = ( + db: FumaDb, + fault: (key: string) => Promise | undefined, +): FumaDb => { + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "findFirst" || prop === "findMany") { + return (table: unknown, query: unknown) => + fault(`${String(table)}.${prop}`) ?? + new Promise((resolve) => { + setImmediate(resolve); + }).then(() => + (Reflect.get(target, prop) as (t: unknown, q: unknown) => Promise).call( + target, + table, + query, + ), + ); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + +/** A credential provider that works during setup and can be armed to fail + * every later read — the site-2 "credential resolution fails" input. The + * armed failure lands one `setImmediate` tick later, like the I/O failure a + * real provider produces: an inline failure would exit the call before the + * plainly-forked integration read's scheduler tick, so the read this test + * wants hanging would never start at all. */ +const armableFailingProvider = () => { + const store = new Map(); + let failWith: string | undefined; + const provider: CredentialProvider = { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => + Effect.suspend(() => { + if (failWith === undefined) return Effect.succeed(store.get(String(id)) ?? null); + const message = failWith; + // Two chained ticks, not one: the credential read launches BEFORE + // the speculative integration fork is even scheduled (dominant-first + // is structural), so a failure delivered on the very next tick would + // interrupt that fork before it ever issued its read. The extra tick + // lets the speculative read genuinely start — and hang — first, so + // the test exercises "failure surfaces while the read hangs" rather + // than "failure surfaces before the read exists". + return Effect.promise( + () => + new Promise((resolve) => { + setImmediate(() => setImmediate(resolve)); + }), + ).pipe(Effect.andThen(Effect.fail(new StorageError({ message, cause: undefined })))); + }), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + }; + return { + provider, + failNow: (message: string) => { + failWith = message; + }, + }; +}; + +describe("speculative read abandonment", () => { + it.effect("unknown-tool error surfaces while the speculative reads hang forever", () => + Effect.gen(function* () { + const faults = makeReadFaults(); + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: withFaultedReads(config.db, faults.fault), + }); + yield* seedRunConnection(executor); + + // From here on the speculative policy and connection reads NEVER + // resolve. The unknown-tool branch needs neither, so the call must + // fail without waiting on them — the previous all-or-nothing shape + // could not fail until every read settled. + faults.hangReads(["tool_policy.findMany", "connection.findFirst"]); + const result = yield* Effect.result(executor.execute(addr("no-such-tool"), {})); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ToolNotFoundError")(result.failure)).toBe(true); + // Both hung reads really started: the branch abandoned in-flight + // reads, it did not skip forking them. + expect(faults.started.has("tool_policy.findMany")).toBe(true); + expect(faults.started.has("connection.findFirst")).toBe(true); + }), + ); + + it.effect("a blocked tool reports ToolBlockedError while the connection read hangs", () => + Effect.gen(function* () { + const faults = makeReadFaults(); + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: withFaultedReads(config.db, faults.fault), + }); + yield* seedRunConnection(executor); + yield* executor.policies.create({ + owner: "org", + pattern: "demo.*", + action: "block", + }); + + // The block branch consumes the policy read but never the connection + // read; a connection read that never resolves must not gate it. + faults.hangReads(["connection.findFirst"]); + const result = yield* Effect.result(executor.execute(addr("run"), {})); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ToolBlockedError")(result.failure)).toBe(true); + expect(faults.started.has("connection.findFirst")).toBe(true); + }), + ); + + it.effect("failing speculative reads neither mask the branch error nor unhandled-reject", () => + Effect.gen(function* () { + const faults = makeReadFaults(); + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: withFaultedReads(config.db, faults.fault), + }); + yield* seedRunConnection(executor); + + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => void unhandled.push(reason); + process.on("unhandledRejection", onUnhandled); + const result = yield* Effect.gen(function* () { + faults.failReads({ + "tool_policy.findMany": "POLICY_READ_FAILED", + "connection.findFirst": "CONNECTION_READ_FAILED", + }); + const out = yield* Effect.result(executor.execute(addr("no-such-tool"), {})); + // Both rejections fired before the call returned; give an + // unobserved one its macrotask turn to reach the process hook. + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); + return out; + }).pipe(Effect.ensuring(Effect.sync(() => process.off("unhandledRejection", onUnhandled)))); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ToolNotFoundError")(result.failure)).toBe(true); + expect(unhandled).toEqual([]); + }), + ); + + it.effect( + "a policy read failure on the consuming path surfaces, ahead of a connection failure", + () => + Effect.gen(function* () { + const faults = makeReadFaults(); + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: withFaultedReads(config.db, faults.fault), + }); + yield* seedRunConnection(executor); + + faults.failReads({ + "tool_policy.findMany": "POLICY_READ_FAILED", + "connection.findFirst": "CONNECTION_READ_FAILED", + }); + const result = yield* Effect.result(executor.execute(addr("run"), {})); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + // The policy read is consumed first, exactly as the sequential code + // ordered the reads, so its failure is the one reported even though + // the connection read failed too. + expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true); + expect((result.failure as StorageError).message).toContain("POLICY_READ_FAILED"); + }), + ); + + it.effect("a credential failure surfaces while the site-2 integration read hangs", () => + Effect.gen(function* () { + const faults = makeReadFaults(); + const armable = armableFailingProvider(); + const config = makeTestConfig({ + plugins: [invokeConcurrencyPlugin(armable.provider)] as const, + }); + const executor = yield* createExecutor({ + ...config, + db: withFaultedReads(config.db, faults.fault), + }); + yield* seedRunConnection(executor); + + // After approval, credential resolution and the integration-row read + // run concurrently. Resolution fails while the integration read never + // resolves: the credential failure must surface without waiting on the + // read — the failure path interrupts it instead of joining it. + faults.hangReads(["integration.findFirst"]); + armable.failNow("CREDENTIAL_READ_FAILED"); + const result = yield* Effect.result(executor.execute(addr("run"), {})); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true); + expect((result.failure as StorageError).message).toBe("CREDENTIAL_READ_FAILED"); + expect(faults.started.has("integration.findFirst")).toBe(true); + }), + ); +}); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 653ab5701..aba6b674c 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -5794,175 +5794,265 @@ export const createExecutor = - b.and( - byOwner(parsed.owner)(b), - b("integration", "=", String(parsed.integration)), - b("connection", "=", String(parsed.connection)), - b("name", "=", String(parsed.tool)), - ), - select: TOOL_INVOCATION_COLUMNS, - }); - if (!row) { - const searchMatches = yield* searchToolRowsForConnection(parsed); - const connectionTools = - searchMatches.length > 0 ? searchMatches : yield* findToolRowsForConnection(parsed); - // An empty catalog on a connection that DOES exist is usually not a - // wrong tool name: discovery produced nothing, most often because the - // upstream rejected the credential. Reporting only the address sends - // the reader after a tool that was never the problem, so name the - // connection and point at the surface that knows the cause. - const connectionExists = - connectionTools.length === 0 && - (yield* findConnectionRow({ - owner: parsed.owner, - integration: parsed.integration, - name: parsed.connection, - })) !== null; - return yield* new ToolNotFoundError({ - address, - suggestions: toolSuggestions(connectionTools), - reason: connectionExists - ? `connection "${parsed.integration}/${parsed.connection}" has no tools; ` + - `check its health for why discovery produced none` - : undefined, - }); - } - - // Resolve policy (owner-ranked). - const toolForPolicy = rowToTool(row); - const policyRules = yield* listActivePolicyRuleSet(); - const annotations = decodeJsonColumn(row.annotations) as ToolAnnotations | undefined; - const policy = yield* resolvePolicyFromRuleSet( - normalizedPolicyId(toolForPolicy), - policyRules, - annotations?.requiresApproval, + // The three storage reads this call needs — the tool row (projected: + // invoke needs routing/policy fields only, never the multi-KB + // input/output schema JSON; `tools.schema` is the schema-bearing + // surface), the active policy rule set, and the connection row — are + // mutually independent, so they run concurrently instead of paying + // three serial round-trips. All three are forked as children of this + // fiber (so interrupting the call still interrupts them) and + // consumed with `Fiber.join` at exactly the points the sequential + // code performed them — a joined fiber resumes with its exact Exit, + // so the caller-visible error for a given input is unchanged: a + // tool-row read failure still dominates, and a policy or connection + // read failure still surfaces only where the old code would have + // executed that read. A branch that returns without needing a + // speculative read must neither wait on it (a hung read must not + // gate an unknown-tool / blocked / plugin-not-loaded error that + // never needed it) nor swallow its failure as an unobserved value — + // the `ensuring` guard below interrupts whatever was not consumed. + // Interrupting a read mid-flight merely abandons the driver promise + // (`fumaEffect` takes no abort signal and installs its rejection + // handler at construction), so an abandoned read cannot + // unhandled-reject; a fiber interrupted before it ever ran issues no + // read at all. + // Launch order is dominant-first. The dominant tool-row read is + // forked FIRST with `startImmediately: true`: an immediate fork + // evaluates the child INLINE (`forkUnsafe` calls `child.evaluate` on + // the spot), and a forked fiber enters its run loop with a FRESH + // operation count (`runLoop` zeroes `currentOpCount`), so the few + // dozen operations between fork and the driver-promise suspension + // cannot reach the cooperative-yield budget (`MaxOpsBeforeYield` + // defaults to 2048) — the read is issued before the speculative + // forks below are even scheduled (a plain fork only queues its child + // on the dispatcher for the next tick). Dominant-first matters + // beyond taste: cloud's postgres pool is `max: 1`, so queries + // pipeline through one connection in issue order — a speculative + // query issued first would sit ahead of the read every branch needs, + // and a slow or lock-blocked speculative query would gate it. Under + // a pathologically small budget override (single digits — a test + // harness setting; 1-2 deadlocks the effect run loop itself) the + // inline launch can park early and a speculative read may issue + // first; that bounded case is accepted rather than suppressed, + // because the only known suppression (a fiber-lifetime + // `PreventSchedulerYield`) is inherited by everything the child runs + // and provably keeps effect timeouts from firing across CPU-bound + // stretches. + const toolRowFiber = yield* Effect.forkChild( + core.findFirst("tool", { + where: (b: AnyCb) => + b.and( + byOwner(parsed.owner)(b), + b("integration", "=", String(parsed.integration)), + b("connection", "=", String(parsed.connection)), + b("name", "=", String(parsed.tool)), + ), + select: TOOL_INVOCATION_COLUMNS, + }), + { startImmediately: true }, ); - if (policy.action === "block") { - return yield* new ToolBlockedError({ - address, - pattern: policy.pattern ?? "*", - }); - } - - const runtime = runtimes.get(row.plugin_id); - if (!runtime) { - return yield* new PluginNotLoadedError({ - address, - pluginId: row.plugin_id, - }); - } - if (!runtime.plugin.invokeTool) { - return yield* new NoHandlerError({ - address, - pluginId: row.plugin_id, - }); - } - - // Find the connection row. - const connectionRow = yield* findConnectionRow({ - owner: parsed.owner, - integration: parsed.integration, - name: parsed.connection, - }); - if (!connectionRow) { - return yield* new ConnectionNotFoundError({ + const policyRulesFiber = yield* Effect.forkChild(listActivePolicyRuleSet()); + const connectionRowFiber = yield* Effect.forkChild( + findConnectionRow({ owner: parsed.owner, integration: parsed.integration, name: parsed.connection, - }); - } + }), + ); + const invokeDynamicTool = Effect.gen(function* () { + const row = yield* Fiber.join(toolRowFiber); + if (!row) { + const searchMatches = yield* searchToolRowsForConnection(parsed); + const connectionTools = + searchMatches.length > 0 ? searchMatches : yield* findToolRowsForConnection(parsed); + // An empty catalog on a connection that DOES exist is usually not a + // wrong tool name: discovery produced nothing, most often because the + // upstream rejected the credential. Reporting only the address sends + // the reader after a tool that was never the problem, so name the + // connection and point at the surface that knows the cause. + // Joining here is the sequential read this branch always + // performed; the short-circuit keeps the suggestion path from + // waiting on a connection read it does not need. + const connectionExists = + connectionTools.length === 0 && (yield* Fiber.join(connectionRowFiber)) !== null; + return yield* new ToolNotFoundError({ + address, + suggestions: toolSuggestions(connectionTools), + reason: connectionExists + ? `connection "${parsed.integration}/${parsed.connection}" has no tools; ` + + `check its health for why discovery produced none` + : undefined, + }); + } + + // Resolve policy (owner-ranked). + const toolForPolicy = rowToTool(row); + const policyRules = yield* Fiber.join(policyRulesFiber); + const annotations = decodeJsonColumn(row.annotations) as ToolAnnotations | undefined; + const policy = yield* resolvePolicyFromRuleSet( + normalizedPolicyId(toolForPolicy), + policyRules, + annotations?.requiresApproval, + ); + if (policy.action === "block") { + return yield* new ToolBlockedError({ + address, + pattern: policy.pattern ?? "*", + }); + } - // Resolve annotations + enforce approval. - let resolvedAnnotations = annotations; - if (policy.action !== "approve" && runtime.plugin.resolveAnnotations) { - const map = yield* runtime.plugin - .resolveAnnotations({ - ctx: runtime.ctx, + const runtime = runtimes.get(row.plugin_id); + if (!runtime) { + return yield* new PluginNotLoadedError({ + address, + pluginId: row.plugin_id, + }); + } + if (!runtime.plugin.invokeTool) { + return yield* new NoHandlerError({ + address, + pluginId: row.plugin_id, + }); + } + + // Join the connection row (read concurrently above). + const connectionRow = yield* Fiber.join(connectionRowFiber); + if (!connectionRow) { + return yield* new ConnectionNotFoundError({ + owner: parsed.owner, + integration: parsed.integration, + name: parsed.connection, + }); + } + + // Resolve annotations + enforce approval. + let resolvedAnnotations = annotations; + if (policy.action !== "approve" && runtime.plugin.resolveAnnotations) { + const map = yield* runtime.plugin + .resolveAnnotations({ + ctx: runtime.ctx, + integration: parsed.integration, + connection: parsed.connection, + toolRows: [row], + }) + .pipe(wrapInvocationError); + resolvedAnnotations = map[String(parsed.tool)] ?? annotations; + } + // When this call is about to pause for approval, validate args + // first: a call that can only fail (missing required path param / + // body) must be rejected here, not after the user grants an approval + // that then goes to waste. Non-pausing calls skip this — invokeTool + // raises the identical failure moments later without the extra pass. + if (approvalRequired(resolvedAnnotations, policy) && runtime.plugin.validateToolArgs) { + yield* runtime.plugin + .validateToolArgs({ ctx: runtime.ctx, toolRow: row, args }) + .pipe(wrapInvocationError); + } + yield* enforceApproval(resolvedAnnotations, address, args, policy, handler); + + // Resolve every named credential input (`variable → value`); `value` is + // the primary `token` for single-input + OAuth callers. The + // integration-row read is independent of credential resolution, so + // the two run concurrently. Both start only after + // `enforceApproval` above completes — a declined call must never + // trigger the token refresh credential resolution can perform. + // Credential resolution is the dominant work here and launches + // first: an immediate fork evaluates it inline to its first + // suspension before the plain integration-row fork is even + // scheduled (best-effort — see the launch-order comment at the + // pre-approval forks for why the rare budget-yield reversal is + // accepted and why suppressing it is off the table: credential + // resolution runs extension-owned provider code, and a + // fiber-lifetime yield guard would disable the provider-call + // timeout across it and leak into the detached refresh fork). The + // integration fork is joined after `values`, so a credential + // resolution failure keeps dominating a storage failure exactly as + // it did when the reads were sequential (`Fiber.join` resumes with + // the credential fiber's exact Exit); on that failure path the + // integration fork is interrupted rather than joined, so a hung + // integration read cannot gate the credential error and a failed + // one is deliberately abandoned, never silently dropped as an + // unobserved value. + const valuesFiber = yield* Effect.forkChild(resolveConnectionValues(connectionRow), { + startImmediately: true, + }); + const integrationRowFiber = yield* Effect.forkChild( + findIntegrationRow(parsed.integration), + ); + const values = yield* Fiber.join(valuesFiber).pipe( + Effect.onError(() => Fiber.interrupt(integrationRowFiber)), + ); + const integrationRow = yield* Fiber.join(integrationRowFiber); + const grantedScopes = grantedScopesFromRow(connectionRow); + const invokeTool = runtime.plugin.invokeTool; + const invokeWith = ( + resolved: Record, + ): Effect.Effect => { + const credential: ToolInvocationCredential = { + owner: parsed.owner, integration: parsed.integration, connection: parsed.connection, - toolRows: [row], - }) - .pipe(wrapInvocationError); - resolvedAnnotations = map[String(parsed.tool)] ?? annotations; - } - // When this call is about to pause for approval, validate args - // first: a call that can only fail (missing required path param / - // body) must be rejected here, not after the user grants an approval - // that then goes to waste. Non-pausing calls skip this — invokeTool - // raises the identical failure moments later without the extra pass. - if (approvalRequired(resolvedAnnotations, policy) && runtime.plugin.validateToolArgs) { - yield* runtime.plugin - .validateToolArgs({ ctx: runtime.ctx, toolRow: row, args }) - .pipe(wrapInvocationError); - } - yield* enforceApproval(resolvedAnnotations, address, args, policy, handler); - - // Resolve every named credential input (`variable → value`); `value` is - // the primary `token` for single-input + OAuth callers. - const values = yield* resolveConnectionValues(connectionRow); - const integrationRow = yield* findIntegrationRow(parsed.integration); - const grantedScopes = grantedScopesFromRow(connectionRow); - const invokeTool = runtime.plugin.invokeTool; - const invokeWith = ( - resolved: Record, - ): Effect.Effect => { - const credential: ToolInvocationCredential = { - owner: parsed.owner, - integration: parsed.integration, - connection: parsed.connection, - template: AuthTemplateSlug.make(connectionRow.template), - value: resolved[PRIMARY_INPUT_VARIABLE] ?? null, - values: resolved, - config: integrationRow ? decodeJsonColumn(integrationRow.config) : undefined, - ...(grantedScopes ? { grantedScopes } : {}), + template: AuthTemplateSlug.make(connectionRow.template), + value: resolved[PRIMARY_INPUT_VARIABLE] ?? null, + values: resolved, + config: integrationRow ? decodeJsonColumn(integrationRow.config) : undefined, + ...(grantedScopes ? { grantedScopes } : {}), + }; + return wrapInvocationError( + invokeTool({ + ctx: runtime.ctx, + toolRow: row, + credential, + args, + elicit: buildElicit(address, args, handler), + invokeOptions: options, + }), + ); }; - return wrapInvocationError( - invokeTool({ - ctx: runtime.ctx, - toolRow: row, - credential, - args, - elicit: buildElicit(address, args, handler), - invokeOptions: options, - }), - ); - }; - const first = yield* invokeWith(values); - // Reactive refresh. `expires_at` is only ever the AS's ADVERTISED - // lifetime; the upstream rejecting the token is the authoritative word - // on whether it is still good. The two diverge routinely: server-side - // revocation, an identity provider's idle-timeout policy shorter than - // the token lifetime, and connections whose AS omitted `expires_in` - // entirely (null expiry → the proactive check never fires, so this is - // their ONLY route back to a working token short of a reconnect). - // - // Deliberately narrow: exactly one retry, only on the 401 that means - // "this credential is not valid", and only for a connection holding a - // refresh token. A 403 is excluded — it means authenticated-but-not- - // permitted, and re-minting the same grant returns the same answer. - // If the retry also fails its result stands, so a genuinely dead grant - // still surfaces the upstream's own auth failure and its reconnect - // guidance rather than a masked one. - const { result, usedValues } = yield* Effect.gen(function* () { - if (!isUnauthorizedToolFailure(first)) return { result: first, usedValues: values }; - const refreshed = yield* forceRefreshConnectionValues(connectionRow).pipe( - // A failed re-mint is not this call's failure to report: the upstream - // already produced an auth failure with recovery guidance, which is - // strictly more actionable than a refresh-plumbing error. Keep it. - Effect.catchTag("CredentialResolutionError", () => Effect.succeed(null)), - ); - if (!refreshed) return { result: first, usedValues: values }; - yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.retried": true }); - return { result: yield* invokeWith(refreshed), usedValues: refreshed }; + const first = yield* invokeWith(values); + // Reactive refresh. `expires_at` is only ever the AS's ADVERTISED + // lifetime; the upstream rejecting the token is the authoritative word + // on whether it is still good. The two diverge routinely: server-side + // revocation, an identity provider's idle-timeout policy shorter than + // the token lifetime, and connections whose AS omitted `expires_in` + // entirely (null expiry → the proactive check never fires, so this is + // their ONLY route back to a working token short of a reconnect). + // + // Deliberately narrow: exactly one retry, only on the 401 that means + // "this credential is not valid", and only for a connection holding a + // refresh token. A 403 is excluded — it means authenticated-but-not- + // permitted, and re-minting the same grant returns the same answer. + // If the retry also fails its result stands, so a genuinely dead grant + // still surfaces the upstream's own auth failure and its reconnect + // guidance rather than a masked one. + const { result, usedValues } = yield* Effect.gen(function* () { + if (!isUnauthorizedToolFailure(first)) return { result: first, usedValues: values }; + const refreshed = yield* forceRefreshConnectionValues(connectionRow).pipe( + // A failed re-mint is not this call's failure to report: the upstream + // already produced an auth failure with recovery guidance, which is + // strictly more actionable than a refresh-plumbing error. Keep it. + Effect.catchTag("CredentialResolutionError", () => Effect.succeed(null)), + ); + if (!refreshed) return { result: first, usedValues: values }; + yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.retried": true }); + return { result: yield* invokeWith(refreshed), usedValues: refreshed }; + }); + yield* healPersistedHealthOnUse(connectionRow, result, usedValues); + return result; }); - yield* healPersistedHealthOnUse(connectionRow, result, usedValues); - return result; + // Interrupting an already-completed (or already-joined) fiber is a + // no-op, so this single guard covers every exit path: a path that + // consumed a read leaves a finished fiber behind, and a path that + // exited early — an early-return branch or a tool-row read failure — + // deliberately abandons the reads it never needed instead of waiting + // on them or dropping their failures unobserved. The tool-row fiber + // is settled by the time any branch past its join runs; it is listed + // so an interruption that lands before the join reaches it promptly. + return yield* Effect.ensuring( + invokeDynamicTool, + Fiber.interruptAll([toolRowFiber, policyRulesFiber, connectionRowFiber]), + ); }).pipe( // Expected tool failures (`ToolResult.fail`) resolve through the // success channel, so the tracer alone would record them as healthy