core/tests/test_contemplation_quality_lane.py

"""Contract tests for the ``contemplation_quality`` eval lane.

W-025 / ADR-0159.  The lane scores the structured output from
``core demo learning-arc --json`` along nine deterministic, non-mutating
quality gates without widening the trust surface.

These tests pin:

  - Case-set integrity (single invocation case, required schema).
  - Lane discovery via the generic eval framework (no CLI wiring needed).
  - ``evaluate_report`` purity over arbitrary dictionaries (well-formed,
    malformed, empty, wrong types).
  - ``run_lane`` input-shape enforcement (single case, source enum).
  - Read-only invariant: lane execution must not produce filesystem writes
    under teaching/, packs/, or engine_state/ during scoring.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from evals.framework import (
    discover_lanes,
    get_lane,
    load_cases,
    load_lane_runner,
)
from evals.contemplation_quality.runner import (
    ContemplationQualityReport,
    LaneReport,
    QualityMetric,
    evaluate_report,
    run_lane,
)


LANE_NAME = "contemplation_quality"
_EVAL_ROOT = Path(__file__).resolve().parent.parent / "evals" / LANE_NAME
_PUBLIC_CASES = _EVAL_ROOT / "public" / "v1" / "cases.jsonl"
_DEV_CASES = _EVAL_ROOT / "dev" / "cases.jsonl"
_CONTRACT = _EVAL_ROOT / "contract.md"


_REQUIRED_METRIC_NAMES: frozenset[str] = frozenset({
    "scene_contract",
    "deterministic_replay_integrity",
    "typed_contemplation_provenance",
    "engine_authored_specificity",
    "grounding_transition",
    "downstream_gain_observed",
    "active_corpus_boundary",
    "pending_not_auto_accepted",
    "stable_proposal_identity_present",
})


# ---------------------------------------------------------------------------
# Case-set integrity
# ---------------------------------------------------------------------------


class TestCaseSetIntegrity:
    def test_public_cases_file_exists(self) -> None:
        assert _PUBLIC_CASES.exists()

    def test_dev_cases_file_exists(self) -> None:
        assert _DEV_CASES.exists()

    def test_contract_file_exists(self) -> None:
        assert _CONTRACT.exists()

    def test_public_case_count_is_one(self) -> None:
        cases = load_cases(_PUBLIC_CASES)
        assert len(cases) == 1

    def test_dev_case_count_is_one(self) -> None:
        cases = load_cases(_DEV_CASES)
        assert len(cases) == 1

    def test_case_required_fields(self) -> None:
        for path in (_PUBLIC_CASES, _DEV_CASES):
            for case in load_cases(path):
                assert "case_id" in case and isinstance(case["case_id"], str)
                assert "source" in case and isinstance(case["source"], str)

    def test_case_source_is_supported_enum(self) -> None:
        for path in (_PUBLIC_CASES, _DEV_CASES):
            for case in load_cases(path):
                assert case["source"] == "learning_arc_demo"


# ---------------------------------------------------------------------------
# Lane discovery via the generic framework
# ---------------------------------------------------------------------------


class TestLaneDiscovery:
    def test_lane_is_discoverable(self) -> None:
        names = {lane.name for lane in discover_lanes()}
        assert LANE_NAME in names

    def test_lane_has_v1_version(self) -> None:
        lane = get_lane(LANE_NAME)
        assert "v1" in lane.versions

    def test_lane_runner_exposes_run_lane(self) -> None:
        lane = get_lane(LANE_NAME)
        runner = load_lane_runner(lane)
        assert hasattr(runner, "run_lane")
        assert hasattr(runner, "evaluate_report")


# ---------------------------------------------------------------------------
# evaluate_report — pure function over a dict
# ---------------------------------------------------------------------------


def _passing_report() -> dict:
    """Synthesize a minimal learning-arc report that satisfies every gate."""
    return {
        "engine_connective": "grounds",
        "engine_object": "truth",
        "learning_arc_closed": True,
        "active_corpus_byte_identical": True,
        "before": {"surface": "I don't know."},
        "after": {"surface": "light grounds truth"},
        "scenes": [
            {
                "scene": "S1_cold_session",
                "detail": {"grounding_source": "none"},
            },
            {
                "scene": "S2_checkpoint_enrichment",
                "detail": {
                    "engine_chain_found": True,
                    "engine_chain": {"connective": "grounds", "object": "truth"},
                },
            },
            {
                "scene": "S3_engine_authored_proposal",
                "detail": {
                    "source_kind": "contemplation",
                    "proposal_id": "proposal-abc123",
                    "state": "pending",
                    "replay_evidence": {
                        "replay_equivalent": True,
                        "regressed_metrics": [],
                    },
                    "proposed_chain": {
                        "connective": "grounds",
                        "object": "truth",
                    },
                },
            },
            {
                "scene": "S4_operator_ratifies",
                "detail": {"active_corpus_byte_identical": True},
            },
            {
                "scene": "S5_grounded_session",
                "detail": {"grounding_source": "teaching"},
            },
        ],
    }


class TestEvaluateReportShape:
    def test_returns_contemplation_quality_report(self) -> None:
        report = evaluate_report(_passing_report())
        assert isinstance(report, ContemplationQualityReport)

    def test_lane_label_is_canonical(self) -> None:
        report = evaluate_report(_passing_report())
        assert report.lane == "contemplation-quality"

    def test_source_label_is_canonical(self) -> None:
        report = evaluate_report(_passing_report())
        assert report.source == "core demo learning-arc --json"

    def test_source_digest_is_sha256_hex(self) -> None:
        report = evaluate_report(_passing_report())
        assert isinstance(report.source_digest, str)
        assert len(report.source_digest) == 64
        int(report.source_digest, 16)  # raises ValueError if non-hex

    def test_all_nine_metrics_present(self) -> None:
        report = evaluate_report(_passing_report())
        names = {m.name for m in report.metrics}
        assert names == _REQUIRED_METRIC_NAMES

    def test_metrics_are_quality_metric_instances(self) -> None:
        report = evaluate_report(_passing_report())
        for metric in report.metrics:
            assert isinstance(metric, QualityMetric)


class TestEvaluateReportDeterminism:
    def test_same_input_yields_same_digest(self) -> None:
        a = evaluate_report(_passing_report())
        b = evaluate_report(_passing_report())
        assert a.source_digest == b.source_digest

    def test_well_formed_report_passes(self) -> None:
        report = evaluate_report(_passing_report())
        assert report.passed is True

    def test_serializable_as_dict(self) -> None:
        report = evaluate_report(_passing_report()).as_dict()
        # Must be JSON-serializable without raising.  Tuples in ``expected``
        # values become lists after a JSON round-trip, which is fine for
        # downstream consumers — the contract here is only serializability.
        encoded = json.dumps(report)
        decoded = json.loads(encoded)
        assert decoded["lane"] == report["lane"]
        assert decoded["source_digest"] == report["source_digest"]
        assert decoded["passed"] is report["passed"]
        assert decoded["score"] == report["score"]
        assert len(decoded["metrics"]) == len(report["metrics"])


class TestEvaluateReportBoundaryViolations:
    """Each gate should fail when its specific invariant is broken."""

    def _mutate_scene(
        self,
        report: dict,
        scene_name: str,
        **detail_overrides,
    ) -> dict:
        for scene in report["scenes"]:
            if scene["scene"] == scene_name:
                scene["detail"] = {**scene["detail"], **detail_overrides}
        return report

    def _failed_metric(
        self,
        report: ContemplationQualityReport,
        name: str,
    ) -> QualityMetric:
        for metric in report.metrics:
            if metric.name == name:
                return metric
        raise AssertionError(f"metric {name!r} not in report")

    def test_scene_contract_fails_on_missing_scene(self) -> None:
        report = _passing_report()
        report["scenes"] = report["scenes"][:-1]
        scored = evaluate_report(report)
        assert self._failed_metric(scored, "scene_contract").passed is False

    def test_replay_integrity_fails_when_not_equivalent(self) -> None:
        report = self._mutate_scene(
            _passing_report(),
            "S3_engine_authored_proposal",
            replay_evidence={
                "replay_equivalent": False,
                "regressed_metrics": ["surface_diff"],
            },
        )
        scored = evaluate_report(report)
        assert (
            self._failed_metric(scored, "deterministic_replay_integrity").passed
            is False
        )

    def test_pending_gate_fails_on_auto_acceptance(self) -> None:
        report = self._mutate_scene(
            _passing_report(),
            "S3_engine_authored_proposal",
            state="accepted",
        )
        scored = evaluate_report(report)
        assert (
            self._failed_metric(scored, "pending_not_auto_accepted").passed
            is False
        )

    def test_active_corpus_boundary_fails_on_byte_drift(self) -> None:
        report = _passing_report()
        report["active_corpus_byte_identical"] = False
        scored = evaluate_report(report)
        assert (
            self._failed_metric(scored, "active_corpus_boundary").passed
            is False
        )

    def test_provenance_gate_fails_without_contemplation_kind(self) -> None:
        report = self._mutate_scene(
            _passing_report(),
            "S3_engine_authored_proposal",
            source_kind="seeded",
        )
        scored = evaluate_report(report)
        assert (
            self._failed_metric(scored, "typed_contemplation_provenance").passed
            is False
        )


class TestEvaluateReportMalformedInput:
    """The pure-function entry point must reject or absorb malformed shapes."""

    def test_non_dict_input_raises_type_error(self) -> None:
        with pytest.raises(TypeError):
            evaluate_report([])  # type: ignore[arg-type]

    def test_none_input_raises_type_error(self) -> None:
        with pytest.raises(TypeError):
            evaluate_report(None)  # type: ignore[arg-type]

    def test_empty_report_produces_all_failing_metrics(self) -> None:
        scored = evaluate_report({})
        assert scored.passed is False
        # All nine metrics still emitted — none are silently skipped.
        assert {m.name for m in scored.metrics} == _REQUIRED_METRIC_NAMES

    def test_malformed_scenes_field_does_not_crash(self) -> None:
        scored = evaluate_report({"scenes": "not-a-list"})
        assert scored.passed is False

    def test_malformed_scene_detail_does_not_crash(self) -> None:
        scored = evaluate_report(
            {"scenes": [{"scene": "S1_cold_session", "detail": "wrong-type"}]}
        )
        assert scored.passed is False


# ---------------------------------------------------------------------------
# run_lane — invocation-contract enforcement
# ---------------------------------------------------------------------------


class TestRunLaneInputContract:
    def test_empty_case_list_rejected(self) -> None:
        with pytest.raises(ValueError, match="exactly one"):
            run_lane([])

    def test_multiple_cases_rejected(self) -> None:
        with pytest.raises(ValueError, match="exactly one"):
            run_lane([
                {"case_id": "a", "source": "learning_arc_demo"},
                {"case_id": "b", "source": "learning_arc_demo"},
            ])

    def test_non_list_input_rejected(self) -> None:
        with pytest.raises(ValueError):
            run_lane("not-a-list")  # type: ignore[arg-type]

    def test_non_dict_case_rejected(self) -> None:
        with pytest.raises(TypeError):
            run_lane(["not-a-dict"])  # type: ignore[list-item]

    def test_unsupported_source_rejected(self) -> None:
        with pytest.raises(ValueError, match="unsupported"):
            run_lane([{"case_id": "x", "source": "external_dataset_v2"}])


# ---------------------------------------------------------------------------
# Read-only invariant — execution must not write outside tempdirs
# ---------------------------------------------------------------------------


class TestReadOnlyInvariant:
    """ADR-0159 read-only invariants.

    The lane must never mutate the active teaching corpus or any pack data
    file.  These are the trust boundaries the proposal/teaching path
    protects: corpus mutation requires ``accept_proposal``, pack mutation
    requires a reviewed pack-mutation ADR path, and neither is exercised
    by the eval lane.

    Note on ``engine_state/``: the lane's downstream demo (run_demo) runs
    a replay-equivalence gate that spawns the cognition lane, whose
    per-case ``ChatRuntime`` instances checkpoint to ``engine_state/`` via
    the runtime path already governed by ADR-0146/0150.  That checkpoint
    surface is a transient runtime artifact, not a teaching/pack write,
    so it is explicitly out of scope for this invariant.
    """

    def _snapshot(self, root: Path) -> dict[str, bytes]:
        snap: dict[str, bytes] = {}
        if not root.exists():
            return snap
        for path in sorted(root.rglob("*")):
            if not path.is_file():
                continue
            rel = path.relative_to(root)
            # Skip Python bytecode caches and the package's own
            # ``__init__.py`` — they are not corpus/pack content.
            if "__pycache__" in rel.parts or rel.suffix in {".pyc", ".pyo"}:
                continue
            snap[str(rel)] = path.read_bytes()
        return snap

    def test_run_lane_does_not_mutate_teaching_or_packs(self) -> None:
        repo_root = Path(__file__).resolve().parent.parent
        guarded = {
            "teaching/corpora": repo_root / "teaching" / "corpora",
            "packs": repo_root / "packs",
            "language_packs/data": repo_root / "language_packs" / "data",
        }
        before = {k: self._snapshot(v) for k, v in guarded.items()}

        result = run_lane(load_cases(_PUBLIC_CASES))

        after = {k: self._snapshot(v) for k, v in guarded.items()}
        for key in guarded:
            assert before[key] == after[key], (
                f"lane execution mutated {key} — trust boundary violated"
            )
        assert isinstance(result, LaneReport)
        assert result.metrics["all_passed"] is True