#!/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())