= ), so it is a
+ consistency check of the implementation, not extra evidence.
+ """
+ inverse = [0] * 512
+ for x, y in enumerate(images):
+ inverse[y] = x
+ # G^{-T} z: bit j of the result is .
+ git = lambda z: sum((bin(inverse[1 << j] & z).count('1') & 1) << j for j in range(H))
+ lift = lambda x, z: (images[x], images[x] ^ git(x ^ z))
+ pair = lambda a, b: (bin(a[0] & b[1]).count('1') + bin(a[1] & b[0]).count('1')) & 1
+ basis = [(1 << i, 0) for i in range(H)] + [(0, 1 << i) for i in range(H)]
+ lifted = [lift(*b) for b in basis]
+ if any(pair(lifted[i], lifted[j]) != pair(basis[i], basis[j]) for i in range(18) for j in range(i + 1, 18)):
+ return False
+
+ def frame(mask):
+ us = points(mask)
+ perp = [p for p in range(512) if all(bin(p & u).count('1') % 2 == 0 for u in us)]
+ return {(u, u ^ p) for u in us for p in perp}
+ return all({lift(*pt) for pt in frame(a)} == frame(b) for a, b in ((ZERO, ZERO), (FULL, FULL), (U, E)))
+
+
+# ----------------------------------------------------------------------------
+# Chronological binary-frame bill (BORROW-WORD contract)
+# ----------------------------------------------------------------------------
+def frame_bill(w, lagrangian=False):
+ rep = w['report']
+ need((rep['v'], 2 * rep['v'] + rep['R'], rep['h']) == (V, N, H), 'word geometry')
+ removed = set(w['removed'])
+ need(len(removed) == len(w['removed']) and all(2 * V <= r < N for r in removed), 'removed helpers')
+ live = [r for r in range(N) if r not in removed]
+ liveset = set(live)
+ B, warps = w['B'], w['warps']
+ for k, (pos, lender, cols, U, E) in enumerate(warps, 1):
+ need(0 <= pos <= len(B), 'warp position', k)
+ need(lender in liveset and not is_target(lender), 'warp lender', k)
+ images = warp_images(cols)
+ su, se = span(U), span(E)
+ need(dim(su) == len(U) == len(E) == dim(se), 'warp frame bases', k)
+ need(image_mask(images, su) == se, 'warp does not map its frame', k)
+ if lagrangian:
+ need(lagrangian_action_ok(images, su, se), 'lifted frame action', k)
+ for name in ('A', 'B', 'restore'):
+ for g in w[name]:
+ if g[0] != 'warp':
+ need(all(r in liveset for r in touched(g)), 'removed/out-of-range register', name, g)
+ if g[0] == 'add' and not is_target(g[1]):
+ need(not is_target(g[2]), 'non-target reads a target', name, g)
+ need(all(g[-1] is None and is_target(g[1]) and g[0] == 'add' for g in w['cut']), 'cut gates')
+ need(all(g[-1] == [] and is_target(g[1]) for g in w['precondition'] + w['compensation']), 'zero-frame wrapper writes')
+ need(all(not is_target(g[1]) for g in w['restore']), 'restore writes a target')
+
+ ports = w['ports']
+ start, end = {}, {}
+ for r in live:
+ if r < V:
+ start[r], end[r] = span((ports[r],)), FULL
+ elif r < 2 * V:
+ start[r], end[r] = ZERO, hyperplane(ports[r - V])
+ else:
+ start[r], end[r] = ZERO, FULL
+ for q, U in w['start_extra'].items():
+ need(int(q) in liveset and int(q) >= 2 * V, 'entrance names a live helper', q)
+ start[int(q)] = span(U)
+ for q, U in w['helper_final'].items():
+ need(int(q) in liveset and int(q) >= 2 * V, 'terminal names a live helper', q)
+ end[int(q)] = span(U)
+
+ current, bill, chain = dict(start), Counter(), {}
+
+ def climb(r, mask, where):
+ need(current[r] & ~mask == 0, 'frame containment fails', where, r)
+ d = dim(mask) - dim(current[r])
+ if d:
+ bill[d] += 1
+ chain.setdefault(r, []).append((where, d, mask))
+ current[r] = mask
+
+ npre = w['prelude_events']
+ for i, g in enumerate(w['A']):
+ if g[0] == 'warp':
+ need(i < npre and len(g) == 5, 'warp outside the prelude', i)
+ _, r, op, fr, to = g
+ need(op != 0 and abs(op) <= len(warps) and warps[abs(op) - 1][1] == r, 'prelude warp lender', i)
+ images = warp_images(warps[abs(op) - 1][2])
+ if op < 0:
+ inverse = [0] * 512
+ for x, y in enumerate(images):
+ inverse[y] = x
+ images = inverse
+ need(image_mask(images, span(fr)) == span(to), 'prelude warp frame action', i)
+ need(current[r] == span(fr), 'prelude warp entry frame', i, r)
+ current[r] = span(to)
+ chain.setdefault(r, []).append((('A', i), 'warp', span(to)))
+ else:
+ if i < npre:
+ need(is_target(g[1]), 'prelude scalar writes a non-target', i)
+ for r in touched(g):
+ climb(r, span(g[-1]), ('A', i))
+ levels = {V + int(t): span(U) for t, U in w['target_levels'].items()}
+ need(all(current[t] == levels.get(t, ZERO) for t in range(V, 2 * V)), 'target cut levels')
+ for q, U in zip(w['centers'], w['center_frames']):
+ U = span(U)
+ need(current[q] == U, 'centre entry frame', q)
+ served = {levels.get(g[1], ZERO) for g in w['cut'] if g[2] == q}
+ for T in sorted(served, key=lambda m: -dim(m)):
+ need(T & ~U == 0, 'centre descent not nested', q)
+ d = dim(U) - dim(T)
+ if d:
+ bill[d] += 1
+ chain.setdefault(q, []).append((('centre',), d, T))
+ U = T
+ at = {}
+ for k, rec in enumerate(warps, 1):
+ at.setdefault(rec[0], []).append(k)
+ for p in range(len(B) + 1):
+ for k in at.get(p, ()):
+ _, lender, _, U, E = warps[k - 1]
+ need(current[lender] == span(U), 'native warp entry frame', k, lender)
+ current[lender] = span(E)
+ chain.setdefault(lender, []).append((('warp', k), 'warp', span(E)))
+ if p < len(B):
+ for r in touched(B[p]):
+ climb(r, span(B[p][-1]), ('B', p))
+ for i, g in enumerate(w['restore']):
+ for r in touched(g):
+ climb(r, span(g[-1]), ('restore', i))
+ for r in live:
+ climb(r, end[r], ('end',))
+ restored = set(map(int, w['helper_final']))
+ need(all(end[r] == FULL for r in live if r not in restored and not is_target(r)), 'a FULL-cleanup register ends low')
+ widths = Counter(dim(end[r]) - dim(start[r]) for r in live if r >= 2 * V)
+ return dict(sorted(bill.items())), dict(sorted(widths.items())), chain, start, end
+
+
+def hygiene(w):
+ """Inherited structural contract: dyadic scalars, power-of-two scales, wrapper and correction access."""
+ entering = set(map(int, w['start_extra']))
+ restored = set(map(int, w['helper_final']))
+ live = set(range(N)) - set(w['removed'])
+ for name in ('precondition', 'compensation', 'A', 'cut', 'B', 'restore', 'cleanup'):
+ for g in w[name]:
+ if g[0] == 'warp':
+ continue
+ q = Q(g[3], g[4]) if g[0] == 'add' else Q(g[2], g[3])
+ need(q != 0 and q.denominator & (q.denominator - 1) == 0, 'non-dyadic scalar', name, g)
+ if g[0] == 'scale':
+ need(abs(q.numerator) & (abs(q.numerator) - 1) == 0, 'non power-of-two scale', name, g)
+ for g in w['precondition'] + w['compensation']:
+ need(not any(r in entering | restored for r in touched(g)), 'wrapper touches an entering or restored helper', g)
+ for c in w['initial_corrections']:
+ word = c['operator_word']
+ need(is_target(c['target']) and c['source'] in live and c['source'] >= 2 * V
+ and c['source'] not in entering, 'correction access', c)
+ need(word and all(w['warps'][abs(o) - 1][1] == c['source'] for o in word), 'correction operator lender', c)
+ q = Q(c['coefficient'])
+ need(q != 0 and q.denominator & (q.denominator - 1) == 0, 'correction coefficient', c)
+ return True
+
+
+# ----------------------------------------------------------------------------
+# Literal inverse cleanup, derived from the contract
+# ----------------------------------------------------------------------------
+def ordinary_forward(w):
+ """Chronological non-wrapper actions whose inverse forms cleanup."""
+ out = [g for g in w['A'][w['prelude_events']:]]
+ at = {}
+ for k, rec in enumerate(w['warps'], 1):
+ at.setdefault(rec[0], []).append(['warp', rec[1], k])
+ for p in range(len(w['B']) + 1):
+ out += at.get(p, [])
+ if p < len(w['B']):
+ out.append(w['B'][p])
+ return out + list(w['restore'])
+
+
+def derive_cleanup(w):
+ restored = set(map(int, w['helper_final']))
+ buf = lambda r: r + N if r in restored else r
+ full = [256, 128, 64, 32, 16, 8, 4, 2, 1]
+ out = []
+ for g in reversed(ordinary_forward(w)):
+ if g[0] == 'warp':
+ out.append(['warp', buf(g[1]), -g[2], full])
+ elif is_target(g[1]):
+ continue
+ elif g[0] == 'add':
+ out.append(['add', buf(g[1]), buf(g[2]), -g[3], g[4], full])
+ else:
+ need(g[0] == 'scale', 'unknown primitive', g)
+ out.append(['scale', buf(g[1]), g[3], g[2], full])
+ return out
+
+
+def same_cleanup(a, b):
+ norm = lambda c: [(g[:-1], span(g[-1])) for g in c]
+ return len(a) == len(b) and norm(a) == norm(b)
+
+
+# ----------------------------------------------------------------------------
+# Exact free-operator semantics. A value is a dict {(input, word): Fraction};
+# word is a reduced tuple of signed operator ids (leftmost = applied last).
+# ----------------------------------------------------------------------------
+def program(w, cleanup=None):
+ """Literal executed sequence: corrections, wrapper, A, cut, B+warps, restore | cleanup."""
+ items = [('corr', c['target'], c['source'], tuple(c['operator_word']), Q(c['coefficient']))
+ for c in w['initial_corrections']]
+ items += w['precondition'] + w['compensation'] + w['A'] + w['cut']
+ at = {}
+ for k, rec in enumerate(w['warps'], 1):
+ at.setdefault(rec[0], []).append(['warp', rec[1], k])
+ for p in range(len(w['B']) + 1):
+ items += at.get(p, [])
+ if p < len(w['B']):
+ items.append(w['B'][p])
+ items += w['restore']
+ return items, list(w['cleanup'] if cleanup is None else cleanup)
+
+
+def dual_op(op):
+ return (1 if op > 0 else -1) * (abs(op) + DUAL)
+
+
+def _left(op, word):
+ return word[1:] if word and word[0] == -op else (op,) + word
+
+
+def _right(word, op):
+ return word[:-1] if word and word[-1] == -op else word + (op,)
+
+
+def _axpy(dst, src, q):
+ for key, x in src.items():
+ y = dst.get(key, 0) + q * x
+ if y:
+ dst[key] = y
+ else:
+ dst.pop(key, None)
+
+
+def forward_rows(w, cleanup=None, dual=False, symbolic_buffers=True, snapshot=None):
+ """Schrodinger pass: every register's value as a combination of initial inputs.
+
+ Live inputs are independent symbols. Private buffers are independent symbols
+ when symbolic_buffers is set (dirty), otherwise zero. The optional snapshot
+ callback sees the rows at the cleanup boundary.
+ """
+ removed = set(w['removed'])
+ live = [r for r in range(N) if r not in removed]
+ restored = sorted(map(int, w['helper_final']))
+ rows = {r: {(r, ()): Q(1)} for r in live}
+ for c in restored:
+ rows[c + N] = {(c + N, ()): Q(1)} if symbolic_buffers else {}
+ prefix, tail = program(w, cleanup)
+
+ def apply(g):
+ kind = g[0]
+ if kind == 'add':
+ t, s, q = g[1], g[2], Q(g[3], g[4])
+ need(t != s and q, 'add gate', g)
+ if dual:
+ t, s, q = s, t, -q
+ _axpy(rows[t], rows[s], q)
+ elif kind == 'scale':
+ q = Q(g[2], g[3])
+ need(q != 0, 'zero scale', g)
+ if dual:
+ q = 1 / q
+ rows[g[1]] = {k: q * x for k, x in rows[g[1]].items()}
+ elif kind == 'warp':
+ op = dual_op(g[2]) if dual else g[2]
+ rows[g[1]] = {(var, _left(op, word)): x for (var, word), x in rows[g[1]].items()}
+ elif kind == 'corr':
+ _, t, s, word, c = g
+ if dual:
+ src, ops, dst, c = rows[t], [-dual_op(o) for o in word], s, -c
+ else:
+ src, ops, dst = rows[s], list(reversed(word)), t
+ moved = {}
+ for (var, wd), x in src.items():
+ for o in ops:
+ wd = _left(o, wd)
+ moved[var, wd] = x
+ _axpy(rows[dst], moved, c)
+ else:
+ raise AuditFailure('unknown primitive ' + repr(g))
+
+ for g in prefix:
+ apply(g)
+ if snapshot is not None:
+ snapshot(rows)
+ for g in tail:
+ apply(g)
+ return rows
+
+
+def reverse_columns(w, cleanup=None, upto_cleanup=False):
+ """Heisenberg pull-back: process the forward program backwards.
+
+ cols[s] = {(output, word): coefficient}: the final output register 'output'
+ contains coefficient*word*x_s. Words grow on the right. Every live input and
+ every private buffer is an independent symbol.
+ """
+ removed = set(w['removed'])
+ live = [r for r in range(N) if r not in removed]
+ restored = sorted(map(int, w['helper_final']))
+ regs = live + [c + N for c in restored]
+ cols = {r: {(r, ()): Q(1)} for r in regs}
+ prefix, tail = program(w, cleanup)
+ seq = prefix if upto_cleanup else prefix + tail
+ for g in reversed(seq):
+ kind = g[0]
+ if kind == 'add':
+ t, s, q = g[1], g[2], Q(g[3], g[4])
+ need(t != s and q, 'add gate', g)
+ _axpy(cols[s], cols[t], q)
+ elif kind == 'scale':
+ q = Q(g[2], g[3])
+ need(q != 0, 'zero scale', g)
+ cols[g[1]] = {k: q * x for k, x in cols[g[1]].items()}
+ elif kind == 'warp':
+ op = g[2]
+ cols[g[1]] = {(out, _right(word, op)): x for (out, word), x in cols[g[1]].items()}
+ elif kind == 'corr':
+ _, t, s, word, c = g
+ moved = {}
+ for (out, wd), x in cols[t].items():
+ for o in word:
+ wd = _right(wd, o)
+ moved[out, wd] = x
+ _axpy(cols[s], moved, c)
+ else:
+ raise AuditFailure('unknown primitive ' + repr(g))
+ return cols
+
+
+def transpose(cols):
+ rows = {}
+ for s, col in cols.items():
+ for (r, word), x in col.items():
+ rows.setdefault(r, {})[s, word] = x
+ return rows
+
+
+def endpoint_defects(w, rows, dual=False):
+ """Compare final rows with the contract on the zero-buffer (live-input) domain.
+
+ Returns (bad live registers, dirty buffers, buffer->live leaks, buffer self-map identity).
+ """
+ removed = set(w['removed'])
+ live = [r for r in range(N) if r not in removed]
+ restored = sorted(map(int, w['helper_final']))
+ bad = []
+ for r in live:
+ expect = {(r, ()): Q(1)}
+ if dual and r < V:
+ expect[r + V, ()] = Q(-1)
+ if not dual and is_target(r):
+ expect[r - V, ()] = Q(1)
+ got = {k: x for k, x in rows.get(r, {}).items() if k[0] < N}
+ if got != expect:
+ bad.append(r)
+ dirty = [c for c in restored if any(k[0] < N for k in rows.get(c + N, {}))]
+ leaks = sorted({r for r in live for k in rows.get(r, {}) if k[0] >= N})
+ buffer_identity = all({k: x for k, x in rows.get(c + N, {}).items() if k[0] >= N} == {(c + N, ()): Q(1)}
+ for c in restored)
+ return bad, dirty, leaks, buffer_identity
+
+
+def boundary_defects(w, rows):
+ """Contract at the cleanup boundary (forward): targets sheared, restored helpers clean and isolated."""
+ restored = set(map(int, w['helper_final']))
+ out = []
+ for t in range(V, 2 * V):
+ if {k: x for k, x in rows[t].items() if k[0] < N} != {(t, ()): Q(1), (t - V, ()): Q(1)}:
+ out.append(('target', t))
+ for c in restored:
+ if rows[c] != {(c, ()): Q(1)}:
+ out.append(('restored', c))
+ for r, row in rows.items():
+ if r < N and r not in restored and any(k[0] in restored for k in row):
+ out.append(('restored input read', r))
+ return out
diff --git a/research/staged544-separated-a2/test_verify.py b/research/staged544-separated-a2/test_verify.py
new file mode 100644
index 0000000000..4a0686dc62
--- /dev/null
+++ b/research/staged544-separated-a2/test_verify.py
@@ -0,0 +1,228 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+"""Bounded mutation controls for the fixed separated-A2 witnesses.
+
+Uses the submitted verifier as the subject under test; it is not an independent
+matrix or full-word semantics proof. Every required check uses explicit errors,
+so the same suite remains effective with Python's optimization enabled.
+"""
+import argparse
+from copy import deepcopy
+import importlib.util
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+
+
+class TestFailure(Exception):
+ pass
+
+
+def need(ok, message):
+ if not ok:
+ raise TestFailure(message)
+
+
+def import_subject(root):
+ sys.path.insert(0, str(root))
+ spec = importlib.util.spec_from_file_location('separated_a2_subject', root / 'verify.py')
+ subject = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(subject)
+ return subject
+
+
+def writable_copy(source, destination):
+ """Fresh files only; never relax permissions on a frozen source export."""
+ shutil.copytree(source, destination, copy_function=shutil.copyfile)
+ for path in destination.rglob('*'):
+ path.chmod(0o755 if path.is_dir() else 0o644)
+ destination.chmod(0o755)
+
+
+def run_cli(root, provider='base'):
+ command = [sys.executable, '-I', '-B']
+ if not __debug__:
+ command.append('-O')
+ command.extend([str(root / 'verify.py'), '--provider', provider, '--quick'])
+ return subprocess.run(command, cwd=root, capture_output=True, text=True)
+
+
+class Controls:
+ def __init__(self, errors):
+ self.errors = errors
+ self.rejected = []
+
+ def rejects(self, name, action):
+ try:
+ action()
+ except self.errors:
+ self.rejected.append(name)
+ return
+ except Exception as error:
+ raise TestFailure(name + ' crashed with ' + type(error).__name__ + ': ' + str(error)) from error
+ raise TestFailure(name + ' was accepted')
+
+ def cli_rejects(self, name, fixture):
+ result = run_cli(fixture)
+ need(result.returncode != 0, name + ' CLI accepted the altered fixture')
+ need('FAIL' in result.stderr, name + ' CLI did not report an explicit validation failure: ' + result.stderr[-500:])
+ self.rejected.append(name)
+
+
+def tree_bytes(root):
+ return {str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
+ for path in root.rglob('*') if path.is_file()}
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('--package', type=Path, default=Path(__file__).resolve().parent)
+ args = parser.parse_args()
+ root = args.package.resolve()
+ before = tree_bytes(root)
+ subject = import_subject(root)
+ core = subject.authenticate()
+ controls = Controls((ValueError, core.AuditFailure))
+ source = subject.load_word('base')
+ sites = list(subject.SITES)
+ separated = next(site for site in sites if not site['known'] and site['section'] == 'B')
+ known = next(site for site in sites if site['known'])
+ indices = separated['indices']
+ gap = next(i for i in range(indices[0], indices[2]) if i not in indices)
+ support = subject.certify_site(source, separated)['support']
+
+ def site_control(name, mutate):
+ word, site = deepcopy(source), deepcopy(separated)
+ mutate(word, site)
+ controls.rejects(name, lambda: subject.certify_site(word, site))
+
+ def set_site(field, value):
+ return lambda _word, site: site.__setitem__(field, value)
+
+ site_control('duplicate selected index', set_site('indices', [indices[0], indices[0], indices[2]]))
+ site_control('reversed selected indices', set_site('indices', list(reversed(indices))))
+ site_control('boolean selected index', set_site('indices', [True, indices[1], indices[2]]))
+ site_control('selected index outside section', set_site('indices', [indices[0], indices[1], len(source['B'])]))
+ site_control('boolean removed position', set_site('removed_position', False))
+ site_control('integer known-site flag', set_site('known', 0))
+ site_control('unexpected site field', set_site('unproved', True))
+ site_control('uncovered section', set_site('section', 'cut'))
+
+ def alter_selected(word, site, position, action):
+ action(word[site['section']][site['indices'][position]])
+
+ site_control('wrong A2 sign', lambda w, s: alter_selected(w, s, 0, lambda g: g.__setitem__(3, -g[3])))
+ site_control('wrong A2 product', lambda w, s: alter_selected(w, s, 2, lambda g: g.__setitem__(3, 2 * g[3])))
+ site_control('boolean A2 coefficient', lambda w, s: alter_selected(w, s, 0, lambda g: g.__setitem__(3, True)))
+ site_control('zero A2 denominator', lambda w, s: alter_selected(w, s, 0, lambda g: g.__setitem__(4, 0)))
+ site_control('mismatched original frame', lambda w, s: alter_selected(w, s, 0, lambda g: g.__setitem__(-1, [512])))
+ site_control('self ADD edge', lambda w, s: alter_selected(w, s, 0, lambda g: g.__setitem__(2, g[1])))
+ site_control('unknown selected primitive', lambda w, s: alter_selected(w, s, 0, lambda g: g.__setitem__(0, 'copy')))
+ site_control('unknown intervening primitive', lambda w, s: w['B'].__setitem__(gap, ['copy', 0, 1, 1, 1, []]))
+ site_control('intervening support dependency', lambda w, s: w['B'].__setitem__(gap, ['scale', support[0], 2, 1, []]))
+ site_control('native warp cut crossing', lambda w, s: w['warps'][0].__setitem__(0, indices[0] + 1))
+ cut_word = deepcopy(source)
+ cut_word['prelude_events'] = known['indices'][0] + 1
+ controls.rejects('prelude cut crossing', lambda: subject.certify_site(cut_word, deepcopy(known)))
+ controls.rejects('boolean include-known flag', lambda: subject.contract(source, 1))
+ controls.rejects('unrecognized provider', lambda: subject.load_word('unproved'))
+
+ positive = []
+ changed_base = None
+ for provider in ('base', 'extra'):
+ original = subject.load_word(provider)
+ changed = subject.contract(original)
+ info = subject.inspect_word(core, changed)
+ need(info['lengths'] == {'A': 4078, 'B': 1867, 'restore': 207, 'cleanup': 5218},
+ 'fixed contraction lengths changed for ' + provider)
+ need(changed['cleanup'] == core.derive_cleanup(changed), 'positive cleanup not literal inverse')
+ need(all(a[1:] == b[1:] for a, b in zip(original['warps'], changed['warps'])),
+ 'positive contraction changed native maps or frames')
+ positive.append(provider + ' fixed contractions, literal cleanup, complete bill')
+ if provider == 'base':
+ changed_base = changed
+
+ def word_control(name, mutate):
+ word = deepcopy(changed_base)
+ mutate(word)
+ controls.rejects(name, lambda: subject.inspect_word(core, word))
+
+ word_control('omitted inverse cleanup ADD', lambda w: w['cleanup'].pop())
+ word_control('duplicated inverse cleanup ADD', lambda w: w['cleanup'].append(deepcopy(w['cleanup'][-1])))
+ word_control('wrong inverse cleanup sign', lambda w: w['cleanup'][0].__setitem__(3, -w['cleanup'][0][3]))
+ word_control('stale original inverse cleanup', lambda w: w.__setitem__('cleanup', deepcopy(source['cleanup'])))
+ word_control('forged declared paid profile', lambda w: w['report']['local_profile'].__setitem__('1', 2439))
+
+ certificate = subject.compute('base', False)
+ subject.audit('base', False, certificate=certificate)
+ positive.append('exact quick certificate accepted')
+
+ def receipt_control(name, mutate):
+ receipt = deepcopy(certificate)
+ mutate(receipt)
+ controls.rejects(name, lambda: subject.audit('base', False, certificate=receipt))
+
+ receipt_control('forged forward saving', lambda r: r.__setitem__('additional_forward_add_saving', 8))
+ receipt_control('forged cleanup saving', lambda r: r.__setitem__('additional_cleanup_add_saving', 6))
+ receipt_control('integer false kappa flag', lambda r: r.__setitem__('new_kappa', 0))
+ receipt_control('unsupported kappa claim', lambda r: r.__setitem__('new_kappa', True))
+ receipt_control('unsupported native claim', lambda r: r.__setitem__('native_operators_changed', True))
+ receipt_control('forged result hash', lambda r: r['words']['separated'].__setitem__('canonical_word_sha256', '0' * 64))
+ receipt_control('forged helper inventory', lambda r: r['words']['separated']['helper_widths'].__setitem__(9, 214))
+ receipt_control('extra unverified receipt field', lambda r: r.__setitem__('proved_by_assert', True))
+
+ # CLI controls run against fresh writable copies; retained source bytes and
+ # the package under test are never changed, even on a read-only export.
+ with tempfile.TemporaryDirectory(prefix='separated-a2-negative-') as directory:
+ fixture = Path(directory) / 'package'
+ writable_copy(root, fixture)
+ baseline = run_cli(fixture)
+ need(baseline.returncode == 0 and 'PASS_STAGED544_SEPARATED_A2' in baseline.stdout,
+ 'unaltered CLI fixture failed: ' + baseline.stderr[-500:])
+ positive.append('unaltered copied CLI fixture accepted')
+
+ for relative in ('source/audit_core.py', 'inputs/BORROW-WORD-base.json.gz'):
+ path = fixture / relative
+ raw = path.read_bytes()
+ path.write_bytes(raw + b'\n')
+ controls.cli_rejects('CLI altered frozen ' + relative, fixture)
+ path.write_bytes(raw)
+
+ receipt_path = fixture / 'data.json'
+ receipt_bytes = receipt_path.read_bytes()
+ data = json.loads(receipt_bytes)
+ forged = deepcopy(data)
+ forged['providers']['base']['additional_cleanup_add_saving'] = 6
+ receipt_path.write_text(json.dumps(forged) + '\n')
+ controls.cli_rejects('CLI forged cleanup saving', fixture)
+ forged = deepcopy(data)
+ forged['schema_version'] = True
+ receipt_path.write_text(json.dumps(forged) + '\n')
+ controls.cli_rejects('CLI boolean root schema', fixture)
+ receipt_path.write_text('{"schema_version":1,"schema_version":1,"providers":' + json.dumps(data['providers']) + '}\n')
+ controls.cli_rejects('CLI duplicate JSON key', fixture)
+ receipt_path.write_bytes(receipt_bytes)
+ source_path = fixture / 'source/audit_core.py'
+ source_bytes = source_path.read_bytes()
+ sibling = Path(directory) / 'same-audit-core.py'
+ sibling.write_bytes(source_bytes)
+ source_path.unlink()
+ source_path.symlink_to(sibling)
+ controls.cli_rejects('CLI symlink source substitution', fixture)
+
+ need(tree_bytes(root) == before, 'the package under test changed')
+ print('PASS_SEPARATED_A2_NEGATIVE_CONTROLS ' + json.dumps(
+ {'positive_controls': len(positive), 'expected_rejections': len(controls.rejected),
+ 'positive_names': positive, 'rejection_names': controls.rejected,
+ 'package_bytes_unchanged': True, 'optimized': not __debug__}, sort_keys=True))
+
+
+if __name__ == '__main__':
+ try:
+ main()
+ except TestFailure as error:
+ raise SystemExit('FAIL: ' + str(error))
diff --git a/research/staged544-separated-a2/verify.py b/research/staged544-separated-a2/verify.py
new file mode 100644
index 0000000000..877e57b883
--- /dev/null
+++ b/research/staged544-separated-a2/verify.py
@@ -0,0 +1,308 @@
+#!/usr/bin/env python3
+"""Seven fixed separated A2 contractions, with complete unchanged paid invoices.
+
+Apache-2.0. ToikY with substantial OpenAI Codex assistance. The exact whole-word
+arithmetic and frame engine is credited retained audit_core.py; this is not a
+new independent interpreter or native/supplier audit. The replay adapter follows
+the ordinary/dual/reverse contract of roger-sato's retained audit_verify.py.
+"""
+from __future__ import annotations
+import argparse
+from copy import deepcopy
+from fractions import Fraction as Q
+import gzip
+import hashlib
+import json
+from pathlib import Path
+import types
+
+HERE = Path(__file__).resolve().parent
+# Exact frozen exports: PR3734 fa041037..., PR3998 389268a6..., PR3945 91f81d13...
+PINS = {
+ 'source/audit_core.py': (19364,'888679ca8502cdcb100420fe5382a052529a494bf76cb332928b86e62612b951','8b5e1d5f2a106402f036b63028ecef31a4e4c462'),
+ 'source/SOURCE-CREDITS.md': (2882,'d4a13d8b8ed9cdb594f00efb3d1a336e09d934423330f02a1514308a720c278b','c054fad87eeaa510865e682ee0c976c4cd728da2'),
+ 'source/PR3998-README.md': (6498,'d935e7d10e2935ca82a3ba952308d73b78e76a9039cb335682f1fc5ea3b580d7','dcd14ef45bcfb102d030e3f87a0bb2bf80d57ade'),
+ 'source/PR3998-PROOF.md': (11251,'0bbec6ba5b03e40e571f7f4a89e5776441a33b6ede41beeb80153ffad39d124f','afa6241fdaee6da1e567eb5ba74b780f9121ba8d'),
+ 'source/PR3945-NATIVE-LINEAGE-NOTICE': (2456,'ee0c73ca35877aad0e0b6adc560a3881cb82276dba4cf2d4fc58cd19e57cc420','380e979f0e84e00a22bd890615d87cc4684ad740'),
+ 'inputs/BORROW-WORD-base.json.gz': (72769,'5ca4be99851f8878cbfe0fd6c454df926b11218f17feefa1664721cf03156699','674ea18ade4af85ca865fb08dc973637a09e30e6'),
+ 'inputs/BORROW-WORD-extra.json.gz': (72766,'965268a9be73d3f62e7ef680d18f108cf064bfe1f44b82157baa61443232904e','b17e2be67e327f04802fd034c344fe4253c13a4a'),
+}
+SITES = (
+ {'section':'A','indices':[1858,1859,1860],'removed_position':2,'known':True},
+ {'section':'B','indices':[4,53,54],'removed_position':0,'known':False},
+ {'section':'B','indices':[7,49,50],'removed_position':0,'known':False},
+ {'section':'restore','indices':[20,21,36],'removed_position':2,'known':False},
+ {'section':'restore','indices':[22,23,38],'removed_position':2,'known':False},
+ {'section':'restore','indices':[26,27,51],'removed_position':2,'known':False},
+ {'section':'restore','indices':[30,31,62],'removed_position':2,'known':False},
+ {'section':'restore','indices':[32,33,73],'removed_position':2,'known':False},
+)
+# These are PR3998's published canonical known-contraction words, including its
+# ordered basis [42,82] rather than the older source's [82,42].
+KNOWN_WORD = {'base':'32415e93a50f0fdab5fa144da05977fe34bc8340bf967b91c7f7f0bc312df994',
+ 'extra':'439ce3544b68e3dffdb0cb4496fbc92330e825592515bc28c95580807c9d232d'}
+
+
+def need(condition, message):
+ if not condition:
+ raise ValueError(message)
+
+
+def unique(pairs):
+ result = {}
+ for key,value in pairs:
+ need(key not in result,'duplicate JSON key')
+ result[key] = value
+ return result
+
+
+def load_json(path):
+ return json.loads(path.read_bytes(),object_pairs_hook=unique)
+
+
+def canonical(value):
+ return json.dumps(value,sort_keys=True,separators=(',',':'))
+
+
+def digest(value):
+ return hashlib.sha256(canonical(value).encode()).hexdigest()
+
+
+def authenticate():
+ for relative,(length,sha,blob) in PINS.items():
+ path = HERE / relative
+ need(not path.is_symlink(),'source symlink: '+relative)
+ raw = path.read_bytes()
+ need(len(raw)==length and hashlib.sha256(raw).hexdigest()==sha,'source bytes: '+relative)
+ need(hashlib.sha1(b'blob '+str(len(raw)).encode()+b'\0'+raw).hexdigest()==blob,
+ 'source Git blob: '+relative)
+ path = HERE / 'source/audit_core.py'
+ core = types.ModuleType('credited_staged544_audit_core')
+ core.__file__ = str(path)
+ exec(compile(path.read_bytes(),str(path),'exec'),core.__dict__)
+ return core
+
+
+def load_word(provider):
+ need(provider in ('base','extra'),'provider domain')
+ authenticate()
+ return json.loads(gzip.decompress((HERE/'inputs'/('BORROW-WORD-'+provider+'.json.gz')).read_bytes()),
+ object_pairs_hook=unique)
+
+
+def touched(gate):
+ need(isinstance(gate,(tuple,list)) and gate,'scalar operation tuple')
+ kind = gate[0]
+ need(kind in ('add','scale'),'unknown scalar primitive barrier')
+ need(len(gate)==(6 if kind=='add' else 5),'scalar operation arity')
+ registers = gate[1:3] if kind=='add' else gate[1:2]
+ need(all(type(r) is int and 0<=r<704 for r in registers),'scalar register domain')
+ if kind=='add':
+ need(registers[0]!=registers[1],'self ADD')
+ return set(registers)
+
+
+def matrix(gates, registers):
+ rows = {r:{s:Q(r==s) for s in registers} for r in registers}
+ for gate in gates:
+ need(gate[0]=='add','A2 ADD alphabet')
+ need(type(gate[3]) is int and type(gate[4]) is int and gate[4]>0 and gate[3]!=0,
+ 'typed nonzero rational coefficient')
+ destination,source = gate[1:3]
+ coefficient = Q(gate[3],gate[4])
+ donor = rows[source].copy()
+ rows[destination] = {r:rows[destination][r]+coefficient*donor[r] for r in registers}
+ return rows
+
+
+def segments(word,section):
+ if section=='A':
+ cuts = [word['prelude_events'],len(word['A'])]
+ elif section=='B':
+ cuts = sorted({0,len(word['B'])}|{record[0] for record in word['warps']})
+ else:
+ cuts = [0,len(word['restore'])]
+ return list(zip(cuts,cuts[1:]))
+
+
+def certify_site(word,site):
+ need(isinstance(site,dict) and set(site)=={'section','indices','removed_position','known'},'site schema')
+ section,indices,position = site['section'],site['indices'],site['removed_position']
+ need(section in ('A','B','restore'),'site section')
+ need(isinstance(indices,(tuple,list)) and len(indices)==3 and
+ all(type(i) is int for i in indices),'typed site indices')
+ need(0<=indices[0]