mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 21:02:34 +01:00
fix(workspaces): admit continuation after migrated recovery
This commit is contained in:
1 parent
1ab43062b2
commit
76ad68e69f
5 files changed
+130
-10
No files matched your search
@@ -67,16 +67,30 @@ unchanged descriptors and task bindings, verified local snapshot bytes, and a
|
||||
single transaction publishing the migration receipt and remote authority.
|
||||
Failed runs with result-less `observed` or `retryable_failure` coordinators can
|
||||
also be preserved after their leases expire and the offline deployment fence is
|
||||
in force. Their coordinator, checkpoint, and unresolved copyback remain intact.
|
||||
Publication journals the original coordinator binding and installs a run-scoped
|
||||
in force. Their coordinator phase, checkpoint, and unresolved copyback remain
|
||||
intact. Publication journals the original coordinator binding, releases only
|
||||
that exact expired lease with a recovery-history receipt, and installs a run-scoped
|
||||
admission fence: a replacement cannot automatically replay that old turn against
|
||||
the removed local authority. Explicit file recovery and a new task continuation
|
||||
use the retained sources and remote workspace. This is not successful recovery,
|
||||
provider termination evidence, or permission to delete the original allocation.
|
||||
After independently verified sandbox preservation, the task accepts the exact
|
||||
migrated snapshot receipt for continuation without requiring a fresh workspace.
|
||||
Accepted results with unfinished copyback remain blockers. Once the original
|
||||
sync receipt is finalized, assessment and status effects can continue from their
|
||||
existing database coordinator independently of the file storage migration.
|
||||
|
||||
Offline maintenance stops writers; it does not silently change workspace
|
||||
semantics or retire task ownership. `parallel_shared_writers_unsupported` needs
|
||||
an operator-approved transition to serialized remote workspace writes.
|
||||
`task_writer_bound` needs the task lifecycle owner to settle its exact checkout
|
||||
and execution binding, even if the run row is terminal. `task_binding_required`
|
||||
needs provenance recovery or a separately approved durable archival path for an
|
||||
orphaned workspace; the importer never invents a task owner. These records stay
|
||||
on the original volume and block detach. The batch reports its first failed
|
||||
workspace through the phase/error receipt; it does not claim to aggregate every
|
||||
possible blocker in advance or skip blocked rows.
|
||||
|
||||
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
|
||||
|
||||
@@ -202,7 +202,13 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
|
||||
expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "replacement", runIds: [f.runId] })).toEqual([]);
|
||||
expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).toEqual([]);
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before);
|
||||
const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (before!.leaseOwner) {
|
||||
expect(after).toMatchObject({ ...before, leaseOwner: null, leaseExpiresAt: null, updatedAt: expect.any(Date),
|
||||
recoveryHistory: [{ event: "legacy_workspace_migration_owner_released", workspaceId: f.workspaceId,
|
||||
operationId: expect.any(String), previousOwner: { phase, leaseOwner: before!.leaseOwner,
|
||||
leaseExpiresAt: before!.leaseExpiresAt!.toISOString() } }] });
|
||||
} else expect(after).toEqual(before);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toMatchObject({
|
||||
status: "failed", runnerProfileJson: { ...(mode === "no-descriptor" ? {} : { nativeWorkspaceSync: f.reference }), sessionCheckpoint: checkpoint,
|
||||
legacyWorkspaceRecoveryMigration: { workspaceId: f.workspaceId } },
|
||||
@@ -252,6 +258,22 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
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("does not release a coordinator whose expired owner changed during upload", async () => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: "journaled-owner",
|
||||
leaseExpiresAt: new Date(Date.now() - 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const put = storage.putFile.bind(storage);
|
||||
storage.putFile = async input => {
|
||||
const result = await put(input);
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: "different-owner" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
return result;
|
||||
};
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }))
|
||||
.rejects.toThrow("remote_workspace_migration_source_changed");
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
|
||||
.toMatchObject({ phase: "observed", leaseOwner: "different-owner", recoveryHistory: [] });
|
||||
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();
|
||||
@@ -667,6 +689,20 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_operation_unresolved");
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
|
||||
});
|
||||
it.each(["parallel", "orphan", "stale-binding"])("keeps offline enrollment blocked on unresolved workspace semantics (%s)", async mode => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: null }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (mode === "parallel") {
|
||||
await db.update(executionWorkspaces).set({ mode: "shared_workspace" }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
await db.update(issues).set({ executionWorkspaceSettings: { sharedWorkspaceConcurrency: "allow" } }).where(eq(issues.id, f.issueId));
|
||||
} else if (mode === "orphan") await db.update(executionWorkspaces).set({ sourceIssueId: null }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
else await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
|
||||
const reason = mode === "parallel" ? "parallel_shared_writers_unsupported" : mode === "orphan" ? "task_binding_required" : "task_writer_bound";
|
||||
await expect(migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).rejects.toThrow(reason);
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
|
||||
expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
|
||||
});
|
||||
it.each(["running", "starting", "unknown", "failed"])("blocks a %s local service after its control lease expires", async status => {
|
||||
const f = await seed(), serviceId = randomUUID();
|
||||
await db.insert(executionWorkspaceRuntimeLeases).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId,
|
||||
|
||||
@@ -338,12 +338,27 @@ export async function migrateRemoteWorkspace(input: {
|
||||
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 };
|
||||
for (const source of reserved.legacyRecoverySources) await tx.update(heartbeatRuns).set({
|
||||
for (const source of reserved.legacyRecoverySources) {
|
||||
// The offline deployment fence has stopped the old controller. The
|
||||
// locked validation above must still match its journaled expired
|
||||
// lease exactly. Release only that coordinator lease so explicit
|
||||
// file recovery can proceed; its phase/checkpoint and the provider's
|
||||
// independent stop obligation remain unchanged.
|
||||
if (source.unresolvedFinalization?.leaseOwner) await tx.update(nativeRunFinalizations).set({
|
||||
leaseOwner: null, leaseExpiresAt: null,
|
||||
recoveryHistory: sql`${nativeRunFinalizations.recoveryHistory} || ${JSON.stringify([{
|
||||
event: "legacy_workspace_migration_owner_released", operationId: operation.id,
|
||||
workspaceId: workspace.id, previousOwner: source.unresolvedFinalization,
|
||||
}])}::jsonb`, updatedAt: new Date(),
|
||||
}).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId),
|
||||
eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId)));
|
||||
await tx.update(heartbeatRuns).set({
|
||||
runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({
|
||||
legacyWorkspaceRecoveryMigration: { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION,
|
||||
operationId: operation.id, workspaceId: workspace.id },
|
||||
})}::jsonb`, updatedAt: new Date(),
|
||||
}).where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, source.runId)));
|
||||
}).where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, source.runId)));
|
||||
}
|
||||
const prior = Array.isArray(metadata.remoteWorkspaceRecoverySourceRunIds) ? metadata.remoteWorkspaceRecoverySourceRunIds : [];
|
||||
metadata.remoteWorkspaceRecoverySourceRunIds = [...new Set([...prior, ...reserved.legacyRecoverySources.map(source => source.runId)])];
|
||||
}
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { and, desc, eq, sql } from "drizzle-orm";
|
||||
import { executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type ExecutionBlocker } from "@paperclipai/shared";
|
||||
import { z } from "zod";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
import { hasMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js";
|
||||
import { readSnapshot } from "./workspace-recovery-snapshot.js";
|
||||
|
||||
export const workspaceRecoverySchema = z.object({
|
||||
schema: z.literal("paperclip.workspace-recovery.v1"),
|
||||
@@ -43,6 +45,31 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick<typeof he
|
||||
const parsedRemote = remoteWorkspaceRunReferenceSchema.safeParse(run.runnerProfileJson?.remoteWorkspace);
|
||||
const remote = parsedRemote.success && parsedRemote.data.state === "saved" ? parsedRemote.data : null;
|
||||
if ((run.runnerProfileJson?.remoteWorkspace != null && !parsedRemote.success) || reference && parsedRemote.success) return null;
|
||||
if (typeof issueId === "string" && reference?.state === "prepared") {
|
||||
let migrated: boolean;
|
||||
try { migrated = await hasMigratedLegacyWorkspaceSource(db, { companyId: run.companyId, runId: run.id, reference }); }
|
||||
catch (error) { if (error instanceof Error && error.message === "legacy_workspace_migration_receipt_invalid") return null; throw error; }
|
||||
if (migrated) {
|
||||
// Migration retires the local merge destination. A verified immutable
|
||||
// attachment is the repair receipt; the old copyback remains unresolved.
|
||||
// Provider stop and current execution ownership are checked separately.
|
||||
const [saved] = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata,
|
||||
evidence: issueRecoveryActions.evidence }).from(workspaceOperations).innerJoin(issueRecoveryActions, and(
|
||||
eq(issueRecoveryActions.companyId, workspaceOperations.companyId), eq(issueRecoveryActions.sourceIssueId, workspaceOperations.issueId),
|
||||
sql`${issueRecoveryActions.id}::text = ${workspaceOperations.metadata}->>'recoveryActionId'`,
|
||||
)).where(and(eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.issueId, issueId),
|
||||
eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.phase, "workspace_recovery_snapshot"),
|
||||
eq(workspaceOperations.status, "succeeded"),
|
||||
sql`${workspaceOperations.metadata}->>'leaseId' = ${reference.leaseId}`,
|
||||
sql`${workspaceOperations.metadata}->>'providerLeaseId' = ${reference.providerLeaseId}`,
|
||||
sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,runId}' = ${run.id}`,
|
||||
sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' = 'recovered'`,
|
||||
)).orderBy(desc(workspaceOperations.createdAt)).limit(1);
|
||||
if (!saved || readWorkspaceRecovery(saved.evidence)?.status !== "recovered") return null;
|
||||
try { readSnapshot(saved.metadata, run.companyId, run.id); } catch { return null; }
|
||||
return saved.id;
|
||||
}
|
||||
}
|
||||
if (typeof issueId !== "string" || (!remote && (!reference || reference.state !== "finalized" || !reference.finalHostSha256))) return null;
|
||||
const [receipt] = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.issueId, issueId),
|
||||
|
||||
@@ -195,7 +195,7 @@ 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(["retained", "removed", "child"])("preserves an old sandbox after storage migration without writing its old host (%s)", async mode => {
|
||||
it.each(["retained", "removed", "child", "expired-owner"])("preserves an old sandbox after storage migration without writing its old host (%s)", async mode => {
|
||||
vi.stubEnv("DATABASE_URL", connectionString);
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32));
|
||||
const f = await seed();
|
||||
@@ -219,6 +219,9 @@ describe("automatic retained workspace recovery", () => {
|
||||
await db.insert(executionWorkspaces).values({ id: f.issueId, companyId: f.companyId, sourceIssueId,
|
||||
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));
|
||||
if (mode === "expired-owner") await db.update(nativeRunFinalizations).set({ phase: "observed",
|
||||
leaseOwner: "stopped-old-controller", leaseExpiresAt: new Date(Date.now() - 60_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
try {
|
||||
await migrateRemoteWorkspace({ db, storage: getStorageService(), companyId: f.companyId,
|
||||
@@ -229,6 +232,8 @@ describe("automatic retained workspace recovery", () => {
|
||||
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" } } });
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ workspaceRepairRequired: false });
|
||||
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 });
|
||||
@@ -236,12 +241,17 @@ describe("automatic retained workspace recovery", () => {
|
||||
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);
|
||||
if (mode === "child") {
|
||||
if (mode === "child" || mode === "expired-owner") {
|
||||
// The parent must not receive the child's task-scoped recovery copy.
|
||||
expect(await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: sourceIssueId,
|
||||
if (mode === "child") expect(await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: sourceIssueId,
|
||||
sourceRunIds: [f.runId], cwd: restored })).toHaveLength(0);
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "false");
|
||||
const nextRunId = randomUUID(), nextLeaseId = randomUUID(), sourceRun = await run(f.runId);
|
||||
const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: "operator", body: "/new" }).returning();
|
||||
expect(await admitExplicitNativeContinuation({ db, companyId: f.companyId, issueId: f.issueId,
|
||||
agentId: sourceRun.agentId, actorType: "user", actorId: "operator", reason: "issue_commented",
|
||||
commentId: comment.id, successorRunId: nextRunId })).toMatchObject({ previousRunId: f.runId, commentId: comment.id });
|
||||
await db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, agentId: sourceRun.agentId,
|
||||
nativeIssueId: f.issueId, runtimeMode: "native", status: "running" });
|
||||
const storage = { ...getStorageService(), provider: "s3" as const };
|
||||
@@ -259,7 +269,25 @@ describe("automatic retained workspace recovery", () => {
|
||||
expect(await readFile(path.join(prepared.remoteRoot, ".paperclip-recovered", copies[0]!.operationId, "files", "saved.txt"), "utf8"))
|
||||
.toBe("finished work\n");
|
||||
}
|
||||
} finally { vi.unstubAllEnvs(); }
|
||||
if (mode === "retained") {
|
||||
const [receipt] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, copies[0]!.operationId));
|
||||
for (const metadata of [{ ...receipt.metadata, leaseId: randomUUID() }, { ...receipt.metadata, providerLeaseId: randomUUID() },
|
||||
{ ...receipt.metadata, recoveryActionId: randomUUID() }, { ...receipt.metadata, archive: { broken: true } }]) {
|
||||
await db.update(workspaceOperations).set({ metadata }).where(eq(workspaceOperations.id, receipt.id));
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
|
||||
}
|
||||
await db.update(workspaceOperations).set({ metadata: receipt.metadata, status: "running" }).where(eq(workspaceOperations.id, receipt.id));
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
|
||||
await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, receipt.id));
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
|
||||
}
|
||||
} finally {
|
||||
// This journey admits a successor; retire that fixture writer before the
|
||||
// next case exercises the instance-wide offline migration fence.
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(and(
|
||||
eq(heartbeatRuns.companyId, f.companyId), inArray(heartbeatRuns.status, ["queued", "running"])));
|
||||
vi.unstubAllEnvs();
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
function boardRecoveryApp() {
|
||||
|
||||
Reference in new issue
Block a user