fix(maintenance): stop counting unowned attachments as attachment debt

polylogue/maintenance/archive_verification.py

@@ -64,7 +64,10 @@
 )
 from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin
 from polylogue.storage.blob_integrity import scan_attachment_coverage, scan_blob_integrity
-from polylogue.storage.blob_liveness import validated_blob_ref_liveness_joins
+from polylogue.storage.blob_liveness import (
+    acquired_attachment_missing_ref_predicate,
+    validated_blob_ref_liveness_joins,
+)
 from polylogue.storage.blob_store import BlobStore
 from polylogue.storage.introspection import table_exists
 from polylogue.storage.raw_failure_lifecycle import read_raw_failure_lifecycle
@@ -1756,10 +1759,9 @@ def _check_blob_reference_closure_for_index(
         attachment_sample = [
             str(row[0])
             for row in index_conn.execute(
-                """
+                f"""
                 SELECT a.attachment_id FROM attachments a
-                WHERE a.acquisition_status = 'acquired'
-                  AND NOT EXISTS (SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id)
+                WHERE {acquired_attachment_missing_ref_predicate()}
                 ORDER BY a.attachment_id LIMIT ?
                 """,
                 (sample_limit,),
@@ -1968,7 +1970,10 @@ def _check_attachment_coverage_at_index_path(
         summary=(
             f"acquired attachment debt: missing_blob={missing:,}, unreachable={unreachable:,}"
             if debt_count
-            else f"all {report.acquired_count:,} acquired attachment(s) have bytes and a live attachment reference"
+            else (
+                f"all {report.acquired_reachable_count:,} acquired attachment(s) have bytes and a live attachment reference"
+                + (f"; {report.acquired_unowned_count:,} retained unowned" if report.acquired_unowned_count else "")
+            )
         ),
         count=debt_count,
         details=details,

polylogue/maintenance/blob_reference_closure.py

@@ -23,6 +23,7 @@
     UnrecoverableAttachmentReason,
     plan_orphaned_attachment_relink,
 )
+from polylogue.storage.blob_liveness import acquired_attachment_missing_ref_predicate
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.migration_runner import (
     validate_backup_manifest_covers_derived_tier,
@@ -563,10 +564,9 @@ def closure_counts(source_conn: sqlite3.Connection, index_conn: sqlite3.Connecti
     )
     attachment_missing = int(
         index_conn.execute(
-            """
+            f"""
             SELECT COUNT(*) FROM attachments a
-            WHERE a.acquisition_status = 'acquired'
-              AND NOT EXISTS (SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id)
+            WHERE {acquired_attachment_missing_ref_predicate()}
             """
         ).fetchone()[0]
     )

polylogue/storage/blob_integrity.py

@@ -38,7 +38,12 @@
 from polylogue.core.raw_coordinates import zip_member_identity_coordinate
 from polylogue.logging import get_logger
 from polylogue.sources.origin_specs import artifact_rule_for_path
-from polylogue.storage.blob_liveness import BlobLivenessProjection, project_index_blob_hashes, project_live_blob_hashes
+from polylogue.storage.blob_liveness import (
+    BlobLivenessProjection,
+    acquired_attachment_missing_ref_predicate,
+    project_index_blob_hashes,
+    project_live_blob_hashes,
+)
 from polylogue.storage.blob_store import BlobNamespaceEntry, BlobStore
 from polylogue.storage.introspection import column_exists as _column_exists
 from polylogue.storage.introspection import table_exists as _table_exists
@@ -2332,14 +2337,18 @@ class AttachmentCoverageReport:
     # disk somewhere, most of which nothing can ever reach".
     acquired_unreachable_count: int
     acquired_unreachable_sample: tuple[str, ...]
+    #: Acquired attachments the writer retained with an ambiguous owner
+    #: (ref_count 0, never swept). Unreferenced by construction, so never
+    #: coverage debt -- and not reachable either, so they are their own term.
+    acquired_unowned_count: int = 0
 
     @property
     def ok(self) -> bool:
         return self.acquired_missing_blob_count == 0
 
     @property
     def acquired_reachable_count(self) -> int:
-        return self.acquired_count - self.acquired_unreachable_count
+        return self.acquired_count - self.acquired_unreachable_count - self.acquired_unowned_count
 
     def to_dict(self) -> dict[str, object]:
         return {
@@ -2349,6 +2358,7 @@ def to_dict(self) -> dict[str, object]:
             "acquired_reachable_count": self.acquired_reachable_count,
             "acquired_unreachable_count": self.acquired_unreachable_count,
             "acquired_unreachable_sample": list(self.acquired_unreachable_sample),
+            "acquired_unowned_count": self.acquired_unowned_count,
             "acquired_missing_blob_count": self.acquired_missing_blob_count,
             "acquired_missing_blob_sample": list(self.acquired_missing_blob_sample),
             "unavailable_count": self.unavailable_count,
@@ -2382,16 +2392,25 @@ def scan_attachment_coverage(
         # as its own dimension, distinct from "bytes missing from the blob
         # store" (acquired_missing_blob_count, below).
         unreachable_rows = conn.execute(
-            """
+            f"""
             SELECT a.attachment_id AS attachment_id
             FROM attachments a
-            WHERE a.acquisition_status = 'acquired'
-              AND NOT EXISTS (
-                  SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id
-              )
+            WHERE {acquired_attachment_missing_ref_predicate()}
             ORDER BY a.attachment_id
             """
         ).fetchall()
+        # The writer's owner-ambiguous retention: unreferenced by construction,
+        # so it is reported as its own dimension rather than as debt.
+        unowned_count = int(
+            conn.execute(
+                """
+                SELECT COUNT(*) FROM attachments a
+                WHERE a.acquisition_status = 'acquired'
+                  AND a.ref_count = 0
+                  AND NOT EXISTS (SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id)
+                """
+            ).fetchone()[0]
+        )
 
     missing_sample: list[str] = []
     missing_count = 0
@@ -2417,6 +2436,7 @@ def scan_attachment_coverage(
         unfetched_count=status_counts.get("unfetched", 0),
         acquired_unreachable_count=len(unreachable_rows),
         acquired_unreachable_sample=unreachable_sample,
+        acquired_unowned_count=unowned_count,
     )
 
 

polylogue/storage/blob_liveness.py

@@ -21,6 +21,23 @@
 from polylogue.storage.introspection import table_exists as _table_exists
 
 
+#: An attachment the writer retained with an ambiguous owner is unreferenced by
+#: construction: ``_write_attachments`` inserts it with ``ref_count`` 0 and
+#: deliberately keeps it out of the ref-count sweep, so it never had a ref to
+#: lose. Reference-closure debt and unreachable-coverage debt both mean "refs
+#: went away without the sweep running", which only a non-zero ``ref_count``
+#: witnesses. Every site that counts ref-less acquired attachments as debt uses
+#: this predicate, so the two states cannot be conflated on one route.
+def acquired_attachment_missing_ref_predicate(alias: str = "a", *, refs_table: str = "attachment_refs") -> str:
+    """SQL predicate for an acquired attachment whose refs went away."""
+    return (
+        f"{alias}.acquisition_status = 'acquired'\n"
+        f"              AND {alias}.ref_count != 0\n"
+        f"              AND NOT EXISTS (SELECT 1 FROM {refs_table} r "
+        f"WHERE r.attachment_id = {alias}.attachment_id)"
+    )
+
+
 class LivenessState(str, Enum):
     LIVE = "live"
     UNREFERENCED = "unreferenced"
@@ -492,6 +509,7 @@ def project_index_blob_hashes(index_conn: sqlite3.Connection) -> BlobLivenessPro
 
 __all__ = [
     "BLOB_OWNERS",
+    "acquired_attachment_missing_ref_predicate",
     "BlobLiveness",
     "BlobLivenessProjection",
     "LivenessState",

tests/unit/maintenance/test_archive_verification.py

@@ -1482,14 +1482,23 @@ def test_full_blob_integrity_red_twin(tmp_path: Path, mutation: str) -> None:
 
 
 def test_acquired_unreachable_attachment_debt_is_blocking(tmp_path: Path) -> None:
-    """Acquired bytes without an attachment_refs edge are not queryable."""
+    """Acquired bytes whose refs went away are not queryable.
+
+    The non-zero ``ref_count`` is what makes this debt rather than the
+    writer's deliberate unowned retention: the sweep set it while refs
+    existed, then the refs disappeared without the sweep running again.
+
+    Anti-vacuity: the companion test below inserts the same row with
+    ``ref_count`` 0 and must stay green, so this cannot pass merely because
+    every ref-less acquired attachment is reported.
+    """
     _seed_coherent_archive(tmp_path)
     blob_hash, size = BlobStore(tmp_path / "blob").write_from_bytes(b"unreachable attachment")
     with _connect(tmp_path / "index.db") as conn:
         conn.execute(
             """
             INSERT INTO attachments(attachment_id, blob_hash, byte_count, acquisition_status, ref_count)
-            VALUES ('unreachable-attachment', ?, ?, 'acquired', 0)
+            VALUES ('unreachable-attachment', ?, ?, 'acquired', 1)
             """,
             (bytes.fromhex(blob_hash), size),
         )
@@ -1504,6 +1513,43 @@ def test_acquired_unreachable_attachment_debt_is_blocking(tmp_path: Path) -> Non
     assert "unreachable-attachment" in check.details[0]
 
 
+def test_owner_ambiguous_attachment_is_not_coverage_or_closure_debt(tmp_path: Path) -> None:
+    """The writer's typed unowned retention is not attachment debt.
+
+    ``_write_attachments`` inserts an attachment whose owning message is
+    ambiguous with ``ref_count`` 0 and keeps it out of the ref-count sweep, so
+    it never had a ref to lose. Both declarations that count ref-less acquired
+    attachments must read it as explained.
+
+    Anti-vacuity: the test above inserts the same row with a non-zero
+    ``ref_count`` and must stay red, so this cannot pass because the checks
+    stopped reporting ref-less acquired attachments at all.
+    """
+    _seed_coherent_archive(tmp_path)
+    blob_hash, size = BlobStore(tmp_path / "blob").write_from_bytes(b"unowned attachment")
+    with _connect(tmp_path / "index.db") as conn:
+        conn.execute(
+            """
+            INSERT INTO attachments(attachment_id, blob_hash, byte_count, acquisition_status, ref_count)
+            VALUES ('unowned-attachment', ?, ?, 'acquired', 0)
+            """,
+            (bytes.fromhex(blob_hash), size),
+        )
+        conn.commit()
+
+    report = verify_archive(tmp_path, checks=("attachment-coverage", "blob-reference-closure"))
+
+    assert not report.blocking, [c.summary for c in report.checks]
+    coverage = _check(report, "attachment-coverage")
+    assert coverage.status is OutcomeStatus.OK
+    assert coverage.evidence["unreachable_count"] == 0
+    assert cast(dict[str, object], coverage.evidence["scan"])["acquired_unowned_count"] == 1
+    assert "retained unowned" in coverage.summary
+    closure = _check(report, "blob-reference-closure")
+    assert closure.status is OutcomeStatus.OK
+    assert closure.evidence["acquired_attachment_missing_ref_count"] == 0
+
+
 def test_orphaned_embedding_ref_trips_embeddings_refs_liveness(tmp_path: Path) -> None:
     _seed_coherent_archive(tmp_path)
     conn = _connect(tmp_path / "embeddings.db")

tests/unit/maintenance/test_rebuild_index_candidate_promotion.py

@@ -235,7 +235,7 @@ def test_rebuild_preflight_rejects_acquired_unreachable_attachment_before_candid
         conn.execute(
             """
             INSERT INTO attachments(attachment_id, blob_hash, byte_count, acquisition_status, ref_count)
-            VALUES ('unreachable-attachment', ?, ?, 'acquired', 0)
+            VALUES ('unreachable-attachment', ?, ?, 'acquired', 1)
             """,
             (bytes.fromhex(blob_hash), size),
         )

tests/unit/storage/test_blob_integrity.py

@@ -492,7 +492,7 @@ def test_scan_attachment_coverage_flags_acquired_row_with_no_attachment_ref(tmp_
         """
         INSERT INTO attachments (
             attachment_id, display_name, media_type, byte_count, blob_hash, acquisition_status, ref_count
-        ) VALUES (?, ?, ?, ?, ?, 'acquired', 0)
+        ) VALUES (?, ?, ?, ?, ?, 'acquired', 1)
         """,
         ("orphan-att-1", "orphan.txt", "text/plain", blob_size, bytes.fromhex(blob_hash)),
     )
@@ -507,6 +507,11 @@ def test_scan_attachment_coverage_flags_acquired_row_with_no_attachment_ref(tmp_
     assert report.acquired_unreachable_count == 1
     assert report.acquired_unreachable_sample == ("orphan-att-1",)
     assert report.acquired_reachable_count == 0
+    # The stale non-zero ref_count is what makes this debt: refs existed when
+    # the sweep last ran and then went away without it running again. A row
+    # inserted with ref_count 0 is the writer's typed unowned retention and is
+    # reported as `unowned_count` instead.
+    assert report.acquired_unowned_count == 0
     # `ok` only tracks missing blob bytes (a distinct dimension) -- bytes ARE
     # present on disk here, so `ok` stays True even though the row is
     # unreachable. Reachability debt is its own signal