1import { SourceControlProviderKind } from "@t3tools/contracts";
2import { assert, it, describe } from "@effect/vitest";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import * as Cause from "effect/Cause";
5import * as Deferred from "effect/Deferred";
6import * as DateTime from "effect/DateTime";
7import * as Duration from "effect/Duration";
8import * as Effect from "effect/Effect";
9import * as Fiber from "effect/Fiber";
10import * as Exit from "effect/Exit";
11import * as FileSystem from "effect/FileSystem";
12import * as Layer from "effect/Layer";
13import * as Logger from "effect/Logger";
14import * as Option from "effect/Option";
15import * as Path from "effect/Path";
16import * as Scope from "effect/Scope";
17import * as Stream from "effect/Stream";
18import * as TestClock from "effect/testing/TestClock";
19import type {
20 BackgroundScope,
21 VcsStatusLocalResult,
22 VcsStatusRemoteResult,
23 VcsStatusResult,
24 VcsStatusStreamEvent,
25} from "@t3tools/contracts";
26import { GitManagerError } from "@t3tools/contracts";
27
28import * as VcsStatusBroadcaster from "./VcsStatusBroadcaster.ts";
29import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
30import * as GitWorkflowService from "../git/GitWorkflowService.ts";
31import { symlinksSupported } from "@t3tools/shared/testing/symlinks";
32
33const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z");
34
35const baseLocalStatus: VcsStatusLocalResult = {
36 isRepo: true,
37 sourceControlProvider: {
38 kind: SourceControlProviderKind.make("github"),
39 name: "GitHub",
40 baseUrl: "https://github.com",
41 },
42 hasPrimaryRemote: true,
43 isDefaultRef: false,
44 refName: "feature/status-broadcast",
45 hasWorkingTreeChanges: false,
46 workingTree: { files: [], insertions: 0, deletions: 0 },
47};
48
49const baseRemoteStatus: VcsStatusRemoteResult = {
50 hasUpstream: true,
51 aheadCount: 0,
52 behindCount: 0,
53 pr: null,
54};
55
56const remoteStatusWithPr: VcsStatusRemoteResult = {
57 ...baseRemoteStatus,
58 pr: {
59 number: 2978,
60 title: "[codex] Rewrite client connection architecture",
61 url: "https://github.com/pingdotgg/t3code/pull/2978",
62 baseRef: "main",
63 headRef: "codex/connection-state-audit",
64 state: "open",
65 },
66};
67
68const baseStatus: VcsStatusResult = {
69 ...baseLocalStatus,
70 ...baseRemoteStatus,
71};
72
73function layerTestFor(state: {
74 currentLocalStatus: VcsStatusLocalResult;
75 currentRemoteStatus: VcsStatusRemoteResult | null;
76 localStatusCalls: number;
77 remoteStatusCalls: number;
78 localInvalidationCalls: number;
79 remoteInvalidationCalls: number;
80 remoteStatusRefreshUpstreamValues?: Array<boolean | undefined>;
81 backgroundWorkEnabled?: boolean;
82 /** Runs before each remote status read, e.g. to hold a fetch open. */
83 beforeRemoteStatus?: Effect.Effect<void>;
84}) {
85 return VcsStatusBroadcaster.layer.pipe(
86 Layer.provideMerge(NodeServices.layer),
87 Layer.provide(layerBackgroundPolicy(() => state.backgroundWorkEnabled !== false)),
88 Layer.provide(
89 Layer.mock(GitWorkflowService.GitWorkflowService)({
90 localStatus: () =>
91 Effect.sync(() => {
92 state.localStatusCalls += 1;
93 return state.currentLocalStatus;
94 }),
95 remoteStatus: (_input, options) =>
96 Effect.suspend(() => state.beforeRemoteStatus ?? Effect.void).pipe(
97 Effect.andThen(
98 Effect.sync(() => {
99 state.remoteStatusCalls += 1;
100 state.remoteStatusRefreshUpstreamValues?.push(options?.refreshUpstream);
101 return state.currentRemoteStatus;
102 }),
103 ),
104 ),
105 invalidateLocalStatus: () =>
106 Effect.sync(() => {
107 state.localInvalidationCalls += 1;
108 }),
109 invalidateRemoteStatus: () =>
110 Effect.sync(() => {
111 state.remoteInvalidationCalls += 1;
112 }),
113 invalidateStatus: () =>
114 Effect.sync(() => {
115 state.localInvalidationCalls += 1;
116 state.remoteInvalidationCalls += 1;
117 }),
118 }),
119 ),
120 );
121}
122
123function layerBackgroundPolicy(shouldRunScopeWork: (scope: BackgroundScope) => boolean) {
124 return Layer.mock(BackgroundPolicy.BackgroundPolicy)({
125 reportClientActivity: () => Effect.void,
126 removeRpcClient: () => Effect.void,
127 reportHostPowerState: () => Effect.void,
128 snapshot: Effect.succeed({
129 hostPower: {
130 source: "unknown",
131 idle: "unknown",
132 idleSeconds: null,
133 locked: "unknown",
134 suspended: false,
135 onBattery: "unknown",
136 lowPowerMode: "unknown",
137 thermalState: "unknown",
138 stale: true,
139 updatedAt: TEST_EPOCH,
140 },
141 leases: [],
142 activeForegroundLeaseCount: 0,
143 activeScopeKeys: [],
144 shouldRunOpportunisticWork: false,
145 updatedAt: TEST_EPOCH,
146 }),
147 streamChanges: Stream.empty,
148 hasDemand: () => Effect.succeed(true),
149 shouldRunScopeWork: (scope) => Effect.sync(() => shouldRunScopeWork(scope)),
150 shouldRunOpportunisticWork: Effect.succeed(true),
151 });
152}
153
154describe("VcsStatusBroadcaster", () => {
155 it.effect.skipIf(!symlinksSupported)(
156 "automatically pulls an enabled clean default branch when status detects it is behind",
157 () => {
158 let remoteStatus: VcsStatusRemoteResult = { ...baseRemoteStatus, behindCount: 2 };
159 let pullCalls = 0;
160 let configuredWorkspaceRoot = "";
161 const localStatus: VcsStatusLocalResult = {
162 ...baseLocalStatus,
163 isDefaultRef: true,
164 refName: "main",
165 };
166 const layerTest = VcsStatusBroadcaster.layer.pipe(
167 Layer.provideMerge(NodeServices.layer),
168 Layer.provide(layerBackgroundPolicy(() => true)),
169 Layer.provide(
170 Layer.succeed(VcsStatusBroadcaster.VcsAutoPullPolicy, {
171 isEnabled: (cwd) => Effect.succeed(cwd === configuredWorkspaceRoot),
172 }),
173 ),
174 Layer.provide(
175 Layer.mock(GitWorkflowService.GitWorkflowService)({
176 localStatus: () => Effect.succeed(localStatus),
177 remoteStatus: () => Effect.succeed(remoteStatus),
178 invalidateLocalStatus: () => Effect.void,
179 invalidateRemoteStatus: () => Effect.void,
180 invalidateStatus: () => Effect.void,
181 pullCurrentBranch: () =>
182 Effect.sync(() => {
183 pullCalls += 1;
184 remoteStatus = { ...remoteStatus, behindCount: 0 };
185 return {
186 status: "pulled" as const,
187 refName: "main",
188 upstreamRef: "origin/main",
189 };
190 }),
191 }),
192 ),
193 );
194
195 return Effect.gen(function* () {
196 const fileSystem = yield* FileSystem.FileSystem;
197 const path = yield* Path.Path;
198 const realDir = yield* fileSystem.makeTempDirectoryScoped({
199 prefix: "t3-vcs-auto-pull-real-",
200 });
201 const linkParent = yield* fileSystem.makeTempDirectoryScoped({
202 prefix: "t3-vcs-auto-pull-link-",
203 });
204 configuredWorkspaceRoot = path.join(linkParent, "repo-link");
205 yield* fileSystem.symlink(realDir, configuredWorkspaceRoot);
206
207 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
208 const status = yield* broadcaster.refreshStatus(configuredWorkspaceRoot);
209
210 assert.equal(pullCalls, 1);
211 assert.equal(status.behindCount, 0);
212 }).pipe(Effect.provide(layerTest));
213 },
214 );
215
216 it.effect("reuses the cached VCS status across repeated reads", () => {
217 const state = {
218 currentLocalStatus: baseLocalStatus,
219 currentRemoteStatus: baseRemoteStatus,
220 localStatusCalls: 0,
221 remoteStatusCalls: 0,
222 localInvalidationCalls: 0,
223 remoteInvalidationCalls: 0,
224 };
225
226 return Effect.gen(function* () {
227 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
228
229 const first = yield* broadcaster.getStatus({ cwd: "/repo" });
230 const second = yield* broadcaster.getStatus({ cwd: "/repo" });
231
232 assert.deepStrictEqual(first, baseStatus);
233 assert.deepStrictEqual(second, baseStatus);
234 assert.equal(state.localStatusCalls, 1);
235 assert.equal(state.remoteStatusCalls, 1);
236 assert.equal(state.localInvalidationCalls, 0);
237 assert.equal(state.remoteInvalidationCalls, 0);
238 }).pipe(Effect.provide(layerTestFor(state)));
239 });
240
241 it.effect("refreshes a loaded cwd without reusing a previous branch's PR", () => {
242 const state = {
243 currentLocalStatus: baseLocalStatus,
244 currentRemoteStatus: baseRemoteStatus,
245 localStatusCalls: 0,
246 remoteStatusCalls: 0,
247 localInvalidationCalls: 0,
248 remoteInvalidationCalls: 0,
249 };
250
251 return Effect.gen(function* () {
252 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
253
254 // Nobody loaded this cwd yet: no host request is spent.
255 assert.isNull(yield* broadcaster.refreshPullRequestStatus("/repo"));
256 assert.equal(state.remoteStatusCalls, 0);
257
258 yield* broadcaster.getStatus({ cwd: "/repo" });
259 assert.equal(state.remoteStatusCalls, 1);
260
261 // Loaded and no PR known: ask GitManager to retry the missing PR.
262 state.currentRemoteStatus = remoteStatusWithPr;
263 const refreshed = yield* broadcaster.refreshPullRequestStatus("/repo");
264 assert.deepStrictEqual(refreshed, remoteStatusWithPr);
265 assert.equal(state.remoteStatusCalls, 2);
266 assert.equal(state.remoteInvalidationCalls, 0);
267
268 // The agent switches branches. The previous branch's PR must not block a read.
269 state.currentLocalStatus = { ...baseLocalStatus, refName: "feature/next" };
270 state.currentRemoteStatus = baseRemoteStatus;
271 yield* broadcaster.refreshLocalStatus("/repo");
272 const refreshedBranch = yield* broadcaster.refreshPullRequestStatus("/repo");
273 assert.deepStrictEqual(refreshedBranch, baseRemoteStatus);
274 assert.equal(state.remoteStatusCalls, 3);
275 }).pipe(Effect.provide(layerTestFor(state)));
276 });
277
278 it.effect("a poll that started before the turn-end refresh cannot overwrite its PR", () => {
279 const releaseFirstPoll = Deferred.makeUnsafe<void>();
280 const firstPollStarted = Deferred.makeUnsafe<void>();
281 let remoteReads = 0;
282 const layer = VcsStatusBroadcaster.layer.pipe(
283 Layer.provideMerge(NodeServices.layer),
284 Layer.provide(layerBackgroundPolicy(() => true)),
285 Layer.provide(
286 Layer.mock(GitWorkflowService.GitWorkflowService)({
287 localStatus: () => Effect.succeed(baseLocalStatus),
288 remoteStatus: () =>
289 Effect.gen(function* () {
290 remoteReads += 1;
291 if (remoteReads === 2) {
292 // Hold an older empty response while the turn-end refresh queues.
293 yield* Deferred.succeed(firstPollStarted, undefined);
294 yield* Deferred.await(releaseFirstPoll);
295 return baseRemoteStatus;
296 }
297 return remoteReads === 1 ? baseRemoteStatus : remoteStatusWithPr;
298 }),
299 invalidateLocalStatus: () => Effect.void,
300 invalidateRemoteStatus: () => Effect.void,
301 invalidateStatus: () => Effect.void,
302 }),
303 ),
304 );
305
306 return Effect.gen(function* () {
307 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
308 yield* broadcaster.getStatus({ cwd: "/repo" });
309
310 const poll = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);
311 yield* Deferred.await(firstPollStarted);
312 const refresh = yield* broadcaster.refreshPullRequestStatus("/repo").pipe(Effect.forkScoped);
313 yield* Deferred.succeed(releaseFirstPoll, undefined);
314 yield* Fiber.join(poll);
315 const refreshed = yield* Fiber.join(refresh);
316
317 assert.deepStrictEqual(refreshed, remoteStatusWithPr);
318 const final = yield* broadcaster.getStatus({ cwd: "/repo" });
319 assert.deepStrictEqual(final.pr, remoteStatusWithPr.pr);
320 }).pipe(Effect.provide(layer), Effect.scoped);
321 });
322
323 it.effect("an initial status read cannot overwrite an explicit refresh", () => {
324 const firstReadStarted = Deferred.makeUnsafe<void>();
325 const releaseFirstRead = Deferred.makeUnsafe<void>();
326 let remoteReads = 0;
327 const layer = VcsStatusBroadcaster.layer.pipe(
328 Layer.provide(FileSystem.layerNoop({ realPath: (path) => Effect.succeed(path) })),
329 Layer.provideMerge(NodeServices.layer),
330 Layer.provide(layerBackgroundPolicy(() => true)),
331 Layer.provide(
332 Layer.mock(GitWorkflowService.GitWorkflowService)({
333 localStatus: () => Effect.succeed(baseLocalStatus),
334 remoteStatus: () =>
335 Effect.gen(function* () {
336 remoteReads += 1;
337 if (remoteReads === 1) {
338 yield* Deferred.succeed(firstReadStarted, undefined);
339 yield* Deferred.await(releaseFirstRead);
340 return baseRemoteStatus;
341 }
342 return remoteStatusWithPr;
343 }),
344 invalidateStatus: () => Effect.void,
345 }),
346 ),
347 );
348 return Effect.gen(function* () {
349 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
350 const initial = yield* broadcaster.getStatus({ cwd: "/repo" }).pipe(Effect.forkScoped);
351 yield* Deferred.await(firstReadStarted);
352 const refresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);
353 // Run ready fibers before releasing the delayed first read.
354 yield* TestClock.adjust(Duration.zero);
355 yield* Deferred.succeed(releaseFirstRead, undefined);
356 yield* Fiber.join(initial);
357 yield* Fiber.join(refresh);
358 assert.deepStrictEqual(
359 (yield* broadcaster.getStatus({ cwd: "/repo" })).pr,
360 remoteStatusWithPr.pr,
361 );
362 }).pipe(Effect.provide(layer), Effect.scoped);
363 });
364
365 it.effect("turn-end refresh skips a loaded cwd when background policy pauses it", () => {
366 const state = {
367 currentLocalStatus: baseLocalStatus,
368 currentRemoteStatus: baseRemoteStatus,
369 localStatusCalls: 0,
370 remoteStatusCalls: 0,
371 localInvalidationCalls: 0,
372 remoteInvalidationCalls: 0,
373 backgroundWorkEnabled: false,
374 };
375 return Effect.gen(function* () {
376 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
377 yield* broadcaster.getStatus({ cwd: "/repo" });
378 yield* broadcaster.refreshPullRequestStatus("/repo");
379 assert.equal(state.remoteStatusCalls, 1);
380 assert.equal(state.remoteInvalidationCalls, 0);
381 }).pipe(Effect.provide(layerTestFor(state)));
382 });
383
384 it.effect("refreshes the cached snapshot after explicit invalidation", () => {
385 const state = {
386 currentLocalStatus: baseLocalStatus,
387 currentRemoteStatus: baseRemoteStatus,
388 localStatusCalls: 0,
389 remoteStatusCalls: 0,
390 localInvalidationCalls: 0,
391 remoteInvalidationCalls: 0,
392 };
393
394 return Effect.gen(function* () {
395 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
396 const initial = yield* broadcaster.getStatus({ cwd: "/repo" });
397
398 state.currentLocalStatus = {
399 ...baseLocalStatus,
400 refName: "feature/updated-status",
401 };
402 state.currentRemoteStatus = {
403 ...baseRemoteStatus,
404 aheadCount: 2,
405 };
406 const refreshed = yield* broadcaster.refreshStatus("/repo");
407 const cached = yield* broadcaster.getStatus({ cwd: "/repo" });
408
409 assert.deepStrictEqual(initial, baseStatus);
410 assert.deepStrictEqual(refreshed, {
411 ...state.currentLocalStatus,
412 ...state.currentRemoteStatus,
413 });
414 assert.deepStrictEqual(cached, {
415 ...state.currentLocalStatus,
416 ...state.currentRemoteStatus,
417 });
418 assert.equal(state.localStatusCalls, 2);
419 assert.equal(state.remoteStatusCalls, 2);
420 assert.equal(state.localInvalidationCalls, 1);
421 assert.equal(state.remoteInvalidationCalls, 1);
422 }).pipe(Effect.provide(layerTestFor(state)));
423 });
424
425 it.effect("keeps the cached snapshot unchanged when a refresh branch fails", () => {
426 const state = {
427 currentLocalStatus: baseLocalStatus,
428 currentRemoteStatus: baseRemoteStatus,
429 localStatusCalls: 0,
430 remoteStatusCalls: 0,
431 localInvalidationCalls: 0,
432 remoteInvalidationCalls: 0,
433 failRemoteStatus: false,
434 };
435 const layerTest = VcsStatusBroadcaster.layer.pipe(
436 Layer.provideMerge(NodeServices.layer),
437 Layer.provide(layerBackgroundPolicy(() => true)),
438 Layer.provide(
439 Layer.mock(GitWorkflowService.GitWorkflowService)({
440 localStatus: () =>
441 Effect.sync(() => {
442 state.localStatusCalls += 1;
443 return state.currentLocalStatus;
444 }),
445 remoteStatus: () =>
446 Effect.suspend(() => {
447 state.remoteStatusCalls += 1;
448 return state.failRemoteStatus
449 ? Effect.fail(
450 new GitManagerError({
451 operation: "VcsStatusBroadcaster.test",
452 cwd: "/repo",
453 detail: "remote status failed",
454 }),
455 )
456 : Effect.succeed(state.currentRemoteStatus);
457 }),
458 invalidateLocalStatus: () =>
459 Effect.sync(() => {
460 state.localInvalidationCalls += 1;
461 }),
462 invalidateRemoteStatus: () =>
463 Effect.sync(() => {
464 state.remoteInvalidationCalls += 1;
465 }),
466 invalidateStatus: () =>
467 Effect.sync(() => {
468 state.localInvalidationCalls += 1;
469 state.remoteInvalidationCalls += 1;
470 }),
471 }),
472 ),
473 );
474
475 return Effect.gen(function* () {
476 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
477 yield* broadcaster.getStatus({ cwd: "/repo" });
478
479 state.currentLocalStatus = {
480 ...baseLocalStatus,
481 refName: "feature/partial-refresh",
482 };
483 state.currentRemoteStatus = {
484 ...baseRemoteStatus,
485 aheadCount: 3,
486 };
487 state.failRemoteStatus = true;
488
489 const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit);
490 const cached = yield* broadcaster.getStatus({ cwd: "/repo" });
491
492 assert.isTrue(Exit.isFailure(refreshExit));
493 assert.deepStrictEqual(cached, baseStatus);
494 }).pipe(Effect.provide(layerTest));
495 });
496
497 it.effect("refreshes only the cached local snapshot when requested", () => {
498 const state = {
499 currentLocalStatus: baseLocalStatus,
500 currentRemoteStatus: baseRemoteStatus,
501 localStatusCalls: 0,
502 remoteStatusCalls: 0,
503 localInvalidationCalls: 0,
504 remoteInvalidationCalls: 0,
505 };
506
507 return Effect.gen(function* () {
508 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
509 const initial = yield* broadcaster.getStatus({ cwd: "/repo" });
510
511 state.currentLocalStatus = {
512 ...baseLocalStatus,
513 refName: "feature/local-only-refresh",
514 hasWorkingTreeChanges: true,
515 };
516
517 const refreshedLocal = yield* broadcaster.refreshLocalStatus("/repo");
518 const cached = yield* broadcaster.getStatus({ cwd: "/repo" });
519
520 assert.deepStrictEqual(initial, baseStatus);
521 assert.deepStrictEqual(refreshedLocal, state.currentLocalStatus);
522 assert.deepStrictEqual(cached, {
523 ...state.currentLocalStatus,
524 ...baseRemoteStatus,
525 });
526 assert.equal(state.localStatusCalls, 2);
527 assert.equal(state.remoteStatusCalls, 1);
528 assert.equal(state.localInvalidationCalls, 1);
529 assert.equal(state.remoteInvalidationCalls, 0);
530 }).pipe(Effect.provide(layerTestFor(state)));
531 });
532
533 it.effect.skipIf(!symlinksSupported)(
534 "normalizes symlinked CWDs before cache lookup and workflow calls",
535 () => {
536 const seenCwds: string[] = [];
537 const state = {
538 currentLocalStatus: baseLocalStatus,
539 currentRemoteStatus: baseRemoteStatus,
540 localStatusCalls: 0,
541 remoteStatusCalls: 0,
542 localInvalidationCalls: 0,
543 remoteInvalidationCalls: 0,
544 };
545 const layerTest = VcsStatusBroadcaster.layer.pipe(
546 Layer.provideMerge(NodeServices.layer),
547 Layer.provide(layerBackgroundPolicy(() => true)),
548 Layer.provide(
549 Layer.mock(GitWorkflowService.GitWorkflowService)({
550 localStatus: (input) =>
551 Effect.sync(() => {
552 seenCwds.push(input.cwd);
553 state.localStatusCalls += 1;
554 return state.currentLocalStatus;
555 }),
556 remoteStatus: (input) =>
557 Effect.sync(() => {
558 seenCwds.push(input.cwd);
559 state.remoteStatusCalls += 1;
560 return state.currentRemoteStatus;
561 }),
562 invalidateLocalStatus: () =>
563 Effect.sync(() => {
564 state.localInvalidationCalls += 1;
565 }),
566 invalidateRemoteStatus: () =>
567 Effect.sync(() => {
568 state.remoteInvalidationCalls += 1;
569 }),
570 } satisfies Partial<GitWorkflowService.GitWorkflowService["Service"]>),
571 ),
572 );
573
574 return Effect.gen(function* () {
575 const fileSystem = yield* FileSystem.FileSystem;
576 const path = yield* Path.Path;
577 const realDir = yield* fileSystem.makeTempDirectoryScoped({
578 prefix: "t3-vcs-status-real-",
579 });
580 const linkParent = yield* fileSystem.makeTempDirectoryScoped({
581 prefix: "t3-vcs-status-link-",
582 });
583 const linkDir = path.join(linkParent, "repo-link");
584 yield* fileSystem.symlink(realDir, linkDir);
585 const realPath = yield* fileSystem.realPath(realDir);
586
587 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
588 yield* broadcaster.getStatus({ cwd: linkDir });
589 yield* broadcaster.getStatus({ cwd: realDir });
590
591 assert.deepStrictEqual(seenCwds, [realPath, realPath]);
592 assert.equal(state.localStatusCalls, 1);
593 assert.equal(state.remoteStatusCalls, 1);
594 }).pipe(Effect.provide(layerTest));
595 },
596 );
597
598 it.effect("passive streams retain cached remote status", () => {
599 const state = {
600 currentLocalStatus: baseLocalStatus,
601 currentRemoteStatus: baseRemoteStatus,
602 localStatusCalls: 0,
603 remoteStatusCalls: 0,
604 localInvalidationCalls: 0,
605 remoteInvalidationCalls: 0,
606 };
607
608 return Effect.gen(function* () {
609 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
610 const snapshotDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
611 const remoteUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
612 yield* Stream.runForEach(
613 broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }),
614 (event) => {
615 if (event._tag === "snapshot") {
616 return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore);
617 }
618 if (event._tag === "remoteUpdated") {
619 return Deferred.succeed(remoteUpdatedDeferred, event).pipe(Effect.ignore);
620 }
621 return Effect.void;
622 },
623 ).pipe(Effect.forkScoped);
624
625 const snapshot = yield* Deferred.await(snapshotDeferred);
626 yield* TestClock.adjust("1 minute");
627 assert.equal(state.remoteStatusCalls, 0);
628 yield* broadcaster.getStatus({ cwd: "/repo" });
629 state.currentRemoteStatus = remoteStatusWithPr;
630 yield* broadcaster.refreshPullRequestStatus("/repo");
631 const remoteUpdated = yield* Deferred.await(remoteUpdatedDeferred);
632
633 assert.deepStrictEqual(snapshot, {
634 _tag: "snapshot",
635 local: baseLocalStatus,
636 remote: null,
637 } satisfies VcsStatusStreamEvent);
638 assert.deepStrictEqual(remoteUpdated, {
639 _tag: "remoteUpdated",
640 remote: remoteStatusWithPr,
641 } satisfies VcsStatusStreamEvent);
642 const cachedSnapshot = yield* Stream.runHead(
643 broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }),
644 );
645 assert.deepStrictEqual(Option.getOrThrow(cachedSnapshot), {
646 _tag: "snapshot",
647 local: baseLocalStatus,
648 remote: remoteStatusWithPr,
649 });
650 }).pipe(Effect.provide(layerTestFor(state)));
651 });
652
653 it.effect("loads remote status once when periodic refreshes are disabled", () => {
654 const state = {
655 currentLocalStatus: baseLocalStatus,
656 currentRemoteStatus: remoteStatusWithPr,
657 localStatusCalls: 0,
658 remoteStatusCalls: 0,
659 localInvalidationCalls: 0,
660 remoteInvalidationCalls: 0,
661 remoteStatusRefreshUpstreamValues: [] as Array<boolean | undefined>,
662 };
663
664 return Effect.gen(function* () {
665 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
666 const scope = yield* Scope.make();
667 const snapshotDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
668 const remoteUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
669 yield* Stream.runForEach(
670 broadcaster.streamStatus(
671 { cwd: "/repo" },
672 { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) },
673 ),
674 (event) => {
675 if (event._tag === "snapshot") {
676 return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore);
677 }
678 if (event._tag === "remoteUpdated") {
679 return Deferred.succeed(remoteUpdatedDeferred, event).pipe(Effect.ignore);
680 }
681 return Effect.void;
682 },
683 ).pipe(Effect.forkIn(scope));
684
685 const snapshot = yield* Deferred.await(snapshotDeferred);
686 const remoteUpdated = yield* Deferred.await(remoteUpdatedDeferred);
687
688 assert.deepStrictEqual(snapshot, {
689 _tag: "snapshot",
690 local: baseLocalStatus,
691 remote: null,
692 } satisfies VcsStatusStreamEvent);
693 assert.deepStrictEqual(remoteUpdated, {
694 _tag: "remoteUpdated",
695 remote: remoteStatusWithPr,
696 } satisfies VcsStatusStreamEvent);
697 assert.equal(state.remoteStatusCalls, 1);
698 assert.equal(state.remoteInvalidationCalls, 0);
699 assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues, [false]);
700
701 yield* TestClock.adjust(Duration.minutes(2));
702 assert.equal(state.remoteStatusCalls, 1);
703 assert.equal(state.remoteInvalidationCalls, 0);
704
705 yield* Scope.close(scope, Exit.void);
706 }).pipe(Effect.provide(Layer.merge(layerTestFor(state), TestClock.layer())));
707 });
708
709 it.effect("retries the initial remote load when periodic refreshes are disabled", () => {
710 const state = {
711 currentLocalStatus: baseLocalStatus,
712 localStatusCalls: 0,
713 remoteStatusCalls: 0,
714 localInvalidationCalls: 0,
715 remoteInvalidationCalls: 0,
716 remoteStatusRefreshUpstreamValues: [] as Array<boolean | undefined>,
717 };
718 const privateCwd = "/private/user/workspace/repo";
719 const nestedCause = new Error("private nested VCS failure");
720 const messages: Array<ReadonlyArray<unknown>> = [];
721 const logger = Logger.make<unknown, void>(({ message }) => {
722 messages.push(message as ReadonlyArray<unknown>);
723 });
724 let firstRemoteAttemptDeferred: Deferred.Deferred<void> | null = null;
725 const layerTest = VcsStatusBroadcaster.layer.pipe(
726 Layer.provideMerge(NodeServices.layer),
727 Layer.provide(layerBackgroundPolicy(() => true)),
728 Layer.provide(
729 Layer.mock(GitWorkflowService.GitWorkflowService)({
730 localStatus: () =>
731 Effect.sync(() => {
732 state.localStatusCalls += 1;
733 return state.currentLocalStatus;
734 }),
735 remoteStatus: (_input, options) =>
736 Effect.suspend(() => {
737 state.remoteStatusCalls += 1;
738 state.remoteStatusRefreshUpstreamValues.push(options?.refreshUpstream);
739 if (state.remoteStatusCalls === 1) {
740 return Effect.fail(
741 new GitManagerError({
742 operation: "VcsStatusBroadcaster.test",
743 cwd: privateCwd,
744 detail: "private initial remote status failure",
745 cause: nestedCause,
746 }),
747 ).pipe(
748 Effect.ensuring(
749 firstRemoteAttemptDeferred
750 ? Deferred.succeed(firstRemoteAttemptDeferred, undefined).pipe(Effect.ignore)
751 : Effect.void,
752 ),
753 );
754 }
755 return Effect.succeed(remoteStatusWithPr);
756 }),
757 invalidateLocalStatus: () =>
758 Effect.sync(() => {
759 state.localInvalidationCalls += 1;
760 }),
761 invalidateRemoteStatus: () =>
762 Effect.sync(() => {
763 state.remoteInvalidationCalls += 1;
764 }),
765 }),
766 ),
767 );
768
769 return Effect.gen(function* () {
770 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
771 const scope = yield* Scope.make();
772 firstRemoteAttemptDeferred = yield* Deferred.make<void>();
773 const remoteUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
774 yield* Stream.runForEach(
775 broadcaster.streamStatus(
776 { cwd: privateCwd },
777 { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) },
778 ),
779 (event) =>
780 event._tag === "remoteUpdated"
781 ? Deferred.succeed(remoteUpdatedDeferred, event).pipe(Effect.ignore)
782 : Effect.void,
783 ).pipe(Effect.forkIn(scope));
784
785 yield* Deferred.await(firstRemoteAttemptDeferred);
786 yield* Effect.yieldNow;
787 assert.equal(state.remoteStatusCalls, 1);
788 assert.deepStrictEqual(
789 messages.find((message) => message[0] === "VCS remote status refresh failed"),
790 [
791 "VCS remote status refresh failed",
792 {
793 cwdLength: privateCwd.length,
794 reasonCount: 1,
795 failureCount: 1,
796 failureTags: ["GitManagerError"],
797 failureOperations: ["VcsStatusBroadcaster.test"],
798 defectCount: 0,
799 defectTags: [],
800 interruptionCount: 0,
801 consecutiveFailures: 1,
802 nextDelayMs: 30_000,
803 },
804 ],
805 );
806
807 yield* TestClock.adjust(Duration.seconds(30));
808 const remoteUpdated = yield* Deferred.await(remoteUpdatedDeferred);
809
810 assert.deepStrictEqual(remoteUpdated, {
811 _tag: "remoteUpdated",
812 remote: remoteStatusWithPr,
813 } satisfies VcsStatusStreamEvent);
814 assert.equal(state.remoteStatusCalls, 2);
815 assert.equal(state.remoteInvalidationCalls, 0);
816 assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues, [false, false]);
817
818 yield* Scope.close(scope, Exit.void);
819 }).pipe(
820 Effect.provide(
821 Layer.mergeAll(
822 layerTest,
823 TestClock.layer(),
824 Logger.layer([logger], { mergeWithExisting: false }),
825 ),
826 ),
827 );
828 });
829
830 it.effect("delays automatic refresh when a cached remote snapshot is available", () => {
831 const state = {
832 currentLocalStatus: baseLocalStatus,
833 currentRemoteStatus: baseRemoteStatus,
834 localStatusCalls: 0,
835 remoteStatusCalls: 0,
836 localInvalidationCalls: 0,
837 remoteInvalidationCalls: 0,
838 };
839
840 return Effect.gen(function* () {
841 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
842 yield* broadcaster.getStatus({ cwd: "/repo" });
843 const scope = yield* Scope.make();
844 const snapshotDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
845 yield* Stream.runForEach(
846 broadcaster.streamStatus(
847 { cwd: "/repo" },
848 { automaticRemoteRefreshInterval: Effect.succeed(Duration.minutes(1)) },
849 ),
850 (event) =>
851 event._tag === "snapshot"
852 ? Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore)
853 : Effect.void,
854 ).pipe(Effect.forkIn(scope));
855
856 yield* Deferred.await(snapshotDeferred);
857 assert.equal(state.remoteStatusCalls, 1);
858 assert.equal(state.remoteInvalidationCalls, 0);
859
860 yield* TestClock.adjust(Duration.seconds(59));
861 assert.equal(state.remoteStatusCalls, 1);
862
863 yield* TestClock.adjust(Duration.seconds(1));
864 yield* Effect.yieldNow;
865 assert.equal(state.remoteStatusCalls, 2);
866 assert.equal(state.remoteInvalidationCalls, 1);
867
868 yield* Scope.close(scope, Exit.void);
869 }).pipe(Effect.provide(Layer.merge(layerTestFor(state), TestClock.layer())));
870 });
871
872 // A push from a terminal moves ahead; a PR merged on the host moves ahead-of-default.
873 it.effect.each([
874 ["a push", { ...baseRemoteStatus, aheadCount: 1, aheadOfDefaultCount: 0 }],
875 ["a merged pull request", { ...baseRemoteStatus, aheadOfDefaultCount: 2 }],
876 ] as const)("re-reads local status when a fetch reflects %s", ([, initialRemote]) => {
877 const state = {
878 currentLocalStatus: baseLocalStatus,
879 currentRemoteStatus: initialRemote as VcsStatusRemoteResult | null,
880 localStatusCalls: 0,
881 remoteStatusCalls: 0,
882 localInvalidationCalls: 0,
883 remoteInvalidationCalls: 0,
884 };
885
886 return Effect.gen(function* () {
887 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
888 yield* broadcaster.getStatus({ cwd: "/repo" });
889 const scope = yield* Scope.make();
890 const snapshotDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
891 const localUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();
892 yield* Stream.runForEach(
893 broadcaster.streamStatus(
894 { cwd: "/repo" },
895 { automaticRemoteRefreshInterval: Effect.succeed(Duration.minutes(1)) },
896 ),
897 (event) =>
898 event._tag === "snapshot"
899 ? Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore)
900 : event._tag === "localUpdated"
901 ? Deferred.succeed(localUpdatedDeferred, event).pipe(Effect.ignore)
902 : Effect.void,
903 ).pipe(Effect.forkIn(scope));
904 yield* Deferred.await(snapshotDeferred);
905
906 // The next fetch moves the base, so the Changes totals shrink.
907 const pushedLocal: VcsStatusLocalResult = {
908 ...baseLocalStatus,
909 branchChanges: { baseRef: "origin/main", insertions: 0, deletions: 0 },
910 };
911 state.currentLocalStatus = pushedLocal;
912 state.currentRemoteStatus = { ...baseRemoteStatus, aheadOfDefaultCount: 0 };
913 yield* TestClock.adjust(Duration.minutes(1));
914
915 assert.deepStrictEqual(yield* Deferred.await(localUpdatedDeferred), {
916 _tag: "localUpdated",
917 local: pushedLocal,
918 } satisfies VcsStatusStreamEvent);
919 yield* Scope.close(scope, Exit.void);
920 }).pipe(Effect.provide(Layer.merge(layerTestFor(state), TestClock.layer())));
921 });
922
923 it.effect("an explicit refresh reads local totals after the fetch", () => {
924 const state: Parameters<typeof layerTestFor>[0] = {
925 currentLocalStatus: baseLocalStatus,
926 currentRemoteStatus: baseRemoteStatus,
927 localStatusCalls: 0,
928 remoteStatusCalls: 0,
929 localInvalidationCalls: 0,
930 remoteInvalidationCalls: 0,
931 };
932
933 return Effect.gen(function* () {
934 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
935 const fetchStarted = yield* Deferred.make<void>();
936 const finishFetch = yield* Deferred.make<void>();
937 state.beforeRemoteStatus = Deferred.succeed(fetchStarted, undefined).pipe(
938 Effect.andThen(Deferred.await(finishFetch)),
939 );
940 const refresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);
941 yield* Deferred.await(fetchStarted);
942
943 // The fetch moves the base while it runs, so totals read before it ends would be stale.
944 const fetchedLocal: VcsStatusLocalResult = {
945 ...baseLocalStatus,
946 branchChanges: { baseRef: "origin/main", insertions: 0, deletions: 0 },
947 };
948 state.currentLocalStatus = fetchedLocal;
949 yield* Deferred.succeed(finishFetch, undefined);
950
951 const status = yield* Fiber.join(refresh);
952 assert.deepStrictEqual(status.branchChanges, fetchedLocal.branchChanges);
953 }).pipe(Effect.provide(layerTestFor(state)));
954 });
955
956 it("backs off remote refresh failures exponentially and honors larger configured intervals", () => {
957 assert.equal(
958 Duration.toMillis(VcsStatusBroadcaster.remoteRefreshFailureDelay(1, Duration.seconds(1))),
959 30_000,
960 );
961 assert.equal(
962 Duration.toMillis(VcsStatusBroadcaster.remoteRefreshFailureDelay(2, Duration.seconds(1))),
963 60_000,
964 );
965 assert.equal(
966 Duration.toMillis(VcsStatusBroadcaster.remoteRefreshFailureDelay(3, Duration.seconds(1))),
967 120_000,
968 );
969 assert.equal(
970 Duration.toMillis(VcsStatusBroadcaster.remoteRefreshFailureDelay(1, Duration.minutes(5))),
971 300_000,
972 );
973 assert.equal(
974 Duration.toMillis(VcsStatusBroadcaster.remoteRefreshFailureDelay(20, Duration.seconds(1))),
975 900_000,
976 );
977 });
978
979 it("summarizes refresh causes without exposing nested failure details", () => {
980 const nestedCause = new Error("private nested failure detail");
981 const failure = new GitManagerError({
982 operation: "VcsStatusBroadcaster.remoteStatus",
983 cwd: "/private/user/workspace/repo",
984 detail: "private Git failure detail",
985 cause: nestedCause,
986 });
987 const cause = Cause.combine(Cause.fail(failure), Cause.die(new TypeError("private defect")));
988
989 assert.deepStrictEqual(VcsStatusBroadcaster.remoteRefreshFailureDiagnostics(cause), {
990 reasonCount: 2,
991 failureCount: 1,
992 failureTags: ["GitManagerError"],
993 failureOperations: ["VcsStatusBroadcaster.remoteStatus"],
994 defectCount: 1,
995 defectTags: ["TypeError"],
996 interruptionCount: 0,
997 });
998 });
999
1000 it.effect("does not start automatic remote refreshes without foreground client demand", () => {
1001 const state = {
1002 currentLocalStatus: baseLocalStatus,
1003 currentRemoteStatus: baseRemoteStatus,
1004 localStatusCalls: 0,
1005 remoteStatusCalls: 0,
1006 localInvalidationCalls: 0,
1007 remoteInvalidationCalls: 0,
1008 };
1009 const layerTest = VcsStatusBroadcaster.layer.pipe(
1010 Layer.provideMerge(NodeServices.layer),
1011 Layer.provide(layerBackgroundPolicy(() => false)),
1012 Layer.provide(
1013 Layer.mock(GitWorkflowService.GitWorkflowService)({
1014 localStatus: () =>
1015 Effect.sync(() => {
1016 state.localStatusCalls += 1;
1017 return state.currentLocalStatus;
1018 }),
1019 remoteStatus: () =>
1020 Effect.sync(() => {
1021 state.remoteStatusCalls += 1;
1022 return state.currentRemoteStatus;
1023 }),
1024 invalidateLocalStatus: () =>
1025 Effect.sync(() => {
1026 state.localInvalidationCalls += 1;
1027 }),
1028 invalidateRemoteStatus: () =>
1029 Effect.sync(() => {
1030 state.remoteInvalidationCalls += 1;
1031 }),
1032 } satisfies Partial<GitWorkflowService.GitWorkflowService["Service"]>),
1033 ),
1034 );
1035
1036 return Effect.gen(function* () {
1037 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
1038 const snapshot = yield* Stream.runHead(
1039 broadcaster.streamStatus(
1040 { cwd: "/repo" },
1041 { automaticRemoteRefreshInterval: Effect.succeed(Duration.seconds(1)) },
1042 ),
1043 );
1044
1045 assert.isTrue(Option.isSome(snapshot));
1046 assert.equal(state.remoteStatusCalls, 0);
1047 assert.equal(state.remoteInvalidationCalls, 0);
1048 }).pipe(Effect.provide(layerTest));
1049 });
1050
1051 it.effect("releases remote demand while passive observers remain", () => {
1052 const state = {
1053 currentLocalStatus: baseLocalStatus,
1054 currentRemoteStatus: baseRemoteStatus,
1055 localStatusCalls: 0,
1056 remoteStatusCalls: 0,
1057 localInvalidationCalls: 0,
1058 remoteInvalidationCalls: 0,
1059 };
1060 let remoteInterruptedDeferred: Deferred.Deferred<void, never> | null = null;
1061 let remoteStartedDeferred: Deferred.Deferred<void, never> | null = null;
1062 const layerTest = VcsStatusBroadcaster.layer.pipe(
1063 Layer.provideMerge(NodeServices.layer),
1064 Layer.provide(layerBackgroundPolicy(() => true)),
1065 Layer.provide(
1066 Layer.mock(GitWorkflowService.GitWorkflowService)({
1067 localStatus: () =>
1068 Effect.sync(() => {
1069 state.localStatusCalls += 1;
1070 return state.currentLocalStatus;
1071 }),
1072 remoteStatus: () =>
1073 Effect.sync(() => {
1074 state.remoteStatusCalls += 1;
1075 }).pipe(
1076 Effect.andThen(
1077 remoteStartedDeferred
1078 ? Deferred.succeed(remoteStartedDeferred, undefined).pipe(Effect.ignore)
1079 : Effect.void,
1080 ),
1081 Effect.andThen(Effect.never as Effect.Effect<VcsStatusRemoteResult | null, never>),
1082 Effect.onInterrupt(() =>
1083 remoteInterruptedDeferred
1084 ? Deferred.succeed(remoteInterruptedDeferred, undefined).pipe(Effect.ignore)
1085 : Effect.void,
1086 ),
1087 ),
1088 invalidateLocalStatus: () =>
1089 Effect.sync(() => {
1090 state.localInvalidationCalls += 1;
1091 }),
1092 invalidateRemoteStatus: () =>
1093 Effect.sync(() => {
1094 state.remoteInvalidationCalls += 1;
1095 }),
1096 } satisfies Partial<GitWorkflowService.GitWorkflowService["Service"]>),
1097 ),
1098 );
1099
1100 return Effect.gen(function* () {
1101 const remoteInterrupted = yield* Deferred.make<void>();
1102 const remoteStarted = yield* Deferred.make<void>();
1103 remoteInterruptedDeferred = remoteInterrupted;
1104 remoteStartedDeferred = remoteStarted;
1105
1106 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
1107 const firstSnapshot = yield* Deferred.make<VcsStatusStreamEvent>();
1108 const secondSnapshot = yield* Deferred.make<VcsStatusStreamEvent>();
1109 const firstScope = yield* Scope.make();
1110 const secondScope = yield* Scope.make();
1111 const passiveSnapshot = yield* Deferred.make<VcsStatusStreamEvent>();
1112 yield* Stream.runForEach(
1113 broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }),
1114 (event) =>
1115 event._tag === "snapshot"
1116 ? Deferred.succeed(passiveSnapshot, event).pipe(Effect.ignore)
1117 : Effect.void,
1118 ).pipe(Effect.forkScoped);
1119 yield* Deferred.await(passiveSnapshot);
1120 assert.equal(state.remoteStatusCalls, 0);
1121 yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) =>
1122 event._tag === "snapshot"
1123 ? Deferred.succeed(firstSnapshot, event).pipe(Effect.ignore)
1124 : Effect.void,
1125 ).pipe(Effect.forkIn(firstScope));
1126 yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) =>
1127 event._tag === "snapshot"
1128 ? Deferred.succeed(secondSnapshot, event).pipe(Effect.ignore)
1129 : Effect.void,
1130 ).pipe(Effect.forkIn(secondScope));
1131
1132 yield* Deferred.await(firstSnapshot);
1133 yield* Deferred.await(secondSnapshot);
1134 yield* Deferred.await(remoteStarted);
1135
1136 assert.equal(state.remoteStatusCalls, 1);
1137
1138 yield* Scope.close(firstScope, Exit.void);
1139 assert.isTrue(Option.isNone(yield* Deferred.poll(remoteInterrupted)));
1140
1141 yield* Scope.close(secondScope, Exit.void).pipe(Effect.forkScoped);
1142 yield* Deferred.await(remoteInterrupted);
1143 assert.isTrue(Option.isSome(yield* Deferred.poll(remoteInterrupted)));
1144 }).pipe(Effect.provide(layerTest));
1145 });
1146});