Overview

Parsing, compiling, and printing QVR source, with the public entry points: load and loads compile source to a Program, parse produces the AST that Compiler lowers and module_to_source prints, and the QIEC lowering checks a module against the kernel. The package re-exports the public names of its modules; the AST node classes live in quivers.dsl.ast_nodes.

dsl

quivers DSL: parse .qvr files into trainable PyTorch models.

The DSL provides a declarative syntax for specifying V-enriched categorical morphism networks that compile to quivers.Program instances (nn.Module subclasses). Parsing is delegated to panproto via the qvr tree-sitter grammar; the AST is a tree of quivers.dsl.ast_nodes didactic Models.

Morphism roles need not be declared: a morphism drawn with sample compiles as latent, one drawn with observe as observed, and any other morphism as a kernel. An explicit [role=...] option always wins over inference.

Quick start

::

from quivers.dsl import load, loads

program = loads('''
    object Resp : FinSet 8
    morphism bias : Resp -> Real 1 ~ Normal(0.0, 5.0)
    program m : Resp -> Resp
        sample b <- bias
        observe y : Resp <- Normal(b, 1.0)
        return y
    export m
''')

program = load("model.qvr")

optimizer = torch.optim.Adam(program.parameters())

DISTRIBUTION_FAMILIES module-attribute

DISTRIBUTION_FAMILIES: frozenset[str] = frozenset((*_FAMILY_PARAMETERS, *FAMILY_ALIASES))

Every distribution-family name accepted in QVR source.

Highlighters consume this set so their surface vocabulary follows the same schemas used by validation and lowering.

FAMILY_ALIASES module-attribute

FAMILY_ALIASES: dict[str, str] = {'Pushforward': 'Transformed', 'Truncate': 'Restrict'}

Surface family names that resolve to another family's registry record.

A key is a name a source may write; its value is the family whose specification builds the distribution.

QVR_DEDUCTION_PROTOCOL module-attribute

QVR_DEDUCTION_PROTOCOL: Protocol = panproto.define_protocol({'name': 'qvr_deduction', 'schema_theory': 'ThBratSchema', 'instance_theory': 'ThBratInstance', 'edge_rules': _DEDUCTION_EDGE_RULES, 'obj_kinds': _DEDUCTION_OBJECT_KINDS, 'constraint_sorts': _DEDUCTION_CONSTRAINT_SORTS})

Panproto protocol for a weighted deductive system.

Vertex kinds: * deduction_system: the top-level declaration. * deduction_atom: an atom of the item algebra. * deduction_rule: a named sequent-style inference rule. * deduction_premise: one premise pattern of a rule. * deduction_conclusion: a rule's conclusion pattern.

Edges: * decl: system :math:\to rule. * atom: system :math:\to atom. * premise: rule :math:\to premise. * conclusion: rule :math:\to conclusion.

Constraint sorts carry the system's semiring, start, depth, and each pattern's textual form. Schema morphisms over this protocol correspond to specializations of deduction systems (e.g., :math:\mathsf{CCG} \subset \mathsf{Lambek} \subset \mathsf{MultimodalLambek}).

QVR_PROGRAM_PROTOCOL module-attribute

QVR_PROGRAM_PROTOCOL: Protocol = panproto.define_protocol({'name': 'qvr_program', 'schema_theory': 'ThBratSchema', 'instance_theory': 'ThBratInstance', 'edge_rules': _EDGE_RULES, 'obj_kinds': _OBJECT_KINDS, 'constraint_sorts': _CONSTRAINT_SORTS})

Panproto protocol for a compiled QVR program.

extract_program_schema writes schemas in this protocol. The vertex kinds, edge rules, and constraint sorts are the ones the module docstring lists.

QIEC_STATEMENT_TYPES module-attribute

QIEC_STATEMENT_TYPES = (surface.QiecIndexDecl, surface.QiecFamilyDecl, surface.QiecEffectDecl, surface.QiecEffectInstanceDecl, surface.QiecHandlerDecl, surface.QiecComputationDecl)

The declaration classes that make up the QIEC surface of a module.

qiec_projection keeps these statements and non_qiec_projection drops them, so the tuple fixes where the two routes divide a module.

QVR_SOURCE_VERSION module-attribute

QVR_SOURCE_VERSION = 'qvr-source/v0.20'

The source protocol a QvrQiecLowerer route accepts.

lower_qvr_to_qiec checks its source_version argument against this identifier, which names the QVR surface the elaboration reads.

Module didactic-model

Module(*, statements: tuple[Statement, ...] = field(default_factory=tuple))

Bases: Model

A complete .qvr program (sequence of statements).

Compiler

Compiler(module: Module, *, module_name: str | None = None, file_path: str = '<source>', lower_qiec: bool = True)

Bases: _DeclarationsMixin, _ProgramsMixin, _StructuralMixin, _DeductionsMixin, _ResolutionMixin, _ExpressionsMixin

Compile a quivers DSL AST into a Program.

The compiler maintains three environments:

• objects: name -> SetObject (discrete finite sets)
• spaces: name -> ContinuousSpace
• morphisms: name -> Morphism or ContinuousMorphism (any morphism-like)

It processes statements in order and compiles the output expression into a Program wrapping the morphism DAG.

PARAMETER	DESCRIPTION
module	The parsed AST. TYPE: Module
lower_qiec	Whether to lower the checked QIEC projection. Tooling sets this to False only for its classic-compiler recovery pass after QIEC has already produced a diagnostic. TYPE: bool DEFAULT: True

Source code in src/quivers/dsl/compiler/core.py

def __init__(
    self,
    module: Module,
    *,
    module_name: str | None = None,
    file_path: str = "<source>",
    lower_qiec: bool = True,
) -> None:
    self._qiec_module: QiecModule | None = None
    # The QIEC projection is lowered on first use, after the
    # statements' own checks when `compile` drives it, so a source
    # error the statement compiler names is reported by it.
    self._qiec_source: Module | None = None
    self._qiec_module_name = module_name
    self._qiec_file_path = file_path
    if lower_qiec and has_qiec_surface(module):
        self._qiec_source = module
        module = non_qiec_projection(module)
    self._module = module
    self._algebra: Algebra = PRODUCT_FUZZY
    self._categories: list[str] = []
    self._rules: dict = {}
    self._bundles: dict[str, tuple[str, ...]] = {}
    self._aliases: dict[str, ObjectExpr] = {}
    self._alias_names: set[str] = set()
    # Built-in transformation catalog: singletons looked up by
    # bare name (``expectation``, ``log_prob``, …) and
    # constructors invoked with arguments (``softmax(B)``,
    # ``bayes_invert(prior)``).  Disjoint from
    # `_transformations`, which holds user define-bound
    # transformations defined inside the module.
    self._trans_singletons: dict = _build_default_trans_singletons()
    self._trans_constructors: dict = _build_default_trans_constructors()
    # User-defined transformations bound via ``define t = …``.
    # Disjoint from `_morphisms`: a ``define`` whose RHS
    # resolves to a transformation lands here; a ``define`` whose
    # RHS resolves to a morphism lands in ``_morphisms``.
    self._transformations: dict = {}
    self._objects: dict[str, SetObject] = {}
    self._spaces: dict = {}
    self._morphisms: dict = {}
    self._groups: dict[str, list[str]] = {}
    self._output_expr: Expr | None = None
    # Parametric-program templates: dependent kernels Π(p:P).Kern(dom(p),cod(p))
    # stored as their unsubstituted AST decl. Instantiated at each call
    # site by parameter substitution + α-renaming of internal latents.
    self._program_templates: dict[str, ProgramDecl] = {}
    # Operadic contraction declarations. Each entry is callable
    # from the DSL at define-binding sites; the value records the
    # compiled `EinsumWiring` plus the declared domain /
    # codomain typing for shape-checking at invocation time.
    self._contractions: dict[str, "_CompiledContraction"] = {}

categories property

categories: list[str]

The declared category atoms.

rules property

rules: dict

The compiled rule schemas from rule declarations.

objects property

objects: dict[str, SetObject]

The compiled object environment.

spaces property

spaces: dict

The compiled space environment.

morphisms property

morphisms: dict

The compiled morphism environment.

algebra property

algebra: Algebra

The active algebra.

programs property

programs: dict

Declared parametric program templates (program NAME(...)).

deductions property

deductions: dict

Declared deduction systems (deduction NAME : ...).

signatures property

signatures: dict

Declared signatures (signature NAME).

encoders property

encoders: dict

Declared encoders (encoder NAME : Sig).

decoders property

decoders: dict

Declared decoders (decoder NAME : Sig).

losses property

losses: dict

Declared loss heads (loss NAME : ... [on=...]).

Keyed by entry name; values are the registered :class:LossEntry records. Empty when no loss decl appears in the module.

bundles property

bundles: dict[str, tuple[str, ...]]

Declared bundle aliases (bundle NAME = R1 | R2 | ...).

contractions property

contractions: dict

Declared tensor-network contractions (contraction NAME : ... [wiring=...]).

transformations property

transformations: dict

User-declared transformation constructors / singletons.

qiec_module property

qiec_module: QiecModule | None

The checked QIEC module, or None when the source has no QIEC surface.

RETURNS	DESCRIPTION
QiecModule | None	The module, lowered on first access.

RAISES	DESCRIPTION
QiecDiagnosticError	If the source's QIEC surface fails to lower, other than by a program gap, which the module records instead.

entry_points

entry_points() -> tuple[EntryPoint, ...]

The executable entry points of the checked module.

RETURNS	DESCRIPTION
tuple[EntryPoint, ...]	Every program and define computation, in declaration order; empty when the source has no checked module.

Source code in src/quivers/dsl/compiler/core.py

def entry_points(self) -> tuple[EntryPoint, ...]:
    """The executable entry points of the checked module.

    Returns
    -------
    tuple[EntryPoint, ...]
        Every ``program`` and ``define`` computation, in declaration
        order; empty when the source has no checked module.
    """
    module = self.qiec_module
    if module is None:
        return ()
    return entry_points(module)

entry

entry(name: str) -> EntryPoint

One entry point of the checked module, by name.

PARAMETER	DESCRIPTION
name	The entry's source name. TYPE: str

RETURNS	DESCRIPTION
EntryPoint	The entry.

RAISES	DESCRIPTION
ExecutionFailure	With code qiec-run-module if the source has no checked module, or qiec-run-computation for an unknown name.

Source code in src/quivers/dsl/compiler/core.py

def entry(self, name: str) -> EntryPoint:
    """One entry point of the checked module, by name.

    Parameters
    ----------
    name : str
        The entry's source name.

    Returns
    -------
    EntryPoint
        The entry.

    Raises
    ------
    ExecutionFailure
        With code ``qiec-run-module`` if the source has no checked
        module, or ``qiec-run-computation`` for an unknown name.
    """
    module = self.qiec_module
    if module is None:
        raise ExecutionFailure(
            ExecutionDiagnostic(
                "qiec-run-module",
                "the source has no checked module and no entry points",
                name,
            )
        )
    if module.gap and not any(point.name == name for point in entry_points(module)):
        raise ExecutionFailure(
            ExecutionDiagnostic(
                "qiec-program-gap",
                f"entry point {name!r} was not lowered: {module.gap}",
                name,
            )
        )
    return entry_point(module, name)

compile

compile() -> Program

Compile the module into a trainable Program.

RETURNS	DESCRIPTION
Program	The compiled nn.Module wrapping the morphism DAG.

RAISES	DESCRIPTION
CompileError	On semantic errors (undefined names, type mismatches, etc.).

Source code in src/quivers/dsl/compiler/core.py

def compile(self) -> Program:
    """Compile the module into a trainable Program.

    Returns
    -------
    Program
        The compiled nn.Module wrapping the morphism DAG.

    Raises
    ------
    CompileError
        On semantic errors (undefined names, type mismatches, etc.).
    """
    for stmt in self._module.statements:
        self._compile_statement(stmt)
    qiec_module = self.qiec_module
    if self._output_expr is None:
        # A module may declare only structural artifacts
        # (signatures, encoders, decoders, losses) with no
        # exported morphism; the returned Program is a container
        # carrying those artifacts.
        program = Program(None)
    elif isinstance(
        self._output_expr, ExprIdent
    ) and self._output_expr.name in getattr(self, "_program_templates", {}):
        # The export names a parametric program template. A
        # template has no root morphism on its own (a function
        # ``Pi (p : P). Kern(dom(p), cod(p))`` rather than a
        # single Kleisli arrow); the returned Program is a
        # container holding the template, which the caller
        # instantiates by attribute access.
        tmpl_name = self._output_expr.name
        program = Program(None)
        invoker = self._make_template_invoker(tmpl_name)
        program.templates = {tmpl_name: invoker}
        # Convenience: ``program.<name>(...)`` works directly.
        object.__setattr__(program, tmpl_name, invoker)
    else:
        root_morphism = self._compile_expr(self._output_expr)
        program = Program(root_morphism)
    # Exports after the first become named accessors on the compiled
    # Program: ``export a`` then ``export b`` makes ``a`` the primary
    # morphism (``program.morphism`` / ``program(...)``) and
    # ``program.b`` the compiled form of the second.
    self._attach_secondary_exports(program)
    # Attach the compiler's deduction and posterior registries to
    # the Program so downstream callers can reach them after
    # `quivers.dsl.load(...)`.
    program.deductions = getattr(self, "_deductions", {})
    program.posteriors = getattr(self, "_posteriors", {})
    program.signatures = getattr(self, "_signatures", {})
    program.encoders = getattr(self, "_encoders", {})
    program.decoders = getattr(self, "_decoders", {})
    program.losses = getattr(self, "_loss_registry", None)
    # Structural components are ordinary PyTorch modules. Register them
    # under stable source-derived names as well as exposing the ergonomic
    # dictionaries above, so ``Program.parameters()``, ``state_dict()``,
    # optimizers, and checkpointing see every learned tensor.
    for role, components in (
        ("encoder", program.encoders),
        ("decoder", program.decoders),
    ):
        for name, component in components.items():
            if isinstance(component, nn.Module):
                program.add_module(f"_qvr_{role}_{name}", component)
    # Preserve the checked projection on the compiled container.  A
    # QIEC-only source thus becomes an inspection/evaluation container;
    # its computation graph is never misrepresented as a PyTorch morphism.
    object.__setattr__(program, "qiec", qiec_module)
    # Wire the loss registry into every compiled deduction so the
    # agenda's rule-firing and chart-completion paths can fire
    # rule-attached and chart-attached losses automatically.
    if program.losses is not None:
        for name, system in program.deductions.items():
            system._loss_registry = program.losses  # type: ignore[attr-defined]
            system._deduction_name = name  # type: ignore[attr-defined]
    return program

compile_env

compile_env() -> dict

Compile all statements and return the full environment.

Useful for inspection without requiring an output declaration.

RETURNS	DESCRIPTION
dict	Combined environment of every declared atom: the active algebra, objects, spaces, morphisms, rules, programs (parametric templates), deductions, signatures, encoders, decoders, losses, bundles, contractions, transformations.

Source code in src/quivers/dsl/compiler/core.py

def compile_env(self) -> dict:
    """Compile all statements and return the full environment.

    Useful for inspection without requiring an output declaration.

    Returns
    -------
    dict
        Combined environment of every declared atom: the active
        algebra, objects, spaces, morphisms, rules, programs
        (parametric templates), deductions, signatures, encoders,
        decoders, losses, bundles, contractions, transformations.
    """
    for stmt in self._module.statements:
        self._compile_statement(stmt)
    env: dict = {}
    env["__algebra__"] = self._algebra
    qiec_module = self.qiec_module
    if qiec_module is not None:
        env["__qiec__"] = qiec_module
    for name, obj in self._objects.items():
        env[name] = obj
    for name, space in self._spaces.items():
        env[name] = space
    for name, morph in self._morphisms.items():
        env[name] = morph
    for name, rule in self._rules.items():
        env[name] = rule
    for name, tmpl in getattr(self, "_program_templates", {}).items():
        env[name] = tmpl
    for name, system in getattr(self, "_deductions", {}).items():
        env[name] = system
    for name, sig in getattr(self, "_signatures", {}).items():
        env[name] = sig
    for name, enc in getattr(self, "_encoders", {}).items():
        env[name] = enc
    for name, dec in getattr(self, "_decoders", {}).items():
        env[name] = dec
    reg = getattr(self, "_loss_registry", None)
    if reg is not None:
        for entry in reg.entries:
            env[entry.name] = entry
    for name, bundle in self._bundles.items():
        env[name] = bundle
    for name, contr in getattr(self, "_contractions", {}).items():
        env[name] = contr
    return env

CompileError

CompileError(message: str, line: int = 0, col: int = 0)

Bases: Exception

Raised when the compiler encounters a semantic error.

PARAMETER	DESCRIPTION
message	Error description. TYPE: str
line	Source line number (0 if unknown). TYPE: int DEFAULT: 0
col	Source column number (0 if unknown). TYPE: int DEFAULT: 0

Source code in src/quivers/dsl/compiler/_prelude.py

def __init__(self, message: str, line: int = 0, col: int = 0) -> None:
    self.line = line
    self.col = col
    loc = f"line {line}, col {col}: " if line else ""
    super().__init__(f"{loc}{message}")

Violation didactic-model

Violation(*, code: str, message: str, line: int, col: int, severity: str = 'error')

Bases: Model

One constraint-solver finding.

Every finding check_constraints reports is an error. Compiler diagnostics that should surface in qvr check output without failing the compile set severity to "warning".

PARAMETER	DESCRIPTION
code	The stable diagnostic code. TYPE: str
message	What is wrong. TYPE: str
line	The one-based source line. TYPE: int
col	The source column. TYPE: int
severity	"error" or "warning". TYPE: str DEFAULT: 'error'

EmitError

Bases: Exception

Raised when an AST value has no valid .qvr surface form.

ParseError

Bases: Exception

Raised when .qvr source fails to parse or wrap into AST nodes.

parse and parse_file raise it with a message that starts with the file path and locates the first syntax error.

QvrLexer

Bases: Lexer

Pygments lexer for .qvr (quivers DSL) files.

The lexer is a thin walker over the in-tree tree-sitter parse; the grammar is the single source of truth. There is no regex approximation: when the shared library can't be loaded, the lexer raises a typed exception so the failure is visible at the rendering site rather than silently emitting a degraded highlight.

get_tokens_unprocessed

get_tokens_unprocessed(text: str) -> Iterator[tuple[int, _TokenType, str]]

Yield (index, token_type, text) tuples for text.

PARAMETER	DESCRIPTION
text	The .qvr source to tokenize. TYPE: str

YIELDS	DESCRIPTION
tuple[int, _TokenType, str]	The character offset, Pygments token type, and text of each leaf of the parse and of the whitespace between leaves.

RAISES	DESCRIPTION
OSError	If the in-tree tree-sitter parser library cannot be loaded.

Source code in src/quivers/dsl/pygments_lexer.py

def get_tokens_unprocessed(
    self, text: str
) -> Iterator[tuple[int, _TokenType, str]]:
    """Yield ``(index, token_type, text)`` tuples for ``text``.

    Parameters
    ----------
    text : str
        The ``.qvr`` source to tokenize.

    Yields
    ------
    tuple[int, _TokenType, str]
        The character offset, Pygments token type, and text of each
        leaf of the parse and of the whitespace between leaves.

    Raises
    ------
    OSError
        If the in-tree tree-sitter parser library cannot be loaded.
    """
    parser, _, _ = _load_parser()
    src_bytes = text.encode("utf-8")
    tree = parser.parse(src_bytes)

    # Walk leaf-first in source order, emitting tokens for
    # each leaf and reproducing inter-leaf whitespace as Text.
    leaves: list[tuple[tree_sitter.Node, str | None, str | None]] = []

    def walk(
        node: tree_sitter.Node,
        parent_kind: str | None,
        field_name: str | None,
    ) -> None:
        if not node.children:
            leaves.append((node, parent_kind, field_name))
            return
        for index, child in enumerate(node.children):
            walk(child, node.type, node.field_name_for_child(index))

    walk(tree.root_node, None, None)

    cursor = 0
    for node, parent_kind, field_name in leaves:
        start = node.start_byte
        end = node.end_byte
        if start > cursor:
            gap = src_bytes[cursor:start].decode("utf-8")
            if gap:
                yield (_byte_to_char_offset(src_bytes, cursor), Text, gap)
        node_text = src_bytes[start:end].decode("utf-8")
        token = _node_kind_to_pygments_token(
            node.type,
            node_text,
            parent_kind,
            field_name,
        )
        yield (_byte_to_char_offset(src_bytes, start), token, node_text)
        cursor = end
    if cursor < len(src_bytes):
        tail = src_bytes[cursor:].decode("utf-8")
        if tail:
            yield (_byte_to_char_offset(src_bytes, cursor), Text, tail)

QiecDiagnosticError

QiecDiagnosticError(message: str, *, code: str, file: str, line: int = 0, column: int = 0, program: str | None = None)

Bases: ValueError

A source-located, stable-code diagnostic from QIEC elaboration.

PARAMETER	DESCRIPTION
message	What was rejected. TYPE: str
code	The stable diagnostic code. TYPE: str
file	The file the source was read from. TYPE: str
line	The one-based line, or zero when unknown. TYPE: int DEFAULT: 0
column	The zero-based column. TYPE: int DEFAULT: 0
program	The program whose elaboration raised the diagnostic, when it was raised inside a program body; None for any other declaration. TYPE: str | None DEFAULT: None

Source code in src/quivers/dsl/qiec_diagnostics.py

def __init__(
    self,
    message: str,
    *,
    code: str,
    file: str,
    line: int = 0,
    column: int = 0,
    program: str | None = None,
) -> None:
    self.message = message
    self.code = code
    self.file = file
    self.line = line
    self.column = column
    self.program = program
    location = f"{file}:{line}:{column}" if line else file
    super().__init__(f"{location}: [{code}] {message}")

CheckedQvrQiec dataclass

CheckedQvrQiec(module: QiecModule)

The checked result passed between Didactic's ordered stages.

PARAMETER	DESCRIPTION
module	The kernel module the stages produced. TYPE: QiecModule

QvrQiecLowerer

Quivers-owned implementation of Didactic's extension-lowering protocol.

check

check(source: QvrQiecSource) -> CheckedQvrQiec

Elaborate a source module and recheck the result.

The recheck is deliberate rather than redundant: it rebuilds a registry from the lowered module alone, so a bug in elaboration surfaces here instead of reaching a backend.

PARAMETER	DESCRIPTION
source	The QIEC projection, its module name, and its file path. TYPE: QvrQiecSource

RETURNS	DESCRIPTION
CheckedQvrQiec	The checked module.

RAISES	DESCRIPTION
QiecDiagnosticError	If elaboration rejects the source, carrying a stable code and a source position.
KernelError	If the lowered module fails its independent recheck.

Source code in src/quivers/dsl/qiec_lowering.py

def check(self, source: QvrQiecSource, /) -> CheckedQvrQiec:
    """Elaborate a source module and recheck the result.

    The recheck is deliberate rather than redundant: it rebuilds a
    registry from the lowered module alone, so a bug in elaboration
    surfaces here instead of reaching a backend.

    Parameters
    ----------
    source : QvrQiecSource
        The QIEC projection, its module name, and its file path.

    Returns
    -------
    CheckedQvrQiec
        The checked module.

    Raises
    ------
    QiecDiagnosticError
        If elaboration rejects the source, carrying a stable code and
        a source position.
    KernelError
        If the lowered module fails its independent recheck.
    """
    module = _Elaborator(source).elaborate()
    validate_module(module)
    return CheckedQvrQiec(module)

lower

lower(checked: CheckedQvrQiec) -> QiecModule

Return the checked module.

Checking and lowering are one step here, because the elaborator produces the kernel module directly, so this unwraps rather than transforms.

PARAMETER	DESCRIPTION
checked	The result of check. TYPE: CheckedQvrQiec

RETURNS	DESCRIPTION
QiecModule	The lowered module.

Source code in src/quivers/dsl/qiec_lowering.py

def lower(self, checked: CheckedQvrQiec, /) -> QiecModule:
    """Return the checked module.

    Checking and lowering are one step here, because the elaborator
    produces the kernel module directly, so this unwraps rather than
    transforms.

    Parameters
    ----------
    checked : CheckedQvrQiec
        The result of `check`.

    Returns
    -------
    QiecModule
        The lowered module.
    """
    return checked.module

validate

validate(target: QiecModule) -> None

Recheck a module and confirm it survives serialization.

PARAMETER	DESCRIPTION
target	The module to validate. TYPE: QiecModule

RAISES	DESCRIPTION
KernelError	If the module fails its recheck, or if a dump and load round trip does not return an equal module. The second is checked because identities downstream are content addressed: a module that changed under serialization would hash differently on the other side of the boundary.

Source code in src/quivers/dsl/qiec_lowering.py

def validate(self, target: QiecModule, /) -> None:
    """Recheck a module and confirm it survives serialization.

    Parameters
    ----------
    target : QiecModule
        The module to validate.

    Raises
    ------
    KernelError
        If the module fails its recheck, or if a dump and load round
        trip does not return an equal module. The second is checked
        because identities downstream are content addressed: a module
        that changed under serialization would hash differently on
        the other side of the boundary.
    """
    validate_module(target)
    decoded = loads(dumps(target))
    if decoded != target:
        raise KernelError("QIEC module changed during canonical serialization")

QvrQiecSource dataclass

QvrQiecSource(syntax: Module, module_name: str, file_path: str = '<source>', elaborate_programs: bool = True)

A parsed QVR projection plus the identity absent from the AST root.

PARAMETER	DESCRIPTION
syntax	The projected module. TYPE: Module
module_name	The name every stable identity derives from. TYPE: str
file_path	The path recorded on diagnostics. TYPE: str DEFAULT: '<source>'
elaborate_programs	Whether program declarations become computations. A caller that has found a program using a construct the elaboration does not yet cover lowers the rest of the module without it. TYPE: bool DEFAULT: True

check_constraints

check_constraints(module: Module) -> list[Violation]

Return every well-formedness violation in module.

Walks the statements in source order, building up the symbol tables (residuated universes, enum sets, free monoids, declared rule / schema / bundle names) incrementally so that out-of-order references are flagged.

PARAMETER	DESCRIPTION
module	The parsed module to check. TYPE: Module

RETURNS	DESCRIPTION
list[Violation]	The findings in source order, empty for a well-formed module.

Source code in src/quivers/dsl/constraints.py

def check_constraints(module: Module) -> list[Violation]:
    """Return every well-formedness violation in ``module``.

    Walks the statements in source order, building up the symbol
    tables (residuated universes, enum sets, free monoids, declared
    rule / schema / bundle names) incrementally so that out-of-order
    references are flagged.

    Parameters
    ----------
    module : Module
        The parsed module to check.

    Returns
    -------
    list[Violation]
        The findings in source order, empty for a well-formed module.
    """
    out: list[Violation] = []

    residuated_universes: dict[str, TypeFreeResiduated] = {}
    enum_sets: dict[str, TypeEnumSet] = {}
    free_monoids: dict[str, TypeFreeMonoid] = {}
    aliased_names: dict[str, ObjectExpr] = {}
    rule_names: set[str] = set()
    schema_names: set[str] = set()
    bundle_names: set[str] = set()
    category_atoms: set[str] = set()

    builtin_schemas = set(SCHEMA_REGISTRY.keys())

    def is_residuated_object(name: str) -> bool:
        """True iff ``name`` resolves (via aliases) to a residuated universe."""
        seen: set[str] = set()
        cur = name
        while cur not in seen:
            seen.add(cur)
            if cur in residuated_universes:
                return True
            if cur in aliased_names:
                rhs = aliased_names[cur]
                if isinstance(rhs, TypeName):
                    cur = rhs.name
                    continue
            return False
        return False

    def is_known_effect(name: str) -> bool:
        """True iff ``name`` matches the conventional effect-naming pattern."""
        return bool(name) and (name[0].isupper() or "_" in name)

    def walk_pattern(
        texpr: ObjectExpr,
        *,
        in_residuated_context: bool,
        line_hint: int = 0,
        col_hint: int = 0,
    ) -> None:
        if isinstance(texpr, TypeName):
            return
        if isinstance(texpr, ObjectProduct):
            for c in texpr.components:
                walk_pattern(
                    c,
                    in_residuated_context=in_residuated_context,
                    line_hint=line_hint,
                    col_hint=col_hint,
                )
            return
        if isinstance(texpr, ObjectCoproduct):
            for c in texpr.components:
                walk_pattern(
                    c,
                    in_residuated_context=in_residuated_context,
                    line_hint=line_hint,
                    col_hint=col_hint,
                )
            return
        if isinstance(texpr, ObjectSlash):
            if not in_residuated_context:
                line = texpr.line or line_hint
                col = texpr.col or col_hint
                out.append(
                    Violation(
                        code="residuated_constraint",
                        message=(
                            f"ObjectSlash {texpr.direction!r} appears "
                            "outside a residuated context; either "
                            "declare a FreeResiduated universe and "
                            "parameterize the schema by it, or remove "
                            "the slash"
                        ),
                        line=line,
                        col=col,
                    )
                )
            walk_pattern(
                texpr.result,
                in_residuated_context=in_residuated_context,
                line_hint=line_hint,
                col_hint=col_hint,
            )
            walk_pattern(
                texpr.argument,
                in_residuated_context=in_residuated_context,
                line_hint=line_hint,
                col_hint=col_hint,
            )
            return
        if isinstance(texpr, ObjectEffectApply):
            if not is_known_effect(texpr.effect):
                line = texpr.line or line_hint
                col = texpr.col or col_hint
                out.append(
                    Violation(
                        code="effect_constraint",
                        message=(
                            f"effect {texpr.effect!r} has no "
                            "recognized naming pattern; effect names "
                            "must start with an uppercase letter or "
                            "contain an underscore"
                        ),
                        line=line,
                        col=col,
                    )
                )
            for arg in texpr.args:
                walk_pattern(
                    arg,
                    in_residuated_context=in_residuated_context,
                    line_hint=line_hint,
                    col_hint=col_hint,
                )
            return

    for stmt in module.statements:
        if isinstance(stmt, ObjectDecl):
            init = stmt.init
            for name in stmt.names:
                if isinstance(init, TypeFreeResiduated):
                    residuated_universes[name] = init
                elif isinstance(init, TypeEnumSet):
                    enum_sets[name] = init
                elif isinstance(init, TypeFreeMonoid):
                    free_monoids[name] = init
                elif isinstance(init, TypeFromExpr):
                    aliased_names[name] = init.expr
        elif isinstance(stmt, CategoryDecl):
            for name in stmt.names:
                category_atoms.add(name)
        elif isinstance(stmt, RuleDecl):
            rule_names.add(stmt.name)
            for prem in stmt.premises:
                walk_pattern(
                    prem,
                    in_residuated_context=True,
                    line_hint=stmt.line,
                    col_hint=stmt.col,
                )
            walk_pattern(
                stmt.conclusion,
                in_residuated_context=True,
                line_hint=stmt.line,
                col_hint=stmt.col,
            )
        elif isinstance(stmt, SchemaDecl):
            schema_names.add(stmt.name)
            in_res = any(
                isinstance(p.type_expr, TypeName)
                and is_residuated_object(p.type_expr.name)
                for p in stmt.parameters
            )
            walk_pattern(
                stmt.domain,
                in_residuated_context=in_res,
                line_hint=stmt.line,
                col_hint=stmt.col,
            )
            walk_pattern(
                stmt.codomain,
                in_residuated_context=in_res,
                line_hint=stmt.line,
                col_hint=stmt.col,
            )
        elif isinstance(stmt, BundleDecl):
            bundle_names.add(stmt.name)
            for member in stmt.rules:
                if (
                    member not in rule_names
                    and member not in schema_names
                    and member not in bundle_names
                    and member not in builtin_schemas
                ):
                    out.append(
                        Violation(
                            code="bundle_unknown_member",
                            message=(
                                f"bundle {stmt.name!r} references "
                                f"unknown member {member!r}; not a "
                                "declared rule / schema / bundle, "
                                "nor a built-in schema"
                            ),
                            line=stmt.line,
                            col=stmt.col,
                        )
                    )

    return out

module_to_source

module_to_source(module: Module) -> str

Serialize a Module AST to canonical .qvr source.

PARAMETER	DESCRIPTION
module	The module to print. TYPE: Module

RETURNS	DESCRIPTION
str	The canonical source, ending in a single newline.

RAISES	DESCRIPTION
EmitError	If a node in module has no valid .qvr surface form.

Source code in src/quivers/dsl/emit.py

def module_to_source(module: Module) -> str:
    """Serialize a `Module` AST to canonical ``.qvr`` source.

    Parameters
    ----------
    module : Module
        The module to print.

    Returns
    -------
    str
        The canonical source, ending in a single newline.

    Raises
    ------
    EmitError
        If a node in ``module`` has no valid ``.qvr`` surface form.
    """
    parts = [_emit_statement(stmt, 0) for stmt in module.statements]
    return "\n\n".join(parts) + "\n"

static_argument_to_source

static_argument_to_source(argument: QiecStaticArgument) -> str

Emit one static argument of an effect or type application.

PARAMETER	DESCRIPTION
argument	A type, an effect written as type syntax, or an index literal. TYPE: QiecStaticArgument

RETURNS	DESCRIPTION
str	The argument's source form, as the module emitter writes it.

Source code in src/quivers/dsl/emit.py

def static_argument_to_source(argument: QiecStaticArgument) -> str:
    """Emit one static argument of an effect or type application.

    Parameters
    ----------
    argument : QiecStaticArgument
        A type, an effect written as type syntax, or an index literal.

    Returns
    -------
    str
        The argument's source form, as the module emitter writes it.
    """
    return _emit_qiec_static_argument(argument)

family_parameter_names

family_parameter_names(family: str) -> tuple[str, ...] | None

Return the canonical QVR keyword order for family.

PARAMETER	DESCRIPTION
family	The family name as written in source. TYPE: str

RETURNS	DESCRIPTION
tuple[str, ...] | None	The keyword names in canonical order, or None when no keyword schema is declared for family.

Source code in src/quivers/dsl/family_schemas.py

def family_parameter_names(family: str) -> tuple[str, ...] | None:
    """Return the canonical QVR keyword order for ``family``.

    Parameters
    ----------
    family : str
        The family name as written in source.

    Returns
    -------
    tuple[str, ...] | None
        The keyword names in canonical order, or None when no keyword
        schema is declared for ``family``.
    """
    return _FAMILY_PARAMETERS.get(family)

family_parameterizations

family_parameterizations(family: str, canonical: tuple[str, ...]) -> tuple[tuple[str, ...], ...]

Return every complete keyword schema accepted by family.

PARAMETER	DESCRIPTION
family	The family name as written in source. TYPE: str
canonical	The parameter names the family's registry record declares, used when no schema for family is declared here. TYPE: tuple[str, ...]

RETURNS	DESCRIPTION
tuple[tuple[str, ...], ...]	One tuple of keyword names per accepted parameterization, empty when the family takes no parameters.

Source code in src/quivers/dsl/family_schemas.py

def family_parameterizations(
    family: str, canonical: tuple[str, ...]
) -> tuple[tuple[str, ...], ...]:
    """Return every complete keyword schema accepted by ``family``.

    Parameters
    ----------
    family : str
        The family name as written in source.
    canonical : tuple[str, ...]
        The parameter names the family's registry record declares,
        used when no schema for ``family`` is declared here.

    Returns
    -------
    tuple[tuple[str, ...], ...]
        One tuple of keyword names per accepted parameterization, empty
        when the family takes no parameters.
    """
    alternatives = _ALTERNATIVE_SCHEMAS.get(family)
    if alternatives is not None:
        return alternatives
    parameters = _FAMILY_PARAMETERS.get(family, canonical)
    return (parameters,) if parameters else ()

free_let_names

free_let_names(expr: LetExprNode, *, bound: frozenset[str] = frozenset(), include_callees: bool = False) -> tuple[str, ...]

Return free value names in first-use order.

Lambda parameters and factor indices bind in their bodies. Call names are declarations by default; callers that model a callee as a host value can include them explicitly.

PARAMETER	DESCRIPTION
expr	The expression to scan. TYPE: LetExprNode
bound	Names already bound around expr, which are not free. TYPE: frozenset[str] DEFAULT: frozenset()
include_callees	Whether the function name of a call counts as a free name. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
tuple[str, ...]	Each free name once, in the order of its first occurrence.

RAISES	DESCRIPTION
TypeError	If the scan reaches an unsupported let-expression variant.

Source code in src/quivers/dsl/let_expr_traversal.py

def free_let_names(
    expr: surface.LetExprNode,
    *,
    bound: frozenset[str] = frozenset(),
    include_callees: bool = False,
) -> tuple[str, ...]:
    """Return free value names in first-use order.

    Lambda parameters and factor indices bind in their bodies. Call names are
    declarations by default; callers that model a callee as a host value can
    include them explicitly.

    Parameters
    ----------
    expr : LetExprNode
        The expression to scan.
    bound : frozenset[str]
        Names already bound around ``expr``, which are not free.
    include_callees : bool
        Whether the function name of a call counts as a free name.

    Returns
    -------
    tuple[str, ...]
        Each free name once, in the order of its first occurrence.

    Raises
    ------
    TypeError
        If the scan reaches an unsupported let-expression variant.
    """
    found: list[str] = []

    def visit(node: surface.LetExprNode, local_bound: frozenset[str]) -> None:
        if isinstance(node, surface.LetExprVar):
            if node.name not in local_bound and node.name not in found:
                found.append(node.name)
            return
        if isinstance(node, surface.LetExprCall):
            if (
                include_callees
                and node.func not in local_bound
                and node.func not in found
            ):
                found.append(node.func)
            for argument in node.args:
                visit(argument, local_bound)
            return
        if isinstance(node, surface.LetExprLambda):
            visit(node.body, local_bound | {node.param})
            return
        if isinstance(node, surface.LetExprFactor):
            inner = local_bound | {binder.var for binder in node.binders}
            if node.body is not None:
                visit(node.body, inner)
            for case in node.cases:
                visit(case.value, inner)
            return
        for child in let_expr_children(node):
            visit(child, local_bound)

    visit(expr, bound)
    return tuple(found)

let_expr_children

let_expr_children(expr: LetExprNode) -> tuple[LetExprNode, ...]

Return the immediate expression children of expr in source order.

PARAMETER	DESCRIPTION
expr	The expression to take apart. TYPE: LetExprNode

RETURNS	DESCRIPTION
tuple[LetExprNode, ...]	The direct subexpressions, empty for a leaf.

RAISES	DESCRIPTION
TypeError	If expr is a let-expression variant this function does not cover.

Source code in src/quivers/dsl/let_expr_traversal.py

def let_expr_children(expr: surface.LetExprNode) -> tuple[surface.LetExprNode, ...]:
    """Return the immediate expression children of ``expr`` in source order.

    Parameters
    ----------
    expr : LetExprNode
        The expression to take apart.

    Returns
    -------
    tuple[LetExprNode, ...]
        The direct subexpressions, empty for a leaf.

    Raises
    ------
    TypeError
        If ``expr`` is a let-expression variant this function does not
        cover.
    """
    if isinstance(
        expr,
        surface.LetExprLiteral
        | surface.LetExprBool
        | surface.LetExprUnit
        | surface.LetExprVar
        | surface.LetExprString,
    ):
        return ()
    if isinstance(expr, surface.LetExprBinOp):
        return (expr.left, expr.right)
    if isinstance(expr, surface.LetExprUnaryOp):
        return (expr.operand,)
    if isinstance(expr, surface.LetExprCall):
        return expr.args
    if isinstance(expr, surface.LetExprIndex):
        return (expr.array, *expr.indices)
    if isinstance(expr, surface.LetExprList | surface.LetExprTuple):
        return expr.items
    if isinstance(expr, surface.LetExprLambda):
        return (expr.body,)
    if isinstance(expr, surface.LetExprFactor):
        body = () if expr.body is None else (expr.body,)
        return (*body, *(case.value for case in expr.cases))
    if isinstance(expr, surface.LetExprMethodCall):
        return (expr.receiver, *expr.args)
    if isinstance(expr, surface.QiecConstructorValue):
        return expr.fields
    raise TypeError(
        f"let_expr_children: unsupported let-expression variant {type(expr).__name__}"
    )

substitute_let_expr

substitute_let_expr(expr: LetExprNode, substitution: Mapping[str, LetExprNode]) -> LetExprNode

Substitute expressions for free variables in expr without capture.

A lambda parameter or factor binder that would capture a free name of a replacement is renamed to a fresh name first.

PARAMETER	DESCRIPTION
expr	The expression to rewrite. TYPE: LetExprNode
substitution	The replacement for each free variable name. TYPE: Mapping[str, LetExprNode]

RETURNS	DESCRIPTION
LetExprNode	expr with every free occurrence of a key replaced.

RAISES	DESCRIPTION
TypeError	If the rewrite reaches an unsupported let-expression variant.

Source code in src/quivers/dsl/let_expr_traversal.py

def substitute_let_expr(
    expr: surface.LetExprNode,
    substitution: Mapping[str, surface.LetExprNode],
) -> surface.LetExprNode:
    """Substitute expressions for free variables in ``expr`` without capture.

    A lambda parameter or factor binder that would capture a free name of a
    replacement is renamed to a fresh name first.

    Parameters
    ----------
    expr : LetExprNode
        The expression to rewrite.
    substitution : Mapping[str, LetExprNode]
        The replacement for each free variable name.

    Returns
    -------
    LetExprNode
        ``expr`` with every free occurrence of a key replaced.

    Raises
    ------
    TypeError
        If the rewrite reaches an unsupported let-expression variant.
    """
    occupied = set(free_let_names(expr, include_callees=True))
    for replacement in substitution.values():
        occupied.update(free_let_names(replacement, include_callees=True))
    for node in walk_let_expr(expr):
        if isinstance(node, surface.LetExprLambda):
            occupied.add(node.param)
        elif isinstance(node, surface.LetExprFactor):
            occupied.update(binder.var for binder in node.binders)

    def fresh(base: str) -> str:
        suffix = 1
        candidate = f"{base}_{suffix}"
        while candidate in occupied:
            suffix += 1
            candidate = f"{base}_{suffix}"
        occupied.add(candidate)
        return candidate

    def replacement_names(
        active: Mapping[str, surface.LetExprNode],
        body: surface.LetExprNode,
    ) -> set[str]:
        used_keys = set(free_let_names(body)) & active.keys()
        return {name for key in used_keys for name in free_let_names(active[key])}

    def replace(
        node: surface.LetExprNode,
        active: Mapping[str, surface.LetExprNode],
    ) -> surface.LetExprNode:
        if isinstance(node, surface.LetExprVar):
            return active.get(node.name, node)
        if isinstance(
            node,
            surface.LetExprLiteral
            | surface.LetExprBool
            | surface.LetExprUnit
            | surface.LetExprString,
        ):
            return node
        if isinstance(node, surface.LetExprBinOp):
            return node.with_(
                left=replace(node.left, active),
                right=replace(node.right, active),
            )
        if isinstance(node, surface.LetExprUnaryOp):
            return node.with_(operand=replace(node.operand, active))
        if isinstance(node, surface.LetExprCall):
            return node.with_(args=tuple(replace(item, active) for item in node.args))
        if isinstance(node, surface.LetExprIndex):
            return node.with_(
                array=replace(node.array, active),
                indices=tuple(replace(item, active) for item in node.indices),
            )
        if isinstance(node, surface.LetExprList | surface.LetExprTuple):
            return node.with_(items=tuple(replace(item, active) for item in node.items))
        if isinstance(node, surface.LetExprLambda):
            inner = {key: value for key, value in active.items() if key != node.param}
            param = node.param
            body = node.body
            if param in replacement_names(inner, body):
                renamed = fresh(param)
                body = replace(
                    body,
                    {
                        param: surface.LetExprVar(
                            name=renamed,
                            line=node.line,
                            col=node.col,
                        )
                    },
                )
                param = renamed
            return node.with_(param=param, body=replace(body, inner))
        if isinstance(node, surface.LetExprFactor):
            bound = {binder.var for binder in node.binders}
            inner = {key: value for key, value in active.items() if key not in bound}
            bodies = (() if node.body is None else (node.body,)) + tuple(
                case.value for case in node.cases
            )
            captures = {
                name for body in bodies for name in replacement_names(inner, body)
            }
            binders = list(node.binders)
            body = node.body
            cases = node.cases
            for position, binder in enumerate(binders):
                if binder.var not in captures:
                    continue
                renamed = fresh(binder.var)
                alpha = {
                    binder.var: surface.LetExprVar(
                        name=renamed,
                        line=binder.line,
                        col=binder.col,
                    )
                }
                body = None if body is None else replace(body, alpha)
                cases = tuple(
                    case.with_(value=replace(case.value, alpha)) for case in cases
                )
                binders[position] = binder.with_(var=renamed)
            return node.with_(
                binders=tuple(binders),
                body=None if body is None else replace(body, inner),
                cases=tuple(
                    case.with_(value=replace(case.value, inner)) for case in cases
                ),
            )
        if isinstance(node, surface.LetExprMethodCall):
            return node.with_(
                receiver=replace(node.receiver, active),
                args=tuple(replace(item, active) for item in node.args),
            )
        if isinstance(node, surface.QiecConstructorValue):
            return node.with_(
                fields=tuple(replace(field, active) for field in node.fields)
            )
        raise TypeError(
            f"substitute_let_expr: unsupported let-expression variant "
            f"{type(node).__name__}"
        )

    return replace(expr, substitution)

walk_let_expr

walk_let_expr(expr: LetExprNode) -> Iterator[LetExprNode]

Yield expr and all descendants in preorder.

PARAMETER	DESCRIPTION
expr	The root of the walk. TYPE: LetExprNode

YIELDS	DESCRIPTION
LetExprNode	expr first, then each subexpression before its own children.

RAISES	DESCRIPTION
TypeError	If the walk reaches an unsupported let-expression variant.

Source code in src/quivers/dsl/let_expr_traversal.py

def walk_let_expr(expr: surface.LetExprNode) -> Iterator[surface.LetExprNode]:
    """Yield ``expr`` and all descendants in preorder.

    Parameters
    ----------
    expr : LetExprNode
        The root of the walk.

    Yields
    ------
    LetExprNode
        ``expr`` first, then each subexpression before its own children.

    Raises
    ------
    TypeError
        If the walk reaches an unsupported let-expression variant.
    """
    yield expr
    for child in let_expr_children(expr):
        yield from walk_let_expr(child)

parse

parse(source: str | bytes, file_path: str = '<source>') -> Module

Parse .qvr source into a Module.

PARAMETER	DESCRIPTION
source	The source text, or its UTF-8 bytes. TYPE: str | bytes
file_path	The path recorded on diagnostics. TYPE: str DEFAULT: '<source>'

RETURNS	DESCRIPTION
Module	The module's statements in source order.

RAISES	DESCRIPTION
ParseError	If the source has a syntax error or a node the AST cannot represent.

Source code in src/quivers/dsl/parser/core.py

def parse(source: str | bytes, file_path: str = "<source>") -> Module:
    """Parse `.qvr` source into a `Module`.

    Parameters
    ----------
    source : str | bytes
        The source text, or its UTF-8 bytes.
    file_path : str
        The path recorded on diagnostics.

    Returns
    -------
    Module
        The module's statements in source order.

    Raises
    ------
    ParseError
        If the source has a syntax error or a node the AST cannot
        represent.
    """
    if isinstance(source, str):
        source_bytes = source.encode("utf-8")
    else:
        source_bytes = source

    try:
        schema = _registry().parse_with_protocol("qvr", source_bytes, file_path)
    except panproto.PanprotoError as exc:
        raise ParseError(f"{file_path}: panproto failed to parse: {exc}") from exc
    tree = _Tree(schema, source_bytes)

    _reject_malformed(tree, file_path)

    root_id = next(
        (vid for vid in tree.vertices if tree.kind(vid) == "source_file"),
        None,
    )
    if root_id is None:
        raise ParseError(
            f"{file_path}: source failed to parse (the tree has no source_file "
            "root); check that every program body ends with a return step"
        )

    statements: list[Statement] = []
    for child in tree.positional(root_id):
        ckind = tree.kind(child)
        if ckind in ("line_comment", "block_comment"):
            # plain comments are extras and are dropped at parse time;
            # `#!` doc comments ride each declaration's `docs` field
            # and are attached by the per-declaration walkers.
            continue
        result = _walk_statement(tree, child)
        if isinstance(result, list):
            statements.extend(result)
        else:
            statements.append(result)
    return Module(statements=tuple(statements))

parse_file

parse_file(path: str | Path) -> Module

Parse a .qvr file at path.

PARAMETER	DESCRIPTION
path	The file to read. TYPE: str | Path

RETURNS	DESCRIPTION
Module	The file's statements in source order.

RAISES	DESCRIPTION
OSError	If the file cannot be read.
ParseError	If the file does not parse.

Source code in src/quivers/dsl/parser/core.py

def parse_file(path: str | Path) -> Module:
    """Parse a `.qvr` file at `path`.

    Parameters
    ----------
    path : str | Path
        The file to read.

    Returns
    -------
    Module
        The file's statements in source order.

    Raises
    ------
    OSError
        If the file cannot be read.
    ParseError
        If the file does not parse.
    """
    p = Path(path)
    return parse(p.read_bytes(), str(p))

extract_deduction_schema

extract_deduction_schema(compiler: Compiler) -> Schema

Produce a panproto.Schema for the compiler's deduction-system environment.

Walks the compiler's _deductions registry and emits one panproto vertex per deduction system, one per rule, one per atom, and per-premise / per-conclusion pattern vertices. The returned schema validates against QVR_DEDUCTION_PROTOCOL and is suitable for panproto.diff_schemas and panproto.auto_generate_lens operations over deduction systems.

PARAMETER	DESCRIPTION
compiler	A compiler whose _deductions registry is populated. TYPE: Compiler

RETURNS	DESCRIPTION
Schema	A schema in the qvr_deduction protocol.

Source code in src/quivers/dsl/program_theory.py

def extract_deduction_schema(compiler: Compiler) -> panproto.Schema:
    """Produce a `panproto.Schema` for the compiler's
    deduction-system environment.

    Walks the compiler's ``_deductions`` registry and emits one
    panproto vertex per deduction system, one per rule, one per
    atom, and per-premise / per-conclusion pattern vertices. The
    returned schema validates against
    `QVR_DEDUCTION_PROTOCOL` and is suitable for
    `panproto.diff_schemas` and
    `panproto.auto_generate_lens` operations over deduction
    systems.

    Parameters
    ----------
    compiler : Compiler
        A compiler whose ``_deductions`` registry is populated.

    Returns
    -------
    panproto.Schema
        A schema in the ``qvr_deduction`` protocol.
    """
    builder = QVR_DEDUCTION_PROTOCOL.schema()
    deductions = getattr(compiler, "_deductions", {})
    for name, system in deductions.items():
        sys_vid = f"deduction:{name}"
        builder.vertex(sys_vid, "deduction_system")
        builder.constraint(sys_vid, "name", name)
        builder.constraint(sys_vid, "semiring", system.semiring.__class__.__name__)
        for rule_idx, rule in enumerate(system.rules):
            rule_vid = f"{sys_vid}/rule:{rule.name}"
            builder.vertex(rule_vid, "deduction_rule")
            builder.constraint(rule_vid, "name", rule.name)
            builder.edge(sys_vid, rule_vid, "decl")
            for prem_idx, premise in enumerate(rule.premises):
                p_vid = f"{rule_vid}/premise:{prem_idx}"
                builder.vertex(p_vid, "deduction_premise")
                builder.constraint(p_vid, "pattern", repr(premise))
                builder.edge(rule_vid, p_vid, "premise")
            conc_vid = f"{rule_vid}/conclusion"
            builder.vertex(conc_vid, "deduction_conclusion")
            builder.constraint(conc_vid, "pattern", repr(rule.conclusion))
            builder.edge(rule_vid, conc_vid, "conclusion")
    return builder.build()

extract_program_schema

extract_program_schema(compiler: Compiler) -> Schema

Produce a panproto.Schema for a compiled program.

Walks the compiler's resolved environment (objects, spaces, morphisms) and its checked kernel module (effects, instances, handlers, computations, and program entry points) and emits a graph of vertices and edges in the QVR_PROGRAM_PROTOCOL protocol. The returned schema validates against that protocol and is suitable for panproto.diff_schemas, panproto.auto_generate_lens, and the rest of panproto's schema-level operations.

PARAMETER	DESCRIPTION
compiler	A quivers.dsl.compiler.Compiler after compile_env (or compile) has populated the resolved environments. TYPE: Compiler

RETURNS	DESCRIPTION
Schema	A program-level Schema in the qvr_program protocol.

Source code in src/quivers/dsl/program_theory.py

def extract_program_schema(compiler: Compiler) -> panproto.Schema:
    """Produce a `panproto.Schema` for a compiled program.

    Walks the compiler's resolved environment (objects, spaces, morphisms)
    and its checked kernel module (effects, instances, handlers,
    computations, and program entry points) and emits a graph of
    vertices and edges in the `QVR_PROGRAM_PROTOCOL` protocol. The
    returned schema validates against that protocol and is suitable for
    `panproto.diff_schemas`, `panproto.auto_generate_lens`, and the rest
    of panproto's schema-level operations.

    Parameters
    ----------
    compiler : Compiler
        A [`quivers.dsl.compiler.Compiler`][quivers.dsl.compiler.Compiler] after `compile_env`
        (or `compile`) has populated the resolved environments.

    Returns
    -------
    panproto.Schema
        A program-level Schema in the ``qvr_program`` protocol.
    """
    builder = QVR_PROGRAM_PROTOCOL.schema()
    writer = _SchemaWriter(builder)

    builder.vertex("program", "program")
    if compiler._algebra is not None:
        builder.constraint("program", "algebra", type(compiler._algebra).__name__)

    # object decls
    for name, obj in compiler._objects.items():
        decl_vid = f"object_decl::{name}"
        builder.vertex(decl_vid, "object_decl")
        builder.constraint(decl_vid, "name", name)
        builder.edge("program", decl_vid, "decl")
        target_vid = writer.write_set_object(obj)
        builder.edge(decl_vid, target_vid, "binds_to")

    # space decls
    for name, space in compiler._spaces.items():
        decl_vid = f"space_decl::{name}"
        builder.vertex(decl_vid, "space_decl")
        builder.constraint(decl_vid, "name", name)
        builder.edge("program", decl_vid, "decl")
        target_vid = writer.write_space(space)
        builder.edge(decl_vid, target_vid, "binds_to")

    # morphism decls: the compiler's _morphisms env holds named primitive
    # morphisms; we record them as morphism_decl vertices with domain/codomain.
    # Composite morphisms (let-bindings, output) are derived rather than
    # recorded directly; the structural Diff-able layer is the named decls.
    for name, morphism in compiler._morphisms.items():
        kind = _classify_morphism_kind(morphism)

        decl_vid = f"{kind}::{name}"
        builder.vertex(decl_vid, kind)
        builder.constraint(decl_vid, "name", name)
        builder.edge("program", decl_vid, "decl")

        dom = getattr(morphism, "domain", None)
        cod = getattr(morphism, "codomain", None)
        if dom is not None:
            dom_vid = writer.write_any(dom)
            builder.edge(decl_vid, dom_vid, "domain")
        if cod is not None:
            cod_vid = writer.write_any(cod)
            builder.edge(decl_vid, cod_vid, "codomain")

    # output decl: the compiler's `_output_expr` holds the AST expression
    # whose compilation produces the program's root morphism. We record an
    # output_decl vertex carrying the expression's source-text form (when
    # available) as a name constraint; the structural diff cares about
    # presence/absence of an output, not its detailed shape.
    if compiler._output_expr is not None:
        out_vid = "output_decl"
        builder.vertex(out_vid, "output_decl")
        # ExprIdent / ExprIdentity carry a single name; composite expressions
        # don't have a single canonical name, so mark them as "<composite>".
        expr = compiler._output_expr
        if isinstance(expr, ExprIdent):
            label = expr.name
        elif isinstance(expr, ExprIdentity):
            label = f"identity({expr.object_name})"
        else:
            label = "<composite>"
        builder.constraint(out_vid, "name", label)
        builder.edge("program", out_vid, "output")

    if compiler.qiec_module is not None:
        _KernelWriter(builder, compiler.qiec_module).write()

    return builder.build()

has_qiec_surface

has_qiec_surface(module: Module) -> bool

Return whether a parsed module has anything to elaborate through QIEC.

PARAMETER	DESCRIPTION
module	The parsed module to inspect. TYPE: Module

RETURNS	DESCRIPTION
bool	True when at least one statement belongs to the QIEC surface or is a program or deduction the elaboration turns into computations. Callers use this to decide whether the QIEC route runs at all, so a module of other declarations pays nothing for it.

Source code in src/quivers/dsl/qiec_lowering.py

def has_qiec_surface(module: surface.Module) -> bool:
    """Return whether a parsed module has anything to elaborate through QIEC.

    Parameters
    ----------
    module : surface.Module
        The parsed module to inspect.

    Returns
    -------
    bool
        True when at least one statement belongs to the QIEC surface or
        is a program or deduction the elaboration turns into
        computations. Callers
        use this to decide whether the QIEC route runs at all, so a
        module of other declarations pays nothing for it.
    """

    return any(
        isinstance(
            statement,
            (
                *QIEC_STATEMENT_TYPES,
                surface.DeductionDecl,
                surface.SignatureDecl,
                surface.EncoderDecl,
                surface.DecoderDecl,
                surface.LossDecl,
            ),
        )
        or (
            isinstance(statement, surface.DefineDecl)
            and isinstance(statement.expr, surface.ExprParser)
        )
        or (
            isinstance(statement, surface.ProgramDecl)
            and (
                statement.type_params is None
                or all(
                    isinstance(parameter, surface.ScalarParam)
                    for parameter in statement.type_params
                )
            )
        )
        for statement in module.statements
    )

lower_qvr_to_qiec

lower_qvr_to_qiec(module: Module, *, module_name: str | None = None, file_path: str = '<source>', source_version: str = QVR_SOURCE_VERSION, target_version: str = QIEC_ABI, elaborate_programs: bool = True) -> QiecModule

Check and lower the QIEC projection of one parsed QVR module.

PARAMETER	DESCRIPTION
module	The parsed module, which may mix QIEC and ordinary declarations. TYPE: Module
module_name	Name the stable identities derive from. None takes it from file_path, so two files of the same stem produce the same identities and a rename is a deliberate act. TYPE: str or None DEFAULT: None
file_path	Path recorded on diagnostics. TYPE: str DEFAULT: '<source>'
source_version	The source protocol the module is written against. TYPE: str DEFAULT: QVR_SOURCE_VERSION
target_version	The kernel ABI to lower to. TYPE: str DEFAULT: QIEC_ABI
elaborate_programs	Whether program declarations become computations. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
QiecModule	The checked, lowered module.

RAISES	DESCRIPTION
QiecDiagnosticError	If the source is rejected, with a stable code and a position. The code qiec-program-gap names a program construct, such as a parsing chart or a network-parameterized morphism, whose elaboration is not yet defined; a caller may lower the module again without its programs.
KernelError	If the lowered module fails its independent recheck.

Source code in src/quivers/dsl/qiec_lowering.py

def lower_qvr_to_qiec(
    module: surface.Module,
    *,
    module_name: str | None = None,
    file_path: str = "<source>",
    source_version: str = QVR_SOURCE_VERSION,
    target_version: str = QIEC_ABI,
    elaborate_programs: bool = True,
) -> QiecModule:
    """Check and lower the QIEC projection of one parsed QVR module.

    Parameters
    ----------
    module : surface.Module
        The parsed module, which may mix QIEC and ordinary declarations.
    module_name : str or None
        Name the stable identities derive from. None takes it from
        `file_path`, so two files of the same stem produce the same
        identities and a rename is a deliberate act.
    file_path : str
        Path recorded on diagnostics.
    source_version : str
        The source protocol the module is written against.
    target_version : str
        The kernel ABI to lower to.
    elaborate_programs : bool
        Whether ``program`` declarations become computations.

    Returns
    -------
    QiecModule
        The checked, lowered module.

    Raises
    ------
    QiecDiagnosticError
        If the source is rejected, with a stable code and a position. The
        code ``qiec-program-gap`` names a program construct, such as a
        parsing chart or a network-parameterized morphism, whose
        elaboration is not yet defined; a caller may lower the module
        again without its programs.
    KernelError
        If the lowered module fails its independent recheck.
    """

    if module_name is None:
        module_name = _module_name(file_path)
    source = QvrQiecSource(
        qiec_projection(module), module_name, file_path, elaborate_programs
    )
    return lower_checked(
        source,
        QvrQiecLowerer(),
        source_version=source_version,
        target_version=target_version,
    )

non_qiec_projection

non_qiec_projection(module: Module) -> Module

Project declarations consumed by the categorical/probabilistic compiler.

PARAMETER	DESCRIPTION
module	The parsed module to project. TYPE: Module

RETURNS	DESCRIPTION
Module	A module of everything outside the QIEC surface, in source order.

Source code in src/quivers/dsl/qiec_lowering.py

def non_qiec_projection(module: surface.Module) -> surface.Module:
    """Project declarations consumed by the categorical/probabilistic compiler.

    Parameters
    ----------
    module : surface.Module
        The parsed module to project.

    Returns
    -------
    surface.Module
        A module of everything outside the QIEC surface, in source
        order.
    """

    return surface.Module(
        statements=tuple(
            statement
            for statement in module.statements
            if not isinstance(statement, QIEC_STATEMENT_TYPES)
        )
    )

qiec_projection

qiec_projection(module: Module) -> Module

Project QIEC declarations from a module with multiple neighborhoods.

PARAMETER	DESCRIPTION
module	The parsed module to project. TYPE: Module

RETURNS	DESCRIPTION
Module	A module of the QIEC statements alone, in source order. The complement is non_qiec_projection, and the two partition the input, so nothing is dropped between the two routes.

Source code in src/quivers/dsl/qiec_lowering.py

def qiec_projection(module: surface.Module) -> surface.Module:
    """Project QIEC declarations from a module with multiple neighborhoods.

    Parameters
    ----------
    module : surface.Module
        The parsed module to project.

    Returns
    -------
    surface.Module
        A module of the QIEC statements alone, in source order. The
        complement is `non_qiec_projection`, and the two partition the
        input, so nothing is dropped between the two routes.
    """

    return surface.Module(
        statements=tuple(
            statement
            for statement in module.statements
            if isinstance(statement, (*QIEC_STATEMENT_TYPES, *PROGRAM_CONTEXT_TYPES))
        )
    )

loads

loads(source: str, *, data: dict | None = None) -> Program

Compile .qvr source text into a trainable Program.

PARAMETER	DESCRIPTION
source	The .qvr source. TYPE: str
data	Maps string keys to tensors (or tensor-like objects) for any from_data("KEY") initializers in the source. The compiler looks each key up at compile time; an unknown key raises CompileError. TYPE: dict DEFAULT: None

RETURNS	DESCRIPTION
Program	The compiled program.

RAISES	DESCRIPTION
ParseError	If source does not parse.
CompileError	If the parsed module does not compile.

Source code in src/quivers/dsl/__init__.py

def loads(
    source: str,
    *,
    data: dict | None = None,
) -> Program:
    """Compile .qvr source text into a trainable Program.

    Parameters
    ----------
    source : str
        The ``.qvr`` source.
    data : dict, optional
        Maps string keys to tensors (or tensor-like objects) for
        any ``from_data("KEY")`` initializers in the source. The
        compiler looks each key up at compile time; an unknown key
        raises `CompileError`.

    Returns
    -------
    Program
        The compiled program.

    Raises
    ------
    ParseError
        If ``source`` does not parse.
    CompileError
        If the parsed module does not compile.
    """
    ast = parse(source)
    compiler = Compiler(ast, module_name="source", file_path="<source>")
    if data is not None:
        compiler.bind_data(data)
    return compiler.compile()

load

load(path: str | Path, *, data: dict | None = None) -> Program

Load and compile a .qvr file into a trainable Program.

PARAMETER	DESCRIPTION
path	The .qvr file. Its stem names the compiled module. TYPE: str | Path
data	Maps string keys to tensors for any from_data("KEY") initializers in the file, as in loads. TYPE: dict DEFAULT: None

RETURNS	DESCRIPTION
Program	The compiled program.

RAISES	DESCRIPTION
OSError	If the file cannot be read.
ParseError	If the file does not parse.
CompileError	If the parsed module does not compile.

Source code in src/quivers/dsl/__init__.py

def load(
    path: str | Path,
    *,
    data: dict | None = None,
) -> Program:
    """Load and compile a .qvr file into a trainable Program.

    Parameters
    ----------
    path : str | Path
        The ``.qvr`` file. Its stem names the compiled module.
    data : dict, optional
        Maps string keys to tensors for any ``from_data("KEY")``
        initializers in the file, as in `loads`.

    Returns
    -------
    Program
        The compiled program.

    Raises
    ------
    OSError
        If the file cannot be read.
    ParseError
        If the file does not parse.
    CompileError
        If the parsed module does not compile.
    """
    ast = parse_file(path)
    compiler = Compiler(ast, module_name=Path(path).stem, file_path=str(path))
    if data is not None:
        compiler.bind_data(data)
    return compiler.compile()