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: {