orchestrata_ultimate.session.cli.kg

Reference › orchestrata_ultimate › session › cli › kg

The knowledge-graph commands of the session CLI. sync runs the KnowledgeGraphSyncer; options select a session number (--session), individual stages (--code, --specs, --traces, --folders, --docs) and graph statistics after the sync (--stats). It exits with an error if the graph connection fails and always closes the connection. The query group reads through KnowledgeGraphContext: file and spec give the context of a file or a spec, undocumented lists code entities without docstrings, untraced lists those without requirement traceability, unchecked lists unchecked requirement items, and search does a full-text search with a result limit.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/cli/kg.py

# mypy: ignore-errors
"""Knowledge graph sync command and query group for the session CLI."""

from __future__ import annotations

import json
import sys
from typing import TYPE_CHECKING

import click

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.session.kg_context import KnowledgeGraphContext
    from orchestrata_ultimate.session.kg_sync import KnowledgeGraphSyncer


@click.command()
@click.option("--session", type=int, default=None, help="Session seq to sync.")
@click.option("--code", is_flag=True, help="Sync code entities only.")
@click.option("--specs", is_flag=True, help="Sync specs only.")
@click.option("--traces", "traces_flag", is_flag=True, help="Sync traces only.")
@click.option("--folders", is_flag=True, help="Sync folders only.")
@click.option("--docs", is_flag=True, help="Sync doc coverage only.")
@click.option("--stats", is_flag=True, help="Show graph stats after sync.")
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-006"])
def sync(
    ctx: click.Context,
    session: int | None,
    code: bool,
    specs: bool,
    traces_flag: bool,
    folders: bool,
    docs: bool,
    stats: bool,
) -> None:
    """Sync project state to knowledge graph."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.kg_sync import KnowledgeGraphSyncer

    syncer = KnowledgeGraphSyncer(str(root))
    if not syncer.connect():
        click.echo("Failed to connect to the graph", err=True)
        sys.exit(1)

    try:
        results = _run_sync_stages(
            syncer, session, code, specs, traces_flag, folders, docs
        )
        _echo_sync_results(results)
        if stats:
            _echo_graph_stats(syncer)
    finally:
        syncer.close()


def _run_sync_stages(
    syncer: KnowledgeGraphSyncer,
    session: int | None,
    code: bool,
    specs: bool,
    traces_flag: bool,
    folders: bool,
    docs: bool,
) -> dict:
    """Run the requested sync stages (or all stages) and return their stats."""
    specific = code or specs or traces_flag or folders or docs
    if not specific:
        return syncer.sync_all(session_seq=session)

    results = {}
    if folders:
        results["folders"] = syncer.sync_folders()
    if code:
        results["code"] = syncer.sync_code()
    if specs:
        results["specs"] = syncer.sync_specs()
    if traces_flag:
        results["traces"] = syncer.sync_traces()
    if docs:
        results["docs"] = syncer.sync_docs()
    return results


def _echo_sync_results(results: dict) -> None:
    """Print one summary line per sync stage plus any errors."""
    for name, st in results.items():
        status = "OK" if st.ok else "ERRORS"
        click.echo(
            f"  {name:<15} nodes={st.nodes_created:>4}  "
            f"rels={st.relationships_created:>4}  "
            f"updated={st.nodes_updated:>4}  "
            f"{st.duration_ms:>7.1f}ms  [{status}]"
        )
        for err in st.errors:
            click.echo(f"    ERROR: {err}", err=True)


def _echo_graph_stats(syncer: KnowledgeGraphSyncer) -> None:
    """Print node/relationship counts for the current graph."""
    graph_stats = syncer.get_stats()
    click.echo("\nGraph Statistics:")
    click.echo("  Nodes:")
    for label, count in sorted(graph_stats["nodes"].items()):
        click.echo(f"    {label:<20} {count:>6}")
    click.echo("  Relationships:")
    for rtype, count in sorted(graph_stats["relationships"].items()):
        click.echo(f"    {rtype:<20} {count:>6}")


@click.group()
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-006"])
def query(ctx: click.Context) -> None:
    """Query the knowledge graph."""


def _get_kg_context(ctx: click.Context) -> KnowledgeGraphContext:
    """Create and connect a KnowledgeGraphContext."""
    from orchestrata_ultimate.session.kg_context import KnowledgeGraphContext

    kgc = KnowledgeGraphContext()
    if not kgc.connect():
        click.echo("Failed to connect to the graph", err=True)
        sys.exit(1)
    return kgc


@click.command(name="file")
@click.argument("path")
@click.pass_context
def query_file(ctx: click.Context, path: str) -> None:
    """Query context for a file."""
    kgc = _get_kg_context(ctx)
    try:
        result = kgc.get_file_context(path)
        click.echo(json.dumps(result, indent=2, default=str))
    finally:
        kgc.close()


@click.command(name="spec")
@click.argument("spec_id")
@click.pass_context
def query_spec(ctx: click.Context, spec_id: str) -> None:
    """Query context for a spec."""
    kgc = _get_kg_context(ctx)
    try:
        result = kgc.get_spec_context(spec_id)
        click.echo(json.dumps(result, indent=2, default=str))
    finally:
        kgc.close()


@click.command()
@click.pass_context
def undocumented(ctx: click.Context) -> None:
    """List code entities missing docstrings."""
    kgc = _get_kg_context(ctx)
    try:
        results = kgc.get_undocumented()
        if not results:
            click.echo("All code entities have docstrings.")
            return
        click.echo(f"{'File':<50} {'Type':<10} {'Name':<30}")
        click.echo("-" * 92)
        for r in results:
            click.echo(f"{r['file']:<50} {r['type']:<10} {r['name']:<30}")
        click.echo(f"\nTotal: {len(results)} undocumented entities")
    finally:
        kgc.close()


@click.command()
@click.pass_context
def untraced(ctx: click.Context) -> None:
    """List code entities with no requirement traceability."""
    kgc = _get_kg_context(ctx)
    try:
        results = kgc.get_untraced()
        if not results:
            click.echo("All code entities are traced to requirements.")
            return
        click.echo(f"{'File':<50} {'Type':<10} {'Name':<30}")
        click.echo("-" * 92)
        for r in results:
            click.echo(f"{r['file']:<50} {r['type']:<10} {r['name']:<30}")
        click.echo(f"\nTotal: {len(results)} untraced entities")
    finally:
        kgc.close()


@click.command()
@click.pass_context
def unchecked(ctx: click.Context) -> None:
    """List unchecked requirement items."""
    kgc = _get_kg_context(ctx)
    try:
        results = kgc.get_unchecked_requirements()
        if not results:
            click.echo("All requirement items are checked.")
            return
        click.echo(f"{'Spec':<10} {'Item ID':<25} {'Text':<50}")
        click.echo("-" * 87)
        for r in results:
            text = r["text"][:47] + "..." if len(r["text"]) > 50 else r["text"]
            click.echo(f"{r['spec']:<10} {r['item_id']:<25} {text:<50}")
        click.echo(f"\nTotal: {len(results)} unchecked items")
    finally:
        kgc.close()


@click.command(name="search")
@click.argument("term")
@click.option("--limit", default=20, help="Max results.")
@click.pass_context
def query_search(ctx: click.Context, term: str, limit: int) -> None:
    """Full-text search across the knowledge graph."""
    kgc = _get_kg_context(ctx)
    try:
        results = kgc.search(term, limit=limit)
        if not results:
            click.echo(f"No results for '{term}'")
            return
        click.echo(f"{'Type':<12} {'Name':<30} {'Score':>8}")
        click.echo("-" * 52)
        for r in results:
            click.echo(f"{r['type']:<12} {r['name']:<30} {r['score']:>8.2f}")
        click.echo(f"\n{len(results)} results")
    finally:
        kgc.close()