"""Finite-entity grounding — compile a typed finite-entity problem to propositional atoms (Phase 2 of the universal-structure plan; the deductive-logic runway doc's PR-1). This is the first **comprehension compiler**: it lowers a structured, finite-entity problem (finite named entities, unary predicates, single-variable universal rules) into the propositional regime the ADR-0206 entailment operator already decides — exactly where `wrong == 0` is structural. A grounded finite-entity problem (each ``predicate(entity)`` pair → one atom) IS propositional and in scope; anything outside the narrow v1 grammar refuses with a typed reason rather than guessing. Lowering law (deterministic): predicate(entity) -> ``predicate_slug__entity_slug`` (atom) negative literal -> ``~atom`` rule body (∧ of literals) -> conjunction of lowered body literals universal rule ∀x. B(x)->H(x)-> for each entity e: ``lower(B[e]) -> lower(H[e])`` query -> the lowered query literal The lowered ``(premises, query)`` is decided by **two independent procedures** (the ROBDD engine and the truth-table oracle); a case counts only when both agree with the gold (INV-25). v1 keeps the grammar intentionally narrow — that narrowness is the firewall, not a weakness. Sealed: no ``chat`` import, no serving path. Deterministic. """ from __future__ import annotations import re from dataclasses import dataclass from typing import Any, Final # Closed refusal vocabulary for the grounding layer — distinct from the entailment # operator's reasons so a malformed case never leaks into an ambiguous failure. UNSUPPORTED_PREDICATE_ARITY: Final[str] = "unsupported_predicate_arity" UNSUPPORTED_QUANTIFIER: Final[str] = "unsupported_quantifier" UNSAFE_SYMBOL: Final[str] = "unsafe_symbol" UNKNOWN_ENTITY: Final[str] = "unknown_entity" UNKNOWN_VARIABLE: Final[str] = "unknown_variable" MALFORMED_CASE: Final[str] = "malformed_case" EMPTY_CASE: Final[str] = "empty_case" GROUNDING_REASONS: Final[frozenset[str]] = frozenset({ UNSUPPORTED_PREDICATE_ARITY, UNSUPPORTED_QUANTIFIER, UNSAFE_SYMBOL, UNKNOWN_ENTITY, UNKNOWN_VARIABLE, MALFORMED_CASE, EMPTY_CASE, }) _ATOM_SEPARATOR: Final[str] = "__" # A slug component: lowercase ASCII letters/digits/single underscores, no leading # digit, no empty, no double-underscore (would collide with the separator). _SLUG_RE: Final[re.Pattern[str]] = re.compile(r"[a-z][a-z0-9]*(?:_[a-z0-9]+)*") class GroundingError(ValueError): """A finite-entity case outside the v1 grammar. Carries a typed reason.""" def __init__(self, reason: str, detail: str = "") -> None: if reason not in GROUNDING_REASONS: raise ValueError(f"unknown grounding reason: {reason!r}") self.reason = reason super().__init__(f"{reason}: {detail}" if detail else reason) @dataclass(frozen=True, slots=True) class GroundedProblem: """The lowered finite-entity problem: propositional premises + query.""" premises: tuple[str, ...] query: str def slug(token: Any) -> str: """Deterministic, collision-safe slug. Rejects (never silently repairs) anything outside ``[a-z][a-z0-9]*(_[a-z0-9]+)*`` — including the separator, leading digits, empties, and double underscores.""" if not isinstance(token, str): raise GroundingError(UNSAFE_SYMBOL, f"non-string symbol {token!r}") s = token.strip().lower() if not s: raise GroundingError(UNSAFE_SYMBOL, "empty symbol") if _ATOM_SEPARATOR in s: raise GroundingError(UNSAFE_SYMBOL, f"symbol {token!r} contains separator '{_ATOM_SEPARATOR}'") if not _SLUG_RE.fullmatch(s): raise GroundingError(UNSAFE_SYMBOL, f"symbol {token!r} is not a safe slug") return s def atom(predicate: Any, entity: Any) -> str: """``predicate(entity)`` → the canonical propositional atom.""" return f"{slug(predicate)}{_ATOM_SEPARATOR}{slug(entity)}" def _literal(predicate: Any, entity: Any, polarity: Any) -> str: if not isinstance(polarity, bool): raise GroundingError(MALFORMED_CASE, f"polarity must be a bool, got {polarity!r}") a = atom(predicate, entity) return a if polarity else f"~{a}" def _require(condition: bool, reason: str, detail: str = "") -> None: if not condition: raise GroundingError(reason, detail) def _check_unary(obj: dict[str, Any], *, allow_var: bool) -> None: """Reject any non-unary / relational / functional shape (v1 is unary only).""" keys = set(obj) # A binary relation / function would carry extra argument keys. extra = keys - {"predicate", "entity", "var", "polarity"} _require(not extra, UNSUPPORTED_PREDICATE_ARITY, f"unexpected keys {sorted(extra)}") has_entity = "entity" in obj has_var = "var" in obj if allow_var: _require(has_entity ^ has_var, MALFORMED_CASE, "literal needs exactly one of entity/var") else: _require(has_entity and not has_var, UNKNOWN_VARIABLE, "free variable outside a rule") def lower_case(case: dict[str, Any]) -> GroundedProblem: """Lower a finite-entity case to ``(premises, query)``. Refuse-first. Schema (v1): {"entities": [str, ...], "facts": [{"predicate": str, "entity": str, "polarity": bool}, ...], "rules": [{"if": [, ...], "then": }, ...], "query": {"predicate": str, "entity": str, "polarity": bool}} """ if not isinstance(case, dict) or not case: raise GroundingError(EMPTY_CASE, "case is empty or not a mapping") entities = case.get("entities") if not (isinstance(entities, list) and entities): raise GroundingError(EMPTY_CASE, "no entities") entity_set = {slug(e) for e in entities} # also validates each entity slug _require(len(entity_set) == len(entities), MALFORMED_CASE, "duplicate entities after slugging") if "query" not in case: raise GroundingError(MALFORMED_CASE, "no query") premises: list[str] = [] facts = case.get("facts", []) or [] rules = case.get("rules", []) or [] _require(isinstance(facts, list), MALFORMED_CASE, "facts must be a list") _require(isinstance(rules, list), MALFORMED_CASE, "rules must be a list") # Facts — ground unary literals over named entities. for fact in facts: _require(isinstance(fact, dict), MALFORMED_CASE, f"fact is not a mapping: {fact!r}") _check_unary(fact, allow_var=False) _require(slug(fact["entity"]) in entity_set, UNKNOWN_ENTITY, str(fact.get("entity"))) premises.append(_literal(fact["predicate"], fact["entity"], fact.get("polarity", True))) # Rules — single-variable universal rules, grounded by explicit entity expansion. for rule in rules: _require(isinstance(rule, dict), MALFORMED_CASE, f"rule is not a mapping: {rule!r}") _require("quantifier" not in rule and "exists" not in rule, UNSUPPORTED_QUANTIFIER, "only implicit single-var universal rules in v1") body = rule.get("if") head = rule.get("then") if not (isinstance(body, list) and body): raise GroundingError(MALFORMED_CASE, "rule body must be a non-empty list") if not isinstance(head, dict): raise GroundingError(MALFORMED_CASE, "rule head must be a literal") var = _rule_variable(body, head) for ent in entities: body_atoms = [_literal(lit["predicate"], ent, lit.get("polarity", True)) for lit in body] head_atom = _literal(head["predicate"], ent, head.get("polarity", True)) body_conj = " & ".join(f"({b})" for b in body_atoms) premises.append(f"({body_conj}) -> ({head_atom})") _ = var # validated; grounding is by explicit entity expansion, not by name query = case["query"] _require(isinstance(query, dict), MALFORMED_CASE, "query must be a literal") _check_unary(query, allow_var=False) _require(slug(query["entity"]) in entity_set, UNKNOWN_ENTITY, str(query.get("entity"))) query_lit = _literal(query["predicate"], query["entity"], query.get("polarity", True)) return GroundedProblem(premises=tuple(premises), query=query_lit) def _rule_variable(body: list[dict[str, Any]], head: dict[str, Any]) -> str: """Confirm the rule is single-variable: every literal uses one shared var, no named entities (those would make it not universal). Returns the var name.""" seen: set[str] = set() for lit in (*body, head): _require(isinstance(lit, dict), MALFORMED_CASE, f"rule literal not a mapping: {lit!r}") _check_unary(lit, allow_var=True) _require("var" in lit, MALFORMED_CASE, "rule literals must use a variable, not a named entity") v = lit["var"] _require(isinstance(v, str) and bool(v.strip()), MALFORMED_CASE, "empty variable") seen.add(v) _require(len(seen) == 1, UNSUPPORTED_QUANTIFIER, f"v1 allows one variable per rule, saw {sorted(seen)}") return seen.pop()