API Reference

Auto-generated reference for the public akms_learn surface. Every symbol below is exported from the package root (from akms_learn import ...). For narrative guidance see the Usage guide; for the model behind these symbols see Core Concepts.

---

Compiler

The nine-stage pipeline entry point, its result bundle, and the stage tuple.

compile_learning_source

akms_learn.compile_learning_source

compile_learning_source(
    request: LearningRequest | dict[str, Any],
    graph_path: str | Path | None = None,
    graph_slice: GraphSlice | dict[str, Any] | None = None,
    output_dir: str | Path | None = None,
    domain_pack_paths: list[str | Path] | None = None,
    source_pack_paths: list[str | Path] | None = None,
) -> CompileResult

Run the 9-stage LSP compiler pipeline.

Parameters

request: A :class:LearningRequest instance OR a raw dict that :func:normalize_request will canonicalise. graph_path: Filesystem path to a graph JSON. Mutually exclusive with graph_slice. graph_slice: Either a :class:GraphSlice instance or a raw dict payload. Mutually exclusive with graph_path. output_dir: If provided, the canonical packet JSON is written to <output_dir>/<request_hash>.json (Stage 9). Directory is created on demand. domain_pack_paths: Optional list of paths to domain_pack.yaml files (or directories containing one). Loaded into a :class:DomainPackRegistry whose descriptors are attached to the packet body. source_pack_paths: Optional list of paths to source-pack YAMLs whose descriptors are attached to the packet body.

Returns

CompileResult Wraps the validated packet, any export paths, warnings, and the stage execution log.

Raises

LearningCapabilityError If a required capability is unavailable, or a domain-pack / source-pack path is missing or invalid. PacketValidationError If the assembled packet violates a hard cross-field invariant. ValueError If neither / both of graph_path and graph_slice are given.

Source code in packages/akms_learn/src/akms_learn/compiler.py

def compile_learning_source(
    request: LearningRequest | dict[str, Any],
    graph_path: str | Path | None = None,
    graph_slice: GraphSlice | dict[str, Any] | None = None,
    output_dir: str | Path | None = None,
    domain_pack_paths: list[str | Path] | None = None,
    source_pack_paths: list[str | Path] | None = None,
) -> CompileResult:
    """Run the 9-stage LSP compiler pipeline.

    Parameters
    ----------
    request:
        A :class:`LearningRequest` instance OR a raw dict that
        :func:`normalize_request` will canonicalise.
    graph_path:
        Filesystem path to a graph JSON. Mutually exclusive with
        ``graph_slice``.
    graph_slice:
        Either a :class:`GraphSlice` instance or a raw dict payload.
        Mutually exclusive with ``graph_path``.
    output_dir:
        If provided, the canonical packet JSON is written to
        ``<output_dir>/<request_hash>.json`` (Stage 9). Directory is created
        on demand.
    domain_pack_paths:
        Optional list of paths to ``domain_pack.yaml`` files (or directories
        containing one). Loaded into a :class:`DomainPackRegistry` whose
        descriptors are attached to the packet body.
    source_pack_paths:
        Optional list of paths to source-pack YAMLs whose descriptors are
        attached to the packet body.

    Returns
    -------
    CompileResult
        Wraps the validated packet, any export paths, warnings, and the
        stage execution log.

    Raises
    ------
    LearningCapabilityError
        If a required capability is unavailable, or a domain-pack /
        source-pack path is missing or invalid.
    PacketValidationError
        If the assembled packet violates a hard cross-field invariant.
    ValueError
        If neither / both of ``graph_path`` and ``graph_slice`` are given.
    """
    accumulator = WarningAccumulator()
    stage_log: list[str] = []

    # -----------------------------------------------------------------
    # Stage 1 — plugin and compatibility check
    # -----------------------------------------------------------------
    plugin = get_plugin()
    #   # ``LearningRequest`` carries no ``akms_schema`` field, so the check
    #       # defaults to "v2": existing requests pass cleanly while mis-typed
    #       # override values supplied via ad-hoc dicts are still rejected.
    requested_schema = str(_request_get(request, "akms_schema", "v2") or "v2")
    if requested_schema not in (
        plugin.supported_akms_schema_min,
        plugin.supported_akms_schema_max,
    ):
        raise LearningCapabilityError(
            f"Unsupported akms_schema {requested_schema!r}; plugin supports "
            f"{plugin.supported_akms_schema_min}..{plugin.supported_akms_schema_max}."
        )

    _check_required_capabilities(request, plugin, domain_pack_paths, source_pack_paths)
    stage_log.append("plugin_compat_check")

    # -----------------------------------------------------------------
    # Stage 2 — request normalization
    # -----------------------------------------------------------------
    normalized = normalize_request(request)
    req_hash = request_hash(normalized)
    stage_log.append("request_normalization")

    # -----------------------------------------------------------------
    # Stage 3 — graph source resolution
    # -----------------------------------------------------------------
    resolved_slice = _ensure_graph_slice(graph_path, graph_slice)
    graph_hash = compute_graph_hash(resolved_slice)
    stage_log.append("graph_source_resolution")

    # -----------------------------------------------------------------
    # Stage 4 — deterministic seed-tag handling
    # -----------------------------------------------------------------
    seed_tags = list(normalized.get("seed_tags") or [])
    filtered_slice = _filter_by_seed_tags(resolved_slice, seed_tags)
    stage_log.append("seed_tag_handling")

    # -----------------------------------------------------------------
    # Stage 5 — slice conversion (preserve provenance, no invention)
    # -----------------------------------------------------------------
    nodes_by_id: dict[str, dict[str, Any]] = {}
    for node in filtered_slice.nodes:
        # Copy so any downstream stage cannot mutate the input dict.
        nodes_by_id[_node_id(node)] = dict(node)
    stage_log.append("slice_conversion")

    # -----------------------------------------------------------------
    # Stage 6 — learning ordering
    # -----------------------------------------------------------------
    # Dispatch through the mode-specific ordering-strategy registry rather than
    # calling order_nodes() directly. The strategy for ``generation_option`` is
    # looked up in :data:`STRATEGY_KEYS`; options without a registered strategy
    # (e.g. ``deterministic_outline``, ``anthology``, ``bundle``) fall back to
    # ``"default"``, which delegates to :func:`order_nodes` unchanged. Modes with
    # a real reordering strategy (``derivation_first``, ``implementation_first``)
    # now have that ordering reflected in the LSP ``reading_order`` — previously
    # the registry was defined but never reached from the compiler.
    gen_opt = (normalized.get("generation_option") or "").strip().lower()
    strategy_key = gen_opt if gen_opt in STRATEGY_KEYS else "default"
    if strategy_key == "multi_granularity":
        # The multi_granularity ordering strategy is request-less and therefore
        # cannot apply the granularity filter, so the LSP reading_order would
        # otherwise be identical across overview/standard/deep_dive variants.
        # Route through the full mode here: it reads request.granularity (and
        # the tag/id/domain conventions) and returns the variant's node subset
        # — overview drops ``fine``-marked nodes, deep_dive keeps all. granularity
        # is intentionally excluded from request_hash (see NORMALIZED_FIELDS), so
        # the variant changes rendered output without changing request identity.
        from akms_learn.modes.multi_granularity import multi_granularity_mode

        mg_result, ordering_warnings = multi_granularity_mode(
            filtered_slice, _as_learning_request(request)
        )
        ordered_ids = list(mg_result.ordered_nodes)
    else:
        ordered_ids, ordering_warnings = get_strategy(strategy_key)(filtered_slice)
    accumulator.extend(ordering_warnings)

    if strategy_key == "adaptive_path":
        # adaptive_path is capability-gated (requires the ``llm`` extra) and its
        # ordering strategy is request-less, so the learner-profile prerequisite
        # skip would otherwise never reach the LSP. When the capability is
        # available, route the default-ordered nodes through the mode so skipped
        # prerequisites are dropped from ``reading_order`` (the slice is never
        # mutated — skips are preserved in the mode's provenance). When the extra
        # is absent the mode is unavailable, so the default ordering is kept
        # unchanged — matching how the capability catalog and bundle generator
        # already treat adaptive_path as unavailable in a clean checkout. The
        # ``active_nodes`` set is intersected against ``ordered_ids`` so reading
        # order is preserved (the result's own list is sorted, not ordered).
        from akms_learn.capability_gates import build_capability_gate

        if build_capability_gate().adaptive_path:
            from akms_learn.modes.adaptive_path import adaptive_path_mode

            ap_result, ap_warnings = adaptive_path_mode(
                filtered_slice, ordered_ids, _as_learning_request(request)
            )
            active = set(ap_result.active_nodes)
            ordered_ids = [nid for nid in ordered_ids if nid in active]
            accumulator.extend(ap_warnings)
    stage_log.append("learning_ordering")

    # -----------------------------------------------------------------
    # Stage 7 — section extraction
    # -----------------------------------------------------------------
    sections_by_node: dict[str, dict[str, Any]] = {}
    for nid in ordered_ids:
        node = nodes_by_id.get(nid, {})
        markdown = node.get("markdown") or node.get("body")
        if not markdown:
            sections_by_node[nid] = {}
            continue
        sections, section_warnings = extract_sections(
            markdown,
            str(node.get("source_path") or "unknown"),
            node_id=nid,
        )
        sections_by_node[nid] = sections
        accumulator.extend(section_warnings)
    stage_log.append("section_extraction")

    # -----------------------------------------------------------------
    # Stage 7b — LLM expansion
    # -----------------------------------------------------------------
    # Strictly between section extraction (stage 7) and packet assembly
    # (stage 8). Runs ONLY when the request opts in; otherwise the step is
    # skipped entirely so the packet stays byte-identical to the deterministic
    # baseline. The mode is a pure function that freezes a deep copy of
    # the deterministic packet BEFORE any provider call and never mutates the
    # inputs, so the surrounding pipeline is unaffected when expansion is off.
    llm_generated_sections: list[GeneratedSection] = []
    llm_provenance: dict[str, Any] = {}
    expansion_request = _build_llm_expansion_request(request)
    if expansion_request is not None:
        # Lazy module import mirrors the multi_granularity / adaptive_path
        # pattern below and avoids an import cycle through ``akms_learn.modes``.
        # Referencing the function via the module (rather than a bound local)
        # keeps it patchable by tests that instrument the pipeline step.
        from akms_learn.modes import llm_expanded as _llm_expanded

        llm_result, llm_warnings = _llm_expanded.llm_expanded_mode(
            filtered_slice,
            ordered_ids,
            _as_learning_request(request),
            expansion_request=expansion_request,
        )
        # Surface mode warnings (e.g. llm_citation_outside_packet,
        # llm_provider_unavailable) without touching the deterministic body.
        accumulator.extend(llm_warnings)
        llm_generated_sections = list(llm_result.generated_sections)
        # Record provenance.llm whenever the expansion step ran — even when the
        # provider returned nothing or every section was rejected — so consumers
        # can see the expansion was attempted. The mode's frozen
        # pre_expansion_packet guarantees the deterministic body is unchanged
        # regardless. ``model/notebook`` is read off the attached sections (all
        # share one provider/model) and is ``None`` when no section survived.
        #
        # This is the AUTHORITATIVE block written onto the PacketBody. It shares
        # ``build_llm_provenance`` (the single canonical shape) with the mode's
        # own ``result.packet`` block so the two cannot drift. ``rejected_count``
        # is read from the mode's block (the mode knows the raw vs. valid count);
        # it defaults to 0 on the fallback paths that don't populate it.
        citation_count = sum(len(s.source_node_ids) for s in llm_generated_sections)
        model_id = llm_generated_sections[0].model if llm_generated_sections else None
        mode_llm_prov = (llm_result.packet.get("provenance") or {}).get("llm") or {}
        # Prefer the provider the mode actually dispatched to: when the request
        # left the provider at its default, the mode auto-selects one by env
        # precedence (nlm → akms → stub), so the mode's own provenance block is
        # the authoritative name. Fall back to the request's provider on paths
        # that don't populate it (e.g. unavailable-provider fallback).
        effective_provider = mode_llm_prov.get("provider") or expansion_request.provider
        llm_provenance = {
            "llm": build_llm_provenance(
                provider=effective_provider,
                model=model_id,
                policy=llm_result.policy,
                section_count=len(llm_generated_sections),
                citation_count=citation_count,
                rejected_count=int(mode_llm_prov.get("rejected_count", 0)),
            )
        }

    # -----------------------------------------------------------------
    # Stage 8 — packet assembly and validation
    # -----------------------------------------------------------------
    # Resolve domain-pack / source-pack provenance up-front so we know which
    # capabilities ended up unsatisfied.
    domain_pack_provenance, _registry = _resolve_domain_pack_provenance(
        domain_pack_paths
    )
    source_pack_provenance = _resolve_source_pack_provenance(source_pack_paths)

    node_views = [
        _build_node_view(nodes_by_id[nid], sections_by_node.get(nid, {}))
        for nid in ordered_ids
        if nid in nodes_by_id
    ]
    # Restrict edges to those whose BOTH endpoints survive in the reading order.
    # For every non-dropping mode all nodes are retained, so this is a no-op; for
    # the node-dropping modes (``multi_granularity`` overview/standard,
    # ``adaptive_path``) it removes edges that would otherwise dangle against a
    # dropped node and fail packet validation. Mirrors the edge filter in
    # :func:`_filter_by_seed_tags`.
    reading_order_ids = set(ordered_ids)
    effective_edges = tuple(
        e
        for e in filtered_slice.edges
        if e.get("from") in reading_order_ids and e.get("to") in reading_order_ids
    )
    edge_views = [_build_edge_view(e) for e in effective_edges]
    pitfall_views = _build_pitfalls(nodes_by_id, ordered_ids, sections_by_node)
    code_links = _build_code_links(effective_edges, nodes_by_id, accumulator)

    # References: derive from each node's extracted References section. Empty for
    # nodes/fixtures that carry no References section (deterministic-baseline safe).
    reference_views = _build_references(ordered_ids, sections_by_node)

    # Assessments: when the assessment_first strategy is selected and its
    # capability extra (``notebook``) is present, generate items and surface them
    # in the packet body. Mirrors the adaptive_path capability-gated dispatch
    # above. The mode reads node["extracted"]; nodes without it yield no items.
    assessment_views: list[AssessmentView] = []
    if strategy_key == "assessment_first":
        from akms_learn.capability_gates import build_capability_gate

        if build_capability_gate().assessment_first:
            from akms_learn.modes.assessment_first import assessment_first_mode

            assessment_result, assessment_warnings = assessment_first_mode(
                filtered_slice, ordered_ids, _as_learning_request(request)
            )
            accumulator.extend(assessment_warnings)
            assessment_views = [
                AssessmentView(**item.model_dump())
                for item in assessment_result.assessment_items
            ]

    body = PacketBody(
        nodes=node_views,
        edges=edge_views,
        pitfalls=pitfall_views,
        code_links=code_links,
        assessments=assessment_views,
        references=reference_views,
        reading_order=list(ordered_ids),
        sections=[],
        domain_pack_provenance=domain_pack_provenance,
        source_pack_provenance=source_pack_provenance,
        generated_sections=llm_generated_sections,
        provenance=llm_provenance,
    )

    compiler_info = CompilerInfo(
        name="akms-learn",
        version="1.0",
        plugin_api=plugin.plugin_api,
    )

    source_info = SourceInfo(
        graph_hash=graph_hash,
        graph_path=str(graph_path) if graph_path is not None else "<in-memory>",
        graph_version=str(filtered_slice.metadata.get("graph_version") or "") or None,
        query_hash=req_hash,
    )

    # ``granularity`` is read off the raw request — it is
    # intentionally excluded from ``normalize_request`` / ``request_hash`` so
    # two requests that differ only in granularity hash identically (see
    # :data:`akms_learn.requests.NORMALIZED_FIELDS`). It is surfaced on the
    # LSP request block so downstream consumers (e.g. the bundle manifest)
    # can read the selected variant without re-running the mode.
    raw_granularity = _request_get(request, "granularity", None)
    # ``rich_html`` — like granularity, read off the raw request and
    # excluded from request_hash; it only toggles the html exporter's rendering.
    raw_rich_html = bool(_request_get(request, "rich_html", False))
    request_info = LearningRequestInfo(
        topic=str(normalized.get("topic") or ""),
        goal=normalized.get("goal") or None,
        audience=normalized.get("audience"),
        depth=normalized.get("depth"),
        generation_option=normalized.get("generation_option"),
        seed_tags=tuple(normalized.get("seed_tags") or ()),
        max_nodes=normalized.get("max_nodes"),
        max_depth=normalized.get("max_depth"),
        include_pitfalls=normalized.get("include_pitfalls"),
        include_code_links=normalized.get("include_code_links"),
        exporters=tuple(normalized.get("exporters") or ()),
        request_hash=req_hash,
        granularity=raw_granularity
        if raw_granularity in ("overview", "standard", "deep_dive")
        else None,
        rich_html=raw_rich_html,
    )

    # Deterministic packet_id derived from request_hash + graph_hash.
    # Byte-stability across identical invocations is required, so no per-call
    # entropy (no uuid, no timestamp suffix) may appear in this identifier.
    packet_id = f"lsp-{req_hash[:16]}-{graph_hash[:8]}"

    packet = LearningSourcePacket(
        packet_id=packet_id,
        created_at=datetime.now(UTC).isoformat(),
        compiler=compiler_info,
        source=source_info,
        request=request_info,
        body=body,
        warnings=accumulator.finalize(),
    )

    # Hard validation + soft-warning accumulation.
    validation_warnings = validate_packet(packet)
    accumulator.extend(validation_warnings)
    # Re-build with the (possibly extended) warning list so consumers see them.
    if validation_warnings:
        packet = packet.model_copy(update={"warnings": accumulator.finalize()})

    #   # Round-trip sanity check — provenance must never be destroyed. We narrow
    #       # the exception type to pydantic's validation error because any other
    #       # failure here is a programming bug, not a packet-shape issue, and should
    #       # surface with its native traceback.
    try:
        LearningSourcePacket.model_validate(packet.model_dump(by_alias=True))
    except pydantic.ValidationError as exc:  # pragma: no cover - defensive
        raise RuntimeError(f"LearningSourcePacket round-trip failed: {exc}") from exc

    stage_log.append("packet_assembly_and_validation")

    # -----------------------------------------------------------------
    # Stage 9 — export
    #
    # Ordering matters: we (1) probe declared exporters and accumulate any
    # ``exporter_unavailable`` warnings, (2) rebuild the final packet so its
    # ``warnings`` field includes the exporter warnings, and (3) only then
    # write the canonical JSON file to disk. Inverting this order causes the
    # persisted JSON to omit warnings that the in-memory packet later carries
    # (regression: ``test_compile_export_warnings_persisted``).
    # -----------------------------------------------------------------
    export_paths: list[Path] = []
    packet_path: Path | None = None

    # (1) Probe declared exporters. ``KNOWN_EXPORTERS`` is the single source
    # of truth for the names the compiler will attempt to dispatch (see
    # ``exporters/__init__.py``). Any name outside that set emits
    # ``exporter_unavailable`` without raising.
    requested_exporters: list[str] = list(normalized.get("exporters") or [])
    for exporter_name in requested_exporters:
        if exporter_name in KNOWN_EXPORTERS:
            module_name = f"akms_learn.exporters.{exporter_name}"
            try:
                mod = __import__(module_name, fromlist=["*"])
            except ImportError:  # pragma: no cover - stubs are importable
                accumulator.append(
                    LearningWarning(
                        severity="warning",
                        code="exporter_unavailable",
                        message=(
                            f"Exporter {exporter_name!r} module not available; skipped."
                        ),
                        source_ref=exporter_name,
                    )
                )
                continue
            # If the module exposes an ``export`` callable, invoke it now.
            if hasattr(mod, "export") and output_dir is not None:
                try:
                    produced = mod.export(packet, Path(output_dir))
                    export_paths.extend(produced)
                # One exporter's failure is isolated to a warning so it never
                # aborts the compile or the other exporters (broad by design).
                except Exception as exc:
                    accumulator.append(
                        LearningWarning(
                            severity="warning",
                            code="exporter_failed",
                            message=(
                                f"Exporter {exporter_name!r} raised an exception: {exc}"
                            ),
                            source_ref=exporter_name,
                        )
                    )
            elif not hasattr(mod, "export") and not hasattr(mod, "render"):
                accumulator.append(
                    LearningWarning(
                        severity="warning",
                        code="exporter_unavailable",
                        message=(
                            f"Exporter {exporter_name!r} is a Phase 1 stub; no artifact produced."
                        ),
                        source_ref=exporter_name,
                    )
                )
        else:
            accumulator.append(
                LearningWarning(
                    severity="warning",
                    code="exporter_unavailable",
                    message=(
                        f"Exporter {exporter_name!r} is not registered; no artifact produced."
                    ),
                    source_ref=exporter_name,
                )
            )

    # (2) Sync the final warning list into the packet BEFORE writing JSON.
    if accumulator.finalize() != list(packet.warnings):
        packet = packet.model_copy(update={"warnings": accumulator.finalize()})

    # (3) Write the canonical packet JSON to disk if an output directory was
    # supplied. The written file is now guaranteed to include every warning
    # the in-memory packet carries.
    if output_dir is not None:
        out_dir = Path(output_dir)
        out_dir.mkdir(parents=True, exist_ok=True)
        packet_path = out_dir / f"{req_hash}.json"
        payload = json.dumps(
            packet.model_dump(by_alias=True, mode="json"),
            indent=2,
            sort_keys=True,
            ensure_ascii=False,
        )
        packet_path.write_text(payload, encoding="utf-8")
        export_paths.append(packet_path)
    stage_log.append("export")

    # Capabilities that the request signalled interest in but that the
    # current pipeline could not satisfy. There is no soft-capability
    # surface yet; we leave it as an explicit empty list for forward compat.
    unavailable_capabilities: list[str] = []

    return CompileResult(
        packet=packet,
        packet_path=packet_path,
        export_paths=export_paths,
        warnings=accumulator.finalize(),
        unavailable_capabilities=unavailable_capabilities,
        stage_log=stage_log,
    )

CompileResult

akms_learn.CompileResult dataclass

CompileResult(
    packet: LearningSourcePacket,
    packet_path: Path | None = None,
    export_paths: list[Path] = list(),
    warnings: list[LearningWarning] = list(),
    unavailable_capabilities: list[str] = list(),
    stage_log: list[str] = list(),
)

Result bundle returned by :func:compile_learning_source.

Fields

packet: The fully-assembled, validated :class:LearningSourcePacket. packet_path: Filesystem path the canonical JSON packet was written to (Stage 9), or None when no output_dir was supplied. export_paths: Paths of every artifact produced by Stage 9 exporters. Always includes packet_path when present. warnings: Accumulated :class:LearningWarning instances from every stage. unavailable_capabilities: Capability strings that were requested but not satisfied by either the static plugin set or the domain-pack registry. stage_log: List of stage names appended after each stage completes. Always ends equal to STAGES on success.

STAGES

akms_learn.STAGES module-attribute

STAGES: tuple[str, ...] = (
    "plugin_compat_check",
    "request_normalization",
    "graph_source_resolution",
    "seed_tag_handling",
    "slice_conversion",
    "learning_ordering",
    "section_extraction",
    "packet_assembly_and_validation",
    "export",
)

---

Requests

The learning-request model, its canonical normalization, and the request hash.

LearningRequest

akms_learn.LearningRequest

Bases: BaseModel

Input model for a learning request (plan §10).

Only these 11 fields contribute to request_hash. UI-only state from Logic-Loom (preview_mode, ui_theme, session_id, ...) is rejected at normalize_request time and never reaches the hash.

audience, depth, generation_option are kept as free-form str rather than Literal to avoid coupling Phase 2 to a frozen enum set; normalize_request lowercases them so case variations collapse to a single canonical form.

normalize_request

akms_learn.normalize_request

normalize_request(
    raw: dict[str, Any] | LearningRequest,
) -> dict[str, Any]

Return the canonical dict representation of a learning request.

The output is a plain dict containing exactly the 11 normalized fields (see :data:NORMALIZED_FIELDS), with extras dropped and lists sorted. Calling :func:request_hash on the result yields a byte-stable SHA-256 digest (plan §10, L203).

Behavior:

• Accepts either a raw dict (e.g. from JSON / Logic-Loom UI) or a validated :class:LearningRequest instance.
• Drops keys not in the 11-field schema — including Logic-Loom UI state such as preview_mode, ui_theme, session_id.
• topic/goal are .strip() only (case preserved).
• audience/depth/generation_option are trimmed + lowercased.
• seed_tags/exporters elements are trimmed + lowercased + sorted.
• Missing optional fields receive documented defaults.

Source code in packages/akms_learn/src/akms_learn/requests.py

def normalize_request(raw: dict[str, Any] | LearningRequest) -> dict[str, Any]:
    """Return the canonical dict representation of a learning request.

    The output is a plain ``dict`` containing exactly the 11 normalized
    fields (see :data:`NORMALIZED_FIELDS`), with extras dropped and lists
    sorted. Calling :func:`request_hash` on the result yields a byte-stable
    SHA-256 digest (plan §10, L203).

    Behavior:

    * Accepts either a raw ``dict`` (e.g. from JSON / Logic-Loom UI) or a
      validated :class:`LearningRequest` instance.
    * Drops keys not in the 11-field schema — including Logic-Loom UI state
      such as ``preview_mode``, ``ui_theme``, ``session_id``.
    * ``topic``/``goal`` are ``.strip()`` only (case preserved).
    * ``audience``/``depth``/``generation_option`` are trimmed + lowercased.
    * ``seed_tags``/``exporters`` elements are trimmed + lowercased + sorted.
    * Missing optional fields receive documented defaults.
    """
    if isinstance(raw, LearningRequest):
        raw_dict: dict[str, Any] = raw.model_dump()
    elif isinstance(raw, dict):
        raw_dict = raw
    else:
        raise TypeError(
            f"normalize_request expects dict or LearningRequest, got {type(raw).__name__}"
        )

    # Pull only the 11 known fields. Anything else is dropped.
    canonical: dict[str, Any] = {
        "topic": _norm_str_trim(raw_dict.get("topic", "")),
        "goal": _norm_str_trim(raw_dict.get("goal", "")),
        "audience": _norm_enum_str(raw_dict.get("audience"), _DEFAULTS["audience"]),
        "depth": _norm_enum_str(raw_dict.get("depth"), _DEFAULTS["depth"]),
        "generation_option": _norm_enum_str(raw_dict.get("generation_option"), ""),
        "seed_tags": _norm_str_list(raw_dict.get("seed_tags")),
        "max_nodes": _coerce_optional_int(raw_dict.get("max_nodes")),
        "max_depth": _coerce_optional_int(raw_dict.get("max_depth")),
        "include_pitfalls": _coerce_bool(
            raw_dict.get("include_pitfalls"), _DEFAULTS["include_pitfalls"]
        ),
        "include_code_links": _coerce_bool(
            raw_dict.get("include_code_links"), _DEFAULTS["include_code_links"]
        ),
        "exporters": _norm_str_list(raw_dict.get("exporters")),
    }
    return canonical

to_canonical_dict

akms_learn.to_canonical_dict

to_canonical_dict(
    req: LearningRequest | dict[str, Any],
) -> dict[str, Any]

Convenience alias for :func:normalize_request.

Provided so call-sites that read more naturally as "give me the canonical dict for this request" don't have to import normalize_request directly. Returns the same canonical form (same 11 keys, same rules).

Source code in packages/akms_learn/src/akms_learn/requests.py

def to_canonical_dict(req: LearningRequest | dict[str, Any]) -> dict[str, Any]:
    """Convenience alias for :func:`normalize_request`.

    Provided so call-sites that read more naturally as "give me the canonical
    dict for this request" don't have to import ``normalize_request``
    directly. Returns the same canonical form (same 11 keys, same rules).
    """
    return normalize_request(req)

request_hash

akms_learn.request_hash

request_hash(normalized: dict[str, Any]) -> str

Return the 64-char hex SHA-256 digest of the canonical JSON form.

The input is expected to be the output of :func:normalize_request (or an equivalently shaped dict). The function is idempotent for pre-normalized input and byte-identical across Python sessions because:

• sort_keys=True makes the JSON form key-order-invariant.
• separators=(",", ":") removes incidental whitespace.
• ensure_ascii=False keeps non-ASCII characters in their native UTF-8 form so the digest does not depend on locale or escape choice.

See :func:normalize_request for the canonical key set and rules.

Source code in packages/akms_learn/src/akms_learn/requests.py

def request_hash(normalized: dict[str, Any]) -> str:
    """Return the 64-char hex SHA-256 digest of the canonical JSON form.

    The input is expected to be the output of :func:`normalize_request`
    (or an equivalently shaped dict). The function is idempotent for
    pre-normalized input and byte-identical across Python sessions because:

    * ``sort_keys=True`` makes the JSON form key-order-invariant.
    * ``separators=(",", ":")`` removes incidental whitespace.
    * ``ensure_ascii=False`` keeps non-ASCII characters in their native
      UTF-8 form so the digest does not depend on locale or escape choice.

    See :func:`normalize_request` for the canonical key set and rules.
    """
    payload = json.dumps(
        normalized,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()

---

Graph import

The in-memory graph slice and helpers, including the demo fixture.

GraphSlice

akms_learn.GraphSlice

Bases: BaseModel

In-memory representation of an AKMS graph slice.

Fields mirror the top-level structure of a graph.json produced by the AKMS compiler:

• nodes – list of node dicts (raw AKMS schema dicts, not typed).
• edges – list of edge dicts.
• metadata – free-form graph-level metadata (version, build_time, …).

The model is frozen so that equal slices hash identically and can be used as dict keys or set members. GraphSlice instances are immutable after construction; mutate by constructing a new instance.

load_graph

akms_learn.load_graph

load_graph(
    graph_path: str | Path | None = None,
    graph_slice: dict[str, Any] | None = None,
) -> GraphSlice

Dispatcher: load a GraphSlice from exactly one of the two sources.

Parameters

graph_path: If provided, read the graph from this filesystem path. graph_slice: If provided, validate this in-memory dict into a GraphSlice.

Returns

GraphSlice

Raises

ValueError If both graph_path and graph_slice are provided (mutual exclusion), or if neither is provided.

Source code in packages/akms_learn/src/akms_learn/graph_import.py

def load_graph(
    graph_path: str | Path | None = None,
    graph_slice: dict[str, Any] | None = None,
) -> GraphSlice:
    """Dispatcher: load a ``GraphSlice`` from exactly one of the two sources.

    Parameters
    ----------
    graph_path:
        If provided, read the graph from this filesystem path.
    graph_slice:
        If provided, validate this in-memory dict into a ``GraphSlice``.

    Returns
    -------
    GraphSlice

    Raises
    ------
    ValueError
        If *both* ``graph_path`` and ``graph_slice`` are provided (mutual
        exclusion), or if *neither* is provided.
    """
    if graph_path is not None and graph_slice is not None:
        raise ValueError(
            "load_graph: 'graph_path' and 'graph_slice' are mutually exclusive — "
            "supply exactly one, not both."
        )
    if graph_path is None and graph_slice is None:
        raise ValueError(
            "load_graph: at least one of 'graph_path' or 'graph_slice' must be provided."
        )
    if graph_path is not None:
        return load_from_path(graph_path)
    return load_from_slice(graph_slice)  # type: ignore[arg-type]

compute_graph_hash

akms_learn.compute_graph_hash

compute_graph_hash(graph_slice: GraphSlice) -> str

Return a deterministic SHA-256 hex digest for graph_slice.

The recipe is identical to request_hash in requests.py:

1. Build a canonical dict from the slice (nodes as list, edges as list, metadata as dict).
2. Serialise with json.dumps(..., sort_keys=True, separators=(',',':'), ensure_ascii=False).
3. Encode as UTF-8.
4. Return hashlib.sha256(...).hexdigest().

The sort_keys=True flag guarantees that dict key ordering inside individual node/edge dicts does NOT affect the digest.

Parameters

graph_slice: A GraphSlice instance (frozen Pydantic model).

Returns

str 64-character lowercase hex digest.

Source code in packages/akms_learn/src/akms_learn/graph_import.py

def compute_graph_hash(graph_slice: GraphSlice) -> str:
    """Return a deterministic SHA-256 hex digest for *graph_slice*.

    The recipe is identical to ``request_hash`` in ``requests.py``:

    1. Build a canonical ``dict`` from the slice (nodes as list, edges as
       list, metadata as dict).
    2. Serialise with ``json.dumps(..., sort_keys=True,
       separators=(',',':'), ensure_ascii=False)``.
    3. Encode as UTF-8.
    4. Return ``hashlib.sha256(...).hexdigest()``.

    The ``sort_keys=True`` flag guarantees that dict key ordering inside
    individual node/edge dicts does NOT affect the digest.

    Parameters
    ----------
    graph_slice:
        A ``GraphSlice`` instance (frozen Pydantic model).

    Returns
    -------
    str
        64-character lowercase hex digest.
    """
    canonical: dict[str, Any] = {
        "edges": list(graph_slice.edges),
        "metadata": graph_slice.metadata,
        "nodes": list(graph_slice.nodes),
    }
    payload = json.dumps(
        canonical,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()

fixture_graph

akms_learn.fixture_graph

fixture_graph() -> GraphSlice

Return a hand-built GraphSlice for use in tests and pipeline demos.

Graph topology (j² return-mapping theme, 6 nodes):

.. code-block:: text

prereq_linear_algebra  ──requires──►  core_j2_return_mapping
prereq_complex_numbers ──requires──►  core_j2_return_mapping
core_j2_return_mapping ──derives──►   deriv_state_space
deriv_state_space      ──implements──► impl_pole_placement
core_j2_return_mapping ──pitfall_of──► pitfall_sign_convention
impl_pole_placement    ──exercise_for──► exercise_verify_poles

Edge types used: requires, derives, implements, pitfall_of, exercise_for.

The fixture satisfies the Phase 3 ordering vocabulary (§12) and is large enough to be reused as the Phase 4 mode fixture.

Source code in packages/akms_learn/src/akms_learn/graph_import.py

def fixture_graph() -> GraphSlice:
    """Return a hand-built ``GraphSlice`` for use in tests and pipeline demos.

    Graph topology (j² return-mapping theme, 6 nodes):

    .. code-block:: text

        prereq_linear_algebra  ──requires──►  core_j2_return_mapping
        prereq_complex_numbers ──requires──►  core_j2_return_mapping
        core_j2_return_mapping ──derives──►   deriv_state_space
        deriv_state_space      ──implements──► impl_pole_placement
        core_j2_return_mapping ──pitfall_of──► pitfall_sign_convention
        impl_pole_placement    ──exercise_for──► exercise_verify_poles

    Edge types used: ``requires``, ``derives``, ``implements``,
    ``pitfall_of``, ``exercise_for``.

    The fixture satisfies the Phase 3 ordering vocabulary (§12) and is
    large enough to be reused as the Phase 4 mode fixture.
    """
    nodes: list[dict[str, Any]] = [
        {
            "node_id": "prereq_linear_algebra",
            "title": "Linear Algebra Foundations",
            "kind": "prerequisite",
            "domain": "mathematics",
            "tags": ["linear_algebra", "matrices", "eigenvalues"],
            "status": "established",
        },
        {
            "node_id": "prereq_complex_numbers",
            "title": "Complex Numbers and the Complex Plane",
            "kind": "prerequisite",
            "domain": "mathematics",
            "tags": ["complex_numbers", "poles", "s_plane"],
            "status": "established",
        },
        {
            "node_id": "core_j2_return_mapping",
            "title": "j² Return Mapping Algorithm",
            "kind": "core_concept",
            "domain": "computational_mechanics",
            "tags": ["j2_plasticity", "return_mapping", "radial_return"],
            "status": "established",
        },
        {
            "node_id": "deriv_state_space",
            "title": "State-Space Form of Elastoplastic Equations",
            "kind": "derivation",
            "domain": "computational_mechanics",
            "tags": ["state_space", "elastoplasticity", "incremental"],
            "status": "tentative",
        },
        {
            "node_id": "impl_pole_placement",
            "title": "Pole Placement Implementation",
            "kind": "implementation",
            "domain": "computational_mechanics",
            "tags": ["pole_placement", "implementation", "python"],
            "status": "tentative",
        },
        {
            "node_id": "pitfall_sign_convention",
            "title": "Sign Convention Pitfall in Stress Update",
            "kind": "pitfall",
            "domain": "computational_mechanics",
            "tags": ["pitfall", "sign_convention", "stress_update"],
            "status": "established",
        },
        {
            "node_id": "exercise_verify_poles",
            "title": "Exercise: Verify Pole Locations Analytically",
            "kind": "exercise",
            "domain": "computational_mechanics",
            "tags": ["exercise", "poles", "verification"],
            "status": "draft",
        },
    ]

    edges: list[dict[str, Any]] = [
        {
            "edge_id": "e_prereq_la_core",
            "from": "prereq_linear_algebra",
            "to": "core_j2_return_mapping",
            "type": "requires",
        },
        {
            "edge_id": "e_prereq_cn_core",
            "from": "prereq_complex_numbers",
            "to": "core_j2_return_mapping",
            "type": "requires",
        },
        {
            "edge_id": "e_core_deriv",
            "from": "core_j2_return_mapping",
            "to": "deriv_state_space",
            "type": "derives",
        },
        {
            "edge_id": "e_deriv_impl",
            "from": "deriv_state_space",
            "to": "impl_pole_placement",
            "type": "implements",
        },
        {
            "edge_id": "e_core_pitfall",
            "from": "core_j2_return_mapping",
            "to": "pitfall_sign_convention",
            "type": "pitfall_of",
        },
        {
            "edge_id": "e_impl_exercise",
            "from": "impl_pole_placement",
            "to": "exercise_verify_poles",
            "type": "exercise_for",
        },
    ]

    metadata: dict[str, Any] = {
        "description": "Minimal fixture graph for j² return-mapping learning path (Phase 3+4 tests)",
        "graph_version": "fixture-v1",
        "node_count": len(nodes),
        "edge_count": len(edges),
    }

    return GraphSlice(
        nodes=tuple(nodes),
        edges=tuple(edges),
        metadata=metadata,
    )

---

Models

The Learning Source Packet and its view models.

LearningSourcePacket

akms_learn.LearningSourcePacket

Bases: BaseModel

Root of the LSP (spec §3, L33-L69).

Holds top-level header fields, the request snapshot, the packet body, and a list of soft warnings accumulated during compilation.

PacketBody

akms_learn.PacketBody

Bases: BaseModel

The body of the LSP — all the rendered view collections (spec §3).

Optional forward-compat fields domain_pack_provenance and source_pack_provenance exist so the domain-pack layer can populate metadata without forcing a v2 schema bump (spec §12).

CompilerInfo

akms_learn.CompilerInfo

Bases: BaseModel

Identifies the compiler that produced the packet (spec §3).

SourceInfo

akms_learn.SourceInfo

Bases: BaseModel

Identifies the AKMS graph + query that produced this packet (spec §3).

graph_hash and graph_path are the stable provenance anchors that let the review bundle reproduce the packet from the same graph.

LearningRequestInfo

akms_learn.LearningRequestInfo

Bases: BaseModel

Normalized request snapshot + hash (spec §3, plan §10).

Only the normalized 11 request fields contribute to request_hash; UI state from Logic-Loom must NOT contribute (plan §10, L203). Hash stability is enforced upstream in request.normalize.

LearningNodeView

akms_learn.LearningNodeView

Bases: BaseModel

A node included in the packet (spec §4).

Required provenance: node_id, source_path, line_range. line_range is a (start, end) tuple of 1-indexed inclusive line numbers into source_path.

LearningEdgeView

akms_learn.LearningEdgeView

Bases: BaseModel

An edge included in the packet (spec §5).

Required provenance: edge_id, source_path, line_range. The from/to semantic endpoints from the spec YAML are mapped via aliases so the dumped form matches the spec key names.

PitfallView

akms_learn.PitfallView

Bases: BaseModel

A pitfall surfaced into the packet (spec §4, extracted.pitfalls + §6).

CodeLinkView

akms_learn.CodeLinkView

Bases: BaseModel

A code link (spec §7).

This view-type carries optional fields that the implementation-first mode populates when walking implements edges:

• source_node_id — the learning/spec node from which the edge starts.
• target — the code-mirror node id (preferred) or the source-file path of the implementation referenced by the edge.
• relation — the edge type that produced this view. Defaults to "implements" since CodeLinkViews are only emitted from implements edges.
• file_path — optional file path of the implementation.
• line_range — optional (start, end) line range.

The original fields (node_id, source_file, symbols, concept, mirror_node_id, explanation_mode) are preserved. CodeLinkView is a view-type and may be extended; the v2 graph schema (node/edge models) remains frozen.

ReferenceView

akms_learn.ReferenceView

Bases: BaseModel

A reference / further-reading entry (spec §9, references array).

AssessmentView

akms_learn.AssessmentView

Bases: BaseModel

Forward-compatibility stub for assessment items (spec §8).

Schema is NOT enforced; arbitrary keys are accepted. An empty assessments=[] is the typical value.

LearningWarning

akms_learn.LearningWarning

Bases: BaseModel

Soft validation issue emitted during LSP compilation.

Severity follows the spec's tri-state: info / warning / error. source_ref is an optional free-form reference to the offending input (e.g. node id, request field, descriptor path).

---

Validation

Packet validation and its hard-error type.

validate_packet

akms_learn.validate_packet

validate_packet(
    packet: LearningSourcePacket,
) -> list[LearningWarning]

Validate cross-field invariants on a compiled LSP.

Returns the list of accumulated soft warnings (possibly empty) on success. Raises :class:PacketValidationError if any hard invariant is violated.

Hard checks run first and short-circuit on failure (a packet that lacks a request_hash is not meaningful to soft-check).

Source code in packages/akms_learn/src/akms_learn/validation.py

def validate_packet(packet: LearningSourcePacket) -> list[LearningWarning]:
    """Validate cross-field invariants on a compiled LSP.

    Returns the list of accumulated soft warnings (possibly empty) on success.
    Raises :class:`PacketValidationError` if any hard invariant is violated.

    Hard checks run first and short-circuit on failure (a packet that lacks a
    ``request_hash`` is not meaningful to soft-check).
    """
    issues: list[str] = []

    # --- Hard check 1: request_hash is present and non-empty.
    request_hash = (packet.request.request_hash or "").strip()
    if not request_hash:
        issues.append("request.request_hash is missing or empty")

    # --- Hard check 2: source.graph_hash is present and non-empty.
    graph_hash = (packet.source.graph_hash or "").strip()
    if not graph_hash:
        issues.append("source.graph_hash is missing or empty")

    # --- Hard check 3: every edge endpoint resolves to a node in the packet.
    known_node_ids = {n.node_id for n in packet.body.nodes}
    for edge in packet.body.edges:
        if edge.from_node not in known_node_ids:
            issues.append(
                f"edge {edge.edge_id!r} 'from' references unknown "
                f"node_id {edge.from_node!r}"
            )
        if edge.to_node not in known_node_ids:
            issues.append(
                f"edge {edge.edge_id!r} 'to' references unknown "
                f"node_id {edge.to_node!r}"
            )

    if issues:
        raise PacketValidationError(issues)

    # --- Soft checks: empty bodies are valid but worth flagging.
    acc = WarningAccumulator()
    if not packet.body.nodes:
        acc.append(
            LearningWarning(
                severity="warning",
                code="empty_nodes",
                message="packet.body.nodes is empty",
                source_ref="body.nodes",
            )
        )
    if not packet.body.edges:
        acc.append(
            LearningWarning(
                severity="info",
                code="empty_edges",
                message="packet.body.edges is empty",
                source_ref="body.edges",
            )
        )

    return acc.finalize()

PacketValidationError

akms_learn.PacketValidationError

PacketValidationError(issues: list[str])

Bases: Exception

Raised when a :class:LearningSourcePacket violates a hard invariant.

Carries one or more textual issues describing the missing field(s) or dangling reference(s). The string form joins issues with "; " so error messages stay informative when surfaced through a single str(exc).

Source code in packages/akms_learn/src/akms_learn/validation.py

def __init__(self, issues: list[str]) -> None:
    self.issues: list[str] = list(issues)
    super().__init__("; ".join(self.issues))

---

Exporters

Exporter entry points. The compiler dispatches these in Stage 9 — callers list exporter names in request.exporters rather than calling these directly.

markdown_export

akms_learn.markdown_export

markdown_export(
    packet: "LearningSourcePacket", output_dir: Path
) -> list[Path]

Write a lesson.md file to output_dir and return its path.

This function is the Exporter Protocol entry point; the compiler's Stage 9 dispatches it automatically when "markdown" appears in request.exporters.

Parameters

packet: The fully-validated :class:~akms_learn.models.LearningSourcePacket produced by :func:~akms_learn.compiler.compile_learning_source. output_dir: Target directory. Created on demand if it does not exist.

Returns

list[Path] A one-element list containing the absolute path to lesson.md.

Source code in packages/akms_learn/src/akms_learn/exporters/markdown.py

def export(
    packet: "LearningSourcePacket",
    output_dir: Path,
    /,
) -> list[Path]:
    """Write a ``lesson.md`` file to *output_dir* and return its path.

    This function is the Exporter Protocol entry point; the compiler's Stage 9
    dispatches it automatically when ``"markdown"`` appears in
    ``request.exporters``.

    Parameters
    ----------
    packet:
        The fully-validated :class:`~akms_learn.models.LearningSourcePacket`
        produced by :func:`~akms_learn.compiler.compile_learning_source`.
    output_dir:
        Target directory.  Created on demand if it does not exist.

    Returns
    -------
    list[Path]
        A one-element list containing the absolute path to ``lesson.md``.
    """
    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    env = Environment(
        loader=FileSystemLoader(str(_TEMPLATES_DIR)),
        trim_blocks=True,
        lstrip_blocks=True,
        keep_trailing_newline=True,
        autoescape=False,
    )

    # Pedagogical dispatch: route the pedagogical mode keys through the
    # 12-slot template + context. Every other mode key keeps the original
    # template + context for byte-identical backward compatibility.
    mode_key = _mode_key(packet)
    if mode_key in PLAN2_MODE_KEYS:
        template = env.get_template(_TEMPLATE_EXPANDED)
        context = _build_plan2_context(packet)
    else:
        template = env.get_template(_TEMPLATE_NAME)
        context = _build_context(packet)
    rendered = template.render(**context)

    lesson_path = out_dir / "lesson.md"
    lesson_path.write_text(rendered, encoding="utf-8")
    return [lesson_path]

bundle_export

akms_learn.bundle_export

bundle_export(
    packet: "LearningSourcePacket", output_dir: Path
) -> list[Path]

Write the 7-artifact Mode 12 bundle to output_dir.

Steps:

1. lesson.md is rendered by delegating to :func:akms_learn.exporters.markdown.export (keeps a single source of truth for Markdown rendering).
2. All remaining artifact payloads are built in memory before any second write hits disk. Manifest is built last because it lists the other artifacts.
3. Artifacts are written in alphabetical order for visual consistency with manifest.json's artifacts field.
4. Empty exports/ and assets/ directories are created and pinned with .gitkeep sentinels so git tracks them.

Returns

list[Path] Sorted (by path string) list of every file written, including the two .gitkeep sentinels.

Source code in packages/akms_learn/src/akms_learn/exporters/bundle.py

def export(
    packet: "LearningSourcePacket",
    output_dir: Path,
    /,
) -> list[Path]:
    """Write the 7-artifact Mode 12 bundle to *output_dir*.

    Steps:

    1. ``lesson.md`` is rendered by delegating to
       :func:`akms_learn.exporters.markdown.export` (keeps a single source of
       truth for Markdown rendering).
    2. All remaining artifact payloads are built in memory before any second
       write hits disk. Manifest is built last because it lists the other
       artifacts.
    3. Artifacts are written in alphabetical order for visual consistency
       with ``manifest.json``'s ``artifacts`` field.
    4. Empty ``exports/`` and ``assets/`` directories are created and pinned
       with ``.gitkeep`` sentinels so git tracks them.

    Returns
    -------
    list[Path]
        Sorted (by path string) list of every file written, including the
        two ``.gitkeep`` sentinels.
    """
    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    # --- 1. lesson.md (delegate to markdown exporter) ----------------------
    # Writes <out_dir>/lesson.md and returns that path. The return value is
    # discarded rather than lost: the same path is rebuilt as `lesson_path`
    # below and is part of the returned list.
    _markdown_exporter.export(packet, out_dir)

    # --- 2. Build all payloads in memory ----------------------------------
    lsp_payload = packet.model_dump(by_alias=True, mode="json")
    lsp_yaml_text = _dump_yaml(lsp_payload)

    concept_map_text = _dump_json(
        {
            "nodes": [n.model_dump(mode="json") for n in packet.body.nodes],
            "edges": [
                e.model_dump(by_alias=True, mode="json") for e in packet.body.edges
            ],
        }
    )

    provenance_text = _dump_json(
        {
            "graph_hash": packet.source.graph_hash,
            "request_hash": packet.request.request_hash,
            "nodes": sorted(
                [
                    {
                        "node_id": n.node_id,
                        "source_path": n.source_path,
                        "line_range": list(n.line_range),
                    }
                    for n in packet.body.nodes
                ],
                key=lambda d: d["node_id"],
            ),
            "edges": sorted(
                [
                    {
                        "edge_id": e.edge_id,
                        "source_path": e.source_path,
                        "line_range": list(e.line_range),
                    }
                    for e in packet.body.edges
                ],
                key=lambda d: d["edge_id"],
            ),
        }
    )

    warnings_text = _dump_json([w.model_dump(mode="json") for w in packet.warnings])

    # Artifact filenames (relative to out_dir), sorted alphabetically.
    artifact_names = sorted(
        [
            "concept_map.json",
            "learning_source_packet.yaml",
            "lesson.md",
            "manifest.json",
            "provenance.json",
            "warnings.json",
        ]
    )

    # Manifest is built LAST because it indexes the other six artifacts.
    manifest_payload = _build_manifest(packet, artifact_names)
    manifest_text = _dump_json(manifest_payload)

    # --- 3. Write artifacts (alphabetical order; lesson.md already written) -
    concept_map_path = out_dir / "concept_map.json"
    lsp_yaml_path = out_dir / "learning_source_packet.yaml"
    lesson_path = out_dir / "lesson.md"  # written by markdown exporter
    manifest_path = out_dir / "manifest.json"
    provenance_path = out_dir / "provenance.json"
    warnings_path = out_dir / "warnings.json"

    _write_text(concept_map_path, concept_map_text)
    _write_text(lsp_yaml_path, lsp_yaml_text)
    # lesson.md is already on disk via the markdown exporter.
    _write_text(manifest_path, manifest_text)
    _write_text(provenance_path, provenance_text)
    _write_text(warnings_path, warnings_text)

    # --- 4. Empty exports/ and assets/ directories with .gitkeep ----------
    exports_dir = out_dir / "exports"
    assets_dir = out_dir / "assets"
    exports_dir.mkdir(parents=True, exist_ok=True)
    assets_dir.mkdir(parents=True, exist_ok=True)

    exports_gitkeep = exports_dir / ".gitkeep"
    assets_gitkeep = assets_dir / ".gitkeep"
    _write_text(exports_gitkeep, "")
    _write_text(assets_gitkeep, "")

    # --- 5. Return sorted absolute paths to every written artifact --------
    written: list[Path] = [
        concept_map_path,
        lsp_yaml_path,
        lesson_path,
        manifest_path,
        provenance_path,
        warnings_path,
        exports_gitkeep,
        assets_gitkeep,
    ]
    return sorted(set(written), key=str)

MANIFEST_VERSION

akms_learn.MANIFEST_VERSION module-attribute

MANIFEST_VERSION: str = 'v1'

---

Domain packs

Metadata-only descriptors, the registry, and the capability error type.

DomainPackDescriptor

akms_learn.DomainPackDescriptor

Bases: BaseModel

Top-level domain pack descriptor (spec §3).

A domain pack declares a curated learning domain. The core compiler MUST be able to load and reason about this descriptor without importing any companion package.

SourcePackDescriptor

akms_learn.SourcePackDescriptor

Bases: BaseModel

A source pack — a companion source repo / package (spec §4).

Source packs are pure metadata declarations. They describe a companion's repository roots, capability surface, and adapter id, but they never trigger any import of the companion code.

DomainPackRegistry

akms_learn.DomainPackRegistry

DomainPackRegistry()

In-memory map of :class:DomainPackDescriptor keyed by descriptor.id.

Deterministic: :meth:ordered_descriptors always returns descriptors sorted alphabetically by id, regardless of insertion order.

Source code in packages/akms_learn/src/akms_learn/domain_packs/registry.py

def __init__(self) -> None:
    self._descriptors: dict[str, DomainPackDescriptor] = {}

register

register(descriptor: DomainPackDescriptor) -> None

Add descriptor to the registry.

Raises:

Type	Description
ValueError	If a descriptor with the same id is already registered. Duplicate ids would make :meth:ordered_descriptors ambiguous.

Source code in packages/akms_learn/src/akms_learn/domain_packs/registry.py

def register(self, descriptor: DomainPackDescriptor) -> None:
    """Add ``descriptor`` to the registry.

    Raises:
        ValueError: If a descriptor with the same id is already
            registered. Duplicate ids would make
            :meth:`ordered_descriptors` ambiguous.
    """
    if descriptor.id in self._descriptors:
        raise ValueError(
            f"DomainPackDescriptor id {descriptor.id!r} is already registered."
        )
    self._descriptors[descriptor.id] = descriptor

get

get(descriptor_id: str) -> DomainPackDescriptor | None

Return the descriptor with id descriptor_id or None.

Source code in packages/akms_learn/src/akms_learn/domain_packs/registry.py

def get(self, descriptor_id: str) -> DomainPackDescriptor | None:
    """Return the descriptor with id ``descriptor_id`` or ``None``."""
    return self._descriptors.get(descriptor_id)

ordered_descriptors

ordered_descriptors() -> list[DomainPackDescriptor]

Return descriptors sorted alphabetically by id.

Deterministic across runs and Python sessions.

Source code in packages/akms_learn/src/akms_learn/domain_packs/registry.py

def ordered_descriptors(self) -> list[DomainPackDescriptor]:
    """Return descriptors sorted alphabetically by id.

    Deterministic across runs and Python sessions.
    """
    return [self._descriptors[k] for k in sorted(self._descriptors)]

load_from_yaml

load_from_yaml(yaml_path: PathLike) -> DomainPackDescriptor

Load a :class:DomainPackDescriptor from yaml_path.

The descriptor is parsed and returned but not automatically registered. Callers may decide whether to register it (this keeps the loader free of registration side effects).

Source code in packages/akms_learn/src/akms_learn/domain_packs/registry.py

def load_from_yaml(self, yaml_path: PathLike) -> DomainPackDescriptor:
    """Load a :class:`DomainPackDescriptor` from ``yaml_path``.

    The descriptor is parsed and returned but **not** automatically
    registered. Callers may decide whether to register it (this keeps
    the loader free of registration side effects).
    """
    return load_descriptor_from_yaml(yaml_path)

build_registry_from_paths

akms_learn.build_registry_from_paths

build_registry_from_paths(
    domain_pack_paths: Sequence[PathLike],
) -> DomainPackRegistry

Load multiple domain-pack YAMLs into a fresh :class:DomainPackRegistry.

Each path is parsed via :func:load_descriptor_from_yaml and registered in input order, but :meth:DomainPackRegistry.ordered_descriptors output remains deterministic (alphabetic by id).

Source code in packages/akms_learn/src/akms_learn/domain_packs/registry.py

def build_registry_from_paths(
    domain_pack_paths: Sequence[PathLike],
) -> DomainPackRegistry:
    """Load multiple domain-pack YAMLs into a fresh :class:`DomainPackRegistry`.

    Each path is parsed via :func:`load_descriptor_from_yaml` and registered
    in input order, but :meth:`DomainPackRegistry.ordered_descriptors`
    output remains deterministic (alphabetic by id).
    """
    registry = DomainPackRegistry()
    for path in domain_pack_paths:
        registry.register(load_descriptor_from_yaml(path))
    return registry

LearningCapabilityError

akms_learn.LearningCapabilityError

Bases: Exception

Raised when a request needs an unavailable required capability.

Per plan §21 rule 5 / spec §4 rule 2: missing source packs degrade to warnings unless the requested mode explicitly requires that pack — in which case the compiler MUST raise this error.