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

Centralize git/GitHub runtime state into WorkspaceGitService

Browse Source 
Consolidate workspace git watch lifecycle, background fetch scheduling,
and GitHub PR polling from session.ts and BackgroundGitFetchManager into
a single daemon-scoped WorkspaceGitService. Sessions now subscribe to
the service for push-driven updates instead of coordinating watchers and
fetch intervals themselves. Workspace payloads gain optional gitRuntime
and githubRuntime fields for richer client-side display.
This commit is contained in:
Mohamed Boudra
2026-04-12 10:38:50 +07:00
parent f35b0b8b7d
commit 40e27f5a00
12 changed files with 1871 additions and 1001 deletions
Download Patch File Download Diff File Expand all files Collapse all files

248
packages/server/src/server/background-git-fetch-manager.test.ts
View File

@@ -1,248 +0,0 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
const execFileMock = vi.hoisted(() =>
vi.fn(
(
_file: string,
_args: string[],
_options: unknown,
callback: (error: Error | null, stdout?: string, stderr?: string) => void,
) => {
callback(null, "", "");
},
),
);
vi.mock("node:child_process", async () => {
const actual = await vi.importActual<typeof import("node:child_process")>("node:child_process");
return {
...actual,
execFile: execFileMock,
};
});
import { BackgroundGitFetchManager } from "./background-git-fetch-manager.js";
async function flushPromises(): Promise<void> {
await Promise.resolve();
await Promise.resolve();
}
function createLogger() {
const logger = {
child: () => logger,
debug: vi.fn(),
warn: vi.fn(),
};
return logger;
}
describe("BackgroundGitFetchManager", () => {
beforeEach(() => {
vi.useFakeTimers();
execFileMock.mockReset();
execFileMock.mockImplementation(
(
_file: string,
_args: string[],
_options: unknown,
callback: (error: Error | null, stdout?: string, stderr?: string) => void,
) => {
callback(null, "", "");
},
);
});
afterEach(() => {
vi.useRealTimers();
});
test("creates a fetch timer for a repo with an origin remote", async () => {
const logger = createLogger();
const manager = new BackgroundGitFetchManager({ logger: logger as any });
const subscription = await manager.subscribe(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" },
vi.fn(),
);
await flushPromises();
const managerAny = manager as any;
const target = managerAny.targets.get("/tmp/repo/.git");
expect(target).toBeDefined();
expect(target.intervalId).toBeTruthy();
expect(execFileMock).toHaveBeenNthCalledWith(
1,
"git",
["remote", "get-url", "origin"],
expect.objectContaining({
cwd: "/tmp/repo",
env: expect.objectContaining({ GIT_TERMINAL_PROMPT: "0" }),
}),
expect.any(Function),
);
expect(execFileMock).toHaveBeenNthCalledWith(
2,
"git",
["fetch", "origin", "--prune"],
expect.objectContaining({
cwd: "/tmp/repo",
env: expect.objectContaining({ GIT_TERMINAL_PROMPT: "0" }),
}),
expect.any(Function),
);
subscription.unsubscribe();
manager.dispose();
});
test("dedupes multiple subscribers for the same repo root behind one timer", async () => {
const logger = createLogger();
const manager = new BackgroundGitFetchManager({ logger: logger as any });
const listenerOne = vi.fn();
const listenerTwo = vi.fn();
const subscriptionOne = await manager.subscribe(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" },
listenerOne,
);
const subscriptionTwo = await manager.subscribe(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo-worktree" },
listenerTwo,
);
await flushPromises();
const managerAny = manager as any;
const target = managerAny.targets.get("/tmp/repo/.git");
expect(managerAny.targets.size).toBe(1);
expect(target.listeners).toEqual(new Set([listenerOne, listenerTwo]));
expect(execFileMock.mock.calls.filter((call) => call[1][0] === "remote")).toHaveLength(1);
subscriptionOne.unsubscribe();
subscriptionTwo.unsubscribe();
manager.dispose();
});
test("cleans up the timer when the last subscriber unsubscribes", async () => {
const logger = createLogger();
const manager = new BackgroundGitFetchManager({ logger: logger as any });
const subscription = await manager.subscribe(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" },
vi.fn(),
);
await flushPromises();
const managerAny = manager as any;
const target = managerAny.targets.get("/tmp/repo/.git");
const intervalId = target.intervalId;
const clearIntervalSpy = vi.spyOn(globalThis, "clearInterval");
subscription.unsubscribe();
expect(clearIntervalSpy).toHaveBeenCalledWith(intervalId);
expect(managerAny.targets.size).toBe(0);
clearIntervalSpy.mockRestore();
manager.dispose();
});
test("logs fetch errors without crashing", async () => {
const logger = createLogger();
execFileMock.mockImplementation(
(
_file: string,
args: string[],
_options: unknown,
callback: (error: Error | null, stdout?: string, stderr?: string) => void,
) => {
if (args[0] === "remote") {
callback(null, "", "");
return;
}
callback(new Error("fetch failed"));
},
);
const manager = new BackgroundGitFetchManager({ logger: logger as any });
await manager.subscribe({ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" }, vi.fn());
await flushPromises();
expect(logger.debug).toHaveBeenCalledWith(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" },
"Running background git fetch",
);
expect(logger.warn).toHaveBeenCalledWith(
{
err: expect.any(Error),
repoGitRoot: "/tmp/repo/.git",
cwd: "/tmp/repo",
},
"Background git fetch failed",
);
manager.dispose();
});
test("calls listeners when a fetch completes", async () => {
const logger = createLogger();
const manager = new BackgroundGitFetchManager({ logger: logger as any });
const listener = vi.fn();
await manager.subscribe({ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" }, listener);
await flushPromises();
expect(listener).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(180_000);
await flushPromises();
expect(listener).toHaveBeenCalledTimes(2);
manager.dispose();
});
test("does not create a timer when the repo has no origin remote", async () => {
const logger = createLogger();
execFileMock.mockImplementation(
(
_file: string,
_args: string[],
_options: unknown,
callback: (error: Error | null, stdout?: string, stderr?: string) => void,
) => {
callback(new Error("missing origin"));
},
);
const manager = new BackgroundGitFetchManager({ logger: logger as any });
const subscription = await manager.subscribe(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" },
vi.fn(),
);
expect((manager as any).targets.size).toBe(0);
subscription.unsubscribe();
manager.dispose();
});
test("dispose clears timers and listeners", async () => {
const logger = createLogger();
const manager = new BackgroundGitFetchManager({ logger: logger as any });
const listener = vi.fn();
await manager.subscribe({ repoGitRoot: "/tmp/repo-one/.git", cwd: "/tmp/repo-one" }, listener);
await manager.subscribe({ repoGitRoot: "/tmp/repo-two/.git", cwd: "/tmp/repo-two" }, vi.fn());
await flushPromises();
const managerAny = manager as any;
const clearIntervalSpy = vi.spyOn(globalThis, "clearInterval");
manager.dispose();
expect(clearIntervalSpy).toHaveBeenCalledTimes(2);
expect(managerAny.targets.size).toBe(0);
clearIntervalSpy.mockRestore();
});
});

150
packages/server/src/server/background-git-fetch-manager.ts
View File

@@ -1,150 +0,0 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import type pino from "pino";
import { READ_ONLY_GIT_ENV } from "./checkout-git-utils.js";
const execFileAsync = promisify(execFile);
const BACKGROUND_GIT_FETCH_INTERVAL_MS = 180_000;
type BackgroundGitFetchTarget = {
repoGitRoot: string;
cwd: string;
listeners: Set<() => void>;
intervalId: NodeJS.Timeout | null;
fetchInFlight: boolean;
};
export class BackgroundGitFetchManager {
private readonly logger: pino.Logger;
private readonly targets = new Map<string, BackgroundGitFetchTarget>();
constructor(options: { logger: pino.Logger }) {
this.logger = options.logger.child({ module: "background-git-fetch-manager" });
}
async subscribe(
params: { repoGitRoot: string; cwd: string },
listener: () => void,
): Promise<{ unsubscribe: () => void }> {
const existingTarget = this.targets.get(params.repoGitRoot);
if (existingTarget) {
existingTarget.listeners.add(listener);
return {
unsubscribe: () => {
this.removeListener(params.repoGitRoot, listener);
},
};
}
const hasOrigin = await this.hasOriginRemote(params.cwd);
if (!hasOrigin) {
return { unsubscribe: () => {} };
}
const targetAfterProbe = this.targets.get(params.repoGitRoot);
if (targetAfterProbe) {
targetAfterProbe.listeners.add(listener);
return {
unsubscribe: () => {
this.removeListener(params.repoGitRoot, listener);
},
};
}
const target: BackgroundGitFetchTarget = {
repoGitRoot: params.repoGitRoot,
cwd: params.cwd,
listeners: new Set([listener]),
intervalId: setInterval(() => {
void this.runFetch(target);
}, BACKGROUND_GIT_FETCH_INTERVAL_MS),
fetchInFlight: false,
};
this.targets.set(params.repoGitRoot, target);
void this.runFetch(target);
return {
unsubscribe: () => {
this.removeListener(params.repoGitRoot, listener);
},
};
}
dispose(): void {
for (const target of this.targets.values()) {
this.closeTarget(target);
}
this.targets.clear();
}
private closeTarget(target: BackgroundGitFetchTarget): void {
if (target.intervalId) {
clearInterval(target.intervalId);
target.intervalId = null;
}
target.listeners.clear();
}
private removeListener(targetKey: string, listener: () => void): void {
const target = this.targets.get(targetKey);
if (!target) {
return;
}
target.listeners.delete(listener);
if (target.listeners.size > 0) {
return;
}
this.closeTarget(target);
this.targets.delete(targetKey);
}
private async hasOriginRemote(cwd: string): Promise<boolean> {
try {
await execFileAsync("git", ["remote", "get-url", "origin"], {
cwd,
env: {
...READ_ONLY_GIT_ENV,
GIT_TERMINAL_PROMPT: "0",
},
});
return true;
} catch {
return false;
}
}
private async runFetch(target: BackgroundGitFetchTarget): Promise<void> {
if (target.fetchInFlight) {
return;
}
target.fetchInFlight = true;
this.logger.debug(
{ repoGitRoot: target.repoGitRoot, cwd: target.cwd },
"Running background git fetch",
);
try {
await execFileAsync("git", ["fetch", "origin", "--prune"], {
cwd: target.cwd,
env: {
...process.env,
GIT_TERMINAL_PROMPT: "0",
},
});
} catch (error) {
this.logger.warn(
{ err: error, repoGitRoot: target.repoGitRoot, cwd: target.cwd },
"Background git fetch failed",
);
} finally {
target.fetchInFlight = false;
for (const listener of target.listeners) {
listener();
}
}
}
}

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

@@ -1,9 +1,8 @@
import { v4 as uuidv4 } from "uuid"; import { v4 as uuidv4 } from "uuid";
import { watch, type FSWatcher } from "node:fs"; import { stat } from "fs/promises";
import { readFile, stat } from "fs/promises";
import { exec, execFile } from "node:child_process"; import { exec, execFile } from "node:child_process";
import { promisify } from "util"; import { promisify } from "util";
import { join, resolve, sep } from "path"; import { resolve, sep } from "path";
import { homedir } from "node:os"; import { homedir } from "node:os";
import { z } from "zod"; import { z } from "zod";
import type { ToolSet } from "ai"; import type { ToolSet } from "ai";
@@ -67,8 +66,8 @@ import {
import { experimental_createMCPClient } from "ai"; import { experimental_createMCPClient } from "ai";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js"; import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js";
import { BackgroundGitFetchManager } from "./background-git-fetch-manager.js";
import type { DaemonConfigStore } from "./daemon-config-store.js"; import type { DaemonConfigStore } from "./daemon-config-store.js";
import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
import { buildProviderRegistry } from "./agent/provider-registry.js"; import { buildProviderRegistry } from "./agent/provider-registry.js";
import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js";
@@ -147,7 +146,6 @@ import { type WorktreeConfig } from "../utils/worktree.js";
import { runAsyncWorktreeBootstrap } from "./worktree-bootstrap.js"; import { runAsyncWorktreeBootstrap } from "./worktree-bootstrap.js";
import { import {
getCheckoutDiff, getCheckoutDiff,
getCheckoutShortstat,
getCheckoutStatus, getCheckoutStatus,
listBranchSuggestions, listBranchSuggestions,
commitChanges, commitChanges,
@@ -156,12 +154,11 @@ import {
pullCurrentBranch, pullCurrentBranch,
pushCurrentBranch, pushCurrentBranch,
createPullRequest, createPullRequest,
getPullRequestStatus,
} from "../utils/checkout-git.js"; } from "../utils/checkout-git.js";
import { getProjectIcon } from "../utils/project-icon.js"; import { getProjectIcon } from "../utils/project-icon.js";
import { expandTilde } from "../utils/path.js"; import { expandTilde } from "../utils/path.js";
import { searchHomeDirectories, searchWorkspaceEntries } from "../utils/directory-suggestions.js"; import { searchHomeDirectories, searchWorkspaceEntries } from "../utils/directory-suggestions.js";
import { READ_ONLY_GIT_ENV, resolveCheckoutGitDir, toCheckoutError } from "./checkout-git-utils.js"; import { READ_ONLY_GIT_ENV, toCheckoutError } from "./checkout-git-utils.js";
import { CheckoutDiffManager } from "./checkout-diff-manager.js"; import { CheckoutDiffManager } from "./checkout-diff-manager.js";
import type { LocalSpeechModelId } from "./speech/providers/local/models.js"; import type { LocalSpeechModelId } from "./speech/providers/local/models.js";
import { toResolver, type Resolvable } from "./speech/provider-resolver.js"; import { toResolver, type Resolvable } from "./speech/provider-resolver.js";
@@ -219,8 +216,6 @@ function clientSupportsFlexibleEditorIds(appVersion: string | null): boolean {
return isAppVersionAtLeast(appVersion, MIN_VERSION_FLEXIBLE_EDITOR_IDS); return isAppVersionAtLeast(appVersion, MIN_VERSION_FLEXIBLE_EDITOR_IDS);
} }
const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 500;
const WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT = "__removed__";
const MAX_TERMINAL_STREAM_SLOTS = 256; const MAX_TERMINAL_STREAM_SLOTS = 256;
function deriveInitialAgentTitle(prompt: string): string | null { function deriveInitialAgentTitle(prompt: string): string | null {
@@ -270,15 +265,6 @@ export function resolveWaitForFinishError(options: {
type ProcessingPhase = "idle" | "transcribing"; type ProcessingPhase = "idle" | "transcribing";
type WorkspaceGitWatchTarget = {
cwd: string;
watchers: FSWatcher[];
debounceTimer: NodeJS.Timeout | null;
refreshPromise: Promise<void> | null;
refreshQueued: boolean;
latestFingerprint: string | null;
};
type ActiveTerminalStream = { type ActiveTerminalStream = {
terminalId: string; terminalId: string;
slot: number; slot: number;
@@ -406,7 +392,7 @@ export type SessionOptions = {
scheduleService: ScheduleService; scheduleService: ScheduleService;
loopService: LoopService; loopService: LoopService;
checkoutDiffManager: CheckoutDiffManager; checkoutDiffManager: CheckoutDiffManager;
backgroundGitFetchManager: BackgroundGitFetchManager; workspaceGitService: WorkspaceGitService;
daemonConfigStore: DaemonConfigStore; daemonConfigStore: DaemonConfigStore;
mcpBaseUrl?: string | null; mcpBaseUrl?: string | null;
stt: Resolvable<SpeechToTextProvider | null>; stt: Resolvable<SpeechToTextProvider | null>;
@@ -589,7 +575,7 @@ export class Session {
private readonly scheduleService: ScheduleService; private readonly scheduleService: ScheduleService;
private readonly loopService: LoopService; private readonly loopService: LoopService;
private readonly checkoutDiffManager: CheckoutDiffManager; private readonly checkoutDiffManager: CheckoutDiffManager;
private readonly backgroundGitFetchManager: BackgroundGitFetchManager; private readonly workspaceGitService: WorkspaceGitService;
private readonly daemonConfigStore: DaemonConfigStore; private readonly daemonConfigStore: DaemonConfigStore;
private readonly mcpBaseUrl: string | null; private readonly mcpBaseUrl: string | null;
private readonly downloadTokenStore: DownloadTokenStore; private readonly downloadTokenStore: DownloadTokenStore;
@@ -618,8 +604,7 @@ export class Session {
private inflightRequests = 0; private inflightRequests = 0;
private peakInflightRequests = 0; private peakInflightRequests = 0;
private readonly checkoutDiffSubscriptions = new Map<string, () => void>(); private readonly checkoutDiffSubscriptions = new Map<string, () => void>();
private readonly workspaceGitWatchTargets = new Map<string, WorkspaceGitWatchTarget>(); private readonly workspaceGitSubscriptions = new Map<string, () => void>();
private readonly workspaceGitFetchSubscriptions = new Map<string, () => void>();
private readonly registerVoiceSpeakHandler?: ( private readonly registerVoiceSpeakHandler?: (
agentId: string, agentId: string,
handler: VoiceSpeakHandler, handler: VoiceSpeakHandler,
@@ -654,7 +639,7 @@ export class Session {
scheduleService, scheduleService,
loopService, loopService,
checkoutDiffManager, checkoutDiffManager,
backgroundGitFetchManager, workspaceGitService,
daemonConfigStore, daemonConfigStore,
mcpBaseUrl, mcpBaseUrl,
stt, stt,
@@ -683,7 +668,7 @@ export class Session {
this.scheduleService = scheduleService; this.scheduleService = scheduleService;
this.loopService = loopService; this.loopService = loopService;
this.checkoutDiffManager = checkoutDiffManager; this.checkoutDiffManager = checkoutDiffManager;
this.backgroundGitFetchManager = backgroundGitFetchManager; this.workspaceGitService = workspaceGitService;
this.daemonConfigStore = daemonConfigStore; this.daemonConfigStore = daemonConfigStore;
this.mcpBaseUrl = mcpBaseUrl ?? null; this.mcpBaseUrl = mcpBaseUrl ?? null;
this.terminalManager = terminalManager; this.terminalManager = terminalManager;
@@ -1444,7 +1429,7 @@ export class Session {
let removedWorkspaceId: string | null = null; let removedWorkspaceId: string | null = null;
if (needsStaleWorkspaceCleanup) { if (needsStaleWorkspaceCleanup) {
await this.workspaceRegistry.archive(staleWorkspace.workspaceId, now); await this.workspaceRegistry.archive(staleWorkspace.workspaceId, now);
this.removeWorkspaceGitWatchTarget(staleWorkspace.workspaceId); this.removeWorkspaceGitSubscription(staleWorkspace.workspaceId);
removedWorkspaceId = staleWorkspace.workspaceId; removedWorkspaceId = staleWorkspace.workspaceId;
} }
@@ -1460,7 +1445,7 @@ export class Session {
await this.workspaceRegistry.upsert(nextWorkspaceRecord); await this.workspaceRegistry.upsert(nextWorkspaceRecord);
if (existing && existing.workspaceId !== resolvedWorkspaceId) { if (existing && existing.workspaceId !== resolvedWorkspaceId) {
await this.workspaceRegistry.archive(existing.workspaceId, now); await this.workspaceRegistry.archive(existing.workspaceId, now);
this.removeWorkspaceGitWatchTarget(existing.workspaceId); this.removeWorkspaceGitSubscription(existing.workspaceId);
removedWorkspaceId ??= existing.workspaceId; removedWorkspaceId ??= existing.workspaceId;
} }
@@ -4181,190 +4166,30 @@ export class Session {
} }
} }
private async resolveWorkspaceGitRefsRoot(gitDir: string): Promise<string> { private removeWorkspaceGitSubscription(cwd: string): void {
try {
const commonDir = (await readFile(join(gitDir, "commondir"), "utf8")).trim();
if (commonDir.length > 0) {
return resolve(gitDir, commonDir);
}
} catch {
// Regular repos do not have a commondir file.
}
return gitDir;
}
private closeWorkspaceGitWatchTarget(target: WorkspaceGitWatchTarget): void {
if (target.debounceTimer) {
clearTimeout(target.debounceTimer);
target.debounceTimer = null;
}
for (const watcher of target.watchers) {
watcher.close();
}
target.watchers = [];
}
private removeWorkspaceGitWatchTarget(cwd: string): void {
const workspaceId = normalizePersistedWorkspaceId(cwd); const workspaceId = normalizePersistedWorkspaceId(cwd);
const target = this.workspaceGitWatchTargets.get(workspaceId); this.workspaceGitSubscriptions.get(workspaceId)?.();
if (!target) { this.workspaceGitSubscriptions.delete(workspaceId);
return;
}
const unsubscribeFetch = this.workspaceGitFetchSubscriptions.get(workspaceId);
unsubscribeFetch?.();
this.workspaceGitFetchSubscriptions.delete(workspaceId);
this.closeWorkspaceGitWatchTarget(target);
this.workspaceGitWatchTargets.delete(workspaceId);
}
private workspaceGitDescriptorFingerprint(workspace: WorkspaceDescriptorPayload | null): string {
if (!workspace) {
return WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT;
}
return JSON.stringify([
workspace.name,
workspace.diffStat ? [workspace.diffStat.additions, workspace.diffStat.deletions] : null,
]);
}
private shouldSkipWorkspaceGitWatchUpdate(
workspaceId: string,
workspace: WorkspaceDescriptorPayload | null,
): boolean {
const target = this.workspaceGitWatchTargets.get(workspaceId);
if (!target) {
return false;
}
const nextFingerprint = this.workspaceGitDescriptorFingerprint(workspace);
if (target.latestFingerprint === nextFingerprint) {
return true;
}
target.latestFingerprint = nextFingerprint;
return false;
}
private rememberWorkspaceGitWatchFingerprint(
workspaceId: string,
workspace: WorkspaceDescriptorPayload | null,
): void {
const target = this.workspaceGitWatchTargets.get(workspaceId);
if (!target) {
return;
}
target.latestFingerprint = this.workspaceGitDescriptorFingerprint(workspace);
}
private primeWorkspaceGitWatchFingerprints(
workspaces: Iterable<WorkspaceDescriptorPayload>,
): void {
for (const workspace of workspaces) {
this.rememberWorkspaceGitWatchFingerprint(workspace.id, workspace);
}
}
private scheduleWorkspaceGitWatchRefresh(target: WorkspaceGitWatchTarget): void {
if (target.debounceTimer) {
clearTimeout(target.debounceTimer);
}
target.debounceTimer = setTimeout(() => {
target.debounceTimer = null;
void this.refreshWorkspaceGitWatchTarget(target);
}, WORKSPACE_GIT_WATCH_DEBOUNCE_MS);
}
private async refreshWorkspaceGitWatchTarget(target: WorkspaceGitWatchTarget): Promise<void> {
if (target.refreshPromise) {
target.refreshQueued = true;
return;
}
target.refreshPromise = (async () => {
do {
target.refreshQueued = false;
await this.emitWorkspaceUpdateForCwd(target.cwd, {
dedupeGitState: true,
});
} while (target.refreshQueued);
})();
try {
await target.refreshPromise;
} finally {
target.refreshPromise = null;
}
}
private async ensureWorkspaceGitWatchTarget(cwd: string): Promise<void> {
const workspaceId = normalizePersistedWorkspaceId(cwd);
if (this.workspaceGitWatchTargets.has(workspaceId)) {
return;
}
const gitDir = await resolveCheckoutGitDir(cwd);
if (!gitDir) {
return;
}
const refsRoot = await this.resolveWorkspaceGitRefsRoot(gitDir);
const target: WorkspaceGitWatchTarget = {
cwd: workspaceId,
watchers: [],
debounceTimer: null,
refreshPromise: null,
refreshQueued: false,
latestFingerprint: null,
};
for (const watchPath of new Set([join(gitDir, "HEAD"), join(refsRoot, "refs", "heads")])) {
let watcher: FSWatcher | null = null;
try {
watcher = watch(watchPath, { recursive: false }, () => {
this.scheduleWorkspaceGitWatchRefresh(target);
});
} catch (error) {
this.sessionLogger.warn(
{ err: error, cwd, watchPath },
"Failed to start workspace git watcher",
);
}
if (!watcher) {
continue;
}
watcher.on("error", (error) => {
this.sessionLogger.warn({ err: error, cwd, watchPath }, "Workspace git watcher error");
});
target.watchers.push(watcher);
}
if (target.watchers.length === 0) {
return;
}
this.workspaceGitWatchTargets.set(workspaceId, target);
const subscription = await this.backgroundGitFetchManager.subscribe(
{ repoGitRoot: refsRoot, cwd: workspaceId },
() => {
const activeTarget = this.workspaceGitWatchTargets.get(workspaceId);
if (activeTarget) {
this.scheduleWorkspaceGitWatchRefresh(activeTarget);
}
},
);
this.workspaceGitFetchSubscriptions.set(workspaceId, subscription.unsubscribe);
} }
private async syncWorkspaceGitWatchTarget( private async syncWorkspaceGitWatchTarget(
cwd: string, cwd: string,
options: { isGit: boolean }, options: { isGit: boolean },
): Promise<void> { ): Promise<void> {
const workspaceId = normalizePersistedWorkspaceId(cwd);
if (!options.isGit) { if (!options.isGit) {
this.removeWorkspaceGitWatchTarget(cwd); this.removeWorkspaceGitSubscription(workspaceId);
return; return;
} }
await this.ensureWorkspaceGitWatchTarget(cwd); if (this.workspaceGitSubscriptions.has(workspaceId)) {
return;
}
const subscription = await this.workspaceGitService.subscribe({ cwd: workspaceId }, () => {
void this.emitWorkspaceUpdateForCwd(workspaceId);
});
this.workspaceGitSubscriptions.set(workspaceId, subscription.unsubscribe);
} }
private async handleSubscribeCheckoutDiffRequest( private async handleSubscribeCheckoutDiffRequest(
@@ -4806,14 +4631,20 @@ export class Session {
const { cwd, requestId } = msg; const { cwd, requestId } = msg;
try { try {
const prStatus = await getPullRequestStatus(cwd); await this.workspaceGitService.refresh(cwd, { priority: "high" });
const snapshot = await this.workspaceGitService.getSnapshot(cwd);
this.emit({ this.emit({
type: "checkout_pr_status_response", type: "checkout_pr_status_response",
payload: { payload: {
cwd, cwd,
status: prStatus.status, status: snapshot.github.pullRequest,
githubFeaturesEnabled: prStatus.githubFeaturesEnabled, githubFeaturesEnabled: snapshot.github.featuresEnabled,
error: null, error: snapshot.github.error
? {
code: "UNKNOWN",
message: snapshot.github.error.message,
}
: null,
requestId, requestId,
}, },
}); });
@@ -5510,6 +5341,35 @@ export class Session {
}; };
} }
private buildWorkspaceGitRuntimePayload(
snapshot: WorkspaceGitRuntimeSnapshot,
): NonNullable<WorkspaceDescriptorPayload["gitRuntime"]> | null {
if (!snapshot.git.isGit) {
return null;
}
return {
currentBranch: snapshot.git.currentBranch,
remoteUrl: snapshot.git.remoteUrl,
isPaseoOwnedWorktree: snapshot.git.isPaseoOwnedWorktree,
isDirty: snapshot.git.isDirty,
aheadBehind: snapshot.git.aheadBehind,
aheadOfOrigin: snapshot.git.aheadOfOrigin,
behindOfOrigin: snapshot.git.behindOfOrigin,
};
}
private buildWorkspaceGitHubRuntimePayload(
snapshot: WorkspaceGitRuntimeSnapshot,
): NonNullable<WorkspaceDescriptorPayload["githubRuntime"]> {
return {
featuresEnabled: snapshot.github.featuresEnabled,
pullRequest: snapshot.github.pullRequest,
error: snapshot.github.error,
refreshedAt: snapshot.github.refreshedAt,
};
}
private async describeWorkspaceRecordWithGitData( private async describeWorkspaceRecordWithGitData(
workspace: PersistedWorkspaceRecord, workspace: PersistedWorkspaceRecord,
projectRecord?: PersistedProjectRecord | null, projectRecord?: PersistedProjectRecord | null,
@@ -5526,14 +5386,16 @@ export class Session {
// Fall back to the persisted label if checkout metadata is unavailable. // Fall back to the persisted label if checkout metadata is unavailable.
} }
let diffStat: { additions: number; deletions: number } | null = null; let snapshot: WorkspaceGitRuntimeSnapshot | null = null;
try { snapshot = this.workspaceGitService.peekSnapshot(workspace.cwd);
diffStat = await getCheckoutShortstat(workspace.cwd);
} catch {
// Non-critical — leave null on failure.
}
return { ...base, name: displayName, diffStat }; return {
...base,
name: displayName,
diffStat: snapshot?.git.diffStat ?? null,
gitRuntime: snapshot ? this.buildWorkspaceGitRuntimePayload(snapshot) : undefined,
githubRuntime: snapshot ? this.buildWorkspaceGitHubRuntimePayload(snapshot) : undefined,
};
} }
private async buildWorkspaceDescriptor(input: { private async buildWorkspaceDescriptor(input: {
@@ -5608,16 +5470,6 @@ export class Session {
return descriptorsByWorkspaceId; return descriptorsByWorkspaceId;
} }
private async listWorkspaceDescriptorsSnapshot(): Promise<WorkspaceDescriptorPayload[]> {
return Array.from(
(
await this.buildWorkspaceDescriptorMap({
includeGitData: false,
})
).values(),
);
}
private resolveRegisteredWorkspaceIdForCwd( private resolveRegisteredWorkspaceIdForCwd(
cwd: string, cwd: string,
workspaces: PersistedWorkspaceRecord[], workspaces: PersistedWorkspaceRecord[],
@@ -5645,7 +5497,13 @@ export class Session {
} }
private async listWorkspaceDescriptors(): Promise<WorkspaceDescriptorPayload[]> { private async listWorkspaceDescriptors(): Promise<WorkspaceDescriptorPayload[]> {
return this.listWorkspaceDescriptorsSnapshot(); return Array.from(
(
await this.buildWorkspaceDescriptorMap({
includeGitData: true,
})
).values(),
);
} }
private normalizeFetchWorkspacesSort( private normalizeFetchWorkspacesSort(
@@ -5958,13 +5816,13 @@ export class Session {
private async archiveWorkspaceRecord(workspaceId: string, archivedAt?: string): Promise<void> { private async archiveWorkspaceRecord(workspaceId: string, archivedAt?: string): Promise<void> {
const existing = await this.workspaceRegistry.get(workspaceId); const existing = await this.workspaceRegistry.get(workspaceId);
if (!existing || existing.archivedAt) { if (!existing || existing.archivedAt) {
this.removeWorkspaceGitWatchTarget(workspaceId); this.removeWorkspaceGitSubscription(workspaceId);
return; return;
} }
const nextArchivedAt = archivedAt ?? new Date().toISOString(); const nextArchivedAt = archivedAt ?? new Date().toISOString();
await this.workspaceRegistry.archive(workspaceId, nextArchivedAt); await this.workspaceRegistry.archive(workspaceId, nextArchivedAt);
this.removeWorkspaceGitWatchTarget(workspaceId); this.removeWorkspaceGitSubscription(workspaceId);
const siblingWorkspaces = (await this.workspaceRegistry.list()).filter( const siblingWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => workspace.projectId === existing.projectId && !workspace.archivedAt, (workspace) => workspace.projectId === existing.projectId && !workspace.archivedAt,
@@ -5993,7 +5851,7 @@ export class Session {
private async emitWorkspaceUpdatesForWorkspaceIds( private async emitWorkspaceUpdatesForWorkspaceIds(
workspaceIds: Iterable<string>, workspaceIds: Iterable<string>,
options?: { dedupeGitState?: boolean; skipReconcile?: boolean }, options?: { skipReconcile?: boolean },
): Promise<void> { ): Promise<void> {
const subscription = this.workspaceUpdatesSubscription; const subscription = this.workspaceUpdatesSubscription;
if (!subscription) { if (!subscription) {
@@ -6018,13 +5876,6 @@ export class Session {
workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter }) workspace && this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })
? workspace ? workspace
: null; : null;
if (
options?.dedupeGitState &&
this.shouldSkipWorkspaceGitWatchUpdate(workspaceId, nextWorkspace)
) {
continue;
}
this.rememberWorkspaceGitWatchFingerprint(workspaceId, nextWorkspace);
if (!nextWorkspace) { if (!nextWorkspace) {
this.bufferOrEmitWorkspaceUpdate(subscription, { this.bufferOrEmitWorkspaceUpdate(subscription, {
@@ -6045,30 +5896,9 @@ export class Session {
} }
} }
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( private async emitWorkspaceUpdateForCwd(
cwd: string, cwd: string,
options?: { dedupeGitState?: boolean }, options?: { skipReconcile?: boolean },
): Promise<void> { ): Promise<void> {
const activeWorkspaces = (await this.workspaceRegistry.list()).filter( const activeWorkspaces = (await this.workspaceRegistry.list()).filter(
(workspace) => !workspace.archivedAt, (workspace) => !workspace.archivedAt,
@@ -6175,7 +6005,6 @@ export class Session {
} }
const payload = await this.listFetchWorkspacesEntries(request); const payload = await this.listFetchWorkspacesEntries(request);
this.primeWorkspaceGitWatchFingerprints(payload.entries);
const snapshotLatestActivityByWorkspaceId = new Map<string, number>(); const snapshotLatestActivityByWorkspaceId = new Map<string, number>();
for (const entry of payload.entries) { for (const entry of payload.entries) {
const parsedLatestActivity = entry.activityAt const parsedLatestActivity = entry.activityAt
@@ -6198,10 +6027,6 @@ export class Session {
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) { if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
this.flushBootstrappedWorkspaceUpdates({ snapshotLatestActivityByWorkspaceId }); this.flushBootstrappedWorkspaceUpdates({ snapshotLatestActivityByWorkspaceId });
void this.reconcileAndEmitWorkspaceUpdates(); void this.reconcileAndEmitWorkspaceUpdates();
this.scheduleWorkspaceGitBootstrapUpdates({
subscriptionId,
workspaces: payload.entries,
});
} }
} catch (error) { } catch (error) {
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) { if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
@@ -6227,7 +6052,9 @@ export class Session {
): Promise<void> { ): Promise<void> {
try { try {
const workspace = await this.ensureWorkspaceRegistered(request.cwd); const workspace = await this.ensureWorkspaceRegistered(request.cwd);
await this.emitWorkspaceUpdateForCwd(workspace.cwd); await this.emitWorkspaceUpdateForCwd(workspace.cwd, {
skipReconcile: true,
});
const descriptor = await this.describeWorkspaceRecordWithGitData(workspace); const descriptor = await this.describeWorkspaceRecordWithGitData(workspace);
this.emit({ this.emit({
type: "open_project_response", type: "open_project_response",
@@ -6349,8 +6176,7 @@ export class Session {
return createWorktreeInBackgroundSession( return createWorktreeInBackgroundSession(
{ {
paseoHome: this.paseoHome, paseoHome: this.paseoHome,
emitWorkspaceUpdateForCwd: (cwd, emitOptions) => emitWorkspaceUpdateForCwd: (cwd) => this.emitWorkspaceUpdateForCwd(cwd),
this.emitWorkspaceUpdateForCwd(cwd, emitOptions),
sessionLogger: this.sessionLogger, sessionLogger: this.sessionLogger,
terminalManager: this.terminalManager, terminalManager: this.terminalManager,
}, },
@@ -7517,14 +7343,10 @@ export class Session {
} }
this.checkoutDiffSubscriptions.clear(); this.checkoutDiffSubscriptions.clear();
for (const unsubscribe of this.workspaceGitFetchSubscriptions.values()) { for (const unsubscribe of this.workspaceGitSubscriptions.values()) {
unsubscribe(); unsubscribe();
} }
this.workspaceGitFetchSubscriptions.clear(); this.workspaceGitSubscriptions.clear();
for (const target of this.workspaceGitWatchTargets.values()) {
this.closeWorkspaceGitWatchTarget(target);
}
this.workspaceGitWatchTargets.clear();
} }
// ============================================================================ // ============================================================================

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

@@ -1,86 +1,84 @@
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { describe, expect, test, vi } from "vitest";
import { tmpdir } from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
const { watchCalls, watchMock } = vi.hoisted(() => {
const hoistedWatchCalls: Array<{
path: string;
listener: () => void;
close: ReturnType<typeof vi.fn>;
}> = [];
const hoistedWatchMock = vi.fn(
(watchPath: string, _options: { recursive: boolean }, listener: () => void) => {
const close = vi.fn();
const watcher = {
close,
on: vi.fn().mockReturnThis(),
};
hoistedWatchCalls.push({
path: watchPath,
listener,
close,
});
return watcher as any;
},
);
return {
watchCalls: hoistedWatchCalls,
watchMock: hoistedWatchMock,
};
});
const resolveCheckoutGitDirMock = vi.hoisted(() => vi.fn(async () => null));
vi.mock("node:fs", async () => {
const actual = await vi.importActual<typeof import("node:fs")>("node:fs");
return {
...actual,
watch: watchMock,
};
});
vi.mock("./checkout-git-utils.js", () => ({
READ_ONLY_GIT_ENV: {
...process.env,
GIT_OPTIONAL_LOCKS: "0",
},
resolveCheckoutGitDir: resolveCheckoutGitDirMock,
toCheckoutError: vi.fn((error: unknown) => ({
message: error instanceof Error ? error.message : String(error),
})),
}));
import { Session } from "./session.js"; import { Session } from "./session.js";
import type {
WorkspaceGitListener,
WorkspaceGitRuntimeSnapshot,
WorkspaceGitService,
} from "./workspace-git-service.js";
function createWorkspaceRuntimeSnapshot(
cwd: string,
overrides?: {
git?: Partial<WorkspaceGitRuntimeSnapshot["git"]>;
github?: Partial<WorkspaceGitRuntimeSnapshot["github"]>;
},
): WorkspaceGitRuntimeSnapshot {
const base: WorkspaceGitRuntimeSnapshot = {
cwd,
git: {
isGit: true,
repoRoot: cwd,
mainRepoRoot: null,
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
isDirty: false,
aheadBehind: { ahead: 0, behind: 0 },
aheadOfOrigin: 0,
behindOfOrigin: 0,
diffStat: { additions: 1, deletions: 0 },
},
github: {
featuresEnabled: true,
pullRequest: null,
error: null,
refreshedAt: "2026-04-12T00:00:00.000Z",
},
};
return {
cwd,
git: {
...base.git,
...overrides?.git,
},
github: {
...base.github,
...overrides?.github,
pullRequest:
overrides?.github && "pullRequest" in overrides.github
? overrides.github.pullRequest ?? null
: base.github.pullRequest,
error:
overrides?.github && "error" in overrides.github
? overrides.github.error ?? null
: base.github.error,
},
};
}
function createSessionForWorkspaceGitWatchTests(): { function createSessionForWorkspaceGitWatchTests(): {
session: Session; session: Session;
emitted: Array<{ type: string; payload: unknown }>; emitted: Array<{ type: string; payload: unknown }>;
backgroundGitFetchManager: { workspaceGitService: WorkspaceGitService & {
subscribe: ReturnType<typeof vi.fn>; subscribe: ReturnType<typeof vi.fn>;
subscriptions: Array<{ peekSnapshot: ReturnType<typeof vi.fn>;
params: { repoGitRoot: string; cwd: string }; getSnapshot: ReturnType<typeof vi.fn>;
listener: () => void; refresh: ReturnType<typeof vi.fn>;
unsubscribe: ReturnType<typeof vi.fn>; dispose: ReturnType<typeof vi.fn>;
}>;
};
logger: {
child: () => unknown;
trace: ReturnType<typeof vi.fn>;
debug: ReturnType<typeof vi.fn>;
info: ReturnType<typeof vi.fn>;
warn: ReturnType<typeof vi.fn>;
error: ReturnType<typeof vi.fn>;
}; };
subscriptions: Array<{
params: { cwd: string };
listener: WorkspaceGitListener;
unsubscribe: ReturnType<typeof vi.fn>;
}>;
} { } {
const emitted: Array<{ type: string; payload: unknown }> = []; const emitted: Array<{ type: string; payload: unknown }> = [];
const projects = new Map<string, any>(); const projects = new Map<string, any>();
const workspaces = new Map<string, any>(); const workspaces = new Map<string, any>();
const backgroundGitFetchSubscriptions: Array<{ const subscriptions: Array<{
params: { repoGitRoot: string; cwd: string }; params: { cwd: string };
listener: () => void; listener: WorkspaceGitListener;
unsubscribe: ReturnType<typeof vi.fn>; unsubscribe: ReturnType<typeof vi.fn>;
}> = []; }> = [];
const logger = { const logger = {
@@ -91,16 +89,23 @@ function createSessionForWorkspaceGitWatchTests(): {
warn: vi.fn(), warn: vi.fn(),
error: vi.fn(), error: vi.fn(),
}; };
const backgroundGitFetchManager = { const workspaceGitService = {
subscribe: vi.fn(async (params: { repoGitRoot: string; cwd: string }, listener: () => void) => { subscribe: vi.fn(async (params: { cwd: string }, listener: WorkspaceGitListener) => {
const unsubscribe = vi.fn(); const unsubscribe = vi.fn();
backgroundGitFetchSubscriptions.push({ subscriptions.push({
params, params,
listener, listener,
unsubscribe, unsubscribe,
}); });
return { unsubscribe }; return {
initial: createWorkspaceRuntimeSnapshot(params.cwd),
unsubscribe,
};
}), }),
peekSnapshot: vi.fn((cwd: string) => createWorkspaceRuntimeSnapshot(cwd)),
getSnapshot: vi.fn(async (cwd: string) => createWorkspaceRuntimeSnapshot(cwd)),
refresh: vi.fn(async () => {}),
dispose: vi.fn(),
}; };
const session = new Session({ const session = new Session({
@@ -179,7 +184,7 @@ function createSessionForWorkspaceGitWatchTests(): {
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
backgroundGitFetchManager: backgroundGitFetchManager as any, workspaceGitService: workspaceGitService as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -191,29 +196,15 @@ function createSessionForWorkspaceGitWatchTests(): {
return { return {
session, session,
emitted, emitted,
backgroundGitFetchManager: { workspaceGitService: workspaceGitService as any,
subscribe: backgroundGitFetchManager.subscribe, subscriptions,
subscriptions: backgroundGitFetchSubscriptions,
},
logger,
}; };
} }
describe("workspace git watch targets", () => { describe("workspace git watch targets", () => {
beforeEach(() => { test("emits one workspace_update when the workspace git service emits a changed snapshot", async () => {
watchCalls.length = 0; const { session, emitted, workspaceGitService, subscriptions } =
watchMock.mockClear(); createSessionForWorkspaceGitWatchTests();
resolveCheckoutGitDirMock.mockReset();
resolveCheckoutGitDirMock.mockResolvedValue(null);
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test("debounces watcher events and skips unchanged branch/diff snapshots", async () => {
const { session, emitted } = createSessionForWorkspaceGitWatchTests();
const sessionAny = session as any; const sessionAny = session as any;
sessionAny.buildProjectPlacement = async (cwd: string) => ({ sessionAny.buildProjectPlacement = async (cwd: string) => ({
@@ -229,7 +220,6 @@ describe("workspace git watch targets", () => {
mainRepoRoot: null, mainRepoRoot: null,
}, },
}); });
resolveCheckoutGitDirMock.mockResolvedValue("/tmp/repo/.git");
sessionAny.workspaceUpdatesSubscription = { sessionAny.workspaceUpdatesSubscription = {
subscriptionId: "sub-1", subscriptionId: "sub-1",
filter: undefined, filter: undefined,
@@ -254,25 +244,27 @@ describe("workspace git watch targets", () => {
sessionAny.buildWorkspaceDescriptorMap = async () => new Map([[descriptor.id, descriptor]]); sessionAny.buildWorkspaceDescriptorMap = async () => new Map([[descriptor.id, descriptor]]);
await sessionAny.ensureWorkspaceRegistered("/tmp/repo"); await sessionAny.ensureWorkspaceRegistered("/tmp/repo");
sessionAny.primeWorkspaceGitWatchFingerprints([descriptor]);
expect(watchCalls.map((entry) => entry.path).sort()).toEqual([ expect(workspaceGitService.subscribe).toHaveBeenCalledWith(
"/tmp/repo/.git/HEAD", { cwd: "/tmp/repo" },
"/tmp/repo/.git/refs/heads", expect.any(Function),
]); );
watchCalls[0]!.listener();
watchCalls[1]!.listener();
await vi.advanceTimersByTimeAsync(500);
expect(emitted.filter((message) => message.type === "workspace_update")).toHaveLength(0);
descriptor = { descriptor = {
...descriptor, ...descriptor,
name: "renamed-branch", name: "renamed-branch",
}; };
watchCalls[0]!.listener();
await vi.advanceTimersByTimeAsync(500); subscriptions[0]?.listener(
createWorkspaceRuntimeSnapshot("/tmp/repo", {
git: {
currentBranch: "renamed-branch",
},
}),
);
await Promise.resolve();
await Promise.resolve();
const workspaceUpdates = emitted.filter( const workspaceUpdates = emitted.filter(
(message) => message.type === "workspace_update", (message) => message.type === "workspace_update",
@@ -287,220 +279,106 @@ describe("workspace git watch targets", () => {
}, },
}); });
descriptor = {
...descriptor,
diffStat: { additions: 3, deletions: 1 },
};
watchCalls[1]!.listener();
await vi.advanceTimersByTimeAsync(500);
expect(emitted.filter((message) => message.type === "workspace_update")).toHaveLength(2);
await session.cleanup(); await session.cleanup();
}); });
test("closes watchers when a workspace is archived and when the session closes", async () => { test("checkout_pr_status_request reads pull request status from the workspace git service snapshot", async () => {
const { session } = createSessionForWorkspaceGitWatchTests(); const { session, emitted, workspaceGitService } = createSessionForWorkspaceGitWatchTests();
const sessionAny = session as any;
sessionAny.buildProjectPlacement = async (cwd: string) => ({ workspaceGitService.getSnapshot.mockResolvedValue(
projectKey: cwd, createWorkspaceRuntimeSnapshot("/tmp/repo", {
projectName: path.basename(cwd), github: {
checkout: { featuresEnabled: true,
cwd, pullRequest: {
isGit: true, url: "https://github.com/acme/repo/pull/456",
currentBranch: "main", title: "Runtime centralization",
remoteUrl: "https://github.com/acme/repo.git", state: "merged",
worktreeRoot: cwd, baseRefName: "main",
isPaseoOwnedWorktree: false, headRefName: "workspace-git-service",
mainRepoRoot: null, isMerged: true,
}, },
}); refreshedAt: "2026-04-12T00:05:00.000Z",
},
resolveCheckoutGitDirMock.mockImplementation(async (cwd: string) => path.join(cwd, ".git")); }),
await sessionAny.ensureWorkspaceRegistered("/tmp/repo-one");
expect(sessionAny.workspaceGitWatchTargets.size).toBe(1);
expect(watchCalls).toHaveLength(2);
await sessionAny.archiveWorkspaceRecord("/tmp/repo-one", "2026-03-21T00:00:00.000Z");
expect(sessionAny.workspaceGitWatchTargets.size).toBe(0);
expect(watchCalls.every((entry) => entry.close.mock.calls.length === 1)).toBe(true);
watchCalls.length = 0;
watchMock.mockClear();
await sessionAny.ensureWorkspaceRegistered("/tmp/repo-two");
expect(sessionAny.workspaceGitWatchTargets.size).toBe(1);
expect(watchCalls).toHaveLength(2);
await session.cleanup();
expect(sessionAny.workspaceGitWatchTargets.size).toBe(0);
expect(watchCalls.every((entry) => entry.close.mock.calls.length === 1)).toBe(true);
});
test("resolves refs from the shared git dir for linked worktrees", async () => {
const { session } = createSessionForWorkspaceGitWatchTests();
const sessionAny = session as any;
const tempDir = mkdtempSync(path.join(tmpdir(), "session-workspace-git-watch-"));
const gitDir = path.join(tempDir, "repo", ".git", "worktrees", "feature");
mkdirSync(gitDir, { recursive: true });
writeFileSync(path.join(gitDir, "commondir"), "../..\n");
try {
expect(await sessionAny.resolveWorkspaceGitRefsRoot(gitDir)).toBe(
path.join(tempDir, "repo", ".git"),
);
} finally {
rmSync(tempDir, { recursive: true, force: true });
await session.cleanup();
}
});
test("subscribes to the background fetch manager when a git watch target is created", async () => {
const { session, backgroundGitFetchManager } = createSessionForWorkspaceGitWatchTests();
const sessionAny = session as any;
sessionAny.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: path.basename(cwd),
checkout: {
cwd,
isGit: true,
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
worktreeRoot: cwd,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
});
resolveCheckoutGitDirMock.mockResolvedValue("/tmp/repo/.git");
await sessionAny.ensureWorkspaceRegistered("/tmp/repo");
expect(backgroundGitFetchManager.subscribe).toHaveBeenCalledWith(
{ repoGitRoot: "/tmp/repo/.git", cwd: "/tmp/repo" },
expect.any(Function),
); );
expect(sessionAny.workspaceGitFetchSubscriptions.size).toBe(1);
await session.cleanup(); await session.handleMessage({
}); type: "checkout_pr_status_request",
test("stores separate background fetch subscriptions per workspace and unsubscribes removed targets", async () => {
const { session, backgroundGitFetchManager } = createSessionForWorkspaceGitWatchTests();
const sessionAny = session as any;
sessionAny.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: path.basename(cwd),
checkout: {
cwd,
isGit: true,
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
worktreeRoot: cwd,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
});
resolveCheckoutGitDirMock.mockImplementation(async (cwd: string) =>
cwd === "/tmp/repo" ? "/tmp/repo/.git" : "/tmp/repo/.git/worktrees/feature",
);
sessionAny.resolveWorkspaceGitRefsRoot = vi.fn(async () => "/tmp/repo/.git");
await sessionAny.ensureWorkspaceRegistered("/tmp/repo");
await sessionAny.ensureWorkspaceRegistered("/tmp/repo-feature");
expect(backgroundGitFetchManager.subscribe).toHaveBeenCalledTimes(2);
expect(backgroundGitFetchManager.subscriptions[0]?.params).toEqual({
repoGitRoot: "/tmp/repo/.git",
cwd: "/tmp/repo", cwd: "/tmp/repo",
}); requestId: "req-pr-status",
expect(backgroundGitFetchManager.subscriptions[1]?.params).toEqual({
repoGitRoot: "/tmp/repo/.git",
cwd: "/tmp/repo-feature",
}); });
sessionAny.removeWorkspaceGitWatchTarget("/tmp/repo"); expect(workspaceGitService.getSnapshot).toHaveBeenCalledWith("/tmp/repo");
expect(emitted.find((message) => message.type === "checkout_pr_status_response")?.payload).toEqual({
expect(backgroundGitFetchManager.subscriptions[0]?.unsubscribe).toHaveBeenCalledTimes(1); cwd: "/tmp/repo",
expect(backgroundGitFetchManager.subscriptions[1]?.unsubscribe).not.toHaveBeenCalled(); status: {
expect(sessionAny.workspaceGitFetchSubscriptions.size).toBe(1); url: "https://github.com/acme/repo/pull/456",
title: "Runtime centralization",
await session.cleanup(); state: "merged",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: true,
},
githubFeaturesEnabled: true,
error: null,
requestId: "req-pr-status",
});
}); });
test("refreshes the workspace when the background fetch manager callback fires and unsubscribes on cleanup", async () => { test("checkout_pr_status_request explicitly refreshes the focused workspace before reading runtime data", async () => {
const { session, emitted, backgroundGitFetchManager } = const { session, emitted, workspaceGitService } = createSessionForWorkspaceGitWatchTests();
createSessionForWorkspaceGitWatchTests(); let refreshed = false;
const sessionAny = session as any;
sessionAny.buildProjectPlacement = async (cwd: string) => ({ workspaceGitService.refresh.mockImplementation(async () => {
projectKey: cwd, refreshed = true;
projectName: path.basename(cwd),
checkout: {
cwd,
isGit: true,
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
worktreeRoot: cwd,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
}); });
resolveCheckoutGitDirMock.mockResolvedValue("/tmp/repo/.git"); workspaceGitService.getSnapshot.mockImplementation(async (cwd: string) =>
sessionAny.workspaceUpdatesSubscription = { createWorkspaceRuntimeSnapshot(cwd, {
subscriptionId: "sub-1", github: {
filter: undefined, pullRequest: refreshed
isBootstrapping: false, ? {
pendingUpdatesByWorkspaceId: new Map(), url: "https://github.com/acme/repo/pull/457",
}; title: "After explicit refresh",
sessionAny.reconcileActiveWorkspaceRecords = async () => new Set(); state: "merged",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: true,
}
: {
url: "https://github.com/acme/repo/pull/456",
title: "Before explicit refresh",
state: "open",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: false,
},
refreshedAt: refreshed ? "2026-04-12T00:10:00.000Z" : "2026-04-12T00:05:00.000Z",
},
}),
);
let descriptor = { await session.handleMessage({
id: "/tmp/repo", type: "checkout_pr_status_request",
projectId: "/tmp/repo", cwd: "/tmp/repo",
projectDisplayName: "repo", requestId: "req-pr-refresh",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "local_checkout",
name: "main",
status: "done",
activityAt: null,
diffStat: { additions: 1, deletions: 0 },
};
sessionAny.buildWorkspaceDescriptorMap = async () => new Map([[descriptor.id, descriptor]]);
await sessionAny.ensureWorkspaceRegistered("/tmp/repo");
sessionAny.primeWorkspaceGitWatchFingerprints([descriptor]);
descriptor = {
...descriptor,
name: "updated-after-fetch",
};
backgroundGitFetchManager.subscriptions[0]?.listener();
await vi.advanceTimersByTimeAsync(500);
const workspaceUpdates = emitted.filter(
(message) => message.type === "workspace_update",
) as any[];
expect(workspaceUpdates).toHaveLength(1);
expect(workspaceUpdates[0]?.payload).toMatchObject({
kind: "upsert",
workspace: {
id: "/tmp/repo",
name: "updated-after-fetch",
},
}); });
await session.cleanup(); expect(workspaceGitService.refresh).toHaveBeenCalledWith("/tmp/repo", {
priority: "high",
expect(backgroundGitFetchManager.subscriptions[0]?.unsubscribe).toHaveBeenCalledTimes(1); });
expect(emitted.find((message) => message.type === "checkout_pr_status_response")?.payload).toEqual({
cwd: "/tmp/repo",
status: {
url: "https://github.com/acme/repo/pull/457",
title: "After explicit refresh",
state: "merged",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: true,
},
githubFeaturesEnabled: true,
error: null,
requestId: "req-pr-refresh",
});
}); });
}); });

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

@@ -5,6 +5,7 @@ import path from "node:path";
import { describe, expect, test, vi } from "vitest"; import { describe, expect, test, vi } from "vitest";
import { Session } from "./session.js"; import { Session } from "./session.js";
import type { AgentSnapshotPayload } from "../shared/messages.js"; import type { AgentSnapshotPayload } from "../shared/messages.js";
import type { WorkspaceGitRuntimeSnapshot } from "./workspace-git-service.js";
import { import {
createPersistedProjectRecord, createPersistedProjectRecord,
createPersistedWorkspaceRecord, createPersistedWorkspaceRecord,
@@ -61,7 +62,123 @@ function makeAgent(input: {
}; };
} }
function createSessionForWorkspaceTests(options: { appVersion?: string | null } = {}): Session { 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,
refreshedAt: 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,
refreshedAt: null,
},
}),
refresh: async () => {},
dispose: () => {},
};
}
function createWorkspaceRuntimeSnapshot(
cwd: string,
overrides?: {
git?: Partial<WorkspaceGitRuntimeSnapshot["git"]>;
github?: Partial<WorkspaceGitRuntimeSnapshot["github"]>;
},
): WorkspaceGitRuntimeSnapshot {
const base: WorkspaceGitRuntimeSnapshot = {
cwd,
git: {
isGit: true,
repoRoot: cwd,
mainRepoRoot: null,
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
isDirty: false,
aheadBehind: { ahead: 0, behind: 0 },
aheadOfOrigin: 0,
behindOfOrigin: 0,
diffStat: { additions: 1, deletions: 0 },
},
github: {
featuresEnabled: true,
pullRequest: {
url: "https://github.com/acme/repo/pull/123",
title: "Runtime payloads",
state: "open",
baseRefName: "main",
headRefName: "feature/runtime-payloads",
isMerged: false,
},
error: null,
refreshedAt: "2026-04-12T00:00:00.000Z",
},
};
return {
cwd,
git: {
...base.git,
...overrides?.git,
},
github: {
...base.github,
...overrides?.github,
pullRequest:
overrides?.github && "pullRequest" in overrides.github
? overrides.github.pullRequest ?? null
: base.github.pullRequest,
error:
overrides?.github && "error" in overrides.github
? overrides.github.error ?? null
: base.github.error,
},
};
}
function createSessionForWorkspaceTests(options: {
appVersion?: string | null;
workspaceGitService?: ReturnType<typeof createNoopWorkspaceGitService>;
} = {}): Session {
const logger = { const logger = {
child: () => logger, child: () => logger,
trace: vi.fn(), trace: vi.fn(),
@@ -121,8 +238,9 @@ function createSessionForWorkspaceTests(options: { appVersion?: string | null }
checkoutDiffWatcherCount: 0, checkoutDiffWatcherCount: 0,
checkoutDiffFallbackRefreshTargetCount: 0, checkoutDiffFallbackRefreshTargetCount: 0,
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
workspaceGitService: (options.workspaceGitService ?? createNoopWorkspaceGitService()) as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -228,6 +346,7 @@ describe("workspace aggregation", () => {
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -371,6 +490,7 @@ describe("workspace aggregation", () => {
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -532,6 +652,7 @@ describe("workspace aggregation", () => {
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -666,6 +787,7 @@ describe("workspace aggregation", () => {
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -933,6 +1055,7 @@ describe("workspace aggregation", () => {
}), }),
dispose: () => {}, dispose: () => {},
} as any, } as any,
workspaceGitService: createNoopWorkspaceGitService() as any,
mcpBaseUrl: null, mcpBaseUrl: null,
stt: null, stt: null,
tts: null, tts: null,
@@ -1685,14 +1808,231 @@ describe("workspace aggregation", () => {
session.describeWorkspaceRecord = vi.fn(async () => baselineDescriptor); session.describeWorkspaceRecord = vi.fn(async () => baselineDescriptor);
session.describeWorkspaceRecordWithGitData = vi.fn(async () => gitDescriptor); session.describeWorkspaceRecordWithGitData = vi.fn(async () => gitDescriptor);
const descriptors = await session.listWorkspaceDescriptorsSnapshot(); const descriptors = Array.from(
(
await session.buildWorkspaceDescriptorMap({
includeGitData: false,
})
).values(),
);
expect(session.describeWorkspaceRecord).toHaveBeenCalledWith(workspace, project); expect(session.describeWorkspaceRecord).toHaveBeenCalledWith(workspace, project);
expect(session.describeWorkspaceRecordWithGitData).not.toHaveBeenCalled(); expect(session.describeWorkspaceRecordWithGitData).not.toHaveBeenCalled();
expect(descriptors).toEqual([baselineDescriptor]); expect(descriptors).toEqual([baselineDescriptor]);
}); });
test("subscribed fetch_workspaces emits git enrichment updates after the baseline snapshot", async () => { test("fetch_workspaces_response reads runtime fields from passive workspace git service snapshots", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const runtimeSnapshot = createWorkspaceRuntimeSnapshot("/tmp/repo", {
git: {
currentBranch: "runtime-branch",
isDirty: true,
aheadBehind: { ahead: 3, behind: 1 },
aheadOfOrigin: 3,
behindOfOrigin: 1,
},
github: {
pullRequest: {
url: "https://github.com/acme/repo/pull/456",
title: "Ship runtime payloads",
state: "open",
baseRefName: "main",
headRefName: "runtime-branch",
isMerged: false,
},
refreshedAt: "2026-04-12T00:05:00.000Z",
},
});
const workspaceGitService = createNoopWorkspaceGitService();
workspaceGitService.peekSnapshot = vi.fn(() => runtimeSnapshot);
workspaceGitService.getSnapshot = vi.fn(async () => {
throw new Error("fetch_workspaces should not trigger per-workspace refreshes");
});
workspaceGitService.subscribe = vi.fn(async () => ({
initial: runtimeSnapshot,
unsubscribe: () => {},
}));
const session = createSessionForWorkspaceTests({
workspaceGitService,
}) 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.emit = (message: any) => emitted.push(message);
session.listAgentPayloads = async () => [];
session.projectRegistry.list = async () => [project];
session.workspaceRegistry.list = async () => [workspace];
session.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: "repo",
checkout: {
cwd,
isGit: true,
currentBranch: runtimeSnapshot.git.currentBranch,
remoteUrl: runtimeSnapshot.git.remoteUrl,
worktreeRoot: cwd,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
});
await session.handleMessage({
type: "fetch_workspaces_request",
requestId: "req-fetch-workspaces-runtime",
});
const response = emitted.find((message) => message.type === "fetch_workspaces_response") as
| { type: "fetch_workspaces_response"; payload: any }
| undefined;
expect(workspaceGitService.peekSnapshot).toHaveBeenCalledWith("/tmp/repo");
expect(workspaceGitService.getSnapshot).not.toHaveBeenCalled();
expect(response?.payload.entries).toEqual([
expect.objectContaining({
id: "/tmp/repo",
gitRuntime: {
currentBranch: "runtime-branch",
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
isDirty: true,
aheadBehind: { ahead: 3, behind: 1 },
aheadOfOrigin: 3,
behindOfOrigin: 1,
},
githubRuntime: {
featuresEnabled: true,
pullRequest: {
url: "https://github.com/acme/repo/pull/456",
title: "Ship runtime payloads",
state: "open",
baseRefName: "main",
headRefName: "runtime-branch",
isMerged: false,
},
error: null,
refreshedAt: "2026-04-12T00:05:00.000Z",
},
}),
]);
});
test("workspace_update includes updated runtime fields", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const runtimeSnapshot = createWorkspaceRuntimeSnapshot("/tmp/repo", {
git: {
currentBranch: "feature/runtime-payloads",
isDirty: true,
},
github: {
pullRequest: {
url: "https://github.com/acme/repo/pull/789",
title: "Updated runtime payloads",
state: "merged",
baseRefName: "main",
headRefName: "feature/runtime-payloads",
isMerged: true,
},
refreshedAt: "2026-04-12T00:10:00.000Z",
},
});
const workspaceGitService = createNoopWorkspaceGitService();
workspaceGitService.peekSnapshot = vi.fn(() => runtimeSnapshot);
workspaceGitService.getSnapshot = vi.fn(async () => {
throw new Error("workspace updates should use passive workspace git snapshots");
});
const session = createSessionForWorkspaceTests({
workspaceGitService,
}) 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.emit = (message: any) => emitted.push(message);
session.workspaceUpdatesSubscription = {
subscriptionId: "sub-runtime",
filter: undefined,
isBootstrapping: false,
pendingUpdatesByWorkspaceId: new Map(),
};
session.reconcileActiveWorkspaceRecords = async () => new Set();
session.listAgentPayloads = async () => [];
session.projectRegistry.list = async () => [project];
session.workspaceRegistry.list = async () => [workspace];
session.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: "repo",
checkout: {
cwd,
isGit: true,
currentBranch: runtimeSnapshot.git.currentBranch,
remoteUrl: runtimeSnapshot.git.remoteUrl,
worktreeRoot: cwd,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
});
await session.emitWorkspaceUpdateForCwd("/tmp/repo", {
skipReconcile: true,
});
expect(workspaceGitService.peekSnapshot).toHaveBeenCalledWith("/tmp/repo");
expect(workspaceGitService.getSnapshot).not.toHaveBeenCalled();
expect(emitted).toContainEqual({
type: "workspace_update",
payload: {
kind: "upsert",
workspace: expect.objectContaining({
id: "/tmp/repo",
gitRuntime: expect.objectContaining({
currentBranch: "feature/runtime-payloads",
isDirty: true,
}),
githubRuntime: expect.objectContaining({
featuresEnabled: true,
pullRequest: expect.objectContaining({
title: "Updated runtime payloads",
isMerged: true,
}),
refreshedAt: "2026-04-12T00:10:00.000Z",
}),
}),
},
});
});
test("subscribed fetch_workspaces includes git enrichment in the initial snapshot", async () => {
const emitted: Array<{ type: string; payload: any }> = []; const emitted: Array<{ type: string; payload: any }> = [];
const session = createSessionForWorkspaceTests() as any; const session = createSessionForWorkspaceTests() as any;
const gitProject = createPersistedProjectRecord({ const gitProject = createPersistedProjectRecord({
@@ -1794,21 +2134,13 @@ describe("workspace aggregation", () => {
), ),
).toEqual([ ).toEqual([
[directoryDescriptor.id, directoryDescriptor.diffStat], [directoryDescriptor.id, directoryDescriptor.diffStat],
[baselineGitDescriptor.id, baselineGitDescriptor.diffStat], [enrichedGitDescriptor.id, enrichedGitDescriptor.diffStat],
]); ]);
const workspaceUpdates = emitted.filter( const workspaceUpdates = emitted.filter(
(message) => message.type === "workspace_update", (message) => message.type === "workspace_update",
) as Array<{ type: "workspace_update"; payload: any }>; ) as Array<{ type: "workspace_update"; payload: any }>;
expect(workspaceUpdates).toEqual([ expect(workspaceUpdates).toEqual([]);
{
type: "workspace_update",
payload: {
kind: "upsert",
workspace: enrichedGitDescriptor,
},
},
]);
expect(session.describeWorkspaceRecordWithGitData).toHaveBeenCalledWith( expect(session.describeWorkspaceRecordWithGitData).toHaveBeenCalledWith(
gitWorkspace, gitWorkspace,
gitProject, gitProject,

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

@@ -12,7 +12,6 @@ import type { FileBackedChatService } from "./chat/chat-service.js";
import type { LoopService } from "./loop-service.js"; import type { LoopService } from "./loop-service.js";
import type { ScheduleService } from "./schedule/service.js"; import type { ScheduleService } from "./schedule/service.js";
import type { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-manager.js"; import type { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-manager.js";
import { BackgroundGitFetchManager } from "./background-git-fetch-manager.js";
import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js"; import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js";
import { import {
type ServerInfoStatusPayload, type ServerInfoStatusPayload,
@@ -31,6 +30,7 @@ import type { AgentProvider } from "./agent/agent-sdk-types.js";
import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js";
import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
import { buildProviderRegistry } from "./agent/provider-registry.js"; import { buildProviderRegistry } from "./agent/provider-registry.js";
import { WorkspaceGitServiceImpl } from "./workspace-git-service.js";
import { PushTokenStore } from "./push/token-store.js"; import { PushTokenStore } from "./push/token-store.js";
import { PushService } from "./push/push-service.js"; import { PushService } from "./push/push-service.js";
import type { SpeechReadinessSnapshot, SpeechService } from "./speech/speech-runtime.js"; import type { SpeechReadinessSnapshot, SpeechService } from "./speech/speech-runtime.js";
@@ -234,7 +234,7 @@ export class VoiceAssistantWebSocketServer {
private readonly loopService: LoopService; private readonly loopService: LoopService;
private readonly scheduleService: ScheduleService; private readonly scheduleService: ScheduleService;
private readonly checkoutDiffManager: CheckoutDiffManager; private readonly checkoutDiffManager: CheckoutDiffManager;
private readonly backgroundGitFetchManager: BackgroundGitFetchManager; private readonly workspaceGitService: WorkspaceGitServiceImpl;
private readonly downloadTokenStore: DownloadTokenStore; private readonly downloadTokenStore: DownloadTokenStore;
private readonly paseoHome: string; private readonly paseoHome: string;
private readonly daemonConfigStore: DaemonConfigStore; private readonly daemonConfigStore: DaemonConfigStore;
@@ -329,8 +329,9 @@ export class VoiceAssistantWebSocketServer {
throw new Error("VoiceAssistantWebSocketServer requires a checkout diff manager."); throw new Error("VoiceAssistantWebSocketServer requires a checkout diff manager.");
} }
this.checkoutDiffManager = checkoutDiffManager; this.checkoutDiffManager = checkoutDiffManager;
this.backgroundGitFetchManager = new BackgroundGitFetchManager({ this.workspaceGitService = new WorkspaceGitServiceImpl({
logger: this.logger, logger: this.logger,
paseoHome,
}); });
this.downloadTokenStore = downloadTokenStore; this.downloadTokenStore = downloadTokenStore;
this.paseoHome = paseoHome; this.paseoHome = paseoHome;
@@ -507,7 +508,7 @@ export class VoiceAssistantWebSocketServer {
await Promise.all(cleanupPromises); await Promise.all(cleanupPromises);
this.providerSnapshotManager.destroy(); this.providerSnapshotManager.destroy();
this.backgroundGitFetchManager.dispose(); this.workspaceGitService.dispose();
this.checkoutDiffManager.dispose(); this.checkoutDiffManager.dispose();
this.pendingConnections.clear(); this.pendingConnections.clear();
this.sessions.clear(); this.sessions.clear();
@@ -638,7 +639,7 @@ export class VoiceAssistantWebSocketServer {
loopService: this.loopService, loopService: this.loopService,
scheduleService: this.scheduleService, scheduleService: this.scheduleService,
checkoutDiffManager: this.checkoutDiffManager, checkoutDiffManager: this.checkoutDiffManager,
backgroundGitFetchManager: this.backgroundGitFetchManager, workspaceGitService: this.workspaceGitService,
daemonConfigStore: this.daemonConfigStore, daemonConfigStore: this.daemonConfigStore,
mcpBaseUrl: this.mcpBaseUrl, mcpBaseUrl: this.mcpBaseUrl,
stt: () => this.speech?.resolveStt() ?? null, stt: () => this.speech?.resolveStt() ?? null,

444
packages/server/src/server/workspace-git-service.test.ts Normal file
View File

@@ -0,0 +1,444 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import type {
CheckoutStatusGit,
PullRequestStatusResult,
} from "../utils/checkout-git.js";
import {
WorkspaceGitServiceImpl,
type WorkspaceGitRuntimeSnapshot,
} from "./workspace-git-service.js";
function createLogger() {
const logger = {
child: () => logger,
debug: vi.fn(),
warn: vi.fn(),
};
return logger;
}
function createSnapshot(
cwd: string,
overrides?: {
git?: Partial<WorkspaceGitRuntimeSnapshot["git"]>;
github?: Partial<WorkspaceGitRuntimeSnapshot["github"]>;
},
): WorkspaceGitRuntimeSnapshot {
const base: WorkspaceGitRuntimeSnapshot = {
cwd,
git: {
isGit: true,
repoRoot: cwd,
mainRepoRoot: null,
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
isDirty: false,
aheadBehind: { ahead: 0, behind: 0 },
aheadOfOrigin: 0,
behindOfOrigin: 0,
diffStat: { additions: 1, deletions: 0 },
},
github: {
featuresEnabled: true,
pullRequest: {
url: "https://github.com/acme/repo/pull/123",
title: "Update feature",
state: "open",
baseRefName: "main",
headRefName: "feature",
isMerged: false,
},
error: null,
refreshedAt: "2026-04-12T00:00:00.000Z",
},
};
return {
cwd,
git: {
...base.git,
...overrides?.git,
},
github: {
...base.github,
...overrides?.github,
pullRequest:
overrides?.github && "pullRequest" in overrides.github
? overrides.github.pullRequest ?? null
: base.github.pullRequest,
error:
overrides?.github && "error" in overrides.github
? overrides.github.error ?? null
: base.github.error,
},
};
}
function createCheckoutStatus(
cwd: string,
overrides?: Partial<CheckoutStatusGit>,
): CheckoutStatusGit {
return {
isGit: true,
repoRoot: cwd,
currentBranch: "main",
isDirty: false,
baseRef: "main",
aheadBehind: { ahead: 0, behind: 0 },
aheadOfOrigin: 0,
behindOfOrigin: 0,
hasRemote: true,
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
...overrides,
};
}
function createPullRequestStatusResult(
overrides?: Partial<PullRequestStatusResult>,
): PullRequestStatusResult {
return {
status: {
url: "https://github.com/acme/repo/pull/123",
title: "Update feature",
state: "open",
baseRefName: "main",
headRefName: "feature",
isMerged: false,
},
githubFeaturesEnabled: true,
...overrides,
};
}
function createWatcher() {
return {
close: vi.fn(),
on: vi.fn().mockReturnThis(),
};
}
async function flushPromises(): Promise<void> {
await Promise.resolve();
await Promise.resolve();
}
function createService(options?: {
getCheckoutStatus?: ReturnType<typeof vi.fn>;
getCheckoutShortstat?: ReturnType<typeof vi.fn>;
getPullRequestStatus?: ReturnType<typeof vi.fn>;
resolveGhPath?: ReturnType<typeof vi.fn>;
resolveAbsoluteGitDir?: ReturnType<typeof vi.fn>;
hasOriginRemote?: ReturnType<typeof vi.fn>;
runGitFetch?: ReturnType<typeof vi.fn>;
watch?: ReturnType<typeof vi.fn>;
now?: () => Date;
}) {
return new WorkspaceGitServiceImpl({
logger: createLogger() as any,
paseoHome: "/tmp/paseo-test",
deps: {
watch: options?.watch ?? (((() => createWatcher()) as unknown) as any),
getCheckoutStatus:
options?.getCheckoutStatus ?? vi.fn(async (cwd: string) => createCheckoutStatus(cwd)),
getCheckoutShortstat:
options?.getCheckoutShortstat ??
vi.fn(async () => ({
additions: 1,
deletions: 0,
})),
getPullRequestStatus:
options?.getPullRequestStatus ??
vi.fn(async () => createPullRequestStatusResult()),
resolveGhPath: options?.resolveGhPath ?? vi.fn(async () => "/usr/bin/gh"),
resolveAbsoluteGitDir: options?.resolveAbsoluteGitDir ?? vi.fn(async () => "/tmp/repo/.git"),
hasOriginRemote: options?.hasOriginRemote ?? vi.fn(async () => false),
runGitFetch: options?.runGitFetch ?? vi.fn(async () => {}),
now: options?.now ?? (() => new Date("2026-04-12T00:00:00.000Z")),
},
});
}
describe("WorkspaceGitServiceImpl", () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test("subscribe returns an initial workspace runtime snapshot", async () => {
const service = createService();
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
expect(subscription.initial).toEqual(createSnapshot("/tmp/repo"));
expect(listener).not.toHaveBeenCalled();
subscription.unsubscribe();
service.dispose();
});
test("getSnapshot populates github pull request state in the runtime snapshot", async () => {
const getPullRequestStatus = vi.fn(async () =>
createPullRequestStatusResult({
status: {
url: "https://github.com/acme/repo/pull/999",
title: "Ship runtime centralization",
state: "open",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: false,
},
}),
);
const service = createService({
getPullRequestStatus,
now: () => new Date("2026-04-12T02:03:04.000Z"),
});
await expect(service.getSnapshot("/tmp/repo")).resolves.toEqual(
createSnapshot("/tmp/repo", {
github: {
pullRequest: {
url: "https://github.com/acme/repo/pull/999",
title: "Ship runtime centralization",
state: "open",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: false,
},
refreshedAt: "2026-04-12T02:03:04.000Z",
},
}),
);
expect(getPullRequestStatus).toHaveBeenCalledTimes(1);
service.dispose();
});
test("multiple listeners on the same workspace share one GitHub pull request lookup", async () => {
const getPullRequestStatus = vi.fn(async () => createPullRequestStatusResult());
const resolveAbsoluteGitDir = vi.fn(async () => "/tmp/repo/.git");
const service = createService({
getPullRequestStatus,
resolveAbsoluteGitDir,
});
const [first, second] = await Promise.all([
service.subscribe({ cwd: "/tmp/repo" }, vi.fn()),
service.subscribe({ cwd: "/tmp/repo" }, vi.fn()),
]);
expect(getPullRequestStatus).toHaveBeenCalledTimes(1);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(1);
expect((service as any).workspaceTargets.size).toBe(1);
first.unsubscribe();
second.unsubscribe();
service.dispose();
});
test("equivalent cwd strings share one workspace target across service entry points", async () => {
const getPullRequestStatus = vi.fn(async () => createPullRequestStatusResult());
const resolveAbsoluteGitDir = vi.fn(async () => "/tmp/repo/.git");
const service = createService({
getPullRequestStatus,
resolveAbsoluteGitDir,
});
const subscription = await service.subscribe({ cwd: "/tmp/repo/." }, vi.fn());
expect(subscription.initial).toEqual(createSnapshot("/tmp/repo"));
expect(service.peekSnapshot("/tmp/repo")).toEqual(createSnapshot("/tmp/repo"));
await service.refresh("/tmp/repo");
await expect(service.getSnapshot("/tmp/repo/.")).resolves.toEqual(createSnapshot("/tmp/repo"));
expect(getPullRequestStatus).toHaveBeenCalledTimes(2);
expect(resolveAbsoluteGitDir).toHaveBeenCalledTimes(1);
expect((service as any).workspaceTargets.size).toBe(1);
subscription.unsubscribe();
service.dispose();
});
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 service = createService({
resolveAbsoluteGitDir: vi.fn(async () => "/tmp/repo/.git"),
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();
expect(hasOriginRemote).toHaveBeenCalledTimes(1);
expect(runGitFetch).toHaveBeenCalledTimes(1);
expect((service as any).repoTargets.size).toBe(1);
await vi.advanceTimersByTimeAsync(180_000);
await flushPromises();
expect(runGitFetch).toHaveBeenCalledTimes(2);
first.unsubscribe();
second.unsubscribe();
service.dispose();
});
test("explicit refresh recomputes github state and notifies listeners", async () => {
const getPullRequestStatus = vi
.fn<() => Promise<PullRequestStatusResult>>()
.mockResolvedValueOnce(
createPullRequestStatusResult({
status: {
url: "https://github.com/acme/repo/pull/123",
title: "Before refresh",
state: "open",
baseRefName: "main",
headRefName: "feature",
isMerged: false,
},
}),
)
.mockResolvedValueOnce(
createPullRequestStatusResult({
status: {
url: "https://github.com/acme/repo/pull/123",
title: "After refresh",
state: "merged",
baseRefName: "main",
headRefName: "feature",
isMerged: true,
},
}),
);
const nowValues = [
new Date("2026-04-12T00:00:00.000Z"),
new Date("2026-04-12T00:05:00.000Z"),
];
const service = createService({
getPullRequestStatus,
now: () => nowValues.shift() ?? new Date("2026-04-12T00:05:00.000Z"),
});
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
expect(subscription.initial.github.pullRequest?.title).toBe("Before refresh");
service.refresh("/tmp/repo");
await (service as any).workspaceTargets.get("/tmp/repo")?.refreshPromise;
await flushPromises();
expect(getPullRequestStatus).toHaveBeenCalledTimes(2);
expect(listener).toHaveBeenCalledTimes(1);
expect(listener).toHaveBeenCalledWith(
createSnapshot("/tmp/repo", {
github: {
pullRequest: {
url: "https://github.com/acme/repo/pull/123",
title: "After refresh",
state: "merged",
baseRefName: "main",
headRefName: "feature",
isMerged: true,
},
refreshedAt: "2026-04-12T00:05:00.000Z",
},
}),
);
subscription.unsubscribe();
service.dispose();
});
test("unchanged runtime snapshots do not emit duplicate updates", async () => {
const getCheckoutStatus = vi
.fn<() => Promise<CheckoutStatusGit>>()
.mockResolvedValueOnce(createCheckoutStatus("/tmp/repo"))
.mockResolvedValueOnce(
createCheckoutStatus("/tmp/repo", {
currentBranch: "feature/runtime-payloads",
aheadBehind: { ahead: 2, behind: 0 },
aheadOfOrigin: 2,
}),
)
.mockResolvedValueOnce(
createCheckoutStatus("/tmp/repo", {
currentBranch: "feature/runtime-payloads",
aheadBehind: { ahead: 2, behind: 0 },
aheadOfOrigin: 2,
}),
);
const getPullRequestStatus = vi
.fn<() => Promise<PullRequestStatusResult>>()
.mockResolvedValue(
createPullRequestStatusResult({
status: {
url: "https://github.com/acme/repo/pull/123",
title: "Runtime payloads",
state: "open",
baseRefName: "main",
headRefName: "feature/runtime-payloads",
isMerged: false,
},
}),
);
const service = createService({
getCheckoutStatus,
getPullRequestStatus,
now: () => new Date("2026-04-12T00:00:00.000Z"),
});
const listener = vi.fn();
const subscription = await service.subscribe({ cwd: "/tmp/repo" }, listener);
expect(subscription.initial.git.currentBranch).toBe("main");
service.refresh("/tmp/repo");
await (service as any).workspaceTargets.get("/tmp/repo")?.refreshPromise;
await flushPromises();
service.refresh("/tmp/repo");
await (service as any).workspaceTargets.get("/tmp/repo")?.refreshPromise;
await flushPromises();
expect(listener).toHaveBeenCalledTimes(1);
expect(listener).toHaveBeenCalledWith(
createSnapshot("/tmp/repo", {
git: {
currentBranch: "feature/runtime-payloads",
aheadBehind: { ahead: 2, behind: 0 },
aheadOfOrigin: 2,
},
github: {
pullRequest: {
url: "https://github.com/acme/repo/pull/123",
title: "Runtime payloads",
state: "open",
baseRefName: "main",
headRefName: "feature/runtime-payloads",
isMerged: false,
},
},
}),
);
subscription.unsubscribe();
service.dispose();
});
});

588
packages/server/src/server/workspace-git-service.ts Normal file
View File

@@ -0,0 +1,588 @@
import { execFile } from "node:child_process";
import { watch, type FSWatcher } from "node:fs";
import { readFile } from "node:fs/promises";
import { join, resolve } from "node:path";
import { promisify } from "node:util";
import type pino from "pino";
import type { CheckoutContext } from "../utils/checkout-git.js";
import {
getCheckoutShortstat,
getCheckoutStatus,
getPullRequestStatus,
hasOriginRemote,
resolveGhPath,
resolveAbsoluteGitDir,
} from "../utils/checkout-git.js";
import { normalizeWorkspaceId } from "./workspace-registry-model.js";
const execFileAsync = promisify(execFile);
const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 500;
const BACKGROUND_GIT_FETCH_INTERVAL_MS = 180_000;
export type WorkspaceGitRuntimeSnapshot = {
cwd: string;
git: {
isGit: boolean;
repoRoot: string | null;
mainRepoRoot: string | null;
currentBranch: string | null;
remoteUrl: string | null;
isPaseoOwnedWorktree: boolean;
isDirty: boolean | null;
aheadBehind: { ahead: number; behind: number } | null;
aheadOfOrigin: number | null;
behindOfOrigin: number | null;
diffStat: { additions: number; deletions: number } | null;
};
github: {
featuresEnabled: boolean;
pullRequest: {
url: string;
title: string;
state: string;
baseRefName: string;
headRefName: string;
isMerged: boolean;
} | null;
error: { message: string } | null;
refreshedAt: string | null;
};
};
export interface WorkspaceGitService {
subscribe(
params: { cwd: string },
listener: WorkspaceGitListener,
): Promise<{
initial: WorkspaceGitRuntimeSnapshot;
unsubscribe: () => void;
}>;
peekSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot | null;
getSnapshot(cwd: string): Promise<WorkspaceGitRuntimeSnapshot>;
refresh(cwd: string, options?: { priority?: "normal" | "high" }): Promise<void>;
dispose(): void;
}
export type WorkspaceGitListener = (snapshot: WorkspaceGitRuntimeSnapshot) => void;
interface WorkspaceGitServiceDependencies {
watch: typeof watch;
getCheckoutStatus: typeof getCheckoutStatus;
getCheckoutShortstat: typeof getCheckoutShortstat;
getPullRequestStatus: typeof getPullRequestStatus;
resolveGhPath: typeof resolveGhPath;
resolveAbsoluteGitDir: (cwd: string) => Promise<string | null>;
hasOriginRemote: (cwd: string) => Promise<boolean>;
runGitFetch: (cwd: string) => Promise<void>;
now: () => Date;
}
interface WorkspaceGitServiceOptions {
logger: pino.Logger;
paseoHome: string;
deps?: Partial<WorkspaceGitServiceDependencies>;
}
interface WorkspaceGitTarget {
cwd: string;
listeners: Set<WorkspaceGitListener>;
watchers: FSWatcher[];
debounceTimer: NodeJS.Timeout | null;
refreshPromise: Promise<void> | null;
refreshQueued: boolean;
latestSnapshot: WorkspaceGitRuntimeSnapshot | null;
latestFingerprint: string | null;
repoGitRoot: string | null;
}
interface RepoGitTarget {
repoGitRoot: string;
cwd: string;
workspaceKeys: Set<string>;
intervalId: NodeJS.Timeout | null;
fetchInFlight: boolean;
}
export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private readonly logger: pino.Logger;
private readonly paseoHome: string;
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>>();
constructor(options: WorkspaceGitServiceOptions) {
this.logger = options.logger.child({ module: "workspace-git-service" });
this.paseoHome = options.paseoHome;
this.deps = {
watch,
getCheckoutStatus: options.deps?.getCheckoutStatus ?? getCheckoutStatus,
getCheckoutShortstat: options.deps?.getCheckoutShortstat ?? getCheckoutShortstat,
getPullRequestStatus: options.deps?.getPullRequestStatus ?? getPullRequestStatus,
resolveGhPath: options.deps?.resolveGhPath ?? resolveGhPath,
resolveAbsoluteGitDir: options.deps?.resolveAbsoluteGitDir ?? resolveAbsoluteGitDir,
hasOriginRemote: options.deps?.hasOriginRemote ?? hasOriginRemote,
runGitFetch: options.deps?.runGitFetch ?? runGitFetch,
now: options.deps?.now ?? (() => new Date()),
};
}
async subscribe(
params: { cwd: string },
listener: WorkspaceGitListener,
): Promise<{
initial: WorkspaceGitRuntimeSnapshot;
unsubscribe: () => void;
}> {
const cwd = normalizeWorkspaceId(params.cwd);
const target = await this.ensureWorkspaceTarget(cwd);
target.listeners.add(listener);
return {
initial: target.latestSnapshot ?? (await this.getSnapshot(cwd)),
unsubscribe: () => {
this.removeWorkspaceListener(cwd, listener);
},
};
}
async getSnapshot(cwd: string): Promise<WorkspaceGitRuntimeSnapshot> {
cwd = normalizeWorkspaceId(cwd);
const target = this.workspaceTargets.get(cwd);
if (target?.latestSnapshot) {
return target.latestSnapshot;
}
return this.refreshSnapshot(cwd);
}
peekSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot | null {
cwd = normalizeWorkspaceId(cwd);
return this.workspaceTargets.get(cwd)?.latestSnapshot ?? null;
}
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);
return;
}
await this.ensureWorkspaceTarget(cwd);
}
dispose(): void {
for (const target of this.workspaceTargets.values()) {
this.closeWorkspaceTarget(target);
}
this.workspaceTargets.clear();
for (const target of this.repoTargets.values()) {
this.closeRepoTarget(target);
}
this.repoTargets.clear();
this.workspaceTargetSetups.clear();
}
private async ensureWorkspaceTarget(cwd: string): Promise<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;
}
private async createWorkspaceTarget(cwd: string): Promise<WorkspaceGitTarget> {
const target: WorkspaceGitTarget = {
cwd,
listeners: new Set(),
watchers: [],
debounceTimer: null,
refreshPromise: null,
refreshQueued: false,
latestSnapshot: null,
latestFingerprint: null,
repoGitRoot: null,
};
const initial = await this.refreshSnapshot(cwd);
this.rememberSnapshot(target, initial);
this.workspaceTargets.set(cwd, target);
const gitDir = await this.deps.resolveAbsoluteGitDir(cwd);
if (!gitDir) {
return target;
}
const repoGitRoot = await this.resolveWorkspaceGitRefsRoot(gitDir);
target.repoGitRoot = repoGitRoot;
this.startWorkspaceWatchers(target, gitDir, repoGitRoot);
await this.ensureRepoTarget(target);
return target;
}
private async resolveWorkspaceGitRefsRoot(gitDir: string): Promise<string> {
try {
const commonDir = (await readFile(join(gitDir, "commondir"), "utf8")).trim();
if (commonDir.length > 0) {
return resolve(gitDir, commonDir);
}
} catch {
return gitDir;
}
return gitDir;
}
private startWorkspaceWatchers(
target: WorkspaceGitTarget,
gitDir: string,
repoGitRoot: string,
): void {
for (const watchPath of new Set([join(gitDir, "HEAD"), join(repoGitRoot, "refs", "heads")])) {
let watcher: FSWatcher | null = null;
try {
watcher = this.deps.watch(watchPath, { recursive: false }, () => {
this.scheduleWorkspaceRefresh(target);
});
} catch (error) {
this.logger.warn({ err: error, cwd: target.cwd, watchPath }, "Failed to start workspace git watcher");
}
if (!watcher) {
continue;
}
watcher.on("error", (error) => {
this.logger.warn({ err: error, cwd: target.cwd, watchPath }, "Workspace git watcher error");
});
target.watchers.push(watcher);
}
}
private async ensureRepoTarget(workspaceTarget: WorkspaceGitTarget): Promise<void> {
const repoGitRoot = workspaceTarget.repoGitRoot;
if (!repoGitRoot) {
return;
}
const existingTarget = this.repoTargets.get(repoGitRoot);
if (existingTarget) {
existingTarget.workspaceKeys.add(workspaceTarget.cwd);
return;
}
const hasOrigin = await this.deps.hasOriginRemote(workspaceTarget.cwd);
if (!hasOrigin) {
return;
}
const targetAfterProbe = this.repoTargets.get(repoGitRoot);
if (targetAfterProbe) {
targetAfterProbe.workspaceKeys.add(workspaceTarget.cwd);
return;
}
const repoTarget: RepoGitTarget = {
repoGitRoot,
cwd: workspaceTarget.cwd,
workspaceKeys: new Set([workspaceTarget.cwd]),
intervalId: setInterval(() => {
void this.runRepoFetch(repoTarget);
}, BACKGROUND_GIT_FETCH_INTERVAL_MS),
fetchInFlight: false,
};
this.repoTargets.set(repoGitRoot, repoTarget);
void this.runRepoFetch(repoTarget);
}
private scheduleWorkspaceRefresh(target: WorkspaceGitTarget): void {
if (target.debounceTimer) {
clearTimeout(target.debounceTimer);
}
target.debounceTimer = setTimeout(() => {
target.debounceTimer = null;
void this.refreshWorkspaceTarget(target);
}, WORKSPACE_GIT_WATCH_DEBOUNCE_MS);
}
private async refreshWorkspaceTarget(target: WorkspaceGitTarget): Promise<void> {
if (target.refreshPromise) {
target.refreshQueued = true;
return;
}
target.refreshPromise = (async () => {
do {
target.refreshQueued = false;
try {
const snapshot = await this.refreshSnapshot(target.cwd);
this.rememberSnapshot(target, snapshot, { notify: true });
} catch (error) {
this.logger.warn({ err: error, cwd: target.cwd }, "Failed to refresh workspace git snapshot");
}
} while (target.refreshQueued);
})();
try {
await target.refreshPromise;
} finally {
target.refreshPromise = null;
}
}
private async refreshSnapshot(cwd: string): Promise<WorkspaceGitRuntimeSnapshot> {
return loadWorkspaceGitRuntimeSnapshot(
cwd,
{ paseoHome: this.paseoHome },
this.deps.now(),
this.deps,
);
}
private rememberSnapshot(
target: WorkspaceGitTarget,
snapshot: WorkspaceGitRuntimeSnapshot,
options?: { notify?: boolean },
): void {
target.latestSnapshot = snapshot;
const fingerprint = JSON.stringify(snapshot);
if (target.latestFingerprint === fingerprint) {
return;
}
target.latestFingerprint = fingerprint;
if (!options?.notify) {
return;
}
for (const listener of target.listeners) {
listener(snapshot);
}
}
private async runRepoFetch(target: RepoGitTarget): Promise<void> {
if (target.fetchInFlight) {
return;
}
target.fetchInFlight = true;
this.logger.debug({ repoGitRoot: target.repoGitRoot, cwd: target.cwd }, "Running background git fetch");
try {
await this.deps.runGitFetch(target.cwd);
} catch (error) {
this.logger.warn(
{ err: error, repoGitRoot: target.repoGitRoot, cwd: target.cwd },
"Background git fetch failed",
);
} finally {
target.fetchInFlight = false;
await Promise.all(
Array.from(target.workspaceKeys, async (workspaceKey) => {
const workspaceTarget = this.workspaceTargets.get(workspaceKey);
if (!workspaceTarget) {
return;
}
await this.refreshWorkspaceTarget(workspaceTarget);
}),
);
}
}
private removeWorkspaceListener(cwd: string, listener: WorkspaceGitListener): void {
const target = this.workspaceTargets.get(cwd);
if (!target) {
return;
}
target.listeners.delete(listener);
if (target.listeners.size > 0) {
return;
}
this.removeWorkspaceTarget(target);
}
private removeWorkspaceTarget(target: WorkspaceGitTarget): void {
if (target.repoGitRoot) {
const repoTarget = this.repoTargets.get(target.repoGitRoot);
repoTarget?.workspaceKeys.delete(target.cwd);
if (repoTarget && repoTarget.workspaceKeys.size === 0) {
this.closeRepoTarget(repoTarget);
this.repoTargets.delete(target.repoGitRoot);
}
}
this.closeWorkspaceTarget(target);
this.workspaceTargets.delete(target.cwd);
}
private closeWorkspaceTarget(target: WorkspaceGitTarget): void {
if (target.debounceTimer) {
clearTimeout(target.debounceTimer);
target.debounceTimer = null;
}
for (const watcher of target.watchers) {
watcher.close();
}
target.watchers = [];
target.listeners.clear();
}
private closeRepoTarget(target: RepoGitTarget): void {
if (target.intervalId) {
clearInterval(target.intervalId);
target.intervalId = null;
}
target.workspaceKeys.clear();
}
}
async function loadWorkspaceGitRuntimeSnapshot(
cwd: string,
context: CheckoutContext,
now: Date,
deps: Pick<
WorkspaceGitServiceDependencies,
"getCheckoutStatus" | "getCheckoutShortstat" | "getPullRequestStatus" | "resolveGhPath"
>,
): Promise<WorkspaceGitRuntimeSnapshot> {
const checkoutStatus = await deps.getCheckoutStatus(cwd, context);
if (!checkoutStatus.isGit) {
return buildNotGitSnapshot(cwd);
}
const [diffStat, github] = await Promise.all([
deps.getCheckoutShortstat(cwd, context),
loadGitHubSnapshot({
cwd,
remoteUrl: checkoutStatus.remoteUrl,
now,
deps,
}),
]);
return {
cwd,
git: {
isGit: true,
repoRoot: checkoutStatus.repoRoot,
mainRepoRoot: checkoutStatus.isPaseoOwnedWorktree ? checkoutStatus.mainRepoRoot : null,
currentBranch: checkoutStatus.currentBranch,
remoteUrl: checkoutStatus.remoteUrl,
isPaseoOwnedWorktree: checkoutStatus.isPaseoOwnedWorktree,
isDirty: checkoutStatus.isDirty,
aheadBehind: checkoutStatus.aheadBehind,
aheadOfOrigin: checkoutStatus.aheadOfOrigin,
behindOfOrigin: checkoutStatus.behindOfOrigin,
diffStat,
},
github,
};
}
async function loadGitHubSnapshot(options: {
cwd: string;
remoteUrl: string | null;
now: Date;
deps: Pick<WorkspaceGitServiceDependencies, "getPullRequestStatus" | "resolveGhPath">;
}): Promise<WorkspaceGitRuntimeSnapshot["github"]> {
if (!hasGitHubRemoteUrl(options.remoteUrl)) {
return {
featuresEnabled: false,
pullRequest: null,
error: null,
refreshedAt: null,
};
}
try {
await options.deps.resolveGhPath();
} catch {
return {
featuresEnabled: false,
pullRequest: null,
error: null,
refreshedAt: null,
};
}
try {
const result = await options.deps.getPullRequestStatus(options.cwd);
return {
featuresEnabled: true,
pullRequest: result.status,
error: null,
refreshedAt: options.now.toISOString(),
};
} catch (error) {
return {
featuresEnabled: true,
pullRequest: null,
error: {
message: error instanceof Error ? error.message : String(error),
},
refreshedAt: options.now.toISOString(),
};
}
}
function hasGitHubRemoteUrl(remoteUrl: string | null): boolean {
if (!remoteUrl) {
return false;
}
return (
remoteUrl.includes("github.com/") ||
remoteUrl.startsWith("git@github.com:") ||
remoteUrl.startsWith("ssh://git@github.com/")
);
}
function buildNotGitSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot {
return {
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,
refreshedAt: null,
},
};
}
async function runGitFetch(cwd: string): Promise<void> {
await execFileAsync("git", ["fetch", "origin", "--prune"], {
cwd,
env: {
...process.env,
GIT_TERMINAL_PROMPT: "0",
},
});
}

2
packages/server/src/server/worktree-session.ts
View File

@@ -95,7 +95,7 @@ type RegisterPendingWorktreeWorkspaceDependencies = {
type CreatePaseoWorktreeInBackgroundDependencies = { type CreatePaseoWorktreeInBackgroundDependencies = {
paseoHome?: string; paseoHome?: string;
emitWorkspaceUpdateForCwd: (cwd: string, options?: { dedupeGitState?: boolean }) => Promise<void>; emitWorkspaceUpdateForCwd: (cwd: string) => Promise<void>;
sessionLogger: Logger; sessionLogger: Logger;
terminalManager: TerminalManager | null; terminalManager: TerminalManager | null;
}; };

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

@@ -1855,6 +1855,50 @@ export const ProjectPlacementPayloadSchema = z.object({
checkout: ProjectCheckoutLitePayloadSchema, checkout: ProjectCheckoutLitePayloadSchema,
}); });
const WorkspaceGitRuntimePayloadSchema = z
.object({
currentBranch: z.string().nullable().optional(),
remoteUrl: z.string().nullable().optional(),
isPaseoOwnedWorktree: z.boolean().optional(),
isDirty: z.boolean().nullable().optional(),
aheadBehind: z
.object({
ahead: z.number(),
behind: z.number(),
})
.nullable()
.optional(),
aheadOfOrigin: z.number().nullable().optional(),
behindOfOrigin: z.number().nullable().optional(),
})
.optional()
.nullable();
const WorkspaceGitHubRuntimePayloadSchema = z
.object({
featuresEnabled: z.boolean().optional(),
pullRequest: z
.object({
url: z.string(),
title: z.string(),
state: z.string(),
baseRefName: z.string(),
headRefName: z.string(),
isMerged: z.boolean(),
})
.nullable()
.optional(),
error: z
.object({
message: z.string(),
})
.nullable()
.optional(),
refreshedAt: z.string().nullable().optional(),
})
.optional()
.nullable();
export const WorkspaceDescriptorPayloadSchema = z.object({ export const WorkspaceDescriptorPayloadSchema = z.object({
id: z.string(), id: z.string(),
projectId: z.string(), projectId: z.string(),
@@ -1872,6 +1916,8 @@ export const WorkspaceDescriptorPayloadSchema = z.object({
}) })
.nullable() .nullable()
.optional(), .optional(),
gitRuntime: WorkspaceGitRuntimePayloadSchema,
githubRuntime: WorkspaceGitHubRuntimePayloadSchema,
}); });
export const AgentUpdateMessageSchema = z.object({ export const AgentUpdateMessageSchema = z.object({

157
packages/server/src/shared/messages.workspaces.test.ts
View File

@@ -1,3 +1,4 @@
import { z } from "zod";
import { describe, expect, test } from "vitest"; import { describe, expect, test } from "vitest";
import { SessionInboundMessageSchema, SessionOutboundMessageSchema } from "./messages.js"; import { SessionInboundMessageSchema, SessionOutboundMessageSchema } from "./messages.js";
@@ -117,6 +118,162 @@ describe("workspace message schemas", () => {
expect(result.success).toBe(false); expect(result.success).toBe(false);
}); });
test("parses fetch_workspaces_response with optional runtime fields", () => {
const parsed = SessionOutboundMessageSchema.parse({
type: "fetch_workspaces_response",
payload: {
requestId: "req-workspaces",
entries: [
{
id: "/tmp/repo",
projectId: "remote:github.com/acme/repo",
projectDisplayName: "acme/repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "local_checkout",
name: "main",
status: "done",
activityAt: null,
diffStat: {
additions: 3,
deletions: 1,
},
gitRuntime: {
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
isDirty: true,
aheadBehind: {
ahead: 2,
behind: 1,
},
aheadOfOrigin: 2,
behindOfOrigin: 1,
},
githubRuntime: {
featuresEnabled: true,
pullRequest: {
url: "https://github.com/acme/repo/pull/123",
title: "Runtime payloads",
state: "open",
baseRefName: "main",
headRefName: "workspace-git-service",
isMerged: false,
},
error: null,
refreshedAt: "2026-04-12T00:00:00.000Z",
},
},
],
pageInfo: {
nextCursor: null,
prevCursor: null,
hasMore: false,
},
},
});
expect(parsed.type).toBe("fetch_workspaces_response");
expect(parsed.payload.entries[0]?.gitRuntime).toMatchObject({
currentBranch: "main",
isDirty: true,
aheadOfOrigin: 2,
});
expect(parsed.payload.entries[0]?.githubRuntime?.pullRequest?.title).toBe("Runtime payloads");
});
test("older workspace parsers ignore additive runtime fields", () => {
const message = {
type: "fetch_workspaces_response",
payload: {
requestId: "req-workspaces",
entries: [
{
id: "/tmp/repo",
projectId: "remote:github.com/acme/repo",
projectDisplayName: "acme/repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "local_checkout",
name: "main",
status: "done",
activityAt: null,
diffStat: null,
gitRuntime: {
currentBranch: "main",
remoteUrl: "https://github.com/acme/repo.git",
isPaseoOwnedWorktree: false,
isDirty: false,
aheadBehind: {
ahead: 0,
behind: 0,
},
aheadOfOrigin: 0,
behindOfOrigin: 0,
},
githubRuntime: {
featuresEnabled: true,
pullRequest: null,
error: null,
refreshedAt: "2026-04-12T00:00:00.000Z",
},
},
],
pageInfo: {
nextCursor: null,
prevCursor: null,
hasMore: false,
},
},
};
const legacyWorkspaceSchema = z.object({
id: z.string(),
projectId: z.string(),
projectDisplayName: z.string(),
projectRootPath: z.string(),
projectKind: z.enum(["git", "non_git"]),
workspaceKind: z.enum(["local_checkout", "worktree", "directory"]),
name: z.string(),
status: z.enum(["needs_input", "failed", "running", "attention", "done"]),
activityAt: z.string().nullable(),
diffStat: z
.object({
additions: z.number(),
deletions: z.number(),
})
.nullable()
.optional(),
});
const legacyMessageSchema = z.object({
type: z.literal("fetch_workspaces_response"),
payload: z.object({
requestId: z.string(),
entries: z.array(legacyWorkspaceSchema),
pageInfo: z.object({
nextCursor: z.string().nullable(),
prevCursor: z.string().nullable(),
hasMore: z.boolean(),
}),
}),
});
const parsed = legacyMessageSchema.parse(message);
expect(parsed.payload.entries[0]).toEqual({
id: "/tmp/repo",
projectId: "remote:github.com/acme/repo",
projectDisplayName: "acme/repo",
projectRootPath: "/tmp/repo",
projectKind: "git",
workspaceKind: "local_checkout",
name: "main",
status: "done",
activityAt: null,
diffStat: null,
});
});
test("parses legacy fetch_agents_response checkout payloads without worktreeRoot", () => { test("parses legacy fetch_agents_response checkout payloads without worktreeRoot", () => {
const result = SessionOutboundMessageSchema.safeParse({ const result = SessionOutboundMessageSchema.safeParse({
type: "fetch_agents_response", type: "fetch_agents_response",

12
packages/server/src/utils/checkout-git.ts
View File

@@ -643,7 +643,7 @@ async function requireGitRepo(cwd: string): Promise<void> {
} }
} }
async function getCurrentBranch(cwd: string): Promise<string | null> { export async function getCurrentBranch(cwd: string): Promise<string | null> {
const { stdout } = await execAsync("git rev-parse --abbrev-ref HEAD", { const { stdout } = await execAsync("git rev-parse --abbrev-ref HEAD", {
cwd, cwd,
env: READ_ONLY_GIT_ENV, env: READ_ONLY_GIT_ENV,
@@ -665,7 +665,7 @@ async function getWorktreeRoot(cwd: string): Promise<string | null> {
} }
} }
async function getMainRepoRoot(cwd: string): Promise<string> { export async function getMainRepoRoot(cwd: string): Promise<string> {
const { stdout: commonDirOut } = await execAsync( const { stdout: commonDirOut } = await execAsync(
"git rev-parse --path-format=absolute --git-common-dir", "git rev-parse --path-format=absolute --git-common-dir",
{ cwd, env: READ_ONLY_GIT_ENV }, { cwd, env: READ_ONLY_GIT_ENV },
@@ -807,7 +807,7 @@ async function isWorkingTreeDirty(cwd: string): Promise<boolean> {
return stdout.trim().length > 0; return stdout.trim().length > 0;
} }
async function getOriginRemoteUrl(cwd: string): Promise<string | null> { export async function getOriginRemoteUrl(cwd: string): Promise<string | null> {
try { try {
const { stdout } = await execAsync("git config --get remote.origin.url", { const { stdout } = await execAsync("git config --get remote.origin.url", {
cwd, cwd,
@@ -820,12 +820,12 @@ async function getOriginRemoteUrl(cwd: string): Promise<string | null> {
} }
} }
async function hasOriginRemote(cwd: string): Promise<boolean> { export async function hasOriginRemote(cwd: string): Promise<boolean> {
const url = await getOriginRemoteUrl(cwd); const url = await getOriginRemoteUrl(cwd);
return url !== null; return url !== null;
} }
async function resolveAbsoluteGitDir(cwd: string): Promise<string | null> { export async function resolveAbsoluteGitDir(cwd: string): Promise<string | null> {
try { try {
const { stdout } = await execAsync("git rev-parse --absolute-git-dir", { const { stdout } = await execAsync("git rev-parse --absolute-git-dir", {
cwd, cwd,
@@ -1850,7 +1850,7 @@ export interface PullRequestStatusResult {
githubFeaturesEnabled: boolean; githubFeaturesEnabled: boolean;
} }
async function resolveGhPath(): Promise<string> { export async function resolveGhPath(): Promise<string> {
if (cachedGhPath === undefined) { if (cachedGhPath === undefined) {
cachedGhPath = await findExecutable("gh"); cachedGhPath = await findExecutable("gh");
} }