fix(server): publish workspace snapshots without UI subscribers (#7681)

Computed snapshots must reach the runtime store and auto-archive even when no UI observes the workspace. Reject retired targets so late refreshes cannot replace newer published state.

packages/server/src/server/auto-archive-on-merge/behavior.test.ts

@@ -2,7 +2,7 @@ import { execFileSync } from "node:child_process";
 import { mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
-import type { Logger } from "pino";
+import pino, { type Logger } from "pino";
 import { afterEach, describe, expect, test, vi } from "vitest";
 
 import {
@@ -10,8 +10,14 @@ import {
   type AutoArchiveOnMergeDependencies,
   type AutoArchiveOnMergeOptions,
 } from "./index.js";
-import type { WorkspaceGitRuntimeSnapshot } from "../workspace-git-service.js";
-import { createWorktree } from "../../utils/worktree.js";
+import {
+  WorkspaceGitServiceImpl,
+  type WorkspaceGitRuntimeSnapshot,
+} from "../workspace-git-service.js";
+import { createGitHubService } from "../../services/github-service.js";
+import { archiveIfSafe } from "./archive-if-safe.js";
+import { killTerminalsForWorkspace } from "../workspace-archive-service.js";
+import { isPaseoOwnedWorktreeCwd, createWorktree } from "../../utils/worktree.js";
 
 type PullRequestState = "open" | "closed" | "merged";
 
@@ -22,6 +28,7 @@ interface PullRequestObservation {
 }
 
 const cleanupPaths: string[] = [];
+const cleanupServices: WorkspaceGitServiceImpl[] = [];
 
 function run(cwd: string, args: string[]): void {
   execFileSync("git", args, { cwd, stdio: "pipe" });
@@ -32,7 +39,7 @@ async function settleObservation(): Promise<void> {
   await new Promise<void>((resolve) => setImmediate(resolve));
 }
 
-async function createWorkspaceJourney() {
+async function createWorkspaceFiles() {
   const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "auto-archive-behavior-")));
   cleanupPaths.push(tempDir);
   const repoDir = path.join(tempDir, "repo");
@@ -51,6 +58,11 @@ async function createWorkspaceJourney() {
 
     paseoHome,
   });
+  return { repoDir, paseoHome, worktree };
+}
+
+async function createWorkspaceJourney() {
+  const { repoDir, worktree } = await createWorkspaceFiles();
   const workspaceId = "workspace-under-test";
   let active = true;
   let currentPullRequest: PullRequestObservation | null = null;
@@ -149,6 +161,95 @@ async function createWorkspaceJourney() {
   };
 }
 
+async function createUnobservedWorkspace(archiveBarrier?: Promise<void>) {
+  const { repoDir, paseoHome, worktree } = await createWorkspaceFiles();
+  const cwd = worktree.worktreePath;
+  run(repoDir, ["remote", "add", "origin", "https://github.com/acme/repo.git"]);
+  run(repoDir, ["update-ref", "refs/remotes/origin/workspace", "HEAD"]);
+  run(cwd, ["branch", "--set-upstream-to=origin/workspace"]);
+  const logger = pino({ level: "silent" });
+  const github = createGitHubService({
+    resolveGhPath: async () => "/test/gh",
+    runner: async () => ({ stdout: "", stderr: "" }),
+  });
+  let observation: PullRequestObservation | null = null;
+  const reads: Array<string | undefined> = [];
+  const archives: string[] = [];
+  let active = true;
+  const service = new WorkspaceGitServiceImpl({
+    logger,
+    paseoHome,
+    deps: {
+      forgeOverrides: { github },
+      getPullRequestStatus: async (_cwd, _forge, options) => {
+        reads.push(options?.reason);
+        return {
+          status: observation && {
+            ...observation,
+            title: observation.headRefName,
+            baseRefName: "main",
+            isMerged: observation.state === "merged",
+          },
+          authState: "authenticated",
+          featuresEnabled: true,
+          githubFeaturesEnabled: true,
+        };
+      },
+    },
+  });
+  cleanupServices.push(service);
+  const options = {
+    logger,
+    paseoHome,
+    daemonConfigStore: { get: () => ({ autoArchiveAfterMerge: true }) },
+    workspaceGitService: service,
+    getAutoArchivedChangeRequestUrl: async () => null,
+    listActiveWorkspaces: async () =>
+      active ? [{ workspaceId: "workspace-under-test", cwd, kind: "worktree" }] : [],
+  } as unknown as AutoArchiveOnMergeOptions;
+  setupAutoArchiveOnMerge(options, {
+    resolvePath: path.resolve,
+    archiveIfSafe: async (input) =>
+      archiveIfSafe({
+        ...input,
+        deps: {
+          isPaseoOwnedWorktreeCwd,
+          killTerminalsForWorkspace,
+          archiveByScope: async (_options, request) => {
+            if (request.scope.kind !== "workspace") throw new Error("Expected workspace archive");
+            archives.push(request.scope.workspaceId);
+            await archiveBarrier;
+            active = false;
+            return {
+              archivedAgentIds: [],
+              archivedWorkspaceIds: [request.scope.workspaceId],
+              removedDirectory: false,
+            };
+          },
+        },
+      }),
+  });
+  return {
+    archives,
+    reads,
+    metrics: () => service.getMetrics(),
+    isActive: () => active,
+    makeDirty: () => writeFileSync(path.join(cwd, "README.md"), "uncommitted work\n"),
+    commitUnpushed: () => {
+      writeFileSync(path.join(cwd, "README.md"), "unpushed work\n");
+      run(cwd, ["add", "README.md"]);
+      run(cwd, ["-c", "commit.gpgsign=false", "commit", "-m", "unpushed"]);
+    },
+    observe: async (state: PullRequestState) => {
+      observation = pullRequest("workspace", state);
+      const snapshot = await service.getSnapshot(cwd, { force: true, reason: "workspace-journey" });
+      expect(snapshot.forge.pullRequest?.state).toBe(state);
+      await settleObservation();
+      return snapshot;
+    },
+  };
+}
+
 function pullRequest(
   headRefName: string,
   state: PullRequestState,
@@ -161,13 +262,77 @@ function pullRequest(
   };
 }
 
-afterEach(() => {
+afterEach(async () => {
+  for (const service of cleanupServices.splice(0)) await service.dispose();
   for (const target of cleanupPaths.splice(0)) {
     rmSync(target, { recursive: true, force: true });
   }
 });
 
 describe("workspace auto-archive behavior", () => {
+  test("archives a clean owned workspace on an unobserved open-to-merged transition only once", async () => {
+    let releaseArchive!: () => void;
+    const archiveBarrier = new Promise<void>((resolve) => {
+      releaseArchive = resolve;
+    });
+    const workspace = await createUnobservedWorkspace(archiveBarrier);
+    try {
+      await workspace.observe("open");
+      expect(workspace.archives).toEqual([]);
+
+      const merged = await workspace.observe("merged");
+      expect(merged.git).toMatchObject({
+        isDirty: false,
+        aheadOfOrigin: 0,
+        isPaseoOwnedWorktree: true,
+      });
+      await expect.poll(() => workspace.archives).toEqual(["workspace-under-test"]);
+      expect(workspace.isActive()).toBe(true);
+      await workspace.observe("merged");
+
+      expect(workspace.metrics().workspaceListenerCount).toBe(0);
+      expect(workspace.archives).toEqual(["workspace-under-test"]);
+      // Confirmation reads the completed snapshot cache; it does not shell out again.
+      expect(workspace.reads).toEqual([
+        "workspace-journey",
+        "workspace-journey",
+        "workspace-journey",
+      ]);
+    } finally {
+      releaseArchive();
+    }
+    await expect.poll(() => workspace.isActive()).toBe(false);
+  });
+
+  test("keeps an unobserved workspace active when its first snapshot is already merged", async () => {
+    const workspace = await createUnobservedWorkspace();
+    await workspace.observe("merged");
+
+    expect(workspace.metrics().workspaceListenerCount).toBe(0);
+    expect(workspace.archives).toEqual([]);
+    expect(workspace.reads).toEqual(["workspace-journey"]);
+  });
+
+  test("keeps dirty work when an unobserved workspace PR merges", async () => {
+    const workspace = await createUnobservedWorkspace();
+    await workspace.observe("open");
+    workspace.makeDirty();
+    const merged = await workspace.observe("merged");
+
+    expect(merged.git.isDirty).toBe(true);
+    expect(workspace.archives).toEqual([]);
+  });
+
+  test("keeps unpushed commits when an unobserved workspace PR merges", async () => {
+    const workspace = await createUnobservedWorkspace();
+    await workspace.observe("open");
+    workspace.commitUnpushed();
+    const merged = await workspace.observe("merged");
+
+    expect(merged.git).toMatchObject({ isDirty: false, aheadOfOrigin: 1 });
+    expect(workspace.archives).toEqual([]);
+  });
+
   test("archives when the checked-out PR changes from open to merged", async () => {
     const workspace = await createWorkspaceJourney();
     workspace.checkout("feature");

packages/server/src/server/workspace-git-service.test.ts

@@ -384,28 +384,84 @@ describe("WorkspaceGitServiceImpl", () => {
     await service.dispose();
   });
 
-  test("onSnapshotUpdated emits only for observed workspace snapshots and can unsubscribe", async () => {
+  test("onSnapshotUpdated delivers without workspace subscribers and can unsubscribe", async () => {
     const service = createService();
-    const snapshotListener = vi.fn();
-    const snapshotSubscription = service.onSnapshotUpdated(snapshotListener);
+    const snapshots: WorkspaceGitRuntimeSnapshot[] = [];
+    const subscription = service.onSnapshotUpdated((snapshot) => snapshots.push(snapshot));
+    try {
+      const initial = await service.getSnapshot(REPO_CWD);
+      expect(snapshots).toEqual([initial]);
+      expect(service.getMetrics().workspaceListenerCount).toBe(0);
 
-    await service.getSnapshot(REPO_CWD, { force: true, reason: "unobserved" });
+      await service.getSnapshot(REPO_CWD);
+      expect(snapshots).toEqual([initial]);
 
-    expect(snapshotListener).not.toHaveBeenCalled();
+      const refreshed = await service.getSnapshot(REPO_CWD, {
+        force: true,
+        reason: "unobserved-refresh",
+      });
+      expect(snapshots).toEqual([initial, refreshed]);
 
-    const workspaceSubscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
-    await service.getSnapshot(REPO_CWD, { force: true, reason: "observed" });
+      subscription.unsubscribe();
+      await service.getSnapshot(REPO_CWD, { force: true, reason: "after-unsubscribe" });
+      expect(snapshots).toEqual([initial, refreshed]);
+    } finally {
+      await service.dispose();
+    }
+  });
 
-    expect(snapshotListener).toHaveBeenCalledTimes(1);
-    expect(snapshotListener).toHaveBeenCalledWith(createSnapshot(REPO_CWD));
+  test("a retired target's late refresh cannot overwrite the runtime store's newer snapshot", async () => {
+    const retiredRead = createDeferred<PullRequestStatusResult>();
+    const retiredReadStarted = createDeferred<void>();
+    const merged = createPullRequestStatusResult();
+    if (!merged.status) throw new Error("Expected pull request status");
+    merged.status = { ...merged.status, state: "merged", isMerged: true };
+    let reads = 0;
+    const service = new WorkspaceGitServiceImpl({
+      logger: createLogger() as unknown as pino.Logger,
+      paseoHome: "/tmp/paseo-test",
+      deps: {
+        ...buildDefaultTestServiceDeps(),
+        getPullRequestStatus: async () => {
+          reads += 1;
+          if (reads === 1) {
+            retiredReadStarted.resolve();
+            return retiredRead.promise;
+          }
+          return merged;
+        },
+      },
+    });
+    const store = createWorkspaceRuntimeStore();
+    const published: WorkspaceGitRuntimeSnapshot[] = [];
+    service.onSnapshotUpdated((snapshot) => {
+      store.set(snapshot);
+      published.push(snapshot);
+    });
+    const subscription = service.registerWorkspace({ cwd: REPO_CWD }, () => {});
+    const oldRefresh = service.getSnapshot(REPO_CWD, { force: true, reason: "retiring-target" });
+    try {
+      await retiredReadStarted.promise;
+      subscription.unsubscribe();
+      expect(service.getMetrics().workspaceTargetCount).toBe(0);
 
-    snapshotSubscription.unsubscribe();
-    await service.getSnapshot(REPO_CWD, { force: true, reason: "after-unsubscribe" });
+      const latest = await service.getSnapshot(REPO_CWD, {
+        force: true,
+        reason: "replacement-target",
+      });
+      expect(latest.forge.pullRequest).toMatchObject({ state: "merged", isMerged: true });
+      expect(store.get(REPO_CWD)).toBe(latest);
 
-    expect(snapshotListener).toHaveBeenCalledTimes(1);
-
-    workspaceSubscription.unsubscribe();
-    service.dispose();
+      retiredRead.resolve(createPullRequestStatusResult());
+      const retired = await oldRefresh;
+      expect(retired.forge.pullRequest).toMatchObject({ state: "open", isMerged: false });
+      expect(store.get(REPO_CWD)).toBe(latest);
+      expect(published).toEqual([latest]);
+    } finally {
+      retiredRead.resolve(createPullRequestStatusResult());
+      await oldRefresh;
+      await service.dispose();
+    }
   });
 
   test("getSnapshot populates github pull request state in the runtime snapshot", async () => {

packages/server/src/server/workspace-git-service.ts

@@ -208,6 +208,7 @@ export interface WorkspaceGitService {
     listener: WorkspaceGitListener,
   ): WorkspaceGitSubscription;
 
+  /** Publishes computed snapshot changes independently of workspace UI subscriptions. */
   onSnapshotUpdated(listener: WorkspaceGitSnapshotUpdatedListener): WorkspaceGitSubscription;
   getCheckout(cwd: string): Promise<ProjectCheckoutLitePayload>;
   getSnapshot(
@@ -334,7 +335,6 @@ interface WorkspaceGitRefreshRequest {
   includeForge: boolean;
   emitUnchanged?: boolean;
   reason: string;
-  notify: boolean;
   queueIfBusy: boolean;
   movedRemoteRefs: Set<string>;
 }
@@ -747,7 +747,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   ): Promise<WorkspaceGitRuntimeSnapshot> {
     this.assertNotDisposed();
     cwd = resolve(cwd);
-    const request = this.normalizeRefreshRequest(options, "getSnapshot", true);
+    const request = this.normalizeRefreshRequest(options, "getSnapshot");
     const target = this.ensureWorkspaceTarget(cwd);
     if (!request.force && target.latestSnapshot) {
       return target.latestSnapshot;
@@ -974,7 +974,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       refreshWorktree: true,
       includeForge: false,
       reason: "refresh",
-      notify: true,
       queueIfBusy: false,
       movedRemoteRefs: new Set(),
     });
@@ -1239,7 +1238,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         refreshWorktree: true,
         includeForge: true,
         reason: "initial",
-        notify: true,
         queueIfBusy: false,
         movedRemoteRefs: new Set(),
       });
@@ -1631,7 +1629,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
             refreshWorktree: true,
             includeForge: false,
             reason: "working-tree-watch-fallback",
-            notify: true,
             queueIfBusy: true,
             movedRemoteRefs: new Set(),
           });
@@ -2691,7 +2688,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
             includeForge: false,
             emitUnchanged: true,
             reason: "git-metadata-watch-fallback",
-            notify: true,
             queueIfBusy: true,
             movedRemoteRefs: new Set(),
           });
@@ -2824,9 +2820,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
           this.rememberForgePrStatusSnapshot(
             target,
             buildForgeSnapshotFromStatus(status, resolution.forge),
-            {
-              notify: true,
-            },
           );
         },
         onError: (error) => {
@@ -2901,9 +2894,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         if (!closed && this.isActiveObservedWorkspaceTarget(target)) {
           latestStatus = status;
           consecutiveErrors = 0;
-          this.rememberForgePrStatusSnapshot(target, buildForgeSnapshotFromStatus(status, forge), {
-            notify: true,
-          });
+          this.rememberForgePrStatusSnapshot(target, buildForgeSnapshotFromStatus(status, forge));
         }
       } catch (error) {
         consecutiveErrors += 1;
@@ -3079,7 +3070,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   private normalizeRefreshRequest(
     options: WorkspaceGitSnapshotOptions | undefined,
     defaultReason: string,
-    notify: boolean,
   ): WorkspaceGitRefreshRequest {
     if (options?.force && !options.reason) {
       throw new Error("WorkspaceGitService.getSnapshot force refresh requires a reason");
@@ -3092,7 +3082,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       refreshWorktree: true,
       includeForge: options?.includeForge ?? true,
       reason: options?.reason ?? defaultReason,
-      notify,
       queueIfBusy: false,
       movedRemoteRefs: new Set(),
     };
@@ -3121,7 +3110,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       includeForge: options?.includeForge ?? false,
       emitUnchanged: options?.emitUnchanged,
       reason: options?.reason ?? "watch",
-      notify: true,
       queueIfBusy: options?.queueIfBusy ?? true,
       movedRemoteRefs: new Set(options?.movedRemoteRefs),
     };
@@ -3160,7 +3148,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         upgradesMovedRefs
           ? request.reason
           : pending.reason,
-      notify: pending.notify || request.notify,
       queueIfBusy: pending.queueIfBusy || request.queueIfBusy,
       movedRemoteRefs,
     };
@@ -3181,7 +3168,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         target.latestSnapshot
       ) {
         this.rememberSnapshot(target, target.latestSnapshot, {
-          notify: request.notify,
           forceEmit: true,
         });
       }
@@ -3198,7 +3184,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         }
         snapshot = admittedSnapshot;
         this.rememberSnapshot(target, snapshot, {
-          notify: request.notify,
           forceEmit:
             request.force || (request.emitUnchanged === true && request.movedRemoteRefs.size > 0),
         });
@@ -3485,7 +3470,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   private rememberForgePrStatusSnapshot(
     target: WorkspaceGitTarget,
     github: WorkspaceGitRuntimeSnapshot["forge"],
-    options?: { notify?: boolean },
   ): void {
     if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
       return;
@@ -3494,7 +3478,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     target.latestForge = github;
     target.latestForgeLoadedAtMs = this.deps.now().getTime();
     this.rememberSnapshot(target, this.combineSnapshot(target), {
-      notify: options?.notify,
       forceEmit: false,
     });
   }
@@ -3502,8 +3485,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   private rememberSnapshot(
     target: WorkspaceGitTarget,
     snapshot: WorkspaceGitRuntimeSnapshot,
-    options?: { forceEmit?: boolean; notify?: boolean },
+    options?: { forceEmit?: boolean },
   ): void {
+    if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+      return;
+    }
     target.latestSnapshot = snapshot;
     if (target.listeners.size > 0) {
       this.updateForgePrStatusPollForTarget(target);
@@ -3514,9 +3500,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       return;
     }
     target.latestFingerprint = fingerprint;
-    if (!options?.notify || target.listeners.size === 0) {
-      return;
-    }
     for (const listener of this.snapshotUpdatedListeners) {
       try {
         listener(snapshot);

packages/server/src/server/workspace-runtime-store/index.test.ts

@@ -1,5 +1,10 @@
+import { execFileSync } from "node:child_process";
+import { mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path, { resolve } from "node:path";
+import pino from "pino";
+import { WorkspaceGitServiceImpl } from "../workspace-git-service.js";
 import { expect, test } from "vitest";
-import { resolve } from "node:path";
 import { createNoGitWorkspaceRuntimeSnapshot } from "../test-utils/workspace-git-service-stub.js";
 import { createWorkspaceRuntimeStore } from "./index.js";
 
@@ -16,3 +21,50 @@ test("retains, replaces, and removes snapshots by resolved workspace directory",
   store.remove(resolve(cwd));
   expect(store.get(cwd)).toBeNull();
 });
+
+test("retains refreshed Git changes without a workspace subscription", async () => {
+  const root = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-runtime-store-")));
+  const repo = path.join(root, "repo");
+  let now = Date.now();
+  const service = new WorkspaceGitServiceImpl({
+    logger: pino({ enabled: false }),
+    paseoHome: path.join(root, "home"),
+    deps: { now: () => new Date(now) },
+  });
+  const store = createWorkspaceRuntimeStore();
+  service.onSnapshotUpdated((snapshot) => store.set(snapshot));
+  try {
+    execFileSync("git", ["init", "-q", "-b", "main", repo]);
+    const git = (...args: string[]) => execFileSync("git", args, { cwd: repo, stdio: "pipe" });
+    writeFileSync(path.join(repo, "README.md"), "# Temp\n");
+    git("add", "README.md");
+    git(
+      "-c",
+      "user.name=T",
+      "-c",
+      "user.email=t@localhost",
+      "-c",
+      "commit.gpgsign=false",
+      "commit",
+      "-qm",
+      "init",
+    );
+    execFileSync("git", ["init", "-q", "--bare", path.join(root, "remote.git")]);
+    git("remote", "add", "origin", path.join(root, "remote.git"));
+    git("push", "-q", "-u", "origin", "main");
+
+    writeFileSync(path.join(repo, "README.md"), "# Temp\none\ntwo\n");
+    await service.refresh(repo);
+    expect(service.getMetrics().workspaceListenerCount).toBe(0);
+    expect((await service.getSnapshot(repo)).git.diffStat).toEqual({ additions: 2, deletions: 0 });
+    expect(store.get(repo)?.git.diffStat).toEqual({ additions: 2, deletions: 0 });
+
+    now += 3_000;
+    writeFileSync(path.join(repo, "README.md"), "# Temp\none\ntwo\nthree\n");
+    await service.refresh(repo);
+    expect(store.get(repo)?.git.diffStat).toEqual({ additions: 3, deletions: 0 });
+  } finally {
+    await service.dispose();
+    rmSync(root, { recursive: true, force: true });
+  }
+});