fix(dataflow): connect the L4 SDG port layer to the statement ddg

.claude/SCHEMA_DECISIONS.md

@@ -202,3 +202,31 @@ MERGE collapses legitimately-distinct edges (per-variable dependences; a
 conditional's true/false pair) and a live Bolt push then materializes fewer
 relationships than the projection produced (caught by the opt-in
 `test_neo4j_bolt.py` count gates).
+
+## L4 graph completeness — port wiring + call anchoring (#115, 1.1.1)
+
+Two connectivity gaps closed in the L4 emission (no vocabulary invented; both
+decisions use surface the keystone already ships):
+
+1. **Statement ↔ port ddg wiring, `prov:["reaching-defs"]`.** The SDG's
+ binding edges — `def stmt → actual_in`, `actual_out → callsite`,
+ `formal_in → first use`, `def/return → formal_out` — existed in the IR
+ (`fg.extra_edges`, wired by `assemble_sdg`) and were emitted by the old v1
+ `program_graphs` projection, but the v2 emission dropped them, leaving the
+ port lattice an island (no end-to-end `flows_to` witness could cross a
+ call). `emit_l4` now emits them onto each callable's `ddg` tagged
+ `prov:["reaching-defs"]` — the label codeanalyzer-typescript already ships
+ for its port-routing edges, so the prov vocabulary stays keystone-shared:
+ `ssa` (L3 syntactic) ⊂ + `points-to` (L4 alias delta) + `reaching-defs`
+ (L4 port bindings). Monotonicity: both L4 families are additive over the
+ untouched ssa set, and every port endpoint exists only at L4.
+2. **Call vertices anchor via `parent`, not the CFG spine.** A call nested in
+ a larger statement (`y = f(x)`) keeps its own `"line:col"` body key and
+ deliberately stays OFF the cfg spine — calls are dataflow satellites of
+ their statement, not control-flow steps. From L3 (when statements exist)
+ such a call carries `parent` = its enclosing statement's local id — the
+ same anchoring `actual_in`/`actual_out` vertices already use. A bare-call
+ statement shares its key with the CFG node (no self-parent). This is a
+ sanctioned `null → value` refinement of `BodyNode.parent` at the L2→L3
+ boundary, mirroring the `callee: null → id` refinement at L1→L2; the
+ superset gates compare body keys, so no gate exception was needed.

CHANGELOG.md

@@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- **L4 SDG port layer is connected to the statement ddg** (#115): the binding
+ edges `def stmt → actual_in`, `actual_out → callsite`, `formal_in → first use`
+ and `return → formal_out` were built by the SDG assembler but dropped by the
+ v2 emission, leaving the interprocedural port lattice an island — no
+ end-to-end `flows_to(def, callee_formal)` path could cross a call. They are
+ now emitted on each callable's `ddg` with `prov:["reaching-defs"]` (the same
+ label codeanalyzer-typescript ships for its port-routing edges). Strictly
+ additive over the L3 `ssa` set, so `L3 ⊆ L4` monotonicity holds.
+- **Nested call vertices are anchored to their statement** (#115): a call inside
+ a larger statement (`y = f(x)`) sits off the CFG spine by design (a dataflow
+ satellite); from L3 it now carries `parent` = its enclosing statement's local
+ id — the same anchoring `actual_in`/`actual_out` vertices use. Sanctioned
+ `null → value` refinement at L2→L3, mirroring `callee: null → id` at L1→L2;
+ recorded in `.claude/SCHEMA_DECISIONS.md`.
+
 ## [1.1.0] - 2026-07-27
 
 ### Changed

codeanalyzer/dataflow/builder.py

@@ -283,6 +283,23 @@ def _span_of(source: str, node) -> Optional["Span"]:
  continue
  pycallable.body[local] = BodyNode(kind=node.kind, span=span)
 
+ # #115: anchor nested call vertices to their statement. A bare-call
+ # statement shares its key with its CFG node (handled above); a call
+ # nested inside a larger statement (`y = f(x)`) has its own key and
+ # no cfg contact, so it carries `parent` = the enclosing statement's
+ # local id — the same anchoring actual_in/actual_out vertices use.
+ for node in pdg.cfg.nodes:
+ if node.ast_node is None:
+ continue
+ stmt_local = im.local(node.id)
+ for call in _calls_in(node.ast_node):
+ call_key = f"{call.lineno}:{call.col_offset}"
+ child = pycallable.body.get(call_key)
+ if child is None or call_key == stmt_local:
+ continue
+ if child.kind == "call":
+ child.parent = stmt_local
+
  if want_cfg:
  pycallable.cfg = [
  CfgEdge(
@@ -437,7 +454,12 @@ def emit_l4(
  points-to provenance and taint are *not* emitted here (later tasks).
  """
  from codeanalyzer.dataflow.identity import IdentityMap
- from codeanalyzer.schema.py_schema import BodyNode, ParamEdge, SummaryEdge
+ from codeanalyzer.schema.py_schema import (
+ BodyNode,
+ DdgEdge,
+ ParamEdge,
+ SummaryEdge,
+ )
 
  # L4 emission is additive (it *appends* summary/param edges), so it must
  # first clear any L4 state a reused cache left on these live objects —
@@ -517,6 +539,41 @@ def emit_l4(
  )
  (app.param_in if e.type == "PARAM_IN" else app.param_out).append(edge)
 
+ # (e) statement ↔ port ddg wiring (#115). The IR's ``extra_edges`` — the
+ # def→actual_in, actual_out→callsite, formal_in→use and def→formal_out
+ # bindings ``assemble_sdg`` wires — are what connect the port lattice to
+ # the statement-level ddg; without them the SDG is two disconnected
+ # graphs and no end-to-end flows_to walk can cross a call. Emitted with
+ # ``prov=["reaching-defs"]`` (the label codeanalyzer-typescript ships for
+ # its port-routing ddg edges, so the vocabulary stays keystone-shared).
+ # CDG-typed extras (callsite → actual_in containment) are skipped — the
+ # actual vertices already carry that anchoring in ``parent``.
+ for sig, fg in ir.functions.items():
+ pycallable = sig_to_callable.get(sig)
+ im = ims.get(sig)
+ if pycallable is None or im is None:
+ continue
+ # Idempotency under cache reuse, mirroring the points-to delta: strip
+ # any reaching-defs edges a prior run appended before re-emitting.
+ pycallable.ddg = [e for e in pycallable.ddg if e.prov != ["reaching-defs"]]
+ seen: set = set()
+ rows = []
+ for e in fg.extra_edges:
+ if e.type != "DDG":
+ continue
+ src, dst = im.local(e.source), im.local(e.target)
+ if src not in pycallable.body or dst not in pycallable.body:
+ continue
+ key = (src, dst, e.var)
+ if key in seen:
+ continue
+ seen.add(key)
+ rows.append(
+ DdgEdge(src=src, dst=dst, var=e.var, prov=["reaching-defs"])
+ )
+ rows.sort(key=lambda r: (r.src, r.dst, r.var or ""))
+ pycallable.ddg.extend(rows)
+
 
 def _ddg_local_set(im, pdg) -> Set[Tuple[str, str, Optional[str]]]:
  """The DDG edges of ``pdg`` as a set of ``(local_src, local_dst, var)``.

test/conftest_v2.py

@@ -68,14 +68,16 @@ def assert_conformant(payload: dict, max_level: int) -> None:
  f"L3 ddg edge must have prov ['ssa'], got {e.get('prov')} in {c['id']}"
  )
  elif max_level >= 4:
- # L4 layers an alias-derived (points-to) def-use delta additively on top
- # of the unchanged L3 ssa edges, so every ddg edge carries exactly one of
- # those two provenances — and no other.
+ # L4 layers two additive deltas on the unchanged L3 ssa edges: the
+ # alias-derived points-to def-use delta, and the statement ↔ port
+ # binding edges (#115) tagged reaching-defs — the keystone-shared label
+ # codeanalyzer-typescript also emits for its port-routing edges. Every
+ # ddg edge carries exactly one of those three provenances — no other.
  for mod, c in _iter_callables(app):
  for e in c.get("ddg", []):
- assert e.get("prov") in (["ssa"], ["points-to"]), (
- f"L4 ddg edge must have prov ['ssa'] or ['points-to'], "
- f"got {e.get('prov')} in {c['id']}"
+ assert e.get("prov") in (["ssa"], ["points-to"], ["reaching-defs"]), (
+ f"L4 ddg edge must have prov ['ssa'], ['points-to'] or "
+ f"['reaching-defs'], got {e.get('prov')} in {c['id']}"
  )
 
  if max_level >= 4:

test/test_v2_l4_ports.py

@@ -0,0 +1,200 @@
+"""Regression tests for #115: the L4 SDG port layer must be *connected* to the
+statement-level ddg, and call vertices must be anchored to their statement.
+
+Before the fix, the interprocedural port lattice (``actual_in → formal_in``,
+``formal_out → actual_out`` via param_in/param_out/summary) was an island: no
+ddg edge touched any port, so an end-to-end ``flows_to(def, callee_formal)``
+walk was inexpressible. The wiring existed in the IR (``fg.extra_edges``,
+emitted by the old v1 ``program_graphs`` projection) but the v2 emission
+dropped it. The restored edges carry ``prov=["reaching-defs"]`` — the same
+label codeanalyzer-typescript ships for its port-routing ddg edges.
+"""
+from pathlib import Path
+
+from codeanalyzer.dataflow.builder import (
+ _base_types,
+ build_function_pdgs,
+ build_program_graphs,
+ emit_ddg_pointsto_delta,
+ emit_l3_body,
+ emit_l4,
+)
+from codeanalyzer.dataflow.scalpel_oracle import make_alias_oracle
+from codeanalyzer.dataflow.syntactic import SyntacticOracle
+from codeanalyzer.schema import PyApplication
+from codeanalyzer.schema.assign_ids import assign_ids
+from codeanalyzer.schema.l1_body import populate_l1_body
+from codeanalyzer.schema.py_schema import PyCallEdge
+from codeanalyzer.syntactic_analysis.symbol_table_builder import SymbolTableBuilder
+
+_SOURCE = """\
+def build(flag):
+ result = flag + 1
+ return result
+
+
+def main():
+ x = 5
+ y = build(x)
+ z = y * 2
+ return z
+"""
+
+
+def _build_l4_app(tmp_path: Path):
+ f = tmp_path / "app.py"
+ f.write_text(_SOURCE, encoding="utf-8")
+ mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f)
+ app = PyApplication(symbol_table={"app.py": mod})
+ sig_to_id = assign_ids(app, "portfix")
+ app.call_graph = [
+ PyCallEdge(src="app.main", dst="app.build", prov=["jedi"], weight=1)
+ ]
+ populate_l1_body(app)
+ syn_infos, _ = build_function_pdgs(
+ app, k=3, oracle_factory=lambda c, fast: SyntacticOracle()
+ )
+ emit_l3_body(app, syn_infos, sig_to_id, graphs={"cfg", "dfg", "pdg"})
+ ir = build_program_graphs(
+ app, k=3,
+ oracle_factory=lambda c, fast: make_alias_oracle(c, fast, _base_types(c)),
+ )
+ emit_l4(app, ir, sig_to_id)
+ emit_ddg_pointsto_delta(app, syn_infos, ir, sig_to_id)
+ mod = app.symbol_table["app.py"]
+ return app, mod.functions["build"], mod.functions["main"]
+
+
+def _edges(c, prov=None):
+ out = set()
+ for e in c.ddg or []:
+ if prov is None or e.prov == prov:
+ out.add((e.src, e.dst, e.var))
+ return out
+
+
+def test_def_stmt_flows_into_actual_in(tmp_path):
+ """`x = 5` (7:4) must feed the argument port of the `build(x)` callsite."""
+ _, _, main = _build_l4_app(tmp_path)
+ rd = _edges(main, prov=["reaching-defs"])
+ assert any(
+ src == "7:4" and dst.endswith("/actual_in:0") for src, dst, _ in rd
+ ), f"missing def→actual_in binding edge; reaching-defs edges: {sorted(rd)}"
+
+
+def test_actual_out_flows_back_to_callsite(tmp_path):
+ """The return-value port must flow back into the callsite statement."""
+ _, _, main = _build_l4_app(tmp_path)
+ rd = _edges(main, prov=["reaching-defs"])
+ assert any(
+ src.endswith("/actual_out") and dst == "8:4" for src, dst, _ in rd
+ ), f"missing actual_out→use binding edge; reaching-defs edges: {sorted(rd)}"
+
+
+def test_formal_in_flows_to_first_use(tmp_path):
+ """Inside the callee, the parameter port must reach its first-use stmt."""
+ _, build, _ = _build_l4_app(tmp_path)
+ rd = _edges(build, prov=["reaching-defs"])
+ assert any(
+ src == "@formal_in:0" and dst == "2:4" for src, dst, _ in rd
+ ), f"missing formal_in→use edge; reaching-defs edges: {sorted(rd)}"
+
+
+def test_return_stmt_flows_into_formal_out(tmp_path):
+ """`return result` (3:4) must feed the callee's formal_out port."""
+ _, build, _ = _build_l4_app(tmp_path)
+ rd = _edges(build, prov=["reaching-defs"])
+ assert any(
+ src == "3:4" and dst == "@formal_out" for src, dst, _ in rd
+ ), f"missing return→formal_out edge; reaching-defs edges: {sorted(rd)}"
+
+
+def test_port_edges_never_replace_or_retag_l3_edges(tmp_path):
+ """The wiring is additive: every ssa edge survives, and no reaching-defs
+ edge duplicates an (src, dst, var) triple that ssa already carries."""
+ _, build, main = _build_l4_app(tmp_path)
+ for c in (build, main):
+ ssa = _edges(c, prov=["ssa"])
+ rd = _edges(c, prov=["reaching-defs"])
+ assert ssa, "L3 ssa edges must still be present"
+ assert not (ssa & rd), "reaching-defs must not duplicate ssa triples"
+
+
+def test_port_edge_endpoints_exist_in_body(tmp_path):
+ _, build, main = _build_l4_app(tmp_path)
+ for c in (build, main):
+ for e in c.ddg or []:
+ assert e.src in c.body, f"dangling ddg src {e.src}"
+ assert e.dst in c.body, f"dangling ddg dst {e.dst}"
+
+
+def test_emission_is_idempotent_under_reemit(tmp_path):
+ """Re-running emit_l4 + the delta against the same live tree (cache-reuse
+ shape) must not duplicate the reaching-defs edges."""
+ f = tmp_path / "app.py"
+ f.write_text(_SOURCE, encoding="utf-8")
+ mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f)
+ app = PyApplication(symbol_table={"app.py": mod})
+ sig_to_id = assign_ids(app, "portfix")
+ app.call_graph = [
+ PyCallEdge(src="app.main", dst="app.build", prov=["jedi"], weight=1)
+ ]
+ populate_l1_body(app)
+ syn_infos, _ = build_function_pdgs(
+ app, k=3, oracle_factory=lambda c, fast: SyntacticOracle()
+ )
+ emit_l3_body(app, syn_infos, sig_to_id, graphs={"cfg", "dfg", "pdg"})
+ ir = build_program_graphs(
+ app, k=3,
+ oracle_factory=lambda c, fast: make_alias_oracle(c, fast, _base_types(c)),
+ )
+ emit_l4(app, ir, sig_to_id)
+ emit_ddg_pointsto_delta(app, syn_infos, ir, sig_to_id)
+ main = app.symbol_table["app.py"].functions["main"]
+ first = sorted((e.src, e.dst, e.var, tuple(e.prov)) for e in main.ddg)
+ emit_l4(app, ir, sig_to_id)
+ emit_ddg_pointsto_delta(app, syn_infos, ir, sig_to_id)
+ main = app.symbol_table["app.py"].functions["main"]
+ second = sorted((e.src, e.dst, e.var, tuple(e.prov)) for e in main.ddg)
+ assert first == second, "re-emission must be idempotent"
+
+
+# ----------------------------------------------------------------------------------------------
+# #115 part 2: call vertices anchored to their statement via `parent`.
+# ----------------------------------------------------------------------------------------------
+
+
+def test_nested_call_vertex_is_parented_to_its_statement(tmp_path):
+ """`y = build(x)`: the call vertex (8:8) floats off the CFG spine; from L3
+ it must carry `parent` = its enclosing statement's local (8:4)."""
+ _, _, main = _build_l4_app(tmp_path)
+ call_nodes = {k: n for k, n in main.body.items() if n.kind == "call"}
+ assert call_nodes, "fixture must materialize a call vertex"
+ for key, node in call_nodes.items():
+ assert node.parent == "8:4", (
+ f"call vertex {key} must be parented to its statement, "
+ f"got parent={node.parent!r}"
+ )
+
+
+def test_bare_call_statement_needs_no_parent(tmp_path):
+ """A bare call (`g(b)`) shares its key with the statement node — no
+ self-parent is emitted."""
+ f = tmp_path / "m.py"
+ f.write_text(
+ "def g(x):\n return x\n\n\ndef f(a):\n g(a)\n return a\n",
+ encoding="utf-8",
+ )
+ mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f)
+ app = PyApplication(symbol_table={"m.py": mod})
+ sig_to_id = assign_ids(app, "barefix")
+ populate_l1_body(app)
+ syn_infos, _ = build_function_pdgs(
+ app, k=3, oracle_factory=lambda c, fast: SyntacticOracle()
+ )
+ emit_l3_body(app, syn_infos, sig_to_id, graphs={"cfg", "dfg", "pdg"})
+ fcallable = app.symbol_table["m.py"].functions["f"]
+ call_nodes = {k: n for k, n in fcallable.body.items() if n.kind == "call"}
+ assert call_nodes, "fixture must materialize the bare call vertex"
+ for key, node in call_nodes.items():
+ assert node.parent != key, "a call must never parent to itself"