diff --git a/Dockerfile b/Dockerfile index 69e7bdc..f3d7ad2 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,4 +14,5 @@ RUN poetry install --only=main EXPOSE 8000 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ CMD curl -f http://localhost:8000/health || exit 1 +ENV L9_LIFECYCLE_HOOK=engine.boot:CodegraphLifecycle CMD ["uvicorn", "engine.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/chassis/__init__.py b/chassis/__init__.py index 9e44683..3e7cdcf 100644 --- a/chassis/__init__.py +++ b/chassis/__init__.py @@ -16,7 +16,7 @@ """ from chassis.actions import execute_action, register_handler, register_handlers -from chassis.app import LifecycleHook, create_app +from chassis.chassis_app import LifecycleHook, create_app from chassis.errors import ( AuthorizationError, ChassisError, diff --git a/domains/revopsos/constellation.yaml b/domains/revopsos/constellation.yaml new file mode 100644 index 0000000..22b16b2 --- /dev/null +++ b/domains/revopsos/constellation.yaml @@ -0,0 +1,177 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [domain] +# tags: [plangraph, constellation, revopsos] +# status: active +# --- /L9_META --- +# RevOpsOS constellation — 8-service revenue operations platform + +constellation: revopsos + +services: + ENRICH: + status: built + description: Contact and company data enrichment from external sources + depends_on: [] + + GRAPH: + status: built + description: Relationship graph engine — maps accounts, contacts, and deals + depends_on: [] + + SCORE: + status: planned + description: ML-based lead and account scoring + depends_on: [ENRICH, GRAPH] + + ROUTE: + status: planned + description: Intelligent lead and opportunity routing to reps/queues + depends_on: [SCORE, GRAPH] + + FORECAST: + status: planned + description: Revenue forecast engine using pipeline + historical signals + depends_on: [SCORE, ROUTE] + + SIGNAL: + status: planned + description: Real-time intent and behavioral signal ingestion + depends_on: [ENRICH] + + HEALTH: + status: planned + description: Account health scoring and churn risk detection + depends_on: [SCORE, ENRICH] + + HANDOFF: + status: planned + description: Sales-to-CS handoff automation and playbook execution + depends_on: [ROUTE, SCORE] + +interfaces: + - name: enrich_inbound + service: ENRICH + direction: inbound + protocol: http + description: Accepts raw contact/company records for enrichment + + - name: enrich_outbound + service: ENRICH + direction: outbound + protocol: http + description: Emits enriched records to downstream services + + - name: graph_inbound + service: GRAPH + direction: inbound + protocol: bolt + description: Receives entity relationship data + + - name: graph_query + service: GRAPH + direction: outbound + protocol: bolt + description: Serves graph queries to SCORE and ROUTE + + - name: score_inbound + service: SCORE + direction: inbound + protocol: http + description: Accepts enriched + graph data for scoring + + - name: score_outbound + service: SCORE + direction: outbound + protocol: http + description: Emits scored leads/accounts to ROUTE, FORECAST, HEALTH, HANDOFF + + - name: route_inbound + service: ROUTE + direction: inbound + protocol: http + description: Accepts scored leads for routing decisions + + - name: route_outbound + service: ROUTE + direction: outbound + protocol: http + description: Emits routing decisions to CRM and HANDOFF + + - name: forecast_inbound + service: FORECAST + direction: inbound + protocol: http + description: Accepts pipeline + score data for forecasting + + - name: signal_inbound + service: SIGNAL + direction: inbound + protocol: webhook + description: Receives real-time behavioral events + + - name: signal_outbound + service: SIGNAL + direction: outbound + protocol: http + description: Forwards processed signals to ENRICH and SCORE + + - name: health_inbound + service: HEALTH + direction: inbound + protocol: http + description: Accepts account data for health scoring + + - name: handoff_inbound + service: HANDOFF + direction: inbound + protocol: http + description: Triggers handoff workflows from routing decisions + +flows: + - from: ENRICH + to: SCORE + label: enriched_data + + - from: GRAPH + to: SCORE + label: graph_context + + - from: SCORE + to: ROUTE + label: scored_leads + + - from: SCORE + to: FORECAST + label: score_signals + + - from: SCORE + to: HEALTH + label: health_inputs + + - from: SCORE + to: HANDOFF + label: handoff_triggers + + - from: ROUTE + to: FORECAST + label: routing_outcomes + + - from: ROUTE + to: HANDOFF + label: handoff_assignments + + - from: SIGNAL + to: ENRICH + label: signal_enrichment + feedback: false + + - from: HANDOFF + to: ROUTE + label: handoff_feedback + feedback: true + + - from: FORECAST + to: SCORE + label: forecast_feedback + feedback: true diff --git a/engine/boot.py b/engine/boot.py new file mode 100644 index 0000000..f37045e --- /dev/null +++ b/engine/boot.py @@ -0,0 +1,97 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [api] +# tags: [chassis, lifecycle, boot] +# status: active +# --- /L9_META --- +"""engine/boot.py — CodegraphLifecycle: chassis LifecycleHook for l9-codegraph.""" +from __future__ import annotations + +import logging +from typing import Any + +from chassis.chassis_app import LifecycleHook +from engine.settings import Settings + +logger = logging.getLogger(__name__) +settings = Settings() + + +class CodegraphLifecycle(LifecycleHook): + """Wires Neo4j connections for CodeGraph + PlanGraph engines.""" + + def __init__(self) -> None: + self._driver: Any = None + + async def startup(self) -> None: + logger.info("CodegraphLifecycle.startup — Neo4j: %s", settings.neo4j_uri) + # Import here to avoid loading neo4j at module level (keeps tests fast) + from neo4j import AsyncGraphDatabase + self._driver = AsyncGraphDatabase.driver( + settings.neo4j_uri, + auth=(settings.neo4j_user, settings.neo4j_password), + ) + logger.info("CodegraphLifecycle.startup complete") + + async def shutdown(self) -> None: + if self._driver: + await self._driver.close() + logger.info("CodegraphLifecycle.shutdown complete") + + async def execute( + self, + action: str, + payload: dict[str, Any], + tenant: str, + trace_id: str, + ) -> dict[str, Any]: + logger.info("execute action=%s tenant=%s trace_id=%s", action, tenant, trace_id) + # Inject tenant + trace_id into payload so handlers can log them + enriched = {**payload, "tenant": tenant, "trace_id": trace_id} + + try: + if action == "search_codegraph": + from engine.codegraph.handler import handle_search_codegraph + data = await handle_search_codegraph(enriched) + elif action == "build_codegraph": + from engine.codegraph.handler import handle_build_codegraph + data = await handle_build_codegraph(enriched) + elif action == "search_plangraph": + from engine.plangraph.handler import handle_search_plangraph + data = await handle_search_plangraph(enriched) + elif action == "build_order": + from engine.plangraph.handler import handle_build_order + data = await handle_build_order(enriched) + elif action == "check_drift": + from engine.plangraph.handler import handle_check_drift + data = await handle_check_drift(enriched) + elif action == "load_constellation": + from engine.plangraph.handler import handle_load_constellation + data = await handle_load_constellation(enriched) + elif action == "health": + data = {"status": "healthy", "service": "l9-codegraph"} + else: + return { + "status": "failed", + "action": action, + "tenant": tenant, + "data": {"error": f"Unknown action: '{action}'"}, + "meta": {"trace_id": trace_id}, + } + except Exception as exc: + logger.exception("Handler failed action=%s: %s", action, exc) + return { + "status": "failed", + "action": action, + "tenant": tenant, + "data": {"error": str(exc)}, + "meta": {"trace_id": trace_id}, + } + + return { + "status": "ok", + "action": action, + "tenant": tenant, + "data": data, + "meta": {"trace_id": trace_id}, + } diff --git a/engine/codegraph/__init__.py b/engine/codegraph/__init__.py new file mode 100644 index 0000000..dfbaa43 --- /dev/null +++ b/engine/codegraph/__init__.py @@ -0,0 +1,12 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [codegraph, repograph] +# status: active +# --- /L9_META --- +"""Code Graph Engine — RepoGraph-inspired structural code intelligence.""" + +from .builder import CodeGraphBuilder +from .retriever import EgoGraphRetriever + +__all__ = ["CodeGraphBuilder", "EgoGraphRetriever"] diff --git a/engine/codegraph/builder.py b/engine/codegraph/builder.py new file mode 100644 index 0000000..b32789d --- /dev/null +++ b/engine/codegraph/builder.py @@ -0,0 +1,143 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [codegraph, builder, neo4j] +# status: active +# --- /L9_META --- +"""CodeGraph builder — parses repo AST and writes CodeDef/CodeRef graph to Neo4j.""" + +from __future__ import annotations + +import structlog +from neo4j import GraphDatabase + +from .parser import CodeLineParser + +logger = structlog.get_logger() + + +class CodeGraphBuilder: + """Parse a local repo and persist structural graph to Neo4j. + + Args: + neo4j_uri: Bolt URI, e.g. bolt://localhost:7687 + neo4j_password: Neo4j password + repo_root: Absolute or relative path to the repo on disk + repo: Repo identifier string, e.g. "owner/name" (used as isolation key) + neo4j_user: Neo4j username (default: neo4j) + """ + + def __init__( + self, + neo4j_uri: str, + neo4j_password: str, + repo_root: str, + repo: str, + neo4j_user: str = "neo4j", + ) -> None: + self.repo = repo + self.repo_root = repo_root + self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password)) + self.parser = CodeLineParser(repo_root) + + def close(self) -> None: + self.driver.close() + + def build(self) -> dict: + """Clear existing graph for this repo and rebuild from source. + + Returns: + {repo, files, definitions, references} + """ + self._ensure_indexes() + self._clear() + + files = self.parser.find_files() + total_defs = 0 + total_refs = 0 + + with self.driver.session() as session: + for filepath in files: + parsed = self.parser.parse_file(filepath) + rel_file = parsed["file"] + defs = parsed["definitions"] + refs = parsed["references"] + + # Write CodeDef nodes + if defs: + session.run( + """ + UNWIND $defs AS d + MERGE (n:CodeDef {name: d, repo: $repo}) + SET n.file = $file, n.language = $lang, n.updated = timestamp() + """, + defs=defs, + repo=self.repo, + file=rel_file, + lang=parsed.get("language", "unknown"), + ) + total_defs += len(defs) + + # Write CodeRef nodes and INVOKES edges + if defs and refs: + for ref_name in refs: + # For each ref in this file, create an INVOKES edge from + # any def in this file that could call it. + # Simplified: link the first def in this file to the ref target. + # More accurate: all defs in this file may reference it. + session.run( + """ + UNWIND $defs AS caller + MATCH (a:CodeDef {name: caller, repo: $repo}) + MERGE (b:CodeDef {name: $ref, repo: $repo}) + ON CREATE SET b.file = 'unknown', b.language = $lang, + b.updated = timestamp(), b.synthetic = true + MERGE (a)-[:INVOKES]->(b) + """, + defs=defs, + ref=ref_name, + repo=self.repo, + lang=parsed.get("language", "unknown"), + ) + total_refs += len(refs) + + logger.debug( + "codegraph.parsed", + file=rel_file, + defs=len(defs), + refs=len(refs), + ) + + logger.info( + "codegraph.built", + repo=self.repo, + files=len(files), + definitions=total_defs, + references=total_refs, + ) + return { + "repo": self.repo, + "files": len(files), + "definitions": total_defs, + "references": total_refs, + } + + def _clear(self) -> None: + """Delete all CodeDef nodes for this repo only (scoped delete).""" + with self.driver.session() as session: + session.run( + "MATCH (n:CodeDef {repo: $repo}) DETACH DELETE n", + repo=self.repo, + ) + logger.info("codegraph.cleared", repo=self.repo) + + def _ensure_indexes(self) -> None: + """Create indexes if they don't exist.""" + with self.driver.session() as session: + try: + session.run( + "CREATE INDEX codedef_repo_name IF NOT EXISTS " + "FOR (n:CodeDef) ON (n.repo, n.name)" + ) + except Exception as e: + logger.debug("index_creation_skipped", error=str(e)) diff --git a/engine/codegraph/handler.py b/engine/codegraph/handler.py new file mode 100644 index 0000000..a654082 --- /dev/null +++ b/engine/codegraph/handler.py @@ -0,0 +1,105 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [codegraph, handler, api] +# status: active +# --- /L9_META --- +"""CodeGraph action handlers — wired into engine/main.py.""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +import structlog + +from engine.settings import Settings + +logger = structlog.get_logger() +_settings = Settings() + + +async def handle_search_codegraph(payload: dict) -> dict: + """Search CodeGraph for a term in a repo. + + Payload: + term (str, required): Function or class name + repo (str, required): "owner/name" repo identifier + hops (int, optional): 1 or 2 (default: 1) + """ + term = payload.get("term") + repo = payload.get("repo") + if not term or not repo: + return {"error": "search_codegraph requires 'term' and 'repo'"} + + hops = int(payload.get("hops", 1)) + + if not _settings.codegraph_enabled: + return {"error": "CodeGraph engine is disabled (L9_CODEGRAPH_ENABLED=false)"} + + from .retriever import EgoGraphRetriever + + retriever = EgoGraphRetriever( + neo4j_uri=_settings.neo4j_uri, + neo4j_password=_settings.neo4j_password, + neo4j_user=_settings.neo4j_user, + ) + try: + result = retriever.search(term=term, repo=repo, hops=hops) + logger.info("codegraph.search", term=term, repo=repo, hops=hops, nodes=len(result["nodes"])) + return result + finally: + retriever.close() + + +async def handle_build_codegraph(payload: dict) -> dict: + """Clone a GitHub repo and build its CodeGraph. + + Payload: + repo (str, required): "owner/name" format + branch (str, optional): branch to clone (default: main) + """ + repo = payload.get("repo") + if not repo: + return {"error": "build_codegraph requires 'repo' (format: owner/name)"} + + branch = payload.get("branch", "main") + + if not _settings.codegraph_enabled: + return {"error": "CodeGraph engine is disabled (L9_CODEGRAPH_ENABLED=false)"} + + from .builder import CodeGraphBuilder + + token = _settings.github_token + if token: + clone_url = f"https://{token}@github.com/{repo}.git" + else: + clone_url = f"https://github.com/{repo}.git" + + with tempfile.TemporaryDirectory(prefix="l9-codegraph-") as tmpdir: + clone_path = Path(tmpdir) / "repo" + try: + import git + + logger.info("codegraph.clone", repo=repo, branch=branch) + git.Repo.clone_from(clone_url, str(clone_path), branch=branch, depth=1) + except Exception as e: + logger.error("codegraph.clone_failed", repo=repo, error=str(e)) + return {"error": f"Clone failed: {e}"} + + builder = CodeGraphBuilder( + neo4j_uri=_settings.neo4j_uri, + neo4j_password=_settings.neo4j_password, + repo_root=str(clone_path), + repo=repo, + neo4j_user=_settings.neo4j_user, + ) + try: + result = builder.build() + except Exception as e: + logger.error("codegraph.build_failed", repo=repo, error=str(e)) + return {"error": f"Build failed: {e}"} + finally: + builder.close() + + return {"status": "built", **result} diff --git a/engine/codegraph/parser.py b/engine/codegraph/parser.py new file mode 100644 index 0000000..f277589 --- /dev/null +++ b/engine/codegraph/parser.py @@ -0,0 +1,350 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [codegraph, parser, tree-sitter, ast] +# status: active +# --- /L9_META --- +"""Multi-language tree-sitter parser for CodeGraph. + +Supports: .py, .ts, .tsx, .js, .jsx +Extracts: function/class definitions and call references. +Skips: .git, node_modules, __pycache__, dist, build +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any + +import structlog + +logger = structlog.get_logger() + +_SKIP_DIRS = frozenset( + { + ".git", + "node_modules", + "__pycache__", + "dist", + "build", + ".venv", + "venv", + ".mypy_cache", + ".ruff_cache", + ".pytest_cache", + } +) + +_BUILTINS = frozenset( + { + "print", + "len", + "range", + "enumerate", + "zip", + "map", + "filter", + "list", + "dict", + "set", + "tuple", + "str", + "int", + "float", + "bool", + "type", + "isinstance", + "issubclass", + "hasattr", + "getattr", + "setattr", + "delattr", + "callable", + "iter", + "next", + "open", + "input", + "super", + "object", + "property", + "staticmethod", + "classmethod", + "abs", + "all", + "any", + "bin", + "chr", + "dir", + "divmod", + "format", + "frozenset", + "hash", + "hex", + "id", + "max", + "min", + "oct", + "ord", + "pow", + "repr", + "reversed", + "round", + "sorted", + "sum", + "vars", + "NotImplemented", + "Ellipsis", + "None", + "True", + "False", + "Exception", + "ValueError", + "TypeError", + "KeyError", + "IndexError", + "AttributeError", + "RuntimeError", + "StopIteration", + "GeneratorExit", + "ImportError", + "OSError", + "IOError", + "FileNotFoundError", + "PermissionError", + "TimeoutError", + "NotImplementedError", + } +) + +# Lazy language map — populated on first import attempt +LANGUAGE_MAP: dict[str, Any] = {} + + +def _init_languages() -> None: + """Lazily initialize tree-sitter language parsers.""" + global LANGUAGE_MAP + if LANGUAGE_MAP: + return + + try: + import tree_sitter_python as tspython + from tree_sitter import Language + + LANGUAGE_MAP["python"] = Language(tspython.language()) + except Exception as e: + logger.warning("tree-sitter-python unavailable", error=str(e)) + + try: + import tree_sitter_typescript as tstypescript + from tree_sitter import Language + + LANGUAGE_MAP["typescript"] = Language(tstypescript.language_typescript()) + LANGUAGE_MAP["tsx"] = Language(tstypescript.language_tsx()) + except Exception as e: + logger.warning("tree-sitter-typescript unavailable", error=str(e)) + + try: + import tree_sitter_javascript as tsjavascript + from tree_sitter import Language + + LANGUAGE_MAP["javascript"] = Language(tsjavascript.language()) + except Exception as e: + logger.warning("tree-sitter-javascript unavailable", error=str(e)) + + +_EXT_TO_LANG: dict[str, str] = { + ".py": "python", + ".ts": "typescript", + ".tsx": "tsx", + ".js": "javascript", + ".jsx": "javascript", +} + + +def _lang_for_file(path: Path) -> str | None: + return _EXT_TO_LANG.get(path.suffix.lower()) + + +def _extract_python_regex(source: str) -> tuple[list[str], list[str]]: + """Fallback regex extraction for Python when tree-sitter is unavailable.""" + defs: list[str] = [] + refs: list[str] = [] + def_pattern = re.compile(r"^(?:async\s+)?def\s+(\w+)|^class\s+(\w+)", re.MULTILINE) + call_pattern = re.compile(r"\b(\w+)\s*\(") + for m in def_pattern.finditer(source): + name = m.group(1) or m.group(2) + if name: + defs.append(name) + def_set = set(defs) + for m in call_pattern.finditer(source): + name = m.group(1) + if name and name not in _BUILTINS and name not in def_set: + refs.append(name) + return defs, refs + + +class CodeLineParser: + """Parse source files and extract CodeDef + CodeRef data.""" + + def __init__(self, repo_root: str) -> None: + self.repo_root = Path(repo_root).resolve() + _init_languages() + + def find_files(self) -> list[Path]: + """Walk repo_root and return all parseable source files.""" + results: list[Path] = [] + for path in self.repo_root.rglob("*"): + if not path.is_file(): + continue + # Skip ignored dirs + parts = path.relative_to(self.repo_root).parts + if any(part in _SKIP_DIRS for part in parts): + continue + if _lang_for_file(path) is not None: + results.append(path) + return results + + def parse_file(self, filepath: Path) -> dict[str, Any]: + """Parse a single file. Returns {file, language, definitions, references}.""" + lang_key = _lang_for_file(filepath) + if lang_key is None: + return {"file": str(filepath), "language": None, "definitions": [], "references": []} + + rel_path = str(filepath.relative_to(self.repo_root)) + + try: + source = filepath.read_text(encoding="utf-8", errors="replace") + except OSError as e: + logger.warning("read_error", file=rel_path, error=str(e)) + return {"file": rel_path, "language": lang_key, "definitions": [], "references": []} + + # Use tree-sitter if available for this language + if lang_key in LANGUAGE_MAP: + defs, refs = self._parse_with_treesitter(source, lang_key, filepath) + elif lang_key == "python": + defs, refs = _extract_python_regex(source) + else: + defs, refs = [], [] + + return { + "file": rel_path, + "language": lang_key, + "definitions": defs, + "references": refs, + } + + def _parse_with_treesitter( + self, source: str, lang_key: str, filepath: Path + ) -> tuple[list[str], list[str]]: + """Use tree-sitter AST to extract definitions and call references.""" + try: + from tree_sitter import Parser + + language = LANGUAGE_MAP[lang_key] + parser = Parser(language) + tree = parser.parse(source.encode("utf-8", errors="replace")) + root = tree.root_node + + defs: list[str] = [] + refs: list[str] = [] + def_set: set[str] = set() + + self._walk(root, source, defs, refs, def_set, lang_key) + + # Deduplicate, filter builtins from refs + unique_defs = list(dict.fromkeys(defs)) + unique_refs = list( + dict.fromkeys(r for r in refs if r not in _BUILTINS and r not in def_set) + ) + return unique_defs, unique_refs + except Exception as e: + logger.warning("treesitter_parse_error", file=str(filepath), error=str(e)) + if lang_key == "python": + return _extract_python_regex(source) + return [], [] + + def _walk( + self, + node: Any, + source: str, + defs: list[str], + refs: list[str], + def_set: set[str], + lang_key: str, + ) -> None: + """Recursively walk AST, collect definitions and call references.""" + self._collect_def(node, source, defs, def_set) + self._collect_ref(node, source, refs) + for child in node.children: + self._walk(child, source, defs, refs, def_set, lang_key) + + def _collect_def(self, node: Any, source: str, defs: list[str], def_set: set[str]) -> None: + """Extract definition name from AST node if applicable.""" + node_type = node.type + named_def_types = { + "function_definition", + "async_function_definition", + "class_definition", + "function_declaration", + "generator_function_declaration", + "method_definition", + "arrow_function", + "class_declaration", + } + if node_type in named_def_types: + name_node = node.child_by_field_name("name") + if name_node: + name = source[name_node.start_byte : name_node.end_byte] + defs.append(name) + def_set.add(name) + elif node_type == "lexical_declaration": + self._collect_lexical_def(node, source, defs, def_set) + + def _collect_lexical_def( + self, node: Any, source: str, defs: list[str], def_set: set[str] + ) -> None: + """Handle const foo = () => {} style declarations.""" + for child in node.children: + if child.type == "variable_declarator": + name_node = child.child_by_field_name("name") + val_node = child.child_by_field_name("value") + if ( + name_node + and val_node + and val_node.type + in ( + "arrow_function", + "function", + "function_expression", + ) + ): + name = source[name_node.start_byte : name_node.end_byte] + defs.append(name) + def_set.add(name) + + def _collect_ref(self, node: Any, source: str, refs: list[str]) -> None: + """Extract call reference from AST node if applicable.""" + node_type = node.type + if node_type in ("call", "call_expression"): + func_node = node.child_by_field_name("function") + if func_node: + name = self._extract_call_name(func_node, source) + if name: + refs.append(name) + + def _extract_call_name(self, node: Any, source: str) -> str | None: + """Extract a clean name from a call's function node.""" + node_type = node.type + if node_type == "identifier": + return source[node.start_byte : node.end_byte] + if node_type == "attribute": + # foo.bar(...) — return just "bar" + attr = node.child_by_field_name("attribute") + if attr: + return source[attr.start_byte : attr.end_byte] + if node_type == "member_expression": + prop = node.child_by_field_name("property") + if prop: + return source[prop.start_byte : prop.end_byte] + return None diff --git a/engine/codegraph/retriever.py b/engine/codegraph/retriever.py new file mode 100644 index 0000000..a096bbb --- /dev/null +++ b/engine/codegraph/retriever.py @@ -0,0 +1,110 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [codegraph, retriever, ego-graph] +# status: active +# --- /L9_META --- +"""EgoGraphRetriever — pure Cypher, no APOC. + +Returns ego-graphs (1–2 hop neighborhoods) for a given term + repo. +""" + +from __future__ import annotations + +import structlog +from neo4j import GraphDatabase + +logger = structlog.get_logger() + + +class EgoGraphRetriever: + """Retrieve ego-graphs from CodeDef/INVOKES graph. + + Args: + neo4j_uri: Bolt URI + neo4j_password: Neo4j password + neo4j_user: Neo4j username (default: neo4j) + """ + + def __init__( + self, + neo4j_uri: str, + neo4j_password: str, + neo4j_user: str = "neo4j", + ) -> None: + self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password)) + + def close(self) -> None: + self.driver.close() + + def search(self, term: str, repo: str, hops: int = 1) -> dict: + """Return ego-graph for `term` in `repo` with up to `hops` traversals. + + Args: + term: Function or class name to look up + repo: Repo identifier (e.g. "owner/name") + hops: 1 or 2 (clamped) + + Returns: + {nodes, edges, flat_text} + """ + hops = min(max(hops, 1), 2) + with self.driver.session() as s: + center = s.run( + "MATCH (c:CodeDef {name: $t, repo: $r}) RETURN c LIMIT 1", + t=term, + r=repo, + ).single() + if not center: + return { + "nodes": [], + "edges": [], + "flat_text": f"No results for '{term}' in '{repo}'", + } + center_node = dict(center["c"]) + + neighbors_result = s.run( + "MATCH (c:CodeDef {name: $t, repo: $r})-[:INVOKES*1..$h]-(n:CodeDef {repo: $r})" + " RETURN DISTINCT n", + t=term, + r=repo, + h=hops, + ) + neighbor_nodes = [dict(r["n"]) for r in neighbors_result] + + all_names = {center_node["name"]} | {n["name"] for n in neighbor_nodes} + edges_result = s.run( + "MATCH (a:CodeDef {repo: $r})-[:INVOKES]->(b:CodeDef {repo: $r})" + " WHERE a.name IN $names AND b.name IN $names" + " RETURN DISTINCT a.name AS fn, b.name AS tn", + r=repo, + names=list(all_names), + ) + edge_list = [{"from": e["fn"], "to": e["tn"], "type": "INVOKES"} for e in edges_result] + + nodes = [center_node] + neighbor_nodes + return { + "nodes": nodes, + "edges": edge_list, + "flat_text": self._flatten(nodes, edge_list, term, repo), + } + + def _flatten( + self, + nodes: list[dict], + edges: list[dict], + term: str, + repo: str, + ) -> str: + """Produce a token-efficient text representation of the ego-graph.""" + lines = [f"# CodeGraph ego-graph: '{term}' in '{repo}'"] + lines.append(f"Nodes ({len(nodes)}):") + for n in nodes: + marker = " [CENTER]" if n.get("name") == term else "" + lang = n.get("language", "?") + file_ = n.get("file", "?") + lines.append(f" - {n['name']} ({lang}) @ {file_}{marker}") + lines.append(f"Edges ({len(edges)}):") + for e in edges: + lines.append(f" - {e['from']} -> {e['to']}") + return "\n".join(lines) diff --git a/engine/main.py b/engine/main.py index 83820db..acfadd8 100644 --- a/engine/main.py +++ b/engine/main.py @@ -4,35 +4,8 @@ # tags: [fastapi, chassis, entrypoint] # status: active # --- /L9_META --- -"""L9 Golden Repo — FastAPI entrypoint. Replace APP_NAME and wire your engine.""" -from __future__ import annotations +"""l9-codegraph — chassis entrypoint. Single ingress via PacketEnvelope.""" +from chassis.chassis_app import create_app +from engine.boot import CodegraphLifecycle -import structlog -from fastapi import FastAPI -from fastapi.responses import JSONResponse - -from engine.settings import Settings - -logger = structlog.get_logger() -settings = Settings() - -app = FastAPI( - title=settings.app_name, - version="0.1.0", - docs_url="/docs" if settings.debug else None, -) - - -@app.get("/health") -async def health() -> JSONResponse: - return JSONResponse({"status": "ok", "service": settings.app_name}) - - -@app.post("/v1/execute") -async def execute(payload: dict) -> JSONResponse: - """Primary action endpoint. action + tenant + payload envelope.""" - action = payload.get("action") - tenant = payload.get("tenant") - logger.info("execute", action=action, tenant=tenant) - # TODO: route to your engine handler - return JSONResponse({"status": "ok", "action": action, "tenant": tenant}) +app = create_app(lifecycle_hook=CodegraphLifecycle()) diff --git a/engine/plangraph/__init__.py b/engine/plangraph/__init__.py new file mode 100644 index 0000000..2a8cdb2 --- /dev/null +++ b/engine/plangraph/__init__.py @@ -0,0 +1,13 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, rpg, planning] +# status: active +# --- /L9_META --- +"""Plan Graph Engine — RPG-inspired service constellation planning.""" + +from .builder import PlanGraphBuilder +from .drift import DriftDetector +from .retriever import PlanGraphRetriever + +__all__ = ["PlanGraphBuilder", "PlanGraphRetriever", "DriftDetector"] diff --git a/engine/plangraph/builder.py b/engine/plangraph/builder.py new file mode 100644 index 0000000..176232c --- /dev/null +++ b/engine/plangraph/builder.py @@ -0,0 +1,167 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, builder, neo4j] +# status: active +# --- /L9_META --- +"""PlanGraph builder — writes constellation service graph to Neo4j.""" + +from __future__ import annotations + +import structlog +from neo4j import GraphDatabase + +logger = structlog.get_logger() + + +class PlanGraphBuilder: + """Build a service constellation graph in Neo4j. + + Args: + neo4j_uri: Bolt URI + neo4j_password: Neo4j password + constellation: Constellation name (isolation key) + neo4j_user: Neo4j username (default: neo4j) + """ + + def __init__( + self, + neo4j_uri: str, + neo4j_password: str, + constellation: str, + neo4j_user: str = "neo4j", + ) -> None: + self.constellation = constellation + self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password)) + + def close(self) -> None: + self.driver.close() + + def load(self, spec: dict) -> dict: + """Load a parsed spec dict into Neo4j. + + Args: + spec: {services, interfaces, flows} from SpecParser.parse() + + Returns: + {constellation, services, interfaces, flows} + """ + self._ensure_indexes() + self._clear() + + services = spec.get("services", []) + interfaces = spec.get("interfaces", []) + flows = spec.get("flows", []) + + with self.driver.session() as session: + # Write PlanService nodes + for svc in services: + session.run( + """ + MERGE (s:PlanService {name: $name, constellation: $c}) + SET s.status = $status, + s.description = $desc, + s.updated = timestamp() + """, + name=svc["name"], + c=self.constellation, + status=svc.get("status", "planned"), + desc=svc.get("description", ""), + ) + + # Write DEPENDS_ON edges + for svc in services: + for dep in svc.get("depends_on", []): + session.run( + """ + MATCH (a:PlanService {name: $name, constellation: $c}) + MERGE (b:PlanService {name: $dep, constellation: $c}) + ON CREATE SET b.status = 'planned', b.updated = timestamp() + MERGE (a)-[:DEPENDS_ON]->(b) + """, + name=svc["name"], + dep=dep, + c=self.constellation, + ) + + # Write PlanInterface nodes + EXPOSES edges + for iface in interfaces: + session.run( + """ + MERGE (i:PlanInterface {name: $iname, constellation: $c}) + SET i.direction = $direction, + i.protocol = $protocol, + i.description = $desc, + i.updated = timestamp() + WITH i + MATCH (s:PlanService {name: $svc, constellation: $c}) + MERGE (s)-[:EXPOSES]->(i) + """, + iname=iface["name"], + c=self.constellation, + direction=iface.get("direction", "inbound"), + protocol=iface.get("protocol", "http"), + desc=iface.get("description", ""), + svc=iface.get("service", ""), + ) + + # Write FLOWS_TO / FEEDS_BACK_TO edges + for flow in flows: + from_svc = flow.get("from", "") + to_svc = flow.get("to", "") + label = flow.get("label", "") + feedback = flow.get("feedback", False) + rel_type = "FEEDS_BACK_TO" if feedback else "FLOWS_TO" + session.run( + f""" + MATCH (a:PlanService {{name: $from_s, constellation: $c}}) + MATCH (b:PlanService {{name: $to_s, constellation: $c}}) + MERGE (a)-[r:{rel_type}]->(b) + SET r.label = $label + """, + from_s=from_svc, + to_s=to_svc, + c=self.constellation, + label=label, + ) + + logger.info( + "plangraph.loaded", + constellation=self.constellation, + services=len(services), + interfaces=len(interfaces), + flows=len(flows), + ) + return { + "constellation": self.constellation, + "services": len(services), + "interfaces": len(interfaces), + "flows": len(flows), + } + + def _clear(self) -> None: + """Delete all nodes for this constellation only.""" + with self.driver.session() as session: + session.run( + "MATCH (n:PlanService {constellation: $c}) DETACH DELETE n", + c=self.constellation, + ) + session.run( + "MATCH (n:PlanInterface {constellation: $c}) DETACH DELETE n", + c=self.constellation, + ) + logger.info("plangraph.cleared", constellation=self.constellation) + + def _ensure_indexes(self) -> None: + with self.driver.session() as session: + try: + session.run( + "CREATE INDEX planservice_constellation IF NOT EXISTS " + "FOR (n:PlanService) ON (n.constellation, n.name)" + ) + session.run( + "CREATE INDEX planiface_constellation IF NOT EXISTS " + "FOR (n:PlanInterface) ON (n.constellation, n.name)" + ) + except Exception as e: + logger.debug("plangraph_index_skip", error=str(e)) diff --git a/engine/plangraph/drift.py b/engine/plangraph/drift.py new file mode 100644 index 0000000..67ee49b --- /dev/null +++ b/engine/plangraph/drift.py @@ -0,0 +1,161 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, drift, validation] +# status: active +# --- /L9_META --- +"""DriftDetector — compares planned service graph vs actual CodeDef nodes.""" + +from __future__ import annotations + +import structlog +from neo4j import GraphDatabase + +logger = structlog.get_logger() + + +class DriftDetector: + """Detect drift between planned constellation and implemented code. + + Args: + neo4j_uri: Bolt URI + neo4j_password: Neo4j password + neo4j_user: Neo4j username (default: neo4j) + """ + + def __init__( + self, + neo4j_uri: str, + neo4j_password: str, + neo4j_user: str = "neo4j", + ) -> None: + self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password)) + + def close(self) -> None: + self.driver.close() + + def check_service(self, service: str, constellation: str, repo: str) -> dict: + """Check drift for a single service. + + Uses explicit `repo` param for CodeDef lookup — no fragile filename inference. + + Args: + service: Service name in the constellation + constellation: Constellation identifier + repo: Repo identifier for CodeDef lookup (e.g. "owner/name") + + Returns: + DriftResult-like dict + """ + with self.driver.session() as s: + # Get planned status + svc_rec = s.run( + "MATCH (n:PlanService {name: $svc, constellation: $c}) RETURN n LIMIT 1", + svc=service, + c=constellation, + ).single() + + if not svc_rec: + return { + "service": service, + "constellation": constellation, + "repo": repo, + "planned_status": "unknown", + "implemented_functions": [], + "is_implemented": False, + "drift_score": 1.0, + "notes": f"Service '{service}' not found in constellation '{constellation}'", + } + + svc_node = dict(svc_rec["n"]) + planned_status = svc_node.get("status", "planned") + + # Look for CodeDef nodes in the repo that match this service name + # Strategy: look for defs whose name contains the service name (case-insensitive) + # and who belong to the specified repo + service_lower = service.lower() + code_rec = s.run( + """ + MATCH (n:CodeDef {repo: $repo}) + WHERE toLower(n.name) CONTAINS $svc_lower + OR toLower(n.file) CONTAINS $svc_lower + RETURN n.name AS name, n.file AS file + LIMIT 50 + """, + repo=repo, + svc_lower=service_lower, + ) + implemented_functions = [r["name"] for r in code_rec] + is_implemented = len(implemented_functions) > 0 + + # Drift score: 0.0 = perfect alignment, 1.0 = total drift + if planned_status in ("built", "deployed"): + drift_score = 0.0 if is_implemented else 0.8 + elif planned_status == "in_progress": + drift_score = 0.0 if is_implemented else 0.4 + else: + # planned — not implemented yet is expected + drift_score = 0.0 + + notes = "" + if planned_status in ("built", "deployed") and not is_implemented: + notes = ( + f"Service marked '{planned_status}' but no matching " + f"CodeDef found in repo '{repo}'" + ) + elif is_implemented: + notes = f"Found {len(implemented_functions)} matching function(s) in repo" + + return { + "service": service, + "constellation": constellation, + "repo": repo, + "planned_status": planned_status, + "implemented_functions": implemented_functions, + "is_implemented": is_implemented, + "drift_score": drift_score, + "notes": notes, + } + + def check_all(self, constellation: str, repo: str) -> dict: + """Check drift for all services in a constellation. + + Args: + constellation: Constellation identifier + repo: Repo identifier for CodeDef lookup + + Returns: + {constellation, repo, results: [...], summary: {...}} + """ + with self.driver.session() as s: + services_rec = s.run( + "MATCH (n:PlanService {constellation: $c}) RETURN n.name AS name", + c=constellation, + ) + service_names = [r["name"] for r in services_rec] + + results = [self.check_service(svc, constellation, repo) for svc in service_names] + + drifted = [r for r in results if r["drift_score"] > 0.0] + avg_drift = sum(r["drift_score"] for r in results) / len(results) if results else 0.0 + + logger.info( + "plangraph.drift_check", + constellation=constellation, + repo=repo, + services=len(results), + drifted=len(drifted), + avg_drift=round(avg_drift, 3), + ) + + return { + "constellation": constellation, + "repo": repo, + "results": results, + "summary": { + "total": len(results), + "drifted": len(drifted), + "clean": len(results) - len(drifted), + "avg_drift_score": round(avg_drift, 3), + }, + } diff --git a/engine/plangraph/handler.py b/engine/plangraph/handler.py new file mode 100644 index 0000000..73d55c9 --- /dev/null +++ b/engine/plangraph/handler.py @@ -0,0 +1,154 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, handler, api] +# status: active +# --- /L9_META --- +"""PlanGraph action handlers — wired into engine/main.py.""" + +from __future__ import annotations + +from pathlib import Path + +import structlog + +from engine.settings import Settings + +logger = structlog.get_logger() +_settings = Settings() + +# Domains directory (relative to repo root) +_DOMAINS_DIR = Path(__file__).parent.parent.parent / "domains" + + +async def handle_search_plangraph(payload: dict) -> dict: + """Search PlanGraph for a service in a constellation. + + Payload: + service (str, required): Service name + constellation (str, required): Constellation identifier + """ + service = payload.get("service") + constellation = payload.get("constellation") + if not service or not constellation: + return {"error": "search_plangraph requires 'service' and 'constellation'"} + + if not _settings.plangraph_enabled: + return {"error": "PlanGraph engine is disabled (L9_PLANGRAPH_ENABLED=false)"} + + from .retriever import PlanGraphRetriever + + retriever = PlanGraphRetriever( + neo4j_uri=_settings.neo4j_uri, + neo4j_password=_settings.neo4j_password, + neo4j_user=_settings.neo4j_user, + ) + try: + result = retriever.search_service(service=service, constellation=constellation) + logger.info("plangraph.search", service=service, constellation=constellation) + return result + finally: + retriever.close() + + +async def handle_build_order(payload: dict) -> dict: + """Get topological build order for a constellation. + + Payload: + constellation (str, required): Constellation identifier + """ + constellation = payload.get("constellation") + if not constellation: + return {"error": "build_order requires 'constellation'"} + + if not _settings.plangraph_enabled: + return {"error": "PlanGraph engine is disabled (L9_PLANGRAPH_ENABLED=false)"} + + from .retriever import PlanGraphRetriever + + retriever = PlanGraphRetriever( + neo4j_uri=_settings.neo4j_uri, + neo4j_password=_settings.neo4j_password, + neo4j_user=_settings.neo4j_user, + ) + try: + result = retriever.build_order(constellation=constellation) + logger.info("plangraph.build_order", constellation=constellation) + return result + finally: + retriever.close() + + +async def handle_check_drift(payload: dict) -> dict: + """Check drift between planned constellation and implemented code. + + Payload: + constellation (str, required): Constellation identifier + repo (str, required): Repo identifier (owner/name) + service (str, optional): Single service (omit for all services) + """ + constellation = payload.get("constellation") + repo = payload.get("repo") + if not constellation or not repo: + return {"error": "check_drift requires 'constellation' and 'repo'"} + + if not _settings.plangraph_enabled: + return {"error": "PlanGraph engine is disabled (L9_PLANGRAPH_ENABLED=false)"} + + from .drift import DriftDetector + + detector = DriftDetector( + neo4j_uri=_settings.neo4j_uri, + neo4j_password=_settings.neo4j_password, + neo4j_user=_settings.neo4j_user, + ) + try: + service = payload.get("service") + if service: + result = detector.check_service(service=service, constellation=constellation, repo=repo) + else: + result = detector.check_all(constellation=constellation, repo=repo) + return result + finally: + detector.close() + + +async def handle_load_constellation(payload: dict) -> dict: + """Load a constellation spec from a YAML file into Neo4j. + + Payload: + constellation (str, required): Constellation identifier + spec_file (str, required): YAML filename (relative to domains//) + spec_dir (str, optional): Override directory path + """ + constellation = payload.get("constellation") + spec_file = payload.get("spec_file") + if not constellation or not spec_file: + return {"error": "load_constellation requires 'constellation' and 'spec_file'"} + + if not _settings.plangraph_enabled: + return {"error": "PlanGraph engine is disabled (L9_PLANGRAPH_ENABLED=false)"} + + spec_dir = payload.get("spec_dir") or str(_DOMAINS_DIR / constellation) + + from .builder import PlanGraphBuilder + from .spec_parser import SpecParser + + parser = SpecParser(spec_dir=spec_dir) + spec = parser.parse(spec_file) + + if not spec.get("services"): + return {"error": f"No services found in spec '{spec_file}' at '{spec_dir}'"} + + builder = PlanGraphBuilder( + neo4j_uri=_settings.neo4j_uri, + neo4j_password=_settings.neo4j_password, + constellation=constellation, + neo4j_user=_settings.neo4j_user, + ) + try: + result = builder.load(spec) + logger.info("plangraph.loaded", constellation=constellation, spec_file=spec_file) + return {"status": "loaded", **result} + finally: + builder.close() diff --git a/engine/plangraph/retriever.py b/engine/plangraph/retriever.py new file mode 100644 index 0000000..67cbd4e --- /dev/null +++ b/engine/plangraph/retriever.py @@ -0,0 +1,188 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, retriever, topology] +# status: active +# --- /L9_META --- +"""PlanGraph retriever — service search and topological build order. + +Pure Cypher only, no APOC. +""" + +from __future__ import annotations + +from collections import deque + +import structlog +from neo4j import GraphDatabase + +logger = structlog.get_logger() + + +class PlanGraphRetriever: + """Query PlanGraph for service neighborhoods and build order. + + Args: + neo4j_uri: Bolt URI + neo4j_password: Neo4j password + neo4j_user: Neo4j username (default: neo4j) + """ + + def __init__( + self, + neo4j_uri: str, + neo4j_password: str, + neo4j_user: str = "neo4j", + ) -> None: + self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password)) + + def close(self) -> None: + self.driver.close() + + def search_service(self, service: str, constellation: str) -> dict: + """Return service node + neighbors (DEPENDS_ON, FLOWS_TO, FEEDS_BACK_TO). + + Args: + service: Service name + constellation: Constellation identifier + + Returns: + {service, nodes, edges, flat_text} + """ + with self.driver.session() as s: + center_rec = s.run( + "MATCH (n:PlanService {name: $svc, constellation: $c}) RETURN n LIMIT 1", + svc=service, + c=constellation, + ).single() + if not center_rec: + msg = f"Service '{service}' not found in constellation '{constellation}'" + return {"service": service, "nodes": [], "edges": [], "flat_text": msg} + center = dict(center_rec["n"]) + + dep_result = s.run( + """ + MATCH (a:PlanService {name: $svc, constellation: $c}) + -[r:DEPENDS_ON|FLOWS_TO|FEEDS_BACK_TO]- + (b:PlanService {constellation: $c}) + RETURN b, type(r) AS rel_type, + startNode(r).name AS rel_from, + endNode(r).name AS rel_to + """, + svc=service, + c=constellation, + ) + neighbor_nodes = [] + edge_list = [] + seen_names: set[str] = {center["name"]} + for rec in dep_result: + n = dict(rec["b"]) + if n["name"] not in seen_names: + neighbor_nodes.append(n) + seen_names.add(n["name"]) + edge_list.append( + { + "from": rec["rel_from"], + "to": rec["rel_to"], + "type": rec["rel_type"], + } + ) + + nodes = [center] + neighbor_nodes + flat_text = self._flatten_service(center, nodes, edge_list) + return {"service": service, "nodes": nodes, "edges": edge_list, "flat_text": flat_text} + + def build_order(self, constellation: str) -> dict: + """Compute topological build order for a constellation. + + Returns: + {constellation, topological_order, parallel_groups, flat_text} + """ + with self.driver.session() as s: + services_rec = s.run( + "MATCH (n:PlanService {constellation: $c}) RETURN n", + c=constellation, + ) + services = [dict(r["n"]) for r in services_rec] + + deps_rec = s.run( + """ + MATCH (a:PlanService {constellation: $c}) + -[:DEPENDS_ON]-> + (b:PlanService {constellation: $c}) + RETURN a.name AS dependent, b.name AS dependency + """, + c=constellation, + ) + dep_map: dict[str, list[str]] = {svc["name"]: [] for svc in services} + for rec in deps_rec: + dep_map[rec["dependent"]].append(rec["dependency"]) + + topo, groups = self._topological_sort(dep_map) + flat_text = self._flatten_build_order(constellation, topo, groups) + return { + "constellation": constellation, + "topological_order": topo, + "parallel_groups": groups, + "flat_text": flat_text, + } + + def _topological_sort(self, dep_map: dict[str, list[str]]) -> tuple[list[str], list[list[str]]]: + """Kahn's algorithm — returns (flat order, parallel groups). + + dep_map: {service_name: [dependencies]} + """ + # in_degree = number of unresolved dependencies per node + in_degree = {n: len(deps) for n, deps in dep_map.items()} + dependents = self._build_dependents(dep_map) + queue: deque[str] = deque(sorted(n for n, d in in_degree.items() if d == 0)) + topo: list[str] = [] + groups: list[list[str]] = [] + + while queue: + level = list(queue) + queue.clear() + groups.append(sorted(level)) + topo.extend(sorted(level)) + for node in level: + for dep_node in dependents.get(node, []): + in_degree[dep_node] -= 1 + if in_degree[dep_node] == 0: + queue.append(dep_node) + + remaining = sorted(n for n in in_degree if n not in topo) + if remaining: + logger.warning("plangraph.cycle_detected", nodes=remaining) + topo.extend(remaining) + groups.append(remaining) + + return topo, groups + + def _build_dependents(self, dep_map: dict[str, list[str]]) -> dict[str, list[str]]: + """Build reverse mapping: node -> list of nodes that depend on it.""" + dependents: dict[str, list[str]] = {n: [] for n in dep_map} + for node, deps in dep_map.items(): + for dep in deps: + if dep not in dependents: + dependents[dep] = [] + dependents[dep].append(node) + return dependents + + def _flatten_service(self, center: dict, nodes: list[dict], edges: list[dict]) -> str: + lines = [f"# PlanGraph: service '{center['name']}' ({center.get('status', '?')})"] + lines.append(f"Neighbors ({len(nodes) - 1}):") + for n in nodes[1:]: + lines.append(f" - {n['name']} [{n.get('status', '?')}]") + lines.append(f"Edges ({len(edges)}):") + for e in edges: + lines.append(f" - {e['from']} -[{e['type']}]-> {e['to']}") + return "\n".join(lines) + + def _flatten_build_order( + self, constellation: str, topo: list[str], groups: list[list[str]] + ) -> str: + lines = [f"# Build order for constellation '{constellation}'"] + for i, group in enumerate(groups, 1): + lines.append(f"Wave {i} (parallel): {', '.join(group)}") + lines.append(f"Full order: {' -> '.join(topo)}") + return "\n".join(lines) diff --git a/engine/plangraph/schema.py b/engine/plangraph/schema.py new file mode 100644 index 0000000..1911a64 --- /dev/null +++ b/engine/plangraph/schema.py @@ -0,0 +1,63 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, schema, pydantic] +# status: active +# --- /L9_META --- +"""PlanGraph Pydantic v2 models.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, Field + + +class ServiceStatus(StrEnum): + planned = "planned" + in_progress = "in_progress" + built = "built" + deployed = "deployed" + + +class InterfaceDirection(StrEnum): + inbound = "inbound" + outbound = "outbound" + + +class PlanServiceNode(BaseModel): + name: str + constellation: str + status: ServiceStatus = ServiceStatus.planned + description: str = "" + depends_on: list[str] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PlanInterfaceNode(BaseModel): + name: str + constellation: str + service: str + direction: InterfaceDirection + protocol: str = "http" + description: str = "" + + +class PlanDataFlowEdge(BaseModel): + from_service: str + to_service: str + constellation: str + label: str = "" + feedback: bool = False + + +class DriftResult(BaseModel): + service: str + constellation: str + repo: str + planned_status: ServiceStatus + implemented_functions: list[str] = Field(default_factory=list) + is_implemented: bool = False + drift_score: float = 0.0 + notes: str = "" diff --git a/engine/plangraph/spec_parser.py b/engine/plangraph/spec_parser.py new file mode 100644 index 0000000..e84bbbf --- /dev/null +++ b/engine/plangraph/spec_parser.py @@ -0,0 +1,131 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [engine] +# tags: [plangraph, spec, yaml, parser] +# status: active +# --- /L9_META --- +"""SpecParser — reads constellation YAML specs into normalized dicts.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import structlog +import yaml + +logger = structlog.get_logger() + + +class SpecParser: + """Parse constellation YAML specification files. + + Args: + spec_dir: Directory containing spec YAML files + """ + + def __init__(self, spec_dir: str) -> None: + self.spec_dir = Path(spec_dir) + + def parse(self, filename: str) -> dict[str, Any]: + """Parse a YAML spec file. + + Returns: + {services: [...], interfaces: [...], flows: [...]} + """ + path = self.spec_dir / filename + try: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) + except FileNotFoundError: + logger.error("spec_not_found", path=str(path)) + return {"services": [], "interfaces": [], "flows": []} + except yaml.YAMLError as e: + logger.error("spec_parse_error", path=str(path), error=str(e)) + return {"services": [], "interfaces": [], "flows": []} + + services = self._extract_services(raw) + interfaces = self._extract_interfaces(raw) + flows = self._extract_flows(raw) + + return {"services": services, "interfaces": interfaces, "flows": flows} + + def _extract_services(self, raw: dict) -> list[dict]: + """Extract service definitions from raw YAML.""" + raw_services = raw.get("services", {}) + if isinstance(raw_services, dict): + result = [] + for name, data in raw_services.items(): + if data is None: + data = {} + svc = { + "name": name, + "status": data.get("status", "planned"), + "description": data.get("description", ""), + "depends_on": data.get("depends_on", []), + "metadata": data.get("metadata", {}), + } + result.append(svc) + return result + if isinstance(raw_services, list): + return [ + { + "name": s.get("name", ""), + "status": s.get("status", "planned"), + "description": s.get("description", ""), + "depends_on": s.get("depends_on", []), + "metadata": s.get("metadata", {}), + } + for s in raw_services + if isinstance(s, dict) + ] + return [] + + def _extract_interfaces(self, raw: dict) -> list[dict]: + """Extract interface definitions from raw YAML.""" + raw_ifaces = raw.get("interfaces", []) + if isinstance(raw_ifaces, list): + return [ + { + "name": i.get("name", ""), + "service": i.get("service", ""), + "direction": i.get("direction", "inbound"), + "protocol": i.get("protocol", "http"), + "description": i.get("description", ""), + } + for i in raw_ifaces + if isinstance(i, dict) + ] + # Dict keyed by service name + if isinstance(raw_ifaces, dict): + result = [] + for svc_name, iface_list in raw_ifaces.items(): + if not isinstance(iface_list, list): + continue + for iface in iface_list: + result.append( + { + "name": iface.get("name", ""), + "service": svc_name, + "direction": iface.get("direction", "inbound"), + "protocol": iface.get("protocol", "http"), + "description": iface.get("description", ""), + } + ) + return result + return [] + + def _extract_flows(self, raw: dict) -> list[dict]: + """Extract data flow edges from raw YAML.""" + raw_flows = raw.get("flows", []) + if not isinstance(raw_flows, list): + return [] + return [ + { + "from": f.get("from", ""), + "to": f.get("to", ""), + "label": f.get("label", ""), + "feedback": f.get("feedback", False), + } + for f in raw_flows + if isinstance(f, dict) + ] diff --git a/engine/settings.py b/engine/settings.py index 155cabf..ac1ca03 100644 --- a/engine/settings.py +++ b/engine/settings.py @@ -1,96 +1,29 @@ -""" -CRM AI Platform — Central Configuration -All environment-driven. No secrets in code. -""" -import os -from dataclasses import dataclass, field -from typing import Optional - - -@dataclass(frozen=True) -class DatabaseConfig: - host: str = os.getenv("CRM_DB_HOST", "localhost") - port: int = int(os.getenv("CRM_DB_PORT", "5432")) - name: str = os.getenv("CRM_DB_NAME", "crm_ai") - user: str = os.getenv("CRM_DB_USER", "crm_service") - password: str = os.getenv("CRM_DB_PASSWORD", "") - pool_min: int = int(os.getenv("CRM_DB_POOL_MIN", "2")) - pool_max: int = int(os.getenv("CRM_DB_POOL_MAX", "10")) - statement_timeout_ms: int = int(os.getenv("CRM_DB_TIMEOUT_MS", "30000")) - - @property - def dsn(self) -> str: - return f"postgresql://{self.user}:{self.password}@{self.host}:{self.port}/{self.name}" - - -@dataclass(frozen=True) -class GoogleAdsConfig: - developer_token: str = os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", "") - client_id: str = os.getenv("GOOGLE_ADS_CLIENT_ID", "") - client_secret: str = os.getenv("GOOGLE_ADS_CLIENT_SECRET", "") - refresh_token: str = os.getenv("GOOGLE_ADS_REFRESH_TOKEN", "") - login_customer_id: str = os.getenv("GOOGLE_ADS_LOGIN_CUSTOMER_ID", "") - customer_id: str = os.getenv("GOOGLE_ADS_CUSTOMER_ID", "") - conversion_action_id: str = os.getenv("GOOGLE_ADS_CONVERSION_ACTION_ID", "") - api_version: str = os.getenv("GOOGLE_ADS_API_VERSION", "v17") - +# --- L9_META --- +# l9_schema: 1 +# layer: [config] +# tags: [settings, config, neo4j] +# status: active +# --- /L9_META --- +"""l9-codegraph settings — all env-driven.""" -@dataclass(frozen=True) -class SalesforceConfig: - instance_url: str = os.getenv("SF_INSTANCE_URL", "") - client_id: str = os.getenv("SF_CLIENT_ID", "") - client_secret: str = os.getenv("SF_CLIENT_SECRET", "") - username: str = os.getenv("SF_USERNAME", "") - password: str = os.getenv("SF_PASSWORD", "") - security_token: str = os.getenv("SF_SECURITY_TOKEN", "") - api_version: str = os.getenv("SF_API_VERSION", "v60.0") - data_cloud_endpoint: str = os.getenv("SF_DATA_CLOUD_ENDPOINT", "") +from __future__ import annotations +import os -@dataclass(frozen=True) -class QualityConfig: - min_export_score: float = float(os.getenv("CRM_MIN_EXPORT_SCORE", "60")) - min_match_score: float = float(os.getenv("CRM_MIN_MATCH_SCORE", "70")) - stale_days_threshold: int = int(os.getenv("CRM_STALE_DAYS", "365")) - duplicate_similarity_threshold: float = float(os.getenv("CRM_DUPE_THRESHOLD", "0.85")) - max_duplicate_batch: int = int(os.getenv("CRM_MAX_DUPE_BATCH", "5000")) - email_verification_enabled: bool = os.getenv("CRM_EMAIL_VERIFY", "true").lower() == "true" - phone_verification_enabled: bool = os.getenv("CRM_PHONE_VERIFY", "true").lower() == "true" - address_verification_enabled: bool = os.getenv("CRM_ADDR_VERIFY", "true").lower() == "true" - - -@dataclass(frozen=True) -class ExportConfig: - enhanced_conversions_batch_size: int = int(os.getenv("CRM_EC_BATCH_SIZE", "2000")) - enhanced_conversions_max_retries: int = int(os.getenv("CRM_EC_MAX_RETRIES", "5")) - enhanced_conversions_retry_delay_s: int = int(os.getenv("CRM_EC_RETRY_DELAY", "300")) - customer_match_batch_size: int = int(os.getenv("CRM_CM_BATCH_SIZE", "500000")) - conversion_lookback_days: int = int(os.getenv("CRM_CONV_LOOKBACK_DAYS", "90")) - export_lock_timeout_s: int = int(os.getenv("CRM_EXPORT_LOCK_TIMEOUT", "600")) - - -@dataclass(frozen=True) -class ObservabilityConfig: - log_level: str = os.getenv("CRM_LOG_LEVEL", "INFO") - metrics_enabled: bool = os.getenv("CRM_METRICS_ENABLED", "true").lower() == "true" - metrics_prefix: str = os.getenv("CRM_METRICS_PREFIX", "crm_ai") - alert_webhook_url: str = os.getenv("CRM_ALERT_WEBHOOK", "") - alert_on_quality_drop: float = float(os.getenv("CRM_ALERT_QUALITY_THRESHOLD", "70")) - alert_on_match_rate_drop: float = float(os.getenv("CRM_ALERT_MATCH_THRESHOLD", "40")) - +from pydantic_settings import BaseSettings -@dataclass(frozen=True) -class PlatformConfig: - db: DatabaseConfig = field(default_factory=DatabaseConfig) - google_ads: GoogleAdsConfig = field(default_factory=GoogleAdsConfig) - salesforce: SalesforceConfig = field(default_factory=SalesforceConfig) - quality: QualityConfig = field(default_factory=QualityConfig) - export: ExportConfig = field(default_factory=ExportConfig) - observability: ObservabilityConfig = field(default_factory=ObservabilityConfig) - default_country: str = os.getenv("CRM_DEFAULT_COUNTRY", "US") - default_currency: str = os.getenv("CRM_DEFAULT_CURRENCY", "USD") - default_timezone: str = os.getenv("CRM_DEFAULT_TIMEZONE", "America/New_York") +class Settings(BaseSettings): + app_name: str = "l9-codegraph" + debug: bool = os.getenv("DEBUG", "false").lower() == "true" + port: int = int(os.getenv("PORT", "8002")) + neo4j_uri: str = os.getenv("NEO4J_BOLT_URI", "bolt://localhost:7687") + neo4j_user: str = os.getenv("NEO4J_USER", "neo4j") + neo4j_password: str = os.getenv("NEO4J_PASSWORD", "") + github_token: str = os.getenv("GITHUB_TOKEN", "") + codegraph_enabled: bool = os.getenv("L9_CODEGRAPH_ENABLED", "true").lower() == "true" + plangraph_enabled: bool = os.getenv("L9_PLANGRAPH_ENABLED", "true").lower() == "true" -def load_config() -> PlatformConfig: - return PlatformConfig() + class Config: + env_file = ".env" + extra = "ignore" diff --git a/pyproject.toml b/pyproject.toml index 1850744..b313d99 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [tool.poetry] -name = "l9-service" +name = "l9-codegraph" version = "0.1.0" description = "L9 microservice — replace with your service name" authors = ["Igor Beylin "] @@ -16,6 +16,12 @@ httpx = "^0.28.0" structlog = "^25.0.0" pyyaml = ">=6.0.0" prometheus-client = "^0.24.1" +neo4j = "^5.0" +tree-sitter = ">=0.21,<0.25" +tree-sitter-python = ">=0.21" +tree-sitter-typescript = ">=0.21" +tree-sitter-javascript = ">=0.21" +gitpython = "^3.1" [tool.poetry.group.dev.dependencies] pytest = "^8.3.0" diff --git a/sonar-project.properties b/sonar-project.properties index da02dac..f7ca264 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -1,4 +1,5 @@ -sonar.projectKey=cryptoxdog_REPLACE_ME +sonar.projectKey=cryptoxdog_l9-codegraph +sonar.projectName=l9-codegraph sonar.organization=cryptoxdog sonar.sources=engine sonar.tests=tests diff --git a/tests/test_codegraph_unit.py b/tests/test_codegraph_unit.py new file mode 100644 index 0000000..fb71887 --- /dev/null +++ b/tests/test_codegraph_unit.py @@ -0,0 +1,129 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [test] +# tags: [codegraph, unit, parser] +# status: active +# --- /L9_META --- +"""CodeGraph unit tests — no Neo4j required.""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +from engine.codegraph.parser import _SKIP_DIRS, CodeLineParser + + +class TestParserFindsPyFiles: + def test_parser_finds_py_files(self) -> None: + """Parser should discover .py files in a directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + # Create some Python files + (root / "module_a.py").write_text("def foo(): pass\n") + (root / "module_b.py").write_text("class Bar: pass\n") + # Non-Python file — should NOT be found + (root / "readme.txt").write_text("hello\n") + + parser = CodeLineParser(str(root)) + files = parser.find_files() + py_files = [f for f in files if f.suffix == ".py"] + + assert len(py_files) == 2 + names = {f.name for f in py_files} + assert "module_a.py" in names + assert "module_b.py" in names + + def test_parser_finds_js_and_ts_files(self) -> None: + """Parser should discover .ts and .js files.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + (root / "index.ts").write_text("function hello() {}\n") + (root / "util.js").write_text("const x = () => {};\n") + + parser = CodeLineParser(str(root)) + files = parser.find_files() + exts = {f.suffix for f in files} + + assert ".ts" in exts + assert ".js" in exts + + +class TestParserExtractsDefinitions: + def test_parser_extracts_function_definition(self) -> None: + """Parser should extract function definitions from Python source.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + src = "def my_function(x, y):\n return x + y\n" + (root / "funcs.py").write_text(src) + + parser = CodeLineParser(str(root)) + result = parser.parse_file(root / "funcs.py") + + assert "my_function" in result["definitions"] + + def test_parser_extracts_class_definition(self) -> None: + """Parser should extract class definitions from Python source.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + src = "class MyClass:\n def method(self):\n pass\n" + (root / "classes.py").write_text(src) + + parser = CodeLineParser(str(root)) + result = parser.parse_file(root / "classes.py") + + assert "MyClass" in result["definitions"] + + def test_parser_extracts_both_function_and_class(self) -> None: + """Parser should extract both function and class defs in same file.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + src = "class ServiceA:\n pass\n\ndef process_data(records):\n return records\n" + (root / "mixed.py").write_text(src) + + parser = CodeLineParser(str(root)) + result = parser.parse_file(root / "mixed.py") + + assert "ServiceA" in result["definitions"] + assert "process_data" in result["definitions"] + + +class TestParserSkipsGitDir: + def test_parser_skips_git_dir(self) -> None: + """Parser should skip files inside .git directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + git_dir = root / ".git" + git_dir.mkdir() + (git_dir / "hooks.py").write_text("def hook(): pass\n") + (root / "real.py").write_text("def real_func(): pass\n") + + parser = CodeLineParser(str(root)) + files = parser.find_files() + file_parts = [str(f.relative_to(root)) for f in files] + + assert not any(".git" in p for p in file_parts), ( + f"Should not include .git files, got: {file_parts}" + ) + assert any("real.py" in p for p in file_parts) + + def test_parser_skips_node_modules(self) -> None: + """Parser should skip node_modules directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + nm = root / "node_modules" / "some-lib" + nm.mkdir(parents=True) + (nm / "index.js").write_text("function lib() {}\n") + (root / "app.js").write_text("function app() {}\n") + + parser = CodeLineParser(str(root)) + files = parser.find_files() + file_parts = [str(f.relative_to(root)) for f in files] + + assert not any("node_modules" in p for p in file_parts) + assert any("app.js" in p for p in file_parts) + + def test_skip_dirs_contains_expected(self) -> None: + """_SKIP_DIRS frozenset should contain standard ignore dirs.""" + for d in (".git", "node_modules", "__pycache__", "dist", "build"): + assert d in _SKIP_DIRS, f"Expected '{d}' in _SKIP_DIRS" diff --git a/tests/test_plangraph_unit.py b/tests/test_plangraph_unit.py new file mode 100644 index 0000000..34dd819 --- /dev/null +++ b/tests/test_plangraph_unit.py @@ -0,0 +1,256 @@ +# --- L9_META --- +# l9_schema: 1 +# layer: [test] +# tags: [plangraph, unit, spec-parser, topology] +# status: active +# --- /L9_META --- +"""PlanGraph unit tests — no Neo4j required.""" + +from __future__ import annotations + +import tempfile +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +import yaml + +from engine.plangraph.retriever import PlanGraphRetriever +from engine.plangraph.spec_parser import SpecParser + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +SAMPLE_SPEC = { + "constellation": "testcon", + "services": { + "ALPHA": { + "status": "built", + "description": "First service", + "depends_on": [], + }, + "BETA": { + "status": "planned", + "description": "Second service", + "depends_on": ["ALPHA"], + }, + "GAMMA": { + "status": "planned", + "description": "Third service", + "depends_on": ["ALPHA"], + }, + "DELTA": { + "status": "planned", + "description": "Fourth service", + "depends_on": ["BETA", "GAMMA"], + }, + }, + "interfaces": [ + {"name": "alpha_in", "service": "ALPHA", "direction": "inbound", "protocol": "http"}, + {"name": "alpha_out", "service": "ALPHA", "direction": "outbound", "protocol": "http"}, + {"name": "beta_in", "service": "BETA", "direction": "inbound", "protocol": "http"}, + ], + "flows": [ + {"from": "ALPHA", "to": "BETA", "label": "data"}, + {"from": "ALPHA", "to": "GAMMA", "label": "context"}, + {"from": "BETA", "to": "DELTA", "label": "processed", "feedback": False}, + {"from": "DELTA", "to": "ALPHA", "label": "feedback", "feedback": True}, + ], +} + + +@pytest.fixture() +def spec_dir_with_yaml(): + """Create a temp dir with a sample constellation.yaml.""" + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "constellation.yaml" + path.write_text(yaml.dump(SAMPLE_SPEC), encoding="utf-8") + yield tmpdir + + +# --------------------------------------------------------------------------- +# test_spec_parser_extracts_services +# --------------------------------------------------------------------------- + + +class TestSpecParserExtractsServices: + def test_spec_parser_extracts_services(self, spec_dir_with_yaml: str) -> None: + """SpecParser should return all 4 services from the YAML.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + services = result["services"] + assert len(services) == 4 + + names = {s["name"] for s in services} + assert names == {"ALPHA", "BETA", "GAMMA", "DELTA"} + + def test_spec_parser_extracts_service_status(self, spec_dir_with_yaml: str) -> None: + """SpecParser should preserve service status.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + by_name = {s["name"]: s for s in result["services"]} + assert by_name["ALPHA"]["status"] == "built" + assert by_name["BETA"]["status"] == "planned" + + def test_spec_parser_extracts_depends_on(self, spec_dir_with_yaml: str) -> None: + """SpecParser should extract depends_on lists.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + by_name = {s["name"]: s for s in result["services"]} + assert by_name["ALPHA"]["depends_on"] == [] + assert "ALPHA" in by_name["BETA"]["depends_on"] + assert set(by_name["DELTA"]["depends_on"]) == {"BETA", "GAMMA"} + + +# --------------------------------------------------------------------------- +# test_spec_parser_extracts_interfaces +# --------------------------------------------------------------------------- + + +class TestSpecParserExtractsInterfaces: + def test_spec_parser_extracts_interfaces(self, spec_dir_with_yaml: str) -> None: + """SpecParser should return all interface definitions.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + interfaces = result["interfaces"] + assert len(interfaces) == 3 + + names = {i["name"] for i in interfaces} + assert "alpha_in" in names + assert "alpha_out" in names + assert "beta_in" in names + + def test_spec_parser_interface_has_direction(self, spec_dir_with_yaml: str) -> None: + """Each interface should have a direction field.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + for iface in result["interfaces"]: + assert "direction" in iface + assert iface["direction"] in ("inbound", "outbound") + + def test_spec_parser_interface_has_service(self, spec_dir_with_yaml: str) -> None: + """Each interface should reference its parent service.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + by_name = {i["name"]: i for i in result["interfaces"]} + assert by_name["alpha_in"]["service"] == "ALPHA" + assert by_name["beta_in"]["service"] == "BETA" + + +# --------------------------------------------------------------------------- +# test_spec_parser_extracts_flows +# --------------------------------------------------------------------------- + + +class TestSpecParserExtractsFlows: + def test_spec_parser_extracts_flows(self, spec_dir_with_yaml: str) -> None: + """SpecParser should return all flow edges.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + flows = result["flows"] + assert len(flows) == 4 + + def test_spec_parser_flow_has_from_to(self, spec_dir_with_yaml: str) -> None: + """Each flow should have 'from' and 'to' fields.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + for flow in result["flows"]: + assert "from" in flow + assert "to" in flow + + def test_spec_parser_flow_feedback_flag(self, spec_dir_with_yaml: str) -> None: + """SpecParser should preserve feedback=True flag on flows.""" + parser = SpecParser(spec_dir_with_yaml) + result = parser.parse("constellation.yaml") + + feedback_flows = [f for f in result["flows"] if f.get("feedback")] + assert len(feedback_flows) == 1 + assert feedback_flows[0]["from"] == "DELTA" + assert feedback_flows[0]["to"] == "ALPHA" + + +# --------------------------------------------------------------------------- +# test_build_order_topological +# --------------------------------------------------------------------------- + + +class TestBuildOrderTopological: + """Test _topological_sort directly — no Neo4j connection needed.""" + + def _make_retriever(self) -> PlanGraphRetriever: + """Create a PlanGraphRetriever with mocked Neo4j driver.""" + retriever = PlanGraphRetriever.__new__(PlanGraphRetriever) + retriever.driver = MagicMock() + return retriever + + def test_topological_sort_simple_chain(self) -> None: + """A -> B -> C should sort to [A, B, C].""" + retriever = self._make_retriever() + dep_map = {"A": [], "B": ["A"], "C": ["B"]} + topo, groups = retriever._topological_sort(dep_map) + + assert topo.index("A") < topo.index("B") + assert topo.index("B") < topo.index("C") + + def test_topological_sort_parallel_groups(self) -> None: + """B and C both depend on A — should be in the same wave.""" + retriever = self._make_retriever() + dep_map = { + "ALPHA": [], + "BETA": ["ALPHA"], + "GAMMA": ["ALPHA"], + "DELTA": ["BETA", "GAMMA"], + } + topo, groups = retriever._topological_sort(dep_map) + + # ALPHA must come first + assert topo[0] == "ALPHA" + # BETA and GAMMA should be in same wave (wave 2) + wave2 = groups[1] + assert set(wave2) == {"BETA", "GAMMA"} + # DELTA must come last + assert topo[-1] == "DELTA" + + def test_topological_sort_no_deps(self) -> None: + """Services with no deps should all be in wave 1.""" + retriever = self._make_retriever() + dep_map = {"A": [], "B": [], "C": []} + topo, groups = retriever._topological_sort(dep_map) + + assert len(groups) == 1 + assert set(groups[0]) == {"A", "B", "C"} + assert set(topo) == {"A", "B", "C"} + + def test_topological_sort_revopsos_shape(self) -> None: + """Test RevOpsOS topology: ENRICH+GRAPH -> SCORE -> ROUTE/HEALTH -> FORECAST/HANDOFF.""" + retriever = self._make_retriever() + dep_map = { + "ENRICH": [], + "GRAPH": [], + "SCORE": ["ENRICH", "GRAPH"], + "ROUTE": ["SCORE", "GRAPH"], + "FORECAST": ["SCORE", "ROUTE"], + "SIGNAL": ["ENRICH"], + "HEALTH": ["SCORE", "ENRICH"], + "HANDOFF": ["ROUTE", "SCORE"], + } + topo, groups = retriever._topological_sort(dep_map) + + # ENRICH and GRAPH must precede SCORE + assert topo.index("ENRICH") < topo.index("SCORE") + assert topo.index("GRAPH") < topo.index("SCORE") + # SCORE must precede ROUTE + assert topo.index("SCORE") < topo.index("ROUTE") + # ROUTE must precede FORECAST + assert topo.index("ROUTE") < topo.index("FORECAST") + # All services present + assert set(topo) == set(dep_map.keys())