codellm-devkit · rahlk · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026
diff --git a/cldk/models/cpg/__init__.py b/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",
+]
diff --git a/cldk/models/cpg/base.py b/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
diff --git a/cldk/models/cpg/models.py b/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
diff --git a/tests/models/cpg/__init__.py b/tests/models/cpg/__init__.py
diff --git a/tests/models/cpg/test_accessor_contract.py b/tests/models/cpg/test_accessor_contract.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
diff --git a/tests/models/cpg/test_base_span.py b/tests/models/cpg/test_base_span.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)
diff --git a/tests/models/cpg/test_containers.py b/tests/models/cpg/test_containers.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)
diff --git a/tests/models/cpg/test_edge_import.py b/tests/models/cpg/test_edge_import.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
diff --git a/tests/models/cpg/test_node.py b/tests/models/cpg/test_node.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"