feat(graph): L3/L4 program-slice engine core (Tasks 1–6 of #270)

cldk/graph/capability.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+from typing import Optional, Dict
+
+
+class CapabilityError(Exception):
+    pass
+
+
+def require(level_needed: int, provider, *, strict: bool, what: str) -> Optional[Dict]:
+    available = provider.max_level()
+    if available >= level_needed:
+        return None
+    if strict:
+        raise CapabilityError(
+            f"{what} requires analysis level {level_needed}; backend is at level {available}. "
+            f"Re-analyze at -a {level_needed} or drop strict=True to degrade.")
+    return {
+        "requested": level_needed,
+        "available": available,
+        "gap": f"{what} requires level {level_needed}; backend at level {available} — "
+               f"reduced result returned; absence of a result here is UNKNOWN, not safety.",
+    }

cldk/graph/engine.py

@@ -0,0 +1,196 @@
+# cldk/graph/engine.py
+from __future__ import annotations
+from typing import Iterable, List, Optional, Tuple
+import networkx as nx
+from cldk.graph.provider import ProgramGraphProvider, resolve_vertex
+from cldk.graph.capability import require
+from cldk.graph.result import SliceResult, FlowResult, FlowPath
+
+
+def _filter_edges(g: nx.MultiDiGraph, families: Iterable[str]) -> nx.MultiDiGraph:
+    fam = set(families)
+    out = nx.MultiDiGraph()
+    out.add_nodes_from(g.nodes(data=True))
+    for u, v, k, d in g.edges(keys=True, data=True):
+        if d.get("family") in fam:
+            out.add_edge(u, v, key=k, **d)
+    return out
+
+
+_TIER_RANK = {"unresolved": 0, "structural": 1, "resolved": 2}
+_RANK_TIER = {v: k for k, v in _TIER_RANK.items()}
+
+# Provisional witness-enumeration bounds (to be tuned against real graphs in a later
+# task): flows_to explores simple paths no deeper than _PATH_CUTOFF hops and stops
+# collecting witnesses at _MAX_PATHS; explain()["truncated"] reports whether the
+# path cap was hit (depth-cutoff drops are not separately detectable and are folded
+# into the same provisional-bounds caveat).
+_MAX_PATHS = 1000
+_PATH_CUTOFF = 64
+
+
+def _ddg_tier(prov) -> str:
+    # Membership, not exact-list: prov is a provenance set in list form, and
+    # ["ssa", "points-to"] is STRONGER evidence than ["points-to"] alone.
+    prov = prov or []
+    if "points-to" in prov:
+        return "resolved"
+    if "ssa" in prov:
+        return "structural"
+    return "unresolved"
+
+
+class Engine:
+    def __init__(self, provider: ProgramGraphProvider):
+        self.p = provider
+
+    def _evidence(self, uris, seeds, roles=None, default_role="def"):
+        # Seeds are always "seed"; other vertices take the verb's default_role
+        # (slices/flows: "def", control_deps: "control", def_use: "use").
+        roles = roles or {}
+        ev = []
+        for u in uris:
+            fl, code = self.p.source_slice(u)
+            ev.append({"uri": u, "file_line": fl, "code": code,
+                       "role": "seed" if u in seeds else roles.get(u, default_role)})
+        return ev
+
+    def _intra(self, seed, edges, backward, strict, what, interprocedural=None,
+               default_role="def") -> SliceResult:
+        note = require(3, self.p, strict=strict, what=what)
+        want_inter = interprocedural if interprocedural is not None else (self.p.max_level() >= 4)
+        if interprocedural is True:
+            inter_note = require(4, self.p, strict=strict, what=f"interprocedural {what}")
+            if inter_note:
+                note = inter_note
+                want_inter = False
+        # Only dataflow (param_in/param_out/summary) crosses callable boundaries, so the
+        # sdg overlay is additionally gated on the ddg family being requested: a cfg- or
+        # cdg-only slice never crosses, even on an L4 backend. want_inter is the single
+        # source of truth — it both gates the overlay and feeds explain()["interprocedural"].
+        want_inter = want_inter and self.p.max_level() >= 4 and "ddg" in set(edges)
+        seeds = resolve_vertex(self.p, seed)
+        g = _filter_edges(self.p.program_graph(self.p.callable_of(seeds[0])), edges)
+        if want_inter:
+            for e in self.p.sdg_edges():
+                g.add_edge(e.src, e.dst, family="sdg", kind=getattr(e, "kind", None),
+                           var=getattr(e, "var", None), prov=getattr(e, "prov", []))
+        walk = g.reverse(copy=False) if backward else g
+        reached = set(seeds)
+        for s in seeds:
+            if s in walk:
+                reached |= nx.descendants(walk, s)
+        # seed-consistency (Task 4 fix, carried here): evidence/uris must equal subgraph nodes,
+        # and a seed is trivially in its own slice.
+        sub = g.subgraph(reached & set(g.nodes())).copy()   # MultiDiGraph
+        for s in seeds:
+            if s not in sub:
+                sub.add_node(s, kind="seed")
+        ev_nodes = sorted(sub.nodes())   # deterministic; evidence set == subgraph nodes
+        explain = {"seed": seeds, "direction": "backward" if backward else "forward",
+                   "edges": list(edges), "level": self.p.max_level(),
+                   "vertices": len(sub), "interprocedural": bool(want_inter)}
+        if note:
+            explain["degraded"] = note
+        return SliceResult(subgraph=sub,
+                           evidence=self._evidence(ev_nodes, set(seeds),
+                                                   default_role=default_role),
+                           _explain=explain)
+
+    def slice_backward(self, seed, *, edges=("cfg", "cdg", "ddg"),
+                       interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult:
+        return self._intra(seed, edges, True, strict, "slice_backward", interprocedural)
+
+    def slice_forward(self, seed, *, edges=("cfg", "cdg", "ddg"),
+                      interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult:
+        return self._intra(seed, edges, False, strict, "slice_forward", interprocedural)
+
+    def control_deps(self, seed, *, strict: bool = False) -> SliceResult:
+        # Control dependence is intraprocedural in this model — only dataflow (param/summary)
+        # crosses boundaries. Force interprocedural=False so the sdg overlay is never merged
+        # into a pure CDG slice, even on an L4 backend.
+        return self._intra(seed, ("cdg",), backward=True, strict=strict,
+                           what="control_deps", interprocedural=False,
+                           default_role="control")
+
+    def _dataflow_graph(self, *callable_uris) -> nx.MultiDiGraph:
+        # Union of the given callables' intra ddg graphs, plus the summary/param_*
+        # (inter) sdg overlay at L4; below L4 this is intraprocedural ddg only.
+        g = nx.MultiDiGraph()
+        for c in dict.fromkeys(callable_uris):          # dedupe, keep order
+            cg = _filter_edges(self.p.program_graph(c), ("ddg",))
+            g.add_nodes_from(cg.nodes(data=True))
+            for u, v, k, d in cg.edges(keys=True, data=True):
+                g.add_edge(u, v, key=k, **d)
+        if self.p.max_level() >= 4:
+            for e in self.p.sdg_edges():
+                g.add_edge(e.src, e.dst, family="sdg", kind=getattr(e, "kind", None),
+                           var=getattr(e, "var", None), prov=getattr(e, "prov", []))
+        return g
+
+    def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResult:
+        # Full flows_to semantics are interprocedural (ddg + param_in/param_out/summary),
+        # which is L4. Below that, non-strict degrades honestly: the note is attached and
+        # the intra-only ddg witnesses that CAN be computed are still returned.
+        note = require(4, self.p, strict=strict, what="flows_to")
+        src = resolve_vertex(self.p, source_seed)[0]
+        dst = resolve_vertex(self.p, sink_seed)[0]
+        # A sink in a different callable is reachable via param_in/param_out/summary,
+        # so the dataflow graph must span BOTH endpoint callables. (Multi-hop flows
+        # through a THIRD callable's interior need the whole-program graph — deferred.)
+        g = self._dataflow_graph(self.p.callable_of(src), self.p.callable_of(dst))
+        paths: List[FlowPath] = []
+        reached = set()
+        truncated = False
+        if src in g and dst in g:
+            # A MultiDiGraph enumerates a route once per parallel-edge combination, yielding
+            # byte-identical duplicate witnesses. Enumerate over a plain-DiGraph VIEW (one path
+            # per distinct node route) and read per-hop parallel evidence from the MultiDiGraph g.
+            routes = nx.DiGraph(g)
+            for path in nx.all_simple_paths(routes, src, dst, cutoff=_PATH_CUTOFF):
+                if len(paths) >= _MAX_PATHS:
+                    truncated = True
+                    break
+                hops, tiers = [], []
+                for a, b in zip(path, path[1:]):
+                    # MultiDiGraph: get_edge_data returns {key: attrdict} over parallel edges.
+                    # Pick the strongest-confidence parallel edge as the hop's evidence (the step
+                    # is as strong as its best evidence; the path is as weak as its weakest step).
+                    parallels = g.get_edge_data(a, b)
+                    best = max(parallels.values(),
+                               key=lambda d: _TIER_RANK[_ddg_tier(d.get("prov", []))])
+                    t = _ddg_tier(best.get("prov", []))
+                    tiers.append(t)
+                    # Intra edges report their family (cfg/cdg/ddg have no kind); sdg
+                    # boundary edges report the concrete kind (param_in/param_out/summary).
+                    hops.append({"from": a, "to": b,
+                                 "kind": best.get("kind") or best.get("family"),
+                                 "var": best.get("var"), "confidence": t})
+                conf = _RANK_TIER[min(_TIER_RANK[t] for t in tiers)] if tiers else "unresolved"
+                paths.append(FlowPath(source=src, sink=dst, hops=hops, confidence=conf))
+                reached.update(path)
+        explain = {"source": src, "sink": dst, "level": self.p.max_level(),
+                   "paths": len(paths), "truncated": truncated}
+        if note:
+            explain["degraded"] = note
+        sub = g.subgraph(reached).copy()
+        return FlowResult(subgraph=sub, evidence=self._evidence(sorted(sub.nodes()), {src, dst}),
+                          _explain=explain, paths=paths)
+
+    def def_use(self, seed, *, strict: bool = False) -> FlowResult:
+        note = require(3, self.p, strict=strict, what="def_use")
+        s = resolve_vertex(self.p, seed)[0]
+        # NOTE: currently scoped to the seed's callable plus sdg endpoints; uses inside
+        # OTHER callables' interiors arrive with the whole-program dataflow graph (deferred).
+        g = self._dataflow_graph(self.p.callable_of(s))
+        reached = {s} | (nx.descendants(g, s) if s in g else set())
+        sub = g.subgraph(reached & set(g.nodes())).copy()
+        if s not in sub:                          # a seed is trivially in its own def-use result
+            sub.add_node(s, kind="seed")
+        explain = {"seed": s, "level": self.p.max_level(), "vertices": len(sub)}
+        if note:
+            explain["degraded"] = note
+        return FlowResult(subgraph=sub,
+                          evidence=self._evidence(sorted(sub.nodes()), {s},
+                                                  default_role="use"),
+                          _explain=explain, paths=[])

cldk/graph/provider.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+import re
+from abc import ABC, abstractmethod
+from typing import List, Tuple, Iterable, Optional, Any
+import networkx as nx
+
+_LOC = re.compile(r"^(?P<file>.+?):(?P<line>\d+)(?::(?P<col>\d+))?$")
+
+
+class ProgramGraphProvider(ABC):
+    """The per-backend data seam the shared engine consumes. Implemented by local backends
+    (from cpg models) and Neo4j backends (from Cypher). Traversal lives in the engine, not here.
+
+    Below the level a verb requires, the engine gates via require(...); providers should
+    still answer program_graph/resolve_location/callable_of structurally — none of these
+    ever need L3+ data to do so."""
+
+    @abstractmethod
+    def program_graph(self, callable_uri: str) -> nx.MultiDiGraph:
+        """Parallel cfg/cdg/ddg edges between the same vertex pair must stay distinct
+        edges, each carrying its own family/var/prov/kind."""
+    @abstractmethod
+    def sdg_edges(self) -> Iterable[Any]: ...
+    @abstractmethod
+    def resolve_location(self, file: str, line: int, col: Optional[int] = None) -> List[str]: ...
+    @abstractmethod
+    def source_slice(self, vertex_uri: str) -> Tuple[Optional[str], Optional[str]]: ...
+    @abstractmethod
+    def callable_of(self, vertex_uri: str) -> Optional[str]: ...
+    @abstractmethod
+    def max_level(self) -> int: ...
+
+
+def resolve_vertex(provider: ProgramGraphProvider, seed: Any) -> List[str]:
+    """Normalize a polymorphic seed to vertex ids: a BodyNode-like object (has .id), a can:// id
+    string, or a 'file:line[:col]' location string."""
+    if hasattr(seed, "id"):
+        return [seed.id]
+    if isinstance(seed, str):
+        if seed.startswith("can://"):
+            return [seed]
+        m = _LOC.match(seed)
+        if m:
+            col = int(m["col"]) if m["col"] is not None else None
+            found = provider.resolve_location(m["file"], int(m["line"]), col)
+            if not found:
+                # An ordinary user miss (no vertex at that line) must surface as a clean
+                # ValueError here — every verb indexes the result, and [] would IndexError.
+                raise ValueError(f"no vertex at location {seed!r}")
+            return found
+    raise ValueError(f"cannot resolve seed to a vertex: {seed!r} "
+                     f"(expected 'file:line[:col]', a can:// id, or a body node)")

cldk/graph/result.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+import json
+from dataclasses import dataclass, field, asdict
+from typing import List, Dict, Literal
+import networkx as nx
+
+Confidence = Literal["resolved", "structural", "unresolved"]
+
+
+@dataclass(frozen=True)
+class FlowPath:
+    source: str
+    sink: str
+    hops: List[Dict] = field(default_factory=list)
+    confidence: Confidence = "unresolved"
+
+
+@dataclass
+class GraphResult:
+    subgraph: nx.MultiDiGraph
+    evidence: List[Dict]
+    _explain: Dict
+
+    def uris(self) -> List[str]:
+        return [e["uri"] for e in self.evidence]
+
+    def explain(self) -> Dict:
+        return dict(self._explain)
+
+    def to_json(self) -> str:
+        return json.dumps({"evidence": self.evidence, "explain": self._explain,
+                           "vertices": list(self.subgraph.nodes)}, sort_keys=True)
+
+    def __len__(self) -> int:
+        return self.subgraph.number_of_nodes()
+
+    def __bool__(self) -> bool:
+        return self.subgraph.number_of_nodes() > 0
+
+
+@dataclass
+class SliceResult(GraphResult):
+    pass
+
+
+@dataclass
+class FlowResult(GraphResult):
+    paths: List[FlowPath] = field(default_factory=list)
+
+    def to_json(self) -> str:
+        base = json.loads(super().to_json())
+        base["paths"] = [asdict(p) for p in self.paths]
+        return json.dumps(base, sort_keys=True)

tests/graph/test_capability.py

@@ -0,0 +1,23 @@
+import pytest
+from cldk.graph.capability import require, CapabilityError
+
+
+class P:
+    def __init__(self, lvl): self._l = lvl
+    def max_level(self): return self._l
+
+
+def test_satisfied_returns_none():
+    assert require(3, P(4), strict=False, what="slice_backward") is None
+
+
+def test_degrade_returns_note():
+    note = require(4, P(3), strict=False, what="interprocedural flows_to")
+    assert note["requested"] == 4 and note["available"] == 3
+    assert "UNKNOWN, not safety" in note["gap"]
+
+
+def test_strict_raises():
+    with pytest.raises(CapabilityError) as e:
+        require(4, P(3), strict=True, what="flows_to")
+    assert "level 4" in str(e.value)