[ISSUE-07] Implement CompatibilitySelector and deterministic route scoring

src/buildcompiler/inventory/__init__.py

@@ -1,5 +1,14 @@
 """Inventory package exports for deterministic lookup/indexing contracts."""
 
+from .compatibility import Lvl1Route, Lvl2Route, RouteScore, RouteSelection
 from .inventory import Inventory
+from .selector import CompatibilitySelector
 
-__all__ = ["Inventory"]
+__all__ = [
+    "CompatibilitySelector",
+    "Inventory",
+    "Lvl1Route",
+    "Lvl2Route",
+    "RouteScore",
+    "RouteSelection",
+]

src/buildcompiler/inventory/compatibility.py

@@ -0,0 +1,58 @@
+"""Deterministic compatibility route models and score ordering."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from buildcompiler.domain import IndexedBackbone, IndexedPlasmid
+
+
+@dataclass(frozen=True)
+class RouteScore:
+    missing_required_products: int = 0
+    missing_domestications: int = 0
+    missing_lvl1_plasmids: int = 0
+    generated_or_planned_materials: int = 0
+    lower_material_state_penalty: int = 0
+    constraint_violations: int = 0
+    total_assemblies: int = 0
+    identity_tiebreak: tuple[str, ...] = ()
+
+    def sort_key(self) -> tuple[int, int, int, int, int, int, int, tuple[str, ...]]:
+        """Lower sort key is a better route."""
+        return (
+            self.constraint_violations,
+            self.missing_required_products,
+            self.missing_domestications,
+            self.missing_lvl1_plasmids,
+            self.generated_or_planned_materials,
+            self.lower_material_state_penalty,
+            self.total_assemblies,
+            self.identity_tiebreak,
+        )
+
+
+@dataclass(frozen=True)
+class Lvl1Route:
+    request_id: str
+    part_identities: tuple[str, ...]
+    selected_part_plasmids: tuple[IndexedPlasmid, ...]
+    missing_part_identities: tuple[str, ...]
+    backbone: IndexedBackbone | None
+    score: RouteScore
+
+
+@dataclass(frozen=True)
+class Lvl2Route:
+    request_id: str
+    region_order: tuple[str, ...]
+    selected_lvl1_plasmids: tuple[IndexedPlasmid, ...]
+    missing_region_identities: tuple[str, ...]
+    backbone: IndexedBackbone | None
+    score: RouteScore
+
+
+@dataclass(frozen=True)
+class RouteSelection:
+    selected: Lvl1Route | Lvl2Route | None
+    rejected: tuple[Lvl1Route | Lvl2Route, ...] = ()

src/buildcompiler/inventory/selector.py

@@ -0,0 +1,165 @@
+"""Deterministic compatibility selector for lvl1/lvl2 route selection."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from itertools import permutations
+from typing import Any
+
+from buildcompiler.api import BuildOptions
+from buildcompiler.domain import BuildStage, MaterialState
+from buildcompiler.inventory.compatibility import Lvl1Route, Lvl2Route, RouteScore, RouteSelection
+from buildcompiler.inventory.inventory import Inventory
+
+
+_STATE_RANK = {
+    MaterialState.PLANNED: 0,
+    MaterialState.GENERATED: 1,
+    MaterialState.ASSEMBLED: 2,
+    MaterialState.TRANSFORMED: 3,
+    MaterialState.PLATED: 4,
+}
+
+
+class CompatibilitySelector:
+    def __init__(self, inventory: Inventory, *, options: BuildOptions | None = None) -> None:
+        self.inventory = inventory
+        self.options = options or BuildOptions()
+
+    def _is_generated_or_planned(self, plasmid: Any) -> bool:
+        source = (plasmid.metadata or {}).get("source", "")
+        if source:
+            return source in {"generated", "planned"}
+        return plasmid.state in {MaterialState.PLANNED, MaterialState.GENERATED}
+
+    def _constraint_filter(self, items: list[Any], constraints: Mapping[str, Any]) -> list[Any]:
+        allowed = set(constraints.get("allowed_identities", []))
+        forbidden = set(constraints.get("forbidden_identities", []))
+        antibiotic = constraints.get("antibiotic")
+        out = []
+        for item in items:
+            if allowed and item.identity not in allowed:
+                continue
+            if item.identity in forbidden:
+                continue
+            if antibiotic and item.metadata.get("antibiotic") != antibiotic:
+                continue
+            out.append(item)
+        return out
+
+    def _best_candidate(self, candidates: list[Any], constraints: Mapping[str, Any]) -> Any | None:
+        filtered = self._constraint_filter(candidates, constraints)
+        if not filtered:
+            return None
+
+        prefer_existing = self.options.selection.prefer_existing_collection_material
+        prefer_state = self.options.selection.prefer_higher_material_state
+
+        def _key(p: Any) -> tuple[int, int, str]:
+            generated_penalty = int(prefer_existing and self._is_generated_or_planned(p))
+            state_penalty = -_STATE_RANK[p.state] if prefer_state else 0
+            return (generated_penalty, state_penalty, p.identity)
+
+        return sorted(filtered, key=_key)[0]
+
+    def select_lvl1_route(self, *, request_id: str, part_identities: Sequence[str], constraints: Mapping[str, Any] | None = None) -> RouteSelection:
+        active_constraints = constraints or {}
+        selected = []
+        missing = []
+        for part_identity in part_identities:
+            candidates = self.inventory.find_single_part_plasmids(part_identity, antibiotic=active_constraints.get("antibiotic"))
+            choice = self._best_candidate(candidates, active_constraints)
+            if choice is None:
+                missing.append(part_identity)
+            else:
+                selected.append(choice)
+
+        backbone = self.inventory.find_backbone(
+            fusion_sites=tuple(active_constraints["fusion_sites"]) if "fusion_sites" in active_constraints else None,
+            antibiotic=active_constraints.get("antibiotic"),
+            stage=BuildStage.ASSEMBLY_LVL1,
+        )
+        score = RouteScore(
+            missing_required_products=len(missing),
+            missing_domestications=len(missing),
+            generated_or_planned_materials=sum(1 for p in selected if self._is_generated_or_planned(p)),
+            lower_material_state_penalty=sum((_STATE_RANK[MaterialState.PLATED] - _STATE_RANK[p.state]) for p in selected)
+            if self.options.selection.prefer_higher_material_state
+            else 0,
+            identity_tiebreak=tuple(sorted(p.identity for p in selected)) + tuple(missing),
+        )
+        route = Lvl1Route(request_id, tuple(part_identities), tuple(selected), tuple(missing), backbone, score)
+        return RouteSelection(selected=route, rejected=())
+
+    def select_lvl2_route(self, *, request_id: str, region_identities: Sequence[str], constraints: Mapping[str, Any] | None = None) -> RouteSelection:
+        active_constraints = constraints or {}
+        max_regions = self.options.planning.lvl2_search.max_exhaustive_region_count
+        allow_large = self.options.planning.lvl2_search.allow_large_order_search
+
+        if "region_order" in active_constraints:
+            constrained_order = tuple(active_constraints["region_order"])
+            requested_regions = tuple(region_identities)
+            if sorted(constrained_order) != sorted(requested_regions):
+                blocked = Lvl2Route(
+                    request_id=request_id,
+                    region_order=constrained_order,
+                    selected_lvl1_plasmids=(),
+                    missing_region_identities=requested_regions,
+                    backbone=None,
+                    score=RouteScore(
+                        missing_required_products=len(requested_regions),
+                        missing_lvl1_plasmids=len(requested_regions),
+                        constraint_violations=1,
+                        identity_tiebreak=requested_regions,
+                    ),
+                )
+                return RouteSelection(selected=None, rejected=(blocked,))
+            orders = [constrained_order]
+        elif len(region_identities) > max_regions and not allow_large:
+            blocked = Lvl2Route(
+                request_id=request_id,
+                region_order=tuple(region_identities),
+                selected_lvl1_plasmids=(),
+                missing_region_identities=tuple(region_identities),
+                backbone=None,
+                score=RouteScore(
+                    missing_required_products=len(region_identities),
+                    missing_lvl1_plasmids=len(region_identities),
+                    constraint_violations=1,
+                    identity_tiebreak=tuple(region_identities),
+                ),
+            )
+            return RouteSelection(selected=None, rejected=(blocked,))
+        else:
+            orders = sorted(set(permutations(region_identities)))
+
+        routes = []
+        for order in orders:
+            selected = []
+            missing = []
+            for region in order:
+                candidates = self.inventory.find_lvl1_region_plasmids(region)
+                choice = self._best_candidate(candidates, active_constraints)
+                if choice is None:
+                    missing.append(region)
+                else:
+                    selected.append(choice)
+            score = RouteScore(
+                missing_required_products=len(missing),
+                missing_lvl1_plasmids=len(missing),
+                generated_or_planned_materials=sum(1 for p in selected if self._is_generated_or_planned(p)),
+                lower_material_state_penalty=sum((_STATE_RANK[MaterialState.PLATED] - _STATE_RANK[p.state]) for p in selected)
+                if self.options.selection.prefer_higher_material_state
+                else 0,
+                total_assemblies=int(bool(missing)),
+                identity_tiebreak=tuple(p.identity for p in selected) + tuple(missing),
+            )
+            backbone = self.inventory.find_backbone(
+                fusion_sites=tuple(active_constraints["fusion_sites"]) if "fusion_sites" in active_constraints else None,
+                antibiotic=active_constraints.get("antibiotic"),
+                stage=BuildStage.ASSEMBLY_LVL2,
+            )
+            routes.append(Lvl2Route(request_id, tuple(order), tuple(selected), tuple(missing), backbone, score))
+
+        ranked = sorted(routes, key=lambda r: r.score.sort_key())
+        return RouteSelection(selected=ranked[0] if ranked else None, rejected=tuple(ranked[1:4]))

tests/unit/inventory/test_compatibility.py

@@ -0,0 +1,13 @@
+from buildcompiler.inventory import RouteScore
+
+
+def test_route_score_prefers_fewer_missing_domestications():
+    assert RouteScore(missing_domestications=0).sort_key() < RouteScore(missing_domestications=1).sort_key()
+
+
+def test_route_score_prefers_fewer_missing_lvl1_plasmids():
+    assert RouteScore(missing_lvl1_plasmids=0).sort_key() < RouteScore(missing_lvl1_plasmids=1).sort_key()
+
+
+def test_route_score_uses_stable_identity_tiebreak():
+    assert RouteScore(identity_tiebreak=("a",)).sort_key() < RouteScore(identity_tiebreak=("b",)).sort_key()

tests/unit/inventory/test_selector.py

@@ -0,0 +1,73 @@
+from buildcompiler.api import BuildOptions
+from buildcompiler.domain import IndexedPlasmid, MaterialState
+from buildcompiler.inventory import CompatibilitySelector, Inventory
+
+
+def _plasmid(identity: str, insert: str, *, state=MaterialState.PLANNED, source="collection") -> IndexedPlasmid:
+    return IndexedPlasmid(
+        identity=identity,
+        state=state,
+        metadata={"insert_identities": [insert], "source": source, "antibiotic": "Ampicillin"},
+    )
+
+
+def test_lvl1_missing_parts_are_reported_not_raised():
+    inv = Inventory(plasmids=[_plasmid("https://e/p1", "https://e/part1")])
+    sel = CompatibilitySelector(inv)
+    route = sel.select_lvl1_route(request_id="r1", part_identities=["https://e/part1", "https://e/part2"]).selected
+    assert route is not None
+    assert route.missing_part_identities == ("https://e/part2",)
+
+
+def test_lvl1_prefers_existing_material_in_tie():
+    inv = Inventory(
+        plasmids=[
+            _plasmid("https://e/a", "https://e/part", source="generated", state=MaterialState.GENERATED),
+            _plasmid("https://e/b", "https://e/part", source="collection", state=MaterialState.GENERATED),
+        ]
+    )
+    sel = CompatibilitySelector(inv)
+    route = sel.select_lvl1_route(request_id="r1", part_identities=["https://e/part"]).selected
+    assert route.selected_part_plasmids[0].identity == "https://e/b"
+
+
+def test_lvl1_hard_constraints_override_selection_preference():
+    inv = Inventory(plasmids=[_plasmid("https://e/a", "https://e/part", source="generated")])
+    opts = BuildOptions()
+    opts.selection.prefer_existing_collection_material = True
+    sel = CompatibilitySelector(inv, options=opts)
+    route = sel.select_lvl1_route(
+        request_id="r1",
+        part_identities=["https://e/part"],
+        constraints={"allowed_identities": ["https://e/a"]},
+    ).selected
+    assert route.selected_part_plasmids[0].identity == "https://e/a"
+
+
+def test_lvl2_large_order_search_not_silent_without_opt_in():
+    inv = Inventory()
+    sel = CompatibilitySelector(inv)
+    out = sel.select_lvl2_route(request_id="r2", region_identities=["a", "b", "c", "d", "e"])
+    assert out.selected is None
+    assert out.rejected
+
+
+def test_lvl2_rejected_alternatives_capped_at_3():
+    inv = Inventory(plasmids=[_plasmid(f"https://e/p{i}", f"https://e/r{i}") for i in range(4)])
+    sel = CompatibilitySelector(inv)
+    out = sel.select_lvl2_route(request_id="r3", region_identities=["https://e/r0", "https://e/r1", "https://e/r2", "https://e/r3"])
+    assert out.selected is not None
+    assert len(out.rejected) == 3
+
+
+def test_lvl2_constrained_order_must_match_requested_regions():
+    inv = Inventory(plasmids=[_plasmid("https://e/p0", "https://e/r0"), _plasmid("https://e/p1", "https://e/r1")])
+    sel = CompatibilitySelector(inv)
+    out = sel.select_lvl2_route(
+        request_id="r4",
+        region_identities=["https://e/r0", "https://e/r1"],
+        constraints={"region_order": ["https://e/r0"]},
+    )
+    assert out.selected is None
+    assert out.rejected
+    assert out.rejected[0].missing_region_identities == ("https://e/r0", "https://e/r1")