fix(runtime): recheck successful source state before pre-copy publication

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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"

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<Parameters<Db["transaction"]>[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<unknown>`${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<Parameters<Db["transaction"]>[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");
   }
 }