fix(workspaces): preserve legacy recovery across storage migration (#18841)

## Thinking Path

> - Paperclip manages durable tasks and agent work through the control
plane.
> - An ephemeral web server needs existing workspace files in shared
storage.
> - Failed historical runs can still own recoverable sandbox files or
unfinished coordinator state.
> - Moving local files must preserve those obligations without replaying
the old agent turn.
> - Migration journals the exact source and fences old copyback; the
recovery owner saves verified sandbox files separately.
> - New user input can continue only after matching preservation
receipts and independent execution-stop checks.
> - Accepted results retain their finalization owner and must have its
durable workspace receipt before storage moves.

## Linked Issues or Issue Description

Builds on merged #18826 and #18835.

**What happened?**

Existing workspaces with failed native copyback could not migrate to
remote storage. Simply changing their storage authority would let later
recovery target a retired local directory, lose child-task attachments,
or leave a recovered task unable to accept a new turn.

**Expected behavior**

Preserve local files, exact sandbox provenance, task bindings,
checkpoints, and unresolved history. Recover surviving sandbox files
into immutable task-scoped attachments after migration, and let a new
authorized user turn use those files without replaying the old
execution.

**Steps to reproduce**

Create a terminal native run with a prepared copyback descriptor and a
retained sandbox. Migrate its local workspace, remove the old host
filesystem, recover the sandbox, and submit a new user turn. Also
exercise a child task reusing its ancestor's workspace and an expired
observed coordinator.

## What Changed

- Journal exact legacy run, task, descriptor, and unfinished coordinator
bindings during offline migration. Descriptor-free runs retain the
complete declared lease/allocation/directory identities; publication
locks and revalidates them.
- Publish a run-scoped fence that prevents automatic restart or copyback
into the retired local authority.
- Release only the exact journaled expired coordinator lease under the
offline deployment fence and locked revalidation, recording the prior
owner in recovery history. Preserve phase, checkpoint, result, and run
status.
- Save recovered sandbox files through the existing immutable snapshot
owner, retaining allocation and provider-stop checks.
- Attach recovered files using the current task's identity when a child
reuses an ancestor workspace.
- Accept matching migration and snapshot receipts for workspace-repair
admission and explicit new-user continuation. Changed coordinator
bindings and invalid receipts remain blocked.
- Require the finalizer's successful workspace_finalize operation before
moving a workspace whose accepted result has finalized copyback; the
existing owner can then finish assessment/status effects.
- Document unresolved workspace-policy, task-ownership, and
missing-provenance cases as separate owner actions that block
detachment.

## Verification

- 569 tests across migration, retained recovery, and explicit
continuation passed on ffcde13f57.
- The final saved-result follow-up passes all 75 migration tests. Four
receipt cases cover prepared copyback, a successful finalization
receipt, a missing receipt, and a failed receipt. The successful case
invokes the actual finalizer after migration and completes the existing
saved operation.
- Five real-file journeys cover a retained host, removed host,
child-task reuse, an expired observed coordinator, and a descriptor-free
run with two separate sandbox allocations. They recover sandbox files,
clear the repair blocker, admit an explicit /new user comment, reserve a
new remote workspace, and verify recovered attachment contents.
- Negative cases reject changed coordinator phase/owner, invalid
migration fences, changed source sets and allocations, mismatched
task/allocation/action receipts, malformed archives, and unfinished
snapshot operations. Descriptor-free continuation requires every
journaled source; one missing archive keeps the hold.
- Twelve PostgreSQL/S3 and maintenance-service tests pass on the exact
final rebuilt runtime, including a plain-Node bundled process,
authenticated original-key export, overlapping task-folder restore, and
complete predecessor filesystem removal. The final server and runtime
bundle builds also pass.
- Head 9ab6a38fa4 passed all 573
migration/recovery/continuation tests, server typecheck, server build,
runtime bundle, and 12 PostgreSQL/S3/bundled-maintenance tests.
- Verification caught an expired-owner continuation rejection (160
recovery cases passed, one failed); the admission fix and subsequent
569-test run passed. Earlier fixture failures and the parent PR's
unrelated full-suite failure remain documented in their verification
history.
- Greptile and Paperclip Review both scored final head
fa58d4a5ce 5/5, with no remaining
findings.
- CI exposed 15 blocker-read projection regressions: descriptor-free
checks performed an extra full run read for ordinary unmigrated tasks.
The follow-up gates the lookup on the migration marker, includes that
marker in the bounded blocker projection, and selects only recovery
fields. All 31 blocker-projection and real-file journey cases plus
server typecheck pass. Final-head CI passed all 52 successful checks,
including repository typecheck, build, server tests, and browser suites.
- Preserved CI failure: the unchanged Runner package had one Cursor
Cloud Stop/checkpoint timing failure (4,903 cases passed). All four
focused local cases passed. Retrying only the failed job and its
verification gate passed without source changes.
- This PR does not claim tenant enrollment or a hosted rolling-deploy
result.

## Risks

Cloud must stop every predecessor deployment before enabling the offline
migration contract. Migration preserves uncertain work; it does not
certify provider termination, invent a completed result, or authorize
sandbox deletion. Missing or ambiguous sources remain held. Preserved
attachments are separate files rather than an automatic merge into newer
task edits. Parallel shared-write policy, stale task ownership, missing
task provenance, and accepted results without a successful finalization
receipt remain enrollment blockers. Original folders and the original
volume remain retained.

## Model Used

OpenAI GPT-6 through Codex, with reasoning, repository tools, and code
execution. This session does not expose a more specific serving model ID
or context-window size.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used with available version and
capability details
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have described the issue in-PR using the bug template
- [x] I have not referenced internal or instance-local issues or links
- [x] My branch name describes the change and contains no internal task
ID
- [x] I have run the focused tests locally and they pass
- [x] I have added tests for the changed behavior
- [x] I have updated relevant documentation
- [x] I have considered and documented risks
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address reviewer findings before merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-11-03 16:17:13 -06:00
1 parent d597918284
commit 0ab3ce7a03
17 files changed
+715 -31

No files matched your search

@@ -45,3 +45,60 @@ 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.
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 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.
For older runs without a synchronization descriptor, the migration journals
all declared recovery lease/allocation/directory identities under row locks.
Continuation requires the unchanged complete set and a verified snapshot for
every source. Missing or ambiguous source identity remains held; a partial
archive cannot certify recovery. The recovery journey covers two allocations,
removal of the old filesystem, and the files appearing in the next remote turn.
Accepted results with unfinished copyback remain blockers. Once the original
sync reference is finalized and its successful `workspace_finalize` operation
receipt exists, 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
recovery source validation and company boundaries.
+1 -1
View File
@@ -17,7 +17,7 @@ import { hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspac
const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution",
"interactionKind", "interactionStatus", "mutation", "interactionId"] as const;
const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace"] as const;
const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "legacyWorkspaceRecoveryMigration"] as const;
const blockerContextColumn = sql<Record<string, unknown> | null>`case
when ${heartbeatRuns.contextSnapshot} is null then null else (
select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb)
+1 -1
View File
@@ -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) {
@@ -6,6 +6,8 @@ import { readTaskRetryIntent } from "./task-retry.js";
import { readSystemRecovery } from "./system-recovery-state.js";
import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js";
import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js";
import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js";
import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js";
import { createHash } from "node:crypto";
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
@@ -380,8 +382,19 @@ export async function admitExplicitNativeContinuation(input: {
// cleanup failed. That immutable result is history, not an active commit.
// Keep it intact and require the same controller/process/lease stop proofs
// before admitting new user input; never apply or replay the old result.
const legacyReference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync);
const migratedSource = coordinator && !coordinator.resultId && !coordinator.leaseOwner
&& ["observed", "retryable_failure"].includes(coordinator.phase)
&& await hasVerifiedWorkspaceRepair(db, run)
? await readMigratedLegacyWorkspaceSource(db, { companyId, runId: run.id, reference: legacyReference }) : null;
// Offline migration fenced this exact unfinished turn and preserved its
// coordinator. A verified attachment permits new user input without
// inventing a terminal result or replaying the old session. All stop and
// ownership checks below still apply.
const preservedFailure = migratedSource?.issueId === issueId
&& migratedSource.unresolvedFinalization?.phase === coordinator?.phase;
if (!cancelledStartup && coordinator && (
(coordinator.phase !== "terminal_failure" && !committedFailure) || coordinator.leaseOwner ||
(coordinator.phase !== "terminal_failure" && !committedFailure && !preservedFailure) || coordinator.leaseOwner ||
coordinator.failureDetail?.successorRunId)) return blocked("controller_settling",
run.status === "cancelled" && !coordinator.leaseOwner
? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details."
@@ -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`,
),
);
@@ -0,0 +1,109 @@
import { isDeepStrictEqual } from "node:util";
import { and, eq, sql } from "drizzle-orm";
import { environmentLeases, 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 | null;
recoveryAllocations?: { leaseId: string; providerLeaseId: string; remoteCwd: string }[] | 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: {
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;
}
/** Record every declared source, or none: partial preservation cannot release
* the task's recovery hold. The recovery owner separately verifies provenance,
* provider termination and archive bytes before writing its receipts. */
export async function readLegacyWorkspaceRecoveryAllocations(db: Pick<Db, "select">, run: {
id: string; companyId: string; nativeIssueId: string | null; resultJson: Record<string, unknown> | null;
}): Promise<LegacyWorkspaceRecoverySource["recoveryAllocations"]> {
const recovery = run.resultJson?.workspaceRestoreRecovery as { schema?: unknown; leaseIds?: unknown } | undefined;
const ids = recovery?.leaseIds;
if (recovery?.schema !== "paperclip.workspace-restore-recovery.v1" || !Array.isArray(ids) || !ids.length
|| ids.some(id => typeof id !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id))
|| new Set(ids).size !== ids.length || !run.nativeIssueId) return null;
const allocations = [];
for (const id of [...ids].sort()) {
const [lease] = await db.select({ providerLeaseId: environmentLeases.providerLeaseId, metadata: environmentLeases.metadata })
.from(environmentLeases).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, id),
eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.issueId, run.nativeIssueId)));
if (!lease?.providerLeaseId || typeof lease.metadata?.remoteCwd !== "string" || !lease.metadata.remoteCwd.startsWith("/")) return null;
allocations.push({ leaseId: id as string, providerLeaseId: lease.providerLeaseId, remoteCwd: lease.metadata.remoteCwd });
}
return allocations;
}
/** 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 readMigratedLegacyWorkspaceSource(db: Pick<Db, "select">, input: {
companyId: string; runId: string; reference: NativeWorkspaceSyncReference | null;
}): Promise<LegacyWorkspaceRecoverySource | null> {
if (input.reference && input.reference.state !== "prepared") return null;
const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
nativeIssueId: heartbeatRuns.nativeIssueId,
resultJson: sql<Record<string, unknown>>`jsonb_build_object('workspaceRestoreRecovery', ${heartbeatRuns.resultJson}->'workspaceRestoreRecovery')`,
runnerProfileJson: sql<Record<string, unknown>>`jsonb_build_object(
'legacyWorkspaceRecoveryMigration', ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration',
'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync',
'remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace')`,
}).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId),
));
const runMarker = run?.runnerProfileJson?.legacyWorkspaceRecoveryMigration as { workspaceId?: unknown } | undefined;
const workspaceId = input.reference?.workspaceId ?? runMarker?.workspaceId;
if (typeof workspaceId !== "string") return null;
const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and(
eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, workspaceId),
));
if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return null;
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");
if (!isDeepStrictEqual(runMarker, {
schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: marker.operationId, workspaceId,
})) 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, workspaceId),
eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"),
eq(workspaceOperations.status, "succeeded"),
));
const sources = operation?.metadata?.legacyRecoverySources;
const source = Array.isArray(sources) ? sources.find(source => source && typeof source === "object"
&& source.runId === input.runId && typeof source.issueId === "string" && isDeepStrictEqual(source.reference, input.reference)) : null;
if (source && source.issueId !== run?.nativeIssueId) throw new Error("legacy_workspace_migration_receipt_invalid");
if (source && !input.reference && (run?.runnerProfileJson?.nativeWorkspaceSync != null
|| run?.runnerProfileJson?.remoteWorkspace != null
|| !isDeepStrictEqual(source.recoveryAllocations ?? null, run ? await readLegacyWorkspaceRecoveryAllocations(db, run) : null)))
throw new Error("legacy_workspace_migration_receipt_invalid");
if (!source) throw new Error("legacy_workspace_migration_receipt_invalid");
return source as LegacyWorkspaceRecoverySource;
}
export async function hasMigratedLegacyWorkspaceSource(...args: Parameters<typeof readMigratedLegacyWorkspaceSource>) {
return Boolean(await readMigratedLegacyWorkspaceSource(...args));
}
@@ -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);
@@ -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, environmentLeases, 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";
@@ -25,6 +25,10 @@ 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 { claimNativeSessionResumptions } from "./native-runtime/native-finalization-reconciler.js";
import { claimNativeRestartRecoveries } from "./native-runtime/native-restart-recovery.js";
import { attemptNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalizer.js";
import { managedProjectCheckoutName } from "./managed-project-checkout-name.js";
import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js";
@@ -51,7 +55,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(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() {
@@ -140,6 +145,182 @@ 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", 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" };
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", "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 === "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));
}
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.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([]);
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 } },
});
});
it.each(["prepared", "finalized", "missing-receipt", "failed-receipt"])("keeps accepted-result finalization independent of migrated file storage (%s)", async mode => {
const state = mode === "prepared" ? "prepared" : "finalized";
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 operationId = randomUUID();
if (mode === "finalized" || mode === "failed-receipt") await db.insert(workspaceOperations).values({ id: operationId,
companyId: f.companyId, executionWorkspaceId: f.workspaceId, heartbeatRunId: f.runId, issueId: f.issueId,
phase: "workspace_finalize", status: mode === "finalized" ? "succeeded" : "failed" });
const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
const migrate = () => migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true });
if (mode !== "finalized") await expect(migrate()).rejects.toThrow(state === "prepared"
? "remote_copyback_unresolved" : "execution_finalization_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 attemptNativeWorkspaceFinalization({ db, runId: f.runId }))
.toMatchObject({ kind: "completed", operation: { id: operationId } });
}
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);
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("rejects a descriptor-free allocation changed during upload", async () => {
const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
await db.insert(environmentLeases).values({ id: f.reference.leaseId, companyId: f.companyId,
issueId: f.issueId, heartbeatRunId: f.runId, provider: "daytona", providerLeaseId: f.reference.providerLeaseId,
status: "released", metadata: { remoteCwd: f.reference.remoteCwd } });
await db.update(heartbeatRuns).set({ runnerProfileJson: {}, resultJson: {
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.reference.leaseId] },
} }).where(eq(heartbeatRuns.id, f.runId));
await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId));
const put = storage.putFile.bind(storage);
storage.putFile = async input => {
const result = await put(input);
await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.reference.leaseId));
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("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();
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 });
@@ -536,6 +717,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,
+121 -17
View File
@@ -4,7 +4,7 @@ import os from "node:os";
import path from "node:path";
import { and, eq, inArray, ne, or, sql } from "drizzle-orm";
import {
executionWorkspaceRepositories, executionWorkspaceRuntimeLeases, executionWorkspaceSnapshots, executionWorkspaces,
environmentLeases, executionWorkspaceRepositories, executionWorkspaceRuntimeLeases, executionWorkspaceSnapshots, executionWorkspaces,
heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, workspaceOperations, workspaceRuntimeServices, type Db,
} from "@paperclipai/db";
import { readRemoteWorkspaceAuthority, remoteWorkspaceSourceSchema, type RemoteWorkspaceAuthority, type RemoteWorkspaceSource } from "@paperclipai/shared";
@@ -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, readLegacyWorkspaceRecoveryAllocations, 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,16 +99,76 @@ 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 [finalization] = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
));
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,
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 || 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 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);
// The finalizer retries from its operation receipt, not the sync descriptor.
// Without that receipt it would try exporting the newly migrated workspace,
// which intentionally has no live location. Keep the local authority until
// the existing owner records completion; then assessment/status can resume.
if (stopped && reference?.state === "finalized") {
const [finalized] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.heartbeatRunId, row.id),
eq(workspaceOperations.issueId, row.run.nativeIssueId!), eq(workspaceOperations.phase, "workspace_finalize"),
eq(workspaceOperations.status, "succeeded"),
)).limit(1);
if (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,
recoveryAllocations: await readLegacyWorkspaceRecoveryAllocations(db, row.run),
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,
@@ -129,11 +192,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 +207,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 +230,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 +247,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 +262,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 +313,15 @@ 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: 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");
for (const allocation of source.recoveryAllocations ?? []) await tx.select({ id: environmentLeases.id })
.from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
eq(environmentLeases.id, allocation.leaseId))).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 +345,36 @@ 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 };
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)));
}
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 } : {}) },
+10 -2
View File
@@ -394,11 +394,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: [{
@@ -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 { readMigratedLegacyWorkspaceSource } 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,39 @@ 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" || !reference && !parsedRemote.success
&& run.runnerProfileJson?.legacyWorkspaceRecoveryMigration != null)) {
let migrated;
try { migrated = await readMigratedLegacyWorkspaceSource(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) {
const allocations = reference ? [{ leaseId: reference.leaseId, providerLeaseId: reference.providerLeaseId }]
: migrated.recoveryAllocations;
if (!allocations?.length) return null;
let receiptId: string | null = null;
for (const allocation of allocations) {
// 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' = ${allocation.leaseId}`,
sql`${workspaceOperations.metadata}->>'providerLeaseId' = ${allocation.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; }
receiptId = saved.id;
}
return receiptId;
}
}
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),
@@ -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 };
+146 -1
View File
@@ -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";
@@ -183,7 +184,7 @@ describe("automatic retained workspace recovery", () => {
ownerType: "board", cause: mode === "native" ? "native_continuation_requires_reconciliation" : "legacy_execution_requires_reconciliation",
fingerprint: runId, nextAction: "Manual repair required", evidence: { runId, workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery } });
const runtime = { resumeRunLease: vi.fn(async (_input: { lease: EnvironmentLease }) => ({ providerLeaseId, metadata: { remoteCwd: remote } })),
retryPendingSandboxTeardown: vi.fn(async () => ({ providerLeaseId, state: "stopped" })) };
retryPendingSandboxTeardown: vi.fn(async (_input: { lease: EnvironmentLease }) => ({ providerLeaseId, state: "stopped" })) };
const environmentRuntime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox",
acquireRunLease: vi.fn(), releaseRunLease: vi.fn(), ...runtime,
resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: [], declared: null, narrowing: null }),
@@ -194,6 +195,150 @@ 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", "expired-owner", "no-descriptor"])("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("native", mode !== "no-descriptor");
// 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: [] };
};
if (mode === "no-descriptor") {
const secondId = randomUUID(), secondProviderId = randomUUID(), secondRemote = path.join(root, secondId);
await cp(f.remote, secondRemote, { recursive: true });
const second: typeof f.lease = { ...f.lease, id: secondId, providerLeaseId: secondProviderId,
metadata: { ...f.lease.metadata, remoteCwd: secondRemote } };
second.metadata = { ...second.metadata,
remoteExecutionTermination: remoteTerminationReceipt(second, { providerLeaseId: secondProviderId, state: "stopped" }) };
await db.insert(environmentLeases).values(second);
targets.set(secondId, { ...targets.get(f.lease.id)!, leaseId: secondId, remoteCwd: secondRemote });
f.runtime.retryPendingSandboxTeardown.mockImplementation(async ({ lease }) => ({
providerLeaseId: lease.providerLeaseId!, state: "stopped",
}));
f.runtime.resumeRunLease.mockImplementation(async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId!,
metadata: { remoteCwd: lease.metadata!.remoteCwd as string } }));
const source = await run(f.runId);
await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson,
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id, secondId] },
} }).where(eq(heartbeatRuns.id, f.runId));
}
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));
if (mode === "expired-owner" || mode === "no-descriptor") 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,
workspaceId: f.issueId, apply: true, writersStopped: 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");
expect(await readFile(path.join(f.host, "saved.txt")).catch(() => null)).toBeNull();
expect(await run(f.runId)).toMatchObject({ status: "failed", runnerProfileJson: mode === "no-descriptor" ? {} : { 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 });
expect(copies).toHaveLength(mode === "no-descriptor" ? 2 : 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);
if (mode === "no-descriptor") {
const source = await run(f.runId);
await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson,
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] },
} }).where(eq(heartbeatRuns.id, f.runId));
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
await db.update(heartbeatRuns).set({ resultJson: source.resultJson }).where(eq(heartbeatRuns.id, f.runId));
for (const copy of copies) {
await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, copy.operationId));
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, copy.operationId));
}
const [allocation] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.lease.id));
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false);
await db.update(environmentLeases).set({ providerLeaseId: allocation.providerLeaseId }).where(eq(environmentLeases.id, f.lease.id));
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
}
if (mode === "child" || mode === "expired-owner" || mode === "no-descriptor") {
// The parent must not receive the child's task-scoped recovery copy.
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();
const continuationInput = { db, companyId: f.companyId, issueId: f.issueId,
agentId: sourceRun.agentId, actorType: "user", actorId: "operator", reason: "issue_commented",
commentId: comment.id, successorRunId: nextRunId };
if (mode === "expired-owner" || mode === "no-descriptor") {
for (const changed of [{ phase: "retryable_failure", leaseOwner: null }, { phase: "observed", leaseOwner: "new-controller" }]) {
await db.update(nativeRunFinalizations).set(changed).where(eq(nativeRunFinalizations.runId, f.runId));
expect(await admitExplicitNativeContinuation({ ...continuationInput, dryRun: true })).toBeNull();
}
await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId));
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...sourceRun.runnerProfileJson,
legacyWorkspaceRecoveryMigration: { invalid: true } } }).where(eq(heartbeatRuns.id, f.runId));
expect(await admitExplicitNativeContinuation({ ...continuationInput, dryRun: true })).toBeNull();
await db.update(heartbeatRuns).set({ runnerProfileJson: sourceRun.runnerProfileJson }).where(eq(heartbeatRuns.id, f.runId));
}
expect(await admitExplicitNativeContinuation(continuationInput)).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 };
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");
}
if (mode === "retained" || mode === "no-descriptor") {
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() {
const app = express();
app.use(express.json());
+1 -1
View File
@@ -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 }); }