digital-drosophila/tools/normal-conditioning-audit/terminal_ledger.py

"""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)