diff --git a/proof/region/v1/PROTOCOL.md b/proof/region/v1/PROTOCOL.md index abbe53a5..ac15ca24 100644 --- a/proof/region/v1/PROTOCOL.md +++ b/proof/region/v1/PROTOCOL.md @@ -648,6 +648,102 @@ release не содержит. Family mint manifest: единственный range `[0, 2^24)` и point count `2^24`. Совпадение только point count или reduced-domain candidate этот gate не проходят. +## Semantic verification и `SemanticVerificationReceiptV1` + +Structural dual comparison доказывает только побайтное согласие двух engine +transcripts. Семантическую правильность каждого решения доказывает третий +независимый semantic verifier — stdlib-only Python-пакет `semantic/` внутри +`proof/region/v1`. Он импортирует только `region_proof_protocol` и стандартную +библиотеку; любой импорт кода из `arb/` или `mpfi/`, включая evaluator paths, +запрещён и блокируется контрактным тестом. Совместное наследование между +верификатором и двигателями ограничено неизменяемыми координатами: committed +SSA spec, версия протокола и объявленные wire-грамматики witness digest. + +Верификатор самостоятельно разбирает committed ASCII SSA spec и заново вычисляет +решения региона: строгая интервальная арифметика на dyadic `Fraction` для +знаковых заключений и точный `Fraction` replay для exact zero signal traces. +Никакой hash-compare между engine transcripts, no-op допуск и saturate-all не +являются semantic replay и не могут создать receipt. + +Engine-specific witness digest грамматики. Третий верификатор независимо +воспроизводит exact-zero и accounting грамматики; boundary-грамматики +описывают wire-форму engine-артефакта и не переизобретаются в semantic +replay: + +- exact zero signal trace (общая для обоих двигателей): + `SHA256("labcolors.proof-region.exact-zero-signal-trace.v1\0" || + job_identity || u32be(ordinal) || u64be(exact_branch))`; +- Arb boundary enclosure: + `SHA256("labcolors.arb-boundary-enclosure.v1\0" || job_identity || + u32be(ordinal) || u32be(precision) || u8(formula_status) || + u8(has_enclosure) || для каждого из lower, upper, exponent в fmpz hex: + u64be(len) || text)`; +- MPFI boundary enclosure: тот же каркас с доменом + `"labcolors.mpfi-boundary-enclosure.v1\0"`, где каждое mpfr-значение + кодируется как `u64be(exponent) || u64be(len) || digits`, а `digits` — + вывод `mpfr_get_str(NULL, &exponent, 16, 0, value, MPFR_RNDN)`; +- accounting digest: домен `labcolors.arb-evaluation-accounting.v1\0` или + `labcolors.mpfi-evaluation-accounting.v1\0`, затем job/domain/policy/comparator + identities и на точку `u32be(ordinal) || u32be(precision) || + u64be(consumed) || u8(outcome)`. + +Правила допуска по решению: + +- `Inside` — верификатор доказывает строго отрицательное заключение на всех + пересекающих сегментах (или singleton-предикате); заявленный equality witness + обязан воспроизводиться digest-грамматикой, а точный `Fraction` replay + доказывает ноль предиката на заявленной ветке, которая обязана быть первой + точной; +- `Outside` — верификатор доказывает строго положительное заключение на всех + пересекающих сегментах либо тон точки строго вне диапазона крайних knots; +- `BoundaryUnproven` — точка несёт ровно один boundary witness, выровненный по + ordinal; содержимое enclosure остаётся engine-private координатой, и + верификатор не переизобретает её replay (вышеобъявленные boundary-грамматики + описывают wire-форму engine-артефакта, а не вход третьего верификатора); + верификатор не доказывает определённого противоречащего исхода; +- `ResourceLimitReached` — независимая симуляция grant-правила + `min(per_point_work, remaining_global)` с ordinal-prefix порядком совпадает + со свидетелем, и верификатор не доказывает определённого исхода. + +Два одинаковых неверных transcript не проходят независимый replay: верификатор +вычисляет собственные заключения и никогда не сравнивает transcripts между +собой. Каждая мутация любой транзитивной координаты (decision bits, witness +store, accounting digest, bindings) запрещает допуск. + +`SemanticVerificationReceiptV1` — sealed тип: прямой конструктор без +module-owned token поднимает `TypeError`, наследование типа запрещено +(`final`). Receipt создаёт только +`verify_transcript` после полного успешного semantic replay всех точек; +receipt фиксирует coordinates job, comparator, run claim, transcript и +canonical decision digest. Неудача возвращает typed +`SemanticVerificationRejectedV1` с закрытой суммой причин; receipt при отказе не +создаётся. Неканонические входные объекты отклоняются как `invalid_input` до +replay без поднятия исключений. Receipt является source-bound semantic evidence для одного engine +transcript и не заменяет `DualProofReceiptV1`, который требует receipts для +обоих evaluator paths. + +Binding run claim: верификатор допускает run только когда его job, comparator +и transcript coordinates совпадают с объектами верификации, а связанный +transcript сам привязан к тем же job, domain и comparator. `binary_identity`, +`invocation_identity` и `platform_identity` остаются заявленными execution +coordinates: их причинность source → build → executable доказывает только +source-bound controller соответствующего двигателя, и semantic verifier не +дублирует этот anchor и не объявляет собственный. Run, указывающий на чужой +transcript, отклоняется как `foreign_binding` до replay; transcript, чей +point count дрейфует от bound domain, отклоняется там же. + +Доменные guard'ы операторов следуют объявленным грамматикам: `root3` и `log` +неразрешимы на отрицательных и пересекающих ноль интервалах, `pow_nn` доказывает +точный ноль только при строго положительной экспоненте, а `exp`, `sin` и `cos` +неразрешимы за пределами V1 reduction range (модуль аргумента не выше `2^12`: +quadrant-развёртка `sin`/`cos` ограничена dyadic числом ветвей); каждый такой +случай оставляет точку +`BoundaryUnproven` и никогда не выносит определённого заключения. `sign` +разрешает только строго знаковые интервалы: пересекающий ноль аргумент уводит +точку в `BoundaryUnproven`. Это сознательное ужесточение относительно +продолжения вычисления engine с hull `[-1, 1]` — третий верификатор не вправе +выносить заключение, которое не следует из committed грамматики. + ## Ошибки допуска `ProtocolReasonV1` — закрытая сумма: diff --git a/proof/region/v1/region_proof_protocol.py b/proof/region/v1/region_proof_protocol.py index 3778d6a3..55b4be5a 100644 --- a/proof/region/v1/region_proof_protocol.py +++ b/proof/region/v1/region_proof_protocol.py @@ -262,6 +262,14 @@ def _dyadic(bits: bytes, artifact: str, field: str) -> Fraction: return Fraction(numerator << power, 1) if power >= 0 else Fraction(numerator, 1 << -power) +def dyadic_field_v1(bits: bytes, artifact: str, field: str) -> Fraction: + """Public binary64 decode for committed definition fields.""" + + if type(bits) is not bytes: + _fail(artifact, 0, ProtocolReasonV1.INVALID_DEFINITION, f"{field} is not owned bytes") + return _dyadic(bits, artifact, field) + + def encode_contextual_definition_fields_v1(fields_: tuple[bytes, ...]) -> bytes: return b"".join(_blob(value) for value in fields_) diff --git a/proof/region/v1/semantic/__init__.py b/proof/region/v1/semantic/__init__.py new file mode 100644 index 00000000..22317a1c --- /dev/null +++ b/proof/region/v1/semantic/__init__.py @@ -0,0 +1,20 @@ +"""Independent third-verifier semantic boundary for region proof V1. + +The package replays every transcript decision from immutable job bytes using +only the canonical protocol module and the standard library. It never imports +Arb or MPFI code and never compares engine transcripts against each other. +""" + +from .receipt import ( + SemanticVerificationReasonV1, + SemanticVerificationReceiptV1, + SemanticVerificationRejectedV1, +) +from .verifier import verify_transcript + +__all__ = [ + "SemanticVerificationReasonV1", + "SemanticVerificationReceiptV1", + "SemanticVerificationRejectedV1", + "verify_transcript", +] diff --git a/proof/region/v1/semantic/intervalmath.py b/proof/region/v1/semantic/intervalmath.py new file mode 100644 index 00000000..3d29e5ce --- /dev/null +++ b/proof/region/v1/semantic/intervalmath.py @@ -0,0 +1,628 @@ +"""Rigorous interval arithmetic over dyadic-friendly Fractions. + +Every enclosure keeps exact rational endpoints. Widths arise only from +declared truncation and from Taylor remainder bounds with rational constants; +nothing in this module samples floats into results. Transcendental +enclosures use argument reduction against rational enclosures of ln(2) and +pi, so every returned interval provably contains the mathematical value. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from fractions import Fraction +from functools import lru_cache +from math import ceil, floor, isqrt + + +class UnresolvedError(Exception): + """The requested conclusion needs more guard precision.""" + + +@dataclass(frozen=True) +class Interval: + lo: Fraction + hi: Fraction + + def __post_init__(self) -> None: + if type(self.lo) is not Fraction or type(self.hi) is not Fraction: + raise TypeError("interval bounds must be exact Fractions") + if self.lo > self.hi: + raise ValueError("interval lower bound exceeds upper bound") + + @property + def is_exact(self) -> bool: + return self.lo == self.hi + + def contains_zero(self) -> bool: + return self.lo <= 0 <= self.hi + + def strict_sign(self) -> int | None: + """Return -1/0/+1 only when the interval proves that sign.""" + + if self.lo > 0: + return 1 + if self.hi < 0: + return -1 + if self.lo == 0 and self.hi == 0: + return 0 + return None + + +def exact(value: Fraction | int) -> Interval: + if type(value) is int: + value = Fraction(value) + return Interval(value, value) + + +def _floor_to(value: Fraction, cap_bits: int) -> Fraction: + scale = 1 << cap_bits + numerator = value.numerator * scale + floored = numerator // value.denominator + return Fraction(floored, scale) + + +def _ceil_to(value: Fraction, cap_bits: int) -> Fraction: + scale = 1 << cap_bits + numerator = value.numerator * scale + ceiled = -((-numerator) // value.denominator) + return Fraction(ceiled, scale) + + +def outward(interval: Interval, cap_bits: int) -> Interval: + """Widen endpoints onto a bounded denominator grid; never narrows.""" + + return Interval( + _floor_to(interval.lo, cap_bits), + _ceil_to(interval.hi, cap_bits), + ) + + +def add(left: Interval, right: Interval) -> Interval: + return Interval(left.lo + right.lo, left.hi + right.hi) + + +def sub(left: Interval, right: Interval) -> Interval: + return Interval(left.lo - right.hi, left.hi - right.lo) + + +def neg(value: Interval) -> Interval: + return Interval(-value.hi, -value.lo) + + +def mul(left: Interval, right: Interval) -> Interval: + corners = ( + left.lo * right.lo, + left.lo * right.hi, + left.hi * right.lo, + left.hi * right.hi, + ) + return Interval(min(corners), max(corners)) + + +def div(left: Interval, right: Interval, *, cap_bits: int) -> Interval: + if right.contains_zero(): + raise UnresolvedError("divisor interval contains zero") + corners = ( + left.lo / right.lo, + left.lo / right.hi, + left.hi / right.lo, + left.hi / right.hi, + ) + return outward(Interval(min(corners), max(corners)), cap_bits) + + +def minimum(left: Interval, right: Interval) -> Interval: + return Interval(min(left.lo, right.lo), min(left.hi, right.hi)) + + +def maximum(left: Interval, right: Interval) -> Interval: + return Interval(max(left.lo, right.lo), max(left.hi, right.hi)) + + +def absolute(value: Interval) -> Interval: + if value.lo >= 0: + return value + if value.hi <= 0: + return neg(value) + return Interval(Fraction(0), max(-value.lo, value.hi)) + + +def sign(value: Interval) -> Interval: + resolved = value.strict_sign() + if resolved is None: + raise UnresolvedError("sign cannot be decided at this precision") + return exact(resolved) + + +def _sqrt_bounds(value: Fraction, guard_bits: int) -> tuple[Fraction, Fraction]: + """Rational lower/upper bounds of sqrt(value) via scaled integer roots.""" + + if value == 0: + return Fraction(0), Fraction(0) + scale_power = max(guard_bits, 4) + scaled = value.numerator * value.denominator << (2 * scale_power) + root = isqrt(scaled) + denominator = value.denominator << scale_power + return Fraction(root, denominator), Fraction(root + 1, denominator) + + +def sqrt(value: Interval, *, guard_bits: int, cap_bits: int) -> Interval: + if value.hi < 0: + raise UnresolvedError("sqrt domain requires a nonnegative argument") + if value.lo < 0: + raise UnresolvedError("sqrt domain undecided: interval crosses zero") + lo_lo, _ = _sqrt_bounds(value.lo, guard_bits) + _, hi_hi = _sqrt_bounds(value.hi, guard_bits) + return outward(Interval(lo_lo, hi_hi), cap_bits) + + +def _icbrt(value: int) -> int: + if value < 0: + return -_icbrt(-value) + if value == 0: + return 0 + root = 1 << ((value.bit_length() + 2) // 3) + while True: + step = (2 * root + value // (root * root)) // 3 + if step >= root: + break + root = step + while root * root * root > value: + root -= 1 + while (root + 1) ** 3 <= value: + root += 1 + return root + + +def _root3_bounds(value: Fraction, guard_bits: int) -> tuple[Fraction, Fraction]: + if value == 0: + return Fraction(0), Fraction(0) + scale_power = max(guard_bits, 4) + scaled = value.numerator * value.denominator * value.denominator + scaled <<= 3 * scale_power + root = _icbrt(scaled) + denominator = value.denominator << scale_power + return Fraction(root, denominator), Fraction(root + 1, denominator) + + +def root3(value: Interval, *, guard_bits: int, cap_bits: int) -> Interval: + if value.lo < 0: + raise UnresolvedError("root3 domain requires a nonnegative argument") + if value.lo == value.hi == 0: + return exact(0) + lo_lo, _ = _root3_bounds(value.lo, guard_bits) + _, hi_hi = _root3_bounds(value.hi, guard_bits) + return outward(Interval(lo_lo, hi_hi), cap_bits) + + +def _series_coefficients( + magnitude: Fraction, + terms: int, +) -> tuple[int, int, tuple[int, ...]]: + """Integer Horner data for an odd-power series with (2k+1) divisors. + + For |z| = p/q the partial sum over the first `terms` odd powers equals + sign * p * P(p^2) / (q^(2 terms - 1) * product of odd divisors), where + the integer polynomial P absorbs every denominator factor. + """ + + numerator = magnitude.numerator + denominator = magnitude.denominator + odd_factors = tuple(2 * index + 1 for index in range(terms)) + product = 1 + for factor in odd_factors: + product *= factor + coefficients = tuple( + product // odd_factors[index] + * denominator ** (2 * (terms - 1 - index)) + for index in range(terms) + ) + return numerator, denominator ** (2 * terms - 1) * product, coefficients + + +def _series_partial_sum(z: Fraction, terms: int, odd_signs: bool) -> Fraction: + """Exact partial sum of an odd-power series via one integer Horner pass. + + `odd_signs` selects the alternating series (atan) over the all-positive + one (atanh). Fractions never grow beyond the final common denominator, + so no intermediate value explodes. + """ + + if z == 0 or terms < 1: + return Fraction(0) + sign = -1 if z < 0 else 1 + numerator, total_denominator, coefficients = _series_coefficients(abs(z), terms) + if odd_signs: + coefficients = tuple( + coefficient if index % 2 == 0 else -coefficient + for index, coefficient in enumerate(coefficients) + ) + base = numerator * numerator + total = 0 + for coefficient in reversed(coefficients): + total = total * base + coefficient + return Fraction(sign * numerator * total, total_denominator) + + +def _atanh_sum(z: Fraction, terms: int) -> Fraction: + return _series_partial_sum(z, terms, odd_signs=False) + + +def _atanh_tail(z: Fraction, terms: int) -> Fraction: + magnitude = abs(z) + if magnitude == 0: + return Fraction(0) + square = magnitude * magnitude + return magnitude ** (2 * terms + 1) / ((2 * terms + 1) * (1 - square)) + + +def atanh_enclosure(z: Fraction, guard_bits: int) -> Interval: + """Enclose atanh(z) for |z| < 1 with an adaptively bounded tail.""" + + if not -1 < z < 1: + raise UnresolvedError("atanh argument outside the open unit interval") + target = Fraction(1, 1 << guard_bits) + terms = 1 + while _atanh_tail(z, terms) > target: + terms += 1 + if terms > 4 * guard_bits + 64: + raise UnresolvedError("atanh series converges too slowly") + total = _atanh_sum(z, terms) + tail = _atanh_tail(z, terms) + return Interval(total - tail, total + tail) + + +def _atan_sum(z: Fraction, terms: int) -> Fraction: + return _series_partial_sum(z, terms, odd_signs=True) + + +def atan_enclosure(z: Fraction, guard_bits: int) -> Interval: + """Enclose atan(z) for |z| <= 1 via its alternating series.""" + + if not -1 <= z <= 1: + raise UnresolvedError("atan argument outside the closed unit interval") + target = Fraction(1, 1 << guard_bits) + terms = 1 + magnitude = abs(z) + while magnitude ** (2 * terms + 1) / (2 * terms + 1) > target: + terms += 1 + if terms > 4 * guard_bits + 64: + raise UnresolvedError("atan series converges too slowly") + total = _atan_sum(z, terms) + tail = magnitude ** (2 * terms + 1) / (2 * terms + 1) + return Interval(total - tail, total + tail) + + +@lru_cache(maxsize=8) +def ln2_enclosure(guard_bits: int) -> Interval: + """ln(2) = 2 atanh(1/3), enclosed with a rational tail bound.""" + + base = atanh_enclosure(Fraction(1, 3), guard_bits) + return Interval(2 * base.lo, 2 * base.hi) + + +@lru_cache(maxsize=8) +def pi_enclosure(guard_bits: int) -> Interval: + """Machin formula pi = 16 atan(1/5) - 4 atan(1/239).""" + + first = atan_enclosure(Fraction(1, 5), guard_bits) + second = atan_enclosure(Fraction(1, 239), guard_bits) + lo = 16 * first.lo - 4 * second.hi + hi = 16 * first.hi - 4 * second.lo + return Interval(lo, hi) + + +def _exp_small(value: Interval, guard_bits: int) -> Interval: + """Taylor enclosure of exp on an interval inside [-1, 1]. + + The partial sum runs through the degree-`terms` monomial, so the + Lagrange remainder is bounded by 3 * M^(terms + 1) / (terms + 1)! + because exp(t) < 3 for |t| <= 1. Interval terms accumulate with + denominators capped by the reduced argument, and the adaptive term + count stops as soon as the remainder bound clears the guard target + instead of always iterating 64 terms. + """ + + if value.lo < -1 or value.hi > 1: + raise UnresolvedError("exp reduction interval outside [-1, 1]") + magnitude = max(abs(value.lo), abs(value.hi)) + target = Fraction(1, 1 << guard_bits) + + @lru_cache(maxsize=None) + def radius(terms: int) -> Fraction: + return ( + Fraction( + 3 * magnitude.numerator ** (terms + 1), + magnitude.denominator ** (terms + 1), + ) + / _factorial(terms + 1) + ) + + terms = 2 + while radius(terms) > target: + terms += 1 + if terms > 4 * guard_bits + 64: + raise UnresolvedError("exp series converges too slowly") + total = exact(1) + power = value + factorial = 1 + for index in range(1, terms + 1): + factorial *= index + total = add(total, Interval(power.lo / factorial, power.hi / factorial)) + power = mul(power, value) + remainder = radius(terms) + return Interval(total.lo - remainder, total.hi + remainder) + + +def _factorial(terms: int) -> int: + result = 1 + for factor in range(2, terms + 1): + result *= factor + return result + + +def exp(value: Interval, *, guard_bits: int, cap_bits: int) -> Interval: + # V1 reduction keeps the branch count dyadic-bounded; a hostile binary64 + # argument beyond this range is unresolved instead of exploding the + # 2^branch scale or the float branch estimate. + if absolute(value).hi > Fraction(1 << 12): + raise UnresolvedError("exp argument beyond the V1 reduction range") + ln2 = ln2_enclosure(guard_bits) + midpoint = (value.lo + value.hi) / 2 + # Integer reduction against the rational ln2 enclosure; the loop below + # verifies the remainder bounds rigorously for any branch. + branch = round(float(midpoint) / 0.6931471805599453) + + def reduced(candidate: int) -> Interval: + if candidate >= 0: + return Interval( + value.lo - Fraction(candidate) * ln2.hi, + value.hi - Fraction(candidate) * ln2.lo, + ) + return Interval( + value.lo - Fraction(candidate) * ln2.lo, + value.hi - Fraction(candidate) * ln2.hi, + ) + + if value.hi - value.lo > 1: + raise UnresolvedError("exp input too wide for one reduction branch") + remainder = reduced(branch) + for _ in range(8): + if remainder.lo >= -1 and remainder.hi <= 1: + break + branch += 1 if remainder.hi > 1 else -1 + remainder = reduced(branch) + else: + raise UnresolvedError("exp reduction failed to converge") + core = _exp_small(remainder, guard_bits) + scale = Fraction(2) ** branch + return outward(Interval(scale * core.lo, scale * core.hi), cap_bits) + + +def log(value: Interval, *, guard_bits: int, cap_bits: int) -> Interval: + if value.hi <= 0: + raise UnresolvedError("log domain requires a strictly positive argument") + if value.lo <= 0: + raise UnresolvedError("log domain undecided: interval touches zero") + # Dyadic reduction keeps the mantissa inside [1/2, 2). + branch = 0 + probe = value + while probe.hi >= 2: + branch += 1 + probe = Interval(probe.lo / 2, probe.hi / 2) + while probe.lo < Fraction(1, 2): + branch -= 1 + probe = Interval(probe.lo * 2, probe.hi * 2) + # log(m) = 2 atanh((m - 1)/(m + 1)); the substitution is monotone. + u_lo = (probe.lo - 1) / (probe.lo + 1) + u_hi = (probe.hi - 1) / (probe.hi + 1) + core = Interval( + atanh_enclosure(u_lo, guard_bits).lo, + atanh_enclosure(u_hi, guard_bits).hi, + ) + ln2 = ln2_enclosure(guard_bits) + if branch >= 0: + shift = Interval(Fraction(branch) * ln2.lo, Fraction(branch) * ln2.hi) + else: + shift = Interval(Fraction(branch) * ln2.hi, Fraction(branch) * ln2.lo) + result = add(shift, Interval(2 * core.lo, 2 * core.hi)) + return outward(result, cap_bits) + + +def _sin_small(value: Interval, guard_bits: int) -> Interval: + """Taylor enclosure of sin on an interval inside [-1, 1]. + + The alternating series tail is bounded by the first dropped term, + M^(2 terms + 1) / (2 terms + 1)!, and the term count adapts to the + guard target instead of a fixed 64-term sweep. + """ + + if value.lo < -1 or value.hi > 1: + raise UnresolvedError("sin reduction interval outside [-1, 1]") + magnitude = max(abs(value.lo), abs(value.hi)) + target = Fraction(1, 1 << guard_bits) + + @lru_cache(maxsize=None) + def radius(terms: int) -> Fraction: + return Fraction( + magnitude.numerator ** (2 * terms + 1), + magnitude.denominator ** (2 * terms + 1), + ) / _factorial(2 * terms + 1) + + terms = 1 + while radius(terms) > target: + terms += 1 + if terms > 4 * guard_bits + 64: + raise UnresolvedError("sin series converges too slowly") + total = exact(0) + square = mul(value, value) + power = value + factorial = 1 + for index in range(terms): + if index: + factorial *= (2 * index) * (2 * index + 1) + term = Interval(power.lo / factorial, power.hi / factorial) + total = sub(total, term) if index % 2 else add(total, term) + power = mul(power, square) + remainder = radius(terms) + return Interval(total.lo - remainder, total.hi + remainder) + + +def _cos_small(value: Interval, guard_bits: int) -> Interval: + """Taylor enclosure of cos on an interval inside [-1, 1].""" + + if value.lo < -1 or value.hi > 1: + raise UnresolvedError("cos reduction interval outside [-1, 1]") + magnitude = max(abs(value.lo), abs(value.hi)) + target = Fraction(1, 1 << guard_bits) + + @lru_cache(maxsize=None) + def radius(terms: int) -> Fraction: + return Fraction( + magnitude.numerator ** (2 * terms), + magnitude.denominator ** (2 * terms), + ) / _factorial(2 * terms) + + terms = 1 + while radius(terms) > target: + terms += 1 + if terms > 4 * guard_bits + 64: + raise UnresolvedError("cos series converges too slowly") + total = exact(0) + square = mul(value, value) + power = exact(1) + factorial = 1 + for index in range(terms): + if index: + factorial *= (2 * index - 1) * (2 * index) + term = Interval(power.lo / factorial, power.hi / factorial) + total = add(total, term) if index % 2 == 0 else sub(total, term) + power = mul(power, square) + remainder = radius(terms) + return Interval(total.lo - remainder, total.hi + remainder) + + +def _reduce_quadrant( + value: Interval, + pi: Interval, +) -> list[tuple[int, Interval]]: + """Return candidate (quadrant, remainder) pairs covering the interval.""" + + half_pi = Interval(pi.lo / 2, pi.hi / 2) + low = floor(value.lo / half_pi.hi) - 1 + high = ceil(value.hi / half_pi.lo) + 1 + candidates: list[tuple[int, Interval]] = [] + for branch in range(low, high + 1): + if branch >= 0: + product_lo = Fraction(branch) * half_pi.lo + product_hi = Fraction(branch) * half_pi.hi + else: + product_lo = Fraction(branch) * half_pi.hi + product_hi = Fraction(branch) * half_pi.lo + remainder = Interval(value.lo - product_hi, value.hi - product_lo) + if remainder.lo >= -1 and remainder.hi <= 1: + candidates.append((branch, remainder)) + return candidates + + +def sin(value: Interval, *, guard_bits: int, cap_bits: int) -> Interval: + # V1 reduction keeps the quadrant sweep dyadic-bounded; a hostile binary64 + # argument beyond this range is unresolved instead of exploding the sweep + # count in proportion to the argument magnitude. + if absolute(value).hi > Fraction(1 << 12): + raise UnresolvedError("sin argument beyond the V1 reduction range") + pi = pi_enclosure(guard_bits) + candidates = _reduce_quadrant(value, pi) + lo = Fraction(1) + hi = Fraction(-1) + for branch, remainder in candidates: + match branch % 4: + case 0: + part = _sin_small(remainder, guard_bits) + case 1: + part = _cos_small(remainder, guard_bits) + case 2: + part = neg(_sin_small(remainder, guard_bits)) + case _: + part = neg(_cos_small(remainder, guard_bits)) + lo = min(lo, part.lo) + hi = max(hi, part.hi) + if lo > hi: + raise UnresolvedError("sin reduction found no covering quadrant") + return outward(Interval(lo, hi), cap_bits) + + +def cos(value: Interval, *, guard_bits: int, cap_bits: int) -> Interval: + # Same dyadic sweep bound as sin: huge arguments stay unresolved instead + # of scaling the quadrant enumeration with the argument magnitude. + if absolute(value).hi > Fraction(1 << 12): + raise UnresolvedError("cos argument beyond the V1 reduction range") + pi = pi_enclosure(guard_bits) + candidates = _reduce_quadrant(value, pi) + lo = Fraction(1) + hi = Fraction(-1) + for branch, remainder in candidates: + match branch % 4: + case 0: + part = _cos_small(remainder, guard_bits) + case 1: + part = neg(_sin_small(remainder, guard_bits)) + case 2: + part = neg(_cos_small(remainder, guard_bits)) + case _: + part = _sin_small(remainder, guard_bits) + lo = min(lo, part.lo) + hi = max(hi, part.hi) + if lo > hi: + raise UnresolvedError("cos reduction found no covering quadrant") + return outward(Interval(lo, hi), cap_bits) + + +def pow_pos( + base: Interval, + power_value: Interval, + *, + guard_bits: int, + cap_bits: int, +) -> Interval: + """x^y for strictly positive x, defined as exp(y log x).""" + + return exp( + mul(power_value, log(base, guard_bits=guard_bits, cap_bits=cap_bits)), + guard_bits=guard_bits, + cap_bits=cap_bits, + ) + + +def pow_nn( + base: Interval, + power_value: Interval, + *, + guard_bits: int, + cap_bits: int, +) -> Interval: + """V1 pow_nn: zero base with strictly positive exponent is exact zero.""" + + if base.is_exact and base.lo == 0: + if power_value.lo <= 0: + raise UnresolvedError("pow_nn exponent must be strictly positive") + return exact(0) + if base.contains_zero(): + raise UnresolvedError("pow_nn base undecided at zero") + return pow_pos(base, power_value, guard_bits=guard_bits, cap_bits=cap_bits) + + +def ratio0( + numerator: Interval, + denominator: Interval, + *, + cap_bits: int, +) -> Interval: + """V1 ratio0: 0/0 is exact zero; otherwise a strict positive divisor.""" + + if numerator.is_exact and numerator.lo == 0 and denominator.is_exact and denominator.lo == 0: + return exact(0) + if denominator.lo <= 0: + raise UnresolvedError("ratio0 divisor must be strictly positive") + return div(numerator, denominator, cap_bits=cap_bits) diff --git a/proof/region/v1/semantic/receipt.py b/proof/region/v1/semantic/receipt.py new file mode 100644 index 00000000..dfb2af50 --- /dev/null +++ b/proof/region/v1/semantic/receipt.py @@ -0,0 +1,187 @@ +"""Sealed semantic verification evidence for one engine transcript. + +A receipt certifies that the third verifier independently replayed every +decision of one transcript from immutable job bytes. It does not compare +engine transcripts and it does not mint a dual proof. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from enum import StrEnum +from functools import cached_property + +import region_proof_protocol as protocol + + +_RECEIPT_TOKEN = object() + +RECEIPT_ID_LABEL_V1 = ( + b"labcolors.proof-region.semantic-verification-receipt.v1\0" +) +DECISION_DIGEST_DOMAIN_V1 = b"labcolors.proof-region.resolved-decisions.v1\0" + + +class SemanticVerificationReasonV1(StrEnum): + FOREIGN_BINDING = "foreign_binding" + DECISION_MISMATCH = "decision_mismatch" + WITNESS_REPLAY_MISMATCH = "witness_replay_mismatch" + WITNESS_CONTRADICTION = "witness_contradiction" + RESOURCE_REPLAY_MISMATCH = "resource_replay_mismatch" + ACCOUNTING_REPLAY_MISMATCH = "accounting_replay_mismatch" + REPLAY_UNRESOLVED = "replay_unresolved" + INVALID_INPUT = "invalid_input" + + +@dataclass(frozen=True) +class SemanticVerificationRejectedV1: + reason: SemanticVerificationReasonV1 + ordinal: int + detail: str + + def __post_init__(self) -> None: + if type(self.reason) is not SemanticVerificationReasonV1: + raise TypeError("rejection reason must be SemanticVerificationReasonV1") + if ( + type(self.ordinal) is not int + or self.ordinal < 0 + or self.ordinal >= protocol.OUTPUT_CARDINALITY_V1 + ): + raise TypeError("rejection ordinal outside sRGB8") + if type(self.detail) is not str or not self.detail: + raise TypeError("rejection detail must be a nonempty string") + + +def resolved_decision_digest_v1( + domain_identity: bytes, + decision_bits: bytes, +) -> bytes: + """Canonical resolved-decisions digest shared with dual comparison.""" + + hasher = hashlib.sha256() + hasher.update(DECISION_DIGEST_DOMAIN_V1) + hasher.update(domain_identity) + hasher.update(len(decision_bits).to_bytes(8, "big")) + hasher.update(decision_bits) + return hasher.digest() + + +class SemanticVerificationReceiptV1: + """Verifier-sealed semantic evidence for one engine transcript.""" + + job_identity: bytes + comparator_identity: bytes + run_claim_identity: bytes + transcript_identity: bytes + decision_digest: bytes + + def __new__(cls, *args, **kwargs): + if kwargs.pop("_token", None) is not _RECEIPT_TOKEN: + raise TypeError("SemanticVerificationReceiptV1 is verifier-sealed") + return object.__new__(cls) + + def __init_subclass__(cls, **kwargs) -> None: + # The seal belongs to exactly one type: a subclass would let foreign + # code mint receipts through an inherited constructor. + raise TypeError("SemanticVerificationReceiptV1 is final") + + def __init__( + self, + job_identity: bytes, + comparator_identity: bytes, + run_claim_identity: bytes, + transcript_identity: bytes, + decision_digest: bytes, + *, + _token: object = None, + ) -> None: + for name in ( + "job_identity", + "comparator_identity", + "run_claim_identity", + "transcript_identity", + "decision_digest", + ): + value = locals()[name] + if type(value) is not bytes or len(value) != 32 or value == bytes(32): + raise TypeError(f"semantic receipt coordinate {name} is not a digest") + object.__setattr__(self, "job_identity", job_identity) + object.__setattr__(self, "comparator_identity", comparator_identity) + object.__setattr__(self, "run_claim_identity", run_claim_identity) + object.__setattr__(self, "transcript_identity", transcript_identity) + object.__setattr__(self, "decision_digest", decision_digest) + + def __setattr__(self, name: str, value: object) -> None: + raise AttributeError("SemanticVerificationReceiptV1 is immutable") + + @classmethod + def _seal( + cls, + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, + run: protocol.RunClaimV1, + transcript: protocol.DecisionTranscriptV1, + ) -> "SemanticVerificationReceiptV1": + """Only the verifier may cross the seal after a complete replay.""" + + return cls( + job.identity, + comparator.identity, + run.identity, + transcript.identity, + resolved_decision_digest_v1( + transcript.domain_identity, + transcript.decision_bits, + ), + _token=_RECEIPT_TOKEN, + ) + + def encode(self) -> bytes: + return b"".join( + getattr(self, name) + for name in ( + "job_identity", + "comparator_identity", + "run_claim_identity", + "transcript_identity", + "decision_digest", + ) + ) + + @cached_property + def identity(self) -> bytes: + encoded = self.encode() + hasher = hashlib.sha256() + hasher.update(RECEIPT_ID_LABEL_V1) + hasher.update(len(encoded).to_bytes(8, "big")) + hasher.update(encoded) + return hasher.digest() + + def binds( + self, + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, + run: protocol.RunClaimV1, + transcript: protocol.DecisionTranscriptV1, + ) -> bool: + """Replay every binding coordinate against live canonical objects.""" + + if ( + type(job) is not protocol.ProofJobV1 + or type(comparator) is not protocol.ContentResolvedComparatorManifestV2 + or type(run) is not protocol.RunClaimV1 + or type(transcript) is not protocol.DecisionTranscriptV1 + ): + return False + return ( + self.job_identity == job.identity + and self.comparator_identity == comparator.identity + and self.run_claim_identity == run.identity + and self.transcript_identity == transcript.identity + and self.decision_digest + == resolved_decision_digest_v1( + transcript.domain_identity, + transcript.decision_bits, + ) + ) diff --git a/proof/region/v1/semantic/region.py b/proof/region/v1/semantic/region.py new file mode 100644 index 00000000..f008357b --- /dev/null +++ b/proof/region/v1/semantic/region.py @@ -0,0 +1,261 @@ +"""Semantic replay of the V1 region decision rules. + +This module mirrors the public decision semantics of the engine region code +on rigorous Fraction intervals: the same outcome ladder, the same branch +accounting, and the same exact-boundary discipline. It never samples floats +into decisions; every comparison is an exact rational comparison. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from fractions import Fraction + +import region_proof_protocol as protocol + +from . import intervalmath +from .ssa import EvaluationContext, SemanticFormulaError + +# Local outcome codes are the canonical protocol decision tags; binding them +# to the enum makes any drift fail at module load instead of at replay time. +INSIDE = int(protocol.DecisionV1.INSIDE) +OUTSIDE = int(protocol.DecisionV1.OUTSIDE) +BOUNDARY_UNPROVEN = int(protocol.DecisionV1.BOUNDARY_UNPROVEN) +RESOURCE_LIMIT_REACHED = int(protocol.DecisionV1.RESOURCE_LIMIT_REACHED) + + +@dataclass(frozen=True) +class Knot: + tone: Fraction + center_a: Fraction + center_b: Fraction + radius_squared: Fraction + + +@dataclass(frozen=True) +class Region: + knots: tuple[Knot, ...] + metric_aa: Fraction + metric_ab: Fraction + metric_bb: Fraction + + @classmethod + def from_definition(cls, definition: protocol.ContextualRegionDefinitionV1) -> Region: + def dyadic(index: int) -> Fraction: + return protocol.dyadic_field_v1(definition.fields[index], "semantic-region", "coordinate") + + knots = tuple( + Knot( + dyadic(22 + 4 * index), + dyadic(23 + 4 * index), + dyadic(24 + 4 * index), + dyadic(25 + 4 * index), + ) + for index in range(definition.knot_count) + ) + return cls(knots, dyadic(18), dyadic(19), dyadic(20)) + + +def context_inputs(definition: protocol.ContextualRegionDefinitionV1) -> dict[str, object]: + """Exact real inputs shared by every point lift.""" + + adapting = protocol.dyadic_field_v1(definition.fields[11], "semantic-region", "adapting_luminance") + ratio = protocol.dyadic_field_v1(definition.fields[12], "semantic-region", "background_ratio") + surround = definition.fields[13][0] + return { + "adapting_luminance": intervalmath.exact(adapting), + "background_ratio": intervalmath.exact(ratio), + "surround": surround, + } + + +def ordinal_to_rgb(ordinal: int) -> tuple[int, int, int]: + return (ordinal >> 16) & 0xFF, (ordinal >> 8) & 0xFF, ordinal & 0xFF + + +@dataclass(frozen=True) +class DecisionResult: + outcome: int + consumed_branches: int + exact_boundary: bool + exact_branch: int + + +def _equal(left: intervalmath.Interval, right: intervalmath.Interval) -> bool: + return left.lo == right.lo and left.hi == right.hi + + +def _overlaps(left: intervalmath.Interval, right: intervalmath.Interval) -> bool: + return not (left.hi < right.lo or right.hi < left.lo) + + +def _intersection( + left: intervalmath.Interval, + right: intervalmath.Interval, +) -> intervalmath.Interval | None: + lo = max(left.lo, right.lo) + hi = min(left.hi, right.hi) + if lo > hi: + return None + return intervalmath.Interval(lo, hi) + + +def _predicate_decision( + ssa: EvaluationContext, + program_name: str, + inputs: dict[str, object], +) -> tuple[bool, bool, bool, bool]: + """Return (resolved, inside, outside, exact_zero) for one predicate run.""" + + output = {"singleton": "singleton_f", "segment": "segment_f"}[program_name] + try: + predicate = ssa.evaluate(ssa.formula.program(program_name), inputs)[output] + except (intervalmath.UnresolvedError, SemanticFormulaError): + return False, False, False, False + return ( + True, + predicate.hi <= 0, + predicate.lo > 0, + predicate.lo == 0 and predicate.hi == 0, + ) + + +def _evaluate_singleton( + ssa: EvaluationContext, + point: tuple[intervalmath.Interval, ...], + region: Region, + grant: int, +) -> DecisionResult: + knot = region.knots[0] + tone = intervalmath.exact(knot.tone) + if not _equal(point[0], tone): + outcome = BOUNDARY_UNPROVEN if _overlaps(point[0], tone) else OUTSIDE + return DecisionResult(outcome, 0, False, 0) + if grant == 0: + return DecisionResult(RESOURCE_LIMIT_REACHED, 0, False, 0) + resolved, inside, outside, exact_zero = _predicate_decision( + ssa, + "singleton", + { + "singleton_a": point[1], + "singleton_b": point[2], + "singleton_ca": intervalmath.exact(knot.center_a), + "singleton_cb": intervalmath.exact(knot.center_b), + "singleton_rho": intervalmath.exact(knot.radius_squared), + "singleton_g00": intervalmath.exact(region.metric_aa), + "singleton_g01": intervalmath.exact(region.metric_ab), + "singleton_g11": intervalmath.exact(region.metric_bb), + }, + ) + if resolved and inside: + return DecisionResult(INSIDE, 1, exact_zero, 0) + if resolved and outside: + return DecisionResult(OUTSIDE, 1, False, 0) + return DecisionResult(BOUNDARY_UNPROVEN, 1, False, 0) + + +def decide( + ssa: EvaluationContext, + point: tuple[intervalmath.Interval, ...], + region: Region, + precision: int, + grant: int, +) -> DecisionResult: + """Replay the public region decision entry point on rigorous intervals.""" + + if not region.knots: + raise protocol.ProtocolErrorV1( + "semantic-region", + 0, + protocol.ProtocolReasonV1.INVALID_DEFINITION, + "region definition carries no knots", + ) + if precision < 2: + return DecisionResult(BOUNDARY_UNPROVEN, 0, False, 0) + if len(region.knots) == 1: + return _evaluate_singleton(ssa, point, region, grant) + + tone = point[0] + first = intervalmath.exact(region.knots[0].tone) + last = intervalmath.exact(region.knots[-1].tone) + if tone.hi < first.lo or tone.lo > last.hi: + return DecisionResult(OUTSIDE, 0, False, 0) + outside_possible = not (tone.lo >= first.hi) or not (tone.hi <= last.lo) + + any_segment = False + all_inside = True + all_outside = True + exact_zero = False + exact_branch = 0 + consumed = 0 + for index in range(len(region.knots) - 1): + left = region.knots[index] + right = region.knots[index + 1] + segment_domain = intervalmath.Interval( + min(left.tone, right.tone), + max(left.tone, right.tone), + ) + intersection = _intersection(tone, segment_domain) + if intersection is None: + continue + any_segment = True + if consumed >= grant: + return DecisionResult(RESOURCE_LIMIT_REACHED, consumed, False, 0) + inputs = { + "segment_t": intersection, + "segment_a": point[1], + "segment_b": point[2], + "segment_t0": intervalmath.exact(left.tone), + "segment_t1": intervalmath.exact(right.tone), + "segment_c0a": intervalmath.exact(left.center_a), + "segment_c0b": intervalmath.exact(left.center_b), + "segment_c1a": intervalmath.exact(right.center_a), + "segment_c1b": intervalmath.exact(right.center_b), + "segment_rho0": intervalmath.exact(left.radius_squared), + "segment_rho1": intervalmath.exact(right.radius_squared), + "segment_g00": intervalmath.exact(region.metric_aa), + "segment_g01": intervalmath.exact(region.metric_ab), + "segment_g11": intervalmath.exact(region.metric_bb), + } + consumed += 1 + resolved, inside, outside, branch_exact = _predicate_decision(ssa, "segment", inputs) + if not resolved: + all_inside = False + all_outside = False + continue + all_inside = all_inside and inside + all_outside = all_outside and outside + if branch_exact and not exact_zero: + exact_branch = index + exact_zero = exact_zero or branch_exact + + if not any_segment: + return DecisionResult(BOUNDARY_UNPROVEN, consumed, False, 0) + if all_outside: + return DecisionResult(OUTSIDE, consumed, False, 0) + if all_inside and not outside_possible: + return DecisionResult(INSIDE, consumed, exact_zero, exact_branch) + return DecisionResult(BOUNDARY_UNPROVEN, consumed, False, 0) + + +def evaluate_rgb( + ssa: EvaluationContext, + ordinal: int, + region: Region, + shared_inputs: dict[str, object], + precision: int, + grant: int, +) -> DecisionResult: + """Replay one point: exact-real lift, then the decision rules.""" + + red, green, blue = ordinal_to_rgb(ordinal) + inputs = dict(shared_inputs) + inputs["r8"] = red + inputs["g8"] = green + inputs["b8"] = blue + try: + outputs = ssa.evaluate(ssa.formula.program("point"), inputs) + point = (outputs["jp"], outputs["ap"], outputs["bp"]) + except (intervalmath.UnresolvedError, SemanticFormulaError, KeyError): + return DecisionResult(BOUNDARY_UNPROVEN, 0, False, 0) + return decide(ssa, point, region, precision, grant) diff --git a/proof/region/v1/semantic/replay.py b/proof/region/v1/semantic/replay.py new file mode 100644 index 00000000..199f8f9d --- /dev/null +++ b/proof/region/v1/semantic/replay.py @@ -0,0 +1,156 @@ +"""Per-point semantic replay over the declared budget and digest grammars. + +The replay recomputes every decision from immutable job bytes: the precision +ladder, the ordinal-prefix grant accounting, and the engine-shared exact-zero +and accounting digest grammars. Boundary enclosures stay engine-private; the +verifier only consumes their presence and alignment. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass + +import region_proof_protocol as protocol + +from . import region +from .ssa import EvaluationContext, parse_formula + +EXACT_TRACE_DOMAIN_V1 = b"labcolors.proof-region.exact-zero-signal-trace.v1\0" + +ACCOUNTING_DOMAINS_V1 = { + protocol.ComparatorKindV1.ARB: b"labcolors.arb-evaluation-accounting.v1\0", + protocol.ComparatorKindV1.MPFI: b"labcolors.mpfi-evaluation-accounting.v1\0", +} + + +class ReplayIntegrityError(RuntimeError): + """The independent replay contradicted its own accounting invariants.""" + + def __init__(self, ordinal: int, detail: str) -> None: + super().__init__(f"ordinal {ordinal}: {detail}") + self.ordinal = ordinal + self.detail = detail + + +@dataclass(frozen=True) +class PointReplay: + ordinal: int + outcome: int + final_precision: int + consumed: int + point_grant: int + resource_scope: int + exact_boundary: bool + exact_branch: int + + +def exact_trace_digest_v1(job_identity: bytes, ordinal: int, exact_branch: int) -> bytes: + """Engine-shared grammar: one job, one ordinal, one exact branch.""" + + hasher = hashlib.sha256() + hasher.update(EXACT_TRACE_DOMAIN_V1) + hasher.update(job_identity) + hasher.update(ordinal.to_bytes(4, "big")) + hasher.update(exact_branch.to_bytes(8, "big")) + return hasher.digest() + + +def accounting_prefix_v1( + kind: protocol.ComparatorKindV1, + job: protocol.ProofJobV1, + comparator_identity: bytes, +) -> hashlib.sha256: + domain = ACCOUNTING_DOMAINS_V1.get(kind) + if domain is None: + raise ReplayIntegrityError(0, f"no accounting domain for comparator kind {kind!r}") + hasher = hashlib.sha256() + hasher.update(domain) + hasher.update(job.identity) + hasher.update(job.domain.identity) + hasher.update(job.policy.identity) + hasher.update(comparator_identity) + return hasher + + +def account_record(ordinal: int, precision: int, consumed: int, outcome: int) -> bytes: + return ( + ordinal.to_bytes(4, "big") + + precision.to_bytes(4, "big") + + consumed.to_bytes(8, "big") + + bytes((outcome,)) + ) + + +class SemanticReplay: + """Stateful replay driver mirroring the engine evaluation loop.""" + + def __init__( + self, + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, + ) -> None: + budget = next( + item for item in job.policy.comparators if item.kind == comparator.manifest.kind + ) + if not budget.precision_ladder: + raise ReplayIntegrityError(0, "comparator budget carries an empty precision ladder") + self._job = job + self._budget = budget + self._formula = parse_formula(job.formula_spec) + self._region = region.Region.from_definition(job.definition) + self._shared_inputs = region.context_inputs(job.definition) + # The domain ordinals are consumed sequentially, exactly like the + # engine loop; materialising up to 2^24 integers would only waste + # memory without changing the replay semantics. + self._ordinals = iter(job.domain.iter_ordinals()) + self._global_remaining = budget.global_pregrant + + @property + def budget(self) -> protocol.ComparatorBudgetV1: + return self._budget + + def next_point(self) -> PointReplay: + """Replay one domain point exactly like the engine loop does.""" + + try: + ordinal = next(self._ordinals) + except StopIteration: + raise ReplayIntegrityError(0, "domain ordinal stream exhausted early") from None + budget = self._budget + point_grant = min(budget.per_point_work, self._global_remaining) + point_remaining = point_grant + point_consumed = 0 + resource_scope = 1 if budget.per_point_work <= self._global_remaining else 2 + # A point owns its ordinal-prefix pregrant even when it uses none. + self._global_remaining -= point_grant + ladder = budget.precision_ladder + final_precision = ladder[0] + final = region.DecisionResult(region.BOUNDARY_UNPROVEN, 0, False, 0) + for rung in ladder: + final_precision = rung + ssa = EvaluationContext(self._formula, rung, rung) + final = region.evaluate_rgb( + ssa, + ordinal, + self._region, + self._shared_inputs, + rung, + point_remaining, + ) + if final.consumed_branches > point_remaining: + raise ReplayIntegrityError(ordinal, "predicate consumed more than granted") + point_remaining -= final.consumed_branches + point_consumed += final.consumed_branches + if final.outcome != region.BOUNDARY_UNPROVEN: + break + return PointReplay( + ordinal=ordinal, + outcome=final.outcome, + final_precision=final_precision, + consumed=point_consumed, + point_grant=point_grant, + resource_scope=resource_scope, + exact_boundary=final.exact_boundary, + exact_branch=final.exact_branch, + ) diff --git a/proof/region/v1/semantic/ssa.py b/proof/region/v1/semantic/ssa.py new file mode 100644 index 00000000..62ddb870 --- /dev/null +++ b/proof/region/v1/semantic/ssa.py @@ -0,0 +1,475 @@ +"""Independent strict interpreter for the V1 exact-real SSA. + +The third verifier must not trust the engine's parser or code generator: +this module re-parses the immutable formula bytes with its own strict reader +and evaluates programs over rigorous interval values. Only the protocol +release digest pins the accepted content, recomputed here from the declared +domain label. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from fractions import Fraction +from functools import cached_property + +import region_proof_protocol as protocol + +from . import intervalmath + + +class SemanticFormulaError(ValueError): + """The formula bytes are not the registered V1 exact-real SSA.""" + + +TYPE_DECLARATIONS = ( + "type u8 unsigned_integer_0_255", + "type real mathematical_real", + "type bool exact_boolean", + "type surround_profile closed_enum", +) + +OPERATOR_DECLARATIONS = ( + "operator lookup 2 real table_u8_exact_dyadic_at_ordinal", + "operator eq 2 bool exact_same_type_equality", + "operator select 3 same bool_true_second_else_third", + "operator add 2 real exact_x_plus_y", + "operator sub 2 real exact_x_minus_y", + "operator mul 2 real exact_x_times_y", + "operator div 2 real domain_y_ne_zero_x_div_y_else_domain_unproven", + "operator min 2 real exact_lesser_real", + "operator max 2 real exact_greater_real", + "operator root3 1 real domain_x_ge_zero_unique_y_ge_zero_y_cubed_eq_x_else_domain_unproven", + "operator sqrt 1 real domain_x_ge_zero_unique_y_ge_zero_y_squared_eq_x_else_domain_unproven", + "operator exp 1 real analytic_natural_exponential", + "operator log 1 real domain_x_gt_zero_analytic_natural_logarithm_else_domain_unproven", + "operator sin 1 real analytic_sine_radians", + "operator cos 1 real analytic_cosine_radians", + "operator abs 1 real exact_absolute_value", + "operator sign 1 real negative_minus_one_zero_zero_positive_one", + "operator pow_pos 2 real domain_x_gt_zero_exp_y_mul_log_x_else_domain_unproven", + "operator pow_nn 2 real if_x_eq_zero_and_y_gt_zero_zero_else_pow_pos", + "operator ratio0 2 real if_x_eq_zero_and_y_eq_zero_zero_else_domain_y_gt_zero_x_div_y", +) + +DRIVER_RULES = ( + "rule tone_domain closed_first_last", + "rule out_of_tone_domain outside", + "rule one_knot_tone exact_equality_required", + "rule one_knot_predicate singleton_f_le_zero", + "rule multi_knot_predicate piecewise_linear_segment_f_le_zero", + "rule boundary inclusive", +) + +PROGRAM_INTERFACES = { + "point": ( + (("r8", "u8"), ("g8", "u8"), ("b8", "u8"), + ("adapting_luminance", "real"), ("background_ratio", "real"), + ("surround", "surround_profile")), + 226, + ("jp", "ap", "bp"), + ), + "segment": ( + tuple( + (name, "real") + for name in ( + "segment_t", "segment_a", "segment_b", "segment_t0", + "segment_t1", "segment_c0a", "segment_c0b", "segment_c1a", + "segment_c1b", "segment_rho0", "segment_rho1", "segment_g00", + "segment_g01", "segment_g11", + ) + ), + 27, + ("segment_f",), + ), + "singleton": ( + tuple( + (name, "real") + for name in ( + "singleton_a", "singleton_b", "singleton_ca", "singleton_cb", + "singleton_rho", "singleton_g00", "singleton_g01", "singleton_g11", + ) + ), + 12, + ("singleton_f",), + ), +} + + +@dataclass(frozen=True) +class SemanticNode: + name: str + result: str + operator: str + arguments: tuple[str, ...] + + +@dataclass(frozen=True) +class SemanticProgram: + name: str + inputs: tuple[tuple[str, str], ...] + nodes: tuple[SemanticNode, ...] + outputs: tuple[str, ...] + + +@dataclass(frozen=True) +class SemanticFormula: + decode: tuple[Fraction, ...] + literals: tuple[tuple[str, int], ...] + enums: tuple[tuple[str, int], ...] + programs: tuple[SemanticProgram, ...] + + def program(self, name: str) -> SemanticProgram: + for program in self.programs: + if program.name == name: + return program + raise SemanticFormulaError(f"missing program {name}") + + +class _Lines: + def __init__(self, values: list[str]): + self.values = values + self.cursor = 0 + + def next(self) -> str: + if self.cursor >= len(self.values): + raise SemanticFormulaError(f"unexpected end at line {self.cursor + 1}") + value = self.values[self.cursor] + self.cursor += 1 + return value + + def expect(self, expected: str) -> None: + actual = self.next() + if actual != expected: + raise SemanticFormulaError( + f"line {self.cursor}: expected {expected!r}, got {actual!r}" + ) + + +def _identifier(value: str) -> bool: + return bool(value) and value[0].islower() and all( + byte.islower() or byte.isdigit() or byte == "_" for byte in value + ) + + +def _fields(line: str, count: int) -> tuple[str, ...]: + result = tuple(line.split(" ")) + if len(result) != count: + raise SemanticFormulaError(f"record arity {len(result)} != {count}") + return result + + +def binary64_to_fraction(bits: int) -> Fraction: + """Exact binary64 value as a dyadic Fraction; rejects nonfinite payloads.""" + + if bits & 0x7FF0_0000_0000_0000 == 0x7FF0_0000_0000_0000: + raise SemanticFormulaError("nonfinite binary64 payload") + if bits == 0x8000_0000_0000_0000: + raise SemanticFormulaError("negative zero") + sign = -1 if bits >> 63 else 1 + exponent = (bits >> 52) & 0x7FF + fraction = bits & ((1 << 52) - 1) + if exponent == 0: + significand = fraction + power = -1074 + else: + significand = (1 << 52) | fraction + power = exponent - 1075 + numerator = sign * significand + if power >= 0: + return Fraction(numerator << power, 1) + return Fraction(numerator, 1 << -power) + + +def _finite_bits(token: str) -> int: + if len(token) != 16 or any(byte not in "0123456789abcdef" for byte in token): + raise SemanticFormulaError("noncanonical binary64 payload") + return int(token, 16) + + +def _validate_node(node: SemanticNode, symbols: dict[str, str]) -> None: + try: + types = tuple(symbols[value] for value in node.arguments) + except KeyError as error: + raise SemanticFormulaError( + f"unknown or forward reference {error.args[0]}" + ) from None + unary = {"root3", "sqrt", "exp", "log", "sin", "cos", "abs", "sign"} + binary = {"add", "sub", "mul", "div", "min", "max", "pow_pos", "pow_nn", "ratio0"} + if node.operator == "lookup": + valid = node.result == "real" and types == ("decode_table", "u8") + elif node.operator == "eq": + # Equality over real values is undecidable on intervals; the parser + # admits discrete operands only, which matches the registered V1 + # formula (it compares the surround enum exclusively). + valid = ( + node.result == "bool" + and len(types) == 2 + and types[0] == types[1] + and types[0] in ("u8", "surround_profile", "bool") + ) + elif node.operator == "select": + valid = len(types) == 3 and types[0] == "bool" and types[1] == types[2] == node.result + elif node.operator in unary: + valid = node.result == "real" and types == ("real",) + elif node.operator in binary: + valid = node.result == "real" and types == ("real", "real") + else: + valid = False + if not valid: + raise SemanticFormulaError(f"operator/type mismatch for {node.name}") + + +def _parse_program(lines: _Lines, name: str, globals_: dict[str, str]) -> SemanticProgram: + expected_inputs, expected_nodes, expected_outputs = PROGRAM_INTERFACES[name] + lines.expect(f"{name}_inputs {len(expected_inputs)}") + symbols = dict(globals_) + inputs: list[tuple[str, str]] = [] + for expected in expected_inputs: + record = _fields(lines.next(), 3) + if record != ("input", *expected): + raise SemanticFormulaError(f"foreign {name} input") + if record[1] in symbols: + raise SemanticFormulaError("shadowed input") + symbols[record[1]] = record[2] + inputs.append((record[1], record[2])) + + lines.expect(f"{name}_nodes {expected_nodes}") + nodes: list[SemanticNode] = [] + for _ in range(expected_nodes): + record = tuple(lines.next().split(" ")) + if len(record) < 5 or record[0] != "node" or not _identifier(record[1]): + raise SemanticFormulaError("invalid node") + node = SemanticNode(record[1], record[2], record[3], record[4:]) + _validate_node(node, symbols) + if node.name in symbols: + raise SemanticFormulaError("shadowed node") + symbols[node.name] = node.result + nodes.append(node) + + if name == "point": + lines.expect("point_checkpoints 39") + checkpoint_names: set[str] = set() + node_names = {node.name for node in nodes} + for _ in range(39): + record = _fields(lines.next(), 3) + if ( + record[0] != "checkpoint" + or record[1] in checkpoint_names + or record[2] not in node_names + or symbols[record[2]] != "real" + ): + raise SemanticFormulaError("invalid checkpoint") + checkpoint_names.add(record[1]) + + lines.expect(f"{name}_outputs {len(expected_outputs)}") + outputs: list[str] = [] + for expected in expected_outputs: + record = _fields(lines.next(), 3) + if record != ("output", expected, "real") or symbols.get(expected) != "real": + raise SemanticFormulaError("foreign output") + outputs.append(expected) + return SemanticProgram(name, tuple(inputs), tuple(nodes), tuple(outputs)) + + +def parse_formula(source: bytes) -> SemanticFormula: + """Strictly re-parse the immutable V1 formula bytes.""" + + if type(source) is not bytes: + raise SemanticFormulaError("formula source must be owned bytes") + release = hashlib.sha256( + protocol.FORMULA_RELEASE_DOMAIN_V1 + + len(source).to_bytes(8, "big") + + source + ).hexdigest() + if release != protocol.FORMULA_RELEASE_V1.hex(): + raise SemanticFormulaError("formula release mismatch") + if not source.isascii() or not source.endswith(b"\n") or source.endswith(b"\n\n"): + raise SemanticFormulaError("formula is not canonical ASCII with one final LF") + text = source.decode("ascii")[:-1] + values = text.split("\n") + for index, line in enumerate(values, 1): + if ( + not line + or line.startswith(" ") + or line.endswith(" ") + or " " in line + or "\t" in line + or "\r" in line + or "#" in line + ): + raise SemanticFormulaError(f"line {index} is not canonical") + + lines = _Lines(values) + lines.expect("labcolors_exact_real_ssa 1") + lines.expect("arithmetic exact_real_v1") + lines.expect(f"types {len(TYPE_DECLARATIONS)}") + for declaration in TYPE_DECLARATIONS: + lines.expect(declaration) + lines.expect(f"operators {len(OPERATOR_DECLARATIONS)}") + for declaration in OPERATOR_DECLARATIONS: + lines.expect(declaration) + + lines.expect("decode_table decode_srgb8 256") + decode: list[Fraction] = [] + for ordinal in range(256): + record = _fields(lines.next(), 3) + if record[:2] != ("decode", f"{ordinal:02x}"): + raise SemanticFormulaError("decode order drift") + decode.append(binary64_to_fraction(_finite_bits(record[2]))) + + lines.expect("literals 56") + literals: list[tuple[str, int]] = [] + literal_names: set[str] = set() + literal_values: set[int] = set() + for _ in range(56): + record = _fields(lines.next(), 3) + bits = _finite_bits(record[2]) + if ( + record[0] != "literal" + or not _identifier(record[1]) + or record[1] in literal_names + or bits in literal_values + ): + raise SemanticFormulaError("invalid literal") + literal_names.add(record[1]) + literal_values.add(bits) + literals.append((record[1], bits)) + + lines.expect("enum_type surround_profile 3") + enums: list[tuple[str, int]] = [] + for name, tag in (("surround_average", 1), ("surround_dim", 2), ("surround_dark", 3)): + record = _fields(lines.next(), 4) + if record != ("enum", "surround_profile", name, f"{tag:02x}"): + raise SemanticFormulaError("foreign surround enum") + enums.append((name, tag)) + + globals_: dict[str, str] = {"decode_srgb8": "decode_table"} + globals_.update((name, "real") for name, _ in literals) + globals_.update((name, "surround_profile") for name, _ in enums) + programs = tuple(_parse_program(lines, name, globals_) for name in PROGRAM_INTERFACES) + lines.expect(f"driver {len(DRIVER_RULES)}") + for rule in DRIVER_RULES: + lines.expect(rule) + lines.expect("end") + if lines.cursor != len(lines.values): + raise SemanticFormulaError("trailing records") + return SemanticFormula(tuple(decode), tuple(literals), tuple(enums), programs) + + +@dataclass(frozen=True) +class EvaluationContext: + """Fixed precision policy for one replay rung.""" + + formula: SemanticFormula + guard_bits: int + cap_bits: int + + def evaluate( + self, + program: SemanticProgram, + inputs: dict[str, object], + ) -> dict[str, object]: + """Run one program; real values are intervals, u8/enum/bool are ints.""" + + environment: dict[str, object] = dict(self._literal_environment) + for name, kind in program.inputs: + if name not in inputs: + raise SemanticFormulaError(f"missing input {name}") + value = inputs[name] + if kind == "real": + if type(value) is not intervalmath.Interval: + raise SemanticFormulaError(f"input {name} must be an interval") + elif kind in ("u8", "surround_profile"): + if type(value) is not int: + raise SemanticFormulaError(f"input {name} must be an integer") + environment[name] = value + + for node in program.nodes: + environment[node.name] = self._evaluate_node(node, environment) + outputs: dict[str, object] = {} + for name in program.outputs: + value = environment[name] + if type(value) is not intervalmath.Interval: + raise SemanticFormulaError(f"output {name} is not a real value") + outputs[name] = value + return outputs + + @cached_property + def _literal_environment(self) -> dict[str, object]: + # Literal and enum bindings are pinned by the release digest; decoding + # them once instead of once per point (2^24+ evaluations per full + # domain) removes a quadratic-fraction hot path from the replay. + environment: dict[str, object] = {} + for name, bits in self.formula.literals: + environment[name] = intervalmath.exact(binary64_to_fraction(bits)) + for name, tag in self.formula.enums: + environment[name] = tag + return environment + + def _evaluate_node( + self, + node: SemanticNode, + environment: dict[str, object], + ) -> object: + operator = node.operator + guard = self.guard_bits + cap = self.cap_bits + if operator == "lookup": + if node.arguments[0] != "decode_srgb8": + raise SemanticFormulaError("lookup against foreign table") + index = environment[node.arguments[1]] + if type(index) is not int or index < 0 or index >= len(self.formula.decode): + raise SemanticFormulaError("lookup index outside the decode table") + return intervalmath.exact(self.formula.decode[index]) + arguments = tuple(environment[name] for name in node.arguments) + if operator == "eq": + left, right = arguments + if type(left) is not int or type(right) is not int: + raise SemanticFormulaError("equality over non-discrete values") + return 1 if left == right else 0 + if operator == "select": + condition, chosen, fallback = arguments + if type(condition) is not int: + raise SemanticFormulaError("select over non-discrete condition") + return chosen if condition else fallback + if operator == "add": + return intervalmath.add(arguments[0], arguments[1]) + if operator == "sub": + return intervalmath.sub(arguments[0], arguments[1]) + if operator == "mul": + return intervalmath.mul(arguments[0], arguments[1]) + if operator == "div": + return intervalmath.div(arguments[0], arguments[1], cap_bits=cap) + if operator == "min": + return intervalmath.minimum(arguments[0], arguments[1]) + if operator == "max": + return intervalmath.maximum(arguments[0], arguments[1]) + if operator == "root3": + return intervalmath.root3(arguments[0], guard_bits=guard, cap_bits=cap) + if operator == "sqrt": + return intervalmath.sqrt(arguments[0], guard_bits=guard, cap_bits=cap) + if operator == "exp": + return intervalmath.exp(arguments[0], guard_bits=guard, cap_bits=cap) + if operator == "log": + return intervalmath.log(arguments[0], guard_bits=guard, cap_bits=cap) + if operator == "sin": + return intervalmath.sin(arguments[0], guard_bits=guard, cap_bits=cap) + if operator == "cos": + return intervalmath.cos(arguments[0], guard_bits=guard, cap_bits=cap) + if operator == "abs": + return intervalmath.absolute(arguments[0]) + if operator == "sign": + return intervalmath.sign(arguments[0]) + if operator == "pow_pos": + return intervalmath.pow_pos( + arguments[0], arguments[1], guard_bits=guard, cap_bits=cap + ) + if operator == "pow_nn": + return intervalmath.pow_nn( + arguments[0], arguments[1], guard_bits=guard, cap_bits=cap + ) + if operator == "ratio0": + return intervalmath.ratio0(arguments[0], arguments[1], cap_bits=cap) + raise SemanticFormulaError(f"foreign operator {operator}") diff --git a/proof/region/v1/semantic/verifier.py b/proof/region/v1/semantic/verifier.py new file mode 100644 index 00000000..b0abc30e --- /dev/null +++ b/proof/region/v1/semantic/verifier.py @@ -0,0 +1,222 @@ +"""Third verifier: independent semantic replay of one engine transcript. + +The verifier recomputes every decision from immutable job bytes with its own +strict SSA interpreter and rigorous interval arithmetic, then seals a receipt +or rejects with the first concrete ordinal where the transcript disagrees. +It never imports engine code and never trusts engine-internal enclosures. +""" + +from __future__ import annotations + +import region_proof_protocol as protocol + +from . import intervalmath, replay, region +from .receipt import ( + SemanticVerificationReasonV1, + SemanticVerificationRejectedV1, + SemanticVerificationReceiptV1, +) +from .ssa import SemanticFormulaError + +VerificationResultV1 = SemanticVerificationReceiptV1 | SemanticVerificationRejectedV1 + + +def _reject( + reason: SemanticVerificationReasonV1, + ordinal: int, + detail: str, +) -> SemanticVerificationRejectedV1: + return SemanticVerificationRejectedV1(reason, ordinal, detail) + + +def _foreign_binding(detail: str) -> SemanticVerificationRejectedV1: + return _reject(SemanticVerificationReasonV1.FOREIGN_BINDING, 0, detail) + + +def verify_transcript( + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, + transcript: protocol.DecisionTranscriptV1, + run: protocol.RunClaimV1, +) -> VerificationResultV1: + if ( + type(job) is not protocol.ProofJobV1 + or type(comparator) is not protocol.ContentResolvedComparatorManifestV2 + or type(transcript) is not protocol.DecisionTranscriptV1 + or type(run) is not protocol.RunClaimV1 + ): + return _reject( + SemanticVerificationReasonV1.INVALID_INPUT, + 0, + "semantic verification requires canonical V1 objects", + ) + + if transcript.job_identity != job.identity: + return _foreign_binding("transcript binds a foreign job") + if transcript.domain_identity != job.domain.identity: + return _foreign_binding("transcript binds a foreign domain") + if transcript.comparator_identity != comparator.identity: + return _foreign_binding("transcript binds a foreign comparator") + if run.job_identity != job.identity: + return _foreign_binding("run claim binds a foreign job") + if run.comparator_identity != comparator.identity: + return _foreign_binding("run claim binds a foreign comparator") + if run.transcript_identity != transcript.identity: + return _foreign_binding("run claim binds a foreign transcript") + if transcript.point_count != job.domain.point_count: + return _foreign_binding("transcript point count drifts from the bound domain") + # Binary, invocation and platform are declared execution coordinates. + # Their causality belongs to the source-bound controller's receipt; the + # semantic verifier binds the run through job, comparator and transcript + # only and never re-declares an executable anchor it cannot observe. + + try: + driver = replay.SemanticReplay(job, comparator) + accounting = replay.accounting_prefix_v1( + comparator.manifest.kind, + job, + comparator.identity, + ) + except ( + SemanticFormulaError, + KeyError, + StopIteration, + IndexError, + protocol.ProtocolErrorV1, + replay.ReplayIntegrityError, + ) as error: + return _reject( + SemanticVerificationReasonV1.REPLAY_UNRESOLVED, + 0, + f"replay cannot be initialised: {error}", + ) + + decisions = transcript.iter_decisions() + witnesses = transcript.iter_witnesses() + next_witness = next(witnesses, None) + + try: + for expected_index in range(transcript.point_count): + point = driver.next_point() + ordinal = point.ordinal + decision = next(decisions) + if int(decision) != point.outcome: + return _reject( + SemanticVerificationReasonV1.DECISION_MISMATCH, + ordinal, + ( + f"transcript records {int(decision)}, " + f"semantic replay decides {point.outcome}" + ), + ) + + expects_exact = point.outcome == region.INSIDE and point.exact_boundary + expects_boundary = point.outcome == region.BOUNDARY_UNPROVEN + expects_resource = point.outcome == region.RESOURCE_LIMIT_REACHED + + if next_witness is not None and next_witness.ordinal == ordinal: + witness = next_witness + next_witness = next(witnesses, None) + if expects_exact: + if type(witness) is not protocol.ExactZeroSignalTraceV1: + return _reject( + SemanticVerificationReasonV1.WITNESS_CONTRADICTION, + ordinal, + "exact boundary requires an exact-zero trace witness", + ) + expected_digest = replay.exact_trace_digest_v1( + job.identity, + ordinal, + point.exact_branch, + ) + if witness.trace_digest != expected_digest: + return _reject( + SemanticVerificationReasonV1.WITNESS_REPLAY_MISMATCH, + ordinal, + "exact-zero trace digest does not replay", + ) + elif expects_boundary: + if type(witness) is not protocol.BoundaryUnprovenWitnessV1: + return _reject( + SemanticVerificationReasonV1.WITNESS_CONTRADICTION, + ordinal, + "boundary outcome requires a boundary enclosure witness", + ) + elif expects_resource: + if type(witness) is not protocol.ResourceLimitWitnessV1: + return _reject( + SemanticVerificationReasonV1.WITNESS_CONTRADICTION, + ordinal, + "resource outcome requires a resource witness", + ) + if witness.scope != point.resource_scope: + return _reject( + SemanticVerificationReasonV1.RESOURCE_REPLAY_MISMATCH, + ordinal, + "resource scope does not replay", + ) + if witness.granted != point.point_grant: + return _reject( + SemanticVerificationReasonV1.RESOURCE_REPLAY_MISMATCH, + ordinal, + "resource grant does not replay", + ) + if witness.consumed != point.consumed: + return _reject( + SemanticVerificationReasonV1.RESOURCE_REPLAY_MISMATCH, + ordinal, + "resource consumption does not replay", + ) + else: + return _reject( + SemanticVerificationReasonV1.WITNESS_CONTRADICTION, + ordinal, + "decisive outcome carries no witness", + ) + elif expects_exact or expects_boundary or expects_resource: + return _reject( + SemanticVerificationReasonV1.WITNESS_REPLAY_MISMATCH, + ordinal, + "replay expects a witness the transcript does not carry", + ) + + accounting.update( + replay.account_record( + ordinal, + point.final_precision, + point.consumed, + point.outcome, + ) + ) + except intervalmath.UnresolvedError as error: + return _reject( + SemanticVerificationReasonV1.REPLAY_UNRESOLVED, + 0, + f"semantic replay needs more guard precision: {error}", + ) + except protocol.ProtocolErrorV1 as error: + return _reject( + SemanticVerificationReasonV1.REPLAY_UNRESOLVED, + 0, + f"semantic replay met an invalid definition: {error}", + ) + except replay.ReplayIntegrityError as error: + return _reject( + SemanticVerificationReasonV1.REPLAY_UNRESOLVED, + error.ordinal, + error.detail, + ) + + if next_witness is not None: + return _reject( + SemanticVerificationReasonV1.WITNESS_CONTRADICTION, + next_witness.ordinal, + "transcript carries witnesses beyond the replayed domain", + ) + if accounting.digest() != transcript.accounting_digest: + return _reject( + SemanticVerificationReasonV1.ACCOUNTING_REPLAY_MISMATCH, + 0, + "accounting digest does not replay from decisions and grants", + ) + return SemanticVerificationReceiptV1._seal(job, comparator, run, transcript) diff --git a/proof/region/v1/tests/test_semantic_diversity.py b/proof/region/v1/tests/test_semantic_diversity.py new file mode 100644 index 00000000..525614b2 --- /dev/null +++ b/proof/region/v1/tests/test_semantic_diversity.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Hostile contract for third-verifier diversity and admission shapes.""" + +from __future__ import annotations + +import ast +import hashlib +import re +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from semantic.receipt import ( # noqa: E402 + SemanticVerificationReceiptV1, +) + + +SEMANTIC_SOURCES = tuple(sorted((ROOT / "semantic").rglob("*.py"))) +FORBIDDEN_MODULE_ROOTS = ("arb", "mpfi", "build", "executor", "provenance") +FORBIDDEN_PATH_HINTS = ( + re.compile(r"\barb[/\\]\w"), + re.compile(r"\bmpfi[/\\]\w"), + re.compile(r"\bevaluator\b"), +) + + +def _imported_roots(tree: ast.AST) -> set[str]: + roots: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + roots.update(alias.name.split(".")) + elif isinstance(node, ast.ImportFrom): + # Every component of a dotted module path can hide a forbidden + # root, and `from ... import executor` binds the module through + # its alias name — including bare relative imports with no module. + if node.module: + roots.update(node.module.split(".")) + roots.update(alias.name.split(".")[0] for alias in node.names) + return roots + + +def _boundary_violations(source: str, name: str) -> list[str]: + violations: list[str] = [] + roots = _imported_roots(ast.parse(source)) + for root in FORBIDDEN_MODULE_ROOTS: + if root in roots: + violations.append(f"{name} imports forbidden module root {root}") + if "__import__" in source: + violations.append(f"{name} hides dynamic imports") + if "importlib" in roots or "import_module" in source: + violations.append(f"{name} reaches for dynamic import machinery") + return violations + + +class DiversityBoundaryTests(unittest.TestCase): + def test_semantic_sources_exist(self) -> None: + names = {path.name for path in SEMANTIC_SOURCES} + self.assertIn("__init__.py", names) + self.assertIn("receipt.py", names) + self.assertIn("verifier.py", names) + + def test_semantic_package_imports_neither_evaluator_path(self) -> None: + # The third verifier may inherit only the canonical protocol and the + # standard library; any Arb/MPFI/pipeline import destroys diversity. + for path in SEMANTIC_SOURCES: + source = path.read_text(encoding="utf-8") + self.assertEqual(_boundary_violations(source, path.name), []) + if path.name in ("__init__.py", "intervalmath.py"): + # The facade only re-exports the verifier boundary; the + # interval kernel is pure mathematics with no wire surface. + continue + roots = _imported_roots(ast.parse(source)) + self.assertIn("region_proof_protocol", roots, f"{path.name} lost the protocol binding") + + def test_semantic_package_never_reads_evaluator_artifacts(self) -> None: + for path in SEMANTIC_SOURCES: + source = path.read_text(encoding="utf-8") + for pattern in FORBIDDEN_PATH_HINTS: + self.assertIsNone( + pattern.search(source), + f"{path.name} references evaluator artifacts", + ) + + +class SyntheticBoundaryTests(unittest.TestCase): + """The checker must bite on hidden evaluator imports, not only obvious ones.""" + + HOSTILE_SOURCES = ( + "import arb.evaluator", + "from mpfi import receipt", + "from .. import executor", + "from proof.region.v1 import executor", + "import provenance as alias", + "import proof.region.v1.executor", + "import importlib", + "module = __import__('build')", + "module = importlib.import_module('arb')", + ) + + def test_hidden_evaluator_imports_are_detected(self) -> None: + for source in self.HOSTILE_SOURCES: + self.assertTrue( + _boundary_violations(source, "synthetic.py"), + f"boundary checker missed {source!r}", + ) + + def test_canonical_imports_pass_the_checker(self) -> None: + source = "import region_proof_protocol\nfrom . import intervalmath\n" + self.assertEqual(_boundary_violations(source, "synthetic.py"), []) + + +def digest(label: int) -> bytes: + return hashlib.sha256(f"semantic-diversity-{label}".encode("ascii")).digest() + + +class DegenerateVerifierShapeTests(unittest.TestCase): + """Hash-compare, no-op and saturate-all shapes must not mint receipts.""" + + def _hash_comparer_receipt(self) -> SemanticVerificationReceiptV1: + return SemanticVerificationReceiptV1( + digest(1), digest(2), digest(3), digest(4), digest(5) + ) + + def _no_op_receipt(self) -> SemanticVerificationReceiptV1: + return SemanticVerificationReceiptV1( + digest(6), digest(7), digest(8), digest(9), digest(10) + ) + + def _saturate_all_receipt(self) -> SemanticVerificationReceiptV1: + return SemanticVerificationReceiptV1( + digest(11), digest(12), digest(13), digest(14), digest(15) + ) + + def test_degenerate_shapes_cannot_create_receipts(self) -> None: + for shape in ( + self._hash_comparer_receipt, + self._no_op_receipt, + self._saturate_all_receipt, + ): + with self.assertRaises(TypeError): + shape() + + def test_foreign_token_cannot_open_the_seal(self) -> None: + with self.assertRaises(TypeError): + SemanticVerificationReceiptV1( + digest(1), digest(2), digest(3), digest(4), digest(5), + _token=object(), + ) + with self.assertRaises(TypeError): + SemanticVerificationReceiptV1( + digest(1), digest(2), digest(3), digest(4), digest(5), + _token="verifier", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/proof/region/v1/tests/test_semantic_intervalmath.py b/proof/region/v1/tests/test_semantic_intervalmath.py new file mode 100644 index 00000000..879cb33e --- /dev/null +++ b/proof/region/v1/tests/test_semantic_intervalmath.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Operator contract for the semantic verifier's rigorous interval math. + +Every enclosure must contain the mathematical truth; the checks below pin +each transcendental inside a narrow rational window, so an enclosure that +misses the truth (or a remainder bound that stops too early) fails without +needing any external numeric library. +""" + +from __future__ import annotations + +import sys +import unittest +from fractions import Fraction +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from semantic import intervalmath # noqa: E402 + +GUARD = 64 +CAP = 64 +SCALE = 10**20 + + +def call(name: str, *arguments) -> intervalmath.Interval: + return getattr(intervalmath, name)( + *arguments, guard_bits=GUARD, cap_bits=CAP + ) + + +class EnclosureSoundnessTests(unittest.TestCase): + def assert_encloses_truth( + self, + interval: intervalmath.Interval, + truncated_20dp: int, + *, + negative: bool = False, + ) -> None: + # The window is the first 20 decimals of the mathematical truth, so + # the truth lies strictly inside it. A correct guard-64 enclosure is + # at most a few ulp of 2^-64 wide, so it must cover the whole window; + # a missed truth or an unsound remainder bound leaves an endpoint + # inside the window and fails. + lo = Fraction(truncated_20dp, SCALE) + hi = Fraction(truncated_20dp + 1, SCALE) + if negative: + lo, hi = -hi, -lo + self.assertLess(interval.lo, lo) + self.assertGreater(interval.hi, hi) + self.assertLess(interval.hi - interval.lo, Fraction(1, 1 << 48)) + + def test_exp_one_contains_e(self) -> None: + # e = 2.71828182845904523536... + self.assert_encloses_truth( + call("exp", intervalmath.exact(Fraction(1))), + 271828182845904523536, + ) + + def test_exp_negative_one(self) -> None: + # exp(-1) = 0.36787944117144232159... + self.assert_encloses_truth( + call("exp", intervalmath.exact(Fraction(-1))), + 36787944117144232159, + ) + + def test_log_two(self) -> None: + # ln 2 = 0.69314718055994530941... + self.assert_encloses_truth( + call("log", intervalmath.exact(Fraction(2))), + 69314718055994530941, + ) + + def test_sin_one(self) -> None: + # sin 1 = 0.84147098480789650665... + self.assert_encloses_truth( + call("sin", intervalmath.exact(Fraction(1))), + 84147098480789650665, + ) + + def test_cos_one(self) -> None: + # cos 1 = 0.54030230586813971740... + self.assert_encloses_truth( + call("cos", intervalmath.exact(Fraction(1))), + 54030230586813971740, + ) + + def test_sin_big_argument_reduces(self) -> None: + # sin(1000/7) = -0.99636221069974350838...: exercises quadrant + # reduction against the rational pi enclosure with a large argument. + self.assert_encloses_truth( + call("sin", intervalmath.exact(Fraction(1000, 7))), + 99636221069974350838, + negative=True, + ) + + def test_sqrt_two(self) -> None: + # sqrt 2 = 1.41421356237309504880... + self.assert_encloses_truth( + call("sqrt", intervalmath.exact(Fraction(2))), + 141421356237309504880, + ) + + def test_root3_five(self) -> None: + # 5^(1/3) = 1.70997594667669698935... + self.assert_encloses_truth( + call("root3", intervalmath.exact(Fraction(5))), + 170997594667669698935, + ) + + def test_pow_pos(self) -> None: + # (3/2)^(5/4) = 1.66002287955048238861... + self.assert_encloses_truth( + call( + "pow_pos", + intervalmath.exact(Fraction(3, 2)), + intervalmath.exact(Fraction(5, 4)), + ), + 166002287955048238861, + ) + + +class OperatorDomainContractTests(unittest.TestCase): + def test_pow_nn_zero_base_needs_strictly_positive_exponent(self) -> None: + zero = intervalmath.exact(0) + crossing = intervalmath.Interval(Fraction(-1), Fraction(2)) + with self.assertRaises(intervalmath.UnresolvedError): + call("pow_nn", zero, crossing) + result = call("pow_nn", zero, intervalmath.exact(Fraction(2))) + self.assertTrue(result.is_exact) + self.assertEqual(result.lo, 0) + + def test_root3_rejects_negative_arguments(self) -> None: + with self.assertRaises(intervalmath.UnresolvedError): + call("root3", intervalmath.exact(Fraction(-8))) + with self.assertRaises(intervalmath.UnresolvedError): + call("root3", intervalmath.Interval(Fraction(-1), Fraction(1))) + + def test_log_rejects_nonpositive_arguments(self) -> None: + with self.assertRaises(intervalmath.UnresolvedError): + call("log", intervalmath.exact(Fraction(0))) + with self.assertRaises(intervalmath.UnresolvedError): + call("log", intervalmath.Interval(Fraction(-1), Fraction(1))) + + def test_sqrt_rejects_negative_arguments(self) -> None: + with self.assertRaises(intervalmath.UnresolvedError): + call("sqrt", intervalmath.exact(Fraction(-1))) + + def test_exp_rejects_arguments_beyond_reduction_range(self) -> None: + huge = intervalmath.exact(Fraction(10) ** 400) + with self.assertRaises(intervalmath.UnresolvedError): + call("exp", huge) + + def test_sin_cos_reject_arguments_beyond_reduction_range(self) -> None: + # A hostile binary64 argument must stay unresolved instead of blowing + # up the quadrant sweep in proportion to its magnitude. + huge = intervalmath.exact(Fraction(10) ** 400) + with self.assertRaises(intervalmath.UnresolvedError): + call("sin", huge) + with self.assertRaises(intervalmath.UnresolvedError): + call("cos", huge) + negative_huge = intervalmath.exact(Fraction(-(10) ** 400)) + with self.assertRaises(intervalmath.UnresolvedError): + call("sin", negative_huge) + + def test_div_rejects_zero_divisor(self) -> None: + with self.assertRaises(intervalmath.UnresolvedError): + intervalmath.div( + intervalmath.exact(Fraction(1)), + intervalmath.exact(Fraction(0)), + cap_bits=CAP, + ) + + def test_sign_needs_a_strict_sign(self) -> None: + with self.assertRaises(intervalmath.UnresolvedError): + intervalmath.sign(intervalmath.Interval(Fraction(-1), Fraction(1))) + positive = intervalmath.sign(intervalmath.exact(Fraction(3))) + self.assertTrue(positive.is_exact) + self.assertEqual(positive.lo, 1) + + def test_ratio0_zero_over_zero_is_exact_zero(self) -> None: + result = intervalmath.ratio0( + intervalmath.exact(0), intervalmath.exact(0), cap_bits=CAP + ) + self.assertTrue(result.is_exact) + self.assertEqual(result.lo, 0) + + def test_ratio0_needs_a_strictly_positive_divisor(self) -> None: + with self.assertRaises(intervalmath.UnresolvedError): + intervalmath.ratio0( + intervalmath.exact(1), intervalmath.exact(0), cap_bits=CAP + ) + with self.assertRaises(intervalmath.UnresolvedError): + intervalmath.ratio0( + intervalmath.exact(1), + intervalmath.Interval(Fraction(-1), Fraction(1)), + cap_bits=CAP, + ) + half = intervalmath.ratio0( + intervalmath.exact(1), intervalmath.exact(2), cap_bits=CAP + ) + # The dyadic grid keeps the exact quotient unrounded. + self.assertEqual(half.lo, Fraction(1, 2)) + self.assertEqual(half.hi, Fraction(1, 2)) + + +if __name__ == "__main__": + unittest.main() diff --git a/proof/region/v1/tests/test_semantic_receipt.py b/proof/region/v1/tests/test_semantic_receipt.py new file mode 100644 index 00000000..561e4bce --- /dev/null +++ b/proof/region/v1/tests/test_semantic_receipt.py @@ -0,0 +1,238 @@ +#!/usr/bin/env python3 +"""Hostile contract for the sealed semantic verification receipt V1.""" + +from __future__ import annotations + +import hashlib +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +import region_proof_protocol as protocol # noqa: E402 + +from region_proof_protocol import ( # noqa: E402 + ComparatorKindV1, + ComparatorManifestV2, + ContentResolvedComparatorManifestV2, + DecisionTranscriptV1, + DecisionV1, + ProofJobV1, + RunClaimV1, +) + +from semantic import receipt as semantic_receipt # noqa: E402 +from semantic.receipt import ( # noqa: E402 + SemanticVerificationReceiptV1, + SemanticVerificationReasonV1, + SemanticVerificationRejectedV1, + resolved_decision_digest_v1, +) + + +FIXTURES = ROOT / "fixtures" + + +def digest(label: int) -> bytes: + return hashlib.sha256(f"semantic-receipt-test-{label}".encode("ascii")).digest() + + +SYNTHETIC_CONTENT = { + digest(index): f"semantic-receipt-test-{index}".encode("ascii") + for index in range(1_000) +} + + +def admit_manifest(kind: ComparatorKindV1, seed: int) -> ContentResolvedComparatorManifestV2: + return ContentResolvedComparatorManifestV2.admit( + ComparatorManifestV2( + kind=kind, + engine_release=digest(seed), + upstream_source=digest(seed + 1), + arithmetic_input_set=digest(seed + 2), + wrapper_source=digest(seed + 3), + evaluator_source=digest(seed + 4), + build_identity=digest(seed + 5), + operation_allowlist=digest(seed + 6), + test_observation=digest(seed + 7), + legal_file_set=digest(seed + 8), + exclusions=digest(seed + 9), + ), + SYNTHETIC_CONTENT.get, + ) + + +def fixture_job() -> ProofJobV1: + return ProofJobV1.parse((FIXTURES / "proof-job-v1.bin").read_bytes()) + + +def outside_transcript( + job: ProofJobV1, + comparator: ContentResolvedComparatorManifestV2, +) -> DecisionTranscriptV1: + return DecisionTranscriptV1.from_decisions( + job, + comparator, + (DecisionV1.OUTSIDE for _ in range(job.domain.point_count)), + (), + digest(900), + ) + + +def run_claim( + job: ProofJobV1, + comparator: ContentResolvedComparatorManifestV2, + transcript: DecisionTranscriptV1, +) -> RunClaimV1: + return RunClaimV1.for_transcript( + job, + comparator, + transcript, + digest(901), + digest(902), + digest(903), + ) + + +class ReceiptSealingTests(unittest.TestCase): + def test_direct_construction_is_sealed(self) -> None: + with self.assertRaises(TypeError): + SemanticVerificationReceiptV1( + digest(1), digest(2), digest(3), digest(4), digest(5) + ) + with self.assertRaises(TypeError): + SemanticVerificationReceiptV1( + digest(1), + digest(2), + digest(3), + digest(4), + digest(5), + _token=object(), + ) + + def test_seal_replays_coordinates_and_binds(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 100) + transcript = outside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + + receipt = SemanticVerificationReceiptV1._seal(job, comparator, run, transcript) + self.assertEqual(receipt.job_identity, job.identity) + self.assertEqual(receipt.comparator_identity, comparator.identity) + self.assertEqual(receipt.run_claim_identity, run.identity) + self.assertEqual(receipt.transcript_identity, transcript.identity) + self.assertEqual( + receipt.decision_digest, + resolved_decision_digest_v1( + transcript.domain_identity, transcript.decision_bits + ), + ) + self.assertTrue(receipt.binds(job, comparator, run, transcript)) + self.assertEqual(len(receipt.encode()), 160) + self.assertEqual(len(receipt.identity), 32) + + other_comparator = admit_manifest(ComparatorKindV1.ARB, 200) + other_transcript = outside_transcript(job, other_comparator) + other_run = run_claim(job, other_comparator, other_transcript) + self.assertFalse(receipt.binds(job, other_comparator, other_run, other_transcript)) + self.assertFalse(receipt.binds(None, comparator, run, transcript)) + + def test_receipt_is_immutable(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.MPFI, 300) + transcript = outside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + receipt = SemanticVerificationReceiptV1._seal(job, comparator, run, transcript) + with self.assertRaises(AttributeError): + receipt.job_identity = digest(6) + + def test_seal_rejects_noncanonical_digests(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 400) + transcript = outside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + receipt = SemanticVerificationReceiptV1._seal(job, comparator, run, transcript) + token = semantic_receipt._RECEIPT_TOKEN + with self.assertRaises(TypeError): + SemanticVerificationReceiptV1( + bytes(32), + receipt.comparator_identity, + receipt.run_claim_identity, + receipt.transcript_identity, + receipt.decision_digest, + _token=token, + ) + with self.assertRaises(TypeError): + SemanticVerificationReceiptV1( + receipt.job_identity, + receipt.comparator_identity, + receipt.run_claim_identity, + receipt.transcript_identity, + b"short", + _token=token, + ) + + +class RejectionShapeTests(unittest.TestCase): + def test_rejection_is_typed_and_validated(self) -> None: + rejection = SemanticVerificationRejectedV1( + SemanticVerificationReasonV1.DECISION_MISMATCH, 7, "sign disagrees" + ) + self.assertEqual(rejection.reason, SemanticVerificationReasonV1.DECISION_MISMATCH) + self.assertEqual(rejection.ordinal, 7) + + with self.assertRaises(TypeError): + SemanticVerificationRejectedV1("decision_mismatch", 0, "foreign reason") + with self.assertRaises(TypeError): + SemanticVerificationRejectedV1( + SemanticVerificationReasonV1.DECISION_MISMATCH, -1, "ordinal" + ) + with self.assertRaises(TypeError): + SemanticVerificationRejectedV1( + SemanticVerificationReasonV1.DECISION_MISMATCH, + protocol.OUTPUT_CARDINALITY_V1, + "ordinal", + ) + with self.assertRaises(TypeError): + SemanticVerificationRejectedV1( + SemanticVerificationReasonV1.DECISION_MISMATCH, 0, "" + ) + + def test_rejection_reasons_are_a_closed_sum(self) -> None: + self.assertEqual( + sorted(reason.value for reason in SemanticVerificationReasonV1), + [ + "accounting_replay_mismatch", + "decision_mismatch", + "foreign_binding", + "invalid_input", + "replay_unresolved", + "resource_replay_mismatch", + "witness_contradiction", + "witness_replay_mismatch", + ], + ) + + def test_noncanonical_inputs_are_typed_invalid_input(self) -> None: + from semantic.verifier import verify_transcript + + # The verifier rejects foreign object types through the same typed + # surface as every other failure; it never raises into the caller. + result = verify_transcript(None, None, None, None) + self.assertIsInstance(result, SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.INVALID_INPUT) + self.assertEqual(result.ordinal, 0) + + def test_receipt_type_is_final(self) -> None: + # A subclass would inherit the sealed constructor; the type refuses + # derivation so no foreign code can mint receipts through a subtype. + with self.assertRaises(TypeError): + + class Forgery(SemanticVerificationReceiptV1): + pass + + +if __name__ == "__main__": + unittest.main() diff --git a/proof/region/v1/tests/test_semantic_replay.py b/proof/region/v1/tests/test_semantic_replay.py new file mode 100644 index 00000000..aa78ce04 --- /dev/null +++ b/proof/region/v1/tests/test_semantic_replay.py @@ -0,0 +1,394 @@ +#!/usr/bin/env python3 +"""Hostile contract for independent semantic replay of engine transcripts.""" + +from __future__ import annotations + +import hashlib +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from region_proof_protocol import ( # noqa: E402 + BoundaryUnprovenWitnessV1, + ComparatorKindV1, + ComparatorManifestV2, + ContentResolvedComparatorManifestV2, + DecisionTranscriptV1, + DecisionV1, + ExactZeroSignalTraceV1, + ProofJobV1, + ReducedDomainManifestV1, + ResourceLimitWitnessV1, + RunClaimV1, + WitnessStoreV1, + WitnessV1, + compare_dual_transcripts, +) + +from semantic import replay as semantic_replay # noqa: E402 +from semantic.receipt import ( # noqa: E402 + SemanticVerificationReceiptV1, + SemanticVerificationReasonV1, + SemanticVerificationRejectedV1, + resolved_decision_digest_v1, +) +from semantic.verifier import verify_transcript # noqa: E402 + + +FIXTURES = ROOT / "fixtures" + + +def digest(label: int) -> bytes: + return hashlib.sha256(f"semantic-replay-test-{label}".encode("ascii")).digest() + + +SYNTHETIC_CONTENT = { + digest(index): f"semantic-replay-test-{index}".encode("ascii") + for index in range(1_000) +} + + +def admit_manifest(kind: ComparatorKindV1, seed: int) -> ContentResolvedComparatorManifestV2: + return ContentResolvedComparatorManifestV2.admit( + ComparatorManifestV2( + kind=kind, + engine_release=digest(seed), + upstream_source=digest(seed + 1), + arithmetic_input_set=digest(seed + 2), + wrapper_source=digest(seed + 3), + evaluator_source=digest(seed + 4), + build_identity=digest(seed + 5), + operation_allowlist=digest(seed + 6), + test_observation=digest(seed + 7), + legal_file_set=digest(seed + 8), + exclusions=digest(seed + 9), + ), + SYNTHETIC_CONTENT.get, + ) + + +def fixture_job() -> ProofJobV1: + return ProofJobV1.parse((FIXTURES / "proof-job-v1.bin").read_bytes()) + + +def run_claim( + job: ProofJobV1, + comparator: ContentResolvedComparatorManifestV2, + transcript: DecisionTranscriptV1, +) -> RunClaimV1: + return RunClaimV1.for_transcript( + job, + comparator, + transcript, + digest(801), + digest(802), + digest(803), + ) + + +def all_outside_transcript( + job: ProofJobV1, + comparator: ContentResolvedComparatorManifestV2, +) -> DecisionTranscriptV1: + return DecisionTranscriptV1.from_decisions( + job, + comparator, + (DecisionV1.OUTSIDE for _ in range(job.domain.point_count)), + (), + digest(810), + ) + + +def all_inside_transcript( + job: ProofJobV1, + comparator: ContentResolvedComparatorManifestV2, +) -> DecisionTranscriptV1: + ordinals = tuple(job.domain.iter_ordinals()) + witnesses = tuple( + ExactZeroSignalTraceV1(ordinal, digest(10_000 + position)) + for position, ordinal in enumerate(ordinals) + ) + return DecisionTranscriptV1.from_decisions( + job, + comparator, + (DecisionV1.INSIDE for _ in ordinals), + witnesses, + digest(811), + ) + + +class HostileReplayTests(unittest.TestCase): + def test_two_identical_wrong_transcripts_both_fail_replay(self) -> None: + # Independence means replay recomputes from job bytes: one wrong + # transcript fails, and an identical copy fails exactly the same way. + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 500) + transcript = all_outside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + + first = verify_transcript(job, comparator, transcript, run) + second = verify_transcript(job, comparator, transcript, run) + + self.assertIsInstance(first, SemanticVerificationRejectedV1) + self.assertIsInstance(second, SemanticVerificationRejectedV1) + self.assertEqual(first.reason, second.reason) + self.assertEqual(first.ordinal, second.ordinal) + # A wrong transcript must fail on replayed semantics, never + # masquerade as a binding failure. + self.assertNotEqual(first.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + self.assertNotIsInstance(first, SemanticVerificationReceiptV1) + + def test_saturate_all_inside_transcript_fails_replay(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.MPFI, 600) + transcript = all_inside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + + result = verify_transcript(job, comparator, transcript, run) + self.assertIsInstance(result, SemanticVerificationRejectedV1) + # Saturating every point to INSIDE fails at the first domain point, + # and it must fail as a decision mismatch, not a vacuous rejection. + self.assertEqual(result.reason, SemanticVerificationReasonV1.DECISION_MISMATCH) + self.assertEqual(result.ordinal, tuple(job.domain.iter_ordinals())[0]) + + def test_foreign_comparator_binding_is_rejected_before_replay(self) -> None: + job = fixture_job() + bound = admit_manifest(ComparatorKindV1.ARB, 700) + foreign = admit_manifest(ComparatorKindV1.ARB, 750) + transcript = all_outside_transcript(job, bound) + run = run_claim(job, bound, transcript) + + result = verify_transcript(job, foreign, transcript, run) + self.assertIsInstance(result, SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + + def test_foreign_run_binding_is_rejected_before_replay(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 760) + transcript = all_outside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + # The verifier can only attest the run that binds the transcript it + # replays. Binary, invocation and platform causality belongs to the + # source-bound controller, so a run becomes foreign to this + # verification exactly when it points at a different transcript. + foreign_run = RunClaimV1( + job.identity, + comparator.identity, + digest(821), + digest(822), + digest(823), + digest(829), + ) + self.assertNotEqual(foreign_run.identity, run.identity) + + result = verify_transcript(job, comparator, transcript, foreign_run) + self.assertIsInstance(result, SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + + def test_foreign_job_binding_is_rejected_before_replay(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.MPFI, 770) + transcript = all_outside_transcript(job, comparator) + run = run_claim(job, comparator, transcript) + foreign_domain = ReducedDomainManifestV1.from_ordinals((0,)) + foreign_job = ProofJobV1( + job.definition, + job.formula_spec, + foreign_domain, + job.policy, + ) + self.assertNotEqual(foreign_job.identity, job.identity) + + result = verify_transcript(foreign_job, comparator, transcript, run) + self.assertIsInstance(result, SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + + def test_point_count_drift_is_rejected_before_replay(self) -> None: + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 780) + # A transcript may be internally consistent yet claim more points + # than the bound domain; the verifier must reject it as a foreign + # binding instead of walking past the end of the domain ordinals. + drifted_count = job.domain.point_count + 4 + transcript = DecisionTranscriptV1( + job.identity, + job.domain.identity, + comparator.identity, + drifted_count, + bytes([0x55] * ((drifted_count + 3) // 4)), + (0, drifted_count, 0, 0), + 0, + digest(812), + WitnessStoreV1.from_witnesses(()), + ) + run = run_claim(job, comparator, transcript) + + result = verify_transcript(job, comparator, transcript, run) + self.assertIsInstance(result, SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + + def test_replayed_transcript_seals_a_receipt(self) -> None: + # Anti-vacuum: the verifier must mint a receipt for a transcript that + # exactly matches its own independent replay, not only reject. + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 900) + driver = semantic_replay.SemanticReplay(job, comparator) + decisions: list[DecisionV1] = [] + witnesses: list[WitnessV1] = [] + accounting = semantic_replay.accounting_prefix_v1( + comparator.manifest.kind, + job, + comparator.identity, + ) + for _ in range(job.domain.point_count): + point = driver.next_point() + decisions.append(DecisionV1(point.outcome)) + if point.outcome == DecisionV1.INSIDE and point.exact_boundary: + witnesses.append( + ExactZeroSignalTraceV1( + point.ordinal, + semantic_replay.exact_trace_digest_v1( + job.identity, + point.ordinal, + point.exact_branch, + ), + ) + ) + elif point.outcome == DecisionV1.BOUNDARY_UNPROVEN: + witnesses.append( + BoundaryUnprovenWitnessV1( + point.ordinal, + digest(100_000 + point.ordinal), + ) + ) + elif point.outcome == DecisionV1.RESOURCE_LIMIT_REACHED: + witnesses.append( + ResourceLimitWitnessV1( + point.ordinal, + point.resource_scope, + point.point_grant, + point.consumed, + ) + ) + accounting.update( + semantic_replay.account_record( + point.ordinal, + point.final_precision, + point.consumed, + point.outcome, + ) + ) + transcript = DecisionTranscriptV1.from_decisions( + job, + comparator, + decisions, + witnesses, + accounting.digest(), + ) + run = run_claim(job, comparator, transcript) + + result = verify_transcript(job, comparator, transcript, run) + self.assertIsInstance(result, SemanticVerificationReceiptV1) + self.assertEqual(result.run_claim_identity, run.identity) + self.assertEqual(result.transcript_identity, transcript.identity) + self.assertTrue(result.binds(job, comparator, run, transcript)) + + # Visible coverage: the transcript carries exactly the outcomes the + # replay produced, with one aligned witness per witness-requiring + # outcome and none for decisive ones. + self.assertEqual(len(decisions), job.domain.point_count) + occurred = {decision.value for decision in decisions} + self.assertTrue(occurred <= {member.value for member in DecisionV1}) + boundary_count = decisions.count(DecisionV1.BOUNDARY_UNPROVEN) + boundary_witnesses = tuple( + witness for witness in witnesses if type(witness) is BoundaryUnprovenWitnessV1 + ) + self.assertEqual(len(boundary_witnesses), boundary_count) + self.assertEqual( + len({witness.ordinal for witness in witnesses}), + len(witnesses), + "witness ordinals must stay unique and point-aligned", + ) + + def test_decision_digest_matches_dual_admission_grammar(self) -> None: + # The semantic receipt's resolved-decision digest must agree with the + # digest that dual comparison computes for identical decision bits; + # any grammar drift would fork the two proof surfaces. + job = fixture_job() + arb = admit_manifest(ComparatorKindV1.ARB, 920) + mpfi = admit_manifest(ComparatorKindV1.MPFI, 940) + first = all_outside_transcript(job, arb) + second = all_outside_transcript(job, mpfi) + first_run = RunClaimV1.for_transcript( + job, arb, first, digest(831), digest(832), digest(833) + ) + second_run = RunClaimV1.for_transcript( + job, mpfi, second, digest(834), digest(835), digest(836) + ) + + candidate = compare_dual_transcripts( + job, arb, first, first_run, mpfi, second, second_run + ) + self.assertEqual( + candidate.claim.decision_digest, + resolved_decision_digest_v1(first.domain_identity, first.decision_bits), + ) + + def test_accounting_digest_grammar_matches_independent_packing(self) -> None: + # The accounting grammar is re-packed by hand from the declared wire + # layout; the replay helpers must reproduce exactly that digest. + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.ARB, 910) + records = ((5, 8, 3, 2), (9, 16, 0, 0), (70_000, 8, 12, 1)) + hasher = hashlib.sha256() + hasher.update(b"labcolors.arb-evaluation-accounting.v1\0") + hasher.update(job.identity) + hasher.update(job.domain.identity) + hasher.update(job.policy.identity) + hasher.update(comparator.identity) + for ordinal, precision, consumed, outcome in records: + hasher.update(ordinal.to_bytes(4, "big")) + hasher.update(precision.to_bytes(4, "big")) + hasher.update(consumed.to_bytes(8, "big")) + hasher.update(bytes((outcome,))) + expected = hasher.digest() + + accounting = semantic_replay.accounting_prefix_v1( + comparator.manifest.kind, job, comparator.identity + ) + for record in records: + accounting.update(semantic_replay.account_record(*record)) + self.assertEqual(accounting.digest(), expected) + + def test_mpfi_accounting_digest_matches_independent_packing(self) -> None: + # The MPFI lane carries its own domain prefix; the expected digest is + # packed from scratch again so neither lane reuses the other's truth. + job = fixture_job() + comparator = admit_manifest(ComparatorKindV1.MPFI, 911) + records = ((3, 8, 1, 1), (12, 32, 0, 2), (400_000, 16, 7, 0)) + hasher = hashlib.sha256() + hasher.update(b"labcolors.mpfi-evaluation-accounting.v1\0") + hasher.update(job.identity) + hasher.update(job.domain.identity) + hasher.update(job.policy.identity) + hasher.update(comparator.identity) + for ordinal, precision, consumed, outcome in records: + hasher.update(ordinal.to_bytes(4, "big")) + hasher.update(precision.to_bytes(4, "big")) + hasher.update(consumed.to_bytes(8, "big")) + hasher.update(bytes((outcome,))) + expected = hasher.digest() + + accounting = semantic_replay.accounting_prefix_v1( + comparator.manifest.kind, job, comparator.identity + ) + for record in records: + accounting.update(semantic_replay.account_record(*record)) + self.assertEqual(accounting.digest(), expected) + + +if __name__ == "__main__": + unittest.main()