From aa3a26184a099519159ad051d7172e40e13c0a5b Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 16:18:21 -0400 Subject: [PATCH 01/19] feat(graph): slice/flow result objects (#270) --- cldk/graph/__init__.py | 0 cldk/graph/result.py | 48 ++++++++++++++++++++++++++++++++++++++ tests/graph/__init__.py | 0 tests/graph/test_result.py | 34 +++++++++++++++++++++++++++ 4 files changed, 82 insertions(+) create mode 100644 cldk/graph/__init__.py create mode 100644 cldk/graph/result.py create mode 100644 tests/graph/__init__.py create mode 100644 tests/graph/test_result.py diff --git a/cldk/graph/__init__.py b/cldk/graph/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/cldk/graph/result.py b/cldk/graph/result.py new file mode 100644 index 00000000..9ec50af8 --- /dev/null +++ b/cldk/graph/result.py @@ -0,0 +1,48 @@ +from __future__ import annotations +import json +from dataclasses import dataclass, field +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.DiGraph + 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) diff --git a/tests/graph/__init__.py b/tests/graph/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/graph/test_result.py b/tests/graph/test_result.py new file mode 100644 index 00000000..191cf9dc --- /dev/null +++ b/tests/graph/test_result.py @@ -0,0 +1,34 @@ +import networkx as nx +from cldk.graph.result import GraphResult, SliceResult, FlowResult, FlowPath + + +def _graph(*nodes): + g = nx.DiGraph() + g.add_nodes_from(nodes) + return g + + +def test_graphresult_len_bool_uris(): + g = _graph("a", "b") + r = SliceResult(subgraph=g, evidence=[{"uri": "a"}, {"uri": "b"}], _explain={"level": 3}) + assert len(r) == 2 + assert bool(r) is True + assert r.uris() == ["a", "b"] + assert r.explain() == {"level": 3} + + +def test_empty_result_is_falsy(): + r = SliceResult(subgraph=_graph(), evidence=[], _explain={}) + assert not r + assert len(r) == 0 + + +def test_flowresult_carries_paths_and_serializes(): + p = FlowPath(source="a", sink="c", + hops=[{"from": "a", "to": "b", "kind": "ddg", "var": "x", "confidence": "structural"}], + confidence="structural") + r = FlowResult(subgraph=_graph("a", "b", "c"), + evidence=[{"uri": "a", "file_line": "m.py:1", "code": "x = 1", "role": "seed"}], + _explain={"level": 4}, paths=[p]) + assert r.paths[0].confidence == "structural" + assert '"file_line": "m.py:1"' in r.to_json() From 8ecaad3b6459bd7936536a5b86da69db831f8139 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 16:28:55 -0400 Subject: [PATCH 02/19] fix(graph): serialize FlowResult paths in to_json (#270) --- cldk/graph/result.py | 7 ++++++- tests/graph/test_result.py | 11 +++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/cldk/graph/result.py b/cldk/graph/result.py index 9ec50af8..6f23df9a 100644 --- a/cldk/graph/result.py +++ b/cldk/graph/result.py @@ -1,6 +1,6 @@ from __future__ import annotations import json -from dataclasses import dataclass, field +from dataclasses import dataclass, field, asdict from typing import List, Dict, Literal import networkx as nx @@ -46,3 +46,8 @@ class SliceResult(GraphResult): @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) diff --git a/tests/graph/test_result.py b/tests/graph/test_result.py index 191cf9dc..c8080163 100644 --- a/tests/graph/test_result.py +++ b/tests/graph/test_result.py @@ -32,3 +32,14 @@ def test_flowresult_carries_paths_and_serializes(): _explain={"level": 4}, paths=[p]) assert r.paths[0].confidence == "structural" assert '"file_line": "m.py:1"' in r.to_json() + + +def test_flowresult_to_json_includes_paths(): + p = FlowPath(source="a", sink="c", + hops=[{"from": "a", "to": "b", "kind": "ddg", "var": "x", "confidence": "structural"}], + confidence="structural") + r = FlowResult(subgraph=_graph("a", "b", "c"), + evidence=[{"uri": "a"}], _explain={"level": 4}, paths=[p]) + dumped = r.to_json() + assert '"confidence": "structural"' in dumped + assert '"var": "x"' in dumped From 5e80c6d312265b83c3c7e964f63ca7ba94a716b2 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 16:33:12 -0400 Subject: [PATCH 03/19] feat(graph): provider ABC seam + polymorphic resolve_vertex (#270) --- cldk/graph/provider.py | 41 ++++++++++++++++++++++++++++++++++++ tests/graph/test_provider.py | 35 ++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+) create mode 100644 cldk/graph/provider.py create mode 100644 tests/graph/test_provider.py diff --git a/cldk/graph/provider.py b/cldk/graph/provider.py new file mode 100644 index 00000000..3c76a1d1 --- /dev/null +++ b/cldk/graph/provider.py @@ -0,0 +1,41 @@ +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.+?):(?P\d+)(?::(?P\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.""" + + @abstractmethod + def program_graph(self, callable_uri: str) -> nx.DiGraph: ... + @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 + return provider.resolve_location(m["file"], int(m["line"]), col) + raise ValueError(f"cannot resolve seed to a vertex: {seed!r} " + f"(expected 'file:line[:col]', a can:// id, or a body node)") diff --git a/tests/graph/test_provider.py b/tests/graph/test_provider.py new file mode 100644 index 00000000..5de36748 --- /dev/null +++ b/tests/graph/test_provider.py @@ -0,0 +1,35 @@ +import pytest +from cldk.graph.provider import resolve_vertex, ProgramGraphProvider + + +class FakeProvider(ProgramGraphProvider): + def program_graph(self, callable_uri): ... + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): + return [f"can://x/{file}/f@{line}:{col or 0}"] + def source_slice(self, vertex_uri): return ("m.py:1", "code") + def callable_of(self, vertex_uri): return "can://x/f" + def max_level(self): return 4 + + +def test_resolve_location_string(): + p = FakeProvider() + assert resolve_vertex(p, "src/m.py:42") == ["can://x/src/m.py/f@42:0"] + assert resolve_vertex(p, "src/m.py:42:5") == ["can://x/src/m.py/f@42:5"] + + +def test_resolve_can_id_passthrough(): + p = FakeProvider() + assert resolve_vertex(p, "can://x/src/m.py/f@42:5") == ["can://x/src/m.py/f@42:5"] + + +def test_resolve_node_object_uses_id(): + p = FakeProvider() + class N: id = "can://x/src/m.py/f@42:5" + assert resolve_vertex(p, N()) == ["can://x/src/m.py/f@42:5"] + + +def test_resolve_rejects_garbage(): + p = FakeProvider() + with pytest.raises(ValueError): + resolve_vertex(p, 12345) From c8bddd7d0bb38e3e7e6bdfde27d9cb40aa30e07a Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 16:40:51 -0400 Subject: [PATCH 04/19] feat(graph): capability gating with honest-degrade + strict (#270) --- cldk/graph/capability.py | 22 ++++++++++++++++++++++ tests/graph/test_capability.py | 23 +++++++++++++++++++++++ 2 files changed, 45 insertions(+) create mode 100644 cldk/graph/capability.py create mode 100644 tests/graph/test_capability.py diff --git a/cldk/graph/capability.py b/cldk/graph/capability.py new file mode 100644 index 00000000..a95f9993 --- /dev/null +++ b/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.", + } diff --git a/tests/graph/test_capability.py b/tests/graph/test_capability.py new file mode 100644 index 00000000..19253061 --- /dev/null +++ b/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) From eed15f74d7a7adbc689f549b09398c669059c2dc Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 16:46:52 -0400 Subject: [PATCH 05/19] feat(graph): engine intraprocedural slices + control_deps (#270) --- cldk/graph/engine.py | 61 ++++++++++++++++++++++++++++++++ tests/graph/test_engine_slice.py | 44 +++++++++++++++++++++++ 2 files changed, 105 insertions(+) create mode 100644 cldk/graph/engine.py create mode 100644 tests/graph/test_engine_slice.py diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py new file mode 100644 index 00000000..06abe986 --- /dev/null +++ b/cldk/graph/engine.py @@ -0,0 +1,61 @@ +# 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.DiGraph, families: Iterable[str]) -> nx.DiGraph: + fam = set(families) + out = nx.DiGraph() + out.add_nodes_from(g.nodes(data=True)) + for u, v, d in g.edges(data=True): + if d.get("family") in fam: + out.add_edge(u, v, **d) + return out + + +class Engine: + def __init__(self, provider: ProgramGraphProvider): + self.p = provider + + def _evidence(self, uris, seeds, roles=None): + 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, "def")}) + return ev + + def _intra(self, seed, edges, backward, strict, what) -> SliceResult: + note = require(3, self.p, strict=strict, what=what) + seeds = resolve_vertex(self.p, seed) + cal = self.p.callable_of(seeds[0]) + g = _filter_edges(self.p.program_graph(cal), edges) + 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) + sub = g.subgraph(reached).copy() + explain = {"seed": seeds, "direction": "backward" if backward else "forward", + "edges": list(edges), "level": self.p.max_level(), + "vertices": len(reached), "interprocedural": False} + if note: + explain["degraded"] = note + return SliceResult(subgraph=sub, evidence=self._evidence(reached, set(seeds)), + _explain=explain) + + def slice_backward(self, seed, *, edges=("cfg", "cdg", "ddg"), + interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult: + return self._intra(seed, edges, backward=True, strict=strict, what="slice_backward") + + def slice_forward(self, seed, *, edges=("cfg", "cdg", "ddg"), + interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult: + return self._intra(seed, edges, backward=False, strict=strict, what="slice_forward") + + def control_deps(self, seed, *, strict: bool = False) -> SliceResult: + return self._intra(seed, ("cdg",), backward=True, strict=strict, what="control_deps") diff --git a/tests/graph/test_engine_slice.py b/tests/graph/test_engine_slice.py new file mode 100644 index 00000000..314d5150 --- /dev/null +++ b/tests/graph/test_engine_slice.py @@ -0,0 +1,44 @@ +# tests/graph/test_engine_slice.py +import networkx as nx +from cldk.graph.engine import Engine +from cldk.graph.provider import ProgramGraphProvider + + +def _callable_graph(): + # entry -> s1(x=1) -> s2(y=x) -> s3(return y); ddg x:s1->s2, y:s2->s3 + g = nx.DiGraph() + for n in ["c@entry", "c@1:0", "c@2:0", "c@3:0"]: + g.add_node(n, kind="statement", span=None) + g.add_edge("c@entry", "c@1:0", family="cfg") + g.add_edge("c@1:0", "c@2:0", family="cfg") + g.add_edge("c@2:0", "c@3:0", family="cfg") + g.add_edge("c@1:0", "c@2:0", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@2:0", "c@3:0", family="ddg", var="y", prov=["points-to"]) + return g + + +class OneCallableProvider(ProgramGraphProvider): + def program_graph(self, callable_uri): return _callable_graph() + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): return [f"c@{line}:{col or 0}"] + def source_slice(self, vertex_uri): return (f"m:{vertex_uri}", vertex_uri) + def callable_of(self, vertex_uri): return "c" + def max_level(self): return 3 + + +def test_backward_slice_exact_set(): + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:3", edges=("cfg", "ddg")) # seed s3 + assert set(r.uris()) == {"c@3:0", "c@2:0", "c@1:0", "c@entry"} + + +def test_forward_slice_exact_set(): + e = Engine(OneCallableProvider()) + r = e.slice_forward("m:1", edges=("ddg",)) # seed s1, ddg only + assert set(r.uris()) == {"c@1:0", "c@2:0", "c@3:0"} + + +def test_ddg_only_backward_from_s3(): + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:3", edges=("ddg",)) + assert set(r.uris()) == {"c@3:0", "c@2:0", "c@1:0"} # follows ddg chain, not entry From 0e54293b50a8b053371a99787ecf533972d85836 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 17:31:21 -0400 Subject: [PATCH 06/19] fix(graph): MultiDiGraph program graph + seed-consistent slices (#270) --- cldk/graph/engine.py | 18 +++++++++------ tests/graph/test_engine_slice.py | 39 ++++++++++++++++++++++++++------ 2 files changed, 43 insertions(+), 14 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 06abe986..a6b16be4 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -7,13 +7,13 @@ from cldk.graph.result import SliceResult, FlowResult, FlowPath -def _filter_edges(g: nx.DiGraph, families: Iterable[str]) -> nx.DiGraph: +def _filter_edges(g: nx.MultiDiGraph, families: Iterable[str]) -> nx.MultiDiGraph: fam = set(families) - out = nx.DiGraph() + out = nx.MultiDiGraph() out.add_nodes_from(g.nodes(data=True)) - for u, v, d in g.edges(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, **d) + out.add_edge(u, v, key=k, **d) return out @@ -40,13 +40,17 @@ def _intra(self, seed, edges, backward, strict, what) -> SliceResult: for s in seeds: if s in walk: reached |= nx.descendants(walk, s) - sub = g.subgraph(reached).copy() + sub = g.subgraph(reached & set(g.nodes())).copy() + for s in seeds: # a seed is trivially in its own slice + if s not in sub: + sub.add_node(s, kind="seed") + ev_nodes = sorted(sub.nodes()) # evidence == subgraph nodes, deterministic explain = {"seed": seeds, "direction": "backward" if backward else "forward", "edges": list(edges), "level": self.p.max_level(), - "vertices": len(reached), "interprocedural": False} + "vertices": len(sub), "interprocedural": False} if note: explain["degraded"] = note - return SliceResult(subgraph=sub, evidence=self._evidence(reached, set(seeds)), + return SliceResult(subgraph=sub, evidence=self._evidence(ev_nodes, set(seeds)), _explain=explain) def slice_backward(self, seed, *, edges=("cfg", "cdg", "ddg"), diff --git a/tests/graph/test_engine_slice.py b/tests/graph/test_engine_slice.py index 314d5150..c4601aa7 100644 --- a/tests/graph/test_engine_slice.py +++ b/tests/graph/test_engine_slice.py @@ -5,15 +5,18 @@ def _callable_graph(): - # entry -> s1(x=1) -> s2(y=x) -> s3(return y); ddg x:s1->s2, y:s2->s3 - g = nx.DiGraph() + # entry -> s1(x=1) -> s2(y=x) -> s3(return y); ddg x:s1->s2, y:s2->s3. + # MultiDiGraph so the cfg fallthrough (s1->s2, s2->s3) and the ddg edges on the + # same statement pairs stay as DISTINCT parallel edges, each keeping its attrs. + g = nx.MultiDiGraph() for n in ["c@entry", "c@1:0", "c@2:0", "c@3:0"]: g.add_node(n, kind="statement", span=None) - g.add_edge("c@entry", "c@1:0", family="cfg") - g.add_edge("c@1:0", "c@2:0", family="cfg") - g.add_edge("c@2:0", "c@3:0", family="cfg") - g.add_edge("c@1:0", "c@2:0", family="ddg", var="x", prov=["ssa"]) - g.add_edge("c@2:0", "c@3:0", family="ddg", var="y", prov=["points-to"]) + g.add_edge("c@entry", "c@1:0", key="cfg", family="cfg") + g.add_edge("c@1:0", "c@2:0", key="cfg", family="cfg") + g.add_edge("c@2:0", "c@3:0", key="cfg", family="cfg") + g.add_edge("c@1:0", "c@2:0", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@2:0", "c@3:0", key="ddg", family="ddg", var="y", prov=["points-to"]) + assert g.number_of_edges() == 5 # genuine parallel edges, not overwrites return g @@ -42,3 +45,25 @@ def test_ddg_only_backward_from_s3(): e = Engine(OneCallableProvider()) r = e.slice_backward("m:3", edges=("ddg",)) assert set(r.uris()) == {"c@3:0", "c@2:0", "c@1:0"} # follows ddg chain, not entry + + +def test_family_scoped_slices_differ(): + # cfg and ddg share the endpoint pairs s1->s2 and s2->s3; a MultiDiGraph keeps + # them as distinct parallel edges, so family-scoped slices must NOT collapse. + e = Engine(OneCallableProvider()) + ddg_set = set(e.slice_backward("m:3", edges=("ddg",)).uris()) + cfg_set = set(e.slice_backward("m:3", edges=("cfg",)).uris()) + assert "c@entry" not in ddg_set # entry reachable only via the cfg chain + assert "c@entry" in cfg_set # cfg fallthrough reaches entry + assert ddg_set != cfg_set # families are distinct, not merged + + +def test_seed_absent_from_graph_is_consistent(): + # a seed resolving to a vertex not in the callable graph is still in its own + # slice; uris()/evidence must equal the subgraph's node set (no contradiction). + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:99", edges=("cfg", "ddg")) # c@99:0 is not a node + assert set(r.uris()) == set(r.subgraph.nodes()) + assert len(r) == r.subgraph.number_of_nodes() + assert "c@99:0" in set(r.uris()) # seed present in its own slice + assert bool(r) is True # non-empty; uris() and bool() agree From 4ecede2be14cd960c8866f62e3ad27ec731f14d9 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 17:37:53 -0400 Subject: [PATCH 07/19] feat(graph): flows_to witnesses + def_use with data-derived confidence (#270) --- cldk/graph/engine.py | 64 ++++++++++++++++++++++++++++++++ tests/graph/test_engine_flows.py | 24 ++++++++++++ 2 files changed, 88 insertions(+) create mode 100644 tests/graph/test_engine_flows.py diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index a6b16be4..d66f37d1 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -17,6 +17,18 @@ def _filter_edges(g: nx.MultiDiGraph, families: Iterable[str]) -> nx.MultiDiGrap return out +_TIER_RANK = {"unresolved": 0, "structural": 1, "resolved": 2} +_RANK_TIER = {v: k for k, v in _TIER_RANK.items()} + + +def _ddg_tier(prov) -> str: + if prov == ["points-to"]: + return "resolved" + if prov == ["ssa"]: + return "structural" + return "unresolved" + + class Engine: def __init__(self, provider: ProgramGraphProvider): self.p = provider @@ -63,3 +75,55 @@ def slice_forward(self, seed, *, edges=("cfg", "cdg", "ddg"), def control_deps(self, seed, *, strict: bool = False) -> SliceResult: return self._intra(seed, ("cdg",), backward=True, strict=strict, what="control_deps") + + def _dataflow_graph(self, callable_uri) -> nx.DiGraph: + # ddg (intra) plus summary/param_* (inter) at L4; here L3 uses ddg only + g = _filter_edges(self.p.program_graph(callable_uri), ("ddg",)) + if self.p.max_level() >= 4: + for e in self.p.sdg_edges(): + g.add_edge(e.src, e.dst, family="sdg", var=getattr(e, "var", None), + prov=getattr(e, "prov", [])) + return g + + def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResult: + note = require(3, self.p, strict=strict, what="flows_to") + src = resolve_vertex(self.p, source_seed)[0] + dst = resolve_vertex(self.p, sink_seed)[0] + g = self._dataflow_graph(self.p.callable_of(src)) + paths: List[FlowPath] = [] + reached = set() + if src in g and dst in g: + for path in nx.all_simple_paths(g, src, dst, cutoff=64): + 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) + hops.append({"from": a, "to": b, "kind": 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)} + if note: + explain["degraded"] = note + return FlowResult(subgraph=g.subgraph(reached).copy(), + evidence=self._evidence(reached, {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] + g = self._dataflow_graph(self.p.callable_of(s)) + reached = {s} | (nx.descendants(g, s) if s in g else set()) + explain = {"seed": s, "level": self.p.max_level(), "vertices": len(reached)} + if note: + explain["degraded"] = note + return FlowResult(subgraph=g.subgraph(reached).copy(), + evidence=self._evidence(reached, {s}), _explain=explain, paths=[]) diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py new file mode 100644 index 00000000..113892a7 --- /dev/null +++ b/tests/graph/test_engine_flows.py @@ -0,0 +1,24 @@ +# tests/graph/test_engine_flows.py +from cldk.graph.engine import Engine +from tests.graph.test_engine_slice import OneCallableProvider + + +def test_flows_to_finds_witness_with_min_confidence(): + e = Engine(OneCallableProvider()) + r = e.flows_to("m:1", "m:3") # s1 -> s2 (ssa) -> s3 (points-to); min = structural + assert bool(r) is True + assert len(r.paths) == 1 + assert [h["from"] for h in r.paths[0].hops] == ["c@1:0", "c@2:0"] + assert r.paths[0].confidence == "structural" + + +def test_flows_to_no_path_is_falsy(): + e = Engine(OneCallableProvider()) + r = e.flows_to("m:3", "m:1") # no forward ddg path s3 -> s1 + assert not r.paths + + +def test_def_use_returns_downstream_uses(): + e = Engine(OneCallableProvider()) + r = e.def_use("m:1") # def at s1 flows to s2, s3 + assert set(r.uris()) == {"c@1:0", "c@2:0", "c@3:0"} From 0fc1fdf271452409cd85d7dc49b43a959a93c069 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 17:56:21 -0400 Subject: [PATCH 08/19] fix(graph): dedup flows_to witnesses over parallel edges; def_use seed-consistency (#270) --- cldk/graph/engine.py | 21 ++++++++---- tests/graph/test_engine_flows.py | 55 +++++++++++++++++++++++++++++++- 2 files changed, 68 insertions(+), 8 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index d66f37d1..1a410783 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -93,7 +93,11 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul paths: List[FlowPath] = [] reached = set() if src in g and dst in g: - for path in nx.all_simple_paths(g, src, dst, cutoff=64): + # 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=64): hops, tiers = [], [] for a, b in zip(path, path[1:]): # MultiDiGraph: get_edge_data returns {key: attrdict} over parallel edges. @@ -113,17 +117,20 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul "paths": len(paths)} if note: explain["degraded"] = note - return FlowResult(subgraph=g.subgraph(reached).copy(), - evidence=self._evidence(reached, {src, dst}), _explain=explain, - paths=paths) + 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] g = self._dataflow_graph(self.p.callable_of(s)) reached = {s} | (nx.descendants(g, s) if s in g else set()) - explain = {"seed": s, "level": self.p.max_level(), "vertices": len(reached)} + 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=g.subgraph(reached).copy(), - evidence=self._evidence(reached, {s}), _explain=explain, paths=[]) + return FlowResult(subgraph=sub, evidence=self._evidence(sorted(sub.nodes()), {s}), + _explain=explain, paths=[]) diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py index 113892a7..abb752d1 100644 --- a/tests/graph/test_engine_flows.py +++ b/tests/graph/test_engine_flows.py @@ -1,14 +1,41 @@ # tests/graph/test_engine_flows.py +import networkx as nx from cldk.graph.engine import Engine +from cldk.graph.provider import ProgramGraphProvider from tests.graph.test_engine_slice import OneCallableProvider +class ParallelEdgeProvider(ProgramGraphProvider): + # ONE node route s1 -> s2 -> s3, but the s1->s2 hop carries TWO parallel ddg edges + # (var x via ssa, var y via points-to). A MultiDiGraph enumerates the route once per + # parallel edge; the engine must collapse to one witness per distinct node route. + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + for n in ["c@1:0", "c@2:0", "c@3:0"]: + g.add_node(n, kind="statement", span=None) + g.add_edge("c@1:0", "c@2:0", key="ddg:x", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@1:0", "c@2:0", key="ddg:y", family="ddg", var="y", prov=["points-to"]) + g.add_edge("c@2:0", "c@3:0", key="ddg", family="ddg", var="z", prov=["ssa"]) + return g + + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): return [f"c@{line}:{col or 0}"] + def source_slice(self, vertex_uri): return (f"m:{vertex_uri}", vertex_uri) + def callable_of(self, vertex_uri): return "c" + def max_level(self): return 3 + + def test_flows_to_finds_witness_with_min_confidence(): e = Engine(OneCallableProvider()) r = e.flows_to("m:1", "m:3") # s1 -> s2 (ssa) -> s3 (points-to); min = structural assert bool(r) is True assert len(r.paths) == 1 - assert [h["from"] for h in r.paths[0].hops] == ["c@1:0", "c@2:0"] + hops = r.paths[0].hops + assert [h["from"] for h in hops] == ["c@1:0", "c@2:0"] + assert [h["to"] for h in hops] == ["c@2:0", "c@3:0"] + assert [h["kind"] for h in hops] == ["ddg", "ddg"] + assert [h["var"] for h in hops] == ["x", "y"] + assert [h["confidence"] for h in hops] == ["structural", "resolved"] assert r.paths[0].confidence == "structural" @@ -16,9 +43,35 @@ def test_flows_to_no_path_is_falsy(): e = Engine(OneCallableProvider()) r = e.flows_to("m:3", "m:1") # no forward ddg path s3 -> s1 assert not r.paths + assert bool(r) is False + + +def test_flows_to_dedups_parallel_edge_routes(): + # one node route, two parallel ddg edges on the first hop -> exactly one witness. + e = Engine(ParallelEdgeProvider()) + r = e.flows_to("m:1", "m:3") + assert len(r.paths) == 1 + p = r.paths[0] + assert [h["from"] for h in p.hops] == ["c@1:0", "c@2:0"] + assert [h["to"] for h in p.hops] == ["c@2:0", "c@3:0"] + # per-hop confidence uses the BEST (max-tier) parallel: s1->s2 resolved (points-to + # beats ssa), s2->s3 structural. Path confidence is the min over hops. + assert [h["confidence"] for h in p.hops] == ["resolved", "structural"] + assert p.confidence == "structural" def test_def_use_returns_downstream_uses(): e = Engine(OneCallableProvider()) r = e.def_use("m:1") # def at s1 flows to s2, s3 assert set(r.uris()) == {"c@1:0", "c@2:0", "c@3:0"} + + +def test_def_use_seed_absent_is_consistent(): + # seed resolving to a vertex not in the dataflow graph is still in its own result; + # uris()/evidence must equal the subgraph node set (no uris/bool contradiction). + e = Engine(OneCallableProvider()) + r = e.def_use("m:99") # c@99:0 is not a node in the ddg graph + assert set(r.uris()) == set(r.subgraph.nodes()) + assert "c@99:0" in set(r.uris()) + assert bool(r) is True + assert len(r) == r.subgraph.number_of_nodes() From 48bf60dc296a90a80ea135c53912966549c922a6 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:11:23 -0400 Subject: [PATCH 09/19] feat(graph): level-driven interprocedural slice depth (#270) --- cldk/graph/engine.py | 29 +++++++++++++------ tests/graph/test_engine_interproc.py | 43 ++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 9 deletions(-) create mode 100644 tests/graph/test_engine_interproc.py diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 1a410783..aba0cfaa 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -42,24 +42,35 @@ def _evidence(self, uris, seeds, roles=None): "role": "seed" if u in seeds else roles.get(u, "def")}) return ev - def _intra(self, seed, edges, backward, strict, what) -> SliceResult: + def _intra(self, seed, edges, backward, strict, what, interprocedural=None) -> 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 seeds = resolve_vertex(self.p, seed) - cal = self.p.callable_of(seeds[0]) - g = _filter_edges(self.p.program_graph(cal), edges) + g = _filter_edges(self.p.program_graph(self.p.callable_of(seeds[0])), edges) + if want_inter and self.p.max_level() >= 4: + for e in self.p.sdg_edges(): + g.add_edge(e.src, e.dst, family="sdg", 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) - sub = g.subgraph(reached & set(g.nodes())).copy() - for s in seeds: # a seed is trivially in its own slice + # 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()) # evidence == subgraph nodes, deterministic + 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": False} + "vertices": len(sub), "interprocedural": bool(want_inter)} if note: explain["degraded"] = note return SliceResult(subgraph=sub, evidence=self._evidence(ev_nodes, set(seeds)), @@ -67,11 +78,11 @@ def _intra(self, seed, edges, backward, strict, what) -> SliceResult: def slice_backward(self, seed, *, edges=("cfg", "cdg", "ddg"), interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult: - return self._intra(seed, edges, backward=True, strict=strict, what="slice_backward") + 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, backward=False, strict=strict, what="slice_forward") + return self._intra(seed, edges, False, strict, "slice_forward", interprocedural) def control_deps(self, seed, *, strict: bool = False) -> SliceResult: return self._intra(seed, ("cdg",), backward=True, strict=strict, what="control_deps") diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py new file mode 100644 index 00000000..3c9d2d27 --- /dev/null +++ b/tests/graph/test_engine_interproc.py @@ -0,0 +1,43 @@ +# tests/graph/test_engine_interproc.py +import networkx as nx +from cldk.graph.engine import Engine +from cldk.graph.provider import ProgramGraphProvider +from cldk.graph.capability import CapabilityError +import pytest + + +class _Edge: + def __init__(self, src, dst): self.src, self.dst, self.var, self.prov = src, dst, "a", ["points-to"] + + +class TwoCallableProvider(ProgramGraphProvider): + # caller c: c@call --param_in--> callee d; d@ret --param_out--> c@after + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() # engine's _filter_edges iterates edges(keys=True) + if callable_uri == "c": + g.add_edge("c@call", "c@after", key="ddg", family="ddg", var="a", prov=["points-to"]) + else: + g.add_node("d@ret", kind="statement", span=None) + return g + def sdg_edges(self): return [_Edge("c@call", "d@in"), _Edge("d@ret", "c@after")] + def resolve_location(self, file, line, col=None): return [f"c@{line}"] + def source_slice(self, vertex_uri): return (vertex_uri, vertex_uri) + def callable_of(self, vertex_uri): return vertex_uri.split("@")[0] + def max_level(self): return 4 + + +def test_interproc_none_crosses_at_l4(): + e = Engine(TwoCallableProvider()) + # resolve_vertex only accepts a .id-bearing object, a can:// id, or a file:line[:col] + # string (see test_provider.py::test_resolve_node_object_uses_id for the same pattern) — + # "c@call" is a raw vertex id, so it must go through the .id-object path. + class Seed: id = "c@call" + r = e.slice_forward(Seed(), edges=("ddg",), interprocedural=None) + assert "d@in" in set(r.uris()) # crossed the param_in boundary + + +def test_explicit_interproc_on_l3_strict_raises(): + class L3(TwoCallableProvider): + def max_level(self): return 3 + with pytest.raises(CapabilityError): + Engine(L3()).slice_forward("c@call", interprocedural=True, strict=True) From 557573a875d6eb1ec8626edb264f0188802c37fa Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:28:07 -0400 Subject: [PATCH 10/19] =?UTF-8?q?fix(graph):=20control=5Fdeps=20is=20intra?= =?UTF-8?q?procedural=20=E2=80=94=20no=20sdg=20overlay=20at=20L4=20(#270)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cldk/graph/engine.py | 6 +++++- tests/graph/test_engine_interproc.py | 24 ++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index aba0cfaa..863b4ea6 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -85,7 +85,11 @@ def slice_forward(self, seed, *, edges=("cfg", "cdg", "ddg"), return self._intra(seed, edges, False, strict, "slice_forward", interprocedural) def control_deps(self, seed, *, strict: bool = False) -> SliceResult: - return self._intra(seed, ("cdg",), backward=True, strict=strict, what="control_deps") + # 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) def _dataflow_graph(self, callable_uri) -> nx.DiGraph: # ddg (intra) plus summary/param_* (inter) at L4; here L3 uses ddg only diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py index 3c9d2d27..ef3dc4f6 100644 --- a/tests/graph/test_engine_interproc.py +++ b/tests/graph/test_engine_interproc.py @@ -41,3 +41,27 @@ class L3(TwoCallableProvider): def max_level(self): return 3 with pytest.raises(CapabilityError): Engine(L3()).slice_forward("c@call", interprocedural=True, strict=True) + + +def test_control_deps_stays_intraprocedural_at_l4(): + # Control dependence has NO interprocedural notion in this model — only dataflow + # (param_in/param_out/summary) crosses callable boundaries. control_deps must force + # interprocedural=False; otherwise, on an L4 backend, _intra defaults to want_inter=True and + # merges the sdg dataflow overlay into a pure CDG slice, and the backward walk pulls in + # dataflow-reachable vertices from other callables with no control-dependence relation. + # + # control_deps is a BACKWARD slice, so a leaking sdg edge must be a forward-ANCESTOR edge of + # the seed: d@in -> c@body means d@in reaches c@body, so a backward slice from c@body WOULD + # pull in d@in if the overlay were applied (verified: it leaks against the unfixed code). + class CDGProvider(TwoCallableProvider): + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() # only a control-dependence edge + g.add_edge("c@guard", "c@body", key="cdg", family="cdg") + return g + def sdg_edges(self): return [_Edge("d@in", "c@body")] # cross-callable DATAFLOW + e = Engine(CDGProvider()) + class Seed: id = "c@body" + r = e.control_deps(Seed()) + assert set(r.uris()) == {"c@guard", "c@body"} # only intra cdg reachability, no d@in + assert "d@in" not in set(r.uris()) # sdg dataflow did NOT cross the boundary + assert r.explain()["interprocedural"] is False # control_deps is always intraprocedural From f9459a64d6618f68be541632543b7519ad7651df Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:50:24 -0400 Subject: [PATCH 11/19] fix(graph): gate the sdg overlay on the ddg edge family (#270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _intra applied the sdg (dataflow) overlay whenever the backend was L4 and interprocedural was wanted, ignoring the edges family filter — so a cfg- or cdg-only slice picked up dataflow vertices from foreign callables. Fold the family gate into want_inter so the overlay predicate and explain()["interprocedural"] stay one source of truth: no dataflow family requested means no boundary crossing. Subsumes the redundant max_level()>=4 check on the overlay branch; control_deps' explicit interprocedural=False stays as belt-and-suspenders. --- cldk/graph/engine.py | 7 ++++++- tests/graph/test_engine_interproc.py | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 863b4ea6..14218795 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -50,9 +50,14 @@ def _intra(self, seed, edges, backward, strict, what, interprocedural=None) -> S 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 and self.p.max_level() >= 4: + if want_inter: for e in self.p.sdg_edges(): g.add_edge(e.src, e.dst, family="sdg", var=getattr(e, "var", None), prov=getattr(e, "prov", [])) diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py index ef3dc4f6..4b8e6ad1 100644 --- a/tests/graph/test_engine_interproc.py +++ b/tests/graph/test_engine_interproc.py @@ -43,6 +43,26 @@ def max_level(self): return 3 Engine(L3()).slice_forward("c@call", interprocedural=True, strict=True) +def test_family_scoped_slice_has_no_sdg_overlay_at_l4(): + # C3: the sdg (dataflow: param_in/param_out/summary) overlay must be gated on the + # ddg family being REQUESTED, not just on level/interprocedural intent. A cfg-only + # backward slice on an L4 backend must not pull in dataflow vertices from other + # callables — only dataflow crosses boundaries, and no dataflow family was asked for. + class CFGProvider(TwoCallableProvider): + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + g.add_edge("c@1", "c@2", key="cfg", family="cfg") + g.add_edge("c@2", "c@3", key="cfg", family="cfg") + return g + def sdg_edges(self): return [_Edge("d@in", "c@2")] # foreign DATAFLOW vertex + e = Engine(CFGProvider()) + class Seed: id = "c@3" + r = e.slice_backward(Seed(), edges=("cfg",)) + assert "d@in" not in set(r.uris()) # no dataflow contamination + assert set(r.uris()) == {"c@1", "c@2", "c@3"} + assert r.explain()["interprocedural"] is False # no dataflow family => no crossing + + def test_control_deps_stays_intraprocedural_at_l4(): # Control dependence has NO interprocedural notion in this model — only dataflow # (param_in/param_out/summary) crosses callable boundaries. control_deps must force From a452296dbc60e58363940004f421149d5afc0fec Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:51:08 -0400 Subject: [PATCH 12/19] fix(graph): flows_to requires L4, honest-degrade to intra ddg at L3 (#270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit flows_to's full semantics are ddg + summary/param_in/param_out — an interprocedural (L4) capability — but it gated on require(3), so an L3 backend silently returned intra-only results as if they were complete. Raise the requirement to L4: non-strict now attaches the degraded note (absence is UNKNOWN, not safety) while still returning the intra ddg witnesses it can compute; strict=True raises CapabilityError. --- cldk/graph/engine.py | 5 ++++- tests/graph/test_engine_flows.py | 20 ++++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 14218795..1ff2ff77 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -106,7 +106,10 @@ def _dataflow_graph(self, callable_uri) -> nx.DiGraph: return g def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResult: - note = require(3, self.p, strict=strict, what="flows_to") + # 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] g = self._dataflow_graph(self.p.callable_of(src)) diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py index abb752d1..c1f75b90 100644 --- a/tests/graph/test_engine_flows.py +++ b/tests/graph/test_engine_flows.py @@ -1,7 +1,9 @@ # tests/graph/test_engine_flows.py import networkx as nx +import pytest from cldk.graph.engine import Engine from cldk.graph.provider import ProgramGraphProvider +from cldk.graph.capability import CapabilityError from tests.graph.test_engine_slice import OneCallableProvider @@ -60,6 +62,24 @@ def test_flows_to_dedups_parallel_edge_routes(): assert p.confidence == "structural" +def test_flows_to_on_l3_degrades_but_still_returns_intra_paths(): + # C1: full flows_to semantics are interprocedural (ddg + param_in/param_out/summary), + # which is L4. On an L3 backend the non-strict call must attach a degraded note AND + # still return the intraprocedural ddg witnesses it can compute — honest degrade, + # not silent completeness and not a refusal. + e = Engine(OneCallableProvider()) # L3 backend + r = e.flows_to("m:1", "m:3") + assert "degraded" in r.explain() + assert r.explain()["degraded"]["requested"] == 4 + assert len(r.paths) == 1 # intra ddg witness still computed + + +def test_flows_to_strict_on_l3_raises(): + e = Engine(OneCallableProvider()) # L3 backend + with pytest.raises(CapabilityError): + e.flows_to("m:1", "m:3", strict=True) + + def test_def_use_returns_downstream_uses(): e = Engine(OneCallableProvider()) r = e.def_use("m:1") # def at s1 flows to s2, s3 From ce2a5b3826ead24978995b44ca1018b94f505e70 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:52:10 -0400 Subject: [PATCH 13/19] fix(graph): flows_to spans source and sink callables (#270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit flows_to built its dataflow graph from the source's callable only, so a sink inside a different callable (reachable via param_in into the callee interior) was unreachable and reported a false "no flow". _dataflow_graph now unions the intra ddg graphs of the given callables before adding the sdg overlay, and flows_to passes both endpoint callables. def_use keeps its single-callable scope with a NOTE — interprocedural completeness lands with the whole-program dataflow graph (deferred to Task 7). --- cldk/graph/engine.py | 19 +++++++++++--- tests/graph/test_engine_interproc.py | 38 ++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 4 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 1ff2ff77..40c12e60 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -96,9 +96,15 @@ def control_deps(self, seed, *, strict: bool = False) -> SliceResult: return self._intra(seed, ("cdg",), backward=True, strict=strict, what="control_deps", interprocedural=False) - def _dataflow_graph(self, callable_uri) -> nx.DiGraph: - # ddg (intra) plus summary/param_* (inter) at L4; here L3 uses ddg only - g = _filter_edges(self.p.program_graph(callable_uri), ("ddg",)) + 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", var=getattr(e, "var", None), @@ -112,7 +118,10 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul 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] - g = self._dataflow_graph(self.p.callable_of(src)) + # 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() if src in g and dst in g: @@ -147,6 +156,8 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul 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() diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py index 4b8e6ad1..f9ecf137 100644 --- a/tests/graph/test_engine_interproc.py +++ b/tests/graph/test_engine_interproc.py @@ -43,6 +43,44 @@ def max_level(self): return 3 Engine(L3()).slice_forward("c@call", interprocedural=True, strict=True) +class _SDGEdge: + def __init__(self, src, dst, kind): + self.src, self.dst, self.kind = src, dst, kind + self.var, self.prov = "a", ["points-to"] + + +class CrossCallableFlowProvider(ProgramGraphProvider): + # caller c: c@src --ddg--> c@call; sdg: c@call --param_in--> d@in; + # callee d: d@in --ddg--> d@sink. The flow c@src -> d@sink exists only if the + # dataflow graph spans BOTH endpoint callables plus the sdg overlay. + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + if callable_uri == "c": + g.add_edge("c@src", "c@call", key="ddg", family="ddg", var="a", prov=["ssa"]) + else: + g.add_edge("d@in", "d@sink", key="ddg", family="ddg", var="a", prov=["ssa"]) + return g + def sdg_edges(self): return [_SDGEdge("c@call", "d@in", "param_in")] + def resolve_location(self, file, line, col=None): return [f"c@{line}"] + def source_slice(self, vertex_uri): return (vertex_uri, vertex_uri) + def callable_of(self, vertex_uri): return vertex_uri.split("@")[0] + def max_level(self): return 4 + + +def test_flows_to_crosses_callable_boundary(): + # C2: a sink in a DIFFERENT callable (reachable via param_in into the callee's + # interior) must be found. Building the dataflow graph from the source's callable + # alone loses the callee's intra ddg edges and yields a false "no flow". + e = Engine(CrossCallableFlowProvider()) + class Src: id = "c@src" + class Snk: id = "d@sink" + r = e.flows_to(Src(), Snk()) + assert len(r.paths) >= 1 # a real cross-callable flow, not empty + p = r.paths[0] + assert [h["from"] for h in p.hops] == ["c@src", "c@call", "d@in"] + assert [h["to"] for h in p.hops] == ["c@call", "d@in", "d@sink"] + + def test_family_scoped_slice_has_no_sdg_overlay_at_l4(): # C3: the sdg (dataflow: param_in/param_out/summary) overlay must be gated on the # ddg family being REQUESTED, not just on level/interprocedural intent. A cfg-only From 7eeb3297bdd7608466ede34a7272e00ffec4d71a Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:52:46 -0400 Subject: [PATCH 14/19] fix(graph): empty location resolution raises ValueError, not IndexError (#270) resolve_location legitimately returns [] when no vertex sits at the given line, but every engine verb indexes resolve_vertex(...)[0], turning an ordinary user miss into an IndexError. Raise a descriptive ValueError at the source in resolve_vertex instead, converting all [0] call sites into a clean failure mode. --- cldk/graph/provider.py | 7 ++++++- tests/graph/test_provider.py | 10 ++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/cldk/graph/provider.py b/cldk/graph/provider.py index 3c76a1d1..606e251c 100644 --- a/cldk/graph/provider.py +++ b/cldk/graph/provider.py @@ -36,6 +36,11 @@ def resolve_vertex(provider: ProgramGraphProvider, seed: Any) -> List[str]: m = _LOC.match(seed) if m: col = int(m["col"]) if m["col"] is not None else None - return provider.resolve_location(m["file"], int(m["line"]), col) + 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)") diff --git a/tests/graph/test_provider.py b/tests/graph/test_provider.py index 5de36748..39923aa3 100644 --- a/tests/graph/test_provider.py +++ b/tests/graph/test_provider.py @@ -33,3 +33,13 @@ def test_resolve_rejects_garbage(): p = FakeProvider() with pytest.raises(ValueError): resolve_vertex(p, 12345) + + +def test_resolve_location_with_no_vertex_raises(): + # I1: resolve_location legitimately returns [] when no vertex sits at that line. + # Every engine verb indexes resolve_vertex(...)[0], so [] must surface as a clean + # ValueError here — not an IndexError at the call site. + class EmptyProvider(FakeProvider): + def resolve_location(self, file, line, col=None): return [] + with pytest.raises(ValueError, match="no vertex at location"): + resolve_vertex(EmptyProvider(), "m.py:99") From a076a761ecf242b42479e2b0df71100787f07138 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:53:20 -0400 Subject: [PATCH 15/19] fix(graph): MultiDiGraph annotations and provider contract docstrings (#270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The graph is an nx.MultiDiGraph everywhere, but three annotations still said nx.DiGraph: ProgramGraphProvider.program_graph, GraphResult.subgraph (and engine._dataflow_graph, already corrected in the C2 commit). Fix the first two and document the provider contract: parallel cfg/cdg/ddg edges between the same vertex pair must stay distinct edges with their own family/var/prov/kind, and sub-L3 providers still answer the structural methods — the engine handles level gating via require(...). --- cldk/graph/provider.py | 10 ++++++++-- cldk/graph/result.py | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/cldk/graph/provider.py b/cldk/graph/provider.py index 606e251c..eb5ea3e6 100644 --- a/cldk/graph/provider.py +++ b/cldk/graph/provider.py @@ -9,10 +9,16 @@ 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.""" + (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.DiGraph: ... + 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 diff --git a/cldk/graph/result.py b/cldk/graph/result.py index 6f23df9a..5d61be0d 100644 --- a/cldk/graph/result.py +++ b/cldk/graph/result.py @@ -17,7 +17,7 @@ class FlowPath: @dataclass class GraphResult: - subgraph: nx.DiGraph + subgraph: nx.MultiDiGraph evidence: List[Dict] _explain: Dict From ef9df6d393037ec2710408132a2b64d84bd9f12a Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:54:14 -0400 Subject: [PATCH 16/19] fix(graph): preserve sdg edge kind in flow witnesses (#270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sdg overlay dropped e.kind (param_in/param_out/summary) on the floor in both _intra and _dataflow_graph, so a FlowPath hop crossing a callable boundary could only say "sdg" — not which boundary edge carried the flow. Carry kind on the overlay edges and have the hop dict prefer the edge kind over the family: intra hops still report cfg/cdg/ddg, boundary hops now report the concrete sdg kind. --- cldk/graph/engine.py | 13 ++++++++----- tests/graph/test_engine_interproc.py | 15 +++++++++++++++ 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 40c12e60..39074a21 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -59,8 +59,8 @@ def _intra(self, seed, edges, backward, strict, what, interprocedural=None) -> S 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", var=getattr(e, "var", None), - prov=getattr(e, "prov", [])) + 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: @@ -107,8 +107,8 @@ def _dataflow_graph(self, *callable_uris) -> nx.MultiDiGraph: 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", var=getattr(e, "var", None), - prov=getattr(e, "prov", [])) + 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: @@ -140,7 +140,10 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul key=lambda d: _TIER_RANK[_ddg_tier(d.get("prov", []))]) t = _ddg_tier(best.get("prov", [])) tiers.append(t) - hops.append({"from": a, "to": b, "kind": best.get("family"), + # 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)) diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py index f9ecf137..7e70419a 100644 --- a/tests/graph/test_engine_interproc.py +++ b/tests/graph/test_engine_interproc.py @@ -81,6 +81,21 @@ class Snk: id = "d@sink" assert [h["to"] for h in p.hops] == ["c@call", "d@in", "d@sink"] +def test_flow_boundary_hop_reports_sdg_kind(): + # I6: a hop crossing the callable boundary must report WHICH sdg edge carried the + # flow (param_in/param_out/summary), not the opaque family name "sdg". Intra hops + # keep reporting their family ("ddg"). + e = Engine(CrossCallableFlowProvider()) + class Src: id = "c@src" + class Snk: id = "d@sink" + p = e.flows_to(Src(), Snk()).paths[0] + kinds = [h["kind"] for h in p.hops] + assert kinds[0] == "ddg" # intra hop: family + assert kinds[1] in {"param_in", "param_out", "summary"} # boundary hop: sdg kind + assert kinds[1] == "param_in" + assert kinds[2] == "ddg" + + def test_family_scoped_slice_has_no_sdg_overlay_at_l4(): # C3: the sdg (dataflow: param_in/param_out/summary) overlay must be gated on the # ddg family being REQUESTED, not just on level/interprocedural intent. A cfg-only From 78c196f41f698c2cfb43a311450d2185202f83ba Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:55:31 -0400 Subject: [PATCH 17/19] =?UTF-8?q?fix(graph):=20per-verb=20evidence=20roles?= =?UTF-8?q?=20=E2=80=94=20control=20and=20use,=20not=20always=20def=20(#27?= =?UTF-8?q?0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _evidence stamped every non-seed vertex "def", which is wrong for control_deps (the guard CONTROLS the seed) and def_use (downstream vertices are USES of the definition). Thread a default_role through _evidence and _intra: slices and flows keep "def", control_deps passes "control", def_use passes "use"; seeds are always "seed". --- cldk/graph/engine.py | 20 ++++++++++++++------ tests/graph/test_engine_flows.py | 10 ++++++++++ tests/graph/test_engine_interproc.py | 16 ++++++++++++++++ tests/graph/test_engine_slice.py | 10 ++++++++++ 4 files changed, 50 insertions(+), 6 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 39074a21..03ffa91f 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -33,16 +33,19 @@ class Engine: def __init__(self, provider: ProgramGraphProvider): self.p = provider - def _evidence(self, uris, seeds, roles=None): + 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, "def")}) + "role": "seed" if u in seeds else roles.get(u, default_role)}) return ev - def _intra(self, seed, edges, backward, strict, what, interprocedural=None) -> SliceResult: + 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: @@ -78,7 +81,9 @@ def _intra(self, seed, edges, backward, strict, what, interprocedural=None) -> S "vertices": len(sub), "interprocedural": bool(want_inter)} if note: explain["degraded"] = note - return SliceResult(subgraph=sub, evidence=self._evidence(ev_nodes, set(seeds)), + 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"), @@ -94,7 +99,8 @@ def control_deps(self, seed, *, strict: bool = False) -> SliceResult: # 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) + 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_* @@ -169,5 +175,7 @@ def def_use(self, seed, *, strict: bool = False) -> FlowResult: 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}), + return FlowResult(subgraph=sub, + evidence=self._evidence(sorted(sub.nodes()), {s}, + default_role="use"), _explain=explain, paths=[]) diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py index c1f75b90..bdb1bb32 100644 --- a/tests/graph/test_engine_flows.py +++ b/tests/graph/test_engine_flows.py @@ -86,6 +86,16 @@ def test_def_use_returns_downstream_uses(): assert set(r.uris()) == {"c@1:0", "c@2:0", "c@3:0"} +def test_def_use_evidence_role_is_use(): + # I4: downstream vertices in a def_use result are USES of the seed's definition. + e = Engine(OneCallableProvider()) + r = e.def_use("m:1") + roles = {ev["uri"]: ev["role"] for ev in r.evidence} + assert roles["c@1:0"] == "seed" + assert roles["c@2:0"] == "use" + assert roles["c@3:0"] == "use" + + def test_def_use_seed_absent_is_consistent(): # seed resolving to a vertex not in the dataflow graph is still in its own result; # uris()/evidence must equal the subgraph node set (no uris/bool contradiction). diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py index 7e70419a..3f4f1f72 100644 --- a/tests/graph/test_engine_interproc.py +++ b/tests/graph/test_engine_interproc.py @@ -138,3 +138,19 @@ class Seed: id = "c@body" assert set(r.uris()) == {"c@guard", "c@body"} # only intra cdg reachability, no d@in assert "d@in" not in set(r.uris()) # sdg dataflow did NOT cross the boundary assert r.explain()["interprocedural"] is False # control_deps is always intraprocedural + + +def test_control_deps_evidence_role_is_control(): + # I4: a non-seed vertex in a control_deps result is there as a controlling guard, + # not as a definition — its evidence role must say so. + class CDGProvider(TwoCallableProvider): + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + g.add_edge("c@guard", "c@body", key="cdg", family="cdg") + return g + e = Engine(CDGProvider()) + class Seed: id = "c@body" + r = e.control_deps(Seed()) + roles = {ev["uri"]: ev["role"] for ev in r.evidence} + assert roles["c@body"] == "seed" + assert roles["c@guard"] == "control" diff --git a/tests/graph/test_engine_slice.py b/tests/graph/test_engine_slice.py index c4601aa7..d183995f 100644 --- a/tests/graph/test_engine_slice.py +++ b/tests/graph/test_engine_slice.py @@ -58,6 +58,16 @@ def test_family_scoped_slices_differ(): assert ddg_set != cfg_set # families are distinct, not merged +def test_slice_evidence_default_role_is_def(): + # I4 guard: slices keep the "def" role for non-seed vertices (only control_deps + # and def_use re-role their evidence). + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:3", edges=("ddg",)) + roles = {ev["uri"]: ev["role"] for ev in r.evidence} + assert roles["c@3:0"] == "seed" + assert roles["c@1:0"] == "def" and roles["c@2:0"] == "def" + + def test_seed_absent_from_graph_is_consistent(): # a seed resolving to a vertex not in the callable graph is still in its own # slice; uris()/evidence must equal the subgraph's node set (no contradiction). From 069950e2bbeb6df5e3cff17112c70f770a407456 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:56:36 -0400 Subject: [PATCH 18/19] fix(graph): bound flows_to witness enumeration, report truncation (#270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit flows_to enumerated simple paths with a hard-coded cutoff=64 and no path cap, and never told the caller when witnesses were dropped. Hoist the bounds to module constants (_PATH_CUTOFF=64, _MAX_PATHS=1000 — provisional, to be tuned on real graphs), stop collecting at _MAX_PATHS, and surface explain()["truncated"] so a partial witness set is never silently presented as complete. --- cldk/graph/engine.py | 16 +++++++++++-- tests/graph/test_engine_flows.py | 39 ++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index 03ffa91f..e8c0b4f8 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -20,6 +20,14 @@ def _filter_edges(g: nx.MultiDiGraph, families: Iterable[str]) -> nx.MultiDiGrap _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: if prov == ["points-to"]: @@ -130,12 +138,16 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul 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=64): + 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. @@ -155,7 +167,7 @@ def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResul 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)} + "paths": len(paths), "truncated": truncated} if note: explain["degraded"] = note sub = g.subgraph(reached).copy() diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py index bdb1bb32..5539313e 100644 --- a/tests/graph/test_engine_flows.py +++ b/tests/graph/test_engine_flows.py @@ -80,6 +80,45 @@ def test_flows_to_strict_on_l3_raises(): e.flows_to("m:1", "m:3", strict=True) +class DiamondProvider(ProgramGraphProvider): + # TWO distinct node routes: c@s -> c@a -> c@t and c@s -> c@b -> c@t. + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + g.add_edge("c@s", "c@a", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@a", "c@t", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@s", "c@b", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@b", "c@t", key="ddg", family="ddg", var="x", prov=["ssa"]) + return g + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): return [f"c@{line}"] + def source_slice(self, vertex_uri): return (vertex_uri, vertex_uri) + def callable_of(self, vertex_uri): return "c" + def max_level(self): return 4 + + +def test_flows_to_sets_truncated_when_path_cap_hit(monkeypatch): + # I5: witness enumeration is bounded; when the cap drops paths the result must SAY + # so via explain()["truncated"], instead of silently presenting a partial set as + # complete. Shrink the cap to 1 so the diamond's second route is dropped. + import cldk.graph.engine as eng + monkeypatch.setattr(eng, "_MAX_PATHS", 1) + e = Engine(DiamondProvider()) + class S: id = "c@s" + class T: id = "c@t" + r = e.flows_to(S(), T()) + assert len(r.paths) == 1 # capped at _MAX_PATHS + assert r.explain()["truncated"] is True + + +def test_flows_to_not_truncated_within_bounds(): + e = Engine(DiamondProvider()) + class S: id = "c@s" + class T: id = "c@t" + r = e.flows_to(S(), T()) + assert len(r.paths) == 2 # both diamond routes enumerated + assert r.explain()["truncated"] is False + + def test_def_use_returns_downstream_uses(): e = Engine(OneCallableProvider()) r = e.def_use("m:1") # def at s1 flows to s2, s3 From bfe7fa71025f25f36a5abb63710565a1618a51ef Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Tue, 14 Jul 2026 18:57:23 -0400 Subject: [PATCH 19/19] fix(graph): _ddg_tier ranks by prov membership, not exact-list match (#270) prov is a provenance set in list form; the exact-list comparisons (prov == ["points-to"] / == ["ssa"]) made the STRONGER combined provenance ["ssa", "points-to"] fall through to "unresolved". Rank by membership: points-to anywhere means resolved, else ssa means structural, else unresolved. --- cldk/graph/engine.py | 7 +++++-- tests/graph/test_engine_flows.py | 12 +++++++++++- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py index e8c0b4f8..673900bb 100644 --- a/cldk/graph/engine.py +++ b/cldk/graph/engine.py @@ -30,9 +30,12 @@ def _filter_edges(g: nx.MultiDiGraph, families: Iterable[str]) -> nx.MultiDiGrap def _ddg_tier(prov) -> str: - if prov == ["points-to"]: + # 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 prov == ["ssa"]: + if "ssa" in prov: return "structural" return "unresolved" diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py index 5539313e..3b6b59ae 100644 --- a/tests/graph/test_engine_flows.py +++ b/tests/graph/test_engine_flows.py @@ -1,12 +1,22 @@ # tests/graph/test_engine_flows.py import networkx as nx import pytest -from cldk.graph.engine import Engine +from cldk.graph.engine import Engine, _ddg_tier from cldk.graph.provider import ProgramGraphProvider from cldk.graph.capability import CapabilityError from tests.graph.test_engine_slice import OneCallableProvider +def test_ddg_tier_uses_membership_not_exact_list(): + # I8: prov is a provenance SET in list form. ["ssa", "points-to"] carries strictly + # MORE evidence than ["points-to"] alone and must rank "resolved" — an exact-list + # comparison would let the stronger provenance fall through to "unresolved". + assert _ddg_tier(["ssa", "points-to"]) == "resolved" + assert _ddg_tier(["points-to"]) == "resolved" + assert _ddg_tier(["ssa"]) == "structural" + assert _ddg_tier([]) == "unresolved" + + class ParallelEdgeProvider(ProgramGraphProvider): # ONE node route s1 -> s2 -> s3, but the s1->s2 hop carries TWO parallel ddg edges # (var x via ssa, var y via points-to). A MultiDiGraph enumerates the route once per