History

sdk: dedup concurrent connection refreshes and surface reauth-required

OpenAPI tool invocations call `ctx.connections.accessToken(id)` to pick up a fresh access token, and the SDK already delegated refresh to a per-provider handler. Two gaps remained: 1. No concurrency dedup — ten parallel invokes against an expired token fired ten `grant_type=refresh_token` POSTs. Now `connectionsAccessToken` registers a `Deferred` under a semaphore; the first caller leads the refresh, every other concurrent caller awaits the same Deferred, and the slot is cleared in `onExit` regardless of outcome. 2. `invalid_grant` from the AS collapsed into a generic `ConnectionRefreshError`. Providers can now set `reauthRequired: true` and the SDK translates it to a new `ConnectionReauthRequiredError` tagged error so UI bridges can prompt sign-in instead of silently retrying. `OAuth2Error` now carries the RFC 6749 §5.2 error code (`invalid_grant`, `invalid_client`, `unauthorized_client`) parsed from the token endpoint response. The openapi connection provider maps those to `reauthRequired`. A stored authorizationCode connection with no refresh token also surfaces as reauth-required, since we literally cannot hit the token endpoint. Tests: `connections.test.ts` gains three cases covering the dedup path (Deferred-gated fake provider proves one `refresh` call across five concurrent `accessToken` invokes), the reauth translation, and the transient failure path that must stay `ConnectionRefreshError`. A new `oauth-refresh.test.ts` exercises the same three scenarios end-to-end through the openapi plugin against a live HttpApi server with a mock token endpoint.

Rhys Sullivan committed 6 months ago

Commit b94faa51 parent e9fad479 files · +640 −48Browse files

Unified · Split

9 files
  1. packages/core/sdk/src/connections.test.ts +0 −0
  2. packages/core/sdk/src/connections.ts +9 −0
  3. packages/core/sdk/src/errors.ts +20 −1
  4. packages/core/sdk/src/executor.ts +148 −45
  5. packages/core/sdk/src/index.ts +1 −0
  6. packages/core/sdk/src/plugin.ts +2 −0
  7. packages/plugins/oauth2/src/index.ts +19 −3
  8. packages/plugins/openapi/src/sdk/oauth-refresh.test.ts +422 −0
  9. packages/plugins/openapi/src/sdk/plugin.ts +20 −0
packages/core/sdk/src/connections.test.ts+0−0

Binary file not shown.

packages/core/sdk/src/connections.ts+9−0
@@ -100,6 +100,15 @@ export class ConnectionRefreshError extends Data.TaggedError( · 99 unmodified lines
100)<{100)<{
101 readonly connectionId: ConnectionId;101 readonly connectionId: ConnectionId;
102 readonly message: string;102 readonly message: string;
103 /**
104 * Set by providers when the refresh failed in a way that the stored
105 * refresh token cannot recover from (RFC 6749 §5.2 `invalid_grant`
106 * — the AS has revoked the grant, the user changed their password,
107 * the refresh token rotated out from under us, ...). The SDK
108 * translates this into a `ConnectionReauthRequiredError` so callers
109 * can prompt the user to sign in again instead of silently retrying.
110 */
111 readonly reauthRequired?: boolean;
103 readonly cause?: unknown;112 readonly cause?: unknown;
104}> {}113}> {}
105 114
packages/core/sdk/src/errors.ts+20−1
@@ -111,6 +111,24 @@ export class ConnectionRefreshNotSupportedError extends Schema.TaggedError<Conne · 110 unmodified lines
111 },111 },
112) {}112) {}
113 113
114/**
115 * Raised by `connections.accessToken(id)` when the provider's refresh
116 * handler reported that the stored refresh token is permanently
117 * invalid (RFC 6749 §5.2 `invalid_grant` and friends). The caller —
118 * typically a tool invocation — surfaces this so the UI can prompt the
119 * user to sign in again. Distinct from `ConnectionRefreshError` so
120 * "the network flaked, retry later" and "the grant is dead, re-auth"
121 * don't collapse into one error tag at the plugin boundary.
122 */
123export class ConnectionReauthRequiredError extends Schema.TaggedError<ConnectionReauthRequiredError>()(
124 "ConnectionReauthRequiredError",
125 {
126 connectionId: ConnectionId,
127 provider: Schema.String,
128 message: Schema.String,
129 },
130) {}
131
114// ---------------------------------------------------------------------------132// ---------------------------------------------------------------------------
115// Union type for convenience in signatures.133// Union type for convenience in signatures.
116// ---------------------------------------------------------------------------134// ---------------------------------------------------------------------------
@@ -127,4 +145,5 @@ export type ExecutorError = · 10 unmodified lines
127 | SecretOwnedByConnectionError145 | SecretOwnedByConnectionError
128 | ConnectionNotFoundError146 | ConnectionNotFoundError
129 | ConnectionProviderNotRegisteredError147 | ConnectionProviderNotRegisteredError
130 | ConnectionRefreshNotSupportedError;148 | ConnectionRefreshNotSupportedError
149 | ConnectionReauthRequiredError;
packages/core/sdk/src/executor.ts+148−45
@@ -1,4 +1,4 @@
1import { Effect, FiberRef, Option, Schema } from "effect";1import { Deferred, Effect, FiberRef, Option, Schema } from "effect";
2import {2import {
3 StorageError,3 StorageError,
4 typedAdapter,4 typedAdapter,
@@ -41,6 +41,7 @@ import { · 36 unmodified lines
41import {41import {
42 ConnectionNotFoundError,42 ConnectionNotFoundError,
43 ConnectionProviderNotRegisteredError,43 ConnectionProviderNotRegisteredError,
44 ConnectionReauthRequiredError,
44 ConnectionRefreshNotSupportedError,45 ConnectionRefreshNotSupportedError,
45 NoHandlerError,46 NoHandlerError,
46 PluginNotLoadedError,47 PluginNotLoadedError,
@@ -219,6 +220,7 @@ export type Executor<TPlugins extends readonly AnyPlugin[] = []> = { · 172 unmodified lines
219 | ConnectionNotFoundError220 | ConnectionNotFoundError
220 | ConnectionProviderNotRegisteredError221 | ConnectionProviderNotRegisteredError
221 | ConnectionRefreshNotSupportedError222 | ConnectionRefreshNotSupportedError
223 | ConnectionReauthRequiredError
222 | ConnectionRefreshError224 | ConnectionRefreshError
223 | StorageFailure225 | StorageFailure
224 >;226 >;
@@ -603,6 +605,25 @@ export const createExecutor = < · 378 unmodified lines
603 // Connection providers keyed by `provider.key` — drive the refresh605 // Connection providers keyed by `provider.key` — drive the refresh
604 // lifecycle for connection-owned tokens.606 // lifecycle for connection-owned tokens.
605 const connectionProviders = new Map<string, ConnectionProvider>();607 const connectionProviders = new Map<string, ConnectionProvider>();
608 // In-flight refresh dedup. `connectionsAccessToken` stamps a
609 // `Deferred` here before calling the provider's `refresh`; parallel
610 // callers that walk in while a refresh is still running observe
611 // the same Deferred and await its resolution instead of hitting
612 // the AS a second time. The map is mutated under a semaphore so
613 // check-or-register is atomic under fiber interleavings.
614 const refreshInFlight = new Map<
615 string,
616 Deferred.Deferred<
617 string,
618 | ConnectionNotFoundError
619 | ConnectionProviderNotRegisteredError
620 | ConnectionRefreshNotSupportedError
621 | ConnectionReauthRequiredError
622 | ConnectionRefreshError
623 | StorageFailure
624 >
625 >();
626 const refreshInFlightLock = Effect.unsafeMakeSemaphore(1);
606 const extensions: Record<string, object> = {};627 const extensions: Record<string, object> = {};
607 628
608 // ------------------------------------------------------------------629 // ------------------------------------------------------------------
@@ -1389,46 +1410,26 @@ export const createExecutor = < · 780 unmodified lines
1389 );1410 );
1390 });1411 });
1391 1412
1392 // accessToken(id) — the single surface plugins use at invoke time.1413 // Typed error union that `connectionsAccessToken` and every helper
1393 // Resolves the backing secret, checks expiry, calls the provider's1414 // that participates in a refresh returns. Pulled out into a type
1394 // refresh handler if we're inside the skew window. New tokens are1415 // alias because it has to match the Deferred's channel exactly —
1395 // written back through the same provider and the connection row is1416 // otherwise concurrent waiters and the leader diverge on the error
1396 // patched with the new expiry.1417 // type.
1397 const connectionsAccessToken = (1418 type AccessTokenError =
1398 id: string,
1399 ): Effect.Effect<
1400 string,
1401 | ConnectionNotFoundError1419 | ConnectionNotFoundError
1402 | ConnectionProviderNotRegisteredError1420 | ConnectionProviderNotRegisteredError
1403 | ConnectionRefreshNotSupportedError1421 | ConnectionRefreshNotSupportedError
1422 | ConnectionReauthRequiredError
1404 | ConnectionRefreshError1423 | ConnectionRefreshError
1405 | StorageFailure1424 | StorageFailure;
1406 > =>1425
1426 // The actual work of a single refresh cycle, factored out so the
1427 // concurrency gate (`connectionsAccessToken`) stays readable. Runs
1428 // for the fiber that wins the `refreshInFlight` race.
1429 const performRefresh = (
1430 ref: ConnectionRef,
1431 ): Effect.Effect<string, AccessTokenError> =>
1407 Effect.gen(function* () {1432 Effect.gen(function* () {
1408 const row = yield* findInnermostConnectionRow(id);
1409 if (!row) {
1410 return yield* Effect.fail(
1411 new ConnectionNotFoundError({
1412 connectionId: ConnectionId.make(id),
1413 }),
1414 );
1415 }
1416 const ref = rowToConnection(row);
1417 const now = Date.now();
1418 const needsRefresh =
1419 ref.expiresAt !== null &&
1420 ref.expiresAt - CONNECTION_REFRESH_SKEW_MS <= now;
1421
1422 if (!needsRefresh) {
1423 const current = yield* secretsGet(
1424 ref.accessTokenSecretId as unknown as string,
1425 );
1426 if (current !== null) return current;
1427 // Fall through to refresh if the stored token vanished — a
1428 // genuinely-missing secret with no way to refresh is a
1429 // hard-failure, same behavior as if `expires_at` had passed.
1430 }
1431
1432 const provider = connectionProviders.get(ref.provider);1433 const provider = connectionProviders.get(ref.provider);
1433 if (!provider) {1434 if (!provider) {
1434 return yield* Effect.fail(1435 return yield* Effect.fail(
@@ -1451,15 +1452,36 @@ export const createExecutor = < · 16 unmodified lines
1451 ? yield* secretsGet(ref.refreshTokenSecretId as unknown as string)1452 ? yield* secretsGet(ref.refreshTokenSecretId as unknown as string)
1452 : null;1453 : null;
1453 1454
1454 const result = yield* provider.refresh({1455 // RFC 6749 §5.2 `invalid_grant` (and anything else the
1455 connectionId: ref.id,1456 // provider tags with `reauthRequired`) is terminal — the
1456 scopeId: ref.scopeId,1457 // stored refresh token can't recover. Translate into the
1457 kind: ref.kind,1458 // caller-visible "re-authenticate" error so the UI can
1458 identityLabel: ref.identityLabel,1459 // prompt sign-in instead of silently retrying.
1459 refreshToken: refreshTokenValue,1460 const rawResult = yield* Effect.either(
1460 providerState: ref.providerState,1461 provider.refresh({
1461 oauthScope: ref.oauthScope,1462 connectionId: ref.id,
1462 });1463 scopeId: ref.scopeId,
1464 kind: ref.kind,
1465 identityLabel: ref.identityLabel,
1466 refreshToken: refreshTokenValue,
1467 providerState: ref.providerState,
1468 oauthScope: ref.oauthScope,
1469 }),
1470 );
1471 if (rawResult._tag === "Left") {
1472 const err = rawResult.left;
1473 if (err.reauthRequired) {
1474 return yield* Effect.fail(
1475 new ConnectionReauthRequiredError({
1476 connectionId: err.connectionId,
1477 provider: ref.provider,
1478 message: err.message,
1479 }),
1480 );
1481 }
1482 return yield* Effect.fail(err);
1483 }
1484 const result = rawResult.right;
1463 1485
1464 yield* connectionsUpdateTokens({1486 yield* connectionsUpdateTokens({
1465 id: ref.id,1487 id: ref.id,
@@ -1473,6 +1495,87 @@ export const createExecutor = < · 7 unmodified lines
1473 return result.accessToken;1495 return result.accessToken;
1474 });1496 });
1475 1497
1498 // accessToken(id) — the single surface plugins use at invoke time.
1499 // Resolves the backing secret, checks expiry, calls the provider's
1500 // refresh handler if we're inside the skew window. New tokens are
1501 // written back through the same provider and the connection row is
1502 // patched with the new expiry.
1503 //
1504 // Concurrent invokes on an expired token all share one refresh.
1505 // The fiber that wins the `refreshInFlightLock` race registers a
1506 // Deferred and performs the refresh; every other concurrent caller
1507 // observes the Deferred and awaits its completion. The Deferred is
1508 // pulled out of the map before the refresh result resolves so
1509 // later invokes don't reuse a completed slot.
1510 const connectionsAccessToken = (
1511 id: string,
1512 ): Effect.Effect<string, AccessTokenError> =>
1513 Effect.gen(function* () {
1514 const row = yield* findInnermostConnectionRow(id);
1515 if (!row) {
1516 return yield* Effect.fail(
1517 new ConnectionNotFoundError({
1518 connectionId: ConnectionId.make(id),
1519 }),
1520 );
1521 }
1522 const ref = rowToConnection(row);
1523 const now = Date.now();
1524 const needsRefresh =
1525 ref.expiresAt !== null &&
1526 ref.expiresAt - CONNECTION_REFRESH_SKEW_MS <= now;
1527
1528 if (!needsRefresh) {
1529 const current = yield* secretsGet(
1530 ref.accessTokenSecretId as unknown as string,
1531 );
1532 if (current !== null) return current;
1533 // Fall through to refresh if the stored token vanished — a
1534 // genuinely-missing secret with no way to refresh is a
1535 // hard-failure, same behavior as if `expires_at` had passed.
1536 }
1537
1538 // Concurrency gate. `action` either returns the fresh access
1539 // token (this fiber did the refresh) or the already-running
1540 // Deferred that another fiber stamped into the map (this fiber
1541 // piggybacks on their refresh).
1542 const action = yield* refreshInFlightLock.withPermits(1)(
1543 Effect.gen(function* () {
1544 const existing = refreshInFlight.get(id);
1545 if (existing) {
1546 return {
1547 kind: "await" as const,
1548 deferred: existing,
1549 };
1550 }
1551 const deferred = yield* Deferred.make<string, AccessTokenError>();
1552 refreshInFlight.set(id, deferred);
1553 return { kind: "lead" as const, deferred };
1554 }),
1555 );
1556
1557 if (action.kind === "await") {
1558 return yield* action.deferred;
1559 }
1560
1561 // Leader path: run the refresh, pipe the outcome into the
1562 // Deferred (so waiters wake up), and then clear the map slot
1563 // regardless of success or failure. Keeping clear+complete in
1564 // the same `onExit` avoids a window where a late waiter sees
1565 // the Deferred after it's been completed but before it's been
1566 // evicted.
1567 return yield* performRefresh(ref).pipe(
1568 Effect.onExit((exit) =>
1569 refreshInFlightLock.withPermits(1)(
1570 Effect.gen(function* () {
1571 refreshInFlight.delete(id);
1572 yield* Deferred.done(action.deferred, exit);
1573 }),
1574 ),
1575 ),
1576 );
1577 });
1578
1476 const connectionsListForCtx = () => connectionsList();1579 const connectionsListForCtx = () => connectionsList();
1477 1580
1478 // ------------------------------------------------------------------1581 // ------------------------------------------------------------------
packages/core/sdk/src/index.ts+1−0
@@ -42,6 +42,7 @@ export { · 41 unmodified lines
42 ConnectionNotFoundError,42 ConnectionNotFoundError,
43 ConnectionProviderNotRegisteredError,43 ConnectionProviderNotRegisteredError,
44 ConnectionRefreshNotSupportedError,44 ConnectionRefreshNotSupportedError,
45 ConnectionReauthRequiredError,
45 type ExecutorError,46 type ExecutorError,
46} from "./errors";47} from "./errors";
47 48
packages/core/sdk/src/plugin.ts+2−0
@@ -30,6 +30,7 @@ import type { · 29 unmodified lines
30import type {30import type {
31 ConnectionNotFoundError,31 ConnectionNotFoundError,
32 ConnectionProviderNotRegisteredError,32 ConnectionProviderNotRegisteredError,
33 ConnectionReauthRequiredError,
33 ConnectionRefreshNotSupportedError,34 ConnectionRefreshNotSupportedError,
34 SecretOwnedByConnectionError,35 SecretOwnedByConnectionError,
35} from "./errors";36} from "./errors";
@@ -193,6 +194,7 @@ export interface PluginCtx<TStore = unknown> { · 157 unmodified lines
193 | ConnectionNotFoundError194 | ConnectionNotFoundError
194 | ConnectionProviderNotRegisteredError195 | ConnectionProviderNotRegisteredError
195 | ConnectionRefreshNotSupportedError196 | ConnectionRefreshNotSupportedError
197 | ConnectionReauthRequiredError
196 | ConnectionRefreshError198 | ConnectionRefreshError
197 | StorageFailure199 | StorageFailure
198 >;200 >;
packages/plugins/oauth2/src/index.ts+19−3
@@ -22,6 +22,13 @@ import { Data, Effect, ParseResult, Schema } from "effect"; · 21 unmodified lines
22 22
23export class OAuth2Error extends Data.TaggedError("OAuth2Error")<{23export class OAuth2Error extends Data.TaggedError("OAuth2Error")<{
24 readonly message: string;24 readonly message: string;
25 /**
26 * RFC 6749 §5.2 error code, when the token endpoint returned one
27 * (`invalid_grant`, `invalid_client`, `unauthorized_client`, ...).
28 * Callers use this to distinguish terminal failures (a refresh token
29 * the AS no longer honours → re-auth required) from transient ones.
30 */
31 readonly error?: string;
25 readonly cause?: unknown;32 readonly cause?: unknown;
26}> {}33}> {}
27 34
@@ -104,8 +111,11 @@ export const buildAuthorizationUrl = (input: BuildAuthorizationUrlInput): string · 76 unmodified lines
104// Token endpoint response parsing111// Token endpoint response parsing
105// ---------------------------------------------------------------------------112// ---------------------------------------------------------------------------
106 113
107const oauth2Error = (message: string, cause?: unknown): OAuth2Error =>114const oauth2Error = (
108 new OAuth2Error({ message, cause });115 message: string,
116 cause?: unknown,
117 error?: string,
118): OAuth2Error => new OAuth2Error({ message, cause, error });
109 119
110// ---------------------------------------------------------------------------120// ---------------------------------------------------------------------------
111// Schemas121// Schemas
@@ -213,7 +223,13 @@ export const decodeTokenResponse = ( · 101 unmodified lines
213 );223 );
214 const description =224 const description =
215 envelope.error_description ?? envelope.error ?? `status ${response.status}`;225 envelope.error_description ?? envelope.error ?? `status ${response.status}`;
216 return yield* Effect.fail(oauth2Error(`OAuth token exchange failed: ${description}`));226 return yield* Effect.fail(
227 oauth2Error(
228 `OAuth token exchange failed: ${description}`,
229 undefined,
230 envelope.error,
231 ),
232 );
217 }233 }
218 234
219 return yield* decodeSuccessBody(record).pipe(235 return yield* decodeSuccessBody(record).pipe(
packages/plugins/openapi/src/sdk/oauth-refresh.test.ts+422−0
@@ -0,0 +1,422 @@
1// ---------------------------------------------------------------------------
2// End-to-end refresh behaviour for the OpenAPI plugin's oauth2 connection
3// provider.
4//
5// The existing `multi-scope-oauth.test.ts` covers sign-in isolation; this
6// file focuses on RFC 6749 §6 refresh behaviour at the plugin boundary:
7//
8// 1. An expired access_token is refreshed transparently before invoke.
9// 2. Concurrent invokes collapse to a single `grant_type=refresh_token`
10// POST — the SDK's dedup applies to the plugin's provider.
11// 3. `invalid_grant` from the token endpoint surfaces as
12// `ConnectionReauthRequiredError` so the UI can prompt sign-in.
13// ---------------------------------------------------------------------------
14
15import { afterEach } from "vitest";
16import { expect, layer } from "@effect/vitest";
17import { Effect, Layer, Schema } from "effect";
18import {
19 HttpApi,
20 HttpApiBuilder,
21 HttpApiEndpoint,
22 HttpApiGroup,
23 HttpClient,
24 HttpServerRequest,
25 OpenApi,
26} from "@effect/platform";
27import { NodeHttpServer } from "@effect/platform-node";
28
29import {
30 ConnectionId,
31 ConnectionReauthRequiredError,
32 CreateConnectionInput,
33 ScopeId,
34 SecretId,
35 Scope,
36 SetSecretInput,
37 TokenMaterial,
38 collectSchemas,
39 createExecutor,
40 definePlugin,
41 makeInMemoryBlobStore,
42 type InvokeOptions,
43 type SecretProvider,
44} from "@executor/sdk";
45import { makeMemoryAdapter } from "@executor/storage-core/testing/memory";
46
47import { openApiPlugin } from "./plugin";
48import { OAuth2Auth } from "./types";
49
50const autoApprove: InvokeOptions = { onElicitation: "accept-all" };
51
52// ---------------------------------------------------------------------------
53// Test API — one endpoint that echoes the Authorization header so we can
54// prove which access token was in flight at invoke time.
55// ---------------------------------------------------------------------------
56
57class EchoHeaders extends Schema.Class<EchoHeaders>("EchoHeaders")({
58 authorization: Schema.optional(Schema.String),
59}) {}
60
61const ItemsGroup = HttpApiGroup.make("items").add(
62 HttpApiEndpoint.get("echoHeaders", "/echo-headers").addSuccess(EchoHeaders),
63);
64
65const TestApi = HttpApi.make("testApi").add(ItemsGroup);
66const specJson = JSON.stringify(OpenApi.fromApi(TestApi));
67
68const ItemsGroupLive = HttpApiBuilder.group(TestApi, "items", (handlers) =>
69 handlers.handle("echoHeaders", () =>
70 Effect.gen(function* () {
71 const req = yield* HttpServerRequest.HttpServerRequest;
72 return new EchoHeaders({
73 authorization: req.headers["authorization"],
74 });
75 }),
76 ),
77);
78
79const ApiLive = HttpApiBuilder.api(TestApi).pipe(Layer.provide(ItemsGroupLive));
80
81const TestLayer = HttpApiBuilder.serve().pipe(
82 Layer.provide(ApiLive),
83 Layer.provideMerge(NodeHttpServer.layerTest),
84);
85
86// ---------------------------------------------------------------------------
87// Token-endpoint mock. Callers supply a handler that sees the parsed body
88// (grant_type, refresh_token, ...) and returns either an RFC 6749 success
89// response or an error envelope. `calls` records every hit for assertions.
90// ---------------------------------------------------------------------------
91
92const originalFetch = globalThis.fetch;
93
94type TokenCall = {
95 readonly body: URLSearchParams;
96};
97
98const mockTokenFetch = (
99 handler: (body: URLSearchParams) => Effect.Effect<Response> | Promise<Response>,
100) => {
101 const calls: TokenCall[] = [];
102 globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => {
103 const bodyText =
104 init?.body instanceof URLSearchParams
105 ? init.body.toString()
106 : typeof init?.body === "string"
107 ? init.body
108 : "";
109 const body = new URLSearchParams(bodyText);
110 calls.push({ body });
111 const out = handler(body);
112 if (out && typeof (out as Promise<Response>).then === "function") {
113 return await (out as Promise<Response>);
114 }
115 return await Effect.runPromise(out as Effect.Effect<Response>);
116 }) as unknown as typeof fetch;
117 return { calls };
118};
119
120afterEach(() => {
121 globalThis.fetch = originalFetch;
122});
123
124// ---------------------------------------------------------------------------
125// Fixture builder. Wires up a single-scope executor with an in-memory
126// secrets provider, the openApi plugin pointed at a live HttpClient, and
127// seeds an expired oauth2 Connection + source pointing at that server.
128// ---------------------------------------------------------------------------
129
130const makeExecutor = () =>
131 Effect.gen(function* () {
132 const secretStore = new Map<string, string>();
133 const keyOf = (scope: string, id: string) => `${scope} ${id}`;
134 const memoryProvider: SecretProvider = {
135 key: "memory",
136 writable: true,
137 get: (id, scope) =>
138 Effect.sync(() => secretStore.get(keyOf(scope, id)) ?? null),
139 set: (id, value, scope) =>
140 Effect.sync(() => {
141 secretStore.set(keyOf(scope, id), value);
142 }),
143 delete: (id, scope) =>
144 Effect.sync(() => secretStore.delete(keyOf(scope, id))),
145 };
146 const memorySecretsPlugin = definePlugin(() => ({
147 id: "memory-secrets" as const,
148 storage: () => ({}),
149 secretProviders: [memoryProvider],
150 }));
151
152 const httpClient = yield* HttpClient.HttpClient;
153 const clientLayer = Layer.succeed(HttpClient.HttpClient, httpClient);
154 const plugins = [
155 openApiPlugin({ httpClientLayer: clientLayer }),
156 memorySecretsPlugin(),
157 ] as const;
158
159 const schema = collectSchemas(plugins);
160 const adapter = makeMemoryAdapter({ schema });
161 const blobs = makeInMemoryBlobStore();
162
163 const scopeId = ScopeId.make("test-scope");
164 const scope = new Scope({
165 id: scopeId,
166 name: "test",
167 createdAt: new Date(),
168 });
169 const executor = yield* createExecutor({
170 scopes: [scope],
171 adapter,
172 blobs,
173 plugins,
174 });
175
176 // Seed client id + secret in the executor scope so the openapi
177 // provider's refresh can resolve them.
178 yield* executor.secrets.set(
179 new SetSecretInput({
180 id: SecretId.make("client_id"),
181 scope: scopeId,
182 name: "Client ID",
183 value: "abc",
184 }),
185 );
186 yield* executor.secrets.set(
187 new SetSecretInput({
188 id: SecretId.make("client_secret"),
189 scope: scopeId,
190 name: "Client Secret",
191 value: "shhh",
192 }),
193 );
194
195 return { executor, scopeId };
196 });
197
198type ExecutorValue = Effect.Effect.Success<
199 ReturnType<typeof makeExecutor>
200>["executor"];
201
202// Seed an authorizationCode Connection with an already-expired access
203// token and a stored refresh token. The test's mock token endpoint
204// decides what comes back on `grant_type=refresh_token`.
205const seedExpiredConnection = (
206 executor: ExecutorValue,
207 scopeId: ScopeId,
208 connectionId: string,
209) =>
210 Effect.gen(function* () {
211 yield* executor.connections.create(
212 new CreateConnectionInput({
213 id: ConnectionId.make(connectionId),
214 scope: scopeId,
215 provider: "openapi:oauth2",
216 kind: "user",
217 identityLabel: "Alice",
218 accessToken: new TokenMaterial({
219 secretId: SecretId.make(`${connectionId}.access_token`),
220 name: "Access",
221 value: "expired-access-v1",
222 }),
223 refreshToken: new TokenMaterial({
224 secretId: SecretId.make(`${connectionId}.refresh_token`),
225 name: "Refresh",
226 value: "refresh-v1",
227 }),
228 expiresAt: Date.now() - 10_000,
229 oauthScope: "read",
230 providerState: {
231 flow: "authorizationCode",
232 tokenUrl: "https://token.example.com/token",
233 clientIdSecretId: "client_id",
234 clientSecretSecretId: "client_secret",
235 scopes: ["read"],
236 },
237 }),
238 );
239 return new OAuth2Auth({
240 kind: "oauth2",
241 connectionId,
242 securitySchemeName: "oauth2",
243 flow: "authorizationCode",
244 tokenUrl: "https://token.example.com/token",
245 authorizationUrl: "https://auth.example.com/authorize",
246 clientIdSecretId: "client_id",
247 clientSecretSecretId: "client_secret",
248 scopes: ["read"],
249 });
250 });
251
252// ---------------------------------------------------------------------------
253// Tests
254// ---------------------------------------------------------------------------
255
256layer(TestLayer)("OpenAPI oauth refresh", (it) => {
257 it.effect(
258 "expired access_token is refreshed via grant_type=refresh_token before invoke",
259 () =>
260 Effect.gen(function* () {
261 const { executor, scopeId } = yield* makeExecutor();
262 const { calls } = mockTokenFetch(
263 () =>
264 Effect.succeed(
265 new Response(
266 JSON.stringify({
267 access_token: "fresh-access-v2",
268 token_type: "Bearer",
269 refresh_token: "refresh-v2",
270 expires_in: 3600,
271 }),
272 { status: 200, headers: { "content-type": "application/json" } },
273 ),
274 ),
275 );
276
277 const auth = yield* seedExpiredConnection(
278 executor,
279 scopeId,
280 "conn-refresh-ok",
281 );
282
283 yield* executor.openapi.addSpec({
284 spec: specJson,
285 scope: scopeId as unknown as string,
286 namespace: "petstore",
287 baseUrl: "",
288 oauth2: auth,
289 });
290
291 const result = (yield* executor.tools.invoke(
292 "petstore.items.echoHeaders",
293 {},
294 autoApprove,
295 )) as { data: { authorization?: string } | null; error: unknown };
296
297 expect(result.error).toBeNull();
298 // Proves the refresh landed: invoke carried the fresh token,
299 // not the expired one we seeded.
300 expect(result.data?.authorization).toBe("Bearer fresh-access-v2");
301 expect(calls).toHaveLength(1);
302 expect(calls[0]!.body.get("grant_type")).toBe("refresh_token");
303 expect(calls[0]!.body.get("refresh_token")).toBe("refresh-v1");
304
305 // Connection row is patched with the new expiry so the next
306 // invoke in-window doesn't trip a second refresh.
307 const conn = yield* executor.connections.get("conn-refresh-ok");
308 expect(conn).not.toBeNull();
309 expect(conn!.expiresAt).not.toBeNull();
310 expect(conn!.expiresAt!).toBeGreaterThan(Date.now() + 3_000_000);
311 }),
312 );
313
314 it.effect(
315 "concurrent invokes with an expired token issue exactly one refresh",
316 () =>
317 Effect.gen(function* () {
318 const { executor, scopeId } = yield* makeExecutor();
319 const { calls } = mockTokenFetch(
320 () =>
321 Effect.succeed(
322 new Response(
323 JSON.stringify({
324 access_token: "fresh-access-v2",
325 token_type: "Bearer",
326 refresh_token: "refresh-v2",
327 expires_in: 3600,
328 }),
329 { status: 200, headers: { "content-type": "application/json" } },
330 ),
331 ),
332 );
333
334 const auth = yield* seedExpiredConnection(
335 executor,
336 scopeId,
337 "conn-refresh-concurrent",
338 );
339
340 yield* executor.openapi.addSpec({
341 spec: specJson,
342 scope: scopeId as unknown as string,
343 namespace: "petstore",
344 baseUrl: "",
345 oauth2: auth,
346 });
347
348 const invokes = yield* Effect.all(
349 [1, 2, 3, 4, 5].map(() =>
350 executor.tools.invoke(
351 "petstore.items.echoHeaders",
352 {},
353 autoApprove,
354 ),
355 ),
356 { concurrency: "unbounded" },
357 );
358
359 for (const r of invokes) {
360 const res = r as {
361 data: { authorization?: string } | null;
362 error: unknown;
363 };
364 expect(res.error).toBeNull();
365 expect(res.data?.authorization).toBe("Bearer fresh-access-v2");
366 }
367 // Critical assertion: the SDK's dedup collapses every parallel
368 // invoke into one call to the token endpoint. Anything more
369 // means we're hammering the AS under load.
370 expect(calls).toHaveLength(1);
371 }),
372 );
373
374 it.effect(
375 "invalid_grant from refresh surfaces as ConnectionReauthRequiredError",
376 () =>
377 Effect.gen(function* () {
378 const { executor, scopeId } = yield* makeExecutor();
379 mockTokenFetch(
380 () =>
381 Effect.succeed(
382 new Response(
383 JSON.stringify({
384 error: "invalid_grant",
385 error_description: "Refresh token revoked",
386 }),
387 { status: 400, headers: { "content-type": "application/json" } },
388 ),
389 ),
390 );
391
392 const auth = yield* seedExpiredConnection(
393 executor,
394 scopeId,
395 "conn-refresh-dead",
396 );
397
398 yield* executor.openapi.addSpec({
399 spec: specJson,
400 scope: scopeId as unknown as string,
401 namespace: "petstore",
402 baseUrl: "",
403 oauth2: auth,
404 });
405
406 // Tool invocation currently wraps connection errors in a
407 // generic Error (see openapi invokeTool), so we assert against
408 // the `accessToken` call directly too — that's the surface
409 // the UI bridges use to trigger re-auth.
410 const flipped = yield* executor.connections
411 .accessToken("conn-refresh-dead")
412 .pipe(Effect.flip);
413 expect(flipped._tag).toBe("ConnectionReauthRequiredError");
414 expect((flipped as ConnectionReauthRequiredError).provider).toBe(
415 "openapi:oauth2",
416 );
417 expect(
418 (flipped as ConnectionReauthRequiredError).message,
419 ).toMatch(/invalid_grant|revoked/i);
420 }),
421 );
422});
packages/plugins/openapi/src/sdk/plugin.ts+20−0
@@ -1078,10 +1078,24 @@ export const openApiPlugin = definePlugin( · 1077 unmodified lines
1078 )1078 )
1079 : null;1079 : null;
1080 1080
1081 // RFC 6749 §5.2 terminal error codes — the AS has told us
1082 // the stored grant is unusable, so no amount of retries
1083 // will recover. Surface these as `reauthRequired: true`
1084 // so the SDK translates to `ConnectionReauthRequiredError`
1085 // at the caller and the UI prompts sign-in.
1086 const REAUTH_REQUIRED_ERRORS = new Set([
1087 "invalid_grant",
1088 "invalid_client",
1089 "unauthorized_client",
1090 ]);
1091
1081 const toRefreshError = (err: OAuth2Error) =>1092 const toRefreshError = (err: OAuth2Error) =>
1082 new ConnectionRefreshError({1093 new ConnectionRefreshError({
1083 connectionId: input.connectionId,1094 connectionId: input.connectionId,
1084 message: err.message,1095 message: err.message,
1096 reauthRequired: err.error
1097 ? REAUTH_REQUIRED_ERRORS.has(err.error)
1098 : false,
1085 cause: err,1099 cause: err,
1086 });1100 });
1087 1101
@@ -1094,10 +1108,16 @@ export const openApiPlugin = definePlugin( · 6 unmodified lines
1094 })1108 })
1095 : (() => {1109 : (() => {
1096 if (input.refreshToken === null) {1110 if (input.refreshToken === null) {
1111 // No refresh token stored — we cannot call the
1112 // token endpoint at all. This is the RFC 6749
1113 // §4.1 "must re-consent" case, modelled as a
1114 // reauth-required refresh failure so callers
1115 // hit the same UI path as an `invalid_grant`.
1097 return Effect.fail(1116 return Effect.fail(
1098 new OAuth2Error({1117 new OAuth2Error({
1099 message:1118 message:
1100 "authorizationCode connection has no refresh token",1119 "authorizationCode connection has no refresh token",
1120 error: "invalid_grant",
1101 }),1121 }),
1102 );1122 );
1103 }1123 }