From 680bec8cacda901d303f39f5c1254eb17b5cee50 Mon Sep 17 00:00:00 2001 From: Dylan Roy <249851642+d6g8k5htny-coder@users.noreply.github.com> Date: Mon, 5 Oct 2026 20:20:25 -0700 Subject: [PATCH] Add optional raw QS child execution evidence without changing rejection rules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dylan Roy — delegated AI engineering. OpenAI / GPT-6 Astra Pro, qs-failure-evidence-20261006-r9; pickup307/6008405929. Three files only: wrapper, new external capture tests, workflow wiring. 31 local focused methods pass in each Python mode. Final18 new methods produce24 expected failing subcases against the exact predecessor. Preserve all91 stored payloads, original13 protocol methods,62 rejection contracts, default run branch and core verification stages exactly. Record binary streams, completed/timeout/execution-error metadata and independent write outcomes outside checkout. Never turn exception into success. Scientific effect NONE. Full genuine and hosted evidence remain separately recorded; no author merge, parent hold or independent review claimed. --- .github/workflows/qs-d3-soft-layer-chain.yml | 6 +- .../qs_d3_soft_layer_chain_20261005/replay.py | 133 +++++++++- tests/test_qs_failure_evidence.py | 241 ++++++++++++++++++ 3 files changed, 376 insertions(+), 4 deletions(-) create mode 100644 tests/test_qs_failure_evidence.py 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()