diff --git a/.github/workflows/qs-d3-soft-layer-chain.yml b/.github/workflows/qs-d3-soft-layer-chain.yml index 2fcf36c84..1a9ed5a45 100644 --- a/.github/workflows/qs-d3-soft-layer-chain.yml +++ b/.github/workflows/qs-d3-soft-layer-chain.yml @@ -5,6 +5,7 @@ name: QS d = 3 soft-layer chain (A3-A3.6) replay paths: - 'frontiers/qs_d3_soft_layer_chain_20261005/**' - 'tests/test_qs_d3_replay_protocol.py' + - 'tests/test_qs_failure_evidence.py' - '.github/workflows/qs-d3-soft-layer-chain.yml' workflow_dispatch: @@ -36,7 +37,8 @@ jobs: out="$RUNNER_TEMP/qs-replay-evidence" mkdir -p "$out/logs" python ${{ matrix.mode }} -m unittest discover -s tests -p test_qs_d3_replay_protocol.py -v 2>&1 | tee "$out/logs/protocol.txt" - python ${{ matrix.mode }} frontiers/qs_d3_soft_layer_chain_20261005/replay.py 2>&1 | tee "$out/logs/replay.txt" + python ${{ matrix.mode }} -m unittest discover -s tests -p test_qs_failure_evidence.py -v 2>&1 | tee "$out/logs/capture-protocol.txt" + QS_REPLAY_EVIDENCE_DIR="$out/children" python ${{ matrix.mode }} frontiers/qs_d3_soft_layer_chain_20261005/replay.py 2>&1 | tee "$out/logs/replay.txt" git diff --exit-code test -z "$(git status --porcelain)" - name: Preserve exact public sources and available execution logs @@ -47,7 +49,7 @@ jobs: out="$RUNNER_TEMP/qs-replay-evidence" mkdir -p "$out" cp -a frontiers/qs_d3_soft_layer_chain_20261005 "$out/packet" - cp tests/test_qs_d3_replay_protocol.py "$out/" + cp tests/test_qs_d3_replay_protocol.py tests/test_qs_failure_evidence.py "$out/" cp .github/workflows/qs-d3-soft-layer-chain.yml "$out/" git rev-parse HEAD > "$out/tested_commit.txt" printf '%s\n' '${{ matrix.mode }}' > "$out/mode.txt" diff --git a/frontiers/qs_d3_soft_layer_chain_20261005/replay.py b/frontiers/qs_d3_soft_layer_chain_20261005/replay.py index cfc71d58a..2b54456b5 100644 --- a/frontiers/qs_d3_soft_layer_chain_20261005/replay.py +++ b/frontiers/qs_d3_soft_layer_chain_20261005/replay.py @@ -22,9 +22,15 @@ Reproducibility only. The analytic notes and their reads carry the mathematics. These finite checks support algebra, as the reads say. Nothing here reads or grades a proof. +Set QS_REPLAY_EVIDENCE_DIR to a NEW directory outside the checkout to retain +raw child streams and process records. Capture is execution evidence, not a verdict. +Without this variable the original invocation/output behavior is unchanged. + Usage: python3 -B -S frontiers/qs_d3_soft_layer_chain_20261005/replay.py (exit 0 iff every check passes) """ import hashlib +import math +import os import json import pathlib import re @@ -32,6 +38,7 @@ import subprocess import sys import tempfile +import time HERE = pathlib.Path(__file__).resolve().parent OWN = {'README.md', 'SOURCES.json', 'replay.py'} @@ -279,12 +286,121 @@ def extraction_failures(root, manifest): return failures +class ChildEvidenceError(RuntimeError): + """Required capture failed; the completed child result remains inspectable.""" + + +class ChildEvidence: + """One fresh external directory, unique records, no output decoding or verdict.""" + + def __init__(self, directory): + if not str(directory).strip(): + raise ValueError('evidence directory must be nonempty') + self.directory = pathlib.Path(directory).resolve() + checkout = HERE.parents[1].resolve() + if self.directory == checkout or checkout in self.directory.parents: + raise ValueError('evidence directory must be outside the checkout') + self.directory.mkdir() # No reuse, overwrite, or implicit parent creation. + self.sequence = 0 + + @staticmethod + def _write(path, data): + with path.open('xb') as stream: + stream.write(data) + + @staticmethod + def _clock(): + try: + value = time.monotonic() + if type(value) not in (int, float) or not math.isfinite(value): + raise ValueError('invalid monotonic clock value') + return value, None + except Exception as exc: + return None, type(exc).__name__ + ': ' + str(exc) + + def _record(self, directory, command, cwd, timeout, start, status, + code, out, err, exception): + stop = self._clock() + clock_error = start[1] or stop[1] + elapsed = None + if clock_error is None: + duration = stop[0] - start[0] + if not math.isfinite(duration) or duration < 0: + clock_error = 'invalid elapsed time' + else: + elapsed = round(duration, 6) # Validate BEFORE rounding. + record = dict(schema=1, sequence=self.sequence, command=list(command), cwd=str(cwd), + timeout_seconds=timeout, status=status, returncode=code, + capture_complete=(status == 'completed'), elapsed_seconds=elapsed, + clock_error=clock_error, exception=exception, + streams={}, evidence_errors=[], + meaning='raw subprocess capture only; no verification verdict') + for name, data in (('stdout', out), ('stderr', err)): + row = dict(available=data is not None, bytes=None, sha256=None, saved=False) + record['streams'][name] = row + if data is None: + continue # Absent capture is not an empty captured byte stream. + if type(data) is not bytes: + record['capture_complete'] = False + record['evidence_errors'].append(name + ': non-bytes capture') + continue + row.update(bytes=len(data), sha256=hashlib.sha256(data).hexdigest()) + try: + self._write(directory / (name + '.bin'), data) + row['saved'] = True + except OSError as exc: + record['evidence_errors'].append(name + ': ' + type(exc).__name__ + ': ' + str(exc)) + try: + self._write(directory / 'process.json', + (json.dumps(record, sort_keys=True, indent=2, allow_nan=False) + '\n').encode('utf-8')) + except OSError as exc: + record['evidence_errors'].append('process.json: ' + type(exc).__name__ + ': ' + str(exc)) + return record + + def run(self, command, *, cwd, timeout): + self.sequence += 1 + directory = self.directory / ('%04d' % self.sequence) + directory.mkdir() # Fail before child execution if the slot is not fresh. + start = self._clock() + try: + proc = subprocess.run(command, capture_output=True, timeout=timeout, cwd=cwd) + except Exception as exc: + is_timeout = isinstance(exc, subprocess.TimeoutExpired) + try: + record = self._record(directory, command, cwd, timeout, start, + 'timeout' if is_timeout else 'execution_error', None, + exc.stdout if is_timeout else None, + exc.stderr if is_timeout else None, + dict(type=type(exc).__name__, message=str(exc))) + exc.qs_evidence = record + if record['evidence_errors']: + exc.add_note('QS capture errors: ' + '; '.join(record['evidence_errors'])) + except Exception as capture_error: + # Even an unexpected capture bug cannot replace the primary exception. + exc.add_note('QS capture could not finish: ' + type(capture_error).__name__ + ': ' + str(capture_error)) + raise + record = self._record(directory, command, cwd, timeout, start, 'completed', + proc.returncode, proc.stdout, proc.stderr, None) + result = proc.returncode, proc.stdout, proc.stderr + if record['evidence_errors']: + error = ChildEvidenceError('QS child evidence could not be completely saved') + error.qs_evidence = record + error.child_result = result + raise error + return result + + +ACTIVE_EVIDENCE = None + + def run(script, *args): + if ACTIVE_EVIDENCE is not None: + return ACTIVE_EVIDENCE.run([sys.executable, *interpreter_flags(), str(script), *args], + timeout=1800, cwd=str(script.parent)) proc = subprocess.run([sys.executable, *interpreter_flags(), str(script), *args], capture_output=True, timeout=1800, cwd=str(script.parent)) return proc.returncode, proc.stdout, proc.stderr - def replay_failures(root, manifest): failures = rejection_inventory_failures(manifest) if failures: @@ -342,7 +458,7 @@ def negative_control_failures(root, manifest): return failures -def main(): +def _replay_main(): manifest = json.loads((HERE / 'SOURCES.json').read_text(encoding='utf-8')) failures = tree_failures(HERE, manifest) if not failures: @@ -366,5 +482,18 @@ def main(): return 0 +def main(): + global ACTIVE_EVIDENCE + directory = os.environ.get('QS_REPLAY_EVIDENCE_DIR') + if directory is None: + return _replay_main() + previous = ACTIVE_EVIDENCE + ACTIVE_EVIDENCE = ChildEvidence(directory) + try: + return _replay_main() + finally: + ACTIVE_EVIDENCE = previous + + if __name__ == '__main__': sys.exit(main()) diff --git a/tests/test_qs_failure_evidence.py b/tests/test_qs_failure_evidence.py new file mode 100644 index 000000000..54686ed50 --- /dev/null +++ b/tests/test_qs_failure_evidence.py @@ -0,0 +1,241 @@ +"""QS raw-child evidence controls. Synthetic children are not mathematical evidence.""" +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +WRAPPER = ROOT / 'frontiers/qs_d3_soft_layer_chain_20261005/replay.py' + + +class ChildEvidenceTests(unittest.TestCase): + def setUp(self): + spec = importlib.util.spec_from_file_location('qs_capture_subject', WRAPPER) + self.q = importlib.util.module_from_spec(spec) + spec.loader.exec_module(self.q) + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + + def recorder(self, name='evidence'): + self.assertTrue(hasattr(self.q, 'ChildEvidence'), + 'QS has no optional raw child-evidence recorder') + return self.q.ChildEvidence(self.root / name) + + def command(self, text): + return [sys.executable, *self.q.interpreter_flags(), '-c', text] + + def record(self, sink, index=1): + return json.loads((sink.directory / ('%04d' % index) / 'process.json').read_text()) + + def check_stream(self, sink, record, key, data, index=1): + row = record['streams'][key] + self.assertEqual(row['available'], data is not None) + if data is None: + self.assertIsNone(row['bytes']); self.assertIsNone(row['sha256']) + self.assertFalse(row['saved']) + self.assertFalse((sink.directory / ('%04d' % index) / (key+'.bin')).exists()) + else: + self.assertEqual(row['bytes'], len(data)) + self.assertEqual(row['sha256'], hashlib.sha256(data).hexdigest()) + self.assertTrue(row['saved']) + self.assertEqual((sink.directory / ('%04d' % index) / (key+'.bin')).read_bytes(), data) + + def test_exact_completed_binary_and_empty_streams(self): + sink = self.recorder() + command = self.command("import os; os.write(1,b'\\x00\\xffQS\\n'); os.write(2,b'')") + result = sink.run(command, cwd=str(self.root), timeout=5) + self.assertEqual(result, (0, b'\x00\xffQS\n', b'')) + record = self.record(sink) + self.assertEqual(record['command'], command) + self.assertEqual(record['cwd'], str(self.root)); self.assertEqual(record['timeout_seconds'], 5) + self.assertEqual(record['status'], 'completed'); self.assertEqual(record['returncode'], 0) + self.assertTrue(record['capture_complete']); self.assertIsNone(record['exception']) + self.check_stream(sink, record, 'stdout', result[1]); self.check_stream(sink, record, 'stderr', b'') + self.assertEqual(record['evidence_errors'], []) + + def test_crash_streams_are_preserved_and_not_accepted(self): + sink = self.recorder() + result = sink.run(self.command("import os; os.write(1,b'\\x00pre'); raise RuntimeError('synthetic unrelated crash')"), + cwd=str(self.root), timeout=5) + self.assertEqual(result[0], 1); self.assertIn(b'RuntimeError: synthetic unrelated crash', result[2]) + record = self.record(sink) + self.assertEqual(record['returncode'], 1) + self.check_stream(sink, record, 'stdout', result[1]); self.check_stream(sink, record, 'stderr', result[2]) + self.assertTrue(self.q.check_rejection(result, self.q.NEGATIVE_HARDGAP)) + + def test_wrong_response_is_preserved_without_changing_matcher(self): + sink = self.recorder() + result = sink.run(self.command("import sys; print('{bad-json}'); sys.exit(1)"), cwd=str(self.root), timeout=5) + self.check_stream(sink, self.record(sink), 'stdout', b'{bad-json}\n') + self.assertTrue(self.q.check_rejection(result, self.q.NEGATIVE_HARDGAP)) + + def test_actual_timeout_preserves_binary_prefix_and_exception(self): + sink = self.recorder() + command = self.command("import os,time; os.write(1,b'\\x00\\xffpartial'); os.write(2,b'\\xfeerr'); time.sleep(5)") + with self.assertRaises(subprocess.TimeoutExpired) as caught: + sink.run(command, cwd=str(self.root), timeout=0.6) + exc = caught.exception; record = self.record(sink) + self.assertIsNone(record['returncode']); self.assertEqual(record['status'], 'timeout') + self.assertFalse(record['capture_complete']); self.assertEqual(record['timeout_seconds'], 0.6) + self.assertEqual(exc.stdout, b'\x00\xffpartial'); self.assertEqual(exc.stderr, b'\xfeerr') + self.check_stream(sink, record, 'stdout', exc.stdout); self.check_stream(sink, record, 'stderr', exc.stderr) + self.assertEqual(exc.qs_evidence, record) + + def test_unavailable_is_not_captured_empty(self): + for index, (out, err) in enumerate(((None,None),(b'',None),(None,b''),(b'',b''))): + sink = self.recorder('case%d'%index) + original = subprocess.TimeoutExpired(['fake'], 7, output=out, stderr=err) + with mock.patch.object(self.q.subprocess, 'run', side_effect=original): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + sink.run(['fake'], cwd=str(self.root), timeout=7) + self.assertIs(caught.exception, original) + record = self.record(sink) + self.check_stream(sink, record, 'stdout', out); self.check_stream(sink, record, 'stderr', err) + + def test_real_spawn_error_has_no_invented_capture(self): + sink = self.recorder() + with self.assertRaises(FileNotFoundError) as caught: + sink.run([str(self.root/'not-an-executable')], cwd=str(self.root), timeout=5) + record = self.record(sink) + self.assertEqual(record['status'], 'execution_error'); self.assertIsNone(record['returncode']) + self.assertEqual(record['exception']['type'], 'FileNotFoundError') + self.check_stream(sink, record, 'stdout', None); self.check_stream(sink, record, 'stderr', None) + self.assertEqual(caught.exception.qs_evidence, record) + + def test_independent_writes_preserve_primary_timeout(self): + for mask in range(8): + with self.subTest(mask=mask): + sink = self.recorder('mask%d'%mask); real = sink._write + names = ['stdout.bin','stderr.bin','process.json']; attempted = [] + def write(path, data): + attempted.append(path.name) + if mask & (1 << names.index(path.name)): + raise OSError('synthetic write failure '+path.name) + return real(path, data) + original = subprocess.TimeoutExpired(['fake'], 5, output=b'out\x00', stderr=b'err\xff') + with mock.patch.object(sink, '_write', side_effect=write), \ + mock.patch.object(self.q.subprocess, 'run', side_effect=original): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + sink.run(['fake'], cwd=str(self.root), timeout=5) + self.assertIs(caught.exception, original); self.assertEqual(attempted, names) + record = original.qs_evidence + self.assertIsNone(record['returncode']) + self.assertEqual(len(record['evidence_errors']), mask.bit_count()) + for bit, name in enumerate(names): + self.assertEqual((sink.directory/'0001'/name).exists(), not bool(mask & (1 << bit))) + + def test_evidence_write_failure_never_promotes_completed_child(self): + for code in (0,1,2): + sink = self.recorder('exit%d'%code) + with mock.patch.object(sink, '_write', side_effect=OSError('synthetic disk failure')): + with self.assertRaises(self.q.ChildEvidenceError) as caught: + sink.run(self.command('import sys; print("data"); sys.exit(%d)'%code), cwd=str(self.root), timeout=5) + self.assertEqual(caught.exception.child_result, (code,b'data\n',b'')) + self.assertEqual(caught.exception.qs_evidence['returncode'], code) + self.assertEqual(len(caught.exception.qs_evidence['evidence_errors']), 3) + + def test_invalid_clock_is_qualified_without_masking_timeout(self): + cases = [(10.,9.9999999),(10.,float('nan')),(10.,float('inf')), + (True,11.),(OSError('clock-start'),11.),(10.,OSError('clock-end'))] + for i, clocks in enumerate(cases): + sink = self.recorder('clock%d'%i) + original = subprocess.TimeoutExpired(['fake'],5) + with mock.patch.object(self.q.time, 'monotonic', side_effect=clocks), \ + mock.patch.object(self.q.subprocess, 'run', side_effect=original): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + sink.run(['fake'],cwd=str(self.root),timeout=5) + self.assertIs(caught.exception,original);record=self.record(sink) + self.assertIsNone(record['elapsed_seconds']);self.assertTrue(record['clock_error']) + + def test_fresh_directory_and_no_overwrite(self): + sink = self.recorder() + with self.assertRaises((FileExistsError,ValueError)): + self.q.ChildEvidence(sink.directory) + (sink.directory/'0001').mkdir() + (sink.directory/'0001'/'sentinel').write_bytes(b'keep') + with mock.patch.object(self.q.subprocess,'run',side_effect=AssertionError('must not run')): + with self.assertRaises(FileExistsError): + sink.run(['fake'],cwd=str(self.root),timeout=5) + self.assertEqual((sink.directory/'0001'/'sentinel').read_bytes(),b'keep') + + def test_evidence_inside_checkout_or_symlink_alias_is_refused(self): + self.assertTrue(hasattr(self.q,'ChildEvidence'),'missing external evidence directory guard') + checkout=self.root/'checkout';(checkout/'frontiers'/'packet').mkdir(parents=True) + with mock.patch.object(self.q,'HERE',checkout/'frontiers'/'packet'): + for directory in (checkout/'evidence',checkout/'frontiers'/'packet'/'evidence'): + with self.assertRaises(ValueError): self.q.ChildEvidence(directory) + (self.root/'alias').symlink_to(checkout,target_is_directory=True) + with self.assertRaises(ValueError): self.q.ChildEvidence(self.root/'alias'/'evidence') + self.assertFalse((checkout/'evidence').exists()) + + def test_separate_records_for_identical_invocations(self): + sink=self.recorder(); command=self.command('print("same")') + for _ in range(2): sink.run(command,cwd=str(self.root),timeout=5) + self.assertEqual(sorted(p.name for p in sink.directory.iterdir()),['0001','0002']) + self.assertEqual(self.record(sink,1)['sequence'],1);self.assertEqual(self.record(sink,2)['sequence'],2) + + def test_run_without_capture_retains_original_contract(self): + script=self.root/'checker.py';script.write_text('print("unchanged")\n') + result=self.q.run(script) + self.assertEqual(result,(0,b'unchanged\n',b'')) + self.assertEqual(sorted(p.name for p in self.root.iterdir()),['checker.py']) + + def test_run_with_capture_keeps_original_flags_cwd_timeout(self): + sink=self.recorder();script=self.root/'checker.py';script.write_text('print("same")\n') + with mock.patch.object(self.q,'ACTIVE_EVIDENCE',sink): result=self.q.run(script) + record=self.record(sink) + self.assertEqual(result,(0,b'same\n',b''));self.assertEqual(record['timeout_seconds'],1800) + self.assertEqual(record['command'],[sys.executable,*self.q.interpreter_flags(),str(script)]) + self.assertEqual(record['cwd'],str(script.parent)) + + def test_main_restores_recorder_and_does_not_continue_after_exception(self): + self.assertTrue(hasattr(self.q,'_replay_main'),'missing scoped evidence opt-in') + destination=self.root/'main-capture'; original=RuntimeError('primary synthetic failure') + def failing(): + self.assertIsNotNone(self.q.ACTIVE_EVIDENCE) + raise original + with mock.patch.dict(os.environ,{'QS_REPLAY_EVIDENCE_DIR':str(destination)}), \ + mock.patch.object(self.q,'_replay_main',side_effect=failing): + with self.assertRaises(RuntimeError) as caught: self.q.main() + self.assertIs(caught.exception,original);self.assertIsNone(self.q.ACTIVE_EVIDENCE) + self.assertTrue(destination.is_dir()) + + def test_intended_rejection_keeps_exact_diagnostic(self): + sink=self.recorder() + command=self.command("import sys; sys.stderr.write('FAILED: X9_hardgap\\n'); sys.exit(1)") + result=sink.run(command,cwd=str(self.root),timeout=5) + self.assertEqual(self.q.check_rejection(result,self.q.NEGATIVE_HARDGAP),[]) + self.check_stream(sink,self.record(sink),'stderr',b'FAILED: X9_hardgap\n') + + def test_real_orchestration_stops_after_first_timeout(self): + self.recorder('capability-probe') + original=subprocess.TimeoutExpired(['synthetic timeout'],1800,output=b'partial') + with mock.patch.dict(os.environ,{'QS_REPLAY_EVIDENCE_DIR':str(self.root/'orchestration')}), \ + mock.patch.object(self.q.subprocess,'run',side_effect=original) as child, \ + mock.patch.object(self.q,'negative_control_failures',side_effect=AssertionError('must not continue')): + with self.assertRaises(subprocess.TimeoutExpired) as caught: self.q.main() + self.assertIs(caught.exception,original);self.assertEqual(child.call_count,1) + self.assertIsNone(self.q.ACTIVE_EVIDENCE) + record=json.loads((self.root/'orchestration'/'0001'/'process.json').read_text()) + self.assertEqual(record['status'],'timeout');self.assertEqual(record['timeout_seconds'],1800) + + def test_workflow_enables_capture_only_for_real_replay_and_preserves_it(self): + text=(ROOT/'.github/workflows/qs-d3-soft-layer-chain.yml').read_text() + self.assertIn("'tests/test_qs_failure_evidence.py'",text) + self.assertIn('-p test_qs_failure_evidence.py -v',text) + self.assertIn('QS_REPLAY_EVIDENCE_DIR="$out/children" python ${{ matrix.mode }}',text) + self.assertIn('path: ${{ runner.temp }}/qs-replay-evidence/',text) + self.assertIn('if-no-files-found: error',text) + self.assertEqual(text.count('if: always()'),2) + self.assertIn('set -euo pipefail',text) + + +if __name__ == '__main__': + unittest.main()