test(server): qualify native backup publication and retention recovery

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

server/src/__tests__/native-harness-backup-expiration.test.ts

@@ -1,5 +1,6 @@
+import { publishNativeHarnessBackup } from "../services/native-runtime/native-harness-backup-publication.js";
 import { registerNativeHarnessBackupExpirationWork } from "../services/native-runtime/native-harness-backup-expiration-work.js";
-import { noteNativeHarnessBackupChange, notifyNativeHarnessBackupExpiration } from "../services/native-harness-backup-expiration-notifications.js";
+import { noteNativeHarnessBackupChange } from "../services/native-harness-backup-expiration-notifications.js";
 import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js";
 import { issueService } from "../services/issues.js";
 import { applyPendingMigrations } from "@paperclipai/db";
@@ -583,6 +584,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     if (history) {
       const session = await seedSession();
       await fs.rm(path.join(session.root, "failover-backups"), { recursive: true });
+      await db.insert(heartbeatRuns).values(Array.from({ length: 500 }, () => ({
+        companyId: session.companyId, agentId: session.agentId, status: "succeeded", issueId: session.issueId,
+        nativeIssueId: session.issueId, nativeSessionId: randomUUID(), finishedAt: now,
+      })));
       for (let index = 0; index < 250; index++) await fs.mkdir(path.join(stateBase,
         createHash("sha256").update(String(index)).digest("hex")));
     }
@@ -609,6 +614,68 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     expect(backupErrors).toEqual([]); await quiet();
   });
 
+  it("ordinary backup publication after quiet schedules its actual file age and leaves a durable cleanup result", async () => {
+    const session = await seedSession();
+    const backupRoot = path.dirname(session.current);
+    const pendingRoot = path.join(backupRoot, "pending-publication");
+    await fs.rename(session.current, pendingRoot);
+    await startBackups(); await settled(); await quiet();
+    const publishedAt = Date.now();
+    await fs.utimes(pendingRoot, new Date(publishedAt), new Date(publishedAt));
+    await publishNativeHarnessBackup({ db, companyId: session.companyId, scopeDigest: session.scope, backupRoot, pendingRoot });
+    const stat = await fs.stat(session.current);
+    const due = stat.mtime.getTime() + 7 * DAY_MS + 1;
+    await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled();
+    expect(await exists(session.current)).toBe(false); expect(await taskSessionCount(session.agentId)).toBe(0);
+    expect(await fs.readdir(backupRoot)).toEqual([]); await quiet();
+  });
+
+  it("a committed earlier close timestamp replaces the later deadline without early cleanup", async () => {
+    const session = await seedSession({ closedDaysAgo: 2 });
+    await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1);
+    const earlier = new Date(now.getTime() - 6 * DAY_MS);
+    await issueService(db).update(session.issueId, { completedAt: earlier });
+    const due = earlier.getTime() + 7 * DAY_MS + 1;
+    await settled(due); await quiet(60_000);
+    await vi.advanceTimersByTimeAsync(due - Date.now()); await settled();
+    expect(await exists(session.current)).toBe(false); await quiet();
+  });
+
+  it("holds nested ordinary close activation until the outer transaction commits", async () => {
+    const session = await seedSession({ issueStatus: "todo" });
+    await startBackups(); await settled();
+    let release!: () => void, written!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const staged = new Promise<void>(resolve => { written = resolve; });
+    const transaction = db.transaction(async outer => {
+      await outer.transaction(nested => issueService(db).update(session.issueId, { status: "done" }, nested));
+      written(); await gate;
+    });
+    await staged;
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(await exists(session.current)).toBe(true);
+    expect(await taskSessionCount(session.agentId)).toBe(1);
+    release(); await transaction;
+    const [closed] = await db.select().from(issues).where(eq(issues.id, session.issueId));
+    const due = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1;
+    await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled();
+    expect(await exists(session.current)).toBe(false); await quiet();
+  });
+
+  it("preserves failed cleanup retry across drain, then finishes its physical tombstone on resume", async () => {
+    const session = await seedSession(); const remove = fs.rm;
+    const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => {
+      if (String(args[0]).includes(".expired-")) throw new Error("controlled drain retry failure");
+      return remove(...args);
+    });
+    await startBackups();
+    expect(backupErrors).toHaveLength(1); admitted = false; fault.mockRestore();
+    await quiet(60_000);
+    expect((await fs.readdir(path.dirname(session.current))).some(name => name.startsWith(".expired-"))).toBe(true);
+    admitted = true; await settled();
+    expect(await fs.readdir(path.dirname(session.current))).toEqual([]); await quiet();
+  });
+
   it("reopen cancels a saved deadline and a later ordinary close restores retention", async () => {
     const session = await seedSession({ closedDaysAgo: 2 });
     await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1);

server/src/services/native-runtime/native-harness-backup-expiration.ts

@@ -764,7 +764,7 @@ export function nativeHarnessBackupExpirationService(
     }));
     if ("reason" in blocker) return null;
     const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS;
-    return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtimeMs, blocker.closedAt) + retentionMs + 1 };
+    return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 };
   }
 
   async function backupScopes(after?: string, limit = 100) {

server/src/services/native-runtime/native-harness-backup-publication.test.ts

@@ -0,0 +1,44 @@
+import { promises as fs } from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, it } from "vitest";
+import { onNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js";
+import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js";
+
+describe("native harness backup publication", () => {
+  let backupRoot = "";
+  let unsubscribe: (() => void) | undefined;
+  afterEach(async () => {
+    unsubscribe?.(); unsubscribe = undefined;
+    if (backupRoot) await fs.rm(backupRoot, { recursive: true, force: true });
+  });
+  async function fixture() {
+    backupRoot = await fs.mkdtemp(path.join(os.tmpdir(), "native-backup-publication-"));
+    const current = path.join(backupRoot, "current"), pendingRoot = path.join(backupRoot, "pending");
+    await fs.mkdir(current); await fs.mkdir(pendingRoot);
+    await fs.writeFile(path.join(current, "manifest.json"), "saved");
+    await fs.writeFile(path.join(pendingRoot, "manifest.json"), "replacement");
+    return { db: {}, companyId: "company", scopeDigest: "scope", backupRoot, pendingRoot, current };
+  }
+  it("restores the existing backup when its lease stamp fails without publishing a false change", async () => {
+    const input = await fixture(); const changes: unknown[] = [];
+    unsubscribe = onNativeHarnessBackupChange(input.db, change => changes.push(change));
+    await expect(publishNativeHarnessBackup({ ...input, stamp: async current => {
+      expect(await fs.readFile(path.join(current, "manifest.json"), "utf8")).toBe("replacement");
+      throw new Error("stamp transaction rejected");
+    } })).rejects.toThrow("stamp transaction rejected");
+    expect(await fs.readFile(path.join(input.current, "manifest.json"), "utf8")).toBe("saved");
+    expect(await fs.readdir(backupRoot)).toEqual(["current"]); expect(changes).toEqual([]);
+  });
+  it("publishes the completed replacement once and removes its predecessor", async () => {
+    const input = await fixture(); const changes: unknown[] = [];
+    unsubscribe = onNativeHarnessBackupChange(input.db, change => changes.push(change));
+    await publishNativeHarnessBackup({ ...input, stamp: async current => {
+      expect(changes).toEqual([]);
+      expect(await fs.readFile(path.join(current, "manifest.json"), "utf8")).toBe("replacement");
+    } });
+    expect(await fs.readFile(path.join(input.current, "manifest.json"), "utf8")).toBe("replacement");
+    expect(await fs.readdir(backupRoot)).toEqual(["current"]);
+    expect(changes).toEqual([{ companyId: input.companyId, scopeDigest: input.scopeDigest }]);
+  });
+});

server/src/services/native-runtime/native-harness-backup-publication.ts

@@ -0,0 +1,28 @@
+import { existsSync, renameSync } from "node:fs";
+import { resolve } from "node:path";
+import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js";
+import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
+
+export async function publishNativeHarnessBackup(input: {
+  db: object; companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string;
+  stamp?: (currentRoot: string) => Promise<void>;
+}) {
+  const currentRoot = resolve(input.backupRoot, "current");
+  const previousRoot = resolve(input.backupRoot, "previous");
+  removeNativeHarnessBackup(previousRoot);
+  let movedCurrent = false;
+  if (existsSync(currentRoot)) {
+    renameSync(currentRoot, previousRoot);
+    movedCurrent = true;
+  }
+  try {
+    renameSync(input.pendingRoot, currentRoot);
+    await input.stamp?.(currentRoot);
+  } catch (error) {
+    if (existsSync(currentRoot)) removeNativeHarnessBackup(currentRoot);
+    if (movedCurrent && existsSync(previousRoot) && !existsSync(currentRoot)) renameSync(previousRoot, currentRoot);
+    throw error;
+  }
+  removeNativeHarnessBackup(previousRoot);
+  noteNativeHarnessBackupChange(input.db, { companyId: input.companyId, scopeDigest: input.scopeDigest });
+}

server/src/services/native-runtime/native-session-executor.ts

@@ -1,4 +1,4 @@
-import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js";
+import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js";
 import { createNativeUsageDelivery } from "./native-usage-delivery.js";
 import { createNativeTurnAccounting, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js";
 export { nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage } from "./native-usage-accounting.js";
@@ -13277,42 +13277,15 @@ async function createRunnerdBackendWithinSessionClaim(
                 });
                 renameSync(temporaryManifest, manifestPath);
 
-                const currentRoot = resolve(backupRoot, "current");
-                const previousRoot = resolve(backupRoot, "previous");
-                removeNativeHarnessBackup(previousRoot);
-                let movedCurrent = false;
-                if (existsSync(currentRoot)) {
-                  renameSync(currentRoot, previousRoot);
-                  movedCurrent = true;
-                }
-                try {
-                  renameSync(pendingRoot, currentRoot);
-                  if (
-                    remoteTarget?.transport === "sandbox" &&
-                    remoteTarget.leaseId
-                  ) {
+                await publishNativeHarnessBackup({
+                  db: input.db, companyId: input.execution.binding.companyId,
+                  scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex"), backupRoot, pendingRoot,
+                  stamp: remoteTarget?.transport === "sandbox" && remoteTarget.leaseId ? async currentRoot => {
                     await recordHarnessBackupStampForCurrentLease({
-                      root: currentRoot,
-                      manifest,
-                      bytes: backupSpanAttributes.bytesTransferred,
+                      root: currentRoot, manifest, bytes: backupSpanAttributes.bytesTransferred,
                     });
-                  }
-                } catch (error) {
-                  if (existsSync(currentRoot)) {
-                    removeNativeHarnessBackup(currentRoot);
-                  }
-                  if (
-                    movedCurrent &&
-                    existsSync(previousRoot) &&
-                    !existsSync(currentRoot)
-                  ) {
-                    renameSync(previousRoot, currentRoot);
-                  }
-                  throw error;
-                }
-                removeNativeHarnessBackup(previousRoot);
-                noteNativeHarnessBackupChange(input.db, { companyId: input.execution.binding.companyId,
-                  scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex") });
+                  } : undefined,
+                });
               } finally {
                 removeNativeHarnessBackup(pendingRoot);
               }