fix(server): reconcile stale workspaces by pruning missing directories and fully-archived agents

Workspaces whose directories were deleted (e.g. cleaned-up worktrees) and
workspaces where every agent has been archived now get auto-archived during
reconciliation, cascading to project archival when no active siblings remain.

Extracts detectStaleWorkspaces() as a pure function in workspace-registry-model.ts
with standalone unit tests.
This commit is contained in:
Mohamed Boudra
2026-03-21 13:04:57 +07:00
parent a042fbbe43
commit 2003f308f3
4 changed files with 413 additions and 37 deletions
+246 -6
View File
@@ -1,6 +1,6 @@
import { v4 as uuidv4 } from 'uuid'
import { watch, type FSWatcher } from 'node:fs'
import { stat } from 'fs/promises'
import { readFile, stat } from 'fs/promises'
import { exec } from 'child_process'
import { promisify } from 'util'
import { join, resolve, sep } from 'path'
@@ -101,6 +101,7 @@ import { AgentStorage, type StoredAgentRecord } from './agent/agent-storage.js'
import { isValidAgentProvider, AGENT_PROVIDER_IDS } from './agent/provider-manifest.js'
import {
buildProjectPlacementForCwd,
detectStaleWorkspaces,
deriveProjectKind,
deriveProjectRootPath,
deriveWorkspaceDisplayName,
@@ -184,6 +185,8 @@ const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>()
const DEFAULT_AGENT_PROVIDER = AGENT_PROVIDER_IDS[0]
const CHECKOUT_DIFF_WATCH_DEBOUNCE_MS = 150
const CHECKOUT_DIFF_FALLBACK_REFRESH_MS = 5_000
const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 500
const WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT = '__removed__'
const TERMINAL_STREAM_WINDOW_BYTES = 256 * 1024
const TERMINAL_STREAM_MAX_PENDING_BYTES = 2 * 1024 * 1024
const TERMINAL_STREAM_MAX_PENDING_CHUNKS = 2048
@@ -245,6 +248,15 @@ type CheckoutDiffWatchTarget = {
latestFingerprint: string | null
}
type WorkspaceGitWatchTarget = {
cwd: string
watchers: FSWatcher[]
debounceTimer: NodeJS.Timeout | null
refreshPromise: Promise<void> | null
refreshQueued: boolean
latestFingerprint: string | null
}
type NormalizedGitOptions = {
baseBranch?: string
createNewBranch: boolean
@@ -600,6 +612,7 @@ export class Session {
private nextTerminalStreamId = 1
private readonly checkoutDiffSubscriptions = new Map<string, { targetKey: string }>()
private readonly checkoutDiffTargets = new Map<string, CheckoutDiffWatchTarget>()
private readonly workspaceGitWatchTargets = new Map<string, WorkspaceGitWatchTarget>()
private readonly voiceAgentMcpStdio: VoiceMcpStdioConfig | null
private readonly localSpeechModelsDir: string
private readonly defaultLocalSpeechModelIds: LocalSpeechModelId[]
@@ -1308,6 +1321,9 @@ export class Session {
const normalizedWorkspaceId = normalizePersistedWorkspaceId(workspaceId)
const existing = await this.workspaceRegistry.get(normalizedWorkspaceId)
const placement = await this.buildProjectPlacement(normalizedWorkspaceId)
await this.syncWorkspaceGitWatchTarget(normalizedWorkspaceId, {
isGit: placement.checkout.isGit,
})
const now = new Date().toISOString()
const nextProjectCreatedAt = existing?.createdAt ?? now
const nextWorkspaceCreatedAt = existing?.createdAt ?? now
@@ -1366,8 +1382,32 @@ export class Session {
private async reconcileActiveWorkspaceRecords(): Promise<Set<string>> {
const changedWorkspaceIds = new Set<string>()
const activeWorkspaces = (await this.workspaceRegistry.list()).filter((workspace) => !workspace.archivedAt)
const staleWorkspaceIds = await detectStaleWorkspaces({
activeWorkspaces,
agentRecords: (await this.agentStorage.list()).map((agent) => ({
cwd: agent.cwd,
archivedAt: agent.archivedAt ?? null,
})),
checkDirectoryExists: async (cwd) => {
try {
await stat(cwd)
return true
} catch {
return false
}
},
})
for (const workspaceId of staleWorkspaceIds) {
await this.archiveWorkspaceRecord(workspaceId)
changedWorkspaceIds.add(workspaceId)
}
for (const workspace of activeWorkspaces) {
if (staleWorkspaceIds.has(workspace.workspaceId)) {
continue
}
const result = await this.reconcileWorkspaceRecord(workspace.workspaceId)
if (result.changed) {
changedWorkspaceIds.add(result.workspace.workspaceId)
@@ -3987,6 +4027,174 @@ export class Session {
}
}
private async resolveWorkspaceGitRefsRoot(gitDir: string): Promise<string> {
try {
const commonDir = (await readFile(join(gitDir, 'commondir'), 'utf8')).trim()
if (commonDir.length > 0) {
return resolve(gitDir, commonDir)
}
} catch {
// Regular repos do not have a commondir file.
}
return gitDir
}
private closeWorkspaceGitWatchTarget(target: WorkspaceGitWatchTarget): void {
if (target.debounceTimer) {
clearTimeout(target.debounceTimer)
target.debounceTimer = null
}
for (const watcher of target.watchers) {
watcher.close()
}
target.watchers = []
}
private removeWorkspaceGitWatchTarget(cwd: string): void {
const workspaceId = normalizePersistedWorkspaceId(cwd)
const target = this.workspaceGitWatchTargets.get(workspaceId)
if (!target) {
return
}
this.closeWorkspaceGitWatchTarget(target)
this.workspaceGitWatchTargets.delete(workspaceId)
}
private workspaceGitDescriptorFingerprint(workspace: WorkspaceDescriptorPayload | null): string {
if (!workspace) {
return WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT
}
return JSON.stringify([
workspace.name,
workspace.diffStat ? [workspace.diffStat.additions, workspace.diffStat.deletions] : null,
])
}
private shouldSkipWorkspaceGitWatchUpdate(
workspaceId: string,
workspace: WorkspaceDescriptorPayload | null
): boolean {
const target = this.workspaceGitWatchTargets.get(workspaceId)
if (!target) {
return false
}
const nextFingerprint = this.workspaceGitDescriptorFingerprint(workspace)
if (target.latestFingerprint === nextFingerprint) {
return true
}
target.latestFingerprint = nextFingerprint
return false
}
private rememberWorkspaceGitWatchFingerprint(
workspaceId: string,
workspace: WorkspaceDescriptorPayload | null
): void {
const target = this.workspaceGitWatchTargets.get(workspaceId)
if (!target) {
return
}
target.latestFingerprint = this.workspaceGitDescriptorFingerprint(workspace)
}
private primeWorkspaceGitWatchFingerprints(workspaces: Iterable<WorkspaceDescriptorPayload>): void {
for (const workspace of workspaces) {
this.rememberWorkspaceGitWatchFingerprint(workspace.id, workspace)
}
}
private scheduleWorkspaceGitWatchRefresh(target: WorkspaceGitWatchTarget): void {
if (target.debounceTimer) {
clearTimeout(target.debounceTimer)
}
target.debounceTimer = setTimeout(() => {
target.debounceTimer = null
void this.refreshWorkspaceGitWatchTarget(target)
}, WORKSPACE_GIT_WATCH_DEBOUNCE_MS)
}
private async refreshWorkspaceGitWatchTarget(target: WorkspaceGitWatchTarget): Promise<void> {
if (target.refreshPromise) {
target.refreshQueued = true
return
}
target.refreshPromise = (async () => {
do {
target.refreshQueued = false
await this.emitWorkspaceUpdateForCwd(target.cwd, {
dedupeGitState: true,
})
} while (target.refreshQueued)
})()
try {
await target.refreshPromise
} finally {
target.refreshPromise = null
}
}
private async ensureWorkspaceGitWatchTarget(cwd: string): Promise<void> {
const workspaceId = normalizePersistedWorkspaceId(cwd)
if (this.workspaceGitWatchTargets.has(workspaceId)) {
return
}
const gitDir = await this.resolveCheckoutGitDir(cwd)
if (!gitDir) {
return
}
const refsRoot = await this.resolveWorkspaceGitRefsRoot(gitDir)
const target: WorkspaceGitWatchTarget = {
cwd: workspaceId,
watchers: [],
debounceTimer: null,
refreshPromise: null,
refreshQueued: false,
latestFingerprint: null,
}
for (const watchPath of new Set([join(gitDir, 'HEAD'), join(refsRoot, 'refs', 'heads')])) {
let watcher: FSWatcher | null = null
try {
watcher = watch(watchPath, { recursive: false }, () => {
this.scheduleWorkspaceGitWatchRefresh(target)
})
} catch (error) {
this.sessionLogger.warn({ err: error, cwd, watchPath }, 'Failed to start workspace git watcher')
}
if (!watcher) {
continue
}
watcher.on('error', (error) => {
this.sessionLogger.warn({ err: error, cwd, watchPath }, 'Workspace git watcher error')
})
target.watchers.push(watcher)
}
if (target.watchers.length === 0) {
return
}
this.workspaceGitWatchTargets.set(workspaceId, target)
}
private async syncWorkspaceGitWatchTarget(
cwd: string,
options: { isGit: boolean }
): Promise<void> {
if (!options.isGit) {
this.removeWorkspaceGitWatchTarget(cwd)
return
}
await this.ensureWorkspaceGitWatchTarget(cwd)
}
private async resolveCheckoutWatchRoot(cwd: string): Promise<string | null> {
try {
const { stdout } = await execAsync('git rev-parse --path-format=absolute --show-toplevel', {
@@ -5731,6 +5939,9 @@ export class Session {
await this.projectRegistry.upsert(nextProjectRecord)
await this.workspaceRegistry.upsert(nextWorkspaceRecord)
await this.syncWorkspaceGitWatchTarget(workspaceId, {
isGit: placement.checkout.isGit,
})
if (
existingWorkspace &&
@@ -5746,11 +5957,13 @@ export class Session {
private async archiveWorkspaceRecord(workspaceId: string, archivedAt?: string): Promise<void> {
const existing = await this.workspaceRegistry.get(workspaceId)
if (!existing || existing.archivedAt) {
this.removeWorkspaceGitWatchTarget(workspaceId)
return
}
const nextArchivedAt = archivedAt ?? new Date().toISOString()
await this.workspaceRegistry.archive(workspaceId, nextArchivedAt)
this.removeWorkspaceGitWatchTarget(workspaceId)
const siblingWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => workspace.projectId === existing.projectId && !workspace.archivedAt
@@ -5760,7 +5973,10 @@ export class Session {
}
}
private async emitWorkspaceUpdateForCwd(cwd: string): Promise<void> {
private async emitWorkspaceUpdateForCwd(
cwd: string,
options?: { dedupeGitState?: boolean }
): Promise<void> {
const subscription = this.workspaceUpdatesSubscription
if (!subscription) {
return
@@ -5774,7 +5990,19 @@ export class Session {
for (const nextWorkspaceId of workspaceIdsToEmit) {
const workspace = descriptorsByWorkspaceId.get(nextWorkspaceId)
if (!workspace || !this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })) {
const nextWorkspace =
workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })
? workspace
: null
if (
options?.dedupeGitState &&
this.shouldSkipWorkspaceGitWatchUpdate(nextWorkspaceId, nextWorkspace)
) {
continue
}
this.rememberWorkspaceGitWatchFingerprint(nextWorkspaceId, nextWorkspace)
if (!nextWorkspace) {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'remove',
id: nextWorkspaceId,
@@ -5784,7 +6012,7 @@ export class Session {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'upsert',
workspace,
workspace: nextWorkspace,
})
}
}
@@ -5810,7 +6038,13 @@ export class Session {
for (const workspaceId of uniqueWorkspaceCwds) {
const workspace = descriptorsByWorkspaceId.get(workspaceId)
if (!workspace || !this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })) {
const nextWorkspace =
workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })
? workspace
: null
this.rememberWorkspaceGitWatchFingerprint(workspaceId, nextWorkspace)
if (!nextWorkspace) {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'remove',
id: workspaceId,
@@ -5820,7 +6054,7 @@ export class Session {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'upsert',
workspace,
workspace: nextWorkspace,
})
}
}
@@ -5906,6 +6140,7 @@ export class Session {
}
const payload = await this.listFetchWorkspacesEntries(request)
this.primeWorkspaceGitWatchFingerprints(payload.entries)
const snapshotLatestActivityByWorkspaceId = new Map<string, number>()
for (const entry of payload.entries) {
const parsedLatestActivity = entry.activityAt
@@ -7242,6 +7477,11 @@ export class Session {
}
this.checkoutDiffTargets.clear()
this.checkoutDiffSubscriptions.clear()
for (const target of this.workspaceGitWatchTargets.values()) {
this.closeWorkspaceGitWatchTarget(target)
}
this.workspaceGitWatchTargets.clear()
}
// ============================================================================
@@ -68,7 +68,7 @@ function createSessionForWorkspaceTests(): Session {
error: vi.fn(),
}
return new Session({
const session = new Session({
clientId: 'test-client',
onMessage: vi.fn(),
logger: logger as any,
@@ -108,7 +108,8 @@ function createSessionForWorkspaceTests(): Session {
stt: null,
tts: null,
terminalManager: null,
})
}) as any
return session
}
describe('workspace aggregation', () => {
@@ -528,11 +529,14 @@ describe('workspace aggregation', () => {
const projects = new Map<string, ReturnType<typeof createPersistedProjectRecord>>()
const workspaces = new Map<string, ReturnType<typeof createPersistedWorkspaceRecord>>()
const mainWorkspaceId = '/tmp/inkwell'
const worktreeWorkspaceId = '/tmp/inkwell/.paseo/worktrees/feature-a'
const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), 'session-workspace-reconcile-')))
const mainWorkspaceId = path.join(tempDir, 'inkwell')
const worktreeWorkspaceId = path.join(mainWorkspaceId, '.paseo', 'worktrees', 'feature-a')
const localProjectId = mainWorkspaceId
const remoteProjectId = 'remote:github.com/zimakki/inkwell'
execSync(`mkdir -p ${JSON.stringify(worktreeWorkspaceId)}`)
projects.set(
localProjectId,
createPersistedProjectRecord({
@@ -595,25 +599,29 @@ describe('workspace aggregation', () => {
},
})
await session.handleMessage({
type: 'open_project_request',
cwd: worktreeWorkspaceId,
requestId: 'req-open-worktree',
})
try {
await session.handleMessage({
type: 'open_project_request',
cwd: worktreeWorkspaceId,
requestId: 'req-open-worktree',
})
expect(workspaces.get(mainWorkspaceId)?.projectId).toBe(remoteProjectId)
expect(workspaces.get(worktreeWorkspaceId)?.projectId).toBe(remoteProjectId)
expect(projects.get(localProjectId)?.archivedAt).toBeTruthy()
expect(workspaces.get(mainWorkspaceId)?.projectId).toBe(remoteProjectId)
expect(workspaces.get(worktreeWorkspaceId)?.projectId).toBe(remoteProjectId)
expect(projects.get(localProjectId)?.archivedAt).toBeTruthy()
const workspaceUpdates = emitted.filter((message) => message.type === 'workspace_update') as any[]
expect(workspaceUpdates).toHaveLength(2)
expect(workspaceUpdates.map((message) => message.payload.workspace.id).sort()).toEqual([
mainWorkspaceId,
worktreeWorkspaceId,
])
expect(
workspaceUpdates.every((message) => message.payload.workspace.projectId === remoteProjectId)
).toBe(true)
const workspaceUpdates = emitted.filter((message) => message.type === 'workspace_update') as any[]
expect(workspaceUpdates).toHaveLength(2)
expect(workspaceUpdates.map((message) => message.payload.workspace.id).sort()).toEqual([
mainWorkspaceId,
worktreeWorkspaceId,
])
expect(
workspaceUpdates.every((message) => message.payload.workspace.projectId === remoteProjectId)
).toBe(true)
} finally {
rmSync(tempDir, { recursive: true, force: true })
}
})
test('fetch_workspaces_request reconciles remote URL changes for existing workspaces', async () => {
@@ -621,11 +629,14 @@ describe('workspace aggregation', () => {
const projects = new Map<string, ReturnType<typeof createPersistedProjectRecord>>()
const workspaces = new Map<string, ReturnType<typeof createPersistedWorkspaceRecord>>()
const mainWorkspaceId = '/tmp/inkwell'
const worktreeWorkspaceId = '/tmp/inkwell/.paseo/worktrees/feature-a'
const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), 'session-workspace-fetch-')))
const mainWorkspaceId = path.join(tempDir, 'inkwell')
const worktreeWorkspaceId = path.join(mainWorkspaceId, '.paseo', 'worktrees', 'feature-a')
const oldProjectId = 'remote:github.com/old-owner/inkwell'
const newProjectId = 'remote:github.com/new-owner/inkwell'
execSync(`mkdir -p ${JSON.stringify(worktreeWorkspaceId)}`)
projects.set(
oldProjectId,
createPersistedProjectRecord({
@@ -687,14 +698,18 @@ describe('workspace aggregation', () => {
},
})
const result = await session.listFetchWorkspacesEntries({
type: 'fetch_workspaces_request',
requestId: 'req-fetch-reconcile',
})
try {
const result = await session.listFetchWorkspacesEntries({
type: 'fetch_workspaces_request',
requestId: 'req-fetch-reconcile',
})
expect(result.entries.map((entry: any) => entry.projectId)).toEqual([newProjectId, newProjectId])
expect(workspaces.get(mainWorkspaceId)?.projectId).toBe(newProjectId)
expect(workspaces.get(worktreeWorkspaceId)?.projectId).toBe(newProjectId)
expect(projects.get(oldProjectId)?.archivedAt).toBeTruthy()
expect(result.entries.map((entry: any) => entry.projectId)).toEqual([newProjectId, newProjectId])
expect(workspaces.get(mainWorkspaceId)?.projectId).toBe(newProjectId)
expect(workspaces.get(worktreeWorkspaceId)?.projectId).toBe(newProjectId)
expect(projects.get(oldProjectId)?.archivedAt).toBeTruthy()
} finally {
rmSync(tempDir, { recursive: true, force: true })
}
})
})
@@ -0,0 +1,78 @@
import { describe, expect, test, vi } from 'vitest'
import { detectStaleWorkspaces } from './workspace-registry-model.js'
import { createPersistedWorkspaceRecord } from './workspace-registry.js'
function createWorkspaceRecord(workspaceId: string) {
return createPersistedWorkspaceRecord({
workspaceId,
projectId: workspaceId,
cwd: workspaceId,
kind: 'directory',
displayName: workspaceId.split('/').at(-1) ?? workspaceId,
createdAt: '2026-03-01T00:00:00.000Z',
updatedAt: '2026-03-01T00:00:00.000Z',
})
}
describe('detectStaleWorkspaces', () => {
test('returns workspace ids whose directories no longer exist', async () => {
const checkDirectoryExists = vi.fn(async (cwd: string) => cwd !== '/tmp/missing')
const staleWorkspaceIds = await detectStaleWorkspaces({
activeWorkspaces: [
createWorkspaceRecord('/tmp/existing'),
createWorkspaceRecord('/tmp/missing'),
],
agentRecords: [],
checkDirectoryExists,
})
expect(Array.from(staleWorkspaceIds)).toEqual(['/tmp/missing'])
expect(checkDirectoryExists.mock.calls).toEqual([['/tmp/existing'], ['/tmp/missing']])
})
test('returns workspace ids when all matching agents are archived', async () => {
const staleWorkspaceIds = await detectStaleWorkspaces({
activeWorkspaces: [
createWorkspaceRecord('/tmp/repo'),
createWorkspaceRecord('/tmp/other'),
],
agentRecords: [
{
cwd: '/tmp/repo',
archivedAt: '2026-03-02T00:00:00.000Z',
},
{
cwd: '/tmp/other',
archivedAt: null,
},
],
checkDirectoryExists: async () => true,
})
expect(Array.from(staleWorkspaceIds)).toEqual(['/tmp/repo'])
})
test('keeps workspaces with no agents or at least one active agent', async () => {
const staleWorkspaceIds = await detectStaleWorkspaces({
activeWorkspaces: [
createWorkspaceRecord('/tmp/active'),
createWorkspaceRecord('/tmp/no-agents'),
],
agentRecords: [
{
cwd: '/tmp/active',
archivedAt: '2026-03-02T00:00:00.000Z',
},
{
cwd: '/tmp/active/../active',
archivedAt: null,
},
],
checkDirectoryExists: async () => true,
})
expect(Array.from(staleWorkspaceIds)).toEqual([])
})
})
@@ -2,9 +2,20 @@ import { resolve } from 'node:path'
import { getCheckoutStatusLite } from '../utils/checkout-git.js'
import type { ProjectCheckoutLitePayload, ProjectPlacementPayload } from '../shared/messages.js'
import type { PersistedWorkspaceRecord } from './workspace-registry.js'
export type PersistedProjectKind = 'git' | 'non_git'
export type PersistedWorkspaceKind = 'local_checkout' | 'worktree' | 'directory'
export type StaleWorkspaceAgentRecord = {
cwd: string
archivedAt: string | null
}
export type DetectStaleWorkspacesInput = {
activeWorkspaces: PersistedWorkspaceRecord[]
agentRecords: StaleWorkspaceAgentRecord[]
checkDirectoryExists: (cwd: string) => Promise<boolean>
}
export function normalizeWorkspaceId(cwd: string): string {
const trimmed = cwd.trim()
@@ -128,6 +139,38 @@ export function deriveWorkspaceKind(checkout: ProjectCheckoutLitePayload): Persi
return checkout.isPaseoOwnedWorktree ? 'worktree' : 'local_checkout'
}
export async function detectStaleWorkspaces(
input: DetectStaleWorkspacesInput
): Promise<Set<string>> {
const staleWorkspaceIds = new Set<string>()
const cwdsWithActiveAgents = new Set<string>()
const cwdsWithAnyAgent = new Set<string>()
for (const agent of input.agentRecords) {
const normalizedCwd = normalizeWorkspaceId(agent.cwd)
cwdsWithAnyAgent.add(normalizedCwd)
if (!agent.archivedAt) {
cwdsWithActiveAgents.add(normalizedCwd)
}
}
for (const workspace of input.activeWorkspaces) {
const dirExists = await input.checkDirectoryExists(workspace.cwd)
if (!dirExists) {
staleWorkspaceIds.add(workspace.workspaceId)
continue
}
const hasAgents = cwdsWithAnyAgent.has(workspace.workspaceId)
const hasActiveAgents = cwdsWithActiveAgents.has(workspace.workspaceId)
if (hasAgents && !hasActiveAgents) {
staleWorkspaceIds.add(workspace.workspaceId)
}
}
return staleWorkspaceIds
}
export async function buildProjectPlacementForCwd(input: {
cwd: string
paseoHome: string