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 (#19715)
## Thinking Path
> - Paperclip coordinates agents, tasks, and their saved execution
state.
> - A replacement server needs native provider backups in shared
storage.
> - The online pre-copy owner accepts successful runs with a committed
result.
> - A sealed backup from a failed or cancelled run can still contain
user history.
> - The existing owner defers those files even after their source
ownership is settled.
> - This change adds a separate preservation owner with exact source and
publication checks.
> - The copy retains the failed outcome and original files while making
the backup readable from a fresh home.
## Linked Issues or Issue Description
Refs #19137 and #19108 for the existing pre-copy and encrypted backup
owners.
**What happened?**
Online pre-copy reports `source_not_successful` for a sealed legacy
backup from a settled failed or cancelled run. The provider files remain
tied to the old disk.
**Expected behavior**
Copy the exact recorded backup when the source has no remaining writer
ownership. Preserve the run status, warnings, result, allocation,
selected session, and local files.
**Steps to reproduce**
1. Create a lease-stamped suspended native backup.
2. Settle its source as a failed run with a terminal-failure coordinator
and no owner.
3. Keep an unrelated execution active.
4. Run canonical online pre-copy. The previous owner defers the source.
## What Changed
- Add a narrow preservation owner for terminal source files. It checks
the physical finalization lock, task, coordinator, run, selected
session, source lease, and source fingerprints.
- Recheck that owner before the encrypted upload receipt and
current-backup publication.
- Reuse the finite workspace keepalive during long source checks and
defer source-local invalid archives.
- Add optional exact-root stamp verification. A valid previous backup
cannot authorize a changed current payload. Existing callers keep their
fallback behavior.
- Bind the archive input to the checked owner and record an activity
event after publication.
- Keep pending object receipts after an uncertain or failed publication.
- Add focused PostgreSQL and filesystem controls. Update the existing
migration runbook.
## Verification
- Real PostgreSQL baseline on unchanged production: the new positive
case fails with `source_not_successful` instead of publishing the
backup.
- Same case passes after the change and decrypts/restores the exact
provider history.
- Final head 4b6cf093be: three focused suites pass, 81 tests in 62.26
seconds, including existing storage and stamp consumers. The canonical
PostgreSQL wrapper exited zero and completed cleanup.
- Controls cover cancelled and failed sources, real committed results,
expired nonnull owners, physical owner/lock contention, late
source/owner changes, previous-tree fallback, and archive root/scope
binding.
- Both review regressions fail with the committed pre-fix owners: a real
30-second database idle timeout expires during a 32-second exact-source
check, and a stamp-valid invalid archive stops the pass. The fixed owner
keeps its transaction live and copies the next valid source. Source
hashes and automatic file restoration were recorded for each leg.
- Final-head strict server TypeScript and module-boundary checks pass
after the master rebase and required Runner/shared declaration rebuild.
- No source migration, provider mutation, or archive apply was executed.
## Risks
This path does not reconstruct missing backups or prove that missing
provider state is recoverable. Such sources remain blockers. It does not
take expired owners or report failed work as successful. Source hashing
adds bounded disk work while source row locks are held; unrelated
executions remain active in the PostgreSQL fixture. Existing archive
size limits, authenticated encryption, object readback, and publication
fencing remain in place. No schema, API route, scheduler, or
rollout-policy change is included.
## Model Used
OpenAI Codex (GPT-6 family; exact runtime model ID and context capacity
are not exposed in this session), with reasoning, repository editing,
code execution, and tool use.
## 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
- [ ] All Paperclip CI gates are green
- [ ] 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
ece09c156a
commit
d1306f338b
6 files changed
+355
-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,29 @@ 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 && ["native_harness_preservation_source_unsettled",
|
||||
"native_harness_archive_invalid"].includes(error.message)
|
||||
|| ((error as { code?: string; cause?: { code?: string } })?.code
|
||||
?? (error as { cause?: { code?: string } })?.cause?.code) === "55P03") {
|
||||
return deferSource(source.scopeDigest, error instanceof Error ? error.message : "native_harness_source_busy", "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); }
|
||||
}
|
||||
|
||||
|
||||
+176
@@ -0,0 +1,176 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { cp, mkdir, readFile, symlink, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agentTaskSessions, createDb, 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";
|
||||
import * as backupStamp from "./native-harness-backup-stamp.js";
|
||||
import { nativeHarnessPrestageDatabaseUrl } from "../../maintenance/native-harness-prestage-runtime.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(options: { base?: string; scopeId?: string; unsafeLink?: boolean } = {}) {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "51".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const base = options.base ?? join(runtimeRoot, randomUUID());
|
||||
if (!options.base) { homes.push(base); await mkdir(base); }
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", base);
|
||||
const sessionScopeId = options.scopeId ?? 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 });
|
||||
if (options.unsafeLink) await symlink("/outside-owned-backup", join(root, "codex-home", "outside-link"));
|
||||
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, base, 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();
|
||||
});
|
||||
it("keeps the command transaction alive while exact payload verification exceeds thirty seconds", async () => {
|
||||
const f = await fixture(), db = createDb(nativeHarnessPrestageDatabaseUrl(environment.current.temporary.connectionString));
|
||||
const original = backupStamp.verifyNativeHarnessBackupStampAsync;
|
||||
let slowed = false;
|
||||
const spy = vi.spyOn(backupStamp, "verifyNativeHarnessBackupStampAsync").mockImplementation(async (...args) => {
|
||||
if (args[3] !== undefined && !slowed) { slowed = true; await new Promise(resolve => setTimeout(resolve, 32_000)); }
|
||||
return original(...args);
|
||||
});
|
||||
try {
|
||||
expect((await db.$client.unsafe("show idle_in_transaction_session_timeout"))[0].idle_in_transaction_session_timeout).toBe("30s");
|
||||
expect(await prestageNativeHarnessBackups(db, { storage: f.storage, signal: AbortSignal.timeout(60_000) }))
|
||||
.toMatchObject({ published: 1, deferred: {} });
|
||||
expect(slowed).toBe(true);
|
||||
} finally { spy.mockRestore(); await db.$client.end({ timeout: 1 }); }
|
||||
}, 60_000);
|
||||
it("defers a stamped archive-invalid source and preserves the next eligible source", async () => {
|
||||
const bad = await fixture({ scopeId: "b", unsafeLink: true });
|
||||
const good = await fixture({ base: bad.base, scopeId: "a" });
|
||||
expect(bad.scopeDigest < good.scopeDigest).toBe(true);
|
||||
expect(await prestageNativeHarnessBackups(bad.db, { storage: bad.storage, signal: AbortSignal.timeout(15_000) }))
|
||||
.toMatchObject({ inspected: 2, published: 1, deferred: { source_changed: 1 } });
|
||||
expect(await bad.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, bad.scopeDigest))).toEqual([]);
|
||||
expect((await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest)))[0].status).toBe("current");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,112 @@
|
||||
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";
|
||||
import { withWorkspaceLockKeepalive } from "./native-workspace-lock-keepalive.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 [settings] = await db.execute<{ idle_timeout_ms: string }>(sql`select
|
||||
extract(epoch from current_setting('idle_in_transaction_session_timeout')::interval) * 1000 as idle_timeout_ms`);
|
||||
await withWorkspaceLockKeepalive(db, Number(settings.idle_timeout_ms), async () => {
|
||||
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