feat(models): cpg — canonical schema-v2 models, modeled once (F2 / #240) (#274)

* feat(cpg): _NullSafeBase + Span (#240)

* feat(cpg): Edge + Import leaf models (#240)

* test(cpg): cover Edge empty-prov default and Import.span default (#240)

* feat(cpg): open-kind Node covering type/callable/body facets (#240)

* feat(cpg): Module/Application/Analyzer/AnalysisPayload + exports (#240)

* test(cpg): cover Application->Module->Node deep composition (#240)

* test(cpg): parse real L1/L4 samples from both analyzers + superset gate (#240)

* fix(cpg): enforce id on durable nodes positionally; body nodes exempt (#240)

* test(cpg): pin the F7/F3 accessor contract against real L4 samples (#240)

* test(cpg): genuinely pin module.functions accessor; strengthen TS source check (#240)

* fix(cpg): add Module.span so the common span field parses (#240)

span is listed as a common field on every node in the keystone (Part
II), module included, but Module had no span field so it degraded to
a raw dict in model_extra instead of parsing as Span. The ts-a4/ts-a1
fixtures emit span on the module node.

* test(cpg): pin cdg/summary/k_limit/TS body+cfg in the accessor contract (#240)

extra="allow" absorbs unknown keys, so deleting a canonical field
leaves every parse test green — the accessor contract is the only
guard. cdg and summary (both in F7's cfg/cdg/ddg/summary read set),
the envelope k_limit, and TS callable body/cfg were unpinned.

Each new assertion dereferences an element/field rather than doing a
bare isinstance check, since a raw dict-of-dicts under extra="allow"
still satisfies isinstance(list)/isinstance(dict) — confirmed by
temporarily removing each field from the model and watching the new
asserts fail before restoring.

Author: rahlk

cldk/models/cpg/__init__.py

@@ -0,0 +1,8 @@
+from cldk.models.cpg.base import _NullSafeBase
+from cldk.models.cpg.models import (
+    Span, Import, Edge, Node, Module, Application, Analyzer, AnalysisPayload,
+)
+
+__all__ = [
+    "AnalysisPayload", "Application", "Module", "Node", "Edge", "Span", "Import", "Analyzer",
+]

cldk/models/cpg/base.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+from pydantic import BaseModel, ConfigDict, model_validator
+
+
+class _NullSafeBase(BaseModel):
+    """Shared base for every canonical (cpg) model. `extra="allow"` so language-specific fields
+    (TS is_tsx/exports, Python package, …) are tolerated and preserved rather than rejected — the
+    device that lets ONE model set parse every analyzer. The before-validator drops None-valued
+    keys so a collection serialized as `null` (Go/Rust/C) falls back to its field default; the one
+    sanctioned null (a body-node `callee`) simply resolves to its `None` default."""
+
+    model_config = ConfigDict(extra="allow")
+
+    @model_validator(mode="before")
+    @classmethod
+    def _drop_nulls(cls, data):
+        if isinstance(data, dict):
+            return {k: v for k, v in data.items() if v is not None}
+        return data

cldk/models/cpg/models.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+from typing import Any, Dict, List, Optional, Tuple
+from pydantic import model_validator
+from cldk.models.cpg.base import _NullSafeBase
+
+
+class Span(_NullSafeBase):
+    start: Tuple[int, int]
+    end: Tuple[int, int]
+    bytes: Tuple[int, int]
+
+
+class Edge(_NullSafeBase):
+    src: str
+    dst: str
+    kind: Optional[str] = None      # cfg edge kind; absent on identity edges
+    var: Optional[str] = None       # ddg access path
+    prov: List[str] = []            # ["jedi"|"pycg"|"tsc"|"jelly"] (call) / ["ssa"|"points-to"] (ddg)
+    weight: int = 1
+
+
+class Import(_NullSafeBase):
+    name: str
+    path: Optional[str] = None
+    alias: Optional[str] = None
+    span: Optional[Span] = None
+
+
+class Node(_NullSafeBase):
+    # Optional because sub-callable body nodes are keyed by local position and omit id; durable
+    # nodes (types/callables/functions/fields) are required to carry it — enforced positionally by
+    # the container validators.
+    id: Optional[str] = None
+    kind: str
+    span: Optional[Span] = None
+    parent: Optional[str] = None
+    # type facet
+    base_types: List[str] = []
+    interfaces: List[str] = []
+    modifiers: List[str] = []
+    decorators: List[Any] = []
+    callables: Dict[str, "Node"] = {}
+    fields: Dict[str, "Node"] = {}
+    # callable facet
+    signature: Optional[str] = None
+    parameters: List[Any] = []
+    return_type: Optional[str] = None
+    error_channel: List[str] = []
+    metrics: Dict[str, Any] = {}
+    refs: Dict[str, Any] = {}
+    body: Dict[str, "Node"] = {}
+    cfg: List[Edge] = []
+    cdg: List[Edge] = []
+    ddg: List[Edge] = []
+    summary: List[Edge] = []
+    # field / body-node facet
+    type: Optional[str] = None
+    callee: Optional[str] = None
+    arguments: List[str] = []
+    of: Optional[str] = None
+    # open vocab
+    tags: Dict[str, str] = {}
+
+    @model_validator(mode="after")
+    def _durable_children_require_id(self):
+        for container in (self.callables, self.fields):
+            for key, node in container.items():
+                if node.id is None:
+                    raise ValueError(f"durable node {key!r} under {self.id or '<body>'!r} is missing required id")
+        return self
+
+
+class Module(_NullSafeBase):
+    id: str
+    kind: str = "module"
+    span: Optional[Span] = None
+    package: Optional[str] = None
+    source: str = ""
+    imports: List[Import] = []
+    types: Dict[str, Node] = {}
+    functions: Dict[str, Node] = {}
+    content_hash: Optional[str] = None
+
+    @model_validator(mode="after")
+    def _durable_children_require_id(self):
+        for container in (self.types, self.functions):
+            for key, node in container.items():
+                if node.id is None:
+                    raise ValueError(f"durable node {key!r} in module {self.id!r} is missing required id")
+        return self
+
+
+class Application(_NullSafeBase):
+    id: str
+    kind: str = "application"
+    symbol_table: Dict[str, Module] = {}
+    call_graph: List[Edge] = []
+    param_in: List[Edge] = []
+    param_out: List[Edge] = []
+
+
+class Analyzer(_NullSafeBase):
+    name: str
+    version: Optional[str] = None
+
+
+class AnalysisPayload(_NullSafeBase):
+    schema_version: str
+    language: str
+    max_level: int
+    k_limit: Optional[int] = None
+    analyzer: Optional[Analyzer] = None
+    application: Application

tests/models/cpg/__init__.py

@@ -0,0 +1,66 @@
+"""Pin the accessors F7 (Task-7 provider) and F3 (views) depend on, against a real L4 sample."""
+import json
+from pathlib import Path
+from cldk.models.cpg import AnalysisPayload
+
+RES = Path(__file__).parent.parent.parent / "resources" / "cpg"
+
+
+def _app(name):
+    return AnalysisPayload(**json.loads((RES / name).read_text())).application
+
+
+def test_symbol_table_module_source_and_containment():
+    app = _app("py-a4.json")
+    mod = app.symbol_table["pkg/mod.py"]
+    assert isinstance(mod.source, str) and mod.source          # module.source (byte-slice base)
+    assert isinstance(mod.types, dict) and isinstance(mod.functions, dict)   # both accessors pinned
+    assert mod.types or mod.functions                          # at least one populated
+
+
+def test_callable_body_and_dataflow_edges_present_at_l4():
+    app = _app("py-a4.json")
+    mod = app.symbol_table["pkg/mod.py"]
+    cls = next(iter(mod.types.values()))
+    call = next(iter(cls.callables.values()))
+    assert call.signature                                      # callable.signature
+    assert call.body                                           # body{} populated at L4
+    assert call.cfg and call.ddg                               # cfg/ddg edge lists
+    # cdg/summary are unpinned elsewhere and are the sole extra="allow" guard for these two
+    # fields (both in F7's cfg/cdg/ddg/summary read set) — dereference an element attribute so a
+    # deleted field (which would fall back to a raw dict under extra="allow") fails loudly.
+    assert isinstance(call.cdg, list) and isinstance(call.summary, list)
+    assert call.cdg[0].src and call.summary[0].src
+    # span.bytes present for slicing a body node
+    some = next(iter(call.body.values()))
+    assert some.span is None or (some.span.bytes and len(some.span.bytes) == 2)
+
+
+def test_envelope_k_limit_at_l4():
+    payload = AnalysisPayload(**json.loads((RES / "py-a4.json").read_text()))
+    assert payload.k_limit == 3
+    # a plain value check alone would still pass via the extra="allow" passthrough even if
+    # k_limit were deleted from the model — assert it's a declared field, not an extras leak.
+    assert "k_limit" not in (payload.model_extra or {})
+
+
+def test_application_interprocedural_edges_at_l4():
+    app = _app("py-a4.json")
+    assert app.call_graph                                      # L2 call graph
+    assert app.param_in and app.param_out                     # L4 SDG param edges
+    for e in app.call_graph:
+        assert e.src.startswith("can://") and e.dst.startswith("can://")   # can:// identity
+
+
+def test_typescript_sample_same_accessors():
+    app = _app("ts-a4.json")
+    mod = next(iter(app.symbol_table.values()))
+    assert isinstance(mod.source, str) and mod.source
+    # a TS type node with callables
+    typ = next((t for t in mod.types.values() if t.callables), None)
+    assert typ is not None and next(iter(typ.callables.values())).signature is not None
+    # resetPassword under type Users: body/cfg must resolve to parsed Node/Edge, not raw dicts,
+    # so the TS analyzer path isn't pinned on source/signature alone.
+    call = typ.callables["resetPassword"]
+    assert isinstance(call.body, dict) and next(iter(call.body.values())).kind
+    assert isinstance(call.cfg, list) and call.cfg[0].kind

tests/models/cpg/test_accessor_contract.py

@@ -0,0 +1,26 @@
+from cldk.models.cpg.models import Span
+from cldk.models.cpg.base import _NullSafeBase
+from pydantic import ConfigDict
+from typing import Dict, List
+
+
+class _M(_NullSafeBase):
+    xs: List[int] = []
+    d: Dict[str, int] = {}
+    opt: int | None = None
+
+
+def test_null_collections_coerce_to_defaults():
+    m = _M(**{"xs": None, "d": None, "opt": None})
+    assert m.xs == [] and m.d == {} and m.opt is None
+
+
+def test_extra_fields_are_allowed_and_preserved():
+    m = _M(**{"xs": [1], "is_tsx": True})       # language-specific extra
+    assert m.xs == [1]
+    assert m.model_extra.get("is_tsx") is True
+
+
+def test_span_parses_byte_offsets():
+    s = Span(**{"start": [1, 0], "end": [4, 2], "bytes": [0, 40]})
+    assert s.start == (1, 0) and s.end == (4, 2) and s.bytes == (0, 40)

tests/models/cpg/test_base_span.py

@@ -0,0 +1,59 @@
+import pytest
+from pydantic import ValidationError
+
+from cldk.models.cpg import AnalysisPayload, Application, Module, Node, Edge, Span, Import, Analyzer
+
+
+def test_envelope_reads_authoritative_level():
+    p = AnalysisPayload(**{
+        "schema_version": "2.0.0", "language": "python", "max_level": 4, "k_limit": 3,
+        "analyzer": {"name": "codeanalyzer-python", "version": "0.4.0"},
+        "application": {"id": "can://python/app", "kind": "application", "symbol_table": {}},
+    })
+    assert p.schema_version == "2.0.0" and p.max_level == 4 and p.k_limit == 3
+    assert p.analyzer.name == "codeanalyzer-python"
+    assert p.application.id == "can://python/app"
+
+
+def test_module_holds_source_and_containment():
+    m = Module(**{"id": "can://python/app/m.py", "kind": "module", "source": "x = 1\n",
+                  "types": {"C": {"id": "can://python/app/m.py/C", "kind": "class"}},
+                  "functions": {"f()": {"id": "can://python/app/m.py/f()", "kind": "function"}}})
+    assert m.source == "x = 1\n"
+    assert m.types["C"].kind == "class" and m.functions["f()"].kind == "function"
+
+
+def test_module_durable_node_missing_id_raises():
+    # a type reached through the durable-containment dict MUST carry the join key id
+    with pytest.raises(ValidationError):
+        Module(**{"id": "m", "types": {"C": {"kind": "class"}}})   # no id on the type
+
+
+def test_application_edge_lists():
+    a = Application(**{"id": "can://python/app", "kind": "application", "symbol_table": {},
+                       "call_graph": [{"src": "a", "dst": "b", "prov": ["jedi"], "weight": 1}],
+                       "param_in": [{"src": "c@in", "dst": "d@in"}]})
+    assert a.call_graph[0].dst == "b" and a.param_in[0].src == "c@in" and a.param_out == []
+
+
+def test_symbol_table_deep_composition():
+    from cldk.models.cpg import Application, Module, Node
+    a = Application(**{
+        "id": "can://python/app", "kind": "application",
+        "symbol_table": {
+            "pkg/m.py": {
+                "id": "can://python/app/pkg/m.py", "kind": "module", "source": "x = 1\n",
+                "types": {"C": {"id": "can://python/app/pkg/m.py/C", "kind": "class",
+                                "callables": {"C.f()": {"id": "can://python/app/pkg/m.py/C/f()",
+                                                        "kind": "method", "signature": "f"}}}},
+                "functions": {"g()": {"id": "can://python/app/pkg/m.py/g()", "kind": "function"}},
+            }
+        },
+    })
+    mod = a.symbol_table["pkg/m.py"]
+    assert isinstance(mod, Module) and mod.source == "x = 1\n"
+    cls = mod.types["C"]
+    assert isinstance(cls, Node) and cls.kind == "class"
+    method = cls.callables["C.f()"]
+    assert isinstance(method, Node) and method.kind == "method" and method.signature == "f"
+    assert isinstance(mod.functions["g()"], Node)

tests/models/cpg/test_containers.py

@@ -0,0 +1,22 @@
+from cldk.models.cpg.models import Edge, Import
+
+
+def test_call_edge_shape():
+    e = Edge(**{"src": "can://p/a#f", "dst": "can://p/a#g", "prov": ["jedi", "pycg"], "weight": 2})
+    assert e.src.endswith("#f") and e.dst.endswith("#g")
+    assert e.prov == ["jedi", "pycg"] and e.weight == 2 and e.kind is None and e.var is None
+
+
+def test_ddg_edge_carries_var_and_prov():
+    e = Edge(**{"src": "a@1:0", "dst": "a@2:0", "var": "x", "prov": ["ssa"]})
+    assert e.var == "x" and e.prov == ["ssa"] and e.weight == 1
+
+
+def test_edge_empty_prov_and_weight_defaults():
+    e = Edge(src="a", dst="b")
+    assert e.prov == [] and e.weight == 1
+
+
+def test_import_optional_fields():
+    i = Import(**{"name": "os"})
+    assert i.name == "os" and i.path is None and i.alias is None and i.span is None

tests/models/cpg/test_edge_import.py

@@ -0,0 +1,44 @@
+import pytest
+from pydantic import ValidationError
+
+from cldk.models.cpg.models import Node
+
+
+def test_class_node_facet():
+    n = Node(**{"id": "can://p/m.py/C", "kind": "class",
+                "callables": {"C.f()": {"id": "can://p/m.py/C/f()", "kind": "method", "signature": "f"}}})
+    assert n.kind == "class"
+    assert n.callables["C.f()"].kind == "method" and n.callables["C.f()"].signature == "f"
+
+
+def test_callable_node_carries_body_and_edges():
+    n = Node(**{"id": "can://p/m.py/f()", "kind": "function", "signature": "f()",
+                "body": {"f@1:0": {"id": "can://p/m.py/f()@1:0", "kind": "statement"}},
+                "cfg": [{"src": "can://p/m.py/f()@1:0", "dst": "can://p/m.py/f()@2:0", "kind": "fallthrough"}],
+                "ddg": [{"src": "can://p/m.py/f()@1:0", "dst": "can://p/m.py/f()@2:0", "var": "x", "prov": ["ssa"]}]})
+    assert set(n.body) == {"f@1:0"}
+    assert n.cfg[0].kind == "fallthrough" and n.ddg[0].var == "x" and n.ddg[0].prov == ["ssa"]
+
+
+def test_call_body_node_callee_refines_from_null():
+    n = Node(**{"id": "a@2:0", "kind": "call", "callee": None, "arguments": ["a@2:0/arg0"]})
+    assert n.kind == "call" and n.callee is None and n.arguments == ["a@2:0/arg0"]
+
+
+def test_language_extra_field_preserved():
+    n = Node(**{"id": "x", "kind": "class", "is_abstract": True})   # a language-specific flag
+    assert n.model_extra.get("is_abstract") is True
+
+
+def test_durable_callable_missing_id_raises():
+    # a callable reached through the durable-containment dict MUST carry the join key id
+    with pytest.raises(ValidationError):
+        Node(**{"id": "can://p/m.py/C", "kind": "class",
+                "callables": {"C.f()": {"kind": "method"}}})   # no id on the callable
+
+
+def test_body_node_missing_id_parses():
+    # body nodes are keyed by local position and legitimately omit id — must NOT raise
+    n = Node(**{"id": "can://p/m.py/f()", "kind": "function",
+                "body": {"1:0": {"kind": "statement"}}})
+    assert n.body["1:0"].id is None and n.body["1:0"].kind == "statement"