mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-06 23:20:53 +01:00
Preserve unresolved coordinators and bind recovery files to the current task
This commit is contained in:
1 parent
420b93c7b7
commit
1ab43062b2
10 files changed
+200
-22
No files matched your search
@@ -65,6 +65,18 @@ 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.
|
||||
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
|
||||
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.
|
||||
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.
|
||||
|
||||
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
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js";
|
||||
import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js";
|
||||
import { settledNativeProcessIdentityCondition } from "../native-local-process-stop.js";
|
||||
import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
|
||||
import { agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
@@ -73,6 +74,7 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
|
||||
scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
|
||||
cursor ? gt(heartbeatRuns.id, cursor) : undefined,
|
||||
kind === "native" ? and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["running", "failed"]),
|
||||
legacyWorkspaceRecoveryNotMigrated(),
|
||||
eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`)
|
||||
: kind === "question" ? and(or(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { readNativeStartupCleanup, readNativeStartupContext } from "../native-runtime/native-startup-cleanup.js";
|
||||
import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js";
|
||||
import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js";
|
||||
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
||||
import { registerLiveRecoveryWork } from "./live-recovery-work.js";
|
||||
@@ -962,6 +963,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
inArray(heartbeatRuns.status, ["running", "failed"]),
|
||||
isNull(nativeRunFinalizations.resultId),
|
||||
eq(nativeRunFinalizations.phase, "retryable_failure"),
|
||||
legacyWorkspaceRecoveryNotMigrated(),
|
||||
or(gt(nativeRunFinalizations.nextAttemptAt, now), eq(nativeRunFinalizations.failureCode, NATIVE_RETAINED_OWNER_WAIT)),
|
||||
),
|
||||
);
|
||||
@@ -2051,6 +2053,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
eq(nativeRunFinalizations.phase, "retryable_failure"),
|
||||
isNull(nativeRunFinalizations.resultId),
|
||||
legacyWorkspaceRecoveryNotMigrated(),
|
||||
sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`,
|
||||
),
|
||||
);
|
||||
|
||||
@@ -1,12 +1,22 @@
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { executionWorkspaces, issues, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { executionWorkspaces, heartbeatRuns, 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 type LegacyWorkspaceRecoverySource = {
|
||||
runId: string; issueId: string; reference: NativeWorkspaceSyncReference | null;
|
||||
unresolvedFinalization?: { phase: string; leaseOwner: string | null; leaseExpiresAt: string | null };
|
||||
};
|
||||
export const LEGACY_WORKSPACE_RECOVERY_MIGRATION = "paperclip.legacy-workspace-recovery-migration.v1";
|
||||
|
||||
/** Offline migration preserves an unresolved coordinator, but it cannot replay
|
||||
* an old execution against a local copyback destination that is no longer its
|
||||
* authority. File recovery and explicit task continuation remain available. */
|
||||
export function legacyWorkspaceRecoveryNotMigrated() {
|
||||
return sql`${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null`;
|
||||
}
|
||||
|
||||
/** 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: {
|
||||
|
||||
@@ -63,6 +63,7 @@ import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold
|
||||
import { readRunCancellation } from "../run-cancellation.js";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
import { retireSupersededNativeRetry } from "./native-superseded-retry.js";
|
||||
import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js";
|
||||
|
||||
function record(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
@@ -115,6 +116,7 @@ export async function claimNativeSessionResumptions(input: {
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
legacyWorkspaceRecoveryNotMigrated(),
|
||||
nativeRunnerOwnershipNotHeldCondition(),
|
||||
isNull(heartbeatRuns.processPid),
|
||||
isNull(heartbeatRuns.processGroupId),
|
||||
@@ -156,7 +158,7 @@ export async function claimNativeSessionResumptions(input: {
|
||||
eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, candidate.runId),
|
||||
eq(heartbeatRuns.nativeIssueId, issue.id),
|
||||
)).for("update").limit(1);
|
||||
if (!run || !coordinator) return false;
|
||||
if (!run || !coordinator || run.runnerProfileJson?.legacyWorkspaceRecoveryMigration != null) return false;
|
||||
if (await retireSupersededNativeRetry(tx as unknown as Db, {
|
||||
run, coordinator, issueExecutionRunId: issue.executionRunId, now,
|
||||
})) return false;
|
||||
|
||||
@@ -39,6 +39,7 @@ import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { retireSupersededNativeRetry } from "./native-superseded-retry.js";
|
||||
import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js";
|
||||
import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js";
|
||||
|
||||
export type NativeControllerIdentity = {
|
||||
@@ -524,6 +525,7 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
legacyWorkspaceRecoveryNotMigrated(),
|
||||
eq(agents.adapterType, "paperclip_runner"),
|
||||
or(inArray(heartbeatRuns.status, ["running", "failed"]), and(
|
||||
eq(heartbeatRuns.status, "interrupted"), eq(nativeRunFinalizations.phase, "observed"),
|
||||
@@ -603,6 +605,8 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
if (!row) {
|
||||
return busy();
|
||||
}
|
||||
if (row.run.runnerProfileJson?.legacyWorkspaceRecoveryMigration != null)
|
||||
return { kind: "blocked", runId: row.run.id, reason: "legacy_workspace_recovery_migrated" } as const;
|
||||
const stopCleanup = row.coordinator.phase === "observed"
|
||||
&& row.coordinator.companyId === row.run.companyId && row.coordinator.issueId === row.run.nativeIssueId
|
||||
&& await readPendingNativeRunStop(tx as unknown as Db, row.run);
|
||||
|
||||
@@ -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, nativeRunFinalizations, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots,
|
||||
import { activityLog, agents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, completionContracts, 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";
|
||||
@@ -26,6 +26,8 @@ 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 { claimNativeSessionResumptions } from "./native-runtime/native-finalization-reconciler.js";
|
||||
import { claimNativeRestartRecoveries } from "./native-runtime/native-restart-recovery.js";
|
||||
import { managedProjectCheckoutName } from "./managed-project-checkout-name.js";
|
||||
import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js";
|
||||
|
||||
@@ -52,7 +54,8 @@ 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(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(nativeRunFinalizations); await db.delete(nativeRunResults); await db.delete(heartbeatRuns); await db.delete(completionContracts);
|
||||
await db.delete(issues); await db.delete(agents); 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() {
|
||||
@@ -143,7 +146,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
});
|
||||
async function failedCopybackFixture() {
|
||||
const f = await seed(), runId = randomUUID();
|
||||
const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Historical agent" }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Historical agent", adapterType: "paperclip_runner" }).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" };
|
||||
@@ -167,11 +170,12 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
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 => {
|
||||
it.each(["controller", "finalizer", "live-finalizer", "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 === "live-finalizer") await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: randomUUID(),
|
||||
leaseExpiresAt: new Date(Date.now() + 60_000) }).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));
|
||||
@@ -180,6 +184,63 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_copyback_unresolved");
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
|
||||
});
|
||||
it.each(["retryable_failure", "observed", "no-descriptor"])("preserves an abandoned %s coordinator without replaying its old agent turn", async mode => {
|
||||
const phase = mode === "no-descriptor" ? "observed" : mode;
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
const [oldRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const checkpoint = { providerSessionId: "original-provider-session", sequence: 17 };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...(mode === "no-descriptor" ? {} : oldRun!.runnerProfileJson),
|
||||
sessionCheckpoint: checkpoint } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(issues).set({ assigneeAgentId: oldRun!.agentId, status: "in_progress" })
|
||||
.where(eq(issues.id, f.issueId));
|
||||
await db.update(nativeRunFinalizations).set({ phase, leaseOwner: phase === "observed" ? "old-controller" : null,
|
||||
leaseExpiresAt: phase === "observed" ? new Date(Date.now() - 60_000) : null,
|
||||
nextAttemptAt: new Date(Date.now() - 30_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true });
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy();
|
||||
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);
|
||||
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 } },
|
||||
});
|
||||
});
|
||||
it.each(["prepared", "finalized"])("keeps accepted-result finalization independent of migrated file storage (%s)", async state => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
const contractId = randomUUID(), resultId = randomUUID();
|
||||
await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, revision: 1,
|
||||
schemaVersion: "v1", policyVersion: "v1", risk: "low", completionAuthority: "agent", incompleteCriteriaPolicy: "block",
|
||||
contractJson: {}, canonicalSha256: "12".repeat(32), createdByActorType: "system", createdByActorId: "test" });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", completionContractId: contractId,
|
||||
runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, state, finalHostSha256: state === "finalized" ? "56".repeat(32) : null } } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, runId: f.runId,
|
||||
completionContractId: contractId, serverFingerprint: "saved-result", schemaStatus: "accepted", resultJson: { saved: true }, canonicalSha256: "34".repeat(32) });
|
||||
await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", resultId }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const migrate = () => migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true });
|
||||
if (state === "prepared") await expect(migrate()).rejects.toThrow("remote_copyback_unresolved");
|
||||
else {
|
||||
await migrate();
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy();
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!.runnerProfileJson)
|
||||
.not.toHaveProperty("legacyWorkspaceRecoveryMigration");
|
||||
}
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before);
|
||||
});
|
||||
it("migrates historical records without borrowing another workspace's finalizer", async () => {
|
||||
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
const otherId = randomUUID(), cwd = path.join(root, "longer-historical-folder"); await fs.mkdir(cwd);
|
||||
await fs.writeFile(path.join(cwd, "history.txt"), "prior task files\n");
|
||||
await db.insert(executionWorkspaces).values({ id: otherId, companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
name: "Historical workspace", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd });
|
||||
await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).toBe(2);
|
||||
for (const id of [f.workspaceId, otherId]) expect(readRemoteWorkspaceAuthority((await workspace(id)).metadata)?.snapshotId).toBeTruthy();
|
||||
});
|
||||
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);
|
||||
|
||||
@@ -109,20 +109,56 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri
|
||||
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 })
|
||||
const [owner] = await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner,
|
||||
leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, resultId: nativeRunFinalizations.resultId })
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.runId, run.id)));
|
||||
const unsettled = owner && ["observed", "retryable_failure"].includes(owner.phase) && owner.resultId === null;
|
||||
const owned = owner?.leaseOwner && (!owner.leaseExpiresAt || owner.leaseExpiresAt > new Date());
|
||||
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 });
|
||||
|| !owner || owned || !(["committed", "terminal_failure"].includes(owner.phase) || unsettled)) reasons.push("remote_copyback_unresolved");
|
||||
else legacyRecoverySources.push({ runId: run.id, issueId: run.issueId, reference,
|
||||
...(unsettled ? { unresolvedFinalization: { phase: owner.phase, leaseOwner: owner.leaseOwner,
|
||||
leaseExpiresAt: owner.leaseExpiresAt?.toISOString() ?? null } } : {}) });
|
||||
}
|
||||
const [finalization] = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
|
||||
const finalizations = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase,
|
||||
leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt,
|
||||
resultId: nativeRunFinalizations.resultId, run: heartbeatRuns }).from(nativeRunFinalizations)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.id, nativeRunFinalizations.runId))).where(and(
|
||||
eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.issueId, workspace.sourceIssueId),
|
||||
sql`${nativeRunFinalizations.phase} not in ('committed', 'terminal_failure')`,
|
||||
)).limit(1) : [];
|
||||
if (finalization) reasons.push("execution_finalization_unresolved");
|
||||
)) : [];
|
||||
for (const row of finalizations) {
|
||||
if (legacyRecoverySources.some(source => source.runId === row.id && source.unresolvedFinalization)) continue;
|
||||
const stopped = input.writersStopped && row.run.runtimeMode === "native"
|
||||
&& ["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(row.run.status)
|
||||
&& !(row.run.controllerLeaseExpiresAt && row.run.controllerLeaseExpiresAt > new Date())
|
||||
&& !(row.leaseOwner && (!row.leaseExpiresAt || row.leaseExpiresAt > new Date()));
|
||||
const reference = readNativeWorkspaceSyncReference(row.run.runnerProfileJson?.nativeWorkspaceSync);
|
||||
// Finalized copyback has no outstanding local writer. Assessment/status
|
||||
// effects stay with their existing durable coordinator after the move.
|
||||
if (stopped && reference?.state === "finalized") continue;
|
||||
if (stopped && reference && reference.workspaceId !== workspace.id) {
|
||||
const [other] = await db.select({ providerType: executionWorkspaces.providerType, cwd: executionWorkspaces.cwd,
|
||||
sourceIssueId: executionWorkspaces.sourceIssueId, metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, workspace.companyId), eq(executionWorkspaces.id, reference.workspaceId)));
|
||||
// The offline batch must migrate every local record before the app can
|
||||
// start. Another workspace's coordinator is checked with its own source.
|
||||
if (other && ((other.cwd && ["local_fs", "git_worktree"].includes(other.providerType)) || readRemoteWorkspaceAuthority(other.metadata))
|
||||
&& await isLegacyWorkspaceTaskRelated(db, { companyId: workspace.companyId, issueId: row.run.nativeIssueId, sourceIssueId: other.sourceIssueId })) continue;
|
||||
}
|
||||
if (stopped && row.run.runnerProfileJson?.nativeWorkspaceSync == null && row.run.runnerProfileJson?.remoteWorkspace == null
|
||||
&& row.resultId === null && ["observed", "retryable_failure"].includes(row.phase)) {
|
||||
legacyRecoverySources.push({ runId: row.id, issueId: row.run.nativeIssueId!, reference: null,
|
||||
unresolvedFinalization: { phase: row.phase, leaseOwner: row.leaseOwner,
|
||||
leaseExpiresAt: row.leaseExpiresAt?.toISOString() ?? null } });
|
||||
continue;
|
||||
}
|
||||
reasons.push("execution_finalization_unresolved");
|
||||
}
|
||||
legacyRecoverySources.sort((a, b) => a.runId.localeCompare(b.runId));
|
||||
const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id),
|
||||
allowedOperation ? ne(workspaceOperations.id, allowedOperation) : undefined,
|
||||
@@ -267,8 +303,12 @@ 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");
|
||||
for (const source of reserved.legacyRecoverySources) {
|
||||
await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations)
|
||||
.where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, source.runId))).for("update");
|
||||
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);
|
||||
@@ -298,6 +338,12 @@ 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({
|
||||
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)));
|
||||
const prior = Array.isArray(metadata.remoteWorkspaceRecoverySourceRunIds) ? metadata.remoteWorkspaceRecoverySourceRunIds : [];
|
||||
metadata.remoteWorkspaceRecoverySourceRunIds = [...new Set([...prior, ...reserved.legacyRecoverySources.map(source => source.runId)])];
|
||||
}
|
||||
|
||||
@@ -372,11 +372,19 @@ export async function prepareRemoteWorkspace(input: {
|
||||
}
|
||||
const recoveredFiles: Array<{ operationId: string; sourceRunId: string; path: string }> = [];
|
||||
const recoverySourceRunIds = input.workspace.metadata?.remoteWorkspaceRecoverySourceRunIds;
|
||||
if (!sameAllocation && Array.isArray(recoverySourceRunIds) && recoverySourceRunIds.length && input.workspace.sourceIssueId) {
|
||||
if (!sameAllocation && Array.isArray(recoverySourceRunIds) && recoverySourceRunIds.length) {
|
||||
// Shared and inherited workspaces retain their original source task. The
|
||||
// current writer's task owns its recovery attachments, including a child's
|
||||
// preserved files recorded during migration of an ancestor's workspace.
|
||||
const [run] = await input.db.select({ nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId,
|
||||
contextIssueId: sql<unknown>`${heartbeatRuns.contextSnapshot}->'issueId'` }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, input.workspace.companyId), eq(heartbeatRuns.id, input.runId)));
|
||||
const issueId = run?.nativeIssueId ?? run?.issueId ?? run?.contextIssueId;
|
||||
if (typeof issueId !== "string") throw new Error("remote_workspace_recovery_task_missing");
|
||||
const scratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-recovery-import-"));
|
||||
try {
|
||||
const recovered = await attachPreservedWorkspaceSnapshots({ db: input.db, companyId: input.workspace.companyId,
|
||||
issueId: input.workspace.sourceIssueId, sourceRunIds: recoverySourceRunIds as string[], cwd: scratch });
|
||||
issueId, sourceRunIds: recoverySourceRunIds as string[], cwd: scratch });
|
||||
for (const item of recovered) {
|
||||
const relative = `.paperclip-recovered/${item.operationId}`;
|
||||
await target.runner!.syncIn!([{ operationId: randomUUID(), files: [{
|
||||
|
||||
@@ -195,7 +195,9 @@ 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 => {
|
||||
it.each(["retained", "removed", "child"])("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();
|
||||
// The fixture's command runner supports native copyback; the provider
|
||||
// snapshot boundary separately needs its real-file syncOut capability.
|
||||
@@ -209,14 +211,19 @@ describe("automatic retained workspace recovery", () => {
|
||||
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,
|
||||
const sourceIssueId = mode === "child" ? randomUUID() : f.issueId;
|
||||
if (mode === "child") {
|
||||
await db.insert(issues).values({ id: sourceIssueId, companyId: f.companyId, title: "Ancestor workspace owner" });
|
||||
await db.update(issues).set({ parentId: sourceIssueId }).where(eq(issues.id, f.issueId));
|
||||
}
|
||||
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));
|
||||
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 });
|
||||
if (mode !== "retained") 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");
|
||||
@@ -229,6 +236,29 @@ 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") {
|
||||
// The parent must not receive the child's task-scoped recovery copy.
|
||||
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);
|
||||
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 };
|
||||
const reserved = await reserveRemoteWorkspace({ db, storage, companyId: f.companyId, issueId: f.issueId,
|
||||
runId: nextRunId, agentId: sourceRun.agentId, environmentId: f.lease.environmentId!,
|
||||
projectId: null, projectWorkspaceId: null, existingWorkspaceId: f.issueId, source: { kind: "empty" } });
|
||||
const [lease] = await db.insert(environmentLeases).values({ id: nextLeaseId, companyId: f.companyId,
|
||||
issueId: f.issueId, heartbeatRunId: nextRunId, environmentId: f.lease.environmentId,
|
||||
provider: "daytona", providerLeaseId: randomUUID(), status: "active" }).returning();
|
||||
const nextRoot = path.join(root, nextRunId); await mkdir(nextRoot);
|
||||
const target = { ...targets.get(f.lease.id)!, leaseId: nextLeaseId, remoteCwd: await realpath(nextRoot) };
|
||||
const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved,
|
||||
runId: nextRunId, lease: lease as EnvironmentLease, target });
|
||||
expect(prepared.workspace.sourceIssueId).toBe(sourceIssueId);
|
||||
expect(await readFile(path.join(prepared.remoteRoot, ".paperclip-recovered", copies[0]!.operationId, "files", "saved.txt"), "utf8"))
|
||||
.toBe("finished work\n");
|
||||
}
|
||||
} finally { vi.unstubAllEnvs(); }
|
||||
}, 30_000);
|
||||
|
||||
|
||||
Reference in new issue
Block a user