jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Defer workspace diff stats until post-bootstrap updates

Browse Source
This commit is contained in:
Mohamed Boudra
2026-04-07 16:09:47 +07:00
parent 682fc54778
commit 21c7761403
2 changed files with 346 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/session.ts
+109 -70
View File
@@ -5367,7 +5367,21 @@ export class Session {
return { ...base, name: displayName, diffStat };
}
private async listWorkspaceDescriptorsSnapshot(): Promise<WorkspaceDescriptorPayload[]> {
private async buildWorkspaceDescriptor(input: {
workspace: PersistedWorkspaceRecord;
projectRecord?: PersistedProjectRecord | null;
includeGitData: boolean;
}): Promise<WorkspaceDescriptorPayload> {
if (input.includeGitData && input.projectRecord?.kind === "git") {
return this.describeWorkspaceRecordWithGitData(input.workspace, input.projectRecord);
}
return this.describeWorkspaceRecord(input.workspace, input.projectRecord);
}
private async buildWorkspaceDescriptorMap(options: {
includeGitData: boolean;
workspaceIds?: Iterable<string>;
}): Promise<Map<string, WorkspaceDescriptorPayload>> {
const [agents, persistedWorkspaces, persistedProjects] = await Promise.all([
this.listAgentPayloads(),
this.workspaceRegistry.list(),
@@ -5381,14 +5395,26 @@ export class Session {
.map((project) => [project.projectId, project] as const),
);
const descriptorsByWorkspaceId = new Map<string, WorkspaceDescriptorPayload>();
const workspaceIds = options.workspaceIds
? new Set(
Array.from(options.workspaceIds, (workspaceId) =>
normalizePersistedWorkspaceId(workspaceId),
),
)
: null;
for (const workspace of activeRecords) {
if (workspaceIds && !workspaceIds.has(workspace.workspaceId)) {
continue;
}
const projectRecord = activeProjects.get(workspace.projectId) ?? null;
descriptorsByWorkspaceId.set(
workspace.workspaceId,
await this.describeWorkspaceRecord(
await this.buildWorkspaceDescriptor({
workspace,
activeProjects.get(workspace.projectId) ?? null,
),
projectRecord,
includeGitData: options.includeGitData,
}),
);
}
@@ -5410,7 +5436,15 @@ export class Session {
existing.activityAt = this.accumulateLatestActivityAt(existing.activityAt, agent);
}
return Array.from(descriptorsByWorkspaceId.values());
return descriptorsByWorkspaceId;
}
private async listWorkspaceDescriptorsSnapshot(): Promise<WorkspaceDescriptorPayload[]> {
return Array.from(
(await this.buildWorkspaceDescriptorMap({
includeGitData: false,
})).values(),
);
}
private resolveRegisteredWorkspaceIdForCwd(
@@ -5770,8 +5804,7 @@ export class Session {
}
private async reconcileAndEmitWorkspaceUpdates(): Promise<void> {
const subscription = this.workspaceUpdatesSubscription;
if (!subscription) {
if (!this.workspaceUpdatesSubscription) {
return;
}
try {
@@ -5779,82 +5812,34 @@ export class Session {
if (changedWorkspaceIds.size === 0) {
return;
}
const all = await this.listWorkspaceDescriptorsSnapshot();
const descriptorsByWorkspaceId = new Map(all.map((entry) => [entry.id, entry] as const));
for (const workspaceId of changedWorkspaceIds) {
const workspace = descriptorsByWorkspaceId.get(workspaceId) ?? null;
if (workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })) {
this.bufferOrEmitWorkspaceUpdate(subscription, { kind: "upsert", workspace });
} else {
this.bufferOrEmitWorkspaceUpdate(subscription, { kind: "remove", id: workspaceId });
}
}
await this.emitWorkspaceUpdatesForWorkspaceIds(changedWorkspaceIds, {
skipReconcile: true,
});
} catch (error) {
this.sessionLogger.error({ err: error }, "Background workspace reconciliation failed");
}
}
private async emitWorkspaceUpdateForCwd(
cwd: string,
options?: { dedupeGitState?: boolean },
private async emitWorkspaceUpdatesForWorkspaceIds(
workspaceIds: Iterable<string>,
options?: { dedupeGitState?: boolean; skipReconcile?: boolean },
): Promise<void> {
const subscription = this.workspaceUpdatesSubscription;
if (!subscription) {
return;
}
const activeWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => !workspace.archivedAt,
const uniqueWorkspaceIds = new Set(
Array.from(workspaceIds, (workspaceId) => normalizePersistedWorkspaceId(workspaceId)),
);
const workspaceId = this.resolveRegisteredWorkspaceIdForCwd(cwd, activeWorkspaces);
const all = await this.listWorkspaceDescriptorsSnapshot();
const descriptorsByWorkspaceId = new Map(all.map((entry) => [entry.id, entry] as const));
const workspace = descriptorsByWorkspaceId.get(workspaceId);
const nextWorkspace =
workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })
? workspace
: null;
if (
options?.dedupeGitState &&
this.shouldSkipWorkspaceGitWatchUpdate(workspaceId, nextWorkspace)
) {
void this.reconcileAndEmitWorkspaceUpdates();
return;
}
this.rememberWorkspaceGitWatchFingerprint(workspaceId, nextWorkspace);
if (!nextWorkspace) {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: "remove",
id: workspaceId,
});
} else {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: "upsert",
workspace: nextWorkspace,
});
}
void this.reconcileAndEmitWorkspaceUpdates();
}
private async emitWorkspaceUpdatesForCwds(cwds: Iterable<string>): Promise<void> {
const subscription = this.workspaceUpdatesSubscription;
if (!subscription) {
if (uniqueWorkspaceIds.size === 0) {
return;
}
const activeWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => !workspace.archivedAt,
);
const uniqueWorkspaceIds = new Set<string>();
for (const cwd of cwds) {
uniqueWorkspaceIds.add(this.resolveRegisteredWorkspaceIdForCwd(cwd, activeWorkspaces));
}
const all = await this.listWorkspaceDescriptorsSnapshot();
const descriptorsByWorkspaceId = new Map(all.map((entry) => [entry.id, entry] as const));
const descriptorsByWorkspaceId = await this.buildWorkspaceDescriptorMap({
workspaceIds: uniqueWorkspaceIds,
includeGitData: true,
});
for (const workspaceId of uniqueWorkspaceIds) {
const workspace = descriptorsByWorkspaceId.get(workspaceId);
@@ -5862,6 +5847,12 @@ export class Session {
workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })
? workspace
: null;
if (
options?.dedupeGitState &&
this.shouldSkipWorkspaceGitWatchUpdate(workspaceId, nextWorkspace)
) {
continue;
}
this.rememberWorkspaceGitWatchFingerprint(workspaceId, nextWorkspace);
if (!nextWorkspace) {
@@ -5878,7 +5869,52 @@ export class Session {
});
}
void this.reconcileAndEmitWorkspaceUpdates();
if (!options?.skipReconcile) {
void this.reconcileAndEmitWorkspaceUpdates();
}
}
private scheduleWorkspaceGitBootstrapUpdates(options: {
subscriptionId: string;
workspaces: Iterable<WorkspaceDescriptorPayload>;
}): void {
const gitWorkspaceIds = Array.from(options.workspaces, (workspace) => workspace)
.filter((workspace) => workspace.projectKind === "git")
.map((workspace) => workspace.id);
if (gitWorkspaceIds.length === 0) {
return;
}
queueMicrotask(() => {
if (this.workspaceUpdatesSubscription?.subscriptionId !== options.subscriptionId) {
return;
}
void this.emitWorkspaceUpdatesForWorkspaceIds(gitWorkspaceIds, {
skipReconcile: true,
});
});
}
private async emitWorkspaceUpdateForCwd(
cwd: string,
options?: { dedupeGitState?: boolean },
): Promise<void> {
const activeWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => !workspace.archivedAt,
);
const workspaceId = this.resolveRegisteredWorkspaceIdForCwd(cwd, activeWorkspaces);
await this.emitWorkspaceUpdatesForWorkspaceIds([workspaceId], options);
}
private async emitWorkspaceUpdatesForCwds(cwds: Iterable<string>): Promise<void> {
const activeWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => !workspace.archivedAt,
);
const uniqueWorkspaceIds = new Set<string>();
for (const cwd of cwds) {
uniqueWorkspaceIds.add(this.resolveRegisteredWorkspaceIdForCwd(cwd, activeWorkspaces));
}
await this.emitWorkspaceUpdatesForWorkspaceIds(uniqueWorkspaceIds);
}
private async handleFetchAgents(
@@ -5991,6 +6027,10 @@ export class Session {
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
this.flushBootstrappedWorkspaceUpdates({ snapshotLatestActivityByWorkspaceId });
void this.reconcileAndEmitWorkspaceUpdates();
this.scheduleWorkspaceGitBootstrapUpdates({
subscriptionId,
workspaces: payload.entries,
});
}
} catch (error) {
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
@@ -6558,7 +6598,6 @@ export class Session {
: result.status === "error"
? "error"
: "idle";
this.emit({
type: "wait_for_finish_response",
payload: { requestId, status, final, error: null, lastMessage: result.lastMessage },
packages/server/src/server/session.workspaces.test.ts
+237 -50
View File
@@ -964,34 +964,42 @@ describe("workspace aggregation", () => {
};
session.reconcileActiveWorkspaceRecords = async () => new Set();
session.listWorkspaceDescriptorsSnapshot = async () => [
{
id: "/tmp/repo",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "non_git",
workspaceKind: "directory",
name: "repo",
status: "running",
activityAt: "2026-03-01T12:00:00.000Z",
},
];
session.buildWorkspaceDescriptorMap = async () =>
new Map([
[
"/tmp/repo",
{
id: "/tmp/repo",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "non_git",
workspaceKind: "directory",
name: "repo",
status: "running",
activityAt: "2026-03-01T12:00:00.000Z",
},
],
]);
await session.emitWorkspaceUpdateForCwd("/tmp/repo");
session.listWorkspaceDescriptorsSnapshot = async () => [
{
id: "/tmp/repo",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "non_git",
workspaceKind: "directory",
name: "repo",
status: "done",
activityAt: null,
},
];
session.buildWorkspaceDescriptorMap = async () =>
new Map([
[
"/tmp/repo",
{
id: "/tmp/repo",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "non_git",
workspaceKind: "directory",
name: "repo",
status: "done",
activityAt: null,
},
],
]);
await session.emitWorkspaceUpdateForCwd("/tmp/repo");
const workspaceUpdates = emitted.filter((message) => message.type === "workspace_update");
@@ -1083,30 +1091,37 @@ describe("workspace aggregation", () => {
};
session.reconcileActiveWorkspaceRecords = async () =>
new Set(["/tmp/repo", "/tmp/repo/worktree"]);
session.listWorkspaceDescriptorsSnapshot = async () => [
{
id: "/tmp/repo",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "local_checkout",
name: "main",
status: "done",
activityAt: null,
},
{
id: "/tmp/repo/worktree",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "worktree",
name: "feature",
status: "running",
activityAt: "2026-03-01T12:00:00.000Z",
},
];
session.buildWorkspaceDescriptorMap = async () =>
new Map([
[
"/tmp/repo",
{
id: "/tmp/repo",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "local_checkout",
name: "main",
status: "done",
activityAt: null,
},
],
[
"/tmp/repo/worktree",
{
id: "/tmp/repo/worktree",
projectId: "/tmp/repo",
projectDisplayName: "repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "worktree",
name: "feature",
status: "running",
activityAt: "2026-03-01T12:00:00.000Z",
},
],
]);
session.onMessage = (message: { type: string; payload: unknown }) => {
emitted.push(message);
};
@@ -1605,4 +1620,176 @@ describe("workspace aggregation", () => {
rmSync(tempDir, { recursive: true, force: true });
}
});
test("listWorkspaceDescriptorsSnapshot keeps git workspaces on the baseline descriptor path", async () => {
const session = createSessionForWorkspaceTests() as any;
const project = createPersistedProjectRecord({
projectId: "/tmp/repo",
rootPath: "/tmp/repo",
kind: "git",
displayName: "repo",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const workspace = createPersistedWorkspaceRecord({
workspaceId: "/tmp/repo",
projectId: project.projectId,
cwd: "/tmp/repo",
kind: "local_checkout",
displayName: "main",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
session.listAgentPayloads = async () => [];
session.projectRegistry.list = async () => [project];
session.workspaceRegistry.list = async () => [workspace];
const baselineDescriptor = {
id: workspace.workspaceId,
projectId: project.projectId,
projectDisplayName: project.displayName,
projectRootPath: project.rootPath,
projectKind: project.kind,
workspaceKind: workspace.kind,
name: "main",
status: "done",
activityAt: null,
diffStat: null,
} as const;
const gitDescriptor = {
...baselineDescriptor,
diffStat: { additions: 3, deletions: 1 },
} as const;
session.describeWorkspaceRecord = vi.fn(async () => baselineDescriptor);
session.describeWorkspaceRecordWithGitData = vi.fn(async () => gitDescriptor);
const descriptors = await session.listWorkspaceDescriptorsSnapshot();
expect(session.describeWorkspaceRecord).toHaveBeenCalledWith(workspace, project);
expect(session.describeWorkspaceRecordWithGitData).not.toHaveBeenCalled();
expect(descriptors).toEqual([baselineDescriptor]);
});
test("subscribed fetch_workspaces emits git enrichment updates after the baseline snapshot", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const session = createSessionForWorkspaceTests() as any;
const gitProject = createPersistedProjectRecord({
projectId: "/tmp/repo",
rootPath: "/tmp/repo",
kind: "git",
displayName: "repo",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const directoryProject = createPersistedProjectRecord({
projectId: "/tmp/docs",
rootPath: "/tmp/docs",
kind: "non_git",
displayName: "docs",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const gitWorkspace = createPersistedWorkspaceRecord({
workspaceId: "/tmp/repo",
projectId: gitProject.projectId,
cwd: "/tmp/repo",
kind: "local_checkout",
displayName: "main",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const directoryWorkspace = createPersistedWorkspaceRecord({
workspaceId: "/tmp/docs",
projectId: directoryProject.projectId,
cwd: "/tmp/docs",
kind: "directory",
displayName: "docs",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const baselineGitDescriptor = {
id: gitWorkspace.workspaceId,
projectId: gitProject.projectId,
projectDisplayName: gitProject.displayName,
projectRootPath: gitProject.rootPath,
projectKind: gitProject.kind,
workspaceKind: gitWorkspace.kind,
name: "main",
status: "done",
activityAt: null,
diffStat: null,
} as const;
const enrichedGitDescriptor = {
...baselineGitDescriptor,
diffStat: { additions: 3, deletions: 1 },
} as const;
const directoryDescriptor = {
id: directoryWorkspace.workspaceId,
projectId: directoryProject.projectId,
projectDisplayName: directoryProject.displayName,
projectRootPath: directoryProject.rootPath,
projectKind: directoryProject.kind,
workspaceKind: directoryWorkspace.kind,
name: "docs",
status: "done",
activityAt: null,
diffStat: null,
} as const;
session.emit = (message: any) => emitted.push(message);
session.listAgentPayloads = async () => [];
session.projectRegistry.list = async () => [gitProject, directoryProject];
session.workspaceRegistry.list = async () => [gitWorkspace, directoryWorkspace];
session.reconcileAndEmitWorkspaceUpdates = vi.fn(async () => {});
session.describeWorkspaceRecord = vi.fn(
async (workspace: typeof gitWorkspace | typeof directoryWorkspace, project: any) => {
if (workspace.workspaceId === gitWorkspace.workspaceId) {
expect(project).toEqual(gitProject);
return baselineGitDescriptor;
}
expect(project).toEqual(directoryProject);
return directoryDescriptor;
},
);
session.describeWorkspaceRecordWithGitData = vi.fn(async () => enrichedGitDescriptor);
await session.handleMessage({
type: "fetch_workspaces_request",
requestId: "req-fetch-workspaces",
subscribe: {},
});
await new Promise((resolve) => setTimeout(resolve, 0));
const response = emitted.find(
(message) => message.type === "fetch_workspaces_response",
) as { type: "fetch_workspaces_response"; payload: any } | undefined;
expect(
response?.payload.entries.map((entry: typeof baselineGitDescriptor | typeof directoryDescriptor) => [
entry.id,
entry.diffStat,
]),
).toEqual([
[directoryDescriptor.id, directoryDescriptor.diffStat],
[baselineGitDescriptor.id, baselineGitDescriptor.diffStat],
]);
const workspaceUpdates = emitted.filter(
(message) => message.type === "workspace_update",
) as Array<{ type: "workspace_update"; payload: any }>;
expect(workspaceUpdates).toEqual([
{
type: "workspace_update",
payload: {
kind: "upsert",
workspace: enrichedGitDescriptor,
},
},
]);
expect(session.describeWorkspaceRecordWithGitData).toHaveBeenCalledWith(
gitWorkspace,
gitProject,
);
});
});