Verify native behavior creation recovery and source authoring contracts

harness/experiments/behavior-creation-native-20260911/verify-recovery.py

@@ -0,0 +1,89 @@
+"""Verify immutable recovery evidence; never access live profile/work or UI."""
+from pathlib import Path
+import copy
+import datetime
+import hashlib
+import json
+import os
+import subprocess
+import time
+
+ROOT = Path(__file__).resolve().parents[3]
+OUT = Path(__file__).resolve().parent
+STAGE = ROOT / '.local/native-qa/behavior-20260911-v4'
+EVIDENCE = STAGE / 'recovery-evidence'
+EXPECTED = STAGE / 'checkpoint-creation-reopen'
+ACTUAL = STAGE / 'checkpoint-recovery-cancel'
+EXE = ROOT / '.local/behavior-creation-oracle-20260911/behavior-creation-contract-v2.exe'
+
+
+def record(path):
+    data = path.read_bytes()
+    return {'path': path.relative_to(ROOT).as_posix(), 'bytes': len(data),
+            'sha256': hashlib.sha256(data).hexdigest()}
+
+
+def inventory(directory):
+    return {p.relative_to(directory).as_posix(): record(p)
+            for p in sorted(directory.rglob('*')) if p.is_file()}
+
+
+artifacts = [record(p) for p in (Path(__file__), OUT / 'manifest.json', EXE,
+    EVIDENCE / 'before-interruption.json', EVIDENCE / 'after-recovery-and-typing.json',
+    EVIDENCE / 'restored-draft.png', STAGE / 'recovery-launch.json', STAGE / 'bin/arcadia-studio.exe')]
+assert artifacts[1]['sha256'] == '356dfbb27dec04764a4b3f2372602f39953b472baac0a3de3e72acd9c1d76b8a'
+assert artifacts[2]['sha256'] == '056eefd444924916bf6ac3c4897ab2ff4a8a4bb892e1dbc16adad32e59bf6a36'
+before = json.loads((EVIDENCE / 'before-interruption.json').read_text(encoding='utf-8-sig'))
+after = json.loads((EVIDENCE / 'after-recovery-and-typing.json').read_text(encoding='utf-8-sig'))
+draft = before['behavior']['draft']
+expected_name = 'Recovery — secours Δ «operating modes»'
+checks = {'literal_name_before': draft['name'] == expected_name,
+          'literal_name_after': after['behavior']['draft']['name'] == expected_name + ' · resumed',
+          'sa_machine': draft['layer'] == 'Sa' and draft['family'] == 'Machine',
+          'selected_exact_system_actor': draft['owner'] == 'c0629449-1209-4b50-afac-f558a9a0fc7f',
+          'first_view_selected': draft['first_view'] is True}
+normalized = copy.deepcopy(after['behavior'])
+if checks['literal_name_after']:
+    normalized['draft']['name'] = expected_name
+checks['complete_behavior_equal_except_deliberate_suffix'] = normalized == before['behavior']
+for field in ('accepted', 'workspace', 'property', 'workflow', 'selected'):
+    checks[field + '_exact_json_equality'] = before[field] == after[field]
+reserved = {field: draft[field] for field in ('element', 'diagram', 'placement')}
+checks['reserved_ids_unchanged'] = all(after['behavior']['draft'][k] == v for k, v in reserved.items())
+inputs = {str(p.relative_to(ROOT)): inventory(p) for p in (EXPECTED, ACTUAL)}
+expected_names = set(inputs[str(EXPECTED.relative_to(ROOT))])
+actual_names = set(inputs[str(ACTUAL.relative_to(ROOT))])
+checks['complete_file_inventory_equal'] = expected_names == actual_names
+byte_results = {name: (EXPECTED / name).read_bytes() == (ACTUAL / name).read_bytes()
+                for name in expected_names if name in actual_names}
+checks['complete_file_bytes_equal'] = all(byte_results.values()) and len(byte_results) == len(expected_names)
+checks['accepted_snapshot_exactly_matches_saved_checkpoint'] = (
+    set(before['accepted']['files']) == expected_names and
+    all(text.encode('utf-8') == (EXPECTED / name).read_bytes()
+        for name, text in before['accepted']['files'].items()))
+checks['reserved_objects_not_committed'] = all(
+    all(value.encode('utf-8') not in (ACTUAL / name).read_bytes() for value in reserved.values())
+    for name in actual_names)
+command = [str(EXE), 'same', str(EXPECTED / 'project.arcadia'), str(ACTUAL / 'project.arcadia')]
+env = os.environ.copy()
+env['PATH'] = str(STAGE / 'bin') + os.pathsep + env.get('PATH', '')
+started = datetime.datetime.now(datetime.timezone.utc).isoformat()
+tick = time.perf_counter()
+run = subprocess.run(command, cwd=ROOT, env=env, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=60)
+elapsed = time.perf_counter() - tick
+checks['oracle_same_pass'] = run.returncode == 0
+checks['input_files_unchanged'] = inputs == {str(p.relative_to(ROOT)): inventory(p) for p in (EXPECTED, ACTUAL)}
+checks['evidence_and_prior_manifest_unchanged'] = artifacts == [record(ROOT / item['path']) for item in artifacts]
+result = {'command_argv': command, 'exit_code': run.returncode,
+          'elapsed_seconds': round(elapsed, 6), 'stdout': run.stdout, 'stderr': run.stderr}
+(OUT / 'recovery-commands-results.log').write_text(json.dumps(result, indent=2) + '\n', encoding='utf-8', newline='\n')
+manifest = {'reviewer': 'Codex-9e8r6t /root/palette_finish', 'started_utc': started,
+            'finished_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
+            'scope': 'Immutable evidence only; native observations supplied separately by root. No live work/profile/UI/Cargo access; no persisted Undo claim.',
+            'artifacts': artifacts, 'input_file_inventories': inputs, 'checks': checks,
+            'reserved_ids': reserved, 'reopen_recovery_cancel_direct_byte_comparisons': byte_results,
+            'comparison_origin': 'checkpoint-creation-reopen', 'command': result,
+            'log': record(OUT / 'recovery-commands-results.log'), 'pass': all(checks.values())}
+(OUT / 'recovery-manifest.json').write_text(json.dumps(manifest, indent=2) + '\n', encoding='utf-8', newline='\n')
+print(json.dumps({'pass': manifest['pass'], 'checks': checks, 'manifest': record(OUT / 'recovery-manifest.json')}))
+raise SystemExit(0 if manifest['pass'] else 1)