diff --git a/docs/design/vss2-02-capsule-solver.md b/docs/design/vss2-02-capsule-solver.md new file mode 100644 index 00000000..9b3f1e8b --- /dev/null +++ b/docs/design/vss2-02-capsule-solver.md @@ -0,0 +1,60 @@ +# VSS2-02: Capsule solve plans, SCC joint solving, and interface summaries + +**Issue:** SLM-66 +**Status:** wiring / fixture evidence. No train, eval, benchmark, model, checkpoint, or ship claim. + +## What was added + +A deterministic capsule solver coordinator in `src/slm_training/dsl/solver/capsule_solver.py`: + +- `CapsuleInterfaceSummary`, `CapsuleProblem`, `CapsuleSolvePlan`, `PerCapsuleResult`, `CapsuleSolveResult`, `Disagreement`, `CapsuleCounters`, `BindingSummary`, `SlotSummary`, `ExternalInput` — immutable, JSON-safe dataclasses with `to_dict`/`from_dict`. +- `build_capsule_solve_plan(graph)` — topological stage planning that: + - Computes dependency stages from the capsule graph. + - Groups independent capsules that share no edges into the same stage. + - Marks strongly-connected components (SCCs) as joint problems so they are solved together. +- `solve_capsule_graph(...)` — dependency-order solve loop that: + - Solves each stage in order. + - Blocks a capsule when a predecessor summary is unknown (conservative predecessor check). + - Materializes predecessor outputs into successor inputs. + - Records local-pass/global-fail verifier disagreements. + - Updates counters for support queries, solver nodes, verifier calls, and joint solves. + +Pack integration in `src/slm_training/dsl/pack.py`: + +- Added optional slots to `DslPack`: + - `capsule_problem_builder` + - `capsule_summary_extractor` + - `capsule_materializer` + - `capsule_local_oracle` + - `capsule_global_oracle` +- Fixed a latent registration ordering bug: `_ensure_builtin_packs()` now uses a `_BUILTINS_LOADED` flag instead of `_PACKS` truthiness, so registering a custom pack before the builtins are loaded no longer hides `openui`/`toy-layout`/`graphql`. + +Serialization helpers: + +- `src/slm_training/dsl/solver/controller.py` — added `to_dict`/`from_dict` to `TerminalOutcome`, `SearchDecision`, `Nogood`, and `SearchResult`. +- `src/slm_training/dsl/solver/closure.py` — added `CertifiedDeduction.from_dict`. +- `src/slm_training/dsl/solver/__init__.py` — exports the capsule solver public API. + +Support oracle tweak: + +- `src/slm_training/dsl/solver/support.py` — the enumerative oracle now prefers an unresolved hole when expanding a multi-hole child, while still falling back to the first hole for singleton-hole expanders. + +## Verified + +- `ruff check` passes. +- `python -m compileall` passes. +- `pytest tests/test_dsl/test_solver_controller.py tests/test_dsl/test_solver_closure.py tests/test_dsl/test_solver_support.py tests/test_dsl/test_solver_state.py tests/test_dsl/test_capsule_solver.py tests/test_data/test_progspec.py -q` → 110 passed. +- `.githooks/check-changed` → all checks passed (`tests/test_dsl`, `tests/test_harnesses/model_build`: 418 passed, 5 skipped, 12 deselected). +- `python -m scripts.repo_policy` ok. +- `git diff --check` clean. + +## Design boundaries preserved + +- The solver is coordinator wiring: it delegates per-capsule search to the existing VSS1 controller/state/closure/support stack. +- No new model, checkpoint, or ship gate is introduced. +- Conservative blocking and disagreement recording keep the contract honest: a missing predecessor summary does not silently become a pass. + +## Caveats + +- This is solver wiring only. Real capsule-level verification needs model-backed oracles, pack hooks for the new `DslPack` slots, and end-to-end fixture runs. +- No model, checkpoint, or ship gate is claimed. diff --git a/src/slm_training/dsl/pack.py b/src/slm_training/dsl/pack.py index 63ef6945..62ba7f8f 100644 --- a/src/slm_training/dsl/pack.py +++ b/src/slm_training/dsl/pack.py @@ -99,6 +99,11 @@ class DslPack: scope_extractor: Callable[..., list[Any]] | None = None prop_order: Callable[[], Mapping[str, Sequence[str]]] | None = None incremental_engine: Callable[[], Any] | None = None + capsule_problem_builder: Callable[..., Any] | None = None + capsule_summary_extractor: Callable[..., Any] | None = None + capsule_materializer: Callable[..., Any] | None = None + capsule_local_oracle: Callable[..., Any] | None = None + capsule_global_oracle: Callable[..., Any] | None = None def filled_slots(self) -> tuple[str, ...]: return tuple( @@ -121,6 +126,7 @@ def require(self, slot: str) -> Any: _PACKS: dict[str, DslPack] = {} +_BUILTINS_LOADED: bool = False # Backend ids that resolve to the openui pack (same language, different parser). _ALIASES = { "openui-lark": "openui", @@ -243,7 +249,8 @@ def _toy_layout_engine() -> Any: def _ensure_builtin_packs() -> None: - if _PACKS: + global _BUILTINS_LOADED + if _BUILTINS_LOADED: return shared_policy = PlaceholderPolicy( placeholder_re=PLACEHOLDER_RE, @@ -292,6 +299,7 @@ def _ensure_builtin_packs() -> None: register_pack(build_graphql_pack()) except Exception: # noqa: BLE001 - graphql pack is optional pass + _BUILTINS_LOADED = True __all__ = [ diff --git a/src/slm_training/dsl/solver/__init__.py b/src/slm_training/dsl/solver/__init__.py index cb91a09b..4aba2263 100644 --- a/src/slm_training/dsl/solver/__init__.py +++ b/src/slm_training/dsl/solver/__init__.py @@ -51,22 +51,44 @@ Verifier, replay_support_certificate, ) +from slm_training.dsl.solver.capsule_solver import ( + BindingSummary, + CapsuleCounters, + CapsuleGlobalVerifier, + CapsuleInterfaceSummary, + CapsuleMaterializer, + CapsuleProblem, + CapsuleProblemBuilder, + CapsuleSolvePlan, + CapsuleSolveResult, + CapsuleSummaryExtractor, + Disagreement, + ExternalInput, + PerCapsuleResult, + SlotSummary, + build_capsule_solve_plan, + solve_capsule_graph, +) __all__ = [ "BaselineRanker", + "BindingSummary", "CandidateRanker", + "CapsuleCounters", + "CapsuleGlobalVerifier", + "CapsuleInterfaceSummary", + "CapsuleMaterializer", + "CapsuleProblem", + "CapsuleProblemBuilder", + "CapsuleSolvePlan", + "CapsuleSolveResult", + "CapsuleSummaryExtractor", "CertifiedDeduction", "ClosureCounters", "ClosureResult", + "Disagreement", "DomainValue", - "Nogood", - "SearchDecision", - "SearchResult", - "SearchStatus", - "TerminalChecker", - "TerminalOutcome", - "default_hole_selector", - "search", + "ExternalInput", "EnumerativeSupportOracle", "EnumerativeSupportProvider", "ExpandStatus", @@ -75,9 +97,15 @@ "HoleDomain", "HoleId", "JsonScalar", + "Nogood", + "PerCapsuleResult", "ProblemExpander", "ReplayResult", "SearchCounters", + "SearchDecision", + "SearchResult", + "SearchStatus", + "SlotSummary", "SolverBounds", "SupportCertificate", "SupportOracle", @@ -85,13 +113,19 @@ "SupportQuery", "SupportResult", "SupportVerdict", + "TerminalChecker", + "TerminalOutcome", "TopologyDomainAdapter", "Verifier", "VerifyOutcome", "VerifyStatus", "WitnessRef", + "build_capsule_solve_plan", "completion_forest_state", + "default_hole_selector", "default_query_order", "exact_closure", "replay_support_certificate", + "search", + "solve_capsule_graph", ] diff --git a/src/slm_training/dsl/solver/capsule_solver.py b/src/slm_training/dsl/solver/capsule_solver.py new file mode 100644 index 00000000..f6f67363 --- /dev/null +++ b/src/slm_training/dsl/solver/capsule_solver.py @@ -0,0 +1,738 @@ +"""Capsule solve plans, SCC joint solving, and interface summaries (VSS2-02). + +This module is the model-independent coordinator that solves verification +capsules in dependency order, treats SCCs as joint finite problems, propagates +explicit interface summaries, and always performs a final whole-program +verification. It keeps the hard/soft separation strict: + +* **Certified deduction** — domain reductions produced by ``exact_closure``. +* **Reversible decision** — branch choices recorded by the search controller. +* **Local nogood** — request-local conflict memory; NOT a certified deduction. +* **Interface summary** — a conservative or exact capsule boundary description. +* **Local/global disagreement** — when a capsule-local oracle passes but the + authoritative whole-program verifier fails. + +The layer is Torch-free. It delegates hole-domain construction, summary +extraction, materialization, and final verification to pack-owned hooks so that +partial packs fail closed when capsule solving is requested. + +Semantics are owned by ``docs/design/verified-scope-solver.md`` and +``docs/design/vss2-02-capsule-solver.md``. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Callable, Protocol + +from slm_training.data.progspec.capsules import ( + CapsuleGraph, + DependencyKind, + VerificationCapsule, +) +from slm_training.dsl.solver.closure import SupportProvider +from slm_training.dsl.solver.controller import ( + CandidateRanker, + SearchResult, + TerminalChecker, + TerminalOutcome, + search, +) +from slm_training.dsl.solver.state import ( + DomainValue, + FiniteDomainState, + SolverBounds, +) +from slm_training.dsl.solver.support import SearchCounters + +JsonValue = Any + + +def _canonical_json(value: Any) -> str: + return json.dumps( + value, allow_nan=False, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + + +def _sha256(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +# --------------------------------------------------------------------------- # +# Interface summary contracts +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True) +class BindingSummary: + """One named binder crossing a capsule boundary. + + ``value`` carries the solved DomainValue when the summary is extracted from a + concrete solved state. It is ``None`` for unresolved inputs or conservative + placeholders. + """ + + name: str + kind: str + origin: str = "" + value: DomainValue | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "kind": self.kind, + "origin": self.origin, + "value": self.value.to_dict() if self.value is not None else None, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> BindingSummary: + value = data.get("value") + return cls( + name=str(data["name"]), + kind=str(data["kind"]), + origin=str(data.get("origin", "")), + value=DomainValue.from_dict(value) if value is not None else None, + ) + + +@dataclass(frozen=True) +class SlotSummary: + """One external slot input required by a capsule.""" + + name: str + kind: str = "slot" + + def to_dict(self) -> dict[str, Any]: + return {"name": self.name, "kind": self.kind} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SlotSummary: + return cls(name=str(data["name"]), kind=str(data.get("kind", "slot"))) + + +@dataclass(frozen=True) +class CapsuleInterfaceSummary: + """Conservative or exact boundary description for one solved capsule.""" + + capsule_id: str + input_bindings: tuple[BindingSummary, ...] + output_bindings: tuple[BindingSummary, ...] + slots: tuple[SlotSummary, ...] + preconditions: tuple[str, ...] + postconditions: tuple[str, ...] + effects: tuple[str, ...] + exceptions: tuple[str, ...] + captures: tuple[str, ...] + conservative: bool + fingerprint: str + + def __post_init__(self) -> None: + if not isinstance(self.capsule_id, str) or not self.capsule_id: + raise ValueError("CapsuleInterfaceSummary requires a non-empty capsule_id") + if not isinstance(self.fingerprint, str) or len(self.fingerprint) != 64: + raise ValueError("CapsuleInterfaceSummary requires a SHA-256 fingerprint") + + def to_dict(self) -> dict[str, Any]: + return { + "capsule_id": self.capsule_id, + "input_bindings": [b.to_dict() for b in self.input_bindings], + "output_bindings": [b.to_dict() for b in self.output_bindings], + "slots": [s.to_dict() for s in self.slots], + "preconditions": list(self.preconditions), + "postconditions": list(self.postconditions), + "effects": list(self.effects), + "exceptions": list(self.exceptions), + "captures": list(self.captures), + "conservative": self.conservative, + "fingerprint": self.fingerprint, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CapsuleInterfaceSummary: + return cls( + capsule_id=str(data["capsule_id"]), + input_bindings=tuple( + BindingSummary.from_dict(d) for d in data.get("input_bindings", []) + ), + output_bindings=tuple( + BindingSummary.from_dict(d) for d in data.get("output_bindings", []) + ), + slots=tuple(SlotSummary.from_dict(d) for d in data.get("slots", [])), + preconditions=tuple(str(v) for v in data.get("preconditions", [])), + postconditions=tuple(str(v) for v in data.get("postconditions", [])), + effects=tuple(str(v) for v in data.get("effects", [])), + exceptions=tuple(str(v) for v in data.get("exceptions", [])), + captures=tuple(str(v) for v in data.get("captures", [])), + conservative=bool(data.get("conservative", False)), + fingerprint=str(data["fingerprint"]), + ) + + @property + def is_exact(self) -> bool: + """An empty interface is still conservative if the flag says so.""" + return not self.conservative + + +@dataclass(frozen=True) +class ExternalInput: + """One typed input arriving from outside the capsule graph.""" + + name: str + kind: str = "external" + + def to_dict(self) -> dict[str, Any]: + return {"name": self.name, "kind": self.kind} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ExternalInput: + return cls(name=str(data["name"]), kind=str(data.get("kind", "external"))) + + +# --------------------------------------------------------------------------- # +# Problem / plan / result contracts +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True) +class CapsuleProblem: + """One capsule plus its finite-domain state and boundary assumptions.""" + + capsule: VerificationCapsule + state: FiniteDomainState + predecessor_summaries: tuple[CapsuleInterfaceSummary, ...] + external_inputs: tuple[ExternalInput, ...] + + def to_dict(self) -> dict[str, Any]: + return { + "capsule": self.capsule.to_dict(), + "state": self.state.to_dict(), + "predecessor_summaries": [s.to_dict() for s in self.predecessor_summaries], + "external_inputs": [e.to_dict() for e in self.external_inputs], + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CapsuleProblem: + return cls( + capsule=VerificationCapsule.from_dict(data["capsule"]), + state=FiniteDomainState.from_dict(data["state"]), + predecessor_summaries=tuple( + CapsuleInterfaceSummary.from_dict(d) + for d in data.get("predecessor_summaries", []) + ), + external_inputs=tuple( + ExternalInput.from_dict(d) for d in data.get("external_inputs", []) + ), + ) + + +@dataclass(frozen=True) +class PerCapsuleResult: + """Outcome of solving one capsule, independent of assembly.""" + + capsule_id: str + status: str + state: FiniteDomainState + summary: CapsuleInterfaceSummary | None + search_result: SearchResult | None + stop_reason: str | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "capsule_id": self.capsule_id, + "status": self.status, + "state": self.state.to_dict(), + "summary": self.summary.to_dict() if self.summary is not None else None, + "search_result": self.search_result.to_dict() if self.search_result is not None else None, + "stop_reason": self.stop_reason, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> PerCapsuleResult: + return cls( + capsule_id=str(data["capsule_id"]), + status=str(data["status"]), + state=FiniteDomainState.from_dict(data["state"]), + summary=CapsuleInterfaceSummary.from_dict(data["summary"]) + if data.get("summary") is not None + else None, + search_result=SearchResult.from_dict(data["search_result"]) + if data.get("search_result") is not None + else None, + stop_reason=data.get("stop_reason"), + ) + + +@dataclass(frozen=True) +class CapsuleSolvePlan: + """Deterministic execution plan for a capsule dependency graph.""" + + graph_fingerprint: str + stages: tuple[tuple[str, ...], ...] + joint_sccs: tuple[tuple[str, ...], ...] + fingerprint: str + + def to_dict(self) -> dict[str, Any]: + return { + "graph_fingerprint": self.graph_fingerprint, + "stages": [list(stage) for stage in self.stages], + "joint_sccs": [list(scc) for scc in self.joint_sccs], + "fingerprint": self.fingerprint, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CapsuleSolvePlan: + return cls( + graph_fingerprint=str(data["graph_fingerprint"]), + stages=tuple(tuple(str(v) for v in stage) for stage in data.get("stages", [])), + joint_sccs=tuple( + tuple(str(v) for v in scc) for scc in data.get("joint_sccs", []) + ), + fingerprint=str(data["fingerprint"]), + ) + + +@dataclass(frozen=True) +class Disagreement: + """Local pass and global pass disagree; neither relabeled as solved.""" + + kind: str + capsule_id: str + detail: str + + def to_dict(self) -> dict[str, Any]: + return {"kind": self.kind, "capsule_id": self.capsule_id, "detail": self.detail} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Disagreement: + return cls( + kind=str(data["kind"]), + capsule_id=str(data["capsule_id"]), + detail=str(data["detail"]), + ) + + +@dataclass(frozen=True) +class CapsuleCounters: + """Aggregated work counters across all capsules and the final verifier.""" + + passes: int = 0 + support_queries: int = 0 + cache_hits: int = 0 + supported: int = 0 + unsupported: int = 0 + unknown: int = 0 + candidates_removed: int = 0 + decisions: int = 0 + backtracks: int = 0 + local_nogoods: int = 0 + expanded_solver_nodes: int = 0 + solver_verifier_calls: int = 0 + capsule_count: int = 0 + joint_count: 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, + "decisions": self.decisions, + "backtracks": self.backtracks, + "local_nogoods": self.local_nogoods, + "expanded_solver_nodes": self.expanded_solver_nodes, + "solver_verifier_calls": self.solver_verifier_calls, + "capsule_count": self.capsule_count, + "joint_count": self.joint_count, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CapsuleCounters: + return cls(**{key: int(data[key]) for key in cls().to_dict()}) + + +@dataclass(frozen=True) +class CapsuleSolveResult: + """Project-level outcome of a capsule solve campaign.""" + + status: str + capsule_results: tuple[PerCapsuleResult, ...] + summaries: tuple[CapsuleInterfaceSummary, ...] + assembled_source: str | None + global_verifier_report: JsonValue | None + local_global_disagreements: tuple[Disagreement, ...] + counters: CapsuleCounters + stop_reason: str | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "status": self.status, + "capsule_results": [r.to_dict() for r in self.capsule_results], + "summaries": [s.to_dict() for s in self.summaries], + "assembled_source": self.assembled_source, + "global_verifier_report": self.global_verifier_report, + "local_global_disagreements": [d.to_dict() for d in self.local_global_disagreements], + "counters": self.counters.to_dict(), + "stop_reason": self.stop_reason, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CapsuleSolveResult: + return cls( + status=str(data["status"]), + capsule_results=tuple( + PerCapsuleResult.from_dict(d) for d in data.get("capsule_results", []) + ), + summaries=tuple( + CapsuleInterfaceSummary.from_dict(d) for d in data.get("summaries", []) + ), + assembled_source=data.get("assembled_source"), + global_verifier_report=data.get("global_verifier_report"), + local_global_disagreements=tuple( + Disagreement.from_dict(d) + for d in data.get("local_global_disagreements", []) + ), + counters=CapsuleCounters.from_dict(data["counters"]), + stop_reason=data.get("stop_reason"), + ) + + +# --------------------------------------------------------------------------- # +# Pack seam protocols +# --------------------------------------------------------------------------- # + + +class CapsuleProblemBuilder(Protocol): + """Pack-owned construction of a finite-domain problem for one capsule.""" + + def build_problem( + self, + capsule: VerificationCapsule, + predecessor_summaries: tuple[CapsuleInterfaceSummary, ...], + external_inputs: tuple[ExternalInput, ...], + bounds: SolverBounds, + ) -> CapsuleProblem: ... + + +class CapsuleSummaryExtractor(Protocol): + """Pack-owned extraction of an interface summary from a solved capsule.""" + + def extract_summary( + self, capsule: VerificationCapsule, state: FiniteDomainState + ) -> CapsuleInterfaceSummary: ... + + +CapsuleMaterializer = Callable[[tuple[PerCapsuleResult, ...]], str | None] + + +class CapsuleGlobalVerifier(Protocol): + """Authoritative whole-program verifier; the local oracle is not final.""" + + def verify(self, source: str | None) -> TerminalOutcome: ... + + +# --------------------------------------------------------------------------- # +# Plan construction +# --------------------------------------------------------------------------- # + + +def _capsule_predecessors( + graph: CapsuleGraph, +) -> dict[str, set[str]]: + """Map each capsule to the set of predecessor capsules from REFERENCE edges.""" + node_to_capsule: dict[str, str] = {} + for capsule in graph.capsules: + for node_id in capsule.node_ids: + node_to_capsule[node_id] = capsule.capsule_id + + predecessors: dict[str, set[str]] = { + capsule.capsule_id: set() for capsule in graph.capsules + } + for edge in graph.edges: + if edge.kind is not DependencyKind.REFERENCE: + continue + source_capsule = node_to_capsule.get(edge.source) + target_capsule = node_to_capsule.get(edge.target) + if source_capsule is None or target_capsule is None: + continue + if source_capsule != target_capsule: + predecessors[target_capsule].add(source_capsule) + return predecessors + + +def build_capsule_solve_plan(graph: CapsuleGraph) -> CapsuleSolvePlan: + """Build a deterministic topological stage plan from a capsule graph. + + Each stage contains capsules whose predecessor capsules are all in earlier + stages. SCCs with more than one member are recorded as joint problems. + """ + predecessors = _capsule_predecessors(graph) + remaining = set(predecessors) + stages: list[tuple[str, ...]] = [] + placed: set[str] = set() + + while remaining: + ready = sorted( + capsule_id + for capsule_id in remaining + if predecessors[capsule_id] <= placed + ) + if not ready: + # The graph is expected to be acyclic; if not, fall back to one + # stage per remaining capsule to avoid infinite looping. + ready = sorted(remaining) + stage = tuple(ready) + stages.append(stage) + placed.update(ready) + remaining.difference_update(ready) + + joint_sccs = tuple( + tuple(sorted(capsule.node_ids)) + for capsule in graph.capsules + if len(capsule.node_ids) > 1 + ) + + plan_payload = _canonical_json( + { + "graph_fingerprint": graph.fingerprint if hasattr(graph, "fingerprint") else graph.spec_id, + "stages": [list(stage) for stage in stages], + "joint_sccs": [list(scc) for scc in joint_sccs], + } + ) + fingerprint = _sha256(plan_payload) + graph_fingerprint = getattr(graph, "fingerprint", graph.spec_id) + + return CapsuleSolvePlan( + graph_fingerprint=graph_fingerprint, + stages=tuple(stages), + joint_sccs=joint_sccs, + fingerprint=fingerprint, + ) + + +# --------------------------------------------------------------------------- # +# Core solve loop +# --------------------------------------------------------------------------- # + + +def _empty_summary( + capsule: VerificationCapsule, *, conservative: bool, reason: str +) -> CapsuleInterfaceSummary: + """A conservative empty-boundary summary for a skipped/failed capsule.""" + payload = _canonical_json( + { + "capsule_id": capsule.capsule_id, + "reason": reason, + "conservative": conservative, + } + ) + return CapsuleInterfaceSummary( + capsule_id=capsule.capsule_id, + input_bindings=(), + output_bindings=(), + slots=tuple(SlotSummary(name=name) for name in capsule.external_dependencies), + preconditions=(), + postconditions=(), + effects=(), + exceptions=(), + captures=(), + conservative=conservative, + fingerprint=_sha256(payload), + ) + + +def solve_capsule_graph( + graph: CapsuleGraph, + *, + builder: CapsuleProblemBuilder, + provider: SupportProvider, + terminal_checker: TerminalChecker, + summary_extractor: CapsuleSummaryExtractor, + materializer: CapsuleMaterializer, + global_verifier: Callable[[str | None], TerminalOutcome], + ranker: CandidateRanker | None = None, + bounds: SolverBounds | None = None, + cache: dict[str, Any] | None = None, + certificate_store: dict[str, Any] | None = None, +) -> CapsuleSolveResult: + """Solve a capsule dependency graph and run the whole-program verifier.""" + plan = build_capsule_solve_plan(graph) + bounds = bounds or SolverBounds( + max_tokens=10_000, + max_nodes=10_000, + max_depth=256, + max_backtracks=1_000, + max_verifier_calls=1_000, + ) + cache = cache or {} + + results_by_id: dict[str, PerCapsuleResult] = {} + summaries_by_id: dict[str, CapsuleInterfaceSummary] = {} + capsule_results: list[PerCapsuleResult] = [] + counters = CapsuleCounters(capsule_count=len(graph.capsules), joint_count=len(plan.joint_sccs)) + + status = "solved" + stop_reason: str | None = None + + for stage_index, stage in enumerate(plan.stages): + for capsule_id in sorted(stage): + capsule = next(c for c in graph.capsules if c.capsule_id == capsule_id) + preds = _capsule_predecessors(graph)[capsule_id] + pred_summaries = tuple(summaries_by_id[p] for p in sorted(preds) if p in summaries_by_id) + + # Unknown/conservative predecessor boundaries cannot license an + # exact successor solve. + if any(s.conservative for s in pred_summaries): + summary = _empty_summary( + capsule, conservative=True, reason="conservative_predecessor" + ) + result = PerCapsuleResult( + capsule_id=capsule_id, + status="unknown", + state=_empty_state(capsule, bounds), + summary=summary, + search_result=None, + stop_reason="conservative_predecessor", + ) + results_by_id[capsule_id] = result + summaries_by_id[capsule_id] = summary + capsule_results.append(result) + status = "unknown" + if stop_reason is None: + stop_reason = "conservative_predecessor" + continue + + external_inputs = tuple( + ExternalInput(name=name) + for name in sorted(capsule.external_dependencies) + ) + problem = builder.build_problem( + capsule, pred_summaries, external_inputs, bounds + ) + + search_result = search( + problem.state, + provider, + terminal_checker, + ranker=ranker, + cache=cache, + certificate_store=certificate_store, + ) + + # Aggregate controller counters. + counters = _add_search_counters(counters, search_result.counters) + + if search_result.status.value != "solved": + summary = _empty_summary( + capsule, + conservative=True, + reason=f"solver:{search_result.status.value}", + ) + result = PerCapsuleResult( + capsule_id=capsule_id, + status=search_result.status.value, + state=search_result.state, + summary=summary, + search_result=search_result, + stop_reason=search_result.stop_reason, + ) + results_by_id[capsule_id] = result + summaries_by_id[capsule_id] = summary + capsule_results.append(result) + status = "unknown" if status == "solved" else status + if stop_reason is None: + stop_reason = f"capsule_{capsule_id}:{search_result.status.value}" + continue + + summary = summary_extractor.extract_summary(capsule, search_result.state) + result = PerCapsuleResult( + capsule_id=capsule_id, + status="solved", + state=search_result.state, + summary=summary, + search_result=search_result, + stop_reason=None, + ) + results_by_id[capsule_id] = result + summaries_by_id[capsule_id] = summary + capsule_results.append(result) + + assembled = materializer(tuple(capsule_results)) + final = global_verifier(assembled) + + disagreements: list[Disagreement] = [] + if final.accepted: + project_status = "solved" if status == "solved" else status + else: + project_status = "unknown" + stop_reason = stop_reason or "global_verifier_rejected" + disagreements.append( + Disagreement( + kind="local_pass_global_fail", + capsule_id="assembly", + detail=final.detail or "global verifier rejected assembled program", + ) + ) + + counters = CapsuleCounters( + passes=counters.passes, + support_queries=counters.support_queries, + cache_hits=counters.cache_hits, + supported=counters.supported, + unsupported=counters.unsupported, + unknown=counters.unknown, + candidates_removed=counters.candidates_removed, + decisions=counters.decisions, + backtracks=counters.backtracks, + local_nogoods=counters.local_nogoods, + expanded_solver_nodes=counters.expanded_solver_nodes, + solver_verifier_calls=counters.solver_verifier_calls + 1, + capsule_count=counters.capsule_count, + joint_count=counters.joint_count, + ) + + return CapsuleSolveResult( + status=project_status, + capsule_results=tuple(capsule_results), + summaries=tuple(summaries_by_id[c] for c in sorted(summaries_by_id)), + assembled_source=assembled, + global_verifier_report=final.report, + local_global_disagreements=tuple(disagreements), + counters=counters, + stop_reason=stop_reason, + ) + + +def _empty_state(capsule: VerificationCapsule, bounds: SolverBounds) -> FiniteDomainState: + """A well-formed but empty state for skipped capsules.""" + return FiniteDomainState( + problem_id=f"{capsule.capsule_id}:skipped", + pack_id="capsule-solver", + constraint_version="vss2-02-v1", + bounds=bounds, + holes=(), + ) + + +def _add_search_counters(counters: CapsuleCounters, sc: SearchCounters) -> CapsuleCounters: + return CapsuleCounters( + passes=counters.passes, + support_queries=counters.support_queries + sc.tokens, + cache_hits=counters.cache_hits, + supported=counters.supported, + unsupported=counters.unsupported, + unknown=counters.unknown, + candidates_removed=counters.candidates_removed, + decisions=counters.decisions + sc.depth, + backtracks=counters.backtracks + sc.backtracks, + local_nogoods=counters.local_nogoods, + expanded_solver_nodes=counters.expanded_solver_nodes + sc.nodes, + solver_verifier_calls=counters.solver_verifier_calls + sc.verifier_calls, + capsule_count=counters.capsule_count, + joint_count=counters.joint_count, + ) diff --git a/src/slm_training/dsl/solver/closure.py b/src/slm_training/dsl/solver/closure.py index f968ada1..0f7bb364 100644 --- a/src/slm_training/dsl/solver/closure.py +++ b/src/slm_training/dsl/solver/closure.py @@ -140,6 +140,17 @@ def to_dict(self) -> dict[str, Any]: "reason": self.reason, } + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CertifiedDeduction: + return cls( + before_fingerprint=str(data["before_fingerprint"]), + after_fingerprint=str(data["after_fingerprint"]), + hole_id=HoleId.from_dict(data["hole_id"]), + removed=tuple(DomainValue.from_dict(d) for d in data.get("removed", [])), + certificate_ids=tuple(str(v) for v in data.get("certificate_ids", [])), + reason=str(data.get("reason", "")), + ) + @dataclass(frozen=True) class ClosureCounters: diff --git a/src/slm_training/dsl/solver/controller.py b/src/slm_training/dsl/solver/controller.py index a01bb96c..b6ff79a0 100644 --- a/src/slm_training/dsl/solver/controller.py +++ b/src/slm_training/dsl/solver/controller.py @@ -96,6 +96,23 @@ class TerminalOutcome: report: JsonValue | None = None detail: str = "" + def to_dict(self) -> dict[str, Any]: + return { + "accepted": self.accepted, + "source": self.source, + "report": self.report, + "detail": self.detail, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> TerminalOutcome: + return cls( + accepted=bool(data["accepted"]), + source=data.get("source"), + report=data.get("report"), + detail=str(data.get("detail", "")), + ) + class TerminalChecker(Protocol): """Materializes a structurally-solved state and runs the final verifier.""" @@ -131,6 +148,21 @@ def to_dict(self) -> dict[str, Any]: "ranker_id": self.ranker_id, } + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SearchDecision: + return cls( + decision_id=str(data["decision_id"]), + before_fingerprint=str(data["before_fingerprint"]), + after_fingerprint=str(data["after_fingerprint"]), + level=int(data["level"]), + hole_id=HoleId.from_dict(data["hole_id"]), + chosen=DomainValue.from_dict(data["chosen"]), + alternatives=tuple( + DomainValue.from_dict(d) for d in data.get("alternatives", []) + ), + ranker_id=str(data["ranker_id"]), + ) + @dataclass(frozen=True) class Nogood: @@ -153,6 +185,21 @@ def to_dict(self) -> dict[str, Any]: "provenance": self.provenance, } + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Nogood: + assignment: list[tuple[HoleId, DomainValue]] = [] + for pair in data.get("assignment", []): + if not isinstance(pair, (list, tuple)) or len(pair) != 2: + raise ValueError("Nogood assignment entries must be [hole, value] pairs") + assignment.append((HoleId.from_dict(pair[0]), DomainValue.from_dict(pair[1]))) + return cls( + problem_id=str(data["problem_id"]), + constraint_version=str(data["constraint_version"]), + bounds=SolverBounds.from_dict(data["bounds"]), + assignment=tuple(assignment), + provenance=str(data.get("provenance", "")), + ) + @property def key(self) -> str: return _canonical_json( @@ -179,6 +226,37 @@ class SearchResult: counters: SearchCounters stop_reason: str | None = None + def to_dict(self) -> dict[str, Any]: + return { + "status": self.status.value, + "state": self.state.to_dict(), + "source": self.source, + "verifier_report": self.verifier_report, + "deductions": [d.to_dict() for d in self.deductions], + "decisions": [d.to_dict() for d in self.decisions], + "nogoods": [n.to_dict() for n in self.nogoods], + "counters": self.counters.to_dict(), + "stop_reason": self.stop_reason, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SearchResult: + return cls( + status=SearchStatus(data["status"]), + state=FiniteDomainState.from_dict(data["state"]), + source=data.get("source"), + verifier_report=data.get("verifier_report"), + deductions=tuple( + CertifiedDeduction.from_dict(d) for d in data.get("deductions", []) + ), + decisions=tuple( + SearchDecision.from_dict(d) for d in data.get("decisions", []) + ), + nogoods=tuple(Nogood.from_dict(d) for d in data.get("nogoods", [])), + counters=SearchCounters.from_dict(data["counters"]), + stop_reason=data.get("stop_reason"), + ) + def default_hole_selector(state: FiniteDomainState) -> HoleId | None: """Smallest live domain first, then canonical ``HoleId`` (deterministic).""" diff --git a/src/slm_training/dsl/solver/support.py b/src/slm_training/dsl/solver/support.py index 2d58d284..d527a12b 100644 --- a/src/slm_training/dsl/solver/support.py +++ b/src/slm_training/dsl/solver/support.py @@ -450,10 +450,15 @@ def check(self, state: FiniteDomainState, query: SupportQuery) -> SupportResult: counters.backtracks += 1 failures["dead:child_bottom"] = failures.get("dead:child_bottom", 0) + 1 continue - # Push every live (hole, value) branch of the child in canonical order - # (reversed so the smallest value is popped first). - child_hole = child.holes[0].hole_id - for child_value in reversed(child.holes[0].values): + # Push every live (hole, value) branch of the child in canonical + # order (reversed so the smallest value is popped first). Prefer an + # unresolved hole to avoid re-expanding a singleton decision path; + # fall back to the first hole for expanders that emit a singleton + # hole and expect the caller to drive it to terminal. + unresolved = [h for h in child.holes if len(h.values) > 1] + chosen_hole = unresolved[0] if unresolved else child.holes[0] + child_hole = chosen_hole.hole_id + for child_value in reversed(chosen_hole.values): stack.append((child, child_hole, child_value, depth + 1)) verdict, exhausted = _decide(witness is not None, incomplete, stop_reason) diff --git a/tests/test_dsl/test_capsule_solver.py b/tests/test_dsl/test_capsule_solver.py new file mode 100644 index 00000000..eda0e07d --- /dev/null +++ b/tests/test_dsl/test_capsule_solver.py @@ -0,0 +1,713 @@ +"""Regression tests for the VSS2-02 capsule solver coordinator.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + +import pytest + +from slm_training.data.progspec.capsules import ( + CapsuleGraph, + DependencyKind, + ScopeEdge, + ScopeNode, + VerificationCapsule, +) +from slm_training.dsl.pack import DslPack, PackSlotUnavailable, get_backend +from slm_training.dsl.solver.capsule_solver import ( + BindingSummary, + CapsuleInterfaceSummary, + CapsuleProblem, + CapsuleSolveResult, + ExternalInput, + SlotSummary, + build_capsule_solve_plan, + solve_capsule_graph, +) +from slm_training.dsl.solver.closure import EnumerativeSupportProvider +from slm_training.dsl.solver.controller import TerminalChecker, TerminalOutcome +from slm_training.dsl.solver.state import DomainValue, FiniteDomainState, HoleDomain, HoleId, SolverBounds +from slm_training.dsl.solver.support import ( + ExpandStatus, + ExpandStep, + Verifier, + VerifyOutcome, + VerifyStatus, +) + + +@dataclass +class CapsuleSpec: + """Test-only description of one fake finite capsule.""" + + holes: dict[str, list[int]] + constraints: Callable[[dict[str, int]], bool] | None = None + outputs: tuple[str, ...] = () + inputs: tuple[str, ...] = () + + +# --------------------------------------------------------------------------- # +# Fake finite pack helpers +# --------------------------------------------------------------------------- # + + +def _hole_id(capsule_id: str, name: str) -> HoleId: + return HoleId(namespace=capsule_id, path=(name,), kind="var") + + +def _int_value(value: int) -> DomainValue: + return DomainValue.create("int", value) + + +class FakeVerifier(Verifier): + """Accepts every terminal program; constraints are enforced by the expander.""" + + profile = "fake-verifier-v1" + + def verify(self, program: str) -> VerifyOutcome: + return VerifyOutcome(status=VerifyStatus.ACCEPT, detail="ok") + + +class FakeExpander: + """Deterministic expander for one fake capsule.""" + + def __init__(self, capsule_id: str, spec: CapsuleSpec, bounds: SolverBounds) -> None: + self.problem_id = capsule_id + self.pack_id = "fake" + self.constraint_version = "v1" + self.bounds = bounds + self.spec = spec + + def successor( + self, state: FiniteDomainState, hole_id: HoleId, value: DomainValue + ) -> ExpandStep: + after = state.with_decision(hole_id, value) + if after.is_structurally_solved: + assignment: dict[str, int] = {} + for hole in after.holes: + name = hole.hole_id.path[0] + assignment[name] = hole.values[0].payload + ok = self.spec.constraints is None or self.spec.constraints(assignment) + if not ok: + return ExpandStep( + status=ExpandStatus.DEAD, detail="constraint_violation" + ) + program = f"{self.problem_id}:{assignment}" + return ExpandStep( + status=ExpandStatus.TERMINAL, program=program, detail="ok" + ) + return ExpandStep( + status=ExpandStatus.CONTINUE, next_state=after, coverage="complete" + ) + + +class FakeBuilder: + """Builds a CapsuleProblem from the SPECS registry.""" + + def __init__(self, specs: dict[str, CapsuleSpec]) -> None: + self.specs = specs + + def build_problem( + self, + capsule: VerificationCapsule, + predecessor_summaries: tuple[CapsuleInterfaceSummary, ...], + external_inputs: tuple[ExternalInput, ...], + bounds: SolverBounds, + ) -> CapsuleProblem: + spec = self.specs[capsule.capsule_id] + allowed: dict[str, list[int]] = { + name: list(values) for name, values in spec.holes.items() + } + + # Predecessor outputs become singleton inputs. + for summary in predecessor_summaries: + for binding in summary.output_bindings: + if binding.value is not None and binding.name in allowed: + allowed[binding.name] = [binding.value.payload] + + # External slots are listed but carry no value in these fixtures. + for ext in external_inputs: + if ext.name in allowed: + # keep the declared domain; the slot name is merely recorded + pass + + holes = tuple( + HoleDomain( + hole_id=_hole_id(capsule.capsule_id, name), + values=tuple(_int_value(v) for v in sorted(allowed[name])), + ) + for name in sorted(allowed) + ) + state = FiniteDomainState( + problem_id=capsule.capsule_id, + pack_id="fake", + constraint_version="v1", + bounds=bounds, + holes=holes, + ) + return CapsuleProblem( + capsule=capsule, + state=state, + predecessor_summaries=predecessor_summaries, + external_inputs=external_inputs, + ) + + +class FakeSummaryExtractor: + """Extracts output bindings from solved singleton holes.""" + + def __init__(self, specs: dict[str, CapsuleSpec]) -> None: + self.specs = specs + + def extract_summary( + self, capsule: VerificationCapsule, state: FiniteDomainState + ) -> CapsuleInterfaceSummary: + spec = self.specs[capsule.capsule_id] + input_bindings = tuple( + BindingSummary(name=name, kind="input") for name in spec.inputs + ) + output_bindings = [] + for name in spec.outputs: + hole_id = _hole_id(capsule.capsule_id, name) + domain = state.domain(hole_id) + value = domain.values[0] if domain.values else None + output_bindings.append( + BindingSummary(name=name, kind="output", value=value) + ) + slots = tuple( + SlotSummary(name=name) for name in sorted(capsule.external_dependencies) + ) + summary = CapsuleInterfaceSummary( + capsule_id=capsule.capsule_id, + input_bindings=input_bindings, + output_bindings=tuple(output_bindings), + slots=slots, + preconditions=(), + postconditions=(), + effects=(), + exceptions=(), + captures=(), + conservative=False, + fingerprint="0" * 64, # replaced below + ) + payload = summary.to_dict() + payload.pop("fingerprint") + import hashlib + import json + + fingerprint = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + return CapsuleInterfaceSummary( + capsule_id=summary.capsule_id, + input_bindings=summary.input_bindings, + output_bindings=summary.output_bindings, + slots=summary.slots, + preconditions=summary.preconditions, + postconditions=summary.postconditions, + effects=summary.effects, + exceptions=summary.exceptions, + captures=summary.captures, + conservative=summary.conservative, + fingerprint=fingerprint, + ) + + +class FakeTerminalChecker(TerminalChecker): + """Local terminal checker accepts any structurally solved state.""" + + def check(self, state: FiniteDomainState) -> TerminalOutcome: + return TerminalOutcome(accepted=True, source="local", report=None) + + +def make_materializer() -> Callable[[tuple[Any, ...]], str]: + def materialize(results: tuple[Any, ...]) -> str: + parts = [] + for result in sorted(results, key=lambda r: r.capsule_id): + if result.status != "solved": + continue + summary = result.summary + assert summary is not None + out = { + b.name: b.value.payload if b.value is not None else None + for b in summary.output_bindings + } + parts.append(f"{result.capsule_id}={out}") + return ";".join(parts) + + return materialize + + +def make_global_verifier(reject_if: Callable[[str | None], bool]) -> Callable[[str | None], TerminalOutcome]: + def verify(source: str | None) -> TerminalOutcome: + if source is None or reject_if(source): + return TerminalOutcome( + accepted=False, detail="global_rejected", report={"source": source} + ) + return TerminalOutcome( + accepted=True, detail="global_accepted", report={"source": source} + ) + + return verify + + +class MultiSupportProvider: + """Dispatches to the per-capsule enumerative provider by problem_id.""" + + def __init__(self, specs: dict[str, CapsuleSpec], bounds: SolverBounds) -> None: + self.specs = specs + self.bounds = bounds + self._providers: dict[str, EnumerativeSupportProvider] = {} + + @property + def backend_version(self) -> str: + return "fake-multi-v1" + + def _provider(self, problem_id: str) -> EnumerativeSupportProvider: + if problem_id not in self._providers: + self._providers[problem_id] = EnumerativeSupportProvider( + FakeExpander(problem_id, self.specs[problem_id], self.bounds), + FakeVerifier(), + ) + return self._providers[problem_id] + + def check(self, state: FiniteDomainState, query: Any) -> Any: + return self._provider(state.problem_id).check(state, query) + + def replay(self, certificate: Any, *, state: FiniteDomainState) -> Any: + return self._provider(state.problem_id).replay(certificate, state=state) + + +def _provider(specs: dict[str, CapsuleSpec], bounds: SolverBounds) -> MultiSupportProvider: + return MultiSupportProvider(specs, bounds) + + +# --------------------------------------------------------------------------- # +# Graph builders +# --------------------------------------------------------------------------- # + + +def _root_node(spec_id: str) -> ScopeNode: + return ScopeNode( + node_id=f"{spec_id}:root", + scope_id=None, + kind="root", + ast_path=(), + member_paths=(), + definitions=(), + external_dependencies=(), + ) + + +def _statement_node(spec_id: str, name: str) -> ScopeNode: + return ScopeNode( + node_id=f"{spec_id}:{name}", + scope_id=f"{spec_id}:{name}", + kind="statement", + ast_path=(name,), + member_paths=(), + definitions=(name,) if not name.startswith(":") else (), + external_dependencies=(), + ) + + +def _capsule(capsule_id: str, node_ids: tuple[str, ...]) -> VerificationCapsule: + return VerificationCapsule( + capsule_id=capsule_id, + node_ids=node_ids, + entry_node_id=node_ids[0], + external_dependencies=(), + ) + + +def _make_graph( + root: ScopeNode, + nodes: tuple[ScopeNode, ...], + edges: tuple[ScopeEdge, ...], + capsules: tuple[VerificationCapsule, ...], + spec_id: str, +) -> CapsuleGraph: + return CapsuleGraph( + root_id=root.node_id, + nodes=nodes, + edges=edges, + capsules=capsules, + spec_id=spec_id, + version=CapsuleGraph.VERSION, + ) + + +def _ref_edge(source: str, target: str) -> ScopeEdge: + return ScopeEdge( + source=source, target=target, kind=DependencyKind.REFERENCE, role="ref" + ) + + +def _external_edge(source: str, role: str) -> ScopeEdge: + return ScopeEdge( + source=source, target=f"{source.split(':')[0]}:root", kind=DependencyKind.EXTERNAL, role=role + ) + + +# --------------------------------------------------------------------------- # +# Tests +# --------------------------------------------------------------------------- # + + +def test_build_plan_stages_are_topological(): + spec_id = "plan-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + b = _statement_node(spec_id, "b") + c = _statement_node(spec_id, "c") + nodes = (root, a, b, c) + edges = ( + _ref_edge(a.node_id, b.node_id), + _ref_edge(b.node_id, c.node_id), + ) + capsules = ( + _capsule(f"{spec_id}:ca", (a.node_id,)), + _capsule(f"{spec_id}:cb", (b.node_id,)), + _capsule(f"{spec_id}:cc", (c.node_id,)), + ) + graph = _make_graph( + root, + nodes=nodes, + edges=edges, + capsules=capsules, + spec_id=spec_id, + ) + plan = build_capsule_solve_plan(graph) + assert plan.stages == ( + (f"{spec_id}:ca",), + (f"{spec_id}:cb",), + (f"{spec_id}:cc",), + ) + assert plan.fingerprint != plan.graph_fingerprint + + +def test_build_plan_groups_independent_capsules(): + spec_id = "independent-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + b = _statement_node(spec_id, "b") + nodes = (root, a, b) + edges = () + capsules = ( + _capsule(f"{spec_id}:ca", (a.node_id,)), + _capsule(f"{spec_id}:cb", (b.node_id,)), + ) + graph = _make_graph( + root, + nodes=nodes, + edges=edges, + capsules=capsules, + spec_id=spec_id, + ) + plan = build_capsule_solve_plan(graph) + assert len(plan.stages) == 1 + assert set(plan.stages[0]) == {f"{spec_id}:ca", f"{spec_id}:cb"} + + +def test_two_node_scc_is_joint_problem(): + spec_id = "scc-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + b = _statement_node(spec_id, "b") + nodes = (root, a, b) + # mutual reference creates a 2-node SCC + edges = ( + _ref_edge(a.node_id, b.node_id), + _ref_edge(b.node_id, a.node_id), + ) + capsules = (_capsule(f"{spec_id}:c_ab", (a.node_id, b.node_id)),) + graph = _make_graph( + root, + nodes=nodes, + edges=edges, + capsules=capsules, + spec_id=spec_id, + ) + plan = build_capsule_solve_plan(graph) + assert plan.joint_sccs == ((a.node_id, b.node_id),) + + specs = { + capsules[0].capsule_id: CapsuleSpec( + holes={"a": [0, 1], "b": [0, 1]}, + constraints=lambda assn: assn["a"] != assn["b"], + outputs=("a", "b"), + ) + } + bounds = SolverBounds( + max_tokens=1000, + max_nodes=1000, + max_depth=10, + max_backtracks=100, + max_verifier_calls=100, + ) + result = solve_capsule_graph( + graph, + builder=FakeBuilder(specs), + provider=_provider(specs, bounds), + terminal_checker=FakeTerminalChecker(), + summary_extractor=FakeSummaryExtractor(specs), + materializer=make_materializer(), + global_verifier=make_global_verifier(lambda _: False), + bounds=bounds, + ) + assert result.status == "solved" + assert result.counters.joint_count == 1 + assert len(result.capsule_results[0].search_result.decisions) >= 1 + # Cyclic dependencies were submitted as one joint problem, not two separate ones. + assert len(result.capsule_results) == 1 + + +def test_predecessor_output_becomes_successor_input(): + spec_id = "chain-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + b = _statement_node(spec_id, "b") + nodes = (root, a, b) + edges = (_ref_edge(a.node_id, b.node_id),) + ca = _capsule(f"{spec_id}:ca", (a.node_id,)) + cb = _capsule(f"{spec_id}:cb", (b.node_id,)) + graph = _make_graph( + root, + nodes=nodes, + edges=edges, + capsules=(ca, cb), + spec_id=spec_id, + ) + specs = { + ca.capsule_id: CapsuleSpec( + holes={"a": [1, 2, 3]}, + constraints=lambda assn: assn["a"] >= 2, + outputs=("a",), + ), + cb.capsule_id: CapsuleSpec( + holes={"a": [1, 2, 3], "b": [1, 2, 3]}, + constraints=lambda assn: assn["b"] == assn["a"], + outputs=("b",), + ), + } + bounds = SolverBounds( + max_tokens=1000, + max_nodes=1000, + max_depth=10, + max_backtracks=100, + max_verifier_calls=100, + ) + result = solve_capsule_graph( + graph, + builder=FakeBuilder(specs), + provider=_provider(specs, bounds), + terminal_checker=FakeTerminalChecker(), + summary_extractor=FakeSummaryExtractor(specs), + materializer=make_materializer(), + global_verifier=make_global_verifier(lambda _: False), + bounds=bounds, + ) + assert result.status == "solved" + ca_result = result.capsule_results[0] + cb_result = result.capsule_results[1] + a_value = next( + b.value.payload for b in ca_result.summary.output_bindings if b.name == "a" + ) + assert a_value == 2 + b_value = next( + b.value.payload for b in cb_result.summary.output_bindings if b.name == "b" + ) + assert b_value == a_value + + +def test_unknown_predecessor_summary_blocks_successor(): + spec_id = "conservative-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + b = _statement_node(spec_id, "b") + nodes = (root, a, b) + edges = (_ref_edge(a.node_id, b.node_id),) + ca = _capsule(f"{spec_id}:ca", (a.node_id,)) + cb = _capsule(f"{spec_id}:cb", (b.node_id,)) + graph = _make_graph( + root, + nodes=nodes, + edges=edges, + capsules=(ca, cb), + spec_id=spec_id, + ) + specs = { + ca.capsule_id: CapsuleSpec(holes={"a": [1]}, outputs=("a",)), + cb.capsule_id: CapsuleSpec(holes={"b": [1]}, outputs=("b",)), + } + bounds = SolverBounds( + max_tokens=100, + max_nodes=100, + max_depth=10, + max_backtracks=10, + max_verifier_calls=10, + ) + + # Force the predecessor summary to be conservative. + class ConservativeExtractor(FakeSummaryExtractor): + def extract_summary(self, capsule, state): + summary = super().extract_summary(capsule, state) + return CapsuleInterfaceSummary( + capsule_id=summary.capsule_id, + input_bindings=summary.input_bindings, + output_bindings=summary.output_bindings, + slots=summary.slots, + preconditions=summary.preconditions, + postconditions=summary.postconditions, + effects=summary.effects, + exceptions=summary.exceptions, + captures=summary.captures, + conservative=True, + fingerprint=summary.fingerprint, + ) + + result = solve_capsule_graph( + graph, + builder=FakeBuilder(specs), + provider=_provider(specs, bounds), + terminal_checker=FakeTerminalChecker(), + summary_extractor=ConservativeExtractor(specs), + materializer=make_materializer(), + global_verifier=make_global_verifier(lambda _: False), + bounds=bounds, + ) + assert result.status == "unknown" + assert result.stop_reason == "conservative_predecessor" + assert result.capsule_results[1].status == "unknown" + + +def test_local_pass_global_fail_is_disagreement(): + spec_id = "disagree-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + nodes = (root, a) + ca = _capsule(f"{spec_id}:ca", (a.node_id,)) + graph = _make_graph( + root, + nodes=nodes, + edges=(), + capsules=(ca,), + spec_id=spec_id, + ) + specs = { + ca.capsule_id: CapsuleSpec(holes={"a": [1]}, outputs=("a",)), + } + bounds = SolverBounds( + max_tokens=100, + max_nodes=100, + max_depth=10, + max_backtracks=10, + max_verifier_calls=10, + ) + result = solve_capsule_graph( + graph, + builder=FakeBuilder(specs), + provider=_provider(specs, bounds), + terminal_checker=FakeTerminalChecker(), + summary_extractor=FakeSummaryExtractor(specs), + materializer=make_materializer(), + global_verifier=make_global_verifier(lambda src: src is not None and "ca=" in src), + bounds=bounds, + ) + assert result.status == "unknown" + assert len(result.local_global_disagreements) == 1 + assert result.local_global_disagreements[0].kind == "local_pass_global_fail" + + +def test_missing_pack_hook_fails_closed(): + import re + + from slm_training.dsl.pack import CONTENT_PROPS, PLACEHOLDER_RE, PlaceholderPolicy, register_pack + + placeholder_policy = PlaceholderPolicy( + placeholder_re=re.compile(PLACEHOLDER_RE.pattern), + content_props=CONTENT_PROPS, + slot_contract=lambda *_: (), + ) + partial = DslPack( + pack_id="partial-capsule-pack", + backend=get_backend("openui"), + placeholder_policy=placeholder_policy, + reward_label="none", + ) + register_pack(partial) + with pytest.raises(PackSlotUnavailable): + partial.require("capsule_problem_builder") + + +def test_summary_round_trips_json(): + summary = CapsuleInterfaceSummary( + capsule_id="c1", + input_bindings=(BindingSummary(name="x", kind="input"),), + output_bindings=(BindingSummary(name="y", kind="output", value=_int_value(7)),), + slots=(SlotSummary(name=":slot"),), + preconditions=("p1",), + postconditions=(), + effects=(), + exceptions=(), + captures=(), + conservative=False, + fingerprint="a" * 64, + ) + recovered = CapsuleInterfaceSummary.from_dict(summary.to_dict()) + assert recovered == summary + assert recovered.is_exact + + +def test_result_round_trips_json(): + spec_id = "roundtrip-test" + root = _root_node(spec_id) + a = _statement_node(spec_id, "a") + ca = _capsule(f"{spec_id}:ca", (a.node_id,)) + graph = _make_graph( + root, + nodes=(root, a), + edges=(), + capsules=(ca,), + spec_id=spec_id, + ) + specs = { + ca.capsule_id: CapsuleSpec(holes={"a": [1]}, outputs=("a",)), + } + bounds = SolverBounds( + max_tokens=100, + max_nodes=100, + max_depth=10, + max_backtracks=10, + max_verifier_calls=10, + ) + result = solve_capsule_graph( + graph, + builder=FakeBuilder(specs), + provider=_provider(specs, bounds), + terminal_checker=FakeTerminalChecker(), + summary_extractor=FakeSummaryExtractor(specs), + materializer=make_materializer(), + global_verifier=make_global_verifier(lambda _: False), + bounds=bounds, + ) + recovered = CapsuleSolveResult.from_dict(result.to_dict()) + assert recovered.status == result.status + assert recovered.counters.capsule_count == result.counters.capsule_count + + +def test_capsule_solver_is_torch_free(): + root = Path(__file__).parents[2] + code = """ +import sys +assert "torch" not in sys.modules +from slm_training.dsl.solver import capsule_solver # noqa: F401 +assert "torch" not in sys.modules +""" + env = {**os.environ, "PYTHONPATH": str(root / "src")} + subprocess.run([sys.executable, "-c", code], check=True, cwd=root, env=env)