fix(server): stop workspace updates triggering full scans (#2379)

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