Execution

Named, checked execution for whole modules.

execution

Named, checked execution for whole QIEC modules.

The evaluator deliberately operates on core terms and process-local attachments. This module supplies the user-facing boundary above it: it selects a named computation, specializes its static telescope, validates host arguments and results, creates a new attachment table for every invocation, and reports failures with stable diagnostic codes.

Runtime providers are explicit. A configuration names registered provider factories (or carries provider objects when used from Python); no module import or ambient attachment registry is consulted by run_named.

TraceObserver

TraceObserver = Callable[['ExecutionTraceEvent'], None]

A callback receiving each event an execution emits, in order.

RuntimeProviderFactory

RuntimeProviderFactory = Callable[[Mapping[str, object]], RuntimeProvider]

Build a runtime provider from the options its configuration entry records.

ExecutionTraceEvent dataclass

ExecutionTraceEvent(sequence: int, event: str, computation: str, detail: Mapping[str, object] = dict())

One stable event emitted by a named QIEC invocation.

PARAMETER	DESCRIPTION
sequence	The event's position in the invocation's trace, from zero. TYPE: int
event	The stable event name, such as "run.started". TYPE: str
computation	The name of the computation being run. TYPE: str
detail	JSON-compatible event data; frozen on construction. TYPE: Mapping[str, object] DEFAULT: dict()

__post_init__

__post_init__() -> None

Freeze detail so a recorded event cannot change after emission.

Source code in src/quivers/qiec/execution.py

def __post_init__(self) -> None:
    """Freeze ``detail`` so a recorded event cannot change after emission."""
    object.__setattr__(self, "detail", MappingProxyType(dict(self.detail)))

to_data

to_data() -> dict[str, object]

Render the event as JSON-compatible data.

RETURNS	DESCRIPTION
dict[str, object]	The sequence number, event name, computation name, and detail mapping, with every detail value made JSON-compatible.

Source code in src/quivers/qiec/execution.py

def to_data(self) -> dict[str, object]:
    """Render the event as JSON-compatible data.

    Returns
    -------
    dict[str, object]
        The sequence number, event name, computation name, and detail
        mapping, with every detail value made JSON-compatible.
    """
    return {
        "sequence": self.sequence,
        "event": self.event,
        "computation": self.computation,
        "detail": _json_value(dict(self.detail)),
    }

ExecutionTraceRecorder dataclass

ExecutionTraceRecorder(events: list[ExecutionTraceEvent] = list())

Collect events while also satisfying the :class:TraceObserver API.

PARAMETER	DESCRIPTION
events	The recorded events, in order; empty to start. TYPE: list[ExecutionTraceEvent] DEFAULT: list()

__call__

__call__(event: ExecutionTraceEvent) -> None

Record one event.

PARAMETER	DESCRIPTION
event	The event the invocation just emitted. TYPE: ExecutionTraceEvent

Source code in src/quivers/qiec/execution.py

def __call__(self, event: ExecutionTraceEvent) -> None:
    """Record one event.

    Parameters
    ----------
    event
        The event the invocation just emitted.
    """
    self.events.append(event)

ExecutionDiagnostic dataclass

ExecutionDiagnostic(code: str, message: str, computation: str | None = None, origin: SourceOrigin | None = None, severity: Literal['error', 'warning', 'note'] = 'error')

A stable failure at the source-to-host execution boundary.

PARAMETER	DESCRIPTION
code	The stable diagnostic code, such as "qiec-run-arity". TYPE: str
message	The human-readable explanation. TYPE: str
computation	The named computation involved, if any. TYPE: str | None DEFAULT: None
origin	The source location involved, if any. TYPE: SourceOrigin | None DEFAULT: None
severity	"error", "warning", or "note". TYPE: Literal['error', 'warning', 'note'] DEFAULT: 'error'

to_data

to_data() -> dict[str, object]

Render the diagnostic as JSON-compatible data.

RETURNS	DESCRIPTION
dict[str, object]	The code, severity, message, computation name, and origin (or None when the failure has no source location).

Source code in src/quivers/qiec/execution.py

def to_data(self) -> dict[str, object]:
    """Render the diagnostic as JSON-compatible data.

    Returns
    -------
    dict[str, object]
        The code, severity, message, computation name, and origin (or
        ``None`` when the failure has no source location).
    """
    return {
        "code": self.code,
        "severity": self.severity,
        "message": self.message,
        "computation": self.computation,
        "origin": self.origin.to_data() if self.origin is not None else None,
    }

ExecutionFailure

ExecutionFailure(diagnostic: ExecutionDiagnostic)

Bases: EvaluationError

Raised with a serializable diagnostic when named execution cannot run.

PARAMETER	DESCRIPTION
diagnostic	The failure, kept on the error as diagnostic; its message becomes the error's message. TYPE: ExecutionDiagnostic

Source code in src/quivers/qiec/execution.py

def __init__(self, diagnostic: ExecutionDiagnostic) -> None:
    super().__init__(diagnostic.message)
    self.diagnostic = diagnostic

RuntimeProvider

Bases: Protocol

An explicit source of process-local handlers and type validators.

name property

name: str

The provider's stable name, reported in traces and result labels.

attach

attach(module: QiecModule, attachments: RuntimeAttachments) -> None

Bind this provider's handlers into a fresh attachment table.

PARAMETER	DESCRIPTION
module	The validated module whose handlers and effects are being served. TYPE: QiecModule
attachments	The per-invocation table receiving runtime handlers. TYPE: RuntimeAttachments

Source code in src/quivers/qiec/execution.py

def attach(self, module: QiecModule, attachments: RuntimeAttachments) -> None:
    """Bind this provider's handlers into a fresh attachment table.

    Parameters
    ----------
    module
        The validated module whose handlers and effects are being served.
    attachments
        The per-invocation table receiving runtime handlers.
    """
    ...

validator_for

validator_for(type_: TypeExpr) -> RuntimeValidator | None

Supply a host validator for a runtime type, if this provider has one.

PARAMETER	DESCRIPTION
type_	The closed QIEC type an argument or result must inhabit. TYPE: TypeExpr

RETURNS	DESCRIPTION
RuntimeValidator | None	A predicate over host values, or None when this provider does not cover the type.

Source code in src/quivers/qiec/execution.py

def validator_for(self, type_: TypeExpr) -> RuntimeValidator | None:
    """Supply a host validator for a runtime type, if this provider has one.

    Parameters
    ----------
    type_
        The closed QIEC type an argument or result must inhabit.

    Returns
    -------
    RuntimeValidator | None
        A predicate over host values, or ``None`` when this provider does
        not cover the type.
    """
    ...

CoreRuntimeProvider dataclass

CoreRuntimeProvider(options: Mapping[str, object] = dict(), name: str = 'core')

Primitive validators plus explicitly configured structural handlers.

options.handlers maps authored handler names to either {"kind": "passthrough"}, {"kind": "state", "initial": ...}, or {"kind": "scripted", "responses": {"op": [...]}}. The source handler still controls types, coverage, and resumption grades; options supply only process-local clause behavior and state.

PARAMETER	DESCRIPTION
options	JSON-compatible configuration; frozen on construction. TYPE: Mapping[str, object] DEFAULT: dict()
name	The provider's name, reported in traces and result labels. TYPE: str DEFAULT: 'core'

__post_init__

__post_init__() -> None

Freeze options so a provider's configuration cannot drift.

Source code in src/quivers/qiec/execution.py

def __post_init__(self) -> None:
    """Freeze ``options`` so a provider's configuration cannot drift."""
    object.__setattr__(self, "options", MappingProxyType(dict(self.options)))

attach

attach(module: QiecModule, attachments: RuntimeAttachments) -> None

Bind one structural handler per configured authored handler.

PARAMETER	DESCRIPTION
module	The validated module whose handlers the configuration names. TYPE: QiecModule
attachments	The per-invocation table receiving the constructed handlers. TYPE: RuntimeAttachments

RAISES

DESCRIPTION

ValueError

If options.handlers is not an object, names a handler the module does not declare, holds a non-object configuration, or requests a kind other than passthrough, scripted, state, or one of the prelude kinds draw, draw-scoring (with the option score_instance), enumerate (with the options grouped and reduction), replay, collect (with the options combine and forkable), score, search (with the option reduction), and param (with the option values).

Source code in src/quivers/qiec/execution.py

def attach(self, module: QiecModule, attachments: RuntimeAttachments) -> None:
    """Bind one structural handler per configured authored handler.

    Parameters
    ----------
    module
        The validated module whose handlers the configuration names.
    attachments
        The per-invocation table receiving the constructed handlers.

    Raises
    ------
    ValueError
        If ``options.handlers`` is not an object, names a handler the module
        does not declare, holds a non-object configuration, or requests a
        kind other than ``passthrough``, ``scripted``, ``state``, or one
        of the prelude kinds ``draw``, ``draw-scoring`` (with the option
        ``score_instance``), ``enumerate`` (with the options
        ``grouped`` and ``reduction``), ``replay``, ``collect`` (with
        the options ``combine`` and ``forkable``), ``score``, ``search`` (with the
        option ``reduction``), and ``param`` (with the option
        ``values``).
    """
    configured = self.options.get("handlers", {})
    if not isinstance(configured, Mapping):
        raise ValueError("core runtime option 'handlers' must be an object")
    definitions = {handler.name: handler for handler in module.handlers}
    unknown = set(configured) - set(definitions)
    if unknown:
        raise ValueError(
            f"core runtime names unknown handlers: {', '.join(sorted(unknown))}"
        )
    operation_names = {
        operation.id: operation.name
        for effect in module.effects
        for operation in effect.operations
    }
    for handler_name, raw_options in configured.items():
        if not isinstance(handler_name, str) or not isinstance(
            raw_options, Mapping
        ):
            raise ValueError("each core handler configuration must be an object")
        definition = definitions[handler_name]
        kind = raw_options.get("kind")
        if kind == "passthrough":
            runtime = self._passthrough(definition, raw_options, operation_names)
        elif kind == "scripted":
            runtime = self._scripted(definition, raw_options, operation_names)
        elif kind == "state":
            runtime = self._state(definition, raw_options, operation_names)
        elif kind == "draw":
            runtime = self._prelude(definition, draw_handler, "Random")
        elif kind == "draw-scoring":
            score_instance = _replay_score_instance(raw_options)
            if score_instance is None:
                raise ValueError(
                    f"core handler {definition.name!r} of kind 'draw-scoring' "
                    "needs a `score_instance`"
                )
            runtime = self._prelude(
                definition,
                lambda result_validator, answer_type: draw_scoring_handler(
                    score_instance=score_instance,
                    result_validator=result_validator,
                    answer_type=answer_type,
                ),
                "Random",
            )
        elif kind == "enumerate":
            runtime = self._prelude(
                definition,
                lambda result_validator, answer_type: enumerate_handler(
                    result_validator,
                    answer_type=answer_type,
                    weight_type=(
                        definition.output_type
                        if _enumerate_grouped(raw_options, definition.name)
                        else None
                    ),
                    reduction=_enumerate_reduction(raw_options, definition.name),
                ),
                "Random",
            )
        elif kind == "replay":
            runtime = self._prelude(
                definition,
                lambda result_validator, answer_type: replay_handler(
                    _replay_values(raw_options, definition.name),
                    result_validator=result_validator,
                    policy=_replay_policy(raw_options, definition.name),
                    score_instance=_replay_score_instance(raw_options),
                    answer_type=answer_type,
                ),
                "Random",
            )
        elif kind == "collect":
            identity, combine = _collect_combine(raw_options, definition.name)
            forkable = raw_options.get("forkable", False)
            if not isinstance(forkable, bool):
                raise ValueError(
                    f"core handler {definition.name!r} option 'forkable' must "
                    "be a boolean"
                )
            runtime = self._prelude(
                definition,
                lambda **_: weight_handler(
                    identity,
                    combine,
                    _weights_validator(_weight_type(definition)),
                    weight_type=_weight_type(definition),
                    answer_type=definition.input_type,
                    answer_validator=_weights_validator(definition.input_type),
                    forkable=forkable,
                )[0],
                "Weight",
            )
        elif kind == "search":
            runtime = self._search(definition, raw_options)
        elif kind == "param":
            runtime = self._param(definition, raw_options)
        elif kind == "score":
            runtime = self._prelude(
                definition,
                lambda **_: score_handler(
                    weight_validator=_require_core_handler_validator(
                        LOG_WEIGHT, definition.name, "weight"
                    ),
                    answer_validator=_require_core_handler_validator(
                        definition.input_type, definition.name, "input"
                    ),
                    expose_total=definition.output_type != definition.input_type,
                )[0],
                "Score",
            )
        else:
            raise ValueError(
                f"core handler {handler_name!r} has unknown kind {kind!r}"
            )
        attachments.bind_handler(runtime)

validator_for

validator_for(type_: TypeExpr) -> RuntimeValidator | None

Supply the primitive validator for a closed runtime type.

PARAMETER	DESCRIPTION
type_	The closed QIEC type an argument or result must inhabit. TYPE: TypeExpr

RETURNS	DESCRIPTION
RuntimeValidator | None	The core validator, or None for types the core runtime cannot check on its own.

Source code in src/quivers/qiec/execution.py

def validator_for(self, type_: TypeExpr) -> RuntimeValidator | None:
    """Supply the primitive validator for a closed runtime type.

    Parameters
    ----------
    type_
        The closed QIEC type an argument or result must inhabit.

    Returns
    -------
    RuntimeValidator | None
        The core validator, or ``None`` for types the core runtime cannot
        check on its own.
    """
    return _core_validator(type_)

RuntimeSelection dataclass

RuntimeSelection(provider: str, options: Mapping[str, object] = dict())

One named provider and its JSON-compatible construction options.

PARAMETER	DESCRIPTION
provider	The registered or entry-point name of the provider; must be non-empty. TYPE: str
options	The options passed to the provider's factory; frozen on construction. TYPE: Mapping[str, object] DEFAULT: dict()

__post_init__

__post_init__() -> None

Reject empty provider names and freeze the options.

RAISES	DESCRIPTION
ValueError	If provider is empty.

Source code in src/quivers/qiec/execution.py

def __post_init__(self) -> None:
    """Reject empty provider names and freeze the options.

    Raises
    ------
    ValueError
        If ``provider`` is empty.
    """
    if not self.provider:
        raise ValueError("runtime provider name cannot be empty")
    object.__setattr__(self, "options", MappingProxyType(dict(self.options)))

RuntimeConfiguration dataclass

RuntimeConfiguration(selections: tuple[RuntimeSelection, ...] = (RuntimeSelection('core'),), providers: tuple[RuntimeProvider, ...] = ())

Complete, explicit runtime selection for one or more invocations.

PARAMETER	DESCRIPTION
selections	Providers to build by name for every invocation; the core provider alone by default. TYPE: tuple[RuntimeSelection, ...] DEFAULT: (RuntimeSelection('core'),)
providers	Already-built provider objects, appended after the selections when used from Python. TYPE: tuple[RuntimeProvider, ...] DEFAULT: ()

label property

label: str

A +-joined label naming every selected and supplied provider.

RETURNS	DESCRIPTION
str	The provider names in order, or "detached" when there are none.

instantiate

instantiate() -> tuple[RuntimeProvider, ...]

Build every named provider and append the programmatic ones.

RETURNS	DESCRIPTION
tuple[RuntimeProvider, ...]	The providers in selection order, followed by providers.

RAISES	DESCRIPTION
ExecutionFailure	With code qiec-run-provider if a selection names an unknown provider, its factory cannot be loaded, or the factory rejects the selection's options.

Source code in src/quivers/qiec/execution.py

def instantiate(self) -> tuple[RuntimeProvider, ...]:
    """Build every named provider and append the programmatic ones.

    Returns
    -------
    tuple[RuntimeProvider, ...]
        The providers in selection order, followed by ``providers``.

    Raises
    ------
    ExecutionFailure
        With code ``qiec-run-provider`` if a selection names an unknown
        provider, its factory cannot be loaded, or the factory rejects the
        selection's options.
    """
    providers: list[RuntimeProvider] = []
    for selection in self.selections:
        try:
            factory = _provider_factory(selection.provider)
        except KeyError as error:
            available = ", ".join(available_runtime_providers()) or "(none)"
            raise ExecutionFailure(
                ExecutionDiagnostic(
                    "qiec-run-provider",
                    f"unknown runtime provider {selection.provider!r}; "
                    f"available providers: {available}",
                )
            ) from error
        except Exception as error:
            raise ExecutionFailure(
                ExecutionDiagnostic(
                    "qiec-run-provider",
                    f"could not load runtime provider {selection.provider!r}: {error}",
                )
            ) from error
        try:
            providers.append(factory(selection.options))
        except Exception as error:
            raise ExecutionFailure(
                ExecutionDiagnostic(
                    "qiec-run-provider",
                    f"runtime provider {selection.provider!r} rejected its "
                    f"configuration: {error}",
                )
            ) from error
    providers.extend(self.providers)
    return tuple(providers)

to_data

to_data() -> dict[str, object]

Render the configuration as JSON-compatible data.

RETURNS	DESCRIPTION
dict[str, object]	The named selections with their options, and the names of any programmatic providers.

Source code in src/quivers/qiec/execution.py

def to_data(self) -> dict[str, object]:
    """Render the configuration as JSON-compatible data.

    Returns
    -------
    dict[str, object]
        The named selections with their options, and the names of any
        programmatic providers.
    """
    return {
        "providers": [
            {"name": selection.provider, "options": dict(selection.options)}
            for selection in self.selections
        ],
        "programmatic_providers": [provider.name for provider in self.providers],
    }

ExecutionResult dataclass

ExecutionResult(computation: str, value: object, result_type: TypeExpr, runtime: str, trace: tuple[ExecutionTraceEvent, ...])

The checked value and trace produced by one named invocation.

PARAMETER	DESCRIPTION
computation	The name of the computation run. TYPE: str
value	The validated host result. TYPE: object
result_type	The computation's result type under its specialization. TYPE: TypeExpr
runtime	The label of the runtime configuration used. TYPE: str
trace	Every event the invocation emitted, in order. TYPE: tuple[ExecutionTraceEvent, ...]

to_data

to_data() -> dict[str, object]

Render the result as JSON-compatible data.

RETURNS	DESCRIPTION
dict[str, object]	ok set to True, the computation name, the JSON-compatible value, the rendered result type, the runtime label, and the trace.

Source code in src/quivers/qiec/execution.py

def to_data(self) -> dict[str, object]:
    """Render the result as JSON-compatible data.

    Returns
    -------
    dict[str, object]
        ``ok`` set to ``True``, the computation name, the JSON-compatible
        value, the rendered result type, the runtime label, and the trace.
    """
    return {
        "ok": True,
        "computation": self.computation,
        "value": _json_value(self.value),
        "result_type": _render_type(self.result_type),
        "runtime": self.runtime,
        "trace": [event.to_data() for event in self.trace],
    }

available_runtime_providers

available_runtime_providers() -> tuple[str, ...]

Return built-in and installed quivers.qiec_runtime entry points.

RETURNS	DESCRIPTION
tuple[str, ...]	Every selectable provider name, sorted.

Source code in src/quivers/qiec/execution.py

def available_runtime_providers() -> tuple[str, ...]:
    """Return built-in and installed ``quivers.qiec_runtime`` entry points.

    Returns
    -------
    tuple[str, ...]
        Every selectable provider name, sorted.
    """

    installed = {
        point.name for point in entry_points().select(group="quivers.qiec_runtime")
    }
    return tuple(sorted({*_PROVIDER_FACTORIES, *installed}))

register_runtime_provider

register_runtime_provider(name: str, factory: RuntimeProviderFactory) -> None

Register a provider factory for explicit configuration by name.

Registration makes a provider selectable; it never activates that provider for an invocation. Replacing an existing name is rejected so plugin load order cannot silently change execution semantics.

PARAMETER	DESCRIPTION
name	The name a configuration uses to select the provider. TYPE: str
factory	The callable building a provider from JSON-compatible options. TYPE: RuntimeProviderFactory

RAISES	DESCRIPTION
ValueError	If name is empty or already registered.

Source code in src/quivers/qiec/execution.py

def register_runtime_provider(name: str, factory: RuntimeProviderFactory) -> None:
    """Register a provider factory for explicit configuration by name.

    Registration makes a provider selectable; it never activates that provider
    for an invocation.  Replacing an existing name is rejected so plugin load
    order cannot silently change execution semantics.

    Parameters
    ----------
    name
        The name a configuration uses to select the provider.
    factory
        The callable building a provider from JSON-compatible options.

    Raises
    ------
    ValueError
        If ``name`` is empty or already registered.
    """

    if not name or name in _PROVIDER_FACTORIES:
        raise ValueError(f"runtime provider {name!r} is already registered or invalid")
    _PROVIDER_FACTORIES[name] = factory

runtime_configuration_from_data

runtime_configuration_from_data(data: object) -> RuntimeConfiguration

Decode the non-executable JSON runtime configuration format.

PARAMETER	DESCRIPTION
data	The decoded JSON document; providers lists names or {"name": ..., "options": {...}} objects and defaults to core. TYPE: object

RETURNS	DESCRIPTION
RuntimeConfiguration	The selections in document order, with no programmatic providers.

RAISES	DESCRIPTION
ValueError	If data is not an object, providers is not a list, an entry is neither a name nor an object with a name, or an entry's options is not an object.

Source code in src/quivers/qiec/execution.py

def runtime_configuration_from_data(data: object) -> RuntimeConfiguration:
    """Decode the non-executable JSON runtime configuration format.

    Parameters
    ----------
    data
        The decoded JSON document; ``providers`` lists names or
        ``{"name": ..., "options": {...}}`` objects and defaults to ``core``.

    Returns
    -------
    RuntimeConfiguration
        The selections in document order, with no programmatic providers.

    Raises
    ------
    ValueError
        If ``data`` is not an object, ``providers`` is not a list, an entry
        is neither a name nor an object with a ``name``, or an entry's
        ``options`` is not an object.
    """

    if not isinstance(data, dict):
        raise ValueError("runtime configuration must be a JSON object")
    raw = data.get("providers", [{"name": "core"}])
    if not isinstance(raw, list):
        raise ValueError("runtime configuration 'providers' must be a list")
    selections: list[RuntimeSelection] = []
    for item in raw:
        if isinstance(item, str):
            selections.append(RuntimeSelection(item))
            continue
        if not isinstance(item, dict) or not isinstance(item.get("name"), str):
            raise ValueError(
                "each runtime provider must be a name or object with 'name'"
            )
        options = item.get("options", {})
        if not isinstance(options, dict):
            raise ValueError("runtime provider 'options' must be an object")
        selections.append(RuntimeSelection(item["name"], options))
    return RuntimeConfiguration(tuple(selections))

load_runtime_configuration

load_runtime_configuration(path: str | Path) -> RuntimeConfiguration

Load an explicit runtime selection without importing executable code.

PARAMETER	DESCRIPTION
path	A JSON file in the format :func:runtime_configuration_from_data accepts. TYPE: str | Path

RETURNS	DESCRIPTION
RuntimeConfiguration	The decoded selection.

RAISES	DESCRIPTION
ValueError	If the file cannot be read, is not valid JSON, or does not decode as a runtime configuration.

Source code in src/quivers/qiec/execution.py

def load_runtime_configuration(path: str | Path) -> RuntimeConfiguration:
    """Load an explicit runtime selection without importing executable code.

    Parameters
    ----------
    path
        A JSON file in the format :func:`runtime_configuration_from_data`
        accepts.

    Returns
    -------
    RuntimeConfiguration
        The decoded selection.

    Raises
    ------
    ValueError
        If the file cannot be read, is not valid JSON, or does not decode
        as a runtime configuration.
    """

    source = Path(path)
    try:
        data = json.loads(source.read_text())
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError(
            f"could not load runtime configuration {source}: {error}"
        ) from error
    return runtime_configuration_from_data(data)

parse_static_arguments

parse_static_arguments(module: QiecModule, computation: str, assignments: tuple[str, ...]) -> tuple[StaticArgument, ...]

Resolve CLI NAME=TERM specializations in telescope order.

This intentionally accepts only closed, nominal terms available in the module: primitive or declared types, closed index constructors/literals, and declared effect applications. Static variables and inferred holes are not runtime configuration values.

PARAMETER	DESCRIPTION
module	The module declaring computation and the names its terms may use. TYPE: QiecModule
computation	The named computation whose telescope is being instantiated. TYPE: str
assignments	NAME=TERM strings, one per telescope binder, in any order. TYPE: tuple[str, ...]

RETURNS	DESCRIPTION
tuple[StaticArgument, ...]	The parsed arguments in the telescope's order.

RAISES	DESCRIPTION
ExecutionFailure	With code qiec-run-computation if the computation is unknown, or qiec-run-static if an assignment is malformed, repeated, names a binder the telescope lacks, omits one it has, or does not parse at the binder's sort.

Source code in src/quivers/qiec/execution.py

def parse_static_arguments(
    module: QiecModule,
    computation: str,
    assignments: tuple[str, ...],
) -> tuple[StaticArgument, ...]:
    """Resolve CLI ``NAME=TERM`` specializations in telescope order.

    This intentionally accepts only closed, nominal terms available in the
    module: primitive or declared types, closed index constructors/literals,
    and declared effect applications.  Static variables and inferred holes are
    not runtime configuration values.

    Parameters
    ----------
    module
        The module declaring ``computation`` and the names its terms may use.
    computation
        The named computation whose telescope is being instantiated.
    assignments
        ``NAME=TERM`` strings, one per telescope binder, in any order.

    Returns
    -------
    tuple[StaticArgument, ...]
        The parsed arguments in the telescope's order.

    Raises
    ------
    ExecutionFailure
        With code ``qiec-run-computation`` if the computation is unknown, or
        ``qiec-run-static`` if an assignment is malformed, repeated, names a
        binder the telescope lacks, omits one it has, or does not parse at
        the binder's sort.
    """

    selected = next(
        (
            candidate
            for candidate in module.computations
            if candidate.name == computation
        ),
        None,
    )
    if selected is None:
        _fail("qiec-run-computation", f"unknown QIEC computation {computation!r}")
    assert selected is not None
    values: dict[str, str] = {}
    for assignment in assignments:
        name, separator, value = assignment.partition("=")
        if not separator or not name or not value:
            _fail(
                "qiec-run-static",
                f"static argument {assignment!r} must have the form NAME=TERM",
                computation,
                selected.origin,
            )
        if name in values:
            _fail(
                "qiec-run-static",
                f"static argument {name!r} was supplied more than once",
                computation,
                selected.origin,
            )
        values[name] = value
    expected = {binder.name for binder in selected.telescope}
    unknown = set(values) - expected
    missing = expected - set(values)
    if unknown or missing:
        detail: list[str] = []
        if missing:
            detail.append(f"missing {', '.join(sorted(missing))}")
        if unknown:
            detail.append(f"unknown {', '.join(sorted(unknown))}")
        _fail(
            "qiec-run-static",
            "invalid static specialization: " + "; ".join(detail),
            computation,
            selected.origin,
        )
    try:
        return tuple(
            _parse_static(module, binder, values[binder.name])
            for binder in selected.telescope
        )
    except (TypeError, ValueError) as error:
        _fail(
            "qiec-run-static",
            f"invalid static specialization for {computation!r}: {error}",
            computation,
            selected.origin,
        )

run_named

run_named(module: QiecModule, computation: str, arguments: tuple[object, ...] = (), *, static_arguments: tuple[StaticArgument, ...] = (), runtime: RuntimeConfiguration | None = None, observer: TraceObserver | None = None, fuel: int | None = None) -> ExecutionResult

Validate and execute one named computation from module.

The attachment table and every configured provider instance are allocated inside this call. Mutable handler state can thus never leak from one named invocation to the next.

PARAMETER	DESCRIPTION
module	The module declaring the computation; it is validated before the body is evaluated. TYPE: QiecModule
computation	The name of the computation to run. TYPE: str
arguments	Host values for the computation's value parameters, in order. TYPE: tuple[object, ...] DEFAULT: ()
static_arguments	Closed terms instantiating the computation's telescope, in order; required whenever the telescope is non-empty. TYPE: tuple[StaticArgument, ...] DEFAULT: ()
runtime	The providers to attach; defaults to the core provider alone. TYPE: RuntimeConfiguration | None DEFAULT: None
observer	A callback receiving every trace event as it is emitted. TYPE: TraceObserver | None DEFAULT: None
fuel	The number of machine steps the run may take, or None for no bound. General recursion may diverge, and a bound is the only way a caller can be sure a run ends. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
ExecutionResult	The validated result value, its specialized type, the runtime label, and the full trace.

RAISES

DESCRIPTION

ExecutionFailure

With code qiec-run-computation for an unknown name, qiec-run-static for a bad specialization, qiec-run-arity for a wrong argument count, qiec-run-provider when a provider cannot be built or attached, qiec-run-validator when no provider covers an argument or result type, qiec-run-argument or qiec-run-result when a value fails validation, qiec-run-module when the module fails validation, qiec-run-fuel when the step budget runs out, and qiec-run-evaluation when evaluation or a runtime clause fails.

Source code in src/quivers/qiec/execution.py

def run_named(
    module: QiecModule,
    computation: str,
    arguments: tuple[object, ...] = (),
    *,
    static_arguments: tuple[StaticArgument, ...] = (),
    runtime: RuntimeConfiguration | None = None,
    observer: TraceObserver | None = None,
    fuel: int | None = None,
) -> ExecutionResult:
    """Validate and execute one named computation from ``module``.

    The attachment table and every configured provider instance are allocated
    inside this call.  Mutable handler state can thus never leak from one named
    invocation to the next.

    Parameters
    ----------
    module
        The module declaring the computation; it is validated before the
        body is evaluated.
    computation
        The name of the computation to run.
    arguments
        Host values for the computation's value parameters, in order.
    static_arguments
        Closed terms instantiating the computation's telescope, in order;
        required whenever the telescope is non-empty.
    runtime
        The providers to attach; defaults to the core provider alone.
    observer
        A callback receiving every trace event as it is emitted.
    fuel
        The number of machine steps the run may take, or ``None`` for no
        bound. General recursion may diverge, and a bound is the only way
        a caller can be sure a run ends.

    Returns
    -------
    ExecutionResult
        The validated result value, its specialized type, the runtime label,
        and the full trace.

    Raises
    ------
    ExecutionFailure
        With code ``qiec-run-computation`` for an unknown name,
        ``qiec-run-static`` for a bad specialization, ``qiec-run-arity`` for
        a wrong argument count, ``qiec-run-provider`` when a provider cannot
        be built or attached, ``qiec-run-validator`` when no provider covers
        an argument or result type, ``qiec-run-argument`` or
        ``qiec-run-result`` when a value fails validation,
        ``qiec-run-module`` when the module fails validation,
        ``qiec-run-fuel`` when the step budget runs out, and
        ``qiec-run-evaluation`` when evaluation or a runtime clause fails.
    """

    selected = next(
        (
            candidate
            for candidate in module.computations
            if candidate.name == computation
        ),
        None,
    )
    if selected is None:
        available = ", ".join(item.name for item in module.computations) or "(none)"
        _fail(
            "qiec-run-computation",
            f"unknown QIEC computation {computation!r}; available: {available}",
            computation,
        )
    assert selected is not None
    config = runtime or RuntimeConfiguration()
    recorder = ExecutionTraceRecorder()
    sequence = 0

    def emit(event: str, detail: Mapping[str, object] | None = None) -> None:
        """Record one trace event and forward it to the observer.

        Parameters
        ----------
        event
            The stable event name.
        detail
            JSON-compatible event data; ``None`` records an empty mapping.
        """
        nonlocal sequence
        trace_event = ExecutionTraceEvent(sequence, event, selected.name, detail or {})
        sequence += 1
        recorder(trace_event)
        if observer is not None:
            observer(trace_event)

    emit("run.started", {"runtime": config.label})
    substitution = _specialize(selected, static_arguments)
    if static_arguments:
        emit(
            "run.specialized",
            {"arguments": [_render_static(argument) for argument in static_arguments]},
        )
    specialized_parameters, specialized_body = _specialize_computation(
        selected, substitution
    )
    specialized_result = substitute_type(selected.type.result, substitution)
    if len(arguments) != len(selected.parameters):
        _fail(
            "qiec-run-arity",
            f"computation {selected.name!r} expects {len(selected.parameters)} "
            f"value arguments, got {len(arguments)}",
            selected.name,
            selected.origin,
        )

    providers = config.instantiate()
    attachments = RuntimeAttachments()
    try:
        for provider in providers:
            provider.attach(module, attachments)
            emit("runtime.attached", {"provider": provider.name})
    except ExecutionFailure:
        raise
    except Exception as error:
        _fail(
            "qiec-run-provider",
            f"runtime provider failed while attaching: {error}",
            selected.name,
            selected.origin,
        )

    for position, (parameter, specialized_parameter, value) in enumerate(
        zip(selected.parameters, specialized_parameters, arguments, strict=True)
    ):
        expected = specialized_parameter.type
        validator = _validator_for(expected, providers)
        if validator is None:
            _fail(
                "qiec-run-validator",
                f"no configured runtime provider can validate argument "
                f"{parameter.name!r} at type {_render_type(expected)}",
                selected.name,
                selected.origin,
            )
        try:
            accepted = validator(value)
        except Exception as error:
            _fail(
                "qiec-run-argument",
                f"validator raised for argument {parameter.name!r}: {error}",
                selected.name,
                selected.origin,
            )
        if accepted is False:
            _fail(
                "qiec-run-argument",
                f"argument {position} ({parameter.name}) does not inhabit "
                f"{_render_type(expected)}: {value!r}",
                selected.name,
                selected.origin,
            )
        emit("argument.validated", {"name": parameter.name, "position": position})

    try:
        registry = validate_module(module)
    except Exception as error:
        _fail(
            "qiec-run-module",
            f"QIEC module failed validation before execution: {error}",
            selected.name,
            selected.origin,
        )
    environment = dict(zip(specialized_parameters, arguments, strict=True))

    def evaluator_trace(event: str, detail: Mapping[str, object]) -> None:
        """Forward an evaluator trace event into the invocation's trace.

        Parameters
        ----------
        event
            The stable event name the evaluator emitted.
        detail
            The event's data.
        """
        emit(event, detail)

    evaluator = Evaluator(
        attachments,
        trace_hook=evaluator_trace,
        module=module,
        fuel=fuel,
        type_validator=lambda type_: _validator_for(type_, providers),
    )
    try:
        value = evaluator.evaluate_checked(specialized_body, registry, environment)
    except ExecutionFailure:
        raise
    except FuelExhaustedError as error:
        emit("run.failed", {"error": type(error).__name__, "steps": error.steps})
        _fail(
            "qiec-run-fuel",
            f"computation {selected.name!r} did not finish within {error.steps} steps",
            selected.name,
            selected.origin,
        )
    except EvaluationError as error:
        emit("run.failed", {"error": type(error).__name__})
        _fail(
            "qiec-run-evaluation",
            f"computation {selected.name!r} failed: {error}",
            selected.name,
            selected.origin,
        )
    except Exception as error:
        emit("run.failed", {"error": type(error).__name__})
        _fail(
            "qiec-run-evaluation",
            f"runtime clause for computation {selected.name!r} failed: {error}",
            selected.name,
            selected.origin,
        )
    validator = _validator_for(specialized_result, providers)
    if validator is None:
        _fail(
            "qiec-run-validator",
            f"no configured runtime provider can validate result type "
            f"{_render_type(specialized_result)}",
            selected.name,
            selected.origin,
        )
    try:
        accepted = validator(value)
    except Exception as error:
        _fail(
            "qiec-run-result",
            f"result validator raised: {error}",
            selected.name,
            selected.origin,
        )
    if accepted is False:
        _fail(
            "qiec-run-result",
            f"result does not inhabit {_render_type(specialized_result)}: {value!r}",
            selected.name,
            selected.origin,
        )
    emit("run.completed", {"result_type": _render_type(specialized_result)})
    return ExecutionResult(
        selected.name,
        value,
        specialized_result,
        config.label,
        tuple(recorder.events),
    )