CLI: qvr

The qvr console script ships with the package as a thin wrapper around the parser, constraint solver, and compiler. Subcommands:

qvr check FILES...

Parse, run the constraint solver, and compile every supplied .qvr file. Emits structured diagnostics; exits 0 on full success, 1 on any error.

Flags:

• --json, emit a single JSON document on stdout containing the full diagnostic list. Suitable for CI / pre-commit hooks.

Diagnostic codes:

• parse, tree-sitter rejected the source.
• compile, the compiler raised :class:CompileError.
• residuated_constraint, a TypeSlash pattern appears outside a residuated context.
• effect_constraint, a TypeEffectApply references an effect whose name doesn't match the conventional pattern.
• bundle_unknown_member, a bundle declaration references a member that isn't a declared rule, schema, bundle, or built-in schema.
• io, file-system error (file not found, permission denied).

qvr migrate PATHS...

Lower .qvr source forward along the QVR grammar release chain. The composer chains the adjacent-pair migrators defined in quivers.cli.migrations so users do not have to know the intermediate versions; pinning the boundary with --from / --to selects a sub-chain when needed.

Flags:

• --from VERSION, source revision in the chain (defaults to the most recent release).
• --to VERSION, target revision (defaults to HEAD).
• --dry-run, report which files would change without writing.
• --output DIR, write migrated copies under DIR instead of overwriting the originals.

Directory arguments are walked recursively; individual .qvr files may also be supplied. The migration tooling is built on the in-tree panproto VCS at grammars/qvr/vcs/ so adding a new release is purely additive to the migrations package.

qvr repl [FILE]

Start the interactive REPL. Without a file, opens an empty session; with a file, loads and elaborates it before dropping to the prompt. See REPL and Language Server.

qvr lsp

Run the Language Server over stdio. Editor extensions (VS Code, Zed, Neovim) invoke this; the protocol is LSP 3.17. See REPL and Language Server.

Module reference

cli

Command-line entry points for quivers.

The main function is registered as the qvr console script in pyproject.toml.

Subcommands:

• qvr check FILES... — parse + compile every supplied .qvr file, emitting structured diagnostics. Exits 0 on full success, non-zero when any file produces an error.
• qvr migrate --from VER --to VER PATHS... — lower .qvr source files from one tagged grammar revision to another, via panproto migrations composed from the in-tree grammars/qvr/vcs chain.

Output format: human-readable by default, structured JSON when --json is supplied. Each diagnostic carries:

• file: source path,
• line, col: 1-indexed source location,
• severity: "error", "warning", or "note",
• code: stable diagnostic code (parse, compile, effect_constraint, residuated_constraint),
• message: human-readable description.

check

qvr check — parse + compile .qvr files and report diagnostics.

Implementation

For each input file:

1. parse(source) is invoked; ParseError produces a code="parse" diagnostic.
2. Compiler(module).compile() is invoked; CompileError produces a code="compile" diagnostic.
3. The constraint solver in quivers.dsl.constraints walks the parsed AST for residuated-universe and effect-typed-application well-formedness violations; each emits a code="residuated_constraint" or code="effect_constraint" diagnostic.

Successful files emit no diagnostics; on success the human-readable mode prints "OK file.qvr".

Exit codes:

• 0 — every file compiled without diagnostics,
• 1 — at least one file produced an error diagnostic,
• 2 — usage / IO error.

Diagnostic dataclass

Diagnostic(file: str, line: int, col: int, severity: Severity, code: str, message: str)

One structured diagnostic message.

main

main(files: list[str], *, json_output: bool = False) -> int

Run qvr check on a list of paths.

PARAMETER	DESCRIPTION
files	Paths to .qvr files. TYPE: list of str
json_output	When True, emit a single JSON document on stdout containing the full diagnostic list. When False (default), emit human-readable lines. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
int	Exit code. 0 on full success; 1 on any error diagnostic.

Source code in src/quivers/cli/check.py

def main(files: list[str], *, json_output: bool = False) -> int:
    """Run ``qvr check`` on a list of paths.

    Parameters
    ----------
    files : list of str
        Paths to ``.qvr`` files.
    json_output : bool
        When True, emit a single JSON document on stdout containing
        the full diagnostic list. When False (default), emit
        human-readable lines.

    Returns
    -------
    int
        Exit code. 0 on full success; 1 on any error diagnostic.
    """
    paths = [Path(f) for f in files]
    all_diags: list[Diagnostic] = []
    for p in paths:
        all_diags.extend(_check_one(p))

    has_error = any(d.severity == "error" for d in all_diags)

    if json_output:
        payload = {
            "files": [str(p) for p in paths],
            "diagnostics": [asdict(d) for d in all_diags],
            "ok": not has_error,
        }
        sys.stdout.write(json.dumps(payload, indent=2))
        sys.stdout.write("\n")
    else:
        files_with_diags: set[str] = set()
        for d in all_diags:
            files_with_diags.add(d.file)
            loc = f"{d.file}:{d.line}:{d.col}" if d.line else d.file
            sys.stderr.write(f"{loc}: {d.severity}[{d.code}]: {d.message}\n")
        for p in paths:
            if str(p) not in files_with_diags:
                sys.stdout.write(f"OK {p}\n")
        if has_error:
            sys.stderr.write(
                f"\n{sum(1 for d in all_diags if d.severity == 'error')} "
                f"error(s) across {len(paths)} file(s)\n"
            )

    return 1 if has_error else 0

migrate

qvr migrate command.

Lowers .qvr source forward along the QVR grammar release chain declared in quivers.cli.migrations.

Surface::

qvr migrate path/to/file.qvr [paths...]
qvr migrate --from v0.10.0 --to HEAD --dry-run docs/examples/source/
qvr migrate --output OUT_DIR --to HEAD path/to/file.qvr

--from defaults to the most recent release on the chain (the penultimate entry of quivers.cli.migrations.CHAIN); --to defaults to HEAD. Both must be members of quivers.cli.migrations.CHAIN; the CLI composes the intermediate adjacent-pair migrators automatically, so adding a new release is purely additive to the migrations package.

--dry-run reports which files would change without writing output. --output DIR writes migrated copies under DIR instead of overwriting the originals.

MigrateError

Bases: Exception

Raised by the migrate CLI on a recoverable user-facing error.

main

main(args: Namespace) -> int

Entry point invoked by the top-level qvr dispatcher.

Returns 0 when every file migrated cleanly; 2 on invalid input.

Source code in src/quivers/cli/migrate.py

def main(args: argparse.Namespace) -> int:
    """Entry point invoked by the top-level ``qvr`` dispatcher.

    Returns 0 when every file migrated cleanly; 2 on invalid input.
    """
    # ``--check`` runs the panproto-VCS coverage check across the
    # migration chain. For each adjacent (from, to) pair, computes
    # the schema diff and reports any source rule removed at the
    # target whose hop migrator has no converter. Non-zero exit
    # if any pair has uncovered removed rules.
    if getattr(args, "check", False):
        reports = vcs_coverage_report()
        any_uncovered = False
        for r in reports:
            print(r.format())
            if r.uncovered_removed:
                any_uncovered = True
        return 1 if any_uncovered else 0

    from_ref = args.from_ref or _default_from_ref()
    to_ref = args.to_ref or "HEAD"

    try:
        migrate_fn = compose_migration(from_ref, to_ref)
        inputs = _walk_inputs(args.paths)
    except (MigrateError, MigrationError) as exc:
        print(f"qvr migrate: {exc}", file=sys.stderr)
        return 2

    if not inputs:
        print("qvr migrate: no .qvr files to migrate", file=sys.stderr)
        return 2

    out_root = Path(args.output) if args.output is not None else None
    changed = 0
    for src_path in inputs:
        source = src_path.read_bytes()
        migrated = migrate_fn(source)
        if migrated == source:
            continue
        changed += 1
        prefix = "would migrate" if args.dry_run else "migrated"
        print(f"{prefix} {src_path}")
        if args.dry_run:
            continue
        target = (out_root / src_path.name) if out_root is not None else src_path
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(migrated)

    if changed == 0:
        print(
            f"qvr migrate: {len(inputs)} file(s) already at {to_ref}",
        )
    return 0

migrations

Registry of structural one-hop .qvr source migrations.

Each adjacent-pair migrator is a callable bytes -> bytes built on the panproto pipeline in quivers.cli.migrations._common: parse with the source revision's tree-sitter parser, build the target schema by walking the source parse tree and emitting target vertices/edges/constraints, then emit canonical target bytes via the target revision's emit_pretty. The third stage is grammar- bound by panproto: a target schema that does not satisfy the target grammar's rules cannot emit.

The CHAIN tuple lists releases in chronological order with "HEAD" always last. Each adjacent pair (CHAIN[i], CHAIN[i+1]) must have a registered migrator in MIGRATORS. compose_migration walks CHAIN between any two listed releases and composes the intermediate hops into a single bytes -> bytes callable.

To add the next release:

1. Tag the release in git.
2. Run python grammars/qvr/vcs/build_schemas.py and python grammars/qvr/vcs/build_parsers.py.
3. Rename the current v<latest>_to_head migrator module to v<latest>_to_v<next> (it now migrates between two pinned revisions), and write a new v<next>_to_head covering the next batch of grammar changes if any.
4. Append the new release to CHAIN and register the migrator(s) in MIGRATORS.

The qvr migrate CLI consumes compose_migration and is agnostic to which pairs are registered.

BlameReport dataclass

BlameReport(rule: str, introduced_at_commit: str | None, introduced_at_tag: str | None, last_present_at_commit: str | None, last_present_at_tag: str | None)

Where in the grammar's history a rule first appeared (or was last seen).

DiffCoverageReport dataclass

DiffCoverageReport(from_ref: str, to_ref: str, added_rules: tuple[str, ...], removed_rules: tuple[str, ...], uncovered_removed: tuple[str, ...])

The schema diff between two revisions, classified against a hop's declared converter dispatch table.

uncovered_removed instance-attribute

uncovered_removed: tuple[str, ...]

Rules removed at the target revision that are also absent from declared_converters. Each one is a source-side vertex kind that the migrator's dispatch will silently pass through as a structural clone — and emit_pretty will then either misrender or drop. These are the actionable misses.

format

format() -> str

Render the report for CLI display.

Source code in src/quivers/cli/migrations/_vcs.py

def format(self) -> str:
    """Render the report for CLI display."""
    lines = [f"{self.from_ref} -> {self.to_ref}:"]
    if not self.added_rules and not self.removed_rules:
        lines.append("    (grammar identical; no diff)")
        return "\n".join(lines)
    if self.removed_rules:
        lines.append(f"    removed: {', '.join(self.removed_rules)}")
    if self.added_rules:
        lines.append(f"    added:   {', '.join(self.added_rules)}")
    if self.uncovered_removed:
        lines.append(
            f"    UNCOVERED removed rules (no converter): "
            f"{', '.join(self.uncovered_removed)}",
        )
    elif self.removed_rules:
        lines.append("    all removed rules have converters [OK]")
    return "\n".join(lines)

MigrationError

Bases: Exception

Raised when the requested migration cannot be composed.

blame_kind

blame_kind(rule: str) -> BlameReport

Report when a tree-sitter rule was introduced or removed in the grammar's VCS history. Used by the migrator's failure path to point the user at the specific release that needs a new converter.

Source code in src/quivers/cli/migrations/_vcs.py

def blame_kind(rule: str) -> BlameReport:
    """Report when a tree-sitter rule was introduced or removed in
    the grammar's VCS history. Used by the migrator's failure
    path to point the user at the specific release that needs a
    new converter."""
    repo = _open_repo()
    head = repo.head() or ""

    introduced_commit: str | None = None
    try:
        info = repo.blame_vertex(head, rule)
        introduced_commit = str(info.get("commit"))  # type: ignore[union-attr]
    except Exception:
        introduced_commit = None

    introduced_tag = (
        _tag_for_commit(repo, introduced_commit) if introduced_commit else None
    )

    # Walk log oldest-first, find the last commit whose schema
    # contains the rule. If the current HEAD schema contains it,
    # introduced_commit is the answer to both "introduced" and
    # "last present." Otherwise scan history.
    last_commit: str | None = None
    for entry in repo.log():
        cid = str(entry["id"])
        try:
            schema = repo.schema_at(cid)
        except Exception:
            continue
        if schema.has_vertex(rule):
            last_commit = cid
            break  # log() is newest-first; the first hit is the last presence.
    last_tag = _tag_for_commit(repo, last_commit) if last_commit else None

    return BlameReport(
        rule=rule,
        introduced_at_commit=introduced_commit,
        introduced_at_tag=introduced_tag,
        last_present_at_commit=last_commit,
        last_present_at_tag=last_tag,
    )

diff_coverage

diff_coverage(from_ref: str, to_ref: str, declared_converters: frozenset[str]) -> DiffCoverageReport

Compute the schema diff between from_ref and to_ref in the VCS, classified against the set of source-side rule names the migrator declares it can handle.

A rule appearing in from_ref's schema but not in to_ref's is a "removed" rule. If a removed rule is a top- level declaration kind that the migrator's _DECL_CONVERTERS dict does not list, the migrator will silently pass it through (likely producing incorrect output); these surface in uncovered_removed.

Identity hops (where from_ref and to_ref share a commit) report no diff.

Source code in src/quivers/cli/migrations/_vcs.py

def diff_coverage(
    from_ref: str,
    to_ref: str,
    declared_converters: frozenset[str],
) -> DiffCoverageReport:
    """Compute the schema diff between ``from_ref`` and ``to_ref``
    in the VCS, classified against the set of source-side rule
    names the migrator declares it can handle.

    A rule appearing in ``from_ref``'s schema but not in
    ``to_ref``'s is a "removed" rule. If a removed rule is a top-
    level declaration kind that the migrator's
    ``_DECL_CONVERTERS`` dict does not list, the migrator will
    silently pass it through (likely producing incorrect output);
    these surface in ``uncovered_removed``.

    Identity hops (where ``from_ref`` and ``to_ref`` share a
    commit) report no diff.
    """
    repo = _open_repo()
    from_id = commit_id_for(from_ref) or _resolve_via_chain(repo, from_ref)
    to_id = commit_id_for(to_ref) or _resolve_via_chain(repo, to_ref)
    if from_id == to_id:
        return DiffCoverageReport(
            from_ref=from_ref,
            to_ref=to_ref,
            added_rules=(),
            removed_rules=(),
            uncovered_removed=(),
        )
    src_schema = repo.schema_at(from_id)
    tgt_schema = repo.schema_at(to_id)
    diff = panproto.diff_schemas(src_schema, tgt_schema)
    diff_dict = diff.to_dict()
    added = tuple(sorted(diff_dict.get("added_vertices", [])))
    removed = tuple(sorted(diff_dict.get("removed_vertices", [])))
    uncovered = tuple(r for r in removed if r not in declared_converters)
    return DiffCoverageReport(
        from_ref=from_ref,
        to_ref=to_ref,
        added_rules=added,
        removed_rules=removed,
        uncovered_removed=uncovered,
    )

commit_id

commit_id(ref: str) -> str

Resolve a release name to its panproto VCS commit id. Cached.

Source code in src/quivers/cli/migrations/__init__.py

def commit_id(ref: str) -> str:
    """Resolve a release name to its panproto VCS commit id. Cached."""
    if not _COMMIT_IDS:
        _COMMIT_IDS.update(_build_commit_index())
    return _COMMIT_IDS.get(ref, "")

vcs_coverage_report

vcs_coverage_report() -> list[DiffCoverageReport]

Run the panproto-VCS-driven coverage check across every adjacent pair in CHAIN. Each report carries the schema diff and the set of removed source rules not covered by the corresponding hop's SOURCE_RULE_COVERAGE.

Use this from qvr migrate --check (CLI) or from a CI test to catch migrators that drift behind grammar changes.

Source code in src/quivers/cli/migrations/__init__.py

def vcs_coverage_report() -> list[DiffCoverageReport]:
    """Run the panproto-VCS-driven coverage check across every
    adjacent pair in `CHAIN`. Each report carries the schema
    diff and the set of removed source rules not covered by the
    corresponding hop's ``SOURCE_RULE_COVERAGE``.

    Use this from ``qvr migrate --check`` (CLI) or from a CI test
    to catch migrators that drift behind grammar changes."""
    reports: list[DiffCoverageReport] = []
    for i in range(len(CHAIN) - 1):
        from_ref = CHAIN[i]
        to_ref = CHAIN[i + 1]
        decl = COVERAGE.get((from_ref, to_ref), frozenset())
        from_id = commit_id(from_ref)
        to_id = commit_id(to_ref)
        if from_id == to_id:
            reports.append(
                DiffCoverageReport(
                    from_ref=from_ref,
                    to_ref=to_ref,
                    added_rules=(),
                    removed_rules=(),
                    uncovered_removed=(),
                )
            )
            continue
        # Re-implement diff_coverage's body using the resolved
        # commit ids directly to avoid double-resolution.
        from quivers.cli.migrations._vcs import _open_repo
        import panproto

        repo = _open_repo()
        src_schema = repo.schema_at(from_id)
        tgt_schema = repo.schema_at(to_id)
        diff_dict = panproto.diff_schemas(src_schema, tgt_schema).to_dict()
        added = tuple(sorted(diff_dict.get("added_vertices", [])))
        removed = tuple(sorted(diff_dict.get("removed_vertices", [])))
        uncovered = tuple(r for r in removed if r not in decl)
        reports.append(
            DiffCoverageReport(
                from_ref=from_ref,
                to_ref=to_ref,
                added_rules=added,
                removed_rules=removed,
                uncovered_removed=uncovered,
            )
        )
    return reports

compose_migration

compose_migration(from_ref: str, to_ref: str) -> _Migrator

Return a single bytes -> bytes callable that composes every adjacent-pair migrator between from_ref and to_ref on CHAIN.

Source code in src/quivers/cli/migrations/__init__.py

def compose_migration(from_ref: str, to_ref: str) -> _Migrator:
    """Return a single ``bytes -> bytes`` callable that composes
    every adjacent-pair migrator between ``from_ref`` and ``to_ref``
    on `CHAIN`."""
    pairs = _chain_slice(from_ref, to_ref)
    if not pairs:
        return _identity
    missing = [pair for pair in pairs if pair not in MIGRATORS]
    if missing:
        raise MigrationError(
            "missing migrator(s) for pair(s): "
            + ", ".join(f"{a} -> {b}" for a, b in missing),
        )
    migrators = [MIGRATORS[pair] for pair in pairs]

    def _composite(source: bytes) -> bytes:
        for step in migrators:
            source = step(source)
        return source

    return _composite

available_targets

available_targets(from_ref: str) -> tuple[str, ...]

Return every revision reachable forward from from_ref on CHAIN (inclusive of from_ref itself).

Source code in src/quivers/cli/migrations/__init__.py

def available_targets(from_ref: str) -> tuple[str, ...]:
    """Return every revision reachable forward from ``from_ref`` on
    `CHAIN` (inclusive of ``from_ref`` itself)."""
    if from_ref not in CHAIN:
        return ()
    i = CHAIN.index(from_ref)
    return CHAIN[i:]