1import { describe, expect, it } from "@effect/vitest";
2import { ProviderInstanceId } from "@t3tools/contracts";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";
6import * as Option from "effect/Option";
7import * as TestClock from "effect/testing/TestClock";
8
9import type { AcpRegistryAvailableCommands, AcpRegistryLiveConfiguration } from "./probe.ts";
10import * as AcpRegistryRuntimeCoordinator from "./AcpRegistryRuntimeCoordinator.ts";
11
12describe("AcpRegistryRuntimeCoordinator", () => {
13 it.effect("suppresses a background probe while foreground startup is active", () =>
14 Effect.gen(function* () {
15 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
16 const foregroundStarted = yield* Deferred.make<void>();
17 const releaseForeground = yield* Deferred.make<void>();
18 const foreground = yield* coordinator
19 .withForegroundStartup(
20 "kilo",
21 Deferred.succeed(foregroundStarted, undefined).pipe(
22 Effect.andThen(Deferred.await(releaseForeground)),
23 ),
24 )
25 .pipe(Effect.forkChild);
26 yield* Deferred.await(foregroundStarted);
27
28 const probed = yield* coordinator.runBackgroundProbe("kilo", Effect.succeed("unexpected"));
29 expect(Option.isNone(probed)).toBe(true);
30
31 yield* Deferred.succeed(releaseForeground, undefined);
32 yield* Fiber.join(foreground);
33 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
34 );
35
36 it.effect("interrupts an active background probe when foreground startup begins", () =>
37 Effect.gen(function* () {
38 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
39 const probeStarted = yield* Deferred.make<void>();
40 const probeFinalized = yield* Deferred.make<void>();
41 const releaseForeground = yield* Deferred.make<void>();
42 const probe = yield* coordinator
43 .runBackgroundProbe(
44 "kilo",
45 Deferred.succeed(probeStarted, undefined).pipe(
46 Effect.andThen(Effect.never),
47 Effect.ensuring(Deferred.succeed(probeFinalized, undefined).pipe(Effect.ignore)),
48 ),
49 )
50 .pipe(Effect.forkChild);
51 yield* Deferred.await(probeStarted);
52
53 const foreground = yield* coordinator
54 .withForegroundStartup("kilo", Deferred.await(releaseForeground))
55 .pipe(Effect.forkChild);
56 expect(Option.isNone(yield* Fiber.join(probe))).toBe(true);
57 yield* Deferred.await(probeFinalized);
58
59 yield* Deferred.succeed(releaseForeground, undefined);
60 yield* Fiber.join(foreground);
61 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
62 );
63
64 it.effect("serializes native session mutations across callers", () =>
65 Effect.gen(function* () {
66 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
67 const firstEntered = yield* Deferred.make<void>();
68 const releaseFirst = yield* Deferred.make<void>();
69 const secondEntered = yield* Deferred.make<void>();
70 const first = yield* coordinator
71 .withSessionMutation(
72 Deferred.succeed(firstEntered, undefined).pipe(
73 Effect.andThen(Deferred.await(releaseFirst)),
74 ),
75 )
76 .pipe(Effect.forkChild);
77 yield* Deferred.await(firstEntered);
78
79 const second = yield* coordinator
80 .withSessionMutation(Deferred.succeed(secondEntered, undefined))
81 .pipe(Effect.forkChild);
82 expect(yield* Deferred.poll(secondEntered)).toEqual(Option.none());
83
84 yield* Deferred.succeed(releaseFirst, undefined);
85 yield* Fiber.join(first);
86 yield* Fiber.join(second);
87 expect(Option.isSome(yield* Deferred.poll(secondEntered))).toBe(true);
88 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
89 );
90
91 it.effect("replays and replaces late command advertisements per provider instance", () =>
92 Effect.gen(function* () {
93 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
94 const codex = ProviderInstanceId.make("acpRegistry_codex");
95 const kilo = ProviderInstanceId.make("acpRegistry_kilo");
96 const firstSeen = yield* Deferred.make<void>();
97 const replacementSeen = yield* Deferred.make<void>();
98 const received: Array<AcpRegistryAvailableCommands> = [];
99
100 yield* coordinator.publishAvailableCommands(codex, {
101 slashCommands: [{ name: "status" }],
102 skills: [{ name: "workspace-skill", path: "acp://skill/workspace-skill", enabled: true }],
103 });
104 const consumer = yield* coordinator
105 .watchAvailableCommands(codex, (commands) =>
106 Effect.gen(function* () {
107 received.push(commands);
108 yield* received.length === 1
109 ? Deferred.succeed(firstSeen, undefined)
110 : Deferred.succeed(replacementSeen, undefined);
111 }),
112 )
113 .pipe(Effect.forkChild);
114 yield* Deferred.await(firstSeen);
115
116 yield* coordinator.publishAvailableCommands(kilo, {
117 slashCommands: [{ name: "review" }],
118 skills: [],
119 });
120 yield* coordinator.publishAvailableCommands(codex, { slashCommands: [], skills: [] });
121 yield* Deferred.await(replacementSeen);
122 yield* Fiber.interrupt(consumer);
123
124 expect(received).toEqual([
125 {
126 slashCommands: [{ name: "status" }],
127 skills: [{ name: "workspace-skill", path: "acp://skill/workspace-skill", enabled: true }],
128 },
129 { slashCommands: [], skills: [] },
130 ]);
131 expect(yield* coordinator.getAvailableCommands(kilo)).toEqual(
132 Option.some({ slashCommands: [{ name: "review" }], skills: [] }),
133 );
134 yield* coordinator.clearAvailableCommands(codex);
135 expect(Option.isNone(yield* coordinator.getAvailableCommands(codex))).toBe(true);
136 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
137 );
138
139 it.effect("replays and replaces live configuration per provider instance", () =>
140 Effect.gen(function* () {
141 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
142 const instanceId = ProviderInstanceId.make("acpRegistry_kilo");
143 const firstSeen = yield* Deferred.make<void>();
144 const replacementSeen = yield* Deferred.make<void>();
145 const received: Array<AcpRegistryLiveConfiguration> = [];
146 const first = {
147 models: [{ id: "sonnet", name: "Sonnet", description: null }],
148 currentModelId: "sonnet",
149 configOptions: [],
150 } satisfies AcpRegistryLiveConfiguration;
151 const replacement = {
152 models: [{ id: "opus", name: "Opus", description: null }],
153 currentModelId: "opus",
154 configOptions: [],
155 } satisfies AcpRegistryLiveConfiguration;
156
157 yield* coordinator.publishLiveConfiguration(instanceId, first);
158 const consumer = yield* coordinator
159 .watchLiveConfiguration(instanceId, (configuration) =>
160 Effect.gen(function* () {
161 received.push(configuration);
162 yield* received.length === 1
163 ? Deferred.succeed(firstSeen, undefined)
164 : Deferred.succeed(replacementSeen, undefined);
165 }),
166 )
167 .pipe(Effect.forkChild);
168 yield* Deferred.await(firstSeen);
169 yield* coordinator.publishLiveConfiguration(instanceId, replacement);
170 yield* Deferred.await(replacementSeen);
171 yield* Fiber.interrupt(consumer);
172
173 expect(received).toEqual([first, replacement]);
174 expect(yield* coordinator.getLiveConfiguration(instanceId)).toEqual(Option.some(replacement));
175 yield* coordinator.clearLiveConfiguration(instanceId);
176 expect(Option.isNone(yield* coordinator.getLiveConfiguration(instanceId))).toBe(true);
177 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
178 );
179
180 it.effect("requires matching user consent before accepting URL authentication", () =>
181 Effect.gen(function* () {
182 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
183 const instanceId = ProviderInstanceId.make("acpRegistry_antigravity");
184 const action = {
185 elicitationId: "login-1",
186 url: "https://accounts.example.com/login",
187 message: "Continue in your browser",
188 };
189 const actionSeen = yield* Deferred.make<void>();
190 const consumer = yield* coordinator
191 .watchUrlAuthAction(instanceId, (current) =>
192 current === null
193 ? Effect.void
194 : Deferred.succeed(actionSeen, undefined).pipe(Effect.asVoid),
195 )
196 .pipe(Effect.forkChild);
197 const request = yield* coordinator
198 .requestUrlAuthentication(instanceId, action)
199 .pipe(Effect.forkChild);
200 yield* Deferred.await(actionSeen);
201
202 expect(yield* coordinator.getUrlAuthAction(instanceId)).toEqual(
203 Option.some({
204 ...action,
205 createdAt: "1970-01-01T00:00:00.000Z",
206 expiresAt: "1970-01-01T00:10:00.000Z",
207 }),
208 );
209
210 expect(
211 yield* coordinator.acceptUrlAuthentication({ instanceId, elicitationId: "stale" }),
212 ).toBe(false);
213 expect(
214 yield* coordinator.acceptUrlAuthentication({ instanceId, elicitationId: "login-1" }),
215 ).toBe(true);
216 expect(yield* Fiber.join(request)).toBe(true);
217 expect(Option.isNone(yield* coordinator.getUrlAuthAction(instanceId))).toBe(true);
218 yield* Fiber.interrupt(consumer);
219 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
220 );
221
222 it.effect("expires stale URL authentication actions without accepting them", () =>
223 Effect.gen(function* () {
224 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
225 const instanceId = ProviderInstanceId.make("acpRegistry_expired");
226 const actionSeen = yield* Deferred.make<void>();
227 const consumer = yield* coordinator
228 .watchUrlAuthAction(instanceId, (current) =>
229 current === null
230 ? Effect.void
231 : Deferred.succeed(actionSeen, undefined).pipe(Effect.asVoid),
232 )
233 .pipe(Effect.forkChild);
234 const request = yield* coordinator
235 .requestUrlAuthentication(instanceId, {
236 elicitationId: "expired-login",
237 url: "https://accounts.example.com/login",
238 message: "Continue in your browser",
239 })
240 .pipe(Effect.forkChild);
241 yield* Deferred.await(actionSeen);
242 yield* TestClock.adjust("10 minutes");
243
244 expect(yield* Fiber.join(request)).toBe(false);
245 expect(
246 yield* coordinator.acceptUrlAuthentication({
247 instanceId,
248 elicitationId: "expired-login",
249 }),
250 ).toBe(false);
251 expect(Option.isNone(yield* coordinator.getUrlAuthAction(instanceId))).toBe(true);
252 yield* Fiber.interrupt(consumer);
253 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
254 );
255
256 it.effect("publishes a replacement URL action before retiring the previous request", () =>
257 Effect.gen(function* () {
258 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
259 const instanceId = ProviderInstanceId.make("acpRegistry_replacement");
260 const received: Array<string | null> = [];
261 const firstSeen = yield* Deferred.make<void>();
262 const replacementSeen = yield* Deferred.make<void>();
263 const consumer = yield* coordinator
264 .watchUrlAuthAction(instanceId, (current) =>
265 Effect.sync(() => {
266 received.push(current?.elicitationId ?? null);
267 }).pipe(
268 Effect.andThen(
269 current?.elicitationId === "login-1"
270 ? Deferred.succeed(firstSeen, undefined).pipe(Effect.asVoid)
271 : current?.elicitationId === "login-2"
272 ? Deferred.succeed(replacementSeen, undefined).pipe(Effect.asVoid)
273 : Effect.void,
274 ),
275 ),
276 )
277 .pipe(Effect.forkChild);
278 const first = yield* coordinator
279 .requestUrlAuthentication(instanceId, {
280 elicitationId: "login-1",
281 url: "https://accounts.example.com/first",
282 message: "Continue with the first login",
283 })
284 .pipe(Effect.forkChild);
285 yield* Deferred.await(firstSeen);
286 const replacement = yield* coordinator
287 .requestUrlAuthentication(instanceId, {
288 elicitationId: "login-2",
289 url: "https://accounts.example.com/replacement",
290 message: "Continue with the replacement login",
291 })
292 .pipe(Effect.forkChild);
293 yield* Deferred.await(replacementSeen);
294
295 expect(yield* Fiber.join(first)).toBe(false);
296 expect(
297 (yield* coordinator.getUrlAuthAction(instanceId)).pipe(Option.getOrThrow),
298 ).toMatchObject({ elicitationId: "login-2" });
299 expect(received.slice(0, 2)).toEqual(["login-1", "login-2"]);
300 expect(
301 yield* coordinator.acceptUrlAuthentication({ instanceId, elicitationId: "login-2" }),
302 ).toBe(true);
303 expect(yield* Fiber.join(replacement)).toBe(true);
304 yield* Fiber.interrupt(consumer);
305 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
306 );
307
308 it.effect("clears a published URL action when its request is interrupted", () =>
309 Effect.gen(function* () {
310 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
311 const instanceId = ProviderInstanceId.make("acpRegistry_interrupted");
312 const actionSeen = yield* Deferred.make<void>();
313 const cleared = yield* Deferred.make<void>();
314 const consumer = yield* coordinator
315 .watchUrlAuthAction(instanceId, (current) =>
316 current === null
317 ? Deferred.succeed(cleared, undefined).pipe(Effect.asVoid)
318 : Deferred.succeed(actionSeen, undefined).pipe(Effect.asVoid),
319 )
320 .pipe(Effect.forkChild);
321 const request = yield* coordinator
322 .requestUrlAuthentication(instanceId, {
323 elicitationId: "interrupted-login",
324 url: "https://accounts.example.com/interrupted",
325 message: "Continue with login",
326 })
327 .pipe(Effect.forkChild);
328 yield* Deferred.await(actionSeen);
329
330 yield* Fiber.interrupt(request);
331 yield* Deferred.await(cleared);
332 expect(Option.isNone(yield* coordinator.getUrlAuthAction(instanceId))).toBe(true);
333 yield* Fiber.interrupt(consumer);
334 }).pipe(Effect.provide(AcpRegistryRuntimeCoordinator.layer), Effect.scoped),
335 );
336});