acdl/core/regression_verify.py

"""Regression-class VERIFY (D-091).

The standard VERIFY stage is diff-scoped: it checks the phase diff only
and never re-runs underlying platform capability. That structural defect
(let 8 NFR-patch phases pass while the platform decayed) is recorded as
D-091. This module provides the regression-class VERIFY that re-runs
capability checks against the current codebase and tags each capability
Verified / Decayed / Broken.

A capability check is a function that takes no args and returns
(status, detail) where status is one of:
  - "Verified"  : the capability runs as advertised
  - "Decayed"   : the capability runs partially / with errors but the
                  core path is intact (e.g. needs revival work)
  - "Broken"    : the capability does not run at all

The regression run fails closed: any non-Verified capability blocks
milestone completion. The result is written to
`.ciagent/REGRESSION_REPORT.md` and a machine-readable JSON file.
"""

from __future__ import annotations

import importlib
import json
import os
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Callable, Dict, List, Optional, Tuple

ROOT = Path(__file__).resolve().parent.parent
CIAgent = ROOT / ".ciagent"

Status = str  # "Verified" | "Decayed" | "Broken"


@dataclass
class CapabilityResult:
    capability_id: str
    name: str
    status: Status
    detail: str
    tier: str  # "local" | "live-aws"
    duration_ms: int


@dataclass
class RegressionReport:
    run_id: str
    run_at_utc: str
    milestone: str
    phase: int
    results: List[CapabilityResult] = field(default_factory=list)

    @property
    def summary(self) -> Dict[str, int]:
        counts = {"Verified": 0, "Decayed": 0, "Broken": 0}
        for r in self.results:
            counts[r.status] = counts.get(r.status, 0) + 1
        return counts

    @property
    def passed(self) -> bool:
        return all(r.status == "Verified" for r in self.results)

    def to_dict(self) -> dict:
        return {
            "run_id": self.run_id,
            "run_at_utc": self.run_at_utc,
            "milestone": self.milestone,
            "phase": self.phase,
            "summary": self.summary,
            "passed": self.passed,
            "results": [asdict(r) for r in self.results],
        }


def _run_subprocess(cmd: List[str], cwd: Optional[str] = None,
                    timeout: int = 120) -> Tuple[int, str, str]:
    """Run a subprocess, return (returncode, stdout, stderr)."""
    try:
        p = subprocess.run(
            cmd, cwd=cwd or str(ROOT), capture_output=True,
            text=True, timeout=timeout,
        )
        return p.returncode, p.stdout, p.stderr
    except subprocess.TimeoutExpired as e:
        return 124, e.stdout or "", e.stderr or ""
    except FileNotFoundError as e:
        return 127, "", str(e)


def _check_subprocess(cmd: List[str], cwd: Optional[str] = None,
                      timeout: int = 120) -> Tuple[Status, str]:
    """Run a subprocess; map returncode to a status."""
    rc, out, err = _run_subprocess(cmd, cwd=cwd, timeout=timeout)
    if rc == 0:
        return "Verified", f"exit 0; {out.strip()[-200:]}"
    if rc == 124:
        return "Decayed", f"timeout after {timeout}s; {err.strip()[-200:]}"
    return "Broken", f"exit {rc}; {err.strip()[-200:]}"


# ---------------------------------------------------------------------------
# Capability checks (seeded for Phase 52; Phase 54 expands the registry).
# Each check is local-only at this stage (Phase 53 adds the local emulators;
# Phase 54 adds the live-AWS tier for the headline E2E).
# ---------------------------------------------------------------------------

def _check_contract_schema_validation() -> Tuple[Status, str]:
    """CAP-001: contract.schema.json validates sample contracts."""
    return _check_subprocess([
        "python3", "-c",
        "import json, yaml, jsonschema; "
        "s=json.load(open('schemas/contract.schema.json')); "
        "[jsonschema.validate(yaml.safe_load(open(f)), s) "
        " for f in ['contracts/static-assets.yaml','contracts/microservice.yaml']]; "
        "print('2 sample contracts validate')",
    ])


def _check_environment_schema_validation() -> Tuple[Status, str]:
    """CAP-002: environment.schema.json validates the env files."""
    return _check_subprocess([
        "python3", "-c",
        "import json, jsonschema; "
        "s=json.load(open('schemas/environment.schema.json')); "
        "[jsonschema.validate(json.load(open(f)), s) "
        " for f in ['core/environments/dev.json']]; "
        "print('env schema validates')",
    ])


def _check_resolver_static_assets() -> Tuple[Status, str]:
    """CAP-003: contract_resolver resolves static-assets to a Target Stack."""
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as t:
        out = t.name
    try:
        return _check_subprocess([
            "python3", "core/contract_resolver.py",
            "contracts/static-assets.yaml", out,
        ])
    finally:
        try:
            os.unlink(out)
        except OSError:
            pass


def _check_resolver_microservice() -> Tuple[Status, str]:
    """CAP-004: contract_resolver resolves the microservice contract."""
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as t:
        out = t.name
    try:
        return _check_subprocess([
            "python3", "core/contract_resolver.py",
            "contracts/microservice.yaml", out,
        ])
    finally:
        try:
            os.unlink(out)
        except OSError:
            pass


def _check_adapter_emits_terraform() -> Tuple[Status, str]:
    """CAP-005: terraform adapter compiles a resolved stack to .tf files."""
    work = tempfile.mkdtemp(prefix="acdl_regr_")
    stack_path = os.path.join(work, "stack.json")
    tf_dir = os.path.join(work, "tf")
    os.makedirs(tf_dir, exist_ok=True)
    rc, out, err = _run_subprocess([
        "python3", "core/contract_resolver.py",
        "contracts/static-assets.yaml", stack_path,
    ])
    if rc != 0:
        return "Broken", f"resolver failed: {err.strip()[-200:]}"
    status, detail = _check_subprocess([
        "python3", "adapters/terraform/adapter.py", stack_path, tf_dir,
    ])
    if status == "Verified":
        main_tf = os.path.join(tf_dir, "main.tf")
        if not os.path.isfile(main_tf) or os.path.getsize(main_tf) == 0:
            return "Broken", "adapter exited 0 but main.tf missing/empty"
    return status, detail


def _check_interpolation() -> Tuple[Status, str]:
    """CAP-006: contract interpolation expands ${env.*} / ${contract.*}."""
    return _check_subprocess([
        "python3", "-c",
        "import sys; sys.path.insert(0,'.'); "
        "from core.contract_resolver import _expand_vars; "
        "ctx={'env':{'environment':'qa','account_id':'123'},'contract':{'module':'ms'}}; "
        "assert _expand_vars('acdl-${env.environment}-${contract.module}', ctx)=='acdl-qa-ms'; "
        "print('interpolation ok')",
    ])


def _check_confidence_signal() -> Tuple[Status, str]:
    """CAP-007: confidence_signal.compute returns a band for a pass/fail input."""
    return _check_subprocess([
        "python3", "-c",
        "import sys, json; sys.path.insert(0,'.'); "
        "import core.confidence_signal as c; "
        "inputs={'policy':[],'validation':{'schema':True,'stack_resolved':True,'tf_validated':True,'tf_planned':True},'freshness':{'age_days':0,'max_age_days':7},'source':{'submitter':'consumer','commit_sha':'x','signed':False},'history':{'prior_rollbacks':0,'prior_policy_fails':0},'nfrs':{'conformance':None}}; "
        "sig=c.compute('cid','dev',inputs); "
        "assert sig.band in ('pass','warn','fail'); "
        "print(f'confidence band={sig.band}')",
    ])


def _check_outbox_writer() -> Tuple[Status, str]:
    """CAP-008: outbox_writer writes a hash-chained event to a temp file."""
    work = tempfile.mkdtemp(prefix="acdl_outbox_")
    event_path = os.path.join(work, "event.json")
    event = {
        "contractId": "regression-test", "eventType": "CONFIDENCE_COMPUTED",
        "ts": "2026-07-27T00:00:00Z", "environment": "dev",
        "stack": "regression", "score": 0.9, "band": "pass",
        "prev_event_hash": "GENESIS",
    }
    with open(event_path, "w") as f:
        json.dump(event, f)
    # The outbox writer writes to DynamoDB in prod; for the regression we
    # verify the hash-chain logic (the testable core) without AWS. The
    # actual DynamoDB write is a live-AWS concern, deferred to Phase 54.
    return _check_subprocess([
        "python3", "-c",
        f"import sys, json; sys.path.insert(0,'.'); "
        f"import core.outbox_writer as w; "
        f"ev=json.load(open('{event_path}')); "
        f"h=w._canonical_hash(ev); "
        f"assert len(h)==64; "
        f"assert w._canonical_hash(ev)==h; "
        f"print('outbox hash chain ok')",
    ])


def _check_pytest_offline() -> Tuple[Status, str]:
    """CAP-009: the offline pytest suite passes (the regression baseline).

    Excludes slow tests (which invoke the full pipeline) and the
    regression test itself (to avoid recursion: this check runs inside
    the regression run)."""
    return _check_subprocess(
        ["python3", "-m", "pytest", "tests/", "-q", "--tb=line",
         "-m", "not slow",
         "--ignore=tests/test_contract_ingestor.py",
         "--ignore=tests/test_verify_regression_mode.py"],
        timeout=180,
    )


def _check_run_ci_check_only() -> Tuple[Status, str]:
    """CAP-010: run_ci.sh reproduces the CI pipeline locally (offline).

    Excluded from the regression's own pytest invocation to avoid
    recursion; invoked directly here."""
    return _check_subprocess(
        ["bash", "scripts/run_ci.sh", "--quiet"], timeout=240,
    )


# Registry: ordered, each entry is (capability_id, name, tier, check_fn).
# Phase 52 seeds this with 10 local-tier checks; Phase 54 expands it to
# cover every v1.1->v1.8 advertised capability and adds the live-AWS tier
# for the headline E2E.
CAPABILITY_REGISTRY: List[Tuple[str, str, str, Callable[[], Tuple[Status, str]]]] = [
    ("CAP-001", "contract.schema.json validates sample contracts", "local",
     _check_contract_schema_validation),
    ("CAP-002", "environment.schema.json validates env files", "local",
     _check_environment_schema_validation),
    ("CAP-003", "contract_resolver resolves static-assets", "local",
     _check_resolver_static_assets),
    ("CAP-004", "contract_resolver resolves microservice", "local",
     _check_resolver_microservice),
    ("CAP-005", "terraform adapter emits .tf files", "local",
     _check_adapter_emits_terraform),
    ("CAP-006", "contract interpolation expands env/contract tokens", "local",
     _check_interpolation),
    ("CAP-007", "confidence_signal.compute returns a band", "local",
     _check_confidence_signal),
    ("CAP-008", "outbox_writer builds a hash-chained item", "local",
     _check_outbox_writer),
    ("CAP-009", "offline pytest suite passes", "local",
     _check_pytest_offline),
    ("CAP-010", "run_ci.sh reproduces CI pipeline locally", "local",
     _check_run_ci_check_only),
]


def run_regression(milestone: str = "v1.10", phase: int = 52,
                   registry: Optional[List] = None) -> RegressionReport:
    """Run every capability check in the registry; return a RegressionReport."""
    reg = registry if registry is not None else CAPABILITY_REGISTRY
    run_id = f"regr-{int(time.time())}"
    run_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    report = RegressionReport(run_id=run_id, run_at_utc=run_at,
                              milestone=milestone, phase=phase)
    for cap_id, name, tier, fn in reg:
        t0 = time.monotonic()
        try:
            status, detail = fn()
        except Exception as e:  # noqa: BLE001
            status, detail = "Broken", f"check raised: {type(e).__name__}: {e}"[:300]
        dur = int((time.monotonic() - t0) * 1000)
        report.results.append(CapabilityResult(
            capability_id=cap_id, name=name, status=status,
            detail=detail, tier=tier, duration_ms=dur,
        ))
    return report


def write_report(report: RegressionReport,
                 md_path: Optional[Path] = None,
                 json_path: Optional[Path] = None) -> Tuple[Path, Path]:
    """Write the report to .ciagent/REGRESSION_REPORT.md + .json."""
    md_path = md_path or (CIAgent / "REGRESSION_REPORT.md")
    json_path = json_path or (CIAgent / "REGRESSION_REPORT.json")
    json_path.write_text(json.dumps(report.to_dict(), indent=2))
    lines = [
        f"# Regression Report — {report.milestone} Phase {report.phase}",
        "",
        f"- **Run ID:** `{report.run_id}`",
        f"- **Run at (UTC):** {report.run_at_utc}",
        f"- **Summary:** {report.summary}",
        f"- **Passed (milestone gate):** {report.passed}",
        "",
        "| Capability | Name | Tier | Status | Duration (ms) | Detail |",
        "|-----------|------|------|--------|--------------|--------|",
    ]
    for r in report.results:
        lines.append(
            f"| {r.capability_id} | {r.name} | {r.tier} | "
            f"**{r.status}** | {r.duration_ms} | {r.detail[:160]} |"
        )
    md_path.write_text("\n".join(lines) + "\n")
    return md_path, json_path


def main() -> int:
    milestone = os.environ.get("ACDL_REGRESSION_MILESTONE", "v1.10")
    phase = int(os.environ.get("ACDL_REGRESSION_PHASE", "52"))
    report = run_regression(milestone=milestone, phase=phase)
    md, js = write_report(report)
    print(f"regression: {report.summary} -> {md}")
    if not report.passed:
        print("FAIL: regression surfaced non-Verified capabilities "
              "(milestone gate blocks)", file=sys.stderr)
        return 1
    print("regression: all capabilities Verified (milestone gate passes)")
    return 0


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