1import type { ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as PubSub from "effect/PubSub";
6import type * as Scope from "effect/Scope";
7
8export interface PreviewAgentTurnChange {
9 readonly threadId: ThreadId;
10 readonly active: boolean;
11}
12
13/**
14 * Whether an agent turn runs on a thread, model thinking included, so its preview tab
15 * stays focused between tool calls. Background commands outlive the turn and do not count.
16 */
17export class PreviewAgentActivity extends Context.Service<
18 PreviewAgentActivity,
19 {
20 /** Marks the thread active until the returned effect runs; it runs at most once. */
21 readonly begin: (threadId: ThreadId) => Effect.Effect<Effect.Effect<void>>;
22 readonly active: (threadId: ThreadId) => Effect.Effect<boolean>;
23 /** Each thread's transitions between no turn and some turn, from subscription on. */
24 readonly changes: Effect.Effect<
25 PubSub.Subscription<PreviewAgentTurnChange>,
26 never,
27 Scope.Scope
28 >;
29 }
30>()("t3/preview/PreviewAgentActivity") {}
31
32const make = Effect.gen(function* () {
33 const changes = yield* PubSub.unbounded<PreviewAgentTurnChange>();
34 // A token per turn, so overlapping turns on one thread end only when the last does.
35 const turns = new Map<ThreadId, Set<symbol>>();
36
37 const begin = Effect.fn("PreviewAgentActivity.begin")(function* (threadId: ThreadId) {
38 const existing = turns.get(threadId);
39 const tokens = existing ?? new Set<symbol>();
40 if (existing === undefined) turns.set(threadId, tokens);
41 const token = Symbol(threadId);
42 tokens.add(token);
43 if (existing === undefined) yield* PubSub.publish(changes, { threadId, active: true });
44 return Effect.suspend(() => {
45 if (!tokens.delete(token) || tokens.size > 0) return Effect.void;
46 if (turns.get(threadId) === tokens) turns.delete(threadId);
47 return PubSub.publish(changes, { threadId, active: false }).pipe(Effect.asVoid);
48 });
49 });
50
51 return PreviewAgentActivity.of({
52 begin,
53 active: Effect.fn("PreviewAgentActivity.active")(function (threadId) {
54 return Effect.sync(() => turns.has(threadId));
55 }),
56 changes: PubSub.subscribe(changes),
57 });
58});
59
60export const layer = Layer.effect(PreviewAgentActivity, make);