From 6dd93f934144dd994e73612fe5dabc6b45a22437 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 20 Aug 2026 09:46:21 -0700 Subject: [PATCH 1/8] Revalidate stale non-healthy connection verdicts on agent reads and successful invocations --- .changeset/health-verdict-revalidate.md | 11 ++ packages/core/sdk/src/connections.test.ts | 190 ++++++++++++++++++++++ packages/core/sdk/src/core-tools.ts | 61 ++++++- packages/core/sdk/src/executor.ts | 67 ++++++-- packages/core/sdk/src/health-check.ts | 9 + packages/core/sdk/src/plugin.ts | 11 ++ 6 files changed, 325 insertions(+), 24 deletions(-) create mode 100644 .changeset/health-verdict-revalidate.md diff --git a/.changeset/health-verdict-revalidate.md b/.changeset/health-verdict-revalidate.md new file mode 100644 index 000000000..8d466e497 --- /dev/null +++ b/.changeset/health-verdict-revalidate.md @@ -0,0 +1,11 @@ +--- +"@executor-js/sdk": patch +--- + +**Stale "unhealthy" verdicts no longer wait for a manual "Check now"** + +A connection's persisted health verdict was only ever re-checked from the web UI, so after one bad probe (a transient upstream error, a refresh that failed once) agents reading `connections.list` kept reporting "unhealthy, reconnect" for a connection that worked fine — invocation auto-refreshes OAuth tokens — until a human opened the page and clicked "Check now". + +Two repair paths make the verdict track reality on its own. The agent-facing `connections.list` now re-runs the same probe as "Check now" before reporting a non-healthy verdict older than a minute, so recovery shows on the next read while repeated lists collapse to one probe per window. And a successful tool invocation through a connection wearing a non-healthy verdict flips it back to healthy — real traffic is stronger evidence than any probe. Tool-sync failure verdicts and grants the authorization server has rejected as `invalid_grant` are deliberately left alone: the first is cleared only by a successful sync, and the second genuinely requires a reconnect. + +`PluginCtx.connections` gains `checkHealth`, the same probe-with-freshness-window the executor surface already exposed. diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 7ed046f65..274516ea6 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -14,6 +14,7 @@ import { createExecutor } from "./executor"; import { definePlugin } from "./plugin"; import type { CredentialProvider } from "./provider"; import { makeTestConfig, makeTestExecutor } from "./testing"; +import { ToolResult } from "./tool-result"; // removed: v1 connection-refresh lifecycle, ConnectionProvider.refresh, // SecretProvider, accessToken token-refresh + in-flight dedup tests — the v2 @@ -718,3 +719,192 @@ describe("execute over a connection", () => { }), ); }); + +// --------------------------------------------------------------------------- +// Sticky-verdict repair: agents read `lastHealth` through coreTools +// connections.list, and nothing else ever re-probes a persisted verdict, so a +// transient failure used to read as "unhealthy, reconnect" until a human +// clicked "Check now". These cover the two repair paths: read-time +// revalidation on the agent list, and heal-on-use from a successful +// invocation. +// --------------------------------------------------------------------------- + +const CORE_LIST = ToolAddress.make("executor.coreTools.connections.list"); +const STALE_MS = 5 * 60 * 1000; + +type ListedConnections = { + readonly connections: readonly { + readonly name: string; + readonly lastHealth: { readonly status: string; readonly detail?: string } | null; + }[]; +}; + +const makeHealthHarness = () => { + const counters = { probes: 0 }; + const plugin = definePlugin(() => ({ + id: "healthdemo" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }), + invokeTool: ({ toolRow, credential, args }) => + Effect.succeed( + (args as { fail?: boolean }).fail === true + ? ToolResult.fail({ code: "upstream_error", message: "boom" }) + : { ran: toolRow.name, value: credential.value }, + ), + checkHealth: () => + Effect.sync(() => { + counters.probes += 1; + return { status: "healthy" as const, checkedAt: Date.now(), detail: "probe ok" }; + }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), + }), + }))(); + + return Effect.gen(function* () { + const config = makeTestConfig({ + plugins: [plugin] as const, + coreTools: { webBaseUrl: "http://localhost:3000" }, + }); + const executor = yield* createExecutor(config); + yield* executor.healthdemo.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + const stamp = (set: Record) => + Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set, + }), + ); + const persisted = () => + executor.connections.get({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + return { executor, counters, stamp, persisted } as const; + }); +}; + +describe("agent read revalidation (coreTools connections.list)", () => { + it.effect("re-probes a stale non-healthy verdict and reports + persists the fresh one", () => + Effect.gen(function* () { + const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + + const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.status).toBe("healthy"); + expect(counters.probes).toBe(1); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("healthy"); + }), + ); + + it.effect("serves a fresh non-healthy verdict without re-probing", () => + Effect.gen(function* () { + const { executor, counters, stamp } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now(), detail: "HTTP 401" }, + }); + + const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.status).toBe("expired"); + expect(counters.probes).toBe(0); + }), + ); + + it.effect("never probes a healthy verdict, however old", () => + Effect.gen(function* () { + const { executor, counters, stamp } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "healthy", checkedAt: Date.now() - STALE_MS, detail: "probe ok" }, + }); + + const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.status).toBe("healthy"); + expect(counters.probes).toBe(0); + }), + ); + + it.effect("leaves tool-sync failure verdicts for sync to clear", () => + Effect.gen(function* () { + const { executor, counters, stamp } = yield* makeHealthHarness(); + const detail = "Tool sync failing: plugin returned an incomplete tool catalog"; + yield* stamp({ + last_health: { status: "degraded", checkedAt: Date.now() - STALE_MS, detail }, + }); + + const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.detail).toBe(detail); + expect(counters.probes).toBe(0); + }), + ); +}); + +describe("heal-on-use", () => { + it.effect("a successful invocation flips a stale non-healthy verdict to healthy", () => + Effect.gen(function* () { + const { executor, stamp, persisted } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ + status: "healthy", + detail: "Tool invocation succeeded.", + }); + }), + ); + + it.effect("an explicit tool failure does not heal", () => + Effect.gen(function* () { + const { executor, stamp, persisted } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), { fail: true }); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("expired"); + }), + ); + + it.effect("a grant recorded invalid_grant-dead is not healed by a lingering token", () => + Effect.gen(function* () { + const { executor, stamp, persisted } = yield* makeHealthHarness(); + yield* stamp({ + provider_state: { oauthReauthRequiredAt: Date.now() }, + last_health: { + status: "expired", + checkedAt: Date.now() - STALE_MS, + detail: "invalid_grant", + }, + }); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("expired"); + }), + ); +}); diff --git a/packages/core/sdk/src/core-tools.ts b/packages/core/sdk/src/core-tools.ts index e6dc78ce7..f55a106f9 100644 --- a/packages/core/sdk/src/core-tools.ts +++ b/packages/core/sdk/src/core-tools.ts @@ -22,7 +22,7 @@ import { type Owner, } from "./ids"; import { definePlugin, tool, type StaticToolSchema } from "./plugin"; -import { HealthCheckResult } from "./health-check"; +import { HealthCheckResult, isToolSyncHealth } from "./health-check"; import { ToolPolicyActionSchema } from "./policies"; import type { Tool } from "./tool"; @@ -408,6 +408,26 @@ const connectionToListItem = (connection: Connection, verbose: boolean) => ({ ...(verbose ? { oauthScope: connection.oauthScope ?? null } : {}), }); +/** How long a non-healthy persisted verdict may be served to an agent before + * it is re-verified. Verdicts are sticky — nothing re-probes them between UI + * visits — so without read-time revalidation an agent keeps reporting + * "unhealthy, reconnect" for a connection that recovered long ago (or was + * never really down: invocation auto-refreshes OAuth tokens, so a stale + * "expired" verdict often describes a working connection). The window is + * short so recovery shows on the next read, but bounds repeated lists from + * hammering a genuinely-down upstream. Healthy verdicts are deliberately + * served as-is: they mislead no one into reconnect guidance, and the UI + * owns their background revalidation. */ +const NON_HEALTHY_REVALIDATE_MS = 60 * 1000; + +/** Whether an agent read must re-verify a persisted verdict before reporting + * it. Only probe-refutable non-healthy verdicts qualify: `unknown` and + * missing verdicts carry no reconnect implication, and a tool-sync failure + * verdict cannot be refuted by a credential probe (a successful sync clears + * it instead). */ +const needsAgentReadRevalidation = (last: HealthCheckResult | null | undefined): boolean => + (last?.status === "expired" || last?.status === "degraded") && !isToolSyncHealth(last); + const toolToOutput = (toolRow: Tool) => ({ address: String(toolRow.address), owner: toolRow.owner, @@ -593,20 +613,45 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { inputSchema: ConnectionsListInputStd, outputSchema: ConnectionsListOutputStd, execute: (input: typeof ConnectionsListInput.Type, { ctx }) => - Effect.map( - ctx.connections.list({ + Effect.gen(function* () { + const connections = yield* ctx.connections.list({ integration: input.integration === undefined ? undefined : IntegrationSlug.make(input.integration), owner: input.owner === undefined ? undefined : (input.owner as Owner), - }), - (connections) => ({ - connections: connections.map((connection) => + }); + // Re-verify sticky non-healthy verdicts before reporting them + // (the same probe as the UI's "Check now", server-cached via + // `ifStaleMs` so repeated lists collapse to one probe per + // window). Quiet on probe failure: the persisted verdict is + // still the best known state, exactly like the UI surfaces. + const revalidated = yield* Effect.forEach( + connections, + (connection) => + needsAgentReadRevalidation(connection.lastHealth) + ? ctx.connections + .checkHealth( + { + owner: connection.owner, + integration: connection.integration, + name: connection.name, + }, + { ifStaleMs: NON_HEALTHY_REVALIDATE_MS }, + ) + .pipe( + Effect.map((health) => ({ ...connection, lastHealth: health })), + Effect.catch(() => Effect.succeed(connection)), + ) + : Effect.succeed(connection), + { concurrency: 4 }, + ); + return { + connections: revalidated.map((connection) => connectionToListItem(connection, input.verbose === true), ), - }), - ), + }; + }), }), tool({ name: "connections.create", diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 04c742063..eec91a99e 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -27,7 +27,12 @@ import type { UpdateConnectionInput, ValidateConnectionInput, } from "./connection"; -import { HealthCheckResult, HealthCheckSpec } from "./health-check"; +import { + HealthCheckResult, + HealthCheckSpec, + isToolSyncHealth, + toolSyncHealthDetailPrefix, +} from "./health-check"; import type { HealthCheckCandidate } from "./health-check"; import { ARTIFACT_SUMMARY_COLUMNS, @@ -168,7 +173,7 @@ import { type OAuthEndpointUrlPolicy, } from "./oauth-helpers"; import { connectionIdentifier } from "./connection-name-identifier"; -import { annotateToolResultOutcome } from "./tool-result"; +import { annotateToolResultOutcome, isToolResult } from "./tool-result"; import { isUnauthorizedToolFailure } from "./auth-tool-failure"; const PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE = 90; @@ -2541,8 +2546,6 @@ export const createExecutor = ({ status: "degraded", checkedAt: Date.now(), @@ -2576,8 +2579,6 @@ export const createExecutor = => @@ -4541,16 +4571,20 @@ export const createExecutor = Effect.succeed(null)), - ); - if (!refreshed) return first; - yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.retried": true }); - return yield* invokeWith(refreshed); + const result = yield* Effect.gen(function* () { + if (!isUnauthorizedToolFailure(first)) return first; + const refreshed = yield* forceRefreshConnectionValues(connectionRow).pipe( + // A failed re-mint is not this call's failure to report: the upstream + // already produced an auth failure with recovery guidance, which is + // strictly more actionable than a refresh-plumbing error. Keep it. + Effect.catchTag("CredentialResolutionError", () => Effect.succeed(null)), + ); + if (!refreshed) return first; + yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.retried": true }); + return yield* invokeWith(refreshed); + }); + yield* healPersistedHealthOnUse(connectionRow, result); + return result; }).pipe( // Expected tool failures (`ToolResult.fail`) resolve through the // success channel, so the tracer alone would record them as healthy @@ -4709,6 +4743,7 @@ export const createExecutor = connectionsUpdate(ref, input), remove: (ref) => connectionsRemove(ref), refresh: (ref) => connectionsRefresh(ref), + checkHealth: (ref, options) => connectionCheckHealth(ref, options), markToolsStale: (ref) => connectionsMarkToolsStale(ref), resolveValue: (ref) => resolveConnectionValueByRef(ref), }, diff --git a/packages/core/sdk/src/health-check.ts b/packages/core/sdk/src/health-check.ts index d553f994f..572a4b207 100644 --- a/packages/core/sdk/src/health-check.ts +++ b/packages/core/sdk/src/health-check.ts @@ -84,6 +84,15 @@ export const HealthCheckResult = Schema.Struct({ }); export type HealthCheckResult = typeof HealthCheckResult.Type; +/** Detail prefix that marks a verdict as produced by tool-catalog sync, not a + * credential probe. Shared vocabulary: sync stamps it, and the surfaces that + * auto-revalidate verdicts skip these — a credential probe cannot refute a + * failed tool sync, and a later successful sync clears the verdict itself. */ +export const toolSyncHealthDetailPrefix = "Tool sync failing"; + +export const isToolSyncHealth = (result: HealthCheckResult | null | undefined): boolean => + result?.detail?.startsWith(toolSyncHealthDetailPrefix) === true; + // --------------------------------------------------------------------------- // HealthCheckCandidate: one operation the user can pick as the health check, // projected from the plugin's stored operations. The editor lists these ranked diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts index 6d4dbc23b..80bb84764 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -231,6 +231,17 @@ export interface PluginCtx { readonly Tool[], ConnectionNotFoundError | IntegrationNotFoundError | StorageFailure >; + /** Run the integration's declared health check against a saved connection + * and persist the verdict. `ifStaleMs` serves the persisted verdict when + * younger than that window, so concurrent readers collapse to one probe; + * omit it to always probe. */ + readonly checkHealth: ( + ref: ConnectionRef, + options?: { readonly ifStaleMs?: number }, + ) => Effect.Effect< + HealthCheckResult, + ConnectionNotFoundError | IntegrationNotFoundError | StorageFailure + >; /** Mark a connection's persisted tool catalog stale (clears its sync * stamp) without re-listing inline. The next tools read re-produces it. * For signals that arrive mid-invocation — e.g. an MCP server sending From 8216bf5dde799a12e6f02c8610c7d406e79c3184 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 02:19:59 -0700 Subject: [PATCH 2/8] Align revalidation tests with the compact connection list and the missing-credential guard --- packages/core/sdk/src/connections.test.ts | 29 +++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 40307a52a..e5442ab2b 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1206,7 +1206,7 @@ describe("agent read revalidation (coreTools connections.list)", () => { it.effect("leaves tool-sync failure verdicts for sync to clear", () => Effect.gen(function* () { - const { executor, counters, stamp } = yield* makeHealthHarness(); + const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); const detail = "Tool sync failing: plugin returned an incomplete tool catalog"; yield* stamp({ last_health: { status: "degraded", checkedAt: Date.now() - STALE_MS, detail }, @@ -1214,8 +1214,14 @@ describe("agent read revalidation (coreTools connections.list)", () => { const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; const listed = out.connections.find((c) => c.name === "main"); - expect(listed?.lastHealth?.detail).toBe(detail); + expect(listed?.lastHealth?.status).toBe("degraded"); expect(counters.probes).toBe(0); + + // The compact list shape omits `detail`, so read the untouched verdict + // off the row: a tool-sync failure is cleared by a successful sync, not + // by a credential probe. + const row = yield* persisted(); + expect(row?.lastHealth?.detail).toBe(detail); }), ); }); @@ -1270,4 +1276,23 @@ describe("heal-on-use", () => { expect(row?.lastHealth?.status).toBe("expired"); }), ); + + it.effect("a call whose credential no longer resolves is not healed", () => + Effect.gen(function* () { + const { executor, stamp, persisted } = yield* makeHealthHarness(); + // The stored credential is gone from the provider. Rendering skips a + // missing placement, so an upstream that answers unauthenticated still + // succeeds — that success says nothing about a credential that no longer + // exists, and healing from it would tell the user to stop reconnecting. + yield* stamp({ + item_ids: { token: "vanished-item" }, + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("expired"); + }), + ); }); From bfd0fa8161de93654923edab00f249489de18b3b Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:43:28 -0700 Subject: [PATCH 3/8] Coalesce concurrent health probes and guard verdict writes against newer state --- packages/core/sdk/src/connections.test.ts | 145 ++++++++++++- packages/core/sdk/src/core-tools.ts | 12 +- packages/core/sdk/src/executor.ts | 250 +++++++++++++++++----- 3 files changed, 342 insertions(+), 65 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index e5442ab2b..739cd4750 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1102,8 +1102,17 @@ type ListedConnections = { }[]; }; -const makeHealthHarness = () => { +const makeHealthHarness = (options?: { + /** Replaces the default instant-healthy probe. Entries are still counted in + * `counters.probes`, so a Deferred-gated probe lets a test hold every + * in-flight health check open and count how many actually started. */ + readonly probe?: Effect.Effect; +}) => { const counters = { probes: 0 }; + // Runs inside every invocation before it returns, so a test can interleave + // a concurrent write (e.g. a refresh discovering invalid_grant) between the + // row load and the heal-on-use decision. + const hooks = { onInvoke: Effect.void as Effect.Effect }; const plugin = definePlugin(() => ({ id: "healthdemo" as const, credentialProviders: [memoryProvider()], @@ -1111,15 +1120,19 @@ const makeHealthHarness = () => { resolveTools: () => Effect.succeed({ tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }), invokeTool: ({ toolRow, credential, args }) => - Effect.succeed( + Effect.as( + hooks.onInvoke, (args as { fail?: boolean }).fail === true ? ToolResult.fail({ code: "upstream_error", message: "boom" }) : { ran: toolRow.name, value: credential.value }, ), checkHealth: () => - Effect.sync(() => { + Effect.suspend(() => { counters.probes += 1; - return { status: "healthy" as const, checkedAt: Date.now(), detail: "probe ok" }; + return ( + options?.probe ?? + Effect.succeed({ status: "healthy" as const, checkedAt: Date.now(), detail: "probe ok" }) + ); }), extension: (ctx) => ({ seed: () => @@ -1154,7 +1167,7 @@ const makeHealthHarness = () => { integration: INTEG, name: ConnectionName.make("main"), }); - return { executor, counters, stamp, persisted } as const; + return { executor, counters, stamp, persisted, hooks } as const; }); }; @@ -1204,6 +1217,86 @@ describe("agent read revalidation (coreTools connections.list)", () => { }), ); + it.effect("concurrent lists past the freshness gate collapse to one probe", () => + Effect.gen(function* () { + const gate = yield* Deferred.make(); + const { executor, counters, stamp, persisted } = yield* makeHealthHarness({ + probe: Deferred.await(gate).pipe( + Effect.map(() => ({ + status: "healthy" as const, + checkedAt: Date.now(), + detail: "probe ok", + })), + ), + }); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + + const lists = yield* Effect.forkChild( + Effect.all( + Array.from({ length: 5 }, () => executor.execute(CORE_LIST, {})), + { concurrency: "unbounded" }, + ), + ); + // The probe is held open by the gate, so NOTHING has been persisted yet: + // every one of the five lists must pass the freshness check and reach + // the probe path. Give them real time to get there — the counter then + // says how many probes actually started. + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 25))); + expect(counters.probes).toBe(1); + + yield* Deferred.succeed(gate, void 0); + const outs = (yield* Fiber.join(lists)) as readonly ListedConnections[]; + for (const out of outs) { + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.status).toBe("healthy"); + } + expect(counters.probes).toBe(1); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("healthy"); + }), + ); + + it.effect("a grant recorded invalid_grant-dead is never auto-revalidated", () => + Effect.gen(function* () { + const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); + yield* stamp({ + provider_state: { + oauthReauthRequiredAt: Date.now(), + oauthReauthRequiredDetail: "invalid_grant", + }, + last_health: { + status: "expired", + checkedAt: Date.now() - STALE_MS, + detail: "invalid_grant", + }, + }); + + // The list serves the persisted verdict without probing: a probe could + // pass on the access token's remaining lifetime and persist "healthy", + // hiding the required reconnect forever. + const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.status).toBe("expired"); + expect(counters.probes).toBe(0); + + // The manual "Check now" (no freshness window) refuses too: only an + // explicit reconnect clears a dead grant. + const manual = yield* executor.connections.checkHealth({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + expect(manual.status).toBe("expired"); + expect(counters.probes).toBe(0); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("expired"); + }), + ); + it.effect("leaves tool-sync failure verdicts for sync to clear", () => Effect.gen(function* () { const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); @@ -1277,6 +1370,48 @@ describe("heal-on-use", () => { }), ); + it.effect("does not overwrite a newer verdict written while the call was in flight", () => + Effect.gen(function* () { + const { executor, stamp, persisted, hooks } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + // A probe (or refresh) lands a NEWER expired verdict while the call is + // in flight. Heal-on-use decided from the row loaded BEFORE invocation; + // it must re-check at write time and leave the newer evidence standing. + const newerDetail = "revoked upstream while the call ran"; + hooks.onInvoke = stamp({ + last_health: { status: "expired", checkedAt: Date.now(), detail: newerDetail }, + }).pipe(Effect.asVoid); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "expired", detail: newerDetail }); + }), + ); + + it.effect("does not resurrect a grant that died while the call was in flight", () => + Effect.gen(function* () { + const { executor, stamp, persisted, hooks } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + }); + // A concurrent refresh discovers invalid_grant mid-call and records the + // dead grant. The invocation still succeeded on the old access token's + // remaining lifetime — healing from it would bury the reconnect. + hooks.onInvoke = stamp({ + provider_state: { oauthReauthRequiredAt: Date.now() }, + last_health: { status: "expired", checkedAt: Date.now(), detail: "invalid_grant" }, + }).pipe(Effect.asVoid); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "expired", detail: "invalid_grant" }); + }), + ); + it.effect("a call whose credential no longer resolves is not healed", () => Effect.gen(function* () { const { executor, stamp, persisted } = yield* makeHealthHarness(); diff --git a/packages/core/sdk/src/core-tools.ts b/packages/core/sdk/src/core-tools.ts index bac3066b5..0618ef86d 100644 --- a/packages/core/sdk/src/core-tools.ts +++ b/packages/core/sdk/src/core-tools.ts @@ -632,10 +632,14 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { owner: input.owner === undefined ? undefined : (input.owner as Owner), }); // Re-verify sticky non-healthy verdicts before reporting them - // (the same probe as the UI's "Check now", server-cached via - // `ifStaleMs` so repeated lists collapse to one probe per - // window). Quiet on probe failure: the persisted verdict is - // still the best known state, exactly like the UI surfaces. + // (the same probe as the UI's "Check now"). The server owns the + // stampede control: `ifStaleMs` caches settled verdicts per + // window, and concurrent readers past that gate coalesce onto + // one in-flight probe per connection. A grant the AS recorded + // dead is never re-probed there — only an explicit reconnect + // clears that verdict. Quiet on probe failure: the persisted + // verdict is still the best known state, exactly like the UI + // surfaces. const revalidated = yield* Effect.forEach( connections, (connection) => diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 9449d1e6d..8be454a6a 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -900,6 +900,37 @@ const decodeOAuthReauthRequiredProviderState = Schema.decodeUnknownOption( const oauthReauthRequiredFromProviderState = (value: unknown) => Option.getOrNull(decodeOAuthReauthRequiredProviderState(decodeJsonColumn(value))); +type OAuthReauthRequiredState = NonNullable< + ReturnType +>; + +// In-flight health probes, shared per connection across every executor holding +// the same root db handle. Read-time revalidation makes `connections.list` a +// probe trigger, so N concurrent readers past the freshness gate must collapse +// to ONE upstream probe per connection, not N — the freshness window alone +// cannot do that (nothing is persisted until the first probe settles). Same +// keyed-Deferred shape as #1537's refresh gate; if both land, the two could +// share one WeakMap-keyed gate helper. Weakly keyed so the map dies with the +// handle and a host that opens and drops handles does not leak one gate per +// handle. Process-local on purpose: a peer isolate probing in parallel is +// wasteful, not harmful, and cross-instance coordination belongs to the +// database. +interface HealthProbeOutcome { + readonly source: "credential_only" | "probe"; + readonly result: HealthCheckResult; +} +type HealthProbeGate = Map>; + +const healthProbeGateByRootDb = new WeakMap(); + +const healthProbeGateFor = (rootDb: object): HealthProbeGate => { + const existing = healthProbeGateByRootDb.get(rootDb); + if (existing) return existing; + const created: HealthProbeGate = new Map(); + healthProbeGateByRootDb.set(rootDb, created); + return created; +}; + const rowToConnection = (row: ConnectionRow): Connection => { const owner = row.owner as Owner; const integration = IntegrationSlug.make(row.integration); @@ -1727,6 +1758,9 @@ export const createExecutor = { if (isToolResult(result) && !result.ok) return Effect.void; if (Object.values(values).some((value) => value == null)) return Effect.void; - const last = Option.getOrNull(decodeLastHealth(row.last_health)); - if (last === null || last.status === "healthy" || last.status === "unknown") { + const observed = Option.getOrNull(decodeLastHealth(row.last_health)); + if (observed === null || observed.status === "healthy" || observed.status === "unknown") { return Effect.void; } - if (isToolSyncHealth(last)) return Effect.void; + if (isToolSyncHealth(observed)) return Effect.void; if (oauthReauthRequiredFromProviderState(row.provider_state) !== null) return Effect.void; const ref: ConnectionRef = { owner: row.owner as Owner, integration: IntegrationSlug.make(row.integration), name: ConnectionName.make(row.name), }; - return persistHealthResult(ref, { - status: "healthy", - checkedAt: Date.now(), - detail: "Tool invocation succeeded.", - }); + return findConnectionRow(ref).pipe( + Effect.flatMap((fresh) => { + if (fresh === null) return Effect.void; + if (oauthReauthRequiredFromProviderState(fresh.provider_state) !== null) { + return Effect.void; + } + const current = Option.getOrNull(decodeLastHealth(fresh.last_health)); + if ( + current === null || + current.status !== observed.status || + current.checkedAt !== observed.checkedAt + ) { + return Effect.void; + } + return persistHealthResult(ref, { + status: "healthy", + checkedAt: Date.now(), + detail: "Tool invocation succeeded.", + }); + }), + Effect.ignore, + ); }); const healthFromCredentialResolutionFailure = ( @@ -3857,7 +3916,7 @@ export const createExecutor = => Effect.annotateCurrentSpan({ @@ -3868,6 +3927,43 @@ export const createExecutor = { + const cached = Option.getOrNull(decodeLastHealth(row.last_health)); + if (cached !== null && cached.status === "expired") return cached; + return { + status: "expired", + checkedAt: reauthState.oauthReauthRequiredAt, + detail: + reauthState.oauthReauthRequiredDetail ?? + "The authorization server rejected this connection's refresh token (invalid_grant). Reconnect to continue.", + }; + }; + + /** Persist a probe verdict unless the grant died while the probe was in + * flight: a concurrent refresh discovering invalid_grant writes the + * authoritative dead-grant state (with its own `expired` verdict), and a + * probe that passed on the old access token's remaining lifetime must + * not bury it. Best-effort, like every verdict write. */ + const persistProbeHealthResult = ( + ref: ConnectionRef, + result: HealthCheckResult, + ): Effect.Effect => + findConnectionRow(ref).pipe( + Effect.flatMap((fresh) => + fresh !== null && oauthReauthRequiredFromProviderState(fresh.provider_state) !== null + ? Effect.void + : persistHealthResult(ref, result), + ), + Effect.ignore, + ); + const connectionCheckHealth = ( ref: ConnectionRef, options?: { @@ -3890,6 +3986,20 @@ export const createExecutor = { + const key = `${tenant}:${connectionKey(connectionRow)}`; + const existing = healthProbeInFlight.get(key); + if (existing) return Deferred.await(existing); + const deferred = Deferred.makeUnsafe(); + // Nothing suspends between the lookup above and this registration, + // so check-and-set is atomic against peer fibers. + healthProbeInFlight.set(key, deferred); + const freshVerdict: Effect.Effect = + spec === undefined && connectionRow.oauth_client != null + ? // No probe operation is declared, so "healthy" here means only + // "the credential resolved (refreshing if due)" — a refresh + // failure is the one real signal this path can produce, and it + // must not hide inside a green span. + oauthCredentialHealthWithoutProbe(connectionRow).pipe( + Effect.tap((result) => persistProbeHealthResult(ref, result)), + Effect.map((result) => ({ source: "credential_only" as const, result })), + ) + : foldCredentialResolutionIntoVerdict( + Effect.gen(function* () { + const values = yield* resolveConnectionValues(connectionRow); + const record = rowToIntegrationRecord( + integrationRow, + describeAuthMethodsForRow(integrationRow), + ); + const grantedScopes = grantedScopesFromRow(connectionRow); + const credential: ToolInvocationCredential = { + owner: connectionRow.owner as Owner, + integration: ref.integration, + connection: ConnectionName.make(connectionRow.name), + template: AuthTemplateSlug.make(connectionRow.template), + value: values[PRIMARY_INPUT_VARIABLE] ?? null, + values, + config: record.config, + ...(grantedScopes ? { grantedScopes } : {}), + }; + // Core resolves the declared spec (its own column) and + // hands it to the plugin; plugins no longer read it out of + // their config. + return yield* foldPluginFailure( + check({ ctx: runtime.ctx, integration: record, credential, spec }), + `Health check for connection "${ref.name}" failed.`, + ); + }), + ).pipe( + // Persist the verdict on the connection row so the accounts + // list shows alive/expired at a glance, AND so the freshness + // gate above has something to serve. A probe that could not + // resolve its credential persists too: it is the connection + // most likely to be re-probed by every surface on every + // mount, so leaving it unwritten is what turns one broken + // connection into unbounded upstream and error traffic. + Effect.tap((result) => persistProbeHealthResult(ref, result)), + Effect.map((result) => ({ source: "probe" as const, result })), + ); + const run = freshVerdict.pipe( + Effect.exit, + Effect.flatMap((exit) => Deferred.done(deferred, exit)), + Effect.ensuring(Effect.sync(() => void healthProbeInFlight.delete(key))), + ); + return Effect.forkDetach(run).pipe(Effect.andThen(Deferred.await(deferred))); + }); + yield* annotateHealthVerdict(outcome.source, outcome.result); + return outcome.result; }).pipe( Effect.withSpan("executor.connection.health.check", { attributes: { From 3ffc26ebece99277082f6ee4ee1816e81ab4ec36 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:51:44 -0700 Subject: [PATCH 4/8] Key the health probe gate structurally to prevent cross-tenant collisions --- packages/core/sdk/src/connections.test.ts | 84 +++++++++++++++++++++++ packages/core/sdk/src/executor.ts | 15 +++- 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 739cd4750..ef582358e 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -17,6 +17,8 @@ import { IntegrationSlug, ProviderItemId, ProviderKey, + Subject, + Tenant, ToolAddress, ToolName, } from "./ids"; @@ -1431,3 +1433,85 @@ describe("heal-on-use", () => { }), ); }); + +describe("health probe gate key integrity", () => { + // The in-flight probe gate is shared across every executor holding the same + // root db handle, so the key must be collision-free across tenants. A + // colon-join is not: tenant and subject are opaque strings that may contain + // colons, so (tenant "a", subject "user:b") and (tenant "a:user", subject + // "b") both read "a:user:user:b::" — and colliding keys + // share one Deferred, serving one tenant's probe outcome (run with ITS + // credentials) as the other tenant's health verdict. + it.effect("colliding colon-join identities run two distinct probes, not one shared gate", () => + Effect.gen(function* () { + const counters = { probes: 0 }; + const gate = yield* Deferred.make(); + // Every probe increments the shared counter and then parks on the gate, + // so both checks are provably in flight at once: nothing is persisted, + // and a collided gate would let the second check join the first probe's + // Deferred instead of starting its own. + const probingPlugin = definePlugin(() => ({ + id: "healthgate" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }), + invokeTool: ({ toolRow, credential }) => + Effect.succeed({ ran: toolRow.name, value: credential.value }), + checkHealth: () => + Effect.suspend(() => { + counters.probes += 1; + return Deferred.await(gate).pipe( + Effect.map(() => ({ + status: "healthy" as const, + checkedAt: Date.now(), + detail: "probe ok", + })), + ); + }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), + }), + })); + + // Both executors share ONE root db handle — and therefore one gate map; + // only the key separates their probes. + const configA = makeTestConfig({ + plugins: [probingPlugin()] as const, + tenant: "a", + subject: "user:b", + }); + const executorA = yield* createExecutor(configA); + const executorB = yield* createExecutor({ + ...configA, + tenant: Tenant.make("a:user"), + subject: Subject.make("b"), + plugins: [probingPlugin()] as const, + }); + yield* executorA.healthgate.seed(); + yield* executorB.healthgate.seed(); + const ref = { + owner: "user", + name: ConnectionName.make("main"), + integration: INTEG, + } as const; + yield* executorA.connections.create({ ...ref, template: TEMPLATE, value: "token-a" }); + yield* executorB.connections.create({ ...ref, template: TEMPLATE, value: "token-b" }); + + const checkA = yield* Effect.forkChild(executorA.connections.checkHealth(ref)); + const checkB = yield* Effect.forkChild(executorB.connections.checkHealth(ref)); + // Give both fibers real time to reach the probe path while the gate + // holds every probe open; the counter then says how many started. + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 25))); + expect(counters.probes).toBe(2); + + yield* Deferred.succeed(gate, void 0); + const resultA = yield* Fiber.join(checkA); + const resultB = yield* Fiber.join(checkB); + expect(resultA.status).toBe("healthy"); + expect(resultB.status).toBe("healthy"); + expect(counters.probes).toBe(2); + }), + ); +}); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 8be454a6a..7c2460236 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -931,6 +931,16 @@ const healthProbeGateFor = (rootDb: object): HealthProbeGate => { return created; }; +/** Gate key for one connection's in-flight probe. Structured (a JSON array), + * never delimiter-joined: `tenant` and `subject` are opaque strings that may + * themselves contain any delimiter, so a colon-join lets distinct identities + * collide — tenant "a" + subject "user:b" reads exactly like tenant "a:user" + * + subject "b" — and colliding identities would share one Deferred, serving + * one tenant's probe outcome (run with ITS credentials) as another tenant's + * health verdict. Same structured-key idiom as #1537's refresh gate. */ +const healthProbeGateKey = (tenant: string, row: ConnectionRow): string => + JSON.stringify([tenant, row.owner, row.subject, row.integration, row.name]); + const rowToConnection = (row: ConnectionRow): Connection => { const owner = row.owner as Owner; const integration = IntegrationSlug.make(row.integration); @@ -1759,7 +1769,8 @@ export const createExecutor = { - const key = `${tenant}:${connectionKey(connectionRow)}`; + const key = healthProbeGateKey(tenant, connectionRow); const existing = healthProbeInFlight.get(key); if (existing) return Deferred.await(existing); const deferred = Deferred.makeUnsafe(); From 767af267f620fee6020e5c5536533a623796c835 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:20:04 -0700 Subject: [PATCH 5/8] Compare-and-swap verdict writes on updated_at so newer state wins --- packages/core/sdk/src/connections.test.ts | 242 +++++++++++++++++++++- packages/core/sdk/src/executor.ts | 40 +++- 2 files changed, 267 insertions(+), 15 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index ef582358e..d55e99a7e 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "@effect/vitest"; import { Deferred, Effect, + Exit, Fiber, Inspectable, Logger, @@ -1108,16 +1109,74 @@ const makeHealthHarness = (options?: { /** Replaces the default instant-healthy probe. Entries are still counted in * `counters.probes`, so a Deferred-gated probe lets a test hold every * in-flight health check open and count how many actually started. */ - readonly probe?: Effect.Effect; + readonly probe?: Effect.Effect; }) => { - const counters = { probes: 0 }; - // Runs inside every invocation before it returns, so a test can interleave - // a concurrent write (e.g. a refresh discovering invalid_grant) between the - // row load and the heal-on-use decision. - const hooks = { onInvoke: Effect.void as Effect.Effect }; + const counters = { probes: 0, resolves: 0 }; + const hooks = { + // Runs inside every invocation before it returns, so a test can interleave + // a concurrent write (e.g. a refresh discovering invalid_grant) between the + // row load and the heal-on-use decision. + onInvoke: Effect.void as Effect.Effect, + // Runs inside every credential-provider read (counted in + // `counters.resolves`), so a Deferred here holds the credential-only + // health path open the way a Deferred probe holds the probing path open. + onResolve: Effect.void as Effect.Effect, + // One-shot: runs immediately before a connection UPDATE that writes + // `last_health` reaches the database — INSIDE a verdict guard's + // check-to-write window, after its fresh read has already been taken. + // Cleared before it runs, so the conflicting write it performs (through + // the unwrapped `config.db`) is not intercepted again. + beforeHealthPersist: null as Effect.Effect | null, + }; + // Wraps the executor's FumaDb handle so `beforeHealthPersist` can commit a + // conflicting write in the exact window the write guards must close. + // `withContext` re-wraps because `createExecutor` rebinds the handle to its + // owner context; without that the interception would be dropped. + const interceptHealthWrites = (db: FumaDb): FumaDb => + new Proxy(db as object, { + get(target, prop, receiver) { + const value = Reflect.get(target, prop, receiver); + if (prop === "withContext") { + return (context: unknown) => + interceptHealthWrites((value as (context: unknown) => FumaDb).call(target, context)); + } + if (prop === "updateMany") { + return async ( + table: string, + updateOptions: { readonly set?: Record }, + ) => { + const hook = hooks.beforeHealthPersist; + if ( + hook !== null && + table === "connection" && + updateOptions.set !== undefined && + "last_health" in updateOptions.set + ) { + hooks.beforeHealthPersist = null; + await Effect.runPromise(hook); + } + return (value as (...args: unknown[]) => Promise).call( + target, + table, + updateOptions, + ); + }; + } + return value; + }, + }) as FumaDb; + const baseProvider = memoryProvider(); + const countingProvider: CredentialProvider = { + ...baseProvider, + get: (id) => + Effect.suspend(() => { + counters.resolves += 1; + return hooks.onResolve.pipe(Effect.andThen(baseProvider.get(id))); + }), + }; const plugin = definePlugin(() => ({ id: "healthdemo" as const, - credentialProviders: [memoryProvider()], + credentialProviders: [countingProvider], storage: () => ({}), resolveTools: () => Effect.succeed({ tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }), @@ -1147,7 +1206,7 @@ const makeHealthHarness = (options?: { plugins: [plugin] as const, coreTools: { webBaseUrl: "http://localhost:3000" }, }); - const executor = yield* createExecutor(config); + const executor = yield* createExecutor({ ...config, db: interceptHealthWrites(config.db) }); yield* executor.healthdemo.seed(); yield* executor.connections.create({ owner: "org", @@ -1434,6 +1493,173 @@ describe("heal-on-use", () => { ); }); +// --------------------------------------------------------------------------- +// The guards above re-take their decision from a fresh row at write time, but +// a re-read alone leaves a window: a conflicting write can commit AFTER the +// fresh read and BEFORE the guard's own UPDATE. These tests commit the +// conflict inside that exact window (`hooks.beforeHealthPersist` fires after +// the guard's fresh read, immediately before its UPDATE reaches the +// database) and assert the guarded write loses: the UPDATE is +// compare-and-swapped on the `updated_at` stamp the fresh read observed, so +// the conflict's bump makes it match zero rows. Every initial stamp ages +// `updated_at` because SQLite stores the stamp at second granularity — the +// conflict's fresh stamp must land in a different granule than the observed +// one for the swap to see it. +// --------------------------------------------------------------------------- + +describe("verdict write guards close the check-to-write window", () => { + const REF = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; + + it.effect("probe persist: a dead grant recorded inside the window survives", () => + Effect.gen(function* () { + const { executor, counters, stamp, persisted, hooks } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + updated_at: new Date(Date.now() - STALE_MS), + }); + // The probe passes on the old access token's remaining lifetime while a + // concurrent refresh discovers invalid_grant. The dead-grant write + // commits after the guard's fresh read (which saw no dead grant) and + // before its UPDATE — the window a re-read alone cannot close. + hooks.beforeHealthPersist = stamp({ + provider_state: { + oauthReauthRequiredAt: Date.now(), + oauthReauthRequiredDetail: "invalid_grant", + }, + last_health: { status: "expired", checkedAt: Date.now(), detail: "invalid_grant" }, + updated_at: new Date(), + }).pipe(Effect.asVoid); + + const result = yield* executor.connections.checkHealth(REF); + expect(result.status).toBe("healthy"); + expect(counters.probes).toBe(1); + + // The dead-grant verdict survived the probe's guarded write... + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "expired", detail: "invalid_grant" }); + + // ...and the next check serves it without probing, as for any dead grant. + const after = yield* executor.connections.checkHealth(REF); + expect(after.status).toBe("expired"); + expect(counters.probes).toBe(1); + }), + ); + + it.effect("heal-on-use: a dead grant recorded inside the window survives", () => + Effect.gen(function* () { + const { executor, stamp, persisted, hooks } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + updated_at: new Date(Date.now() - STALE_MS), + }); + // Heal-on-use re-reads, sees the stale verdict it observed at load and + // no dead grant, and decides to heal — then the dead-grant write + // commits before its UPDATE. + hooks.beforeHealthPersist = stamp({ + provider_state: { oauthReauthRequiredAt: Date.now() }, + last_health: { status: "expired", checkedAt: Date.now(), detail: "invalid_grant" }, + updated_at: new Date(), + }).pipe(Effect.asVoid); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "expired", detail: "invalid_grant" }); + }), + ); + + it.effect("heal-on-use: a newer verdict written inside the window survives", () => + Effect.gen(function* () { + const { executor, stamp, persisted, hooks } = yield* makeHealthHarness(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + updated_at: new Date(Date.now() - STALE_MS), + }); + const newerDetail = "revoked upstream while the heal was in flight"; + hooks.beforeHealthPersist = stamp({ + last_health: { status: "expired", checkedAt: Date.now(), detail: newerDetail }, + updated_at: new Date(), + }).pipe(Effect.asVoid); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "expired", detail: newerDetail }); + }), + ); +}); + +describe("credential-only health path", () => { + it.effect("concurrent checks collapse to one credential resolution", () => + Effect.gen(function* () { + const gate = yield* Deferred.make(); + const { executor, counters, stamp, persisted, hooks } = yield* makeHealthHarness(); + // No declared probe spec + an OAuth client on the row routes checkHealth + // down the credential-only path: the verdict is "the credential + // resolved", produced without invoking the plugin probe. That path runs + // behind the same in-flight gate as probing, so concurrent checks must + // collapse to ONE resolution. + yield* stamp({ oauth_client: "acme", expires_at: null }); + hooks.onResolve = Deferred.await(gate); + counters.resolves = 0; + const ref = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; + + const checks = yield* Effect.forkChild( + Effect.all([executor.connections.checkHealth(ref), executor.connections.checkHealth(ref)], { + concurrency: "unbounded", + }), + ); + // The resolution is held open by the gate, so nothing has been + // persisted: both checks must reach the credential path. Give them real + // time to get there — the counter then says how many resolutions started. + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 25))); + expect(counters.resolves).toBe(1); + + yield* Deferred.succeed(gate, void 0); + const [first, second] = yield* Fiber.join(checks); + expect(first.status).toBe("healthy"); + expect(second.status).toBe("healthy"); + expect(counters.resolves).toBe(1); + expect(counters.probes).toBe(0); + + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("healthy"); + }), + ); +}); + +describe("health probe gate lifecycle", () => { + it.effect("a failed probe clears its gate entry for the next check", () => + Effect.gen(function* () { + let firstProbe = true; + const { executor, counters } = yield* makeHealthHarness({ + probe: Effect.suspend(() => { + if (firstProbe) { + firstProbe = false; + return Effect.fail("probe exploded" as const); + } + return Effect.succeed({ + status: "healthy" as const, + checkedAt: Date.now(), + detail: "probe ok", + }); + }), + }); + const ref = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; + + const first = yield* executor.connections.checkHealth(ref).pipe(Effect.exit); + expect(Exit.isFailure(first)).toBe(true); + + // A leaked gate entry would hand this check the first probe's settled + // Deferred (failing again without probing); a second probe run proves + // the failure removed the entry. + const second = yield* executor.connections.checkHealth(ref); + expect(second.status).toBe("healthy"); + expect(counters.probes).toBe(2); + }), + ); +}); + describe("health probe gate key integrity", () => { // The in-flight probe gate is shared across every executor holding the same // root db handle, so the key must be collision-free across tenants. A diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 7c2460236..12edb5e28 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3762,8 +3762,27 @@ export const createExecutor = ({ status: "unknown", checkedAt: Date.now() }); + /** Persist a verdict with a compare-and-swap on `updated_at`: the single + * UPDATE commits only while the row still carries the stamp the caller's + * fresh read observed, so a write landing between that read and this one + * (a refresh recording invalid_grant, a newer verdict) makes the WHERE + * match zero rows — the newer state wins and the loser is a silent no-op. + * + * `updated_at` is the version token because every write that touches + * `last_health` or `provider_state` bumps it inside the same statement + * (grep `updateMany("connection"`; keep it that way), while the json + * columns themselves can never appear in a WHERE clause — Postgres maps + * them to `json`, which has no comparison operators, so a value-guarded + * UPDATE would raise at runtime and, behind `Effect.ignore`, silently + * disable verdict persistence. The stamp is millisecond-grained on + * Postgres and second-grained on SQLite; a conflicting write inside the + * same granule as the row's previous stamp can still slip the swap, but + * even then a buried dead grant stays authoritative at read time: + * `deadGrantVerdict` answers from `provider_state`, which no verdict + * write touches. Best-effort, like every verdict write. */ const persistHealthResult = ( ref: ConnectionRef, + observedUpdatedAt: Date, result: HealthCheckResult, ): Effect.Effect => core @@ -3773,6 +3792,7 @@ export const createExecutor = => findConnectionRow(ref).pipe( Effect.flatMap((fresh) => - fresh !== null && oauthReauthRequiredFromProviderState(fresh.provider_state) !== null + fresh === null || oauthReauthRequiredFromProviderState(fresh.provider_state) !== null ? Effect.void - : persistHealthResult(ref, result), + : persistHealthResult(ref, fresh.updated_at, result), ), Effect.ignore, ); From c6a29575d31ee8648e274d439bbfdf25b6198ae0 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:39:34 -0700 Subject: [PATCH 6/8] Join tools_synced_at into the verdict CAS so same-second sync failures survive --- packages/core/sdk/src/connections.test.ts | 103 +++++++++++++++++++++- packages/core/sdk/src/executor.ts | 45 ++++++---- 2 files changed, 132 insertions(+), 16 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index d55e99a7e..4459fb3ed 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1228,7 +1228,15 @@ const makeHealthHarness = (options?: { integration: INTEG, name: ConnectionName.make("main"), }); - return { executor, counters, stamp, persisted, hooks } as const; + // The public connection shape omits `tools_synced_at`; read the raw row + // for tests asserting a conflicting sync's stamp survived a guard. + const rawRow = () => + Effect.promise(() => + config.db.findFirst("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + }), + ); + return { executor, counters, stamp, persisted, rawRow, hooks } as const; }); }; @@ -1587,6 +1595,99 @@ describe("verdict write guards close the check-to-write window", () => { expect(row?.lastHealth).toMatchObject({ status: "expired", detail: newerDetail }); }), ); + + // The three tests above age `updated_at` so the conflict's bump lands in a + // different SQLite second granule. The three below do the opposite: the + // conflict reuses the EXACT stamp the guard's fresh read observed — the + // same-second collision `updated_at` alone cannot see — so only the + // `tools_synced_at` leg of the swap can refuse the guarded write. The + // conflict is the one collision that must never be buried: a failing tool + // sync stores its degraded verdict TOGETHER with a fresh `tools_synced_at`, + // so a guard overwriting it with "healthy" would also leave the catalog + // looking just-synced and hide the failure for the full sync TTL. + + const SYNC_DETAIL = "Tool sync failing: plugin returned an incomplete tool catalog"; + + it.effect("probe persist: a failing sync landing in the same second survives", () => + Effect.gen(function* () { + const { executor, counters, stamp, persisted, rawRow, hooks } = yield* makeHealthHarness(); + const observedUpdatedAt = new Date(); + const observedSyncedAt = Date.now() - STALE_MS; + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + updated_at: observedUpdatedAt, + tools_synced_at: observedSyncedAt, + }); + const freshSyncedAt = Date.now(); + hooks.beforeHealthPersist = stamp({ + tools_synced_at: freshSyncedAt, + last_health: { status: "degraded", checkedAt: Date.now(), detail: SYNC_DETAIL }, + updated_at: observedUpdatedAt, + }).pipe(Effect.asVoid); + + const result = yield* executor.connections.checkHealth(REF); + expect(result.status).toBe("healthy"); + expect(counters.probes).toBe(1); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "degraded", detail: SYNC_DETAIL }); + const raw = yield* rawRow(); + expect(Number(raw?.tools_synced_at)).toBe(freshSyncedAt); + }), + ); + + it.effect("heal-on-use: a failing sync landing in the same second survives", () => + Effect.gen(function* () { + const { executor, stamp, persisted, rawRow, hooks } = yield* makeHealthHarness(); + const observedUpdatedAt = new Date(); + const observedSyncedAt = Date.now() - STALE_MS; + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + updated_at: observedUpdatedAt, + tools_synced_at: observedSyncedAt, + }); + // `+ 1` rather than `Date.now()`: the invocation itself may re-stamp + // `tools_synced_at` before the heal's fresh read, and the conflicting + // sync stamp must be guaranteed to differ from whatever that read + // observed — an aged value + 1 can match neither it nor "now". + const freshSyncedAt = observedSyncedAt + 1; + hooks.beforeHealthPersist = stamp({ + tools_synced_at: freshSyncedAt, + last_health: { status: "degraded", checkedAt: Date.now(), detail: SYNC_DETAIL }, + updated_at: observedUpdatedAt, + }).pipe(Effect.asVoid); + + yield* executor.execute(ToolAddress.make("tools.vercel.org.main.deploy"), {}); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "degraded", detail: SYNC_DETAIL }); + const raw = yield* rawRow(); + expect(Number(raw?.tools_synced_at)).toBe(freshSyncedAt); + }), + ); + + it.effect("probe persist: the sync leg guards a never-synced row (NULL stamp)", () => + Effect.gen(function* () { + const { executor, stamp, persisted, hooks } = yield* makeHealthHarness(); + const observedUpdatedAt = new Date(); + yield* stamp({ + last_health: { status: "expired", checkedAt: Date.now() - STALE_MS, detail: "HTTP 401" }, + updated_at: observedUpdatedAt, + tools_synced_at: null, + }); + hooks.beforeHealthPersist = stamp({ + tools_synced_at: Date.now(), + last_health: { status: "degraded", checkedAt: Date.now(), detail: SYNC_DETAIL }, + updated_at: observedUpdatedAt, + }).pipe(Effect.asVoid); + + const result = yield* executor.connections.checkHealth(REF); + expect(result.status).toBe("healthy"); + + const row = yield* persisted(); + expect(row?.lastHealth).toMatchObject({ status: "degraded", detail: SYNC_DETAIL }); + }), + ); }); describe("credential-only health path", () => { diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 12edb5e28..1d189e131 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3775,14 +3775,25 @@ export const createExecutor = , result: HealthCheckResult, ): Effect.Effect => core @@ -3792,7 +3803,10 @@ export const createExecutor = fresh === null || oauthReauthRequiredFromProviderState(fresh.provider_state) !== null ? Effect.void - : persistHealthResult(ref, fresh.updated_at, result), + : persistHealthResult(ref, fresh, result), ), Effect.ignore, ); From 4f2c97d0773e635a4b68185cb38539b7f850f056 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 13:17:15 -0700 Subject: [PATCH 7/8] Keep the dead-grant expired verdict authoritative over sync failures --- packages/core/sdk/src/connections.test.ts | 121 ++++++++++++++++++++++ packages/core/sdk/src/executor.ts | 51 +++++++-- 2 files changed, 161 insertions(+), 11 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 4459fb3ed..b1e7fcf55 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -793,6 +793,93 @@ describe("tool catalog sync safety", () => { ), ); + it.effect("a failing sync does not bury a dead grant's expired verdict", () => + 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: "upstream rejected the credential", + } + : { + 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: {}, + }), + }), + }))(); + 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", + }); + + // The refresh recorder's authoritative write: the sync's own + // credential resolution discovered invalid_grant, so by the time the + // sync fails, the row records the dead grant WITH its expired verdict. + const expired = { + status: "expired" as const, + checkedAt: Date.now(), + detail: "invalid_grant: Grant not found", + }; + 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, + last_health: expired, + provider_state: { + oauthReauthRequiredAt: Date.now(), + oauthReauthRequiredDetail: "invalid_grant: Grant not found", + }, + }, + }), + ); + yield* executor.tools.list({ integration: INTEG }); + + // "expired, reconnect" outranks "tool sync failing": nothing would + // re-assert the dead grant's verdict (it is never probed), while the + // reconnect that clears it re-syncs tools anyway. + const connection = yield* executor.connections.get({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + expect(connection?.lastHealth).toMatchObject(expired); + + // The sync time still stamps, so the failing catalog is not + // re-attempted on every read. + const row = yield* Effect.promise(() => + config.db.findFirst("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + }), + ); + expect(row?.tools_synced_at).not.toBeNull(); + }), + ), + ); + it.effect("successful sync preserves genuine health-check records", () => Effect.scoped( Effect.gen(function* () { @@ -1366,6 +1453,40 @@ describe("agent read revalidation (coreTools connections.list)", () => { }), ); + it.effect("check now repairs a dead-grant verdict buried by a racing writer", () => + Effect.gen(function* () { + const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); + const detail = "invalid_grant: Grant not found"; + // A racing writer's verdict landed after the recorder's: the row reads + // "degraded" while provider_state still records the dead grant. + yield* stamp({ + provider_state: { oauthReauthRequiredAt: Date.now(), oauthReauthRequiredDetail: detail }, + last_health: { + status: "degraded", + checkedAt: Date.now(), + detail: "Tool sync failing: upstream rejected the credential", + }, + }); + + // Still no probe — the dead grant refuses those — but the served + // verdict is the authoritative expired one, not the buried degraded. + const manual = yield* executor.connections.checkHealth({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + expect(manual.status).toBe("expired"); + expect(manual.detail).toBe(detail); + expect(counters.probes).toBe(0); + + // And it re-persisted, so plain row reads agree with what every health + // read serves. + const row = yield* persisted(); + expect(row?.lastHealth?.status).toBe("expired"); + expect(row?.lastHealth?.detail).toBe(detail); + }), + ); + it.effect("leaves tool-sync failure verdicts for sync to clear", () => Effect.gen(function* () { const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 1d189e131..d9e790354 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3091,15 +3091,33 @@ export const createExecutor = - core.updateMany("connection", { - where: connectionWhere, - set: { - tools_synced_at: Date.now(), - last_health: toolSyncHealth(reason), - updated_at: new Date(), - }, - }); + findConnectionRow(ref).pipe( + Effect.flatMap((fresh) => + core.updateMany("connection", { + where: connectionWhere, + set: + fresh !== null && + oauthReauthRequiredFromProviderState(fresh.provider_state) !== null + ? { tools_synced_at: Date.now() } + : { + 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 @@ -4043,12 +4061,23 @@ export const createExecutor = Date: Fri, 28 Aug 2026 13:36:13 -0700 Subject: [PATCH 8/8] Derive dead-grant expired verdicts at read time instead of repairing rows --- packages/core/sdk/src/connections.test.ts | 65 ++++++++++++++++--- packages/core/sdk/src/executor.ts | 79 +++++++++++++---------- 2 files changed, 103 insertions(+), 41 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index b1e7fcf55..2caf2195d 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1453,9 +1453,9 @@ describe("agent read revalidation (coreTools connections.list)", () => { }), ); - it.effect("check now repairs a dead-grant verdict buried by a racing writer", () => + it.effect("a buried dead-grant verdict presents expired on every read, without a write", () => Effect.gen(function* () { - const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); + const { executor, counters, stamp, persisted, rawRow } = yield* makeHealthHarness(); const detail = "invalid_grant: Grant not found"; // A racing writer's verdict landed after the recorder's: the row reads // "degraded" while provider_state still records the dead grant. @@ -1467,9 +1467,11 @@ describe("agent read revalidation (coreTools connections.list)", () => { detail: "Tool sync failing: upstream rejected the credential", }, }); + const before = yield* rawRow(); - // Still no probe — the dead grant refuses those — but the served - // verdict is the authoritative expired one, not the buried degraded. + // Check now: still no probe — the dead grant refuses those — and the + // served verdict is the authoritative expired one, not the buried + // degraded. const manual = yield* executor.connections.checkHealth({ owner: "org", integration: INTEG, @@ -1479,11 +1481,58 @@ describe("agent read revalidation (coreTools connections.list)", () => { expect(manual.detail).toBe(detail); expect(counters.probes).toBe(0); - // And it re-persisted, so plain row reads agree with what every health - // read serves. + // connections.get and the agent list present the same derivation. const row = yield* persisted(); - expect(row?.lastHealth?.status).toBe("expired"); - expect(row?.lastHealth?.detail).toBe(detail); + expect(row?.lastHealth).toMatchObject({ status: "expired", detail }); + const out = (yield* executor.execute(CORE_LIST, {})) as ListedConnections; + const listed = out.connections.find((c) => c.name === "main"); + expect(listed?.lastHealth?.status).toBe("expired"); + expect(counters.probes).toBe(0); + + // Derivation, not repair: no read wrote anything back. A repair write + // could race a concurrent reconnect and stamp the old grant's expired + // verdict onto the fresh connection; presenting from `provider_state` + // needs no write, so there is nothing to race. + const after = yield* rawRow(); + expect(after?.updated_at).toEqual(before?.updated_at); + expect(after?.last_health).toEqual(before?.last_health); + }), + ); + + it.effect("reconnect clearing the dead grant ends the derivation on reads", () => + Effect.gen(function* () { + const { executor, counters, stamp, persisted } = yield* makeHealthHarness(); + yield* stamp({ + provider_state: { + oauthReauthRequiredAt: Date.now(), + oauthReauthRequiredDetail: "invalid_grant", + }, + last_health: { + status: "degraded", + checkedAt: Date.now(), + detail: "Tool sync failing: upstream rejected the credential", + }, + }); + const buried = yield* persisted(); + expect(buried?.lastHealth?.status).toBe("expired"); + + // The reconnect mint rewrites `provider_state` wholesale and clears the + // old grant's verdict. That alone must end the expired presentation — + // no repair write exists to resurrect the old grant's verdict onto the + // fresh connection. + yield* stamp({ provider_state: null, last_health: null, updated_at: new Date() }); + + const fresh = yield* persisted(); + expect(fresh?.lastHealth).toBeNull(); + + // And probing is re-opened for the new grant. + const check = yield* executor.connections.checkHealth({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + expect(check.status).toBe("healthy"); + expect(counters.probes).toBe(1); }), ); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index d9e790354..cc6d798a8 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -941,6 +941,42 @@ const healthProbeGateFor = (rootDb: object): HealthProbeGate => { const healthProbeGateKey = (tenant: string, row: ConnectionRow): string => JSON.stringify([tenant, row.owner, row.subject, row.integration, row.name]); +/** The verdict a recorded dead grant answers every health read with. The + * persisted `expired` verdict (written together with the dead-grant + * state) is served as-is; a row whose verdict a racing writer buried (or + * that somehow lacks one) gets an expired verdict synthesized from the + * recorded rejection, unpersisted. */ +const deadGrantVerdict = ( + reauthState: OAuthReauthRequiredState, + row: ConnectionRow, +): HealthCheckResult => { + const cached = Option.getOrNull(decodeLastHealth(row.last_health)); + if (cached !== null && cached.status === "expired") return cached; + return { + status: "expired", + checkedAt: reauthState.oauthReauthRequiredAt, + detail: + reauthState.oauthReauthRequiredDetail ?? + "The authorization server rejected this connection's refresh token (invalid_grant). Reconnect to continue.", + }; +}; + +/** The health a connection row presents on every API read. Derived, never + * written back: while `provider_state` records a dead grant, the row + * presents the dead grant's expired verdict regardless of what a racing + * writer left in `last_health`, so a buried verdict cannot mislead any + * reader. A repair WRITE here instead would race the reconnect mint — a + * stale repair that observed the pre-reconnect dead grant can pass the + * verdict CAS inside one SQLite `updated_at` second and stamp the OLD + * grant's expired verdict onto the fresh connection. The reconnect mint + * rewrites `provider_state` wholesale, which ends this derivation with no + * write to race. */ +const presentedLastHealth = (row: ConnectionRow): HealthCheckResult | null => { + const reauthState = oauthReauthRequiredFromProviderState(row.provider_state); + if (reauthState !== null) return deadGrantVerdict(reauthState, row); + return Option.getOrNull(decodeLastHealth(row.last_health)); +}; + const rowToConnection = (row: ConnectionRow): Connection => { const owner = row.owner as Owner; const integration = IntegrationSlug.make(row.integration); @@ -960,7 +996,7 @@ const rowToConnection = (row: ConnectionRow): Connection => { row.oauth_client_owner == null ? null : (String(row.oauth_client_owner) as Owner), oauthScope: row.oauth_scope == null ? null : String(row.oauth_scope), missingOAuthScopes: missingOAuthScopesFromProviderState(row.provider_state), - lastHealth: Option.getOrNull(decodeLastHealth(row.last_health)), + lastHealth: presentedLastHealth(row), }; }; @@ -3993,25 +4029,6 @@ export const createExecutor = { - const cached = Option.getOrNull(decodeLastHealth(row.last_health)); - if (cached !== null && cached.status === "expired") return cached; - return { - status: "expired", - checkedAt: reauthState.oauthReauthRequiredAt, - detail: - reauthState.oauthReauthRequiredDetail ?? - "The authorization server rejected this connection's refresh token (invalid_grant). Reconnect to continue.", - }; - }; - /** Persist a probe verdict unless the grant died while the probe was in * flight: a concurrent refresh discovering invalid_grant writes the * authoritative dead-grant state (with its own `expired` verdict), and a @@ -4063,21 +4080,17 @@ export const createExecutor =