mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-06 23:20:53 +01:00
fix(workspaces): preserve legacy recovery across storage migration
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
2a16dcf629
commit
420b93c7b7
9 files changed
+240
-20
No files matched your search
@@ -45,3 +45,27 @@ yet make the whole tenant eligible for enrollment.
|
||||
bindings, archived sources, overlap/writer races, and unavailable sources.
|
||||
- Unrelated tenants, production services, and agent task execution remain outside
|
||||
the migration's mutation scope.
|
||||
|
||||
## Legacy recovery storage transition
|
||||
|
||||
The migration owner must distinguish file location from recovery completion.
|
||||
An offline import can preserve a terminal failed run's local workspace without
|
||||
claiming that its remote files were recovered. The snapshot and migration
|
||||
receipt bind the original run and sync descriptor; the failed run, provider
|
||||
allocation, recovery action, and unresolved state stay recorded.
|
||||
|
||||
After publication, the native sync owner must reject copyback to the old local
|
||||
folder. Explicit recovery still validates the exact provider allocation and
|
||||
stop evidence, then preserves remote files as an immutable recovery attachment
|
||||
instead of merging into retired local authority. New remote allocations attach
|
||||
those verified recovery copies through the existing attachment owner. Missing
|
||||
or ambiguous remote sources remain unresolved; migration never authorizes
|
||||
replaying the old agent turn or destroying its allocation.
|
||||
|
||||
Publication requires the offline deployment fence, terminal unowned runs,
|
||||
unchanged descriptors and task bindings, verified local snapshot bytes, and a
|
||||
single transaction publishing the migration receipt and remote authority.
|
||||
Controller leases, in-flight recovery, active writers, and uncertain workspace
|
||||
operations remain blockers. Tests must prove migration leaves failure state
|
||||
intact, rejects stale copyback even if the old folder exists, and preserves
|
||||
recovery source validation and company boundaries.
|
||||
@@ -272,7 +272,7 @@ function preserveWorkspaceMetadataForPatch(
|
||||
if (patch[key] !== undefined && patch[key] !== current[key]) throw conflict("Remote workspace coordinates belong to its sandbox storage");
|
||||
}
|
||||
}
|
||||
const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles"] as const;
|
||||
const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles", "legacyWorkspaceRecoveryMigration"] as const;
|
||||
if (patch.metadata !== undefined && remoteAuthorityKeys.some(key => current.metadata?.[key] !== undefined || patch.metadata?.[key] !== undefined)) {
|
||||
nextPatch.metadata = { ...patch.metadata };
|
||||
for (const key of remoteAuthorityKeys) {
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { executionWorkspaces, issues, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import type { NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
|
||||
export type LegacyWorkspaceRecoverySource = { runId: string; issueId: string; reference: NativeWorkspaceSyncReference };
|
||||
export const LEGACY_WORKSPACE_RECOVERY_MIGRATION = "paperclip.legacy-workspace-recovery-migration.v1";
|
||||
|
||||
/** Child tasks can explicitly reuse an ancestor's workspace. The run's own
|
||||
* task remains the scope for its recovery files; migration never rebinds it. */
|
||||
export async function isLegacyWorkspaceTaskRelated(db: Pick<Db, "select">, input: {
|
||||
companyId: string; issueId: string | null; sourceIssueId: string | null;
|
||||
}) {
|
||||
let current = input.issueId;
|
||||
const visited = new Set<string>();
|
||||
while (current && visited.size < 32 && !visited.has(current)) {
|
||||
if (current === input.sourceIssueId) return true;
|
||||
visited.add(current);
|
||||
const [issue] = await db.select({ parentId: issues.parentId }).from(issues)
|
||||
.where(and(eq(issues.companyId, input.companyId), eq(issues.id, current)));
|
||||
current = issue?.parentId ?? null;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/** This receipt changes where recovery may save files; it never certifies that
|
||||
* a provider stopped or that its files were recovered. The recovery owner must
|
||||
* still validate allocation provenance and termination before exporting. */
|
||||
export async function hasMigratedLegacyWorkspaceSource(db: Pick<Db, "select">, input: {
|
||||
companyId: string; runId: string; reference: NativeWorkspaceSyncReference;
|
||||
}) {
|
||||
if (input.reference.state !== "prepared") return false;
|
||||
const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.reference.workspaceId),
|
||||
));
|
||||
if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return false;
|
||||
const marker = workspace?.metadata?.legacyWorkspaceRecoveryMigration as { schema?: unknown; operationId?: unknown } | undefined;
|
||||
if (marker?.schema !== LEGACY_WORKSPACE_RECOVERY_MIGRATION || typeof marker.operationId !== "string")
|
||||
throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
const [operation] = await db.select({ metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.executionWorkspaceId, input.reference.workspaceId),
|
||||
eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"),
|
||||
eq(workspaceOperations.status, "succeeded"),
|
||||
));
|
||||
const sources = operation?.metadata?.legacyRecoverySources;
|
||||
if (!Array.isArray(sources) || !sources.some(source => source && typeof source === "object"
|
||||
&& source.runId === input.runId && isDeepStrictEqual(source.reference, input.reference)))
|
||||
throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
return true;
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { hasMigratedLegacyWorkspaceSource } from "../legacy-workspace-recovery-migration.js";
|
||||
import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
|
||||
import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js";
|
||||
import { measureRunPhase } from "../execution-preparation-timing.js";
|
||||
@@ -921,7 +922,7 @@ export async function prepareNativeWorkspaceSync(input: {
|
||||
throw new Error("native_remote_recovery_lease_mismatch");
|
||||
}
|
||||
const run = await measureRunPhase("workspace_sync.run_read", () => input.db
|
||||
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson })
|
||||
.select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.runId))
|
||||
.limit(1)
|
||||
@@ -1184,7 +1185,7 @@ export async function resumeNativeWorkspaceSync(input: {
|
||||
throw new Error("workspace_sync_out_unrecoverable");
|
||||
}
|
||||
const run = await input.db
|
||||
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson })
|
||||
.select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.runId))
|
||||
.limit(1)
|
||||
@@ -1193,6 +1194,8 @@ export async function resumeNativeWorkspaceSync(input: {
|
||||
parseObject(run?.runnerProfileJson).nativeWorkspaceSync,
|
||||
);
|
||||
if (!reference) return false;
|
||||
if (run && await hasMigratedLegacyWorkspaceSource(input.db, { companyId: run.companyId, runId: input.runId, reference }))
|
||||
throw new Error("native_workspace_sync_source_migrated");
|
||||
const existing = await readDescriptor({ runId: input.runId, reference });
|
||||
const providerLeaseId =
|
||||
input.target.sandboxLeaseAcquisition?.providerLeaseId ??
|
||||
|
||||
@@ -8,7 +8,7 @@ import { createServer, type Server } from "node:http";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { activityLog, agents, heartbeatRuns, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots,
|
||||
import { activityLog, agents, heartbeatRuns, nativeRunFinalizations, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots,
|
||||
executionWorkspaceRuntimeLeases, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
|
||||
@@ -25,6 +25,7 @@ import type { TaskDirectoryRestorePointRef } from "./isolated-task-directory-ret
|
||||
import { resolveManagedProjectWorkspaceDir } from "../home-paths.js";
|
||||
import { cleanupExecutionWorkspaceArtifacts } from "./workspace-runtime.js";
|
||||
import { workspaceOperationService } from "./workspace-operations.js";
|
||||
import { resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js";
|
||||
import { managedProjectCheckoutName } from "./managed-project-checkout-name.js";
|
||||
import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js";
|
||||
|
||||
@@ -51,7 +52,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
await Promise.all(servers.splice(0).map(server => new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()))));
|
||||
vi.unstubAllEnvs(); await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots);
|
||||
await db.delete(workspaceRuntimeServices); await db.delete(executionWorkspaceRuntimeLeases);
|
||||
await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects);
|
||||
await db.delete(nativeRunFinalizations); await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects);
|
||||
await db.delete(companies); await fs.rm(root, { recursive: true, force: true });
|
||||
});
|
||||
async function seed() {
|
||||
@@ -140,6 +141,75 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: reference!.manifestDigest });
|
||||
expect(await fs.readFile(path.join(targetDir, ".paperclip/worktrees/child/child.txt"), "utf8")).toBe("saved child edits\n");
|
||||
});
|
||||
async function failedCopybackFixture() {
|
||||
const f = await seed(), runId = randomUUID();
|
||||
const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Historical agent" }).returning();
|
||||
const reference = { schema: "paperclip.native-workspace-sync/v3", state: "prepared", workspaceId: f.workspaceId,
|
||||
leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/workspaces/task",
|
||||
descriptorSha256: "12".repeat(32), baselineSha256: "34".repeat(32), finalHostSha256: null, resourceDisposition: "stop_and_retain" };
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: agent!.id, nativeIssueId: f.issueId,
|
||||
runtimeMode: "native", status: "failed", finishedAt: new Date(), runnerProfileJson: { nativeWorkspaceSync: reference } });
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId, phase: "terminal_failure" });
|
||||
return { ...f, runId, reference };
|
||||
}
|
||||
it("carries unresolved copyback forward during offline migration without declaring recovery complete", async () => {
|
||||
const f = await failedCopybackFixture();
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_copyback_unresolved");
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true });
|
||||
const row = await workspace(f.workspaceId);
|
||||
expect(readRemoteWorkspaceAuthority(row.metadata)?.snapshotId).toBeTruthy();
|
||||
expect(row.metadata?.remoteWorkspaceRecoverySourceRunIds).toEqual([f.runId]);
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(run).toMatchObject({ status: "failed", runnerProfileJson: { nativeWorkspaceSync: f.reference } });
|
||||
expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
|
||||
// An old copyback must not recreate or mutate the predecessor's local folder.
|
||||
await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } }))
|
||||
.rejects.toThrow("native_workspace_sync_source_migrated");
|
||||
});
|
||||
it.each(["controller", "finalizer", "retryable", "task", "descriptor"])("keeps unsafe legacy copyback blocked (%s)", async mode => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
if (mode === "controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (mode === "finalizer") await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (mode === "retryable") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (mode === "task") {
|
||||
const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Different task" }).returning();
|
||||
await db.update(executionWorkspaces).set({ sourceIssueId: other!.id }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
}
|
||||
if (mode === "descriptor") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, descriptorSha256: "invalid" } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_copyback_unresolved");
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
|
||||
});
|
||||
it("rejects a legacy descriptor changed while its local files are uploading", async () => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
const put = storage.putFile.bind(storage);
|
||||
storage.putFile = async input => {
|
||||
const result = await put(input);
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, providerLeaseId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
return result;
|
||||
};
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_workspace_migration_source_changed");
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
|
||||
});
|
||||
it("preserves a child task's recovery reference to its ancestor's workspace", async () => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Parent task" }).returning();
|
||||
await db.update(issues).set({ parentId: parent!.id }).where(eq(issues.id, f.issueId));
|
||||
await db.update(executionWorkspaces).set({ sourceIssueId: parent!.id }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true });
|
||||
expect((await workspace(f.workspaceId)).sourceIssueId).toBe(parent!.id);
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.executionWorkspaceId).toBe(f.workspaceId);
|
||||
await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } }))
|
||||
.rejects.toThrow("native_workspace_sync_source_migrated");
|
||||
});
|
||||
it("fails closed if the durable legacy migration receipt changes", async () => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true });
|
||||
const row = await workspace(f.workspaceId);
|
||||
await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, legacyWorkspaceRecoveryMigration: { operationId: randomUUID() } } }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } }))
|
||||
.rejects.toThrow("legacy_workspace_migration_receipt_invalid");
|
||||
});
|
||||
it("blocks distinct legacy records that alias the same folder", async () => {
|
||||
const f = await seed(), id = randomUUID();
|
||||
await db.insert(executionWorkspaces).values({ id, companyId: f.companyId, name: "Historical alias", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: f.cwd });
|
||||
|
||||
@@ -17,7 +17,9 @@ import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-rete
|
||||
import { materializeTaskDirectorySnapshot, removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js";
|
||||
import { loadTaskDirectoryRestorePoint, mayBeIsolatedTaskDirectoryWorkspace, readTaskDirectoryRetention, validateTaskDirectoryRestorePointRef } from "./isolated-task-directory-retention.js";
|
||||
import { inspectArchivedGitMigrationSource, materializeArchivedGitMigrationSource, type ArchivedGitMigrationSource } from "./remote-workspace-migration-git.js";
|
||||
import { prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
import { isLegacyWorkspaceTaskRelated, LEGACY_WORKSPACE_RECOVERY_MIGRATION, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js";
|
||||
import { offlineWorkspaceMigrationBlockers, prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js";
|
||||
|
||||
const SCHEMA = "paperclip.remote-workspace-migration.v1";
|
||||
type Workspace = typeof executionWorkspaces.$inferSelect;
|
||||
@@ -46,7 +48,8 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri
|
||||
const marker = object(workspace.metadata?.remoteWorkspaceMigration);
|
||||
const allowedOperation = input.operationId ?? (marker.schema === SCHEMA && typeof marker.operationId === "string" ? marker.operationId : undefined);
|
||||
const reasons: string[] = [];
|
||||
if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null };
|
||||
if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] };
|
||||
if (input.writersStopped) reasons.push(...await offlineWorkspaceMigrationBlockers(db));
|
||||
if (!["local_fs", "git_worktree"].includes(workspace.providerType)) reasons.push("workspace_provider_unsupported");
|
||||
if (!["active", "idle", "in_review", "archived"].includes(workspace.status)) reasons.push("workspace_not_active");
|
||||
if (!workspace.sourceIssueId) reasons.push("task_binding_required");
|
||||
@@ -96,11 +99,25 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri
|
||||
or ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'workspaceId' = ${workspace.id})`,
|
||||
)).limit(1);
|
||||
if (run) reasons.push("active_execution");
|
||||
const [unexported] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, workspace.companyId),
|
||||
const unexported = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode,
|
||||
issueId: heartbeatRuns.nativeIssueId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt,
|
||||
reference: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`,
|
||||
}).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, workspace.companyId),
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'workspaceId' = ${workspace.id}`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared'`,
|
||||
)).limit(1);
|
||||
if (unexported) reasons.push("remote_copyback_unresolved");
|
||||
));
|
||||
const legacyRecoverySources: LegacyWorkspaceRecoverySource[] = [];
|
||||
for (const run of unexported.sort((a, b) => a.id.localeCompare(b.id))) {
|
||||
const reference = readNativeWorkspaceSyncReference(run.reference);
|
||||
const [owner] = await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner })
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.runId, run.id)));
|
||||
if (!input.writersStopped || !reference || run.runtimeMode !== "native" || !run.issueId
|
||||
|| !await isLegacyWorkspaceTaskRelated(db, { companyId: workspace.companyId, issueId: run.issueId, sourceIssueId: workspace.sourceIssueId })
|
||||
|| !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status)
|
||||
|| run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()
|
||||
|| !owner || owner.leaseOwner || !["committed", "terminal_failure"].includes(owner.phase)) reasons.push("remote_copyback_unresolved");
|
||||
else legacyRecoverySources.push({ runId: run.id, issueId: run.issueId, reference });
|
||||
}
|
||||
const [finalization] = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.issueId, workspace.sourceIssueId),
|
||||
sql`${nativeRunFinalizations.phase} not in ('committed', 'terminal_failure')`,
|
||||
@@ -129,11 +146,14 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri
|
||||
) : undefined),
|
||||
)).limit(1);
|
||||
if (service) reasons.push("workspace_runtime_service_unsettled");
|
||||
const [recovery] = workspace.sourceIssueId ? await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, workspace.companyId), eq(issueRecoveryActions.sourceIssueId, workspace.sourceIssueId),
|
||||
const recoveryIssueIds = [...new Set([workspace.sourceIssueId, ...legacyRecoverySources.map(source => source.issueId)].filter((id): id is string => Boolean(id)))];
|
||||
const recoveries = recoveryIssueIds.length ? await db.select({ id: issueRecoveryActions.id, evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, workspace.companyId), inArray(issueRecoveryActions.sourceIssueId, recoveryIssueIds),
|
||||
inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
)).limit(1) : [];
|
||||
if (recovery) reasons.push("workspace_recovery_unresolved");
|
||||
)) : [];
|
||||
if (recoveries.some(recovery => !input.writersStopped
|
||||
|| !legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId)
|
||||
|| (recovery.evidence?.systemRecovery as { state?: unknown } | undefined)?.state === "running")) reasons.push("workspace_recovery_unresolved");
|
||||
const repositories = await db.select().from(executionWorkspaceRepositories).where(and(eq(executionWorkspaceRepositories.companyId, workspace.companyId),
|
||||
eq(executionWorkspaceRepositories.executionWorkspaceId, workspace.id)));
|
||||
if (repositories.some(repository => repository.state !== "ready")) reasons.push("repository_preparation_unresolved");
|
||||
@@ -141,7 +161,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri
|
||||
ref: archivedGitSource?.commit ?? workspace.baseRef, branchName: workspace.branchName } : { kind: "empty" };
|
||||
if (repositories.length) source.repositories = repositories.map(repository => ({ id: repository.id, relativePath: repository.relativePath,
|
||||
repoUrl: repository.repoUrl, ref: repository.requestedRef, ...(repository.pinnedCommit ? { pinnedCommit: repository.pinnedCommit } : {}) }));
|
||||
return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource };
|
||||
return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource, legacyRecoverySources };
|
||||
}
|
||||
|
||||
export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) {
|
||||
@@ -164,7 +184,7 @@ export async function migrateRemoteWorkspace(input: {
|
||||
if (!workspace) throw new Error("remote_workspace_migration_workspace_missing");
|
||||
if (workspace.sourceIssueId) await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, workspace.companyId), eq(issues.id, workspace.sourceIssueId))).for("update");
|
||||
const current = await inspect(tx as unknown as Db, input);
|
||||
if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null };
|
||||
if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] };
|
||||
if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`);
|
||||
const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source);
|
||||
const sourceFingerprint = fingerprint(workspace);
|
||||
@@ -181,12 +201,14 @@ export async function migrateRemoteWorkspace(input: {
|
||||
if (issueBinding !== current.issueBinding) throw new Error("remote_workspace_migration_source_changed");
|
||||
if (Boolean(operation.metadata?.retainedDirectory) !== Boolean(current.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed");
|
||||
if (canonical(operation.metadata?.archivedGitSource ?? null) !== canonical(gitSourceProof(current.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed");
|
||||
return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource };
|
||||
if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(current.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed");
|
||||
return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources };
|
||||
}
|
||||
const [operation] = await tx.insert(workspaceOperations).values({ companyId: workspace.companyId, executionWorkspaceId: workspace.id,
|
||||
issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running",
|
||||
metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding,
|
||||
nestedWorkspaceScope: { version: 1, state: "pending" },
|
||||
...(current.legacyRecoverySources.length ? { legacyRecoverySources: current.legacyRecoverySources } : {}),
|
||||
...(current.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}),
|
||||
...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id,
|
||||
generation: current.retainedDirectory.generation, digest: current.retainedDirectory.restorePoint.digest } } : {}) } }).returning();
|
||||
@@ -194,7 +216,7 @@ export async function migrateRemoteWorkspace(input: {
|
||||
await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspaceMigration: {
|
||||
schema: SCHEMA, operationId: operation!.id, state: "running", sourceFingerprint, source,
|
||||
} }, updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id));
|
||||
return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource };
|
||||
return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources };
|
||||
});
|
||||
if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true };
|
||||
let operation = reserved.operation!;
|
||||
@@ -245,6 +267,8 @@ export async function migrateRemoteWorkspace(input: {
|
||||
input.signal?.throwIfAborted();
|
||||
await lock.execute(sql`select 1`);
|
||||
const publication = await input.db.transaction(async tx => {
|
||||
for (const source of reserved.legacyRecoverySources) await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, source.runId))).for("update");
|
||||
const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update");
|
||||
if (!workspace) throw new Error("remote_workspace_migration_workspace_missing");
|
||||
const existing = readRemoteWorkspaceAuthority(workspace.metadata);
|
||||
@@ -268,9 +292,15 @@ export async function migrateRemoteWorkspace(input: {
|
||||
if (!saved || !["committed", "superseded"].includes(saved.status) || saved.manifestDigest !== child.manifestDigest
|
||||
|| saved.sourceOperationId !== child.operationId) throw new Error("nested_workspace_snapshot_changed");
|
||||
}
|
||||
if (canonical(validation.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed");
|
||||
const authority: RemoteWorkspaceAuthority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: ref.id,
|
||||
source: reserved.source, location: null, writerRunId: null, dirty: false };
|
||||
const metadata: Record<string, unknown> = { ...workspace.metadata, remoteWorkspace: authority }; delete metadata.remoteWorkspaceMigration;
|
||||
if (reserved.legacyRecoverySources.length) {
|
||||
metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id };
|
||||
const prior = Array.isArray(metadata.remoteWorkspaceRecoverySourceRunIds) ? metadata.remoteWorkspaceRecoverySourceRunIds : [];
|
||||
metadata.remoteWorkspaceRecoverySourceRunIds = [...new Set([...prior, ...reserved.legacyRecoverySources.map(source => source.runId)])];
|
||||
}
|
||||
await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`);
|
||||
await tx.update(executionWorkspaceSnapshots).set({ status: "committed", deleteAfter: null,
|
||||
metadata: { ...uploaded.metadata, ...(nested.length ? { importedWorkspaceChildren: nested } : {}) },
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { hasMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js";
|
||||
import { retiredReusableLeasePredecessors } from "./environment-lease-handoff.js";
|
||||
import { and, eq, inArray, ne, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, environments, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
@@ -211,7 +212,10 @@ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRec
|
||||
}
|
||||
const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId));
|
||||
if (!environment) throw new SystemRecoveryFailure("source_unavailable", "service_required");
|
||||
sources.push({ lease, environment, reference: legacyDescriptor, remoteReference: remoteDescriptor,
|
||||
const preservationOnly = legacyDescriptor ? await hasMigratedLegacyWorkspaceSource(db, {
|
||||
companyId: input.companyId, runId: run.id, reference: legacyDescriptor,
|
||||
}) : false;
|
||||
sources.push({ lease, environment, reference: legacyDescriptor, preservationOnly, remoteReference: remoteDescriptor,
|
||||
remoteWorkspace: remoteDescriptor ? remoteWorkspace : null, remoteCwd });
|
||||
}
|
||||
return { run, action, sources };
|
||||
|
||||
@@ -33,6 +33,7 @@ import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspac
|
||||
import { readSystemRecovery } from "./system-recovery-state.js";
|
||||
import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js";
|
||||
import { getStorageService } from "../storage/index.js";
|
||||
import { migrateRemoteWorkspace } from "./remote-workspace-migration.js";
|
||||
import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js";
|
||||
import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js";
|
||||
import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js";
|
||||
@@ -194,6 +195,43 @@ describe("automatic retained workspace recovery", () => {
|
||||
const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
|
||||
const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!;
|
||||
|
||||
it.each([false, true])("preserves an old sandbox after storage migration without writing its old host (removed=%s)", async removeHost => {
|
||||
const f = await seed();
|
||||
// The fixture's command runner supports native copyback; the provider
|
||||
// snapshot boundary separately needs its real-file syncOut capability.
|
||||
targets.get(f.lease.id)!.runner!.syncIn = async operations => {
|
||||
for (const operation of operations) for (const file of operation.files)
|
||||
await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false });
|
||||
return { filesTransferred: 2, bytesTransferred: 0, operations: [] };
|
||||
};
|
||||
targets.get(f.lease.id)!.runner!.syncOut = async operations => {
|
||||
for (const operation of operations) for (const file of operation.files)
|
||||
await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false });
|
||||
return { filesTransferred: 2, bytesTransferred: 0, operations: [] };
|
||||
};
|
||||
await db.insert(executionWorkspaces).values({ id: f.issueId, companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
name: "Existing files", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: f.host });
|
||||
await db.update(issues).set({ executionWorkspaceId: f.issueId }).where(eq(issues.id, f.issueId));
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
try {
|
||||
await migrateRemoteWorkspace({ db, storage: getStorageService(), companyId: f.companyId,
|
||||
workspaceId: f.issueId, apply: true, writersStopped: true });
|
||||
if (removeHost) await rm(f.host, { recursive: true });
|
||||
await requestWorkspaceRecovery({ ...f, idempotencyKey: randomUUID(), actorId: "board" });
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete");
|
||||
expect(await readFile(path.join(f.host, "saved.txt")).catch(() => null)).toBeNull();
|
||||
expect(await run(f.runId)).toMatchObject({ status: "failed", runnerProfileJson: { nativeWorkspaceSync: { state: "prepared" } } });
|
||||
const restored = path.join(root, randomUUID()); await mkdir(restored);
|
||||
const copies = await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: f.issueId,
|
||||
sourceRunIds: [f.runId], cwd: restored });
|
||||
expect(copies).toHaveLength(1);
|
||||
expect(await readFile(path.join(copies[0]!.path, "files", "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
} finally { vi.unstubAllEnvs(); }
|
||||
}, 30_000);
|
||||
|
||||
function boardRecoveryApp() {
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
|
||||
@@ -206,7 +206,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ
|
||||
continue;
|
||||
}
|
||||
}
|
||||
let reference = source.reference;
|
||||
let reference = source.preservationOnly ? null : source.reference;
|
||||
if (reference) {
|
||||
try { await readNativeWorkspaceSyncLocalCwd({ runId: input.runId, companyId: input.companyId,
|
||||
workspaceId: reference.workspaceId, reference }); }
|
||||
|
||||
Reference in new issue
Block a user