mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 16:02:36 +01:00
fix(runtime): preserve settled failed harness backups online
Add a source-scoped preservation owner for sealed terminal native backup files. Recheck task, coordinator, selection, source lease, physical lock and exact-current payload at publication without changing failed outcomes, warnings or original files. Keep missing and uncertain sources blocked. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
03498f90ee
commit
ddabe48660
6 files changed
+318
-10
No files matched your search
@@ -317,14 +317,26 @@ It starts no scheduler, does not drain tasks, and always reports
|
||||
`enrollmentReady: false`. Check serving memory headroom before running a separate
|
||||
process on a tenant. Keep the deployment on normal releases.
|
||||
|
||||
Only a successful, finished source with a released committed coordinator and an
|
||||
accepted result can publish through this path. At both database fences, hold the
|
||||
A successful, finished source with a released committed coordinator and an
|
||||
accepted result publishes through its normal runtime owner. At both database fences, hold the
|
||||
task share lock and recheck its checkout, related queued/running executions,
|
||||
recorded session and Runner, exact backup-stamp lease, and unchanged local
|
||||
manifest and suspended state. A concurrent turn, moved source, revoked stamp or
|
||||
changed owner defers that source. Unrelated executions continue unchanged.
|
||||
No lease, result, task or lifecycle state is rewritten to make a source eligible.
|
||||
|
||||
The same entry can preserve sealed failed or cancelled backup files through a
|
||||
separate source owner. The source must retain its exact selected task/session,
|
||||
released committed result or terminal-failure coordinator, and recorded backup
|
||||
stamp. Every publication checks the physical finalization lock, task checkout,
|
||||
source run and coordinator, source lease, manifest, and payload hashes. A remaining
|
||||
coordinator or physical owner blocks the copy even after its deadline expires.
|
||||
Historical PID fields are preserved and are not termination proof. A valid
|
||||
`previous` tree cannot authenticate a changed `current` tree. Failed publication
|
||||
retains the pending encrypted object receipt and every original file. The copy
|
||||
does not change the failed outcome, warnings, result, allocation or session.
|
||||
It does not prove that an absent backup was recovered or authorize detachment.
|
||||
|
||||
Run the controller-only phase separately so it does not re-read provider
|
||||
archives that were already verified:
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import { nativeHarnessBackupOwner, persistNativeHarnessBackup, verifyDurableNati
|
||||
import { MAX_NATIVE_HARNESS_BYTES, MAX_NATIVE_HARNESS_OBJECT_BYTES, nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js";
|
||||
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
|
||||
import { NativeHarnessBackupSourceError, readNativeHarnessBackupSource } from "./native-harness-backup-source.js";
|
||||
import { settledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js";
|
||||
|
||||
export async function hasLocalNativeHarnessObligations(db: Db): Promise<boolean> {
|
||||
const rows = await db.execute(sql`select 1 from agent_task_sessions s
|
||||
@@ -181,7 +182,25 @@ export async function prestageNativeHarnessBackups(db: Db, input: { storage: Sto
|
||||
scopeDigest: source.scopeDigest, manifestSha256: source.manifestSha256, signal });
|
||||
return "current";
|
||||
}
|
||||
if (run.status !== "succeeded" || !run.finishedAt) return "source_not_successful";
|
||||
if (run.status !== "succeeded" || !run.finishedAt) {
|
||||
if (!run.finishedAt || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)) return "source_not_successful";
|
||||
const providerLeaseId = await verifiedSourceLease(db, source, run, signal);
|
||||
if (!providerLeaseId) return "source_unverified";
|
||||
try {
|
||||
const owner = await settledNativeHarnessBackupOwner(db, { companyId: run.companyId, issueId: run.issueId,
|
||||
runId: source.runId, sessionId: run.sessionId, runnerInstanceId: run.runnerId, scopeDigest: source.scopeDigest,
|
||||
root: source.root, manifestSha256: source.manifestSha256, providerLeaseId, signal });
|
||||
if (!owner) return "source_not_successful";
|
||||
await persistNativeHarnessBackup({ db, storage: input.storage, owner, root: source.root,
|
||||
scopeDigest: source.scopeDigest, signal, maxArchiveBytes: source.archiveLimit });
|
||||
return "published";
|
||||
} catch (error) {
|
||||
signal.throwIfAborted();
|
||||
if (error instanceof NativeHarnessBackupSourceError || error instanceof Error && error.message === "native_harness_preservation_source_unsettled"
|
||||
|| (error as { code?: string })?.code === "55P03") return "source_changed";
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.issueId),
|
||||
eq(nativeRunFinalizations.runId, source.runId)));
|
||||
|
||||
@@ -211,10 +211,14 @@ export function verifyNativeHarnessBackupStamp(value: unknown, expectedProviderL
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function verifyNativeHarnessBackupStampAsync(value: unknown, expectedProviderLeaseId: string, signal: AbortSignal): Promise<boolean> {
|
||||
export async function verifyNativeHarnessBackupStampAsync(value: unknown, expectedProviderLeaseId: string, signal: AbortSignal,
|
||||
exactRoot?: string): Promise<boolean> {
|
||||
const stamp = readNativeHarnessBackupStamp(value, expectedProviderLeaseId);
|
||||
if (!stamp) return false;
|
||||
for (const candidate of backupCandidates(stamp)) {
|
||||
// A preservation publisher must authenticate the tree it actually copied;
|
||||
// an intact previous backup cannot authorize a changed current payload.
|
||||
if (exactRoot !== undefined && resolve(candidate) !== resolve(exactRoot)) continue;
|
||||
signal.throwIfAborted();
|
||||
try {
|
||||
const directories = matchingBackupDirectories(candidate, stamp);
|
||||
|
||||
@@ -23,6 +23,8 @@ import { pinNativeProviderArtifactsForBackup, releaseRetiredNativeProviderArtifa
|
||||
type NativeProviderArtifactReceipt, type NativeProviderArtifactSource } from "./native-provider-artifacts.js";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
import type { SettledControllerArchiveOwner } from "./native-controller-archive-source.js";
|
||||
import type { SettledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js";
|
||||
import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js";
|
||||
|
||||
const SCHEMA = "paperclip.native-harness-object.v1";
|
||||
type Binding = { companyId: string; scopeDigest: string; id: string };
|
||||
@@ -97,9 +99,13 @@ export async function nativeHarnessBackupOwner(db: Db, input: { companyId: strin
|
||||
if (!row || !row.leaseOwner && (row.phase !== "committed" || !row.resultId)) fail("owner_missing");
|
||||
return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointFenceOwner;
|
||||
}
|
||||
async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner,
|
||||
inactiveSource?: InactiveSource) {
|
||||
async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner | SettledNativeHarnessBackupOwner,
|
||||
inactiveSource?: InactiveSource, signal?: AbortSignal) {
|
||||
if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner");
|
||||
if ("kind" in owner && owner.kind === "settled_native_harness_backup") {
|
||||
await (await import("./native-harness-preservation-source.js")).assertSettledNativeHarnessBackupSource(tx as unknown as Db, owner, signal);
|
||||
return;
|
||||
}
|
||||
if ("kind" in owner && owner.kind === "settled_controller_archive") {
|
||||
await (await import("./native-controller-archive-source.js")).assertSettledControllerArchiveSource(tx as unknown as Db, owner);
|
||||
return;
|
||||
@@ -188,12 +194,14 @@ export async function assertNativeHarnessBackupOwner(db: Db, owner: NativeRunner
|
||||
}
|
||||
|
||||
export async function persistNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner;
|
||||
db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner | SettledNativeHarnessBackupOwner;
|
||||
scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number;
|
||||
inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string };
|
||||
providerArtifacts?: NativeProviderArtifactSource;
|
||||
}): Promise<void> {
|
||||
if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid");
|
||||
if ("kind" in input.owner && input.owner.kind === "settled_native_harness_backup"
|
||||
&& (input.scopeDigest !== input.owner.scopeDigest || input.root !== input.owner.root)) fail("preservation_source_changed");
|
||||
if (input.providerArtifacts) {
|
||||
if ("kind" in input.owner) fail("provider_artifacts_require_runtime_owner");
|
||||
const receipt = await retainNativeProviderArtifacts({ ...input, owner: input.owner, source: input.providerArtifacts });
|
||||
@@ -206,10 +214,13 @@ export async function persistNativeHarnessBackup(input: {
|
||||
const manifestBytes = source?.manifestBytes ?? await readFile(join(input.root, "manifest.json"));
|
||||
const providerArtifacts = JSON.parse(manifestBytes.toString("utf8")).providerArtifacts as NativeProviderArtifactReceipt | undefined;
|
||||
const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`;
|
||||
if ("kind" in input.owner && input.owner.kind === "settled_native_harness_backup"
|
||||
&& manifestSha256 !== input.owner.manifestSha256) fail("preservation_source_changed");
|
||||
const inactiveSource = input.inactiveSource ? { ...input.inactiveSource, root: input.root, manifestSha256, scopeDigest: input.scopeDigest } : undefined;
|
||||
const binding = { companyId: input.owner.companyId, scopeDigest: input.scopeDigest, id: randomUUID() };
|
||||
const scratch = await mkdtemp(join(os.tmpdir(), "pc-harness-upload-"));
|
||||
const key = randomBytes(32);
|
||||
const publications: ActivityPublication[] = [];
|
||||
try {
|
||||
const archive = join(scratch, "archive"), encrypted = join(scratch, "encrypted");
|
||||
await captureNativeHarnessArchive(input.root, archive, input.signal, input.maxArchiveBytes, providerArtifacts ? ["provider-artifacts"] : []);
|
||||
@@ -223,7 +234,7 @@ export async function persistNativeHarnessBackup(input: {
|
||||
key: key.toString("base64"), iv: iv.toString("base64"), tag: cipher.getAuthTag().toString("base64"), plaintext })) };
|
||||
if (ref.byteSize > MAX_NATIVE_HARNESS_OBJECT_BYTES) fail("size_limit_exceeded");
|
||||
await input.db.transaction(async tx => {
|
||||
await fence(tx, input.owner, inactiveSource);
|
||||
await fence(tx, input.owner, inactiveSource, input.signal);
|
||||
await tx.insert(nativeHarnessBackups).values({ ...binding, issueId: input.owner.issueId, runId: input.owner.runId,
|
||||
manifestSha256, reference: ref });
|
||||
if (providerArtifacts) await pinNativeProviderArtifactsForBackup(tx, { companyId: input.owner.companyId,
|
||||
@@ -234,7 +245,7 @@ export async function persistNativeHarnessBackup(input: {
|
||||
byteSize: ref.byteSize, sha256: ref.sha256, signal: input.signal });
|
||||
await receive(input.storage, ref, undefined, input.signal);
|
||||
await input.db.transaction(async tx => {
|
||||
await fence(tx, input.owner, inactiveSource);
|
||||
await fence(tx, input.owner, inactiveSource, input.signal);
|
||||
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`${ref.companyId}:${ref.scopeDigest}`}, 0))`);
|
||||
const [pending] = await tx.select({ status: nativeHarnessBackups.status }).from(nativeHarnessBackups)
|
||||
.where(eq(nativeHarnessBackups.id, ref.id)).for("update");
|
||||
@@ -244,12 +255,19 @@ export async function persistNativeHarnessBackup(input: {
|
||||
eq(nativeHarnessBackups.status, "current"), sql`${nativeHarnessBackups.runId} <> ${input.owner.runId}::uuid`,
|
||||
sql`${heartbeatRuns.createdAt} >= (select created_at from heartbeat_runs where id = ${input.owner.runId}::uuid)`)).limit(1);
|
||||
if (successor) fail("owner_superseded");
|
||||
await fence(tx, input.owner, inactiveSource);
|
||||
await fence(tx, input.owner, inactiveSource, input.signal);
|
||||
await tx.update(nativeHarnessBackups).set({ status: "superseded", updatedAt: new Date() })
|
||||
.where(and(scope(ref), eq(nativeHarnessBackups.status, "current")));
|
||||
await tx.update(nativeHarnessBackups).set({ status: "current", updatedAt: new Date() })
|
||||
.where(and(eq(nativeHarnessBackups.id, ref.id), eq(nativeHarnessBackups.status, "pending")));
|
||||
if ("kind" in input.owner && input.owner.kind === "settled_native_harness_backup") {
|
||||
await logActivity(tx as unknown as Db, { companyId: input.owner.companyId, actorType: "system", actorId: "native-harness-preservation",
|
||||
action: "native_harness_backup_preserved", entityType: "heartbeat_run", entityId: input.owner.runId,
|
||||
agentId: input.owner.agentId, runId: input.owner.runId,
|
||||
details: { scopeDigest: ref.scopeDigest, manifestSha256, backupId: ref.id, originalFilesRetained: true } }, publications);
|
||||
}
|
||||
});
|
||||
for (const publication of publications) publishActivity(publication);
|
||||
} finally { key.fill(0); await removeNativeHarnessBackup(scratch); }
|
||||
}
|
||||
|
||||
|
||||
+148
@@ -0,0 +1,148 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { cp, mkdir, readFile, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agentTaskSessions, environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
|
||||
import { createStorageService } from "../../storage/service.js";
|
||||
import { buildNativeHarnessBackupManifest } from "./native-session-executor.js";
|
||||
import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { prestageNativeHarnessBackups } from "./native-harness-backup-migration.js";
|
||||
import { persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { settledNativeHarnessBackupOwner } from "./native-harness-preservation-source.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
|
||||
|
||||
describeEmbeddedPostgres("online preservation of settled legacy harness backups", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
const homes: string[] = [];
|
||||
afterEach(async () => { for (const home of homes.splice(0)) await removeNativeHarnessBackup(home); vi.unstubAllEnvs(); });
|
||||
async function fixture() {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "51".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const base = join(runtimeRoot, randomUUID()); homes.push(base); await mkdir(base);
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", base);
|
||||
const sessionScopeId = randomUUID(), scopeDigest = createHash("sha256").update(sessionScopeId).digest("hex");
|
||||
const root = join(base, scopeDigest, "failover-backups", "current");
|
||||
await mkdir(join(root, "codex-home", "sessions"), { recursive: true }); await mkdir(join(root, "runner"));
|
||||
const saved = { providerSessionId: randomUUID(), history: "surviving failed provider history" };
|
||||
await writeFile(join(root, "codex-home", "sessions", "history.json"), JSON.stringify(saved), { mode: 0o600 });
|
||||
await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ runId: f.runId, lifecycle: "suspended" }));
|
||||
const normalizedSessionId = randomUUID(), runnerInstanceId = randomUUID(), providerLeaseId = randomUUID();
|
||||
const execution = { provider: { kind: "codex" }, binding: { companyId, issueId: f.issueId, runId: f.runId,
|
||||
agentId, executionWorkspaceId: randomUUID() }, workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as never;
|
||||
const manifest = buildNativeHarnessBackupManifest({ backupRoot: root, execution, runnerInstanceId,
|
||||
providerSessionIdentity: { providerSessionId: saved.providerSessionId }, sourceProviderLeaseId: providerLeaseId });
|
||||
await writeFile(join(root, "manifest.json"), JSON.stringify(manifest));
|
||||
const stamp = createNativeHarnessBackupStamp({ manifestPath: join(root, "manifest.json"), sessionScopeId,
|
||||
authorizedProviderLeaseId: providerLeaseId, runnerInstanceId, normalizedSessionId, completedAt: manifest.completedAt });
|
||||
await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), nativeSessionId: normalizedSessionId,
|
||||
runnerInstanceId, processPid: 123456, processStartedAt: new Date(1) }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null,
|
||||
recoveryState: "original_failure_retained" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await f.db.update(issues).set({ status: "done", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
await f.db.insert(agentTaskSessions).values({ companyId, agentId, adapterType: "paperclip_runner", taskKey: f.issueId, lastRunId: f.runId });
|
||||
await f.db.insert(environmentLeases).values({ companyId, issueId: f.issueId, heartbeatRunId: f.runId, providerLeaseId,
|
||||
status: "expired", metadata: { nativeHarnessBackup: stamp } });
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(join(runtimeRoot, "durable-objects")));
|
||||
return { ...f, companyId, root, scopeDigest, storage, saved, stamp, normalizedSessionId, runnerInstanceId, providerLeaseId };
|
||||
}
|
||||
it("preserves failed backup bytes online while retaining failure and unrelated active work", async () => {
|
||||
const f = await fixture(), active = await environment.seedRun(randomUUID());
|
||||
const [before] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [coordinator] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toEqual({ inspected: 1, published: 1, current: 0, deferred: {} });
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(before);
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(coordinator);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active.runId)))[0].status).toBe("running");
|
||||
expect(await readFile(join(f.root, "codex-home", "sessions", "history.json"), "utf8")).toBe(JSON.stringify(f.saved));
|
||||
const restored = join(environment.current.runtimeRoot, randomUUID()); homes.push(restored);
|
||||
expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: restored })).toBe(true);
|
||||
expect(JSON.parse(await readFile(join(restored, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(f.saved);
|
||||
expect((await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)))[0].status).toBe("current");
|
||||
});
|
||||
it.each([["failed", "committed"], ["cancelled", "terminal_failure"], ["cancelled", "committed"]])("preserves %s source bytes with %s ownership without changing its status", async (status, phase) => {
|
||||
const f = await fixture();
|
||||
await f.db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (phase === "committed") {
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId,
|
||||
completionContractId: run.completionContractId!, serverFingerprint: "settled-historical-result", schemaStatus: "accepted",
|
||||
resultJson: { summary: "Original settled result" }, canonicalSha256: "settled-historical-result" }).returning();
|
||||
await f.db.update(nativeRunFinalizations).set({ phase, resultId: result.id }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
}
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toMatchObject({ published: 1, deferred: {} });
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe(status);
|
||||
});
|
||||
it("never takes a nonnull expired coordinator or durable physical owner", async () => {
|
||||
const f = await fixture(), put = vi.spyOn(f.storage, "putFile");
|
||||
await f.db.update(nativeRunFinalizations).set({ leaseOwner: "original-owner", leaseExpiresAt: new Date(0) })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toMatchObject({ published: 0, deferred: { source_not_successful: 1 } });
|
||||
await f.db.update(nativeRunFinalizations).set({ leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceFinalizationOwner: { token: "physical-owner" } } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toMatchObject({ published: 0, deferred: { source_changed: 1 } });
|
||||
expect(put).not.toHaveBeenCalled();
|
||||
});
|
||||
it("respects the physical advisory lock while unrelated work stays active", async () => {
|
||||
const f = await fixture(), put = vi.spyOn(f.storage, "putFile");
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; }), ready = new Promise<void>(resolve => { entered = resolve; });
|
||||
const holder = f.db.transaction(async tx => { expect(await tryLockNativeWorkspaceFinalization(tx as never, f)).toBe(true); entered(); await gate; });
|
||||
try {
|
||||
await ready;
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toMatchObject({ published: 0, deferred: { source_changed: 1 } });
|
||||
expect(put).not.toHaveBeenCalled();
|
||||
} finally { release(); await holder; }
|
||||
});
|
||||
it.each(["owner", "payload"])("retains the uploaded pending receipt when %s changes before publication", async change => {
|
||||
const f = await fixture(), put = f.storage.putFile.bind(f.storage);
|
||||
vi.spyOn(f.storage, "putFile").mockImplementation(async input => {
|
||||
const result = await put(input);
|
||||
if (change === "owner") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "new-owner", leaseExpiresAt: new Date(0) })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
else await writeFile(join(f.root, "codex-home", "sessions", "history.json"), "newer surviving files");
|
||||
return result;
|
||||
});
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toMatchObject({ published: 0, deferred: { source_changed: 1 } });
|
||||
expect(await f.db.select({ status: nativeHarnessBackups.status }).from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)))
|
||||
.toEqual([{ status: "pending" }]);
|
||||
if (change === "owner") expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].leaseOwner).toBe("new-owner");
|
||||
else expect(await readFile(join(f.root, "codex-home", "sessions", "history.json"), "utf8")).toBe("newer surviving files");
|
||||
});
|
||||
it("does not authenticate a changed current payload from an intact previous backup", async () => {
|
||||
const f = await fixture(), previous = join(f.root, "..", "previous");
|
||||
await cp(f.root, previous, { recursive: true });
|
||||
await writeFile(join(f.root, "codex-home", "sessions", "history.json"), "changed current payload");
|
||||
const signal = AbortSignal.timeout(15_000);
|
||||
expect(await verifyNativeHarnessBackupStampAsync(f.stamp, f.providerLeaseId, signal)).toBe(true);
|
||||
expect(await verifyNativeHarnessBackupStampAsync(f.stamp, f.providerLeaseId, signal, f.root)).toBe(false);
|
||||
expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal })).toMatchObject({ published: 0 });
|
||||
expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]);
|
||||
});
|
||||
it("binds the archive payload and scope to the checked preservation owner", async () => {
|
||||
const f = await fixture(), signal = AbortSignal.timeout(15_000);
|
||||
const owner = await settledNativeHarnessBackupOwner(f.db, { ...f, sessionId: f.normalizedSessionId,
|
||||
manifestSha256: f.stamp.manifestSha256, signal });
|
||||
expect(owner).not.toBeNull();
|
||||
const put = vi.spyOn(f.storage, "putFile");
|
||||
await expect(persistNativeHarnessBackup({ ...f, owner: owner!, root: join(f.root, "..", "previous"), signal }))
|
||||
.rejects.toThrow("preservation_source_changed");
|
||||
await expect(persistNativeHarnessBackup({ ...f, owner: owner!, scopeDigest: "f".repeat(64), signal }))
|
||||
.rejects.toThrow("preservation_source_changed");
|
||||
expect(put).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,107 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { resolve } from "node:path";
|
||||
import { and, eq, inArray, or, sql } from "drizzle-orm";
|
||||
import { agentTaskSessions, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { adapterExecutionControls } from "../adapter-execution-control.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
|
||||
import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js";
|
||||
import { readNativeHarnessBackupSource } from "./native-harness-backup-source.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
type Coordinator = typeof nativeRunFinalizations.$inferSelect;
|
||||
type Lease = typeof environmentLeases.$inferSelect;
|
||||
export type SettledNativeHarnessBackupOwner = {
|
||||
kind: "settled_native_harness_backup";
|
||||
companyId: string; issueId: string; agentId: string; runId: string;
|
||||
sessionId: string; runnerInstanceId: string; scopeDigest: string; root: string; manifestSha256: string;
|
||||
selectedSessionId: string; selectedLastRunId: string; sourceLeaseId: string; providerLeaseId: string;
|
||||
runDigest: string; coordinatorDigest: string; leaseDigest: string;
|
||||
};
|
||||
const runDigest = (run: Run) => nativeSha256({ status: run.status, finishedAt: run.finishedAt?.toISOString(),
|
||||
controllerBootId: run.controllerBootId, controllerLeaseExpiresAt: run.controllerLeaseExpiresAt?.toISOString(),
|
||||
processPid: run.processPid, processGroupId: run.processGroupId, processStartedAt: run.processStartedAt?.toISOString(),
|
||||
runnerInstanceId: run.runnerInstanceId, runnerProfileJson: run.runnerProfileJson });
|
||||
const coordinatorDigest = (row: Coordinator) => nativeSha256({ phase: row.phase, resultId: row.resultId,
|
||||
controllerBootId: row.controllerBootId, controllerGeneration: row.controllerGeneration, recoveryState: row.recoveryState });
|
||||
const leaseDigest = (row: Lease) => nativeSha256({ heartbeatRunId: row.heartbeatRunId, status: row.status,
|
||||
releasedAt: row.releasedAt?.toISOString(), cleanupStatus: row.cleanupStatus, providerLeaseId: row.providerLeaseId,
|
||||
nativeHarnessBackup: row.metadata?.nativeHarnessBackup });
|
||||
const settled = (row: Coordinator) => row.leaseOwner === null && row.nextAttemptAt === null
|
||||
&& (row.phase === "terminal_failure" || row.phase === "committed" && row.resultId !== null);
|
||||
const fail = (): never => { throw new Error("native_harness_preservation_source_unsettled"); };
|
||||
|
||||
/** Copy sealed historical files without certifying provider termination, changing
|
||||
* a failed run, or taking an expired coordinator. Every publication rechecks it. */
|
||||
export async function settledNativeHarnessBackupOwner(db: Db, input: {
|
||||
companyId: string; issueId: string; runId: string; sessionId: string; runnerInstanceId: string;
|
||||
scopeDigest: string; root: string; manifestSha256: string; providerLeaseId: string; signal: AbortSignal;
|
||||
}): Promise<SettledNativeHarnessBackupOwner | null> {
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId)));
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId),
|
||||
eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId)));
|
||||
if (!run || !coordinator || !settled(coordinator)) return null;
|
||||
const selected = await db.select({ id: agentTaskSessions.id, lastRunId: agentTaskSessions.lastRunId }).from(agentTaskSessions)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, agentTaskSessions.companyId), eq(heartbeatRuns.agentId, agentTaskSessions.agentId),
|
||||
eq(heartbeatRuns.id, agentTaskSessions.lastRunId))).where(and(eq(agentTaskSessions.companyId, input.companyId),
|
||||
eq(agentTaskSessions.agentId, run.agentId), eq(agentTaskSessions.adapterType, "paperclip_runner"),
|
||||
eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.sessionId))).limit(2);
|
||||
if (selected.length !== 1 || !selected[0]!.lastRunId) return null;
|
||||
const leases = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
|
||||
eq(environmentLeases.issueId, input.issueId), eq(environmentLeases.providerLeaseId, input.providerLeaseId),
|
||||
sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${input.manifestSha256}`));
|
||||
const lease = leases.find(row => readNativeHarnessBackupStamp(row.metadata?.nativeHarnessBackup, input.providerLeaseId)?.sessionScopeSha256 === input.scopeDigest);
|
||||
if (!lease) return null;
|
||||
const owner: SettledNativeHarnessBackupOwner = { kind: "settled_native_harness_backup", companyId: input.companyId,
|
||||
issueId: input.issueId, agentId: run.agentId, runId: input.runId, sessionId: input.sessionId, runnerInstanceId: input.runnerInstanceId,
|
||||
scopeDigest: input.scopeDigest, root: input.root, manifestSha256: input.manifestSha256,
|
||||
selectedSessionId: selected[0]!.id, selectedLastRunId: selected[0]!.lastRunId,
|
||||
sourceLeaseId: lease.id, providerLeaseId: input.providerLeaseId,
|
||||
runDigest: runDigest(run), coordinatorDigest: coordinatorDigest(coordinator), leaseDigest: leaseDigest(lease) };
|
||||
await db.transaction(tx => assertSettledNativeHarnessBackupSource(tx as unknown as Db, owner, input.signal));
|
||||
return owner;
|
||||
}
|
||||
|
||||
export async function assertSettledNativeHarnessBackupSource(db: Db, owner: SettledNativeHarnessBackupOwner,
|
||||
signal = AbortSignal.timeout(300_000)) {
|
||||
signal.throwIfAborted();
|
||||
await db.execute(sql`set local lock_timeout = '1s'`);
|
||||
if (!await tryLockNativeWorkspaceFinalization(db, owner)) fail();
|
||||
const [task] = await db.select().from(issues).where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("update");
|
||||
if (!task || task.executionRunId || task.checkoutRunId) fail();
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, owner.companyId),
|
||||
eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.runId, owner.runId))).for("update");
|
||||
if (!coordinator || !settled(coordinator) || coordinatorDigest(coordinator) !== owner.coordinatorDigest) fail();
|
||||
const [selection] = await db.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, owner.companyId),
|
||||
eq(agentTaskSessions.agentId, owner.agentId), eq(agentTaskSessions.id, owner.selectedSessionId),
|
||||
eq(agentTaskSessions.adapterType, "paperclip_runner"),
|
||||
eq(agentTaskSessions.lastRunId, owner.selectedLastRunId))).for("share");
|
||||
const [selectedRun] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId),
|
||||
eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.id, owner.selectedLastRunId),
|
||||
eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId)));
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId),
|
||||
eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.id, owner.runId), eq(heartbeatRuns.nativeIssueId, owner.issueId),
|
||||
eq(heartbeatRuns.nativeSessionId, owner.sessionId), eq(heartbeatRuns.runnerInstanceId, owner.runnerInstanceId), eq(heartbeatRuns.runtimeMode, "native"))).for("share");
|
||||
const [clock] = await db.execute<{ now: Date }>(sql`select clock_timestamp() as now`);
|
||||
if (!selection || !selectedRun || !run?.finishedAt || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)
|
||||
|| run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date(clock.now)
|
||||
|| adapterExecutionControls.has(run.id) || run.runnerProfileJson?.nativeWorkspaceFinalizationOwner != null
|
||||
|| runDigest(run) !== owner.runDigest) fail();
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId),
|
||||
inArray(heartbeatRuns.status, ["queued", "running"]), or(eq(heartbeatRuns.nativeIssueId, owner.issueId),
|
||||
eq(heartbeatRuns.issueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId)))).limit(1);
|
||||
if (active) fail();
|
||||
const [lease] = await db.select().from(environmentLeases).where(and(eq(environmentLeases.id, owner.sourceLeaseId),
|
||||
eq(environmentLeases.companyId, owner.companyId), eq(environmentLeases.issueId, owner.issueId),
|
||||
eq(environmentLeases.providerLeaseId, owner.providerLeaseId))).for("share");
|
||||
const stamp = readNativeHarnessBackupStamp(lease?.metadata?.nativeHarnessBackup, owner.providerLeaseId);
|
||||
if (!lease || !stamp || leaseDigest(lease) !== owner.leaseDigest || stamp.sessionScopeSha256 !== owner.scopeDigest
|
||||
|| stamp.manifestSha256 !== owner.manifestSha256 || stamp.normalizedSessionId !== owner.sessionId
|
||||
|| stamp.runnerInstanceId !== owner.runnerInstanceId) fail();
|
||||
if (!/^[a-f0-9]{64}$/.test(owner.scopeDigest)
|
||||
|| resolve(owner.root) !== resolve(nativeHarnessStateBase(), owner.scopeDigest, "failover-backups", "current")) fail();
|
||||
const source = await readNativeHarnessBackupSource(owner.root);
|
||||
if (!source || source.state.runId !== owner.runId || source.state.lifecycle !== "suspended"
|
||||
|| `sha256:${createHash("sha256").update(source.manifestBytes).digest("hex")}` !== owner.manifestSha256
|
||||
|| !await verifyNativeHarnessBackupStampAsync(stamp, owner.providerLeaseId, signal, owner.root)) fail();
|
||||
}
|
||||
Reference in new issue
Block a user