#!/usr/bin/env python3 """Read-only replay of C932--C1131, or a completed prefix via --through N. The journal writer is never imported. Standard output is the JSON report; --output may additionally save that report. Replay itself cannot write files. The supplied packet is trusted research code, not an arbitrary-code sandbox. """ from pathlib import Path from fractions import Fraction from datetime import datetime import argparse import ast import contextlib import copy import hashlib import io import json import os import re import runpy import sys import types ROOT = Path(__file__).resolve().parent.parent FIRST, LAST = 932, 1131 STEP_COUNT = LAST - FIRST + 1 PREDECESSOR_HASH = '55a6faf783579d9119c15adafb3697ab140daddaeda4547bcd674196a4951a9f' PRIMARY_HASH = 'a5ea84562101158b60d0cf296765d6eff38e7a2abda4e74ad1b353dfd13b9530' REPLAY_ACTIVE = False class VerificationError(RuntimeError): pass def require(condition, message): if not condition: raise VerificationError(message) def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def record_digest(row): return hashlib.sha256(json.dumps({k: v for k, v in row.items() if k != 'sha256'}, sort_keys=True).encode()).hexdigest() def clean(value): if isinstance(value, Fraction): return value.numerator if value.denominator == 1 else str(value) if isinstance(value, dict): return {str(k): clean(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [clean(v) for v in value] return value class InverseValue(int): """Track values descended from inv(), including ordinary integer arithmetic.""" @staticmethod def wrap(value): return InverseValue(value) if isinstance(value, int) else value def __add__(self, other): return self.wrap(int.__add__(self, other)) def __radd__(self, other): return self.wrap(int.__add__(self, other)) def __sub__(self, other): return self.wrap(int.__sub__(self, other)) def __rsub__(self, other): return self.wrap(int.__rsub__(self, other)) def __mul__(self, other): return self.wrap(int.__mul__(self, other)) def __rmul__(self, other): return self.wrap(int.__mul__(self, other)) def __floordiv__(self, other): return self.wrap(int.__floordiv__(self, other)) def __rfloordiv__(self, other): return self.wrap(int.__rfloordiv__(self, other)) def __mod__(self, other): return self.wrap(int.__mod__(self, other)) def __rmod__(self, other): return self.wrap(int.__rmod__(self, other)) def __pow__(self, other, modulo=None): return self.wrap(int.__pow__(self, other, modulo)) def __neg__(self): return InverseValue(int.__neg__(self)) def __pos__(self): return InverseValue(self) def __abs__(self): return InverseValue(int.__abs__(self)) def audit_read_only(event, args): if not REPLAY_ACTIVE: return if event == 'open': mode = args[1] if len(args) > 1 else None flags = args[2] if len(args) > 2 else 0 write_flags = os.O_WRONLY | os.O_RDWR | os.O_APPEND | os.O_CREAT | os.O_TRUNC require(not (isinstance(mode, str) and any(c in mode for c in 'wax+')), 'Replay attempted a write-capable open') require(not (isinstance(flags, int) and flags & write_flags), 'Replay attempted a write-capable file descriptor') forbidden = ('subprocess.', 'socket.', 'ctypes.', 'os.exec', 'os.spawn') require(not event.startswith(forbidden), f'Replay attempted forbidden action: {event}') require(event not in {'os.remove', 'os.rmdir', 'os.rename', 'os.mkdir', 'os.link', 'os.symlink', 'os.truncate', 'os.chmod', 'os.chown', 'os.utime', 'os.system', 'os.fork', 'os.chdir'}, f'Replay attempted mutation: {event}') def current_candidates(name, workspace, bindings): if name in bindings: paths = bindings[name] paths = [paths] if isinstance(paths, str) else paths require(isinstance(paths, list) and all(isinstance(p, str) for p in paths), f'Invalid current-source binding: {name}') return [workspace / p for p in paths] if len(name) > 3 and name[:2].isdigit() and name[2] == '-': return [workspace / 'project_sources' / name] special = { 'File51a.md': ['project_sources/39-File_51a.Rounded_Scaffold_Mod5_Architecture.md'], 'File52c_latest.md': ['upload/File_52c.Rounded_Whole_Span_Inverse_Detailed_Study_Draft (2)(1).md'], 'Strategy.md': ['upload/490d_Unification_Research_Strategy_v0_2_20260906.md', 'c482_c531/evidence/sources/Strategy.md'], 'C931_checkpoint.md': ['c832_c931/deliverables/490d_Chronological_Families_Checkpoint_After_C931.md'], 'C931_synthesis.md': ['c832_c931/deliverables/490d_Chronological_Families_Integrated_Synthesis_C931.md'], 'C931_reading_guide.md': ['c832_c931/deliverables/490d_Chronological_Families_Reading_Guide_C931.md'], 'C931_technical.md': ['c832_c931/deliverables/490d_Chronological_Families_Technical_Companion_C931.md'], } return [workspace / p for p in special.get(name, [])] def verify_sources(workspace, require_current): manifest = json.loads((ROOT / 'evidence/SOURCE_MANIFEST.json').read_text()) require(isinstance(manifest, dict) and manifest, 'Missing source manifest') require(manifest.get('File52c_latest.md') == PRIMARY_HASH, 'Latest File52c identity mismatch') binding_path = ROOT / 'evidence/SOURCE_BINDINGS.json' bindings = json.loads(binding_path.read_text()) if binding_path.is_file() else {} require(isinstance(bindings, dict), 'Invalid optional SOURCE_BINDINGS.json') current, absent = [], [] for name, expected in manifest.items(): require(Path(name).name == name, f'Unsafe source snapshot path: {name}') snapshot = ROOT / 'evidence/sources' / name require(digest(snapshot) == expected, f'Source snapshot hash mismatch: {name}') available = [p for p in current_candidates(name, workspace, bindings) if p.is_file()] if not available: absent.append(name) for path in available: require(digest(path) == expected, f'Current source differs from frozen snapshot: {path}') current.append({'snapshot': name, 'current_path': str(path), 'sha256': expected}) require(not require_current or not absent, f'Current originals unavailable: {absent}') predecessor = json.loads((ROOT / 'evidence/C931_PREDECESSOR.json').read_text()) require(predecessor['step'] == 931, 'Wrong predecessor step') require(record_digest(predecessor) == predecessor['sha256'] == PREDECESSOR_HASH, 'C931 predecessor hash mismatch') # C931 closes with a list of fully bound final artifacts. Match source # aliases to those artifact identities rather than assuming an older schema. final_artifacts = predecessor['results']['final_artifacts'] require(isinstance(final_artifacts, list) and final_artifacts, 'Missing C931 final artifact bindings') deliveries = {Path(item['path']).name: item for item in final_artifacts} require(len(deliveries) == len(final_artifacts), 'Ambiguous C931 final artifact basenames') required_bindings = { 'C931_checkpoint.md': '490d_Chronological_Families_Checkpoint_After_C931.md', 'C931_synthesis.md': '490d_Chronological_Families_Integrated_Synthesis_C931.md', 'C931_technical.md': '490d_Chronological_Families_Technical_Companion_C931.md', } optional_bindings = { 'C931_reading_guide.md': '490d_Chronological_Families_Reading_Guide_C931.md', } for alias, filename in {**required_bindings, **{k: v for k, v in optional_bindings.items() if k in manifest}}.items(): require(alias in manifest, f'Missing required predecessor source snapshot: {alias}') item = deliveries[filename] require(item['sha256'] == manifest[alias], f'C931 artifact binding mismatch: {alias}') require((ROOT / 'evidence/sources' / alias).stat().st_size == item['bytes'], f'C931 artifact byte count mismatch: {alias}') old_journal = workspace / 'c832_c931/journal.json' predecessor_live = False if old_journal.is_file(): old_rows = json.loads(old_journal.read_text()) prior = [r for r in old_rows if r.get('step') == 931] require(prior == [predecessor], 'Current C931 record differs from predecessor snapshot') predecessor_live = True return manifest, current, absent, predecessor_live def verify_versioned_inputs(rows): """Validate explicit draft versions; journal metadata remains authoritative.""" manifest_path = ROOT / 'evidence/VERSIONED_INPUTS.json' if not manifest_path.is_file(): return {}, [] manifest = json.loads(manifest_path.read_text()) require(isinstance(manifest, dict) and set(manifest) == {'schema_version', 'bindings'}, 'Invalid versioned-input manifest schema') require(manifest['schema_version'] == 1 and isinstance(manifest['bindings'], list) and manifest['bindings'], 'Invalid versioned-input manifest version/bindings') by_step, verified = {}, [] records = {row['step']: row for row in rows} def safe_path(value, prefix): require(isinstance(value, str) and value and '\\' not in value, 'Invalid versioned-input path') path = Path(value) require(not path.is_absolute() and '..' not in path.parts and path.as_posix() == value, f'Unsafe versioned-input path: {value}') require(path.parts[:len(prefix)] == prefix, f'Invalid versioned-input path scope: {value}') resolved = (ROOT / path).resolve() require(ROOT in resolved.parents and resolved.is_file(), f'Missing or escaping versioned input: {value}') return ROOT / path for item in manifest['bindings']: require(isinstance(item, dict) and set(item) == {'logical_path', 'snapshot_path', 'sha256', 'steps', 'journal_input_field', 'mode'}, 'Invalid versioned-input binding schema') logical = safe_path(item['logical_path'], ('prep',)) snapshot = safe_path(item['snapshot_path'], ('evidence', 'input_versions')) require(isinstance(item['sha256'], str) and re.fullmatch(r'[0-9a-f]{64}', item['sha256']), 'Invalid versioned-input SHA-256') require(item['journal_input_field'] == 'draft_sha256', 'Versioned input must bind the recorded draft_sha256 field') require(item['mode'] in {'historical', 'current'}, 'Invalid versioned-input mode') steps = item['steps'] require(isinstance(steps, list) and steps and all(type(n) is int and FIRST <= n <= LAST for n in steps), 'Invalid versioned-input steps') require(steps == sorted(set(steps)), 'Duplicate or unordered versioned-input steps') require(digest(snapshot) == item['sha256'], f'Versioned-input snapshot hash mismatch: {item["snapshot_path"]}') if item['mode'] == 'current': require(digest(logical) == item['sha256'], f'Current versioned-input hash mismatch: {item["logical_path"]}') for number in steps: # Prefix packets can carry bindings for later, unselected actions. # A binding is activated only against a verified selected record. if number not in records: continue row = records[number] require(row['inputs'].get(item['journal_input_field']) == item['sha256'], f'C{number}: versioned-input binding differs from journal draft hash') require(item['logical_path'] in row['sources'], f'C{number}: versioned input is not a declared source') step_bindings = by_step.setdefault(number, {}) require(item['logical_path'] not in step_bindings, f'C{number}: duplicate versioned-input binding') step_bindings[item['logical_path']] = item.copy() verified.append({'step': number, **item}) return by_step, verified def replay(row, versioned_inputs=None): global REPLAY_ACTIVE number = row['step'] script = ROOT / 'evidence' / f's{number}.py' tree = ast.parse(script.read_text(), filename=str(script)) # An additional static check complements runtime inverse-value tracking. for node in ast.walk(tree): if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == 'inv': require(not any(isinstance(inner, ast.Call) and isinstance(inner.func, ast.Name) and inner.func.id == 'inv' for arg in node.args for inner in ast.walk(arg)), f'C{number}: nested inverse execution') if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): require(node.name not in {'inv', 'inverse'}, f'C{number}: replay inverse guard was redefined') calls = {'begin': 0, 'finish': 0, 'inverse': 0} versioned_inputs = versioned_inputs or {} historical_reads = {name: 0 for name, item in versioned_inputs.items() if item['mode'] == 'historical'} historical_paths = {str(ROOT / name): item for name, item in versioned_inputs.items() if item['mode'] == 'historical'} class ReplayPath(type(ROOT)): """Redirect only declared ROOT-derived read_text/read_bytes calls.""" def historical_bytes(self): item = historical_paths.get(str(self)) if item is None: return None data = (ROOT / item['snapshot_path']).read_bytes() require(hashlib.sha256(data).hexdigest() == item['sha256'], f'C{number}: historical input changed during replay') historical_reads[item['logical_path']] += 1 return data def read_bytes(self): data = self.historical_bytes() return super().read_bytes() if data is None else data def read_text(self, encoding=None, errors=None): data = self.historical_bytes() if data is None: return super().read_text(encoding=encoding, errors=errors) with io.TextIOWrapper(io.BytesIO(data), encoding=encoding, errors=errors) as handle: return handle.read() def inverse(n): require(isinstance(n, int) and not isinstance(n, bool) and n >= 0, f'C{number}: inverse input must be a nonnegative integer') require(not isinstance(n, InverseValue), f'C{number}: forbidden inverse of an inverse-derived value') calls['inverse'] += 1 if n == 0: return InverseValue(0) # Independent implementation: string-based core extraction, not the production loop. text = str(n) core = text.rstrip('0') return InverseValue(int(core[::-1]) * 10 ** (len(text) - len(core))) def begin(n, title, question, inputs, sources): calls['begin'] += 1 require(calls['begin'] == 1 and calls['finish'] == 0, f'C{number}: invalid begin sequence') supplied = (n, title, question, clean(inputs), sources) recorded = tuple(row[k] for k in ('step', 'title', 'question', 'inputs', 'sources')) require(supplied == recorded, f'C{number}: input/source metadata mismatch') return copy.deepcopy({k: row[k] for k in ('step', 'title', 'question', 'inputs', 'sources', 'opened_utc', 'predecessor_sha256')}) def finish(item, results, finding, next_question, checks): calls['finish'] += 1 require(calls['begin'] == calls['finish'] == 1, f'C{number}: invalid finish sequence') for key in ('step', 'title', 'question', 'inputs', 'sources', 'opened_utc', 'predecessor_sha256'): require(item.get(key) == row[key], f'C{number}: begin record mutated: {key}') require(isinstance(checks, dict) and checks and all(v is True for v in checks.values()), f'C{number}: checks are not all Boolean True') require((clean(results), finding, next_question, checks) == (row['results'], row['finding'], row['reassessment'], row['checks']), f'C{number}: result/finding/reassessment/check mismatch') def artifact(relative_path, content): # Recompute each generation event even when a later step revised the file. # The historical generated bytes are bound by the result record's digest. path = Path(relative_path) require(not path.is_absolute() and '..' not in path.parts, f'C{number}: unsafe artifact path') data = content.encode() return {'path': relative_path, 'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data)} fake = types.ModuleType('research') exported = {'Path': Path, 'F': Fraction, 'ROOT': ReplayPath(ROOT), 'E': Fraction(25, 23), 'P': Fraction(70, 69), 'J': Fraction(300, 299), 'json': json, 'hashlib': hashlib, 'clean': clean, 'inv': inverse, 'begin': begin, 'finish': finish, 'REPLAY': True, 'artifact': artifact} for key, value in exported.items(): setattr(fake, key, value) previous = sys.modules.get('research') sys.modules['research'] = fake capture = io.StringIO() try: REPLAY_ACTIVE = True with contextlib.redirect_stdout(capture): runpy.run_path(str(script), run_name=f'replay_{number}') finally: REPLAY_ACTIVE = False if previous is None: sys.modules.pop('research', None) else: sys.modules['research'] = previous require(calls['begin'] == calls['finish'] == 1, f'C{number}: missing begin or finish') require(all(count > 0 for count in historical_reads.values()), f'C{number}: declared historical input was not read') return {'step': number, 'script_sha256': digest(script), 'checks': len(row['checks']), 'single_pass_inverse_calls': calls['inverse'], 'historical_input_reads': historical_reads} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--through', type=int, default=LAST, help='Completed last step to replay;1130 is the199-action preseal prefix; default1131 requires all200') parser.add_argument('--workspace', type=Path, default=ROOT.parent, help='Optional workspace holding current originals; default packet parent') parser.add_argument('--require-current-sources', action='store_true', help='Fail if any current original is absent; omit for extracted portable packets') parser.add_argument('--output', type=Path, help='Optional report path; the only file written by this runner') args = parser.parse_args() require(FIRST <= args.through <= LAST, f'--through must be in {FIRST}..{LAST}') raw = (ROOT / 'journal.json').read_bytes() rows = json.loads(raw) require(isinstance(rows, list) and rows, 'Empty journal') require(len(rows) <= STEP_COUNT, 'Journal exceeds requested200-action range') require([r['step'] for r in rows] == list(range(FIRST, FIRST + len(rows))), 'Nonsequential/duplicate journal steps') require(rows[-1]['step'] >= args.through, f'Requested C{args.through} is unfinished; use --through {rows[-1]["step"]}') selected = [r for r in rows if r['step'] <= args.through] if args.through == LAST: require(len(rows) == STEP_COUNT, 'Final verification requires all200 completed actions') require(not (ROOT / 'pending.json').exists(), 'Final verification found an unfinished pending step') required = {'step', 'title', 'question', 'inputs', 'sources', 'opened_utc', 'predecessor_sha256', 'results', 'finding', 'reassessment', 'checks', 'closed_utc', 'sha256'} expected_previous = 'C931:' + PREDECESSOR_HASH previous_close = None for row in selected: require(set(row) == required, f'C{row["step"]}: unexpected record schema') require(record_digest(row) == row['sha256'], f'C{row["step"]}: record hash mismatch') require(row['predecessor_sha256'] == expected_previous, f'C{row["step"]}: predecessor chain mismatch') require(all(isinstance(row[k], str) and row[k].strip() for k in ('title', 'question', 'finding', 'reassessment')), f'C{row["step"]}: empty research metadata') require(isinstance(row['sources'], list) and row['sources'], f'C{row["step"]}: missing sources') opened, closed = datetime.fromisoformat(row['opened_utc']), datetime.fromisoformat(row['closed_utc']) require(opened.tzinfo is not None and closed.tzinfo is not None and opened <= closed, f'C{row["step"]}: invalid timestamps') require(previous_close is None or previous_close <= opened, f'C{row["step"]}: overlapping numbered sequence') previous_close, expected_previous = closed, row['sha256'] manifest, current, absent, predecessor_live = verify_sources(args.workspace.resolve(), args.require_current_sources) versioned_inputs, versioned_bindings = verify_versioned_inputs(selected) sys.dont_write_bytecode = True sys.addaudithook(audit_read_only) replays = [replay(row, versioned_inputs.get(row['step'])) for row in selected] # A lead agent may append later completed steps during an intermediate run. # Existing records must remain byte-equivalent after canonical serialization; # final verification additionally requires the entire file to stay unchanged. after_raw = (ROOT / 'journal.json').read_bytes() after_rows = json.loads(after_raw) require(after_rows[:len(rows)] == rows, 'Journal prefix mutated during replay') require([r['step'] for r in after_rows] == list(range(FIRST, FIRST + len(after_rows))), 'Invalid concurrent journal append') if args.through == LAST: require(after_raw == raw, 'Final verification journal changed during replay') report = { 'status': 'PASS', 'first': FIRST, 'last': args.through, 'sequential_steps': len(selected), 'final_200_step_verification': args.through == LAST, 'requested_steps': STEP_COUNT, 'preseal_199_step_verification': args.through == LAST - 1, 'journal_steps_available_at_read': len(rows), 'journal_sha256_at_read': hashlib.sha256(raw).hexdigest(), 'last_record_sha256': expected_previous, 'record_metadata_and_results_replayed': True, 'journal_hash_chain_verified': True, 'C931_predecessor_snapshot_verified': True, 'C931_current_record_checked': predecessor_live, 'source_snapshot_hashes_verified': len(manifest), 'current_source_copies_checked': len(current), 'current_sources': current, 'current_originals_absent': absent, 'latest_File52c_sha256': PRIMARY_HASH, 'versioned_input_bindings_verified': versioned_bindings, 'historical_input_read_count': sum(sum(r['historical_input_reads'].values()) for r in replays), 'step_checks_replayed': sum(r['checks'] for r in replays), 'independent_reversal_used': True, 'single_pass_inverse_calls': sum(r['single_pass_inverse_calls'] for r in replays), 'second_inverse_guard_violations': 0, 'inverse_guard_scope': 'Nested inv calls and inversion of provenance-tracked inv outputs are rejected. This is not a proof against arbitrary manual reimplementation or deliberate provenance erasure.', 'replay_writes_blocked': True, 'journal_prefix_unchanged_after_replay': True, 'journal_bytes_unchanged_after_replay': after_raw == raw, 'concurrent_completed_steps_appended': len(after_rows) - len(rows), 'replays': replays, 'independent_review_notes': sorted(p.name for p in (ROOT / 'evidence').glob('REVIEW_*.md')), 'scope': 'Exact record and script replay, source identity, and declared operation checks. Check counts are not independent discoveries or statistical evidence. Historical source suites were not rerun.' } text = json.dumps(report, indent=2, ensure_ascii=False) + '\n' if args.output: target = args.output.resolve() require(target != ROOT / 'journal.json' and target.suffix == '.json', 'Output must be a separate JSON report') target.write_text(text) print(text, end='') if __name__ == '__main__': try: main() except (VerificationError, KeyError, ValueError, OSError) as exc: print(json.dumps({'status': 'FAIL', 'error': str(exc)}, ensure_ascii=False), file=sys.stderr) sys.exit(1)