History

Preserve catalogs on failed sync (#1335)

Rhys Sullivan committed 4 months ago

Commit efd09e91 parent c1893c56 files · +610 −28Browse files

Unified · Split

6 files
  1. packages/core/sdk/src/connections.test.ts +260 −1
  2. packages/core/sdk/src/executor.ts +77 −18
  3. packages/core/sdk/src/plugin.ts +3 −0
  4. packages/plugins/graphql/src/sdk/plugin.ts +23 −4
  5. packages/plugins/openapi/src/sdk/backing.ts +18 −5
  6. packages/plugins/openapi/src/sdk/spec-blob.test.ts +229 −0
packages/core/sdk/src/connections.test.ts+260−1
@@ -10,9 +10,10 @@ import { · 9 unmodified lines
10 ToolAddress,10 ToolAddress,
11 ToolName,11 ToolName,
12} from "./ids";12} from "./ids";
13import { createExecutor } from "./executor";
13import { definePlugin } from "./plugin";14import { definePlugin } from "./plugin";
14import type { CredentialProvider } from "./provider";15import type { CredentialProvider } from "./provider";
15import { makeTestExecutor } from "./testing";16import { makeTestConfig, makeTestExecutor } from "./testing";
16 17
17// removed: v1 connection-refresh lifecycle, ConnectionProvider.refresh,18// removed: v1 connection-refresh lifecycle, ConnectionProvider.refresh,
18// SecretProvider, accessToken token-refresh + in-flight dedup tests — the v219// SecretProvider, accessToken token-refresh + in-flight dedup tests — the v2
@@ -359,6 +360,264 @@ describe("connections.refresh", () => { · 340 unmodified lines
359 );360 );
360});361});
361 362
363describe("tool catalog sync safety", () => {
364 it.effect(
365 "background sync preserves a nonzero catalog when a plugin returns authoritative empty",
366 () =>
367 Effect.scoped(
368 Effect.gen(function* () {
369 let empty = false;
370 const guardedPlugin = definePlugin(() => ({
371 id: "guarded" as const,
372 credentialProviders: [memoryProvider()],
373 storage: () => ({}),
374 resolveTools: () =>
375 Effect.sync(() => ({
376 tools: empty
377 ? []
378 : [
379 { name: ToolName.make("deploy"), description: "deploy" },
380 { name: ToolName.make("list"), description: "list" },
381 ],
382 })),
383 invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }),
384 extension: (ctx) => ({
385 seed: () =>
386 ctx.core.integrations.register({
387 slug: INTEG,
388 description: "Vercel",
389 config: {},
390 }),
391 }),
392 }))();
393 const config = makeTestConfig({ plugins: [guardedPlugin] as const });
394 const executor = yield* createExecutor(config);
395 yield* executor.guarded.seed();
396 yield* executor.connections.create({
397 owner: "org",
398 name: ConnectionName.make("main"),
399 integration: INTEG,
400 template: TEMPLATE,
401 value: "secret-token",
402 });
403
404 empty = true;
405 yield* Effect.promise(() =>
406 config.db.updateMany("connection", {
407 where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")),
408 set: { tools_synced_at: null },
409 }),
410 );
411 const tools = yield* executor.tools.list({ integration: INTEG });
412 const connection = yield* executor.connections.get({
413 owner: "org",
414 integration: INTEG,
415 name: ConnectionName.make("main"),
416 });
417
418 expect(tools.map((tool) => String(tool.name)).sort()).toEqual(["deploy", "list"]);
419 expect(connection?.lastHealth).toMatchObject({
420 status: "degraded",
421 detail: expect.stringContaining("authoritative empty catalog"),
422 });
423 }),
424 ),
425 );
426
427 it.effect("explicit refresh accepts an authoritative empty catalog", () =>
428 Effect.scoped(
429 Effect.gen(function* () {
430 let empty = false;
431 const guardedPlugin = definePlugin(() => ({
432 id: "guarded" as const,
433 credentialProviders: [memoryProvider()],
434 storage: () => ({}),
435 resolveTools: () =>
436 Effect.sync(() => ({
437 tools: empty
438 ? []
439 : [
440 { name: ToolName.make("deploy"), description: "deploy" },
441 { name: ToolName.make("list"), description: "list" },
442 ],
443 })),
444 invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }),
445 extension: (ctx) => ({
446 seed: () =>
447 ctx.core.integrations.register({
448 slug: INTEG,
449 description: "Vercel",
450 config: {},
451 }),
452 }),
453 }))();
454 const executor = yield* createExecutor(
455 makeTestConfig({ plugins: [guardedPlugin] as const }),
456 );
457 yield* executor.guarded.seed();
458 yield* executor.connections.create({
459 owner: "org",
460 name: ConnectionName.make("main"),
461 integration: INTEG,
462 template: TEMPLATE,
463 value: "secret-token",
464 });
465
466 empty = true;
467 const refreshed = yield* executor.connections.refresh({
468 owner: "org",
469 integration: INTEG,
470 name: ConnectionName.make("main"),
471 });
472 const tools = yield* executor.tools.list({ integration: INTEG });
473
474 expect(refreshed).toEqual([]);
475 expect(tools).toEqual([]);
476 }),
477 ),
478 );
479
480 it.effect("successful sync clears a prior tool-sync failure health record", () =>
481 Effect.scoped(
482 Effect.gen(function* () {
483 let incomplete = false;
484 const guardedPlugin = definePlugin(() => ({
485 id: "guarded" as const,
486 credentialProviders: [memoryProvider()],
487 storage: () => ({}),
488 resolveTools: () =>
489 Effect.sync(() =>
490 incomplete
491 ? {
492 tools: [],
493 incomplete: true,
494 incompleteReason: "temporary catalog outage",
495 }
496 : {
497 tools: [
498 { name: ToolName.make("deploy"), description: "deploy" },
499 { name: ToolName.make("list"), description: "list" },
500 ],
501 },
502 ),
503 invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }),
504 extension: (ctx) => ({
505 seed: () =>
506 ctx.core.integrations.register({
507 slug: INTEG,
508 description: "Vercel",
509 config: {},
510 }),
511 }),
512 }))();
513 const config = makeTestConfig({ plugins: [guardedPlugin] as const });
514 const executor = yield* createExecutor(config);
515 yield* executor.guarded.seed();
516 yield* executor.connections.create({
517 owner: "org",
518 name: ConnectionName.make("main"),
519 integration: INTEG,
520 template: TEMPLATE,
521 value: "secret-token",
522 });
523
524 incomplete = true;
525 yield* Effect.promise(() =>
526 config.db.updateMany("connection", {
527 where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")),
528 set: { tools_synced_at: null },
529 }),
530 );
531 yield* executor.tools.list({ integration: INTEG });
532 expect(
533 (yield* executor.connections.get({
534 owner: "org",
535 integration: INTEG,
536 name: ConnectionName.make("main"),
537 }))?.lastHealth,
538 ).toMatchObject({
539 status: "degraded",
540 detail: expect.stringContaining("temporary catalog outage"),
541 });
542
543 incomplete = false;
544 yield* Effect.promise(() =>
545 config.db.updateMany("connection", {
546 where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")),
547 set: { tools_synced_at: null },
548 }),
549 );
550 yield* executor.tools.list({ integration: INTEG });
551 const connection = yield* executor.connections.get({
552 owner: "org",
553 integration: INTEG,
554 name: ConnectionName.make("main"),
555 });
556
557 expect(connection?.lastHealth).toBeNull();
558 }),
559 ),
560 );
561
562 it.effect("successful sync preserves genuine health-check records", () =>
563 Effect.scoped(
564 Effect.gen(function* () {
565 const guardedPlugin = definePlugin(() => ({
566 id: "guarded" as const,
567 credentialProviders: [memoryProvider()],
568 storage: () => ({}),
569 resolveTools: () =>
570 Effect.succeed({
571 tools: [
572 { name: ToolName.make("deploy"), description: "deploy" },
573 { name: ToolName.make("list"), description: "list" },
574 ],
575 }),
576 invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }),
577 extension: (ctx) => ({
578 seed: () =>
579 ctx.core.integrations.register({
580 slug: INTEG,
581 description: "Vercel",
582 config: {},
583 }),
584 }),
585 }))();
586 const config = makeTestConfig({ plugins: [guardedPlugin] as const });
587 const executor = yield* createExecutor(config);
588 yield* executor.guarded.seed();
589 yield* executor.connections.create({
590 owner: "org",
591 name: ConnectionName.make("main"),
592 integration: INTEG,
593 template: TEMPLATE,
594 value: "secret-token",
595 });
596
597 const health = {
598 status: "degraded" as const,
599 checkedAt: Date.now(),
600 detail: "health check returned HTTP 503",
601 };
602 yield* Effect.promise(() =>
603 config.db.updateMany("connection", {
604 where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")),
605 set: { tools_synced_at: null, last_health: health },
606 }),
607 );
608 yield* executor.tools.list({ integration: INTEG });
609 const connection = yield* executor.connections.get({
610 owner: "org",
611 integration: INTEG,
612 name: ConnectionName.make("main"),
613 });
614
615 expect(connection?.lastHealth).toMatchObject(health);
616 }),
617 ),
618 );
619});
620
362describe("execute over a connection", () => {621describe("execute over a connection", () => {
363 it.effect("resolves the credential value and hands it to invokeTool", () =>622 it.effect("resolves the credential value and hands it to invokeTool", () =>
364 Effect.gen(function* () {623 Effect.gen(function* () {
packages/core/sdk/src/executor.ts+77−18
@@ -2002,9 +2002,21 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 2001 unmodified lines
2002 // Per-connection tool production2002 // Per-connection tool production
2003 // ------------------------------------------------------------------2003 // ------------------------------------------------------------------
2004 2004
2005 const toolSyncHealthDetailPrefix = "Tool sync failing";
2006
2007 const toolSyncHealth = (reason: string): HealthCheckResult => ({
2008 status: "degraded",
2009 checkedAt: Date.now(),
2010 detail: `${toolSyncHealthDetailPrefix}: ${reason}`,
2011 });
2012
2013 const syncHealthReason = (result: ResolveToolsResult): string =>
2014 result.incompleteReason ?? "plugin returned an incomplete tool catalog";
2015
2005 const produceConnectionTools = (2016 const produceConnectionTools = (
2006 integrationRow: IntegrationRow,2017 integrationRow: IntegrationRow,
2007 ref: ConnectionRef,2018 ref: ConnectionRef,
2019 mode: "explicit" | "background" = "explicit",
2008 ): Effect.Effect<readonly Tool[], IntegrationNotFoundError | StorageFailure> =>2020 ): Effect.Effect<readonly Tool[], IntegrationNotFoundError | StorageFailure> =>
2009 Effect.gen(function* () {2021 Effect.gen(function* () {
2010 const runtime = runtimes.get(integrationRow.plugin_id);2022 const runtime = runtimes.get(integrationRow.plugin_id);
@@ -2019,18 +2031,39 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 8 unmodified lines
2019 b("integration", "=", String(ref.integration)),2031 b("integration", "=", String(ref.integration)),
2020 b("connection", "=", String(ref.name)),2032 b("connection", "=", String(ref.name)),
2021 );2033 );
2034 const connectionWhere = (b: AnyCb) =>
2035 b.and(
2036 byOwner(owner)(b),
2037 b("integration", "=", String(ref.integration)),
2038 b("name", "=", String(ref.name)),
2039 );
2040 const isToolSyncHealth = (health: HealthCheckResult | null): boolean =>
2041 health?.detail?.startsWith(toolSyncHealthDetailPrefix) === true;
2042 const syncedSet = (row: ConnectionRow | null) => {
2043 const health = row ? Option.getOrNull(decodeLastHealth(row.last_health)) : null;
2044 return isToolSyncHealth(health)
2045 ? { tools_synced_at: Date.now(), last_health: null, updated_at: new Date() }
2046 : { tools_synced_at: Date.now() };
2047 };
2022 // Every exit stamps the sync time — including the cleanup paths that2048 // Every exit stamps the sync time — including the cleanup paths that
2023 // produce zero tools — so the stale-catalog check (`config_revised_at`2049 // produce zero tools — so the stale-catalog check (`config_revised_at`
2024 // vs `tools_synced_at`) doesn't re-attempt this connection per read.2050 // vs `tools_synced_at`) doesn't re-attempt this connection per read.
2025 const stampSynced = core.updateMany("connection", {2051 // Successful syncs also clear stale sync-failure health records, while
2026 where: (b: AnyCb) =>2052 // preserving genuine health-check outcomes.
2027 b.and(2053 const stampSynced = (row: ConnectionRow | null) =>
2028 byOwner(owner)(b),2054 core.updateMany("connection", {
2029 b("integration", "=", String(ref.integration)),2055 where: connectionWhere,
2030 b("name", "=", String(ref.name)),2056 set: syncedSet(row),
2031 ),2057 });
2032 set: { tools_synced_at: Date.now() },2058 const stampSyncedWithHealth = (reason: string) =>
2033 });2059 core.updateMany("connection", {
2060 where: connectionWhere,
2061 set: {
2062 tools_synced_at: Date.now(),
2063 last_health: toolSyncHealth(reason),
2064 updated_at: new Date(),
2065 },
2066 });
2034 2067
2035 // Defense in depth (and cleanup for rows created before the create-time2068 // Defense in depth (and cleanup for rows created before the create-time
2036 // guard, or emptied by an external edit): a credentialed non-OAuth2069 // guard, or emptied by an external edit): a credentialed non-OAuth
@@ -2051,7 +2084,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 14 unmodified lines
2051 Effect.gen(function* () {2084 Effect.gen(function* () {
2052 yield* core.deleteMany("tool", { where });2085 yield* core.deleteMany("tool", { where });
2053 yield* core.deleteMany("definition", { where });2086 yield* core.deleteMany("definition", { where });
2054 yield* stampSynced;2087 yield* stampSynced(existingRow);
2055 }),2088 }),
2056 );2089 );
2057 return [];2090 return [];
@@ -2063,7 +2096,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 5 unmodified lines
2063 Effect.gen(function* () {2096 Effect.gen(function* () {
2064 yield* core.deleteMany("tool", { where });2097 yield* core.deleteMany("tool", { where });
2065 yield* core.deleteMany("definition", { where });2098 yield* core.deleteMany("definition", { where });
2066 yield* stampSynced;2099 yield* stampSynced(existingRow);
2067 }),2100 }),
2068 );2101 );
2069 return [];2102 return [];
@@ -2092,11 +2125,33 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 22 unmodified lines
2092 // over a transient outage — and stamp the sync time anyway so a down2125 // over a transient outage — and stamp the sync time anyway so a down
2093 // server isn't re-dialed on every read; the freshness TTL re-attempts2126 // server isn't re-dialed on every read; the freshness TTL re-attempts
2094 // later.2127 // later.
2095 yield* stampSynced;2128 const reason = syncHealthReason(result);
2129 yield* stampSyncedWithHealth(reason);
2130 yield* Effect.logWarning("executor tool sync preserved catalog", {
2131 reason,
2132 integration: String(ref.integration),
2133 connection: String(ref.name),
2134 });
2096 const keptRows = yield* core.findMany("tool", { where });2135 const keptRows = yield* core.findMany("tool", { where });
2097 return keptRows.map((row) => rowToTool(row as ConnectionToolRow));2136 return keptRows.map((row) => rowToTool(row as ConnectionToolRow));
2098 }2137 }
2099 2138
2139 if (mode === "background" && result.tools.length === 0) {
2140 const keptRows = yield* core.findMany("tool", { where });
2141 if (keptRows.length > 0) {
2142 const reason =
2143 "background tool sync produced an authoritative empty catalog for a connection with existing tools";
2144 yield* stampSyncedWithHealth(reason);
2145 yield* Effect.logWarning("executor tool sync preserved nonzero catalog", {
2146 reason,
2147 integration: String(ref.integration),
2148 connection: String(ref.name),
2149 existingToolCount: keptRows.length,
2150 });
2151 return keptRows.map((row) => rowToTool(row as ConnectionToolRow));
2152 }
2153 }
2154
2100 const now = new Date();2155 const now = new Date();
2101 const toolRows = result.tools.map((tool: ToolDef) => ({2156 const toolRows = result.tools.map((tool: ToolDef) => ({
2102 tenant: keys.tenant,2157 tenant: keys.tenant,
@@ -2132,7 +2187,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 29 unmodified lines
2132 yield* core.deleteMany("definition", { where });2187 yield* core.deleteMany("definition", { where });
2133 yield* core.createMany("tool", toolRows);2188 yield* core.createMany("tool", toolRows);
2134 yield* core.createMany("definition", definitionRows);2189 yield* core.createMany("definition", definitionRows);
2135 yield* stampSynced;2190 yield* stampSynced(existingRow);
2136 }),2191 }),
2137 );2192 );
2138 2193
@@ -2965,11 +3020,15 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 826 unmodified lines
2965 syncedAt < cutoff;3020 syncedAt < cutoff;
2966 if (!staleMarked && !configRevised && !expired) continue;3021 if (!staleMarked && !configRevised && !expired) continue;
2967 3022
2968 yield* produceConnectionTools(integrationRow, {3023 yield* produceConnectionTools(
2969 owner: connection.owner as Owner,3024 integrationRow,
2970 integration: IntegrationSlug.make(connection.integration),3025 {
2971 name: ConnectionName.make(connection.name),3026 owner: connection.owner as Owner,
2972 }).pipe(3027 integration: IntegrationSlug.make(connection.integration),
3028 name: ConnectionName.make(connection.name),
3029 },
3030 "background",
3031 ).pipe(
2973 Effect.catch(() => Effect.succeed([] as readonly Tool[])),3032 Effect.catch(() => Effect.succeed([] as readonly Tool[])),
2974 Effect.withSpan("executor.tools.sync_stale", {3033 Effect.withSpan("executor.tools.sync_stale", {
2975 attributes: {3034 attributes: {
packages/core/sdk/src/plugin.ts+3−0
@@ -297,6 +297,9 @@ export interface ResolveToolsResult { · 296 unmodified lines
297 * listing is authoritative, including a genuine "this source has zero297 * listing is authoritative, including a genuine "this source has zero
298 * tools". */298 * tools". */
299 readonly incomplete?: boolean;299 readonly incomplete?: boolean;
300 /** Human-readable reason for an incomplete listing. Persisted by core when it
301 * preserves the prior catalog so operators can see why data is stale. */
302 readonly incompleteReason?: string;
300}303}
301 304
302// ---------------------------------------------------------------------------305// ---------------------------------------------------------------------------
packages/plugins/graphql/src/sdk/plugin.ts+23−4
@@ -975,10 +975,17 @@ export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => { · 974 unmodified lines
975 readonly httpClientLayer: Layer.Layer<HttpClient.HttpClient>;975 readonly httpClientLayer: Layer.Layer<HttpClient.HttpClient>;
976 }) =>976 }) =>
977 Effect.gen(function* () {977 Effect.gen(function* () {
978 const incomplete = (reason: string) => ({
979 tools: [] as readonly ToolDef[],
980 incomplete: true,
981 incompleteReason: reason,
982 });
978 const decoded = yield* decodeGraphqlIntegrationConfig(config).pipe(Effect.option);983 const decoded = yield* decodeGraphqlIntegrationConfig(config).pipe(Effect.option);
979 if (Option.isNone(decoded)) return { tools: [] };984 if (Option.isNone(decoded)) return { tools: [] };
980 const graphqlConfig = decoded.value;985 const graphqlConfig = decoded.value;
981 const introspectionJson = yield* loadIntrospectionJson(storage, graphqlConfig);986 const introspectionJson = yield* loadIntrospectionJson(storage, graphqlConfig).pipe(
987 Effect.catch(() => Effect.succeed(null)),
988 );
982 // Live introspection (no stored snapshot) needs the connection's989 // Live introspection (no stored snapshot) needs the connection's
983 // credential inputs for auth-required endpoints; resolve them lazily.990 // credential inputs for auth-required endpoints; resolve them lazily.
984 const values =991 const values =
@@ -994,15 +1001,27 @@ export const graphqlPlugin = definePlugin((options?: GraphqlPluginOptions) => { · 9 unmodified lines
994 template,1001 template,
995 options?.httpClientLayer ?? httpClientLayer,1002 options?.httpClientLayer ?? httpClientLayer,
996 ).pipe(Effect.option);1003 ).pipe(Effect.option);
997 if (Option.isNone(introspection)) return { tools: [] };1004 if (Option.isNone(introspection)) {
1005 return incomplete("GraphQL introspection could not be loaded.");
1006 }
998 const extracted = yield* extract(introspection.value).pipe(Effect.option);1007 const extracted = yield* extract(introspection.value).pipe(Effect.option);
999 if (Option.isNone(extracted)) return { tools: [] };1008 if (Option.isNone(extracted)) {
1009 return incomplete("GraphQL introspection result could not be converted to tools.");
1010 }
1000 const prepared = prepareOperations(extracted.value.result.fields, introspection.value);1011 const prepared = prepareOperations(extracted.value.result.fields, introspection.value);
1001 return {1012 return {
1002 tools: buildToolDefs(prepared),1013 tools: buildToolDefs(prepared),
1003 definitions: extracted.value.definitions,1014 definitions: extracted.value.definitions,
1004 };1015 };
1005 }).pipe(Effect.catch(() => Effect.succeed({ tools: [] as readonly ToolDef[] }))),1016 }).pipe(
1017 Effect.catch(() =>
1018 Effect.succeed({
1019 tools: [] as readonly ToolDef[],
1020 incomplete: true,
1021 incompleteReason: "GraphQL tool catalog could not be resolved.",
1022 }),
1023 ),
1024 ),
1006 1025
1007 // -----------------------------------------------------------------------1026 // -----------------------------------------------------------------------
1008 // Invoke one of a connection's tools. Look up the operation by integration1027 // Invoke one of a connection's tools. Look up the operation by integration
packages/plugins/openapi/src/sdk/backing.ts+18−5
@@ -579,24 +579,37 @@ export const resolveOpenApiBackedTools = ({ · 578 unmodified lines
579 readonly storage: OpenapiStore;579 readonly storage: OpenapiStore;
580}): Effect.Effect<ResolveToolsResult, StorageFailure> =>580}): Effect.Effect<ResolveToolsResult, StorageFailure> =>
581 Effect.gen(function* () {581 Effect.gen(function* () {
582 const incomplete = (reason: string): ResolveToolsResult => ({
583 tools: [],
584 definitions: {},
585 incomplete: true,
586 incompleteReason: reason,
587 });
582 const openApiConfig = decodeOpenApiIntegrationConfig(config);588 const openApiConfig = decodeOpenApiIntegrationConfig(config);
583 if (!openApiConfig) return { tools: [], definitions: {} };589 if (!openApiConfig) return { tools: [], definitions: {} };
584 if (openApiConfig.specHash != null) {590 if (openApiConfig.specHash != null) {
585 const defsJson = yield* storage.getDefs(openApiConfig.specHash);591 const defsJson = yield* storage
592 .getDefs(openApiConfig.specHash)
593 .pipe(Effect.catch(() => Effect.succeed(null)));
586 if (defsJson != null) {594 if (defsJson != null) {
587 const definitions = Option.getOrNull(decodeDefsJson(defsJson));595 const definitions = Option.getOrNull(decodeDefsJson(defsJson));
588 if (definitions != null) {596 if (definitions != null) {
589 const ops = yield* storage.listOperations(String(integration.slug));597 const ops = yield* storage
598 .listOperations(String(integration.slug))
599 .pipe(Effect.catch(() => Effect.succeed(null)));
600 if (ops == null) return incomplete("OpenAPI operation bindings could not be loaded.");
590 return { tools: ops.map(toolDefFromStoredOperation), definitions };601 return { tools: ops.map(toolDefFromStoredOperation), definitions };
591 }602 }
592 }603 }
593 }604 }
594 const specText = yield* loadOpenApiSpecText(storage, openApiConfig);605 const specText = yield* loadOpenApiSpecText(storage, openApiConfig).pipe(
595 if (specText == null) return { tools: [], definitions: {} };606 Effect.catch(() => Effect.succeed(null)),
607 );
608 if (specText == null) return incomplete("OpenAPI spec blob could not be loaded.");
596 const compiled = yield* compileOpenApiSpecCached(openApiConfig.specHash, specText).pipe(609 const compiled = yield* compileOpenApiSpecCached(openApiConfig.specHash, specText).pipe(
597 Effect.catch(() => Effect.succeed(null)),610 Effect.catch(() => Effect.succeed(null)),
598 );611 );
599 if (!compiled) return { tools: [], definitions: {} };612 if (!compiled) return incomplete("OpenAPI spec could not be parsed.");
600 return {613 return {
601 tools: openApiToolDefsFromCompiled(compiled),614 tools: openApiToolDefsFromCompiled(compiled),
602 definitions: compiled.hoistedDefs,615 definitions: compiled.hoistedDefs,
packages/plugins/openapi/src/sdk/spec-blob.test.ts+229−0
@@ -32,6 +32,7 @@ import { variable } from "@executor-js/sdk/http-auth"; · 31 unmodified lines
32import { openApiPlugin } from "./plugin";32import { openApiPlugin } from "./plugin";
33import type { OpenapiStore } from "./store";33import type { OpenapiStore } from "./store";
34import type { AuthenticationInput } from "./types";34import type { AuthenticationInput } from "./types";
35import { defsBlobKey, specBlobKey } from "./store";
35import {36import {
36 makeOpenApiHttpApiTestSourceConfig,37 makeOpenApiHttpApiTestSourceConfig,
37 serveOpenApiHttpApiTestServer,38 serveOpenApiHttpApiTestServer,
@@ -65,12 +66,49 @@ const specText = () => { · 27 unmodified lines
65 return spec.url;66 return spec.url;
66};67};
67 68
69const specTextWithDefinition = () =>
70 JSON.stringify({
71 openapi: "3.1.0",
72 info: { title: "Catalog", version: "1.0.0" },
73 paths: {
74 "/items": {
75 get: {
76 operationId: "listItems",
77 responses: {
78 "200": {
79 description: "ok",
80 content: {
81 "application/json": {
82 schema: {
83 type: "array",
84 items: { $ref: "#/components/schemas/Item" },
85 },
86 },
87 },
88 },
89 },
90 },
91 },
92 },
93 components: {
94 schemas: {
95 Item: {
96 type: "object",
97 properties: { id: { type: "string" } },
98 required: ["id"],
99 },
100 },
101 },
102 });
103
68const apiKeyTemplate: AuthenticationInput = {104const apiKeyTemplate: AuthenticationInput = {
69 slug: AuthTemplateSlug.make("apiKey"),105 slug: AuthTemplateSlug.make("apiKey"),
70 type: "apiKey",106 type: "apiKey",
71 headers: { "x-api-key": [variable("token")] },107 headers: { "x-api-key": [variable("token")] },
72};108};
73 109
110const openApiBlobNamespace = "o:test-tenant/openapi";
111
74describe("OpenAPI plugin — spec blob storage", () => {112describe("OpenAPI plugin — spec blob storage", () => {
75 it.effect("addSpec stores a content pointer, not the inline spec text", () =>113 it.effect("addSpec stores a content pointer, not the inline spec text", () =>
76 Effect.scoped(114 Effect.scoped(
@@ -125,6 +163,197 @@ describe("OpenAPI plugin — spec blob storage", () => { · 48 unmodified lines
125 ),163 ),
126 );164 );
127 165
166 it.effect(
167 "stale sync preserves tools and definitions when the spec blob parses to a non-object",
168 () =>
169 Effect.scoped(
170 Effect.gen(function* () {
171 const config = makeTestConfig({ plugins: testPlugins() });
172 const executor = yield* createExecutor(config);
173 const text = specTextWithDefinition();
174 const hash = yield* sha256Hex(text);
175
176 yield* executor.openapi.addSpec({
177 spec: { kind: "blob", value: text },
178 slug: "corrupt_blob",
179 });
180 yield* executor.connections.create({
181 owner: "org",
182 name: ConnectionName.make("main"),
183 integration: IntegrationSlug.make("corrupt_blob"),
184 template: AuthTemplateSlug.make("none"),
185 values: {},
186 });
187
188 const beforeTools = yield* executor.tools.list({
189 integration: IntegrationSlug.make("corrupt_blob"),
190 });
191 const beforeDefinitions = yield* Effect.promise(() =>
192 config.db.findMany("definition", {
193 where: (b) => b("integration", "=", "corrupt_blob"),
194 }),
195 );
196 expect(beforeTools.length).toBeGreaterThan(0);
197 expect(beforeDefinitions.length).toBeGreaterThan(0);
198
199 yield* Effect.promise(() =>
200 config.db.updateMany("blob", {
201 where: (b) =>
202 b.and(b("namespace", "=", openApiBlobNamespace), b("key", "=", specBlobKey(hash))),
203 set: { value: JSON.stringify("not an object") },
204 }),
205 );
206 yield* Effect.promise(() =>
207 config.db.deleteMany("blob", {
208 where: (b) =>
209 b.and(b("namespace", "=", openApiBlobNamespace), b("key", "=", defsBlobKey(hash))),
210 }),
211 );
212 yield* Effect.promise(() =>
213 config.db.updateMany("connection", {
214 where: (b) => b.and(b("integration", "=", "corrupt_blob"), b("name", "=", "main")),
215 set: { tools_synced_at: null },
216 }),
217 );
218
219 const afterTools = yield* executor.tools.list({
220 integration: IntegrationSlug.make("corrupt_blob"),
221 });
222 const afterDefinitions = yield* Effect.promise(() =>
223 config.db.findMany("definition", {
224 where: (b) => b("integration", "=", "corrupt_blob"),
225 }),
226 );
227 const connection = yield* executor.connections.get({
228 owner: "org",
229 integration: IntegrationSlug.make("corrupt_blob"),
230 name: ConnectionName.make("main"),
231 });
232
233 expect(afterTools.map((tool) => String(tool.name)).sort()).toEqual(
234 beforeTools.map((tool) => String(tool.name)).sort(),
235 );
236 expect(afterDefinitions).toHaveLength(beforeDefinitions.length);
237 expect(connection?.lastHealth).toMatchObject({
238 status: "degraded",
239 detail: expect.stringContaining("OpenAPI spec could not be parsed"),
240 });
241 }),
242 ),
243 );
244
245 it.effect("stale sync preserves tools and definitions when the spec blob is missing", () =>
246 Effect.scoped(
247 Effect.gen(function* () {
248 const config = makeTestConfig({ plugins: testPlugins() });
249 const executor = yield* createExecutor(config);
250 const text = specTextWithDefinition();
251 const hash = yield* sha256Hex(text);
252
253 yield* executor.openapi.addSpec({
254 spec: { kind: "blob", value: text },
255 slug: "missing_blob",
256 });
257 yield* executor.connections.create({
258 owner: "org",
259 name: ConnectionName.make("main"),
260 integration: IntegrationSlug.make("missing_blob"),
261 template: AuthTemplateSlug.make("none"),
262 values: {},
263 });
264
265 const beforeTools = yield* executor.tools.list({
266 integration: IntegrationSlug.make("missing_blob"),
267 });
268 const beforeDefinitions = yield* Effect.promise(() =>
269 config.db.findMany("definition", {
270 where: (b) => b("integration", "=", "missing_blob"),
271 }),
272 );
273 expect(beforeTools.length).toBeGreaterThan(0);
274 expect(beforeDefinitions.length).toBeGreaterThan(0);
275
276 yield* Effect.promise(() =>
277 config.db.deleteMany("blob", {
278 where: (b) =>
279 b.and(
280 b("namespace", "=", openApiBlobNamespace),
281 b("key", "in", [specBlobKey(hash), defsBlobKey(hash)]),
282 ),
283 }),
284 );
285 yield* Effect.promise(() =>
286 config.db.updateMany("connection", {
287 where: (b) => b.and(b("integration", "=", "missing_blob"), b("name", "=", "main")),
288 set: { tools_synced_at: null },
289 }),
290 );
291
292 const afterTools = yield* executor.tools.list({
293 integration: IntegrationSlug.make("missing_blob"),
294 });
295 const afterDefinitions = yield* Effect.promise(() =>
296 config.db.findMany("definition", {
297 where: (b) => b("integration", "=", "missing_blob"),
298 }),
299 );
300 const connection = yield* executor.connections.get({
301 owner: "org",
302 integration: IntegrationSlug.make("missing_blob"),
303 name: ConnectionName.make("main"),
304 });
305
306 expect(afterTools.map((tool) => String(tool.name)).sort()).toEqual(
307 beforeTools.map((tool) => String(tool.name)).sort(),
308 );
309 expect(afterDefinitions).toHaveLength(beforeDefinitions.length);
310 expect(connection?.lastHealth).toMatchObject({
311 status: "degraded",
312 detail: expect.stringContaining("OpenAPI spec blob could not be loaded"),
313 });
314 }),
315 ),
316 );
317
318 it.effect("explicit spec refresh accepts a valid OpenAPI document with zero operations", () =>
319 Effect.scoped(
320 Effect.gen(function* () {
321 const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() }));
322 const emptySpec = JSON.stringify({
323 openapi: "3.1.0",
324 info: { title: "Empty", version: "1.0.0" },
325 paths: {},
326 });
327
328 yield* executor.openapi.addSpec({
329 spec: { kind: "blob", value: specText() },
330 slug: "healthy_zero",
331 });
332 yield* executor.connections.create({
333 owner: "org",
334 name: ConnectionName.make("main"),
335 integration: IntegrationSlug.make("healthy_zero"),
336 template: AuthTemplateSlug.make("none"),
337 values: {},
338 });
339 expect(
340 (yield* executor.tools.list({ integration: IntegrationSlug.make("healthy_zero") }))
341 .length,
342 ).toBeGreaterThan(0);
343
344 const update = yield* executor.openapi.updateSpec("healthy_zero", {
345 spec: { kind: "blob", value: emptySpec },
346 });
347 const tools = yield* executor.tools.list({
348 integration: IntegrationSlug.make("healthy_zero"),
349 });
350
351 expect(update.toolCount).toBe(0);
352 expect(tools).toEqual([]);
353 }),
354 ),
355 );
356
128 it.effect("resolveTools reads the spec from the store, never an inline field", () =>357 it.effect("resolveTools reads the spec from the store, never an inline field", () =>
129 Effect.gen(function* () {358 Effect.gen(function* () {
130 const plugin = openApiPlugin({ httpClientLayer: FetchHttpClient.layer });359 const plugin = openApiPlugin({ httpClientLayer: FetchHttpClient.layer });