Skip to content

API

The pipeline composes the stages; everything below it is a total function over plain data and can be used on its own.

Compose the stages. This module holds no logic of its own.

Every stage is a total function over plain data, so composition is ordinary function application and there is nothing here to test except that the stages fit together. That is not nothing: the failures worth catching at this level are the ones no single stage can see.

The important one is identity agreement. Five producers mint IRIs — the extraction pass, name resolution, member-write resolution, the def-use graph and the control-flow graph. If two of them name one entity differently, the graph contains two disconnected nodes where it should contain one, and every unit test still passes because each producer is self-consistent.

analyze(source, *, module='', file='<source>')

Run every observation and inference stage over one module.

Parameters:

Name Type Description Default
source str

The module's text.

required
module str

Its dotted import path.

''
file str

A label for spans.

'<source>'

Returns:

Type Description
dict

facts, names, writes, flow and plan: what was seen, what it refers to, what was assigned to which typed member, where each value came from, and how control moves between steps.

Source code in src\awl\pipeline.py
def analyze(source: str, *, module: str = "", file: str = "<source>") -> dict[str, Any]:
    """Run every observation and inference stage over one module.

    Parameters
    ----------
    source : str
        The module's text.
    module : str
        Its dotted import path.
    file : str, optional
        A label for spans.

    Returns
    -------
    dict
        ``facts``, ``names``, ``writes``, ``flow`` and ``plan``: what was seen,
        what it refers to, what was assigned to which typed member, where each
        value came from, and how control moves between steps.
    """
    observed = facts.extract(source, module=module, file=file)
    return {
        "file": file,
        "module": module,
        "facts": observed,
        "names": resolve(observed),
        "writes": resolve_writes(observed),
        "flow": dataflow.analyze(source, module=module, file=file),
        "plan": controlflow.analyze(source, module=module, file=file),
    }

resolve_module(module, origin)

Return the absolute path an import names.

from .params import ChargeParam inside battery.procedure names battery.params. Extraction records the origin exactly as written, because that is what the source says; turning it into a path is an inference and belongs here.

Source code in src\awl\pipeline.py
def resolve_module(module: str, origin: str) -> str:
    """Return the absolute path an import names.

    ``from .params import ChargeParam`` inside ``battery.procedure`` names
    ``battery.params``. Extraction records the origin exactly as written,
    because that is what the source says; turning it into a path is an
    inference and belongs here.
    """
    if not origin.startswith("."):
        return origin
    depth = len(origin) - len(origin.lstrip("."))
    parts = module.split(".")[: -depth or None]
    tail = origin.lstrip(".")
    return ".".join([*parts, tail]) if tail else ".".join(parts)

to_ast_doc(source, *, module='', profile='ast', file='<source>', index=None)

Parse, elide by profile, and collapse resolved constructors.

Parameters:

Name Type Description Default
source str

The module's text.

required
module str

Its dotted import path.

''
profile str

One of awl.vocab.PROFILES.

'ast'
file str

A label for spans.

'<source>'
index dict

Module path to source, for the modules this one imports from. A parameter object is nearly always defined in another file, so without it the collapse almost never fires.

None

Returns:

Type Description
dict

An AstDoc. Only constructors whose callee resolved are collapsed; an unresolved name stays a plain call.

Source code in src\awl\pipeline.py
def to_ast_doc(
    source: str,
    *,
    module: str = "",
    profile: str = "ast",
    file: str = "<source>",
    index: dict[str, str] | None = None,
) -> Any:
    """Parse, elide by profile, and collapse resolved constructors.

    Parameters
    ----------
    source : str
        The module's text.
    module : str
        Its dotted import path.
    profile : str, optional
        One of ``awl.vocab.PROFILES``.
    file : str, optional
        A label for spans.
    index : dict, optional
        Module path to source, for the modules this one imports from. A
        parameter object is nearly always defined in another file, so without
        it the collapse almost never fires.

    Returns
    -------
    dict
        An ``AstDoc``. Only constructors whose callee resolved are collapsed;
        an unresolved name stays a plain call.
    """
    observed = facts.extract(source, module=module, file=file)
    return _tree(source, observed, module=module, profile=profile, index=index or {})

to_compact(source, *, module='', profile='ast', spans=False, file='<source>', index=None)

Run the chain and return the editor model.

Parameters:

Name Type Description Default
spans bool

Keep source spans, which are the join key for in-place patching and for the trace overlay. Not needed to regenerate code.

False
Source code in src\awl\pipeline.py
def to_compact(
    source: str,
    *,
    module: str = "",
    profile: str = "ast",
    spans: bool = False,
    file: str = "<source>",
    index: dict[str, str] | None = None,
) -> Any:
    """Run the chain and return the editor model.

    Parameters
    ----------
    spans : bool, optional
        Keep source spans, which are the join key for in-place patching and
        for the trace overlay. Not needed to regenerate code.
    """
    observed = facts.extract(source, module=module, file=file)
    tree = _tree(source, observed, module=module, profile=profile, index=index or {}, spans=spans)
    return compact.encode(tree, keep_spans=spans)

to_document(source, *, module='', profile='ast', file='<source>', index=None, layers=None, spans=None, identities=None, trivia=None)

Build the JSON-LD document a profile calls for.

Parameters:

Name Type Description Default
source str

The module's text.

required
module str

Its dotted import path.

''
profile str

A named set of generator parameters: which wrappers are transparent, which types go opaque, whether keywords fold, and which lookups run. Defaults to ast, the profile whose obligation is complete value provenance. The reduced profiles are paused, and until each is defined by the class of question it must answer, choosing one only makes the document smaller and no better.

'ast'
file str

A label for spans.

'<source>'
index dict

Module path to source, for the modules this one imports from.

None
layers tuple of str

Overrides the profile's lookups, for a caller who wants one of them on its own. Asking for ("document",) gives the tree and nothing derived from it, which is what a reader comparing notations needs.

None
spans bool

Overrides the profile's :data:awl.vocab.MATERIALIZES_SPANS. On for every profile, because a span locates a node in the file it came from, which is what a patch and a trace overlay are written against.

None
identities bool

Overrides the profile's :data:awl.vocab.MATERIALIZES_IDENTITIES. On for every profile: it names each statement with the identity the plan mints for it, so the tree's statement and the plan's step are one node rather than two that happen to sit at the same coordinates. Needs spans, which is what matches the two sides at build time.

None
trivia bool

Overrides the profile's :data:awl.vocab.MATERIALIZES_TRIVIA. On for every profile: it carries the comment written about each statement, which the syntax tree has no node for. Needs spans too, since a comment is placed against the span of the statement it describes.

None

Returns:

Type Description
dict

A @context and a @graph. This is the artefact; RDF is one serialization of it and the editor model is the first entry in it.

Raises:

Type Description
ValueError

If a layer is not one of :data:awl.vocab.LAYERS, rather than quietly returning less than was asked for.

Source code in src\awl\pipeline.py
def to_document(
    source: str,
    *,
    module: str = "",
    profile: str = "ast",
    file: str = "<source>",
    index: dict[str, str] | None = None,
    layers: tuple[str, ...] | None = None,
    spans: bool | None = None,
    identities: bool | None = None,
    trivia: bool | None = None,
) -> dict[str, Any]:
    """Build the JSON-LD document a profile calls for.

    Parameters
    ----------
    source : str
        The module's text.
    module : str
        Its dotted import path.
    profile : str, optional
        A named set of generator parameters: which wrappers are transparent,
        which types go opaque, whether keywords fold, and which lookups run.
        Defaults to ``ast``, the profile whose obligation is complete value
        provenance. The reduced profiles are paused, and until each is defined
        by the class of question it must answer, choosing one only makes the
        document smaller and no better.
    file : str, optional
        A label for spans.
    index : dict, optional
        Module path to source, for the modules this one imports from.
    layers : tuple of str, optional
        Overrides the profile's lookups, for a caller who wants one of them on
        its own. Asking for ``("document",)`` gives the tree and nothing
        derived from it, which is what a reader comparing notations needs.
    spans : bool, optional
        Overrides the profile's :data:`awl.vocab.MATERIALIZES_SPANS`. On for
        every profile, because a span locates a node in the file it came from,
        which is what a patch and a trace overlay are written against.
    identities : bool, optional
        Overrides the profile's :data:`awl.vocab.MATERIALIZES_IDENTITIES`. On
        for every profile: it names each statement with the identity the plan
        mints for it, so the tree's statement and the plan's step are one node
        rather than two that happen to sit at the same coordinates. Needs
        ``spans``, which is what matches the two sides at build time.
    trivia : bool, optional
        Overrides the profile's :data:`awl.vocab.MATERIALIZES_TRIVIA`. On for
        every profile: it carries the comment written about each statement,
        which the syntax tree has no node for. Needs ``spans`` too, since a
        comment is placed against the span of the statement it describes.

    Returns
    -------
    dict
        A ``@context`` and a ``@graph``. This is the artefact; RDF is one
        serialization of it and the editor model is the first entry in it.

    Raises
    ------
    ValueError
        If a layer is not one of :data:`awl.vocab.LAYERS`, rather than quietly
        returning less than was asked for.
    """
    selected = vocab.LOOKUPS[profile] if layers is None else layers
    located = vocab.MATERIALIZES_SPANS[profile] if spans is None else spans
    named = vocab.MATERIALIZES_IDENTITIES[profile] if identities is None else identities
    noted = vocab.MATERIALIZES_TRIVIA[profile] if trivia is None else trivia
    unknown = set(selected) - set(vocab.LAYERS)
    if unknown:
        raise ValueError(f"unknown layers {sorted(unknown)}; expected some of {list(vocab.LAYERS)}")

    observed = facts.extract(source, module=module, file=file)
    # The same resolution the collapse used. Building the context from this
    # module's types alone left an imported class resolving through @vocab, so
    # the node was typed awl:ChargeParam while the collapse had named it
    # py/tier3_oold.params/ChargeParam: one entity, two IRIs.
    types, _ = _collapsible(observed, module, index or {})
    built = context.build_context(list(types.values()))

    graph: list[Any] = []
    for part in _layers(
        source,
        observed,
        selected,
        module=module,
        profile=profile,
        file=file,
        index=index,
        spans=located,
        # A statement is matched to its step by span, so without one there is
        # nothing to name and asking for both is asking for neither.
        identities=named and located,
        # Same bargain: a comment is placed against the span of the statement
        # it was written about.
        trivia=noted and located,
    ):
        graph.extend(part.get("@graph", [part]))
    return {"@context": built["@context"], "@graph": graph}

to_graph(source, *, module='', profile='ast', file='<source>', index=None, layers=None, spans=None, identities=None, trivia=None)

Serialize :func:to_document as RDF, taking the same parameters.

Returns:

Type Description
Graph
Notes

There is nothing here but a change of notation. What the graph contains is decided by the profile when the document is built, so the two cannot say different things about one program.

Source code in src\awl\pipeline.py
def to_graph(
    source: str,
    *,
    module: str = "",
    profile: str = "ast",
    file: str = "<source>",
    index: dict[str, str] | None = None,
    layers: tuple[str, ...] | None = None,
    spans: bool | None = None,
    identities: bool | None = None,
    trivia: bool | None = None,
):
    """Serialize :func:`to_document` as RDF, taking the same parameters.

    Returns
    -------
    rdflib.Graph

    Notes
    -----
    There is nothing here but a change of notation. What the graph contains is
    decided by the profile when the document is built, so the two cannot say
    different things about one program.
    """
    return rdf.to_graph(
        to_document(
            source,
            module=module,
            profile=profile,
            file=file,
            index=index,
            layers=layers,
            spans=spans,
            identities=identities,
            trivia=trivia,
        )
    )

trace_run(source, call, *, module='', file='<source>')

Run call under instrumentation and join the events onto the plan.

Parameters:

Name Type Description Default
source str

The text of the module being run, so its plan can be built.

required
call callable

Invoked with no arguments.

required

Returns:

Type Description
dict

One entry per step: whether it ran, on which loop iterations, and for a branch which outcomes were observed.

Source code in src\awl\pipeline.py
def trace_run(
    source: str,
    call,
    *,
    module: str = "",
    file: str = "<source>",
) -> dict[str, Any]:
    """Run *call* under instrumentation and join the events onto the plan.

    Parameters
    ----------
    source : str
        The text of the module being run, so its plan can be built.
    call : callable
        Invoked with no arguments.

    Returns
    -------
    dict
        One entry per step: whether it ran, on which loop iterations, and for a
        branch which outcomes were observed.
    """
    from awl.trace import trace

    events = trace(call)
    plan = controlflow.analyze(source, module=module, file=file)
    return execution.join(plan, events)

Stages

Observation: what the source says, with nothing inferred.

Extraction that guesses cannot be audited, and a wrong guess is indistinguishable from an observation once it is in the graph. So this module records from battery.params import ChargeParam as an import fact with an alias hop and stops. Whether ChargeParam at a given call site is that class is awl.resolve's judgement, and it carries a confidence tier.

User code is never imported. Tier 3 of the corpus references an experimental oold branch that need not be installed, and the notation is statically visible, so static analysis is sufficient.

extract(source, *, module, file='<source>')

Read symbol facts and type info out of one module.

Parameters:

Name Type Description Default
source str

The module's text.

required
module str

Its dotted import path, used to mint identities.

required
file str

A label for spans.

'<source>'

Returns:

Type Description
dict

Conforms to symbol-facts.schema.json. types is populated only for classes deriving from LinkedBaseModel; a plain dataclass yields a declaration with fields and no TypeInfo, which is the tier 2 rung.

Notes

Never imports the module, and never decides what a name refers to.

Source code in src\awl\facts.py
def extract(source: str, *, module: str, file: str = "<source>") -> dict[str, Any]:
    """Read symbol facts and type info out of one module.

    Parameters
    ----------
    source : str
        The module's text.
    module : str
        Its dotted import path, used to mint identities.
    file : str, optional
        A label for spans.

    Returns
    -------
    dict
        Conforms to ``symbol-facts.schema.json``. ``types`` is populated only
        for classes deriving from ``LinkedBaseModel``; a plain dataclass yields
        a declaration with fields and no ``TypeInfo``, which is the tier 2 rung.

    Notes
    -----
    Never imports the module, and never decides what a name refers to.
    """
    walk = _Walk(module, file)
    walk.visit(ast.parse(source))
    return {
        "file": file,
        "module": module,
        "declarations": walk.declarations,
        "imports": walk.imports,
        "aliases": walk.aliases,
        "exports": walk.exports,
        "uses": walk.uses,
        "writes": walk.writes,
        "bindings": walk.bindings,
        "types": walk.types,
    }

Inference: bind names to identities, and say how sure we are.

Confidence is part of the data model, not metadata attached to it. Once meaning can be retrofitted by a model, declared and inferred semantics must never be indistinguishable: a pipeline that cannot mark its own output is worse than no pipeline.

Nothing here is a general resolver. The judgement is a lookup in an import table plus an optional hop through another module's exports.

resolve(facts, *, index=None, scheme='py')

Bind each use to an identity with a confidence tier.

Parameters:

Name Type Description Default
facts dict

A SymbolFacts document.

required
index dict

Module path to that module's SymbolFacts. Supplying it lets a re-export be followed one hop, which is a deduction and is therefore marked INFERRED. Without it, resolution is per file and an import binds to the module named at the import site, which is what the source literally says.

None
scheme str

Language dimension passed through to minting.

'py'

Returns:

Type Description
dict

Conforms to resolved-names.schema.json. A binding with no identity is AMBIGUOUS, never a fabricated identity: a wrong identity merges two entities, which is worse than admitting ignorance.

Source code in src\awl\resolve.py
def resolve(
    facts: dict[str, Any],
    *,
    index: dict[str, Any] | None = None,
    scheme: str = "py",
) -> dict[str, Any]:
    """Bind each use to an identity with a confidence tier.

    Parameters
    ----------
    facts : dict
        A ``SymbolFacts`` document.
    index : dict, optional
        Module path to that module's ``SymbolFacts``. Supplying it lets a
        re-export be followed one hop, which is a deduction and is therefore
        marked ``INFERRED``. Without it, resolution is per file and an import
        binds to the module named at the import site, which is what the source
        literally says.
    scheme : str, optional
        Language dimension passed through to minting.

    Returns
    -------
    dict
        Conforms to ``resolved-names.schema.json``. A binding with no identity
        is ``AMBIGUOUS``, never a fabricated identity: a wrong identity merges
        two entities, which is worse than admitting ignorance.
    """
    index = index or {}
    imports = facts.get("imports", [])
    by_name = {entry["local_name"]: entry for entry in imports}
    declarations = facts.get("declarations", [])
    declared = {entry.get("name") for entry in declarations}
    has_star = any(entry.get("is_star") for entry in imports)
    module = facts.get("module", "")

    bindings = []
    for use in facts.get("uses", []):
        name = use["local_name"]
        identity: dict[str, Any] | None = None

        if _is_overloaded(declarations, name):
            confidence = AMBIGUOUS
        elif name in by_name:
            identity, confidence = _bind_import(by_name[name], index, scheme)
        elif name in declared:
            identity = mint(scheme=scheme, module=module, symbol=name)
            confidence = EXTRACTED
        else:
            # Reached through a star import, or simply not in the scanned set.
            # Nothing in the surveyed prior art resolves a star import soundly,
            # so it stays flagged and the domain schema forbids it.
            confidence = AMBIGUOUS

        binding: dict[str, Any] = {
            "local_name": name,
            "span": use["span"],
            "confidence": confidence,
        }
        if identity is not None:
            binding["identity"] = identity
        if confidence is AMBIGUOUS and has_star:
            binding["reason"] = "reached through a star import"
        bindings.append(binding)

    return {"file": facts["file"], "bindings": bindings}

resolve_writes(facts, *, scheme='py', imported=None)

Resolve each attribute write to the member it targets.

Parameters:

Name Type Description Default
facts dict

A SymbolFacts document, whose writes carry attribute paths, whose bindings carry local aliases, and whose declarations carry the annotations to walk them against.

required
scheme str

Language dimension passed through to minting.

'py'
imported dict

Classes declared in the modules this one imports from, by name. A write walks declared fields, and a declaration lives in the module that made it, so without these an imported class resolves for the collapse and not for the write: known and unknown in the same pass.

None

Returns:

Type Description
dict

{"file": ..., "writes": [...]}. Each entry carries the path as written, the canonical member path, the member assigned, the class declaring it, the class the chain started from, the range, and a confidence tier.

Notes

This turns "an attribute named e_mod was assigned at line 74" into "the modulus of elasticity of a tensile test specimen, reached from a tensile test dataset, was assigned at line 74, by ModulusOfElasticity.from_pint".

Bindings and writes are replayed in source order within each function, so a local alias is in scope for the writes that follow it and the rooted chain is carried across it. s = dataset.specimen followed by s.e_mod = ... therefore produces the same member, owner and root as the direct form, differing only in confidence.

A hop that cannot be walked yields AMBIGUOUS rather than a guess.

Source code in src\awl\_writes.py
def resolve_writes(
    facts: dict[str, Any],
    *,
    scheme: str = "py",
    imported: dict[str, dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Resolve each attribute write to the member it targets.

    Parameters
    ----------
    facts : dict
        A ``SymbolFacts`` document, whose ``writes`` carry attribute paths,
        whose ``bindings`` carry local aliases, and whose ``declarations``
        carry the annotations to walk them against.
    scheme : str, optional
        Language dimension passed through to minting.
    imported : dict, optional
        Classes declared in the modules this one imports from, by name. A
        write walks declared fields, and a declaration lives in the module that
        made it, so without these an imported class resolves for the collapse
        and not for the write: known and unknown in the same pass.

    Returns
    -------
    dict
        ``{"file": ..., "writes": [...]}``. Each entry carries the path as
        written, the canonical member path, the member assigned, the class
        declaring it, the class the chain started from, the range, and a
        confidence tier.

    Notes
    -----
    This turns "an attribute named ``e_mod`` was assigned at line 74" into "the
    modulus of elasticity of a tensile test specimen, reached from a tensile
    test dataset, was assigned at line 74, by ``ModulusOfElasticity.from_pint``".

    Bindings and writes are replayed in **source order** within each function,
    so a local alias is in scope for the writes that follow it and the rooted
    chain is carried across it. ``s = dataset.specimen`` followed by ``s.e_mod
    = ...`` therefore produces the same member, owner and root as the direct
    form, differing only in confidence.

    A hop that cannot be walked yields ``AMBIGUOUS`` rather than a guess.
    """
    # A local declaration wins: a class declared here is what a name here
    # means, whatever a module it imports from happens to call the same thing.
    classes = {**(imported or {}), **_classes(facts)}
    declared = {
        name: entry["identity"]["iri"] for name, entry in classes.items() if (entry.get("identity") or {}).get("iri")
    }
    imports = {entry["local_name"]: entry for entry in facts.get("imports", [])}
    module = facts.get("module", "")

    def identity_for(class_name: str) -> str:
        """Return the identity of a class, as whoever declared it minted it.

        The declaration is asked first. Re-minting from the import as written
        gets the module wrong for anything reached indirectly: the range of an
        imported class's field is declared where that class is, not where it
        was imported to.
        """
        if class_name in declared:
            return declared[class_name]
        entry = imports.get(class_name)
        if entry is not None:
            return mint(scheme=scheme, module=entry["from_module"], symbol=class_name)["iri"]
        return mint(scheme=scheme, module=module, symbol=class_name)["iri"]

    def position(event: dict[str, Any]) -> tuple[int, int]:
        span = event.get("span") or {}
        return span.get("start_line", 0), span.get("start_col", 0)

    events: list[dict[str, Any]] = [
        *({"kind": "binding", **entry} for entry in facts.get("bindings", [])),
        *({"kind": "write", **entry} for entry in facts.get("writes", [])),
    ]
    events.sort(key=position)

    environments: dict[str | None, dict[str, _Bound]] = {}
    for entry in facts.get("declarations", []):
        if entry.get("kind") != "function":
            continue
        environments[entry["name"]] = {
            parameter["name"]: _Bound(target, EXTRACTED)
            for parameter in entry.get("parameters", [])
            if (target := annotation_target(parameter.get("annotation")))
        }

    resolved = []
    for event in events:
        environment = environments.setdefault(event.get("in_function"), {})

        if event["kind"] == "binding":
            bound = _bound_from(event, environment, classes)
            if bound is None:
                # A rebinding to something unknown drops the old type rather
                # than leaving a stale one in scope.
                environment.pop(event["name"], None)
            else:
                environment[event["name"]] = bound
            continue

        resolved.append(_describe(event, _walk(event["path"], environment, classes), identity_for))

    return {"file": facts["file"], "writes": resolved}

Profile-driven cutoff: order, fold, elide.

Three behaviours, not one. Order materializes the two orderings, so statement sequence is queryable without walking an RDF collection. Fold merges a keyword argument into its call while keeping its name. Elide unwraps a transparent wrapper into its parent and reduces an opaque expression to a single node holding its source text.

The distinction between folding and eliding a keyword is load-bearing. Treating keyword as transparent and splicing out its value discards the argument name, turning charge(ChargeParam(target_voltage=4.2, c_rate=0.23)) into ChargeParam(4.2, 0.23). That is a semantic change rather than a formatting one, and it also leaves the collapse with nothing to match on.

elide(doc, *, profile='ast', source='')

Apply a cutoff profile to an AST document.

Parameters:

Name Type Description Default
doc dict or list

An AstDoc.

required
profile str

One of awl.vocab.PROFILES.

'ast'
source str

Original text, used to populate source_text on opaque nodes. Without it an opaque node keeps its type but loses its expression.

''

Returns:

Type Description
dict or list

A new document; the input is not mutated.

Raises:

Type Description
KeyError

If the profile is unknown, rather than silently doing nothing.

Source code in src\awl\elide.py
def elide(doc: Any, *, profile: str = "ast", source: str = "") -> Any:
    """Apply a cutoff profile to an AST document.

    Parameters
    ----------
    doc : dict or list
        An ``AstDoc``.
    profile : str
        One of ``awl.vocab.PROFILES``.
    source : str, optional
        Original text, used to populate ``source_text`` on opaque nodes. Without
        it an opaque node keeps its type but loses its expression.

    Returns
    -------
    dict or list
        A new document; the input is not mutated.

    Raises
    ------
    KeyError
        If the profile is unknown, rather than silently doing nothing.
    """
    return _walk(doc, TRANSPARENT[profile], OPAQUE[profile], FOLDS_KEYWORDS[profile], source)

rewrap_statements(node)

Put back the Expr wrappers that elision dropped.

Python needs a statement to hold an expression in a body. Which items need one is entirely derivable — anything in a statement list that is not itself a statement — so dropping the wrapper costs nothing and restoring it needs no record of what was removed. That is what makes Expr elidable on the profile that regenerates code, where a lossy elision would not be.

Source code in src\awl\elide.py
def rewrap_statements(node: dict[str, Any]) -> dict[str, Any]:
    """Put back the ``Expr`` wrappers that elision dropped.

    Python needs a statement to hold an expression in a body. *Which* items
    need one is entirely derivable — anything in a statement list that is not
    itself a statement — so dropping the wrapper costs nothing and restoring
    it needs no record of what was removed. That is what makes ``Expr``
    elidable on the profile that regenerates code, where a lossy elision would
    not be.
    """
    statements = statement_types()
    out = dict(node)
    for field in ORDERED_FIELDS:
        items = out.get(field)
        if not isinstance(items, list):
            continue
        out[field] = [
            item
            if not isinstance(item, dict) or item.get("_type") in statements or "_type" not in item
            else {
                "_type": "Expr",
                "value": item,
                **{key: item[key] for key in _POSITION if key in item},
            }
            for item in items
        ]
    return out

unfold(doc)

Restore everything elision rewrote reversibly.

Parameters:

Name Type Description Default
doc dict or list

An AstDoc produced by :func:elide.

required

Returns:

Type Description
dict or list

A document whose calls carry keywords again and whose bodies carry their statement wrappers, ready to unparse.

Source code in src\awl\elide.py
def unfold(doc: Any) -> Any:
    """Restore everything elision rewrote reversibly.

    Parameters
    ----------
    doc : dict or list
        An ``AstDoc`` produced by :func:`elide`.

    Returns
    -------
    dict or list
        A document whose calls carry ``keywords`` again and whose bodies carry
        their statement wrappers, ready to unparse.
    """
    if isinstance(doc, list):
        return [unfold(item) for item in doc]
    if not isinstance(doc, dict):
        return doc
    return rewrap_statements(unfold_node({key: unfold(value) for key, value in doc.items()}))

unfold_node(node, *, type_key='_type')

Reverse keyword folding for one node.

Parameters:

Name Type Description Default
node dict

A node that may carry keyword_arguments.

required
type_key str

The key naming a node type. The compact form spells it type and the intermediate form _type; parameterising it keeps one implementation of how folding reverses, rather than two that drift.

'_type'

Returns:

Type Description
dict

The node with its keywords list restored, or unchanged.

Notes

Folding is a rewrite, not a loss, and this is what makes that true. Every profile folds because of it, including the faithful one: without a working inverse the profile that regenerates code could not fold, and the constructor collapse could never fire where it is most useful.

Source code in src\awl\elide.py
def unfold_node(node: dict[str, Any], *, type_key: str = "_type") -> dict[str, Any]:
    """Reverse keyword folding for one node.

    Parameters
    ----------
    node : dict
        A node that may carry ``keyword_arguments``.
    type_key : str, optional
        The key naming a node type. The compact form spells it ``type`` and
        the intermediate form ``_type``; parameterising it keeps one
        implementation of how folding reverses, rather than two that drift.

    Returns
    -------
    dict
        The node with its ``keywords`` list restored, or unchanged.

    Notes
    -----
    Folding is a rewrite, not a loss, and this is what makes that true. Every
    profile folds because of it, including the faithful one: without a working
    inverse the profile that regenerates code could not fold, and the
    constructor collapse could never fire where it is most useful.
    """
    folded = node.get("keyword_arguments")
    if not isinstance(folded, dict):
        return node
    out = {key: value for key, value in node.items() if key != "keyword_arguments"}
    position = {key: node[key] for key in _POSITION if key in node}
    # Anything already here had no name to fold under, `**kwargs` being the
    # only case. Overwriting the list rather than extending it dropped it.
    survived = list(out.get("keywords", []) or [])
    out["keywords"] = [
        {
            type_key: "keyword",
            "arg": name,
            "value": {k: v for k, v in value.items() if k not in _ORDERING} if isinstance(value, dict) else value,
            **position,
        }
        for name, value in folded.items()
    ] + survived
    return out

Type-driven collapse of call subtrees into typed nodes, and its inverse.

A call whose callee resolves to an annotated type is rewritten into a single node interpreted by that type's own context, so the call site carries meaning without the author writing any linked data.

The mechanism is what is new here, not the goal. Other systems reach the same semantic target by attaching meaning to a declared signature; this rewrites a call site's subtree using the resolved callee's context. Code-indexing systems keep the call as a call and merely point at the callee.

The collapse is a representation change rather than a projection, so it has an inverse. Without :func:expand an editor could display a workflow and never write one back.

callee_of(node)

Return the local class name a collapsed node names, or None.

Read from @type when it carries a bare term. A term resolves through the context and is therefore both the class name and, once mapped, the IRI; a CURIE or absolute IRI is emitted verbatim and names no local class. That distinction is what removes the need for a separate _callee.

Source code in src\awl\collapse.py
def callee_of(node: dict[str, Any]) -> str | None:
    """Return the local class name a collapsed node names, or None.

    Read from ``@type`` when it carries a bare term. A term resolves through
    the context and is therefore both the class name and, once mapped, the
    IRI; a CURIE or absolute IRI is emitted verbatim and names no local class.
    That distinction is what removes the need for a separate ``_callee``.
    """
    declared = node.get("@type")
    if isinstance(declared, str):
        declared = [declared]
    for candidate in declared or []:
        if isinstance(candidate, str) and ":" not in candidate and "/" not in candidate:
            return candidate
    return None

collapse(doc, *, types, resolved, embed_context=True, keep_spans=False)

Rewrite resolved constructor calls into typed nodes.

Parameters:

Name Type Description Default
doc dict or list

An AstDoc.

required
types dict

TypeInfo keyed by symbol name.

required
resolved dict

Local name to symbol name. A name absent here is not collapsed, which is how an ambiguous binding stays a plain call: a wrong type is worse than no type.

required
embed_context bool

Emit a per-node @context. Turn it off when the document already carries a context naming these types, which is the compact form: a node is then just its class name and its data.

True
keep_spans bool

Emit the source span as "@". Needed only to patch the original file in place; regenerating code from the document does not use it. Off by default, matching the compact encoder.

False

Returns:

Type Description
dict or list

A new document.

Source code in src\awl\collapse.py
def collapse(
    doc: Any,
    *,
    types: dict[str, Any],
    resolved: dict[str, str],
    embed_context: bool = True,
    keep_spans: bool = False,
) -> Any:
    """Rewrite resolved constructor calls into typed nodes.

    Parameters
    ----------
    doc : dict or list
        An ``AstDoc``.
    types : dict
        ``TypeInfo`` keyed by symbol name.
    resolved : dict
        Local name to symbol name. A name absent here is not collapsed, which
        is how an ambiguous binding stays a plain call: a wrong type is worse
        than no type.
    embed_context : bool, optional
        Emit a per-node ``@context``. Turn it off when the document already
        carries a context naming these types, which is the compact form: a
        node is then just its class name and its data.
    keep_spans : bool, optional
        Emit the source span as ``"@"``. Needed only to patch the original
        file in place; regenerating code from the document does not use it.
        Off by default, matching the compact encoder.

    Returns
    -------
    dict or list
        A new document.
    """
    options: dict[str, Any] = {
        "types": types,
        "resolved": resolved,
        "embed_context": embed_context,
        "keep_spans": keep_spans,
    }
    if isinstance(doc, list):
        return [collapse(item, **options) for item in doc]
    if not isinstance(doc, dict):
        return doc

    if doc.get("_type") == "Call":
        callee = doc.get("func", {}).get("id")
        symbol = resolved.get(callee) if callee else None
        info = types.get(symbol) if symbol else None
        if info is not None:
            return _typed_node(doc, info, callee, **options)

    return {key: collapse(value, **options) for key, value in doc.items()}

expand(node)

Rebuild a Call document from a typed node.

Parameters:

Name Type Description Default
node dict or list

A document that may contain collapsed nodes.

required

Returns:

Type Description
dict or list

An AstDoc with every typed node turned back into a call.

Raises:

Type Description
ValueError

If a typed node carries no _callee. Emitting a call with a guessed name would be worse than refusing.

Notes

Field insertion order is the keyword order, which JSON objects and Python dictionaries both preserve, so the restored call reads as it was written.

Source code in src\awl\collapse.py
def expand(node: Any) -> Any:
    """Rebuild a ``Call`` document from a typed node.

    Parameters
    ----------
    node : dict or list
        A document that may contain collapsed nodes.

    Returns
    -------
    dict or list
        An ``AstDoc`` with every typed node turned back into a call.

    Raises
    ------
    ValueError
        If a typed node carries no ``_callee``. Emitting a call with a guessed
        name would be worse than refusing.

    Notes
    -----
    Field insertion order is the keyword order, which JSON objects and Python
    dictionaries both preserve, so the restored call reads as it was written.
    """
    if isinstance(node, list):
        return [expand(item) for item in node]
    if not isinstance(node, dict):
        return node
    callee = callee_of(node)
    if "@type" in node and callee is None:
        raise ValueError("typed node names no local class, so it cannot be expanded; expected a bare term in @type")
    if callee is None:
        expanded = {key: expand(value) for key, value in node.items()}
        return unfold_node(expanded)

    span = node.get("span") or [1, 0, 1, 0]
    position = {
        "lineno": span[0],
        "col_offset": span[1],
        "end_lineno": span[2],
        "end_col_offset": span[3],
    }
    keywords = []
    for name, value in node.items():
        if name in _RESERVED:
            continue
        inner = expand(value) if isinstance(value, dict) else {"_type": "Constant", "value": value, **position}
        keywords.append({"_type": "keyword", "arg": name, "value": inner, **position})

    return {
        "_type": "Call",
        "func": {"_type": "Name", "id": callee, "ctx": {"_type": "Load"}, **position},
        "args": [],
        "keywords": keywords,
        **position,
    }

Compact AST codec: the editor model for AWL-LD workflows.

Both ends speak AstDoc, so this composes with the elision stage.

Three node forms, named rather than punctuated:

========================== ========================================= {"@type": "While", ...} a typed node, the same key a collapsed constructor uses {"literal": 4.2} a constant {"var": "i"} a name reference ========================== =========================================

The node type uses the JSON-LD keyword rather than a plain word, and that is a correctness requirement rather than a style choice. A node carries its fields as sibling keys, and AST field names are arbitrary identifiers: ExceptHandler has a field literally called type. Spelling the node type type silently destroyed every try/except in the standard library. @ cannot appear in a Python identifier, so the keyword namespace is the only one a field can never occupy.

literal and var stay plain words because they are complete nodes on their own and never sit beside fields, so nothing can collide with them.

An earlier revision used _, c and $, chosen to save bytes before the document was JSON-LD. They saved about four percent and cost a reader having to learn a private punctuation scheme sitting next to type, in a document meant to be edited by hand. type in particular now means one thing everywhere: an ast node type and a collapsed class name are both "what this is".

@value was the obvious JSON-LD choice for a literal and is unusable here: a value object may carry only @value, type, @language, @index and @direction, so a literal could not also carry argument_index.

decode(node)

Decode the compact form back into a live AST node.

Parameters:

Name Type Description Default
node dict or list

A CompactDoc.

required

Returns:

Type Description
AST or list

Ready for :func:ast.fix_missing_locations and :func:ast.unparse.

Notes

Every field of the target class is rebuilt from cls._fields, which is what makes the encoder's omissions safe. Copying only the present keys works on a recent interpreter and breaks on an older one: 3.11 raises AttributeError unparsing a Module with no type_ignores where 3.13 returns the source.

Source code in src\awl\compact.py
def decode(node: Any) -> Any:
    """Decode the compact form back into a live AST node.

    Parameters
    ----------
    node : dict or list
        A ``CompactDoc``.

    Returns
    -------
    ast.AST or list
        Ready for :func:`ast.fix_missing_locations` and :func:`ast.unparse`.

    Notes
    -----
    Every field of the target class is rebuilt from ``cls._fields``, which is
    what makes the encoder's omissions safe. Copying only the present keys
    works on a recent interpreter and breaks on an older one: 3.11 raises
    ``AttributeError`` unparsing a ``Module`` with no ``type_ignores`` where
    3.13 returns the source.
    """
    if isinstance(node, list):
        return [decode(item) for item in node]
    if not isinstance(node, dict):
        return node

    if "@type" in node and not _is_ast_node(node):
        return _decode_collapsed(node)
    # Minus the drop list, because a shorthand can arrive carrying a
    # materialized ordering: renumbering a statement list stamps `order` onto
    # every item, and a docstring is an item. Reading that as a typed node
    # asked a literal for its `@type` and raised, which made adding a step to
    # any body holding a docstring fail on the way back out.
    plain = set(node) - _DROP
    if "literal" in node and plain <= {"literal", *_SHORTHAND_EXTRA}:
        from awl.astdoc import from_doc

        return ast.Constant(value=from_doc(node["literal"]))
    if "var" in node and plain <= {"var", *_SHORTHAND_EXTRA}:
        return ast.Name(id=node["var"], ctx=ast.Load())

    from awl.elide import unfold_node

    node = unfold_node(node, type_key="@type")
    cls = getattr(ast, node["@type"])
    return cls(**{field: _decode_field(node, cls, field) for field in cls._fields})

dumps(doc, *, width=88, indent=1, _depth=0)

Serialize a compact document, inlining whatever fits.

Parameters:

Name Type Description Default
doc Any

A CompactDoc, or any JSON-serializable value.

required
width int

The column a line may reach before its structure is broken open.

88
indent int

Spaces per level.

1

Returns:

Type Description
str

JSON in which a small structure stays on one line and a large one breaks.

Notes

json.dumps(indent=...) puts every element of every structure on its own line, so {"var": "i"} costs three lines and a two-argument call costs a page. That is not more readable, only taller: the shape of a node is easiest to see when the node fits on one line.

Source code in src\awl\compact.py
def dumps(doc: Any, *, width: int = 88, indent: int = 1, _depth: int = 0) -> str:
    """Serialize a compact document, inlining whatever fits.

    Parameters
    ----------
    doc : Any
        A ``CompactDoc``, or any JSON-serializable value.
    width : int, optional
        The column a line may reach before its structure is broken open.
    indent : int, optional
        Spaces per level.

    Returns
    -------
    str
        JSON in which a small structure stays on one line and a large one
        breaks.

    Notes
    -----
    ``json.dumps(indent=...)`` puts every element of every structure on its own
    line, so ``{"var": "i"}`` costs three lines and a two-argument call costs a
    page. That is not more readable, only taller: the shape of a node is
    easiest to see when the node fits on one line.
    """
    pad = " " * (indent * _depth)
    inner = " " * (indent * (_depth + 1))

    if isinstance(doc, dict):
        if not doc:
            return "{}"
        flat = (
            "{"
            + ", ".join(json.dumps(key) + ": " + dumps(value, width=width, indent=0) for key, value in doc.items())
            + "}"
        )
        if len(pad) + len(flat) <= width and NEWLINE not in flat:
            return flat
        parts = [
            inner + json.dumps(key) + ": " + dumps(value, width=width, indent=indent, _depth=_depth + 1)
            for key, value in doc.items()
        ]
        return "{" + NEWLINE + ("," + NEWLINE).join(parts) + NEWLINE + pad + "}"

    if isinstance(doc, list):
        if not doc:
            return "[]"
        flat = "[" + ", ".join(dumps(item, width=width, indent=0) for item in doc) + "]"
        if len(pad) + len(flat) <= width and NEWLINE not in flat:
            return flat
        parts = [inner + dumps(item, width=width, indent=indent, _depth=_depth + 1) for item in doc]
        return "[" + NEWLINE + ("," + NEWLINE).join(parts) + NEWLINE + pad + "]"

    return json.dumps(doc)

encode(doc, *, keep_spans=False)

Encode an AstDoc into the compact form.

Parameters:

Name Type Description Default
doc dict or list

An AstDoc, from ast2json or from the elision stage.

required
keep_spans bool

Keep a span array. The editor addresses edits by span and the trace overlay attributes events by span, so without it there is no join key between the editor, the trace and the source. Costs about two percentage points of size, so the RDF projection leaves it off.

False

Returns:

Type Description
dict or list

A CompactDoc.

Notes

Rules: _type becomes type; a bare Constant becomes {"literal": value}; a bare Name becomes {"var": id}; an operator becomes its bare name; ctx, positions, nulls, empty lists and the derivable orderings are dropped.

Source code in src\awl\compact.py
def encode(doc: Any, *, keep_spans: bool = False) -> Any:
    """Encode an ``AstDoc`` into the compact form.

    Parameters
    ----------
    doc : dict or list
        An ``AstDoc``, from ``ast2json`` or from the elision stage.
    keep_spans : bool, optional
        Keep a ``span`` array. The editor addresses edits by span
        and the trace overlay attributes events by span, so without it there is
        no join key between the editor, the trace and the source. Costs about
        two percentage points of size, so the RDF projection leaves it off.

    Returns
    -------
    dict or list
        A ``CompactDoc``.

    Notes
    -----
    Rules: ``_type`` becomes ``type``; a bare ``Constant`` becomes
    ``{"literal": value}``; a bare ``Name`` becomes ``{"var": id}``; an
    operator becomes its bare name; ``ctx``, positions, nulls, empty lists and
    the derivable orderings are dropped.
    """
    if isinstance(doc, list):
        return [encode(item, keep_spans=keep_spans) for item in doc]
    if not isinstance(doc, dict):
        return doc

    node = _shorthand(doc)
    if node is None:
        return _encode_long(doc, keep_spans=keep_spans)

    if keep_spans and "lineno" in doc:
        node["span"] = _span(doc)
    return node

Return doc with each name reference pointing at what it refers to.

Parameters:

Name Type Description Default
doc dict or list or scalar

A CompactDoc.

required
bindings list of dict

What :func:awl.resolve.resolve produced for the same module.

required

Returns:

Type Description
dict or list or scalar

A new document. A var node gains refers_to when the name it holds could be resolved, and is left alone when it could not.

Notes

Joined by span where the document has one, because a name is scope-blind and two functions may both call something named run. Where it has no span the join falls back to the name, and only when every use of that name in the module resolved to one identity: an ambiguous name is left unresolved rather than pointed at whichever binding was seen last.

The editor model does not carry this. It regenerates source from the name as written, and what that name refers to is a judgement with a confidence behind it, which the names lookup records as such.

Source code in src\awl\compact.py
def link_names(doc: Any, bindings: list[dict[str, Any]]) -> Any:
    """Return *doc* with each name reference pointing at what it refers to.

    Parameters
    ----------
    doc : dict or list or scalar
        A ``CompactDoc``.
    bindings : list of dict
        What :func:`awl.resolve.resolve` produced for the same module.

    Returns
    -------
    dict or list or scalar
        A new document. A ``var`` node gains ``refers_to`` when the name it
        holds could be resolved, and is left alone when it could not.

    Notes
    -----
    Joined by span where the document has one, because a name is scope-blind
    and two functions may both call something named ``run``. Where it has no
    span the join falls back to the name, and only when every use of that name
    in the module resolved to one identity: an ambiguous name is left
    unresolved rather than pointed at whichever binding was seen last.

    The editor model does not carry this. It regenerates source from the name
    as written, and what that name refers to is a judgement with a confidence
    behind it, which the ``names`` lookup records as such.
    """
    by_span: dict[tuple[Any, Any], str] = {}
    by_name: dict[str, set[str]] = {}
    for binding in bindings:
        identity = (binding.get("identity") or {}).get("iri")
        span = binding.get("span") or {}
        if not identity:
            continue
        if span:
            by_span[(span.get("start_line"), span.get("start_col"))] = identity
        by_name.setdefault(binding.get("local_name", ""), set()).add(identity)

    unambiguous = {name: next(iter(found)) for name, found in by_name.items() if len(found) == 1}
    return _link(doc, by_span, unambiguous)

Return doc with each statement carrying the identity the plan minted for it.

Parameters:

Name Type Description Default
doc dict or list or scalar

A CompactDoc, with spans kept.

required
steps list of dict

What :func:awl.controlflow.analyze produced for the same module.

required

Returns:

Type Description
dict or list or scalar

A new document. A statement the plan recorded gains @id; every other node is left anonymous.

Notes

The plan already mints an identity for each statement, from the same ast.stmt the tree is encoded from, and then throws it away on the tree's side. The two layers then described one statement as two nodes, joined only by carrying equal span coordinates, so a query that wanted the plan's successor and the tree's callee had to match four numbers to say "the same statement". That is matching by coincidence: it is the objection to joining on a name, one level down.

With the identity on both, there is nothing to join. The tree's statement and the plan's step are the same subject, and the plan's next and when_true land on the node that carries the source.

Matched here by span, but only within one parse of one file, where the coordinates come from the same tree that produced both sides. That is a build-time lookup, not something a reader of the document has to repeat.

Source code in src\awl\compact.py
def link_steps(doc: Any, steps: list[dict[str, Any]]) -> Any:
    """Return *doc* with each statement carrying the identity the plan minted for it.

    Parameters
    ----------
    doc : dict or list or scalar
        A ``CompactDoc``, with spans kept.
    steps : list of dict
        What :func:`awl.controlflow.analyze` produced for the same module.

    Returns
    -------
    dict or list or scalar
        A new document. A statement the plan recorded gains ``@id``; every
        other node is left anonymous.

    Notes
    -----
    The plan already mints an identity for each statement, from the same
    ``ast.stmt`` the tree is encoded from, and then throws it away on the tree's
    side. The two layers then described one statement as two nodes, joined only
    by carrying equal span coordinates, so a query that wanted the plan's
    successor and the tree's callee had to match four numbers to say "the same
    statement". That is matching by coincidence: it is the objection to joining
    on a name, one level down.

    With the identity on both, there is nothing to join. The tree's statement
    and the plan's step are the same subject, and the plan's ``next`` and
    ``when_true`` land on the node that carries the source.

    Matched here by span, but only within one parse of one file, where the
    coordinates come from the same tree that produced both sides. That is a
    build-time lookup, not something a reader of the document has to repeat.
    """
    by_span = {
        (span.get("start_line"), span.get("start_col"), span.get("end_line"), span.get("end_col")): step["id"]
        for step in steps
        for span in [step.get("span") or {}]
        if step.get("id") and span
    }
    return _identify(doc, by_span)

Return doc with each statement carrying the comment written about it.

Parameters:

Name Type Description Default
doc dict or list or scalar

A CompactDoc, with spans kept.

required
source str

The text the document was built from.

required

Returns:

Type Description
dict or list or scalar

A new document. A statement gains comment; a block whose last lines are comments gains <slot>_footer; the module gains header and footer for what belongs to the file rather than to any statement.

Notes

The syntax tree has no comment node, so without this a comment exists only in the source and no projection of the document can see it. Derived from comment positions against the spans the document already carries, rather than from a second parse: a concrete-syntax library would say the same thing and would make the document layer depend on one.

One line, and only one. A comment beside a statement is unambiguously about it. The single line directly above it, at its own indentation, usually is. A run above that is not claimed: nothing here can tell a banner from commented-out code, and claiming four lines as one statement's explanation would put dead code into a field that reads as prose.

Two places that are not statements, both of which strand a comment otherwise: the end of a block, after its last statement, and the head and tail of the file. A comment at the end of a body would otherwise attach to whatever follows the block, which is outside it.

Source code in src\awl\compact.py
def link_trivia(doc: Any, source: str) -> Any:
    """Return *doc* with each statement carrying the comment written about it.

    Parameters
    ----------
    doc : dict or list or scalar
        A ``CompactDoc``, with spans kept.
    source : str
        The text the document was built from.

    Returns
    -------
    dict or list or scalar
        A new document. A statement gains ``comment``; a block whose last lines
        are comments gains ``<slot>_footer``; the module gains ``header`` and
        ``footer`` for what belongs to the file rather than to any statement.

    Notes
    -----
    The syntax tree has no comment node, so without this a comment exists only
    in the source and no projection of the document can see it. Derived from
    comment positions against the spans the document already carries, rather
    than from a second parse: a concrete-syntax library would say the same thing
    and would make the document layer depend on one.

    **One line, and only one.** A comment beside a statement is unambiguously
    about it. The single line directly above it, at its own indentation, usually
    is. A run above that is *not* claimed: nothing here can tell a banner from
    commented-out code, and claiming four lines as one statement's explanation
    would put dead code into a field that reads as prose.

    **Two places that are not statements**, both of which strand a comment
    otherwise: the end of a block, after its last statement, and the head and
    tail of the file. A comment at the end of a body would otherwise attach to
    whatever follows the block, which is outside it.
    """
    marks = dict(_comment_marks(source))
    lines = source.splitlines()
    stamped = _attach(doc, marks, lines, nearest=_nearest_on_line(doc))
    if not isinstance(stamped, dict):
        return stamped

    body = stamped.get("body")
    first = _first_span(body)
    last = _last_span(body)
    header = [_note(line, marks.pop(line)) for line in sorted(marks) if first is None or line < first[0]]
    footer = [_note(line, marks.pop(line)) for line in sorted(marks) if last is not None and line > last[2]]

    # The last header line, when it sits directly above the first statement with
    # no blank between, is that statement's note and not the file's. Leaving it
    # in the header shows a file with a comment over its first block and a block
    # with no comment, which reads as a defect rather than as a rule.
    first_note = body[0].get("comment") if body and isinstance(body[0], dict) else None
    directly_above = (
        header
        and first is not None
        and header[-1]["span"][0] == first[0] - 1
        # At the statement's own column, which is the rule `_note_for` applies
        # everywhere else. Without it an indented comment over a column-zero
        # statement became its note here and stayed invisible to
        # `writeback.trivia`, so the document showed a note the editor could
        # not see and writing one added a second.
        and header[-1]["span"][1] == first[1]
    )
    if directly_above and first_note is None and body:
        moved = header.pop()
        body[0]["comment"] = {"text": moved["text"], "where": "above", "span": moved["span"]}

    if header:
        stamped["header"] = header
    if footer:
        stamped["footer"] = footer
    return stamped

name_spans(doc, *, file='')

Return doc with each span's four numbers named.

Parameters:

Name Type Description Default
doc dict or list or scalar

A CompactDoc, whose spans are [line, col, end_line, end_col].

required
file str

Recorded in each span, since a position means nothing without it.

''

Returns:

Type Description
dict or list or scalar

A new document.

Notes

Two shapes for one fact, which is worth stating rather than hiding. The editor holds a span as four numbers because it patches source with them and reads them by position. A document that carries meaning cannot: a bare array says which four numbers, never which is the line and which the column, and a consumer has to know the order by convention. The plan and the def-use graph already name them, so naming them here is what keeps one spelling across the whole document rather than two.

Source code in src\awl\compact.py
def name_spans(doc: Any, *, file: str = "") -> Any:
    """Return *doc* with each span's four numbers named.

    Parameters
    ----------
    doc : dict or list or scalar
        A ``CompactDoc``, whose spans are ``[line, col, end_line, end_col]``.
    file : str, optional
        Recorded in each span, since a position means nothing without it.

    Returns
    -------
    dict or list or scalar
        A new document.

    Notes
    -----
    Two shapes for one fact, which is worth stating rather than hiding. The
    editor holds a span as four numbers because it patches source with them and
    reads them by position. A document that carries meaning cannot: a bare
    array says which four numbers, never which is the line and which the
    column, and a consumer has to know the order by convention. The plan and
    the def-use graph already name them, so naming them here is what keeps one
    spelling across the whole document rather than two.
    """
    if isinstance(doc, list):
        return [name_spans(item, file=file) for item in doc]
    if not isinstance(doc, dict):
        return doc

    named = {key: name_spans(value, file=file) for key, value in doc.items() if key != "span"}
    span = doc.get("span")
    if isinstance(span, list) and len(span) == 4:
        start_line, start_col, end_line, end_col = span
        named["span"] = {
            "file": file,
            "start_line": start_line,
            "start_col": start_col,
            "end_line": end_line,
            "end_col": end_col,
        }
    elif span is not None:
        named["span"] = name_spans(span, file=file)
    return named

number_items(doc)

Return doc with every ordered statement carrying its sibling slot.

Parameters:

Name Type Description Default
doc dict or list or scalar

A CompactDoc.

required

Returns:

Type Description
dict or list or scalar

A new document.

Notes

The editor model leaves the number out, because the array already says it and an editor that reorders a body would leave it stale. A document that is going to be projected cannot: @container: @list yields an RDF collection, a collection yields members rather than positions, and SPARQL 1.1 property paths have only *, + and ?, so a query downstream cannot count the rdf:rest hops back. This integer is the query surface, added at the one point where position stops being recoverable.

Source code in src\awl\compact.py
def number_items(doc: Any) -> Any:
    """Return *doc* with every ordered statement carrying its sibling slot.

    Parameters
    ----------
    doc : dict or list or scalar
        A ``CompactDoc``.

    Returns
    -------
    dict or list or scalar
        A new document.

    Notes
    -----
    The editor model leaves the number out, because the array already says it
    and an editor that reorders a body would leave it stale. A document that is
    going to be projected cannot: ``@container: @list`` yields an RDF
    collection, a collection yields members rather than positions, and SPARQL
    1.1 property paths have only ``*``, ``+`` and ``?``, so a query downstream
    cannot count the ``rdf:rest`` hops back. This integer is the query surface,
    added at the one point where position stops being recoverable.
    """
    if isinstance(doc, list):
        return [number_items(item) for item in doc]
    if not isinstance(doc, dict):
        return doc

    numbered = {key: number_items(value) for key, value in doc.items()}
    for field in ORDERED_FIELDS:
        sequence = numbered.get(field)
        if not isinstance(sequence, list):
            continue
        numbered[field] = [
            {**item, "order": position} if isinstance(item, dict) else item for position, item in enumerate(sequence)
        ]
    return numbered

Reaching definitions and def-use edges: the backbone of value provenance.

An ordered syntax tree says what the code is. It does not say where a value came from, because nothing links a name being read to the binding that produced it. This module adds that link, which is what makes the ast profile able to answer provenance questions rather than only structural ones.

Two properties worth stating, because both were initially got wrong.

Provenance does not need typing. Following a value back through linear["strain"].pint.to_base_units().pint.magnitude never requires knowing what those calls mean, only that the value flowed through them. Chains that are impossible to type are ordinary to trace.

A name may have several reaching definitions. Taking only the most recent one silently drops a dependency whenever a value is assigned in both arms of a branch, so definitions are tracked as sets and merged at every join.

Known limitation: a comprehension target is not a definition

ys = [f(x) for x in xs] binds x, and this module does not record it. Comprehensions have their own scope in Python 3, so binding x in the enclosing environment would be wrong, and binding it correctly needs a nested scope the walker does not model.

The chain does not break: ys still depends on xs and on f, because the dependency is taken from every name read across the whole expression. What is unresolvable is x itself, so a question of the form "which binding does this comprehension variable refer to" has no answer. The same applies to the target of a generator expression and to a walrus inside one. Worth fixing when comprehension-heavy code matters; the fix is a nested scope, not a special case.

analyze(source, *, module='', file='<source>')

Return the def-use graph of one module.

Parameters:

Name Type Description Default
source str

The module's text.

required
module str

Its dotted import path, used to mint definition identities.

''
file str

A label for spans.

'<source>'

Returns:

Type Description
dict

definitions are name bindings, each with the definitions it depends_on and the callee that produced_by it. writes are assignments to attribute paths, carrying the same two edges.

Notes

Definitions are tracked as sets and merged at every control-flow join, so a name bound in both arms of a branch reaches its uses through both. A loop body is replayed twice, which picks up a loop-carried dependency without iterating to a fixpoint.

Nothing here requires a type. That is the point: a chain that cannot be typed can still be traced.

Source code in src\awl\dataflow.py
def analyze(source: str, *, module: str = "", file: str = "<source>") -> dict[str, Any]:
    """Return the def-use graph of one module.

    Parameters
    ----------
    source : str
        The module's text.
    module : str, optional
        Its dotted import path, used to mint definition identities.
    file : str, optional
        A label for spans.

    Returns
    -------
    dict
        ``definitions`` are name bindings, each with the definitions it
        ``depends_on`` and the callee that ``produced_by`` it. ``writes`` are
        assignments to attribute paths, carrying the same two edges.

    Notes
    -----
    Definitions are tracked as sets and merged at every control-flow join, so a
    name bound in both arms of a branch reaches its uses through both. A loop
    body is replayed twice, which picks up a loop-carried dependency without
    iterating to a fixpoint.

    Nothing here requires a type. That is the point: a chain that cannot be
    typed can still be traced.
    """
    analysis = _Analysis(module, file)
    tree = ast.parse(source)
    walker = _Walker(analysis)
    walker.block(tree.body, {}, "")
    return {
        "file": file,
        "module": module,
        "definitions": analysis.definitions,
        "writes": analysis.writes,
        "conditions": analysis.conditions,
    }

The control-flow graph: which steps are connected by execution logic.

An ordered statement list says a step comes second. It does not say a step runs only if a condition held, or repeatedly, or not at all on some path. That is what a plan is, and it is what an ordered body cannot express.

Edges are typed by the reason control moves, so a query asks "which steps run when this test is true" rather than reconstructing it from nesting:

============ =========================================================== next unconditional sequence when_true the test held when_false the test did not hold, including falling past an if each_item one pass of a for exhausted the iterable ran out repeat the back edge closing a loop ============ ===========================================================

Joining with a trace answers the other half of the same profile's obligation, what actually ran, because both sides are keyed by source span.

analyze(source, *, module='', file='<source>')

Return the control-flow graph of one module.

Parameters:

Name Type Description Default
source str

The module's text.

required
module str

Its dotted import path, used to mint step identities.

''
file str

A label for spans.

'<source>'

Returns:

Type Description
dict

steps carry a neutral node type, the callee when the step is a call, the condition when it is a branch or loop, and a span. edges are typed by the reason control moves.

Notes

Each function gets its own subgraph, since control does not flow between them without a call. Steps are statements rather than expressions: that is the granularity a plan is written at, and it keeps the graph small enough to render.

A span is on every step, so a trace joins to this graph without any further machinery.

Source code in src\awl\controlflow.py
def analyze(source: str, *, module: str = "", file: str = "<source>") -> dict[str, Any]:
    """Return the control-flow graph of one module.

    Parameters
    ----------
    source : str
        The module's text.
    module : str, optional
        Its dotted import path, used to mint step identities.
    file : str, optional
        A label for spans.

    Returns
    -------
    dict
        ``steps`` carry a neutral node type, the callee when the step is a
        call, the condition when it is a branch or loop, and a span.
        ``edges`` are typed by the reason control moves.

    Notes
    -----
    Each function gets its own subgraph, since control does not flow between
    them without a call. Steps are statements rather than expressions: that is
    the granularity a plan is written at, and it keeps the graph small enough
    to render.

    A span is on every step, so a trace joins to this graph without any
    further machinery.
    """
    graph = _Graph(module, file)
    tree = ast.parse(source)
    builder = _Builder(graph)
    builder.sequence(tree.body, "")
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
            builder.sequence(node.body, node.name)
    return {"file": file, "module": module, "steps": graph.steps, "edges": graph.edges}

as_document(graph, *, spans=True)

Turn an analysed graph into JSON-LD nodes with typed edge predicates.

Parameters:

Name Type Description Default
graph dict

The output of :func:analyze.

required
spans bool

Locate each step. Turned off when the tree is in the same document and already names its statements with these identities: the step and the statement are then one node, and it would carry the same four numbers twice, once from each side.

True

Returns:

Type Description
dict

A @graph of steps, each carrying its outgoing edges as properties named for the reason control moves.

Notes

Reifying an edge as its own node would need two joins to cross one edge and would put repeat and when_true behind a literal comparison. A predicate per reason keeps a path expression usable, which is what makes "every step reachable while this test holds" a one-line query.

Source code in src\awl\controlflow.py
def as_document(graph: dict[str, Any], *, spans: bool = True) -> dict[str, Any]:
    """Turn an analysed graph into JSON-LD nodes with typed edge predicates.

    Parameters
    ----------
    graph : dict
        The output of :func:`analyze`.
    spans : bool, optional
        Locate each step. Turned off when the tree is in the same document and
        already names its statements with these identities: the step and the
        statement are then one node, and it would carry the same four numbers
        twice, once from each side.

    Returns
    -------
    dict
        A ``@graph`` of steps, each carrying its outgoing edges as properties
        named for the reason control moves.

    Notes
    -----
    Reifying an edge as its own node would need two joins to cross one edge
    and would put ``repeat`` and ``when_true`` behind a literal comparison. A
    predicate per reason keeps a path expression usable, which is what makes
    "every step reachable while this test holds" a one-line query.
    """
    outgoing: dict[str, dict[str, list[str]]] = {}
    for edge in graph["edges"]:
        outgoing.setdefault(edge["from"], {}).setdefault(edge["kind"], []).append(edge["to"])

    nodes = []
    for step in graph["steps"]:
        node: dict[str, Any] = {
            "@id": step["id"],
            # Co-typed: it is a plan step, and it is a control structure. Two
            # properties for that would be two spellings of one relation, and
            # JSON-LD expresses co-typing as a list natively.
            "_type": ["Step", step["node_type"]],
            # A label, deliberately not a type: a second language frontend
            # should add a name here, never a new type.
            "parser_type_name": step["parser_type_name"],
        }
        if spans:
            node["span"] = step["span"]
        for key in ("callee", "condition", "scope"):
            if step.get(key):
                node[key] = step[key]
        for kind, targets in outgoing.get(step["id"], {}).items():
            node[kind] = [{"@id": target} for target in targets]
        nodes.append(node)
    return {"@graph": nodes}

Projection

Build the JSON-LD context that projects an AstDoc to correct RDF.

The cheapest lever in the design: it retargets the whole projection without touching the walker. Most of the defects measured in the previous hand-written context were context work, not walker work. On the running example that context rendered a collapsed constructor as an empty blank node, because target_voltage and c_rate were not terms in it and JSON-LD drops unmapped terms.

Two halves, and they are built differently on purpose.

The static half is the AST vocabulary. It is not written here at all: it lives in ast-doc.schema.json, which is one file that is both the shape and the mapping, so the terms are defined once. Its property names are chosen (when_true reads as awl:whenTrue), which is precisely what @vocab cannot derive, so each carries an @id.

The dynamic half is one term per class, scoped to itself. A class's field names are the author's, carried verbatim, so @vocab yields exactly the right IRI and a field that needs no coercion costs no term at all. Where the class declares its own context, that is pulled rather than derived.

build_context(types=None)

Build the document-level context.

Parameters:

Name Type Description Default
types list of dict

TypeInfo documents. A field annotated float gets an explicit xsd:double coercion.

None

Returns:

Type Description
dict

Conforms to context-doc.schema.json.

Notes

Ordered statement lists get @container: @list, which JSON-LD 1.1 API section 8.3 defines as producing an RDF collection. That yields members rather than positions: recovering an index means counting rdf:rest hops, and SPARQL 1.1 property paths have only *, + and ?. So order is emitted as a materialized integer alongside, and it is the query surface.

Two alternatives are ruled out rather than overlooked. JSON-LD index maps are defined for keys that have "no semantic meaning", which disqualifies them by the specification's own framing, and SHACL sh:order is non-validating form-layout metadata.

The numeric coercion guards a cross-language hazard. Python preserves 4.0 as a float, so it costs nothing here; JavaScript cannot, since JSON.parse('4.0') === JSON.parse('4'). Without the coercion a voltage written 4.0 silently becomes an integer on the way through a JavaScript processor.

Source code in src\awl\context.py
def build_context(types: list[dict[str, Any]] | None = None) -> dict[str, Any]:
    """Build the document-level context.

    Parameters
    ----------
    types : list of dict, optional
        ``TypeInfo`` documents. A field annotated ``float`` gets an explicit
        ``xsd:double`` coercion.

    Returns
    -------
    dict
        Conforms to ``context-doc.schema.json``.

    Notes
    -----
    Ordered statement lists get ``@container: @list``, which JSON-LD 1.1 API
    section 8.3 defines as producing an RDF collection. That yields members
    rather than positions: recovering an index means counting ``rdf:rest``
    hops, and SPARQL 1.1 property paths have only ``*``, ``+`` and ``?``. So
    ``order`` is emitted as a materialized integer alongside, and it is the
    query surface.

    Two alternatives are ruled out rather than overlooked. JSON-LD index maps
    are defined for keys that have "no semantic meaning", which disqualifies
    them by the specification's own framing, and SHACL ``sh:order`` is
    non-validating form-layout metadata.

    The numeric coercion guards a **cross-language** hazard. Python preserves
    ``4.0`` as a float, so it costs nothing here; JavaScript cannot, since
    ``JSON.parse('4.0') === JSON.parse('4')``. Without the coercion a voltage
    written ``4.0`` silently becomes an integer on the way through a
    JavaScript processor.
    """
    # The static half of the vocabulary lives in ast-doc.schema.json, which is
    # an OO-LD document: one file that is both the shape and the mapping.
    # Holding it here as well would define the same terms twice, and the two
    # would drift the first time either was edited.
    from awl import contracts

    context: dict[str, Any] = dict(contracts.load_schema("ast-doc")["@context"])

    for info in types or []:
        _add_type(context, info)
    return {"@context": context}

declared_prefixes(info)

Return the prefix bindings a class declares.

A term whose value is a plain string ending in a delimiter is a namespace, which is the same rule JSON-LD 1.1 uses to decide whether a term may be used as a prefix.

Source code in src\awl\context.py
def declared_prefixes(info: dict[str, Any]) -> dict[str, str]:
    """Return the prefix bindings a class declares.

    A term whose value is a plain string ending in a delimiter is a namespace,
    which is the same rule JSON-LD 1.1 uses to decide whether a term may be
    used as a prefix.
    """
    return {
        term: value
        for term, value in _declared_terms(info.get("declared_context")).items()
        if isinstance(value, str) and value[-1:] in ("#", "/", ":")
    }

declared_type_iris(info)

Return the instance types a class declares, as the class means them.

The reference schemas write the type tag as a bare term and let the class's own context say what it resolves to: type defaults to ["QuantityValue"] and the context maps QuantityValue to qudt:QuantityValue. Read verbatim, that term would resolve against the document instead and land back on the minted Python identity, which is the one thing it is not: the class named an ontology class, and saying so is the whole point of declaring it.

The minted identity is not replaced by it. Both are true of the instance and the graph carries both, because every other producer, from member writes to the def-use graph, joins on the minted one.

Source code in src\awl\context.py
def declared_type_iris(info: dict[str, Any]) -> list[str]:
    """Return the instance types a class declares, as the class means them.

    The reference schemas write the type tag as a bare term and let the class's
    own context say what it resolves to: ``type`` defaults to
    ``["QuantityValue"]`` and the context maps ``QuantityValue`` to
    ``qudt:QuantityValue``. Read verbatim, that term would resolve against the
    document instead and land back on the minted Python identity, which is the
    one thing it is not: the class named an ontology class, and saying so is
    the whole point of declaring it.

    The minted identity is not replaced by it. Both are true of the instance
    and the graph carries both, because every other producer, from member
    writes to the def-use graph, joins on the minted one.
    """
    declared = _declared_terms(info.get("declared_context"))
    resolved = []
    for name in info.get("declared_types") or []:
        term = declared.get(name)
        if isinstance(term, dict):
            term = term.get("@id")
        resolved.append(term if isinstance(term, str) else name)
    return resolved

type_terms(info)

Return the terms that hold inside a node of this type.

Parameters:

Name Type Description Default
info dict

A TypeInfo.

required

Returns:

Type Description
dict

A JSON-LD context. Used both here, scoped under the class term, and by awl.collapse for the context it embeds in a standalone node, so the two cannot name one property two ways.

Notes

The class namespace arrives as @vocab rather than as one term per field, so a field that needs no coercion costs nothing at all. What is left is only what @vocab cannot express: a declared term, a datatype, or a reference.

A type-scoped context does not propagate to nested node objects (JSON-LD 1.1, 4.1.9), which is what keeps a plain Call nested inside a collapsed node in the AST vocabulary instead of dragging it into the class namespace.

Source code in src\awl\context.py
def type_terms(info: dict[str, Any]) -> dict[str, Any]:
    """Return the terms that hold inside a node of this type.

    Parameters
    ----------
    info : dict
        A ``TypeInfo``.

    Returns
    -------
    dict
        A JSON-LD context. Used both here, scoped under the class term, and by
        awl.collapse for the context it embeds in a standalone node, so the two
        cannot name one property two ways.

    Notes
    -----
    The class namespace arrives as ``@vocab`` rather than as one term per
    field, so a field that needs no coercion costs nothing at all. What is
    left is only what ``@vocab`` cannot express: a declared term, a datatype,
    or a reference.

    A type-scoped context does not propagate to nested node objects (JSON-LD
    1.1, 4.1.9), which is what keeps a plain ``Call`` nested inside a collapsed
    node in the AST vocabulary instead of dragging it into the class namespace.
    """
    symbol = (info.get("identity") or {}).get("symbol")
    identity = (info.get("identity") or {}).get("iri") or ""
    # A relative IRI in @vocab is not expanded, it is concatenated as written,
    # so a declared CURIE such as ex:ChargeParam would mint ex:ChargeParam#size
    # rather than failing. Better no namespace than a fabricated one.
    terms: dict[str, Any] = {"@vocab": f"{identity}#"} if "://" in identity else {}
    declared = _declared_terms(info.get("declared_context"))
    # The class term and its prefixes are the document's business, not this
    # node's: what a class is called cannot be settled from inside a node that
    # is already known to be one. See _add_type and declared_type_iris.
    prefixes = declared_prefixes(info)
    terms.update({term: value for term, value in declared.items() if term != symbol and term not in prefixes})

    for field in info.get("fields", []):
        if field["name"] in terms:
            continue
        term = _term_for(field)
        if term is not None:
            terms[field["name"]] = term
    return terms

Project a document to RDF, and read it back where that is honestly possible.

Export attaches a context and materializes schema-declared instance types, per the OO-LD rule that tooling exporting an instance must do so: a JSON-LD-only consumer sees the instance and its context but never the schema.

Import is scoped, and the scope is a refusal. The reduced profiles elide by construction, so no importer can recover what was dropped, and a silent partial reconstruction is worse than an error because the caller cannot tell which one they received.

A triple carries no attributes, so a confidence tier cannot ride on an edge. Named graphs carry it instead. That is not optional once meaning can be retrofitted by a model: a pipeline that cannot mark its own output is worse than no pipeline.

from_graph(graph, *, profile='ast')

Reconstruct a document from RDF.

Parameters:

Name Type Description Default
graph Graph
required
profile str

The profile the graph was produced at.

'ast'

Returns:

Type Description
Any

The expanded JSON-LD form.

Raises:

Type Description
ValueError

If profile elides, because the dropped nodes are unrecoverable and a partial reconstruction would be indistinguishable from a complete one.

Notes

Two preconditions for a faithful round trip are met and two are not. Ordering survives, verified: @container: @list produces an rdf:List. Unmapped keys survive, verified: an embedded @vocab keeps them. Numeric fidelity on the way back is not verified, and precision loss was measured in the other direction. Recovering the exact tree shape needs JSON-LD framing, which is not implemented here.

So this returns the expanded form, not a document identical to the input. The authoritative round trip remains document to source, never RDF to source.

Source code in src\awl\rdf.py
def from_graph(graph, *, profile: str = "ast") -> Any:
    """Reconstruct a document from RDF.

    Parameters
    ----------
    graph : rdflib.Graph
    profile : str
        The profile the graph was produced at.

    Returns
    -------
    Any
        The expanded JSON-LD form.

    Raises
    ------
    ValueError
        If *profile* elides, because the dropped nodes are unrecoverable and a
        partial reconstruction would be indistinguishable from a complete one.

    Notes
    -----
    Two preconditions for a faithful round trip are met and two are not.
    Ordering survives, verified: ``@container: @list`` produces an
    ``rdf:List``. Unmapped keys survive, verified: an embedded ``@vocab`` keeps
    them. Numeric fidelity on the way back is **not** verified, and precision
    loss was measured in the other direction. Recovering the exact tree shape
    needs JSON-LD framing, which is **not implemented here**.

    So this returns the expanded form, not a document identical to the input.
    The authoritative round trip remains document to source, never RDF to
    source.
    """
    if profile in LOSSY_PROFILES:
        raise ValueError(
            f"profile {profile!r} elides by construction, so RDF cannot be read back; "
            "only the 'ast' profile round-trips"
        )
    from pyld import jsonld

    # Serialized as n-triples, not n-quads: rdflib refuses n-quads for a store
    # that is not context-aware, and n-triples is a subset of the n-quads
    # grammar, so the parser on the other side accepts it unchanged.
    return jsonld.from_rdf(graph.serialize(format="nt"), {"useNativeTypes": True})

to_dataset(doc, *, confidence='EXTRACTED', context=None)

Return an rdflib Dataset with the statements in a tier-named graph.

Parameters:

Name Type Description Default
doc dict

An AstDoc.

required
confidence str

One of the keys of :data:CONFIDENCE_GRAPHS.

'EXTRACTED'
context dict

A ContextDoc.

None

Returns:

Type Description
Dataset

Raises:

Type Description
KeyError

If the tier is unknown, rather than silently defaulting to EXTRACTED, which would make retrofitted meaning look declared.

Source code in src\awl\rdf.py
def to_dataset(
    doc: dict[str, Any],
    *,
    confidence: str = "EXTRACTED",
    context: dict[str, Any] | None = None,
):
    """Return an rdflib Dataset with the statements in a tier-named graph.

    Parameters
    ----------
    doc : dict
        An ``AstDoc``.
    confidence : str
        One of the keys of :data:`CONFIDENCE_GRAPHS`.
    context : dict, optional
        A ``ContextDoc``.

    Returns
    -------
    rdflib.Dataset

    Raises
    ------
    KeyError
        If the tier is unknown, rather than silently defaulting to
        ``EXTRACTED``, which would make retrofitted meaning look declared.
    """
    from rdflib import Dataset, URIRef

    name = CONFIDENCE_GRAPHS[confidence]
    dataset = Dataset()
    graph = dataset.graph(URIRef(name))
    graph.parse(data=json.dumps(to_jsonld(doc, context=context)), format="json-ld")
    return dataset

to_graph(doc, *, context=None)

Return an rdflib Graph for doc.

Parameters:

Name Type Description Default
doc dict

An AstDoc.

required
context dict

A ContextDoc.

None

Returns:

Type Description
Graph
Source code in src\awl\rdf.py
def to_graph(doc: dict[str, Any], *, context: dict[str, Any] | None = None):
    """Return an rdflib Graph for *doc*.

    Parameters
    ----------
    doc : dict
        An ``AstDoc``.
    context : dict, optional
        A ``ContextDoc``.

    Returns
    -------
    rdflib.Graph
    """
    from rdflib import Graph

    graph = Graph()
    graph.parse(data=json.dumps(to_jsonld(doc, context=context)), format="json-ld")
    return graph

to_jsonld(doc, *, context=None)

Attach a context to a document.

Parameters:

Name Type Description Default
doc dict

An AstDoc, possibly containing collapsed nodes.

required
context dict

A ContextDoc. Built from the vocabulary alone when omitted.

None

Returns:

Type Description
dict

A JSON-LD document. A collapsed node keeps its own embedded context, which is what maps its fields to the type's namespace rather than to this vocabulary.

Source code in src\awl\rdf.py
def to_jsonld(doc: dict[str, Any], *, context: dict[str, Any] | None = None) -> dict[str, Any]:
    """Attach a context to a document.

    Parameters
    ----------
    doc : dict
        An ``AstDoc``, possibly containing collapsed nodes.
    context : dict, optional
        A ``ContextDoc``. Built from the vocabulary alone when omitted.

    Returns
    -------
    dict
        A JSON-LD document. A collapsed node keeps its own embedded context,
        which is what maps its fields to the type's namespace rather than to
        this vocabulary.
    """
    from awl.context import build_context

    resolved = context or build_context()
    return {"@context": resolved["@context"], **doc}

Editing and execution

The headless editing model: palette, edits, and domain validation.

Pure functions over a compact document. No browser, no rendering. Every edit returns a new document and a patch that can be applied to source, so the visual and textual representations never diverge: an edit that returned only the new document would force the writer to diff two trees and guess which source range changed.

add_step(doc, *, into, node)

Append a step to a statement list.

Parameters:

Name Type Description Default
doc dict

A compact document. Not mutated.

required
into list

Path to the statement list, e.g. ["body"]. The slot is created when it is absent.

required
node dict

The compact node to append.

required

Returns:

Type Description
tuple

The new document with order renumbered, and a structural patch.

Notes

A statement list that is empty is not in the document at all: :func:awl.compact.encode drops an empty required list, so an if with no else carries no orelse key. Appending is the edit that gives it one, and this is where that belongs: an editor that created the key itself would be writing the encoding by hand, and one that did not raised KeyError: 'orelse' on the first statement dropped into an empty branch.

Source code in src\awl\editor.py
def add_step(
    doc: dict[str, Any], *, into: list[Any], node: dict[str, Any]
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """Append a step to a statement list.

    Parameters
    ----------
    doc : dict
        A compact document. Not mutated.
    into : list
        Path to the statement list, e.g. ``["body"]``. The slot is created when
        it is absent.
    node : dict
        The compact node to append.

    Returns
    -------
    tuple
        The new document with ``order`` renumbered, and a structural patch.

    Notes
    -----
    A statement list that is empty is not in the document at all:
    :func:`awl.compact.encode` drops an empty required list, so an ``if`` with
    no ``else`` carries no ``orelse`` key. Appending is the edit that gives it
    one, and this is where that belongs: an editor that created the key itself
    would be writing the encoding by hand, and one that did not raised
    ``KeyError: 'orelse'`` on the first statement dropped into an empty branch.
    """
    out = copy.deepcopy(doc)
    holder = _descend(out, into[:-1]) if into[:-1] else out
    if into and isinstance(holder, dict) and into[-1] not in holder:
        holder[into[-1]] = []
    steps = _descend(out, into)
    steps.append(copy.deepcopy(node))
    _renumber(steps)
    return out, _structural("insert", into, code=_unparse(node))

delete_step(doc, *, path)

Remove a step, renumbering the rest.

Parameters:

Name Type Description Default
doc dict

A compact document. Not mutated.

required
path list

Path to the step, ending in its index, e.g. ["body", 0].

required

Returns:

Type Description
tuple

The new document and a structural patch.

Source code in src\awl\editor.py
def delete_step(doc: dict[str, Any], *, path: list[Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """Remove a step, renumbering the rest.

    Parameters
    ----------
    doc : dict
        A compact document. Not mutated.
    path : list
        Path to the step, ending in its index, e.g. ``["body", 0]``.

    Returns
    -------
    tuple
        The new document and a structural patch.
    """
    out = copy.deepcopy(doc)
    steps = _descend(out, path[:-1])
    removed = steps.pop(path[-1])
    _renumber(steps)
    return out, _structural("delete", path, code=_unparse(removed))

palette(schema, type_schemas=None)

Derive the placeable node types.

Parameters:

Name Type Description Default
schema dict

The workflow domain schema. Its enum constraints say what is legal.

required
type_schemas list of dict

OO-LD class schemas. Each contributes a typed entry carrying the fields its form needs, so a ChargeParam entry already knows its fields, their ranges and their widgets.

None

Returns:

Type Description
list of dict

Each has name and kind, deduplicated, in schema order. Empty when the schema declares no enum, which means the domain permits nothing rather than everything.

Notes

Read from ordinary JSON Schema enum constraints, with no vendor keyword. A document here is the syntax tree as JSON, so plain JSON Schema already expresses the restriction. That matters twice over: any standard validator enforces it, where a vendor keyword would have to be ignored by a generic validator and would turn the constraint into a hint; and the schema stays one artefact rather than encoding the same rule twice.

The two sources are both needed. The domain schema alone gives a list of names; the class schemas are what make an entry usable as a form.

Source code in src\awl\editor.py
def palette(schema: dict[str, Any], type_schemas: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
    """Derive the placeable node types.

    Parameters
    ----------
    schema : dict
        The workflow domain schema. Its ``enum`` constraints say what is legal.
    type_schemas : list of dict, optional
        OO-LD class schemas. Each contributes a typed entry carrying the fields
        its form needs, so a ``ChargeParam`` entry already knows its fields,
        their ranges and their widgets.

    Returns
    -------
    list of dict
        Each has ``name`` and ``kind``, deduplicated, in schema order. Empty
        when the schema declares no enum, which means the domain permits
        nothing rather than everything.

    Notes
    -----
    Read from ordinary JSON Schema ``enum`` constraints, with no vendor
    keyword. A document here *is* the syntax tree as JSON, so plain JSON Schema
    already expresses the restriction. That matters twice over: any standard
    validator enforces it, where a vendor keyword would have to be ignored by
    a generic validator and would turn the constraint into a hint; and the
    schema stays one artefact rather than encoding the same rule twice.

    The two sources are both needed. The domain schema alone gives a list of
    names; the class schemas are what make an entry usable as a form.
    """
    seen: set[tuple[str, str]] = set()
    entries: list[dict[str, Any]] = []
    for prop, kind in _PALETTE_SOURCES:
        found: list[str] = []
        _collect_enum(schema, prop, found)
        for name in found:
            if (kind, name) not in seen:
                seen.add((kind, name))
                entries.append({"name": name, "kind": kind})

    for type_schema in type_schemas or []:
        name = type_schema.get("title") or type_schema.get("$id", "")
        if (("type", name)) in seen:
            continue
        seen.add(("type", name))
        entries.append({
            "name": name,
            "kind": "type",
            "fields": dict(type_schema.get("properties", {})),
            "declared_types": list(type_schema.get("x-oold-instance-rdf-type", [])),
        })
    return entries

reorder(doc, *, path, frm, to)

Move a step within a statement list.

Parameters:

Name Type Description Default
doc dict

A compact document. Not mutated.

required
path list

Path to the statement list.

required
frm int

Source and destination positions.

required
to int

Source and destination positions.

required

Returns:

Type Description
tuple

The new document with order renumbered, and a structural patch.

Source code in src\awl\editor.py
def reorder(doc: dict[str, Any], *, path: list[Any], frm: int, to: int) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """Move a step within a statement list.

    Parameters
    ----------
    doc : dict
        A compact document. Not mutated.
    path : list
        Path to the statement list.
    frm, to : int
        Source and destination positions.

    Returns
    -------
    tuple
        The new document with ``order`` renumbered, and a structural patch.
    """
    out = copy.deepcopy(doc)
    steps = _descend(out, path)
    steps.insert(to, steps.pop(frm))
    _renumber(steps)
    return out, _structural("reorder", path, frm=frm, to=to)

set_literal(doc, *, path, value, span)

Replace a literal, returning the new document and a source patch.

Parameters:

Name Type Description Default
doc dict

A compact document. Not mutated.

required
path list

Keys and indices locating the literal node.

required
value Any

The new literal value.

required
span dict

start and end character offsets of the literal in the source.

required

Returns:

Type Description
tuple

The new document, and a patch that can be applied by span rather than by regenerating the file.

Source code in src\awl\editor.py
def set_literal(
    doc: dict[str, Any], *, path: list[Any], value: Any, span: dict[str, int]
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """Replace a literal, returning the new document and a source patch.

    Parameters
    ----------
    doc : dict
        A compact document. Not mutated.
    path : list
        Keys and indices locating the literal node.
    value : Any
        The new literal value.
    span : dict
        ``start`` and ``end`` character offsets of the literal in the source.

    Returns
    -------
    tuple
        The new document, and a patch that can be applied by span rather than
        by regenerating the file.
    """
    out = copy.deepcopy(doc)
    target = _descend(out, path[:-1]) if path[:-1] else out
    # The span is kept, not replaced. Writing a bare {"literal": value} left
    # the node with no position, so a canvas drawn from spans lost it and it
    # could never be edited a second time: the first edit deleted the thing
    # the next one would have pointed at.
    previous = target[path[-1]] if isinstance(target, list) or path[-1] in target else None
    replacement: dict[str, Any] = {"literal": value}
    if isinstance(previous, dict) and "span" in previous:
        replacement["span"] = previous["span"]
    target[path[-1]] = replacement
    return out, [{"start": span["start"], "end": span["end"], "text": repr(value)}]

validate_domain(node, schema)

Check one node against the workflow domain schema.

Parameters:

Name Type Description Default
node dict

The node to check.

required
schema dict

The workflow domain schema.

required

Raises:

Type Description
ValidationError

If the node is not permitted. Deliberately the library's own error and not a wrapped one, so the message names the offending enum.

Source code in src\awl\editor.py
def validate_domain(node: dict[str, Any], schema: dict[str, Any]) -> None:
    """Check one node against the workflow domain schema.

    Parameters
    ----------
    node : dict
        The node to check.
    schema : dict
        The workflow domain schema.

    Raises
    ------
    jsonschema.ValidationError
        If the node is not permitted. Deliberately the library's own error and
        not a wrapped one, so the message names the offending enum.
    """
    import jsonschema

    jsonschema.validate(node, schema)

Apply editor changes back to source without disturbing untouched text.

Two tiers. A value edit patches the original text by source span and is byte-exact everywhere else. A structural edit goes through libcst, which attaches comments and whitespace to named slots so trivia travels with the node it belongs to.

Regenerating the file from the AST is not an option for the retrofit path. ast.parse has no comment node at all, so ast.unparse discards every comment and reflows the layout. For workflows authored in the editor that is irrelevant; for editing a laboratory's existing procedure files it is fatal, and those are exactly the files this path targets.

apply_edits(source, edits)

Apply non-overlapping span edits to source.

Parameters:

Name Type Description Default
source str

Original text.

required
edits list of dict

Each has start, end and text.

required

Returns:

Type Description
str

The patched source. Every character outside an edited span is identical to the input: this is not a formatter and must not reflow, reindent or normalise quotes.

Raises:

Type Description
ValueError

If two edits overlap, which would make the result order-dependent.

Notes

Edits are applied right to left so earlier offsets stay valid.

Source code in src\awl\writeback.py
def apply_edits(source: str, edits: list[dict[str, Any]]) -> str:
    """Apply non-overlapping span edits to *source*.

    Parameters
    ----------
    source : str
        Original text.
    edits : list of dict
        Each has ``start``, ``end`` and ``text``.

    Returns
    -------
    str
        The patched source. Every character outside an edited span is
        identical to the input: this is not a formatter and must not reflow,
        reindent or normalise quotes.

    Raises
    ------
    ValueError
        If two edits overlap, which would make the result order-dependent.

    Notes
    -----
    Edits are applied right to left so earlier offsets stay valid.
    """
    ordered = sorted(edits, key=lambda edit: edit["start"])
    for earlier, later in pairwise(ordered):
        if earlier["end"] > later["start"]:
            raise ValueError(f"overlapping edits: {earlier} and {later}")
    out = source
    for edit in reversed(ordered):
        out = out[: edit["start"]] + edit["text"] + out[edit["end"] :]
    return out

comment_text(note, where, indent)

Return what a note edit writes into the range :func:trivia reported.

Parameters:

Name Type Description Default
note str

What the reader typed, with or without a #.

required
where str

above for a comment on a line of its own; anything else writes it beside the statement, which is where a first note goes.

required
indent int

The statement's own column, for a note written on a line of its own.

required

Returns:

Type Description
str

The replacement text, empty when the note was cleared.

Source code in src\awl\writeback.py
def comment_text(note: str, where: str, indent: int) -> str:
    """Return what a note edit writes into the range :func:`trivia` reported.

    Parameters
    ----------
    note : str
        What the reader typed, with or without a ``#``.
    where : str
        ``above`` for a comment on a line of its own; anything else writes it
        beside the statement, which is where a first note goes.
    indent : int
        The statement's own column, for a note written on a line of its own.

    Returns
    -------
    str
        The replacement text, empty when the note was cleared.
    """
    # One line, because a comment is one line. A note carrying a newline was
    # written through verbatim, so "note\nimport os" added an import to the
    # file and the reparse accepted it.
    body = " ".join(note.strip().lstrip("#").split())
    if not body:
        return ""
    if where == "above":
        return f"{' ' * indent}# {body}\n"
    return f"  # {body}"

comments(source)

Return every comment in source, by the line it is on.

Parameters:

Name Type Description Default
source str

The module's text.

required

Returns:

Type Description
dict

Line number to (start_col, end_col, text), the text without its # and the columns covering the comment token itself.

Notes

The tokenizer rather than a regular expression, because a # inside a string is not a comment and telling the two apart is what a tokenizer is for. Half-written source tokenizes as far as it gets and the rest is dropped, which is the same bargain an editor's source pane makes: a file that does not parse changes nothing.

Source code in src\awl\writeback.py
def comments(source: str) -> dict[int, tuple[int, int, str]]:
    """Return every comment in *source*, by the line it is on.

    Parameters
    ----------
    source : str
        The module's text.

    Returns
    -------
    dict
        Line number to ``(start_col, end_col, text)``, the text without its
        ``#`` and the columns covering the comment token itself.

    Notes
    -----
    The tokenizer rather than a regular expression, because a ``#`` inside a
    string is not a comment and telling the two apart is what a tokenizer is
    for. Half-written source tokenizes as far as it gets and the rest is
    dropped, which is the same bargain an editor's source pane makes: a file
    that does not parse changes nothing.
    """
    found: dict[int, tuple[int, int, str]] = {}
    tokens: list[tokenize.TokenInfo] = []
    with contextlib.suppress(tokenize.TokenError, IndentationError, SyntaxError, ValueError):
        tokens.extend(tokenize.generate_tokens(io.StringIO(source).readline))
    for entry in tokens:
        if entry.type == token_module.COMMENT:
            found[entry.start[0]] = (entry.start[1], entry.end[1], entry.string.lstrip("#").strip())
    return found

insert_statement(source, *, into, code, leading_comment=None)

Append a statement to a control-structure body, preserving trivia.

Parameters:

Name Type Description Default
source str

Original text.

required
into str

The control structure to insert into. Only "while" is implemented.

required
code str

The statement's expression source, e.g. "rest(600)".

required
leading_comment str or None

A comment line emitted above the new statement, including its #.

None

Returns:

Type Description
str

The patched source, with every pre-existing comment intact.

Raises:

Type Description
NotImplementedError

If into names a structure with no insertion rule, rather than silently returning the source unchanged.

Notes

Span splicing cannot do this: it has no opinion about which comment belongs to which statement, so a moved node cannot carry its own. libcst attaches trivia to named slots and can.

Source code in src\awl\writeback.py
def insert_statement(
    source: str,
    *,
    into: str,
    code: str,
    leading_comment: str | None = None,
) -> str:
    """Append a statement to a control-structure body, preserving trivia.

    Parameters
    ----------
    source : str
        Original text.
    into : str
        The control structure to insert into. Only ``"while"`` is implemented.
    code : str
        The statement's expression source, e.g. ``"rest(600)"``.
    leading_comment : str or None, optional
        A comment line emitted above the new statement, including its ``#``.

    Returns
    -------
    str
        The patched source, with every pre-existing comment intact.

    Raises
    ------
    NotImplementedError
        If *into* names a structure with no insertion rule, rather than
        silently returning the source unchanged.

    Notes
    -----
    Span splicing cannot do this: it has no opinion about which comment
    belongs to which statement, so a moved node cannot carry its own. libcst
    attaches trivia to named slots and can.
    """
    if into != "while":
        raise NotImplementedError(f"insertion into {into!r} is not implemented")

    import libcst as cst

    leading = [cst.EmptyLine(comment=cst.Comment(leading_comment))] if leading_comment else []
    statement = cst.SimpleStatementLine(body=[cst.Expr(cst.parse_expression(code))], leading_lines=leading)

    class _Insert(cst.CSTTransformer):
        def leave_While(self, original_node, updated_node):
            return updated_node.with_changes(
                body=updated_node.body.with_changes(body=[*updated_node.body.body, statement])
            )

    return cst.parse_module(source).visit(_Insert()).code

offsets(source, span)

Return the character offsets of a document span.

Parameters:

Name Type Description Default
source str

The original text.

required
span list of int

[start_line, start_col, end_line, end_col], one-based lines and zero-based columns, as a document carries them.

required

Returns:

Type Description
dict

start and end character offsets, which is what a patch needs.

Notes

Two coordinate systems meet here and neither can be dropped. A document locates a node by line and column, because that is what a parser reports and what survives an edit elsewhere in the file. A patch has to name character offsets, because that is the only way to replace a range without reflowing anything around it. Columns are byte-free: Python reports col_offset in UTF-8 bytes on some paths and in characters here, and this uses the character reading the document was built with.

Source code in src\awl\writeback.py
def offsets(source: str, span: list[int]) -> dict[str, int]:
    """Return the character offsets of a document span.

    Parameters
    ----------
    source : str
        The original text.
    span : list of int
        ``[start_line, start_col, end_line, end_col]``, one-based lines and
        zero-based columns, as a document carries them.

    Returns
    -------
    dict
        ``start`` and ``end`` character offsets, which is what a patch needs.

    Notes
    -----
    Two coordinate systems meet here and neither can be dropped. A document
    locates a node by line and column, because that is what a parser reports
    and what survives an edit elsewhere in the file. A patch has to name
    character offsets, because that is the only way to replace a range without
    reflowing anything around it. Columns are byte-free: Python reports
    ``col_offset`` in UTF-8 bytes on some paths and in characters here, and
    this uses the character reading the document was built with.
    """
    start_line, start_col, end_line, end_col = span
    lines = source.splitlines(keepends=True)
    starts = [0]
    for line in lines:
        starts.append(starts[-1] + len(line))
    return {
        "start": starts[start_line - 1] + start_col,
        "end": starts[end_line - 1] + end_col,
    }

span_of(source, predicate)

Return the character span of the first AST node satisfying predicate.

Parameters:

Name Type Description Default
source str

The original text.

required
predicate callable

Receives an :class:ast.AST node, returns bool.

required

Returns:

Type Description
tuple of int

(start, end) character offsets into source.

Raises:

Type Description
LookupError

If no node matches, so a silent no-op edit is impossible.

Notes

A keyword node spans the whole name=value pair, so an editor patching a value must target the inner node or it overwrites the parameter name too.

Source code in src\awl\writeback.py
def span_of(source: str, predicate: Callable[[ast.AST], bool]) -> tuple[int, int]:
    """Return the character span of the first AST node satisfying *predicate*.

    Parameters
    ----------
    source : str
        The original text.
    predicate : callable
        Receives an :class:`ast.AST` node, returns bool.

    Returns
    -------
    tuple of int
        ``(start, end)`` character offsets into *source*.

    Raises
    ------
    LookupError
        If no node matches, so a silent no-op edit is impossible.

    Notes
    -----
    A ``keyword`` node spans the whole ``name=value`` pair, so an editor
    patching a *value* must target the inner node or it overwrites the
    parameter name too.
    """
    import asttokens

    # Parsed here rather than via parse=True so the tree is known to exist:
    # ASTTokens.tree is Optional, and a None slipping into ast.walk would fail
    # far from the cause.
    tree = ast.parse(source)
    atok = asttokens.ASTTokens(source, tree=tree)
    for node in ast.walk(tree):
        if predicate(node):
            return atok.get_text_range(node)
    raise LookupError("no node matched the predicate")

trivia(source, span)

Return the comment the statement at span carries, and the range an edit rewrites.

Parameters:

Name Type Description Default
source str

The module's text.

required
span list or None

[start_line, start_col, end_line, end_col].

required

Returns:

Type Description
dict

text, the comment without its #; span, the range to rewrite, zero width where there is no comment yet and one would be written; and where, beside, above or "".

Notes

Here rather than in an editor, because the syntax tree has no comment node and every canvas that wants one would otherwise tokenize the source itself. The write half needs nothing new: the range this returns goes to :func:apply_edits like any other span patch, so a note survives being edited and the rest of the file does not move.

Two places, and only two. A comment beside the statement, on the line it starts on, is unambiguously about it. One above it, on the line before at the statement's own indentation, usually is.

Only the line immediately above, never a run of them. Nothing here can tell a banner from commented-out code, and claiming a four line block as one statement's explanation would put dead code into an input that writes it back as prose. A run keeps its lines as trivia and the nearest one is the note, which is a rule rather than an answer: there is no answer without a concrete-syntax tree.

Source code in src\awl\writeback.py
def trivia(source: str, span: list[int] | None) -> dict[str, Any]:
    """Return the comment the statement at *span* carries, and the range an edit rewrites.

    Parameters
    ----------
    source : str
        The module's text.
    span : list or None
        ``[start_line, start_col, end_line, end_col]``.

    Returns
    -------
    dict
        ``text``, the comment without its ``#``; ``span``, the range to rewrite,
        zero width where there is no comment yet and one would be written; and
        ``where``, ``beside``, ``above`` or ``""``.

    Notes
    -----
    Here rather than in an editor, because the syntax tree has no comment node
    and every canvas that wants one would otherwise tokenize the source itself.
    The write half needs nothing new: the range this returns goes to
    :func:`apply_edits` like any other span patch, so a note survives being
    edited and the rest of the file does not move.

    Two places, and only two. A comment **beside** the statement, on the line it
    starts on, is unambiguously about it. One **above** it, on the line before at
    the statement's own indentation, usually is.

    Only the line immediately above, never a run of them. Nothing here can tell
    a banner from commented-out code, and claiming a four line block as one
    statement's explanation would put dead code into an input that writes it
    back as prose. A run keeps its lines as trivia and the nearest one is the
    note, which is a rule rather than an answer: there is no answer without a
    concrete-syntax tree.
    """
    if not span:
        return {"text": "", "span": [], "where": ""}
    line, column = span[0], span[1]
    lines = source.splitlines()
    marks = comments(source)

    beside = marks.get(line)
    if beside is not None and beside[0] > column:
        # From the end of the code to the end of the comment, so removing a note
        # takes the run of spaces before it rather than leaving a ragged tail.
        code = len(lines[line - 1][: beside[0]].rstrip()) if line <= len(lines) else beside[0]
        return {"text": beside[2], "span": [line, code, line, beside[1]], "where": "beside"}

    above = marks.get(line - 1)
    if above is not None and above[0] == column and _only_comment(lines, line - 1):
        # The whole line, newline included, so an emptied note leaves no blank
        # line where the comment was.
        return {"text": above[2], "span": [line - 1, 0, line, 0], "where": "above"}

    # At the end of the statement, which is not always the end of its first
    # line. A statement whose first line ends inside a multi-line string put the
    # note *inside the literal*: writing one on a docstring produced
    # ``\"\"\"  # NOTE`` and silently changed what the string said, while
    # reporting success and leaving `reformats()` false.
    at = _closing_line(source, span)
    tail = len(lines[at - 1].rstrip()) if at <= len(lines) else column
    return {"text": "", "span": [at, tail, at, tail], "where": ""}

Runtime tracing: what ran, along which branch, for how many iterations.

Uses sys.settrace with opcode tracing for expression-level position, plus an interleaved sys.setprofile to name C callees the static analysis cannot see. Both hooks are independent slots and can be active at once, which is what lets a C call be attributed to a source position.

Only one trace function and one profile function exist per thread, so this conflicts with debuggers, coverage and profilers. The previous hooks are saved and restored rather than cleared, so a surrounding coverage run survives, but the two cannot observe the same code at the same time.

One consequence to read correctly: CPython does not trace the trace function itself, so every line reached from inside the callback is invisible to any coverage tool, and this module reports a low percentage no matter how well it is tested. The span algebra is therefore also exercised directly, outside the callback, where measurement works.

trace(fn, *, capture_c_calls=True, seconds=None)

Run fn under instrumentation and return the events it produced.

Parameters:

Name Type Description Default
fn callable

Invoked with no arguments.

required
capture_c_calls bool

Also install a profile hook, so C callees are named. settrace alone cannot see them, and arg.__module__ + arg.__qualname__ is exactly the identity awl.ids mints.

True
seconds float

Give up after this long, raising :class:TimeoutError from inside the run. None waits forever.

An editor places loops, so it runs code nobody has read: a palette template of while i < 10: dropped into a body that never increments i hangs the interpreter and takes the editor with it. The tracer is already on every line, so it is the one place that can stop it.

None

Returns:

Type Description
list of dict

Each conforms to trace-event.schema.json. index is a total order, so event sequence never depends on list position alone.

Notes

Overhead is large: f_trace_opcodes fires once per instruction. This is the mechanism for inspecting a run, not for production. sys.monitoring with only BRANCH and CALL enabled is the cheap alternative, at the cost of the frame object and therefore of locals.

Threads need threading.settrace separately; that is not handled here.

Source code in src\awl\trace.py
def trace(
    fn: Callable[[], Any],
    *,
    capture_c_calls: bool = True,
    seconds: float | None = None,
) -> list[dict[str, Any]]:
    """Run *fn* under instrumentation and return the events it produced.

    Parameters
    ----------
    fn : callable
        Invoked with no arguments.
    capture_c_calls : bool, optional
        Also install a profile hook, so C callees are named. ``settrace`` alone
        cannot see them, and ``arg.__module__ + arg.__qualname__`` is exactly
        the identity awl.ids mints.
    seconds : float, optional
        Give up after this long, raising :class:`TimeoutError` from inside the
        run. ``None`` waits forever.

        An editor places loops, so it runs code nobody has read: a palette
        template of ``while i < 10:`` dropped into a body that never increments
        ``i`` hangs the interpreter and takes the editor with it. The tracer is
        already on every line, so it is the one place that can stop it.

    Returns
    -------
    list of dict
        Each conforms to ``trace-event.schema.json``. ``index`` is a total
        order, so event sequence never depends on list position alone.

    Notes
    -----
    Overhead is large: ``f_trace_opcodes`` fires once per instruction. This is
    the mechanism for inspecting a run, not for production. ``sys.monitoring``
    with only ``BRANCH`` and ``CALL`` enabled is the cheap alternative, at the
    cost of the frame object and therefore of locals.

    Threads need ``threading.settrace`` separately; that is not handled here.
    """
    recorder = _Recorder()

    deadline = None if seconds is None else time.monotonic() + seconds

    def tracer(frame: FrameType, event: str, arg: Any = None):
        if deadline is not None and time.monotonic() > deadline:
            raise TimeoutError(f"the run passed {seconds:g}s and was stopped")
        frame.f_trace_opcodes = True
        if event == "call":
            recorder.enter_frame(frame)
        span = _position(frame)
        if span is not None:
            iteration = recorder.observe(frame, span)
            recorder.emit(
                kind="line" if event == "opcode" else event,
                span=span,
                iteration=iteration,
            )
        if event == "return":
            recorder.leave_frame(frame)
        return tracer

    def profiler(frame: FrameType, event: str, arg: Any) -> None:
        if event in ("c_call", "c_return"):
            module = getattr(arg, "__module__", "") or ""
            qualname = getattr(arg, "__qualname__", repr(arg))
            recorder.emit(kind=event, callee=f"{module}.{qualname}".lstrip("."))

    # Saved and restored rather than cleared: clearing would silently disable a
    # surrounding coverage run for the rest of the process.
    previous_trace = sys.gettrace()
    previous_profile = sys.getprofile()
    sys.settrace(tracer)
    if capture_c_calls:
        sys.setprofile(profiler)
    try:
        fn()
    finally:
        sys.setprofile(previous_profile)
        sys.settrace(previous_trace)
    return recorder.events

Join a recorded run onto the plan: what actually ran, and how often.

The plan says a step may run, under a condition, possibly repeatedly. A trace says what happened. Both are keyed by source span, so the join needs no extra machinery and no instrumentation of the plan.

What this makes answerable, and what nothing in the surveyed prior art records: which arm of a branch was taken on a given run, and how many times a loop body actually executed. Static workflow languages omit branching, orchestrators expand it away before recording, and systems with real control flow keep the branch and iteration logic out of their provenance graphs.

A step that never ran is reported as such rather than omitted. "Present in the plan and absent from the run" is an answer; a missing node is not.

join(plan, events)

Attribute trace events to plan steps.

Parameters:

Name Type Description Default
plan dict

A control-flow graph, as returned by awl.controlflow.analyze.

required
events list of dict

TraceEvent documents, as returned by awl.trace.trace.

required

Returns:

Type Description
dict

{"file": ..., "executions": [...]}. One entry per step, carrying whether it ran, how many trace events were attributed to it, the loop ordinals it ran under, and for a branch the outcomes recorded. A step that never ran carries executed: false.

event_count is the number of attributed events, not the number of times the step executed: opcode tracing fires many times per statement. The honest execution count is iteration_count, which is derived from the loop ordinals the tracer recorded.

Notes

Events are matched to steps by exact span first and by line second. The tracer reports expression-level positions, so an event usually falls inside a statement rather than on it; without the line-level fallback almost nothing would match, which would look like a run that did nothing.

Source code in src\awl\execution.py
def join(plan: dict[str, Any], events: list[dict[str, Any]]) -> dict[str, Any]:
    """Attribute trace events to plan steps.

    Parameters
    ----------
    plan : dict
        A control-flow graph, as returned by ``awl.controlflow.analyze``.
    events : list of dict
        ``TraceEvent`` documents, as returned by ``awl.trace.trace``.

    Returns
    -------
    dict
        ``{"file": ..., "executions": [...]}``. One entry per step, carrying
        whether it ran, how many trace events were attributed to it, the loop
        ordinals it ran under, and for a branch the outcomes recorded. A step
        that never ran carries ``executed: false``.

        ``event_count`` is the number of attributed events, **not** the number
        of times the step executed: opcode tracing fires many times per
        statement. The honest execution count is ``iteration_count``, which is
        derived from the loop ordinals the tracer recorded.

    Notes
    -----
    Events are matched to steps by exact span first and by line second. The
    tracer reports expression-level positions, so an event usually falls
    *inside* a statement rather than on it; without the line-level fallback
    almost nothing would match, which would look like a run that did nothing.
    """
    exact, by_line = _index(plan["steps"])

    observed: dict[str, dict[str, Any]] = {
        step["id"]: {
            "step": step["id"],
            "callee": step.get("callee"),
            "condition": step.get("condition"),
            "span": step["span"],
            "executed": False,
            "event_count": 0,
            "iterations": set(),
            "branch_taken": set(),
        }
        for step in plan["steps"]
    }

    for event in events:
        span = event.get("span")
        matched = exact.get(_key(span) or ()) or by_line.get(_line_key(span) or ())
        if not matched:
            continue
        for step in matched:
            record = observed[step["id"]]
            record["executed"] = True
            record["event_count"] += 1
            if event.get("iteration") is not None:
                record["iterations"].add(event["iteration"])
            if event.get("kind") == "branch" and "taken" in event:
                record["branch_taken"].add(bool(event["taken"]))

    return {
        "file": plan["file"],
        "executions": [_finish(record) for record in observed.values()],
    }