diff --git a/reasoning/clauses.py b/reasoning/clauses.py index d018d8d..8fea15f 100644 --- a/reasoning/clauses.py +++ b/reasoning/clauses.py @@ -8,7 +8,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Hashable, Iterable, Iterator, TypeAlias, cast +from typing import Hashable, Iterable, Iterator, Sequence, TypeAlias, cast Literal: TypeAlias = int | bool @@ -96,6 +96,78 @@ def format_clauses(self) -> str: return " & ".join(self.format_clause(clause) for clause in self.data) or "True" +class SharedClauses: + """A pre-translated, immutable block of integer clauses. + + Some clause theories are *fixed*: the same clauses are needed in every + query, with only the atom bindings varying. Such a theory is translated + into integer clauses once and replayed into each query's database, so + per-query work is one ``ClauseDB.literal`` call per bound atom plus a + clause replay instead of a full re-emission. Sharing stops at + construction time: every receiving database gets its own copies of the + clauses and stays independently mutable, and no solver ever observes + shared state between queries. + + The block stores clauses over *template positions* — positive integers + from 1 to :attr:`width`. :meth:`replay` binds each position to a + variable id of the receiving database. Auxiliary variables of the + compiled theory must occupy the topmost positions of the block; callers + that compile theories containing none (the known-facts template) use + the default ``auxiliary_count=0``. + """ + + __slots__ = ("clauses", "width", "auxiliary_count") + + def __init__(self, clauses: Iterable[Iterable[int]], width: int, + auxiliary_count: int = 0) -> None: + self.clauses: tuple[tuple[int, ...], ...] = tuple( + tuple(clause) for clause in clauses) + self.width = width + self.auxiliary_count = auxiliary_count + if any(abs(literal) > width + for clause in self.clauses for literal in clause): + raise ValueError("clause literal exceeds the block width") + if not 0 <= auxiliary_count <= width: + raise ValueError("auxiliary count outside the block width") + + def replay(self, db: ClauseDB, binding: Sequence[int]) -> None: + """Add the block to *db*, binding position *i* to ``binding[i]``. + + The binding must map template positions to distinct positive + variable ids of *db*; known-facts bindings map the template's + predicate positions through ``ClauseDB.literal``, which does. When + the binding is consecutive from *db*'s next free variable — every + atom freshly minted — the stored clauses are appended verbatim; + otherwise they are re-based through the binding. Re-based clauses + skip ``add_clause`` normalization legitimately: template clauses + contain no duplicate or tautological literals by construction, and + distinct positions cannot collapse onto one id or onto negations of + one another (ids are positive and ``literal`` never reuses an id for + a different atom), so every re-based clause is already a normalized + duplicate-free integer set. + """ + if len(binding) != self.width: + raise ValueError("binding does not cover the block width") + base = db._next_variable + for position, variable in enumerate(binding): + if variable != base + position: + self._rebase(db, binding) + return + db.data.extend(map(set, self.clauses)) + db._next_variable = base + self.width + + def _rebase(self, db: ClauseDB, binding: Sequence[int]) -> None: + data = db.data + for clause in self.clauses: + data.append({ + binding[literal - 1] if literal > 0 else -binding[-literal - 1] + for literal in clause}) + # Keep the next free id above every bound position so a later + # ``literal`` call cannot mint an id a binding already names. + if db._next_variable <= max(binding): + db._next_variable = max(binding) + 1 + + @dataclass(frozen=True) class Formula: """An internal Boolean connective. Anything else is an opaque atom.""" @@ -362,6 +434,33 @@ def assert_formula(formula: object, db: ClauseDB) -> None: if antecedent is not None and consequent is not None: db.add_clause((*(_negate(literal) for literal in antecedent), *consequent)) return + # A -> (Or_j(And_j) -> C) flattens to (A & Or_j(And_j)) -> C, one + # clause per disjunct. Distribution is confined to this negative + # polarity: the same shapes in consequent position would expand + # exponentially and must keep falling through to Tseitin compilation. + # The shape checks below never allocate variables, so every shape + # that is not fast-pathed reaches Tseitin with the database exactly + # as before. + if antecedent is not None and isinstance(formula.args[1], Formula) \ + and formula.args[1].op == "implies": + nested = formula.args[1] + disjuncts = _cnf_disjunct_formulas(nested.args[0]) + if disjuncts is not None and _flat_disjunction(nested.args[1]): + nested_consequent = _disjunction_literals(nested.args[1], db) + for disjunct in disjuncts: + db.add_clause((*(_negate(literal) for literal in antecedent), + *(_negate(_as_literal(term, db)) + for term in disjunct), + *nested_consequent)) + return + # (Or_j(And_j)) -> C is one clause per disjunct, no auxiliaries. + disjuncts = _cnf_disjunct_formulas(formula.args[0]) + if disjuncts is not None and consequent is not None: + for disjunct in disjuncts: + db.add_clause((*(_negate(_as_literal(term, db)) + for term in disjunct), + *consequent)) + return if isinstance(formula, Formula) and formula.op == "equivalent": literals = [_as_literal(arg, db) for arg in formula.args] if all(arg is not None for arg in literals): @@ -394,8 +493,43 @@ def _disjunction_literals(formula: object, db: ClauseDB) -> list[Literal] | None return None +def _is_literal_formula(formula: object) -> bool: + """Whether ``_as_literal`` would succeed for *formula*, without + allocating any variable.""" + if formula is True or formula is False: + return True + if isinstance(formula, Formula): + return formula.op == "not" and _is_literal_formula(formula.args[0]) + return True + + +def _flat_disjunction(formula: object) -> bool: + """Whether *formula* is a literal or a flat disjunction of them.""" + terms = formula.args if isinstance(formula, Formula) and formula.op == "or" else (formula,) + return all(_is_literal_formula(term) for term in terms) + + +def _cnf_disjunct_formulas(formula: object) -> tuple[tuple[object, ...], ...] | None: + """Return the disjuncts of an Or-of-Ands of literal-shaped terms. + + Each disjunct is a conjunction of literal-shaped terms, held as the raw + term formulas so callers materialize variable ids only when they commit + to emitting the distributed clauses. ``None`` when *formula* does not + have this shape. + """ + if not (isinstance(formula, Formula) and formula.op == "or"): + return None + disjuncts = [] + for arg in formula.args: + terms = arg.args if isinstance(arg, Formula) and arg.op == "and" else (arg,) + if not all(_is_literal_formula(term) for term in terms): + return None + disjuncts.append(terms) + return tuple(disjuncts) + + __all__ = [ "AND", "OR", "NOT", "IMPLIES", "EQUIVALENT", "XOR", "ITE", - "ClauseDB", "Formula", "Literal", "assert_formula", "compile_formula", - "iter_atoms", + "ClauseDB", "Formula", "Literal", "SharedClauses", + "assert_formula", "compile_formula", "iter_atoms", ] diff --git a/reasoning/satask.py b/reasoning/satask.py index d913b45..367e162 100644 --- a/reasoning/satask.py +++ b/reasoning/satask.py @@ -1,18 +1,74 @@ """SymPy-facing entry points for the independent SAT reasoning core.""" from __future__ import annotations -from typing import Iterable +from functools import lru_cache +from typing import Any, Iterable, cast -from .clauses import ClauseDB, assert_formula, compile_formula, iter_atoms +from sympy.core.kind import NumberKind, UndefinedKind +from sympy.matrices.kind import MatrixKind + +from .clauses import ( + ClauseDB, Formula, assert_formula, compile_formula, iter_atoms, +) from .discovery import discover_facts, relevant_subjects from .engine import ReasoningEngine +from .knownfacts import template as known_facts_template from .lra_adapter import build_lra_theory +from .predicates import AppliedPredicate as LocalAppliedPredicate +from .sathandlers import class_fact_registry from .sympy_adapter import ( NormalizedFormula, SympyAdapter, normalize, to_formula, ) from .sympy_types import SymPyExpr +def _memo_argument(value: SymPyExpr | bool) -> SymPyExpr: + """Return a cache key for a public input without bool/int collisions. + + Python bools are ints (``True == 1`` and ``False == 0`` compare equal and + hash equal), so raw values would let a cached ``satask(True, ...)`` answer + a later ``satask(1, ...)``, which raises TypeError. Keying on the SymPy + counterparts (as ``normalize`` interprets them) separates the values; the + keys stay hashable and every other value is unaffected. SymPy objects + themselves never equal Python numbers they are not equal to (SymPy Booleans + are ``Basic``, so ``hash(S.true) != hash(1)``), and SymPy Booleans mapped + through :func:`normalize` are exactly these constants, while raw SymPy + expressions still key on identity-by-value hashing. + """ + if value is True: + from sympy import true + return true + if value is False: + from sympy import false + return false + return value + + +def _memo_iteration(iterations: object) -> tuple[Any, object] | None: + """Return a cache key for the iteration limit without int collisions. + + The result must be unhashable-safe: an unknown type raises TypeError at + key construction, and :func:`satask` then recomputes uncached. Tagging + with the runtime type separates values that compare and hash equal on + the value alone, such as ``False``, ``0.0``, or ``numpy.int64(0)`` + against ``0``: the former pairs raise ValueError while ``0`` is valid. + """ + if iterations is None: + return None + return _IterationKey((type(iterations), iterations)) + + +class _IterationKey(tuple): + """Marker subclass wrapping an iteration limit inside the memo key.""" + + +def _unwrap(iterations: object) -> object | None: + # Invert _memo_iteration by marker type, never by value inspection. + if type(iterations) is _IterationKey: + return iterations[1] + return iterations + + def _iteration_limit(iterations: object) -> int | None: # Keep accepting the former oo default at the public boundary. from sympy import oo @@ -48,9 +104,69 @@ def satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool = True, value raises TypeError, and SymPy ``Q`` applications become local :mod:`reasoning.predicates` applications whose arguments are the original SymPy expressions. + + Results are memoized on the full argument tuple in a bounded LRU cache + (``_MEMO_MAXSIZE`` entries), so repeated identical queries cost a dict + lookup. Exceptions are never cached: inconsistent assumptions raise + again on every call. Trivial queries — the proposition is an assumed + literal, or follows from one through a single known-facts clause, or + there is nothing to assume — are answered directly from the known-facts + template without building the clause database; the fast path only fires + when the full pipeline provably produces the same value, which excludes + ``use_lra_theory`` (the theory can decide literals the clauses cannot). + Like the rest of the package, the cache assumes single-threaded use. """ + try: + return _satask_memoized(_memo_argument(proposition), + _memo_argument(assumptions), use_known_facts, + _memo_iteration(iterations), early_return, + use_lra_theory) + except TypeError: + # The arguments are unhashable (lru_cache rejects them) or ``normalize`` + # rejected a non-expression input. Recompute uncached so the original + # result or error is produced exactly as before. + return _satask(proposition, assumptions, use_known_facts, iterations, + early_return, use_lra_theory) + + +# Bound the memoization cache; each entry pins its key expressions. +_MEMO_MAXSIZE = 4096 + +# Sentinel distinguishing "the fast path cannot answer" from a None answer. +_NO_FAST_ANSWER: Any = object() + + +@lru_cache(maxsize=_MEMO_MAXSIZE) +def _satask_memoized(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool, + use_known_facts: bool, iterations: object, + early_return: bool, use_lra_theory: bool) -> bool | None: + return _satask(proposition, assumptions, use_known_facts, + _unwrap(iterations), early_return, use_lra_theory) + + +def _satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool, + use_known_facts: bool, iterations: object, + early_return: bool, use_lra_theory: bool) -> bool | None: prop_formula: NormalizedFormula = normalize(proposition) assump_formula: NormalizedFormula = normalize(assumptions) + # Validate here so an invalid limit raises even when the fast path answers. + _iteration_limit(iterations) + # With the LRA theory enabled, the clause database alone no longer + # determines the answer: the theory can entail or refute a degenerate + # relation literal (Q.gt(x, x), ~Q.eq(x, x), ...) and can even make a + # single literal assumption inconsistent. No fast-path answer is sound. + if not use_lra_theory: + answer = _fast_path_answer(prop_formula, assump_formula, use_known_facts) + if answer is not _NO_FAST_ANSWER: + return cast("bool | None", answer) + return _satask_pipeline(prop_formula, assump_formula, use_known_facts, + iterations, early_return, use_lra_theory) + + +def _satask_pipeline(prop_formula: NormalizedFormula, + assump_formula: NormalizedFormula, + use_known_facts: bool, iterations: object, + early_return: bool, use_lra_theory: bool) -> bool | None: db = get_all_relevant_facts(prop_formula, assump_formula, use_known_facts, iterations) assert_formula(assump_formula, db) query = compile_formula(prop_formula, db) @@ -67,6 +183,115 @@ def satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool = True, return engine.ask(query, early_return=early_return) +def _as_literal(formula: NormalizedFormula + ) -> tuple[LocalAppliedPredicate, bool] | None: + """Return the atom and polarity when *formula* is a single literal.""" + if isinstance(formula, LocalAppliedPredicate): + return formula, True + if isinstance(formula, Formula) and formula.op == "not": + argument = formula.args[0] + if isinstance(argument, LocalAppliedPredicate): + return argument, False + return None + + +def _fact_free(arguments: tuple[object, ...]) -> bool: + """Return whether no registered handler produces facts for the arguments.""" + for argument in arguments: + singles, multiples = class_fact_registry[type(argument)] + if singles or multiples: + return False + return True + + +def _template_selection(arguments: tuple[object, ...]) -> tuple[bool, bool]: + """Mirror the fact-family selection of ``SympyAdapter.add_known_facts``.""" + numbers = any(cast("Any", argument).kind in (NumberKind, UndefinedKind) + for argument in arguments) + matrices = any(cast("Any", argument).kind == MatrixKind(NumberKind) + for argument in arguments) + return numbers, matrices + + +@lru_cache(maxsize=3) +def _implication_edges(numbers: bool, matrices: bool + ) -> tuple[dict[str, int], dict[int, frozenset[int]]]: + """One-step implications of the known-facts template. + + Every binary clause ``a | b`` forces ``b`` under ``~a`` and ``a`` under + ``~b``; the result maps each signed predicate index to the signed indices + unit propagation derives from it in a single step. Only binary clauses + are used, so these are exactly the single-hop known-facts consequences. + """ + names, clauses = known_facts_template(numbers, matrices) + index = {name: position + 1 for position, name in enumerate(names)} + edges: dict[int, set[int]] = {} + for clause in clauses: + if len(clause) == 2: + left, right = clause + edges.setdefault(-left, set()).add(right) + edges.setdefault(-right, set()).add(left) + return index, {key: frozenset(values) for key, values in edges.items()} + + +def _fast_path_answer(prop_formula: NormalizedFormula, + assump_formula: NormalizedFormula, + use_known_facts: bool) -> bool | None | Any: + """Answer trivial literal queries without building the clause database. + + Return ``_NO_FAST_ANSWER`` unless the full pipeline provably produces the + same value. The fast path requires the proposition to be a single + literal whose arguments have no registered expression facts, and either: + + - no assumptions at all: the database then holds only known-fact + templates, which contain no unit clauses, so nothing about the query + is forced and the engine answers None; + - one assumption literal over the same arguments: consistency then + reduces to the template staying satisfiable under one unit, which + holds for every vendored predicate (verified by the test suite), and + the answer is decided by the same atom, or by one binary template + clause when ``use_known_facts`` holds. + + Anything else — conjunctions, differing arguments, fact-bearing subjects + such as numbers (whose literals can be contradictory) — is left to the + full pipeline. + """ + prop_literal = _as_literal(prop_formula) + if prop_literal is None: + return _NO_FAST_ANSWER + prop_atom, prop_sign = prop_literal + if not _fact_free(prop_atom.arguments): + return _NO_FAST_ANSWER + + if assump_formula is True: + return None + + assump_literal = _as_literal(assump_formula) + if assump_literal is None: + return _NO_FAST_ANSWER + assump_atom, assump_sign = assump_literal + if assump_atom.arguments != prop_atom.arguments: + return _NO_FAST_ANSWER + + if prop_atom == assump_atom: + return prop_sign == assump_sign + + if not use_known_facts: + return _NO_FAST_ANSWER + index, edges = _implication_edges(*_template_selection(prop_atom.arguments)) + source = index.get(assump_atom.name) + target = index.get(prop_atom.name) + if source is None or target is None: + return _NO_FAST_ANSWER + source_literal = source if assump_sign else -source + target_literal = target if prop_sign else -target + if target_literal in edges.get(source_literal, ()): + return True + if -target_literal in edges.get(source_literal, ()): + return False + return _NO_FAST_ANSWER + + def extract_predargs(proposition: object, assumptions: object = None) -> set[object]: """Find subjects connected to the proposition through assumption symbols.""" diff --git a/reasoning/sympy_adapter.py b/reasoning/sympy_adapter.py index 66abb2c..09a8f4d 100644 --- a/reasoning/sympy_adapter.py +++ b/reasoning/sympy_adapter.py @@ -12,8 +12,8 @@ from sympy.matrices.kind import MatrixKind from .clauses import ( - ClauseDB, Formula, AND, OR, NOT, IMPLIES, EQUIVALENT, XOR, ITE as IF, - iter_atoms, + ClauseDB, Formula, SharedClauses, AND, OR, NOT, IMPLIES, EQUIVALENT, + XOR, ITE as IF, iter_atoms, ) from .knownfacts import template as known_facts_template from .predicates import AppliedPredicate as LocalAppliedPredicate @@ -113,6 +113,70 @@ def _known_template(numbers: bool, matrices: bool) -> tuple[ return predicates, clauses +@lru_cache(maxsize=4) +def _shared_known_facts(numbers: bool, matrices: bool) -> SharedClauses: + """Compile the known-facts theory into a shared clause block. + + The known-facts template is a fixed finite theory: the same clauses are + needed in every query, one per subject with only the atom bindings + varying. Compiling it once into a :class:`~reasoning.clauses. + SharedClauses` block removes the per-call re-translation; each query + still gets its own database and replays its own copies of the clauses + into it, so nothing is shared with the solver at solve time. + """ + predicates, clauses = _known_template(numbers, matrices) + return SharedClauses(clauses, width=len(predicates)) + + +class _SubjectBindings: + """Intern the per-subject template bindings across queries. + + Each subject's binding is a tuple of variable ids, one per template + predicate, minted in the receiving database by ``db.literal``. Variable + ids are not stable across databases, so a cached binding cannot be + replayed into a fresh database directly — but its *layout* is: whenever + the minting would start at the same base id (every atom fresh, the only + shape ``satask`` produces), the ids repeat exactly, and the minting + loop collapses to two dict fills. The applied atoms are cached + alongside the binding so a hit fills ``encoding`` and ``symbols`` + without rebuilding or re-hashing them; a hit therefore leaves the + database exactly as a sequence of ``db.literal`` calls would have. + Bindings that would not repeat (atoms pre-existing in the database) are + simply not cached, keeping every other caller on the minting path. + """ + + __slots__ = ("_cache",) + + def __init__(self) -> None: + self._cache: dict[tuple[bool, bool, object], + tuple[tuple[int, ...], tuple[LocalAppliedPredicate, ...]]] = {} + + def bind(self, subject: SymPyExpr, predicates: list[LocalPredicate], + db: ClauseDB, selection: tuple[bool, bool]) -> tuple[int, ...]: + # The selection is part of the key: template selections have + # different widths, and a layout minted under one must never be + # replayed against another. + key = (selection[0], selection[1], subject) + cached = self._cache.get(key) + if cached is not None and db._next_variable == cached[0][0]: + binding, atoms = cached + encoding = db.encoding + symbols = db.symbols + for variable, atom in zip(binding, atoms): + encoding[atom] = variable + symbols[variable] = atom + db._next_variable += len(binding) + return binding + atoms = tuple(predicate(subject) for predicate in predicates) + binding = tuple(db.literal(atom) for atom in atoms) + if db._next_variable == binding[0] + len(binding): + self._cache[key] = (binding, atoms) + return binding + + +_subject_bindings = _SubjectBindings() + + class SympyAdapter: def relevance_keys(self, atom: Any) -> set[SymPyExpr]: if isinstance(atom, LocalAppliedPredicate): @@ -138,15 +202,9 @@ def facts_for(self, subject: SymPyExpr) -> Iterable[object]: def add_known_facts(self, subjects: Iterable[SymPyExpr], db: ClauseDB) -> None: numbers = any(expr.kind in (NumberKind, UndefinedKind) for expr in subjects) matrices = any(expr.kind == MatrixKind(NumberKind) for expr in subjects) - predicates, clauses = _known_template(numbers, matrices) + block = _shared_known_facts(numbers, matrices) + predicates, _ = _known_template(numbers, matrices) + selection = (numbers, matrices) for subject in subjects: - mapping: list[int | None] = [None] + [ - db.literal(predicate(subject)) for predicate in predicates - ] - for clause in clauses: - literals = [ - cast(int, mapping[lit]) if lit > 0 - else -cast(int, mapping[-lit]) - for lit in clause - ] - db.add_clause(literals) + block.replay(db, _subject_bindings.bind(subject, predicates, db, + selection)) diff --git a/reasoning/tests/test_persistent_facts.py b/reasoning/tests/test_persistent_facts.py new file mode 100644 index 0000000..664202f --- /dev/null +++ b/reasoning/tests/test_persistent_facts.py @@ -0,0 +1,305 @@ +"""Persistent known-facts compilation and per-subject binding reuse. + +``add_known_facts`` instantiates a fixed finite theory: the vendored +known-facts template is translated into integer clauses once (lazily, on +first use) and each query only binds its subjects into the shared block. +These tests lock that structure down: + +- the compiled block is built once and reused across many queries with + distinct symbols (instrumented construction counter), +- per-subject bindings are reused across fresh databases whenever the + variable layout repeats (the only layout ``satask`` produces), +- replay leaves every receiving database exactly as the previous + per-clause re-emission did — clauses, auxiliaries, and the atom + encoding are compared byte for byte against the oracle, +- answers are unchanged across a broad battery: the full-pipeline oracle + over every ordered predicate pair in both polarities, and the issue #3 + harness shapes. +""" +from __future__ import annotations + +from typing import Any + +import pytest +from sympy import Abs, Add, And, Not, Q, sqrt, Symbol +from sympy.core.relational import Eq +from sympy.matrices.expressions.matexpr import MatrixSymbol + +from reasoning import sympy_adapter +from reasoning.clauses import ClauseDB, SharedClauses, assert_formula, iter_atoms +from reasoning.discovery import discover_facts +from reasoning.knownfacts import MATRIX_PREDICATES, NUMBER_PREDICATES +from reasoning.satask import ( + _satask_pipeline, get_all_relevant_facts, normalize, satask, +) +from reasoning.sympy_adapter import ( + SympyAdapter, _shared_known_facts, _subject_bindings, _known_template, +) + +x, y, z = Symbol("x"), Symbol("y"), Symbol("z") +A = MatrixSymbol("A", 2, 2) + + +@pytest.fixture(autouse=True) +def isolate_caches(): + """Every test starts from cold template and binding caches.""" + _shared_known_facts.cache_clear() + _subject_bindings._cache.clear() + _satask_memoized_clear() + yield + _shared_known_facts.cache_clear() + _subject_bindings._cache.clear() + _satask_memoized_clear() + + +def _satask_memoized_clear() -> None: + from reasoning.satask import _satask_memoized + _satask_memoized.cache_clear() + + +@pytest.fixture() +def build_counter(monkeypatch: pytest.MonkeyPatch) -> list[int]: + """Count SharedClauses constructions across the test.""" + calls = [] + original = SharedClauses.__init__ + + def counting(self: SharedClauses, *args: Any, **kwargs: Any) -> None: + calls.append(1) + original(self, *args, **kwargs) + + monkeypatch.setattr(SharedClauses, "__init__", counting) + return calls + + +def oracle_db(prop: Any, assump: Any) -> tuple[ + list[frozenset[int]], list[int], list[tuple[Any, int]]]: + """Database contents as the per-clause re-emission produced them.""" + db = get_all_relevant_facts(normalize(prop), normalize(assump), True, None) + assert_formula(normalize(assump), db) + return (sorted(frozenset(clause) for clause in db.data), + sorted(db.auxiliaries), + sorted(((atom, variable) for atom, variable in db.encoding.items()), + key=lambda item: (item[0].name, repr(item[0].arguments)))) + + +def current_db(prop: Any, assump: Any) -> tuple[ + list[frozenset[int]], list[int], list[tuple[Any, int]]]: + """Database contents through the shared-block path.""" + db = get_all_relevant_facts(normalize(prop), normalize(assump), True, None) + assert_formula(normalize(assump), db) + return (sorted(frozenset(clause) for clause in db.data), + sorted(db.auxiliaries), + sorted(((atom, variable) for atom, variable in db.encoding.items()), + key=lambda item: (item[0].name, repr(item[0].arguments)))) + + +# --------------------------------------------------------------------------- +# template reuse actually reuses +# --------------------------------------------------------------------------- + +def test_shared_block_is_built_once_across_queries(build_counter) -> None: + for index in range(25): + symbol = Symbol(f"reuse{index}") + # Composite assumptions defeat the fast path, exercising the + # known-facts pipeline on every iteration. + assert satask(Q.real(symbol), + Q.rational(symbol) & Q.commutative(symbol)) is True + assert satask(Q.integer(symbol) & Q.real(symbol), + Q.even(symbol)) is True + # One block per template selection ever requested, never per call. + assert 1 <= len(build_counter) <= 4 + + +def test_binding_cache_is_actually_hit() -> None: + # Poison the minting path; a warm subject must reuse its binding. + adapter = SympyAdapter() + db1 = ClauseDB() + adapter.add_known_facts([x], db1) + binding1 = _subject_bindings._cache[(True, False, x)][0] + + def explode(atom: Any) -> int: + raise AssertionError("db.literal must not run on a binding cache hit") + + db2 = ClauseDB() + monkey_literal = ClauseDB.literal + monkeypatch = pytest.MonkeyPatch() + try: + monkeypatch.setattr(ClauseDB, "literal", explode) + adapter.add_known_facts([x], db2) + finally: + monkeypatch.undo() + del monkey_literal + binding2 = _subject_bindings._cache[(True, False, x)][0] + assert binding1 == binding2 + assert db2.encoding[iter_next_atom()] == 1 + + +def iter_next_atom() -> Any: + predicates, _ = _known_template(True, False) + return predicates[0](x) + + +def test_binding_cache_hit_replicates_db_literal(monkeypatch: pytest.MonkeyPatch) -> None: + # The cached-hit fill must leave encoding and symbols exactly as the + # minting path would, including the atom objects themselves. + adapter = SympyAdapter() + minted = ClauseDB() + adapter.add_known_facts([x, y], minted) + + fresh = ClauseDB() + adapter.add_known_facts([x, y], fresh) # second call: cache hits + for atom, variable in minted.encoding.items(): + assert fresh.encoding[atom] == variable + assert fresh.symbols[variable] is atom + assert fresh._next_variable == minted._next_variable + + +def test_binding_cache_does_not_cross_template_selections() -> None: + # A binding minted under the numbers-only template must never be + # replayed against the merged (numbers+matrices) template. + adapter = SympyAdapter() + numbers_only = ClauseDB() + adapter.add_known_facts([x], numbers_only) + numbers_binding = _subject_bindings._cache[(True, False, x)][0] + + merged = ClauseDB() + adapter.add_known_facts([A, x], merged) + merged_binding = _subject_bindings._cache[(True, True, x)][0] + assert len(merged_binding) > len(numbers_binding) + + +def test_stale_layout_falls_back_to_minting() -> None: + # A database whose next free variable does not match the cached + # binding's base must fall back to db.literal, not replay stale ids. + from reasoning.predicates import AppliedPredicate as LocalApplied + from reasoning.predicates import Predicate as LocalPredicate + + adapter = SympyAdapter() + warm = ClauseDB() + adapter.add_known_facts([x], warm) + + populated = ClauseDB() + # Pre-occupy variable 1 so the cached layout cannot apply. + populated.literal(LocalApplied(LocalPredicate("custom_predicate"), (x,))) + adapter.add_known_facts([x], populated) + assert populated.encoding[LocalApplied(LocalPredicate("custom_predicate"), (x,))] == 1 + predicates, _ = _known_template(True, False) + for position, predicate in enumerate(predicates): + assert populated.encoding[predicate(x)] == position + 2 + + +# --------------------------------------------------------------------------- +# databases byte-identical to the re-emission oracle +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("prop,assump", [ + (Q.real(x), Q.rational(x)), + (Q.real(x), True), + (Q.real(x + y), And(Q.rational(x), Q.rational(y))), + (Q.positive(x * y), And(Q.negative(x), Q.negative(y))), + (Q.zero(Abs(x * y)), Q.zero(x)), + (Q.real(sqrt(x)), Q.nonnegative(x)), + (Q.diagonal(A), Q.invertible(A)), + (Q.positive(Add(*[Symbol(f"pp{i}", positive=True) for i in range(20)], + *[Symbol(f"pn{i}", negative=True) for i in range(20)])), + And(*[Q.positive(Symbol(f"pp{i}")) for i in range(20)], + *[Q.negative(Symbol(f"pn{i}")) for i in range(20)])), +]) +def test_database_matches_reemission_oracle(prop: Any, assump: Any) -> None: + assert current_db(prop, assump) == oracle_db(prop, assump) + + +# --------------------------------------------------------------------------- +# answers unchanged across a broad battery +# --------------------------------------------------------------------------- + +def literal(predicate: Any, subject: Any, positive: bool) -> Any: + return predicate(subject) if positive else Not(predicate(subject)) + + +@pytest.mark.parametrize("prop_sign", [True, False]) +@pytest.mark.parametrize("assump_sign", [True, False]) +def test_battery_number_predicate_pairs(prop_sign: bool, assump_sign: bool) -> None: + # Every ordered pair of known number predicates, both polarities, over + # fresh subjects; the full pipeline is the oracle. + for name in NUMBER_PREDICATES: + prop = getattr(Q, name, None) + if prop is None: + continue + for assump_name in NUMBER_PREDICATES: + assump = getattr(Q, assump_name, None) + if assump is None: + continue + symbol = Symbol(f"b_{name}_{assump_name}") + compare_with_oracle(literal(prop, symbol, prop_sign), + literal(assump, symbol, assump_sign)) + + +def test_battery_matrix_predicate_pairs() -> None: + for prop_name in MATRIX_PREDICATES: + for assump_name in MATRIX_PREDICATES: + for prop_sign in (True, False): + prop = getattr(Q, prop_name) + assump = getattr(Q, assump_name) + matrix = MatrixSymbol(f"m_{prop_name}_{assump_name}_{prop_sign}", 2, 2) + compare_with_oracle( + prop(matrix) if prop_sign else Not(prop(matrix)), assump(matrix)) + + +def test_battery_no_assumptions_and_composites() -> None: + for name in NUMBER_PREDICATES + MATRIX_PREDICATES: + predicate = getattr(Q, name, None) + if predicate is None: + continue + compare_with_oracle(predicate(x)) + compare_with_oracle(Not(predicate(x))) + # Composite propositions/assumptions, relations, mixed subjects, and + # the shapes the memo tests use. + compare_with_oracle(Q.real(x) & Q.nonnegative(x), Q.positive(x)) + compare_with_oracle(Q.zero(x * y), Q.zero(x)) + compare_with_oracle(Eq(x, x), Q.real(x)) + compare_with_oracle(Q.real(x + y), Q.rational(x) & Q.rational(y)) + compare_with_oracle(Q.positive(Abs(x)), Q.real(x)) + compare_with_oracle(Q.real(sqrt(x)), Q.nonnegative(x)) + compare_with_oracle(Q.real(x), ~Q.real(x)) + compare_with_oracle(Q.real(x), Q.imaginary(x)) + with pytest.raises(ValueError): + satask(Q.real(x), Q.real(x) & ~Q.real(x)) + with pytest.raises(TypeError): + satask(x, Q.real(x)) + + +def test_battery_issue3_harness_shapes() -> None: + # The 13 shapes of the issue #3 reproduction script, with fresh + # subjects per case, against the full-pipeline oracle. + from sympy import exp, sin + + def pair(prop: Any, assump: Any) -> None: + compare_with_oracle(prop, assump) + + pair(Q.positive(x), Q.positive(x)) + pair(Q.real(Symbol("r1")), Q.rational(Symbol("r1"))) + pair(Q.integer(Symbol("e1")), Q.even(Symbol("e1"))) + pair(Q.positive(x + y), And(Q.positive(x), Q.positive(y))) + pair(Q.positive(x * y), And(Q.negative(x), Q.negative(y))) + pair(Q.positive(x + y), And(Q.real(x), Q.real(y))) + pair(Q.real(x ** 2), Q.real(x)) + pair(Q.positive(exp(x)), Q.real(x)) + pair(Q.real(sin(x)), Q.real(x)) + pair(Q.nonnegative(Abs(x)), Q.real(x)) + pair(Q.real(sqrt(x)), Q.nonnegative(x)) + four = [Symbol("f1", positive=True), Symbol("f2", negative=True)] + pair(Q.positive(four[0] + four[1] + four[0] + four[1]), + And(Q.positive(four[0]), Q.negative(four[1]))) + big = [Symbol(f"bg{i}", positive=True) for i in range(40)] + pair(Q.positive(Add(*big)), And(*[Q.positive(v) for v in big])) + + +def compare_with_oracle(prop: Any, assump: Any = True, + **kw: Any) -> None: + """Assert the public entry and the full pipeline agree bit for bit.""" + expected = _satask_pipeline(normalize(prop), normalize(assump), + kw.get("use_known_facts", True), + kw.get("iterations"), kw.get("early_return", False), + kw.get("use_lra_theory", False)) + assert satask(prop, assump, **kw) is expected diff --git a/reasoning/tests/test_polarity_encoding.py b/reasoning/tests/test_polarity_encoding.py new file mode 100644 index 0000000..1533cc6 --- /dev/null +++ b/reasoning/tests/test_polarity_encoding.py @@ -0,0 +1,288 @@ +"""Polarity-aware distribution of CNF-able antecedents (issue #2). + +``assert_formula`` compiles ``Implies(Or(And-of-literals...), consequent)`` +(and the right-nested form with a conjunction prefix) by distributing in the +negative polarity: one clause per disjunct, no Tseitin auxiliaries. These +tests lock down + +- the quadratic-to-linear clause/auxiliary counts of the Add + sign-determination handler facts, +- that consequent-position Or-of-Ands still goes through Tseitin unchanged, +- a strong equivalence property: for every total assignment (exhaustive up + to 8 atoms, random beyond), the asserted formula is true exactly when all + emitted clauses are satisfied. +""" +from itertools import product +import random + +import pytest +from sympy import Add, Q, Symbol + +from reasoning.clauses import ( + AND, IMPLIES, NOT, OR, ClauseDB, Formula, assert_formula, compile_formula, + iter_atoms, +) +from reasoning.sathandlers import class_fact_registry + + +def _formula_value(formula: object, values: dict[object, bool]) -> bool: + if formula is True: + return True + if formula is False: + return False + if not isinstance(formula, Formula): + return values[formula] + args = [_formula_value(arg, values) for arg in formula.args] + if formula.op == "and": + return all(args) + if formula.op == "or": + return any(args) + if formula.op == "not": + return not args[0] + if formula.op == "implies": + return not args[0] or args[1] + raise AssertionError(formula.op) + + +def _clauses_hold(db: ClauseDB, values: dict[object, bool]) -> bool: + """Every emitted clause is satisfied under the atom assignment.""" + assignment = {db.literal(atom): value for atom, value in values.items()} + return _clauses_hold_variables(db, assignment) + + +def _clauses_hold_variables(db: ClauseDB, + assignment: dict[int, bool]) -> bool: + return all(any(assignment[abs(literal)] == (literal > 0) for literal in clause) + for clause in db.data) + + +def _can_extend(db: ClauseDB, values: dict[object, bool]) -> bool: + """Some assignment of the auxiliary variables satisfies all clauses.""" + fixed = {db.literal(atom): value for atom, value in values.items()} + free = [variable for variable in db.variables if variable not in fixed] + for bits in product((False, True), repeat=len(free)): + extended = dict(fixed) + extended.update(zip(free, bits)) + if _clauses_hold_variables(db, extended): + return True + return False + + +def _exactly_one(items: tuple) -> object: + return OR(*(AND(item, *(NOT(other) for other in items if other is not item)) + for item in items)) + + +def _assert_equivalent(formula: object) -> ClauseDB: + """Assert *formula* and check clauses ⟺ formula over every assignment.""" + db = ClauseDB() + assert_formula(formula, db) + atoms = tuple(dict.fromkeys(iter_atoms(formula))) + assert len(atoms) <= 10 + for bits in product((False, True), repeat=len(atoms)): + values = dict(zip(atoms, bits)) + assert _clauses_hold(db, values) == _formula_value(formula, values) + return db + + +def _canonical(db: ClauseDB) -> tuple: + """Numbering-invariant clause structure (atom names, aux by first use).""" + renamed: dict[int, str] = {} + + def name(literal: int) -> str: + variable = abs(literal) + if variable in db.auxiliaries: + if variable not in renamed: + renamed[variable] = f"@{len(renamed) + 1}" + term = renamed[variable] + else: + term = repr(db.symbols[variable]) + return term if literal > 0 else "~" + term + + return tuple(frozenset(name(literal) for literal in clause) + for clause in db.data) + + +def _assert_matches_reference_tseitin(formula: object) -> ClauseDB: + """Non-fast-pathed shapes must emit exactly the Tseitin clauses.""" + db = ClauseDB() + assert_formula(formula, db) + assert db.auxiliaries # no distribution happened + reference = ClauseDB() + literal = compile_formula(formula, reference) + reference.add_clause((literal,)) + assert _canonical(db) == _canonical(reference) + assert len(db.auxiliaries) == len(reference.auxiliaries) + return db + + +# Reproducer shapes from issue #2 (Add sign-determination handler facts). + +def _sign_fact_disjuncts(n: int) -> object: + return OR(*(AND(*[('p%d' % (i + 1)) if i == j else NOT('p%d' % (i + 1)) + for i in range(n)]) for j in range(n))) + + +@pytest.mark.parametrize("n", [4, 10, 40]) +def test_negative_position_counts_drop_from_quadratic_to_linear(n: int) -> None: + # Before: n=4 -> 37 clauses/8 aux, n=10 -> 139/14, n=40 -> 1729/44. + nested = IMPLIES(AND(*['r%d' % (i + 1) for i in range(n)]), + IMPLIES(_sign_fact_disjuncts(n), 'q')) + db = ClauseDB() + assert_formula(nested, db) + assert (len(db.data), len(db.auxiliaries)) == (n, 0) + + plain = IMPLIES(_sign_fact_disjuncts(n), NOT('q')) + db = ClauseDB() + assert_formula(plain, db) + assert (len(db.data), len(db.auxiliaries)) == (n, 0) + + +def test_consequent_position_still_compiles_through_tseitin() -> None: + _assert_matches_reference_tseitin(IMPLIES('a', OR(AND('b', 'c'), AND('d', 'e')))) + + +def test_consequent_position_nested_implies_still_compiles_through_tseitin() -> None: + _assert_matches_reference_tseitin(IMPLIES(OR(AND('a', 'b'), AND('c', 'd')), + IMPLIES('q', 'r'))) + + +def test_complex_antecedent_disjuncts_still_compile_through_tseitin() -> None: + # An Or inside an And disjunct is not a conjunction of literals. + _assert_matches_reference_tseitin(Formula("implies", ( + Formula("or", ( + Formula("and", ('a', Formula("or", ('b', 'c')))), + AND('d', 'e'))), + 'q'))) + + +# Exhaustive equivalence: fast-path family emits no auxiliaries, and the +# clause set is true exactly when the formula is. + +@pytest.mark.parametrize("formula", [ + IMPLIES(OR(AND('a', 'b'), AND('c', 'd')), 'q'), + IMPLIES(OR('a', 'b', 'c'), NOT('q')), + IMPLIES(OR(AND('a', 'b', 'c'), AND('d', 'e', 'f')), OR('q', NOT('r'))), + IMPLIES(AND('a', 'b'), + IMPLIES(OR(AND('c', 'd'), AND('e', 'f')), NOT('g'))), + IMPLIES(OR(AND('a'), AND('b', 'c'), 'd', AND('e', 'f')), 'q'), + IMPLIES(AND('a', 'b'), + IMPLIES(OR(AND('a', 'c'), AND('b', 'd')), 'q')), + # Constants inside the antecedent (bypass the simplifying constructors). + Formula("implies", ( + Formula("or", ( + Formula("and", ('a', True)), + Formula("and", (False, 'b')), + AND('c', 'd'))), + 'q')), + # A self-contradictory disjunct contributes a tautological clause. + Formula("implies", ( + Formula("or", ( + Formula("and", ('a', NOT('a'))), + AND('b', 'c'))), + 'q')), + # Duplicated literals and single-disjunct Or. + Formula("implies", ( + Formula("or", ( + Formula("and", ('a', 'a', NOT('b'))), + Formula("and", ('c', NOT('d'))))), + 'q')), + Formula("implies", (Formula("or", (AND('a', 'b'),)), 'q')), + # A constant consequent. + Formula("implies", (OR(AND('a', 'b'), AND('c', 'd')), False)), +]) +def test_fast_path_formula_equivalent_to_clauses_exhaustively(formula: object) -> None: + db = _assert_equivalent(formula) + assert db.auxiliaries == set() + + +@pytest.mark.parametrize("formula", [ + IMPLIES('a', OR(AND('b', 'c'), AND('d', 'e'))), + Formula("implies", ( + Formula("or", ( + AND('a', 'b'), + Formula("or", ('c', 'd')))), + 'q')), +]) +def test_fallback_formula_equisatisfiable_exhaustively(formula: object) -> None: + db = ClauseDB() + assert_formula(formula, db) + assert db.auxiliaries + atoms = tuple(dict.fromkeys(iter_atoms(formula))) + for bits in product((False, True), repeat=len(atoms)): + values = dict(zip(atoms, bits)) + assert _can_extend(db, values) == _formula_value(formula, values) + + +# Random shapes: exhaustive up to 8 atoms, sampled beyond. + +def _random_fast_path_formula(rng: random.Random, atoms: list[str]) -> object: + disjuncts = [] + for _ in range(rng.randint(2, 4)): + width = rng.randint(1, 3) + literals = [atom if rng.random() < 0.5 else NOT(atom) + for atom in rng.sample(atoms, min(width, len(atoms)))] + disjuncts.append(AND(*literals) if len(literals) > 1 else literals[0]) + consequent_literals = [atom if rng.random() < 0.5 else NOT(atom) + for atom in rng.sample(atoms, rng.randint(1, 2))] + consequent = OR(*consequent_literals) + disjunction = Formula("or", tuple(disjuncts)) if len(disjuncts) > 1 else disjuncts[0] + if rng.random() < 0.5: + prefix = [atom if rng.random() < 0.5 else NOT(atom) + for atom in rng.sample(atoms, rng.randint(1, 2))] + return IMPLIES(AND(*prefix), IMPLIES(disjunction, consequent)) + return IMPLIES(disjunction, consequent) + + +def test_random_fast_path_formulas_equivalent_to_clauses() -> None: + rng = random.Random(20260922) + checked = 0 + for trial in range(24): + size = 8 if trial < 12 else rng.randint(9, 12) + atoms = [f'v{trial}_{index}' for index in range(size)] + formula = _random_fast_path_formula(rng, atoms) + db = ClauseDB() + assert_formula(formula, db) + assert db.auxiliaries == set() + if size <= 8: + assignments = (dict(zip(atoms, bits)) + for bits in product((False, True), repeat=size)) + else: + assignments = (dict(zip(atoms, rng.choices([False, True], k=size))) + for _ in range(256)) + for values in assignments: + checked += 1 + assert _clauses_hold(db, values) == _formula_value(formula, values) + assert checked >= 12 * 2 ** 8 + 12 * 256 + + +def test_random_fallback_formulas_equisatisfiable() -> None: + rng = random.Random(409) + for trial in range(8): + atoms = [f'w{trial}_{index}' for index in range(5)] + formula = IMPLIES(rng.choice(atoms), + OR(AND(rng.choice(atoms), rng.choice(atoms)), + AND(rng.choice(atoms), rng.choice(atoms)))) + db = ClauseDB() + assert_formula(formula, db) + assert db.auxiliaries + for bits in product((False, True), repeat=len(atoms)): + values = dict(zip(atoms, bits)) + assert _can_extend(db, values) == _formula_value(formula, values) + + +# The real Add sign-determination handler facts, exactly as discovery +# asserts them, verified exhaustively and now auxiliary-free. + +def test_add_handler_sign_facts_equivalent_and_auxiliary_free() -> None: + terms = [Symbol(f's{index}') for index in range(4)] + expr = Add(*terms) + db = ClauseDB() + for fact in sorted(class_fact_registry(expr), key=repr): + assert_formula(fact, db) + assert db.auxiliaries == set() + assert len(db.data) == 13 # 5 flat + 4 (exactly-one integer) + 4 (nested irrational) + + for fact in sorted(class_fact_registry(expr), key=repr): + fact_db = _assert_equivalent(fact) + assert fact_db.auxiliaries == set() diff --git a/reasoning/tests/test_satask_memo.py b/reasoning/tests/test_satask_memo.py new file mode 100644 index 0000000..bfdff17 --- /dev/null +++ b/reasoning/tests/test_satask_memo.py @@ -0,0 +1,271 @@ +"""Memoization and fast-path behavior of :mod:`reasoning.satask`. + +The optimizations inside ``satask`` must be semantically transparent: the +memoized public entry point and the fast path must return exactly what the +full pipeline returns for every query. These tests treat the full pipeline +(:func:`reasoning.satask._satask_pipeline`, which builds the clause database, +runs discovery, and invokes the solver) as the oracle and compare against it +bit for bit, including raised exceptions. The oracle deliberately bypasses +both the memoization and the fast path. +""" +from __future__ import annotations + +from typing import Any + +import pytest +from sympy import Abs, Not, Q, sqrt +from sympy import oo +from sympy.core.relational import Eq +from sympy.core.symbol import symbols +from sympy.matrices.expressions.matexpr import MatrixSymbol + +from reasoning.knownfacts import MATRIX_PREDICATES, NUMBER_PREDICATES +from reasoning.satask import ( + _MEMO_MAXSIZE, _satask_pipeline, normalize, satask, _satask_memoized, +) + +x, y = symbols("x y") +A = MatrixSymbol("A", 2, 2) + + +@pytest.fixture(autouse=True) +def clear_memo(): + """Isolate every test from cache state left behind by other tests.""" + _satask_memoized.cache_clear() + yield + _satask_memoized.cache_clear() + + +def oracle(prop: Any, assump: Any = True, **kw: Any) -> Any: + """Run the full pipeline directly: no memoization, no fast path.""" + return _satask_pipeline(normalize(prop), normalize(assump), + kw.get("use_known_facts", True), + kw.get("iterations"), kw.get("early_return", False), + kw.get("use_lra_theory", False)) + + +def compare(prop: Any, assump: Any = True, **kw: Any) -> None: + """Assert public and full-pipeline answers agree bit for bit.""" + try: + expected = oracle(prop, assump, **kw) + except Exception as exc: # noqa: BLE001 - the raise is reproduced below + with pytest.raises(type(exc), match=f"^{__import__('re').escape(str(exc))}$"): + satask(prop, assump, **kw) + return + assert satask(prop, assump, **kw) is expected + + +def literal(predicate: Any, positive: bool) -> Any: + return predicate(x) if positive else Not(predicate(x)) + + +def test_memo_returns_identical_repeated_results() -> None: + # Warm cache: True, False, and None answers all survive repeated calls. + assert satask(Q.real(x), Q.rational(x)) is True + assert satask(Q.real(x), ~Q.real(x)) is False + assert satask(Q.real(x)) is None + assert satask(Q.positive(x), Q.positive(x)) is True + for _ in range(3): + assert satask(Q.real(x), Q.rational(x)) is True + assert satask(Q.real(x), ~Q.real(x)) is False + assert satask(Q.real(x)) is None + + +def test_memo_cache_is_actually_hit() -> None: + # Poison every pipeline entry point; a warm repeat must not rerun them. + import reasoning.satask as module + + def explode(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("the full pipeline must not run on a cache hit") + + assert satask(Q.real(x), Q.rational(x)) is True + saved = {name: getattr(module, name) + for name in ("normalize", "get_all_relevant_facts")} + try: + for name in saved: + setattr(module, name, explode) + assert satask(Q.real(x), Q.rational(x)) is True + finally: + for name, function in saved.items(): + setattr(module, name, function) + + +def test_memo_does_not_leak_across_assumptions_or_props() -> None: + # Same proposition under different assumptions; same assumptions over + # different propositions; also: cache order does not matter. + pairs = [ + (Q.real(x), Q.rational(x)), # True + (Q.real(x), ~Q.rational(x)), # None + (Q.positive(x), Q.even(x)), # None: different family + (Q.integer(x), Q.even(x)), # True + (Q.real(x), Q.imaginary(x)), # True via negative edge + (Q.real(x), ~Q.imaginary(x)), # None + (Q.real(x) & Q.real(y), Q.rational(x) & Q.rational(y)), # full pipeline + (Q.real(x), Q.rational(y)), # different subjects, full pipeline + ] + for prop, assump in pairs: + compare(prop, assump) + # Reverse stress: cache the None entries after the True entries. + _satask_memoized.cache_clear() + for prop, assump in reversed(pairs): + compare(prop, assump) + + +def test_memo_never_mixes_boolean_and_integer_inputs() -> None: + # Python bools compare and hash equal to 0/1, so a key on the raw values + # would let satask(False) answer satask(0) or satask(True) answer + # satask(1). Booleans (valid queries) and integers (which normalize + # rejects with TypeError) must stay apart in both argument positions. + assert satask(False) is False + with pytest.raises(TypeError): + satask(0) + assert satask(True) is True + with pytest.raises(TypeError): + satask(1) + with pytest.raises(TypeError): + satask(Q.real(x), 0) + with pytest.raises(TypeError): + satask(Q.real(x), 1) + assert satask(Q.real(x), True) is None + with pytest.raises(ValueError): + satask(Q.real(x), False) + + +def test_memo_does_not_cache_exceptions() -> None: + # Inconsistent assumptions raise on every call; the failure must not + # poison later, valid queries over the same expressions. + for _ in range(3): + with pytest.raises(ValueError): + satask(Q.real(x), Q.real(x) & ~Q.real(x)) + assert satask(Q.real(x)) is None + assert satask(Q.real(x), Q.real(x)) is True + + +def test_memo_iteration_values_are_kept_apart() -> None: + # iterations=0 is accepted while False and 0.0 raise ValueError; keys for + # values that compare equal as numbers must not share cache slots. + for value in (0, 5, oo): + prop, assump = Q.real(x), Q.rational(x) + assert satask(prop, assump, iterations=value) == oracle( + prop, assump, iterations=value) + with pytest.raises(ValueError): + satask(Q.real(x), Q.rational(x), iterations=False) + with pytest.raises(ValueError): + satask(Q.real(x), Q.rational(x), iterations=0.0) + with pytest.raises(ValueError): + satask(Q.real(x), Q.rational(x), iterations="1") + with pytest.raises(ValueError): + satask(Q.real(x), Q.rational(x), iterations=-1) + + +def test_fast_path_matches_full_pipeline_on_leaf_cases() -> None: + # The leaf regimes from the issue: declared facts and known single-hop + # inferences, including a negated proposition and a negated assumption. + for prop, assump in [ + (Q.positive(x), Q.positive(x)), + (Q.real(x), Q.rational(x)), + (Q.integer(x), Q.even(x)), + (~Q.real(x), Q.rational(x)), + (Q.real(x), ~Q.rational(x)), + (Q.real(x), Q.imaginary(x)), + ]: + compare(prop, assump) + + +def test_lra_theory_disables_the_fast_path() -> None: + # The LRA theory decides degenerate relation literals that the clause + # database alone cannot, and can make a single literal assumption + # inconsistent, so the fast path must never answer when it is enabled. + # Every case here was answered wrongly (or without raising) by an earlier + # fast path that ignored ``use_lra_theory``. + for prop, assump, expected in [ + (Q.eq(x, x), ~Q.eq(x, x), ValueError), + (Q.gt(x, x), Q.gt(x, x), ValueError), + (Q.real(x), ~Q.eq(x, x), ValueError), + (~Q.eq(x, x), True, False), + (Q.gt(x, x), True, False), + (~Q.gt(x, x), True, True), + (Q.ge(x, x), True, True), + (Q.le(x, x), True, True), + (Q.eq(x, 0), Q.eq(x, 0), True), + (Q.eq(x, y), Q.eq(x, y), True), + (Q.real(x), Q.eq(x, 0), None), + (Q.zero(x), Q.eq(x, 0), None), + ]: + compare(prop, assump, use_lra_theory=True) + if expected is ValueError: + with pytest.raises(ValueError): + satask(prop, assump, use_lra_theory=True) + else: + assert satask(prop, assump, use_lra_theory=True) is expected + + +def test_not_fast_path_cases_still_work() -> None: + # Composite propositions, differing subjects, relations, and transfer + # facts must fall through to the full pipeline intact. + compare(Q.real(x) & Q.nonnegative(x), Q.positive(x)) + compare(Q.zero(x * y), Q.zero(x)) + compare(Eq(x, x), Q.real(x)) + compare(Q.real(x + y), Q.rational(x) & Q.rational(y)) + compare(Q.positive(Abs(x)), Q.real(x)) + compare(Q.real(sqrt(x)), Q.nonnegative(x)) + with pytest.raises(TypeError): + satask(x, Q.real(x)) + with pytest.raises(TypeError): + satask("junk") + + +def test_inconsistent_assumptions_match() -> None: + # Inconsistent assumptions raise through both the public entry point and + # the full pipeline, identically, regardless of cache state. The sample + # contradiction mirrors test_satask_early_return in test_satask.py: it is + # invisible to root-level unit propagation, so satask(early_return=True) + # still answers without raising, and the fast path defers to the full + # pipeline for every such composite assumption anyway. + contradiction = (Q.real(x) & (Q.positive(x) | Q.negative(x)) + & (Q.positive(x) >> Q.zero(x)) + & (Q.negative(x) >> Q.zero(x))) + public_raises(ValueError, Q.real(x), contradiction) + assert satask(Q.real(x), contradiction, early_return=True) is True + # Root-level-propagatable contradictions raise even with early_return: + public_raises(ValueError, Q.real(x), Q.real(x) & ~Q.real(x)) + with pytest.raises(ValueError): + satask(Q.real(x), Q.real(x) & ~Q.real(x), early_return=True) + compare(Q.real(x), Q.positive(x) & Q.negative(x)) + compare(Q.positive(x), ~Q.positive(x)) + compare(Q.real(x), ~Q.real(x)) + + +def public_raises(exception: type[BaseException], + prop: Any, assump: Any = True, **kw: Any) -> None: + with pytest.raises(exception): + satask(prop, assump, **kw) + + +def test_fast_path_matches_full_pipeline_single_hops_matrix() -> None: + predicates = [getattr(Q, name) for name in MATRIX_PREDICATES] + for prop in predicates: + for assump in predicates: + for prop_positive in (True, False): + compare(prop(A) if prop_positive else Not(prop(A)), assump(A)) + + +@pytest.mark.parametrize("prop_sign", [True, False]) +@pytest.mark.parametrize("assump_sign", [True, False]) +def test_fast_path_matches_full_pipeline_single_hops( + prop_sign: bool, assump_sign: bool) -> None: + # Exhaust the known-facts template directly: every ordered pair of known + # number predicates, in both directions and both polarities. The full + # pipeline is the oracle; whatever pair is not a single hop is exercised + # by it, so any fast-path divergence will show up. + predicates = {name: getattr(Q, name) for name in NUMBER_PREDICATES + if getattr(Q, name, None) is not None} + for prop_name in predicates: + for assump_name in predicates: + compare(literal(predicates[prop_name], prop_sign), + literal(predicates[assump_name], assump_sign)) + + +def test_memo_size_policy_is_bounded() -> None: + # Cache growth policy: a bounded LRU with a documented generous maxsize. + assert _MEMO_MAXSIZE == 4096