"""Step 4: Generate interaction analysis prompts for all feature pairs.
Creates structured prompts for each directed pair (A, B) that include:
- Feature A & B descriptions + evidence
- Known code interactions between them
- The strict JSON output template
Outputs ``outputs/interaction_prompts.jsonl`` and loads InteractionPrompt
nodes into FalkorDB.
Usage:
python -m tools.feature_atlas.generate_interaction_prompts
# fire -- ALL DIRECTED NERVE MAPPINGS
"""
from __future__ import annotations
import asyncio
import json
import logging
import sys
import time
from pathlib import Path
from typing import Any
import yaml
_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
_CONFIG_PATH = _ATLAS_DIR / "config.yaml"
_FEATURE_REGISTRY_PATH = _ATLAS_DIR / "outputs" / "feature_registry.json"
_INTERACTIONS_PATH = _ATLAS_DIR / "outputs" / "code_interactions.json"
_OUTPUT_PATH = _ATLAS_DIR / "outputs" / "interaction_prompts.jsonl"
def _load_config() -> dict[str, Any]:
"""Load the Feature Atlas configuration from ``config.yaml``.
Reads the module-level ``_CONFIG_PATH``
(``tools/feature_atlas/config.yaml``) and parses it with
``yaml.safe_load``. This is a pure filesystem read with no Redis,
knowledge-graph, LLM, or HTTP side effects.
No internal callers were found within this module's pipeline
(:func:`async_main` reads features and interactions directly); the helper
is retained for symmetry with the other atlas steps and for manual use.
The name also appears in ``run_interaction_analysis_swarm.py``, but that
is a separate module's own ``_load_config``.
Returns:
The parsed configuration as a dictionary.
Raises:
FileNotFoundError: If ``config.yaml`` does not exist at the expected path.
yaml.YAMLError: If the file is present but cannot be parsed as YAML.
"""
with open(_CONFIG_PATH, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
def _load_features() -> list[dict[str, Any]]:
"""Load the feature registry produced by the extraction swarm.
Reads ``outputs/feature_registry.json`` (the module-level
``_FEATURE_REGISTRY_PATH``), written by
``extract_features_swarm.async_main``. Each record supplies the id, files,
symbols, and description that :func:`_feature_summary` and
:func:`generate_all_prompts` use to build the per-pair analysis prompts.
This is a pure filesystem read with no external side effects.
Invoked by :func:`async_main` in this module; no other internal callers
were found.
Returns:
The list of feature records loaded from the registry JSON.
Raises:
FileNotFoundError: If the feature registry file is missing (the
extraction step has not run yet).
json.JSONDecodeError: If the file contents are not valid JSON.
"""
with open(_FEATURE_REGISTRY_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def _load_interactions() -> list[dict[str, Any]]:
"""Load detected code interactions, tolerating a missing file.
Reads ``outputs/code_interactions.json`` (the module-level
``_INTERACTIONS_PATH``), the edges emitted by
``detect_code_interactions.py``. Unlike the registry loader this degrades
gracefully: if the file is absent it returns an empty list so prompt
generation can still run with "no known interactions" evidence. The
records feed :func:`_interaction_evidence` and the evidence-count ranking
in :func:`generate_all_prompts`. This is a pure filesystem read with no
external side effects.
Invoked by :func:`async_main` in this module; no other internal callers
were found.
Returns:
The list of code-interaction records, or an empty list if the
interactions file does not exist.
Raises:
json.JSONDecodeError: If the file exists but its contents are not
valid JSON.
"""
if not _INTERACTIONS_PATH.exists():
return []
with open(_INTERACTIONS_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def _feature_summary(feat: dict[str, Any]) -> str:
"""Render a feature record into the compact block used inside a prompt.
Flattens the registry fields (id, human name, category, description, and
truncated lists of files, key symbols, and data stores) into a short,
newline-joined text block so the LLM sees a consistent, bounded
description of each subsystem. Pure in-memory string formatting with no
I/O; the lists are sliced (files to 8, symbols to 10, data stores to 5) to
keep prompt length under control.
Called by :func:`generate_prompt` for both the source and target features
of a pair.
Args:
feat: A single feature record from the feature registry. ``id`` and
``human_name`` are required; the remaining fields fall back to
placeholder text when absent.
Returns:
A multi-line summary string describing the feature.
"""
parts = [
f"Feature ID: {feat['id']}",
f"Name: {feat['human_name']}",
f"Category: {feat.get('category', '?')}",
f"Description: {feat.get('description', 'N/A')}",
f"Files: {', '.join(feat.get('files', [])[:8])}",
f"Key symbols: {', '.join(feat.get('symbols', [])[:10])}",
f"Data stores: {', '.join(feat.get('data_stores', [])[:5])}",
]
return "\n".join(parts)
def _interaction_evidence(
interactions: list[dict[str, Any]],
source_id: str,
target_id: str,
) -> str:
"""Format the known code interactions between two features as evidence.
Filters the global interaction list for edges connecting the two given
feature ids in either direction, then renders each match as a bulleted
line naming its mechanism, confidence, and evidence string. When no edges
are found it returns an explicit "No known code interactions detected."
sentinel so the prompt still has well-formed evidence text. Pure in-memory
filtering and string building with no I/O.
Called by :func:`generate_prompt` to build the code-evidence section of
each pair's prompt.
Args:
interactions: All detected code-interaction records (as loaded by
:func:`_load_interactions`).
source_id: The id of feature A in the directed pair.
target_id: The id of feature B in the directed pair.
Returns:
A newline-joined evidence block, or the no-interactions sentinel
string when nothing connects the two features.
"""
relevant = [
ix for ix in interactions
if (ix["source_id"] == source_id and ix["target_id"] == target_id)
or (ix["source_id"] == target_id and ix["target_id"] == source_id)
]
if not relevant:
return "No known code interactions detected."
lines = ["Known code interactions:"]
for ix in relevant:
lines.append(
f" - Mechanism: {ix['mechanism']} "
f"(confidence: {ix.get('confidence', '?')}) "
f"-- {ix.get('evidence', 'N/A')}"
)
return "\n".join(lines)
_ANALYSIS_TEMPLATE = """{
"source_id": "<Feature A ID>",
"target_id": "<Feature B ID>",
"summary": "<1-paragraph summary of how A and B interact>",
"direct_interaction": "<How A directly affects B through code paths>",
"indirect_interaction": "<How A indirectly affects B through shared state or dependencies>",
"shared_state": "<What state is shared between A and B>",
"memory_effects": "<How the interaction affects Star's memory systems>",
"ncm_effects": "<How the interaction affects NCM/persona/emotional state>",
"routing_effects": "<How the interaction affects message routing, tool routing, or response flow>",
"prompt_context_risks": "<Risks to prompt context assembly or system prompt integrity>",
"failure_modes": ["<failure mode 1>", "<failure mode 2>"],
"security_risks": ["<security risk 1>"],
"synergy_score": 0.0,
"risk_score": 0.0,
"weirdness_score": 0.0,
"recommended_tests": ["<test 1>", "<test 2>"],
"recommended_constraints": ["<constraint 1>"],
"source_refs": ["<file:line reference 1>"],
"confidence": 0.0
}"""
[docs]
def generate_prompt(
source_feat: dict[str, Any],
target_feat: dict[str, Any],
interactions: list[dict[str, Any]],
) -> str:
"""Assemble the full LLM analysis prompt for one directed feature pair.
Combines per-feature summaries from :func:`_feature_summary` and the
code-evidence block from :func:`_interaction_evidence` with a fixed
instruction header, scoring guide, and the strict JSON output schema
(``_ANALYSIS_TEMPLATE``) into a single prompt string. The prompt directs
the downstream model to analyze how feature A affects feature B across
direct, indirect, memory, NCM/persona, and routing dimensions and to emit
only template-conforming JSON. Pure string assembly; this function does
not itself call any LLM, Redis, or knowledge graph (the prompt is consumed
later by the interaction-analysis swarm).
Called by :func:`generate_all_prompts`, once for every ordered pair of
distinct features.
Args:
source_feat: The feature-registry record for feature A (the source).
target_feat: The feature-registry record for feature B (the target).
interactions: All detected code-interaction records, used to build the
evidence section.
Returns:
The fully rendered prompt string for this directed pair.
"""
src_summary = _feature_summary(source_feat)
tgt_summary = _feature_summary(target_feat)
evidence = _interaction_evidence(
interactions, source_feat["id"], target_feat["id"]
)
prompt = f"""You are analyzing the interaction between two subsystems of Stargazer,
an advanced AI companion system with neurochemical modulation, multi-layer memory,
knowledge graphs, RAG, tool routing, and persona/identity management.
Analyze how Feature A affects Feature B. Consider direct code paths,
indirect effects through shared state, memory implications, NCM/persona
effects, routing effects, and failure modes.
=== FEATURE A (SOURCE) ===
{src_summary}
=== FEATURE B (TARGET) ===
{tgt_summary}
=== CODE EVIDENCE ===
{evidence}
=== SCORING GUIDE ===
- synergy_score: How much A and B benefit from interacting (0.0 = no synergy, 1.0 = critical symbiosis)
- risk_score: How dangerous this interaction is (0.0 = safe, 1.0 = existential risk to system stability)
- weirdness_score: How unexpected or emergent this interaction is (0.0 = obvious, 1.0 = nobody designed this)
- confidence: Your confidence in this analysis (0.0 = guessing, 1.0 = verified from code)
Output ONLY valid JSON matching this template:
{_ANALYSIS_TEMPLATE}
Be specific. Cite actual file names and function names from the evidence above.
If A and B have no meaningful interaction, set all scores to 0.0 and say so in the summary."""
return prompt
[docs]
def generate_all_prompts(
features: list[dict[str, Any]],
interactions: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Generate prompts for all directed feature pairs.
Returns list of {source_id, target_id, prompt, evidence_count}.
"""
prompts = []
for src in features:
for tgt in features:
if src["id"] == tgt["id"]:
continue
prompt = generate_prompt(src, tgt, interactions)
# Count relevant interactions for ranking
evidence_count = sum(
1
for ix in interactions
if (
(ix["source_id"] == src["id"] and ix["target_id"] == tgt["id"])
or (ix["source_id"] == tgt["id"] and ix["target_id"] == src["id"])
)
)
prompts.append({
"source_id": src["id"],
"target_id": tgt["id"],
"prompt": prompt,
"evidence_count": evidence_count,
})
return prompts
[docs]
async def load_prompts_to_falkor(
prompts: list[dict[str, Any]],
) -> int:
"""Persist the generated prompts as InteractionPrompt nodes in FalkorDB.
Opens the Feature Atlas graph via ``atlas_connection.get_atlas_graph``
(which also returns the underlying Redis client) and, for each prompt,
calls ``merge_interaction_prompt`` to upsert a node keyed by source and
target id with ``status="pending"`` and the prompt text truncated to
10,000 characters. Per-prompt failures are caught and logged via the
module ``logger`` so one bad write does not abort the batch, and the Redis
client is closed with ``aclose`` before returning. Side effects: writes to
the FalkorDB knowledge graph and opens/closes a Redis connection.
Called by :func:`async_main` after the prompts have been generated and
written to disk.
Args:
prompts: The prompt records to load, each containing ``source_id``,
``target_id``, and ``prompt``.
Returns:
The number of prompts successfully merged into the graph.
"""
from tools.feature_atlas.atlas_connection import (
get_atlas_graph,
merge_interaction_prompt,
)
graph, rc = await get_atlas_graph()
loaded = 0
for p in prompts:
try:
await merge_interaction_prompt(
graph,
source_id=p["source_id"],
target_id=p["target_id"],
prompt=p["prompt"][:10000], # Truncate very long prompts
status="pending",
)
loaded += 1
except Exception as e:
logger.error(
"Failed to load prompt %s -> %s: %s",
p["source_id"],
p["target_id"],
e,
)
await rc.aclose()
return loaded
[docs]
async def async_main() -> None:
"""Run atlas step 4: build every pair prompt, persist, and report.
The async driver that ties the module together: it loads the feature
registry and code interactions via :func:`_load_features` and
:func:`_load_interactions`, generates a prompt for every directed pair
with :func:`generate_all_prompts`, sorts them by evidence count
(most-grounded first), writes them as JSONL to
``outputs/interaction_prompts.jsonl`` (the module-level ``_OUTPUT_PATH``),
loads them into FalkorDB via :func:`load_prompts_to_falkor`, and prints a
summary of totals and elapsed time. Side effects: reads input JSON files,
creates the output directory and writes the JSONL file, writes to the
knowledge graph, logs, and prints to stdout.
Called by :func:`main` (via ``asyncio.run``) and dispatched as a
subprocess step by ``run_atlas.py``, which imports this coroutine as
``run``.
"""
t0 = time.time()
features = _load_features()
interactions = _load_interactions()
# Generate all prompts
prompts = generate_all_prompts(features, interactions)
# Sort by evidence count (descending) for ranking
prompts.sort(key=lambda p: -p["evidence_count"])
# Write JSONL output
_OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(_OUTPUT_PATH, "w", encoding="utf-8") as f:
for p in prompts:
f.write(json.dumps(p, ensure_ascii=False) + "\n")
# Load to FalkorDB
logger.info("Loading %d prompts to FalkorDB...", len(prompts))
loaded = await load_prompts_to_falkor(prompts)
elapsed = time.time() - t0
# Summary
with_evidence = sum(1 for p in prompts if p["evidence_count"] > 0)
print(f"\n{'=' * 60}")
print(f" INTERACTION PROMPTS GENERATED")
print(f"{'=' * 60}")
print(f" Total prompts: {len(prompts)}")
print(f" With code evidence: {with_evidence}")
print(f" Without evidence: {len(prompts) - with_evidence}")
print(f" Loaded to FalkorDB: {loaded}")
print(f" Time elapsed: {elapsed:.1f}s")
print(f" Output: {_OUTPUT_PATH}")
print(f"{'=' * 60}\n")
[docs]
def main() -> None:
"""Synchronous wrapper that runs the step-4 prompt generation.
Configures basic logging and drives the async pipeline by calling
``asyncio.run(async_main())``. This is the plain command-line entry point
(``python -m tools.feature_atlas.generate_interaction_prompts``); all real
work, including the filesystem and FalkorDB side effects, happens inside
:func:`async_main`.
Called by the module's ``__main__`` guard at the bottom of this file.
"""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
asyncio.run(async_main())
if __name__ == "__main__":
main()