1import * as Cause from "effect/Cause";
2import * as Context from "effect/Context";
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 FileSystem from "effect/FileSystem";
9import * as Layer from "effect/Layer";
10import * as Path from "effect/Path";
11import * as PubSub from "effect/PubSub";
12import * as Ref from "effect/Ref";
13import * as Schedule from "effect/Schedule";
14import * as Scope from "effect/Scope";
15import * as Stream from "effect/Stream";
16import * as SynchronizedRef from "effect/SynchronizedRef";
17import type {
18 GitManagerServiceError,
19 VcsStatusInput,
20 VcsBranchStatus,
21 VcsStatusLocalResult,
22 VcsStatusRemoteResult,
23 VcsStatusResult,
24 VcsStatusStreamEvent,
25 VcsStatusSubscriptionInput,
26} from "@t3tools/contracts";
27import { DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL } from "@t3tools/contracts";
28import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
29import { mergeGitStatusParts } from "@t3tools/shared/git";
30import * as KeyedLock from "@t3tools/shared/KeyedLock";
31import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
32
33import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
34import * as GitWorkflowService from "../git/GitWorkflowService.ts";
35import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
36import * as ServerSettings from "../serverSettings.ts";
37
38const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30);
39const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15);
40const MAX_FAILURE_DIAGNOSTIC_VALUES = 8;
41// A held index.lock fails the read, and its removal moves no HEAD to trigger another.
42const CHECKOUT_REFRESH_RETRY = Schedule.exponential(Duration.seconds(1)).pipe(
43 Schedule.modifyDelay(({ duration }) =>
44 Effect.succeed(Duration.min(duration, Duration.seconds(30))),
45 ),
46);
47const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128;
48
49function boundedDiagnosticValue(value: string): string {
50 return value.slice(0, MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH);
51}
52
53function diagnosticValueTag(value: unknown): string {
54 try {
55 if (
56 typeof value === "object" &&
57 value !== null &&
58 "_tag" in value &&
59 typeof value._tag === "string"
60 ) {
61 return boundedDiagnosticValue(value._tag);
62 }
63 if (value instanceof Error) {
64 return boundedDiagnosticValue(value.name);
65 }
66 return typeof value;
67 } catch {
68 return "Uninspectable";
69 }
70}
71
72function diagnosticFailureOperation(value: unknown): string | undefined {
73 try {
74 if (
75 typeof value === "object" &&
76 value !== null &&
77 "operation" in value &&
78 typeof value.operation === "string"
79 ) {
80 return boundedDiagnosticValue(value.operation);
81 }
82 } catch {
83 return undefined;
84 }
85 return undefined;
86}
87
88function addUniqueDiagnosticValue(values: Array<string>, value: string | undefined): void {
89 if (
90 value !== undefined &&
91 values.length < MAX_FAILURE_DIAGNOSTIC_VALUES &&
92 !values.includes(value)
93 ) {
94 values.push(value);
95 }
96}
97
98function remoteRefreshFailureDiagnostics(cause: Cause.Cause<unknown>) {
99 const failureTags: Array<string> = [];
100 const failureOperations: Array<string> = [];
101 const defectTags: Array<string> = [];
102 let failureCount = 0;
103 let defectCount = 0;
104 let interruptionCount = 0;
105
106 for (const reason of cause.reasons) {
107 if (Cause.isFailReason(reason)) {
108 failureCount += 1;
109 addUniqueDiagnosticValue(failureTags, diagnosticValueTag(reason.error));
110 addUniqueDiagnosticValue(failureOperations, diagnosticFailureOperation(reason.error));
111 continue;
112 }
113 if (Cause.isDieReason(reason)) {
114 defectCount += 1;
115 addUniqueDiagnosticValue(defectTags, diagnosticValueTag(reason.defect));
116 continue;
117 }
118 interruptionCount += 1;
119 }
120
121 return {
122 reasonCount: cause.reasons.length,
123 failureCount,
124 failureTags,
125 failureOperations,
126 defectCount,
127 defectTags,
128 interruptionCount,
129 };
130}
131
132interface VcsStatusChange {
133 readonly cwd: string;
134 readonly event: VcsStatusStreamEvent;
135}
136
137interface CachedValue<T> {
138 readonly fingerprint: string;
139 readonly value: T;
140}
141
142interface CachedRemoteStatus extends CachedValue<VcsStatusRemoteResult | null> {
143 readonly branch: string | null;
144}
145
146interface CachedVcsStatus {
147 readonly local: CachedValue<VcsStatusLocalResult> | null;
148 readonly remote: CachedRemoteStatus | null;
149}
150
151interface ActiveHeadWatcher {
152 readonly fiber: Fiber.Fiber<void, never>;
153 readonly ready: Deferred.Deferred<void>;
154 readonly subscriberCount: number;
155}
156
157interface ActiveRemotePoller {
158 readonly fiber: Fiber.Fiber<void, never>;
159 readonly subscriberCount: number;
160 readonly demandCwds: Ref.Ref<ReadonlyMap<string, number>>;
161}
162
163export class VcsAutomaticFetchInterval extends Context.Reference<Effect.Effect<Duration.Duration>>(
164 "t3/vcs/VcsAutomaticFetchInterval",
165 { defaultValue: () => Effect.succeed(DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL) },
166) {}
167
168export const layerAutomaticFetchInterval = Layer.effect(
169 VcsAutomaticFetchInterval,
170 Effect.gen(function* () {
171 const serverSettings = yield* ServerSettings.ServerSettingsService;
172 return serverSettings.getSettings.pipe(
173 Effect.map(
174 (settings) => resolveServerBackgroundActivitySettings(settings).automaticGitFetchInterval,
175 ),
176 Effect.orElseSucceed(() => DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL),
177 );
178 }),
179);
180
181export class VcsAutoPullPolicy extends Context.Reference<{
182 readonly isEnabled: (cwd: string) => Effect.Effect<boolean, never>;
183}>("t3/vcs/VcsAutoPullPolicy", {
184 defaultValue: () => ({ isEnabled: () => Effect.succeed(false) }),
185}) {}
186
187export const layerAutoPullPolicy = Layer.effect(
188 VcsAutoPullPolicy,
189 Effect.gen(function* () {
190 const projects = yield* ProjectStore.ProjectStoreV2;
191 const serverSettings = yield* ServerSettings.ServerSettingsService;
192 return {
193 isEnabled: Effect.fn("VcsAutoPullPolicy.isEnabled")(
194 function* (cwd: string) {
195 const project = yield* projects.findActiveByWorkspaceRoot(cwd);
196 if (project._tag === "None") return false;
197 const settings = yield* serverSettings.getSettings;
198 return resolveProjectSettings(settings, project.value.projectId).settings.defaultAutoPull;
199 },
200 Effect.orElseSucceed(() => false),
201 ),
202 };
203 }),
204);
205
206function remoteRefreshFailureDelay(
207 consecutiveFailures: number,
208 configuredInterval: Duration.Duration,
209) {
210 const exponent = Math.max(0, consecutiveFailures - 1);
211 const backoffMs =
212 Duration.toMillis(VCS_STATUS_REFRESH_FAILURE_BASE_DELAY) * Math.pow(2, exponent);
213 const cappedBackoff = Duration.min(
214 Duration.millis(backoffMs),
215 VCS_STATUS_REFRESH_FAILURE_MAX_DELAY,
216 );
217 return Duration.max(configuredInterval, cappedBackoff);
218}
219
220export class VcsStatusBroadcaster extends Context.Service<
221 VcsStatusBroadcaster,
222 {
223 readonly streamBranchStatus: (
224 input: VcsStatusInput,
225 ) => Stream.Stream<VcsBranchStatus, GitManagerServiceError>;
226 readonly refreshBranchStatus: (
227 input: VcsStatusInput,
228 ) => Effect.Effect<VcsBranchStatus, GitManagerServiceError>;
229 readonly streamBranchChanges: Stream.Stream<{
230 readonly cwd: string;
231 readonly refName: string | null;
232 }>;
233 readonly getStatus: (
234 input: VcsStatusInput,
235 ) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;
236 readonly refreshLocalStatus: (
237 cwd: string,
238 ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>;
239 readonly refreshStatus: (
240 cwd: string,
241 options?: { readonly localOnly?: boolean },
242 ) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;
243 readonly refreshPullRequestStatus: (
244 cwd: string,
245 ) => Effect.Effect<VcsStatusRemoteResult | null, GitManagerServiceError>;
246 readonly streamStatus: (
247 input: VcsStatusSubscriptionInput,
248 ) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>;
249 }
250>()("t3/vcs/VcsStatusBroadcaster") {}
251
252function fingerprintStatusPart(status: unknown): string {
253 return JSON.stringify(status);
254}
255
256const normalizeCwd = (cwd: string) =>
257 Effect.service(FileSystem.FileSystem).pipe(
258 Effect.flatMap((fs) => fs.realPath(cwd)),
259 Effect.orElseSucceed(() => cwd),
260 );
261
262/** @public */
263export const make = Effect.gen(function* () {
264 const automaticFetchInterval = yield* VcsAutomaticFetchInterval;
265 const autoPullPolicy = yield* VcsAutoPullPolicy;
266 const workflow = yield* GitWorkflowService.GitWorkflowService;
267 const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
268 const fs = yield* FileSystem.FileSystem;
269 const path = yield* Path.Path;
270 const changesPubSub = yield* Effect.acquireRelease(
271 PubSub.unbounded<VcsStatusChange>(),
272 (pubsub) => PubSub.shutdown(pubsub),
273 );
274 const branchChangesPubSub = yield* Effect.acquireRelease(
275 PubSub.unbounded<{
276 readonly cwd: string;
277 readonly refName: string | null;
278 readonly isRepo: boolean;
279 readonly revision: number;
280 }>(),
281 (pubsub) => PubSub.shutdown(pubsub),
282 );
283 const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
284 Scope.close(scope, Exit.void),
285 );
286 const detailedSubscribers = new Map<string, number>();
287 const branches = new Map<
288 string,
289 { readonly refName: string | null; readonly isRepo: boolean; readonly revision: number }
290 >();
291 let branchRevision = 0;
292 const publishBranch = Effect.fn("VcsStatusBroadcaster.publishBranch")(function* (
293 cwd: string,
294 { refName, isRepo }: { readonly refName: string | null; readonly isRepo: boolean },
295 ) {
296 const previous = branches.get(cwd);
297 if (previous?.refName === refName && previous.isRepo === isRepo) return;
298 const revision = ++branchRevision;
299 branches.set(cwd, { refName, isRepo, revision });
300 yield* PubSub.publish(branchChangesPubSub, { cwd, refName, isRepo, revision });
301 });
302 const localRefreshLocks = yield* KeyedLock.make<string>();
303 const cacheRef = yield* Ref.make(new Map<string, CachedVcsStatus>());
304 // One permit per cwd for remote reads that write the cache. Without it a
305 // periodic poll that started before an agent creates a PR can finish after the
306 // turn-end refresh and overwrite the fresh PR with its stale `pr: null`.
307 const remoteWriteLocks = yield* KeyedLock.make<string>();
308 const withRemoteWriteLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) =>
309 remoteWriteLocks.withLock(cwd, effect);
310 const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>());
311 const headWatchersRef = yield* SynchronizedRef.make(new Map<string, ActiveHeadWatcher>());
312
313 const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* (
314 cwd: string,
315 ) {
316 return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null));
317 });
318
319 const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")(
320 function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) {
321 const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local };
322 const [localChanged, remoteCleared] = yield* Ref.modify(cacheRef, (cache) => {
323 const previous = cache.get(cwd) ?? { local: null, remote: null };
324 const remote = previous.remote?.branch === local.refName ? previous.remote : null;
325 return [
326 [previous.local?.fingerprint !== nextLocal.fingerprint, previous.remote !== remote],
327 new Map(cache).set(cwd, { local: nextLocal, remote }),
328 ] as const;
329 });
330 yield* publishBranch(cwd, local);
331 if (options?.publish && remoteCleared) {
332 yield* PubSub.publish(changesPubSub, {
333 cwd,
334 event: { _tag: "remoteUpdated", remote: null },
335 });
336 }
337 if (options?.publish && localChanged) {
338 yield* PubSub.publish(changesPubSub, {
339 cwd,
340 event: { _tag: "localUpdated", local },
341 });
342 }
343 return local;
344 },
345 );
346
347 const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")(
348 function* (cwd: string, remote: CachedRemoteStatus | null, options?: { publish?: boolean }) {
349 const [changed, value] = yield* Ref.modify(cacheRef, (cache) => {
350 const previous = cache.get(cwd) ?? { local: null, remote: null };
351 const next = remote?.branch === previous.local?.value.refName ? remote : null;
352 return [
353 [previous.remote?.fingerprint !== next?.fingerprint, next?.value ?? null],
354 new Map(cache).set(cwd, { ...previous, remote: next }),
355 ] as const;
356 });
357 if (options?.publish && changed) {
358 yield* PubSub.publish(changesPubSub, {
359 cwd,
360 event: { _tag: "remoteUpdated", remote: value },
361 });
362 }
363 return value;
364 },
365 );
366
367 const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (
368 cwd: string,
369 ) {
370 const local = yield* workflow.localStatus({ cwd });
371 return yield* updateCachedLocalStatus(cwd, local);
372 });
373
374 const getOrLoadLocalStatusUnlocked = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(
375 function* (cwd: string) {
376 const cached = yield* getCachedStatus(cwd);
377 if (cached?.local) {
378 const branch = yield* workflow.currentBranch(cwd);
379 if (branch === cached.local.value.refName) return cached.local.value;
380 yield* workflow.invalidateLocalStatus(cwd);
381 return yield* updateCachedLocalStatus(cwd, yield* workflow.localStatus({ cwd }), {
382 publish: true,
383 });
384 }
385 return yield* loadLocalStatus(cwd);
386 },
387 );
388
389 const getOrLoadLocalStatus = (cwd: string) =>
390 localRefreshLocks.withLock(cwd, getOrLoadLocalStatusUnlocked(cwd));
391
392 const getStatusParts = Effect.fn("VcsStatusBroadcaster.getStatusParts")(function* (cwd: string) {
393 const cached = yield* getCachedStatus(cwd);
394 if (cached?.local) {
395 return { local: cached.local.value, remote: cached.remote?.value ?? null };
396 }
397 return { local: yield* getOrLoadLocalStatus(cwd), remote: null };
398 });
399
400 const withFileSystem = Effect.provideService(FileSystem.FileSystem, fs);
401
402 const getStatus: VcsStatusBroadcaster["Service"]["getStatus"] = Effect.fn(
403 "VcsStatusBroadcaster.getStatus",
404 )(function* (input) {
405 const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
406 yield* getOrLoadLocalStatus(cwd);
407 const cached = yield* getCachedStatus(cwd);
408 if (cached?.local && cached.remote) {
409 return mergeGitStatusParts(cached.local.value, cached.remote.value);
410 }
411 return yield* withRemoteWriteLock(
412 cwd,
413 Effect.gen(function* () {
414 const latest = yield* getCachedStatus(cwd);
415 const refreshUpstream = !Duration.isZero(yield* automaticFetchInterval);
416 const remote = latest?.remote ?? (yield* readRemoteStatus(cwd, { refreshUpstream }));
417 yield* updateCachedRemoteStatus(cwd, remote);
418 const parts = yield* getStatusParts(cwd);
419 return mergeGitStatusParts(parts.local, parts.remote);
420 }),
421 );
422 });
423
424 // A refresh that started after a request saw every change that request could mean, so queued
425 // callers share it instead of each running git status again.
426 const localRefreshRequests = new Map<string, number>();
427 const localRefreshResults = new Map<
428 string,
429 { readonly startedAt: number; readonly local: VcsStatusLocalResult }
430 >();
431 const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")(
432 function* (cwd: string) {
433 const requestedAt = (localRefreshRequests.get(cwd) ?? 0) + 1;
434 localRefreshRequests.set(cwd, requestedAt);
435 return yield* localRefreshLocks.withLock(
436 cwd,
437 Effect.gen(function* () {
438 const shared = localRefreshResults.get(cwd);
439 if (shared !== undefined && shared.startedAt >= requestedAt) return shared.local;
440 const startedAt = localRefreshRequests.get(cwd) ?? requestedAt;
441 yield* workflow.invalidateLocalStatus(cwd);
442 const local = yield* workflow.localStatus({ cwd });
443 localRefreshResults.set(cwd, { startedAt, local });
444 return yield* updateCachedLocalStatus(cwd, local, { publish: true });
445 }),
446 );
447 },
448 );
449
450 const refreshLocalStatus: VcsStatusBroadcaster["Service"]["refreshLocalStatus"] = Effect.fn(
451 "VcsStatusBroadcaster.refreshLocalStatus",
452 )(function* (rawCwd) {
453 const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
454 const local = yield* refreshLocalStatusCore(cwd);
455 const cached = yield* getCachedStatus(cwd);
456 // Divergence is local Git data; a slow forge lookup must not delay turn-end updates.
457 if (cached?.remote?.value?.hasUpstream) {
458 const divergence = yield* workflow.upstreamDivergence(cwd);
459 const latest = yield* getCachedStatus(cwd);
460 if (divergence && latest?.remote?.value && latest.remote.branch === divergence.refName) {
461 const value = {
462 ...latest.remote.value,
463 aheadCount: divergence.aheadCount,
464 behindCount: divergence.behindCount,
465 };
466 yield* updateCachedRemoteStatus(
467 cwd,
468 {
469 branch: divergence.refName,
470 value,
471 fingerprint: fingerprintStatusPart(value),
472 },
473 { publish: true },
474 );
475 }
476 }
477 return local;
478 });
479
480 const readRemoteStatus = Effect.fn("VcsStatusBroadcaster.readRemoteStatus")(function* (
481 cwd: string,
482 options?: Parameters<GitWorkflowService.GitWorkflowService["Service"]["remoteStatus"]>[1],
483 ) {
484 const branch = yield* workflow.currentBranch(cwd);
485 // GitManager's remote TTL cache is keyed by cwd, not the checked-out branch.
486 yield* workflow.invalidateRemoteStatus(cwd);
487 const previousRemote = (yield* getCachedStatus(cwd))?.remote;
488 const remote = yield* workflow.remoteStatus({ cwd }, options);
489 const latest = yield* getCachedStatus(cwd);
490 // An overlapping turn refresh and fetch can both move divergence during the forge lookup.
491 const refreshed =
492 remote?.hasUpstream && latest?.remote !== previousRemote && latest?.remote?.branch === branch
493 ? yield* workflow.upstreamDivergence(cwd)
494 : null;
495 const value =
496 remote && refreshed?.refName === branch
497 ? { ...remote, aheadCount: refreshed.aheadCount, behindCount: refreshed.behindCount }
498 : remote;
499 const currentBranch = yield* workflow.currentBranch(cwd);
500 const cached = yield* getCachedStatus(cwd);
501 if (branch !== currentBranch || cached?.local?.value.refName !== currentBranch) {
502 yield* refreshLocalStatusCore(cwd);
503 }
504 return branch === currentBranch
505 ? { branch, value, fingerprint: fingerprintStatusPart(value) }
506 : null;
507 });
508
509 const maybeAutoPull = Effect.fn("VcsStatusBroadcaster.maybeAutoPull")(function* (
510 cwd: string,
511 remoteRead: CachedRemoteStatus | null,
512 policyCwds: ReadonlyArray<string>,
513 ) {
514 const remote = remoteRead?.value ?? null;
515 let pullAttempted = false;
516 return yield* Effect.gen(function* () {
517 const autoPullEnabled = (yield* Effect.forEach(policyCwds, autoPullPolicy.isEnabled, {
518 concurrency: "unbounded",
519 })).some(Boolean);
520 if (
521 remote === null ||
522 !remote.hasUpstream ||
523 remote.aheadCount > 0 ||
524 remote.behindCount <= 0 ||
525 !autoPullEnabled
526 ) {
527 return null;
528 }
529
530 yield* workflow.invalidateLocalStatus(cwd);
531 const local = yield* workflow.localStatus({ cwd });
532 if (
533 !local.isRepo ||
534 !local.isDefaultRef ||
535 local.hasWorkingTreeChanges ||
536 local.refName !== remoteRead?.branch
537 )
538 return null;
539
540 pullAttempted = true;
541 yield* workflow.pullCurrentBranch(cwd);
542 yield* workflow.invalidateStatus(cwd);
543 yield* refreshLocalStatusCore(cwd);
544 const refreshedRemote = yield* readRemoteStatus(cwd, { refreshUpstream: false });
545 yield* updateCachedRemoteStatus(cwd, refreshedRemote, { publish: true });
546 return yield* getStatusParts(cwd);
547 }).pipe(
548 Effect.catch((error) =>
549 Effect.gen(function* () {
550 yield* Effect.logWarning("Automatic project pull failed", {
551 cwdLength: cwd.length,
552 errorTag: error._tag,
553 });
554 if (!pullAttempted) return null;
555 // A pull or its follow-up reads can fail after local state has changed.
556 // Let a failed reconciliation reach the caller instead of publishing old status.
557 yield* refreshLocalStatusCore(cwd);
558 yield* updateCachedRemoteStatus(cwd, remoteRead, { publish: true });
559 return yield* getStatusParts(cwd);
560 }),
561 ),
562 );
563 });
564
565 const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (
566 cwd: string,
567 options?: {
568 readonly refreshUpstream?: boolean;
569 readonly policyCwds?: ReadonlyArray<string>;
570 },
571 ) {
572 return yield* withRemoteWriteLock(
573 cwd,
574 Effect.gen(function* () {
575 const remoteRead = yield* readRemoteStatus(cwd, options);
576 const pulled =
577 options?.refreshUpstream === false
578 ? null
579 : yield* maybeAutoPull(cwd, remoteRead, options?.policyCwds ?? [cwd]);
580 if (pulled !== null) return pulled.remote;
581 // Base refs and merge bases can move without changing divergence counts.
582 yield* refreshLocalStatusCore(cwd);
583 return yield* updateCachedRemoteStatus(cwd, remoteRead, { publish: true });
584 }),
585 );
586 });
587
588 const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn(
589 "VcsStatusBroadcaster.refreshStatus",
590 )(function* (rawCwd, options) {
591 const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
592 yield* refreshLocalStatusCore(cwd);
593 if (options?.localOnly) {
594 // Never cache a remote here: readers take `value: null` as "read and found no PR".
595 const parts = yield* getStatusParts(cwd);
596 return mergeGitStatusParts(parts.local, parts.remote);
597 }
598 return yield* withRemoteWriteLock(
599 cwd,
600 Effect.gen(function* () {
601 const refreshUpstream = !Duration.isZero(yield* automaticFetchInterval);
602 // Bump the PR lookup epoch after earlier remote writers have finished.
603 yield* workflow.invalidateStatus(cwd);
604 const remoteRead = yield* readRemoteStatus(cwd, { refreshUpstream });
605 // Fetch can move the base used for local Changes totals.
606 yield* refreshLocalStatusCore(cwd);
607 // Without a fetch, the cached behind count cannot justify an automatic pull.
608 const pulled = refreshUpstream ? yield* maybeAutoPull(cwd, remoteRead, [rawCwd]) : null;
609 if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote);
610 yield* updateCachedRemoteStatus(cwd, remoteRead, { publish: true });
611 const parts = yield* getStatusParts(cwd);
612 return mergeGitStatusParts(parts.local, parts.remote);
613 }),
614 );
615 });
616
617 const refreshPullRequestStatus: VcsStatusBroadcaster["Service"]["refreshPullRequestStatus"] =
618 Effect.fn("VcsStatusBroadcaster.refreshPullRequestStatus")(function* (rawCwd) {
619 const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
620 return yield* withRemoteWriteLock(
621 cwd,
622 Effect.gen(function* () {
623 const cached = yield* getCachedStatus(cwd);
624 if (!cached || cached.remote?.value === null) return null;
625 // Turn completion must update divergence even when background polling is paused.
626 const remoteRead = yield* readRemoteStatus(cwd, {
627 refreshUpstream: false,
628 refreshMissingPullRequest: true,
629 });
630 return yield* updateCachedRemoteStatus(cwd, remoteRead, { publish: true });
631 }),
632 );
633 });
634
635 // A subscription owns this loop; keep each poll's spans out of a never-ending parent span.
636 const makeRemoteRefreshLoop = Effect.fnUntraced(function* (
637 cwd: string,
638 demandCwdsRef: Ref.Ref<ReadonlyMap<string, number>>,
639 refreshImmediately: boolean,
640 ) {
641 const consecutiveFailuresRef = yield* Ref.make(0);
642 const needsInitialRefreshRef = yield* Ref.make(refreshImmediately);
643 const refreshRemoteStatusIfEnabled = Effect.gen(function* () {
644 const configuredInterval = yield* automaticFetchInterval;
645 const activeInterval = Duration.isZero(configuredInterval)
646 ? DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL
647 : configuredInterval;
648 const needsInitialRefresh = yield* Ref.get(needsInitialRefreshRef);
649 if (Duration.isZero(configuredInterval) && !needsInitialRefresh) {
650 return activeInterval;
651 }
652
653 const demandCwds = yield* Ref.get(demandCwdsRef);
654 const shouldRun =
655 needsInitialRefresh ||
656 (yield* Effect.forEach(
657 [...demandCwds.keys()],
658 (demandCwd) =>
659 backgroundPolicy.shouldRunScopeWork({
660 type: "vcs-status",
661 cwd: demandCwd,
662 }),
663 { concurrency: "unbounded" },
664 )).some(Boolean);
665 if (!shouldRun) {
666 return activeInterval;
667 }
668
669 const exit = yield* refreshRemoteStatus(cwd, {
670 refreshUpstream: !Duration.isZero(configuredInterval),
671 policyCwds: [...demandCwds.keys()],
672 }).pipe(Effect.exit);
673 if (Exit.isSuccess(exit)) {
674 yield* Ref.set(needsInitialRefreshRef, false);
675 yield* Ref.set(consecutiveFailuresRef, 0);
676 return activeInterval;
677 }
678
679 const interruptionReasons = exit.cause.reasons.filter(Cause.isInterruptReason);
680 if (interruptionReasons.length > 0) {
681 return yield* Effect.failCause(Cause.fromReasons<never>(interruptionReasons));
682 }
683
684 const consecutiveFailures = yield* Ref.updateAndGet(
685 consecutiveFailuresRef,
686 (count) => count + 1,
687 );
688 const nextDelay = remoteRefreshFailureDelay(consecutiveFailures, activeInterval);
689 yield* Effect.logWarning("VCS remote status refresh failed", {
690 cwdLength: cwd.length,
691 ...remoteRefreshFailureDiagnostics(exit.cause),
692 consecutiveFailures,
693 nextDelayMs: Duration.toMillis(nextDelay),
694 });
695 return nextDelay;
696 });
697
698 if (!refreshImmediately) {
699 const configuredInterval = yield* automaticFetchInterval;
700 yield* Effect.sleep(
701 Duration.isZero(configuredInterval)
702 ? DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL
703 : configuredInterval,
704 );
705 }
706
707 return yield* refreshRemoteStatusIfEnabled.pipe(
708 Effect.repeat(
709 Schedule.identity<Duration.Duration>().pipe(
710 Schedule.addDelay(({ output: delay }) => Effect.succeed(delay)),
711 ),
712 ),
713 Effect.asVoid,
714 );
715 });
716
717 const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* (
718 cwd: string,
719 demandCwd: string,
720 refreshImmediately: boolean,
721 ) {
722 yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => {
723 const existing = activePollers.get(cwd);
724 if (existing) {
725 return Ref.update(existing.demandCwds, (demandCwds) => {
726 const next = new Map(demandCwds);
727 next.set(demandCwd, (next.get(demandCwd) ?? 0) + 1);
728 return next;
729 }).pipe(
730 Effect.map(() => {
731 const nextPollers = new Map(activePollers);
732 nextPollers.set(cwd, {
733 ...existing,
734 subscriberCount: existing.subscriberCount + 1,
735 });
736 return [undefined, nextPollers] as const;
737 }),
738 );
739 }
740
741 return Ref.make<ReadonlyMap<string, number>>(new Map([[demandCwd, 1]])).pipe(
742 Effect.flatMap((demandCwds) =>
743 makeRemoteRefreshLoop(cwd, demandCwds, refreshImmediately).pipe(
744 Effect.forkIn(broadcasterScope),
745 Effect.map((fiber) => {
746 const nextPollers = new Map(activePollers);
747 nextPollers.set(cwd, {
748 fiber,
749 subscriberCount: 1,
750 demandCwds,
751 });
752 return [undefined, nextPollers] as const;
753 }),
754 ),
755 ),
756 );
757 });
758 });
759
760 const releaseRemotePoller = Effect.fn("VcsStatusBroadcaster.releaseRemotePoller")(function* (
761 cwd: string,
762 demandCwd: string,
763 ) {
764 const pollerToInterrupt = yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => {
765 const existing = activePollers.get(cwd);
766 if (!existing) {
767 return Effect.succeed([null, activePollers] as const);
768 }
769
770 if (existing.subscriberCount > 1) {
771 return Ref.update(existing.demandCwds, (demandCwds) => {
772 const nextDemandCwds = new Map(demandCwds);
773 const count = nextDemandCwds.get(demandCwd) ?? 0;
774 if (count <= 1) {
775 nextDemandCwds.delete(demandCwd);
776 } else {
777 nextDemandCwds.set(demandCwd, count - 1);
778 }
779 return nextDemandCwds;
780 }).pipe(
781 Effect.as([
782 null,
783 new Map(activePollers).set(cwd, {
784 ...existing,
785 subscriberCount: existing.subscriberCount - 1,
786 }),
787 ] as const),
788 );
789 }
790
791 return Effect.succeed([
792 existing.fiber,
793 new Map([...activePollers].filter(([activeCwd]) => activeCwd !== cwd)),
794 ] as const);
795 });
796
797 if (pollerToInterrupt) {
798 yield* Fiber.interrupt(pollerToInterrupt).pipe(Effect.ignore);
799 }
800 });
801
802 // Serialize branch publication and cache invalidation with detailed reads and refreshes.
803 const refreshBranchStatusCore = Effect.fn("VcsStatusBroadcaster.refreshBranchStatusCore")(
804 (cwd: string) =>
805 localRefreshLocks.withLock(
806 cwd,
807 Effect.gen(function* () {
808 if (detailedSubscribers.has(cwd)) {
809 const local = yield* getOrLoadLocalStatusUnlocked(cwd);
810 return {
811 refName: local.refName,
812 isRepo: local.isRepo,
813 revision: branches.get(cwd)?.revision ?? 0,
814 };
815 }
816 const refName = yield* workflow.currentBranch(cwd);
817 // A null branch is a detached HEAD or a checkout that is gone, such as a removed worktree.
818 const isRepo = refName !== null || (yield* workflow.repositoryRoot(cwd)) !== null;
819 // HEAD cannot certify working-tree totals, even when it has not moved.
820 localRefreshResults.delete(cwd);
821 yield* workflow.invalidateLocalStatus(cwd);
822 yield* Ref.update(cacheRef, (cache) => {
823 const previous = cache.get(cwd);
824 return previous ? new Map(cache).set(cwd, { ...previous, local: null }) : cache;
825 });
826 if (detailedSubscribers.has(cwd)) {
827 const local = yield* workflow.localStatus({ cwd });
828 yield* updateCachedLocalStatus(cwd, local, { publish: true });
829 return {
830 refName: local.refName,
831 isRepo: local.isRepo,
832 revision: branches.get(cwd)?.revision ?? 0,
833 };
834 }
835 yield* publishBranch(cwd, { refName, isRepo });
836 return { refName, isRepo, revision: branches.get(cwd)?.revision ?? 0 };
837 }),
838 ),
839 );
840
841 // A checkout outside T3 Code, such as `git switch` in the terminal, calls no refresh RPC.
842 // Rereading remote too, without a fetch, sends the new branch with its own PR and counts.
843 const refreshCheckedOutRef = Effect.fn("VcsStatusBroadcaster.refreshCheckedOutRef")(function* (
844 cwd: string,
845 ) {
846 if (!detailedSubscribers.has(cwd)) {
847 yield* refreshBranchStatusCore(cwd);
848 } else if ((yield* getCachedStatus(cwd))?.remote) {
849 yield* refreshRemoteStatus(cwd, { refreshUpstream: false });
850 } else {
851 yield* refreshLocalStatusCore(cwd);
852 }
853 });
854
855 // Runs for the subscription lifetime, so tracing would leave a span open indefinitely.
856 const watchHead = Effect.fnUntraced(function* (cwd: string, ready: Deferred.Deferred<void>) {
857 const headPath = yield* workflow.headPath(cwd);
858 if (headPath === null) return;
859 const headDirectory = path.dirname(headPath);
860 // Git replaces HEAD by renaming HEAD.lock, which a watch on the file itself would lose.
861 yield* fs
862 .watch(headDirectory, {
863 onReady: () => Deferred.doneUnsafe(ready, Exit.void),
864 })
865 .pipe(
866 // On Windows a watch outlives its deleted directory and reports it ~150k times a second.
867 Stream.takeUntilEffect((event) =>
868 event._tag === "Remove"
869 ? fs.exists(headDirectory).pipe(Effect.map((exists) => !exists))
870 : Effect.succeed(false),
871 ),
872 Stream.filter((event) => path.basename(event.path) === "HEAD"),
873 // A rebase moves HEAD many times; refresh once for whatever arrived during a refresh.
874 Stream.buffer({ capacity: 1, strategy: "sliding" }),
875 Stream.runForEach(() =>
876 refreshCheckedOutRef(cwd).pipe(
877 Effect.retry(CHECKOUT_REFRESH_RETRY),
878 Effect.catchCause((cause) =>
879 Effect.logWarning("VCS checked-out ref refresh failed", {
880 cwdLength: cwd.length,
881 ...remoteRefreshFailureDiagnostics(cause),
882 }),
883 ),
884 ),
885 ),
886 );
887 });
888
889 const retainHeadWatcher = Effect.fn("VcsStatusBroadcaster.retainHeadWatcher")(function* (
890 cwd: string,
891 ) {
892 const ready = yield* SynchronizedRef.modifyEffect(headWatchersRef, (activeWatchers) => {
893 const existing = activeWatchers.get(cwd);
894 if (existing && existing.fiber.pollUnsafe() === undefined) {
895 const nextWatchers = new Map(activeWatchers).set(cwd, {
896 ...existing,
897 subscriberCount: existing.subscriberCount + 1,
898 });
899 return Effect.succeed([existing.ready, nextWatchers] as const);
900 }
901 // A watch that ended, say before the repository existed, restarts for everyone holding it.
902 return Effect.gen(function* () {
903 const ready = yield* Deferred.make<void>();
904 const fiber = yield* watchHead(cwd, ready).pipe(
905 Effect.catchCause((cause) =>
906 Effect.logWarning("VCS HEAD watch failed", {
907 cwdLength: cwd.length,
908 ...remoteRefreshFailureDiagnostics(cause),
909 }),
910 ),
911 // Subscribers wait for the watch, so one that cannot start must not hold them.
912 Effect.ensuring(Deferred.succeed(ready, undefined)),
913 Effect.forkIn(broadcasterScope),
914 );
915 const nextWatchers = new Map(activeWatchers).set(cwd, {
916 fiber,
917 ready,
918 subscriberCount: (existing?.subscriberCount ?? 0) + 1,
919 });
920 return [ready, nextWatchers] as const;
921 });
922 });
923 yield* Deferred.await(ready);
924 });
925
926 const releaseHeadWatcher = Effect.fn("VcsStatusBroadcaster.releaseHeadWatcher")(function* (
927 cwd: string,
928 ) {
929 const watcherToInterrupt = yield* SynchronizedRef.modify(headWatchersRef, (activeWatchers) => {
930 const existing = activeWatchers.get(cwd);
931 if (!existing) return [null, activeWatchers] as const;
932 const nextWatchers = new Map(activeWatchers);
933 if (existing.subscriberCount > 1) {
934 nextWatchers.set(cwd, { ...existing, subscriberCount: existing.subscriberCount - 1 });
935 return [null, nextWatchers] as const;
936 }
937 nextWatchers.delete(cwd);
938 return [existing.fiber, nextWatchers] as const;
939 });
940 if (watcherToInterrupt) {
941 yield* Fiber.interrupt(watcherToInterrupt).pipe(Effect.ignore);
942 }
943 });
944
945 const streamStatus: VcsStatusBroadcaster["Service"]["streamStatus"] = (input) =>
946 Stream.unwrap(
947 Effect.gen(function* () {
948 const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
949 yield* Effect.acquireRelease(
950 Effect.sync(() => {
951 const previous = detailedSubscribers.get(cwd) ?? 0;
952 detailedSubscribers.set(cwd, previous + 1);
953 }),
954 () =>
955 Effect.sync(() => {
956 const remaining = (detailedSubscribers.get(cwd) ?? 1) - 1;
957 if (remaining === 0) detailedSubscribers.delete(cwd);
958 else detailedSubscribers.set(cwd, remaining);
959 }),
960 );
961 const subscription = yield* PubSub.subscribe(changesPubSub);
962 // Watch before reading so a checkout in between still publishes.
963 yield* Effect.acquireRelease(retainHeadWatcher(cwd), () => releaseHeadWatcher(cwd));
964 const initialLocal = yield* getOrLoadLocalStatus(cwd);
965 const cachedStatus = yield* getCachedStatus(cwd);
966 const initialRemote = cachedStatus?.remote?.value ?? null;
967 if (input.includeRemote !== false) {
968 yield* retainRemotePoller(
969 cwd,
970 input.cwd,
971 cachedStatus?.remote === null || cachedStatus?.remote === undefined,
972 );
973 }
974 const release =
975 input.includeRemote === false
976 ? Effect.void
977 : releaseRemotePoller(cwd, input.cwd).pipe(Effect.ignore, Effect.asVoid);
978
979 return Stream.concat(
980 Stream.make({
981 _tag: "snapshot" as const,
982 local: initialLocal,
983 remote: initialRemote,
984 }),
985 Stream.fromSubscription(subscription).pipe(
986 Stream.filter((event) => event.cwd === cwd),
987 Stream.map((event) => event.event),
988 ),
989 ).pipe(Stream.ensuring(release));
990 }),
991 );
992
993 const refreshBranchStatus: VcsStatusBroadcaster["Service"]["refreshBranchStatus"] = Effect.fn(
994 "VcsStatusBroadcaster.refreshBranchStatus",
995 )(function* (input) {
996 const { refName, isRepo } = yield* refreshBranchStatusCore(
997 yield* withFileSystem(normalizeCwd(input.cwd)),
998 );
999 return { refName, isRepo };
1000 });
1001
1002 const streamBranchStatus: VcsStatusBroadcaster["Service"]["streamBranchStatus"] = (input) =>
1003 Stream.unwrap(
1004 Effect.gen(function* () {
1005 const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
1006 const subscription = yield* PubSub.subscribe(branchChangesPubSub);
1007 yield* Effect.acquireRelease(retainHeadWatcher(cwd), () => releaseHeadWatcher(cwd));
1008 const branch = yield* refreshBranchStatusCore(cwd);
1009 return Stream.concat(
1010 Stream.make({ refName: branch.refName, isRepo: branch.isRepo }),
1011 Stream.fromSubscription(subscription).pipe(
1012 Stream.filter((event) => event.cwd === cwd && event.revision > branch.revision),
1013 Stream.map(({ refName, isRepo }) => ({ refName, isRepo })),
1014 ),
1015 ).pipe(
1016 Stream.changesWith(
1017 (left, right) => left.refName === right.refName && left.isRepo === right.isRepo,
1018 ),
1019 );
1020 }),
1021 );
1022
1023 return VcsStatusBroadcaster.of({
1024 streamBranchStatus,
1025 refreshBranchStatus,
1026 streamBranchChanges: Stream.fromPubSub(branchChangesPubSub).pipe(
1027 Stream.map(({ cwd, refName }) => ({ cwd, refName })),
1028 ),
1029 getStatus,
1030 refreshLocalStatus,
1031 refreshStatus,
1032 refreshPullRequestStatus,
1033 streamStatus,
1034 });
1035});
1036
1037export const layer = Layer.effect(VcsStatusBroadcaster, make);