From 64101ca18edeb756fd277d5db67fb1aa1deba861 Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Mon, 21 Sep 2026 23:03:05 -0700 Subject: [PATCH 1/7] perf: memoize satask results on the query tuple Cache every satask result, including None answers, in a bounded lru_cache at the public entry point, keyed on the original (hashable) arguments -- the answer depends only on the arguments, so no intermediate state can take part. Exceptions are never cached, so inconsistent assumptions keep raising on every call. Repeated identical queries now cost a dict lookup (~0.5us instead of ~480us for a leaf query). Two key-normalization rules guard against Python-int collisions, since bools compare and hash equal to ints (True == 1, 0 == False): Python booleans are keyed on the SymPy constants that normalize maps them to anyway, and the iteration limit is keyed with its runtime type so iterations=False/0.0 cannot inherit an iterations=0 result. Unhashable inputs raise at key construction and fall through to the uncached pipeline, so behavior (including raised errors) stays identical to before. Part of #3. --- reasoning/satask.py | 83 ++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 82 insertions(+), 1 deletion(-) diff --git a/reasoning/satask.py b/reasoning/satask.py index d913b45..a767852 100644 --- a/reasoning/satask.py +++ b/reasoning/satask.py @@ -1,7 +1,8 @@ """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 .discovery import discover_facts, relevant_subjects @@ -13,6 +14,53 @@ 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,7 +96,40 @@ 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. 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 + +@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) db = get_all_relevant_facts(prop_formula, assump_formula, use_known_facts, iterations) From e9ca115b94f4534ce105a2d7eaf8b09c2991f020 Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Mon, 21 Sep 2026 23:03:28 -0700 Subject: [PATCH 2/7] perf: answer trivial literal queries through a fast path Before building the clause database, check whether the proposition is a single literal over fact-free subjects (no registered expression handlers) and its assumptions are either absent or exactly one literal over the same arguments. In those narrow shapes the database provably contains nothing but the known-facts template plus one unit (or nothing at all), so the answer follows from the template's one-step implication graph: - no assumptions: the template has no unit clauses, so the answer is None; - same atom: the unit itself (True or False); - otherwise: single-hop consequences of the known-facts template's binary clauses, reusing reasoning.knownfacts rather than any hard-coded pairs, with the fact-family selection mirroring SympyAdapter.add_known_facts. Anything else (conjunctions, negated assumptions that are not single-hop, relations, fact-bearing subjects such as numbers) falls through to the full pipeline untouched, and answers stay bit-identical. Leaf queries drop from ~480us to ~13-40us, within the measured cost of normalize plus one lattice lookup. Part of #3. --- reasoning/satask.py | 134 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 132 insertions(+), 2 deletions(-) diff --git a/reasoning/satask.py b/reasoning/satask.py index a767852..96c8c0e 100644 --- a/reasoning/satask.py +++ b/reasoning/satask.py @@ -4,10 +4,18 @@ 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, ) @@ -100,7 +108,11 @@ def satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool = True, 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. Like the rest + 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. Like the rest of the package, the cache assumes single-threaded use. """ try: @@ -119,6 +131,10 @@ def satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool = True, # 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, @@ -132,6 +148,11 @@ def _satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool, 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) + answer = _fast_path_answer(prop_formula, assump_formula, use_known_facts) + if answer is not _NO_FAST_ANSWER: + return cast("bool | None", answer) db = get_all_relevant_facts(prop_formula, assump_formula, use_known_facts, iterations) assert_formula(assump_formula, db) query = compile_formula(prop_formula, db) @@ -148,6 +169,115 @@ def _satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool, 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.""" From 11292c86ea35464a93de758d7dd663d912b4846a Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Mon, 21 Sep 2026 23:03:45 -0700 Subject: [PATCH 3/7] test: lock the satask memoization and fast path to the full pipeline New module treats reasoning.satask._satask (full pipeline: clause database, discovery, solver) as the oracle for every query, public and fast answers must agree bit for bit, including raised exceptions and cache-independent ordering. Coverage: - memo repeats (True/False/None), exception-free repeats (no poison), - Python-bool/integer key collisions (satask(0) vs satask(False); iterations=False/0.0 vs 0), - no leaks across assumption or proposition differences, forward and reverse cache-fill order, - leaf fast-path shapes from the issue, exhaustively over the single-hop pairs of the known-facts number and matrix templates, both polarities, - full-pipeline shapes that must NOT take the fast path, - inconsistent assumptions raising identically through both paths (root-level-propagatable vs propagation-invisible, mirroring test_satask_early_return). Part of #3. --- reasoning/tests/test_satask_memo.py | 241 ++++++++++++++++++++++++++++ 1 file changed, 241 insertions(+) create mode 100644 reasoning/tests/test_satask_memo.py diff --git a/reasoning/tests/test_satask_memo.py b/reasoning/tests/test_satask_memo.py new file mode 100644 index 0000000..a79edd2 --- /dev/null +++ b/reasoning/tests/test_satask_memo.py @@ -0,0 +1,241 @@ +"""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`, which builds the clause database, runs +discovery, and invokes the solver) as the oracle and compare against it bit +for bit, including raised exceptions. +""" +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, _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(prop, 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)): + 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_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 From 91477a5f9ee28b498466dd8010ddb773e3bb883b Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Tue, 22 Sep 2026 00:08:06 -0700 Subject: [PATCH 4/7] test: make the memo test oracle bypass the fast path _satask still contained the fast-path probe, so the exhaustive satask-vs-pipeline battery in test_satask_memo.py compared the fast path against itself and pinned nothing. Extract the pipeline body into _satask_pipeline (no behavior change) and use it as the oracle; also match exception messages, not just types, in compare(). --- reasoning/satask.py | 8 ++++++++ reasoning/tests/test_satask_memo.py | 18 ++++++++++-------- 2 files changed, 18 insertions(+), 8 deletions(-) diff --git a/reasoning/satask.py b/reasoning/satask.py index 96c8c0e..fd9b89b 100644 --- a/reasoning/satask.py +++ b/reasoning/satask.py @@ -153,6 +153,14 @@ def _satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool, 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) diff --git a/reasoning/tests/test_satask_memo.py b/reasoning/tests/test_satask_memo.py index a79edd2..7b34ba8 100644 --- a/reasoning/tests/test_satask_memo.py +++ b/reasoning/tests/test_satask_memo.py @@ -3,9 +3,10 @@ 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`, which builds the clause database, runs -discovery, and invokes the solver) as the oracle and compare against it bit -for bit, including raised exceptions. +(: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 @@ -20,7 +21,7 @@ from reasoning.knownfacts import MATRIX_PREDICATES, NUMBER_PREDICATES from reasoning.satask import ( - _MEMO_MAXSIZE, satask, _satask, _satask_memoized, + _MEMO_MAXSIZE, _satask_pipeline, normalize, satask, _satask_memoized, ) x, y = symbols("x y") @@ -37,9 +38,10 @@ def clear_memo(): def oracle(prop: Any, assump: Any = True, **kw: Any) -> Any: """Run the full pipeline directly: no memoization, no fast path.""" - return _satask(prop, assump, kw.get("use_known_facts", True), - kw.get("iterations"), kw.get("early_return", False), - kw.get("use_lra_theory", False)) + 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: @@ -47,7 +49,7 @@ def compare(prop: Any, assump: Any = True, **kw: Any) -> None: try: expected = oracle(prop, assump, **kw) except Exception as exc: # noqa: BLE001 - the raise is reproduced below - with pytest.raises(type(exc)): + with pytest.raises(type(exc), match=f"^{__import__('re').escape(str(exc))}$"): satask(prop, assump, **kw) return assert satask(prop, assump, **kw) is expected From f2f63acf3b02618520154b0141cead1226568c75 Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Tue, 22 Sep 2026 00:08:55 -0700 Subject: [PATCH 5/7] perf: never answer through the fast path when the LRA theory is on The fast path decided single-literal queries from the clause database alone, but with use_lra_theory=True the theory participates in every consistency check and query: it entails or refutes degenerate relation literals the clauses cannot (satask(~Q.eq(x, x), use_lra_theory=True) returned None instead of False, ~Q.gt/Q.ge/Q.le similarly), and it can make a single literal assumption inconsistent, so satask(Q.eq(x, x), ~Q.eq(x, x), use_lra_theory=True) returned False instead of raising ValueError. Gate the fast path off for the whole LRA mode; parity is restored and the mode is what the theory is documented to change. --- reasoning/satask.py | 16 +++++++++++----- reasoning/tests/test_satask_memo.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 5 deletions(-) diff --git a/reasoning/satask.py b/reasoning/satask.py index fd9b89b..367e162 100644 --- a/reasoning/satask.py +++ b/reasoning/satask.py @@ -112,8 +112,9 @@ def satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool = True, 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. Like the rest - of the package, the cache assumes single-threaded use. + 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), @@ -150,9 +151,14 @@ def _satask(proposition: SymPyExpr | bool, assumptions: SymPyExpr | bool, assump_formula: NormalizedFormula = normalize(assumptions) # Validate here so an invalid limit raises even when the fast path answers. _iteration_limit(iterations) - answer = _fast_path_answer(prop_formula, assump_formula, use_known_facts) - if answer is not _NO_FAST_ANSWER: - return cast("bool | None", answer) + # 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) diff --git a/reasoning/tests/test_satask_memo.py b/reasoning/tests/test_satask_memo.py index 7b34ba8..bfdff17 100644 --- a/reasoning/tests/test_satask_memo.py +++ b/reasoning/tests/test_satask_memo.py @@ -172,6 +172,34 @@ def test_fast_path_matches_full_pipeline_on_leaf_cases() -> None: 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. From 44a7f2c2be99d1f03b07d898da45249db47b00e5 Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Tue, 22 Sep 2026 01:09:55 -0700 Subject: [PATCH 6/7] perf: distribute CNF-able antecedents in negative polarity only The two Add sign-determination facts (exactly-one-arg shapes from sathandlers) fell through to Tseitin compilation in assert_formula and blew up quadratically: n auxiliary variables times (n + 1) biconditional clauses per exactly-one antecedent. On a 40-term positive-sum query discovery emitted 3,499 clauses (1,685 + 1,729 from exactly these two facts, 86 auxiliaries, database 6,740 clauses); the isolated n=40 reproducer needed 1,729 clauses / 44 auxiliaries. assert_formula now distributes in negative polarity only: - Implies(Or(And-of-literals...), flat consequent) becomes one clause per disjunct (negated disjunct literals + consequent literals), - Implies(And-of-literals, Implies(Or(And-of-literals...), flat C)) flattens the right nesting and distributes the same way. No auxiliaries. Distribution never applies in consequent position; those shapes (and every other non-fast-pathed shape) keep Tseitin. The new shape checks allocate no variables, so non-fast-pathed shapes reach compile_formula with a byte-identical database (clauses, aux and encoding verified exactly against the pre-change code). Divergences from the issue's idealized shape: the quadratic facts are the non-nested exactly-one-integer fact Implies(Or(And...), Not(atom)) and the nested real -> exactly-one-irrational fact (the issue described both as nested); distribution is confined to a single Or-of-Ands per implication, so deeper right-nesting still falls through to Tseitin. n=40: reproducer 1,729/44 -> 40/0; discovery 3,499 -> 165 clauses, aux 86 -> 0, database 6,740 -> 3,406 clauses. Issue #3 harness: all 13 answers identical, 40-term row 50,651us -> 39,373us, geomean 19x -> 18x (run noise). reasoning/tests/test_polarity_encoding.py locks the counts (n=4, 10, 40), the Tseitin preservation of consequent-position shapes, and a strong equivalence property: for every total assignment (exhaustive up to 8 atoms, sampled to 12; real Add handler facts exhaustive) the asserted formula is true exactly when every emitted clause is true. --- reasoning/clauses.py | 62 +++++ reasoning/tests/test_polarity_encoding.py | 288 ++++++++++++++++++++++ 2 files changed, 350 insertions(+) create mode 100644 reasoning/tests/test_polarity_encoding.py diff --git a/reasoning/clauses.py b/reasoning/clauses.py index d018d8d..7b21ad6 100644 --- a/reasoning/clauses.py +++ b/reasoning/clauses.py @@ -362,6 +362,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,6 +421,41 @@ 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", 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() From 009f0305df4328551bbac3cc74b48a05c78e8175 Mon Sep 17 00:00:00 2001 From: tilorc-bot Date: Tue, 22 Sep 2026 01:52:39 -0700 Subject: [PATCH 7/7] perf: compile known facts once and replay them per query add_known_facts re-translated the fixed known-facts template into integer clauses on every call: 81 clauses and 31 db.literal mints per subject, per query (3,240 clauses / 6.8ms for the 40-subject case; ~27% of a non-fast-path trivial pipeline). The template is a fixed finite theory, so the translation is now compiled once, lazily, into a SharedClauses block (clauses over template positions) and each query only binds its subjects: a per-subject binding cache (keyed by template selection + subject, stored in the adapter module) skips the 31 atom mints whenever the variable layout repeats -- exactly the fresh-database layout satask always produces -- and SharedClauses.replay appends the pre-translated clause sets verbatim. Non-consecutive layouts fall back to full minting and re-basing, so external callers stay correct. Sharing stops at construction time: every database receives its own clause copies and no solver ever observes shared state between queries, which keeps each query's engine/db exactly as isolated as before. The databases are byte-identical to the per-clause re-emission (clauses, auxiliaries, and atom encoding verified against an oracle across 8 shapes, including the 40-subject one), and no add_clause normalization is skipped that could ever fire: template clauses are duplicate- and tautology-free by construction, and distinct positions cannot collapse. Solver/engine reuse was evaluated and skipped: the solver clones itself on every assumed solve (solver.copy), mutates its clause list, sentinel watch sets, and occurrence counts during search, and its add() cannot introduce variables, so a persistent engine would need trail-recorded root propagation, pre-allocated variable spaces, and a clone that shares the immutable clause store -- deep surgery obsoleted by the planned DPLL(T) work (design doc section 4 re-scopes item 3 to exactly this template-level reuse). Micro (medians, same machine, interleaved runs): - add_known_facts(40 subjects): 6,631us -> 1,788us (3.7x); template translation now once per process instead of per call - fresh-symbol pipeline query (2-lit prop, 1-lit assum): 1,068 -> 694us - e2e fast-pathed trivial query: unchanged (memo/fast-path floor, 5.3us) Issue #3 harness (13 shapes): all answers identical, geomean 19x -> 17x with every pipeline-bound row improving 5-25% (leaf rows 39-47us, 40-term 33.6 -> 33.1ms; run noise dominates the solver-bound rows). New reasoning/tests/test_persistent_facts.py: builds-once counter, cache hit without db.literal, selection-keyed isolation, stale-layout fallback, byte-identical database oracle, and the full-pipeline answer battery over all ordered predicate pairs in both polarities plus the 13 harness shapes. Suite: 215 passed, 1 xfailed. --- reasoning/clauses.py | 78 +++++- reasoning/sympy_adapter.py | 84 ++++++- reasoning/tests/test_persistent_facts.py | 305 +++++++++++++++++++++++ 3 files changed, 451 insertions(+), 16 deletions(-) create mode 100644 reasoning/tests/test_persistent_facts.py diff --git a/reasoning/clauses.py b/reasoning/clauses.py index 7b21ad6..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.""" @@ -458,6 +530,6 @@ def _cnf_disjunct_formulas(formula: object) -> tuple[tuple[object, ...], ...] | __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/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