diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index f4498fe..8049ca1 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -65,6 +65,18 @@ The simple retained-row joins currently reproduce: - 4,878 of 4,916 canonical edge rows can be accounted for by a staged relation; historical duplicate multiplicity still needs an explicit replay rule. +Re-run this aggregate, read-only coverage audit against any restored local +clone: + +```bash +.venv/bin/python ops/audit_kb_rebuild_coverage.py \ + --container \ + --database teleo +``` + +The auditor emits no claim bodies or source excerpts and separates snapshot +recovery from source-derived recompilation readiness. + The remaining gaps are concrete rather than mysterious: - 30 claims were created after or outside the mapped import; diff --git a/docs/reports/leo-working-state-20260709/kb-source-recompile-coverage-20260713.md b/docs/reports/leo-working-state-20260709/kb-source-recompile-coverage-20260713.md new file mode 100644 index 0000000..a9062c1 --- /dev/null +++ b/docs/reports/leo-working-state-20260709/kb-source-recompile-coverage-20260713.md @@ -0,0 +1,71 @@ +# KB Source Recompile Coverage + +Generated UTC: `2026-07-13T08:51:50.309130+00:00` + +Status: **partial** + +This audit ran inside a repeatable-read, read-only transaction against a fresh +restored canonical database. It emitted aggregate counts only: no claim bodies, +source excerpts, credentials, or database writes. The disposable container was +removed and a later Docker inspection confirmed it was absent. + +## What Can Be Rebuilt From Retained Import State + +| Surface | Covered | Canonical | Coverage | +|---|---:|---:|---:| +| Claims with canonical mappings | 1,807 | 1,837 | 98.37% | +| Sources with canonical mappings | 4,145 | 4,145 | 100.00% | +| Eligible mapped claim field projection | 1,807 | 1,807 | 100.00% | +| Eligible staged source field projection | 3,861 | 3,861 | 100.00% | +| Evidence links from staged claim-source rows | 4,254 | 4,670 | 91.09% | +| Edge rows accounted for by resolved staged relations | 4,878 | 4,916 | 99.23% | +| Applied proposals with replayable apply payload | 1 | 2 | 50.00% | + +The claim projection compares type, title-to-text, status, confidence, tags, +creator, and applicable legacy creation time. The source projection compares +type, URL, storage path, excerpt, hash, capture time, and creator. Relation +coverage does not by itself reconstruct row IDs, timestamps, creators, or +historical duplicate multiplicity. + +## Retained Import State + +- Import runs: `2` +- Inventory items: `11,624` (`5,812` per run) +- Staged claims: `2,277` +- Staged sources: `4,715` +- Staged claim-source rows: `5,195` +- Staged claim-edge rows: `6,593` +- Canonical mappings: `5,952` + +Both imports point to the same retained 5,812-file Forgejo-era workspace +inventory. The second run reflects the corrected classifier mapping. + +## Exact Remaining Gaps + +- `30` canonical claims have no canonical mapping. +- `284` source mappings do not resolve to a staged source in their recorded + first import run. +- `66` distinct staged claim keys and `85` staged source keys have no canonical + mapping. +- `155` staged claim-source rows cannot resolve both mapped endpoints. +- `416` canonical evidence links have no exact staged relation projection. +- `700` staged edges are not resolved; another `152` resolved staged edges + cannot resolve both mapped endpoints. +- `38` canonical edge rows have no matching resolved staged relation. +- `1` applied proposal has no non-empty replayable `apply_payload`. + +## Command + +```bash +.venv/bin/python ops/audit_kb_rebuild_coverage.py \ + --container \ + --database teleo +``` + +## Interpretation + +Fast exact snapshot recovery is complete and separately proven. Source-derived +recompilation is close for the original mapped claim/source graph, but it is not +yet a complete blank-database rebuild. The practical next step is to preserve +the current snapshot as genesis, backfill the listed gaps, and require every +future accepted change to carry a replayable apply payload and row receipt. diff --git a/ops/audit_kb_rebuild_coverage.py b/ops/audit_kb_rebuild_coverage.py new file mode 100644 index 0000000..de79442 --- /dev/null +++ b/ops/audit_kb_rebuild_coverage.py @@ -0,0 +1,1069 @@ +#!/usr/bin/env python3 +"""Audit how much of the canonical KB can be rebuilt from staged source data. + +The auditor only issues aggregate SELECT queries inside a repeatable-read, +read-only transaction. It reports external snapshot recovery separately from +source-derived recompilation because those are materially different guarantees. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +SAFE_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\Z") + +REQUIRED_COLUMNS: dict[tuple[str, str], tuple[str, ...]] = { + ("public", "agents"): ("id", "handle"), + ("public", "claims"): ( + "id", + "type", + "text", + "status", + "confidence", + "tags", + "created_by", + "superseded_by", + "created_at", + "updated_at", + ), + ("public", "sources"): ( + "id", + "source_type", + "url", + "storage_path", + "excerpt", + "hash", + "captured_at", + "created_by", + "created_at", + ), + ("public", "claim_evidence"): ("claim_id", "source_id", "role", "weight", "created_at"), + ("public", "claim_edges"): ("from_claim", "to_claim", "edge_type", "created_at"), + ("kb_stage", "canonical_mappings"): ("legacy_key", "target_table", "target_id", "first_run_id"), + ("kb_stage", "import_runs"): ( + "id", + "label", + "repo_path", + "inventory_path", + "started_at", + "completed_at", + ), + ("kb_stage", "inventory_items"): ("run_id",), + ("kb_stage", "staged_claims"): ( + "run_id", + "path", + "title", + "body", + "claim_type", + "status", + "confidence_numeric", + "tags", + "created_by_handle", + "created_at_legacy", + "source_ref", + "sourced_from", + "warnings", + "raw_frontmatter", + ), + ("kb_stage", "staged_sources"): ( + "run_id", + "source_key", + "path", + "source_type", + "url", + "storage_path", + "title", + "excerpt", + "hash", + "captured_at", + "created_by_handle", + "warnings", + "raw_frontmatter", + ), + ("kb_stage", "staged_claim_sources"): ("run_id", "claim_path", "source_key", "role", "weight"), + ("kb_stage", "staged_claim_edges"): ( + "run_id", + "from_path", + "to_path", + "edge_type", + "resolution_state", + ), + ("kb_stage", "kb_proposals"): ("status", "payload", "applied_at"), +} + + +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str + stderr: str + + +CommandRunner = Callable[[Sequence[str], str, int], CommandResult] + + +def _sql_values(rows: Sequence[Sequence[str]]) -> str: + return ",\n ".join( + "(" + ", ".join("'" + value.replace("'", "''") + "'" for value in row) + ")" for row in rows + ) + + +def build_audit_sql() -> str: + relation_rows = [(schema, table) for schema, table in REQUIRED_COLUMNS] + column_rows = [ + (schema, table, column) for (schema, table), columns in REQUIRED_COLUMNS.items() for column in columns + ] + relation_values = _sql_values(relation_rows) + column_values = _sql_values(column_rows) + + return f"""\ +\\set ON_ERROR_STOP on +BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY; + +WITH required_relations(schema_name, table_name) AS ( + VALUES + {relation_values} +), +required_columns(schema_name, table_name, column_name) AS ( + VALUES + {column_values} +) +SELECT ( + NOT EXISTS ( + SELECT 1 + FROM required_relations r + WHERE to_regclass(r.schema_name || '.' || r.table_name) IS NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM required_columns r + LEFT JOIN information_schema.columns c + ON c.table_schema = r.schema_name + AND c.table_name = r.table_name + AND c.column_name = r.column_name + WHERE c.column_name IS NULL + ) +) AS schema_ok +\\gset + +\\if :schema_ok +WITH claim_map AS ( + SELECT legacy_key, target_id, first_run_id + FROM kb_stage.canonical_mappings + WHERE target_table = 'claims' +), +source_map AS ( + SELECT legacy_key, target_id, first_run_id + FROM kb_stage.canonical_mappings + WHERE target_table = 'sources' +), +mapped_claim_stage AS ( + SELECT + cm.target_id, + sc.*, + c.type AS canonical_type, + c.text AS canonical_text, + c.status AS canonical_status, + c.confidence AS canonical_confidence, + c.tags AS canonical_tags, + c.created_by AS canonical_created_by, + c.superseded_by AS canonical_superseded_by, + c.created_at AS canonical_created_at, + c.updated_at AS canonical_updated_at, + actor.id AS staged_actor_id + FROM claim_map cm + JOIN kb_stage.staged_claims sc + ON sc.run_id = cm.first_run_id + AND sc.path = cm.legacy_key + JOIN public.claims c ON c.id = cm.target_id + LEFT JOIN public.agents actor ON actor.handle = sc.created_by_handle +), +mapped_source_stage AS ( + SELECT + sm.target_id, + ss.*, + s.source_type AS canonical_source_type, + s.url AS canonical_url, + s.storage_path AS canonical_storage_path, + s.excerpt AS canonical_excerpt, + s.hash AS canonical_hash, + s.captured_at AS canonical_captured_at, + s.created_by AS canonical_created_by, + s.created_at AS canonical_created_at, + actor.id AS staged_actor_id + FROM source_map sm + JOIN kb_stage.staged_sources ss + ON ss.run_id = sm.first_run_id + AND ss.source_key = sm.legacy_key + JOIN public.sources s ON s.id = sm.target_id + LEFT JOIN public.agents actor ON actor.handle = ss.created_by_handle +), +evidence_projected AS ( + SELECT + scs.claim_path, + scs.source_key, + cm.target_id AS claim_id, + sm.target_id AS source_id, + scs.role, + scs.weight + FROM kb_stage.staged_claim_sources scs + LEFT JOIN claim_map cm ON cm.legacy_key = scs.claim_path + LEFT JOIN source_map sm ON sm.legacy_key = scs.source_key +), +evidence_distinct AS ( + SELECT DISTINCT claim_id, source_id, role, weight + FROM evidence_projected + WHERE claim_id IS NOT NULL AND source_id IS NOT NULL +), +edge_projected AS ( + SELECT + sce.resolution_state, + from_map.target_id AS from_claim, + to_map.target_id AS to_claim, + sce.edge_type + FROM kb_stage.staged_claim_edges sce + LEFT JOIN claim_map from_map ON from_map.legacy_key = sce.from_path + LEFT JOIN claim_map to_map ON to_map.legacy_key = sce.to_path +), +edge_distinct AS ( + SELECT DISTINCT from_claim, to_claim, edge_type + FROM edge_projected + WHERE resolution_state = 'resolved' + AND from_claim IS NOT NULL + AND to_claim IS NOT NULL +), +latest_import AS ( + SELECT max(completed_at) AS completed_at + FROM kb_stage.import_runs +), +run_rows AS ( + SELECT + r.id, + r.label, + r.repo_path, + r.inventory_path, + r.started_at, + r.completed_at, + (SELECT count(*) FROM kb_stage.inventory_items i WHERE i.run_id = r.id) AS inventory_items, + (SELECT count(*) FROM kb_stage.staged_claims sc WHERE sc.run_id = r.id) AS staged_claims, + (SELECT count(*) FROM kb_stage.staged_sources ss WHERE ss.run_id = r.id) AS staged_sources, + (SELECT count(*) FROM kb_stage.staged_claim_sources scs WHERE scs.run_id = r.id) AS staged_claim_sources, + (SELECT count(*) FROM kb_stage.staged_claim_edges sce WHERE sce.run_id = r.id) AS staged_claim_edges + FROM kb_stage.import_runs r +), +proposal_status_counts AS ( + SELECT status, count(*) AS rows + FROM kb_stage.kb_proposals + GROUP BY status +) +SELECT jsonb_build_object( + 'identity', jsonb_build_object( + 'database', current_database(), + 'transaction_read_only', current_setting('transaction_read_only'), + 'server_version_num', current_setting('server_version_num') + ), + 'canonical_counts', jsonb_build_object( + 'claims', (SELECT count(*) FROM public.claims), + 'sources', (SELECT count(*) FROM public.sources), + 'claim_evidence', (SELECT count(*) FROM public.claim_evidence), + 'claim_edges', (SELECT count(*) FROM public.claim_edges) + ), + 'staging_counts', jsonb_build_object( + 'canonical_mappings', (SELECT count(*) FROM kb_stage.canonical_mappings), + 'import_runs', (SELECT count(*) FROM kb_stage.import_runs), + 'inventory_items', (SELECT count(*) FROM kb_stage.inventory_items), + 'staged_claims', (SELECT count(*) FROM kb_stage.staged_claims), + 'staged_sources', (SELECT count(*) FROM kb_stage.staged_sources), + 'staged_claim_sources', (SELECT count(*) FROM kb_stage.staged_claim_sources), + 'staged_claim_edges', (SELECT count(*) FROM kb_stage.staged_claim_edges), + 'kb_proposals', (SELECT count(*) FROM kb_stage.kb_proposals) + ), + 'mappings', jsonb_build_object( + 'claims', jsonb_build_object( + 'rows', (SELECT count(*) FROM claim_map), + 'legacy_keys', (SELECT count(DISTINCT legacy_key) FROM claim_map), + 'target_ids', (SELECT count(DISTINCT target_id) FROM claim_map), + 'missing_canonical_targets', ( + SELECT count(*) FROM claim_map cm + WHERE NOT EXISTS (SELECT 1 FROM public.claims c WHERE c.id = cm.target_id) + ), + 'with_first_run_staged_row', ( + SELECT count(*) FROM claim_map cm + WHERE EXISTS ( + SELECT 1 FROM kb_stage.staged_claims sc + WHERE sc.run_id = cm.first_run_id AND sc.path = cm.legacy_key + ) + ), + 'canonical_target_rows_covered', ( + SELECT count(DISTINCT cm.target_id) + FROM claim_map cm JOIN public.claims c ON c.id = cm.target_id + ), + 'canonical_target_rows_unmapped', ( + SELECT count(*) FROM public.claims c + WHERE NOT EXISTS (SELECT 1 FROM claim_map cm WHERE cm.target_id = c.id) + ), + 'staged_distinct_keys', (SELECT count(DISTINCT path) FROM kb_stage.staged_claims), + 'staged_distinct_keys_unmapped', ( + SELECT count(*) + FROM (SELECT DISTINCT path FROM kb_stage.staged_claims) staged + WHERE NOT EXISTS (SELECT 1 FROM claim_map cm WHERE cm.legacy_key = staged.path) + ) + ), + 'sources', jsonb_build_object( + 'rows', (SELECT count(*) FROM source_map), + 'legacy_keys', (SELECT count(DISTINCT legacy_key) FROM source_map), + 'target_ids', (SELECT count(DISTINCT target_id) FROM source_map), + 'missing_canonical_targets', ( + SELECT count(*) FROM source_map sm + WHERE NOT EXISTS (SELECT 1 FROM public.sources s WHERE s.id = sm.target_id) + ), + 'with_first_run_staged_row', ( + SELECT count(*) FROM source_map sm + WHERE EXISTS ( + SELECT 1 FROM kb_stage.staged_sources ss + WHERE ss.run_id = sm.first_run_id AND ss.source_key = sm.legacy_key + ) + ), + 'canonical_target_rows_covered', ( + SELECT count(DISTINCT sm.target_id) + FROM source_map sm JOIN public.sources s ON s.id = sm.target_id + ), + 'canonical_target_rows_unmapped', ( + SELECT count(*) FROM public.sources s + WHERE NOT EXISTS (SELECT 1 FROM source_map sm WHERE sm.target_id = s.id) + ), + 'staged_distinct_keys', (SELECT count(DISTINCT source_key) FROM kb_stage.staged_sources), + 'staged_distinct_keys_unmapped', ( + SELECT count(*) + FROM (SELECT DISTINCT source_key FROM kb_stage.staged_sources) staged + WHERE NOT EXISTS (SELECT 1 FROM source_map sm WHERE sm.legacy_key = staged.source_key) + ) + ), + 'unsupported_target_rows', ( + SELECT count(*) FROM kb_stage.canonical_mappings + WHERE target_table NOT IN ('claims', 'sources') + ) + ), + 'claim_parity', jsonb_build_object( + 'compared_fields', jsonb_build_array( + 'type', 'text_from_title', 'status', 'confidence', 'tags', 'created_by', + 'created_at_when_legacy_timestamp_exists' + ), + 'eligible_rows', (SELECT count(*) FROM mapped_claim_stage), + 'exact_projection_rows', ( + SELECT count(*) FROM mapped_claim_stage m + WHERE m.canonical_type = m.claim_type + AND m.canonical_text = m.title + AND m.canonical_status = m.status + AND m.canonical_confidence IS NOT DISTINCT FROM m.confidence_numeric + AND m.canonical_tags IS NOT DISTINCT FROM m.tags + AND m.canonical_created_by IS NOT DISTINCT FROM m.staged_actor_id + AND (m.created_at_legacy IS NULL OR m.canonical_created_at = m.created_at_legacy) + ), + 'field_matches', jsonb_build_object( + 'type', (SELECT count(*) FROM mapped_claim_stage WHERE canonical_type = claim_type), + 'text_from_title', (SELECT count(*) FROM mapped_claim_stage WHERE canonical_text = title), + 'status', (SELECT count(*) FROM mapped_claim_stage WHERE canonical_status = status), + 'confidence', ( + SELECT count(*) FROM mapped_claim_stage + WHERE canonical_confidence IS NOT DISTINCT FROM confidence_numeric + ), + 'tags', ( + SELECT count(*) FROM mapped_claim_stage + WHERE canonical_tags IS NOT DISTINCT FROM tags + ), + 'created_by', ( + SELECT count(*) FROM mapped_claim_stage + WHERE canonical_created_by IS NOT DISTINCT FROM staged_actor_id + ), + 'legacy_created_at_applicable', ( + SELECT count(*) FROM mapped_claim_stage WHERE created_at_legacy IS NOT NULL + ), + 'legacy_created_at', ( + SELECT count(*) FROM mapped_claim_stage + WHERE created_at_legacy IS NOT NULL AND canonical_created_at = created_at_legacy + ) + ), + 'projection_limits', jsonb_build_object( + 'creator_handles_without_agent_row', ( + SELECT count(*) FROM mapped_claim_stage + WHERE created_by_handle IS NOT NULL AND staged_actor_id IS NULL + ), + 'staged_body_nonempty', ( + SELECT count(*) FROM mapped_claim_stage WHERE nullif(btrim(body), '') IS NOT NULL + ), + 'staged_source_ref_nonempty', ( + SELECT count(*) FROM mapped_claim_stage WHERE nullif(btrim(source_ref), '') IS NOT NULL + ), + 'staged_sourced_from_nonempty', ( + SELECT count(*) FROM mapped_claim_stage WHERE nullif(btrim(sourced_from), '') IS NOT NULL + ), + 'staged_raw_frontmatter_nonempty', ( + SELECT count(*) FROM mapped_claim_stage WHERE raw_frontmatter <> '{{}}'::jsonb + ), + 'staged_warnings_nonempty', ( + SELECT count(*) FROM mapped_claim_stage WHERE cardinality(warnings) > 0 + ), + 'canonical_superseded_by_nonnull', ( + SELECT count(*) FROM mapped_claim_stage WHERE canonical_superseded_by IS NOT NULL + ), + 'canonical_updated_after_created', ( + SELECT count(*) FROM mapped_claim_stage WHERE canonical_updated_at > canonical_created_at + ) + ) + ), + 'source_parity', jsonb_build_object( + 'compared_fields', jsonb_build_array( + 'source_type', 'url', 'storage_path', 'excerpt', 'hash', 'captured_at', 'created_by' + ), + 'eligible_rows', (SELECT count(*) FROM mapped_source_stage), + 'exact_projection_rows', ( + SELECT count(*) FROM mapped_source_stage m + WHERE m.canonical_source_type = m.source_type + AND m.canonical_url IS NOT DISTINCT FROM m.url + AND m.canonical_storage_path IS NOT DISTINCT FROM m.storage_path + AND m.canonical_excerpt IS NOT DISTINCT FROM m.excerpt + AND m.canonical_hash IS NOT DISTINCT FROM m.hash + AND m.canonical_captured_at IS NOT DISTINCT FROM m.captured_at + AND m.canonical_created_by IS NOT DISTINCT FROM m.staged_actor_id + ), + 'field_matches', jsonb_build_object( + 'source_type', ( + SELECT count(*) FROM mapped_source_stage WHERE canonical_source_type = source_type + ), + 'url', ( + SELECT count(*) FROM mapped_source_stage WHERE canonical_url IS NOT DISTINCT FROM url + ), + 'storage_path', ( + SELECT count(*) FROM mapped_source_stage + WHERE canonical_storage_path IS NOT DISTINCT FROM storage_path + ), + 'excerpt', ( + SELECT count(*) FROM mapped_source_stage WHERE canonical_excerpt IS NOT DISTINCT FROM excerpt + ), + 'hash', ( + SELECT count(*) FROM mapped_source_stage WHERE canonical_hash IS NOT DISTINCT FROM hash + ), + 'captured_at', ( + SELECT count(*) FROM mapped_source_stage + WHERE canonical_captured_at IS NOT DISTINCT FROM captured_at + ), + 'created_by', ( + SELECT count(*) FROM mapped_source_stage + WHERE canonical_created_by IS NOT DISTINCT FROM staged_actor_id + ) + ), + 'projection_limits', jsonb_build_object( + 'creator_handles_without_agent_row', ( + SELECT count(*) FROM mapped_source_stage + WHERE created_by_handle IS NOT NULL AND staged_actor_id IS NULL + ), + 'staged_path_nonempty', ( + SELECT count(*) FROM mapped_source_stage WHERE nullif(btrim(path), '') IS NOT NULL + ), + 'staged_title_nonempty', ( + SELECT count(*) FROM mapped_source_stage WHERE nullif(btrim(title), '') IS NOT NULL + ), + 'staged_raw_frontmatter_nonempty', ( + SELECT count(*) FROM mapped_source_stage WHERE raw_frontmatter <> '{{}}'::jsonb + ), + 'staged_warnings_nonempty', ( + SELECT count(*) FROM mapped_source_stage WHERE cardinality(warnings) > 0 + ), + 'canonical_created_at_without_source_projection', (SELECT count(*) FROM mapped_source_stage) + ) + ), + 'claim_evidence_reconstruction', jsonb_build_object( + 'compared_fields', jsonb_build_array('claim_id', 'source_id', 'role', 'weight'), + 'not_replayed_fields', jsonb_build_array('created_by', 'created_at'), + 'staged_rows', (SELECT count(*) FROM evidence_projected), + 'rows_with_mapped_claim', (SELECT count(*) FROM evidence_projected WHERE claim_id IS NOT NULL), + 'rows_with_mapped_source', (SELECT count(*) FROM evidence_projected WHERE source_id IS NOT NULL), + 'rows_with_both_mapped_endpoints', ( + SELECT count(*) FROM evidence_projected WHERE claim_id IS NOT NULL AND source_id IS NOT NULL + ), + 'distinct_projected_rows', (SELECT count(*) FROM evidence_distinct), + 'projected_rows_present_in_canonical', ( + SELECT count(*) FROM evidence_distinct e + WHERE EXISTS ( + SELECT 1 FROM public.claim_evidence ce + WHERE ce.claim_id = e.claim_id + AND ce.source_id = e.source_id + AND ce.role = e.role + AND ce.weight IS NOT DISTINCT FROM e.weight + ) + ), + 'canonical_rows_covered', ( + SELECT count(*) FROM public.claim_evidence ce + WHERE EXISTS ( + SELECT 1 FROM evidence_distinct e + WHERE ce.claim_id = e.claim_id + AND ce.source_id = e.source_id + AND ce.role = e.role + AND ce.weight IS NOT DISTINCT FROM e.weight + ) + ), + 'canonical_rows_total', (SELECT count(*) FROM public.claim_evidence) + ), + 'claim_edge_reconstruction', jsonb_build_object( + 'compared_fields', jsonb_build_array('from_claim', 'to_claim', 'edge_type'), + 'not_replayed_fields', jsonb_build_array( + 'id', 'weight', 'created_by', 'created_at', 'duplicate_multiplicity' + ), + 'staged_rows', (SELECT count(*) FROM edge_projected), + 'resolution_state_resolved_rows', ( + SELECT count(*) FROM edge_projected WHERE resolution_state = 'resolved' + ), + 'resolved_rows_with_from_mapping', ( + SELECT count(*) FROM edge_projected + WHERE resolution_state = 'resolved' AND from_claim IS NOT NULL + ), + 'resolved_rows_with_to_mapping', ( + SELECT count(*) FROM edge_projected + WHERE resolution_state = 'resolved' AND to_claim IS NOT NULL + ), + 'resolved_rows_with_both_mapped_endpoints', ( + SELECT count(*) FROM edge_projected + WHERE resolution_state = 'resolved' AND from_claim IS NOT NULL AND to_claim IS NOT NULL + ), + 'distinct_projected_rows', (SELECT count(*) FROM edge_distinct), + 'projected_rows_present_in_canonical', ( + SELECT count(*) FROM edge_distinct e + WHERE EXISTS ( + SELECT 1 FROM public.claim_edges ce + WHERE ce.from_claim = e.from_claim + AND ce.to_claim = e.to_claim + AND ce.edge_type = e.edge_type + ) + ), + 'canonical_rows_covered', ( + SELECT count(*) FROM public.claim_edges ce + WHERE EXISTS ( + SELECT 1 FROM edge_distinct e + WHERE ce.from_claim = e.from_claim + AND ce.to_claim = e.to_claim + AND ce.edge_type = e.edge_type + ) + ), + 'canonical_rows_total', (SELECT count(*) FROM public.claim_edges) + ), + 'later_and_unmapped', jsonb_build_object( + 'latest_import_completed_at', (SELECT completed_at FROM latest_import), + 'canonical_created_after_latest_import', jsonb_build_object( + 'claims', CASE WHEN (SELECT completed_at FROM latest_import) IS NULL THEN NULL ELSE ( + SELECT count(*) FROM public.claims c + WHERE c.created_at > (SELECT completed_at FROM latest_import) + ) END, + 'sources', CASE WHEN (SELECT completed_at FROM latest_import) IS NULL THEN NULL ELSE ( + SELECT count(*) FROM public.sources s + WHERE s.created_at > (SELECT completed_at FROM latest_import) + ) END, + 'claim_evidence', CASE WHEN (SELECT completed_at FROM latest_import) IS NULL THEN NULL ELSE ( + SELECT count(*) FROM public.claim_evidence ce + WHERE ce.created_at > (SELECT completed_at FROM latest_import) + ) END, + 'claim_edges', CASE WHEN (SELECT completed_at FROM latest_import) IS NULL THEN NULL ELSE ( + SELECT count(*) FROM public.claim_edges ce + WHERE ce.created_at > (SELECT completed_at FROM latest_import) + ) END + ), + 'canonical_without_mapping', jsonb_build_object( + 'claims', ( + SELECT count(*) FROM public.claims c + WHERE NOT EXISTS (SELECT 1 FROM claim_map cm WHERE cm.target_id = c.id) + ), + 'sources', ( + SELECT count(*) FROM public.sources s + WHERE NOT EXISTS (SELECT 1 FROM source_map sm WHERE sm.target_id = s.id) + ) + ), + 'canonical_relations_not_reconstructible', jsonb_build_object( + 'claim_evidence', ( + SELECT count(*) FROM public.claim_evidence ce + WHERE NOT EXISTS ( + SELECT 1 FROM evidence_distinct e + WHERE ce.claim_id = e.claim_id + AND ce.source_id = e.source_id + AND ce.role = e.role + AND ce.weight IS NOT DISTINCT FROM e.weight + ) + ), + 'claim_edges', ( + SELECT count(*) FROM public.claim_edges ce + WHERE NOT EXISTS ( + SELECT 1 FROM edge_distinct e + WHERE ce.from_claim = e.from_claim + AND ce.to_claim = e.to_claim + AND ce.edge_type = e.edge_type + ) + ) + ), + 'staged_distinct_keys_without_mapping', jsonb_build_object( + 'claims', ( + SELECT count(*) FROM (SELECT DISTINCT path FROM kb_stage.staged_claims) staged + WHERE NOT EXISTS (SELECT 1 FROM claim_map cm WHERE cm.legacy_key = staged.path) + ), + 'sources', ( + SELECT count(*) FROM (SELECT DISTINCT source_key FROM kb_stage.staged_sources) staged + WHERE NOT EXISTS (SELECT 1 FROM source_map sm WHERE sm.legacy_key = staged.source_key) + ) + ) + ), + 'proposal_apply_payload', jsonb_build_object( + 'total_rows', (SELECT count(*) FROM kb_stage.kb_proposals), + 'status_counts', COALESCE( + (SELECT jsonb_object_agg(status, rows ORDER BY status) FROM proposal_status_counts), + '{{}}'::jsonb + ), + 'with_apply_payload_key', ( + SELECT count(*) FROM kb_stage.kb_proposals WHERE payload ? 'apply_payload' + ), + 'with_valid_apply_payload', ( + SELECT count(*) FROM kb_stage.kb_proposals + WHERE jsonb_typeof(payload -> 'apply_payload') = 'object' + AND payload -> 'apply_payload' <> '{{}}'::jsonb + ), + 'applied_rows', (SELECT count(*) FROM kb_stage.kb_proposals WHERE applied_at IS NOT NULL), + 'applied_with_valid_apply_payload', ( + SELECT count(*) FROM kb_stage.kb_proposals + WHERE applied_at IS NOT NULL + AND jsonb_typeof(payload -> 'apply_payload') = 'object' + AND payload -> 'apply_payload' <> '{{}}'::jsonb + ) + ), + 'import_runs', COALESCE( + (SELECT jsonb_agg(to_jsonb(run_rows) ORDER BY started_at) FROM run_rows), + '[]'::jsonb + ) +)::text; +\\else +WITH required_relations(schema_name, table_name) AS ( + VALUES + {relation_values} +), +required_columns(schema_name, table_name, column_name) AS ( + VALUES + {column_values} +), +missing_relations AS ( + SELECT schema_name || '.' || table_name AS relation + FROM required_relations + WHERE to_regclass(schema_name || '.' || table_name) IS NULL +), +missing_columns AS ( + SELECT r.schema_name || '.' || r.table_name || '.' || r.column_name AS column_name + FROM required_columns r + LEFT JOIN information_schema.columns c + ON c.table_schema = r.schema_name + AND c.table_name = r.table_name + AND c.column_name = r.column_name + WHERE c.column_name IS NULL +) +SELECT jsonb_build_object( + 'status', 'error', + 'error', 'expected_schema_absent', + 'missing_relations', COALESCE( + (SELECT jsonb_agg(relation ORDER BY relation) FROM missing_relations), + '[]'::jsonb + ), + 'missing_columns', COALESCE( + (SELECT jsonb_agg(column_name ORDER BY column_name) FROM missing_columns), + '[]'::jsonb + ) +)::text; +ROLLBACK; +\\quit 3 +\\endif + +ROLLBACK; +""" + + +def run_command(command: Sequence[str], stdin: str, timeout: int) -> CommandResult: + completed = subprocess.run( + list(command), + input=stdin, + text=True, + capture_output=True, + check=False, + timeout=timeout, + ) + return CommandResult(completed.returncode, completed.stdout, completed.stderr) + + +def _validate_name(label: str, value: str) -> None: + if not SAFE_NAME_RE.fullmatch(value): + raise ValueError(f"{label} must match {SAFE_NAME_RE.pattern}") + + +def _parse_json_output(stdout: str) -> dict[str, Any] | None: + for line in reversed(stdout.splitlines()): + candidate = line.strip() + if not candidate: + continue + try: + parsed = json.loads(candidate) + except json.JSONDecodeError: + continue + if isinstance(parsed, dict): + return parsed + return None + + +def _coverage(covered: int, total: int) -> dict[str, int | float]: + gap = max(total - covered, 0) + percent = 100.0 if total == 0 and covered == 0 else round((covered / total) * 100, 2) if total else 0.0 + return {"covered": covered, "total": total, "gap": gap, "percent": percent} + + +def _error_report( + *, + container: str, + database: str, + error: str, + details: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "schema_version": "teleo.kb_rebuild_coverage.v1", + "generated_at_utc": datetime.now(UTC).isoformat(), + "audit_status": "error", + "target": {"container": container, "database": database}, + "safety": { + "transaction_mode": "repeatable_read_read_only", + "database_mutation_attempted": False, + }, + "error": error, + "details": details or {}, + } + + +def _require_mapping(value: Any, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ValueError(f"missing or invalid aggregate section: {label}") + return value + + +def build_report(raw: dict[str, Any], *, container: str, database: str) -> dict[str, Any]: + identity = _require_mapping(raw.get("identity"), "identity") + canonical = _require_mapping(raw.get("canonical_counts"), "canonical_counts") + staging = _require_mapping(raw.get("staging_counts"), "staging_counts") + mappings = _require_mapping(raw.get("mappings"), "mappings") + claims_mapping = _require_mapping(mappings.get("claims"), "mappings.claims") + sources_mapping = _require_mapping(mappings.get("sources"), "mappings.sources") + claim_parity = _require_mapping(raw.get("claim_parity"), "claim_parity") + source_parity = _require_mapping(raw.get("source_parity"), "source_parity") + evidence = _require_mapping(raw.get("claim_evidence_reconstruction"), "claim_evidence_reconstruction") + edges = _require_mapping(raw.get("claim_edge_reconstruction"), "claim_edge_reconstruction") + later = _require_mapping(raw.get("later_and_unmapped"), "later_and_unmapped") + proposals = _require_mapping(raw.get("proposal_apply_payload"), "proposal_apply_payload") + import_runs = raw.get("import_runs") + if not isinstance(import_runs, list): + raise ValueError("missing or invalid aggregate section: import_runs") + + claim_mapping_rows = int(claims_mapping["rows"]) + source_mapping_rows = int(sources_mapping["rows"]) + claim_stage_origins = int(claims_mapping["with_first_run_staged_row"]) + source_stage_origins = int(sources_mapping["with_first_run_staged_row"]) + claim_exact = int(claim_parity["exact_projection_rows"]) + claim_eligible = int(claim_parity["eligible_rows"]) + source_exact = int(source_parity["exact_projection_rows"]) + source_eligible = int(source_parity["eligible_rows"]) + + claims_mapping_report = { + **claims_mapping, + "canonical_coverage": _coverage(int(claims_mapping["canonical_target_rows_covered"]), int(canonical["claims"])), + "first_run_stage_origin_coverage": _coverage(claim_stage_origins, claim_mapping_rows), + "staged_key_mapping_coverage": _coverage( + int(claims_mapping["staged_distinct_keys"]) - int(claims_mapping["staged_distinct_keys_unmapped"]), + int(claims_mapping["staged_distinct_keys"]), + ), + } + sources_mapping_report = { + **sources_mapping, + "canonical_coverage": _coverage( + int(sources_mapping["canonical_target_rows_covered"]), int(canonical["sources"]) + ), + "first_run_stage_origin_coverage": _coverage(source_stage_origins, source_mapping_rows), + "staged_key_mapping_coverage": _coverage( + int(sources_mapping["staged_distinct_keys"]) - int(sources_mapping["staged_distinct_keys_unmapped"]), + int(sources_mapping["staged_distinct_keys"]), + ), + } + claim_parity_report = { + **claim_parity, + "exact_projection_coverage": _coverage(claim_exact, claim_eligible), + } + source_parity_report = { + **source_parity, + "exact_projection_coverage": _coverage(source_exact, source_eligible), + } + evidence_report = { + **evidence, + "staged_projection_presence": _coverage( + int(evidence["projected_rows_present_in_canonical"]), int(evidence["distinct_projected_rows"]) + ), + "canonical_reconstruction_coverage": _coverage( + int(evidence["canonical_rows_covered"]), int(evidence["canonical_rows_total"]) + ), + } + edge_report = { + **edges, + "staged_projection_presence": _coverage( + int(edges["projected_rows_present_in_canonical"]), int(edges["distinct_projected_rows"]) + ), + "canonical_reconstruction_coverage": _coverage( + int(edges["canonical_rows_covered"]), int(edges["canonical_rows_total"]) + ), + } + proposal_report = { + **proposals, + "all_proposal_payload_coverage": _coverage( + int(proposals["with_valid_apply_payload"]), int(proposals["total_rows"]) + ), + "applied_proposal_payload_coverage": _coverage( + int(proposals["applied_with_valid_apply_payload"]), int(proposals["applied_rows"]) + ), + } + + gaps: list[dict[str, Any]] = [] + + def add_gap(code: str, count: int, explanation: str) -> None: + if count > 0: + gaps.append({"code": code, "count": count, "explanation": explanation}) + + add_gap( + "canonical_claims_without_mapping", + int(claims_mapping["canonical_target_rows_unmapped"]), + "Canonical claim rows have no canonical_mappings origin.", + ) + add_gap( + "canonical_sources_without_mapping", + int(sources_mapping["canonical_target_rows_unmapped"]), + "Canonical source rows have no canonical_mappings origin.", + ) + add_gap( + "claim_mappings_without_first_run_stage", + claim_mapping_rows - claim_stage_origins, + "Claim mappings do not resolve to the staged claim row in their recorded first import run.", + ) + add_gap( + "source_mappings_without_first_run_stage", + source_mapping_rows - source_stage_origins, + "Source mappings do not resolve to the staged source row in their recorded first import run.", + ) + add_gap( + "mapped_claim_projection_mismatch", + claim_eligible - claim_exact, + "Mapped staged claim fields do not exactly match their canonical projection.", + ) + add_gap( + "mapped_source_projection_mismatch", + source_eligible - source_exact, + "Mapped staged source fields do not exactly match their canonical projection.", + ) + add_gap( + "staged_claim_keys_without_mapping", + int(claims_mapping["staged_distinct_keys_unmapped"]), + "Distinct staged claim keys have no canonical mapping.", + ) + add_gap( + "staged_source_keys_without_mapping", + int(sources_mapping["staged_distinct_keys_unmapped"]), + "Distinct staged source keys have no canonical mapping.", + ) + add_gap( + "mapping_rows_with_missing_canonical_target", + int(claims_mapping["missing_canonical_targets"]) + int(sources_mapping["missing_canonical_targets"]), + "Canonical mappings point to target rows that are absent.", + ) + add_gap( + "unsupported_mapping_target_rows", + int(mappings["unsupported_target_rows"]), + "Canonical mappings target tables outside the audited claims/sources rebuild contract.", + ) + add_gap( + "staged_claim_source_rows_with_unmapped_endpoint", + int(evidence["staged_rows"]) - int(evidence["rows_with_both_mapped_endpoints"]), + "Staged claim-source links cannot resolve both canonical endpoints.", + ) + add_gap( + "staged_evidence_projection_missing_in_canonical", + int(evidence["distinct_projected_rows"]) - int(evidence["projected_rows_present_in_canonical"]), + "A reconstructible staged evidence tuple is absent from canonical claim_evidence.", + ) + add_gap( + "canonical_claim_evidence_not_reconstructible", + int(evidence["canonical_rows_total"]) - int(evidence["canonical_rows_covered"]), + "Canonical claim_evidence rows have no exact staged_claim_sources projection.", + ) + add_gap( + "staged_edges_not_resolved", + int(edges["staged_rows"]) - int(edges["resolution_state_resolved_rows"]), + "Staged claim edges are not marked resolved.", + ) + add_gap( + "resolved_staged_edges_with_unmapped_endpoint", + int(edges["resolution_state_resolved_rows"]) - int(edges["resolved_rows_with_both_mapped_endpoints"]), + "Resolved staged edges cannot resolve both canonical claim endpoints.", + ) + add_gap( + "staged_edge_projection_missing_in_canonical", + int(edges["distinct_projected_rows"]) - int(edges["projected_rows_present_in_canonical"]), + "A reconstructible staged edge tuple is absent from canonical claim_edges.", + ) + add_gap( + "canonical_claim_edges_not_reconstructible", + int(edges["canonical_rows_total"]) - int(edges["canonical_rows_covered"]), + "Canonical claim_edges rows have no exact resolved staged edge projection.", + ) + add_gap( + "applied_proposals_without_valid_apply_payload", + int(proposals["applied_rows"]) - int(proposals["applied_with_valid_apply_payload"]), + "Applied proposal rows cannot be replayed from a non-empty object apply_payload.", + ) + + source_status = "complete" if not gaps else "partial" + return { + "schema_version": "teleo.kb_rebuild_coverage.v1", + "generated_at_utc": datetime.now(UTC).isoformat(), + "audit_status": "complete", + "target": {"container": container, "database": database, "database_readback": identity.get("database")}, + "safety": { + "transaction_mode": "repeatable_read_read_only", + "transaction_read_only_readback": identity.get("transaction_read_only"), + "database_mutation_attempted": False, + "contains_private_bodies_or_source_excerpts": False, + }, + "exact_snapshot_recovery": { + "status": "external_snapshot_artifact_not_audited", + "basis": "A complete database snapshot preserves canonical and staging state without semantic recompilation.", + "verified_by_this_database_only_auditor": False, + "current_state_counts": {"canonical": canonical, "staging": staging}, + "proof_still_required": [ + "snapshot checksum", + "schema and constraint parity", + "restored row-count and row-hash parity", + ], + }, + "source_derived_recompile_readiness": { + "status": source_status, + "canonical_counts": canonical, + "canonical_mappings": { + "claims": claims_mapping_report, + "sources": sources_mapping_report, + "unsupported_target_rows": mappings["unsupported_target_rows"], + }, + "staged_to_canonical_field_parity": { + "claims": claim_parity_report, + "sources": source_parity_report, + }, + "claim_evidence_from_staged_claim_sources": evidence_report, + "claim_edges_from_resolved_staged_claim_edges": edge_report, + "later_and_unmapped_rows": later, + "proposal_apply_payload_coverage": proposal_report, + "import_run_metadata": import_runs, + "explicit_gaps": gaps, + }, + } + + +def audit_database( + *, + container: str, + database: str, + db_user: str = "postgres", + timeout: int = 120, + runner: CommandRunner = run_command, +) -> dict[str, Any]: + _validate_name("container", container) + _validate_name("database", database) + _validate_name("db_user", db_user) + command = [ + "docker", + "exec", + "-i", + container, + "psql", + "-X", + "-U", + db_user, + "-d", + database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + ] + try: + completed = runner(command, build_audit_sql(), timeout) + except (OSError, subprocess.TimeoutExpired) as exc: + return _error_report( + container=container, + database=database, + error="docker_psql_execution_failed", + details={"exception": type(exc).__name__, "message": str(exc)[:500]}, + ) + + parsed = _parse_json_output(completed.stdout) + if parsed and parsed.get("status") == "error": + return _error_report( + container=container, + database=database, + error=str(parsed.get("error") or "audit_query_failed"), + details={ + "missing_relations": parsed.get("missing_relations", []), + "missing_columns": parsed.get("missing_columns", []), + }, + ) + if completed.returncode != 0 or parsed is None: + return _error_report( + container=container, + database=database, + error="audit_query_failed", + details={ + "returncode": completed.returncode, + "stderr_excerpt": completed.stderr.strip()[-1000:], + }, + ) + try: + return build_report(parsed, container=container, database=database) + except (KeyError, TypeError, ValueError) as exc: + return _error_report( + container=container, + database=database, + error="malformed_audit_result", + details={"message": str(exc)}, + ) + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--container", required=True, help="Local Docker PostgreSQL container name") + parser.add_argument("--database", required=True, help="Database name inside the container") + parser.add_argument("--db-user", default="postgres", help="Database role used for the read-only transaction") + parser.add_argument("--timeout", type=int, default=120, help="Docker/psql timeout in seconds") + parser.add_argument("--compact", action="store_true", help="Emit one-line JSON instead of indented JSON") + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + args = parse_args(argv) + report = audit_database( + container=args.container, + database=args.database, + db_user=args.db_user, + timeout=args.timeout, + ) + print(json.dumps(report, indent=None if args.compact else 2, sort_keys=True)) + return 0 if report.get("audit_status") == "complete" else 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_audit_kb_rebuild_coverage.py b/tests/test_audit_kb_rebuild_coverage.py new file mode 100644 index 0000000..b083442 --- /dev/null +++ b/tests/test_audit_kb_rebuild_coverage.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import json +import re +from collections.abc import Sequence +from typing import Any + +import pytest + +from ops.audit_kb_rebuild_coverage import ( + CommandResult, + audit_database, + build_audit_sql, + build_report, +) + + +def aggregate_fixture() -> dict[str, Any]: + return { + "identity": {"database": "teleo", "transaction_read_only": "on", "server_version_num": "160004"}, + "canonical_counts": {"claims": 10, "sources": 8, "claim_evidence": 12, "claim_edges": 9}, + "staging_counts": { + "canonical_mappings": 17, + "import_runs": 1, + "inventory_items": 20, + "staged_claims": 11, + "staged_sources": 8, + "staged_claim_sources": 13, + "staged_claim_edges": 10, + "kb_proposals": 4, + }, + "mappings": { + "claims": { + "rows": 9, + "legacy_keys": 9, + "target_ids": 9, + "missing_canonical_targets": 0, + "with_first_run_staged_row": 9, + "canonical_target_rows_covered": 9, + "canonical_target_rows_unmapped": 1, + "staged_distinct_keys": 10, + "staged_distinct_keys_unmapped": 1, + }, + "sources": { + "rows": 8, + "legacy_keys": 8, + "target_ids": 8, + "missing_canonical_targets": 0, + "with_first_run_staged_row": 7, + "canonical_target_rows_covered": 8, + "canonical_target_rows_unmapped": 0, + "staged_distinct_keys": 8, + "staged_distinct_keys_unmapped": 0, + }, + "unsupported_target_rows": 0, + }, + "claim_parity": { + "compared_fields": ["type", "text_from_title"], + "eligible_rows": 9, + "exact_projection_rows": 9, + "field_matches": {"type": 9, "text_from_title": 9}, + "projection_limits": {"staged_body_nonempty": 9}, + }, + "source_parity": { + "compared_fields": ["source_type", "hash"], + "eligible_rows": 7, + "exact_projection_rows": 6, + "field_matches": {"source_type": 7, "hash": 6}, + "projection_limits": {"staged_title_nonempty": 7}, + }, + "claim_evidence_reconstruction": { + "compared_fields": ["claim_id", "source_id", "role", "weight"], + "not_replayed_fields": ["created_by", "created_at"], + "staged_rows": 13, + "rows_with_mapped_claim": 12, + "rows_with_mapped_source": 13, + "rows_with_both_mapped_endpoints": 12, + "distinct_projected_rows": 11, + "projected_rows_present_in_canonical": 11, + "canonical_rows_covered": 11, + "canonical_rows_total": 12, + }, + "claim_edge_reconstruction": { + "compared_fields": ["from_claim", "to_claim", "edge_type"], + "not_replayed_fields": ["id", "weight", "created_by", "created_at", "duplicate_multiplicity"], + "staged_rows": 10, + "resolution_state_resolved_rows": 9, + "resolved_rows_with_from_mapping": 9, + "resolved_rows_with_to_mapping": 8, + "resolved_rows_with_both_mapped_endpoints": 8, + "distinct_projected_rows": 8, + "projected_rows_present_in_canonical": 8, + "canonical_rows_covered": 8, + "canonical_rows_total": 9, + }, + "later_and_unmapped": { + "latest_import_completed_at": "2026-01-01T00:00:00+00:00", + "canonical_created_after_latest_import": { + "claims": 1, + "sources": 0, + "claim_evidence": 1, + "claim_edges": 1, + }, + "canonical_without_mapping": {"claims": 1, "sources": 0}, + "canonical_relations_not_reconstructible": {"claim_evidence": 1, "claim_edges": 1}, + "staged_distinct_keys_without_mapping": {"claims": 1, "sources": 0}, + }, + "proposal_apply_payload": { + "total_rows": 4, + "status_counts": {"applied": 2, "pending_review": 2}, + "with_apply_payload_key": 2, + "with_valid_apply_payload": 2, + "applied_rows": 2, + "applied_with_valid_apply_payload": 1, + }, + "import_runs": [ + { + "id": "run-id", + "label": "fixture-run", + "repo_path": "/workspace", + "inventory_path": "/inventory/fixture.jsonl", + "started_at": "2026-01-01T00:00:00+00:00", + "completed_at": "2026-01-01T00:01:00+00:00", + "inventory_items": 20, + "staged_claims": 11, + "staged_sources": 8, + "staged_claim_sources": 13, + "staged_claim_edges": 10, + } + ], + } + + +def test_sql_is_guarded_and_read_only() -> None: + sql = build_audit_sql() + + assert "BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY" in sql + assert "current_setting('transaction_read_only')" in sql + assert "expected_schema_absent" in sql + assert "\\if :schema_ok" in sql + assert sql.count("ROLLBACK;") >= 2 + assert not re.search(r"\b(INSERT|UPDATE|DELETE|CREATE|ALTER|DROP|TRUNCATE|COPY)\b", sql, re.IGNORECASE) + + +def test_report_separates_snapshot_recovery_from_source_recompile_and_calculates_gaps() -> None: + report = build_report(aggregate_fixture(), container="teleo-test", database="teleo") + semantic = report["source_derived_recompile_readiness"] + + assert report["audit_status"] == "complete" + assert report["exact_snapshot_recovery"]["verified_by_this_database_only_auditor"] is False + assert report["exact_snapshot_recovery"]["current_state_counts"]["canonical"]["claims"] == 10 + assert semantic["status"] == "partial" + assert semantic["canonical_mappings"]["claims"]["canonical_coverage"] == { + "covered": 9, + "total": 10, + "gap": 1, + "percent": 90.0, + } + assert semantic["canonical_mappings"]["sources"]["first_run_stage_origin_coverage"]["percent"] == 87.5 + assert semantic["claim_evidence_from_staged_claim_sources"]["canonical_reconstruction_coverage"]["percent"] == 91.67 + assert semantic["claim_evidence_from_staged_claim_sources"]["not_replayed_fields"] == [ + "created_by", + "created_at", + ] + assert "duplicate_multiplicity" in semantic["claim_edges_from_resolved_staged_claim_edges"]["not_replayed_fields"] + assert semantic["proposal_apply_payload_coverage"]["applied_proposal_payload_coverage"]["percent"] == 50.0 + gap_codes = {gap["code"] for gap in semantic["explicit_gaps"]} + assert "canonical_claims_without_mapping" in gap_codes + assert "source_mappings_without_first_run_stage" in gap_codes + assert "mapped_source_projection_mismatch" in gap_codes + assert "canonical_claim_evidence_not_reconstructible" in gap_codes + assert "canonical_claim_edges_not_reconstructible" in gap_codes + assert "applied_proposals_without_valid_apply_payload" in gap_codes + assert semantic["import_run_metadata"][0]["inventory_path"] == "/inventory/fixture.jsonl" + assert report["safety"]["contains_private_bodies_or_source_excerpts"] is False + + +def test_audit_uses_named_container_database_and_injected_runner() -> None: + captured: dict[str, Any] = {} + + def fake_runner(command: Sequence[str], stdin: str, timeout: int) -> CommandResult: + captured.update(command=list(command), stdin=stdin, timeout=timeout) + return CommandResult(0, json.dumps(aggregate_fixture()) + "\n", "") + + report = audit_database( + container="teleo-local", + database="teleo", + db_user="postgres", + timeout=45, + runner=fake_runner, + ) + + assert captured["command"] == [ + "docker", + "exec", + "-i", + "teleo-local", + "psql", + "-X", + "-U", + "postgres", + "-d", + "teleo", + "-Atq", + "-v", + "ON_ERROR_STOP=1", + ] + assert captured["timeout"] == 45 + assert "READ ONLY" in captured["stdin"] + assert report["target"]["container"] == "teleo-local" + assert report["safety"]["transaction_read_only_readback"] == "on" + + +def test_missing_schema_fails_closed_with_structured_json() -> None: + def fake_runner(_command: Sequence[str], _stdin: str, _timeout: int) -> CommandResult: + payload = { + "status": "error", + "error": "expected_schema_absent", + "missing_relations": ["kb_stage.canonical_mappings"], + "missing_columns": ["kb_stage.canonical_mappings.legacy_key"], + } + return CommandResult(3, json.dumps(payload) + "\n", "") + + report = audit_database(container="empty-postgres", database="teleo", runner=fake_runner) + + assert report["audit_status"] == "error" + assert report["error"] == "expected_schema_absent" + assert report["details"]["missing_relations"] == ["kb_stage.canonical_mappings"] + assert report["details"]["missing_columns"] == ["kb_stage.canonical_mappings.legacy_key"] + assert "source_derived_recompile_readiness" not in report + + +def test_invalid_docker_identifier_is_rejected_before_runner() -> None: + called = False + + def fake_runner(_command: Sequence[str], _stdin: str, _timeout: int) -> CommandResult: + nonlocal called + called = True + return CommandResult(0, "{}", "") + + with pytest.raises(ValueError, match="container must match"): + audit_database(container="--privileged", database="teleo", runner=fake_runner) + + assert called is False