[ISSUE-10] Implement domestication planner/service with approvals

src/buildcompiler/planning/__init__.py

@@ -1,6 +1,14 @@
 """Planning package exports."""
 
+from .domestication import DomesticationPlan, DomesticationPlanner, SequenceEditProposal
 from .full_build_planner import FullBuildPlanner
 from .models import BuildPlan, UnsupportedPlanningRecord
 
-__all__ = ["BuildPlan", "UnsupportedPlanningRecord", "FullBuildPlanner"]
+__all__ = [
+    "BuildPlan",
+    "UnsupportedPlanningRecord",
+    "FullBuildPlanner",
+    "DomesticationPlan",
+    "DomesticationPlanner",
+    "SequenceEditProposal",
+]

src/buildcompiler/planning/domestication.py

@@ -0,0 +1,83 @@
+"""Deterministic domestication planning helpers."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+import sbol2
+
+from buildcompiler.domain import BuildStage, BuildWarning
+from buildcompiler.planning.validation import classify_part_role
+
+
+@dataclass
+class SequenceEditProposal:
+    source_identity: str
+    enzyme_name: str
+    site_sequence: str
+    position: int
+    original_sequence: str
+    proposed_sequence: str
+    reason: str
+    approved: bool = False
+
+
+@dataclass
+class DomesticationPlan:
+    part_identity: str
+    part_display_id: str | None
+    part_role: str
+    backbone_identity: str | None = None
+    restriction_enzyme_name: str = "BsaI"
+    ligase_name: str = "T4_DNA_ligase"
+    sequence_edit_proposals: list[SequenceEditProposal] = field(default_factory=list)
+    warnings: list[BuildWarning] = field(default_factory=list)
+
+
+class DomesticationPlanner:
+    """Pure planner for domestication requirements for a single part."""
+
+    _BSAI_SITES = ("GGTCTC", "GAGACC")
+
+    def plan(self, part_component: sbol2.ComponentDefinition) -> DomesticationPlan:
+        part_role = classify_part_role(part_component)
+        if part_role is None:
+            raise ValueError(
+                f"Unsupported domestication role for part {part_component.identity}; expected promoter/rbs/cds/terminator"
+            )
+
+        sequence = self._resolve_sequence(part_component)
+        proposals: list[SequenceEditProposal] = []
+        for site in self._BSAI_SITES:
+            start = 0
+            while True:
+                index = sequence.find(site, start)
+                if index < 0:
+                    break
+                proposals.append(
+                    SequenceEditProposal(
+                        source_identity=part_component.identity,
+                        enzyme_name="BsaI",
+                        site_sequence=site,
+                        position=index,
+                        original_sequence=site,
+                        proposed_sequence=f"{site[:-1]}A",
+                        reason="Internal BsaI recognition site detected; human-reviewed edit required.",
+                    )
+                )
+                start = index + 1
+
+        return DomesticationPlan(
+            part_identity=part_component.identity,
+            part_display_id=part_component.displayId,
+            part_role=part_role,
+            sequence_edit_proposals=proposals,
+        )
+
+    def _resolve_sequence(self, part_component: sbol2.ComponentDefinition) -> str:
+        for sequence_ref in part_component.sequences:
+            sequence_obj = part_component.doc.find(sequence_ref) if part_component.doc else None
+            elements = getattr(sequence_obj, "elements", None)
+            if isinstance(elements, str) and elements:
+                return elements.upper()
+        raise ValueError(f"Part {part_component.identity} is missing a usable DNA sequence")

src/buildcompiler/sbol/__init__.py

@@ -1,12 +1,16 @@
 """SBOL package exports for clean architecture contracts."""
 
 from .assembly import AssemblyJob, AssemblySbolResult, AssemblyService
+from .domestication import DomesticationJob, DomesticationSbolResult, DomesticationService
 from .resolver import PullPolicy, SbolResolver
 
 __all__ = [
     "AssemblyJob",
     "AssemblySbolResult",
     "AssemblyService",
+    "DomesticationJob",
+    "DomesticationSbolResult",
+    "DomesticationService",
     "PullPolicy",
     "SbolResolver",
 ]

src/buildcompiler/sbol/domestication.py

@@ -0,0 +1,83 @@
+"""SBOL domestication service."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import sbol2
+
+from buildcompiler.domain import IndexedBackbone, IndexedPlasmid, IndexedReagent, MaterialState
+
+
+@dataclass
+class DomesticationJob:
+    part_identity: str
+    part_display_id: str | None
+    part_component: Any
+    backbone: IndexedBackbone
+    restriction_enzyme: IndexedReagent
+    ligase: IndexedReagent
+    source_document: Any
+    target_document: Any
+    sequence_edit_proposals: list[Any] = field(default_factory=list)
+
+
+@dataclass
+class DomesticationSbolResult:
+    product: IndexedPlasmid
+    stage_document: Any
+    artifacts: dict[str, Any] = field(default_factory=dict)
+    logs: list[str] = field(default_factory=list)
+
+
+class DomesticationService:
+    def run(self, job: DomesticationJob) -> DomesticationSbolResult:
+        component = self._ensure_component(job.part_component)
+        product_identity = f"{component.identity}/domesticated"
+        product_display_id = f"{job.part_display_id or component.displayId or component.identity.rsplit('/', 1)[-1]}_lvl0"
+
+        product_component = sbol2.ComponentDefinition(product_identity)
+        product_component.displayId = product_display_id
+        product_component.name = f"Domesticated {component.displayId or component.identity.rsplit('/', 1)[-1]}"
+        product_component.roles = list(component.roles)
+        job.target_document.addComponentDefinition(product_component)
+
+        implementation_identity = f"{product_identity}_implementation"
+        product_implementation = sbol2.Implementation(implementation_identity)
+        product_implementation.built = product_component.identity
+        job.target_document.addImplementation(product_implementation)
+
+        metadata = {
+            "source_stage": "domestication",
+            "source_part_identity": job.part_identity,
+            "insert_identities": [job.part_identity],
+            "implementation_identity": product_implementation.identity,
+            "backbone_identity": job.backbone.identity,
+            "restriction_enzyme": {
+                "identity": job.restriction_enzyme.identity,
+                "name": job.restriction_enzyme.name,
+            },
+            "ligase": {"identity": job.ligase.identity, "name": job.ligase.name},
+            "sequence_edit_proposals": [proposal.__dict__.copy() for proposal in job.sequence_edit_proposals],
+        }
+        product = IndexedPlasmid(
+            identity=product_component.identity,
+            display_id=product_display_id,
+            name=product_component.name,
+            state=MaterialState.GENERATED,
+            roles=list(component.roles),
+            metadata=metadata,
+            sbol_component=product_component,
+        )
+        return DomesticationSbolResult(
+            product=product,
+            stage_document=job.target_document,
+            artifacts={"domestication": metadata},
+            logs=[f"Generated domesticated lvl0 product {product_identity}."],
+        )
+
+    def _ensure_component(self, component: Any) -> sbol2.ComponentDefinition:
+        if not isinstance(component, sbol2.ComponentDefinition):
+            raise ValueError("DomesticationJob.part_component must be an sbol2.ComponentDefinition")
+        return component

src/buildcompiler/stages/__init__.py

@@ -1,5 +1,6 @@
 """Stage exports."""
 
 from .assembly_lvl1 import AssemblyLvl1Stage
+from .domestication import DomesticationStage
 
-__all__ = ["AssemblyLvl1Stage"]
+__all__ = ["AssemblyLvl1Stage", "DomesticationStage"]

src/buildcompiler/stages/domestication.py

@@ -0,0 +1,149 @@
+"""Domestication stage orchestration."""
+
+from __future__ import annotations
+
+import sbol2
+
+from buildcompiler.api import BuildOptions, ProtocolMode
+from buildcompiler.domain import (
+    ApprovalStatus,
+    BuildRequest,
+    BuildStage,
+    BuildWarning,
+    MissingBuildInput,
+    RequiredApproval,
+    StageResult,
+    StageStatus,
+)
+from buildcompiler.inventory import Inventory
+from buildcompiler.planning.domestication import DomesticationPlanner
+from buildcompiler.sbol.domestication import DomesticationJob, DomesticationService
+
+
+class DomesticationStage:
+    def __init__(
+        self,
+        *,
+        inventory: Inventory,
+        domestication_planner: DomesticationPlanner | None = None,
+        domestication_service: DomesticationService | None = None,
+        options: BuildOptions | None = None,
+    ) -> None:
+        self.inventory = inventory
+        self.domestication_planner = domestication_planner or DomesticationPlanner()
+        self.domestication_service = domestication_service or DomesticationService()
+        self.options = options or BuildOptions()
+
+    def run(self, request: BuildRequest, *, source_document: sbol2.Document, target_document: sbol2.Document) -> StageResult:
+        part_component = source_document.find(request.source_identity)
+        if not isinstance(part_component, sbol2.ComponentDefinition):
+            for candidate in source_document.componentDefinitions:
+                if (
+                    candidate.identity == request.source_identity
+                    or candidate.persistentIdentity == request.source_identity
+                    or candidate.displayId == request.source_identity
+                    or candidate.identity.endswith(f"/{request.source_identity}/1")
+                    or candidate.persistentIdentity.endswith(f"/{request.source_identity}")
+                ):
+                    part_component = candidate
+                    break
+        if not isinstance(part_component, sbol2.ComponentDefinition):
+            return StageResult(
+                id=f"{request.id}:{BuildStage.DOMESTICATION.value}",
+                stage=BuildStage.DOMESTICATION,
+                status=StageStatus.FAILED,
+                request_ids=[request.id],
+                logs=[f"Failed domestication: source part {request.source_identity} not found."],
+            )
+        try:
+            plan = self.domestication_planner.plan(part_component)
+        except ValueError as exc:
+            return StageResult(
+                id=f"{request.id}:{BuildStage.DOMESTICATION.value}",
+                stage=BuildStage.DOMESTICATION,
+                status=StageStatus.FAILED,
+                request_ids=[request.id],
+                warnings=[BuildWarning(code="domestication.invalid_input", message=str(exc), stage=BuildStage.DOMESTICATION, source_identity=request.source_identity)],
+                logs=[str(exc)],
+            )
+
+        missing_inputs: list[MissingBuildInput] = []
+        backbone = self.inventory.find_backbone(stage=BuildStage.DOMESTICATION)
+        if backbone is None:
+            missing_inputs.append(MissingBuildInput(BuildStage.DOMESTICATION, request.source_identity, "backbone", None, "backbone", "fatal", "No domestication backbone found in inventory."))
+
+        restriction = self.inventory.find_restriction_enzyme(self.options.reagents.default_restriction_enzyme)
+        if restriction is None:
+            missing_inputs.append(MissingBuildInput(BuildStage.DOMESTICATION, request.source_identity, self.options.reagents.default_restriction_enzyme, self.options.reagents.default_restriction_enzyme, "restriction_enzyme", "fatal", "Required domestication restriction enzyme missing from inventory."))
+
+        ligase = self.inventory.find_ligase(self.options.reagents.default_ligase)
+        if ligase is None:
+            missing_inputs.append(MissingBuildInput(BuildStage.DOMESTICATION, request.source_identity, self.options.reagents.default_ligase, self.options.reagents.default_ligase, "ligase", "fatal", "Required domestication ligase missing from inventory."))
+
+        if missing_inputs:
+            return StageResult(
+                id=f"{request.id}:{BuildStage.DOMESTICATION.value}",
+                stage=BuildStage.DOMESTICATION,
+                status=StageStatus.BLOCKED,
+                request_ids=[request.id],
+                missing_inputs=missing_inputs,
+                logs=["Domestication blocked on missing backbone/reagents."],
+                protocol_artifacts={"sequence_edit_proposals": [proposal.__dict__.copy() for proposal in plan.sequence_edit_proposals]},
+            )
+
+        approvals: list[RequiredApproval] = []
+        if plan.sequence_edit_proposals:
+            approval_id = f"domestication-edit:{request.source_identity}"
+            process_approved = "domestication_sequence_edit" in self.options.approvals.approved_processes
+            id_approved = approval_id in self.options.approvals.approved_approval_ids
+            allow_edits = self.options.domestication.allow_sequence_domestication_edits
+            protocol_mode_active = self.options.protocol.mode != ProtocolMode.NONE
+            if (not allow_edits) or (protocol_mode_active and not (process_approved or id_approved)):
+                approvals.append(
+                    RequiredApproval(
+                        status=ApprovalStatus.REQUIRED,
+                        process="domestication_sequence_edit",
+                        reason="Sequence edits were proposed and require explicit human approval.",
+                        metadata={
+                            "approval_id": approval_id,
+                            "part_identity": request.source_identity,
+                            "proposals": [proposal.__dict__.copy() for proposal in plan.sequence_edit_proposals],
+                        },
+                    )
+                )
+
+        if approvals:
+            return StageResult(
+                id=f"{request.id}:{BuildStage.DOMESTICATION.value}",
+                stage=BuildStage.DOMESTICATION,
+                status=StageStatus.BLOCKED,
+                request_ids=[request.id],
+                required_approvals=approvals,
+                protocol_artifacts={"sequence_edit_proposals": [proposal.__dict__.copy() for proposal in plan.sequence_edit_proposals]},
+                logs=["Domestication blocked pending sequence-edit approval."],
+            )
+
+        result = self.domestication_service.run(
+            DomesticationJob(
+                part_identity=request.source_identity,
+                part_display_id=request.source_display_id,
+                part_component=part_component,
+                backbone=backbone,
+                restriction_enzyme=restriction,
+                ligase=ligase,
+                source_document=source_document,
+                target_document=target_document,
+                sequence_edit_proposals=plan.sequence_edit_proposals,
+            )
+        )
+        self.inventory.add_generated_product(result.product)
+        return StageResult(
+            id=f"{request.id}:{BuildStage.DOMESTICATION.value}",
+            stage=BuildStage.DOMESTICATION,
+            status=StageStatus.SUCCESS,
+            request_ids=[request.id],
+            products=[result.product],
+            sbol_document=result.stage_document,
+            protocol_artifacts={"sequence_edit_proposals": [proposal.__dict__.copy() for proposal in plan.sequence_edit_proposals], **result.artifacts},
+            logs=result.logs,
+        )