diff --git a/.github/workflows/c6-residual-closure.yml b/.github/workflows/c6-residual-closure.yml index c99afc4f1..c8577515b 100644 --- a/.github/workflows/c6-residual-closure.yml +++ b/.github/workflows/c6-residual-closure.yml @@ -5,6 +5,9 @@ name: C6 residual closure checks paths: - 'reviews/c6_residual_closure_20260930/**' - '.github/workflows/c6-residual-closure.yml' + - 'tools/c6_residual_child_contract.py' + - 'tools/c6_residual_child_contract/**' + - 'tests/test_c6_residual_child_contract.py' - 'frontiers/spectral_cluster_closure_20260929/PROOF.md' - 'frontiers/spectral_cluster_closure_20260929/REVIEW_RECORD.md' - 'frontiers/spectral_cluster_closure_20260929/README.md' @@ -34,6 +37,10 @@ jobs: - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 with: python-version: '3.11.16' + - name: Refuse unrelated C6 residual child failures in both Python modes + run: | + python -B -S -m unittest discover -s tests -p test_c6_residual_child_contract.py -v + python -B -O -S -m unittest discover -s tests -p test_c6_residual_child_contract.py -v - name: Verify sources, replay both modes (baseline or installed register), reject mutants env: PYTHONDONTWRITEBYTECODE: '1' @@ -41,6 +48,8 @@ jobs: set -euo pipefail python -B -S - <<'PY' import hashlib, json, pathlib, subprocess, sys + sys.path.insert(0, str(pathlib.Path('tools').resolve())) + from c6_residual_child_contract import validate_baseline, validate_mutant root = pathlib.Path('reviews/c6_residual_closure_20260930').resolve() manifest = json.loads((root / 'SOURCE_FILES.json').read_text())['files'] present = sorted(p.name for p in root.iterdir()) @@ -57,19 +66,21 @@ jobs: script = str(root / 'residual_check.py') mutants = ['partial-install', 'allow-symlink', 'no-hash', 'stale-fingerprint', 'drop-required-edge', 'executed-flag', 'controlling-true', 'tail-reversed', 'pair-identity-broken', 'route-c-broken', 'drop-review-needle', 'installed-drift', 'snapshot-omitted', 'record-unreviewed', 'reviewed-text-drift'] for flags in (['-B', '-S'], ['-B', '-O', '-S']): - out = subprocess.run([sys.executable, *flags, script], - check=True, capture_output=True, timeout=300).stdout + base_proc = subprocess.run([sys.executable, *flags, script], + capture_output=True, timeout=300) + out = base_proc.stdout state = json.loads(out).get('register_state') if state not in ('baseline', 'installed'): raise RuntimeError('checker reports register_state ' + repr(state)) pinned = 'RESULTS.json' if state == 'baseline' else 'RESULTS_INSTALLED.json' - if out != (root / pinned).read_bytes(): + expected = (root / pinned).read_bytes() + if out != expected: raise RuntimeError('output differs from ' + pinned + ' under ' + ' '.join(flags)) + baseline = validate_baseline(base_proc, expected) for m in mutants: - rc = subprocess.run([sys.executable, *flags, script, '--mutant', m], - capture_output=True, timeout=300).returncode - if rc != 1: - raise RuntimeError('mutant ' + m + ' not rejected') + proc = subprocess.run([sys.executable, *flags, script, '--mutant', m], + capture_output=True, timeout=300) + validate_mutant(m, proc, baseline) print(json.dumps({'scientific_effect': 'NONE', 'mathematical_acceptance': False})) PY git diff --exit-code diff --git a/tests/test_c6_residual_child_contract.py b/tests/test_c6_residual_child_contract.py new file mode 100644 index 000000000..ab3f2152c --- /dev/null +++ b/tests/test_c6_residual_child_contract.py @@ -0,0 +1,170 @@ +"""External controls for C6 residual child output admissibility (no proof edits).""" +import copy +import importlib.util +import json +import pathlib +import shutil +import subprocess +import sys +import tempfile +import unittest +from subprocess import CompletedProcess + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / 'tools')) +from c6_residual_child_contract import (CHECK_NAMES, EXPECTED_FAILURES_BY_STATE, + validate_baseline, validate_mutant) + + +def packet(state='baseline'): + return dict(object='C6-RESIDUAL-CLOSURE-20260930-v1', + checks={name: True for name in CHECK_NAMES}, passed=True, + inventory_files=13, component_nodes=10, scope='pinned mathematical scope', + register_state=state, live_state={'state': state}, gate={}) + + +def output(payload, *, returncode=1, stderr=b''): + return CompletedProcess(['residual_check.py'], returncode, + stdout=json.dumps(payload, sort_keys=True).encode(), stderr=stderr) + + +class ChildContracts(unittest.TestCase): + def test_baseline_in_both_register_states(self): + for state in ('baseline', 'installed'): + original = packet(state) + out = output(original, returncode=0) + with self.subTest(state=state): + self.assertEqual(validate_baseline(out, out.stdout), original) + with self.assertRaises(ValueError): + validate_baseline(output(original, returncode=0, stderr=b'warning'), out.stdout) + with self.assertRaises(ValueError): + validate_baseline(output(original, returncode=0), b'changed') + + def test_all_fifteen_frozen_mutant_failure_sets_in_both_states(self): + for state in ('baseline', 'installed'): + baseline = packet(state) + for mutant, expected_failed in EXPECTED_FAILURES_BY_STATE[state].items(): + result = copy.deepcopy(baseline) + result['passed'] = False + for name in expected_failed: + result['checks'][name] = False + if 'LIVE' in expected_failed: + result['register_state'] = 'partial' + result['live_state']['state'] = 'partial' + with self.subTest(state=state, mutant=mutant): + self.assertEqual(validate_mutant(mutant, output(result), baseline), result) + bad = copy.deepcopy(result) + bad['checks'] = dict.fromkeys(CHECK_NAMES, True) + bad['checks']['IDENTITIES'] = False + if expected_failed != frozenset(('IDENTITIES',)): + with self.assertRaises(ValueError): + validate_mutant(mutant, output(bad), baseline) + if 'LIVE' not in expected_failed: + bad_state = copy.deepcopy(result) + bad_state['register_state'] = 'partial' + bad_state['live_state']['state'] = 'partial' + with self.assertRaises(ValueError): + validate_mutant(mutant, output(bad_state), baseline) + + def test_three_state_dependent_mutants_against_checker_installed_graph(self): + repo = pathlib.Path(__file__).resolve().parents[1] + checker_path = (repo / 'reviews/c6_residual_closure_20260930' / + 'residual_check.py') + module_spec = importlib.util.spec_from_file_location('c6_residual_checker', + checker_path) + checker = importlib.util.module_from_spec(module_spec) + module_spec.loader.exec_module(checker) + + source_graph = json.loads( + (repo / 'frontiers/downstream_gate_20260925/GRAPH.json').read_text()) + proposal = json.loads( + (repo / 'reviews/c6_residual_closure_20260930' / + 'PROPOSED_TRANSITIONS.json').read_text()) + installed = checker.build(proposal, source_graph, 'final')[1] + + with tempfile.TemporaryDirectory() as tmp: + root = pathlib.Path(tmp) + shutil.copytree(repo / 'frontiers', root / 'frontiers') + shutil.copytree(repo / 'reviews', root / 'reviews') + graph_path = root / 'frontiers/downstream_gate_20260925/GRAPH.json' + graph_path.write_text(json.dumps(installed, indent=2, sort_keys=True) + '\n') + script = root / 'reviews/c6_residual_closure_20260930/residual_check.py' + pinned = (root / 'reviews/c6_residual_closure_20260930' / + 'RESULTS_INSTALLED.json').read_bytes() + + for flags in (('-B', '-S'), ('-B', '-O', '-S')): + base_proc = subprocess.run( + [sys.executable, *flags, str(script)], + cwd=root, capture_output=True, timeout=300) + baseline = validate_baseline(base_proc, pinned) + self.assertEqual(baseline['register_state'], 'installed') + for mutant in ('stale-fingerprint', 'record-unreviewed', + 'installed-drift'): + proc = subprocess.run( + [sys.executable, *flags, str(script), '--mutant', mutant], + cwd=root, capture_output=True, timeout=300) + with self.subTest(flags=flags, mutant=mutant): + validate_mutant(mutant, proc, baseline) + + def test_crash_silence_wrong_exit_and_extra_stderr_are_refused(self): + baseline = packet() + example = copy.deepcopy(baseline) + example['passed'] = False + example['checks']['DEDUCTION'] = False + for child in [CompletedProcess([], 1, b'', b''), + CompletedProcess([], 1, b'', b'Traceback: unrelated crash'), + output(example, returncode=0), + output(example, stderr=b'warning'), + CompletedProcess([], 1, b'not-json', b'')]: + with self.subTest(child=child): + with self.assertRaises(ValueError): + validate_mutant('tail-reversed', child, baseline) + + def test_structural_invalidity_and_type_confusion(self): + baseline = packet() + good = copy.deepcopy(baseline) + good['passed'] = False + good['checks']['DEDUCTION'] = False + mutations = [] + for key, value in [('object','DIFFERENT'), ('passed',0), ('register_state','fake'), + ('scope','changed'), ('inventory_files',True), + ('component_nodes','10'), ('gate',None)]: + changed = copy.deepcopy(good) + changed[key] = value + mutations.append(changed) + changed = copy.deepcopy(good) + changed['checks']['DEDUCTION'] = 0 + mutations.append(changed) + changed = copy.deepcopy(good) + del changed['checks']['DEDUCTION'] + mutations.append(changed) + changed = copy.deepcopy(good) + changed['checks']['OTHER'] = False + mutations.append(changed) + changed = copy.deepcopy(good) + changed['extra'] = 'field' + mutations.append(changed) + changed = copy.deepcopy(good) + changed['live_state']['state'] = 'installed' + mutations.append(changed) + for changed in mutations: + with self.subTest(changed=changed): + with self.assertRaises(ValueError): + validate_mutant('tail-reversed', output(changed), baseline) + + def test_workflow_wires_strict_child_validation(self): + workflow = (pathlib.Path(__file__).resolve().parents[1] / + '.github/workflows/c6-residual-closure.yml').read_text() + self.assertIn('from c6_residual_child_contract import validate_baseline, validate_mutant', workflow) + self.assertIn('validate_baseline(base_proc, expected)', workflow) + self.assertIn('validate_mutant(m, proc, baseline)', workflow) + + def test_duplicate_fields_nonfinite_and_extra_json(self): + baseline = packet() + for data in (b'{"passed":false,"passed":false}', b'{"x":NaN}', + b'{"x":Infinity}', b'{}{}'): + with self.subTest(data=data): + with self.assertRaises(ValueError): + validate_mutant('tail-reversed', CompletedProcess([],1,data,b''), baseline) + +if __name__ == '__main__': + unittest.main() diff --git a/tools/c6_residual_child_contract.py b/tools/c6_residual_child_contract.py new file mode 100644 index 000000000..31b1407f8 --- /dev/null +++ b/tools/c6_residual_child_contract.py @@ -0,0 +1,126 @@ +"""Strict child-result contracts for the C6 residual closure workflow. + +Engineering guard only. The eight named checks and expected mutant failure +sets are derived from the pinned residual_check.py branches. Baseline-state +sets use the original Python 3.11.16 hosted observations; installed-state +overrides use checker-constructed installed-graph replays. No mathematical +proof is rerun here. +""" + +import json + +CHECK_NAMES = frozenset(( + "IDENTITIES", "VERDICTS", "LIVE", "REVIEWED", "DEDUCTION", + "TRANSITIONS", "GATE", "NEGATIVES", +)) +EXPECTED_FAILURES = { + "partial-install": frozenset(("TRANSITIONS",)), + "allow-symlink": frozenset(("NEGATIVES",)), + "no-hash": frozenset(("NEGATIVES",)), + "stale-fingerprint": frozenset(("TRANSITIONS", "GATE")), + "drop-required-edge": frozenset(("TRANSITIONS", "GATE")), + "executed-flag": frozenset(("TRANSITIONS", "GATE")), + "controlling-true": frozenset(("TRANSITIONS", "GATE")), + "tail-reversed": frozenset(("DEDUCTION",)), + "pair-identity-broken": frozenset(("DEDUCTION",)), + "route-c-broken": frozenset(("DEDUCTION",)), + "drop-review-needle": frozenset(("VERDICTS",)), + "installed-drift": frozenset(("TRANSITIONS", "GATE")), + "snapshot-omitted": frozenset(("GATE",)), + "record-unreviewed": frozenset(("REVIEWED", "TRANSITIONS", "GATE")), + "reviewed-text-drift": frozenset(("REVIEWED",)), +} +EXPECTED_FAILURES_BY_STATE = { + "baseline": EXPECTED_FAILURES, + "installed": dict(EXPECTED_FAILURES, **{ + "stale-fingerprint": frozenset(("GATE", "LIVE", "TRANSITIONS")), + "installed-drift": frozenset(("TRANSITIONS",)), + "record-unreviewed": frozenset(("GATE", "LIVE", "REVIEWED", + "TRANSITIONS")), + }), +} +REPORT_KEYS = frozenset(( + "checks", "component_nodes", "gate", "inventory_files", "live_state", + "object", "passed", "register_state", "scope", +)) + + +def _require(predicate, message): + if not predicate: + raise ValueError("C6 residual child contract: " + message) + + +def _loads(raw): + _require(type(raw) is bytes, "stdout must be bytes") + + def unique(pairs): + obj = {} + for key, value in pairs: + _require(key not in obj, "duplicate JSON key: " + key) + obj[key] = value + return obj + + def reject_constant(value): + raise ValueError("nonfinite JSON constant: " + value) + + try: + return json.loads(raw, object_pairs_hook=unique, parse_constant=reject_constant) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError("C6 residual child contract: invalid JSON stdout") from exc + + +def _check_common(report, baseline, expected_failures=frozenset()): + _require(type(report) is dict and set(report) == REPORT_KEYS, + "missing or extra report fields") + _require(report["object"] == baseline["object"] == "C6-RESIDUAL-CLOSURE-20260930-v1", + "incorrect object") + _require(type(report["inventory_files"]) is int and report["inventory_files"] == baseline["inventory_files"], + "inventory identity changed") + _require(type(report["component_nodes"]) is int and report["component_nodes"] == baseline["component_nodes"], + "component identity changed") + _require(type(report["scope"]) is str and report["scope"] == baseline["scope"], + "source scope changed") + expected_state = ("partial" if "LIVE" in expected_failures + else baseline["register_state"]) + _require(baseline["register_state"] in ("baseline", "installed") and + report["register_state"] == expected_state, + "register state changed") + _require(type(report["live_state"]) is dict and + report["live_state"].get("state") == report["register_state"], + "live state does not match register state") + _require(type(report["gate"]) is dict, "gate report must be an object") + checks = report["checks"] + _require(type(checks) is dict and set(checks) == CHECK_NAMES and + all(type(v) is bool for v in checks.values()), + "invalid or incomplete eight-check result") + _require(type(report["passed"]) is bool, "passed must be a Boolean") + return checks + + +def validate_baseline(process, pinned_bytes): + """Bind the genuine zero-mutant result to its already-pinned source bytes.""" + _require(process.returncode == 0, "baseline exit must be zero") + _require(process.stderr == b"", "baseline wrote unexpected stderr") + _require(process.stdout == pinned_bytes, "baseline stdout differs from pinned result") + report = _loads(process.stdout) + _check_common(report, report) + _require(report["passed"] is True and all(report["checks"].values()), + "baseline must pass all eight checks") + return report + + +def validate_mutant(mutant, process, baseline): + """Reject crashes, silence, malformed output and failures for the wrong reason.""" + register_state = baseline.get("register_state") + _require(register_state in EXPECTED_FAILURES_BY_STATE, + "unregistered baseline state") + expected = EXPECTED_FAILURES_BY_STATE[register_state] + _require(mutant in expected, "unregistered mutant label") + _require(process.returncode == 1, "mutant exit must be one") + _require(process.stderr == b"", "mutant wrote unexpected stderr") + report = _loads(process.stdout) + checks = _check_common(report, baseline, expected[mutant]) + failed = frozenset(name for name, value in checks.items() if value is False) + _require(report["passed"] is False and failed == expected[mutant], + "mutant failed a different check set: " + mutant) + return report