From 980712965a38456de44366df8d4746bd4ada76f7 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:39:00 -0700 Subject: [PATCH 1/2] fix(processor): bound the complete solver operation (#1108) --- ...e-1108-asr-530-operation-bounded-solver.md | 39 ++ ...4-solver-operational-safety-remediation.md | 94 +++++ docs/requirements/ASR-530/requirement.md | 6 + .../raes_processor/satisfiability/_service.py | 40 +- .../raes_processor/satisfiability/_solver.py | 244 ++++++++++-- .../tests/test_scenario_satisfiability.py | 365 ++++++++++++++++++ .../formal/scenario-satisfiability/README.md | 28 +- 7 files changed, 774 insertions(+), 42 deletions(-) create mode 100644 docs/decisions/issue-1108-asr-530-operation-bounded-solver.md create mode 100644 docs/decisions/issue-1114-solver-operational-safety-remediation.md diff --git a/docs/decisions/issue-1108-asr-530-operation-bounded-solver.md b/docs/decisions/issue-1108-asr-530-operation-bounded-solver.md new file mode 100644 index 000000000..a34a9ab43 --- /dev/null +++ b/docs/decisions/issue-1108-asr-530-operation-bounded-solver.md @@ -0,0 +1,39 @@ +# Issue 1108 / ASR-530 Operation-Bounded Solver + +Date: 2026-08-11 + +Issue: #1108. Requirement: ASR-530. Related: #1114 and closed issue #826. + +## Decision + +One satisfiability analysis builds one deterministic incremental QF_LIA solver +session. Each normalized clause is encoded once behind a Boolean assumption. +The initial decision, canonical witness probes, and sorted-deletion core probes +select clauses and fixed domain indices through assumptions; they do not rebuild +the variable and expression graph. + +A monotonic 5000 ms operation deadline starts before solver construction and +covers expression construction, every Z3 call, and deterministic result +selection. Each check receives the remaining operation time, rounded up to a +positive millisecond and capped at the governed timeout. A result that returns +after the deadline is an operational failure, never SAT or UNSAT evidence. +The existing derived check-count budget remains an independent cardinality +bound. + +## Compatibility and Nonclaims + +The normalized theory, solver package/logic/options, initial decision, canonical +lexicographic witness order, sorted-deletion subset-minimal core order, evidence +schema, and replay digest remain unchanged. The configuration field +`timeout_ms=5000` now honestly bounds the whole solver operation rather than +each individual check. This is an in-process resource bound; it is not a claim +of hard process isolation against a compromised native Z3 library. + +## Verification + +Differential property tests compare complete SAT assignments and UNSAT cores +with the incumbent rebuilding algorithm over generated bounded models. Tests +count one solver construction, force expiry during construction, before a +check, inside repeated selection, and immediately after a nominal Z3 result, +and verify that no partial evidence crosses the service boundary. A high-check +shape guards against renewed per-check graph construction. diff --git a/docs/decisions/issue-1114-solver-operational-safety-remediation.md b/docs/decisions/issue-1114-solver-operational-safety-remediation.md new file mode 100644 index 000000000..913dd7849 --- /dev/null +++ b/docs/decisions/issue-1114-solver-operational-safety-remediation.md @@ -0,0 +1,94 @@ +# Issue 1114 Solver Operational-Safety Remediation + +Date: 2026-08-11 + +Issue: #1114. Requirement: ASR-530. Related lineage: closed issue #826. + +This note records the bounded operational-safety correction without modifying +the immutable issue-826 preflight. It does not change the v1 theory, +solver package pin, solver configuration contract, witness order, or +core-reduction order. + +## Gap Claim + +Z3 may return `unknown` under the configured per-check timeout. The witness and +core loops compared only against their desired decisive result, so `unknown` +could silently skip a canonical witness candidate or retain a removable core +clause while the emitted evidence still claimed the governed selection rule. +Duplicate clause ids also reached tracked assumptions through a collapsing +dictionary, and empty memberships relied on an implicit zero-argument `Or`. + +## Existing Surface Audit + +The normalized contract already requires unique clause ids and finite bounded +symbols, domains, and clauses. The service already translates adapter failure +to `SatisfiabilityOperationalError`, and the evidence contract already records +the 5000 ms per-check timeout. The shared `_check()` seam is used by the initial +decision, witness selection, and core reduction, making it the single complete +place to reject `unknown`. + +## Lineage And Precedent + +ADR-086 and the normative satisfiability specification distinguish completed +SAT/UNSAT outcomes from operational failure. The issue-826 preflight explicitly +forbids mapping unknown, timeout, or exhaustion to a semantic outcome and +requires bounded repeated checks for canonical witnesses and reduced cores. + +## Literature And Practice + +SMT solver APIs expose `sat`, `unsat`, and `unknown` as distinct results; a +resource-limited `unknown` is not evidence for either decision. Deterministic +core reduction and lexicographic witness selection therefore require every +probe to complete decisively before their stronger labels may be published. + +## Alternatives Considered + +1. **Retain the old branch behavior.** Rejected because it can publish an + overclaim that appears only as later replay drift. +2. **Map unknown to `unsupported`.** Rejected because translation coverage is + complete; solver non-completion is an operational failure. +3. **Retry unknown automatically.** Rejected because unrecorded retries change + the operational profile and can make latency unbounded. +4. **Add a caller-configurable check limit to the v1 wire contract.** Rejected + as an unnecessary profile/schema change. +5. **Reject unknown centrally and derive a finite check budget from the bound + normalized model.** Chosen. + +## Chosen Architecture + +Before constructing tracked assumptions, the adapter defensively rejects +duplicate clause ids. Empty membership is explicitly `false`. `_check()` raises +a structured `SolverOperationalError` for `unknown`, recording phase, check +count, derived check budget, the governed 5000 ms timeout, and Z3's bounded +reason string. The service preserves those safe fields on +`SatisfiabilityOperationalError`; it emits no partial evidence. + +For model symbols `S`, domains `D(s)`, and clauses `C`, the run budget is: + +```text +B = 1 + max(|C|, sum(|D(s)| for s in S)) +``` + +One check is the initial decision. Only one of the bounded branches then runs: +at most one deletion probe per clause, or at most one feasibility probe per +domain member. The budget is thus derived from digest-bound normalized input, +not a hidden configuration knob, and its maximum is bounded by the published +contract cardinalities. + +## Documentation Defense + +The normative solver section now states the decisive-result rule, explicit +empty-membership semantics, derived budget, and observable operational-error +fields. No schema is regenerated because completed evidence is unchanged and +an operational failure emits no evidence envelope. + +## Verification Plan + +- Force the first or second solver call to return unknown and prove that no + outcome, witness, or core evidence is emitted. +- Assert phase, call count, budget, timeout, and reason survive the service + boundary. +- Inject duplicate clause identity past model validation and require a bounded + adapter error rather than a raw Z3 exception. +- Exercise an empty target domain as explicit unsatisfiability and exhaust a + synthetic zero-check budget before solver construction. diff --git a/docs/requirements/ASR-530/requirement.md b/docs/requirements/ASR-530/requirement.md index fb9a9d0f4..7a0407f74 100644 --- a/docs/requirements/ASR-530/requirement.md +++ b/docs/requirements/ASR-530/requirement.md @@ -52,5 +52,11 @@ Agent-assisted development can produce internally coherent code and documentatio - TESTS → TEST `implementations/python/tests/test_scenario_satisfiability.py` (Scenario satisfiability analysis and evidence tests) - TESTS → TEST `implementations/python/tests/test_satisfiability_cli.py` (Governed satisfiability CLI tests) - IMPLEMENTS → GITHUB_ISSUE `826` (Governed whole-scenario constraint satisfiability and solver evidence) +- IMPLEMENTS → GITHUB_ISSUE `1114` (Fail-closed incomplete solver checks and defensive solver-boundary validation) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1114-solver-operational-safety-remediation.md` (Incomplete-check operational-safety remediation) +- DOCUMENTS → GITHUB_ISSUE `1108` (Complete-operation satisfiability deadline) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1108-asr-530-operation-bounded-solver.md` (Incremental solver and monotonic deadline decision) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_processor/satisfiability/_solver.py` (Single-session solver and operation-wide deadline) +- TESTS → TEST `implementations/python/tests/test_scenario_satisfiability.py` (Differential, timeout, and construction-count regressions) - IMPLEMENTS → PROOF `docs/research/formal-semantic-validation/bundles/retest-v2.json` (Formal semantic validation atomic retest evidence release v2) - IMPLEMENTS → GITHUB_ISSUE `828` (Re-test formal semantic validation, satisfiability, and exploit-path claims) diff --git a/implementations/python/packages/raes_processor/satisfiability/_service.py b/implementations/python/packages/raes_processor/satisfiability/_service.py index f65a1cdce..14c12f662 100644 --- a/implementations/python/packages/raes_processor/satisfiability/_service.py +++ b/implementations/python/packages/raes_processor/satisfiability/_service.py @@ -27,7 +27,7 @@ canonical_contract_digest, ) -from ._solver import SolverOperationalError, solve_model +from ._solver import SOLVER_TIMEOUT_MS, SolverOperationalError, solve_model from ._translation import translate_scenario ANALYSIS_PROFILE = "raes-finite-domain-satisfiability-v1" @@ -40,6 +40,33 @@ class SatisfiabilityEvidenceError(ValueError): class SatisfiabilityOperationalError(RuntimeError): """The production analyzer failed outside the typed outcome domain.""" + def __init__( + self, + message: str, + *, + solver_phase: str | None = None, + solver_check_count: int | None = None, + solver_check_budget: int | None = None, + solver_timeout_ms: int | None = None, + solver_reason: str | None = None, + ) -> None: + self.solver_phase = solver_phase + self.solver_check_count = solver_check_count + self.solver_check_budget = solver_check_budget + self.solver_timeout_ms = solver_timeout_ms + self.solver_reason = solver_reason + details = [] + if solver_phase is not None: + details.append(f"phase={solver_phase}") + if solver_check_count is not None and solver_check_budget is not None: + details.append(f"check={solver_check_count}/{solver_check_budget}") + if solver_timeout_ms is not None: + details.append(f"timeout_ms={solver_timeout_ms}") + if solver_reason is not None: + details.append(f"reason={solver_reason}") + rendered = f"{message} ({', '.join(details)})" if details else message + super().__init__(rendered) + def analyze_scenario_file( path: Path, @@ -86,7 +113,14 @@ def analyze_scenario_file( try: result = solve_model(translation.model) except SolverOperationalError as exc: - raise SatisfiabilityOperationalError("the pinned solver did not complete") from exc + raise SatisfiabilityOperationalError( + "the pinned solver did not complete", + solver_phase=exc.phase, + solver_check_count=exc.check_count, + solver_check_budget=exc.check_budget, + solver_timeout_ms=exc.timeout_ms, + solver_reason=exc.reason, + ) from exc if result.outcome is SatisfiabilityOutcome.SATISFIABLE: assert result.assignment is not None instantiated = instantiate_scenario( @@ -149,7 +183,7 @@ def _solver_configuration() -> SolverConfigurationModel: engine_version=engine_version, logic="QF_LIA", random_seed=0, - timeout_ms=5000, + timeout_ms=SOLVER_TIMEOUT_MS, threads=1, auto_config=False, model=True, diff --git a/implementations/python/packages/raes_processor/satisfiability/_solver.py b/implementations/python/packages/raes_processor/satisfiability/_solver.py index 41ecb4419..10ad2ce71 100644 --- a/implementations/python/packages/raes_processor/satisfiability/_solver.py +++ b/implementations/python/packages/raes_processor/satisfiability/_solver.py @@ -2,15 +2,89 @@ from __future__ import annotations -from dataclasses import dataclass +import time +from dataclasses import dataclass, field +from math import ceil +from typing import Literal import z3 from raes_contracts.satisfiability import NormalizedConstraintModel, SatisfiabilityOutcome +SOLVER_TIMEOUT_MS: Literal[5000] = 5000 +_MAX_OPERATIONAL_REASON_CHARS = 256 +_NANOSECONDS_PER_MILLISECOND = 1_000_000 + + +def _monotonic_ns() -> int: + return time.monotonic_ns() + class SolverOperationalError(RuntimeError): """The pinned adapter could not produce a completed governed outcome.""" + def __init__( + self, + message: str, + *, + phase: str, + check_count: int | None = None, + check_budget: int | None = None, + reason: str | None = None, + ) -> None: + self.phase = phase + self.check_count = check_count + self.check_budget = check_budget + self.timeout_ms = SOLVER_TIMEOUT_MS + self.reason = reason + details = [f"phase={phase}", f"timeout_ms={SOLVER_TIMEOUT_MS}"] + if check_count is not None and check_budget is not None: + details.append(f"check={check_count}/{check_budget}") + if reason is not None: + details.append(f"reason={reason}") + super().__init__(f"{message} ({', '.join(details)})") + + +@dataclass +class _CheckBudget: + """Check-count and monotonic wall-time bounds for one solver operation.""" + + max_checks: int + checks_used: int = 0 + started_ns: int = field(default_factory=_monotonic_ns) + deadline_ns: int = field(init=False) + + def __post_init__(self) -> None: + self.deadline_ns = self.started_ns + SOLVER_TIMEOUT_MS * _NANOSECONDS_PER_MILLISECOND + + def checkpoint(self, phase: str, *, check_count: int | None = None) -> None: + if _monotonic_ns() >= self.deadline_ns: + raise SolverOperationalError( + "solver operation deadline exhausted", + phase=phase, + check_count=self.checks_used if check_count is None else check_count, + check_budget=self.max_checks, + reason="operation-deadline-exhausted", + ) + + def remaining_timeout_ms(self, phase: str, *, check_count: int) -> int: + remaining_ns = self.deadline_ns - _monotonic_ns() + if remaining_ns <= 0: + self.checkpoint(phase, check_count=check_count) + return min(SOLVER_TIMEOUT_MS, max(1, ceil(remaining_ns / _NANOSECONDS_PER_MILLISECOND))) + + def consume(self, phase: str) -> int: + self.checkpoint(phase) + if self.checks_used >= self.max_checks: + raise SolverOperationalError( + "solver check budget exhausted", + phase=phase, + check_count=self.checks_used, + check_budget=self.max_checks, + reason="derived-check-budget-exhausted", + ) + self.checks_used += 1 + return self.checks_used + @dataclass(frozen=True) class SolverResult: @@ -22,82 +96,176 @@ class SolverResult: def solve_model(model: NormalizedConstraintModel) -> SolverResult: """Solve one normalized model and select deterministic portable evidence.""" + _require_unique_clause_ids(model) all_clause_ids = tuple(clause.clause_id for clause in model.clauses) - status = _check(model, all_clause_ids, {}) - if status == z3.sat: + budget = _solver_check_budget(model) + session = _SolverSession(model, budget) + # ``_check`` fails loudly on ``z3.unknown``, so a non-satisfiable decision + # here is decisively unsatisfiable rather than an incomplete timeout. + if session.check(all_clause_ids, {}, phase="initial-decision") == z3.sat: return SolverResult( outcome=SatisfiabilityOutcome.SATISFIABLE, - assignment=_select_witness(model, all_clause_ids), + assignment=_select_witness(model, all_clause_ids, session), ) - if status == z3.unsat: - return SolverResult( - outcome=SatisfiabilityOutcome.UNSATISFIABLE, - core=_reduce_unsat_core(model, all_clause_ids), + return SolverResult( + outcome=SatisfiabilityOutcome.UNSATISFIABLE, + core=_reduce_unsat_core(all_clause_ids, session), + ) + + +def _require_unique_clause_ids(model: NormalizedConstraintModel) -> None: + """Reject collapsed clause identity before tracked-assumption construction.""" + + clause_ids = [clause.clause_id for clause in model.clauses] + if len(set(clause_ids)) != len(clause_ids): + raise SolverOperationalError( + "normalized model contains duplicate clause ids", + phase="model-validation", + reason="duplicate-clause-id", ) - raise SolverOperationalError("solver returned an incomplete result") + + +def _solver_check_budget(model: NormalizedConstraintModel) -> _CheckBudget: + """Bound the two possible finite algorithms without a second profile knob.""" + + witness_checks = sum(len(symbol.domain) for symbol in model.symbols) + core_checks = len(model.clauses) + return _CheckBudget(max_checks=1 + max(witness_checks, core_checks)) def _select_witness( model: NormalizedConstraintModel, all_clause_ids: tuple[str, ...], + session: _SolverSession, ) -> dict[str, str | int | bool]: fixed: dict[str, int] = {} assignment: dict[str, str | int | bool] = {} for symbol in model.symbols: for index, value in enumerate(symbol.domain): candidate = {**fixed, symbol.symbol_id: index} - if _check(model, all_clause_ids, candidate) == z3.sat: + if ( + session.check( + all_clause_ids, + candidate, + phase="witness-selection", + ) + == z3.sat + ): fixed[symbol.symbol_id] = index assignment[symbol.variable] = value break else: # This is guarded by the initial satisfiable result. - raise SolverOperationalError("deterministic witness selection failed") + raise SolverOperationalError( + "deterministic witness selection failed", + phase="witness-selection", + check_count=session.budget.checks_used, + check_budget=session.budget.max_checks, + reason="no-feasible-domain-member-after-sat", + ) return assignment def _reduce_unsat_core( - model: NormalizedConstraintModel, all_clause_ids: tuple[str, ...], + session: _SolverSession, ) -> tuple[str, ...]: core = list(all_clause_ids) for clause_id in all_clause_ids: candidate = tuple(item for item in core if item != clause_id) - if _check(model, candidate, {}) == z3.unsat: + if session.check(candidate, {}, phase="core-reduction") == z3.unsat: core = list(candidate) return tuple(sorted(core)) +class _SolverSession: + """One incrementally queried expression graph for one normalized model.""" + + def __init__(self, model: NormalizedConstraintModel, budget: _CheckBudget) -> None: + self.budget = budget + budget.checkpoint("model-construction", check_count=budget.checks_used) + self.solver = z3.SolverFor("QF_LIA") + self.solver.set( + random_seed=0, + timeout=SOLVER_TIMEOUT_MS, + threads=1, + auto_config=False, + model=True, + unsat_core=True, + ) + self.variables: dict[str, z3.ArithRef] = {} + for symbol in model.symbols: + self.variables[symbol.symbol_id] = z3.Int(_z3_name(symbol.symbol_id)) + budget.checkpoint("model-construction", check_count=budget.checks_used) + self.trackers: dict[str, z3.BoolRef] = {} + symbols = {item.symbol_id: item for item in model.symbols} + for clause in model.clauses: + symbol = symbols[clause.symbol_id] + indexes = [] + for index, value in enumerate(symbol.domain): + if any(_scalar_equal(value, allowed) for allowed in clause.allowed_values): + indexes.append(index) + budget.checkpoint("model-construction", check_count=budget.checks_used) + expression = ( + z3.Or(*(self.variables[symbol.symbol_id] == index for index in indexes)) + if indexes + else z3.BoolVal(False) + ) + tracker = z3.Bool(_z3_name(clause.clause_id)) + self.trackers[clause.clause_id] = tracker + self.solver.add(z3.Implies(tracker, expression)) + budget.checkpoint("model-construction", check_count=budget.checks_used) + budget.checkpoint("model-construction", check_count=budget.checks_used) + + def check( + self, + clause_ids: tuple[str, ...], + fixed: dict[str, int], + *, + phase: str, + ) -> z3.CheckSatResult: + check_count = self.budget.consume(phase) + return self._check_consumed(clause_ids, fixed, phase=phase, check_count=check_count) + + def _check_consumed( + self, + clause_ids: tuple[str, ...], + fixed: dict[str, int], + *, + phase: str, + check_count: int, + ) -> z3.CheckSatResult: + assumptions = [self.trackers[clause_id] for clause_id in clause_ids] + assumptions.extend(self.variables[symbol_id] == index for symbol_id, index in fixed.items()) + timeout_ms = self.budget.remaining_timeout_ms(phase, check_count=check_count) + self.solver.set(timeout=timeout_ms) + result = self.solver.check(*assumptions) + self.budget.checkpoint(phase, check_count=check_count) + if result == z3.unknown: + reason = " ".join(self.solver.reason_unknown().split())[:_MAX_OPERATIONAL_REASON_CHARS] or "unspecified" + raise SolverOperationalError( + "solver returned unknown", + phase=phase, + check_count=check_count, + check_budget=self.budget.max_checks, + reason=reason, + ) + return result + + def _check( model: NormalizedConstraintModel, clause_ids: tuple[str, ...], fixed: dict[str, int], + *, + budget: _CheckBudget, + phase: str, ) -> z3.CheckSatResult: - solver = z3.SolverFor("QF_LIA") - solver.set( - random_seed=0, - timeout=5000, - threads=1, - auto_config=False, - model=True, - unsat_core=True, - ) - variables = {symbol.symbol_id: z3.Int(_z3_name(symbol.symbol_id)) for symbol in model.symbols} - clauses = {item.clause_id: item for item in model.clauses} - symbols = {item.symbol_id: item for item in model.symbols} - for clause_id in clause_ids: - clause = clauses[clause_id] - symbol = symbols[clause.symbol_id] - indexes = [ - index - for index, value in enumerate(symbol.domain) - if any(_scalar_equal(value, allowed) for allowed in clause.allowed_values) - ] - expression = z3.Or(*(variables[symbol.symbol_id] == index for index in indexes)) - solver.assert_and_track(expression, z3.Bool(_z3_name(clause_id))) - for symbol_id, index in fixed.items(): - solver.add(variables[symbol_id] == index) - return solver.check() + """Compatibility seam for one governed probe; production reuses a session.""" + + check_count = budget.consume(phase) + session = _SolverSession(model, budget) + return session._check_consumed(clause_ids, fixed, phase=phase, check_count=check_count) def _scalar_equal(left: object, right: object) -> bool: diff --git a/implementations/python/tests/test_scenario_satisfiability.py b/implementations/python/tests/test_scenario_satisfiability.py index 288d464bf..3b7c62acc 100644 --- a/implementations/python/tests/test_scenario_satisfiability.py +++ b/implementations/python/tests/test_scenario_satisfiability.py @@ -5,18 +5,38 @@ import json from copy import deepcopy from pathlib import Path +from types import SimpleNamespace import pytest +import raes_processor.satisfiability._solver as solver_adapter +import z3 +from hypothesis import given, settings +from hypothesis import strategies as st from pydantic import ValidationError from raes_contracts.satisfiability import ( + ConstraintClauseKind, + ConstraintClauseModel, + ConstraintSort, + ConstraintSymbolModel, + NormalizedConstraintModel, SatisfiabilityOutcome, ScenarioSatisfiabilityEvidenceModel, ) from raes_processor.satisfiability import ( SatisfiabilityEvidenceError, + SatisfiabilityOperationalError, analyze_scenario_file, replay_satisfiability_evidence, ) +from raes_processor.satisfiability._solver import ( + SOLVER_TIMEOUT_MS, + SolverOperationalError, + _check, + _CheckBudget, + _select_witness, + _solver_check_budget, + solve_model, +) _SATISFIABLE = """\ name: satisfiable-control @@ -63,6 +83,21 @@ cpu: ${cpu_count} """ +_EMPTY_TARGET_DOMAIN = """\ +name: empty-target-domain +variables: + copies: + type: integer + allowed_values: [0] + required: true +nodes: + target: + type: vm +infrastructure: + target: + count: ${copies} +""" + def _write(tmp_path: Path, name: str, content: str) -> Path: path = tmp_path / name @@ -313,3 +348,333 @@ def test_published_valid_and_invalid_fixtures_enforce_outcome_joins() -> None: ScenarioSatisfiabilityEvidenceModel.model_validate_json(invalid_payload) assert "outcome must select exactly one matching payload" in str(exc_info.value) + + +def _force_unknown_on_call(monkeypatch: pytest.MonkeyPatch, target_call: int) -> None: + """Make the target solver check report unknown as a timeout proxy.""" + + original = z3.Solver.check + state = {"calls": 0} + + def patched(self: z3.Solver, *assumptions: object) -> z3.CheckSatResult: + state["calls"] += 1 + if state["calls"] == target_call: + return z3.unknown + return original(self, *assumptions) + + monkeypatch.setattr(z3.Solver, "check", patched) + + +def test_unknown_during_initial_decision_fails_without_evidence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) + _force_unknown_on_call(monkeypatch, target_call=1) + + with pytest.raises(SatisfiabilityOperationalError) as exc_info: + analyze_scenario_file(source) + + error = exc_info.value + assert error.solver_phase == "initial-decision" + assert error.solver_check_count == 1 + assert error.solver_check_budget is not None and error.solver_check_budget >= 1 + + +def test_unknown_during_core_reduction_fails_loudly_with_budget_context( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source = _write(tmp_path, "unsatisfiable.sdl.yaml", _UNSATISFIABLE) + # Call one is the decisive UNSAT check; call two is the first deletion + # probe, where accepting unknown would forge subset minimality. + _force_unknown_on_call(monkeypatch, target_call=2) + + with pytest.raises(SatisfiabilityOperationalError) as exc_info: + analyze_scenario_file(source) + + error = exc_info.value + assert error.solver_phase == "core-reduction" + assert error.solver_check_count == 2 + assert error.solver_check_budget is not None and error.solver_check_budget >= 2 + assert error.solver_timeout_ms == SOLVER_TIMEOUT_MS + assert error.solver_reason + + +def test_unknown_during_witness_selection_fails_loudly_with_budget_context( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) + # Call one decides SAT; call two probes the canonical first domain value, + # where accepting unknown would forge lexicographic selection. + _force_unknown_on_call(monkeypatch, target_call=2) + monkeypatch.setattr(z3.Solver, "reason_unknown", lambda _solver: "timeout\n" + ("x" * 300)) + + with pytest.raises(SatisfiabilityOperationalError) as exc_info: + analyze_scenario_file(source) + + error = exc_info.value + assert error.solver_phase == "witness-selection" + assert error.solver_check_count == 2 + assert error.solver_check_budget is not None and error.solver_check_budget >= 2 + assert error.solver_timeout_ms == SOLVER_TIMEOUT_MS + assert error.solver_reason is not None + assert len(error.solver_reason) == 256 + assert "\n" not in error.solver_reason + + +def test_operational_errors_render_complete_and_minimal_context() -> None: + detailed = SatisfiabilityOperationalError( + "solver failed", + solver_phase="witness-selection", + solver_check_count=2, + solver_check_budget=5, + solver_timeout_ms=SOLVER_TIMEOUT_MS, + solver_reason="timeout", + ) + assert str(detailed) == ( + f"solver failed (phase=witness-selection, check=2/5, timeout_ms={SOLVER_TIMEOUT_MS}, reason=timeout)" + ) + + minimal_service = SatisfiabilityOperationalError("analysis failed") + assert str(minimal_service) == "analysis failed" + + minimal = SolverOperationalError("solver failed", phase="initial-decision") + assert minimal.reason is None + assert str(minimal) == f"solver failed (phase=initial-decision, timeout_ms={SOLVER_TIMEOUT_MS})" + + +def test_witness_selection_fails_if_a_prior_sat_decision_cannot_be_reproduced(tmp_path: Path) -> None: + source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) + model = analyze_scenario_file(source).normalized_model + session = SimpleNamespace( + check=lambda *_args, **_kwargs: z3.unsat, + budget=SimpleNamespace(checks_used=1, max_checks=2), + ) + + with pytest.raises(SolverOperationalError, match="deterministic witness selection failed") as raised: + _select_witness( + model, + tuple(clause.clause_id for clause in model.clauses), + session, + ) + + assert raised.value.reason == "no-feasible-domain-member-after-sat" + + +def test_duplicate_clause_ids_rejected_at_solver_boundary(tmp_path: Path) -> None: + source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) + model = analyze_scenario_file(source).normalized_model + # ``model_copy`` bypasses contract validation to inject collapsed identity; + # the adapter must reject it before tracked-assumption construction. + duplicated = model.model_copy(update={"clauses": model.clauses + (model.clauses[0],)}) + + with pytest.raises(SolverOperationalError, match="duplicate clause ids") as exc_info: + solve_model(duplicated) + + assert exc_info.value.phase == "model-validation" + assert exc_info.value.reason == "duplicate-clause-id" + + +def test_empty_target_domain_is_explicitly_unsatisfiable(tmp_path: Path) -> None: + source = _write(tmp_path, "empty-target-domain.sdl.yaml", _EMPTY_TARGET_DOMAIN) + + evidence = analyze_scenario_file(source) + + assert evidence.outcome is SatisfiabilityOutcome.UNSATISFIABLE + assert evidence.unsat_core is not None + assert any(clause.allowed_values == () for clause in evidence.normalized_model.clauses) + + +def test_derived_solver_check_budget_fails_closed_before_extra_check(tmp_path: Path) -> None: + source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) + model = analyze_scenario_file(source).normalized_model + + with pytest.raises(SolverOperationalError, match="check budget exhausted") as exc_info: + _check( + model, + tuple(clause.clause_id for clause in model.clauses), + {}, + budget=_CheckBudget(max_checks=0), + phase="test-probe", + ) + + assert exc_info.value.check_count == 0 + assert exc_info.value.check_budget == 0 + assert exc_info.value.reason == "derived-check-budget-exhausted" + + +def test_compatibility_check_seam_runs_one_complete_probe(tmp_path: Path) -> None: + source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) + model = analyze_scenario_file(source).normalized_model + clause_ids = tuple(clause.clause_id for clause in model.clauses) + + result = _check(model, clause_ids, {}, budget=_CheckBudget(max_checks=1), phase="test-probe") + + assert result == z3.sat + + +def test_remaining_timeout_fails_when_no_operation_time_remains(monkeypatch: pytest.MonkeyPatch) -> None: + budget = _CheckBudget(max_checks=1, started_ns=0) + monkeypatch.setattr(solver_adapter.time, "monotonic_ns", lambda: budget.deadline_ns) + + with pytest.raises(SolverOperationalError, match="operation deadline exhausted") as raised: + budget.remaining_timeout_ms("test-probe", check_count=1) + + assert raised.value.phase == "test-probe" + assert raised.value.reason == "operation-deadline-exhausted" + + +def _normalized_model(domain_sizes: list[int], masks_by_symbol: list[list[int]]) -> NormalizedConstraintModel: + symbols = tuple( + ConstraintSymbolModel( + symbol_id=f"symbol:{symbol_index:03d}", + variable=f"value_{symbol_index:03d}", + sort=ConstraintSort.INTEGER, + domain=tuple(sorted(range(domain_size), key=lambda value: str(value).encode("utf-8"))), + ) + for symbol_index, domain_size in enumerate(domain_sizes) + ) + clauses = tuple( + ConstraintClauseModel( + clause_id=f"clause:{symbol_index:03d}:{clause_index:03d}", + kind=ConstraintClauseKind.TARGET_DOMAIN, + symbol_id=symbols[symbol_index].symbol_id, + source_address=f"/nodes/node_{symbol_index:03d}/constraint_{clause_index:03d}", + allowed_values=tuple(value for value in symbol.domain if mask & (1 << value)), + ) + for symbol_index, (symbol, masks) in enumerate(zip(symbols, masks_by_symbol, strict=True)) + for clause_index, mask in enumerate(masks) + ) + return NormalizedConstraintModel( + profile="raes-finite-domain-constraints/v1", + theory_profile="raes-finite-domain-theory/v1", + translation_profile="raes-sdl-authoring-translation/v1", + source_digest="sha256:" + "1" * 64, + authored_digest={ + "profile": "raes-sdl-semantic/v1", + "algorithm": "sha256", + "value": "sha256:" + "2" * 64, + }, + symbols=symbols, + clauses=clauses, + ) + + +@st.composite +def _bounded_models(draw: st.DrawFn) -> NormalizedConstraintModel: + domain_sizes = draw(st.lists(st.integers(min_value=1, max_value=5), min_size=1, max_size=4)) + masks_by_symbol = [] + for domain_size in domain_sizes: + masks_by_symbol.append( + draw( + st.lists( + st.integers(min_value=0, max_value=(1 << domain_size) - 1), + min_size=0, + max_size=4, + ) + ) + ) + return _normalized_model(domain_sizes, masks_by_symbol) + + +def _finite_reference( + model: NormalizedConstraintModel, +) -> tuple[SatisfiabilityOutcome, dict[str, str | int | bool] | tuple[str, ...]]: + clauses = {clause.clause_id: clause for clause in model.clauses} + all_clause_ids = tuple(clauses) + + def feasible(clause_ids: tuple[str, ...], fixed: dict[str, int]) -> bool: + selected = [clauses[clause_id] for clause_id in clause_ids] + for symbol in model.symbols: + indexes = [fixed[symbol.symbol_id]] if symbol.symbol_id in fixed else list(range(len(symbol.domain))) + for clause in selected: + if clause.symbol_id == symbol.symbol_id: + indexes = [index for index in indexes if symbol.domain[index] in clause.allowed_values] + if not indexes: + return False + return True + + if feasible(all_clause_ids, {}): + fixed: dict[str, int] = {} + assignment: dict[str, str | int | bool] = {} + for symbol in model.symbols: + for index, value in enumerate(symbol.domain): + if feasible(all_clause_ids, {**fixed, symbol.symbol_id: index}): + fixed[symbol.symbol_id] = index + assignment[symbol.variable] = value + break + return SatisfiabilityOutcome.SATISFIABLE, assignment + core = list(all_clause_ids) + for clause_id in all_clause_ids: + candidate = tuple(item for item in core if item != clause_id) + if not feasible(candidate, {}): + core = list(candidate) + return SatisfiabilityOutcome.UNSATISFIABLE, tuple(sorted(core)) + + +@given(_bounded_models()) +@settings(max_examples=100, deadline=None) +def test_incremental_solver_matches_finite_reference(model: NormalizedConstraintModel) -> None: + result = solve_model(model) + expected_outcome, expected_evidence = _finite_reference(model) + + assert result.outcome is expected_outcome + assert (result.assignment if result.assignment is not None else result.core) == expected_evidence + + +def test_one_solver_is_constructed_for_all_checks(monkeypatch: pytest.MonkeyPatch) -> None: + model = _normalized_model([4, 4], [[0b1000], [0b1000]]) + original = solver_adapter.z3.SolverFor + constructions = 0 + + def counted(logic: str): + nonlocal constructions + constructions += 1 + return original(logic) + + monkeypatch.setattr(solver_adapter.z3, "SolverFor", counted) + + result = solve_model(model) + + assert result.assignment == {"value_000": 3, "value_001": 3} + assert constructions == 1 + + +def test_published_maximum_witness_shape_has_a_finite_derived_budget() -> None: + model = _normalized_model([256] * 128, [[] for _index in range(128)]) + + assert _solver_check_budget(model).max_checks == 32_769 + + +def test_operation_deadline_covers_model_construction(monkeypatch: pytest.MonkeyPatch) -> None: + clock = iter((0, SOLVER_TIMEOUT_MS * 1_000_000)) + monkeypatch.setattr(solver_adapter.time, "monotonic_ns", clock.__next__) + + with pytest.raises(SolverOperationalError, match="operation deadline exhausted") as raised: + solve_model(_normalized_model([1], [[]])) + + assert raised.value.phase == "model-construction" + assert raised.value.check_count == 0 + assert raised.value.reason == "operation-deadline-exhausted" + + +def test_result_returned_after_operation_deadline_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + clock = {"now": 0} + original_check = z3.Solver.check + + def complete_after_deadline(self: z3.Solver, *assumptions: object) -> z3.CheckSatResult: + result = original_check(self, *assumptions) + clock["now"] = SOLVER_TIMEOUT_MS * 1_000_000 + return result + + monkeypatch.setattr(solver_adapter.time, "monotonic_ns", lambda: clock["now"]) + monkeypatch.setattr(z3.Solver, "check", complete_after_deadline) + + with pytest.raises(SolverOperationalError, match="operation deadline exhausted") as raised: + solve_model(_normalized_model([1], [[]])) + + assert raised.value.phase == "initial-decision" + assert raised.value.check_count == 1 + assert raised.value.reason == "operation-deadline-exhausted" diff --git a/specs/formal/scenario-satisfiability/README.md b/specs/formal/scenario-satisfiability/README.md index 36d6375ce..faae1268e 100644 --- a/specs/formal/scenario-satisfiability/README.md +++ b/specs/formal/scenario-satisfiability/README.md @@ -69,11 +69,37 @@ silently dropped. ## Solver Semantics The v1 adapter SHALL use Z3 package `4.16.0.0`, engine `4.16.0`, `QF_LIA`, seed -zero, timeout 5000 ms, one thread, automatic configuration disabled, and model +zero, operation timeout 5000 ms, one thread, automatic configuration disabled, and model and unsat-core production enabled. Scalar domain members SHALL be encoded as integer indices. The published solver configuration SHALL contain every one of these choices and be digest-bound. +Every solver check MUST complete with `sat` or `unsat` before it contributes to +a completed outcome, witness-selection claim, or core-reduction claim. Empty +membership clauses MUST be asserted as false explicitly. The adapter MUST +defensively reject duplicate clause ids before tracked-assumption construction, +even though the normalized model contract independently forbids them. + +One analysis MUST construct one expression graph and query it incrementally. +The 5000 ms monotonic deadline covers expression construction, all repeated +checks, and deterministic witness or core selection. Each native check receives +at most the remaining operation time, and a result returned after the deadline +is an operational failure. + +For symbols `S`, domains `D(s)`, and clauses `C`, one analysis has the derived +check budget: + +```text +B = 1 + max(|C|, sum(|D(s)| for s in S)) +``` + +The initial decision consumes one check. Only witness selection or core +reduction then runs, so the former consumes at most one check per domain member +and the latter at most one per clause. Exhausting this derived budget is an +operational failure. The reference implementation SHALL preserve phase, check +count, check budget, 5000 ms operation timeout, and bounded solver reason on its +operational-error boundary; it SHALL NOT emit partial evidence. + For a satisfiable model, the witness SHALL select the first feasible value for each symbol in canonical symbol and domain order while preserving previous choices. The resulting binding SHALL pass normal scenario instantiation and From fee5ce512e25e2f31be32682ac39be696606710a Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:17:23 -0700 Subject: [PATCH 2/2] test(processor): clarify solver failure assertions (#1108) --- .../tests/test_scenario_satisfiability.py | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/implementations/python/tests/test_scenario_satisfiability.py b/implementations/python/tests/test_scenario_satisfiability.py index 3b7c62acc..e243d7310 100644 --- a/implementations/python/tests/test_scenario_satisfiability.py +++ b/implementations/python/tests/test_scenario_satisfiability.py @@ -378,7 +378,8 @@ def test_unknown_during_initial_decision_fails_without_evidence( error = exc_info.value assert error.solver_phase == "initial-decision" assert error.solver_check_count == 1 - assert error.solver_check_budget is not None and error.solver_check_budget >= 1 + assert error.solver_check_budget is not None + assert error.solver_check_budget >= 1 def test_unknown_during_core_reduction_fails_loudly_with_budget_context( @@ -396,7 +397,8 @@ def test_unknown_during_core_reduction_fails_loudly_with_budget_context( error = exc_info.value assert error.solver_phase == "core-reduction" assert error.solver_check_count == 2 - assert error.solver_check_budget is not None and error.solver_check_budget >= 2 + assert error.solver_check_budget is not None + assert error.solver_check_budget >= 2 assert error.solver_timeout_ms == SOLVER_TIMEOUT_MS assert error.solver_reason @@ -417,7 +419,8 @@ def test_unknown_during_witness_selection_fails_loudly_with_budget_context( error = exc_info.value assert error.solver_phase == "witness-selection" assert error.solver_check_count == 2 - assert error.solver_check_budget is not None and error.solver_check_budget >= 2 + assert error.solver_check_budget is not None + assert error.solver_check_budget >= 2 assert error.solver_timeout_ms == SOLVER_TIMEOUT_MS assert error.solver_reason is not None assert len(error.solver_reason) == 256 @@ -490,13 +493,16 @@ def test_empty_target_domain_is_explicitly_unsatisfiable(tmp_path: Path) -> None def test_derived_solver_check_budget_fails_closed_before_extra_check(tmp_path: Path) -> None: source = _write(tmp_path, "satisfiable.sdl.yaml", _SATISFIABLE) model = analyze_scenario_file(source).normalized_model + clause_ids = tuple(clause.clause_id for clause in model.clauses) + fixed_indexes: dict[str, int] = {} + budget = _CheckBudget(max_checks=0) with pytest.raises(SolverOperationalError, match="check budget exhausted") as exc_info: _check( model, - tuple(clause.clause_id for clause in model.clauses), - {}, - budget=_CheckBudget(max_checks=0), + clause_ids, + fixed_indexes, + budget=budget, phase="test-probe", ) @@ -651,9 +657,10 @@ def test_published_maximum_witness_shape_has_a_finite_derived_budget() -> None: def test_operation_deadline_covers_model_construction(monkeypatch: pytest.MonkeyPatch) -> None: clock = iter((0, SOLVER_TIMEOUT_MS * 1_000_000)) monkeypatch.setattr(solver_adapter.time, "monotonic_ns", clock.__next__) + model = _normalized_model([1], [[]]) with pytest.raises(SolverOperationalError, match="operation deadline exhausted") as raised: - solve_model(_normalized_model([1], [[]])) + solve_model(model) assert raised.value.phase == "model-construction" assert raised.value.check_count == 0 @@ -671,9 +678,10 @@ def complete_after_deadline(self: z3.Solver, *assumptions: object) -> z3.CheckSa monkeypatch.setattr(solver_adapter.time, "monotonic_ns", lambda: clock["now"]) monkeypatch.setattr(z3.Solver, "check", complete_after_deadline) + model = _normalized_model([1], [[]]) with pytest.raises(SolverOperationalError, match="operation deadline exhausted") as raised: - solve_model(_normalized_model([1], [[]])) + solve_model(model) assert raised.value.phase == "initial-decision" assert raised.value.check_count == 1