diff --git a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts index 6474ed3381b..fde6df5306c 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts @@ -224,7 +224,7 @@ describeEmbeddedPostgres("durable native provider backups", () => { await expect(migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })) .rejects.toThrow("native_harness_migration_source_unsettled"); }); - it.each(["source_advanced", "queued_turn", "lease_revoked", "owner_changed"])("keeps the published source when %s happens during pre-copy upload", async change => { + it.each(["source_advanced", "queued_turn", "lease_revoked", "owner_changed", "run_failed"])("keeps the published source when %s happens during pre-copy upload", async change => { const f = await inactiveFixture(); await persistNativeHarnessBackup(f); const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); const put = f.storage.putFile.bind(f.storage); @@ -237,6 +237,7 @@ describeEmbeddedPostgres("durable native provider backups", () => { .where(eq(environmentLeases.providerLeaseId, f.providerLeaseId)); if (change === "owner_changed") await f.db.update(nativeRunFinalizations).set({ controllerGeneration: 2 }) .where(eq(nativeRunFinalizations.runId, f.runId)); + if (change === "run_failed") await f.db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId)); return result; }); await expect(persistNativeHarnessBackup(f)).rejects.toThrow(change === "owner_changed" ? "native_runner_checkpoint_owner_lost" diff --git a/server/src/services/native-runtime/native-harness-backup-storage.ts b/server/src/services/native-runtime/native-harness-backup-storage.ts index c235e620d62..cc1b8f96a04 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -6,7 +6,7 @@ import { join } from "node:path"; import os from "node:os"; import { pipeline } from "node:stream/promises"; import { Transform } from "node:stream"; -import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { and, eq, inArray, isNotNull, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import type { StorageService } from "../../storage/types.js"; @@ -152,7 +152,8 @@ async function fence(tx: Parameters[0]>[0], owner: const [binding] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, owner.authorityRunId), eq(heartbeatRuns.companyId, owner.companyId), eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, inactiveSource.normalizedSessionId), - eq(heartbeatRuns.runnerInstanceId, inactiveSource.runnerInstanceId))); + eq(heartbeatRuns.runnerInstanceId, inactiveSource.runnerInstanceId), eq(heartbeatRuns.status, "succeeded"), + isNotNull(heartbeatRuns.finishedAt))); const [lease] = await tx.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'` }) .from(environmentLeases).where(and(eq(environmentLeases.companyId, owner.companyId), eq(environmentLeases.issueId, owner.issueId), eq(environmentLeases.providerLeaseId, inactiveSource.providerLeaseId), @@ -165,8 +166,11 @@ async function fence(tx: Parameters[0]>[0], owner: // The task share lock serializes publication with execution checkout. A // turn that ran during upload also changes the immutable backup manifest. const manifest = await readFile(join(inactiveSource.root, "manifest.json")); + const identity = JSON.parse(manifest.toString("utf8")); const state = JSON.parse(await readFile(join(inactiveSource.root, "runner", "runner-state.json"), "utf8")); if (`sha256:${createHash("sha256").update(manifest).digest("hex")}` !== inactiveSource.manifestSha256 + || identity.normalizedSessionId !== stamp.normalizedSessionId || identity.runnerInstanceId !== stamp.runnerInstanceId + || identity.sourceProviderLeaseId !== stamp.sourceProviderLeaseId || identity.completedAt !== stamp.completedAt || state.runId !== owner.authorityRunId || state.lifecycle !== "suspended") fail("prestage_source_changed"); } }