LightNest / experiments /morphology_data.py
Integrate latest architecture breakthrough, harder tasks, and evaluation suite
e64c5d2 verified 5 days ago
37.6 kB
"""Counterexample-derived, sealed data for bounded neural architecture search.
This generator never loads a predictor. Labels are construction-time expression
ancestry, checked against two RV32 executors. Only request/current bytes are
model inputs; targets, graph provenance, and execution cases are external labels.
The previous architecture experiment remains immutable.
"""
from __future__ import annotations

import argparse
from collections import Counter, defaultdict
from copy import deepcopy
from datetime import datetime, timezone
import hashlib
import importlib.util
import itertools
import json
from pathlib import Path
import random
import secrets
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from experiments.architecture_data import (
    Builder, EDIT_COMMANDS, NEG_COMMANDS, EDGE, TRAIN_REGS, pack, s32,
    cases_for as old_cases_for,
)
from lightnest import rv32

BASE_COMMIT = 'fcbf22b77aa308cc23f4f0d4cee84071ec486984'
OLD_AUDITOR = 'work/architecture-20261017/audit_evaluation.py'
OLD_AUDITOR_SHA = 'b1e47e27ebdc2d312f74419ea674a2d32c6f4bf18882279f61e09b22282f43b8'
PUBLIC_SEEDS = dict(train=982451653, development=982451707, selection=982451761,
                    replay=982451809, mixture_order=982451863, retention=982451909)
FAMILIES = (
    ('copy_chain', 'copy_chain', (1, 2, 3), 1),
    ('rename_chain', 'rename_chain', (1, 2, 3), 1),
    ('overwrite_kill', 'overwrite_kill', (1, 2, 3), 1),
    ('overwrite_retained', 'overwrite_retained', (1, 2, 3), 1),
    ('shared_dag', 'shared_dag', (1, 2), 2),
    ('two_edits', 'chain', (1, 2), 2),
    ('three_edits', 'chain', (1, 2), 3),
    ('shared_overwrite', 'shared_overwrite', (1, 2), 3),
)
EXTENDED = (
    ('long_copy', 'copy_chain', (5, 7), 1),
    ('long_rename', 'rename_chain', (5, 7), 1),
    ('deep_overwrite_kill', 'overwrite_kill', (5, 7), 1),
    ('deep_overwrite_retained', 'overwrite_retained', (5, 7), 1),
    ('deep_shared_dag', 'shared_dag', (5, 7), 2),
    ('four_edits', 'chain', (1, 2), 4),
    ('six_edits', 'chain', (1, 2), 6),
    ('deep_shared_overwrite', 'shared_overwrite', (5, 7), 3),
)
RETENTION_COUNTS = dict(simple_edit=14, far_pair=20, two_edits=12, three_edits=12,
                        shared_dag=12, overwrite=10, already_correct=16,
                        negation=16, dead_only=16)


def sha(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def write_json(path, value):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open('x') as stream:
        json.dump(value, stream, sort_keys=True, separators=(',', ':'), allow_nan=False)
        stream.write('\n')


def write_jsonl(path, rows):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open('x') as stream:
        for row in rows:
            stream.write(json.dumps(row, sort_keys=True, separators=(',', ':'), allow_nan=False) + '\n')


def independent_executor():
    """Reuse the frozen second VM, without loading its data or any models."""
    path = ROOT / OLD_AUDITOR
    if sha(path) != OLD_AUDITOR_SHA:
        raise ValueError('independent VM source differs from the previous frozen auditor')
    spec = importlib.util.spec_from_file_location('morphology_independent_vm', path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module.independent_execute


class MorphBuilder(Builder):
    def immediate(self, register, immediate, out=None):
        out = self.reg() if out is None else out
        node = len(self.nodes)
        self.nodes.append(dict(kind='immediate', source=self.reg_values[register],
                               immediate=immediate, word=len(self.words), register=out))
        self.words.append(rv32.addi(f'x{out}', f'x{register}', immediate))
        self.reg_values[out] = node
        return out

    def copy(self, register, out=None):
        out = register if out is None else out
        node = len(self.nodes)
        self.nodes.append(dict(kind='copy', source=self.reg_values[register],
                               word=len(self.words), register=out))
        self.words.append(rv32.addi(f'x{out}', f'x{register}', 0))
        self.reg_values[out] = node
        return out

    def shift(self, register, amount, out=None):
        out = self.reg() if out is None else out
        node = len(self.nodes)
        self.nodes.append(dict(kind='shift', source=self.reg_values[register],
                               amount=amount, word=len(self.words), register=out))
        self.words.append(rv32.slli(f'x{out}', f'x{register}', amount))
        self.reg_values[out] = node
        return out

    def copies(self, register, depth, rename=False):
        alternate = self.reg() if rename else register
        for _ in range(depth):
            old = register
            register = self.copy(register, out=alternate)
            if rename:
                alternate = old
        return register

    def finish_morphology(self, out, command, category, episode_id, *, negate=False):
        output_node = self.reg_values[out]
        live = set()

        def visit(index):
            if index in live:
                return
            live.add(index)
            item = self.nodes[index]
            if item['kind'] == 'op':
                visit(item['left'])
                visit(item['right'])
            elif item['kind'] in ('copy', 'shift', 'immediate'):
                visit(item['source'])

        visit(output_node)
        words = self.words + [rv32.addi('x10', f'x{out}', 0), rv32.EBREAK]
        target_words = list(words)
        changed = []
        for index in sorted(live):
            item = self.nodes[index]
            if item['kind'] == 'op' and item['operation'] == self.source and not negate:
                rd, left, right = item['registers']
                target_words[item['word']] = getattr(rv32, self.destination)(
                    f'x{rd}', f'x{left}', f'x{right}')
                changed.append(index)
        binary, target = pack(words), pack(target_words)
        return dict(id=episode_id, episode_id=episode_id, category=category,
                    command=command, binary=binary.hex(), target=target.hex(),
                    edits=[dict(offset=i, value=y) for i, (x, y) in enumerate(zip(binary, target)) if x != y],
                    source_operation=self.source, destination_operation=self.destination,
                    oracle=dict(nodes=deepcopy(self.nodes), output_node=output_node,
                                changed_nodes=changed, destination=self.destination,
                                input_count=self.input_count))


def expression_expected(oracle, inputs):
    """Arithmetic over immutable construction nodes, without instruction decode."""
    changed = set(oracle['changed_nodes'])
    values = []
    for index, item in enumerate(oracle['nodes']):
        kind = item['kind']
        if kind == 'input':
            value = inputs[item['index']]
        elif kind in ('copy', 'shift', 'immediate'):
            if not 0 <= item['source'] < index:
                raise ValueError('invalid construction-time dependency')
            value = values[item['source']]
            if kind == 'shift':
                value <<= item['amount']
            elif kind == 'immediate':
                value += item['immediate']
        elif kind == 'op':
            if not 0 <= item['left'] < index or not 0 <= item['right'] < index:
                raise ValueError('invalid construction-time dependency')
            left, right = values[item['left']], values[item['right']]
            operation = oracle['destination'] if index in changed else item['operation']
            if operation not in ('add', 'sub'):
                raise ValueError('unknown arithmetic operation')
            value = left + right if operation == 'add' else left - right
        else:
            raise ValueError('unknown construction-time node')
        values.append(s32(value))
    return values[oracle['output_node']]


def make_cases(oracle, rng):
    count = oracle['input_count']
    inputs = [[EDGE[(i + j) % len(EDGE)] for j in range(count)] for i in range(len(EDGE))]
    inputs += [[1 if j == i else 0 for j in range(count)] for i in range(count)]
    inputs += [[rng.randint(-2**31, 2**31 - 1) for _ in range(count)] for _ in range(8)]
    return [[values, expression_expected(oracle, values)] for values in inputs]


def branch(builder, inputs, shape, depth, required):
    b = builder
    root = b.op(inputs[0], inputs[1])
    if shape in ('copy_chain', 'rename_chain', 'chain'):
        out = b.copies(root, depth, rename=shape == 'rename_chain')
        for index in range(1, required):
            b.nops(1)
            out = b.op(out, inputs[(index + 1) % len(inputs)], out=out)
            out = b.copies(out, depth)
        return out
    if shape == 'overwrite_kill':
        # The old definition has a consumer, but its copied value is discarded.
        b.copy(root, out=b.reg())
        b.nops(1)
        b.op(inputs[1], inputs[2], out=root)
        return b.copies(root, depth)
    if shape == 'overwrite_retained':
        # The saved old definition remains live even after its register is reused.
        saved = b.copy(root, out=b.reg())
        b.nops(1)
        b.op(inputs[1], inputs[2], out=root)
        b.copy(root)
        return b.copies(saved, depth)
    if shape in ('shared_dag', 'shared_overwrite'):
        saved = b.copy(root, out=b.reg())
        doubled = b.shift(root, 1)
        if shape == 'shared_overwrite':
            b.op(inputs[2], inputs[0], out=root)
            b.copy(root)  # This overwritten, consumed definition stays dead.
            saved = b.op(saved, inputs[2], out=saved)
            b.copy(saved)
        out = b.op(saved, doubled, out=root)
        return b.copies(out, depth)
    raise ValueError('unknown branch shape')


def source_consumers(binary):
    """Ordinary last-definition reads; this diagnostic is never a model label."""
    last, consumed, source_words = {}, set(), {}
    for index in range(len(binary) // 4):
        word = int.from_bytes(binary[index * 4:index * 4 + 4], 'little')
        opcode = word & 127
        rd, left, right = (word >> 7) & 31, (word >> 15) & 31, (word >> 20) & 31
        if word == rv32.EBREAK:
            reads, write = (10,), None
        elif opcode == 0x33:
            reads, write = (left, right), rd
            source_words[index] = 'sub' if word >> 25 == 32 else 'add'
        elif opcode in (0x03, 0x13):
            reads, write = (left,), rd
        else:
            raise ValueError('unexpected generated instruction')
        for register in reads:
            if register and register in last:
                consumed.add(last[register])
        if write:
            last[write] = index
    return consumed, source_words


def make_group(rng, spec, group_index, split, tier='main', nuisance_kind=None):
    family, shape, depths, required = spec
    source, destination = ('add', 'sub') if group_index % 2 == 0 else ('sub', 'add')
    depth = rng.choice(depths)
    b = MorphBuilder(rng, rng.sample(TRAIN_REGS, len(TRAIN_REGS)), source, destination)
    inputs = [b.load() for _ in range(3)]
    padding_word = len(b.words)
    b.nops(rng.randint(1, 3))
    junk_node = len(b.nodes)
    junk_word = len(b.words)
    junk = b.immediate(inputs[2], rng.randint(-31, 31))
    b.copy(junk)
    b.nops(5)
    nuisance_kind = nuisance_kind or ('dead_value', 'irrelevant_bytes')[(group_index // 2) % 2]
    if nuisance_kind not in ('dead_value', 'irrelevant_bytes'):
        raise ValueError('unknown invariant perturbation')
    nuisance_word = junk_word if nuisance_kind == 'dead_value' else padding_word
    outputs = []
    positions = []
    for _ in range(2):
        before = len(b.nodes)
        outputs.append(branch(b, inputs, shape, depth, required))
        positions.append([item['word'] * 4 + 3 for item in b.nodes[before:]
                          if item['kind'] == 'op' and item['operation'] == source])
        b.nops(rng.randint(1, 3))
    b.nops(6)  # Return choice is outside every candidate's +/-16-byte view.
    group_id = f'{split}/{tier}/{family}/group-{group_index:03d}'
    command = rng.choice(EDIT_COMMANDS).format(a=source, b=destination)
    category = family if tier == 'main' else f'extended/{family}'
    rows = []
    for variant, out in enumerate(outputs):
        row = b.finish_morphology(out, command, category, group_id + f'/n0/v{variant}/edit')
        row.update(family=family, tier=tier, depth=depth, split=split,
                   pair_id=group_id + '/return/n0', origin_pair_id=group_id, variant=variant,
                   is_noop=False, noop_kind=None, branch_source_positions=positions,
                   branch_output_nodes=[b.reg_values[register] for register in outputs],
                   nuisance_variant=0, nuisance_kind=nuisance_kind,
                   invariance_pair_id=group_id + f'/invariance/v{variant}/edit')
        row['cases'] = make_cases(row['oracle'], rng)
        if len(row['edits']) != required:
            raise ValueError('construction produced an unexpected number of edits')
        rows.append(row)
        if variant == group_index % 2:
            noop = deepcopy(row)
            noop.update(binary=row['target'], edits=[], noop_kind='already_repaired')
            for node in noop['oracle']['changed_nodes']:
                noop['oracle']['nodes'][node]['operation'] = destination
            noop['oracle']['changed_nodes'] = []
        else:
            negative = rng.choice(NEG_COMMANDS).format(a=source, b=destination)
            noop = b.finish_morphology(out, negative, category, group_id, negate=True)
            noop.update({key: row[key] for key in (
                'family', 'tier', 'depth', 'split', 'origin_pair_id', 'variant',
                'branch_source_positions', 'branch_output_nodes', 'nuisance_variant',
                'nuisance_kind')})
            noop.update(noop_kind='negation')
        noop.update(id=group_id + f'/n0/v{variant}/noop', episode_id=group_id + f'/n0/v{variant}/noop',
                    pair_id=None, is_noop=True,
                    invariance_pair_id=group_id + f'/invariance/v{variant}/noop')
        noop['cases'] = [[values, expression_expected(noop['oracle'], values)]
                         for values, _ in row['cases']]
        rows.append(noop)
    first, second = (bytes.fromhex(rows[index]['binary']) for index in (0, 2))
    if first[:-8] != second[:-8] or first[-8:-4] == second[-8:-4]:
        raise ValueError('paired programs must differ only in returned register')
    for pos in itertools.chain.from_iterable(positions):
        if first[max(0, pos - 16):pos + 17] != second[max(0, pos - 16):pos + 17]:
            raise ValueError('a local candidate window reveals return choice')
    perturbed_rows = []
    for row in rows:
        other = deepcopy(row)
        other.update(id=row['id'].replace('/n0/', '/n1/'),
                     episode_id=row['episode_id'].replace('/n0/', '/n1/'),
                     nuisance_variant=1,
                     pair_id=group_id + '/return/n1' if row['edits'] else None)
        for field in ('binary', 'target'):
            raw = bytearray.fromhex(row[field])
            raw[nuisance_word * 4 + 2] ^= 16  # ADDI immediate bit zero.
            other[field] = bytes(raw).hex()
        if nuisance_kind == 'dead_value':
            other['oracle']['nodes'][junk_node]['immediate'] ^= 1
        if other['edits'] != row['edits']:
            raise ValueError('irrelevant perturbation changed the requested patch')
        a, b_raw = bytes.fromhex(row['binary']), bytes.fromhex(other['binary'])
        if a == b_raw:
            raise ValueError('invariance control has no input perturbation')
        for pos in itertools.chain.from_iterable(positions):
            if a[max(0, pos - 16):pos + 17] != b_raw[max(0, pos - 16):pos + 17]:
                raise ValueError('irrelevant perturbation changes a candidate local window')
        perturbed_rows.append(other)
    rows.extend(perturbed_rows)
    return rows


def generate(seed, groups_per_family, split, specs=FAMILIES, tier='main'):
    rng = random.Random(seed)
    result = []
    for family_index, spec in enumerate(specs):
        for group in range(groups_per_family):
            nuisance = ('dead_value', 'irrelevant_bytes')[(family_index + group // 2) % 2]
            result.extend(make_group(rng, spec, group, split, tier, nuisance))
    return result


def expand_subsets(row):
    """All current-state subsets for up to three public required edits."""
    edits = row['edits']
    if len(edits) > 3:
        raise ValueError('extended final tasks cannot be expanded as training states')
    states = []
    for mask in range(1 << len(edits)):
        binary = bytearray.fromhex(row['binary'])
        remaining = []
        for index, edit in enumerate(edits):
            if mask & (1 << index):
                binary[edit['offset']] = edit['value']
            else:
                remaining.append(edit)
        states.append({key: row[key] for key in ('episode_id', 'category', 'command', 'target')} |
                      dict(id=row['id'] + f'/subset-{mask}', binary=bytes(binary).hex(),
                           edits=remaining, provenance='new_public_counterexamples',
                           source='new', origin_id=row['origin_pair_id'],
                           pair_id=row.get('pair_id'), variant=row['variant'],
                           nuisance_variant=row['nuisance_variant'], state_mask=mask,
                           required_at_origin=len(edits),
                           pair_state_key=(row['pair_id'] + f'#mask={mask}') if row.get('pair_id') else None,
                           invariance_state_key=row['invariance_pair_id'] + f'#mask={mask}'))
    return states


def validate_rows(rows, execute, *, morphology=True):
    ids, executions, required_checks = set(), 0, 0
    for row in rows:
        if row['id'] in ids:
            raise ValueError('duplicate episode identifier')
        ids.add(row['id'])
        initial, target = bytes.fromhex(row['binary']), bytes.fromhex(row['target'])
        if not initial or len(initial) != len(target) or len(initial) % 4 or len(initial) > 512:
            raise ValueError('invalid program bounds')
        changes = [dict(offset=i, value=y) for i, (x, y) in enumerate(zip(initial, target)) if x != y]
        if changes != row['edits'] or not row['cases']:
            raise ValueError('incorrect byte label or missing execution cases')
        if len(changes) > 6:
            raise ValueError('too many required edits for the fixed action budget')
        original_fails = False
        for inputs, expected in row['cases']:
            if morphology and expression_expected(row['oracle'], inputs) != expected:
                raise ValueError('cases disagree with expression oracle')
            if execute(target, inputs) != expected or rv32.simulate(target, inputs) != expected:
                raise ValueError('target disagrees with one of the independent executors')
            executions += 2
            original_fails |= execute(initial, inputs) != expected
            executions += 1
        if changes and not original_fails:
            raise ValueError('edit task lacks a semantic counterexample')
        if not changes and initial != target:
            raise ValueError('invalid no-op')
        if morphology:
            if row['edits']:
                original_oracle = deepcopy(row['oracle'])
                original_oracle['changed_nodes'] = []
                for inputs, _ in row['cases']:
                    candidate_values = []
                    for node in row['branch_output_nodes']:
                        original_oracle['output_node'] = node
                        candidate_values.append(expression_expected(original_oracle, inputs))
                    if candidate_values[0] != candidate_values[1]:
                        raise ValueError('candidate values reveal the return choice')
            consumed, sources = source_consumers(initial)
            for pos in itertools.chain.from_iterable(row['branch_source_positions']):
                if pos // 4 not in consumed:
                    raise ValueError('a source candidate lacks a direct consumer')
            dead_source = [word for word, op in sources.items() if op == row['source_operation']
                           and word * 4 + 3 not in {e['offset'] for e in changes}]
            if not dead_source:
                raise ValueError('the preservation control has no unrequested source operation')
            # Every requested operation has a detectable effect when omitted from
            # an otherwise complete repair. No algebraically cancelled labels.
            for edit in changes:
                omitted = bytearray(target)
                omitted[edit['offset']] = initial[edit['offset']]
                observed = False
                for inputs, expected in row['cases']:
                    executions += 1
                    if execute(bytes(omitted), inputs) != expected:
                        observed = True
                        break
                if not observed:
                    raise ValueError('an individual required edit is semantically vacuous')
                required_checks += 1
    return dict(episodes=len(rows), execution_checks=executions,
                individually_necessary_edit_checks=required_checks,
                min_bytes=min(len(bytes.fromhex(r['binary'])) for r in rows),
                max_bytes=max(len(bytes.fromhex(r['binary'])) for r in rows))


def make_retention(old_rows):
    rng = random.Random(PUBLIC_SEEDS['retention'])
    selected = []
    for category, number in RETENTION_COUNTS.items():
        candidates = [r for r in old_rows if r['category'] == category]
        if category == 'far_pair':
            groups = defaultdict(list)
            for row in candidates:
                groups[row['pair_id']].append(row)
            if any(len(group) != 2 for group in groups.values()):
                raise ValueError('incomplete old public retention pairs')
            choices = rng.sample(sorted(groups), number // 2)
            chosen = [row for pair in choices for row in groups[pair]]
        else:
            chosen = rng.sample(candidates, number)
        for original in chosen:
            row = deepcopy(original)
            row['cases'] = old_cases_for(row, rng)
            row.update(family=category, tier='retention', split='retention',
                       is_noop=not bool(row['edits']), provenance='old_public_development')
            selected.append(row)
    return selected


def summarize_scored(episodes, scored):
    """Complete, disjoint edit/no-op accounting; no all-STOP edit credit."""
    if not episodes or len(scored) != len(episodes):
        raise ValueError('missing evaluation episodes')
    if [r['id'] for r in scored] != [r['id'] for r in episodes]:
        raise ValueError('evaluation identifiers/order differ from the committed input')
    if len({r['id'] for r in episodes}) != len(episodes):
        raise ValueError('duplicate committed episode identifier')
    groups = defaultdict(Counter)
    pairs = defaultdict(list)
    invariance = defaultdict(list)
    for episode, score in zip(episodes, scored):
        required = len(episode['edits'])
        if score['required_changes'] != required or score['category'] != episode['category']:
            raise ValueError('score metadata disagrees with episode')
        edit = required > 0
        for name in ('all', 'tier/' + episode.get('tier', 'main'), 'category/' + episode['category']):
            group = groups[name]
            group['total'] += 1
            group['edit_total' if edit else 'noop_total'] += 1
            group['edit_strict' if edit else 'noop_strict'] += bool(score['strict_success'])
            for metric in ('strict_success', 'exact_final', 'semantic_final', 'explicit_stop',
                           'false_edit_on_correct', 'proposed_edit_steps'):
                group[metric] += int(score[metric])
        if episode.get('pair_id') is not None:
            pairs[episode['pair_id']].append((episode, score))
        if episode.get('invariance_pair_id') is not None:
            invariance[episode['invariance_pair_id']].append((episode, score))
    pair_groups = defaultdict(Counter)
    for pair in pairs.values():
        if len(pair) != 2 or {e['variant'] for e, _ in pair} != {0, 1}:
            raise ValueError('incomplete return-choice pair')
        if any(not e['edits'] for e, _ in pair):
            raise ValueError('paired editing metric contains no-op tasks')
        episode = pair[0][0]
        for name in ('all', 'tier/' + episode.get('tier', 'main'), 'category/' + episode['category']):
            pair_groups[name]['pair_total'] += 1
            pair_groups[name]['pair_both_strict'] += all(s['strict_success'] for _, s in pair)
    invariant_groups = defaultdict(Counter)
    for pair in invariance.values():
        if len(pair) != 2 or {e['nuisance_variant'] for e, _ in pair} != {0, 1}:
            raise ValueError('incomplete irrelevant-perturbation pair')
        if pair[0][0]['edits'] != pair[1][0]['edits']:
            raise ValueError('invariance references request different edit sets')
        if pair[0][0]['variant'] != pair[1][0]['variant']:
            raise ValueError('invariance pair changes the requested return branch')
        deltas = []
        for episode, score in pair:
            initial, final = bytes.fromhex(episode['binary']), bytes.fromhex(score['final_hex'])
            deltas.append((len(final), tuple((i, y) for i, (x, y) in enumerate(zip(initial, final)) if x != y)))
        episode = pair[0][0]
        kind = 'edit' if episode['edits'] else 'noop'
        both = all(score['strict_success'] for _, score in pair)
        for name in ('all', 'tier/' + episode.get('tier', 'main'), 'category/' + episode['category']):
            group = invariant_groups[name]
            group['invariance_total'] += 1
            group['invariance_both_strict'] += both
            group['invariance_predicted_delta_agreement'] += deltas[0] == deltas[1]
            group['invariance_' + kind + '_total'] += 1
            group['invariance_' + kind + '_both_strict'] += both
    result = {}
    for name, values in groups.items():
        values.update(pair_groups[name])
        values.update(invariant_groups[name])
        record = dict(values)
        for numerator, denominator, rate in (
            ('edit_strict', 'edit_total', 'edit_strict_rate'),
            ('noop_strict', 'noop_total', 'noop_strict_rate'),
            ('pair_both_strict', 'pair_total', 'pair_both_strict_rate')):
            record[rate] = values[numerator] / values[denominator] if values[denominator] else None
        result[name] = record
    return result


def freeze(output):
    output = Path(output)
    output.mkdir(parents=True, exist_ok=True)
    if any(output.iterdir()):
        raise ValueError('output must be empty; refuse to replace any frozen data')
    execute = independent_executor()
    train_episodes = generate(PUBLIC_SEEDS['train'], 8, 'train')
    development = generate(PUBLIC_SEEDS['development'], 2, 'development')
    selection = generate(PUBLIC_SEEDS['selection'], 2, 'selection')
    train_new = [state for row in train_episodes for state in expand_subsets(row)]
    dev_states = [state for row in development for state in expand_subsets(row)]
    old_train_path = ROOT / 'work/architecture-20261017/data/train.jsonl'
    old_dev_path = ROOT / 'work/architecture-20261017/data/development-episodes.json'
    old_states = [json.loads(line) for line in old_train_path.read_text().splitlines()]
    replay_count = (len(train_new) + 1) // 3
    replay_indices = random.Random(PUBLIC_SEEDS['replay']).sample(range(len(old_states)), replay_count)
    replay = [dict(old_states[index], provenance='old_public_training_replay', source='replay',
                   origin_id=old_states[index].get('episode_id'), pair_id=None, variant=None,
                   nuisance_variant=None, state_mask=None, pair_state_key=None,
                   invariance_state_key=None, old_source_index=index)
              for index in replay_indices]
    train = train_new + replay
    random.Random(PUBLIC_SEEDS['mixture_order']).shuffle(train)
    retention = make_retention(json.loads(old_dev_path.read_text()))
    # Only this independent data steward reads the new cases before model freeze.
    seed = secrets.randbits(128)
    final_main = generate(seed, 2, 'final', FAMILIES)
    final_extended = generate(seed ^ (1 << 127), 2, 'final', EXTENDED, 'extended')
    final = final_main + final_extended
    sets = dict(train=train_episodes, development=development, selection=selection,
                retention=retention, final_main=final_main, final_extended=final_extended)
    validation = {name: validate_rows(rows, execute, morphology=name != 'retention')
                  for name, rows in sets.items()}
    if any(validation[name]['max_bytes'] > 256 for name in ('train', 'development', 'selection')):
        raise ValueError('public data exceeds the promised 256-byte bound')
    # Expanded states are included in overlap checks, including old replay.
    overlap_sets = dict(train=train, development=dev_states, selection=selection,
                        retention=retention, final=final)
    overlaps = {}
    for left, right in itertools.combinations(overlap_sets, 2):
        a = {r['binary'] for r in overlap_sets[left]}
        b = {r['binary'] for r in overlap_sets[right]}
        overlaps[left + '__' + right] = len(a & b)
    if any(overlaps.values()):
        raise ValueError('split overlap in exact program bytes')
    outputs = {
        'train-new.jsonl': train_new, 'train.jsonl': train,
        'train-episodes.json': train_episodes, 'development.jsonl': dev_states,
        'development-episodes.json': development, 'selection.json': selection,
        'retention.json': retention, 'sealed/final.json': final,
    }
    for relative, rows in outputs.items():
        (write_jsonl if relative.endswith('.jsonl') else write_json)(output / relative, rows)
    write_json(output / 'sealed/seed.json', dict(seed=seed, bits=128,
        origin='secrets.randbits(128); data steward only until frozen model inventory'))
    write_json(output / 'replay-manifest.json', dict(
        source=str(old_train_path.relative_to(ROOT)), source_sha256=sha(old_train_path),
        source_rows=len(old_states), selected_indices=replay_indices,
        sample_seed=PUBLIC_SEEDS['replay'], mixture_order_seed=PUBLIC_SEEDS['mixture_order'],
        new_states=len(train_new), replay_states=len(replay), total_states=len(train),
        replay_fraction=len(replay) / len(train),
        selected_ids=[r['id'] for r in replay]))
    write_json(output / 'retention-manifest.json', dict(
        source=str(old_dev_path.relative_to(ROOT)), source_sha256=sha(old_dev_path),
        seed=PUBLIC_SEEDS['retention'], counts=RETENTION_COUNTS,
        selected_ids=[r['id'] for r in retention],
        use='Public selection retention only. Never added to the training mixture.'))
    files = {str(path.relative_to(output)): sha(path) for path in sorted(output.rglob('*')) if path.is_file()}
    sources = {relative: sha(ROOT / relative) for relative in (
        'experiments/morphology_data.py', 'experiments/architecture_data.py',
        'tests/test_morphology_data.py',
        'experiments/architecture_eval.py', 'lightnest/rv32.py',
        'lightnest/iterative_patch.py', OLD_AUDITOR)}
    protocol = dict(
        created_utc=datetime.now(timezone.utc).isoformat(), base_git_commit=BASE_COMMIT,
        experiment='Counterexample-derived data for bounded inherited neural architecture search',
        public_seeds=PUBLIC_SEEDS, files=files, sources=sources,
        counts={name: dict(total=len(rows), edit=sum(bool(r['edits']) for r in rows),
                           noop=sum(not r['edits'] for r in rows),
                           pairs=len({r['pair_id'] for r in rows if r.get('pair_id')}),
                           invariance_pairs=len({r['invariance_pair_id'] for r in rows if r.get('invariance_pair_id')}),
                           categories=dict(Counter(r['category'] for r in rows)),
                           required_edits=dict(Counter(str(len(r['edits'])) for r in rows)))
                for name, rows in sets.items()},
        train_states=len(train), new_train_states=len(train_new),
        replay_states=len(replay), replay_fraction=len(replay) / len(train),
        development_states=len(dev_states), public_families=FAMILIES, extended_families=EXTENDED,
        validation=validation, exact_binary_overlap=overlaps,
        grouping='Each origin has eight rows: two returned-register variants times two irrelevant-perturbation variants times edit/no-op. At each nuisance setting only the two editing return variants share pair_id. invariance_pair_id links the two nuisance variants with identical return choice and required edits, separately for edit and no-op. All eight share origin_pair_id. No-ops are equally split between already repaired and explicit negation.',
        construction='Both independent candidate branches contain direct consumers. Last-definition overwrites, copies retaining overwritten values, shared definitions, and 1-3 public / 4-6 held-out relevant edits require preserving dead source operations. Shared DAGs use a one-bit shift to prevent algebraic cancellation.',
        invariance='Perturb either an ADDI immediate in a third never-returned computation or an ADDI x0 padding immediate. No instruction moves and all candidate +/-16-byte neighborhoods are identical. Both candidate branches retain identical input values and operations, so their original numerical outputs are equal in every paired case. The third dead computation may change value; it is never a returned candidate.',
        training='All subsets of up to three required edits are supervised as current states. Fixed old public training replay approximates 25 percent by row count. Every arm and seed reads the identical combined file; no final data or final feedback enters training.',
        training_pair_keys='Explicit source/new-or-replay, origin_id, pair_id, variant, nuisance_variant, state_mask, pair_state_key, and invariance_state_key support a reconstructible paired sampler. A return pair with mask zero differs only in return choice. Nonzero masks represent the same subset index of each program\'s own valid edits, so repaired offsets differ between returned branches; they are not claimed to differ only in return bytes. Original no-op/negation states have no return pair_state_key and must retain training exposure.',
        model_input='Only command and current bytes. Never oracle nodes, category/depth, target, edit labels, cases, pair IDs, or evaluation feedback.',
        scoring='Reuse architecture_eval.score_episode with 8 steps including explicit model STOP, total changed-byte budget 8, and no semantic validator in the loop. Strict success requires exact target, every execution case, and model STOP; any edit attempt on a correct input invalidates no-op strict credit.',
        reporting='Separate edit strict rate, no-op strict rate/false edits, returned-register pair-both strict rate, irrelevant-perturbation pair-both strict and delivered-edit-set agreement, semantic-only, exact bytes, explicit STOP. An all-STOP model can have perfect invariance agreement but zero editing credit. Report final_main 128 and final_extended 128 separately, every declared arm and seed; no best-seed/depth selection from final.',
        selection='Development is public engineering feedback. Selection and retention are public architecture-selection inputs, not fresh test claims. Coordinator must freeze its selection/promotion rule and all final checkpoint hashes before opening sealed/final.json.',
        sealed_access='Only the independent data steward generated/validated final cases. Training/model agents must not open sealed files. Coordinator executes one formal final pass after the complete model inventory is frozen.',
        limits='Straight-line RV32 add/sub tasks with LW, ADDI identity copies, SLLI scaffolding, and EBREAK; no branches, loops, aliasing, compiler optimization, arbitrary language, or real OSS repair claim.',
    )
    write_json(output / 'protocol.json', protocol)
    return dict(protocol_sha256=sha(output / 'protocol.json'),
                train_sha256=files['train.jsonl'], selection_sha256=files['selection.json'],
                retention_sha256=files['retention.json'], final_sha256=files['sealed/final.json'],
                counts={key: {k: v for k, v in value.items() if k != 'categories'}
                        for key, value in protocol['counts'].items()},
                training_states=len(train), replay_states=len(replay),
                replay_fraction=len(replay) / len(train), validation=validation,
                all_exact_binary_overlaps_zero=not any(overlaps.values()))


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--output', default='work/morphology-20261017/data')
    args = parser.parse_args()
    print(json.dumps(freeze(args.output), sort_keys=True, indent=2))