From 3fd58f2cf89579be7d128ba593ea98d689ee880e Mon Sep 17 00:00:00 2001 From: Digital Drosophila Archive Date: Wed, 30 Sep 2026 21:38:56 +0800 Subject: [PATCH] Add post-exit normal acquisition audit with frozen gate regression --- docs/progress/phase4ar-normal-acquisition.md | 19 ++ .../phase4ar-supervisor-start-binding.json | 14 + .../phase4ar-terminal-audit-cpu-tests.json | 24 ++ tools/normal-conditioning-audit/README.md | 61 +++++ .../normal_finalize.py | 227 +++++++++++++++++ .../normal_outcome.py | 239 ++++++++++++++++++ .../terminal_ledger.py | 237 +++++++++++++++++ .../test_normal_finalize.py | 121 +++++++++ .../test_normal_outcome.py | 176 +++++++++++++ .../test_terminal_ledger.py | 228 +++++++++++++++++ 10 files changed, 1346 insertions(+) create mode 100644 registry/phase4ar-normal-conditioning-active/phase4ar-supervisor-start-binding.json create mode 100644 registry/phase4ar-normal-conditioning-active/phase4ar-terminal-audit-cpu-tests.json create mode 100644 tools/normal-conditioning-audit/README.md create mode 100644 tools/normal-conditioning-audit/normal_finalize.py create mode 100644 tools/normal-conditioning-audit/normal_outcome.py create mode 100644 tools/normal-conditioning-audit/terminal_ledger.py create mode 100644 tools/normal-conditioning-audit/test_normal_finalize.py create mode 100644 tools/normal-conditioning-audit/test_normal_outcome.py create mode 100644 tools/normal-conditioning-audit/test_terminal_ledger.py diff --git a/docs/progress/phase4ar-normal-acquisition.md b/docs/progress/phase4ar-normal-acquisition.md index ff88aa4..7182627 100644 --- a/docs/progress/phase4ar-normal-acquisition.md +++ b/docs/progress/phase4ar-normal-acquisition.md @@ -66,3 +66,22 @@ delayed reward 或任何游戏、动作解码器、RL、视觉、Web/3D。 不读取大体量脑输出；其进度结果不是完整性报告或科学结论。 正在执行的 child 在没有真实退出记录时明确保持 UNKNOWN，不能由 PID 或一个提前写出的 results 文件推定成功。完整 run 完成后仍需最终独立审计。 + +## 终审工具准备进展（不改变正在运行的实验） + +独立终审工具已完成服务器联合测试：42/42 通过，0 失败、0 跳过。 +其中学习结果判定测试通过冻结的原指标代码生成六神经元合成报告， +包含正常通过、终点未达 3/5、单一 H 对照来源失败、严格等号不通过等情形。 +这些不是本次真实学习结果，也没有读取中途学习指标。 + +`tools/normal-conditioning-audit/normal_finalize.py check` 已在服务器实测： +发现注册进程仍在运行后按设计拒绝终审，未创建终审输出目录。 +真正的完整审计和固定 gate 分析只能在全部任务结束、实际退出为 0 后进行。 +终审产物将位于独立的 `phase4ar-normal-conditioning-final-audit/`， +不会向已完成的原实验目录追加或替换文件。 + +终审保留原账本的全部字段，不改 PASS/FAIL 标志；完成记录在账本检查后 +才生成这一文件边界由单独的真实文件清单核对。任何额外文件、变更、 +未完成进程或来源矛盾均不得被当成完整的学习阳性/阴性结论。 +即便普通 acquisition 通过，delayed/reversal/retention 仍未运行，不能 +生成候选模型、宣称学习已经可用或自动启动后续实验。 diff --git a/registry/phase4ar-normal-conditioning-active/phase4ar-supervisor-start-binding.json b/registry/phase4ar-normal-conditioning-active/phase4ar-supervisor-start-binding.json new file mode 100644 index 0000000..74dcc20 --- /dev/null +++ b/registry/phase4ar-normal-conditioning-active/phase4ar-supervisor-start-binding.json @@ -0,0 +1,14 @@ +{ + "schema": "phase4ar-supervisor-start-binding-v1", + "registration_sha256": "b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740", + "observation_scope": "Actual immutable startup record bytes read while the already launched run was live; not an exit or completion receipt", + "source_commit": "80aeb2093db655329774eff7b32fb6e417b8b40b", + "supervisor_pid": 441824, + "files": { + "launch-attempt.json": {"bytes": 229, "sha256": "93274a58b83a3e508e378f3972a62c2e43a461bc4a890314f6e38530fa05719c"}, + "launch.json": {"bytes": 523, "sha256": "94c7dba031149451a0d653dd7c083fa5c918b9363fa1caef31b988d02d72672e"}, + "supervisor-start.json": {"bytes": 187, "sha256": "3fa980cfcf0993bb7aa29de87eff31f5bc48be215a2cfbc03b715d4b84672839"} + }, + "learning_success_claimed": false, + "automatic_retry_authorized": false +} diff --git a/registry/phase4ar-normal-conditioning-active/phase4ar-terminal-audit-cpu-tests.json b/registry/phase4ar-normal-conditioning-active/phase4ar-terminal-audit-cpu-tests.json new file mode 100644 index 0000000..a4450ab --- /dev/null +++ b/registry/phase4ar-normal-conditioning-active/phase4ar-terminal-audit-cpu-tests.json @@ -0,0 +1,24 @@ +{ + "errors": 0, + "failures": 0, + "no_neural_modules": true, + "pass_all": true, + "schema": "normal-terminal-audit-CPU-tests-v1", + "scope": "Synthetic temporary files and six-neuron CPU fixtures through frozen metrics; no actual scientific outcomes or GPU executed", + "skips": 0, + "source_sha256": { + "README.md": "4873b85049c9b9e17cb29991d8fd805b674248fbf44e1026ce575804ce3ca270", + "normal_finalize.py": "ef387892ec4d82adfd47c103a831b47adb4aeed978c7017fe371be33a8f9e5f9", + "normal_outcome.py": "ab2667622b1ea5374cef7508afd063f2419952de08e98a07389378a23e593d5b", + "progress_normal.py": "081d5de098c9a249a13394f9669c76d79dd421ca28765397d3a9d8bbbde051d0", + "terminal_ledger.py": "73f4cfdd47104d12d58c34f3bd6cff8fb5292b2e9d0566220b0f1932ef1cdd52", + "test_normal_finalize.py": "8d956a6b3777abe0a6a92aa8a521d5efb02091a115ec1ef0d866a632dbc33d45", + "test_normal_outcome.py": "88fe863b3ddd224817555af29db2b24e51147690cfc1d6b0af0f306436cc24ac", + "test_progress_normal.py": "f1a34c8d0933cb7226cbeea8afed7321df771e6ad8b1979072af13a5e5bb3c9c", + "test_terminal_ledger.py": "ade5738490f2c2770b916154690531a75cd5759829a2cc2ebc9d52422b8890d4" + }, + "sources_unchanged": true, + "test_output": "test_actual_live_process_refuses_even_with_complete_files_and_no_writes (test_normal_finalize.FinalizeTests) ... ok\ntest_actual_process_query_required_not_caller_assumption (test_normal_finalize.FinalizeTests) ... ok\ntest_argv_classifier_ignores_grep_substrings_and_includes_real_children (test_normal_finalize.FinalizeTests) ... ok\ntest_changed_startup_binding_is_rejected (test_normal_finalize.FinalizeTests) ... ok\ntest_completion_inventory_or_registration_mismatch_rejected (test_normal_finalize.FinalizeTests) ... ok\ntest_exclusive_save_never_overwrites (test_normal_finalize.FinalizeTests) ... ok\ntest_failed_CPU_child_preserves_exit_and_refuses_same_name_retry (test_normal_finalize.FinalizeTests) ... ok\ntest_finalization_checks_live_guard_before_source_or_output_changes (test_normal_finalize.FinalizeTests) ... ok\ntest_missing_process_without_real_exit_does_not_authorize_terminal_audit (test_normal_finalize.FinalizeTests) ... ok\ntest_nonzero_parent_exit_is_not_scientific_failure_or_success (test_normal_finalize.FinalizeTests) ... ok\ntest_real_harmless_CPU_child_exit_and_log_hashes_are_saved (test_normal_finalize.FinalizeTests) ... ok\ntest_stopped_record_always_refuses_completion (test_normal_finalize.FinalizeTests) ... ok\ntest_success_only_means_ready_not_learning_and_does_not_hash_brain_outputs (test_normal_finalize.FinalizeTests) ... ok\ntest_H_one_control_branch_failure_and_equality_are_strict (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_actual_source_digest_registration_and_integrity_fail_closed (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_canonical_scope_unicode_no_newline_and_nonfinite_rejected (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_exact_cell_replicate_curve_inventory_missing_not_scientific_negative (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_missing_outer_and_failed_analysis_are_incomplete_not_scientific_fail (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_pinned_real_frozen_gates_compact_output_pass_and_no_mutation (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_rehashed_top_boolean_and_predicate_mutations_cannot_forge_success (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_terminal_same_replicate_fail_cannot_use_intermediate_or_other_cell (test_normal_outcome.NormalOutcomeTests) ... ok\ntest_failed_child_and_real_supervisor_exit_and_stop_record (test_progress_normal.ProgressTests) ... ok\ntest_latest_receipt_mutation_is_unknown_not_metadata_success (test_progress_normal.ProgressTests) ... ok\ntest_missing_fields_bad_sequence_or_registration_are_unknown (test_progress_normal.ProgressTests) ... ok\ntest_no_registration_not_learning_or_inferred_preparation_liveness (test_progress_normal.ProgressTests) ... ok\ntest_probe_and_train_commits_separate_and_no_neural_files_read (test_progress_normal.ProgressTests) ... ok\ntest_registered_and_launch_pid_are_not_completion (test_progress_normal.ProgressTests) ... ok\ntest_sampler_started_vs_committed_and_latest_actual_exit (test_progress_normal.ProgressTests) ... ok\ntest_written_result_never_substitutes_for_verify_commit (test_progress_normal.ProgressTests) ... ok\ntest_authentic_reports_preserved_and_exact_tree_inventory_read_only (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_duplicate_json_keys_rejected (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_exit_code_bool_or_failed_or_wrong_command_refused (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_external_startup_pin_required_and_tamper_rejected (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_extra_file_and_changed_build_rejected (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_historical_pass_and_counters_cannot_be_rewritten (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_historical_payload_and_registered_source_mutation_rejected (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_missing_completion_rejects_without_creating_controls (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_missing_exit_or_live_lock_refused (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_missing_header_cannot_auto_initialize (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_pending_start_or_stopped_marker_not_a_terminal_result (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_schema_and_filesystem_task_key_match_actual_frozen_source (test_terminal_ledger.TerminalLedgerTests) ... ok\ntest_supervisor_log_change_during_read_rejected_at_end (test_terminal_ledger.TerminalLedgerTests) ... ok\n\n----------------------------------------------------------------------\nRan 42 tests in 44.228s\n\nOK\n", + "tests": 42, + "wall_seconds": 44.32410157699633 +} diff --git a/tools/normal-conditioning-audit/README.md b/tools/normal-conditioning-audit/README.md new file mode 100644 index 0000000..4303638 --- /dev/null +++ b/tools/normal-conditioning-audit/README.md @@ -0,0 +1,61 @@ +# Phase 4AR read-only progress and post-run audit + +These tools are separate from the 16 registered scientific source files. They +never launch a brain, change plastic delta, resample inputs, resume a partial +epoch or authorize another assay. Tests use synthetic temporary data only. + +## While the experiment runs + +`progress_normal.py` reads the small committed ledger records. It distinguishes +completed CPU-verified epochs, completed probes, pending processes and mere +result files. It is not a whole-tree integrity check or a learning gate. + +`normal_finalize.py check` makes a real Linux process query and examines the +actual terminal receipts. It intentionally refuses a live or incomplete run, +even when a completion-looking file exists. An absent PID alone proves nothing. +No final-audit directory or result is created by this check. + +## After all registered work has ended + +An operator must first provision this exact, empty, separately owned directory: + +`/data/digital-drosophila/experiments/phase4ar-normal-conditioning-final-audit/` + +Then run the committed `normal_finalize.py audit` with the existing science +environment and `-B`. The driver refuses uncommitted audit code, a nonempty +destination, missing/failed exit evidence, a live runtime or a stopped phase. +There is no automatic retry after a partial finalization. + +The terminal ledger's authentic original PASS was written before +`audit/execution-complete.json` existed. A fresh unmodified ledger audit will +therefore report that exact one extra file. `terminal_ledger.py` retains both +original reports unchanged. It accepts this boundary only if every other field, +the full order, head, payload and compiler-cache identity agree, and both byte +totals differ by exactly that completion file's actual length. It never deletes +the file or changes a failure flag to PASS. Any other difference is rejected. + +The terminal inventory includes all experiment payload, registered external +inputs and all six administrative supervisor files. Startup identities were +independently bound while the run was live. The driver additionally checks real +process quiescence before and after analysis; file existence is not a lock or +an atomic filesystem snapshot. + +The frozen `normal_metrics.analyze` then runs in a fresh CPU process against all +33,920 task records and 3,920 probes. Its real exit and file bytes are recorded. +The old frozen data are reverified in a separate integrity process. Full +terminal inventory is read again after analysis. All new reports and their +manifest/checksums live outside the original experiment. + +`normal_outcome.py` checks the internally consistent fixed terminal report: +all eight cells, the same-replicate conjunction, all controls and both H control +source mappings. Incomplete/invalid evidence is not a scientific negative result. +It cannot authenticate a caller's evidence flags itself; the driver must supply +the actual verified files and process evidence. + +Even `NORMAL-ACQUISITION-PASS` is **not usable learning** or a model candidate. +Delayed reward, reversal and retention remain unrun and are not automatically +authorized. No parameter, seed or endpoint is chosen using intermediate results. + +Historical integrity checks do not rerun the five old reference trials and are +never labeled as new numerical regression. Git stores code and selected small +evidence, not a second copy of the bulk experiment dataset. diff --git a/tools/normal-conditioning-audit/normal_finalize.py b/tools/normal-conditioning-audit/normal_finalize.py new file mode 100644 index 0000000..2bdf5c9 --- /dev/null +++ b/tools/normal-conditioning-audit/normal_finalize.py @@ -0,0 +1,227 @@ +"""Post-completion ONLY: fresh terminal inventory, fixed science, historical seals. + +No training, repair, retry or change to the registered model/experiment. Output +goes to an independently provisioned EMPTY sibling directory. The real normal +runner's final ledger report is retained unchanged, even though a subsequent +full-tree audit sees its separately authenticated completion report as extra. +""" +from __future__ import annotations +import argparse +from datetime import datetime, timezone +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import shlex +import shutil +import subprocess +import sys +import time + +from progress_normal import safe, read_json, require + +ROOT = Path('/data/digital-drosophila') +MODEL = ROOT/'models/male-cns-lif/normal-conditioning-v1' +EXP = ROOT/'experiments/phase4ar-normal-conditioning' +CONTROL = ROOT/'logs/phase4ar-normal-conditioning' +OUT = ROOT/'experiments/phase4ar-normal-conditioning-final-audit' +TOOLS = Path(__file__).resolve().parent +PYTHON = ROOT/'models/male-cns-lif/.venv/bin/python' +REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740' +START_PINS = { + 'launch-attempt.json': dict(bytes=229,sha256='93274a58b83a3e508e378f3972a62c2e43a461bc4a890314f6e38530fa05719c'), + 'launch.json': dict(bytes=523,sha256='94c7dba031149451a0d653dd7c083fa5c918b9363fa1caef31b988d02d72672e'), + 'supervisor-start.json': dict(bytes=187,sha256='3fa980cfcf0993bb7aa29de87eff31f5bc48be215a2cfbc03b715d4b84672839'), +} + +def identity(path): + p=safe(path);require(p.is_file(),'Missing regular artifact: '+str(p)) + size=0;h=hashlib.sha256() + with p.open('rb') as f: + for part in iter(lambda:f.read(8<<20),b''):size+=len(part);h.update(part) + return dict(bytes=size,sha256=h.hexdigest()) + +def read(path): + return read_json(path)[0] + +def save(path,value): + with safe(path).open('x',encoding='utf-8',newline='\n') as f: + json.dump(value,f,sort_keys=True,indent=2,ensure_ascii=False,allow_nan=False);f.write('\n') + +def verify(pins): + for p,sha in pins.items():require(identity(p)['sha256']==sha,'Actual dependency changed: '+str(p)) + +def live_runtime_rows(ps_text,model,exp): + """Exact argv tokens, not substring matching or a missing-PID success rule.""" + runtime={str(model/n) for n in ('run_normal.py','normal_runner.py','normal_inputs.py')} + worker=str(model.parent/'conditioning-interface-v2/isolated_workers.py') + matches=[] + for line in ps_text.splitlines(): + fields=line.strip().split(None,1) + if len(fields)!=2:continue + try:pid=int(fields[0]);argv=shlex.split(fields[1]) + except (ValueError,TypeError):continue + if not argv:continue + owned=bool(runtime.intersection(argv)) + if worker in argv and '--job' in argv: + i=argv.index('--job')+1 + owned=owned or (i=reserve+(4<<30),'No space for terminal audit while preserving registered disk reserve') + sources=source_identity();save(OUT/'source-identity.json',sources);save(OUT/'readiness-before.json',ready) + try: + from terminal_ledger import terminal_inventory + from normal_outcome import summarize_normal_analysis, analysis_digest + terminal=terminal_inventory(EXP,REG_SHA,MODEL,supervisor_dir=CONTROL,supervisor_start_pins=START_PINS) + inventory=terminal.pop('verified_inventory') + save(OUT/'terminal-before.json',terminal);save(OUT/'verified-inventory.json',inventory) + command=[str(PYTHON),'-B',str(TOOLS/'normal_finalize.py'),'analyze-child', + '--inventory-sha256',identity(OUT/'verified-inventory.json')['sha256'], + '--terminal-sha256',identity(OUT/'terminal-before.json')['sha256']] + observed=child('analysis',command,43200) + analysis_sha=identity(OUT/'analysis.json')['sha256'];analysis=json.loads((OUT/'analysis.json').read_text()) + child('historical-integrity',[str(PYTHON),'-B',str(MODEL/'normal_prepare.py'),'integrity', + '--output',str(OUT/'historical-integrity-after.json')],1800) + historical=json.loads((OUT/'historical-integrity-after.json').read_text()) + require(historical.get('complete') is True,'Historical integrity failed') + ready_after=readiness();save(OUT/'readiness-after.json',ready_after) + require(ready_after['exit_identity']==ready['exit_identity'] and ready_after['completion_identity']==ready['completion_identity'], + 'Terminal control bytes changed while analyzing') + final=terminal_inventory(EXP,REG_SHA,MODEL,supervisor_dir=CONTROL,supervisor_start_pins=START_PINS) + require(final.pop('verified_inventory')==inventory,'Actual full inventory changed during science audit') + require(final==terminal,'Terminal audit changed during science analysis') + save(OUT/'terminal-after.json',final) + for p,info in sources['files'].items():require(identity(p)==info,'Terminal source changed during audit') + require(identity(OUT/'analysis.json')['sha256']==analysis_sha,'Actual analysis bytes changed') + # This outer binding contains actual byte/exit verification, never only a caller's PASS Boolean. + binding=dict(registration_sha256=REG_SHA,terminal_reconciliation_pass=True,historical_integrity_after_pass=True, + analysis_sha256=analysis_digest(analysis),analysis_file_sha256=analysis_sha, + actual_parent_exit_code=observed['exit_code'], + analyzed_source_sha256=read(EXP/'protocol/preregistered.json')['source_sha256'][str(MODEL/'normal_metrics.py')]) + outcome=summarize_normal_analysis(analysis,binding) + save(OUT/'outer-evidence.json',binding);save(OUT/'final-status.json',outcome) + return dict(outcome=outcome,seal=seal_output()) + except BaseException as exc: + if not (OUT/'audit-stopped.json').exists():save(OUT/'audit-stopped.json',dict(status='INCOMPLETE-AUDIT', + error_type=type(exc).__name__,error=str(exc),learning_failure_inferred=False,automatic_retry=False)) + raise + +if __name__=='__main__': + parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('mode',choices=('check','audit','analyze-child')) + parser.add_argument('--inventory-sha256');parser.add_argument('--terminal-sha256');args=parser.parse_args() + if args.mode=='check':print(json.dumps(readiness(),indent=2)) + elif args.mode=='audit':print(json.dumps(finalize(),indent=2)) + else: + require(args.inventory_sha256 and args.terminal_sha256,'Actual outer report hashes required') + analyze_child(args.inventory_sha256,args.terminal_sha256) diff --git a/tools/normal-conditioning-audit/normal_outcome.py b/tools/normal-conditioning-audit/normal_outcome.py new file mode 100644 index 0000000..d20a942 --- /dev/null +++ b/tools/normal-conditioning-audit/normal_outcome.py @@ -0,0 +1,239 @@ +"""Pure normal-only outcome aggregation; NO disk, processes or new gates. + +outer_evidence MUST be supplied by the real terminal auditor after complete +ledger/checksum and historical-integrity reconciliation, and an actual +parent-observed successful invocation of the fixed normal_metrics.analyze. +The caller must authenticate that process, source, consumed authority and saved +report bytes. analysis_sha256 below binds canonical parsed JSON, NOT the raw +pretty-printed file SHA (the caller must authenticate that separately). + +This function cannot establish any of those external facts from a Boolean or a +digest. It checks their declared binding and the frozen report's internal +structure/predicates only. Normal PASS is not candidate, usable learning, +behavior, full legacy-H reproduction, or permission to run later assays. +""" +from __future__ import annotations + +import hashlib +import json +import math + +import numpy as np + +ANALYSIS_SOURCE_SHA256 = '82af63d8fadd5a6b67083f5e303ebe88674b76f9a2a8a8fff082b5c828483533' +METRICS_SOURCE_SHA256 = '47319e21db3c8b030b1bafbf0703cd86fb588d23bb29436accbacdba73d08c3c' +CANONICAL_SCOPE = "JSON sort_keys=True separators=(',',':') ensure_ascii=False allow_nan=False; UTF-8; no newline" +GATE_SCHEMA = 'conditioning-learning-gates-v1' +H_MAPPING = 'each-DAN-control-source-separately-both-pass-v1' +MODES = ('physiological', 'matched_energy') +DANS = ('PAM15', 'PPL103') +CUES = ('4-ethylphenol', '3-hexanone') +ARMS = {'paired': 'paired', 'd_zero': 'D=0', 'frozen': 'frozen', 'random_reward': 'random reward', + 'unpaired': 'unpaired', 'cs_only': 'CS-only', 'dan_only': 'DAN-only'} +CONTROLS = tuple(ARMS.values())[1:] +POINTS = (0, 1, 2, 5, 10, 20, 50) +POPS = ('MBON', 'legacyDownstream', 'nonMBONDownstream') +PREDICATES = ('raw_signed_direction', 'beyond_each_actual_control', 'conditioned_final_g_exceeds_neutral', + 'conditioned_count_or_defined_population_first_latency_change', 'all_MBON_AB_euclidean_gain_positive', + 'legacyDownstream_AB_euclidean_gain_positive', 'nonMBONDownstream_AB_euclidean_gain_positive') +SCOPE_FALSE = ('external_authority_verified', 'scientific_learning_pass_claimed', + 'candidate_generated', 'execution_authorized', 'data_backed_claimed') +OUTER_KEYS = ('registration_sha256', 'terminal_reconciliation_pass', 'historical_integrity_after_pass', + 'analysis_sha256', 'actual_parent_exit_code', 'analyzed_source_sha256') + + +class Incomplete(ValueError): + pass + + +def require(ok, message): + if not ok: + raise ValueError(message) + + +def analysis_digest(report): + """Canonical content digest; caller separately authenticates saved bytes.""" + return hashlib.sha256(json.dumps(report, sort_keys=True, separators=(',', ':'), + ensure_ascii=False, allow_nan=False).encode('utf-8')).hexdigest() + + +def is_sha(value): + return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value) + + +def finite(value): + require(type(value) in (int, float) and math.isfinite(value), 'Missing/nonfinite scientific scalar') + return value + + +def exact_inventory(values, expected, label): + require(len(values) == len(set(values)), 'Duplicate ' + label) + actual = set(values) + require(actual <= expected, 'Unexpected ' + label) + if actual != expected: + raise Incomplete('Missing ' + label) + + +def scope(value): + require(value['schema'] == GATE_SCHEMA and all(value[k] is False for k in SCOPE_FALSE) and + value['frozen_metrics_source_sha256'] == METRICS_SOURCE_SHA256, 'Unsupported conditional-report authority claim/source') + + +def compact_curve(row, blocks): + """Check existing compact fields, not unretained raw-vector calculations.""" + metric = row['metrics']; cue = blocks[row['block_id']]['conditioned_cue'] + neutral = next(c for c in CUES if c != cue) + raw = metric['rawB'] + for phase in ('pre', 'post'): + require(set(raw[phase]) == set(CUES), 'Curve cue coverage differs') + for v in raw[phase].values(): finite(v) + require(finite(metric['rawS']) == (raw['post'][cue] - raw['post'][neutral]) - + (raw['pre'][cue] - raw['pre'][neutral]), 'Compact raw S/B relation differs') + require(set(metric['separation']) == set(POPS) and set(metric['cues']) == set(CUES), 'Curve population/cue coverage differs') + for value in metric['separation'].values(): + require(finite(value['gain']) == finite(value['post']) - finite(value['pre']), 'Compact separation relation differs') + for value in metric['cues'].values(): + require(set(value['populations']) == set(POPS) | {'MBON01', 'MBON12'}, 'Missing compact population') + for field in ('state_pre', 'state_post'): + require(set(value[field]) == {'step', 'neuron_idx', 'counts', 'v', 'g', 'ref'} and + all(is_sha(v) for v in value[field].values()), 'Missing compact state hashes') + + +def _summarize(report, outer): + if not isinstance(outer, dict) or any(k not in outer or outer[k] is None for k in OUTER_KEYS): + raise Incomplete('Complete outer audit and actual analysis process evidence required') + for key in ('terminal_reconciliation_pass', 'historical_integrity_after_pass'): + require(type(outer[key]) is bool and outer[key], 'Outer integrity not satisfied: ' + key) + require(type(outer['actual_parent_exit_code']) is int, 'Actual analysis parent exit must be an integer') + if outer['actual_parent_exit_code'] != 0: + raise Incomplete('Analysis process did not complete successfully') + require(outer['analyzed_source_sha256'] == ANALYSIS_SOURCE_SHA256 and is_sha(outer['registration_sha256']) and + outer['analysis_sha256'] == analysis_digest(report), 'Analysis source/content/registration authority differs') + require(report['schema'] == 'normal-scientific-analysis-v1' and report['status'] == 'CONDITIONAL-PENDING-OUTER-INTEGRITY' and + report['registration_sha256'] == outer['registration_sha256'] and is_sha(report['normal_plan_sha256']), 'Wrong normal report identity/status') + require(all(report[k] is False for k in ('subsequent_assays_authorized', 'candidate_generated', 'learning_pass_claimed', + 'scientific_learning_pass_claimed', 'data_backed_claimed', 'external_full_tree_anatomy_historical_integrity_verified_here')) and + report['subsequent_assays_status'] == 'NOT-RUN', 'Report contains out-of-scope science/authorization claims') + if report['analyzed_probe_calls'] != 3920 or report['curve_points'] != 1960: + raise Incomplete('Complete 3,920 probes and 1,960 curve points required') + require(type(report['analyzed_probe_calls']) is type(report['curve_points']) is int, 'Inventory counts must be integers') + registry, authority, a = report['registry'], report['authority'], report['analysis'] + require(registry['schema'] == 'conditioning-gate-registry-descriptor-v1' and registry['protocol_sha256'] == outer['registration_sha256'] and + authority['schema'] == 'conditioning-data-authority-descriptor-v1' and authority['registry_sha256'] == analysis_digest(registry) and + authority['population_binding_sha256'] == registry['population_binding_sha256'] and + all(is_sha(authority[k]) for k in ('verification_receipt_sha256', 'execution_inventory_sha256', 'population_binding_sha256')), + 'Registry/authority descriptor binding differs') + blocks = registry['blocks'] + expected_cells = {(m, d, c) for m in MODES for d in DANS for c in CUES} + identities = [(b['mode'], b['dan_type'], b['conditioned_cue'], b['replicate']) for b in blocks.values()] + exact_inventory(identities, {(*cell, r) for cell in expected_cells for r in range(1, 6)}, '40 registered replicate identities') + require(all(type(b['replicate']) is int for b in blocks.values()), 'Replicate labels must be integers') + curves = a['curves'] + curve_keys = [(r['block_id'], r['arm'], r['checkpoint']) for r in curves] + exact_inventory(curve_keys, {(b, arm, p) for b in blocks for arm in ARMS for p in POINTS}, 'complete curve inventory') + require(all(type(r['checkpoint']) is int for r in curves), 'Checkpoint labels must be integers') + for row in curves: compact_curve(row, blocks) + terminal = {(r['block_id'], r['arm']): r['metrics'] for r in curves if r['checkpoint'] == 50} + cells = a['cells']; cell_keys = [(c['mode'], c['dan_type'], c['conditioned_cue']) for c in cells] + exact_inventory(cell_keys, expected_cells, 'eight normal cells') + table, cell_lookup = [], {} + for cell, identity in zip(cells, cell_keys): + scope(cell) + require(cell['assay'] == 'normal' and cell['registry_sha256'] == authority['registry_sha256'] and + cell['external_data_authority_descriptor_sha256'] == analysis_digest(authority) and + type(cell['required_majority']) is int and cell['required_majority'] == 3 and + type(cell['training_replicates']) is int and cell['training_replicates'] == 5 and + cell['acquisition_controls'] == list(CONTROLS) and + cell['downstream_policy'] == 'unchanged acquisition requires both legacyDownstream and nonMBONDownstream positive gains', + 'Cell rule/authority/control coverage differs') + expected_ids = {key for key, b in blocks.items() if (b['mode'], b['dan_type'], b['conditioned_cue']) == identity} + rows = cell['replicates'] + exact_inventory([r['replicate_id'] for r in rows], expected_ids, 'same-cell independent replicates') + for row in rows: + block_id = row['replicate_id']; paired = terminal[(block_id, 'paired')] + s = finite(paired['rawS']) + controls = {legacy: finite(terminal[(block_id, name)]['rawS']) for name, legacy in ARMS.items() if name != 'paired'} + beyond = {name: s > value if identity[1] == 'PAM15' else s < value for name, value in controls.items()} + cue = identity[2]; neutral = next(c for c in CUES if c != cue) + conditioned = paired['cues'][cue]['populations']['MBON']; other = paired['cues'][neutral]['populations']['MBON'] + require(type(conditioned['spike_count_l1']) is int and conditioned['spike_count_l1'] >= 0 and + type(conditioned['defined_first_latency_changed']) is bool, 'Functional count/latency schema differs') + gains = {p: finite(paired['separation'][p]['gain']) for p in POPS} + predicates = dict(zip(PREDICATES, ( + s > 0 if identity[1] == 'PAM15' else s < 0, all(beyond.values()), + finite(conditioned['final_delta_g_max_abs']) > finite(other['final_delta_g_max_abs']), + conditioned['spike_count_l1'] > 0 or conditioned['defined_first_latency_changed'], + gains['MBON'] > 0, gains['legacyDownstream'] > 0, gains['nonMBONDownstream'] > 0))) + require(row['rawS'] == s and row['raw_control_S'] == controls and row['population_separation_gains'] == gains and + row['beyond_controls'] == beyond and all(type(x) is bool for x in row['beyond_controls'].values()) and + row['predicates'] == predicates and all(type(x) is bool for x in row['predicates'].values()) and + row['conjunction_pass'] is all(predicates.values()), 'Replicate differs from frozen terminal conjunction') + count = sum(r['conjunction_pass'] for r in rows); passed = count >= 3 + require(type(cell['successful_same_replicate_conjunctions']) is int and cell['successful_same_replicate_conjunctions'] == count and + cell['predicate_pass'] is passed and cell['status'] == ('CONDITIONAL-PREDICATES-PASS' if passed else 'CONDITIONAL-PREDICATES-FAIL'), + 'Cell majority/result differs from same-replicate conjunctions') + require(set(cell['legacy_control_MBON_gains']) == {'D=0', 'random reward', 'frozen'}, 'Missing H controls') + for name, legacy in (('d_zero', 'D=0'), ('random_reward', 'random reward'), ('frozen', 'frozen')): + require(cell['legacy_control_MBON_gains'][legacy] == {b: terminal[(b, name)]['separation']['MBON']['gain'] for b in expected_ids}, + 'H controls differ from terminal source curves') + table.append(dict(mode=identity[0], dan_type=identity[1], conditioned_cue=identity[2], assay='normal', + status='CONDITIONAL-PASS' if passed else 'CONDITIONAL-FAIL', successful_replicates=count)) + cell_lookup[identity] = cell + cohort = a['cohort']; scope(cohort) + normal_pass = all(c['predicate_pass'] for c in cells) + require(cohort['status'] == 'COMPLETE' and cohort['required_assays'] == ['normal'] and cohort['complete_inventory'] is True and + cohort['legacy_H_and_integrity_gates_included'] is False and cohort['all_functional_predicates_pass'] is normal_pass and + cohort['table'] == sorted(table, key=lambda r: (r['mode'], r['dan_type'], r['conditioned_cue'], r['assay'])), 'Cohort summary differs') + legacy = a['legacy_H']; scope(legacy) + require(legacy['status'] == 'CONDITIONAL-AUDIT' and legacy['declared_mapping'] == H_MAPPING and + legacy['original_H_data_layout_reproduced'] is False and legacy['new_stricter_control_source_mapping'] is True, 'H mapping changed') + exact_inventory([(g['mode'], g['conditioned_cue']) for g in legacy['groups']], {(m, c) for m in MODES for c in CUES}, 'four separate H groups') + for group in legacy['groups']: + require(set(group['control_sources']) == set(DANS), 'Both H control sources required') + for source, value in group['control_sources'].items(): + means = value['protocol_means'] + require(set(means) == {'appetitive-conditioning', 'aversive-conditioning', 'd-zero-control', 'random-reward-control', 'frozen-brain-control'}, + 'H mean labels differ') + for v in means.values(): finite(v) + # Same five-element NumPy mean as frozen metrics.py; this is a + # consistency check on the existing H mapping, not another gate. + expected_means = {} + for dan, label in (('PAM15', 'appetitive-conditioning'), ('PPL103', 'aversive-conditioning')): + rows = cell_lookup[(group['mode'], dan, group['conditioned_cue'])]['replicates'] + expected_means[label] = float(np.mean([r['population_separation_gains']['MBON'] for r in rows])) + controls = cell_lookup[(group['mode'], source, group['conditioned_cue'])] + for name, label in (('D=0', 'd-zero-control'), ('random reward', 'random-reward-control'), ('frozen', 'frozen-brain-control')): + expected_means[label] = float(np.mean([controls['legacy_control_MBON_gains'][name][r['replicate_id']] + for r in controls['replicates']])) + require(means == expected_means, 'H means differ from exact branch/replicate control sources') + floor = min(means['appetitive-conditioning'], means['aversive-conditioning']) + ceiling = max(means[n] for n in ('d-zero-control', 'random-reward-control', 'frozen-brain-control')) + require(value['schema'] == 'phase4an-metrics-draft-v1' and value['audit'] == 'legacy_H_mean_control_predicate_only' and + value['source_reference'] == 'frozen phase4h.py:412-414' and + value['AN_to_legacy_protocol_mapping_authority'] == 'caller; no automatic pooling' and + value['condition_floor'] == floor and value['control_ceiling'] == ceiling and + value['predicate_pass'] is (floor > ceiling) and value['full_old_H_gate_computed'] is False and + value['scientific_learning_pass_claimed'] is False, 'Frozen strict H scalar predicate differs') + require(group['both_pass'] is all(v['predicate_pass'] for v in group['control_sources'].values()), 'H group conjunction differs') + h_pass = all(g['both_pass'] for g in legacy['groups']) + require(legacy['both_pass'] is h_pass, 'H total conjunction differs') + return dict(status='NORMAL-ACQUISITION-PASS' if normal_pass and h_pass else 'NORMAL-ACQUISITION-FAIL', + normal_acquisition_pass=normal_pass and h_pass, all_eight_cell_conjunctions_pass=normal_pass, + both_H_control_sources_all_groups_pass=h_pass, cell_table=cohort['table'], + registration_sha256=outer['registration_sha256'], analysis_sha256=outer['analysis_sha256']) + + +def summarize_normal_analysis(report, outer_evidence): + """Classify only complete, bound normal reports; missing evidence is not FAIL.""" + base = dict(schema='normal-outcome-summary-v1', learning_pass_claimed=False, usable_learning_pass=False, + scientific_learning_pass_claimed=False, candidate_generated=False, subsequent_assays_authorized=False, + subsequent_assays={'delayed': 'NOT-RUN', 'reversal': 'NOT-RUN', 'retention': 'NOT-RUN'}, + outer_evidence_authenticated_here=False, raw_scientific_arrays_recomputed_here=False, + canonical_scope=CANONICAL_SCOPE, + scope='Registered normal acquisition only; external audit and process authentication are caller obligations') + try: + return dict(base, **_summarize(report, outer_evidence)) + except (Incomplete, KeyError) as error: + return dict(base, status='INCOMPLETE', normal_acquisition_pass=None, reason=str(error)) + except (ValueError, TypeError, AttributeError, IndexError) as error: + return dict(base, status='INTEGRITY-FAIL', normal_acquisition_pass=None, reason=str(error)) diff --git a/tools/normal-conditioning-audit/terminal_ledger.py b/tools/normal-conditioning-audit/terminal_ledger.py new file mode 100644 index 0000000..acaf5ad --- /dev/null +++ b/tools/normal-conditioning-audit/terminal_ledger.py @@ -0,0 +1,237 @@ +"""Read-only post-exit reconciliation, never a learning gate or process runner. + +The original ledger reports stay unchanged. A newly recomputed final_verify +must FAIL for exactly its one post-audit completion file. The authentic earlier +PASS report embedded in that file can then be supplied verbatim to metrics. +The caller owns actual OS quiescence checks; a missing PID is not exit evidence. +No report, manifest, ledger, lock, temporary or bytecode file is written here. +""" +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +import re +import types + +SCHEMA = 'normal-terminal-ledger-reconciliation-v1' +COMPLETION = 'audit/execution-complete.json' +SUPERVISOR_FILES = ('launch-attempt.json', 'launch.json', 'supervisor-start.json', + 'exit.json', 'stdout.log', 'stderr.log') +START_FILES = SUPERVISOR_FILES[:3] +SHA = re.compile('[0-9a-f]{64}') + + +def require(value, message): + if not value: raise ValueError(message) + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False) + + +def path(value): + result = Path(value) + require(result.is_absolute() and '..' not in result.parts and not any(p.is_symlink() or + (hasattr(p, 'is_junction') and p.is_junction()) for p in (result, *result.parents)), 'Unsafe path') + return result + + +def identity(value): + target = path(value); require(target.is_file(), 'Missing regular file: ' + str(target)) + digest = hashlib.sha256(); size = 0 + with target.open('rb') as stream: + for block in iter(lambda: stream.read(8 << 20), b''): + digest.update(block); size += len(block) + return dict(bytes=size, sha256=digest.hexdigest()) + + +def validate_identity(value): + require(type(value) is dict and set(value) == {'bytes', 'sha256'} and type(value['bytes']) is int and + value['bytes'] >= 0 and type(value['sha256']) is str and SHA.fullmatch(value['sha256']), 'Invalid external file identity') + + +def parse(raw): + def pairs(items): + result = {} + for key, value in items: + require(key not in result, 'Duplicate JSON key'); result[key] = value + return result + return json.loads(raw.decode('utf-8'), object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(ValueError('Nonfinite JSON'))) + + +class Reader: + def __init__(self): self.inventory = {} + + def bind(self, value, *, sha=None, expected=None): + target = path(value); key = str(target) + if expected is not None: validate_identity(expected) + if sha is not None: require(type(sha) is str and SHA.fullmatch(sha), 'Invalid expected SHA') + if key not in self.inventory: self.inventory[key] = identity(target) + actual = self.inventory[key] + require((expected is None or actual == expected) and (sha is None or actual['sha256'] == sha), + 'Actual file identity conflicts with authority: ' + key) + return actual + + def raw(self, value): + actual = self.bind(value); raw = path(value).read_bytes() + require(len(raw) == actual['bytes'] and hashlib.sha256(raw).hexdigest() == actual['sha256'], + 'Parsed file changed during audit') + return raw + + def json(self, value): return parse(self.raw(value)) + + +def _expected_entries(plan, reg): + require(plan.get('schema') == 'normal-conditioning-plan-v1' and + canonical(plan['inventory']) == canonical(reg['normal_inventory']), 'Registered plan inventory differs') + result = ['input-generation']; used = set(); train = probe = 0 + for task in plan['tasks']: + name = task['task_id']; require(type(name) is str and name and name not in used, 'Duplicate/invalid task ID') + used.add(name); require(task['kind'] in ('train', 'probe'), 'Unknown task kind') + key = 'task-' + hashlib.sha256(name.encode()).hexdigest() + suffixes = ('--brain', '--update', '--verify') if task['kind'] == 'train' else ('--brain',) + result.extend(key + suffix for suffix in suffixes) + train += task['kind'] == 'train'; probe += task['kind'] == 'probe' + require(plan['inventory']['train_calls'] == train and plan['inventory']['probe_calls'] == probe and + plan['inventory']['brain_calls'] == train + probe and plan['inventory']['original_E_calls'] == train and + plan['inventory']['training_cpu_validation_calls'] == train, 'Plan process-count accounting differs') + return result + + +def _supervisor(reader, root, regsha, model, start_pins): + require(type(start_pins) is dict and set(start_pins) == set(START_FILES), 'All three external startup identities required') + require(root.is_dir() and set(p.name for p in root.iterdir()) == set(SUPERVISOR_FILES), + 'Supervisor files incomplete/extra; no live/partial terminal audit') + before = {} + for name in SUPERVISOR_FILES: + before[name] = reader.bind(root / name, expected=start_pins[name] if name in start_pins else None) + attempt, launch, start, exit_record = (reader.json(root / name) for name in SUPERVISOR_FILES[:4]) + require(all(value.get('registration_sha256') == regsha for value in (attempt, launch, start, exit_record)), + 'Supervisor registration differs') + require(attempt.get('complete') is False and attempt.get('terminal_state') == 'UNKNOWN' and + attempt.get('automatic_retry') is False, 'Original launch attempt differs') + require(launch.get('launch_observed') is True and launch.get('final_exit_observed') is False and + launch.get('not_evidence_of_completion') is True and type(launch.get('supervisor_pid')) is int and + launch['supervisor_pid'] > 0 and type(start.get('pid')) is int and + launch['supervisor_pid'] == start['pid'] and start.get('complete') is False, 'Supervisor start/PID binding differs') + require(type(exit_record.get('exit_code')) is int and exit_record['exit_code'] == 0 and + exit_record.get('final_exit_observed') is True and exit_record.get('complete') is True and + exit_record.get('error') is None and exit_record.get('learning_success_claimed') is False and + exit_record.get('scientific_analysis_required') is True, 'No actual successful parent-observed controller exit') + a, b = launch.get('argv'), exit_record.get('argv') + require(type(a) is list and type(b) is list and len(a) == 6 and len(b) == 5 and + type(a[0]) is str and Path(a[0]).is_absolute() and b[0] == a[0] and + a[1:] == ['-B', str(model / 'run_normal.py'), 'supervise', '--registration-sha256', regsha] and + b[1:] == ['-B', str(model / 'normal_runner.py'), '--registration-sha256', regsha], 'Supervisor executable/argv binding differs') + return before, dict(schema='normal-supervisor-terminal-evidence-v1', parent_observed_controller_exit_code=0, + supervisor_pid=start['pid'], startup_external_pins_verified=True, file_identities=before, + current_OS_liveness_verified_here=False, external_quiescence_required=True, + launch_PID_is_not_completion_evidence=True, learning_success_claimed=False) + + +def _reconcile(recorded, current, expected, regsha, completion_bytes): + require(recorded.get('schema') == current.get('schema') == 'normal-append-ledger-v1-final-audit' and + recorded.get('registration_sha256') == current.get('registration_sha256') == regsha, + 'Final ledger schema/authority differs') + require(recorded.get('pass_all') is True and current.get('pass_all') is False and + recorded.get('entry_ids') == recorded.get('expected_entry_ids') == expected and + recorded.get('expected_order_exact') is True, 'Historical/current final-audit scope differs') + require(all(recorded.get(k) == [] for k in ('extra_files', 'missing_files', 'changed_files', 'unsafe_files', + 'errors', 'uncommitted_entry_ids')) and recorded.get('uncommitted_state_unknown') is False and + recorded.get('output_cap_respected') is True and recorded.get('free_reserve_respected') is True and + recorded.get('scientific_gate_claimed') is False and recorded.get('retry_authorized') is False and + recorded.get('checked_before_future_final_report') is True, 'Original completed ledger was not valid') + require(current.get('extra_files') == [COMPLETION], 'Only the post-ledger completion file may be extra') + delta_keys = {'pass_all', 'extra_files', 'actual_total_bytes', 'immutable_and_ledger_bytes'} + require(set(recorded) == set(current) and canonical({k:v for k,v in recorded.items() if k not in delta_keys}) == + canonical({k:v for k,v in current.items() if k not in delta_keys}), 'Non-completion final-audit field changed') + for key in ('actual_total_bytes', 'immutable_and_ledger_bytes'): + require(type(recorded[key]) is int and type(current[key]) is int and + current[key] == recorded[key] + completion_bytes, 'Completion byte reconciliation failed: ' + key) + + +def terminal_inventory(exp, regsha, model_path, *, supervisor_dir, supervisor_start_pins): + """Read all terminal evidence, return metrics arguments without writing. + + Caller must first verify actual process quiescence, independently of these + receipts. The authentic historical final_ledger is returned unchanged; the + recomputed false report remains false. This function does not run metrics, + verify all historical experiment seals, or decide biological learning. + """ + exp, model, supervisor = path(exp), path(model_path), path(supervisor_dir) + require(exp.is_dir() and model.is_dir() and supervisor.is_dir() and not supervisor.is_relative_to(exp), + 'Existing separate experiment/model/supervisor roots required') + reader = Reader(); regpath = exp / 'protocol/preregistered.json' + reader.bind(regpath, sha=regsha); reg = reader.json(regpath) + require(reg.get('schema') == 'normal-conditioning-registration-v1' and + reg.get('scientific_scope') == 'normal-acquisition-only' and reg.get('subsequent_assays_authorized') is False and + reg['administrative_supervisor_directory'] == str(supervisor), 'Wrong terminal authority/scope') + # Reject live/partial evidence before the expensive full-tree reread. + require((exp / COMPLETION).is_file() and not (exp / 'audit/stopped.json').exists() and + not (exp / '.append-ledger/writer.lock').exists(), 'Live/stopped/incomplete experiment cannot be finalized') + completion_before = reader.bind(exp / COMPLETION); completion = reader.json(exp / COMPLETION) + require(completion.get('complete') is True and completion.get('registration_sha256') == regsha and + completion.get('learning_success_claimed') is False, 'Completion authority differs') + supervisor_before, supervisor_audit = _supervisor(reader, supervisor, regsha, model, supervisor_start_pins) + for name in ('source_sha256', 'input_sha256', 'prepared_sha256'): + require(type(reg[name]) is dict and reg[name], 'Missing registered file inventory') + for p, expected in reg[name].items(): reader.bind(p, sha=expected) + for p, expected in reg['source_sha256'].items(): + require(path(p).parent == model, 'Registered model directory differs') + snapshot = exp / 'protocol/source-snapshot' / Path(p).name + require(reg['prepared_sha256'].get(str(snapshot)) == expected, 'Registered source snapshot differs') + reader.bind(snapshot, sha=expected) + planpath = exp / 'protocol/normal-plan.json' + require(str(planpath) in reg['prepared_sha256'], 'Plan lacks prepared-file authority') + plan = reader.json(planpath); expected_ids = _expected_entries(plan, reg) + require(canonical(completion['inventory']) == canonical(plan['inventory']), 'Completion inventory differs') + helper = model / 'append_ledger.py'; require(str(helper) in reg['source_sha256'], 'Unregistered ledger helper') + module = types.ModuleType('_normal_terminal_frozen_ledger'); module.__file__ = str(helper) + exec(compile(reader.raw(helper), str(helper), 'exec'), module.__dict__) + control = exp / '.append-ledger' + require((control / 'header.json').is_file() and (control / 'started').is_dir() and (control / 'entries').is_dir(), + 'Existing complete ledger required; automatic creation forbidden') + header_before = reader.bind(control / 'header.json') + initial = {} + for p, expected in reg['prepared_sha256'].items(): + require(path(p).is_relative_to(exp), 'Prepared path outside experiment') + initial[Path(p).relative_to(exp).as_posix()] = reader.bind(p, sha=expected) + initial['protocol/preregistered.json'] = reader.bind(regpath, sha=regsha) + ledger = module.Ledger(exp, regsha, initial, reg['budget']['output_cap_bytes'], reg['budget']['minimum_free_reserve_bytes']) + current = ledger.final_verify(expected_ids) + recorded = completion['ledger'] + _reconcile(recorded, current, expected_ids, regsha, completion_before['bytes']) + expected_tree = {**ledger._tracked, **ledger._controls, **current['shared_mutable_build_inventory'], + COMPLETION: completion_before} + tree_files = set() + for p in exp.rglob('*'): + path(p) + if p.is_dir(): continue + require(p.is_file(), 'Nonregular experiment artifact') + rel = p.relative_to(exp).as_posix(); tree_files.add(rel) + require(rel in expected_tree, 'Unexpected terminal experiment file: ' + rel) + reader.bind(p, expected=expected_tree[rel]) + require(tree_files == set(expected_tree), 'Missing terminal experiment file') + for number in range(1, len(expected_ids) + 1): + entry = reader.json(control / f'entries/{number:016d}.json') + for row in entry['consumed_inputs'] + entry['new_outputs']: + reader.bind(row['path'], expected={k:row[k] for k in ('bytes', 'sha256')}) + # Fresh small authority reads bracket all heavy reads; no cross-call cache. + require(identity(exp / COMPLETION) == completion_before and identity(control / 'header.json') == header_before and + identity(regpath) == reader.inventory[str(regpath)], 'Terminal authority changed during audit') + ledger._check_tail() + require(not (control / 'writer.lock').exists() and not (exp / 'audit/stopped.json').exists(), 'Writer/failure appeared during audit') + require(set(p.name for p in supervisor.iterdir()) == set(SUPERVISOR_FILES) and + {name:identity(supervisor/name) for name in SUPERVISOR_FILES} == supervisor_before, 'Supervisor files changed during audit') + return dict(schema=SCHEMA, status='PASS-TERMINAL-LEDGER-RECONCILIATION-NOT-LEARNING', + terminal_reconciliation_pass=True, registration_sha256=regsha, + recorded_final_ledger=recorded, current_final_ledger=current, final_ledger=recorded, + verified_inventory=dict(sorted(reader.inventory.items())), supervisor_audit=supervisor_audit, + completion_file=dict(path=str(exp/COMPLETION), **completion_before), + preserved_report_semantics='Historical true report remains true; recomputed false report remains false; only one post-audit completion artifact reconciled', + historical_seals_verified_here=False, metrics_executed=False, scientific_learning_pass_claimed=False, + candidate_generated=False, external_OS_quiescence_required=True, + read_scope='Actual file SHA/bytes; invocation-local cache only. Frozen Ledger reopen/final_verify additionally rehash. Not an atomic snapshot.', + files_written=False) diff --git a/tools/normal-conditioning-audit/test_normal_finalize.py b/tools/normal-conditioning-audit/test_normal_finalize.py new file mode 100644 index 0000000..f54a9d7 --- /dev/null +++ b/tools/normal-conditioning-audit/test_normal_finalize.py @@ -0,0 +1,121 @@ +"""Synthetic-only terminal guard tests, no science, training or final metrics.""" +import copy +import json +import shlex +import sys +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch +import normal_finalize as F + +class FinalizeTests(unittest.TestCase): + def setUp(self): + self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup) + self.root=Path(self.temp.name).resolve();self.exp=self.root/'exp';self.control=self.root/'control';self.model=self.root/'model' + for p in (self.exp/'protocol',self.exp/'audit',self.control,self.model):p.mkdir(parents=True) + self.reg=dict(schema='normal-conditioning-registration-v1',scientific_scope='normal-acquisition-only', + subsequent_assays_authorized=False,normal_inventory={'fixture':True}) + F.save(self.exp/'protocol/preregistered.json',self.reg) + self.sha=F.identity(self.exp/'protocol/preregistered.json')['sha256'] + for name in ('launch-attempt.json','launch.json','supervisor-start.json'): + F.save(self.control/name,dict(synthetic=True,registration_sha256=self.sha)) + self.pins={p.name:F.identity(p) for p in self.control.iterdir()} + self.kwargs=dict(exp=self.exp,control=self.control,model=self.model,regsha=self.sha,start_pins=self.pins, + census=lambda:dict(actual_OS_query=True,matching_live_processes=[])) + + def terminal(self): + F.save(self.control/'exit.json',dict(registration_sha256=self.sha, + argv=[str(F.PYTHON),'-B',str(self.model/'normal_runner.py'),'--registration-sha256',self.sha], + exit_code=0,final_exit_observed=True,complete=True,error=None, + learning_success_claimed=False,scientific_analysis_required=True)) + F.save(self.exp/'audit/execution-complete.json',dict(complete=True,registration_sha256=self.sha, + inventory=self.reg['normal_inventory'],learning_success_claimed=False)) + + def rewrite_fixture(self,path,update): + value=F.read(path);value.update(update) + # Test-owned file corruption, not application editing/retry behavior. + path.write_text(json.dumps(value),encoding='utf-8') + + def test_actual_live_process_refuses_even_with_complete_files_and_no_writes(self): + self.terminal();before=sorted(str(p) for p in self.root.rglob('*')) + kwargs={**self.kwargs,'census':lambda:dict(actual_OS_query=True,matching_live_processes=[{'pid':123}])} + with self.assertRaisesRegex(ValueError,'still live'):F.readiness(**kwargs) + self.assertEqual(before,sorted(str(p) for p in self.root.rglob('*'))) + + def test_missing_process_without_real_exit_does_not_authorize_terminal_audit(self): + with self.assertRaisesRegex(ValueError,'missing process alone'):F.readiness(**self.kwargs) + + def test_success_only_means_ready_not_learning_and_does_not_hash_brain_outputs(self): + self.terminal();F.save(self.exp/'unrelated-heavy-placeholder.json',{'not_read':True}) + checked=[];original=F.identity + def identity(path):checked.append(str(path));return original(path) + with patch.object(F,'identity',side_effect=identity):result=F.readiness(**self.kwargs) + self.assertTrue(result['ready']);self.assertFalse(result['scientific_success_inferred']) + self.assertNotIn(str(self.exp/'unrelated-heavy-placeholder.json'),checked) + + def test_changed_startup_binding_is_rejected(self): + self.terminal();self.rewrite_fixture(self.control/'launch.json',{'tampered':True}) + with self.assertRaisesRegex(ValueError,'Startup'):F.readiness(**self.kwargs) + + def test_nonzero_parent_exit_is_not_scientific_failure_or_success(self): + self.terminal();self.rewrite_fixture(self.control/'exit.json',{'exit_code':1,'complete':False}) + with self.assertRaisesRegex(ValueError,'successful run exit'):F.readiness(**self.kwargs) + + def test_completion_inventory_or_registration_mismatch_rejected(self): + self.terminal();self.rewrite_fixture(self.exp/'audit/execution-complete.json',{'inventory':{'shortened':True}}) + with self.assertRaisesRegex(ValueError,'completion binding'):F.readiness(**self.kwargs) + + def test_stopped_record_always_refuses_completion(self): + self.terminal();F.save(self.exp/'audit/stopped.json',dict(complete=False)) + with self.assertRaisesRegex(ValueError,'Stopped run'):F.readiness(**self.kwargs) + + def test_actual_process_query_required_not_caller_assumption(self): + self.terminal();kw={**self.kwargs,'census':lambda:dict(matching_live_processes=[])} + with self.assertRaisesRegex(ValueError,'Actual process'):F.readiness(**kw) + + def test_argv_classifier_ignores_grep_substrings_and_includes_real_children(self): + worker=self.model.parent/'conditioning-interface-v2/isolated_workers.py' + validator=self.model.parent/'conditioning-execution-v1/validate_update_cpu.py' + def row(pid,*argv):return str(pid)+' '+shlex.join(argv) + rows='\n'.join([ + row(10,'python','-B',str(self.model/'normal_runner.py'),'--registration-sha256','x'), + row(11,'python','-B',str(worker),'brain','--job',str(self.exp/'jobs/x.json')), + row(12,'python','-B',str(validator),'--authority',str(self.exp/'jobs/x.json')), + row(13,'bash','-c','grep '+str(self.model/'normal_runner.py')), + row(14,'python','-B',str(worker),'brain','--job','/different/exp/jobs/x.json'), + row(15,'python','-B',str(self.model/'normal_inputs.py'),'--registration-sha256','x')]) + self.assertEqual([r['pid'] for r in F.live_runtime_rows(rows,self.model,self.exp)],[10,11,12,15]) + + def test_exclusive_save_never_overwrites(self): + path=self.root/'owned.json';F.save(path,{'first':True});before=path.read_bytes() + with self.assertRaises(FileExistsError):F.save(path,{'second':True}) + self.assertEqual(before,path.read_bytes()) + + def test_real_harmless_CPU_child_exit_and_log_hashes_are_saved(self): + out=self.root/'new-audit';out.mkdir() + with patch.object(F,'OUT',out): + receipt=F.child('fixture',[sys.executable,'-B','-c','print("synthetic CPU fixture only")'],10) + self.assertEqual(receipt['exit_code'],0);self.assertTrue(receipt['actual_parent_observation']) + self.assertEqual(receipt['stdout'],F.identity(out/'fixture.stdout.log')) + self.assertIn('synthetic CPU fixture only',(out/'fixture.stdout.log').read_text()) + + def test_failed_CPU_child_preserves_exit_and_refuses_same_name_retry(self): + out=self.root/'failed-audit';out.mkdir() + with patch.object(F,'OUT',out): + with self.assertRaisesRegex(ValueError,'Audit child failed'): + F.child('fixture',[sys.executable,'-B','-c','raise SystemExit(7)'],10) + receipt=F.read(out/'fixture-exit.json') + self.assertEqual(receipt['exit_code'],7);self.assertFalse(receipt['complete']) + with self.assertRaises(FileExistsError):F.child('fixture',[sys.executable,'-B','-c','pass'],10) + + def test_finalization_checks_live_guard_before_source_or_output_changes(self): + out=self.root/'should-stay-empty';out.mkdir() + with patch.object(F,'PYTHON',Path(sys.executable)),patch.object(F,'OUT',out),\ + patch.object(F,'readiness',side_effect=ValueError('still live')),\ + patch.object(F,'source_identity') as source: + with self.assertRaisesRegex(ValueError,'still live'):F.finalize() + source.assert_not_called() + self.assertEqual(list(out.iterdir()),[]) + +if __name__=='__main__':unittest.main() diff --git a/tools/normal-conditioning-audit/test_normal_outcome.py b/tools/normal-conditioning-audit/test_normal_outcome.py new file mode 100644 index 0000000..09621ba --- /dev/null +++ b/tools/normal-conditioning-audit/test_normal_outcome.py @@ -0,0 +1,176 @@ +"""Synthetic six-neuron CPU fixtures only; NO experiment states or outcomes. + +The fixture is generated by SHA-pinned real frozen G/M through normal_metrics, +not by a second hand-written imitation of its predicates or report schema. +""" +import copy +import hashlib +import importlib.util +import json +from pathlib import Path +import unittest + +import normal_outcome as O + + +def load_fixture_module(): + pins = { + 'normal_metrics.py': O.ANALYSIS_SOURCE_SHA256, + 'test_normal_metrics.py': '204586b30edbe57bb294c1d8f8ad309863e5bcb3baad32118955ac3d4adcdefb', + } + candidates = (Path(__file__).resolve().parent.parent / 'normal-conditioning-v1', + Path('/data/digital-drosophila/models/male-cns-lif/normal-conditioning-v1')) + source = next((p for p in candidates if all((p / n).is_file() and + hashlib.sha256((p / n).read_bytes()).hexdigest() == sha for n, sha in pins.items())), None) + if source is None: + raise AssertionError('No exact pinned frozen synthetic fixture source at either declared location') + spec = importlib.util.spec_from_file_location('_outcome_frozen_fixture', source / 'test_normal_metrics.py') + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def synthetic_report(fixture_module, scenario='pass'): + F = fixture_module; N = F.N + nodes, pops, registry, authority, probes, arms = F.fixture() + first = next(iter(registry['blocks'])) + target = registry['blocks'][first] + # Only one cell changes; all intermediate successes remain available but + # must not rescue a failing terminal checkpoint. + for (block, arm, point, cue), record in probes.items(): + b = registry['blocks'][block] + same = all(b[k] == target[k] for k in ('mode', 'dan_type', 'conditioned_cue')) + if scenario == 'terminal_fail' and same and b['replicate'] == 3 and point == 50: + record['active'] = False + if scenario in ('h_fail', 'h_equal') and same and arm == 'd_zero' and point > 0: + record['active'] = N.ODORS[cue] != b['conditioned_cue'] and ( + scenario == 'h_fail' or b['replicate'] <= 3) + scientific = N.scientific_reports(probes, arms, registry=registry, authority=authority, + nodes=nodes, populations=pops, load_state=F.state_for) + scientific['legacy_H'] = N.legacy_audit(scientific['cells'], N.G.H_MAPPING) + return dict(schema='normal-scientific-analysis-v1', status='CONDITIONAL-PENDING-OUTER-INTEGRITY', + registration_sha256=registry['protocol_sha256'], normal_plan_sha256='5' * 64, + registry=registry, authority=authority, analysis=scientific, + analyzed_probe_calls=3920, curve_points=1960, consumed_file_count=123, + subsequent_assays_status='NOT-RUN', subsequent_assays_authorized=False, + candidate_generated=False, learning_pass_claimed=False, scientific_learning_pass_claimed=False, + data_backed_claimed=False, external_full_tree_anatomy_historical_integrity_verified_here=False, + retained_raw_evidence='synthetic arrays generated in memory; no actual experiment read') + + +def binding(report): + # This is simulated external evidence, not an actual audit or authorization. + return dict(registration_sha256=report['registration_sha256'], terminal_reconciliation_pass=True, + historical_integrity_after_pass=True, analysis_sha256=O.analysis_digest(report), + actual_parent_exit_code=0, analyzed_source_sha256=O.ANALYSIS_SOURCE_SHA256, + analysis_file_sha256='a' * 64) + + +class NormalOutcomeTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + F = load_fixture_module() + cls.reports = {name: synthetic_report(F, name) for name in ('pass', 'terminal_fail', 'h_fail', 'h_equal')} + + def outcome(self, report, evidence=None): + return O.summarize_normal_analysis(report, binding(report) if evidence is None else evidence) + + def test_pinned_real_frozen_gates_compact_output_pass_and_no_mutation(self): + report = self.reports['pass']; before = O.analysis_digest(report) + actual = self.outcome(report) + self.assertEqual(actual['status'], 'NORMAL-ACQUISITION-PASS', actual) + self.assertTrue(actual['normal_acquisition_pass']) + self.assertEqual(len(actual['cell_table']), 8) + self.assertTrue(all(r['successful_replicates'] == 3 for r in actual['cell_table'])) + self.assertEqual(len(report['analysis']['curves']), 1960) + self.assertEqual(before, O.analysis_digest(report)) + for key in ('learning_pass_claimed', 'usable_learning_pass', 'candidate_generated', + 'scientific_learning_pass_claimed', 'subsequent_assays_authorized', + 'outer_evidence_authenticated_here', 'raw_scientific_arrays_recomputed_here'): + self.assertIs(actual[key], False) + self.assertEqual(set(actual['subsequent_assays'].values()), {'NOT-RUN'}) + + def test_terminal_same_replicate_fail_cannot_use_intermediate_or_other_cell(self): + report = self.reports['terminal_fail']; actual = self.outcome(report) + self.assertEqual(actual['status'], 'NORMAL-ACQUISITION-FAIL', actual) + self.assertFalse(actual['all_eight_cell_conjunctions_pass']) + self.assertTrue(actual['both_H_control_sources_all_groups_pass']) + self.assertEqual(sorted(r['successful_replicates'] for r in actual['cell_table']), [2] + [3] * 7) + + def test_H_one_control_branch_failure_and_equality_are_strict(self): + for name in ('h_fail', 'h_equal'): + with self.subTest(name=name): + report = self.reports[name]; actual = self.outcome(report) + self.assertEqual(actual['status'], 'NORMAL-ACQUISITION-FAIL', actual) + self.assertTrue(actual['all_eight_cell_conjunctions_pass']) + self.assertFalse(actual['both_H_control_sources_all_groups_pass']) + audits = [v for g in report['analysis']['legacy_H']['groups'] for v in g['control_sources'].values()] + self.assertEqual(sum(not v['predicate_pass'] for v in audits), 1) + if name == 'h_equal': + bad = next(v for v in audits if not v['predicate_pass']) + self.assertEqual(bad['condition_floor'], bad['control_ceiling']) + + def test_missing_outer_and_failed_analysis_are_incomplete_not_scientific_fail(self): + report = self.reports['pass'] + for key in O.OUTER_KEYS: + value = binding(report); del value[key] + actual = self.outcome(report, value) + self.assertEqual(actual['status'], 'INCOMPLETE', (key, actual)) + self.assertIsNone(actual['normal_acquisition_pass']) + value = binding(report); value['actual_parent_exit_code'] = 9 + self.assertEqual(self.outcome(report, value)['status'], 'INCOMPLETE') + + def test_actual_source_digest_registration_and_integrity_fail_closed(self): + report = self.reports['pass'] + for key, wrong in ( + ('terminal_reconciliation_pass', False), ('historical_integrity_after_pass', False), + ('terminal_reconciliation_pass', 1), ('analyzed_source_sha256', '0' * 64), + ('analysis_sha256', '0' * 64), ('registration_sha256', '0' * 64), + ('actual_parent_exit_code', False), + ): + value = binding(report); value[key] = wrong + actual = self.outcome(report, value) + self.assertEqual(actual['status'], 'INTEGRITY-FAIL', (key, actual)) + self.assertIsNone(actual['normal_acquisition_pass']) + + def test_exact_cell_replicate_curve_inventory_missing_not_scientific_negative(self): + for collection in ('cells', 'curves'): + report = copy.deepcopy(self.reports['pass']); report['analysis'][collection].pop() + self.assertEqual(self.outcome(report)['status'], 'INCOMPLETE') + report = copy.deepcopy(self.reports['pass']) + report['analysis'][collection][-1] = report['analysis'][collection][0] + self.assertEqual(self.outcome(report)['status'], 'INTEGRITY-FAIL') + report = copy.deepcopy(self.reports['pass']) + report['analysis']['cells'][0]['replicates'][0]['replicate_id'] = report['analysis']['cells'][1]['replicates'][0]['replicate_id'] + self.assertEqual(self.outcome(report)['status'], 'INTEGRITY-FAIL') + report = copy.deepcopy(self.reports['pass']); del report['analysis']['curves'][0]['metrics'] + self.assertEqual(self.outcome(report)['status'], 'INCOMPLETE') + + def test_rehashed_top_boolean_and_predicate_mutations_cannot_forge_success(self): + for alter in ( + lambda a: a['cohort'].__setitem__('all_functional_predicates_pass', True), + lambda a: a['cells'][0].__setitem__('predicate_pass', True), + lambda a: a['cells'][0]['replicates'][2].__setitem__('conjunction_pass', True), + lambda a: a['cells'][0]['replicates'][2]['predicates'].__setitem__('raw_signed_direction', True), + ): + report = copy.deepcopy(self.reports['terminal_fail']); alter(report['analysis']) + self.assertEqual(self.outcome(report)['status'], 'INTEGRITY-FAIL') + report = copy.deepcopy(self.reports['h_fail']) + report['analysis']['legacy_H']['both_pass'] = True + self.assertEqual(self.outcome(report)['status'], 'INTEGRITY-FAIL') + report = copy.deepcopy(self.reports['pass']) + report['analysis']['legacy_H']['groups'][0]['control_sources'].pop('PAM15') + self.assertEqual(self.outcome(report)['status'], 'INTEGRITY-FAIL') + + def test_canonical_scope_unicode_no_newline_and_nonfinite_rejected(self): + value = {'z': '\u89c6\u89c9', 'a': [1, 2.5]} + raw = json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False).encode('utf-8') + self.assertEqual(O.analysis_digest(value), hashlib.sha256(raw).hexdigest()) + self.assertNotEqual(O.analysis_digest(value), hashlib.sha256(raw + b'\n').hexdigest()) + with self.assertRaises(ValueError): O.analysis_digest({'nonfinite': float('nan')}) + report = copy.deepcopy(self.reports['pass']); report['curve_points'] = 1959 + self.assertEqual(self.outcome(report)['status'], 'INCOMPLETE') + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/normal-conditioning-audit/test_terminal_ledger.py b/tools/normal-conditioning-audit/test_terminal_ledger.py new file mode 100644 index 0000000..71f55f6 --- /dev/null +++ b/tools/normal-conditioning-audit/test_terminal_ledger.py @@ -0,0 +1,228 @@ +"""Temporary-file accounting tests only; no experiment, runtime or metrics run.""" +import ast +import copy +import hashlib +import json +from pathlib import Path +import shutil +import sys +import tempfile +import types +import unittest +from unittest import mock + +import terminal_ledger as T + +FROZEN_SHA = '1f981f61cd26950e7d4b5393be70b82753ea7e721c0bb75d05f8fd1e92a453fb' +FROZEN_CANDIDATES = ( + Path(__file__).resolve().parents[1] / 'normal-conditioning-v1/append_ledger.py', + Path('/data/digital-drosophila/models/male-cns-lif/normal-conditioning-v1/append_ledger.py'), +) +FROZEN = next((p for p in FROZEN_CANDIDATES if p.is_file() and + hashlib.sha256(p.read_bytes()).hexdigest() == FROZEN_SHA), None) +if FROZEN is None: + raise RuntimeError('No explicitly supported frozen ledger location has the required SHA-256') + + +def write_json(p, value): + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes((T.canonical(value) + '\n').encode()) + + +def all_hashes(root): + return {str(p.relative_to(root)): T.identity(p) for p in root.rglob('*') if p.is_file()} + + +class Fixture: + """Two actual successful toy ledger entries, not scientific output.""" + def __init__(self, root): + self.root = root + self.exp, self.model, self.supervisor = (root / name for name in ('exp', 'model', 'supervisor')) + for p in (self.exp, self.model, self.supervisor): p.mkdir() + shutil.copyfile(FROZEN, self.model / 'append_ledger.py') + for name in ('run_normal.py', 'normal_runner.py'): + (self.model / name).write_text('# synthetic non-executable source\n', encoding='utf-8') + self.external = root / 'external.dat'; self.external.write_bytes(b'read-only synthetic input') + self.source = {str(p): T.identity(p)['sha256'] for p in self.model.iterdir()} + snapshot = self.exp / 'protocol/source-snapshot'; snapshot.mkdir(parents=True) + for p in self.model.iterdir(): shutil.copyfile(p, snapshot / p.name) + inventory = dict(train_calls=0, probe_calls=1, brain_calls=1, original_E_calls=0, + training_cpu_validation_calls=0) + plan = dict(schema='normal-conditioning-plan-v1', inventory=inventory, + tasks=[dict(task_id='synthetic/probe', kind='probe')]) + write_json(self.exp / 'protocol/normal-plan.json', plan) + prepared = {str(p): T.identity(p)['sha256'] for p in (self.exp / 'protocol').rglob('*') if p.is_file()} + self.reg = dict(schema='normal-conditioning-registration-v1', scientific_scope='normal-acquisition-only', + subsequent_assays_authorized=False, administrative_supervisor_directory=str(self.supervisor), + source_sha256=self.source, input_sha256={str(self.external): T.identity(self.external)['sha256']}, + prepared_sha256=prepared, normal_inventory=inventory, + budget=dict(output_cap_bytes=16 << 20, minimum_free_reserve_bytes=0)) + regpath = self.exp / 'protocol/preregistered.json'; write_json(regpath, self.reg) + self.sha = T.identity(regpath)['sha256'] + initial = {Path(p).relative_to(self.exp).as_posix(): T.identity(p) for p in prepared} + initial['protocol/preregistered.json'] = T.identity(regpath) + ledger_module = types.ModuleType('_toy_frozen_ledger') + exec(compile(FROZEN.read_bytes(), str(FROZEN), 'exec'), ledger_module.__dict__) + self.ledger = ledger_module.Ledger(self.exp, self.sha, initial, 16 << 20, 0) + self.ids = T._expected_entries(plan, self.reg) + for n, name in enumerate(self.ids): + job = self.exp / 'jobs' / (name + '.json') + write_json(job, dict(registration_sha256=self.sha, toy=True, entry_id=name)) + self.ledger.begin(name, job, dict(scope='synthetic receipt')) + receipt = self.exp / 'receipts' / (name + '.json') + write_json(receipt, dict(entry_id=name, registration_sha256=self.sha, + job_sha256=T.identity(job)['sha256'], complete=True, exit_code=0, + timed_out=False, exit_observed_by_parent=True)) + output = self.exp / ('payload-' + str(n) + '.dat'); output.write_bytes(b'known payload') + self.ledger.append(name, {str(p): T.identity(p) for p in (regpath, job, self.external)}, + {str(p): T.identity(p) for p in (job, receipt, output)}, + dict(job_file=str(job), process_receipt_file=str(receipt), + scope='synthetic accounting only', observed_returncode=0)) + build = self.exp / 'extension-build'; build.mkdir(); (build / 'mock.bin').write_bytes(b'build') + self.original = self.ledger.final_verify(self.ids) + assert self.original['pass_all'] is True + self.completion = dict(complete=True, registration_sha256=self.sha, inventory=inventory, + ledger=self.original, learning_success_claimed=False) + write_json(self.exp / T.COMPLETION, self.completion) + self.supervisor_values = { + 'launch-attempt.json': dict(registration_sha256=self.sha, complete=False, + terminal_state='UNKNOWN', automatic_retry=False), + 'launch.json': dict(registration_sha256=self.sha, supervisor_pid=123, + launch_observed=True, final_exit_observed=False, not_evidence_of_completion=True, + argv=[sys.executable, '-B', str(self.model/'run_normal.py'), 'supervise', '--registration-sha256', self.sha]), + 'supervisor-start.json': dict(registration_sha256=self.sha, pid=123, complete=False), + 'exit.json': dict(registration_sha256=self.sha, complete=True, exit_code=0, + final_exit_observed=True, error=None, learning_success_claimed=False, + scientific_analysis_required=True, + argv=[sys.executable, '-B', str(self.model/'normal_runner.py'), '--registration-sha256', self.sha]), + } + for name, value in self.supervisor_values.items(): write_json(self.supervisor/name, value) + for name in ('stdout.log', 'stderr.log'): (self.supervisor/name).write_bytes(b'synthetic terminal log\n') + self.start_pins = {name: T.identity(self.supervisor/name) for name in T.START_FILES} + + def run(self): + return T.terminal_inventory(self.exp, self.sha, self.model, supervisor_dir=self.supervisor, + supervisor_start_pins=self.start_pins) + + +class TerminalLedgerTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name).resolve() + self.f = Fixture(self.root) + + def tearDown(self): self.temporary.cleanup() + + def test_authentic_reports_preserved_and_exact_tree_inventory_read_only(self): + before = all_hashes(self.root); value = self.f.run() + self.assertEqual(before, all_hashes(self.root)) + self.assertTrue(value['terminal_reconciliation_pass']) + self.assertEqual(value['recorded_final_ledger'], self.f.original) + self.assertEqual(value['final_ledger'], self.f.original) + self.assertFalse(value['current_final_ledger']['pass_all']) + self.assertEqual(value['current_final_ledger']['extra_files'], [T.COMPLETION]) + for key in ('actual_total_bytes', 'immutable_and_ledger_bytes'): + self.assertEqual(value['current_final_ledger'][key], + self.f.original[key] + T.identity(self.f.exp/T.COMPLETION)['bytes']) + for p in self.root.rglob('*'): + if p.is_file(): self.assertEqual(value['verified_inventory'][str(p)], T.identity(p)) + for key in ('historical_seals_verified_here', 'metrics_executed', 'scientific_learning_pass_claimed', + 'candidate_generated', 'files_written'): + self.assertIs(value[key], False) + self.assertIs(value['external_OS_quiescence_required'], True) + + def test_schema_and_filesystem_task_key_match_actual_frozen_source(self): + model = FROZEN.parent + plan = ast.parse((model/'normal_plan.py').read_bytes()) + schema = next(ast.literal_eval(node.value) for node in plan.body if isinstance(node, ast.Assign) + and any(isinstance(target, ast.Name) and target.id == 'SCHEMA' for target in node.targets)) + self.assertEqual(schema, 'normal-conditioning-plan-v1') + source = ast.parse((model/'normal_io.py').read_bytes()) + function = next(node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == 'task_key') + namespace = dict(hashlib=hashlib) + exec(compile(ast.Module(body=[function], type_ignores=[]), '', 'exec'), namespace) + key = namespace['task_key'](dict(task_id='synthetic/probe')) + self.assertEqual(self.f.ids, ['input-generation', key+'--brain']) + + def test_missing_completion_rejects_without_creating_controls(self): + (self.f.exp/T.COMPLETION).unlink() + shutil.rmtree(self.f.exp/'.append-ledger') + before = all_hashes(self.root) + with self.assertRaises(ValueError): self.f.run() + self.assertEqual(before, all_hashes(self.root)) + self.assertFalse((self.f.exp/'.append-ledger').exists()) + + def test_missing_header_cannot_auto_initialize(self): + shutil.rmtree(self.f.exp/'.append-ledger') + before = all_hashes(self.root) + with self.assertRaisesRegex(ValueError, 'automatic creation forbidden'): self.f.run() + self.assertEqual(before, all_hashes(self.root)) + + def test_missing_exit_or_live_lock_refused(self): + (self.f.supervisor/'exit.json').unlink() + with self.assertRaisesRegex(ValueError, 'incomplete/extra'): self.f.run() + write_json(self.f.supervisor/'exit.json', self.f.supervisor_values['exit.json']) + (self.f.exp/'.append-ledger/writer.lock').write_bytes(b'live') + with self.assertRaisesRegex(ValueError, 'Live/stopped/incomplete'): self.f.run() + + def test_external_startup_pin_required_and_tamper_rejected(self): + pins = self.f.start_pins + self.f.start_pins = dict(pins); self.f.start_pins.pop('launch.json') + with self.assertRaisesRegex(ValueError, 'three external'): self.f.run() + self.f.start_pins = pins + target = self.f.supervisor/'launch.json'; target.write_bytes(target.read_bytes()+b' ') + with self.assertRaisesRegex(ValueError, 'conflicts with authority'): self.f.run() + + def test_exit_code_bool_or_failed_or_wrong_command_refused(self): + for code in (False, None, 1): + record = dict(self.f.supervisor_values['exit.json'], exit_code=code) + write_json(self.f.supervisor/'exit.json', record) + with self.assertRaisesRegex(ValueError, 'parent-observed'): self.f.run() + record = copy.deepcopy(self.f.supervisor_values['exit.json']); record['argv'][2] = str(self.f.model/'run_normal.py') + write_json(self.f.supervisor/'exit.json', record) + with self.assertRaisesRegex(ValueError, 'argv binding'): self.f.run() + + def test_extra_file_and_changed_build_rejected(self): + extra = self.f.exp/'unregistered.dat'; extra.write_bytes(b'x') + with self.assertRaisesRegex(ValueError, 'Only the post-ledger'): self.f.run() + extra.unlink(); (self.f.exp/'extension-build/mock.bin').write_bytes(b'BUILD') + with self.assertRaisesRegex(ValueError, 'Non-completion'): self.f.run() + + def test_historical_payload_and_registered_source_mutation_rejected(self): + target = self.f.exp/'payload-0.dat'; original = target.read_bytes(); target.write_bytes(b'x'*len(original)) + with self.assertRaises(ValueError): self.f.run() + target.write_bytes(original) + (self.f.model/'normal_runner.py').write_bytes(b'changed source') + with self.assertRaisesRegex(ValueError, 'conflicts with authority'): self.f.run() + + def test_historical_pass_and_counters_cannot_be_rewritten(self): + for key, replacement in (('pass_all', False), ('actual_total_bytes', 0), ('head_sha256', '0'*64)): + completion = copy.deepcopy(self.f.completion); completion['ledger'][key] = replacement + write_json(self.f.exp/T.COMPLETION, completion) + with self.assertRaises(ValueError): self.f.run() + self.assertEqual(self.f.original, self.f.completion['ledger']) + + def test_supervisor_log_change_during_read_rejected_at_end(self): + real = T.identity; log = self.f.supervisor/'stdout.log'; seen = 0 + def changing(p): + nonlocal seen + result = real(p) + if Path(p) == log: + seen += 1 + if seen == 1: log.write_bytes(b'written after initial audit read') + return result + with mock.patch.object(T, 'identity', side_effect=changing): + with self.assertRaisesRegex(ValueError, 'Supervisor files changed'): self.f.run() + + def test_pending_start_or_stopped_marker_not_a_terminal_result(self): + name = 'unexpected-start'; job = self.f.exp/'jobs/unexpected.json'; write_json(job, {'toy': True}) + self.f.ledger.begin(name, job) + with self.assertRaises(ValueError): self.f.run() + + def test_duplicate_json_keys_rejected(self): + target = self.f.supervisor/'exit.json'; raw = target.read_text(encoding='utf-8') + target.write_text(raw[:-2] + ',"complete":true}\n', encoding='utf-8') + with self.assertRaisesRegex(ValueError, 'Duplicate JSON key'): self.f.run() + + +if __name__ == '__main__': unittest.main()