feat(cleanup): reclaim orphaned attachments — rows and store blobs (#3068)

PostAttachment's two FKs are both ON DELETE SET NULL, so a deleted post or
artist left the row behind rather than taking it. Nothing ever pruned those
rows, and nothing in the repo had ever unlinked a file under the attachment
store — so both rows and bytes accumulated permanently, invisible to every
existing diagnostic.

Why a disk->DB reconciliation rather than a row sweep: the store is
sha-addressed and idempotent, so ONE blob backs MANY rows. Deleting a row
does not free its blob, and since the artist cascade (#3066) now deletes
its attachment rows outright, a freed blob has no DB pointer left to find
it by. Walking the store and asking "does any row still reference this
sha?" catches orphans from every cause, including ones no future delete
path will think to report.

Preview and apply share `_orphan_attachment_conditions` (rule 93). The
dry-run derives its surviving-sha set by NEGATING that same predicate, so
it is honest about blobs the delete would free rather than counting them as
still-referenced — the one place this was easy to get backwards, so it has
its own parity test.

Guards, each with a reason:
- A blob is written before its row commits, so a just-stored file legitimately
  has no referencing row. Files under 6h are never judged — same guard and
  reasoning as ORPHAN_TEMP_MIN_AGE_HOURS.
- `.partial` staging files belong to cleanup_orphaned_temp_files; skipped
  rather than raced.
- The sha is parsed as the first 64 chars, not via Path.stem: store() takes
  the extension from the source filename, and a URL-encoded basename yields a
  multi-dot suffix that would make stem eat part of the sha.
- A 900s walk budget reports partial=True instead of running to the task's
  hard limit (rule 89).
- TASK_STUCK_THRESHOLD_MINUTES override at 30 (= time_limit 25 + 5). Without
  it a healthy 20-minute walk is phantom-flagged 'RecoverySweep' at the bare
  5-min default — the #883 failure class; its invariant test is mirrored here.

Defaults to the safe preview at both the task and the route, unlike the other
maintenance triggers: this apply unlinks files. Operator-triggered only,
never on a beat.

Ships with its UI (rule 27): AttachmentReclaimCard in Cleanup → Duplicates &
leftovers, built on the existing useMaintenanceTask/MaintenanceTile shapes, so
a run survives navigating away. Surfaces files_failed and partial explicitly,
since both change what the numbers mean.

Also promotes humanBytes to utils/bytes.js — it was byte-identical in
VideoDedupCard and GatedPurgeCard and this card would have been the third
copy. The three divergent `formatBytes` helpers are deliberately left alone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/test_api_admin.py

@@ -677,3 +677,34 @@ async def test_reset_content_tagging_apply_requires_confirm_token(client, db):
     )
     assert resp.status_code == 200
     assert (await resp.get_json())["deleted"] == 1
+
+
+@pytest.mark.asyncio
+async def test_trigger_reclaim_attachments_defaults_to_preview(client, monkeypatch):
+    """Unlike the other maintenance triggers, this one's apply unlinks FILES —
+    so an empty body must mean preview, not apply."""
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.reclaim_orphaned_attachments_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post("/api/admin/maintenance/reclaim-attachments", json={})
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert calls[0][1] == {"dry_run": True}
+
+
+@pytest.mark.asyncio
+async def test_trigger_reclaim_attachments_threads_apply(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.reclaim_orphaned_attachments_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post(
+        "/api/admin/maintenance/reclaim-attachments", json={"dry_run": False},
+    )
+    assert resp.status_code == 202
+    assert calls[0][1] == {"dry_run": False}

tests/test_cleanup_service.py

@@ -5,6 +5,7 @@ side effects use tmp_path. Assertions on mutated rows use COLUMN
 SELECTS per reference_async_coredml_test_assertions — never
 re-read ORM attributes after a service mutates and re-fetches.
 """
+import os
 from datetime import UTC, datetime
 
 import pytest
@@ -1081,3 +1082,176 @@ def test_reconcile_preserves_from_attachment_on_provenance_collision(db_sync, tm
         .where(ImageProvenance.image_record_id == img_id)
     ).all()
     assert rows == [(native_id, att_id)]
+
+
+# --- reclaim_orphaned_attachments -----------------------------------
+
+
+def _store_blob(root, sha, *, ext=".pdf", age_hours=48, data=b"blob"):
+    """Write a file into the sha-addressed attachment store, aged past the
+    min-age guard by default."""
+    d = root / "attachments" / sha[:3]
+    d.mkdir(parents=True, exist_ok=True)
+    p = d / f"{sha}{ext}"
+    p.write_bytes(data)
+    old = datetime.now(UTC).timestamp() - age_hours * 3600
+    os.utime(p, (old, old))
+    return p
+
+
+def _attachment(db_sync, *, sha, post=None, artist=None):
+    att = PostAttachment(
+        post_id=post.id if post else None,
+        artist_id=artist.id if artist else None,
+        sha256=sha, path=f"/store/{sha[:3]}/f.pdf",
+        original_filename="f.pdf", ext=".pdf", size_bytes=4,
+    )
+    db_sync.add(att)
+    db_sync.flush()
+    return att
+
+
+def test_reclaim_attachments_dry_run_projects_without_mutating(db_sync, tmp_path):
+    a = _make_artist(db_sync, slug="recl-dry")
+    p = Post(artist_id=a.id, external_post_id="rd-1")
+    db_sync.add(p)
+    db_sync.flush()
+    kept_sha, orphan_sha = "aa11".ljust(64, "0"), "bb22".ljust(64, "0")
+    _attachment(db_sync, sha=kept_sha, post=p, artist=a)
+    _attachment(db_sync, sha=orphan_sha)            # both FKs NULL → orphan
+    db_sync.commit()
+    kept_blob = _store_blob(tmp_path, kept_sha)
+    orphan_blob = _store_blob(tmp_path, orphan_sha)
+
+    result = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=True,
+    )
+    assert result["rows"] == 1
+    assert result["files"] == 1
+    assert result["bytes"] == orphan_blob.stat().st_size
+
+    # Nothing actually happened.
+    assert kept_blob.exists() and orphan_blob.exists()
+    assert db_sync.execute(
+        select(func.count(PostAttachment.id))
+    ).scalar_one() == 2
+
+
+def test_reclaim_attachments_apply_deletes_rows_and_unlinks_blobs(db_sync, tmp_path):
+    a = _make_artist(db_sync, slug="recl-apply")
+    p = Post(artist_id=a.id, external_post_id="ra-1")
+    db_sync.add(p)
+    db_sync.flush()
+    kept_sha, orphan_sha = "cc33".ljust(64, "0"), "dd44".ljust(64, "0")
+    _attachment(db_sync, sha=kept_sha, post=p, artist=a)
+    _attachment(db_sync, sha=orphan_sha)
+    db_sync.commit()
+    kept_blob = _store_blob(tmp_path, kept_sha)
+    orphan_blob = _store_blob(tmp_path, orphan_sha)
+
+    result = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=False,
+    )
+    assert result["rows"] == 1
+    assert result["files"] == 1
+
+    assert kept_blob.exists()          # still referenced
+    assert not orphan_blob.exists()    # nothing points at it any more
+    surviving = db_sync.execute(select(PostAttachment.sha256)).scalars().all()
+    assert surviving == [kept_sha]
+
+
+def test_reclaim_attachments_preview_matches_apply(db_sync, tmp_path):
+    """Rule 93 — the dry-run's numbers are what the apply does. The projection
+    has to negate the orphan predicate to be honest about blobs the delete is
+    about to free, so this is the assertion that catches getting that backwards.
+    """
+    orphan_sha = "ee55".ljust(64, "0")
+    _attachment(db_sync, sha=orphan_sha)
+    db_sync.commit()
+    _store_blob(tmp_path, orphan_sha)
+
+    projected = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=True,
+    )
+    applied = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=False,
+    )
+    for key in ("rows", "files", "bytes"):
+        assert projected[key] == applied[key], key
+    assert applied["rows"] == 1 and applied["files"] == 1
+
+
+def test_reclaim_attachments_keeps_shared_blob_while_any_row_remains(db_sync, tmp_path):
+    """The refcount case this whole sweep exists for: one sha-addressed blob
+    backs several rows, so deleting SOME of them must not free the file."""
+    a = _make_artist(db_sync, slug="recl-shared")
+    p = Post(artist_id=a.id, external_post_id="rs-1")
+    db_sync.add(p)
+    db_sync.flush()
+    sha = "ff66".ljust(64, "0")
+    _attachment(db_sync, sha=sha, post=p, artist=a)   # attributed — survives
+    _attachment(db_sync, sha=sha)                     # orphan — deleted
+    db_sync.commit()
+    blob = _store_blob(tmp_path, sha)
+
+    result = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=False,
+    )
+    assert result["rows"] == 1     # the orphan row went
+    assert result["files"] == 0    # the blob did NOT
+    assert blob.exists()
+
+
+def test_reclaim_attachments_spares_filesystem_import_rows(db_sync, tmp_path):
+    """post_id NULL with an artist_id is the deliberate filesystem-import shape
+    (importer._capture_attachment), not an orphan — it is still attributed."""
+    a = _make_artist(db_sync, slug="recl-fsimport")
+    sha = "1177".ljust(64, "0")
+    _attachment(db_sync, sha=sha, artist=a)    # post NULL, artist set
+    db_sync.commit()
+    blob = _store_blob(tmp_path, sha)
+
+    result = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=False,
+    )
+    assert result["rows"] == 0
+    assert result["files"] == 0
+    assert blob.exists()
+    assert db_sync.execute(
+        select(func.count(PostAttachment.id))
+    ).scalar_one() == 1
+
+
+def test_reclaim_attachments_skips_recent_and_staging_files(db_sync, tmp_path):
+    """A blob is written BEFORE its row commits, so a just-stored file with no
+    row is in-flight, not orphaned. `.partial` staging files belong to
+    cleanup_orphaned_temp_files and must be left alone either way."""
+    fresh_sha, staged_sha = "2288".ljust(64, "0"), "3399".ljust(64, "0")
+    fresh = _store_blob(tmp_path, fresh_sha, age_hours=0)
+    staged = _store_blob(tmp_path, staged_sha, ext=".pdf.partial")
+    db_sync.commit()
+
+    result = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=False,
+    )
+    assert result["files"] == 0
+    assert result["skipped_recent"] == 1
+    assert fresh.exists() and staged.exists()
+
+
+def test_reclaim_attachments_ignores_non_sha_named_files(db_sync, tmp_path):
+    """The walk must only judge files it can identify as store blobs — anything
+    else under the root is none of its business."""
+    d = tmp_path / "attachments" / "zzz"
+    d.mkdir(parents=True)
+    stray = d / "notes.txt"
+    stray.write_text("not a blob")
+    old = datetime.now(UTC).timestamp() - 48 * 3600
+    os.utime(stray, (old, old))
+
+    result = cleanup_service.reclaim_orphaned_attachments(
+        db_sync, images_root=tmp_path, dry_run=False,
+    )
+    assert result["files"] == 0
+    assert stray.exists()

tests/test_maintenance.py

@@ -778,3 +778,17 @@ def test_vacuum_analyze_runs_over_high_churn_tables():
 
     result = vacuum_analyze.apply().get()
     assert result["vacuumed"] == list(VACUUM_TABLES)
+
+
+def test_reclaim_attachments_stuck_threshold_exceeds_hard_time_limit():
+    """#883's invariant, applied to the attachment reclaim: a task whose stall
+    threshold is under its own hard limit gets phantom-flagged 'RecoverySweep'
+    while it is still healthily running."""
+    from backend.app.tasks.admin import reclaim_orphaned_attachments_task
+    from backend.app.tasks.maintenance import TASK_STUCK_THRESHOLD_MINUTES
+
+    hard_minutes = reclaim_orphaned_attachments_task.time_limit / 60
+    override = TASK_STUCK_THRESHOLD_MINUTES[
+        "backend.app.tasks.admin.reclaim_orphaned_attachments_task"
+    ]
+    assert override >= hard_minutes