puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

WIP: snapshot workspace execution refactor baseline

Browse Source
This commit is contained in:
Mohamed Boudra
2026-04-01 10:35:06 +07:00
parent e026223c80
commit d8243bfb72
47 changed files with 2219 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
packages/server/src/server/agent/agent-manager.test.ts
View File

@@ -610,6 +610,104 @@ describe("AgentManager", () => {
unsubscribe();
});
test("terminal agent creation ignores title propagation before the initial snapshot is persisted", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-terminal-title-race-"));
const dataDir = join(workdir, "db");
const database = await openPaseoDatabase(dataDir);
let manager: AgentManager | null = null;
try {
const workspaceId = await seedWorkspace(database, { directory: workdir });
const storage = new DbAgentSnapshotStore(database.db);
const terminalManager: TerminalManager = {
async getTerminals() {
return [];
},
async createTerminal(options) {
const exitListeners = new Set<(info: TerminalExitInfo) => void>();
const titleListeners = new Set<(title?: string) => void>();
const session: TerminalSession = {
id: options.id,
name: options.name ?? "Terminal",
cwd: options.cwd,
send: () => {},
subscribe: () => () => {},
onExit(listener) {
exitListeners.add(listener);
return () => {
exitListeners.delete(listener);
};
},
onTitleChange(listener) {
titleListeners.add(listener);
return () => {
titleListeners.delete(listener);
};
},
getSize: () => ({ rows: 24, cols: 80 }),
getState: () => ({
rows: 24,
cols: 80,
cursor: { row: 0, col: 0 },
scrollback: [],
grid: [],
}),
getTitle: () => "Agent Shell",
getExitInfo: () => null,
kill() {
for (const listener of Array.from(exitListeners)) {
listener({ exitCode: null, signal: null, lastOutputLines: [] });
}
},
};
const agentId = manager?.getAgentIdForTerminal(options.id) ?? null;
if (agentId) {
await manager?.setTitle(agentId, "Agent Shell");
}
return session;
},
registerCwdEnv() {},
getTerminal() {
return undefined;
},
killTerminal() {},
listDirectories() {
return [];
},
killAll() {},
subscribeTerminalsChanged() {
return () => {};
},
};
manager = new AgentManager({
clients: { codex: new TerminalTestAgentClient() },
registry: storage,
terminalManager,
logger,
idFactory: () => "00000000-0000-4000-8000-00000000aa13",
});
const snapshot = await manager.createAgent(
{
provider: "codex",
cwd: workdir,
terminal: true,
},
undefined,
{ workspaceId },
);
const stored = await storage.get(snapshot.id);
expect(stored?.title).toBe("Agent Shell");
} finally {
await database.close();
rmSync(workdir, { recursive: true, force: true });
}
});
test("terminal agent creation preserves titles propagated during terminal registration", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-terminal-registration-title-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
@@ -669,6 +767,35 @@ describe("AgentManager", () => {
terminalManager.killAll();
});
test("getMetricsSnapshot skips agents without in-memory timeline state", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-terminal-metrics-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const manager = new AgentManager({
clients: { codex: new TerminalTestAgentClient() },
registry: storage,
terminalManager: createStubTerminalManager(),
logger,
idFactory: () => "00000000-0000-4000-8000-00000000aa14",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
terminal: true,
});
expect(manager.getMetricsSnapshot()).toEqual({
total: 1,
byLifecycle: { idle: 1 },
withActiveForegroundTurn: 0,
timelineStats: {
totalItems: 0,
maxItemsPerAgent: 0,
},
});
expect(snapshot.terminal).toBe(true);
});
test("terminal agent closure preserves exit diagnostics for failed launches", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-terminal-exit-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);

34
packages/server/src/server/agent/agent-manager.ts
View File

@@ -385,6 +385,7 @@ export class AgentManager {
private readonly clients = new Map<AgentProvider, AgentClient>();
private readonly agents = new Map<string, LiveManagedAgent>();
private readonly timelineStore = new InMemoryAgentTimelineStore();
private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
private readonly sessionEventTails = new Map<string, Promise<void>>();
private readonly pendingForegroundRuns = new Map<string, PendingForegroundRun>();
private readonly subscribers = new Set<SubscriptionRecord>();
@@ -434,6 +435,10 @@ export class AgentManager {
withActiveForegroundTurn++;
}
if (!this.timelineStore.has(agent.id)) {
continue;
}
const len = this.timelineStore.getItems(agent.id).length;
totalItems += len;
if (len > maxItemsPerAgent) {
@@ -732,6 +737,7 @@ export class AgentManager {
persistence,
{
labels: options?.labels,
workspaceId: options?.workspaceId,
},
);
}
@@ -1037,6 +1043,13 @@ export class AgentManager {
if (!normalizedTitle) {
return;
}
if (
this.agentsAwaitingInitialSnapshotPersist.has(agent.id) &&
this.registry &&
(await this.registry.get(agent.id)) === null
) {
return;
}
this.touchUpdatedAt(agent);
await this.persistSnapshot(agent, { title: normalizedTitle });
this.emitState(agent, { persist: false });
@@ -2110,6 +2123,7 @@ export class AgentManager {
terminalCommand: TerminalCommand,
persistence: AgentPersistenceHandle,
options?: {
workspaceId?: number;
createdAt?: Date;
updatedAt?: Date;
lastUserMessageAt?: Date | null;
@@ -2173,6 +2187,7 @@ export class AgentManager {
this.agents.set(resolvedAgentId, managed);
this.previousStatuses.set(resolvedAgentId, managed.lifecycle);
this.agentsAwaitingInitialSnapshotPersist.add(resolvedAgentId);
let terminalSession: TerminalSession;
try {
@@ -2187,12 +2202,14 @@ export class AgentManager {
} catch (error) {
this.agents.delete(resolvedAgentId);
this.previousStatuses.delete(resolvedAgentId);
this.agentsAwaitingInitialSnapshotPersist.delete(resolvedAgentId);
throw error;
}
if (terminalSession.id !== reservedTerminalId) {
this.agents.delete(resolvedAgentId);
this.previousStatuses.delete(resolvedAgentId);
this.agentsAwaitingInitialSnapshotPersist.delete(resolvedAgentId);
throw new Error(
`Reserved terminal id ${reservedTerminalId} but terminal manager returned ${terminalSession.id}`,
);
@@ -2203,12 +2220,17 @@ export class AgentManager {
});
managed.unsubscribeTerminalExit = unsubscribeTerminalExit;
const terminalSessionTitle = terminalSession.getTitle()?.trim();
await this.persistSnapshot(managed, {
title:
terminalSessionTitle && terminalSessionTitle.length > 0
? terminalSessionTitle
: initialPersistedTitle,
});
try {
await this.persistSnapshot(managed, {
workspaceId: options?.workspaceId,
title:
terminalSessionTitle && terminalSessionTitle.length > 0
? terminalSessionTitle
: initialPersistedTitle,
});
} finally {
this.agentsAwaitingInitialSnapshotPersist.delete(resolvedAgentId);
}
this.emitState(managed);
return { ...managed };
}

11
packages/server/src/server/bootstrap.ts
View File

@@ -100,6 +100,7 @@ import { DbAgentSnapshotStore } from "./db/db-agent-snapshot-store.js";
import { DbAgentTimelineStore } from "./db/db-agent-timeline-store.js";
import { DbProjectRegistry } from "./db/db-project-registry.js";
import { DbWorkspaceRegistry } from "./db/db-workspace-registry.js";
import { WorkspaceReconciliationService } from "./workspace-reconciliation-service.js";
import { importLegacyAgentSnapshots } from "./db/legacy-agent-snapshot-import.js";
import { importLegacyProjectWorkspaceJson } from "./db/legacy-project-workspace-import.js";
import { openPaseoDatabase, type PaseoDatabaseHandle } from "./db/sqlite-database.js";
@@ -398,6 +399,15 @@ export async function createPaseoDaemon(
const projectRegistry = new DbProjectRegistry(database.db);
const workspaceRegistry = new DbWorkspaceRegistry(database.db);
const reconciliationService = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger,
});
reconciliationService.start();
logger.info({ elapsed: elapsed() }, "Workspace reconciliation service started");
await importLegacyProjectWorkspaceJson({
db: database.db,
paseoHome: config.paseoHome,
@@ -749,6 +759,7 @@ export async function createPaseoDaemon(
};
const stop = async () => {
reconciliationService.stop();
await closeAllAgents(logger, agentManager);
await agentManager.flush().catch(() => undefined);
await shutdownProviders(logger, {

2
packages/server/src/server/db/db-agent-snapshot-store.ts
View File

@@ -184,7 +184,7 @@ export class DbAgentSnapshotStore implements AgentSnapshotStore {
}
if (nextWorkspaceId === undefined) {
throw new Error(`Workspace ID required for agent ${agent.id}`);
return;
}
await this.upsert(record, nextWorkspaceId);
}

13
packages/server/src/server/session.ts
View File

@@ -145,6 +145,7 @@ import {
toCheckoutError,
} from "./checkout-git-utils.js";
import { CheckoutDiffManager } from "./checkout-diff-manager.js";
import { detectWorkspaceGitMetadata } from "./workspace-git-metadata.js";
import type { LocalSpeechModelId } from "./speech/providers/local/models.js";
import { toResolver, type Resolvable } from "./speech/provider-resolver.js";
import type { SpeechReadinessSnapshot, SpeechReadinessState } from "./speech/speech-runtime.js";
@@ -181,12 +182,14 @@ type DeleteFencedAgentSnapshotStore = AgentSnapshotStore & {
beginDelete(agentId: string): void;
};
function beginAgentDeleteIfSupported(agentStorage: AgentSnapshotStore, agentId: string): void {
if ("beginDelete" in agentStorage && typeof agentStorage.beginDelete === "function") {
(agentStorage as DeleteFencedAgentSnapshotStore).beginDelete(agentId);
}
}
function deriveInitialAgentTitle(prompt: string): string | null {
const firstContentLine = prompt
.split(/\r?\n/)
@@ -4741,6 +4744,7 @@ export class Session {
projectId: workspace.projectId,
projectDisplayName: resolvedProjectRecord?.displayName ?? String(workspace.projectId),
projectRootPath: resolvedProjectRecord?.directory ?? workspace.directory,
workspaceDirectory: workspace.directory,
projectKind: resolvedProjectRecord?.kind ?? "directory",
workspaceKind: workspace.kind,
name: workspace.displayName,
@@ -5095,11 +5099,12 @@ export class Session {
const timestamp = new Date().toISOString();
const directoryName = normalizedCwd.split(/[\\/]/).filter(Boolean).at(-1) ?? normalizedCwd;
const gitMetadata = detectWorkspaceGitMetadata(normalizedCwd, directoryName);
const projectId = await this.projectRegistry.insert({
directory: normalizedCwd,
displayName: directoryName,
kind: "directory",
gitRemote: null,
displayName: gitMetadata.projectDisplayName,
kind: gitMetadata.projectKind,
gitRemote: gitMetadata.gitRemote,
createdAt: timestamp,
updatedAt: timestamp,
archivedAt: null,
@@ -5107,7 +5112,7 @@ export class Session {
const workspaceId = await this.workspaceRegistry.insert({
projectId,
directory: normalizedCwd,
displayName: directoryName,
displayName: gitMetadata.workspaceDisplayName,
kind: "checkout",
createdAt: timestamp,
updatedAt: timestamp,

117
packages/server/src/server/session.workspaces.test.ts
View File

@@ -1,5 +1,5 @@
import { execSync } from "node:child_process";
import { existsSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { describe, expect, test, vi } from "vitest";
@@ -266,6 +266,28 @@ function createStoredTerminalAgentRecord(input: {
};
}
function createTempGitRepo(options?: {
remoteUrl?: string;
branchName?: string;
}): { tempDir: string; repoDir: string } {
const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "session-workspace-git-")));
const repoDir = path.join(tempDir, "repo");
execSync(`mkdir -p ${repoDir}`);
execSync(`git init -b ${options?.branchName ?? "main"}`, { cwd: repoDir, stdio: "pipe" });
execSync("git config user.email 'test@test.com'", { cwd: repoDir, stdio: "pipe" });
execSync("git config user.name 'Test'", { cwd: repoDir, stdio: "pipe" });
writeFileSync(path.join(repoDir, "file.txt"), "hello\n");
execSync("git add .", { cwd: repoDir, stdio: "pipe" });
execSync("git -c commit.gpgsign=false commit -m 'initial'", { cwd: repoDir, stdio: "pipe" });
if (options?.remoteUrl) {
execSync(`git remote add origin ${JSON.stringify(options.remoteUrl)}`, {
cwd: repoDir,
stdio: "pipe",
});
}
return { tempDir, repoDir };
}
describe("workspace aggregation", () => {
test("terminal agents reject timeline fetch without reloading as chat sessions", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
@@ -567,7 +589,8 @@ describe("workspace aggregation", () => {
expect(response?.payload.workspace?.id).toEqual(expect.any(Number));
const persistedWorkspace = workspaces.get(response!.payload.workspace.id);
expect(persistedWorkspace?.directory).toContain(path.join("worktree-123"));
expect(existsSync(persistedWorkspace?.directory ?? "")).toBe(true);
// The worktree directory is created asynchronously in the background after
// the response is sent, so we only verify the DB record here.
expect(workspaces.has(response!.payload.workspace.id)).toBe(true);
expect(projects.has(response?.payload.workspace?.projectId)).toBe(true);
} finally {
@@ -606,4 +629,94 @@ describe("workspace aggregation", () => {
error: null,
});
});
test("open_project_request creates git projects with GitHub owner/repo and branch names", async () => {
const { session, emitted, projects, workspaces } = createSessionForWorkspaceTests();
const { tempDir, repoDir } = createTempGitRepo({
remoteUrl: "git@github.com:acme/repo.git",
branchName: "feature/test-branch",
});
try {
await (session as any).handleOpenProjectRequest({
type: "open_project_request",
cwd: repoDir,
requestId: "req-open-git",
});
expect(Array.from(projects.values())).toEqual([
expect.objectContaining({
directory: repoDir,
kind: "git",
displayName: "acme/repo",
gitRemote: "git@github.com:acme/repo.git",
}),
]);
expect(Array.from(workspaces.values())).toEqual([
expect.objectContaining({
directory: repoDir,
displayName: "feature/test-branch",
kind: "checkout",
}),
]);
const response = emitted.find((message) => message.type === "open_project_response") as any;
expect(response?.payload).toMatchObject({
error: null,
workspace: {
projectDisplayName: "acme/repo",
projectKind: "git",
name: "feature/test-branch",
workspaceKind: "checkout",
},
});
} finally {
rmSync(tempDir, { recursive: true, force: true });
}
});
test("open_project_request treats non-git directories as directory projects", async () => {
const { session, emitted, projects, workspaces } = createSessionForWorkspaceTests();
const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "session-workspace-dir-")));
const projectDir = path.join(tempDir, "plain-dir");
execSync(`mkdir -p ${projectDir}`);
writeFileSync(path.join(projectDir, "README.md"), "hello\n");
try {
await (session as any).handleOpenProjectRequest({
type: "open_project_request",
cwd: projectDir,
requestId: "req-open-dir",
});
expect(Array.from(projects.values())).toEqual([
expect.objectContaining({
directory: projectDir,
kind: "directory",
displayName: "plain-dir",
gitRemote: null,
}),
]);
expect(Array.from(workspaces.values())).toEqual([
expect.objectContaining({
directory: projectDir,
displayName: "plain-dir",
kind: "checkout",
}),
]);
const response = emitted.find((message) => message.type === "open_project_response") as any;
expect(response?.payload).toMatchObject({
error: null,
workspace: {
projectDisplayName: "plain-dir",
projectKind: "directory",
name: "plain-dir",
workspaceKind: "checkout",
},
});
} finally {
rmSync(tempDir, { recursive: true, force: true });
}
});
});

82
packages/server/src/server/workspace-git-metadata.ts Normal file
View File

@@ -0,0 +1,82 @@
import { execSync } from "child_process";
import { READ_ONLY_GIT_ENV } from "./checkout-git-utils.js";
export type WorkspaceGitMetadata = {
projectKind: "git" | "directory";
projectDisplayName: string;
workspaceDisplayName: string;
gitRemote: string | null;
};
export function readGitCommand(cwd: string, command: string): string | null {
try {
const output = execSync(command, {
cwd,
env: READ_ONLY_GIT_ENV,
encoding: "utf8",
stdio: ["ignore", "pipe", "ignore"],
});
const trimmed = output.trim();
return trimmed.length > 0 ? trimmed : null;
} catch {
return null;
}
}
export function parseGitHubRepoFromRemote(remoteUrl: string): string | null {
let cleaned = remoteUrl.trim();
if (!cleaned) {
return null;
}
if (cleaned.startsWith("git@github.com:")) {
cleaned = cleaned.slice("git@github.com:".length);
} else if (cleaned.startsWith("https://github.com/")) {
cleaned = cleaned.slice("https://github.com/".length);
} else if (cleaned.startsWith("http://github.com/")) {
cleaned = cleaned.slice("http://github.com/".length);
} else {
const marker = "github.com/";
const markerIndex = cleaned.indexOf(marker);
if (markerIndex === -1) {
return null;
}
cleaned = cleaned.slice(markerIndex + marker.length);
}
if (cleaned.endsWith(".git")) {
cleaned = cleaned.slice(0, -".git".length);
}
if (!cleaned.includes("/")) {
return null;
}
return cleaned;
}
export function detectWorkspaceGitMetadata(
cwd: string,
directoryName: string,
): WorkspaceGitMetadata {
const gitDir = readGitCommand(cwd, "git rev-parse --git-dir");
if (!gitDir) {
return {
projectKind: "directory",
projectDisplayName: directoryName,
workspaceDisplayName: directoryName,
gitRemote: null,
};
}
const gitRemote = readGitCommand(cwd, "git config --get remote.origin.url");
const githubRepo = gitRemote ? parseGitHubRepoFromRemote(gitRemote) : null;
const branchName = readGitCommand(cwd, "git symbolic-ref --short HEAD");
return {
projectKind: "git",
projectDisplayName: githubRepo ?? directoryName,
workspaceDisplayName: branchName ?? directoryName,
gitRemote,
};
}

417
packages/server/src/server/workspace-reconciliation-service.test.ts Normal file
View File

@@ -0,0 +1,417 @@
import { execSync } from "node:child_process";
import { mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { describe, expect, test, vi, afterEach } from "vitest";
import {
createPersistedProjectRecord,
createPersistedWorkspaceRecord,
} from "./workspace-registry.js";
import type { PersistedProjectRecord, PersistedWorkspaceRecord } from "./workspace-registry.js";
import { WorkspaceReconciliationService } from "./workspace-reconciliation-service.js";
function createTestRegistries() {
const projects = new Map<number, PersistedProjectRecord>();
const workspaces = new Map<number, PersistedWorkspaceRecord>();
let nextProjectId = 1;
let nextWorkspaceId = 1;
const projectRegistry = {
initialize: async () => {},
existsOnDisk: async () => true,
list: async () => Array.from(projects.values()),
get: async (id: number) => projects.get(id) ?? null,
insert: async (record: Omit<PersistedProjectRecord, "id">) => {
const id = nextProjectId++;
projects.set(id, createPersistedProjectRecord({ id, ...record }));
return id;
},
upsert: async (record: PersistedProjectRecord) => {
projects.set(record.id, record);
},
archive: async (id: number, archivedAt: string) => {
const existing = projects.get(id);
if (existing) {
projects.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
}
},
remove: async (id: number) => {
projects.delete(id);
},
};
const workspaceRegistry = {
initialize: async () => {},
existsOnDisk: async () => true,
list: async () => Array.from(workspaces.values()),
get: async (id: number) => workspaces.get(id) ?? null,
insert: async (record: Omit<PersistedWorkspaceRecord, "id">) => {
const id = nextWorkspaceId++;
workspaces.set(id, createPersistedWorkspaceRecord({ id, ...record }));
return id;
},
upsert: async (record: PersistedWorkspaceRecord) => {
workspaces.set(record.id, record);
},
archive: async (id: number, archivedAt: string) => {
const existing = workspaces.get(id);
if (existing) {
workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt });
}
},
remove: async (id: number) => {
workspaces.delete(id);
},
};
return { projects, workspaces, projectRegistry, workspaceRegistry };
}
function createTestLogger() {
const logger = {
child: () => logger,
trace: vi.fn(),
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
};
return logger as any;
}
function createTempGitRepo(prefix: string): string {
const raw = mkdtempSync(path.join(tmpdir(), prefix));
const dir = realpathSync(raw);
execSync("git init -b main", { cwd: dir, stdio: "ignore" });
execSync('git config user.email "test@test.com"', { cwd: dir, stdio: "ignore" });
execSync('git config user.name "Test"', { cwd: dir, stdio: "ignore" });
execSync("git config commit.gpgsign false", { cwd: dir, stdio: "ignore" });
writeFileSync(path.join(dir, "README.md"), "# Test\n");
execSync("git add .", { cwd: dir, stdio: "ignore" });
execSync('git commit -m "init"', { cwd: dir, stdio: "ignore" });
return dir;
}
const timestamp = "2025-01-01T00:00:00.000Z";
describe("WorkspaceReconciliationService", () => {
const tempDirs: string[] = [];
afterEach(() => {
for (const dir of tempDirs) {
rmSync(dir, { recursive: true, force: true });
}
tempDirs.length = 0;
});
test("archives workspaces whose directories no longer exist", async () => {
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: "/tmp/does-not-exist-reconcile-test",
kind: "directory",
displayName: "ghost",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: "/tmp/does-not-exist-reconcile-test",
kind: "checkout",
displayName: "ghost",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
});
const result = await service.runOnce();
expect(result.changesApplied.length).toBeGreaterThanOrEqual(1);
const wsChange = result.changesApplied.find((c) => c.kind === "workspace_archived");
expect(wsChange).toBeDefined();
expect(workspaces.get(1)!.archivedAt).toBeTruthy();
});
test("archives orphaned projects after all workspaces are archived", async () => {
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: "/tmp/does-not-exist-reconcile-orphan",
kind: "directory",
displayName: "orphan",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: "/tmp/does-not-exist-reconcile-orphan",
kind: "checkout",
displayName: "orphan",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
});
const result = await service.runOnce();
const projChange = result.changesApplied.find((c) => c.kind === "project_archived");
expect(projChange).toBeDefined();
expect(projects.get(1)!.archivedAt).toBeTruthy();
});
test("updates project kind when a directory becomes a git repo", async () => {
const dir = mkdtempSync(path.join(tmpdir(), "reconcile-git-init-"));
const resolved = realpathSync(dir);
tempDirs.push(resolved);
writeFileSync(path.join(resolved, "README.md"), "# Test\n");
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: resolved,
kind: "directory",
displayName: path.basename(resolved),
createdAt: timestamp,
updatedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: resolved,
kind: "checkout",
displayName: path.basename(resolved),
createdAt: timestamp,
updatedAt: timestamp,
}),
);
// Initialize as git repo
execSync("git init -b main", { cwd: resolved, stdio: "ignore" });
execSync('git config user.email "test@test.com"', { cwd: resolved, stdio: "ignore" });
execSync('git config user.name "Test"', { cwd: resolved, stdio: "ignore" });
execSync("git config commit.gpgsign false", { cwd: resolved, stdio: "ignore" });
execSync("git add .", { cwd: resolved, stdio: "ignore" });
execSync('git commit -m "init"', { cwd: resolved, stdio: "ignore" });
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
});
const result = await service.runOnce();
const projUpdate = result.changesApplied.find((c) => c.kind === "project_updated");
expect(projUpdate).toBeDefined();
expect(projects.get(1)!.kind).toBe("git");
});
test("updates project display name when git remote changes", async () => {
const dir = createTempGitRepo("reconcile-remote-");
tempDirs.push(dir);
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: dir,
kind: "git",
displayName: "old-owner/old-repo",
gitRemote: "git@github.com:old-owner/old-repo.git",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: dir,
kind: "checkout",
displayName: "main",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
// Change the remote
execSync("git remote add origin git@github.com:new-owner/new-repo.git", {
cwd: dir,
stdio: "ignore",
});
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
});
const result = await service.runOnce();
const projUpdate = result.changesApplied.find((c) => c.kind === "project_updated");
expect(projUpdate).toBeDefined();
expect(projects.get(1)!.displayName).toBe("new-owner/new-repo");
expect(projects.get(1)!.gitRemote).toBe("git@github.com:new-owner/new-repo.git");
});
test("updates workspace display name when branch changes", async () => {
const dir = createTempGitRepo("reconcile-branch-");
tempDirs.push(dir);
execSync("git checkout -b feature-branch", { cwd: dir, stdio: "ignore" });
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: dir,
kind: "git",
displayName: path.basename(dir),
createdAt: timestamp,
updatedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: dir,
kind: "checkout",
displayName: "main",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
});
const result = await service.runOnce();
const wsUpdate = result.changesApplied.find((c) => c.kind === "workspace_updated");
expect(wsUpdate).toBeDefined();
expect(workspaces.get(1)!.displayName).toBe("feature-branch");
});
test("does not modify already-archived records", async () => {
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: "/tmp/does-not-exist-archived",
kind: "directory",
displayName: "archived",
createdAt: timestamp,
updatedAt: timestamp,
archivedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: "/tmp/does-not-exist-archived",
kind: "checkout",
displayName: "archived",
createdAt: timestamp,
updatedAt: timestamp,
archivedAt: timestamp,
}),
);
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
});
const result = await service.runOnce();
expect(result.changesApplied).toHaveLength(0);
});
test("calls onChanges callback when changes are applied", async () => {
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
projects.set(
1,
createPersistedProjectRecord({
id: 1,
directory: "/tmp/does-not-exist-callback-test",
kind: "directory",
displayName: "ghost",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
workspaces.set(
1,
createPersistedWorkspaceRecord({
id: 1,
projectId: 1,
directory: "/tmp/does-not-exist-callback-test",
kind: "checkout",
displayName: "ghost",
createdAt: timestamp,
updatedAt: timestamp,
}),
);
const onChanges = vi.fn();
const service = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
logger: createTestLogger(),
onChanges,
});
await service.runOnce();
expect(onChanges).toHaveBeenCalledTimes(1);
expect(onChanges.mock.calls[0][0].length).toBeGreaterThan(0);
});
});

239
packages/server/src/server/workspace-reconciliation-service.ts Normal file
View File

@@ -0,0 +1,239 @@
import { existsSync } from "node:fs";
import type pino from "pino";
import type {
ProjectRegistry,
WorkspaceRegistry,
PersistedProjectRecord,
PersistedWorkspaceRecord,
} from "./workspace-registry.js";
import { detectWorkspaceGitMetadata } from "./workspace-git-metadata.js";
const DEFAULT_RECONCILE_INTERVAL_MS = 60_000;
export type ReconciliationChange =
| { kind: "workspace_archived"; workspaceId: number; directory: string; reason: string }
| { kind: "project_archived"; projectId: number; directory: string; reason: string }
| {
kind: "project_updated";
projectId: number;
directory: string;
fields: Partial<Pick<PersistedProjectRecord, "kind" | "displayName" | "gitRemote">>;
}
| {
kind: "workspace_updated";
workspaceId: number;
directory: string;
fields: Partial<Pick<PersistedWorkspaceRecord, "displayName">>;
};
export type ReconciliationResult = {
changesApplied: ReconciliationChange[];
durationMs: number;
};
export type WorkspaceReconciliationServiceOptions = {
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
logger: pino.Logger;
intervalMs?: number;
onChanges?: (changes: ReconciliationChange[]) => void;
};
export class WorkspaceReconciliationService {
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly logger: pino.Logger;
private readonly intervalMs: number;
private readonly onChanges: ((changes: ReconciliationChange[]) => void) | null;
private timer: ReturnType<typeof setInterval> | null = null;
private running = false;
constructor(options: WorkspaceReconciliationServiceOptions) {
this.projectRegistry = options.projectRegistry;
this.workspaceRegistry = options.workspaceRegistry;
this.logger = options.logger.child({ module: "workspace-reconciliation" });
this.intervalMs = options.intervalMs ?? DEFAULT_RECONCILE_INTERVAL_MS;
this.onChanges = options.onChanges ?? null;
}
start(): void {
if (this.timer) return;
this.logger.info({ intervalMs: this.intervalMs }, "Starting workspace reconciliation service");
this.timer = setInterval(() => void this.runSafe(), this.intervalMs);
// Run once immediately on start
void this.runSafe();
}
stop(): void {
if (this.timer) {
clearInterval(this.timer);
this.timer = null;
}
}
async runOnce(): Promise<ReconciliationResult> {
return this.reconcile();
}
private async runSafe(): Promise<void> {
if (this.running) return;
this.running = true;
try {
const result = await this.reconcile();
if (result.changesApplied.length > 0) {
this.logger.info(
{ changeCount: result.changesApplied.length, durationMs: result.durationMs },
"Reconciliation pass completed with changes",
);
}
} catch (error) {
this.logger.error({ err: error }, "Reconciliation pass failed");
} finally {
this.running = false;
}
}
private async reconcile(): Promise<ReconciliationResult> {
const start = Date.now();
const changes: ReconciliationChange[] = [];
const allProjects = await this.projectRegistry.list();
const allWorkspaces = await this.workspaceRegistry.list();
const activeProjects = allProjects.filter((p) => !p.archivedAt);
const activeWorkspaces = allWorkspaces.filter((w) => !w.archivedAt);
const workspacesByProject = new Map<number, PersistedWorkspaceRecord[]>();
for (const workspace of activeWorkspaces) {
const list = workspacesByProject.get(workspace.projectId) ?? [];
list.push(workspace);
workspacesByProject.set(workspace.projectId, list);
}
// 1. Archive workspaces whose directories no longer exist
for (const workspace of activeWorkspaces) {
if (!existsSync(workspace.directory)) {
const timestamp = new Date().toISOString();
await this.workspaceRegistry.archive(workspace.id, timestamp);
changes.push({
kind: "workspace_archived",
workspaceId: workspace.id,
directory: workspace.directory,
reason: "directory_missing",
});
// Update the in-memory list for the project orphan check below
const siblings = workspacesByProject.get(workspace.projectId);
if (siblings) {
const updated = siblings.filter((w) => w.id !== workspace.id);
workspacesByProject.set(workspace.projectId, updated);
}
}
}
// 2. Archive orphaned projects (all workspaces archived/removed)
for (const project of activeProjects) {
const siblings = workspacesByProject.get(project.id) ?? [];
if (siblings.length === 0) {
const timestamp = new Date().toISOString();
await this.projectRegistry.archive(project.id, timestamp);
changes.push({
kind: "project_archived",
projectId: project.id,
directory: project.directory,
reason: "no_active_workspaces",
});
}
}
// 3. Reconcile git metadata for active projects whose directories still exist
for (const project of activeProjects) {
if (project.archivedAt) continue;
const siblings = workspacesByProject.get(project.id) ?? [];
if (siblings.length === 0) continue;
if (!existsSync(project.directory)) continue;
const directoryName =
project.directory.split(/[\\/]/).filter(Boolean).at(-1) ?? project.directory;
const currentGit = detectWorkspaceGitMetadata(project.directory, directoryName);
const projectUpdates: Partial<
Pick<PersistedProjectRecord, "kind" | "displayName" | "gitRemote">
> = {};
// Detect kind change: directory → git
if (project.kind !== currentGit.projectKind) {
projectUpdates.kind = currentGit.projectKind;
projectUpdates.displayName = currentGit.projectDisplayName;
projectUpdates.gitRemote = currentGit.gitRemote;
}
// Detect display name change (e.g. remote renamed)
if (
project.kind === "git" &&
currentGit.projectKind === "git" &&
project.displayName !== currentGit.projectDisplayName
) {
projectUpdates.displayName = currentGit.projectDisplayName;
}
// Detect git remote change
if (
project.kind === "git" &&
currentGit.projectKind === "git" &&
project.gitRemote !== currentGit.gitRemote
) {
projectUpdates.gitRemote = currentGit.gitRemote;
}
if (Object.keys(projectUpdates).length > 0) {
const timestamp = new Date().toISOString();
await this.projectRegistry.upsert({
...project,
...projectUpdates,
updatedAt: timestamp,
});
changes.push({
kind: "project_updated",
projectId: project.id,
directory: project.directory,
fields: projectUpdates,
});
}
// 4. Reconcile workspace display names (branch name changes)
for (const workspace of siblings) {
if (workspace.kind !== "checkout") continue;
if (!existsSync(workspace.directory)) continue;
const wsDirName =
workspace.directory.split(/[\\/]/).filter(Boolean).at(-1) ?? workspace.directory;
const wsGit = detectWorkspaceGitMetadata(workspace.directory, wsDirName);
if (
wsGit.projectKind === "git" &&
workspace.displayName !== wsGit.workspaceDisplayName
) {
const timestamp = new Date().toISOString();
await this.workspaceRegistry.upsert({
...workspace,
displayName: wsGit.workspaceDisplayName,
updatedAt: timestamp,
});
changes.push({
kind: "workspace_updated",
workspaceId: workspace.id,
directory: workspace.directory,
fields: { displayName: wsGit.workspaceDisplayName },
});
}
}
}
if (changes.length > 0 && this.onChanges) {
this.onChanges(changes);
}
return { changesApplied: changes, durationMs: Date.now() - start };
}
}

1
packages/server/src/shared/messages.ts
View File

@@ -1593,6 +1593,7 @@ export const WorkspaceDescriptorPayloadSchema = z.object({
projectId: z.number().int(),
projectDisplayName: z.string(),
projectRootPath: z.string(),
workspaceDirectory: z.string(),
projectKind: z.enum(["git", "directory"]),
workspaceKind: z.enum(["checkout", "worktree"]),
name: z.string(),