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