diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index a183ac501..0f58f2fd0 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -10,9 +10,10 @@ import { ToolAddress, ToolName, } from "./ids"; +import { createExecutor } from "./executor"; import { definePlugin } from "./plugin"; import type { CredentialProvider } from "./provider"; -import { makeTestExecutor } from "./testing"; +import { makeTestConfig, makeTestExecutor } from "./testing"; // removed: v1 connection-refresh lifecycle, ConnectionProvider.refresh, // SecretProvider, accessToken token-refresh + in-flight dedup tests — the v2 @@ -359,6 +360,264 @@ describe("connections.refresh", () => { ); }); +describe("tool catalog sync safety", () => { + it.effect( + "background sync preserves a nonzero catalog when a plugin returns authoritative empty", + () => + Effect.scoped( + Effect.gen(function* () { + let empty = false; + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.sync(() => ({ + tools: empty + ? [] + : [ + { name: ToolName.make("deploy"), description: "deploy" }, + { name: ToolName.make("list"), description: "list" }, + ], + })), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + const config = makeTestConfig({ plugins: [guardedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.guarded.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + + empty = true; + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + const tools = yield* executor.tools.list({ integration: INTEG }); + const connection = yield* executor.connections.get({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + + expect(tools.map((tool) => String(tool.name)).sort()).toEqual(["deploy", "list"]); + expect(connection?.lastHealth).toMatchObject({ + status: "degraded", + detail: expect.stringContaining("authoritative empty catalog"), + }); + }), + ), + ); + + it.effect("explicit refresh accepts an authoritative empty catalog", () => + Effect.scoped( + Effect.gen(function* () { + let empty = false; + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.sync(() => ({ + tools: empty + ? [] + : [ + { name: ToolName.make("deploy"), description: "deploy" }, + { name: ToolName.make("list"), description: "list" }, + ], + })), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + const executor = yield* createExecutor( + makeTestConfig({ plugins: [guardedPlugin] as const }), + ); + yield* executor.guarded.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + + empty = true; + const refreshed = yield* executor.connections.refresh({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + const tools = yield* executor.tools.list({ integration: INTEG }); + + expect(refreshed).toEqual([]); + expect(tools).toEqual([]); + }), + ), + ); + + it.effect("successful sync clears a prior tool-sync failure health record", () => + Effect.scoped( + Effect.gen(function* () { + let incomplete = false; + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.sync(() => + incomplete + ? { + tools: [], + incomplete: true, + incompleteReason: "temporary catalog outage", + } + : { + tools: [ + { name: ToolName.make("deploy"), description: "deploy" }, + { name: ToolName.make("list"), description: "list" }, + ], + }, + ), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + const config = makeTestConfig({ plugins: [guardedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.guarded.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + + incomplete = true; + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + yield* executor.tools.list({ integration: INTEG }); + expect( + (yield* executor.connections.get({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }))?.lastHealth, + ).toMatchObject({ + status: "degraded", + detail: expect.stringContaining("temporary catalog outage"), + }); + + incomplete = false; + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + yield* executor.tools.list({ integration: INTEG }); + const connection = yield* executor.connections.get({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + + expect(connection?.lastHealth).toBeNull(); + }), + ), + ); + + it.effect("successful sync preserves genuine health-check records", () => + Effect.scoped( + Effect.gen(function* () { + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ + tools: [ + { name: ToolName.make("deploy"), description: "deploy" }, + { name: ToolName.make("list"), description: "list" }, + ], + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + const config = makeTestConfig({ plugins: [guardedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.guarded.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + + const health = { + status: "degraded" as const, + checkedAt: Date.now(), + detail: "health check returned HTTP 503", + }; + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set: { tools_synced_at: null, last_health: health }, + }), + ); + yield* executor.tools.list({ integration: INTEG }); + const connection = yield* executor.connections.get({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + + expect(connection?.lastHealth).toMatchObject(health); + }), + ), + ); +}); + describe("execute over a connection", () => { it.effect("resolves the credential value and hands it to invokeTool", () => Effect.gen(function* () { diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 071d4a21b..e290e1f4e 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -2002,9 +2002,21 @@ export const createExecutor = ({ + status: "degraded", + checkedAt: Date.now(), + detail: `${toolSyncHealthDetailPrefix}: ${reason}`, + }); + + const syncHealthReason = (result: ResolveToolsResult): string => + result.incompleteReason ?? "plugin returned an incomplete tool catalog"; + const produceConnectionTools = ( integrationRow: IntegrationRow, ref: ConnectionRef, + mode: "explicit" | "background" = "explicit", ): Effect.Effect => Effect.gen(function* () { const runtime = runtimes.get(integrationRow.plugin_id); @@ -2019,18 +2031,39 @@ export const createExecutor = - b.and( - byOwner(owner)(b), - b("integration", "=", String(ref.integration)), - b("name", "=", String(ref.name)), - ), - set: { tools_synced_at: Date.now() }, - }); + // Successful syncs also clear stale sync-failure health records, while + // preserving genuine health-check outcomes. + const stampSynced = (row: ConnectionRow | null) => + core.updateMany("connection", { + where: connectionWhere, + set: syncedSet(row), + }); + const stampSyncedWithHealth = (reason: string) => + core.updateMany("connection", { + where: connectionWhere, + set: { + tools_synced_at: Date.now(), + last_health: toolSyncHealth(reason), + updated_at: new Date(), + }, + }); // Defense in depth (and cleanup for rows created before the create-time // guard, or emptied by an external edit): a credentialed non-OAuth @@ -2051,7 +2084,7 @@ export const createExecutor = rowToTool(row as ConnectionToolRow)); } + if (mode === "background" && result.tools.length === 0) { + const keptRows = yield* core.findMany("tool", { where }); + if (keptRows.length > 0) { + const reason = + "background tool sync produced an authoritative empty catalog for a connection with existing tools"; + yield* stampSyncedWithHealth(reason); + yield* Effect.logWarning("executor tool sync preserved nonzero catalog", { + reason, + integration: String(ref.integration), + connection: String(ref.name), + existingToolCount: keptRows.length, + }); + return keptRows.map((row) => rowToTool(row as ConnectionToolRow)); + } + } + const now = new Date(); const toolRows = result.tools.map((tool: ToolDef) => ({ tenant: keys.tenant, @@ -2132,7 +2187,7 @@ export const createExecutor = Effect.succeed([] as readonly Tool[])), Effect.withSpan("executor.tools.sync_stale", { attributes: { diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts index 31dfa07eb..bb3f0e18d 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -297,6 +297,9 @@ export interface ResolveToolsResult { * listing is authoritative, including a genuine "this source has zero * tools". */ readonly incomplete?: boolean; + /** Human-readable reason for an incomplete listing. Persisted by core when it + * preserves the prior catalog so operators can see why data is stale. */ + readonly incompleteReason?: string; } // --------------------------------------------------------------------------- diff --git a/packages/plugins/graphql/src/sdk/plugin.ts b/packages/plugins/graphql/src/sdk/plugin.ts index 3ada4e0f4..1938de164 100644 --- a/packages/plugins/graphql/src/sdk/plugin.ts +++ b/packages/plugins/graphql/src/sdk/plugin.ts @@ -975,10 +975,17 @@ export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => { readonly httpClientLayer: Layer.Layer; }) => Effect.gen(function* () { + const incomplete = (reason: string) => ({ + tools: [] as readonly ToolDef[], + incomplete: true, + incompleteReason: reason, + }); const decoded = yield* decodeGraphqlIntegrationConfig(config).pipe(Effect.option); if (Option.isNone(decoded)) return { tools: [] }; const graphqlConfig = decoded.value; - const introspectionJson = yield* loadIntrospectionJson(storage, graphqlConfig); + const introspectionJson = yield* loadIntrospectionJson(storage, graphqlConfig).pipe( + Effect.catch(() => Effect.succeed(null)), + ); // Live introspection (no stored snapshot) needs the connection's // credential inputs for auth-required endpoints; resolve them lazily. const values = @@ -994,15 +1001,27 @@ export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => { template, options?.httpClientLayer ?? httpClientLayer, ).pipe(Effect.option); - if (Option.isNone(introspection)) return { tools: [] }; + if (Option.isNone(introspection)) { + return incomplete("GraphQL introspection could not be loaded."); + } const extracted = yield* extract(introspection.value).pipe(Effect.option); - if (Option.isNone(extracted)) return { tools: [] }; + if (Option.isNone(extracted)) { + return incomplete("GraphQL introspection result could not be converted to tools."); + } const prepared = prepareOperations(extracted.value.result.fields, introspection.value); return { tools: buildToolDefs(prepared), definitions: extracted.value.definitions, }; - }).pipe(Effect.catch(() => Effect.succeed({ tools: [] as readonly ToolDef[] }))), + }).pipe( + Effect.catch(() => + Effect.succeed({ + tools: [] as readonly ToolDef[], + incomplete: true, + incompleteReason: "GraphQL tool catalog could not be resolved.", + }), + ), + ), // ----------------------------------------------------------------------- // Invoke one of a connection's tools. Look up the operation by integration diff --git a/packages/plugins/openapi/src/sdk/backing.ts b/packages/plugins/openapi/src/sdk/backing.ts index 2ec415764..736be3b77 100644 --- a/packages/plugins/openapi/src/sdk/backing.ts +++ b/packages/plugins/openapi/src/sdk/backing.ts @@ -579,24 +579,37 @@ export const resolveOpenApiBackedTools = ({ readonly storage: OpenapiStore; }): Effect.Effect => Effect.gen(function* () { + const incomplete = (reason: string): ResolveToolsResult => ({ + tools: [], + definitions: {}, + incomplete: true, + incompleteReason: reason, + }); const openApiConfig = decodeOpenApiIntegrationConfig(config); if (!openApiConfig) return { tools: [], definitions: {} }; if (openApiConfig.specHash != null) { - const defsJson = yield* storage.getDefs(openApiConfig.specHash); + const defsJson = yield* storage + .getDefs(openApiConfig.specHash) + .pipe(Effect.catch(() => Effect.succeed(null))); if (defsJson != null) { const definitions = Option.getOrNull(decodeDefsJson(defsJson)); if (definitions != null) { - const ops = yield* storage.listOperations(String(integration.slug)); + const ops = yield* storage + .listOperations(String(integration.slug)) + .pipe(Effect.catch(() => Effect.succeed(null))); + if (ops == null) return incomplete("OpenAPI operation bindings could not be loaded."); return { tools: ops.map(toolDefFromStoredOperation), definitions }; } } } - const specText = yield* loadOpenApiSpecText(storage, openApiConfig); - if (specText == null) return { tools: [], definitions: {} }; + const specText = yield* loadOpenApiSpecText(storage, openApiConfig).pipe( + Effect.catch(() => Effect.succeed(null)), + ); + if (specText == null) return incomplete("OpenAPI spec blob could not be loaded."); const compiled = yield* compileOpenApiSpecCached(openApiConfig.specHash, specText).pipe( Effect.catch(() => Effect.succeed(null)), ); - if (!compiled) return { tools: [], definitions: {} }; + if (!compiled) return incomplete("OpenAPI spec could not be parsed."); return { tools: openApiToolDefsFromCompiled(compiled), definitions: compiled.hoistedDefs, diff --git a/packages/plugins/openapi/src/sdk/spec-blob.test.ts b/packages/plugins/openapi/src/sdk/spec-blob.test.ts index 186050633..a75378f08 100644 --- a/packages/plugins/openapi/src/sdk/spec-blob.test.ts +++ b/packages/plugins/openapi/src/sdk/spec-blob.test.ts @@ -32,6 +32,7 @@ import { variable } from "@executor-js/sdk/http-auth"; import { openApiPlugin } from "./plugin"; import type { OpenapiStore } from "./store"; import type { AuthenticationInput } from "./types"; +import { defsBlobKey, specBlobKey } from "./store"; import { makeOpenApiHttpApiTestSourceConfig, serveOpenApiHttpApiTestServer, @@ -65,12 +66,49 @@ const specText = () => { return spec.url; }; +const specTextWithDefinition = () => + JSON.stringify({ + openapi: "3.1.0", + info: { title: "Catalog", version: "1.0.0" }, + paths: { + "/items": { + get: { + operationId: "listItems", + responses: { + "200": { + description: "ok", + content: { + "application/json": { + schema: { + type: "array", + items: { $ref: "#/components/schemas/Item" }, + }, + }, + }, + }, + }, + }, + }, + }, + components: { + schemas: { + Item: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + }, + }, + }, + }); + const apiKeyTemplate: AuthenticationInput = { slug: AuthTemplateSlug.make("apiKey"), type: "apiKey", headers: { "x-api-key": [variable("token")] }, }; +const openApiBlobNamespace = "o:test-tenant/openapi"; + describe("OpenAPI plugin — spec blob storage", () => { it.effect("addSpec stores a content pointer, not the inline spec text", () => Effect.scoped( @@ -125,6 +163,197 @@ describe("OpenAPI plugin — spec blob storage", () => { ), ); + it.effect( + "stale sync preserves tools and definitions when the spec blob parses to a non-object", + () => + Effect.scoped( + Effect.gen(function* () { + const config = makeTestConfig({ plugins: testPlugins() }); + const executor = yield* createExecutor(config); + const text = specTextWithDefinition(); + const hash = yield* sha256Hex(text); + + yield* executor.openapi.addSpec({ + spec: { kind: "blob", value: text }, + slug: "corrupt_blob", + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make("corrupt_blob"), + template: AuthTemplateSlug.make("none"), + values: {}, + }); + + const beforeTools = yield* executor.tools.list({ + integration: IntegrationSlug.make("corrupt_blob"), + }); + const beforeDefinitions = yield* Effect.promise(() => + config.db.findMany("definition", { + where: (b) => b("integration", "=", "corrupt_blob"), + }), + ); + expect(beforeTools.length).toBeGreaterThan(0); + expect(beforeDefinitions.length).toBeGreaterThan(0); + + yield* Effect.promise(() => + config.db.updateMany("blob", { + where: (b) => + b.and(b("namespace", "=", openApiBlobNamespace), b("key", "=", specBlobKey(hash))), + set: { value: JSON.stringify("not an object") }, + }), + ); + yield* Effect.promise(() => + config.db.deleteMany("blob", { + where: (b) => + b.and(b("namespace", "=", openApiBlobNamespace), b("key", "=", defsBlobKey(hash))), + }), + ); + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", "corrupt_blob"), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + + const afterTools = yield* executor.tools.list({ + integration: IntegrationSlug.make("corrupt_blob"), + }); + const afterDefinitions = yield* Effect.promise(() => + config.db.findMany("definition", { + where: (b) => b("integration", "=", "corrupt_blob"), + }), + ); + const connection = yield* executor.connections.get({ + owner: "org", + integration: IntegrationSlug.make("corrupt_blob"), + name: ConnectionName.make("main"), + }); + + expect(afterTools.map((tool) => String(tool.name)).sort()).toEqual( + beforeTools.map((tool) => String(tool.name)).sort(), + ); + expect(afterDefinitions).toHaveLength(beforeDefinitions.length); + expect(connection?.lastHealth).toMatchObject({ + status: "degraded", + detail: expect.stringContaining("OpenAPI spec could not be parsed"), + }); + }), + ), + ); + + it.effect("stale sync preserves tools and definitions when the spec blob is missing", () => + Effect.scoped( + Effect.gen(function* () { + const config = makeTestConfig({ plugins: testPlugins() }); + const executor = yield* createExecutor(config); + const text = specTextWithDefinition(); + const hash = yield* sha256Hex(text); + + yield* executor.openapi.addSpec({ + spec: { kind: "blob", value: text }, + slug: "missing_blob", + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make("missing_blob"), + template: AuthTemplateSlug.make("none"), + values: {}, + }); + + const beforeTools = yield* executor.tools.list({ + integration: IntegrationSlug.make("missing_blob"), + }); + const beforeDefinitions = yield* Effect.promise(() => + config.db.findMany("definition", { + where: (b) => b("integration", "=", "missing_blob"), + }), + ); + expect(beforeTools.length).toBeGreaterThan(0); + expect(beforeDefinitions.length).toBeGreaterThan(0); + + yield* Effect.promise(() => + config.db.deleteMany("blob", { + where: (b) => + b.and( + b("namespace", "=", openApiBlobNamespace), + b("key", "in", [specBlobKey(hash), defsBlobKey(hash)]), + ), + }), + ); + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", "missing_blob"), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + + const afterTools = yield* executor.tools.list({ + integration: IntegrationSlug.make("missing_blob"), + }); + const afterDefinitions = yield* Effect.promise(() => + config.db.findMany("definition", { + where: (b) => b("integration", "=", "missing_blob"), + }), + ); + const connection = yield* executor.connections.get({ + owner: "org", + integration: IntegrationSlug.make("missing_blob"), + name: ConnectionName.make("main"), + }); + + expect(afterTools.map((tool) => String(tool.name)).sort()).toEqual( + beforeTools.map((tool) => String(tool.name)).sort(), + ); + expect(afterDefinitions).toHaveLength(beforeDefinitions.length); + expect(connection?.lastHealth).toMatchObject({ + status: "degraded", + detail: expect.stringContaining("OpenAPI spec blob could not be loaded"), + }); + }), + ), + ); + + it.effect("explicit spec refresh accepts a valid OpenAPI document with zero operations", () => + Effect.scoped( + Effect.gen(function* () { + const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() })); + const emptySpec = JSON.stringify({ + openapi: "3.1.0", + info: { title: "Empty", version: "1.0.0" }, + paths: {}, + }); + + yield* executor.openapi.addSpec({ + spec: { kind: "blob", value: specText() }, + slug: "healthy_zero", + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make("healthy_zero"), + template: AuthTemplateSlug.make("none"), + values: {}, + }); + expect( + (yield* executor.tools.list({ integration: IntegrationSlug.make("healthy_zero") })) + .length, + ).toBeGreaterThan(0); + + const update = yield* executor.openapi.updateSpec("healthy_zero", { + spec: { kind: "blob", value: emptySpec }, + }); + const tools = yield* executor.tools.list({ + integration: IntegrationSlug.make("healthy_zero"), + }); + + expect(update.toolCount).toBe(0); + expect(tools).toEqual([]); + }), + ), + ); + it.effect("resolveTools reads the spec from the store, never an inline field", () => Effect.gen(function* () { const plugin = openApiPlugin({ httpClientLayer: FetchHttpClient.layer });