fix(server): scope backup invalidation to retained session references

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

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

@@ -1,3 +1,6 @@
+import { notifyNativeEnvironmentSettlement } from "../services/native-environment-settlement-notifications.js";
+import { notifyRetainedNativeSessionCleanup, onRetainedNativeSessionCleanupRelease, releaseRetainedNativeSessionCleanup } from "../services/retained-native-session-cleanup-notifications.js";
+import { environmentService } from "../services/environments.js";
 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";
@@ -14,7 +17,7 @@ import os from "node:os";
 import path from "node:path";
 import { promises as fs } from "node:fs";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
-import { eq } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
 import {
   agents,
   agentTaskSessions,
@@ -84,6 +87,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
   });
 
   async function seedSession(input: {
+    companyId?: string; agentId?: string;
     issueStatus?: string;
     closedDaysAgo?: number;
     runStatus?: string;
@@ -93,9 +97,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     cleanupActive?: boolean;
   } = {}) {
     if (!stateBase) stateBase = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-durable-sessions-"));
-    const companyId = randomUUID();
-    await db.insert(companies).values({ id: companyId, name: "Backups", issuePrefix: `B${companyId.slice(0, 6)}` });
-    const [agent] = await db.insert(agents).values({ companyId, name: "Coder" }).returning();
+    const companyId = input.companyId ?? randomUUID();
+    if (!input.companyId) await db.insert(companies).values({ id: companyId, name: "Backups", issuePrefix: `B${companyId.slice(0, 6)}` });
+    const [agent] = input.agentId ? [{ id: input.agentId }]
+      : await db.insert(agents).values({ companyId, name: "Coder" }).returning();
     const issueStatus = input.issueStatus ?? "done";
     const closedAt = new Date(now.getTime() - (input.closedDaysAgo ?? 10) * DAY_MS);
     const [issue] = await db.insert(issues).values({
@@ -733,6 +738,153 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     await quiet();
   });
 
+  it("coalesces repeated task writes to one retained root in a large company with the same historical agent", async () => {
+    const own = await seedSession({ issueStatus: "todo" });
+    const unrelated = [];
+    for (let index = 0; index < 80; index++) unrelated.push(await seedSession({ companyId: own.companyId,
+      agentId: own.agentId, issueStatus: index % 2 ? "todo" : "done", closedDaysAgo: 2,
+      runStatus: index % 3 ? "succeeded" : "running" }));
+    const due = now.getTime() + 5 * DAY_MS + 1;
+    await startBackups({ isSessionBusy: digest => unrelated.some((entry, index) => index % 4 === 0 && entry.scope === digest) });
+    await settled(due); await quiet(60_000);
+    const reads = vi.spyOn(fs, "readFile");
+    const queries: Array<{ query: string; parameters: unknown[] }> = [];
+    const previous = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query, parameters) => { queries.push({ query, parameters }); };
+    try {
+      await db.transaction(async tx => {
+        await issueService(db).update(own.issueId, { status: "done" }, tx);
+        for (let index = 0; index < 12; index++) await issueService(db).update(own.issueId,
+          { completedAt: new Date(now.getTime() - DAY_MS) }, tx);
+      });
+      const writes = queries.length;
+      await settled(due);
+      expect(reads.mock.calls.map(call => String(call[0]))).toEqual([path.join(own.root, "control-plane", "control-plane-state.json")]);
+      expect(queries.length - writes).toBeLessThanOrEqual(20);
+      const otherIds = unrelated.flatMap(entry => [entry.issueId, entry.runId, entry.sessionId]);
+      expect(queries.filter(query => query.parameters.some(parameter => otherIds.some(id => String(parameter).includes(id))))).toEqual([]);
+    } finally { reads.mockRestore(); db.$client.options.debug = previous; }
+    await issueService(db).update(own.issueId, { status: "todo" }); await settled(due);
+    await quiet(); expect(await exists(own.current)).toBe(true);
+  });
+
+  it("indexes protected session relations so a new run, historical task close and exact process release reactivate only its backup", async () => {
+    const own = await seedSession({ closedDaysAgo: 2 });
+    const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" });
+    const [linked] = await db.insert(issues).values({ companyId: own.companyId, title: "New linked task", status: "todo" }).returning();
+    const due = now.getTime() + 5 * DAY_MS + 1;
+    await startBackups(); await settled(due); await quiet(60_000);
+    const reads = vi.spyOn(fs, "readFile");
+    const runId = randomUUID();
+    try {
+      await db.transaction(async tx => {
+        await notifyNativeEnvironmentSettlement(tx, { companyId: own.companyId, runId, issueId: linked!.id });
+        await tx.insert(heartbeatRuns).values({ id: runId, companyId: own.companyId, agentId: own.agentId,
+          nativeSessionId: own.sessionId, nativeIssueId: linked!.id, issueId: other.issueId, status: "running" });
+      });
+      await settled();
+      expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true);
+      expect(reads).toHaveBeenCalled(); reads.mockClear();
+      await issueService(db).update(linked!.id, { status: "done" }); await settled();
+      expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true); reads.mockClear();
+      await db.transaction(async tx => {
+        await notifyRetainedNativeSessionCleanup(tx, { companyId: own.companyId, runId, agentId: own.agentId });
+        await tx.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId));
+      });
+      const [closed] = await db.select().from(issues).where(eq(issues.id, linked!.id));
+      const later = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1;
+      await settled(later); reads.mockClear();
+      const cleanupScope = { companyId: own.companyId, agentId: own.agentId };
+      const observed = vi.fn(); const stop = onRetainedNativeSessionCleanupRelease(db, observed);
+      releaseRetainedNativeSessionCleanup(db, cleanupScope, { companyId: own.companyId, runId });
+      expect(observed).toHaveBeenCalledWith(cleanupScope, { companyId: own.companyId, runId }); stop();
+      await settled(later);
+      expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true);
+      expect(reads).toHaveBeenCalled();
+      await issueService(db).update(linked!.id, { status: "todo" }); await settled();
+    } finally { reads.mockRestore(); }
+    expect(await exists(other.current)).toBe(true); await quiet();
+  });
+
+  it("ordinary lease release without a run uses its source identity instead of other retained roots", async () => {
+    const own = await seedSession({ leaseStatus: "retained" });
+    const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" });
+    await startBackups(); await settled(); await quiet(60_000);
+    const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.companyId, own.companyId));
+    const reads = vi.spyOn(fs, "readFile");
+    try {
+      await environmentService(db).releaseLease(lease!.id, "expired"); await settled();
+      expect(reads).toHaveBeenCalled();
+      expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true);
+    } finally { reads.mockRestore(); }
+    expect(await exists(own.current)).toBe(false); expect(await exists(other.current)).toBe(true); await quiet();
+  });
+
+  it("retains the future deadline when a lease metadata edit removes its last backup reference", async () => {
+    const own = await seedSession({ leaseStatus: "retained", closedDaysAgo: 2 });
+    const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" });
+    await startBackups(); await settled();
+    const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.companyId, own.companyId));
+    const reads = vi.spyOn(fs, "readFile"); const due = now.getTime() + 5 * DAY_MS + 1;
+    try {
+      await environmentService(db).updateLeaseMetadata(lease!.id, {}); await settled(due);
+      expect(reads).toHaveBeenCalled(); expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true);
+    } finally { reads.mockRestore(); }
+    await vi.advanceTimersByTimeAsync(due - Date.now()); await settled();
+    expect(await exists(own.current)).toBe(false); expect(await exists(other.current)).toBe(true); await quiet();
+  });
+
+  it("allows an ordinary task reopen while backup byte counting is held outside the retirement lock", async () => {
+    const own = await seedSession(); const lstat = fs.lstat;
+    const entered = Promise.withResolvers<void>(), release = Promise.withResolvers<void>();
+    const held = vi.spyOn(fs, "lstat").mockImplementation(async (...args) => {
+      if (String(args[0]) === path.join(own.current, "codex-home", "history.jsonl")) {
+        entered.resolve(); await release.promise;
+      }
+      return lstat(...args);
+    });
+    const startup = startBackups(); await entered.promise;
+    try {
+      await db.transaction(async tx => {
+        await tx.execute(sql`set local lock_timeout = '500ms'`);
+        await issueService(db).update(own.issueId, { status: "todo" }, tx);
+      });
+    } finally { release.resolve(); await startup; held.mockRestore(); }
+    await settled(); expect(await exists(own.current)).toBe(true); expect(backupErrors).toEqual([]); await quiet();
+  });
+
+  it.each(["EIO", "EACCES"])("retains a failed identity read as real retry debt (%s)", async code => {
+    const own = await seedSession(); const read = fs.readFile;
+    const fault = vi.spyOn(fs, "readFile").mockImplementation(async (...args) => {
+      if (String(args[0]) === path.join(own.root, "control-plane", "control-plane-state.json")) {
+        throw Object.assign(new Error("controlled identity I/O failure"), { code });
+      }
+      return read(...args);
+    });
+    await startBackups(); expect(backupErrors).toHaveLength(1);
+    const retry = coordinator!.nextWakeAt()!; expect(retry - Date.now()).toBeGreaterThanOrEqual(29_000);
+    expect(await exists(own.current)).toBe(true); fault.mockRestore(); await quiet(5000);
+    await vi.advanceTimersByTimeAsync(retry - Date.now()); await settled();
+    expect(await exists(own.current)).toBe(false); expect(await taskSessionCount(own.agentId)).toBe(0); await quiet();
+  });
+
+  it("reconstructs future retention without issue row locks", async () => {
+    await seedSession({ closedDaysAgo: 2 }); const queries: string[] = [];
+    const previous = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query) => { queries.push(query); };
+    try {
+      await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1);
+      expect(queries.some(query => /for\s+share/i.test(query))).toBe(false);
+    } finally { db.$client.options.debug = previous; }
+    await quiet();
+  });
+
+  it("rejects destructive backup scopes outside the controller digest namespace", async () => {
+    await emptyStateBase(); const remove = vi.spyOn(fs, "rm");
+    await expect(service().expireBackup("../outside")).rejects.toThrow("Invalid native harness backup scope");
+    expect(remove).not.toHaveBeenCalled(); remove.mockRestore();
+  });
+
   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) => {

server/src/services/environments.ts

@@ -1701,7 +1701,7 @@ export function environmentService(db: Db) {
           cleanupStatus: environmentLeases.cleanupStatus, releasedAt: environmentLeases.releasedAt,
         }).from(environmentLeases).where(condition).for("update");
         if (!previous) return null;
-        await notifyEnvironmentCleanup(tx, { companyId: previous.companyId, runId: previous.heartbeatRunId ?? undefined });
+        await notifyEnvironmentCleanup(tx, { companyId: previous.companyId, runId: previous.heartbeatRunId ?? undefined, leaseId: id });
         if (previous.heartbeatRunId && (previous.status !== status ||
             (options?.cleanupStatus !== undefined && previous.cleanupStatus !== options.cleanupStatus) ||
             (status !== "retained" && !previous.releasedAt))) {
@@ -1821,7 +1821,7 @@ export function environmentService(db: Db) {
         const [lease] = await tx.select({ companyId: environmentLeases.companyId, heartbeatRunId: environmentLeases.heartbeatRunId, metadata: environmentLeases.metadata })
           .from(environmentLeases).where(eq(environmentLeases.id, id)).for("update");
         if (!lease) return null;
-        await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined });
+        await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined, leaseId: id });
         return tx.update(environmentLeases)
         .set({
           metadata: preserveReusableLeaseHandoff(lease.metadata, withoutCleanupVerification(metadata)),

server/src/services/issues.ts

@@ -11488,7 +11488,7 @@ export function issueService(db: Db) {
           patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number;
         }
         if (patch.status !== undefined || patch.completedAt !== undefined || patch.cancelledAt !== undefined) {
-          await notifyNativeHarnessBackupExpiration(tx, { companyId: receiptExisting.companyId });
+          await notifyNativeHarnessBackupExpiration(tx, { companyId: receiptExisting.companyId, issueId: id });
         }
         if ((patch.status !== undefined && patch.status !== receiptExisting.status)
           || patch.completedAt !== undefined || patch.cancelledAt !== undefined

server/src/services/native-environment-settlement-notifications.ts

@@ -2,9 +2,9 @@ import { notifyNativeHarnessBackupExpiration } from "./native-harness-backup-exp
 import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string };
+export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string; leaseId?: string };
 export async function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) {
-  await notifyNativeHarnessBackupExpiration(tx, scope && { companyId: scope.companyId });
+  await notifyNativeHarnessBackupExpiration(tx, scope);
   if (scope) await notifySkillVersionRetention(tx, scope.companyId);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeEnvironmentSettlement,
     scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*");

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

@@ -1,9 +1,16 @@
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export type NativeHarnessBackupExpirationScope = { companyId: string; scopeDigest?: string };
+export type NativeHarnessBackupExpirationScope = {
+  companyId: string; scopeDigest?: string; leaseId?: string; runId?: string; issueId?: string; agentId?: string; sessionId?: string;
+};
+
+export function nativeHarnessBackupScopeHint(scope: NativeHarnessBackupExpirationScope) {
+  return JSON.stringify({ companyId: scope.companyId, scopeDigest: scope.scopeDigest, leaseId: scope.leaseId,
+    runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId, sessionId: scope.sessionId });
+}
 
 export function notifyNativeHarnessBackupExpiration(tx: object, scope?: NativeHarnessBackupExpirationScope) {
-  return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeHarnessBackupExpiration, scope ? JSON.stringify(scope) : "*");
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeHarnessBackupExpiration, scope ? nativeHarnessBackupScopeHint(scope) : "*");
 }
 
 const listeners = new WeakMap<object, Set<(scope: NativeHarnessBackupExpirationScope) => void>>();

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

@@ -2,12 +2,14 @@ import type { Db } from "@paperclipai/db";
 import { logger } from "../../middleware/logger.js";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
-import { onNativeHarnessBackupChange, type NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js";
+import { nativeHarnessBackupScopeHint, onNativeHarnessBackupChange, type NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js";
 import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js";
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
-import { nativeHarnessBackupExpirationService, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js";
+import { nativeHarnessBackupExpirationService, type NativeHarnessBackupDependencies, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js";
 
-type Entry = { id: string; companyId: string; agentId: string; dueAt: number };
+type Dependencies = { companyId: string; agentId: string; sessionId: string; runIds: Set<string>; issueIds: Set<string>; leaseIds: Set<string> };
+type Entry = { id: string; companyId: string; agentId: string; dueAt: number; dependencies: Dependencies };
+type Reference = "company" | "agent" | "session" | "run" | "issue" | "lease";
 
 export function registerNativeHarnessBackupExpirationWork(
   coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, db: Db,
@@ -15,46 +17,94 @@ 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);
+  const sources = new Map<string, Dependencies>();
+  const index = new Map<string, Set<string>>();
+  const key = (companyId: string, kind: Reference, id = "") => JSON.stringify([companyId, kind, id]);
+  const sessionKey = (agentId: string, sessionId: string) => JSON.stringify([agentId, sessionId]);
+  function references(source: Dependencies) {
+    return [key(source.companyId, "company"), key(source.companyId, "agent", source.agentId),
+      key(source.companyId, "session", sessionKey(source.agentId, source.sessionId)),
+      ...[...source.runIds].map(id => key(source.companyId, "run", id)),
+      ...[...source.issueIds].map(id => key(source.companyId, "issue", id)),
+      ...[...source.leaseIds].map(id => key(source.companyId, "lease", id))];
+  }
+  function add(reference: string, id: string) {
+    let roots = index.get(reference);
+    if (!roots) { roots = new Set(); index.set(reference, roots); }
+    roots.add(id);
+  }
+  function remember(id: string, current?: NativeHarnessBackupDependencies) {
+    const previous = sources.get(id);
+    if (previous) for (const reference of references(previous)) {
+      const roots = index.get(reference)!;
+      roots.delete(id); if (!roots.size) index.delete(reference);
+    }
+    if (!current) { sources.delete(id); return undefined; }
+    const source = { companyId: current.companyId, agentId: current.agentId,
+      sessionId: current.sessionId, runIds: new Set(current.runIds), issueIds: new Set(current.issueIds), leaseIds: new Set(current.leaseIds) };
+    sources.set(id, source);
+    for (const reference of references(source)) add(reference, id);
+    return source;
+  }
+  function selected(scope: NativeHarnessBackupExpirationScope): Set<string> {
+    if (scope.scopeDigest) return new Set([scope.scopeDigest]);
+    const reference = scope.leaseId ? key(scope.companyId, "lease", scope.leaseId)
+      : scope.runId ? key(scope.companyId, "run", scope.runId)
+      : scope.sessionId && scope.agentId ? key(scope.companyId, "session", sessionKey(scope.agentId, scope.sessionId))
+      : scope.issueId ? key(scope.companyId, "issue", scope.issueId)
+      : scope.agentId ? key(scope.companyId, "agent", scope.agentId) : key(scope.companyId, "company");
+    return new Set(index.get(reference) ?? []);
+  }
+  async function discoverReferences(scope: NativeHarnessBackupExpirationScope) {
+    if (scope.scopeDigest || (!scope.runId && !scope.leaseId)) return;
+    const resolved = await service.resolveBackupScope(scope);
+    const roots = selected(scope);
+    if (resolved.scopeDigest) roots.add(resolved.scopeDigest);
+    const related = resolved.sessionId && resolved.agentId
+      ? selected({ companyId: scope.companyId, sessionId: resolved.sessionId, agentId: resolved.agentId })
+      : resolved.issueId ? selected({ companyId: scope.companyId, issueId: resolved.issueId }) : [];
+    for (const id of related) roots.add(id);
+    for (const id of roots) {
+      const source = sources.get(id);
+      if (!source || source.companyId !== scope.companyId) continue;
+      if (scope.runId) { source.runIds.add(scope.runId); add(key(scope.companyId, "run", scope.runId), id); }
+      if (scope.leaseId) { source.leaseIds.add(scope.leaseId); add(key(scope.companyId, "lease", scope.leaseId), 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 dependencies = remember(id, source?.dependencies);
+    return source && dependencies && source.dueAt !== null ? { ...source, dependencies, dueAt: source.dueAt } : null;
+  }
+  function matches(entry: Entry, scope?: NativeHarnessBackupExpirationScope) {
+    if (!scope) return true;
+    if (entry.companyId !== scope.companyId) return false;
+    if (scope.scopeDigest) return entry.id === scope.scopeDigest;
+    if (scope.leaseId) return entry.dependencies.leaseIds.has(scope.leaseId);
+    if (scope.runId) return entry.dependencies.runIds.has(scope.runId);
+    if (scope.sessionId) return entry.dependencies.sessionId === scope.sessionId
+      && (!scope.agentId || entry.agentId === scope.agentId);
+    if (scope.issueId) return entry.dependencies.issueIds.has(scope.issueId);
+    return !scope.agentId || entry.agentId === scope.agentId;
   }
   const worker = registerScopedDeadlineWork(coordinator, {
     queue: DELIVERY_QUEUES.nativeHarnessBackupExpiration, batchSize: options.maxRemovals ?? 25,
     decodeScope: (hint: string): NativeHarnessBackupExpirationScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
-    matches: (entry: Entry, scope?: NativeHarnessBackupExpirationScope) => !scope || entry.companyId === scope.companyId
-      && (!scope.scopeDigest || entry.id === scope.scopeDigest),
+    matches,
     load: async (scope, signal) => {
-      const key = scope ? JSON.stringify(scope) : "*";
+      const hint = scope ? nativeHarnessBackupScopeHint(scope) : "*";
+      if (scope && !cursors.has(hint)) await discoverReferences(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 }
-        : indexed ? { names: indexed.slice(0, limit), complete: indexed.length <= limit }
-        : await service.backupScopes(cursors.get(key), limit);
+      const indexed = scope ? [...selected(scope)].filter(id => !cursors.get(hint) || id > cursors.get(hint)!).sort() : null;
+      const page = indexed ? { names: indexed.slice(0, limit), complete: indexed.length <= limit }
+        : await service.backupScopes(cursors.get(hint), limit);
       const rows: Entry[] = [];
       for (const name of page.names) {
         if (signal?.aborted || !canRun()) throw new Error("Backup expiration discovery paused");
         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)!);
+      if (page.complete) cursors.delete(hint); else cursors.set(hint, page.names.at(-1)!);
       return { rows, complete: page.complete };
     },
     check: async (entry, signal) => {
@@ -71,10 +121,7 @@ export function registerNativeHarnessBackupExpirationWork(
       return remaining;
     },
   }, canRun);
-  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 })));
+  const unsubscribe = onNativeHarnessBackupChange(db, scope => worker.invalidate(nativeHarnessBackupScopeHint(scope)));
+  const release = onRetainedNativeSessionCleanupRelease(db, (_scope, source) => worker.invalidate(nativeHarnessBackupScopeHint(source)));
   return { ...worker, async stop() { unsubscribe(); release(); await worker.stop(); } };
 }

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

@@ -17,6 +17,7 @@ import { nativeHarnessStateBase } from "./native-harness-backup-stamp.js";
 import { nativeSessionScopeDigestIsBusy } from "./native-session-executor.js";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js";
 
 export const NATIVE_HARNESS_BACKUP_EXPIRATION_INTERVAL_MS = 60 * 60 * 1000;
 export const NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS = 2 * 60 * 1000;
@@ -146,7 +147,9 @@ async function readSessionIdentity(root: string): Promise<SessionIdentity | null
     const stat = await fs.lstat(statePath);
     if (!stat.isFile() || stat.size > MAX_CONTROL_PLANE_STATE_BYTES) return null;
     const state = JSON.parse(await fs.readFile(statePath, "utf8")) as { identity?: Record<string, unknown> };
+    if (!state || typeof state !== "object" || Array.isArray(state)) return null;
     const identity = state.identity ?? {};
+    if (!identity || typeof identity !== "object" || Array.isArray(identity)) return null;
     const runId = identity.runId;
     const normalizedSessionId = identity.normalizedSessionId;
     const environmentLeaseId = identity.environmentLeaseId;
@@ -160,16 +163,21 @@ async function readSessionIdentity(root: string): Promise<SessionIdentity | null
           ? environmentLeaseId
           : null,
     };
-  } catch {
-    return null;
+  } catch (error) {
+    if (error instanceof SyntaxError || ["ENOENT", "ENOTDIR"].includes((error as NodeJS.ErrnoException).code ?? "")) return null;
+    throw error;
   }
 }
 
+export type NativeHarnessBackupDependencies = {
+  companyId: string; agentId: string; sessionId: string; runIds: string[]; issueIds: string[]; leaseIds: string[];
+};
+
 type Tx = Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 async function sessionRetentionBlocker(
   tx: Tx,
-  input: { scopeDigest: string | null; identity: SessionIdentity; closedBefore: Date; checkpoint?: boolean },
+  input: { scopeDigest: string | null; identity: SessionIdentity; closedBefore: Date; checkpoint?: boolean; lock?: boolean },
 ): Promise<{ reason: string } | { companyId: string; agentId: string; runIds: string[]; issueIds: string[]; closedAt: number }> {
   const [owner] = await tx
     .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId })
@@ -197,7 +205,7 @@ async function sessionRetentionBlocker(
   if (input.checkpoint && runs.some(run => run.finishedAt! >= input.closedBefore)) return { reason: "recent_run" };
   const issueIds = [...new Set(runs.map((run) => run.issueId).filter((id): id is string => Boolean(id)))];
   if (issueIds.length === 0) return { reason: "issue_unknown" };
-  const issueRows = await tx
+  const issueQuery = tx
     .select({
       id: issues.id,
       status: issues.status,
@@ -205,8 +213,8 @@ async function sessionRetentionBlocker(
       closedAt: sql<Date | null>`coalesce(${issues.completedAt}, ${issues.cancelledAt})`,
     })
     .from(issues)
-    .where(and(eq(issues.companyId, owner.companyId), inArray(issues.id, issueIds)))
-    .for("share");
+    .where(and(eq(issues.companyId, owner.companyId), inArray(issues.id, issueIds)));
+  const issueRows = await (input.lock === false ? issueQuery : issueQuery.for("share"));
   if (issueRows.length !== issueIds.length) return { reason: "issue_unknown" };
   for (const issue of issueRows) {
     if (!CLOSED_ISSUE_STATUSES.includes(issue.status)) return { reason: "issue_open" };
@@ -246,7 +254,7 @@ async function sessionRetentionBlocker(
     .where(and(inArray(environmentLeases.status, LIVE_LEASE_STATUSES), or(...leaseReferences)))
     .limit(1);
   if (lease) return { reason: "live_lease" };
-  return { ...owner, runIds, issueIds, closedAt: Math.max(...issueRows.map(issue => new Date(issue.closedAt!).getTime())) };
+  return { ...owner, runIds, issueIds, closedAt: issueRows.reduce((latest, issue) => Math.max(latest, new Date(issue.closedAt!).getTime()), Number.NEGATIVE_INFINITY) };
 }
 
 async function hasLiveRun(db: Db, owner: { companyId: string; agentId: string }, sessionId: string) {
@@ -379,6 +387,7 @@ export function nativeHarnessBackupExpirationService(
     const identity = await readSessionIdentity(root);
     if (!identity) return skip("identity_unknown");
 
+    const bytes = await allocatedBytes(current);
     const id = randomUUID();
     const retiring = path.join(backupRoot, `${RETIRING_PREFIX}${id}`);
     const trash = path.join(backupRoot, `${EXPIRED_PREFIX}${id}`);
@@ -390,7 +399,7 @@ export function nativeHarnessBackupExpirationService(
           scopeDigest, identity, closedBefore: options.closedBefore,
         });
         if ("reason" in blocker) return blocker;
-        const bytes = await allocatedBytes(current);
+        if ((await fs.stat(current)).mtime >= options.closedBefore) return { reason: "recent_backup" };
         if (options.dryRun) return { cleared: 0, bytes };
         if (options.isSessionBusy(scopeDigest)) return { reason: "session_active" };
         await notifyExternalConversationSettlement(tx, { companyId: blocker.companyId });
@@ -739,6 +748,43 @@ export function nativeHarnessBackupExpirationService(
     });
   }
 
+  async function backupDependencies(scopeDigest: string, identity: SessionIdentity): Promise<NativeHarnessBackupDependencies | null> {
+    const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
+      issueId: sql<string | null>`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`,
+    }).from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
+    if (!owner) return null;
+    const runs = await db.select({ id: heartbeatRuns.id,
+      issueId: sql<string | null>`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`,
+    }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId),
+      eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.nativeSessionId, identity.normalizedSessionId)));
+    const runIds = [...new Set([identity.runId, ...runs.map(run => run.id)])];
+    const issueIds = [...new Set([owner.issueId, ...runs.map(run => run.issueId)].filter((id): id is string => Boolean(id)))];
+    const leases = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
+      eq(environmentLeases.companyId, owner.companyId), or(inArray(environmentLeases.heartbeatRunId, runIds),
+        sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'sessionScopeSha256' = ${scopeDigest}`,
+        identity.environmentLeaseId ? eq(environmentLeases.id, identity.environmentLeaseId) : undefined)));
+    return { companyId: owner.companyId, agentId: owner.agentId, sessionId: identity.normalizedSessionId,
+      runIds, issueIds, leaseIds: leases.map(lease => lease.id) };
+  }
+
+  async function resolveBackupScope(scope: NativeHarnessBackupExpirationScope) {
+    let runId = scope.runId;
+    let issueId = scope.issueId;
+    let scopeDigest = scope.scopeDigest;
+    if (scope.leaseId) {
+      const [lease] = await db.select({ runId: environmentLeases.heartbeatRunId, issueId: environmentLeases.issueId,
+        scopeDigest: sql<string | null>`${environmentLeases.metadata}->'nativeHarnessBackup'->>'sessionScopeSha256'`,
+      }).from(environmentLeases).where(and(eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId)));
+      runId = lease?.runId ?? runId; issueId = lease?.issueId ?? issueId;
+      scopeDigest = lease?.scopeDigest && SCOPE_PATTERN.test(lease.scopeDigest) ? lease.scopeDigest : scopeDigest;
+    }
+    const [run] = runId ? await db.select({ agentId: heartbeatRuns.agentId, sessionId: heartbeatRuns.nativeSessionId,
+      issueId: sql<string | null>`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`,
+    }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.id, runId))) : [];
+    return { runId, issueId: run?.issueId ?? issueId, agentId: run?.agentId ?? scope.agentId,
+      sessionId: run?.sessionId ?? scope.sessionId, scopeDigest };
+  }
+
   async function reconcileBackup(scopeDigest: string) {
     const base = path.resolve(defaults.stateBase ?? nativeHarnessStateBase());
     const root = path.join(base, scopeDigest);
@@ -751,19 +797,18 @@ export function nativeHarnessBackupExpirationService(
     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;
-    const source = { id: scopeDigest, ...owner };
+    const dependencies = await backupDependencies(scopeDigest, identity);
+    if (!dependencies) return null;
+    const source = { id: scopeDigest, companyId: dependencies.companyId, agentId: dependencies.agentId, dependencies };
     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),
+      scopeDigest, identity, closedBefore: new Date(8_000_000_000_000_000), lock: false,
     }));
     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 };
+    return { ...source, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 };
   }
 
   async function backupScopes(after?: string, limit = 100) {
@@ -776,6 +821,7 @@ export function nativeHarnessBackupExpirationService(
   }
 
   async function expireBackup(scopeDigest: string) {
+    if (!SCOPE_PATTERN.test(scopeDigest)) throw new Error("Invalid native harness backup scope");
     const result = emptyResult();
     const root = path.join(path.resolve(defaults.stateBase ?? nativeHarnessStateBase()), scopeDigest);
     const dryRun = defaults.dryRun ?? false;
@@ -788,5 +834,5 @@ export function nativeHarnessBackupExpirationService(
   }
 
   return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration,
-    reconcileBackup, backupScopes, expireBackup };
+    reconcileBackup, backupScopes, expireBackup, resolveBackupScope };
 }

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

@@ -8092,7 +8092,8 @@ export async function executePaperclipNativeSession(input: {
     }
     releaseRetainedNativeSessionCleanup(input.db, {
       companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
-    });
+    }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
+      issueId: input.execution.binding.issueId });
   }
 }
 
@@ -9474,7 +9475,8 @@ async function executePaperclipNativeSessionWithinScope(
                   warmNativeSessions.set(warmSessionId, {
                     onReleased: () => releaseRetainedNativeSessionCleanup(input.db, {
                       companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
-                    }),
+                    }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
+                      issueId: input.execution.binding.issueId }),
                     pluginTaskId: usesPluginTask && input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "sandbox"
                       ? input.runnerExecutionTarget.sandboxLeaseAcquisition?.providerLeaseId : undefined,
                     runId: input.execution.binding.runId,
@@ -11815,7 +11817,8 @@ export async function createRunnerdBackend(input: {
     initializingSessionToolAuthorities.delete(sessionScopeId);
     releaseRetainedNativeSessionCleanup(input.db, {
       companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
-    });
+    }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
+      issueId: input.execution.binding.issueId });
   }
 }
 
@@ -12860,7 +12863,7 @@ async function createRunnerdBackendWithinSessionClaim(
       })
       .where(eq(environmentLeases.id, leaseId))
       .returning({ id: environmentLeases.id })
-      .then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined });
+      .then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined, leaseId });
     if (!updated) throw new Error("runner_harness_backup_lease_missing");
   };
 

server/src/services/retained-native-session-cleanup-notifications.ts

@@ -10,15 +10,15 @@ export type RetainedNativeSessionCleanupScope = {
 };
 
 export async function notifyRetainedNativeSessionCleanup(tx: object, scope: RetainedNativeSessionCleanupScope) {
-  await notifyNativeHarnessBackupExpiration(tx, { companyId: scope.companyId });
+  await notifyNativeHarnessBackupExpiration(tx, scope);
   await notifySkillVersionRetention(tx, scope.companyId);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.retainedNativeSessionCleanup, JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId }));
 }
 
 
 
-const listeners = new WeakMap<object, Set<(scope: RetainedNativeSessionCleanupScope) => void>>();
-export function onRetainedNativeSessionCleanupRelease(db: object, listener: (scope: RetainedNativeSessionCleanupScope) => void) {
+const listeners = new WeakMap<object, Set<(scope: RetainedNativeSessionCleanupScope, source: RetainedNativeSessionCleanupScope) => void>>();
+export function onRetainedNativeSessionCleanupRelease(db: object, listener: (scope: RetainedNativeSessionCleanupScope, source: RetainedNativeSessionCleanupScope) => void) {
   let owned = listeners.get(db);
   if (!owned) { owned = new Set(); listeners.set(db, owned); }
   owned.add(listener);
@@ -27,6 +27,7 @@ export function onRetainedNativeSessionCleanupRelease(db: object, listener: (sco
     if (!owned.size && listeners.get(db) === owned) listeners.delete(db);
   };
 }
-export function releaseRetainedNativeSessionCleanup(db: object, scope: RetainedNativeSessionCleanupScope) {
-  for (const listener of listeners.get(db) ?? []) listener(scope);
+export function releaseRetainedNativeSessionCleanup(db: object, scope: RetainedNativeSessionCleanupScope,
+  source: RetainedNativeSessionCleanupScope = scope) {
+  for (const listener of listeners.get(db) ?? []) listener(scope, source);
 }