"""Evidence Graph generation schema, validation, and query contracts."""

from __future__ import annotations

import hashlib
import json
import sqlite3
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
from pathlib import Path

import pytest

from tests.code_kernel_helpers import (
    basic_graph_records,
    make_normalized_analysis_for_records,
    make_unminted_verified_subclass,
    open_v3_fixture,
    snapshot_for_records,
)

V3_EXTENSION_NAMES = frozenset(
    {
        "analyzer_run",
        "run_capability",
        "analysis_scope",
        "expected_source",
        "coverage",
        "symbol_claim",
        "relationship_claim",
        "diagnostic",
        "diagnostic_related",
        "slice_activation",
        "validity",
        "analyzer_run_scope",
        "analyzer_run_publication",
        "run_capability_reverse",
        "analysis_scope_run",
        "expected_source_reverse",
        "coverage_capability",
        "symbol_identity",
        "symbol_source_span",
        "relationship_source",
        "relationship_target",
        "relationship_source_span",
        "diagnostic_source_span",
        "diagnostic_related_source_span",
        "one_selected_slice",
        "slice_run",
        "validity_symbol_once",
        "validity_relationship_once",
        "validity_diagnostic_once",
        "validity_status",
    }
)
V3_EXTENSION_SCHEMA_SHA256 = "088316ec6aa481e52bdadb4534e8ab50bd7e76d8fbf2e6395da591214544efec"

SCRIPTS = Path(__file__).resolve().parent.parent / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))


def _sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _records(source: bytes = b"def caller():\n    callee()\n# [[Decision]]\n"):
    decision_start = source.index(b"[[Decision]]")
    return {
        "sources": [
            {
                "source_id": "src-code",
                "relative_path": "src/app.py",
                "sha256": _sha(source),
                "size": len(source),
                "media_type": "text/x-python",
                "language": "python",
                "git_oid": None,
            }
        ],
        "source_bytes": {"src-code": source},
        "nodes": [
            {
                "node_id": "caller",
                "kind": "function",
                "identity_scheme": "python-qualified-name/v1",
                "identity_key": "src.app:caller",
                "metadata": {"name": "caller"},
            },
            {
                "node_id": "callee",
                "kind": "function",
                "identity_scheme": "python-qualified-name/v1",
                "identity_key": "src.app:callee",
                "metadata": {"name": "callee"},
            },
            {
                "node_id": "decision",
                "kind": "decision",
                "identity_scheme": "wiki-slug/v1",
                "identity_key": "decision",
                "metadata": {},
            },
        ],
        "occurrences": [
            {
                "occurrence_id": "occ-caller",
                "node_id": "caller",
                "source_id": "src-code",
                "role": "definition",
                "byte_start": 0,
                "byte_end": 13,
                "line_start": 1,
                "line_end": 1,
            }
        ],
        "assertions": [
            {
                "assertion_id": "call",
                "source_node_id": "caller",
                "edge_type": "CALLS",
                "target_node_id": "callee",
                "literal": None,
                "confidence": "high",
                "authority": "ai-derived",
                "resolution": "resolved",
                "extractor": "python/v1",
            },
            {
                "assertion_id": "documents",
                "source_node_id": "decision",
                "edge_type": "DOCUMENTS",
                "target_node_id": "caller",
                "literal": None,
                "confidence": "high",
                "authority": "user",
                "resolution": "resolved",
                "extractor": "wiki/v1",
            },
        ],
        "observations": [
            {
                "observation_id": "dynamic-call",
                "source_node_id": "caller",
                "edge_type": "CALLS",
                "target_text": "runtime_target",
                "reason": "dynamic_dispatch",
                "extractor": "python/v1",
            }
        ],
        "evidence": [
            {
                "evidence_id": "ev-call",
                "assertion_id": "call",
                "observation_id": None,
                "source_id": "src-code",
                "byte_start": 18,
                "byte_end": 26,
                "span_sha256": _sha(source[18:26]),
            },
            {
                "evidence_id": "ev-documents",
                "assertion_id": "documents",
                "observation_id": None,
                "source_id": "src-code",
                "byte_start": decision_start,
                "byte_end": decision_start + len(b"[[Decision]]"),
                "span_sha256": _sha(b"[[Decision]]"),
            },
        ],
        "dependencies": [
            {
                "dependency_id": "dep-code",
                "dependent_node_id": "decision",
                "dependency_node_id": "caller",
                "kind": "documents",
                "source_id": "src-code",
            }
        ],
    }


def _create(tmp_path: Path, **overrides):
    import evidence_graph

    records = _records()
    records.update(overrides)
    path = tmp_path / "evidence.sqlite3"
    evidence_graph.create_generation_database(path, **records)
    return evidence_graph.EvidenceGraph(path, state_root=tmp_path)


def test_database_schema_is_explicit_and_returned(tmp_path: Path) -> None:
    from code_intelligence import verify_native_analysis
    from evidence_graph import GraphSchema, create_generation_database

    records = basic_graph_records()
    v2 = create_generation_database(
        tmp_path / "v2.sqlite3", schema=GraphSchema.V2, **records
    )
    from repository_scope import resolve_repository_scope

    repository = tmp_path / "repository"
    repository.mkdir()
    scope = resolve_repository_scope(repository)
    verified = verify_native_analysis(
        snapshot_for_records(records), make_normalized_analysis_for_records(records, scope)
    )
    v3 = create_generation_database(
        tmp_path / "v3.sqlite3",
        schema=GraphSchema.V3,
        verified_analyses=(verified,),
        publication_generation_id="v3",
        publication_expected_active=None,
        repository_scope=scope,
        **records,
    )

    assert (v2, v3) == (GraphSchema.V2, GraphSchema.V3)
    with sqlite3.connect(tmp_path / "v2.sqlite3") as database:
        assert database.execute("PRAGMA user_version").fetchone()[0] == 2
    with sqlite3.connect(tmp_path / "v3.sqlite3") as database:
        assert database.execute("PRAGMA user_version").fetchone()[0] == 3


def test_omitted_database_schema_remains_v2_for_legacy_callers(tmp_path: Path) -> None:
    from evidence_graph import GraphSchema, create_generation_database

    selected = create_generation_database(
        tmp_path / "legacy.sqlite3", **basic_graph_records()
    )

    assert selected == GraphSchema.V2


def test_v3_has_normalized_relationships_without_unresolved_v2_assertions(
    tmp_path: Path,
) -> None:
    graph = open_v3_fixture(tmp_path)
    try:
        assert graph._database.execute(
            "SELECT count(*) FROM relationship_claim"
        ).fetchone()[0] > 0
        assert graph._database.execute(
            "SELECT count(*) FROM assertion WHERE resolution != 'resolved'"
        ).fetchone()[0] == 0
        assert graph._database.execute(
            "SELECT count(*) FROM analysis_scope s JOIN analyzer_run r USING (run_id) "
            "WHERE s.source_manifest_sha256 != r.source_manifest_sha256"
        ).fetchone()[0] == 0
    finally:
        graph.close()


def _verified_v3_database(tmp_path: Path) -> Path:
    from code_intelligence import verify_native_analysis
    from evidence_graph import GraphSchema, create_generation_database
    from repository_scope import resolve_repository_scope

    records = basic_graph_records()
    repository = tmp_path / "repository"
    repository.mkdir(exist_ok=True)
    scope = resolve_repository_scope(repository)
    verified = verify_native_analysis(
        snapshot_for_records(records), make_normalized_analysis_for_records(records, scope)
    )
    path = tmp_path / "verified-v3.sqlite3"
    create_generation_database(
        path,
        schema=GraphSchema.V3,
        verified_analyses=(verified,),
        publication_generation_id="verified-v3",
        publication_expected_active="prior",
        repository_scope=scope,
        **records,
    )
    return path


def test_v3_extension_has_exact_table_index_and_fk_signature(tmp_path: Path) -> None:
    path = _verified_v3_database(tmp_path)
    bind_slots = ",".join("?" for _ in V3_EXTENSION_NAMES)
    with sqlite3.connect(f"{path.as_uri()}?mode=ro", uri=True) as database:
        rows = database.execute(
            f"SELECT type,name,tbl_name,sql FROM sqlite_schema "
            f"WHERE name IN ({bind_slots}) ORDER BY type,name",
            sorted(V3_EXTENSION_NAMES),
        ).fetchall()
        encoded = json.dumps(rows, ensure_ascii=True, separators=(",", ":")).encode()
        assert len(rows) == 30
        assert hashlib.sha256(encoded).hexdigest() == V3_EXTENSION_SCHEMA_SHA256
        assert database.execute("PRAGMA foreign_key_check").fetchall() == []


def test_v2_rejects_verified_analysis_batches(tmp_path: Path) -> None:
    from code_intelligence import verify_native_analysis
    from evidence_graph import GraphSchema, create_generation_database

    records = basic_graph_records()
    verified = verify_native_analysis(
        snapshot_for_records(records), make_normalized_analysis_for_records(records)
    )
    with pytest.raises(ValueError, match="v3"):
        create_generation_database(
            tmp_path / "v2.sqlite3",
            schema=GraphSchema.V2,
            verified_analyses=(verified,),
            **records,
        )


def test_v3_writer_rejects_unminted_verified_subclass_before_publish(
    tmp_path: Path,
) -> None:
    from evidence_graph import GraphSchema, create_generation_database
    from repository_scope import resolve_repository_scope

    records = basic_graph_records()
    repository = tmp_path / "repository"
    repository.mkdir()
    scope = resolve_repository_scope(repository)
    path = tmp_path / "forged.sqlite3"
    with pytest.raises(TypeError, match="VerifiedAnalysisBatch"):
        create_generation_database(
            path,
            schema=GraphSchema.V3,
            verified_analyses=(make_unminted_verified_subclass(records),),
            publication_generation_id="forged",
            publication_expected_active=None,
            repository_scope=scope,
            **records,
        )
    assert not path.exists()


def test_v3_expansion_ceiling_rejects_before_database_creation(
    tmp_path: Path, monkeypatch
) -> None:
    from dataclasses import replace

    import evidence_graph
    from code_intelligence import verify_native_analysis
    from repository_scope import resolve_repository_scope

    records = basic_graph_records()
    repository = tmp_path / "repository"
    repository.mkdir()
    scope = resolve_repository_scope(repository)
    analysis = make_normalized_analysis_for_records(records)
    analysis = replace(
        analysis,
        run=replace(
            analysis.run,
            repository_id=scope.repository_id,
            checkout_id=scope.checkout_id,
        ),
    )
    verified = verify_native_analysis(snapshot_for_records(records), analysis)
    monkeypatch.setattr(evidence_graph, "MAX_VALIDATION_ROWS", 1)
    path = tmp_path / "bounded.sqlite3"

    with pytest.raises(ValueError, match="row ceiling"):
        evidence_graph.create_generation_database(
            path,
            schema=evidence_graph.GraphSchema.V3,
            verified_analyses=(verified,),
            publication_generation_id="generation",
            publication_expected_active=None,
            repository_scope=scope,
            **records,
        )
    assert not path.exists()


def _damage_v3(path: Path, damage: str) -> None:
    with sqlite3.connect(path) as database:
        if damage == "missing-scope":
            database.execute("DELETE FROM analysis_scope")
        elif damage == "extra-scope":
            empty_set = hashlib.sha256(b"[]").hexdigest()
            database.execute(
                "INSERT INTO analysis_scope SELECT 'scope:extra',run_id,"
                "source_manifest_sha256,build_target||'-extra',build_configuration,0,?,"
                "generated_sources,dependency_resolution,analyzer_support "
                "FROM analysis_scope LIMIT 1",
                (empty_set,),
            )
        elif damage == "duplicate-target-configuration":
            empty_set = hashlib.sha256(b"[]").hexdigest()
            database.execute(
                "INSERT INTO analysis_scope SELECT 'scope:duplicate',run_id,"
                "source_manifest_sha256,build_target,build_configuration,0,?,"
                "generated_sources,dependency_resolution,analyzer_support "
                "FROM analysis_scope LIMIT 1",
                (empty_set,),
            )
        elif damage == "scope-set-hash":
            database.execute(
                "UPDATE analyzer_run SET expected_scope_set_sha256=?", ("f" * 64,)
            )
        elif damage == "scope-source-manifest":
            database.execute(
                "UPDATE analysis_scope SET source_manifest_sha256=?", ("f" * 64,)
            )
        elif damage == "missing-capability":
            database.execute(
                "DELETE FROM run_capability WHERE capability="
                "(SELECT capability FROM run_capability LIMIT 1)"
            )
        elif damage == "extra-capability":
            run_id = database.execute("SELECT run_id FROM analyzer_run").fetchone()[0]
            database.execute(
                "INSERT INTO run_capability VALUES (?, 'diagnostics')", (run_id,)
            )
        elif damage == "capability-set-hash":
            database.execute(
                "UPDATE analyzer_run SET declared_capabilities_sha256=?", ("f" * 64,)
            )
        elif damage == "missing-source":
            database.execute(
                "DELETE FROM expected_source WHERE source_id="
                "(SELECT source_id FROM expected_source LIMIT 1)"
            )
        elif damage == "extra-source":
            empty_sha = hashlib.sha256(b"").hexdigest()
            database.execute(
                "INSERT INTO source VALUES "
                "('source:extra','extra.py',?,0,'text/x-python','python',NULL,X'')",
                (empty_sha,),
            )
            scope_id, run_id = database.execute(
                "SELECT scope_id,run_id FROM analysis_scope"
            ).fetchone()
            database.execute(
                "INSERT INTO expected_source VALUES (?,?, 'source:extra',?,'included')",
                (scope_id, run_id, empty_sha),
            )
        elif damage == "source-set-hash":
            database.execute(
                "UPDATE analysis_scope SET expected_source_set_sha256=?", ("f" * 64,)
            )
        elif damage == "missing-coverage":
            database.execute(
                "DELETE FROM coverage WHERE capability="
                "(SELECT capability FROM coverage LIMIT 1)"
            )
        elif damage == "extra-coverage":
            database.execute("INSERT INTO coverage SELECT * FROM coverage LIMIT 1")
        elif damage == "duplicate-selected-slice":
            database.execute(
                "INSERT INTO slice_activation SELECT slice_id||':duplicate',slice_key,"
                "run_id,scope_id,capability,1,selection_reason "
                "FROM slice_activation LIMIT 1"
            )
        elif damage == "missing-selected-slice":
            database.execute("UPDATE slice_activation SET selected=0 WHERE selected=1")
        elif damage == "invalid-validity-subject":
            database.execute(
                "INSERT INTO validity VALUES "
                "('validity:invalid',NULL,NULL,NULL,'current',NULL)"
            )
        elif damage in {"wrong-repository", "wrong-checkout"}:
            column = "repository_id" if damage == "wrong-repository" else "checkout_id"
            database.execute(f"UPDATE analyzer_run SET {column}={column}||':wrong'")
        elif damage == "wrong-publication-generation":
            database.execute(
                "UPDATE analyzer_run SET publication_generation_id='wrong-generation'"
            )
        elif damage == "wrong-expected-active":
            database.execute(
                "UPDATE analyzer_run SET publication_expected_active='wrong-active'"
            )
        elif damage in {"bad-analysis-hash", "bad-component-hash"}:
            column = "analysis_sha256" if damage == "bad-analysis-hash" else "manifest_sha256"
            database.execute(f"UPDATE analyzer_run SET {column}=?", ("f" * 64,))
        elif damage == "expected-source-hash-mismatch":
            from reliable_memory import canonical_json_bytes

            database.execute("UPDATE expected_source SET source_sha256=?", ("f" * 64,))
            scope_id = database.execute("SELECT scope_id FROM analysis_scope").fetchone()[0]
            rows = database.execute(
                "SELECT source_id,source_sha256,disposition FROM expected_source "
                "WHERE scope_id=? ORDER BY source_id",
                (scope_id,),
            ).fetchall()
            digest = hashlib.sha256(
                canonical_json_bytes(
                    [
                        {"source_id": row[0], "sha256": row[1], "disposition": row[2]}
                        for row in rows
                    ]
                )
            ).hexdigest()
            database.execute(
                "UPDATE analysis_scope SET expected_source_set_sha256=? WHERE scope_id=?",
                (digest, scope_id),
            )
        elif damage.startswith("eof-"):
            table = damage.removeprefix("eof-")
            database.execute(
                f"UPDATE {table} SET byte_end="
                f"(SELECT size + 1 FROM source WHERE source_id={table}.source_id)"
            )
        else:
            raise AssertionError(damage)


@pytest.mark.parametrize(
    "damage",
    [
        "missing-scope",
        "extra-scope",
        "duplicate-target-configuration",
        "scope-set-hash",
        "scope-source-manifest",
        "missing-capability",
        "extra-capability",
        "capability-set-hash",
        "missing-source",
        "extra-source",
        "source-set-hash",
        "missing-coverage",
        "extra-coverage",
        "duplicate-selected-slice",
        "missing-selected-slice",
        "invalid-validity-subject",
        "wrong-repository",
        "wrong-checkout",
        "wrong-publication-generation",
        "wrong-expected-active",
        "bad-analysis-hash",
        "bad-component-hash",
        "expected-source-hash-mismatch",
        "eof-symbol_claim",
        "eof-relationship_claim",
        "eof-diagnostic",
        "eof-diagnostic_related",
    ],
)
def test_v3_closed_world_storage_fails_closed(tmp_path: Path, damage: str) -> None:
    from evidence_graph import GraphSchema, validate_generation_database
    from repository_scope import resolve_repository_scope

    path = _verified_v3_database(tmp_path)
    with pytest.raises((sqlite3.IntegrityError, ValueError)):
        _damage_v3(path, damage)
        validate_generation_database(
            path,
            schema=GraphSchema.V3,
            publication_generation_id="verified-v3",
            publication_expected_active="prior",
            repository_scope=resolve_repository_scope(tmp_path / "repository"),
        )


def test_database_construction_cancellation_stops_lazy_records_without_publish(tmp_path):
    import evidence_graph

    records = _records()
    checks = 0

    def cancelled():
        nonlocal checks
        checks += 1
        return checks >= 3

    with pytest.raises(TimeoutError, match="cancel"):
        evidence_graph.create_generation_database(
            tmp_path / "evidence.sqlite3", **records, cancelled=cancelled
        )

    assert not (tmp_path / "evidence.sqlite3").exists()


def test_bounded_node_and_edge_lookup_supports_store_facades(tmp_path):
    graph = _create(tmp_path)

    assert [item["node_id"] for item in graph.find_nodes(kinds=("function",), name="caller")] == [
        "caller"
    ]
    assert [item["assertion_id"] for item in graph.edges(edge_types=("CALLS",))] == ["call"]
    assert graph.find_nodes(kinds=("function",), name="missing") == []
    assert graph.evidence(assertion_id="call")[0]["line_start"] == 2


def test_manifest_schema_is_closed_and_bounded():
    from reliable_memory import validate_schema

    schema = json.loads((SCRIPTS / "schemas/evidence-graph-manifest-v1.json").read_text())
    valid = {
        "generation_id": "gen-1",
        "schema_version": "corpus-generation/v1",
        "collector_version": "collector/v1",
        "extractor_version": "extractor/v1",
        "tokenizer_version": "tokenizer/v1",
        "tokenizer_config_sha256": "0" * 64,
        "embedding_model_id": None,
        "embedding_model_revision": None,
        "vector_dimensions": None,
        "graph_schema_version": "evidence-graph/v2",
        "graph_extractor_version": "graph-extractor/v1",
        "source_manifest_sha256": "1" * 64,
        "artifacts": [
            {"path": "evidence.sqlite3", "size": 4096, "sha256": "2" * 64},
            {"path": "source-manifest.json", "size": 1024, "sha256": "1" * 64},
        ],
        "vector_state": "absent",
    }

    assert schema["$schema"] == "https://json-schema.org/draft/2020-12/schema"
    assert schema["additionalProperties"] is False
    validate_schema(valid, SCRIPTS / "schemas/evidence-graph-manifest-v1.json")
    validate_schema(
        {
            **valid,
            "graph_schema_version": None,
            "graph_extractor_version": None,
        },
        SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
    )
    v2 = {
        **valid,
        "schema_version": "corpus-generation/v2",
        "repository_scope": _valid_repository_scope_schema_value(),
        "artifacts": [
            *valid["artifacts"],
            {"path": "search.sqlite3", "size": 4096, "sha256": "3" * 64},
        ],
    }
    for graph_schema in ("evidence-graph/v2", "evidence-graph/v3"):
        validate_schema(
            {**v2, "graph_schema_version": graph_schema},
            SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
        )
    with pytest.raises(ValueError, match="oneOf|enum|graph_schema_version"):
        validate_schema(
            {
                **v2,
                "graph_schema_version": None,
                "graph_extractor_version": None,
            },
            SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
        )
    with pytest.raises(ValueError, match="oneOf|const|graph_schema_version"):
        validate_schema(
            {**valid, "graph_schema_version": "other/v1"},
            SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
        )
    with pytest.raises(ValueError, match="unknown|additional"):
        validate_schema(
            {**valid, "sources": []}, SCRIPTS / "schemas/evidence-graph-manifest-v1.json"
        )


def _manifest_with_repository_scope(scope: dict[str, object]) -> dict[str, object]:
    return {
        "generation_id": "gen-1",
        "schema_version": "corpus-generation/v1",
        "collector_version": "collector/v1",
        "extractor_version": "extractor/v1",
        "tokenizer_version": "tokenizer/v1",
        "tokenizer_config_sha256": "0" * 64,
        "embedding_model_id": None,
        "embedding_model_revision": None,
        "vector_dimensions": None,
        "graph_schema_version": "evidence-graph/v2",
        "graph_extractor_version": "graph-extractor/v1",
        "source_manifest_sha256": "1" * 64,
        "repository_scope": scope,
        "artifacts": [
            {"path": "evidence.sqlite3", "size": 4096, "sha256": "2" * 64},
        ],
        "vector_state": "absent",
    }


def _valid_repository_scope_schema_value() -> dict[str, object]:
    return {
        "schema_version": "repository-scope/v1",
        "repository_id": "repository:" + "a" * 64,
        "checkout_id": "checkout:" + "b" * 64,
        "checkout_root": "C:/repository",
        "git_common_dir": "C:/repository/.git",
        "git_commit": "c" * 40,
    }


def test_manifest_schema_accepts_closed_repository_scope():
    from reliable_memory import validate_schema

    validate_schema(
        _manifest_with_repository_scope(_valid_repository_scope_schema_value()),
        SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
    )


def test_manifest_schema_accepts_non_git_scope_only_with_null_commit():
    from reliable_memory import validate_schema

    scope = _valid_repository_scope_schema_value()
    scope["git_common_dir"] = None
    scope["git_commit"] = None

    validate_schema(
        _manifest_with_repository_scope(scope),
        SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
    )


@pytest.mark.parametrize(
    "mutate",
    [
        lambda scope: scope.pop("checkout_id"),
        lambda scope: scope.update(unknown=True),
        lambda scope: scope.update(checkout_root="relative/repository"),
        lambda scope: scope.update(git_common_dir="relative/.git"),
        lambda scope: scope.update(repository_id="repository:bad"),
        lambda scope: scope.update(checkout_id="checkout:" + "A" * 64),
        lambda scope: scope.update(git_commit="not-a-commit"),
        lambda scope: scope.update(checkout_root="C:/" + "x" * 4096),
        lambda scope: scope.update(checkout_root="c:/repository"),
        lambda scope: scope.update(checkout_root="C:\\repository"),
        lambda scope: scope.update(checkout_root="C:/repository/../other"),
        lambda scope: scope.update(git_common_dir="/repository//.git"),
        lambda scope: scope.update(checkout_root="C:/repository\x00other"),
        lambda scope: scope.update(git_common_dir="C:/repository/.git\x00other"),
        lambda scope: scope.update(git_common_dir=None),
    ],
)
def test_manifest_schema_rejects_invalid_repository_scope(mutate):
    from reliable_memory import validate_schema

    scope = _valid_repository_scope_schema_value()
    mutate(scope)

    with pytest.raises(ValueError):
        validate_schema(
            _manifest_with_repository_scope(scope),
            SCRIPTS / "schemas/evidence-graph-manifest-v1.json",
        )


def test_graph_source_rows_use_shared_corpus_manifest_contract():
    import corpus_snapshot

    sources = _records()["sources"]
    second = {
        "source_id": "src-doc",
        "relative_path": "knowledge/notes/decision.md",
        "sha256": _sha(b"decision"),
        "size": len(b"decision"),
        "media_type": "text/markdown",
        "language": "markdown",
        "git_oid": "abc123",
    }
    shared = [
        {
            "logical_id": source["source_id"],
            "relative_path": source["relative_path"],
            "sha256": source["sha256"],
        }
        for source in [second, *sources]
    ]
    policy = {"daily_paths": [], "code_roots": [], "include_historical": False, "as_of": None}

    forward = corpus_snapshot.canonical_source_manifest_sha256(shared, policy)
    reverse = corpus_snapshot.canonical_source_manifest_sha256(reversed(shared), policy)

    assert forward == reverse


def test_generation_database_has_canonical_tables_indexes_and_pragmas(tmp_path):
    graph = _create(tmp_path)
    graph.close()

    with closing(sqlite3.connect(tmp_path / "evidence.sqlite3")) as database:
        tables = {
            row[0]
            for row in database.execute(
                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            )
        }
        indexes = {
            row[0]
            for row in database.execute(
                "SELECT name FROM sqlite_master WHERE type='index' ORDER BY name"
            )
        }
        evidence_indexes = {
            row[1]: tuple(
                column[2]
                for column in database.execute(f"PRAGMA index_info({row[1]})")
            )
            for row in database.execute("PRAGMA index_list(evidence)")
            if row[2] == 0
        }
        validation_plan = database.execute(
            """
EXPLAIN QUERY PLAN
SELECT 1 FROM assertion a
WHERE a.resolution='resolved'
  AND NOT EXISTS (SELECT 1 FROM evidence e WHERE e.assertion_id=a.assertion_id)
LIMIT 1
"""
        ).fetchall()
        assert database.execute("PRAGMA journal_mode").fetchone()[0].lower() == "delete"
        assert database.execute("PRAGMA synchronous").fetchone()[0] == 2
        assert database.execute("PRAGMA user_version").fetchone()[0] == 2
        assert database.execute("PRAGMA foreign_key_check").fetchall() == []

    assert tables == {
        "assertion",
        "dependency",
        "evidence",
        "node",
        "observation",
        "occurrence",
        "source",
    }
    assert {
        "assertion_reverse",
        "assertion_resolution",
        "assertion_traversal",
        "dependency_invalidation",
        "dependency_reverse",
        "evidence_assertion",
        "evidence_source_span",
        "node_kind",
        "observation_resolution",
        "occurrence_source_span",
    } <= indexes
    assert evidence_indexes["evidence_assertion"][:1] == ("assertion_id",)
    evidence_steps = [detail for *_prefix, detail in validation_plan if " e" in detail]
    assert any("SEARCH e" in detail for detail in evidence_steps), evidence_steps
    assert all("SCAN e" not in detail for detail in evidence_steps), evidence_steps
    assert not (tmp_path / "evidence.sqlite3-wal").exists()


def test_logical_nodes_are_separate_from_occurrences_and_metadata_is_canonical(tmp_path):
    graph = _create(tmp_path)

    node = graph.node("caller")
    occurrences = graph.occurrences("caller")

    assert node["identity_key"] == "src.app:caller"
    assert node["metadata"] == {"name": "caller"}
    assert occurrences == [
        {
            "occurrence_id": "occ-caller",
            "node_id": "caller",
            "source_id": "src-code",
            "relative_path": "src/app.py",
            "source_sha256": _records()["sources"][0]["sha256"],
            "role": "definition",
            "byte_start": 0,
            "byte_end": 13,
            "line_start": 1,
            "line_end": 1,
        }
    ]
    graph.close()


@pytest.mark.parametrize("damage", ["source_hash", "range", "span_hash", "unknown_field"])
def test_create_fails_closed_on_invalid_sources_evidence_and_records(tmp_path, damage):
    import evidence_graph

    records = _records()
    if damage == "source_hash":
        records["sources"][0]["sha256"] = "0" * 64
    elif damage == "range":
        records["evidence"][0]["byte_end"] = len(records["source_bytes"]["src-code"]) + 1
    elif damage == "span_hash":
        records["evidence"][0]["span_sha256"] = "0" * 64
    else:
        records["nodes"][0]["unknown"] = True

    with pytest.raises((TypeError, ValueError), match="hash|range|unknown|closed"):
        evidence_graph.create_generation_database(tmp_path / "evidence.sqlite3", **records)
    assert not (tmp_path / "evidence.sqlite3").exists()


def test_every_resolved_assertion_requires_nonempty_half_open_evidence(tmp_path):
    import evidence_graph

    records = _records()
    records["evidence"] = records["evidence"][:1]
    with pytest.raises(ValueError, match="resolved assertion.*evidence"):
        evidence_graph.create_generation_database(tmp_path / "missing.sqlite3", **records)

    graph = _create(tmp_path)
    graph.close()
    with closing(sqlite3.connect(tmp_path / "evidence.sqlite3")) as database:
        database.row_factory = sqlite3.Row
        database.execute("DELETE FROM evidence WHERE assertion_id='documents'")
        with pytest.raises(ValueError, match="integrity"):
            evidence_graph._validate_connection(database)

    records = _records()
    records["evidence"][0].update(
        byte_end=records["evidence"][0]["byte_start"], span_sha256=_sha(b"")
    )
    with pytest.raises(ValueError, match="non-empty|range"):
        evidence_graph.create_generation_database(tmp_path / "empty.sqlite3", **records)

    records = _records()
    records["evidence"][0]["byte_end"] += 1
    records["evidence"][0]["span_sha256"] = _sha(b"callee()\n")
    evidence_graph.create_generation_database(tmp_path / "half-open.sqlite3", **records)


def test_resolved_literal_assertion_is_valid(tmp_path):
    import evidence_graph

    records = _records()
    records["assertions"].append(
        {
            "assertion_id": "literal",
            "source_node_id": "caller",
            "edge_type": "RETURNS",
            "target_node_id": None,
            "literal": {"type": "str"},
            "confidence": "high",
            "authority": "ai-derived",
            "resolution": "resolved",
            "extractor": "python/v1",
        }
    )
    records["evidence"].append(
        {
            "evidence_id": "ev-literal",
            "assertion_id": "literal",
            "observation_id": None,
            "source_id": "src-code",
            "byte_start": 0,
            "byte_end": 3,
            "span_sha256": _sha(records["source_bytes"]["src-code"][:3]),
        }
    )

    evidence_graph.create_generation_database(tmp_path / "literal.sqlite3", **records)
    graph = evidence_graph.EvidenceGraph(tmp_path / "literal.sqlite3", state_root=tmp_path)
    graph.close()


def test_resolved_node_assertion_requires_a_concrete_target(tmp_path):
    import evidence_graph

    records = _records()
    records["assertions"][0].update(target_node_id=None, literal=None)

    with pytest.raises(ValueError, match="target node or literal"):
        evidence_graph.create_generation_database(tmp_path / "missing-target.sqlite3", **records)


def test_unresolved_or_ambiguous_assertions_use_observations_instead(tmp_path):
    import evidence_graph

    for resolution in ("unresolved", "ambiguous"):
        records = _records()
        records["assertions"][0].update(
            resolution=resolution,
            target_node_id=None,
            literal=None,
        )
        with pytest.raises(ValueError, match="observation|resolved"):
            evidence_graph.create_generation_database(tmp_path / f"{resolution}.sqlite3", **records)


@pytest.mark.parametrize(
    "occurrence",
    [
        {"byte_start": 0, "byte_end": 13, "line_start": 2, "line_end": 2},
        {"byte_start": 0, "byte_end": 13, "line_start": 1, "line_end": 2},
        {"byte_start": 0, "byte_end": 14, "line_start": 1, "line_end": 1},
        {"byte_start": 4, "byte_end": 4, "line_start": 1, "line_end": 1},
    ],
)
def test_occurrence_lines_must_exactly_match_nonempty_source_byte_span(tmp_path, occurrence):
    import evidence_graph

    records = _records()
    records["occurrences"][0].update(occurrence)

    with pytest.raises(ValueError, match="occurrence.*line|non-empty|range"):
        evidence_graph.create_generation_database(tmp_path / "bad-lines.sqlite3", **records)


def test_multiline_occurrence_uses_exact_half_open_byte_line_mapping(tmp_path):
    import evidence_graph

    records = _records()
    records["occurrences"][0].update(
        byte_start=4,
        byte_end=20,
        line_start=1,
        line_end=2,
    )

    evidence_graph.create_generation_database(tmp_path / "multiline.sqlite3", **records)

    records = _records()
    records["occurrences"][0].update(byte_end=14, line_start=1, line_end=2)
    evidence_graph.create_generation_database(tmp_path / "newline-end.sqlite3", **records)


def test_evidence_source_binding_is_verified(tmp_path):
    import evidence_graph

    records = _records()
    other = b"XXXXXXXX"
    records["sources"].append(
        {
            "source_id": "other",
            "relative_path": "other.py",
            "sha256": _sha(other),
            "size": len(other),
            "media_type": "text/x-python",
            "language": "python",
            "git_oid": None,
        }
    )
    records["source_bytes"]["other"] = other
    records["evidence"][0].update(source_id="other", byte_start=0, byte_end=len(other))
    with pytest.raises(ValueError, match="source|hash"):
        evidence_graph.create_generation_database(tmp_path / "wrong-source.sqlite3", **records)


def test_generation_publication_is_exclusive_under_concurrent_destination_race(tmp_path):
    import evidence_graph

    destination = tmp_path / "evidence.sqlite3"

    def publish():
        try:
            evidence_graph.create_generation_database(destination, **_records())
        except FileExistsError:
            return "lost"
        return "won"

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = sorted(pool.map(lambda _value: publish(), range(2)))

    assert results == ["lost", "won"]
    graph = evidence_graph.EvidenceGraph(destination, state_root=tmp_path)
    assert graph.node("caller")["identity_key"] == "src.app:caller"
    graph.close()


def test_unresolved_observations_use_controlled_reasons_without_fake_nodes(tmp_path):
    graph = _create(tmp_path)

    assert graph.unresolved() == [
        {
            "observation_id": "dynamic-call",
            "source_node_id": "caller",
            "edge_type": "CALLS",
            "target_text": "runtime_target",
            "reason": "dynamic_dispatch",
            "extractor": "python/v1",
        }
    ]
    graph.close()

    records = _records()
    records["observations"][0]["reason"] = "made_up"
    with pytest.raises(ValueError, match="reason"):
        import evidence_graph

        evidence_graph.create_generation_database(tmp_path / "bad.sqlite3", **records)


def test_bounded_queries_cover_both_directions_paths_dependencies_and_evidence(tmp_path):
    graph = _create(tmp_path)

    assert [row["node_id"] for row in graph.neighbors("caller", direction="out")] == ["callee"]
    assert [row["node_id"] for row in graph.neighbors("callee", direction="in")] == ["caller"]
    assert graph.path("caller", "callee")[0]["assertion_ids"] == ["call"]
    assert [row["node_id"] for row in graph.callers("callee")] == ["caller"]
    assert [row["node_id"] for row in graph.callees("caller")] == ["callee"]
    assert [row["node_id"] for row in graph.dependencies("decision")] == ["caller"]
    assert [row["node_id"] for row in graph.code_to_doc("caller")] == ["decision"]
    assert [row["node_id"] for row in graph.doc_to_code("decision")] == ["caller"]
    assert graph.evidence(assertion_id="call")[0]["span_sha256"] == _sha(b"callee()")
    graph.close()


def test_evidence_spans_return_hash_bindings_without_selecting_source_content(
    tmp_path, monkeypatch
):
    graph = _create(tmp_path)
    statements = []
    execute = graph._execute

    def tracked(sql, *args, **kwargs):
        statements.append(sql)
        return execute(sql, *args, **kwargs)

    monkeypatch.setattr(graph, "_execute", tracked)

    spans = graph.evidence_spans(assertion_id="call")

    assert spans == [
        {
            "evidence_id": "ev-call",
            "assertion_id": "call",
            "observation_id": None,
            "source_id": "src-code",
            "byte_start": 18,
            "byte_end": 26,
            "span_sha256": _sha(b"callee()"),
            "relative_path": "src/app.py",
            "source_sha256": _records()["sources"][0]["sha256"],
        }
    ]
    assert statements
    assert all("content" not in statement.casefold() for statement in statements)
    graph.close()


def test_dependencies_walk_transitively_with_a_depth_bound(tmp_path):
    records = _records()
    records["nodes"].append(
        {
            "node_id": "package",
            "kind": "package",
            "identity_scheme": "package/v1",
            "identity_key": "runtime",
            "metadata": {},
        }
    )
    records["dependencies"].append(
        {
            "dependency_id": "dep-runtime",
            "dependent_node_id": "caller",
            "dependency_node_id": "package",
            "kind": "imports",
            "source_id": "src-code",
        }
    )
    graph = _create(tmp_path, **records)

    assert [row["node_id"] for row in graph.dependencies("decision", max_depth=1)] == ["caller"]
    assert [row["node_id"] for row in graph.dependencies("decision", max_depth=2)] == [
        "caller",
        "package",
    ]
    graph.close()


def test_query_limits_depth_rows_deadlines_and_closed_enums(tmp_path):
    graph = _create(tmp_path)

    with pytest.raises(ValueError, match="direction"):
        graph.neighbors("caller", direction="sideways")
    with pytest.raises(ValueError, match="max_depth"):
        graph.path("caller", "callee", max_depth=0)
    with pytest.raises(ValueError, match="max_rows"):
        graph.unresolved(max_rows=0)
    with pytest.raises(TimeoutError, match="deadline"):
        graph.neighbors("caller", deadline=time.monotonic() - 1)
    with pytest.raises(ValueError, match="edge_types"):
        graph.neighbors("caller", edge_types=[f"EDGE_{number}" for number in range(65)])
    graph.close()


def test_recursive_traversal_is_deterministic_delimiter_safe_and_work_bounded(tmp_path):
    import evidence_graph

    records = _records()
    records["nodes"].extend(
        {
            "node_id": identifier,
            "kind": "function",
            "identity_scheme": "test/v1",
            "identity_key": identifier,
            "metadata": {},
        }
        for identifier in ("branch-a", "branch-b", "branch-c")
    )
    for number, (source, target) in enumerate(
        (
            ("callee", "branch-c"),
            ("caller", "branch-b"),
            ("caller", "branch-a"),
            ("branch-a", "caller"),
        )
    ):
        assertion_id = f"edge-{number}"
        records["assertions"].append(
            {
                "assertion_id": assertion_id,
                "source_node_id": source,
                "edge_type": "CALLS",
                "target_node_id": target,
                "literal": None,
                "confidence": "high",
                "authority": "ai-derived",
                "resolution": "resolved",
                "extractor": "test/v1",
            }
        )
        records["evidence"].append(
            {
                "evidence_id": f"evidence-{number}",
                "assertion_id": assertion_id,
                "observation_id": None,
                "source_id": "src-code",
                "byte_start": 18,
                "byte_end": 26,
                "span_sha256": _sha(b"callee()"),
            }
        )
    graph = _create(tmp_path, **records)

    first = graph.neighbors("caller", max_depth=3, max_rows=10, max_work=20)
    second = graph.neighbors("caller", max_depth=3, max_rows=10, max_work=20)
    assert first == second
    assert [row["node_id"] for row in first] == ["branch-a", "branch-b", "callee", "branch-c"]
    with pytest.raises(ValueError, match="work"):
        graph.neighbors("caller", max_depth=3, max_rows=10, max_work=2)
    graph.close()

    bad = _records()
    bad["nodes"][0]["node_id"] = "caller,alias"
    with pytest.raises(ValueError, match="node_id"):
        evidence_graph.create_generation_database(tmp_path / "unsafe-id.sqlite3", **bad)


def test_database_is_opened_query_only_and_path_must_remain_in_state_root(tmp_path):
    graph = _create(tmp_path)
    assert graph._database.execute("PRAGMA query_only").fetchone()[0] == 1
    with pytest.raises(sqlite3.OperationalError):
        graph._database.execute("DELETE FROM node")
    graph.close()

    import evidence_graph

    with pytest.raises((PermissionError, ValueError)):
        evidence_graph.EvidenceGraph(tmp_path / "evidence.sqlite3", state_root=tmp_path / "other")
