History

interrupt speculative reads on branches that never consume them

Rhys Sullivan committed 2 months ago

Commit 60be4171 parent 98885112 files · +437 −178Browse files

Unified · Split

2 files
  1. packages/core/sdk/src/executor.test.ts +233 −1
  2. packages/core/sdk/src/executor.ts +204 −177
packages/core/sdk/src/executor.test.ts+233−1
@@ -4,7 +4,7 @@ import { Data, Effect, Predicate, Result } from "effect"; · 3 unmodified lines
4import { ElicitationResponse, type ElicitationHandler } from "./elicitation";4import { ElicitationResponse, type ElicitationHandler } from "./elicitation";
5import { ToolNotFoundError } from "./errors";5import { ToolNotFoundError } from "./errors";
6import { createExecutor } from "./executor";6import { createExecutor } from "./executor";
7import type { FumaDb } from "./fuma-runtime";7import { StorageError, type FumaDb } from "./fuma-runtime";
8import {8import {
9 AuthTemplateSlug,9 AuthTemplateSlug,
10 ConnectionName,10 ConnectionName,
@@ -1013,3 +1013,235 @@ describe("execute read concurrency", () => { · 1002 unmodified lines
1013 }),1013 }),
1014 );1014 );
1015});1015});
1016
1017// ---------------------------------------------------------------------------
1018// Speculative read abandonment. The concurrent reads above are forked, and a
1019// branch that returns without needing one must interrupt it instead of
1020// consuming it — so a read that hangs cannot gate an error that never needed
1021// it, and a read that fails cannot mask that error or leak as an unhandled
1022// rejection. Reads a branch DOES need keep failing exactly where the
1023// sequential code failed.
1024// ---------------------------------------------------------------------------
1025
1026/** Deterministic read faults keyed by `<table>.<method>`: nothing is armed
1027 * until the test says so (setup reads pass through untouched), and the test
1028 * can check that an armed read really started before it was abandoned. */
1029const makeReadFaults = () => {
1030 let hung: ReadonlySet<string> = new Set();
1031 let failed: ReadonlyMap<string, string> = new Map();
1032 const started = new Set<string>();
1033 return {
1034 hangReads: (keys: readonly string[]) => {
1035 hung = new Set(keys);
1036 },
1037 failReads: (byCode: Readonly<Record<string, string>>) => {
1038 failed = new Map(Object.entries(byCode));
1039 },
1040 started,
1041 fault: (key: string): Promise<never> | undefined => {
1042 if (hung.has(key)) {
1043 started.add(key);
1044 // Never settles. The driver promise takes no abort signal, so an
1045 // interrupted read is abandoned exactly like this in production.
1046 return new Promise<never>(() => {});
1047 }
1048 const code = failed.get(key);
1049 if (code === undefined) return undefined;
1050 started.add(key);
1051 // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: simulate the raw driver-promise rejection that fumaEffect normalizes into a StorageFailure
1052 return Promise.reject(Object.assign(new Error(code), { code }));
1053 },
1054 };
1055};
1056
1057/** Wrap a test `FumaDb` so armed reads hang or reject; every other read is
1058 * untouched. `withContext` re-wraps so context-bound handles stay faulted. */
1059const withFaultedReads = (
1060 db: FumaDb,
1061 fault: (key: string) => Promise<never> | undefined,
1062): FumaDb => {
1063 const wrap = (inner: FumaDb): FumaDb =>
1064 new Proxy(inner, {
1065 get(target, prop) {
1066 if (prop === "withContext") {
1067 return (context: unknown) =>
1068 wrap((target.withContext as (c: unknown) => FumaDb)(context));
1069 }
1070 if (prop === "findFirst" || prop === "findMany") {
1071 return (table: unknown, query: unknown) =>
1072 fault(`${String(table)}.${prop}`) ??
1073 (Reflect.get(target, prop) as (t: unknown, q: unknown) => Promise<unknown>).call(
1074 target,
1075 table,
1076 query,
1077 );
1078 }
1079 return Reflect.get(target, prop);
1080 },
1081 });
1082 return wrap(db);
1083};
1084
1085/** A credential provider that works during setup and can be armed to fail
1086 * every later read — the site-2 "credential resolution fails" input. */
1087const armableFailingProvider = () => {
1088 const store = new Map<string, string>();
1089 let failWith: string | undefined;
1090 const provider: CredentialProvider = {
1091 key: ProviderKey.make("memory"),
1092 writable: true,
1093 get: (id) =>
1094 Effect.suspend(() =>
1095 failWith === undefined
1096 ? Effect.succeed(store.get(String(id)) ?? null)
1097 : Effect.fail(new StorageError({ message: failWith, cause: undefined })),
1098 ),
1099 set: (id, value) => Effect.sync(() => void store.set(String(id), value)),
1100 };
1101 return {
1102 provider,
1103 failNow: (message: string) => {
1104 failWith = message;
1105 },
1106 };
1107};
1108
1109describe("speculative read abandonment", () => {
1110 it.effect("unknown-tool error surfaces while the speculative reads hang forever", () =>
1111 Effect.gen(function* () {
1112 const faults = makeReadFaults();
1113 const config = makeTestConfig({ plugins: [demoPlugin] as const });
1114 const executor = yield* createExecutor({
1115 ...config,
1116 db: withFaultedReads(config.db, faults.fault),
1117 });
1118 yield* seedRunConnection(executor);
1119
1120 // From here on the speculative policy and connection reads NEVER
1121 // resolve. The unknown-tool branch needs neither, so the call must
1122 // fail without waiting on them — the previous all-or-nothing shape
1123 // could not fail until every read settled.
1124 faults.hangReads(["tool_policy.findMany", "connection.findFirst"]);
1125 const result = yield* Effect.result(executor.execute(addr("no-such-tool"), {}));
1126 expect(Result.isFailure(result)).toBe(true);
1127 if (!Result.isFailure(result)) return;
1128 expect(Predicate.isTagged("ToolNotFoundError")(result.failure)).toBe(true);
1129 // Both hung reads really started: the branch abandoned in-flight
1130 // reads, it did not skip forking them.
1131 expect(faults.started.has("tool_policy.findMany")).toBe(true);
1132 expect(faults.started.has("connection.findFirst")).toBe(true);
1133 }),
1134 );
1135
1136 it.effect("a blocked tool reports ToolBlockedError while the connection read hangs", () =>
1137 Effect.gen(function* () {
1138 const faults = makeReadFaults();
1139 const config = makeTestConfig({ plugins: [demoPlugin] as const });
1140 const executor = yield* createExecutor({
1141 ...config,
1142 db: withFaultedReads(config.db, faults.fault),
1143 });
1144 yield* seedRunConnection(executor);
1145 yield* executor.policies.create({
1146 owner: "org",
1147 pattern: "demo.*",
1148 action: "block",
1149 });
1150
1151 // The block branch consumes the policy read but never the connection
1152 // read; a connection read that never resolves must not gate it.
1153 faults.hangReads(["connection.findFirst"]);
1154 const result = yield* Effect.result(executor.execute(addr("run"), {}));
1155 expect(Result.isFailure(result)).toBe(true);
1156 if (!Result.isFailure(result)) return;
1157 expect(Predicate.isTagged("ToolBlockedError")(result.failure)).toBe(true);
1158 expect(faults.started.has("connection.findFirst")).toBe(true);
1159 }),
1160 );
1161
1162 it.effect("failing speculative reads neither mask the branch error nor unhandled-reject", () =>
1163 Effect.gen(function* () {
1164 const faults = makeReadFaults();
1165 const config = makeTestConfig({ plugins: [demoPlugin] as const });
1166 const executor = yield* createExecutor({
1167 ...config,
1168 db: withFaultedReads(config.db, faults.fault),
1169 });
1170 yield* seedRunConnection(executor);
1171
1172 const unhandled: unknown[] = [];
1173 const onUnhandled = (reason: unknown) => void unhandled.push(reason);
1174 process.on("unhandledRejection", onUnhandled);
1175 const result = yield* Effect.gen(function* () {
1176 faults.failReads({
1177 "tool_policy.findMany": "POLICY_READ_FAILED",
1178 "connection.findFirst": "CONNECTION_READ_FAILED",
1179 });
1180 const out = yield* Effect.result(executor.execute(addr("no-such-tool"), {}));
1181 // Both rejections fired before the call returned; give an
1182 // unobserved one its macrotask turn to reach the process hook.
1183 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 0)));
1184 return out;
1185 }).pipe(Effect.ensuring(Effect.sync(() => process.off("unhandledRejection", onUnhandled))));
1186 expect(Result.isFailure(result)).toBe(true);
1187 if (!Result.isFailure(result)) return;
1188 expect(Predicate.isTagged("ToolNotFoundError")(result.failure)).toBe(true);
1189 expect(unhandled).toEqual([]);
1190 }),
1191 );
1192
1193 it.effect(
1194 "a policy read failure on the consuming path surfaces, ahead of a connection failure",
1195 () =>
1196 Effect.gen(function* () {
1197 const faults = makeReadFaults();
1198 const config = makeTestConfig({ plugins: [demoPlugin] as const });
1199 const executor = yield* createExecutor({
1200 ...config,
1201 db: withFaultedReads(config.db, faults.fault),
1202 });
1203 yield* seedRunConnection(executor);
1204
1205 faults.failReads({
1206 "tool_policy.findMany": "POLICY_READ_FAILED",
1207 "connection.findFirst": "CONNECTION_READ_FAILED",
1208 });
1209 const result = yield* Effect.result(executor.execute(addr("run"), {}));
1210 expect(Result.isFailure(result)).toBe(true);
1211 if (!Result.isFailure(result)) return;
1212 // The policy read is consumed first, exactly as the sequential code
1213 // ordered the reads, so its failure is the one reported even though
1214 // the connection read failed too.
1215 expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true);
1216 expect((result.failure as StorageError).message).toContain("POLICY_READ_FAILED");
1217 }),
1218 );
1219
1220 it.effect("a credential failure surfaces while the site-2 integration read hangs", () =>
1221 Effect.gen(function* () {
1222 const faults = makeReadFaults();
1223 const armable = armableFailingProvider();
1224 const config = makeTestConfig({
1225 plugins: [invokeConcurrencyPlugin(armable.provider)] as const,
1226 });
1227 const executor = yield* createExecutor({
1228 ...config,
1229 db: withFaultedReads(config.db, faults.fault),
1230 });
1231 yield* seedRunConnection(executor);
1232
1233 // After approval, credential resolution and the integration-row read
1234 // run concurrently. Resolution fails while the integration read never
1235 // resolves: the credential failure must surface without waiting on the
1236 // read — the failure path interrupts it instead of joining it.
1237 faults.hangReads(["integration.findFirst"]);
1238 armable.failNow("CREDENTIAL_READ_FAILED");
1239 const result = yield* Effect.result(executor.execute(addr("run"), {}));
1240 expect(Result.isFailure(result)).toBe(true);
1241 if (!Result.isFailure(result)) return;
1242 expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true);
1243 expect((result.failure as StorageError).message).toBe("CREDENTIAL_READ_FAILED");
1244 expect(faults.started.has("integration.findFirst")).toBe(true);
1245 }),
1246 );
1247});
packages/core/sdk/src/executor.ts+204−177
@@ -5800,194 +5800,221 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 5799 unmodified lines
5800 // surface), the active policy rule set, and the connection row — are5800 // surface), the active policy rule set, and the connection row — are
5801 // mutually independent, so they run concurrently instead of paying5801 // mutually independent, so they run concurrently instead of paying
5802 // three serial round-trips. The policy and connection reads are5802 // three serial round-trips. The policy and connection reads are
5803 // captured as Exits and unwrapped at exactly the points the5803 // forked as children of this fiber (so interrupting the call still
5804 // sequential code performed them, so the caller-visible error for a5804 // interrupts them) and consumed with `Fiber.join` at exactly the
5805 // given input is unchanged: a tool-row read failure still dominates,5805 // points the sequential code performed them, so the caller-visible
5806 // and a policy or connection read failure still surfaces only where5806 // error for a given input is unchanged: a tool-row read failure
5807 // the old code would have executed that read.5807 // still dominates, and a policy or connection read failure still
5808 const { row, policyRulesExit, connectionRowExit } = yield* Effect.all(5808 // surfaces only where the old code would have executed that read.
5809 {5809 // A branch that returns without needing a speculative read must
5810 row: core.findFirst("tool", {5810 // neither wait on it (a hung read must not gate an unknown-tool /
5811 where: (b: AnyCb) =>5811 // blocked / plugin-not-loaded error that never needed it) nor
5812 b.and(5812 // swallow its failure as an unobserved value — the `ensuring` guard
5813 byOwner(parsed.owner)(b),5813 // below interrupts whatever was not consumed. Interrupting a read
5814 b("integration", "=", String(parsed.integration)),5814 // mid-flight merely abandons the driver promise (`fumaEffect` takes
5815 b("connection", "=", String(parsed.connection)),5815 // no abort signal and installs its rejection handler at
5816 b("name", "=", String(parsed.tool)),5816 // construction), so an abandoned read cannot unhandled-reject.
5817 ),5817 const policyRulesFiber = yield* Effect.forkChild(listActivePolicyRuleSet(), {
5818 select: TOOL_INVOCATION_COLUMNS,5818 startImmediately: true,
5819 }),5819 });
5820 policyRulesExit: Effect.exit(listActivePolicyRuleSet()),5820 const connectionRowFiber = yield* Effect.forkChild(
5821 connectionRowExit: Effect.exit(5821 findConnectionRow({
5822 findConnectionRow({5822 owner: parsed.owner,
5823 owner: parsed.owner,5823 integration: parsed.integration,
5824 integration: parsed.integration,5824 name: parsed.connection,
5825 name: parsed.connection,5825 }),
5826 }),5826 { startImmediately: true },
5827 ),
5828 },
5829 { concurrency: 3 },
5830 );5827 );
5831 if (!row) {5828 const invokeDynamicTool = Effect.gen(function* () {
5832 const searchMatches = yield* searchToolRowsForConnection(parsed);5829 const row = yield* core.findFirst("tool", {
5833 const connectionTools =5830 where: (b: AnyCb) =>
5834 searchMatches.length > 0 ? searchMatches : yield* findToolRowsForConnection(parsed);5831 b.and(
5835 // An empty catalog on a connection that DOES exist is usually not a5832 byOwner(parsed.owner)(b),
5836 // wrong tool name: discovery produced nothing, most often because the5833 b("integration", "=", String(parsed.integration)),
5837 // upstream rejected the credential. Reporting only the address sends5834 b("connection", "=", String(parsed.connection)),
5838 // the reader after a tool that was never the problem, so name the5835 b("name", "=", String(parsed.tool)),
5839 // connection and point at the surface that knows the cause.5836 ),
5840 const connectionExists =5837 select: TOOL_INVOCATION_COLUMNS,
5841 connectionTools.length === 0 && (yield* connectionRowExit) !== null;
5842 return yield* new ToolNotFoundError({
5843 address,
5844 suggestions: toolSuggestions(connectionTools),
5845 reason: connectionExists
5846 ? `connection "${parsed.integration}/${parsed.connection}" has no tools; ` +
5847 `check its health for why discovery produced none`
5848 : undefined,
5849 });5838 });
5850 }5839 if (!row) {
5840 const searchMatches = yield* searchToolRowsForConnection(parsed);
5841 const connectionTools =
5842 searchMatches.length > 0 ? searchMatches : yield* findToolRowsForConnection(parsed);
5843 // An empty catalog on a connection that DOES exist is usually not a
5844 // wrong tool name: discovery produced nothing, most often because the
5845 // upstream rejected the credential. Reporting only the address sends
5846 // the reader after a tool that was never the problem, so name the
5847 // connection and point at the surface that knows the cause.
5848 // Joining here is the sequential read this branch always
5849 // performed; the short-circuit keeps the suggestion path from
5850 // waiting on a connection read it does not need.
5851 const connectionExists =
5852 connectionTools.length === 0 && (yield* Fiber.join(connectionRowFiber)) !== null;
5853 return yield* new ToolNotFoundError({
5854 address,
5855 suggestions: toolSuggestions(connectionTools),
5856 reason: connectionExists
5857 ? `connection "${parsed.integration}/${parsed.connection}" has no tools; ` +
5858 `check its health for why discovery produced none`
5859 : undefined,
5860 });
5861 }
5851 5862
5852 // Resolve policy (owner-ranked).5863 // Resolve policy (owner-ranked).
5853 const toolForPolicy = rowToTool(row);5864 const toolForPolicy = rowToTool(row);
5854 const policyRules = yield* policyRulesExit;5865 const policyRules = yield* Fiber.join(policyRulesFiber);
5855 const annotations = decodeJsonColumn(row.annotations) as ToolAnnotations | undefined;5866 const annotations = decodeJsonColumn(row.annotations) as ToolAnnotations | undefined;
5856 const policy = yield* resolvePolicyFromRuleSet(5867 const policy = yield* resolvePolicyFromRuleSet(
5857 normalizedPolicyId(toolForPolicy),5868 normalizedPolicyId(toolForPolicy),
5858 policyRules,5869 policyRules,
5859 annotations?.requiresApproval,5870 annotations?.requiresApproval,
5860 );5871 );
5861 if (policy.action === "block") {5872 if (policy.action === "block") {
5862 return yield* new ToolBlockedError({5873 return yield* new ToolBlockedError({
5863 address,5874 address,
5864 pattern: policy.pattern ?? "*",5875 pattern: policy.pattern ?? "*",
5865 });5876 });
5866 }5877 }
5867 5878
5868 const runtime = runtimes.get(row.plugin_id);5879 const runtime = runtimes.get(row.plugin_id);
5869 if (!runtime) {5880 if (!runtime) {
5870 return yield* new PluginNotLoadedError({5881 return yield* new PluginNotLoadedError({
5871 address,5882 address,
5872 pluginId: row.plugin_id,5883 pluginId: row.plugin_id,
5873 });5884 });
5874 }5885 }
5875 if (!runtime.plugin.invokeTool) {5886 if (!runtime.plugin.invokeTool) {
5876 return yield* new NoHandlerError({5887 return yield* new NoHandlerError({
5877 address,5888 address,
5878 pluginId: row.plugin_id,5889 pluginId: row.plugin_id,
5879 });5890 });
5880 }5891 }
5881 5892
5882 // Unwrap the connection row (read concurrently above).5893 // Join the connection row (read concurrently above).
5883 const connectionRow = yield* connectionRowExit;5894 const connectionRow = yield* Fiber.join(connectionRowFiber);
5884 if (!connectionRow) {5895 if (!connectionRow) {
5885 return yield* new ConnectionNotFoundError({5896 return yield* new ConnectionNotFoundError({
5886 owner: parsed.owner,5897 owner: parsed.owner,
5887 integration: parsed.integration,5898 integration: parsed.integration,
5888 name: parsed.connection,5899 name: parsed.connection,
5889 });5900 });
5890 }5901 }
5891 5902
5892 // Resolve annotations + enforce approval.5903 // Resolve annotations + enforce approval.
5893 let resolvedAnnotations = annotations;5904 let resolvedAnnotations = annotations;
5894 if (policy.action !== "approve" && runtime.plugin.resolveAnnotations) {5905 if (policy.action !== "approve" && runtime.plugin.resolveAnnotations) {
5895 const map = yield* runtime.plugin5906 const map = yield* runtime.plugin
5896 .resolveAnnotations({5907 .resolveAnnotations({
5897 ctx: runtime.ctx,5908 ctx: runtime.ctx,
5909 integration: parsed.integration,
5910 connection: parsed.connection,
5911 toolRows: [row],
5912 })
5913 .pipe(wrapInvocationError);
5914 resolvedAnnotations = map[String(parsed.tool)] ?? annotations;
5915 }
5916 // When this call is about to pause for approval, validate args
5917 // first: a call that can only fail (missing required path param /
5918 // body) must be rejected here, not after the user grants an approval
5919 // that then goes to waste. Non-pausing calls skip this — invokeTool
5920 // raises the identical failure moments later without the extra pass.
5921 if (approvalRequired(resolvedAnnotations, policy) && runtime.plugin.validateToolArgs) {
5922 yield* runtime.plugin
5923 .validateToolArgs({ ctx: runtime.ctx, toolRow: row, args })
5924 .pipe(wrapInvocationError);
5925 }
5926 yield* enforceApproval(resolvedAnnotations, address, args, policy, handler);
5927
5928 // Resolve every named credential input (`variable → value`); `value` is
5929 // the primary `token` for single-input + OAuth callers. The
5930 // integration-row read is independent of credential resolution, so
5931 // it is forked and the two run concurrently. Both start only after
5932 // `enforceApproval` above completes — a declined call must never
5933 // trigger the token refresh credential resolution can perform. The
5934 // fork is joined after `values`, so a credential resolution failure
5935 // keeps dominating a storage failure exactly as it did when the
5936 // reads were sequential; on that failure path the fork is
5937 // interrupted rather than joined, so a hung integration read cannot
5938 // gate the credential error and a failed one is deliberately
5939 // abandoned, never silently dropped as an unobserved value.
5940 const integrationRowFiber = yield* Effect.forkChild(
5941 findIntegrationRow(parsed.integration),
5942 {
5943 startImmediately: true,
5944 },
5945 );
5946 const values = yield* resolveConnectionValues(connectionRow).pipe(
5947 Effect.onError(() => Fiber.interrupt(integrationRowFiber)),
5948 );
5949 const integrationRow = yield* Fiber.join(integrationRowFiber);
5950 const grantedScopes = grantedScopesFromRow(connectionRow);
5951 const invokeTool = runtime.plugin.invokeTool;
5952 const invokeWith = (
5953 resolved: Record<string, string | null>,
5954 ): Effect.Effect<unknown, ToolInvocationError> => {
5955 const credential: ToolInvocationCredential = {
5956 owner: parsed.owner,
5898 integration: parsed.integration,5957 integration: parsed.integration,
5899 connection: parsed.connection,5958 connection: parsed.connection,
5900 toolRows: [row],5959 template: AuthTemplateSlug.make(connectionRow.template),
5901 })5960 value: resolved[PRIMARY_INPUT_VARIABLE] ?? null,
5902 .pipe(wrapInvocationError);5961 values: resolved,
5903 resolvedAnnotations = map[String(parsed.tool)] ?? annotations;5962 config: integrationRow ? decodeJsonColumn(integrationRow.config) : undefined,
5904 }5963 ...(grantedScopes ? { grantedScopes } : {}),
5905 // When this call is about to pause for approval, validate args5964 };
5906 // first: a call that can only fail (missing required path param /5965 return wrapInvocationError(
5907 // body) must be rejected here, not after the user grants an approval5966 invokeTool({
5908 // that then goes to waste. Non-pausing calls skip this — invokeTool5967 ctx: runtime.ctx,
5909 // raises the identical failure moments later without the extra pass.5968 toolRow: row,
5910 if (approvalRequired(resolvedAnnotations, policy) && runtime.plugin.validateToolArgs) {5969 credential,
5911 yield* runtime.plugin5970 args,
5912 .validateToolArgs({ ctx: runtime.ctx, toolRow: row, args })5971 elicit: buildElicit(address, args, handler),
5913 .pipe(wrapInvocationError);5972 invokeOptions: options,
5914 }5973 }),
5915 yield* enforceApproval(resolvedAnnotations, address, args, policy, handler);5974 );
5916
5917 // Resolve every named credential input (`variable → value`); `value` is
5918 // the primary `token` for single-input + OAuth callers. The
5919 // integration-row read is independent of credential resolution, so
5920 // the two run concurrently. Both start only after `enforceApproval`
5921 // above completes — a declined call must never trigger the token
5922 // refresh credential resolution can perform. The integration read is
5923 // captured as an Exit and unwrapped after `values`, so a credential
5924 // resolution failure keeps dominating a storage failure exactly as
5925 // it did when the reads were sequential.
5926 const { values, integrationRowExit } = yield* Effect.all(
5927 {
5928 values: resolveConnectionValues(connectionRow),
5929 integrationRowExit: Effect.exit(findIntegrationRow(parsed.integration)),
5930 },
5931 { concurrency: 2 },
5932 );
5933 const integrationRow = yield* integrationRowExit;
5934 const grantedScopes = grantedScopesFromRow(connectionRow);
5935 const invokeTool = runtime.plugin.invokeTool;
5936 const invokeWith = (
5937 resolved: Record<string, string | null>,
5938 ): Effect.Effect<unknown, ToolInvocationError> => {
5939 const credential: ToolInvocationCredential = {
5940 owner: parsed.owner,
5941 integration: parsed.integration,
5942 connection: parsed.connection,
5943 template: AuthTemplateSlug.make(connectionRow.template),
5944 value: resolved[PRIMARY_INPUT_VARIABLE] ?? null,
5945 values: resolved,
5946 config: integrationRow ? decodeJsonColumn(integrationRow.config) : undefined,
5947 ...(grantedScopes ? { grantedScopes } : {}),
5948 };5975 };
5949 return wrapInvocationError(
5950 invokeTool({
5951 ctx: runtime.ctx,
5952 toolRow: row,
5953 credential,
5954 args,
5955 elicit: buildElicit(address, args, handler),
5956 invokeOptions: options,
5957 }),
5958 );
5959 };
5960 5976
5961 const first = yield* invokeWith(values);5977 const first = yield* invokeWith(values);
5962 // Reactive refresh. `expires_at` is only ever the AS's ADVERTISED5978 // Reactive refresh. `expires_at` is only ever the AS's ADVERTISED
5963 // lifetime; the upstream rejecting the token is the authoritative word5979 // lifetime; the upstream rejecting the token is the authoritative word
5964 // on whether it is still good. The two diverge routinely: server-side5980 // on whether it is still good. The two diverge routinely: server-side
5965 // revocation, an identity provider's idle-timeout policy shorter than5981 // revocation, an identity provider's idle-timeout policy shorter than
5966 // the token lifetime, and connections whose AS omitted `expires_in`5982 // the token lifetime, and connections whose AS omitted `expires_in`
5967 // entirely (null expiry → the proactive check never fires, so this is5983 // entirely (null expiry → the proactive check never fires, so this is
5968 // their ONLY route back to a working token short of a reconnect).5984 // their ONLY route back to a working token short of a reconnect).
5969 //5985 //
5970 // Deliberately narrow: exactly one retry, only on the 401 that means5986 // Deliberately narrow: exactly one retry, only on the 401 that means
5971 // "this credential is not valid", and only for a connection holding a5987 // "this credential is not valid", and only for a connection holding a
5972 // refresh token. A 403 is excluded — it means authenticated-but-not-5988 // refresh token. A 403 is excluded — it means authenticated-but-not-
5973 // permitted, and re-minting the same grant returns the same answer.5989 // permitted, and re-minting the same grant returns the same answer.
5974 // If the retry also fails its result stands, so a genuinely dead grant5990 // If the retry also fails its result stands, so a genuinely dead grant
5975 // still surfaces the upstream's own auth failure and its reconnect5991 // still surfaces the upstream's own auth failure and its reconnect
5976 // guidance rather than a masked one.5992 // guidance rather than a masked one.
5977 const { result, usedValues } = yield* Effect.gen(function* () {5993 const { result, usedValues } = yield* Effect.gen(function* () {
5978 if (!isUnauthorizedToolFailure(first)) return { result: first, usedValues: values };5994 if (!isUnauthorizedToolFailure(first)) return { result: first, usedValues: values };
5979 const refreshed = yield* forceRefreshConnectionValues(connectionRow).pipe(5995 const refreshed = yield* forceRefreshConnectionValues(connectionRow).pipe(
5980 // A failed re-mint is not this call's failure to report: the upstream5996 // A failed re-mint is not this call's failure to report: the upstream
5981 // already produced an auth failure with recovery guidance, which is5997 // already produced an auth failure with recovery guidance, which is
5982 // strictly more actionable than a refresh-plumbing error. Keep it.5998 // strictly more actionable than a refresh-plumbing error. Keep it.
5983 Effect.catchTag("CredentialResolutionError", () => Effect.succeed(null)),5999 Effect.catchTag("CredentialResolutionError", () => Effect.succeed(null)),
5984 );6000 );
5985 if (!refreshed) return { result: first, usedValues: values };6001 if (!refreshed) return { result: first, usedValues: values };
5986 yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.retried": true });6002 yield* Effect.annotateCurrentSpan({ "executor.oauth.refresh.retried": true });
5987 return { result: yield* invokeWith(refreshed), usedValues: refreshed };6003 return { result: yield* invokeWith(refreshed), usedValues: refreshed };
6004 });
6005 yield* healPersistedHealthOnUse(connectionRow, result, usedValues);
6006 return result;
5988 });6007 });
5989 yield* healPersistedHealthOnUse(connectionRow, result, usedValues);6008 // Interrupting an already-completed (or already-joined) fiber is a
5990 return result;6009 // no-op, so this single guard covers every exit path: a path that
6010 // consumed a speculative read leaves a finished fiber behind, and a
6011 // path that exited early — an early-return branch or a tool-row read
6012 // failure — deliberately abandons the reads it never needed instead
6013 // of waiting on them or dropping their failures unobserved.
6014 return yield* Effect.ensuring(
6015 invokeDynamicTool,
6016 Fiber.interruptAll([policyRulesFiber, connectionRowFiber]),
6017 );
5991 }).pipe(6018 }).pipe(
5992 // Expected tool failures (`ToolResult.fail`) resolve through the6019 // Expected tool failures (`ToolResult.fail`) resolve through the
5993 // success channel, so the tracer alone would record them as healthy6020 // success channel, so the tracer alone would record them as healthy