"""Frozen instruction-relation and sequential-edit experiment data. Labels come from a generated expression DAG and the independent assembler. The predictor never receives this DAG, target bytes, execution tests or labels. This module is an experiment generator, not an inference fallback. """ from __future__ import annotations import argparse from collections import Counter from datetime import datetime, timezone import hashlib 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 lightnest import rv32 TRAIN_REGS = tuple(i for i in range(1, 20) if i != 10) HELD_REGS = tuple(range(20, 32)) EDIT_COMMANDS = ( 'change {a} to {b}', 'replace {a} with {b} in the returned computation', 'change the output-producing {a} to {b}', 'change every {a} that contributes to the output to {b}', ) NEW_COMMANDS = ( 'please replace returned {a} operations with {b}', 'switch the {a} operations that determine the output to {b}', ) NEG_COMMANDS = ( 'do not change {a} to {b}', 'do not replace {a} with {b} in the returned computation', ) EDGE = (-2147483648, -1, 0, 1, 2147483647) MAIN_CATEGORIES = ('simple_edit', 'far_pair', 'two_edits', 'three_edits', 'shared_dag', 'overwrite', 'already_correct', 'negation', 'dead_only', 'new_register', 'new_wording') EXTENDED_CATEGORIES = ('four_edits', 'six_edits', 'long_program') def digest_bytes(data): return hashlib.sha256(data).hexdigest() def write_json(path, value): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False) + '\n') def pack(words): return b''.join(word.to_bytes(4, 'little') for word in words) def s32(value): return (value + 2**31) % 2**32 - 2**31 class Builder: """Keep expression nodes separate from generated instruction bytes.""" def __init__(self, rng, regs, source, destination, gap=(1, 4)): self.rng, self.regs = rng, iter(regs) self.source, self.destination, self.gap = source, destination, gap self.words, self.nodes, self.reg_values = [], [], {} self.input_count = 0 def reg(self): return next(self.regs) def nops(self, count=None): if count is None: count = self.rng.randint(*self.gap) self.words += [rv32.addi('x0', 'x0', self.rng.randint(-2048, 2047)) for _ in range(count)] def load(self): reg = self.reg() node = len(self.nodes) self.nodes.append({'kind': 'input', 'index': self.input_count}) self.words.append(rv32.lw(f'x{reg}', self.input_count * 4, 'x0')) self.input_count += 1 self.reg_values[reg] = node return reg def op(self, left, right, out=None, operation=None): out = self.reg() if out is None else out operation = operation or self.source node = len(self.nodes) self.nodes.append({'kind': 'op', 'operation': operation, 'left': self.reg_values[left], 'right': self.reg_values[right], 'word': len(self.words), 'registers': [out, left, right]}) self.words.append(getattr(rv32, operation)(f'x{out}', f'x{left}', f'x{right}')) self.reg_values[out] = node return out def finish(self, out, command, category, episode_id, negate=False): output_node = self.reg_values[out] live = set() def visit(node): if node in live: return live.add(node) item = self.nodes[node] if item['kind'] == 'op': visit(item['left']); visit(item['right']) visit(output_node) words = self.words + [rv32.addi('x10', f'x{out}', 0), rv32.EBREAK] original, targets = pack(words), list(words) changed_nodes = [] 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'] targets[item['word']] = getattr(rv32, self.destination)(f'x{rd}', f'x{left}', f'x{right}') changed_nodes.append(index) target = pack(targets) oracle = {'nodes': self.nodes, 'output_node': output_node, 'changed_nodes': changed_nodes, 'destination': self.destination, 'input_count': self.input_count} return {'id': episode_id, 'episode_id': episode_id, 'category': category, 'command': command, 'binary': original.hex(), 'target': target.hex(), 'edits': [{'offset': i, 'value': y} for i, (x, y) in enumerate(zip(original, target)) if x != y], 'binary_sha256': digest_bytes(original), 'oracle': oracle, 'source_operation': self.source, 'destination_operation': self.destination} def expected(oracle, inputs): """Independent arithmetic over construction-time expression nodes.""" values = [] changes = set(oracle['changed_nodes']) for index, node in enumerate(oracle['nodes']): if node['kind'] == 'input': value = inputs[node['index']] else: operation = oracle['destination'] if index in changes else node['operation'] left, right = values[node['left']], values[node['right']] value = left + right if operation == 'add' else left - right values.append(s32(value)) return values[oracle['output_node']] def cases_for(row, rng): count = row['oracle']['input_count'] xs = [[EDGE[(i + j) % len(EDGE)] for j in range(count)] for i in range(len(EDGE))] xs += [[(1 << min(j, 30)) if j == i else 0 for j in range(count)] for i in range(count)] xs += [[rng.randint(-2**31, 2**31 - 1) for _ in range(count)] for _ in range(16)] return [[values, expected(row['oracle'], values)] for values in xs] def make_episode(rng, category, episode_id, *, split='train'): source, destination = rng.choice((('add', 'sub'), ('sub', 'add'))) regs = rng.sample(HELD_REGS if category == 'new_register' else TRAIN_REGS, len(HELD_REGS if category == 'new_register' else TRAIN_REGS)) command = rng.choice(NEW_COMMANDS if category == 'new_wording' else EDIT_COMMANDS).format(a=source, b=destination) gap = (9, 12) if category == 'far_pair' else (1, 4) b = Builder(rng, regs, source, destination, gap) if category == 'far_pair': left, right = b.load(), b.load() b.nops(6) first = b.op(left, right) first_pos = (len(b.words) - 1) * 4 + 3 b.nops() second = b.op(left, right) second_pos = (len(b.words) - 1) * 4 + 3 b.nops() rows = [b.finish(out, command, category, episode_id + f'/variant-{i}') for i, out in enumerate((first, second))] for i, row in enumerate(rows): row.update(pair_id=episode_id, variant=i, candidate_positions=[first_pos, second_pos]) # Paired program differences are beyond each local candidate view. raw = [bytes.fromhex(row['binary']) for row in rows] for pos in (first_pos, second_pos): assert raw[0][pos - 16:pos + 17] == raw[1][pos - 16:pos + 17] return rows depth = {'two_edits': 2, 'three_edits': 3, 'four_edits': 4, 'six_edits': 6, 'long_program': 3, 'new_register': 2, 'new_wording': 2}.get(category, 1) if category == 'shared_dag': depth = 3 inputs = [b.load() for _ in range(max(depth + 1, 3))] b.nops() dead = b.op(inputs[-1], inputs[0]) b.nops() first_out = b.reg() if category == 'overwrite': b.op(inputs[0], inputs[1], out=first_out) b.nops(10) actual = destination if category in ('already_correct', 'dead_only') else source out = b.op(inputs[0], inputs[1], out=first_out, operation=actual) shared = out for level in range(1, depth): b.nops() b.op(inputs[-1], inputs[0], out=dead) b.nops() # Shared DAG has one value used along two paths; chain reuses one register. if category == 'shared_dag': if level == 1: out = b.op(out, inputs[2], operation=actual) else: out = b.op(out, shared, operation=actual) else: out = b.op(out, inputs[level + 1], out=out, operation=actual) if category == 'dead_only': b.nops(8) b.op(inputs[0], inputs[1], out=dead) if category == 'long_program': # Length extension changes offsets without adding target labels. b.nops(max(0, 126 - len(b.words))) else: b.nops(9 if category == 'overwrite' else None) negate = category == 'negation' if negate: command = rng.choice(NEG_COMMANDS).format(a=source, b=destination) return [b.finish(out, command, category, episode_id, negate=negate)] def expand_states(row, rng): """Supervised current-state examples, including valid intermediates and STOP.""" binary = bytearray.fromhex(row['binary']) remaining = list(row['edits']) order = remaining.copy() rng.shuffle(order) result = [] for step in range(len(order) + 1): result.append({key: row[key] for key in ('episode_id', 'category', 'command', 'target')} | {'id': row['id'] + f'/state-{step}', 'binary': bytes(binary).hex(), 'edits': list(remaining)}) if step < len(order): edit = order[step] binary[edit['offset']] = edit['value'] remaining.remove(edit) return result def generate_training(seed, episodes, split): rng = random.Random(seed) categories = ('simple_edit', 'far_pair', 'two_edits', 'three_edits', 'shared_dag', 'overwrite', 'already_correct', 'negation', 'dead_only') rows, originals = [], [] for i in range(episodes): category = categories[i % len(categories)] for row in make_episode(rng, category, f'{split}/{i:05d}', split=split): # Spot-independent semantic checks on ALL construction-time targets. for xs, value in cases_for(row, rng)[:5]: assert rv32.simulate(bytes.fromhex(row['target']), xs) == value originals.append(row) rows.extend(expand_states(row, rng)) return rows, originals def write_jsonl(path, rows): path.parent.mkdir(parents=True, exist_ok=True) with path.open('w') as out: for row in rows: out.write(json.dumps(row, sort_keys=True, separators=(',', ':'), allow_nan=False) + '\n') def freeze(output, train_episodes=1440, dev_episodes=180): output = Path(output) if (output / 'data-protocol.json').exists(): raise ValueError('refuse to overwrite an existing frozen experiment') train, train_originals = generate_training(817301, train_episodes, 'train') dev, dev_originals = generate_training(817303, dev_episodes, 'development') write_jsonl(output / 'data/train.jsonl', train) write_jsonl(output / 'data/development.jsonl', dev) write_json(output / 'data/development-episodes.json', dev_originals) # New final seed is private until all checkpoints are fixed. It never enters training. seed = secrets.randbits(128) rng = random.Random(seed) final = [] for category in MAIN_CATEGORIES + EXTENDED_CATEGORIES: n = 32 if category == 'far_pair' else 48 if category in EXTENDED_CATEGORIES else 64 for i in range(n): rows = make_episode(rng, category, f'final/{category}/{i:03d}', split='final') for row in rows: row['cases'] = cases_for(row, rng) for xs, value in row['cases']: if rv32.simulate(bytes.fromhex(row['target']), xs) != value: raise ValueError('target disagrees with independent arithmetic oracle') final.append(row) write_json(output / 'sealed/final.json', final) write_json(output / 'sealed/seed.json', {'seed': seed}) train_keys = {(r['command'], r['binary']) for r in train} dev_keys = {(r['command'], r['binary']) for r in dev} final_keys = {(r['command'], r['binary']) for r in final} train_binary = {r['binary'] for r in train} dev_binary = {r['binary'] for r in dev} final_binary = {r['binary'] for r in final} overlap = {'train_development_inputs': len(train_keys & dev_keys), 'train_final_inputs': len(train_keys & final_keys), 'development_final_inputs': len(dev_keys & final_keys), 'train_final_binary': len(train_binary & final_binary), 'development_final_binary': len(dev_binary & final_binary)} if any(overlap.values()): raise ValueError('unexpected exact overlap') protocol = { 'created_utc': datetime.now(timezone.utc).isoformat(), 'base_git_commit': '4945313eab3f3e20c2b71a0ee0e079402d148ac3', 'task': 'Change add/sub operations in the returned computation; preserve dead operations; obey negation.', 'main_categories': list(MAIN_CATEGORIES), 'extended_categories': list(EXTENDED_CATEGORIES), 'final_counts': dict(Counter(r['category'] for r in final)), 'training_episodes_requested': train_episodes, 'training_programs': len(train_originals), 'training_states': len(train), 'development_programs': len(dev_originals), 'development_states': len(dev), 'training_labels': 'All valid remaining edit locations and absolute replacement bytes; empty list means STOP.', 'training_scope': 'Generated straight-line RV32 add/sub programs with load and copy; 1-3 relevant edits.', 'final_scope': 'New random programs, register holdout, wording holdout, 4/6 edits and 512-byte extension.', 'new_registers': list(HELD_REGS), 'training_registers': list(TRAIN_REGS), 'final_evaluation': 'Only after final model hashes are fixed; evaluate both 4 and 8 message rounds without choosing the best.', 'max_actions': 8, 'scoring': 'Exact target bytes AND all independent execution cases AND explicit model STOP; also report semantic-only and raw one-step.', 'inference_verifier': 'Structure only, no hidden expected output or edit labels; final tests never guide candidate choice.', 'old_capacity_640': 'Public regression set, not a fresh final evaluation.', 'overlap': overlap, 'files': {str(p.relative_to(output)): digest_bytes(p.read_bytes()) for p in ( output / 'data/train.jsonl', output / 'data/development.jsonl', output / 'data/development-episodes.json', output / 'sealed/final.json', output / 'sealed/seed.json')}, 'generator_sha256': digest_bytes(Path(__file__).read_bytes()), } write_json(output / 'data-protocol.json', protocol) return protocol if __name__ == '__main__': p = argparse.ArgumentParser() p.add_argument('--output', default='work/architecture-20261017') p.add_argument('--train-episodes', type=int, default=1440) p.add_argument('--dev-episodes', type=int, default=180) a = p.parse_args() print(json.dumps(freeze(a.output, a.train_episodes, a.dev_episodes), indent=2))