orchestrata_ultimate.core.tracing.runtime_callgraph

Reference › orchestrata_ultimate › core › tracing › runtime_callgraph

A pytest plugin that records the runtime caller-to-callee call graph during a test run (SPEC-023 traceability). The static call graph in plans/ast-ultimate.json cannot see dependency-injection construction, registry dispatch or CXP message handling; edges captured here are merged into it so that forward reachability can follow whole paths from an entry point through a wired handler to a leaf.

It uses sys.monitoring CALL events, which give the caller's code object and the callee directly. Each call site returns DISABLE after its first observation, so it fires once and a full-suite capture costs a one-time hit per site. A polymorphic site therefore records only its first callee in each process, and the union across the batched executor's worker processes recovers the alternatives. _is_first_party, _module_of and _qname keep only edges whose both ends are inside the package and name them as the AST does (module path with dots, then the qualified name). The plugin is loaded with -p orchestrata_ultimate.core.tracing.runtime_callgraph and is active only when ORCH_RCG_DIR is set; otherwise it does nothing. pytest_configure starts monitoring, and pytest_sessionfinish writes edges-<pid>.jsonl into that directory, one pair per line. The script scripts/build_runtime_callgraph.py merges the files.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/tracing/runtime_callgraph.py

"""Runtime call-graph capture — a pytest plugin (SPEC-023 traceability).

Records the *runtime* caller->callee call graph across a test run via
``sys.monitoring`` (PEP 669) ``CALL`` events, so the static call graph in
``plans/ast-ultimate.json`` (each function's ``calls`` field) can be augmented
with the edges static analysis structurally cannot see: dependency-injection
construction, registry/dispatch routing, and CXP message handling. Merged into
the AST, these edges make forward reachability (``reaches/2``) trace full call
paths end-to-end (entry point -> DI-wired handler -> leaf).

Why ``CALL`` events: the event fires at each call site with the CALLER's code
object and the CALLEE, yielding a caller->callee edge directly — no frame-stack
walking. Returning ``DISABLE`` after first observation makes each call site fire
exactly once, so full-suite capture costs a one-time hit per site rather than a
per-call slowdown. (Caveat: a polymorphic site — one that dispatches to several
callees — records only its first callee *per process*; the union across the
batched executor's many worker processes recovers the alternates in practice.)

Qualified names match the AST convention exactly: ``<module>.<co_qualname>``
where ``<module>`` is the file path under ``src/orchestrata_ultimate/`` with
``/`` -> ``.`` and ``.py`` stripped (so ``session/cli/main.py``'s ``main`` ->
``session.cli.main.main``; a method ``Cls.m`` -> ``session.cli.main.Cls.m``).
Only edges whose BOTH endpoints are first-party (under the package) are kept —
that is exactly the AST's node set.

Loadable via ``-p orchestrata_ultimate.core.tracing.runtime_callgraph``. Active
only when ``ORCH_RCG_DIR`` is set (the output directory); otherwise a no-op, so
it is safe to leave loaded. On session finish each worker writes
``edges-<pid>.jsonl`` (one ``["caller","callee"]`` per line) into that dir; a
merge step (``scripts/build_runtime_callgraph.py``) unions them.
"""

from __future__ import annotations

import os
import sys
from types import CodeType
from typing import Any

# ── configuration ─────────────────────────────────────────────────────────────
_PKG_MARKER = "src/orchestrata_ultimate/"
_TOOL_NAME = "orch-runtime-callgraph"

# ── per-process capture state ─────────────────────────────────────────────────
# (id(caller_code), id(callee_code)) pairs seen — the cheap hot-path key.
_seen: set[tuple[int, int]] = set()
# id(code) -> code object; a strong ref, so id() cannot be reused by a GC'd
# code object (module-level function/method code lives for the process anyway).
_code_by_id: dict[int, CodeType] = {}
_tool_id: int | None = None


def _is_first_party(code: CodeType) -> bool:
    return _PKG_MARKER in code.co_filename


def _module_of(filename: str) -> str:
    rel = filename.split(_PKG_MARKER, 1)[1]
    mod = rel.removesuffix(".py").replace("/", ".")
    # a package's __init__.py IS the package module in the AST (path has no
    # trailing ``.__init__``) — normalize so runtime qnames join cleanly.
    return mod.removesuffix(".__init__")


def _qname(code: CodeType) -> str:
    return f"{_module_of(code.co_filename)}.{code.co_qualname}"


def _on_call(code: CodeType, offset: int, callee: object, arg0: object) -> Any:
    """``CALL`` handler: record a first-party caller->callee edge, then disable
    this call site so it never fires again in this process."""
    disable = sys.monitoring.DISABLE
    callee_code = getattr(callee, "__code__", None)
    if callee_code is None:  # C function / builtin — cannot be an AST node
        return disable
    if not (_is_first_party(code) and _is_first_party(callee_code)):
        return disable
    pair = (id(code), id(callee_code))
    if pair not in _seen:
        _seen.add(pair)
        _code_by_id[pair[0]] = code
        _code_by_id[pair[1]] = callee_code
    return disable


# ── pytest hooks ──────────────────────────────────────────────────────────────
def pytest_configure(config: object) -> None:
    """Start monitoring if ORCH_RCG_DIR is set; otherwise stay dormant."""
    global _tool_id
    if not os.environ.get("ORCH_RCG_DIR"):
        return
    mon = sys.monitoring
    # Claim the first free tool id (avoid clashing with coverage's DEBUGGER/
    # PROFILER ids if those tools are also active).
    for tid in range(6):
        if mon.get_tool(tid) is None:
            _tool_id = tid
            break
    if _tool_id is None:
        return
    mon.use_tool_id(_tool_id, _TOOL_NAME)
    mon.register_callback(_tool_id, mon.events.CALL, _on_call)
    mon.set_events(_tool_id, mon.events.CALL)


def pytest_sessionfinish(session: object, exitstatus: int) -> None:
    """Resolve id-pairs to qualified names and dump this worker's edges."""
    global _tool_id
    out_dir = os.environ.get("ORCH_RCG_DIR")
    if _tool_id is None or not out_dir:
        return
    mon = sys.monitoring
    try:
        mon.set_events(_tool_id, 0)
        mon.free_tool_id(_tool_id)
    finally:
        _tool_id = None
    lines: list[str] = []
    for caller_id, callee_id in _seen:
        caller = _code_by_id.get(caller_id)
        callee = _code_by_id.get(callee_id)
        if caller is None or callee is None:
            continue
        cq, ce = _qname(caller), _qname(callee)
        if cq != ce:  # drop trivial self-recursion noise at the qname level
            lines.append(f'["{cq}", "{ce}"]')
    if not lines:
        return
    path = os.path.join(out_dir, f"edges-{os.getpid()}.jsonl")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write("\n".join(sorted(set(lines))) + "\n")