Open History to see the last commit
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
24export class WorktreeSetupTracker extends Context.Service<
25 WorktreeSetupTracker,
26 {
27 readonly begin: (input: {
28 readonly threadId: ThreadId;
29 readonly branch: string | null;
30 readonly baseRef: string | null;
31 readonly stages: ReadonlyArray<WorktreeSetupStageId>;
32 readonly fiber: Fiber.Fiber<unknown, unknown> | null;
33 }) => Effect.Effect<void>;
34 readonly update: (
35 threadId: ThreadId,
36 mutate: (snapshot: WorktreeSetupSnapshot) => WorktreeSetupSnapshot,
37 ) => Effect.Effect<void>;
38 readonly stage: (
39 threadId: ThreadId,
40 stageId: WorktreeSetupStageId,
41 patch: Partial<Omit<WorktreeSetupStage, "id">>,
42 ) => Effect.Effect<void>;
43 readonly stageStatus: (
44 threadId: ThreadId,
45 stageId: WorktreeSetupStageId,
46 status: WorktreeSetupStageStatus,
47 detail?: string | null,
48 ) => Effect.Effect<void>;
49 readonly appendTail: (
50 threadId: ThreadId,
51 stageId: WorktreeSetupStageId,
52 line: string,
53 ) => Effect.Effect<void>;
54 readonly finish: (
55 threadId: ThreadId,
56 phase: "done" | "failed" | "cancelled",
57 error?: string | null,
58 ) => Effect.Effect<WorktreeSetupSnapshot | null>;
59 /** Call just before dispatch so late cancellation cannot roll back a started agent's thread. */
60 readonly markUncancellable: (threadId: ThreadId) => Effect.Effect<void>;
61 readonly cancel: (threadId: ThreadId) => Effect.Effect<boolean>;
62 readonly get: (threadId: ThreadId) => Effect.Effect<WorktreeSetupSnapshot | null>;
63 readonly stream: (threadId: ThreadId) => Stream.Stream<WorktreeSetupSnapshot | null>;
64 }
65>()("t3/project/WorktreeSetupTracker") {}
66
67const TAIL_LINE_LIMIT = 4;
68
69function clampText(text: string, maxLength: number): string {
70 return text.length <= maxLength ? text : `${text.slice(0, maxLength - 1)}\u2026`;
71}
72
73const clampDetail = (detail: string | null): string | null =>
74 detail === null ? null : clampText(detail, WORKTREE_SETUP_DETAIL_MAX_LENGTH);
75const FINISHED_RETENTION = "30 seconds";
76
77const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
78
79interface TrackedSetup {
80 readonly snapshot: WorktreeSetupSnapshot;
81 readonly fiber: Fiber.Fiber<unknown, unknown> | null;
82}
83
84function emptyStage(id: WorktreeSetupStageId): WorktreeSetupStage {
85 return {
86 id,
87 status: "pending",
88 startedAt: null,
89 endedAt: null,
90 percent: null,
91 detail: null,
92 tail: [],
93 };
94}
95
96export const make = Effect.gen(function* () {
97 const setups = yield* Ref.make(new Map<ThreadId, TrackedSetup>());
98 const changes = yield* PubSub.unbounded<{
99 readonly threadId: ThreadId;
100 readonly snapshot: WorktreeSetupSnapshot | null;
101 }>();
102 const retentionFibers = new Map<ThreadId, Fiber.Fiber<void, never>>();
103 // Keep sequences increasing across setups so existing subscribers accept the next setup's first snapshot.
104 const lastSequenceByThread = new Map<ThreadId, number>();
105
106 const publish = (threadId: ThreadId, snapshot: WorktreeSetupSnapshot | null) =>
107 PubSub.publish(changes, { threadId, snapshot }).pipe(Effect.asVoid);
108
109 const modify = (
110 threadId: ThreadId,
111 mutate: (tracked: TrackedSetup) => TrackedSetup,
112 ): Effect.Effect<WorktreeSetupSnapshot | null> =>
113 Ref.modify(setups, (current) => {
114 const existing = current.get(threadId);
115 if (!existing) return [null, current] as const;
116 const nextTracked = mutate(existing);
117 const nextSnapshot = {
118 ...nextTracked.snapshot,
119 sequence: existing.snapshot.sequence + 1,
120 };
121 lastSequenceByThread.set(threadId, nextSnapshot.sequence);
122 const next = new Map(current);
123 next.set(threadId, { ...nextTracked, snapshot: nextSnapshot });
124 return [nextSnapshot, next] as const;
125 }).pipe(Effect.tap((snapshot) => (snapshot ? publish(threadId, snapshot) : Effect.void)));
126
127 const clearRetention = (threadId: ThreadId) => {
128 const fiber = retentionFibers.get(threadId);
129 retentionFibers.delete(threadId);
130 return fiber ? Fiber.interrupt(fiber).pipe(Effect.ignore) : Effect.void;
131 };
132
133 const remove = (threadId: ThreadId) =>
134 Ref.update(setups, (current) => {
135 if (!current.has(threadId)) return current;
136 const next = new Map(current);
137 next.delete(threadId);
138 return next;
139 }).pipe(
140 Effect.andThen(publish(threadId, null)),
141 // After retention, subscribers see null and accept any sequence; the counter can reset.
142 Effect.tap(() => Effect.sync(() => lastSequenceByThread.delete(threadId))),
143 );
144
145 const begin: WorktreeSetupTracker["Service"]["begin"] = Effect.fn("WorktreeSetupTracker.begin")(
146 function* (input) {
147 yield* clearRetention(input.threadId);
148 const startedAt = yield* nowIso;
149 const ordered = WORKTREE_SETUP_STAGE_ORDER.filter((id) => input.stages.includes(id));
150 const snapshot: WorktreeSetupSnapshot = {
151 threadId: input.threadId,
152 phase: "running",
153 startedAt,
154 endedAt: null,
155 branch: input.branch,
156 baseRef: input.baseRef,
157 worktreePath: null,
158 setupScript: null,
159 stages: ordered.map(emptyStage),
160 error: null,
161 sequence: (lastSequenceByThread.get(input.threadId) ?? -1) + 1,
162 };
163 lastSequenceByThread.set(input.threadId, snapshot.sequence);
164 yield* Ref.update(setups, (current) => {
165 const next = new Map(current);
166 next.set(input.threadId, { snapshot, fiber: input.fiber });
167 return next;
168 });
169 yield* publish(input.threadId, snapshot);
170 },
171 );
172
173 const update: WorktreeSetupTracker["Service"]["update"] = (threadId, mutate) =>
174 modify(threadId, (tracked) => ({ ...tracked, snapshot: mutate(tracked.snapshot) })).pipe(
175 Effect.asVoid,
176 );
177
178 const stage: WorktreeSetupTracker["Service"]["stage"] = (threadId, stageId, patch) =>
179 update(threadId, (snapshot) => ({
180 ...snapshot,
181 stages: snapshot.stages.map((entry) =>
182 entry.id === stageId
183 ? {
184 ...entry,
185 ...patch,
186 ...(patch.detail === undefined ? {} : { detail: clampDetail(patch.detail ?? null) }),
187 }
188 : entry,
189 ),
190 }));
191
192 const stageStatus: WorktreeSetupTracker["Service"]["stageStatus"] = (
193 threadId,
194 stageId,
195 status,
196 detail,
197 ) =>
198 nowIso.pipe(
199 Effect.flatMap((at) =>
200 update(threadId, (snapshot) => ({
201 ...snapshot,
202 stages: snapshot.stages.map((entry) => {
203 if (entry.id !== stageId) return entry;
204 const startedAt = entry.startedAt ?? (status === "pending" ? null : at);
205 const endedAt =
206 status === "running" || status === "pending" ? null : (entry.endedAt ?? at);
207 return {
208 ...entry,
209 status,
210 startedAt,
211 endedAt,
212 ...(detail === undefined ? {} : { detail: clampDetail(detail ?? null) }),
213 };
214 }),
215 })),
216 ),
217 );
218
219 const appendTail: WorktreeSetupTracker["Service"]["appendTail"] = (threadId, stageId, line) =>
220 update(threadId, (snapshot) => ({
221 ...snapshot,
222 stages: snapshot.stages.map((entry) =>
223 entry.id === stageId
224 ? {
225 ...entry,
226 tail: [...entry.tail, clampText(line, WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH)].slice(
227 -TAIL_LINE_LIMIT,
228 ),
229 }
230 : entry,
231 ),
232 }));
233
234 const finish: WorktreeSetupTracker["Service"]["finish"] = Effect.fn(
235 "WorktreeSetupTracker.finish",
236 )(function* (threadId, phase, error) {
237 const endedAt = yield* nowIso;
238 const snapshot = yield* modify(threadId, (tracked) => ({
239 fiber: null,
240 snapshot: {
241 ...tracked.snapshot,
242 phase,
243 endedAt,
244 error:
245 error === undefined || error === null
246 ? null
247 : clampText(error, WORKTREE_SETUP_ERROR_MAX_LENGTH),
248 stages: tracked.snapshot.stages.map((entry) =>
249 entry.status === "running"
250 ? {
251 ...entry,
252 status: phase === "done" ? "done" : phase === "cancelled" ? "skipped" : "failed",
253 endedAt,
254 }
255 : entry,
256 ),
257 },
258 }));
259 if (!snapshot) return null;
260 yield* clearRetention(threadId);
261 const fiber = yield* remove(threadId).pipe(
262 Effect.delay(FINISHED_RETENTION),
263 Effect.ensuring(
264 Effect.sync(() => {
265 if (retentionFibers.get(threadId) === fiber) retentionFibers.delete(threadId);
266 }),
267 ),
268 Effect.forkDetach,
269 );
270 retentionFibers.set(threadId, fiber);
271 return snapshot;
272 });
273
274 const markUncancellable: WorktreeSetupTracker["Service"]["markUncancellable"] = (threadId) =>
275 Ref.update(setups, (current) => {
276 const existing = current.get(threadId);
277 if (!existing || existing.fiber === null) return current;
278 const next = new Map(current);
279 next.set(threadId, { ...existing, fiber: null });
280 return next;
281 });
282
283 const cancel: WorktreeSetupTracker["Service"]["cancel"] = Effect.fn(
284 "WorktreeSetupTracker.cancel",
285 )(function* (threadId) {
286 const current = yield* Ref.get(setups);
287 const tracked = current.get(threadId);
288 if (!tracked || tracked.snapshot.phase !== "running" || !tracked.fiber) {
289 return false;
290 }
291 yield* Fiber.interrupt(tracked.fiber);
292 return true;
293 });
294
295 const get: WorktreeSetupTracker["Service"]["get"] = (threadId) =>
296 Ref.get(setups).pipe(Effect.map((current) => current.get(threadId)?.snapshot ?? null));
297
298 // One-slot sliding mailboxes bound slow WebSockets to the newest setup snapshot.
299 const stream: WorktreeSetupTracker["Service"]["stream"] = (threadId) =>
300 Stream.callback<WorktreeSetupSnapshot | null>(
301 (mailbox) =>
302 Effect.gen(function* () {
303 const subscription = yield* PubSub.subscribe(changes);
304 const initial = yield* get(threadId);
305 // The initial snapshot already includes queued changes; drop them to avoid stale replay.
306 let lastSequence = initial?.sequence ?? -1;
307 Queue.offerUnsafe(mailbox, initial);
308 yield* Stream.fromSubscription(subscription).pipe(
309 Stream.runForEach((change) =>
310 Effect.sync(() => {
311 if (change.threadId !== threadId) return;
312 if (change.snapshot !== null && change.snapshot.sequence <= lastSequence) return;
313 lastSequence = change.snapshot?.sequence ?? -1;
314 Queue.offerUnsafe(mailbox, change.snapshot);
315 }),
316 ),
317 Effect.forkScoped,
318 );
319 }),
320 { bufferSize: 1, strategy: "sliding" },
321 );
322
323 return WorktreeSetupTracker.of({
324 begin,
325 update,
326 stage,
327 stageStatus,
328 appendTail,
329 finish,
330 markUncancellable,
331 cancel,
332 get,
333 stream,
334 });
335});
336
337export const layer = Layer.effect(WorktreeSetupTracker, make);