fix(server): refresh Changes totals after remote polls (#28600)

Author: Yash-Singh1

apps/server/src/vcs/VcsStatusBroadcaster.test.ts

@@ -656,9 +656,12 @@ describe("VcsStatusBroadcaster", () => {
     },
   );
 
-  it.effect("an explicit refresh reads local totals after the fetch", () => {
+  it.effect("a remote poll refreshes Changes totals when divergence counts stay the same", () => {
     const state: Parameters<typeof layerTestFor>[0] = {
-      currentLocalStatus: baseLocalStatus,
+      currentLocalStatus: {
+        ...baseLocalStatus,
+        branchChanges: { baseRef: "origin/main", insertions: 745, deletions: 24 },
+      },
       currentRemoteStatus: baseRemoteStatus,
       localStatusCalls: 0,
       remoteStatusCalls: 0,
@@ -668,24 +671,79 @@ describe("VcsStatusBroadcaster", () => {
 
     return Effect.gen(function* () {
       const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
-      const fetchStarted = yield* Deferred.make<void>();
-      const finishFetch = yield* Deferred.make<void>();
-      state.beforeRemoteStatus = Deferred.succeed(fetchStarted, undefined).pipe(
-        Effect.andThen(Deferred.await(finishFetch)),
+      yield* broadcaster.getStatus({ cwd: "/repo" });
+      const subscribed = yield* Deferred.make<void>();
+      const polled = yield* Deferred.make<void>();
+      yield* broadcaster.streamStatus({ cwd: "/repo" }).pipe(
+        Stream.runForEach((event) => {
+          if (event._tag === "snapshot") return Deferred.succeed(subscribed, undefined);
+          if (event._tag === "remoteUpdated") return Deferred.succeed(polled, undefined);
+          return Effect.void;
+        }),
+        Effect.forkScoped,
       );
-      const refresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);
-      yield* Deferred.await(fetchStarted);
-
-      const fetchedLocal: VcsStatusLocalResult = {
+      yield* Deferred.await(subscribed);
+      state.currentLocalStatus = {
         ...baseLocalStatus,
-        branchChanges: { baseRef: "origin/main", insertions: 0, deletions: 0 },
+        branchChanges: { baseRef: "origin/main", insertions: 107, deletions: 4 },
       };
-      state.currentLocalStatus = fetchedLocal;
-      yield* Deferred.succeed(finishFetch, undefined);
+      state.currentRemoteStatus = remoteStatusWithPr;
+      yield* TestClock.adjust("30 seconds");
+      yield* Deferred.await(polled);
+      const status = yield* broadcaster.getStatus({ cwd: "/repo" });
+      assert.deepStrictEqual(status.branchChanges, state.currentLocalStatus.branchChanges);
+    }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped);
+  });
 
-      const status = yield* Fiber.join(refresh);
-      assert.deepStrictEqual(status.branchChanges, fetchedLocal.branchChanges);
-    }).pipe(Effect.provide(layerTestFor(state)));
+  it.effect("an explicit refresh reads local totals after the fetch", () => {
+    return Effect.gen(function* () {
+      const fetchStarted = yield* Deferred.make<void>();
+      const finishFetch = yield* Deferred.make<void>();
+      let local: VcsStatusLocalResult = {
+        ...baseLocalStatus,
+        branchChanges: { baseRef: "origin/main", insertions: 745, deletions: 24 },
+      };
+      let cachedLocal: VcsStatusLocalResult | undefined;
+      const readLocalStatus = () => Effect.sync(() => (cachedLocal ??= local));
+      const layerTest = VcsStatusBroadcaster.layer.pipe(
+        Layer.provideMerge(NodeServices.layer),
+        Layer.provide(layerBackgroundPolicy(() => true)),
+        Layer.provide(
+          Layer.mock(GitWorkflowService.GitWorkflowService)({
+            localStatus: readLocalStatus,
+            invalidateStatus: () =>
+              Effect.sync(() => {
+                cachedLocal = undefined;
+              }),
+            invalidateLocalStatus: () =>
+              Effect.sync(() => {
+                cachedLocal = undefined;
+              }),
+            currentBranch: () => Effect.succeed(baseLocalStatus.refName),
+            invalidateRemoteStatus: () => Effect.void,
+            remoteStatus: () =>
+              Deferred.succeed(fetchStarted, undefined).pipe(
+                Effect.andThen(Deferred.await(finishFetch)),
+                Effect.as(baseRemoteStatus),
+              ),
+          }),
+        ),
+      );
+      yield* Effect.gen(function* () {
+        const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
+        const refresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped);
+        yield* Deferred.await(fetchStarted);
+        // Another reader can refill the local cache while fetch is still running.
+        yield* readLocalStatus();
+        local = {
+          ...baseLocalStatus,
+          branchChanges: { baseRef: "origin/main", insertions: 0, deletions: 0 },
+        };
+        yield* Deferred.succeed(finishFetch, undefined);
+        const status = yield* Fiber.join(refresh);
+        assert.deepStrictEqual(status.branchChanges, local.branchChanges);
+      }).pipe(Effect.provide(layerTest));
+    });
   });
 
   it.effect("publishes local status while an earlier lookup holds the remote lock", () => {

apps/server/src/vcs/VcsStatusBroadcaster.ts

@@ -477,23 +477,14 @@ export const make = Effect.gen(function* () {
     return yield* withRemoteWriteLock(
       cwd,
       Effect.gen(function* () {
-        const previousRemote = (yield* getCachedStatus(cwd))?.remote?.value;
         const remoteRead = yield* readRemoteStatus(cwd, options);
-        const remote = remoteRead?.value;
         const pulled =
           options?.refreshUpstream === false
             ? null
             : yield* maybeAutoPull(cwd, remoteRead, options?.policyCwds ?? [cwd]);
         if (pulled !== null) return pulled.remote;
-        if (
-          remote &&
-          (!previousRemote ||
-            previousRemote.aheadCount !== remote.aheadCount ||
-            previousRemote.behindCount !== remote.behindCount ||
-            previousRemote.aheadOfDefaultCount !== remote.aheadOfDefaultCount)
-        ) {
-          yield* refreshLocalStatusCore(cwd);
-        }
+        // Base refs and merge bases can move without changing divergence counts.
+        yield* refreshLocalStatusCore(cwd);
         return yield* updateCachedRemoteStatus(cwd, remoteRead, { publish: true });
       }),
     );
@@ -530,8 +521,7 @@ export const make = Effect.gen(function* () {
         yield* workflow.invalidateStatus(cwd);
         const remoteRead = yield* readRemoteStatus(cwd, { refreshUpstream });
         // Fetch can move the base used for local Changes totals.
-        const local = yield* workflow.localStatus({ cwd });
-        yield* updateCachedLocalStatus(cwd, local, { publish: true });
+        yield* refreshLocalStatusCore(cwd);
         // Without a fetch, the cached behind count cannot justify an automatic pull.
         const pulled = refreshUpstream ? yield* maybeAutoPull(cwd, remoteRead, [rawCwd]) : null;
         if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote);