scripts.graph_query

Reference › scripts › graph_query

scripts/graph_query.py is a command-line client for querying the operational knowledge graph. A human or an agent runs it; nothing imports it. It uses @traces from orchestrata_ultimate.core.base.traces and the LLM client in orchestrata_ultimate.llm.client.

_load_env loads .env before any endpoint is resolved (RUL-019). Global options are --uri (default from the KG_ROUTING_URI environment variable), --user, --password and --database (default from KG_DATABASE, the operational orchkg; pass memgraph for the semantic and RAG database). _make_syncer builds a GraphSyncer from them. The subcommands, dispatched by main, are methods-without-tests (methods with no TESTED_BY relationship), dead-code (private methods with no incoming CALLS), deep-inheritance, untested-entrypoints, class-methods, callers, sync-tests and cypher, which runs an arbitrary Cypher query (SPEC-021 Wave-1b port). Most take --limit. For cypher, _looks_like_write_query runs a keyword pre-check for CREATE, MERGE, DELETE, SET, REMOVE and DROP after _strip_cypher_noise removes comments and string literals. main prints an exception as ERROR: and exits 1.

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/graph_query.py

#!/usr/bin/env python3
"""CLI for querying the operational knowledge graph.

Subcommands
-----------
methods-without-tests   List methods with no TESTED_BY relationship.
dead-code               List private methods with no incoming CALLS.
deep-inheritance        List classes with longest inheritance chains.
untested-entrypoints    List called methods that are not tested.
class-methods           List all methods for a given class.
callers                 List methods/functions that call a given method.
sync-tests              Sync test AST and create TESTED_BY relationships.
cypher                  Execute an arbitrary, project-scoped Cypher query.

Connection defaults to the ``KG_ROUTING_URI`` (falling back to ``KG_URI``)
environment variable; the orchgraph cluster negotiates no auth, so there is
no username/password to configure.  CLI flags override env vars.

Two databases share the cluster: the operational KG (``orchkg``, the
default, taken from ``KG_DATABASE``) and the shared semantic/RAG graph
(``memgraph``).  ``--database memgraph`` (a top-level option, so it goes
before the subcommand) points any subcommand -- in particular the read-only
``cypher`` -- at the semantic database.

Examples
--------
    python scripts/graph_query.py methods-without-tests --limit 20
    python scripts/graph_query.py dead-code --limit 10
    python scripts/graph_query.py deep-inheritance --limit 5
    python scripts/graph_query.py class-methods --class QaVolk
    python scripts/graph_query.py callers --method process
    python scripts/graph_query.py sync-tests --ast plans/ast-ultimate.json --project orchestrata-ultimate
    python scripts/graph_query.py cypher "MATCH (n) RETURN count(n) AS n"
    python scripts/graph_query.py cypher \\
        "MATCH (f:File) WHERE f.id STARTS WITH \\$project_prefix RETURN f.id LIMIT 5" \\
        --project orchestrata-ultimate

``cypher`` is the SPEC-021 Wave-1b port of the old OpenCode
``tools/{kg_query,kg_query_inner,graph_query,simple_kg_query}.py`` tools
(see ``plans/spec-021-port-map.md`` sec 3.5) -- consolidated into one
read-only-by-default, parameter-scoped capability rather than four
near-duplicate scripts each with their own stale monorepo ``sys.path``
hacking. It deliberately does NOT auto-rewrite the caller's Cypher to
inject project scoping the way the old ``kg_query_inner.py`` did: that
tool's ``apply_project_scoping()`` regex-rewrote arbitrary queries against
a 30+-relationship-type reachability pattern anchored on a bare
``directory.includes("orchestrata")`` substring check on cwd, which leaked
across sibling checkouts (``orchestrata-ultimate``, ``orchestrata-backup-*``,
...) -- see port-map bug #6. Here, ``--project`` only binds ``$project`` /
``$project_prefix`` (``"{project}:"``) parameters for the caller's own
``WHERE ... STARTS WITH $project_prefix`` clause, matching the canonical
``{project}:{natural_id}`` id scheme (``plans/kg-project-scoped-identity.md``)
-- never a ``CONTAINS``/substring rewrite.
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import os
import re
import sys
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

# Write-clause keywords the ``cypher`` subcommand refuses to run unless
# --allow-writes is passed. Matched as whole words (case-insensitive) after
# _strip_cypher_noise() removes string/backtick literals and comments, so a
# property VALUE containing e.g. "CREATE" doesn't false-positive.
_WRITE_CLAUSE_RE = re.compile(
    r"\b(CREATE|MERGE|DELETE|DETACH|SET|REMOVE|DROP)\b", re.IGNORECASE
)


@traces(spec="021", story="US-006", reqs=["REQ-021-207"])
def _strip_cypher_noise(cypher: str) -> str:
    """Best-effort strip of comments/string literals for the pre-flight check.

    NOT a Cypher parser -- deliberately so (a hand-rolled grammar would
    itself be a bug surface). This only exists to avoid the pre-check
    false-positiving on a write keyword that merely appears *inside* a
    string. The authoritative read-only guarantee is Neo4j's own
    server-side read-transaction access-mode rejection in ``_cmd_cypher``
    below (``BoltSyncRepository.execute_read``), which genuinely parses
    the query and rejects any real write operation -- this pre-check is
    only a faster, friendlier first error message.
    """
    text = re.sub(r"/\*.*?\*/", " ", cypher, flags=re.DOTALL)
    text = re.sub(r"//[^\n]*", " ", text)
    text = re.sub(r"'(?:[^'\\]|\\.)*'", " ", text)
    text = re.sub(r'"(?:[^"\\]|\\.)*"', " ", text)
    text = re.sub(r"`(?:[^`\\]|\\.)*`", " ", text)
    return text


@traces(spec="021", story="US-006", reqs=["REQ-021-207"])
def _looks_like_write_query(cypher: str) -> bool:
    """Keyword pre-check for CREATE/MERGE/DELETE/SET/REMOVE/DROP.

    See ``_strip_cypher_noise`` and ``_cmd_cypher`` docstrings for why this
    is intentionally NOT the enforcement boundary.
    """
    return bool(_WRITE_CLAUSE_RE.search(_strip_cypher_noise(cypher)))


_sync_path = (
    Path(__file__).resolve().parent.parent
    / "src"
    / "orchestrata_ultimate"
    / "knowledge_graph"
    / "sync.py"
)
_spec = importlib.util.spec_from_file_location(
    "orchestrata_ultimate.knowledge_graph.sync", _sync_path
)
if _spec is None or _spec.loader is None:
    raise ImportError(f"Cannot load sync module from {_sync_path}")
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
GraphSyncer = _mod.GraphSyncer


@traces(spec="021", story="US-006", reqs=["REQ-021-207"])
def _make_syncer(args: argparse.Namespace) -> GraphSyncer:
    """Instantiate GraphSyncer from CLI args (override env vars if supplied)."""
    uri = getattr(args, "uri", None) or None
    user = getattr(args, "user", None) or None
    password = getattr(args, "password", None) or None
    database = getattr(args, "database", None) or os.environ.get("KG_DATABASE") or None
    return GraphSyncer(uri=uri, username=user, password=password, database=database)


def _cmd_methods_without_tests(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        results = syncer.query_methods_without_tests(limit=args.limit)
        print(f"\n=== Methods without tests (limit={args.limit}) ===")
        for r in results:
            print(f"  {r.get('qualified_name', r.get('name', '?'))}")
        print(f"\nTotal: {len(results)}")
        return 0
    finally:
        syncer.close()


def _cmd_dead_code(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        results = syncer.query_dead_code(limit=args.limit)
        print(f"\n=== Dead code candidates (limit={args.limit}) ===")
        for r in results:
            print(f"  {r.get('qualified_name', r.get('name', '?'))}")
        print(f"\nTotal: {len(results)}")
        return 0
    finally:
        syncer.close()


def _cmd_deep_inheritance(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        results = syncer.query_deep_inheritance(limit=args.limit)
        print(f"\n=== Deep inheritance chains (limit={args.limit}) ===")
        for r in results:
            cls = r.get("class", {})
            print(
                f"  {cls.get('qualified_name', cls.get('name', '?'))} "
                f"(length={r.get('chain_length', 0)})"
            )
        print(f"\nTotal: {len(results)}")
        return 0
    finally:
        syncer.close()


def _cmd_untested_entrypoints(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        results = syncer.query_untested_entrypoints(limit=args.limit)
        print(f"\n=== Untested entrypoints (limit={args.limit}) ===")
        for r in results:
            print(f"  {r.get('qualified_name', r.get('name', '?'))}")
        print(f"\nTotal: {len(results)}")
        return 0
    finally:
        syncer.close()


def _cmd_class_methods(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        results = syncer.query_class_methods(class_name=args.class_name)
        print(f"\n=== Methods for class '{args.class_name}' ===")
        for r in results:
            async_flag = "async " if r.get("is_async") else ""
            print(
                f"  {async_flag}{r.get('name', '?')} -> "
                f"{r.get('return_type', 'None')} (line {r.get('line', '?')})"
            )
        print(f"\nTotal: {len(results)}")
        return 0
    finally:
        syncer.close()


def _cmd_callers(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        results = syncer.query_callers(method_name=args.method)
        print(f"\n=== Callers of method '{args.method}' ===")
        for r in results:
            print(f"  {r.get('qualified_name', r.get('name', '?'))}")
        print(f"\nTotal: {len(results)}")
        return 0
    finally:
        syncer.close()


def _cmd_sync_tests(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        counts = syncer.sync_tests(args.ast, args.project)
        print("\n=== Test sync complete ===")
        print(f"  Test nodes created:   {counts['test_nodes_created']}")
        print(f"  TESTED_BY edges:      {counts['tested_by_edges']}")
        return 0
    finally:
        syncer.close()


@traces(spec="021", story="US-006", reqs=["REQ-021-207", "REQ-021-208"])
def _cmd_cypher(args: argparse.Namespace) -> int:
    """Execute an arbitrary Cypher query -- SPEC-021 Wave-1b port.

    Read-only by default: runs through ``BoltSyncRepository.execute_read``,
    a genuine Neo4j read-only transaction -- the server parses the query and
    rejects any real write operation with a ``ClientError`` regardless of
    what the client-side ``_looks_like_write_query`` pre-check catches.
    ``--allow-writes`` switches to ``execute_write`` instead.

    ``--project`` does not rewrite the query text (see the module docstring
    and port-map bug #6) -- it only binds ``$project`` /
    ``$project_prefix`` parameters for the caller's own
    ``WHERE n.id STARTS WITH $project_prefix`` clause, per the canonical
    ``{project}:{natural_id}`` id scheme.
    """
    try:
        params: dict[str, Any] = json.loads(args.params) if args.params else {}
    except json.JSONDecodeError as exc:
        raise ValueError(f"--params must be valid JSON: {exc}") from exc
    if not isinstance(params, dict):
        raise ValueError("--params must decode to a JSON object")

    if args.project:
        params.setdefault("project", args.project)
        params.setdefault("project_prefix", f"{args.project}:")

    if _looks_like_write_query(args.cypher) and not args.allow_writes:
        raise ValueError(
            "query appears to contain a write clause "
            "(CREATE/MERGE/DELETE/SET/REMOVE/DROP) -- pass --allow-writes to "
            "run it. This is a best-effort keyword pre-check, not a full "
            "Cypher parser; the authoritative guard is Neo4j's own "
            "server-side read-transaction access-mode rejection."
        )

    syncer = _make_syncer(args)
    syncer.connect()
    try:
        repo = syncer._connected_repo
        if args.allow_writes:
            results = repo.execute_write(args.cypher, params)
        else:
            results = repo.execute_read(args.cypher, params)
        print(
            json.dumps(
                {"success": True, "data": results, "count": len(results)},
                indent=2,
                default=str,
            )
        )
        return 0
    finally:
        syncer.close()


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="graph_query",
        description="Query the operational knowledge graph.",
    )
    parser.add_argument(
        "--uri",
        default=None,
        help="Graph Bolt URI (default: KG_ROUTING_URI env var).",
    )
    parser.add_argument(
        "--user",
        default=None,
        help="Graph username (default: none -- orchgraph negotiates no auth).",
    )
    parser.add_argument(
        "--password",
        default=None,
        help="Graph password (default: none -- orchgraph negotiates no auth).",
    )

    parser.add_argument(
        "--database",
        default=None,
        help=(
            "Graph database on the shared cluster (default: KG_DATABASE env "
            "var, i.e. the operational 'orchkg'; pass 'memgraph' for the "
            "semantic/RAG database)."
        ),
    )

    sub = parser.add_subparsers(dest="command", required=True)

    # methods-without-tests
    mwt_p = sub.add_parser(
        "methods-without-tests",
        help="List methods with no TESTED_BY relationship.",
    )
    mwt_p.add_argument("--limit", type=int, default=20, help="Result limit.")

    # dead-code
    dc_p = sub.add_parser(
        "dead-code", help="List private methods with no incoming CALLS."
    )
    dc_p.add_argument("--limit", type=int, default=20, help="Result limit.")

    # deep-inheritance
    di_p = sub.add_parser(
        "deep-inheritance", help="List classes with longest inheritance chains."
    )
    di_p.add_argument("--limit", type=int, default=20, help="Result limit.")

    # untested-entrypoints
    ue_p = sub.add_parser(
        "untested-entrypoints",
        help="List called methods that are not tested.",
    )
    ue_p.add_argument("--limit", type=int, default=20, help="Result limit.")

    # class-methods
    cm_p = sub.add_parser("class-methods", help="List all methods for a class.")
    cm_p.add_argument("--class", dest="class_name", required=True, help="Class name.")

    # callers
    cl_p = sub.add_parser("callers", help="List callers of a method.")
    cl_p.add_argument("--method", required=True, help="Method name.")

    # sync-tests
    st_p = sub.add_parser(
        "sync-tests", help="Sync test AST and create TESTED_BY relationships."
    )
    st_p.add_argument("--ast", required=True, help="Path to test AST JSON file.")
    st_p.add_argument(
        "--project",
        required=True,
        help="Project name — scopes Test ids and Method/Function match targets.",
    )

    # cypher
    cy_p = sub.add_parser(
        "cypher",
        help=(
            "Execute an arbitrary Cypher query " "(read-only unless --allow-writes)."
        ),
    )
    cy_p.add_argument("cypher", help="Cypher query text.")
    cy_p.add_argument(
        "--params",
        default=None,
        help='JSON object of query parameters, e.g. \'{"name": "Foo"}\'.',
    )
    cy_p.add_argument(
        "--project",
        default=None,
        help=(
            "Project id to scope by. Binds $project and $project_prefix "
            "('{project}:') query parameters -- write your own WHERE clause "
            "against them (e.g. 'WHERE n.id STARTS WITH $project_prefix'). "
            "Does NOT auto-rewrite the query text (see module docstring)."
        ),
    )
    cy_p.add_argument(
        "--allow-writes",
        action="store_true",
        help=(
            "Permit CREATE/MERGE/DELETE/SET/REMOVE/DROP. Without this flag "
            "the query runs in a genuine Neo4j read-only transaction."
        ),
    )

    return parser


def _load_env() -> None:
    """Load ``.env`` before any endpoint is resolved (RUL-019).

    ``GraphSyncer`` reads ``KG_ROUTING_URI``/``KG_URI`` from ``os.environ`` with
    NO hardcoded fallback, so an entrypoint that never loads ``.env`` fails with
    "NEO4J_URI is not set" no matter how the fleet is configured. This script
    was missing the load entirely: it was reachable only by exporting the keys
    by hand, which silently broke every caller that did not -- including every
    opencode ``kg-*`` tool, all of which route through ``cypher`` here.

    Note ``_load_dotenv`` RETURNS a merged mapping and does NOT mutate
    ``os.environ``; calling it for its side effect alone is a no-op. The
    returned values are applied here without overwriting anything already
    exported, so an explicit ``KG_URI=... graph_query.py`` still wins.
    """
    try:
        from orchestrata_ultimate.llm.client import _load_dotenv
    except ImportError:
        return
    for key, value in (_load_dotenv() or {}).items():
        os.environ.setdefault(key, value)


def main(argv: list[str] | None = None) -> None:
    _load_env()
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "methods-without-tests": _cmd_methods_without_tests,
        "dead-code": _cmd_dead_code,
        "deep-inheritance": _cmd_deep_inheritance,
        "untested-entrypoints": _cmd_untested_entrypoints,
        "class-methods": _cmd_class_methods,
        "callers": _cmd_callers,
        "sync-tests": _cmd_sync_tests,
        "cypher": _cmd_cypher,
    }
    handler = dispatch.get(args.command)
    if handler is None:
        parser.print_help()
        sys.exit(1)

    try:
        exit_code = handler(args)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    sys.exit(exit_code)


if __name__ == "__main__":
    main()