diff --git a/src/lab/compiler.py b/src/lab/compiler.py
index 2c206b3..62fdb9d 100644
--- a/src/lab/compiler.py
+++ b/src/lab/compiler.py
@@ -11,6 +11,7 @@
 
 import lab.documents as documents
 from lab._version import __version__
+from lab.artifacts import write_bundle
 from lab.deck import Deck
 from lab.experiments.cloning.stages.assembly import build_assembly
 from lab.experiments.cloning.stages.plating import build_plating
@@ -22,13 +23,15 @@
     Transformation,
     TransformationRequest,
 )
-from lab.model import Distribute, Mix, RecordedProtocol, TargetPlan, Transfer, encode
+from lab.model import RecordedProtocol, encode
+from lab.operations import Distribute, Mix, Transfer
 from lab.protocol import Protocol
 from lab.samples import Location, OutputManifest
+from lab.target import TargetPlan
 from lab.targets.liquid_handler import LiquidHandler
 from lab.targets.lower import lower_deck
 from lab.targets.manual import Manual
-from lab.validation import CompileError, logical_bindings, validate
+from lab.validation import CompileError, count_trace, logical_bindings, validate
 
 
 class Target(Interface):
@@ -69,6 +72,10 @@ def plan_json(self) -> str:
                     {"location": encode(location), "volume": encode(volume)}
                     for location, volume in self.final_volumes
                 ],
+                "final_counts": [
+                    {"location": encode(location), "count": count}
+                    for location, count in count_trace(self.protocol)[-1].items()
+                ],
                 "source_sha256": (
                     hashlib.sha256(self.target.source.encode()).hexdigest()
                     if self.target.source is not None
@@ -81,9 +88,44 @@ def plan_json(self) -> str:
     def digest(self) -> str:
         return hashlib.sha256(self.plan_json.encode()).hexdigest()
 
+    @property
+    def source_map(self) -> tuple[dict[str, object], ...]:
+        """One-based inclusive generated line spans for semantic operations."""
+        source = self.target.source
+        if source is None:
+            return ()
+        lines = source.splitlines()
+        markers = [
+            (index + 1, line.strip().removeprefix("# lab:step "))
+            for index, line in enumerate(lines)
+            if line.strip().startswith("# lab:step ")
+        ]
+        ends = {
+            line.strip().removeprefix("# lab:end "): index + 1
+            for index, line in enumerate(lines)
+            if line.strip().startswith("# lab:end ")
+        }
+        expected = tuple(step.identity for step in self.protocol.steps)
+        if tuple(identity for _, identity in markers) != expected or set(ends) != set(expected):
+            raise CompileError("Generated source does not map every protocol step exactly once")
+        return tuple(
+            {
+                "step": identity,
+                "file": "protocol.py",
+                "start_line": start,
+                "end_line": ends[identity],
+            }
+            for start, identity in markers
+        )
+
     @property
     def files(self) -> dict[str, str]:
-        result = {"plan.json": self.plan_json, "protocol.html": documents.render(self)}
+        result = {
+            "plan.json": self.plan_json,
+            "protocol.html": documents.render(self),
+            "protocol.json": self.protocol.semantic_json,
+            "source-map.json": canonical_json(self.source_map),
+        }
         if self.protocol.output_sample_ids:
             result["manifest.json"] = canonical_json(self.manifest.to_dict())
         if self.target.source is not None:
@@ -92,16 +134,60 @@ def files(self) -> dict[str, str]:
 
     def write(self, directory: str | Path) -> Path:
         """Write a bundle. Refuse to replace any different existing artifact."""
-        directory = Path(directory)
-        files = self.files
-        for name, text in files.items():
-            path = directory / name
-            if path.exists() and path.read_text(encoding="utf-8") != text:
-                raise FileExistsError(f"{path} already contains a different artifact")
-        directory.mkdir(parents=True, exist_ok=True)
-        for name, text in files.items():
-            (directory / name).write_text(text, encoding="utf-8")
-        return directory
+        return write_bundle(directory, self.files)
+
+
+def _compile_protocol(
+    protocol: Protocol | RecordedProtocol,
+    hardware: Target | Deck,
+    *,
+    liquid_handler: LiquidHandler | None = None,
+) -> Compilation:
+    """Compile offline for one piece of hardware.
+
+    A ``Deck`` contains shared requirements and optional Lab-owned layouts. The
+    selected backend validates and translates its layout or supported preset.
+    Concrete backend targets are also accepted for low-level integrations.
+    A document target such as ``Manual()`` has no robot, so ``liquid_handler`` is omitted.
+    """
+    recorded = protocol.snapshot() if isinstance(protocol, Protocol) else protocol
+    authored_deck = hardware if isinstance(hardware, Deck) else None
+    if isinstance(hardware, Deck):
+        if not isinstance(liquid_handler, LiquidHandler):
+            raise TypeError(
+                "Pass liquid_handler=LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR."
+            )
+        liquid = tuple(
+            step.volume for step in recorded.steps if isinstance(step, (Transfer, Mix, Distribute))
+        )
+        requirements = {container.id: container.labware for container in hardware.containers}
+        for resource in recorded.resources:
+            spec = requirements.get(resource.name)
+            if spec is None or (spec.rows, spec.columns) != (resource.rows, resource.columns):
+                raise CompileError(
+                    f"Deck requirements must match the protocol geometry for {resource.name}."
+                )
+            if resource.capacity > spec.capacity_ul:
+                raise CompileError(f"Protocol capacity exceeds the deck limit for {resource.name}.")
+        hardware = lower_deck(hardware, liquid_handler, liquid)
+    declared = getattr(hardware, "liquid_handler", None)
+    if isinstance(declared, LiquidHandler) and liquid_handler != declared:
+        raise TypeError(
+            f"This hardware is LiquidHandler.{declared.name}. "
+            f"Pass liquid_handler=LiquidHandler.{declared.name}."
+        )
+    if liquid_handler is not None and not isinstance(declared, LiquidHandler):
+        raise TypeError("This hardware does not name a LiquidHandler.")
+    if liquid_handler is not None and not isinstance(liquid_handler, LiquidHandler):
+        raise TypeError("Pass LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR.")
+    validate(recorded, logical_bindings(recorded))
+    prepared = hardware.prepare(recorded)
+    if authored_deck is not None:
+        configuration = json.loads(prepared.configuration_json)
+        configuration["lab_deck"] = encode(authored_deck)
+        prepared = replace(prepared, configuration_json=json.dumps(configuration, sort_keys=True))
+    volumes = validate(recorded, prepared.bindings)
+    return Compilation(recorded, prepared, tuple(volumes.items()))
 
 
 class _DefaultOutput:
@@ -109,7 +195,14 @@ class _DefaultOutput:
 
 
 _DEFAULT_OUTPUT = _DefaultOutput()
-_BUNDLE_FILES = ("protocol.html", "plan.json", "manifest.json", "protocol.py")
+_BUNDLE_FILES = (
+    "protocol.html",
+    "plan.json",
+    "manifest.json",
+    "protocol.py",
+    "protocol.json",
+    "source-map.json",
+)
 
 
 def _segment(value: str) -> str:
@@ -145,7 +238,7 @@ def _write_output(directory: Path, files: dict[str, str]) -> None:
 
 @overload
 def compile(
-    protocol: Protocol | Assembly | AssemblyRequest,
+    protocol: Protocol | RecordedProtocol | Assembly | AssemblyRequest,
     target: Target | None = None,
     *,
     deck: Deck | None = None,
@@ -182,6 +275,7 @@ def compile(
 def compile(
     protocol: (
         Protocol
+        | RecordedProtocol
         | Assembly
         | AssemblyRequest
         | Transformation
@@ -208,20 +302,18 @@ def compile(
     backend validates and translates its layout or supported preset. Concrete
     backend targets are also accepted for low-level integrations.
     """
+    if inputs is not None and not isinstance(
+        protocol, (Transformation, TransformationRequest, PlatingRequest)
+    ):
+        raise TypeError("Pass inputs with a transformation or plating request.")
     if isinstance(target, Deck):
         raise TypeError("Pass a deck with deck=.")
     if deck is not None and target is not None:
         raise TypeError("Pass a target or a deck.")
     if deck is None and target is None and liquid_handler is not None:
         raise TypeError("Pass deck= with liquid_handler.")
-    hardware: Target | Deck = (
-        deck if deck is not None else target if target is not None else Manual()
-    )
-    if inputs is not None and not isinstance(
-        protocol, (Transformation, TransformationRequest, PlatingRequest)
-    ):
-        raise TypeError("Pass inputs with a transformation or plating request.")
-    work: Protocol
+    hardware = deck if deck is not None else target if target is not None else Manual()
+    work: Protocol | RecordedProtocol
     if isinstance(protocol, Assembly):
         work = build_assembly(AssemblyRequest(id=protocol.id, assemblies=(protocol,)))
     elif isinstance(protocol, AssemblyRequest):
@@ -237,44 +329,7 @@ def compile(
         work = build_plating(protocol, inputs=inputs)
     else:
         work = protocol
-    recorded = work.snapshot()
-    authored_deck = hardware if isinstance(hardware, Deck) else None
-    if isinstance(hardware, Deck):
-        if not isinstance(liquid_handler, LiquidHandler):
-            raise TypeError(
-                "Pass liquid_handler=LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR."
-            )
-        liquid = tuple(
-            step.volume for step in recorded.steps if isinstance(step, (Transfer, Mix, Distribute))
-        )
-        requirements = {container.id: container.labware for container in hardware.containers}
-        for resource in recorded.resources:
-            spec = requirements.get(resource.name)
-            if spec is None or (spec.rows, spec.columns) != (resource.rows, resource.columns):
-                raise CompileError(
-                    f"Deck requirements must match the protocol geometry for {resource.name}."
-                )
-            if resource.capacity > spec.capacity_ul:
-                raise CompileError(f"Protocol capacity exceeds the deck limit for {resource.name}.")
-        hardware = lower_deck(hardware, liquid_handler, liquid)
-    declared = getattr(hardware, "liquid_handler", None)
-    if isinstance(declared, LiquidHandler) and liquid_handler != declared:
-        raise TypeError(
-            f"This hardware is LiquidHandler.{declared.name}. "
-            f"Pass liquid_handler=LiquidHandler.{declared.name}."
-        )
-    if liquid_handler is not None and not isinstance(declared, LiquidHandler):
-        raise TypeError("This hardware does not name a LiquidHandler.")
-    if liquid_handler is not None and not isinstance(liquid_handler, LiquidHandler):
-        raise TypeError("Pass LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR.")
-    validate(recorded, logical_bindings(recorded))
-    prepared = hardware.prepare(recorded)
-    if authored_deck is not None:
-        configuration = json.loads(prepared.configuration_json)
-        configuration["lab_deck"] = encode(authored_deck)
-        prepared = replace(prepared, configuration_json=json.dumps(configuration, sort_keys=True))
-    volumes = validate(recorded, prepared.bindings)
-    compilation = Compilation(recorded, prepared, tuple(volumes.items()))
+    compilation = _compile_protocol(work, hardware, liquid_handler=liquid_handler)
     directory = _output_directory(compilation, to)
     if directory is not None:
         _write_output(directory, compilation.files)
diff --git a/src/lab/documents.py b/src/lab/documents.py
index 8ef51ea..885927d 100644
--- a/src/lab/documents.py
+++ b/src/lab/documents.py
@@ -3,8 +3,9 @@
 from html import escape
 from typing import Any
 
-from lab.model import (
+from lab.operations import (
     Distribute,
+    ExternalPreparation,
     ManualInstruction,
     Mix,
     SetTemperature,
@@ -18,8 +19,9 @@
 
 def describe(step: Step) -> str:
     match step:
-        case Transfer(source, destination, volume, _):
-            return f"Transfer {number(volume)} µL from {source} to {destination}."
+        case Transfer(source, destination, volume, _, height):
+            placement = "" if height is None else f" at {number(height)} mm above the well bottom"
+            return f"Transfer {number(volume)} µL from {source} to {destination}{placement}."
         case Distribute(source, destinations, volume, air_gap, _):
             wells = ", ".join(str(destination) for destination in destinations)
             gap = f" Air gap {number(air_gap)} µL." if air_gap is not None else ""
@@ -48,6 +50,22 @@ def describe(step: Step) -> str:
             return f"Hold {resource} at {number(celsius)} °C."
         case ManualInstruction(text, _):
             return f"Operator: {text}"
+        case ExternalPreparation(procedure, instructions, inputs, outputs, _):
+
+            def describe_ports(ports: tuple) -> str:
+                return "; ".join(
+                    f"{port.count} unit(s) at {port.location}"
+                    if port.count
+                    else f"{number(port.volume_ul)} µL at {port.location}"
+                    for port in ports
+                )
+
+            return (
+                f"External procedure <{procedure}>: {instructions} "
+                f"Consume {describe_ports(inputs)}. Expected output: {describe_ports(outputs)}. "
+                "Record completion and measured output before treating material "
+                "as available inventory."
+            )
         case _:
             raise TypeError(f"Unsupported step: {type(step).__name__}")
 
@@ -55,11 +73,23 @@ def describe(step: Step) -> str:
 def render(compilation: Any) -> str:
     p, target = compilation.protocol, compilation.target
     bindings = {binding.location: binding for binding in target.bindings}
+    samples = {sample.id: sample for sample in p.samples}
     resources = []
     for resource in p.resources:
         fills = (
             "; ".join(
-                f"{fill.well}: {fill.material}, {number(fill.volume)} µL" for fill in resource.fills
+                [
+                    f"{fill.well}: {fill.material}, {number(fill.volume)} µL"
+                    for fill in resource.fills
+                ]
+                + [
+                    f"{place.location.well}: {samples[place.sample_id].label}, "
+                    f"{samples[place.sample_id].count} unit(s)"
+                    for place in p.placements
+                    if place.location.resource == resource.name
+                    and place.sample_id in p.input_sample_ids
+                    and samples[place.sample_id].count is not None
+                ]
             )
             or "Initially empty"
         )
diff --git a/src/lab/equipment.py b/src/lab/equipment.py
index 5cc3a66..d74ce2a 100644
--- a/src/lab/equipment.py
+++ b/src/lab/equipment.py
@@ -37,6 +37,7 @@ class TipRackModel(Enum):
     OPENTRONS_20_UL = "opentrons_20_ul"
     OPENTRONS_300_UL = "opentrons_300_ul"
     FLEX_200_UL = "flex_200_ul"
+    FLEX_50_UL = "flex_50_ul"
     HAMILTON_50_UL = "hamilton_50_ul"
     HAMILTON_300_UL = "hamilton_300_ul"
 
@@ -45,6 +46,7 @@ class PipetteModel(Enum):
     P20_SINGLE_GEN2 = "p20_single_gen2"
     P300_SINGLE_GEN2 = "p300_single_gen2"
     FLEX_1CHANNEL_1000 = "flex_1channel_1000"
+    FLEX_1CHANNEL_50 = "flex_1channel_50"
 
 
 class Mount(Enum):
diff --git a/src/lab/model.py b/src/lab/model.py
index 86fcb0a..222f064 100644
--- a/src/lab/model.py
+++ b/src/lab/model.py
@@ -1,112 +1,28 @@
 """The entire recorded protocol vocabulary. No callbacks or device objects."""
 
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, replace
 from decimal import Decimal
 from enum import Enum
 from typing import Any
 
-from lab.samples import Location, OutputManifest, Sample, SamplePlacement
+from lab.artifacts import canonical_json, digest
+from lab.operations import (
+    Distribute,
+    ExternalPreparation,
+    ManualInstruction,
+    Mix,
+    Resource,
+    SetTemperature,
+    Step,
+    Thermocycle,
+    Transfer,
+    Wait,
+)
+from lab.provenance.types import require_iri
+from lab.samples import OutputManifest, Sample, SamplePlacement
 from lab.units import number
 
 
-@dataclass(frozen=True)
-class Origin:
-    file: str
-    line: int
-
-
-@dataclass(frozen=True)
-class Fill:
-    well: str
-    material: str
-    volume: Decimal
-
-
-@dataclass(frozen=True)
-class Resource:
-    name: str
-    rows: int
-    columns: int
-    capacity: Decimal
-    dead_volume: Decimal
-    fills: tuple[Fill, ...] = ()
-
-    @property
-    def wells(self) -> tuple[str, ...]:
-        return tuple(
-            f"{chr(65 + row)}{column + 1}"
-            for row in range(self.rows)
-            for column in range(self.columns)
-        )
-
-
-@dataclass(frozen=True)
-class Transfer:
-    source: Location
-    destination: Location
-    volume: Decimal
-    origin: Origin
-
-
-@dataclass(frozen=True)
-class Distribute:
-    """One aspiration shared across destinations. ``air_gap`` is air, not liquid volume."""
-
-    source: Location
-    destinations: tuple[Location, ...]
-    volume: Decimal
-    air_gap: Decimal | None
-    origin: Origin
-
-
-@dataclass(frozen=True)
-class Mix:
-    location: Location
-    volume: Decimal
-    cycles: int
-    origin: Origin
-
-
-@dataclass(frozen=True)
-class Wait:
-    seconds: Decimal
-    origin: Origin
-
-
-@dataclass(frozen=True)
-class Hold:
-    celsius: Decimal
-    seconds: Decimal
-
-
-@dataclass(frozen=True)
-class Thermocycle:
-    resource: str
-    profile: tuple[Hold, ...]
-    cycles: int
-    lid_celsius: Decimal | None
-    origin: Origin
-    block_volume: Decimal | None = None
-
-
-@dataclass(frozen=True)
-class SetTemperature:
-    """Hold a plate's controlling module at one temperature. Liquid handling may continue."""
-
-    resource: str
-    celsius: Decimal
-    origin: Origin
-
-
-@dataclass(frozen=True)
-class ManualInstruction:
-    text: str
-    origin: Origin
-
-
-Step = Transfer | Distribute | Mix | Wait | Thermocycle | SetTemperature | ManualInstruction
-
-
 @dataclass(frozen=True)
 class RecordedProtocol:
     name: str
@@ -117,37 +33,63 @@ class RecordedProtocol:
     placements: tuple[SamplePlacement, ...] = ()
     input_sample_ids: tuple[str, ...] = ()
     output_sample_ids: tuple[str, ...] = ()
+    identity: str | None = None
+
+    def __post_init__(self) -> None:
+        for field in (
+            self.resources,
+            self.steps,
+            self.samples,
+            self.placements,
+            self.input_sample_ids,
+            self.output_sample_ids,
+        ):
+            if not isinstance(field, tuple):
+                raise TypeError("Recorded protocol collections must be immutable tuples")
+        if any(not isinstance(resource.fills, tuple) for resource in self.resources):
+            raise TypeError("Recorded initial fills must be immutable tuples")
+        if any(
+            isinstance(step, Thermocycle) and not isinstance(step.profile, tuple)
+            for step in self.steps
+        ):
+            raise TypeError("Thermal profiles must be immutable tuples")
+        if any(
+            isinstance(step, Distribute) and not isinstance(step.destinations, tuple)
+            for step in self.steps
+        ):
+            raise TypeError("Distribution destinations must be immutable tuples")
+        identity = self.identity or "urn:lab:protocol:" + digest(semantic(self))
+        require_iri(identity)
+        object.__setattr__(self, "identity", identity)
+        steps = tuple(
+            replace(step, identity=step.identity or identity + f"/step_{index + 1}")
+            for index, step in enumerate(self.steps)
+        )
+        if len({step.identity for step in steps}) != len(steps):
+            raise ValueError("Step identities must be unique")
+        for step in steps:
+            require_iri(str(step.identity))
+        object.__setattr__(self, "steps", steps)
+
+    @property
+    def semantic_json(self) -> str:
+        return canonical_json(semantic(self))
+
+    @property
+    def digest(self) -> str:
+        return digest(semantic(self))
 
     def output_manifest(self) -> OutputManifest:
         """Project declared outputs from this snapshot using logical locations."""
         samples = {sample.id: sample for sample in self.samples}
         placements = {placement.sample_id: placement for placement in self.placements}
         return OutputManifest(
-            protocol_id=self.name,
+            protocol_id=str(self.identity),
             samples=tuple(samples[sample_id] for sample_id in self.output_sample_ids),
             placements=tuple(placements[sample_id] for sample_id in self.output_sample_ids),
         )
 
 
-@dataclass(frozen=True)
-class Binding:
-    """One exact physical location and its usable constraints, in microlitres."""
-
-    location: Location
-    physical: str
-    capacity: Decimal
-    dead_volume: Decimal
-
-
-@dataclass(frozen=True)
-class TargetPlan:
-    name: str
-    bindings: tuple[Binding, ...]
-    configuration_json: str
-    source: str | None = None
-    setup: tuple[str, ...] = ()
-
-
 def encode(value: Any) -> Any:
     """JSON-compatible data for the small, closed vocabulary."""
     if isinstance(value, Decimal):
@@ -159,10 +101,33 @@ def encode(value: Any) -> Any:
     if isinstance(value, dict):
         return {key: encode(item) for key, item in value.items()}
     if isinstance(
-        value, (Transfer, Distribute, Mix, Wait, Thermocycle, SetTemperature, ManualInstruction)
+        value,
+        (
+            Transfer,
+            Distribute,
+            Mix,
+            Wait,
+            Thermocycle,
+            SetTemperature,
+            ManualInstruction,
+            ExternalPreparation,
+        ),
     ):
         return {"kind": type(value).__name__, **encode(asdict(value))}
     if hasattr(value, "__dataclass_fields__"):
         # Do not recursively use asdict here: it would erase step discriminants.
         return {name: encode(getattr(value, name)) for name in value.__dataclass_fields__}
     return value
+
+
+def semantic(value: Any) -> Any:
+    """Encode meaning without developer source paths or line numbers."""
+
+    def clean(item: Any) -> Any:
+        if isinstance(item, dict):
+            return {key: clean(val) for key, val in item.items() if key != "origin"}
+        if isinstance(item, list):
+            return [clean(val) for val in item]
+        return item
+
+    return clean(encode(value))
diff --git a/src/lab/operations.py b/src/lab/operations.py
new file mode 100644
index 0000000..617df37
--- /dev/null
+++ b/src/lab/operations.py
@@ -0,0 +1,178 @@
+"""Immutable semantic operations with persistent step identities."""
+
+from dataclasses import dataclass
+from decimal import Decimal
+
+from lab.provenance.types import require_iri
+from lab.samples import Location
+
+
+@dataclass(frozen=True)
+class Origin:
+    file: str
+    line: int
+
+
+@dataclass(frozen=True)
+class Fill:
+    well: str
+    material: str
+    volume: Decimal
+
+
+@dataclass(frozen=True)
+class Resource:
+    name: str
+    rows: int
+    columns: int
+    capacity: Decimal
+    dead_volume: Decimal
+    fills: tuple[Fill, ...] = ()
+
+    @property
+    def wells(self) -> tuple[str, ...]:
+        return tuple(
+            f"{chr(65 + row)}{column + 1}"
+            for row in range(self.rows)
+            for column in range(self.columns)
+        )
+
+
+@dataclass(frozen=True, kw_only=True)
+class Operation:
+    identity: str | None = None
+
+
+@dataclass(frozen=True)
+class Transfer(Operation):
+    source: Location
+    destination: Location
+    volume: Decimal
+    origin: Origin
+    destination_height_mm: Decimal | None = None
+
+    def __post_init__(self) -> None:
+        if self.destination_height_mm is not None and (
+            not isinstance(self.destination_height_mm, Decimal)
+            or not self.destination_height_mm.is_finite()
+            or self.destination_height_mm < 0
+        ):
+            raise ValueError("Destination height must be a nonnegative finite Decimal")
+
+
+@dataclass(frozen=True)
+class Distribute(Operation):
+    """Reuse one tip across destinations, splitting aspirations to fit the target.
+
+    ``air_gap`` is air, not liquid volume.
+    """
+
+    source: Location
+    destinations: tuple[Location, ...]
+    volume: Decimal
+    air_gap: Decimal | None
+    origin: Origin
+
+
+@dataclass(frozen=True)
+class Mix(Operation):
+    location: Location
+    volume: Decimal
+    cycles: int
+    origin: Origin
+
+
+@dataclass(frozen=True)
+class Wait(Operation):
+    seconds: Decimal
+    origin: Origin
+
+
+@dataclass(frozen=True)
+class Hold:
+    celsius: Decimal
+    seconds: Decimal
+
+
+@dataclass(frozen=True)
+class Thermocycle(Operation):
+    resource: str
+    profile: tuple[Hold, ...]
+    cycles: int
+    lid_celsius: Decimal | None
+    origin: Origin
+    block_volume: Decimal | None = None
+
+
+@dataclass(frozen=True)
+class SetTemperature(Operation):
+    """Hold a plate's controlling module at one temperature. Liquid handling may continue."""
+
+    resource: str
+    celsius: Decimal
+    origin: Origin
+
+
+@dataclass(frozen=True)
+class ManualInstruction(Operation):
+    text: str
+    origin: Origin
+
+
+@dataclass(frozen=True, kw_only=True)
+class MaterialPort:
+    location: Location
+    volume_ul: Decimal = Decimal(0)
+    count: int = 0
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.location, Location):
+            raise TypeError("A material port needs a logical Location")
+        if (
+            not isinstance(self.volume_ul, Decimal)
+            or not self.volume_ul.is_finite()
+            or self.volume_ul < 0
+            or type(self.count) is not int
+            or self.count < 0
+            or (self.volume_ul > 0) == (self.count > 0)
+        ):
+            raise ValueError("A material port specifies either positive volume or count")
+
+
+@dataclass(frozen=True)
+class ExternalPreparation(Operation):
+    procedure: str
+    instructions: str
+    inputs: tuple[MaterialPort, ...]
+    outputs: tuple[MaterialPort, ...]
+    origin: Origin
+
+    def __post_init__(self) -> None:
+        require_iri(self.procedure)
+        if not isinstance(self.instructions, str) or not self.instructions.strip():
+            raise ValueError("Supply external procedure instructions")
+        for ports in (self.inputs, self.outputs):
+            if (
+                not isinstance(ports, tuple)
+                or not ports
+                or not all(isinstance(p, MaterialPort) for p in ports)
+            ):
+                raise TypeError(
+                    "External inputs and outputs must be nonempty tuples of material ports"
+                )
+            if len({port.location for port in ports}) != len(ports):
+                raise ValueError("Combine quantities for repeated material ports")
+        if {p.location for p in self.inputs} & {p.location for p in self.outputs}:
+            raise ValueError("External preparation outputs require distinct locations")
+
+
+Step = (
+    Transfer
+    | Distribute
+    | Mix
+    | Wait
+    | Thermocycle
+    | SetTemperature
+    | ManualInstruction
+    | ExternalPreparation
+)
diff --git a/src/lab/protocol.py b/src/lab/protocol.py
index 6ba2fe2..2dae84e 100644
--- a/src/lab/protocol.py
+++ b/src/lab/protocol.py
@@ -7,14 +7,16 @@
 from pathlib import Path
 from typing import Any
 
-from lab.model import (
+from lab.model import RecordedProtocol
+from lab.operations import (
     Distribute,
+    ExternalPreparation,
     Fill,
     Hold,
     ManualInstruction,
+    MaterialPort,
     Mix,
     Origin,
-    RecordedProtocol,
     Resource,
     SetTemperature,
     Step,
@@ -77,9 +79,10 @@ def _text(value: str, label: str) -> str:
 class Protocol:
     """Record sequential work. Construction and compilation never operate hardware."""
 
-    def __init__(self, name: str, *, description: str = "") -> None:
+    def __init__(self, name: str, *, description: str = "", identity: str | None = None) -> None:
         self.name = _text(name, "Protocol name")
         self.description = description
+        self.identity = identity
         self._owner = object()
         self._resources: dict[str, Resource] = {}
         self._steps: list[Step] = []
@@ -150,6 +153,21 @@ def _location(self, well: Well) -> Location:
             raise ValueError("Unknown well")
         return Location(well.resource, well.name)
 
+    def external_preparation(
+        self,
+        *,
+        procedure: str,
+        instructions: str,
+        inputs: tuple[MaterialPort, ...],
+        outputs: tuple[MaterialPort, ...],
+    ) -> None:
+        """Specify operator work and its expected material balance, without executing it."""
+        for port in (*inputs, *outputs):
+            resource = self._resources.get(port.location.resource)
+            if resource is None or port.location.well not in resource.wells:
+                raise ValueError("External material ports must reference this protocol's wells")
+        self._steps.append(ExternalPreparation(procedure, instructions, inputs, outputs, _origin()))
+
     def load(self, well: Well, material: str, *, volume: Any) -> None:
         """Declare initial contents before recording any steps; this is not a transfer."""
         if self._steps:
@@ -191,11 +209,18 @@ def add_sample(
         if is_output:
             self._output_sample_ids.append(sample.id)
 
-    def transfer(self, source: Well, destination: Well, *, volume: Any) -> None:
+    def transfer(
+        self, source: Well, destination: Well, *, volume: Any, destination_height: Any = None
+    ) -> None:
         start, end = self._location(source), self._location(destination)
         if start == end:
             raise ValueError("Transfer source and destination must differ")
-        self._steps.append(Transfer(start, end, magnitude(volume, "microliter"), _origin()))
+        height = (
+            None
+            if destination_height is None
+            else magnitude(destination_height, "millimeter", positive=False)
+        )
+        self._steps.append(Transfer(start, end, magnitude(volume, "microliter"), _origin(), height))
 
     def distribute(
         self,
@@ -280,4 +305,5 @@ def snapshot(self) -> RecordedProtocol:
             tuple(self._placements),
             tuple(self._input_sample_ids),
             tuple(self._output_sample_ids),
+            self.identity,
         )
diff --git a/src/lab/samples.py b/src/lab/samples.py
index 04c8a76..3e54aea 100644
--- a/src/lab/samples.py
+++ b/src/lab/samples.py
@@ -2,6 +2,9 @@
 
 from dataclasses import dataclass
 
+from lab.inventory import MaterialForm
+from lab.provenance import Component, Implementation, Ref
+
 
 @dataclass(frozen=True)
 class Location:
@@ -16,6 +19,8 @@ def __str__(self) -> str:
 
 @dataclass(frozen=True, slots=True, kw_only=True)
 class Sample:
+    """Material at a location; parent_ids are contributions, not genetic ancestry."""
+
     id: str
     material_identity: str
     label: str
@@ -26,8 +31,25 @@ class Sample:
     source_protocol_id: str | None = None
     contents: tuple[str, ...] = ()
     dilution: int | None = None
+    design: Ref[Component] | None = None
+    implementation: Ref[Implementation] | None = None
+    form: MaterialForm | None = None
+    count: int | None = None
 
     def __post_init__(self) -> None:
+        if self.form is not None and not isinstance(self.form, MaterialForm):
+            raise TypeError("Sample form must be a MaterialForm")
+        if self.form is not None and self.form.counted:
+            if type(self.count) is not int or self.count < 1:
+                raise ValueError("Counted samples need a positive integer count")
+        elif self.count is not None:
+            raise ValueError("Only counted material forms have a count")
+        if self.design is not None and (
+            not isinstance(self.design, Ref) or self.material_identity != self.design.identity
+        ):
+            raise ValueError("Sample material identity must match its design reference")
+        if self.implementation is not None and not isinstance(self.implementation, Ref):
+            raise TypeError("Sample implementation must be a reference")
         if not all(
             isinstance(value, str) and value.strip()
             for value in (self.id, self.material_identity, self.label, self.role)
@@ -91,9 +113,17 @@ def to_dict(self) -> dict[str, object]:
                 {
                     "sample_id": sample.id,
                     "material_identity": sample.material_identity,
+                    "design": None if sample.design is None else sample.design.identity,
+                    "implementation": (
+                        None if sample.implementation is None else sample.implementation.identity
+                    ),
                     "label": sample.label,
                     "parent_sample_ids": list(sample.parent_ids),
                     "replicate": sample.replicate,
+                    "role": sample.role,
+                    "form": None if sample.form is None else sample.form.value,
+                    "count": sample.count,
+                    "dilution": sample.dilution,
                     "source_sample_id": sample.source_sample_id,
                     "source_protocol_id": sample.source_protocol_id,
                     "contents": list(sample.contents),
diff --git a/src/lab/target.py b/src/lab/target.py
new file mode 100644
index 0000000..9d62b92
--- /dev/null
+++ b/src/lab/target.py
@@ -0,0 +1,25 @@
+"""Frozen physical target artifacts, independent of device SDKs."""
+
+from dataclasses import dataclass
+from decimal import Decimal
+
+from lab.samples import Location
+
+
+@dataclass(frozen=True)
+class Binding:
+    """One exact physical location and its usable constraints, in microlitres."""
+
+    location: Location
+    physical: str
+    capacity: Decimal
+    dead_volume: Decimal
+
+
+@dataclass(frozen=True)
+class TargetPlan:
+    name: str
+    bindings: tuple[Binding, ...]
+    configuration_json: str
+    source: str | None = None
+    setup: tuple[str, ...] = ()
diff --git a/src/lab/targets/manual.py b/src/lab/targets/manual.py
index d31b161..8f6676c 100644
--- a/src/lab/targets/manual.py
+++ b/src/lab/targets/manual.py
@@ -1,6 +1,7 @@
 from dataclasses import dataclass
 
-from lab.model import RecordedProtocol, TargetPlan
+from lab.model import RecordedProtocol
+from lab.target import TargetPlan
 from lab.validation import logical_bindings
 
 
diff --git a/src/lab/targets/opentrons.py b/src/lab/targets/opentrons.py
index 2f9291e..ea360d5 100644
--- a/src/lab/targets/opentrons.py
+++ b/src/lab/targets/opentrons.py
@@ -10,19 +10,18 @@
 from lab.documents import describe
 from lab.equipment import LabwareModel, ModuleModel, TipRackModel
 from lab.labware import LabwareKind
-from lab.model import (
-    Binding,
+from lab.model import RecordedProtocol
+from lab.operations import (
     Distribute,
     ManualInstruction,
     Mix,
-    RecordedProtocol,
     SetTemperature,
-    TargetPlan,
     Thermocycle,
     Transfer,
     Wait,
 )
 from lab.samples import Location
+from lab.target import Binding, TargetPlan
 from lab.targets.liquid_handler import LiquidHandler
 from lab.units import number
 from lab.validation import CompileError, step_error, volume_trace
@@ -86,11 +85,10 @@ def prepare(self, protocol: RecordedProtocol) -> TargetPlan:
             "p300_single_gen2": ("OT-2", 20, 300, "opentrons_96_tiprack_300ul"),
             "p20_single_gen2": ("OT-2", 1, 20, "opentrons_96_tiprack_20ul"),
             "flex_1channel_1000": ("Flex", 5, 1000, "opentrons_flex_96_tiprack_200ul"),
+            "flex_1channel_50": ("Flex", 1, 50, "opentrons_flex_96_tiprack_50ul"),
         }
         pipette_specs = [(self.pipette, self.mount, self.tip_racks, "pipette")]
         if self.small_pipette is not None:
-            if self.robot != "OT-2":
-                raise CompileError("A second pipette is supported on the OT-2")
             if not self.small_tip_racks:
                 raise CompileError("The second pipette needs its own tip racks")
             if self.small_mount not in ("left", "right") or self.small_mount == self.mount:
@@ -103,7 +101,7 @@ def prepare(self, protocol: RecordedProtocol) -> TargetPlan:
             if model_name not in models or models[model_name][0] != self.robot:
                 raise CompileError(
                     "Supported pipettes: OT-2 P20 single GEN2, OT-2 P300 single GEN2, "
-                    "and Flex 1-channel 1000"
+                    "and Flex 1-channel 50 or 1000"
                 )
             if mount not in ("left", "right"):
                 raise CompileError("Pipette mount must be left or right")
@@ -259,6 +257,7 @@ def select_pipette(volume: Decimal) -> _Pipette:
             return min(fits, key=lambda pipette: pipette.maximum)
 
         for index, step in enumerate(protocol.steps):
+            commands.append(f"# lab:step {step.identity}")
             commands.append(f"context.comment({describe(step)!r})")
             if isinstance(step, (Transfer, Mix, Distribute)):
                 try:
@@ -273,13 +272,26 @@ def select_pipette(volume: Decimal) -> _Pipette:
                 if step.volume > tip_capacity:
                     raise step_error(index, step, "Volume is outside the pipette/tip range")
                 tool = pipette.variable
+                if pipette.model == "flex_1channel_50":
+                    commands.append(f"{tool}.configure_for_volume({number(step.volume)})")
+                if isinstance(step, Distribute):
+                    mode_max = (
+                        Decimal(30)
+                        if pipette.model == "flex_1channel_50" and step.volume < 5
+                        else pipette.maximum
+                    )
+                    if step.volume + (step.air_gap or Decimal(0)) > min(mode_max, tip_capacity):
+                        raise step_error(index, step, "Liquid and air gap exceed the volume mode")
                 commands.append(f"{tool}.pick_up_tip({tip})")
                 amount = number(step.volume)
                 if isinstance(step, Transfer):
+                    destination = refs[step.destination]
+                    if step.destination_height_mm is not None:
+                        destination += f".bottom(z={number(step.destination_height_mm)})"
                     commands.extend(
                         (
                             f"{tool}.aspirate({amount}, {refs[step.source]})",
-                            f"{tool}.dispense({amount}, {refs[step.destination]})",
+                            f"{tool}.dispense({amount}, {destination})",
                         )
                     )
                 elif isinstance(step, Distribute):
@@ -351,6 +363,7 @@ def select_pipette(volume: Decimal) -> _Pipette:
                 commands.append(f"context.pause({step.text!r})")
             else:
                 raise step_error(index, step, "Unsupported step")
+            commands.append(f"# lab:end {step.identity}")
         source = (
             '"""Generated by Lab. Each liquid operation uses a fresh tip."""\n\n'
             "from opentrons import protocol_api\n\n"
@@ -432,11 +445,27 @@ def lower_deck(
     if robot == "Flex":
         if "D1" in taken:
             raise CompileError("Flex tip rack slot D1 is already in use")
+        small = not volumes or any(volume <= 50 for volume in volumes)
+        large = any(volume > 50 for volume in volumes)
+        second_tips = (
+            (Labware("opentrons_flex_96_tiprack_50ul", _open_slot(("D2", "C2", "B2"), taken)),)
+            if small and large
+            else ()
+        )
         return Opentrons(
             robot="Flex",
-            pipette="flex_1channel_1000",
+            pipette="flex_1channel_1000" if large else "flex_1channel_50",
             labware=labware,
-            tip_racks=(Labware("opentrons_flex_96_tiprack_200ul", "D1"),),
+            tip_racks=(
+                Labware(
+                    "opentrons_flex_96_tiprack_200ul"
+                    if large
+                    else "opentrons_flex_96_tiprack_50ul",
+                    "D1",
+                ),
+            ),
+            small_pipette="flex_1channel_50" if small and large else None,
+            small_tip_racks=second_tips,
             thermocycler=thermocycler,
         )
     small = any(volume <= 20 for volume in volumes) or not any(volume > 20 for volume in volumes)
@@ -525,6 +554,7 @@ def _open_slot(slots: tuple[str, ...], taken: set[str]) -> str:
     TipRackModel.OPENTRONS_20_UL: "opentrons_96_tiprack_20ul",
     TipRackModel.OPENTRONS_300_UL: "opentrons_96_tiprack_300ul",
     TipRackModel.FLEX_200_UL: "opentrons_flex_96_tiprack_200ul",
+    TipRackModel.FLEX_50_UL: "opentrons_flex_96_tiprack_50ul",
 }
 
 
@@ -538,10 +568,8 @@ def lower_layout(layout: DeckLayout) -> Opentrons:
             f"{robot} does not support carriers, independent channels, "
             "or external thermal handoffs."
         )
-    if not 1 <= len(layout.pipettes) <= (2 if robot == "OT-2" else 1):
-        raise CompileError(
-            f"Configure one {'or two pipettes' if robot == 'OT-2' else 'pipette'} for {robot}."
-        )
+    if not 1 <= len(layout.pipettes) <= 2:
+        raise CompileError(f"Configure one or two pipettes for {robot}.")
     assigned_tips = [rack for pipette in layout.pipettes for rack in pipette.tip_racks]
     if len(set(assigned_tips)) != len(assigned_tips):
         raise CompileError("The Opentrons backend requires separate tip racks for each pipette.")
diff --git a/src/lab/targets/star.py b/src/lab/targets/star.py
index 44c7b46..eaeec4e 100644
--- a/src/lab/targets/star.py
+++ b/src/lab/targets/star.py
@@ -25,20 +25,18 @@
 from lab.documents import describe
 from lab.equipment import CarrierModel, LabwareModel, TipRackModel
 from lab.labware import LabwareKind
-from lab.model import (
-    Binding,
+from lab.model import RecordedProtocol, encode
+from lab.operations import (
     Distribute,
     ManualInstruction,
     Mix,
-    RecordedProtocol,
     SetTemperature,
-    TargetPlan,
     Thermocycle,
     Transfer,
     Wait,
-    encode,
 )
 from lab.samples import Location
+from lab.target import Binding, TargetPlan
 from lab.targets.liquid_handler import LiquidHandler
 from lab.units import magnitude, number, uL
 from lab.validation import CompileError, step_error
@@ -167,6 +165,7 @@ def on_deck(resource: Any) -> None:
         commands = []
         tip_index = 0
         for index, step in enumerate(protocol.steps):
+            commands.append(f"# lab:step {step.identity}")
             if isinstance(step, (Thermocycle, SetTemperature)):
                 if (
                     self.external_thermal_resources is not None
@@ -201,12 +200,17 @@ def on_deck(resource: Any) -> None:
                     f"await lh.pick_up_tips([deck.get_resource({spot.name!r})], {channels})"
                 )
                 if isinstance(step, Transfer):
+                    height = (
+                        ""
+                        if step.destination_height_mm is None
+                        else f", liquid_height=[{number(step.destination_height_mm)}]"
+                    )
                     commands.extend(
                         (
                             f"await lh.aspirate([{refs[step.source]}], "
                             f"vols=[{amount}], {channels})",
                             f"await lh.dispense([{refs[step.destination]}], "
-                            f"vols=[{amount}], {channels})",
+                            f"vols=[{amount}], {channels}{height})",
                         )
                     )
                 elif isinstance(step, Distribute):
@@ -259,11 +263,12 @@ def on_deck(resource: Any) -> None:
                 if not isinstance(plate, ItemizedResource):
                     raise step_error(index, step, "A held temperature must bind a whole plate")
                 commands.append(
-                    f"await thermocycle(deck.get_resource({plate.name!r}), "
-                    f"profile=[({number(step.celsius)}, 0)], cycles=1, lid_temperature=None)"
+                    f"await set_temperature(deck.get_resource({plate.name!r}), "
+                    f"temperature={number(step.celsius)})"
                 )
             else:
                 raise step_error(index, step, "Unsupported step")
+            commands.append(f"# lab:end {step.identity}")
         initialize = []
         for resource in protocol.resources:
             fills = {fill.well: fill.volume for fill in resource.fills}
@@ -275,6 +280,8 @@ def on_deck(resource: Any) -> None:
         has_thermal = any(
             isinstance(step, (Thermocycle, SetTemperature)) for step in protocol.steps
         )
+        has_cycle = any(isinstance(step, Thermocycle) for step in protocol.steps)
+        has_hold = any(isinstance(step, SetTemperature) for step in protocol.steps)
         thermal_import = "from inspect import iscoroutinefunction\n" if has_thermal else ""
         source = (
             '"""Generated PyLabRobot protocol. Direct execution uses a software backend."""\n\n'
@@ -285,7 +292,8 @@ def on_deck(resource: Any) -> None:
             "from pylabrobot.resources import Resource\n\n"
             f"DECK_JSON = {serialized_deck!r}\n"
             f"SDK_VERSION = {version('pylabrobot')!r}\n\n"
-            "async def run(backend, *, confirm=None, thermocycle=None, sleep=asyncio.sleep):\n"
+            "async def run(backend, *, confirm=None, thermocycle=None, "
+            "set_temperature=None, sleep=asyncio.sleep):\n"
             "    if version('pylabrobot') != SDK_VERSION:\n"
             "        raise RuntimeError(f'This artifact requires pylabrobot=={SDK_VERSION}')\n"
         )
@@ -294,12 +302,18 @@ def on_deck(resource: Any) -> None:
                 "    if confirm is None:\n"
                 "        raise ValueError('Supply confirm for explicit operator steps')\n"
             )
-        if has_thermal:
+        if has_cycle:
             source += (
                 "    if not iscoroutinefunction(thermocycle):\n"
                 "        raise ValueError('Supply an async thermocycle callback for the external "
                 "thermal device; it must return the plate to its original position')\n"
             )
+        if has_hold:
+            source += (
+                "    if not iscoroutinefunction(set_temperature):\n"
+                "        raise ValueError('Supply an async set_temperature callback; "
+                "the hold must persist while the plate remains accessible for pipetting')\n"
+            )
         source += (
             "    deck = Resource.deserialize(json.loads(DECK_JSON))\n"
             + "\n".join(f"    {line}" for line in initialize)
@@ -310,13 +324,20 @@ def on_deck(resource: Any) -> None:
             + "\n    finally:\n        await lh.stop()\n"
             "    return lh\n\n"
         )
-        if has_thermal:
+        if has_cycle:
             source += (
                 "async def preview_thermocycle(plate, *, profile, cycles, lid_temperature):\n"
                 "    print(f'Simulation only: thermal profile for {plate.name}: {profile}; "
                 "{cycles} cycles; lid {lid_temperature} C')\n\n"
             )
-        thermal_callback = ", thermocycle=preview_thermocycle" if has_thermal else ""
+        if has_hold:
+            source += (
+                "async def preview_set_temperature(plate, *, temperature):\n"
+                "    print(f'Simulation only: persistent hold for {plate.name}: "
+                "{temperature} C')\n\n"
+            )
+        thermal_callback = ", thermocycle=preview_thermocycle" if has_cycle else ""
+        thermal_callback += ", set_temperature=preview_set_temperature" if has_hold else ""
         source += (
             "if __name__ == '__main__':\n"
             "    asyncio.run(run(LiquidHandlerChatterboxBackend(), confirm=input"
@@ -437,7 +458,7 @@ def place(resource: Any, location: Any, expected_carrier: CarrierModel) -> None:
 
 
 def lower_deck(deck: Deck, *, volumes: tuple[Decimal, ...]) -> STAR:
-    """Resolve an ambient plate preset; other equipment needs a Lab DeckLayout."""
+    """Resolve plate carriers, with explicit runtime contracts for thermal plates."""
     if len(deck.containers) > 10:
         raise CompileError("The STAR plate preset holds at most ten plates; provide a DeckLayout.")
     placements = []
@@ -453,6 +474,7 @@ def lower_deck(deck: Deck, *, volumes: tuple[Decimal, ...]) -> STAR:
         if not isinstance(container, DeckContainer) or container.site not in (
             DeckSite.PLATES,
             DeckSite.MORE_PLATES,
+            DeckSite.THERMOCYCLER,
         ):
             raise CompileError(
                 f"No STAR preset for {container.id}; provide a Lab DeckLayout "
@@ -493,5 +515,10 @@ def lower_deck(deck: Deck, *, volumes: tuple[Decimal, ...]) -> STAR:
                 tip_racks=("tips",),
             ),
         ),
+        external_thermal_resources=tuple(
+            container.id
+            for container in deck.containers
+            if isinstance(container, DeckContainer) and container.site == DeckSite.THERMOCYCLER
+        ),
     )
     return lower_layout(deck, layout)
diff --git a/src/lab/validation.py b/src/lab/validation.py
index 0f1a6e1..0bd85a7 100644
--- a/src/lab/validation.py
+++ b/src/lab/validation.py
@@ -2,17 +2,19 @@
 
 from decimal import Decimal
 
-from lab.model import (
-    Binding,
+from lab.inventory import MaterialForm
+from lab.model import RecordedProtocol
+from lab.operations import (
     Distribute,
+    ExternalPreparation,
     Mix,
-    RecordedProtocol,
     SetTemperature,
     Step,
     Thermocycle,
     Transfer,
 )
 from lab.samples import Location
+from lab.target import Binding
 from lab.units import number
 
 
@@ -26,6 +28,7 @@ def step_error(index: int, step: Step, message: str) -> CompileError:
 
 def validate(protocol: RecordedProtocol, bindings: tuple[Binding, ...]) -> dict[Location, Decimal]:
     validate_samples(protocol)
+    count_trace(protocol)
     return volume_trace(protocol, bindings)[-1]
 
 
@@ -102,10 +105,20 @@ def volume_trace(
             if fill.volume > capacities[location]:
                 raise CompileError(f"Initial volume exceeds the bound capacity of {location}")
             volumes[location] = fill.volume
+    samples = {sample.id: sample for sample in protocol.samples}
+    counted = {p.location for p in protocol.placements if samples[p.sample_id].count is not None}
+    if any(volumes[location] for location in counted):
+        raise CompileError("Counted materials cannot be loaded as liquid volumes")
     states = [volumes.copy()]
     for index, step in enumerate(protocol.steps):
         if isinstance(step, (Transfer, Mix)):
             source = step.source if isinstance(step, Transfer) else step.location
+            if source in counted:
+                raise step_error(
+                    index,
+                    step,
+                    "Counted material requires explicit external preparation before pipetting",
+                )
             if source not in volumes:
                 raise step_error(index, step, f"Unknown source {source}")
             available = max(Decimal(0), volumes[source] - dead[source])
@@ -124,6 +137,8 @@ def volume_trace(
                 volumes[source] -= step.volume
                 volumes[step.destination] += step.volume
         elif isinstance(step, Distribute):
+            if step.source in counted:
+                raise step_error(index, step, "Counted material cannot be pipetted")
             if step.source not in volumes:
                 raise step_error(index, step, f"Unknown source {step.source}")
             needed = step.volume * len(step.destinations)
@@ -143,6 +158,23 @@ def volume_trace(
             volumes[step.source] -= needed
             for destination in step.destinations:
                 volumes[destination] += step.volume
+        elif isinstance(step, ExternalPreparation):
+            for port in step.inputs:
+                if port.location not in volumes:
+                    raise step_error(index, step, "Unknown external input location")
+                if port.volume_ul > max(Decimal(0), volumes[port.location] - dead[port.location]):
+                    raise step_error(
+                        index, step, "External preparation exceeds available input volume"
+                    )
+                volumes[port.location] -= port.volume_ul
+            for port in step.outputs:
+                if port.location not in volumes or volumes[port.location]:
+                    raise step_error(
+                        index, step, "External output must occupy a known empty location"
+                    )
+                if port.volume_ul > capacities[port.location]:
+                    raise step_error(index, step, "External output exceeds container capacity")
+                volumes[port.location] = port.volume_ul
         elif isinstance(step, Thermocycle):
             contents = [v for loc, v in volumes.items() if loc.resource == step.resource]
             if not contents:
@@ -156,6 +188,58 @@ def volume_trace(
     return tuple(states)
 
 
+def count_trace(protocol: RecordedProtocol) -> tuple[dict[Location, int], ...]:
+    """Count whole materials separately; pipetting never consumes those counts."""
+    samples = {sample.id: sample for sample in protocol.samples}
+    at = {place.location: samples[place.sample_id] for place in protocol.placements}
+    counted = {location: sample for location, sample in at.items() if sample.count is not None}
+    counts = {
+        location: int(sample.count or 0) if sample.id in protocol.input_sample_ids else 0
+        for location, sample in counted.items()
+    }
+    states = [counts.copy()]
+    for index, step in enumerate(protocol.steps):
+        if isinstance(step, ExternalPreparation):
+            for port in (*step.inputs, *step.outputs):
+                if port.location not in at or (port.count > 0) != (port.location in counted):
+                    raise step_error(
+                        index, step, "External ports must match declared material forms"
+                    )
+            for port in step.inputs:
+                if port.count:
+                    if port.count > counts[port.location]:
+                        raise step_error(
+                            index, step, "External preparation exceeds available material count"
+                        )
+                    counts[port.location] -= port.count
+            for port in step.outputs:
+                if port.count:
+                    if counts[port.location]:
+                        raise step_error(
+                            index, step, "External counted output requires an empty location"
+                        )
+                    counts[port.location] = port.count
+        elif isinstance(step, (Transfer, Distribute)):
+            destinations = (step.destination,) if isinstance(step, Transfer) else step.destinations
+            for destination in destinations:
+                if destination in counted:
+                    if (
+                        counted[destination].form is not MaterialForm.PLATED_SAMPLE
+                        or counts[destination]
+                    ):
+                        raise step_error(
+                            index, step, "A deposition requires an empty plated-sample location"
+                        )
+                    counts[destination] = 1
+        states.append(counts.copy())
+    for location, sample in counted.items():
+        if sample.id in protocol.output_sample_ids and counts[location] != sample.count:
+            raise CompileError(
+                "Declared output count does not match the protocol's material balance"
+            )
+    return tuple(states)
+
+
 def logical_bindings(protocol: RecordedProtocol) -> tuple[Binding, ...]:
     return tuple(
         Binding(
diff --git a/tests/test_layouts.py b/tests/test_layouts.py
index 36e4561..7ef760d 100644
--- a/tests/test_layouts.py
+++ b/tests/test_layouts.py
@@ -232,4 +232,4 @@ def test_star_thermal_steps_require_a_declared_handoff():
         deck=replace(example_deck(), layouts=(layout,)),
         liquid_handler=LiquidHandler.STAR,
     )
-    assert "Supply an async thermocycle callback" in bundle.files["protocol.py"]
+    assert "Supply an async set_temperature callback" in bundle.files["protocol.py"]
diff --git a/tests/test_protocol_metadata.py b/tests/test_protocol_metadata.py
new file mode 100644
index 0000000..309a18a
--- /dev/null
+++ b/tests/test_protocol_metadata.py
@@ -0,0 +1,48 @@
+from dataclasses import replace
+from io import StringIO
+
+import pytest
+
+import lab
+from lab import Protocol, seconds, uL
+from lab.experiments.cloning.decks import assembly_deck
+from lab.operations import Origin
+from lab.targets import LiquidHandler
+
+
+def test_semantic_identity_is_independent_of_authoring_paths():
+    p = Protocol("Same meaning")
+    p.wait(1 * seconds)
+    original = p.snapshot()
+    moved = replace(
+        original,
+        steps=tuple(
+            replace(
+                step,
+                origin=Origin("/another/machine/experiment.py", 800),
+            )
+            for step in original.steps
+        ),
+    )
+    assert moved.digest == original.digest
+    assert moved.steps[0].identity == original.steps[0].identity
+    p.wait(2 * seconds)
+    assert original.digest != p.snapshot().digest
+    with pytest.raises(ValueError, match="unique"):
+        replace(original, steps=(*original.steps, *original.steps))
+
+
+@pytest.mark.integration
+def test_flex_switches_volume_modes_before_picking_up_fresh_tips():
+    simulator = pytest.importorskip("opentrons.simulate")
+
+    p = Protocol("Mixed volumes")
+    source = p.plate("reagents", shape=(4, 6), capacity=1500 * uL)
+    target = p.plate("products", capacity=100 * uL)
+    p.load(source["A1"], "water", volume=200 * uL)
+    for index, volume in enumerate((1, 4, 5, 30, 50), 1):
+        p.transfer(source["A1"], target[f"A{index}"], volume=volume * uL)
+    artifact = lab.compile(p, deck=assembly_deck(), liquid_handler=LiquidHandler.FLEX)
+    source_text = artifact.files["protocol.py"]
+    assert source_text.count("configure_for_volume") == 5
+    simulator.simulate(StringIO(source_text))
diff --git a/tests/test_stages.py b/tests/test_stages.py
index 38bba47..9abed65 100644
--- a/tests/test_stages.py
+++ b/tests/test_stages.py
@@ -24,7 +24,7 @@
     transformation_deck,
 )
 from lab.experiments.cloning.addresses import well_name
-from lab.model import Mix, Transfer
+from lab.operations import Mix, Transfer
 from lab.part import Part
 from lab.samples import Location
 from lab.targets import Labware, LiquidHandler, Manual
@@ -137,7 +137,7 @@ def test_transformation_uses_caller_defined_materials():
     )
     compiled = lab.compile(transformation, Manual())
     assert compiled.protocol.name == transformation.id
-    assert compiled.manifest.protocol_id == transformation.id
+    assert compiled.manifest.protocol_id == compiled.protocol.identity
     assert {sample.material_identity for sample in compiled.manifest.samples} == {"custom-strain"}
     assert {
         sample.material_identity for sample in compiled.protocol.samples if sample.role == "dna"