| """One-time blind evaluation of a hash-frozen native LightNest descendant. |
| |
| This program performs no training or checkpoint selection. Both native ELF |
| programs are called through their --propose inference interface. Exact output |
| actions are checked with an independent Python RV32 execution oracle. Dumped |
| native weights are also loaded into Torch to verify inference parity and that |
| the feature extractor was preserved. A required final ELF hash prevents silent |
| replacement after freeze. The pre-existing blind dataset hash is required. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import itertools |
| import json |
| import random |
| import subprocess |
| import sys |
| import time |
| from collections import Counter |
| from pathlib import Path |
|
|
| from head_learning_diagnostic import features |
| from native_optimizer_diagnostic import native_model |
|
|
| EDGE = (-2147483648, -2147483647, -1, 0, 1, 2147483646, 2147483647) |
| DATA_SHA = "a69b27443c7dfc30ba39505d2d9ce643a7ddd09c4cd2b637e7ed331cd8499fb2" |
|
|
|
|
| def digest(path): |
| return hashlib.sha256(Path(path).read_bytes()).hexdigest() |
|
|
|
|
| def signed32(x): |
| return ((x + (1 << 31)) % (1 << 32)) - (1 << 31) |
|
|
|
|
| def oracle(op, xs): |
| return signed32(xs[0] + xs[1] if op == "add" else xs[0] - xs[1]) |
|
|
|
|
| def wilson(k, n): |
| if not n: return None |
| z = 1.959963984540054 |
| center = (k/n + z*z/(2*n)) / (1 + z*z/n) |
| width = z*((k/n)*(1-k/n)/n + z*z/(4*n*n))**.5 / (1+z*z/n) |
| return [center-width, center+width] |
|
|
|
|
| def run(args): |
| sys.path.insert(0, str(args.source)) |
| import torch |
| from lightnest import rv32 |
| from lightnest.learned_patch import encode |
| torch.set_num_threads(2) |
| torch.set_num_interop_threads(1) |
| if digest(args.final) != args.final_sha256: |
| raise ValueError("final ELF differs from frozen hash") |
| if digest(args.dataset) != DATA_SHA: |
| raise ValueError("pre-existing blind dataset differs from frozen hash") |
| args.output.mkdir(parents=True, exist_ok=False) |
| cpu, wall = time.process_time(), time.monotonic() |
| paths = dict(baseline=args.baseline.resolve(), final=args.final.resolve()) |
| hashes = {k: digest(p) for k, p in paths.items()} |
| states = {k: json.loads(subprocess.run([str(p), "--state"], check=True, stdout=subprocess.PIPE).stdout) |
| for k, p in paths.items()} |
| protocol = dict(dataset=str(args.dataset.resolve()), dataset_sha256=DATA_SHA, |
| executables={k: dict(path=str(p), sha256=hashes[k], state=states[k]) for k, p in paths.items()}, |
| evaluation="one raw and one confidence-gated single-byte proposal from actual native ELF", |
| success="all independent RV32 executions correct; originally correct programs require NONE", |
| confidence_gate=.55, |
| extra_inputs=dict(signed32_edge_values=EDGE, cartesian_pairs=49, random_pairs=32, seed=1720170491), |
| oracle="infer unique add/sub truth function from sealed execution cases, then add boundary/random cases", |
| no_selection="final program fixed by hash before opening blind; no training command is invoked", |
| parity="native --propose versus Torch loaded from actual --dump-weights", |
| uncertainty="Wilson intervals are descriptive; templated cases are not independent real-world projects", |
| recorded_seen_datasets={str(p.resolve()): digest(p) for p in args.seen_data}, |
| evaluator_sha256=digest(Path(__file__))) |
| (args.output / "protocol.json").write_text(json.dumps(protocol, indent=2)) |
| (args.output / "source-snapshot.py").write_text(Path(__file__).read_text()) |
| |
| rows = json.loads(args.dataset.read_text()) |
| blind_keys = {(r["command"], r["binary_hex"]) for r in rows} |
| if len(blind_keys) != len(rows): raise ValueError("duplicate blind inputs") |
| overlap = {} |
| for path in args.seen_data: |
| seen = json.loads(path.read_text()) |
| keys = {(r["command"], r["binary_hex"]) for r in seen} |
| overlap[str(path.resolve())] = len(blind_keys & keys) |
| if any(overlap.values()): raise ValueError("blind inputs overlap recorded training or selection data") |
| rng = random.Random(1720170491) |
| windows = args.output / "windows" |
| windows.mkdir() |
| totals = Counter() |
| for i, r in enumerate(rows): |
| r["binary"] = bytes.fromhex(r["binary_hex"]) |
| possible = [op for op in ("add", "sub") if all(oracle(op, x) == y for x, y in r["cases"])] |
| if len(possible) != 1: raise ValueError("oracle is not uniquely specified by sealed cases") |
| r["oracle_operation"] = possible[0] |
| xs = list(itertools.product(EDGE, EDGE)) |
| xs += [(rng.randint(-(1 << 31), (1 << 31)-1), rng.randint(-(1 << 31), (1 << 31)-1)) for _ in range(32)] |
| r["all_cases"] = r["cases"] + [[list(x), oracle(possible[0], x)] for x in xs] |
| original = [rv32.simulate(r["binary"], x) for x, _ in r["all_cases"]] |
| r["already"] = all(y == got for (_, y), got in zip(r["all_cases"], original)) |
| r["category"] = "negation" if r["family"] == "negation" else "old_none" if r["already"] else "old_positive" |
| if r["family"] == "negation" and not r["already"]: |
| raise ValueError("negation original is not semantically correct") |
| r["window_path"] = windows / f"{i:04d}.bin" |
| r["window_path"].write_bytes(r["binary"]) |
| totals[r["category"]] += 1 |
| models, proposals, counts = {}, {}, {} |
| parity_errors = [] |
| simulator_comparisons = 0 |
| simulator_errors = [] |
| max_confidence_error = 0. |
|
|
| def record_result(r, p, v): |
| if p != 256 and not (0 <= p < len(r["binary"]) and 0 <= v <= 255): |
| raise ValueError(f"invalid native action: position={p}, byte={v}") |
| candidate = bytearray(r["binary"]) |
| if p != 256: candidate[p] = v |
| outputs = [rv32.simulate(bytes(candidate), xs) for xs, _ in r["all_cases"]] |
| mismatch = [k for k, ((_, expected), got) in enumerate(zip(r["all_cases"], outputs)) if expected != got] |
| semantic = not mismatch |
| success = semantic and (not r["already"] or p == 256) |
| failure = None |
| if mismatch: |
| k = mismatch[0] |
| failure = dict(input=r["all_cases"][k][0], expected=r["all_cases"][k][1], got=outputs[k]) |
| reason = "pass" if success else "unnecessary_proposal" if semantic else "abstained_on_needed_edit" if p == 256 else "wrong_semantics" |
| return dict(success=success, semantic=semantic, reason=reason, position=p, byte=v, |
| actual_edit=p != 256 and candidate != r["binary"], |
| execution_cases=len(outputs), failed_execution_cases=len(mismatch), first_failure=failure, |
| output_hex=candidate.hex()) |
|
|
| for name, program in paths.items(): |
| model, payload = native_model(program, torch) |
| (args.output / f"{name}-weights.bin").write_bytes(payload) |
| models[name] = model |
| h, joint = features(model, rows, encode, torch) |
| with torch.inference_mode(): |
| scores = model.pointer(h).squeeze(-1) |
| ns = torch.tensor([len(r["binary"]) for r in rows]) |
| scores.masked_fill_(torch.arange(256)[None, :] >= ns[:, None], -1e9) |
| scores = torch.cat((scores, model.none(joint)), 1) |
| probs = scores.softmax(-1) |
| bits = model.replacement(h) |
| arm = [] |
| tally = {mode: Counter() for mode in ("raw", "gated")} |
| with (args.output / f"{name}.jsonl").open("w") as out: |
| for i, r in enumerate(rows): |
| native = json.loads(subprocess.run([str(program), "--propose", r["command"], str(r["window_path"])], |
| check=True, stdout=subprocess.PIPE).stdout) |
| p = int(scores[i].argmax()) |
| v = 0 if p == 256 else sum(int(bits[i, p, k] >= 0) << k for k in range(8)) |
| g = p |
| confidence = float(probs[i, p]) |
| bit_conf = 1. |
| if p != 256: |
| bp = bits[i, p].sigmoid() |
| bit_conf = float(torch.maximum(bp, 1-bp).min()) |
| if min(confidence, bit_conf) < .55 or v == r["binary"][p]: g = 256 |
| python = dict(position=p, byte=v, gated_position=g, gated_byte=v, |
| pointer_confidence=confidence, min_bit_confidence=bit_conf) |
| if any(native[k] != python[k] for k in ("position", "byte", "gated_position", "gated_byte")): |
| parity_errors.append(dict(arm=name, index=i, native=native, torch=python)) |
| max_confidence_error = max(max_confidence_error, abs(native["pointer_confidence"]-confidence), abs(native["min_bit_confidence"]-bit_conf)) |
| result = dict(index=i, family=r["family"], category=r["category"], command=r["command"], |
| input_hex=r["binary_hex"], oracle_operation=r["oracle_operation"], |
| original_correct=r["already"], native=native, torch=python) |
| for mode, pointer_key, byte_key in (("raw", "position", "byte"), ("gated", "gated_position", "gated_byte")): |
| scored = record_result(r, native[pointer_key], native[byte_key]) |
| result[mode] = scored |
| tally[mode][r["category"]] += scored["success"] |
| tally[mode]["semantic_total"] += scored["semantic"] |
| tally[mode]["strict_total"] += scored["success"] |
| |
| |
| if i % 16 == 0: |
| for variant in ("original", "raw", "gated"): |
| b = r["binary"] if variant == "original" else bytes.fromhex(result[variant]["output_hex"]) |
| window = windows / "execution-check.bin" |
| window.write_bytes(b) |
| for xs, _ in r["all_cases"][:3]: |
| got = json.loads(subprocess.run([str(program), "--execute", str(window), str(xs[0]), str(xs[1])], |
| check=True, stdout=subprocess.PIPE).stdout) |
| expected = rv32.simulate(b, xs) |
| simulator_comparisons += 1 |
| if got != expected: simulator_errors.append(dict(arm=name, index=i, variant=variant, input=xs, native=got, python=expected)) |
| out.write(json.dumps(result) + "\n") |
| arm.append(result) |
| proposals[name] = arm |
| counts[name] = {mode: dict(tally[mode]) for mode in tally} |
| print(json.dumps(dict(arm=name, counts=counts[name], totals=totals)), flush=True) |
| diffs = {} |
| for key, x in models["baseline"].state_dict().items(): |
| y = models["final"].state_dict()[key] |
| changed = int((x != y).sum()) |
| diffs[key] = dict(changed=changed, count=x.numel(), max_abs_difference=float((x-y).abs().max())) |
| frozen_unchanged = all(v["changed"] == 0 for k, v in diffs.items() |
| if not k.startswith(("pointer.", "none.", "replacement."))) |
| paired = {} |
| intervals = {} |
| for mode in ("raw", "gated"): |
| paired[mode], intervals[mode] = {}, {} |
| for category, total in totals.items(): |
| cc = Counter() |
| for a, b in zip(proposals["baseline"], proposals["final"]): |
| if a["category"] != category: continue |
| cc[("pass" if a[mode]["success"] else "fail") + "_to_" + ("pass" if b[mode]["success"] else "fail")] += 1 |
| paired[mode][category] = dict(cc) |
| intervals[mode][category] = {arm: wilson(counts[arm][mode].get(category, 0), total) for arm in paths} |
| unchanged_hashes = all(digest(p) == hashes[k] for k, p in paths.items()) |
| report = dict(protocol=protocol, totals=dict(totals), counts=counts, paired=paired, |
| exact_input_overlap_with_recorded_seen=overlap, |
| descriptive_wilson95=intervals, parameter_changes=diffs, |
| frozen_feature_extractor_unchanged=frozen_unchanged, |
| native_torch_action_parity_errors=parity_errors, |
| max_confidence_absolute_error=max_confidence_error, |
| native_python_simulator_comparisons=simulator_comparisons, |
| native_python_simulator_errors=simulator_errors, |
| executable_hashes_unchanged=unchanged_hashes, |
| blind_evaluated=True, no_checkpoint_selection=True, |
| CPU_seconds=time.process_time()-cpu, wall_seconds=time.monotonic()-wall) |
| (args.output / "report.json").write_text(json.dumps(report, indent=2)) |
| names = dict(old_positive="旧任务：需要修改", old_none="旧任务：原已正确", negation="否定指令：必须 NONE") |
| md = ["# 冻结原生子代的独立盲测", "", |
| "本次评测在最终 ELF 的 SHA-256 冻结后开启。没有训练、调整阈值或用测试结果选择检查点。", "", |
| f"共 {len(rows)} 个独立例子；每例使用 {len(rows[0]['all_cases'])} 个执行输入，其中包括原冻结判据、49 个边界组合和32个新增随机输入。", "", |
| "| 输出方式 | 任务 | 父代 | 最终子代 | 新成功 | 回归失败 |", "|---|---|---:|---:|---:|---:|"] |
| for mode in ("raw", "gated"): |
| for category, total in totals.items(): |
| pair = paired[mode][category] |
| md.append(f"| {mode} | {names[category]} | {counts['baseline'][mode].get(category, 0)}/{total} | {counts['final'][mode].get(category, 0)}/{total} | {pair.get('fail_to_pass', 0)} | {pair.get('pass_to_fail', 0)} |") |
| md += ["", "## 独立核验", "", |
| f"- 实际 C 提案与导出权重的 Torch 提案不一致：{len(parity_errors)}。", |
| f"- C 与 Python RV32 执行对照：{simulator_comparisons} 次；不一致 {len(simulator_errors)} 次。", |
| f"- 冻结编码器保持不变：{frozen_unchanged}。", |
| f"- 评测前后 ELF 哈希保持不变：{unchanged_hashes}。", |
| f"- 在已提供训练/开发文件中发现的完全重复输入：{sum(overlap.values())}。", "", |
| "## 解释范围", "", |
| "原已正确的任务必须明确选择 NONE 才计严格成功。低置信度门控可提高不修改类任务的成功率，同时也可能拒绝正确修改，因此 raw 与门控分别报告。", "", |
| "这些结果属于短 RV32 加减法字节修改与否定指令。即使通过所有列出的有限执行输入，也不是对所有输入的形式化证明。模板相关的例子不能当作独立抽取的真实软件项目；本报告不据此主张一般自然语言理解、复杂工程开发或超过开源项目。", "", |
| "详细失败动作、置信度、首个执行反例、参数差异和父子成对转移见 JSONL 与 report.json。", ""] |
| (args.output / "RESULTS.md").write_text("\n".join(md)) |
| if not unchanged_hashes or not frozen_unchanged or parity_errors or simulator_errors: |
| raise RuntimeError("independent verification mismatch; report retained for diagnosis") |
|
|
|
|
| if __name__ == "__main__": |
| p = argparse.ArgumentParser() |
| p.add_argument("--source", type=Path, required=True) |
| p.add_argument("--baseline", type=Path, required=True) |
| p.add_argument("--final", type=Path, required=True) |
| p.add_argument("--final-sha256", required=True) |
| p.add_argument("--dataset", type=Path, required=True) |
| p.add_argument("--seen-data", type=Path, nargs="*", default=[]) |
| p.add_argument("--output", type=Path, required=True) |
| run(p.parse_args()) |
|
|
