1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import {
4 CheckpointScopeId,
5 ProjectId,
6 RunId,
7 ThreadId,
8 type OrchestrationV2ThreadShell,
9 type Project,
10 type RepositoryIdentity,
11} from "@t3tools/contracts";
12import * as Deferred from "effect/Deferred";
13import * as Effect from "effect/Effect";
14import * as Fiber from "effect/Fiber";
15import * as FileSystem from "effect/FileSystem";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as PubSub from "effect/PubSub";
19import * as TestClock from "effect/testing/TestClock";
20
21import * as ProcessRunner from "../processRunner.ts";
22import * as ProjectEnrichment from "../project/ProjectEnrichmentService.ts";
23import * as ProjectFaviconResolver from "../project/ProjectFaviconResolver.ts";
24import * as ProjectService from "../project/ProjectService.ts";
25import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
26import * as PullRequestService from "../pullRequest/PullRequestService.ts";
27import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
28import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";
29import * as CheckpointCapture from "./CheckpointCaptureService.ts";
30import * as ProjectionStore from "./ProjectionStore.ts";
31import * as RunFinalization from "./RunFinalizationService.ts";
32
33// Without a project the observer never touches enrichment, so its methods stay unimplemented.
34const layerNoProject = Layer.mergeAll(
35 Layer.mock(ProjectService.ProjectService)({ getById: () => Effect.succeedNone }),
36 Layer.mock(ProjectEnrichment.ProjectEnrichmentService)({}),
37);
38
39it.effect("completes finalization when the post-checkpoint workspace refresh fails", () => {
40 const threadId = ThreadId.make("thread_refresh_failure");
41 const runId = RunId.make("run_refresh_failure");
42 const scopeId = CheckpointScopeId.make("scope_refresh_failure");
43 const captured: string[] = [];
44 const layer = RunFinalization.layer.pipe(
45 Layer.provide(
46 Layer.mergeAll(
47 Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({
48 execute: () =>
49 Effect.sync(() => {
50 captured.push("checkpoint");
51 }),
52 }),
53 Layer.mock(ProjectionStore.ProjectionStoreV2)({
54 getCheckpointContext: () =>
55 Effect.succeed({
56 runs: [],
57 checkpointScopes: [{ id: scopeId, runId, kind: "root_run" as const, cwd: "/repo" }],
58 checkpoints: [],
59 }),
60 }),
61 Layer.succeed(RunFinalization.RunFinalizationObserver, {
62 refresh: () =>
63 Effect.gen(function* () {
64 captured.push("refresh");
65 return yield* new RunFinalization.RunFinalizationRefreshError({
66 cwd: "/repo",
67 cause: new Error("Git command timed out"),
68 });
69 }),
70 refreshAfterTurn: () => Effect.void,
71 }),
72 ),
73 ),
74 );
75 return Effect.gen(function* () {
76 const service = yield* RunFinalization.RunFinalizationService;
77 yield* service.finalize({ threadId, runId, scopeId });
78 assert.deepEqual(captured, ["checkpoint", "refresh"]);
79 }).pipe(Effect.provide(layer));
80});
81
82it.effect.each(
83 (
84 [
85 {
86 label: "discovers a new PR for the completed run's branch",
87 branch: "feature",
88 checkedOut: "feature",
89 activeRun: null,
90 expected: ["/repo"],
91 },
92 {
93 label: "leaves the default branch's PR cache alone",
94 branch: "main",
95 checkedOut: "main",
96 activeRun: null,
97 expected: [],
98 },
99 {
100 label: "does not refresh another thread's checkout",
101 branch: "feature",
102 checkedOut: "other",
103 activeRun: null,
104 expected: [],
105 },
106 {
107 label: "does not refresh during a newer active run",
108 branch: "feature",
109 checkedOut: "feature",
110 activeRun: "newer-run",
111 expected: [],
112 },
113 ] as const
114 ).map((scenario) => [scenario.label, scenario] as const),
115)("%s", ([, scenario]) => {
116 const refreshed: string[] = [];
117 const lookedUp = Deferred.makeUnsafe<void>();
118 const threadId = ThreadId.make("thread-pr-refresh");
119 const runId = RunId.make("completed-run");
120 const layer = RunFinalization.layerObserver.pipe(
121 Layer.provide(
122 Layer.mergeAll(
123 layerNoProject,
124 Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }),
125 Layer.mock(PullRequestService.PullRequestService)({
126 refreshAfterTurn: () => Effect.void,
127 }),
128 Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({
129 refreshLocalStatus: () =>
130 Effect.succeed({
131 isRepo: true,
132 hasPrimaryRemote: true,
133 isDefaultRef: scenario.checkedOut === "main",
134 refName: scenario.checkedOut,
135 hasWorkingTreeChanges: false,
136 workingTree: { files: [], insertions: 0, deletions: 0 },
137 }),
138 refreshStatus: () =>
139 Effect.die("turn completion must preserve known PRs and lookup backoff"),
140 // Requested synchronously by refresh; the lookup itself runs in the background.
141 refreshPullRequestStatus: (cwd) => {
142 refreshed.push(cwd);
143 return Deferred.succeed(lookedUp, undefined).pipe(Effect.as(null));
144 },
145 }),
146 Layer.mock(ProjectionStore.ProjectionStoreV2)({
147 getThreadShell: () =>
148 Effect.succeed({
149 id: threadId,
150 branch: scenario.branch,
151 activeRunId: scenario.activeRun === null ? null : RunId.make(scenario.activeRun),
152 } as OrchestrationV2ThreadShell),
153 }),
154 ),
155 ),
156 );
157 return Effect.gen(function* () {
158 const observer = yield* RunFinalization.RunFinalizationObserver;
159 yield* observer.refresh({ cwd: "/repo", threadId, runId });
160 if (scenario.expected.length > 0) yield* Deferred.await(lookedUp);
161 assert.deepEqual(refreshed, [...scenario.expected]);
162 }).pipe(Effect.provide(layer));
163});
164
165it.effect("returns from the post-turn refresh without waiting for the PR lookup", () => {
166 const threadId = ThreadId.make("thread-slow-forge");
167 const runId = RunId.make("completed-run");
168 const layer = RunFinalization.layerObserver.pipe(
169 Layer.provide(
170 Layer.mergeAll(
171 layerNoProject,
172 Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }),
173 Layer.mock(PullRequestService.PullRequestService)({
174 refreshAfterTurn: () => Effect.void,
175 }),
176 Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({
177 refreshLocalStatus: () =>
178 Effect.succeed({
179 isRepo: true,
180 hasPrimaryRemote: true,
181 isDefaultRef: false,
182 refName: "feature",
183 hasWorkingTreeChanges: false,
184 workingTree: { files: [], insertions: 0, deletions: 0 },
185 }),
186 refreshPullRequestStatus: () => Effect.never,
187 }),
188 Layer.mock(ProjectionStore.ProjectionStoreV2)({
189 getThreadShell: () =>
190 Effect.succeed({
191 id: threadId,
192 branch: "feature",
193 activeRunId: null,
194 } as OrchestrationV2ThreadShell),
195 }),
196 ),
197 ),
198 );
199 return Effect.gen(function* () {
200 const observer = yield* RunFinalization.RunFinalizationObserver;
201 const refresh = yield* observer
202 .refresh({ cwd: "/repo", threadId, runId })
203 .pipe(Effect.timeoutOption("1 minute"), Effect.forkChild({ startImmediately: true }));
204 yield* TestClock.adjust("1 minute");
205 const result = yield* Fiber.join(refresh);
206 assert.isTrue(
207 Option.isSome(result),
208 "turn finalization waited on the forge PR lookup, blocking the thread's next turn",
209 );
210 }).pipe(Effect.provide(layer));
211});
212
213it.effect("replaces a still-running PR lookup when a newer turn finishes", () => {
214 const threadId = ThreadId.make("thread-back-to-back");
215 const runId = RunId.make("completed-run");
216 const firstInterrupted = Deferred.makeUnsafe<void>();
217 const secondLookedUp = Deferred.makeUnsafe<void>();
218 let lookups = 0;
219 const layer = RunFinalization.layerObserver.pipe(
220 Layer.provide(
221 Layer.mergeAll(
222 layerNoProject,
223 Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }),
224 Layer.mock(PullRequestService.PullRequestService)({
225 refreshAfterTurn: () => Effect.void,
226 }),
227 Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({
228 refreshLocalStatus: () =>
229 Effect.succeed({
230 isRepo: true,
231 hasPrimaryRemote: true,
232 isDefaultRef: false,
233 refName: "feature",
234 hasWorkingTreeChanges: false,
235 workingTree: { files: [], insertions: 0, deletions: 0 },
236 }),
237 refreshPullRequestStatus: () =>
238 ++lookups === 1
239 ? Effect.never.pipe(
240 Effect.onInterrupt(() => Deferred.succeed(firstInterrupted, undefined)),
241 )
242 : Deferred.succeed(secondLookedUp, undefined).pipe(Effect.as(null)),
243 }),
244 Layer.mock(ProjectionStore.ProjectionStoreV2)({
245 getThreadShell: () =>
246 Effect.succeed({
247 id: threadId,
248 branch: "feature",
249 activeRunId: null,
250 } as OrchestrationV2ThreadShell),
251 }),
252 ),
253 ),
254 );
255 return Effect.gen(function* () {
256 const observer = yield* RunFinalization.RunFinalizationObserver;
257 yield* observer.refresh({ cwd: "/repo", threadId, runId });
258 yield* observer.refresh({ cwd: "/repo", threadId, runId });
259 yield* Deferred.await(firstInterrupted);
260 yield* Deferred.await(secondLookedUp);
261 assert.strictEqual(lookups, 2);
262 }).pipe(Effect.provide(layer));
263});
264
265const identityProjectId = ProjectId.make("project-identity-refresh");
266const identityThreadId = ThreadId.make("thread-identity-refresh");
267const identityRunId = RunId.make("completed-run");
268
269const layerIdentityRefresh = <R>(
270 workspaceRoot: string,
271 resolver: Layer.Layer<RepositoryIdentityResolver.RepositoryIdentityResolver, never, R>,
272) =>
273 RunFinalization.layerObserver.pipe(
274 Layer.provideMerge(
275 Layer.mergeAll(
276 Layer.effect(ProjectEnrichment.ProjectEnrichmentService, ProjectEnrichment.make()).pipe(
277 Layer.provide(
278 Layer.mergeAll(
279 resolver,
280 Layer.succeed(ProjectFaviconResolver.ProjectFaviconResolver, {
281 invalidate: () => Effect.void,
282 resolvePath: () => Effect.succeed(null),
283 }),
284 ),
285 ),
286 ),
287 Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }),
288 Layer.mock(PullRequestService.PullRequestService)({ refreshAfterTurn: () => Effect.void }),
289 Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({
290 refreshLocalStatus: () =>
291 Effect.succeed({
292 isRepo: true,
293 hasPrimaryRemote: true,
294 isDefaultRef: true,
295 refName: "main",
296 hasWorkingTreeChanges: false,
297 workingTree: { files: [], insertions: 0, deletions: 0 },
298 }),
299 }),
300 Layer.mock(ProjectionStore.ProjectionStoreV2)({
301 getThreadShell: () =>
302 Effect.succeed({
303 id: identityThreadId,
304 projectId: identityProjectId,
305 branch: "main",
306 activeRunId: null,
307 } as OrchestrationV2ThreadShell),
308 }),
309 Layer.mock(ProjectService.ProjectService)({
310 getById: () => Effect.succeedSome({ id: identityProjectId, workspaceRoot } as Project),
311 }),
312 ),
313 ),
314 );
315
316it.effect("resolves the identity of a project whose remote was added during the turn", () =>
317 Effect.gen(function* () {
318 const fileSystem = yield* FileSystem.FileSystem;
319 const processRunner = yield* ProcessRunner.ProcessRunner;
320 const workspaceRoot = yield* fileSystem.makeTempDirectoryScoped({
321 prefix: "t3-run-finalization-identity-test-",
322 });
323 const git = (args: ReadonlyArray<string>) =>
324 processRunner.run({ command: "git", args: ["-C", workspaceRoot, ...args] });
325 yield* git(["init"]);
326
327 yield* Effect.gen(function* () {
328 const enrichment = yield* ProjectEnrichment.ProjectEnrichmentService;
329 // A shell read before the turn caches the folder as having no repository.
330 assert.isNull(yield* enrichment.readRepositoryIdentity(workspaceRoot));
331
332 yield* git(["remote", "add", "origin", "https://github.com/t3-triage/example-29929.git"]);
333 const changes = yield* enrichment.subscribeChanges;
334 const observer = yield* RunFinalization.RunFinalizationObserver;
335 yield* observer.refresh({
336 cwd: workspaceRoot,
337 threadId: identityThreadId,
338 runId: identityRunId,
339 });
340
341 const change = yield* PubSub.take(changes);
342 assert.strictEqual(
343 change.enrichment.repositoryIdentity?.locator.remoteUrl,
344 "https://github.com/t3-triage/example-29929.git",
345 "the project's repository identity is still null after the turn added a remote",
346 );
347 }).pipe(
348 Effect.scoped,
349 Effect.provide(layerIdentityRefresh(workspaceRoot, RepositoryIdentityResolver.layer)),
350 );
351 }).pipe(
352 Effect.scoped,
353 Effect.provide(ProcessRunner.layer.pipe(Layer.provideMerge(NodeServices.layer))),
354 ),
355);
356
357it.effect("returns from the post-turn refresh while the identity resolver is still running", () => {
358 const workspaceRoot = "/repo-stubbed-identity";
359 const identity: RepositoryIdentity = {
360 canonicalKey: "example.test/stubbed-identity",
361 locator: {
362 source: "git-remote",
363 remoteName: "origin",
364 remoteUrl: "https://example.test/stubbed-identity.git",
365 },
366 };
367 const releaseResolver = Deferred.makeUnsafe<void>();
368 let remoteAdded = false;
369 const resolver = Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, {
370 resolve: (_cwd, options) =>
371 options?.refresh
372 ? Deferred.await(releaseResolver).pipe(
373 Effect.map(() => {
374 remoteAdded = true;
375 return identity;
376 }),
377 )
378 : Effect.sync(() => (remoteAdded ? identity : null)),
379 });
380 return Effect.gen(function* () {
381 const enrichment = yield* ProjectEnrichment.ProjectEnrichmentService;
382 assert.isNull(yield* enrichment.readRepositoryIdentity(workspaceRoot));
383 const changes = yield* enrichment.subscribeChanges;
384 const observer = yield* RunFinalization.RunFinalizationObserver;
385 const refresh = yield* observer
386 .refresh({ cwd: workspaceRoot, threadId: identityThreadId, runId: identityRunId })
387 .pipe(Effect.timeoutOption("1 minute"), Effect.forkChild({ startImmediately: true }));
388 yield* TestClock.adjust("1 minute");
389 assert.isTrue(
390 Option.isSome(yield* Fiber.join(refresh)),
391 "turn finalization waited on the identity resolver, blocking the thread's next turn",
392 );
393
394 yield* Deferred.succeed(releaseResolver, undefined);
395 const change = yield* PubSub.take(changes);
396 assert.strictEqual(change.enrichment.repositoryIdentity?.canonicalKey, identity.canonicalKey);
397 }).pipe(Effect.scoped, Effect.provide(layerIdentityRefresh(workspaceRoot, resolver)));
398});