feat(dataflow): program_graphs emission, -a 3, --graphs, --graph-field-depth

program_graphs schema section (PyProgramGraphs and friends, versioned
1.0.0 independently of the application schema) attached to
PyApplication; -a extended to 3 (cumulative: level 3 keeps PyCG
enrichment); --graphs cfg,dfg,pdg,sdg selector with strict validation
(unknown values and level<3 usage exit non-zero, never silently fall
back); --graph-field-depth k-limit knob recorded in the output. -a 1/2
emit no program_graphs and their pipeline is untouched. (#67)

Author: rahlk

codeanalyzer/__main__.py

@@ -114,11 +114,31 @@ def main(
         typer.Option(
             "-a",
             "--analysis-level",
-            help="Analysis depth: 1=symbol table+Jedi call graph, 2=+PyCG call graph.",
+            help="Analysis depth: 1=symbol table+Jedi call graph, 2=+PyCG call "
+            "graph, 3=+native dataflow graphs (CFG/PDG/SDG).",
             min=1,
-            max=2,
+            max=3,
         ),
     ] = 1,
+    graphs: Annotated[
+        str,
+        typer.Option(
+            "--graphs",
+            help="Level 3 only: comma-separated program-graph sections to emit "
+            "(cfg, dfg, pdg, sdg). Default: all. `dfg` emits the PDG's data "
+            "edges only; `sdg` implies the dependence edges it stitches.",
+        ),
+    ] = "cfg,dfg,pdg,sdg",
+    graph_field_depth: Annotated[
+        int,
+        typer.Option(
+            "--graph-field-depth",
+            help="Level 3 only: k-limit on access-path depth (x.f.g.h with "
+            "k=3 becomes x.f.g.*). Mandatory bound — it is what guarantees "
+            "the interprocedural fixpoint terminates.",
+            min=1,
+        ),
+    ] = 3,
     using_ray: Annotated[
         bool,
         typer.Option("--ray/--no-ray", help="Enable Ray for distributed analysis."),
@@ -243,6 +263,27 @@ def main(
         ),
     ] = 50,
 ):
+    # Flag validation (strict: unrecognized values error out, never fall back).
+    selected_graphs = [g.strip() for g in graphs.split(",") if g.strip()]
+    from codeanalyzer.dataflow.builder import VALID_GRAPHS
+
+    unknown_graphs = [g for g in selected_graphs if g not in VALID_GRAPHS]
+    if unknown_graphs:
+        logger.error(
+            f"Unrecognized --graphs value(s): {', '.join(unknown_graphs)} "
+            f"(valid: {', '.join(VALID_GRAPHS)})."
+        )
+        raise typer.Exit(code=2)
+    if not selected_graphs:
+        logger.error("--graphs requires at least one of: " + ", ".join(VALID_GRAPHS))
+        raise typer.Exit(code=2)
+    if analysis_level < 3 and graphs != "cfg,dfg,pdg,sdg":
+        logger.error("--graphs is a level-3 option; pass -a 3 to emit program graphs.")
+        raise typer.Exit(code=2)
+    if analysis_level < 3 and graph_field_depth != 3:
+        logger.error("--graph-field-depth is a level-3 option; pass -a 3.")
+        raise typer.Exit(code=2)
+
     options = AnalysisOptions(
         input=input,
         output=output,
@@ -254,6 +295,8 @@ def main(
         neo4j_password=neo4j_password,
         neo4j_database=neo4j_database,
         analysis_level=analysis_level,
+        graphs=",".join(selected_graphs),
+        graph_field_depth=graph_field_depth,
         using_ray=using_ray,
         rebuild_analysis=rebuild_analysis,
         skip_tests=skip_tests,

codeanalyzer/core.py

@@ -454,10 +454,28 @@ def analyze(self) -> PyApplication:
             .external_symbols(external_symbols)
             .build()
         )
-        
+
+        if self.analysis_level >= 3:
+            # Level 3: native dataflow graphs (CFG/PDG/SDG) over the same
+            # signatures, gated so -a 1/-a 2 timings stay untouched.
+            from codeanalyzer.dataflow.builder import (
+                build_program_graphs,
+                to_program_graphs,
+            )
+
+            t0_l3 = time.perf_counter()
+            ir = build_program_graphs(app, k=self.options.graph_field_depth)
+            app.program_graphs = to_program_graphs(
+                ir, set(self.options.graphs.split(","))
+            )
+            logger.info(
+                "✅ Program graphs: %d functions, %d SDG edges in %.1fs",
+                len(ir.functions), len(ir.sdg_edges), time.perf_counter() - t0_l3,
+            )
+
         # Save to cache
         self._save_analysis_cache(app, cache_file)
-        
+
         return app
 
     def _load_pyapplication_from_cache(self, cache_file: Path) -> PyApplication:

codeanalyzer/dataflow/builder.py

@@ -253,3 +253,101 @@ def build_program_graphs(
 
     summaries = compute_summaries(infos, sorted(set(call_edges)))
     return assemble_sdg(infos, summaries, k)
+
+
+VALID_GRAPHS = ("cfg", "dfg", "pdg", "sdg")
+
+
+def to_program_graphs(ir: ProgramGraphsIR, graphs: Set[str]):
+    """Project the IR onto the ``program_graphs`` schema section, scoped by
+    the ``--graphs`` selector. ``dfg`` emits the PDG's DDG edges only;
+    ``sdg`` implies the dependence edges it is stitched over."""
+    from codeanalyzer.schema.py_schema import (
+        PyCFG,
+        PyCFGEdge,
+        PyFunctionGraphs,
+        PyGraphNode,
+        PyParamNode,
+        PyPDG,
+        PyPDGEdge,
+        PyProgramGraphs,
+        PySDGEdge,
+        PySDGEndpoint,
+    )
+
+    want_pdg = bool({"pdg", "sdg"} & graphs)
+    want_dfg = want_pdg or "dfg" in graphs
+    functions: Dict[str, "PyFunctionGraphs"] = {}
+    for sig in sorted(ir.functions):
+        fg = ir.functions[sig]
+        out = PyFunctionGraphs()
+        if "cfg" in graphs:
+            out.cfg = PyCFG(
+                nodes=[
+                    PyGraphNode(
+                        id=n.id,
+                        kind=n.kind,
+                        start_line=n.start_line,
+                        end_line=n.end_line,
+                        start_column=n.start_column,
+                        end_column=n.end_column,
+                    )
+                    for n in fg.pdg.cfg.nodes
+                ],
+                edges=[
+                    PyCFGEdge(source=e.source, target=e.target, kind=e.kind)
+                    for e in fg.pdg.cfg.edges
+                ],
+            )
+        edges: List["PyPDGEdge"] = []
+        if want_pdg:
+            edges.extend(
+                PyPDGEdge(source=e.source, target=e.target, type="CDG")
+                for e in fg.pdg.edges
+                if e.type == "CDG"
+            )
+        if want_dfg:
+            edges.extend(
+                PyPDGEdge(source=e.source, target=e.target, type="DDG", var=e.var)
+                for e in fg.ddg
+            )
+            edges.extend(
+                PyPDGEdge(source=e.source, target=e.target, type=e.type, var=e.var)
+                for e in fg.extra_edges
+                if e.type == "DDG" or want_pdg
+            )
+        if edges:
+            edges.sort(key=lambda e: (e.source, e.target, e.type, e.var or ""))
+            out.pdg = PyPDG(edges=edges)
+        if "sdg" in graphs:
+            out.param_nodes = [
+                PyParamNode(
+                    id=p.id,
+                    kind=p.kind,
+                    var=p.var,
+                    call_node=p.call_node,
+                    start_line=p.start_line,
+                    end_line=p.end_line,
+                )
+                for p in fg.param_nodes
+            ]
+        functions[sig] = out
+
+    sdg_edges = []
+    if "sdg" in graphs:
+        sdg_edges = [
+            PySDGEdge(
+                source=PySDGEndpoint(signature=e.source_sig, node=e.source_node),
+                target=PySDGEndpoint(signature=e.target_sig, node=e.target_node),
+                type=e.type,
+                var=e.var,
+            )
+            for e in ir.sdg_edges
+        ]
+
+    return PyProgramGraphs(
+        schema_version="1.0.0",
+        k_limit=ir.k_limit,
+        functions=functions,
+        sdg_edges=sdg_edges,
+    )

codeanalyzer/options/options.py

@@ -49,6 +49,10 @@ class AnalysisOptions:
     neo4j_password: str = "neo4j"
     neo4j_database: Optional[str] = None
     analysis_level: int = 1
+    # Level-3 dataflow knobs: which program graphs to emit (csv of
+    # cfg|dfg|pdg|sdg) and the access-path k-limit.
+    graphs: str = "cfg,dfg,pdg,sdg"
+    graph_field_depth: int = 3
     using_ray: bool = False
     rebuild_analysis: bool = False
     skip_tests: bool = True

codeanalyzer/schema/__init__.py

@@ -5,12 +5,22 @@
     PyApplication,
     PyCallable,
     PyCallableParameter,
+    PyCFG,
+    PyCFGEdge,
     PyClass,
     PyClassAttribute,
     PyComment,
     PyExternalSymbol,
+    PyFunctionGraphs,
+    PyGraphNode,
     PyImport,
     PyModule,
+    PyParamNode,
+    PyPDG,
+    PyPDGEdge,
+    PyProgramGraphs,
+    PySDGEdge,
+    PySDGEndpoint,
     PyVariableDeclaration,
 )
 
@@ -25,6 +35,16 @@
     "PyCallable",
     "PyClassAttribute",
     "PyCallableParameter",
+    "PyGraphNode",
+    "PyCFGEdge",
+    "PyPDGEdge",
+    "PyParamNode",
+    "PyCFG",
+    "PyPDG",
+    "PyFunctionGraphs",
+    "PySDGEndpoint",
+    "PySDGEdge",
+    "PyProgramGraphs",
 ]
 
 try:

codeanalyzer/schema/py_schema.py

@@ -369,6 +369,135 @@ class PyExternalSymbol(BaseModel):
     module: Optional[str] = None  # best-effort owning module, e.g. "requests"
 
 
+@builder
+@msgpk
+class PyGraphNode(BaseModel):
+    """A CFG node of one callable's level-3 graphs. ``id`` is the source-span
+    order index within the callable (synthetic ENTRY = 0, EXIT = last CFG id);
+    ``(signature, id)`` is the cross-section join key."""
+
+    id: int
+    kind: Literal[
+        "entry", "exit", "statement", "branch", "loop", "return", "raise", "handler"
+    ] = "statement"
+    start_line: int = -1
+    end_line: int = -1
+    start_column: int = -1
+    end_column: int = -1
+
+
+@builder
+@msgpk
+class PyCFGEdge(BaseModel):
+    """Control-flow successor edge (shared cross-language kind vocabulary)."""
+
+    source: int
+    target: int
+    kind: Literal[
+        "fallthrough",
+        "true",
+        "false",
+        "switch_case",
+        "loop_back",
+        "exception",
+        "return",
+        "break",
+        "continue",
+        "yield",
+        "await_resume",
+    ] = "fallthrough"
+
+
+@builder
+@msgpk
+class PyPDGEdge(BaseModel):
+    """Dependence edge: control (``CDG``) or data (``DDG``, labeled with the
+    k-limited access path being read)."""
+
+    source: int
+    target: int
+    type: Literal["CDG", "DDG"] = "DDG"
+    var: Optional[str] = None
+
+
+@builder
+@msgpk
+class PyParamNode(BaseModel):
+    """HRB parameter-passing node, sharing the owning callable's id space
+    (allocated after EXIT). ``call_node`` is the owning callsite statement for
+    actuals; ``var`` is the parameter name, ``<return>``, ``<capture>:name``,
+    or ``<global>:module::name``."""
+
+    id: int
+    kind: Literal["formal_in", "formal_out", "actual_in", "actual_out"]
+    var: str
+    call_node: Optional[int] = None
+    start_line: int = -1
+    end_line: int = -1
+
+
+@builder
+@msgpk
+class PyCFG(BaseModel):
+    """One callable's control-flow graph."""
+
+    nodes: List[PyGraphNode] = []
+    edges: List[PyCFGEdge] = []
+
+
+@builder
+@msgpk
+class PyPDG(BaseModel):
+    """One callable's dependence edges (over the same node ids as the CFG
+    plus its parameter nodes)."""
+
+    edges: List[PyPDGEdge] = []
+
+
+@builder
+@msgpk
+class PyFunctionGraphs(BaseModel):
+    """The per-callable level-3 sections, keyed by signature."""
+
+    cfg: Optional[PyCFG] = None
+    pdg: Optional[PyPDG] = None
+    param_nodes: List[PyParamNode] = []
+
+
+@builder
+@msgpk
+class PySDGEndpoint(BaseModel):
+    """A ``(signature, node)`` reference into a function's emitted graphs."""
+
+    signature: str
+    node: int
+
+
+@builder
+@msgpk
+class PySDGEdge(BaseModel):
+    """Interprocedural dependence edge. ``CALL``/``PARAM_IN``/``PARAM_OUT``
+    cross functions; ``SUMMARY`` connects a callsite's actual_in to its
+    actual_out within the caller (the callee's transitive flow)."""
+
+    source: PySDGEndpoint
+    target: PySDGEndpoint
+    type: Literal["CALL", "PARAM_IN", "PARAM_OUT", "SUMMARY"]
+    var: Optional[str] = None
+
+
+@builder
+@msgpk
+class PyProgramGraphs(BaseModel):
+    """The optional level-3 top-level section of ``analysis.json`` (present
+    only at ``-a 3``), versioned independently of the application schema."""
+
+    schema_version: str = "1.0.0"
+    k_limit: int = 3
+    functions: Dict[str, PyFunctionGraphs] = {}
+    sdg_edges: List[PySDGEdge] = []
+
+
 @builder
 @msgpk
 class PyApplication(BaseModel):
@@ -380,3 +509,5 @@ class PyApplication(BaseModel):
     # builtin members), keyed by signature. Populated by the analyzer so every
     # backend (JSON and Neo4j) shares one authoritative external-symbol set.
     external_symbols: Dict[str, PyExternalSymbol] = {}
+    # Level-3 native dataflow graphs (CFG/PDG/SDG); None below -a 3.
+    program_graphs: Optional[PyProgramGraphs] = None