diff --git a/.github/workflows/qs-d3-soft-layer-chain.yml b/.github/workflows/qs-d3-soft-layer-chain.yml index 2fcf36c84..ba878de64 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_d3_replay_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_d3_replay_evidence.py -v 2>&1 | tee "$out/logs/process-evidence-tests.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 @@ -48,6 +50,7 @@ jobs: 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_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..3415538b0 100644 --- a/frontiers/qs_d3_soft_layer_chain_20261005/replay.py +++ b/frontiers/qs_d3_soft_layer_chain_20261005/replay.py @@ -23,9 +23,13 @@ algebra, as the reads say. Nothing here reads or grades a proof. Usage: python3 -B -S frontiers/qs_d3_soft_layer_chain_20261005/replay.py (exit 0 iff every check passes) +Optional QS_REPLAY_EVIDENCE_DIR (outside the checkout) retains raw child streams and process records. +Each invocation gets a new directory; records describe execution, never mathematical acceptance. +Required evidence I/O fails the run; a timeout/spawn error remains the primary exception. """ import hashlib import json +import os import pathlib import re import shutil @@ -279,9 +283,70 @@ def extraction_failures(root, manifest): return failures +PROCESS_TIMEOUT_SECONDS = 1800 # Unchanged production budget; tests shorten it for real timeout children. + + +def evidence_directory(script): + """An explicit destination opts in; never put generated evidence in source directories.""" + raw = os.environ.get('QS_REPLAY_EVIDENCE_DIR') + if raw is None: + return None + if not raw: + raise ValueError('QS_REPLAY_EVIDENCE_DIR must not be empty') + root = pathlib.Path(raw).expanduser().resolve() + if root.is_relative_to(HERE.resolve().parents[1]) or root.is_relative_to(script.parent.resolve()): + raise ValueError('QS replay evidence must be outside the checkout and child source directory') + root.mkdir(parents=True, exist_ok=True) + return pathlib.Path(tempfile.mkdtemp(prefix='invocation-', dir=root)) + + +def save_process_evidence(folder, record, out, err): + """Attempt every write independently. None is unavailable, not an observed empty stream.""" + record['streams'] = {} + record['evidence_errors'] = [] + for name, data in (('stdout', out), ('stderr', err)): + entry = dict(available=data is not None, bytes=len(data) if data is not None else None, + sha256=hashlib.sha256(data).hexdigest() if data is not None else None, saved=False) + record['streams'][name] = entry + try: + (folder / (name + '.bin')).write_bytes(data if data is not None else b'') + entry['saved'] = True + except Exception as exc: + record['evidence_errors'].append('%s.bin: %s: %s' % (name, type(exc).__name__, exc)) + try: + data = (json.dumps(record, sort_keys=True, indent=2, allow_nan=False) + '\n').encode('utf-8') + (folder / 'process.json').write_bytes(data) + except Exception as exc: + record['evidence_errors'].append('process.json: %s: %s' % (type(exc).__name__, exc)) + + def run(script, *args): - proc = subprocess.run([sys.executable, *interpreter_flags(), str(script), *args], capture_output=True, - timeout=1800, cwd=str(script.parent)) + command = [sys.executable, *interpreter_flags(), str(script), *args] + folder = evidence_directory(script) # Invalid/unwritable setup stops before starting a child. + record = dict(schema=1, command=command, cwd=str(script.parent), + timeout_seconds=PROCESS_TIMEOUT_SECONDS, scientific_effect='NONE') + try: + proc = subprocess.run(command, capture_output=True, timeout=PROCESS_TIMEOUT_SECONDS, + cwd=str(script.parent)) + except (subprocess.TimeoutExpired, OSError) as exc: + if folder is not None: + record.update(status='timeout' if isinstance(exc, subprocess.TimeoutExpired) else 'spawn_error', + returncode=None, exception=dict(type=type(exc).__name__, message=str(exc))) + exc.evidence_record = record + try: + save_process_evidence(folder, record, getattr(exc, 'stdout', None), getattr(exc, 'stderr', None)) + if record['evidence_errors']: + exc.add_note('QS replay evidence errors: ' + '; '.join(record['evidence_errors'])) + except Exception as secondary: + exc.add_note('QS replay evidence recording failed: %s: %s' % (type(secondary).__name__, secondary)) + raise # Never replace the primary timeout or launch error with a recording error. + if folder is not None: + record.update(status='completed', returncode=proc.returncode, exception=None) + save_process_evidence(folder, record, proc.stdout, proc.stderr) + if record['evidence_errors']: + exc = RuntimeError('QS replay evidence errors: ' + '; '.join(record['evidence_errors'])) + exc.evidence_record = record + raise exc # Even exit 0 is not success when explicitly required evidence is missing. return proc.returncode, proc.stdout, proc.stderr diff --git a/tests/test_qs_d3_replay_evidence.py b/tests/test_qs_d3_replay_evidence.py new file mode 100644 index 000000000..87aa624b3 --- /dev/null +++ b/tests/test_qs_d3_replay_evidence.py @@ -0,0 +1,215 @@ +"""Optional raw-process evidence; 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.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +PACKET = ROOT / 'frontiers/qs_d3_soft_layer_chain_20261005' +ENV = 'QS_REPLAY_EVIDENCE_DIR' + + +class RawEvidenceTests(unittest.TestCase): + def setUp(self): + spec = importlib.util.spec_from_file_location('qs_raw_subject', PACKET / 'replay.py') + self.m = importlib.util.module_from_spec(spec) + spec.loader.exec_module(self.m) + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + scripts = self.root / 'scripts' + scripts.mkdir() + self.script = scripts / 'fixture.py' + self.evidence = self.root / 'evidence' + self.env = patch.dict(os.environ, {ENV: str(self.evidence)}) + self.env.start() + self.addCleanup(self.env.stop) + + def child(self, text): + self.script.write_text(text, encoding='utf-8') + return self.script + + def record(self): + records = list(self.evidence.glob('invocation-*/process.json')) + self.assertEqual(len(records), 1, 'one independently retained child process record required') + return records[0].parent, json.loads(records[0].read_bytes()) + + def verify_stream(self, folder, rec, name, value): + stream = rec['streams'][name] + self.assertEqual(stream['available'], value is not None) + self.assertEqual(stream['bytes'], len(value) if value is not None else None) + self.assertEqual(stream['sha256'], hashlib.sha256(value).hexdigest() if value is not None else None) + self.assertTrue(stream['saved']) + self.assertEqual((folder / (name + '.bin')).read_bytes(), value if value is not None else b'') + + def test_no_capture_preserves_original_three_value_result(self): + os.environ.pop(ENV) + s = self.child("import os,sys\nos.write(1,b'\\xffok');os.write(2,b'\\x00err');sys.exit(7)\n") + self.assertEqual(self.m.run(s), (7, b'\xffok', b'\x00err')) + self.assertFalse(self.evidence.exists()) + + def test_real_crash_retains_full_binary_streams_and_invocation(self): + s = self.child("import os\nos.write(1,b'\\x00\\xffpartial')\nraise RuntimeError('synthetic crash')\n") + result = self.m.run(s, '--diagnostic') + self.assertEqual(result[0], 1) + self.assertIn(b'RuntimeError: synthetic crash', result[2]) + folder, rec = self.record() + self.assertEqual(rec['command'], [sys.executable, *self.m.interpreter_flags(), str(s), '--diagnostic']) + self.assertEqual(rec['cwd'], str(s.parent)) + self.assertEqual(rec['timeout_seconds'], 1800) + self.assertEqual(rec['status'], 'completed') + self.assertEqual(rec['returncode'], 1) + self.assertNotIn('passed', rec) + self.assertEqual(rec['evidence_errors'], []) + self.verify_stream(folder, rec, 'stdout', result[1]) + self.verify_stream(folder, rec, 'stderr', result[2]) + + def test_success_and_intended_failure_outputs_are_not_reinterpreted(self): + s = self.child("import os,sys\nos.write(1,b'answer\\n');os.write(2,b'diagnostic\\n');sys.exit(int(sys.argv[1]))\n") + for code in (0, 1, 2): + with self.subTest(code=code): + result = self.m.run(s, str(code)) + self.assertEqual(result, (code, b'answer\n', b'diagnostic\n')) + records = [json.loads(p.read_bytes()) for p in self.evidence.glob('invocation-*/process.json')] + self.assertEqual(sorted(r['returncode'] for r in records), [0, 1, 2]) + self.assertTrue(all('passed' not in r for r in records)) + + def test_repeated_invocations_never_overwrite(self): + s = self.child("print('first')\n") + self.m.run(s) + before = {p: p.read_bytes() for p in self.evidence.glob('invocation-*/*')} + self.assertEqual(len(before), 3) + self.m.run(s) + self.assertEqual(len(list(self.evidence.glob('invocation-*'))), 2) + for p, data in before.items(): + self.assertEqual(p.read_bytes(), data) + + def test_real_timeout_keeps_partial_bytes_and_timeout_exception(self): + self.assertTrue(hasattr(self.m, 'PROCESS_TIMEOUT_SECONDS'), 'explicit unchanged production timeout required') + s = self.child("import os,time\nos.write(1,b'\\xffout');os.write(2,b'\\x00err');time.sleep(10)\n") + with patch.object(self.m, 'PROCESS_TIMEOUT_SECONDS', 0.5): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + self.m.run(s) + folder, rec = self.record() + self.assertEqual(rec['status'], 'timeout') + self.assertIsNone(rec['returncode']) + self.assertEqual(rec['timeout_seconds'], 0.5) + self.assertEqual(caught.exception.output, b'\xffout') + self.assertEqual(caught.exception.stderr, b'\x00err') + self.verify_stream(folder, rec, 'stdout', caught.exception.output) + self.verify_stream(folder, rec, 'stderr', caught.exception.stderr) + self.assertEqual(caught.exception.evidence_record, rec) + + def test_unavailable_empty_and_binary_timeout_capture_are_distinct(self): + for out in (None, b'', b'\xff\x00'): + for err in (None, b'', b'\x80err'): + with self.subTest(out=out, err=err), tempfile.TemporaryDirectory() as td: + target = Path(td) / 'records' + exc = subprocess.TimeoutExpired(['synthetic'], 1800, output=out, stderr=err) + with patch.dict(os.environ, {ENV: str(target)}), patch.object(self.m.subprocess, 'run', side_effect=exc): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + self.m.run(self.script) + self.assertIs(caught.exception, exc) + paths = list(target.glob('invocation-*/process.json')) + self.assertEqual(len(paths), 1) + rec = json.loads(paths[0].read_bytes()) + self.verify_stream(paths[0].parent, rec, 'stdout', out) + self.verify_stream(paths[0].parent, rec, 'stderr', err) + self.assertIsNone(rec['returncode']) + + def test_real_spawn_error_retains_unknown_output_and_original_error(self): + s = self.child("print('must not execute')\n") + missing = str(self.root / 'missing-interpreter') + with patch.object(self.m.sys, 'executable', missing): + with self.assertRaises(FileNotFoundError) as caught: + self.m.run(s) + folder, rec = self.record() + self.assertEqual(rec['status'], 'spawn_error') + self.assertIsNone(rec['returncode']) + self.assertEqual(rec['exception']['type'], 'FileNotFoundError') + self.verify_stream(folder, rec, 'stdout', None) + self.verify_stream(folder, rec, 'stderr', None) + self.assertEqual(caught.exception.evidence_record, rec) + + def test_each_write_failure_still_attempts_other_evidence(self): + original = Path.write_bytes + for failed in ('stdout.bin', 'stderr.bin', 'process.json'): + with self.subTest(failed=failed), tempfile.TemporaryDirectory() as td: + target = Path(td) / 'records' + attempted = [] + def write(path, data): + attempted.append(path.name) + if path.name == failed: + raise OSError('synthetic write refusal') + return original(path, data) + exc = subprocess.TimeoutExpired(['synthetic'], 1800, output=b'out', stderr=b'err') + with patch.dict(os.environ, {ENV: str(target)}), patch.object(self.m.subprocess, 'run', side_effect=exc), patch.object(Path, 'write_bytes', write): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + self.m.run(self.script) + self.assertIs(caught.exception, exc) + self.assertEqual(attempted, ['stdout.bin', 'stderr.bin', 'process.json']) + rec = caught.exception.evidence_record + self.assertEqual(len(rec['evidence_errors']), 1) + self.assertIn(failed, rec['evidence_errors'][0]) + self.assertTrue(any('evidence' in note for note in caught.exception.__notes__)) + folder = next(target.glob('invocation-*')) + for name in {'stdout.bin', 'stderr.bin', 'process.json'} - {failed}: + self.assertTrue((folder / name).is_file()) + + def test_all_write_failures_do_not_replace_primary_timeout(self): + exc = subprocess.TimeoutExpired(['synthetic'], 1800, output=b'out', stderr=b'err') + with patch.object(self.m.subprocess, 'run', side_effect=exc), patch.object(Path, 'write_bytes', side_effect=OSError('synthetic full disk')): + with self.assertRaises(subprocess.TimeoutExpired) as caught: + self.m.run(self.script) + self.assertIs(caught.exception, exc) + self.assertTrue(hasattr(exc, 'evidence_record'), 'retain failure metadata in memory') + self.assertEqual(len(exc.evidence_record['evidence_errors']), 3) + self.assertIsNone(exc.evidence_record['returncode']) + + def test_completed_child_with_failed_evidence_cannot_report_success(self): + self.child("print('actual completed child')\n") + with patch.object(Path, 'write_bytes', side_effect=OSError('synthetic full disk')): + with self.assertRaises(RuntimeError) as caught: + self.m.run(self.script) + self.assertEqual(caught.exception.evidence_record['returncode'], 0) + self.assertEqual(len(caught.exception.evidence_record['evidence_errors']), 3) + + def test_invalid_destination_stops_before_child(self): + for target in (PACKET / 'forbidden-evidence', self.script.parent / 'forbidden-evidence'): + with self.subTest(target=target), patch.dict(os.environ, {ENV: str(target)}), patch.object(self.m.subprocess, 'run') as child: + with self.assertRaises(ValueError): + self.m.run(self.script) + child.assert_not_called() + self.assertFalse(target.exists()) + self.evidence.write_text('not a directory') + with patch.object(self.m.subprocess, 'run') as child: + with self.assertRaises(OSError): + self.m.run(self.script) + child.assert_not_called() + + def test_failed_custody_creates_no_child_evidence(self): + with patch.object(self.m, 'tree_failures', return_value=['synthetic custody refusal']), patch.object(self.m.subprocess, 'run') as child: + self.assertEqual(self.m.main(), 1) + child.assert_not_called() + self.assertFalse(self.evidence.exists()) + + def test_workflow_wires_tests_and_capture_outside_source_tree(self): + text = (ROOT / '.github/workflows/qs-d3-soft-layer-chain.yml').read_text() + self.assertIn("'tests/test_qs_d3_replay_evidence.py'", text) + self.assertIn('-p test_qs_d3_replay_evidence.py', text) + self.assertIn('QS_REPLAY_EVIDENCE_DIR="$out/children" python', text) + self.assertIn('cp tests/test_qs_d3_replay_evidence.py "$out/"', text) + self.assertIn('set -euo pipefail', text) + self.assertIn('git diff --exit-code', text) + self.assertIn('contents: read', text) + self.assertIn("python-version: '3.11.16'", text) + + +if __name__ == '__main__': + unittest.main()