210 lines
8.0 KiB
Python
210 lines
8.0 KiB
Python
"""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('atlas/titan-iac','origin','acceptance1',e.CONFIRMATION,expected_head='1'*40,expected_base_sha='2'*40)\n"
|
|
"assert e._cleanup(R(),object(),q,'/api/v1/repos/atlas/titan-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('atlas/titan-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",
|
|
),
|
|
)
|
|
|
|
|
|
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
|