refactor(server): extract workspace git-observer into a deep module (#1714)

* refactor(server): extract workspace git-observer into a deep module

Move the workspace git-observer cluster out of session.ts into
session/workspace-git-observer/ behind createWorkspaceGitObserverService(deps).
The module owns the per-cwd watch targets and the WorkspaceGitService
subscription handles, so the registration / dedupe / branch-change / teardown
lifecycle lives in one place instead of being reached into by hand from the
emit hot-path, the archive path, CheckoutSession's host, and dispose.

session.ts: 6401 -> 6219 lines. The caller surface narrows: archive teardown
calls removeForWorkspaceId / removeForCwd (no longer resolving watch targets and
cwds itself), and dispose() replaces iterating a raw subscription Map.

Sheds dead code orphaned by 40e27f5a0 (FSWatcher watching moved into
WorkspaceGitService): the workspaceGitFetchSubscriptions Map (never written), the
WorkspaceGitWatchTarget watchers/debounceTimer/refreshPromise/refreshQueued
fields, and the no-op closeWorkspaceGitWatchTarget method.

The module is unit-tested with injected fakes (zero mocks); the existing
session.workspace-git-watch.test.ts integration suite is repointed through the
new boundary and still passes.

Behavior is preserved verbatim, including a latent quirk where syncObservers
seeds descriptor state with a directory in the workspaceId slot (an effective
no-op) -- flagged as a follow-up, not changed here.

* refactor(server): clear watch targets on dispose; document shouldSkipUpdate dedupe

Addresses Greptile review on PR #1714:
- dispose() now clears watchTargets alongside subscriptions so post-teardown
  state is consistent (dispose is terminal; behavior-preserving).
- Document shouldSkipUpdate as a check-and-record dedupe gate (mutation on the
  changed path is intentional, preserved from the original).

* test(server): make git-observer test paths platform-portable

server-tests (windows-latest) failed: the test asserted raw POSIX path literals
against the service's resolve(cwd)-normalized output, which becomes D:\repo\ws1
on Windows. Resolve the test cwds the same way so assertions hold on every
platform (matches the path.resolve pattern in session.workspace-git-watch.test).
Test-only fix; production normalization is unchanged and correct.

packages/server/src/server/session.ts

@@ -1,6 +1,5 @@
 import equal from "fast-deep-equal";
 import { v4 as uuidv4 } from "uuid";
-import type { FSWatcher } from "node:fs";
 import { stat } from "node:fs/promises";
 import { basename, normalize, resolve, sep } from "path";
 import { homedir } from "node:os";
@@ -137,6 +136,10 @@ import { wrapSpokenInput } from "./voice-config.js";
 import { isVoicePermissionAllowed } from "./voice-permission-policy.js";
 import { VoiceSession } from "./session/voice/voice-session.js";
 import { CheckoutSession } from "./session/checkout/checkout-session.js";
+import {
+  createWorkspaceGitObserverService,
+  type WorkspaceGitObserverService,
+} from "./session/workspace-git-observer/workspace-git-observer-service.js";
 import {
   createAgentStructuredTextGeneration,
   createGitMetadataGenerator,
@@ -213,8 +216,6 @@ import { type WorktreeConfig, createWorktree } from "../utils/worktree.js";
 import { runGitCommand } from "../utils/run-git-command.js";
 import { CreateAgentLifecycleDispatch } from "./agent/create-agent-lifecycle-dispatch.js";
 
-const WORKSPACE_GIT_WATCH_REMOVED_STATE_KEY = "__removed__";
-
 // TODO: Remove once all app store clients are on >=0.1.45 and understand arbitrary provider strings.
 // Clients before 0.1.45 validate providers with z.enum(["claude", "codex", "opencode"]) and reject
 // the entire session message if they encounter an unknown provider.
@@ -323,17 +324,6 @@ export function resolveWaitForFinishError(options: {
   return typeof message === "string" && message.trim().length > 0 ? message : "Agent failed";
 }
 
-interface WorkspaceGitWatchTarget {
-  cwd: string;
-  workspaceId: string;
-  watchers: FSWatcher[];
-  debounceTimer: ReturnType<typeof setTimeout> | null;
-  refreshPromise: Promise<void> | null;
-  refreshQueued: boolean;
-  latestDescriptorStateKey: string | null;
-  lastBranchName: string | null;
-}
-
 export interface SessionRuntimeMetrics {
   terminalDirectorySubscriptionCount: number;
   terminalSubscriptionCount: number;
@@ -584,11 +574,6 @@ export class Session {
   private readonly providerSnapshotManager: ProviderSnapshotManager;
   private readonly serviceProxy: ServiceProxySubsystem | null;
   private readonly scriptRuntimeStore: WorkspaceScriptRuntimeStore | null;
-  private readonly onBranchChanged?: (
-    workspaceId: string,
-    oldBranch: string | null,
-    newBranch: string | null,
-  ) => void;
   private readonly getDaemonTcpPort: (() => number | null) | null;
   private readonly getDaemonTcpHost: (() => string | null) | null;
   private readonly serviceProxyPublicBaseUrl: string | null;
@@ -596,10 +581,8 @@ export class Session {
   private readonly terminalController: TerminalSessionController;
   private inflightRequests = 0;
   private peakInflightRequests = 0;
-  private readonly workspaceGitWatchTargets = new Map<string, WorkspaceGitWatchTarget>();
   private readonly workspaceSetupSnapshots: Map<string, WorkspaceSetupSnapshot>;
-  private readonly workspaceGitFetchSubscriptions = new Map<string, () => void>();
-  private readonly workspaceGitSubscriptions = new Map<string, () => void>();
+  private readonly workspaceGitObserver: WorkspaceGitObserverService;
   private readonly workspaceDirectory: WorkspaceDirectory;
   private readonly voiceSession: VoiceSession;
   private readonly checkoutSession: CheckoutSession;
@@ -711,7 +694,7 @@ export class Session {
         emit: (msg) => this.emit(msg),
         emitWorkspaceUpdateForCwd: (cwd) => this.emitWorkspaceUpdateForCwd(cwd),
         handleWorkspaceGitBranchSnapshot: (cwd, branchName) =>
-          this.handleWorkspaceGitBranchSnapshot(cwd, branchName),
+          this.workspaceGitObserver.handleBranchSnapshot(cwd, branchName),
         renameCurrentBranch: (cwd, branch) => this.renameCurrentBranch(cwd, branch),
       },
       gitMutation: this.gitMutation,
@@ -731,6 +714,17 @@ export class Session {
       worktreesRoot: this.worktreesRoot,
       logger: this.sessionLogger,
     });
+    this.workspaceGitObserver = createWorkspaceGitObserverService({
+      workspaceGitService: this.workspaceGitService,
+      describeWorkspaceRecordWithGitData: (workspace) =>
+        this.describeWorkspaceRecordWithGitData(workspace),
+      emitWorkspaceUpdateForCwd: (cwd) => this.emitWorkspaceUpdateForCwd(cwd),
+      emitWorkspaceUpdateForWorkspaceId: (workspaceId) =>
+        this.emitWorkspaceUpdateForWorkspaceId(workspaceId),
+      emitStatusUpdate: (cwd, snapshot) => this.checkoutSession.emitStatusUpdate(cwd, snapshot),
+      onBranchChanged,
+      logger: this.sessionLogger,
+    });
     this.chatScheduleLoopSession = new ChatScheduleLoopSession({
       host: {
         emit: (msg) => this.emit(msg),
@@ -849,7 +843,6 @@ export class Session {
     this.serviceProxy = serviceProxy ?? null;
     this.scriptRuntimeStore = scriptRuntimeStore ?? null;
     this.workspaceSetupSnapshots = workspaceSetupSnapshots ?? new Map();
-    this.onBranchChanged = onBranchChanged;
     this.getDaemonTcpPort = getDaemonTcpPort ?? null;
     this.getDaemonTcpHost = getDaemonTcpHost ?? null;
     this.serviceProxyPublicBaseUrl = serviceProxyPublicBaseUrl ?? null;
@@ -922,8 +915,7 @@ export class Session {
   }
 
   async syncWorkspaceGitObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
-    const descriptor = await this.describeWorkspaceRecordWithGitData(workspace);
-    this.syncWorkspaceGitObservers([descriptor]);
+    await this.workspaceGitObserver.syncObserverForWorkspace(workspace);
   }
 
   async emitWorkspaceUpdateForWorkspaceId(workspaceId: string): Promise<void> {
@@ -950,8 +942,7 @@ export class Session {
   }
 
   async warmWorkspaceGitDataForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
-    await this.syncWorkspaceGitObserverForWorkspace(workspace);
-    await this.emitWorkspaceUpdateForWorkspaceId(workspace.workspaceId);
+    await this.workspaceGitObserver.warmGitData(workspace);
   }
 
   /**
@@ -3475,158 +3466,6 @@ export class Session {
     }
   }
 
-  private closeWorkspaceGitWatchTarget(target: WorkspaceGitWatchTarget): void {
-    if (target.debounceTimer) {
-      clearTimeout(target.debounceTimer);
-      target.debounceTimer = null;
-    }
-    for (const watcher of target.watchers) {
-      try {
-        watcher.close();
-      } catch {
-        // Ignore watcher close errors
-      }
-    }
-    target.watchers.length = 0;
-  }
-
-  private async removeWorkspaceGitWatchTarget(cwd: string): Promise<void> {
-    const normalizedCwd = resolve(cwd);
-    const target = this.workspaceGitWatchTargets.get(normalizedCwd);
-    if (target) {
-      this.closeWorkspaceGitWatchTarget(target);
-      this.workspaceGitWatchTargets.delete(normalizedCwd);
-    }
-  }
-
-  private removeWorkspaceGitSubscription(cwd: string): void {
-    const normalizedCwd = resolve(cwd);
-    const target = this.workspaceGitWatchTargets.get(normalizedCwd);
-    if (target) {
-      const unsubscribeFetch = this.workspaceGitFetchSubscriptions.get(normalizedCwd);
-      unsubscribeFetch?.();
-      this.workspaceGitFetchSubscriptions.delete(normalizedCwd);
-      this.closeWorkspaceGitWatchTarget(target);
-      this.workspaceGitWatchTargets.delete(normalizedCwd);
-    }
-    this.workspaceGitSubscriptions.get(normalizedCwd)?.();
-    this.workspaceGitSubscriptions.delete(normalizedCwd);
-  }
-
-  private workspaceGitDescriptorStateKey(workspace: WorkspaceDescriptorPayload | null): string {
-    if (!workspace) {
-      return WORKSPACE_GIT_WATCH_REMOVED_STATE_KEY;
-    }
-    return JSON.stringify([
-      workspace.name,
-      workspace.diffStat ? [workspace.diffStat.additions, workspace.diffStat.deletions] : null,
-    ]);
-  }
-
-  private resolveWorkspaceGitWatchTarget(workspaceId: string): WorkspaceGitWatchTarget | null {
-    for (const target of this.workspaceGitWatchTargets.values()) {
-      if (target.workspaceId === workspaceId) {
-        return target;
-      }
-    }
-    return null;
-  }
-
-  private shouldSkipWorkspaceGitWatchUpdate(
-    workspaceId: string,
-    workspace: WorkspaceDescriptorPayload | null,
-  ): boolean {
-    const target = this.resolveWorkspaceGitWatchTarget(workspaceId);
-    if (!target) {
-      return false;
-    }
-    const nextStateKey = this.workspaceGitDescriptorStateKey(workspace);
-    if (target.latestDescriptorStateKey === nextStateKey) {
-      return true;
-    }
-    target.latestDescriptorStateKey = nextStateKey;
-    return false;
-  }
-
-  private rememberWorkspaceGitDescriptorState(
-    workspaceId: string,
-    workspace: WorkspaceDescriptorPayload | null,
-  ): void {
-    const target = this.resolveWorkspaceGitWatchTarget(workspaceId);
-    if (!target) {
-      return;
-    }
-    target.latestDescriptorStateKey = this.workspaceGitDescriptorStateKey(workspace);
-    target.lastBranchName = workspace?.name ?? null;
-  }
-
-  private handleWorkspaceGitBranchSnapshot(cwd: string, branchName: string | null): void {
-    const target = this.workspaceGitWatchTargets.get(resolve(cwd));
-    if (!target) {
-      return;
-    }
-
-    const previousBranchName = target.lastBranchName;
-    if (branchName === previousBranchName) {
-      return;
-    }
-
-    target.lastBranchName = branchName;
-    this.onBranchChanged?.(target.workspaceId, previousBranchName, branchName);
-  }
-
-  private syncWorkspaceGitObservers(workspaces: Iterable<WorkspaceDescriptorPayload>): void {
-    for (const workspace of workspaces) {
-      this.syncWorkspaceGitObserver(workspace.workspaceDirectory, {
-        isGit: workspace.projectKind === "git",
-        workspaceId: workspace.id,
-      });
-      this.rememberWorkspaceGitDescriptorState(workspace.workspaceDirectory, workspace);
-    }
-  }
-
-  private syncWorkspaceGitObserver(
-    cwd: string,
-    options: { isGit: boolean; workspaceId: string },
-  ): void {
-    const normalizedCwd = resolve(cwd);
-    if (!options.isGit) {
-      this.removeWorkspaceGitSubscription(normalizedCwd);
-      return;
-    }
-
-    if (this.workspaceGitSubscriptions.has(normalizedCwd)) {
-      return;
-    }
-
-    const target: WorkspaceGitWatchTarget = {
-      cwd: normalizedCwd,
-      workspaceId: options.workspaceId,
-      watchers: [],
-      debounceTimer: null,
-      refreshPromise: null,
-      refreshQueued: false,
-      latestDescriptorStateKey: null,
-      lastBranchName: null,
-    };
-    this.workspaceGitWatchTargets.set(normalizedCwd, target);
-
-    const subscription = this.workspaceGitService.registerWorkspace(
-      { cwd: normalizedCwd },
-      (snapshot) => {
-        this.handleWorkspaceGitBranchSnapshot(normalizedCwd, snapshot.git.currentBranch ?? null);
-        void this.emitWorkspaceUpdateForCwd(normalizedCwd).catch((error) => {
-          this.sessionLogger.warn(
-            { err: error, cwd: normalizedCwd },
-            "Failed to emit workspace update after git branch snapshot",
-          );
-        });
-        this.checkoutSession.emitStatusUpdate(normalizedCwd, snapshot);
-      },
-    );
-    this.workspaceGitSubscriptions.set(normalizedCwd, subscription.unsubscribe);
-  }
-
   private async handlePaseoWorktreeListRequest(
     msg: Extract<SessionInboundMessage, { type: "paseo_worktree_list_request" }>,
   ): Promise<void> {
@@ -4439,10 +4278,7 @@ export class Session {
       workspaceRegistry: this.workspaceRegistry,
     });
     if (!existingWorkspace) {
-      const watchTarget = this.resolveWorkspaceGitWatchTarget(workspaceId);
-      if (watchTarget) {
-        this.removeWorkspaceGitSubscription(watchTarget.cwd);
-      }
+      this.workspaceGitObserver.removeForWorkspaceId(workspaceId);
       return;
     }
 
@@ -4475,9 +4311,8 @@ export class Session {
     workspaceId: string;
     cwd: string;
   }): Promise<void> {
-    await this.removeWorkspaceGitWatchTarget(input.cwd);
+    this.workspaceGitObserver.removeForCwd(input.cwd);
     this.scriptRuntimeStore?.removeForWorkspace(input.workspaceId);
-    this.removeWorkspaceGitSubscription(input.cwd);
   }
 
   private async reconcileAndEmitWorkspaceUpdates(): Promise<void> {
@@ -4567,11 +4402,11 @@ export class Session {
           : null;
       if (
         options?.dedupeGitState &&
-        this.shouldSkipWorkspaceGitWatchUpdate(workspaceId, nextWorkspace)
+        this.workspaceGitObserver.shouldSkipUpdate(workspaceId, nextWorkspace)
       ) {
         continue;
       }
-      this.recordWorkspaceGitDescriptorState(workspaceId, nextWorkspace);
+      this.workspaceGitObserver.recordDescriptorState(workspaceId, nextWorkspace);
 
       if (!nextWorkspace) {
         subscription.lastEmittedByWorkspaceId.delete(workspaceId);
@@ -4604,20 +4439,6 @@ export class Session {
     }
   }
 
-  private recordWorkspaceGitDescriptorState(
-    workspaceId: string,
-    nextWorkspace: WorkspaceDescriptorPayload | null,
-  ): void {
-    const watchTarget = this.resolveWorkspaceGitWatchTarget(workspaceId);
-    if (watchTarget && this.onBranchChanged) {
-      const newBranchName = nextWorkspace?.name ?? null;
-      if (newBranchName !== watchTarget.lastBranchName) {
-        this.onBranchChanged(workspaceId, watchTarget.lastBranchName, newBranchName);
-      }
-    }
-    this.rememberWorkspaceGitDescriptorState(workspaceId, nextWorkspace);
-  }
-
   private async buildWorkspaceRemoveUpdatePayload(
     workspaceId: string,
     removedProjectId?: string,
@@ -4835,7 +4656,7 @@ export class Session {
       }
 
       const payload = await this.listFetchWorkspacesEntries(request);
-      this.syncWorkspaceGitObservers(payload.entries);
+      this.workspaceGitObserver.syncObservers(payload.entries);
       this.sessionLogger.debug(
         {
           requestId: request.requestId,
@@ -6179,9 +6000,6 @@ export class Session {
 
     this.checkoutSession.cleanup();
 
-    for (const unsubscribe of this.workspaceGitSubscriptions.values()) {
-      unsubscribe();
-    }
-    this.workspaceGitSubscriptions.clear();
+    this.workspaceGitObserver.dispose();
   }
 }

packages/server/src/server/session.workspace-git-watch.test.ts

@@ -13,12 +13,13 @@ import type {
   WorkspaceGitRuntimeSnapshot,
   WorkspaceGitService,
 } from "./workspace-git-service.js";
-import type { SessionOutboundMessage } from "./messages.js";
+import type { SessionOutboundMessage, WorkspaceDescriptorPayload } from "./messages.js";
 import {
   createPersistedProjectRecord,
   createPersistedWorkspaceRecord,
 } from "./workspace-registry.js";
 import { createNoopWorkspaceGitService } from "./test-utils/workspace-git-service-stub.js";
+import type { WorkspaceGitObserverService } from "./session/workspace-git-observer/workspace-git-observer-service.js";
 
 interface SessionInternals {
   workspaceUpdatesSubscription: {
@@ -29,10 +30,23 @@ interface SessionInternals {
     lastEmittedByWorkspaceId: Map<string, unknown>;
   };
   buildWorkspaceDescriptorMap: () => Promise<Map<string, unknown>>;
-  syncWorkspaceGitObserver(cwd: string, details: { isGit: boolean; workspaceId: string }): void;
+  workspaceGitObserver: WorkspaceGitObserverService;
   listAgentPayloads: () => Promise<unknown[]>;
 }
 
+// The observer's single public registration entry is syncObservers(descriptors); these
+// integration tests drive it with a minimal git descriptor for one workspace, then push
+// snapshots through the captured WorkspaceGitService listener.
+function syncGitObserver(session: Session, cwd: string, workspaceId: string): void {
+  asInternals<SessionInternals>(session).workspaceGitObserver.syncObservers([
+    {
+      id: workspaceId,
+      workspaceDirectory: cwd,
+      projectKind: "git",
+    } as unknown as WorkspaceDescriptorPayload,
+  ]);
+}
+
 type CheckoutStatusUpdatePayload = Extract<
   SessionOutboundMessage,
   { type: "checkout_status_update" }
@@ -324,7 +338,7 @@ describe("workspace git watch targets", () => {
 
     sessionAny.buildWorkspaceDescriptorMap = async () => new Map([[descriptor.id, descriptor]]);
 
-    sessionAny.syncWorkspaceGitObserver(REPO_CWD, { isGit: true, workspaceId: "ws-10" });
+    syncGitObserver(session, REPO_CWD, "ws-10");
 
     expect(workspaceGitService.registerWorkspace).toHaveBeenCalledWith(
       { cwd: REPO_CWD },
@@ -383,7 +397,7 @@ describe("workspace git watch targets", () => {
       lastEmittedByWorkspaceId: new Map(),
     };
 
-    sessionAny.syncWorkspaceGitObserver(REPO_CWD, { isGit: true, workspaceId: "ws-10" });
+    syncGitObserver(session, REPO_CWD, "ws-10");
     emitted.length = 0;
 
     subscriptions[0]?.listener(
@@ -466,7 +480,7 @@ describe("workspace git watch targets", () => {
       name: "old-branch",
     });
 
-    sessionAny.syncWorkspaceGitObserver("/tmp/repo", { isGit: true, workspaceId: "ws-10" });
+    syncGitObserver(session, "/tmp/repo", "ws-10");
 
     subscriptions[0]?.listener(
       createWorkspaceRuntimeSnapshot("/tmp/repo", {
@@ -542,14 +556,14 @@ describe("workspace git watch targets", () => {
       name: "main",
     });
 
-    sessionAny.syncWorkspaceGitObserver(REPO_CWD, { isGit: true, workspaceId: "ws-10" });
+    syncGitObserver(session, REPO_CWD, "ws-10");
     expect(subscriptions).toHaveLength(1);
 
     await sessionAny.archiveWorkspaceRecord("ws-10");
 
     // Re-observing the directory establishes a fresh subscription only if archive
     // tore down the prior one, which is keyed by cwd — not the opaque workspace id.
-    sessionAny.syncWorkspaceGitObserver(REPO_CWD, { isGit: true, workspaceId: "ws-10" });
+    syncGitObserver(session, REPO_CWD, "ws-10");
     expect(subscriptions).toHaveLength(2);
 
     await session.cleanup();
@@ -575,7 +589,7 @@ describe("workspace git watch targets", () => {
       lastEmittedByWorkspaceId: new Map(),
     };
 
-    sessionAny.syncWorkspaceGitObserver(REPO_CWD, { isGit: true, workspaceId: "ws-10" });
+    syncGitObserver(session, REPO_CWD, "ws-10");
     emitted.length = 0;
 
     subscriptions[0]?.listener(

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

@@ -0,0 +1,287 @@
+import { resolve } from "node:path";
+import type pino from "pino";
+import { describe, expect, test } from "vitest";
+import type { WorkspaceDescriptorPayload } from "../../messages.js";
+import type {
+  WorkspaceGitListener,
+  WorkspaceGitRuntimeSnapshot,
+  WorkspaceGitService,
+} from "../../workspace-git-service.js";
+import type { PersistedWorkspaceRecord } from "../../workspace-registry.js";
+import { createWorkspaceGitObserverService } from "./workspace-git-observer-service.js";
+
+// Watch targets are keyed by resolve(cwd), which is platform-dependent (POSIX vs Windows
+// drive paths). Resolve the test cwds the same way so assertions hold on every platform.
+const WS1 = resolve("/repo/ws1");
+const WS2 = resolve("/repo/ws2");
+
+// The service reads only WorkspaceGitService.registerWorkspace plus a handful of injected
+// session callbacks. The harness below implements exactly that slice as in-memory adapters:
+// registerWorkspace captures the per-cwd listener so a test can drive a git snapshot, and the
+// callbacks are capture-arrays. No mocks — the seams are the injected ports.
+
+function makeDescriptor(overrides: {
+  id: string;
+  workspaceDirectory: string;
+  projectKind?: string;
+  name?: string | null;
+  diffStat?: { additions: number; deletions: number } | null;
+}): WorkspaceDescriptorPayload {
+  return {
+    id: overrides.id,
+    workspaceDirectory: overrides.workspaceDirectory,
+    projectKind: overrides.projectKind ?? "git",
+    name: overrides.name ?? null,
+    diffStat: overrides.diffStat ?? null,
+  } as unknown as WorkspaceDescriptorPayload;
+}
+
+function makeSnapshot(cwd: string, currentBranch: string | null): WorkspaceGitRuntimeSnapshot {
+  return { cwd, git: { currentBranch } } as unknown as WorkspaceGitRuntimeSnapshot;
+}
+
+function makeRecord(workspaceId: string): PersistedWorkspaceRecord {
+  return { workspaceId } as unknown as PersistedWorkspaceRecord;
+}
+
+function flushMicrotasks(): Promise<void> {
+  return new Promise((done) => setImmediate(done));
+}
+
+function buildHarness(opts: { emitCwdRejects?: boolean } = {}) {
+  const listeners = new Map<string, WorkspaceGitListener>();
+  const registerCalls: string[] = [];
+  const unsubscribeCalls: string[] = [];
+  const emitCwdCalls: string[] = [];
+  const emitWorkspaceIdCalls: string[] = [];
+  const statusCalls: Array<{ cwd: string; branch: string | null }> = [];
+  const branchChanges: Array<[string, string | null, string | null]> = [];
+  const warnCalls: unknown[][] = [];
+  const describeCalls: PersistedWorkspaceRecord[] = [];
+  let describeResult: WorkspaceDescriptorPayload | null = null;
+
+  const workspaceGitService: Pick<WorkspaceGitService, "registerWorkspace"> = {
+    registerWorkspace({ cwd }, listener) {
+      registerCalls.push(cwd);
+      listeners.set(cwd, listener);
+      return {
+        unsubscribe() {
+          unsubscribeCalls.push(cwd);
+          listeners.delete(cwd);
+        },
+      };
+    },
+  };
+
+  const service = createWorkspaceGitObserverService({
+    workspaceGitService,
+    describeWorkspaceRecordWithGitData: async (workspace) => {
+      describeCalls.push(workspace);
+      if (!describeResult) {
+        throw new Error("describeResult not set");
+      }
+      return describeResult;
+    },
+    emitWorkspaceUpdateForCwd: async (cwd) => {
+      emitCwdCalls.push(cwd);
+      if (opts.emitCwdRejects) {
+        throw new Error("emit boom");
+      }
+    },
+    emitWorkspaceUpdateForWorkspaceId: async (workspaceId) => {
+      emitWorkspaceIdCalls.push(workspaceId);
+    },
+    emitStatusUpdate: (cwd, snapshot) => {
+      statusCalls.push({ cwd, branch: snapshot.git.currentBranch ?? null });
+    },
+    onBranchChanged: (workspaceId, oldBranch, newBranch) => {
+      branchChanges.push([workspaceId, oldBranch, newBranch]);
+    },
+    logger: { warn: (...args: unknown[]) => warnCalls.push(args) } as unknown as pino.Logger,
+  });
+
+  function emitSnapshot(cwd: string, branch: string | null): void {
+    const listener = listeners.get(cwd);
+    if (!listener) {
+      throw new Error(`no listener registered for ${cwd}`);
+    }
+    listener(makeSnapshot(cwd, branch));
+  }
+
+  return {
+    service,
+    emitSnapshot,
+    registerCalls,
+    unsubscribeCalls,
+    emitCwdCalls,
+    emitWorkspaceIdCalls,
+    statusCalls,
+    branchChanges,
+    warnCalls,
+    describeCalls,
+    setDescribeResult: (descriptor: WorkspaceDescriptorPayload) => {
+      describeResult = descriptor;
+    },
+  };
+}
+
+describe("syncObservers", () => {
+  test("registers a WorkspaceGitService subscription for a git workspace", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    expect(h.registerCalls).toEqual([WS1]);
+  });
+
+  test("does not register a non-git workspace", () => {
+    const h = buildHarness();
+    h.service.syncObservers([
+      makeDescriptor({ id: "ws1", workspaceDirectory: WS1, projectKind: "directory" }),
+    ]);
+    expect(h.registerCalls).toEqual([]);
+  });
+
+  test("is idempotent — re-syncing the same git workspace does not re-register", () => {
+    const h = buildHarness();
+    const descriptor = makeDescriptor({ id: "ws1", workspaceDirectory: WS1 });
+    h.service.syncObservers([descriptor]);
+    h.service.syncObservers([descriptor]);
+    expect(h.registerCalls).toEqual([WS1]);
+  });
+
+  test("tears down the subscription when a git workspace becomes non-git", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.service.syncObservers([
+      makeDescriptor({ id: "ws1", workspaceDirectory: WS1, projectKind: "directory" }),
+    ]);
+    expect(h.unsubscribeCalls).toEqual([WS1]);
+  });
+});
+
+describe("git snapshot listener", () => {
+  test("fans a snapshot out to branch-change, workspace-update, and status-update", async () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.emitSnapshot(WS1, "feature");
+    await flushMicrotasks();
+    expect(h.branchChanges).toEqual([["ws1", null, "feature"]]);
+    expect(h.emitCwdCalls).toEqual([WS1]);
+    expect(h.statusCalls).toEqual([{ cwd: WS1, branch: "feature" }]);
+  });
+
+  test("does not re-fire onBranchChanged when the branch is unchanged", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.emitSnapshot(WS1, "feature");
+    h.emitSnapshot(WS1, "feature");
+    expect(h.branchChanges).toEqual([["ws1", null, "feature"]]);
+  });
+
+  test("logs and swallows an emit failure without skipping the status update", async () => {
+    const h = buildHarness({ emitCwdRejects: true });
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    expect(() => h.emitSnapshot(WS1, "feature")).not.toThrow();
+    expect(h.statusCalls).toEqual([{ cwd: WS1, branch: "feature" }]);
+    await flushMicrotasks();
+    expect(h.warnCalls).toHaveLength(1);
+  });
+});
+
+describe("shouldSkipUpdate", () => {
+  test("returns false when no observer exists for the workspace", () => {
+    const h = buildHarness();
+    expect(h.service.shouldSkipUpdate("unknown", null)).toBe(false);
+  });
+
+  test("skips a repeat descriptor state and re-emits when it changes", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    const a = makeDescriptor({ id: "ws1", workspaceDirectory: WS1, name: "main" });
+    const b = makeDescriptor({ id: "ws1", workspaceDirectory: WS1, name: "feature" });
+    expect(h.service.shouldSkipUpdate("ws1", a)).toBe(false);
+    expect(h.service.shouldSkipUpdate("ws1", a)).toBe(true);
+    expect(h.service.shouldSkipUpdate("ws1", b)).toBe(false);
+  });
+});
+
+describe("recordDescriptorState", () => {
+  test("fires onBranchChanged once per branch name transition", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    const feature = makeDescriptor({ id: "ws1", workspaceDirectory: WS1, name: "feature" });
+    h.service.recordDescriptorState("ws1", feature);
+    h.service.recordDescriptorState("ws1", feature);
+    expect(h.branchChanges).toEqual([["ws1", null, "feature"]]);
+  });
+
+  test("does nothing for an unknown workspace", () => {
+    const h = buildHarness();
+    h.service.recordDescriptorState(
+      "unknown",
+      makeDescriptor({ id: "x", workspaceDirectory: "/x" }),
+    );
+    expect(h.branchChanges).toEqual([]);
+  });
+});
+
+describe("teardown", () => {
+  test("removeForWorkspaceId unsubscribes the matching observer", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.service.removeForWorkspaceId("ws1");
+    expect(h.unsubscribeCalls).toEqual([WS1]);
+  });
+
+  test("removeForWorkspaceId is a no-op for an unknown workspace", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.service.removeForWorkspaceId("nope");
+    expect(h.unsubscribeCalls).toEqual([]);
+  });
+
+  test("removeForCwd unsubscribes and stops the observer", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.service.removeForCwd(WS1);
+    expect(h.unsubscribeCalls).toEqual([WS1]);
+    expect(() => h.emitSnapshot(WS1, "x")).toThrow();
+  });
+
+  test("dispose releases every live subscription", () => {
+    const h = buildHarness();
+    h.service.syncObservers([
+      makeDescriptor({ id: "ws1", workspaceDirectory: WS1 }),
+      makeDescriptor({ id: "ws2", workspaceDirectory: WS2 }),
+    ]);
+    h.service.dispose();
+    expect(h.unsubscribeCalls.sort()).toEqual([WS1, WS2]);
+  });
+
+  test("dispose clears watch targets so post-teardown lookups find nothing", () => {
+    const h = buildHarness();
+    h.service.syncObservers([makeDescriptor({ id: "ws1", workspaceDirectory: WS1 })]);
+    h.service.dispose();
+    const descriptor = makeDescriptor({ id: "ws1", workspaceDirectory: WS1, name: "main" });
+    expect(h.service.shouldSkipUpdate("ws1", descriptor)).toBe(false);
+    h.service.recordDescriptorState("ws1", descriptor);
+    expect(h.branchChanges).toEqual([]);
+  });
+});
+
+describe("syncObserverForWorkspace / warmGitData", () => {
+  test("describes the record then registers the observer", async () => {
+    const h = buildHarness();
+    h.setDescribeResult(makeDescriptor({ id: "ws1", workspaceDirectory: WS1 }));
+    await h.service.syncObserverForWorkspace(makeRecord("ws1"));
+    expect(h.describeCalls).toEqual([makeRecord("ws1")]);
+    expect(h.registerCalls).toEqual([WS1]);
+  });
+
+  test("warmGitData registers the observer and emits a workspace update", async () => {
+    const h = buildHarness();
+    h.setDescribeResult(makeDescriptor({ id: "ws1", workspaceDirectory: WS1 }));
+    await h.service.warmGitData(makeRecord("ws1"));
+    expect(h.registerCalls).toEqual([WS1]);
+    expect(h.emitWorkspaceIdCalls).toEqual(["ws1"]);
+  });
+});

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

@@ -0,0 +1,227 @@
+import { resolve } from "node:path";
+import type pino from "pino";
+import type { WorkspaceDescriptorPayload } from "../../messages.js";
+import type {
+  WorkspaceGitRuntimeSnapshot,
+  WorkspaceGitService,
+} from "../../workspace-git-service.js";
+import type { PersistedWorkspaceRecord } from "../../workspace-registry.js";
+
+const WORKSPACE_GIT_WATCH_REMOVED_STATE_KEY = "__removed__";
+
+interface WorkspaceGitWatchTarget {
+  cwd: string;
+  workspaceId: string;
+  latestDescriptorStateKey: string | null;
+  lastBranchName: string | null;
+}
+
+/**
+ * Observes a workspace's git state on disk (via WorkspaceGitService) and drives the
+ * live update fan-out: branch-change notifications, workspace-card refreshes, and
+ * checkout status updates. It owns the per-cwd watch targets and the WorkspaceGitService
+ * subscription handles, so the registration / dedupe / teardown lifecycle lives in one
+ * module instead of being smeared across the client session.
+ *
+ * Branch changes reach `onBranchChanged` from two paths that share `lastBranchName`: the
+ * on-disk snapshot listener (handleBranchSnapshot) and the workspace-emit loop
+ * (recordDescriptorState). Both stay inside this module so the shared state is coherent.
+ */
+export interface WorkspaceGitObserverService {
+  syncObservers(workspaces: Iterable<WorkspaceDescriptorPayload>): void;
+  syncObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void>;
+  warmGitData(workspace: PersistedWorkspaceRecord): Promise<void>;
+  // Check-and-record dedupe gate: returns true when the descriptor state is unchanged
+  // for this workspace, and otherwise advances the recorded state key as a side effect.
+  shouldSkipUpdate(workspaceId: string, workspace: WorkspaceDescriptorPayload | null): boolean;
+  recordDescriptorState(workspaceId: string, workspace: WorkspaceDescriptorPayload | null): void;
+  handleBranchSnapshot(cwd: string, branchName: string | null): void;
+  removeForWorkspaceId(workspaceId: string): void;
+  removeForCwd(cwd: string): void;
+  dispose(): void;
+}
+
+export function createWorkspaceGitObserverService(deps: {
+  workspaceGitService: Pick<WorkspaceGitService, "registerWorkspace">;
+  describeWorkspaceRecordWithGitData: (
+    workspace: PersistedWorkspaceRecord,
+  ) => Promise<WorkspaceDescriptorPayload>;
+  emitWorkspaceUpdateForCwd: (cwd: string) => Promise<void>;
+  emitWorkspaceUpdateForWorkspaceId: (workspaceId: string) => Promise<void>;
+  emitStatusUpdate: (cwd: string, snapshot: WorkspaceGitRuntimeSnapshot) => void;
+  onBranchChanged?: (
+    workspaceId: string,
+    oldBranch: string | null,
+    newBranch: string | null,
+  ) => void;
+  logger: pino.Logger;
+}): WorkspaceGitObserverService {
+  const {
+    workspaceGitService,
+    describeWorkspaceRecordWithGitData,
+    emitWorkspaceUpdateForCwd,
+    emitWorkspaceUpdateForWorkspaceId,
+    emitStatusUpdate,
+    onBranchChanged,
+    logger,
+  } = deps;
+
+  const watchTargets = new Map<string, WorkspaceGitWatchTarget>();
+  const subscriptions = new Map<string, () => void>();
+
+  function descriptorStateKey(workspace: WorkspaceDescriptorPayload | null): string {
+    if (!workspace) {
+      return WORKSPACE_GIT_WATCH_REMOVED_STATE_KEY;
+    }
+    return JSON.stringify([
+      workspace.name,
+      workspace.diffStat ? [workspace.diffStat.additions, workspace.diffStat.deletions] : null,
+    ]);
+  }
+
+  function resolveTargetByWorkspaceId(workspaceId: string): WorkspaceGitWatchTarget | null {
+    for (const target of watchTargets.values()) {
+      if (target.workspaceId === workspaceId) {
+        return target;
+      }
+    }
+    return null;
+  }
+
+  function rememberDescriptorState(
+    workspaceId: string,
+    workspace: WorkspaceDescriptorPayload | null,
+  ): void {
+    const target = resolveTargetByWorkspaceId(workspaceId);
+    if (!target) {
+      return;
+    }
+    target.latestDescriptorStateKey = descriptorStateKey(workspace);
+    target.lastBranchName = workspace?.name ?? null;
+  }
+
+  function removeForCwd(cwd: string): void {
+    const normalizedCwd = resolve(cwd);
+    watchTargets.delete(normalizedCwd);
+    subscriptions.get(normalizedCwd)?.();
+    subscriptions.delete(normalizedCwd);
+  }
+
+  function handleBranchSnapshot(cwd: string, branchName: string | null): void {
+    const target = watchTargets.get(resolve(cwd));
+    if (!target) {
+      return;
+    }
+
+    const previousBranchName = target.lastBranchName;
+    if (branchName === previousBranchName) {
+      return;
+    }
+
+    target.lastBranchName = branchName;
+    onBranchChanged?.(target.workspaceId, previousBranchName, branchName);
+  }
+
+  function syncObserver(cwd: string, options: { isGit: boolean; workspaceId: string }): void {
+    const normalizedCwd = resolve(cwd);
+    if (!options.isGit) {
+      removeForCwd(normalizedCwd);
+      return;
+    }
+
+    if (subscriptions.has(normalizedCwd)) {
+      return;
+    }
+
+    const target: WorkspaceGitWatchTarget = {
+      cwd: normalizedCwd,
+      workspaceId: options.workspaceId,
+      latestDescriptorStateKey: null,
+      lastBranchName: null,
+    };
+    watchTargets.set(normalizedCwd, target);
+
+    const subscription = workspaceGitService.registerWorkspace(
+      { cwd: normalizedCwd },
+      (snapshot) => {
+        handleBranchSnapshot(normalizedCwd, snapshot.git.currentBranch ?? null);
+        void emitWorkspaceUpdateForCwd(normalizedCwd).catch((error) => {
+          logger.warn(
+            { err: error, cwd: normalizedCwd },
+            "Failed to emit workspace update after git branch snapshot",
+          );
+        });
+        emitStatusUpdate(normalizedCwd, snapshot);
+      },
+    );
+    subscriptions.set(normalizedCwd, subscription.unsubscribe);
+  }
+
+  function syncObservers(workspaces: Iterable<WorkspaceDescriptorPayload>): void {
+    for (const workspace of workspaces) {
+      syncObserver(workspace.workspaceDirectory, {
+        isGit: workspace.projectKind === "git",
+        workspaceId: workspace.id,
+      });
+      rememberDescriptorState(workspace.workspaceDirectory, workspace);
+    }
+  }
+
+  async function syncObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
+    const descriptor = await describeWorkspaceRecordWithGitData(workspace);
+    syncObservers([descriptor]);
+  }
+
+  return {
+    syncObservers,
+    syncObserverForWorkspace,
+
+    async warmGitData(workspace) {
+      await syncObserverForWorkspace(workspace);
+      await emitWorkspaceUpdateForWorkspaceId(workspace.workspaceId);
+    },
+
+    shouldSkipUpdate(workspaceId, workspace) {
+      const target = resolveTargetByWorkspaceId(workspaceId);
+      if (!target) {
+        return false;
+      }
+      const nextStateKey = descriptorStateKey(workspace);
+      if (target.latestDescriptorStateKey === nextStateKey) {
+        return true;
+      }
+      target.latestDescriptorStateKey = nextStateKey;
+      return false;
+    },
+
+    recordDescriptorState(workspaceId, nextWorkspace) {
+      const target = resolveTargetByWorkspaceId(workspaceId);
+      if (target && onBranchChanged) {
+        const newBranchName = nextWorkspace?.name ?? null;
+        if (newBranchName !== target.lastBranchName) {
+          onBranchChanged(workspaceId, target.lastBranchName, newBranchName);
+        }
+      }
+      rememberDescriptorState(workspaceId, nextWorkspace);
+    },
+
+    handleBranchSnapshot,
+
+    removeForWorkspaceId(workspaceId) {
+      const target = resolveTargetByWorkspaceId(workspaceId);
+      if (target) {
+        removeForCwd(target.cwd);
+      }
+    },
+
+    removeForCwd,
+
+    dispose() {
+      for (const unsubscribe of subscriptions.values()) {
+        unsubscribe();
+      }
+      subscriptions.clear();
+      watchTargets.clear();
+    },
+  };
+}