diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index d18cf9a80..80b48ef8a 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/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, diff --git a/polylogue/maintenance/blob_reference_closure.py b/polylogue/maintenance/blob_reference_closure.py index 173589e20..728082a13 100644 --- a/polylogue/maintenance/blob_reference_closure.py +++ b/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] ) diff --git a/polylogue/storage/blob_integrity.py b/polylogue/storage/blob_integrity.py index 4768fb249..0af2b23dd 100644 --- a/polylogue/storage/blob_integrity.py +++ b/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,6 +2337,10 @@ 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: @@ -2339,7 +2348,7 @@ def ok(self) -> bool: @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, ) diff --git a/polylogue/storage/blob_liveness.py b/polylogue/storage/blob_liveness.py index d3393f725..37e769247 100644 --- a/polylogue/storage/blob_liveness.py +++ b/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", diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index fb8f1767a..28292f26b 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/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") diff --git a/tests/unit/maintenance/test_rebuild_index_candidate_promotion.py b/tests/unit/maintenance/test_rebuild_index_candidate_promotion.py index 8f9ed3315..8b2c3ce91 100644 --- a/tests/unit/maintenance/test_rebuild_index_candidate_promotion.py +++ b/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), ) diff --git a/tests/unit/storage/test_blob_integrity.py b/tests/unit/storage/test_blob_integrity.py index e0b95c4fa..d6ad5719d 100644 --- a/tests/unit/storage/test_blob_integrity.py +++ b/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