diff --git a/.changeset/connection-create-conflict.md b/.changeset/connection-create-conflict.md new file mode 100644 index 0000000000..7bc40fa492 --- /dev/null +++ b/.changeset/connection-create-conflict.md @@ -0,0 +1,11 @@ +--- +"@executor-js/sdk": patch +--- + +**Creating a connection over an existing one is rejected instead of silently overwriting it** + +`connections.create` used to upsert: a create with the same (owner, integration, name) replaced the saved connection and, for a pasted value, overwrote the stored secret itself. It now fails with the new `ConnectionAlreadyExistsError` and leaves the existing connection untouched. Remove the connection first, or pick a different name. + +This adds one error to the wire contract: the `POST /connections` endpoint can answer **HTTP 409** with tag `ConnectionAlreadyExistsError`, and the `connections.create` core tool resolves the same case as `{ ok: false, error: { code: "connection_already_exists" } }`. The core tool now also resolves the other expected input failures the same way instead of as opaque internal errors: `integration_not_found` for an unknown integration and `invalid_connection_input` for an invalid input. The change is additive — no existing status, field, or success shape moves. + +OAuth is unaffected. Fresh OAuth connects already resolve a taken name to the next free suffix through `newConnection`, and reconnect still re-mints the same connection on purpose. diff --git a/e2e/cloud/connections-credentials.test.ts b/e2e/cloud/connections-credentials.test.ts index d94ce374a8..5f970a0a86 100644 --- a/e2e/cloud/connections-credentials.test.ts +++ b/e2e/cloud/connections-credentials.test.ts @@ -3,9 +3,11 @@ // identified by (owner, integration, name), with its value stored through the // real vault. The product promises under test: the secret goes in but NEVER // comes back out of any endpoint; metadata round-trips; re-creating the same -// connection replaces it instead of duplicating; removal really removes; and -// unknown connections fail with a typed not-found error. +// connection is rejected as a conflict instead of silently replacing it; +// removal really removes; and unknown connections fail with a typed +// not-found error. import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; import { expect } from "@effect/vitest"; import { Effect } from "effect"; @@ -22,7 +24,8 @@ type Client = HttpApiClient.ForApi; const TEMPLATE_API_KEY = AuthTemplateSlug.make("apiKey"); -/** Minimal OpenAPI spec with a single GET /ping — never contacted here. */ +/** Minimal OpenAPI spec with a single GET /ping — only the conflict scenario + * ever contacts it, to prove which stored secret the connection resolves. */ const pingSpec = JSON.stringify({ openapi: "3.0.3", info: { title: "Ping API", version: "1.0.0" }, @@ -33,15 +36,58 @@ const pingSpec = JSON.stringify({ }, }); -/** Registers a fresh apiKey-authenticated integration for connections to bind to. */ -const registerIntegration = (client: Client) => +type CaptureUpstream = { + readonly url: string; + /** Every Authorization header `GET /ping` has received, in order. */ + readonly authorizationHeaders: () => readonly string[]; + readonly close: () => void; +}; + +/** Upstream on 127.0.0.1 that records the Authorization header of every + * `GET /ping`. This is how a scenario proves WHICH stored secret a connection + * resolves, since no endpoint ever echoes the value itself. */ +const serveCaptureUpstream = () => + Effect.acquireRelease( + Effect.callback((resume) => { + const headers: string[] = []; + const server = createServer((request, response) => { + if (request.method === "GET" && (request.url ?? "").startsWith("/ping")) { + headers.push(request.headers.authorization ?? ""); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ pong: true })); + 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}`, + authorizationHeaders: () => [...headers], + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (upstream) => Effect.sync(upstream.close), + ); + +/** Registers a fresh apiKey-authenticated integration for connections to bind + * to. Identifier-safe slug: it becomes a property path in sandbox code. */ +const registerIntegration = (client: Client, baseUrl = "http://127.0.0.1:59999") => Effect.gen(function* () { - const slug = IntegrationSlug.make(`conn-scn-${randomBytes(4).toString("hex")}`); + const slug = IntegrationSlug.make(`connscn${randomBytes(4).toString("hex")}`); yield* client.openapi.addSpec({ payload: { spec: { kind: "blob", value: pingSpec }, slug, - baseUrl: "http://127.0.0.1:59999", // never contacted during registration + baseUrl, // the default is never contacted during registration authenticationTemplate: [ { slug: "apiKey", @@ -102,48 +148,82 @@ scenario( ); scenario( - "Connections · re-creating the same connection replaces it instead of duplicating", + "Connections · re-creating the same connection is rejected and leaves the original intact", {}, - Effect.gen(function* () { - const target = yield* Target; - const { client: apiClient } = yield* Api; - const identity = yield* target.newIdentity(); - const client = yield* apiClient(api, identity); - const integration = yield* registerIntegration(client); - const name = freshConnectionName(); + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: apiClient } = yield* Api; + const identity = yield* target.newIdentity(); + const client = yield* apiClient(api, identity); + const upstream = yield* serveCaptureUpstream(); + const integration = yield* registerIntegration(client, upstream.url); + const name = freshConnectionName(); - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration, - template: TEMPLATE_API_KEY, - identityLabel: "first key", - value: "first-value", - }, - }); - const first = yield* client.connections.list({ query: { integration } }); - expect( - first.filter((connection) => connection.name === name).map((c) => c.identityLabel), - "the first create stores one row with its label", - ).toEqual(["first key"]); + yield* client.connections.create({ + payload: { + owner: "org", + name, + integration, + template: TEMPLATE_API_KEY, + identityLabel: "first key", + value: "first-value", + }, + }); + const first = yield* client.connections.list({ query: { integration } }); + expect( + first.filter((connection) => connection.name === name).map((c) => c.identityLabel), + "the first create stores one row with its label", + ).toEqual(["first key"]); - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration, - template: TEMPLATE_API_KEY, - identityLabel: "rotated key", - value: "second-value", - }, - }); - const second = yield* client.connections.list({ query: { integration } }); - expect( - second.filter((connection) => connection.name === name).map((c) => c.identityLabel), - "re-creating the same (owner, integration, name) updates the row in place", - ).toEqual(["rotated key"]); - }), + const error = yield* client.connections + .create({ + payload: { + owner: "org", + name, + integration, + template: TEMPLATE_API_KEY, + identityLabel: "clobber attempt", + value: "second-value", + }, + }) + .pipe(Effect.flip); + expect( + (error as { _tag?: string })._tag, + "re-creating the same (owner, integration, name) fails with the typed conflict", + ).toBe("ConnectionAlreadyExistsError"); + + const second = yield* client.connections.list({ query: { integration } }); + expect( + second.filter((connection) => connection.name === name).map((c) => c.identityLabel), + "the rejected create left the original row untouched", + ).toEqual(["first key"]); + + // The stored SECRET is intact too, not just the metadata: invoking + // through the original connection must still authenticate upstream with + // the first value. The pasted value's provider item id is derived from + // the connection name, so a rejected create that wrote before losing + // would have replaced the credential while every metadata read above + // still looked untouched. + const tools = yield* client.tools.list({ query: { integration } }); + const address = tools + .map((tool) => String(tool.address)) + .find((toolAddress) => toolAddress.includes(".ping.")); + expect(address, "the ping tool is in the catalog").toBeDefined(); + if (address === undefined) return; + + const execution = yield* client.executions.execute({ + payload: { + code: [`const result = await ${address}({});`, "return result;"].join("\n"), + }, + }); + expect(execution.status, "the invoke completes").toBe("completed"); + expect( + upstream.authorizationHeaders(), + "the original connection still authenticates with the first value", + ).toEqual(["Bearer first-value"]); + }), + ), ); scenario( diff --git a/e2e/scenarios/health-checks.test.ts b/e2e/scenarios/health-checks.test.ts index c4db028d4c..cb8adf53c0 100644 --- a/e2e/scenarios/health-checks.test.ts +++ b/e2e/scenarios/health-checks.test.ts @@ -141,14 +141,20 @@ const discriminatedUnionSpec = (baseUrl: string): string => { }; /** A real node:http identity API on 127.0.0.1. `GET /me` returns the account - * JSON only when the bearer token matches `validToken`; any other token is a - * 401 (the "the dev token got revoked" case the health check classifies as - * expired). Closed by the scope's finalizer. */ + * JSON only when the bearer token matches the CURRENT `validToken`; any other + * token is a 401 (the "the dev token got revoked" case the health check + * classifies as expired). `revoke()` rotates the server-side token so a saved + * key stops working mid-scenario. Closed by the scope's finalizer. */ const serveIdentityApi = (validToken: string) => Effect.acquireRelease( - Effect.callback<{ readonly url: string; readonly close: () => void }>((resume) => { + Effect.callback<{ + readonly url: string; + readonly revoke: () => void; + readonly close: () => void; + }>((resume) => { + let currentToken = validToken; const server = createServer((request, response) => { - const authorized = request.headers["authorization"] === `Bearer ${validToken}`; + const authorized = request.headers["authorization"] === `Bearer ${currentToken}`; if (request.method === "GET" && (request.url ?? "").startsWith("/me")) { if (!authorized) { response.writeHead(401, { "content-type": "application/json" }); @@ -168,6 +174,9 @@ const serveIdentityApi = (validToken: string) => resume( Effect.succeed({ url: `http://127.0.0.1:${port}`, + revoke: () => { + currentToken = `revoked_${validToken}`; + }, close: () => { server.close(); server.closeAllConnections(); @@ -323,17 +332,9 @@ scenario( expect(healthy.httpStatus, "the saved probe saw the 200").toBe(200); expect(healthy.identity, "the saved probe derives the account identity").toBe(IDENTITY); - // Re-creating the same (owner, integration, name) replaces the stored - // key in place: now the connection holds a key the server rejects. - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration: slug, - template: TEMPLATE, - value: "rotated-away", - }, - }); + // The server revokes the key: the connection's saved value now gets + // a 401 (the "dev token got revoked" case). + server.revoke(); const expired = yield* client.connections.checkHealth({ params: { owner: "org", integration: slug, name }, query: {}, @@ -642,22 +643,14 @@ scenario( }, }); - // Seed a verdict, then re-create the connection with a DEAD key: the - // persisted verdict (healthy) and reality (expired) now disagree. + // Seed a verdict, then revoke the key server-side: the persisted + // verdict (healthy) and reality (expired) now disagree. const seeded = yield* client.connections.checkHealth({ params: { owner: "org", integration: slug, name }, query: {}, }); expect(seeded.status, "the seed probe is healthy").toBe("healthy"); - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration: slug, - template: TEMPLATE, - value: "rotated-away", - }, - }); + server.revoke(); // Within the freshness window the CACHED verdict comes back with no // probe, so the dead key still reads healthy (stale by design). diff --git a/e2e/scenarios/no-auth-connection.test.ts b/e2e/scenarios/no-auth-connection.test.ts index 7b47555bfe..2135b88705 100644 --- a/e2e/scenarios/no-auth-connection.test.ts +++ b/e2e/scenarios/no-auth-connection.test.ts @@ -93,6 +93,19 @@ const created = await tools.executor.coreTools.connections.create({ return created.ok ? { ok: true, connection: created.data } : { ok: false, error: created.error }; `; +// Create is never a replace: re-creating the SAME (owner, integration, name) +// must fail with ConnectionAlreadyExistsError instead of silently upserting +// over the existing connection. +const createDuplicateConnectionCode = (slug: string) => ` +const created = await tools.executor.coreTools.connections.create({ + owner: "org", + name: "public", + integration: ${JSON.stringify(slug)}, + template: "none", +}); +return created.ok ? { ok: true, connection: created.data } : { ok: false, error: created.error }; +`; + // The relaxed filter must still reject an origin on a no-auth create — an // empty `inputs: {}` is a (degenerate) origin and a credential the connection // can't hold, so it stays a validation failure. @@ -189,6 +202,19 @@ scenario( rejected.ok, `a no-auth create with an empty inputs origin is rejected: ${JSON.stringify(rejected)}`, ).toBe(false); + + // 5. Create is never a replace: re-creating the same (owner, + // integration, name) fails with a conflict instead of silently + // editing over the existing connection. + const duplicate = yield* executeJson(session, createDuplicateConnectionCode(integration)); + expect( + duplicate.ok, + `re-creating an existing connection name is rejected: ${JSON.stringify(duplicate)}`, + ).toBe(false); + expect( + JSON.stringify(duplicate.error), + "the failure names the conflict so callers can act on it", + ).toContain("already exists"); }).pipe( // Selfhost shares one workspace identity — leaked connections fail other // scenarios' zero-state assertions, so drop everything this run made. diff --git a/e2e/selfhost/connection-duplicate-name-demo.test.ts b/e2e/selfhost/connection-duplicate-name-demo.test.ts new file mode 100644 index 0000000000..e02c432ab5 --- /dev/null +++ b/e2e/selfhost/connection-duplicate-name-demo.test.ts @@ -0,0 +1,89 @@ +// Demo recording (AFTER the fix): adding a second connection without typing a +// name derives the same default name as the first — the create is REJECTED +// with the conflict error and the original connection survives untouched. +// Video is the artifact. +import { randomBytes } from "node:crypto"; + +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 { IntegrationSlug } from "@executor-js/sdk/shared"; + +import { scenario } from "../src/scenario"; +import { Api, Browser, Target } from "../src/services"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const bearerSpec = (): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Bearer Fixture", version: "1.0.0" }, + servers: [{ url: "https://api.bearerfix.test" }], + security: [{ bearerAuth: [] }], + components: { securitySchemes: { bearerAuth: { type: "http", scheme: "bearer" } } }, + paths: { + "/ping": { get: { operationId: "ping", responses: { "200": { description: "ok" } } } }, + }, + }); + +scenario( + "Connections · a second connection with the default name is rejected, not overwritten", + {}, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeApiClient } = yield* Api; + const browser = yield* Browser; + const identity = yield* target.newIdentity(); + const apiClient = yield* makeApiClient(api, identity); + const slug = `dup_name_demo_${randomBytes(4).toString("hex")}`; + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* apiClient.openapi.addSpec({ + payload: { spec: { kind: "blob", value: bearerSpec() }, slug }, + }); + + yield* browser.session(identity, async ({ page, step }) => { + const addConnection = async (key: string) => { + await page.getByRole("button", { name: "Add connection" }).first().click(); + await page.getByRole("heading", { name: /Add connection/ }).waitFor(); + const dialog = page.getByRole("dialog", { name: /Add connection/ }); + await dialog.locator('input[type="password"]').first().fill(key); + await dialog.getByRole("button", { name: "Continue" }).click(); + // Leave the display name empty: the default derives the SAME + // connection name both times. + await dialog.getByRole("button", { name: "Add connection" }).click(); + }; + + await step("Add the first connection with the default name", async () => { + await page.goto(`/integrations/${slug}`, { waitUntil: "networkidle" }); + await page.getByText("Connections").first().waitFor(); + await addConnection("first-key"); + await page.getByText("Connection added").waitFor(); + }); + + await step("Add a second connection, also leaving the name empty", async () => { + await page.waitForTimeout(1_000); // let the first toast clear + await addConnection("second-key"); + }); + + await step("The duplicate is rejected with the conflict error", async () => { + await page.getByText(/already exists/).waitFor(); + await page.waitForTimeout(2_000); // hold the error on screen + }); + }); + + const connections = yield* apiClient.connections.list({ + query: { integration: IntegrationSlug.make(slug) }, + }); + expect(connections.length, "the original connection is the only one").toBe(1); + }), + apiClient.openapi + .removeSpec({ params: { slug: IntegrationSlug.make(slug) } }) + .pipe(Effect.ignore), + ); + }), + ), +); diff --git a/packages/core/api/src/connections/api.ts b/packages/core/api/src/connections/api.ts index c93e983cb3..dff1f65312 100644 --- a/packages/core/api/src/connections/api.ts +++ b/packages/core/api/src/connections/api.ts @@ -13,6 +13,7 @@ import { Predicate, Schema } from "effect"; import { AuthTemplateSlug, ConnectionAddress, + ConnectionAlreadyExistsError, ConnectionName, ConnectionNotFoundError, CredentialProviderNotRegisteredError, @@ -167,6 +168,9 @@ const ConnectionNotFound = ConnectionNotFoundError.annotate({ const IntegrationNotFound = IntegrationNotFoundError.annotate({ httpApiStatus: 404, }); +const ConnectionAlreadyExists = ConnectionAlreadyExistsError.annotate({ + httpApiStatus: 409, +}); const CredentialProviderNotRegistered = CredentialProviderNotRegisteredError.annotate({ httpApiStatus: 409, }); @@ -193,6 +197,7 @@ export const ConnectionsApi = HttpApiGroup.make("connections") error: [ InternalError, IntegrationNotFound, + ConnectionAlreadyExists, CredentialProviderNotRegistered, InvalidConnectionInput, ], diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 2caf2195de..eb8a52ba88 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "@effect/vitest"; import { + Cause, Deferred, Effect, Exit, @@ -23,6 +24,7 @@ import { ToolAddress, ToolName, } from "./ids"; +import { ConnectionAlreadyExistsError } from "./errors"; import { createExecutor } from "./executor"; import { StorageError, type FumaDb } from "./fuma-runtime"; import { HealthCheckResult } from "./health-check"; @@ -178,6 +180,269 @@ describe("connections.create", () => { }), ); + // Create is never a replace: a second create with the same (owner, + // integration, name) must fail with ConnectionAlreadyExistsError and leave + // the first connection fully intact — including its stored secret, which a + // silent upsert would overwrite (the pasted value's item id is derived from + // the name, so the provider write alone clobbers it). + it.effect("rejects a duplicate (owner, integration, name) and keeps the original intact", () => + Effect.gen(function* () { + const executor = yield* setup(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "original-token", + description: "original", + }); + + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "clobbered-token", + description: "clobbered", + }), + ); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ConnectionAlreadyExistsError")(result.failure)).toBe(true); + + // The original row and its secret both survived. + const connections = yield* executor.connections.list(); + expect(connections.length).toBe(1); + expect(connections[0]?.description).toBe("original"); + const value = yield* executor.demo.resolveValue("org", "main"); + expect(value).toBe("original-token"); + }), + ); + + // Names collide AFTER identifier normalization: "my-api-key" and "my api key" + // both normalize to myApiKey, so the second must be rejected even though the + // raw inputs differ. + it.effect("rejects a duplicate that only collides after name normalization", () => + Effect.gen(function* () { + const executor = yield* setup(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("my-api-key"), + integration: INTEG, + template: TEMPLATE, + value: "v1", + }); + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("my api key"), + integration: INTEG, + template: TEMPLATE, + value: "v2", + }), + ); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ConnectionAlreadyExistsError")(result.failure)).toBe(true); + const value = yield* executor.demo.resolveValue("org", "myApiKey"); + expect(value).toBe("v1"); + }), + ); + + // The race the early duplicate check cannot answer: two creates for the same + // (owner, integration, name) in flight at once. The row insert picks the + // winner and the loser gets the typed 409 — and, the load-bearing part, the + // provider write is winner-only. A pasted value's item id is deterministic, + // so a pre-insert write from the LOSING create would silently replace the + // winner's stored secret while the winner's row keeps resolving through it. + // The gate parks the first create inside its provider write, so the second + // create runs its full duplicate handling while the first is mid-flight. + it.effect("concurrent creates: one winner, a typed 409, and the winner's secret intact", () => + Effect.scoped( + Effect.gen(function* () { + const firstWriteEntered = yield* Deferred.make(); + const releaseFirstWrite = yield* Deferred.make(); + const store = new Map(); + let writes = 0; + const gatedProvider: CredentialProvider = { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => + Effect.gen(function* () { + writes += 1; + if (writes === 1) { + yield* Deferred.succeed(firstWriteEntered, undefined); + yield* Deferred.await(releaseFirstWrite); + } + store.set(String(id), value); + }), + }; + const gatedPlugin = definePlugin(() => ({ + id: "gated" as const, + credentialProviders: [gatedProvider], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ + tools: [{ name: ToolName.make("deploy"), description: "deploy" }], + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + resolveValue: (name: string) => + ctx.connections.resolveValue({ + owner: "org", + integration: INTEG, + name: ConnectionName.make(name), + }), + }), + }))(); + const config = makeTestConfig({ plugins: [gatedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.gated.seed(); + + const createWith = (value: string) => + Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value, + }), + ); + + const firstFiber = yield* Effect.forkChild(createWith("first-value")); + yield* Deferred.await(firstWriteEntered); + const second = yield* createWith("second-value"); + yield* Deferred.succeed(releaseFirstWrite, undefined); + const first = yield* Fiber.join(firstFiber); + + const attempts = [ + { result: first, value: "first-value" }, + { result: second, value: "second-value" }, + ]; + const winners = attempts.filter((attempt) => Result.isSuccess(attempt.result)); + const losers = attempts.filter((attempt) => Result.isFailure(attempt.result)); + expect(winners).toHaveLength(1); + expect(losers).toHaveLength(1); + const loser = losers[0]; + if (!loser || !Result.isFailure(loser.result)) return; + expect(loser.result.failure).toBeInstanceOf(ConnectionAlreadyExistsError); + + // Exactly one connection survived, and it resolves to the WINNER's + // value — the losing create never reached the provider. + const connections = yield* executor.connections.list(); + expect(connections).toHaveLength(1); + const value = yield* executor.gated.resolveValue("main"); + expect(value).toBe(winners[0]?.value); + }), + ), + ); + + // When both creates observe absence, both reach the insert and the primary + // key breaks the tie — the loser must still get the typed 409, not a raw + // unique-constraint storage failure. The proxy blinds every connection-table + // read for the second create, so its early and transactional checks both + // miss the existing row and its insert genuinely collides in the database. + it.effect("maps a lost insert race to the typed 409, not a storage failure", () => + Effect.gen(function* () { + let blind = false; + const blindfoldConnectionReads = (db: FumaDb): 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 === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => + blind && table === "connection" + ? Promise.resolve(null) + : (target.findFirst as (t: unknown, q: unknown) => Promise)( + table, + query, + ); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); + }; + + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: blindfoldConnectionReads(config.db), + }); + yield* executor.demo.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "first-value", + }); + + blind = true; + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "second-value", + }), + ); + blind = false; + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(result.failure).toBeInstanceOf(ConnectionAlreadyExistsError); + + // The losing insert wrote nothing: the original secret is untouched. + const connections = yield* executor.connections.list(); + expect(connections).toHaveLength(1); + const value = yield* executor.demo.resolveValue("org", "main"); + expect(value).toBe("first-value"); + }), + ); + + it.effect("allows the same name under a different owner", () => + Effect.gen(function* () { + const executor = yield* setup(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "org-token", + }); + const personal = yield* executor.connections.create({ + owner: "user", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "user-token", + }); + expect(String(personal.address)).toBe("tools.vercel.user.main"); + expect((yield* executor.connections.list()).length).toBe(2); + }), + ); + it.effect("external `from` references a provider item without writing it", () => Effect.gen(function* () { const executor = yield* setup(); @@ -315,6 +580,853 @@ describe("connections.create", () => { ); }); +// --------------------------------------------------------------------------- +// Credential-write compensation. The row insert and the provider write cannot +// be atomic — the provider may live outside the database — so the create +// sequences them: the provider is touched only after this create wins the row +// insert, and a write that does not complete must tear down everything it +// already stored. The worst outcome is a committed row whose credentials were +// never written: it 409s every retry while resolving nothing. +// --------------------------------------------------------------------------- + +const trackingProvider = ( + store: Map, + overrides?: Partial>, +): CredentialProvider => ({ + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + delete: (id) => Effect.sync(() => void store.delete(String(id))), + ...overrides, +}); + +const durabilityPlugin = (provider: CredentialProvider) => + definePlugin(() => ({ + id: "durable" as const, + credentialProviders: [provider], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), + }), + }))(); + +/** Wrap a test `FumaDb` so deletes on the `connection` table can be made to + * fail on demand — the raw driver-level failure the compensating delete must + * survive loudly. A rejection during the delete statement is ambiguous on an + * auto-commit adapter (the statement may have executed first), so this + * failure is reported as unconfirmed, never as definitively stranded. + * Transactions hand out wrapped handles too, so the guarded delete inside + * the compensation transaction is covered. */ +const failableConnectionDeletes = (db: FumaDb, shouldFail: () => boolean): 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 === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "deleteMany") { + return (table: unknown, query: unknown) => + shouldFail() && table === "connection" + ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + Promise.reject(new StorageError({ message: "delete refused", cause: undefined })) + : (target.deleteMany as (t: unknown, q: unknown) => Promise)(table, query); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + +/** Wrap a test `FumaDb` so compensation fails strictly BEFORE its guarded + * delete statement is issued: the identity read that opens the compensation + * transaction rejects. Only a pre-attempt failure keeps the definitive + * stranded-row claim truthful — a rejection during the delete statement + * itself is reported as unconfirmed instead (see + * `failableConnectionDeletes`). The read is identified by sequence: the + * first `connection` read after this create's row insert. The conflict-check + * read runs before the insert and no other `connection` read happens in + * between, so that read is compensation's. Transactions hand out wrapped + * handles too. */ +const failableCompensationRowDelete = (db: FumaDb, shouldFail: () => boolean): FumaDb => { + let insertSeen = false; + 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 === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "create") { + return async (table: unknown, values: unknown) => { + const row = await (target.create as (t: unknown, v: unknown) => Promise)( + table, + values, + ); + if (table === "connection") insertSeen = true; + return row; + }; + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => + shouldFail() && insertSeen && table === "connection" + ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + Promise.reject( + new StorageError({ message: "identity read refused", cause: undefined }), + ) + : (target.findFirst as (t: unknown, q: unknown) => Promise)(table, query); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + +/** Wrap a test `FumaDb` so one armed `connection` read observes a stale row. + * This models the read/delete race inside the compensation transaction: under + * read-committed isolation the pre-delete identity read can see this create's + * row while a concurrent remove/recreate has already replaced it by the time + * the guarded delete runs. SQLite serializes the whole transaction, so that + * interleaving cannot be produced with real concurrency here — the wrapper + * reproduces the exact observation order instead: the armed read returns the + * create's own (captured) row while the table already holds the successor; + * every other read, including the confirmation read after the guarded + * delete, sees the real table. */ +const staleCompensationRead = (db: FumaDb, state: { armed: boolean }): FumaDb => { + let captured: Record | null = null; + 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 === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "create") { + return async (table: unknown, values: unknown) => { + const row = await ( + target.create as (t: unknown, v: unknown) => Promise> + )(table, values); + // Keep the FIRST inserted connection row — the raced create's own. + if (table === "connection" && captured === null) captured = row; + return row; + }; + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => { + if (table === "connection" && state.armed && captured !== null) { + state.armed = false; + return Promise.resolve(captured); + } + return (target.findFirst as (t: unknown, q: unknown) => Promise)(table, query); + }; + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + +/** Wrap a test `FumaDb` so the confirmation read AFTER the guarded delete + * fails at the driver. While armed, the first `connection` read that follows + * a `connection` delete rejects; every other statement passes through. + * Transactions hand out wrapped handles too, so the read inside the + * compensation transaction is covered. */ +const failableConfirmationRead = (db: FumaDb, state: { armed: boolean }): FumaDb => { + let deleteSeen = false; + 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 === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "deleteMany") { + return (table: unknown, query: unknown) => { + if (state.armed && table === "connection") deleteSeen = true; + return (target.deleteMany as (t: unknown, q: unknown) => Promise)( + table, + query, + ); + }; + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => { + if (state.armed && deleteSeen && table === "connection") { + state.armed = false; + deleteSeen = false; + // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + return Promise.reject( + new StorageError({ message: "confirmation read refused", cause: undefined }), + ); + } + return (target.findFirst as (t: unknown, q: unknown) => Promise)(table, query); + }; + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + +describe("connections.create credential-write compensation", () => { + // Interruption is not an error: error-channel compensation never sees it. A + // create interrupted mid-write must still tear down what it already did — + // the committed row and every item that landed before the interrupt. + it.effect("an interrupted create removes the row and the items it already wrote", () => + Effect.scoped( + Effect.gen(function* () { + const secondWriteEntered = yield* Deferred.make(); + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Deferred.succeed(secondWriteEntered, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.sync(() => void store.set(String(id), value)), + }); + const executor = yield* makeTestExecutor({ + plugins: [durabilityPlugin(provider)] as const, + }); + yield* executor.durable.seed(); + + const fiber = yield* Effect.forkChild( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + yield* Deferred.await(secondWriteEntered); + yield* Fiber.interrupt(fiber); + + // The first item had landed before the interrupt; compensation removed + // it together with the row it belonged to. + expect(store.size).toBe(0); + expect(yield* executor.connections.list()).toEqual([]); + }), + ), + ); + + // One provider.set can succeed and a later one fail. Deleting only the row + // leaves the earlier secret at its deterministic item id, waiting to be + // adopted by the next create of the same name. Compensation must remove the + // items already written, not just the row. + it.effect("a failed later variable write cleans up the earlier items and the row", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const }); + yield* executor.durable.seed(); + + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true); + // Neither half survives: the first item is gone with the row. + expect(store.size).toBe(0); + expect(yield* executor.connections.list()).toEqual([]); + }), + ); + + // A provider can expose `set` without `delete`. Compensation then cannot + // undo the items already written — that is acceptable only if it is loud: + // a warning must name the item that may be stranded, never a silent skip. + it.effect("warns about possibly stranded items when the provider has no delete", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + delete: undefined, + }); + const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const }); + yield* executor.durable.seed(); + + const warnings: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Warn") { + warnings.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true); + // The row is gone, but the first item cannot be undone without a + // provider delete ... + expect(yield* executor.connections.list()).toEqual([]); + expect(store.size).toBe(1); + // ... and the create said so, naming the item. + expect(warnings.some((line) => line.includes("stranded"))).toBe(true); + expect(warnings.some((line) => line.includes("first"))).toBe(true); + }), + ); + + // The compensating delete can fail before its guarded delete statement is + // even issued (here: the identity read that opens the compensation + // transaction rejects). Nothing can have been deleted, so the surviving + // credential-less row is definitively stranded — and swallowing that + // failure hides it behind an error that never mentions it. The create must + // fail with an error that NAMES the stranded connection so an operator can + // act on it. + it.effect("names the stranded connection when the compensating delete fails", () => + Effect.gen(function* () { + let failRowDelete = false; + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableCompensationRowDelete(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + failRowDelete = true; + + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + const failure = result.failure; + expect(Predicate.isTagged("StorageError")(failure)).toBe(true); + if (!Predicate.isTagged("StorageError")(failure)) return; + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + // Pre-attempt failure: the stranded claim is definitive and stated. + expect(failure.message).toContain("stranded"); + // The original write failure is retained as the cause, not replaced. + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure.cause)).toBe(true); + if (!isStorageError(failure.cause)) return; + expect(failure.cause.message).toBe("provider write refused"); + + // Non-vacuous: the compensating delete really did fail, so the + // row the error names is still there. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + }), + ); + + // Compensation can be slow (provider calls). In that window a concurrent + // remove can free the name and a new create can take it, writing fresh + // secrets at the SAME deterministic item ids. Late compensation must then + // recognize that the row is no longer the one it inserted — identified by + // the storage surrogate row id — and touch neither the replacement row nor + // its credentials. Losing compensation to a concurrent remove is correct: + // the remover already cleaned up. + it.effect("late compensation leaves a concurrent replacement untouched", () => + Effect.scoped( + Effect.gen(function* () { + const secondWriteEntered = yield* Deferred.make(); + const releaseSecondWrite = yield* Deferred.make(); + const store = new Map(); + let parkNextSecondWrite = true; + const provider = trackingProvider(store, { + set: (id, value) => { + if (String(id).endsWith(":second") && parkNextSecondWrite) { + parkNextSecondWrite = false; + return Deferred.succeed(secondWriteEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseSecondWrite)), + Effect.andThen( + Effect.fail( + new StorageError({ message: "provider write refused", cause: undefined }), + ), + ), + ); + } + return Effect.sync(() => void store.set(String(id), value)); + }, + }); + const executor = yield* makeTestExecutor({ + plugins: [durabilityPlugin(provider)] as const, + }); + yield* executor.durable.seed(); + + const fiber = yield* Effect.forkChild( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "a-1", second: "a-2" }, + }), + ); + yield* Deferred.await(secondWriteEntered); + + // While the first create is parked in its provider write, the user + // removes the connection and recreates it with different secrets. + yield* executor.connections.remove({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "c-1", second: "c-2" }, + }); + + // Release the parked write: the first create fails and compensates + // late, against a name it no longer owns. + yield* Deferred.succeed(releaseSecondWrite, undefined); + const exit = yield* Fiber.await(fiber); + expect(Exit.isFailure(exit)).toBe(true); + + // The replacement row AND its credentials survive. + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + expect(store.get("connection:org:vercel:main:first")).toBe("c-1"); + expect(store.get("connection:org:vercel:main:second")).toBe("c-2"); + }), + ), + ); + + // fumadb's `deleteMany` returns void, so the guarded delete cannot report + // whether it removed anything. Under read-committed isolation the identity + // read and the delete can straddle a concurrent remove/recreate: the read + // sees this create's row, then the delete matches ZERO rows because a + // successor already holds the name. Treating that zero-row delete as "our + // row is gone, the items are ours to undo" destroys the successor's freshly + // written secrets at the same deterministic item ids. The confirmation read + // after the guarded delete, in the same transaction, must observe the + // surviving row, skip ALL item deletion, and say so. + it.effect("a zero-row guarded delete never touches a successor's credentials", () => + Effect.scoped( + Effect.gen(function* () { + const secondWriteEntered = yield* Deferred.make(); + const releaseSecondWrite = yield* Deferred.make(); + const store = new Map(); + let parkNextSecondWrite = true; + const provider = trackingProvider(store, { + set: (id, value) => { + if (String(id).endsWith(":second") && parkNextSecondWrite) { + parkNextSecondWrite = false; + return Deferred.succeed(secondWriteEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseSecondWrite)), + Effect.andThen( + Effect.fail( + new StorageError({ message: "provider write refused", cause: undefined }), + ), + ), + ); + } + return Effect.sync(() => void store.set(String(id), value)); + }, + }); + const raceState = { armed: false }; + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: staleCompensationRead(config.db, raceState), + }); + yield* executor.durable.seed(); + + const infos: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Info") { + infos.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const fiber = yield* Effect.forkChild( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "a-1", second: "a-2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + yield* Deferred.await(secondWriteEntered); + + // While the first create is parked in its provider write, the user + // removes the connection and recreates it with different secrets. + yield* executor.connections.remove({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "c-1", second: "c-2" }, + }); + + // Arm the stale read and release the parked write: compensation's + // identity read sees the raced create's own row (the race), its + // guarded delete then removes zero rows. + raceState.armed = true; + yield* Deferred.succeed(releaseSecondWrite, undefined); + const exit = yield* Fiber.await(fiber); + expect(Exit.isFailure(exit)).toBe(true); + + // The successor row AND its credentials survive untouched, and the + // skip was reported, not silent. + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + expect(store.get("connection:org:vercel:main:first")).toBe("c-1"); + expect(store.get("connection:org:vercel:main:second")).toBe("c-2"); + expect(infos.some((line) => line.includes("removed nothing"))).toBe(true); + }), + ), + ); + + // The confirmation read after the guarded delete can itself fail. On an + // interactive adapter that failure rolls the delete back with the + // transaction, so "stranded" would be truthful — but on an auto-commit + // adapter (Cloudflare D1 runs `interactiveTransactions: false`) every + // statement commits immediately: the delete has already removed the row + // when the read fails, and a stranded-row claim would be false. The row + // state is genuinely unknown at this layer, so the create must say exactly + // that — skip ALL credential-item deletion and report the row as + // unconfirmed, never as stranded. + it.effect("a failed confirmation read skips item cleanup and reports the row unconfirmed", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const readState = { armed: false }; + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableConfirmationRead(config.db, readState), + }); + yield* executor.durable.seed(); + readState.armed = true; + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + + // Non-vacuous: the armed rejection fired, so the statement that failed + // really was the confirmation read, after the guarded delete ran. + expect(readState.armed).toBe(false); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + const failure = result.failure; + expect(Predicate.isTagged("StorageError")(failure)).toBe(true); + if (!Predicate.isTagged("StorageError")(failure)) return; + // The error names the connection and reports the unconfirmed state; it + // must NOT claim the row is stranded — on an auto-commit adapter the + // delete already committed and the row is gone. + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + expect(failure.message).toContain("could not be confirmed"); + expect(failure.message).not.toContain("stranded"); + // The original write failure is retained as the cause, not replaced. + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure.cause)).toBe(true); + if (!isStorageError(failure.cause)) return; + expect(failure.cause.message).toBe("provider write refused"); + + // The unknown outcome skips ALL credential-item deletion: the item that + // landed before the failed write is untouched. + expect(store.get("connection:org:vercel:main:first")).toBe("1"); + // The log reports the unconfirmed state, not a stranded-row claim. + expect(errors.some((line) => line.includes("could not confirm"))).toBe(true); + expect(errors.every((line) => !line.includes("stranded a connection row"))).toBe(true); + }), + ); + + // The guarded delete STATEMENT can itself reject. On an interactive adapter + // the rejection rolls the transaction back and the row survives — but on an + // auto-commit adapter (Cloudflare D1) the statement may have executed + // before the rejection surfaced, so a definitive stranded-row claim would + // be false. Once the delete has been attempted, the row state is genuinely + // unknown at this layer: the create must skip ALL credential-item deletion + // and report the delete as unconfirmed, never as stranded. + it.effect("a rejected delete statement skips item cleanup and reports the row unconfirmed", () => + Effect.gen(function* () { + let failRowDelete = false; + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableConnectionDeletes(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + failRowDelete = true; + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + const failure = result.failure; + expect(Predicate.isTagged("StorageError")(failure)).toBe(true); + if (!Predicate.isTagged("StorageError")(failure)) return; + // The error names the connection and reports the unconfirmed state; it + // must NOT claim the row is stranded — on an auto-commit adapter the + // delete may already have executed before the rejection surfaced. + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + expect(failure.message).toContain("could not be confirmed"); + expect(failure.message).not.toContain("stranded"); + // The original write failure is retained as the cause, not replaced. + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure.cause)).toBe(true); + if (!isStorageError(failure.cause)) return; + expect(failure.cause.message).toBe("provider write refused"); + + // The unknown outcome skips ALL credential-item deletion: the item that + // landed before the failed write is untouched. + expect(store.get("connection:org:vercel:main:first")).toBe("1"); + // The log reports the unconfirmed state, not a stranded-row claim. + expect(errors.some((line) => line.includes("could not confirm"))).toBe(true); + expect(errors.every((line) => !line.includes("stranded a connection row"))).toBe(true); + + // Non-vacuous: the armed proxy rejected the delete statement without + // running it, so on this interactive adapter the row survived. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + }), + ); + + // A provider write can die with a defect instead of failing. The stranded- + // row promise must hold there too: a defect followed by a compensating + // delete that fails before its delete statement is issued surfaces the + // same typed StorageError naming the stranded connection, not an anonymous + // crash. + it.effect("a defect followed by a failed row delete still names the stranded connection", () => + Effect.gen(function* () { + let failRowDelete = false; + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.die("provider crashed") + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableCompensationRowDelete(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + failRowDelete = true; + + const exit = yield* Effect.exit( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (!Exit.isFailure(exit)) return; + // Not an anonymous defect: the typed error is on the failure channel. + expect(Cause.hasFails(exit.cause)).toBe(true); + const failure = Cause.squash(exit.cause); + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure)).toBe(true); + if (!isStorageError(failure)) return; + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + expect(failure.message).toContain("stranded"); + // The original defect is retained as the cause, not replaced. + expect(failure.cause).toBe("provider crashed"); + + // Non-vacuous: the row the error names is still there. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + }), + ); + + // Interruption cannot carry a typed error — interrupting wins over failing + // — so when an interrupted create cannot delete its row (compensation + // fails before the delete statement is issued, leaving the row + // definitively stranded), the stranded row is reported through a loud + // error log and the create stays an interruption. The items that landed + // before the interrupt stay with the stranded row: credential teardown is + // gated on the row delete succeeding. + it.effect("an interrupted create with a failed row delete logs the stranded row", () => + Effect.scoped( + Effect.gen(function* () { + let failRowDelete = false; + const secondWriteEntered = yield* Deferred.make(); + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Deferred.succeed(secondWriteEntered, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableCompensationRowDelete(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const fiber = yield* Effect.forkChild( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + yield* Deferred.await(secondWriteEntered); + failRowDelete = true; + yield* Fiber.interrupt(fiber); + const exit = yield* Fiber.await(fiber); + + // Still an interruption — and the stranded row was reported loudly. + expect(Exit.isFailure(exit)).toBe(true); + if (!Exit.isFailure(exit)) return; + expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true); + expect(errors.some((line) => line.includes("stranded"))).toBe(true); + + // Non-vacuous: the row survived the failed delete, and the item that + // landed before the interrupt stayed with it. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + expect(store.size).toBe(1); + }), + ), + ); +}); + describe("connections.list / get", () => { it.effect("only includes full health diagnostics in verbose core tool output", () => Effect.gen(function* () { diff --git a/packages/core/sdk/src/core-tools.ts b/packages/core/sdk/src/core-tools.ts index 0618ef86dd..82e2b8fe5e 100644 --- a/packages/core/sdk/src/core-tools.ts +++ b/packages/core/sdk/src/core-tools.ts @@ -25,6 +25,7 @@ import { definePlugin, tool, type StaticToolSchema } from "./plugin"; import { HealthCheckResult, isToolSyncHealth } from "./health-check"; import { ToolPolicyActionSchema } from "./policies"; import type { Tool } from "./tool"; +import { ToolResult } from "./tool-result"; const schemaToStandard = (schema: Schema.Decoder): StaticToolSchema => Schema.toStandardSchemaV1(Schema.toStandardJSONSchemaV1(schema) as never) as StaticToolSchema< @@ -670,7 +671,7 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { tool({ name: "connections.create", description: - 'Low-level create or replace for a saved connection from provider item references. For a no-auth integration (public MCP server, public REST API), pass `template: "none"` with no `from`/`inputs` to wire it up directly. For normal API keys/tokens, use `connections.createHandoff` so the user enters the credential in the web UI. OAuth credentials should use `oauth.start`.', + 'Low-level create for a saved connection from provider item references. Fails if a connection with the same owner, integration, and name already exists (remove it first, or pick a different name). For a no-auth integration (public MCP server, public REST API), pass `template: "none"` with no `from`/`inputs` to wire it up directly. For normal API keys/tokens, use `connections.createHandoff` so the user enters the credential in the web UI. OAuth credentials should use `oauth.start`.', inputSchema: ConnectionCreateInputStd, outputSchema: ConnectionOutputStd, // Creating a connection binds a credential reference and roots a new @@ -681,9 +682,25 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { // approval-gated (the v1 `sources.configure` carried the same guard). annotations: { requiresApproval: true }, execute: (input: typeof ConnectionCreateInput.Type, { ctx }) => - Effect.map( - ctx.connections.create(createConnectionInputFromTool(input)), - connectionToOutput, + ctx.connections.create(createConnectionInputFromTool(input)).pipe( + Effect.map(connectionToOutput), + // Expected, caller-actionable failures resolve as ToolResult.fail + // (the sandbox sees `{ ok: false, error }`); anything else stays + // a defect and surfaces as the opaque internal-error generic. + Effect.catchTags({ + ConnectionAlreadyExistsError: (error) => + Effect.succeed( + ToolResult.fail({ code: "connection_already_exists", message: error.message }), + ), + IntegrationNotFoundError: (error) => + Effect.succeed( + ToolResult.fail({ code: "integration_not_found", message: error.message }), + ), + InvalidConnectionInputError: (error) => + Effect.succeed( + ToolResult.fail({ code: "invalid_connection_input", message: error.message }), + ), + }), ), }), tool({ diff --git a/packages/core/sdk/src/errors.ts b/packages/core/sdk/src/errors.ts index 4e8927ec04..9922287f76 100644 --- a/packages/core/sdk/src/errors.ts +++ b/packages/core/sdk/src/errors.ts @@ -162,6 +162,26 @@ export class ConnectionNotFoundError extends Schema.TaggedErrorClass()( + "ConnectionAlreadyExistsError", + { + owner: Owner, + integration: IntegrationSlug, + name: ConnectionName, + }, + { httpApiStatus: 409 }, +) { + override get message(): string { + return `A connection named "${this.name}" already exists for ${this.integration} (${this.owner}). Choose a different name, or remove the existing connection first.`; + } +} + /** A connection create request was rejected before anything was written: the * input is structurally invalid (no credential inputs for a credentialed * template, mixed pasted/external origins, …) or targets owner `user` in a diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index b24a286eae..d8afd12d23 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,12 +1,15 @@ import { + Cause, Deferred, Duration, Effect, + Exit, Fiber, Inspectable, Layer, Option, Predicate, + Ref, Schema, Semaphore, } from "effect"; @@ -81,6 +84,7 @@ import { } from "./artifact"; import { ArtifactNotFoundError, + ConnectionAlreadyExistsError, ConnectionNotFoundError, CredentialProviderNotRegisteredError, CredentialResolutionError, @@ -385,6 +389,7 @@ export type Executor = { ) => Effect.Effect< Connection, | IntegrationNotFoundError + | ConnectionAlreadyExistsError | CredentialProviderNotRegisteredError | InvalidConnectionInputError | StorageFailure @@ -3437,6 +3442,7 @@ export const createExecutor = = {}; + // Pasted-value provider writes, built here but run only AFTER this + // create wins the row insert below. Each entry carries its own undo + // so a write that does not complete can tear down exactly the items + // it already stored. + const pastedWrites: Array<{ + readonly itemId: ProviderItemId; + readonly write: Effect.Effect; + readonly remove: Effect.Effect | null; + }> = []; if (external.length > 0 && pasted.length > 0) { return yield* new InvalidConnectionInputError({ message: "A connection cannot mix pasted and external-provider inputs.", @@ -3519,8 +3552,17 @@ export const createExecutor = storageFailureFromUnknown("invalid owner", cause), }); const now = new Date(); - yield* transaction( + // The storage surrogate id of the row THIS create inserted. The + // composite key (owner, integration, name) can change hands while a + // failed create is still compensating, and `created_at` round-trips + // at second precision, so neither identifies OUR row — only `row_id` + // does. `FumaRow` deliberately hides `row_id` from domain rows, so it + // is read through a narrow cast. Every adapter generates it ORM-side + // on insert; a create result without it is a broken storage contract + // and fails here, inside the transaction, before any provider write. + const rowIdOf = (row: unknown): string | null => { + const value = row == null ? null : (row as Record)["row_id"]; + return typeof value === "string" ? value : null; + }; + const insertedRowId = yield* transaction( Effect.gen(function* () { const existing = yield* findConnectionRow({ owner: input.owner, integration: input.integration, name, }); - const set: Record = { + if (existing) { + return yield* new ConnectionAlreadyExistsError({ + owner: input.owner, + integration: input.integration, + name, + }); + } + const inserted = yield* core.create("connection", { + tenant: keys.tenant, + owner: keys.owner, + subject: keys.subject, + integration: String(input.integration), + name: String(name), template: String(input.template), provider: providerKey, item_ids: itemIds, identity_label: input.identityLabel ?? null, - // Re-saving a credential keeps an existing curated description - // unless the caller explicitly provides one. - ...(input.description !== undefined ? { description: input.description } : {}), + description: input.description ?? null, + oauth_client: null, + refresh_item_id: null, + expires_at: null, + oauth_scope: null, + provider_state: null, + created_at: now, updated_at: now, - }; - if (existing) { - yield* core.updateMany("connection", { - where: (b: AnyCb) => - b.and( - byOwner(input.owner)(b), - b("integration", "=", String(input.integration)), - b("name", "=", String(name)), - ), - set, - }); - } else { - yield* core.create("connection", { - tenant: keys.tenant, - owner: keys.owner, - subject: keys.subject, - integration: String(input.integration), - name: String(name), - template: String(input.template), - provider: providerKey, - item_ids: itemIds, - identity_label: input.identityLabel ?? null, - description: input.description ?? null, - oauth_client: null, - refresh_item_id: null, - expires_at: null, - oauth_scope: null, - provider_state: null, - created_at: now, - updated_at: now, + }); + const rowId = rowIdOf(inserted); + if (rowId === null) { + return yield* new StorageError({ + message: + "Storage adapter did not return the inserted connection row's row_id; the create cannot be compensated safely.", + cause: undefined, }); } + return rowId; }), + ).pipe( + // Both racers can observe absence and reach the insert; the primary + // key then picks the winner. Map the loser's constraint violation to + // the same typed 409 the pre-checks produce. + Effect.catchTag("UniqueViolationError", () => + Effect.fail( + new ConnectionAlreadyExistsError({ + owner: input.owner, + integration: input.integration, + name, + }), + ), + ), ); + // Winner-only credential write: only the create whose row insert + // committed may touch the provider — a pasted value's item id is + // deterministic, so a losing create would clobber the winner's (or a + // pre-existing connection's) secret. The writes run inline, straight + // after the transactional insert above and BEFORE the connection's + // tools are produced below: GraphQL/MCP plugins do authenticated + // introspection via `getValues()` at tool-production time, so the + // credentials must exist by then or the catalog is discovered + // empty/incomplete and never re-discovered. + // + // Known limitation (pre-existing, not addressed here): `transaction` + // nests by pass-through, so a create running inside an enclosing + // plugin `ctx.transaction` writes credentials before the OUTER + // commit. If that transaction rolls back, the row vanishes with it + // but the credential items survive as orphans at their deterministic + // ids — inert until the next same-shaped create overwrites them. + // External credential stores cannot join a database transaction, and + // deferring the write past the outer commit was tried and reverted: + // it broke the tool-production ordering above and could not guarantee + // the deferred hook runs exactly once under interruption. + if (pastedWrites.length > 0) { + const written: ProviderItemId[] = []; + const writeAll = Effect.gen(function* () { + for (const entry of pastedWrites) { + yield* entry.write; + written.push(entry.itemId); + } + }); + + // While the committed row exists no concurrent create can win, so + // on an incomplete write it is ours to tear down — a surviving row + // whose item_ids were never stored would 409 every retry while + // failing every invocation with `connection_value_missing`. But + // "ours" needs proof before anything is deleted: the composite key + // (owner, integration, name) can change hands while compensation is + // still pending (provider calls can be slow) — a concurrent remove + // frees the name, a new create takes it and writes fresh secrets at + // the SAME deterministic item ids. The one column that tells our + // row apart from such a successor is `insertedRowId`, so the row + // delete carries it in its WHERE (guarded delete), and the identity + // check runs in the same transaction as the delete so both see one + // consistent row. + // + // Order matters: the ROW is deleted first, and the credential items + // are undone only when the guarded delete actually removed OUR row. + // If the row is already gone or replaced, losing compensation is + // correct — the remover already cleaned up, and the deterministic + // item ids may by now carry the successor's secrets, so deleting + // them here would clobber a healthy connection. Nothing here is + // silent: every failed or impossible undo is logged, and + // `rowOutcome` converts a stranded row into an error that names it. + // + // Known limitations, accepted deliberately: provider credential + // stores expose no conditional delete, so perfect cleanup under a + // concurrent remove/recreate is impossible at this layer, and no + // further machinery is built for it. + // - Under concurrent remove/recreate, compensation may skip item + // deletion, leaving orphaned credential values at the + // deterministic item ids. Orphans are inert without a row and the + // next same-shaped create overwrites them; orphans are preferred + // over the alternative, clobbering a live successor's secrets. + // - A successor that overwrites one variable, fails before the + // next, and then also fails its own compensating row delete + // leaves a stranded connection that can resolve one stale + // predecessor value. Closing this needs provider-side conditional + // deletes, which do not exist; the stranded state is surfaced + // loudly as the typed StorageError below, naming the connection. + // - On a non-transactional adapter (statements auto-commit, no + // rollback — Cloudflare D1) the guarded delete may already have + // committed when its own rejection surfaces or when the + // confirmation read fails; the items are left in place as inert + // orphans. + const rowOutcomeRef = yield* Ref.make< + "removed" | "superseded" | "overtaken" | "failed" | "unknown" + >("removed"); + const logContext = { + owner: input.owner, + integration: String(input.integration), + connection: String(name), + }; + const compensate = Effect.gen(function* () { + // Progress marker for the transaction below. It distinguishes + // "compensation failed before the guarded delete was issued" + // (nothing can have been deleted; a surviving row is truthfully + // stranded) from "the delete was attempted". Set BEFORE the + // delete statement is issued, not after it resolves: a rejection + // DURING the statement is already ambiguous on an auto-commit + // adapter (D1), where the delete may have executed before the + // rejection surfaced. Deliberately a plain mutable outside the + // transaction: a rollback cannot un-set it, which is the point — + // it records that the statement was issued, not committed state. + // On an interactive adapter a failure from the attempt onward + // rolls the delete back; on an auto-commit adapter the delete + // may already have committed. This layer cannot tell which world + // it is in, so any failure from the attempt onward is reported + // as "unknown", never as a stranded row. + let rowDeleteAttempted = false; + const rowOutcome = yield* transaction( + Effect.gen(function* () { + const current = yield* findConnectionRow({ + owner: input.owner, + integration: input.integration, + name, + }); + if (rowIdOf(current) !== insertedRowId) { + return "superseded" as const; + } + // From here on a failure can no longer prove the row + // survived: the statement below may execute before its + // rejection surfaces. + rowDeleteAttempted = true; + yield* core.deleteMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(input.owner)(b), + b("integration", "=", String(input.integration)), + b("name", "=", String(name)), + // Even if the row changed hands between the read above + // and this statement, only OUR row can match. + b("row_id", "=", insertedRowId), + ), + }); + // `deleteMany` returns void, so whether the guarded delete + // removed OUR row cannot be read off its result — and the + // identity read above and the delete can straddle a + // concurrent remove/recreate under weak isolation. Confirm + // against the table instead, in this same transaction: the + // guarded delete could only ever match our row, so any row + // still holding the name is a successor (or restored + // original) — our delete removed nothing, and the surviving + // row's owner owns both the name and the credential items. + // Only when no row remains is ours provably gone and the + // items ours to undo. A successor inserting after this + // transaction commits can still interleave with the item + // deletes below; that residual is accepted (see the + // known-limitations note above). + const survivor = yield* findConnectionRow({ + owner: input.owner, + integration: input.integration, + name, + }); + if (survivor !== null) { + return "overtaken" as const; + } + return "removed" as const; + }), + ).pipe( + Effect.catchCause((cause) => + rowDeleteAttempted + ? Effect.logError( + "executor connection create could not confirm its compensating row delete: the connection row may be deleted or stranded", + { ...logContext, cause }, + ).pipe(Effect.as("unknown" as const)) + : Effect.logError( + "executor connection create stranded a connection row it could not delete", + { ...logContext, cause }, + ).pipe(Effect.as("failed" as const)), + ), + ); + yield* Ref.set(rowOutcomeRef, rowOutcome); + if (rowOutcome === "superseded") { + // A concurrent remove took our row, and a successor may already + // own the name and the item ids. The remover cleaned up; + // nothing left here is ours to touch. + yield* Effect.logInfo( + "executor connection create skipped compensation: the connection row was already removed or replaced", + logContext, + ); + return; + } + if (rowOutcome === "overtaken") { + // The guarded delete removed nothing and another row now holds + // the name: a concurrent remove/recreate interleaved between + // the identity read and the delete. The surviving row's owner + // owns the name and the credential items; deleting the items + // here would destroy that live connection's secrets. + yield* Effect.logInfo( + "executor connection create skipped credential cleanup: its guarded row delete removed nothing and another connection now holds the name; the surviving connection owns the credential items", + logContext, + ); + return; + } + if (rowOutcome === "failed") { + // Compensation failed before the row delete was even issued, + // so the row — still ours — keeps holding the name together + // with the items that already landed. Leave the items in + // place (they belong to the + // stranded row the caller is told to remove) and let the exit + // handling below surface the error. + return; + } + if (rowOutcome === "unknown") { + // The guarded delete was attempted but its outcome could not + // be confirmed — the statement itself rejected, or the + // confirmation read after it failed — so whether OUR row + // survived cannot be known: an interactive adapter rolled the + // delete back with the transaction (row stranded), a + // non-transactional adapter may have already committed it (row + // gone). Deleting the items under a surviving row + // would strand it valueless, so ALL item deletion is skipped; + // the exit handling below reports the unconfirmed state. + return; + } + for (const entry of pastedWrites) { + if (!written.includes(entry.itemId)) continue; + if (entry.remove === null) { + // A provider exposing `set` without `delete` cannot undo its + // own writes; say so instead of silently skipping. + yield* Effect.logWarning( + "executor connection create cannot undo a credential write: the provider has no delete, so a partial credential may be stranded", + { ...logContext, item: String(entry.itemId) }, + ); + continue; + } + yield* entry.remove.pipe( + Effect.catchCause((cause) => + Effect.logError("executor connection create failed to undo a credential write", { + ...logContext, + item: String(entry.itemId), + cause, + }), + ), + ); + } + }); + + // `onExit`, not `tapError`: compensation must also run when the + // write is interrupted or dies with a defect. The stranded-row + // promise must hold on every one of those exit shapes, so the exit + // is captured and re-raised by hand: a typed failure or a defect + // that left the row behind becomes the StorageError below, while an + // interruption cannot carry a typed error at all (interrupting wins + // over failing) — for it the loud log inside `compensate` is the + // only signal, and the interruption is re-raised untouched. + const writeExit = yield* writeAll.pipe( + Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)), + Effect.exit, + ); + if (Exit.isFailure(writeExit)) { + const rowOutcome = yield* Ref.get(rowOutcomeRef); + if (rowOutcome === "failed" && !Cause.hasInterruptsOnly(writeExit.cause)) { + return yield* new StorageError({ + message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded with incomplete credentials and must be removed manually.`, + cause: Cause.squash(writeExit.cause), + }); + } + if (rowOutcome === "unknown" && !Cause.hasInterruptsOnly(writeExit.cause)) { + return yield* new StorageError({ + message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and its compensating delete could not be confirmed: the connection row may be deleted or may remain with incomplete credentials; its credential items were left in place.`, + cause: Cause.squash(writeExit.cause), + }); + } + return yield* Effect.failCause(writeExit.cause); + } + } + // Record the sighting. The request seam (`makeScopedExecutor`) already // does this for every hosted call, so this is the belt for direct // SDK/CLI callers that never pass through it — a connecting principal @@ -3628,8 +3945,10 @@ export const createExecutor = => diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index a69979b585..fcbc173d20 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -74,6 +74,7 @@ export { IntegrationNotFoundError, IntegrationAlreadyExistsError, IntegrationRemovalNotAllowedError, + ConnectionAlreadyExistsError, ConnectionNotFoundError, CredentialProviderNotRegisteredError, CredentialResolutionError, diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts index 52df246eae..3f11fe5523 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -41,6 +41,7 @@ import type { InvokeOptions, } from "./elicitation"; import type { + ConnectionAlreadyExistsError, ExecuteError, ConnectionNotFoundError, CredentialProviderNotRegisteredError, @@ -208,6 +209,7 @@ export interface PluginCtx { ) => Effect.Effect< Connection, | IntegrationNotFoundError + | ConnectionAlreadyExistsError | CredentialProviderNotRegisteredError | InvalidConnectionInputError | StorageFailure diff --git a/packages/core/sdk/src/shared.ts b/packages/core/sdk/src/shared.ts index d0cd24da7a..e50db54ae3 100644 --- a/packages/core/sdk/src/shared.ts +++ b/packages/core/sdk/src/shared.ts @@ -59,6 +59,7 @@ export { IntegrationNotFoundError, IntegrationAlreadyExistsError, IntegrationRemovalNotAllowedError, + ConnectionAlreadyExistsError, ConnectionNotFoundError, InvalidConnectionInputError, CredentialProviderNotRegisteredError, diff --git a/packages/plugins/mcp/src/sdk/plugin.ts b/packages/plugins/mcp/src/sdk/plugin.ts index 27165206ae..96e9dc7ae6 100644 --- a/packages/plugins/mcp/src/sdk/plugin.ts +++ b/packages/plugins/mcp/src/sdk/plugin.ts @@ -1036,6 +1036,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { Effect.fail( new McpConnectionError({ transport: "stdio", message: cause.message }), ), + ConnectionAlreadyExistsError: (cause) => + Effect.fail( + new McpConnectionError({ transport: "stdio", message: cause.message }), + ), CredentialProviderNotRegisteredError: (cause) => Effect.fail( new McpConnectionError({ transport: "stdio", message: cause.message }), diff --git a/packages/react/src/components/add-account-modal.tsx b/packages/react/src/components/add-account-modal.tsx index 5c2309ca95..77bade18d9 100644 --- a/packages/react/src/components/add-account-modal.tsx +++ b/packages/react/src/components/add-account-modal.tsx @@ -1,6 +1,8 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useAtomRefresh, useAtomSet, useAtomValue } from "@effect/atom-react"; import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"; import { ConnectionName, @@ -606,6 +608,19 @@ export const connectionLabelForHost = ( organizationId: string | null, ): string => label.trim() || `${ownerLabelForHost(owner, organizationId)} ${integrationName}`; +/** The create endpoint rejected the name as taken (409). Like + * `isIntegrationAlreadyExistsExit`: the error's `message` is a getter derived + * from its fields, so it doesn't survive the wire — match the tag and rebuild + * the message client-side. */ +export const isConnectionAlreadyExistsExit = (exit: Exit.Exit): boolean => + Option.match(Exit.findErrorOption(exit), { + onNone: () => false, + onSome: Predicate.isTagged("ConnectionAlreadyExistsError"), + }); + +export const connectionExistsMessage = (label: string): string => + `A connection named "${label}" already exists. Pick a different name, or remove the existing connection first.`; + /** The default owner a new connection is saved under when the user makes no * explicit choice. Personal: a connection is most often a personal credential. */ export const DEFAULT_CONNECTION_OWNER: Owner = "user"; @@ -1747,7 +1762,8 @@ function AddAccountModalView(props: AddAccountModalProps) { const showSavedToPicker = !oauthRegistering && savedToOptions.length > 1; // OAuth mints a NEW connection per connect, so its preview shows the // uniquified name (`personalGmail2`); credential saves keep the plain - // derivation (they surface an explicit overwrite through the same name). + // derivation, because a save under a taken name is a conflict the server + // rejects rather than something the client silently renames around. const callableName = isOAuth ? previewConnectionName(label, savedToOwner) : connectionNameFrom(label, savedToOwner, integrationName, organizationId); @@ -1982,7 +1998,16 @@ function AddAccountModalView(props: AddAccountModalProps) { }); if (Exit.isFailure(exit)) { setSubmitting(false); - toast.error(messageFromExit(exit, "Failed to add connection")); + // The conflict error's message is a getter derived from its fields, so + // it doesn't survive the wire — rebuild it from the tag (the same + // pattern as isIntegrationAlreadyExistsExit). + toast.error( + isConnectionAlreadyExistsExit(exit) + ? connectionExistsMessage( + connectionLabelForHost(label, owner, integrationName, organizationId), + ) + : messageFromExit(exit, "Failed to add connection"), + ); return; } toast.success("Connection added");