Open History to see the last commit
1import {
2 type AcpRegistryAcceptUrlAuthInput,
3 type AcpRegistryUrlAuthAction,
4 type ProviderInstanceId,
5} from "@t3tools/contracts";
6import * as Context from "effect/Context";
7import * as Deferred from "effect/Deferred";
8import * as DateTime from "effect/DateTime";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as PubSub from "effect/PubSub";
14import * as Ref from "effect/Ref";
15import * as Semaphore from "effect/Semaphore";
16import * as Stream from "effect/Stream";
17
18import type { AcpRegistryAvailableCommands, AcpRegistryLiveConfiguration } from "./probe.ts";
19
20interface AvailableCommandsUpdate {
21 readonly instanceId: ProviderInstanceId;
22 readonly commands: AcpRegistryAvailableCommands;
23}
24
25interface LiveConfigurationUpdate {
26 readonly instanceId: ProviderInstanceId;
27 readonly configuration: AcpRegistryLiveConfiguration;
28}
29
30interface UrlAuthActionUpdate {
31 readonly instanceId: ProviderInstanceId;
32 readonly action: AcpRegistryUrlAuthAction | null;
33}
34
35const URL_AUTH_ACTION_TTL = Duration.minutes(10);
36
37interface PendingUrlAuthAction {
38 readonly action: AcpRegistryUrlAuthAction;
39 readonly consent: Deferred.Deferred<boolean>;
40 readonly expiresAt: DateTime.Utc;
41}
42
43export class AcpRegistryRuntimeCoordinator extends Context.Service<
44 AcpRegistryRuntimeCoordinator,
45 {
46 readonly withForegroundStartup: <A, E, R>(
47 agentId: string,
48 effect: Effect.Effect<A, E, R>,
49 ) => Effect.Effect<A, E, R>;
50 readonly runBackgroundProbe: <A, E, R>(
51 agentId: string,
52 effect: Effect.Effect<A, E, R>,
53 ) => Effect.Effect<Option.Option<A>, E, R>;
54 readonly withSessionMutation: <A, E, R>(
55 effect: Effect.Effect<A, E, R>,
56 ) => Effect.Effect<A, E, R>;
57 readonly clearAvailableCommands: (instanceId: ProviderInstanceId) => Effect.Effect<void>;
58 readonly publishAvailableCommands: (
59 instanceId: ProviderInstanceId,
60 commands: AcpRegistryAvailableCommands,
61 ) => Effect.Effect<void>;
62 readonly getAvailableCommands: (
63 instanceId: ProviderInstanceId,
64 ) => Effect.Effect<Option.Option<AcpRegistryAvailableCommands>>;
65 readonly watchAvailableCommands: (
66 instanceId: ProviderInstanceId,
67 onUpdate: (commands: AcpRegistryAvailableCommands) => Effect.Effect<void>,
68 ) => Effect.Effect<void>;
69 readonly clearLiveConfiguration: (instanceId: ProviderInstanceId) => Effect.Effect<void>;
70 readonly publishLiveConfiguration: (
71 instanceId: ProviderInstanceId,
72 configuration: AcpRegistryLiveConfiguration,
73 ) => Effect.Effect<void>;
74 readonly getLiveConfiguration: (
75 instanceId: ProviderInstanceId,
76 ) => Effect.Effect<Option.Option<AcpRegistryLiveConfiguration>>;
77 readonly watchLiveConfiguration: (
78 instanceId: ProviderInstanceId,
79 onUpdate: (configuration: AcpRegistryLiveConfiguration) => Effect.Effect<void>,
80 ) => Effect.Effect<void>;
81 readonly requestUrlAuthentication: (
82 instanceId: ProviderInstanceId,
83 action: AcpRegistryUrlAuthAction,
84 ) => Effect.Effect<boolean>;
85 readonly acceptUrlAuthentication: (
86 input: AcpRegistryAcceptUrlAuthInput,
87 ) => Effect.Effect<boolean>;
88 readonly getUrlAuthAction: (
89 instanceId: ProviderInstanceId,
90 ) => Effect.Effect<Option.Option<AcpRegistryUrlAuthAction>>;
91 readonly watchUrlAuthAction: (
92 instanceId: ProviderInstanceId,
93 onUpdate: (action: AcpRegistryUrlAuthAction | null) => Effect.Effect<void>,
94 ) => Effect.Effect<void>;
95 }
96>()("@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator") {}
97
98const make = Effect.gen(function* () {
99 const foregroundStarts = yield* PubSub.unbounded<string>();
100 const activeForegroundCounts = yield* Ref.make(new Map<string, number>());
101 const availableCommandsUpdates = yield* PubSub.unbounded<AvailableCommandsUpdate>();
102 const availableCommandsByInstance = yield* Ref.make(
103 new Map<ProviderInstanceId, AcpRegistryAvailableCommands>(),
104 );
105 const liveConfigurationUpdates = yield* PubSub.unbounded<LiveConfigurationUpdate>();
106 const liveConfigurationByInstance = yield* Ref.make(
107 new Map<ProviderInstanceId, AcpRegistryLiveConfiguration>(),
108 );
109 const urlAuthActionUpdates = yield* PubSub.unbounded<UrlAuthActionUpdate>();
110 const pendingUrlAuthActions = yield* Ref.make(
111 new Map<ProviderInstanceId, PendingUrlAuthAction>(),
112 );
113 const urlAuthActionPermit = yield* Semaphore.make(1);
114 const sessionMutationPermit = yield* Semaphore.make(1);
115
116 const withForegroundStartup: AcpRegistryRuntimeCoordinator["Service"]["withForegroundStartup"] = (
117 agentId,
118 effect,
119 ) =>
120 Effect.acquireUseRelease(
121 Ref.update(activeForegroundCounts, (current) => {
122 const next = new Map(current);
123 next.set(agentId, (next.get(agentId) ?? 0) + 1);
124 return next;
125 }).pipe(Effect.andThen(PubSub.publish(foregroundStarts, agentId))),
126 () => effect,
127 () =>
128 Ref.update(activeForegroundCounts, (current) => {
129 const next = new Map(current);
130 const remaining = (next.get(agentId) ?? 1) - 1;
131 if (remaining <= 0) next.delete(agentId);
132 else next.set(agentId, remaining);
133 return next;
134 }),
135 );
136
137 const runBackgroundProbe: AcpRegistryRuntimeCoordinator["Service"]["runBackgroundProbe"] = (
138 agentId,
139 effect,
140 ) =>
141 Effect.scoped(
142 Effect.gen(function* () {
143 // Subscribe first so foreground starts cannot slip between the state check and probe.
144 const subscription = yield* PubSub.subscribe(foregroundStarts);
145 if ((yield* Ref.get(activeForegroundCounts)).has(agentId)) {
146 return Option.none();
147 }
148 const foregroundStarted = Stream.fromSubscription(subscription).pipe(
149 Stream.filter((candidate) => candidate === agentId),
150 Stream.runHead,
151 Effect.as(Option.none()),
152 );
153 return yield* Effect.raceFirst(effect.pipe(Effect.asSome), foregroundStarted);
154 }),
155 );
156
157 const withSessionMutation: AcpRegistryRuntimeCoordinator["Service"]["withSessionMutation"] = (
158 effect,
159 ) => sessionMutationPermit.withPermit(effect);
160
161 const clearAvailableCommands: AcpRegistryRuntimeCoordinator["Service"]["clearAvailableCommands"] =
162 (instanceId) =>
163 Ref.update(availableCommandsByInstance, (current) => {
164 const next = new Map(current);
165 next.delete(instanceId);
166 return next;
167 });
168
169 const publishAvailableCommands: AcpRegistryRuntimeCoordinator["Service"]["publishAvailableCommands"] =
170 (instanceId, commands) =>
171 Ref.update(availableCommandsByInstance, (current) => {
172 const next = new Map(current);
173 next.set(instanceId, commands);
174 return next;
175 }).pipe(
176 Effect.andThen(
177 PubSub.publish(availableCommandsUpdates, {
178 instanceId,
179 commands,
180 }),
181 ),
182 Effect.asVoid,
183 );
184
185 const getAvailableCommands: AcpRegistryRuntimeCoordinator["Service"]["getAvailableCommands"] =
186 Effect.fn("AcpRegistryRuntimeCoordinator.getAvailableCommands")(function* (instanceId) {
187 return Option.fromNullishOr((yield* Ref.get(availableCommandsByInstance)).get(instanceId));
188 });
189
190 const watchAvailableCommands: AcpRegistryRuntimeCoordinator["Service"]["watchAvailableCommands"] =
191 (instanceId, onUpdate) =>
192 Effect.scoped(
193 Effect.gen(function* () {
194 // Subscribe before reading to avoid missing publications.
195 const subscription = yield* PubSub.subscribe(availableCommandsUpdates);
196 const current = yield* getAvailableCommands(instanceId);
197 if (Option.isSome(current)) {
198 yield* onUpdate(current.value);
199 }
200 yield* Stream.fromSubscription(subscription).pipe(
201 Stream.filter((update) => update.instanceId === instanceId),
202 Stream.runForEach((update) => onUpdate(update.commands)),
203 );
204 }),
205 );
206
207 const clearLiveConfiguration: AcpRegistryRuntimeCoordinator["Service"]["clearLiveConfiguration"] =
208 (instanceId) =>
209 Ref.update(liveConfigurationByInstance, (current) => {
210 const next = new Map(current);
211 next.delete(instanceId);
212 return next;
213 });
214
215 const publishLiveConfiguration: AcpRegistryRuntimeCoordinator["Service"]["publishLiveConfiguration"] =
216 (instanceId, configuration) =>
217 Ref.update(liveConfigurationByInstance, (current) =>
218 new Map(current).set(instanceId, configuration),
219 ).pipe(
220 Effect.andThen(
221 PubSub.publish(liveConfigurationUpdates, {
222 instanceId,
223 configuration,
224 }),
225 ),
226 Effect.asVoid,
227 );
228
229 const getLiveConfiguration: AcpRegistryRuntimeCoordinator["Service"]["getLiveConfiguration"] =
230 Effect.fn("AcpRegistryRuntimeCoordinator.getLiveConfiguration")(function* (instanceId) {
231 return Option.fromNullishOr((yield* Ref.get(liveConfigurationByInstance)).get(instanceId));
232 });
233
234 const watchLiveConfiguration: AcpRegistryRuntimeCoordinator["Service"]["watchLiveConfiguration"] =
235 (instanceId, onUpdate) =>
236 Effect.scoped(
237 Effect.gen(function* () {
238 const subscription = yield* PubSub.subscribe(liveConfigurationUpdates);
239 const current = yield* getLiveConfiguration(instanceId);
240 if (Option.isSome(current)) {
241 yield* onUpdate(current.value);
242 }
243 yield* Stream.fromSubscription(subscription).pipe(
244 Stream.filter((update) => update.instanceId === instanceId),
245 Stream.runForEach((update) => onUpdate(update.configuration)),
246 );
247 }),
248 );
249
250 const publishUrlAuthAction = (
251 instanceId: ProviderInstanceId,
252 action: AcpRegistryUrlAuthAction | null,
253 ) => PubSub.publish(urlAuthActionUpdates, { instanceId, action }).pipe(Effect.asVoid);
254
255 const requestUrlAuthentication: AcpRegistryRuntimeCoordinator["Service"]["requestUrlAuthentication"] =
256 Effect.fn("AcpRegistryRuntimeCoordinator.requestUrlAuthentication")(
257 function* (instanceId, action) {
258 const consent = yield* Deferred.make<boolean>();
259 const createdAt = yield* DateTime.now;
260 const expiresAt = DateTime.addDuration(createdAt, URL_AUTH_ACTION_TTL);
261 const publishedAction = {
262 ...action,
263 createdAt: DateTime.formatIso(createdAt),
264 expiresAt: DateTime.formatIso(expiresAt),
265 } satisfies AcpRegistryUrlAuthAction;
266 const cleanup = Ref.modify(pendingUrlAuthActions, (current) => {
267 if (current.get(instanceId)?.consent !== consent) {
268 return [false, current] as const;
269 }
270 const next = new Map(current);
271 next.delete(instanceId);
272 return [true, next] as const;
273 }).pipe(
274 Effect.flatMap((removed) =>
275 removed ? publishUrlAuthAction(instanceId, null) : Effect.void,
276 ),
277 );
278 return yield* Effect.gen(function* () {
279 yield* urlAuthActionPermit.withPermits(1)(
280 Effect.gen(function* () {
281 const previous = yield* Ref.modify(pendingUrlAuthActions, (current) => {
282 const next = new Map(current);
283 const existing = next.get(instanceId);
284 next.set(instanceId, { action: publishedAction, consent, expiresAt });
285 return [existing, next] as const;
286 });
287 if (previous !== undefined) {
288 yield* Deferred.succeed(previous.consent, false).pipe(Effect.ignore);
289 }
290 yield* publishUrlAuthAction(instanceId, publishedAction);
291 }),
292 );
293 return yield* Deferred.await(consent).pipe(
294 Effect.timeoutOrElse({
295 duration: URL_AUTH_ACTION_TTL,
296 orElse: () => Effect.succeed(false),
297 }),
298 );
299 }).pipe(Effect.ensuring(cleanup));
300 },
301 );
302
303 const acceptUrlAuthentication: AcpRegistryRuntimeCoordinator["Service"]["acceptUrlAuthentication"] =
304 Effect.fn("AcpRegistryRuntimeCoordinator.acceptUrlAuthentication")(function* (input) {
305 const pending = (yield* Ref.get(pendingUrlAuthActions)).get(input.instanceId);
306 if (pending?.action.elicitationId !== input.elicitationId) return false;
307 if (DateTime.isGreaterThanOrEqualTo(yield* DateTime.now, pending.expiresAt)) {
308 yield* Deferred.succeed(pending.consent, false).pipe(Effect.ignore);
309 return false;
310 }
311 return yield* Deferred.succeed(pending.consent, true);
312 });
313
314 const getUrlAuthAction: AcpRegistryRuntimeCoordinator["Service"]["getUrlAuthAction"] = Effect.fn(
315 "AcpRegistryRuntimeCoordinator.getUrlAuthAction",
316 )(function* (instanceId) {
317 return Option.fromNullishOr((yield* Ref.get(pendingUrlAuthActions)).get(instanceId)?.action);
318 });
319
320 const watchUrlAuthAction: AcpRegistryRuntimeCoordinator["Service"]["watchUrlAuthAction"] = (
321 instanceId,
322 onUpdate,
323 ) =>
324 Effect.scoped(
325 Effect.gen(function* () {
326 const subscription = yield* PubSub.subscribe(urlAuthActionUpdates);
327 const current = yield* getUrlAuthAction(instanceId);
328 if (Option.isSome(current)) {
329 yield* onUpdate(current.value);
330 }
331 yield* Stream.fromSubscription(subscription).pipe(
332 Stream.filter((update) => update.instanceId === instanceId),
333 Stream.runForEach((update) => onUpdate(update.action)),
334 );
335 }),
336 );
337
338 return AcpRegistryRuntimeCoordinator.of({
339 withForegroundStartup,
340 runBackgroundProbe,
341 withSessionMutation,
342 clearAvailableCommands,
343 publishAvailableCommands,
344 getAvailableCommands,
345 watchAvailableCommands,
346 clearLiveConfiguration,
347 publishLiveConfiguration,
348 getLiveConfiguration,
349 watchLiveConfiguration,
350 requestUrlAuthentication,
351 acceptUrlAuthentication,
352 getUrlAuthAction,
353 watchUrlAuthAction,
354 });
355});
356
357export const layer = Layer.effect(AcpRegistryRuntimeCoordinator, make);