feat(lifecycle): sys_file orphan reclaim — reap guards + attachment tombstone hooks (#2755)

Part 1 of #2755: deleting a sys_attachment join row orphaned the backing
sys_file row and its storage bytes forever.

- objectql: LifecycleService reap guards (registerReapGuard) — a guarded
  object is never blind-deleted; candidates are fetched in batches, the
  guard confirms (after external cleanup) or vetoes each row, an erroring
  guard fails safe, and engines without find skip guarded objects.
- service-storage: sys_file gains the 'attachments' scope option, a
  deleted_at tombstone field, and an ADR-0057 lifecycle declaration
  (ttl 30d on deleted_at for orphans; retention 7d onlyWhen pending for
  abandoned uploads). Tombstone hooks on sys_attachment mark a file
  deleted when its last join row goes away (attachments scope only) and
  un-tombstone on re-attach. The sys_file reap guard re-verifies zero
  references at sweep time and deletes the storage bytes before
  confirming the row reap.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0187NT3Qer9oep5dCRb9b8Lt

Author: claude

packages/objectql/src/index.ts

@@ -63,6 +63,7 @@ export type {
   LifecycleObjectLike,
   LifecycleSettingsLike,
   LifecycleGovernanceAlert,
+  LifecycleReapGuard,
 } from './lifecycle/lifecycle-service.js';
 export { parseLifecycleDuration } from './lifecycle/duration.js';
 export { lifecycleSettingsManifest } from './lifecycle/lifecycle-settings.js';

packages/objectql/src/lifecycle/lifecycle-service.test.ts

@@ -17,10 +17,13 @@ function captureEngine(
     deleteImpl?: (object: string, options: any) => any;
     driver?: Record<string, unknown>;
     datasources?: Record<string, unknown>;
+    /** When set, the engine exposes `find` (guarded-reap candidate reads). */
+    findImpl?: (object: string, options: any) => any;
   } = {},
 ) {
   const deletes: Array<{ object: string; where: any; multi: any; context: any }> = [];
-  const engine = {
+  const finds: Array<{ object: string; where: any; limit: any; context: any }> = [];
+  const engine: any = {
     registry: { getAllObjects: () => objects },
     async delete(object: string, options: any) {
       deletes.push({ object, where: options?.where, multi: options?.multi, context: options?.context });
@@ -33,7 +36,13 @@ function captureEngine(
       return ds;
     },
   };
-  return { engine, deletes };
+  if (opts.findImpl) {
+    engine.find = async (object: string, options: any) => {
+      finds.push({ object, where: options?.where, limit: options?.limit, context: options?.context });
+      return opts.findImpl!(object, options);
+    };
+  }
+  return { engine, deletes, finds };
 }
 
 function service(engine: any, extra: Partial<ConstructorParameters<typeof LifecycleService>[0]> = {}) {
@@ -307,6 +316,107 @@ describe('LifecycleService.sweep — Reaper', () => {
   });
 });
 
+describe('LifecycleService.sweep — reap guard', () => {
+  const guarded: LifecycleObjectLike[] = [
+    { name: 'sys_file', lifecycle: { class: 'transient', ttl: { field: 'deleted_at', expireAfter: '30d' } } },
+  ];
+
+  it('deletes only guard-confirmed ids, by $in, after fetching candidates with the cutoff filter', async () => {
+    const rows = [
+      { id: 'f1', deleted_at: '2020-01-01T00:00:00Z' },
+      { id: 'f2', deleted_at: '2020-01-02T00:00:00Z' },
+      { id: 'f3', deleted_at: '2020-01-03T00:00:00Z' },
+    ];
+    const { engine, deletes, finds } = captureEngine(guarded, { findImpl: () => rows });
+    const svc = service(engine);
+    const guard = vi.fn(async () => ['f1', 'f3']); // vetoes f2
+    svc.registerReapGuard('sys_file', guard);
+
+    const report = await svc.sweep();
+
+    expect(finds).toHaveLength(1);
+    expect(finds[0].where).toEqual({ deleted_at: { $lt: isoCutoff('30d') } });
+    expect(finds[0].context).toEqual({ isSystem: true, positions: [], permissions: [] });
+    expect(guard).toHaveBeenCalledWith('sys_file', rows);
+    expect(deletes).toHaveLength(1);
+    expect(deletes[0].where).toEqual({ id: { $in: ['f1', 'f3'] } });
+    expect(deletes[0].multi).toBe(true);
+    expect(report.swept).toEqual([
+      { object: 'sys_file', class: 'transient', policy: 'ttl', cutoff: isoCutoff('30d'), deleted: 2 },
+    ]);
+    expect(report.errors).toEqual([]);
+  });
+
+  it('an erroring guard fails safe: no rows deleted, error reported', async () => {
+    const { engine, deletes } = captureEngine(guarded, { findImpl: () => [{ id: 'f1' }] });
+    const svc = service(engine);
+    svc.registerReapGuard('sys_file', async () => {
+      throw new Error('storage unreachable');
+    });
+
+    const report = await svc.sweep();
+
+    expect(deletes).toHaveLength(0);
+    expect(report.swept).toEqual([]);
+    expect(report.errors).toEqual([{ object: 'sys_file', error: 'storage unreachable' }]);
+  });
+
+  it('a guard on one object never changes the blind reap of others (regression pin)', async () => {
+    const { engine, deletes } = captureEngine(
+      [
+        ...guarded,
+        { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } },
+      ],
+      { findImpl: () => [] },
+    );
+    const svc = service(engine);
+    svc.registerReapGuard('sys_file', async () => []);
+
+    const report = await svc.sweep();
+
+    // sys_file: no candidates → no delete. sys_job_run: classic blind reap.
+    expect(deletes).toHaveLength(1);
+    expect(deletes[0].object).toBe('sys_job_run');
+    expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('30d') } });
+    expect(report.errors).toEqual([]);
+  });
+
+  it('a guarded object on an engine without find is skipped, never blind-deleted', async () => {
+    const { engine, deletes } = captureEngine(guarded); // no findImpl → no engine.find
+    const svc = service(engine);
+    svc.registerReapGuard('sys_file', async () => ['f1']);
+
+    const report = await svc.sweep();
+
+    expect(deletes).toHaveLength(0);
+    expect(report.swept).toEqual([]);
+    expect(report.skipped).toEqual([{ object: 'sys_file', reason: 'reap-guard-unsupported' }]);
+  });
+
+  it('drains full batches but stops the pass when a batch is not fully confirmed', async () => {
+    // Two "pages" of 500, then a short page. All of page 1 confirmed → loop
+    // continues; page 2 only partially confirmed → pass ends (vetoed rows
+    // would be re-fetched forever within one sweep).
+    const page = (n: number, prefix: string) =>
+      Array.from({ length: n }, (_, i) => ({ id: `${prefix}${i}`, deleted_at: '2020-01-01T00:00:00Z' }));
+    const pages = [page(500, 'a'), page(500, 'b')];
+    let call = 0;
+    const { engine, deletes, finds } = captureEngine(guarded, { findImpl: () => pages[call++] ?? [] });
+    const svc = service(engine);
+    svc.registerReapGuard('sys_file', async (_object, rows) =>
+      rows[0].id === 'a0' ? rows.map((r: any) => r.id) : rows.slice(0, 10).map((r: any) => r.id),
+    );
+
+    const report = await svc.sweep();
+
+    expect(finds).toHaveLength(2);
+    expect(deletes).toHaveLength(2);
+    expect((deletes[0].where.id.$in as string[]).length).toBe(500);
+    expect((deletes[1].where.id.$in as string[]).length).toBe(10);
+    expect(report.swept[0].deleted).toBe(510);
+  });
+});
+
 describe('LifecycleService.sweep — Archiver (P3)', () => {
   const AUDIT_OBJ: LifecycleObjectLike = {
     name: 'sys_audit_log',

packages/objectql/src/lifecycle/lifecycle-service.ts

@@ -200,6 +200,28 @@ interface ArchiveCapableDriver {
 const ARCHIVE_BATCH_SIZE = 500;
 const ARCHIVE_MAX_BATCHES_PER_SWEEP = 20;
 
+/** Guarded reap batching — same posture as the Archiver: bound one sweep's
+ * work, drain the backlog across sweeps. */
+const REAP_GUARD_BATCH_SIZE = 500;
+const REAP_GUARD_MAX_BATCHES_PER_SWEEP = 20;
+
+/**
+ * Reap guard (ADR-0057 amendment): a domain callback consulted by the Reaper
+ * before rows of the guarded object are deleted. The guard receives the
+ * candidate rows and returns the ids it CONFIRMS for deletion — performing
+ * any external cleanup (e.g. storage-byte reclaim) for those ids before
+ * returning. Ids not returned are kept this sweep (vetoed — e.g. the row
+ * regained references since it was marked).
+ *
+ * Guards are registered at runtime (`registerReapGuard`), not declared in the
+ * spec: detection and scheduling stay inside the single platform sweep
+ * (ADR-0057 §3.3 — a guard is a domain callback, not a second sweeper).
+ */
+export type LifecycleReapGuard = (
+  object: string,
+  rows: Array<Record<string, unknown>>,
+) => Promise<Array<string | number>>;
+
 export class LifecycleService {
   private readonly now: () => number;
   private timer: ReturnType<typeof setInterval> | undefined;
@@ -209,6 +231,8 @@ export class LifecycleService {
   private lastCounts = new Map<string, number>();
   /** Governance snapshot for the sweep in flight. */
   private governance: GovernanceSnapshot = DEFAULT_GOVERNANCE;
+  /** Per-object reap guards ({@link LifecycleReapGuard}). */
+  private readonly reapGuards = new Map<string, LifecycleReapGuard>();
 
   constructor(private readonly opts: LifecycleServiceOptions) {
     this.now = opts.now ?? (() => Date.now());
@@ -241,6 +265,17 @@ export class LifecycleService {
     this.timer = undefined;
   }
 
+  /**
+   * Register a {@link LifecycleReapGuard} for one object. From then on the
+   * Reaper never blind-deletes that object's rows: candidates are fetched,
+   * the guard confirms (after external cleanup) or vetoes each row, and only
+   * confirmed ids are deleted. One guard per object (last registration wins —
+   * guards are platform wiring, not user surface).
+   */
+  registerReapGuard(object: string, guard: LifecycleReapGuard): void {
+    this.reapGuards.set(object, guard);
+  }
+
   /**
    * Apply every declared lifecycle policy once. Safe to call directly (the
    * dogfood growth gate and `db:clean`-style tooling do); re-entrant calls
@@ -604,56 +639,89 @@ export class LifecycleService {
     // rows outside it (live workflow state) are retained regardless of age.
     const scope = onlyWhen ?? {};
 
+    // A guarded object is NEVER blind-deleted: without row reads the guard
+    // cannot confirm, so the reap is skipped (fail-safe), not degraded.
+    const guard = this.reapGuards.get(object);
+    if (guard && typeof engine.find !== 'function') {
+      if (!report.skipped.some((s) => s.object === object && s.reason === 'reap-guard-unsupported')) {
+        report.skipped.push({ object, reason: 'reap-guard-unsupported' });
+      }
+      return 0;
+    }
+
     let total: number | undefined = 0;
-    const accumulate = (res: unknown) => {
-      const n = countDeleted(res);
+    const accumulate = (n: number | undefined) => {
       if (n === undefined) total = undefined;
       else if (total !== undefined) total += n;
     };
+    const reapWhere = async (where: Record<string, unknown>): Promise<number | undefined> =>
+      guard
+        ? this.guardedReap(engine, object, guard, where)
+        : countDeleted(await engine.delete(object, { where, multi: true, context: { ...SYSTEM_CTX } }));
 
     if (tenantWindows.length === 0) {
-      accumulate(
-        await engine.delete(object, {
-          where: { [field]: { $lt: cutoff }, ...scope },
-          multi: true,
-          context: { ...SYSTEM_CTX },
-        }),
-      );
+      accumulate(await reapWhere({ [field]: { $lt: cutoff }, ...scope }));
     } else {
       // Tenant-level windows (P4): each overriding tenant gets its own
       // cutoff on its own rows…
       for (const t of tenantWindows) {
         const tMs = this.effectiveWindowMs(t[overrideKey], windowMs, `${object} (tenant ${t.tenantId})`);
         const tCutoff = new Date(this.now() - tMs).toISOString();
-        accumulate(
-          await engine.delete(object, {
-            where: { [field]: { $lt: tCutoff }, organization_id: t.tenantId, ...scope },
-            multi: true,
-            context: { ...SYSTEM_CTX },
-          }),
-        );
+        accumulate(await reapWhere({ [field]: { $lt: tCutoff }, organization_id: t.tenantId, ...scope }));
       }
       // …and the global pass covers everyone else, INCLUDING rows with no
       // organization (a bare `$nin` would silently skip NULL-org rows).
       accumulate(
-        await engine.delete(object, {
-          where: {
-            [field]: { $lt: cutoff },
-            $or: [
-              { organization_id: { $nin: tenantWindows.map((t) => t.tenantId) } },
-              { organization_id: null },
-            ],
-            ...scope,
-          },
-          multi: true,
-          context: { ...SYSTEM_CTX },
+        await reapWhere({
+          [field]: { $lt: cutoff },
+          $or: [
+            { organization_id: { $nin: tenantWindows.map((t) => t.tenantId) } },
+            { organization_id: null },
+          ],
+          ...scope,
         }),
       );
     }
 
     report.swept.push({ object, class: lc.class, policy, cutoff, deleted: total });
     return total;
   }
+
+  /**
+   * Guarded reap: fetch candidate rows in batches, let the guard confirm
+   * (after performing external cleanup) or veto each, delete only confirmed
+   * ids. A guard error propagates to the per-object handler in `sweep()` —
+   * an erroring guard must never fail open into deletion. A batch that isn't
+   * fully confirmed ends the pass: vetoed rows still match the cutoff filter
+   * and would be re-fetched forever; the next sweep retries them.
+   */
+  private async guardedReap(
+    engine: LifecycleEngineLike,
+    object: string,
+    guard: LifecycleReapGuard,
+    where: Record<string, unknown>,
+  ): Promise<number> {
+    let total = 0;
+    for (let batch = 0; batch < REAP_GUARD_MAX_BATCHES_PER_SWEEP; batch++) {
+      const rows = await engine.find!(object, {
+        where,
+        limit: REAP_GUARD_BATCH_SIZE,
+        context: { ...SYSTEM_CTX },
+      });
+      if (!rows?.length) break;
+      const confirmed = (await guard(object, rows)).filter((id) => id !== null && id !== undefined);
+      if (confirmed.length > 0) {
+        await engine.delete(object, {
+          where: { id: { $in: confirmed } },
+          multi: true,
+          context: { ...SYSTEM_CTX },
+        });
+        total += confirmed.length;
+      }
+      if (confirmed.length < rows.length || rows.length < REAP_GUARD_BATCH_SIZE) break;
+    }
+    return total;
+  }
 }
 
 /** Best-effort row-count extraction from a driver's delete result. */