LightNest / experiments /dependency_diagnostics.py

Integrate latest architecture breakthrough, harder tasks, and evaluation suite

e64c5d2 verified 14 days ago

20.6 kB

"""Predeclared public-only read-disable diagnostic for terminal slot models.
This does not train, select, promote, or inspect sealed cases. It compares normal
and zero_read inference for all 128 committed public development episodes. The
old scorer supplies full action audits; ordinary first-input forwards also
record learned allocation distributions and latent states. Read-head variation
is computed separately per head and across episode-mean distributions.
Only terminal slot8/slot16 models for the two declared seeds are accepted. The
same loaded tensors serve both modes. Disk hashes, source hashes and in-memory
state fingerprints are checked before, between and after inference.
"""
from __future__ import annotations

import argparse
from collections import Counter, defaultdict
from dataclasses import asdict
from datetime import datetime, timezone
import gzip
import hashlib
import io
import json
from pathlib import Path
import resource
import sys
import time

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import torch
from experiments.architecture_eval import score_episode, summarize
from experiments.morphology_data import summarize_scored
from lightnest.slot_patch import SlotPredictor, load_slot, state_fingerprint

SCHEMA = 'lightnest-public-slot-read-disable-v1'
PUBLIC_DATA = ROOT / 'work/morphology-20261017/data/development-episodes.json'
PUBLIC_PROTOCOL = ROOT / 'work/morphology-20261017/data/protocol.json'
EXPECTED_DATA_SHA256 = '819034832d42fca7638c9c4b649af06b8f077fb91971dd640b8516392d03e4cb'
SEEDS = (104729, 130363)
ARMS = {'slot8': 8, 'slot16': 16}
MODES = ('normal', 'zero_read')
SOURCE_FILES = (
    'experiments/dependency_diagnostics.py', 'experiments/architecture_eval.py',
    'experiments/morphology_data.py', 'lightnest/slot_patch.py',
    'lightnest/relation_patch.py', 'lightnest/iterative_patch.py', 'lightnest/rv32.py',
)


def sha(path):
    digest = hashlib.sha256()
    with Path(path).open('rb') as source:
        for block in iter(lambda: source.read(1024 * 1024), b''):
            digest.update(block)
    return digest.hexdigest()


def resolved_public_path(path):
    path = Path(path)
    path = (path if path.is_absolute() else ROOT / path).resolve()
    if not path.is_relative_to(ROOT) or any(part.lower() in ('sealed', 'evaluation-final', 'final-test')
                                          for part in path.parts):
        raise ValueError('diagnostic inputs must be public files inside the repository')
    if not path.is_file():
        raise ValueError('missing public input: ' + str(path))
    return path


def file_record(path):
    path = resolved_public_path(path)
    return dict(path=str(path.relative_to(ROOT)), sha256=sha(path), bytes=path.stat().st_size)


def check_record(record):
    path = resolved_public_path(record['path'])
    if path.stat().st_size != record['bytes'] or sha(path) != record['sha256']:
        raise ValueError('frozen public diagnostic input changed: ' + record['path'])
    return path


def write_json(path, data):
    with Path(path).open('x', encoding='utf-8') as output:
        json.dump(data, output, ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False)
        output.write('\n')


def entropy(weights):
    """Entropy on the last axis; use float64 when aggregating recorded weights."""
    return -(weights * weights.clamp_min(1e-300).log()).sum(-1)


class AllocationAccumulator:
    """One normal, initial-state forward per episode; no labels enter this object."""
    def __init__(self):
        self.entries = []

    def add(self, diagnostic):
        valid = diagnostic['valid']
        if valid.shape[0] != 1:
            raise ValueError('allocation collection expects one initial episode at a time')
        reads = diagnostic['read_weights'][0, valid[0]].detach().double().cpu()
        writes = diagnostic['write_weights'][0, valid[0]].detach().double().cpu()
        strength = diagnostic['write_strength'][0, valid[0]].detach().double().cpu()
        bank = diagnostic['slot_state'][0].detach().double().cpu()
        if not reads.numel() or reads.shape[1] != 2 or writes.shape != (reads.shape[0], reads.shape[2]):
            raise ValueError('invalid allocation diagnostic shapes')
        self.entries.append(dict(
            instructions=len(reads), read_mean=reads.mean(0), read_entropy=entropy(reads).mean(0),
            write_mean=writes.mean(0), write_entropy=entropy(writes).mean(),
            mean_strength=strength.mean(), mean_actual_write_fraction=(writes * strength[:, None]).mean(0),
            state_squared_mean=bank.square().mean(),
            content_squared_mean=bank[:, :bank.shape[1] // 2].square().mean(),
            source_squared_mean=bank[:, bank.shape[1] // 2:].square().mean(),
        ))
        # Full allocations and final latent bank are kept at float32 precision.
        # No 1,025-way logits or all-step hidden tensors inflate the artifact.
        return dict(
            valid_instruction_count=len(reads), read_weights=reads.float().tolist(),
            write_weights=writes.float().tolist(), write_strength=strength.float().tolist(),
            final_slot_state=bank.float().tolist(),
            return_query_initial=diagnostic['return_query_initial'][0].detach().cpu().tolist(),
            return_query=diagnostic['return_query'][0].detach().cpu().tolist(),
            query_read_weights=diagnostic['query_read_weights'][:, 0].detach().cpu().tolist(),
            query_trace_weights=diagnostic['query_trace_weights'][:, 0, valid[0]].detach().cpu().tolist(),
        )

    def summarize(self):
        if not self.entries:
            raise ValueError('no normal allocation observations')
        means = torch.stack([entry['read_mean'] for entry in self.entries])
        entropies = torch.stack([entry['read_entropy'] for entry in self.entries])
        write_means = torch.stack([entry['write_mean'] for entry in self.entries])
        write_entropies = torch.stack([entry['write_entropy'] for entry in self.entries])
        # A fixed difference between read heads contributes zero to both JS
        # quantities here: every head is aggregated separately before averaging.
        read_js = entropy(means.mean(0)) - entropy(means).mean(0)
        write_js = entropy(write_means.mean(0)) - entropy(write_means).mean(0)
        read_within = (entropy(means) - entropies).mean(0)
        write_within = (entropy(write_means) - write_entropies).mean()
        return dict(
            episodes=len(self.entries), valid_instructions=sum(e['instructions'] for e in self.entries),
            weighting='Equal episode weight. Each episode first averages its valid instruction allocations.',
            mean_read_mass_per_head=means.mean(0).tolist(), mean_write_mass=write_means.mean(0).tolist(),
            read_entropy_nats_per_head=entropies.mean(0).tolist(),
            read_between_episode_js_nats_per_head=read_js.tolist(),
            read_between_episode_js_nats_mean_heads=float(read_js.mean()),
            read_within_episode_variation_nats_per_head=read_within.tolist(),
            write_entropy_nats=float(write_entropies.mean()),
            write_between_episode_js_nats=float(write_js),
            write_within_episode_variation_nats=float(write_within),
            mean_write_strength=float(torch.stack([e['mean_strength'] for e in self.entries]).mean()),
            mean_actual_write_fraction=torch.stack([e['mean_actual_write_fraction'] for e in self.entries]).mean(0).tolist(),
            latent_slot_rms=float(torch.stack([e['state_squared_mean'] for e in self.entries]).mean().sqrt()),
            latent_content_channel_rms=float(torch.stack([e['content_squared_mean'] for e in self.entries]).mean().sqrt()),
            intended_source_channel_rms=float(torch.stack([e['source_squared_mean'] for e in self.entries]).mean().sqrt()),
            interpretation='Allocation variation and latent RMS are utilization measurements. '
                           'The source channel is an intended role, not verified integer or provenance semantics.',
        )


class InitialForwardRecorder:
    """Model-shaped view: retain diagnostics only for the first input in a rollout.
    SlotPredictor retains its original decision logic. This view supplies no
    episode metadata, labels, execution tests or feedback to the model.
    """
    def __init__(self, model, accumulator):
        self.model = model
        self.config = model.config
        self.none_index = model.none_index
        self.accumulator = accumulator
        self.pending = True
        self.initial = None

    def eval(self):
        self.model.eval()
        return self

    def __call__(self, *encoded, memory_ablation='none'):
        if self.pending:
            scores, bits, diagnostic = self.model(*encoded, return_diagnostics=True,
                                                  memory_ablation=memory_ablation)
            self.initial = self.accumulator.add(diagnostic)
            self.pending = False
            return scores, bits
        return self.model(*encoded, memory_ablation=memory_ablation)


def action_key(score):
    transitions = score['audit']['transitions']
    decision = transitions[0]['decision'] if transitions else None
    if decision is None:
        return None
    # Confidence changes alone are not discrete action changes.
    return dict(stop=decision['stop'], edits=decision['edits'], reason=decision['reason'])


def compare_modes(episodes, normal, disabled):
    if not (len(episodes) == len(normal) == len(disabled)):
        raise ValueError('paired ablation results are incomplete')
    groups = defaultdict(Counter)
    for episode, original, ablated in zip(episodes, normal, disabled):
        if episode['id'] != original['id'] or episode['id'] != ablated['id']:
            raise ValueError('ablation result identifiers are misaligned')
        for category in ('all', 'edit' if episode['edits'] else 'noop', 'category/' + episode['category']):
            counter = groups[category]
            counter['total'] += 1
            first, second = original['strict_success'], ablated['strict_success']
            counter['both_strict'] += bool(first and second)
            counter['normal_only_strict'] += bool(first and not second)
            counter['zero_read_only_strict'] += bool(second and not first)
            counter['neither_strict'] += bool(not first and not second)
            counter['final_bytes_different'] += original['final_hex'] != ablated['final_hex']
            counter['first_action_different'] += action_key(original) != action_key(ablated)
            counter['stop_reason_different'] += original['stop_reason'] != ablated['stop_reason']
    return {name: dict(counts) for name, counts in groups.items()}


def declaration(checkpoint, report, data, output, threads):
    checkpoint = resolved_public_path(checkpoint)
    report = resolved_public_path(report if report is not None else checkpoint.parent / 'report.json')
    data = resolved_public_path(data)
    if data != PUBLIC_DATA.resolve() or sha(data) != EXPECTED_DATA_SHA256:
        raise ValueError('this diagnostic is restricted to the committed 128 public development episodes')
    training = json.loads(report.read_text())
    arm, seed = training['arm']['name'], training['seed']
    if (training.get('complete') is not True or arm not in ARMS or seed not in SEEDS
            or training['last'].get('stage') != 4 or training['last'].get('kind') != 'slot'):
        raise ValueError('a completed terminal slot8/slot16 run for a declared seed is required')
    committed = training['last']['checkpoint']
    if resolved_public_path(committed['path']) != checkpoint or sha(checkpoint) != committed['sha256']:
        raise ValueError('checkpoint is not the report-committed terminal model')
    public_protocol = json.loads(PUBLIC_PROTOCOL.read_text())
    if public_protocol['files']['development-episodes.json'] != EXPECTED_DATA_SHA256:
        raise ValueError('public data commitment differs from the diagnostic declaration')
    if threads not in (1, 2):
        raise ValueError('diagnostic threads must be one or two')
    output = (ROOT / output).resolve()
    allowed = (ROOT / 'work/morphology-20261017/diagnostics').resolve()
    if not output.is_relative_to(allowed) or output == allowed:
        raise ValueError('output must be a distinct run folder below work/morphology-20261017/diagnostics')
    if output.exists():
        raise ValueError('refuse to repeat or overwrite a diagnostic run')
    if any(path.is_relative_to(output) for path in (checkpoint, report, data)):
        raise ValueError('output must not contain input artifacts')
    return dict(
        schema=SCHEMA, created_utc=datetime.now(timezone.utc).isoformat(),
        model_id=f'{arm}-seed{seed}-terminal', arm=arm, seed=seed,
        checkpoint=file_record(checkpoint), terminal_report=file_record(report),
        data=file_record(data), data_protocol=file_record(PUBLIC_PROTOCOL),
        expected_episodes=128, expected_edit=64, expected_noop=64,
        expected_return_pairs=32, expected_invariance_pairs=64,
        sources=[file_record(relative) for relative in SOURCE_FILES],
        modes=list(MODES), allocation_mode='normal only, first unmodified input per episode',
        dense_inference=True, threads=threads,
        bounds=dict(max_steps_including_stop=8, confidence_threshold=0.,
                    max_total_changed_bytes=8, max_output_bytes=1024),
        protocol='Public mechanism diagnostic, not a fresh test, architecture selection, '
                 'training signal, promotion condition or basis for changing defaults. '
                 'Every declared terminal slot8/slot16 seed gets both modes once. '
                 'zero_read is an intervention on retrieved vectors; memory writes and '
                 'global instruction-trace retrieval remain in the computation. '
                 'No additional no_memory inference is included.',
        interpretation='A read-disable effect demonstrates dependence on that pathway under this intervention. '
                       'It does not by itself establish correct value provenance, integer execution, '
                       'minimal causal sufficiency or held-out generalization.',
        output=str(output.relative_to(ROOT)),
    )


def verify_frozen(frozen, model=None, fingerprint=None):
    for record in [frozen['checkpoint'], frozen['terminal_report'], frozen['data'],
                   frozen['data_protocol'], *frozen['sources']]:
        check_record(record)
    if model is not None and state_fingerprint(model) != fingerprint:
        raise AssertionError('diagnostic inference changed in-memory model tensors')


def run(checkpoint, data=PUBLIC_DATA, *, output, report=None, threads=2):
    frozen = declaration(checkpoint, report, data, output, threads)
    folder = ROOT / frozen['output']
    folder.mkdir(parents=True)
    protocol_path = folder / 'protocol.json'
    write_json(protocol_path, frozen)
    protocol_sha = sha(protocol_path)
    torch.set_num_threads(threads)
    torch.set_num_interop_threads(1)
    checkpoint = check_record(frozen['checkpoint'])
    model = load_slot(checkpoint).eval()
    if (model.config.mode != 'memory' or model.config.slots != ARMS[frozen['arm']]
            or model.config.width != 64 or model.config.query_steps != 2 or model.parameter_count != 118035):
        raise ValueError('terminal checkpoint config differs from its declared memory arm')
    initial_fingerprint = state_fingerprint(model)
    verify_frozen(frozen, model, initial_fingerprint)
    # Case contents are opened only after the per-run public declaration exists.
    episodes = json.loads(check_record(frozen['data']).read_text())
    if (len(episodes) != 128 or len({e['id'] for e in episodes}) != 128
            or sum(bool(e['edits']) for e in episodes) != 64):
        raise ValueError('public episode counts differ from the predeclared scope')
    before = dict(state_sha256=initial_fingerprint, checkpoint_sha256=sha(checkpoint),
                  config=asdict(model.config), parameters=model.parameter_count)
    write_json(folder / 'before.json', before)
    all_scores, reports = {}, {}
    allocations = AllocationAccumulator()
    for mode in MODES:
        trace_path = folder / (mode + '.jsonl.gz')
        scored = []
        wall_start, cpu_start = time.perf_counter(), time.process_time()
        with trace_path.open('xb') as raw:
            with gzip.GzipFile(filename='', fileobj=raw, mode='wb', mtime=0, compresslevel=6) as compressed:
                with io.TextIOWrapper(compressed, encoding='utf-8') as output_stream:
                    for episode in episodes:
                        view = InitialForwardRecorder(model, allocations) if mode == 'normal' else model
                        predictor = SlotPredictor(view, threshold=0.,
                                                  memory_ablation='none' if mode == 'normal' else 'zero_read')
                        score = score_episode(predictor, episode,
                                              predictor_id=frozen['model_id'] + '/' + mode, max_steps=8)
                        row = dict(mode=mode, scored=score)
                        if mode == 'normal':
                            if view.initial is None:
                                raise AssertionError('normal episode did not record its initial forward')
                            row['initial_forward'] = view.initial
                        output_stream.write(json.dumps(row, ensure_ascii=False, separators=(',', ':'),
                                                      allow_nan=False) + '\n')
                        scored.append(score)
        verify_frozen(frozen, model, initial_fingerprint)
        if sha(protocol_path) != protocol_sha:
            raise ValueError('diagnostic protocol changed during inference')
        groups = summarize_scored(episodes, scored)
        if (groups['all']['pair_total'] != 32 or groups['all']['invariance_total'] != 64):
            raise ValueError('public pair/invariance groups differ from the declaration')
        reports[mode] = dict(
            architecture_summary=summarize(scored), morphology_groups=groups,
            stops=dict(Counter(s['stop_reason'] for s in scored)), trace=file_record(trace_path),
            wall_seconds=time.perf_counter() - wall_start, cpu_seconds=time.process_time() - cpu_start,
            state_sha256_after=state_fingerprint(model), checkpoint_sha256_after=sha(checkpoint),
        )
        all_scores[mode] = scored
        print(json.dumps(dict(event='public_diagnostic_mode_complete', model=frozen['model_id'],
                              mode=mode, strict=groups['all']['strict_success'], episodes=len(scored))), flush=True)
    result = dict(
        schema=SCHEMA, protocol=file_record(protocol_path), model_id=frozen['model_id'],
        completed_utc=datetime.now(timezone.utc).isoformat(), before=before, modes=reports,
        paired_intervention=compare_modes(episodes, all_scores['normal'], all_scores['zero_read']),
        normal_initial_allocations=allocations.summarize(),
        weights_unchanged=True, disk_checkpoint_unchanged=True, source_and_data_unchanged=True,
        final_state_sha256=state_fingerprint(model), final_checkpoint_sha256=sha(checkpoint),
        threads=threads, max_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,
        interpretation=frozen['interpretation'], selection_used=False, training_performed=False,
        sealed_cases_read=False,
    )
    verify_frozen(frozen, model, initial_fingerprint)
    if sha(protocol_path) != protocol_sha:
        raise ValueError('diagnostic protocol changed at completion')
    write_json(folder / 'report.json', result)
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--checkpoint', required=True, type=Path)
    parser.add_argument('--data', type=Path, default=PUBLIC_DATA)
    parser.add_argument('--terminal-report', type=Path)
    parser.add_argument('--output', required=True, type=Path)
    parser.add_argument('--threads', type=int, default=2)
    args = parser.parse_args()
    result = run(args.checkpoint, args.data, output=args.output, report=args.terminal_report, threads=args.threads)
    print(json.dumps(dict(model=result['model_id'], report=str(args.output / 'report.json'),
                          unchanged=result['weights_unchanged'])), flush=True)


if __name__ == '__main__':
    main()

	"""Predeclared public-only read-disable diagnostic for terminal slot models.

	This does not train, select, promote, or inspect sealed cases. It compares normal
	and zero_read inference for all 128 committed public development episodes. The
	old scorer supplies full action audits; ordinary first-input forwards also
	record learned allocation distributions and latent states. Read-head variation
	is computed separately per head and across episode-mean distributions.

	Only terminal slot8/slot16 models for the two declared seeds are accepted. The
	same loaded tensors serve both modes. Disk hashes, source hashes and in-memory
	state fingerprints are checked before, between and after inference.
	"""
	from __future__ import annotations

	import argparse
	from collections import Counter, defaultdict
	from dataclasses import asdict
	from datetime import datetime, timezone
	import gzip
	import hashlib
	import io
	import json
	from pathlib import Path
	import resource
	import sys
	import time

	ROOT = Path(__file__).resolve().parents[1]
	if str(ROOT) not in sys.path:
	sys.path.insert(0, str(ROOT))

	import torch
	from experiments.architecture_eval import score_episode, summarize
	from experiments.morphology_data import summarize_scored
	from lightnest.slot_patch import SlotPredictor, load_slot, state_fingerprint

	SCHEMA = 'lightnest-public-slot-read-disable-v1'
	PUBLIC_DATA = ROOT / 'work/morphology-20261017/data/development-episodes.json'
	PUBLIC_PROTOCOL = ROOT / 'work/morphology-20261017/data/protocol.json'
	EXPECTED_DATA_SHA256 = '819034832d42fca7638c9c4b649af06b8f077fb91971dd640b8516392d03e4cb'
	SEEDS = (104729, 130363)
	ARMS = {'slot8': 8, 'slot16': 16}
	MODES = ('normal', 'zero_read')
	SOURCE_FILES = (
	'experiments/dependency_diagnostics.py', 'experiments/architecture_eval.py',
	'experiments/morphology_data.py', 'lightnest/slot_patch.py',
	'lightnest/relation_patch.py', 'lightnest/iterative_patch.py', 'lightnest/rv32.py',
	)


	def sha(path):
	digest = hashlib.sha256()
	with Path(path).open('rb') as source:
	for block in iter(lambda: source.read(1024 * 1024), b''):
	digest.update(block)
	return digest.hexdigest()


	def resolved_public_path(path):
	path = Path(path)
	path = (path if path.is_absolute() else ROOT / path).resolve()
	if not path.is_relative_to(ROOT) or any(part.lower() in ('sealed', 'evaluation-final', 'final-test')
	for part in path.parts):
	raise ValueError('diagnostic inputs must be public files inside the repository')
	if not path.is_file():
	raise ValueError('missing public input: ' + str(path))
	return path


	def file_record(path):
	path = resolved_public_path(path)
	return dict(path=str(path.relative_to(ROOT)), sha256=sha(path), bytes=path.stat().st_size)


	def check_record(record):
	path = resolved_public_path(record['path'])
	if path.stat().st_size != record['bytes'] or sha(path) != record['sha256']:
	raise ValueError('frozen public diagnostic input changed: ' + record['path'])
	return path


	def write_json(path, data):
	with Path(path).open('x', encoding='utf-8') as output:
	json.dump(data, output, ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False)
	output.write('\n')


	def entropy(weights):
	"""Entropy on the last axis; use float64 when aggregating recorded weights."""
	return -(weights * weights.clamp_min(1e-300).log()).sum(-1)


	class AllocationAccumulator:
	"""One normal, initial-state forward per episode; no labels enter this object."""
	def __init__(self):
	self.entries = []

	def add(self, diagnostic):
	valid = diagnostic['valid']
	if valid.shape[0] != 1:
	raise ValueError('allocation collection expects one initial episode at a time')
	reads = diagnostic['read_weights'][0, valid[0]].detach().double().cpu()
	writes = diagnostic['write_weights'][0, valid[0]].detach().double().cpu()
	strength = diagnostic['write_strength'][0, valid[0]].detach().double().cpu()
	bank = diagnostic['slot_state'][0].detach().double().cpu()
	if not reads.numel() or reads.shape[1] != 2 or writes.shape != (reads.shape[0], reads.shape[2]):
	raise ValueError('invalid allocation diagnostic shapes')
	self.entries.append(dict(
	instructions=len(reads), read_mean=reads.mean(0), read_entropy=entropy(reads).mean(0),
	write_mean=writes.mean(0), write_entropy=entropy(writes).mean(),
	mean_strength=strength.mean(), mean_actual_write_fraction=(writes * strength[:, None]).mean(0),
	state_squared_mean=bank.square().mean(),
	content_squared_mean=bank[:, :bank.shape[1] // 2].square().mean(),
	source_squared_mean=bank[:, bank.shape[1] // 2:].square().mean(),
	))
	# Full allocations and final latent bank are kept at float32 precision.
	# No 1,025-way logits or all-step hidden tensors inflate the artifact.
	return dict(
	valid_instruction_count=len(reads), read_weights=reads.float().tolist(),
	write_weights=writes.float().tolist(), write_strength=strength.float().tolist(),
	final_slot_state=bank.float().tolist(),
	return_query_initial=diagnostic['return_query_initial'][0].detach().cpu().tolist(),
	return_query=diagnostic['return_query'][0].detach().cpu().tolist(),
	query_read_weights=diagnostic['query_read_weights'][:, 0].detach().cpu().tolist(),
	query_trace_weights=diagnostic['query_trace_weights'][:, 0, valid[0]].detach().cpu().tolist(),
	)

	def summarize(self):
	if not self.entries:
	raise ValueError('no normal allocation observations')
	means = torch.stack([entry['read_mean'] for entry in self.entries])
	entropies = torch.stack([entry['read_entropy'] for entry in self.entries])
	write_means = torch.stack([entry['write_mean'] for entry in self.entries])
	write_entropies = torch.stack([entry['write_entropy'] for entry in self.entries])
	# A fixed difference between read heads contributes zero to both JS
	# quantities here: every head is aggregated separately before averaging.
	read_js = entropy(means.mean(0)) - entropy(means).mean(0)
	write_js = entropy(write_means.mean(0)) - entropy(write_means).mean(0)
	read_within = (entropy(means) - entropies).mean(0)
	write_within = (entropy(write_means) - write_entropies).mean()
	return dict(
	episodes=len(self.entries), valid_instructions=sum(e['instructions'] for e in self.entries),
	weighting='Equal episode weight. Each episode first averages its valid instruction allocations.',
	mean_read_mass_per_head=means.mean(0).tolist(), mean_write_mass=write_means.mean(0).tolist(),
	read_entropy_nats_per_head=entropies.mean(0).tolist(),
	read_between_episode_js_nats_per_head=read_js.tolist(),
	read_between_episode_js_nats_mean_heads=float(read_js.mean()),
	read_within_episode_variation_nats_per_head=read_within.tolist(),
	write_entropy_nats=float(write_entropies.mean()),
	write_between_episode_js_nats=float(write_js),
	write_within_episode_variation_nats=float(write_within),
	mean_write_strength=float(torch.stack([e['mean_strength'] for e in self.entries]).mean()),
	mean_actual_write_fraction=torch.stack([e['mean_actual_write_fraction'] for e in self.entries]).mean(0).tolist(),
	latent_slot_rms=float(torch.stack([e['state_squared_mean'] for e in self.entries]).mean().sqrt()),
	latent_content_channel_rms=float(torch.stack([e['content_squared_mean'] for e in self.entries]).mean().sqrt()),
	intended_source_channel_rms=float(torch.stack([e['source_squared_mean'] for e in self.entries]).mean().sqrt()),
	interpretation='Allocation variation and latent RMS are utilization measurements. '
	'The source channel is an intended role, not verified integer or provenance semantics.',
	)


	class InitialForwardRecorder:
	"""Model-shaped view: retain diagnostics only for the first input in a rollout.

	SlotPredictor retains its original decision logic. This view supplies no
	episode metadata, labels, execution tests or feedback to the model.
	"""
	def __init__(self, model, accumulator):
	self.model = model
	self.config = model.config
	self.none_index = model.none_index
	self.accumulator = accumulator
	self.pending = True
	self.initial = None

	def eval(self):
	self.model.eval()
	return self

	def __call__(self, *encoded, memory_ablation='none'):
	if self.pending:
	scores, bits, diagnostic = self.model(*encoded, return_diagnostics=True,
	memory_ablation=memory_ablation)
	self.initial = self.accumulator.add(diagnostic)
	self.pending = False
	return scores, bits
	return self.model(*encoded, memory_ablation=memory_ablation)


	def action_key(score):
	transitions = score['audit']['transitions']
	decision = transitions[0]['decision'] if transitions else None
	if decision is None:
	return None
	# Confidence changes alone are not discrete action changes.
	return dict(stop=decision['stop'], edits=decision['edits'], reason=decision['reason'])


	def compare_modes(episodes, normal, disabled):
	if not (len(episodes) == len(normal) == len(disabled)):
	raise ValueError('paired ablation results are incomplete')
	groups = defaultdict(Counter)
	for episode, original, ablated in zip(episodes, normal, disabled):
	if episode['id'] != original['id'] or episode['id'] != ablated['id']:
	raise ValueError('ablation result identifiers are misaligned')
	for category in ('all', 'edit' if episode['edits'] else 'noop', 'category/' + episode['category']):
	counter = groups[category]
	counter['total'] += 1
	first, second = original['strict_success'], ablated['strict_success']
	counter['both_strict'] += bool(first and second)
	counter['normal_only_strict'] += bool(first and not second)
	counter['zero_read_only_strict'] += bool(second and not first)
	counter['neither_strict'] += bool(not first and not second)
	counter['final_bytes_different'] += original['final_hex'] != ablated['final_hex']
	counter['first_action_different'] += action_key(original) != action_key(ablated)
	counter['stop_reason_different'] += original['stop_reason'] != ablated['stop_reason']
	return {name: dict(counts) for name, counts in groups.items()}


	def declaration(checkpoint, report, data, output, threads):
	checkpoint = resolved_public_path(checkpoint)
	report = resolved_public_path(report if report is not None else checkpoint.parent / 'report.json')
	data = resolved_public_path(data)
	if data != PUBLIC_DATA.resolve() or sha(data) != EXPECTED_DATA_SHA256:
	raise ValueError('this diagnostic is restricted to the committed 128 public development episodes')
	training = json.loads(report.read_text())
	arm, seed = training['arm']['name'], training['seed']
	if (training.get('complete') is not True or arm not in ARMS or seed not in SEEDS
	or training['last'].get('stage') != 4 or training['last'].get('kind') != 'slot'):
	raise ValueError('a completed terminal slot8/slot16 run for a declared seed is required')
	committed = training['last']['checkpoint']
	if resolved_public_path(committed['path']) != checkpoint or sha(checkpoint) != committed['sha256']:
	raise ValueError('checkpoint is not the report-committed terminal model')
	public_protocol = json.loads(PUBLIC_PROTOCOL.read_text())
	if public_protocol['files']['development-episodes.json'] != EXPECTED_DATA_SHA256:
	raise ValueError('public data commitment differs from the diagnostic declaration')
	if threads not in (1, 2):
	raise ValueError('diagnostic threads must be one or two')
	output = (ROOT / output).resolve()
	allowed = (ROOT / 'work/morphology-20261017/diagnostics').resolve()
	if not output.is_relative_to(allowed) or output == allowed:
	raise ValueError('output must be a distinct run folder below work/morphology-20261017/diagnostics')
	if output.exists():
	raise ValueError('refuse to repeat or overwrite a diagnostic run')
	if any(path.is_relative_to(output) for path in (checkpoint, report, data)):
	raise ValueError('output must not contain input artifacts')
	return dict(
	schema=SCHEMA, created_utc=datetime.now(timezone.utc).isoformat(),
	model_id=f'{arm}-seed{seed}-terminal', arm=arm, seed=seed,
	checkpoint=file_record(checkpoint), terminal_report=file_record(report),
	data=file_record(data), data_protocol=file_record(PUBLIC_PROTOCOL),
	expected_episodes=128, expected_edit=64, expected_noop=64,
	expected_return_pairs=32, expected_invariance_pairs=64,
	sources=[file_record(relative) for relative in SOURCE_FILES],
	modes=list(MODES), allocation_mode='normal only, first unmodified input per episode',
	dense_inference=True, threads=threads,
	bounds=dict(max_steps_including_stop=8, confidence_threshold=0.,
	max_total_changed_bytes=8, max_output_bytes=1024),
	protocol='Public mechanism diagnostic, not a fresh test, architecture selection, '
	'training signal, promotion condition or basis for changing defaults. '
	'Every declared terminal slot8/slot16 seed gets both modes once. '
	'zero_read is an intervention on retrieved vectors; memory writes and '
	'global instruction-trace retrieval remain in the computation. '
	'No additional no_memory inference is included.',
	interpretation='A read-disable effect demonstrates dependence on that pathway under this intervention. '
	'It does not by itself establish correct value provenance, integer execution, '
	'minimal causal sufficiency or held-out generalization.',
	output=str(output.relative_to(ROOT)),
	)


	def verify_frozen(frozen, model=None, fingerprint=None):
	for record in [frozen['checkpoint'], frozen['terminal_report'], frozen['data'],
	frozen['data_protocol'], *frozen['sources']]:
	check_record(record)
	if model is not None and state_fingerprint(model) != fingerprint:
	raise AssertionError('diagnostic inference changed in-memory model tensors')


	def run(checkpoint, data=PUBLIC_DATA, *, output, report=None, threads=2):
	frozen = declaration(checkpoint, report, data, output, threads)
	folder = ROOT / frozen['output']
	folder.mkdir(parents=True)
	protocol_path = folder / 'protocol.json'
	write_json(protocol_path, frozen)
	protocol_sha = sha(protocol_path)
	torch.set_num_threads(threads)
	torch.set_num_interop_threads(1)
	checkpoint = check_record(frozen['checkpoint'])
	model = load_slot(checkpoint).eval()
	if (model.config.mode != 'memory' or model.config.slots != ARMS[frozen['arm']]
	or model.config.width != 64 or model.config.query_steps != 2 or model.parameter_count != 118035):
	raise ValueError('terminal checkpoint config differs from its declared memory arm')
	initial_fingerprint = state_fingerprint(model)
	verify_frozen(frozen, model, initial_fingerprint)
	# Case contents are opened only after the per-run public declaration exists.
	episodes = json.loads(check_record(frozen['data']).read_text())
	if (len(episodes) != 128 or len({e['id'] for e in episodes}) != 128
	or sum(bool(e['edits']) for e in episodes) != 64):
	raise ValueError('public episode counts differ from the predeclared scope')
	before = dict(state_sha256=initial_fingerprint, checkpoint_sha256=sha(checkpoint),
	config=asdict(model.config), parameters=model.parameter_count)
	write_json(folder / 'before.json', before)
	all_scores, reports = {}, {}
	allocations = AllocationAccumulator()
	for mode in MODES:
	trace_path = folder / (mode + '.jsonl.gz')
	scored = []
	wall_start, cpu_start = time.perf_counter(), time.process_time()
	with trace_path.open('xb') as raw:
	with gzip.GzipFile(filename='', fileobj=raw, mode='wb', mtime=0, compresslevel=6) as compressed:
	with io.TextIOWrapper(compressed, encoding='utf-8') as output_stream:
	for episode in episodes:
	view = InitialForwardRecorder(model, allocations) if mode == 'normal' else model
	predictor = SlotPredictor(view, threshold=0.,
	memory_ablation='none' if mode == 'normal' else 'zero_read')
	score = score_episode(predictor, episode,
	predictor_id=frozen['model_id'] + '/' + mode, max_steps=8)
	row = dict(mode=mode, scored=score)
	if mode == 'normal':
	if view.initial is None:
	raise AssertionError('normal episode did not record its initial forward')
	row['initial_forward'] = view.initial
	output_stream.write(json.dumps(row, ensure_ascii=False, separators=(',', ':'),
	allow_nan=False) + '\n')
	scored.append(score)
	verify_frozen(frozen, model, initial_fingerprint)
	if sha(protocol_path) != protocol_sha:
	raise ValueError('diagnostic protocol changed during inference')
	groups = summarize_scored(episodes, scored)
	if (groups['all']['pair_total'] != 32 or groups['all']['invariance_total'] != 64):
	raise ValueError('public pair/invariance groups differ from the declaration')
	reports[mode] = dict(
	architecture_summary=summarize(scored), morphology_groups=groups,
	stops=dict(Counter(s['stop_reason'] for s in scored)), trace=file_record(trace_path),
	wall_seconds=time.perf_counter() - wall_start, cpu_seconds=time.process_time() - cpu_start,
	state_sha256_after=state_fingerprint(model), checkpoint_sha256_after=sha(checkpoint),
	)
	all_scores[mode] = scored
	print(json.dumps(dict(event='public_diagnostic_mode_complete', model=frozen['model_id'],
	mode=mode, strict=groups['all']['strict_success'], episodes=len(scored))), flush=True)
	result = dict(
	schema=SCHEMA, protocol=file_record(protocol_path), model_id=frozen['model_id'],
	completed_utc=datetime.now(timezone.utc).isoformat(), before=before, modes=reports,
	paired_intervention=compare_modes(episodes, all_scores['normal'], all_scores['zero_read']),
	normal_initial_allocations=allocations.summarize(),
	weights_unchanged=True, disk_checkpoint_unchanged=True, source_and_data_unchanged=True,
	final_state_sha256=state_fingerprint(model), final_checkpoint_sha256=sha(checkpoint),
	threads=threads, max_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,
	interpretation=frozen['interpretation'], selection_used=False, training_performed=False,
	sealed_cases_read=False,
	)
	verify_frozen(frozen, model, initial_fingerprint)
	if sha(protocol_path) != protocol_sha:
	raise ValueError('diagnostic protocol changed at completion')
	write_json(folder / 'report.json', result)
	return result


	def main():
	parser = argparse.ArgumentParser(description=__doc__)
	parser.add_argument('--checkpoint', required=True, type=Path)
	parser.add_argument('--data', type=Path, default=PUBLIC_DATA)
	parser.add_argument('--terminal-report', type=Path)
	parser.add_argument('--output', required=True, type=Path)
	parser.add_argument('--threads', type=int, default=2)
	args = parser.parse_args()
	result = run(args.checkpoint, args.data, output=args.output, report=args.terminal_report, threads=args.threads)
	print(json.dumps(dict(model=result['model_id'], report=str(args.output / 'report.json'),
	unchanged=result['weights_unchanged'])), flush=True)


	if __name__ == '__main__':
	main()