scripts.build_runtime_callgraph

Reference › scripts › build_runtime_callgraph

Merges a captured runtime call graph into plans/ast-ultimate.json (SPEC-023), so the code graph holds calls observed at run time as well as those visible in the source. ast_live_sync imports it to keep runtime edges across syncs, and tests/scripts/test_build_runtime_callgraph.py and tests/scripts/test_code_intel.py import it. It writes JSON through atomic_json.

Invocation: main has two subcommands. build reads capture files from an edges directory and writes the merged AST and a report (_cmd_build, build). reaugment re-applies a previously saved edge set to a freshly produced AST (_cmd_reaugment, reaugment). With no subcommand it prints help and exits 1.

The flow: read_edges loads caller-callee pairs from the capture directory and counts them, _normalize and _is_scaffolding clean the names and drop test scaffolding, and collect_ast_nodes indexes the functions in the AST. partition_edges splits the edges into those whose ends both exist in the AST (resolved) and those that do not. compute_di_dispatch_edges adds edges for calls that go through dependency-injected interfaces, and augment_ast writes the edges into the AST. node_coverage computes the share of functions touched by a runtime edge, and write_report records it with the unresolved count.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/build_runtime_callgraph.py

#!/usr/bin/env python3
"""Merge runtime call-graph capture into ``plans/ast-ultimate.json`` (SPEC-023
traceability payoff).

The runtime call-graph capture plugin
(``orchestrata_ultimate.core.tracing.runtime_callgraph``, already built and
verified -- not touched here) writes one ``edges-<pid>.jsonl`` file per pytest
worker into ``$ORCH_RCG_DIR`` on session finish: one ``["caller_qname",
"callee_qname"]`` JSON array per line, captured via ``sys.monitoring`` CALL
events. Qnames match the AST's own convention exactly (``<module>.<co_qualname>``)
-- see that module's docstring for the full derivation.

This script is the offline merge step:

1. Union every worker's edges into one deduped set, dropping ``<locals>``/
   ``<generic ...>`` scaffolding endpoints (frame-internal code objects that
   have no AST node) and normalizing module-level callers (the plugin emits
   these as ``X.<module>`` -- the literal ``co_qualname`` CPython gives a
   module's top-level code object; this collapses to the AST's own module
   path ``X``).
2. Partitions the cleaned edges against the AST's node set (every module path
   + every class/function/method ``qualified_name``) into AST-resolved (both
   endpoints are real AST nodes) vs unresolved (an endpoint is dunder/dynamic/
   external-package glue with no AST node -- expected, not a bug).
3. Writes ``plans/runtime-callgraph.json`` -- a deterministic summary (sorted
   edges, no timestamps) of the AST-resolved runtime graph, headlined by
   ``di_dispatch_edges``: runtime edges the static ``calls`` extraction
   entirely misses (DI construction, registry/dispatch routing, CXP message
   handlers -- anything routed through ``self.x()``/a factory/a lookup table
   rather than a syntactically resolvable name).
4. Augments ``plans/ast-ultimate.json`` IN PLACE: every function/method whose
   qualified_name is an AST-resolved edge's caller gets a ``runtime_calls``
   field (sorted callee qnames); every other function/method gets ``[]``.
   This is what lets ``scripts/code_intel.py``'s Prolog KB (``codebase_db.pl``)
   build the MERGED static+runtime call graph and ``trace-path`` end-to-end
   through DI/dispatch edges pure static analysis cannot see.
5. Writes ``reports/runtime-callgraph.md`` -- the human-readable digest.

Usage::

    uv run python scripts/build_runtime_callgraph.py build \\
        --edges-dir /path/to/rcg_full --ast plans/ast-ultimate.json

Idempotent and safe to re-run: step 4 always recomputes every function's
``runtime_calls`` from scratch (no accumulation across runs) from whatever
edges-dir is passed this time.

A ``serialize_codebase_ast.py serialize`` re-run drops ``runtime_calls`` (it
doesn't know about the runtime capture -- only this script writes that
field), and rebuilding it from scratch normally means re-running the whole
traced suite to regenerate ``edges-*.jsonl``. The ``reaugment`` subcommand
avoids that: it re-applies ``runtime_calls`` to a freshly-regenerated AST
straight from the committed ``plans/runtime-callgraph.json`` summary::

    uv run python scripts/build_runtime_callgraph.py reaugment \\
        --ast plans/ast-ultimate.json --from plans/runtime-callgraph.json
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any, cast

from orchestrata_ultimate.core.atomic_json import dump_json_atomic

REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_AST = REPO_ROOT / "plans" / "ast-ultimate.json"
DEFAULT_OUT_JSON = REPO_ROOT / "plans" / "runtime-callgraph.json"
DEFAULT_REPORT = REPO_ROOT / "reports" / "runtime-callgraph.md"

# The literal co_qualname CPython assigns a module's top-level code object
# (see runtime_callgraph.py's _qname()) -- the plugin emits module-body
# callers as "<module path>.<module>"; this collapses that back to the
# AST's own module-path convention.
_MODULE_SUFFIX = ".<module>"
# Frame-internal code objects with no AST node: nested-function closures
# (<locals>) and PEP 695 generic-parameter scaffolding (<generic ...>).
_SCAFFOLD_MARKERS = ("<locals>", "<generic")


def _is_scaffolding(qname: str) -> bool:
    return any(marker in qname for marker in _SCAFFOLD_MARKERS)


def _normalize(qname: str) -> str:
    if qname.endswith(_MODULE_SUFFIX):
        return qname[: -len(_MODULE_SUFFIX)]
    return qname


def read_edges(edges_dir: Path) -> tuple[set[tuple[str, str]], int]:
    """Union every ``edges-*.jsonl`` in ``edges_dir`` into one deduped edge
    set, dropping scaffolding endpoints and normalizing module-level
    callers. Returns ``(edges, file_count)``."""
    edges: set[tuple[str, str]] = set()
    files = sorted(edges_dir.glob("edges-*.jsonl"))
    for path in files:
        with open(path, encoding="utf-8") as fh:
            for raw_line in fh:
                line = raw_line.strip()
                if not line:
                    continue
                pair = json.loads(line)
                caller, callee = pair[0], pair[1]
                if _is_scaffolding(caller) or _is_scaffolding(callee):
                    continue
                edges.add((_normalize(caller), _normalize(callee)))
    return edges, len(files)


def load_ast(ast_path: Path) -> dict[str, Any]:
    with open(ast_path, encoding="utf-8") as f:
        return cast(dict[str, Any], json.load(f))


def collect_ast_nodes(
    ast_data: dict[str, Any],
) -> tuple[set[str], dict[str, dict[str, Any]]]:
    """Return ``(valid_node_qnames, function_index)``.

    ``valid_node_qnames`` is every module path plus every class/function/
    method ``qualified_name`` -- the full set an edge endpoint can resolve
    against. ``function_index`` maps a function/method ``qualified_name`` to
    its live AST dict (a reference INTO ``ast_data``, not a copy) so
    :func:`augment_ast` can mutate it in place.
    """
    nodes: set[str] = set()
    func_index: dict[str, dict[str, Any]] = {}
    for module in ast_data.get("modules", []):
        nodes.add(module["path"])
        for fn in module.get("functions", []):
            nodes.add(fn["qualified_name"])
            func_index[fn["qualified_name"]] = fn
        for cls in module.get("classes", []):
            nodes.add(cls["qualified_name"])
            for method in cls.get("methods", []):
                nodes.add(method["qualified_name"])
                func_index[method["qualified_name"]] = method
    return nodes, func_index


def partition_edges(
    edges: set[tuple[str, str]], valid_nodes: set[str]
) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]:
    """Split edges into (ast_resolved, unresolved) -- both endpoints must be
    valid AST node qnames to count as resolved."""
    resolved: list[tuple[str, str]] = []
    unresolved: list[tuple[str, str]] = []
    for caller, callee in edges:
        if caller in valid_nodes and callee in valid_nodes:
            resolved.append((caller, callee))
        else:
            unresolved.append((caller, callee))
    return resolved, unresolved


def compute_di_dispatch_edges(
    resolved_edges: list[tuple[str, str]], func_index: dict[str, dict[str, Any]]
) -> list[tuple[str, str]]:
    """AST-resolved runtime edges whose callee does NOT appear verbatim in
    the caller's static ``calls`` (raw call-site text) list.

    The raw ``calls`` field holds unresolved call-site text (``"self.foo"``,
    ``"factory.build"``, a bare local name) -- essentially never a callee's
    full qualified name -- so an edge only drops OUT of this set if the
    fully-qualified runtime callee happens to appear verbatim in the
    caller's raw list. In practice that means nearly every runtime edge
    lands here: this IS the payoff -- the DI/dispatch/CXP edges static
    `calls` extraction structurally cannot resolve to a concrete target.
    A caller with no function-index entry (a class or module top-level
    caller, which carries no ``calls`` field at all) always counts as
    "missed" (empty raw list).
    """
    out: list[tuple[str, str]] = []
    for caller, callee in resolved_edges:
        fn = func_index.get(caller)
        raw_calls = fn.get("calls", []) if fn is not None else []
        if callee not in raw_calls:
            out.append((caller, callee))
    return out


def augment_ast(
    func_index: dict[str, dict[str, Any]], resolved_edges: list[tuple[str, str]]
) -> None:
    """Set ``runtime_calls`` (sorted callee qnames) on every function/method
    in ``func_index``, mutating the AST dicts in place. Functions with no
    runtime edges get ``[]``. Idempotent -- always recomputed from scratch,
    never accumulated across calls."""
    by_caller: dict[str, set[str]] = {}
    for caller, callee in resolved_edges:
        by_caller.setdefault(caller, set()).add(callee)
    for qname, fn in func_index.items():
        fn["runtime_calls"] = sorted(by_caller.get(qname, set()))


def node_coverage(
    func_index: dict[str, dict[str, Any]], resolved_edges: list[tuple[str, str]]
) -> float:
    """% of AST functions/methods that appear as caller OR callee in at
    least one AST-resolved runtime edge."""
    touched: set[str] = {
        n for caller, callee in resolved_edges for n in (caller, callee)
    }
    total = len(func_index)
    if total == 0:
        return 0.0
    covered = sum(1 for qname in func_index if qname in touched)
    return round(covered * 100.0 / total, 2)


def write_report(
    path: Path,
    result: dict[str, Any],
    coverage: float,
    unresolved_count: int,
) -> None:
    di_edges = result["di_dispatch_edges"]
    top = di_edges[:20]
    lines = [
        "# Runtime call-graph report",
        "",
        "Generated by `scripts/build_runtime_callgraph.py` from the SPEC-023 "
        "runtime call-graph capture (`core/tracing/runtime_callgraph.py`, a "
        "`sys.monitoring`-based pytest plugin) merged against "
        "`plans/ast-ultimate.json`.",
        "",
        "## Summary",
        "",
        f"- Total runtime edges (deduped union across worker files): "
        f"{result['edge_count']}",
        f"- Distinct runtime-graph nodes: {result['node_count']}",
        f"- AST-resolved edges (both endpoints are real AST nodes): "
        f"{result['ast_resolved_edge_count']}",
        f"- Unresolved edges (an endpoint has no AST node -- e.g. stdlib/"
        f"third-party glue, dunder machinery; expected, not a bug): "
        f"{unresolved_count}",
        f"- Node coverage: {coverage}% of AST functions/methods have >=1 "
        f"runtime call in or out",
        f"- Edge source files merged: {result['generated_from']['edge_files']}",
        "",
        "## Headline finding: the DI/dispatch/CXP edges static analysis cannot see",
        "",
        "Every edge below was observed at runtime (`sys.monitoring` CALL "
        "events during a real test run) but the callee does not appear "
        "verbatim in the caller's static `calls` list -- these are "
        "dependency-injection construction, registry/dispatch routing, and "
        "CXP message-handler edges: exactly the class of call raw-text call "
        "extraction cannot resolve to a concrete target (`self.x()` through "
        "a subclass, a factory-built handler, a lookup-table dispatch).",
        "",
        f"Top {len(top)} of {len(di_edges)} total (sorted):",
        "",
        "| caller | callee |",
        "|---|---|",
    ]
    for caller, callee in top:
        lines.append(f"| `{caller}` | `{callee}` |")
    lines.append("")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def build(
    edges_dir: Path, ast_path: Path, out_json: Path, out_report: Path
) -> dict[str, Any]:
    edges, file_count = read_edges(edges_dir)
    ast_data = load_ast(ast_path)
    valid_nodes, func_index = collect_ast_nodes(ast_data)
    resolved, unresolved = partition_edges(edges, valid_nodes)
    resolved_sorted = sorted(resolved)
    di_dispatch = sorted(compute_di_dispatch_edges(resolved, func_index))

    # Key order matches the brief's spec exactly -- json.dump with no
    # sort_keys preserves this dict's insertion order.
    result: dict[str, Any] = {
        "edge_count": len(edges),
        "node_count": len({q for pair in edges for q in pair}),
        "ast_resolved_edge_count": len(resolved),
        "edges": [list(pair) for pair in resolved_sorted],
        "di_dispatch_edges": [list(pair) for pair in di_dispatch],
        "generated_from": {"edge_files": file_count},
    }

    # Atomic. Both of these land on artifacts a concurrent gate run /
    # AST daemon also writes; a truncate-in-place write spliced two
    # generations of plans/ast-ultimate.json together on 2026-08-12 and
    # hard-crashed the whole battery in gates._runtime_call_count.
    dump_json_atomic(out_json, result, indent=2, trailing_newline=True)

    augment_ast(func_index, resolved)
    dump_json_atomic(ast_path, ast_data, indent=2, trailing_newline=True)

    coverage = node_coverage(func_index, resolved)
    write_report(out_report, result, coverage, len(unresolved))
    return result


def reaugment(from_path: Path, ast_path: Path) -> dict[str, Any]:
    """Re-apply ``runtime_calls`` to ``ast_path`` from a previously-written
    ``runtime-callgraph.json`` (``from_path``'s ``edges`` list), without
    needing the scratch edges-dir or a full suite re-run.

    Loads ``func_index`` + calls :func:`augment_ast` exactly as :func:`build`
    does -- this recovers what a fresh ``serialize_codebase_ast.py serialize``
    run drops (``runtime_calls`` is written by THIS script, not by the AST
    serializer). Idempotent for the same reason ``augment_ast`` is: it always
    recomputes from scratch.
    """
    with open(from_path, encoding="utf-8") as f:
        summary = json.load(f)
    edges: list[tuple[str, str]] = [(pair[0], pair[1]) for pair in summary["edges"]]

    ast_data = load_ast(ast_path)
    _, func_index = collect_ast_nodes(ast_data)
    augment_ast(func_index, edges)

    dump_json_atomic(ast_path, ast_data, indent=2, trailing_newline=True)

    functions_touched = sum(1 for fn in func_index.values() if fn["runtime_calls"])
    return {
        "reaugmented_from": str(from_path),
        "edges_applied": len(edges),
        "functions_touched": functions_touched,
    }


def _cmd_reaugment(args: argparse.Namespace) -> int:
    from_path = Path(args.from_)
    ast_path = Path(args.ast)
    if not from_path.exists():
        print(
            f"ERROR: runtime-callgraph summary not found: {from_path}", file=sys.stderr
        )
        return 1
    if not ast_path.exists():
        print(f"ERROR: AST snapshot not found: {ast_path}", file=sys.stderr)
        return 1
    result = reaugment(from_path, ast_path)
    print(json.dumps(result, indent=2))
    return 0


def _cmd_build(args: argparse.Namespace) -> int:
    edges_dir = Path(args.edges_dir)
    ast_path = Path(args.ast)
    if not edges_dir.is_dir():
        print(f"ERROR: edges dir not found: {edges_dir}", file=sys.stderr)
        return 1
    if not ast_path.exists():
        print(f"ERROR: AST snapshot not found: {ast_path}", file=sys.stderr)
        return 1
    result = build(edges_dir, ast_path, Path(args.out_json), Path(args.out_report))
    print(
        json.dumps(
            {
                "edge_count": result["edge_count"],
                "node_count": result["node_count"],
                "ast_resolved_edge_count": result["ast_resolved_edge_count"],
                "di_dispatch_edge_count": len(result["di_dispatch_edges"]),
                "edge_files": result["generated_from"]["edge_files"],
            },
            indent=2,
        )
    )
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="build_runtime_callgraph",
        description="Merge runtime call-graph edges into plans/ast-ultimate.json.",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    b = sub.add_parser(
        "build", help="Merge edges-*.jsonl, AST-augment in place, write report."
    )
    b.add_argument(
        "--edges-dir", required=True, help="Directory containing edges-*.jsonl."
    )
    b.add_argument(
        "--ast",
        default=str(DEFAULT_AST),
        help=f"AST JSON snapshot to augment in place (default: {DEFAULT_AST}).",
    )
    b.add_argument(
        "--out-json",
        default=str(DEFAULT_OUT_JSON),
        help=f"Merge summary output path (default: {DEFAULT_OUT_JSON}).",
    )
    b.add_argument(
        "--out-report",
        default=str(DEFAULT_REPORT),
        help=f"Markdown report output path (default: {DEFAULT_REPORT}).",
    )
    r = sub.add_parser(
        "reaugment",
        help=(
            "Re-apply runtime_calls to a freshly-regenerated AST from a "
            "committed runtime-callgraph.json -- no edges-dir or suite "
            "re-run needed."
        ),
    )
    r.add_argument(
        "--ast",
        default=str(DEFAULT_AST),
        help=f"AST JSON snapshot to augment in place (default: {DEFAULT_AST}).",
    )
    r.add_argument(
        "--from",
        dest="from_",
        default=str(DEFAULT_OUT_JSON),
        help=(
            "runtime-callgraph.json summary to re-apply edges from "
            f"(default: {DEFAULT_OUT_JSON})."
        ),
    )
    return parser


def main(argv: list[str] | None = None) -> None:
    parser = _build_parser()
    args = parser.parse_args(argv)
    if args.command == "build":
        sys.exit(_cmd_build(args))
    if args.command == "reaugment":
        sys.exit(_cmd_reaugment(args))
    parser.print_help()
    sys.exit(1)


if __name__ == "__main__":
    main()