diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 2c35ed14a..1f977f788 100644 --- a/packages/server/src/server/session.ts +++ b/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 | null; - refreshPromise: Promise | 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(); private readonly workspaceSetupSnapshots: Map; - private readonly workspaceGitFetchSubscriptions = new Map void>(); - private readonly workspaceGitSubscriptions = new Map 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 { - const descriptor = await this.describeWorkspaceRecordWithGitData(workspace); - this.syncWorkspaceGitObservers([descriptor]); + await this.workspaceGitObserver.syncObserverForWorkspace(workspace); } async emitWorkspaceUpdateForWorkspaceId(workspaceId: string): Promise { @@ -950,8 +942,7 @@ export class Session { } async warmWorkspaceGitDataForWorkspace(workspace: PersistedWorkspaceRecord): Promise { - 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 { - 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): 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, ): Promise { @@ -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 { - await this.removeWorkspaceGitWatchTarget(input.cwd); + this.workspaceGitObserver.removeForCwd(input.cwd); this.scriptRuntimeStore?.removeForWorkspace(input.workspaceId); - this.removeWorkspaceGitSubscription(input.cwd); } private async reconcileAndEmitWorkspaceUpdates(): Promise { @@ -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(); } } diff --git a/packages/server/src/server/session.workspace-git-watch.test.ts b/packages/server/src/server/session.workspace-git-watch.test.ts index 1690a42b1..38a590476 100644 --- a/packages/server/src/server/session.workspace-git-watch.test.ts +++ b/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; }; buildWorkspaceDescriptorMap: () => Promise>; - syncWorkspaceGitObserver(cwd: string, details: { isGit: boolean; workspaceId: string }): void; + workspaceGitObserver: WorkspaceGitObserverService; listAgentPayloads: () => Promise; } +// 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(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( diff --git a/packages/server/src/server/session/workspace-git-observer/workspace-git-observer-service.test.ts b/packages/server/src/server/session/workspace-git-observer/workspace-git-observer-service.test.ts new file mode 100644 index 000000000..08d37068e --- /dev/null +++ b/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 { + return new Promise((done) => setImmediate(done)); +} + +function buildHarness(opts: { emitCwdRejects?: boolean } = {}) { + const listeners = new Map(); + 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 = { + 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"]); + }); +}); diff --git a/packages/server/src/server/session/workspace-git-observer/workspace-git-observer-service.ts b/packages/server/src/server/session/workspace-git-observer/workspace-git-observer-service.ts new file mode 100644 index 000000000..75aac14de --- /dev/null +++ b/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): void; + syncObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise; + warmGitData(workspace: PersistedWorkspaceRecord): Promise; + // 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; + describeWorkspaceRecordWithGitData: ( + workspace: PersistedWorkspaceRecord, + ) => Promise; + emitWorkspaceUpdateForCwd: (cwd: string) => Promise; + emitWorkspaceUpdateForWorkspaceId: (workspaceId: string) => Promise; + 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(); + const subscriptions = new Map 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): 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 { + 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(); + }, + }; +}