1import type {
2 ThreadId,
3 WorktreeSetupSnapshot,
4 WorktreeSetupStage,
5 WorktreeSetupStageId,
6 WorktreeSetupStageStatus,
7} from "@t3tools/contracts";
8import {
9 WORKTREE_SETUP_DETAIL_MAX_LENGTH,
10 WORKTREE_SETUP_ERROR_MAX_LENGTH,
11 WORKTREE_SETUP_STAGE_ORDER,
12 WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH,
13} from "@t3tools/contracts";
14import * as Context from "effect/Context";
15import * as DateTime from "effect/DateTime";
16import * as Effect from "effect/Effect";
17import * as Fiber from "effect/Fiber";
18import * as Layer from "effect/Layer";
19import * as PubSub from "effect/PubSub";
20import * as Queue from "effect/Queue";
21import * as Ref from "effect/Ref";
22import * as Stream from "effect/Stream";
23
24/**
25 * Tracks the live stages of a bootstrap worktree setup per thread so clients
26 * can render a progress card while the first turn is still being prepared.
27 *
28 * State is memory only. It exists from the first `begin` until the turn starts
29 * or the setup fails, plus a short grace window so a client that subscribes
30 * late still sees the final state. Nothing here is persisted or event-sourced:
31 * the durable record of a setup is the thread's worktree path and the setup
32 * script activities, both of which already exist.
33 */
34export class WorktreeSetupTracker extends Context.Service<
35 WorktreeSetupTracker,
36 {
37 /** Creates a fresh running snapshot for the thread, replacing any prior one. */
38 readonly begin: (input: {
39 readonly threadId: ThreadId;
40 readonly branch: string | null;
41 readonly baseRef: string | null;
42 readonly stages: ReadonlyArray<WorktreeSetupStageId>;
43 /** Interrupting this fiber cancels the bootstrap. */
44 readonly fiber: Fiber.Fiber<unknown, unknown> | null;
45 }) => Effect.Effect<void>;
46 readonly update: (
47 threadId: ThreadId,
48 mutate: (snapshot: WorktreeSetupSnapshot) => WorktreeSetupSnapshot,
49 ) => Effect.Effect<void>;
50 readonly stage: (
51 threadId: ThreadId,
52 stageId: WorktreeSetupStageId,
53 patch: Partial<Omit<WorktreeSetupStage, "id">>,
54 ) => Effect.Effect<void>;
55 readonly stageStatus: (
56 threadId: ThreadId,
57 stageId: WorktreeSetupStageId,
58 status: WorktreeSetupStageStatus,
59 detail?: string | null,
60 ) => Effect.Effect<void>;
61 readonly appendTail: (
62 threadId: ThreadId,
63 stageId: WorktreeSetupStageId,
64 line: string,
65 ) => Effect.Effect<void>;
66 /** Returns the settled snapshot, or null when nothing was tracked. */
67 readonly finish: (
68 threadId: ThreadId,
69 phase: "done" | "failed" | "cancelled",
70 error?: string | null,
71 ) => Effect.Effect<WorktreeSetupSnapshot | null>;
72 /**
73 * Drops the cancel handle. Called right before the turn is dispatched so a
74 * late cancel cannot roll back a thread whose agent has already started.
75 */
76 readonly markUncancellable: (threadId: ThreadId) => Effect.Effect<void>;
77 /**
78 * Interrupts the running bootstrap and waits for it to unwind, so the
79 * caller's dispatch has already failed and rolled back when this returns.
80 * Returns false when nothing is running or the setup is past cancellation.
81 */
82 readonly cancel: (threadId: ThreadId) => Effect.Effect<boolean>;
83 readonly get: (threadId: ThreadId) => Effect.Effect<WorktreeSetupSnapshot | null>;
84 /** Emits the current snapshot (or null) first, then every change until unsubscribed. */
85 readonly stream: (threadId: ThreadId) => Stream.Stream<WorktreeSetupSnapshot | null>;
86 }
87>()("t3/project/WorktreeSetupTracker") {}
88
89const TAIL_LINE_LIMIT = 4;
90
91/** Keeps free text inside the contract limit, ending in an ellipsis when cut. */
92function clampText(text: string, maxLength: number): string {
93 return text.length <= maxLength ? text : `${text.slice(0, maxLength - 1)}\u2026`;
94}
95
96const clampDetail = (detail: string | null): string | null =>
97 detail === null ? null : clampText(detail, WORKTREE_SETUP_DETAIL_MAX_LENGTH);
98/** Finished snapshots stay visible this long so a late subscriber sees the outcome. */
99const FINISHED_RETENTION = "30 seconds";
100
101const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
102
103interface TrackedSetup {
104 readonly snapshot: WorktreeSetupSnapshot;
105 readonly fiber: Fiber.Fiber<unknown, unknown> | null;
106}
107
108function emptyStage(id: WorktreeSetupStageId): WorktreeSetupStage {
109 return {
110 id,
111 status: "pending",
112 startedAt: null,
113 endedAt: null,
114 percent: null,
115 detail: null,
116 tail: [],
117 };
118}
119
120export const make = Effect.gen(function* () {
121 const setups = yield* Ref.make(new Map<ThreadId, TrackedSetup>());
122 const changes = yield* PubSub.unbounded<{
123 readonly threadId: ThreadId;
124 readonly snapshot: WorktreeSetupSnapshot | null;
125 }>();
126 const retentionFibers = new Map<ThreadId, Fiber.Fiber<void, never>>();
127 // Sequences keep increasing across setups of the same thread so a stream
128 // opened during a previous setup still accepts the next one's first snapshot.
129 const lastSequenceByThread = new Map<ThreadId, number>();
130
131 const publish = (threadId: ThreadId, snapshot: WorktreeSetupSnapshot | null) =>
132 PubSub.publish(changes, { threadId, snapshot }).pipe(Effect.asVoid);
133
134 const modify = (
135 threadId: ThreadId,
136 mutate: (tracked: TrackedSetup) => TrackedSetup,
137 ): Effect.Effect<WorktreeSetupSnapshot | null> =>
138 Ref.modify(setups, (current) => {
139 const existing = current.get(threadId);
140 if (!existing) return [null, current] as const;
141 const nextTracked = mutate(existing);
142 const nextSnapshot = {
143 ...nextTracked.snapshot,
144 sequence: existing.snapshot.sequence + 1,
145 };
146 lastSequenceByThread.set(threadId, nextSnapshot.sequence);
147 const next = new Map(current);
148 next.set(threadId, { ...nextTracked, snapshot: nextSnapshot });
149 return [nextSnapshot, next] as const;
150 }).pipe(Effect.tap((snapshot) => (snapshot ? publish(threadId, snapshot) : Effect.void)));
151
152 const clearRetention = (threadId: ThreadId) => {
153 const fiber = retentionFibers.get(threadId);
154 retentionFibers.delete(threadId);
155 return fiber ? Fiber.interrupt(fiber).pipe(Effect.ignore) : Effect.void;
156 };
157
158 const remove = (threadId: ThreadId) =>
159 Ref.update(setups, (current) => {
160 if (!current.has(threadId)) return current;
161 const next = new Map(current);
162 next.delete(threadId);
163 return next;
164 }).pipe(
165 Effect.andThen(publish(threadId, null)),
166 // A subscriber that outlives retention sees `null` here and accepts any
167 // sequence after it, so the counter can start over for this thread.
168 Effect.tap(() => Effect.sync(() => lastSequenceByThread.delete(threadId))),
169 );
170
171 const begin: WorktreeSetupTracker["Service"]["begin"] = (input) =>
172 Effect.gen(function* () {
173 yield* clearRetention(input.threadId);
174 const startedAt = yield* nowIso;
175 const ordered = WORKTREE_SETUP_STAGE_ORDER.filter((id) => input.stages.includes(id));
176 const snapshot: WorktreeSetupSnapshot = {
177 threadId: input.threadId,
178 phase: "running",
179 startedAt,
180 endedAt: null,
181 branch: input.branch,
182 baseRef: input.baseRef,
183 worktreePath: null,
184 setupScript: null,
185 stages: ordered.map(emptyStage),
186 error: null,
187 sequence: (lastSequenceByThread.get(input.threadId) ?? -1) + 1,
188 };
189 lastSequenceByThread.set(input.threadId, snapshot.sequence);
190 yield* Ref.update(setups, (current) => {
191 const next = new Map(current);
192 next.set(input.threadId, { snapshot, fiber: input.fiber });
193 return next;
194 });
195 yield* publish(input.threadId, snapshot);
196 });
197
198 const update: WorktreeSetupTracker["Service"]["update"] = (threadId, mutate) =>
199 modify(threadId, (tracked) => ({ ...tracked, snapshot: mutate(tracked.snapshot) })).pipe(
200 Effect.asVoid,
201 );
202
203 const stage: WorktreeSetupTracker["Service"]["stage"] = (threadId, stageId, patch) =>
204 update(threadId, (snapshot) => ({
205 ...snapshot,
206 stages: snapshot.stages.map((entry) =>
207 entry.id === stageId
208 ? {
209 ...entry,
210 ...patch,
211 ...(patch.detail === undefined ? {} : { detail: clampDetail(patch.detail ?? null) }),
212 }
213 : entry,
214 ),
215 }));
216
217 const stageStatus: WorktreeSetupTracker["Service"]["stageStatus"] = (
218 threadId,
219 stageId,
220 status,
221 detail,
222 ) =>
223 nowIso.pipe(
224 Effect.flatMap((at) =>
225 update(threadId, (snapshot) => ({
226 ...snapshot,
227 stages: snapshot.stages.map((entry) => {
228 if (entry.id !== stageId) return entry;
229 const startedAt = entry.startedAt ?? (status === "pending" ? null : at);
230 const endedAt =
231 status === "running" || status === "pending" ? null : (entry.endedAt ?? at);
232 return {
233 ...entry,
234 status,
235 startedAt,
236 endedAt,
237 ...(detail === undefined ? {} : { detail: clampDetail(detail ?? null) }),
238 };
239 }),
240 })),
241 ),
242 );
243
244 const appendTail: WorktreeSetupTracker["Service"]["appendTail"] = (threadId, stageId, line) =>
245 update(threadId, (snapshot) => ({
246 ...snapshot,
247 stages: snapshot.stages.map((entry) =>
248 entry.id === stageId
249 ? {
250 ...entry,
251 tail: [...entry.tail, clampText(line, WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH)].slice(
252 -TAIL_LINE_LIMIT,
253 ),
254 }
255 : entry,
256 ),
257 }));
258
259 const finish: WorktreeSetupTracker["Service"]["finish"] = (threadId, phase, error) =>
260 Effect.gen(function* () {
261 const endedAt = yield* nowIso;
262 const snapshot = yield* modify(threadId, (tracked) => ({
263 fiber: null,
264 snapshot: {
265 ...tracked.snapshot,
266 phase,
267 endedAt,
268 error:
269 error === undefined || error === null
270 ? null
271 : clampText(error, WORKTREE_SETUP_ERROR_MAX_LENGTH),
272 stages: tracked.snapshot.stages.map((entry) =>
273 entry.status === "running"
274 ? {
275 ...entry,
276 status: phase === "done" ? "done" : phase === "cancelled" ? "skipped" : "failed",
277 endedAt,
278 }
279 : entry,
280 ),
281 },
282 }));
283 if (!snapshot) return null;
284 yield* clearRetention(threadId);
285 const fiber = yield* remove(threadId).pipe(
286 Effect.delay(FINISHED_RETENTION),
287 Effect.ensuring(
288 Effect.sync(() => {
289 // Only drop our own entry: a newer setup may have replaced it.
290 if (retentionFibers.get(threadId) === fiber) retentionFibers.delete(threadId);
291 }),
292 ),
293 Effect.forkDetach,
294 );
295 retentionFibers.set(threadId, fiber);
296 return snapshot;
297 });
298
299 const markUncancellable: WorktreeSetupTracker["Service"]["markUncancellable"] = (threadId) =>
300 Ref.update(setups, (current) => {
301 const existing = current.get(threadId);
302 if (!existing || existing.fiber === null) return current;
303 const next = new Map(current);
304 next.set(threadId, { ...existing, fiber: null });
305 return next;
306 });
307
308 const cancel: WorktreeSetupTracker["Service"]["cancel"] = (threadId) =>
309 Effect.gen(function* () {
310 const current = yield* Ref.get(setups);
311 const tracked = current.get(threadId);
312 if (!tracked || tracked.snapshot.phase !== "running" || !tracked.fiber) {
313 return false;
314 }
315 yield* Fiber.interrupt(tracked.fiber);
316 return true;
317 });
318
319 const get: WorktreeSetupTracker["Service"]["get"] = (threadId) =>
320 Ref.get(setups).pipe(Effect.map((current) => current.get(threadId)?.snapshot ?? null));
321
322 /**
323 * Each subscriber gets a one-slot sliding mailbox: a slow WebSocket only
324 * ever holds the newest snapshot, so a chatty setup script cannot grow the
325 * server heap. Snapshots are whole states, so skipping intermediates is safe.
326 */
327 const stream: WorktreeSetupTracker["Service"]["stream"] = (threadId) =>
328 Stream.callback<WorktreeSetupSnapshot | null>(
329 (mailbox) =>
330 Effect.gen(function* () {
331 const subscription = yield* PubSub.subscribe(changes);
332 const initial = yield* get(threadId);
333 // Changes published between subscribing and reading `initial` are
334 // already folded into it. Drop them so the client never steps back.
335 let lastSequence = initial?.sequence ?? -1;
336 Queue.offerUnsafe(mailbox, initial);
337 yield* Stream.fromSubscription(subscription).pipe(
338 Stream.runForEach((change) =>
339 Effect.sync(() => {
340 if (change.threadId !== threadId) return;
341 if (change.snapshot !== null && change.snapshot.sequence <= lastSequence) return;
342 lastSequence = change.snapshot?.sequence ?? -1;
343 Queue.offerUnsafe(mailbox, change.snapshot);
344 }),
345 ),
346 Effect.forkScoped,
347 );
348 }),
349 { bufferSize: 1, strategy: "sliding" },
350 );
351
352 return WorktreeSetupTracker.of({
353 begin,
354 update,
355 stage,
356 stageStatus,
357 appendTail,
358 finish,
359 markUncancellable,
360 cancel,
361 get,
362 stream,
363 });
364});
365
366export const layer = Layer.effect(WorktreeSetupTracker, make);