dizzy.engine

dizzy.engine

The DIZZY runtime kit's engine layer — how a declared feature gets RUN.

dizzy generate turns a .feat.yaml into schemas, contracts and stubs. This package is the other half: the machinery that schedules those elements at runtime, and — crucially — does so without knowing what they are.

Five pieces here, plus a scheduling shell per execution model:

• loop — Engine: the control loop. Command -> procedure -> events -> projections -> policies -> commands, with the read-model commit boundary between the fold and the dispatch. It is keyed by generated class, so it names nothing.
• store — EventStore over dagstore, a content-addressed event DAG: the truth an engine appends to before anything else runs.
• rebuild / replicate — the two things you can do with a stream besides run it forward: refold it into the read models (the recoverability test), and pull a peer's facts and fold those through the same projections. Both take the projection runners as an argument, so neither knows a feature.
• registry — FeatGraph: the feat file read into an app's topology, with every declared command and event resolved to its generated pydantic class by DIZZY's naming convention. This is what makes a shell generic. The feat already declares everything; a shell that hard-codes any of it has copied the design out of the artifact, which is the one thing DIZZY exists to prevent.
• ports — HostApp / ShellServices / Runtime and the CommandQueue / TelemetryBus protocols: the seam through which everything app-specific reaches a shell. An app publishes one HostApp; a shell resolves it from $DIZZY_HOST_APP and needs nothing else.

Scheduling shells (installed via extras, so a host pays only for the one it runs):

• dizzy.engine.st — single process: a durable sqlite command queue with atomic claim + lanes, and an in-process telemetry ring. Stdlib only.
• dizzy.engine.mp — a fleet: Dramatiq/Redis workers, pool-routed, with telemetry over Redis pub/sub. Needs dizzy[mp].

The shells differ ONLY in scheduling — who holds the command queue, who runs the workers, where telemetry lands. The engine they drive and the wiring that binds a feature to it are shared, and both shells execute them verbatim.

They do NOT, however, claim the same semantics, and that difference is deliberate rather than incidental. Because the engine hands every policy-dispatched command to the shell's queue, the shell owns the command phase — so it is the shell, not the engine, that decides how many commands run at once. st drains one lane in one process and is sequentially consistent: one legal interleaving, which is DIZZY's defined semantics. mp runs N workers under at-least-once delivery and is knowingly weaker, relying on confluent projections to absorb the reordering and the duplicates. A host picks the shell whose guarantee it needs.

The engine reads and writes read models only through the runners the wiring registers, so it carries no ORM: dizzy.engine costs pyyaml and pydantic and nothing else. The mp shell's broker dependencies stay behind its extra, so importing dizzy.engine never drags in a broker.

Engine

One process's control loop over a registered topology.

Source code in dizzy/src/dizzy/engine/loop.py

class Engine:
    """One process's control loop over a registered topology."""

    def __init__(
        self,
        command_queue: Any,
        store: Any,
        observer: EventObserver | None = None,
        bus: BusEmit | None = None,
        commit: Callable[[], None] | None = None,
        otel: Any = None,
    ):
        self.command_queue = command_queue
        """External (shell-owned) command queue — where dispatches go."""
        self.store = store
        """Append-only event store: the truth."""
        self.observer = observer
        self.bus = bus
        self.commit = commit
        """Called once per event, after its projections fold and before its
        policies dispatch. ``None`` means the app has no read-model
        transaction to close (see the ordering rule above)."""
        self.otel = otel if otel is not None else NullOtel()
        self._events: deque = deque()
        self.current_event = None
        """The event whose policies are currently running, for wiring that
        needs to correlate a dispatch back to its cause."""
        self._procedures: dict[type, tuple[str, CommandRunner]] = {}
        self._projections: dict[type, list[tuple[str, ProjectionRunner]]] = defaultdict(list)
        self._policies: dict[type, list[tuple[str, PolicyRunner]]] = defaultdict(list)
        self._duration_hist: Any = None

    # ── Registration (the wiring calls these; mirrors the feat topology) ─────

    def register_procedure(
        self, command_type: type, runner: CommandRunner, name: str | None = None
    ) -> None:
        self._procedures[command_type] = (name or command_type.__name__, runner)

    def register_projection(
        self, event_type: type, runner: ProjectionRunner, name: str | None = None
    ) -> None:
        self._projections[event_type].append((name or event_type.__name__, runner))

    def register_policy(
        self, event_type: type, runner: PolicyRunner, name: str | None = None
    ) -> None:
        self._policies[event_type].append((name or event_type.__name__, runner))

    def projection_runners(self) -> dict[type, list[tuple[str, ProjectionRunner]]]:
        """The event class -> ``[(name, runner)]`` map, as registered.

        This is exactly what :func:`dizzy.engine.rebuild.rebuild` and
        :func:`dizzy.engine.replicate.fold_envelopes` take, so a host that has
        already wired an engine does not wire the data loop a second time to
        refold or to replicate — one registration, three triggers.
        """
        return dict(self._projections)

    # ── Emit closures handed to elements ────────────────────────────────────

    def emit_event(self, event: Any) -> None:
        """A procedure emitted an event -> onto the local event queue."""
        self._events.append(event)

    def dispatch_command(self, command: Any) -> None:
        """A policy dispatched a command -> onto the external queue."""
        self.command_queue.put(command)

    def discard_pending_events(self) -> None:
        """Drop events emitted but not yet drained.

        A shell calls this after a failed command so un-appended events cannot
        leak into the NEXT command on a long-lived process.
        """
        self._events.clear()

    # ── Self-observation ────────────────────────────────────────────────────

    def _bus_emit(self, record: dict) -> None:
        """Emit a bus record, stamped with the active trace id (if any) so a
        log line can be linked back to its trace."""
        if self.bus is None:
            return
        trace_id = self.otel.trace_id_hex()
        if trace_id:
            record["trace_id"] = trace_id
        self.bus(record)

    def _record_duration(self, kind: str, name: str, outcome: str, ms: float) -> None:
        """Histogram of element run durations (no-op without an OTel meter)."""
        if self._duration_hist is None:
            meter = getattr(self.otel, "meter", None)
            if meter is None:
                return
            self._duration_hist = meter().create_histogram(
                "dizzy.element.duration",
                unit="ms",
                description="element (procedure/projection/policy) run duration",
            )
        self._duration_hist.record(
            ms, {"dizzy.kind": kind, "dizzy.name": name, "dizzy.outcome": outcome}
        )

    def _timed(self, kind: str, name: str, trigger: str, fn: Callable[[], None]) -> None:
        """Run one element, emitting a start-end record to the bus (if any) and
        an OTel span (a no-op under the null provider).

        Exceptions propagate after being recorded — the engine observes, it
        does not swallow.
        """
        if self.bus is None:
            return fn()
        with self.otel.tracer().start_as_current_span(
            f"{kind} {name}",
            record_exception=True,
            attributes={"dizzy.kind": kind, "dizzy.name": name, "dizzy.trigger": trigger},
        ):
            t0 = time.monotonic()
            try:
                fn()
            except Exception as exc:
                dur = round((time.monotonic() - t0) * 1000, 1)
                self._record_duration(kind, name, "error", dur)
                self._bus_emit(
                    {
                        "kind": kind,
                        "name": name,
                        "trigger": trigger,
                        "duration_ms": dur,
                        "outcome": "error",
                        "detail": f"{type(exc).__name__}: {exc}",
                    }
                )
                raise
            dur = round((time.monotonic() - t0) * 1000, 1)
            self._record_duration(kind, name, "ok", dur)
            self._bus_emit(
                {
                    "kind": kind,
                    "name": name,
                    "trigger": trigger,
                    "duration_ms": dur,
                    "outcome": "ok",
                    "detail": "",
                }
            )

    # ── The loop ────────────────────────────────────────────────────────────

    def run_command(self, command: Any) -> None:
        """Run one command to quiescence: its procedure, then every event that
        cascades from it. Returns when the local event queue is empty; any
        command a policy dispatched is by then on the external queue."""
        entry = self._procedures.get(type(command))
        if entry is None:
            raise KeyError(
                f"no procedure registered for {type(command).__name__} — the "
                f"wiring did not register it"
            )
        name, runner = entry
        self._timed("procedure", name, type(command).__name__, lambda: runner(command))
        self._drain_events()

    def _drain_events(self) -> None:
        while self._events:
            event = self._events.popleft()
            # The store is the truth: append FIRST, get the envelope.
            envelope = self.store.append(event)
            if self.bus is not None:
                self._bus_emit(
                    {
                        "kind": "event",
                        "name": envelope.type,
                        "trigger": envelope.id[:12],
                        "duration_ms": None,
                        "outcome": "appended",
                        "detail": "",
                    }
                )
            # Data loop: projections fold the event, seeing the envelope's
            # ingested_at as their second argument.
            for name, projection in self._projections.get(type(event), []):
                self._timed(
                    "projection",
                    name,
                    envelope.type,
                    # Every free name is bound as a default: the lambda runs
                    # inside this iteration, but a late-binding closure over a
                    # loop variable is a trap waiting for the first caller who
                    # defers it.
                    lambda p=projection, e=event, at=envelope.ingested_at: p(e, at),
                )
            if self.commit is not None:
                self.commit()  # the event is folded, atomically, NOW
            if self.observer is not None:
                self.observer(envelope.type, event)
            # Reactivity loop: policies dispatch commands onto the external queue.
            self.current_event = event
            try:
                for name, policy in self._policies.get(type(event), []):
                    self._timed("policy", name, envelope.type, lambda p=policy, e=event: p(e))
            finally:
                self.current_event = None

command_queue = command_queue instance-attribute

External (shell-owned) command queue — where dispatches go.

store = store instance-attribute

Append-only event store: the truth.

commit = commit instance-attribute

Called once per event, after its projections fold and before its policies dispatch. None means the app has no read-model transaction to close (see the ordering rule above).

current_event = None instance-attribute

The event whose policies are currently running, for wiring that needs to correlate a dispatch back to its cause.

projection_runners()

The event class -> [(name, runner)] map, as registered.

This is exactly what :func:dizzy.engine.rebuild.rebuild and :func:dizzy.engine.replicate.fold_envelopes take, so a host that has already wired an engine does not wire the data loop a second time to refold or to replicate — one registration, three triggers.

Source code in dizzy/src/dizzy/engine/loop.py

def projection_runners(self) -> dict[type, list[tuple[str, ProjectionRunner]]]:
    """The event class -> ``[(name, runner)]`` map, as registered.

    This is exactly what :func:`dizzy.engine.rebuild.rebuild` and
    :func:`dizzy.engine.replicate.fold_envelopes` take, so a host that has
    already wired an engine does not wire the data loop a second time to
    refold or to replicate — one registration, three triggers.
    """
    return dict(self._projections)

emit_event(event)

A procedure emitted an event -> onto the local event queue.

Source code in dizzy/src/dizzy/engine/loop.py

def emit_event(self, event: Any) -> None:
    """A procedure emitted an event -> onto the local event queue."""
    self._events.append(event)

dispatch_command(command)

A policy dispatched a command -> onto the external queue.

Source code in dizzy/src/dizzy/engine/loop.py

def dispatch_command(self, command: Any) -> None:
    """A policy dispatched a command -> onto the external queue."""
    self.command_queue.put(command)

discard_pending_events()

Drop events emitted but not yet drained.

A shell calls this after a failed command so un-appended events cannot leak into the NEXT command on a long-lived process.

Source code in dizzy/src/dizzy/engine/loop.py

def discard_pending_events(self) -> None:
    """Drop events emitted but not yet drained.

    A shell calls this after a failed command so un-appended events cannot
    leak into the NEXT command on a long-lived process.
    """
    self._events.clear()

run_command(command)

Run one command to quiescence: its procedure, then every event that cascades from it. Returns when the local event queue is empty; any command a policy dispatched is by then on the external queue.

Source code in dizzy/src/dizzy/engine/loop.py

def run_command(self, command: Any) -> None:
    """Run one command to quiescence: its procedure, then every event that
    cascades from it. Returns when the local event queue is empty; any
    command a policy dispatched is by then on the external queue."""
    entry = self._procedures.get(type(command))
    if entry is None:
        raise KeyError(
            f"no procedure registered for {type(command).__name__} — the "
            f"wiring did not register it"
        )
    name, runner = entry
    self._timed("procedure", name, type(command).__name__, lambda: runner(command))
    self._drain_events()

CommandQueue

Bases: Protocol

Where a policy's dispatch goes. The engine holds one of these.

Source code in dizzy/src/dizzy/engine/ports.py

@runtime_checkable
class CommandQueue(Protocol):
    """Where a policy's dispatch goes. The engine holds one of these."""

    def put(self, command: Any, origin: str = "policy") -> Any: ...

    def qsize(self) -> int: ...

HostApp dataclass

Everything a scheduling shell needs to run an app it knows nothing about.

Source code in dizzy/src/dizzy/engine/ports.py

@dataclass
class HostApp:
    """Everything a scheduling shell needs to run an app it knows nothing about."""

    graph: FeatGraph
    build_runtime: Callable[[ShellServices], Runtime]

    routes: Callable[[], Mapping[str, tuple[str, dict]]] = dict
    """command name -> (pool, broker message options). Unlisted commands go to
    the default pool. Where the route table COMES from — a manifest, a config
    file, a constant — is the app's business, not a shell's."""

    otel: Any = field(default_factory=NullOtel)

    origin_for: Callable[[Any, Any], str | None] = _no_origin
    """(current_event, command_being_dispatched) -> correlation string, or None
    for the default. Lets an app thread its own causality (e.g. a tool call's
    identity) through a dispatch without the shell knowing those names."""

    on_command_done: Callable[..., Any] = _no_hook
    """(origin, status, detail, emitted) after a command finishes — the app's
    place to close out whatever *origin* referred to.

    Returning TRUTHY on a failure means "handled": the app turned the failure
    into a fact, so the shell must not also let the broker retry the side
    effect. A falsy return re-raises, keeping at-least-once delivery."""

    span_attrs: Callable[[str], Mapping[str, Any]] = _no_attrs
    """origin -> extra tracing attributes. Whatever an app encodes in an
    origin string is the app's to decode, but a shell that simply dropped it
    would make traces unsearchable by the app's own identifiers — so the
    decoding gets a door rather than being deleted."""

    service_name: str = "dizzy-worker"

    @staticmethod
    def resolve(spec: str | None = None) -> HostApp:
        """Load the app manifest named by *spec* or ``$DIZZY_HOST_APP``.

        Spec form is ``module:attr``; *attr* may be a ``HostApp`` or a
        zero-argument callable returning one (the usual choice — it defers the
        app's imports to worker-boot time).
        """
        spec = spec or os.environ.get("DIZZY_HOST_APP") or ""
        if ":" not in spec:
            raise RuntimeError(
                f"$DIZZY_HOST_APP={spec!r} is not a 'module:attr' spec — set it "
                f"to a module path and the name of a HostApp (or of a callable "
                f"returning one)"
                if spec
                else "no host app: set $DIZZY_HOST_APP to 'module:attr' naming a "
                "HostApp (or a callable returning one)"
            )
        module_name, _, attr = spec.partition(":")
        # Every failure below names the variable: an operator reading a systemd
        # journal sees a bare ImportError otherwise, with nothing to act on.
        try:
            module = importlib.import_module(module_name)
        except ImportError as exc:
            raise RuntimeError(
                f"$DIZZY_HOST_APP={spec!r}: cannot import {module_name!r} "
                f"({exc}) — is it on sys.path from this process's cwd?"
            ) from exc
        try:
            obj = getattr(module, attr)
        except AttributeError as exc:
            raise RuntimeError(
                f"$DIZZY_HOST_APP={spec!r}: {module_name!r} has no {attr!r}"
            ) from exc
        try:
            app = obj() if callable(obj) and not isinstance(obj, HostApp) else obj
        except Exception as exc:
            raise RuntimeError(
                f"$DIZZY_HOST_APP={spec!r}: building the HostApp raised {type(exc).__name__}: {exc}"
            ) from exc
        if not isinstance(app, HostApp):
            raise TypeError(
                f"$DIZZY_HOST_APP={spec!r} resolved to {type(app).__name__}, not HostApp"
            )
        return app

routes = dict class-attribute instance-attribute

command name -> (pool, broker message options). Unlisted commands go to the default pool. Where the route table COMES from — a manifest, a config file, a constant — is the app's business, not a shell's.

origin_for = _no_origin class-attribute instance-attribute

(current_event, command_being_dispatched) -> correlation string, or None for the default. Lets an app thread its own causality (e.g. a tool call's identity) through a dispatch without the shell knowing those names.

on_command_done = _no_hook class-attribute instance-attribute

(origin, status, detail, emitted) after a command finishes — the app's place to close out whatever origin referred to.

Returning TRUTHY on a failure means "handled": the app turned the failure into a fact, so the shell must not also let the broker retry the side effect. A falsy return re-raises, keeping at-least-once delivery.

span_attrs = _no_attrs class-attribute instance-attribute

origin -> extra tracing attributes. Whatever an app encodes in an origin string is the app's to decode, but a shell that simply dropped it would make traces unsearchable by the app's own identifiers — so the decoding gets a door rather than being deleted.

resolve(spec=None) staticmethod

Load the app manifest named by spec or $DIZZY_HOST_APP.

Spec form is module:attr; attr may be a HostApp or a zero-argument callable returning one (the usual choice — it defers the app's imports to worker-boot time).

Source code in dizzy/src/dizzy/engine/ports.py

@staticmethod
def resolve(spec: str | None = None) -> HostApp:
    """Load the app manifest named by *spec* or ``$DIZZY_HOST_APP``.

    Spec form is ``module:attr``; *attr* may be a ``HostApp`` or a
    zero-argument callable returning one (the usual choice — it defers the
    app's imports to worker-boot time).
    """
    spec = spec or os.environ.get("DIZZY_HOST_APP") or ""
    if ":" not in spec:
        raise RuntimeError(
            f"$DIZZY_HOST_APP={spec!r} is not a 'module:attr' spec — set it "
            f"to a module path and the name of a HostApp (or of a callable "
            f"returning one)"
            if spec
            else "no host app: set $DIZZY_HOST_APP to 'module:attr' naming a "
            "HostApp (or a callable returning one)"
        )
    module_name, _, attr = spec.partition(":")
    # Every failure below names the variable: an operator reading a systemd
    # journal sees a bare ImportError otherwise, with nothing to act on.
    try:
        module = importlib.import_module(module_name)
    except ImportError as exc:
        raise RuntimeError(
            f"$DIZZY_HOST_APP={spec!r}: cannot import {module_name!r} "
            f"({exc}) — is it on sys.path from this process's cwd?"
        ) from exc
    try:
        obj = getattr(module, attr)
    except AttributeError as exc:
        raise RuntimeError(
            f"$DIZZY_HOST_APP={spec!r}: {module_name!r} has no {attr!r}"
        ) from exc
    try:
        app = obj() if callable(obj) and not isinstance(obj, HostApp) else obj
    except Exception as exc:
        raise RuntimeError(
            f"$DIZZY_HOST_APP={spec!r}: building the HostApp raised {type(exc).__name__}: {exc}"
        ) from exc
    if not isinstance(app, HostApp):
        raise TypeError(
            f"$DIZZY_HOST_APP={spec!r} resolved to {type(app).__name__}, not HostApp"
        )
    return app

NullOtel

Satisfies the tracing surface a shell uses, doing nothing.

Source code in dizzy/src/dizzy/engine/ports.py

class NullOtel:
    """Satisfies the tracing surface a shell uses, doing nothing."""

    def init(self, service_name: str) -> None:
        pass

    def tracer(self) -> Any:
        return _NullTracer()

    def meter(self) -> Any:
        return _NullMeter()

    def inject(self) -> dict:
        return {}

    def extract(self, carrier: dict | None) -> Any:
        return None

    def trace_id_hex(self) -> str:
        return ""

Runtime dataclass

One process's live engine, as the app built it.

Source code in dizzy/src/dizzy/engine/ports.py

@dataclass
class Runtime:
    """One process's live engine, as the app built it."""

    engine: Any
    session: Any = None
    """The read-model session, when there is one — the shell rolls it back
    after a failed command so a partial fold can't leak into the next."""
    refresh: Callable[[], None] = lambda: None
    """Re-hydrate mutable environment before each command (secrets can change
    under a long-lived worker). Derive the field list from
    ``graph.environment`` rather than listing it."""

session = None class-attribute instance-attribute

The read-model session, when there is one — the shell rolls it back after a failed command so a partial fold can't leak into the next.

refresh = lambda: None class-attribute instance-attribute

Re-hydrate mutable environment before each command (secrets can change under a long-lived worker). Derive the field list from graph.environment rather than listing it.

ShellServices dataclass

The shell's side of the contract, passed to build_runtime.

The app builds its Engine around these: dispatches go to command_queue, observations to publish. Telemetry sinks are the app's to construct — a sink that must cross the process boundary is just one that closes over publish, which keeps the shell ignorant of the app's payload shapes.

Source code in dizzy/src/dizzy/engine/ports.py

@dataclass
class ShellServices:
    """The shell's side of the contract, passed to ``build_runtime``.

    The app builds its Engine around these: dispatches go to *command_queue*,
    observations to *publish*. Telemetry sinks are the app's to construct —
    a sink that must cross the process boundary is just one that closes over
    *publish*, which keeps the shell ignorant of the app's payload shapes.
    """

    command_queue: Any
    publish: Publish
    pool: str = "default"

    observer: Callable[[str, Any], None] = _noop_observer
    """The SHELL's event observer, which the app MUST call from whatever
    observer it passes to its engine builder.

    The engine takes exactly one observer, so an app that installs its own
    without chaining this one silently unplugs the shell: mp collects the
    events a command emitted here, and ``on_command_done`` receives that list.
    Dropping it degrades every result to "no events emitted" rather than
    failing, which is why it is stated as a requirement and not a nicety.
    Use :func:`chain_observers` if you have nothing app-specific to add.
    """

observer = _noop_observer class-attribute instance-attribute

The SHELL's event observer, which the app MUST call from whatever observer it passes to its engine builder.

The engine takes exactly one observer, so an app that installs its own without chaining this one silently unplugs the shell: mp collects the events a command emitted here, and on_command_done receives that list. Dropping it degrades every result to "no events emitted" rather than failing, which is why it is stated as a requirement and not a nicety. Use :func:chain_observers if you have nothing app-specific to add.

TelemetryBus

Bases: Protocol

Host-level observation — never events, never load-bearing.

Source code in dizzy/src/dizzy/engine/ports.py

@runtime_checkable
class TelemetryBus(Protocol):
    """Host-level observation — never events, never load-bearing."""

    def emit(self, record: dict) -> None: ...

FeatGraph

An app's declared topology, with its generated classes resolved.

Construct with :meth:load. Cheap to hold; every resolution is cached.

Source code in dizzy/src/dizzy/engine/registry.py

class FeatGraph:
    """An app's declared topology, with its generated classes resolved.

    Construct with :meth:`load`. Cheap to hold; every resolution is cached.
    """

    def __init__(
        self, feat_path: Path, raw: dict[str, Any], def_package: str = DEFAULT_DEF_PACKAGE
    ):
        self.feat_path = feat_path
        self.raw = raw
        self.def_package = def_package

    @classmethod
    def load(
        cls, feat_path: str | Path | None = None, def_package: str = DEFAULT_DEF_PACKAGE
    ) -> FeatGraph:
        import yaml

        path = Path(feat_path) if feat_path else find_feat()
        raw = yaml.safe_load(path.read_text()) or {}
        if not isinstance(raw, dict):
            raise RuntimeError(
                f"{path.name} is not a feat file: its top level is "
                f"{type(raw).__name__}, expected a mapping of sections"
            )
        graph = cls(path, raw, def_package)
        graph._check_shape()
        return graph

    def _check_shape(self) -> None:
        """Fail at load, naming the section — not later, deep in a stdlib call.

        A section written as a list or a bare string is legal YAML and a
        common slip; left alone, a list section half-works (``names()`` looks
        right, ``entry()`` blows up with AttributeError) and a string section
        is iterated one CHARACTER at a time.
        """
        for section in SECTIONS:
            value = self.raw.get(section)
            if value is None or isinstance(value, dict):
                continue
            hint = (
                " — a list of names is not a section; each entry needs a name: description mapping"
                if isinstance(value, list)
                else ""
            )
            raise RuntimeError(
                f"{self.feat_path.name}: section {section!r} is a "
                f"{type(value).__name__}, expected a mapping{hint}"
            )
        for section in SECTIONS:
            for name in self.raw.get(section) or {}:
                check_name(name, section, self.feat_path.name)

    # ── Declared names (no imports needed) ───────────────────────────────────

    def _section(self, section: str) -> dict[str, Any]:
        if section not in SECTIONS:
            raise KeyError(
                f"{section!r} is not a feat section — expected one of {', '.join(SECTIONS)}"
            )
        return self.raw.get(section) or {}

    def names(self, section: str) -> tuple[str, ...]:
        """The names the feat declares in *section*, in feat order."""
        return tuple(self._section(section))

    def entry(self, section: str, name: str) -> dict[str, Any]:
        """One declaration, normalized to a dict.

        A bare string is a description-only entry (how the feat spells most
        commands); a NULL value is a declared-but-unwritten entry, which is
        normal while drafting — it is present, just empty.
        """
        declared = self._section(section)
        if name not in declared:
            raise KeyError(f"{section}.{name} is not declared in {self.feat_path.name}")
        value = declared[name]
        if value is None:
            return {}
        if isinstance(value, str):
            return {"description": value}
        if isinstance(value, dict):
            return dict(value)
        raise RuntimeError(
            f"{self.feat_path.name}: {section}.{name} is a "
            f"{type(value).__name__}, expected a mapping or a description string"
        )

    @property
    def environment(self) -> tuple[str, ...]:
        """Environment field names — what a shell must re-hydrate per command.

        Derived, so adding an env shape to the feat needs no shell change.
        """
        return self.names("environment")

    @property
    def telemetry(self) -> tuple[str, ...]:
        """Telemetry sink names — the ports a shell may re-route as transport."""
        return self.names("telemetry")

    # ── Resolved classes (lazy per section) ──────────────────────────────────

    def _resolve(self, section: str) -> dict[str, type]:
        module_name = f"{self.def_package}.{_CLASS_SECTIONS[section]}"
        module = importlib.import_module(module_name)
        out: dict[str, type] = {}
        missing: list[str] = []
        for name in self.names(section):
            cls = getattr(module, camel_case(name), None)
            if isinstance(cls, type):
                out[name] = cls
            else:
                missing.append(f"{name} ({camel_case(name)})")
        if missing:
            raise RuntimeError(
                f"{self.feat_path.name} declares {section} that {module_name} "
                f"does not provide: {', '.join(missing)} — regenerate with "
                f"`dizzy generate definitions`"
            )
        return out

    @cached_property
    def commands(self) -> dict[str, type]:
        """Command name -> generated pydantic class."""
        return self._resolve("commands")

    @cached_property
    def events(self) -> dict[str, type]:
        """Event name -> generated pydantic class."""
        return self._resolve("events")

    def command_class(self, name: str) -> type:
        cls = self.commands.get(name)
        if cls is None:
            raise KeyError(f"unknown command {name!r} — not declared in {self.feat_path.name}")
        return cls

    def command_name(self, command: Any) -> str:
        """The feat name of a command INSTANCE (or class)."""
        cls = command if isinstance(command, type) else type(command)
        return snake_case(cls.__name__)

    def event_name(self, event: Any) -> str:
        cls = event if isinstance(event, type) else type(event)
        return snake_case(cls.__name__)

    # ── Validation ───────────────────────────────────────────────────────────

    def validate_registered(self, registered: dict[str, set[str]]) -> None:
        """Assert an app's registered elements are exactly what the feat declares.

        *registered* maps a topology section to the names the app actually
        wired. Replaces the hand-maintained ``_REGISTERED`` literal: the feat
        side is read, so only the app's own wiring must be reported.
        """
        problems: list[str] = []
        for section, wired in registered.items():
            # A typo'd section key would otherwise compare against an empty
            # set: silently PASSING when nothing is wired, and blaming the app
            # for "not wiring" every real element when something is.
            declared = set(self.names(section))
            if declared != set(wired):
                problems.append(
                    f"{section}: not wired={sorted(declared - set(wired))} "
                    f"not in feat={sorted(set(wired) - declared)}"
                )
        if problems:
            raise RuntimeError(
                f"wiring/feat mismatch against {self.feat_path.name}: " + "; ".join(problems)
            )

environment property

Environment field names — what a shell must re-hydrate per command.

Derived, so adding an env shape to the feat needs no shell change.

telemetry property

Telemetry sink names — the ports a shell may re-route as transport.

commands cached property

Command name -> generated pydantic class.

events cached property

Event name -> generated pydantic class.

names(section)

The names the feat declares in section, in feat order.

Source code in dizzy/src/dizzy/engine/registry.py

def names(self, section: str) -> tuple[str, ...]:
    """The names the feat declares in *section*, in feat order."""
    return tuple(self._section(section))

entry(section, name)

One declaration, normalized to a dict.

A bare string is a description-only entry (how the feat spells most commands); a NULL value is a declared-but-unwritten entry, which is normal while drafting — it is present, just empty.

Source code in dizzy/src/dizzy/engine/registry.py

def entry(self, section: str, name: str) -> dict[str, Any]:
    """One declaration, normalized to a dict.

    A bare string is a description-only entry (how the feat spells most
    commands); a NULL value is a declared-but-unwritten entry, which is
    normal while drafting — it is present, just empty.
    """
    declared = self._section(section)
    if name not in declared:
        raise KeyError(f"{section}.{name} is not declared in {self.feat_path.name}")
    value = declared[name]
    if value is None:
        return {}
    if isinstance(value, str):
        return {"description": value}
    if isinstance(value, dict):
        return dict(value)
    raise RuntimeError(
        f"{self.feat_path.name}: {section}.{name} is a "
        f"{type(value).__name__}, expected a mapping or a description string"
    )

command_name(command)

The feat name of a command INSTANCE (or class).

Source code in dizzy/src/dizzy/engine/registry.py

def command_name(self, command: Any) -> str:
    """The feat name of a command INSTANCE (or class)."""
    cls = command if isinstance(command, type) else type(command)
    return snake_case(cls.__name__)

validate_registered(registered)

Assert an app's registered elements are exactly what the feat declares.

registered maps a topology section to the names the app actually wired. Replaces the hand-maintained _REGISTERED literal: the feat side is read, so only the app's own wiring must be reported.

Source code in dizzy/src/dizzy/engine/registry.py

def validate_registered(self, registered: dict[str, set[str]]) -> None:
    """Assert an app's registered elements are exactly what the feat declares.

    *registered* maps a topology section to the names the app actually
    wired. Replaces the hand-maintained ``_REGISTERED`` literal: the feat
    side is read, so only the app's own wiring must be reported.
    """
    problems: list[str] = []
    for section, wired in registered.items():
        # A typo'd section key would otherwise compare against an empty
        # set: silently PASSING when nothing is wired, and blaming the app
        # for "not wiring" every real element when something is.
        declared = set(self.names(section))
        if declared != set(wired):
            problems.append(
                f"{section}: not wired={sorted(declared - set(wired))} "
                f"not in feat={sorted(set(wired) - declared)}"
            )
    if problems:
        raise RuntimeError(
            f"wiring/feat mismatch against {self.feat_path.name}: " + "; ".join(problems)
        )

Envelope dataclass

One appended fact, as the stream knows it.

Source code in dizzy/src/dizzy/engine/store.py

@dataclass
class Envelope:
    """One appended fact, as the stream knows it."""

    id: str
    """Content hash — the event's stream identity."""
    type: str
    """Event name in snake_case, i.e. the name the feat declares."""
    ingested_at: datetime
    """UTC, stamped at first append."""
    payload: dict
    parents: tuple = ()
    seq: int = -1
    """DERIVED iteration index, not stored."""

id instance-attribute

Content hash — the event's stream identity.

type instance-attribute

Event name in snake_case, i.e. the name the feat declares.

ingested_at instance-attribute

UTC, stamped at first append.

seq = -1 class-attribute instance-attribute

DERIVED iteration index, not stored.

EventStore

Content-addressed event store. Path from arg > $DIZZY_STORE_PATH > default.

Source code in dizzy/src/dizzy/engine/store.py

class EventStore:
    """Content-addressed event store. Path from arg > ``$DIZZY_STORE_PATH`` > default."""

    def __init__(
        self,
        path: str | Path | None = None,
        event_classes: Mapping[str, type] | None = None,
        graph: FeatGraph | None = None,
    ):
        """*event_classes* maps feat event name -> class, for
        :meth:`reconstruct_event`. Omit both it and *graph* and the store reads
        the ambient feat file when (and only when) something first reconstructs.
        """
        if path is None:
            path = os.environ.get("DIZZY_STORE_PATH") or DEFAULT_STORE_PATH
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()
        self.dag = DagStore(str(self.path), check_same_thread=False)
        self._event_classes = dict(event_classes) if event_classes is not None else None
        self._graph = graph

    @property
    def event_classes(self) -> Mapping[str, type]:
        """The feat's event map, resolved on first use.

        Deferred because appending needs no classes: a worker that only writes
        the stream should not pay to import the generated definitions package.
        """
        if self._event_classes is None:
            self._event_classes = dict((self._graph or default_graph()).events)
        return self._event_classes

    def append(self, event: Any, ingested_at: datetime | None = None) -> Envelope:
        """Append one event and return its envelope.

        ``ingested_at`` is stamped NOW unless supplied — a caller supplies it
        only when replaying or replicating an already-stamped fact.
        """
        stamped = ingested_at or datetime.now(UTC)
        wrapped = {
            "ingested_at": stamped.isoformat(),
            "event": _stringify_floats(event.model_dump(mode="json")),
        }
        with self._lock:
            dag_event = self.dag.append(snake_case(type(event).__name__), wrapped)
            seq = len(self.dag) - 1
        return Envelope(
            id=dag_event.id,
            type=dag_event.type,
            ingested_at=stamped,
            payload=wrapped["event"],
            parents=dag_event.parents,
            seq=seq,
        )

    def iterate(self) -> Iterator[Envelope]:
        """Yield all envelopes in canonical (topological) order."""
        with self._lock:
            events = list(self.dag.iterate())
        for i, ev in enumerate(events):
            yield Envelope(
                id=ev.id,
                type=ev.type,
                ingested_at=datetime.fromisoformat(ev.payload["ingested_at"]),
                payload=ev.payload["event"],
                parents=ev.parents,
                seq=i,
            )

    def heads(self) -> tuple:
        with self._lock:
            return self.dag.heads()

    # ── Replication surface ─────────────────────────────────────────────────
    #
    # Replicated facts arrive already hashed and already stamped, so they do
    # not go through append(): they are ADDED, keeping the id the peer minted.

    def add_replicated(self, event: Any) -> Envelope:
        """Ingest a dagstore event fetched from a peer, returning its envelope.

        The hash is verified on arrival by the DAG, and every parent must
        already be present — replication delivers ancestry first.
        """
        with self._lock:
            self.dag.add(event)
        return Envelope(
            id=event.id,
            type=event.type,
            ingested_at=datetime.fromisoformat(event.payload["ingested_at"]),
            payload=event.payload["event"],
            parents=event.parents,
        )

    def raw_event(self, event_id: str) -> Any:
        """The stored dagstore event, wrapper payload and all — what a peer
        asks for over the wire. Raises KeyError if absent."""
        with self._lock:
            return self.dag.get(event_id)

    def __contains__(self, event_id: str) -> bool:
        with self._lock:
            return event_id in self.dag

    def __len__(self) -> int:
        with self._lock:
            return len(self.dag)

    def reconstruct_event(self, envelope: Envelope) -> Any:
        """Rebuild the event instance from an envelope."""
        return reconstruct_event(envelope, self.event_classes)

event_classes property

The feat's event map, resolved on first use.

Deferred because appending needs no classes: a worker that only writes the stream should not pay to import the generated definitions package.

__init__(path=None, event_classes=None, graph=None)

event_classes maps feat event name -> class, for :meth:reconstruct_event. Omit both it and graph and the store reads the ambient feat file when (and only when) something first reconstructs.

Source code in dizzy/src/dizzy/engine/store.py

def __init__(
    self,
    path: str | Path | None = None,
    event_classes: Mapping[str, type] | None = None,
    graph: FeatGraph | None = None,
):
    """*event_classes* maps feat event name -> class, for
    :meth:`reconstruct_event`. Omit both it and *graph* and the store reads
    the ambient feat file when (and only when) something first reconstructs.
    """
    if path is None:
        path = os.environ.get("DIZZY_STORE_PATH") or DEFAULT_STORE_PATH
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.Lock()
    self.dag = DagStore(str(self.path), check_same_thread=False)
    self._event_classes = dict(event_classes) if event_classes is not None else None
    self._graph = graph

append(event, ingested_at=None)

Append one event and return its envelope.

ingested_at is stamped NOW unless supplied — a caller supplies it only when replaying or replicating an already-stamped fact.

Source code in dizzy/src/dizzy/engine/store.py

def append(self, event: Any, ingested_at: datetime | None = None) -> Envelope:
    """Append one event and return its envelope.

    ``ingested_at`` is stamped NOW unless supplied — a caller supplies it
    only when replaying or replicating an already-stamped fact.
    """
    stamped = ingested_at or datetime.now(UTC)
    wrapped = {
        "ingested_at": stamped.isoformat(),
        "event": _stringify_floats(event.model_dump(mode="json")),
    }
    with self._lock:
        dag_event = self.dag.append(snake_case(type(event).__name__), wrapped)
        seq = len(self.dag) - 1
    return Envelope(
        id=dag_event.id,
        type=dag_event.type,
        ingested_at=stamped,
        payload=wrapped["event"],
        parents=dag_event.parents,
        seq=seq,
    )

iterate()

Yield all envelopes in canonical (topological) order.

Source code in dizzy/src/dizzy/engine/store.py

def iterate(self) -> Iterator[Envelope]:
    """Yield all envelopes in canonical (topological) order."""
    with self._lock:
        events = list(self.dag.iterate())
    for i, ev in enumerate(events):
        yield Envelope(
            id=ev.id,
            type=ev.type,
            ingested_at=datetime.fromisoformat(ev.payload["ingested_at"]),
            payload=ev.payload["event"],
            parents=ev.parents,
            seq=i,
        )

add_replicated(event)

Ingest a dagstore event fetched from a peer, returning its envelope.

The hash is verified on arrival by the DAG, and every parent must already be present — replication delivers ancestry first.

Source code in dizzy/src/dizzy/engine/store.py

def add_replicated(self, event: Any) -> Envelope:
    """Ingest a dagstore event fetched from a peer, returning its envelope.

    The hash is verified on arrival by the DAG, and every parent must
    already be present — replication delivers ancestry first.
    """
    with self._lock:
        self.dag.add(event)
    return Envelope(
        id=event.id,
        type=event.type,
        ingested_at=datetime.fromisoformat(event.payload["ingested_at"]),
        payload=event.payload["event"],
        parents=event.parents,
    )

raw_event(event_id)

The stored dagstore event, wrapper payload and all — what a peer asks for over the wire. Raises KeyError if absent.

Source code in dizzy/src/dizzy/engine/store.py

def raw_event(self, event_id: str) -> Any:
    """The stored dagstore event, wrapper payload and all — what a peer
    asks for over the wire. Raises KeyError if absent."""
    with self._lock:
        return self.dag.get(event_id)

reconstruct_event(envelope)

Rebuild the event instance from an envelope.

Source code in dizzy/src/dizzy/engine/store.py

def reconstruct_event(self, envelope: Envelope) -> Any:
    """Rebuild the event instance from an envelope."""
    return reconstruct_event(envelope, self.event_classes)

chain_observers(*observers)

Compose observers into the one the engine accepts, in order.

Source code in dizzy/src/dizzy/engine/ports.py

def chain_observers(*observers: Callable[[str, Any], None]) -> Callable[[str, Any], None]:
    """Compose observers into the one the engine accepts, in order."""

    def observer(name: str, event: Any) -> None:
        for obs in observers:
            obs(name, event)

    return observer

null_app(build_runtime, feat_path=None)

The minimal HostApp: a feat file and a way to build the engine.

Source code in dizzy/src/dizzy/engine/ports.py

def null_app(
    build_runtime: Callable[[ShellServices], Runtime], feat_path: str | None = None
) -> HostApp:
    """The minimal HostApp: a feat file and a way to build the engine."""
    return HostApp(graph=default_graph(feat_path), build_runtime=build_runtime)

camel_case(name)

classify_image -> ClassifyImage.

LinkML's camelcase semantics (that generator produced the classes, so this must match it, not merely resemble it): split on non-word runs and underscores, upper the first character of each part, keep the rest.

Source code in dizzy/src/dizzy/engine/registry.py

def camel_case(name: str) -> str:
    """``classify_image`` -> ``ClassifyImage``.

    LinkML's ``camelcase`` semantics (that generator produced the classes, so
    this must match it, not merely resemble it): split on non-word runs and
    underscores, upper the first character of each part, keep the rest.
    """
    return "".join(f"{p[0].upper()}{p[1:]}" for p in re.split(r"[\W_]+", name) if p)

check_name(name, section, feat_name)

Reject a declared name that does not survive the round trip.

camel_case splits on any non-word run, so classify-image, classifyImage and classify_image all collapse to ClassifyImage. Left unchecked, a typo'd feat entry resolves to its NEIGHBOUR's class and the reverse lookup (which routes commands) silently maps it back to the wrong name — the stale-generation check would pass on a broken feat. A name is only well formed if snake_case(camel_case(name)) == name.

Also catches YAML's scalar keys: 123: or on: parse to int/bool and would otherwise die inside re with no mention of the feat file.

Source code in dizzy/src/dizzy/engine/registry.py

def check_name(name: Any, section: str, feat_name: str) -> str:
    """Reject a declared name that does not survive the round trip.

    ``camel_case`` splits on any non-word run, so ``classify-image``,
    ``classifyImage`` and ``classify_image`` all collapse to ``ClassifyImage``.
    Left unchecked, a typo'd feat entry resolves to its NEIGHBOUR's class and
    the reverse lookup (which routes commands) silently maps it back to the
    wrong name — the stale-generation check would pass on a broken feat. A
    name is only well formed if ``snake_case(camel_case(name)) == name``.

    Also catches YAML's scalar keys: ``123:`` or ``on:`` parse to int/bool and
    would otherwise die inside ``re`` with no mention of the feat file.
    """
    if not isinstance(name, str):
        raise RuntimeError(
            f"{feat_name}: {section} declares a non-string name {name!r} "
            f"({type(name).__name__}) — quote it (YAML reads 123, yes, on, off "
            f"as scalars)"
        )
    if not name or snake_case(camel_case(name)) != name:
        raise RuntimeError(
            f"{feat_name}: {section} declares {name!r}, which is not "
            f"snake_case — it would resolve to {camel_case(name)!r}, the same "
            f"class as {snake_case(camel_case(name))!r}"
        )
    return name

find_feat(start=None)

Locate the app's feat file.

$DIZZY_FEAT_PATH wins. Otherwise walk up from start (default: the working directory) looking for exactly one *.feat.yaml. This is how a worker boots knowing only where it is — no app import required.

Source code in dizzy/src/dizzy/engine/registry.py

def find_feat(start: Path | None = None) -> Path:
    """Locate the app's feat file.

    ``$DIZZY_FEAT_PATH`` wins. Otherwise walk up from *start* (default: the
    working directory) looking for exactly one ``*.feat.yaml``. This is how a
    worker boots knowing only where it is — no app import required.
    """
    env = os.environ.get("DIZZY_FEAT_PATH")
    if env is not None and env.strip() == "":
        # An exported-but-empty variable is a misconfiguration, not "unset":
        # falling through to the walk-up would silently pick up whatever feat
        # happens to be nearest, which is worse than failing.
        raise RuntimeError(
            "$DIZZY_FEAT_PATH is set but empty — unset it to "
            "search upward, or point it at a feat file"
        )
    if env:
        path = Path(env).expanduser()
        if not path.is_file():
            raise FileNotFoundError(
                f"$DIZZY_FEAT_PATH is not a file: {path}"
                + (" (it is a directory)" if path.is_dir() else "")
            )
        return path
    here = (start or Path.cwd()).resolve()
    for directory in (here, *here.parents):
        found = sorted(directory.glob("*.feat.yaml"))
        if len(found) == 1:
            return found[0]
        if len(found) > 1:
            raise RuntimeError(
                f"{directory} holds {len(found)} feat files "
                f"({', '.join(p.name for p in found)}) — set $DIZZY_FEAT_PATH"
            )
    raise FileNotFoundError(f"no *.feat.yaml found from {here} upward — set $DIZZY_FEAT_PATH")

graph(feat_path=None, def_package=DEFAULT_DEF_PACKAGE)

Process-wide FeatGraph cache — a worker parses its feat once.

Keyed on the RESOLVED path, not on nothing: an earlier version cached a single graph, so once any caller had built one, a later $DIZZY_FEAT_PATH (or a different cwd, since discovery walks up) silently handed back the first caller's feat. Re-resolving per call is two env reads; parsing is what the cache is for.

Source code in dizzy/src/dizzy/engine/registry.py

def graph(feat_path: str | Path | None = None, def_package: str = DEFAULT_DEF_PACKAGE) -> FeatGraph:
    """Process-wide FeatGraph cache — a worker parses its feat once.

    Keyed on the RESOLVED path, not on nothing: an earlier version cached a
    single graph, so once any caller had built one, a later `$DIZZY_FEAT_PATH`
    (or a different cwd, since discovery walks up) silently handed back the
    first caller's feat. Re-resolving per call is two env reads; parsing is
    what the cache is for.
    """
    path = Path(feat_path) if feat_path is not None else find_feat()
    key = (path.resolve(), def_package)
    if key not in _graphs:
        _graphs[key] = FeatGraph.load(path, def_package)
    return _graphs[key]

reset_graph()

Drop the cache — for tests that rewrite a feat file in place.

Source code in dizzy/src/dizzy/engine/registry.py

def reset_graph() -> None:
    """Drop the cache — for tests that rewrite a feat file in place."""
    _graphs.clear()

snake_case(name)

ClassifyImage -> classify_image — the inverse of camel_case.

Source code in dizzy/src/dizzy/engine/registry.py

def snake_case(name: str) -> str:
    """``ClassifyImage`` -> ``classify_image`` — the inverse of camel_case."""
    return re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()

reconstruct_event(envelope, event_classes)

Rebuild the event instance an envelope stands for.

event_classes is a feat-name -> class map; FeatGraph.events is one.

Source code in dizzy/src/dizzy/engine/store.py

def reconstruct_event(envelope: Envelope, event_classes: Mapping[str, type]) -> Any:
    """Rebuild the event instance an envelope stands for.

    *event_classes* is a feat-name -> class map; ``FeatGraph.events`` is one.
    """
    cls = event_classes.get(envelope.type)
    if cls is None:
        raise KeyError(
            f"unknown event type in stream: {envelope.type!r} — the feat does "
            f"not declare it, so this stream was written by a different "
            f"feature (or by a newer version of this one)"
        )
    return cls(**envelope.payload)