From 26ccde014218f7d906d1251dc295fbc1437ede8a Mon Sep 17 00:00:00 2001 From: JoyboyBrian Date: Thu, 22 Oct 2026 11:36:43 -0700 Subject: [PATCH] Execute one transported CRT repair key on binary tapes --- research/crt-transported-key-tape/README.md | 172 ++ research/crt-transported-key-tape/RESULT.json | 1987 +++++++++++++++++ .../crt-transported-key-tape/SOURCES.json | 36 + research/crt-transported-key-tape/TASK.json | 92 + research/crt-transported-key-tape/VENDOR.json | 8 + research/crt-transported-key-tape/VERIFY.json | 22 + research/crt-transported-key-tape/engine.py | 224 ++ .../vendor/pr4716-caller.py | 270 +++ research/crt-transported-key-tape/verify.py | 208 ++ 9 files changed, 3019 insertions(+) create mode 100644 research/crt-transported-key-tape/README.md create mode 100644 research/crt-transported-key-tape/RESULT.json create mode 100644 research/crt-transported-key-tape/SOURCES.json create mode 100644 research/crt-transported-key-tape/TASK.json create mode 100644 research/crt-transported-key-tape/VENDOR.json create mode 100644 research/crt-transported-key-tape/VERIFY.json create mode 100644 research/crt-transported-key-tape/engine.py create mode 100644 research/crt-transported-key-tape/vendor/pr4716-caller.py create mode 100644 research/crt-transported-key-tape/verify.py diff --git a/research/crt-transported-key-tape/README.md b/research/crt-transported-key-tape/README.md new file mode 100644 index 0000000000..a552817b33 --- /dev/null +++ b/research/crt-transported-key-tape/README.md @@ -0,0 +1,172 @@ +# A transported CRT key on binary tapes + +**Fixed finite implementation and tape-action census; no new κ or uniform +machine-time bound.** This takes the physical-key handoff in +[PR4716](https://github.com/CrocSwap/integer-mult-bounds/pull/4716), exact head +`d9913fe24f119f9d431f324dbbe7d90de409a856`. Its unchanged caller is retained in +[`vendor/pr4716-caller.py`](vendor/pr4716-caller.py), with origin/hash in +[`VENDOR.json`](VENDOR.json). The three main-branch reference modules are pinned +in [`SOURCES.json`](SOURCES.json). No cached foreign Git object is needed. + +The new [`engine.py`](engine.py) evaluates one literal exceptional key +`T(S^-1(q))`. Current coordinates are copied anew at each of its three source +queries. CRT reconstruction, modular updates and lookup selection use binary +tape reads and bit flags. A complete 544-bit read-only source table is scanned +at every query; no dynamic data word indexes a Python table. Work and scratch +are erased and every head returns. The program never invokes the predecessor +oracle or converts a runtime data word to a Python integer. + +## Exact boundary + +This is PR4716's transported root with widths `(4,4,5,5)`, moduli +`(11,13,17,19)`, root target axis 2, reservoir axis 3, row 6, +`H=G=1`, `K=5`. The geometry is below production constants. There are 17 active +Word bits `(u,t,y)` with `y` of width 15 and one back spectator. The input and +output are the actual 18-bit **MSB-first** address encoding +`u,t,y14,...,y0,back`; the spectator is physically copied and preserved. + +Five fixed tape roles are reused: protected input (18 bits), protected ROM +(544 bits), owned output (18 bits), private work (18 bits) and private scratch +(64 bits). Output, work and scratch start clean. No arbitrary dirty-scratch +contract is claimed. Each tape has a fixed bounded extent and begins with its +head at cell 0. Each read, write and one-cell move is an event. `seek` emits +every intervening unit move; it does not jump over cells for free. + +Pre-admission validation of these fixed lengths/bit symbols, Python object +allocation, parameter/descriptor compilation, modular-inverse setup, ROM +generation and observer logging are outside the invocation census. The +read-only ROM must already contain the source-authenticated row-6 tables for +all 17 offsets. It is an explicit borrowed input, not a free general-purpose +data oracle. This does **not** execute PR4192's uniform arena compiler or +adopt PR4650's all-parameter tariff. + +The Python program represents the fixed loop schedule and Boolean finite +control. Exact event totals exclude control-instruction time. Thus they are +**tape actions**, not total bit steps, wall-clock predictions or a bound for +arbitrary `p,ell`. In particular this is not a certificate of +`E(p,ell) = 2^(4 ell)(p+2)^16`. + +## Why the local arithmetic is the intended key + +The source-bound physical descriptors, expressed as little-endian bit +positions in current `y`, are + +``` +a: (12,0,1,2) b: (3,4,5,6) target: (7,8,9,10,11). +``` + +The verifier derives them from the retained placement, including the +transported high bit; they are not selected from a successful witness. In the +MSB-first work tape `y[j]` is cell `16-j`. + +For valid `a<11,b<13`, the retained cached CRT identity specializes to +`k=(78a+11b) mod143`, then `offset=5k mod17`. Invalid padded left coordinates +disable the shift by using offset zero. The verifier checks the constants +against the original modulus-only setup. `add_constant` is the binary full +adder. `remainder` shifts one input bit into the remainder and subtracts the +fixed modulus if necessary; its invariant is `0<=r> bit) & 1 + register.write(bit, a ^ b ^ carry) + carry = (a & b) | (a & carry) | (b & carry) + + +def less_constant(register, constant): + less, equal = 0, 1 + for bit in reversed(range(register.width)): + a, b = register.read(bit), (constant >> bit) & 1 + less |= equal & (1 - a) & b + equal &= 1 - (a ^ b) + return less + + +def multiply_constant(source, constant, destination): + destination.clear() + for bit in range(source.width): + if source.read(bit): + add_constant(destination, constant << bit) + + +def remainder(source, modulus, destination): + """Binary long division with one bit of slack: 2*remainder+bit fits.""" + need(2 * modulus <= 1 << destination.width, "remainder register slack") + destination.clear() + for bit in reversed(range(source.width)): + for j in reversed(range(1, destination.width)): + destination.write(j, destination.read(j - 1)) + destination.write(0, source.read(bit)) + if not less_constant(destination, modulus): + add_constant(destination, -modulus) + + +def copy_register(source, destination): + need(source.width == destination.width, "equal copy widths") + for bit in range(source.width): + destination.write(bit, source.read(bit)) + + +class KeyMachine: + """Fixed 18-bit interface: u,t,y[14..0],back; all five tapes are reused. + + Input and ROM are borrowed, output must initially be clean, and work/scratch + are clean private regions. The back bit is a copied spectator. ROM consists + of 17 rows of 32 bits for the fixed Beneš layer. Its compilation is setup. + """ + def __init__(self, address_bits, rom, coordinate_bits, mode="correct"): + need(len(address_bits) == 18 and len(rom) == 544, "fixed fixture interface") + need(tuple(map(len, coordinate_bits)) == (4, 4, 5), "complete descriptors") + need(all(0 <= p < 15 for row in coordinate_bits for p in row), "y-only descriptors") + need(mode in ("correct", "stale_second", "stale_third", "drop_high", + "omit_clear", "omit_return", "bad_weight", "invalid_left_enabled"), + "known fault mode") + self.events, self.snapshots = [], [] + self.input = Tape("input", address_bits, self.events, True) + self.rom = Tape("rom", rom, self.events, True) + self.output = Tape("output", [0] * 18, self.events) + self.work = Tape("work", [0] * 18, self.events) + self.scratch = Tape("scratch", [0] * 64, self.events) + self.tapes = (self.input, self.rom, self.output, self.work, self.scratch) + self.coordinate_bits, self.mode = coordinate_bits, mode + self.a = Register(self.scratch, 0, 4) + self.b = Register(self.scratch, 4, 4) + self.target = Register(self.scratch, 8, 5) + self.acc = Register(self.scratch, 13, 12) + self.rem = Register(self.scratch, 25, 9) + self.product = Register(self.scratch, 34, 12) + self.offset = Register(self.scratch, 46, 6) + self.saved = Register(self.scratch, 52, 6) + self.query_count = 0 + + def phase(self, name): + for tape in self.tapes: + tape.phase = name + + def query(self): + self.query_count += 1 + epoch = self.query_count + self.phase("query-%d-descriptors" % epoch) + for register, positions in zip((self.a, self.b, self.target), self.coordinate_bits): + for bit, ybit in enumerate(positions): + register.write(bit, self.work.read(16 - ybit)) + self.phase("query-%d-arithmetic" % epoch) + valid = less_constant(self.a, 11) & less_constant(self.b, 13) + self.acc.clear() + if valid or self.mode == "invalid_left_enabled": + for bit in range(4): + if self.a.read(bit): + add_constant(self.acc, (77 if self.mode == "bad_weight" else 78) << bit) + if self.b.read(bit): + add_constant(self.acc, 11 << bit) + remainder(self.acc, 143, self.rem) + multiply_constant(self.rem, 5, self.product) + remainder(self.product, 17, self.offset) + if self.mode in ("stale_second", "stale_third"): + if epoch == 1: + copy_register(self.offset, self.saved) + elif (self.mode == "stale_second" and epoch == 2 + or self.mode == "stale_third" and epoch == 3): + copy_register(self.saved, self.offset) + self.phase("query-%d-rom-scan" % epoch) + answer, seen = 0, 0 + # Both loops are fixed control; no data word becomes a Python index. + for fixed_offset in range(17): + for fixed_target in range(32): + match = 1 + for bit in range(6): + match &= 1 - (self.offset.read(bit) ^ ((fixed_offset >> bit) & 1)) + for bit in range(5): + match &= 1 - (self.target.read(bit) ^ ((fixed_target >> bit) & 1)) + need(not (seen & match), "unique ROM match") + seen |= match + answer |= match & self.rom.read(32 * fixed_offset + fixed_target) + need(seen == 1, "complete ROM match") + # Observer-only snapshots are never consulted by the tape program. + self.snapshots.append((tuple(self.work.cells), tuple(self.scratch.cells), answer)) + self.phase("query-%d-cleanup" % epoch) + for register in (self.a, self.b, self.target, self.acc, self.rem, self.product, self.offset): + register.clear() + self.rom.seek(0) + self.scratch.seek(0) + self.work.seek(0) + return answer + + def change_y(self, start, enabled, negative): + carry = enabled + stop = 10 if self.mode == "drop_high" else 15 + for bit in range(start, stop): + previous = self.work.read(16 - bit) + self.work.write(16 - bit, previous ^ carry) + carry &= previous ^ negative + + def run(self): + for bit in range(18): + self.work.write(bit, self.input.read(bit)) + for step in reversed(range(10)): + self.phase("inverse-step-%d" % step) + if step in (9, 4): + value = self.query() + self.phase("inverse-step-%d" % step) + self.work.write(0, self.work.read(0) ^ value) + elif step in (8, 3): + self.work.write(1, self.work.read(1) ^ self.work.read(7) ^ self.work.read(0)) + elif step in (7, 2): + self.change_y(9, self.work.read(0), 1 - self.work.read(1)) + elif step in (6, 1): + self.work.write(1, self.work.read(1) ^ self.work.read(7)) + else: + self.change_y(10, self.work.read(0) & self.work.read(1), 1) + value = self.query() + self.phase("ideal-switch") + self.work.write(7, self.work.read(7) ^ value) + self.phase("copy-output") + for bit in range(18): + self.output.write(bit, self.work.read(bit)) + self.phase("final-cleanup") + if self.mode != "omit_clear": + for tape in (self.work, self.scratch): + for bit in range(len(tape.cells)): + tape.write(bit, 0) + if self.mode != "omit_return": + for tape in self.tapes: + tape.seek(0) + return tuple(self.output.cells) diff --git a/research/crt-transported-key-tape/vendor/pr4716-caller.py b/research/crt-transported-key-tape/vendor/pr4716-caller.py new file mode 100644 index 0000000000..641ce5a6fa --- /dev/null +++ b/research/crt-transported-key-tape/vendor/pr4716-caller.py @@ -0,0 +1,270 @@ +"""Current-state CRT controls with modulus-only inverses hoisted explicitly. + +FixedCRT(left_moduli, target_modulus, right_prefix=1, inverse=...) prepares m+1 inverses +once, then evaluate(residues) performs no modular inversion. RootCaller binds +that evaluator to one actual balanced-tree root axis, transported through the +three disjoint exchanges of the retained reservoir construction. Callers pass +already authenticated crt, allocator and Word modules; this module loads no +files, prints nothing, and performs no I/O. + +The root fixture uses widths (4,4,5,5), moduli (11,13,17,19), target axis 2, +reservoir axis 3, L=18, and H=G=1, K=5. These are proper declared widths for +the prime moduli, but the geometry is below the production size cutoff. The +fixture tests actual CRT control semantics, not the production constants or +a machine-time bound. Optional memoization is by the CURRENT y argument. + +Cinna Davis supplies the CRT/network construction; PR4169/4174 supply the +inverse and placement interfaces; PR4377/4409 motivate the query-budget +audit. Hiroki KUMAZAKI (@kumagi), with substantial OpenAI Codex assistance. +Apache-2.0. See README and PROOF for the complete attribution and limits. +""" +from __future__ import annotations + +from dataclasses import asdict, dataclass +from math import prod + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def modular_inverse(value, modulus): + return pow(value, -1, modulus) + + +@dataclass(frozen=True) +class CRTValue: + residues: tuple[int, ...] + valid: bool + k_A: int + offset: int + + +class FixedCRT: + """Locally scaled CRT reconstruction for fixed pairwise-coprime moduli. + + If P_i is the product of the left moduli preceding i, the current input + residue is k_A/P_i modulo s_i. Each term therefore uses + ((residue_i * P_i) mod s_i) * (S/s_i) * (S/s_i)^(-1) mod s_i. + The target offset is k_A * (S*right_prefix)^(-1) modulo target_modulus. + + setup_inverse_calls, evaluation_calls and reconstruction_calls are host + operation counters, not bit-step costs. The first counts exactly the + actual modular-inverse provider calls below. The optional inverse keyword + accepts a two-argument provider; by default it is pow(value,-1,modulus). + evaluate invokes no inverse helper. The setup inversions themselves check + that the required fixed moduli and target prefix are coprime. + Invalid padded residues disable the node's shift, as in CRT.md section 1. + """ + + def __init__(self, left_moduli, target_modulus, right_prefix=1, *, inverse=None): + self.left_moduli = tuple(left_moduli) + require(self.left_moduli and all(type(m) is int and m >= 2 for m in self.left_moduli), + "nonempty left group of integer moduli at least two") + require(type(target_modulus) is int and target_modulus >= 2, "positive target modulus") + require(type(right_prefix) is int and right_prefix >= 1, "positive right-child prefix") + inverse = modular_inverse if inverse is None else inverse + require(callable(inverse), "modular inverse implementation must be callable") + self.target_modulus = target_modulus + self.right_prefix = right_prefix + self.product = prod(self.left_moduli) + self.setup_inverse_calls = 0 + self.evaluation_calls = 0 + self.reconstruction_calls = 0 + self.terms = [] + prefix = 1 + for modulus in self.left_moduli: + complement = self.product // modulus + coefficient_inverse = inverse(complement, modulus) + self.setup_inverse_calls += 1 + require(type(coefficient_inverse) is int + and complement * coefficient_inverse % modulus == 1, + "left complements must be invertible modulo their own moduli") + self.terms.append((modulus, prefix, complement, coefficient_inverse)) + prefix *= modulus + self.terms = tuple(self.terms) + self.prefixes = tuple(term[1] for term in self.terms) + self.complements = tuple(term[2] for term in self.terms) + self.target_inverse = inverse(self.product * right_prefix, target_modulus) + self.setup_inverse_calls += 1 + require(type(self.target_inverse) is int + and self.product * right_prefix * self.target_inverse % target_modulus == 1, + "target prefix must be invertible") + + def evaluate(self, residues): + """Read current residues and return the node's current valid/offset state.""" + residues = tuple(residues) + require(len(residues) == len(self.left_moduli) + and all(type(value) is int and value >= 0 for value in residues), + "one nonnegative current residue per left modulus") + self.evaluation_calls += 1 + valid = all(value < modulus for value, modulus in zip(residues, self.left_moduli)) + if not valid: + return CRTValue(residues, False, 0, 0) + self.reconstruction_calls += 1 + total = 0 + for value, (modulus, prefix, complement, inverse) in zip(residues, self.terms): + total += ((value * prefix) % modulus) * complement * inverse + integer = total % self.product + return CRTValue(residues, True, integer, + integer * self.target_inverse % self.target_modulus) + + def check_reference(self, value, crt): + """Verification-only comparison with the passed retained decode_group.""" + reference = crt.decode_group(value.residues, self.left_moduli) if value.valid else 0 + require(reference == value.k_A, "fixed-inverse reconstruction differs from source") + require(value.offset == reference * self.target_inverse % self.target_modulus, + "fixed-inverse target offset differs from source") + return True + + +@dataclass(frozen=True) +class Query: + """One source-function evaluation at its actual current y state. + + target is the reconstructed target-axis value. bit is the Boolean source + output. local_bit and target_y_bit identify the physical switch target. + coordinates contains the current axes 0, 1 and 2; omitted axis 3 is dirty. + """ + + current_y: int + row: int + coordinates: tuple[int, int, int] + residues: tuple[int, int] + valid: bool + k_A: int + offset: int + target: int + local_bit: int + target_y_bit: int + bit: int + + def record(self): + return asdict(self) + + +class RootCaller: + """Fixed real CRT root-axis fixture, with exact transported coordinates. + + root = RootCaller(crt_module, allocator_module, word_module, inverse=...) + root.rows # local rows having a complete compact guard + root.word(row) # retained Word with the transported target + root.evaluate(row, current_y) # Query with complete current-state trace + root.source(row, trace, cache) # callable y -> (control_bit,) + root.stale_offset_source(row,c) # deliberately wrong fixed-offset callback + + source(..., cache=True) memoizes Query values by its current y argument; + it does not retain the first offset for later, different addresses. Such + finite memoization is a replay accelerator and is not priced as a runtime + algorithm. The source's networks are cached by their actual offset. + """ + + widths = (4, 4, 5, 5) + moduli = (11, 13, 17, 19) + ell = 5 + target_axis = 2 + reservoir_axis = 3 + H = 1 + G = 1 + K = 5 + + def __init__(self, crt, allocator, word_module, *, inverse=None): + self.crt = crt + self.Word = word_module.Word + self.L = sum(self.widths) + self.Y = self.L - 3 * self.H + self.placement = allocator.Placement(self.L, self.H, ((2, 3), (3, 4), (4, 5))) + axes = [] + end = self.L + for width in self.widths: + axes.append((end - width, end)) + end -= width + self.axes = tuple(axes) + self.descriptors = tuple(self.placement.map_interval(*interval) for interval in self.axes) + self._read_parts = tuple(tuple((self.L - part.image_interval[1], + part.image_interval[1] - part.image_interval[0]) + for part in fragments) + for fragments in self.descriptors[:3]) + self.fixed = FixedCRT(self.moduli[:2], self.moduli[self.target_axis], inverse=inverse) + self._networks = {} + self.schedule = tuple(bit for bit, table in self.network(0)) + self.rows = tuple(row for row in range(len(self.schedule)) + if self.target_y_bit(row) + self.K <= self.Y) + + @property + def setup_inverse_calls(self): + return self.fixed.setup_inverse_calls + + @property + def network_builds(self): + return len(self._networks) + + def network(self, offset): + require(type(offset) is int and 0 <= offset < self.moduli[self.target_axis], + "canonical target offset") + if offset not in self._networks: + self._networks[offset] = self.crt.benes(self.crt.rotation( + 1 << self.widths[self.target_axis], self.moduli[self.target_axis], offset)) + return self._networks[offset] + + def target_y_bit(self, row): + require(type(row) is int and 0 <= row < len(self.schedule), "local network row") + original = self.axes[self.target_axis][1] - 1 - self.schedule[row] + physical = self.placement.map_position(original) + require(2 * self.H <= physical < self.L - self.H, "target must be in the middle field") + return self.L - self.H - 1 - physical + + def word(self, row): + require(row in self.rows, "row uses the separate tail branch rather than a complete guard") + return self.Word(self.G, self.H, self.Y, (self.target_y_bit(row),), self.K) + + def coordinates(self, current_y): + require(type(current_y) is int and 0 <= current_y < (1 << self.Y), + "complete current middle field") + placed = current_y << self.H + result = [] + for parts in self._read_parts: + value = 0 + for shift, width in parts: + value = (value << width) | ((placed >> shift) & ((1 << width) - 1)) + result.append(value) + return tuple(result) + + def evaluate(self, row, current_y): + target_y_bit = self.target_y_bit(row) + coordinates = self.coordinates(current_y) + value = self.fixed.evaluate(coordinates[:2]) + local_bit, table = self.network(value.offset)[row] + require(local_bit == self.schedule[row], "common retained Beneš schedule") + return Query(current_y, row, coordinates, value.residues, value.valid, + value.k_A, value.offset, coordinates[2], local_bit, + target_y_bit, table[coordinates[2]]) + + def source(self, row, trace=None, cache=False): + """Return the actual source; optionally append every callback's Query record.""" + self.target_y_bit(row) + values = {} if cache else None + + def callback(current_y): + if values is None: + query = self.evaluate(row, current_y) + else: + if current_y not in values: + values[current_y] = self.evaluate(row, current_y) + query = values[current_y] + if trace is not None: + trace.append(query.record()) + return (query.bit,) + + return callback + + def stale_offset_source(self, row, offset): + """Negative control: query live target values in one incorrectly frozen network.""" + self.target_y_bit(row) + table = self.network(offset)[row][1] + + def callback(current_y): + return (table[self.coordinates(current_y)[2]],) + + return callback diff --git a/research/crt-transported-key-tape/verify.py b/research/crt-transported-key-tape/verify.py new file mode 100644 index 0000000000..47bc812c01 --- /dev/null +++ b/research/crt-transported-key-tape/verify.py @@ -0,0 +1,208 @@ +"""Source-bound finite tape replay and independently interpreted event ledger. + +JoyboyBrian, with substantial OpenAI Codex assistance. Apache-2.0. +Observer decoding and predecessor oracle arithmetic are outside the runtime. +""" +from collections import Counter +import argparse +import hashlib +import json +from pathlib import Path +import sys +import types + +PACKAGE = Path(__file__).resolve().parent +ROOT = PACKAGE.parents[1] + + +def need(condition, message): + if not condition: + raise ValueError(message) + + +def load(name, path, data): + module = types.ModuleType(name) + module.__file__ = str(path) + sys.modules[name] = module + exec(compile(data, str(path), "exec"), module.__dict__) + return module + + +def sources(): + manifest = json.loads((PACKAGE / "SOURCES.json").read_text()) + authenticated = {} + for row in manifest["files"]: + if row["usage"] != "input": + continue + data = (ROOT / row["path"]).read_bytes() + need(hashlib.sha256(data).hexdigest() == row["sha256"], "source SHA256 drift") + need(hashlib.sha1(b"blob " + str(len(data)).encode() + b"\0" + data).hexdigest() + == row["git_blob"], "source blob drift") + authenticated[row["path"]] = data + vendor = json.loads((PACKAGE / "VENDOR.json").read_text()) + data = (PACKAGE / vendor["path"]).read_bytes() + need(hashlib.sha256(data).hexdigest() == vendor["sha256"], "retained caller drift") + return authenticated, data + + +def bits(value, width): + return tuple((value >> bit) & 1 for bit in range(width)) + + +def integer(values): + return sum(value << bit for bit, value in enumerate(values)) + + +def state(values): + return values[0], values[1], integer(reversed(values[2:17])) + + +def audit(machine, initial): + """Reconstruct all tapes from unit events without using Tape methods.""" + cells = {key: list(value) for key, value in initial.items()} + heads = dict.fromkeys(cells, 0) + counts, phases = Counter(), Counter() + for phase, name, op, position, value in machine.events: + need(position == heads[name], "event head continuity") + need(0 <= position < len(cells[name]), "event bounds") + if op == "M": + need(value in (-1, 1), "unit head move") + heads[name] += value + need(0 <= heads[name] < len(cells[name]), "moved head bounds") + elif op == "R": + need(value == cells[name][position], "event read content") + elif op == "W": + need(name not in ("input", "rom") and value in (0, 1), "event ownership") + cells[name][position] = value + else: + raise ValueError("unknown event") + counts[op] += 1 + phases[phase] += 1 + for tape in machine.tapes: + need(cells[tape.name] == tape.cells and heads[tape.name] == tape.head, + "independent event final state") + need(tape.head == 0, "head restoration") + need(cells["input"] == list(initial["input"]), "borrowed input unchanged") + need(cells["rom"] == list(initial["rom"]), "borrowed ROM unchanged") + need(not any(cells["work"]) and not any(cells["scratch"]), "private erasure") + need(machine.query_count == 3, "exact three query epochs") + encoded = json.dumps(machine.events, separators=(",", ":")).encode() + return dict(actions=len(machine.events), operations=dict(sorted(counts.items())), + phases=dict(sorted(phases.items())), trace_sha256=hashlib.sha256(encoded).hexdigest()) + + +def fixtures(): + data, retained = sources() + paths = ("research/unconditional-batched-crt/check_batched_crt.py", + "research/crt-reservoir-placement/allocator.py", + "research/crt-arbitrary-source-repair/verify.py") + modules = [load("key_reference_%d" % n, ROOT / path, data[path]) + for n, path in enumerate(paths)] + caller = load("key_retained_caller", PACKAGE / "vendor/pr4716-caller.py", retained) + root = caller.RootCaller(*modules) + word = root.word(6) + need(root.fixed.prefixes == (1, 11) and root.fixed.product == 143 + and root.fixed.target_inverse == 5 and word.targets == (9,), "fixed caller scope") + weights = tuple(p * c * a % 143 for m, p, c, a in root.fixed.terms) + need(weights == (78, 11), "exact cached CRT weights") + # Compile physical coordinates from the pinned placement, not from witnesses. + coordinates = [] + for axis, width in enumerate(root.widths[:3]): + positions = [] + for bit in range(width): + candidates = [ybit for ybit in range(15) + if root.coordinates(1 << ybit)[axis] == 1 << bit] + need(len(candidates) == 1, "unique transported descriptor") + positions.append(candidates[0]) + coordinates.append(tuple(positions)) + rom = tuple(value for offset in range(17) for value in root.network(offset)[6][1]) + need(len(rom) == 544, "complete readonly row ROM") + engine = load("key_engine", PACKAGE / "engine.py", (PACKAGE / "engine.py").read_bytes()) + return root, word, rom, tuple(coordinates), engine + + +def run_case(root, word, rom, coordinates, engine, value, back, mode="correct"): + address = (value[0], value[1], *reversed(bits(value[2], 15)), back) + machine = engine.KeyMachine(address, rom, coordinates, mode) + initial = {tape.name: tuple(tape.cells) for tape in machine.tapes} + output = machine.run() + trace = [] + source = root.source(6, trace) + origin = word.run(value, source, inverse=True) + expected = word.ideal(origin, source) + need(state(output) == expected and output[17] == back, "literal key equality") + epochs = [] + for snapshot, oracle in zip(machine.snapshots, trace): + active, scratch, answer = snapshot + actual = dict(current_y=integer(reversed(active[2:17])), + coordinates=[integer(scratch[0:4]), integer(scratch[4:8]), + integer(scratch[8:13])], + offset=integer(scratch[46:52]), bit=answer) + need(actual["current_y"] == oracle["current_y"] + and tuple(actual["coordinates"]) == oracle["coordinates"] + and actual["offset"] == oracle["offset"] and answer == oracle["bit"], + "current-state query trace") + epochs.append(actual) + invoice = audit(machine, initial) + return dict(input=list(value), back=back, key=list(expected), epochs=epochs, **invoice) + + +def verify(): + root, word, rom, coordinates, engine = fixtures() + # Known discriminators, invalid/padded fields, wrap boundaries and mixed high bits. + ys = (4, 6, 0, 511, 512, 1023, 1024, 16383, 16384, 30724, 31236, 32767) + cases = [run_case(root, word, rom, coordinates, engine, (u, t, y), (u ^ t)) + for y in ys for u, t in ((0, 1), (1, 1))] + need([x["current_y"] for x in cases[0]["epochs"]] == [4, 31236, 31236], "first witness") + need([x["current_y"] for x in cases[1]["epochs"]] == [4, 4, 31236], "last-query witness") + controls = [] + for mode, value in (("stale_second", (0, 1, 4)), ("stale_third", (1, 1, 4)), + ("drop_high", (0, 1, 4)), ("omit_clear", (0, 1, 4)), + ("omit_return", (0, 1, 4)), ("bad_weight", (0, 1, 4)), + ("invalid_left_enabled", (0, 1, 6))): + try: + run_case(root, word, rom, coordinates, engine, value, 0, mode) + except ValueError as exc: + controls.append({"mode": mode, "rejected_by": str(exc)}) + else: + raise ValueError("undetected control: " + mode) + wrong = (tuple(reversed(coordinates[0])), *coordinates[1:]) + try: + run_case(root, word, rom, wrong, engine, (0, 1, 4), 0) + except ValueError as exc: + controls.append({"mode": "wrong_transport_descriptor", "rejected_by": str(exc)}) + else: + raise ValueError("undetected descriptor control") + # Event completeness: even a correct output does not certify a corrupted ledger. + machine = engine.KeyMachine((0, 1, *reversed(bits(4, 15)), 0), rom, coordinates) + initial = {tape.name: tuple(tape.cells) for tape in machine.tapes} + machine.run() + index = next(i for i, e in enumerate(machine.events) if e[2] == "M") + machine.events.pop(index) + try: + audit(machine, initial) + except ValueError as exc: + controls.append({"mode": "missing_unit_move_event", "rejected_by": str(exc)}) + else: + raise ValueError("undetected event ledger control") + return dict(status="PASS_CRT_TRANSPORTED_KEY_TAPE", cases=cases, controls=controls, + coordinate_y_bits=coordinates, rom_bits=544, + rom_sha256=hashlib.sha256(bytes(rom)).hexdigest(), + scope="Fixed finite tape-action census; setup and Boolean control time excluded") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--write-result", action="store_true") + args = parser.parse_args() + result = verify() + encoded = json.dumps(result, sort_keys=True, indent=2) + "\n" + if args.write_result: + (PACKAGE / "RESULT.json").write_text(encoded) + else: + need((PACKAGE / "RESULT.json").read_text() == encoded, "frozen receipt equality") + print(result["status"], len(result["cases"]), "cases", len(result["controls"]), "controls") + + +if __name__ == "__main__": + main()