core/evals/demo_composition/runner.py

"""Runner for evals/demo_composition/ (ADR-0098).

Exercises shipped :class:`DemoCommand` adapters and asserts ADR-0098's
two structural invariants:

- ``demo_json_byte_equality``: two runs of the same adapter produce
  byte-identical JSON.
- ``demo_composition_no_side_effects``: a single adapter run does not
  mutate load-bearing global state.

The slowest adapter (anchor-lens-tour) is excluded from this lane to
keep wall time bounded; it is exercised by
``tests/test_demo_composition.py`` instead.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from core.demos.audit_tour_adapter import AuditTourDemo
from core.demos.contract import (
    DemoResult,
    capture_state,
    verify_no_global_state_mutation,
)
from core.demos.tour_adapters import (
    OrthogonalityTourDemo,
    RegisterTourDemo,
)


# Skip anchor-lens-tour in the lane because of its high wall time.
_ADAPTERS: tuple[tuple[str, Any], ...] = (
    ("audit-tour", AuditTourDemo()),
    ("register-tour", RegisterTourDemo()),
    ("orthogonality-tour", OrthogonalityTourDemo()),
)


def _run_byte_equality(adapter: Any, *, tmp_root: Path) -> dict[str, Any]:
    case_dir = tmp_root / f"{adapter.demo_id}_byte"
    case_dir.mkdir(parents=True, exist_ok=True)
    out_a = case_dir / "a"
    out_b = case_dir / "b"
    r_a = adapter.run(output_dir=out_a)
    bytes_a = r_a.json_path.read_bytes()
    r_b = adapter.run(output_dir=out_b)
    bytes_b = r_b.json_path.read_bytes()
    return {
        "case_id": f"{adapter.demo_id}_byte_equality",
        "passed": bytes_a == bytes_b,
        "details": {
            "sha256": hashlib.sha256(bytes_a).hexdigest(),
            "all_claims_supported_a": r_a.all_claims_supported,
            "all_claims_supported_b": r_b.all_claims_supported,
        },
        "divergence": None if bytes_a == bytes_b else "JSON bytes differ",
    }


def _run_state_mutation_check(adapter: Any, *, tmp_root: Path) -> dict[str, Any]:
    case_dir = tmp_root / f"{adapter.demo_id}_state"
    case_dir.mkdir(parents=True, exist_ok=True)
    before = capture_state()
    adapter.run(output_dir=case_dir)
    after = capture_state()
    passed, divergences = verify_no_global_state_mutation(before=before, after=after)
    return {
        "case_id": f"{adapter.demo_id}_no_state_mutation",
        "passed": passed,
        "details": {"divergences": list(divergences)},
        "divergence": None if passed else f"state divergences: {divergences}",
    }


def _run_composition_read_only(tmp_root: Path) -> dict[str, Any]:
    """Compose two adapter results into a composite claim set.

    Verifies the showcase pattern (ADR-0099 preview): reading two
    DemoResult objects and producing a composite without mutating
    either. Each adapter is invoked exactly once.
    """
    out_dir = tmp_root / "composition"
    out_dir.mkdir(parents=True, exist_ok=True)
    audit = AuditTourDemo().run(output_dir=out_dir)
    register = RegisterTourDemo().run(output_dir=out_dir)
    # Composition is a tuple of evidence locators; nothing is mutated.
    composite_claims = tuple(audit.claims) + tuple(register.claims)
    composite_evidence = {**audit.evidence, **register.evidence}
    composite_supported = (
        audit.all_claims_supported and register.all_claims_supported
    )
    return {
        "case_id": "composition_read_only",
        "passed": True,
        "details": {
            "audit_claims": len(audit.claims),
            "register_claims": len(register.claims),
            "composite_claims": len(composite_claims),
            "composite_supported": composite_supported,
            "evidence_size": len(composite_evidence),
        },
        "divergence": None,
    }


@dataclass(frozen=True, slots=True)
class _StatefulFixtureAdapter:
    """Negative control: deliberately mutates an env var during run."""

    demo_id: str = "stateful-fixture"
    claim_contract_version: int = 1

    def run(self, *, output_dir: Path, seed: int | None = None) -> DemoResult:
        import os

        from core.demos.contract import CLAIM_CONTRACT_VERSION, Claim, canonical_json

        os.environ["CORE_STATEFUL_FIXTURE_FLAG"] = "1"
        output_dir.mkdir(parents=True, exist_ok=True)
        json_path = output_dir / f"{self.demo_id}.json"
        claim = Claim(
            claim_id="x",
            statement="negative control",
            supported=True,
            evidence_locator="self",
        )
        json_path.write_bytes(canonical_json({"demo_id": self.demo_id}))
        return DemoResult(
            demo_id=self.demo_id,
            claim_contract_version=CLAIM_CONTRACT_VERSION,
            claims=(claim,),
            evidence={claim.claim_id: "self"},
            all_claims_supported=True,
            json_path=json_path,
            trace_features={},
        )


def _run_stateful_fixture_rejected(tmp_root: Path) -> dict[str, Any]:
    import os

    case_dir = tmp_root / "stateful"
    case_dir.mkdir(parents=True, exist_ok=True)
    # Make sure the env var is not set before; clean up on the way out.
    os.environ.pop("CORE_STATEFUL_FIXTURE_FLAG", None)
    before = capture_state()
    _StatefulFixtureAdapter().run(output_dir=case_dir)
    after = capture_state()
    os.environ.pop("CORE_STATEFUL_FIXTURE_FLAG", None)
    passed, divergences = verify_no_global_state_mutation(before=before, after=after)
    # NEGATIVE control: the detector MUST flag this fixture.
    expected_to_fail = not passed
    return {
        "case_id": "stateful_fixture_rejected",
        "passed": expected_to_fail,
        "details": {"divergences": list(divergences)},
        "divergence": (
            None if expected_to_fail else "negative control did not produce divergences"
        ),
    }


def run() -> dict[str, Any]:
    import os

    # Hermetic engine state (same pattern as public_demo): lived engine_state/
    # on the developer's machine or a dirty CI workspace must not leak into
    # adapter JSON hashes and break the lane pin.
    os.environ.setdefault(
        "CORE_ENGINE_STATE_DIR",
        tempfile.mkdtemp(prefix="demo_composition_engine_state_"),
    )

    tmp_root = Path(tempfile.mkdtemp(prefix="demo_composition_lane_"))
    try:
        cases: list[dict[str, Any]] = []
        for _, adapter in _ADAPTERS:
            cases.append(_run_byte_equality(adapter, tmp_root=tmp_root))
            cases.append(_run_state_mutation_check(adapter, tmp_root=tmp_root))
        cases.append(_run_composition_read_only(tmp_root))
        cases.append(_run_stateful_fixture_rejected(tmp_root))
    finally:
        shutil.rmtree(tmp_root, ignore_errors=True)
    return {
        "lane": "demo_composition",
        "lane_version": "v1",
        "split": "dev",
        "adr": "ADR-0098",
        "invariants": [
            "demo_json_byte_equality",
            "demo_composition_no_side_effects",
        ],
        "adapters_exercised": [a.demo_id for _, a in _ADAPTERS],
        "total_cases": len(cases),
        "passed_cases": sum(1 for c in cases if c["passed"]),
        "failed_cases": sum(1 for c in cases if not c["passed"]),
        "all_passed": all(c["passed"] for c in cases),
        "cases": cases,
    }


def _canonical_json(payload: dict[str, Any]) -> bytes:
    return json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n"


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="demo_composition lane runner")
    parser.add_argument("--report", type=Path, default=None)
    args = parser.parse_args(argv)

    summary = run()
    lane_dir = Path(__file__).resolve().parent
    report_path = args.report or (lane_dir / "results" / "v1_dev.json")
    report_path.parent.mkdir(parents=True, exist_ok=True)
    payload_bytes = _canonical_json(summary)
    report_path.write_bytes(payload_bytes)

    sha = hashlib.sha256(payload_bytes).hexdigest()
    print(f"report: {report_path}")
    print(f"sha256: {sha}")
    print(f"passed: {summary['passed_cases']}/{summary['total_cases']}")
    return 0 if summary["all_passed"] else 1


if __name__ == "__main__":
    sys.exit(main())