arborist/bench/batteries/b_5s.py
russell@unturf.com d6f3834ab7
pi_star: code-py-ast@v1 graduates from stub — first non-text canonicalizer
Implements the cross-modality discipline that ticket #000015 spelled
out. Until now only text + claim_lattice + memory carriers existed
in code; the multimodal story was theoretical.

code-py-ast@v1 algorithm:

1. Parse UTF-8 bytes via stdlib ast.parse.
2. Walk the AST; emit deterministic S-expression
   "(NodeType field1=val1 ...)" with sorted fields, lists as
   "[elem0 elem1 ...]", primitives via repr().
3. Source positions (lineno/col_offset) skipped naturally — not in
   ast._fields.

Equivalence classes preserved (verified by tests):

- Whitespace, indentation, blank lines
- Comments
- String quote style ('x' vs "x")
- Operator spacing (1+2 vs 1 + 2)
- Trailing semicolons (x=1;y=2 vs x=1\ny=2)

Equivalence classes kept distinct:

- Identifier names (def foo vs def bar)
- Operator types (Add vs Sub)
- Argument order in calls (f(x,y) vs f(y,x))

Projective, not invertible — canonical bytes are S-expression text,
NOT valid Python. Re-canonicalizing the canonical output is
undefined; idempotency tests run on the original raw input only.

Surface:

- arborist/pi_star/code.py — full implementation, replaces stub
- bench/batteries/base.py — PHASE_1_CARRIERS adds "code"
- bench/batteries/b_5s.py — runner accepts both pi_star_ref
  (cross-modality canonical) and pi_star (Phase 1a legacy) keys;
  backward-compat shim
- bench/fixtures/5s/syntax-code-v1.jsonl — 10 Python parse-pass
  fixtures
- bench/fixtures/5s/semantics-code-v1.jsonl — 12 equivalence-class
  fixtures (whitespace/comments/quote-style/operator collapse;
  identifier/operator/argument-order remain distinct)
- Makefile: bench-5s-code target
- tests/test_pi_star.py: stubs parametrize drops code-py-ast
  (graduated); 11 new tests for code-py-ast@v1 covering equivalence
  classes, distinguishing classes, syntax-error rejection, non-bytes
  rejection, determinism, empty-source handling, equivalence_class_id
  format

Three remaining stubs (logic-kernel, time-series-quantized,
tabular-pinned) keep their NotImplementedError contract.

Full suite: 1221 passed, 36 skipped. Cross-modality discipline now
has actual code-level proof, not just schema metadata.
2026-05-08 08:50:07 -04:00

604 lines
21 KiB
Python

"""5S battery — Syntax / Semantics / Syllogism / Synthesis / Semiotics.
Phase 1a (landed in ticket #000021): Syntax + Semantics over the
existing π* registry.
Phase 1b (ticket #000023, this file): Syllogism, Synthesis,
Semiotics — all carrier-aware (text / claim_lattice / prose carriers
in v1; future-carrier hooks via fixture metadata that v1 rejects
explicitly).
No LLM-as-judge anywhere. All runners deterministic.
"""
from __future__ import annotations
from pathlib import Path
from bench.batteries.base import (
BatteryResult,
TaskResult,
fixture_digest,
fixture_meta,
iter_tasks,
validate_carrier,
)
def _runtime_digest() -> str:
"""SHA-256 fingerprint of the active π* registry.
Includes name@version of every registered π* so a registry change
surfaces in the runtime_digest.
"""
import hashlib
from arborist.pi_star import list_keys
payload = "\n".join(list_keys()).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def _carrier_check(task: dict) -> tuple[bool, str]:
"""Return (ok, reason). ok=False means task fails immediately."""
reason = validate_carrier(task)
if reason is not None:
return False, reason
return True, ""
# ---------------------------------------------------------------------
# Phase 1a: Syntax + Semantics (landed in ticket #000021)
# ---------------------------------------------------------------------
def run_syntax(fixtures_path: Path) -> BatteryResult:
"""Each task asserts that the named π* canonicalizes the input
without raising. expected="pass" is the only supported expected;
other values fail loudly."""
from arborist.pi_star import get
per_task: list[TaskResult] = []
pass_count = 0
fail_count = 0
for task in iter_tasks(fixtures_path):
task_id = task["id"]
ok, reason = _carrier_check(task)
if not ok:
per_task.append(
TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
)
fail_count += 1
continue
detail: dict = {}
try:
pi_star = get(task.get("pi_star_ref", task.get("pi_star")))
pi_star.canonicalize(task["input"].encode("utf-8"))
ok = task.get("expected", "pass") == "pass"
if not ok:
detail["reason"] = "canonicalize succeeded but expected != pass"
except NotImplementedError:
detail["reason"] = "pi_star is a stub"
ok = False
except Exception as exc: # noqa: BLE001
detail["reason"] = f"{type(exc).__name__}: {exc}"
ok = False
if ok:
pass_count += 1
else:
fail_count += 1
per_task.append(TaskResult(task_id=task_id, passed=ok, detail=detail))
total = pass_count + fail_count
pass_rate = pass_count / total if total else 0.0
meta = fixture_meta(fixtures_path)
return BatteryResult(
battery="5s",
sub_battery=meta.get("sub_battery", "syntax"),
fixture_path=str(fixtures_path),
fixture_digest=fixture_digest(fixtures_path),
pass_count=pass_count,
fail_count=fail_count,
metrics={"parse_pass_rate": pass_rate},
per_task=per_task,
runtime_digest=_runtime_digest(),
)
def run_semantics(fixtures_path: Path) -> BatteryResult:
"""Each task: two inputs + a π*; passes if canonicalize agrees with
expected_equivalent."""
from arborist.pi_star import get
per_task: list[TaskResult] = []
pass_count = 0
fail_count = 0
for task in iter_tasks(fixtures_path):
task_id = task["id"]
ok, reason = _carrier_check(task)
if not ok:
per_task.append(
TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
)
fail_count += 1
continue
detail: dict = {}
try:
pi_star = get(task.get("pi_star_ref", task.get("pi_star")))
ca = pi_star.canonicalize(task["input_a"].encode("utf-8"))
cb = pi_star.canonicalize(task["input_b"].encode("utf-8"))
equivalent = ca == cb
ok = equivalent == bool(task["expected_equivalent"])
if not ok:
detail["reason"] = (
f"observed_equivalent={equivalent} "
f"expected={task['expected_equivalent']}"
)
except Exception as exc: # noqa: BLE001
detail["reason"] = f"{type(exc).__name__}: {exc}"
ok = False
if ok:
pass_count += 1
else:
fail_count += 1
per_task.append(TaskResult(task_id=task_id, passed=ok, detail=detail))
total = pass_count + fail_count
rate = pass_count / total if total else 0.0
meta = fixture_meta(fixtures_path)
return BatteryResult(
battery="5s",
sub_battery=meta.get("sub_battery", "semantics"),
fixture_path=str(fixtures_path),
fixture_digest=fixture_digest(fixtures_path),
pass_count=pass_count,
fail_count=fail_count,
metrics={"equivalence_recovery_rate": rate},
per_task=per_task,
runtime_digest=_runtime_digest(),
)
# ---------------------------------------------------------------------
# Phase 1b: Syllogism (ticket #000023)
# ---------------------------------------------------------------------
# Whitelist of syllogistic rules the deterministic runner can license.
# Each rule has a small kernel that returns (valid: bool, reason: str).
def _syllogism_rule_categorical_transitivity(
premises: list[dict], candidate: dict
) -> tuple[bool, str]:
"""All A are B; All B are C ⇒ All A are C.
Premise text shape: ``All <X> are <Y>.`` (also accepts "is").
Order-agnostic: premises may appear in either chain order so a
natural English ordering ("All mammals are animals; All dogs are
mammals") and the textbook ordering ("All dogs are mammals; All
mammals are animals") both validate.
"""
if len(premises) != 2:
return False, "categorical_transitivity needs exactly 2 premises"
p1 = _parse_categorical(premises[0]["claim"])
p2 = _parse_categorical(premises[1]["claim"])
c = _parse_categorical(candidate["claim"])
if p1 is None or p2 is None or c is None:
return False, "claim does not match 'All X are Y.' pattern"
# Try both orderings so fixture authors can write premises in
# natural English order (general first) or textbook order
# (specific first).
for first, second in [(p1, p2), (p2, p1)]:
if first[1] == second[0]:
if c[0] == first[0] and c[1] == second[1]:
return True, ""
return False, "candidate endpoints do not match chain"
return False, f"premises do not chain: {p1} vs {p2}"
def _parse_categorical(text: str) -> tuple[str, str] | None:
"""Parse 'All <X> are <Y>.' or 'All <X> is <Y>.' into (X, Y)."""
s = text.strip().rstrip(".")
lower = s.lower()
if not lower.startswith("all "):
return None
rest = s[4:]
for connector in (" are ", " is "):
parts = rest.split(connector, 1)
if len(parts) == 2:
return parts[0].strip(), parts[1].strip()
return None
def _syllogism_rule_invalid_converse(
premises: list[dict], candidate: dict
) -> tuple[bool, str]:
"""Candidate is the converse of a premise. Must FAIL.
A fixture tagged ``rule="invalid_converse"`` with ``expected="fail"``
asserts that the runner correctly identifies an invalid step.
"""
if len(premises) < 1:
return False, "invalid_converse needs at least 1 premise"
c = _parse_categorical(candidate["claim"])
if c is None:
return False, "candidate not a categorical claim"
for prem in premises:
p = _parse_categorical(prem["claim"])
if p is None:
continue
if p[0] == c[1] and p[1] == c[0]:
# Converse detected — invalid step.
return False, "invalid converse of premise"
# Not actually a converse — the fixture is mis-tagged.
return False, "no converse pair detected; fixture mis-tagged"
def _syllogism_rule_missing_premise(
premises: list[dict], candidate: dict
) -> tuple[bool, str]:
"""Candidate references evidence not present in premises. Must FAIL."""
available_flat: set[str] = set()
for p in premises:
for i in p.get("evidence_ids", []):
available_flat.add(i)
used = set(candidate.get("uses", []))
missing = used - available_flat
if missing:
return False, f"candidate references missing evidence ids: {sorted(missing)}"
return True, "all evidence present"
def _syllogism_rule_chain_3(
premises: list[dict], candidate: dict
) -> tuple[bool, str]:
"""Three-step transitive chain: A→B, B→C, C→D ⇒ A→D.
Order-agnostic over premise permutations.
"""
from itertools import permutations
if len(premises) != 3:
return False, "chain_3 needs exactly 3 premises"
parsed = [_parse_categorical(p["claim"]) for p in premises]
c = _parse_categorical(candidate["claim"])
if any(x is None for x in parsed) or c is None:
return False, "claim parse failure"
for ordering in permutations(parsed):
a, b, d = ordering
if a[1] == b[0] and b[1] == d[0]:
if c[0] == a[0] and c[1] == d[1]:
return True, ""
return False, "no valid chain ordering matches candidate endpoints"
_SYLLOGISM_RULES = {
"categorical_transitivity": _syllogism_rule_categorical_transitivity,
"invalid_converse": _syllogism_rule_invalid_converse,
"missing_premise": _syllogism_rule_missing_premise,
"chain_3": _syllogism_rule_chain_3,
}
def run_syllogism(fixtures_path: Path) -> BatteryResult:
"""Stepwise entailment validity. Deterministic, no LLM.
Each task names a ``rule`` from :data:`_SYLLOGISM_RULES`. The
runner applies the rule's kernel; the task passes if the kernel's
verdict matches ``expected`` (``"pass"`` ⇒ valid step;
``"fail"`` ⇒ invalid step correctly identified).
"""
per_task: list[TaskResult] = []
pass_count = 0
fail_count = 0
for task in iter_tasks(fixtures_path):
task_id = task["id"]
ok, reason = _carrier_check(task)
if not ok:
per_task.append(
TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
)
fail_count += 1
continue
rule = task.get("rule")
kernel = _SYLLOGISM_RULES.get(rule)
if kernel is None:
per_task.append(
TaskResult(
task_id=task_id,
passed=False,
detail={"reason": f"unknown_rule: {rule!r}"},
)
)
fail_count += 1
continue
expected = task.get("expected", "pass")
valid, why = kernel(task["premises"], task["candidate_step"])
observed = "pass" if valid else "fail"
passed = observed == expected
detail = {
"rule": rule,
"expected": expected,
"observed": observed,
"kernel_reason": why,
}
per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))
if passed:
pass_count += 1
else:
fail_count += 1
total = pass_count + fail_count
rate = pass_count / total if total else 0.0
meta = fixture_meta(fixtures_path)
return BatteryResult(
battery="5s",
sub_battery=meta.get("sub_battery", "syllogism"),
fixture_path=str(fixtures_path),
fixture_digest=fixture_digest(fixtures_path),
pass_count=pass_count,
fail_count=fail_count,
metrics={"step_validity_rate": rate},
per_task=per_task,
runtime_digest=_runtime_digest(),
)
# ---------------------------------------------------------------------
# Phase 1b: Synthesis (ticket #000023)
# ---------------------------------------------------------------------
def _synthesis_check(task: dict) -> tuple[bool, str]:
"""Validate that the expected_derivation is supported by the fact_set.
Each derivation step lists ``uses`` (fact IDs from the fact_set).
A derivation is valid iff every ``uses`` ID exists in the
fact_set AND each derivation claim's text appears as a substring
of at least one cited fact's text (lower-cased, whitespace-
normalized). This is structural support, not semantic entailment;
enough to drive Phase 1b without an LLM judge.
"""
facts = {f["id"]: f["text"] for f in task.get("fact_set", [])}
derivation = task.get("expected_derivation", [])
for step in derivation:
uses = step.get("uses", [])
for fid in uses:
if fid not in facts:
return False, f"derivation step uses unknown fact id {fid!r}"
claim_text = _normalize(step["claim"])
any_support = False
for fid in uses:
fact_text = _normalize(facts[fid])
if claim_text in fact_text or _token_subset(claim_text, fact_text):
any_support = True
break
if not any_support:
return False, f"step {step['claim']!r} not supported by cited facts"
return True, "all steps supported"
_STOPWORDS = frozenset({
"a", "an", "the", "is", "are", "was", "were", "be", "been", "being",
"of", "on", "in", "to", "for", "with", "by", "and", "or", "as",
"that", "which", "who", "whom", "whose", "this", "these", "those",
"has", "have", "had", "do", "does", "did", "from", "at", "into",
"over", "under", "above", "below", "per",
})
def _normalize(s: str) -> str:
return " ".join(s.lower().strip().split())
def _content_tokens(s: str) -> set[str]:
"""Lower-cased tokens with stopwords removed, punctuation stripped."""
out = set()
for tok in s.lower().split():
clean = tok.strip(".,;:!?\"'()[]{}")
if clean and clean not in _STOPWORDS:
out.add(clean)
return out
def _token_subset(a: str, b: str) -> bool:
"""True if every content token in a appears in b.
Stricter than the previous overlap-percentage check — catches
single-token entity swaps (e.g., 'K2' vs 'Mount Everest').
"""
a_tokens = _content_tokens(a)
b_tokens = _content_tokens(b)
if not a_tokens:
return False
return a_tokens.issubset(b_tokens)
def run_synthesis(fixtures_path: Path) -> BatteryResult:
"""Goal + fact_set + expected_derivation → pass/fail.
Pure structural check, no LLM, no retrieval. Phase 1b stays
deterministic.
"""
per_task: list[TaskResult] = []
pass_count = 0
fail_count = 0
for task in iter_tasks(fixtures_path):
task_id = task["id"]
ok, reason = _carrier_check(task)
if not ok:
per_task.append(
TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
)
fail_count += 1
continue
expected = task.get("expected", "pass")
valid, why = _synthesis_check(task)
observed = "pass" if valid else "fail"
passed = observed == expected
detail = {
"expected": expected,
"observed": observed,
"reason": why,
}
per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))
if passed:
pass_count += 1
else:
fail_count += 1
total = pass_count + fail_count
rate = pass_count / total if total else 0.0
meta = fixture_meta(fixtures_path)
return BatteryResult(
battery="5s",
sub_battery=meta.get("sub_battery", "synthesis"),
fixture_path=str(fixtures_path),
fixture_digest=fixture_digest(fixtures_path),
pass_count=pass_count,
fail_count=fail_count,
metrics={"derivation_pass_rate": rate},
per_task=per_task,
runtime_digest=_runtime_digest(),
)
# ---------------------------------------------------------------------
# Phase 1b: Semiotics (ticket #000023)
# ---------------------------------------------------------------------
def _semiotic_swap_check(task: dict) -> tuple[bool, str]:
"""Apply ``swap.from`` → ``swap.to`` in source_text; canonicalize
both via the named π*; compare.
For a synonym swap with ``expected_invariant=True``: the canonical
forms should differ ONLY in the swapped token (the rest is
identical). For a non-equivalent swap with
``expected_invariant=False``: canonical forms should differ
semantically (we approximate this as: the swapped strings remain
different after canonicalization, AND the invariant_fields would
not all match — Phase 1b only checks the surface canonicalization).
"""
from arborist.pi_star import get
pi_star = get(task["pi_star_ref"])
src = task["source_text"].encode("utf-8")
swap = task["swap"]
swapped = task["source_text"].replace(swap["from"], swap["to"]).encode("utf-8")
canon_src = pi_star.canonicalize(src)
canon_swapped = pi_star.canonicalize(swapped)
if task["expected_invariant"]:
# For Phase 1b: a synonym swap should yield a canonical-form
# change limited to the swap. We assert the two forms differ
# by exactly the swap (the canonicalizer should preserve the
# rest of the prose). The structural check is: every word in
# canon_src that isn't in `swap.from` survives in canon_swapped.
return _swap_preserves_invariant(
canon_src, canon_swapped, swap["from"], swap["to"]
)
# expected_invariant=False: a non-synonym swap should produce
# canonical forms that differ in ways NOT explainable solely by
# the swap. The simplest deterministic check: swap was applied
# and canonical forms are not byte-equal (they wouldn't be — the
# swap text is different). We assert the swap WAS detected as
# non-trivial — i.e., the canonical forms DIFFER even after
# accounting for the swap text. A correct non-equivalent fixture
# has distinct canonical "essence" beyond the surface swap.
return _swap_changes_essence(
canon_src, canon_swapped, swap["from"], swap["to"]
)
def _swap_preserves_invariant(
canon_src: bytes, canon_swapped: bytes, frm: str, to: str
) -> tuple[bool, str]:
"""Does the swap produce only the expected substitution?"""
# Re-substitute in canonical form: replacing `to` with `frm` in
# canon_swapped should reproduce canon_src.
rebuilt = canon_swapped.decode("utf-8").replace(to, frm).encode("utf-8")
if rebuilt == canon_src:
return True, "synonym swap preserved invariant"
return False, "swap changed more than the substituted token"
def _swap_changes_essence(
canon_src: bytes, canon_swapped: bytes, frm: str, to: str
) -> tuple[bool, str]:
"""Does the swap change more than just the surface token?
Re-substituting `to` → `frm` should NOT yield canon_src for a
non-synonym swap (the meaning shift surfaces somewhere). Phase
1b uses the cheap surface heuristic: forms differ even after
re-substitution.
"""
rebuilt = canon_swapped.decode("utf-8").replace(to, frm).encode("utf-8")
if rebuilt != canon_src:
return True, "non-equivalent swap detected"
return False, "swap was actually a synonym; fixture mis-tagged"
def run_semiotics(fixtures_path: Path) -> BatteryResult:
"""Invariance under controlled sign substitution.
v1 fixtures use ``concept_relations.relation_kind='synonym'`` for
positives and explicit non-synonyms for negatives. Hidden-channel
/ steganography fixtures are reserved for Phase 2 — defensive
detection only, never generation.
"""
per_task: list[TaskResult] = []
pass_count = 0
fail_count = 0
for task in iter_tasks(fixtures_path):
task_id = task["id"]
ok, reason = _carrier_check(task)
if not ok:
per_task.append(
TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
)
fail_count += 1
continue
try:
valid, why = _semiotic_swap_check(task)
expected = task["expected_invariant"]
observed = valid
passed = observed == expected
except Exception as exc: # noqa: BLE001
valid = False
why = f"{type(exc).__name__}: {exc}"
passed = False
detail = {
"expected_invariant": task["expected_invariant"],
"observed_invariant": valid,
"reason": why,
"relation_kind": task.get("swap", {}).get("relation_kind"),
}
per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))
if passed:
pass_count += 1
else:
fail_count += 1
total = pass_count + fail_count
rate = pass_count / total if total else 0.0
meta = fixture_meta(fixtures_path)
return BatteryResult(
battery="5s",
sub_battery=meta.get("sub_battery", "semiotics"),
fixture_path=str(fixtures_path),
fixture_digest=fixture_digest(fixtures_path),
pass_count=pass_count,
fail_count=fail_count,
metrics={"invariance_under_swap": rate},
per_task=per_task,
runtime_digest=_runtime_digest(),
)
SUB_BATTERIES = {
"syntax": run_syntax,
"semantics": run_semantics,
"syllogism": run_syllogism,
"synthesis": run_synthesis,
"semiotics": run_semiotics,
}