Skip to content

:>> [[ABIO]] → ABIO Docs → Modules ABIO expr API Reference

Expr Module

The Expr language: forms, the environment, the registry, the interpreter, the YAML tags, includes. The user-facing specification is ABIO Expr Spec; the Python side is ABIO Expr Python API.

alienbio.expr

alienbio.expr — the Expr language (M47): forms, the interpreter, the head registry, the three spellings. See the vault's ABIO Expr Spec and ABIO Expr Python API.

from alienbio.expr import X, Env, evaluate, fn, expander

env = Env.standard(seed=7)
evaluate(X.lognormal(1.0, 0.3), env)          # a draw
evaluate(X.parse("max(2, poisson(3))"), env)  # the inline spelling

Ctx dataclass

The per-node context: seed, path (for messages), trust, limits.

Source code in src/alienbio/expr/env.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass(frozen=True)
class Ctx:
    """The per-node context: seed, path (for messages), trust, limits."""

    seed: Seed
    path: str = ""
    trusted: bool = False
    limits: Limits = field(default_factory=Limits)
    #: Template instance name -> the document path that claimed it; shared by
    #: every child ctx of one load (``replace`` copies the reference), so a
    #: second path claiming a name refuses (T054 #5).
    instances: dict[str, str] = field(default_factory=dict)

    @property
    def rng(self) -> np.random.Generator:
        return self.seed.rng()

    def child(self, label: str) -> "Ctx":
        path = f"{self.path}.{label}" if self.path else label
        return replace(self, seed=self.seed.child(label), path=path)

Env

Bindings + heads + context. Immutable in spirit: every child / bind / with_* returns a new Env sharing what is unchanged.

Source code in src/alienbio/expr/env.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
class Env:
    """Bindings + heads + context. Immutable in spirit: every ``child`` /
    ``bind`` / ``with_*`` returns a new ``Env`` sharing what is unchanged."""

    def __init__(self, bindings: Scope, registry: Registry, ctx: Ctx, ns: str = "", depth: int = 0) -> None:
        self.bindings = bindings
        self.registry = registry
        self.ctx = ctx
        self.ns = ns
        self.depth = depth

    # ---- construction -----------------------------------------------------

    @classmethod
    def standard(
        cls,
        seed: Union[int, Seed] = 0,
        *,
        trusted: bool = False,
        registry: Optional[Registry] = None,
        limits: Optional[Limits] = None,
        bindings: Optional[Mapping[str, Any]] = None,
    ) -> "Env":
        """An environment over the default registry with root seed ``seed``."""
        from . import heads as _heads  # noqa: F401  (registers the builtin heads)
        from ..suite import expr_heads as _suite_heads  # noqa: F401  (layers 0-2: blocks, worlds)
        from ..suite import expr_experiment as _suite_experiment  # noqa: F401  (layers 3-6: tasks, experiments, agents)

        root_seed = seed if isinstance(seed, Seed) else Seed(int(seed))
        ctx = Ctx(seed=root_seed, trusted=trusted, limits=limits or Limits())
        return cls(Scope(dict(bindings or {}), name="root"), registry or _default_registry, ctx)

    def child(self, label: str) -> "Env":
        """The env for a named sub-node: same scope, child seed, extended path."""
        return Env(self.bindings, self.registry, self.ctx.child(str(label)), self.ns, self.depth + 1)

    def bind(self, **values: Any) -> "Env":
        """A child scope holding ``values`` (already evaluated)."""
        return Env(self.bindings.child(dict(values)), self.registry, self.ctx, self.ns, self.depth)

    def scope(self, data: Optional[Mapping[str, Any]] = None, parent: Optional[Scope] = None) -> "Env":
        """A child scope over ``parent`` (default: this env's bindings)."""
        base = parent if parent is not None else self.bindings
        return Env(base.child(dict(data or {})), self.registry, self.ctx, self.ns, self.depth)

    def with_seed(self, seed: Seed) -> "Env":
        return Env(self.bindings, self.registry, replace(self.ctx, seed=seed), self.ns, self.depth)

    def with_ctx(self, ctx: Ctx) -> "Env":
        return Env(self.bindings, self.registry, ctx, self.ns, self.depth)

    def with_ns(self, ns: str) -> "Env":
        return Env(self.bindings, self.registry, self.ctx, ns, self.depth)

    @property
    def path(self) -> str:
        return self.ctx.path

    def error(self, message: str) -> ExprError:
        return ExprError(message, self.ctx.path)

    # ---- lookup -----------------------------------------------------------

    def lookup(self, path: str) -> Any:
        """Resolve a dotted name: first segment in the scope chain, then steps."""
        first, *rest = path.split(".")
        try:
            value = self.bindings[first]
        except KeyError:
            raise self.error(f"unbound name {first!r}") from None
        value = self._force(first, value)
        for step in rest:
            value = self._step(value, step, path)
        return value

    def _force(self, name: str, value: Any) -> Any:
        if not isinstance(value, Lazy):
            return value
        if value.state == "done":
            return value.value
        if value.state == "evaluating":
            raise self.error(f"cyclic definition of {name!r}")
        value.state = "evaluating"
        from .interp import evaluate

        try:
            value.value = evaluate(value.form, value.env.child(name))
        except RecursionError:
            # Belt to the depth cap's braces: whatever shape slips past
            # ``limits.depth`` still surfaces as the spec's error, not a
            # Python traceback (T051 box 4).
            raise self.error(f"evaluation of {name!r} too deep (Python recursion limit); raise nothing above limits.depth") from None
        value.state = "done"
        return value.value

    def _step(self, value: Any, step: str, path: str) -> Any:
        from ..spec_lang.safe_eval import UnsafeExpressionError, _check_attr_name

        if isinstance(value, Mapping) and step in value:
            return value[step]
        if isinstance(value, (list, tuple)) and step.isdigit():
            try:
                return value[int(step)]
            except IndexError:
                raise self.error(f"index {step} out of range in {path!r}") from None
        try:
            _check_attr_name(step)
        except UnsafeExpressionError as exc:
            raise self.error(str(exc)) from None
        if hasattr(value, step):
            return getattr(value, step)
        raise self.error(f"{path!r}: no key or attribute {step!r} on {type(value).__name__}")

    def head(self, name: str) -> Head:
        """A head by name: a locally bound head (a YAML template) shadows the registry."""
        first = name.split(".")[0]
        if first in self.bindings:
            value = self.bindings[first]
            # A binding shadows a registered head only when it IS a head — a
            # template already evaluated, or a not-yet-forced `!template` form.
            # A document key that merely shares a head's name (`world: !world
            # {...}`) must not be forced here: that would evaluate the very
            # node being evaluated (a false "cyclic definition").
            if isinstance(value, Lazy):
                if value.state == "done" and isinstance(value.value, Head):
                    return value.value
                if value.state == "done" and callable(value.value):
                    return _callable_head(first, value.value)
                if value.state == "pending":
                    from .form import Call as _Call, is_form as _is_form

                    if isinstance(value.form, _Call) and value.form.head == "template":
                        forced = self._force(first, value)
                        if isinstance(forced, Head):
                            return forced
                    # a resolved !py object / a .py include's function: plain
                    # data that happens to be callable — never a form, so
                    # forcing it evaluates nothing.
                    elif not _is_form(value.form) and not isinstance(value.form, (dict, list)) and callable(value.form):
                        return _callable_head(first, self._force(first, value))
            elif isinstance(value, Head):
                return value
            elif callable(value):
                return _callable_head(first, value)
            elif isinstance(value, Head):
                return value
        try:
            return self.registry.get(name)
        except KeyError:
            raise self.error(f"unknown head {name!r}") from None

    # ---- files ------------------------------------------------------------

    def load(self, source: Union[str, Path], *, text: Optional[str] = None, base: Optional[Union[str, Path]] = None) -> "Env":
        """Load a YAML document into a child scope: every top-level key becomes
        a (lazy) binding. Includes and ``!py`` references are resolved first
        (:mod:`alienbio.expr.include`) relative to ``base`` — the file's own
        directory by default — under this env's trust. Returns the env whose
        scope holds the bindings."""
        from .include import hydrate, include_bindings
        from .yaml_tags import load_text

        src = Path(source)
        if text is None:
            text = src.read_text()
        base_dir = Path(base).resolve() if base is not None else (src.parent.resolve() if src.is_file() else Path.cwd())
        seen = frozenset({src.resolve()}) if src.is_file() else frozenset()
        data = load_text(text)
        if not isinstance(data, Mapping):
            raise ExprError("a spec file must be a mapping at the top level", str(source))
        data = dict(data)
        entries = data.pop("_includes_", None)
        included: dict[str, Any] = {}
        modules: dict[str, Mapping[str, Any]] = {}
        if entries is not None:
            included, modules = include_bindings(entries, base_dir, trusted=self.ctx.trusted, seen=seen)
        data = hydrate(data, base=base_dir, trusted=self.ctx.trusted, seen=seen, modules=modules)
        for key, form in included.items():
            data.setdefault(str(key), form)
        env = self.scope({}, parent=self.bindings)
        for key, form in data.items():
            env.bindings[str(key)] = Lazy(form, env)
        return env

    def hydrate(self, data: Any, *, base: Optional[Union[str, Path]] = None) -> Any:
        """Resolve includes / ``!py`` references inside already-loaded forms."""
        from .include import hydrate

        return hydrate(data, base=Path(base).resolve() if base is not None else Path.cwd(), trusted=self.ctx.trusted)

    # ---- pools ------------------------------------------------------------

    def pool(self, name: Any) -> str:
        """A pool name as seen from this scope (M47.5): inside a template
        instance a name is namespaced by the instance (``krel.ME1``) unless it
        arrived as an argument, in which case it is whatever the caller's
        scope called it — that is how a parent wires its children. Outside any
        template the name is itself."""
        text = str(name)
        passed = self.bindings.get(PASSED_KEY)
        if isinstance(passed, Mapping) and text in passed:
            return str(passed[text])
        instance = self.bindings.get(INSTANCE_KEY)
        return f"{instance}.{text}" if instance else text

    def force_all(self) -> dict[str, Any]:
        """Evaluate every lazy binding in this scope (not parents); returns them."""
        out: dict[str, Any] = {}
        for key in list(self.bindings.local_keys()):
            out[key] = self._force(key, self.bindings[key])
        return out

standard(seed=0, *, trusted=False, registry=None, limits=None, bindings=None) classmethod

An environment over the default registry with root seed seed.

Source code in src/alienbio/expr/env.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def standard(
    cls,
    seed: Union[int, Seed] = 0,
    *,
    trusted: bool = False,
    registry: Optional[Registry] = None,
    limits: Optional[Limits] = None,
    bindings: Optional[Mapping[str, Any]] = None,
) -> "Env":
    """An environment over the default registry with root seed ``seed``."""
    from . import heads as _heads  # noqa: F401  (registers the builtin heads)
    from ..suite import expr_heads as _suite_heads  # noqa: F401  (layers 0-2: blocks, worlds)
    from ..suite import expr_experiment as _suite_experiment  # noqa: F401  (layers 3-6: tasks, experiments, agents)

    root_seed = seed if isinstance(seed, Seed) else Seed(int(seed))
    ctx = Ctx(seed=root_seed, trusted=trusted, limits=limits or Limits())
    return cls(Scope(dict(bindings or {}), name="root"), registry or _default_registry, ctx)

child(label)

The env for a named sub-node: same scope, child seed, extended path.

Source code in src/alienbio/expr/env.py 
150
151
152
def child(self, label: str) -> "Env":
    """The env for a named sub-node: same scope, child seed, extended path."""
    return Env(self.bindings, self.registry, self.ctx.child(str(label)), self.ns, self.depth + 1)

bind(**values)

A child scope holding values (already evaluated).

Source code in src/alienbio/expr/env.py 
154
155
156
def bind(self, **values: Any) -> "Env":
    """A child scope holding ``values`` (already evaluated)."""
    return Env(self.bindings.child(dict(values)), self.registry, self.ctx, self.ns, self.depth)

scope(data=None, parent=None)

A child scope over parent (default: this env's bindings).

Source code in src/alienbio/expr/env.py 
158
159
160
161
def scope(self, data: Optional[Mapping[str, Any]] = None, parent: Optional[Scope] = None) -> "Env":
    """A child scope over ``parent`` (default: this env's bindings)."""
    base = parent if parent is not None else self.bindings
    return Env(base.child(dict(data or {})), self.registry, self.ctx, self.ns, self.depth)

lookup(path)

Resolve a dotted name: first segment in the scope chain, then steps.

Source code in src/alienbio/expr/env.py 
181
182
183
184
185
186
187
188
189
190
191
def lookup(self, path: str) -> Any:
    """Resolve a dotted name: first segment in the scope chain, then steps."""
    first, *rest = path.split(".")
    try:
        value = self.bindings[first]
    except KeyError:
        raise self.error(f"unbound name {first!r}") from None
    value = self._force(first, value)
    for step in rest:
        value = self._step(value, step, path)
    return value

head(name)

A head by name: a locally bound head (a YAML template) shadows the registry.

Source code in src/alienbio/expr/env.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def head(self, name: str) -> Head:
    """A head by name: a locally bound head (a YAML template) shadows the registry."""
    first = name.split(".")[0]
    if first in self.bindings:
        value = self.bindings[first]
        # A binding shadows a registered head only when it IS a head — a
        # template already evaluated, or a not-yet-forced `!template` form.
        # A document key that merely shares a head's name (`world: !world
        # {...}`) must not be forced here: that would evaluate the very
        # node being evaluated (a false "cyclic definition").
        if isinstance(value, Lazy):
            if value.state == "done" and isinstance(value.value, Head):
                return value.value
            if value.state == "done" and callable(value.value):
                return _callable_head(first, value.value)
            if value.state == "pending":
                from .form import Call as _Call, is_form as _is_form

                if isinstance(value.form, _Call) and value.form.head == "template":
                    forced = self._force(first, value)
                    if isinstance(forced, Head):
                        return forced
                # a resolved !py object / a .py include's function: plain
                # data that happens to be callable — never a form, so
                # forcing it evaluates nothing.
                elif not _is_form(value.form) and not isinstance(value.form, (dict, list)) and callable(value.form):
                    return _callable_head(first, self._force(first, value))
        elif isinstance(value, Head):
            return value
        elif callable(value):
            return _callable_head(first, value)
        elif isinstance(value, Head):
            return value
    try:
        return self.registry.get(name)
    except KeyError:
        raise self.error(f"unknown head {name!r}") from None

load(source, *, text=None, base=None)

Load a YAML document into a child scope: every top-level key becomes a (lazy) binding. Includes and !py references are resolved first (:mod:alienbio.expr.include) relative to base — the file's own directory by default — under this env's trust. Returns the env whose scope holds the bindings.

Source code in src/alienbio/expr/env.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def load(self, source: Union[str, Path], *, text: Optional[str] = None, base: Optional[Union[str, Path]] = None) -> "Env":
    """Load a YAML document into a child scope: every top-level key becomes
    a (lazy) binding. Includes and ``!py`` references are resolved first
    (:mod:`alienbio.expr.include`) relative to ``base`` — the file's own
    directory by default — under this env's trust. Returns the env whose
    scope holds the bindings."""
    from .include import hydrate, include_bindings
    from .yaml_tags import load_text

    src = Path(source)
    if text is None:
        text = src.read_text()
    base_dir = Path(base).resolve() if base is not None else (src.parent.resolve() if src.is_file() else Path.cwd())
    seen = frozenset({src.resolve()}) if src.is_file() else frozenset()
    data = load_text(text)
    if not isinstance(data, Mapping):
        raise ExprError("a spec file must be a mapping at the top level", str(source))
    data = dict(data)
    entries = data.pop("_includes_", None)
    included: dict[str, Any] = {}
    modules: dict[str, Mapping[str, Any]] = {}
    if entries is not None:
        included, modules = include_bindings(entries, base_dir, trusted=self.ctx.trusted, seen=seen)
    data = hydrate(data, base=base_dir, trusted=self.ctx.trusted, seen=seen, modules=modules)
    for key, form in included.items():
        data.setdefault(str(key), form)
    env = self.scope({}, parent=self.bindings)
    for key, form in data.items():
        env.bindings[str(key)] = Lazy(form, env)
    return env

hydrate(data, *, base=None)

Resolve includes / !py references inside already-loaded forms.

Source code in src/alienbio/expr/env.py 
302
303
304
305
306
def hydrate(self, data: Any, *, base: Optional[Union[str, Path]] = None) -> Any:
    """Resolve includes / ``!py`` references inside already-loaded forms."""
    from .include import hydrate

    return hydrate(data, base=Path(base).resolve() if base is not None else Path.cwd(), trusted=self.ctx.trusted)

pool(name)

A pool name as seen from this scope (M47.5): inside a template instance a name is namespaced by the instance (krel.ME1) unless it arrived as an argument, in which case it is whatever the caller's scope called it — that is how a parent wires its children. Outside any template the name is itself.

Source code in src/alienbio/expr/env.py 
310
311
312
313
314
315
316
317
318
319
320
321
def pool(self, name: Any) -> str:
    """A pool name as seen from this scope (M47.5): inside a template
    instance a name is namespaced by the instance (``krel.ME1``) unless it
    arrived as an argument, in which case it is whatever the caller's
    scope called it — that is how a parent wires its children. Outside any
    template the name is itself."""
    text = str(name)
    passed = self.bindings.get(PASSED_KEY)
    if isinstance(passed, Mapping) and text in passed:
        return str(passed[text])
    instance = self.bindings.get(INSTANCE_KEY)
    return f"{instance}.{text}" if instance else text

force_all()

Evaluate every lazy binding in this scope (not parents); returns them.

Source code in src/alienbio/expr/env.py 
323
324
325
326
327
328
def force_all(self) -> dict[str, Any]:
    """Evaluate every lazy binding in this scope (not parents); returns them."""
    out: dict[str, Any] = {}
    for key in list(self.bindings.local_keys()):
        out[key] = self._force(key, self.bindings[key])
    return out

ExprError

Bases: ValueError

Every Expr failure, with the path of the node that failed. A ValueError: a document that fails to evaluate is an invalid value.

Source code in src/alienbio/expr/env.py 
24
25
26
27
28
29
30
31
class ExprError(ValueError):
    """Every Expr failure, with the path of the node that failed. A
    ``ValueError``: a document that fails to evaluate is an invalid value."""

    def __init__(self, message: str, path: str = "") -> None:
        self.path = path
        self.message = message
        super().__init__(f"{path}: {message}" if path else message)

Limits dataclass

Caps the interpreter enforces — exceeding one is an error, never a truncation.

Source code in src/alienbio/expr/env.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@dataclass(frozen=True)
class Limits:
    """Caps the interpreter enforces — exceeding one is an error, never a truncation."""

    entities: int = 1_000_000  # elements one `each`/`range`/`list` may produce, and in total per session
    depth: int = 100  # evaluation nesting (T051 box 4: one template level is ~5.5 Python frames, so 200 sat above the interpreter's own limit and could never fire)
    attempts: int = 8  # guard retries (M47.5)
    meter: Meter = field(default_factory=Meter, compare=False, repr=False)

    def charge(self, n: int, path: str, what: str) -> None:
        """Count ``n`` elements against the session total; raise past the cap."""
        if n > self.entities:
            raise ExprError(f"{what}: {n} elements exceeds limits.entities={self.entities}", path)
        self.meter.allocated += n
        if self.meter.allocated > self.entities:
            raise ExprError(
                f"{what}: {self.meter.allocated} elements allocated in this evaluation exceeds limits.entities={self.entities}",
                path,
            )

charge(n, path, what)

Count n elements against the session total; raise past the cap.

Source code in src/alienbio/expr/env.py 
56
57
58
59
60
61
62
63
64
65
def charge(self, n: int, path: str, what: str) -> None:
    """Count ``n`` elements against the session total; raise past the cap."""
    if n > self.entities:
        raise ExprError(f"{what}: {n} elements exceeds limits.entities={self.entities}", path)
    self.meter.allocated += n
    if self.meter.allocated > self.entities:
        raise ExprError(
            f"{what}: {self.meter.allocated} elements allocated in this evaluation exceeds limits.entities={self.entities}",
            path,
        )

UnsafeSpecError

Bases: Exception

Raised when an untrusted spec asks for code execution (a .py include, !py) or a file outside its own directory.

Source code in src/alienbio/expr/include.py 
29
30
31
class UnsafeSpecError(Exception):
    """Raised when an untrusted spec asks for code execution (a ``.py``
    include, ``!py``) or a file outside its own directory."""

Call dataclass

head(*args, **kwargs) as data. head names a registered function, expander, template or special form; the arguments are forms.

Source code in src/alienbio/expr/form.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclass(frozen=True)
class Call:
    """``head(*args, **kwargs)`` as data. ``head`` names a registered function,
    expander, template or special form; the arguments are forms."""

    head: str
    args: tuple[Any, ...] = ()
    kwargs: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "args", tuple(self.args))
        object.__setattr__(self, "kwargs", dict(self.kwargs))

    def __repr__(self) -> str:
        parts = [repr(a) for a in self.args] + [f"{k}={v!r}" for k, v in self.kwargs.items()]
        return f"{self.head}({', '.join(parts)})"

Name dataclass

A lookup: path is ident(.ident)* — the first segment resolves in the scope chain, each further segment steps into the value (mapping key, then attribute, then sequence index).

Source code in src/alienbio/expr/form.py 
20
21
22
23
24
25
26
27
28
29
@dataclass(frozen=True)
class Name:
    """A lookup: ``path`` is ``ident(.ident)*`` — the first segment resolves in
    the scope chain, each further segment steps into the value (mapping key,
    then attribute, then sequence index)."""

    path: str

    def __repr__(self) -> str:
        return f"Name({self.path!r})"

Quoted dataclass

A form held as a value. Evaluating a Quoted yields a :class:~alienbio.expr.interp.QuotedForm — the form closed over the environment it was written in, sampleable as a Dist.

Source code in src/alienbio/expr/form.py 
50
51
52
53
54
55
56
57
58
59
@dataclass(frozen=True)
class Quoted:
    """A form held as a value. Evaluating a ``Quoted`` yields a
    :class:`~alienbio.expr.interp.QuotedForm` — the form closed over the
    environment it was written in, sampleable as a ``Dist``."""

    form: Any

    def __repr__(self) -> str:
        return f"Quoted({self.form!r})"

QuotedForm dataclass

A quoted form as a value: the form plus the environment it was written in. It is a suite.dist.Dist (sample(seed) evaluates the form under that seed) and exposes run for evaluation with extra bindings.

Source code in src/alienbio/expr/interp.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@dataclass(frozen=True, eq=False)
class QuotedForm:
    """A quoted form as a value: the form plus the environment it was written
    in. It *is* a ``suite.dist.Dist`` (``sample(seed)`` evaluates the form
    under that seed) and exposes ``run`` for evaluation with extra bindings."""

    form: Any
    env: Env

    def sample(self, seed: Seed) -> Any:
        return evaluate(self.form, self.env.with_seed(seed))

    def run(self, bindings: Optional[Mapping[str, Any]] = None, *, seed: Optional[Seed] = None) -> Any:
        env = self.env if seed is None else self.env.with_seed(seed)
        if bindings:
            env = env.bind(**dict(bindings))
        return evaluate(self.form, env)

    def __eq__(self, other: object) -> bool:
        return isinstance(other, QuotedForm) and other.form == self.form

    def __hash__(self) -> int:
        return hash(repr(self.form))

    def __repr__(self) -> str:
        return f"QuotedForm({self.form!r})"

TemplateHead

Bases: Head

A head made by the template special form: positional parameter names, keyword parameters with default forms, a body, and the defining environment (closure). Calling it evaluates the body in a child scope of the definition scope, under the call's seed and namespace.

Source code in src/alienbio/expr/interp.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class TemplateHead(Head):
    """A head made by the ``template`` special form: positional parameter
    names, keyword parameters with default forms, a body, and the defining
    environment (closure). Calling it evaluates the body in a child scope of
    the definition scope, under the *call's* seed and namespace."""

    def __init__(
        self,
        name: str,
        positional: Sequence[str],
        params: Mapping[str, Any],
        body: Any,
        env: Env,
        pools: Optional[Sequence[str]] = None,
    ) -> None:
        super().__init__(name=name, kind="template", fn=self.expand, meta={"summary": f"template {name}"})
        self.positional = tuple(positional)
        self.params = dict(params)
        self.body = body
        self.env_def = env
        #: T054 #5 — which parameters carry POOL names. When declared, only
        #: their strings keep the caller's spelling inside the body; when
        #: absent (every template written before this), every string argument
        #: does, and an argument equal to an internal pool name de-namespaces it.
        self.pools = tuple(pools) if pools is not None else None

    def expand(self, args: Sequence[Any], kwargs: Mapping[str, Any], env: Env) -> Any:
        if len(args) > len(self.positional):
            raise env.error(
                f"template {self.name!r} takes {len(self.positional)} positional argument(s), got {len(args)}"
            )
        bound: dict[str, Any] = {}
        for i, pname in enumerate(self.positional):
            if i < len(args):
                bound[pname] = evaluate(args[i], env.child(pname))
            elif pname in kwargs:
                bound[pname] = evaluate(kwargs[pname], env.child(pname))
            else:
                raise env.error(f"template {self.name!r}: missing positional argument {pname!r}")
        for key, form in kwargs.items():
            if key in self.positional:
                continue
            if key not in self.params:
                raise env.error(f"template {self.name!r} has no parameter {key!r}")
            bound[key] = evaluate(form, env.child(key))
        # Defaults evaluate per call, in the call's seed, in the definition scope
        # extended by what is bound so far (a default may use an earlier parameter).
        # T051 box 4 — a template level counts toward ``limits.depth`` (a
        # body that IS the recursive call never went through ``Env.child``,
        # so self-recursion ended in Python's RecursionError, not ExprError).
        scope_env = Env(self.env_def.bindings.child(bound), env.registry, env.ctx, env.path or env.ns, env.depth + 1)
        for key, default in self.params.items():
            if key not in bound:
                bound[key] = evaluate(default, scope_env.child(key))
                scope_env.bindings[key] = bound[key]
        # M47.5 — pool names that arrived as arguments keep the caller's
        # spelling; everything else the body names is namespaced by this
        # instance (``Env.pool``).
        passed: dict[str, str] = {}
        pool_params = bound.keys() if self.pools is None else [k for k in bound if k in self.pools]
        for key in pool_params:
            for text in _strings_in(bound[key]):
                passed.setdefault(text, env.pool(text))
        scope_env.bindings[PASSED_KEY] = passed
        # The instance is named by the key the call is bound to, nested under
        # the enclosing instance (``krel``; ``c1.krel`` inside instance ``c1``).
        label = (env.path.rsplit(".", 1)[-1] if env.path else "") or self.name
        parent = env.bindings.get(INSTANCE_KEY)
        instance = f"{parent}.{label}" if parent else label
        # T054 #5: two calls with the same key under different parents
        # (``a.cell`` / ``b.cell``) used to share one namespace and merge their
        # molecules silently. The instance name is claimed per document path;
        # a second path claiming it refuses (a retry at the same path does not).
        prior = env.ctx.instances.setdefault(instance, env.path)
        if prior != env.path:
            raise env.error(
                f"template {self.name!r}: instance name {instance!r} is already taken by the call at "
                f"{prior!r}; two instances would share every pool — bind them to distinct keys"
            )
        scope_env.bindings[INSTANCE_KEY] = instance
        return evaluate(self.body, scope_env)

GuardViolation

Bases: Exception

Raised by a guard to reject what a call produced. offenders names the elements (keys of a produced mapping, dotted for depth) that on_fail: prune may drop; a violation without them cannot be pruned.

Source code in src/alienbio/expr/registry.py 
18
19
20
21
22
23
24
25
26
class GuardViolation(Exception):
    """Raised by a guard to reject what a call produced. ``offenders`` names
    the elements (keys of a produced mapping, dotted for depth) that
    ``on_fail: prune`` may drop; a violation without them cannot be pruned."""

    def __init__(self, message: str = "guard failed", *, offenders: Collection[str] = ()) -> None:
        super().__init__(message)
        self.message = message
        self.offenders = tuple(offenders)

Registry

A name → :class:Head table with kind-filtered views.

Source code in src/alienbio/expr/registry.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class Registry:
    """A name → :class:`Head` table with kind-filtered views."""

    def __init__(self, heads: Optional[dict[str, Head]] = None, *, kinds: Optional[Collection[str]] = None) -> None:
        self._heads: dict[str, Head] = heads if heads is not None else {}
        self._kinds: Optional[frozenset[str]] = frozenset(kinds) if kinds is not None else None

    def register(self, head: Head, *, replace: bool = False) -> Head:
        if self._kinds is not None:
            raise ValueError("cannot register into a registry view")
        if head.name in self._heads and not replace:
            existing = self._heads[head.name]
            if existing.fn is not head.fn and not _same_definition(existing.fn, head.fn):
                raise ValueError(
                    f"head {head.name!r} is already registered ({existing.kind}, "
                    f"{_where(existing.fn)}); pass replace=True to shadow it on purpose"
                )
        self._heads[head.name] = head
        return head

    def get(self, name: str) -> Head:
        head = self._heads.get(name)
        if head is None or (self._kinds is not None and head.kind not in self._kinds and not head.is_special):
            raise KeyError(name)
        return head

    def __contains__(self, name: object) -> bool:
        try:
            self.get(str(name))
            return True
        except KeyError:
            return False

    def names(self) -> list[str]:
        return sorted(n for n in self._heads if n in self)

    def view(self, kinds: Iterable[str]) -> "Registry":
        """A registry showing only ``kinds`` (special forms always show)."""
        return Registry(self._heads, kinds=set(kinds))

    def describe(self) -> list[dict[str, Any]]:
        out = []
        for name in self.names():
            h = self._heads[name]
            try:
                sig = str(inspect.signature(h.fn))
            except (TypeError, ValueError):
                sig = "(...)"
            out.append({"name": name, "kind": h.kind, "signature": sig, "summary": h.meta.get("summary", "")})
        return out

view(kinds)

A registry showing only kinds (special forms always show).

Source code in src/alienbio/expr/registry.py 
115
116
117
def view(self, kinds: Iterable[str]) -> "Registry":
    """A registry showing only ``kinds`` (special forms always show)."""
    return Registry(self._heads, kinds=set(kinds))

ExprLoader

Bases: SafeLoader

yaml.SafeLoader plus the Expr tags. A subclass, so the global SafeLoader (and every other loader in the process) is untouched.

Source code in src/alienbio/expr/yaml_tags.py 
32
33
34
class ExprLoader(yaml.SafeLoader):
    """``yaml.SafeLoader`` plus the Expr tags. A subclass, so the global
    SafeLoader (and every other loader in the process) is untouched."""

contains_form(value)

True if value is, or contains anywhere inside data, a tagged form.

Source code in src/alienbio/expr/form.py 
95
96
97
def contains_form(value: Any) -> bool:
    """True if ``value`` is, or contains anywhere inside data, a tagged form."""
    return any(True for _ in walk(value) if is_form(_))

is_form(value)

True for the three tagged shapes — a literal or data is "a form" too, but only these three carry meaning beyond their Python value.

Source code in src/alienbio/expr/form.py 
89
90
91
92
def is_form(value: Any) -> bool:
    """True for the three tagged shapes — a literal or data is "a form" too, but
    only these three carry meaning beyond their Python value."""
    return isinstance(value, (Name, Call, Quoted))

walk(value)

Pre-order walk over a form tree, yielding every node (data included).

Source code in src/alienbio/expr/form.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def walk(value: Any) -> Iterator[Any]:
    """Pre-order walk over a form tree, yielding every node (data included)."""
    yield value
    if isinstance(value, Call):
        for a in value.args:
            yield from walk(a)
        for v in value.kwargs.values():
            yield from walk(v)
    elif isinstance(value, Quoted):
        yield from walk(value.form)
    elif isinstance(value, dict):
        for v in value.values():
            yield from walk(v)
    elif isinstance(value, (list, tuple)):
        for v in value:
            yield from walk(v)

evaluate(form, env)

Evaluate form in env.

Source code in src/alienbio/expr/interp.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def evaluate(form: Any, env: Env) -> Any:
    """Evaluate ``form`` in ``env``."""
    if env.depth > env.ctx.limits.depth:
        raise env.error(f"evaluation deeper than limits.depth={env.ctx.limits.depth}")
    if isinstance(form, Name):
        return env.lookup(form.path)
    if isinstance(form, Quoted):
        return QuotedForm(form.form, env)
    if isinstance(form, Call):
        return _call(form, env)
    if isinstance(form, Lazy):
        return env._force("<lazy>", form)
    if isinstance(form, (Include, PyRef)):
        raise env.error(f"{form!r} was not resolved at load — includes are resolved by Env.load / Env.hydrate")
    if isinstance(form, dict):
        if "_type" in form:
            # M47.6 — ``{_type: Reaction, ...}`` is the untagged spelling of
            # ``!Reaction {...}`` (kept for saved worlds).
            head = form["_type"]
            if not isinstance(head, str):
                raise env.error(f"_type must name a head, got {head!r}")
            return _call(Call(head, (), {k: v for k, v in form.items() if k != "_type"}), env)
        return {k: evaluate(v, env.child(str(k))) for k, v in form.items()}
    if isinstance(form, (list, tuple)):
        return [evaluate(v, env.child(str(i))) for i, v in enumerate(form)]
    return form

expander(_f=None, *, name=None, guarded=False, guarded_params=(), into=registry, replace=False, **meta)

Register an expander head: fn(args, kwargs, env) receives the argument forms (unevaluated) and returns a form the interpreter then evaluates under the call's seed.

Source code in src/alienbio/expr/registry.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def expander(
    _f: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
    guarded: bool = False,
    guarded_params: Collection[str] = (),
    into: Registry = registry,
    replace: bool = False,
    **meta: Any,
) -> Any:
    """Register an expander head: ``fn(args, kwargs, env)`` receives the
    argument **forms** (unevaluated) and returns a form the interpreter then
    evaluates under the call's seed."""
    return _decorate("expander", _f, name=name, guarded=guarded, guarded_params=guarded_params, into=into, meta=meta, replace=replace)

fn(_f=None, *, name=None, kind='fn', guarded=False, guarded_params=(), into=registry, replace=False, **meta)

Register a function head: its arguments arrive evaluated. A keyword-only ctx / env parameter is injected, never passed by the spec. kind is the flavor tag (dist, rate, scoring, ...).

Source code in src/alienbio/expr/registry.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def fn(
    _f: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
    kind: str = "fn",
    guarded: bool = False,
    guarded_params: Collection[str] = (),
    into: Registry = registry,
    replace: bool = False,
    **meta: Any,
) -> Any:
    """Register a function head: its arguments arrive **evaluated**. A
    keyword-only ``ctx`` / ``env`` parameter is injected, never passed by the
    spec. ``kind`` is the flavor tag (``dist``, ``rate``, ``scoring``, ...)."""
    if kind not in FUNCTION_KINDS:
        raise ValueError(f"@fn: unknown kind {kind!r}; expected one of {sorted(FUNCTION_KINDS)}")
    return _decorate(kind, _f, name=name, guarded=guarded, guarded_params=guarded_params, into=into, meta=meta, replace=replace)

guard(_f=None, *, name=None, into=registry, replace=False, **meta)

Register a guard: fn(value, ctx, **params) returns False or raises :class:GuardViolation to reject what a call produced. A call's guards: [...] lists guards by name (defaults) or as calls (parameters); on_fail: retry | prune | reject decides what a failure does (M47.5).

Source code in src/alienbio/expr/registry.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
def guard(
    _f: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
    into: Registry = registry,
    replace: bool = False,
    **meta: Any,
) -> Any:
    """Register a guard: ``fn(value, ctx, **params)`` returns ``False`` or
    raises :class:`GuardViolation` to reject what a call produced. A call's
    ``guards: [...]`` lists guards by name (defaults) or as calls (parameters);
    ``on_fail: retry | prune | reject`` decides what a failure does (M47.5)."""
    return _decorate("guard", _f, name=name, guarded=False, guarded_params=(), into=into, meta=meta, replace=replace)

dump_structural(form)

Form -> YAML text in the structural spelling (loads back to an equal form).

Source code in src/alienbio/expr/yaml_tags.py 
170
171
172
173
174
def dump_structural(form: Any) -> str:
    """Form -> YAML text in the structural spelling (loads back to an equal form)."""
    buf = io.StringIO()
    yaml.dump(form, buf, Dumper=ExprDumper, sort_keys=False, default_flow_style=None, allow_unicode=True)
    return buf.getvalue()

load_text(text)

YAML text -> forms (data with Name / Call / Quoted where tagged).

Source code in src/alienbio/expr/yaml_tags.py 
124
125
126
127
128
129
def load_text(text: str) -> Any:
    """YAML text -> forms (data with Name / Call / Quoted where tagged)."""
    try:
        return yaml.load(text, Loader=ExprLoader)  # noqa: S506 - ExprLoader is a SafeLoader
    except yaml.YAMLError as exc:
        raise ExprError(f"YAML error: {exc}") from None