Graph Compiler¶

akms.graph.build_graph ¶

build_graph.py — The Merge Compiler (§2.3 of system design).

Compiles the unified NetworkX DiGraph from all sources

Global nodes from ~/.claude/akms/nodes/ (or $AKMS_GLOBAL_VAULT)
Local nodes from /knowledge/local-nodes/
Code-mirror nodes from /knowledge/code-mirror/ (canonical schema-validated)
Local state overlay from local_state.yaml
Serialize to graph.json

All operations are deterministic. Output is stable across runs.

resolve_global_vault ¶

resolve_global_vault(
    explicit: str | Path | None = None,
    config: Any | None = None,
) -> Path

Resolve the global vault path with documented precedence (F-06).

Precedence (highest wins): 1. explicit — a caller-supplied path (CLI flag, test override). 2. AKMS_GLOBAL_VAULT environment variable. 3. config.global_vault when config is supplied. 4. Default ~/.claude/akms/nodes.

All returned paths have ~ expanded. This is the single source of truth for vault resolution across build_graph, graph_status, and any orchestrator handler that needs to pre-resolve a vault path.

Source code in packages/akms/src/akms/graph/build_graph.py

def resolve_global_vault(
    explicit: str | Path | None = None,
    config: Any | None = None,
) -> Path:
    """Resolve the global vault path with documented precedence (F-06).

    Precedence (highest wins):
      1. ``explicit`` — a caller-supplied path (CLI flag, test override).
      2. ``AKMS_GLOBAL_VAULT`` environment variable.
      3. ``config.global_vault`` when ``config`` is supplied.
      4. Default ``~/.claude/akms/nodes``.

    All returned paths have ``~`` expanded. This is the single source of
    truth for vault resolution across ``build_graph``, ``graph_status``,
    and any orchestrator handler that needs to pre-resolve a vault path.
    """
    if explicit is not None:
        return Path(explicit).expanduser()
    env = os.environ.get("AKMS_GLOBAL_VAULT")
    if env:
        return Path(env).expanduser()
    config_vault = getattr(config, "global_vault", None) if config is not None else None
    if config_vault:
        return Path(str(config_vault)).expanduser()
    return Path("~/.claude/akms/nodes").expanduser()

build_graph ¶

build_graph(
    repo_root: str | Path,
    global_vault: str | Path | None = None,
    output_path: str | Path | None = None,
    config: Any | None = None,
    strict: bool = False,
) -> nx.DiGraph

Compile the unified knowledge graph from all sources.

Parameters:

Name	Type	Description	Default
`repo_root`	`str \| Path`	Path to the repository root (contains knowledge/).	required
`global_vault`	`str \| Path \| None`	Override path to global vault. If None, resolved via precedence: AKMS_GLOBAL_VAULT env var > `config.global_vault` (when `config` is supplied) > default `~/.claude/akms/nodes`.	`None`
`output_path`	`str \| Path \| None`	Override path for graph.json output. If None, writes to /knowledge/graph/graph.json.	`None`
`config`	`Any \| None`	Optional PropagationConfig. When provided, `config.global_vault` becomes the third step in the precedence chain (honored if no explicit arg and no env var is set).	`None`

Returns:

Type	Description
`DiGraph`	The compiled NetworkX DiGraph.

Raises:

Type	Description
`SchemaVersionError`	If any source has wrong schema version.
`SchemaValidationError`	If any source has invalid schema.

Source code in packages/akms/src/akms/graph/build_graph.py

@traced("akms.build_graph")
def build_graph(
    repo_root: str | Path,
    global_vault: str | Path | None = None,
    output_path: str | Path | None = None,
    config: Any | None = None,
    strict: bool = False,
) -> nx.DiGraph:
    """Compile the unified knowledge graph from all sources.

    Args:
        repo_root: Path to the repository root (contains knowledge/).
        global_vault: Override path to global vault. If None, resolved via
                      precedence: AKMS_GLOBAL_VAULT env var > ``config.global_vault``
                      (when ``config`` is supplied) > default ``~/.claude/akms/nodes``.
        output_path: Override path for graph.json output. If None, writes to
                     <repo_root>/knowledge/graph/graph.json.
        config: Optional PropagationConfig. When provided, ``config.global_vault``
                becomes the third step in the precedence chain (honored if no
                explicit arg and no env var is set).

    Returns:
        The compiled NetworkX DiGraph.

    Raises:
        SchemaVersionError: If any source has wrong schema version.
        SchemaValidationError: If any source has invalid schema.
    """
    repo_root = Path(repo_root)
    knowledge_dir = repo_root / "knowledge"
    graph_dir = knowledge_dir / "graph"

    vault_path = resolve_global_vault(explicit=global_vault, config=config)

    if output_path is None:
        output_path = graph_dir / "graph.json"
    else:
        output_path = Path(output_path)

    G = nx.DiGraph()
    warnings: list[str] = []
    skipped_files: list[dict] = []  # surfaced by graph_status
    repo_id = repo_root.name

    # ── Step 1: Load Global Nodes ────────────────────────────────────
    global_files = _collect_md_files(vault_path)
    for md_path in global_files:
        data = _load_node_frontmatter(
            md_path, strict=strict, skipped_accumulator=skipped_files
        )
        if data is None:
            continue

        try:
            node = parse_node_frontmatter_from_dict(
                data, is_local=False, path=str(md_path)
            )
        except (SchemaVersionError, SchemaValidationError):
            raise  # Fatal — halt on schema errors per FR-G08

        node_id = node.id
        attrs = node.model_dump()

        # Extract edges before adding node
        edges = attrs.pop("edges", [])

        # Add origin marker
        attrs["node_origin"] = "global"
        # confidence_default = the global seed value (for inspectability)
        attrs["confidence_default"] = attrs["confidence"]
        # Default experiential state (may be overridden by overlay in step 4)
        attrs["activations"] = 0
        attrs["last_activated"] = None

        G.add_node(node_id, **attrs)

        # Add structural edges
        for edge in edges:
            G.add_edge(
                node_id,
                edge["to"],
                type=edge["type"],
                weight=edge["weight"],
                note=edge.get("note", ""),
                edge_origin="global",
            )

    logger.info("Loaded %d global nodes from %s", len(global_files), vault_path)

    # ── Step 2: Load Local Nodes ─────────────────────────────────────
    local_nodes_dir = knowledge_dir / "local-nodes"
    local_files = _collect_md_files(local_nodes_dir)
    local_count = 0

    for md_path in local_files:
        data = _load_node_frontmatter(
            md_path, strict=strict, skipped_accumulator=skipped_files
        )
        if data is None:
            continue

        try:
            node = parse_node_frontmatter_from_dict(
                data, is_local=True, path=str(md_path)
            )
        except (SchemaVersionError, SchemaValidationError):
            raise

        node_id = node.id

        # Skip on id collision with global node
        if node_id in G and G.nodes[node_id].get("node_origin") == "global":
            msg = (
                f"Local node '{node_id}' collides with global node — "
                f"skipping local (file: {md_path})"
            )
            warnings.append(msg)
            logger.warning(msg)
            continue

        attrs = node.model_dump()
        edges = attrs.pop("edges", [])
        attrs["node_origin"] = "local"
        attrs["confidence_default"] = attrs["confidence"]
        attrs["activations"] = 0
        attrs["last_activated"] = None

        G.add_node(node_id, **attrs)

        for edge in edges:
            G.add_edge(
                node_id,
                edge["to"],
                type=edge["type"],
                weight=edge["weight"],
                note=edge.get("note", ""),
                edge_origin="local",
            )

        local_count += 1

    logger.info("Loaded %d local nodes", local_count)

    # ── Step 3: Load Code-Mirror Nodes ───────────────────────────────
    mirror_dir = knowledge_dir / "code-mirror"
    mirror_files = _collect_md_files(mirror_dir)
    mirror_count = 0

    for md_path in mirror_files:
        data = _load_node_frontmatter(
            md_path, strict=strict, skipped_accumulator=skipped_files
        )
        if data is None:
            continue

        try:
            node = parse_node_frontmatter_from_dict(
                data,
                is_code_mirror=True,
                path=str(md_path),
            )
        except (SchemaVersionError, SchemaValidationError):
            raise

        attrs = node.model_dump()
        node_id = attrs["id"]
        attrs["node_origin"] = "code-mirror"
        attrs["confidence_default"] = attrs.get("confidence", 1.0)
        attrs["activations"] = 0
        attrs["last_activated"] = None

        G.add_node(node_id, **attrs)
        mirror_count += 1

    logger.info("Loaded %d code-mirror nodes", mirror_count)

    # ── Step 4: Apply Local Overlay ──────────────────────────────────
    overlay_path = graph_dir / "local_state.yaml"
    if overlay_path.exists():
        overlay = parse_local_state(overlay_path)
        repo_id = overlay.repo_id or repo_root.name

        # 4a. Override per-node state
        for node_id, state in overlay.nodes.items():
            if node_id not in G:
                msg = (
                    f"Orphaned overlay entry: node '{node_id}' in "
                    f"local_state.yaml but not in graph"
                )
                warnings.append(msg)
                logger.warning(msg)
                continue

            state_dict = state.model_dump(exclude_none=True)
            # Convert date to string for JSON serialization
            if "last_activated" in state_dict and state_dict["last_activated"]:
                state_dict["last_activated"] = str(state_dict["last_activated"])

            G.nodes[node_id].update(state_dict)

        # 4b. Add local edges
        for edge in overlay.local_edges:
            G.add_edge(
                edge.from_node,
                edge.to,
                type=edge.type,
                weight=edge.weight,
                note=edge.note,
                edge_origin="local",
            )

        # 4c. Create session nodes
        for session_id, session in overlay.session_nodes.items():
            attrs = {
                "id": session_id,
                "title": session.title,
                "domain": "session",
                "tags": session.tags,
                "status": "established",
                "confidence": 1.0,
                "confidence_default": 1.0,
                "source": "generated",
                "auto_update": True,
                "node_origin": "local",
                "outcome": session.outcome,
                "content_ref": session.content_ref,
                "phase": session.phase,
                "akms_schema": AKMS_SCHEMA_VERSION,
                "activations": 0,
                "last_activated": None,
            }
            G.add_node(session_id, **attrs)

        logger.info(
            "Applied overlay: %d node overrides, %d local edges, %d session nodes",
            len(overlay.nodes),
            len(overlay.local_edges),
            len(overlay.session_nodes),
        )

    # ── Step 5: Serialize ────────────────────────────────────────────
    graph_data = _serialize_graph(G, vault_path, repo_id)

    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        json.dump(graph_data, f, sort_keys=True, indent=2, default=str)

    logger.info(
        "Compiled graph: %d nodes, %d edges → %s",
        G.number_of_nodes(),
        G.number_of_edges(),
        output_path,
    )

    if warnings:
        logger.info("Build warnings (%d):", len(warnings))
        for w in warnings:
            logger.info("  - %s", w)

    # Preserve skipped-file details through graph.graph attrs so
    # graph_status can report non-fatal parse failures instead of
    # silently omitting sources.
    G.graph["skipped_files"] = list(skipped_files)

    return G

load_graph ¶

load_graph(path: str | Path) -> nx.DiGraph

Load a compiled graph.json back into a NetworkX DiGraph.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to graph.json.	required

Returns:

Type	Description
`DiGraph`	The reconstructed DiGraph.

Source code in packages/akms/src/akms/graph/build_graph.py

def load_graph(path: str | Path) -> nx.DiGraph:
    """Load a compiled graph.json back into a NetworkX DiGraph.

    Args:
        path: Path to graph.json.

    Returns:
        The reconstructed DiGraph.
    """
    path = Path(path)
    with open(path) as f:
        data = json.load(f)

    G = nx.DiGraph()

    for node_data in data.get("nodes", []):
        node_id = node_data.pop("id")
        G.add_node(node_id, id=node_id, **node_data)

    for link_data in data.get("links", []):
        source = link_data.pop("source")
        target = link_data.pop("target")
        G.add_edge(source, target, **link_data)

    return G