C6 residual: reject unrelated mutant failures with typed child-result contracts

.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,13 +37,19 @@ 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'
         run: |
           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

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()

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