Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -22,19 +22,32 @@ 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:
# ``_check`` fails loudly on ``z3.unknown``, so a non-satisfiable decision
# here is decisively unsatisfiable rather than an incomplete timeout.
if _check(model, all_clause_ids, {}) == z3.sat:
return SolverResult(
outcome=SatisfiabilityOutcome.SATISFIABLE,
assignment=_select_witness(model, all_clause_ids),
)
if status == z3.unsat:
return SolverResult(
outcome=SatisfiabilityOutcome.UNSATISFIABLE,
core=_reduce_unsat_core(model, all_clause_ids),
)
raise SolverOperationalError("solver returned an incomplete result")
return SolverResult(
outcome=SatisfiabilityOutcome.UNSATISFIABLE,
core=_reduce_unsat_core(model, all_clause_ids),
)


def _require_unique_clause_ids(model: NormalizedConstraintModel) -> None:
"""Reject collapsed clause identity before it can corrupt tracked evidence.

``_check`` tracks each clause by a name derived from its id; duplicate ids
collapse in the lookup table while z3 rejects the repeated assumption, so the
invariant is enforced explicitly at the solver boundary rather than silently.
"""

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")


def _select_witness(
Expand Down Expand Up @@ -93,11 +106,20 @@ def _check(
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))
# An empty allowed set is an unsatisfiable membership; assert false
# explicitly instead of depending on a zero-argument ``z3.Or``.
expression = (
z3.Or(*(variables[symbol.symbol_id] == index for index in indexes)) if indexes else z3.BoolVal(False)
)
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()
result = solver.check()
if result == z3.unknown:
# A per-call timeout returns unknown; treating it as a decisive answer
# would silently break unsat-core minimality or witness canonicality.
raise SolverOperationalError(f"solver returned unknown: {solver.reason_unknown()}")
return result


def _scalar_equal(left: object, right: object) -> bool:
Expand Down
84 changes: 84 additions & 0 deletions implementations/python/tests/test_scenario_satisfiability.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,19 @@
from pathlib import Path

import pytest
import z3
from pydantic import ValidationError
from raes_contracts.satisfiability import (
SatisfiabilityOutcome,
ScenarioSatisfiabilityEvidenceModel,
)
from raes_processor.satisfiability import (
SatisfiabilityEvidenceError,
SatisfiabilityOperationalError,
analyze_scenario_file,
replay_satisfiability_evidence,
)
from raes_processor.satisfiability._solver import SolverOperationalError, solve_model

_SATISFIABLE = """\
name: satisfiable-control
Expand Down Expand Up @@ -63,6 +66,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
Expand Down Expand Up @@ -313,3 +331,69 @@ 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_call``-th z3 check report ``unknown`` (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_core_reduction_fails_loudly_without_forging_minimality(
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 an undetected timeout would silently retain a droppable clause and
# publish a core claiming subset-minimality it never established.
_force_unknown_on_call(monkeypatch, target_call=2)

with pytest.raises(SatisfiabilityOperationalError):
analyze_scenario_file(source)


def test_unknown_during_witness_selection_fails_loudly_without_forging_canonicality(
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 an undetected timeout would skip it and forge a non-lexicographic
# witness while still claiming canonical-lexicographic selection.
_force_unknown_on_call(monkeypatch, target_call=2)

with pytest.raises(SatisfiabilityOperationalError):
analyze_scenario_file(source)


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 a collapsed clause
# id; the solver must reject it rather than track the assumption twice.
duplicated = model.model_copy(update={"clauses": model.clauses + (model.clauses[0],)})

with pytest.raises(SolverOperationalError, match="duplicate clause ids"):
solve_model(duplicated)


def test_empty_target_domain_is_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
# The count target admits only values >= 1, so a {0} domain produces an empty
# membership clause that must resolve to false, not a bare zero-argument Or.
assert any(clause.allowed_values == () for clause in evidence.normalized_model.clauses)
Loading