Integrate expired-backup retry and consumer contract fixes

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

@@ -868,6 +868,30 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     expect(await exists(own.current)).toBe(false); expect(await taskSessionCount(own.agentId)).toBe(0); await quiet();
   });
 
+  it.each(["EIO", "EACCES"])("retains expired-only cleanup after a directory read failure (%s)", async code => {
+    const session = await seedSession();
+    const backupRoot = path.dirname(session.current);
+    const expired = path.join(backupRoot, `.expired-${randomUUID()}`);
+    await fs.rename(session.current, expired);
+    const read = fs.readdir;
+    const fault = vi.spyOn(fs, "readdir").mockImplementation(async (...args) => {
+      if (String(args[0]) === backupRoot) throw Object.assign(new Error("controlled cleanup listing failure"), { code });
+      return read(...args);
+    });
+    await startBackups();
+    expect(backupErrors).toHaveLength(1);
+    const retry = coordinator!.nextWakeAt();
+    expect(retry).not.toBeNull();
+    expect(retry! - Date.now()).toBeGreaterThanOrEqual(29_000);
+    expect(await exists(session.current)).toBe(false);
+    expect(await exists(expired)).toBe(true);
+    fault.mockRestore();
+    await quiet(5000);
+    await vi.advanceTimersByTimeAsync(retry! - Date.now()); await settled();
+    expect(await fs.readdir(backupRoot)).toEqual([]);
+    await quiet();
+  });
+
   it("reconstructs future retention without issue row locks", async () => {
     await seedSession({ closedDaysAgo: 2 }); const queries: string[] = [];
     const previous = db.$client.options.debug;

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

@@ -352,7 +352,7 @@ export function nativeHarnessBackupExpirationService(
 
   async function settleLeftovers(backupRoot: string) {
     const current = path.join(backupRoot, "current");
-    for (const name of await fs.readdir(backupRoot).catch(() => [] as string[])) {
+    for (const name of (await fs.readdir(backupRoot).catch(missingPath)) ?? []) {
       const target = path.join(backupRoot, name);
       if (name.startsWith(RETIRING_PREFIX) && UUID_PATTERN.test(name.slice(RETIRING_PREFIX.length))) {
         if (!(await isRealDirectory(current))) await fs.rename(target, current);

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

@@ -5103,7 +5103,10 @@ describe("retained native cleanup activation", () => {
       }
       const outcome = await pendingOutcome;
       if (mode === "legacy_busy_copy") {
-        expect(maintenanceReleased).toHaveBeenCalledExactlyOnceWith({ companyId: run.companyId, runId: run.id });
+        expect(maintenanceReleased).toHaveBeenCalledExactlyOnceWith(
+          { companyId: run.companyId, runId: run.id },
+          { companyId: run.companyId, runId: run.id },
+        );
         unsubscribeMaintenance?.();
       }
       if (canonicalSource) {
@@ -6425,7 +6428,10 @@ describe("native startup cancellation fence", () => {
       })).rejects.toThrow("native_cancellation_pending_recovery");
       expect(cancel).toHaveBeenCalledOnce();
       expect(submit).not.toHaveBeenCalled();
-      expect(released).toHaveBeenCalledWith({ companyId: execution.binding.companyId, agentId: execution.binding.agentId });
+      expect(released).toHaveBeenCalledWith(
+        { companyId: execution.binding.companyId, agentId: execution.binding.agentId },
+        { companyId: execution.binding.companyId, runId: execution.binding.runId, issueId: execution.binding.issueId },
+      );
     } finally { unsubscribe(); }
   });
 });
@@ -9872,9 +9878,10 @@ describe("native warm session supervision", () => {
       );
       expect(released).not.toHaveBeenCalled();
       releaseClose();
-      await vi.waitFor(() => expect(released).toHaveBeenCalledExactlyOnceWith({
-        companyId: base.binding.companyId, agentId: base.binding.agentId,
-      }));
+      await vi.waitFor(() => expect(released).toHaveBeenCalledExactlyOnceWith(
+        { companyId: base.binding.companyId, agentId: base.binding.agentId },
+        { companyId: base.binding.companyId, runId: base.binding.runId, issueId: base.binding.issueId },
+      ));
     } finally { releaseClose(); unsubscribe(); }
   });
 
@@ -12453,7 +12460,10 @@ describe("runnerd provider runtime wiring", () => {
       releaseEnvironment();
       releaseAssigned();
       const result = await preparation;
-      expect(released).toHaveBeenCalledExactlyOnceWith({ companyId: execution.binding.companyId, agentId: execution.binding.agentId });
+      expect(released).toHaveBeenCalledExactlyOnceWith(
+        { companyId: execution.binding.companyId, agentId: execution.binding.agentId },
+        { companyId: execution.binding.companyId, runId: execution.binding.runId, issueId: execution.binding.issueId },
+      );
       if (failure !== "success") {
         expect(result).toMatchObject({ error: { message: failure === "assigned" ? "assigned failed" : "environment failed" } });
         expect(state.createBackend).not.toHaveBeenCalled();