feat: extend repository intelligence with symbol calls and test impact v2

.github/workflows/repository-intelligence.yml

@@ -2,7 +2,7 @@ name: Repository Intelligence CI
 on:
   pull_request:
   push:
-    branches: ["feat/repository-intelligence-v1"]
+    branches: ["feat/repository-intelligence-v1", "feat/repository-intelligence-v2"]
 permissions:
   contents: read
 jobs:

README.md

@@ -2,17 +2,21 @@
 
 This repository is being evolved from file-dump prompting scripts into a deterministic repository-intelligence engine for governed software development.
 
-## v1 capabilities
+## v2 capabilities
 
 - repository file inventory with SHA-256 content identity;
 - Python top-level symbol graph;
 - Python import dependency graph;
+- conservative Python symbol-call graph;
 - relative JS/TS import dependency graph;
+- test-to-code dependency edges;
+- top-level repository component mapping;
 - reverse dependency change-impact analysis;
+- impacted test and component identification;
 - deterministic graph and impact digests;
 - CLI suitable for CI and OpenHands/agent consumption.
 
-No LLM is required for v1 graph construction. This keeps repository structure and change-impact evidence reproducible.
+No LLM is required for graph construction. This keeps repository structure and change-impact evidence reproducible.
 
 ## CLI
 
@@ -21,6 +25,10 @@ repo-intel graph /path/to/repo --out repository-graph.json
 repo-intel impact repository-graph.json src/example.py --depth 3 --out change-impact.json
 ```
 
+## Static-analysis truth boundary
+
+The graph is intentionally conservative. It does not claim complete semantic program analysis. Dynamic imports, reflection, monkey-patching, generated code, runtime dependency injection, and many cross-language call relationships require later analyzers.
+
 ## Legacy scripts
 
-`base_print.py` and `base_print_ai_model.py` are preserved as historical utilities. They are not the architectural foundation of the new repository-intelligence layer.
+`base_print.py` and `base_print_ai_model.py` are preserved as historical utilities. They are not the architectural foundation of the repository-intelligence layer.

repo_intelligence/graph.py

@@ -1,7 +1,7 @@
-"""Deterministic repository graph construction.
+"""Deterministic repository graph construction v2.
 
-v1 intentionally avoids LLM dependence. It builds a reproducible graph from
-repository files, Python imports/symbols, and selected JS/TS imports.
+Builds reproducible file, symbol, import, test-to-code, and conservative Python
+symbol-call relationships without requiring an LLM.
 """
 from __future__ import annotations
 
@@ -10,7 +10,6 @@
 import json
 import os
 import re
-from dataclasses import dataclass, asdict
 from pathlib import Path
 from typing import Iterable
 
@@ -51,27 +50,79 @@ def _resolve_py_import(root:Path,current:Path,module:str,level:int)->str|None:
         base=base[:keep]
     target=".".join([*base,*([module] if module else [])]).strip(".") if level else module
     if not target: return None
-    candidates=[root/Path(*target.split(".")).with_suffix(".py"),root/Path(*target.split("."))/ "__init__.py"]
+    candidates=[
+        root/Path(*target.split(".")).with_suffix(".py"),
+        root/Path(*target.split("."))/"__init__.py",
+    ]
     for c in candidates:
         if c.exists(): return c.relative_to(root).as_posix()
     return None
 
 
-def _parse_python(root:Path,path:Path,text:str)->tuple[list[dict],list[str]]:
-    symbols=[]; deps=[]
+def _symbol_id(file_path:str,name:str)->str:
+    return f"{file_path}::{name}"
+
+
+def _called_names(node:ast.AST)->set[str]:
+    names=set()
+    for child in ast.walk(node):
+        if isinstance(child,ast.Call):
+            fn=child.func
+            if isinstance(fn,ast.Name):
+                names.add(fn.id)
+            elif isinstance(fn,ast.Attribute):
+                names.add(fn.attr)
+    return names
+
+
+def _parse_python(root:Path,path:Path,text:str)->tuple[list[dict],list[str],list[dict]]:
+    symbols=[]; deps=[]; call_candidates=[]
     try: tree=ast.parse(text)
-    except SyntaxError: return symbols,deps
+    except SyntaxError: return symbols,deps,call_candidates
+
+    local_names={
+        node.name
+        for node in tree.body
+        if isinstance(node,(ast.FunctionDef,ast.AsyncFunctionDef,ast.ClassDef))
+    }
+    imported_symbols:dict[str,tuple[str,str]]={}
+
     for node in tree.body:
         if isinstance(node,(ast.FunctionDef,ast.AsyncFunctionDef,ast.ClassDef)):
-            symbols.append({"name":node.name,"kind":"class" if isinstance(node,ast.ClassDef) else "function","line":getattr(node,"lineno",None)})
+            symbols.append({
+                "name":node.name,
+                "kind":"class" if isinstance(node,ast.ClassDef) else "function",
+                "line":getattr(node,"lineno",None),
+            })
         elif isinstance(node,ast.Import):
             for alias in node.names:
                 d=_resolve_py_import(root,path,alias.name,0)
                 if d: deps.append(d)
         elif isinstance(node,ast.ImportFrom):
             d=_resolve_py_import(root,path,node.module or "",node.level)
-            if d: deps.append(d)
-    return sorted(symbols,key=lambda x:(x["name"],x["kind"],x["line"] or 0)),sorted(set(deps))
+            if d:
+                deps.append(d)
+                for alias in node.names:
+                    if alias.name!="*":
+                        imported_symbols[alias.asname or alias.name]=(d,alias.name)
+
+    rel=path.relative_to(root).as_posix()
+    for node in tree.body:
+        if not isinstance(node,(ast.FunctionDef,ast.AsyncFunctionDef)):
+            continue
+        caller=_symbol_id(rel,node.name)
+        for name in sorted(_called_names(node)):
+            if name in local_names:
+                call_candidates.append({"from":caller,"to":_symbol_id(rel,name),"type":"calls"})
+            elif name in imported_symbols:
+                dep_file,dep_name=imported_symbols[name]
+                call_candidates.append({"from":caller,"to":_symbol_id(dep_file,dep_name),"type":"calls"})
+
+    return (
+        sorted(symbols,key=lambda x:(x["name"],x["kind"],x["line"] or 0)),
+        sorted(set(deps)),
+        sorted(call_candidates,key=lambda x:(x["from"],x["to"],x["type"])),
+    )
 
 
 def _parse_js_like(root:Path,path:Path,text:str)->list[str]:
@@ -80,35 +131,100 @@ def _parse_js_like(root:Path,path:Path,text:str)->list[str]:
         spec=next((g for g in m.groups() if g),None)
         if not spec or not spec.startswith("."): continue
         base=(path.parent/spec).resolve()
-        for c in [base,base.with_suffix(".js"),base.with_suffix(".jsx"),base.with_suffix(".ts"),base.with_suffix(".tsx"),base/"index.js",base/"index.ts"]:
+        for c in [
+            base,base.with_suffix(".js"),base.with_suffix(".jsx"),
+            base.with_suffix(".ts"),base.with_suffix(".tsx"),
+            base/"index.js",base/"index.ts",
+        ]:
             try:
                 if c.exists() and c.is_file() and root.resolve() in c.resolve().parents:
                     out.append(c.relative_to(root).as_posix()); break
-            except OSError: pass
+            except OSError:
+                pass
     return sorted(set(out))
 
 
+def _is_test_file(path:str)->bool:
+    p=Path(path)
+    return (
+        any(part in {"test","tests","spec","specs"} for part in p.parts[:-1])
+        or p.name.startswith("test_")
+        or p.name.endswith("_test.py")
+        or p.name.endswith(".spec.js")
+        or p.name.endswith(".spec.ts")
+        or p.name.endswith(".test.js")
+        or p.name.endswith(".test.ts")
+    )
+
+
 def build_repository_graph(root:str|Path)->dict:
     root=Path(root).resolve()
     if not root.is_dir(): raise ValueError("root must be an existing directory")
-    files=[]; edges=[]; symbols=[]
+
+    files=[]; edges=[]; symbols=[]; symbol_edges=[]
+    deps_by_file:dict[str,list[str]]={}
+
     for path in _iter_files(root):
         rel=path.relative_to(root).as_posix()
         raw=path.read_bytes()
         try: text=raw.decode("utf-8")
         except UnicodeDecodeError: continue
-        entry={"path":rel,"bytes":len(raw),"sha256":_sha(raw),"extension":path.suffix.lower()}
+
+        files.append({
+            "path":rel,
+            "bytes":len(raw),
+            "sha256":_sha(raw),
+            "extension":path.suffix.lower(),
+            "is_test":_is_test_file(rel),
+        })
+
         deps=[]
         if path.suffix.lower()==".py":
-            syms,deps=_parse_python(root,path,text)
-            for s in syms: symbols.append({"file":rel,**s})
+            syms,deps,calls=_parse_python(root,path,text)
+            for s in syms:
+                symbols.append({"id":_symbol_id(rel,s["name"]),"file":rel,**s})
+            symbol_edges.extend(calls)
         elif path.suffix.lower() in {".js",".jsx",".ts",".tsx"}:
             deps=_parse_js_like(root,path,text)
-        files.append(entry)
-        for dep in deps: edges.append({"from":rel,"to":dep,"type":"imports"})
+
+        deps_by_file[rel]=deps
+        for dep in deps:
+            edges.append({"from":rel,"to":dep,"type":"imports"})
+
+    known_symbols={s["id"] for s in symbols}
+    symbol_edges=[
+        e for e in symbol_edges
+        if e["from"] in known_symbols and e["to"] in known_symbols
+    ]
+
+    test_files={f["path"] for f in files if f["is_test"]}
+    for test_file in sorted(test_files):
+        for dep in deps_by_file.get(test_file,[]):
+            if dep not in test_files:
+                edges.append({"from":test_file,"to":dep,"type":"tests"})
+
+    components=[]
+    for f in files:
+        path=f["path"]
+        parts=Path(path).parts
+        component=parts[0] if len(parts)>1 else "."
+        components.append({"file":path,"component":component})
+
     files=sorted(files,key=lambda x:x["path"])
-    edges=sorted(edges,key=lambda x:(x["from"],x["to"],x["type"]))
+    edges=sorted({(e["from"],e["to"],e["type"]) for e in edges})
+    edges=[{"from":a,"to":b,"type":t} for a,b,t in edges]
     symbols=sorted(symbols,key=lambda x:(x["file"],x["line"] or 0,x["name"]))
-    core={"schema_version":"1.0","root_name":root.name,"files":files,"symbols":symbols,"edges":edges}
+    symbol_edges=sorted(symbol_edges,key=lambda x:(x["from"],x["to"],x["type"]))
+    components=sorted(components,key=lambda x:(x["component"],x["file"]))
+
+    core={
+        "schema_version":"2.0",
+        "root_name":root.name,
+        "files":files,
+        "components":components,
+        "symbols":symbols,
+        "edges":edges,
+        "symbol_edges":symbol_edges,
+    }
     core["graph_sha256"]=_sha(_canon(core).encode())
     return core

repo_intelligence/impact.py

@@ -1,4 +1,4 @@
-"""Change-impact analysis over a deterministic repository graph."""
+"""Change-impact analysis over a deterministic repository graph v2."""
 from __future__ import annotations
 
 import hashlib,json
@@ -18,20 +18,60 @@ def analyze_change_impact(graph:dict,changed_paths:list[str],max_depth:int=3)->d
     known={f["path"] for f in graph.get("files",[]) if isinstance(f,dict) and "path" in f}
     changed=sorted(set(changed_paths))
     unknown=sorted(p for p in changed if p not in known)
+
     reverse={p:set() for p in known}
+    edge_types:dict[tuple[str,str],set[str]]={}
     for e in graph.get("edges",[]):
-        if isinstance(e,dict) and e.get("type")=="imports" and e.get("from") in known and e.get("to") in known:
-            reverse[e["to"]].add(e["from"])
+        if not isinstance(e,dict): continue
+        src=e.get("from"); dst=e.get("to"); typ=e.get("type")
+        if src in known and dst in known and typ in {"imports","tests"}:
+            reverse[dst].add(src)
+            edge_types.setdefault((dst,src),set()).add(str(typ))
+
     distance={p:0 for p in changed if p in known}
+    reasons={p:{"changed"} for p in distance}
     q=deque(sorted(distance))
     while q:
         cur=q.popleft(); d=distance[cur]
         if d>=max_depth: continue
         for dep in sorted(reverse.get(cur,())):
-            if dep not in distance:
-                distance[dep]=d+1; q.append(dep)
-    impacted=[{"path":p,"distance":distance[p]} for p in sorted(distance,key=lambda x:(distance[x],x))]
-    symbols=[s for s in graph.get("symbols",[]) if isinstance(s,dict) and s.get("file") in distance]
-    core={"schema_version":"1.0","graph_sha256":graph.get("graph_sha256"),"changed_paths":changed,"unknown_paths":unknown,"max_depth":max_depth,"impacted_files":impacted,"impacted_symbols":symbols}
+            candidate=d+1
+            rel_types=edge_types.get((cur,dep),{"dependency"})
+            if dep not in distance or candidate<distance[dep]:
+                distance[dep]=candidate
+                reasons[dep]=set(rel_types)
+                q.append(dep)
+            elif candidate==distance[dep]:
+                reasons.setdefault(dep,set()).update(rel_types)
+
+    impacted=[
+        {"path":p,"distance":distance[p],"reasons":sorted(reasons.get(p,()))}
+        for p in sorted(distance,key=lambda x:(distance[x],x))
+    ]
+    symbols=[
+        s for s in graph.get("symbols",[])
+        if isinstance(s,dict) and s.get("file") in distance
+    ]
+    tests=[
+        item for item in impacted
+        if any(f.get("path")==item["path"] and f.get("is_test") is True for f in graph.get("files",[]))
+    ]
+    components=sorted({
+        c.get("component")
+        for c in graph.get("components",[])
+        if isinstance(c,dict) and c.get("file") in distance and isinstance(c.get("component"),str)
+    })
+
+    core={
+        "schema_version":"2.0",
+        "graph_sha256":graph.get("graph_sha256"),
+        "changed_paths":changed,
+        "unknown_paths":unknown,
+        "max_depth":max_depth,
+        "impacted_files":impacted,
+        "impacted_symbols":symbols,
+        "impacted_tests":tests,
+        "impacted_components":components,
+    }
     core["impact_sha256"]=_sha(_canon(core))
     return core

tests/test_repository_intelligence.py

@@ -1,5 +1,4 @@
 from pathlib import Path
-import json
 from repo_intelligence.graph import build_repository_graph
 from repo_intelligence.impact import analyze_change_impact
 
@@ -10,8 +9,38 @@ def test_graph_and_impact_are_deterministic(tmp_path:Path):
     g1=build_repository_graph(tmp_path); g2=build_repository_graph(tmp_path)
     assert g1==g2
     assert {"from":"a.py","to":"b.py","type":"imports"} in g1["edges"]
+    assert {"from":"a.py::g","to":"b.py::f","type":"calls"} in g1["symbol_edges"]
     impact=analyze_change_impact(g1,["b.py"])
-    assert impact["impacted_files"]==[{"path":"b.py","distance":0},{"path":"a.py","distance":1}]
+    assert impact["impacted_files"]==[
+        {"path":"b.py","distance":0,"reasons":["changed"]},
+        {"path":"a.py","distance":1,"reasons":["imports"]},
+    ]
+
+
+def test_test_mapping_and_component_impact(tmp_path:Path):
+    (tmp_path/"src").mkdir()
+    (tmp_path/"tests").mkdir()
+    (tmp_path/"src"/"calc.py").write_text("def add(a,b):\n    return a+b\n",encoding="utf-8")
+    (tmp_path/"tests"/"test_calc.py").write_text(
+        "from src.calc import add\n\ndef test_add():\n    assert add(1,2)==3\n",
+        encoding="utf-8",
+    )
+    graph=build_repository_graph(tmp_path)
+    assert {"from":"tests/test_calc.py","to":"src/calc.py","type":"tests"} in graph["edges"]
+    impact=analyze_change_impact(graph,["src/calc.py"])
+    assert impact["impacted_tests"]==[
+        {"path":"tests/test_calc.py","distance":1,"reasons":["imports","tests"]}
+    ]
+    assert impact["impacted_components"]==["src","tests"]
+
+
+def test_local_symbol_call_edge(tmp_path:Path):
+    (tmp_path/"mod.py").write_text(
+        "def helper():\n    return 1\n\ndef caller():\n    return helper()\n",
+        encoding="utf-8",
+    )
+    graph=build_repository_graph(tmp_path)
+    assert {"from":"mod.py::caller","to":"mod.py::helper","type":"calls"} in graph["symbol_edges"]
 
 
 def test_unknown_changed_path_is_reported(tmp_path:Path):