"""Step 8: Generate the Sarah demo report.
Queries the atlas graph for all features, top risks, top synergies, and
concrete examples, then composes a polished markdown report designed to
be copy-pasted into a Discord message or shown on screen.
Can optionally use Gemini Flash to generate a narrative summary.
Usage:
python -m tools.feature_atlas.export_sarah_demo
# fire skull heart -- THE BODY SPEAKS TO SARAH
"""
from __future__ import annotations
import asyncio
import json
import logging
import sys
import time
from datetime import datetime
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__)
_ATLAS_DIR = Path(__file__).resolve().parent
_OUTPUT_PATH = _ATLAS_DIR / "outputs" / "sarah_demo.md"
async def _query_safe(graph: Any, cypher: str, params: dict | None = None) -> list:
"""Run a read-only Cypher query against the atlas graph, swallowing errors.
A defensive wrapper around the FalkorDB graph's ``ro_query`` used
throughout report generation so that a single failing or empty query never
aborts the whole demo. It executes the supplied read-only Cypher, returns
the result set (or an empty list when there are no rows), and on any
exception logs a warning and returns an empty list instead of propagating.
The side effect is the read-only network round trip to FalkorDB.
Invoked repeatedly by :func:`generate_demo_report` in this module to fetch
feature counts, risks, synergies, examples, and recommended tests; no other
internal callers were found.
Args:
graph: The connected FalkorDB atlas graph handle.
cypher: The read-only Cypher query string to execute.
params: Optional query parameters bound into the Cypher.
Returns:
The query result set as a list of rows, or an empty list when the query
returns nothing or raises.
"""
try:
result = await graph.ro_query(cypher, params=params)
return result.result_set or []
except Exception as e:
logger.warning("Query failed: %s", e)
return []
[docs]
async def generate_demo_report(graph: Any) -> str:
"""Compose the full Sarah demo report from the atlas graph as markdown.
Builds the polished, copy-paste-ready report that showcases the Feature
Interaction Atlas. It issues a sequence of read-only Cypher queries through
:func:`_query_safe` against the FalkorDB atlas graph to gather node and edge
counts, the full feature table with confidence badges, the top ten
highest-risk and highest-synergy interactions, a few concrete examples with
source-file references and recommended tests, an aggregated list of top
recommended tests, and a trailing reference block of the Cypher queries
used. The assembled sections are joined into one markdown document. The only
side effects are the read-only queries against FalkorDB; this function does
not write to disk or mutate the graph.
Invoked by :func:`async_main` in this module and also imported by
:func:`tools.feature_atlas.query_atlas.async_main`, which reuses it to
render the same report from the query CLI; no other internal callers were
found.
Args:
graph: The connected FalkorDB atlas graph handle to query.
Returns:
The complete demo report as a single markdown-formatted string.
"""
now = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
sections: list[str] = []
# Header
sections.append(f"""# Stargazer Feature Interaction Atlas v0
## "The Bodygraph Demo"
> *Stargazer can now inspect her own codebase, identify her major organs,
> map which organs touch, and show the highest-risk and highest-value
> feature interactions.*
Generated: {now}
Graph: `stargazer_feature_interaction_atlas`
---
""")
# Stats
feat_count = await _query_safe(graph, "MATCH (f:Feature) RETURN count(f)")
edge_count = await _query_safe(
graph, "MATCH ()-[e:CODE_INTERACTS_WITH]->() RETURN count(e)"
)
prompt_count = await _query_safe(
graph, "MATCH (p:InteractionPrompt) RETURN count(p)"
)
analysis_count = await _query_safe(
graph, "MATCH (a:InteractionAnalysis) RETURN count(a)"
)
fc = feat_count[0][0] if feat_count else 0
ec = edge_count[0][0] if edge_count else 0
pc = prompt_count[0][0] if prompt_count else 0
ac = analysis_count[0][0] if analysis_count else 0
sections.append(f"""### Atlas Overview
| Metric | Count |
|--------|-------|
| Feature Nodes | {fc} |
| Code Interaction Edges | {ec} |
| Interaction Prompts | {pc} |
| Analyses Completed | {ac} |
""")
# Feature list
features = await _query_safe(
graph,
"MATCH (f:Feature) "
"RETURN f.id, f.human_name, f.category, f.confidence "
"ORDER BY f.category, f.id"
)
if features:
sections.append("### All Features\n")
sections.append("| Feature | Category | Confidence |")
sections.append("|---------|----------|------------|")
for row in features:
fid, name, cat, conf = row
conf = conf or 0.0
conf_emoji = (
"\u2705" if conf >= 0.7
else "\u26a0\ufe0f" if conf >= 0.4
else "\u274c"
)
sections.append(f"| **{fid}** ({name}) | {cat} | {conf_emoji} {conf:.2f} |")
sections.append("")
# Top 10 Risks
risks = await _query_safe(
graph,
"MATCH (a:InteractionAnalysis) "
"RETURN a.source_id, a.target_id, a.risk_score, a.synergy_score, "
"a.summary, a.failure_modes "
"ORDER BY a.risk_score DESC LIMIT 10"
)
if risks:
sections.append("\n### Top 10 Highest-Risk Interactions\n")
for i, row in enumerate(risks, 1):
src, tgt, risk, syn, summary, failure_json = row
risk = risk or 0.0
syn = syn or 0.0
failures = []
if failure_json:
try:
failures = json.loads(failure_json) if isinstance(failure_json, str) else failure_json
except json.JSONDecodeError:
pass
sections.append(f"**{i}. {src} \u2192 {tgt}** (risk: {risk:.2f}, synergy: {syn:.2f})")
if summary:
sections.append(f"> {summary}")
if failures:
sections.append(f" - Failure modes: {'; '.join(failures[:3])}")
sections.append("")
# Top 10 Synergies
synergies = await _query_safe(
graph,
"MATCH (a:InteractionAnalysis) "
"RETURN a.source_id, a.target_id, a.synergy_score, a.risk_score, "
"a.summary "
"ORDER BY a.synergy_score DESC LIMIT 10"
)
if synergies:
sections.append("\n### Top 10 Highest-Synergy Interactions\n")
for i, row in enumerate(synergies, 1):
src, tgt, syn, risk, summary = row
syn = syn or 0.0
risk = risk or 0.0
sections.append(f"**{i}. {src} \u2192 {tgt}** (synergy: {syn:.2f}, risk: {risk:.2f})")
if summary:
sections.append(f"> {summary}")
sections.append("")
# Concrete examples with source refs
examples = await _query_safe(
graph,
"MATCH (a:InteractionAnalysis) "
"WHERE a.source_refs IS NOT NULL AND a.source_refs <> '[]' "
"RETURN a.source_id, a.target_id, a.direct_interaction, "
"a.source_refs, a.recommended_tests "
"ORDER BY a.risk_score DESC LIMIT 3"
)
if examples:
sections.append("\n### Concrete Examples (with source file refs)\n")
for i, row in enumerate(examples, 1):
src, tgt, direct, refs_json, tests_json = row
refs = []
tests = []
if refs_json:
try:
refs = json.loads(refs_json) if isinstance(refs_json, str) else refs_json
except json.JSONDecodeError:
pass
if tests_json:
try:
tests = json.loads(tests_json) if isinstance(tests_json, str) else tests_json
except json.JSONDecodeError:
pass
sections.append(f"**Example {i}: {src} \u2192 {tgt}**\n")
if direct:
sections.append(f"*Direct interaction:* {direct}\n")
if refs:
sections.append("*Source files:*")
for ref in refs[:5]:
sections.append(f" - `{ref}`")
if tests:
sections.append("*Recommended tests:*")
for test in tests[:3]:
sections.append(f" - {test}")
sections.append("")
# Recommended tests (aggregated from all analyses)
all_tests = await _query_safe(
graph,
"MATCH (a:InteractionAnalysis) "
"WHERE a.recommended_tests IS NOT NULL AND a.recommended_tests <> '[]' "
"RETURN a.source_id, a.target_id, a.recommended_tests, a.risk_score "
"ORDER BY a.risk_score DESC LIMIT 20"
)
if all_tests:
seen_tests: list[str] = []
sections.append("\n### Top Recommended Tests\n")
for row in all_tests:
src, tgt, tests_json, risk = row
try:
tests = json.loads(tests_json) if isinstance(tests_json, str) else tests_json
except json.JSONDecodeError:
continue
for test in tests:
if test not in seen_tests and len(seen_tests) < 10:
seen_tests.append(test)
sections.append(f"{len(seen_tests)}. {test} *({src} \u2192 {tgt})*")
sections.append("")
# Cypher queries used
sections.append("""
---
### Cypher Queries Used
```cypher
-- All features
MATCH (f:Feature) RETURN f.id, f.human_name, f.category, f.confidence
ORDER BY f.category, f.id
-- Top 10 risks
MATCH (a:InteractionAnalysis)
RETURN a.source_id, a.target_id, a.risk_score, a.summary
ORDER BY a.risk_score DESC LIMIT 10
-- Top 10 synergies
MATCH (a:InteractionAnalysis)
RETURN a.source_id, a.target_id, a.synergy_score, a.summary
ORDER BY a.synergy_score DESC LIMIT 10
-- Feature neighborhood
MATCH (f:Feature {id: 'ToolRouter'})-[e:CODE_INTERACTS_WITH]-(t:Feature)
RETURN t.id, e.mechanism, e.confidence
ORDER BY e.confidence DESC
-- Full pair analysis
MATCH (a:InteractionAnalysis {source_id: 'NCMStateEngine', target_id: 'PersonaPrefillLayer'})
RETURN a
```
---
*Generated by Stargazer Feature Interaction Atlas v0*
*Graph: stargazer_feature_interaction_atlas*
""")
return "\n".join(sections)
[docs]
async def async_main() -> None:
"""Generate the Sarah demo report and write it to the outputs directory.
The async driver for step 8 of the atlas pipeline. It opens the FalkorDB
atlas connection via
:func:`tools.feature_atlas.atlas_connection.get_atlas_graph`, builds the
markdown report with :func:`generate_demo_report`, writes it to
``outputs/sarah_demo.md``, closes the underlying Redis client, and then
prints both a short completion banner (output path, size, elapsed time) and
the full report to stdout. Side effects span the read-only FalkorDB queries
made during report generation and the filesystem write of the report file.
Invoked by :func:`main` in this module's ``__main__`` guard and imported as
``async_main`` by :func:`tools.feature_atlas.run_atlas.step_export_demo`
(the ``export-demo`` step of the atlas runner); no other internal callers
were found.
Returns:
None.
"""
from tools.feature_atlas.atlas_connection import get_atlas_graph
t0 = time.time()
graph, rc = await get_atlas_graph()
report = await generate_demo_report(graph)
# Write to file
_OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(_OUTPUT_PATH, "w", encoding="utf-8") as f:
f.write(report)
await rc.aclose()
elapsed = time.time() - t0
print(f"\n{'=' * 60}")
print(f" SARAH DEMO REPORT GENERATED")
print(f"{'=' * 60}")
print(f" Output: {_OUTPUT_PATH}")
print(f" Size: {len(report):,} chars")
print(f" Time elapsed: {elapsed:.1f}s")
print(f"{'=' * 60}\n")
# Also print the report
print(report)
[docs]
def main() -> None:
"""Synchronous entry point for the Sarah demo report step.
Configures root logging at WARNING level (to keep the printed report clean)
and drives the async flow by calling ``asyncio.run(async_main())``, which
queries the FalkorDB atlas graph, writes ``outputs/sarah_demo.md``, and
prints the report. All FalkorDB and filesystem side effects happen
transitively inside :func:`async_main`; this wrapper only sets up logging
and starts the event loop.
Invoked from the module's ``if __name__ == "__main__"`` guard via
``python -m tools.feature_atlas.export_sarah_demo``; no other internal
callers were found.
Returns:
None.
"""
logging.basicConfig(
level=logging.WARNING,
format="%(asctime)s [%(levelname)s] %(message)s",
)
asyncio.run(async_main())
if __name__ == "__main__":
main()