Checking

Validation and type and effect inference for the reference kernel.

checking

Validation and type/effect inference for the QIEC reference kernel.

NEVER module-attribute

NEVER = ResumptionUse(0, 0)

A computation that cannot invoke the continuation.

ONCE module-attribute

ONCE = ResumptionUse(1, 1)

A computation that invokes the continuation exactly once.

KernelError

KernelError(message: str, code: str = 'qiec-kind')

Bases: TypeError

A static QIEC kernel rejection.

Carries a stable diagnostic code alongside its message. The code is attached where the rejection is raised, because only that site knows which condition fired; classifying by message text afterwards is a guess that silently changes whenever the prose is reworded.

PARAMETER	DESCRIPTION
message	What was rejected, in prose, for a human reader. TYPE: str
code	A member of the published code set. Defaults to "qiec-kind", which is where a rejection with no more specific classification belongs. TYPE: str DEFAULT: 'qiec-kind'

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

def __init__(self, message: str, code: str = "qiec-kind") -> None:
    super().__init__(message)
    self.message = message
    self.code = code

StaticScope dataclass

StaticScope(types: tuple[tuple[str, Kind], ...] = (), indices: tuple[tuple[str, IndexSort], ...] = (), effects: tuple[str, ...] = ())

The three disjoint namespaces visible in a declaration body.

PARAMETER	DESCRIPTION
types	Type binders in scope, each with its kind. TYPE: tuple[tuple[str, Kind], ...] DEFAULT: ()
indices	Index binders in scope, each with its sort. TYPE: tuple[tuple[str, IndexSort], ...] DEFAULT: ()
effects	Effect binder names in scope. TYPE: tuple[str, ...] DEFAULT: ()

names property

names: frozenset[str]

Every static name bound here, across all three namespaces.

RETURNS	DESCRIPTION
frozenset[str]	The union of the type, index, and effect binder names. The namespaces are disjoint by construction, so this is what shadowing is tested against.

extend

extend(telescope: Telescope, *, subject: str) -> StaticScope

Return this scope widened by a telescope's binders.

PARAMETER	DESCRIPTION
telescope	Binders to add. Each must be a type, index, or effect binder. TYPE: Telescope
subject	What is being extended, named in any diagnostic so the error points at the declaration rather than at the scope. TYPE: str

RETURNS	DESCRIPTION
StaticScope	A scope carrying the original binders and the new ones.

RAISES	DESCRIPTION
KernelError	If a binder is of an unknown class, or if its name is already bound. Shadowing is rejected rather than resolved so a static name always denotes one thing within a declaration.

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

def extend(self, telescope: Telescope, *, subject: str) -> StaticScope:
    """Return this scope widened by a telescope's binders.

    Parameters
    ----------
    telescope : Telescope
        Binders to add. Each must be a type, index, or effect binder.
    subject : str
        What is being extended, named in any diagnostic so the error
        points at the declaration rather than at the scope.

    Returns
    -------
    StaticScope
        A scope carrying the original binders and the new ones.

    Raises
    ------
    KernelError
        If a binder is of an unknown class, or if its name is already
        bound. Shadowing is rejected rather than resolved so a static
        name always denotes one thing within a declaration.
    """
    types = list(self.types)
    indices = list(self.indices)
    effects = list(self.effects)
    seen = set(self.names)
    for binder in telescope:
        name = getattr(binder, "name", None)
        if not isinstance(name, str):  # pragma: no cover - closed binder union
            raise KernelError(f"unknown binder in {subject}: {binder!r}")
        if name in seen:
            raise KernelError(f"static binder {name!r} is shadowed in {subject}")
        seen.add(name)
        if isinstance(binder, TypeBinder):
            types.append((binder.name, binder.kind))
        elif isinstance(binder, IndexBinder):
            indices.append((binder.name, binder.sort))
        elif isinstance(binder, EffectBinder):
            effects.append(binder.name)
        else:  # pragma: no cover - closed binder union
            raise KernelError(f"unknown binder in {subject}: {binder!r}")
    return StaticScope(tuple(types), tuple(indices), tuple(effects))

type_kind

type_kind(name: str) -> Kind | None

The kind of a type binder in scope.

PARAMETER	DESCRIPTION
name	The binder name to resolve. TYPE: str

RETURNS	DESCRIPTION
Kind or None	The binder's kind, or None when name is not a type binder here. None does not mean unbound: the name may be an index or effect binder instead.

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

def type_kind(self, name: str) -> Kind | None:
    """The kind of a type binder in scope.

    Parameters
    ----------
    name : str
        The binder name to resolve.

    Returns
    -------
    Kind or None
        The binder's kind, or None when `name` is not a type binder
        here. None does not mean unbound: the name may be an index or
        effect binder instead.
    """
    return next((kind for candidate, kind in self.types if candidate == name), None)

index_sort

index_sort(name: str) -> IndexSort | None

The sort of an index binder in scope.

PARAMETER	DESCRIPTION
name	The binder name to resolve. TYPE: str

RETURNS	DESCRIPTION
IndexSort or None	The binder's sort, or None when name is not an index binder here.

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

def index_sort(self, name: str) -> IndexSort | None:
    """The sort of an index binder in scope.

    Parameters
    ----------
    name : str
        The binder name to resolve.

    Returns
    -------
    IndexSort or None
        The binder's sort, or None when `name` is not an index binder
        here.
    """
    return next(
        (sort for candidate, sort in self.indices if candidate == name), None
    )

ComputationSignature dataclass

ComputationSignature(id: ComputationId, name: str, telescope: Telescope, parameters: tuple[TypeExpr, ...], result: TypeExpr, effects: EffectRow)

What a call needs to know about a computation it invokes.

A signature rather than a declaration, and deliberately so: a call resolves against this, not against the callee's body. That is what lets every signature be collected before any body is checked, which in turn makes a forward call and mutual recursion ordinary rather than requiring a forward declaration.

PARAMETER	DESCRIPTION
id	The declaration's stable identity, which the call carries. TYPE: ComputationId
name	The authored name, used in diagnostics. TYPE: str
telescope	Static binders the computation takes. TYPE: Telescope
parameters	Value parameter types, in order, before instantiation. TYPE: tuple[TypeExpr, ...]
result	The result type before instantiation. TYPE: TypeExpr
effects	The row the computation performs, before instantiation. TYPE: EffectRow

KernelRegistry dataclass

KernelRegistry(families: dict[FamilyId, FamilyDecl] = dict(), constructors: dict[ConstructorId, ConstructorDecl] = dict(), effects: dict[EffectId, EffectDef] = dict(), operations: dict[OperationId, tuple[EffectDef, OperationDef]] = dict(), handlers: dict[HandlerId, HandlerDef] = dict(), type_constructors: dict[TypeId, TypeConstructorRef] = dict(), computations: dict[ComputationId, ComputationSignature] = dict(), instance_names: dict[EffectInstanceId, str] = dict())

Resolved declarations used by the reference checker.

Every table is keyed by stable identity, so a lookup never depends on a display name.

PARAMETER	DESCRIPTION
families	Indexed data families by identity. TYPE: dict[FamilyId, FamilyDecl] DEFAULT: dict()
constructors	Family constructors by identity. TYPE: dict[ConstructorId, ConstructorDecl] DEFAULT: dict()
effects	Effect interfaces by identity. TYPE: dict[EffectId, EffectDef] DEFAULT: dict()
operations	Each operation by identity, paired with the interface declaring it. TYPE: dict[OperationId, tuple[EffectDef, OperationDef]] DEFAULT: dict()
handlers	Handler declarations by identity. TYPE: dict[HandlerId, HandlerDef] DEFAULT: dict()
type_constructors	Type constructors by identity, recording the telescope each is known with. TYPE: dict[TypeId, TypeConstructorRef] DEFAULT: dict()
computations	Named computation signatures by identity. TYPE: dict[ComputationId, ComputationSignature] DEFAULT: dict()
instance_names	The source name of each lexical effect instance, for rendering rows in diagnostics. TYPE: dict[EffectInstanceId, str] DEFAULT: dict()

validate_static

validate_static(term: StaticArgument, scope: StaticScope | None = None) -> None

Validate a static term against every declaration known here.

PARAMETER	DESCRIPTION
term	The static term to check. TYPE: StaticArgument
scope	Binders in scope at the term's position. None means the empty scope, so any variable reference is out of scope. TYPE: StaticScope or None DEFAULT: None

RAISES	DESCRIPTION
KernelError	If the term is ill-kinded, references an unbound static variable, or applies a known declaration wrongly.

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

def validate_static(
    self,
    term: StaticArgument,
    scope: StaticScope | None = None,
) -> None:
    """Validate a static term against every declaration known here.

    Parameters
    ----------
    term : StaticArgument
        The static term to check.
    scope : StaticScope or None
        Binders in scope at the term's position. None means the empty
        scope, so any variable reference is out of scope.

    Raises
    ------
    KernelError
        If the term is ill-kinded, references an unbound static
        variable, or applies a known declaration wrongly.
    """
    check_static(term, scope, registry=self)
    self._record_type_constructors(term)

validate_type

validate_type(type_: TypeExpr, scope: StaticScope | None = None) -> None

Validate a value type against every declaration known here.

PARAMETER	DESCRIPTION
type_	The value type to check. TYPE: TypeExpr
scope	Binders in scope at the type's position. None means the empty scope. TYPE: StaticScope or None DEFAULT: None

RAISES	DESCRIPTION
KernelError	If the type is ill-kinded or references an unbound variable.

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

def validate_type(
    self,
    type_: TypeExpr,
    scope: StaticScope | None = None,
) -> None:
    """Validate a value type against every declaration known here.

    Parameters
    ----------
    type_ : TypeExpr
        The value type to check.
    scope : StaticScope or None
        Binders in scope at the type's position. None means the empty
        scope.

    Raises
    ------
    KernelError
        If the type is ill-kinded or references an unbound variable.
    """
    check_type(type_, scope, registry=self)
    self._record_type_constructors(type_)

validate_effect_row

validate_effect_row(row: EffectRow) -> None

Validate all concrete interface applications in an effect row.

PARAMETER	DESCRIPTION
row	The row whose entries to check. An open tail carries no application and so is not checked here. TYPE: EffectRow

RAISES	DESCRIPTION
KernelError	If an entry applies its interface wrongly.

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

def validate_effect_row(self, row: EffectRow) -> None:
    """Validate all concrete interface applications in an effect row.

    Parameters
    ----------
    row : EffectRow
        The row whose entries to check. An open tail carries no
        application and so is not checked here.

    Raises
    ------
    KernelError
        If an entry applies its interface wrongly.
    """
    for entry in row.entries:
        self.validate_static(entry.effect)

validate_computation_type

validate_computation_type(type_: ComputationType) -> None

Validate both strata of a computation type.

PARAMETER	DESCRIPTION
type_	The computation type whose effect row and result type to check. TYPE: ComputationType

RAISES	DESCRIPTION
KernelError	If either stratum is ill-formed.

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

def validate_computation_type(self, type_: ComputationType) -> None:
    """Validate both strata of a computation type.

    Parameters
    ----------
    type_ : ComputationType
        The computation type whose effect row and result type to
        check.

    Raises
    ------
    KernelError
        If either stratum is ill-formed.
    """
    self.validate_effect_row(type_.effects)
    self.validate_type(type_.result)

register_family

register_family(family: FamilyDecl) -> None

Record an indexed family declaration.

PARAMETER	DESCRIPTION
family	The family to register, with its uniform parameters and its refinable indices. TYPE: FamilyDecl

RAISES	DESCRIPTION
KernelError	If a family with this identity is already registered, if two of its binders share a name, or if its type constructor disagrees with one already recorded.

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

def register_family(self, family: FamilyDecl) -> None:
    """Record an indexed family declaration.

    Parameters
    ----------
    family : FamilyDecl
        The family to register, with its uniform parameters and its
        refinable indices.

    Raises
    ------
    KernelError
        If a family with this identity is already registered, if two
        of its binders share a name, or if its type constructor
        disagrees with one already recorded.
    """
    if family.id in self.families:
        raise KernelError(f"family already registered: {family.name!r}")
    StaticScope().extend(
        (*family.parameters, *family.indices),
        subject=f"family {family.name!r}",
    )
    self._record_type_constructor(family.type_constructor)
    self.families[family.id] = family

register_constructor

register_constructor(constructor: ConstructorDecl) -> None

Record a constructor against the family that declares it.

PARAMETER	DESCRIPTION
constructor	The constructor to register. Its family must already be registered and must list this constructor. TYPE: ConstructorDecl

RAISES	DESCRIPTION
KernelError	If the constructor is already registered, if its family is unknown or does not declare it, if it returns the wrong number of result indices, if its telescope shadows a family binder, or if a field type or result index is ill-formed.

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

def register_constructor(self, constructor: ConstructorDecl) -> None:
    """Record a constructor against the family that declares it.

    Parameters
    ----------
    constructor : ConstructorDecl
        The constructor to register. Its family must already be
        registered and must list this constructor.

    Raises
    ------
    KernelError
        If the constructor is already registered, if its family is
        unknown or does not declare it, if it returns the wrong
        number of result indices, if its telescope shadows a family
        binder, or if a field type or result index is ill-formed.
    """
    if constructor.id in self.constructors:
        raise KernelError(f"constructor already registered: {constructor.name!r}")
    family = self.families.get(constructor.family)
    if family is None:
        raise KernelError(f"unknown family for constructor {constructor.name!r}")
    if constructor.id not in family.constructors:
        raise KernelError(
            f"constructor {constructor.name!r} is not declared by family {family.name!r}"
        )
    if len(constructor.result_indices) != len(family.indices):
        raise KernelError(
            f"constructor {constructor.name!r} returns "
            f"{len(constructor.result_indices)} indices; expected {len(family.indices)}"
        )
    family_names = {binder.name for binder in (*family.parameters, *family.indices)}
    constructor_names = {binder.name for binder in constructor.telescope}
    shadowed = family_names & constructor_names
    if shadowed:
        raise KernelError(
            f"constructor {constructor.name!r} shadows family binders: {shadowed!r}"
        )
    scope = (
        StaticScope()
        .extend(
            family.parameters,
            subject=f"family parameters of {family.name!r}",
        )
        .extend(
            constructor.telescope,
            subject=f"constructor {constructor.name!r}",
        )
    )
    for field_definition in constructor.fields:
        self.validate_type(field_definition.type, scope)
    for result, binder in zip(
        constructor.result_indices,
        family.indices,
        strict=True,
    ):
        self.validate_static(result, scope)
        _check_binder_argument(
            binder,
            result,
            subject=f"result index of constructor {constructor.name!r}",
        )
    self.constructors[constructor.id] = constructor

register_effect

register_effect(effect: EffectDef) -> None

Record an effect interface and index its operations.

Each operation's signature is checked in a scope carrying the interface binders and then its own, and may mention the interface being declared.

PARAMETER	DESCRIPTION
effect	The interface to register. TYPE: EffectDef

RAISES	DESCRIPTION
KernelError	If the interface or one of its operations is already registered, if a binder is shadowed, or if an argument or result type is ill-formed.

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

def register_effect(self, effect: EffectDef) -> None:
    """Record an effect interface and index its operations.

    Each operation's signature is checked in a scope carrying the
    interface binders and then its own, and may mention the interface
    being declared.

    Parameters
    ----------
    effect : EffectDef
        The interface to register.

    Raises
    ------
    KernelError
        If the interface or one of its operations is already
        registered, if a binder is shadowed, or if an argument or
        result type is ill-formed.
    """
    if effect.ref.id in self.effects:
        raise KernelError(f"effect already registered: {effect.ref.name!r}")
    interface_scope = StaticScope().extend(
        effect.telescope,
        subject=f"effect {effect.ref.name!r}",
    )
    for operation in effect.operations:
        if operation.id in self.operations:
            raise KernelError(f"operation already registered: {operation.name!r}")
        operation_scope = interface_scope.extend(
            operation.telescope,
            subject=f"operation {operation.name!r}",
        )
        for argument in operation.arguments:
            self.validate_type(argument.type, operation_scope)
            self._validate_effect_references(argument.type, (effect,))
        self.validate_type(operation.result_type, operation_scope)
        self._validate_effect_references(operation.result_type, (effect,))
    self.effects[effect.ref.id] = effect
    for operation in effect.operations:
        self.operations[operation.id] = (effect, operation)

register_handler

register_handler(handler: HandlerDef) -> None

Record a handler against the interface it handles.

Coverage is checked here rather than at construction, because it is a claim about the interface's operations and only the registry knows those.

PARAMETER	DESCRIPTION
handler	The handler signature to register. TYPE: HandlerDef

RAISES

DESCRIPTION

KernelError

If the handler is already registered, if its telescope shadows a binder, if its input, output, or introduced row is ill-formed, if the interface it names is unknown or wrongly applied, if a clause covers an operation the interface does not declare, or if the handler claims totality while leaving an operation uncovered, or if an authored clause body fails to check, answers the wrong type, or resumes more often than its grade allows.

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

def register_handler(self, handler: HandlerDef) -> None:
    """Record a handler against the interface it handles.

    Coverage is checked here rather than at construction, because it
    is a claim about the interface's operations and only the registry
    knows those.

    Parameters
    ----------
    handler : HandlerDef
        The handler signature to register.

    Raises
    ------
    KernelError
        If the handler is already registered, if its telescope
        shadows a binder, if its input, output, or introduced row is
        ill-formed, if the interface it names is unknown or wrongly
        applied, if a clause covers an operation the interface does
        not declare, or if the handler claims totality while leaving
        an operation uncovered, or if an authored clause body fails
        to check, answers the wrong type, or resumes more often than
        its grade allows.
    """
    if handler.id in self.handlers:
        raise KernelError(f"handler already registered: {handler.name!r}")
    scope = StaticScope().extend(
        handler.telescope,
        subject=f"handler {handler.name!r}",
    )
    self.validate_static(handler.effect, scope)
    self.validate_type(handler.input_type, scope)
    self.validate_type(handler.output_type, scope)
    for entry in handler.introduced.entries:
        self.validate_static(entry.effect, scope)
        introduced = self.effects.get(entry.effect.id)
        if introduced is None or not introduced.matches(entry.effect):
            raise KernelError(
                f"handler {handler.name!r} introduces an unknown effect interface"
            )
    effect = self.effects.get(handler.effect.id)
    if effect is None or not effect.matches(handler.effect):
        raise KernelError(f"unknown effect interface for handler {handler.name!r}")
    declared = {operation.id for operation in effect.operations}
    clauses = {clause.operation for clause in handler.clauses}
    unknown = clauses - declared
    if unknown:
        raise KernelError(
            f"handler has clauses for unknown operations: {unknown!r}"
        )
    if handler.total and clauses != declared:
        missing = declared - clauses
        raise KernelError(f"total handler is missing operations: {missing!r}")
    self._check_handler_bodies(
        handler,
        effect,
        substitution=instantiate_telescope(
            effect.telescope, handler.effect.arguments
        ),
    )
    self.handlers[handler.id] = handler

register_computation

register_computation(signature: ComputationSignature) -> None

Record a computation's signature so calls can resolve to it.

Signatures are registered before any body is checked. A call therefore resolves whether the callee is declared above or below it, and mutual recursion needs no forward declaration.

PARAMETER	DESCRIPTION
signature	The signature to record. TYPE: ComputationSignature

RAISES	DESCRIPTION
KernelError	If a computation with this identity is already registered, if its telescope shadows a binder, or if a parameter type, the result type, or the declared row is ill-formed. A duplicate identity is rejected rather than overwritten, since a call carrying that identity would otherwise resolve to whichever declaration happened to register last.

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

def register_computation(self, signature: ComputationSignature) -> None:
    """Record a computation's signature so calls can resolve to it.

    Signatures are registered before any body is checked. A call
    therefore resolves whether the callee is declared above or below
    it, and mutual recursion needs no forward declaration.

    Parameters
    ----------
    signature : ComputationSignature
        The signature to record.

    Raises
    ------
    KernelError
        If a computation with this identity is already registered, if
        its telescope shadows a binder, or if a parameter type, the
        result type, or the declared row is ill-formed. A duplicate
        identity is rejected rather than overwritten, since a call
        carrying that identity would otherwise resolve to whichever
        declaration happened to register last.
    """
    if signature.id in self.computations:
        raise KernelError(f"computation already registered: {signature.name!r}")
    scope = StaticScope().extend(
        signature.telescope,
        subject=f"computation {signature.name!r}",
    )
    for parameter in signature.parameters:
        self.validate_type(parameter, scope)
    self.validate_type(signature.result, scope)
    self.validate_effect_row(signature.effects)
    self.computations[signature.id] = signature

computation

computation(computation: ComputationId) -> ComputationSignature

The registered signature with a given identity.

PARAMETER	DESCRIPTION
computation	The identity to resolve. TYPE: ComputationId

RETURNS	DESCRIPTION
ComputationSignature	The registered signature.

RAISES	DESCRIPTION
KernelError	If no computation is registered under that identity, which is how a call to an undeclared name is caught.

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

def computation(self, computation: ComputationId) -> ComputationSignature:
    """The registered signature with a given identity.

    Parameters
    ----------
    computation : ComputationId
        The identity to resolve.

    Returns
    -------
    ComputationSignature
        The registered signature.

    Raises
    ------
    KernelError
        If no computation is registered under that identity, which is
        how a call to an undeclared name is caught.
    """
    try:
        return self.computations[computation]
    except KeyError as exc:
        raise KernelError(
            f"unknown computation {computation}", "qiec-call"
        ) from exc

constructor

constructor(constructor: ConstructorId) -> ConstructorDecl

The registered constructor with a given identity.

PARAMETER	DESCRIPTION
constructor	The identity to resolve. TYPE: ConstructorId

RETURNS	DESCRIPTION
ConstructorDecl	The registered declaration.

RAISES	DESCRIPTION
KernelError	If no constructor is registered under that identity. The underlying KeyError is wrapped so callers handle one kernel error class rather than two.

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

def constructor(self, constructor: ConstructorId) -> ConstructorDecl:
    """The registered constructor with a given identity.

    Parameters
    ----------
    constructor : ConstructorId
        The identity to resolve.

    Returns
    -------
    ConstructorDecl
        The registered declaration.

    Raises
    ------
    KernelError
        If no constructor is registered under that identity. The
        underlying `KeyError` is wrapped so callers handle one
        kernel error class rather than two.
    """
    try:
        return self.constructors[constructor]
    except KeyError as exc:
        raise KernelError(f"unknown constructor {constructor}") from exc

handler

handler(handler: HandlerId) -> HandlerDef

The registered handler with a given identity.

PARAMETER	DESCRIPTION
handler	The identity to resolve. TYPE: HandlerId

RETURNS	DESCRIPTION
HandlerDef	The registered signature.

RAISES	DESCRIPTION
KernelError	If no handler is registered under that identity.

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

def handler(self, handler: HandlerId) -> HandlerDef:
    """The registered handler with a given identity.

    Parameters
    ----------
    handler : HandlerId
        The identity to resolve.

    Returns
    -------
    HandlerDef
        The registered signature.

    Raises
    ------
    KernelError
        If no handler is registered under that identity.
    """
    try:
        return self.handlers[handler]
    except KeyError as exc:
        raise KernelError(f"unknown handler {handler}") from exc

operation

operation(operation: OperationId) -> tuple[EffectDef, OperationDef]

The operation with a given identity, and its owning interface.

PARAMETER	DESCRIPTION
operation	The identity to resolve. TYPE: OperationId

RETURNS	DESCRIPTION
tuple[EffectDef, OperationDef]	The interface that declares the operation, and the operation itself. The interface comes back too because checking a request needs both.

RAISES	DESCRIPTION
KernelError	If no operation is registered under that identity.

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

def operation(self, operation: OperationId) -> tuple[EffectDef, OperationDef]:
    """The operation with a given identity, and its owning interface.

    Parameters
    ----------
    operation : OperationId
        The identity to resolve.

    Returns
    -------
    tuple[EffectDef, OperationDef]
        The interface that declares the operation, and the operation
        itself. The interface comes back too because checking a
        request needs both.

    Raises
    ------
    KernelError
        If no operation is registered under that identity.
    """
    try:
        return self.operations[operation]
    except KeyError as exc:
        raise KernelError(f"unknown operation {operation}") from exc

family_for_type

family_for_type(type_: TypeApplication) -> FamilyDecl

The indexed family a type application is an instance of.

PARAMETER	DESCRIPTION
type_	The applied type whose family to find. TYPE: TypeApplication

RETURNS	DESCRIPTION
FamilyDecl	The family whose type constructor the application names.

RAISES	DESCRIPTION
KernelError	If the application is ill-formed, or names a constructor no registered family declares, which is how a case analysis on a non-family type is rejected.

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

def family_for_type(self, type_: TypeApplication) -> FamilyDecl:
    """The indexed family a type application is an instance of.

    Parameters
    ----------
    type_ : TypeApplication
        The applied type whose family to find.

    Returns
    -------
    FamilyDecl
        The family whose type constructor the application names.

    Raises
    ------
    KernelError
        If the application is ill-formed, or names a constructor no
        registered family declares, which is how a case analysis on
        a non-family type is rejected.
    """
    self.validate_static(type_)
    for family in self.families.values():
        if family.type_constructor == type_.constructor:
            return family
    raise KernelError(f"unknown indexed family type {type_.constructor.name!r}")

ResumptionType dataclass

ResumptionType(result: TypeExpr, answer: TypeExpr, effects: EffectRow)

The continuation available inside one handler clause body.

A resumption is not a value and cannot be stored, so it is not a local binding. It is a capability of the position instead: available inside a clause body and nowhere else.

PARAMETER	DESCRIPTION
result	What the resumed operation supplies, so a resume must carry a value of this type. TYPE: TypeExpr
answer	What resuming produces, which is the clause's own answer type. TYPE: TypeExpr
effects	The row resuming performs. TYPE: EffectRow

CheckContext dataclass

CheckContext(locals: tuple[Local, ...] = (), givens: tuple[BranchGiven, ...] = (), static_scopes: tuple[StaticScopeId, ...] = (), static_variables: tuple[StaticVariableId, ...] = (), resumption: ResumptionType | None = None)

What is in scope at one point in a term being checked.

PARAMETER	DESCRIPTION
locals	Value bindings, innermost last. TYPE: tuple[Local, ...] DEFAULT: ()
givens	Equality evidence introduced by enclosing case branches, which is what makes an index refinement usable in the branch body. TYPE: tuple[BranchGiven, ...] DEFAULT: ()
static_scopes	Static scopes already entered. Tracked so a branch cannot reuse an enclosing scope's identity and silently capture its variables. TYPE: tuple[StaticScopeId, ...] DEFAULT: ()
static_variables	Rigid variables currently active, tracked for the same reason. TYPE: tuple[StaticVariableId, ...] DEFAULT: ()

resumption class-attribute instance-attribute

resumption: ResumptionType | None = None

The continuation, when checking a handler clause body.

None everywhere else, which is what makes resume outside a clause a checker error rather than a term with no meaning.

extend

extend(local: Local) -> CheckContext

Return this context with one more value binding.

PARAMETER	DESCRIPTION
local	The binding to add. TYPE: Local

RETURNS	DESCRIPTION
CheckContext	A context carrying the new binding innermost.

RAISES	DESCRIPTION
KernelError	If the name is already bound. Rebinding is rejected rather than shadowed, so a name denotes one value throughout a body.

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

def extend(self, local: Local) -> CheckContext:
    """Return this context with one more value binding.

    Parameters
    ----------
    local : Local
        The binding to add.

    Returns
    -------
    CheckContext
        A context carrying the new binding innermost.

    Raises
    ------
    KernelError
        If the name is already bound. Rebinding is rejected rather
        than shadowed, so a name denotes one value throughout a body.
    """
    if any(existing.name == local.name for existing in self.locals):
        raise KernelError(f"local already bound: {local.name!r}")
    return CheckContext(
        (*self.locals, local),
        self.givens,
        self.static_scopes,
        self.static_variables,
        self.resumption,
    )

with_givens

with_givens(givens: tuple[BranchGiven, ...]) -> CheckContext

Return this context with further equality evidence in scope.

PARAMETER	DESCRIPTION
givens	Evidence a case branch introduces by matching a constructor. TYPE: tuple[BranchGiven, ...]

RETURNS	DESCRIPTION
CheckContext	A context in which that evidence may be appealed to.

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

def with_givens(self, givens: tuple[BranchGiven, ...]) -> CheckContext:
    """Return this context with further equality evidence in scope.

    Parameters
    ----------
    givens : tuple[BranchGiven, ...]
        Evidence a case branch introduces by matching a constructor.

    Returns
    -------
    CheckContext
        A context in which that evidence may be appealed to.
    """
    return CheckContext(
        self.locals,
        (*self.givens, *givens),
        self.static_scopes,
        self.static_variables,
        self.resumption,
    )

with_static_scope

with_static_scope(scope: StaticScopeId, variables: tuple[StaticVariableId, ...] = ()) -> CheckContext

Return this context inside a fresh static scope.

PARAMETER	DESCRIPTION
scope	Identity of the scope being entered. TYPE: StaticScopeId
variables	Rigid variables the scope introduces. TYPE: tuple[StaticVariableId, ...] DEFAULT: ()

RETURNS	DESCRIPTION
CheckContext	A context recording the scope and its variables as active.

RAISES	DESCRIPTION
KernelError	If the scope or one of its variables is already active. Reuse would let a branch's rigid variable be confused with an enclosing one, which is what keeps branch refinements from leaking out of the branch.

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

def with_static_scope(
    self,
    scope: StaticScopeId,
    variables: tuple[StaticVariableId, ...] = (),
) -> CheckContext:
    """Return this context inside a fresh static scope.

    Parameters
    ----------
    scope : StaticScopeId
        Identity of the scope being entered.
    variables : tuple[StaticVariableId, ...]
        Rigid variables the scope introduces.

    Returns
    -------
    CheckContext
        A context recording the scope and its variables as active.

    Raises
    ------
    KernelError
        If the scope or one of its variables is already active. Reuse
        would let a branch's rigid variable be confused with an
        enclosing one, which is what keeps branch refinements from
        leaking out of the branch.
    """
    if scope in self.static_scopes:
        raise KernelError("a case branch reused an enclosing static scope")
    if set(variables) & set(self.static_variables):
        raise KernelError("a case branch reused an active static variable")
    return CheckContext(
        self.locals,
        self.givens,
        (*self.static_scopes, scope),
        (*self.static_variables, *variables),
        self.resumption,
    )

with_resumption

with_resumption(resumption: ResumptionType) -> CheckContext

Return this context inside a handler clause body.

PARAMETER	DESCRIPTION
resumption	The continuation the clause body may invoke. TYPE: ResumptionType

RETURNS	DESCRIPTION
CheckContext	A context in which resume is available. A clause body nested inside another handler's clause replaces the outer resumption rather than stacking, since resume names the innermost clause's continuation and there is no syntax for reaching past it.

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

def with_resumption(self, resumption: ResumptionType) -> CheckContext:
    """Return this context inside a handler clause body.

    Parameters
    ----------
    resumption : ResumptionType
        The continuation the clause body may invoke.

    Returns
    -------
    CheckContext
        A context in which `resume` is available. A clause body
        nested inside another handler's clause replaces the outer
        resumption rather than stacking, since `resume` names the
        innermost clause's continuation and there is no syntax for
        reaching past it.
    """
    return CheckContext(
        self.locals,
        self.givens,
        self.static_scopes,
        self.static_variables,
        resumption,
    )

local_type

local_type(name: str) -> TypeExpr | None

The type of a value binding in scope.

PARAMETER	DESCRIPTION
name	The bound name to resolve. TYPE: str

RETURNS	DESCRIPTION
TypeExpr or None	The binding's type, or None when the name is not bound here. The innermost binding wins, though extend rejects rebinding, so at most one can match.

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

def local_type(self, name: str) -> TypeExpr | None:
    """The type of a value binding in scope.

    Parameters
    ----------
    name : str
        The bound name to resolve.

    Returns
    -------
    TypeExpr or None
        The binding's type, or None when the name is not bound here.
        The innermost binding wins, though `extend` rejects rebinding,
        so at most one can match.
    """
    return next(
        (local.type for local in reversed(self.locals) if local.name == name),
        None,
    )

given

given(equality: EqualityId) -> BranchGiven | None

The equality evidence with a given identity, if in scope.

PARAMETER	DESCRIPTION
equality	The identity to resolve. TYPE: EqualityId

RETURNS	DESCRIPTION
BranchGiven or None	The evidence, or None when it is not in scope, which is how an appeal to a refinement outside its branch is caught.

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

def given(self, equality: EqualityId) -> BranchGiven | None:
    """The equality evidence with a given identity, if in scope.

    Parameters
    ----------
    equality : EqualityId
        The identity to resolve.

    Returns
    -------
    BranchGiven or None
        The evidence, or None when it is not in scope, which is how
        an appeal to a refinement outside its branch is caught.
    """
    return next((given for given in self.givens if given.id == equality), None)

ResumptionUse dataclass

ResumptionUse(minimum: int, maximum: int | None)

How many times a clause body may invoke its continuation.

An interval rather than a count, because a case splits the paths through a body and different branches may resume different numbers of times. minimum is what every path does at least, maximum what some path does at most.

PARAMETER	DESCRIPTION
minimum	The fewest invocations on any path. TYPE: int
maximum	The most on any path, or None for unbounded. None is the top of the lattice and is used wherever the analysis cannot bound the count, so an unbounded use is never mistaken for a small one. TYPE: int or None

then

then(other: ResumptionUse) -> ResumptionUse

Sequential composition: both run, so the counts add.

PARAMETER	DESCRIPTION
other	What runs after this. TYPE: ResumptionUse

RETURNS	DESCRIPTION
ResumptionUse	The combined use. Unbounded on either side stays unbounded.

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

def then(self, other: ResumptionUse) -> ResumptionUse:
    """Sequential composition: both run, so the counts add.

    Parameters
    ----------
    other : ResumptionUse
        What runs after this.

    Returns
    -------
    ResumptionUse
        The combined use. Unbounded on either side stays unbounded.
    """
    maximum = (
        None
        if self.maximum is None or other.maximum is None
        else self.maximum + other.maximum
    )
    return ResumptionUse(self.minimum + other.minimum, maximum)

join

join(other: ResumptionUse) -> ResumptionUse

Alternation: one path or the other, so the interval widens.

PARAMETER	DESCRIPTION
other	The alternative path's use. TYPE: ResumptionUse

RETURNS	DESCRIPTION
ResumptionUse	The interval covering both. The minimum drops to the smaller because some path now does that few, and the maximum rises to the larger for the same reason.

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

def join(self, other: ResumptionUse) -> ResumptionUse:
    """Alternation: one path or the other, so the interval widens.

    Parameters
    ----------
    other : ResumptionUse
        The alternative path's use.

    Returns
    -------
    ResumptionUse
        The interval covering both. The minimum drops to the smaller
        because some path now does that few, and the maximum rises to
        the larger for the same reason.
    """
    maximum = (
        None
        if self.maximum is None or other.maximum is None
        else max(self.maximum, other.maximum)
    )
    return ResumptionUse(min(self.minimum, other.minimum), maximum)

check_static

check_static(term: StaticArgument, scope: StaticScope | None = None, *, registry: KernelRegistry | None = None) -> None

Check the intrinsic kind/sort structure of one static term.

PARAMETER	DESCRIPTION
term	The static term to check. TYPE: StaticArgument
scope	Binders in scope. None checks the term's intrinsic structure alone and permits no variable reference, which is the right mode for a closed declaration. TYPE: StaticScope or None DEFAULT: None
registry	Registry to resolve interface applications against. None skips that check, leaving only the structural one. TYPE: KernelRegistry or None DEFAULT: None

RAISES	DESCRIPTION
KernelError	If the term is ill-kinded, references an unbound or wrongly kinded static variable, carries a scoped variable where a declaration is expected, or applies a known interface wrongly.

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

def check_static(
    term: StaticArgument,
    scope: StaticScope | None = None,
    *,
    registry: KernelRegistry | None = None,
) -> None:
    """Check the intrinsic kind/sort structure of one static term.

    Parameters
    ----------
    term : StaticArgument
        The static term to check.
    scope : StaticScope or None
        Binders in scope. None checks the term's intrinsic structure
        alone and permits no variable reference, which is the right mode
        for a closed declaration.
    registry : KernelRegistry or None
        Registry to resolve interface applications against. None skips
        that check, leaving only the structural one.

    Raises
    ------
    KernelError
        If the term is ill-kinded, references an unbound or wrongly
        kinded static variable, carries a scoped variable where a
        declaration is expected, or applies a known interface wrongly.
    """
    if registry is not None:
        registry._validate_effect_references(term)
    if isinstance(term, TypeVariable):
        if scope is not None:
            if term.identity is not None:
                raise KernelError(
                    "a scoped type variable cannot occur in a declaration"
                )
            bound_kind = scope.type_kind(term.name)
            if bound_kind is None:
                raise KernelError(f"unbound type variable {term.name!r}")
            if bound_kind != term.kind:
                raise KernelError(f"type variable {term.name!r} has the wrong kind")
        return
    if isinstance(term, TypeApplication):
        if len(term.constructor.telescope) != len(term.arguments):
            raise KernelError(
                f"type constructor {term.constructor.name!r} expects "
                f"{len(term.constructor.telescope)} static arguments"
            )
        for binder, argument in zip(
            term.constructor.telescope,
            term.arguments,
            strict=True,
        ):
            _check_binder_argument(
                binder,
                argument,
                subject=f"argument of type constructor {term.constructor.name!r}",
            )
            check_static(argument, scope)
        return
    if isinstance(term, FunctionType):
        check_type(term.parameter, scope)
        check_type(term.result, scope)
        return
    if isinstance(term, EqualityType):
        check_static(term.left, scope)
        check_static(term.right, scope)
        if isinstance(term.kind, (NatSort, ShapeSort, ContextSort, UserIndexSort)):
            index_nodes = (IndexVariable, IndexLiteral, IndexConstructor, ShapeIndex)
            if not isinstance(term.left, index_nodes) or not isinstance(
                term.right, index_nodes
            ):
                raise KernelError("index equality endpoints must be index terms")
            if not _sort_matches(term.kind, index_sort(term.left)):  # type: ignore[arg-type]
                raise KernelError("left equality endpoint has the wrong index sort")
            if not _sort_matches(term.kind, index_sort(term.right)):  # type: ignore[arg-type]
                raise KernelError("right equality endpoint has the wrong index sort")
        else:
            if (
                static_kind(term.left) != term.kind
                or static_kind(term.right) != term.kind
            ):
                raise KernelError("equality endpoints have the wrong kind")
        return
    if isinstance(term, EffectRef):
        for argument in term.arguments:
            check_static(argument, scope)
        return
    if isinstance(term, EffectVariable):
        if scope is not None:
            if term.identity is not None:
                raise KernelError(
                    "a scoped effect variable cannot occur in a declaration"
                )
            if term.name not in scope.effects:
                raise KernelError(f"unbound effect variable {term.name!r}")
        return
    if isinstance(term, IndexConstructor):
        for argument in term.arguments:
            check_static(argument, scope)
        return
    if isinstance(term, ShapeIndex):
        index_sort(term)
        for dimension in term.dimensions:
            check_static(dimension, scope)
        return
    if isinstance(term, IndexVariable):
        if scope is not None:
            if term.identity is not None:
                raise KernelError(
                    "a scoped index variable cannot occur in a declaration"
                )
            bound_sort = scope.index_sort(term.name)
            if bound_sort is None:
                raise KernelError(f"unbound index variable {term.name!r}")
            if not _sort_matches(bound_sort, term.sort):
                raise KernelError(f"index variable {term.name!r} has the wrong sort")
        return
    if isinstance(term, IndexLiteral):
        return
    raise KernelError(f"unknown static term {term!r}")

check_type

check_type(type_: TypeExpr, scope: StaticScope | None = None, *, registry: KernelRegistry | None = None) -> None

Check that a type expression is intrinsically well-kinded.

PARAMETER	DESCRIPTION
type_	The type expression to check. TYPE: TypeExpr
scope	Binders in scope, or None for a closed type. TYPE: StaticScope or None DEFAULT: None
registry	Registry to resolve interface applications against, or None. TYPE: KernelRegistry or None DEFAULT: None

RAISES	DESCRIPTION
KernelError	If the expression is ill-kinded, or is a well-formed static term that is not a type, such as an index where a type is required.

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

def check_type(
    type_: TypeExpr,
    scope: StaticScope | None = None,
    *,
    registry: KernelRegistry | None = None,
) -> None:
    """Check that a type expression is intrinsically well-kinded.

    Parameters
    ----------
    type_ : TypeExpr
        The type expression to check.
    scope : StaticScope or None
        Binders in scope, or None for a closed type.
    registry : KernelRegistry or None
        Registry to resolve interface applications against, or None.

    Raises
    ------
    KernelError
        If the expression is ill-kinded, or is a well-formed static term
        that is not a type, such as an index where a type is required.
    """
    check_static(type_, scope, registry=registry)
    if static_kind(type_) != TYPE:
        raise KernelError(f"expected a type, got {render_static(type_)}")

check_evidence

check_evidence(evidence: EqualityEvidence, context: CheckContext) -> EqualityType

Check an appeal to equality evidence and return what it proves.

PARAMETER	DESCRIPTION
evidence	The appeal being made, either reflexivity or an appeal to a branch given. TYPE: EqualityEvidence
context	Scope the appeal is made in, which decides whether a branch given is still available. TYPE: CheckContext

RETURNS	DESCRIPTION
EqualityType	The equality the evidence establishes.

RAISES	DESCRIPTION
KernelError	If the stated equality is ill-kinded, if reflexivity is claimed between unequal endpoints, if a branch given is appealed to outside its branch or with an equality other than the one it carries, or if the evidence is of an unknown class.

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

def check_evidence(evidence: EqualityEvidence, context: CheckContext) -> EqualityType:
    """Check an appeal to equality evidence and return what it proves.

    Parameters
    ----------
    evidence : EqualityEvidence
        The appeal being made, either reflexivity or an appeal to a
        branch given.
    context : CheckContext
        Scope the appeal is made in, which decides whether a branch given
        is still available.

    Returns
    -------
    EqualityType
        The equality the evidence establishes.

    Raises
    ------
    KernelError
        If the stated equality is ill-kinded, if reflexivity is claimed
        between unequal endpoints, if a branch given is appealed to
        outside its branch or with an equality other than the one it
        carries, or if the evidence is of an unknown class.
    """
    check_type(evidence.equality)
    if isinstance(evidence, Reflexivity):
        if evidence.equality.left != evidence.equality.right:
            raise KernelError("reflexivity evidence has unequal endpoints")
        return evidence.equality
    if isinstance(evidence, BranchGiven):
        available = context.given(evidence.id)
        if available is None or available.equality != evidence.equality:
            raise KernelError("branch equality evidence escaped or was forged")
        return evidence.equality
    raise KernelError(f"unknown equality evidence {evidence!r}")

infer_value

infer_value(value: Value, registry: KernelRegistry, context: CheckContext = CheckContext()) -> TypeExpr

Infer the type of a pure value term.

PARAMETER	DESCRIPTION
value	The value to check. TYPE: Value
registry	Declarations the value's types are resolved against. TYPE: KernelRegistry
context	Bindings and evidence in scope. The default empty context suits a closed value. TYPE: CheckContext DEFAULT: CheckContext()

RETURNS	DESCRIPTION
TypeExpr	The inferred type.

RAISES	DESCRIPTION
KernelError	If a local is unbound or carries a type other than the one the term claims, if a literal's value does not inhabit its type, if a constructor is applied wrongly, if equality evidence is forged or out of scope, or if an inferred type mentions a rigid variable that escapes its branch.

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

def infer_value(
    value: Value,
    registry: KernelRegistry,
    context: CheckContext = CheckContext(),
) -> TypeExpr:
    """Infer the type of a pure value term.

    Parameters
    ----------
    value : Value
        The value to check.
    registry : KernelRegistry
        Declarations the value's types are resolved against.
    context : CheckContext
        Bindings and evidence in scope. The default empty context suits a
        closed value.

    Returns
    -------
    TypeExpr
        The inferred type.

    Raises
    ------
    KernelError
        If a local is unbound or carries a type other than the one the
        term claims, if a literal's value does not inhabit its type, if
        a constructor is applied wrongly, if equality evidence is forged
        or out of scope, or if an inferred type mentions a rigid variable
        that escapes its branch.
    """
    if isinstance(value, Var):
        actual = context.local_type(value.local.name)
        if actual is None or actual != value.local.type:
            raise KernelError(f"unbound or mistyped local {value.local.name!r}")
        registry.validate_type(actual)
        _check_static_variable_scope(actual, context, subject="local type")
        return actual
    if isinstance(value, LiteralValue):
        registry.validate_type(value.type)
        _check_static_variable_scope(value.type, context, subject="literal type")
        _check_literal(value.value, value.type)
        return value.type
    if isinstance(value, AttachmentRef):
        registry.validate_type(value.type)
        _check_static_variable_scope(value.type, context, subject="attachment type")
        return value.type
    if isinstance(value, EvidenceValue):
        equality = check_evidence(value.evidence, context)
        registry.validate_type(equality)
        _check_static_variable_scope(equality, context, subject="equality evidence")
        return equality
    if isinstance(value, TransportValue):
        equality = check_evidence(value.evidence, context)
        registry.validate_type(equality)
        _check_static_variable_scope(equality, context, subject="transport equality")
        source = infer_value(value.value, registry, context)
        registry.validate_type(value.target_type)
        _check_static_variable_scope(
            value.target_type,
            context,
            subject="transport target",
        )
        if equality.left != source or equality.right != value.target_type:
            raise KernelError("transport endpoints do not match value and target types")
        return value.target_type
    if isinstance(value, PrimitiveApplication):
        signature = PRIMITIVES.get(value.name)
        if signature is None:
            raise KernelError(f"unknown primitive {value.name!r}", "qiec-primitive")
        if signature.id != value.primitive:
            raise KernelError(
                f"primitive {value.name!r} is applied under the identity of "
                "another primitive",
                "qiec-primitive",
            )
        if len(value.arguments) != len(signature.parameters):
            raise KernelError(
                f"primitive {value.name!r} takes {len(signature.parameters)} "
                f"argument(s), got {len(value.arguments)}",
                "qiec-primitive",
            )
        broadcast: tuple[IndexTerm, ...] | None = None
        for position, (argument, expected) in enumerate(
            zip(value.arguments, signature.parameters, strict=True)
        ):
            actual = infer_value(argument, registry, context)
            if actual == expected:
                continue
            shape = tensor_shape(actual)
            if shape is None or shape[0] != expected:
                raise KernelError(
                    f"argument {position} of primitive {value.name!r} has type "
                    f"{actual!r}; the primitive takes {expected!r} or a Tensor "
                    "of it",
                    "qiec-primitive",
                )
            if broadcast is not None and tuple(shape[1]) != broadcast:
                raise KernelError(
                    f"primitive {value.name!r} is applied to tensors of shapes "
                    f"{broadcast!r} and {tuple(shape[1])!r}, which do not agree",
                    "qiec-primitive",
                )
            broadcast = tuple(shape[1])
        expected_result: TypeExpr = (
            signature.result
            if broadcast is None
            else tensor_type(signature.result, broadcast)
        )
        if value.result_type != expected_result:
            raise KernelError(
                f"primitive {value.name!r} produces {render_static(expected_result)}, not "
                f"the claimed {render_static(value.result_type)}",
                "qiec-primitive",
            )
        return expected_result
    if isinstance(value, DistributionValue):
        return _infer_distribution(value, registry, context)
    if isinstance(value, SiteValue):
        if not value.label:
            raise KernelError("a site label cannot be empty", "qiec-distribution")
        if not (
            isinstance(value.result_type, TypeApplication)
            and value.result_type.constructor == SITE_CONSTRUCTOR
            and len(value.result_type.arguments) == 1
        ):
            raise KernelError(
                f"site {value.label!r} claims {render_static(value.result_type)}, not a Site",
                "qiec-distribution",
            )
        registry.validate_type(value.result_type)
        _check_static_variable_scope(value.result_type, context, subject="site type")
        return value.result_type
    if isinstance(value, LogDensity):
        sampleable = infer_value(value.sampleable, registry, context)
        element = sampled_element(sampleable)
        if element is None:
            raise KernelError(
                f"log_prob takes a Sampleable, not {sampleable!r}",
                "qiec-distribution",
            )
        actual = infer_value(value.value, registry, context)
        if actual != element:
            raise KernelError(
                f"log_prob evaluates a distribution over {render_static(element)} at a value "
                f"of type {render_static(actual)}",
                "qiec-distribution",
            )
        if not value.batch:
            return LOG_WEIGHT
        if not (
            isinstance(value.sampleable, DistributionValue)
            and value.sampleable.plate.batch == value.batch
        ):
            raise KernelError(
                "log_prob keeps its weights apart over batch axes only for a "
                "construction plated over those axes",
                "qiec-distribution",
            )
        return tensor_type(LOG_WEIGHT, tuple(axis.size for axis in value.batch))
    if isinstance(value, Gather):
        return _infer_gather(value, registry, context)
    if isinstance(value, WeightSum):
        actual = infer_value(value.value, registry, context)
        shape = tensor_shape(actual)
        if actual != LOG_WEIGHT and (shape is None or shape[0] != LOG_WEIGHT):
            raise KernelError(
                f"a weight sum totals log weights, not {actual!r}",
                "qiec-distribution",
            )
        return LOG_WEIGHT
    if isinstance(value, SegmentSum):
        return _infer_segment_sum(value, registry, context)
    if isinstance(value, KernelMatrix):
        actual = infer_value(value.inputs, registry, context)
        shape = tensor_shape(actual)
        if shape is None or shape[0] != REAL or len(shape[1]) != 1:
            raise KernelError(
                f"a kernel matrix takes a Tensor[Real] of rank 1, not {actual!r}",
                "qiec-distribution",
            )
        if value.kernel != "rbf":
            raise KernelError(
                f"unknown kernel {value.kernel!r}; the kernels are rbf",
                "qiec-distribution",
            )
        if not value.length_scale > 0 or value.jitter < 0:
            raise KernelError(
                "a kernel matrix needs a positive length scale and a "
                "nonnegative jitter",
                "qiec-distribution",
            )
        expected = tensor_type(REAL, (shape[1][0], shape[1][0]))
        if value.result_type != expected:
            raise KernelError(
                f"kernel matrix has type {render_static(expected)}, not the claimed "
                f"{render_static(value.result_type)}",
                "qiec-distribution",
            )
        return expected
    if isinstance(value, AffineMap):
        return _infer_affine_map(value, registry, context)
    if isinstance(value, TableMap):
        return _infer_table_map(value, registry, context)
    if isinstance(value, Reduction):
        shape = tensor_shape(infer_value(value.value, registry, context))
        if shape is None:
            raise KernelError(
                f"a {value.operator} reduction takes a Tensor", "qiec-primitive"
            )
        element = shape[0]
        integral = value.operator in ("sum", "max", "min", "prod")
        if element != REAL and not (integral and element == INT):
            raise KernelError(
                f"a {value.operator} reduction takes a Tensor[Real]"
                + (" or a Tensor[Int]" if integral else "")
                + f", not one of {element!r}",
                "qiec-primitive",
            )
        if value.result_type != element:
            raise KernelError(
                f"reduction has type {render_static(element)}, not the claimed "
                f"{render_static(value.result_type)}",
                "qiec-primitive",
            )
        return element
    if isinstance(value, Rowwise):
        actual = infer_value(value.value, registry, context)
        shape = tensor_shape(actual)
        if shape is None or shape[0] != REAL:
            raise KernelError(
                f"a {value.operator} takes a Tensor[Real], not {actual!r}",
                "qiec-primitive",
            )
        if value.result_type != actual:
            raise KernelError(
                f"{value.operator} keeps its argument's type {render_static(actual)}, not the "
                f"claimed {render_static(value.result_type)}",
                "qiec-primitive",
            )
        return actual
    if isinstance(value, Comprehension):
        if value.binder.type != INT:
            raise KernelError(
                "a comprehension's binder is an Int index", "qiec-primitive"
            )
        registry.validate_type(value.result_type)
        body = infer_value(value.body, registry, context.extend(value.binder))
        inner = tensor_shape(body)
        expected = (
            tensor_type(body, (value.extent,))
            if inner is None
            else tensor_type(inner[0], (value.extent, *inner[1]))
        )
        if value.result_type != expected:
            raise KernelError(
                f"comprehension has type {render_static(expected)}, not the claimed "
                f"{render_static(value.result_type)}",
                "qiec-primitive",
            )
        _check_static_variable_scope(expected, context, subject="comprehension type")
        return expected
    if isinstance(value, TupleValue):
        components = tuple(infer_value(item, registry, context) for item in value.items)
        expected = product_type(*components)
        if value.result_type != expected:
            raise KernelError(
                f"tuple has type {render_static(expected)}, not the claimed {render_static(value.result_type)}",
                "qiec-primitive",
            )
        registry.validate_type(expected)
        _check_static_variable_scope(expected, context, subject="tuple type")
        return expected
    if isinstance(value, TensorValue):
        return _infer_tensor(value, registry, context)
    if isinstance(value, Projection):
        source = infer_value(value.value, registry, context)
        if not (
            isinstance(source, TypeApplication)
            and source.constructor.name.startswith("Product")
            and source.constructor.id == product_type(*source.arguments).constructor.id  # type: ignore[arg-type]
        ):
            raise KernelError(
                f"projection from a value of non-product type {render_static(source)}",
                "qiec-primitive",
            )
        if not 0 <= value.position < len(source.arguments):
            raise KernelError(
                f"projection position {value.position} is outside a product of "
                f"{len(source.arguments)} components",
                "qiec-primitive",
            )
        component = source.arguments[value.position]
        if value.result_type != component:
            raise KernelError(
                f"projection {value.position} has type {render_static(component)}, not the "
                f"claimed {render_static(value.result_type)}",
                "qiec-primitive",
            )
        return component  # type: ignore[return-value]
    if isinstance(value, ConstructorValue):
        constructor = registry.constructor(value.constructor)
        family = registry.families[constructor.family]
        for argument in value.static_arguments:
            registry.validate_static(argument)
            _check_static_variable_scope(
                argument,
                context,
                subject="constructor static argument",
            )
        registry.validate_type(value.result_type)
        _check_static_variable_scope(
            value.result_type,
            context,
            subject="constructor result type",
        )
        parameter_count = len(family.parameters)
        local_count = len(constructor.telescope)
        if len(value.static_arguments) != parameter_count + local_count:
            raise KernelError(
                f"constructor {constructor.name!r} expects "
                f"{parameter_count + local_count} static arguments"
            )
        parameters = instantiate_telescope(
            family.parameters,
            value.static_arguments[:parameter_count],
        )
        locals_ = instantiate_telescope(
            constructor.telescope,
            value.static_arguments[parameter_count:],
        )
        substitution = _combine_substitutions(parameters, locals_)
        if len(value.fields) != len(constructor.fields):
            raise KernelError(
                f"constructor {constructor.name!r} expects {len(constructor.fields)} fields"
            )
        for field_value, field in zip(value.fields, constructor.fields, strict=True):
            actual = infer_value(field_value, registry, context)
            expected = substitute_type(field.type, substitution)
            if actual != expected:
                raise KernelError(
                    f"field {field.name!r} has type {render_static(actual)}, expected {render_static(expected)}"
                )
        parameter_arguments = value.static_arguments[:parameter_count]
        result_indices = tuple(
            substitute_static(index, substitution)
            for index in constructor.result_indices
        )
        expected_result = TypeApplication(
            family.type_constructor,
            (*parameter_arguments, *result_indices),
        )
        if value.result_type != expected_result:
            raise KernelError(
                f"constructor result is {render_static(value.result_type)}, expected {render_static(expected_result)}"
            )
        return expected_result
    raise KernelError(f"unknown value term {value!r}")

check_request

check_request(request: EffectRequest, registry: KernelRegistry, context: CheckContext) -> tuple[EffectRef, TypeExpr]

Check one effect request and report what performing it yields.

PARAMETER	DESCRIPTION
request	The request to check, naming an instance, an interface application, an operation, and the arguments it carries. TYPE: EffectRequest
registry	Declarations the request is resolved against. TYPE: KernelRegistry
context	Scope the request is made in. TYPE: CheckContext

RETURNS	DESCRIPTION
tuple[EffectRef, TypeExpr]	The interface application the request performs, and the type resuming it supplies. The interface comes back because the caller needs it to build the row this request adds to.

RAISES	DESCRIPTION
KernelError	If the operation is unknown, if the named interface does not own it, if the static or value arguments do not match the operation's signature, if the declared result type disagrees with the instantiated one, or if any part mentions a rigid variable out of scope.

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

def check_request(
    request: EffectRequest,
    registry: KernelRegistry,
    context: CheckContext,
) -> tuple[EffectRef, TypeExpr]:
    """Check one effect request and report what performing it yields.

    Parameters
    ----------
    request : EffectRequest
        The request to check, naming an instance, an interface
        application, an operation, and the arguments it carries.
    registry : KernelRegistry
        Declarations the request is resolved against.
    context : CheckContext
        Scope the request is made in.

    Returns
    -------
    tuple[EffectRef, TypeExpr]
        The interface application the request performs, and the type
        resuming it supplies. The interface comes back because the caller
        needs it to build the row this request adds to.

    Raises
    ------
    KernelError
        If the operation is unknown, if the named interface does not own
        it, if the static or value arguments do not match the operation's
        signature, if the declared result type disagrees with the
        instantiated one, or if any part mentions a rigid variable out of
        scope.
    """
    registry.validate_static(request.effect)
    _check_static_variable_scope(request.effect, context, subject="request effect")
    for argument in request.static_arguments:
        registry.validate_static(argument)
        _check_static_variable_scope(
            argument,
            context,
            subject="request static argument",
        )
    registry.validate_type(request.result_type)
    _check_static_variable_scope(
        request.result_type,
        context,
        subject="request result type",
    )
    effect, operation_object = registry.operation(request.operation)
    if not effect.matches(request.effect):
        raise KernelError("request effect interface does not own the operation")
    interface_substitution = instantiate_telescope(
        effect.telescope,
        request.effect.arguments,
    )
    argument_types, result_type = instantiate_operation(
        operation_object,
        request.static_arguments,
        interface_substitution,
    )
    if len(request.arguments) != len(argument_types):
        raise KernelError(
            f"operation {operation_object.name!r} expects {len(argument_types)} arguments"
        )
    for argument, expected in zip(request.arguments, argument_types, strict=True):
        actual = infer_value(argument, registry, context)
        if actual != expected:
            raise KernelError(
                f"operation argument has type {render_static(actual)}, expected {render_static(expected)}"
            )
    if request.result_type != result_type:
        raise KernelError(
            f"request result is {render_static(request.result_type)}, expected {render_static(result_type)}"
        )
    return request.effect, result_type

infer_computation

infer_computation(computation: Computation, registry: KernelRegistry, context: CheckContext = CheckContext()) -> ComputationType

Infer the result type and effect row of a computation.

This is the kernel's entry point for checking effectful code. The row it returns is what remains unhandled: a Handle discharges its instance, so a fully handled computation comes back with the empty row.

PARAMETER	DESCRIPTION
computation	The term to check. TYPE: Computation
registry	Declarations the term is resolved against. TYPE: KernelRegistry
context	Bindings and evidence in scope. The default empty context suits a closed computation. TYPE: CheckContext DEFAULT: CheckContext()

RETURNS	DESCRIPTION
ComputationType	The result type paired with the effects still outstanding.

RAISES	DESCRIPTION
KernelError	If two case branches share a static scope, if a request, value, or branch fails to check, if a handler's interface does not match the instance it discharges, if branch types or rows fail to unify, or if a rigid variable escapes the branch that introduced it.

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

def infer_computation(
    computation: Computation,
    registry: KernelRegistry,
    context: CheckContext = CheckContext(),
) -> ComputationType:
    """Infer the result type and effect row of a computation.

    This is the kernel's entry point for checking effectful code. The
    row it returns is what remains unhandled: a `Handle` discharges its
    instance, so a fully handled computation comes back with the empty
    row.

    Parameters
    ----------
    computation : Computation
        The term to check.
    registry : KernelRegistry
        Declarations the term is resolved against.
    context : CheckContext
        Bindings and evidence in scope. The default empty context suits a
        closed computation.

    Returns
    -------
    ComputationType
        The result type paired with the effects still outstanding.

    Raises
    ------
    KernelError
        If two case branches share a static scope, if a request, value,
        or branch fails to check, if a handler's interface does not match
        the instance it discharges, if branch types or rows fail to
        unify, or if a rigid variable escapes the branch that introduced
        it.
    """
    static_scopes = _computation_static_scopes(computation)
    if len(set(static_scopes)) != len(static_scopes):
        raise KernelError("case branch static scopes must be globally fresh")
    return _infer_computation(computation, registry, context)

resumption_use

resumption_use(computation: Computation) -> ResumptionUse

Count the continuation invocations along the paths of a body.

A resume is lexical to the clause that encloses it, so a call contributes nothing: the callee has no access to this clause's continuation and cannot invoke it however it is written. That is what keeps the analysis exact rather than conservative across recursion.

PARAMETER	DESCRIPTION
computation	The clause body to analyse. TYPE: Computation

RETURNS	DESCRIPTION
ResumptionUse	The interval of invocation counts over the body's paths.

RAISES	DESCRIPTION
KernelError	If the term is of an unknown computation class.

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

def resumption_use(computation: Computation) -> ResumptionUse:
    """Count the continuation invocations along the paths of a body.

    A `resume` is lexical to the clause that encloses it, so a call
    contributes nothing: the callee has no access to this clause's
    continuation and cannot invoke it however it is written. That is what
    keeps the analysis exact rather than conservative across recursion.

    Parameters
    ----------
    computation : Computation
        The clause body to analyse.

    Returns
    -------
    ResumptionUse
        The interval of invocation counts over the body's paths.

    Raises
    ------
    KernelError
        If the term is of an unknown computation class.
    """
    if isinstance(computation, Resume):
        return ONCE
    if isinstance(computation, Return | Perform | Call):
        return NEVER
    if isinstance(computation, Bind):
        return resumption_use(computation.first).then(resumption_use(computation.then))
    if isinstance(computation, NewInstance):
        return resumption_use(computation.body)
    if isinstance(computation, Handle):
        # A `resume` inside a handled computation still names this
        # clause's continuation: handling an instance does not introduce
        # a new one, and the nested handler's own clauses are separate
        # declarations analysed on their own.
        return resumption_use(computation.computation)
    if isinstance(computation, If):
        return resumption_use(computation.then).join(
            resumption_use(computation.otherwise)
        )
    if isinstance(computation, Case):
        if not computation.branches:
            return NEVER
        uses = [resumption_use(branch.body) for branch in computation.branches]
        result = uses[0]
        for use in uses[1:]:
            result = result.join(use)
        return result
    raise KernelError(f"unknown computation term {computation!r}")

check_resumption_grade

check_resumption_grade(grade: ResumptionGrade, use: ResumptionUse, *, subject: str) -> None

Hold a clause body to the grade its declaration promises.

PARAMETER	DESCRIPTION
grade	The declared grade. TYPE: ResumptionGrade
use	What the body actually does, from resumption_use. TYPE: ResumptionUse
subject	What is being checked, named in any diagnostic. TYPE: str

RAISES	DESCRIPTION
KernelError	If the body can resume more often than the grade allows, or, for a linear grade, if some path fails to resume at all. A grade is a promise other code relies on: a handler declared 0 may be compiled without keeping the continuation alive, so resuming anyway is not a stylistic matter.

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

def check_resumption_grade(
    grade: ResumptionGrade,
    use: ResumptionUse,
    *,
    subject: str,
) -> None:
    """Hold a clause body to the grade its declaration promises.

    Parameters
    ----------
    grade : ResumptionGrade
        The declared grade.
    use : ResumptionUse
        What the body actually does, from `resumption_use`.
    subject : str
        What is being checked, named in any diagnostic.

    Raises
    ------
    KernelError
        If the body can resume more often than the grade allows, or, for
        a linear grade, if some path fails to resume at all. A grade is a
        promise other code relies on: a handler declared `0` may be
        compiled without keeping the continuation alive, so resuming
        anyway is not a stylistic matter.
    """
    maximum = use.maximum
    if grade is ResumptionGrade.ZERO:
        if maximum != 0:
            raise KernelError(
                f"{subject} declares grade 0 but can resume "
                f"{'any number of' if maximum is None else maximum} time(s)",
                "qiec-resumption",
            )
        return
    if grade is ResumptionGrade.AFFINE:
        if maximum is None or maximum > 1:
            raise KernelError(
                f"{subject} declares an affine grade but can resume "
                f"{'any number of' if maximum is None else maximum} times",
                "qiec-resumption",
            )
        return
    if grade is ResumptionGrade.LINEAR:
        if maximum is None or maximum > 1:
            raise KernelError(
                f"{subject} declares a linear grade but can resume "
                f"{'any number of' if maximum is None else maximum} times",
                "qiec-resumption",
            )
        if use.minimum < 1:
            raise KernelError(
                f"{subject} declares a linear grade but some path through "
                f"its body does not resume at all",
                "qiec-resumption",
            )
        return