Skip to content
2 changes: 1 addition & 1 deletion docs/ontology/patterns.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ project_domains:
fqn_prefix: "cgis.resolver"
expected_pattern: pipeline_stage
profile: python
drift_tolerance: 0.40 # v2 measured ≈0.35 (tv_calls .65; IMPORTS layer empty → excluded)
drift_tolerance: 0.40 # v2 measured ≈0.25 post-#115 split (tv_imp 0.00 — real 021C chain engine→symbols→indices; residual is tv_calls)

- name: "pipeline"
fqn_prefix: "cgis.pipeline"
Expand Down
21 changes: 21 additions & 0 deletions docs/specs/2026-06-11-resolver-split-design.md
Original file line number Diff line number Diff line change
Expand Up @@ -245,3 +245,24 @@ assumed**:
- `uplift.py` and `resolver/__init__.py` re-export changes beyond what the
new modules need.
- Slice 2 of #161 (symbol-level imports) — separate effort.

## Amendment 1 (2026-06-11): honest import chain instead of re-exports

PR #180 review and issue #182 established that §3.3's pre-authorized fix
("symbols re-exports what the facade needs") merely laundered the
engine→indices dependency through a passthrough re-export — the IMPORTS layer
measured as a 021C chain while the real dependency graph stayed a triangle.

Superseding §2.1/§2.4/§2.5 details:

- `SELF_PREFIX` / `RAW_CLASS_PREFIX` are public constants in `core/models.py`
(Edge-target encoding conventions, same family as `VIRTUAL_FILE_PATH`).
- `SymbolResolver.__init__(nodes, edges)` builds its own index via
`IndexBuilder` and exposes it as the public `index` attribute. The
"index complete before resolution" ordering invariant is now structural.
- `ResolverEngine` imports only `cgis.resolver.symbols` (+ core) — the
021C chain engine→symbols→indices is the true dependency graph, not a
measurement artifact. No re-exports, no `noqa: PLC0414`.
- `RAW_DEP_PREFIX` remains in `engine.py` (#161 contract).

Metric-hardening against passthrough re-exports in general is tracked in #182.
912 changes: 912 additions & 0 deletions docs/specs/plans/2026-06-11-resolver-split.md

Large diffs are not rendered by default.

6 changes: 6 additions & 0 deletions src/cgis/core/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,12 @@ class NodeNamespace(StrEnum):

VIRTUAL_FILE_PATH = "EXTERNAL"

# Edge-target encoding conventions for raw (pre-resolution) targets.
# These mirror the raw_call: / raw_dep: prefixes used by extractors and the
# resolver, but apply to self-method dispatch and class-inheritance edges.
SELF_PREFIX = "self."
RAW_CLASS_PREFIX = "raw_class:"


class NodeType(StrEnum):
"""All possible node types in the Code Graph."""
Expand Down
362 changes: 48 additions & 314 deletions src/cgis/resolver/engine.py

Large diffs are not rendered by default.

184 changes: 184 additions & 0 deletions src/cgis/resolver/indices.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
"""Symbol indices for FQN resolution: built once by IndexBuilder, read-only after."""

import builtins
import os
import sys
from dataclasses import dataclass

from cgis.core.models import SELF_PREFIX, Node, NodeNamespace, NodeType

_BUILTINS: frozenset[str] = frozenset(dir(builtins))


@dataclass(frozen=True)
class SymbolIndex:
"""Immutable lookup indices over the extracted node set.

Built by IndexBuilder, consumed by SymbolResolver and ResolverEngine.
Frozen by convention (field rebinding prevented); the contained dicts
are never mutated after construction.
"""

# node id (FQN) -> Node
nodes: dict[str, Node]
# name -> list of FQNs
global_symbols: dict[str, list[str]]
# (file_path, name) -> list of FQNs (list handles conditional redefinitions)
file_global_symbols: dict[tuple[str, str], list[str]]
# class_fqn -> {method_name -> method_fqn}
class_methods: dict[str, dict[str, str]]
# DI-alias (VARIABLE) indices for raw_dep: resolution; kept separate
# from global_symbols so call resolution behavior does not change.
variable_symbols: dict[str, list[str]]
file_variable_symbols: dict[tuple[str, str], list[str]]
# normalized file_path -> {local_alias: target_fqn} (from FILE node import_map)
file_imports: dict[str, dict[str, str]]
# suffix_fqn -> [full_node_ids] (handles src/ layout prefix mismatch)
suffix_map: dict[str, list[str]]
# top-level root segments of all internal nodes (for classify)
internal_roots: set[str]
# root segments of absolute imports (anything else is UNKNOWN)
external_roots: set[str]

def map_to_node_fqn(self, imported_fqn: str) -> str | None:
"""Resolve an imported FQN to an actual node in the graph.

Handles two common prefix-mismatch patterns:
- Import has extra package prefix: `cgis.resolver.engine.X` → node `resolver.engine.X`
(strip leading segments from the imported FQN)
- Node has extra layout prefix: `cgis.pipeline.X` → node `src.cgis.pipeline.X`
(look up in suffix_map built from node IDs)

Returns None when the target is ambiguous or not in the graph.
"""
if imported_fqn in self.nodes:
return imported_fqn
# Node has extra layout prefix (e.g. src/) — most precise match first
candidates = self.suffix_map.get(imported_fqn, [])
if len(candidates) == 1:
return candidates[0]
# Strip leading segments from the imported FQN (import has extra prefix)
parts = imported_fqn.split(".")
for i in range(1, len(parts)):
candidate = ".".join(parts[i:])
if candidate in self.nodes:
return candidate
return None

def classify_fqn(self, fqn: str) -> NodeNamespace:
"""Classify an FQN as STDLIB, INTERNAL, EXTERNAL, or UNKNOWN.

UNKNOWN means the root segment was not found in internal roots,
stdlib/builtins, or any known import-map external root.
"""
if fqn.startswith((".", SELF_PREFIX)):
return NodeNamespace.INTERNAL
root = fqn.split(".", maxsplit=1)[0]
if root in self.internal_roots:
return NodeNamespace.INTERNAL
if root in sys.stdlib_module_names or root in _BUILTINS:
return NodeNamespace.STDLIB
if root in self.external_roots:
return NodeNamespace.EXTERNAL
return NodeNamespace.UNKNOWN

def is_variable_node(self, fqn: str) -> bool:
"""Return True when fqn names an existing VARIABLE node in the graph."""
node = self.nodes.get(fqn)
return node is not None and node.type == NodeType.VARIABLE

def normalized_file_path(self, source_fqn: str, edge_file_path: str | None) -> str | None:
"""Return the normalized file path for a source FQN, falling back to edge_file_path."""
source_node = self.nodes.get(source_fqn)
if source_node:
return os.path.normpath(source_node.file_path)
return os.path.normpath(edge_file_path) if edge_file_path else None


class IndexBuilder:
"""Builds a SymbolIndex from extracted nodes (Phase 1: indexing).

Takes only nodes — never edges. The inheritance tree is deliberately
NOT built here: resolving EXTENDS targets requires symbol resolution,
so it belongs to SymbolResolver (spec §2.4).
"""

def build(self, nodes: list[Node]) -> SymbolIndex:
"""Index all nodes for fast resolution and return the frozen index."""
nodes_by_id = {n.id: n for n in nodes}
global_symbols: dict[str, list[str]] = {}
file_global_symbols: dict[tuple[str, str], list[str]] = {}
class_methods: dict[str, dict[str, str]] = {}
variable_symbols: dict[str, list[str]] = {}
file_variable_symbols: dict[tuple[str, str], list[str]] = {}
file_imports: dict[str, dict[str, str]] = {}
suffix_map: dict[str, list[str]] = {}
internal_roots: set[str] = set()

for node in nodes_by_id.values():
# Index FILE-level import maps
if node.type == NodeType.FILE:
normalized = os.path.normpath(node.file_path)
file_imports[normalized] = node.metadata.get("import_map") or {}
Comment thread
zaebee marked this conversation as resolved.

# Index global functions/symbols
if node.type in (NodeType.FUNCTION, NodeType.CLASS):
global_symbols.setdefault(node.name, []).append(node.id)
file_global_symbols.setdefault(
(os.path.normpath(node.file_path), node.name), []
).append(node.id)

# Index methods within classes
if node.type == NodeType.METHOD:
class_fqn, sep, _ = node.id.rpartition(".")
if sep:
class_methods.setdefault(class_fqn, {})[node.name] = node.id

# Index DI aliases for raw_dep: candidate resolution
if node.type == NodeType.VARIABLE:
variable_symbols.setdefault(node.name, []).append(node.id)
file_variable_symbols.setdefault(
(os.path.normpath(node.file_path), node.name), []
).append(node.id)

# Build suffix map for src/-layout prefix normalization:
# "src.cgis.pipeline.X" → suffix "cgis.pipeline.X" also points to the node
self._add_node_to_suffix_map(node.id, suffix_map, internal_roots)

return SymbolIndex(
nodes=nodes_by_id,
global_symbols=global_symbols,
file_global_symbols=file_global_symbols,
class_methods=class_methods,
variable_symbols=variable_symbols,
file_variable_symbols=file_variable_symbols,
file_imports=file_imports,
suffix_map=suffix_map,
internal_roots=internal_roots,
external_roots=self._build_external_roots(file_imports),
)

def _add_node_to_suffix_map(
self, node_id: str, suffix_map: dict[str, list[str]], internal_roots: set[str]
) -> None:
"""Add node to suffix map and internal roots based on its ID."""
parts = node_id.split(".")
internal_roots.add(parts[0])
if len(parts) > 1 and parts[0] in {"src", "lib"}:
internal_roots.add(parts[1])
for i in range(1, len(parts)):
suffix = ".".join(parts[i:])
suffix_map.setdefault(suffix, []).append(node_id)

def _build_external_roots(self, file_imports: dict[str, dict[str, str]]) -> set[str]:
"""Build set of known external root modules from file import maps.

Any root not in internal_roots, stdlib, or external_roots
is classified as UNKNOWN by classify_fqn.
"""
return {
val.split(".", maxsplit=1)[0]
for import_map in file_imports.values()
for val in import_map.values()
if val and not val.startswith(".")
}
166 changes: 166 additions & 0 deletions src/cgis/resolver/symbols.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
"""Symbol resolution strategies over a SymbolIndex."""

from cgis.core.models import RAW_CLASS_PREFIX, Edge, EdgeType, Node
from cgis.resolver.indices import IndexBuilder, SymbolIndex


class SymbolResolver:
"""Maps raw symbol names to graph FQNs.

Strategy chain per call site: local variable types, the consuming file's
import map, then the global symbol index with same-file preference.
Holds the inheritance tree (a resolution product built from EXTENDS
edges — not an index, see spec §2.4).

The constructor accepts nodes and edges directly; it builds the
SymbolIndex internally via IndexBuilder and exposes it as the public
``index`` attribute so callers (e.g. ResolverEngine) can read the index
without building it themselves.
"""

Comment thread
zaebee marked this conversation as resolved.
def __init__(self, nodes: list[Node], edges: list[Edge]) -> None:
"""Build the symbol index from nodes, then the inheritance tree from EXTENDS edges."""
self.index: SymbolIndex = IndexBuilder().build(nodes)
# class_fqn -> [resolved parent FQNs] built from EXTENDS edges
self._inheritance_tree: dict[str, list[str]] = {}
for edge in edges:
if edge.type == EdgeType.EXTENDS:
raw = edge.target.removeprefix(RAW_CLASS_PREFIX)
resolved = self.resolve_class_ref(raw, edge.source, edge.file_path)
self._inheritance_tree.setdefault(edge.source, []).append(resolved or raw)

def resolve_class_ref(
self, name: str, source_fqn: str, edge_file_path: str | None
) -> str | None:
"""Resolve a class name from an EXTENDS edge target to a graph FQN."""
file_path = self.index.normalized_file_path(source_fqn, edge_file_path)
if file_path:
result = self._resolve_via_import_map(name, file_path)
if result:
return result
# Global symbol index is keyed by short name; for dotted refs like `models.BaseModel`
# strip the module prefix and verify the resolved FQN ends with the full dotted name.
query_name = name.rsplit(".", maxsplit=1)[-1]
resolved = self._resolve_via_global_symbols(query_name, file_path)
if resolved and ("." not in name or resolved == name or resolved.endswith(f".{name}")):
return resolved
return None

def resolve_self_call(self, source_fqn: str, method_name: str) -> str | None:
"""Attempts to find a method on the class that owns source, traversing inheritance.

Walks up the FQN segments to handle nested functions (e.g. mod.Cls.method.inner).
"""
parts = source_fqn.split(".")
for i in range(len(parts) - 1, 0, -1):
candidate = ".".join(parts[:i])
if candidate in self.index.class_methods:
return self._resolve_method_on_class_hierarchy(candidate, method_name, set())
return None

def resolve_global_call(
self, name: str, source_fqn: str, edge_file_path: str | None = None
) -> str | None:
"""Resolve a global call using local types, import map, then global symbol index."""
source_node = self.index.nodes.get(source_fqn)
file_path = self.index.normalized_file_path(source_fqn, edge_file_path)

if source_node:
result = self._resolve_local_type_call(name, source_node)
if result:
return result

if file_path:
result = self._resolve_via_import_map(name, file_path)
if result:
return result

return self._resolve_via_global_symbols(name, file_path)

def resolve_dep_candidate(
self, name: str, source_fqn: str, edge_file_path: str | None
) -> str | None:
"""Resolve a raw_dep: candidate to a VARIABLE (DI alias) node, or None.

Order: the consuming file's import map first, then the VARIABLE-only
symbol index with same-file preference. Returns None for anything that
is not an existing VARIABLE node — the caller drops the edge.
A globally-unique match is accepted even when the name is not importable
in the consuming file — matching resolve_global_call's behavior.

When ``name`` is present in the file's import map, the import is
authoritative: return the hit only if it is a VARIABLE node, otherwise
return ``None`` immediately — never fall through to the global
``variable_symbols`` index. This prevents an explicitly imported
class from being shadowed by a same-named DI alias in another module.
"""
file_path = self.index.normalized_file_path(source_fqn, edge_file_path)
if file_path:
via_import = self._resolve_via_import_map(name, file_path)
if via_import:
return via_import if self.index.is_variable_node(via_import) else None
candidates = self.index.variable_symbols.get(name, [])
if len(candidates) == 1:
return candidates[0]
if candidates and file_path:
same_file = self.index.file_variable_symbols.get((file_path, name), [])
if len(same_file) == 1:
return same_file[0]
return None

def _resolve_method_on_class_hierarchy(
self, class_fqn: str, method_name: str, visited: set[str]
) -> str | None:
"""DFS over EXTENDS edges to find method_name defined in class_fqn or any ancestor."""
if class_fqn in visited:
return None
visited.add(class_fqn)
direct = self.index.class_methods.get(class_fqn, {}).get(method_name)
if direct:
return direct
for parent_fqn in self._inheritance_tree.get(class_fqn, []):
result = self._resolve_method_on_class_hierarchy(parent_fqn, method_name, visited)
if result:
return result
return None

def _resolve_via_import_map(self, name: str, file_path: str) -> str | None:
"""Look up name in the file's import map (direct and module-prefixed calls)."""
file_import_map = self.index.file_imports.get(file_path, {})

if name in file_import_map:
target_fqn = file_import_map[name]
return self.index.map_to_node_fqn(target_fqn) or target_fqn

first_part = name.split(".", maxsplit=1)[0]
if first_part in file_import_map and "." in name:
rest = name[len(first_part) + 1 :]
target_fqn = f"{file_import_map[first_part]}.{rest}"
return self.index.map_to_node_fqn(target_fqn) or target_fqn

return None

def _resolve_local_type_call(self, name: str, source_node: Node) -> str | None:
"""Resolve `var.method` using local_types metadata on the source node."""
if "." not in name:
return None
var_name, method_name = name.split(".", maxsplit=1)
local_types: dict[str, str] = source_node.metadata.get("local_types") or {}
Comment thread
zaebee marked this conversation as resolved.
class_fqn = local_types.get(var_name)
if not class_fqn:
return None
candidate = f"{class_fqn}.{method_name}"
return self.index.map_to_node_fqn(candidate) or candidate

def _resolve_via_global_symbols(self, name: str, file_path: str | None) -> str | None:
"""Look up name in the global symbol index, preferring same-file candidates."""
candidates = self.index.global_symbols.get(name, [])
if not candidates:
return None
if len(candidates) == 1:
return candidates[0]
if file_path:
same_file = self.index.file_global_symbols.get((file_path, name), [])
if len(same_file) == 1:
return same_file[0]
return None
Loading
Loading