From 1ffa2f821a14c073867c430fd3d406c82294e1a4 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 23 Jul 2026 19:16:28 +0200 Subject: [PATCH] Stop archived workspaces from running background Git checks (#2355) * fix(workspaces): stop archived workspace subscriptions Workspace mutations were persisted globally but projected only through the initiating session, leaving other sessions' Git observers alive after archive. Publish mutations from the shared registry so every session releases its local observer and receives exactly one update. * fix(workspaces): serialize workspace mutation observers * fix(workspaces): finish mutation cleanup Queued workspace mutations now stop at the session cleanup boundary and remove any observer created across an asynchronous cleanup race. Project removal can still enrich an earlier removal delta for legacy clients. * fix(workspaces): honor global mutation boundaries Serialize project and workspace lifecycle notifications per session, keep filtered subscriptions from owning unrelated Git observers, and broadcast final project removals to every session that saw the project. Update strict Session harnesses for the registry subscriptions. * fix(workspaces): preserve initial agent status Carry initial-agent intent through global workspace mutations and publish the optimistic running status until the agent contribution takes over. This removes the invalid transient Done state exposed by global subscriptions. --- packages/server/src/server/bootstrap.ts | 14 +- .../src/server/paseo-worktree-service.ts | 1 + packages/server/src/server/session.ts | 236 ++++++++- .../session.workspace-git-watch.test.ts | 2 + ...on.workspace-resolution-invariants.test.ts | 2 + .../src/server/session.workspaces.test.ts | 501 +++++++++++++++++- .../workspace-provisioning-service.ts | 15 +- .../snapshot-mutation-ownership.test.ts | 2 + ...orkspace-archive-record-scoped.e2e.test.ts | 89 +++- .../server/src/server/workspace-registry.ts | 90 +++- 10 files changed, 910 insertions(+), 42 deletions(-) diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 4d3d48929..b22cb0e1a 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -178,6 +178,7 @@ import type { } from "./agent/provider-launch-config.js"; import type { PersistedConfig } from "./persisted-config.js"; import { createServiceProxySubsystem, type ServiceProxySubsystem } from "./service-proxy.js"; +import { releaseWorkspaceServicePortPlan } from "./workspace-service-port-registry.js"; import { ScriptHealthMonitor } from "./script-health-monitor.js"; import { createScriptStatusEmitter } from "./script-status-projection.js"; import { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js"; @@ -867,18 +868,13 @@ export async function createPaseoDaemon( workspaceGitService, }); const archiveWorkspaceRecordExternal = async (workspaceId: string) => { - const sessions = wsServer?.listTrustedSessions() ?? []; - if (sessions.length > 0) { - await Promise.all( - sessions.map((session) => session.archiveWorkspaceRecordForExternalMutation(workspaceId)), - ); - return; - } - - await archivePersistedWorkspaceRecord({ + const existingWorkspace = await archivePersistedWorkspaceRecord({ workspaceId, workspaceRegistry, }); + if (!existingWorkspace || existingWorkspace.archivedAt) return; + scriptRuntimeStore.removeForWorkspace(workspaceId); + releaseWorkspaceServicePortPlan(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/paseo-worktree-service.ts b/packages/server/src/server/paseo-worktree-service.ts index 818d98753..f09a81338 100644 --- a/packages/server/src/server/paseo-worktree-service.ts +++ b/packages/server/src/server/paseo-worktree-service.ts @@ -91,6 +91,7 @@ export async function createPaseoWorktree( branch: createdWorktree.worktree.branchName || null, baseBranch: resolveIntentBaseBranch(createdWorktree.intent), title: input.title?.trim() || resolveFirstAgentPromptTitle(input.firstAgentContext), + expectsInitialAgent: Boolean(input.firstAgentContext), }); deps.github.invalidate({ cwd: createdWorktree.worktree.worktreePath }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 2192ad69b..1c8de87a0 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -128,7 +128,9 @@ import { resolveWorkspaceName, type PersistedProjectRecord, type PersistedWorkspaceRecord, + type ProjectMutation, type ProjectRegistry, + type WorkspaceMutation, type WorkspaceRegistry, } from "./workspace-registry.js"; import { wrapSpokenInput } from "./voice-config.js"; @@ -364,6 +366,7 @@ interface WorkspaceUpdatesSubscriptionState { isBootstrapping: boolean; pendingUpdatesByWorkspaceId: Map; lastEmittedByWorkspaceId: Map; + visibleEmptyProjectIds?: Set; } class SessionRequestError extends Error { @@ -585,6 +588,10 @@ export class Session { private readonly daemonConfigStore: DaemonConfigStore; private readonly pushTokenStore: PushTokenStore; private unsubscribeAgentEvents: (() => void) | null = null; + private unsubscribeProjectMutations: (() => void) | null = null; + private unsubscribeWorkspaceMutations: (() => void) | null = null; + private registryMutationQueue: Promise = Promise.resolve(); + private isCleanedUp = false; private viewedTimelineAgentIds = new Set(); private readonly viewedTimelineAgentIdsBySource = new Map>(); private readonly clientCapabilitiesBySource = new Map>(); @@ -981,6 +988,7 @@ export class Session { }); this.subscribeToAgentEvents(); + this.subscribeToRegistryMutations(); this.sessionLogger.trace({}, "agent.session.lifecycle.created"); } @@ -1127,18 +1135,26 @@ export class Session { await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId], { skipReconcile: true }); } - private async emitCreatedWorkspaceUpdate(workspace: WorkspaceDescriptorPayload): Promise { + private async emitCreatedWorkspaceUpdate( + workspace: WorkspaceDescriptorPayload, + optimisticStatus?: WorkspaceDescriptorPayload["status"], + ): Promise { if (this.workspaceUpdatesSubscription) { - await this.emitWorkspaceUpdateForWorkspaceId(workspace.id); + await this.emitWorkspaceUpdatesForWorkspaceIds([workspace.id], { + skipReconcile: true, + ...(optimisticStatus ? { optimisticStatus } : {}), + }); return; } // COMPAT(workspaceCreateCausalUpdate): added in v0.1.106, remove after 2027-01-12. // Older clients create before subscribing and require the causal update beside the response. - this.emit({ type: "workspace_update", payload: { kind: "upsert", workspace } }); - } - - async archiveWorkspaceRecordForExternalMutation(workspaceId: string): Promise { - await this.archiveWorkspaceRecord(workspaceId); + this.emit({ + type: "workspace_update", + payload: { + kind: "upsert", + workspace: optimisticStatus ? { ...workspace, status: optimisticStatus } : workspace, + }, + }); } markWorkspaceArchivingForExternalMutation( @@ -1348,6 +1364,137 @@ export class Session { this.providerCatalogSession.start(); } + private subscribeToRegistryMutations(): void { + this.unsubscribeProjectMutations?.(); + this.unsubscribeProjectMutations = + this.projectRegistry.subscribeToMutations?.((mutation) => + this.enqueueRegistryMutation(() => this.handleProjectMutation(mutation)), + ) ?? null; + this.unsubscribeWorkspaceMutations?.(); + this.unsubscribeWorkspaceMutations = + this.workspaceRegistry.subscribeToMutations?.((mutation) => + this.enqueueRegistryMutation(() => this.handleWorkspaceMutation(mutation)), + ) ?? null; + } + + private enqueueRegistryMutation(handleMutation: () => Promise): Promise { + const next = this.registryMutationQueue.then(handleMutation); + this.registryMutationQueue = next.catch(() => {}); + return next; + } + + private async handleWorkspaceMutation(mutation: WorkspaceMutation): Promise { + try { + if (this.isCleanedUp) { + return; + } + if ( + mutation.kind === "archive" || + mutation.kind === "remove" || + mutation.workspace?.archivedAt + ) { + this.workspaceGitObserver.removeForWorkspaceId(mutation.workspaceId); + } else { + await this.syncWorkspaceMutationObserver(mutation); + } + if (this.isCleanedUp) { + return; + } + await this.emitWorkspaceUpdatesForWorkspaceIds([mutation.workspaceId], { + skipReconcile: true, + ...(mutation.expectsInitialAgent ? { optimisticStatus: "running" } : {}), + }); + } catch (error) { + this.sessionLogger.warn( + { err: error, workspaceId: mutation.workspaceId, mutationKind: mutation.kind }, + "Failed to apply workspace mutation to session", + ); + } + } + + private async syncWorkspaceMutationObserver(mutation: WorkspaceMutation): Promise { + const subscription = this.workspaceUpdatesSubscription; + if (!mutation.workspace || !subscription) { + return; + } + const descriptorsByWorkspaceId = await this.buildWorkspaceDescriptorMap({ + workspaceIds: [mutation.workspaceId], + includeGitData: false, + }); + const descriptor = descriptorsByWorkspaceId.get(mutation.workspaceId); + if ( + !descriptor || + !this.matchesWorkspaceFilter({ workspace: descriptor, filter: subscription.filter }) + ) { + this.workspaceGitObserver.removeForWorkspaceId(mutation.workspaceId); + return; + } + const currentWorkspace = await this.workspaceRegistry.get(mutation.workspaceId); + if (!currentWorkspace || currentWorkspace.archivedAt) { + this.workspaceGitObserver.removeForWorkspaceId(mutation.workspaceId); + return; + } + await this.workspaceGitObserver.syncObserverForWorkspace(currentWorkspace); + if (this.isCleanedUp) { + this.workspaceGitObserver.removeForWorkspaceId(mutation.workspaceId); + } + } + + private async handleProjectMutation(mutation: ProjectMutation): Promise { + try { + const subscription = this.workspaceUpdatesSubscription; + if (this.isCleanedUp || !subscription) { + return; + } + const projectWorkspaceIds = (await this.workspaceRegistry.list()) + .filter((workspace) => workspace.projectId === mutation.projectId) + .map((workspace) => workspace.workspaceId); + + if (mutation.kind === "remove") { + const visibleWorkspaceIds = projectWorkspaceIds.filter((workspaceId) => { + const lastEmitted = subscription.lastEmittedByWorkspaceId.get(workspaceId); + return ( + lastEmitted?.kind === "upsert" || + (lastEmitted?.kind === "remove" && + lastEmitted.emptyProject?.projectId === mutation.projectId) + ); + }); + let updateIds = visibleWorkspaceIds; + if ( + updateIds.length === 0 && + subscription.visibleEmptyProjectIds?.has(mutation.projectId) + ) { + updateIds = [mutation.projectId]; + } + if (updateIds.length === 0) { + return; + } + for (const workspaceId of projectWorkspaceIds) { + this.workspaceGitObserver.removeForWorkspaceId(workspaceId); + } + await this.emitWorkspaceUpdatesForWorkspaceIds(updateIds, { + skipReconcile: true, + removedProjectId: mutation.projectId, + }); + return; + } + + if (mutation.kind === "archive" || mutation.project?.archivedAt) { + for (const workspaceId of projectWorkspaceIds) { + this.workspaceGitObserver.removeForWorkspaceId(workspaceId); + } + } + await this.emitWorkspaceUpdatesForWorkspaceIds(projectWorkspaceIds, { + skipReconcile: true, + }); + } catch (error) { + this.sessionLogger.warn( + { err: error, projectId: mutation.projectId, mutationKind: mutation.kind }, + "Failed to apply project mutation to session", + ); + } + } + private subscribeToAgentEvents(): void { if (this.unsubscribeAgentEvents) { this.unsubscribeAgentEvents(); @@ -4222,6 +4369,16 @@ export class Session { subscription: WorkspaceUpdatesSubscriptionState, payload: WorkspaceUpdatePayload, ): void { + if (payload.kind === "upsert") { + subscription.visibleEmptyProjectIds?.delete(payload.workspace.projectId); + } else { + if (payload.emptyProject) { + subscription.visibleEmptyProjectIds?.add(payload.emptyProject.projectId); + } + if (payload.removedProjectId) { + subscription.visibleEmptyProjectIds?.delete(payload.removedProjectId); + } + } if (subscription.isBootstrapping) { const workspaceId = payload.kind === "upsert" ? payload.workspace.id : payload.id; subscription.pendingUpdatesByWorkspaceId.set(workspaceId, payload); @@ -4468,7 +4625,12 @@ export class Session { private async emitWorkspaceUpdatesForWorkspaceIds( workspaceIds: Iterable, - options?: { skipReconcile?: boolean; dedupeGitState?: boolean; removedProjectId?: string }, + options?: { + skipReconcile?: boolean; + dedupeGitState?: boolean; + removedProjectId?: string; + optimisticStatus?: WorkspaceDescriptorPayload["status"]; + }, ): Promise { const subscription = this.workspaceUpdatesSubscription; if (!subscription) { @@ -4487,10 +4649,15 @@ export class Session { for (const workspaceId of uniqueWorkspaceIds) { const workspace = descriptorsByWorkspaceId.get(workspaceId); - const nextWorkspace = + const filteredWorkspace = workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter }) ? workspace : null; + const nextWorkspace = this.applyOptimisticWorkspaceStatus( + filteredWorkspace, + options?.optimisticStatus, + ); + const lastEmitted = subscription.lastEmittedByWorkspaceId.get(workspaceId); if ( options?.dedupeGitState && this.workspaceGitObserver.shouldSkipUpdate(workspaceId, nextWorkspace) @@ -4500,7 +4667,7 @@ export class Session { this.workspaceGitObserver.recordDescriptorState(workspaceId, nextWorkspace); if (!nextWorkspace) { - if (workspace && !subscription.lastEmittedByWorkspaceId.has(workspaceId)) { + if (this.shouldSkipWorkspaceRemoval(lastEmitted, options?.removedProjectId)) { continue; } subscription.lastEmittedByWorkspaceId.delete(workspaceId); @@ -4516,7 +4683,6 @@ export class Session { workspace: nextWorkspace, }; - const lastEmitted = subscription.lastEmittedByWorkspaceId.get(workspaceId); if ( lastEmitted && lastEmitted.kind === "upsert" && @@ -4533,6 +4699,26 @@ export class Session { } } + private applyOptimisticWorkspaceStatus( + workspace: WorkspaceDescriptorPayload | null, + optimisticStatus: WorkspaceDescriptorPayload["status"] | undefined, + ): WorkspaceDescriptorPayload | null { + if (!workspace || !optimisticStatus) { + return workspace; + } + return { ...workspace, status: optimisticStatus }; + } + + private shouldSkipWorkspaceRemoval( + lastEmitted: WorkspaceUpdatePayload | undefined, + removedProjectId: string | undefined, + ): boolean { + if (lastEmitted?.kind === "remove") { + return !removedProjectId || lastEmitted.removedProjectId === removedProjectId; + } + return !lastEmitted && !removedProjectId; + } + private async buildWorkspaceRemoveUpdatePayload( workspaceId: string, removedProjectId?: string, @@ -4744,6 +4930,7 @@ export class Session { isBootstrapping: true, pendingUpdatesByWorkspaceId: new Map(), lastEmittedByWorkspaceId: new Map(), + visibleEmptyProjectIds: new Set(), }; } @@ -4759,7 +4946,12 @@ export class Session { "fetch_workspaces_response_ready", ); const snapshot = this.buildBootstrapSnapshot(payload.entries); - this.seedWorkspaceSubscriptionSnapshot(subscriptionId, request.filter, payload.entries); + this.seedWorkspaceSubscriptionSnapshot( + subscriptionId, + request.filter, + payload.entries, + payload.emptyProjects, + ); this.emit({ type: "fetch_workspaces_response", @@ -4824,6 +5016,7 @@ export class Session { subscriptionId: string | null, filter: FetchWorkspacesRequestFilter | undefined, entries: FetchWorkspacesResponseEntry[], + emptyProjects: WorkspaceProjectDescriptorPayload[], ): void { const subscription = this.workspaceUpdatesSubscription; if (!subscription) return; @@ -4835,6 +5028,9 @@ export class Session { workspace: entry, }); } + for (const project of emptyProjects) { + subscription.visibleEmptyProjectIds?.add(project.projectId); + } } private async registerWorkspaceForImportedAgent( @@ -4910,6 +5106,7 @@ export class Session { cwd, explicitTitle ?? promptTitle, request.source.projectId, + { expectsInitialAgent: Boolean(request.firstAgentContext) }, ); await this.syncWorkspaceGitObserverForWorkspace(workspace); const descriptor = await this.describeWorkspaceRecord(workspace); @@ -4922,7 +5119,10 @@ export class Session { error: null, }, }); - await this.emitCreatedWorkspaceUpdate(descriptor); + await this.emitCreatedWorkspaceUpdate( + descriptor, + request.firstAgentContext ? "running" : undefined, + ); void this.workspaceGitService .getSnapshot(workspace.cwd, { force: true, includeForge: true, reason: "open_project" }) .catch((error) => { @@ -4997,7 +5197,10 @@ export class Session { error: null, }, }); - await this.emitCreatedWorkspaceUpdate(descriptor); + await this.emitCreatedWorkspaceUpdate( + descriptor, + request.firstAgentContext ? "running" : undefined, + ); } private async handleOpenProjectRequest( @@ -6310,11 +6513,16 @@ export class Session { */ public async cleanup(): Promise { this.sessionLogger.trace({}, "agent.session.lifecycle.cleanup"); + this.isCleanedUp = true; if (this.unsubscribeAgentEvents) { this.unsubscribeAgentEvents(); this.unsubscribeAgentEvents = null; } + this.unsubscribeProjectMutations?.(); + this.unsubscribeProjectMutations = null; + this.unsubscribeWorkspaceMutations?.(); + this.unsubscribeWorkspaceMutations = null; this.agentUpdates.dispose(); await this.hubExecutionController?.cleanup(); if (this.unsubscribeTerminalWorkspaceContributionEvents) { 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 06daa4726..00f3f780f 100644 --- a/packages/server/src/server/session.workspace-git-watch.test.ts +++ b/packages/server/src/server/session.workspace-git-watch.test.ts @@ -198,6 +198,7 @@ function createSessionForWorkspaceGitWatchTests(options?: { get: async () => null, }), projectRegistry: createStub({ + subscribeToMutations: () => () => {}, initialize: async () => {}, existsOnDisk: async () => true, list: async () => Array.from(projects.values()), @@ -215,6 +216,7 @@ function createSessionForWorkspaceGitWatchTests(options?: { }, }), workspaceRegistry: createStub({ + subscribeToMutations: () => () => {}, initialize: async () => {}, existsOnDisk: async () => true, list: async () => Array.from(workspaces.values()), diff --git a/packages/server/src/server/session.workspace-resolution-invariants.test.ts b/packages/server/src/server/session.workspace-resolution-invariants.test.ts index 277320e54..fdc3a8aa4 100644 --- a/packages/server/src/server/session.workspace-resolution-invariants.test.ts +++ b/packages/server/src/server/session.workspace-resolution-invariants.test.ts @@ -109,6 +109,7 @@ function createHarness(input: { get: async () => null, }), projectRegistry: createStub({ + subscribeToMutations: () => () => {}, initialize: async () => {}, existsOnDisk: async () => true, list: async () => Array.from(projects.values()), @@ -141,6 +142,7 @@ function createHarness(input: { }, }), workspaceRegistry: createStub({ + subscribeToMutations: () => () => {}, initialize: async () => {}, existsOnDisk: async () => true, list: async () => Array.from(workspaces.values()), diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 2b38acba4..c7c0ed247 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -70,6 +70,7 @@ import { createPersistedWorkspaceRecord, type PersistedProjectRecord, type PersistedWorkspaceRecord, + type WorkspaceMutation, } from "./workspace-registry.js"; const REPO_CWD = path.resolve("/tmp/repo"); @@ -3409,7 +3410,27 @@ test("archiving the last workspace emits a remove carrying the now-empty project filter: undefined, isBootstrapping: false, pendingUpdatesByWorkspaceId: new Map(), - lastEmittedByWorkspaceId: new Map(), + lastEmittedByWorkspaceId: new Map([ + [ + archivedWorkspace.workspaceId, + { + kind: "upsert", + workspace: { + id: archivedWorkspace.workspaceId, + projectId: project.projectId, + projectDisplayName: project.displayName, + projectRootPath: project.rootPath, + workspaceDirectory: archivedWorkspace.cwd, + projectKind: project.kind, + workspaceKind: archivedWorkspace.kind, + name: archivedWorkspace.displayName, + status: "done", + activityAt: null, + diffStat: null, + }, + }, + ], + ]), }; session.reconcileActiveWorkspaceRecords = async () => new Set(); // The archived workspace no longer resolves to an active descriptor. @@ -3578,7 +3599,9 @@ test("project.remove.request removes an already-empty project", async () => { filter: undefined, isBootstrapping: false, pendingUpdatesByWorkspaceId: new Map(), - lastEmittedByWorkspaceId: new Map(), + lastEmittedByWorkspaceId: new Map([ + [archivedWorkspace.workspaceId, { kind: "remove", id: archivedWorkspace.workspaceId }], + ]), }; session.reconcileActiveWorkspaceRecords = async () => new Set(); session.listAgentPayloads = async () => []; @@ -6927,6 +6950,456 @@ test("subscribed fetch_workspaces includes git enrichment in the initial snapsho ); }); +test("workspace mutation handling does not let a delayed upsert recreate an archived observer", async () => { + let mutationListener: ((mutation: WorkspaceMutation) => void | Promise) | null = null; + const registerCalls: string[] = []; + const unsubscribeCalls: string[] = []; + const project = createPersistedProjectRecord({ + projectId: "proj-race", + rootPath: REPO_CWD, + kind: "git", + displayName: "repo", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const workspace = createPersistedWorkspaceRecord({ + workspaceId: "ws-race", + projectId: project.projectId, + cwd: REPO_CWD, + kind: "local_checkout", + displayName: "main", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const descriptor = { + id: workspace.workspaceId, + projectId: project.projectId, + projectDisplayName: project.displayName, + projectRootPath: project.rootPath, + workspaceDirectory: workspace.cwd, + projectKind: project.kind, + workspaceKind: workspace.kind, + name: workspace.displayName, + status: "done", + activityAt: null, + diffStat: null, + } as WorkspaceDescriptorPayload; + let listedWorkspaces: PersistedWorkspaceRecord[] = []; + const workspaceRegistry: SessionOptions["workspaceRegistry"] = { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => listedWorkspaces, + get: async (workspaceId: string) => (workspaceId === workspace.workspaceId ? workspace : null), + update: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + subscribeToMutations: (listener) => { + mutationListener = listener; + return () => { + mutationListener = null; + }; + }, + }; + const session = createSessionForWorkspaceTests({ + projectRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [project], + get: async (projectId: string) => (projectId === project.projectId ? project : null), + getOrCreateActiveByRoot: async () => project, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + }, + workspaceRegistry, + workspaceGitService: createNoopWorkspaceGitService({ + registerWorkspace: ({ cwd }) => { + registerCalls.push(path.resolve(cwd)); + return { + unsubscribe: () => { + unsubscribeCalls.push(path.resolve(cwd)); + }, + }; + }, + }), + }); + session.emitWorkspaceUpdatesForWorkspaceIds = async () => {}; + + await session.handleMessage({ + type: "fetch_workspaces_request", + requestId: "req-fetch-workspaces-subscribe", + subscribe: {}, + }); + listedWorkspaces = [workspace]; + + let resumeDescribe!: () => void; + const describeStarted = new Promise((resolveStarted) => { + session.describeWorkspaceRecordWithGitData = async () => { + resolveStarted(); + await new Promise((resolveResume) => { + resumeDescribe = resolveResume; + }); + return descriptor; + }; + }); + + const upsertMutation = mutationListener?.({ + kind: "upsert", + workspaceId: workspace.workspaceId, + workspace, + }); + expect(upsertMutation).toBeDefined(); + const upsertPromise = Promise.resolve(upsertMutation); + await describeStarted; + + const archivedWorkspace = { ...workspace, archivedAt: "2026-03-02T12:00:00.000Z" }; + const archivePromise = Promise.resolve( + mutationListener?.({ + kind: "archive", + workspaceId: workspace.workspaceId, + workspace: archivedWorkspace, + }), + ); + + await Promise.resolve(); + expect(unsubscribeCalls).toEqual([]); + + resumeDescribe(); + await upsertPromise; + await archivePromise; + + expect(registerCalls).toEqual([REPO_CWD]); + expect(unsubscribeCalls).toEqual([REPO_CWD]); +}); + +test("workspace mutations outside a filtered subscription neither watch nor emit", async () => { + const emitted: SessionOutboundMessage[] = []; + let mutationListener: ((mutation: WorkspaceMutation) => void | Promise) | null = null; + const registerCalls: string[] = []; + const project = createPersistedProjectRecord({ + projectId: "proj-filtered-mutation", + rootPath: REPO_CWD, + kind: "git", + displayName: "repo", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const workspace = createPersistedWorkspaceRecord({ + workspaceId: "ws-filtered-mutation", + projectId: project.projectId, + cwd: REPO_CWD, + kind: "local_checkout", + displayName: "main", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const descriptor = { + id: workspace.workspaceId, + projectId: project.projectId, + projectDisplayName: project.displayName, + projectRootPath: project.rootPath, + workspaceDirectory: workspace.cwd, + projectKind: project.kind, + workspaceKind: workspace.kind, + name: workspace.displayName, + status: "done", + activityAt: null, + diffStat: null, + } as WorkspaceDescriptorPayload; + const workspaceRegistry: SessionOptions["workspaceRegistry"] = { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [workspace], + get: async (workspaceId: string) => (workspaceId === workspace.workspaceId ? workspace : null), + update: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + subscribeToMutations: (listener) => { + mutationListener = listener; + return () => { + mutationListener = null; + }; + }, + }; + const session = createSessionForWorkspaceTests({ + onMessage: (message) => emitted.push(message), + projectRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [project], + get: async (projectId: string) => (projectId === project.projectId ? project : null), + getOrCreateActiveByRoot: async () => project, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + }, + workspaceRegistry, + workspaceGitService: createNoopWorkspaceGitService({ + registerWorkspace: ({ cwd }) => { + registerCalls.push(path.resolve(cwd)); + return { unsubscribe: () => {} }; + }, + }), + }); + session.listFetchWorkspacesEntries = async () => ({ + entries: [], + emptyProjects: [], + pageInfo: { nextCursor: null, prevCursor: null, hasMore: false }, + }); + session.buildWorkspaceDescriptorMap = async () => new Map([[descriptor.id, descriptor]]); + + await session.handleMessage({ + type: "fetch_workspaces_request", + requestId: "req-filtered-mutation", + filter: { projectId: "some-other-project" }, + subscribe: {}, + }); + emitted.length = 0; + + await mutationListener?.({ + kind: "upsert", + workspaceId: workspace.workspaceId, + workspace, + }); + session.buildWorkspaceDescriptorMap = async () => new Map(); + await mutationListener?.({ + kind: "archive", + workspaceId: workspace.workspaceId, + workspace: { ...workspace, archivedAt: "2026-03-02T12:00:00.000Z" }, + }); + + expect(registerCalls).toEqual([]); + expect(filterByType(emitted, "workspace_update")).toEqual([]); +}); + +test("project removal mutation broadcasts the final delta to another subscribed session", async () => { + const emitted: SessionOutboundMessage[] = []; + const project = createPersistedProjectRecord({ + projectId: "proj-global-remove", + rootPath: REPO_CWD, + kind: "git", + displayName: "repo", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const workspace = createPersistedWorkspaceRecord({ + workspaceId: "ws-global-remove", + projectId: project.projectId, + cwd: REPO_CWD, + kind: "local_checkout", + displayName: "main", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-02T12:00:00.000Z", + archivedAt: "2026-03-02T12:00:00.000Z", + }); + let projectMutationListener: + | Parameters>[0] + | null = null; + const projectRegistry: SessionOptions["projectRegistry"] = { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [], + get: async () => null, + getOrCreateActiveByRoot: async () => project, + upsert: async () => {}, + archive: async () => {}, + remove: async (projectId) => { + await projectMutationListener?.({ kind: "remove", projectId, project: null }); + }, + subscribeToMutations: (listener) => { + projectMutationListener = listener; + return () => { + projectMutationListener = null; + }; + }, + }; + const session = createSessionForWorkspaceTests({ + onMessage: (message) => emitted.push(message), + projectRegistry, + workspaceRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [workspace], + get: async (workspaceId) => (workspaceId === workspace.workspaceId ? workspace : null), + update: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + }, + }); + session.workspaceUpdatesSubscription = { + subscriptionId: "sub-global-remove", + filter: undefined, + isBootstrapping: false, + pendingUpdatesByWorkspaceId: new Map(), + lastEmittedByWorkspaceId: new Map([ + [ + workspace.workspaceId, + { + kind: "remove", + id: workspace.workspaceId, + emptyProject: { + projectId: project.projectId, + projectDisplayName: project.displayName, + projectCustomName: null, + projectRootPath: project.rootPath, + projectKind: project.kind, + }, + }, + ], + ]), + }; + + await projectRegistry.remove(project.projectId); + + expect(filterByType(emitted, "workspace_update")).toEqual([ + { + type: "workspace_update", + payload: { + kind: "remove", + id: workspace.workspaceId, + removedProjectId: project.projectId, + }, + }, + ]); +}); + +test("workspace mutation handling drops queued observer sync after session cleanup", async () => { + let mutationListener: ((mutation: WorkspaceMutation) => void | Promise) | null = null; + const registerCalls: string[] = []; + const unsubscribeCalls: string[] = []; + const project = createPersistedProjectRecord({ + projectId: "proj-cleanup-race", + rootPath: REPO_CWD, + kind: "git", + displayName: "repo", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const firstWorkspace = createPersistedWorkspaceRecord({ + workspaceId: "ws-cleanup-first", + projectId: project.projectId, + cwd: REPO_CWD, + kind: "local_checkout", + displayName: "main", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const queuedWorkspace = createPersistedWorkspaceRecord({ + workspaceId: "ws-cleanup-queued", + projectId: project.projectId, + cwd: "/tmp/repo/queued", + kind: "local_checkout", + displayName: "queued", + createdAt: "2026-03-01T12:00:00.000Z", + updatedAt: "2026-03-01T12:00:00.000Z", + }); + const workspaces = new Map([ + [firstWorkspace.workspaceId, firstWorkspace], + [queuedWorkspace.workspaceId, queuedWorkspace], + ]); + let listedWorkspaces: PersistedWorkspaceRecord[] = []; + const workspaceRegistry: SessionOptions["workspaceRegistry"] = { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => listedWorkspaces, + get: async (workspaceId: string) => workspaces.get(workspaceId) ?? null, + update: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + subscribeToMutations: (listener) => { + mutationListener = listener; + return () => { + mutationListener = null; + }; + }, + }; + const session = createSessionForWorkspaceTests({ + projectRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [project], + get: async (projectId: string) => (projectId === project.projectId ? project : null), + getOrCreateActiveByRoot: async () => project, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + }, + workspaceRegistry, + workspaceGitService: createNoopWorkspaceGitService({ + registerWorkspace: ({ cwd }) => { + registerCalls.push(path.resolve(cwd)); + return { + unsubscribe: () => { + unsubscribeCalls.push(path.resolve(cwd)); + }, + }; + }, + }), + }); + + await session.handleMessage({ + type: "fetch_workspaces_request", + requestId: "req-fetch-workspaces-cleanup-subscribe", + subscribe: {}, + }); + listedWorkspaces = Array.from(workspaces.values()); + + session.describeWorkspaceRecordWithGitData = async (workspace) => + ({ + id: workspace.workspaceId, + projectId: workspace.projectId, + projectDisplayName: project.displayName, + projectRootPath: project.rootPath, + workspaceDirectory: workspace.cwd, + projectKind: project.kind, + workspaceKind: workspace.kind, + name: workspace.displayName, + status: "done", + activityAt: null, + diffStat: null, + }) as WorkspaceDescriptorPayload; + + let resumeFirstEmit!: () => void; + const firstEmitStarted = new Promise((resolveStarted) => { + session.emitWorkspaceUpdatesForWorkspaceIds = async () => { + resolveStarted(); + await new Promise((resolveResume) => { + resumeFirstEmit = resolveResume; + }); + }; + }); + + const firstMutationPromise = Promise.resolve( + mutationListener?.({ + kind: "upsert", + workspaceId: firstWorkspace.workspaceId, + workspace: firstWorkspace, + }), + ); + expect(firstMutationPromise).toBeDefined(); + await firstEmitStarted; + + const queuedMutationPromise = Promise.resolve( + mutationListener?.({ + kind: "upsert", + workspaceId: queuedWorkspace.workspaceId, + workspace: queuedWorkspace, + }), + ); + + await session.cleanup(); + resumeFirstEmit(); + await firstMutationPromise; + await queuedMutationPromise; + + expect(registerCalls).toEqual([REPO_CWD]); + expect(unsubscribeCalls).toEqual([REPO_CWD]); +}); + test("a workspace leaving a filtered subscription after bootstrap emits a removal", async () => { const emitted: SessionOutboundMessage[] = []; const session = asTestSession( @@ -8302,6 +8775,7 @@ test("workspace.create.response persists the first prompt as the initial title", test("workspace create emits through a matching workspace subscription", async () => { const emitted: SessionOutboundMessage[] = []; const workspaces = new Map>(); + let mutationListener: ((mutation: WorkspaceMutation) => void | Promise) | null = null; const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message), workspaceRegistry: { @@ -8309,11 +8783,23 @@ test("workspace create emits through a matching workspace subscription", async ( existsOnDisk: async () => true, list: async () => Array.from(workspaces.values()), get: async (workspaceId) => workspaces.get(workspaceId) ?? null, - upsert: async (workspace) => { + upsert: async (workspace, context) => { workspaces.set(workspace.workspaceId, workspace); + await mutationListener?.({ + kind: "upsert", + workspaceId: workspace.workspaceId, + workspace, + expectsInitialAgent: context?.expectsInitialAgent, + }); }, archive: async () => {}, remove: async () => {}, + subscribeToMutations: (listener) => { + mutationListener = listener; + return () => { + mutationListener = null; + }; + }, }, }); session.listAgentPayloads = async () => []; @@ -8321,7 +8807,7 @@ test("workspace create emits through a matching workspace subscription", async ( await session.handleMessage({ type: "fetch_workspaces_request", requestId: "req-subscribe-create-match", - filter: { query: "repo" }, + filter: { query: "Implement" }, subscribe: { subscriptionId: "sub-create-match" }, }); emitted.length = 0; @@ -8329,9 +8815,14 @@ test("workspace create emits through a matching workspace subscription", async ( type: "workspace.create.request", requestId: "req-create-match", source: { kind: "directory", path: REPO_CWD }, + firstAgentContext: { prompt: "Implement the requested change" }, }); - expect(filterByType(emitted, "workspace_update")).toHaveLength(1); + const statuses = filterByType(emitted, "workspace_update").flatMap((message) => + message.payload.kind === "upsert" ? [message.payload.workspace.status] : [], + ); + expect(statuses).toContain("running"); + expect(statuses).not.toContain("done"); }); test("workspace create stays out of a non-matching workspace subscription", async () => { diff --git a/packages/server/src/server/session/workspace-provisioning/workspace-provisioning-service.ts b/packages/server/src/server/session/workspace-provisioning/workspace-provisioning-service.ts index 103033803..98d076ecf 100644 --- a/packages/server/src/server/session/workspace-provisioning/workspace-provisioning-service.ts +++ b/packages/server/src/server/session/workspace-provisioning/workspace-provisioning-service.ts @@ -42,6 +42,7 @@ export interface CreateWorktreeWorkspaceInput { branch: string | null; baseBranch: string | null; title: string | null; + expectsInitialAgent?: boolean; } export interface WorkspaceProvisioningService { @@ -55,6 +56,7 @@ export interface WorkspaceProvisioningService { cwd: string, title?: string | null, projectId?: string, + context?: { expectsInitialAgent?: boolean }, ): Promise; createWorkspaceForWorktree( input: CreateWorktreeWorkspaceInput, @@ -175,6 +177,7 @@ export function createWorkspaceProvisioningService(deps: { cwd: string, title?: string | null, projectId?: string, + context?: { expectsInitialAgent?: boolean }, ): Promise { const normalizedCwd = resolve(cwd); const checkout = await workspaceGitService.getCheckout(normalizedCwd); @@ -191,7 +194,7 @@ export function createWorkspaceProvisioningService(deps: { createdAt: timestamp, updatedAt: timestamp, }); - await workspaceRegistry.upsert(workspace); + await workspaceRegistry.upsert(workspace, context); return workspace; } @@ -223,7 +226,9 @@ export function createWorkspaceProvisioningService(deps: { createdAt: timestamp, updatedAt: timestamp, }); - await workspaceRegistry.upsert(workspace); + await workspaceRegistry.upsert(workspace, { + expectsInitialAgent: input.expectsInitialAgent, + }); return workspace; } @@ -294,7 +299,11 @@ export function createWorkspaceProvisioningService(deps: { ): Promise { if (input.createdWorktree) return input.createdWorktree.workspace.workspaceId; if (input.requestedWorkspaceId) return input.requestedWorkspaceId; - return (await createWorkspaceForDirectory(input.cwd, input.initialTitle)).workspaceId; + return ( + await createWorkspaceForDirectory(input.cwd, input.initialTitle, undefined, { + expectsInitialAgent: true, + }) + ).workspaceId; } async function ensureWorkspaceRecordUnarchived( diff --git a/packages/server/src/server/snapshot-mutation-ownership.test.ts b/packages/server/src/server/snapshot-mutation-ownership.test.ts index 87308a3de..373364348 100644 --- a/packages/server/src/server/snapshot-mutation-ownership.test.ts +++ b/packages/server/src/server/snapshot-mutation-ownership.test.ts @@ -119,6 +119,7 @@ describe("snapshot mutation ownership boundary", () => { upsert: directStorageWrite, }), projectRegistry: createStub({ + subscribeToMutations: () => () => {}, initialize: async () => {}, existsOnDisk: async () => true, list: async () => [], @@ -128,6 +129,7 @@ describe("snapshot mutation ownership boundary", () => { remove: async () => {}, }), workspaceRegistry: createStub({ + subscribeToMutations: () => () => {}, initialize: async () => {}, existsOnDisk: async () => true, list: async () => [], diff --git a/packages/server/src/server/workspace-archive-record-scoped.e2e.test.ts b/packages/server/src/server/workspace-archive-record-scoped.e2e.test.ts index 50cd9d20e..5414a8cce 100644 --- a/packages/server/src/server/workspace-archive-record-scoped.e2e.test.ts +++ b/packages/server/src/server/workspace-archive-record-scoped.e2e.test.ts @@ -5,7 +5,11 @@ import path from "node:path"; import { afterEach, beforeEach, expect, test } from "vitest"; import { getFullAccessConfig } from "./daemon-e2e/agent-configs.js"; -import { createDaemonTestContext, type DaemonTestContext } from "./test-utils/index.js"; +import { + createDaemonTestContext, + DaemonClient, + type DaemonTestContext, +} from "./test-utils/index.js"; // Model B archive is scoped to a single workspace RECORD (by workspaceId), not // to a directory on disk. A directory can back multiple workspaces, so archiving @@ -85,6 +89,33 @@ async function terminalIdsForWorkspace(cwd: string, workspaceId: string): Promis return new Set(listed.terminals.map((terminal) => terminal.id)); } +function collectWorkspaceRemovals(client: DaemonClient): { + workspaceIds: string[]; + stop: () => void; +} { + const workspaceIds: string[] = []; + const stop = client.on("workspace_update", (message) => { + if (message.payload.kind === "remove") { + workspaceIds.push(message.payload.id); + } + }); + return { workspaceIds, stop }; +} + +function collectWorkspaceTitles(client: DaemonClient): { + workspaces: Array<{ id: string; name: string; title: string | null }>; + stop: () => void; +} { + const workspaces: Array<{ id: string; name: string; title: string | null }> = []; + const stop = client.on("workspace_update", (message) => { + if (message.payload.kind === "upsert") { + const { id, name, title } = message.payload.workspace; + workspaces.push({ id, name, title }); + } + }); + return { workspaces, stop }; +} + test("archiving one of two workspaces sharing a cwd spares the sibling and the directory", async () => { const cwd = makeTempDir("workspace-archive-shared-cwd-"); @@ -161,6 +192,62 @@ test("archiving one of two workspaces sharing a cwd spares the sibling and the d await ctx.client.killTerminal(terminalBId); }, 60000); +test("archiving a workspace removes it from every subscribed client", async () => { + const cwd = makeTempDir("workspace-archive-global-"); + const workspaceId = await createLocalWorkspace(cwd, "shared workspace"); + const observer = new DaemonClient({ + url: `ws://127.0.0.1:${ctx.daemon.port}/ws`, + reconnect: { enabled: false }, + }); + + await observer.connect(); + const initiatingClientRemovals = collectWorkspaceRemovals(ctx.client); + const removals = collectWorkspaceRemovals(observer); + + try { + await ctx.client.fetchWorkspaces({ subscribe: { subscriptionId: "workspace-initiator" } }); + await observer.fetchWorkspaces({ subscribe: { subscriptionId: "workspace-observer" } }); + + const archive = await ctx.client.archiveWorkspace(workspaceId); + await observer.ping({ requestId: "archive-observer-barrier" }); + + expect(archive.error).toBe(null); + expect(initiatingClientRemovals.workspaceIds).toEqual([workspaceId]); + expect(removals.workspaceIds).toEqual([workspaceId]); + } finally { + initiatingClientRemovals.stop(); + removals.stop(); + await observer.close(); + } +}); + +test("renaming a workspace updates every subscribed client", async () => { + const cwd = makeTempDir("workspace-rename-global-"); + const workspaceId = await createLocalWorkspace(cwd, "shared workspace"); + const observer = new DaemonClient({ + url: `ws://127.0.0.1:${ctx.daemon.port}/ws`, + reconnect: { enabled: false }, + }); + + await observer.connect(); + const titles = collectWorkspaceTitles(observer); + + try { + await observer.fetchWorkspaces({ subscribe: { subscriptionId: "workspace-observer" } }); + + const renamed = await ctx.client.setWorkspaceTitle(workspaceId, "Renamed workspace"); + await observer.ping({ requestId: "rename-observer-barrier" }); + + expect(renamed).toEqual({ title: "Renamed workspace" }); + expect(titles.workspaces).toEqual([ + { id: workspaceId, name: "Renamed workspace", title: "Renamed workspace" }, + ]); + } finally { + titles.stop(); + await observer.close(); + } +}); + test("archiving the last reference to a worktree removes it from disk regardless of the disk flag", async () => { const repoDir = createGitRepo(); diff --git a/packages/server/src/server/workspace-registry.ts b/packages/server/src/server/workspace-registry.ts index f219d9158..14359f80c 100644 --- a/packages/server/src/server/workspace-registry.ts +++ b/packages/server/src/server/workspace-registry.ts @@ -78,6 +78,23 @@ const PersistedWorkspaceRecordSchema = z.object({ export type PersistedProjectRecord = z.infer; export type PersistedWorkspaceRecord = z.infer; +export interface WorkspaceMutation { + kind: "upsert" | "archive" | "remove"; + workspaceId: string; + workspace: PersistedWorkspaceRecord | null; + expectsInitialAgent?: boolean; +} + +export interface WorkspaceMutationContext { + expectsInitialAgent?: boolean; +} + +export interface ProjectMutation { + kind: "upsert" | "archive" | "remove"; + projectId: string; + project: PersistedProjectRecord | null; +} + export interface ProjectRegistry { initialize(): Promise; existsOnDisk(): Promise; @@ -93,13 +110,7 @@ export interface ProjectRegistry { archive(projectId: string, archivedAt: string): Promise; remove(projectId: string): Promise; /** Central lifecycle seam for daemon-global project observers. */ - subscribeToMutations?( - listener: (mutation: { - kind: "upsert" | "archive" | "remove"; - projectId: string; - project: PersistedProjectRecord | null; - }) => void | Promise, - ): () => void; + subscribeToMutations?(listener: (mutation: ProjectMutation) => void | Promise): () => void; } export interface WorkspaceRegistry { @@ -111,9 +122,13 @@ export interface WorkspaceRegistry { workspaceId: string, updater: (record: PersistedWorkspaceRecord) => PersistedWorkspaceRecord, ): Promise; - upsert(record: PersistedWorkspaceRecord): Promise; + upsert(record: PersistedWorkspaceRecord, context?: WorkspaceMutationContext): Promise; archive(workspaceId: string, archivedAt: string): Promise; remove(workspaceId: string): Promise; + /** Central lifecycle seam for daemon-global workspace observers. */ + subscribeToMutations?( + listener: (mutation: WorkspaceMutation) => void | Promise, + ): () => void; } type RegistryRecord = PersistedProjectRecord | PersistedWorkspaceRecord; @@ -186,10 +201,14 @@ class FileBackedRegistry { } async archive(id: string, archivedAt: string): Promise { + await this.archiveIfPresent(id, archivedAt); + } + + protected async archiveIfPresent(id: string, archivedAt: string): Promise { await this.load(); const existing = this.cache.get(id); - if (!existing) return; - await this.persistArchive(existing, archivedAt); + if (!existing) return null; + return this.persistArchive(existing, archivedAt); } protected async archiveIfActive(id: string, archivedAt: string): Promise { @@ -372,6 +391,10 @@ export class FileBackedWorkspaceRegistry extends FileBackedRegistry implements WorkspaceRegistry { + private readonly mutationListeners = new Set< + (mutation: WorkspaceMutation) => void | Promise + >(); + constructor(filePath: string, logger: Logger) { super({ filePath, @@ -381,6 +404,53 @@ export class FileBackedWorkspaceRegistry component: "workspaces", }); } + + subscribeToMutations( + listener: (mutation: WorkspaceMutation) => void | Promise, + ): () => void { + this.mutationListeners.add(listener); + return () => this.mutationListeners.delete(listener); + } + + override async update( + workspaceId: string, + updater: (record: PersistedWorkspaceRecord) => PersistedWorkspaceRecord, + ): Promise { + const workspace = await super.update(workspaceId, updater); + if (workspace) { + await this.notifyMutation({ kind: "upsert", workspaceId, workspace }); + } + return workspace; + } + + override async upsert( + record: PersistedWorkspaceRecord, + context?: WorkspaceMutationContext, + ): Promise { + await super.upsert(record); + await this.notifyMutation({ + kind: "upsert", + workspaceId: record.workspaceId, + workspace: record, + ...(context?.expectsInitialAgent ? { expectsInitialAgent: true } : {}), + }); + } + + override async archive(workspaceId: string, archivedAt: string): Promise { + const workspace = await this.archiveIfPresent(workspaceId, archivedAt); + if (!workspace) return; + await this.notifyMutation({ kind: "archive", workspaceId, workspace }); + } + + override async remove(workspaceId: string): Promise { + const workspace = await this.removeIfPresent(workspaceId); + if (!workspace) return; + await this.notifyMutation({ kind: "remove", workspaceId, workspace: null }); + } + + private async notifyMutation(mutation: WorkspaceMutation): Promise { + await Promise.all([...this.mutationListeners].map((listener) => listener(mutation))); + } } export function createPersistedProjectRecord(input: {