#!/usr/bin/env python3 """Read-only replay of C1132--C1431, 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 = 1132, 1431 STEP_COUNT = LAST - FIRST + 1 STRATEGY_HASH = '9c9aa357f5483b3af1dbb5f0025ee514fa28160576a037be03037aa49bf01476' STRATEGY_CHECKPOINTS = tuple(range(1181, 1432, 50)) PREDECESSOR_HASH = '342be9d0c96ff1b6290524df190b12c0e065d492e91ccb546a7841edd5f72f01' 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 = { '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'], } 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}') inherited = json.loads((ROOT / 'evidence/INHERITED_MANIFEST.json').read_text()) require(isinstance(inherited, dict) and inherited, 'Missing inherited-input manifest') for relative, expected in inherited.items(): path = safe_packet_path(relative, prefix=('inherited',)) require(digest(path) == expected, f'Inherited input hash mismatch: {relative}') predecessor = json.loads((ROOT / 'evidence/PREDECESSOR.json').read_text()) require(predecessor['step'] == FIRST - 1, 'Wrong predecessor step') require(record_digest(predecessor) == predecessor['sha256'] == PREDECESSOR_HASH, 'C1131 predecessor hash mismatch') artifacts = predecessor['results']['final_artifacts'] require(isinstance(artifacts, list) and artifacts, 'Missing predecessor final artifact bindings') by_path = {item['path']: item for item in artifacts} require(len(by_path) == len(artifacts), 'Duplicate predecessor final artifact paths') bindings = json.loads((ROOT / 'evidence/PREDECESSOR_BINDINGS.json').read_text()) require(isinstance(bindings, dict) and len(bindings) == len(artifacts), 'Predecessor snapshots must cover all final artifacts') require(len(set(bindings.values())) == len(bindings) and set(bindings.values()) == set(by_path), 'Incomplete or duplicate predecessor artifact coverage') for alias, original_path in bindings.items(): path = safe_packet_path(alias, prefix=('inherited',)) item = by_path[original_path] require(alias in inherited and inherited[alias] == item['sha256'], f'Predecessor inherited-manifest mismatch: {alias}') require(digest(path) == item['sha256'] and path.stat().st_size == item['bytes'], f'Predecessor final artifact identity mismatch: {alias}') prior_journal_name = 'inherited/C1131_journal.json' require(prior_journal_name in inherited, 'Missing inherited C1131 journal identity') prior_rows = json.loads((ROOT / prior_journal_name).read_text()) require(prior_rows and prior_rows[-1] == predecessor, 'Inherited journal does not end in exact C1131 predecessor') for index, row in enumerate(prior_rows): require(record_digest(row) == row['sha256'], 'Inherited predecessor journal record hash mismatch') if index: require(row['step'] == prior_rows[index-1]['step'] + 1 and row['predecessor_sha256'] == prior_rows[index-1]['sha256'], 'Inherited predecessor journal chain mismatch') old_journal = workspace / 'c932_c1131/journal.json' predecessor_live = False if old_journal.is_file(): require(json.loads(old_journal.read_text()) == prior_rows, 'Current predecessor journal differs from inherited snapshot') predecessor_live = True return manifest, current, absent, predecessor_live, inherited, bindings def safe_packet_path(value, prefix=()): require(isinstance(value, str) and value and '\\' not in value, 'Invalid packet-relative path') path = Path(value) require(not path.is_absolute() and '..' not in path.parts and path.as_posix() == value, f'Unsafe packet path: {value}') require(path.parts[:len(prefix)] == prefix, f'Invalid packet path scope: {value}') resolved = (ROOT / path).resolve() require(ROOT in resolved.parents and resolved.is_file(), f'Missing or escaping packet file: {value}') return ROOT / path def strategy_section_index(data): """Partition every source byte at Markdown headings, preserving line endings.""" require(isinstance(data, bytes) and data, 'Empty Strategy full text') lines = data.splitlines(keepends=True) starts = [i for i, line in enumerate(lines) if re.match(rb'#{1,6}[ \t]+', line)] if not starts or starts[0] != 0: starts.insert(0, 0) result = [] for position, first in enumerate(starts): stop = starts[position+1] if position+1 < len(starts) else len(lines) content = b''.join(lines[first:stop]) heading = lines[first].decode('utf-8').strip() if re.match(rb'#{1,6}[ \t]+', lines[first]) else '(preamble)' result.append({'heading': heading, 'start_line': first+1, 'end_line': stop, 'bytes': len(content), 'sha256': hashlib.sha256(content).hexdigest()}) require(sum(item['bytes'] for item in result) == len(data), 'Incomplete Strategy section partition') return result def verify_strategy_reviews(rows): """Unnumbered full reviews follow each completed fifty-action block.""" source = ROOT / 'evidence/sources/Strategy.md' data = source.read_bytes() expected_hash = hashlib.sha256(data).hexdigest() require(expected_hash == STRATEGY_HASH, 'Strategy full-text identity mismatch') manifest = json.loads((ROOT / 'evidence/SOURCE_MANIFEST.json').read_text()) require(manifest.get('Strategy.md') == expected_hash, 'Strategy source-manifest identity mismatch') sections = strategy_section_index(data) records = {row['step']: row for row in rows} reviews = [] for number in STRATEGY_CHECKPOINTS: if number not in records: continue path = safe_packet_path(f'evidence/strategy_reviews/review_{number}.json', prefix=('evidence', 'strategy_reviews')) item = json.loads(path.read_text()) required = {'schema_version', 'checkpoint', 'completed_record_sha256', 'block', 'strategy', 'full_text_reviewed', 'sections', 'assessment', 'reviewed_utc'} require(isinstance(item, dict) and set(item) == required, f'C{number}: invalid Strategy review schema') require(type(item['schema_version']) is int and item['schema_version'] == 1 and type(item['checkpoint']) is int and item['checkpoint'] == number, f'C{number}: wrong Strategy review version/checkpoint') require(record_digest(records[number]) == records[number]['sha256'] == item['completed_record_sha256'], f'C{number}: Strategy review does not bind the completed record') require(item['block'] == {'first': number-49, 'last': number, 'actions': 50}, f'C{number}: invalid fifty-action review block') require(item['strategy'] == {'snapshot': 'Strategy.md', 'sha256': expected_hash, 'bytes': len(data)}, f'C{number}: Strategy review source identity mismatch') require(item['full_text_reviewed'] is True and item['sections'] == sections, f'C{number}: incomplete Strategy full-text/section coverage') assessment_keys = {'strategy_alignment', 'last_fifty_actions', 'explanatory_progress', 'remaining_uncertainty', 'next_best_step'} assessment = item['assessment'] require(isinstance(assessment, dict) and set(assessment) == assessment_keys and all(isinstance(value, str) and value.strip() for value in assessment.values()), f'C{number}: incomplete substantive Strategy assessment') reviewed = datetime.fromisoformat(item['reviewed_utc']) closed = datetime.fromisoformat(records[number]['closed_utc']) require(reviewed.tzinfo is not None and reviewed >= closed, f'C{number}: Strategy review must follow completion') if number+1 in records: require(reviewed <= datetime.fromisoformat(records[number+1]['opened_utc']), f'C{number}: next action began before Strategy review') reviews.append({'checkpoint': number, 'path': str(path.relative_to(ROOT)), 'sha256': digest(path), 'completed_record_sha256': item['completed_record_sha256'], 'strategy_sha256': expected_hash, 'full_text_bytes': len(data), 'section_count': len(sections), 'reviewed_utc': item['reviewed_utc']}) return reviews def replay(row): 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} 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() if number == 1424 and relative_path == 'model/reader_package_verification.json': # C1424 iterates a set of ZIP names. Link report order is presentation only. # Preserve every value and multiplicity; accept no other result difference. recorded_binding = row['results']['verification'] stored = (ROOT / relative_path).read_bytes() require(hashlib.sha256(stored).hexdigest() == recorded_binding['sha256'] and len(stored) == recorded_binding['bytes'], 'C1424: stored link report identity mismatch') generated_object, stored_object = json.loads(data), json.loads(stored) for obj in (generated_object, stored_object): require(isinstance(obj, dict) and isinstance(obj.get('local_links'), list), 'C1424: invalid link report') obj['local_links'] = sorted(obj['local_links'], key=lambda item: json.dumps(item, sort_keys=True)) require(json.dumps(generated_object, sort_keys=True) == json.dumps(stored_object, sort_keys=True), 'C1424: substantive link report mismatch') data = stored return {'path': relative_path, 'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data)} fake = types.ModuleType('research') exported = {'Path': Path, 'F': Fraction, 'ROOT': 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') return {'step': number, 'script_sha256': digest(script), 'checks': len(row['checks']), 'single_pass_inverse_calls': calls['inverse']} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--through', type=int, default=LAST, help='Completed last step to replay;1430 is the299-action preseal prefix; default1431 requires all300') 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 requested300-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 all300 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 = 'C1131:' + 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, inherited, predecessor_bindings = verify_sources(args.workspace.resolve(), args.require_current_sources) strategy_reviews = verify_strategy_reviews(rows) sys.dont_write_bytecode = True sys.addaudithook(audit_read_only) replays = [replay(row) 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_300_step_verification': args.through == LAST, 'requested_steps': STEP_COUNT, 'preseal_299_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, 'C1131_predecessor_snapshot_verified': True, 'C1131_current_record_checked': predecessor_live, 'inherited_input_hashes_verified': len(inherited), 'predecessor_final_artifacts_verified': len(predecessor_bindings), 'source_snapshot_hashes_verified': len(manifest), 'current_source_copies_checked': len(current), 'current_sources': current, 'current_originals_absent': absent, 'latest_File52c_sha256': PRIMARY_HASH, 'strategy_review_certificates': strategy_reviews, 'strategy_review_count': len(strategy_reviews), 'all_six_strategy_reviews_verified': len(strategy_reviews) == 6, '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')), 'presentation_order_normalization': {'step': 1424, 'artifact': 'model/reader_package_verification.json', 'field': 'local_links', 'rule': 'Order-only JSON list normalization; values, multiplicities, other fields and original artifact hash remain exact.'}, 'scope': 'Exact record and script replay, source identity, and declared operation checks, with declared order-only normalization of the C1424 link report. 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)