diff --git a/docs/design/verified-scope-solver.md b/docs/design/verified-scope-solver.md index d9de9e8f..e7fffed6 100644 --- a/docs/design/verified-scope-solver.md +++ b/docs/design/verified-scope-solver.md @@ -200,6 +200,54 @@ one verdict with a replayable `SupportCertificate`: Non-goals honored: no multi-candidate closure loop, no model ranking, no SMT dependency, no decode/runtime flag, and no production-scale performance claim. +## Implemented exact closure (VSS1-01 / SLM-61) + +[`dsl/solver/closure.py`](../../src/slm_training/dsl/solver/closure.py) adds the +monotone fixed-point operator `exact_closure(state, provider, *, query_order, cache, +certificate_store, max_queries)`. It is a **pure orchestration layer** over the +VSS0-04 oracle (a `SupportProvider` = the oracle + its replay checker) — it never +reimplements the support search — and removes only candidates carrying a +replay-valid `UNSUPPORTED` certificate. Reference fixed-point: + +```text +def exact_closure(state, provider): + while True: + if state.is_bottom: return certified_bottom(state) # every value certified-removed + holes = unresolved_holes(state) # domains with > 1 value + if not holes: return structurally_solved(state) # every domain singleton + removals = {} + for hole in order(holes): # smallest domain, then HoleId + for value in hole.values: # canonical value order + r = provider.check(state, query(state, hole, value)) # cached by (fp,hole,value,backend) + if r.verdict is SUPPORTED: keep_witness(r) # kept; alternatives not collapsed + elif r.verdict is UNSUPPORTED: + if provider.replay(r.certificate, state=state).ok: # REQUIRED: replay vs pass-start state + removals[hole].add(value, because=r.certificate) # DESTRUCTIVE — cites certificate + else: record_unknown(value) # stale/tampered proof never removes + else: record_unknown(value) # UNKNOWN never shrinks a domain + if not removals: return fixed_point(state) + state = apply_all(removals, against=state) # one new state per pass (pass-consistent) +``` + +- **Pass consistency.** Every deduction in a pass is computed against the same + pass-start state and applied together to form one new immutable state; the next + pass re-queries against the new fingerprint. A proof that went stale after a + reduction is therefore never reused — the cache is keyed by `state.fingerprint` + (plus hole/value/backend version), so a shrunk state is a cache miss and forces a + fresh query, and every removal replays against the exact pre-refinement state. +- **Three distinct terminal states.** *Structurally solved* = every domain is a + singleton (a candidate chosen per hole; still not a verified terminal). *Certified + bottom* = some domain is empty **because every one of its values was removed under + an accepted certificate** (`reached_fixed_point=True`). *Partial closure* = a + `max_queries` budget stopped the loop with live domains (`reached_fixed_point=False`, + `stop_reason="budget:max_queries"`) — an honest partial, never a bottom or a + solved claim. +- **Honesty.** `UNKNOWN` (and any `UNSUPPORTED` whose certificate fails replay) + never decreases a domain; `result.state` is always a subset of the input and every + `CertifiedDeduction` cites one certificate id per removed value. `CertifiedDeduction` + stores certificate **references** (digests) plus compact summaries; full + certificates live in a caller-provided `certificate_store`. + ## Reference support semantics | Verdict | Requirement | Removal permitted? | diff --git a/src/slm_training/dsl/solver/__init__.py b/src/slm_training/dsl/solver/__init__.py index b3db1d9b..eb41629c 100644 --- a/src/slm_training/dsl/solver/__init__.py +++ b/src/slm_training/dsl/solver/__init__.py @@ -13,6 +13,16 @@ SolverBounds, SupportVerdict, ) +from slm_training.dsl.solver.closure import ( + CertifiedDeduction, + ClosureCounters, + ClosureResult, + EnumerativeSupportProvider, + SupportProvider, + WitnessRef, + default_query_order, + exact_closure, +) from slm_training.dsl.solver.support import ( EnumerativeSupportOracle, ExpandStatus, @@ -31,8 +41,12 @@ ) __all__ = [ + "CertifiedDeduction", + "ClosureCounters", + "ClosureResult", "DomainValue", "EnumerativeSupportOracle", + "EnumerativeSupportProvider", "ExpandStatus", "ExpandStep", "FiniteDomainState", @@ -45,6 +59,7 @@ "SolverBounds", "SupportCertificate", "SupportOracle", + "SupportProvider", "SupportQuery", "SupportResult", "SupportVerdict", @@ -52,6 +67,9 @@ "Verifier", "VerifyOutcome", "VerifyStatus", + "WitnessRef", "completion_forest_state", + "default_query_order", + "exact_closure", "replay_support_certificate", ] diff --git a/src/slm_training/dsl/solver/closure.py b/src/slm_training/dsl/solver/closure.py new file mode 100644 index 00000000..f968ada1 --- /dev/null +++ b/src/slm_training/dsl/solver/closure.py @@ -0,0 +1,358 @@ +"""Certificate-checked exact closure over finite domains (VSS1-01). + +`exact_closure` is a deterministic, pure orchestration layer on top of the VSS0-04 +support oracle. It repeatedly queries candidate support and removes **only** +candidates carrying a replay-valid ``UNSUPPORTED`` certificate, until it reaches a +fixed point, certified bottom, an all-singleton state, or an honest partial result +under a work budget. It never reimplements the support search; it drives a +:class:`SupportProvider` (the oracle plus its replay checker) and records every +destructive step as a :class:`CertifiedDeduction`. + +Monotonicity and honesty: + +* ``SUPPORTED`` and ``UNKNOWN`` values are always kept; only replay-valid + ``UNSUPPORTED`` values are removed. A domain becomes **certified bottom** only + when *every* value was removed with an accepted certificate. +* All deductions in one pass are computed against a single consistent pass-start + state and applied together to produce one new immutable state; the next pass + re-queries against the new fingerprint so a proof that went stale after a + reduction is never reused. +* Soft scores/logits are not accepted anywhere in this API. + +Semantics are owned by ``docs/design/verified-scope-solver.md``. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Callable, Protocol + +from slm_training.dsl.solver.state import ( + DomainValue, + FiniteDomainState, + HoleDomain, + HoleId, + SupportVerdict, +) +from slm_training.dsl.solver.support import ( + EnumerativeSupportOracle, + ProblemExpander, + ReplayResult, + SupportCertificate, + SupportQuery, + SupportResult, + Verifier, + replay_support_certificate, +) + + +def _canonical_json(value: Any) -> str: + return json.dumps( + value, allow_nan=False, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + + +# --------------------------------------------------------------------------- # +# Provider seam: an oracle that can also replay its own certificates +# --------------------------------------------------------------------------- # + + +class SupportProvider(Protocol): + """A support oracle plus its replay checker and a stable backend version.""" + + @property + def backend_version(self) -> str: ... + + def check(self, state: FiniteDomainState, query: SupportQuery) -> SupportResult: ... + + def replay( + self, certificate: SupportCertificate, *, state: FiniteDomainState + ) -> ReplayResult: ... + + +class EnumerativeSupportProvider: + """Bundles :class:`EnumerativeSupportOracle` with certificate replay.""" + + def __init__(self, expander: ProblemExpander, verifier: Verifier) -> None: + self._expander = expander + self._verifier = verifier + self._oracle = EnumerativeSupportOracle(expander, verifier) + + @property + def backend_version(self) -> str: + return f"enumerative/{self._verifier.profile}" + + def check(self, state: FiniteDomainState, query: SupportQuery) -> SupportResult: + return self._oracle.check(state, query) + + def replay( + self, certificate: SupportCertificate, *, state: FiniteDomainState + ) -> ReplayResult: + return replay_support_certificate( + certificate, state=state, expander=self._expander, verifier=self._verifier + ) + + +# --------------------------------------------------------------------------- # +# Closure records +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True) +class WitnessRef: + """A kept ``SUPPORTED`` candidate and its certificate/witness digests.""" + + hole_id: HoleId + value: DomainValue + certificate_id: str + witness_digest: str | None + witness_source: str | None + + def to_dict(self) -> dict[str, Any]: + return { + "hole_id": self.hole_id.to_dict(), + "value": self.value.to_dict(), + "certificate_id": self.certificate_id, + "witness_digest": self.witness_digest, + "witness_source": self.witness_source, + } + + +@dataclass(frozen=True) +class CertifiedDeduction: + """One destructive domain reduction, each removal citing a certificate.""" + + before_fingerprint: str + after_fingerprint: str + hole_id: HoleId + removed: tuple[DomainValue, ...] + certificate_ids: tuple[str, ...] + reason: str + + def to_dict(self) -> dict[str, Any]: + return { + "before_fingerprint": self.before_fingerprint, + "after_fingerprint": self.after_fingerprint, + "hole_id": self.hole_id.to_dict(), + "removed": [value.to_dict() for value in self.removed], + "certificate_ids": list(self.certificate_ids), + "reason": self.reason, + } + + +@dataclass(frozen=True) +class ClosureCounters: + passes: int = 0 + support_queries: int = 0 + cache_hits: int = 0 + supported: int = 0 + unsupported: int = 0 + unknown: int = 0 + candidates_removed: int = 0 + verifier_calls: int = 0 + expanded_nodes: int = 0 + + def to_dict(self) -> dict[str, int]: + return { + "passes": self.passes, + "support_queries": self.support_queries, + "cache_hits": self.cache_hits, + "supported": self.supported, + "unsupported": self.unsupported, + "unknown": self.unknown, + "candidates_removed": self.candidates_removed, + "verifier_calls": self.verifier_calls, + "expanded_nodes": self.expanded_nodes, + } + + +@dataclass(frozen=True) +class ClosureResult: + state: FiniteDomainState + deductions: tuple[CertifiedDeduction, ...] + unknown_queries: tuple[SupportQuery, ...] + witnesses: tuple[WitnessRef, ...] + counters: ClosureCounters + reached_fixed_point: bool + stop_reason: str | None = None + + +# --------------------------------------------------------------------------- # +# Query order +# --------------------------------------------------------------------------- # + +QueryOrder = Callable[[FiniteDomainState], list[HoleDomain]] +DEFAULT_QUERY_ORDER = "smallest-domain-then-hole-id-v1" + + +def default_query_order(state: FiniteDomainState) -> list[HoleDomain]: + """Unresolved holes: smallest live domain first, then canonical ``HoleId``.""" + unresolved = [hole for hole in state.holes if len(hole.values) > 1] + return sorted(unresolved, key=lambda hole: (len(hole.values), hole.hole_id.sort_key)) + + +class _MutCounters: + __slots__ = ( + "passes", "support_queries", "cache_hits", "supported", "unsupported", + "unknown", "candidates_removed", "verifier_calls", "expanded_nodes", + ) + + def __init__(self) -> None: + for name in self.__slots__: + setattr(self, name, 0) + + def snapshot(self) -> ClosureCounters: + return ClosureCounters(**{name: getattr(self, name) for name in self.__slots__}) + + +def _cache_key( + state: FiniteDomainState, hole_id: HoleId, value: DomainValue, backend_version: str +) -> str: + # state.fingerprint already subsumes pack/constraint/bounds/hole domains. + return _canonical_json( + [state.fingerprint, hole_id.to_dict(), value.to_dict(), backend_version] + ) + + +# --------------------------------------------------------------------------- # +# Exact closure +# --------------------------------------------------------------------------- # + + +def exact_closure( + state: FiniteDomainState, + provider: SupportProvider, + *, + query_order: QueryOrder = default_query_order, + cache: dict[str, SupportResult] | None = None, + certificate_store: dict[str, SupportCertificate] | None = None, + max_queries: int | None = None, +) -> ClosureResult: + """Deterministic monotone fixed point that removes only certified candidates.""" + if not isinstance(state, FiniteDomainState): + raise ValueError("exact_closure requires a FiniteDomainState") + counters = _MutCounters() + deductions: list[CertifiedDeduction] = [] + unknown_queries: list[SupportQuery] = [] + witnesses: list[WitnessRef] = [] + current = state + stop_reason: str | None = None + reached = False + + while True: + if current.is_bottom: + reached = True # certified bottom (only certified removals got us here) + break + ordered = query_order(current) + if not ordered: # every domain singleton + reached = True + break + counters.passes += 1 + pass_removals: dict[HoleId, tuple[tuple[DomainValue, ...], tuple[str, ...]]] = {} + budget_hit = False + + for hole in ordered: + removed: list[DomainValue] = [] + cert_ids: list[str] = [] + for value in hole.values: + if max_queries is not None and counters.support_queries >= max_queries: + budget_hit = True + break + query = SupportQuery( + state_fingerprint=current.fingerprint, + hole_id=hole.hole_id, + candidate=value, + ) + key = _cache_key(current, hole.hole_id, value, provider.backend_version) + if cache is not None and key in cache: + result = cache[key] + counters.cache_hits += 1 + else: + result = provider.check(current, query) + counters.support_queries += 1 + counters.verifier_calls += result.counters.verifier_calls + counters.expanded_nodes += result.counters.nodes + if cache is not None: + cache[key] = result + + if result.verdict is SupportVerdict.SUPPORTED: + counters.supported += 1 + witnesses.append( + WitnessRef( + hole_id=hole.hole_id, + value=value, + certificate_id=result.certificate.digest, + witness_digest=result.certificate.witness_digest, + witness_source=result.certificate.witness_source, + ) + ) + if certificate_store is not None: + certificate_store[result.certificate.digest] = result.certificate + elif result.verdict is SupportVerdict.UNSUPPORTED: + counters.unsupported += 1 + # Replay/validate against the exact pre-refinement state before + # removing anything. A stale or tampered proof does not remove. + replay = provider.replay(result.certificate, state=current) + if replay.ok and replay.verdict is SupportVerdict.UNSUPPORTED: + removed.append(value) + cert_ids.append(result.certificate.digest) + if certificate_store is not None: + certificate_store[result.certificate.digest] = result.certificate + else: + counters.unknown += 1 + unknown_queries.append(query) + else: # UNKNOWN + counters.unknown += 1 + unknown_queries.append(query) + if removed: + retained = tuple(v for v in hole.values if v not in set(removed)) + pass_removals[hole.hole_id] = (tuple(removed), tuple(cert_ids)) + del retained # retained is recomputed below from the pass-start state + if budget_hit: + break + + if not pass_removals: + reached = not budget_hit + if budget_hit: + stop_reason = "budget:max_queries" + break + + # Apply every deduction against the single pass-start state. + before_fp = current.fingerprint + new_state = current + for hole_id in sorted(pass_removals, key=lambda h: h.sort_key): + removed, _cert_ids = pass_removals[hole_id] + live = current.domain(hole_id).values + retained = tuple(v for v in live if v not in set(removed)) + new_state = new_state.refine(hole_id, retained) + after_fp = new_state.fingerprint + for hole_id in sorted(pass_removals, key=lambda h: h.sort_key): + removed, cert_ids_t = pass_removals[hole_id] + counters.candidates_removed += len(removed) + deductions.append( + CertifiedDeduction( + before_fingerprint=before_fp, + after_fingerprint=after_fp, + hole_id=hole_id, + removed=removed, + certificate_ids=cert_ids_t, + reason="certified_unsupported", + ) + ) + current = new_state + + if budget_hit: + stop_reason = "budget:max_queries" + reached = False + break + + return ClosureResult( + state=current, + deductions=tuple(deductions), + unknown_queries=tuple(unknown_queries), + witnesses=tuple(witnesses), + counters=counters.snapshot(), + reached_fixed_point=reached, + stop_reason=stop_reason, + ) diff --git a/tests/test_dsl/test_solver_closure.py b/tests/test_dsl/test_solver_closure.py new file mode 100644 index 00000000..eca7d24e --- /dev/null +++ b/tests/test_dsl/test_solver_closure.py @@ -0,0 +1,326 @@ +"""VSS1-01 (SLM-61): certificate-checked exact closure over finite domains. + +Closure is a pure orchestration layer over the VSS0-04 oracle, so it is tested +against a stub :class:`SupportProvider` whose verdicts are an exact function of the +current state — this pins multi-pass propagation, stale/tampered-certificate +rejection, certified bottom, partial-budget stops, determinism, and cache identity +without depending on the oracle's search. Torch-free. +""" + +from __future__ import annotations + +import hashlib +import inspect + +from slm_training.dsl.solver.state import ( + DomainValue, + FiniteDomainState, + HoleDomain, + HoleId, + SolverBounds, + SupportVerdict, +) +from slm_training.dsl.solver.support import ( + ReplayResult, + SupportCertificate, + SupportQuery, + SupportResult, + SearchCounters, +) +from slm_training.dsl.solver.closure import ( + default_query_order, + exact_closure, +) + +_BOUNDS = SolverBounds( + max_tokens=1000, max_nodes=1000, max_depth=32, max_backtracks=1000, + max_verifier_calls=1000, +) + + +def _sha(text: str) -> str: + return hashlib.sha256(text.encode()).hexdigest() + + +def _val(name: str) -> DomainValue: + return DomainValue.create("v", {"v": name}) + + +def _hole(name: str) -> HoleId: + return HoleId(namespace="closure-test", path=(name,), kind="slot") + + +def _state(domains: dict[str, list[str]], *, constraint_version: str = "v1") -> FiniteDomainState: + holes = tuple( + HoleDomain(_hole(name), tuple(_val(v) for v in values)) + for name, values in domains.items() + ) + return FiniteDomainState( + problem_id="closure-test", + pack_id="fixture", + constraint_version=constraint_version, + bounds=_BOUNDS, + holes=holes, + ) + + +class _RuleProvider: + """Stub oracle whose verdict is an exact function of the current state. + + Certificates are canonical for ``(state, query, verdict)``, so an honest + replay reproduces them and any tamper/stale mismatch is caught. + """ + + def __init__(self, rule, *, profile: str = "stub-v1"): + self._rule = rule + self._profile = profile + + @property + def backend_version(self) -> str: + return f"stub/{self._profile}" + + def _cert(self, state, query, verdict) -> SupportCertificate: + common = dict( + schema_version=1, query=query, verdict=verdict, + problem_id=state.problem_id, pack_id=state.pack_id, + constraint_version=state.constraint_version, bounds=state.bounds, + search_order="canonical-domain-value-v1", + explored_state_fingerprints=(), verifier_profile=self._profile, + ) + if verdict is SupportVerdict.UNSUPPORTED: + return SupportCertificate( + **common, coverage_observations=("complete",), exhausted=True, + ) + if verdict is SupportVerdict.SUPPORTED: + witness = f"witness:{query.candidate.payload_json}" + return SupportCertificate( + **common, coverage_observations=("complete",), + witness_source="stub", witness_digest=_sha(witness), exhausted=False, + ) + return SupportCertificate( + **common, coverage_observations=("partial",), exhausted=False, + stop_reason="incomplete", + ) + + def check(self, state, query) -> SupportResult: + verdict = self._rule(state, query.hole_id, query.candidate) + cert = self._cert(state, query, verdict) + witness = ( + f"witness:{query.candidate.payload_json}" + if verdict is SupportVerdict.SUPPORTED + else None + ) + return SupportResult(verdict, cert, witness=witness, counters=SearchCounters(nodes=1)) + + def replay(self, certificate, *, state) -> ReplayResult: + violations: list[str] = [] + if state.fingerprint != certificate.query.state_fingerprint: + violations.append("stale fingerprint") + recomputed = self._rule(state, certificate.query.hole_id, certificate.query.candidate) + if recomputed != certificate.verdict: + violations.append("verdict changed") + if certificate.verdict is SupportVerdict.UNSUPPORTED: + if not certificate.exhausted: + violations.append("not exhausted") + if any(c in {"partial", "none"} for c in certificate.coverage_observations): + violations.append("incomplete coverage") + expected = self._cert(state, certificate.query, recomputed) + if expected.digest != certificate.digest: + violations.append("digest mismatch") + return ReplayResult(ok=not violations, verdict=certificate.verdict, violations=tuple(violations)) + + +def _v(value: DomainValue) -> str: + return value.payload["v"] + + +# --------------------------------------------------------------------------- # +# Multi-pass propagation +# --------------------------------------------------------------------------- # + + +def _propagation_rule(state, hole_id, value): + name = _v(value) + if name in {"a1", "b1"}: + return SupportVerdict.UNSUPPORTED + if name in {"a3", "b2"}: + return SupportVerdict.SUPPORTED + if name == "a2": + b = next(h for h in state.holes if h.hole_id == _hole("B")) + b1_present = any(_v(dv) == "b1" for dv in b.values) + return SupportVerdict.UNKNOWN if b1_present else SupportVerdict.UNSUPPORTED + return SupportVerdict.UNKNOWN + + +def test_multipass_propagation_removes_only_after_earlier_reduction(): + state = _state({"A": ["a1", "a2", "a3"], "B": ["b1", "b2"]}) + result = exact_closure(state, _RuleProvider(_propagation_rule)) + assert result.reached_fixed_point + live = {h.hole_id.path[0]: sorted(_v(v) for v in h.values) for h in result.state.holes} + assert live == {"A": ["a3"], "B": ["b2"]} + # a2 was removed in a *later* pass than a1/b1 (needs >= 2 passes). + assert result.counters.passes >= 2 + # a2 was UNKNOWN in pass 1 and UNSUPPORTED in pass 2 -> it appears in a later + # deduction, and was recorded unknown at least once. + removed_names = {_v(v) for d in result.deductions for v in d.removed} + assert removed_names == {"a1", "b1", "a2"} + assert any(q.candidate.payload["v"] == "a2" for q in result.unknown_queries) + # SUPPORTED witnesses are kept and recorded. + assert {w.value.payload["v"] for w in result.witnesses} >= {"a3", "b2"} + + +def test_every_removed_value_has_a_certificate_and_state_is_subset(): + state = _state({"A": ["a1", "a2", "a3"], "B": ["b1", "b2"]}) + result = exact_closure(state, _RuleProvider(_propagation_rule)) + orig = {_v(v) for h in state.holes for v in h.values} + kept = {_v(v) for h in result.state.holes for v in h.values} + assert kept <= orig # monotone subset + for deduction in result.deductions: + assert len(deduction.certificate_ids) == len(deduction.removed) + assert all(cid for cid in deduction.certificate_ids) + + +# --------------------------------------------------------------------------- # +# Honesty: UNKNOWN never removes; tampered / stale proofs never mutate state +# --------------------------------------------------------------------------- # + + +def test_unknown_and_supported_are_kept(): + # Everything is UNKNOWN or SUPPORTED -> no removals, immediate fixed point. + def rule(state, hole_id, value): + return SupportVerdict.SUPPORTED if _v(value) == "keep" else SupportVerdict.UNKNOWN + + state = _state({"A": ["keep", "maybe", "dunno"]}) + result = exact_closure(state, _RuleProvider(rule)) + assert result.reached_fixed_point + assert result.state == state # unchanged + assert not result.deductions + assert result.counters.candidates_removed == 0 + + +def test_tampered_unsupported_certificate_cannot_change_state(): + # Verdict says UNSUPPORTED but the certificate is forged (exhausted=False), + # so replay fails and nothing is removed. + class _Dishonest(_RuleProvider): + def check(self, state, query): + result = super().check(state, query) + if result.verdict is SupportVerdict.UNSUPPORTED: + from dataclasses import replace + forged = replace(result.certificate, exhausted=False) + return SupportResult(result.verdict, forged, counters=result.counters) + return result + + state = _state({"A": ["a1", "a3"]}) # a1 unsupported, a3 supported + result = exact_closure(state, _Dishonest(_propagation_rule)) + assert result.state == state # identity unchanged + assert not result.deductions + # The value was seen as unsupported but demoted to unknown because replay failed. + assert result.counters.candidates_removed == 0 + assert any(q.candidate.payload["v"] == "a1" for q in result.unknown_queries) + + +def test_stale_certificate_is_rejected_by_replay(): + provider = _RuleProvider(_propagation_rule) + s0 = _state({"A": ["a1", "a3"], "B": ["b1", "b2"]}) + query = SupportQuery(s0.fingerprint, _hole("A"), _val("a1")) + result = provider.check(s0, query) + # Replay the s0 certificate against a *different* state -> stale. + s1 = _state({"A": ["a1", "a3"], "B": ["b2"]}) + replay = provider.replay(result.certificate, state=s1) + assert not replay.ok + assert any("stale" in v for v in replay.violations) + + +# --------------------------------------------------------------------------- # +# Certified bottom + partial budget +# --------------------------------------------------------------------------- # + + +def test_all_unsupported_yields_certified_bottom(): + def rule(state, hole_id, value): + return SupportVerdict.UNSUPPORTED + + state = _state({"A": ["a1", "a2"]}) + result = exact_closure(state, _RuleProvider(rule)) + assert result.state.is_bottom + assert result.reached_fixed_point + assert result.counters.candidates_removed == 2 + + +def test_partial_budget_stops_with_live_domains(): + def rule(state, hole_id, value): + return SupportVerdict.UNSUPPORTED if _v(value) == "a1" else SupportVerdict.UNKNOWN + + state = _state({"A": ["a1", "a2", "a3"], "B": ["b1", "b2", "b3"]}) + result = exact_closure(state, _RuleProvider(rule), max_queries=1) + assert not result.reached_fixed_point + assert result.stop_reason == "budget:max_queries" + assert result.counters.support_queries <= 1 + # Domains remain live. + assert any(len(h.values) > 1 for h in result.state.holes) + + +# --------------------------------------------------------------------------- # +# Determinism + cache identity +# --------------------------------------------------------------------------- # + + +def test_closure_is_deterministic(): + state = _state({"A": ["a1", "a2", "a3"], "B": ["b1", "b2"]}) + a = exact_closure(state, _RuleProvider(_propagation_rule)) + b = exact_closure(state, _RuleProvider(_propagation_rule)) + assert a.state == b.state + assert [d.to_dict() for d in a.deductions] == [d.to_dict() for d in b.deductions] + assert a.counters.to_dict() == b.counters.to_dict() + + +def test_cache_reuses_results_by_full_identity(): + state = _state({"A": ["a1", "a2", "a3"], "B": ["b1", "b2"]}) + cache: dict = {} + first = exact_closure(state, _RuleProvider(_propagation_rule), cache=cache) + assert first.counters.cache_hits == 0 + # A second closure over the same state + a shared cache reuses stored results. + second = exact_closure(state, _RuleProvider(_propagation_rule), cache=cache) + assert second.counters.cache_hits > 0 + assert second.state == first.state + # A different backend version must miss the cache (identity includes it). + third = exact_closure( + state, _RuleProvider(_propagation_rule, profile="other-vX"), cache=cache + ) + assert third.counters.support_queries > 0 + + +def test_default_query_order_is_smallest_domain_first(): + state = _state({"A": ["a1", "a2", "a3"], "B": ["b1", "b2"], "C": ["c1"]}) + order = default_query_order(state) + names = [h.hole_id.path[0] for h in order] + assert names == ["B", "A"] # C is singleton (resolved), B(2) before A(3) + + +def test_certificate_store_collects_full_certificates(): + state = _state({"A": ["a1", "a3"]}) + store: dict = {} + result = exact_closure(state, _RuleProvider(_propagation_rule), certificate_store=store) + # Every deduction certificate id resolves to a stored full certificate. + for deduction in result.deductions: + for cid in deduction.certificate_ids: + assert cid in store + assert store[cid].digest == cid + + +def test_records_round_trip_json(): + state = _state({"A": ["a1", "a3"], "B": ["b1", "b2"]}) + result = exact_closure(state, _RuleProvider(_propagation_rule)) + import json + + for deduction in result.deductions: + blob = json.dumps(deduction.to_dict(), sort_keys=True) + assert "before_fingerprint" in json.loads(blob) + for witness in result.witnesses: + assert "certificate_id" in witness.to_dict() + + +def test_solver_closure_is_torch_free(): + import slm_training.dsl.solver.closure as closure_mod + + assert "torch" not in inspect.getsource(closure_mod)