lael-labs/executor

.changeset/oauth-dead-grant-classification.md+5−0

@@ -0,0 +1,5 @@
	1	---
	2	"executor": patch
	3	---
	4
	5	Fix: OAuth refresh rejections with non-spec error bodies (e.g. Datadog) now surface as expired connections with a reconnect path, and definitively dead refresh tokens are no longer retried against the authorization server

packages/core/sdk/src/executor.ts+88−1

@@ -730,6 +730,19 @@ const missingOAuthScopesFromProviderState = (value: unknown): readonly string[] · 729 unmodified lines
730	: [];	730	: [];
731	};	731	};
732		732
		733	/** Epoch ms of the definitive refresh rejection recorded on `provider_state`,
		734	* or null. Set when the AS rejects the grant itself (RFC 6749 invalid_grant —
		735	* retrying cannot change the verdict); cleared by the reconnect mint, which
		736	* rewrites `provider_state` wholesale. While set, refresh attempts are
		737	* skipped: the pre-fix behavior re-sent a known-dead grant to the AS every
		738	* proactive cycle, forever, and surfaced nothing to the user. */
		739	const oauthReauthRequiredAtFromProviderState = (value: unknown): number \| null => {
		740	const decoded = decodeJsonColumn(value);
		741	if (decoded == null \|\| typeof decoded !== "object" \|\| Array.isArray(decoded)) return null;
		742	const at = (decoded as Record<string, unknown>).oauthReauthRequiredAt;
		743	return typeof at === "number" ? at : null;
		744	};
		745
733	const rowToConnection = (row: ConnectionRow): Connection => {	746	const rowToConnection = (row: ConnectionRow): Connection => {
734	const owner = row.owner as Owner;	747	const owner = row.owner as Owner;
735	const integration = IntegrationSlug.make(row.integration);	748	const integration = IntegrationSlug.make(row.integration);
@@ -1691,6 +1704,44 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 955 unmodified lines
1691	* upstream 401 on a token we believed was still valid (`reactive`). */	1704	* upstream 401 on a token we believed was still valid (`reactive`). */
1692	type RefreshTrigger = "proactive" \| "reactive";	1705	type RefreshTrigger = "proactive" \| "reactive";
1693		1706
		1707	/** Record the AS's invalid_grant verdict on the row so later refreshes
		1708	* skip the doomed token request, and stamp `last_health` expired so the
		1709	* accounts list shows the dead connection at a glance instead of only
		1710	* after a manual probe. Merges into `provider_state` (preserving
		1711	* `missingOAuthScopes`); the reconnect mint rewrites the column wholesale,
		1712	* which is what re-arms refresh. Best-effort: a bookkeeping write failure
		1713	* must not mask the refresh failure being reported. */
		1714	const markRefreshGrantDead = (
		1715	row: ConnectionRow,
		1716	detail: string,
		1717	): Effect.Effect<void, never> => {
		1718	const existingState = decodeJsonColumn(row.provider_state);
		1719	const mergedState =
		1720	existingState != null && typeof existingState === "object" && !Array.isArray(existingState)
		1721	? (existingState as Record<string, unknown>)
		1722	: {};
		1723	const health: HealthCheckResult = {
		1724	status: "expired",
		1725	checkedAt: Date.now(),
		1726	detail,
		1727	};
		1728	return core
		1729	.updateMany("connection", {
		1730	where: (b: AnyCb) =>
		1731	b.and(
		1732	byOwner(row.owner as Owner)(b),
		1733	b("integration", "=", String(row.integration)),
		1734	b("name", "=", String(row.name)),
		1735	),
		1736	set: {
		1737	provider_state: { ...mergedState, oauthReauthRequiredAt: Date.now() },
		1738	last_health: health,
		1739	updated_at: new Date(),
		1740	},
		1741	})
		1742	.pipe(Effect.ignore);
		1743	};
		1744
1694	// Perform the actual refresh-token grant and persist the rotated material.	1745	// Perform the actual refresh-token grant and persist the rotated material.
1695	const performTokenRefresh = (	1746	const performTokenRefresh = (
1696	row: ConnectionRow,	1747	row: ConnectionRow,
@@ -1708,6 +1759,20 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 11 unmodified lines
1708	reauthRequired: true,	1759	reauthRequired: true,
1709	});	1760	});
1710		1761
		1762	// A recorded invalid_grant is the AS's standing verdict on this grant:
		1763	// re-sending it cannot succeed, so don't. Fail as reauth-required
		1764	// without a token request — the reconnect mint rewrites
		1765	// `provider_state` and thereby re-arms refresh. Without this gate a
		1766	// dead connection re-sent its dead grant on every proactive cycle,
		1767	// indefinitely (owner.com's Datadog connections: 100+ identical
		1768	// rejections over two days, surfacing nothing).
		1769	if (oauthReauthRequiredAtFromProviderState(row.provider_state) !== null) {
		1770	yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.skipped_known_dead": true });
		1771	return yield* reauth(
		1772	"The authorization server rejected this connection's refresh token (invalid_grant). Reconnect to continue.",
		1773	);
		1774	}
		1775
1711	// Load the backing app by the owner STORED on the connection (a Personal	1776	// Load the backing app by the owner STORED on the connection (a Personal
1712	// connection may be backed by a shared Workspace app) — no derivation.	1777	// connection may be backed by a shared Workspace app) — no derivation.
1713	const clientOwner = (row.oauth_client_owner ?? row.owner) as Owner;	1778	const clientOwner = (row.oauth_client_owner ?? row.owner) as Owner;
@@ -1813,6 +1878,16 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 99 unmodified lines
1813	cause,	1878	cause,
1814	});	1879	});
1815	}),	1880	}),
		1881	// Persist the definitive verdict so the NEXT refresh skips
		1882	// the doomed grant (see the known-dead gate above) and the
		1883	// connection shows `expired` without waiting for a probe.
		1884	Effect.tapError((error) =>
		1885	Predicate.isTagged(error, "CredentialResolutionError") &&
		1886	error.reauthRequired === true
		1887	? // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: CredentialResolutionError carries a typed `message` field
		1888	markRefreshGrantDead(row, error.message)
		1889	: Effect.void,
		1890	),
1816	);	1891	);
1817	});	1892	});
1818		1893
@@ -1879,6 +1954,12 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 60 unmodified lines
1879	),	1954	),
1880	Effect.withSpan("executor.oauth.refresh", {	1955	Effect.withSpan("executor.oauth.refresh", {
1881	attributes: {	1956	attributes: {
		1957	// Tenant + subject make refresh outcomes answerable PER CUSTOMER
		1958	// ("is org X's Datadog refresh healthy?") — without them the only
		1959	// grouping dimensions were integration-wide. Opaque ids, never
		1960	// emails or org names.
		1961	"executor.tenant": tenant,
		1962	...(subject != null ? { "executor.subject": subject } : {}),
1882	"executor.integration": String(row.integration),	1963	"executor.integration": String(row.integration),
1883	"executor.connection": String(row.name),	1964	"executor.connection": String(row.name),
1884	// Which path drove this refresh: the expiry check ahead of a call,	1965	// Which path drove this refresh: the expiry check ahead of a call,
@@ -3199,6 +3280,8 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 1314 unmodified lines
3199	}).pipe(	3280	}).pipe(
3200	Effect.withSpan("executor.connection.health.check", {	3281	Effect.withSpan("executor.connection.health.check", {
3201	attributes: {	3282	attributes: {
		3283	"executor.tenant": tenant,
		3284	...(subject != null ? { "executor.subject": subject } : {}),
3202	"executor.integration": String(ref.integration),	3285	"executor.integration": String(ref.integration),
3203	"executor.connection": String(ref.name),	3286	"executor.connection": String(ref.name),
3204	},	3287	},
@@ -3254,7 +3337,11 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 49 unmodified lines
3254	return result;	3337	return result;
3255	}).pipe(	3338	}).pipe(
3256	Effect.withSpan("executor.connection.validate", {	3339	Effect.withSpan("executor.connection.validate", {
3257	attributes: { "executor.integration": String(input.integration) },	3340	attributes: {
		3341	"executor.tenant": tenant,
		3342	...(subject != null ? { "executor.subject": subject } : {}),
		3343	"executor.integration": String(input.integration),
		3344	},
3258	}),	3345	}),
3259	);	3346	);
3260		3347

packages/core/sdk/src/oauth-flow.test.ts+110−0

@@ -962,6 +962,116 @@ describe("oauth token refresh in resolveConnectionValue", () => { · 961 unmodified lines
962	),	962	),
963	);	963	);
964		964
		965	it.effect("a definitively rejected grant is not re-sent to the AS on later refreshes", () =>
		966	Effect.scoped(
		967	Effect.gen(function* () {
		968	const server = yield* serveOAuthTestServer({
		969	scopes: ["read"],
		970	supportRefresh: false,
		971	tokenExpiresInSeconds: 0,
		972	invalidRefreshTokenDescription: "Grant revoked",
		973	});
		974	const harness = yield* makeTestWorkspaceHarness({ plugins });
		975	const { executor, config } = harness;
		976	yield* executor.acme.seed();
		977
		978	yield* executor.oauth.createClient({
		979	owner: "org",
		980	slug: CLIENT,
		981	authorizationUrl: server.authorizationEndpoint,
		982	tokenUrl: server.tokenEndpoint,
		983	grant: "authorization_code",
		984	clientId: "test-client",
		985	clientSecret: "test-secret",
		986	resource: server.mcpResourceUrl,
		987	});
		988
		989	const started = yield* executor.oauth.start({
		990	owner: "org",
		991	client: CLIENT,
		992	clientOwner: "org",
		993	name: ConnectionName.make("main"),
		994	integration: INTEG,
		995	template: TEMPLATE,
		996	});
		997	expect(started.status).toBe("redirect");
		998	if (started.status !== "redirect") return;
		999	const callback = yield* server.completeAuthorizationCodeFlow({
		1000	authorizationUrl: started.authorizationUrl,
		1001	});
		1002	yield* executor.oauth.complete({ state: started.state, code: callback.code });
		1003
		1004	yield* Effect.promise(() =>
		1005	config.db.updateMany("connection", {
		1006	where: (b) => b("name", "=", "main"),
		1007	set: { expires_at: Date.now() - 60_000 },
		1008	}),
		1009	);
		1010	yield* server.clearRequests;
		1011
		1012	// First resolve: the refresh grant reaches the AS and is rejected
		1013	// with invalid_grant — the AS's definitive verdict.
		1014	const first = yield* Effect.flip(
		1015	executor.execute(ToolAddress.make("tools.acme.org.main.whoami"), {}),
		1016	);
		1017	expect(JSON.stringify(first)).toContain("invalid_grant");
		1018
		1019	// The verdict is persisted: the dead-grant marker plus an expired
		1020	// health record, without waiting for a probe.
		1021	const row = yield* Effect.promise(() =>
		1022	config.db.findFirst("connection", { where: (b) => b("name", "=", "main") }),
		1023	);
		1024	expect(
		1025	(row?.provider_state as { oauthReauthRequiredAt?: number } \| null)?.oauthReauthRequiredAt,
		1026	).toEqual(expect.any(Number));
		1027	expect(row?.last_health).toMatchObject({ status: "expired" });
		1028
		1029	const grantRequests = () =>
		1030	server.requests.pipe(
		1031	Effect.map(
		1032	(all) =>
		1033	all.filter((r) => r.path === "/token" && r.body.includes("refresh_token")).length,
		1034	),
		1035	);
		1036	const sentBefore = yield* grantRequests();
		1037	expect(sentBefore).toBe(1);
		1038
		1039	// Later resolves still fail reauth-required, but WITHOUT re-sending
		1040	// the dead grant: the token endpoint sees no further traffic.
		1041	const second = yield* Effect.flip(
		1042	executor.execute(ToolAddress.make("tools.acme.org.main.whoami"), {}),
		1043	);
		1044	expect(JSON.stringify(second)).toContain("Reconnect");
		1045	expect(yield* grantRequests()).toBe(sentBefore);
		1046
		1047	// Reconnecting mints a fresh grant and re-arms refresh: the marker is
		1048	// gone and resolution works again.
		1049	const restarted = yield* executor.oauth.start({
		1050	owner: "org",
		1051	client: CLIENT,
		1052	clientOwner: "org",
		1053	name: ConnectionName.make("main"),
		1054	integration: INTEG,
		1055	template: TEMPLATE,
		1056	});
		1057	expect(restarted.status).toBe("redirect");
		1058	if (restarted.status !== "redirect") return;
		1059	const reCallback = yield* server.completeAuthorizationCodeFlow({
		1060	authorizationUrl: restarted.authorizationUrl,
		1061	});
		1062	yield* executor.oauth.complete({ state: restarted.state, code: reCallback.code });
		1063
		1064	const cleared = yield* Effect.promise(() =>
		1065	config.db.findFirst("connection", { where: (b) => b("name", "=", "main") }),
		1066	);
		1067	expect(
		1068	(cleared?.provider_state as { oauthReauthRequiredAt?: number } \| null)
		1069	?.oauthReauthRequiredAt,
		1070	).toBeUndefined();
		1071	}),
		1072	),
		1073	);
		1074
965	it.effect(	1075	it.effect(
966	"checkHealth reports healthy from OAuth credential resolution when no probe is configured",	1076	"checkHealth reports healthy from OAuth credential resolution when no probe is configured",
967	() =>	1077	() =>

packages/core/sdk/src/oauth-helpers.test.ts+62−0

@@ -871,6 +871,68 @@ describe("refreshAccessToken", () => { · 870 unmodified lines
871	}),	871	}),
872	),	872	),
873	);	873	);
		874
		875	// Datadog answers refresh grants with a non-conform envelope; the §5.2 code
		876	// must still be recovered so invalid_grant classifies as reauth-required
		877	// instead of a retried-forever transient (owner.com prod, 2026-07-30).
		878	it.effect("recovers invalid_grant from Datadog's non-conform errors array", () =>
		879	withTokenEndpoint(
		880	() =>
		881	json(400, {
		882	errors: ["invalid_grant - Invalid or expired refresh token or code verifier."],
		883	}),
		884	({ tokenUrl }) =>
		885	Effect.gen(function* () {
		886	const error = yield* Effect.flip(
		887	refreshAccessToken({ tokenUrl, clientId: "cid", refreshToken: "old" }),
		888	);
		889	expect(error).toBeInstanceOf(OAuth2Error);
		890	expect((error as OAuth2Error).error).toBe("invalid_grant");
		891	}),
		892	),
		893	);
		894
		895	it.effect("recovers a bare non-conform `error` string outside the spec envelope shape", () =>
		896	withTokenEndpoint(
		897	() => json(400, { error: "invalid_grant", detail: 42 }),
		898	({ tokenUrl }) =>
		899	Effect.gen(function* () {
		900	const error = yield* Effect.flip(
		901	refreshAccessToken({ tokenUrl, clientId: "cid", refreshToken: "old" }),
		902	);
		903	expect(error).toBeInstanceOf(OAuth2Error);
		904	expect((error as OAuth2Error).error).toBe("invalid_grant");
		905	}),
		906	),
		907	);
		908
		909	it.effect("does not invent a code from free-text error bodies", () =>
		910	withTokenEndpoint(
		911	() => json(400, { errors: ["something went wrong, try again later"] }),
		912	({ tokenUrl }) =>
		913	Effect.gen(function* () {
		914	const error = yield* Effect.flip(
		915	refreshAccessToken({ tokenUrl, clientId: "cid", refreshToken: "old" }),
		916	);
		917	expect(error).toBeInstanceOf(OAuth2Error);
		918	expect((error as OAuth2Error).error).toBeUndefined();
		919	}),
		920	),
		921	);
		922
		923	it.effect("does not probe non-conform bodies on 5xx responses", () =>
		924	withTokenEndpoint(
		925	() => json(502, { errors: ["invalid_grant - upstream proxy noise"] }),
		926	({ tokenUrl }) =>
		927	Effect.gen(function* () {
		928	const error = yield* Effect.flip(
		929	refreshAccessToken({ tokenUrl, clientId: "cid", refreshToken: "old" }),
		930	);
		931	expect(error).toBeInstanceOf(OAuth2Error);
		932	expect((error as OAuth2Error).error).toBeUndefined();
		933	}),
		934	),
		935	);
874	});	936	});
875		937
876	describe("shouldRefreshToken", () => {	938	describe("shouldRefreshToken", () => {

packages/core/sdk/src/oauth-helpers.ts+71−10

@@ -320,6 +320,55 @@ const bodyPreviewFromResponse = async (response: Response): Promise<string \| und · 319 unmodified lines
320	return redacted.length > 500 ? `${redacted.slice(0, 500)}...` : redacted;	320	return redacted.length > 500 ? `${redacted.slice(0, 500)}...` : redacted;
321	};	321	};
322		322
		323	// RFC 6749 §5.2's closed set. Only these are ever recovered from a
		324	// non-conform body: a free-text match against an open set would let an
		325	// arbitrary error message masquerade as an AS verdict.
		326	const RFC6749_TOKEN_ERROR_CODES = [
		327	"invalid_request",
		328	"invalid_client",
		329	"invalid_grant",
		330	"unauthorized_client",
		331	"unsupported_grant_type",
		332	"invalid_scope",
		333	] as const;
		334
		335	const rfc6749CodeFromCandidate = (candidate: unknown): string \| undefined => {
		336	if (typeof candidate !== "string") return undefined;
		337	const trimmed = candidate.trim();
		338	return RFC6749_TOKEN_ERROR_CODES.find(
		339	(code) => trimmed === code \|\| trimmed.startsWith(`${code} `) \|\| trimmed.startsWith(`${code}:`),
		340	);
		341	};
		342
		343	/** Recover the AS's §5.2 verdict from an error body oauth4webapi refused to
		344	* parse. Some ASes wrap the code in a non-conform envelope — Datadog answers
		345	* refresh grants with `{"errors": ["invalid_grant - Invalid or expired
		346	* refresh token or code verifier."]}` — and without this probe a definitive
		347	* `invalid_grant` (dead refresh token, reconnect required) is classified as
		348	* a transient failure and retried forever instead of surfacing a re-auth. */
		349	const oauthErrorCodeFromNonConformBody = (text: string): string \| undefined => {
		350	const parsed: unknown = (() => {
		351	// oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: probing an untrusted upstream body that already failed spec parsing; a parse failure just means "no recoverable code"
		352	try {
		353	// oxlint-disable-next-line executor/no-json-parse -- boundary: same untrusted-body probe; the value is only structurally inspected against the closed §5.2 code set, never decoded into domain types
		354	return JSON.parse(text) as unknown;
		355	} catch {
		356	return undefined;
		357	}
		358	})();
		359	if (typeof parsed !== "object" \|\| parsed === null) return undefined;
		360	const envelope = parsed as { readonly error?: unknown; readonly errors?: unknown };
		361	const direct = rfc6749CodeFromCandidate(envelope.error);
		362	if (direct) return direct;
		363	if (Array.isArray(envelope.errors)) {
		364	for (const entry of envelope.errors) {
		365	const code = rfc6749CodeFromCandidate(entry);
		366	if (code) return code;
		367	}
		368	}
		369	return undefined;
		370	};
		371
323	const toOAuth2Error = (cause: unknown): OAuth2Error => {	372	const toOAuth2Error = (cause: unknown): OAuth2Error => {
324	if (isOAuth2Error(cause)) return cause;	373	if (isOAuth2Error(cause)) return cause;
325	if (typeof cause === "object" && cause !== null) {	374	if (typeof cause === "object" && cause !== null) {
@@ -352,16 +401,28 @@ const toOAuth2ErrorWithHttpSummary = (cause: unknown): Effect.Effect<OAuth2Error · 26 unmodified lines
352	const base = toOAuth2Error(cause);	401	const base = toOAuth2Error(cause);
353	const response = responseFromOAuthErrorCause(cause);	402	const response = responseFromOAuthErrorCause(cause);
354	if (!response) return Effect.succeed(base);	403	if (!response) return Effect.succeed(base);
355	return Effect.promise(() => tokenEndpointHttpSummary(response)).pipe(	404	return Effect.promise(async () => {
356	Effect.map(	405	const summary = await tokenEndpointHttpSummary(response);
357	(summary) =>	406	// A 4xx the spec parser refused may still carry the AS's verdict in a
358	new OAuth2Error({	407	// non-conform envelope; recover it so classification (invalid_grant →
359	message: `${base.message} (${summary})`,	408	// reauth-required) sees the code instead of a code-less "transient".
360	error: base.error,	409	const recovered =
361	cause,	410	base.error === undefined && response.status >= 400 && response.status < 500
362	}),	411	? oauthErrorCodeFromNonConformBody(
363	),	412	await Promise.resolve()
364	);	413	.then(() => response.clone().text())
		414	.then(
		415	(value) => value,
		416	() => "",
		417	),
		418	)
		419	: undefined;
		420	return new OAuth2Error({
		421	message: `${base.message} (${summary})`,
		422	error: base.error ?? recovered,
		423	cause,
		424	});
		425	});
365	};	426	};
366		427
367	const failOAuth2WithHttpSummary = (cause: unknown): Effect.Effect<never, OAuth2Error> =>	428	const failOAuth2WithHttpSummary = (cause: unknown): Effect.Effect<never, OAuth2Error> =>

packages/core/sdk/src/oauth-service.ts+7−1

       return connection;
     }).pipe(
       Effect.withSpan("executor.oauth.complete", {
-        attributes: { "executor.oauth.grant": "authorization_code" },
+        attributes: {
+          "executor.oauth.grant": "authorization_code",
+          // Same per-customer dimensions as executor.oauth.refresh, so a
+          // connect and its later refresh failures group under one tenant.
+          "executor.tenant": deps.tenant,
+          ...(deps.subject != null ? { "executor.subject": deps.subject } : {}),
+        },
       }),
     );