mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix(workspaces): preserve finalized recovery holds through offline import (#19180)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - Offline workspace import must save files before local volume
removal.
> - A task recovery hold can remain after native file finalization
succeeds.
> - The old importer rejects that hold even when its file source is
exact and no controller owns the run.
> - Clearing the hold would discard the task's recovery obligation.
> - This pull request saves files and carries each exact recovery source
forward without resolving the hold.
> - Accepted-result finalizers keep their existing retry path.
## Linked Issues or Issue Description
Refs #19144, #18943, #19038. This PR builds on merged #19144.
**What happened?**
Offline import reports `workspace_recovery_unresolved` for terminal
native runs with finalized workspace references and active task recovery
holds. Some holds also name an exact historical file allocation.
**Expected behavior**
Save the workspace durably. Keep task warnings, recovery rows, results,
coordinator phases and task pointers unchanged. Retain exact future file
recovery and accepted-result finalization retries.
**Steps to reproduce**
1. Retain a local task workspace and an active recovery hold for a
finalized native run.
2. Stop all application writers through offline maintenance.
3. Attempt workspace import.
4. Observe that the old importer rejects the hold.
## What Changed
- Add an owner module that checks the exact terminal native run,
finalized reference, released ownership and task relationship.
- Require the normal finalizer's successful company/run/task receipt for
an accepted-result retry. Support historical receipts without a
workspace foreign key.
- Validate recorded file allocations with the existing recovery-source
owner.
- Journal and recheck hold evidence. Lock recovery, coordinator, run and
allocation rows before publication.
- Give finalized file sources a separate migration marker. Keep legacy
execution-replay suppression out of accepted-result retry paths.
- Use the owning parser for older finalized references that omit
optional fields.
- Redirect future file recovery to immutable snapshot preservation. Keep
provider-stop and source checks intact.
- Verify native copyback or snapshot receipts for each exact allocation.
Require proof for every source, including completed mixed repairs.
- Retain existing successful copyback receipts. Keep a historical
folder's obligation on its real source workspace.
- Retain the finalized migration marker through the execution blocker
projection. Discover finalized child-task sources from their exact
ancestor workspace reference.
- Document the receipt and lock contracts.
## Verification
- Fresh local run at head 6ebe535751: 208
tests pass across the new preservation suite, migration suite,
publication concurrency suite and legacy restore recovery suite.
- `node ../node_modules/vitest/vitest.mjs run --config vitest.config.ts
src/services/finalized-workspace-recovery-migration.test.ts
src/services/remote-workspace-migration-concurrency.test.ts
src/services/remote-workspace-migration.test.ts
src/__tests__/legacy-workspace-restore-recovery.test.ts` from `server/`.
- Comment-only cleanup `a858d203c68a6c86fd39e30fb63dc56c1a303429`
removes added prose comments under CONTRIBUTING.md. TypeScript 6.0.3
output with `removeComments: true` is byte-identical before and after
for all six changed TypeScript files. Runtime behavior and the 208-test
evidence are unchanged; no new tests were added for this cleanup.
- The two file-restoration cases fail against the original importer with
`workspace_recovery_unresolved`.
- Positive tests delete the old folder and restore the saved archive.
They compare unchanged task, recovery, result and coordinator rows.
- Negative controls cover unknown kinds, mismatched sources, malformed
file markers, active ownership, active recovery and changed evidence
during upload.
- Red/green tests exercise getExecutionBlocker and the ancestor import
path. An accepted result without a successful finalize receipt remains
blocked.
- The mixed-receipt regression fails before the review fix and passes
afterward. A missing secondary receipt still blocks repair.
- Existing and future repair receipts remain readable. Accepted-result
finalization still returns its original successful receipt.
- Local server typechecking passed before the master replay. The
post-replay local check found a stale linked Runner declaration in
unchanged vendor code. Clean current-head CI Typecheck + Release
Registry and Build both pass.
- Greptile gives current head `a858d203c68a6c86fd39e30fb63dc56c1a303429`
5/5. All actionable findings are fixed and resolved.
- The prior CI run had one workspace shard cancelled after its spot
runner received a shutdown signal. No test assertion failed in that job.
That incomplete run was superseded by the equivalent comment-only
cleanup. The fresh full CI run passed on the current head; the original
infrastructure cancellation remains recorded here.
- [Current-head full
CI](https://github.com/paperclipai/paperclip/actions/runs/41970385102)
passes: 54 success, one neutral and two skipped statuses, with no failed
or pending checks. Typecheck, tests and build pass.
- Ponytail's source review rates the behavior head 5/5. It did not
execute tests; the local and CI runs above supply execution evidence.
- No provider, serving migration or copied-data deployment acceptance is
claimed.
## Risks
- This path depends on the existing offline deployment fence. It does
not prove that providers stopped.
- Unknown or unverified source boundaries remain blocked.
- Migration receipts must remain exact. Missing or changed finalized
file-source receipts fail closed.
- Root maintainers own merge, deployment and copied-data acceptance
#19144 is already merged.
## Model Used
OpenAI GPT-6.1 Codex, exact model ID `gpt-6.1-sol`. Used for reasoning,
code editing, shell tools and test execution. The runtime does not
expose its 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 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 either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
---------
Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
48b57e0c94
commit
dbbde744d2
7 files changed
+448
-35
No files matched your search
@@ -836,6 +836,25 @@ with `NOWAIT`. A real `55P03` aborts the publication transaction and reports
|
||||
archive and local files remain available for a later administrative retry. Other
|
||||
database errors propagate. No automatic retry or lock-strength reduction occurs.
|
||||
|
||||
Offline import can preserve an active task recovery hold after its exact native
|
||||
workspace reference is finalized. It checks the terminal run, released controller
|
||||
and coordinator ownership, task relationship, and any recorded file allocations.
|
||||
An accepted result with a retrying coordinator also needs the successful
|
||||
company/run/task `workspace_finalize` receipt consumed by the normal finalizer.
|
||||
Historical receipts can have no workspace foreign key. Import retains recovery
|
||||
rows, task pointers, results and coordinator phases. It does not establish provider
|
||||
termination or resolve the hold. It locks and rechecks the saved evidence before
|
||||
publication. An unrelated historical folder keeps the obligation on its real
|
||||
source workspace.
|
||||
|
||||
Finalized file sources receive a separate `finalizedWorkspaceRecoveryMigration`
|
||||
run marker and an exact migration-operation receipt. They do not receive the
|
||||
legacy execution-replay suppression marker. Future recovery exports an immutable
|
||||
snapshot of the recorded allocation instead of merging into the retired local
|
||||
folder. The existing source, provider-stop and archive checks still apply.
|
||||
Completed copyback recovery receipts remain valid. Unknown or unverified recovery
|
||||
sources continue to block import.
|
||||
|
||||
The workspace fingerprint pins the recorded source. An arriving reservation can
|
||||
reach a legacy workspace before refusing it, so publication must follow this
|
||||
order too. The lifecycle advisory lock alone does not fence runtime operations.
|
||||
|
||||
@@ -18,7 +18,7 @@ import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-run
|
||||
|
||||
const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution",
|
||||
"interactionKind", "interactionStatus", "mutation", "interactionId"] as const;
|
||||
const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "legacyWorkspaceRecoveryMigration"] as const;
|
||||
const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "legacyWorkspaceRecoveryMigration", "finalizedWorkspaceRecoveryMigration"] 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)
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, closeRegisteredClients, companies, completionContracts, createDb, environmentLeases, environments,
|
||||
executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { createStorageService } from "../storage/service.js";
|
||||
import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js";
|
||||
import { inspectWorkspaceRecoverySource } from "./workspace-recovery-source.js";
|
||||
import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js";
|
||||
import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js";
|
||||
import { createSystemRecovery } from "./system-recovery-state.js";
|
||||
import { attemptNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalizer.js";
|
||||
import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js";
|
||||
import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("finalized workspace recovery preservation", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>, db: ReturnType<typeof createDb>, root: string;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("finalized-recovery-migration-"); db = createDb(database.connectionString);
|
||||
root = await fs.mkdtemp(path.join(os.tmpdir(), "finalized-recovery-migration-"));
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
|
||||
}, 90_000);
|
||||
afterAll(async () => {
|
||||
if (database) await closeRegisteredClients(database.connectionString);
|
||||
await database?.cleanup(); if (root) await fs.rm(root, { recursive: true, force: true }); vi.unstubAllEnvs();
|
||||
});
|
||||
async function fixture(files = false) {
|
||||
const companyId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(), cwd = path.join(root, workspaceId);
|
||||
await fs.mkdir(cwd); await fs.writeFile(path.join(cwd, "saved.txt"), "original held files\n");
|
||||
await db.insert(companies).values({ id: companyId, name: "Finalized recovery", issuePrefix: `F${companyId.slice(0, 6)}` });
|
||||
const [agent] = await db.insert(agents).values({ companyId, name: "Historical worker", adapterType: "paperclip_runner" }).returning();
|
||||
const [issue] = await db.insert(issues).values({ companyId, title: "Held task", status: "blocked" }).returning();
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issue.id,
|
||||
mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd, name: "Held files" });
|
||||
await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issue.id));
|
||||
const reference = { schema: "paperclip.native-workspace-sync/v3" as const, state: "finalized" as const, workspaceId,
|
||||
leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/workspaces/task", descriptorSha256: "12".repeat(32),
|
||||
baselineSha256: "34".repeat(32), finalHostSha256: "56".repeat(32), resourceDisposition: "destroy" as const };
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId: agent.id, nativeIssueId: issue.id,
|
||||
status: "failed", runtimeMode: "native", runnerProfileJson: { nativeWorkspaceSync: reference }, resultJson: { retained: "result" } });
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId, phase: "terminal_failure" });
|
||||
if (files) {
|
||||
const [environment] = await db.insert(environments).values({ name: `Historical environment ${workspaceId}`, driver: "daytona" }).returning();
|
||||
await db.insert(environmentLeases).values({ id: reference.leaseId, companyId, environmentId: environment.id,
|
||||
issueId: issue.id, heartbeatRunId: runId, executionWorkspaceId: workspaceId, provider: "daytona",
|
||||
providerLeaseId: reference.providerLeaseId, leasePolicy: "ephemeral", status: "expired", metadata: { remoteCwd: reference.remoteCwd } });
|
||||
}
|
||||
const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issue.id,
|
||||
kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(),
|
||||
nextAction: "Preserve task hold", evidence: { runId, nativeFailureBlock: { retained: "hold" }, ...(files ? {
|
||||
systemRecovery: { ...createSystemRecovery(runId), state: "exhausted" },
|
||||
workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId, status: "blocked", stage: "verifying_stop", attempt: 4, requestIds: [] },
|
||||
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId, leaseIds: [reference.leaseId] },
|
||||
} : {}) } }).returning();
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(path.join(root, `${workspaceId}-objects`)));
|
||||
return { companyId, workspaceId, runId, issueId: issue.id, cwd, reference, action, storage };
|
||||
}
|
||||
async function preserved(f: Awaited<ReturnType<typeof fixture>>) {
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
return { issue, run, owner, action };
|
||||
}
|
||||
it.each([false, true])("saves held files while preserving task/coordinator/result bytes (file source: %s)", async files => {
|
||||
const f = await fixture(files), before = await preserved(f);
|
||||
await expect(migrateRemoteWorkspace({ db, ...f, writersStopped: false, apply: true })).rejects.toThrow("workspace_recovery_unresolved");
|
||||
await migrateRemoteWorkspace({ db, ...f, writersStopped: true, apply: true });
|
||||
const after = await preserved(f);
|
||||
expect(after.issue).toEqual(before.issue); expect(after.owner).toEqual(before.owner); expect(after.action).toEqual(before.action);
|
||||
if (files) {
|
||||
expect(after.run).toEqual({ ...before.run, updatedAt: expect.any(Date), runnerProfileJson: { ...before.run.runnerProfileJson,
|
||||
finalizedWorkspaceRecoveryMigration: { workspaceId: f.workspaceId, operationId: expect.any(String), schema: "paperclip.legacy-workspace-recovery-migration.v1" } } });
|
||||
const inspected = await inspectWorkspaceRecoverySource(db, { ...f, actionId: f.action.id });
|
||||
expect(inspected.sources).toEqual([expect.objectContaining({ preservationOnly: true, reference: f.reference })]);
|
||||
expect(inspected.sources[0].lease.metadata).toEqual({ remoteCwd: f.reference.remoteCwd });
|
||||
} else expect(after.run).toEqual(before.run);
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata)!;
|
||||
const ref = (await getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }))!;
|
||||
await fs.rm(f.cwd, { recursive: true });
|
||||
const archivePath = path.join(root, `${f.workspaceId}.pcws`), targetDir = path.join(root, `${f.workspaceId}-restored`);
|
||||
await loadWorkspaceSnapshotArchive({ storage: f.storage, reference: ref, targetPath: archivePath });
|
||||
await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest });
|
||||
expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("original held files\n");
|
||||
});
|
||||
it.each([true, false])("requires a successful receipt for accepted-result retries (receipt: %s)", async saved => {
|
||||
const f = await fixture(), contractId = randomUUID(), resultId = randomUUID(), operationId = 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 }).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));
|
||||
if (saved) await db.insert(workspaceOperations).values({ id: operationId, companyId: f.companyId, executionWorkspaceId: null,
|
||||
issueId: f.issueId, heartbeatRunId: f.runId, phase: "workspace_finalize", status: "succeeded" });
|
||||
const before = await preserved(f);
|
||||
if (!saved) {
|
||||
await expect(migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true })).rejects.toThrow("workspace_recovery_unresolved");
|
||||
expect(await preserved(f)).toEqual(before);
|
||||
return;
|
||||
}
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
expect(await preserved(f)).toEqual(before);
|
||||
expect(await attemptNativeWorkspaceFinalization({ db, runId: f.runId })).toMatchObject({ kind: "completed", operation: { id: operationId } });
|
||||
});
|
||||
it.each(["source-run", "controller", "owner", "phase", "descriptor", "file-marker", "allocation", "running-recovery", "unknown-kind"])("keeps an unproven %s hold blocked", async mode => {
|
||||
const f = await fixture(true);
|
||||
if (mode === "source-run") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, runId: randomUUID() } }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
if (mode === "controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (mode === "owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "expired-owner", leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (mode === "phase") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (mode === "descriptor") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, finalHostSha256: null } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (mode === "file-marker") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, historicalWorkspaceSource: { schema: "unverified" } } }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
if (mode === "allocation") await db.update(environmentLeases).set({ heartbeatRunId: null }).where(eq(environmentLeases.id, f.reference.leaseId));
|
||||
if (mode === "running-recovery") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence,
|
||||
systemRecovery: { ...createSystemRecovery(f.runId), state: "running" } } }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
if (mode === "unknown-kind") await db.update(issueRecoveryActions).set({ kind: "unknown_kind" }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
expect(await inventoryRemoteWorkspaceMigration(db, { ...f, writersStopped: true })).toMatchObject({ eligible: false, reasons: expect.arrayContaining(["workspace_recovery_unresolved"]) });
|
||||
});
|
||||
it.each(["action", "allocation", "coordinator"])("rechecks the exact %s after upload", async mode => {
|
||||
const f = await fixture(true), put = f.storage.putFile.bind(f.storage);
|
||||
f.storage.putFile = async input => { const result = await put(input);
|
||||
if (mode === "action") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, retainedChange: true } }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
if (mode === "allocation") await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.reference.leaseId));
|
||||
if (mode === "coordinator") await db.update(nativeRunFinalizations).set({ failureCode: "changed" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
return result;
|
||||
};
|
||||
await expect(migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true })).rejects.toThrow();
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
expect(readRemoteWorkspaceAuthority(workspace.metadata)).toBeNull();
|
||||
});
|
||||
it("keeps a historical folder separate from its real recovery source", async () => {
|
||||
const f = await fixture(true), workspaceId = randomUUID(), cwd = path.join(root, workspaceId); await fs.mkdir(cwd);
|
||||
await fs.writeFile(path.join(cwd, "history.txt"), "historical files\n");
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd, name: "Other historical folder" });
|
||||
await migrateRemoteWorkspace({ db, ...f, workspaceId, apply: true, writersStopped: true });
|
||||
const [historical] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId));
|
||||
expect(historical.metadata).not.toHaveProperty("legacyWorkspaceRecoveryMigration");
|
||||
expect((await preserved(f)).run.runnerProfileJson).not.toHaveProperty("finalizedWorkspaceRecoveryMigration");
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
expect(await readMigratedLegacyWorkspaceSource(db, { ...f, reference: f.reference })).toMatchObject({ reference: f.reference });
|
||||
});
|
||||
it("discovers a finalized child task's held source when importing its ancestor workspace", async () => {
|
||||
const f = await fixture(true);
|
||||
const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Workspace owner" }).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));
|
||||
const before = await preserved(f);
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
expect((await preserved(f)).issue).toEqual(before.issue);
|
||||
expect((await preserved(f)).action).toEqual(before.action);
|
||||
const inspected = await inspectWorkspaceRecoverySource(db, { ...f, actionId: f.action.id });
|
||||
expect(inspected.sources).toEqual([expect.objectContaining({ preservationOnly: true, reference: f.reference })]);
|
||||
expect((await preserved(f)).run.runnerProfileJson).toHaveProperty("finalizedWorkspaceRecoveryMigration.workspaceId", f.workspaceId);
|
||||
});
|
||||
it.each(["old-copyback", "new-snapshot"])("retains verified repair through migration and the execution blocker (%s)", async mode => {
|
||||
const f = await fixture(true), operationId = randomUUID();
|
||||
await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, workspaceRestoreFailure: "restore_unsafe_archive",
|
||||
systemRecovery: { ...createSystemRecovery(f.runId), state: "complete" },
|
||||
workspaceRecovery: { ...(f.action.evidence.workspaceRecovery as object), status: "recovered" },
|
||||
} }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
const file = (suffix: string) => ({ objectKey: `${f.companyId}/workspace-recovery/${f.runId}/${f.reference.leaseId}/${"78".repeat(32)}.${suffix}`,
|
||||
sha256: "78".repeat(32), byteSize: 1 });
|
||||
await db.insert(workspaceOperations).values({ id: operationId, companyId: f.companyId, issueId: f.issueId,
|
||||
heartbeatRunId: f.runId, phase: mode === "old-copyback" ? "workspace_recovery" : "workspace_recovery_snapshot", status: "succeeded",
|
||||
metadata: { schema: "paperclip.workspace-recovery-snapshot.v1", recoveryActionId: f.action.id,
|
||||
leaseId: f.reference.leaseId, providerLeaseId: f.reference.providerLeaseId,
|
||||
descriptorSha256: f.reference.descriptorSha256, finalHostSha256: f.reference.finalHostSha256,
|
||||
archive: file("tar"), manifest: file("jsonl") } });
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
expect(await verifiedWorkspaceRepairReceipt(db, (await preserved(f)).run)).toBe(operationId);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({
|
||||
recoveryActionId: f.action.id, workspaceRepairRequired: false,
|
||||
});
|
||||
});
|
||||
it("requires every allocation and preserves mixed completed repair receipts", async () => {
|
||||
const f = await fixture(true), otherLeaseId = randomUUID(), otherProviderLeaseId = randomUUID();
|
||||
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.reference.leaseId));
|
||||
await db.insert(environmentLeases).values({ ...lease, id: otherLeaseId, providerLeaseId: otherProviderLeaseId });
|
||||
await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence,
|
||||
systemRecovery: { ...createSystemRecovery(f.runId), state: "complete" },
|
||||
workspaceRecovery: { ...(f.action.evidence.workspaceRecovery as object), status: "recovered" },
|
||||
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: f.runId,
|
||||
leaseIds: [f.reference.leaseId, otherLeaseId] },
|
||||
} }).where(eq(issueRecoveryActions.id, f.action.id));
|
||||
await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
heartbeatRunId: f.runId, phase: "workspace_recovery", status: "succeeded", metadata: {
|
||||
recoveryActionId: f.action.id, leaseId: f.reference.leaseId, providerLeaseId: f.reference.providerLeaseId,
|
||||
descriptorSha256: f.reference.descriptorSha256, finalHostSha256: f.reference.finalHostSha256 } });
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
expect(await verifiedWorkspaceRepairReceipt(db, (await preserved(f)).run)).toBeNull();
|
||||
const file = (suffix: string) => ({ objectKey: `${f.companyId}/workspace-recovery/${f.runId}/${otherLeaseId}/${"78".repeat(32)}.${suffix}`,
|
||||
sha256: "78".repeat(32), byteSize: 1 });
|
||||
await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
heartbeatRunId: f.runId, phase: "workspace_recovery_snapshot", status: "succeeded", metadata: {
|
||||
schema: "paperclip.workspace-recovery-snapshot.v1", recoveryActionId: f.action.id,
|
||||
leaseId: otherLeaseId, providerLeaseId: otherProviderLeaseId,
|
||||
archive: file("tar"), manifest: file("jsonl") } });
|
||||
expect(await verifiedWorkspaceRepairReceipt(db, (await preserved(f)).run)).not.toBeNull();
|
||||
});
|
||||
it("preserves an older finalized reference with omitted optional fields", async () => {
|
||||
const f = await fixture(true);
|
||||
const { resourceDisposition: _disposition, ...legacy } = f.reference;
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: {
|
||||
...legacy, schema: "paperclip.native-workspace-sync/v1",
|
||||
} } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
const inspected = await inspectWorkspaceRecoverySource(db, { ...f, actionId: f.action.id });
|
||||
expect(inspected.sources).toEqual([expect.objectContaining({ preservationOnly: true,
|
||||
reference: { ...legacy, schema: "paperclip.native-workspace-sync/v1", resourceDisposition: null } })]);
|
||||
});
|
||||
it("requires the exact receipt and leaves ordinary finalized references alone", async () => {
|
||||
const f = await fixture(true);
|
||||
expect(await readMigratedLegacyWorkspaceSource(db, { ...f, reference: f.reference })).toBeNull();
|
||||
await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true });
|
||||
await expect(readMigratedLegacyWorkspaceSource(db, { ...f, reference: { ...f.reference, descriptorSha256: "78".repeat(32) } }))
|
||||
.rejects.toThrow("legacy_workspace_migration_receipt_invalid");
|
||||
await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.reference.leaseId));
|
||||
await expect(readMigratedLegacyWorkspaceSource(db, { ...f, reference: f.reference })).rejects.toThrow("legacy_workspace_migration_receipt_invalid");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,93 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { executionWorkspaces, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, nativeRunResults,
|
||||
workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { adapterExecutionControls } from "./adapter-execution-control.js";
|
||||
import { isLegacyWorkspaceTaskRelated, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
import { readSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js";
|
||||
import { readWorkspaceRecovery } from "./workspace-recovery-receipt.js";
|
||||
import { inspectWorkspaceRecoverySource, terminalWorkspaceRunStates } from "./workspace-recovery-source.js";
|
||||
|
||||
type ReadDb = Pick<Db, "select">;
|
||||
type Recovery = Pick<typeof issueRecoveryActions.$inferSelect,
|
||||
"id" | "sourceIssueId" | "kind" | "cause" | "evidence" | "status" | "updatedAt">;
|
||||
export type FinalizedWorkspaceRecoveryHold = {
|
||||
actionId: string; runId: string; sourceWorkspaceId: string; fingerprint: string;
|
||||
source: LegacyWorkspaceRecoverySource | null;
|
||||
};
|
||||
|
||||
function canonical(value: unknown): string {
|
||||
if (value instanceof Date) return JSON.stringify(value.toISOString());
|
||||
if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`;
|
||||
if (value && typeof value === "object") return `{${Object.entries(value).sort(([a], [b]) => a.localeCompare(b))
|
||||
.map(([key, item]) => `${JSON.stringify(key)}:${canonical(item)}`).join(",")}}`;
|
||||
return JSON.stringify(value) ?? "null";
|
||||
}
|
||||
|
||||
export async function inspectFinalizedWorkspaceRecoveryHold(db: ReadDb, input: {
|
||||
companyId: string; workspaceId: string; recovery: Recovery;
|
||||
}): Promise<FinalizedWorkspaceRecoveryHold | null> {
|
||||
const action = input.recovery;
|
||||
if (action.kind !== "active_run_watchdog" || typeof action.evidence.runId !== "string"
|
||||
|| !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(action.evidence.runId)) return null;
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, action.evidence.runId)));
|
||||
const reference = readNativeWorkspaceSyncReference(run?.runnerProfileJson?.nativeWorkspaceSync);
|
||||
if (!run || run.runtimeMode !== "native" || run.nativeIssueId !== action.sourceIssueId
|
||||
|| !terminalWorkspaceRunStates.includes(run.status) || !reference || reference.state !== "finalized"
|
||||
|| !reference.finalHostSha256 || run.runnerProfileJson?.remoteWorkspace != null
|
||||
|| adapterExecutionControls.has(run.id) || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return null;
|
||||
const [owner] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, run.id),
|
||||
eq(nativeRunFinalizations.issueId, action.sourceIssueId)));
|
||||
if (!owner || owner.leaseOwner) return null;
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, reference.workspaceId)));
|
||||
if (!workspace || !await isLegacyWorkspaceTaskRelated(db, { companyId: input.companyId,
|
||||
issueId: run.nativeIssueId, sourceIssueId: workspace.sourceIssueId })) return null;
|
||||
if (reference.workspaceId !== input.workspaceId && !readRemoteWorkspaceAuthority(workspace.metadata)
|
||||
&& (!workspace.cwd || !["local_fs", "git_worktree"].includes(workspace.providerType))) return null;
|
||||
let finalizeReceipt: string | null = null;
|
||||
if (!["committed", "terminal_failure"].includes(owner.phase)) {
|
||||
if (owner.phase !== "retryable_failure" || !owner.resultId || !run.completionContractId) return null;
|
||||
const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
|
||||
eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.issueId, action.sourceIssueId),
|
||||
eq(nativeRunResults.runId, run.id), eq(nativeRunResults.id, owner.resultId),
|
||||
eq(nativeRunResults.completionContractId, run.completionContractId), eq(nativeRunResults.schemaStatus, "accepted")));
|
||||
const [receipt] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.issueId, action.sourceIssueId),
|
||||
eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.phase, "workspace_finalize"),
|
||||
eq(workspaceOperations.status, "succeeded")));
|
||||
if (!result || !receipt) return null;
|
||||
finalizeReceipt = receipt.id;
|
||||
}
|
||||
const fileKeys = ["systemRecovery", "workspaceRecovery", "workspaceRestoreRecovery", "historicalWorkspaceSource"];
|
||||
let source: LegacyWorkspaceRecoverySource | null = null;
|
||||
if (fileKeys.some(key => action.evidence[key] != null)) {
|
||||
const system = readSystemRecovery(action.evidence), recovery = readWorkspaceRecovery(action.evidence);
|
||||
if (!system || system.sourceRunId !== run.id || system.state === "running" || !recovery || recovery.runId !== run.id
|
||||
|| ["queued", "recovering"].includes(recovery.status)
|
||||
|| action.evidence.workspaceRestoreRecovery != null
|
||||
&& !isDeepStrictEqual(action.evidence.workspaceRestoreRecovery, run.resultJson?.workspaceRestoreRecovery)) return null;
|
||||
try {
|
||||
const inspected = await inspectWorkspaceRecoverySource(db as Db, {
|
||||
companyId: input.companyId, issueId: action.sourceIssueId, runId: run.id, actionId: action.id,
|
||||
});
|
||||
source = { runId: run.id, issueId: action.sourceIssueId, reference,
|
||||
recoveryAllocations: inspected.sources.map(item => ({ leaseId: item.lease.id,
|
||||
providerLeaseId: item.lease.providerLeaseId!, remoteCwd: item.remoteCwd })).sort((a, b) => a.leaseId.localeCompare(b.leaseId)) };
|
||||
} catch (error) {
|
||||
if (error instanceof SystemRecoveryFailure) return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return { actionId: action.id, runId: run.id, sourceWorkspaceId: reference.workspaceId,
|
||||
source: reference.workspaceId === input.workspaceId ? source : null,
|
||||
fingerprint: createHash("sha256").update(canonical({ action, reference,
|
||||
status: run.status, controllerLeaseExpiresAt: run.controllerLeaseExpiresAt?.toISOString() ?? null,
|
||||
completionContractId: run.completionContractId, restoreRecovery: run.resultJson?.workspaceRestoreRecovery,
|
||||
owner, finalizeReceipt, source })).digest("hex") };
|
||||
}
|
||||
@@ -2,7 +2,7 @@ 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";
|
||||
import { readNativeWorkspaceSyncReference, type NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
|
||||
export type LegacyWorkspaceRecoverySource = {
|
||||
runId: string; issueId: string; reference: NativeWorkspaceSyncReference | null;
|
||||
@@ -63,24 +63,31 @@ export async function readLegacyWorkspaceRecoveryAllocations(db: Pick<Db, "selec
|
||||
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',
|
||||
'finalizedWorkspaceRecoveryMigration', ${heartbeatRuns.runnerProfileJson}->'finalizedWorkspaceRecoveryMigration',
|
||||
'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 runMarker = run?.runnerProfileJson?.[input.reference?.state === "finalized"
|
||||
? "finalizedWorkspaceRecoveryMigration" : "legacyWorkspaceRecoveryMigration"] as { workspaceId?: unknown } | undefined;
|
||||
if (input.reference?.state === "finalized" && runMarker == null) return null;
|
||||
if (input.reference?.state === "finalized" && !isDeepStrictEqual(input.reference, readNativeWorkspaceSyncReference(run?.runnerProfileJson?.nativeWorkspaceSync)))
|
||||
throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
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;
|
||||
if (!readRemoteWorkspaceAuthority(workspace?.metadata)) {
|
||||
if (input.reference?.state === "finalized" && runMarker != null) throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
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");
|
||||
@@ -100,6 +107,16 @@ export async function readMigratedLegacyWorkspaceSource(db: Pick<Db, "select">,
|
||||
|| run?.runnerProfileJson?.remoteWorkspace != null
|
||||
|| !isDeepStrictEqual(source.recoveryAllocations ?? null, run ? await readLegacyWorkspaceRecoveryAllocations(db, run) : null)))
|
||||
throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
if (source?.recoveryAllocations) {
|
||||
for (const allocation of source.recoveryAllocations) {
|
||||
const [lease] = await db.select({ providerLeaseId: environmentLeases.providerLeaseId, metadata: environmentLeases.metadata })
|
||||
.from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
|
||||
eq(environmentLeases.id, allocation.leaseId), eq(environmentLeases.heartbeatRunId, input.runId),
|
||||
eq(environmentLeases.issueId, source.issueId)));
|
||||
if (!lease || lease.providerLeaseId !== allocation.providerLeaseId || lease.metadata?.remoteCwd !== allocation.remoteCwd)
|
||||
throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
}
|
||||
}
|
||||
if (!source) throw new Error("legacy_workspace_migration_receipt_invalid");
|
||||
return source as LegacyWorkspaceRecoverySource;
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceSourceSchema, type RemoteWorkspaceAuthority, type RemoteWorkspaceSource } from "@paperclipai/shared";
|
||||
import { offlineMigrationExecutionWhere } from "./offline-migration-execution.js";
|
||||
import { inspectFinalizedWorkspaceRecoveryHold, type FinalizedWorkspaceRecoveryHold } from "./finalized-workspace-recovery-migration.js";
|
||||
import { workspaceExportCauses } from "./workspace-recovery-source.js";
|
||||
import type { StorageService } from "../storage/types.js";
|
||||
import { persistActivity, publishActivity } from "./activity-log.js";
|
||||
@@ -59,7 +60,7 @@ 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, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] };
|
||||
if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[] };
|
||||
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");
|
||||
@@ -110,15 +111,15 @@ 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, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode,
|
||||
const referencedRuns = 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'`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' in ('prepared', 'finalized')`,
|
||||
));
|
||||
const legacyRecoverySources: LegacyWorkspaceRecoverySource[] = [];
|
||||
for (const run of unexported.sort((a, b) => a.id.localeCompare(b.id))) {
|
||||
for (const run of referencedRuns.filter(run => object(run.reference).state === "prepared").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 })
|
||||
@@ -203,23 +204,36 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri
|
||||
) : undefined),
|
||||
)).limit(1);
|
||||
if (service) reasons.push("workspace_runtime_service_unsettled");
|
||||
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, kind: issueRecoveryActions.kind,
|
||||
const recoveryIssueIds = [...new Set([workspace.sourceIssueId, ...referencedRuns.map(run => run.issueId),
|
||||
...legacyRecoverySources.map(source => source.issueId)].filter((id): id is string => Boolean(id)))];
|
||||
const recoveries = recoveryIssueIds.length ? await db.select({ id: issueRecoveryActions.id, sourceIssueId: issueRecoveryActions.sourceIssueId,
|
||||
status: issueRecoveryActions.status, updatedAt: issueRecoveryActions.updatedAt, kind: issueRecoveryActions.kind,
|
||||
cause: issueRecoveryActions.cause, evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, workspace.companyId), inArray(issueRecoveryActions.sourceIssueId, recoveryIssueIds),
|
||||
inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
)) : [];
|
||||
if (recoveries.some(recovery => {
|
||||
if (!input.writersStopped || object(recovery.evidence?.systemRecovery).state === "running") return true;
|
||||
if (legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId)) return false;
|
||||
const finalizedRecoveryHolds: FinalizedWorkspaceRecoveryHold[] = [];
|
||||
for (const recovery of recoveries.sort((a, b) => a.id.localeCompare(b.id))) {
|
||||
if (!input.writersStopped || object(recovery.evidence?.systemRecovery).state === "running") {
|
||||
reasons.push("workspace_recovery_unresolved"); continue;
|
||||
}
|
||||
if (legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId && source.reference?.state !== "finalized")) continue;
|
||||
const held = await inspectFinalizedWorkspaceRecoveryHold(db, { companyId: workspace.companyId,
|
||||
workspaceId: workspace.id, recovery });
|
||||
if (held) {
|
||||
finalizedRecoveryHolds.push(held);
|
||||
if (held.source && !legacyRecoverySources.some(source => source.runId === held.source!.runId)) legacyRecoverySources.push(held.source);
|
||||
continue;
|
||||
}
|
||||
// These warnings govern task disposition, not file authority. Keep them
|
||||
// unchanged while the offline owner saves the workspace. Unknown recovery
|
||||
// kinds and any declared file source still require explicit preservation.
|
||||
return !["stranded_assigned_issue", "configuration_validation", "missing_disposition", "deliberate_wait_without_target"].includes(recovery.kind)
|
||||
if (!["stranded_assigned_issue", "configuration_validation", "missing_disposition", "deliberate_wait_without_target"].includes(recovery.kind)
|
||||
|| workspaceExportCauses.includes(recovery.cause)
|
||||
|| ["workspaceRestoreRecovery", "workspaceRecovery", "historicalWorkspaceSource", "systemRecovery", "nativeFailureBlock"]
|
||||
.some(key => recovery.evidence?.[key] != null);
|
||||
})) reasons.push("workspace_recovery_unresolved");
|
||||
.some(key => recovery.evidence?.[key] != null)) reasons.push("workspace_recovery_unresolved");
|
||||
}
|
||||
legacyRecoverySources.sort((a, b) => a.runId.localeCompare(b.runId));
|
||||
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");
|
||||
@@ -227,7 +241,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, legacyRecoverySources };
|
||||
return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource, legacyRecoverySources, finalizedRecoveryHolds };
|
||||
}
|
||||
|
||||
export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) {
|
||||
@@ -254,7 +268,7 @@ export async function migrateRemoteWorkspace(input: {
|
||||
if (!workspace) throw new Error("remote_workspace_migration_workspace_missing");
|
||||
if (workspace.sourceIssueId !== selected.sourceIssueId) throw new Error("remote_workspace_migration_source_changed");
|
||||
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, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] };
|
||||
if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[] };
|
||||
if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`);
|
||||
const nativeSessionContinuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace);
|
||||
const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source);
|
||||
@@ -273,14 +287,16 @@ 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");
|
||||
if (canonical(operation.metadata?.finalizedRecoveryHolds ?? []) !== canonical(current.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed");
|
||||
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 };
|
||||
return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds };
|
||||
}
|
||||
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,
|
||||
...(nativeSessionContinuity ? { nativeSessionContinuity } : {}),
|
||||
nestedWorkspaceScope: { version: 1, state: "pending" },
|
||||
...(current.finalizedRecoveryHolds.length ? { finalizedRecoveryHolds: current.finalizedRecoveryHolds } : {}),
|
||||
...(current.legacyRecoverySources.length ? { legacyRecoverySources: current.legacyRecoverySources } : {}),
|
||||
...(current.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}),
|
||||
...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id,
|
||||
@@ -289,7 +305,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, legacyRecoverySources: current.legacyRecoverySources };
|
||||
return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds };
|
||||
});
|
||||
if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true };
|
||||
let operation = reserved.operation!;
|
||||
@@ -341,11 +357,17 @@ export async function migrateRemoteWorkspace(input: {
|
||||
await lock.execute(sql`select 1`);
|
||||
const publication = await input.db.transaction(async tx => {
|
||||
await lockMigrationParents(tx, input.companyId, reserved.workspace.sourceIssueId);
|
||||
for (const source of reserved.legacyRecoverySources) {
|
||||
for (const held of reserved.finalizedRecoveryHolds) {
|
||||
await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, held.actionId))).for("update", { noWait: true });
|
||||
}
|
||||
const protectedRunIds = [...new Set([...reserved.legacyRecoverySources.map(source => source.runId),
|
||||
...reserved.finalizedRecoveryHolds.map(held => held.runId)])].sort();
|
||||
for (const runId of protectedRunIds) {
|
||||
await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations)
|
||||
.where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, source.runId))).for("update", { noWait: true });
|
||||
.where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, runId))).for("update", { noWait: true });
|
||||
await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, source.runId))).for("update", { noWait: true });
|
||||
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, runId))).for("update", { noWait: true });
|
||||
}
|
||||
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");
|
||||
@@ -374,6 +396,7 @@ 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.finalizedRecoveryHolds) !== canonical(reserved.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed");
|
||||
if (canonical(validation.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed");
|
||||
const continuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace);
|
||||
if (canonical(continuity) !== canonical(operation.metadata?.nativeSessionContinuity ?? null)) throw new Error("native_workspace_migration_source_changed");
|
||||
@@ -384,6 +407,8 @@ export async function migrateRemoteWorkspace(input: {
|
||||
if (reserved.legacyRecoverySources.length) {
|
||||
metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id };
|
||||
for (const source of reserved.legacyRecoverySources) {
|
||||
const markerKey = source.reference?.state === "finalized"
|
||||
? "finalizedWorkspaceRecoveryMigration" : "legacyWorkspaceRecoveryMigration";
|
||||
// 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
|
||||
@@ -399,7 +424,7 @@ export async function migrateRemoteWorkspace(input: {
|
||||
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,
|
||||
[markerKey]: { 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)));
|
||||
|
||||
@@ -2,7 +2,7 @@ import { and, desc, eq, sql, type AnyColumn } 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 { readNativeWorkspaceSyncReference, type NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js";
|
||||
import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js";
|
||||
import { readSnapshot } from "./workspace-recovery-snapshot.js";
|
||||
|
||||
@@ -62,6 +62,28 @@ export function workspaceRecoveryDisplay(evidence: Record<string, unknown>, runI
|
||||
};
|
||||
}
|
||||
|
||||
async function verifiedNativeCopybackReceipt(db: Db, input: {
|
||||
companyId: string; issueId: string; runId: string; reference: NativeWorkspaceSyncReference;
|
||||
}) {
|
||||
const { reference } = input;
|
||||
if (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, input.companyId),
|
||||
eq(workspaceOperations.issueId, input.issueId), eq(workspaceOperations.heartbeatRunId, input.runId),
|
||||
eq(workspaceOperations.phase, "workspace_recovery"), eq(workspaceOperations.status, "succeeded"),
|
||||
sql`${workspaceOperations.metadata}->>'leaseId' = ${reference.leaseId}`,
|
||||
sql`${workspaceOperations.metadata}->>'providerLeaseId' = ${reference.providerLeaseId}`,
|
||||
sql`${workspaceOperations.metadata}->>'descriptorSha256' = ${reference.descriptorSha256}`,
|
||||
sql`${workspaceOperations.metadata}->>'finalHostSha256' = ${reference.finalHostSha256}`,
|
||||
)).limit(1);
|
||||
if (!receipt || typeof receipt.metadata?.recoveryActionId !== "string") return null;
|
||||
const [action] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId),
|
||||
eq(issueRecoveryActions.id, receipt.metadata.recoveryActionId)));
|
||||
const state = readWorkspaceRecovery(action?.evidence ?? null);
|
||||
return state?.runId === input.runId && state.status === "recovered" ? receipt.id : null;
|
||||
}
|
||||
|
||||
/** Only the host writes workspace operations. Adapter output and a user-supplied
|
||||
* repair note cannot substitute for this exact copyback receipt. Stop/authority
|
||||
* are checked separately at every continuation admission. */
|
||||
@@ -72,20 +94,18 @@ 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
|
||||
if (typeof issueId === "string" && (reference?.state === "prepared" || reference?.state === "finalized"
|
||||
&& run.runnerProfileJson?.finalizedWorkspaceRecoveryMigration != null || !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;
|
||||
const allocations = migrated.recoveryAllocations ?? (reference
|
||||
? [{ leaseId: reference.leaseId, providerLeaseId: reference.providerLeaseId }] : null);
|
||||
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),
|
||||
@@ -98,7 +118,14 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick<typeof he
|
||||
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;
|
||||
if (!saved || readWorkspaceRecovery(saved.evidence)?.status !== "recovered") {
|
||||
const copyback = reference && allocation.leaseId === reference.leaseId
|
||||
&& allocation.providerLeaseId === reference.providerLeaseId
|
||||
? await verifiedNativeCopybackReceipt(db, { companyId: run.companyId, issueId, runId: run.id, reference }) : null;
|
||||
if (!copyback) return null;
|
||||
receiptId = copyback;
|
||||
continue;
|
||||
}
|
||||
try { readSnapshot(saved.metadata, run.companyId, run.id); } catch { return null; }
|
||||
receiptId = saved.id;
|
||||
}
|
||||
@@ -106,6 +133,7 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick<typeof he
|
||||
}
|
||||
}
|
||||
if (typeof issueId !== "string" || (!remote && (!reference || reference.state !== "finalized" || !reference.finalHostSha256))) return null;
|
||||
if (!remote) return verifiedNativeCopybackReceipt(db, { companyId: run.companyId, issueId, runId: run.id, reference: reference! });
|
||||
const [receipt] = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.issueId, issueId),
|
||||
eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.phase, "workspace_recovery"),
|
||||
@@ -128,8 +156,7 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick<typeof he
|
||||
eq(executionWorkspaceSnapshots.epoch, remote.epoch), eq(executionWorkspaceSnapshots.sourceRunId, run.id),
|
||||
));
|
||||
if (!snapshot || snapshot.sha256 !== proof.sha256 || snapshot.generation !== proof.generation) return null;
|
||||
} else if (!reference || proof.leaseId !== reference.leaseId || proof.providerLeaseId !== reference.providerLeaseId
|
||||
|| proof.descriptorSha256 !== reference.descriptorSha256 || proof.finalHostSha256 !== reference.finalHostSha256) return null;
|
||||
}
|
||||
const [action] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
eq(issueRecoveryActions.id, proof.recoveryActionId),
|
||||
|
||||
Reference in new issue
Block a user