diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index b22cb0e1a..725a67733 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -92,10 +92,7 @@ function formatListenTarget(listenTarget: ListenTarget | null): string | null { export async function fanOutReconciledWorkspaceUpdates(input: { sessions: Iterable<{ syncWorkspaceGitObserversForExternalWorkspaceIds(workspaceIds: Iterable): Promise; - emitWorkspaceUpdatesForExternalWorkspaceIds( - workspaceIds: Iterable, - options: { skipReconcile: boolean }, - ): Promise; + emitWorkspaceUpdatesForExternalWorkspaceIds(workspaceIds: Iterable): Promise; }>; workspaceIds: readonly string[]; logger: Pick; @@ -111,9 +108,7 @@ export async function fanOutReconciledWorkspaceUpdates(input: { ); } try { - await session.emitWorkspaceUpdatesForExternalWorkspaceIds(input.workspaceIds, { - skipReconcile: true, - }); + await session.emitWorkspaceUpdatesForExternalWorkspaceIds(input.workspaceIds); } catch (error) { input.logger.warn({ err: error }, "Failed to emit workspace updates after reconciliation"); } @@ -842,12 +837,17 @@ export async function createPaseoDaemon( logger, }); logger.info({ elapsed: elapsed() }, "Workspace registries bootstrapped"); + const teardownArchivedWorkspaceRuntime = (workspaceId: string): void => { + scriptRuntimeStore.removeForWorkspace(workspaceId); + releaseWorkspaceServicePortPlan(workspaceId); + }; const workspaceReconciliation = new WorkspaceReconciliationService({ projectRegistry, workspaceRegistry, logger, workspaceGitService, onProjectUpdate: (update) => wsServer?.publishProjectUpdate(update), + onWorkspaceArchived: teardownArchivedWorkspaceRuntime, onWorkspacesChanged: async (workspaceIds) => { await fanOutReconciledWorkspaceUpdates({ sessions: wsServer?.listTrustedSessions() ?? [], @@ -873,8 +873,7 @@ export async function createPaseoDaemon( workspaceRegistry, }); if (!existingWorkspace || existingWorkspace.archivedAt) return; - scriptRuntimeStore.removeForWorkspace(workspaceId); - releaseWorkspaceServicePortPlan(workspaceId); + teardownArchivedWorkspaceRuntime(workspaceId); }; // external path→workspace adapter, not ownership: archive-by-path requests that // arrive with a worktree path and no workspaceId (old clients / CLI). diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 9f21c4833..c060450c9 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -162,7 +162,6 @@ import { archiveWorkspaceContents, requireActiveWorkspaceForArchive, } from "./workspace-archive-service.js"; -import { WorkspaceReconciliationService } from "./workspace-reconciliation-service.js"; import type { ServiceProxySubsystem } from "./service-proxy.js"; import { renameCurrentBranch as renameCurrentBranchDefault } from "../utils/checkout-git.js"; import { @@ -206,7 +205,6 @@ import type { ForgeService } from "../services/forge-service.js"; import type { ProviderUsageService } from "../services/quota-fetcher/service.js"; import { summarizeFetchWorkspacesEntries, - workspaceIdsForProjects, workspaceIdsOnCheckout, WorkspaceDirectory, type WorkspaceUpdatesFilter, @@ -1135,7 +1133,7 @@ export class Session { } async emitWorkspaceUpdateForWorkspaceId(workspaceId: string): Promise { - await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId], { skipReconcile: true }); + await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId]); } private async emitCreatedWorkspaceUpdate( @@ -1143,10 +1141,10 @@ export class Session { optimisticStatus?: WorkspaceDescriptorPayload["status"], ): Promise { if (this.workspaceUpdatesSubscription) { - await this.emitWorkspaceUpdatesForWorkspaceIds([workspace.id], { - skipReconcile: true, - ...(optimisticStatus ? { optimisticStatus } : {}), - }); + await this.emitWorkspaceUpdatesForWorkspaceIds( + [workspace.id], + optimisticStatus ? { optimisticStatus } : undefined, + ); return; } // COMPAT(workspaceCreateCausalUpdate): added in v0.1.106, remove after 2027-01-12. @@ -1171,11 +1169,8 @@ export class Session { this.clearWorkspaceArchiving(workspaceIds); } - async emitWorkspaceUpdatesForExternalWorkspaceIds( - workspaceIds: Iterable, - options?: { skipReconcile?: boolean }, - ): Promise { - await this.emitWorkspaceUpdatesForWorkspaceIds(workspaceIds, options); + async emitWorkspaceUpdatesForExternalWorkspaceIds(workspaceIds: Iterable): Promise { + await this.emitWorkspaceUpdatesForWorkspaceIds(workspaceIds); } async syncWorkspaceGitObserversForExternalWorkspaceIds( @@ -1407,10 +1402,10 @@ export class Session { if (this.isCleanedUp) { return; } - await this.emitWorkspaceUpdatesForWorkspaceIds([mutation.workspaceId], { - skipReconcile: true, - ...(mutation.expectsInitialAgent ? { optimisticStatus: "running" } : {}), - }); + await this.emitWorkspaceUpdatesForWorkspaceIds( + [mutation.workspaceId], + mutation.expectsInitialAgent ? { optimisticStatus: "running" } : undefined, + ); } catch (error) { this.sessionLogger.warn( { err: error, workspaceId: mutation.workspaceId, mutationKind: mutation.kind }, @@ -1480,7 +1475,6 @@ export class Session { this.workspaceGitObserver.removeForWorkspaceId(workspaceId); } await this.emitWorkspaceUpdatesForWorkspaceIds(updateIds, { - skipReconcile: true, removedProjectId: mutation.projectId, }); return; @@ -1491,9 +1485,7 @@ export class Session { this.workspaceGitObserver.removeForWorkspaceId(workspaceId); } } - await this.emitWorkspaceUpdatesForWorkspaceIds(projectWorkspaceIds, { - skipReconcile: true, - }); + await this.emitWorkspaceUpdatesForWorkspaceIds(projectWorkspaceIds); } catch (error) { this.sessionLogger.warn( { err: error, projectId: mutation.projectId, mutationKind: mutation.kind }, @@ -2412,9 +2404,7 @@ export class Session { } } - await this.emitWorkspaceUpdatesForWorkspaceIds(affectedWorkspaceIds, { - skipReconcile: true, - }); + await this.emitWorkspaceUpdatesForWorkspaceIds(affectedWorkspaceIds); this.emit({ type: "agent.detach.response", @@ -2619,9 +2609,7 @@ export class Session { .filter((workspace) => workspace.projectId === projectId) .map((workspace) => workspace.workspaceId); if (affectedWorkspaceIds.length > 0) { - await this.emitWorkspaceUpdatesForWorkspaceIds(affectedWorkspaceIds, { - skipReconcile: true, - }); + await this.emitWorkspaceUpdatesForWorkspaceIds(affectedWorkspaceIds); } } catch (error) { this.sessionLogger.error( @@ -2666,9 +2654,7 @@ export class Session { if (activeWorkspaceIds.length > 0) { this.markWorkspaceArchiving(activeWorkspaceIds, new Date().toISOString()); - await this.emitWorkspaceUpdatesForWorkspaceIds(activeWorkspaceIds, { - skipReconcile: true, - }); + await this.emitWorkspaceUpdatesForWorkspaceIds(activeWorkspaceIds); } const removedWorkspaceIds: string[] = []; @@ -2700,7 +2686,6 @@ export class Session { ? removedWorkspaceIds : [projectWorkspaces[0]?.workspaceId ?? projectId]; await this.emitWorkspaceUpdatesForWorkspaceIds(updateIds, { - skipReconcile: true, removedProjectId: projectId, }); @@ -2785,9 +2770,7 @@ export class Session { }, }); - await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId], { - skipReconcile: true, - }); + await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId]); } catch (error) { this.sessionLogger.error( { err: error, workspaceId, requestId }, @@ -2842,7 +2825,7 @@ export class Session { return; } emitResponse(true, nextPinnedAt, null); - await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId], { skipReconcile: true }); + await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId]); } catch (error) { this.sessionLogger.error( { ...logContext, err: error }, @@ -4573,67 +4556,9 @@ export class Session { releaseWorkspaceServicePortPlan(workspaceId); } - private async reconcileAndEmitWorkspaceUpdates(): Promise { - if (!this.workspaceUpdatesSubscription) { - return; - } - try { - const changedWorkspaceIds = await this.reconcileActiveWorkspaceRecords(); - if (changedWorkspaceIds.size === 0) { - return; - } - await this.emitWorkspaceUpdatesForWorkspaceIds(changedWorkspaceIds, { - skipReconcile: true, - }); - } catch (error) { - this.sessionLogger.error({ err: error }, "Background workspace reconciliation failed"); - } - } - - private async reconcileActiveWorkspaceRecords(): Promise> { - const service = new WorkspaceReconciliationService({ - projectRegistry: this.projectRegistry, - workspaceRegistry: this.workspaceRegistry, - logger: this.sessionLogger, - workspaceGitService: this.workspaceGitService, - }); - const result = await service.runOnce(); - const changedWorkspaceIds = new Set(); - const changedProjectIds = new Set(); - - await Promise.all( - result.changesApplied.map(async (change) => { - switch (change.kind) { - case "workspace_archived": - await this.teardownArchivedWorkspace(change.workspaceId); - changedWorkspaceIds.add(change.workspaceId); - break; - case "workspace_updated": - changedWorkspaceIds.add(change.workspaceId); - break; - case "project_updated": - changedProjectIds.add(change.projectId); - break; - } - }), - ); - - if (changedProjectIds.size > 0) { - for (const workspaceId of workspaceIdsForProjects( - await this.workspaceRegistry.list(), - changedProjectIds, - )) { - changedWorkspaceIds.add(workspaceId); - } - } - - return changedWorkspaceIds; - } - private async emitWorkspaceUpdatesForWorkspaceIds( workspaceIds: Iterable, options?: { - skipReconcile?: boolean; dedupeGitState?: boolean; removedProjectId?: string; optimisticStatus?: WorkspaceDescriptorPayload["status"]; @@ -4700,10 +4625,6 @@ export class Session { this.bufferOrEmitWorkspaceUpdate(subscription, nextPayload); } - - if (!options?.skipReconcile) { - void this.reconcileAndEmitWorkspaceUpdates(); - } } private applyOptimisticWorkspaceStatus( @@ -4764,9 +4685,7 @@ export class Session { if (!event.workspaceId) { return; } - await this.emitWorkspaceUpdatesForWorkspaceIds([event.workspaceId], { - skipReconcile: true, - }); + await this.emitWorkspaceUpdatesForWorkspaceIds([event.workspaceId]); } // A git fact (branch, diff, dirty, PR) changed at `cwd`. Every workspace whose @@ -4777,7 +4696,6 @@ export class Session { private async emitWorkspaceUpdateForCwd( cwd: string, options?: { - skipReconcile?: boolean; dedupeGitState?: boolean; }, ): Promise { @@ -4971,7 +4889,6 @@ export class Session { if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) { this.flushBootstrappedWorkspaceUpdates(snapshot); - void this.reconcileAndEmitWorkspaceUpdates(); } } catch (error) { if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) { diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index c7c0ed247..8a87e5064 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -10,6 +10,7 @@ import { } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; +import { setImmediate as waitForImmediate } from "node:timers/promises"; import { afterEach, expect, test, vi } from "vitest"; import { z } from "zod"; @@ -78,6 +79,10 @@ const UNREGISTERED_CWD = path.resolve("/tmp/unregistered"); const terminalManagers: TerminalManager[] = []; +async function flushWorkspaceUpdateBackgroundWork(): Promise { + await waitForImmediate(); +} + afterEach(async () => { while (terminalManagers.length > 0) { const manager = terminalManagers.pop(); @@ -130,14 +135,12 @@ interface SessionTestAccess { agentUpdates: AgentUpdatesService; workspaceUpdatesSubscription: unknown; interruptAgentIfRunning(agentId: string): unknown; - reconcileActiveWorkspaceRecords(...args: unknown[]): Promise>; reconcileWorkspaceRecord(workspaceId: string): Promise<{ changed: boolean; workspace?: Record | null; removedWorkspaceId?: string | null; [key: string]: unknown; }>; - reconcileAndEmitWorkspaceUpdates(...args: unknown[]): Promise; handleArchiveAgentRequest(agentId: string, requestId: string): Promise; handleMessage(message: unknown): Promise; handleCreatePaseoWorktreeRequest(params: unknown): Promise; @@ -155,10 +158,7 @@ interface SessionTestAccess { clearWorkspaceArchiving(workspaceIds: Iterable): void; emitWorkspaceUpdateForCwd(...args: unknown[]): Promise; emitWorkspaceUpdatesForWorkspaceIds(...args: unknown[]): Promise; - emitWorkspaceUpdatesForExternalWorkspaceIds( - workspaceIds: Iterable, - options?: { skipReconcile?: boolean }, - ): Promise; + emitWorkspaceUpdatesForExternalWorkspaceIds(workspaceIds: Iterable): Promise; updateClientCapabilities(capabilities: Record | null): void; emit(message: unknown): void; onMessage(message: unknown): void; @@ -1234,59 +1234,6 @@ test("unsupported persisted agents are excluded from active lists but preserved ); }); -test("workspace reconciliation reports archived workspaces to subscribed clients", async () => { - const missingCwd = path.join(tmpdir(), `paseo-missing-workspace-${Date.now()}`); - rmSync(missingCwd, { recursive: true, force: true }); - const projects = new Map([ - [ - "proj-missing", - createPersistedProjectRecord({ - projectId: "proj-missing", - rootPath: missingCwd, - kind: "non_git", - displayName: "missing", - createdAt: "2026-03-01T12:00:00.000Z", - updatedAt: "2026-03-01T12:00:00.000Z", - }), - ], - ]); - const workspaces = new Map([ - [ - "ws-missing", - createPersistedWorkspaceRecord({ - workspaceId: "ws-missing", - projectId: "proj-missing", - cwd: missingCwd, - kind: "directory", - displayName: "missing", - createdAt: "2026-03-01T12:00:00.000Z", - updatedAt: "2026-03-01T12:00:00.000Z", - }), - ], - ]); - const session = createSessionForWorkspaceTests(); - session.projectRegistry.list = async () => Array.from(projects.values()); - session.projectRegistry.archive = async (projectId: string, archivedAt: string) => { - const project = projects.get(projectId); - if (project) { - projects.set(projectId, { ...project, archivedAt }); - } - }; - session.workspaceRegistry.list = async () => Array.from(workspaces.values()); - session.workspaceRegistry.archive = async (workspaceId: string, archivedAt: string) => { - const workspace = workspaces.get(workspaceId); - if (workspace) { - workspaces.set(workspaceId, { ...workspace, archivedAt }); - } - }; - - const changedWorkspaceIds = await session.reconcileActiveWorkspaceRecords(); - - expect(changedWorkspaceIds).toEqual(new Set(["ws-missing"])); - expect(workspaces.get("ws-missing")?.archivedAt).toBeTruthy(); - expect(projects.get("proj-missing")?.archivedAt).toBeFalsy(); -}); - test("agent_update placement does not refresh git snapshots", async () => { const emitted: SessionOutboundMessage[] = []; const getSnapshot = vi.fn(async () => { @@ -3316,8 +3263,6 @@ test("workspace update stream keeps persisted workspace visible after agents sto pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); - session.buildWorkspaceDescriptorMap = async () => new Map([ [ @@ -3432,13 +3377,10 @@ test("archiving the last workspace emits a remove carrying the now-empty project ], ]), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); // The archived workspace no longer resolves to an active descriptor. session.buildWorkspaceDescriptorMap = async () => new Map(); - await session.emitWorkspaceUpdatesForWorkspaceIds([archivedWorkspace.workspaceId], { - skipReconcile: true, - }); + await session.emitWorkspaceUpdatesForWorkspaceIds([archivedWorkspace.workspaceId]); const removeUpdate = filterByType(emitted, "workspace_update").find( (message) => message.payload.kind === "remove", @@ -3503,7 +3445,6 @@ test("project.remove.request archives active workspaces and removes the project pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); session.listAgentPayloads = async () => []; session.buildWorkspaceDescriptorMap = async (options: { workspaceIds?: Iterable }) => { const workspaceIds = Array.from(options.workspaceIds ?? workspaces.keys()); @@ -3603,7 +3544,6 @@ test("project.remove.request removes an already-empty project", async () => { [archivedWorkspace.workspaceId, { kind: "remove", id: archivedWorkspace.workspaceId }], ]), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); session.listAgentPayloads = async () => []; session.buildWorkspaceDescriptorMap = async () => new Map(); @@ -3773,14 +3713,19 @@ test("create paseo worktree response preserves an explicit non-Git project", asy expect(projects.get(explicitProject.projectId)).toEqual(explicitProject); }); -test("workspace update fanout for multiple cwd values is deduplicated", async () => { +test("workspace updates stay scoped to the matching cwd", async () => { const emitted: SessionOutboundMessage[] = []; + const archivedWorkspaceIds: string[] = []; + const missingRoot = path.join(tmpdir(), `paseo-scoped-workspace-${Date.now()}`); + rmSync(missingRoot, { recursive: true, force: true }); + const mainCwd = path.join(missingRoot, "main"); + const featureCwd = path.join(missingRoot, "feature"); const session = createSessionForWorkspaceTests(); session.workspaceRegistry.list = async () => [ createPersistedWorkspaceRecord({ workspaceId: "ws-repo-main", projectId: "proj-repo-main", - cwd: REPO_CWD, + cwd: mainCwd, kind: "local_checkout", displayName: "main", createdAt: "2026-03-01T12:00:00.000Z", @@ -3789,13 +3734,16 @@ test("workspace update fanout for multiple cwd values is deduplicated", async () createPersistedWorkspaceRecord({ workspaceId: "ws-repo-feature", projectId: "proj-repo-main", - cwd: "/tmp/repo/worktree", + cwd: featureCwd, kind: "worktree", displayName: "feature", createdAt: "2026-03-01T12:00:00.000Z", updatedAt: "2026-03-01T12:00:00.000Z", }), ]; + session.workspaceRegistry.archive = async (workspaceId) => { + archivedWorkspaceIds.push(workspaceId); + }; session.workspaceUpdatesSubscription = { subscriptionId: "sub-dedup", filter: undefined, @@ -3803,31 +3751,15 @@ test("workspace update fanout for multiple cwd values is deduplicated", async () pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => - new Set(["ws-repo-main", "ws-repo-feature"]); session.buildWorkspaceDescriptorMap = async () => new Map([ - [ - "ws-repo-main", - { - id: "ws-repo-main", - projectId: "proj-repo-main", - projectDisplayName: "repo", - projectRootPath: REPO_CWD, - projectKind: "git", - workspaceKind: "local_checkout", - name: "main", - status: "done", - activityAt: null, - }, - ], [ "ws-repo-feature", { id: "ws-repo-feature", projectId: "proj-repo-main", projectDisplayName: "repo", - projectRootPath: REPO_CWD, + projectRootPath: mainCwd, projectKind: "git", workspaceKind: "worktree", name: "feature", @@ -3840,17 +3772,20 @@ test("workspace update fanout for multiple cwd values is deduplicated", async () if (isSessionOutboundMessage(message)) emitted.push(message); }; - await session.emitWorkspaceUpdateForCwd("/tmp/repo/worktree"); - await new Promise((resolve) => setTimeout(resolve, 0)); + await session.emitWorkspaceUpdateForCwd(featureCwd); + await flushWorkspaceUpdateBackgroundWork(); const workspaceUpdates = filterByType(emitted, "workspace_update"); - expect(workspaceUpdates).toHaveLength(2); - expect(workspaceUpdates.map((entry) => entry.payload.kind)).toEqual(["upsert", "upsert"]); - expect( - workspaceUpdates - .map((entry) => (entry.payload.kind === "upsert" ? entry.payload.workspace.id : null)) - .sort((a, b) => String(a).localeCompare(String(b))), - ).toEqual(["ws-repo-feature", "ws-repo-main"]); + expect(workspaceUpdates).toEqual([ + { + type: "workspace_update", + payload: { + kind: "upsert", + workspace: expect.objectContaining({ id: "ws-repo-feature" }), + }, + }, + ]); + expect(archivedWorkspaceIds).toEqual([]); }); test("open_project_request registers a workspace before any agent exists", async () => { @@ -4243,8 +4178,6 @@ test("open_project_request emits a workspace_update with githubRuntime once the pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); - await session.handleMessage({ type: "open_project_request", cwd, @@ -4892,7 +4825,6 @@ test("workspace recovery stays accepted when git observer warming fails", async pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); session.listAgentPayloads = async () => []; await session.handleMessage({ @@ -6458,19 +6390,14 @@ test("emitWorkspaceUpdatesForWorkspaceIds includes archiving state and dedupes u pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); session.listAgentPayloads = async () => []; session.projectRegistry.list = async () => [project]; session.workspaceRegistry.list = async () => [workspace]; session.markWorkspaceArchiving([workspace.workspaceId], archivingAt); - await session.emitWorkspaceUpdatesForWorkspaceIds([workspace.workspaceId], { - skipReconcile: true, - }); - await session.emitWorkspaceUpdatesForWorkspaceIds([workspace.workspaceId], { - skipReconcile: true, - }); + await session.emitWorkspaceUpdatesForWorkspaceIds([workspace.workspaceId]); + await session.emitWorkspaceUpdatesForWorkspaceIds([workspace.workspaceId]); expect(emitted).toEqual([ { @@ -6486,7 +6413,7 @@ test("emitWorkspaceUpdatesForWorkspaceIds includes archiving state and dedupes u ]); }); -test("external workspace updates emit one deduplicated batch without reconciling", async () => { +test("external workspace updates emit one deduplicated batch", async () => { const emitted: SessionOutboundMessage[] = []; const session = createSessionForWorkspaceTests(); const project = createPersistedProjectRecord({ @@ -6536,10 +6463,11 @@ test("external workspace updates emit one deduplicated batch without reconciling if (isSessionOutboundMessage(message)) emitted.push(message); }; - await session.emitWorkspaceUpdatesForExternalWorkspaceIds( - [main.workspaceId, feature.workspaceId, main.workspaceId], - { skipReconcile: true }, - ); + await session.emitWorkspaceUpdatesForExternalWorkspaceIds([ + main.workspaceId, + feature.workspaceId, + main.workspaceId, + ]); expect(filterByType(emitted, "workspace_update")).toEqual([ { @@ -6795,7 +6723,6 @@ test("workspace_update includes updated runtime fields", async () => { pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), }; - session.reconcileActiveWorkspaceRecords = async () => new Set(); session.listAgentPayloads = async () => []; session.projectRegistry.list = async () => [project]; session.workspaceRegistry.list = async () => [workspace]; @@ -6813,9 +6740,7 @@ test("workspace_update includes updated runtime fields", async () => { }, }); - await session.emitWorkspaceUpdateForCwd(REPO_CWD, { - skipReconcile: true, - }); + await session.emitWorkspaceUpdateForCwd(REPO_CWD); expect(peekSnapshotRuntimeUpdate).toHaveBeenCalledWith(REPO_CWD); expect(emitted).toContainEqual({ @@ -6914,7 +6839,6 @@ test("subscribed fetch_workspaces includes git enrichment in the initial snapsho session.listAgentPayloads = async () => []; session.projectRegistry.list = async () => [gitProject, directoryProject]; session.workspaceRegistry.list = async () => [gitWorkspace, directoryWorkspace]; - session.reconcileAndEmitWorkspaceUpdates = vi.fn(async () => {}); const describeWorkspaceRecordSubscribed = vi.fn( async (workspace: typeof gitWorkspace, project: unknown) => { if (workspace.workspaceId === gitWorkspace.workspaceId) { @@ -7426,15 +7350,13 @@ test("a workspace leaving a filtered subscription after bootstrap emits a remova session.listFetchWorkspacesEntries = async () => listing; session.buildWorkspaceDescriptorMap = async () => new Map(currentDescriptor ? [[currentDescriptor.id, currentDescriptor]] : []); - session.reconcileAndEmitWorkspaceUpdates = async () => undefined; - const bootstrap = session.handleMessage({ type: "fetch_workspaces_request", requestId: "req-buffered-filter", filter: { query: "repo" }, subscribe: { subscriptionId: "sub-buffered-filter" }, }); - await session.emitWorkspaceUpdatesForWorkspaceIds([descriptor.id], { skipReconcile: true }); + await session.emitWorkspaceUpdatesForWorkspaceIds([descriptor.id]); finishListing({ entries: [], emptyProjects: [], @@ -7448,7 +7370,7 @@ test("a workspace leaving a filtered subscription after bootstrap emits a remova emitted.length = 0; currentDescriptor = { ...descriptor, name: "other work" }; - await session.emitWorkspaceUpdatesForWorkspaceIds([descriptor.id], { skipReconcile: true }); + await session.emitWorkspaceUpdatesForWorkspaceIds([descriptor.id]); expect(filterByType(emitted, "workspace_update")).toEqual([ { diff --git a/packages/server/src/server/workspace-reconciliation-service.test.ts b/packages/server/src/server/workspace-reconciliation-service.test.ts index 4cdebd624..26559458b 100644 --- a/packages/server/src/server/workspace-reconciliation-service.test.ts +++ b/packages/server/src/server/workspace-reconciliation-service.test.ts @@ -474,6 +474,7 @@ describe("WorkspaceReconciliationService", () => { test("archives workspaces whose directories no longer exist", async () => { const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries(); + const archivedWorkspaceIds: string[] = []; projects.set( "p1", @@ -503,14 +504,23 @@ describe("WorkspaceReconciliationService", () => { projectRegistry, workspaceRegistry, logger: createTestLogger(), + onWorkspaceArchived: (workspaceId) => { + archivedWorkspaceIds.push(workspaceId); + }, }); const result = await service.runOnce(); - expect(result.changesApplied.length).toBeGreaterThanOrEqual(1); - const wsChange = result.changesApplied.find((c) => c.kind === "workspace_archived"); - expect(wsChange).toBeDefined(); - expect(workspaces.get("w1")!.archivedAt).toBeTruthy(); + expect(result.changesApplied).toEqual([ + { + kind: "workspace_archived", + workspaceId: "w1", + directory: "/tmp/does-not-exist-reconcile-test", + reason: "directory_missing", + }, + ]); + expect(archivedWorkspaceIds).toEqual(["w1"]); + expect(workspaces.get("w1")?.archivedAt).toEqual(expect.any(String)); }); test("keeps a project active after all its workspaces are archived", async () => { diff --git a/packages/server/src/server/workspace-reconciliation-service.ts b/packages/server/src/server/workspace-reconciliation-service.ts index 28ca9e219..fa1810358 100644 --- a/packages/server/src/server/workspace-reconciliation-service.ts +++ b/packages/server/src/server/workspace-reconciliation-service.ts @@ -87,6 +87,7 @@ export interface WorkspaceReconciliationServiceOptions { onChanges?: (changes: ReconciliationChange[]) => void; workspaceGitService?: Pick; onProjectUpdate?: (update: ProjectUpdate) => void; + onWorkspaceArchived?: (workspaceId: string) => void | Promise; onWorkspacesChanged?: (workspaceIds: string[]) => Promise; watchProjectRoot?: ProjectRootWatch; clock?: ReconciliationClock; @@ -116,6 +117,7 @@ export class WorkspaceReconciliationService { private readonly onChanges: ((changes: ReconciliationChange[]) => void) | null; private readonly workspaceGitService: Pick | null; private readonly onProjectUpdate: ((update: ProjectUpdate) => void) | null; + private readonly onWorkspaceArchived: ((workspaceId: string) => void | Promise) | null; private readonly onWorkspacesChanged: ((workspaceIds: string[]) => Promise) | null; private readonly watchProjectRoot: ProjectRootWatch; private readonly clock: ReconciliationClock; @@ -137,6 +139,7 @@ export class WorkspaceReconciliationService { this.onChanges = options.onChanges ?? null; this.workspaceGitService = options.workspaceGitService ?? null; this.onProjectUpdate = options.onProjectUpdate ?? null; + this.onWorkspaceArchived = options.onWorkspaceArchived ?? null; this.onWorkspacesChanged = options.onWorkspacesChanged ?? null; this.watchProjectRoot = options.watchProjectRoot ?? watchProjectRoot; this.clock = options.clock ?? systemClock; @@ -237,6 +240,7 @@ export class WorkspaceReconciliationService { missingWorkspaces.map(async (workspace) => { const timestamp = new Date().toISOString(); await this.workspaceRegistry.archive(workspace.workspaceId, timestamp); + await this.onWorkspaceArchived?.(workspace.workspaceId); changes.push({ kind: "workspace_archived", workspaceId: workspace.workspaceId,