"""Step 7: Interactive CLI for querying the Feature Interaction Atlas.
Provides commands for inspecting features, top risks/synergies,
feature neighborhoods, and generating the Sarah demo report.
Style matches ``memory_search.py`` -- ANSI terminal colors, clean
formatting, designed for SSH access.
Usage:
python -m tools.feature_atlas.query_atlas stats
python -m tools.feature_atlas.query_atlas list-features
python -m tools.feature_atlas.query_atlas top-risks
python -m tools.feature_atlas.query_atlas top-synergies
python -m tools.feature_atlas.query_atlas feature-neighborhood CoreMemory
python -m tools.feature_atlas.query_atlas explain-pair NCMStateEngine PersonaPrefillLayer
python -m tools.feature_atlas.query_atlas missing-evidence
python -m tools.feature_atlas.query_atlas sarah-demo
python -m tools.feature_atlas.query_atlas --interactive
# fire skull spider -- THE WITCH READS HER OWN X-RAY
"""
from __future__ import annotations
import argparse
import asyncio
import json
import logging
import sys
import textwrap
from pathlib import Path
from typing import Any
_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
if str(_PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(_PROJECT_ROOT))
logger = logging.getLogger(__name__)
# -- ANSI colors (matching memory_search.py) ----------------------------------
[docs]
class C:
"""ANSI color/style escape codes for the Atlas CLI's terminal output.
A namespace of raw SGR escape sequences (``RESET``, ``BOLD``, ``DIM`` and the
color constants) plus small :func:`staticmethod` helpers such as
:meth:`header` that wrap text in a style and reset it. Mirrors the palette in
``memory_search.py`` and is used by the ``cmd_*`` query functions and
``interactive_mode`` to colorize results when this module runs as a CLI.
"""
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
PURPLE = "\033[35m"
CYAN = "\033[36m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
RED = "\033[31m"
MAGENTA = "\033[95m"
WHITE = "\033[97m"
[docs]
@staticmethod
def entity(text: str) -> str:
"""Wrap ``text`` in bold-cyan ANSI codes to highlight an entity name.
Used to make feature ids and interaction endpoints stand out in
terminal output. Called by ``cmd_list_features``, ``cmd_top_risks``,
``cmd_top_synergies``, ``cmd_feature_neighborhood``, and
``cmd_explain_pair`` to style feature/interaction identifiers.
Args:
text: The entity label (e.g. a feature id) to colorize.
Returns:
str: ``text`` wrapped in the bold-cyan/reset escape sequence.
"""
return f"{C.BOLD}{C.CYAN}{text}{C.RESET}"
[docs]
@staticmethod
def score_color(score: float) -> str:
"""Pick an ANSI color for a 0-1 score using fixed risk thresholds.
Maps scores to a traffic-light scheme: red for ``>= 0.7`` (high),
yellow for ``>= 0.4`` (medium), and green otherwise (low). Called by
``C.bar`` to color the filled portion of its meter; no other callers
were found in the repository.
Args:
score: A score in the ``[0.0, 1.0]`` range.
Returns:
str: The bare ANSI color escape (``C.RED``, ``C.YELLOW``, or
``C.GREEN``) corresponding to the score's severity band.
"""
if score >= 0.7:
return C.RED
elif score >= 0.4:
return C.YELLOW
else:
return C.GREEN
[docs]
@staticmethod
def bar(score: float, width: int = 20) -> str:
"""Render a colored fixed-width text meter for a 0-1 score.
Fills ``int(score * width)`` cells with ``#`` and pads the remainder
with dimmed ``.`` characters, coloring the filled run via
``C.score_color``. Used throughout this module to visualize risk,
synergy, weirdness, and confidence values in ``cmd_stats``,
``cmd_list_features``, ``cmd_top_risks``, ``cmd_top_synergies``,
``cmd_feature_neighborhood``, and ``cmd_explain_pair``.
Args:
score: A score in the ``[0.0, 1.0]`` range determining the fill
fraction and color.
width: Total number of cells in the bar (default 20).
Returns:
str: The colored, reset-terminated meter string.
"""
filled = int(score * width)
color = C.score_color(score)
return f"{color}{'#' * filled}{C.DIM}{'.' * (width - filled)}{C.RESET}"
# -- Query functions -----------------------------------------------------------
[docs]
async def cmd_stats(graph: Any) -> None:
"""Print summary counts and average scores for the atlas graph.
Renders a terminal overview of the Feature Interaction Atlas: node and
edge tallies (features, ``CODE_INTERACTS_WITH`` edges, interaction
prompts, completed analyses, pending prompts), a per-category feature
breakdown, and average risk/synergy/weirdness/confidence scores across
all ``InteractionAnalysis`` nodes. Each count comes from a read-only
Cypher call to ``graph.ro_query`` against the FalkorDB atlas namespace
(``stargazer_feature_interaction_atlas``); per-query failures are caught
and printed in red rather than aborting the whole summary. Output goes to
stdout via ``print`` using the ANSI helpers ``C.header`` and ``C.bar``.
Called by ``interactive_mode`` (the ``stats`` REPL command) and by
``async_main`` for the ``stats`` subcommand.
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
Returns:
None. Results are written to stdout.
"""
print(C.header("\n -- Feature Interaction Atlas: Statistics --\n"))
queries = {
"Features": "MATCH (f:Feature) RETURN count(f)",
"Interactions (edges)": (
"MATCH ()-[e:CODE_INTERACTS_WITH]->() RETURN count(e)"
),
"Interaction Prompts": "MATCH (p:InteractionPrompt) RETURN count(p)",
"Analyses Completed": "MATCH (a:InteractionAnalysis) RETURN count(a)",
"Prompts Pending": (
"MATCH (p:InteractionPrompt {status: 'pending'}) RETURN count(p)"
),
}
for label, cypher in queries.items():
try:
result = await graph.ro_query(cypher)
count = result.result_set[0][0] if result.result_set else 0
print(f" {C.BOLD}{label:30s}{C.RESET} {C.GREEN}{count:>6,}{C.RESET}")
except Exception as e:
print(f" {C.BOLD}{label:30s}{C.RESET} {C.RED}ERROR: {e}{C.RESET}")
# Category breakdown
try:
result = await graph.ro_query(
"MATCH (f:Feature) RETURN f.category, count(f) ORDER BY count(f) DESC"
)
if result.result_set:
print(f"\n {C.BOLD}By Category:{C.RESET}")
for row in result.result_set:
print(f" {C.YELLOW}{row[0]:20s}{C.RESET} {row[1]:>3}")
except Exception:
pass
# Average scores (if analyses exist)
try:
result = await graph.ro_query(
"MATCH (a:InteractionAnalysis) "
"RETURN avg(a.risk_score), avg(a.synergy_score), "
"avg(a.weirdness_score), avg(a.confidence)"
)
if result.result_set and result.result_set[0][0] is not None:
row = result.result_set[0]
print(f"\n {C.BOLD}Average Scores:{C.RESET}")
print(f" Risk: {C.bar(row[0])} {row[0]:.3f}")
print(f" Synergy: {C.bar(row[1])} {row[1]:.3f}")
print(f" Weirdness: {C.bar(row[2])} {row[2]:.3f}")
print(f" Confidence:{C.bar(row[3])} {row[3]:.3f}")
except Exception:
pass
print()
[docs]
async def cmd_list_features(graph: Any) -> None:
"""Print every ``Feature`` node grouped by category.
Lists all features in the atlas, sorted by category then id, emitting a
bold category header whenever the category changes and one line per
feature with a confidence meter and human-readable name. The feature rows
come from a single read-only ``graph.ro_query`` against the FalkorDB
atlas; output is written to stdout via ``print`` using ``C.header``,
``C.entity``, and ``C.bar``. When the graph holds no features it prints a
dim "No features found" notice and returns early. Called by
``interactive_mode`` (the ``features``/``ls`` REPL command) and by
``async_main`` for the ``list-features`` subcommand.
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
Returns:
None. Results are written to stdout.
"""
print(C.header("\n -- All Features --\n"))
result = await graph.ro_query(
"MATCH (f:Feature) "
"RETURN f.id, f.human_name, f.category, f.confidence "
"ORDER BY f.category, f.id"
)
if not result.result_set:
print(f" {C.DIM}No features found.{C.RESET}\n")
return
current_cat = None
for row in result.result_set:
fid, name, cat, conf = row
conf = conf or 0.0
if cat != current_cat:
current_cat = cat
print(f"\n {C.BOLD}{C.YELLOW}[{cat}]{C.RESET}")
conf_bar = C.bar(conf, 10)
print(f" {C.entity(f'{fid:30s}')} {conf_bar} {conf:.2f} {C.DIM}{name}{C.RESET}")
print()
[docs]
async def cmd_top_risks(graph: Any, limit: int = 10) -> None:
"""Print the highest-risk analyzed interactions, worst first.
Queries ``InteractionAnalysis`` nodes ordered by ``risk_score`` descending
and prints up to ``limit`` of them, each as a source-to-target heading
followed by risk, synergy, and weirdness meters and a word-wrapped
summary. Data comes from a single read-only ``graph.ro_query`` against the
FalkorDB atlas; output is written to stdout via ``print`` using
``C.header``, ``C.entity``, ``C.bar``, and ``textwrap.fill``. When no
analyses exist it prints a dim notice and returns early. Called by
``interactive_mode`` (the ``risks`` REPL command, which parses an optional
numeric argument) and by ``async_main`` for the ``top-risks`` subcommand
(which passes ``args.limit``).
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
limit: Maximum number of interactions to display (default 10). This
value is interpolated directly into the Cypher ``LIMIT`` clause.
Returns:
None. Results are written to stdout.
"""
print(C.header(f"\n -- Top {limit} Highest Risk Interactions --\n"))
result = await graph.ro_query(
"MATCH (a:InteractionAnalysis) "
"RETURN a.source_id, a.target_id, a.risk_score, a.synergy_score, "
"a.weirdness_score, a.summary, a.confidence "
f"ORDER BY a.risk_score DESC LIMIT {limit}"
)
if not result.result_set:
print(f" {C.DIM}No analyses found.{C.RESET}\n")
return
for i, row in enumerate(result.result_set, 1):
src, tgt, risk, syn, weird, summary, conf = row
risk = risk or 0.0
syn = syn or 0.0
weird = weird or 0.0
conf = conf or 0.0
print(f" {C.BOLD}{C.RED}{i:2d}.{C.RESET} {C.entity(src)} -> {C.entity(tgt)}")
print(f" Risk: {C.bar(risk)} {risk:.3f}")
print(f" Synergy: {C.bar(syn)} {syn:.3f}")
print(f" Weird: {C.bar(weird)} {weird:.3f}")
if summary:
wrapped = textwrap.fill(
summary, width=70, initial_indent=" ", subsequent_indent=" "
)
print(f"{C.WHITE}{wrapped}{C.RESET}")
print()
[docs]
async def cmd_top_synergies(graph: Any, limit: int = 10) -> None:
"""Print the highest-synergy analyzed interactions, best first.
The synergy-ranked counterpart to ``cmd_top_risks``: queries
``InteractionAnalysis`` nodes ordered by ``synergy_score`` descending and
prints up to ``limit`` of them, each as a source-to-target heading
followed by synergy, risk, and weirdness meters and a word-wrapped
summary. Data comes from a single read-only ``graph.ro_query`` against the
FalkorDB atlas; output is written to stdout via ``print`` using
``C.header``, ``C.entity``, ``C.bar``, and ``textwrap.fill``. When no
analyses exist it prints a dim notice and returns early. Called by
``interactive_mode`` (the ``synergies`` REPL command, with an optional
numeric argument) and by ``async_main`` for the ``top-synergies``
subcommand (which passes ``args.limit``).
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
limit: Maximum number of interactions to display (default 10). This
value is interpolated directly into the Cypher ``LIMIT`` clause.
Returns:
None. Results are written to stdout.
"""
print(C.header(f"\n -- Top {limit} Highest Synergy Interactions --\n"))
result = await graph.ro_query(
"MATCH (a:InteractionAnalysis) "
"RETURN a.source_id, a.target_id, a.synergy_score, a.risk_score, "
"a.weirdness_score, a.summary, a.confidence "
f"ORDER BY a.synergy_score DESC LIMIT {limit}"
)
if not result.result_set:
print(f" {C.DIM}No analyses found.{C.RESET}\n")
return
for i, row in enumerate(result.result_set, 1):
src, tgt, syn, risk, weird, summary, conf = row
syn = syn or 0.0
risk = risk or 0.0
weird = weird or 0.0
print(f" {C.BOLD}{C.GREEN}{i:2d}.{C.RESET} {C.entity(src)} -> {C.entity(tgt)}")
print(f" Synergy: {C.bar(syn)} {syn:.3f}")
print(f" Risk: {C.bar(risk)} {risk:.3f}")
print(f" Weird: {C.bar(weird)} {weird:.3f}")
if summary:
wrapped = textwrap.fill(
summary, width=70, initial_indent=" ", subsequent_indent=" "
)
print(f"{C.WHITE}{wrapped}{C.RESET}")
print()
[docs]
async def cmd_feature_neighborhood(graph: Any, feature_id: str) -> None:
"""Print a full profile of one feature and everything touching it.
Gives a 360-degree view of a single ``Feature``: its metadata (name,
category, confidence meter, description, and first ten source files),
then its outgoing and incoming ``CODE_INTERACTS_WITH`` edges with
mechanism, confidence, and a truncated evidence snippet, and finally up to
ten ``InteractionAnalysis`` rows that name it as source or target. This
issues four separate parameterized read-only ``graph.ro_query`` calls
against the FalkorDB atlas (all keyed on ``feature_id`` via ``$id``), and
decodes the JSON ``files`` blob with ``json.loads``. Output goes to stdout
via ``print`` using ``C.header``, ``C.entity``, ``C.bar``, and
``textwrap.fill``; a missing feature prints a red notice and returns
early. Called by ``interactive_mode`` (the ``hood``/``neighborhood`` REPL
command) and by ``async_main`` for the ``feature-neighborhood``
subcommand.
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
feature_id: The ``id`` of the feature to inspect, bound to the ``$id``
Cypher parameter.
Returns:
None. Results are written to stdout.
"""
print(C.header(f"\n -- Neighborhood: {feature_id} --\n"))
# Feature info
result = await graph.ro_query(
"MATCH (f:Feature {id: $id}) "
"RETURN f.human_name, f.category, f.confidence, f.description, f.files",
params={"id": feature_id},
)
if not result.result_set:
print(f" {C.RED}Feature '{feature_id}' not found.{C.RESET}\n")
return
row = result.result_set[0]
name, cat, conf, desc, files_json = row
print(f" {C.BOLD}Name:{C.RESET} {name}")
print(f" {C.BOLD}Category:{C.RESET} {cat}")
print(f" {C.BOLD}Confidence:{C.RESET} {C.bar(conf or 0)} {conf or 0:.2f}")
if desc:
wrapped = textwrap.fill(desc, width=70, initial_indent=" ", subsequent_indent=" ")
print(f" {C.BOLD}Description:{C.RESET}")
print(f"{C.WHITE}{wrapped}{C.RESET}")
if files_json:
try:
files = json.loads(files_json)
print(f" {C.BOLD}Files:{C.RESET}")
for f in files[:10]:
print(f" {C.DIM}{f}{C.RESET}")
except json.JSONDecodeError:
pass
# Outgoing interactions
print(C.header(f"\n Outgoing CODE_INTERACTS_WITH:"))
result = await graph.ro_query(
"MATCH (f:Feature {id: $id})-[e:CODE_INTERACTS_WITH]->(t:Feature) "
"RETURN t.id, e.mechanism, e.confidence, e.evidence "
"ORDER BY e.confidence DESC",
params={"id": feature_id},
)
if result.result_set:
for row in result.result_set:
tgt, mech, conf, ev = row
print(f" -> {C.entity(tgt)} [{C.YELLOW}{mech}{C.RESET}] conf={conf or 0:.2f}")
if ev:
print(f" {C.DIM}{ev[:100]}{C.RESET}")
else:
print(f" {C.DIM}None{C.RESET}")
# Incoming interactions
print(C.header(f"\n Incoming CODE_INTERACTS_WITH:"))
result = await graph.ro_query(
"MATCH (s:Feature)-[e:CODE_INTERACTS_WITH]->(f:Feature {id: $id}) "
"RETURN s.id, e.mechanism, e.confidence, e.evidence "
"ORDER BY e.confidence DESC",
params={"id": feature_id},
)
if result.result_set:
for row in result.result_set:
src, mech, conf, ev = row
print(f" <- {C.entity(src)} [{C.YELLOW}{mech}{C.RESET}] conf={conf or 0:.2f}")
if ev:
print(f" {C.DIM}{ev[:100]}{C.RESET}")
else:
print(f" {C.DIM}None{C.RESET}")
# Analyses involving this feature
print(C.header(f"\n Analyses:"))
result = await graph.ro_query(
"MATCH (a:InteractionAnalysis) "
"WHERE a.source_id = $id OR a.target_id = $id "
"RETURN a.source_id, a.target_id, a.risk_score, a.synergy_score, "
"a.weirdness_score, a.summary "
"ORDER BY a.risk_score DESC LIMIT 10",
params={"id": feature_id},
)
if result.result_set:
for row in result.result_set:
src, tgt, risk, syn, weird, summary = row
risk = risk or 0.0
syn = syn or 0.0
weird = weird or 0.0
direction = f"{src} -> {tgt}"
print(
f" {C.entity(direction):50s} "
f"R={risk:.2f} S={syn:.2f} W={weird:.2f}"
)
else:
print(f" {C.DIM}None{C.RESET}")
print()
[docs]
async def cmd_explain_pair(
graph: Any, source_id: str, target_id: str
) -> None:
"""Print the complete stored analysis for one ordered feature pair.
Looks up the single ``InteractionAnalysis`` matching ``source_id`` ->
``target_id`` and renders every recorded field: the narrative sections
(summary, direct and indirect interaction, shared state, memory, NCM,
routing, and prompt-context effects), the JSON list fields (failure
modes, security risks, recommended tests and constraints, source refs),
and the four numeric scores with meters. The data comes from one
parameterized read-only ``graph.ro_query`` against the FalkorDB atlas
(bound via ``$s`` and ``$t``); list fields are parsed with ``json.loads``
when stored as strings. Output is written to stdout via ``print`` using
``C.header``, ``C.bar``, and ``textwrap.fill``; a missing pair prints a
dim notice and returns early. Called by ``interactive_mode`` (the
``pair``/``explain`` REPL command) and by ``async_main`` for the
``explain-pair`` subcommand.
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
source_id: The interaction source feature id, bound to ``$s``.
target_id: The interaction target feature id, bound to ``$t``.
Returns:
None. Results are written to stdout.
"""
print(C.header(f"\n -- Pair Analysis: {source_id} -> {target_id} --\n"))
result = await graph.ro_query(
"MATCH (a:InteractionAnalysis {source_id: $s, target_id: $t}) "
"RETURN a.summary, a.direct_interaction, a.indirect_interaction, "
"a.shared_state, a.memory_effects, a.ncm_effects, "
"a.routing_effects, a.prompt_context_risks, "
"a.failure_modes, a.security_risks, "
"a.synergy_score, a.risk_score, a.weirdness_score, "
"a.recommended_tests, a.recommended_constraints, "
"a.source_refs, a.confidence",
params={"s": source_id, "t": target_id},
)
if not result.result_set:
print(f" {C.DIM}No analysis found for this pair.{C.RESET}\n")
return
row = result.result_set[0]
fields = [
"Summary", "Direct Interaction", "Indirect Interaction",
"Shared State", "Memory Effects", "NCM Effects",
"Routing Effects", "Prompt Context Risks",
]
for i, label in enumerate(fields):
val = row[i] or ""
if val:
print(f" {C.BOLD}{label}:{C.RESET}")
wrapped = textwrap.fill(
val, width=72, initial_indent=" ", subsequent_indent=" "
)
print(f"{C.WHITE}{wrapped}{C.RESET}\n")
# Lists
for label, idx in [
("Failure Modes", 8),
("Security Risks", 9),
("Recommended Tests", 13),
("Recommended Constraints", 14),
("Source Refs", 15),
]:
raw = row[idx]
if raw:
try:
items = json.loads(raw) if isinstance(raw, str) else raw
if items:
print(f" {C.BOLD}{label}:{C.RESET}")
for item in items:
print(f" - {C.WHITE}{item}{C.RESET}")
print()
except json.JSONDecodeError:
pass
# Scores
print(f" {C.BOLD}Scores:{C.RESET}")
for label, idx in [("Synergy", 10), ("Risk", 11), ("Weirdness", 12), ("Confidence", 16)]:
val = row[idx] or 0.0
print(f" {label:15s} {C.bar(val)} {val:.3f}")
print()
[docs]
async def cmd_missing_evidence(graph: Any) -> None:
"""Print features whose confidence is below 0.5, weakest first.
Surfaces under-evidenced features so a human can prioritize follow-up:
selects every ``Feature`` with ``confidence < 0.5``, ordered ascending,
and prints each id in red alongside its confidence, its source-file count
(derived by ``json.loads`` on the stored ``files`` blob), and its human
name. Data comes from one read-only ``graph.ro_query`` against the
FalkorDB atlas; output is written to stdout via ``print`` using
``C.header``. When no feature falls below the threshold it prints a green
"all features have confidence >= 0.5" message and returns early. Called by
``interactive_mode`` (the ``missing`` REPL command) and by ``async_main``
for the ``missing-evidence`` subcommand.
Args:
graph: The FalkorDB atlas graph handle exposing ``ro_query``.
Returns:
None. Results are written to stdout.
"""
print(C.header("\n -- Features with Low Confidence (<0.5) --\n"))
result = await graph.ro_query(
"MATCH (f:Feature) WHERE f.confidence < 0.5 "
"RETURN f.id, f.human_name, f.confidence, f.files "
"ORDER BY f.confidence ASC"
)
if not result.result_set:
print(f" {C.GREEN}All features have confidence >= 0.5{C.RESET}\n")
return
for row in result.result_set:
fid, name, conf, files_json = row
conf = conf or 0.0
file_count = 0
if files_json:
try:
file_count = len(json.loads(files_json))
except json.JSONDecodeError:
pass
print(
f" {C.RED}{fid:30s}{C.RESET} "
f"conf={conf:.2f} files={file_count} "
f"{C.DIM}{name}{C.RESET}"
)
print()
[docs]
async def cmd_sarah_demo(graph: Any) -> None:
"""Build and print the curated "Sarah" demo report inline.
Lazily imports ``generate_demo_report`` from
``tools.feature_atlas.export_sarah_demo`` (kept inside the function so the
rest of the CLI works even if that module is unavailable), awaits it
against the live atlas graph to assemble the formatted demo string, and
prints the result to stdout. The report itself runs further read-only
queries against the FalkorDB atlas inside ``generate_demo_report``. Called
by ``interactive_mode`` (the ``demo``/``sarah`` REPL command) and by
``async_main`` for the ``sarah-demo`` subcommand.
Args:
graph: The FalkorDB atlas graph handle passed through to
``generate_demo_report``.
Returns:
None. The rendered report is written to stdout.
"""
from tools.feature_atlas.export_sarah_demo import generate_demo_report
report = await generate_demo_report(graph)
print(report)
[docs]
def print_banner() -> None:
"""Print the boxed ASCII banner for the atlas query CLI.
Writes a purple/bold framed title block ("FEATURE INTERACTION ATLAS" /
"Stargazer Bodygraph") to stdout via ``print``. Purely cosmetic with no
graph or I/O side effects beyond the terminal write. Called by
``interactive_mode`` once at REPL startup and by ``async_main`` before
each non-interactive subcommand so single-shot invocations are also
branded. (Note: ``memory_search.py`` defines its own same-named banner;
they are independent.)
Returns:
None. The banner is written to stdout.
"""
print(f"""
{C.PURPLE}{C.BOLD}+{'=' * 70}+
| FEATURE INTERACTION ATLAS |
| Stargazer Bodygraph -- Self-Inspection System v0 |
+{'=' * 70}+{C.RESET}
""")
[docs]
async def interactive_mode(graph: Any) -> None:
"""Run the blocking REPL loop that dispatches atlas query commands.
Prints the banner and a command cheat-sheet, then loops reading lines from
stdin via the builtin ``input`` (so it blocks the event loop on terminal
I/O), splitting each into a command plus arguments and awaiting the
matching ``cmd_*`` coroutine: ``stats``, ``features``/``ls``,
``risks``/``synergies`` (with an optional numeric limit), ``hood``,
``pair``, ``missing``, and ``demo``. ``quit``/``exit``/``q`` and EOF or
Ctrl-C break the loop; unknown commands and any exception raised by a
handler are reported in red and the loop continues. All work flows through
the shared ``graph`` handle into the FalkorDB atlas. Called by
``async_main`` when the ``--interactive`` flag is set or the command
defaults to ``interactive``.
Args:
graph: The FalkorDB atlas graph handle passed to each dispatched
``cmd_*`` coroutine.
Returns:
None. Returns when the user quits or input ends.
"""
print_banner()
print(f" {C.DIM}Commands:{C.RESET}")
print(f" {C.CYAN}stats{C.RESET} Atlas statistics")
print(f" {C.CYAN}features{C.RESET} List all features")
print(f" {C.CYAN}risks [N]{C.RESET} Top N highest risk interactions")
print(f" {C.CYAN}synergies [N]{C.RESET} Top N highest synergy interactions")
print(f" {C.CYAN}hood <feature_id>{C.RESET} Feature neighborhood")
print(f" {C.CYAN}pair <src> <tgt>{C.RESET} Explain a specific pair")
print(f" {C.CYAN}missing{C.RESET} Low-confidence features")
print(f" {C.CYAN}demo{C.RESET} Generate Sarah demo report")
print(f" {C.CYAN}quit{C.RESET} Exit")
print()
while True:
try:
raw = input(f" {C.PURPLE}atlas>{C.RESET} ").strip()
except (EOFError, KeyboardInterrupt):
print()
break
if not raw:
continue
parts = raw.split()
cmd = parts[0].lower()
try:
if cmd in ("quit", "exit", "q"):
break
elif cmd == "stats":
await cmd_stats(graph)
elif cmd in ("features", "list-features", "ls"):
await cmd_list_features(graph)
elif cmd in ("risks", "top-risks"):
limit = int(parts[1]) if len(parts) > 1 else 10
await cmd_top_risks(graph, limit)
elif cmd in ("synergies", "top-synergies"):
limit = int(parts[1]) if len(parts) > 1 else 10
await cmd_top_synergies(graph, limit)
elif cmd in ("hood", "neighborhood", "feature-neighborhood"):
if len(parts) < 2:
print(f" {C.RED}Usage: hood <feature_id>{C.RESET}")
else:
await cmd_feature_neighborhood(graph, parts[1])
elif cmd in ("pair", "explain-pair", "explain"):
if len(parts) < 3:
print(f" {C.RED}Usage: pair <source_id> <target_id>{C.RESET}")
else:
await cmd_explain_pair(graph, parts[1], parts[2])
elif cmd in ("missing", "missing-evidence"):
await cmd_missing_evidence(graph)
elif cmd in ("demo", "sarah-demo", "sarah"):
await cmd_sarah_demo(graph)
else:
print(f" {C.RED}Unknown command: {cmd}{C.RESET}")
except Exception as e:
print(f" {C.RED}Error: {e}{C.RESET}")
[docs]
async def async_main() -> None:
"""Parse CLI arguments, open the atlas graph, and dispatch one command.
The asynchronous core of the CLI: it builds the ``argparse`` parser
(positional command plus extra ``args``, ``--limit``/``-n``, and
``--interactive``/``-i``), opens a connection to the FalkorDB atlas via
``get_atlas_graph`` (lazily imported from
``tools.feature_atlas.atlas_connection``), then routes to the appropriate
handler: ``interactive_mode`` for the interactive flag/command, or one of
the ``cmd_*`` coroutines (preceded by ``print_banner``) for a single-shot
subcommand. ``feature-neighborhood`` and ``explain-pair`` validate that
their positional ids are present and otherwise print a usage hint. The
Redis connection returned alongside the graph is always closed via
``rc.aclose()`` in a ``finally`` block. Called by ``main`` through
``asyncio.run(async_main())``.
Returns:
None.
"""
from tools.feature_atlas.atlas_connection import get_atlas_graph
parser = argparse.ArgumentParser(
description="Feature Interaction Atlas -- Query CLI"
)
parser.add_argument(
"command",
nargs="?",
default="interactive",
choices=[
"stats",
"list-features",
"top-risks",
"top-synergies",
"feature-neighborhood",
"explain-pair",
"missing-evidence",
"sarah-demo",
"interactive",
],
)
parser.add_argument("args", nargs="*", default=[])
parser.add_argument("--limit", "-n", type=int, default=10)
parser.add_argument("--interactive", "-i", action="store_true")
args = parser.parse_args()
graph, rc = await get_atlas_graph()
try:
if args.interactive or args.command == "interactive":
await interactive_mode(graph)
elif args.command == "stats":
print_banner()
await cmd_stats(graph)
elif args.command == "list-features":
print_banner()
await cmd_list_features(graph)
elif args.command == "top-risks":
print_banner()
await cmd_top_risks(graph, args.limit)
elif args.command == "top-synergies":
print_banner()
await cmd_top_synergies(graph, args.limit)
elif args.command == "feature-neighborhood":
if not args.args:
print(f"{C.RED}Usage: query_atlas feature-neighborhood <feature_id>{C.RESET}")
else:
print_banner()
await cmd_feature_neighborhood(graph, args.args[0])
elif args.command == "explain-pair":
if len(args.args) < 2:
print(f"{C.RED}Usage: query_atlas explain-pair <source_id> <target_id>{C.RESET}")
else:
print_banner()
await cmd_explain_pair(graph, args.args[0], args.args[1])
elif args.command == "missing-evidence":
print_banner()
await cmd_missing_evidence(graph)
elif args.command == "sarah-demo":
print_banner()
await cmd_sarah_demo(graph)
finally:
await rc.aclose()
[docs]
def main() -> None:
"""Configure logging and run the async CLI to completion.
The synchronous process entry point: sets up basic ``WARNING``-level
logging and then drives the whole CLI via ``asyncio.run(async_main())``,
which parses arguments, opens the atlas graph, and dispatches the chosen
command. Invoked from this module's ``__main__`` guard (e.g.
``python -m tools.feature_atlas.query_atlas ...``); no other internal
caller exists.
Returns:
None.
"""
logging.basicConfig(
level=logging.WARNING,
format="%(asctime)s [%(levelname)s] %(message)s",
)
asyncio.run(async_main())
if __name__ == "__main__":
main()