diff --git a/.changeset/oauth-refresh-evidence.md b/.changeset/oauth-refresh-evidence.md new file mode 100644 index 000000000..826bf9881 --- /dev/null +++ b/.changeset/oauth-refresh-evidence.md @@ -0,0 +1,18 @@ +--- +"@executor-js/sdk": patch +"@executor-js/plugin-mcp": patch +"@executor-js/plugin-openapi": patch +"@executor-js/plugin-graphql": patch +--- + +Stop recording a permanent **Expired** verdict without evidence, and stop the refresh races that produced one. + +A refresher whose grant is refused now reads the stored refresh token again; when a peer instance rotated it during the request, the call adopts the access token that peer persisted and records nothing. Previously the loser of a concurrent refresh wrote `oauthReauthRequiredAt` onto a connection that still held a valid rotated refresh token, and every surface then answered `expired` without probing — no tool call could refresh it again, only a re-authorization recovered it. The record write is also skipped when `expires_at` moved forward during the grant, which is the same peer success read from the row. + +`isPermanentTokenRejection` no longer reads 408, 425, or 429 as a definitive refusal, so one rate-limited minute at a token endpoint no longer ends a grant; those statuses behave like a 5xx and the next call retries. A refresh response that omits `expires_in` (RFC 6749 makes it optional) no longer erases `expires_at`: the mint records the advertised lifetime in `provider_state.oauthTokenLifetimeMs` and a refresh derives the expiry from it, instead of disabling proactive refresh for the rest of the connection's life. + +`connections.checkHealth` re-mints once and probes again before it answers `expired` for an OAuth connection, so a revoked token, an idle timeout shorter than the advertised lifetime, or a null `expires_at` no longer shows a working connection as dead until a tool call heals it. The plugin is asked first with or without a declared health-check spec, so a plugin whose probe needs no spec (MCP lists tools) gives its OAuth connections a real verdict; only a plugin that answers `unknown` falls back to the credential-only verdict, and that verdict now reports `expired` when a credential value resolves to nothing. + +The MCP liveness probe takes the invocation pool's lease instead of dialling a second connection, bounded by the shared 15s discovery deadline; a probe of a stdio server no longer starts a second child process, which single-instance servers (Chrome DevTools MCP, Playwright MCP, `docker run -i`) refused — reporting a live, serving connection as broken on every page mount. A 403 scope shortfall on a probe reads `degraded` instead of `expired`, from either an RFC 6750 `WWW-Authenticate` challenge or a body marker. The GraphQL probe no longer reads a transport failure's prose as a dead credential (`connect EACCES: permission denied` on a socket is not an authentication verdict). + +The test authorization server's `/mcp` resource endpoint now speaks JSON-RPC honestly — the request's own id, an empty catalog for `tools/list`, silence for notifications — where the old canned reply used a fixed id and left every completed handshake waiting forever for its `tools/list` response. diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 51c7f8d11..92fe43586 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -2575,6 +2575,11 @@ const makeHealthHarness = (options?: { * `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; + /** Answer `unknown` when core passes no declared spec, the way the protocol + * plugins do: they have no operation to dial, so core falls back to the + * credential-only verdict. Without this the harness plugin answers every + * check, and the fallback is unreachable. */ + readonly declineWithoutSpec?: boolean; }) => { const counters = { probes: 0, resolves: 0 }; const hooks = { @@ -2660,9 +2665,16 @@ const makeHealthHarness = (options?: { ? ToolResult.fail({ code: "upstream_error", message: "boom" }) : { ran: toolRow.name, value: credential.value }, ), - checkHealth: () => + checkHealth: ({ spec }) => Effect.suspend(() => { counters.probes += 1; + if (options?.declineWithoutSpec === true && spec === undefined) { + return Effect.succeed({ + status: "unknown" as const, + checkedAt: Date.now(), + detail: "No health check configured.", + }); + } return ( options?.probe ?? Effect.succeed({ status: "healthy" as const, checkedAt: Date.now(), detail: "probe ok" }) @@ -3298,12 +3310,15 @@ describe("credential-only health path", () => { // parallel suite load the forked checks may not have finished their row // loads yet, and the counter reads 0. const entered = 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. + const { executor, counters, stamp, persisted, hooks } = yield* makeHealthHarness({ + declineWithoutSpec: true, + }); + // No declared probe spec + an OAuth client on the row: the plugin is + // asked first, declines for want of an operation to dial, and core falls + // back to the credential-only verdict — "the credential resolved", + // produced from the SAME resolution the plugin's probe used, so nothing + // refreshes twice. 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.succeed(entered, void 0).pipe( Effect.andThen(Deferred.await(gate)), @@ -3329,7 +3344,10 @@ describe("credential-only health path", () => { expect(first.status).toBe("healthy"); expect(second.status).toBe("healthy"); expect(counters.resolves).toBe(1); - expect(counters.probes).toBe(0); + // Both checks shared ONE gate entry, so the plugin was asked once and + // declined once; the verdict both callers received is the fallback. + expect(counters.probes).toBe(1); + expect(first.detail).toBe("Credential resolved (no probe configured)."); const row = yield* persisted(); expect(row?.lastHealth?.status).toBe("healthy"); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index cd89ab907..4af36d4e0 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1013,6 +1013,30 @@ const decodeOAuthReauthRequiredProviderState = Schema.decodeUnknownOption( const oauthReauthRequiredFromProviderState = (value: unknown) => Option.getOrNull(decodeOAuthReauthRequiredProviderState(decodeJsonColumn(value))); +/** `provider_state` as a merge base: the object it holds, or an empty one. Every + * writer merges rather than replaces, so a concurrent record survives. */ +const providerStateRecord = (value: unknown): Record => { + const decoded = decodeJsonColumn(value); + return decoded != null && typeof decoded === "object" && !Array.isArray(decoded) + ? (decoded as Record) + : {}; +}; + +const decodeTokenLifetimeState = Schema.decodeUnknownOption( + Schema.Struct({ oauthTokenLifetimeMs: Schema.Number }), +); + +/** The access-token lifetime this grant advertised, in ms, when it ever + * advertised one. RFC 6749 makes `expires_in` optional: writing a null + * `expires_at` from a refresh that omits it erased the only input the + * proactive refresh has, permanently. */ +const rememberedTokenLifetimeMs = (value: unknown): number | null => { + const decoded = Option.getOrNull(decodeTokenLifetimeState(decodeJsonColumn(value))); + return decoded === null || !Number.isFinite(decoded.oauthTokenLifetimeMs) + ? null + : decoded.oauthTokenLifetimeMs; +}; + type OAuthReauthRequiredState = NonNullable< ReturnType >; @@ -2284,6 +2308,11 @@ export const createExecutor = => { - const existingState = decodeJsonColumn(row.provider_state); - const mergedState = - existingState != null && typeof existingState === "object" && !Array.isArray(existingState) - ? (existingState as Record) - : {}; - const health: HealthCheckResult = { - status: "expired", - checkedAt: Date.now(), - detail, - reason, + const ref: ConnectionRef = { + owner: row.owner as Owner, + integration: IntegrationSlug.make(row.integration), + name: ConnectionName.make(row.name), }; - return core - .updateMany("connection", { + const observedExpiry = row.expires_at == null ? null : Number(row.expires_at); + const record = (target: ConnectionRow): Effect.Effect => { + const mergedState = providerStateRecord(target.provider_state); + const health: HealthCheckResult = { + status: "expired", + checkedAt: Date.now(), + detail, + reason, + }; + return core.updateMany("connection", { where: (b: AnyCb) => b.and( - byOwner(row.owner as Owner)(b), - b("integration", "=", String(row.integration)), - b("name", "=", String(row.name)), + byOwner(target.owner as Owner)(b), + b("integration", "=", String(target.integration)), + b("name", "=", String(target.name)), ), set: { provider_state: { @@ -2333,8 +2364,28 @@ export const createExecutor = { + if (fresh === null) return record(row); + const freshExpiry = fresh.expires_at == null ? null : Number(fresh.expires_at); + const peerRefreshed = + freshExpiry !== null && (observedExpiry === null || freshExpiry > observedExpiry); + return peerRefreshed + ? Effect.annotateCurrentSpan({ + "executor.oauth.dead_grant.skipped_peer_refresh": true, + }) + : record(fresh); + }), + Effect.ignore, + ); }; /** Write a re-minted token back: a ROTATED refresh token into the refresh @@ -2383,13 +2434,32 @@ export const createExecutor = = { - expires_at: nextExpiresAt, + expires_at: lifetimeMs === null ? null : Date.now() + lifetimeMs, updated_at: new Date(), }; if (token.scope !== undefined) set.oauth_scope = token.scope; + if (advertisedLifetimeMs !== null && advertisedLifetimeMs !== rememberedLifetimeMs) { + // Merge into a freshly read `provider_state`, so a concurrent write + // (a dead-grant record, a missing-scope set) survives. + const ref: ConnectionRef = { + owner: row.owner as Owner, + integration: IntegrationSlug.make(row.integration), + name: ConnectionName.make(row.name), + }; + const fresh = yield* findConnectionRow(ref); + set.provider_state = { + ...providerStateRecord((fresh ?? row).provider_state), + oauthTokenLifetimeMs: advertisedLifetimeMs, + }; + } yield* core.updateMany("connection", { where: (b: AnyCb) => b.and( @@ -2739,7 +2809,36 @@ export const createExecutor = => + Effect.gen(function* () { + if (error.reauthRequired !== true) return yield* Effect.fail(error); + const sent = storedRefreshToken; + if (sent === undefined || row.refresh_item_id == null) { + return yield* Effect.fail(error); + } + const current = yield* provider.get(ProviderItemId.make(row.refresh_item_id)); + if (current === null || current === sent) return yield* Effect.fail(error); + const tokenItemId = + connectionItemIds(row)[PRIMARY_INPUT_VARIABLE] ?? + `connection:${row.owner}:${row.integration}:${row.name}:${PRIMARY_INPUT_VARIABLE}`; + const access = yield* provider.get(ProviderItemId.make(tokenItemId)); + if (access === null) return yield* Effect.fail(error); + return access; + }); + const token: OAuth2TokenResponse | AdoptedAccessToken = clientRow.grant === "client_credentials" ? yield* exchangeClientCredentials({ tokenUrl, @@ -2874,6 +2973,11 @@ export const createExecutor = + adoptPeerRotatedToken(error), + ), // Persist the definitive verdict so the NEXT refresh skips // the doomed grant (see the known-dead gate above) and the // connection shows `expired` without waiting for a probe. @@ -2887,6 +2991,13 @@ export const createExecutor = => - foldCredentialResolutionIntoVerdict( - resolveConnectionValues(row).pipe( - Effect.as({ - status: "healthy" as const, + /** The verdict when nothing can probe: "the credential resolved" is the only + * signal this path produces. A refresh failure reaches the caller as a + * folded CredentialResolutionError instead. A null value means the stored + * credential is gone — rendering omits that placement, and an upstream + * that answers unauthenticated would otherwise look alive. */ + const credentialOnlyVerdict = (values: Record): HealthCheckResult => + Object.values(values).some((value) => value == null) + ? { + status: "expired", + checkedAt: Date.now(), + detail: "Connection has no resolvable credential value.", + reason: "credential_missing", + } + : { + status: "healthy", checkedAt: Date.now(), detail: "Credential resolved (no probe configured).", - }), - ), - ); + }; // Resolve an in-flight credential's value map (key-first validation) without // saving anything. Mirrors `resolveConnectionValues` for the saved-row path: @@ -5234,64 +5356,90 @@ export const createExecutor = = {}; 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, - yield* 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, - })), + foldCredentialResolutionIntoVerdict( + Effect.gen(function* () { + const record = rowToIntegrationRecord( + integrationRow, + yield* describeAuthMethodsForRow(integrationRow), ); + const grantedScopes = grantedScopesFromRow(connectionRow); + const probe = ( + values: Record, + ): Effect.Effect => { + 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 foldPluginFailure( + check({ + ctx: runtime.ctx, + integration: record, + credential, + spec, + }), + `Health check for connection "${ref.name}" failed.`, + ); + }; + const values = yield* resolveConnectionValues(connectionRow); + resolvedValues = values; + const first = yield* probe(values); + // A probe's `expired` is only as good as the credential it was + // handed. The invoke path re-mints once on a 401; the probe did + // not, so a revoked token, an idle timeout, or a null + // `expires_at` showed a working connection as dead until a tool + // call healed it. One forced refresh and one re-probe, OAuth + // connections only; a refusal keeps the probe's own verdict. + if (first.status !== "expired" || connectionRow.oauth_client == null) { + return first; + } + const refreshed = yield* forceRefreshConnectionValues(connectionRow).pipe( + Effect.catchTag("CredentialResolutionError", () => Effect.succeed(null)), + ); + if (refreshed === null) return first; + yield* Effect.annotateCurrentSpan({ + "executor.health.refresh_retried": true, + }); + return yield* probe(refreshed); + }), + ).pipe( + // Ask the plugin first, even with no declared spec: a plugin + // whose checkHealth needs no spec (MCP lists tools) gives a real + // verdict. Only a plugin that answers `unknown` falls back to the + // credential-only verdict, computed from the values the probe + // already resolved so nothing refreshes twice. + Effect.map((result) => + spec === undefined && + connectionRow.oauth_client != null && + result.status === "unknown" + ? { + source: "credential_only" as const, + result: credentialOnlyVerdict(resolvedValues), + } + : { source: "probe" as const, result }, + ), + // 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((outcome) => persistProbeHealthResult(ref, outcome.result)), + ); const run = freshVerdict.pipe( Effect.exit, Effect.flatMap((exit) => Deferred.done(deferred, exit)), diff --git a/packages/core/sdk/src/health-check.test.ts b/packages/core/sdk/src/health-check.test.ts index 61b652267..c3f650912 100644 --- a/packages/core/sdk/src/health-check.test.ts +++ b/packages/core/sdk/src/health-check.test.ts @@ -137,4 +137,36 @@ describe("classifyProbeResponse", () => { ); } }); + + // A scope shortfall authenticated: the credential works, the grant is + // narrower than this operation needs, and the remedy is a NEW CONSENT — which + // the connection's `missingOAuthScopes` already offers. Reporting it as + // `expired` told the user the connection was dead and sent them through a + // reconnect that could not widen the grant. + it("classifies an RFC 6750 insufficient_scope challenge as degraded", () => { + expect( + classifyProbeResponse(403, undefined, { + "www-authenticate": 'Bearer error="insufficient_scope", scope="read write"', + }), + ).toBe("degraded"); + }); + + it("classifies a body-named insufficient_scope as degraded", () => { + expect(classifyProbeResponse(403, { error: "insufficient_scope" })).toBe("degraded"); + expect( + classifyProbeResponse(403, { + error: { code: 403, details: [{ reason: "ACCESS_TOKEN_SCOPE_INSUFFICIENT" }] }, + }), + ).toBe("degraded"); + }); + + it("keeps the configuration carve-out ahead of the scope one", () => { + expect( + classifyProbeResponse( + 403, + { error: { errors: [{ reason: "accessNotConfigured" }], code: 403 } }, + { "www-authenticate": 'Bearer error="insufficient_scope"' }, + ), + ).toBe("misconfigured"); + }); }); diff --git a/packages/core/sdk/src/health-check.ts b/packages/core/sdk/src/health-check.ts index a4d6f1cb2..088012d13 100644 --- a/packages/core/sdk/src/health-check.ts +++ b/packages/core/sdk/src/health-check.ts @@ -17,6 +17,8 @@ import { Schema } from "effect"; +import { detectInsufficientScope } from "./insufficient-scope"; + // --------------------------------------------------------------------------- // Status: the five states a connection can be in. `expired` is the one this // whole feature exists for (Google's 7-day dev-token revocation): the credential @@ -241,17 +243,25 @@ const errorReasonMarkers = (body: unknown): string[] => { return markers; }; -/** Classify a probe response from its status AND body. Everything is - * `classifyHttpStatus` except one carve-out: a 403 whose error body carries a - * known configuration reason (Google `accessNotConfigured` / - * `SERVICE_DISABLED`) is `misconfigured`, not `expired`: the credential - * authenticated; the upstream API is disabled in the OAuth client's project, - * and only enabling it there (not reconnecting) fixes it. */ -export const classifyProbeResponse = (status: number, body: unknown): HealthStatus => { +/** Classify a probe response from its status, body, and (optionally) headers. + * Everything is `classifyHttpStatus` except two 403 carve-outs, both of which + * authenticated: a known configuration reason (Google `accessNotConfigured` / + * `SERVICE_DISABLED`) is `misconfigured`, and a scope shortfall (RFC 6750 + * `insufficient_scope`) is `degraded` — the remedy is a new consent, not a + * reconnect, so `expired` would send the user through a flow that cannot fix + * it. */ +export const classifyProbeResponse = ( + status: number, + body: unknown, + headers?: Record, +): HealthStatus => { const byStatus = classifyHttpStatus(status); if (status !== 403 || byStatus !== "expired") return byStatus; - return errorReasonMarkers(body).some((reason) => CONFIGURATION_403_REASONS.has(reason)) - ? "misconfigured" + if (errorReasonMarkers(body).some((reason) => CONFIGURATION_403_REASONS.has(reason))) { + return "misconfigured"; + } + return detectInsufficientScope({ body, ...(headers === undefined ? {} : { headers }) }) !== null + ? "degraded" : "expired"; }; diff --git a/packages/core/sdk/src/oauth-flow.test.ts b/packages/core/sdk/src/oauth-flow.test.ts index 7496ce1ef..a3f82ae5e 100644 --- a/packages/core/sdk/src/oauth-flow.test.ts +++ b/packages/core/sdk/src/oauth-flow.test.ts @@ -54,11 +54,17 @@ const oauthPlugin = definePlugin(() => ({ }, // Echo the resolved credential value (the OAuth access token) back out. invokeTool: ({ credential }) => Effect.succeed({ token: credential.value }), - checkHealth: ({ credential }) => - Effect.succeed({ - status: credential.value === null ? "expired" : "healthy", - checkedAt: Date.now(), - }), + // Mirrors the protocol plugins: with no declared probe operation there is + // nothing to dial, so the plugin answers `unknown` and core falls back to the + // credential-only verdict. A plugin that CAN answer without a spec (MCP lists + // tools) is asked first and gives a real verdict. + checkHealth: ({ credential, spec }) => + spec === undefined + ? Effect.succeed({ status: "unknown" as const, checkedAt: Date.now() }) + : Effect.succeed({ + status: credential.value === null ? ("expired" as const) : ("healthy" as const), + checkedAt: Date.now(), + }), extension: (ctx) => ({ seed: (scopes: readonly string[] = []) => ctx.core.integrations.register({ @@ -2085,7 +2091,7 @@ describe("oauth token refresh in resolveConnectionValue", () => { where: (b) => b("name", "=", "main"), }), ); - expect(row?.provider_state).toEqual({ missingOAuthScopes: ["write"] }); + expect(row?.provider_state).toMatchObject({ missingOAuthScopes: ["write"] }); const listed = yield* executor.connections.list({ integration: INTEG }); expect(listed[0]?.missingOAuthScopes).toEqual(["write"]); }), @@ -2136,7 +2142,12 @@ describe("oauth token refresh in resolveConnectionValue", () => { where: (b) => b("name", "=", "main"), }), ); - expect(row?.provider_state).toBeNull(); + // No missing-scope record. `provider_state` itself is not null: the + // mint records the advertised token lifetime there so a later refresh + // whose response omits `expires_in` can still derive an expiry. + expect( + (row?.provider_state as { missingOAuthScopes?: unknown } | null)?.missingOAuthScopes, + ).toBeUndefined(); }), ), ); diff --git a/packages/core/sdk/src/oauth-helpers.test.ts b/packages/core/sdk/src/oauth-helpers.test.ts index d94e8c34a..1ecfb6e7b 100644 --- a/packages/core/sdk/src/oauth-helpers.test.ts +++ b/packages/core/sdk/src/oauth-helpers.test.ts @@ -1781,6 +1781,48 @@ describe("refreshAccessToken", () => { expect(isPermanentTokenRejection(error)).toBe(false); }), ); + + // A 4xx that describes THIS MINUTE rather than this grant must stay + // retryable: the identical request can succeed moments later. 429 is the + // authorization server asking us to come back, 408 is its own request + // timeout, 425 is a transport-level replay refusal. Reading any of them as + // definitive ended connections permanently on one rate-limited minute — and + // rate limiting is likelier the more refreshers race for one grant. + for (const status of [408, 425, 429] as const) { + it.effect(`keeps a ${status} response transient`, () => + withTokenEndpoint( + () => HttpServerResponse.text("slow down", { status }), + ({ tokenUrl }) => + Effect.gen(function* () { + const error = yield* Effect.flip( + refreshAccessToken({ tokenUrl, clientId: "cid", refreshToken: "old" }), + ); + expect(error.status).toBe(status); + expect(error.error).toBeUndefined(); + expect(isPermanentTokenRejection(error)).toBe(false); + }), + ), + ); + } + + // The definitive 4xx stay definitive: §5.2 mandates 400 for a grant the AS + // will not honour, and a token endpoint answering 404 does not start + // existing on the next attempt. + for (const status of [400, 404] as const) { + it.effect(`keeps a text/plain ${status} response definitive`, () => + withTokenEndpoint( + () => HttpServerResponse.text("your session has expired", { status }), + ({ tokenUrl }) => + Effect.gen(function* () { + const error = yield* Effect.flip( + refreshAccessToken({ tokenUrl, clientId: "cid", refreshToken: "old" }), + ); + expect(error.status).toBe(status); + expect(isPermanentTokenRejection(error)).toBe(true); + }), + ), + ); + } }); describe("shouldRefreshToken", () => { diff --git a/packages/core/sdk/src/oauth-helpers.ts b/packages/core/sdk/src/oauth-helpers.ts index c6e3209fe..a5c151708 100644 --- a/packages/core/sdk/src/oauth-helpers.ts +++ b/packages/core/sdk/src/oauth-helpers.ts @@ -73,22 +73,34 @@ export class OAuth2Error extends Data.TaggedError("OAuth2Error")<{ export const isUnusableSuccessTokenResponse = (error: OAuth2Error): boolean => error.status !== undefined && error.status < 300; +/** 4xx statuses that describe THIS MINUTE rather than this grant. A 429 is the + * authorization server asking us to come back, a 408 is its own request + * timeout, and a 425 is a transport-level replay refusal — none of them is a + * verdict on the refresh token, and re-sending the identical grant later can + * succeed. Treating them as definitive ended connections permanently on one + * rate-limited minute, which is likelier the more refreshers race. */ +const TRANSIENT_4XX_STATUSES: ReadonlySet = new Set([408, 425, 429]); + /** * Did the token endpoint answer in a way that re-sending the identical grant * cannot change? * - * Yes for a 4xx — §5.2 mandates 400 for a grant the authorization server will - * not honour, 401/403 are refusals, and a token endpoint answering 404 does not - * start existing on the next attempt — and yes for a 2xx that carried no usable - * token, because the server called it a success and still issued nothing. + * Yes for a 4xx that is not one of {@link TRANSIENT_4XX_STATUSES} — §5.2 + * mandates 400 for a grant the authorization server will not honour, 401/403 + * are refusals, and a token endpoint answering 404 does not start existing on + * the next attempt — and yes for a 2xx that carried no usable token, because + * the server called it a success and still issued nothing. * - * No for a 5xx (the AS is having a bad minute) and no when there is no response - * at all (transport). Those are exactly the failures a later attempt survives, - * so they must stay retryable. + * No for a 5xx (the AS is having a bad minute), no for a rate-limited or + * timed-out 4xx, and no when there is no response at all (transport). Those are + * exactly the failures a later attempt survives, so they must stay retryable. */ export const isPermanentTokenRejection = (error: OAuth2Error): boolean => isUnusableSuccessTokenResponse(error) || - (error.status !== undefined && error.status >= 400 && error.status < 500); + (error.status !== undefined && + error.status >= 400 && + error.status < 500 && + !TRANSIENT_4XX_STATUSES.has(error.status)); // --------------------------------------------------------------------------- // Token response shape (RFC 6749 §5.1) diff --git a/packages/core/sdk/src/oauth-refresh-evidence.test.ts b/packages/core/sdk/src/oauth-refresh-evidence.test.ts new file mode 100644 index 000000000..bf27aa464 --- /dev/null +++ b/packages/core/sdk/src/oauth-refresh-evidence.test.ts @@ -0,0 +1,638 @@ +// Regression coverage for the connection status and OAuth refresh defects that +// produced a permanent, wrong **Expired**: a refresher that loses a rotation +// race, a rate-limited token endpoint, a probe that answered without +// refreshing, and a refresh response that omits `expires_in`. +// +// One database, one credential store, two executor instances, one root database +// handle each. The in-flight refresh gate is keyed on the handle, so two +// instances do not share a gate — the cloud app's per-request `DbService` +// rebuild and any multi-process self-host both have this shape. The tests +// assert on the connection ROW, which is where the wrong status was written: +// `oauth-flow.test.ts` already covers this shape for the credential store. + +import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; + +import { describe, expect, it } from "@effect/vitest"; +import { Deferred, Effect, Fiber, Option, Schema } from "effect"; +import * as Exit from "effect/Exit"; +import { withQueryContext } from "@executor-js/fumadb/query"; + +import { authToolFailure } from "./auth-tool-failure"; +import { createExecutor } from "./executor"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + OAuthClientSlug, + ProviderKey, + ToolAddress, + ToolName, +} from "./ids"; +import { definePlugin } from "./plugin"; +import type { CredentialProvider } from "./provider"; +import { makeTestConfig } from "./test-config"; +import { serveOAuthTestServer } from "./testing/oauth-test-server"; +import { ToolResult } from "./tool-result"; + +const TENANT = "test-tenant"; +const SUBJECT = "test-subject"; +const INTEG = IntegrationSlug.make("acme"); +const TEMPLATE = AuthTemplateSlug.make("oauth"); +const CLIENT = OAuthClientSlug.make("acme-app"); +const NAME = ConnectionName.make("main"); +const ADDRESS = ToolAddress.make("tools.acme.org.main.whoami"); +const REF = { owner: "org" as const, integration: INTEG, name: NAME }; + +// --------------------------------------------------------------------------- +// Plugin: an upstream that honours every access token except the revoked ones. +// `checkHealth` authenticates the same way `invokeTool` does, so a revoked +// token reads 401 on both paths — the divergence under test is what CORE does +// with each, not what the plugin reports. +// --------------------------------------------------------------------------- + +interface UpstreamState { + readonly revoked: Set; + readonly calls: string[]; + readonly probes: string[]; +} + +const makeUpstreamPlugin = (state: UpstreamState) => + definePlugin(() => ({ + id: "acme" as const, + storage: () => ({}), + resolveTools: () => + Effect.succeed({ tools: [{ name: ToolName.make("whoami"), description: "whoami" }] }), + describeAuthMethods: () => [ + { + id: "oauth", + label: "OAuth2", + kind: "oauth" as const, + template: String(TEMPLATE), + oauth: { scopes: [] }, + }, + ], + invokeTool: ({ credential }) => { + const token = credential.value; + state.calls.push(String(token)); + if (token !== null && !state.revoked.has(token)) { + return Effect.succeed(ToolResult.ok({ token })); + } + return Effect.succeed( + authToolFailure({ + code: "connection_rejected", + status: 401, + message: "Upstream rejected credentials with HTTP 401.", + integration: { id: String(credential.integration) }, + credential: { kind: "upstream", label: String(credential.connection) }, + }), + ); + }, + checkHealth: ({ credential }) => { + const token = credential.value; + state.probes.push(String(token)); + if (token !== null && !state.revoked.has(token)) { + return Effect.succeed({ status: "healthy" as const, checkedAt: Date.now() }); + } + return Effect.succeed({ + status: "expired" as const, + httpStatus: 401, + checkedAt: Date.now(), + detail: "The endpoint rejected the credential with HTTP 401.", + reason: "upstream_status" as const, + }); + }, + extension: (ctx) => ({ + seed: () => ctx.core.integrations.register({ slug: INTEG, description: "Acme", config: {} }), + }), + }))(); + +// --------------------------------------------------------------------------- +// Shared credential store, with a seam that can hold ONE reader between its +// read of the stored refresh token and whatever it does next. That seam is the +// race window; the same one `oauth-flow.test.ts` opens. +// --------------------------------------------------------------------------- + +interface SharedStore { + readonly provider: CredentialProvider; + readonly values: Map; + readonly writes: string[]; + /** Arm the one-shot pause on the next refresh-token read. */ + readonly arm: () => void; +} + +const makeSharedStore = (input: { + readonly pausedAtRead: Deferred.Deferred; + readonly resumeFromRead: Deferred.Deferred; +}): SharedStore => { + const values = new Map(); + const writes: string[] = []; + let pauseNextRefreshRead = false; + return { + values, + writes, + arm: () => { + pauseNextRefreshRead = true; + }, + provider: { + key: ProviderKey.make("shared-memory"), + writable: true, + get: (id) => + Effect.gen(function* () { + const value = values.get(String(id)) ?? null; + if (pauseNextRefreshRead && String(id).endsWith(":refresh")) { + pauseNextRefreshRead = false; + yield* Deferred.succeed(input.pausedAtRead, undefined); + yield* Deferred.await(input.resumeFromRead); + } + return value; + }), + set: (id, value) => + Effect.sync(() => { + writes.push(String(id)); + values.set(String(id), value); + }), + delete: (id) => Effect.sync(() => void values.delete(String(id))), + }, + }; +}; + +// --------------------------------------------------------------------------- +// One database + one store, two executor instances, one completed OAuth +// connection. `expire` forces the next resolve to refresh. +// --------------------------------------------------------------------------- + +const makeRace = (options?: { readonly healthCheck?: boolean }) => + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const state: UpstreamState = { revoked: new Set(), calls: [], probes: [] }; + const pausedAtRead = yield* Deferred.make(); + const resumeFromRead = yield* Deferred.make(); + const store = makeSharedStore({ pausedAtRead, resumeFromRead }); + const config = { + ...makeTestConfig({ + plugins: [makeUpstreamPlugin(state)] as const, + tenant: TENANT, + subject: SUBJECT, + }), + providers: [store.provider], + }; + const a = yield* createExecutor(config); + // A SECOND root db handle onto the same database = a second instance. The + // in-flight refresh gate is keyed on handle identity, so this is exactly + // the boundary the gate cannot see across. + const b = yield* createExecutor({ + ...config, + db: withQueryContext(config.testDb.db, { tenant: TENANT, subject: SUBJECT }), + }); + yield* Effect.addFinalizer(() => a.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => b.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => + Effect.promise(() => config.testDb.close()).pipe(Effect.ignore), + ); + + yield* a.acme.seed(); + yield* a.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + if (options?.healthCheck === true) { + // A declared health check puts the connection on the PROBING path rather + // than the credential-only one. + yield* a.integrations.healthCheck.set(INTEG, { operation: "whoami" }); + } + const started = yield* a.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: NAME, + integration: INTEG, + template: TEMPLATE, + }); + if (started.status !== "redirect") return yield* Effect.die("expected a redirect start"); + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + yield* a.oauth.complete({ state: started.state, code: callback.code }); + + return { + server, + state, + store, + a, + b, + config, + pausedAtRead, + resumeFromRead, + expire: () => + Effect.promise(() => + config.db.updateMany("connection", { + where: (builder) => builder("name", "=", String(NAME)), + set: { expires_at: Date.now() - 60_000 }, + }), + ), + rawRow: () => + Effect.promise(() => + config.db.findFirst("connection", { + where: (builder) => builder("name", "=", String(NAME)), + }), + ), + refreshItemId: () => [...store.values.keys()].find((key) => key.endsWith(":refresh")), + } as const; + }); + +type Race = Effect.Success>; + +/** Run `use` against a freshly connected two-instance race. */ +const withRace = ( + options: { readonly healthCheck?: boolean }, + use: (race: Race) => Effect.Effect, +) => Effect.scoped(Effect.flatMap(makeRace(options), use)); + +const refreshGrants = (requests: readonly { readonly path: string; readonly body: string }[]) => + requests.filter((r) => r.path === "/token" && r.body.includes("grant_type=refresh_token")); + +const DeadGrantState = Schema.Struct({ oauthReauthRequiredAt: Schema.Number }); +const decodeDeadGrantObject = Schema.decodeUnknownOption(DeadGrantState); +const decodeDeadGrantJson = Schema.decodeUnknownOption(Schema.fromJsonString(DeadGrantState)); + +/** The recorded dead-grant stamp, read off a connection row's + * `provider_state` (a JSON column: an object on some adapters, encoded text + * on others). Takes `unknown` because the row comes back from the raw query + * surface, and normalises it through Schema rather than a cast. */ +const RowProviderState = Schema.Struct({ provider_state: Schema.optional(Schema.Unknown) }); +const decodeRowProviderState = Schema.decodeUnknownOption(RowProviderState); + +const deadGrantStamp = (row: unknown): number | undefined => { + const value = Option.getOrUndefined( + Option.map(decodeRowProviderState(row), (decoded) => decoded.provider_state), + ); + if (value === undefined || value === null) return undefined; + const fromObject = Option.getOrUndefined(decodeDeadGrantObject(value)); + if (fromObject !== undefined) return fromObject.oauthReauthRequiredAt; + return Option.getOrUndefined( + Option.map(decodeDeadGrantJson(value), (state) => state.oauthReauthRequiredAt), + ); +}; + +// --------------------------------------------------------------------------- +// A refresher that loses the rotation race adopts the peer's token. +// --------------------------------------------------------------------------- + +/** Run the race: A reads the stored refresh token and stalls, B wins and + * rotates it, A resumes and redeems the consumed token. Returns the rotated + * token and A's own outcome, so a test can assert what the loser did with the + * refusal. */ +const runRotationRace = (race: Race) => + Effect.gen(function* () { + const refreshItemId = race.refreshItemId(); + expect(refreshItemId, "the connection stored a refresh token").toBeDefined(); + const originalRefreshToken = race.store.values.get(refreshItemId!); + yield* race.expire(); + + race.store.arm(); + const loser = yield* Effect.forkChild(Effect.exit(race.a.execute(ADDRESS, {}))); + yield* Deferred.await(race.pausedAtRead); + + // B wins: it spends that token, the AS rotates it, B stores the successor. + yield* race.b.execute(ADDRESS, {}); + const rotatedRefreshToken = race.store.values.get(refreshItemId!); + expect(rotatedRefreshToken, "the winner rotated the stored refresh token").not.toBe( + originalRefreshToken, + ); + + // A resumes and redeems a token the authorization server already consumed. + yield* Deferred.succeed(race.resumeFromRead, undefined); + const loserExit = yield* Fiber.join(loser); + + // The store still holds B's valid rotated token: this connection is not out + // of credentials, it lost a race. + expect(race.store.values.get(refreshItemId!)).toBe(rotatedRefreshToken); + return { + refreshItemId: refreshItemId!, + rotatedRefreshToken: rotatedRefreshToken!, + loserExit, + }; + }); + +describe("a lost rotation race is not a dead grant", () => { + it.effect("the loser adopts the peer's token and the connection keeps refreshing", () => + withRace({}, (race) => + Effect.gen(function* () { + const { loserExit } = yield* runRotationRace(race); + + // The loser's own call recovered: it read the refresh item back, saw + // that a peer had replaced the value it sent, and used the access token + // that peer persisted. + expect(Exit.isSuccess(loserExit), "the losing refresher still served its call").toBe(true); + + // No permanent rejection record, because the grant is alive. + expect(deadGrantStamp(yield* race.rawRow()), "no dead grant is recorded").toBeUndefined(); + const health = yield* race.b.connections.checkHealth(REF); + expect(health.status, "and no surface answers expired").not.toBe("expired"); + + // The winner still refreshes with its own rotated token when the access + // token next expires. + yield* race.expire(); + yield* race.server.clearRequests; + const next = yield* Effect.exit(race.b.execute(ADDRESS, {})); + expect(Exit.isSuccess(next), "the winner refreshes again on the next expiry").toBe(true); + expect( + refreshGrants(yield* race.server.requests).length, + "the authorization server received that grant", + ).toBeGreaterThan(0); + }), + ), + ); +}); + +// --------------------------------------------------------------------------- +// One temporary 4xx response does not end a grant. +// --------------------------------------------------------------------------- + +interface FlakyEndpoint { + readonly url: string; + readonly attempts: () => number; + readonly close: () => void; +} + +/** Token endpoint that rate-limits the FIRST refresh grant and forwards the + * rest to the real authorization server: one bad minute, then healthy. */ +const serveFlakyTokenEndpoint = (upstream: string) => + Effect.acquireRelease( + Effect.callback((resume) => { + let attempts = 0; + const forward = async ( + req: IncomingMessage, + res: ServerResponse, + body: string, + ): Promise => { + // oxlint-disable-next-line executor/no-raw-fetch -- boundary: test fixture proxying form-encoded token requests to the test authorization server; it must not carry the SDK's own HttpClient layer into the endpoint under test + const response = await fetch(upstream, { + method: req.method, + headers: { + "content-type": req.headers["content-type"] ?? "application/x-www-form-urlencoded", + ...(typeof req.headers["authorization"] === "string" + ? { authorization: req.headers["authorization"] } + : {}), + }, + body: body.length > 0 ? body : undefined, + }); + const text = await response.text(); + res.writeHead(response.status, { + "content-type": response.headers.get("content-type") ?? "application/json", + }); + res.end(text); + }; + const server: Server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + if (body.includes("grant_type=refresh_token")) { + attempts += 1; + if (attempts === 1) { + res.writeHead(429, { "content-type": "text/plain; charset=utf-8" }); + res.end("Too Many Requests: slow down"); + return; + } + } + // oxlint-disable-next-line executor/no-promise-catch -- boundary: plain node:http handler in a test fixture standing in for a flaky upstream + void forward(req, res, body).catch(() => { + res.writeHead(502, { "content-type": "text/plain" }); + res.end("proxy failed"); + }); + }); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address !== null ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}/token`, + attempts: () => attempts, + close: () => server.close(), + }), + ); + }); + }), + (handle) => Effect.sync(() => handle.close()), + ); + +/** Connect, point the backing app at a token endpoint that rate-limits once, + * and take that first failing refresh. */ +const withRateLimitedRefresh = ( + use: (input: { + readonly race: Race; + readonly flaky: FlakyEndpoint; + readonly firstCallSucceeded: boolean; + }) => Effect.Effect, +) => + Effect.scoped( + Effect.gen(function* () { + const race = yield* makeRace({}); + const flaky = yield* serveFlakyTokenEndpoint(race.server.tokenEndpoint); + yield* Effect.promise(() => + race.config.db.updateMany("oauth_client", { + where: (builder) => builder("slug", "=", String(CLIENT)), + set: { token_url: flaky.url }, + }), + ); + yield* race.expire(); + const first = yield* Effect.exit(race.a.execute(ADDRESS, {})); + expect(Exit.isSuccess(first), "the rate-limited refresh fails the call").toBe(false); + expect(flaky.attempts(), "the token endpoint was asked once").toBe(1); + return yield* use({ race, flaky, firstCallSucceeded: Exit.isSuccess(first) }); + }), + ); + +describe("a rate-limited refresh stays retryable", () => { + it.effect("a 429 from the token endpoint does not end the grant", () => + withRateLimitedRefresh(({ race, flaky }) => + Effect.gen(function* () { + // The endpoint forwards every grant after the first to the real + // authorization server, so it is healthy from here on. The next call + // must reach it. + const second = yield* Effect.exit(race.a.execute(ADDRESS, {})); + expect(Exit.isSuccess(second), "the retry refreshed and the call succeeded").toBe(true); + expect(flaky.attempts(), "executor asked the token endpoint again").toBeGreaterThan(1); + + expect( + deadGrantStamp(yield* race.rawRow()), + "a rate limit is not a permanent rejection", + ).toBeUndefined(); + const health = yield* race.a.connections.checkHealth(REF); + expect(health.status, "and the connection does not read as expired").not.toBe("expired"); + }), + ), + ); +}); + +// --------------------------------------------------------------------------- +// A refresh response without `expires_in` keeps the advertised lifetime. +// --------------------------------------------------------------------------- + +interface StrippingEndpoint { + readonly url: string; + readonly attempts: () => number; + readonly close: () => void; +} + +/** A token endpoint that answers a refresh grant with a valid token response + * that OMITS `expires_in`, which RFC 6749 permits, and rotates the refresh + * token like the real one does. It stands in for an authorization server that + * advertised a lifetime on the code exchange and then stopped repeating it. */ +const serveExpiresInStrippingEndpoint = () => + Effect.acquireRelease( + Effect.callback((resume) => { + let attempts = 0; + const server: Server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + if (!body.includes("grant_type=refresh_token")) { + res.writeHead(400, { "content-type": "text/plain" }); + res.end("this fixture answers refresh grants only"); + return; + } + attempts += 1; + res.writeHead(200, { "content-type": "application/json", "cache-control": "no-store" }); + res.end( + `{"access_token":"at_stripped_${attempts}","refresh_token":"rt_stripped_${attempts}","token_type":"Bearer"}`, + ); + void req; + }); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address !== null ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}/token`, + attempts: () => attempts, + close: () => server.close(), + }), + ); + }); + }), + (handle) => Effect.sync(() => handle.close()), + ); + +/** `connection.expires_at` off the raw row, which adapters return as a number or + * a string. */ +const RowExpiry = Schema.Struct({ expires_at: Schema.optional(Schema.Unknown) }); +const decodeRowExpiry = Schema.decodeUnknownOption(RowExpiry); +const rowExpiresAt = (row: unknown): number | null => { + const value = Option.getOrUndefined(decodeRowExpiry(row))?.expires_at; + // A bigint column: adapters hand back a number, a string, or a BigInt. + if (typeof value === "number") return value; + if (typeof value === "bigint") return Number(value); + if (typeof value === "string") return Number(value); + return null; +}; + +// --------------------------------------------------------------------------- +// The probe refreshes before it answers expired. +// --------------------------------------------------------------------------- + +/** Connect with a declared health check, then revoke the live access token + * upstream while `expires_at` still says it is good for an hour. */ +const withRevokedToken = (use: (race: Race) => Effect.Effect) => + withRace({ healthCheck: true }, (race) => + Effect.gen(function* () { + for (const token of yield* race.server.issuedAccessTokens) race.state.revoked.add(token); + yield* race.server.clearRequests; + return yield* use(race); + }), + ); + +describe("the probe refreshes before it answers expired", () => { + it.effect("a revoked token that the refresh can replace probes healthy", () => + withRevokedToken((race) => + Effect.gen(function* () { + const verdict = yield* race.a.connections.checkHealth(REF); + expect(verdict.status, "the probe re-minted instead of reporting expired").toBe("healthy"); + expect( + refreshGrants(yield* race.server.requests).length, + "the probe sent a refresh grant", + ).toBeGreaterThan(0); + + // The persisted verdict agrees, so every surface reads healthy without + // waiting for a tool call to heal it. + const persisted = yield* race.a.connections.get(REF); + expect(persisted?.lastHealth?.status, "the healthy verdict is persisted").toBe("healthy"); + }), + ), + ); + + it.effect("a refused refresh still answers expired from the probe", () => + withRace({ healthCheck: true }, (race) => + Effect.gen(function* () { + // No revocation and no expiry: the probe answers from the credential it + // resolved. A refusal is what the persisted-expired contract in + // `connection-health-verdict.test.ts` covers; this asserts the retry + // did not turn the probe into a second grant on a healthy connection. + yield* race.server.clearRequests; + const verdict = yield* race.a.connections.checkHealth(REF); + expect(verdict.status).toBe("healthy"); + expect( + refreshGrants(yield* race.server.requests), + "a healthy probe sends no refresh grant", + ).toHaveLength(0); + }), + ), + ); +}); + +// --------------------------------------------------------------------------- +// A refresh response without `expires_in` must not erase the expiry. +// --------------------------------------------------------------------------- + +describe("a refresh response that omits expires_in", () => { + it.effect("keeps the advertised lifetime, so proactive refresh survives", () => + Effect.scoped( + Effect.gen(function* () { + const race = yield* makeRace({}); + const mintedExpiry = rowExpiresAt(yield* race.rawRow()); + expect(mintedExpiry, "the mint recorded the advertised expiry").not.toBeNull(); + expect( + (mintedExpiry ?? 0) - Date.now(), + "and the test authorization server advertised an hour", + ).toBeGreaterThan(30 * 60_000); + + const stripping = yield* serveExpiresInStrippingEndpoint(); + yield* Effect.promise(() => + race.config.db.updateMany("oauth_client", { + where: (builder) => builder("slug", "=", String(CLIENT)), + set: { token_url: stripping.url }, + }), + ); + yield* race.expire(); + + const first = yield* Effect.exit(race.a.execute(ADDRESS, {})); + expect(Exit.isSuccess(first), "the refresh succeeded").toBe(true); + expect(stripping.attempts(), "and it went to the endpoint that omits expires_in").toBe(1); + + // This wrote null before the fix, which disabled the proactive check + // for the rest of the connection's life and left every later call to + // the reactive 401 path. + const refreshedExpiry = rowExpiresAt(yield* race.rawRow()); + expect(refreshedExpiry, "the expiry survived a response without expires_in").not.toBeNull(); + expect( + (refreshedExpiry ?? 0) - Date.now(), + "and it carries the lifetime the grant advertised", + ).toBeGreaterThan(30 * 60_000); + + // The proactive path still works, so the next call needs no grant. + const second = yield* Effect.exit(race.a.execute(ADDRESS, {})); + expect(Exit.isSuccess(second), "the next call used the stored token").toBe(true); + expect(stripping.attempts(), "and sent no second grant").toBe(1); + }), + ), + ); +}); diff --git a/packages/core/sdk/src/testing/oauth-test-server.ts b/packages/core/sdk/src/testing/oauth-test-server.ts index 0380e389f..3887ae698 100644 --- a/packages/core/sdk/src/testing/oauth-test-server.ts +++ b/packages/core/sdk/src/testing/oauth-test-server.ts @@ -512,6 +512,9 @@ const SUPPORTED_SUBJECT_TOKEN_TYPES = new Set([ const JwksUriMetadata = Schema.Struct({ jwks_uri: Schema.String }); const decodeJwksUriMetadata = Schema.decodeUnknownOption(JwksUriMetadata); +/** One JSON-RPC frame off the `/mcp` resource endpoint's body. */ +const decodeMcpResourceFrame = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); + /** Resolve a trusted IdP's signing keys the way a Resource Authorization Server * does: read its RFC 8414 metadata, follow `jwks_uri`, fetch the key set. Any * failure yields `None`, which the caller reports as `invalid_grant` — the @@ -1076,10 +1079,44 @@ export const serveOAuthTestServer = ( }, ); } + // A minimal but honest MCP resource server. The old canned reply used + // a FIXED json-rpc id and answered notifications too, so a client + // that completed the handshake waited forever for its `tools/list` + // response: every catalog sync and every liveness probe against this + // endpoint timed out at the discovery deadline. Answer the request's + // OWN id, answer `tools/list` with an empty catalog, and stay silent + // for notifications, which is the protocol. + const frame = Option.getOrUndefined(decodeMcpResourceFrame(body)); + const frameIsRecord = frame !== null && typeof frame === "object"; + const frameRecord = frameIsRecord ? (frame as Record) : {}; + const frameMethod = + typeof frameRecord["method"] === "string" ? frameRecord["method"] : ""; + const hasId = "id" in frameRecord && frameRecord["id"] !== undefined; + if (!hasId) { + // Accepted with no body: a notification has no reply. + return HttpServerResponse.empty({ status: 202 }); + } + const params = + frameRecord["params"] !== null && typeof frameRecord["params"] === "object" + ? (frameRecord["params"] as Record) + : {}; + const reply = + frameMethod === "tools/list" + ? { tools: [] } + : frameMethod === "initialize" + ? { + protocolVersion: + typeof params["protocolVersion"] === "string" + ? params["protocolVersion"] + : "2025-06-18", + capabilities: { tools: {} }, + serverInfo: { name: "oauth-test-server", version: "0.0.0" }, + } + : {}; return jsonResponse(200, { jsonrpc: "2.0", - id: 1, - result: { protocolVersion: "2025-06-18", capabilities: {} }, + id: frameRecord["id"], + result: reply, }); } diff --git a/packages/plugins/graphql/src/sdk/health-classification.test.ts b/packages/plugins/graphql/src/sdk/health-classification.test.ts new file mode 100644 index 000000000..eb1d0b548 --- /dev/null +++ b/packages/plugins/graphql/src/sdk/health-classification.test.ts @@ -0,0 +1,48 @@ +// The prose a transport failure carries is the operating system's or the HTTP +// client's, not the upstream's verdict on a credential. Reading it as one sent +// users to re-enter a secret that was never the problem: `EACCES: permission +// denied` on a socket matches any pattern looking for the word "permission". + +import { describe, expect, it } from "@effect/vitest"; + +import { GraphqlIntrospectionError } from "./errors"; +import { healthFromIntrospectionError } from "./plugin"; + +const classify = (input: { + readonly reason?: "network" | "graphql-errors" | "invalid-json"; + readonly status?: number; + readonly upstreamMessage?: string; +}) => + healthFromIntrospectionError( + new GraphqlIntrospectionError({ + message: "introspection failed", + ...(input.reason === undefined ? {} : { reason: input.reason }), + ...(input.status === undefined ? {} : { status: input.status }), + ...(input.upstreamMessage === undefined ? {} : { upstreamMessage: input.upstreamMessage }), + }), + Date.now(), + ); + +describe("GraphQL liveness classification", () => { + it("does not read a transport failure's prose as a dead credential", () => { + const verdict = classify({ + reason: "network", + upstreamMessage: "connect EACCES: permission denied /run/graphql.sock", + }); + expect(verdict.status).not.toBe("expired"); + }); + + it("still reads an authentication failure the upstream named as expired", () => { + const verdict = classify({ + reason: "graphql-errors", + upstreamMessage: "Authentication required: invalid token", + }); + expect(verdict.status).toBe("expired"); + }); + + it("classifies an HTTP 401 on its status whatever else is said", () => { + const verdict = classify({ status: 401, upstreamMessage: "nope" }); + expect(verdict.status).toBe("expired"); + expect(verdict.httpStatus).toBe(401); + }); +}); diff --git a/packages/plugins/graphql/src/sdk/plugin.ts b/packages/plugins/graphql/src/sdk/plugin.ts index 176e6d772..dfe3f77f4 100644 --- a/packages/plugins/graphql/src/sdk/plugin.ts +++ b/packages/plugins/graphql/src/sdk/plugin.ts @@ -114,6 +114,14 @@ const appendUpstreamMessage = (detail: string, message?: string): string => ? `${detail} Upstream said: ${truncateHealthDetail(message)}` : detail; +/** Whether an upstream's own prose names an authentication failure. + * + * Text matching, so it is deliberately the WEAKER signal: it is consulted only + * when no HTTP status classified the failure, and never for a transport + * failure. An OS-level refusal carries the word "permission" too — `EACCES: + * permission denied` on a socket or a binary — and reading that as a dead + * credential sent the user to re-enter a secret that was never the problem. + * The caller excludes `reason: "network"` for exactly that case. */ const isAuthMessage = (message: string | undefined): boolean => message !== undefined && /authoriz|authenticat|forbidden|permission|credential|api.?key|access denied|access token|invalid token|token expired|logged in|sign in/i.test( @@ -141,7 +149,14 @@ const missingCredentialVariables = ( }); }; -const healthFromIntrospectionError = ( +/** Classify one introspection failure as a health verdict. + * + * Exported for tests (not re-exported from `sdk/index.ts`, so this widens no + * public API): the classification rules — which prose counts as an + * authentication failure, and which reason may never be read from prose — are + * the whole behavior under test, and reaching them through a live introspection + * would test the transport instead. */ +export const healthFromIntrospectionError = ( error: GraphqlIntrospectionError, checkedAt: number, ): HealthCheckResult => { @@ -160,7 +175,12 @@ const healthFromIntrospectionError = ( }; } - if (httpStatus === 401 || httpStatus === 403 || isAuthMessage(upstream)) { + // A transport failure never classifies from prose: its message is the OS's or + // the HTTP client's, and "permission denied" there names a socket, not a + // credential (see `isAuthMessage`). An HTTP 401/403 still classifies on its + // own status whatever the reason. + const proseSaysAuth = error.reason !== "network" && isAuthMessage(upstream); + if (httpStatus === 401 || httpStatus === 403 || proseSaysAuth) { const statusDetail = httpStatus === 401 || httpStatus === 403 ? `The endpoint rejected the credential with HTTP ${httpStatus}.` diff --git a/packages/plugins/mcp/src/sdk/discover.ts b/packages/plugins/mcp/src/sdk/discover.ts index d3c672361..19756bf37 100644 --- a/packages/plugins/mcp/src/sdk/discover.ts +++ b/packages/plugins/mcp/src/sdk/discover.ts @@ -5,7 +5,11 @@ import { Duration, Effect, Option, Predicate, Schema } from "effect"; import { hasNestedOAuthReauthorization, type McpConnection, type McpConnector } from "./connection"; -import { McpToolDiscoveryError } from "./errors"; +import { + type McpConnectionError, + type McpOAuthReauthorizationRequired, + McpToolDiscoveryError, +} from "./errors"; import { createMcpConnector, type ConnectorInput } from "./connection"; import { httpStatusFromCause } from "./http-status"; import { @@ -29,7 +33,8 @@ const MAX_LIST_TOOLS_PAGES = 100; // (`probeMcpEndpointShape`'s `timeoutMs = 8_000`) at a slightly longer // bound since a real handshake + listTools round-trip is heavier than the // shape probe's single unauth POST. -const DEFAULT_DISCOVER_TIMEOUT = Duration.seconds(15); +/** The shared deadline for one discovery: dial plus list. */ +export const DEFAULT_DISCOVER_TIMEOUT = Duration.seconds(15); // Teardown is best-effort and paid for by the request that performed discovery. // A remote transport may accept close and then never settle, so use the same @@ -160,6 +165,70 @@ export const discoverToolsFromInput = ( }), ); +/** Turn a connection failure into the discovery failure every caller of this + * module handles. A caller that takes its connection from the invocation pool + * meets the raw connector errors itself (the pool dials, this module only + * lists), so the mapping is shared. Preserves the handshake HTTP status and a + * connect-level timeout, which the liveness health check classifies on. */ +export const connectionFailureToDiscoveryError = ( + failure: McpConnectionError | McpOAuthReauthorizationRequired, +): McpToolDiscoveryError => { + const httpStatus = Predicate.isTagged(failure, "McpConnectionError") + ? failure.httpStatus + : undefined; + const reauthorizationRequired = Predicate.isTagged(failure, "McpOAuthReauthorizationRequired"); + const timedOut = + Predicate.isTagged(failure, "McpConnectionError") && failure.failureKind === "timeout"; + return new McpToolDiscoveryError({ + stage: "connect", + message: `Failed connecting to MCP server: ${failure.message}`, + ...(httpStatus !== undefined ? { httpStatus } : {}), + ...(reauthorizationRequired ? { reauthorizationRequired: true } : {}), + ...(timedOut ? { timedOut } : {}), + }); +}; + +/** Bound a discovery step with the shared deadline and the shared timeout + * error, so every path answers a wedged server identically — and every path is + * bounded: the pool's own dial has no deadline, so a pooled caller must wrap + * the whole lease. On timeout the lease releases and the pool closes the + * connection, so nothing is left behind. */ +export const withDiscoveryTimeout = ( + effect: Effect.Effect, + timeoutMs: number, +): Effect.Effect => + effect.pipe( + Effect.timeoutOrElse({ + duration: Duration.millis(timeoutMs), + orElse: () => + Effect.fail( + new McpToolDiscoveryError({ + stage: "connect", + message: `MCP discovery timed out after ${timeoutMs}ms`, + timedOut: true, + }), + ), + }), + ); + +/** The listing half of discovery, over a connection the caller owns. + * `discoverTools` dials, lists, and closes; a caller holding a pool lease must + * not close it. Same listing work and deadline, no teardown. */ +export const discoverToolsFromConnection = ( + connection: McpConnection, + timeoutMs: number = Duration.toMillis(DEFAULT_DISCOVER_TIMEOUT), +): Effect.Effect => + withDiscoveryTimeout( + Effect.gen(function* () { + // Decline elicitation explicitly; see the same call in `discoverTools`. + connection.client.setRequestHandler("elicitation/create", () => + Promise.resolve({ action: "decline" }), + ); + return yield* listAllTools(connection); + }), + timeoutMs, + ); + /** * Connect to an MCP server and discover all available tools. * Returns the parsed manifest containing server metadata and tool entries. @@ -184,68 +253,34 @@ export const discoverTools = ( connector: McpConnector, timeoutMs: number = Duration.toMillis(DEFAULT_DISCOVER_TIMEOUT), ): Effect.Effect => - Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - // Acquire connection - const connection = yield* restore( - connector.pipe( - Effect.mapError((failure) => { - // Preserve the handshake HTTP status (401/403 = auth wall) and a - // connect-level timeout so the liveness health check can classify - // structurally — dropping `failureKind: "timeout"` here is what - // made a timed-out handshake read as a generic probe failure. - const httpStatus = Predicate.isTagged(failure, "McpConnectionError") - ? failure.httpStatus - : undefined; - const reauthorizationRequired = Predicate.isTagged( - failure, - "McpOAuthReauthorizationRequired", - ); - const timedOut = - Predicate.isTagged(failure, "McpConnectionError") && - failure.failureKind === "timeout"; - return new McpToolDiscoveryError({ - stage: "connect", - message: `Failed connecting to MCP server: ${failure.message}`, - ...(httpStatus !== undefined ? { httpStatus } : {}), - ...(reauthorizationRequired ? { reauthorizationRequired: true } : {}), - ...(timedOut ? { timedOut } : {}), - }); - }), - ), - ); + withDiscoveryTimeout( + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + // Acquire connection + const connection = yield* restore( + connector.pipe(Effect.mapError(connectionFailureToDiscoveryError)), + ); - // The connection advertises the elicitation capability (connection.ts), - // so a server may elicit mid-listTools — the Codex desktop plugins do - // this for first-use approvals. Discovery has no user to route the - // request to (unlike the invoke path's bridge in invoke.ts), and a - // handler-less request would surface as a method-not-found error on the - // server's side of an otherwise healthy sync. Decline explicitly: the - // server completes the list with whatever it allows unapproved. - connection.client.setRequestHandler("elicitation/create", () => - Promise.resolve({ action: "decline" }), - ); + // The connection advertises the elicitation capability (connection.ts), + // so a server may elicit mid-listTools — the Codex desktop plugins do + // this for first-use approvals. Discovery has no user to route the + // request to (unlike the invoke path's bridge in invoke.ts), and a + // handler-less request would surface as a method-not-found error on the + // server's side of an otherwise healthy sync. Decline explicitly: the + // server completes the list with whatever it allows unapproved. + connection.client.setRequestHandler("elicitation/create", () => + Promise.resolve({ action: "decline" }), + ); - const manifest = yield* restore(listAllTools(connection)).pipe( - Effect.onExit(() => closeConnection(connection)), - ); + const manifest = yield* restore(listAllTools(connection)).pipe( + Effect.onExit(() => closeConnection(connection)), + ); - return manifest; - }), - ).pipe( - Effect.timeoutOrElse({ - duration: Duration.millis(timeoutMs), - orElse: () => - Effect.fail( - new McpToolDiscoveryError({ - stage: "connect", - message: `MCP discovery timed out after ${timeoutMs}ms`, - timedOut: true, - }), - ), - }), + return manifest; + }), + ), + timeoutMs, ); - const closeConnection = (connection: { readonly close: () => Promise; }): Effect.Effect => diff --git a/packages/plugins/mcp/src/sdk/mcp-liveness-second-spawn.test.ts b/packages/plugins/mcp/src/sdk/mcp-liveness-second-spawn.test.ts new file mode 100644 index 000000000..5ff0744ee --- /dev/null +++ b/packages/plugins/mcp/src/sdk/mcp-liveness-second-spawn.test.ts @@ -0,0 +1,113 @@ +// A liveness probe must reuse the invocation pool's connection instead of +// dialling a second one. For a stdio server a fresh dial starts a second child +// process, and the common local servers permit one instance only (Chrome +// DevTools MCP, Playwright MCP, `docker run -i`): the second child cannot +// start, so the probe reported a live, serving server as broken — once per page +// mount, because the UI re-probes every non-healthy verdict. +// +// The fixture refuses to start while a live process holds its lock, so two +// probes pass only when the second reuses the first one's child. +// +// `it.live`: this measures real child processes, so it needs the wall clock. + +import { existsSync, mkdtempSync, readFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; +import { FetchHttpClient } from "effect/unstable/http"; + +import { mcpPlugin } from "./plugin"; + +const fixture = fileURLToPath(new URL("./stdio-single-instance-test-server.ts", import.meta.url)); + +type Verdict = { readonly status: string; readonly detail?: string; readonly reason?: string }; + +type CheckHealth = (input: { + readonly ctx: { readonly httpClientLayer: typeof FetchHttpClient.layer }; + readonly credential: { + readonly config: unknown; + readonly values: Record; + readonly template: string | null; + readonly owner: string; + readonly connection: string; + readonly integration: string; + }; +}) => Effect.Effect; + +/** One plugin instance, so both probes share its connection pool — the same + * lifetime the pool has in a host. */ +const pluginCheckHealth = (): CheckHealth => { + const plugin = mcpPlugin({ dangerouslyAllowStdioMCP: true }); + const seam = (plugin as { readonly checkHealth?: CheckHealth }).checkHealth; + // The seam is part of the plugin contract. A build without it cannot run this + // scenario at all, so die rather than invent a verdict. + return seam ?? ((() => Effect.die("mcpPlugin no longer exposes checkHealth")) as CheckHealth); +}; + +const spawnedPids = (log: string): readonly number[] => + existsSync(log) + ? readFileSync(log, "utf8") + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line) => Number(line)) + : []; + +/** Stop every child the fixture logged. The pool keeps an idle child alive by + * design, and a test must not leave one behind. */ +const stopSpawned = (log: string): Effect.Effect => + Effect.sync(() => { + for (const pid of spawnedPids(log)) { + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: kill throws ESRCH when the child already exited, which is the desired state + try { + process.kill(pid, "SIGTERM"); + } catch { + // already gone + } + } + }); + +describe("MCP liveness probe against a single-instance local stdio server", () => { + it.live("reuses the pooled child, so a second probe starts no second process", () => + Effect.gen(function* () { + const dir = mkdtempSync(join(tmpdir(), "mcp-single-instance-")); + const lockFile = join(dir, "lock"); + const spawnLog = join(dir, "spawns"); + const config = { + transport: "stdio" as const, + command: "bun", + args: ["run", fixture, lockFile, spawnLog], + }; + const checkHealth = pluginCheckHealth(); + const credential = { + config, + values: {}, + template: null, + owner: "user", + connection: "main", + integration: "single_instance_mcp", + }; + const ctx = { httpClientLayer: FetchHttpClient.layer }; + + yield* Effect.acquireUseRelease( + Effect.void, + () => + Effect.gen(function* () { + const first = yield* checkHealth({ ctx, credential }); + expect(first.status, "the first probe dials and the server answers").toBe("healthy"); + expect(spawnedPids(spawnLog), "and it started exactly one child").toHaveLength(1); + + // The pooled child is alive and still holds the lock, so a second + // dial could not start. This probe passes only by reuse. + expect(existsSync(lockFile), "the first child is still running").toBe(true); + const second = yield* checkHealth({ ctx, credential }); + expect(second.status, "the second probe reads the same live server").toBe("healthy"); + expect(spawnedPids(spawnLog), "and it started no second child").toHaveLength(1); + }), + () => stopSpawned(spawnLog), + ); + }), + ); +}); diff --git a/packages/plugins/mcp/src/sdk/plugin.ts b/packages/plugins/mcp/src/sdk/plugin.ts index 079b89dce..cf7dd1230 100644 --- a/packages/plugins/mcp/src/sdk/plugin.ts +++ b/packages/plugins/mcp/src/sdk/plugin.ts @@ -1,4 +1,4 @@ -import { Effect, Layer, Option, Result, Schema } from "effect"; +import { Duration, Effect, Layer, Option, Predicate, Result, Schema } from "effect"; import type { HttpClient } from "effect/unstable/http"; import type { OAuthClientProvider } from "@modelcontextprotocol/client"; @@ -46,7 +46,13 @@ import { import type { CodexPluginEntry } from "./codex-plugins"; import { createMcpConnector, type ConnectorInput, type McpConnector } from "./connection"; import { createMcpConnectionPool } from "./connection-pool"; -import { discoverToolsFromInput } from "./discover"; +import { + connectionFailureToDiscoveryError, + DEFAULT_DISCOVER_TIMEOUT, + discoverToolsFromConnection, + discoverToolsFromInput, + withDiscoveryTimeout, +} from "./discover"; import { McpConnectionError, type McpConnectionFailureKind, @@ -1969,7 +1975,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { } } } - const connector = yield* buildConnectorInput( + const connectorInput = yield* buildConnectorInput( parsed, credential.values, credential.template === null ? null : String(credential.template), @@ -1977,7 +1983,49 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { options?.httpClientLayer ?? ctx.httpClientLayer, ); - return yield* discoverToolsFromInput(connector).pipe( + // Take the invocation pool's lease when this connection is poolable, so + // the probe reuses the session or child process tool calls already hold. + // A fresh dial starts a second child on a stdio server, and the common + // local servers permit one instance only (Chrome DevTools MCP, + // Playwright MCP, `docker run -i`) — the probe then failed a live, + // serving server, once per page mount. The key matches the invoke + // path's, which is what makes the lease hit the same entry. + const poolKey = isPoolableConnectorInput(connectorInput) + ? yield* connectionPoolKey( + connectorInput, + String(credential.template), + credential.values, + { + owner: String(credential.owner), + connection: String(credential.connection), + }, + ) + : undefined; + const discovery: Effect.Effect = + poolKey === undefined + ? Effect.asVoid(discoverToolsFromInput(connectorInput)) + : // The whole lease is bounded: the pool's own dial has no + // deadline. + withDiscoveryTimeout( + connectionPool.withConnection( + poolKey, + createMcpConnector(connectorInput), + (connection) => discoverToolsFromConnection(connection), + ), + Duration.toMillis(DEFAULT_DISCOVER_TIMEOUT), + ).pipe( + Effect.asVoid, + // The pool dials, so the raw connector failures surface here + // instead of inside `discoverTools`; map them through the same + // classification that path uses. + Effect.mapError((error) => + Predicate.isTagged(error, "McpToolDiscoveryError") + ? error + : connectionFailureToDiscoveryError(error), + ), + ); + + return yield* discovery.pipe( Effect.map( () => ({ status: "healthy" as const, checkedAt: Date.now() }) satisfies HealthCheckResult, diff --git a/packages/plugins/mcp/src/sdk/stdio-single-instance-test-server.ts b/packages/plugins/mcp/src/sdk/stdio-single-instance-test-server.ts new file mode 100644 index 000000000..a752fc15f --- /dev/null +++ b/packages/plugins/mcp/src/sdk/stdio-single-instance-test-server.ts @@ -0,0 +1,133 @@ +// Fixture for mcp-liveness-second-spawn.test.ts. A stdio MCP server that +// models a SINGLE-INSTANCE local server — the shape Chrome DevTools MCP, +// Playwright MCP and anything else that owns a browser, a debug port or a +// lock file has: a second concurrent process cannot start, and says so on +// stderr before exiting non-zero. +// +// argv[2] is the lock file, argv[3] a spawn log the test reads to count how +// many child processes a code path created. Every spawn appends its PID, so +// "did the health probe reuse a connection or start a new process?" is +// answerable from the file alone. + +import { appendFileSync, existsSync, readFileSync, unlinkSync, writeFileSync } from "node:fs"; + +const lockFile = process.argv[2]; +const spawnLog = process.argv[3]; +if (lockFile === undefined || spawnLog === undefined) { + process.stderr.write("usage: stdio-single-instance-test-server.ts \n"); + process.exit(2); +} + +appendFileSync(spawnLog, `${process.pid}\n`); + +const isAlive = (pid: number): boolean => { + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: standalone non-Effect fixture process; kill(pid, 0) reports "gone" only by throwing + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +}; + +if (existsSync(lockFile)) { + const holder = Number(readFileSync(lockFile, "utf8").trim()); + if (Number.isFinite(holder) && holder !== process.pid && isAlive(holder)) { + // Exactly what a single-instance local server does when something already + // owns the resource: refuse to start and exit non-zero. + process.stderr.write( + `single-instance server: another instance (${holder}) is already running\n`, + ); + process.exit(1); + } +} + +writeFileSync(lockFile, String(process.pid)); + +const release = (): void => { + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: fixture teardown must not throw on an already-removed lock + try { + if (existsSync(lockFile) && readFileSync(lockFile, "utf8").trim() === String(process.pid)) { + unlinkSync(lockFile); + } + } catch { + // already gone + } +}; +process.on("exit", release); +process.on("SIGTERM", () => { + release(); + process.exit(0); +}); + +const respond = (message: object): void => { + process.stdout.write(`${JSON.stringify(message)}\n`); +}; + +const handle = (line: string): void => { + if (!line.trim()) return; + let request: { + id?: number; + method?: string; + params?: { protocolVersion?: string; name?: string; arguments?: Record }; + }; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: standalone fixture process; a malformed frame is dropped like a real server would + try { + // oxlint-disable-next-line executor/no-json-parse -- boundary: hand-rolled JSON-RPC framing is the fixture's entire purpose + request = JSON.parse(line); + } catch { + return; + } + if (request.method === "initialize") { + respond({ + jsonrpc: "2.0", + id: request.id, + result: { + protocolVersion: request.params?.protocolVersion, + capabilities: { tools: {} }, + serverInfo: { name: "stdio-single-instance-test-server", version: "0.0.0" }, + }, + }); + } else if (request.method === "tools/list") { + respond({ + jsonrpc: "2.0", + id: request.id, + result: { + tools: [ + { + name: "whoami", + description: "whoami", + inputSchema: { type: "object", properties: {}, additionalProperties: false }, + }, + ], + }, + }); + } else if (request.method === "tools/call") { + respond({ + jsonrpc: "2.0", + id: request.id, + result: { + content: [{ type: "text", text: `served by ${process.pid}` }], + isError: false, + }, + }); + } else if (request.id !== undefined) { + respond({ jsonrpc: "2.0", id: request.id, result: {} }); + } +}; + +let buffer = ""; +process.stdin.setEncoding("utf8"); +process.stdin.on("data", (chunk: string) => { + buffer += chunk; + let newline = buffer.indexOf("\n"); + while (newline >= 0) { + handle(buffer.slice(0, newline)); + buffer = buffer.slice(newline + 1); + newline = buffer.indexOf("\n"); + } +}); +process.stdin.on("end", () => { + release(); + process.exit(0); +}); diff --git a/packages/plugins/openapi/src/sdk/backing.ts b/packages/plugins/openapi/src/sdk/backing.ts index 82cd6dcdb..d336556af 100644 --- a/packages/plugins/openapi/src/sdk/backing.ts +++ b/packages/plugins/openapi/src/sdk/backing.ts @@ -1013,8 +1013,14 @@ export const checkHealthOpenApi = (input: { } // Body-aware: a configuration 403 (Google accessNotConfigured / - // SERVICE_DISABLED) reads misconfigured, not expired. - const status = classifyProbeResponse(probe.result.status, probe.result.error); + // SERVICE_DISABLED) reads misconfigured, and a scope shortfall reads + // degraded, not expired — both authenticated, and neither is fixed by a + // reconnect. + const status = classifyProbeResponse( + probe.result.status, + probe.result.error, + probe.result.headers, + ); const rawIdentity = status === "healthy" ? extractIdentity(probe.result.data, spec.identityField) : undefined; // The identity is read straight off the raw body, so unlike the sample it