[ISSUE-08] Port AssemblyService behind clean SBOL assembly interface

src/buildcompiler/sbol/__init__.py

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

src/buildcompiler/sbol/assembly.py

@@ -0,0 +1,185 @@
+"""SBOL assembly service wrapping legacy Golden Gate behavior."""
+
+from dataclasses import dataclass, field
+
+import sbol2
+
+from buildcompiler.domain import (
+    BuildStage,
+    IndexedBackbone,
+    IndexedPlasmid,
+    IndexedReagent,
+    MaterialState,
+)
+from buildcompiler.sbol2build import Assembly
+
+
+@dataclass
+class _LegacyPlasmidAdapter:
+    plasmid_definition: sbol2.ComponentDefinition
+    plasmid_implementations: list[sbol2.Implementation]
+
+
+@dataclass
+class AssemblyJob:
+    """Normalized assembly inputs plus source/target SBOL documents."""
+
+    stage: BuildStage
+    product_identity: str
+    product_display_id: str
+    part_plasmids: list[IndexedPlasmid]
+    backbone: IndexedBackbone
+    restriction_enzyme: IndexedReagent
+    ligase: IndexedReagent
+    source_document: sbol2.Document
+    target_document: sbol2.Document
+    include_extracted_parts: bool = False
+
+
+@dataclass
+class AssemblySbolResult:
+    """Assembly output contract: normalized products plus stage document."""
+
+    products: list[IndexedPlasmid]
+    stage_document: sbol2.Document
+    activity_identity: str
+    logs: list[str] = field(default_factory=list)
+
+
+class AssemblyService:
+    """Service wrapper that preserves legacy assembly internals behind normalized contracts."""
+
+    def run(self, job: AssemblyJob) -> AssemblySbolResult:
+        if not job.part_plasmids:
+            raise ValueError("AssemblyJob.part_plasmids must contain at least one plasmid")
+
+        legacy_parts = [
+            self._record_to_legacy_plasmid(record, job.source_document, "part_plasmids")
+            for record in job.part_plasmids
+        ]
+        legacy_backbone = self._record_to_legacy_plasmid(
+            IndexedPlasmid(
+                identity=job.backbone.identity,
+                display_id=job.backbone.display_id,
+                name=job.backbone.name,
+                metadata=job.backbone.metadata,
+                sbol_component=job.backbone.sbol_component,
+            ),
+            job.source_document,
+            "backbone",
+        )
+
+        restriction_impl = self._implementation_from_record(
+            job.restriction_enzyme, job.source_document
+        )
+        ligase_impl = self._implementation_from_record(job.ligase, job.source_document)
+
+        composite_prefix = job.product_display_id or job.product_identity.split("/")[-1]
+        legacy_assembly = Assembly(
+            part_plasmids=legacy_parts,
+            backbone_plasmid=legacy_backbone,
+            restriction_enzyme=restriction_impl,
+            ligase=ligase_impl,
+            source_document=job.source_document,
+            final_document=job.target_document,
+            composite_prefix=composite_prefix,
+        )
+        legacy_products, final_doc = legacy_assembly.run(
+            include_extracted_parts=job.include_extracted_parts
+        )
+
+        products = [
+            self._indexed_product_from_legacy_product(plasmid, job)
+            for plasmid in legacy_products
+        ]
+        logs = [
+            f"Assembled {len(products)} product(s) at stage {job.stage.value}.",
+            f"Assembly activity: {legacy_assembly.assembly_activity.identity}",
+        ]
+
+        return AssemblySbolResult(
+            products=products,
+            stage_document=final_doc,
+            activity_identity=legacy_assembly.assembly_activity.identity,
+            logs=logs,
+        )
+
+    def _record_to_legacy_plasmid(
+        self,
+        record: IndexedPlasmid,
+        source_document: sbol2.Document,
+        field_name: str,
+    ) -> _LegacyPlasmidAdapter:
+        component = self._component_from_record(record, source_document, field_name)
+        implementation = self._implementation_from_plasmid_record(record, source_document)
+        return _LegacyPlasmidAdapter(component, [implementation])
+
+    def _component_from_record(
+        self,
+        record: IndexedPlasmid,
+        source_document: sbol2.Document,
+        field_name: str,
+    ) -> sbol2.ComponentDefinition:
+        component = record.sbol_component or source_document.find(record.identity)
+        if component is None:
+            raise ValueError(
+                f"Missing SBOL ComponentDefinition for {field_name} record {record.identity}"
+            )
+        if not isinstance(component, sbol2.ComponentDefinition):
+            raise ValueError(
+                f"{field_name} record {record.identity} must resolve to sbol2.ComponentDefinition"
+            )
+        return component
+
+    def _implementation_from_plasmid_record(
+        self, record: IndexedPlasmid, source_document: sbol2.Document
+    ) -> sbol2.Implementation:
+        impl_identity = record.metadata.get("implementation_identity")
+        implementation = source_document.find(impl_identity) if impl_identity else None
+
+        if implementation is None:
+            component = self._component_from_record(record, source_document, "plasmid")
+            matches = [
+                impl
+                for impl in source_document.implementations
+                if isinstance(impl, sbol2.Implementation) and impl.built == component.identity
+            ]
+            implementation = matches[0] if matches else None
+
+        if implementation is None:
+            raise ValueError(
+                f"Missing SBOL Implementation for plasmid {record.identity}; "
+                "set metadata['implementation_identity'] or include implementation in source_document"
+            )
+        return implementation
+
+    def _implementation_from_record(
+        self, record: IndexedReagent, source_document: sbol2.Document
+    ) -> sbol2.Implementation:
+        impl_identity = record.metadata.get("implementation_identity") or record.identity
+        implementation = source_document.find(impl_identity)
+        if not isinstance(implementation, sbol2.Implementation):
+            raise ValueError(
+                "Missing SBOL Implementation for reagent "
+                f"{record.identity}; expected metadata['implementation_identity'] or identity to resolve"
+            )
+        return implementation
+
+    def _indexed_product_from_legacy_product(
+        self, product, job: AssemblyJob
+    ) -> IndexedPlasmid:
+        component = product.plasmid_definition
+        return IndexedPlasmid(
+            identity=component.identity,
+            display_id=component.displayId,
+            name=component.name,
+            state=MaterialState.GENERATED,
+            roles=list(component.roles),
+            metadata={
+                "source_stage": job.stage.value,
+                "source_product_identity": job.product_identity,
+                "source_product_display_id": job.product_display_id,
+                "assembly_activity_identity": product.plasmid_implementations[0].wasGeneratedBy,
+            },
+            sbol_component=component,
+        )

tests/unit/sbol/test_assembly_service.py

@@ -0,0 +1,81 @@
+import sbol2
+import pytest
+
+from buildcompiler.domain import (
+    BuildStage,
+    IndexedBackbone,
+    IndexedPlasmid,
+    IndexedReagent,
+    MaterialState,
+)
+from buildcompiler.sbol import AssemblyJob, AssemblySbolResult, AssemblyService
+
+
+def test_assembly_service_runs_and_returns_normalized_products(monkeypatch):
+    source = sbol2.Document()
+    product_component = sbol2.ComponentDefinition("assembled_product")
+    source.add(product_component)
+    product_impl = sbol2.Implementation("assembled_product_impl")
+    product_impl.built = product_component.identity
+    source.add(product_impl)
+    product_impl.wasGeneratedBy = "https://example.org/activity/assembly"
+
+    class FakeLegacyAssembly:
+        def __init__(self, **kwargs):
+            self.assembly_activity = sbol2.Activity("fake_assembly")
+
+        def run(self, include_extracted_parts=False):
+            return [type("LegacyProduct", (), {"plasmid_definition": product_component, "plasmid_implementations": [product_impl]})()], source
+
+    monkeypatch.setattr("buildcompiler.sbol.assembly.Assembly", FakeLegacyAssembly)
+
+    service = AssemblyService()
+    result = service.run(
+        AssemblyJob(
+            stage=BuildStage.ASSEMBLY_LVL1,
+            product_identity="https://example.org/products/p001",
+            product_display_id="p001",
+            part_plasmids=[
+                IndexedPlasmid(
+                    identity=product_component.identity,
+                    sbol_component=product_component,
+                    metadata={"implementation_identity": product_impl.identity},
+                )
+            ],
+            backbone=IndexedBackbone(
+                identity=product_component.identity,
+                sbol_component=product_component,
+                metadata={"implementation_identity": product_impl.identity},
+            ),
+            restriction_enzyme=IndexedReagent(identity=product_impl.identity),
+            ligase=IndexedReagent(identity=product_impl.identity),
+            source_document=source,
+            target_document=sbol2.Document(),
+        )
+    )
+
+    assert isinstance(result, AssemblySbolResult)
+    assert result.products
+    assert isinstance(result.products[0], IndexedPlasmid)
+    assert result.products[0].state == MaterialState.GENERATED
+    assert result.activity_identity
+
+
+def test_assembly_service_raises_clear_error_for_missing_component():
+    doc = sbol2.Document()
+    service = AssemblyService()
+
+    with pytest.raises(ValueError, match="Missing SBOL ComponentDefinition"):
+        service.run(
+            AssemblyJob(
+                stage=BuildStage.ASSEMBLY_LVL1,
+                product_identity="https://example.org/products/p001",
+                product_display_id="p001",
+                part_plasmids=[IndexedPlasmid(identity="https://example.org/missing")],
+                backbone=IndexedBackbone(identity="https://example.org/backbone"),
+                restriction_enzyme=IndexedReagent(identity="https://example.org/reagent/re"),
+                ligase=IndexedReagent(identity="https://example.org/reagent/ligase"),
+                source_document=doc,
+                target_document=sbol2.Document(),
+            )
+        )