From 7900e502c67694b92f53066111b79694b8cf6051 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 14:56:15 +0200 Subject: [PATCH] feat(maintenance): add source-conservation and population-coverage gates Add the `source-conservation` verify-archive declaration: every acquired source item types into exactly one term citing its rule, and every index session, message, block, and attachment ref traces back to a raw row. Phantom sessions (polylogue-b508) are reported by lineage and never deleted. A source file that is gone while the archive still retains its raw payload bytes is typed accounting (`source_missing`); only genuine byte loss (`source_lost`) blocks, alongside unexplained, unclassified, orphan, and phantom terms. Add `devtools gate population-coverage`: every origin, detector route, and artifact kind is declared and witnessed. Coverage comes from the declarations, so a stored `recognized_unparsed` support status is not itself a declaration, and the gate evaluates a source inventory only when `--archive-root` names one - its result never depends on ambient machine state. `planner-stats` now covers the documented set including `session_links`, and the coherent fixture ANALYZEs the whole index tier. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- devtools/gate.py | 7 + devtools/verify_population_coverage.py | 317 +++++++++ docs/maintenance.md | 3 +- polylogue/maintenance/archive_verification.py | 141 ++-- polylogue/maintenance/source_conservation.py | 626 ++++++++++++++++++ .../test_verify_population_coverage.py | 184 +++++ .../maintenance/test_archive_verification.py | 6 +- .../maintenance/test_source_conservation.py | 449 +++++++++++++ 8 files changed, 1674 insertions(+), 59 deletions(-) create mode 100644 devtools/verify_population_coverage.py create mode 100644 polylogue/maintenance/source_conservation.py create mode 100644 tests/unit/devtools/test_verify_population_coverage.py create mode 100644 tests/unit/maintenance/test_source_conservation.py diff --git a/devtools/gate.py b/devtools/gate.py index e9fc54c697..4c9943b061 100644 --- a/devtools/gate.py +++ b/devtools/gate.py @@ -189,6 +189,13 @@ def mypy_command(*, root: Path = ROOT) -> list[str]: ("devtools.schema_inference_gate",), label="gate schema-inference-gate", ), + Gate( + "population-coverage", + "Verify every origin, detector route, and artifact kind in the source inventory is declared and witnessed.", + "module", + ("devtools.verify_population_coverage",), + label="gate population-coverage", + ), Gate( "agent-integration", "Verify manual compilation, parser examples, continuation, native delivery, and packaging.", diff --git a/devtools/verify_population_coverage.py b/devtools/verify_population_coverage.py new file mode 100644 index 0000000000..69b59bed83 --- /dev/null +++ b/devtools/verify_population_coverage.py @@ -0,0 +1,317 @@ +"""``devtools gate population-coverage``: the real source inventory is declared and witnessed. + +Two halves, both read-only and neither able to create a fixture or a parser: + +* **Declarations**: every executable ``OriginSpec`` has a capability-matrix + entry whose witness fixtures exist on disk, and every non-executable origin + carries an unsupported receipt. Runs without an archive. +* **Inventory**: every origin, detector route (``detected_provider``), and + artifact kind observed in an archive's ``source.db`` maps to a declared + parser route, a declared artifact rule, or a typed unsupported exclusion. + A construct nothing declares is reported as typed unsupported evidence and + fails the gate. + +Ordinary value variation inside a declared construct is not a construct; a +new origin token, detector route, or artifact kind is. +""" + +from __future__ import annotations + +import argparse +import json +import sqlite3 +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +from polylogue.archive.artifact_taxonomy.models import ArtifactKind +from polylogue.core.enums import ArtifactSupportStatus, Provider +from polylogue.core.sources import origin_from_provider +from polylogue.sources.origin_specs import ORIGIN_SPECS, OriginSpec +from polylogue.storage.introspection import table_exists +from tests.infra.origin_capability_matrix import CapabilityManifest, load_manifest + +REPO_ROOT = Path(__file__).resolve().parents[1] + +COVERED = "covered" +UNSUPPORTED_DECLARED = "unsupported_declared" +UNCOVERED = "uncovered" + +#: Artifact kinds a session-bearing parser route consumes directly. +_SESSION_BEARING_KINDS: frozenset[str] = frozenset( + { + ArtifactKind.SESSION_DOCUMENT.value, + ArtifactKind.SESSION_RECORD_STREAM.value, + ArtifactKind.AGENT_TRANSCRIPT.value, + ArtifactKind.COORDINATOR_SESSION_STREAM.value, + } +) + + +@dataclass(frozen=True, slots=True) +class CoverageConstruct: + """One construct of the population and how it is covered.""" + + family: str + key: str + status: str + route: str + witness: str + count: int = 0 + + def to_dict(self) -> dict[str, object]: + return { + "family": self.family, + "key": self.key, + "status": self.status, + "route": self.route, + "witness": self.witness, + "count": self.count, + } + + +@dataclass(frozen=True, slots=True) +class PopulationCoverageReport: + archive_root: str | None + inventory_evaluated: bool + constructs: tuple[CoverageConstruct, ...] + + @property + def uncovered(self) -> tuple[CoverageConstruct, ...]: + return tuple(construct for construct in self.constructs if construct.status == UNCOVERED) + + @property + def ok(self) -> bool: + return not self.uncovered + + def to_dict(self) -> dict[str, object]: + counts: dict[str, int] = {} + for construct in self.constructs: + counts[construct.status] = counts.get(construct.status, 0) + 1 + return { + "ok": self.ok, + "archive_root": self.archive_root, + "inventory_evaluated": self.inventory_evaluated, + "summary": counts, + "constructs": [construct.to_dict() for construct in self.constructs], + } + + +def _spec_by_origin(specs: Sequence[OriginSpec]) -> dict[str, OriginSpec]: + return {spec.origin.value: spec for spec in specs} + + +def _matrix_witness(manifest: CapabilityManifest, origin: str) -> tuple[str, str] | None: + """Return ``(status, witness)`` for an origin from the capability matrix.""" + for entry in manifest.entries: + if entry.origin.value != origin: + continue + if entry.unsupported is not None: + return UNSUPPORTED_DECLARED, f"matrix unsupported: {entry.unsupported.reason}" + present = [witness.fixture_path for witness in entry.witnesses if (REPO_ROOT / witness.fixture_path).is_file()] + if not present: + return None + return COVERED, ";".join(present) + return None + + +def declaration_constructs( + *, specs: Sequence[OriginSpec] = ORIGIN_SPECS, manifest: CapabilityManifest | None = None +) -> tuple[CoverageConstruct, ...]: + """Every declared origin has a witness (executable) or an unsupported receipt.""" + manifest = manifest if manifest is not None else load_manifest() + out: list[CoverageConstruct] = [] + for spec in specs: + origin = spec.origin.value + witness = _matrix_witness(manifest, origin) + route = ";".join(spec.parser_paths) or f"lifecycle:{spec.lifecycle}" + if spec.lifecycle == "executable": + if witness is None or witness[0] != COVERED: + out.append(CoverageConstruct("origin-declaration", origin, UNCOVERED, route, "no matrix witness")) + else: + out.append(CoverageConstruct("origin-declaration", origin, COVERED, route, witness[1])) + elif witness is None: + out.append(CoverageConstruct("origin-declaration", origin, UNCOVERED, route, "no unsupported receipt")) + else: + out.append(CoverageConstruct("origin-declaration", origin, UNSUPPORTED_DECLARED, route, witness[1])) + return tuple(out) + + +def inventory_constructs( + source_db: Path, + *, + specs: Sequence[OriginSpec] = ORIGIN_SPECS, + manifest: CapabilityManifest | None = None, +) -> tuple[CoverageConstruct, ...]: + """Classify every origin, detector route, and artifact kind in ``source_db``.""" + manifest = manifest if manifest is not None else load_manifest() + by_origin = _spec_by_origin(specs) + executable_wires: dict[str, str] = { + provider.value: spec.origin.value + for spec in specs + if spec.lifecycle == "executable" + for provider in spec.provider_wires + } + out: list[CoverageConstruct] = [] + conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) + try: + for origin, count in conn.execute("SELECT origin, COUNT(*) FROM raw_sessions GROUP BY origin"): + origin = str(origin) + spec = by_origin.get(origin) + if spec is None: + out.append(CoverageConstruct("origin", origin, UNCOVERED, "no OriginSpec", "none", int(count))) + continue + witness = _matrix_witness(manifest, origin) + if spec.lifecycle == "executable" and witness is not None and witness[0] == COVERED: + out.append( + CoverageConstruct("origin", origin, COVERED, ";".join(spec.parser_paths), witness[1], int(count)) + ) + elif witness is not None and witness[0] == UNSUPPORTED_DECLARED: + out.append( + CoverageConstruct( + "origin", origin, UNSUPPORTED_DECLARED, f"lifecycle:{spec.lifecycle}", witness[1], int(count) + ) + ) + else: + out.append( + CoverageConstruct( + "origin", origin, UNCOVERED, f"lifecycle:{spec.lifecycle}", "no matrix witness", int(count) + ) + ) + + columns = {str(row[1]) for row in conn.execute("PRAGMA table_info(raw_sessions)")} + if "detected_provider" in columns: + for origin, provider, count in conn.execute( + """ + SELECT origin, detected_provider, COUNT(*) FROM raw_sessions + WHERE detected_provider IS NOT NULL GROUP BY origin, detected_provider + """ + ): + key = f"{origin}/{provider}" + declared_origin = executable_wires.get(str(provider)) + wire = Provider.from_string(str(provider)) + mapped = origin_from_provider(wire).value if wire is not Provider.UNKNOWN else None + if declared_origin is None or mapped != str(origin): + out.append( + CoverageConstruct( + "detector-route", key, UNCOVERED, "no executable provider wire", "none", int(count) + ) + ) + else: + spec = by_origin[declared_origin] + out.append( + CoverageConstruct( + "detector-route", + key, + COVERED, + ";".join(binding.predicate_path for binding in spec.detector_bindings) + or ";".join(spec.parser_paths), + ";".join(spec.coverage_refs), + int(count), + ) + ) + + if table_exists(conn, "raw_artifacts"): + for origin, kind, support, count in conn.execute( + "SELECT origin, artifact_kind, support_status, COUNT(*) FROM raw_artifacts GROUP BY 1, 2, 3" + ): + out.append(_artifact_construct(by_origin, manifest, str(origin), str(kind), str(support), int(count))) + finally: + conn.close() + return tuple(out) + + +def _artifact_construct( + by_origin: dict[str, OriginSpec], + manifest: CapabilityManifest, + origin: str, + kind: str, + support: str, + count: int, +) -> CoverageConstruct: + key = f"{origin}/{kind}/{support}" + spec = by_origin.get(origin) + known_kind = kind in {member.value for member in ArtifactKind} + if spec is None or not known_kind or kind == ArtifactKind.UNKNOWN.value: + return CoverageConstruct("artifact-kind", key, UNCOVERED, "no artifact declaration", "none", count) + if support == ArtifactSupportStatus.UNSUPPORTED_PARSEABLE.value: + return CoverageConstruct( + "artifact-kind", key, UNSUPPORTED_DECLARED, "artifact taxonomy: unsupported_parseable", "taxonomy", count + ) + for rule in spec.artifact_rules: + if rule.kind == kind: + route = rule.parser_path or f"parse_policy:{rule.parse_policy}" + return CoverageConstruct("artifact-kind", key, COVERED, route, rule.coverage_role, count) + if kind in _SESSION_BEARING_KINDS and spec.lifecycle == "executable": + witness = _matrix_witness(manifest, origin) + if witness is not None and witness[0] == COVERED: + return CoverageConstruct("artifact-kind", key, COVERED, ";".join(spec.parser_paths), witness[1], count) + if kind == ArtifactKind.HOOK_EVENT.value: + return CoverageConstruct("artifact-kind", key, COVERED, "raw hook event capture", "hook_event", count) + return CoverageConstruct("artifact-kind", key, UNCOVERED, "no artifact rule for origin", "none", count) + + +def evaluate_population_coverage( + archive_root: Path | None, + *, + specs: Sequence[OriginSpec] = ORIGIN_SPECS, + manifest: CapabilityManifest | None = None, +) -> PopulationCoverageReport: + manifest = manifest if manifest is not None else load_manifest() + constructs = list(declaration_constructs(specs=specs, manifest=manifest)) + source_db = archive_root / "source.db" if archive_root is not None else None + evaluated = source_db is not None and source_db.is_file() + if evaluated and source_db is not None: + constructs.extend(inventory_constructs(source_db, specs=specs, manifest=manifest)) + return PopulationCoverageReport( + archive_root=str(archive_root) if archive_root is not None else None, + inventory_evaluated=evaluated, + constructs=tuple(constructs), + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Verify every origin, detector route, and artifact kind in the source inventory is declared." + ) + parser.add_argument( + "--archive-root", + type=Path, + default=None, + help="evaluate the source inventory at this archive root; declarations alone are checked without it", + ) + parser.add_argument("--json", action="store_true", dest="as_json") + args = parser.parse_args(argv) + report = evaluate_population_coverage(args.archive_root) + if args.as_json: + print(json.dumps(report.to_dict(), sort_keys=True)) + else: + print(f"Population coverage: {'PASS' if report.ok else 'FAIL'}") + print( + f"Inventory: {'evaluated at ' + str(report.archive_root) if report.inventory_evaluated else 'not evaluated (no source.db)'}" + ) + summary: dict[str, int] = {} + for construct in report.constructs: + summary[construct.status] = summary.get(construct.status, 0) + 1 + for status, n in sorted(summary.items()): + print(f" {status}: {n}") + for construct in report.uncovered: + print(f" UNCOVERED {construct.family} {construct.key} ({construct.count:,}): {construct.route}") + return 0 if report.ok else 1 + + +__all__ = [ + "COVERED", + "UNCOVERED", + "UNSUPPORTED_DECLARED", + "CoverageConstruct", + "PopulationCoverageReport", + "declaration_constructs", + "evaluate_population_coverage", + "inventory_constructs", + "main", +] + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/maintenance.md b/docs/maintenance.md index c096b8be3f..26d4c3c127 100644 --- a/docs/maintenance.md +++ b/docs/maintenance.md @@ -419,9 +419,10 @@ extensible registry): | `tier-schema` | Every tier file (source/index/embeddings/user/ops) exists at its current `PRAGMA user_version`. | | `pointer-coherence` | The conventional `index.db` path and the active `.index-active-pointer` generation agree (an interrupted blue-green promotion leaves these diverged — polylogue-k8kj class). | | `source-index-coverage` | Every raw logical head is materialized, has an explicit terminal disposition, or is quarantined, and every index session's `raw_id` still resolves to a real raw row (orphans). The raw source population, not the derived census ledger, defines the coverage universe. | +| `source-conservation` | Every acquired source item (each `raw_sessions` row, hook event, history sidecar) is materialized or carries a typed exclusion citing its rule (revision superseded, byte-duplicate receipt, parse failure, validation rejection, declared non-session artifact kind, decode failure, census verdict, pending); a raw row whose source file no longer exists on disk is `source_missing` when its raw payload bytes are still retained and `source_lost` when they are not. Reverse: every session traces to a raw row that is not a declared non-session artifact (phantom sessions, polylogue-b508, are reported and never deleted), and every message, block, and attachment ref traces to its owner. Unexplained, unclassified, lost-source, orphan, and phantom terms block; pending is a warning. The acceptance instrument for a rebuilt archive: zero blocking terms. | | `fts-parity` | `messages_fts`/`blocks_command_trigram` exactly cover their source `blocks` rows, archive-wide, with the worst-offending sessions surfaced by name. | | `lineage-sanity` | `session_links.resolved_dst_session_id` and `branch_point_message_id` resolve to real sessions/messages (the latter is deliberately not a foreign key — see the data-model docs). | -| `planner-stats` | `sqlite_stat1` covers `blocks`/`messages`/`session_links` (warn-level: a fresh generation without `ANALYZE` picks pathological query plans, polylogue-l3tk class). | +| `planner-stats` | `sqlite_stat1` covers `blocks`/`messages`/`session_links`/`action_pairs` (warn-level: a fresh generation without `ANALYZE` picks pathological query plans, polylogue-l3tk class). | | `counts-summary` | Archive-wide session/message/block counts and an origin breakdown — the numbers-freeze starting point for an operator handoff. | Exit code is non-zero when any check reports `error` (or, with `--strict`, diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index f07bb0d3a8..b31adad832 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -40,7 +40,6 @@ from pathlib import Path from typing import Any, Literal -from polylogue.archive.revision_authority import logical_head_cohort_sql from polylogue.archive.topology.edge import HOOK_AUTHORITATIVE_LINK_METHOD, HOOK_CONTRADICTED_LINK_METHOD from polylogue.core.json import JSONDocument, json_document from polylogue.core.outcomes import ( @@ -58,6 +57,11 @@ audit_chatgpt_content_conservation, audit_revision_fidelity, ) +from polylogue.maintenance.source_conservation import ( + audit_source_conservation, + logical_head_cohort_expr, + valid_byte_duplicate_supersession_expr, +) 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 @@ -76,7 +80,7 @@ #: index-tier tables the planner-stats check expects ``ANALYZE`` coverage for #: (polylogue-l3tk: fresh generations without stats pick pathological plans). -_PLANNER_STATS_COVERED_TABLES: tuple[str, ...] = ("blocks", "messages", "action_pairs") +_PLANNER_STATS_COVERED_TABLES: tuple[str, ...] = ("blocks", "messages", "session_links", "action_pairs") @dataclass @@ -332,53 +336,6 @@ def _check_pointer_coherence(archive_root: Path, _sample_limit: int) -> ArchiveV # --------------------------------------------------------------------------- -def _valid_byte_duplicate_supersession_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str: - """Return the receipt predicate shared by source/index coverage checks. - - A supersession receipt is authority only when it still names the same - bytes and an index materialization of the recorded duplicate twin. Keep - the predicate in one place so backlog freshness cannot classify a receipt - differently from source-index coverage. - """ - if not table_exists(conn, "raw_byte_duplicate_supersession_receipts"): - return "0" - return f""" - EXISTS( - SELECT 1 - FROM raw_byte_duplicate_supersession_receipts receipt - JOIN raw_sessions twin ON twin.raw_id = receipt.duplicate_of_raw_id - JOIN idx_tier.sessions twin_session - ON twin_session.raw_id = twin.raw_id - AND twin_session.session_id = receipt.duplicate_of_session_id - WHERE receipt.raw_id = {raw_alias}.raw_id - AND receipt.blob_hash = {raw_alias}.blob_hash - AND receipt.blob_size = {raw_alias}.blob_size - AND twin.blob_hash = {raw_alias}.blob_hash - AND twin.blob_size = {raw_alias}.blob_size - AND twin.origin IS {raw_alias}.origin - AND twin.source_path IS {raw_alias}.source_path - AND twin.source_index IS {raw_alias}.source_index - ) - """ - - -def _logical_head_cohort_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str: - """Return the durable identity used to group raw revisions into one head. - - A full-revision row retired into membership governance intentionally loses - its raw-level ``logical_source_key``. Its single retained membership key - remains the authoritative identity, so use it before the legacy - native-id/path fallback. Shared raws can hold several membership keys; - they have no one raw-level cohort and must keep that fallback instead of - being arbitrarily assigned to one member. - """ - return logical_head_cohort_sql( - conn, - raw_alias=raw_alias, - has_memberships=table_exists(conn, "raw_session_memberships"), - ) - - def _check_source_index_coverage(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: return _check_source_index_coverage_at_index_path(archive_root, _resolve_index_path(archive_root), sample_limit) @@ -426,8 +383,8 @@ def _check_source_index_coverage_at_index_path( census_expr = ( "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" ) - valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r") - logical_cohort_expr = _logical_head_cohort_expr(conn, raw_alias="r") + valid_supersession_expr = valid_byte_duplicate_supersession_expr(conn, raw_alias="r") + logical_cohort_expr = logical_head_cohort_expr(conn, raw_alias="r") # A read-only connection (``query_only=ON``, connection-wide, not # per-attached-db) cannot ``CREATE TEMP VIEW`` -- the temp schema @@ -627,6 +584,65 @@ def _check_source_index_coverage_at_candidate( return _check_source_index_coverage_at_index_path(archive_root, index_path, sample_limit) +def _check_source_conservation(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: + return _check_source_conservation_at_index_path(archive_root, _resolve_index_path(archive_root), sample_limit) + + +def _check_source_conservation_at_index_path( + archive_root: Path, index_path: Path, sample_limit: int +) -> ArchiveVerificationCheck: + """Every acquired source item and every index row types into one term. + + Forward: each ``raw_sessions`` row (not only logical heads), hook event, + and history sidecar is materialized or carries a typed exclusion whose + rule the report cites; an on-disk probe turns a raw row whose source file + vanished into ``source_missing``. Reverse: every session traces to a raw + row that is not a declared non-session artifact, and every message, + block, and attachment ref traces to its owner. Blocking terms are the + unexplained ones (``unexplained``, ``unclassified_shape``, + ``source_lost``, orphans, phantoms); a source file that is gone while its + raw payload bytes are retained (``source_missing``) is typed accounting, + and ``pending`` plus hook events whose session file was never acquired are + warnings. + """ + name = "source-conservation" + source_path = _tier_path(archive_root, ArchiveTier.SOURCE) + if not source_path.exists() or not index_path.exists(): + return _skip_check(name, "source.db or index.db not present") + try: + conn = _open_ro(source_path) + except sqlite3.Error as exc: + return _error_check(name, f"could not open source.db: {exc}", exc=exc) + try: + try: + conn.execute("ATTACH DATABASE ? AS idx_tier", (f"file:{index_path}?mode=ro",)) + except sqlite3.Error as exc: + return _error_check(name, f"could not attach index.db: {exc}", exc=exc) + try: + report = audit_source_conservation(conn, archive_root=archive_root, sample_limit=sample_limit) + except sqlite3.Error as exc: + return _error_check(name, f"could not read source/index tiers: {exc}", exc=exc) + finally: + conn.close() + + if report.blocking_count: + status = OutcomeStatus.ERROR + elif report.warning_count: + status = OutcomeStatus.WARNING + else: + status = OutcomeStatus.OK + return ArchiveVerificationCheck( + name=name, + status=status, + summary=report.summary(), + count=report.blocking_count, + details=[ + f"{term.name}:{item}" for term in report.terms if term.blocking and term.count for item in term.sample + ], + evidence=dict(report.to_json()), + ) + + def archive_verification_owner_adapters( archive_root: Path, *, @@ -846,6 +862,23 @@ def archive_verification_migrated_owner_adapters( else _check_source_index_coverage(archive_root, sample_limit) ), ), + _declared_owner( + name="source-conservation", + semantic_owner="source-materialization", + applicable_routes=frozenset({_ROUTE_LIVE}), + production_route="source-to-index replay", + population=( + "source.db.raw_sessions", + "source.db.raw_artifacts", + "source.db.raw_hook_events", + "index.db.sessions", + "index.db.messages", + "index.db.blocks", + "index.db.attachment_refs", + ), + owned_reference="test_deleted_source_file_without_retained_bytes_trips_source_conservation", + check=lambda: _check_source_conservation(archive_root, sample_limit), + ), _declared_owner( name="hook-authority-topology-conflict", semantic_owner="topology", @@ -2377,7 +2410,7 @@ def _check_planner_stats( return ArchiveVerificationCheck( name="planner-stats", status=OutcomeStatus.OK, - summary="sqlite_stat1 covers blocks/messages/action_pairs", + summary="sqlite_stat1 covers blocks/messages/session_links/action_pairs", evidence={"covered_tables": sorted(analyzed), "missing_tables": []}, ) @@ -2711,8 +2744,8 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: """ has_census = table_exists(conn, "raw_membership_census") census_expr = "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" - valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r") - logical_cohort_expr = _logical_head_cohort_expr(conn, raw_alias="r") + valid_supersession_expr = valid_byte_duplicate_supersession_expr(conn, raw_alias="r") + logical_cohort_expr = logical_head_cohort_expr(conn, raw_alias="r") row = conn.execute( f""" WITH heads AS ( diff --git a/polylogue/maintenance/source_conservation.py b/polylogue/maintenance/source_conservation.py new file mode 100644 index 0000000000..71140e3da9 --- /dev/null +++ b/polylogue/maintenance/source_conservation.py @@ -0,0 +1,626 @@ +"""Source-to-archive conservation: every acquired source item types into one term. + +Backs the ``source-conservation`` owner check of ``verify-archive``. The +forward universe is the durable acquisition ledger (``raw_sessions``, +``raw_hook_events``, ``history_sidecars``); the reverse universe is every +index row (sessions, messages, blocks, attachment refs). Each item lands in +exactly one term, in a fixed precedence, and each term cites the rule that +explains it. An item no rule explains is an *unexplained* term and turns the +check red; a typed exclusion never does. + +Phantom sessions (polylogue-b508) are a reverse-direction term: an index +session whose only source lineage is a declared non-session artifact +(sidecar, workflow journal, tool-result fragment, metadata fragment) or whose +identity carries an artifact-derived shape. They are reported as a +current-producer failure and never deleted here. +""" + +from __future__ import annotations + +import sqlite3 +from collections.abc import Iterable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from polylogue.archive.revision_authority import logical_head_cohort_sql +from polylogue.core.json import JSONDocument, json_document +from polylogue.sources.origin_specs import ORIGIN_SPECS, OriginArtifactRule +from polylogue.storage.introspection import table_exists + +#: Identity prefixes that name provider fragments, never conversations: +#: ``toolu_`` is a tool_use block id (tool-result fragment) and ``wf_`` is a +#: workflow run id (workflow journal/snapshot). +FRAGMENT_IDENTITY_PREFIXES: tuple[str, ...] = ("toolu_", "wf_") + +#: Identity suffixes left behind when a sidecar filename stem is mistaken for +#: a session id. Each entry names the artifact kind whose declared path +#: pattern produces it. +ARTIFACT_IDENTITY_SUFFIXES: tuple[tuple[str, str], ...] = ( + (".meta", "agent_sidecar_meta"), + (".metadata", "agent_sidecar_meta"), +) + +_TERM_SOURCE_MISSING = "source_missing" +_TERM_SOURCE_LOST = "source_lost" +_TERM_MATERIALIZED = "materialized" +_TERM_REVISION_SUPERSEDED = "revision_superseded" +_TERM_BYTE_DUPLICATE = "byte_duplicate_superseded" +_TERM_PARSE_FAILURE = "parse_failure" +_TERM_VALIDATION_REJECTED = "validation_rejected" +_TERM_NON_SESSION_ARTIFACT = "non_session_artifact" +_TERM_DECODE_FAILED = "decode_failed" +_TERM_CENSUS_NON_SESSION = "census_non_session" +_TERM_UNCLASSIFIED_SHAPE = "unclassified_shape" +_TERM_PENDING = "pending" +_TERM_UNEXPLAINED = "unexplained" + +_TERM_HOOK_MATERIALIZED = "hook_session_materialized" +_TERM_HOOK_ACQUIRED = "hook_session_acquired" +_TERM_HOOK_NO_SESSION_ID = "hook_without_session_id" +_TERM_HOOK_NO_SOURCE = "hook_without_source_session" +_TERM_SIDECAR_RETAINED = "sidecar_retained" + +_TERM_SESSION_WITHOUT_RAW = "session_without_raw" +_TERM_SESSION_ORPHAN = "session_orphan" +_TERM_PHANTOM_LINEAGE = "phantom_declared_non_session_lineage" +_TERM_PHANTOM_IDENTITY = "phantom_fragment_identity" +_TERM_MESSAGE_ORPHAN = "message_orphan" +_TERM_BLOCK_ORPHAN = "block_orphan" +_TERM_ATTACHMENT_REF_ORPHAN = "attachment_ref_orphan" +_TERM_ATTACHMENT_UNREFERENCED = "attachment_unreferenced" + +_RULES: dict[str, str] = { + _TERM_SOURCE_MISSING: ("acquired source file no longer exists on disk; the archive retains its raw payload bytes"), + _TERM_SOURCE_LOST: ( + "acquired source file no longer exists on disk and no raw payload blob is retained; the bytes are gone" + ), + _TERM_MATERIALIZED: "index session carries this raw_id", + _TERM_REVISION_SUPERSEDED: "another revision of the same logical source is materialized", + _TERM_BYTE_DUPLICATE: "content-bound byte-duplicate supersession receipt names a materialized twin", + _TERM_PARSE_FAILURE: "raw_sessions.parse_error records the typed parser refusal", + _TERM_VALIDATION_REJECTED: "raw_sessions.validation_status = 'failed' records the schema refusal", + _TERM_NON_SESSION_ARTIFACT: "raw_artifacts declares the item a non-session artifact kind", + _TERM_DECODE_FAILED: "raw_artifacts.decode_error records the typed decode failure", + _TERM_CENSUS_NON_SESSION: "raw_membership_census recorded a terminal non-session verdict", + _TERM_UNCLASSIFIED_SHAPE: "artifact taxonomy holds no classification (unknown/unknown); a rule is missing", + _TERM_PENDING: "acquired; convergence has not parsed it yet", + _TERM_UNEXPLAINED: "parsed without refusal, yet no index session and no exclusion rule applies", + _TERM_HOOK_MATERIALIZED: "hook event names a materialized session", + _TERM_HOOK_ACQUIRED: "hook event names an acquired raw session (typed by that raw's term)", + _TERM_HOOK_NO_SESSION_ID: "hook event carries no session identity", + _TERM_HOOK_NO_SOURCE: "hook event names a session whose file was never acquired", + _TERM_SIDECAR_RETAINED: "history sidecar is evidence for a session, never a session", + _TERM_SESSION_WITHOUT_RAW: "index session has no raw_id", + _TERM_SESSION_ORPHAN: "index session raw_id names no raw_sessions row", + _TERM_PHANTOM_LINEAGE: "session lineage is a declared non-session artifact (sidecar/journal/fragment/metadata)", + _TERM_PHANTOM_IDENTITY: "session identity carries a fragment or sidecar shape", + _TERM_MESSAGE_ORPHAN: "message names no session", + _TERM_BLOCK_ORPHAN: "block names no message", + _TERM_ATTACHMENT_REF_ORPHAN: "attachment ref names no message", + _TERM_ATTACHMENT_UNREFERENCED: "attachment has no ref and therefore no source lineage", +} + +_BLOCKING: frozenset[str] = frozenset( + { + _TERM_SOURCE_LOST, + _TERM_UNCLASSIFIED_SHAPE, + _TERM_UNEXPLAINED, + _TERM_SESSION_WITHOUT_RAW, + _TERM_SESSION_ORPHAN, + _TERM_PHANTOM_LINEAGE, + _TERM_PHANTOM_IDENTITY, + _TERM_MESSAGE_ORPHAN, + _TERM_BLOCK_ORPHAN, + _TERM_ATTACHMENT_REF_ORPHAN, + _TERM_ATTACHMENT_UNREFERENCED, + } +) + +_WARNING: frozenset[str] = frozenset({_TERM_PENDING, _TERM_HOOK_NO_SOURCE}) + + +@dataclass(frozen=True, slots=True) +class ConservationTerm: + """One typed outcome: how many items it explains and why.""" + + name: str + count: int + rule: str + blocking: bool + sample: tuple[str, ...] = () + breakdown: dict[str, int] = field(default_factory=dict) + + def to_json(self) -> JSONDocument: + return json_document( + { + "count": self.count, + "rule": self.rule, + "blocking": self.blocking, + "sample": list(self.sample), + "breakdown": dict(sorted(self.breakdown.items())), + } + ) + + +@dataclass(frozen=True, slots=True) +class SourceConservationReport: + """Both directions of the source/archive equation, every term typed.""" + + forward_total: int + hook_total: int + sidecar_total: int + session_total: int + terms: tuple[ConservationTerm, ...] + + @property + def blocking_count(self) -> int: + return sum(term.count for term in self.terms if term.blocking) + + @property + def warning_count(self) -> int: + return sum(term.count for term in self.terms if term.name in _WARNING) + + def term(self, name: str) -> ConservationTerm: + for term in self.terms: + if term.name == name: + return term + raise KeyError(name) + + def summary(self) -> str: + parts = [ + f"{self.forward_total:,} raw item(s), {self.hook_total:,} hook event(s), " + f"{self.session_total:,} index session(s)" + ] + for term in self.terms: + if term.count and term.name != _TERM_MATERIALIZED: + marker = "!" if term.blocking else "" + parts.append(f"{term.name}={term.count:,}{marker}") + return "; ".join(parts) + + def to_json(self) -> JSONDocument: + return json_document( + { + "forward_total": self.forward_total, + "hook_total": self.hook_total, + "sidecar_total": self.sidecar_total, + "session_total": self.session_total, + "blocking_count": self.blocking_count, + "warning_count": self.warning_count, + "terms": {term.name: term.to_json() for term in self.terms}, + } + ) + + +def valid_byte_duplicate_supersession_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str: + """Return the receipt predicate shared by source/index coverage checks. + + A supersession receipt is authority only when it still names the same + bytes and an index materialization of the recorded duplicate twin. Keep + the predicate in one place so backlog freshness cannot classify a receipt + differently from source-index coverage. + """ + if not table_exists(conn, "raw_byte_duplicate_supersession_receipts"): + return "0" + return f""" + EXISTS( + SELECT 1 + FROM raw_byte_duplicate_supersession_receipts receipt + JOIN raw_sessions twin ON twin.raw_id = receipt.duplicate_of_raw_id + JOIN idx_tier.sessions twin_session + ON twin_session.raw_id = twin.raw_id + AND twin_session.session_id = receipt.duplicate_of_session_id + WHERE receipt.raw_id = {raw_alias}.raw_id + AND receipt.blob_hash = {raw_alias}.blob_hash + AND receipt.blob_size = {raw_alias}.blob_size + AND twin.blob_hash = {raw_alias}.blob_hash + AND twin.blob_size = {raw_alias}.blob_size + AND twin.origin IS {raw_alias}.origin + AND twin.source_path IS {raw_alias}.source_path + AND twin.source_index IS {raw_alias}.source_index + ) + """ + + +def logical_head_cohort_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str: + """Return the durable identity used to group raw revisions into one head. + + A full-revision row retired into membership governance intentionally loses + its raw-level ``logical_source_key``. Its single retained membership key + remains the authoritative identity, so use it before the legacy + native-id/path fallback. Shared raws can hold several membership keys; + they have no one raw-level cohort and must keep that fallback instead of + being arbitrarily assigned to one member. + """ + return logical_head_cohort_sql( + conn, + raw_alias=raw_alias, + has_memberships=table_exists(conn, "raw_session_memberships"), + ) + + +def _non_session_rules_by_origin() -> dict[str, tuple[OriginArtifactRule, ...]]: + return { + spec.origin.value: tuple(rule for rule in spec.artifact_rules if rule.parse_policy != "session") + for spec in ORIGIN_SPECS + } + + +def _declared_non_session_rule( + rules_by_origin: dict[str, tuple[OriginArtifactRule, ...]], origin: str, source_path: str | None +) -> OriginArtifactRule | None: + if not source_path: + return None + for rule in rules_by_origin.get(origin, ()): + if rule.matches(source_path): + return rule + return None + + +def fragment_identity_shape(native_id: str) -> str | None: + """Return the declared fragment/sidecar shape a session identity carries, if any.""" + for prefix in FRAGMENT_IDENTITY_PREFIXES: + if native_id.startswith(prefix): + return f"prefix:{prefix}" + for suffix, kind in ARTIFACT_IDENTITY_SUFFIXES: + if native_id.endswith(suffix): + return f"suffix:{suffix}:{kind}" + return None + + +def _source_exists(archive_root: Path, source_path: str) -> bool: + path = Path(source_path) + if not path.is_absolute(): + path = archive_root / path + return path.exists() + + +def _raw_term_case(conn: sqlite3.Connection) -> tuple[str, str]: + """Return the ``heads`` CTE and the CASE expression typing every raw row.""" + has_artifacts = table_exists(conn, "raw_artifacts") + has_census = table_exists(conn, "raw_membership_census") + census_expr = "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" + kind_expr = ( + "(SELECT a.artifact_kind FROM raw_artifacts a WHERE a.raw_id = r.raw_id ORDER BY a.artifact_id LIMIT 1)" + if has_artifacts + else "NULL" + ) + support_expr = ( + "(SELECT a.support_status FROM raw_artifacts a WHERE a.raw_id = r.raw_id ORDER BY a.artifact_id LIMIT 1)" + if has_artifacts + else "NULL" + ) + parse_as_session_expr = ( + "(SELECT a.parse_as_session FROM raw_artifacts a WHERE a.raw_id = r.raw_id ORDER BY a.artifact_id LIMIT 1)" + if has_artifacts + else "NULL" + ) + retained_expr = ( + "(r.blob_hash IS NOT NULL AND EXISTS(SELECT 1 FROM blob_refs b WHERE b.blob_hash = r.blob_hash))" + if table_exists(conn, "blob_refs") + else "(r.blob_hash IS NOT NULL)" + ) + supersession_expr = valid_byte_duplicate_supersession_expr(conn, raw_alias="r") + cohort_expr = logical_head_cohort_expr(conn, raw_alias="r") + heads_cte = f""" + WITH heads AS ( + SELECT + r.raw_id, + r.origin, + r.source_path, + r.parse_error, + r.parsed_at_ms, + r.validation_status, + {census_expr} AS census_status, + {kind_expr} AS artifact_kind, + {support_expr} AS support_status, + {parse_as_session_expr} AS parse_as_session, + {supersession_expr} AS valid_supersession, + {retained_expr} AS bytes_retained, + EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id) AS self_indexed, + MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id)) + OVER (PARTITION BY r.origin, {cohort_expr}) AS any_indexed + FROM raw_sessions r + ) + """ + term_case = f""" + CASE + WHEN self_indexed = 1 THEN '{_TERM_MATERIALIZED}' + WHEN any_indexed = 1 THEN '{_TERM_REVISION_SUPERSEDED}' + WHEN valid_supersession = 1 THEN '{_TERM_BYTE_DUPLICATE}' + WHEN parse_error IS NOT NULL THEN '{_TERM_PARSE_FAILURE}' + WHEN validation_status = 'failed' THEN '{_TERM_VALIDATION_REJECTED}' + WHEN parse_as_session = 0 AND artifact_kind IS NOT NULL AND artifact_kind != 'unknown' + THEN '{_TERM_NON_SESSION_ARTIFACT}' + WHEN support_status = 'decode_failed' THEN '{_TERM_DECODE_FAILED}' + WHEN census_status IN ('non_session', 'failed') THEN '{_TERM_CENSUS_NON_SESSION}' + WHEN artifact_kind = 'unknown' THEN '{_TERM_UNCLASSIFIED_SHAPE}' + WHEN parsed_at_ms IS NULL THEN '{_TERM_PENDING}' + ELSE '{_TERM_UNEXPLAINED}' + END + """ + return heads_cte, term_case + + +def _sample(rows: Iterable[tuple[Any, ...]], limit: int) -> tuple[str, ...]: + out: list[str] = [] + for row in rows: + if len(out) >= limit: + break + out.append(str(row[0])) + return tuple(out) + + +def audit_source_conservation( + conn: sqlite3.Connection, + *, + archive_root: Path, + sample_limit: int = 10, + probe_filesystem: bool = True, +) -> SourceConservationReport: + """Type every acquired source item and every index row; ``conn`` is the + source tier with the index tier attached as ``idx_tier`` (read-only).""" + heads_cte, term_case = _raw_term_case(conn) + forward_total = int(conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone()[0]) + + typed_rows = conn.execute( + f"{heads_cte} SELECT raw_id, origin, source_path, artifact_kind, bytes_retained, {term_case} AS term FROM heads" + ).fetchall() + + counts: dict[str, int] = {} + samples: dict[str, list[str]] = {} + breakdowns: dict[str, dict[str, int]] = {} + missing_paths: dict[str, bool] = {} + for raw_id, origin, source_path, artifact_kind, bytes_retained, term in typed_rows: + if probe_filesystem: + present = missing_paths.get(source_path) + if present is None: + present = _source_exists(archive_root, str(source_path)) + missing_paths[source_path] = present + if not present: + term = _TERM_SOURCE_MISSING if bytes_retained else _TERM_SOURCE_LOST + counts[term] = counts.get(term, 0) + 1 + bucket = samples.setdefault(term, []) + if len(bucket) < sample_limit: + bucket.append(str(raw_id)) + key = str(origin) if term != _TERM_NON_SESSION_ARTIFACT else f"{origin}:{artifact_kind}" + by = breakdowns.setdefault(term, {}) + by[key] = by.get(key, 0) + 1 + + # Hook events. + hook_total = 0 + hook_counts: dict[str, int] = {} + hook_samples: dict[str, tuple[str, ...]] = {} + if table_exists(conn, "raw_hook_events"): + hook_case = f""" + CASE + WHEN h.session_native_id IS NULL THEN '{_TERM_HOOK_NO_SESSION_ID}' + WHEN EXISTS(SELECT 1 FROM idx_tier.sessions s + WHERE s.session_id = h.origin || ':' || h.session_native_id) + THEN '{_TERM_HOOK_MATERIALIZED}' + WHEN EXISTS(SELECT 1 FROM raw_sessions r + WHERE r.origin = h.origin AND r.native_id = h.session_native_id) + THEN '{_TERM_HOOK_ACQUIRED}' + ELSE '{_TERM_HOOK_NO_SOURCE}' + END + """ + for term, count in conn.execute( + f"SELECT {hook_case} AS term, COUNT(*) FROM raw_hook_events h GROUP BY term" + ).fetchall(): + hook_counts[str(term)] = int(count) + hook_total += int(count) + hook_samples[_TERM_HOOK_NO_SOURCE] = _sample( + conn.execute( + f"SELECT h.hook_event_id FROM raw_hook_events h WHERE ({hook_case}) = ? LIMIT ?", + (_TERM_HOOK_NO_SOURCE, sample_limit), + ), + sample_limit, + ) + + sidecar_total = 0 + if table_exists(conn, "history_sidecars"): + sidecar_total = int(conn.execute("SELECT COUNT(*) FROM history_sidecars").fetchone()[0]) + + # Reverse direction. + session_total = int(conn.execute("SELECT COUNT(*) FROM idx_tier.sessions").fetchone()[0]) + without_raw = conn.execute( + "SELECT session_id FROM idx_tier.sessions WHERE raw_id IS NULL ORDER BY session_id" + ).fetchall() + orphans = conn.execute( + """ + SELECT s.session_id FROM idx_tier.sessions s + WHERE s.raw_id IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM raw_sessions r WHERE r.raw_id = s.raw_id) + ORDER BY s.session_id + """ + ).fetchall() + + has_artifacts = table_exists(conn, "raw_artifacts") + parse_as_session_expr = ( + "(SELECT a.parse_as_session FROM raw_artifacts a WHERE a.raw_id = r.raw_id ORDER BY a.artifact_id LIMIT 1)" + if has_artifacts + else "NULL" + ) + kind_expr = ( + "(SELECT a.artifact_kind FROM raw_artifacts a WHERE a.raw_id = r.raw_id ORDER BY a.artifact_id LIMIT 1)" + if has_artifacts + else "NULL" + ) + rules_by_origin = _non_session_rules_by_origin() + phantom_lineage: list[str] = [] + phantom_lineage_breakdown: dict[str, int] = {} + phantom_identity: list[str] = [] + phantom_identity_breakdown: dict[str, int] = {} + for session_id, native_id, origin, source_path, parse_as_session, artifact_kind in conn.execute( + f""" + SELECT s.session_id, s.native_id, s.origin, r.source_path, {parse_as_session_expr}, {kind_expr} + FROM idx_tier.sessions s + JOIN raw_sessions r ON r.raw_id = s.raw_id + """ + ): + lineage_class: str | None = None + if parse_as_session == 0 and artifact_kind is not None and artifact_kind != "unknown": + lineage_class = f"artifact:{artifact_kind}" + else: + rule = _declared_non_session_rule(rules_by_origin, str(origin), source_path) + if rule is not None: + lineage_class = f"rule:{rule.kind}" + if lineage_class is not None: + phantom_lineage.append(str(session_id)) + phantom_lineage_breakdown[lineage_class] = phantom_lineage_breakdown.get(lineage_class, 0) + 1 + continue + shape = fragment_identity_shape(str(native_id)) + if shape is not None: + phantom_identity.append(str(session_id)) + phantom_identity_breakdown[shape] = phantom_identity_breakdown.get(shape, 0) + 1 + + message_orphans = conn.execute( + """ + SELECT m.message_id FROM idx_tier.messages m + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.sessions s WHERE s.session_id = m.session_id) + LIMIT ? + """, + (sample_limit,), + ).fetchall() + message_orphan_count = int( + conn.execute( + """ + SELECT COUNT(*) FROM idx_tier.messages m + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.sessions s WHERE s.session_id = m.session_id) + """ + ).fetchone()[0] + ) + block_orphan_count = int( + conn.execute( + """ + SELECT COUNT(*) FROM idx_tier.blocks b + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.messages m WHERE m.message_id = b.message_id) + """ + ).fetchone()[0] + ) + block_orphans = conn.execute( + """ + SELECT b.block_id FROM idx_tier.blocks b + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.messages m WHERE m.message_id = b.message_id) + LIMIT ? + """, + (sample_limit,), + ).fetchall() + attachment_ref_orphan_count = 0 + attachment_ref_orphans: list[tuple[Any, ...]] = [] + attachment_unreferenced_count = 0 + attachment_unreferenced: list[tuple[Any, ...]] = [] + if table_exists(conn, "attachment_refs", schema="idx_tier"): + attachment_ref_orphan_count = int( + conn.execute( + """ + SELECT COUNT(*) FROM idx_tier.attachment_refs ar + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.messages m WHERE m.message_id = ar.message_id) + """ + ).fetchone()[0] + ) + attachment_ref_orphans = conn.execute( + """ + SELECT ar.ref_id FROM idx_tier.attachment_refs ar + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.messages m WHERE m.message_id = ar.message_id) + LIMIT ? + """, + (sample_limit,), + ).fetchall() + attachment_unreferenced_count = int( + conn.execute( + """ + SELECT COUNT(*) FROM idx_tier.attachments a + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.attachment_refs ar WHERE ar.attachment_id = a.attachment_id) + """ + ).fetchone()[0] + ) + attachment_unreferenced = conn.execute( + """ + SELECT a.attachment_id FROM idx_tier.attachments a + WHERE NOT EXISTS (SELECT 1 FROM idx_tier.attachment_refs ar WHERE ar.attachment_id = a.attachment_id) + LIMIT ? + """, + (sample_limit,), + ).fetchall() + + def _term( + name: str, count: int, sample: tuple[str, ...] = (), breakdown: dict[str, int] | None = None + ) -> ConservationTerm: + return ConservationTerm( + name=name, + count=count, + rule=_RULES[name], + blocking=name in _BLOCKING, + sample=sample, + breakdown=dict(breakdown or {}), + ) + + forward_order = ( + _TERM_SOURCE_MISSING, + _TERM_SOURCE_LOST, + _TERM_MATERIALIZED, + _TERM_REVISION_SUPERSEDED, + _TERM_BYTE_DUPLICATE, + _TERM_PARSE_FAILURE, + _TERM_VALIDATION_REJECTED, + _TERM_NON_SESSION_ARTIFACT, + _TERM_DECODE_FAILED, + _TERM_CENSUS_NON_SESSION, + _TERM_UNCLASSIFIED_SHAPE, + _TERM_PENDING, + _TERM_UNEXPLAINED, + ) + terms: list[ConservationTerm] = [ + _term(name, counts.get(name, 0), tuple(samples.get(name, ())), breakdowns.get(name)) for name in forward_order + ] + for name in (_TERM_HOOK_MATERIALIZED, _TERM_HOOK_ACQUIRED, _TERM_HOOK_NO_SESSION_ID, _TERM_HOOK_NO_SOURCE): + terms.append(_term(name, hook_counts.get(name, 0), hook_samples.get(name, ()))) + terms.append(_term(_TERM_SIDECAR_RETAINED, sidecar_total)) + terms.extend( + ( + _term(_TERM_SESSION_WITHOUT_RAW, len(without_raw), _sample(without_raw, sample_limit)), + _term(_TERM_SESSION_ORPHAN, len(orphans), _sample(orphans, sample_limit)), + _term( + _TERM_PHANTOM_LINEAGE, + len(phantom_lineage), + tuple(phantom_lineage[:sample_limit]), + phantom_lineage_breakdown, + ), + _term( + _TERM_PHANTOM_IDENTITY, + len(phantom_identity), + tuple(phantom_identity[:sample_limit]), + phantom_identity_breakdown, + ), + _term(_TERM_MESSAGE_ORPHAN, message_orphan_count, _sample(message_orphans, sample_limit)), + _term(_TERM_BLOCK_ORPHAN, block_orphan_count, _sample(block_orphans, sample_limit)), + _term( + _TERM_ATTACHMENT_REF_ORPHAN, attachment_ref_orphan_count, _sample(attachment_ref_orphans, sample_limit) + ), + _term( + _TERM_ATTACHMENT_UNREFERENCED, + attachment_unreferenced_count, + _sample(attachment_unreferenced, sample_limit), + ), + ) + ) + return SourceConservationReport( + forward_total=forward_total, + hook_total=hook_total, + sidecar_total=sidecar_total, + session_total=session_total, + terms=tuple(terms), + ) + + +__all__ = [ + "ARTIFACT_IDENTITY_SUFFIXES", + "FRAGMENT_IDENTITY_PREFIXES", + "ConservationTerm", + "SourceConservationReport", + "audit_source_conservation", + "fragment_identity_shape", + "logical_head_cohort_expr", + "valid_byte_duplicate_supersession_expr", +] diff --git a/tests/unit/devtools/test_verify_population_coverage.py b/tests/unit/devtools/test_verify_population_coverage.py new file mode 100644 index 0000000000..535d524f27 --- /dev/null +++ b/tests/unit/devtools/test_verify_population_coverage.py @@ -0,0 +1,184 @@ +"""Red twins for ``devtools gate population-coverage``. + +Anti-vacuity: removing one origin declaration, one artifact rule, or one +matrix witness turns the respective construct ``uncovered``; an unknown +artifact kind is typed unsupported evidence; the gate creates nothing. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import replace +from pathlib import Path + +import pytest + +from devtools.verify_population_coverage import ( + COVERED, + UNCOVERED, + UNSUPPORTED_DECLARED, + CoverageConstruct, + declaration_constructs, + evaluate_population_coverage, + inventory_constructs, + main, +) +from polylogue.core.enums import Origin +from polylogue.sources.origin_specs import ORIGIN_SPECS +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from tests.infra.origin_capability_matrix import load_manifest + +FIXTURE_ROOT = Path(__file__).resolve().parents[3] / "tests" / "fixtures" + + +def _by_key(constructs: tuple[CoverageConstruct, ...], family: str) -> dict[str, CoverageConstruct]: + return {construct.key: construct for construct in constructs if construct.family == family} + + +def _seed_inventory(root: Path) -> Path: + initialize_active_archive_root(root) + conn = sqlite3.connect(root / "source.db") + try: + conn.execute( + """ + INSERT INTO raw_sessions(raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms, + detected_provider) + VALUES ('raw-codex', 'codex-session', 'c1', '/src/c1.jsonl', ?, 10, 100, 'codex'), + ('raw-claude', 'claude-code-session', 'a1', '/src/agent-1.meta.json', ?, 10, 100, 'claude-code') + """, + (b"a" * 32, b"b" * 32), + ) + conn.execute( + """ + INSERT INTO raw_artifacts(artifact_id, raw_id, origin, source_path, artifact_kind, support_status, + classification_reason, parse_as_session, first_observed_at_ms, + last_observed_at_ms) + VALUES ('art-codex', 'raw-codex', 'codex-session', '/src/c1.jsonl', 'session_record_stream', + 'supported_parseable', 'test', 1, 100, 100), + ('art-claude', 'raw-claude', 'claude-code-session', '/src/agent-1.meta.json', + 'agent_sidecar_meta', 'recognized_unparsed', 'test', 0, 100, 100) + """ + ) + conn.commit() + finally: + conn.close() + return root / "source.db" + + +def test_every_declared_origin_is_witnessed_or_declared_unsupported() -> None: + constructs = declaration_constructs() + assert constructs + assert not [c for c in constructs if c.status == UNCOVERED] + statuses = {c.key: c.status for c in constructs} + assert statuses[Origin.CODEX_SESSION.value] == COVERED + assert statuses[Origin.BEADS_ISSUE.value] == UNSUPPORTED_DECLARED + + +def test_seeded_inventory_is_fully_covered(tmp_path: Path) -> None: + constructs = inventory_constructs(_seed_inventory(tmp_path)) + assert not [c for c in constructs if c.status == UNCOVERED], constructs + origins = _by_key(constructs, "origin") + assert origins["codex-session"].count == 1 + routes = _by_key(constructs, "detector-route") + assert routes["codex-session/codex"].status == COVERED + kinds = _by_key(constructs, "artifact-kind") + assert kinds["claude-code-session/agent_sidecar_meta/recognized_unparsed"].status == COVERED + assert kinds["claude-code-session/agent_sidecar_meta/recognized_unparsed"].witness == "attempt_meta" + assert kinds["codex-session/session_record_stream/supported_parseable"].status == COVERED + + +def test_removed_origin_declaration_turns_its_inventory_uncovered(tmp_path: Path) -> None: + """Anti-vacuity: the inventory half must consult the declarations, not the enum.""" + specs = tuple(spec for spec in ORIGIN_SPECS if spec.origin is not Origin.CODEX_SESSION) + constructs = inventory_constructs(_seed_inventory(tmp_path), specs=specs) + origins = _by_key(constructs, "origin") + assert origins["codex-session"].status == UNCOVERED + assert origins["codex-session"].route == "no OriginSpec" + routes = _by_key(constructs, "detector-route") + assert routes["codex-session/codex"].status == UNCOVERED + + +def test_removed_artifact_rule_turns_its_kind_uncovered(tmp_path: Path) -> None: + specs = tuple( + replace(spec, artifact_rules=()) if spec.origin is Origin.CLAUDE_CODE_SESSION else spec for spec in ORIGIN_SPECS + ) + constructs = inventory_constructs(_seed_inventory(tmp_path), specs=specs) + kinds = _by_key(constructs, "artifact-kind") + assert kinds["claude-code-session/agent_sidecar_meta/recognized_unparsed"].status == UNCOVERED + + +def test_removed_matrix_witness_turns_declaration_uncovered() -> None: + manifest = load_manifest() + stripped = replace( + manifest, + entries=tuple( + replace(entry, witnesses=()) if entry.origin is Origin.CODEX_SESSION else entry + for entry in manifest.entries + ), + ) + constructs = declaration_constructs(manifest=stripped) + statuses = {c.key: c for c in constructs} + assert statuses[Origin.CODEX_SESSION.value].status == UNCOVERED + assert statuses[Origin.CODEX_SESSION.value].witness == "no matrix witness" + + +def test_unknown_artifact_kind_is_typed_unsupported_evidence(tmp_path: Path) -> None: + source_db = _seed_inventory(tmp_path) + conn = sqlite3.connect(source_db) + try: + conn.execute( + """ + INSERT INTO raw_sessions(raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms) + VALUES ('raw-odd', 'aistudio-drive', NULL, '/src/odd.json', ?, 10, 100) + """, + (b"c" * 32,), + ) + conn.execute( + """ + INSERT INTO raw_artifacts(artifact_id, raw_id, origin, source_path, artifact_kind, support_status, + classification_reason, parse_as_session, first_observed_at_ms, + last_observed_at_ms) + VALUES ('art-odd', 'raw-odd', 'aistudio-drive', '/src/odd.json', 'unknown', 'unknown', 'test', 0, 100, 100) + """ + ) + conn.commit() + finally: + conn.close() + report = evaluate_population_coverage(tmp_path) + assert not report.ok + assert [c.key for c in report.uncovered] == ["aistudio-drive/unknown/unknown"] + assert report.uncovered[0].route == "no artifact declaration" + + +def test_gate_reports_static_only_without_an_archive_and_writes_nothing( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + before = sorted(path for path in FIXTURE_ROOT.rglob("*") if path.is_file()) + assert main(["--archive-root", str(tmp_path / "absent"), "--json"]) == 0 + payload = capsys.readouterr().out + assert '"inventory_evaluated": false' in payload + assert main(["--archive-root", str(tmp_path / "absent")]) == 0 + assert "not evaluated" in capsys.readouterr().out + after = sorted(path for path in FIXTURE_ROOT.rglob("*") if path.is_file()) + assert before == after + + +def test_gate_exit_code_follows_inventory_coverage(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + _seed_inventory(tmp_path) + assert main(["--archive-root", str(tmp_path)]) == 0 + assert "PASS" in capsys.readouterr().out + conn = sqlite3.connect(tmp_path / "source.db") + try: + conn.execute( + """ + INSERT INTO raw_sessions(raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms, + detected_provider) + VALUES ('raw-mis', 'codex-session', 'x', '/src/x.jsonl', ?, 10, 100, 'chatgpt') + """, + (b"d" * 32,), + ) + conn.commit() + finally: + conn.close() + assert main(["--archive-root", str(tmp_path)]) == 1 + assert "UNCOVERED detector-route codex-session/chatgpt" in capsys.readouterr().out diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index bd1e0dcad1..7425a2a22e 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -239,13 +239,11 @@ def _seed_coherent_archive(root: Path) -> None: index_conn.execute( """ INSERT INTO blocks(message_id, session_id, position, block_type, text) - VALUES ('codex-session:session:0.0', 'codex-session:session', 0, 'text', 'hello world') + VALUES ('codex-session:session:p:0.0', 'codex-session:session', 0, 'text', 'hello world') """ ) index_conn.commit() - index_conn.execute("ANALYZE blocks") - index_conn.execute("ANALYZE messages") - index_conn.execute("ANALYZE session_links") + index_conn.execute("ANALYZE") index_conn.commit() finally: index_conn.close() diff --git a/tests/unit/maintenance/test_source_conservation.py b/tests/unit/maintenance/test_source_conservation.py new file mode 100644 index 0000000000..ff996a6fa3 --- /dev/null +++ b/tests/unit/maintenance/test_source_conservation.py @@ -0,0 +1,449 @@ +"""Red twins for the ``source-conservation`` owner check of ``verify-archive``. + +Each test names the mutation that would make it vacuous: a deleted source +file, an injected unadmitted index row, a session materialized from a +declared non-session artifact, a fragment-shaped identity, a parsed raw that +no rule explains. The archive is built through the production tier bootstrap +with real source files under ``tmp_path``; no ambient data is read. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from polylogue.core.outcomes import OutcomeStatus +from polylogue.maintenance.archive_verification import ( + ArchiveVerificationCheck, + ArchiveVerificationReport, + archive_verification_names_for_route, + verify_archive, +) +from polylogue.maintenance.source_conservation import ( + FRAGMENT_IDENTITY_PREFIXES, + fragment_identity_shape, +) +from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin +from polylogue.storage.blob_store import BlobStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + +CHECK = "source-conservation" + + +def _check(report: ArchiveVerificationReport) -> ArchiveVerificationCheck: + matches = [c for c in report.checks if c.name == CHECK] + assert len(matches) == 1 + match = matches[0] + assert isinstance(match, ArchiveVerificationCheck) + return match + + +def _terms(check: ArchiveVerificationCheck) -> dict[str, dict[str, object]]: + terms = check.evidence["terms"] + assert isinstance(terms, dict) + return terms + + +def _count(check: ArchiveVerificationCheck, term: str) -> int: + value = _terms(check)[term]["count"] + assert isinstance(value, int) + return value + + +def _write_source(root: Path, name: str, payload: bytes) -> Path: + path = root / "sources" / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload) + return path + + +def _insert_raw( + conn: sqlite3.Connection, + *, + raw_id: str, + origin: str, + native_id: str | None, + source_path: Path, + blob_hash: str, + parsed: bool, +) -> None: + conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms, parsed_at_ms + ) VALUES (?, ?, ?, ?, ?, 10, 100, ?) + """, + (raw_id, origin, native_id, str(source_path), bytes.fromhex(blob_hash), 100 if parsed else None), + ) + conn.execute( + """ + INSERT INTO blob_refs(blob_hash, ref_id, ref_type, source_path, size_bytes, acquired_at_ms) + VALUES (?, ?, 'raw_payload', ?, 10, 100) + """, + (bytes.fromhex(blob_hash), raw_id, str(source_path)), + ) + + +def _insert_artifact( + conn: sqlite3.Connection, + *, + raw_id: str, + origin: str, + source_path: Path, + kind: str, + support: str, + parse_as_session: bool, +) -> None: + conn.execute( + """ + INSERT INTO raw_artifacts( + artifact_id, raw_id, origin, source_path, artifact_kind, support_status, + classification_reason, parse_as_session, first_observed_at_ms, last_observed_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, 'test', ?, 100, 100) + """, + (f"artifact-{raw_id}", raw_id, origin, str(source_path), kind, support, int(parse_as_session)), + ) + + +def _insert_session(conn: sqlite3.Connection, *, origin: str, native_id: str, raw_id: str | None) -> str: + conn.execute( + """ + INSERT INTO sessions( + native_id, origin, raw_id, parser_fingerprint, lowering_fingerprint, content_hash, message_count + ) VALUES (?, ?, ?, ?, ?, ?, 1) + """, + (native_id, origin, raw_id, parser_fingerprint_for_origin(origin), lowering_fingerprint(), b"s" * 32), + ) + session_id = f"{origin}:{native_id}" + conn.execute( + """ + INSERT INTO messages(session_id, position, role, material_origin, content_hash) + VALUES (?, 0, 'user', 'human_authored', ?) + """, + (session_id, b"m" * 32), + ) + conn.execute( + """ + INSERT INTO blocks(message_id, session_id, position, block_type, text) + VALUES (?, ?, 0, 'text', 'hello world') + """, + (f"{session_id}:p:0.0", session_id), + ) + return session_id + + +def _seed(root: Path) -> tuple[Path, Path]: + """One materialized session, one declared non-session sidecar; both sources on disk.""" + initialize_active_archive_root(root) + store = BlobStore(root / "blob") + session_source = _write_source(root, "session.jsonl", b"session payload") + sidecar_source = _write_source(root, "subagents/agent-1.meta.json", b"{}") + session_hash = store.write_from_bytes(b"session payload")[0] + sidecar_hash = store.write_from_bytes(b"{}")[0] + source_conn = sqlite3.connect(root / "source.db") + try: + _insert_raw( + source_conn, + raw_id="raw-session", + origin="claude-code-session", + native_id="session", + source_path=session_source, + blob_hash=session_hash, + parsed=True, + ) + _insert_artifact( + source_conn, + raw_id="raw-session", + origin="claude-code-session", + source_path=session_source, + kind="session_record_stream", + support="supported_parseable", + parse_as_session=True, + ) + _insert_raw( + source_conn, + raw_id="raw-sidecar", + origin="claude-code-session", + native_id=None, + source_path=sidecar_source, + blob_hash=sidecar_hash, + parsed=False, + ) + _insert_artifact( + source_conn, + raw_id="raw-sidecar", + origin="claude-code-session", + source_path=sidecar_source, + kind="agent_sidecar_meta", + support="recognized_unparsed", + parse_as_session=False, + ) + source_conn.commit() + finally: + source_conn.close() + index_conn = sqlite3.connect(root / "index.db") + try: + _insert_session(index_conn, origin="claude-code-session", native_id="session", raw_id="raw-session") + index_conn.commit() + finally: + index_conn.close() + return session_source, sidecar_source + + +def _run(root: Path) -> ArchiveVerificationCheck: + return _check(verify_archive(root, checks=(CHECK,))) + + +def test_source_conservation_is_declared_for_the_live_route() -> None: + assert CHECK in archive_verification_names_for_route("live-archive") + + +def test_coherent_archive_types_every_item_and_is_green(tmp_path: Path) -> None: + _seed(tmp_path) + check = _run(tmp_path) + assert check.status is OutcomeStatus.OK, check.summary + assert _count(check, "materialized") == 1 + assert _count(check, "non_session_artifact") == 1 + assert _terms(check)["non_session_artifact"]["breakdown"] == {"claude-code-session:agent_sidecar_meta": 1} + assert check.evidence["forward_total"] == 2 + assert check.evidence["blocking_count"] == 0 + assert all(term["rule"] for term in _terms(check).values()) + + +def test_deleted_source_file_retypes_the_raw_as_source_missing(tmp_path: Path) -> None: + """Anti-vacuity: without the on-disk probe the raw stays typed ``materialized``. + + The source file is gone but its raw payload bytes are still retained, so + the content is conserved: the term is accounting, not a blocker. + """ + session_source, _ = _seed(tmp_path) + assert _count(_run(tmp_path), "source_missing") == 0 + session_source.unlink() + check = _run(tmp_path) + assert check.status is OutcomeStatus.OK, check.summary + assert _count(check, "source_missing") == 1 + assert _terms(check)["source_missing"]["sample"] == ["raw-session"] + assert _terms(check)["source_missing"]["blocking"] is False + assert _count(check, "materialized") == 0 + assert _count(check, "source_lost") == 0 + + +def test_deleted_source_file_without_retained_bytes_trips_source_conservation(tmp_path: Path) -> None: + """Anti-vacuity: without the retained-bytes join this reads as the non-blocking term. + + Both the acquired file and the raw payload blob ref are gone, so nothing + in the archive holds the bytes any more. + """ + session_source, _ = _seed(tmp_path) + session_source.unlink() + source_conn = sqlite3.connect(tmp_path / "source.db") + try: + source_conn.execute("DELETE FROM blob_refs WHERE ref_id = 'raw-session'") + source_conn.commit() + finally: + source_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.ERROR + assert _count(check, "source_lost") == 1 + assert _terms(check)["source_lost"]["sample"] == ["raw-session"] + assert _count(check, "source_missing") == 0 + assert "source_lost:raw-session" in check.details + + +def test_injected_unadmitted_session_trips_source_conservation(tmp_path: Path) -> None: + """Anti-vacuity: without the reverse join an index row with no raw is invisible.""" + _seed(tmp_path) + index_conn = sqlite3.connect(tmp_path / "index.db") + try: + _insert_session(index_conn, origin="codex-session", native_id="ghost", raw_id="raw-never-acquired") + _insert_session(index_conn, origin="codex-session", native_id="rawless", raw_id=None) + index_conn.commit() + finally: + index_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.ERROR + assert _count(check, "session_orphan") == 1 + assert _terms(check)["session_orphan"]["sample"] == ["codex-session:ghost"] + assert _count(check, "session_without_raw") == 1 + assert check.count == 2 + + +def test_session_from_declared_non_session_artifact_is_a_phantom(tmp_path: Path) -> None: + """polylogue-b508: lineage, not filename, makes the phantom; the row is reported, never deleted.""" + _seed(tmp_path) + index_conn = sqlite3.connect(tmp_path / "index.db") + try: + phantom = _insert_session(index_conn, origin="claude-code-session", native_id="agent-1", raw_id="raw-sidecar") + index_conn.commit() + finally: + index_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.ERROR + assert _count(check, "phantom_declared_non_session_lineage") == 1 + assert _terms(check)["phantom_declared_non_session_lineage"]["breakdown"] == {"artifact:agent_sidecar_meta": 1} + assert _terms(check)["phantom_declared_non_session_lineage"]["sample"] == [phantom] + # The sidecar raw is now materialized (by the phantom) and no longer a non-session exclusion. + assert _count(check, "non_session_artifact") == 0 + index_conn = sqlite3.connect(tmp_path / "index.db") + try: + assert index_conn.execute("SELECT COUNT(*) FROM sessions WHERE session_id = ?", (phantom,)).fetchone()[0] == 1 + finally: + index_conn.close() + + +def test_session_with_declared_rule_path_is_a_phantom_without_artifact_row(tmp_path: Path) -> None: + """The origin's artifact rules classify lineage when raw_artifacts holds no row.""" + _seed(tmp_path) + journal = _write_source(tmp_path, "subagents/workflows/run-1/journal.jsonl", b"{}") + blob_hash = BlobStore(tmp_path / "blob").write_from_bytes(b"journal")[0] + source_conn = sqlite3.connect(tmp_path / "source.db") + try: + _insert_raw( + source_conn, + raw_id="raw-journal", + origin="claude-code-session", + native_id="run-1", + source_path=journal, + blob_hash=blob_hash, + parsed=True, + ) + source_conn.commit() + finally: + source_conn.close() + index_conn = sqlite3.connect(tmp_path / "index.db") + try: + _insert_session(index_conn, origin="claude-code-session", native_id="run-1", raw_id="raw-journal") + index_conn.commit() + finally: + index_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.ERROR + assert _terms(check)["phantom_declared_non_session_lineage"]["breakdown"] == {"rule:workflow_journal": 1} + + +def test_fragment_shaped_identity_is_a_phantom(tmp_path: Path) -> None: + _seed(tmp_path) + fragment = _write_source(tmp_path, "fragment.jsonl", b"fragment") + blob_hash = BlobStore(tmp_path / "blob").write_from_bytes(b"fragment")[0] + source_conn = sqlite3.connect(tmp_path / "source.db") + try: + _insert_raw( + source_conn, + raw_id="raw-fragment", + origin="claude-code-session", + native_id="toolu_01abc", + source_path=fragment, + blob_hash=blob_hash, + parsed=True, + ) + source_conn.commit() + finally: + source_conn.close() + index_conn = sqlite3.connect(tmp_path / "index.db") + try: + _insert_session(index_conn, origin="claude-code-session", native_id="toolu_01abc", raw_id="raw-fragment") + index_conn.commit() + finally: + index_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.ERROR + assert _terms(check)["phantom_fragment_identity"]["breakdown"] == {"prefix:toolu_": 1} + + +def test_fragment_identity_shapes_cover_each_declared_prefix_and_meta_suffix() -> None: + for prefix in FRAGMENT_IDENTITY_PREFIXES: + assert fragment_identity_shape(f"{prefix}x") == f"prefix:{prefix}" + assert fragment_identity_shape("agent-af4e.meta") == "suffix:.meta:agent_sidecar_meta" + assert fragment_identity_shape("5ecdb160-agent-af4e") is None + + +def test_parsed_raw_without_session_or_rule_is_unexplained(tmp_path: Path) -> None: + _seed(tmp_path) + stray = _write_source(tmp_path, "stray.json", b"{}") + blob_hash = BlobStore(tmp_path / "blob").write_from_bytes(b"stray")[0] + source_conn = sqlite3.connect(tmp_path / "source.db") + try: + _insert_raw( + source_conn, + raw_id="raw-stray", + origin="aistudio-drive", + native_id=None, + source_path=stray, + blob_hash=blob_hash, + parsed=True, + ) + source_conn.commit() + finally: + source_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.ERROR + assert _count(check, "unexplained") == 1 + assert _terms(check)["unexplained"]["sample"] == ["raw-stray"] + + +def test_unparsed_raw_is_pending_and_only_a_warning(tmp_path: Path) -> None: + _seed(tmp_path) + fresh = _write_source(tmp_path, "fresh.jsonl", b"fresh") + blob_hash = BlobStore(tmp_path / "blob").write_from_bytes(b"fresh")[0] + source_conn = sqlite3.connect(tmp_path / "source.db") + try: + _insert_raw( + source_conn, + raw_id="raw-fresh", + origin="codex-session", + native_id="fresh", + source_path=fresh, + blob_hash=blob_hash, + parsed=False, + ) + source_conn.commit() + finally: + source_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.WARNING + assert _count(check, "pending") == 1 + assert check.evidence["blocking_count"] == 0 + + +def test_parse_failure_is_a_typed_exclusion(tmp_path: Path) -> None: + _seed(tmp_path) + broken = _write_source(tmp_path, "broken.json", b"{") + blob_hash = BlobStore(tmp_path / "blob").write_from_bytes(b"{")[0] + source_conn = sqlite3.connect(tmp_path / "source.db") + try: + _insert_raw( + source_conn, + raw_id="raw-broken", + origin="chatgpt-export", + native_id="broken", + source_path=broken, + blob_hash=blob_hash, + parsed=True, + ) + source_conn.execute("UPDATE raw_sessions SET parse_error = 'transform: boom' WHERE raw_id = 'raw-broken'") + source_conn.commit() + finally: + source_conn.close() + check = _run(tmp_path) + assert check.status is OutcomeStatus.OK + assert _count(check, "parse_failure") == 1 + + +def test_check_json_carries_every_term_with_its_rule(tmp_path: Path) -> None: + _seed(tmp_path) + payload = _run(tmp_path).to_json() + evidence = payload["evidence"] + assert isinstance(evidence, dict) + terms = evidence["terms"] + assert isinstance(terms, dict) + assert { + "materialized", + "source_missing", + "source_lost", + "unexplained", + "phantom_declared_non_session_lineage", + } <= set(terms) + for term in terms.values(): + assert isinstance(term, dict) + assert isinstance(term["rule"], str) and term["rule"] + assert isinstance(term["blocking"], bool)