diff --git a/CONTRACT.md b/CONTRACT.md index 0fe3500..4d79299 100644 --- a/CONTRACT.md +++ b/CONTRACT.md @@ -47,3 +47,18 @@ complete the action. to run before `snapgene_open` when a map is requested. - `workflow_status` is the complete reproducibility log and must remain JSON serializable. + +## Gibson MVP + +`BiologyService.gibson_assemble(fragments, *, overlaps, name, circular=True, +min_overlap=15)` returns a new GenBank `ConstructRef`. Fragments are ordered, +pre-oriented linear inputs; overlap lengths explicitly describe exact suffix/prefix +junctions (including last-to-first for circular products). No primer design or +assembly-efficiency prediction is performed. Every fragment must retain a non-overlap +interior. A/C/G/T only. Source features and checked junctions are retained in the +product; provenance is saved in its GenBank comment. + +The workflow/MCP `gibson_assemble` operation can start from an empty workflow, +records all fragment references/settings, and resets validation and SnapGene state. +Validate before saving. `DNA_MAKER_SNAPGENE_ADAPTER` is optional for biology-only +startup; explicitly configured invalid adapters still report configuration errors. diff --git a/README.md b/README.md index 958e5ff..6c402c0 100644 --- a/README.md +++ b/README.md @@ -340,3 +340,58 @@ directory when needed. > Read inputs/pEGFP-N1.gb. Report its length, topology, and annotated features. Find EcoRI and BamHI restriction > sites, clearly stating the coordinate convention. Run validation and report each check, then show the workflow > status. Do not modify, save, or open any files. + +## Gibson assembly without SnapGene + +A local MVP simulates the product of **ordered, pre-oriented linear fragments +with exact, pre-existing overlaps**. It checks every junction, removes duplicate +overlap sequence, maps annotations (including across the circular origin), and +writes a GenBank product with junction annotations and provenance. It does not +design primers, choose fragment order/orientation, or predict experimental yield. + +Run the self-contained synthetic presentation demo: + +```sh +uv run python -m dnamaker.gibson_demo +``` + +It creates a fresh `artifacts/gibson-demo-*/` directory containing input GenBank +files, `assembled.gb`, `verification.json`, and `workflow.json`. The independently +checked result is 160 bp: 110 bp vector + 90 bp insert − two 20 bp overlaps. +If launching Python from an AppImage shell causes interpreter issues, use +`env -u APPIMAGE .venv/bin/python -m dnamaker.gibson_demo`. + +For the MCP server on Linux, only the biology adapter is required: + +```sh +DNA_MAKER_BIOLOGY_ADAPTER=dnamaker.service:create_biology_adapter uv run dna-maker-mcp +``` + +Call `start_workflow`, then `gibson_assemble` with these arguments (substitute +actual workspace-relative fragment paths and the returned workflow ID): + +```json +{ + "workflow_id": "", + "fragments": [ + {"path": "inputs/linear_vector.gb", "format": "genbank"}, + {"path": "inputs/overlapped_insert.gb", "format": "genbank"} + ], + "overlaps": [20, 20], + "name": "assembled_product", + "circular": true, + "min_overlap": 15 +} +``` + +Overlap `i` is the suffix of fragment `i` matching the prefix of the next fragment. +Circular products need one overlap per fragment, including last-to-first closure; +linear products need one fewer. Explicit lengths resolve the intended junction +when repeated sequence offers several matches; the engine does not check for +alternative experimental assemblies. Each fragment must contribute sequence +outside its two overlaps. Inputs must contain only A/C/G/T and be linear. +The default minimum is a software acceptance threshold, not an efficiency guarantee. + +After assembly, call `validate_construct`, then `save_construct`. Assembly creates +a new artifact, resets prior validation, and logs the inputs and overlap settings. +SnapGene actions return `snapgene_unavailable` when no SnapGene adapter is configured. diff --git a/agent/adapters.py b/agent/adapters.py index 1629c33..607edb4 100644 --- a/agent/adapters.py +++ b/agent/adapters.py @@ -14,6 +14,11 @@ def replace_region( self, construct: ConstructRef, *, target: str, replacement_sequence: str, replacement_name: str | None ) -> ConstructRef: ... + def gibson_assemble( + self, fragments: list[ConstructRef], *, overlaps: list[int], name: str, + circular: bool = True, min_overlap: int = 15, + ) -> ConstructRef: ... + def add_annotation( self, construct: ConstructRef, *, name: str, feature_type: str, start: int, end: int, strand: int ) -> ConstructRef: ... diff --git a/agent/bootstrap.py b/agent/bootstrap.py index 5cd76cc..d04c90a 100644 --- a/agent/bootstrap.py +++ b/agent/bootstrap.py @@ -22,8 +22,18 @@ def _load_factory(setting: str) -> Any: return factory() +class UnavailableSnapGene: + """Permit biology-only workflows without a desktop installation.""" + + def __getattr__(self, name: str) -> Any: + def unavailable(*args: Any, **kwargs: Any) -> Any: + raise WorkflowError("snapgene_unavailable", "The SnapGene adapter has not been configured.") + return unavailable + + def manager_from_environment() -> WorkflowManager: """Build the production manager from factories owned by Persons 1 and 2.""" biology = _load_factory("DNA_MAKER_BIOLOGY_ADAPTER") - snapgene = _load_factory("DNA_MAKER_SNAPGENE_ADAPTER") + snapgene = (_load_factory("DNA_MAKER_SNAPGENE_ADAPTER") + if os.environ.get("DNA_MAKER_SNAPGENE_ADAPTER") else UnavailableSnapGene()) return WorkflowManager(biology, snapgene) diff --git a/agent/prompts/codex_instructions.md b/agent/prompts/codex_instructions.md index 3857878..c4a705f 100644 --- a/agent/prompts/codex_instructions.md +++ b/agent/prompts/codex_instructions.md @@ -20,3 +20,9 @@ specified. Do not ask for confirmation of a mutation. The biology engine owns biological correctness. The SnapGene adapter owns GUI and SnapGene-specific failures. Do not reinterpret or hide their errors. + +For Gibson simulation, use `gibson_assemble` with ordered, already oriented linear +fragment references and explicit exact overlap lengths. Circular products require +last-to-first overlap too. This MVP does not design primers or prepare fragments; +never describe ordinary region replacement as Gibson assembly. Validate the returned +product before saving. Biology-only workflows do not require SnapGene. diff --git a/agent/server.py b/agent/server.py index 9bfc681..f6b1491 100644 --- a/agent/server.py +++ b/agent/server.py @@ -81,6 +81,22 @@ def replace_region(workflow_id: str, target: str, replacement_sequence: str, rep return _run(lambda: manager.replace_region(workflow_id, target=target, replacement_sequence=replacement_sequence, replacement_name=replacement_name)) +@mcp.tool() +def gibson_assemble( + workflow_id: str, fragments: list[ConstructRef], overlaps: list[int], name: str, + circular: bool = True, min_overlap: int = 15, +) -> dict[str, Any]: + """Simulate Gibson from ordered, oriented linear fragments with exact overlaps. + + Supply n overlaps for n circular fragments, n-1 for linear. Last circular + overlap joins last to first. No primer design or orientation inference. + Returns a new GenBank product; call validate_construct before saving. + """ + return _run(lambda: manager.gibson_assemble( + workflow_id, fragments=fragments, overlaps=overlaps, name=name, + circular=circular, min_overlap=min_overlap)) + + @mcp.tool() def add_annotation(workflow_id: str, name: str, feature_type: str, start: int, end: int, strand: int = 1) -> dict[str, Any]: """Add a feature using 0-based, end-exclusive coordinates; validate afterwards.""" diff --git a/agent/session.py b/agent/session.py index 6e12563..a539a5e 100644 --- a/agent/session.py +++ b/agent/session.py @@ -48,6 +48,19 @@ def replace_region(self, workflow_id: str, *, target: str, replacement_sequence: result = self.biology.replace_region(session.current_construct, target=target, replacement_sequence=replacement_sequence, replacement_name=replacement_name) return self._mutation(session, "replace_region", {"target": target, "replacement_sequence": replacement_sequence, "replacement_name": replacement_name}, result) + def gibson_assemble( + self, workflow_id: str, *, fragments: list[ConstructRef], overlaps: list[int], + name: str, circular: bool = True, min_overlap: int = 15, + ) -> dict: + session = self.get(workflow_id) + result = self.biology.gibson_assemble(fragments, overlaps=overlaps, name=name, + circular=circular, min_overlap=min_overlap) + return self._mutation(session, "gibson_assemble", { + "fragments": [fragment.to_dict() for fragment in fragments], + "overlaps": overlaps, "name": name, "circular": circular, + "min_overlap": min_overlap, + }, result) + def add_annotation(self, workflow_id: str, *, name: str, feature_type: str, start: int, end: int, strand: int) -> dict: session = self._require_construct(workflow_id) result = self.biology.add_annotation(session.current_construct, name=name, feature_type=feature_type, start=start, end=end, strand=strand) diff --git a/docs/GIBSON_DEMO.md b/docs/GIBSON_DEMO.md new file mode 100644 index 0000000..3569dde --- /dev/null +++ b/docs/GIBSON_DEMO.md @@ -0,0 +1,48 @@ +# Gibson assembly presentation + +Run from the project root: + +```sh +uv run python -m dnamaker.gibson_demo +``` + +If the application launcher affects Python, use: + +```sh +env -u APPIMAGE .venv/bin/python -m dnamaker.gibson_demo +``` + +## What to say + +“DNAmaker now simulates Gibson assembly locally without SnapGene. We provide two +linear DNA fragments with matching end overlaps. The engine checks both junctions, +collapses the shared sequence, preserves annotations, and saves a circular GenBank +product. The workflow requires validation before saving and records the operations.” + +“The demonstration combines a 110-base synthetic vector fragment and a 90-base +synthetic insert. The two shared 20-base overlaps occur once in the final product, +so the result is 160 bases. The verification checks the expected sequence, length, +circular topology, preserved annotations, both junctions, unchanged source files, +and the validation gate.” + +## What is included + +- Real Python assembly engine, file-backed adapter, workflow operation, and MCP tool. +- Circular and linear products from two or more ordered fragments. +- Explicit exact overlap lengths; invalid joins fail with structured errors. +- GenBank output, junction annotations, provenance, and workflow log. +- Linux biology-only startup; Windows and SnapGene are unnecessary. + +## Scope of this demonstration + +Sequences are synthetic teaching fixtures, not a functional plasmid. Input fragments +must already be linear, oriented, and contain their overlaps. Primer design, +automatic fragment preparation/order/orientation, mismatch handling, alternative +assembly detection, experimental efficiency prediction, and SnapGene GUI/history +are outside this MVP. Structural validation does not establish experimental success. + +## Artifacts + +Every run creates a unique `artifacts/gibson-demo-*/` folder. Open `assembled.gb` +for the sequence and features, `verification.json` for pass/fail evidence, and +`workflow.json` for the operation history. diff --git a/findings.md b/findings.md new file mode 100644 index 0000000..8dfe6fa --- /dev/null +++ b/findings.md @@ -0,0 +1,6 @@ +The biology engine supports GenBank/FASTA/JSON and shared ConstructRef mutations. +Bootstrap currently requires both adapters; native desktop access is unavailable. +Explicit overlap lengths avoid silently choosing ambiguous repeat junctions. +Assembly retains both copies of source annotations if sources independently annotate a shared overlap. +Circular origin mapping preserves positive- and negative-strand feature sequences in GenBank round trips. +FastMCP converts structured fragment dictionaries to ConstructRef successfully. diff --git a/progress.md b/progress.md new file mode 100644 index 0000000..e292ad9 --- /dev/null +++ b/progress.md @@ -0,0 +1,7 @@ +Started Gibson MVP for a ten-minute presentation deadline. Working tree was clean. +Implemented assembly engine, adapter/workflow/MCP operation, and optional SnapGene startup. +Full test suite: 49 passed, 2 skipped (live SnapGene), one dependency forward-reference warning. +Demo passed seven independent verification checks, output in artifacts/gibson-demo-20260912-192118-eb75ef/. +Added README/contract/agent instructions and docs/GIBSON_DEMO.md presentation notes. +PR CI follow-up: reproduced two pre-existing repository-wide lint failures in tests/test_biology_adapter.py (I001) and tests/test_mcp_stdio.py (SIM117). Sorted imports and combined async context managers without changing behavior. Full CI lint command passes; pytest remains 49 passed, 2 skipped, one dependency warning. Earlier changed-files lint scope missed these baseline errors. +Merged origin/main at 88aa90a using the repository's merge-commit convention. Resolved README conflict by retaining main's corrected SnapGene prompts and this branch's Gibson guide. Preserved new manuals and real MCP SnapGene runner. Verification: locked dependency sync, full CI lint, 49 passing tests (2 live tests skipped, existing dependency warning), successful wheel/sdist build, passing Gibson demo, and new runner --help smoke check. diff --git a/src/dnamaker/assembly.py b/src/dnamaker/assembly.py new file mode 100644 index 0000000..1cb0328 --- /dev/null +++ b/src/dnamaker/assembly.py @@ -0,0 +1,161 @@ +"""Deterministic Gibson product simulation from ordered, pre-overlapped fragments. + +This predicts sequence from specified exact junctions; it does not design primers, +choose fragment order/orientation, or predict experimental assembly efficiency. +""" + +from __future__ import annotations + +import json + +from .errors import BiologyError +from .models import Construct, Feature, Location +from .validation import validate_construct_data + + +def assemble_gibson( + fragments: list[Construct], + *, + overlaps: list[int], + name: str, + circular: bool = True, + min_overlap: int = 15, +) -> Construct: + """Join each suffix to the next prefix; last overlap closes circular products. + + All inputs must be linear, unambiguous A/C/G/T, and already oriented. + Each fragment must contribute bases outside its incoming/outgoing overlaps. + Overlap lengths are explicit so repeated ends cannot silently select a join. + """ + if len(fragments) < 2 or not name.strip(): + raise BiologyError( + "invalid_assembly", "Provide at least two fragments and a product name." + ) + if type(min_overlap) is not int or min_overlap < 1: + raise BiologyError("invalid_overlap", "min_overlap must be a positive integer.") + expected = len(fragments) if circular else len(fragments) - 1 + if len(overlaps) != expected: + raise BiologyError( + "invalid_overlap", + f"Provide exactly {expected} overlap lengths in junction order.", + ) + for index, fragment in enumerate(fragments): + valid, checks = validate_construct_data(fragment) + if not valid or set(fragment.sequence) - set("ACGT"): + raise BiologyError( + "invalid_fragment", + "Fragments must have valid annotations and unambiguous A/C/G/T sequence.", + {"fragment": index, "checks": checks}, + ) + if fragment.topology != "linear": + raise BiologyError( + "fragment_not_linear", + "Linearize each fragment before assembly.", + {"fragment": index}, + ) + for index, length in enumerate(overlaps): + left, right = fragments[index], fragments[(index + 1) % len(fragments)] + if type(length) is not int or not min_overlap <= length < min( + len(left.sequence), len(right.sequence) + ): + raise BiologyError( + "invalid_overlap", + "Overlap must meet min_overlap and be shorter than both fragments.", + {"junction": index, "overlap": length}, + ) + if left.sequence[-length:] != right.sequence[:length]: + raise BiologyError( + "overlap_mismatch", + "Fragment suffix does not exactly match the next prefix.", + { + "junction": index, + "left": left.name, + "right": right.name, + "overlap": length, + }, + ) + for index, fragment in enumerate(fragments): + incoming = overlaps[index - 1] if index or circular else 0 + outgoing = overlaps[index] if index < len(overlaps) else 0 + if incoming + outgoing >= len(fragment.sequence): + raise BiologyError( + "overlapping_junctions", + "Each fragment must contain sequence outside its junction overlaps.", + {"fragment": index}, + ) + + sequence = fragments[0].sequence + offsets = [0] + for index, fragment in enumerate(fragments[1:]): + offsets.append(len(sequence) - overlaps[index]) + sequence += fragment.sequence[overlaps[index] :] + if circular: + sequence = sequence[: -overlaps[-1]] + size = len(sequence) + features = [] + for fragment, offset in zip(fragments, offsets): + for feature in fragment.features: + parts = [] + for part in feature.locations(): + start, end = part.start + offset, part.end + offset + if end <= size: + mapped = [Location(start=start, end=end)] + elif start >= size: + mapped = [Location(start=start - size, end=end - size)] + else: + mapped = [ + Location(start=start, end=size), + Location(start=0, end=end - size), + ] + if feature.strand == -1: + mapped.reverse() + parts.extend(mapped) + features.append( + feature.model_copy( + deep=True, + update={ + "start": min(p.start for p in parts), + "end": max(p.end for p in parts), + "parts": parts if len(parts) > 1 else None, + }, + ) + ) + # Retain source annotations and identify every checked junction in the product. + for index, length in enumerate(overlaps): + start = offsets[index + 1] if index + 1 < len(offsets) else 0 + features.append( + Feature( + name=f"Gibson junction {index + 1}", + type="misc_feature", + start=start, + end=start + length, + strand=1, + qualifiers={ + "note": [ + f"Exact {length} bp overlap: {fragments[index].name} -> {fragments[(index + 1) % len(fragments)].name}" + ] + }, + ) + ) + provenance = { + "method": "Gibson exact-overlap simulation", + "fragments": [f.name for f in fragments], + "overlaps": overlaps, + "circular": circular, + "min_overlap": min_overlap, + } + product = Construct( + name=name, + sequence=sequence, + topology="circular" if circular else "linear", + features=features, + metadata={"comment": json.dumps(provenance)}, + ) + valid, checks = validate_construct_data(product) + if not valid: + raise BiologyError( + "invalid_assembly_product", + "Assembled product failed structural validation.", + {"checks": checks}, + ) + return product diff --git a/src/dnamaker/gibson_demo.py b/src/dnamaker/gibson_demo.py new file mode 100644 index 0000000..76f4613 --- /dev/null +++ b/src/dnamaker/gibson_demo.py @@ -0,0 +1,132 @@ +"""Run a self-contained, synthetic Gibson presentation demo without SnapGene. + +uv run python -m dnamaker.gibson_demo +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +from agent.bootstrap import UnavailableSnapGene +from agent.errors import WorkflowError +from agent.models import ConstructRef +from agent.session import WorkflowManager + +from .io import load_construct, write_construct +from .models import Construct, Feature +from .service import BiologyServiceAdapter + + +def main() -> None: + workspace = Path.cwd() + run = ( + Path("artifacts") + / f"gibson-demo-{datetime.now(UTC):%Y%m%d-%H%M%S}-{uuid4().hex[:6]}" + ) + left = "ACGTCAGTACGATCGTAGCA" + right = "TGCACTGATCGATGCTACGT" + backbone = "GATTACA" * 10 + insert = "CCGTA" * 10 + fragments = [ + Construct( + name="synthetic_vector", + sequence=left + backbone + right, + features=[ + Feature( + name="Synthetic backbone", type="misc_feature", start=20, end=90 + ) + ], + ), + Construct( + name="synthetic_insert", + sequence=right + insert + left, + features=[ + Feature(name="Synthetic insert", type="misc_feature", start=20, end=70) + ], + ), + ] + refs = [] + for fragment in fragments: + path = run / f"{fragment.name}.gb" + write_construct(fragment, workspace / path, "genbank") + refs.append(ConstructRef(path.as_posix())) + originals = [(workspace / ref.path).read_bytes() for ref in refs] + manager = WorkflowManager( + BiologyServiceAdapter(workspace, run / "intermediate"), UnavailableSnapGene() + ) + workflow = manager.start( + "Assemble two synthetic overlapping fragments into a circular product." + ) + manager.gibson_assemble( + workflow.id, fragments=refs, overlaps=[20, 20], name="Gibson_demo" + ) + gate_checked = False + try: + manager.save_construct( + workflow.id, + output_path=(run / "premature.gb").as_posix(), + output_format="genbank", + ) + except WorkflowError as error: + if error.code != "validation_required": + raise + gate_checked = True + validation = manager.validate_construct(workflow.id) + if not validation["valid"]: + raise RuntimeError("Product failed validation") + saved = manager.save_construct( + workflow.id, + output_path=(run / "assembled.gb").as_posix(), + output_format="genbank", + ) + product = load_construct(workspace / saved["construct"]["path"], "genbank") + checks = { + "expected_sequence": product.sequence == left + backbone + right + insert, + "expected_length": len(product.sequence) == 160, + "circular": product.topology == "circular", + "source_files_unchanged": originals + == [(workspace / ref.path).read_bytes() for ref in refs], + "validation_gate": gate_checked, + "annotations_preserved": {f.name for f in product.features} + >= {"Synthetic backbone", "Synthetic insert"}, + "two_junctions_annotated": sum( + f.name.startswith("Gibson junction") for f in product.features + ) + == 2, + } + if not all(checks.values()): + raise RuntimeError(f"Independent demo verification failed: {checks}") + report = { + "demo": "Synthetic Gibson exact-overlap simulation", + "fragments_bp": [110, 90], + "overlaps_bp": [20, 20], + "product_bp": len(product.sequence), + "topology": product.topology, + "checks": checks, + "product": saved["construct"], + "limitations": "Pre-oriented fragments; explicit exact overlaps; no primer design or wet-lab efficiency prediction.", + } + (workspace / run / "verification.json").write_text( + json.dumps(report, indent=2) + "\n" + ) + (workspace / run / "workflow.json").write_text( + json.dumps(workflow.to_dict(), indent=2) + "\n" + ) + print("GIBSON ASSEMBLY — local Python demo (no SnapGene)") + print("110 bp vector + 90 bp insert - two 20 bp overlaps = 160 bp circular product") + print( + "PASS: sequence, length, topology, annotations, junctions, unchanged inputs, validation gate" + ) + print(f"Product: {run / 'assembled.gb'}") + print(f"Verification: {run / 'verification.json'}") + print(f"Workflow log: {run / 'workflow.json'}") + print( + "Synthetic sequence demo; primer design and experimental efficiency are outside this MVP." + ) + + +if __name__ == "__main__": + main() diff --git a/src/dnamaker/service.py b/src/dnamaker/service.py index e95aeed..9a44e78 100644 --- a/src/dnamaker/service.py +++ b/src/dnamaker/service.py @@ -174,6 +174,17 @@ def replace_region( ) return self._write_mutation(modified, "replace") + def gibson_assemble( + self, fragments: list[ConstructRef], *, overlaps: list[int], name: str, + circular: bool = True, min_overlap: int = 15, + ) -> ConstructRef: + from .assembly import assemble_gibson + + loaded = [self._load(reference)[0] for reference in fragments] + product = assemble_gibson(loaded, overlaps=overlaps, name=name, + circular=circular, min_overlap=min_overlap) + return self._write_mutation(product, "gibson") + def add_annotation( self, construct: ConstructRef, diff --git a/task_plan.md b/task_plan.md new file mode 100644 index 0000000..5e29b4c --- /dev/null +++ b/task_plan.md @@ -0,0 +1,13 @@ +# Gibson MVP +1. Implement ordered exact-overlap assembly and annotation mapping — complete. +2. Expose adapter/workflow/MCP and allow biology-only startup — complete. +3. Test junctions, circular features, validation gates; run presentation demo — complete. +Scope: pre-oriented linear fragments, explicit overlap lengths, no primer design. + +## Errors encountered +- AppImage launcher redirected Python executable discovery; run with APPIMAGE unset or via uv. +- Lint found unused test imports/import ordering; fixed with ruff. +4. Diagnose and fix PR CI lint failures — complete; remote CI passed. +- Full-repository lint exposed two existing test style violations; fixed and rechecked with the exact CI command. + +5. Merge latest main, preserve both documentation changes, build and verify — complete. diff --git a/tests/test_biology_adapter.py b/tests/test_biology_adapter.py index 6025d3f..f4b2c53 100644 --- a/tests/test_biology_adapter.py +++ b/tests/test_biology_adapter.py @@ -3,8 +3,8 @@ import shutil from pathlib import Path -from Bio import SeqIO import pytest +from Bio import SeqIO from agent.models import ConstructRef from agent.session import WorkflowManager diff --git a/tests/test_gibson.py b/tests/test_gibson.py new file mode 100644 index 0000000..9a6606d --- /dev/null +++ b/tests/test_gibson.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import os +from unittest.mock import patch + +import pytest +from Bio.Seq import Seq + +from agent.bootstrap import UnavailableSnapGene, manager_from_environment +from agent.errors import WorkflowError +from agent.models import ConstructRef +from agent.session import WorkflowManager +from dnamaker.assembly import assemble_gibson +from dnamaker.errors import BiologyError +from dnamaker.io import load_construct, write_construct +from dnamaker.models import Construct, Feature +from dnamaker.service import BiologyServiceAdapter + +LEFT = "ACGTCAGTACGATCGTAGCA" +RIGHT = "TGCACTGATCGATGCTACGT" +CORE = "GATTACA" * 10 +INSERT = "CCGTA" * 10 + + +def fragments(): + return [ + Construct(name="vector", sequence=LEFT + CORE + RIGHT), + Construct(name="insert", sequence=RIGHT + INSERT + LEFT), + ] + + +def test_circular_product_and_junctions(): + product = assemble_gibson(fragments(), overlaps=[20, 20], name="product") + assert product.sequence == LEFT + CORE + RIGHT + INSERT + assert product.topology == "circular" + assert [(f.start, f.end) for f in product.features] == [(90, 110), (0, 20)] + + +def test_linear_three_fragments(): + a = Construct(name="a", sequence=CORE + LEFT) + b = Construct(name="b", sequence=LEFT + INSERT + RIGHT) + c = Construct(name="c", sequence=RIGHT + CORE) + product = assemble_gibson( + [a, b, c], overlaps=[20, 20], name="linear", circular=False + ) + assert product.sequence == CORE + LEFT + INSERT + RIGHT + CORE + assert product.topology == "linear" + + +@pytest.mark.parametrize("strand", [1, -1]) +def test_origin_crossing_features_preserve_sequence_after_genbank_roundtrip( + tmp_path, strand +): + inputs = fragments() + # This insert feature crosses the product origin after the final overlap is collapsed. + inputs[1].features = [ + Feature(name="crossing", type="misc_feature", start=65, end=85, strand=strand) + ] + product = assemble_gibson(inputs, overlaps=[20, 20], name="product") + path = tmp_path / "product.gb" + write_construct(product, path, "genbank") + loaded = load_construct(path, "genbank") + feature = loaded.features[0] + assert len(feature.parts) == 2 + extracted = "".join( + str(Seq(loaded.sequence[p.start : p.end]).reverse_complement()) + if strand == -1 + else loaded.sequence[p.start : p.end] + for p in feature.locations() + ) + expected = inputs[1].sequence[65:85] + assert extracted == ( + str(Seq(expected).reverse_complement()) if strand == -1 else expected + ) + assert "Gibson exact-overlap" in loaded.metadata["comment"] + + +@pytest.mark.parametrize( + "overlaps,code", + [ + ([19, 20], "overlap_mismatch"), + ([20], "invalid_overlap"), + ([True, 20], "invalid_overlap"), + ([0, 20], "invalid_overlap"), + ], +) +def test_bad_junctions(overlaps, code): + with pytest.raises(BiologyError) as error: + assemble_gibson(fragments(), overlaps=overlaps, name="bad") + assert error.value.code == code + + +def test_closing_junction_is_checked(): + inputs = fragments() + inputs[1].sequence = inputs[1].sequence[:-1] + "T" + with pytest.raises(BiologyError) as error: + assemble_gibson(inputs, overlaps=[20, 20], name="bad") + assert error.value.code == "overlap_mismatch" + assert error.value.details["junction"] == 1 + + +@pytest.mark.parametrize( + "change,code", [("circular", "fragment_not_linear"), ("N", "invalid_fragment")] +) +def test_invalid_inputs(change, code): + inputs = fragments() + if change == "circular": + inputs[0].topology = change + else: + inputs[0].sequence = "N" + inputs[0].sequence[1:] + with pytest.raises(BiologyError) as error: + assemble_gibson(inputs, overlaps=[20, 20], name="bad") + assert error.value.code == code + + +def test_overlapping_junctions_rejected(): + inputs = [ + Construct(name="a", sequence="A" * 30), + Construct(name="b", sequence="A" * 30), + ] + with pytest.raises(BiologyError) as error: + assemble_gibson(inputs, overlaps=[20, 20], name="bad") + assert error.value.code == "overlapping_junctions" + + +def test_real_workflow_and_biology_only_bootstrap(tmp_path): + refs = [] + for index, fragment in enumerate(fragments()): + path = tmp_path / f"fragment-{index}.gb" + write_construct(fragment, path, "genbank") + refs.append(ConstructRef(path.name)) + with patch.dict( + os.environ, + { + "DNA_MAKER_BIOLOGY_ADAPTER": "dnamaker.service:create_biology_adapter", + "DNA_MAKER_WORKSPACE": str(tmp_path), + }, + clear=True, + ): + manager = manager_from_environment() + workflow = manager.start("Gibson demo") + manager.read_construct(workflow.id, refs[0]) + manager.validate_construct(workflow.id) + result = manager.gibson_assemble( + workflow.id, fragments=refs, overlaps=[20, 20], name="demo" + ) + assert result["validation_required"] + with pytest.raises(WorkflowError, match="Validate"): + manager.save_construct( + workflow.id, output_path="result.gb", output_format="genbank" + ) + assert manager.validate_construct(workflow.id)["valid"] + manager.save_construct( + workflow.id, output_path="result.gb", output_format="genbank" + ) + product = load_construct(tmp_path / "result.gb", "genbank") + assert product.sequence == LEFT + CORE + RIGHT + INSERT + assert workflow.operations[2].name == "gibson_assemble" + with pytest.raises(WorkflowError) as error: + manager.snapgene_convert(workflow.id, output_path="result.dna") + assert error.value.code == "snapgene_unavailable" + + +def test_mcp_tool_accepts_structured_fragment_refs(tmp_path): + import asyncio + + from agent import server + + refs = [] + for index, fragment in enumerate(fragments()): + path = tmp_path / f"fragment-{index}.gb" + write_construct(fragment, path, "genbank") + refs.append({"path": path.name, "format": "genbank"}) + manager = WorkflowManager(BiologyServiceAdapter(tmp_path), UnavailableSnapGene()) + workflow = manager.start("MCP Gibson") + with patch.object(server, "manager", manager): + # Exercise FastMCP's JSON-to-dataclass conversion, not just the Python function. + tool = server.mcp._tool_manager.get_tool("gibson_assemble") + result = asyncio.run( + tool.run( + { + "workflow_id": workflow.id, + "fragments": refs, + "overlaps": [20, 20], + "name": "mcp_product", + } + ) + ) + assert result["ok"] + assert result["result"]["validation_required"] diff --git a/tests/test_mcp_stdio.py b/tests/test_mcp_stdio.py index 2452776..e626f79 100644 --- a/tests/test_mcp_stdio.py +++ b/tests/test_mcp_stdio.py @@ -33,77 +33,79 @@ async def run() -> None: args=["-m", "agent.server"], env=environment, ) - async with stdio_client(parameters) as (read, write): - async with ClientSession(read, write) as session: - await session.initialize() - tools = await session.list_tools() - names = {tool.name for tool in tools.tools} - assert {"read_construct", "replace_region", "validate_construct", "snapgene_convert", "snapgene_render", "snapgene_open"} <= names + async with ( + stdio_client(parameters) as (read, write), + ClientSession(read, write) as session, + ): + await session.initialize() + tools = await session.list_tools() + names = {tool.name for tool in tools.tools} + assert {"read_construct", "replace_region", "validate_construct", "snapgene_convert", "snapgene_render", "snapgene_open"} <= names - started = await session.call_tool( - "start_workflow", {"request": "Replace target and open the result."} - ) - workflow_id = started.structuredContent["result"]["workflow_id"] + started = await session.call_tool( + "start_workflow", {"request": "Replace target and open the result."} + ) + workflow_id = started.structuredContent["result"]["workflow_id"] - read_result = await session.call_tool( - "read_construct", - {"workflow_id": workflow_id, "path": "inputs/pEGFP-N1.gb"}, - ) - assert read_result.structuredContent["ok"] is True + read_result = await session.call_tool( + "read_construct", + {"workflow_id": workflow_id, "path": "inputs/pEGFP-N1.gb"}, + ) + assert read_result.structuredContent["ok"] is True - failed_edit = await session.call_tool( - "replace_region", - { - "workflow_id": workflow_id, - "target": "missing", - "replacement_sequence": "CCCC", - }, - ) - assert failed_edit.structuredContent == { - "ok": False, - "error": { - "code": "target_not_found", - "message": "No feature named 'missing' was found.", - "details": {"target": "missing"}, - }, - } + failed_edit = await session.call_tool( + "replace_region", + { + "workflow_id": workflow_id, + "target": "missing", + "replacement_sequence": "CCCC", + }, + ) + assert failed_edit.structuredContent == { + "ok": False, + "error": { + "code": "target_not_found", + "message": "No feature named 'missing' was found.", + "details": {"target": "missing"}, + }, + } - await session.call_tool( - "replace_region", - { - "workflow_id": workflow_id, - "target": "EGFP", - "replacement_sequence": str(mCherry.seq), - "replacement_name": "mCherry", - }, - ) - validation = await session.call_tool( - "validate_construct", {"workflow_id": workflow_id} - ) - assert validation.structuredContent["result"]["valid"] is True - converted = await session.call_tool( - "snapgene_convert", - {"workflow_id": workflow_id, "output_path": "outputs/result.dna"}, - ) - converted_ref = converted.structuredContent["result"]["construct"] - assert converted_ref["format"] == "snapgene" - rendered = await session.call_tool( - "snapgene_render", - {"workflow_id": workflow_id, "output_path": "outputs/result.png"}, - ) - assert rendered.structuredContent["result"]["map_path"] == "outputs/result.png" - opened = await session.call_tool("snapgene_open", {"workflow_id": workflow_id}) - assert opened.structuredContent["result"]["opened"] == converted_ref + await session.call_tool( + "replace_region", + { + "workflow_id": workflow_id, + "target": "EGFP", + "replacement_sequence": str(mCherry.seq), + "replacement_name": "mCherry", + }, + ) + validation = await session.call_tool( + "validate_construct", {"workflow_id": workflow_id} + ) + assert validation.structuredContent["result"]["valid"] is True + converted = await session.call_tool( + "snapgene_convert", + {"workflow_id": workflow_id, "output_path": "outputs/result.dna"}, + ) + converted_ref = converted.structuredContent["result"]["construct"] + assert converted_ref["format"] == "snapgene" + rendered = await session.call_tool( + "snapgene_render", + {"workflow_id": workflow_id, "output_path": "outputs/result.png"}, + ) + assert rendered.structuredContent["result"]["map_path"] == "outputs/result.png" + opened = await session.call_tool("snapgene_open", {"workflow_id": workflow_id}) + assert opened.structuredContent["result"]["opened"] == converted_ref - status = await session.call_tool("workflow_status", {"workflow_id": workflow_id}) - operations = status.structuredContent["result"]["operations"] - assert [operation["name"] for operation in operations] == [ - "read_construct", - "replace_region", - "validate_construct", - "snapgene_convert", - "snapgene_render", - "snapgene_open", - ] + status = await session.call_tool("workflow_status", {"workflow_id": workflow_id}) + operations = status.structuredContent["result"]["operations"] + assert [operation["name"] for operation in operations] == [ + "read_construct", + "replace_region", + "validate_construct", + "snapgene_convert", + "snapgene_render", + "snapgene_open", + ] asyncio.run(run())