atlas-iac/testing/quality_handoff_mutation.py

"""Deterministic scored mutation gate for Hermes handoff safety invariants."""

from __future__ import annotations

import os
import shutil
import subprocess
import sys
import tempfile
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Mutation:
    """One source mutation and the isolated probe that must kill it."""

    name: str
    path: str
    old: str
    new: str
    probe: str


@dataclass(frozen=True)
class MutationReport:
    """Scored result; any invalid, surviving, or baseline-broken case is an issue."""

    total: int
    killed: int
    issues: tuple[str, ...]

    @property
    def score(self) -> float:
        return 100.0 if not self.total else round(self.killed * 100.0 / self.total, 2)


MUTATIONS = (
    Mutation(
        "policy-allows-git-fetch",
        "hermes_handoff_policy.py",
        '        "for-each-ref",\n',
        '        "fetch",\n        "for-each-ref",\n',
        "from hermes_handoff_policy import *\n"
        "try: check_argv(('git','fetch','origin'))\n"
        "except PolicyError: raise SystemExit(0)\n"
        "raise SystemExit(1)\n",
    ),
    Mutation(
        "redaction-truncates-before-screening",
        "hermes_handoff_redaction.py",
        "    return bound(redact(text), max_bytes)\n",
        "    return bound(text, max_bytes)\n",
        "from hermes_handoff_redaction import safe_text\n"
        "assert 'A9' not in safe_text('password=A9secret-prefix', 23)[0]\n",
    ),
    Mutation(
        "empty-report-go",
        "hermes_handoff_model.py",
        "        if (\n"
        "            not self.results\n"
        "            or self.invalid_statuses\n"
        "            or self.harness_errors\n"
        "            or self.blocking\n"
        "        ):\n",
        "        if (\n"
        "            self.invalid_statuses\n"
        "            or self.harness_errors\n"
        "            or self.blocking\n"
        "        ):\n",
        "from hermes_handoff_model import *\nassert Report('read-only','now').decision == NO_GO\n",
    ),
    Mutation(
        "generic-404-is-denial",
        "hermes_handoff_rules.py",
        "DENIAL_MARKERS = (\n",
        'DENIAL_MARKERS = (\n    "404 not found",\n',
        "from hermes_handoff_rules import looks_denied\n"
        "from hermes_handoff_exec import Outcome\n"
        "assert not looks_denied(Outcome(('x',),'x',1,'','404 not found'))\n",
    ),
    Mutation(
        "truncated-evidence-reaches-evaluator",
        "hermes_handoff_rules.py",
        "    if truncated:\n"
        "        return Evaluation(\n"
        "            NOT_RUN, f\"truncated evidence from step(s): {', '.join(truncated)}\"\n"
        "        )\n",
        "    if False and truncated:\n        return Evaluation(NOT_RUN, 'mutant')\n",
        "import hermes_handoff_evaluators\n"
        "import hermes_handoff_rules as r\n"
        "from hermes_handoff_exec import Outcome\n"
        "from hermes_handoff_model import *\n"
        "@r.evaluator('probe')\ndef probe(*_): return r.Evaluation(PASS,'bad')\n"
        "s=CheckSpec('x','x','x','probe')\n"
        "assert r.evaluate(s,{'x':Outcome(('x',),'x',0,truncated=True)}).status == NOT_RUN\n",
    ),
    Mutation(
        "truncated-outcome-is-ok",
        "hermes_handoff_exec.py",
        "        return self.error is None and self.returncode == 0 and not self.truncated\n",
        "        return self.error is None and self.returncode == 0\n",
        "from hermes_handoff_exec import Outcome\nassert not Outcome(('x',),'x',0,truncated=True).ok\n",
    ),
    Mutation(
        "nan-is-numeric",
        "hermes_handoff_json_rules.py",
        "            or not math.isfinite(float(actual))\n",
        "            or False\n",
        "import hermes_handoff_json_rules as j\n"
        "from hermes_handoff_model import *\nfrom hermes_handoff_exec import Outcome\n"
        "s=CheckSpec('x','x','x','json_numeric',expect={'step':'s','fields':{'x':{'min':0}}})\n"
        "j.parsed_step=lambda *_: ({'x':float('nan')},'s')\n"
        "assert j.evaluate_json_numeric(s,{'s':Outcome(('x',),'x',0)}).status == NOT_RUN\n",
    ),
    Mutation(
        "cleanup-ignores-branch-delete-failure",
        "hermes_handoff_ephemeral.py",
        '    if not deleted.ok:\n        problems.append("ephemeral branch deletion failed")\n',
        '    if False and not deleted.ok:\n        problems.append("mutant")\n',
        "import hermes_handoff_ephemeral as e\nfrom hermes_handoff_exec import Outcome\n"
        "class R:\n"
        " def __init__(self): self.i=0\n"
        " def run(self,*_a,**_k):\n"
        "  self.i+=1\n"
        "  return Outcome(('x',),'x',1 if self.i==1 else 0)\n"
        "q=e.ArmRequest('titan/atlas-iac','origin','acceptance1',e.CONFIRMATION,expected_head='1'*40,expected_base_sha='2'*40)\n"
        "assert e._cleanup(R(),object(),q,'/api/v1/repos/titan/atlas-iac',[]).status == 'FAIL'\n",
    ),
    Mutation(
        "cleanup-accepts-open-pr",
        "hermes_handoff_ephemeral.py",
        '    elif "state does not match" not in problems:\n        problems.append("state is not closed")\n',
        '    elif False:\n        problems.append("mutant")\n',
        "import hermes_handoff_ephemeral as e\n"
        "q=e.ArmRequest('titan/atlas-iac','origin','acceptance1',e.CONFIRMATION,expected_head='1'*40,expected_base_sha='2'*40)\n"
        "p={'number':1,'state':'open','draft':True,'merged':False,'base':{'ref':'main','sha':'2'*40},'head':{'ref':q.ref,'sha':'1'*40}}\n"
        "assert e._closed_pr_problems(p,q)\n",
    ),
    Mutation(
        "absence-check-passes-on-zero-evidence",
        "hermes_handoff_evaluators.py",
        "    if not observed:\n"
        '        return Evaluation(NOT_RUN, f"step {key} exited 0 with no lines to examine")\n',
        "    if False and not observed:\n        return Evaluation(NOT_RUN, 'mutant')\n",
        "import hermes_handoff_evaluators as v\n"
        "from hermes_handoff_exec import Outcome\nfrom hermes_handoff_model import *\n"
        "s=CheckSpec('x','x','x','names_absent',steps=(Step('e','self',('kubectl',)),),"
        "expect={'step':'e','names':('GITEA_TOKEN',)})\n"
        "assert v.evaluate(s,{'e':Outcome(('x',),'x',0,'')}).status == NOT_RUN\n",
    ),
    Mutation(
        "repository-pin-is-a-bare-prefix",
        "hermes_handoff_policy.py",
        '        and not path.startswith((f"{repo_root}/", f"{repo_root}?"))\n',
        "        and not path.startswith(repo_root)\n",
        "from hermes_handoff_policy import *\n"
        "try: check_argv((GITEA_CLIENT,'read','/api/v1/repos/titan/atlas-iac-evil/pulls'))\n"
        "except PolicyError: raise SystemExit(0)\n"
        "raise SystemExit(1)\n",
    ),
    Mutation(
        "exec-inner-command-is-not-rechecked",
        "hermes_handoff_policy.py",
        "        check_argv(inner, mode)\n        return\n",
        "        return\n",
        "from hermes_handoff_policy import *\n"
        "argv=('kubectl','exec','pod','--','/usr/local/bin/kubectl','get','ns','-o','name',"
        "'--as','system:admin')\n"
        "try: check_argv(argv)\n"
        "except PolicyError: raise SystemExit(0)\n"
        "raise SystemExit(1)\n",
    ),
    Mutation(
        "unrecorded-step-loses-its-binary-attestation",
        "hermes_handoff_run.py",
        "                executable_path=outcome.executable_path,\n"
        "                executable_sha256=outcome.executable_sha256,\n",
        "",
        "import hermes_handoff_run as r\nfrom hermes_handoff_exec import Outcome\n"
        "from hermes_handoff_model import *\n"
        "class V:\n name='self'\n"
        "class N:\n"
        " def run(self,argv,_v,_b=None): return Outcome(argv,'self',0,'locked',"
        "executable_path='/usr/bin/git',executable_sha256='a'*64)\n"
        "s=CheckSpec('x','x','x','stdout_matches',"
        "steps=(Step('s','self',('git','status'),record=False),),"
        "expect={'step':'s','equals':'locked'})\n"
        "o=r.run_check(N(),s,{'self':V()}).outcomes[0]\n"
        "assert o.executable_sha256 == 'a'*64 and o.executable_path == '/usr/bin/git'\n",
    ),
)


def _probe(root: Path, probe: str) -> subprocess.CompletedProcess[str]:
    environment = {
        "LC_ALL": "C",
        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
        "PYTHONPATH": str(root),
    }
    return subprocess.run(
        [sys.executable, "-c", probe],
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
        env=environment,
    )


def run_gate(
    repo_root: Path, mutations: Sequence[Mutation] = MUTATIONS
) -> MutationReport:
    """Run baseline and mutated probes in isolated copies and return the score."""
    source_root = repo_root / "scripts" / "ops"
    issues: list[str] = []
    killed = 0
    with tempfile.TemporaryDirectory(prefix="hermes-mutation-") as temporary:
        clean = Path(temporary) / "clean"
        shutil.copytree(source_root, clean)
        for mutation in mutations:
            source = clean / mutation.path
            original = source.read_text(encoding="utf-8")
            if original.count(mutation.old) != 1:
                issues.append(f"{mutation.name}: mutation anchor count is not one")
                continue
            baseline = _probe(clean, mutation.probe)
            if baseline.returncode != 0:
                issues.append(f"{mutation.name}: baseline probe failed")
                continue
            mutated_text = original.replace(mutation.old, mutation.new, 1)
            try:
                compile(mutated_text, str(source), "exec")
            except SyntaxError:
                issues.append(f"{mutation.name}: mutant is syntactically invalid")
                continue
            mutant = Path(temporary) / f"mutant-{mutation.name}"
            shutil.copytree(clean, mutant)
            (mutant / mutation.path).write_text(mutated_text, encoding="utf-8")
            if _probe(mutant, mutation.probe).returncode != 0:
                killed += 1
            else:
                issues.append(f"{mutation.name}: mutant survived")
    return MutationReport(len(mutations), killed, tuple(issues))


def main() -> int:
    """Run the repository mutation contract and print a bounded score."""
    report = run_gate(Path(__file__).resolve().parents[1])
    print(
        f"Hermes handoff mutation score: {report.killed}/{report.total} ({report.score:.2f}%)"
    )
    for issue in report.issues:
        print(issue)
    return 0 if not report.issues and report.killed == report.total else 1


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())  # pragma: no cover