refactor: reshape analysis levels as a fluent AnalysisPipeline

codeanalyzer/dataflow/scalpel_oracle.py

@@ -253,7 +253,8 @@ def make_alias_oracle(pycallable, func_ast, base_types) -> object:
     Returns a :class:`ScalpelAliasOracle` when ``python-scalpel`` is importable
     *and* builds successfully on ``func_ast``; otherwise logs once (INFO) and
     returns a :class:`TypeBasedAliasOracle` over ``base_types``.  Never raises —
-    mirrors how ``core._get_pycg_call_graph`` degrades on a missing/failed PyCG.
+    mirrors how ``pipeline.passes.pycg_call_graph_edges`` degrades on a
+    missing/failed PyCG.
     """
     fallback = TypeBasedAliasOracle(base_types)
     try:

codeanalyzer/pipeline/__init__.py

@@ -0,0 +1,4 @@
+from codeanalyzer.pipeline.context import AnalysisContext
+from codeanalyzer.pipeline.pipeline import AnalysisPipeline
+
+__all__ = ["AnalysisContext", "AnalysisPipeline"]

codeanalyzer/pipeline/context.py

@@ -0,0 +1,32 @@
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from codeanalyzer.options import AnalysisOptions
+from codeanalyzer.schema import PyApplication, PyModule
+
+
+@dataclass
+class AnalysisContext:
+    """Mutable carrier threaded through the AnalysisPipeline passes.
+
+    Inputs are set at construction; produced artifacts start as ``None`` and are
+    filled in chain order: ``symbol_table`` -> ``app``/``sig_to_id`` ->
+    ``infos`` -> ``ir``. ``infos`` (L3 PDGs) is deliberately reused by the L4
+    pass, so its ordering in the chain matters.
+    """
+
+    # inputs
+    options: AnalysisOptions
+    project_dir: Path
+    virtualenv: Optional[Path]
+    analysis_level: int
+    app_name: str
+    cached_symbol_table: Dict[str, PyModule] = field(default_factory=dict)
+
+    # produced by passes (loosely typed to avoid importing dataflow types here)
+    symbol_table: Optional[Dict[str, PyModule]] = None
+    app: Optional[PyApplication] = None
+    sig_to_id: Optional[Dict[str, str]] = None
+    infos: Optional[Dict[str, Any]] = None   # Dict[str, FunctionInfo]
+    ir: Optional[Any] = None                 # ProgramGraphsIR

codeanalyzer/pipeline/passes.py

@@ -0,0 +1,137 @@
+from pathlib import Path
+from typing import Dict, List
+
+from codeanalyzer.options import AnalysisOptions
+from codeanalyzer.pipeline.context import AnalysisContext
+from codeanalyzer.pipeline.symbol_table import build_symbol_table
+from codeanalyzer.provenance import repository_info
+from codeanalyzer.schema import PyApplication, PyExternalSymbol
+from codeanalyzer.schema.assign_ids import assign_ids
+from codeanalyzer.schema.call_graph_ids import reidentify_call_graph
+from codeanalyzer.schema.l1_body import populate_l1_body
+from codeanalyzer.schema.l2_callees import backfill_callees
+from codeanalyzer.schema.py_schema import PyCallEdge
+from codeanalyzer.semantic_analysis.call_graph import (
+    filter_external_edges, jedi_call_graph_edges, merge_edges,
+    resolve_unresolved_constructors,
+)
+from codeanalyzer.semantic_analysis.pycg import PyCG, PyCGExceptions
+from codeanalyzer.syntactic_analysis.import_resolver import resolve_imports
+from codeanalyzer.utils import logger
+
+
+def home_external_symbols(
+    app: PyApplication, app_id: str, sig_to_id: Dict[str, str]
+) -> Dict[str, PyExternalSymbol]:
+    """Home every call-graph endpoint that is not a declared callable onto a
+    ``can://…/@external/<module>/<name>`` id. Moved verbatim from
+    ``Codeanalyzer._home_external_symbols`` (static method, no ``self``)."""
+    externals: Dict[str, PyExternalSymbol] = {}
+    for edge in app.call_graph:
+        for sig in (edge.src, edge.dst):
+            if sig in sig_to_id:
+                continue
+            module, name = sig.rsplit(".", 1) if "." in sig else (None, sig)
+            ext_id = f"{app_id}/@external/{module}/{name}" if module else \
+                f"{app_id}/@external/{name}"
+            sig_to_id[sig] = ext_id
+            externals[ext_id] = PyExternalSymbol(id=ext_id, name=name, module=module)
+    return externals
+
+
+def pycg_call_graph_edges(
+    project_dir: Path, symbol_table, jedi_edges, options: AnalysisOptions
+) -> List[PyCallEdge]:
+    """Build PyCG-resolved call edges, degrading to Jedi-only on failure. Moved
+    from ``Codeanalyzer._get_pycg_call_graph`` (``self.X`` -> ``options.X`` /
+    ``project_dir``)."""
+    try:
+        pycg = PyCG(
+            project_dir,
+            skip_tests=options.skip_tests,
+            shard=options.pycg_shard,
+            shard_ceiling=options.pycg_shard_ceiling,
+            shard_timeout=options.pycg_shard_timeout,
+            shard_strategy=options.pycg_shard_strategy,
+            max_iter=options.pycg_max_iter,
+            using_ray=options.using_ray,
+        )
+        return pycg.build_call_graph_edges(symbol_table, jedi_edges=jedi_edges)
+    except PyCGExceptions.PyCGImportError as exc:
+        logger.warning(f"PyCG not installed — level 2 edges will be Jedi-only: {exc}")
+        return []
+    except PyCGExceptions.PyCGAnalysisError as exc:
+        logger.warning(f"PyCG analysis failed — level 2 edges will be Jedi-only: {exc}")
+        logger.debug("PyCG full traceback:", exc_info=True)
+        return []
+
+
+def _pass_symbol_table(ctx: AnalysisContext) -> None:
+    symbol_table = build_symbol_table(
+        ctx.project_dir, ctx.virtualenv, ctx.options, ctx.cached_symbol_table
+    )
+    resolve_unresolved_constructors(symbol_table)
+    ctx.symbol_table = symbol_table
+
+
+def _pass_call_graph(ctx: AnalysisContext) -> None:
+    assert ctx.symbol_table is not None, "call_graph pass requires the symbol-table pass"
+    st = ctx.symbol_table
+
+    call_graph = list(jedi_call_graph_edges(st))
+    if ctx.analysis_level >= 2:
+        pycg_edges = pycg_call_graph_edges(
+            ctx.project_dir, st, call_graph, ctx.options
+        )
+        call_graph = merge_edges(call_graph, pycg_edges)
+    call_graph = filter_external_edges(call_graph, st)
+    call_graph.sort(key=lambda e: (e.src, e.dst))
+
+    app = PyApplication.builder().symbol_table(st).call_graph(call_graph).build()
+    resolve_imports(app, ctx.project_dir)
+    app.repository = repository_info(ctx.project_dir)
+
+    sig_to_id = assign_ids(app, ctx.app_name)
+    app.external_symbols = home_external_symbols(app, app.id, sig_to_id)
+    populate_l1_body(app)
+    if ctx.analysis_level >= 2:
+        backfill_callees(app, sig_to_id)
+    reidentify_call_graph(app, sig_to_id)
+
+    ctx.app = app
+    ctx.sig_to_id = sig_to_id
+
+
+def _pass_intraproc_dataflow(ctx: AnalysisContext) -> None:
+    assert ctx.app is not None and ctx.sig_to_id is not None, \
+        "intraproc dataflow pass requires the call-graph pass"
+    from codeanalyzer.dataflow.builder import build_function_pdgs, emit_l3_body
+    from codeanalyzer.dataflow.syntactic import SyntacticOracle
+
+    infos, _func_asts = build_function_pdgs(
+        ctx.app,
+        k=ctx.options.graph_field_depth,
+        oracle_factory=lambda c, fast: SyntacticOracle(),
+    )
+    emit_l3_body(ctx.app, infos, ctx.sig_to_id, set(ctx.options.graphs.split(",")))
+    ctx.infos = infos
+
+
+def _pass_interproc_dataflow(ctx: AnalysisContext) -> None:
+    assert ctx.app is not None and ctx.sig_to_id is not None, \
+        "interproc dataflow pass requires the call-graph pass"
+    assert ctx.infos is not None, \
+        "interproc dataflow pass requires the intraproc pass (it reuses its PDGs)"
+    from codeanalyzer.dataflow.builder import (
+        _base_types, build_program_graphs, emit_ddg_pointsto_delta, emit_l4,
+    )
+    from codeanalyzer.dataflow.scalpel_oracle import make_alias_oracle
+
+    ir = build_program_graphs(
+        ctx.app,
+        k=ctx.options.graph_field_depth,
+        oracle_factory=lambda c, fast: make_alias_oracle(c, fast, _base_types(c)),
+    )
+    emit_l4(ctx.app, ir, ctx.sig_to_id)
+    emit_ddg_pointsto_delta(ctx.app, ctx.infos, ir, ctx.sig_to_id)
+    ctx.ir = ir

codeanalyzer/pipeline/pipeline.py

@@ -0,0 +1,53 @@
+import time
+
+from codeanalyzer.pipeline.context import AnalysisContext
+from codeanalyzer.pipeline.passes import (
+    _pass_call_graph, _pass_interproc_dataflow,
+    _pass_intraproc_dataflow, _pass_symbol_table,
+)
+from codeanalyzer.provenance import analyzer_info
+from codeanalyzer.schema import Analysis
+from codeanalyzer.utils import logger
+
+
+class AnalysisPipeline:
+    """Fluent chain of analysis passes over a shared AnalysisContext.
+
+    Each ``.with_*`` runs one pass through the shared ``_run`` gate: a pass
+    below its intrinsic ``min_level`` is a logged no-op. ``.build()`` assembles
+    the ``Analysis`` envelope from the produced context.
+    """
+
+    def __init__(self, ctx: AnalysisContext):
+        self.ctx = ctx
+
+    def with_symbol_table(self):
+        return self._run("symbol_table", 1, _pass_symbol_table)
+
+    def with_call_graph(self):
+        return self._run("call_graph", 1, _pass_call_graph)
+
+    def with_intraproc_dataflow(self):
+        return self._run("intraproc_dataflow", 3, _pass_intraproc_dataflow)
+
+    def with_interproc_dataflow(self):
+        return self._run("interproc_dataflow", 4, _pass_interproc_dataflow)
+
+    def _run(self, name, min_level, fn):
+        if self.ctx.analysis_level < min_level:
+            logger.info("⏭️  %s: skipped (level %d < %d)", name,
+                        self.ctx.analysis_level, min_level)
+            return self
+        t0 = time.perf_counter()
+        fn(self.ctx)
+        logger.info("✅ %s: %.1fs", name, time.perf_counter() - t0)
+        return self
+
+    def build(self) -> Analysis:
+        return Analysis(
+            max_level=self.ctx.analysis_level,
+            k_limit=self.ctx.options.graph_field_depth
+            if self.ctx.analysis_level >= 3 else None,
+            analyzer=analyzer_info(self.ctx.analysis_level),
+            application=self.ctx.app,
+        )