Refresh working tree diff after turns

packages/client-runtime/src/state/review.test.ts

@@ -0,0 +1,40 @@
+import { EnvironmentId } from "@t3tools/contracts";
+import { describe, expect, it } from "@effect/vitest";
+import { Atom, AtomRegistry } from "effect/unstable/reactivity";
+
+import { invalidateReviewDiffPreviews, reviewDiffPreviewRevisionAtom } from "./review.ts";
+
+const TARGET = {
+  environmentId: EnvironmentId.make("environment-1"),
+};
+
+describe("review diff preview invalidation", () => {
+  it("scopes revision changes to the invalidated environment", () => {
+    const registry = AtomRegistry.make();
+    const otherEnvironment = {
+      environmentId: EnvironmentId.make("environment-2"),
+    };
+
+    expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(0);
+    expect(registry.get(reviewDiffPreviewRevisionAtom(otherEnvironment))).toBe(0);
+
+    invalidateReviewDiffPreviews(registry, TARGET);
+
+    expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(1);
+    expect(registry.get(reviewDiffPreviewRevisionAtom(otherEnvironment))).toBe(0);
+    registry.dispose();
+  });
+
+  it("notifies reactive dependents without replacing their atom identity", () => {
+    const registry = AtomRegistry.make();
+    const dependent = Atom.make((get) => get(reviewDiffPreviewRevisionAtom(TARGET)));
+    const unmount = registry.mount(dependent);
+
+    expect(registry.get(dependent)).toBe(0);
+    invalidateReviewDiffPreviews(registry, TARGET);
+    expect(registry.get(dependent)).toBe(1);
+
+    unmount();
+    registry.dispose();
+  });
+});

packages/client-runtime/src/state/review.ts

@@ -1,9 +1,31 @@
-import { WS_METHODS } from "@t3tools/contracts";
-import { Atom } from "effect/unstable/reactivity";
+import { type EnvironmentId, WS_METHODS } from "@t3tools/contracts";
+import { Atom, type AtomRegistry } from "effect/unstable/reactivity";
 
 import { createEnvironmentRpcQueryAtomFamily } from "./runtime.ts";
 import type { EnvironmentRegistry } from "../connection/registry.ts";
 
+export interface ReviewDiffPreviewInvalidationTarget {
+  readonly environmentId: EnvironmentId;
+}
+
+const diffPreviewRevisionByEnvironment = Atom.family((environmentId: EnvironmentId) =>
+  Atom.make(0).pipe(
+    Atom.keepAlive,
+    Atom.withLabel(`environment-data:review:diff-preview-revision:${environmentId}`),
+  ),
+);
+
+export function reviewDiffPreviewRevisionAtom(target: ReviewDiffPreviewInvalidationTarget) {
+  return diffPreviewRevisionByEnvironment(target.environmentId);
+}
+
+export function invalidateReviewDiffPreviews(
+  registry: AtomRegistry.AtomRegistry,
+  target: ReviewDiffPreviewInvalidationTarget,
+): void {
+  registry.update(reviewDiffPreviewRevisionAtom(target), (revision) => revision + 1);
+}
+
 export function createReviewEnvironmentAtoms<R, E>(
   runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
 ) {
@@ -12,6 +34,7 @@ export function createReviewEnvironmentAtoms<R, E>(
       label: "environment-data:review:diff-preview",
       tag: WS_METHODS.reviewGetDiffPreview,
       staleTimeMs: 5_000,
+      refreshSignal: reviewDiffPreviewRevisionAtom,
     }),
   };
 }

packages/client-runtime/src/state/runtime.ts

@@ -52,6 +52,10 @@ interface EnvironmentQueryAtomOptions<Input, A, E, R> extends EnvironmentAtomOpt
   readonly staleTimeMs?: number;
   readonly idleTtlMs?: number;
   readonly refreshIntervalMs?: number;
+  readonly refreshSignal?: (target: {
+    readonly environmentId: EnvironmentIdType;
+    readonly input: Input;
+  }) => Atom.Atom<unknown>;
 }
 
 interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> {
@@ -488,6 +492,9 @@ function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>(
     const idleTtlMs = options.idleTtlMs ?? 5 * 60_000;
     const queryAtom = runtime
       .atom((get) => {
+        if (options.refreshSignal !== undefined) {
+          get(options.refreshSignal(target));
+        }
         const generation = Option.getOrNull(
           AsyncResult.value(get(rpcGenerationAtom(target.environmentId))),
         );
@@ -576,6 +583,10 @@ export function createEnvironmentRpcQueryAtomFamily<R, ER, TTag extends Environm
     readonly staleTimeMs?: number;
     readonly idleTtlMs?: number;
     readonly refreshIntervalMs?: number;
+    readonly refreshSignal?: (target: {
+      readonly environmentId: EnvironmentIdType;
+      readonly input: EnvironmentRpcInput<TTag>;
+    }) => Atom.Atom<unknown>;
   },
 ) {
   return createEnvironmentQueryAtomFamily(runtime, {
@@ -585,6 +596,7 @@ export function createEnvironmentRpcQueryAtomFamily<R, ER, TTag extends Environm
     ...(options.refreshIntervalMs === undefined
       ? {}
       : { refreshIntervalMs: options.refreshIntervalMs }),
+    ...(options.refreshSignal === undefined ? {} : { refreshSignal: options.refreshSignal }),
     execute: (input: EnvironmentRpcInput<TTag>) => request(options.tag, input),
   });
 }

packages/client-runtime/src/state/vcs.test.ts

@@ -3,6 +3,8 @@ import {
   WS_METHODS,
   type VcsListRefsInput,
   type VcsListRefsResult,
+  type VcsStatusLocalResult,
+  type VcsStatusRemoteResult,
 } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
@@ -32,7 +34,9 @@ import {
   commitVcsRefsRefresh,
   createVcsEnvironmentAtoms,
   makeCachedVcsRefsChanges,
+  projectVcsStatusChanges,
 } from "./vcs.ts";
+import { reviewDiffPreviewRevisionAtom } from "./review.ts";
 import {
   invalidateCachedVcsRefs,
   invalidateVcsRefs,
@@ -82,6 +86,36 @@ const LIVE_REFS: VcsListRefsResult = {
   ],
 };
 
+const LOCAL_STATUS_WITH_FILE: VcsStatusLocalResult = {
+  isRepo: true,
+  hasPrimaryRemote: true,
+  isDefaultRef: false,
+  refName: "feature",
+  hasWorkingTreeChanges: true,
+  workingTree: {
+    files: [{ path: "temporary.txt", insertions: 1, deletions: 0 }],
+    insertions: 1,
+    deletions: 0,
+  },
+};
+
+const LOCAL_STATUS_AFTER_DELETION: VcsStatusLocalResult = {
+  ...LOCAL_STATUS_WITH_FILE,
+  hasWorkingTreeChanges: false,
+  workingTree: {
+    files: [],
+    insertions: 0,
+    deletions: 0,
+  },
+};
+
+const REMOTE_STATUS: VcsStatusRemoteResult = {
+  hasUpstream: true,
+  aheadCount: 0,
+  behindCount: 0,
+  pr: null,
+};
+
 function session(client: WsRpcProtocolClient): RpcSession {
   return {
     client,
@@ -114,6 +148,45 @@ function cacheWithRefs(
 }
 
 describe("cached VCS refs", () => {
+  it.effect(
+    "invalidates diff previews after a changed status event, not the initial snapshot",
+    () =>
+      Effect.gen(function* () {
+        const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
+          Effect.sync(() => registry.dispose()),
+        );
+        const revisionAtom = reviewDiffPreviewRevisionAtom(TARGET);
+        const revisions: number[] = [];
+
+        const statuses = yield* projectVcsStatusChanges(
+          TARGET.environmentId,
+          Stream.make(
+            {
+              _tag: "snapshot" as const,
+              local: LOCAL_STATUS_WITH_FILE,
+              remote: REMOTE_STATUS,
+            },
+            {
+              _tag: "localUpdated" as const,
+              local: LOCAL_STATUS_AFTER_DELETION,
+            },
+          ),
+        ).pipe(
+          Stream.tap(() =>
+            Effect.sync(() => {
+              revisions.push(registry.get(revisionAtom));
+            }),
+          ),
+          Stream.provideService(AtomRegistry.AtomRegistry, registry),
+          Stream.runCollect,
+        );
+
+        expect(statuses).toHaveLength(2);
+        expect(statuses[1]?.workingTree.files).toEqual([]);
+        expect(revisions).toEqual([0, 1]);
+      }),
+  );
+
   it("invalidates all ref streams in the mutated environment", () => {
     const registry = AtomRegistry.make();
     const environment = {
@@ -260,6 +333,7 @@ describe("cached VCS refs", () => {
           revision: 1,
           persistedCacheReadable: true,
         });
+        expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(1);
 
         expect(
           yield* commitVcsRefsRefresh(registry, cache, {
@@ -329,6 +403,7 @@ describe("cached VCS refs", () => {
         expect(AsyncResult.isFailure(result)).toBe(true);
         expect(yield* Ref.get(clears)).toBe(1);
         expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1);
+        expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(1);
 
         const refreshResult = yield* Effect.promise(() =>
           atoms.refreshStatus.run(registry, {
@@ -340,6 +415,7 @@ describe("cached VCS refs", () => {
         expect(AsyncResult.isSuccess(refreshResult)).toBe(true);
         expect(yield* Ref.get(clears)).toBe(2);
         expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2);
+        expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(2);
       }),
     ),
   );

packages/client-runtime/src/state/vcs.ts

@@ -3,6 +3,7 @@ import {
   type VcsListRefsInput,
   type VcsListRefsResult,
   type VcsStatusResult,
+  type VcsStatusStreamEvent,
   WS_METHODS,
 } from "@t3tools/contracts";
 import { applyGitStatusStreamEvent } from "@t3tools/shared/git";
@@ -15,14 +16,15 @@ import * as Stream from "effect/Stream";
 import * as SubscriptionRef from "effect/SubscriptionRef";
 import { Atom, AtomRegistry } from "effect/unstable/reactivity";
 
-import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts";
+import { createEnvironmentRpcCommand } from "./runtime.ts";
 import type { EnvironmentRegistry } from "../connection/registry.ts";
 import { EnvironmentSupervisor } from "../connection/supervisor.ts";
 import { safeErrorLogAttributes } from "../errors/safeLog.ts";
 import { EnvironmentCacheStore } from "../platform/persistence.ts";
 import { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
 import { followStreamInEnvironment } from "./runtime.ts";
 import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
+import { invalidateReviewDiffPreviews } from "./review.ts";
 import {
   invalidateCachedVcsRefs,
   vcsRefsCacheStateAtom,
@@ -233,6 +235,34 @@ export function cachedVcsRefsChanges(
   );
 }
 
+export function projectVcsStatusChanges<E, R>(
+  environmentId: EnvironmentId,
+  events: Stream.Stream<VcsStatusStreamEvent, E, R>,
+) {
+  return events.pipe(
+    Stream.mapAccumEffect(
+      () => null as VcsStatusResult | null,
+      (current, event) => {
+        const next = applyGitStatusStreamEvent(current, event);
+        return Effect.gen(function* () {
+          if (current !== null) {
+            const registry = yield* AtomRegistry.AtomRegistry;
+            invalidateReviewDiffPreviews(registry, { environmentId });
+          }
+          return [next, [next]] as const;
+        });
+      },
+    ),
+  );
+}
+
+export function vcsStatusChanges(
+  environmentId: EnvironmentId,
+  input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>,
+) {
+  return projectVcsStatusChanges(environmentId, subscribe(WS_METHODS.subscribeVcsStatus, input));
+}
+
 export function createVcsEnvironmentAtoms<R, E>(
   runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
 ) {
@@ -259,6 +289,23 @@ export function createVcsEnvironmentAtoms<R, E>(
     readonly environmentId: EnvironmentId;
     readonly input: VcsListRefsInput;
   }) => listRefsByEnvironment(target.environmentId)(JSON.stringify(target.input));
+  const statusByEnvironment = Atom.family((environmentId: EnvironmentId) =>
+    Atom.family((inputKey: string) => {
+      const input = JSON.parse(inputKey) as EnvironmentRpcInput<
+        typeof WS_METHODS.subscribeVcsStatus
+      >;
+      return runtime
+        .atom(followStreamInEnvironment(environmentId, vcsStatusChanges(environmentId, input)))
+        .pipe(
+          Atom.setIdleTTL(5 * 60_000),
+          Atom.withLabel(`environment-data:vcs:status:${environmentId}:${inputKey}`),
+        );
+    }),
+  );
+  const status = (target: {
+    readonly environmentId: EnvironmentId;
+    readonly input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>;
+  }) => statusByEnvironment(target.environmentId)(JSON.stringify(target.input));
   const invalidateRefs = (
     target: { readonly environmentId: EnvironmentId; readonly input: { readonly cwd: string } },
     registry: AtomRegistry.AtomRegistry,
@@ -270,19 +317,7 @@ export function createVcsEnvironmentAtoms<R, E>(
 
   return {
     listRefs,
-    status: createEnvironmentSubscriptionAtomFamily(runtime, {
-      label: "environment-data:vcs:status",
-      subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>) =>
-        subscribe(WS_METHODS.subscribeVcsStatus, input).pipe(
-          Stream.mapAccum(
-            () => null as VcsStatusResult | null,
-            (current, event) => {
-              const next = applyGitStatusStreamEvent(current, event);
-              return [next, [next]] as const;
-            },
-          ),
-        ),
-    }),
+    status,
     pull: createEnvironmentRpcCommand(runtime, {
       label: "environment-data:vcs:pull",
       tag: WS_METHODS.vcsPull,

packages/client-runtime/src/state/vcsRefInvalidation.ts

@@ -5,6 +5,7 @@ import { Atom, type AtomRegistry } from "effect/unstable/reactivity";
 
 import { safeErrorLogAttributes } from "../errors/safeLog.ts";
 import { EnvironmentCacheStore } from "../platform/persistence.ts";
+import { invalidateReviewDiffPreviews } from "./review.ts";
 
 export interface VcsRefsInvalidationTarget {
   readonly environmentId: EnvironmentId;
@@ -78,6 +79,7 @@ export const invalidateCachedVcsRefs = Effect.fn("VcsRefsState.invalidateCached"
         ),
       );
       invalidateVcsRefs(registry, target, persistedCacheReadable);
+      invalidateReviewDiffPreviews(registry, target);
     }),
   );
 });