fix(server): isolate backup discovery and recover orphan cleanup

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

doc/plans/2026-10-30-backup-expiration-sleep.md

@@ -14,7 +14,9 @@ must invalidate the source when it changes. A busy process or failed I/O leaves
 real unresolved cleanup and uses the shared retry backoff. Dry-run reports an
 eligible source once and waits for source invalidation or restart.
 
-Startup pages existing disk scopes. Roots without backup or cleanup tombstones
+Startup pages existing disk scopes and indexes company ownership, including
+protected backups. Company notifications inspect only that index; local backup
+publication adds newly created roots. Roots without backup or cleanup tombstones
 perform no backup SQL. After discovery and all obligations settle, there is no
 recurring SQL or filesystem scan. Startup reconstructs future work; inspecting a
 source never changes its business timestamps. Runtime task close/reopen and
@@ -32,7 +34,9 @@ no permanent database listener and does not add a polling table or migration.
 The existing issue share locks now remain held through the retirement rename.
 This fences task reopen while the backup becomes unavailable. A failed database
 transaction restores `.retiring-*`; a process interruption is recovered from
-that existing file journal at startup. After database settlement, `.expired-*`
+that existing file journal at startup. Journal reconciliation runs before
+identity or owner lookup, so an agent purge cannot strand expired physical
+cleanup or prevent restoration of a retiring backup. After database settlement, `.expired-*`
 records outstanding physical removal so a partial delete retries idempotently.
 Ownership, session identity, live run, lease and local execution safety checks
 remain in the backup owner. No host-sleep blocker or expiry-on-use rule changes.

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

@@ -1,3 +1,4 @@
+import { agentService } from "../services/agents.js";
 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 } from "../services/native-harness-backup-expiration-notifications.js";
@@ -711,6 +712,64 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     await quiet();
   });
 
+  it("company-only notifications inspect only the indexed company's blocked backup sources", async () => {
+    const own = await seedSession({ issueStatus: "todo" });
+    const other = await seedSession({ issueStatus: "todo" });
+    await startBackups({ maxEntries: 1 }); await settled(); await quiet();
+    const sql: Array<{ query: string; parameters: unknown[] }> = [];
+    const previous = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query, parameters) => { sql.push({ query, parameters }); };
+    const directories = vi.spyOn(fs, "readdir");
+    try {
+      const closed = await issueService(db).update(own.issueId, { status: "done" });
+      await settled(new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1);
+      const unrelated = [other.companyId, other.agentId, other.runId, other.sessionId, other.issueId];
+      expect(sql.filter(query => query.parameters.some(parameter =>
+        unrelated.some(id => String(parameter).includes(id))))).toEqual([]);
+      expect(directories.mock.calls.some(call => call[0] === stateBase
+        || String(call[0]).startsWith(other.root))).toBe(false);
+      expect(await exists(other.current)).toBe(true);
+    } finally { db.$client.options.debug = previous; directories.mockRestore(); }
+    await quiet();
+  });
+
+  it.each([false, true])("finishes cleanup journals after ordinary agent deletion (restart %s)", async restart => {
+    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 orphan cleanup failure");
+      return remove(...args);
+    });
+    await startBackups();
+    expect(backupErrors).toHaveLength(1);
+    const retryAt = coordinator!.nextWakeAt()!;
+    const backupRoot = path.dirname(session.current);
+    expect((await fs.readdir(backupRoot)).some(name => name.startsWith(".expired-"))).toBe(true);
+    if (restart) { await backupWorker!.stop(); await coordinator!.stop(); }
+    await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }).where(eq(agents.id, session.agentId));
+    await agentService(db).remove(session.agentId);
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, session.runId))).toEqual([]);
+    fault.mockRestore();
+    if (restart) await startBackups(); else await vi.advanceTimersByTimeAsync(retryAt - Date.now());
+    await settled();
+    expect(await fs.readdir(backupRoot)).toEqual([]); expect(await taskSessionCount(session.agentId)).toBe(0);
+    await quiet();
+  });
+
+  it("restores an orphan retiring journal without expiring the recovered backup or unrelated names", async () => {
+    const session = await seedSession(); const backupRoot = path.dirname(session.current);
+    const retiring = path.join(backupRoot, `.retiring-${randomUUID()}`);
+    await fs.rename(session.current, retiring);
+    await fs.mkdir(path.join(backupRoot, ".expired-unrecognized"));
+    await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }).where(eq(agents.id, session.agentId));
+    await agentService(db).remove(session.agentId);
+    await fs.rm(path.join(session.root, "control-plane"), { recursive: true });
+    await startBackups(); await settled();
+    expect(await exists(session.current)).toBe(true); expect(await exists(retiring)).toBe(false);
+    expect(await exists(path.join(session.current, "manifest.json"))).toBe(true);
+    expect(await exists(path.join(backupRoot, ".expired-unrecognized"))).toBe(true);
+    await quiet();
+  });
+
   it("retains live lease references and only removes the owning company's backup after lease release", async () => {
     const first = await seedSession({ leaseStatus: "retained" });
     const other = await seedSession({ leaseStatus: "retained" });

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -566,6 +566,8 @@ vi.mock("../services/native-runtime/native-harness-backup-expiration.js", () =>
     tick: vi.fn(async () => ({ checked: 0, candidates: 0, evidenceCandidates: 0, failed: 0, truncated: false })),
     noteEvidence: vi.fn(), nextEvidenceAt: vi.fn(() => null),
     tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn(),
+    backupScopes: vi.fn(async () => ({ names: [], complete: true })),
+    reconcileBackup: vi.fn(async () => null), expireBackup: vi.fn(),
   })),
 }));
 
@@ -676,9 +678,12 @@ describe("startServer feedback export wiring", () => {
     loadConfigMock.mockReturnValue(buildTestConfig({ nativeHarnessBackupExpirationEnabled: enabled }));
     await startServer(); await settleStartupHousekeeping();
     expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.nativeRunnerEvidenceRetention)).toHaveLength(enabled ? 1 : 0);
-    expect(vi.mocked(nativeHarnessBackupExpirationService).mock.calls.map(([, options]) => options?.duty)).toEqual(enabled ? ["backups", "evidence"] : ["backups"]);
+    expect(vi.mocked(nativeHarnessBackupExpirationService).mock.calls.map(([, options]) => options?.duty)).toEqual(enabled ? ["backups", "backups", "evidence"] : ["backups"]);
     const backups = vi.mocked(nativeHarnessBackupExpirationService).mock.results[0]!.value;
     expect(backups.registerCheckpointExpiration).toHaveBeenCalledTimes(enabled ? 1 : 0);
+    expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.nativeHarnessBackupExpiration)).toHaveLength(enabled ? 1 : 0);
+    expect(backups.tick).not.toHaveBeenCalled();
+    expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.nativeHarnessBackupExpiration)).toEqual([]);
   });
 
   it("tracks held evidence reconstruction after HTTP readiness", async () => {
@@ -686,6 +691,9 @@ describe("startServer feedback export wiring", () => {
     const pending = Promise.withResolvers<{ truncated: boolean }>();
     vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ tick: vi.fn(), noteEvidence: vi.fn(), nextEvidenceAt: () => null,
       tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn() } as never);
+    vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({
+      backupScopes: vi.fn(async () => ({ names: [], complete: true })), reconcileBackup: vi.fn(), expireBackup: vi.fn(),
+    } as never);
     vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ tick: vi.fn(() => pending.promise), noteEvidence: vi.fn(), nextEvidenceAt: () => null,
       tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn() } as never);
     try {

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

@@ -15,6 +15,26 @@ export function registerNativeHarnessBackupExpirationWork(
 ) {
   const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "backups" });
   const cursors = new Map<string, string>();
+  const roots = new Map<string, Set<string>>();
+  const owners = new Map<string, string>();
+  function remember(id: string, companyId?: string) {
+    const previous = owners.get(id);
+    if (previous && previous !== companyId) {
+      const companyRoots = roots.get(previous)!;
+      companyRoots.delete(id);
+      if (!companyRoots.size) roots.delete(previous);
+      owners.delete(id);
+    }
+    if (!companyId) return;
+    let companyRoots = roots.get(companyId);
+    if (!companyRoots) { companyRoots = new Set(); roots.set(companyId, companyRoots); }
+    companyRoots.add(id); owners.set(id, companyId);
+  }
+  async function reconcile(id: string): Promise<Entry | null> {
+    const source = await service.reconcileBackup(id);
+    remember(id, source?.companyId);
+    return source && source.dueAt !== null ? { ...source, dueAt: source.dueAt } : null;
+  }
   const worker = registerScopedDeadlineWork(coordinator, {
     queue: DELIVERY_QUEUES.nativeHarnessBackupExpiration, batchSize: options.maxRemovals ?? 25,
     decodeScope: (hint: string): NativeHarnessBackupExpirationScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
@@ -22,12 +42,16 @@ export function registerNativeHarnessBackupExpirationWork(
       && (!scope.scopeDigest || entry.id === scope.scopeDigest),
     load: async (scope, signal) => {
       const key = scope ? JSON.stringify(scope) : "*";
+      const limit = options.maxEntries ?? 100;
+      const indexed = scope ? [...roots.get(scope.companyId) ?? []]
+        .filter(id => !cursors.get(key) || id > cursors.get(key)!).sort() : null;
       const page = scope?.scopeDigest ? { names: [scope.scopeDigest], complete: true }
-        : await service.backupScopes(cursors.get(key), options.maxEntries ?? 100);
+        : indexed ? { names: indexed.slice(0, limit), complete: indexed.length <= limit }
+        : await service.backupScopes(cursors.get(key), limit);
       const rows: Entry[] = [];
       for (const name of page.names) {
         if (signal?.aborted || !canRun()) throw new Error("Backup expiration discovery paused");
-        const entry = await service.inspectBackup(name);
+        const entry = await reconcile(name);
         if (entry && (!scope || entry.companyId === scope.companyId)) rows.push(entry);
       }
       if (page.complete) cursors.delete(key); else cursors.set(key, page.names.at(-1)!);
@@ -35,19 +59,22 @@ export function registerNativeHarnessBackupExpirationWork(
     },
     check: async (entry, signal) => {
       if (signal?.aborted || !canRun()) throw new Error("Backup expiration paused");
-      const current = await service.inspectBackup(entry.id);
+      const current = await reconcile(entry.id);
       if (!current || current.companyId !== entry.companyId) return null;
       if (current.dueAt > Date.now()) return current;
       const result = await service.expireBackup(entry.id);
       if (result.candidates || result.failed) logger.info({ ...result, dryRun: options.dryRun ?? false }, "native harness backup expiration");
       if (options.dryRun) return null;
-      const remaining = await service.inspectBackup(entry.id);
+      const remaining = await reconcile(entry.id);
       if (!remaining) return null;
       if (result.skipped.session_active) throw new Error("Backup session is active");
       return remaining;
     },
   }, canRun);
-  const unsubscribe = onNativeHarnessBackupChange(db, scope => worker.invalidate(JSON.stringify(scope)));
+  const unsubscribe = onNativeHarnessBackupChange(db, scope => {
+    if (scope.scopeDigest) remember(scope.scopeDigest, scope.companyId);
+    worker.invalidate(JSON.stringify(scope));
+  });
   const release = onRetainedNativeSessionCleanupRelease(db, scope => worker.invalidate(JSON.stringify({ companyId: scope.companyId })));
   return { ...worker, async stop() { unsubscribe(); release(); await worker.stop(); } };
 }

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

@@ -739,30 +739,29 @@ export function nativeHarnessBackupExpirationService(
     });
   }
 
-  async function inspectBackup(scopeDigest: string) {
+  async function reconcileBackup(scopeDigest: string) {
     const base = path.resolve(defaults.stateBase ?? nativeHarnessStateBase());
     const root = path.join(base, scopeDigest);
     if (!SCOPE_PATTERN.test(scopeDigest) || !(await isRealDirectory(root))) return null;
     const backupRoot = path.join(root, "failover-backups");
     if (!(await isRealDirectory(backupRoot))) return null;
-    const names = await fs.readdir(backupRoot);
+    if (!defaults.dryRun) await settleLeftovers(backupRoot);
     const current = path.join(backupRoot, "current");
     const stat = await fs.lstat(current).catch(missingPath);
-    const leftovers = names.some(name => [RETIRING_PREFIX, EXPIRED_PREFIX].some(prefix =>
-      name.startsWith(prefix) && UUID_PATTERN.test(name.slice(prefix.length))));
-    if ((!stat?.isDirectory() || stat.isSymbolicLink()) && !leftovers) return null;
-    if (defaults.dryRun && !stat?.isDirectory()) return null;
+    if (!stat?.isDirectory() || stat.isSymbolicLink()) return null;
     const identity = await readSessionIdentity(root);
     if (!identity) return null;
     const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId })
       .from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
     if (!owner) return null;
-    if (leftovers && !defaults.dryRun) return { id: scopeDigest, ...owner, dueAt: 0 };
-    if (await fs.lstat(path.join(root, "cleanup-activation.json")).then(() => true, () => false)) return null;
+    const source = { id: scopeDigest, ...owner };
+    if (await fs.lstat(path.join(root, "cleanup-activation.json")).then(() => true, () => false)) {
+      return { ...source, dueAt: null };
+    }
     const blocker = await db.transaction(tx => sessionRetentionBlocker(tx, {
       scopeDigest, identity, closedBefore: new Date(8_000_000_000_000_000),
     }));
-    if ("reason" in blocker) return null;
+    if ("reason" in blocker) return { ...source, dueAt: null };
     const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS;
     return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 };
   }
@@ -789,5 +788,5 @@ export function nativeHarnessBackupExpirationService(
   }
 
   return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration,
-    inspectBackup, backupScopes, expireBackup };
+    reconcileBackup, backupScopes, expireBackup };
 }