Make workspace archive self-healing during setup (#2987)
* fix(server): make workspace archive self-healing Background setup could keep writing after its workspace record was archived, causing disk cleanup to fail and the initiating client to restore a local ghost. Track setup per daemon so archive can stop it, and let repeated archive requests finish residual cleanup after the record has committed. * fix(server): await setup process termination Archive must not proceed from setup cancellation until the process-tree terminator has completed. Join that promise to the streamed command's existing finish path so child close cannot release workspace cleanup early.
This commit is contained in:
@@ -117,6 +117,7 @@ export async function fanOutReconciledWorkspaceUpdates(input: {
|
||||
}
|
||||
|
||||
import { VoiceAssistantWebSocketServer } from "./websocket-server.js";
|
||||
import { WorkspaceSetupRuntime } from "./workspace-setup-runtime.js";
|
||||
import { createGitHubService } from "../services/github-service.js";
|
||||
import { createPaseoWorktree as createRegisteredPaseoWorktree } from "./paseo-worktree-service.js";
|
||||
import { createWorkspaceProvisioningService } from "./session/workspace-provisioning/workspace-provisioning-service.js";
|
||||
@@ -605,6 +606,7 @@ export async function createPaseoDaemon(
|
||||
publicBaseUrl: serviceProxyPublicBaseUrl,
|
||||
});
|
||||
const scriptRuntimeStore = new WorkspaceScriptRuntimeStore();
|
||||
const workspaceSetupRuntime = new WorkspaceSetupRuntime();
|
||||
const configuredHostnames = config.hostnames ?? config.allowedHosts;
|
||||
let wsServer: VoiceAssistantWebSocketServer | null = null;
|
||||
let serviceProxyListenTarget: ListenTarget | null = null;
|
||||
@@ -1039,6 +1041,8 @@ export async function createPaseoDaemon(
|
||||
await emitWorkspaceUpdatesExternal([workspaceId]);
|
||||
},
|
||||
cacheWorkspaceSetupSnapshot: () => {},
|
||||
startWorkspaceSetup: (workspaceId, operation) =>
|
||||
workspaceSetupRuntime.start(workspaceId, operation),
|
||||
emit: emitExternalSessionMessage,
|
||||
sessionLogger: logger,
|
||||
terminalManager,
|
||||
@@ -1079,12 +1083,14 @@ export async function createPaseoDaemon(
|
||||
agentStorage,
|
||||
findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
getWorkspace: (workspaceIdToGet) => workspaceRegistry.get(workspaceIdToGet),
|
||||
archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
markWorkspaceArchiving: markWorkspaceArchivingExternal,
|
||||
clearWorkspaceArchiving: clearWorkspaceArchivingExternal,
|
||||
killTerminalsForWorkspace: (workspaceIdToKill) =>
|
||||
killTerminalsForWorkspace({ terminalManager, sessionLogger: logger }, workspaceIdToKill),
|
||||
stopWorkspaceSetup: (workspaceIdToStop) => workspaceSetupRuntime.stop(workspaceIdToStop),
|
||||
sessionLogger: logger,
|
||||
},
|
||||
{ scope: { kind: "workspace", workspaceId }, requestId },
|
||||
@@ -1187,6 +1193,7 @@ export async function createPaseoDaemon(
|
||||
agentStorage,
|
||||
findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
getWorkspace: (workspaceIdToGet) => workspaceRegistry.get(workspaceIdToGet),
|
||||
archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
markWorkspaceArchiving: markWorkspaceArchivingExternal,
|
||||
@@ -1199,6 +1206,7 @@ export async function createPaseoDaemon(
|
||||
},
|
||||
workspaceIdToKill,
|
||||
),
|
||||
stopWorkspaceSetup: (workspaceIdToStop) => workspaceSetupRuntime.stop(workspaceIdToStop),
|
||||
sessionLogger: logger,
|
||||
},
|
||||
{
|
||||
@@ -1564,6 +1572,7 @@ export async function createPaseoDaemon(
|
||||
serviceProxyPublicBaseUrl,
|
||||
browserToolsBroker,
|
||||
hubRelationships,
|
||||
workspaceSetupRuntime,
|
||||
);
|
||||
relayRuntime = createRelayRuntime({
|
||||
config: {
|
||||
|
||||
@@ -173,7 +173,6 @@ import { PushTokenStore } from "./push/token-store.js";
|
||||
import {
|
||||
archivePersistedWorkspaceRecord,
|
||||
archiveWorkspaceContents,
|
||||
requireActiveWorkspaceForArchive,
|
||||
} from "./workspace-archive-service.js";
|
||||
import type { ServiceProxySubsystem } from "./service-proxy.js";
|
||||
import { renameCurrentBranch as renameCurrentBranchDefault } from "../utils/checkout-git.js";
|
||||
@@ -240,6 +239,13 @@ import {
|
||||
handleWorkspaceSetupStatusRequest as handleWorkspaceSetupStatusRequestMessage,
|
||||
} from "./worktree-session.js";
|
||||
import { archiveByScope, type ActiveWorkspaceRef } from "./workspace-archive-service.js";
|
||||
import { WorkspaceSetupRuntime } from "./workspace-setup-runtime.js";
|
||||
|
||||
function resolveWorkspaceSetupRuntime(
|
||||
runtime: WorkspaceSetupRuntime | undefined,
|
||||
): WorkspaceSetupRuntime {
|
||||
return runtime ?? new WorkspaceSetupRuntime();
|
||||
}
|
||||
import { WorktreeRequestError, toWorktreeWireError } from "./worktree-errors.js";
|
||||
import { parseGitRemoteLocation } from "@getpaseo/protocol/git-remote";
|
||||
import {
|
||||
@@ -458,6 +464,7 @@ export interface SessionOptions {
|
||||
serviceProxy?: ServiceProxySubsystem;
|
||||
scriptRuntimeStore?: WorkspaceScriptRuntimeStore;
|
||||
workspaceSetupSnapshots?: Map<string, WorkspaceSetupSnapshot>;
|
||||
workspaceSetupRuntime?: WorkspaceSetupRuntime;
|
||||
onBranchChanged?: (
|
||||
workspaceId: string,
|
||||
oldBranch: string | null,
|
||||
@@ -649,6 +656,7 @@ export class Session {
|
||||
private inflightRequests = 0;
|
||||
private peakInflightRequests = 0;
|
||||
private readonly workspaceSetupSnapshots: Map<string, WorkspaceSetupSnapshot>;
|
||||
private readonly workspaceSetupRuntime: WorkspaceSetupRuntime;
|
||||
private readonly workspaceGitObserver: WorkspaceGitObserverService;
|
||||
private readonly workspaceDirectory: WorkspaceDirectory;
|
||||
private readonly voiceSession: VoiceSession;
|
||||
@@ -704,6 +712,7 @@ export class Session {
|
||||
serviceProxy,
|
||||
scriptRuntimeStore,
|
||||
workspaceSetupSnapshots,
|
||||
workspaceSetupRuntime,
|
||||
onBranchChanged,
|
||||
getDaemonTcpPort,
|
||||
getDaemonTcpHost,
|
||||
@@ -955,6 +964,7 @@ export class Session {
|
||||
this.serviceProxy = serviceProxy ?? null;
|
||||
this.scriptRuntimeStore = scriptRuntimeStore ?? null;
|
||||
this.workspaceSetupSnapshots = workspaceSetupSnapshots ?? new Map();
|
||||
this.workspaceSetupRuntime = resolveWorkspaceSetupRuntime(workspaceSetupRuntime);
|
||||
this.getDaemonTcpPort = getDaemonTcpPort ?? null;
|
||||
this.getDaemonTcpHost = getDaemonTcpHost ?? null;
|
||||
this.serviceProxyPublicBaseUrl = serviceProxyPublicBaseUrl ?? null;
|
||||
@@ -5890,6 +5900,8 @@ export class Session {
|
||||
this.workspaceAutoName.scheduleForWorktree(autoNameInput, {
|
||||
currentSelection: this.getFocusedAgentSelectionForCwd(autoNameInput.workspace.cwd),
|
||||
}),
|
||||
startWorkspaceSetup: (workspaceId, operation) =>
|
||||
this.workspaceSetupRuntime.start(workspaceId, operation),
|
||||
emitWorkspaceUpdateForWorkspaceId: (workspaceId) =>
|
||||
this.emitWorkspaceUpdateForWorkspaceId(workspaceId),
|
||||
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => {
|
||||
@@ -5929,10 +5941,10 @@ export class Session {
|
||||
request: Extract<SessionInboundMessage, { type: "archive_workspace_request" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const existing = await requireActiveWorkspaceForArchive(
|
||||
{ listActiveWorkspaces: () => this.listActiveWorkspaceRefs() },
|
||||
request.workspaceId,
|
||||
);
|
||||
const existing = await this.workspaceRegistry.get(request.workspaceId);
|
||||
if (!existing) {
|
||||
throw new Error(`Workspace not found: ${request.workspaceId}`);
|
||||
}
|
||||
|
||||
await archiveByScope(
|
||||
{
|
||||
@@ -5943,6 +5955,7 @@ export class Session {
|
||||
agentManager: this.agentManager,
|
||||
agentStorage: this.agentStorage,
|
||||
findWorkspaceIdForCwd: (cwd) => this.findWorkspaceIdForCwd(cwd),
|
||||
getWorkspace: (workspaceId) => this.workspaceRegistry.get(workspaceId),
|
||||
listActiveWorkspaces: () => this.listActiveWorkspaceRefs(),
|
||||
archiveWorkspaceRecord: (workspaceId) => this.archiveWorkspaceRecord(workspaceId),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds) =>
|
||||
@@ -5952,6 +5965,7 @@ export class Session {
|
||||
clearWorkspaceArchiving: (workspaceIds) => this.clearWorkspaceArchiving(workspaceIds),
|
||||
killTerminalsForWorkspace: (workspaceId) =>
|
||||
this.terminalController.killTerminalsForWorkspace(workspaceId),
|
||||
stopWorkspaceSetup: (workspaceId) => this.workspaceSetupRuntime.stop(workspaceId),
|
||||
sessionLogger: this.sessionLogger,
|
||||
},
|
||||
{
|
||||
|
||||
@@ -34,6 +34,7 @@ import type { HostnamesConfig } from "./hostnames.js";
|
||||
import { isHostnameAllowed } from "./hostnames.js";
|
||||
import { Session, type SessionLifecycleIntent, type SessionRuntimeMetrics } from "./session.js";
|
||||
import type { HubRelationshipManagement } from "./hub/relationship-controller.js";
|
||||
import { WorkspaceSetupRuntime } from "./workspace-setup-runtime.js";
|
||||
import type { HubExecutionAgents } from "./hub/daemon-executions.js";
|
||||
import type { AgentProvider } from "./agent/agent-sdk-types.js";
|
||||
import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
|
||||
@@ -537,6 +538,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
private readonly voiceSpeakHandlers = new Map<string, VoiceSpeakHandler>();
|
||||
private readonly voiceCallerContexts = new Map<string, VoiceCallerContext>();
|
||||
private readonly workspaceSetupSnapshots = new Map<string, WorkspaceSetupSnapshot>();
|
||||
private readonly workspaceSetupRuntime: WorkspaceSetupRuntime;
|
||||
private readonly providerSnapshotManager: ProviderSnapshotManager;
|
||||
private onLifecycleIntent!: ((intent: SessionLifecycleIntent) => void) | null;
|
||||
private onBranchChanged!:
|
||||
@@ -604,8 +606,10 @@ export class VoiceAssistantWebSocketServer {
|
||||
serviceProxyPublicBaseUrl?: string | null,
|
||||
browserToolsBroker?: BrowserToolsBroker | null,
|
||||
hubRelationships?: HubRelationshipManagement | null,
|
||||
workspaceSetupRuntime: WorkspaceSetupRuntime = new WorkspaceSetupRuntime(),
|
||||
) {
|
||||
this.logger = logger.child({ module: "websocket-server" });
|
||||
this.workspaceSetupRuntime = workspaceSetupRuntime;
|
||||
this.advertiseDaemonStatusRpc = wsConfig.daemonStatusRpc !== false;
|
||||
this.advertiseRelayConfig = wsConfig.relayConfig !== false;
|
||||
this.serverId = serverId;
|
||||
@@ -1348,6 +1352,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
serviceProxy: this.serviceProxy ?? undefined,
|
||||
scriptRuntimeStore: this.scriptRuntimeStore ?? undefined,
|
||||
workspaceSetupSnapshots: this.workspaceSetupSnapshots,
|
||||
workspaceSetupRuntime: this.workspaceSetupRuntime,
|
||||
onBranchChanged: this.onBranchChanged ?? undefined,
|
||||
getDaemonTcpPort: this.getDaemonTcpPort ?? undefined,
|
||||
getDaemonTcpHost: this.getDaemonTcpHost ?? undefined,
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { existsSync, mkdtempSync, rmSync } from "node:fs";
|
||||
import { execFileSync, spawn } from "node:child_process";
|
||||
import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, expect, test } from "vitest";
|
||||
@@ -300,6 +300,118 @@ test("archiving the last reference to a worktree removes it from disk regardless
|
||||
await expect.poll(() => existsSync(deleteDir), { timeout: 10000, interval: 100 }).toBe(false);
|
||||
}, 60000);
|
||||
|
||||
test.skipIf(process.platform === "win32")(
|
||||
"archiving a worktree is repeatable while background setup is still writing into it",
|
||||
async () => {
|
||||
const repoDir = createGitRepo();
|
||||
const setupStartedPath = path.join(repoDir, "setup-started");
|
||||
const stopSetupPath = path.join(repoDir, "stop-setup");
|
||||
writeFileSync(
|
||||
path.join(repoDir, "paseo.json"),
|
||||
JSON.stringify({
|
||||
worktree: {
|
||||
setup: [
|
||||
`node -e "const fs=require('fs'),path=require('path');const source=process.env.PASEO_SOURCE_CHECKOUT_PATH;const worktree=process.env.PASEO_WORKTREE_PATH;const target=path.join(worktree,'node_modules/react-native-svg/lib/typescript');fs.writeFileSync(path.join(source,'setup-started'),'started');while(!fs.existsSync(path.join(source,'stop-setup'))){try{fs.mkdirSync(target,{recursive:true});fs.writeFileSync(path.join(target,'active'),String(Date.now()))}catch{}}"`,
|
||||
],
|
||||
},
|
||||
}),
|
||||
);
|
||||
execFileSync("git", ["add", "paseo.json"], { cwd: repoDir, stdio: "pipe" });
|
||||
execFileSync(
|
||||
"git",
|
||||
["-c", "commit.gpgsign=false", "commit", "-m", "add active worktree setup"],
|
||||
{ cwd: repoDir, stdio: "pipe" },
|
||||
);
|
||||
|
||||
const result = await ctx.client.createWorkspace({
|
||||
source: {
|
||||
kind: "worktree",
|
||||
cwd: repoDir,
|
||||
worktreeSlug: "archive-during-setup",
|
||||
baseBranch: "main",
|
||||
},
|
||||
});
|
||||
const workspace = result.workspace;
|
||||
if (!workspace?.workspaceDirectory) {
|
||||
throw new Error(result.error ?? "Failed to create worktree workspace");
|
||||
}
|
||||
|
||||
try {
|
||||
await expect.poll(() => existsSync(setupStartedPath), { timeout: 10000 }).toBe(true);
|
||||
|
||||
const archive = await ctx.client.archiveWorkspace(workspace.id);
|
||||
const retry = await ctx.client.archiveWorkspace(workspace.id);
|
||||
|
||||
expect({ archiveError: archive.error, retryError: retry.error }).toEqual({
|
||||
archiveError: null,
|
||||
retryError: null,
|
||||
});
|
||||
expect((await activeWorkspaceIds()).has(workspace.id)).toBe(false);
|
||||
expect(existsSync(workspace.workspaceDirectory)).toBe(false);
|
||||
} finally {
|
||||
writeFileSync(stopSetupPath, "stop\n");
|
||||
}
|
||||
},
|
||||
60000,
|
||||
);
|
||||
|
||||
test.skipIf(process.platform === "win32")(
|
||||
"repeating archive removes a residual worktree for an already-archived workspace",
|
||||
async () => {
|
||||
const repoDir = createGitRepo();
|
||||
const result = await ctx.client.createWorkspace({
|
||||
source: {
|
||||
kind: "worktree",
|
||||
cwd: repoDir,
|
||||
worktreeSlug: "self-heal-archive",
|
||||
baseBranch: "main",
|
||||
},
|
||||
});
|
||||
const workspace = result.workspace;
|
||||
if (!workspace?.workspaceDirectory) {
|
||||
throw new Error(result.error ?? "Failed to create worktree workspace");
|
||||
}
|
||||
|
||||
const writerStartedPath = path.join(repoDir, "writer-started");
|
||||
const stopWriterPath = path.join(repoDir, "stop-writer");
|
||||
const target = path.join(
|
||||
workspace.workspaceDirectory,
|
||||
"node_modules/react-native-svg/lib/typescript",
|
||||
);
|
||||
const writer = spawn(
|
||||
process.execPath,
|
||||
[
|
||||
"-e",
|
||||
`const fs=require('fs'),path=require('path');const target=${JSON.stringify(target)};const started=${JSON.stringify(writerStartedPath)};const stop=${JSON.stringify(stopWriterPath)};fs.writeFileSync(started,'started');while(!fs.existsSync(stop)){try{fs.mkdirSync(target,{recursive:true});fs.writeFileSync(path.join(target,'active'),String(Date.now()))}catch{}}`,
|
||||
],
|
||||
{ stdio: "ignore" },
|
||||
);
|
||||
const writerExit = new Promise<void>((resolve) => writer.once("exit", () => resolve()));
|
||||
|
||||
try {
|
||||
await expect.poll(() => existsSync(writerStartedPath), { timeout: 10000 }).toBe(true);
|
||||
|
||||
const firstArchive = await ctx.client.archiveWorkspace(workspace.id);
|
||||
|
||||
expect(firstArchive.error).toBeNull();
|
||||
expect((await activeWorkspaceIds()).has(workspace.id)).toBe(false);
|
||||
expect(existsSync(workspace.workspaceDirectory)).toBe(true);
|
||||
|
||||
writeFileSync(stopWriterPath, "stop\n");
|
||||
await writerExit;
|
||||
|
||||
const retry = await ctx.client.archiveWorkspace(workspace.id);
|
||||
|
||||
expect(retry.error).toBeNull();
|
||||
expect(existsSync(workspace.workspaceDirectory)).toBe(false);
|
||||
} finally {
|
||||
writeFileSync(stopWriterPath, "stop\n");
|
||||
await writerExit;
|
||||
}
|
||||
},
|
||||
60000,
|
||||
);
|
||||
|
||||
test("worktree archive targets the explicit workspaceId when a directory backs multiple workspaces", async () => {
|
||||
const repoDir = createGitRepo();
|
||||
|
||||
|
||||
@@ -38,6 +38,7 @@ export interface ArchiveDependencies {
|
||||
// path uniquely identifies a worktree workspace; this is a directory lookup for
|
||||
// the archive target, not status/ownership.
|
||||
findWorkspaceIdForCwd: (cwd: string) => Promise<string | null>;
|
||||
getWorkspace?: (workspaceId: string) => Promise<PersistedWorkspaceRecord | null>;
|
||||
// Active (non-archived) workspaces, used to decide whether the workspace being
|
||||
// archived is the last reference to its backing worktree directory, and to
|
||||
// break a same-cwd tie in favor of the worktree-kind record when archiving by
|
||||
@@ -48,6 +49,7 @@ export interface ArchiveDependencies {
|
||||
markWorkspaceArchiving: (workspaceIds: Iterable<string>, archivingAt: string) => void;
|
||||
clearWorkspaceArchiving: (workspaceIds: Iterable<string>) => void;
|
||||
killTerminalsForWorkspace: (workspaceId: string) => Promise<void>;
|
||||
stopWorkspaceSetup?: (workspaceId: string) => Promise<void>;
|
||||
sessionLogger?: Logger;
|
||||
}
|
||||
|
||||
@@ -95,6 +97,7 @@ interface BackingDirectory {
|
||||
interface ArchiveTarget {
|
||||
backing: BackingDirectory | null;
|
||||
teardownTargets: Array<{ workspaceId: string | null; cwd: string }>;
|
||||
setupWorkspaceIds: string[];
|
||||
workspaceIds: string[];
|
||||
}
|
||||
|
||||
@@ -129,6 +132,8 @@ async function archiveByScopeWithPriority(
|
||||
const target = await resolveArchiveTarget(dependencies, request.scope);
|
||||
const targetWorkspaceIds = target.workspaceIds;
|
||||
|
||||
await stopWorkspaceSetups(dependencies, target.setupWorkspaceIds, request.requestId);
|
||||
|
||||
if (targetWorkspaceIds.length > 0) {
|
||||
dependencies.markWorkspaceArchiving(targetWorkspaceIds, new Date().toISOString());
|
||||
}
|
||||
@@ -190,18 +195,22 @@ async function resolveArchiveTarget(
|
||||
|
||||
if (scope.kind === "workspace") {
|
||||
const workspaceId = scope.workspaceId;
|
||||
const record = activeWorkspaces.find((workspace) => workspace.workspaceId === workspaceId);
|
||||
const record =
|
||||
(await dependencies.getWorkspace?.(workspaceId)) ??
|
||||
activeWorkspaces.find((workspace) => workspace.workspaceId === workspaceId);
|
||||
if (!record) {
|
||||
dependencies.sessionLogger?.warn(
|
||||
{ workspaceId },
|
||||
"Workspace not found for archive-by-scope; skipping",
|
||||
);
|
||||
return { backing: null, teardownTargets: [], workspaceIds: [] };
|
||||
return { backing: null, teardownTargets: [], setupWorkspaceIds: [], workspaceIds: [] };
|
||||
}
|
||||
const isArchived = "archivedAt" in record && Boolean(record.archivedAt);
|
||||
return {
|
||||
backing: await resolveWorkspaceBackingDirectory(record, dependencies),
|
||||
teardownTargets: [{ workspaceId, cwd: record.cwd }],
|
||||
workspaceIds: [workspaceId],
|
||||
teardownTargets: isArchived ? [] : [{ workspaceId, cwd: record.cwd }],
|
||||
setupWorkspaceIds: [workspaceId],
|
||||
workspaceIds: isArchived ? [] : [workspaceId],
|
||||
};
|
||||
}
|
||||
|
||||
@@ -230,10 +239,32 @@ async function resolveArchiveTarget(
|
||||
cwd: workspace.cwd,
|
||||
}))
|
||||
: [{ workspaceId: null, cwd: scope.targetPath }],
|
||||
setupWorkspaceIds: targetWorkspaces.map((workspace) => workspace.workspaceId),
|
||||
workspaceIds: targetWorkspaces.map((workspace) => workspace.workspaceId),
|
||||
};
|
||||
}
|
||||
|
||||
async function stopWorkspaceSetups(
|
||||
dependencies: ArchiveDependencies,
|
||||
workspaceIds: string[],
|
||||
requestId: string,
|
||||
): Promise<void> {
|
||||
if (!dependencies.stopWorkspaceSetup) {
|
||||
return;
|
||||
}
|
||||
const results = await Promise.allSettled(
|
||||
workspaceIds.map((workspaceId) => dependencies.stopWorkspaceSetup!(workspaceId)),
|
||||
);
|
||||
for (const [index, result] of results.entries()) {
|
||||
if (result?.status === "rejected") {
|
||||
dependencies.sessionLogger?.warn(
|
||||
{ err: result.reason, workspaceId: workspaceIds[index], requestId },
|
||||
"Failed to stop workspace setup during archive; continuing",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function resolveWorkspaceBackingDirectory(
|
||||
workspace: ActiveWorkspaceRef,
|
||||
dependencies: Pick<ArchiveDependencies, "paseoHome" | "paseoWorktreesBaseRoot">,
|
||||
@@ -379,14 +410,11 @@ async function maybeRemoveDirectory(
|
||||
dependencies.github.invalidate({ cwd: backing.path });
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error instanceof WorktreeTeardownError) {
|
||||
dependencies.sessionLogger?.warn(
|
||||
{ err: error, targetPath: backing.path, requestId: request.requestId },
|
||||
"Worktree disk removal failed during archive; workspace already archived",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
throw error;
|
||||
dependencies.sessionLogger?.warn(
|
||||
{ err: error, targetPath: backing.path, requestId: request.requestId },
|
||||
"Worktree disk removal failed during archive; workspace already archived",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
export type WorkspaceSetupOperation = (signal: AbortSignal) => Promise<void>;
|
||||
|
||||
interface WorkspaceSetupRun {
|
||||
controller: AbortController;
|
||||
completion: Promise<void>;
|
||||
}
|
||||
|
||||
export class WorkspaceSetupRuntime {
|
||||
private readonly runs = new Map<string, WorkspaceSetupRun>();
|
||||
|
||||
start(workspaceId: string, operation: WorkspaceSetupOperation): void {
|
||||
const controller = new AbortController();
|
||||
const run: WorkspaceSetupRun = {
|
||||
controller,
|
||||
completion: Promise.resolve(),
|
||||
};
|
||||
this.runs.set(workspaceId, run);
|
||||
run.completion = Promise.resolve()
|
||||
.then(() => operation(controller.signal))
|
||||
.catch(() => undefined)
|
||||
.finally(() => {
|
||||
if (this.runs.get(workspaceId) === run) {
|
||||
this.runs.delete(workspaceId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async stop(workspaceId: string): Promise<void> {
|
||||
const run = this.runs.get(workspaceId);
|
||||
if (!run) {
|
||||
return;
|
||||
}
|
||||
run.controller.abort();
|
||||
await run.completion;
|
||||
}
|
||||
}
|
||||
@@ -47,6 +47,7 @@ import {
|
||||
createPaseoWorktreeCommand,
|
||||
listPaseoWorktreesCommand,
|
||||
} from "./worktree/commands.js";
|
||||
import type { WorkspaceSetupOperation } from "./workspace-setup-runtime.js";
|
||||
|
||||
const SAFE_GIT_REF_PATTERN = /^[A-Za-z0-9._/-]+$/;
|
||||
|
||||
@@ -123,6 +124,7 @@ interface CreatePaseoWorktreeWorkflowDependencies extends CreatePaseoWorktreeInB
|
||||
workspace: PersistedWorkspaceRecord;
|
||||
firstAgentContext: FirstAgentContext;
|
||||
}) => void;
|
||||
startWorkspaceSetup?: (workspaceId: string, operation: WorkspaceSetupOperation) => void;
|
||||
}
|
||||
|
||||
interface AgentWorktreeSetupContinuationInput {
|
||||
@@ -619,16 +621,26 @@ export async function createPaseoWorktreeWorkflow(
|
||||
);
|
||||
});
|
||||
if (setupContinuation.kind === "workspace") {
|
||||
void runWorktreeSetupInBackground(dependencies, {
|
||||
requestCwd: input.cwd,
|
||||
repoRoot: createdWorktree.repoRoot,
|
||||
workspaceId: workspace.workspaceId,
|
||||
worktree: createdWorktree.worktree,
|
||||
shouldBootstrap: createdWorktree.created,
|
||||
slug,
|
||||
worktreePath: createdWorktree.worktree.worktreePath,
|
||||
workspaceCwd: workspace.cwd,
|
||||
});
|
||||
const runSetup = (signal: AbortSignal) =>
|
||||
runWorktreeSetupInBackground(
|
||||
dependencies,
|
||||
{
|
||||
requestCwd: input.cwd,
|
||||
repoRoot: createdWorktree.repoRoot,
|
||||
workspaceId: workspace.workspaceId,
|
||||
worktree: createdWorktree.worktree,
|
||||
shouldBootstrap: createdWorktree.created,
|
||||
slug,
|
||||
worktreePath: createdWorktree.worktree.worktreePath,
|
||||
workspaceCwd: workspace.cwd,
|
||||
},
|
||||
signal,
|
||||
);
|
||||
if (dependencies.startWorkspaceSetup) {
|
||||
dependencies.startWorkspaceSetup(workspace.workspaceId, runSetup);
|
||||
} else {
|
||||
void runSetup(new AbortController().signal);
|
||||
}
|
||||
}
|
||||
}, 0);
|
||||
|
||||
@@ -687,6 +699,7 @@ export async function runWorktreeSetupInBackground(
|
||||
worktreePath: string;
|
||||
workspaceCwd?: string;
|
||||
},
|
||||
signal?: AbortSignal,
|
||||
): Promise<void> {
|
||||
let worktree: WorktreeConfig = options.worktree;
|
||||
let setupResults: WorktreeSetupCommandResult[] = [];
|
||||
@@ -746,6 +759,7 @@ export async function runWorktreeSetupInBackground(
|
||||
cleanupOnFailure: false,
|
||||
repoRootPath: options.repoRoot,
|
||||
runtimeEnv,
|
||||
signal,
|
||||
onEvent: (event) => {
|
||||
applyWorktreeSetupProgressEvent(progressAccumulator, event);
|
||||
emitSetupProgress("running", null);
|
||||
|
||||
@@ -37,6 +37,7 @@ import { createExternalProcessEnv } from "../server/paseo-env.js";
|
||||
import { parseGitRevParsePath, resolveGitRevParsePath } from "./git-rev-parse-path.js";
|
||||
import { validateBranchSlug } from "@getpaseo/protocol/branch-slug";
|
||||
import { expandTilde, getRealpathAwareRelativePath, isPathInsideRoot } from "./path.js";
|
||||
import { terminateWithTreeKill } from "./tree-kill.js";
|
||||
|
||||
export { slugify, validateBranchSlug } from "@getpaseo/protocol/branch-slug";
|
||||
|
||||
@@ -449,6 +450,7 @@ async function execSetupCommandStreamed(options: {
|
||||
env: NodeJS.ProcessEnv;
|
||||
index: number;
|
||||
total: number;
|
||||
signal?: AbortSignal;
|
||||
onEvent?: (event: WorktreeSetupCommandProgressEvent) => void;
|
||||
}): Promise<WorktreeSetupCommandResult> {
|
||||
return new Promise((resolvePromise) => {
|
||||
@@ -456,6 +458,7 @@ async function execSetupCommandStreamed(options: {
|
||||
const stdoutChunks: string[] = [];
|
||||
const stderrChunks: string[] = [];
|
||||
let settled = false;
|
||||
let termination: Promise<unknown> | null = null;
|
||||
|
||||
const emitOutput = (stream: "stdout" | "stderr", chunk: string) => {
|
||||
const text = stripAnsi(chunk);
|
||||
@@ -478,11 +481,13 @@ async function execSetupCommandStreamed(options: {
|
||||
});
|
||||
};
|
||||
|
||||
const finish = (exitCode: number | null) => {
|
||||
const finish = async (exitCode: number | null) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
options.signal?.removeEventListener("abort", abort);
|
||||
await termination;
|
||||
const result: WorktreeSetupCommandResult = {
|
||||
command: options.command,
|
||||
cwd: options.cwd,
|
||||
@@ -521,6 +526,18 @@ async function execSetupCommandStreamed(options: {
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
|
||||
const abort = () => {
|
||||
termination ??= terminateWithTreeKill(child, {
|
||||
gracefulTimeoutMs: 1000,
|
||||
forceTimeoutMs: 1000,
|
||||
});
|
||||
};
|
||||
if (options.signal?.aborted) {
|
||||
abort();
|
||||
} else {
|
||||
options.signal?.addEventListener("abort", abort, { once: true });
|
||||
}
|
||||
|
||||
child.stdout?.on("data", (chunk: Buffer | string) => {
|
||||
emitOutput("stdout", chunk.toString());
|
||||
});
|
||||
@@ -531,11 +548,11 @@ async function execSetupCommandStreamed(options: {
|
||||
|
||||
child.on("error", (error) => {
|
||||
emitOutput("stderr", error instanceof Error ? error.message : String(error));
|
||||
finish(null);
|
||||
void finish(null);
|
||||
});
|
||||
|
||||
child.on("close", (code) => {
|
||||
finish(typeof code === "number" ? code : null);
|
||||
void finish(typeof code === "number" ? code : null);
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -621,6 +638,7 @@ export async function runWorktreeSetupCommands(options: {
|
||||
cleanupOnFailure: boolean;
|
||||
repoRootPath?: string;
|
||||
runtimeEnv?: WorktreeRuntimeEnv;
|
||||
signal?: AbortSignal;
|
||||
onEvent?: (event: WorktreeSetupCommandProgressEvent) => void;
|
||||
}): Promise<WorktreeSetupCommandResult[]> {
|
||||
// Read paseo.json from the worktree (it will have the same content as the source repo)
|
||||
@@ -647,6 +665,7 @@ export async function runWorktreeSetupCommands(options: {
|
||||
env: setupEnv,
|
||||
index: index + 1,
|
||||
total: setupCommands.length,
|
||||
signal: options.signal,
|
||||
onEvent: options.onEvent,
|
||||
})
|
||||
: await execSetupCommand(cmd, {
|
||||
|
||||
Reference in New Issue
Block a user