From 9ae1d82f4e248dd830e32c4d2a0330b62cddc1dd Mon Sep 17 00:00:00 2001 From: Nick Petalas Date: Wed, 28 Oct 2026 12:57:23 +0000 Subject: [PATCH] feat(research): minimize the paired residue dispatcher bank --- research/shared-residue-dispatcher/PROOF.md | 118 +++++++++++ research/shared-residue-dispatcher/README.md | 123 ++++++++++++ .../shared-residue-dispatcher/SOURCES.json | 18 ++ research/shared-residue-dispatcher/TASK.json | 110 +++++++++++ .../shared-residue-dispatcher/VERIFY.json | 32 +++ .../shared-residue-dispatcher/dispatcher.py | 134 +++++++++++++ research/shared-residue-dispatcher/verify.py | 186 ++++++++++++++++++ 7 files changed, 721 insertions(+) create mode 100644 research/shared-residue-dispatcher/PROOF.md create mode 100644 research/shared-residue-dispatcher/README.md create mode 100644 research/shared-residue-dispatcher/SOURCES.json create mode 100644 research/shared-residue-dispatcher/TASK.json create mode 100644 research/shared-residue-dispatcher/VERIFY.json create mode 100644 research/shared-residue-dispatcher/dispatcher.py create mode 100644 research/shared-residue-dispatcher/verify.py diff --git a/research/shared-residue-dispatcher/PROOF.md b/research/shared-residue-dispatcher/PROOF.md new file mode 100644 index 0000000000..346ae141ac --- /dev/null +++ b/research/shared-residue-dispatcher/PROOF.md @@ -0,0 +1,118 @@ +# A minimal shared forward dispatcher + +## Model and inherited graph + +Fix `a,b` in `{1,...,12}`. The four output labels are ordered `00,01,10,11`. +Use zxv's ordered-coin graph from [PR #6072](https://github.com/CrocSwap/integer-mult-bounds/pull/6072), +at `9011bf83e85ecd0443c789cf03259442f750877d`. This section states the graph +definition so the new result has no runtime dependency on that pending PR. +Its sampling law and expected-bit optimality are inherited, not new claims. + +Put `D=169`, `L=156` and + +\[ +w=((13-a)(13-b),(13-a)b,a(13-b),ab),\qquad +r_{i,k}=2^k w_i\bmod D,\qquad +n_k=\frac{\sum_i r_{i,k}}D. +\] + +Internal states are `(k,j)` with `0<=k= DENOMINATOR] + row = [] + for coin in (0, 1): + child = 2 * node + coin + if child < len(leaves): + row.append(leaves[child]) + else: + row.append(ids[((phase + 1) % PERIOD, child - len(leaves))]) + table.append(tuple(row)) + return tuple(table) + + +def disjoint_bank(): + """Give the 144 graphs common terminal labels, in lexicographic pair order.""" + internal = MAX_STATES - 4 + table, roots = [], [] + for a in range(1, 13): + for b in range(1, 13): + offset = len(table) + roots.append(offset) + table.extend(tuple(offset + t if t < PAIR_STATES else + internal + t - PAIR_STATES for t in row) + for row in pair_graph(a, b)) + table.extend((internal + i, internal + i) for i in range(4)) + return tuple(table), tuple(roots) + + +def validate_graph(table): + require(5 <= len(table) <= MAX_STATES, "graph size out of bounds") + require(all(len(row) == 2 for row in table), "binary graph required") + require(all(type(t) is int and 0 <= t < len(table) + for row in table for t in row), "invalid target") + internal = len(table) - 4 + require(all(table[internal + i] == (internal + i, internal + i) + for i in range(4)), "four fixed absorbing output labels required") + + +def refine(table, colors): + """One Moore refinement; empty-word observations separate the terminals.""" + internal = len(table) - 4 + ids, refined = {}, [] + for state, (left, right) in enumerate(table): + observation = 0 if state < internal else 1 + state - internal + signature = (observation, colors[left], colors[right]) + refined.append(ids.setdefault(signature, len(ids))) + return tuple(refined) + + +def minimize(table, roots): + validate_graph(table) + require(len(roots) <= PAIR_COUNT and all(type(r) is int and + 0 <= r < len(table) - 4 for r in roots), "invalid entry table") + colors = (0,) * (len(table) - 4) + (1, 2, 3, 4) + history = [colors] + for _ in range(MAX_REFINEMENTS): + updated = refine(table, colors) + history.append(updated) + stable = len(set(updated)) == len(set(colors)) + colors = updated + if stable: + break + else: + raise ValueError("refinement exceeded its finite bound") + count = len(set(colors)) + quotient = [None] * count + for state, row in enumerate(table): + translated = tuple(colors[t] for t in row) + prior = quotient[colors[state]] + require(prior is None or prior == translated, "unstable quotient") + quotient[colors[state]] = translated + require(all(row is not None for row in quotient), "unfilled quotient") + return tuple(quotient), tuple(colors[r] for r in roots), colors, tuple(history) + + +def pack_codes(codes, width): + """Pack little-endian fixed-width codes; unused high bits are zero.""" + require(type(width) is int and 1 <= width <= 16, "invalid code width") + require(len(codes) <= 2 * MAX_STATES, "too many packed codes") + require(all(type(c) is int and 0 <= c < 1 << width for c in codes), + "code does not fit width") + packed = bytearray((len(codes) * width + 7) // 8) + for index, code in enumerate(codes): + position = index * width + for bit in range(width): + if code & (1 << bit): + packed[(position + bit) // 8] |= 1 << ((position + bit) % 8) + return bytes(packed) + + +def unpack_codes(packed, count, width): + require(type(count) is int and 0 <= count <= 2 * MAX_STATES, + "invalid code count") + require(type(width) is int and 1 <= width <= 16, "invalid code width") + require(len(packed) == (count * width + 7) // 8, "wrong payload length") + codes = [] + for index in range(count): + position = index * width + code = sum(((packed[(position + bit) // 8] >> + ((position + bit) % 8)) & 1) << bit for bit in range(width)) + codes.append(code) + return tuple(codes) diff --git a/research/shared-residue-dispatcher/verify.py b/research/shared-residue-dispatcher/verify.py new file mode 100644 index 0000000000..e78ac558f4 --- /dev/null +++ b/research/shared-residue-dispatcher/verify.py @@ -0,0 +1,186 @@ +"""Certify the shared dispatcher, minimality partition and packed representation. + +Produced by Nick Petalas. Apache-2.0. +""" +import argparse +from hashlib import sha256 +from itertools import product +import json +from pathlib import Path +import runpy + +HERE = Path(__file__).resolve().parent +implementation = runpy.run_path(str(HERE / "dispatcher.py")) +require = implementation["require"] + + +def quotient_certificate(table, roots, quotient, entries, mapping): + """A complete homomorphism check proves all tapes, including nontermination.""" + n, m = len(table) - 4, len(quotient) - 4 + require(len(mapping) == len(table), "wrong mapping size") + require(set(mapping) == set(range(len(quotient))), "mapping not onto") + require(tuple(mapping[n + i] for i in range(4)) == tuple(m + i for i in range(4)), + "output labels not preserved") + require(all(mapping[s] < m for s in range(n)), "premature output") + require(tuple(mapping[r] for r in roots) == entries, "wrong entry pointer") + require(all(tuple(mapping[t] for t in row) == quotient[mapping[s]] + for s, row in enumerate(table)), "transition does not commute") + + +def partition_certificate(table, history, mapping): + """Check each refinement independently by sorting signatures, not interning.""" + n = len(table) - 4 + require(history[0] == (0,) * n + (1, 2, 3, 4), "wrong empty observation") + for before, after in zip(history, history[1:]): + signatures = [(0 if s < n else 1 + s - n, + before[row[0]], before[row[1]]) + for s, row in enumerate(table)] + order = sorted(range(len(table)), key=signatures.__getitem__) + require(all((signatures[u] == signatures[v]) == (after[u] == after[v]) + for u, v in zip(order, order[1:])), "wrong refinement split") + require(len(set(after)) == len(set(signatures)), "reused class name") + require(history[-1] == mapping, "wrong final partition") + require(len(set(history[-1])) == len(set(history[-2])), "not stable") + + +def reachable(table, roots): + seen = set(roots) + frontier = list(roots) + for index in range(len(table)): + if index == len(frontier): + break + for target in table[frontier[index]]: + if target not in seen: + seen.add(target) + frontier.append(target) + require(len(seen) == len(table), "unreachable states in minimality count") + + +def run(table, start, bits): + internal = len(table) - 4 + state = start + if state >= internal: + return state - internal, 0 + for used, bit in enumerate(bits, 1): + state = table[state][bit] + if state >= internal: + return state - internal, used + return None, len(bits) + + +def literal_controls(table, roots, quotient, entries): + count = 0 + for root, entry in zip(roots, entries): + for bits in product((0, 1), repeat=8): + require(run(table, root, bits) == run(quotient, entry, bits), + "literal tape changed") + count += 1 + return count + + +def sharp_separation(table, history): + """Extract a pair not distinguished at depth 8 but distinguished at depth 9.""" + older, newer = history[8], history[9] + representatives = {} + pair = None + for state, color in enumerate(older): + prior = representatives.setdefault(color, state) + if newer[prior] != newer[state]: + pair = (prior, state) + break + require(pair is not None, "missing sharp depth-nine pair") + left, right = pair + bits = [] + for depth in range(9, 0, -1): + if history[0][left] != history[0][right]: + break + choices = [b for b in (0, 1) if + history[depth - 1][table[left][b]] != + history[depth - 1][table[right][b]]] + require(bool(choices), "missing distinguishing continuation") + bit = choices[0] + bits.append(bit) + left, right = table[left][bit], table[right][bit] + require(len(bits) == 9 and run(table, pair[0], bits) != run(table, pair[1], bits), + "incorrect depth-nine witness") + return {"states": pair, "coins": "".join(map(str, bits))} + + +def rejection_controls(table, roots, quotient, entries, mapping, history): + rejected = 0 + # Corrupt different parts of the complete certificate, not sample tapes. + for kind in ("output", "entry", "edge", "map", "partition"): + q, e, mp, h = list(quotient), list(entries), list(mapping), list(history) + if kind == "output": + mp[-1], mp[-2] = mp[-2], mp[-1] + elif kind == "entry": + e[0] = (e[0] + 1) % (len(q) - 4) + elif kind == "edge": + q[0] = (len(q) - 1, len(q) - 1) + elif kind == "map": + mp[0] = (mp[0] + 1) % (len(q) - 4) + else: + h[1] = (0,) * (len(table) - 4) + (1, 2, 3, 4) + try: + quotient_certificate(table, roots, q, e, mp) + partition_certificate(table, h, mp) + except ValueError: + rejected += 1 + require(rejected == 5, "corrupted certificate accepted") + return rejected + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--receipt", type=Path) + parser.add_argument("--emit-dir", type=Path) + args = parser.parse_args() + table, roots = implementation["disjoint_bank"]() + quotient, entries, mapping, history = implementation["minimize"](table, roots) + quotient_certificate(table, roots, quotient, entries, mapping) + partition_certificate(table, history, mapping) + reachable(table, roots) + reachable(quotient, entries) + counts = [len(set(colors)) for colors in history] + require(counts == [5, 15, 56, 282, 1850, 7426, 14824, 20274, + 22450, 22486, 22486], "unexpected exact partition counts") + require(len(quotient) - 4 == 22482, "unexpected internal-state minimum") + width = (len(quotient) - 1).bit_length() + codes = tuple(t for row in quotient[:-4] for t in row) + payload = implementation["pack_codes"](codes, width) + entry_payload = implementation["pack_codes"](entries, width) + require(implementation["unpack_codes"](payload, len(codes), width) == codes, + "transition packing changed codes") + require(implementation["unpack_codes"](entry_payload, len(entries), width) == entries, + "entry packing changed codes") + require((width, len(payload), len(entry_payload)) == (15, 84308, 270), + "wrong packed bank accounting") + tapes = literal_controls(table, roots, quotient, entries) + controls = rejection_controls(table, roots, quotient, entries, mapping, history) + result = {"status": "PASS_SHARED_RESIDUE_DISPATCHER", "configurations": len(roots), + "original_internal_states": len(table) - 4, + "minimal_internal_states": len(quotient) - 4, + "partition_counts": counts, "literal_tapes": tapes, + "rejected_corruptions": controls, "code_width": width, + "transition_bytes": len(payload), "entry_bytes": len(entry_payload), + "shared_bank_bytes": len(payload) + len(entry_payload), + "separate_bank_bytes": 144 * 702, + "saved_bytes": 144 * 702 - len(payload) - len(entry_payload), + "transition_sha256": sha256(payload).hexdigest(), + "entries_sha256": sha256(entry_payload).hexdigest(), + "depth_nine_witness": sharp_separation(table, history), + "scope": "fixed ordered-coin forward behavior; no physical tariff or new kappa"} + if args.emit_dir is not None: + args.emit_dir.mkdir(parents=True, exist_ok=True) + (args.emit_dir / "transitions.bin").write_bytes(payload) + (args.emit_dir / "entries.bin").write_bytes(entry_payload) + if args.receipt is not None: + hashes = {p.name: sha256(p.read_bytes()).hexdigest() + for p in sorted(HERE.iterdir()) if p.is_file()} + args.receipt.write_text(json.dumps({"files_sha256": hashes, "result": result}, + indent=2, sort_keys=True) + "\n") + print(json.dumps(result, sort_keys=True)) + + +if __name__ == "__main__": + main()