Stream and Markers

stream(...) projects a macro sample into a stream of discrete units.

Marker panels can be declared with markers(...) and attached to a stream. This belongs with material-side workflows: it maps material from a container into unit-level iteration rather than describing a separate execution family.

Markers

let panel = markers(["CD3", "CD19"]);

markers([...]) declares a controlled set of marker names for later use. Marker names are not hard-coded by the grammar.

Stream

let events = stream(sample = cells, unit = ObservationUnit.SINGLE_CELL, panel = panel);

Arguments:

Argument	Required	Meaning
sample	yes	Input macro sample container.
unit	yes	ObservationUnit.SINGLE_CELL or an installed observation-unit member. Legacy text remains accepted.
panel	no	Marker panel reference.

stream(...) returns a unit_stream_ref.

Unit Types

ObservationUnit.SINGLE_CELL identifies the standard single-cell unit; single_cell remains an equivalent text spelling. Historical labels such as molecule and ion remain text rather than registered members. Custom types inherit ObservationUnitBase and are installed by the Python host for the compilation and execution scope:

import culsma.enum_services
from culsma.domains.stream import ObservationUnitBase, STANDARD_OBSERVATION_UNITS
from culsma.parser import parse
from culsma.frontend.resolver import resolve_program
from culsma.pipeline.compile import compile_ast
from culsma.pipeline.validate import validate
from culsma.pipeline.typecheck import typecheck
from culsma.pipeline.plan import lower_ir_to_plan
from culsma.runtime.executor import run
from culsma.driver.stub import StubDriver

class LabUnit(ObservationUnitBase):
    DROPLET = "lab_droplet"

units = STANDARD_OBSERVATION_UNITS.with_type(LabUnit, "example.observation.unit", version=1)
source = "protocol Demo { let cells=tube(); let events=stream(sample=cells, unit=LabUnit.DROPLET); }"
with units.activate():
    compiled = compile_ast(resolve_program(parse(source)).prepared_program)
    assert validate(compiled.ir, analysis=compiled.analysis, enforce_binding=True).ok
    assert typecheck(compiled.ir, analysis=compiled.analysis).ok
    plan = lower_ir_to_plan(compiled.ir)
    assert not plan.diagnostics
    result = run(plan=plan, driver=StubDriver())
    assert result.ok

This example demonstrates type identity, not droplet generation. A type name or member does not supply observation data or inherit single-cell behavior. Extensions persist with stable ID, version and member; replay requires the matching installed registry. Leaving the scope removes the extension, including for retained decoders. Cross-family multiple inheritance and source-name collisions are rejected.

Unit Lifetime

Units do not exist independently outside the stream workflow. They are normally introduced by a repeat binding over a stream:

repeat cell in events {
    keep_tube << [cell];
}

When a unit is collected back into a physical container, use transfer syntax:

keep_tube << [cell];

Examples

let molecules = stream(sample = prepared_tube, unit = molecule);
let ions = stream(sample = ion_chamber, unit = ion);

Streams are for discrete unit workflows. They are not a general replacement for container transfer.