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

Make workspace git registration nonblocking (#542)

Browse Source
This commit is contained in:
Mohamed Boudra
2026-04-23 18:21:59 +08:00
committed by GitHub
parent 083cd34db4
commit dfc8223bcc
13 changed files with 557 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
packages/app/src/hooks/use-sidebar-workspaces-list.test.ts
View File

@@ -85,6 +85,17 @@ function DisabledRenderCountProbe({
return null;
}
function HookResultProbe({
onResult,
serverId,
}: {
onResult: (result: ReturnType<typeof useSidebarWorkspacesList>) => void;
serverId: string;
}): null {
onResult(useSidebarWorkspacesList({ serverId }));
return null;
}
describe("applyStoredOrdering", () => {
it("keeps unknown items on the baseline while applying stored order", () => {
const result = applyStoredOrdering({
@@ -206,6 +217,7 @@ describe("useSidebarWorkspacesList", () => {
act(() => {
getHostRuntimeStore().syncHosts([]);
useSessionStore.getState().clearSession("srv-disabled");
useSessionStore.getState().clearSession("srv-loading");
useSidebarOrderStore.setState({
projectOrderByServerId: {},
workspaceOrderByServerAndProject: {},
@@ -234,6 +246,30 @@ describe("useSidebarWorkspacesList", () => {
expect(useSidebarOrderStore.getState().workspaceOrderByServerAndProject).toEqual({});
});
it("keeps the sidebar in initial load until workspace hydration succeeds", async () => {
const onResult = vi.fn();
act(() => {
useSessionStore.getState().initializeSession("srv-loading", null as unknown as DaemonClient);
});
container = document.createElement("div");
document.body.appendChild(container);
root = createRoot(container);
await act(async () => {
root?.render(React.createElement(HookResultProbe, { serverId: "srv-loading", onResult }));
});
expect(onResult).toHaveBeenLastCalledWith(
expect.objectContaining({
projects: [],
isLoading: true,
isInitialLoad: true,
}),
);
});
it("does not subscribe to order updates while disabled", async () => {
const onRender = vi.fn();

3
packages/app/src/hooks/use-sidebar-workspaces-list.ts
View File

@@ -303,8 +303,7 @@ export function useSidebarWorkspacesList(options?: {
})();
}, [connectionStatus, isActive, runtime, serverId]);
const isLoading =
isActive && Boolean(serverId) && connectionStatus === "online" && !hasHydratedWorkspaces;
const isLoading = isActive && Boolean(serverId) && !hasHydratedWorkspaces;
const isInitialLoad = isLoading && projects.length === 0;
const isRevalidating = false;

8
packages/server/src/server/agent/mcp-server.test.ts
View File

@@ -1274,11 +1274,9 @@ describe("agent snapshot MCP serialization", () => {
const tool = (server as any)._registeredTools["list_agents"];
const response = await tool.callback({});
expect(response.structuredContent.agents.map((agent: { id: string }) => agent.id)).toEqual([
"in-cwd",
"in-child-cwd",
"stored-in-cwd",
]);
const agentIds = response.structuredContent.agents.map((agent: { id: string }) => agent.id);
expect(agentIds).toHaveLength(3);
expect(new Set(agentIds)).toEqual(new Set(["in-cwd", "in-child-cwd", "stored-in-cwd"]));
});
it("allows explicit cwd, status, archive, time, and limit filters for list_agents", async () => {

3
packages/server/src/server/paseo-worktree-service.test.ts
View File

@@ -236,8 +236,7 @@ function createGitHubServiceStub(): GitHubService {
function createWorkspaceGitServiceStub(): WorkspaceGitService {
return {
subscribe: async (params) => ({
initial: createWorkspaceGitSnapshot(params.cwd),
registerWorkspace: () => ({
unsubscribe: () => {},
}),
peekSnapshot: (cwd) => createWorkspaceGitSnapshot(cwd),

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

@@ -217,7 +217,7 @@ import { killTerminalsUnderPath as killWorktreeTerminalsUnderPath } from "./pase
import { toWorktreeWireError } from "./worktree-errors.js";
const MAX_INITIAL_AGENT_TITLE_CHARS = Math.min(60, MAX_EXPLICIT_AGENT_TITLE_CHARS);
const WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT = "__removed__";
const WORKSPACE_GIT_WATCH_REMOVED_STATE_KEY = "__removed__";
type GitMutationRefreshReason =
| "commit-changes"
| "pull"
@@ -326,7 +326,7 @@ type WorkspaceGitWatchTarget = {
debounceTimer: ReturnType<typeof setTimeout> | null;
refreshPromise: Promise<void> | null;
refreshQueued: boolean;
latestFingerprint: string | null;
latestDescriptorStateKey: string | null;
lastBranchName: string | null;
};
@@ -897,11 +897,9 @@ export class Session {
}
}
async primeWorkspaceGitWatchFingerprintForWorkspace(
workspace: PersistedWorkspaceRecord,
): Promise<void> {
async syncWorkspaceGitObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
const descriptor = await this.describeWorkspaceRecordWithGitData(workspace);
await this.primeWorkspaceGitWatchFingerprints([descriptor]);
this.syncWorkspaceGitObservers([descriptor]);
}
async emitWorkspaceUpdateForWorkspaceId(workspaceId: string): Promise<void> {
@@ -917,7 +915,7 @@ export class Session {
}
async warmWorkspaceGitDataForWorkspace(workspace: PersistedWorkspaceRecord): Promise<void> {
await this.primeWorkspaceGitWatchFingerprintForWorkspace(workspace);
await this.syncWorkspaceGitObserverForWorkspace(workspace);
await this.emitWorkspaceUpdateForWorkspaceId(workspace.workspaceId);
}
@@ -4360,9 +4358,9 @@ export class Session {
this.workspaceGitSubscriptions.delete(normalizedCwd);
}
private workspaceGitDescriptorFingerprint(workspace: WorkspaceDescriptorPayload | null): string {
private workspaceGitDescriptorStateKey(workspace: WorkspaceDescriptorPayload | null): string {
if (!workspace) {
return WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT;
return WORKSPACE_GIT_WATCH_REMOVED_STATE_KEY;
}
return JSON.stringify([
workspace.name,
@@ -4378,15 +4376,15 @@ export class Session {
if (!target) {
return false;
}
const nextFingerprint = this.workspaceGitDescriptorFingerprint(workspace);
if (target.latestFingerprint === nextFingerprint) {
const nextStateKey = this.workspaceGitDescriptorStateKey(workspace);
if (target.latestDescriptorStateKey === nextStateKey) {
return true;
}
target.latestFingerprint = nextFingerprint;
target.latestDescriptorStateKey = nextStateKey;
return false;
}
private rememberWorkspaceGitWatchFingerprint(
private rememberWorkspaceGitDescriptorState(
workspaceId: string,
workspace: WorkspaceDescriptorPayload | null,
): void {
@@ -4394,29 +4392,20 @@ export class Session {
if (!target) {
return;
}
target.latestFingerprint = this.workspaceGitDescriptorFingerprint(workspace);
target.latestDescriptorStateKey = this.workspaceGitDescriptorStateKey(workspace);
target.lastBranchName = workspace?.name ?? null;
}
private async primeWorkspaceGitWatchFingerprints(
workspaces: Iterable<WorkspaceDescriptorPayload>,
): Promise<void> {
private syncWorkspaceGitObservers(workspaces: Iterable<WorkspaceDescriptorPayload>): void {
for (const workspace of workspaces) {
const persistedWorkspace = await this.workspaceRegistry.get(workspace.id);
if (!persistedWorkspace) {
continue;
}
await this.syncWorkspaceGitWatchTarget(persistedWorkspace.cwd, {
this.syncWorkspaceGitObserver(workspace.workspaceDirectory, {
isGit: workspace.projectKind === "git",
});
this.rememberWorkspaceGitWatchFingerprint(persistedWorkspace.cwd, workspace);
this.rememberWorkspaceGitDescriptorState(workspace.workspaceDirectory, workspace);
}
}
private async syncWorkspaceGitWatchTarget(
cwd: string,
options: { isGit: boolean },
): Promise<void> {
private syncWorkspaceGitObserver(cwd: string, options: { isGit: boolean }): void {
const normalizedCwd = normalizePersistedWorkspaceId(cwd);
if (!options.isGit) {
this.removeWorkspaceGitSubscription(normalizedCwd);
@@ -4427,7 +4416,7 @@ export class Session {
return;
}
const subscription = await this.workspaceGitService.subscribe(
const subscription = this.workspaceGitService.registerWorkspace(
{ cwd: normalizedCwd },
(snapshot) => {
void this.emitWorkspaceUpdateForCwd(normalizedCwd);
@@ -6523,7 +6512,7 @@ export class Session {
this.onBranchChanged(workspaceId, watchTarget.lastBranchName, newBranchName);
}
}
this.rememberWorkspaceGitWatchFingerprint(workspaceId, nextWorkspace);
this.rememberWorkspaceGitDescriptorState(workspaceId, nextWorkspace);
if (!nextWorkspace) {
subscription.lastEmittedByWorkspaceId.delete(workspaceId);
@@ -6694,7 +6683,7 @@ export class Session {
}
const payload = await this.listFetchWorkspacesEntries(request);
await this.primeWorkspaceGitWatchFingerprints(payload.entries);
this.syncWorkspaceGitObservers(payload.entries);
this.sessionLogger.debug(
{
requestId: request.requestId,

21
packages/server/src/server/session.workspace-git-watch.test.ts
View File

@@ -9,6 +9,7 @@ import {
createPersistedProjectRecord,
createPersistedWorkspaceRecord,
} from "./workspace-registry.js";
import { createNoopWorkspaceGitService } from "./test-utils/workspace-git-service-stub.js";
function createWorkspaceRuntimeSnapshot(
cwd: string,
@@ -68,7 +69,7 @@ function createSessionForWorkspaceGitWatchTests(): {
projects: Map<string, ReturnType<typeof createPersistedProjectRecord>>;
workspaces: Map<string, ReturnType<typeof createPersistedWorkspaceRecord>>;
workspaceGitService: WorkspaceGitService & {
subscribe: ReturnType<typeof vi.fn>;
registerWorkspace: ReturnType<typeof vi.fn>;
peekSnapshot: ReturnType<typeof vi.fn>;
getSnapshot: ReturnType<typeof vi.fn>;
refresh: ReturnType<typeof vi.fn>;
@@ -99,7 +100,8 @@ function createSessionForWorkspaceGitWatchTests(): {
error: vi.fn(),
};
const workspaceGitService = {
subscribe: vi.fn(async (params: { cwd: string }, listener: WorkspaceGitListener) => {
...createNoopWorkspaceGitService(),
registerWorkspace: vi.fn((params: { cwd: string }, listener: WorkspaceGitListener) => {
const unsubscribe = vi.fn();
subscriptions.push({
params,
@@ -107,7 +109,6 @@ function createSessionForWorkspaceGitWatchTests(): {
unsubscribe,
});
return {
initial: createWorkspaceRuntimeSnapshot(params.cwd),
unsubscribe,
};
}),
@@ -186,7 +187,7 @@ function createSessionForWorkspaceGitWatchTests(): {
}),
dispose: () => {},
} as any,
workspaceGitService: workspaceGitService as any,
workspaceGitService,
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -200,7 +201,7 @@ function createSessionForWorkspaceGitWatchTests(): {
emitted,
projects,
workspaces,
workspaceGitService: workspaceGitService as any,
workspaceGitService,
subscriptions,
};
}
@@ -275,9 +276,9 @@ describe("workspace git watch targets", () => {
sessionAny.buildWorkspaceDescriptorMap = async () => new Map([[descriptor.id, descriptor]]);
await sessionAny.syncWorkspaceGitWatchTarget("/tmp/repo", { isGit: true });
sessionAny.syncWorkspaceGitObserver("/tmp/repo", { isGit: true });
expect(workspaceGitService.subscribe).toHaveBeenCalledWith(
expect(workspaceGitService.registerWorkspace).toHaveBeenCalledWith(
{ cwd: "/tmp/repo" },
expect.any(Function),
);
@@ -334,7 +335,7 @@ describe("workspace git watch targets", () => {
lastEmittedByWorkspaceId: new Map(),
};
await sessionAny.syncWorkspaceGitWatchTarget("/tmp/repo", { isGit: true });
sessionAny.syncWorkspaceGitObserver("/tmp/repo", { isGit: true });
emitted.length = 0;
subscriptions[0]?.listener(
@@ -369,7 +370,7 @@ describe("workspace git watch targets", () => {
error: null,
requestId: "subscription:/tmp/repo",
});
expect(workspaceGitService.subscribe).toHaveBeenCalledWith(
expect(workspaceGitService.registerWorkspace).toHaveBeenCalledWith(
{ cwd: "/tmp/repo" },
expect.any(Function),
);
@@ -397,7 +398,7 @@ describe("workspace git watch targets", () => {
lastEmittedByWorkspaceId: new Map(),
};
await sessionAny.syncWorkspaceGitWatchTarget("/tmp/repo", { isGit: true });
sessionAny.syncWorkspaceGitObserver("/tmp/repo", { isGit: true });
emitted.length = 0;
subscriptions[0]?.listener(

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

@@ -10,6 +10,7 @@ import type {
SessionOutboundMessage,
} from "../shared/messages.js";
import type { WorkspaceGitRuntimeSnapshot } from "./workspace-git-service.js";
import { createNoopWorkspaceGitService } from "./test-utils/workspace-git-service-stub.js";
import {
createPersistedProjectRecord,
createPersistedWorkspaceRecord,
@@ -112,67 +113,6 @@ function agentIdsFromEntries(entries: Array<{ agent: Pick<AgentSnapshotPayload,
return entries.map((entry) => entry.agent.id);
}
function createNoopWorkspaceGitService() {
return {
subscribe: async (params: { cwd: string }) => ({
initial: {
cwd: params.cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
},
unsubscribe: () => {},
}),
peekSnapshot: (_cwd: string) => null,
getSnapshot: async (cwd: string) => ({
cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
}),
resolveRepoRemoteUrl: async () => null,
resolveRepoRoot: async (cwd: string) => cwd,
resolveDefaultBranch: async () => "main",
refresh: async () => {},
requestWorkingTreeWatch: async (cwd: string) => ({
repoRoot: cwd,
unsubscribe: () => {},
}),
scheduleRefreshForCwd: () => {},
dispose: () => {},
};
}
function createWorkspaceRuntimeSnapshot(
cwd: string,
overrides?: {
@@ -301,7 +241,7 @@ function createSessionForWorkspaceTests(
}),
dispose: () => {},
} as any,
workspaceGitService: (options.workspaceGitService ?? createNoopWorkspaceGitService()) as any,
workspaceGitService: options.workspaceGitService ?? createNoopWorkspaceGitService(),
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -593,7 +533,7 @@ describe("workspace aggregation", () => {
}),
dispose: () => {},
} as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
workspaceGitService: createNoopWorkspaceGitService(),
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -745,7 +685,7 @@ describe("workspace aggregation", () => {
}),
dispose: () => {},
} as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
workspaceGitService: createNoopWorkspaceGitService(),
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -924,7 +864,7 @@ describe("workspace aggregation", () => {
}),
dispose: () => {},
} as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
workspaceGitService: createNoopWorkspaceGitService(),
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -1066,7 +1006,7 @@ describe("workspace aggregation", () => {
}),
dispose: () => {},
} as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
workspaceGitService: createNoopWorkspaceGitService(),
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -1683,7 +1623,7 @@ describe("workspace aggregation", () => {
}),
dispose: () => {},
} as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
workspaceGitService: createNoopWorkspaceGitService(),
mcpBaseUrl: null,
stt: null,
tts: null,
@@ -2646,8 +2586,7 @@ describe("workspace aggregation", () => {
});
const workspaceGitService = createNoopWorkspaceGitService();
workspaceGitService.peekSnapshot = vi.fn(() => runtimeSnapshot);
workspaceGitService.subscribe = vi.fn(async () => ({
initial: runtimeSnapshot,
workspaceGitService.registerWorkspace = vi.fn(() => ({
unsubscribe: () => {},
}));
@@ -2728,6 +2667,64 @@ describe("workspace aggregation", () => {
]);
});
test("fetch_workspaces_response emits before cold registration-triggered git work starts", async () => {
const events: string[] = [];
const emitted: Array<{ type: string; payload: any }> = [];
const workspaceGitService = createNoopWorkspaceGitService();
const getSnapshot = vi.fn(async (cwd: string) => {
events.push(`git:${cwd}`);
return createWorkspaceRuntimeSnapshot(cwd);
});
workspaceGitService.getSnapshot = getSnapshot;
workspaceGitService.registerWorkspace = vi.fn((params: { cwd: string }) => {
queueMicrotask(() => {
void getSnapshot(params.cwd);
});
return {
unsubscribe: () => {},
};
});
const session = createSessionForWorkspaceTests({
workspaceGitService,
}) as any;
const project = createPersistedProjectRecord({
projectId: "proj-fetch-boundary",
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: "ws-fetch-boundary",
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.emit = (message: any) => {
if (message.type === "fetch_workspaces_response") {
events.push("response");
}
emitted.push(message);
};
session.listAgentPayloads = async () => [];
session.projectRegistry.list = async () => [project];
session.workspaceRegistry.list = async () => [workspace];
await session.handleMessage({
type: "fetch_workspaces_request",
requestId: "req-fetch-workspaces-boundary",
subscribe: {},
});
expect(emitted.find((message) => message.type === "fetch_workspaces_response")).toBeDefined();
expect(events[0]).toBe("response");
});
test("workspace_update includes updated runtime fields", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const runtimeSnapshot = createWorkspaceRuntimeSnapshot("/tmp/repo", {
@@ -2865,6 +2862,7 @@ describe("workspace aggregation", () => {
projectId: gitProject.projectId,
projectDisplayName: gitProject.displayName,
projectRootPath: gitProject.rootPath,
workspaceDirectory: gitWorkspace.cwd,
projectKind: gitProject.kind,
workspaceKind: gitWorkspace.kind,
name: "main",
@@ -2881,6 +2879,7 @@ describe("workspace aggregation", () => {
projectId: directoryProject.projectId,
projectDisplayName: directoryProject.displayName,
projectRootPath: directoryProject.rootPath,
workspaceDirectory: directoryWorkspace.cwd,
projectKind: directoryProject.kind,
workspaceKind: directoryWorkspace.kind,
name: "docs",

76
packages/server/src/server/test-utils/workspace-git-service-stub.ts Normal file
View File

@@ -0,0 +1,76 @@
import { basename } from "node:path";
import type { CheckoutDiffResult } from "../../utils/checkout-git.js";
import {
buildWorkspaceGitMetadataFromSnapshot,
type WorkspaceGitMetadata,
} from "../workspace-git-metadata.js";
import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "../workspace-git-service.js";
export function createNoGitWorkspaceRuntimeSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot {
return {
cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
baseRef: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
hasRemote: false,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
};
}
export function createNoopWorkspaceGitService(
overrides: Partial<WorkspaceGitService> = {},
): WorkspaceGitService {
const service: WorkspaceGitService = {
registerWorkspace: () => ({
unsubscribe: () => {},
}),
peekSnapshot: () => null,
getSnapshot: async (cwd: string) => createNoGitWorkspaceRuntimeSnapshot(cwd),
getCheckoutDiff: async (): Promise<CheckoutDiffResult> => ({ diff: "" }),
validateBranchRef: async () => ({ kind: "not-found" }),
hasLocalBranch: async () => false,
suggestBranchesForCwd: async () => [],
listStashes: async () => [],
listWorktrees: async () => [],
getWorkspaceGitMetadata: async (cwd: string, options): Promise<WorkspaceGitMetadata> => {
const snapshot = createNoGitWorkspaceRuntimeSnapshot(cwd);
return buildWorkspaceGitMetadataFromSnapshot({
cwd,
directoryName: options?.directoryName ?? basename(cwd),
isGit: snapshot.git.isGit,
repoRoot: snapshot.git.repoRoot,
mainRepoRoot: snapshot.git.mainRepoRoot,
currentBranch: snapshot.git.currentBranch,
remoteUrl: snapshot.git.remoteUrl,
});
},
resolveRepoRoot: async (cwd: string) => cwd,
resolveDefaultBranch: async () => "main",
resolveRepoRemoteUrl: async () => null,
refresh: async () => {},
requestWorkingTreeWatch: async () => ({
repoRoot: null,
unsubscribe: () => {},
}),
scheduleRefreshForCwd: () => {},
dispose: () => {},
...overrides,
};
return service;
}

109
packages/server/src/server/websocket-server.ts
View File

@@ -1,6 +1,6 @@
import { WebSocketServer } from "ws";
import type { Server as HTTPServer } from "http";
import { join } from "path";
import { basename, join } from "path";
import { hostname as getHostname } from "node:os";
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentStorage } from "./agent/agent-storage.js";
@@ -35,7 +35,8 @@ import type {
} from "./agent/provider-launch-config.js";
import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
import { buildProviderRegistry } from "./agent/provider-registry.js";
import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";
import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
import { buildWorkspaceGitMetadataFromSnapshot } from "./workspace-git-metadata.js";
import { PushTokenStore } from "./push/token-store.js";
import { PushService } from "./push/push-service.js";
import type { ScriptHealthState } from "./script-health-monitor.js";
@@ -67,62 +68,68 @@ type WebSocketServerConfig = {
type WebSocketRuntimeMetrics = SessionRuntimeMetrics & CheckoutDiffMetrics;
function createFallbackWorkspaceGitService(): WorkspaceGitServiceImpl {
function createFallbackWorkspaceGitSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot {
return {
subscribe: async ({ cwd }: { cwd: string }) => ({
initial: {
cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
},
cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
baseRef: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
hasRemote: false,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
};
}
function createFallbackWorkspaceGitService(): WorkspaceGitService {
return {
registerWorkspace: () => ({
unsubscribe: () => {},
}),
peekSnapshot: () => null,
getSnapshot: async (cwd: string) => ({
cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
}),
getSnapshot: async (cwd: string) => createFallbackWorkspaceGitSnapshot(cwd),
getCheckoutDiff: async () => ({ diff: "" }),
validateBranchRef: async () => ({ kind: "not-found" }),
hasLocalBranch: async () => false,
suggestBranchesForCwd: async () => [],
listStashes: async () => [],
listWorktrees: async () => [],
getWorkspaceGitMetadata: async (cwd: string, options) => {
const snapshot = createFallbackWorkspaceGitSnapshot(cwd);
return buildWorkspaceGitMetadataFromSnapshot({
cwd,
directoryName: options?.directoryName ?? basename(cwd),
isGit: snapshot.git.isGit,
repoRoot: snapshot.git.repoRoot,
mainRepoRoot: snapshot.git.mainRepoRoot,
currentBranch: snapshot.git.currentBranch,
remoteUrl: snapshot.git.remoteUrl,
});
},
resolveRepoRoot: async (cwd: string) => cwd,
resolveDefaultBranch: async () => "main",
resolveRepoRemoteUrl: async () => null,
refresh: async () => {},
requestWorkingTreeWatch: async (cwd: string) => ({
repoRoot: cwd,
requestWorkingTreeWatch: async () => ({
repoRoot: null,
unsubscribe: () => {},
}),
scheduleRefreshForCwd: () => {},
dispose: () => {},
} as unknown as WorkspaceGitServiceImpl;
};
}
function createNoopProjectRegistry(): ProjectRegistry {
@@ -297,7 +304,7 @@ export class VoiceAssistantWebSocketServer {
private readonly scheduleService: ScheduleService;
private readonly checkoutDiffManager: CheckoutDiffManager;
private readonly github: GitHubService;
private readonly workspaceGitService: WorkspaceGitServiceImpl;
private readonly workspaceGitService: WorkspaceGitService;
private readonly downloadTokenStore: DownloadTokenStore;
private readonly paseoHome: string;
private readonly daemonConfigStore: DaemonConfigStore;
@@ -394,7 +401,7 @@ export class VoiceAssistantWebSocketServer {
getDaemonTcpPort?: () => number | null,
getDaemonTcpHost?: () => string | null,
resolveScriptHealth?: (hostname: string) => ScriptHealthState | null,
workspaceGitService?: WorkspaceGitServiceImpl,
workspaceGitService?: WorkspaceGitService,
github?: GitHubService,
) {
this.logger = logger.child({ module: "websocket-server" });

157
packages/server/src/server/workspace-git-service.primitive.test.ts
View File

@@ -295,6 +295,33 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
service.dispose();
});
test("registerWorkspace returns a subscription without waiting for a cold snapshot", async () => {
const checkoutStatusDeferred = createDeferred<CheckoutStatusGit>();
const getCheckoutStatus = vi.fn(async () => checkoutStatusDeferred.promise);
const service = createService({ getCheckoutStatus });
const listener = vi.fn();
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
expect(subscription).toEqual({ unsubscribe: expect.any(Function) });
expect(getCheckoutStatus).not.toHaveBeenCalled();
expect(listener).not.toHaveBeenCalled();
expect(service.peekSnapshot("/tmp/repo")).toBeNull();
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(service.peekSnapshot("/tmp/repo")).toBeNull();
checkoutStatusDeferred.resolve(createCheckoutStatus("/tmp/repo"));
await expect(service.getSnapshot("/tmp/repo")).resolves.toEqual(createSnapshot("/tmp/repo"));
expect(service.peekSnapshot("/tmp/repo")).toEqual(createSnapshot("/tmp/repo"));
subscription.unsubscribe();
service.dispose();
});
test("forced getSnapshot bypasses the internal min-gap and re-shells", async () => {
let nowMs = 0;
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
@@ -315,8 +342,10 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
test("forced getSnapshot emits even when the fingerprint matches", async () => {
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({ getCheckoutStatus });
await service.getSnapshot("/tmp/repo");
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
await service.getSnapshot("/tmp/repo", { force: true, reason: "test" });
@@ -336,8 +365,10 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getCheckoutStatus,
now: () => new Date(nowMs),
});
await service.getSnapshot("/tmp/repo");
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
nowMs = 3_000;
await service.refresh("/tmp/repo");
@@ -566,7 +597,8 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getPullRequestStatus,
now: () => new Date(nowMs),
});
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
@@ -582,6 +614,103 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
service.dispose();
});
test("self-heal retries workspace observation setup while a listener remains active", async () => {
let nowMs = 0;
const resolveAbsoluteGitDir = vi
.fn<() => Promise<string | null>>()
.mockRejectedValueOnce(new Error("git dir temporarily unavailable"))
.mockResolvedValue("/tmp/repo/.git");
const watch = vi.fn(() => createWatcher() as never);
const service = createService({
resolveAbsoluteGitDir,
watch,
now: () => new Date(nowMs),
});
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(1);
expect(watch).not.toHaveBeenCalled();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
await flushPromises();
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(2);
expect(resolveAbsoluteGitDir).toHaveBeenLastCalledWith("/tmp/repo");
subscription.unsubscribe();
service.dispose();
});
test("stale workspace watcher callbacks do not refresh after unsubscribe", async () => {
const watchCallbacks: Array<() => void> = [];
const watch = vi.fn(
(_watchPath: string, _options: { recursive: boolean }, callback: () => void) => {
watchCallbacks.push(callback);
return createWatcher() as never;
},
);
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({
getCheckoutStatus,
resolveAbsoluteGitDir: vi.fn(async () => "/tmp/repo/.git"),
watch,
});
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
await vi.waitFor(() => {
expect(watchCallbacks.length).toBeGreaterThan(0);
});
const callsBeforeStaleCallback = getCheckoutStatus.mock.calls.length;
subscription.unsubscribe();
watchCallbacks[0]?.();
await vi.advanceTimersByTimeAsync(500);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(callsBeforeStaleCallback);
service.dispose();
});
test("stale GitHub poll callbacks do not refresh after unsubscribe", async () => {
let pollStatus: (() => void) | null = null;
const pollUnsubscribe = vi.fn();
const github = {
...createGitHubServiceStub(),
retainCurrentPullRequestStatusPoll: vi.fn((options: { onStatus: () => void }) => {
pollStatus = options.onStatus;
return { unsubscribe: pollUnsubscribe };
}),
};
const getCheckoutStatus = vi.fn(async (cwd: string) => createCheckoutStatus(cwd));
const service = createService({
getCheckoutStatus,
github,
});
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
await vi.waitFor(() => {
expect(github.retainCurrentPullRequestStatusPoll).toHaveBeenCalledTimes(1);
});
const callsBeforeStaleCallback = getCheckoutStatus.mock.calls.length;
subscription.unsubscribe();
pollStatus?.();
await flushPromises();
expect(pollUnsubscribe).toHaveBeenCalledTimes(1);
expect(getCheckoutStatus).toHaveBeenCalledTimes(callsBeforeStaleCallback);
service.dispose();
});
test("subscription starts GitHub self-heal reads within the fast poll window", async () => {
let nowMs = 0;
const githubReadCalls: Array<{ reason: string | undefined; tickMs: number }> = [];
@@ -611,7 +740,8 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
github,
now: () => new Date(nowMs),
});
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
nowMs = 20_000;
await vi.advanceTimersByTimeAsync(20_000);
@@ -643,7 +773,8 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getCheckoutStatus,
github,
});
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
expect(retainCurrentPullRequestStatusPoll).not.toHaveBeenCalled();
@@ -658,8 +789,9 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getCheckoutStatus,
now: () => new Date(nowMs),
});
const first = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const second = await service.subscribe({ cwd: "/tmp/repo/." }, vi.fn());
const first = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
const second = service.registerWorkspace({ cwd: "/tmp/repo/." }, vi.fn());
await flushPromises();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
@@ -679,14 +811,14 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getCheckoutStatus,
now: () => new Date(nowMs),
});
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
subscription.unsubscribe();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(getCheckoutStatus).toHaveBeenCalledTimes(0);
service.dispose();
});
@@ -698,14 +830,14 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getCheckoutStatus,
now: () => new Date(nowMs),
});
await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
service.dispose();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);
await flushPromises();
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(getCheckoutStatus).toHaveBeenCalledTimes(0);
});
test("self-heal poll coalesces with a concurrent direct getSnapshot call", async () => {
@@ -719,7 +851,8 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
getCheckoutStatus,
now: () => new Date(nowMs),
});
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
nowMs = 60_000;
await vi.advanceTimersByTimeAsync(60_000);

65
packages/server/src/server/workspace-git-service.test.ts
View File

@@ -223,14 +223,16 @@ describe("WorkspaceGitServiceImpl", () => {
vi.useRealTimers();
});
test("subscribe returns an initial workspace runtime snapshot", async () => {
test("registerWorkspace returns a subscription without an initial snapshot contract", async () => {
const service = createService();
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
expect(subscription.initial).toEqual(createSnapshot("/tmp/repo"));
expect(subscription).toEqual({ unsubscribe: expect.any(Function) });
expect("initial" in subscription).toBe(false);
expect(listener).not.toHaveBeenCalled();
expect(service.peekSnapshot("/tmp/repo")).toBeNull();
subscription.unsubscribe();
service.dispose();
@@ -282,7 +284,8 @@ describe("WorkspaceGitServiceImpl", () => {
now: () => new Date(nowMs),
});
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
await service.getSnapshot("/tmp/repo");
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
nowMs += 3_000;
await service.refresh("/tmp/repo");
@@ -294,7 +297,7 @@ describe("WorkspaceGitServiceImpl", () => {
service.dispose();
});
test("cold getSnapshot calls share one workspace target setup and cache the snapshot", async () => {
test("cold getSnapshot calls share one workspace target and cache the snapshot", async () => {
const checkoutStatusDeferred = createDeferred<CheckoutStatusGit>();
const getCheckoutStatus = vi.fn(async () => checkoutStatusDeferred.promise);
const getPullRequestStatus = vi.fn(async () => createPullRequestStatusResult());
@@ -313,8 +316,6 @@ describe("WorkspaceGitServiceImpl", () => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(getPullRequestStatus).toHaveBeenCalledTimes(0);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(0);
expect((service as any).workspaceTargets.size).toBe(1);
expect((service as any).workspaceTargetSetups.size).toBe(1);
checkoutStatusDeferred.resolve(createCheckoutStatus("/tmp/repo"));
@@ -325,8 +326,7 @@ describe("WorkspaceGitServiceImpl", () => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(getPullRequestStatus).toHaveBeenCalledTimes(1);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(1);
expect((service as any).workspaceTargets.size).toBe(1);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(0);
expect(service.peekSnapshot("/tmp/repo")).toEqual(createSnapshot("/tmp/repo"));
await expect(service.getSnapshot("/tmp/repo")).resolves.toEqual(createSnapshot("/tmp/repo"));
@@ -347,14 +347,12 @@ describe("WorkspaceGitServiceImpl", () => {
now: () => new Date(nowMs),
});
const [first, second] = await Promise.all([
service.subscribe({ cwd: "/tmp/repo" }, vi.fn()),
service.subscribe({ cwd: "/tmp/repo" }, vi.fn()),
]);
const first = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
const second = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
expect(getPullRequestStatus).toHaveBeenCalledTimes(1);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(1);
expect((service as any).workspaceTargets.size).toBe(1);
first.unsubscribe();
second.unsubscribe();
@@ -372,9 +370,9 @@ describe("WorkspaceGitServiceImpl", () => {
now: () => new Date(nowMs),
});
const subscription = await service.subscribe({ cwd: "/tmp/repo/." }, vi.fn());
const subscription = service.registerWorkspace({ cwd: "/tmp/repo/." }, vi.fn());
expect(subscription.initial).toEqual(createSnapshot("/tmp/repo"));
await expect(service.getSnapshot("/tmp/repo/.")).resolves.toEqual(createSnapshot("/tmp/repo"));
expect(service.peekSnapshot("/tmp/repo")).toEqual(createSnapshot("/tmp/repo"));
nowMs += 3_000;
@@ -383,7 +381,6 @@ describe("WorkspaceGitServiceImpl", () => {
expect(getPullRequestStatus).toHaveBeenCalledTimes(2);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(1);
expect((service as any).workspaceTargets.size).toBe(1);
subscription.unsubscribe();
service.dispose();
@@ -392,20 +389,23 @@ describe("WorkspaceGitServiceImpl", () => {
test("repo-level fetch intervals are shared for workspaces in the same repo", async () => {
const runGitFetch = vi.fn(async () => {});
const hasOriginRemote = vi.fn(async () => true);
const resolveAbsoluteGitDir = vi.fn(async () => "/tmp/repo/.git");
const service = createService({
resolveAbsoluteGitDir: vi.fn(async () => "/tmp/repo/.git"),
resolveAbsoluteGitDir,
hasOriginRemote,
runGitFetch,
});
const first = await service.subscribe({ cwd: "/tmp/repo" }, vi.fn());
const second = await service.subscribe({ cwd: "/tmp/repo/packages/server" }, vi.fn());
await flushPromises();
const first = service.registerWorkspace({ cwd: "/tmp/repo" }, vi.fn());
const second = service.registerWorkspace({ cwd: "/tmp/repo/packages/server" }, vi.fn());
await vi.waitFor(() => {
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(2);
expect(runGitFetch).toHaveBeenCalledTimes(1);
});
expect(hasOriginRemote).toHaveBeenCalledTimes(1);
expect(runGitFetch).toHaveBeenCalledTimes(1);
expect((service as any).repoTargets.size).toBe(1);
await vi.advanceTimersByTimeAsync(180_000);
await flushPromises();
@@ -452,9 +452,10 @@ describe("WorkspaceGitServiceImpl", () => {
});
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
const initialSnapshot = await service.getSnapshot("/tmp/repo");
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
expect(subscription.initial.github.pullRequest?.title).toBe("Before refresh");
expect(initialSnapshot.github.pullRequest?.title).toBe("Before refresh");
await service.refresh("/tmp/repo");
await flushPromises();
@@ -521,9 +522,10 @@ describe("WorkspaceGitServiceImpl", () => {
});
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
const initialSnapshot = await service.getSnapshot("/tmp/repo");
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
expect(subscription.initial.git.currentBranch).toBe("main");
expect(initialSnapshot.git.currentBranch).toBe("main");
nowMs += 3_000;
await service.refresh("/tmp/repo");
@@ -564,7 +566,8 @@ describe("WorkspaceGitServiceImpl", () => {
});
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
await service.getSnapshot("/tmp/repo");
const subscription = service.registerWorkspace({ cwd: "/tmp/repo" }, listener);
await service.getSnapshot("/tmp/repo", {
force: true,
@@ -765,10 +768,14 @@ describe("WorkspaceGitServiceImpl", () => {
const service = createService({ getCheckoutShortstat, watch });
const workspaceListener = vi.fn();
const workspaceSubscription = await service.subscribe({ cwd: "/tmp/repo" }, workspaceListener);
const initialSnapshot = await service.getSnapshot("/tmp/repo");
const workspaceSubscription = service.registerWorkspace(
{ cwd: "/tmp/repo" },
workspaceListener,
);
const diffSubscription = await service.requestWorkingTreeWatch("/tmp/repo", vi.fn());
expect(workspaceSubscription.initial.git.diffStat).toEqual({ additions: 1, deletions: 0 });
expect(initialSnapshot.git.diffStat).toEqual({ additions: 1, deletions: 0 });
const repoRootWatch = watchCallbacks.find((entry) => entry.path === "/tmp/repo");
expect(repoRootWatch).toBeDefined();

151
packages/server/src/server/workspace-git-service.ts
View File

@@ -83,13 +83,10 @@ export type WorkspaceGitRuntimeSnapshot = {
};
export interface WorkspaceGitService {
subscribe(
registerWorkspace(
params: { cwd: string },
listener: WorkspaceGitListener,
): Promise<{
initial: WorkspaceGitRuntimeSnapshot;
unsubscribe: () => void;
}>;
): WorkspaceGitSubscription;
peekSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot | null;
getSnapshot(
@@ -139,6 +136,10 @@ export interface WorkspaceGitService {
export type WorkspaceGitListener = (snapshot: WorkspaceGitRuntimeSnapshot) => void;
export type WorkspaceGitSubscription = {
unsubscribe: () => void;
};
export type WorkspaceGitReadOptions =
| {
force?: false;
@@ -246,6 +247,9 @@ interface WorkspaceGitTarget {
latestFingerprint: string | null;
lastShellOutAtMs: number | null;
repoGitRoot: string | null;
observationSetupPromise: Promise<void> | null;
observationSetupComplete: boolean;
closed: boolean;
}
interface RepoGitTarget {
@@ -281,7 +285,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private readonly deps: WorkspaceGitServiceDependencies;
private readonly workspaceTargets = new Map<string, WorkspaceGitTarget>();
private readonly repoTargets = new Map<string, RepoGitTarget>();
private readonly workspaceTargetSetups = new Map<string, Promise<WorkspaceGitTarget>>();
private readonly workingTreeWatchTargets = new Map<string, WorkingTreeWatchTarget>();
private readonly workingTreeWatchSetups = new Map<string, Promise<WorkingTreeWatchTarget>>();
private readonly branchValidationCache = new Map<
@@ -337,22 +340,22 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
};
}
async subscribe(
registerWorkspace(
params: { cwd: string },
listener: WorkspaceGitListener,
): Promise<{
initial: WorkspaceGitRuntimeSnapshot;
unsubscribe: () => void;
}> {
): WorkspaceGitSubscription {
const cwd = normalizeWorkspaceId(params.cwd);
const target = await this.ensureWorkspaceTarget(cwd);
const target = this.ensureWorkspaceTarget(cwd);
target.listeners.add(listener);
if (target.listeners.size === 1) {
this.startWorkspaceSubscriptionTimers(target);
}
if (!target.latestSnapshot) {
this.scheduleInitialWorkspaceRefresh(target);
}
this.scheduleWorkspaceObservationSetup(target);
return {
initial: target.latestSnapshot ?? (await this.getSnapshot(cwd)),
unsubscribe: () => {
this.removeWorkspaceListener(cwd, listener);
},
@@ -365,7 +368,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
): Promise<WorkspaceGitRuntimeSnapshot> {
cwd = normalizeWorkspaceId(cwd);
const request = this.normalizeRefreshRequest(options, "getSnapshot", true);
const target = await this.ensureWorkspaceTarget(cwd);
const target = this.ensureWorkspaceTarget(cwd);
if (!request.force && this.isSnapshotWarm(target)) {
return target.latestSnapshot;
}
@@ -544,18 +547,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
async refresh(cwd: string, _options?: { priority?: "normal" | "high" }): Promise<void> {
cwd = normalizeWorkspaceId(cwd);
const target = this.workspaceTargets.get(cwd);
if (target) {
await this.refreshWorkspaceTarget(target, {
force: false,
includeGitHub: false,
reason: "refresh",
notify: true,
});
return;
}
await this.ensureWorkspaceTarget(cwd);
const target = this.ensureWorkspaceTarget(cwd);
await this.refreshWorkspaceTarget(target, {
force: false,
includeGitHub: false,
reason: "refresh",
notify: true,
});
this.scheduleWorkspaceObservationSetup(target);
}
async requestWorkingTreeWatch(
@@ -587,7 +586,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
this.closeWorkspaceTarget(target);
}
this.workspaceTargets.clear();
this.workspaceTargetSetups.clear();
for (const target of this.repoTargets.values()) {
this.closeRepoTarget(target);
@@ -601,22 +599,13 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
this.workingTreeWatchSetups.clear();
}
private async ensureWorkspaceTarget(cwd: string): Promise<WorkspaceGitTarget> {
private ensureWorkspaceTarget(cwd: string): WorkspaceGitTarget {
const existingTarget = this.workspaceTargets.get(cwd);
if (existingTarget) {
return existingTarget;
}
const existingSetup = this.workspaceTargetSetups.get(cwd);
if (existingSetup) {
return existingSetup;
}
const setup = this.createWorkspaceTarget(cwd).finally(() => {
this.workspaceTargetSetups.delete(cwd);
});
this.workspaceTargetSetups.set(cwd, setup);
return setup;
return this.createWorkspaceTarget(cwd);
}
private readAuxiliaryCache<T>(
@@ -698,7 +687,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return setup;
}
private async createWorkspaceTarget(cwd: string): Promise<WorkspaceGitTarget> {
private createWorkspaceTarget(cwd: string): WorkspaceGitTarget {
const target: WorkspaceGitTarget = {
cwd,
listeners: new Set(),
@@ -713,35 +702,85 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
latestFingerprint: null,
lastShellOutAtMs: null,
repoGitRoot: null,
observationSetupPromise: null,
observationSetupComplete: false,
closed: false,
};
this.workspaceTargets.set(cwd, target);
return target;
}
try {
await this.requestWorkspaceSnapshot(target, {
private scheduleInitialWorkspaceRefresh(target: WorkspaceGitTarget): void {
queueMicrotask(() => {
if (!this.isActiveObservedWorkspaceTarget(target) || target.latestSnapshot) {
return;
}
void this.refreshWorkspaceTarget(target, {
force: false,
includeGitHub: true,
reason: "initial",
notify: false,
notify: true,
});
});
}
const gitDir = await this.deps.resolveAbsoluteGitDir(cwd);
private scheduleWorkspaceObservationSetup(target: WorkspaceGitTarget): void {
if (
target.observationSetupComplete ||
target.observationSetupPromise ||
!this.isActiveObservedWorkspaceTarget(target)
) {
return;
}
target.observationSetupPromise = Promise.resolve()
.then(() => this.setupWorkspaceObservation(target))
.catch((error) => {
this.logger.warn(
{ err: error, cwd: target.cwd },
"Failed to set up workspace git observation",
);
})
.finally(() => {
target.observationSetupPromise = null;
});
}
private async setupWorkspaceObservation(target: WorkspaceGitTarget): Promise<void> {
try {
const gitDir = await this.deps.resolveAbsoluteGitDir(target.cwd);
if (!this.isActiveObservedWorkspaceTarget(target)) {
return;
}
if (!gitDir) {
return target;
target.observationSetupComplete = true;
return;
}
const repoGitRoot = await this.resolveWorkspaceGitRefsRoot(gitDir);
if (!this.isActiveObservedWorkspaceTarget(target)) {
return;
}
target.repoGitRoot = repoGitRoot;
this.startWorkspaceWatchers(target, gitDir, repoGitRoot);
await this.ensureRepoTarget(target);
return target;
if (this.isActiveObservedWorkspaceTarget(target)) {
target.observationSetupComplete = true;
}
} catch (error) {
this.closeWorkspaceTarget(target);
this.workspaceTargets.delete(cwd);
throw error;
}
}
private isActiveObservedWorkspaceTarget(target: WorkspaceGitTarget): boolean {
return (
!target.closed &&
target.listeners.size > 0 &&
this.workspaceTargets.get(target.cwd) === target
);
}
private async createWorkingTreeWatchTarget(cwd: string): Promise<WorkingTreeWatchTarget> {
const repoRoot = await this.resolveCheckoutWatchRoot(cwd);
const target: WorkingTreeWatchTarget = {
@@ -862,7 +901,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private async ensureRepoTarget(workspaceTarget: WorkspaceGitTarget): Promise<void> {
const repoGitRoot = workspaceTarget.repoGitRoot;
if (!repoGitRoot) {
if (!repoGitRoot || !this.isActiveObservedWorkspaceTarget(workspaceTarget)) {
return;
}
@@ -873,6 +912,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
const hasOrigin = await this.deps.hasOriginRemote(workspaceTarget.cwd);
if (!this.isActiveObservedWorkspaceTarget(workspaceTarget)) {
return;
}
if (!hasOrigin) {
return;
}
@@ -904,7 +946,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
typeof targetOrCwd === "string"
? this.workspaceTargets.get(normalizeWorkspaceId(targetOrCwd))
: targetOrCwd;
if (!target) {
if (!target || target.closed || this.workspaceTargets.get(target.cwd) !== target) {
return;
}
@@ -913,6 +955,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
target.debounceTimer = setTimeout(() => {
if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
return;
}
target.debounceTimer = null;
void this.refreshWorkspaceTarget(target, {
force: options?.force === true,
@@ -926,6 +971,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private startWorkspaceSubscriptionTimers(target: WorkspaceGitTarget): void {
if (!target.selfHealTimer) {
target.selfHealTimer = setInterval(() => {
this.scheduleWorkspaceObservationSetup(target);
this.getSnapshot(target.cwd, { reason: "self-heal-git" }).catch((error) => {
this.logger.warn(
{ err: error, cwd: target.cwd, reason: "self-heal-git" },
@@ -965,6 +1011,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
cwd: target.cwd,
headRef,
onStatus: () => {
if (!this.isActiveObservedWorkspaceTarget(target)) {
return;
}
void this.refreshWorkspaceTarget(target, {
force: false,
includeGitHub: true,
@@ -1137,6 +1186,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
target: WorkspaceGitTarget,
request: WorkspaceGitRefreshRequest,
): Promise<void> {
if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
return;
}
try {
await this.requestWorkspaceSnapshot(target, request);
} catch (error) {
@@ -1397,6 +1449,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
private closeWorkspaceTarget(target: WorkspaceGitTarget): void {
target.closed = true;
if (target.debounceTimer) {
clearTimeout(target.debounceTimer);
target.debounceTimer = null;

60
packages/server/src/server/workspace-registry-bootstrap.test.ts
View File

@@ -6,69 +6,11 @@ import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { createTestLogger } from "../test-utils/test-logger.js";
import { AgentStorage } from "./agent/agent-storage.js";
import { createNoopWorkspaceGitService } from "./test-utils/workspace-git-service-stub.js";
import type { WorkspaceGitService } from "./workspace-git-service.js";
import { FileBackedProjectRegistry, FileBackedWorkspaceRegistry } from "./workspace-registry.js";
import { bootstrapWorkspaceRegistries } from "./workspace-registry-bootstrap.js";
function createNoopWorkspaceGitService(): WorkspaceGitService {
return {
subscribe: async (params) => ({
initial: {
cwd: params.cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
},
unsubscribe: () => {},
}),
peekSnapshot: () => null,
getSnapshot: async (cwd) => ({
cwd,
git: {
isGit: false,
repoRoot: null,
mainRepoRoot: null,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
diffStat: null,
},
github: {
featuresEnabled: false,
pullRequest: null,
error: null,
},
}),
resolveRepoRemoteUrl: async () => null,
refresh: async () => {},
requestWorkingTreeWatch: async () => ({
repoRoot: null,
unsubscribe: () => {},
}),
scheduleRefreshForCwd: () => {},
dispose: () => {},
};
}
describe("bootstrapWorkspaceRegistries", () => {
let tmpDir: string;
let paseoHome: string;