| @@ -1,7 +1,10 @@ |
| 1 | 1 | import { describe, expect, it } from "@effect/vitest"; |
| 2 | 2 | import { Data, Effect, Predicate, Result } from "effect"; |
| 3 | 3 | |
| 4 | +import { ElicitationResponse, type ElicitationHandler } from "./elicitation"; |
| 4 | 5 | import { ToolNotFoundError } from "./errors"; |
| 6 | +import { createExecutor } from "./executor"; |
| 7 | +import type { FumaDb } from "./fuma-runtime"; |
| 5 | 8 | import { |
| 6 | 9 | AuthTemplateSlug, |
| 7 | 10 | ConnectionName, |
| @@ -15,7 +18,7 @@ import { · 7 unmodified lines |
| 15 | 18 | import { definePlugin } from "./plugin"; |
| 16 | 19 | import type { CredentialProvider } from "./provider"; |
| 17 | 20 | import { IntegrationDetectionResult } from "./types"; |
| 18 | | −import { makeTestExecutor, memoryCredentialsPlugin } from "./testing"; |
| 21 | +import { makeTestConfig, makeTestExecutor, memoryCredentialsPlugin } from "./testing"; |
| 19 | 22 | import { serveOAuthTestServer } from "./testing/oauth-test-server"; |
| 20 | 23 | |
| 21 | 24 | // removed: v1 secret browser-handoff, source.configure, case-insensitive tool-id |
| @@ -783,3 +786,230 @@ describe("muscle memory (observed output shapes)", () => { · 761 unmodified lines |
| 783 | 786 | }), |
| 784 | 787 | ); |
| 785 | 788 | }); |
| 789 | + |
| 790 | +// --------------------------------------------------------------------------- |
| 791 | +// Dynamic tool-call read concurrency. The invoke path runs its independent |
| 792 | +// storage reads concurrently: tool row + policy rules + connection row before |
| 793 | +// approval, then credential resolution + integration row after approval. |
| 794 | +// These tests pin the overlap itself AND the invariants the overlap must not |
| 795 | +// disturb — most importantly that a declined approval never starts credential |
| 796 | +// resolution (which can trigger an upstream token refresh). |
| 797 | +// --------------------------------------------------------------------------- |
| 798 | + |
| 799 | +/** Records which armed reads are simultaneously in flight. Each armed read |
| 800 | + * registers on entry, then HOLDS its result until every other armed read has |
| 801 | + * also started — so genuinely concurrent reads all overlap and the probe |
| 802 | + * observes it. A sequential ordering would deadlock on that barrier, so a |
| 803 | + * fallback timer releases each held read; the run then completes with |
| 804 | + * `sawAllInFlight() === false` and the assertion (not a hang) reports the |
| 805 | + * regression. */ |
| 806 | +const makeReadOverlapProbe = () => { |
| 807 | + let targets: ReadonlySet<string> = new Set(); |
| 808 | + let sawAll = false; |
| 809 | + const started = new Set<string>(); |
| 810 | + const inFlight = new Set<string>(); |
| 811 | + let release: (() => void) | undefined; |
| 812 | + let barrier: Promise<void> | undefined; |
| 813 | + const observe = async (key: string): Promise<void> => { |
| 814 | + if (barrier === undefined || !targets.has(key) || started.has(key)) return; |
| 815 | + started.add(key); |
| 816 | + inFlight.add(key); |
| 817 | + if (started.size === targets.size) { |
| 818 | + if (inFlight.size === targets.size) sawAll = true; |
| 819 | + release?.(); |
| 820 | + } |
| 821 | + await Promise.race([ |
| 822 | + barrier, |
| 823 | + new Promise<void>((resolve) => { |
| 824 | + setTimeout(resolve, 250); |
| 825 | + }), |
| 826 | + ]); |
| 827 | + inFlight.delete(key); |
| 828 | + }; |
| 829 | + return { |
| 830 | + arm: (keys: readonly string[]) => { |
| 831 | + targets = new Set(keys); |
| 832 | + barrier = new Promise<void>((resolve) => { |
| 833 | + release = resolve; |
| 834 | + }); |
| 835 | + }, |
| 836 | + observe, |
| 837 | + sawAllInFlight: () => sawAll, |
| 838 | + }; |
| 839 | +}; |
| 840 | + |
| 841 | +/** Wrap a test `FumaDb` so every read reports to the probe as |
| 842 | + * `<table>.<method>` before it executes. `withContext` re-wraps so the |
| 843 | + * executor's context-bound handles stay observed. */ |
| 844 | +const withObservedReads = (db: FumaDb, observe: (key: string) => Promise<void>): FumaDb => { |
| 845 | + const wrap = (inner: FumaDb): FumaDb => |
| 846 | + new Proxy(inner, { |
| 847 | + get(target, prop) { |
| 848 | + if (prop === "withContext") { |
| 849 | + return (context: unknown) => |
| 850 | + wrap((target.withContext as (c: unknown) => FumaDb)(context)); |
| 851 | + } |
| 852 | + if (prop === "findFirst" || prop === "findMany") { |
| 853 | + return async (table: unknown, query: unknown) => { |
| 854 | + await observe(`${String(table)}.${prop}`); |
| 855 | + return (Reflect.get(target, prop) as (t: unknown, q: unknown) => Promise<unknown>).call( |
| 856 | + target, |
| 857 | + table, |
| 858 | + query, |
| 859 | + ); |
| 860 | + }; |
| 861 | + } |
| 862 | + return Reflect.get(target, prop); |
| 863 | + }, |
| 864 | + }); |
| 865 | + return wrap(db); |
| 866 | +}; |
| 867 | + |
| 868 | +const countingProvider = (calls: { count: number }, onGet?: () => Promise<void>) => { |
| 869 | + const store = new Map<string, string>(); |
| 870 | + const provider: CredentialProvider = { |
| 871 | + key: ProviderKey.make("memory"), |
| 872 | + writable: true, |
| 873 | + get: (id) => |
| 874 | + Effect.promise(async () => { |
| 875 | + calls.count++; |
| 876 | + await onGet?.(); |
| 877 | + return store.get(String(id)) ?? null; |
| 878 | + }), |
| 879 | + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), |
| 880 | + }; |
| 881 | + return provider; |
| 882 | +}; |
| 883 | + |
| 884 | +const invokeConcurrencyPlugin = (provider: CredentialProvider) => |
| 885 | + definePlugin(() => ({ |
| 886 | + id: "demo" as const, |
| 887 | + credentialProviders: [provider], |
| 888 | + storage: () => ({}), |
| 889 | + resolveTools: () => |
| 890 | + Effect.succeed({ tools: [{ name: ToolName.make("run"), description: "run" }] }), |
| 891 | + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), |
| 892 | + extension: (ctx) => ({ |
| 893 | + seed: () => ctx.core.integrations.register({ slug: INTEG, description: "Demo", config: {} }), |
| 894 | + }), |
| 895 | + }))(); |
| 896 | + |
| 897 | +const seedRunConnection = < |
| 898 | + E extends { |
| 899 | + readonly demo: { readonly seed: () => Effect.Effect<unknown, unknown> }; |
| 900 | + readonly connections: { |
| 901 | + readonly create: (input: { |
| 902 | + owner: "org"; |
| 903 | + name: ConnectionName; |
| 904 | + integration: IntegrationSlug; |
| 905 | + template: AuthTemplateSlug; |
| 906 | + from: { provider: ProviderKey; id: ProviderItemId }; |
| 907 | + }) => Effect.Effect<unknown, unknown>; |
| 908 | + }; |
| 909 | + }, |
| 910 | +>( |
| 911 | + executor: E, |
| 912 | +) => |
| 913 | + Effect.gen(function* () { |
| 914 | + yield* executor.demo.seed(); |
| 915 | + yield* executor.connections.create({ |
| 916 | + owner: "org", |
| 917 | + name: CONN, |
| 918 | + integration: INTEG, |
| 919 | + template: TEMPLATE, |
| 920 | + from: { |
| 921 | + provider: ProviderKey.make("memory"), |
| 922 | + id: ProviderItemId.make("v"), |
| 923 | + }, |
| 924 | + }); |
| 925 | + }); |
| 926 | + |
| 927 | +describe("execute read concurrency", () => { |
| 928 | + it.effect("overlaps the tool, policy, and connection reads on the invoke path", () => |
| 929 | + Effect.gen(function* () { |
| 930 | + const probe = makeReadOverlapProbe(); |
| 931 | + const config = makeTestConfig({ plugins: [demoPlugin] as const }); |
| 932 | + const executor = yield* createExecutor({ |
| 933 | + ...config, |
| 934 | + db: withObservedReads(config.db, probe.observe), |
| 935 | + }); |
| 936 | + yield* seedRunConnection(executor); |
| 937 | + |
| 938 | + probe.arm(["tool.findFirst", "tool_policy.findMany", "connection.findFirst"]); |
| 939 | + const out = yield* executor.execute(addr("run"), {}); |
| 940 | + expect(out).toEqual({ ran: "run" }); |
| 941 | + // All three pre-approval reads were in flight at the same instant. |
| 942 | + expect(probe.sawAllInFlight()).toBe(true); |
| 943 | + }), |
| 944 | + ); |
| 945 | + |
| 946 | + it.effect("overlaps credential resolution with the integration-row read after approval", () => |
| 947 | + Effect.gen(function* () { |
| 948 | + const probe = makeReadOverlapProbe(); |
| 949 | + const calls = { count: 0 }; |
| 950 | + const provider = countingProvider(calls, () => probe.observe("credential.get")); |
| 951 | + const config = makeTestConfig({ |
| 952 | + plugins: [invokeConcurrencyPlugin(provider)] as const, |
| 953 | + }); |
| 954 | + const executor = yield* createExecutor({ |
| 955 | + ...config, |
| 956 | + db: withObservedReads(config.db, probe.observe), |
| 957 | + }); |
| 958 | + yield* seedRunConnection(executor); |
| 959 | + |
| 960 | + probe.arm(["credential.get", "integration.findFirst"]); |
| 961 | + const out = yield* executor.execute(addr("run"), {}); |
| 962 | + expect(out).toEqual({ ran: "run" }); |
| 963 | + expect(probe.sawAllInFlight()).toBe(true); |
| 964 | + }), |
| 965 | + ); |
| 966 | + |
| 967 | + it.effect("a declined approval never resolves credentials", () => |
| 968 | + Effect.gen(function* () { |
| 969 | + const calls = { count: 0 }; |
| 970 | + const executor = yield* makeTestExecutor({ |
| 971 | + plugins: [invokeConcurrencyPlugin(countingProvider(calls))] as const, |
| 972 | + }); |
| 973 | + yield* seedRunConnection(executor); |
| 974 | + yield* executor.policies.create({ |
| 975 | + owner: "org", |
| 976 | + pattern: "demo.*", |
| 977 | + action: "require_approval", |
| 978 | + }); |
| 979 | + |
| 980 | + // Setup (connection create / tool sync) may read the credential; only |
| 981 | + // reads issued by the declined call itself are the regression signal. |
| 982 | + calls.count = 0; |
| 983 | + const decliningHandler: ElicitationHandler = () => |
| 984 | + Effect.succeed(ElicitationResponse.make({ action: "decline" })); |
| 985 | + const result = yield* Effect.result( |
| 986 | + executor.execute(addr("run"), {}, { onElicitation: decliningHandler }), |
| 987 | + ); |
| 988 | + expect(Result.isFailure(result)).toBe(true); |
| 989 | + if (!Result.isFailure(result)) return; |
| 990 | + expect(Predicate.isTagged("ElicitationDeclinedError")(result.failure)).toBe(true); |
| 991 | + // The decline happened BEFORE credential resolution started: a token |
| 992 | + // refresh (a network side effect) must never fire for a declined call. |
| 993 | + expect(calls.count).toBe(0); |
| 994 | + }), |
| 995 | + ); |
| 996 | + |
| 997 | + it.effect("fails with ConnectionNotFoundError when the tool row outlives its connection", () => |
| 998 | + Effect.gen(function* () { |
| 999 | + const config = makeTestConfig({ plugins: [demoPlugin] as const }); |
| 1000 | + const executor = yield* createExecutor(config); |
| 1001 | + yield* seedRunConnection(executor); |
| 1002 | + |
| 1003 | + // Remove ONLY the connection row, leaving the tool rows behind — the |
| 1004 | + // inconsistent state the ConnectionNotFoundError branch reports. The |
| 1005 | + // concurrent connection read must still surface this error, not a |
| 1006 | + // policy or tool-row failure. |
| 1007 | + yield* Effect.promise(() => config.db.deleteMany("connection", {})); |
| 1008 | + |
| 1009 | + const result = yield* Effect.result(executor.execute(addr("run"), {})); |
| 1010 | + expect(Result.isFailure(result)).toBe(true); |
| 1011 | + if (!Result.isFailure(result)) return; |
| 1012 | + expect(Predicate.isTagged("ConnectionNotFoundError")(result.failure)).toBe(true); |
| 1013 | + }), |
| 1014 | + ); |
| 1015 | +}); |