[ISSUE-06] Implement planner, classifier, and combinatorial expansion

src/buildcompiler/planning/__init__.py

@@ -1 +1,6 @@
-"""Package scaffolding for clean architecture."""
+"""Planning package exports."""
+
+from .full_build_planner import FullBuildPlanner
+from .models import BuildPlan, UnsupportedPlanningRecord
+
+__all__ = ["BuildPlan", "UnsupportedPlanningRecord", "FullBuildPlanner"]

src/buildcompiler/planning/classifier.py

@@ -0,0 +1,112 @@
+"""Design classification helpers for planning."""
+
+from __future__ import annotations
+
+import re
+from collections import Counter
+
+import sbol2
+
+from buildcompiler.domain import BuildRequest, BuildStage, DesignKind
+from buildcompiler.planning.models import UnsupportedPlanningRecord
+from buildcompiler.planning.validation import classify_part_role
+
+RECOMMENDED_LVL1_PARTS = ("promoter", "rbs", "cds", "terminator")
+
+
+def _stable_slug(identity: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", identity.lower()).strip("-")[-48:]
+
+
+def request_id_for(
+    stage: BuildStage,
+    source_identity: str,
+    source_display_id: str | None,
+    *,
+    variant_index: int | None = None,
+) -> str:
+    base = source_display_id or _stable_slug(source_identity)
+    rid = f"{stage.value}:{base}"
+    if variant_index is not None:
+        rid = f"{rid}:v{variant_index}"
+    return rid
+
+
+def classify_non_combinatorial(
+    design: object,
+) -> BuildRequest | UnsupportedPlanningRecord:
+    if isinstance(design, sbol2.ModuleDefinition):
+        return BuildRequest(
+            request_id_for(BuildStage.ASSEMBLY_LVL2, design.identity, design.displayId),
+            BuildStage.ASSEMBLY_LVL2,
+            design.identity,
+            design.displayId,
+            DesignKind.MODULE_DEFINITION,
+        )
+
+    if isinstance(design, sbol2.ComponentDefinition):
+        count = len(design.components)
+        if count > 1:
+            observed_roles: list[str] = []
+            for component in design.components:
+                target = (
+                    component.doc.find(component.definition)
+                    if getattr(component, "doc", None) is not None
+                    else None
+                )
+                if isinstance(target, sbol2.ComponentDefinition):
+                    role = classify_part_role(target)
+                    if role is not None:
+                        observed_roles.append(role)
+
+            counts = Counter(observed_roles)
+            missing = [role for role in RECOMMENDED_LVL1_PARTS if counts[role] != 1]
+            has_role_evidence = len(observed_roles) > 0
+            if count != 4 or (has_role_evidence and missing):
+                return UnsupportedPlanningRecord(
+                    design.identity,
+                    design.displayId,
+                    DesignKind.COMPONENT_DEFINITION,
+                    "Warning: Level-1 planning expects exactly four parts (promoter, RBS, CDS, terminator).",
+                    {
+                        "component_count": count,
+                        "observed_role_counts": {
+                            role: counts.get(role, 0)
+                            for role in RECOMMENDED_LVL1_PARTS
+                        },
+                        "suggested_parts": list(RECOMMENDED_LVL1_PARTS),
+                    },
+                )
+            return BuildRequest(
+                request_id_for(
+                    BuildStage.ASSEMBLY_LVL1, design.identity, design.displayId
+                ),
+                BuildStage.ASSEMBLY_LVL1,
+                design.identity,
+                design.displayId,
+                DesignKind.COMPONENT_DEFINITION,
+            )
+        if count <= 1 and classify_part_role(design) is not None:
+            return BuildRequest(
+                request_id_for(
+                    BuildStage.DOMESTICATION, design.identity, design.displayId
+                ),
+                BuildStage.DOMESTICATION,
+                design.identity,
+                design.displayId,
+                DesignKind.COMPONENT_DEFINITION,
+            )
+        return UnsupportedPlanningRecord(
+            design.identity,
+            design.displayId,
+            DesignKind.COMPONENT_DEFINITION,
+            "ComponentDefinition is single/empty with unsupported role.",
+            {"component_count": count, "roles": list(design.roles)},
+        )
+
+    return UnsupportedPlanningRecord(
+        getattr(design, "identity", str(design)),
+        getattr(design, "displayId", None),
+        DesignKind.UNSUPPORTED,
+        f"Unsupported design type: {type(design).__name__}",
+    )

src/buildcompiler/planning/combinatorial.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+import itertools
+import sbol2
+from buildcompiler.api import BuildOptions
+from buildcompiler.domain import BuildRequest, BuildStage, BuildWarning, DesignKind
+from buildcompiler.planning.classifier import request_id_for
+from buildcompiler.planning.models import UnsupportedPlanningRecord
+from buildcompiler.planning.validation import ROLE_TO_NAME
+
+
+def _collect_variant_sets(derivation):
+    variables = list(derivation.variableComponents)
+    variables.sort(
+        key=lambda variable: (str(getattr(variable, "variable", "")), variable.identity)
+    )
+    return variables, [sorted(list(vc.variants), key=str) for vc in variables]
+
+
+def expand_combinatorial_derivation(
+    derivation: sbol2.CombinatorialDerivation, *, options: BuildOptions
+):
+    warnings = []
+    unsupported = []
+    requests = []
+    variables, variant_sets = _collect_variant_sets(derivation)
+    if not variant_sets or any(len(v) == 0 for v in variant_sets):
+        unsupported.append(
+            UnsupportedPlanningRecord(
+                derivation.identity,
+                derivation.displayId,
+                DesignKind.COMBINATORIAL_DERIVATION,
+                "Variable component has no listed variants.",
+            )
+        )
+        return requests, unsupported, warnings
+    total = 1
+    for v in variant_sets:
+        total *= len(v)
+    if (
+        total > options.planning.combinatorial.max_variants
+        and not options.planning.combinatorial.allow_large_expansion
+    ):
+        warnings.append(
+            BuildWarning(
+                "planning.combinatorial.expansion_blocked",
+                "Combinatorial expansion exceeds max_variants and is blocked.",
+                BuildStage.ASSEMBLY_LVL1,
+                derivation.identity,
+                {
+                    "variant_count": total,
+                    "max_variants": options.planning.combinatorial.max_variants,
+                },
+            )
+        )
+        unsupported.append(
+            UnsupportedPlanningRecord(
+                derivation.identity,
+                derivation.displayId,
+                DesignKind.COMBINATORIAL_DERIVATION,
+                "Combinatorial expansion blocked by max_variants.",
+                {"variant_count": total},
+            )
+        )
+        return requests, unsupported, warnings
+    doc = derivation.doc
+    valid = 0
+    for idx, chosen in enumerate(itertools.product(*variant_sets)):
+        roles = []
+        for pid in chosen:
+            obj = doc.find(pid) if doc is not None else None
+            if isinstance(obj, sbol2.ComponentDefinition):
+                matching = [ROLE_TO_NAME[r] for r in obj.roles if r in ROLE_TO_NAME]
+                if len(matching) == 1:
+                    roles.append(matching[0])
+        if sorted(roles) != ["cds", "promoter", "rbs", "terminator"]:
+            warnings.append(
+                BuildWarning(
+                    "planning.combinatorial.invalid_variant",
+                    "Skipping invalid combinatorial variant.",
+                    BuildStage.ASSEMBLY_LVL1,
+                    derivation.identity,
+                    {"variant_index": idx},
+                )
+            )
+            continue
+        requests.append(
+            BuildRequest(
+                request_id_for(
+                    BuildStage.ASSEMBLY_LVL1,
+                    derivation.identity,
+                    derivation.displayId,
+                    variant_index=idx,
+                ),
+                BuildStage.ASSEMBLY_LVL1,
+                derivation.identity,
+                derivation.displayId,
+                DesignKind.COMBINATORIAL_DERIVATION,
+                parent_group=derivation.identity,
+                variant_index=idx,
+                constraints={"part_order": list(chosen)},
+            )
+        )
+        valid += 1
+    if valid == 0:
+        unsupported.append(
+            UnsupportedPlanningRecord(
+                derivation.identity,
+                derivation.displayId,
+                DesignKind.COMBINATORIAL_DERIVATION,
+                "All combinatorial variants were invalid.",
+            )
+        )
+    return requests, unsupported, warnings

src/buildcompiler/planning/full_build_planner.py

@@ -0,0 +1,49 @@
+"""Pure full-build planner orchestration."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import sbol2
+
+from buildcompiler.api import BuildOptions
+from buildcompiler.planning.classifier import classify_non_combinatorial
+from buildcompiler.planning.combinatorial import expand_combinatorial_derivation
+from buildcompiler.planning.models import BuildPlan, UnsupportedPlanningRecord
+from buildcompiler.sbol import SbolResolver
+
+
+class FullBuildPlanner:
+    def __init__(
+        self,
+        *,
+        options: BuildOptions | None = None,
+        resolver: SbolResolver | None = None,
+    ):
+        self.options = options or BuildOptions()
+        self.resolver = resolver
+
+    def plan(
+        self, abstract_designs: Iterable[object], *, options: BuildOptions | None = None
+    ) -> BuildPlan:
+        active = options or self.options
+        out = BuildPlan()
+        for design in abstract_designs:
+            if isinstance(design, sbol2.CombinatorialDerivation):
+                reqs, unsupported, warnings = expand_combinatorial_derivation(
+                    design, options=active
+                )
+                out.lvl1_requests.extend(reqs)
+                out.unsupported.extend(unsupported)
+                out.warnings.extend(warnings)
+                continue
+            classified = classify_non_combinatorial(design)
+            if isinstance(classified, UnsupportedPlanningRecord):
+                out.unsupported.append(classified)
+            elif classified.stage.value == "assembly_lvl2":
+                out.lvl2_requests.append(classified)
+            elif classified.stage.value == "assembly_lvl1":
+                out.lvl1_requests.append(classified)
+            else:
+                out.domestication_requests.append(classified)
+        return out

src/buildcompiler/planning/models.py

@@ -0,0 +1,24 @@
+"""Planning contracts for full-build planning output."""
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from buildcompiler.domain import BuildRequest, BuildWarning, DesignKind
+
+
+@dataclass
+class UnsupportedPlanningRecord:
+    source_identity: str
+    source_display_id: str | None
+    source_kind: DesignKind
+    reason: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class BuildPlan:
+    lvl2_requests: list[BuildRequest] = field(default_factory=list)
+    lvl1_requests: list[BuildRequest] = field(default_factory=list)
+    domestication_requests: list[BuildRequest] = field(default_factory=list)
+    unsupported: list[UnsupportedPlanningRecord] = field(default_factory=list)
+    warnings: list[BuildWarning] = field(default_factory=list)