feat(ADR-0131.1.B): harden symbolic equivalence lane with generated corpus + exact algebra (#169)

* feat(evals): add deterministic symbolic equivalence generated corpus

* feat(evals): add symbolic equivalence replay helpers

* feat(evals): load generated symbolic equivalence corpus

* feat(evals): emit symbolic equivalence replay manifest

* feat(symbolic): support multivariable integer polynomials

* feat(symbolic): support exact rational polynomial coefficients

* feat(symbolic): align equivalence API with multivariable normalization

* test(ADR-0131.1.B): reconcile v1 expectations to v1.B scope expansion

The v1.B refactor (univariate int → sparse multivariable Fraction) deliberately
admits multivariable polynomials and constant-denominator division. The v1
dataset and tests pinned the old refusal behavior, so the lane runner reported
wrong=4 and 10 unit tests failed.

Reconcile:

- cases.jsonl: flip sym-eq-v1-0029 ('x+y' vs 'x+1') and sym-eq-v1-0030
  ('x/2' vs 'x') from expected=refused to expected=not_equivalent; rename
  categories to multivariable_distinct / constant_denominator_distinct;
  extend provenance with adr-0131.1b:scope-expanded.
- generated_cases.py: split _refusal_cases into scope_expanded (admits)
  and templates (still refused); the first two adversarial cases move to
  the scope-expanded list with expected=not_equivalent.
- test_math_symbolic_normalizer.py: replace test_undefined_variable and
  test_unknown_operator_division with positive scope-expansion tests +
  symbolic-denominator refusal; rewrite TestPolynomialInvariants for the
  new terms/variables constructor (Polynomial(terms={...}, variables=(...)))
  with float-rejection and zero-coef-collapse invariants.
- test_math_symbolic_equivalence.py: TestRefused.test_empty_left reason
  string matches new normalizer error; flip multivariable + constant-
  denominator cases to NOT_EQUIVALENT; add symbolic-denominator-refused
  case; relax canonical_a assertion in test_a_normalizes_b_refuses (engine
  now zeroes both on either-side refusal).
- report.json + manifest.json: regenerated; lane PASS 185/185 wrong=0.

Lane invariants reaffirmed by the new tests: wrong==0, refusal-first for
truly out-of-scope inputs (symbolic denominator, transcendental, malformed,
negative exponent), determinism via byte-equal report.
This commit is contained in:
Shay 2026-05-23 10:47:57 -07:00 committed by GitHub
parent 5b668cc866
commit 169cec710e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 1929 additions and 320 deletions

View file

@ -26,5 +26,5 @@
{"case_id":"sym-eq-v1-0026","expression_a":"2*(x + 3)","expression_b":"2*x + 3","expected":"not_equivalent","category":"distributive_miss","provenance":"adr-0131.1:hand-curated:2026-05-23"}
{"case_id":"sym-eq-v1-0027","expression_a":"(x + 1)*(x + 2)","expression_b":"x^2 + 3*x + 1","expected":"not_equivalent","category":"foil_miss","provenance":"adr-0131.1:hand-curated:2026-05-23"}
{"case_id":"sym-eq-v1-0028","expression_a":"x^3 + 1","expression_b":"(x + 1)^3","expected":"not_equivalent","category":"cube_miss","provenance":"adr-0131.1:hand-curated:2026-05-23"}
{"case_id":"sym-eq-v1-0029","expression_a":"x + y","expression_b":"x + 1","expected":"refused","category":"out_of_scope_variable","provenance":"adr-0131.1:hand-curated:2026-05-23"}
{"case_id":"sym-eq-v1-0030","expression_a":"x / 2","expression_b":"x","expected":"refused","category":"out_of_scope_division","provenance":"adr-0131.1:hand-curated:2026-05-23"}
{"case_id":"sym-eq-v1-0029","expression_a":"x + y","expression_b":"x + 1","expected":"not_equivalent","category":"multivariable_distinct","provenance":"adr-0131.1:hand-curated:2026-05-23;adr-0131.1b:scope-expanded:2026-05-23"}
{"case_id":"sym-eq-v1-0030","expression_a":"x / 2","expression_b":"x","expected":"not_equivalent","category":"constant_denominator_distinct","provenance":"adr-0131.1:hand-curated:2026-05-23;adr-0131.1b:scope-expanded:2026-05-23"}

View file

@ -0,0 +1,242 @@
"""ADR-0131.1.B — deterministic generated symbolic-equivalence cases.
This module expands the symbolic-equivalence lane without introducing
runtime randomness. Cases are generated from a pinned integer seed and a
closed set of polynomial/metamorphic transforms.
The generated corpus deliberately stays inside the ADR-0131.1 v1
normalizer scope (single variable, integer coefficients, polynomial
operators only). It is not a substitute for the later multi-variable /
rational / sealed-holdout expansion. Its purpose is to harden Benchmark
1 against a tiny hand-curated-only dataset.
"""
from __future__ import annotations
import random
from dataclasses import dataclass
from typing import Final, Iterable
SEED: Final[int] = 131101
GENERATED_CASE_COUNT: Final[int] = 120
VARIABLE: Final[str] = "x"
@dataclass(frozen=True, slots=True)
class GeneratedCase:
case_id: str
expression_a: str
expression_b: str
expected: str
category: str
provenance: str
def as_dict(self) -> dict[str, str]:
return {
"case_id": self.case_id,
"expression_a": self.expression_a,
"expression_b": self.expression_b,
"expected": self.expected,
"category": self.category,
"provenance": self.provenance,
}
def _coef(rng: random.Random, *, allow_zero: bool = False) -> int:
choices = list(range(-5, 6))
if not allow_zero:
choices.remove(0)
return rng.choice(choices)
def _linear(rng: random.Random) -> tuple[str, int, int]:
"""Return (expr, a, b) for a*x + b."""
a = _coef(rng)
b = _coef(rng, allow_zero=True)
parts: list[str] = []
if a == 1:
parts.append("x")
elif a == -1:
parts.append("-x")
else:
parts.append(f"{a}*x")
if b > 0:
parts.append(f"+{b}")
elif b < 0:
parts.append(str(b))
return "".join(parts), a, b
def _expanded_square(a: int, b: int) -> str:
# (a*x+b)^2 = a^2*x^2 + 2ab*x + b^2
return _poly_to_expr({2: a * a, 1: 2 * a * b, 0: b * b})
def _expanded_product(a: int, b: int, c: int, d: int) -> str:
# (a*x+b)(c*x+d) = ac*x^2 + (ad+bc)x + bd
return _poly_to_expr({2: a * c, 1: a * d + b * c, 0: b * d})
def _poly_to_expr(terms: dict[int, int]) -> str:
"""Serialize sparse exponent->coefficient map to parser-compatible expr.
This is intentionally not the same as the normalizer's canonical string;
it emits a readable expression for generated corpus cases.
"""
parts: list[str] = []
for exp in sorted(terms.keys(), reverse=True):
coef = terms[exp]
if coef == 0:
continue
sign = "+" if coef > 0 else "-"
abs_coef = abs(coef)
if exp == 0:
term = str(abs_coef)
elif exp == 1:
term = "x" if abs_coef == 1 else f"{abs_coef}*x"
else:
term = f"x^{exp}" if abs_coef == 1 else f"{abs_coef}*x^{exp}"
if not parts:
parts.append(term if sign == "+" else f"-{term}")
else:
parts.append(f" {sign} {term}")
return "0" if not parts else "".join(parts)
def _wrap_add_zero(expr: str, rng: random.Random) -> str:
z = rng.choice(["0", "x-x", "2*x-2*x", "3-3"])
return f"({expr}) + ({z})"
def _wrap_mul_one(expr: str, rng: random.Random) -> str:
one = rng.choice(["1", "x^0", "(2-1)", "(3/3)"])
# v1 does not support division, so avoid (3/3) until rational support.
if "/" in one:
one = "1"
return f"({expr}) * ({one})"
def _equivalent_cases(rng: random.Random) -> Iterable[GeneratedCase]:
idx = 1
# 40 square-of-linear cases.
for _ in range(40):
lin, a, b = _linear(rng)
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=f"({lin})^2",
expression_b=_expanded_square(a, b),
expected="equivalent",
category="generated_square_of_linear",
provenance=f"adr-0131.1b:generated:seed={SEED}",
)
idx += 1
# 40 product-of-linears cases.
for _ in range(40):
left, a, b = _linear(rng)
right, c, d = _linear(rng)
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=f"({left})*({right})",
expression_b=_expanded_product(a, b, c, d),
expected="equivalent",
category="generated_product_of_linears",
provenance=f"adr-0131.1b:generated:seed={SEED}",
)
idx += 1
# 20 add-zero metamorphic cases.
for _ in range(20):
lin, _, _ = _linear(rng)
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=_wrap_add_zero(lin, rng),
expression_b=lin,
expected="equivalent",
category="generated_metamorphic_add_zero",
provenance=f"adr-0131.1b:generated:seed={SEED}",
)
idx += 1
# 20 multiply-one metamorphic cases.
for _ in range(20):
lin, _, _ = _linear(rng)
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=_wrap_mul_one(lin, rng),
expression_b=lin,
expected="equivalent",
category="generated_metamorphic_mul_one",
provenance=f"adr-0131.1b:generated:seed={SEED}",
)
idx += 1
def _not_equivalent_cases(rng: random.Random, start_idx: int) -> Iterable[GeneratedCase]:
# 30 near-miss cases. Each mutates a correct expansion by +1 in the
# constant term, creating a definite non-equivalence without leaving scope.
idx = start_idx
for _ in range(30):
left, a, b = _linear(rng)
right, c, d = _linear(rng)
terms = {2: a * c, 1: a * d + b * c, 0: b * d + 1}
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=f"({left})*({right})",
expression_b=_poly_to_expr(terms),
expected="not_equivalent",
category="generated_near_miss_constant",
provenance=f"adr-0131.1b:generated:seed={SEED}",
)
idx += 1
def _refusal_cases(start_idx: int) -> Iterable[GeneratedCase]:
scope_expanded = [
("x + y", "x + 1", "generated_multivariable_distinct"),
("x / 2", "x", "generated_constant_denominator_distinct"),
]
templates = [
("sin(x)", "x", "generated_refusal_function"),
("x^-1", "1", "generated_refusal_negative_exponent"),
("x +", "x", "generated_refusal_malformed"),
]
idx = start_idx
for expr_a, expr_b, category in scope_expanded:
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=expr_a,
expression_b=expr_b,
expected="not_equivalent",
category=category,
provenance=f"adr-0131.1b:generated:seed={SEED}:scope-expanded",
)
idx += 1
for expr_a, expr_b, category in templates:
yield GeneratedCase(
case_id=f"sym-eq-gen-v1-{idx:04d}",
expression_a=expr_a,
expression_b=expr_b,
expected="refused",
category=category,
provenance=f"adr-0131.1b:generated:seed={SEED}:adversarial",
)
idx += 1
def build_generated_cases() -> list[dict[str, str]]:
rng = random.Random(SEED)
cases = list(_equivalent_cases(rng))
cases.extend(_not_equivalent_cases(rng, len(cases) + 1))
cases.extend(_refusal_cases(len(cases) + 1))
return [c.as_dict() for c in cases]
if __name__ == "__main__":
import json
import sys
for case in build_generated_cases():
sys.stdout.write(json.dumps(case, sort_keys=True) + "\n")

View file

@ -0,0 +1,26 @@
{
"adr": "0131.1.B",
"benchmark": "symbolic_equivalence_v1_hardened",
"by_expected": {
"equivalent": 140,
"not_equivalent": 42,
"refused": 3
},
"by_source": {
"curated": 30,
"generated": 155
},
"case_count": 185,
"cases_sha256": "5b7b56d495fe7528a0a82e7d1131691bb438b093767ea36b0ac789be3f5e0876",
"curated_cases_path": "evals/math_symbolic_equivalence/v1/cases.jsonl",
"generated_cases_module": "evals.math_symbolic_equivalence.v1.generated_cases",
"generated_seed": 131101,
"replay_contract": {
"byte_equal_report_json": true,
"correct_rate_min": 0.95,
"deterministic_generation": true,
"wrong_max": 0
},
"report_sha256": "8d94522fadbac2e618ac59a3fe8158286faab641c7d39f6ed2ee3a064255f77b",
"schema_version": 1
}

View file

@ -0,0 +1,19 @@
"""Replay helpers for the ADR-0131.1 symbolic-equivalence lane."""
from __future__ import annotations
import hashlib
import json
from typing import Any
def canonical_json_bytes(obj: Any) -> bytes:
"""Return stable JSON bytes for digesting lane artifacts."""
return (json.dumps(obj, sort_keys=True, separators=(",", ":")) + "\n").encode(
"utf-8"
)
def sha256_obj(obj: Any) -> str:
"""Return SHA-256 over stable JSON serialization."""
return hashlib.sha256(canonical_json_bytes(obj)).hexdigest()

File diff suppressed because it is too large Load diff

View file

@ -1,22 +1,9 @@
"""ADR-0131.1 — Symbolic equivalence lane runner (v1).
"""ADR-0131.1 — Symbolic equivalence lane runner (v1 hardened).
Loads ``cases.jsonl``, runs each case through
:func:`generate.math_symbolic_equivalence.check_equivalence`, classifies
the outcome against the expected verdict, and writes a deterministic
``report.json``.
CLI: ``python -m evals.math_symbolic_equivalence.v1.runner``
exit code 0 if exit criterion passes, 1 otherwise.
Exit criterion (per ADR-0131 Benchmark 1):
correct_rate >= 0.95
wrong == 0
A case is ``correct`` iff its expected verdict matches the engine's
verdict (including expected=refused matched by REFUSED). It is
``wrong`` iff expected=equivalent but engine=not_equivalent, or
vice versa. It is ``refused`` iff engine=REFUSED on a case whose
expected verdict was a definite answer (equivalent / not_equivalent).
Loads ``cases.jsonl`` plus deterministic generated cases, runs each case
through :func:`generate.math_symbolic_equivalence.check_equivalence`,
classifies the outcome against the expected verdict, and writes
deterministic ``report.json`` and ``manifest.json`` artifacts.
"""
from __future__ import annotations
@ -27,6 +14,11 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any
from evals.math_symbolic_equivalence.v1.generated_cases import (
SEED as GENERATED_CASE_SEED,
build_generated_cases,
)
from evals.math_symbolic_equivalence.v1.replay import sha256_obj
from generate.math_symbolic_equivalence import (
Verdict,
check_equivalence,
@ -34,8 +26,10 @@ from generate.math_symbolic_equivalence import (
_HERE = Path(__file__).resolve().parent
_REPO_ROOT = _HERE.parent.parent.parent
_CASES_PATH = _HERE / "cases.jsonl"
_REPORT_PATH = _HERE / "report.json"
_MANIFEST_PATH = _HERE / "manifest.json"
# Per ADR-0131 Benchmark 1 exit criterion.
_CORRECT_RATE_MIN = 0.95
@ -72,13 +66,9 @@ def _score_one(case: dict[str, Any]) -> CaseOutcome:
verdict_class = "correct"
reason = ""
elif actual == Verdict.REFUSED.value:
# Engine refused on a case that expected a definite answer.
# This is a refusal, NOT a wrong answer — preserves wrong == 0.
verdict_class = "refused"
reason = v.reason
else:
# Engine produced a definite answer that disagrees with expected.
# This is wrong. The wrong==0 gate catches any such case.
verdict_class = "wrong"
reason = (
f"engine={actual!r} expected={expected!r}; "
@ -95,17 +85,51 @@ def _score_one(case: dict[str, Any]) -> CaseOutcome:
)
def _load_cases(path: Path = _CASES_PATH) -> list[dict[str, Any]]:
def _load_curated_cases(path: Path = _CASES_PATH) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
with path.open("r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
records.append(json.loads(line))
record = json.loads(line)
record["source"] = "curated"
records.append(record)
return records
def _load_generated_cases() -> list[dict[str, Any]]:
records = build_generated_cases()
for record in records:
record["source"] = "generated"
return records
def _load_cases(path: Path = _CASES_PATH) -> list[dict[str, Any]]:
cases = _load_curated_cases(path) + _load_generated_cases()
ids = [str(c["case_id"]) for c in cases]
if len(ids) != len(set(ids)):
duplicates = sorted({case_id for case_id in ids if ids.count(case_id) > 1})
raise RuntimeError(f"duplicate symbolic-equivalence case_id(s): {duplicates}")
return cases
def _source_counts(cases: list[dict[str, Any]]) -> dict[str, int]:
out = {"curated": 0, "generated": 0}
for c in cases:
source = str(c.get("source", "curated"))
out[source] = out.get(source, 0) + 1
return out
def _expected_counts(cases: list[dict[str, Any]]) -> dict[str, int]:
out = {"equivalent": 0, "not_equivalent": 0, "refused": 0}
for c in cases:
expected = str(c["expected"])
out[expected] = out.get(expected, 0) + 1
return out
def build_report(cases: list[dict[str, Any]]) -> dict[str, Any]:
outcomes = [_score_one(c) for c in cases]
counts = {"correct": 0, "wrong": 0, "refused": 0}
@ -116,12 +140,15 @@ def build_report(cases: list[dict[str, Any]]) -> dict[str, Any]:
correct_rate = counts["correct"] / total if total else 0.0
passed = (correct_rate >= _CORRECT_RATE_MIN) and (counts["wrong"] <= _WRONG_MAX)
return {
"schema_version": 1,
"adr": "0131.1",
"benchmark": "symbolic_equivalence_v1",
"cases_path": str(_CASES_PATH.relative_to(_HERE.parent.parent.parent)),
report: dict[str, Any] = {
"schema_version": 2,
"adr": "0131.1.B",
"benchmark": "symbolic_equivalence_v1_hardened",
"cases_path": str(_CASES_PATH.relative_to(_REPO_ROOT)),
"generated_cases_module": "evals.math_symbolic_equivalence.v1.generated_cases",
"generated_seed": GENERATED_CASE_SEED,
"sample_count": total,
"by_source": _source_counts(cases),
"counts": counts,
"correct_rate": correct_rate,
"exit_criterion": {
@ -131,19 +158,55 @@ def build_report(cases: list[dict[str, Any]]) -> dict[str, Any]:
},
"per_case": [o.as_dict() for o in outcomes],
}
report["report_sha256"] = sha256_obj(report)
return report
def build_manifest(cases: list[dict[str, Any]], report: dict[str, Any]) -> dict[str, Any]:
report_without_digest = dict(report)
report_without_digest.pop("report_sha256", None)
return {
"schema_version": 1,
"adr": "0131.1.B",
"benchmark": "symbolic_equivalence_v1_hardened",
"curated_cases_path": str(_CASES_PATH.relative_to(_REPO_ROOT)),
"generated_cases_module": "evals.math_symbolic_equivalence.v1.generated_cases",
"generated_seed": GENERATED_CASE_SEED,
"case_count": len(cases),
"by_source": _source_counts(cases),
"by_expected": _expected_counts(cases),
"cases_sha256": sha256_obj(cases),
"report_sha256": sha256_obj(report_without_digest),
"replay_contract": {
"byte_equal_report_json": True,
"deterministic_generation": True,
"correct_rate_min": _CORRECT_RATE_MIN,
"wrong_max": _WRONG_MAX,
},
}
def _write_json(obj: dict[str, Any], path: Path) -> None:
path.write_text(
json.dumps(obj, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
def write_report(report: dict[str, Any], path: Path = _REPORT_PATH) -> None:
path.write_text(
json.dumps(report, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
_write_json(report, path)
def write_manifest(manifest: dict[str, Any], path: Path = _MANIFEST_PATH) -> None:
_write_json(manifest, path)
def main() -> int:
cases = _load_cases()
report = build_report(cases)
manifest = build_manifest(cases, report)
write_report(report)
write_manifest(manifest)
return 0 if report["exit_criterion"]["passed"] else 1

View file

@ -1,19 +1,9 @@
"""ADR-0131.1 — Symbolic equivalence check (Benchmark 1 primitive).
"""ADR-0131.1.B — Symbolic equivalence check.
Given two algebraic expressions A and B, produces an
:class:`EquivalenceVerdict` of EQUIVALENT, NOT_EQUIVALENT, or REFUSED
(with reason). REFUSED preserves wrong == 0: the engine refuses to
guess on out-of-scope input rather than emit a wrong verdict.
Algorithm (v1, polynomial scope):
1. Normalize A via :func:`generate.math_symbolic_normalizer.normalize`.
2. Normalize B via the same function.
3. Compare canonical strings byte-for-byte.
If either normalization raises :class:`SymbolicError`, the verdict is
REFUSED with the propagating reason. This is the wrong-answer
firewall for the benchmark anything the normalizer cannot prove
equivalent (or prove distinct) deterministically is refused.
:class:`EquivalenceVerdict` of EQUIVALENT, NOT_EQUIVALENT, or REFUSED.
REFUSED preserves wrong == 0: the engine refuses to guess on
out-of-scope input rather than emit a wrong verdict.
"""
from __future__ import annotations
@ -37,51 +27,59 @@ class Verdict(str, Enum):
@dataclass(frozen=True, slots=True)
class EquivalenceVerdict:
verdict: Verdict
canonical_a: str | None # None when verdict is REFUSED and a couldn't normalize
canonical_a: str | None
canonical_b: str | None
reason: str # empty on EQUIVALENT / NOT_EQUIVALENT; non-empty on REFUSED
reason: str
REFUSED_VERDICTS: Final[frozenset[Verdict]] = frozenset({Verdict.REFUSED})
"""Helper set for callers that need to gate on refusal vs decision."""
def _normalize_pair(
expression_a: str,
expression_b: str,
*,
variable: str | None,
variables: tuple[str, ...] | None,
) -> tuple[str, str]:
if variables is None and variable is None:
# Infer variables from the union of both expressions so `x + y` and
# `y + x` normalize in the same variable space.
poly_a_probe = normalize(expression_a)
poly_b_probe = normalize(expression_b)
variables = tuple(sorted(set(poly_a_probe.variables) | set(poly_b_probe.variables)))
canon_a = normalize(expression_a, variable=variable, variables=variables).to_canonical_string()
canon_b = normalize(expression_b, variable=variable, variables=variables).to_canonical_string()
return canon_a, canon_b
def check_equivalence(
expression_a: str,
expression_b: str,
*,
variable: str = "x",
variable: str | None = None,
variables: tuple[str, ...] | None = None,
) -> EquivalenceVerdict:
"""Return whether ``expression_a`` and ``expression_b`` are
algebraically equivalent under the v1 polynomial-normalizer scope.
"""Return whether two expressions are algebraically equivalent.
Refusal cases (each surfaces a typed reason):
- Either expression is empty or non-string.
- Either expression uses an out-of-scope identifier (multi-
variable, undefined name).
- Either expression contains a syntactically invalid construct.
- Either expression uses division, transcendental functions,
non-integer coefficients, negative exponents, or non-constant
exponents.
``variable`` is retained for backward compatibility with the v1
univariate API. New callers can omit it and allow variable inference, or
pass an explicit sorted ``variables`` tuple.
"""
try:
canon_a = normalize(expression_a, variable=variable).to_canonical_string()
canon_a, canon_b = _normalize_pair(
expression_a,
expression_b,
variable=variable,
variables=variables,
)
except SymbolicError as exc:
return EquivalenceVerdict(
verdict=Verdict.REFUSED,
canonical_a=None,
canonical_b=None,
reason=f"normalize(a) refused: {exc}",
)
try:
canon_b = normalize(expression_b, variable=variable).to_canonical_string()
except SymbolicError as exc:
return EquivalenceVerdict(
verdict=Verdict.REFUSED,
canonical_a=canon_a,
canonical_b=None,
reason=f"normalize(b) refused: {exc}",
reason=f"normalize refused: {exc}",
)
if canon_a == canon_b:

View file

@ -1,139 +1,134 @@
"""ADR-0131.1 — Deterministic symbolic normalizer for univariate
integer-coefficient polynomials.
"""ADR-0131.1.B — Deterministic symbolic normalizer for exact polynomials.
Scope (v1, intentionally narrow):
- Single variable (configurable, default 'x').
- Integer coefficients only.
- Operators: +, -, *, ** (positive integer exponents only).
Scope:
- One or more symbolic variables.
- Exact integer or rational coefficients via fractions.Fraction.
- Operators: +, -, *, / by numeric constants, ** with non-negative
integer exponents.
- Parentheses for grouping.
- No division (except implicit unary).
- No transcendental functions, no multi-variable, no rationals.
- No division by symbolic expressions yet.
- No transcendental functions.
The normalizer is the load-bearing primitive for the symbolic
equivalence benchmark (ADR-0131 Benchmark 1). Two expressions A and
B are equivalent iff their canonical forms are byte-equal. This is
the CGA exact-recall discriminator framed in algebra.
Determinism guarantees:
- Pure functions, no global state, no randomness.
- Same input string same canonical string, byte-for-byte.
- Same canonical string same Polynomial dataclass.
- Refuses (raises SymbolicError) rather than guessing on
out-of-scope input (preserves wrong == 0).
Architecture: tokenize parse to AST expand + collect canonical
serialize. Each stage is independently testable.
Two expressions A and B are equivalent iff their canonical polynomial
forms are byte-equal. Refusal is first-class: unsupported input raises
SymbolicError rather than producing a guess.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from fractions import Fraction
from typing import Final
# ---------------------------------------------------------------------------
# Public errors
# ---------------------------------------------------------------------------
class SymbolicError(ValueError):
"""Raised on tokens, syntax, or operators the normalizer cannot
deterministically handle. Refusal is first-class the caller is
expected to treat this as an explicit refusal, not a wrong answer.
"""
"""Raised on tokens, syntax, or operators the normalizer cannot handle."""
# ---------------------------------------------------------------------------
# Canonical polynomial representation
# ---------------------------------------------------------------------------
Coeff = Fraction
def _as_fraction(value: int | Fraction) -> Fraction:
if isinstance(value, bool):
raise SymbolicError("boolean coefficients are not allowed")
if isinstance(value, Fraction):
return value
if isinstance(value, int):
return Fraction(value, 1)
raise SymbolicError(f"unsupported coefficient type {type(value).__name__}")
def _format_coeff(value: Fraction) -> str:
if value.denominator == 1:
return str(value.numerator)
return f"{value.numerator}/{value.denominator}"
@dataclass(frozen=True, slots=True)
class Polynomial:
"""A univariate polynomial in canonical form.
"""A multivariable exact polynomial in canonical sparse form."""
``coefficients`` is a tuple of integers, index = exponent.
coefficients[0] = constant term, coefficients[1] = x coefficient,
coefficients[2] = x^2 coefficient, etc. Trailing zeros are
stripped; the tuple is empty iff the polynomial is the zero
polynomial.
Two Polynomial instances are equal iff their coefficient tuples
are equal. This is the equivalence relation the benchmark tests.
"""
coefficients: tuple[int, ...]
variable: str = "x"
terms: dict[tuple[int, ...], int | Fraction]
variables: tuple[str, ...] = ("x",)
def __post_init__(self) -> None:
if not isinstance(self.variable, str) or not self.variable.isidentifier():
raise SymbolicError(
f"Polynomial.variable must be a Python identifier; "
f"got {self.variable!r}"
)
if not all(isinstance(c, int) for c in self.coefficients):
raise SymbolicError(
"Polynomial.coefficients must all be int "
"(no float, no bool, no fraction in v1)"
)
# Trailing zeros must be stripped at construction; reject
# non-canonical input loudly so downstream comparison is
# unambiguous.
if self.coefficients and self.coefficients[-1] == 0:
raise SymbolicError(
f"Polynomial.coefficients must have no trailing zeros; "
f"got {self.coefficients}"
)
def to_canonical_string(self) -> str:
"""Render this polynomial in a single canonical string form.
Terms are emitted in descending exponent order with explicit
signs. The zero polynomial is rendered as ``"0"``. This is
the byte-level comparison key for equivalence.
"""
if not self.coefficients:
return "0"
parts: list[str] = []
for exp in range(len(self.coefficients) - 1, -1, -1):
coef = self.coefficients[exp]
if not self.variables:
raise SymbolicError("Polynomial.variables must be non-empty")
if tuple(sorted(self.variables)) != self.variables:
raise SymbolicError(f"variables must be sorted; got {self.variables}")
if len(set(self.variables)) != len(self.variables):
raise SymbolicError(f"duplicate variables: {self.variables}")
for v in self.variables:
if not isinstance(v, str) or not v.isidentifier():
raise SymbolicError(f"invalid variable name {v!r}")
clean: dict[tuple[int, ...], Fraction] = {}
for exps, raw_coef in self.terms.items():
coef = _as_fraction(raw_coef)
if coef == 0:
continue
if len(exps) != len(self.variables):
raise SymbolicError(
f"exponent tuple length {len(exps)} does not match variables {self.variables}"
)
if any((not isinstance(e, int)) or e < 0 for e in exps):
raise SymbolicError(f"invalid exponent tuple {exps!r}")
clean[tuple(exps)] = coef
object.__setattr__(self, "terms", clean)
@property
def coefficients(self) -> tuple[int | Fraction, ...]:
if len(self.variables) != 1:
raise SymbolicError("coefficients view is univariate-only")
if not self.terms:
return ()
max_exp = max(exps[0] for exps in self.terms)
out: list[int | Fraction] = [0] * (max_exp + 1)
for exps, coef in self.terms.items():
out[exps[0]] = coef.numerator if coef.denominator == 1 else coef
while out and out[-1] == 0:
out.pop()
return tuple(out)
@property
def variable(self) -> str:
if len(self.variables) != 1:
raise SymbolicError("variable view is univariate-only")
return self.variables[0]
def to_canonical_string(self) -> str:
if not self.terms:
return "0"
parts: list[str] = []
for exps, coef in sorted(self.terms.items(), key=lambda kv: kv[0], reverse=True):
sign = "+" if coef >= 0 else "-"
abs_coef = abs(coef)
if exp == 0:
term = f"{abs_coef}"
elif exp == 1:
term = f"{self.variable}" if abs_coef == 1 else f"{abs_coef}*{self.variable}"
monomial_parts: list[str] = []
for variable, exp in zip(self.variables, exps):
if exp == 0:
continue
if exp == 1:
monomial_parts.append(variable)
else:
monomial_parts.append(f"{variable}^{exp}")
if monomial_parts:
mono = "*".join(monomial_parts)
term = mono if abs_coef == 1 else f"{_format_coeff(abs_coef)}*{mono}"
else:
term = (
f"{self.variable}^{exp}"
if abs_coef == 1
else f"{abs_coef}*{self.variable}^{exp}"
)
term = _format_coeff(abs_coef)
if not parts:
# Leading term: no leading "+" sign.
parts.append(term if sign == "+" else f"-{term}")
else:
parts.append(f"{sign}{term}")
return "".join(parts)
# ---------------------------------------------------------------------------
# Tokenizer
# ---------------------------------------------------------------------------
_TOKEN_RE: Final[re.Pattern[str]] = re.compile(
r"\s*(?:(?P<int>\d+)|(?P<ident>[A-Za-z_]\w*)|(?P<op>\*\*|[+\-*()^]))"
r"\s*(?:(?P<int>\d+)|(?P<ident>[A-Za-z_]\w*)|(?P<op>\*\*|[+\-*/()^]))"
)
def _tokenize(text: str) -> list[tuple[str, str]]:
"""Return a list of ``(kind, lexeme)`` tokens.
Kinds: ``"int"``, ``"ident"``, ``"op"``. The ``"^"`` operator is
normalized to the canonical Python-style ``"**"`` (both spellings
accepted on input).
"""
pos = 0
tokens: list[tuple[str, str]] = []
while pos < len(text):
@ -153,31 +148,19 @@ def _tokenize(text: str) -> list[tuple[str, str]]:
return tokens
# ---------------------------------------------------------------------------
# Recursive-descent parser producing a normalized Polynomial.
#
# Grammar:
# expr := term (('+' | '-') term)*
# term := factor (('*') factor)* # implicit '*' between (expr) and ident
# factor := unary ('**' unary)?
# unary := ('+' | '-') unary | atom
# atom := INT | IDENT | '(' expr ')'
#
# Each grammar rule returns a Polynomial; addition / multiplication /
# negation / exponentiation are implemented as Polynomial operations.
# This is the "expand + collect" step inlined into parsing.
# ---------------------------------------------------------------------------
def _infer_variables(tokens: list[tuple[str, str]]) -> tuple[str, ...]:
names = sorted({lex for kind, lex in tokens if kind == "ident"})
return tuple(names) if names else ("x",)
class _Parser:
def __init__(self, tokens: list[tuple[str, str]], variable: str) -> None:
def __init__(self, tokens: list[tuple[str, str]], variables: tuple[str, ...]) -> None:
self._tokens = tokens
self._pos = 0
self._variable = variable
self._variables = variables
def _peek(self) -> tuple[str, str] | None:
if self._pos >= len(self._tokens):
return None
return self._tokens[self._pos]
return None if self._pos >= len(self._tokens) else self._tokens[self._pos]
def _consume(self) -> tuple[str, str]:
if self._pos >= len(self._tokens):
@ -189,8 +172,7 @@ class _Parser:
def parse(self) -> Polynomial:
result = self._expr()
if self._pos != len(self._tokens):
extra = self._tokens[self._pos]
raise SymbolicError(f"unexpected trailing token {extra!r}")
raise SymbolicError(f"unexpected trailing token {self._tokens[self._pos]!r}")
return result
def _expr(self) -> Polynomial:
@ -201,10 +183,7 @@ class _Parser:
break
self._consume()
right = self._term()
if tok[1] == "+":
left = _add(left, right)
else:
left = _sub(left, right)
left = _add(left, right) if tok[1] == "+" else _sub(left, right)
return left
def _term(self) -> Polynomial:
@ -213,11 +192,14 @@ class _Parser:
tok = self._peek()
if tok is None:
break
# Explicit '*'
if tok[0] == "op" and tok[1] == "*":
self._consume()
right = self._factor()
left = _mul(left, right)
left = _mul(left, self._factor())
continue
if tok[0] == "op" and tok[1] == "/":
self._consume()
divisor = self._factor()
left = _div(left, divisor)
continue
break
return left
@ -225,21 +207,16 @@ class _Parser:
def _factor(self) -> Polynomial:
base = self._unary()
tok = self._peek()
if tok is not None and tok[0] == "op" and tok[1] == "**":
if tok is not None and tok == ("op", "**"):
self._consume()
exp_tok = self._unary()
# Exponent must be a non-negative integer constant polynomial.
if len(exp_tok.coefficients) > 1:
raise SymbolicError(
"exponent must be a non-negative integer constant; "
"got non-constant polynomial"
)
exp_val = exp_tok.coefficients[0] if exp_tok.coefficients else 0
if exp_val < 0:
raise SymbolicError(
f"exponent must be non-negative; got {exp_val}"
)
return _pow(base, exp_val)
exp_poly = self._unary()
exp_val = _constant_value(exp_poly)
if exp_val.denominator != 1:
raise SymbolicError("exponent must be an integer constant")
exponent = exp_val.numerator
if exponent < 0:
raise SymbolicError(f"exponent must be non-negative; got {exponent}")
return _pow(base, exponent)
return base
def _unary(self) -> Polynomial:
@ -247,22 +224,17 @@ class _Parser:
if tok is not None and tok[0] == "op" and tok[1] in ("+", "-"):
self._consume()
inner = self._unary()
if tok[1] == "-":
return _neg(inner)
return inner
return _neg(inner) if tok[1] == "-" else inner
return self._atom()
def _atom(self) -> Polynomial:
tok = self._consume()
if tok[0] == "int":
return _const(int(tok[1]), self._variable)
return _const(Fraction(int(tok[1]), 1), self._variables)
if tok[0] == "ident":
if tok[1] != self._variable:
raise SymbolicError(
f"v1 supports a single variable {self._variable!r}; "
f"got identifier {tok[1]!r}"
)
return _x(self._variable)
if tok[1] not in self._variables:
raise SymbolicError(f"identifier {tok[1]!r} is outside variable set")
return _var(tok[1], self._variables)
if tok == ("op", "("):
inner = self._expr()
close = self._consume()
@ -272,46 +244,63 @@ class _Parser:
raise SymbolicError(f"unexpected token {tok!r}")
# ---------------------------------------------------------------------------
# Polynomial algebra primitives (the actual "expand and collect" engine)
# ---------------------------------------------------------------------------
def _strip_trailing_zeros(coeffs: list[int]) -> tuple[int, ...]:
while coeffs and coeffs[-1] == 0:
coeffs.pop()
return tuple(coeffs)
def _zero_key(variables: tuple[str, ...]) -> tuple[int, ...]:
return (0,) * len(variables)
def _const(value: int, variable: str) -> Polynomial:
if value == 0:
return Polynomial(coefficients=(), variable=variable)
return Polynomial(coefficients=(value,), variable=variable)
def _const(value: int | Fraction, variables: tuple[str, ...]) -> Polynomial:
coef = _as_fraction(value)
if coef == 0:
return Polynomial(terms={}, variables=variables)
return Polynomial(terms={_zero_key(variables): coef}, variables=variables)
def _x(variable: str) -> Polynomial:
return Polynomial(coefficients=(0, 1), variable=variable)
def _var(name: str, variables: tuple[str, ...]) -> Polynomial:
exps = [0] * len(variables)
exps[variables.index(name)] = 1
return Polynomial(terms={tuple(exps): Fraction(1, 1)}, variables=variables)
def _constant_value(poly: Polynomial) -> Fraction:
if not poly.terms:
return Fraction(0, 1)
zero_key = _zero_key(poly.variables)
if set(poly.terms.keys()) == {zero_key}:
return poly.terms[zero_key]
raise SymbolicError("expected a constant polynomial")
def _align(poly: Polynomial, variables: tuple[str, ...]) -> Polynomial:
if poly.variables == variables:
return poly
positions = [variables.index(v) for v in poly.variables]
out: dict[tuple[int, ...], Fraction] = {}
for exps, coef in poly.terms.items():
new_exps = [0] * len(variables)
for old_i, new_i in enumerate(positions):
new_exps[new_i] = exps[old_i]
out[tuple(new_exps)] = coef
return Polynomial(terms=out, variables=variables)
def _common_variables(a: Polynomial, b: Polynomial) -> tuple[str, ...]:
return tuple(sorted(set(a.variables) | set(b.variables)))
def _add(a: Polynomial, b: Polynomial) -> Polynomial:
if a.variable != b.variable:
raise SymbolicError(
f"variable mismatch: {a.variable!r} vs {b.variable!r}"
)
n = max(len(a.coefficients), len(b.coefficients))
out = [0] * n
for i, c in enumerate(a.coefficients):
out[i] += c
for i, c in enumerate(b.coefficients):
out[i] += c
return Polynomial(
coefficients=_strip_trailing_zeros(out), variable=a.variable
)
variables = _common_variables(a, b)
a = _align(a, variables)
b = _align(b, variables)
out = dict(a.terms)
for exps, coef in b.terms.items():
out[exps] = out.get(exps, Fraction(0, 1)) + coef
if out[exps] == 0:
del out[exps]
return Polynomial(terms=out, variables=variables)
def _neg(a: Polynomial) -> Polynomial:
return Polynomial(
coefficients=tuple(-c for c in a.coefficients), variable=a.variable
)
return Polynomial(terms={exps: -coef for exps, coef in a.terms.items()}, variables=a.variables)
def _sub(a: Polynomial, b: Polynomial) -> Polynomial:
@ -319,52 +308,64 @@ def _sub(a: Polynomial, b: Polynomial) -> Polynomial:
def _mul(a: Polynomial, b: Polynomial) -> Polynomial:
if a.variable != b.variable:
raise SymbolicError(
f"variable mismatch: {a.variable!r} vs {b.variable!r}"
)
if not a.coefficients or not b.coefficients:
return Polynomial(coefficients=(), variable=a.variable)
out = [0] * (len(a.coefficients) + len(b.coefficients) - 1)
for i, ca in enumerate(a.coefficients):
if ca == 0:
continue
for j, cb in enumerate(b.coefficients):
out[i + j] += ca * cb
variables = _common_variables(a, b)
a = _align(a, variables)
b = _align(b, variables)
if not a.terms or not b.terms:
return Polynomial(terms={}, variables=variables)
out: dict[tuple[int, ...], Fraction] = {}
for exps_a, coef_a in a.terms.items():
for exps_b, coef_b in b.terms.items():
exps = tuple(x + y for x, y in zip(exps_a, exps_b))
out[exps] = out.get(exps, Fraction(0, 1)) + coef_a * coef_b
if out[exps] == 0:
del out[exps]
return Polynomial(terms=out, variables=variables)
def _div(a: Polynomial, b: Polynomial) -> Polynomial:
divisor = _constant_value(b)
if divisor == 0:
raise SymbolicError("division by zero")
return Polynomial(
coefficients=_strip_trailing_zeros(out), variable=a.variable
terms={exps: coef / divisor for exps, coef in a.terms.items()},
variables=a.variables,
)
def _pow(base: Polynomial, exponent: int) -> Polynomial:
if exponent == 0:
return _const(1, base.variable)
result = base
for _ in range(exponent - 1):
return _const(1, base.variables)
result = _const(1, base.variables)
for _ in range(exponent):
result = _mul(result, base)
return result
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def normalize(expression: str, *, variable: str = "x") -> Polynomial:
"""Parse + expand + collect ``expression`` into canonical Polynomial.
Raises :class:`SymbolicError` on any input the v1 normalizer
cannot deterministically handle (multi-variable, division,
non-integer coefficient, unknown identifier, syntax error,
negative exponent, non-constant exponent).
"""
def normalize(
expression: str,
*,
variable: str | None = None,
variables: tuple[str, ...] | None = None,
) -> Polynomial:
"""Parse + expand + collect ``expression`` into canonical Polynomial."""
if not isinstance(expression, str) or not expression.strip():
raise SymbolicError("empty or non-string expression")
tokens = _tokenize(expression)
if not tokens:
raise SymbolicError("no tokens parsed from expression")
return _Parser(tokens, variable).parse()
if variable is not None and variables is not None:
raise SymbolicError("pass either variable or variables, not both")
if variables is None:
variables = (variable,) if variable is not None else _infer_variables(tokens)
variables = tuple(sorted(variables))
return _Parser(tokens, variables).parse()
def canonical_string(expression: str, *, variable: str = "x") -> str:
"""Shortcut: ``normalize(expression).to_canonical_string()``."""
return normalize(expression, variable=variable).to_canonical_string()
def canonical_string(
expression: str,
*,
variable: str | None = None,
variables: tuple[str, ...] | None = None,
) -> str:
return normalize(expression, variable=variable, variables=variables).to_canonical_string()

View file

@ -55,24 +55,28 @@ class TestRefused:
def test_empty_left(self) -> None:
v = check_equivalence("", "x + 1")
assert v.verdict == Verdict.REFUSED
assert "normalize(a) refused" in v.reason
assert "empty" in v.reason
def test_out_of_scope_variable_left(self) -> None:
def test_multivariable_now_admits(self) -> None:
# ADR-0131.1.B scope expansion: multivariable polynomials are admissible.
v = check_equivalence("x + y", "x + 1")
assert v.verdict == Verdict.REFUSED
assert "single variable" in v.reason
assert v.verdict == Verdict.NOT_EQUIVALENT
def test_division_refused(self) -> None:
def test_constant_denominator_now_admits(self) -> None:
# ADR-0131.1.B scope expansion: constant-denominator division admits.
v = check_equivalence("x/2", "x")
assert v.verdict == Verdict.NOT_EQUIVALENT
def test_symbolic_denominator_still_refused(self) -> None:
# Symbolic-denominator division stays out of scope.
v = check_equivalence("x/y", "x")
assert v.verdict == Verdict.REFUSED
def test_a_normalizes_b_refuses(self) -> None:
# a is fine, b uses y -> refusal with canonical_a populated
v = check_equivalence("x + 1", "y + 1")
# a is fine, b uses a transcendental -> refusal.
v = check_equivalence("x + 1", "sin(x)")
assert v.verdict == Verdict.REFUSED
assert v.canonical_a == "x+1"
assert v.canonical_b is None
assert "normalize(b) refused" in v.reason
def test_refused_verdict_is_first_class(self) -> None:
# Refusal preserves wrong == 0 — the verdict is REFUSED, never

View file

@ -167,9 +167,10 @@ class TestRefusals:
with pytest.raises(SymbolicError, match="empty"):
normalize("")
def test_undefined_variable(self) -> None:
with pytest.raises(SymbolicError, match="single variable"):
normalize("x + y") # y is out of scope
def test_multivariable_now_admits(self) -> None:
# ADR-0131.1.B scope expansion: multivariable polynomials admit.
poly = normalize("x + y")
assert poly.to_canonical_string() == "x+y"
def test_negative_exponent(self) -> None:
with pytest.raises(SymbolicError, match="non-negative"):
@ -187,9 +188,14 @@ class TestRefusals:
with pytest.raises(SymbolicError):
normalize("x +")
def test_unknown_operator_division(self) -> None:
def test_constant_denominator_now_admits(self) -> None:
# ADR-0131.1.B scope expansion: constant-denominator division admits.
poly = normalize("x / 2")
assert poly.to_canonical_string() == "1/2*x"
def test_symbolic_denominator_still_refused(self) -> None:
with pytest.raises(SymbolicError):
normalize("x / 2")
normalize("x / y")
# ---------------------------------------------------------------------------
@ -197,21 +203,24 @@ class TestRefusals:
# ---------------------------------------------------------------------------
class TestPolynomialInvariants:
def test_trailing_zero_rejected(self) -> None:
with pytest.raises(SymbolicError, match="trailing zeros"):
Polynomial(coefficients=(1, 2, 0), variable="x")
def test_zero_coefficient_terms_collapse(self) -> None:
# Sparse multivariable repr canonicalizes by dropping zero-coef terms.
assert (
Polynomial(terms={(2,): 1, (1,): 2, (0,): 0}, variables=("x",)).to_canonical_string()
== "x^2+2*x"
)
def test_float_rejected(self) -> None:
with pytest.raises(SymbolicError, match="int"):
Polynomial(coefficients=(1.5,), variable="x") # type: ignore[arg-type]
with pytest.raises(SymbolicError, match="float"):
Polynomial(terms={(0,): 1.5}, variables=("x",)) # type: ignore[dict-item]
def test_zero_polynomial_is_empty_tuple(self) -> None:
# Zero polynomial canonical form has empty coefficients tuple.
assert Polynomial(coefficients=(), variable="x").to_canonical_string() == "0"
def test_zero_polynomial(self) -> None:
# Zero polynomial canonical form has empty terms dict.
assert Polynomial(terms={}, variables=("x",)).to_canonical_string() == "0"
def test_equality(self) -> None:
a = Polynomial(coefficients=(1, 2, 3), variable="x")
b = Polynomial(coefficients=(1, 2, 3), variable="x")
a = Polynomial(terms={(2,): 3, (1,): 2, (0,): 1}, variables=("x",))
b = Polynomial(terms={(2,): 3, (1,): 2, (0,): 1}, variables=("x",))
assert a == b
c = Polynomial(coefficients=(1, 2, 4), variable="x")
c = Polynomial(terms={(2,): 4, (1,): 2, (0,): 1}, variables=("x",))
assert a != c