feat(dataflow): stage 8b — CPG projection through the Neo4j emitter

CFGNode label (merge key id = <signature>#<node_id>) carrying both CFG
statements and HRB parameter nodes, plus the shared cross-language edge
vocabulary HAS_CFG_NODE / CFG_NEXT / CDG / DDG / PARAM_IN / PARAM_OUT /
SUMMARY (deliberately unprefixed — parity clause). Additive
schema.neo4j.json bump to 1.2.0; sample app extended so the conformance
tests exercise every new row family; count-parity and no-dangling gates
on the real fixture at -a 3. CALL stays at the callable level (PY_CALLS
twin). (#67)

Author: rahlk

codeanalyzer/neo4j/project.py

@@ -74,9 +74,105 @@ def project(app: PyApplication, app_name: str) -> GraphRows:
             "PY_CALLS", src, tgt, _call_edge_props(e.weight, list(e.provenance or []))
         )
 
+    # Level-3 CPG overlay (present only at -a 3): the same program_graphs IR
+    # projected as :CFGNode nodes and the shared cross-language edge types.
+    if app.program_graphs is not None:
+        _project_program_graphs(b, app)
+
     return b.finish()
 
 
+# ----------------------------------------------------------------------------------------------
+# Level-3 CPG overlay
+# ----------------------------------------------------------------------------------------------
+
+
+def _signature_modules(app: PyApplication) -> dict:
+    """signature → owning module file_key, for CFGNode `_module` provenance."""
+    from codeanalyzer.semantic_analysis.call_graph import _walk_module_callables
+
+    out: dict = {}
+    for file_key, mod in app.symbol_table.items():
+        for c in _walk_module_callables(mod):
+            out[c.signature] = file_key
+    return out
+
+
+def _cfg_node_ref(b: RowBuilder, sig: str, node_id: int) -> NodeRef:
+    return NodeRef("CFGNode", "id", f"{sig}#{node_id}")
+
+
+def _project_program_graphs(b: RowBuilder, app: PyApplication) -> None:
+    """CFG/PDG/SDG rows: node label ``CFGNode`` (merge key ``id`` =
+    ``<signature>#<node_id>``) and edge types ``HAS_CFG_NODE`` / ``CFG_NEXT``
+    (prop ``kind``) / ``CDG`` / ``DDG`` (prop ``var``) / ``PARAM_IN`` /
+    ``PARAM_OUT`` / ``SUMMARY`` — the shared cross-language vocabulary, so no
+    ``PY_`` prefix. Parameter nodes ride the same label with their HRB kinds
+    plus ``var``/``call_node`` props (an additive, recorded extension)."""
+    pg = app.program_graphs
+    sig_module = _signature_modules(app)
+
+    for sig, fg in pg.functions.items():
+        owner = _sym(sig)
+        module = sig_module.get(sig)
+        for n in (fg.cfg.nodes if fg.cfg else []):
+            ref = b.node(
+                ["CFGNode"],
+                "id",
+                f"{sig}#{n.id}",
+                prune(
+                    {
+                        "kind": n.kind,
+                        "start_line": n.start_line,
+                        "end_line": n.end_line,
+                        "_module": module,
+                    }
+                ),
+            )
+            b.edge("HAS_CFG_NODE", owner, ref)
+        for p in fg.param_nodes or []:
+            ref = b.node(
+                ["CFGNode"],
+                "id",
+                f"{sig}#{p.id}",
+                prune(
+                    {
+                        "kind": p.kind,
+                        "var": p.var,
+                        "call_node": p.call_node,
+                        "start_line": p.start_line,
+                        "end_line": p.end_line,
+                        "_module": module,
+                    }
+                ),
+            )
+            b.edge("HAS_CFG_NODE", owner, ref)
+        for e in (fg.cfg.edges if fg.cfg else []):
+            b.edge(
+                "CFG_NEXT",
+                _cfg_node_ref(b, sig, e.source),
+                _cfg_node_ref(b, sig, e.target),
+                {"kind": e.kind},
+            )
+        for e in (fg.pdg.edges if fg.pdg else []):
+            b.edge(
+                e.type,  # CDG | DDG
+                _cfg_node_ref(b, sig, e.source),
+                _cfg_node_ref(b, sig, e.target),
+                prune({"var": e.var}),
+            )
+
+    for e in pg.sdg_edges:
+        if e.type == "CALL":
+            continue  # the callable-level PY_CALLS twin already carries calls
+        b.edge(
+            e.type,  # PARAM_IN | PARAM_OUT | SUMMARY
+            _cfg_node_ref(b, e.source.signature, e.source.node),
+            _cfg_node_ref(b, e.target.signature, e.target.node),
+            prune({"var": e.var}),
+        )
+
+
 def _sym(signature: str) -> NodeRef:
     return NodeRef("PySymbol", "signature", signature)
 

codeanalyzer/neo4j/schema.py

@@ -35,7 +35,7 @@
 from dataclasses import dataclass, field
 from typing import Dict, List
 
-SCHEMA_VERSION = "1.1.0"
+SCHEMA_VERSION = "1.2.0"
 
 # PropType ∈ {"string", "integer", "float", "boolean", "string[]", "integer[]"}.
 
@@ -176,6 +176,23 @@ class RelType:
             "_module": "string",
         },
     ),
+    # Level-3 CPG overlay (present only at -a 3). The label and edge types
+    # below are the shared cross-language dataflow vocabulary — deliberately
+    # NOT PY_-prefixed. `id` = "<signature>#<node_id>"; parameter-passing
+    # nodes (formal/actual in/out) ride the same label with `var`/`call_node`.
+    NodeLabel(
+        "CFGNode",
+        "CFGNode",
+        "id",
+        {
+            "id": "string",
+            "kind": "string",
+            "var": "string",
+            "call_node": "integer",
+            **_SPAN,
+            "_module": "string",
+        },
+    ),
 ]
 
 _DECL_TARGETS = ["PyClass", "PyCallable"]
@@ -203,6 +220,14 @@ class RelType:
         {"imported_names": "string[]", "aliases": "string[]"},
     ),
     RelType("PY_DECORATED_BY", ["PyCallable"], ["PyDecorator"]),
+    # Level-3 CPG overlay (shared cross-language vocabulary, -a 3 only).
+    RelType("HAS_CFG_NODE", ["PyCallable"], ["CFGNode"]),
+    RelType("CFG_NEXT", ["CFGNode"], ["CFGNode"], {"kind": "string"}),
+    RelType("CDG", ["CFGNode"], ["CFGNode"]),
+    RelType("DDG", ["CFGNode"], ["CFGNode"], {"var": "string"}),
+    RelType("PARAM_IN", ["CFGNode"], ["CFGNode"], {"var": "string"}),
+    RelType("PARAM_OUT", ["CFGNode"], ["CFGNode"], {"var": "string"}),
+    RelType("SUMMARY", ["CFGNode"], ["CFGNode"]),
 ]
 
 

schema.neo4j.json

@@ -1,5 +1,5 @@
 {
-  "schema_version": "1.1.0",
+  "schema_version": "1.2.0",
   "generator": "codeanalyzer-python",
   "marker_labels": [],
   "node_labels": [
@@ -135,6 +135,20 @@
         "end_line": "integer",
         "_module": "string"
       }
+    },
+    {
+      "label": "CFGNode",
+      "merge_label": "CFGNode",
+      "key": "id",
+      "properties": {
+        "id": "string",
+        "kind": "string",
+        "var": "string",
+        "call_node": "integer",
+        "start_line": "integer",
+        "end_line": "integer",
+        "_module": "string"
+      }
     }
   ],
   "relationship_types": [
@@ -260,6 +274,84 @@
         "PyDecorator"
       ],
       "properties": {}
+    },
+    {
+      "type": "HAS_CFG_NODE",
+      "from": [
+        "PyCallable"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {}
+    },
+    {
+      "type": "CFG_NEXT",
+      "from": [
+        "CFGNode"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {
+        "kind": "string"
+      }
+    },
+    {
+      "type": "CDG",
+      "from": [
+        "CFGNode"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {}
+    },
+    {
+      "type": "DDG",
+      "from": [
+        "CFGNode"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {
+        "var": "string"
+      }
+    },
+    {
+      "type": "PARAM_IN",
+      "from": [
+        "CFGNode"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {
+        "var": "string"
+      }
+    },
+    {
+      "type": "PARAM_OUT",
+      "from": [
+        "CFGNode"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {
+        "var": "string"
+      }
+    },
+    {
+      "type": "SUMMARY",
+      "from": [
+        "CFGNode"
+      ],
+      "to": [
+        "CFGNode"
+      ],
+      "properties": {}
     }
   ],
   "constraints": [
@@ -270,7 +362,8 @@
     "CREATE CONSTRAINT pydecorator_name IF NOT EXISTS FOR (x:PyDecorator) REQUIRE x.name IS UNIQUE",
     "CREATE CONSTRAINT pycallsite_id IF NOT EXISTS FOR (x:PyCallSite) REQUIRE x.id IS UNIQUE",
     "CREATE CONSTRAINT pyattribute_id IF NOT EXISTS FOR (x:PyAttribute) REQUIRE x.id IS UNIQUE",
-    "CREATE CONSTRAINT pyvariable_id IF NOT EXISTS FOR (x:PyVariable) REQUIRE x.id IS UNIQUE"
+    "CREATE CONSTRAINT pyvariable_id IF NOT EXISTS FOR (x:PyVariable) REQUIRE x.id IS UNIQUE",
+    "CREATE CONSTRAINT cfgnode_id IF NOT EXISTS FOR (x:CFGNode) REQUIRE x.id IS UNIQUE"
   ],
   "indexes": [
     "CREATE INDEX py_callable_name IF NOT EXISTS FOR (c:PyCallable) ON (c.name)",

test/sample_graph_app.py

@@ -11,12 +11,22 @@
 from codeanalyzer.schema import (
     PyApplication,
     PyCallable,
+    PyCFG,
+    PyCFGEdge,
     PyClass,
     PyClassAttribute,
     PyComment,
     PyExternalSymbol,
+    PyFunctionGraphs,
+    PyGraphNode,
     PyImport,
     PyModule,
+    PyParamNode,
+    PyPDG,
+    PyPDGEdge,
+    PyProgramGraphs,
+    PySDGEdge,
+    PySDGEndpoint,
     PyVariableDeclaration,
 )
 from codeanalyzer.schema.py_schema import PyCallEdge, PyCallsite
@@ -147,10 +157,90 @@ def make_sample_app() -> PyApplication:
         ),
     ]
 
+    # A miniature level-3 section exercising every CPG row family:
+    # helper's CFG (entry → callsite stmt → exit), a CDG/DDG pair, its HRB
+    # parameter nodes, and PARAM_IN/PARAM_OUT/SUMMARY edges into announce.
+    helper_graphs = PyFunctionGraphs(
+        cfg=PyCFG(
+            nodes=[
+                PyGraphNode(id=0, kind="entry", start_line=17, end_line=17),
+                PyGraphNode(id=1, kind="statement", start_line=18, end_line=18),
+                PyGraphNode(id=2, kind="exit", start_line=20, end_line=20),
+            ],
+            edges=[
+                PyCFGEdge(source=0, target=1, kind="fallthrough"),
+                PyCFGEdge(source=1, target=2, kind="return"),
+                PyCFGEdge(source=1, target=2, kind="exception"),
+            ],
+        ),
+        pdg=PyPDG(
+            edges=[
+                PyPDGEdge(source=0, target=1, type="CDG"),
+                PyPDGEdge(source=0, target=1, type="DDG", var="url"),
+            ]
+        ),
+        param_nodes=[
+            PyParamNode(id=3, kind="formal_out", var="<return>", start_line=20, end_line=20),
+            PyParamNode(id=4, kind="actual_in", var="self", call_node=1, start_line=18, end_line=18),
+            PyParamNode(id=5, kind="actual_out", var="<return>", call_node=1, start_line=18, end_line=18),
+        ],
+    )
+    announce_graphs = PyFunctionGraphs(
+        cfg=PyCFG(
+            nodes=[
+                PyGraphNode(id=0, kind="entry", start_line=10, end_line=10),
+                PyGraphNode(id=1, kind="return", start_line=11, end_line=11),
+                PyGraphNode(id=2, kind="exit", start_line=12, end_line=12),
+            ],
+            edges=[
+                PyCFGEdge(source=0, target=1, kind="fallthrough"),
+                PyCFGEdge(source=1, target=2, kind="return"),
+            ],
+        ),
+        pdg=PyPDG(edges=[PyPDGEdge(source=0, target=1, type="CDG")]),
+        param_nodes=[
+            PyParamNode(id=3, kind="formal_in", var="self", start_line=10, end_line=10),
+            PyParamNode(id=4, kind="formal_out", var="<return>", start_line=12, end_line=12),
+        ],
+    )
+    program_graphs = PyProgramGraphs(
+        schema_version="1.0.0",
+        k_limit=3,
+        functions={
+            "src.service.helper": helper_graphs,
+            "src.service.Service.announce": announce_graphs,
+        },
+        sdg_edges=[
+            PySDGEdge(
+                source=PySDGEndpoint(signature="src.service.helper", node=1),
+                target=PySDGEndpoint(signature="src.service.Service.announce", node=0),
+                type="CALL",
+            ),
+            PySDGEdge(
+                source=PySDGEndpoint(signature="src.service.helper", node=4),
+                target=PySDGEndpoint(signature="src.service.Service.announce", node=3),
+                type="PARAM_IN",
+                var="self",
+            ),
+            PySDGEdge(
+                source=PySDGEndpoint(signature="src.service.Service.announce", node=4),
+                target=PySDGEndpoint(signature="src.service.helper", node=5),
+                type="PARAM_OUT",
+                var="<return>",
+            ),
+            PySDGEdge(
+                source=PySDGEndpoint(signature="src.service.helper", node=4),
+                target=PySDGEndpoint(signature="src.service.helper", node=5),
+                type="SUMMARY",
+            ),
+        ],
+    )
+
     return PyApplication(
         symbol_table={"src/service.py": service_mod, "src/util.py": util_mod},
         call_graph=call_graph,
         # The ghost edge's target (requests.get) is a library member, recorded as a
         # first-class external symbol so the projection emits a :PyExternal for it.
         external_symbols={"requests.get": PyExternalSymbol(name="get", module="requests")},
+        program_graphs=program_graphs,
     )