puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-09 18:04:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(server): persist workspace recovery before restoring agents

Browse Source 
Remove unused undo-list tracking and pass named restore requests. A failed workspace record write leaves its agents archived for retry.
This commit is contained in:
Mohamed Boudra
2026-11-09 16:10:27 +01:00
parent c7b067c19b
commit 58e2b2de9c
7 changed files with 115 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/bootstrap.ts
+7 -11
View File
@@ -978,17 +978,13 @@ export async function createPaseoDaemon(
}
});
const workspaceProvisioning = createWorkspaceProvisioningService({
unarchiveWorkspaceAgents: (workspaceId, archivedAt, allowLegacyFallback) =>
unarchiveWorkspaceAgents(
{
agentStorage,
unarchiveAgent: (id) => unarchiveAgentState(agentStorage, agentManager, id),
sessionLogger: logger,
},
workspaceId,
archivedAt,
allowLegacyFallback,
),
unarchiveWorkspaceAgents: (request) =>
unarchiveWorkspaceAgents({
...request,
agentStorage,
unarchiveAgent: (id) => unarchiveAgentState(agentStorage, agentManager, id),
sessionLogger: logger,
}),
lifecycle: pluginRuntime,
serverId,
projectRegistry,
packages/server/src/server/session.ts
+7 -11
View File
@@ -1004,17 +1004,13 @@ export class Session {
});
this.workspaceAutoName = workspaceAutoName;
this.workspaceProvisioning = createWorkspaceProvisioningService({
unarchiveWorkspaceAgents: (workspaceId, archivedAt, allowLegacyFallback) =>
unarchiveWorkspaceAgents(
{
agentStorage: this.agentStorage,
unarchiveAgent: (id) => unarchiveAgentState(this.agentStorage, this.agentManager, id),
sessionLogger: this.sessionLogger,
},
workspaceId,
archivedAt,
allowLegacyFallback,
),
unarchiveWorkspaceAgents: (request) =>
unarchiveWorkspaceAgents({
...request,
agentStorage: this.agentStorage,
unarchiveAgent: (id) => unarchiveAgentState(this.agentStorage, this.agentManager, id),
sessionLogger: this.sessionLogger,
}),
withCheckoutLock: this.workspaceBaseBranch.withCheckoutLock,
lifecycle: this.pluginRuntime,
serverId,
packages/server/src/server/session/workspace-provisioning/workspace-provisioning-service.test.ts
+43
View File
@@ -19,6 +19,7 @@ import {
FileBackedProjectRegistry,
FileBackedWorkspaceRegistry,
type PersistedProjectRecord,
type PersistedWorkspaceRecord,
createPersistedWorkspaceRecord,
type WorkspaceRegistry,
} from "../../workspace-registry.js";
@@ -1125,3 +1126,45 @@ test("a directory that goes away while the git read is in flight keeps its place
branch: "feature/vanishing",
});
});
test("a failed workspace unarchive write keeps agents archived and permits retry", async () => {
const repo = path.join(tmpDir, "restore-write-failure");
mkdirSync(repo, { recursive: true });
const workspace = await provisioning.findOrCreateWorkspaceForDirectory(repo);
await workspaceRegistry.archive(workspace.workspaceId, ARCHIVED_AT);
const archived = (await workspaceRegistry.get(workspace.workspaceId))!;
class FailingWorkspaceRegistry extends FileBackedWorkspaceRegistry {
failWrites = true;
override async upsert(record: PersistedWorkspaceRecord): Promise<void> {
if (this.failWrites && !record.archivedAt) throw new Error("Workspace write failed");
await super.upsert(record);
}
}
const registry = new FailingWorkspaceRegistry(
path.join(tmpDir, "projects", "workspaces.json"),
logger,
);
await registry.initialize();
let agentArchived = true;
const recovery = createWorkspaceProvisioningService({
workspaceRegistry: registry,
projectRegistry,
workspaceGitService: gitService(),
isDirectory,
logger,
unarchiveWorkspaceAgents: async () => {
agentArchived = false;
},
});
await expect(recovery.ensureWorkspaceRecordUnarchived(archived)).rejects.toThrow(
"Workspace write failed",
);
expect((await registry.get(workspace.workspaceId))?.archivedAt).toBe(ARCHIVED_AT);
expect(agentArchived).toBe(true);
registry.failWrites = false;
await recovery.ensureWorkspaceRecordUnarchived(archived);
expect((await registry.get(workspace.workspaceId))?.archivedAt).toBeNull();
expect(agentArchived).toBe(false);
});
packages/server/src/server/session/workspace-provisioning/workspace-provisioning-service.ts
+8 -12
View File
@@ -1,3 +1,4 @@
import type { WorkspaceArchiveRestoreRequest } from "../../workspace-archive-service.js";
import type { JsonValue } from "@getpaseo/protocol/agent-types";
import type { PluginLifecycle } from "../../plugins/lifecycle/index.js";
import { describeHookWorkspace } from "../../plugins/lifecycle/index.js";
@@ -105,11 +106,7 @@ export class WorkspaceProvisioningError extends Error {
}
export function createWorkspaceProvisioningService(deps: {
unarchiveWorkspaceAgents: (
workspaceId: string,
archivedAt: string,
allowLegacyFallback: boolean,
) => Promise<void>;
unarchiveWorkspaceAgents: (request: WorkspaceArchiveRestoreRequest) => Promise<void>;
serverId?: string;
workspaceRegistry: WorkspaceRegistry;
projectRegistry: ProjectRegistry;
@@ -487,17 +484,16 @@ export function createWorkspaceProvisioningService(deps: {
});
}
}
if (next) await workspaceRegistry.upsert(next);
// A canceled archive supplies a modern stamp before the record itself is archived.
if (archivedAt) {
await deps.unarchiveWorkspaceAgents(
workspace.workspaceId,
await deps.unarchiveWorkspaceAgents({
workspaceId: workspace.workspaceId,
archivedAt,
Boolean(workspace.archivedAt),
);
allowLegacyFallback: Boolean(workspace.archivedAt),
});
}
if (!next) return workspace;
await workspaceRegistry.upsert(next);
return next;
return next ?? workspace;
}
async function refreshWorkspaceRecord(
packages/server/src/server/workspace-archive-operations.test.ts
+3 -5
View File
@@ -19,10 +19,9 @@ function deferred<T = void>(): Deferred<T> {
return { promise, resolve, reject };
}
function archived(newlyArchivedAgentIds: string[]): ArchiveResult {
function archived(agentIds: string[]): ArchiveResult {
return {
archivedAgentIds: newlyArchivedAgentIds,
newlyArchivedAgentIds,
archivedAgentIds: agentIds,
archivedWorkspaceIds: ["ws"],
removedDirectory: false,
};
@@ -46,7 +45,7 @@ function parkedArchive(agentIds: string[]) {
signal = runSignal;
started.resolve();
await release.promise;
if (runSignal.aborted) throw new WorkspaceArchiveCanceledError(agentIds, false, "stamp");
if (runSignal.aborted) throw new WorkspaceArchiveCanceledError("stamp");
return archived(agentIds);
},
};
@@ -210,7 +209,6 @@ describe("WorkspaceArchiveOperations", () => {
await operations.archive("ws", async () => archived(["agent-a"]));
await operations.archive("ws", async () => ({
archivedAgentIds: ["agent-a"],
newlyArchivedAgentIds: [],
archivedWorkspaceIds: [],
removedDirectory: false,
}));
packages/server/src/server/workspace-archive-service.test.ts
+26 -88
View File
@@ -1060,7 +1060,7 @@ describe("archiveByScope cancellation", () => {
} as Pick<AgentStorage, "listByWorkspace">;
}
test("an abort while agents archive keeps the record and reports only the agents it took down", async () => {
test("an abort while agents archive keeps the workspace record active", async () => {
const { tempDir, repoDir } = createGitRepo();
const paseoHome = path.join(tempDir, ".paseo");
const worktree = await createPaseoOwnedWorktree(repoDir, paseoHome, "cancel-agents");
@@ -1090,16 +1090,12 @@ describe("archiveByScope cancellation", () => {
await agentArchive.reached;
controller.abort();
agentArchive.open();
const error = await canceledArchive(archive);
await canceledArchive(archive);
expect({
newlyArchivedAgentIds: error.newlyArchivedAgentIds,
recordArchived: error.recordArchived,
activeWorkspaceIds: (await deps.listActiveWorkspaces()).map((w) => w.workspaceId),
directoryExists: existsSync(worktree.worktreePath),
}).toEqual({
newlyArchivedAgentIds: ["agent-active"],
recordArchived: false,
activeWorkspaceIds: [workspaceId],
directoryExists: true,
});
@@ -1133,13 +1129,12 @@ describe("archiveByScope cancellation", () => {
await recordArchive.reached;
controller.abort();
recordArchive.open();
const error = await canceledArchive(archive);
await canceledArchive(archive);
expect({
recordArchived: error.recordArchived,
activeWorkspaceIds: (await deps.listActiveWorkspaces()).map((w) => w.workspaceId),
directoryExists: existsSync(worktree.worktreePath),
}).toEqual({ recordArchived: true, activeWorkspaceIds: [], directoryExists: true });
}).toEqual({ activeWorkspaceIds: [], directoryExists: true });
});
test("an archive aborted before it starts touches nothing", async () => {
@@ -1155,7 +1150,7 @@ describe("archiveByScope cancellation", () => {
const controller = new AbortController();
controller.abort();
const error = await canceledArchive(
await canceledArchive(
archiveByScope(deps, {
scope: { kind: "workspace", workspaceId },
requestId: "req-cancel-before",
@@ -1163,63 +1158,10 @@ describe("archiveByScope cancellation", () => {
}),
);
expect(error.newlyArchivedAgentIds).toEqual([]);
expect(deps.archivedSnapshotIds).toEqual([]);
expect(deps.markWorkspaceArchiving).not.toHaveBeenCalled();
expect(existsSync(worktree.worktreePath)).toBe(true);
});
test("a completed archive reports the agents it newly archived", async () => {
const { tempDir, repoDir } = createGitRepo();
const paseoHome = path.join(tempDir, ".paseo");
const worktree = await createPaseoOwnedWorktree(repoDir, paseoHome, "newly-archived");
const workspaceId = "ws-newly-archived";
const deps = createArchiveDeps({
paseoHome,
activeWorkspaces: [{ workspaceId, cwd: worktree.worktreePath, kind: "worktree" }],
});
deps.agentStorage = storedAgents(workspaceId, [
{ id: "agent-active", archivedAt: null },
{ id: "agent-already-archived", archivedAt: "2026-01-01T00:00:00.000Z" },
]);
const result = await archiveByScope(deps, {
scope: { kind: "workspace", workspaceId },
requestId: "req-newly-archived",
signal: new AbortController().signal,
});
expect(result.archivedAgentIds.sort()).toEqual(["agent-active", "agent-already-archived"]);
expect(result.newlyArchivedAgentIds).toEqual(["agent-active"]);
});
test("an archived agent loaded to read its history is not reported as newly archived", async () => {
const { tempDir, repoDir } = createGitRepo();
const paseoHome = path.join(tempDir, ".paseo");
const worktree = await createPaseoOwnedWorktree(repoDir, paseoHome, "loaded-archived");
const workspaceId = "ws-loaded-archived";
const deps = createArchiveDeps({
paseoHome,
activeWorkspaces: [{ workspaceId, cwd: worktree.worktreePath, kind: "worktree" }],
});
deps.agentStorage = storedAgents(workspaceId, [
{ id: "agent-active", archivedAt: null },
{ id: "agent-history", archivedAt: "2026-01-01T00:00:00.000Z" },
]);
deps.agentManager.listAgents = () =>
[
{ id: "agent-active", workspaceId },
{ id: "agent-history", workspaceId },
] as ManagedAgent[];
const result = await archiveByScope(deps, {
scope: { kind: "workspace", workspaceId },
requestId: "req-loaded-archived",
signal: new AbortController().signal,
});
expect(result.newlyArchivedAgentIds).toEqual(["agent-active"]);
});
});
describe("resolveWorkspaceIdAtPath", () => {
@@ -1317,38 +1259,34 @@ test.each([
"restores only agents belonging to the workspace archive ($expected)",
async ({ stamp, agents, expected }) => {
const restored: string[] = [];
await unarchiveWorkspaceAgents(
{
agentStorage: {
listByWorkspace: async () =>
agents.map(([id, archivedAt]) => ({ id, archivedAt })) as StoredAgentRecord[],
},
unarchiveAgent: async (id) => {
restored.push(id);
},
await unarchiveWorkspaceAgents({
workspaceId: "workspace",
archivedAt: stamp,
agentStorage: {
listByWorkspace: async () =>
agents.map(([id, archivedAt]) => ({ id, archivedAt })) as StoredAgentRecord[],
},
"workspace",
stamp,
);
unarchiveAgent: async (id) => {
restored.push(id);
},
});
expect(restored).toEqual(expected);
},
);
test("a canceled archive with no matching agents leaves earlier closed agents archived", async () => {
const restored: string[] = [];
await unarchiveWorkspaceAgents(
{
agentStorage: {
listByWorkspace: async () =>
[{ id: "closed", archivedAt: "2026-11-09T09:59:59.000Z" }] as StoredAgentRecord[],
},
unarchiveAgent: async (id) => {
restored.push(id);
},
await unarchiveWorkspaceAgents({
workspaceId: "workspace",
archivedAt: "2026-11-09T10:00:00.000Z",
allowLegacyFallback: false,
agentStorage: {
listByWorkspace: async () =>
[{ id: "closed", archivedAt: "2026-11-09T09:59:59.000Z" }] as StoredAgentRecord[],
},
"workspace",
"2026-11-09T10:00:00.000Z",
false,
);
unarchiveAgent: async (id) => {
restored.push(id);
},
});
expect(restored).toEqual([]);
});
packages/server/src/server/workspace-archive-service.ts
+21 -69
View File
@@ -67,9 +67,6 @@ export type ArchiveScope =
export interface ArchiveResult {
archivedAgentIds: string[];
// Agents this request moved from active to archived. `archivedAgentIds` also lists agents that
// were already archived; undo must restore only the ones this archive took down.
newlyArchivedAgentIds: string[];
archivedWorkspaceIds: string[];
removedDirectory: boolean;
// Set when the record archived but its worktree directory was left on
@@ -86,17 +83,9 @@ export interface ArchiveByScopeRequest {
signal?: AbortSignal;
}
/**
* The archive stopped at a checkpoint because its signal aborted. Whatever ran before the
* checkpoint stays done; `newlyArchivedAgentIds` and `recordArchived` say how far it got so the
* canceller can put it back.
*/
// Cancellation carries the operation stamp even before the workspace record is archived.
export class WorkspaceArchiveCanceledError extends Error {
constructor(
public readonly newlyArchivedAgentIds: string[],
public readonly recordArchived: boolean,
public readonly archivedAt: string,
) {
constructor(public readonly archivedAt: string) {
super("Workspace archive was canceled");
this.name = "WorkspaceArchiveCanceledError";
}
@@ -167,14 +156,9 @@ async function archiveByScopeWithPriority(
request: ArchiveByScopeRequest,
): Promise<ArchiveResult> {
const archivedAt = new Date().toISOString();
const progress: ArchiveProgress = { newlyArchivedAgentIds: new Set(), recordArchived: false };
const checkpoint = () => {
if (request.signal?.aborted) {
throw new WorkspaceArchiveCanceledError(
Array.from(progress.newlyArchivedAgentIds),
progress.recordArchived,
archivedAt,
);
throw new WorkspaceArchiveCanceledError(archivedAt);
}
};
@@ -201,7 +185,7 @@ async function archiveByScopeWithPriority(
const { archivedAgents, archivedWorkspaceIds } = await archiveTargetRecords(
dependencies,
targetWorkspaceIds,
{ requestId: request.requestId, checkpoint, progress, archivedAt },
{ requestId: request.requestId, checkpoint, archivedAt },
);
checkpoint();
@@ -232,7 +216,6 @@ async function archiveByScopeWithPriority(
return {
archivedAgentIds: Array.from(archivedAgents),
newlyArchivedAgentIds: Array.from(progress.newlyArchivedAgentIds),
archivedWorkspaceIds,
removedDirectory,
...(directoryLeftBehind ? { directoryLeftBehind } : {}),
@@ -245,11 +228,6 @@ async function archiveByScopeWithPriority(
}
}
interface ArchiveProgress {
newlyArchivedAgentIds: Set<string>;
recordArchived: boolean;
}
async function resolveArchiveTarget(
dependencies: ArchiveDependencies,
scope: ArchiveScope,
@@ -399,7 +377,6 @@ async function archiveTargetRecords(
options: {
requestId: string;
checkpoint: () => void;
progress: ArchiveProgress;
archivedAt: string;
},
): Promise<{ archivedAgents: Set<string>; archivedWorkspaceIds: string[] }> {
@@ -408,20 +385,16 @@ async function archiveTargetRecords(
const results = await Promise.allSettled(
targetWorkspaceIds.map(async (workspaceId) => {
const contents = await archiveWorkspaceContentsWithProgress(
const contents = await archiveWorkspaceContents(
dependencies,
workspaceId,
options.archivedAt,
);
for (const agentId of contents.newlyArchivedAgentIds) {
options.progress.newlyArchivedAgentIds.add(agentId);
}
// Agents and terminals are down, the record is not: an abort here leaves the workspace
// listed, and the canceller restores the agents it reports.
// listed, and cancellation restores agents carrying this operation stamp.
options.checkpoint();
await dependencies.archiveWorkspaceRecord(workspaceId, options.archivedAt);
options.progress.recordArchived = true;
return { workspaceId, agents: contents.agentIds };
return { workspaceId, agents: contents };
}),
);
@@ -432,11 +405,7 @@ async function archiveTargetRecords(
archivedAgents.add(agentId);
}
} else if (result.reason instanceof WorkspaceArchiveCanceledError) {
throw new WorkspaceArchiveCanceledError(
Array.from(options.progress.newlyArchivedAgentIds),
options.progress.recordArchived,
options.archivedAt,
);
throw result.reason;
} else {
dependencies.sessionLogger?.warn(
{ err: result.reason, requestId: options.requestId },
@@ -589,15 +558,6 @@ export async function archiveWorkspaceContents(
workspaceId: string,
archivedAt = new Date().toISOString(),
): Promise<Set<string>> {
return (await archiveWorkspaceContentsWithProgress(dependencies, workspaceId, archivedAt))
.agentIds;
}
async function archiveWorkspaceContentsWithProgress(
dependencies: ArchiveWorkspaceContentsDependencies,
workspaceId: string,
archivedAt: string,
): Promise<{ agentIds: Set<string>; newlyArchivedAgentIds: string[] }> {
const archivedAgents = new Set<string>();
const liveAgents = dependencies.agentManager
@@ -627,15 +587,6 @@ async function archiveWorkspaceContentsWithProgress(
...matchingStoredRecords.filter((record) => !record.archivedAt).map((record) => record.id),
]),
];
// Undo brings back only what this archive took down. Internal agents are retired rather than
// archived, and an agent loaded only to read its history was archived before this request.
const restorableAgentIds = new Set(agentIdsToArchive);
for (const agent of liveAgents) {
if (agent.internal) restorableAgentIds.delete(agent.id);
}
for (const record of matchingStoredRecords) {
if (record.archivedAt) restorableAgentIds.delete(record.id);
}
const [agentResults, [terminalResult]] = await Promise.all([
Promise.allSettled(
agentIdsToArchive.map((agentId) =>
@@ -647,7 +598,6 @@ async function archiveWorkspaceContentsWithProgress(
Promise.allSettled([dependencies.killTerminalsForWorkspace(workspaceId)]),
]);
const newlyArchivedAgentIds: string[] = [];
for (const [index, result] of agentResults.entries()) {
const agentId = agentIdsToArchive[index]!;
if (result.status === "rejected") {
@@ -655,8 +605,6 @@ async function archiveWorkspaceContentsWithProgress(
{ err: result.reason, workspaceId, agentId },
"Workspace archive teardown step failed; continuing",
);
} else if (restorableAgentIds.has(agentId)) {
newlyArchivedAgentIds.push(agentId);
}
}
if (terminalResult?.status === "rejected") {
@@ -666,7 +614,7 @@ async function archiveWorkspaceContentsWithProgress(
);
}
return { agentIds: archivedAgents, newlyArchivedAgentIds };
return archivedAgents;
}
// True when, after archiving
@@ -764,19 +712,23 @@ export async function archivePersistedWorkspaceRecord(input: {
return existingWorkspace;
}
interface UnarchiveWorkspaceAgentsDependencies {
export interface WorkspaceArchiveRestoreRequest {
workspaceId: string;
archivedAt: string;
allowLegacyFallback?: boolean;
}
interface UnarchiveWorkspaceAgentsInput extends WorkspaceArchiveRestoreRequest {
agentStorage: Pick<AgentStorage, "listByWorkspace">;
unarchiveAgent: (agentId: string) => Promise<unknown>;
sessionLogger?: Logger;
}
export async function unarchiveWorkspaceAgents(
dependencies: UnarchiveWorkspaceAgentsDependencies,
workspaceId: string,
archivedAt: string,
allowLegacyFallback = true,
input: UnarchiveWorkspaceAgentsInput,
): Promise<void> {
const records = await dependencies.agentStorage.listByWorkspace(workspaceId);
const { workspaceId, archivedAt, allowLegacyFallback = true } = input;
const records = await input.agentStorage.listByWorkspace(workspaceId);
const exact = records.filter((record) => record.archivedAt === archivedAt);
// COMPAT(workspaceArchiveStamp): added in v0.15.3; remove after 2027-05-09 once old workspace archives no longer need restoration.
const matching =
@@ -788,11 +740,11 @@ export async function unarchiveWorkspaceAgents(
return age >= 0 && age <= 60_000;
});
const results = await Promise.allSettled(
matching.map((record) => dependencies.unarchiveAgent(record.id)),
matching.map((record) => input.unarchiveAgent(record.id)),
);
for (const [index, result] of results.entries()) {
if (result.status === "rejected") {
dependencies.sessionLogger?.warn(
input.sessionLogger?.warn(
{ err: result.reason, workspaceId, agentId: matching[index]!.id },
"Failed to unarchive workspace agent",
);