1import { describe, expect, it } from "@effect/vitest";
2import { ThreadId, WorktreeSetupSnapshot } from "@t3tools/contracts";
3import * as Deferred from "effect/Deferred";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";
8import * as Schema from "effect/Schema";
9import * as Stream from "effect/Stream";
10import * as TestClock from "effect/testing/TestClock";
11
12import * as WorktreeSetupTracker from "./WorktreeSetupTracker.ts";
13
14const threadId = ThreadId.make("thread-1");
15
16describe("WorktreeSetupTracker", () => {
17 it.effect("records stage transitions, checkout progress, and the final phase", () =>
18 Effect.gen(function* () {
19 const tracker = yield* WorktreeSetupTracker.make;
20 yield* tracker.begin({
21 threadId,
22 branch: "feature",
23 baseRef: "main",
24 stages: ["checkout", "fetch", "agent"],
25 fiber: null,
26 });
27
28 const initial = yield* tracker.get(threadId);
29 // Stages are reordered into the canonical setup order.
30 expect(initial?.stages.map((stage) => stage.id)).toEqual(["fetch", "checkout", "agent"]);
31 expect(initial?.phase).toBe("running");
32
33 yield* tracker.stageStatus(threadId, "fetch", "running");
34 yield* tracker.stageStatus(threadId, "fetch", "done", "origin/main at abc1234");
35 yield* tracker.stageStatus(threadId, "checkout", "running");
36 yield* tracker.stage(threadId, "checkout", { percent: 42, detail: "42 / 100 files" });
37 yield* tracker.finish(threadId, "failed", "boom");
38
39 const final = yield* tracker.get(threadId);
40 expect(final?.phase).toBe("failed");
41 expect(final?.error).toBe("boom");
42 const [fetch, checkout, agent] = final?.stages ?? [];
43 expect(fetch).toMatchObject({ status: "done", detail: "origin/main at abc1234" });
44 expect(fetch?.startedAt).not.toBeNull();
45 expect(fetch?.endedAt).not.toBeNull();
46 // A stage still running when the setup fails is marked failed.
47 expect(checkout).toMatchObject({ status: "failed", percent: 42 });
48 expect(agent?.status).toBe("pending");
49 expect(final?.sequence).toBeGreaterThan(initial?.sequence ?? 0);
50 }),
51 );
52
53 it.effect("stream emits the current snapshot first and then only newer ones", () =>
54 Effect.gen(function* () {
55 const tracker = yield* WorktreeSetupTracker.make;
56 yield* tracker.begin({
57 threadId,
58 branch: null,
59 baseRef: null,
60 stages: ["agent"],
61 fiber: null,
62 });
63
64 const collected = yield* tracker.stream(threadId).pipe(
65 Stream.takeUntil((snapshot) => snapshot?.sequence === 2),
66 Stream.runCollect,
67 Effect.forkChild,
68 );
69 yield* Effect.yieldNow;
70 yield* tracker.stageStatus(threadId, "agent", "running");
71 yield* tracker.appendTail(threadId, "agent", "line 1");
72
73 const snapshots = yield* Fiber.join(collected);
74 const sequences = snapshots.map((snapshot) => snapshot?.sequence ?? -1);
75 expect(sequences.at(-1)).toBe(2);
76 // Delivery is latest-value per subscriber, so intermediates may be
77 // skipped but never delivered out of order.
78 expect(sequences).toEqual([...sequences].toSorted((a, b) => a - b));
79 expect(snapshots.at(-1)?.stages[0]?.tail).toEqual(["line 1"]);
80 }),
81 );
82
83 it.effect("stream never steps back behind the snapshot it started from", () =>
84 Effect.gen(function* () {
85 const tracker = yield* WorktreeSetupTracker.make;
86 yield* tracker.begin({
87 threadId,
88 branch: null,
89 baseRef: null,
90 stages: ["agent"],
91 fiber: null,
92 });
93 yield* tracker.stageStatus(threadId, "agent", "running");
94 yield* tracker.stageStatus(threadId, "agent", "done");
95
96 // A late subscriber starts at sequence 2 and must never see 0 or 1.
97 const collected = yield* tracker.stream(threadId).pipe(
98 Stream.takeUntil((snapshot) => snapshot?.phase === "done"),
99 Stream.runCollect,
100 Effect.forkChild,
101 );
102 yield* Effect.yieldNow;
103 yield* tracker.finish(threadId, "done");
104
105 const snapshots = yield* Fiber.join(collected);
106 expect(snapshots.length).toBeGreaterThan(0);
107 expect(snapshots.every((snapshot) => (snapshot?.sequence ?? -1) >= 2)).toBe(true);
108 expect(snapshots.at(-1)?.phase).toBe("done");
109 }),
110 );
111
112 it.effect("a new setup on the same thread keeps sequences increasing", () =>
113 Effect.gen(function* () {
114 const tracker = yield* WorktreeSetupTracker.make;
115 yield* tracker.begin({
116 threadId,
117 branch: "first",
118 baseRef: null,
119 stages: ["agent"],
120 fiber: null,
121 });
122 yield* tracker.finish(threadId, "failed", "boom");
123 const failedSequence = (yield* tracker.get(threadId))?.sequence ?? -1;
124
125 // A stream opened on the failed setup must still receive the next one.
126 const collected = yield* tracker.stream(threadId).pipe(
127 Stream.takeUntil((snapshot) => snapshot?.branch === "second"),
128 Stream.runCollect,
129 Effect.forkChild,
130 );
131 yield* Effect.yieldNow;
132 yield* tracker.begin({
133 threadId,
134 branch: "second",
135 baseRef: null,
136 stages: ["agent"],
137 fiber: null,
138 });
139
140 const snapshots = yield* Fiber.join(collected);
141 const last = snapshots.at(-1);
142 expect(last?.phase).toBe("running");
143 expect(last?.sequence).toBeGreaterThan(failedSequence);
144 }),
145 );
146
147 it.effect("finished setups are dropped after the retention window", () =>
148 Effect.gen(function* () {
149 const tracker = yield* WorktreeSetupTracker.make;
150 yield* tracker.begin({
151 threadId,
152 branch: null,
153 baseRef: null,
154 stages: ["agent"],
155 fiber: null,
156 });
157 yield* tracker.finish(threadId, "done");
158 expect((yield* tracker.get(threadId))?.phase).toBe("done");
159
160 yield* TestClock.adjust(Duration.seconds(31));
161 expect(yield* tracker.get(threadId)).toBeNull();
162 }),
163 );
164
165 it.effect("cancel interrupts the bootstrap fiber and reports whether one was running", () =>
166 Effect.gen(function* () {
167 const tracker = yield* WorktreeSetupTracker.make;
168 const started = yield* Deferred.make<void>();
169 const fiber = yield* Deferred.succeed(started, undefined).pipe(
170 Effect.andThen(Effect.never),
171 Effect.forkChild,
172 );
173 yield* Deferred.await(started);
174 yield* tracker.begin({ threadId, branch: null, baseRef: null, stages: ["agent"], fiber });
175
176 expect(yield* tracker.cancel(threadId)).toBe(true);
177 // cancel returns only after the bootstrap fiber has unwound.
178 const exit = yield* Fiber.await(fiber);
179 expect(Exit.hasInterrupts(exit)).toBe(true);
180
181 yield* tracker.finish(threadId, "cancelled");
182 expect(yield* tracker.cancel(threadId)).toBe(false);
183 expect(yield* tracker.cancel(ThreadId.make("unknown"))).toBe(false);
184 }),
185 );
186
187 it.effect("markUncancellable makes a later cancel a no-op while the setup keeps running", () =>
188 Effect.gen(function* () {
189 const tracker = yield* WorktreeSetupTracker.make;
190 const fiber = yield* Effect.forkChild(Effect.never);
191 yield* tracker.begin({ threadId, branch: null, baseRef: null, stages: ["agent"], fiber });
192
193 yield* tracker.markUncancellable(threadId);
194 expect(yield* tracker.cancel(threadId)).toBe(false);
195 expect((yield* tracker.get(threadId))?.phase).toBe("running");
196 yield* Fiber.interrupt(fiber);
197 }),
198 );
199
200 it.effect("clamps free text to the contract limits before publishing", () =>
201 Effect.gen(function* () {
202 const tracker = yield* WorktreeSetupTracker.make;
203 yield* tracker.begin({
204 threadId,
205 branch: null,
206 baseRef: null,
207 stages: ["checkout", "setup-script"],
208 fiber: null,
209 });
210 const long = "x".repeat(2_000);
211
212 yield* tracker.stageStatus(threadId, "checkout", "done", long);
213 yield* tracker.stage(threadId, "setup-script", { detail: long });
214 yield* tracker.appendTail(threadId, "setup-script", long);
215 yield* tracker.finish(threadId, "failed", long);
216
217 const snapshot = yield* tracker.get(threadId);
218 expect(snapshot?.stages[0]?.detail?.length).toBe(200);
219 expect(snapshot?.stages[1]?.detail?.length).toBe(200);
220 expect(snapshot?.stages[1]?.tail[0]?.length).toBe(400);
221 expect(snapshot?.error?.length).toBe(1000);
222 // The wire schema must accept what the tracker publishes.
223 expect(Schema.is(WorktreeSetupSnapshot)(snapshot)).toBe(true);
224 }),
225 );
226});