← Files Code Ontology CompanionARCHIVED FILE
skills/manage-code-ontology/scripts/code_ontology_core.py
159 KB · Oct 5, 2026 · 18:30 UTC
#!/usr/bin/env python3
"""Build and explore a privacy-preserving local code ontology.
This utility uses only the Python standard library. It performs static analysis;
it never imports, builds, or executes the target repository.
"""
from __future__ import annotations
import argparse
import ast
import datetime as dt
import hashlib
import html
import json
import os
import re
import stat
import sys
import tempfile
from collections import Counter, deque
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import quote
# Preserve the Explorer 1.0 vocabulary and URIs so previously exported RDF
# remains importable without a migration. Companion provenance uses a separate
# namespace in companion.py.
SCHEMA_VERSION = "1.0"
QUALITY_CONTRACT_VERSION = "1.0"
PLUGIN_VERSION = "0.6.0"
ONTOLOGY_NS = "https://battle-doll.github.io/code-ontology-explorer/schema#"
VISUALIZATION_ASSET_DIR = Path(__file__).resolve().parent.parent / "assets"
VISUALIZATION_MAX_VISIBLE_NODES = 240
VISUALIZATION_DIFF_ITEM_LIMIT = 500
VISUALIZATION_VENDOR_ASSETS = {
"cytoscape": (
"vendor/cytoscape-3.34.0.min.js",
"9c2a3bf2592e0b14a1f7bec07c03a54f16dedf32af9cd0af155c716aa6c87bc3",
),
"elk": (
"vendor/elkjs-0.12.0.bundled.js",
"1222e44f953ce7746af23801e723708f8e6f436b8b377a6a5fc7552f34a307b3",
),
}
SUPPORTED_SUFFIXES = {".java": "Java", ".py": "Python"}
MAX_SOURCE_BYTES = 2 * 1024 * 1024
MAX_SOURCE_FILES = 25_000
MAX_TOTAL_SOURCE_BYTES = 512 * 1024 * 1024
MAX_GRAPH_NODES = 500_000
MAX_GRAPH_EDGES = 1_000_000
MAX_IMPACT_RESULTS = 2_000
MAX_QUERY_RESULTS = 200
DEPENDENCY_RELATIONSHIPS = frozenset({
"CALLS", "INJECTS", "EXTENDS", "IMPLEMENTS", "DECLARES_BEAN",
"READS_POLICY_LEAF", "GUARDS_RUNTIME_BRANCH",
})
GENERIC_CONCEPT_TYPES = frozenset({"FrameworkConcept", "Annotation", "PipelineRole"})
MAX_EDGE_EVIDENCE_ITEMS = 16
MAX_EVIDENCE_LIMITATIONS = 16
MAX_EVIDENCE_PATH_LENGTH = 4_096
MAX_EVIDENCE_LINE = 10_000_000
MAX_PYTHON_AST_NODES = 250_000
MAX_PYTHON_AST_DEPTH = 200
WINDOWS_REPARSE_POINT = 0x400
EVIDENCE_BASES = {
"direct_syntax",
"resolved_static",
"framework_semantic",
"name_heuristic",
}
RUNTIME_STATUSES = {"not_applicable", "runtime_unknown"}
RUNTIME_SENSITIVE_RELATIONSHIPS = {
"INJECTS",
"MANAGED_AS",
"MAY_BE_PROXIED_BY",
"DECLARES_BEAN",
"GUARDS_RUNTIME_BRANCH",
}
EDGE_EVIDENCE_DEFAULTS = {
"DECLARES": ("core.declares", "direct_syntax"),
"IMPORTS": ("core.imports", "direct_syntax"),
"EXTENDS": ("core.extends", "resolved_static"),
"IMPLEMENTS": ("java.implements", "resolved_static"),
"ANNOTATED_BY": ("java.annotation", "direct_syntax"),
"DECORATED_BY": ("python.decorator", "direct_syntax"),
"INJECTS": ("java.spring.injection", "framework_semantic"),
"DECLARES_BEAN": ("java.spring.bean_factory", "framework_semantic"),
"MANAGED_AS": ("java.spring.managed", "framework_semantic"),
"MAY_BE_PROXIED_BY": ("java.spring.proxy_signal", "framework_semantic"),
"CALLS": ("core.calls", "resolved_static"),
"HAS_PIPELINE_ROLE": ("python.pipeline_role", "name_heuristic"),
"READS_POLICY_LEAF": ("java.policy.read", "direct_syntax"),
"DECLARES_RUNTIME_BRANCH": ("java.policy.branch", "direct_syntax"),
"GUARDS_RUNTIME_BRANCH": ("java.policy.guard", "resolved_static"),
}
EXCLUDED_DIRECTORIES = {
".git",
".hg",
".svn",
".idea",
".vscode",
".gradle",
".venv",
"venv",
"env",
"node_modules",
"target",
"build",
"dist",
"out",
".next",
"__pycache__",
".tox",
".mypy_cache",
".pytest_cache",
"coverage",
}
SENSITIVE_NAMES = {
".env",
".env.local",
".env.production",
"credentials",
"credentials.json",
"secrets.json",
"id_rsa",
"id_ed25519",
}
SENSITIVE_SUFFIXES = {
".pem",
".key",
".p12",
".pfx",
".jks",
".keystore",
".kdbx",
}
SPRING_STEREOTYPES = {
"Component",
"Service",
"Repository",
"Controller",
"RestController",
"Configuration",
"SpringBootApplication",
}
SPRING_INJECTION = {"Autowired", "Inject", "Resource"}
SPRING_AOP = {"Aspect", "Before", "After", "AfterReturning", "AfterThrowing", "Around", "Pointcut"}
SPRING_PROXY = {
"Transactional",
"Async",
"Cacheable",
"CacheEvict",
"CachePut",
"Secured",
"PreAuthorize",
"PostAuthorize",
"Retryable",
}
JAVA_ANNOTATION_PATTERN = (
r"@[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)*"
r"(?:\s*\((?:[^()]|\([^()]*\))*\))?"
)
SPRING_ANNOTATION_TYPES = {
"Component": {"org.springframework.stereotype.Component"},
"Service": {"org.springframework.stereotype.Service"},
"Repository": {"org.springframework.stereotype.Repository"},
"Controller": {"org.springframework.stereotype.Controller"},
"RestController": {"org.springframework.web.bind.annotation.RestController"},
"Configuration": {"org.springframework.context.annotation.Configuration"},
"SpringBootApplication": {"org.springframework.boot.autoconfigure.SpringBootApplication"},
"Bean": {"org.springframework.context.annotation.Bean"},
"Autowired": {"org.springframework.beans.factory.annotation.Autowired"},
"Inject": {"jakarta.inject.Inject", "javax.inject.Inject"},
"Resource": {"jakarta.annotation.Resource", "javax.annotation.Resource"},
"Aspect": {"org.aspectj.lang.annotation.Aspect"},
"Before": {"org.aspectj.lang.annotation.Before"},
"After": {"org.aspectj.lang.annotation.After"},
"AfterReturning": {"org.aspectj.lang.annotation.AfterReturning"},
"AfterThrowing": {"org.aspectj.lang.annotation.AfterThrowing"},
"Around": {"org.aspectj.lang.annotation.Around"},
"Pointcut": {"org.aspectj.lang.annotation.Pointcut"},
"Transactional": {
"org.springframework.transaction.annotation.Transactional",
"jakarta.transaction.Transactional",
"javax.transaction.Transactional",
},
"Async": {"org.springframework.scheduling.annotation.Async"},
"Cacheable": {"org.springframework.cache.annotation.Cacheable"},
"CacheEvict": {"org.springframework.cache.annotation.CacheEvict"},
"CachePut": {"org.springframework.cache.annotation.CachePut"},
"Secured": {"org.springframework.security.access.annotation.Secured"},
"PreAuthorize": {"org.springframework.security.access.prepost.PreAuthorize"},
"PostAuthorize": {"org.springframework.security.access.prepost.PostAuthorize"},
"Retryable": {"org.springframework.retry.annotation.Retryable"},
}
JAVA_LANG_TYPES = {
"void",
"boolean",
"byte",
"short",
"int",
"long",
"float",
"double",
"char",
"String",
"Object",
"Integer",
"Long",
"Double",
"Float",
"Short",
"Byte",
"Boolean",
"Character",
"Number",
"Enum",
"Record",
"Class",
"Throwable",
"Exception",
"RuntimeException",
"Iterable",
}
JAVA_UTIL_TYPES = {
"Collection",
"Collections",
"List",
"ArrayList",
"LinkedList",
"Set",
"HashSet",
"Map",
"HashMap",
"Optional",
"Queue",
"Deque",
"Iterator",
}
JAVA_TYPE_RE = re.compile(
rf"(?P<annotations>(?:{JAVA_ANNOTATION_PATTERN}\s*)*)"
r"(?:(?:public|protected|private|abstract|final|static|sealed|non-sealed)\s+)*"
r"(?P<kind>class|interface|enum|record)\s+(?P<name>[A-Za-z_]\w*)",
re.MULTILINE,
)
JAVA_METHOD_RE = re.compile(
r"^[ \t]*"
rf"(?P<annotations>(?:{JAVA_ANNOTATION_PATTERN}\s*)*)"
r"(?:(?:public|protected|private|abstract|final|static|synchronized|native|default|strictfp)\s+)*"
r"(?!return\b|throw\b|new\b|if\b|for\b|while\b|switch\b|catch\b|do\b|else\b)"
r"(?:<[^>{};]+>\s+)?"
r"(?P<return>[A-Za-z_][\w.$<>\[\], ?]*)\s+"
r"(?P<name>[A-Za-z_]\w*)\s*\((?P<params>(?:[^()]|\([^()]*\))*)\)"
r"\s*(?:throws\s+[^{;]+)?(?P<ending>\{|;)",
re.MULTILINE,
)
JAVA_FIELD_RE = re.compile(
rf"(?P<annotations>(?:{JAVA_ANNOTATION_PATTERN}\s*)+)"
r"(?:(?:public|protected|private|final|static|volatile|transient)\s+)*"
r"(?P<type>[A-Za-z_][\w.$<>\[\], ?]*)\s+(?P<name>[A-Za-z_]\w*)\s*(?:=[^;]*)?;",
re.MULTILINE,
)
JAVA_ANNOTATION_RE = re.compile(r"@([A-Za-z_]\w*(?:\.[A-Za-z_]\w*)*)")
JAVA_POLICY_ACCESSORS = ("policyBool", "policyDecimal", "policyInt")
JAVA_POLICY_LEAF_RE = re.compile(
r"[A-Za-z][A-Za-z0-9_-]*(?:\.[A-Za-z][A-Za-z0-9_-]*){1,15}"
)
JAVA_POLICY_READ_RE = re.compile(
r"(?:(?P<variable>[A-Za-z_]\w*)\s*=\s*)?"
r"(?:this\s*\.\s*)?"
r"(?P<accessor>" + "|".join(JAVA_POLICY_ACCESSORS) + r")\s*\("
r"(?:(?![;{}]).)*?"
r'"(?P<leaf>[A-Za-z][A-Za-z0-9_-]*(?:\.[A-Za-z][A-Za-z0-9_-]*){1,15})"',
re.DOTALL,
)
JAVA_ASSIGNMENT_RE = re.compile(
r"(?:^|[;{}])\s*"
r"(?:final\s+)?"
r"(?:[A-Za-z_][\w.$<>\[\], ?]*\s+)?"
r"(?P<variable>[A-Za-z_]\w*)\s*=(?!=)"
r"(?P<expression>[^;{}]*);",
re.MULTILINE,
)
JAVA_CONTROL_RE = re.compile(r"\b(?P<kind>if|while|switch|for)\s*\(")
JAVA_CALL_RE = re.compile(
r"(?<![\w$.])"
r"(?:(?P<qualifier>[A-Za-z_]\w*)\s*\.\s*)?"
r"(?P<name>[A-Za-z_]\w*)\s*\("
)
JAVA_ANNOTATION_TOKEN_RE = re.compile(r"@[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)*")
JAVA_NON_CALL_KEYWORDS = {
"assert",
"catch",
"do",
"else",
"for",
"if",
"new",
"super",
"switch",
"synchronized",
"this",
"throw",
"try",
"while",
}
class OntologyError(RuntimeError):
"""Expected, user-actionable failure."""
def _iso_now() -> str:
return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
def _resolve_dir(raw_path: str, label: str) -> Path:
raw = Path(raw_path).expanduser()
try:
directory_stat = raw.lstat()
except OSError:
raise OntologyError(f"{label} is not a readable directory.")
if _is_link_like(directory_stat):
raise OntologyError(f"{label} may not be a symbolic link or reparse point.")
if not stat.S_ISDIR(directory_stat.st_mode):
raise OntologyError(f"{label} is not a readable directory.")
return raw.resolve()
def _is_relative_to(path: Path, parent: Path) -> bool:
try:
path.relative_to(parent)
return True
except ValueError:
return False
def _is_sensitive_file(path: Path) -> bool:
lowered = path.name.lower()
normalized_stem = path.stem.lower().replace("-", "_").replace(".", "_")
if lowered in SENSITIVE_NAMES or path.suffix.lower() in SENSITIVE_SUFFIXES:
return True
sensitive_stems = {
"api_key",
"access_token",
"auth_token",
"credential",
"credentials",
"id_ed25519",
"id_rsa",
"key",
"private_key",
"secret",
"secrets",
"token",
}
return (
lowered.startswith(".env.")
or "credential" in lowered
or "secret" in lowered
or normalized_stem in sensitive_stems
or normalized_stem.endswith(("_token", "_secret", "_key", "_credentials"))
)
def _is_link_like(file_stat: os.stat_result) -> bool:
attributes = getattr(file_stat, "st_file_attributes", 0)
return stat.S_ISLNK(file_stat.st_mode) or bool(attributes & WINDOWS_REPARSE_POINT)
def discover_sources(repo: Path) -> tuple[list[Path], Counter[str]]:
"""Return safe source files and skip statistics without following symlinks."""
sources: list[Path] = []
skipped: Counter[str] = Counter()
total_source_bytes = 0
for current, directories, filenames in os.walk(repo, topdown=True, followlinks=False):
current_path = Path(current)
kept_dirs: list[str] = []
for name in sorted(directories):
candidate = current_path / name
if name in EXCLUDED_DIRECTORIES:
skipped["excluded_directory"] += 1
continue
try:
directory_stat = candidate.lstat()
except OSError:
skipped["unreadable"] += 1
continue
if _is_link_like(directory_stat):
skipped["symlink_or_reparse"] += 1
elif not stat.S_ISDIR(directory_stat.st_mode):
skipped["special_file"] += 1
else:
kept_dirs.append(name)
directories[:] = kept_dirs
for name in sorted(filenames):
candidate = current_path / name
if _is_sensitive_file(candidate):
skipped["sensitive_name"] += 1
continue
if candidate.suffix.lower() not in SUPPORTED_SUFFIXES:
continue
try:
file_stat = candidate.lstat()
except OSError:
skipped["unreadable"] += 1
continue
if _is_link_like(file_stat):
skipped["symlink_or_reparse"] += 1
continue
if not stat.S_ISREG(file_stat.st_mode):
skipped["special_file"] += 1
continue
size = file_stat.st_size
if size > MAX_SOURCE_BYTES:
skipped["too_large"] += 1
continue
if len(sources) >= MAX_SOURCE_FILES:
raise OntologyError(
f"Repository exceeds the {MAX_SOURCE_FILES}-source-file limit."
)
if total_source_bytes + size > MAX_TOTAL_SOURCE_BYTES:
raise OntologyError(
"Repository exceeds the total supported source-byte limit."
)
sources.append(candidate)
total_source_bytes += size
return sorted(sources), skipped
def _same_file(left: os.stat_result, right: os.stat_result) -> bool:
try:
return os.path.samestat(left, right)
except (AttributeError, OSError):
return (left.st_dev, left.st_ino) == (right.st_dev, right.st_ino)
def _stable_file_metadata(file_stat: os.stat_result) -> tuple[int, ...]:
metadata = (
file_stat.st_size,
getattr(file_stat, "st_mtime_ns", int(file_stat.st_mtime * 1_000_000_000)),
)
if os.name == "nt":
# Python 3.12 deprecates st_ctime[_ns] on Windows, where it still means
# creation time and may differ between path and handle stat calls. File
# identity is checked separately; size and mtime remain change guards.
return metadata
return metadata + (
getattr(file_stat, "st_ctime_ns", int(file_stat.st_ctime * 1_000_000_000)),
)
def _safe_read_bytes(path: Path) -> bytes:
try:
initial_stat = path.lstat()
except OSError as exc:
reason = exc.strerror or exc.__class__.__name__
raise OntologyError(f"Could not read {path.name}: {reason}") from exc
if _is_link_like(initial_stat) or not stat.S_ISREG(initial_stat.st_mode):
raise OntologyError(f"Refusing to read a linked or non-regular source: {path.name}")
flags = os.O_RDONLY
flags |= getattr(os, "O_BINARY", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
flags |= getattr(os, "O_NONBLOCK", 0)
try:
descriptor = os.open(path, flags)
except OSError as exc:
reason = exc.strerror or exc.__class__.__name__
raise OntologyError(f"Could not read {path.name}: {reason}") from exc
try:
handle = os.fdopen(descriptor, "rb", closefd=True)
except Exception:
os.close(descriptor)
raise
try:
with handle:
opened_stat = os.fstat(handle.fileno())
if _is_link_like(opened_stat) or not stat.S_ISREG(opened_stat.st_mode):
raise OntologyError(f"Refusing to read a non-regular source: {path.name}")
if not _same_file(initial_stat, opened_stat):
raise OntologyError(f"Source changed before it could be read: {path.name}")
if _stable_file_metadata(initial_stat) != _stable_file_metadata(opened_stat):
raise OntologyError(f"Source changed before it could be read: {path.name}")
if opened_stat.st_size > MAX_SOURCE_BYTES:
raise OntologyError(
f"Source exceeds the {MAX_SOURCE_BYTES}-byte limit: {path.name}"
)
raw = handle.read(MAX_SOURCE_BYTES + 1)
final_stat = os.fstat(handle.fileno())
except OSError as exc:
reason = exc.strerror or exc.__class__.__name__
raise OntologyError(f"Could not read {path.name}: {reason}") from exc
if len(raw) > MAX_SOURCE_BYTES:
raise OntologyError(f"Source exceeds the {MAX_SOURCE_BYTES}-byte limit: {path.name}")
if not _same_file(opened_stat, final_stat):
raise OntologyError(f"Source changed while it was being read: {path.name}")
if _stable_file_metadata(opened_stat) != _stable_file_metadata(final_stat):
raise OntologyError(f"Source changed while it was being read: {path.name}")
try:
current_stat = path.lstat()
except OSError as exc:
reason = exc.strerror or exc.__class__.__name__
raise OntologyError(f"Could not verify {path.name}: {reason}") from exc
if _is_link_like(current_stat) or not stat.S_ISREG(current_stat.st_mode):
raise OntologyError(f"Refusing to read a linked or non-regular source: {path.name}")
if not _same_file(final_stat, current_stat):
raise OntologyError(f"Source changed while it was being read: {path.name}")
if _stable_file_metadata(final_stat) != _stable_file_metadata(current_stat):
raise OntologyError(f"Source changed while it was being read: {path.name}")
return raw
def _safe_read(path: Path) -> str:
raw = _safe_read_bytes(path)
return raw.decode("utf-8", errors="replace")
def _node_id(language: str, kind: str, qualified_name: str) -> str:
return f"{language.lower()}:{kind.lower()}:{qualified_name}"
def _line_span(source: str, start: int, end: int | None = None) -> tuple[int, int]:
"""Return a stable one-based line span without retaining source text."""
bounded_start = max(0, min(int(start), len(source)))
bounded_end = max(bounded_start, min(int(end if end is not None else start), len(source)))
return source.count("\n", 0, bounded_start) + 1, source.count("\n", 0, bounded_end) + 1
def _portable_relative_path(value: str) -> bool:
if (
not value
or len(value) > MAX_EVIDENCE_PATH_LENGTH
or any(ord(character) < 32 or ord(character) == 127 for character in value)
or value.startswith(("/", "\\"))
or re.match(r"^[A-Za-z]:[\\/]", value)
):
return False
normalized = value.replace("\\", "/")
return not Path(normalized).is_absolute() and ".." not in normalized.split("/")
def _edge_evidence_key(item: dict[str, Any]) -> str:
return json.dumps(
item,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
def _bounded_edge_evidence(values: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
"""Deduplicate, canonically order, and bound evidence independent of set order."""
unique: dict[str, dict[str, Any]] = {}
for item in values:
key = _edge_evidence_key(item)
unique.setdefault(key, item)
return [unique[key] for key in sorted(unique)[:MAX_EDGE_EVIDENCE_ITEMS]]
def _normalized_edge_evidence(
*,
edge_type: str,
rule_id: str | None,
basis: str | None,
runtime_status: str | None,
path: str | None,
line_start: int | None,
line_end: int | None,
limitations: Iterable[str] | None,
) -> dict[str, Any]:
default_rule, default_basis = EDGE_EVIDENCE_DEFAULTS.get(
edge_type,
(f"core.{edge_type.casefold()}", "direct_syntax"),
)
selected_rule = rule_id or default_rule
selected_basis = basis or default_basis
selected_runtime = runtime_status or (
"runtime_unknown" if edge_type in RUNTIME_SENSITIVE_RELATIONSHIPS else "not_applicable"
)
if not re.fullmatch(r"[a-z][a-z0-9_.-]{2,79}", selected_rule):
raise OntologyError(f"Invalid relationship evidence rule identifier: {selected_rule}")
if selected_basis not in EVIDENCE_BASES:
raise OntologyError(f"Unsupported relationship evidence basis: {selected_basis}")
if selected_runtime not in RUNTIME_STATUSES:
raise OntologyError(f"Unsupported relationship runtime status: {selected_runtime}")
item: dict[str, Any] = {
"rule_id": selected_rule,
"basis": selected_basis,
"runtime_status": selected_runtime,
}
if isinstance(path, str) and _portable_relative_path(path):
item["path"] = path.replace("\\", "/")
if (
isinstance(line_start, int)
and not isinstance(line_start, bool)
and 1 <= line_start <= MAX_EVIDENCE_LINE
):
item["line_start"] = line_start
selected_line_end = line_end if isinstance(line_end, int) else line_start
item["line_end"] = min(
MAX_EVIDENCE_LINE,
max(line_start, int(selected_line_end)),
)
normalized_limitations = {
value
for value in (limitations or ())
if isinstance(value, str) and re.fullmatch(r"[a-z][a-z0-9_.-]{2,79}", value)
}
required_limitation = None
if selected_runtime == "runtime_unknown":
required_limitation = "runtime.activation_not_observed"
normalized_limitations.add(required_limitation)
bounded_limitations = sorted(normalized_limitations)[:MAX_EVIDENCE_LIMITATIONS]
if required_limitation and required_limitation not in bounded_limitations:
bounded_limitations[-1:] = [required_limitation]
bounded_limitations.sort()
if bounded_limitations:
item["limitations"] = bounded_limitations
return item
def _adapter_quality(language: str) -> dict[str, Any]:
"""Describe bounded analyzer support without implying completeness."""
if language == "Java":
return {
"status": "partial",
"capabilities": {
"annotations": "partial",
"calls": "partial",
"declarations": "partial",
"dependency_injection": "partial",
"explicit_type_imports": "supported",
"imports": "partial",
"inheritance": "partial",
"runtime_activation": "unsupported",
"runtime_dispatch": "unsupported",
},
"unsupported_runtime": [
"active_application_context",
"dynamic_dispatch",
"generated_code",
"reflection",
"runtime_conditions",
],
}
if language == "Python":
return {
"status": "partial",
"capabilities": {
"calls": "partial",
"decorators": "partial",
"declarations": "partial",
"imports": "partial",
"inheritance": "partial",
"pipeline_roles": "partial",
"runtime_dispatch": "unsupported",
"runtime_imports": "unsupported",
},
"unsupported_runtime": [
"dynamic_imports",
"descriptor_dispatch",
"generated_code",
"monkey_patching",
"runtime_metaprogramming",
],
}
return {
"status": "unsupported",
"capabilities": {},
"unsupported_runtime": ["adapter_not_available"],
}
class Graph:
def __init__(self, repository_name: str) -> None:
self.repository_name = repository_name
self.nodes: dict[str, dict[str, Any]] = {}
self.edges: set[tuple[str, str, str]] = set()
self.edge_evidence: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
self.warnings: list[dict[str, str]] = []
def add_node(
self,
node_id: str,
node_type: str,
name: str,
language: str,
path: str | None = None,
qualified_name: str | None = None,
metadata: dict[str, Any] | None = None,
) -> str:
record: dict[str, Any] = {
"id": node_id,
"type": node_type,
"name": name,
"language": language,
}
if path:
record["path"] = path
if qualified_name:
record["qualified_name"] = qualified_name
if metadata:
clean_metadata = {
key: value
for key, value in metadata.items()
if value not in (None, "", [], {})
}
if clean_metadata:
record["metadata"] = clean_metadata
if node_id in self.nodes:
current = self.nodes[node_id]
if current.get("type") in {
"ExternalModule",
"ExternalType",
"ExternalCallable",
} and record.get("type") not in {
"ExternalModule",
"ExternalType",
"ExternalCallable",
}:
current["type"] = record["type"]
for key, value in record.items():
if key not in current or not current[key]:
current[key] = value
else:
if len(self.nodes) >= MAX_GRAPH_NODES:
raise OntologyError(
f"Ontology exceeds the {MAX_GRAPH_NODES}-node safety limit."
)
self.nodes[node_id] = record
return node_id
def add_edge(
self,
source: str,
target: str,
edge_type: str,
*,
rule_id: str | None = None,
basis: str | None = None,
runtime_status: str | None = None,
path: str | None = None,
line_start: int | None = None,
line_end: int | None = None,
limitations: Iterable[str] | None = None,
) -> None:
if source != target:
edge = (source, target, edge_type)
if (
edge not in self.edges
and len(self.edges) >= MAX_GRAPH_EDGES
):
raise OntologyError(
f"Ontology exceeds the {MAX_GRAPH_EDGES}-edge safety limit."
)
self.edges.add(edge)
source_node = self.nodes.get(source, {})
target_node = self.nodes.get(target, {})
selected_path = path or source_node.get("path") or target_node.get("path")
source_metadata = source_node.get("metadata", {})
if not isinstance(source_metadata, dict):
source_metadata = {}
selected_line_start = line_start or source_metadata.get("line_start")
selected_line_end = line_end or source_metadata.get("line_end")
item = _normalized_edge_evidence(
edge_type=edge_type,
rule_id=rule_id,
basis=basis,
runtime_status=runtime_status,
path=selected_path if isinstance(selected_path, str) else None,
line_start=(
selected_line_start if isinstance(selected_line_start, int) else None
),
line_end=(selected_line_end if isinstance(selected_line_end, int) else None),
limitations=limitations,
)
existing = self.edge_evidence.setdefault(edge, [])
if item not in existing:
self.edge_evidence[edge] = _bounded_edge_evidence((*existing, item))
def add_external_type(self, language: str, qualified_name: str) -> str:
simple_name = qualified_name.rsplit(".", 1)[-1]
return self.add_node(
_node_id(language, "external_type", qualified_name),
"ExternalType",
simple_name,
language,
qualified_name=qualified_name,
)
def add_annotation(
self,
annotation: str,
*,
qualified_name: str | None = None,
semantic_name: str | None = None,
) -> str:
groups: list[str] = []
if semantic_name in SPRING_STEREOTYPES:
groups.append("SpringBean")
if semantic_name in SPRING_INJECTION:
groups.append("DependencyInjection")
if semantic_name in SPRING_AOP:
groups.append("AspectOrAdvice")
if semantic_name in SPRING_PROXY:
groups.append("ProxyOrInterceptor")
identity = qualified_name or annotation
return self.add_node(
f"framework:annotation:{identity}",
"FrameworkAnnotation",
f"@{annotation}",
"Framework",
qualified_name=identity,
metadata={"semantic_groups": groups},
)
def add_warning(self, relative_path: str, message: str) -> None:
self.warnings.append({"path": relative_path, "message": message})
def reconcile_references(self) -> None:
"""Redirect external Java placeholders to analyzed internal types."""
internal_by_name: dict[tuple[str, str], str] = {}
internal_by_simple: dict[tuple[str, str], list[str]] = {}
for node_id, node in self.nodes.items():
if node["type"] in {
"Class",
"Interface",
"Enum",
"Record",
"Function",
"AsyncFunction",
"Method",
"AsyncMethod",
}:
qualified_name = node.get("qualified_name")
if qualified_name:
internal_by_name[(node["language"], qualified_name)] = node_id
internal_by_simple.setdefault(
(node["language"], node["name"]), []
).append(node_id)
redirects: dict[str, str] = {}
for node_id, node in self.nodes.items():
if node["type"] not in {"ExternalType", "ExternalCallable"}:
continue
qualified_name = node.get("qualified_name")
target = internal_by_name.get((node["language"], qualified_name))
if (
target is None
and node["type"] == "ExternalType"
and node["language"] == "Java"
and isinstance(qualified_name, str)
and "." not in qualified_name
):
candidates = internal_by_simple.get(("Java", node["name"]), [])
if len(candidates) == 1:
target = candidates[0]
if target:
redirects[node_id] = target
if not redirects:
return
reconciled: set[tuple[str, str, str]] = set()
reconciled_evidence: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
for source, target, edge_type in self.edges:
new_source = redirects.get(source, source)
new_target = redirects.get(target, target)
if new_source != new_target:
old_edge = (source, target, edge_type)
new_edge = (new_source, new_target, edge_type)
reconciled.add(new_edge)
merged = reconciled_evidence.setdefault(new_edge, [])
for item in self.edge_evidence.get(old_edge, []):
if item not in merged:
merged.append(item)
self.edges = reconciled
self.edge_evidence = {
edge: _bounded_edge_evidence(items)
for edge, items in reconciled_evidence.items()
}
for node_id in redirects:
self.nodes.pop(node_id, None)
def document(self, source_counts: Counter[str], skipped: Counter[str]) -> dict[str, Any]:
self.reconcile_references()
nodes = sorted(self.nodes.values(), key=lambda item: item["id"])
edges = []
for source, target, edge_type in sorted(self.edges):
edge = (source, target, edge_type)
evidence = _bounded_edge_evidence(self.edge_evidence.get(edge, []))
edges.append(
{
"source": source,
"target": target,
"type": edge_type,
"evidence": evidence,
}
)
documented_edges = sum(bool(edge["evidence"]) for edge in edges)
span_documented_edges = sum(
any("path" in item and "line_start" in item for item in edge["evidence"])
for edge in edges
)
basis_counts = Counter(
item["basis"] for edge in edges for item in edge["evidence"]
)
runtime_status_counts = Counter(
item["runtime_status"] for edge in edges for item in edge["evidence"]
)
adapters = {
language: {
**_adapter_quality(language),
"detected": language in source_counts,
}
for language in ("Java", "Python")
}
return {
"schema_version": SCHEMA_VERSION,
"generated_at": _iso_now(),
"generator": {"name": "Code Ontology Companion", "version": PLUGIN_VERSION},
"repository": {
"name": self.repository_name,
"path_policy": "repository-relative-only",
},
"privacy": {
"contains_source_text": False,
"contains_comments": False,
"contains_absolute_paths": False,
"contains_file_hashes": False,
"contains_arbitrary_string_literals": False,
"contains_policy_identifiers": any(
node["type"] == "PolicyLeaf" for node in nodes
),
},
"statistics": {
"source_files": dict(sorted(source_counts.items())),
"nodes": len(nodes),
"edges": len(edges),
"node_types": dict(sorted(Counter(node["type"] for node in nodes).items())),
"edge_types": dict(sorted(Counter(edge["type"] for edge in edges).items())),
"skipped": dict(sorted(skipped.items())),
"warnings": len(self.warnings),
},
"quality": {
"contract_version": QUALITY_CONTRACT_VERSION,
"relationship_evidence": {
"total_edges": len(edges),
"documented_edges": documented_edges,
"missing_evidence": len(edges) - documented_edges,
"coverage_percent": round(
(documented_edges * 100.0 / len(edges)) if edges else 100.0,
3,
),
"source_span_edges": span_documented_edges,
"source_span_coverage_percent": round(
(span_documented_edges * 100.0 / len(edges)) if edges else 100.0,
3,
),
"basis_counts": dict(sorted(basis_counts.items())),
"runtime_status_counts": dict(sorted(runtime_status_counts.items())),
},
"adapters": adapters,
"interpretation": (
"Qualitative static evidence, not a probability or runtime verdict. "
"Unsupported and runtime-unknown capabilities require independent evidence."
),
},
"nodes": nodes,
"edges": edges,
"warnings": sorted(self.warnings, key=lambda item: (item["path"], item["message"])),
}
def _annotations(text: str) -> list[str]:
return sorted(set(JAVA_ANNOTATION_RE.findall(text)))
def _annotation_details(
annotation: str,
imports: dict[str, str],
wildcard_imports: set[str],
package_name: str,
same_package_types: set[str] | None = None,
) -> tuple[str, str, str | None]:
simple_name = annotation.rsplit(".", 1)[-1]
qualified_name: str | None
if "." in annotation:
qualified_name = annotation
else:
qualified_name = imports.get(simple_name)
accepted = SPRING_ANNOTATION_TYPES.get(simple_name, set())
semantic_name = simple_name if qualified_name in accepted else None
if (
semantic_name is None
and qualified_name is None
and len(wildcard_imports) == 1
and same_package_types is not None
and simple_name not in same_package_types
):
wildcard = next(iter(wildcard_imports))
candidate = f"{wildcard}.{simple_name}"
if candidate in accepted:
qualified_name = candidate
semantic_name = simple_name
if qualified_name is None:
qualified_name = f"{package_name}.{simple_name}" if package_name else simple_name
return simple_name, qualified_name, semantic_name
def _resolved_annotations(
text: str,
imports: dict[str, str],
wildcard_imports: set[str],
package_name: str,
same_package_types: set[str] | None = None,
) -> list[tuple[str, str, str | None]]:
return [
_annotation_details(
annotation,
imports,
wildcard_imports,
package_name,
same_package_types,
)
for annotation in _annotations(text)
]
def _java_package_and_declared_types(source: str) -> tuple[str, set[str]]:
stripped = _strip_java_comments_and_literals(source)
package_match = re.search(r"^\s*package\s+([\w.]+)\s*;", stripped, re.MULTILINE)
package_name = package_match.group(1) if package_match else ""
return package_name, {match.group("name") for match in JAVA_TYPE_RE.finditer(stripped)}
def _strip_java_comments_and_literals(source: str) -> str:
"""Replace comments and string/char contents while keeping code positions."""
result = list(source)
index = 0
state = "code"
while index < len(source):
char = source[index]
next_char = source[index + 1] if index + 1 < len(source) else ""
if state == "code":
if char == "/" and next_char == "/":
result[index] = result[index + 1] = " "
state = "line_comment"
index += 2
continue
if char == "/" and next_char == "*":
result[index] = result[index + 1] = " "
state = "block_comment"
index += 2
continue
if source.startswith('"""', index):
result[index] = result[index + 1] = result[index + 2] = " "
state = "text_block"
index += 3
continue
if char == '"':
result[index] = " "
state = "string"
elif char == "'":
result[index] = " "
state = "char"
elif state == "line_comment":
if char == "\n":
state = "code"
else:
result[index] = " "
elif state == "block_comment":
if char == "*" and next_char == "/":
result[index] = result[index + 1] = " "
state = "code"
index += 2
continue
if char != "\n":
result[index] = " "
elif state == "text_block":
if source.startswith('"""', index):
result[index] = result[index + 1] = result[index + 2] = " "
state = "code"
index += 3
continue
if char != "\n":
result[index] = " "
elif state in {"string", "char"}:
quote_char = '"' if state == "string" else "'"
if char == "\\":
result[index] = " "
if index + 1 < len(source):
if source[index + 1] != "\n":
result[index + 1] = " "
index += 2
continue
if char == quote_char:
result[index] = " "
state = "code"
elif char != "\n":
result[index] = " "
index += 1
return "".join(result)
def _java_policy_scan_source(source: str) -> str:
"""Keep code and safe dotted policy identifiers, masking all other literals."""
result = list(source)
index = 0
state = "code"
literal_start = -1
while index < len(source):
char = source[index]
next_char = source[index + 1] if index + 1 < len(source) else ""
if state == "code":
if char == "/" and next_char == "/":
result[index] = result[index + 1] = " "
state = "line_comment"
index += 2
continue
if char == "/" and next_char == "*":
result[index] = result[index + 1] = " "
state = "block_comment"
index += 2
continue
if source.startswith('"""', index):
result[index] = result[index + 1] = result[index + 2] = " "
state = "text_block"
index += 3
continue
if char == '"':
literal_start = index
state = "string"
elif char == "'":
result[index] = " "
state = "char"
elif state == "line_comment":
if char == "\n":
state = "code"
else:
result[index] = " "
elif state == "block_comment":
if char == "*" and next_char == "/":
result[index] = result[index + 1] = " "
state = "code"
index += 2
continue
if char != "\n":
result[index] = " "
elif state == "char":
if char == "\\":
result[index] = " "
if index + 1 < len(source):
if source[index + 1] != "\n":
result[index + 1] = " "
index += 2
continue
result[index] = " "
if char == "'":
state = "code"
elif state == "text_block":
if source.startswith('"""', index):
result[index] = result[index + 1] = result[index + 2] = " "
state = "code"
index += 3
continue
if char != "\n":
result[index] = " "
elif state == "string":
if char == "\\":
if index + 1 < len(source):
index += 2
continue
elif char == '"':
literal = source[literal_start + 1 : index]
if not JAVA_POLICY_LEAF_RE.fullmatch(literal):
for masked in range(literal_start, index + 1):
if source[masked] != "\n":
result[masked] = " "
state = "code"
index += 1
if state == "string" and literal_start >= 0:
for masked in range(literal_start, len(source)):
if source[masked] != "\n":
result[masked] = " "
return "".join(result)
def _matching_java_parenthesis(source: str, opening: int, limit: int) -> int:
depth = 0
for index in range(opening, min(limit, len(source))):
if source[index] == "(":
depth += 1
elif source[index] == ")":
depth -= 1
if depth == 0:
return index
return min(limit, len(source))
def _java_identifier_used(text: str, identifier: str) -> bool:
return bool(re.search(rf"\b{re.escape(identifier)}\b", text))
def _java_branch_has_body(source: str, condition_end: int) -> bool:
index = condition_end + 1
while index < len(source) and source[index].isspace():
index += 1
if index >= len(source):
return False
if source[index] == "{":
closing = _matching_java_brace(source, index)
return bool(source[index + 1 : closing].strip())
statement_end = source.find(";", index)
if statement_end < 0:
return False
return bool(source[index:statement_end].strip())
def _java_previous_identifier(source: str, position: int, lower_bound: int) -> str:
index = position - 1
while index >= lower_bound and source[index].isspace():
index -= 1
end = index + 1
while index >= lower_bound and (source[index].isalnum() or source[index] in "_$"):
index -= 1
return source[index + 1 : end]
def _java_call_argument_count(
source: str,
opening: int,
closing: int,
) -> int | None:
"""Count top-level call arguments without treating literal commas as syntax."""
if not (0 <= opening < closing < len(source)):
return None
parentheses = brackets = braces = angles = 0
commas = 0
has_token = False
state = "code"
index = opening + 1
while index < closing:
char = source[index]
next_char = source[index + 1] if index + 1 < closing else ""
if state == "code":
if char == "/" and next_char == "/":
state = "line_comment"
index += 2
continue
if char == "/" and next_char == "*":
state = "block_comment"
index += 2
continue
if source.startswith('\"\"\"', index):
has_token = True
state = "text_block"
index += 3
continue
if char == '"':
has_token = True
state = "string"
index += 1
continue
if char == "'":
has_token = True
state = "char"
index += 1
continue
if char.isspace():
index += 1
continue
if char == "(":
parentheses += 1
elif char == ")":
if not parentheses:
return None
parentheses -= 1
elif char == "[":
brackets += 1
elif char == "]":
if not brackets:
return None
brackets -= 1
elif char == "{":
braces += 1
elif char == "}":
if not braces:
return None
braces -= 1
elif char == "<":
angles += 1
elif char == ">" and angles:
angles -= 1
elif char == "," and not (parentheses or brackets or braces or angles):
commas += 1
index += 1
continue
has_token = True
elif state == "line_comment":
if char == "\n":
state = "code"
elif state == "block_comment":
if char == "*" and next_char == "/":
state = "code"
index += 2
continue
elif state == "text_block":
if source.startswith('\"\"\"', index):
state = "code"
index += 3
continue
elif state in {"string", "char"}:
quote_char = '"' if state == "string" else "'"
if char == "\\":
index += 2
continue
if char == quote_char:
state = "code"
index += 1
if state != "code" or any((parentheses, brackets, braces, angles)):
return None
return commas + 1 if has_token else 0
def _java_annotation_spans(source: str, start: int, end: int) -> list[tuple[int, int]]:
spans: list[tuple[int, int]] = []
for match in JAVA_ANNOTATION_TOKEN_RE.finditer(source, start, end):
cursor = match.end()
while cursor < end and source[cursor].isspace():
cursor += 1
if cursor < end and source[cursor] == "(":
closing = _matching_java_parenthesis(source, cursor, end)
spans.append((match.start(), min(closing + 1, end)))
else:
spans.append((match.start(), match.end()))
return spans
def _add_java_call_edges(
graph: Graph,
methods: list[dict[str, Any]],
declared_methods: list[dict[str, Any]],
code_source: str,
original_source: str,
imports: dict[str, str],
relative_path: str,
) -> None:
declared_by_owner: dict[tuple[str, str, int], set[str]] = {}
for declared in declared_methods:
key = (
str(declared["owner_id"]),
str(declared["name"]),
int(declared["parameter_count"]),
)
declared_by_owner.setdefault(key, set()).add(str(declared["id"]))
nested_ranges: dict[str, list[tuple[int, int]]] = {
str(method["id"]): [] for method in methods
}
method_stack: list[dict[str, Any]] = []
for method in sorted(methods, key=lambda item: int(item["body_start"])):
method_start = int(method["body_start"])
method_end = int(method["body_end"])
while method_stack and not (
int(method_stack[-1]["body_start"]) <= method_start
and method_end <= int(method_stack[-1]["body_end"])
):
method_stack.pop()
if method_stack:
nested_ranges[str(method_stack[-1]["id"])].append(
(method_start, method_end)
)
method_stack.append(method)
for method in methods:
body_start = int(method["body_start"])
body_end = int(method["body_end"])
# A value binding may legally shadow an imported type. Decline that
# qualifier rather than claiming a static type call from spelling alone.
shadowed = set(method.get("parameter_names", []))
shadowed.update(
field.group("name") for field in JAVA_FIELD_RE.finditer(code_source)
if int(method.get("owner_body_start", 0)) <= field.start() < int(method.get("owner_body_end", len(code_source)))
and not any(int(other["body_start"]) <= field.start() < int(other["body_end"]) for other in methods)
)
body = code_source[body_start:body_end]
for imported_name in imports:
if re.search(r"\b[A-Za-z_$][\w$<>.\[\],?]*\s+" + re.escape(imported_name) + r"\b\s*(?:=|;|:|,|\))", body):
shadowed.add(imported_name)
annotation_spans = _java_annotation_spans(code_source, body_start, body_end)
for match in JAVA_CALL_RE.finditer(code_source, body_start, body_end):
position = match.start()
if any(span_start <= position < span_end for span_start, span_end in annotation_spans):
continue
if any(
nested_start <= position < nested_end
for nested_start, nested_end in nested_ranges[str(method["id"])]
):
continue
name = match.group("name")
qualifier = match.group("qualifier")
if name in JAVA_NON_CALL_KEYWORDS:
continue
if _java_previous_identifier(code_source, position, body_start) == "new":
continue
opening = code_source.find("(", match.start(), match.end())
closing = _matching_java_parenthesis(code_source, opening, body_end)
if closing >= body_end or code_source[closing] != ")":
continue
cursor = closing + 1
while cursor < body_end and code_source[cursor].isspace():
cursor += 1
if cursor < body_end and (
code_source[cursor] == "{"
or re.match(r"throws\b", code_source[cursor:body_end])
):
continue
call_line_start, call_line_end = _line_span(
code_source,
position,
closing,
)
if qualifier in {None, "this"}:
argument_count = _java_call_argument_count(
original_source,
opening,
closing,
)
if argument_count is None:
continue
targets = declared_by_owner.get(
(str(method["owner_id"]), name, argument_count),
set(),
)
if len(targets) == 1:
graph.add_edge(
str(method["id"]),
next(iter(targets)),
"CALLS",
rule_id="java.call.same_owner",
basis="resolved_static",
path=relative_path,
line_start=call_line_start,
line_end=call_line_end,
limitations=("java.dynamic_dispatch_not_resolved",),
)
continue
imported_type = imports.get(qualifier)
if imported_type is None or qualifier in shadowed:
continue
qualified_name = f"{imported_type}.{name}"
callable_id = graph.add_node(
_node_id("java", "callable", qualified_name),
"ExternalCallable",
name,
"Java",
qualified_name=qualified_name,
)
graph.add_edge(
str(method["id"]),
callable_id,
"CALLS",
rule_id="java.call.imported_static",
basis="resolved_static",
path=relative_path,
line_start=call_line_start,
line_end=call_line_end,
limitations=("java.external_overload_not_resolved",),
)
def _add_java_policy_runtime_edges(
graph: Graph,
method: dict[str, Any],
policy_source: str,
code_source: str,
relative_path: str,
) -> None:
body_start = int(method["body_start"])
body_end = int(method["body_end"])
policy_body = policy_source[body_start:body_end]
code_body = code_source[body_start:body_end]
reads: list[dict[str, Any]] = []
for match in JAVA_POLICY_READ_RE.finditer(policy_body):
leaf = match.group("leaf")
if not JAVA_POLICY_LEAF_RE.fullmatch(leaf):
continue
leaf_id = graph.add_node(
_node_id("policy", "leaf", leaf),
"PolicyLeaf",
leaf.rsplit(".", 1)[-1],
"Policy",
qualified_name=leaf,
metadata={"accessor": match.group("accessor")},
)
read_line_start, read_line_end = _line_span(
policy_source,
body_start + match.start(),
body_start + match.end(),
)
graph.add_edge(
method["id"],
leaf_id,
"READS_POLICY_LEAF",
rule_id="java.policy.read",
basis="direct_syntax",
path=relative_path,
line_start=read_line_start,
line_end=read_line_end,
)
reads.append(
{
"leaf": leaf,
"leaf_id": leaf_id,
"variable": match.group("variable"),
"start": match.start(),
"end": match.end(),
}
)
if not reads:
return
assignments = [
{
"start": match.start(),
"end": match.end(),
"variable": match.group("variable"),
"expression": match.group("expression"),
}
for match in JAVA_ASSIGNMENT_RE.finditer(code_body)
]
conditions: list[tuple[int, str, str, int, int]] = []
for ordinal, match in enumerate(JAVA_CONTROL_RE.finditer(code_body), 1):
opening = code_body.find("(", match.start(), match.end())
closing = _matching_java_parenthesis(code_body, opening, len(code_body))
if closing >= len(code_body):
continue
conditions.append(
(
ordinal,
match.group("kind"),
code_body[opening + 1 : closing],
opening,
closing,
)
)
for read in reads:
for ordinal, control_kind, condition, opening, closing in conditions:
direct_call = read["start"] >= opening and read["end"] <= closing
if not direct_call and opening < read["end"]:
continue
tainted = {read["variable"]} if read["variable"] else set()
for assignment in assignments:
if (
assignment["start"] < read["end"]
or assignment["end"] > opening
):
continue
expression_is_tainted = any(
_java_identifier_used(assignment["expression"], variable)
for variable in tainted
)
assigned = assignment["variable"]
if assigned in tainted and not expression_is_tainted:
tainted.remove(assigned)
if expression_is_tainted:
tainted.add(assigned)
if not direct_call and not any(
_java_identifier_used(condition, variable) for variable in tainted
):
continue
if not _java_branch_has_body(code_body, closing):
continue
branch_qualified = f"{method['qualified_name']}#branch:{ordinal}"
branch_id = graph.add_node(
_node_id("java", "runtime_branch", branch_qualified),
"RuntimeBranch",
f"{control_kind} branch {ordinal}",
"Java",
path=relative_path,
qualified_name=branch_qualified,
metadata={"control_kind": control_kind, "ordinal": ordinal},
)
branch_line_start, branch_line_end = _line_span(
code_source,
body_start + opening,
body_start + closing,
)
graph.add_edge(
method["id"],
branch_id,
"DECLARES_RUNTIME_BRANCH",
rule_id="java.policy.branch",
basis="direct_syntax",
path=relative_path,
line_start=branch_line_start,
line_end=branch_line_end,
)
graph.add_edge(
read["leaf_id"],
branch_id,
"GUARDS_RUNTIME_BRANCH",
rule_id="java.policy.guard",
basis="resolved_static",
runtime_status="runtime_unknown",
path=relative_path,
line_start=branch_line_start,
line_end=branch_line_end,
limitations=("runtime.branch_execution_not_observed",),
)
def _strip_balanced_java_generics(value: str) -> str:
result: list[str] = []
depth = 0
for char in value:
if char == "<":
depth += 1
continue
if char == ">" and depth:
depth -= 1
continue
if depth == 0:
result.append(char)
return "".join(result)
def _clean_java_type(raw_type: str) -> str:
value = re.sub(JAVA_ANNOTATION_PATTERN + r"\s*", "", raw_type)
value = _strip_balanced_java_generics(value)
value = value.replace("[]", "").replace("...", "").strip()
value = re.sub(r"\s+", " ", value)
return value.split()[-1] if value else ""
def _resolve_java_type(
type_name: str,
imports: dict[str, str],
package_name: str,
wildcard_imports: set[str] | None = None,
) -> str:
clean = _clean_java_type(type_name)
if not clean:
return "unknown"
if "." in clean:
root, remainder = clean.split(".", 1)
if root in imports:
return f"{imports[root]}.{remainder}"
if root[:1].islower():
return clean
return f"{package_name}.{clean}" if package_name else clean
if clean in imports:
return imports[clean]
if clean in JAVA_LANG_TYPES:
return f"java.lang.{clean}"
if clean in JAVA_UTIL_TYPES:
return f"java.util.{clean}"
if wildcard_imports:
return clean
return f"{package_name}.{clean}" if package_name else clean
def _add_java_annotation_edges(
graph: Graph,
subject: str,
annotations: Iterable[tuple[str, str, str | None]],
*,
path: str | None = None,
line_start: int | None = None,
line_end: int | None = None,
) -> set[str]:
semantics: set[str] = set()
for annotation, qualified_name, semantic_name in annotations:
annotation_id = graph.add_annotation(
annotation,
qualified_name=qualified_name,
semantic_name=semantic_name,
)
graph.add_edge(
subject,
annotation_id,
"ANNOTATED_BY",
rule_id="java.annotation",
basis="direct_syntax",
path=path,
line_start=line_start,
line_end=line_end,
)
if semantic_name:
semantics.add(semantic_name)
if semantic_name in SPRING_STEREOTYPES:
graph.add_edge(
subject,
"framework:spring:bean",
"MANAGED_AS",
rule_id="java.spring.stereotype",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=path,
line_start=line_start,
line_end=line_end,
limitations=("spring.application_context_not_observed",),
)
graph.add_node(
"framework:spring:bean",
"FrameworkConcept",
"Spring-managed bean",
"Framework",
)
if semantic_name in SPRING_PROXY:
graph.add_edge(
subject,
"framework:spring:proxy",
"MAY_BE_PROXIED_BY",
rule_id="java.spring.proxy_annotation",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=path,
line_start=line_start,
line_end=line_end,
limitations=("spring.proxy_activation_not_observed",),
)
graph.add_node(
"framework:spring:proxy",
"FrameworkConcept",
"Spring proxy or interceptor",
"Framework",
)
return semantics
def _matching_java_brace(source: str, opening: int) -> int:
depth = 0
for index in range(opening, len(source)):
if source[index] == "{":
depth += 1
elif source[index] == "}":
depth -= 1
if depth == 0:
return index
return len(source)
def _java_type_header(source: str, start: int) -> tuple[str, int]:
parentheses = 0
brackets = 0
angles = 0
nested_braces = 0
index = start
while index < len(source):
char = source[index]
if char == "(":
parentheses += 1
elif char == ")" and parentheses:
parentheses -= 1
elif char == "[":
brackets += 1
elif char == "]" and brackets:
brackets -= 1
elif char == "<":
angles += 1
elif char == ">" and angles:
angles -= 1
elif char == "{":
if parentheses or brackets or angles or nested_braces:
nested_braces += 1
else:
return source[start:index], index
elif char == "}" and nested_braces:
nested_braces -= 1
elif char == ";" and not (parentheses or brackets or angles or nested_braces):
return source[start:index], -1
index += 1
return source[start:], -1
def _remove_leading_balanced(value: str, opening: str, closing: str) -> str:
stripped = value.lstrip()
if not stripped.startswith(opening):
return stripped
depth = 0
for index, char in enumerate(stripped):
if char == opening:
depth += 1
elif char == closing:
depth -= 1
if depth == 0:
return stripped[index + 1 :].lstrip()
return stripped
def _java_scope_for(scopes: list[dict[str, Any]], position: int) -> dict[str, Any] | None:
candidates = [
scope
for scope in scopes
if scope["body_start"] < position < scope["body_end"]
]
if not candidates:
return None
return max(candidates, key=lambda scope: scope["body_start"])
def _java_direct_member_matches(
source: str,
pattern: re.Pattern[str],
body_start: int,
body_end: int,
) -> list[re.Match[str]]:
"""Return pattern matches at the direct member depth of one Java type."""
matches: list[re.Match[str]] = []
depth = 0
cursor = body_start + 1
for match in pattern.finditer(source, cursor, body_end):
for character in source[cursor : match.start()]:
if character == "{":
depth += 1
elif character == "}" and depth:
depth -= 1
if depth == 0:
matches.append(match)
cursor = match.start()
return matches
def _split_java_parameters(parameters: str) -> list[str]:
values: list[str] = []
start = 0
depth = 0
for index, char in enumerate(parameters):
if char in "<([{":
depth += 1
elif char in ">)]}" and depth:
depth -= 1
elif char == "," and depth == 0:
values.append(parameters[start:index].strip())
start = index + 1
final = parameters[start:].strip()
if final:
values.append(final)
return values
def _java_hierarchy(header: str, kind: str) -> tuple[list[str], list[str]]:
header = _remove_leading_balanced(header, "<", ">")
if kind == "record":
header = _remove_leading_balanced(header, "(", ")")
extends_values: list[str] = []
implements_values: list[str] = []
extends_match = re.search(
r"\bextends\s+(?P<value>.*?)(?=\bimplements\b|\bpermits\b|$)",
header,
)
if extends_match:
values = _split_java_parameters(extends_match.group("value"))
extends_values.extend(values if kind == "interface" else values[:1])
implements_match = re.search(
r"\bimplements\s+(?P<value>.*?)(?=\bpermits\b|$)",
header,
)
if implements_match:
implements_values.extend(_split_java_parameters(implements_match.group("value")))
return extends_values, implements_values
def _java_parameter_types(
parameters: str,
imports: dict[str, str],
package_name: str,
wildcard_imports: set[str] | None = None,
) -> list[str]:
resolved: list[str] = []
for parameter in _split_java_parameters(parameters):
without_annotations = re.sub(
JAVA_ANNOTATION_PATTERN + r"\s*",
"",
parameter,
).strip()
without_modifiers = re.sub(r"^(?:final|volatile|transient)\s+", "", without_annotations)
pieces = without_modifiers.split()
if len(pieces) < 2:
continue
resolved.append(
_resolve_java_type(
" ".join(pieces[:-1]),
imports,
package_name,
wildcard_imports,
)
)
return resolved
def analyze_java(
graph: Graph,
repo: Path,
path: Path,
*,
repository_package_types: dict[str, set[str]] | None = None,
) -> None:
relative_path = path.relative_to(repo).as_posix()
original = _safe_read(path)
source = _strip_java_comments_and_literals(original)
package_match = re.search(r"^\s*package\s+([\w.]+)\s*;", source, re.MULTILINE)
package_name = package_match.group(1) if package_match else ""
_, local_declared_types = _java_package_and_declared_types(original)
same_package_types = (
set(repository_package_types.get(package_name, set())) | local_declared_types
if repository_package_types is not None
else None
)
import_matches = list(
re.finditer(
r"^\s*import\s+(?P<static>static\s+)?(?P<qualified>[\w.]+(?:\.\*)?)\s*;",
source,
re.MULTILINE,
)
)
imports = {
match.group("qualified").rsplit(".", 1)[-1]: match.group("qualified")
for match in import_matches
if not match.group("static") and not match.group("qualified").endswith(".*")
}
wildcard_imports = {
match.group("qualified")[:-2]
for match in import_matches
if not match.group("static") and match.group("qualified").endswith(".*")
}
module_name = package_name or relative_path.removesuffix(".java").replace("/", ".")
module_id = graph.add_node(
_node_id("java", "package", module_name),
"Package",
module_name.rsplit(".", 1)[-1],
"Java",
path=relative_path,
qualified_name=module_name,
metadata={"line_start": 1, "line_end": 1},
)
for imported in sorted(imports.values()):
imported_id = graph.add_external_type("Java", imported)
import_match = next(
(
match
for match in import_matches
if match.group("qualified") == imported
),
None,
)
import_lines = (
_line_span(source, import_match.start(), import_match.end())
if import_match is not None
else (1, 1)
)
graph.add_edge(
module_id,
imported_id,
"IMPORTS",
rule_id="java.import",
basis="direct_syntax",
path=relative_path,
line_start=import_lines[0],
line_end=import_lines[1],
)
raw_scopes: list[dict[str, Any]] = []
for match in JAVA_TYPE_RE.finditer(source):
header, body_start = _java_type_header(source, match.end())
if body_start < 0:
graph.add_warning(relative_path, f"Could not locate body for Java type {match.group('name')}")
continue
raw_scopes.append(
{
"match": match,
"header": header,
"start": match.start(),
"body_start": body_start,
"body_end": _matching_java_brace(source, body_start),
}
)
scopes: list[dict[str, Any]] = []
for raw_scope in sorted(raw_scopes, key=lambda item: item["start"]):
match = raw_scope["match"]
type_name = match.group("name")
parent = _java_scope_for(scopes, raw_scope["start"])
if parent:
qualified_name = f"{parent['qualified_name']}.{type_name}"
else:
qualified_name = f"{package_name}.{type_name}" if package_name else type_name
node_type = {
"class": "Class",
"interface": "Interface",
"enum": "Enum",
"record": "Record",
}[match.group("kind")]
type_line_start, type_line_end = _line_span(source, match.start(), match.end())
type_id = graph.add_node(
_node_id("java", match.group("kind"), qualified_name),
node_type,
type_name,
"Java",
path=relative_path,
qualified_name=qualified_name,
metadata={"line_start": type_line_start, "line_end": type_line_end},
)
graph.add_edge(
parent["id"] if parent else module_id,
type_id,
"DECLARES",
rule_id="java.type_declaration",
basis="direct_syntax",
path=relative_path,
line_start=type_line_start,
line_end=type_line_end,
)
type_semantics = _add_java_annotation_edges(
graph,
type_id,
_resolved_annotations(
match.group("annotations") or "",
imports,
wildcard_imports,
package_name,
same_package_types,
),
path=relative_path,
line_start=type_line_start,
line_end=type_line_end,
)
scope = {
**raw_scope,
"id": type_id,
"qualified_name": qualified_name,
"simple_name": type_name,
"annotation_semantics": type_semantics,
}
scopes.append(scope)
extends_values, implements_values = _java_hierarchy(
raw_scope["header"],
match.group("kind"),
)
for extends_value in extends_values:
target_name = _resolve_java_type(
extends_value, imports, package_name, wildcard_imports
)
graph.add_edge(
type_id,
graph.add_external_type("Java", target_name),
"EXTENDS",
rule_id="java.extends",
basis="resolved_static",
path=relative_path,
line_start=type_line_start,
line_end=type_line_end,
limitations=("java.dynamic_type_resolution_not_observed",),
)
for interface in implements_values:
target_name = _resolve_java_type(
interface, imports, package_name, wildcard_imports
)
graph.add_edge(
type_id,
graph.add_external_type("Java", target_name),
"IMPLEMENTS",
rule_id="java.implements",
basis="resolved_static",
path=relative_path,
line_start=type_line_start,
line_end=type_line_end,
)
if not scopes:
return
methods: list[dict[str, Any]] = []
declared_methods: list[dict[str, Any]] = []
for match in JAVA_METHOD_RE.finditer(source):
owner = _java_scope_for(scopes, match.start())
if owner is None:
continue
method_name = match.group("name")
if method_name == owner["simple_name"]:
continue
parameter_types = _java_parameter_types(
match.group("params"), imports, package_name, wildcard_imports
)
signature = ",".join(parameter_types)
method_qualified = f"{owner['qualified_name']}#{method_name}({signature})"
annotations = _resolved_annotations(
match.group("annotations") or "",
imports,
wildcard_imports,
package_name,
same_package_types,
)
method_line_start, method_line_end = _line_span(source, match.start(), match.end())
method_id = graph.add_node(
_node_id("java", "method", method_qualified),
"Method",
method_name,
"Java",
path=relative_path,
qualified_name=method_qualified,
metadata={
"return_type": _resolve_java_type(
match.group("return"), imports, package_name, wildcard_imports
),
"parameter_types": parameter_types,
"line_start": method_line_start,
"line_end": method_line_end,
},
)
graph.add_edge(
owner["id"],
method_id,
"DECLARES",
rule_id="java.method_declaration",
basis="direct_syntax",
path=relative_path,
line_start=method_line_start,
line_end=method_line_end,
)
declared_method = {
"id": method_id,
"owner_id": owner["id"],
"name": method_name,
"parameter_count": len(parameter_types),
"owner_body_start": owner["body_start"],
"owner_body_end": owner["body_end"],
"parameter_names": [
identifiers[-1]
for parameter in _split_java_parameters(match.group("params"))
if (identifiers := re.findall(r"[A-Za-z_$][\w$]*", parameter))
],
}
declared_methods.append(declared_method)
method_semantics = _add_java_annotation_edges(
graph,
method_id,
annotations,
path=relative_path,
line_start=method_line_start,
line_end=method_line_end,
)
if match.group("ending") == "{":
body_start = source.find("{", match.end() - 1, owner["body_end"])
if body_start >= 0:
methods.append(
{
**declared_method,
"qualified_name": method_qualified,
"body_start": body_start + 1,
"body_end": _matching_java_brace(source, body_start),
}
)
if "Bean" in method_semantics:
return_type = _resolve_java_type(
match.group("return"), imports, package_name, wildcard_imports
)
bean_id = graph.add_external_type("Java", return_type)
graph.add_node(
"framework:spring:bean",
"FrameworkConcept",
"Spring-managed bean",
"Framework",
)
graph.add_edge(
method_id,
bean_id,
"DECLARES_BEAN",
rule_id="java.spring.bean_method",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=relative_path,
line_start=method_line_start,
line_end=method_line_end,
limitations=("spring.bean_registration_not_observed",),
)
graph.add_edge(
bean_id,
"framework:spring:bean",
"MANAGED_AS",
rule_id="java.spring.bean_return_type",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=relative_path,
line_start=method_line_start,
line_end=method_line_end,
limitations=("spring.bean_registration_not_observed",),
)
for target_name in parameter_types:
if target_name.startswith("java.lang."):
continue
graph.add_edge(
method_id,
graph.add_external_type("Java", target_name),
"INJECTS",
rule_id="java.spring.bean_parameter",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=relative_path,
line_start=method_line_start,
line_end=method_line_end,
limitations=("spring.bean_resolution_not_observed",),
)
if method_semantics.intersection(SPRING_INJECTION):
for target_name in parameter_types:
if target_name.startswith("java.lang."):
continue
graph.add_edge(
owner["id"],
graph.add_external_type("Java", target_name),
"INJECTS",
rule_id="java.spring.method_injection",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=relative_path,
line_start=method_line_start,
line_end=method_line_end,
limitations=("spring.bean_resolution_not_observed",),
)
_add_java_call_edges(
graph,
methods,
declared_methods,
source,
original,
imports,
relative_path,
)
policy_source = _java_policy_scan_source(original)
for method in methods:
_add_java_policy_runtime_edges(
graph,
method,
policy_source,
source,
relative_path,
)
for match in JAVA_FIELD_RE.finditer(source):
owner = _java_scope_for(scopes, match.start())
if owner is None:
continue
annotations = _resolved_annotations(
match.group("annotations") or "",
imports,
wildcard_imports,
package_name,
same_package_types,
)
field_line_start, field_line_end = _line_span(source, match.start(), match.end())
annotation_semantics = _add_java_annotation_edges(
graph,
owner["id"],
annotations,
path=relative_path,
line_start=field_line_start,
line_end=field_line_end,
)
if not annotation_semantics.intersection(SPRING_INJECTION):
continue
target_name = _resolve_java_type(
match.group("type"), imports, package_name, wildcard_imports
)
target_id = graph.add_external_type("Java", target_name)
graph.add_edge(
owner["id"],
target_id,
"INJECTS",
rule_id="java.spring.field_injection",
basis="framework_semantic",
runtime_status="runtime_unknown",
path=relative_path,
line_start=field_line_start,
line_end=field_line_end,
limitations=("spring.bean_resolution_not_observed",),
)
for owner in scopes:
constructor_re = re.compile(
rf"(?:^[ \t]*|(?<=[{{}};])[ \t]*)"
rf"(?P<annotations>(?:{JAVA_ANNOTATION_PATTERN}\s*)*)"
rf"(?:(?:public|protected|private)\s+)?"
rf"(?:<[^;{{}}]+>\s+)?"
rf"{re.escape(owner['simple_name'])}\s*"
rf"\((?P<params>(?:[^()]|\([^()]*\))*)\)"
rf"\s*(?:throws\s+[^{{;]+)?\{{",
re.MULTILINE,
)
constructors = [
constructor
for constructor in _java_direct_member_matches(
source,
constructor_re,
owner["body_start"],
owner["body_end"],
)
if _java_scope_for(scopes, constructor.start()) is owner
]
for constructor in constructors:
constructor_line_start, constructor_line_end = _line_span(
source, constructor.start(), constructor.end()
)
constructor_annotations = _resolved_annotations(
constructor.group("annotations") or "",
imports,
wildcard_imports,
package_name,
same_package_types,
)
constructor_semantics = {
semantic
for _, _, semantic in constructor_annotations
if semantic is not None
}
is_managed_type = bool(
owner["annotation_semantics"].intersection(SPRING_STEREOTYPES)
)
is_explicit_injection = bool(
constructor_semantics.intersection(SPRING_INJECTION)
)
if not is_explicit_injection and not (is_managed_type and len(constructors) == 1):
continue
for target_name in _java_parameter_types(
constructor.group("params"),
imports,
package_name,
wildcard_imports,
):
if target_name.startswith("java.lang."):
continue
graph.add_edge(
owner["id"],
graph.add_external_type("Java", target_name),
"INJECTS",
rule_id=(
"java.spring.explicit_constructor_injection"
if is_explicit_injection
else "java.spring.single_constructor_injection"
),
basis="framework_semantic",
runtime_status="runtime_unknown",
path=relative_path,
line_start=constructor_line_start,
line_end=constructor_line_end,
limitations=("spring.bean_resolution_not_observed",),
)
def _python_name(node: ast.AST | None) -> str:
if node is None:
return ""
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
parent = _python_name(node.value)
return f"{parent}.{node.attr}" if parent else node.attr
if isinstance(node, ast.Call):
return _python_name(node.func)
if isinstance(node, ast.Subscript):
return _python_name(node.value)
return ""
def _pipeline_role(name: str) -> str | None:
tokenized = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", name)
tokens = {
token
for token in re.split(r"[^A-Za-z0-9]+", tokenized.casefold())
if token
}
groups = {
"Extract": {
"extract", "extractor", "fetch", "fetcher", "read", "reader",
"ingest", "ingestion", "collect", "collector", "source",
},
"Transform": {
"transform", "transformer", "clean", "cleaner", "normalize",
"normalizer", "enrich", "enricher", "map", "mapper", "parse", "parser",
},
"Load": {
"load", "loader", "write", "writer", "persist", "persistence",
"publish", "publisher", "sink", "store", "storage",
},
"Validate": {
"validate", "validator", "verify", "verifier", "quality", "check", "checker",
},
"Orchestrate": {
"pipeline", "workflow", "orchestrate", "orchestrator", "schedule",
"scheduler", "run", "runner",
},
}
for role, role_tokens in groups.items():
if tokens.intersection(role_tokens):
return role
return None
def _python_local_bindings(node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
names = {
argument.arg
for argument in (
[*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs]
+ ([node.args.vararg] if node.args.vararg else [])
+ ([node.args.kwarg] if node.args.kwarg else [])
)
}
global_names: set[str] = set()
nonlocal_names: set[str] = set()
class Collector(ast.NodeVisitor):
def visit_Name(self, child: ast.Name) -> None:
if isinstance(child.ctx, ast.Store):
names.add(child.id)
def visit_Import(self, child: ast.Import) -> None:
for alias in child.names:
names.add(alias.asname or alias.name.split(".", 1)[0])
def visit_ImportFrom(self, child: ast.ImportFrom) -> None:
for alias in child.names:
if alias.name != "*":
names.add(alias.asname or alias.name)
def visit_FunctionDef(self, child: ast.FunctionDef) -> None:
names.add(child.name)
def visit_AsyncFunctionDef(self, child: ast.AsyncFunctionDef) -> None:
names.add(child.name)
def visit_ClassDef(self, child: ast.ClassDef) -> None:
names.add(child.name)
def visit_ExceptHandler(self, child: ast.ExceptHandler) -> None:
if child.name:
names.add(child.name)
self.generic_visit(child)
def _visit_comprehension(
self,
generators: list[ast.comprehension],
values: list[ast.AST],
) -> None:
# Python 3 comprehension targets have their own implicit scope and
# must not shadow a binding in the containing function.
for generator in generators:
self.visit(generator.iter)
for condition in generator.ifs:
self.visit(condition)
for value in values:
self.visit(value)
def visit_ListComp(self, child: ast.ListComp) -> None:
self._visit_comprehension(child.generators, [child.elt])
def visit_SetComp(self, child: ast.SetComp) -> None:
self._visit_comprehension(child.generators, [child.elt])
def visit_GeneratorExp(self, child: ast.GeneratorExp) -> None:
self._visit_comprehension(child.generators, [child.elt])
def visit_DictComp(self, child: ast.DictComp) -> None:
self._visit_comprehension(child.generators, [child.key, child.value])
def visit_Lambda(self, child: ast.Lambda) -> None:
return
def visit_Global(self, child: ast.Global) -> None:
global_names.update(child.names)
def visit_Nonlocal(self, child: ast.Nonlocal) -> None:
nonlocal_names.update(child.names)
collector = Collector()
for statement in node.body:
collector.visit(statement)
return names - global_names - nonlocal_names
def _python_import_from_module(package_name: str, node: ast.ImportFrom) -> str:
if node.level <= 0:
return node.module or ""
parts = [part for part in package_name.split(".") if part]
ascents = max(0, node.level - 1)
if ascents > len(parts):
return node.module or ""
if ascents:
parts = parts[:-ascents]
if node.module:
parts.extend(node.module.split("."))
return ".".join(parts)
def _python_global_bindings(node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
names: set[str] = set()
class Collector(ast.NodeVisitor):
def visit_Global(self, child: ast.Global) -> None:
names.update(child.names)
def visit_FunctionDef(self, child: ast.FunctionDef) -> None:
return
def visit_AsyncFunctionDef(self, child: ast.AsyncFunctionDef) -> None:
return
def visit_ClassDef(self, child: ast.ClassDef) -> None:
return
def visit_Lambda(self, child: ast.Lambda) -> None:
return
collector = Collector()
for statement in node.body:
collector.visit(statement)
return names
def _python_import_aliases(
node: ast.Module | ast.FunctionDef | ast.AsyncFunctionDef,
package_name: str,
) -> dict[str, str]:
"""Collect unambiguous imports for one lexical Python scope."""
aliases: dict[str, str] = {}
competing_bindings: set[str] = set()
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
competing_bindings.update(
argument.arg
for argument in (
[*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs]
+ ([node.args.vararg] if node.args.vararg else [])
+ ([node.args.kwarg] if node.args.kwarg else [])
)
)
class Collector(ast.NodeVisitor):
def _record_import(self, local_name: str, target: str) -> None:
previous = aliases.get(local_name)
if previous is not None and previous != target:
competing_bindings.add(local_name)
aliases[local_name] = target
def visit_Name(self, child: ast.Name) -> None:
if isinstance(child.ctx, ast.Store):
competing_bindings.add(child.id)
def visit_Import(self, child: ast.Import) -> None:
for alias in child.names:
local_name = alias.asname or alias.name.split(".", 1)[0]
target = alias.name if alias.asname else local_name
self._record_import(local_name, target)
def visit_ImportFrom(self, child: ast.ImportFrom) -> None:
target_module = _python_import_from_module(package_name, child)
for alias in child.names:
if alias.name == "*":
continue
local_name = alias.asname or alias.name
qualified = f"{target_module}.{alias.name}" if target_module else alias.name
self._record_import(local_name, qualified)
def visit_FunctionDef(self, child: ast.FunctionDef) -> None:
competing_bindings.add(child.name)
def visit_AsyncFunctionDef(self, child: ast.AsyncFunctionDef) -> None:
competing_bindings.add(child.name)
def visit_ClassDef(self, child: ast.ClassDef) -> None:
competing_bindings.add(child.name)
def visit_Lambda(self, child: ast.Lambda) -> None:
return
def visit_ExceptHandler(self, child: ast.ExceptHandler) -> None:
if child.name:
competing_bindings.add(child.name)
self.generic_visit(child)
def _visit_comprehension(
self,
generators: list[ast.comprehension],
values: list[ast.AST],
) -> None:
for generator in generators:
self.visit(generator.iter)
for condition in generator.ifs:
self.visit(condition)
for value in values:
self.visit(value)
def visit_ListComp(self, child: ast.ListComp) -> None:
self._visit_comprehension(child.generators, [child.elt])
def visit_SetComp(self, child: ast.SetComp) -> None:
self._visit_comprehension(child.generators, [child.elt])
def visit_GeneratorExp(self, child: ast.GeneratorExp) -> None:
self._visit_comprehension(child.generators, [child.elt])
def visit_DictComp(self, child: ast.DictComp) -> None:
self._visit_comprehension(child.generators, [child.key, child.value])
collector = Collector()
for statement in node.body:
collector.visit(statement)
return {
name: target
for name, target in aliases.items()
if name not in competing_bindings
}
def _python_target_bindings(target: ast.AST) -> set[str]:
return {
child.id
for child in ast.walk(target)
if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Store)
}
def _validate_python_ast_limits(tree: ast.AST) -> None:
"""Bound recursive visitor work using an iterative AST walk."""
stack: list[tuple[ast.AST, int]] = [(tree, 1)]
discovered = 1
while stack:
node, depth = stack.pop()
if depth > MAX_PYTHON_AST_DEPTH:
raise OntologyError(
f"Python AST exceeds the {MAX_PYTHON_AST_DEPTH}-depth safety limit."
)
child_depth = depth + 1
for child in ast.iter_child_nodes(node):
discovered += 1
if discovered > MAX_PYTHON_AST_NODES:
raise OntologyError(
f"Python AST exceeds the {MAX_PYTHON_AST_NODES}-node safety limit."
)
if child_depth > MAX_PYTHON_AST_DEPTH:
raise OntologyError(
f"Python AST exceeds the {MAX_PYTHON_AST_DEPTH}-depth safety limit."
)
stack.append((child, child_depth))
class PythonVisitor(ast.NodeVisitor):
def __init__(
self,
graph: Graph,
module_id: str,
module_name: str,
relative_path: str,
tree: ast.Module,
*,
is_package: bool,
) -> None:
self.graph = graph
self.module_id = module_id
self.module_name = module_name
self.relative_path = relative_path
self.scope: list[tuple[str, str]] = [(module_id, module_name)]
self.class_scope: list[str] = []
self.local_bindings: list[set[str]] = []
self.local_symbols: list[dict[str, str]] = []
self.local_import_aliases: list[dict[str, str]] = []
self.local_globals: list[set[str]] = []
self.module_functions = {
child.name
for child in tree.body
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef))
}
self.module_classes = {
child.name for child in tree.body if isinstance(child, ast.ClassDef)
}
self.class_methods = {
f"{module_name}.{child.name}": {
member.name
for member in child.body
if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef))
}
for child in tree.body
if isinstance(child, ast.ClassDef)
}
self.package_name = (
module_name
if is_package
else module_name.rsplit(".", 1)[0] if "." in module_name else ""
)
self.import_aliases = _python_import_aliases(tree, self.package_name)
@property
def owner(self) -> tuple[str, str]:
return self.scope[-1]
def _import_from_module(self, node: ast.ImportFrom) -> str:
return _python_import_from_module(self.package_name, node)
def _resolve_reference(self, name: str) -> str:
if not name:
return name
pieces = name.split(".")
if self.class_scope and pieces[0] in {"self", "cls"} and len(pieces) > 1:
return ".".join([self.class_scope[-1], *pieces[1:]])
for index in range(len(self.local_bindings) - 1, -1, -1):
if pieces[0] in self.local_globals[index]:
break
local_symbol = self.local_symbols[index].get(pieces[0])
if local_symbol:
return ".".join([local_symbol, *pieces[1:]])
local_import = self.local_import_aliases[index].get(pieces[0])
if local_import:
return ".".join([local_import, *pieces[1:]])
if pieces[0] in self.local_bindings[index]:
return name
imported = self.import_aliases.get(pieces[0])
if imported:
return ".".join([imported, *pieces[1:]]) if len(pieces) > 1 else imported
if len(pieces) == 1:
if name in self.module_functions or name in self.module_classes:
return f"{self.module_name}.{name}"
return name
def _add_decorators(self, subject_id: str, decorators: list[ast.expr]) -> None:
for decorator in decorators:
name = self._resolve_reference(_python_name(decorator))
if not name:
continue
decorator_id = self.graph.add_node(
_node_id("python", "decorator", name),
"Decorator",
name.rsplit(".", 1)[-1],
"Python",
qualified_name=name,
)
self.graph.add_edge(
subject_id,
decorator_id,
"DECORATED_BY",
rule_id="python.decorator",
basis="direct_syntax",
path=self.relative_path,
line_start=getattr(decorator, "lineno", None),
line_end=getattr(decorator, "end_lineno", None),
)
def _add_pipeline_role(
self,
subject_id: str,
name: str,
node: ast.AST | None = None,
) -> None:
role = _pipeline_role(name)
if role:
role_id = self.graph.add_node(
f"pipeline:role:{role.lower()}",
"PipelineRole",
role,
"Concept",
)
self.graph.add_edge(
subject_id,
role_id,
"HAS_PIPELINE_ROLE",
rule_id="python.pipeline_role.name_tokens",
basis="name_heuristic",
path=self.relative_path,
line_start=getattr(node, "lineno", None),
line_end=getattr(node, "end_lineno", None),
limitations=("python.role_name_heuristic",),
)
def visit_Import(self, node: ast.Import) -> None:
for alias in node.names:
target_id = self.graph.add_node(
_node_id("python", "module", alias.name),
"ExternalModule",
alias.name.rsplit(".", 1)[-1],
"Python",
qualified_name=alias.name,
)
self.graph.add_edge(
self.module_id,
target_id,
"IMPORTS",
rule_id="python.import",
basis="direct_syntax",
path=self.relative_path,
line_start=getattr(node, "lineno", None),
line_end=getattr(node, "end_lineno", None),
)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
target_module = self._import_from_module(node)
if target_module:
target_id = self.graph.add_node(
_node_id("python", "module", target_module),
"ExternalModule",
target_module.rsplit(".", 1)[-1],
"Python",
qualified_name=target_module,
)
self.graph.add_edge(
self.module_id,
target_id,
"IMPORTS",
rule_id="python.import_from",
basis="resolved_static",
path=self.relative_path,
line_start=getattr(node, "lineno", None),
line_end=getattr(node, "end_lineno", None),
limitations=("python.import_execution_not_observed",),
)
def visit_ClassDef(self, node: ast.ClassDef) -> None:
owner_id, owner_name = self.owner
qualified = f"{owner_name}.{node.name}"
class_id = self.graph.add_node(
_node_id("python", "class", qualified),
"Class",
node.name,
"Python",
path=self.relative_path,
qualified_name=qualified,
metadata={
"line_start": node.lineno,
"line_end": getattr(node, "end_lineno", node.lineno),
},
)
self.graph.add_edge(
owner_id,
class_id,
"DECLARES",
rule_id="python.class_declaration",
basis="direct_syntax",
path=self.relative_path,
line_start=node.lineno,
line_end=getattr(node, "end_lineno", node.lineno),
)
self._add_decorators(class_id, node.decorator_list)
self._add_pipeline_role(class_id, node.name, node)
for base in node.bases:
base_name = self._resolve_reference(_python_name(base))
if base_name:
self.graph.add_edge(
class_id,
self.graph.add_external_type("Python", base_name),
"EXTENDS",
rule_id="python.inheritance",
basis="resolved_static",
path=self.relative_path,
line_start=getattr(base, "lineno", node.lineno),
line_end=getattr(base, "end_lineno", node.lineno),
limitations=("python.dynamic_base_resolution_not_observed",),
)
self.scope.append((class_id, qualified))
self.class_scope.append(qualified)
for child in node.body:
self.visit(child)
self.class_scope.pop()
self.scope.pop()
def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
owner_id, owner_name = self.owner
qualified = f"{owner_name}.{node.name}"
node_type = "AsyncFunction" if isinstance(node, ast.AsyncFunctionDef) else "Function"
if self.graph.nodes.get(owner_id, {}).get("type") == "Class":
node_type = "AsyncMethod" if isinstance(node, ast.AsyncFunctionDef) else "Method"
function_id = self.graph.add_node(
_node_id("python", node_type, qualified),
node_type,
node.name,
"Python",
path=self.relative_path,
qualified_name=qualified,
metadata={
"parameter_count": (
len(node.args.posonlyargs)
+ len(node.args.args)
+ len(node.args.kwonlyargs)
+ int(node.args.vararg is not None)
+ int(node.args.kwarg is not None)
),
"line_start": node.lineno,
"line_end": getattr(node, "end_lineno", node.lineno),
},
)
self.graph.add_edge(
owner_id,
function_id,
"DECLARES",
rule_id="python.function_declaration",
basis="direct_syntax",
path=self.relative_path,
line_start=node.lineno,
line_end=getattr(node, "end_lineno", node.lineno),
)
self._add_decorators(function_id, node.decorator_list)
self._add_pipeline_role(function_id, node.name, node)
self.scope.append((function_id, qualified))
self.local_bindings.append(_python_local_bindings(node))
self.local_globals.append(_python_global_bindings(node))
self.local_import_aliases.append(
_python_import_aliases(node, self.package_name)
)
self.local_symbols.append(
{
child.name: f"{qualified}.{child.name}"
for child in node.body
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
}
)
for child in node.body:
self.visit(child)
self.local_symbols.pop()
self.local_import_aliases.pop()
self.local_globals.pop()
self.local_bindings.pop()
self.scope.pop()
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
self._visit_function(node)
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
self._visit_function(node)
def _visit_comprehension(
self,
generators: list[ast.comprehension],
values: list[ast.AST],
) -> None:
pushed = 0
try:
for generator in generators:
# Each iterable is evaluated before its own target is bound;
# targets from earlier generators remain visible.
self.visit(generator.iter)
self.local_bindings.append(_python_target_bindings(generator.target))
self.local_symbols.append({})
self.local_import_aliases.append({})
self.local_globals.append(set())
pushed += 1
for condition in generator.ifs:
self.visit(condition)
for value in values:
self.visit(value)
finally:
for _ in range(pushed):
self.local_import_aliases.pop()
self.local_symbols.pop()
self.local_globals.pop()
self.local_bindings.pop()
def visit_ListComp(self, node: ast.ListComp) -> None:
self._visit_comprehension(node.generators, [node.elt])
def visit_SetComp(self, node: ast.SetComp) -> None:
self._visit_comprehension(node.generators, [node.elt])
def visit_GeneratorExp(self, node: ast.GeneratorExp) -> None:
self._visit_comprehension(node.generators, [node.elt])
def visit_DictComp(self, node: ast.DictComp) -> None:
self._visit_comprehension(node.generators, [node.key, node.value])
def visit_Lambda(self, node: ast.Lambda) -> None:
for default in [*node.args.defaults, *node.args.kw_defaults]:
if default is not None:
self.visit(default)
bindings = {
argument.arg
for argument in (
[*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs]
+ ([node.args.vararg] if node.args.vararg else [])
+ ([node.args.kwarg] if node.args.kwarg else [])
)
}
self.local_bindings.append(bindings)
self.local_symbols.append({})
self.local_import_aliases.append({})
self.local_globals.append(set())
try:
self.visit(node.body)
finally:
self.local_globals.pop()
self.local_import_aliases.pop()
self.local_symbols.pop()
self.local_bindings.pop()
def visit_Call(self, node: ast.Call) -> None:
lexical_name = _python_name(node.func)
call_name = self._resolve_reference(lexical_name)
if call_name:
call_id = self.graph.add_node(
_node_id("python", "callable", call_name),
"ExternalCallable",
call_name.rsplit(".", 1)[-1],
"Python",
qualified_name=call_name,
)
self.graph.add_edge(
self.owner[0],
call_id,
"CALLS",
rule_id="python.call.lexical_resolution" if call_name != lexical_name else "python.call.unresolved_reference",
basis="resolved_static" if call_name != lexical_name else "direct_syntax",
path=self.relative_path,
line_start=getattr(node, "lineno", None),
line_end=getattr(node, "end_lineno", None),
limitations=("python.runtime_dispatch_not_observed",) if call_name != lexical_name else (
"python.runtime_dispatch_not_observed", "python.call_target_not_resolved",
),
)
self.generic_visit(node)
def analyze_python(graph: Graph, repo: Path, path: Path) -> None:
relative_path = path.relative_to(repo).as_posix()
module_relative = relative_path
src_is_package = False
try:
src_init_stat = (repo / "src" / "__init__.py").lstat()
src_is_package = not _is_link_like(src_init_stat) and stat.S_ISREG(
src_init_stat.st_mode
)
except OSError:
pass
if module_relative.startswith("src/") and not src_is_package:
module_relative = module_relative[len("src/") :]
module_name = module_relative.removesuffix(".py").replace("/", ".")
if module_name.endswith(".__init__"):
module_name = module_name[: -len(".__init__")]
try:
tree = ast.parse(_safe_read(path), filename=relative_path)
except SyntaxError as exc:
graph.add_warning(relative_path, f"Python syntax could not be parsed at line {exc.lineno or '?'}")
return
except RecursionError as exc:
raise OntologyError("Python syntax exceeds the parser nesting safety limit.") from exc
_validate_python_ast_limits(tree)
module_id = graph.add_node(
_node_id("python", "module", module_name),
"Module",
module_name.rsplit(".", 1)[-1],
"Python",
path=relative_path,
qualified_name=module_name,
metadata={"line_start": 1, "line_end": 1},
)
try:
PythonVisitor(
graph,
module_id,
module_name,
relative_path,
tree,
is_package=Path(relative_path).name == "__init__.py",
).visit(tree)
except RecursionError as exc:
raise OntologyError("Python AST traversal exceeded the nesting safety limit.") from exc
def preflight_document(repo: Path) -> dict[str, Any]:
sources, skipped = discover_sources(repo)
by_language = Counter(SUPPORTED_SUFFIXES[path.suffix.lower()] for path in sources)
return {
"status": "ready" if sources else "no_supported_sources",
"repository_name": repo.name,
"supported_languages": dict(sorted(by_language.items())),
"adapter_coverage": {
language: _adapter_quality(language) for language in sorted(by_language)
},
"source_file_count": len(sources),
"skipped": dict(sorted(skipped.items())),
"limits": {
"max_source_bytes": MAX_SOURCE_BYTES,
"max_source_files": MAX_SOURCE_FILES,
"max_total_source_bytes": MAX_TOTAL_SOURCE_BYTES,
"max_graph_nodes": MAX_GRAPH_NODES,
"max_graph_edges": MAX_GRAPH_EDGES,
"max_impact_results": MAX_IMPACT_RESULTS,
"max_python_ast_nodes": MAX_PYTHON_AST_NODES,
"max_python_ast_depth": MAX_PYTHON_AST_DEPTH,
"follows_symlinks": False,
"executes_source": False,
"network_access": False,
"writes_during_preflight": False,
},
"next_step": (
"Review this summary and obtain the repository owner's confirmation before indexing."
if sources
else "Point --repo to a Java or Python source repository."
),
}
def build_document(repo: Path) -> dict[str, Any]:
sources, skipped = discover_sources(repo)
graph = Graph(repo.name)
source_counts: Counter[str] = Counter()
repository_package_types: dict[str, set[str]] | None = {}
for path in sources:
if SUPPORTED_SUFFIXES[path.suffix.lower()] != "Java":
continue
try:
package_name, declared_types = _java_package_and_declared_types(
_safe_read(path)
)
except (OntologyError, UnicodeError):
# An incomplete name index cannot prove that a Spring wildcard is
# unshadowed, so disable wildcard semantics for this build.
repository_package_types = None
break
repository_package_types.setdefault(package_name, set()).update(declared_types)
for path in sources:
language = SUPPORTED_SUFFIXES[path.suffix.lower()]
source_counts[language] += 1
try:
if language == "Java":
analyze_java(
graph,
repo,
path,
repository_package_types=repository_package_types,
)
elif language == "Python":
analyze_python(graph, repo, path)
except (OntologyError, UnicodeError) as exc:
graph.add_warning(path.relative_to(repo).as_posix(), str(exc))
return graph.document(source_counts, skipped)
def _turtle_literal(value: Any) -> str:
rendered = (
str(value)
.replace("\\", "\\\\")
.replace('"', '\\"')
.replace("\r", "\\r")
.replace("\n", "\\n")
.replace("\t", "\\t")
)
return f'"{rendered}"'
def _turtle_uri(node_id: str) -> str:
return f"<urn:code-ontology:node:{quote(node_id, safe='')}>"
def render_turtle(document: dict[str, Any]) -> str:
lines = [
"@prefix co: <https://battle-doll.github.io/code-ontology-explorer/schema#> .",
"@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .",
"@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .",
"@prefix prov: <http://www.w3.org/ns/prov#> .",
"@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .",
"",
]
for node in document["nodes"]:
subject = _turtle_uri(node["id"])
lines.append(f"{subject} a co:{node['type']} ;")
predicates = [
("rdfs:label", _turtle_literal(node["name"])),
("co:language", _turtle_literal(node["language"])),
]
if node.get("qualified_name"):
predicates.append(("co:qualifiedName", _turtle_literal(node["qualified_name"])))
if node.get("path"):
predicates.append(("co:relativePath", _turtle_literal(node["path"])))
for index, (predicate, value) in enumerate(predicates):
ending = " ." if index == len(predicates) - 1 else " ;"
lines.append(f" {predicate} {value}{ending}")
lines.append("")
for edge in document["edges"]:
predicate = re.sub(r"[^A-Za-z0-9]", "", edge["type"].title())
lines.append(
f"{_turtle_uri(edge['source'])} co:{predicate} {_turtle_uri(edge['target'])} ."
)
for evidence in edge.get("evidence", []):
canonical = json.dumps(
{
"source": edge["source"],
"target": edge["target"],
"type": edge["type"],
"evidence": evidence,
},
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
evidence_uri = (
"<urn:code-ontology:relationship-evidence:"
f"{hashlib.sha256(canonical).hexdigest()}>"
)
evidence_predicates = [
("rdf:subject", _turtle_uri(edge["source"])),
("rdf:predicate", f"co:{predicate}"),
("rdf:object", _turtle_uri(edge["target"])),
("co:relationshipType", _turtle_literal(edge["type"])),
("co:ruleId", _turtle_literal(evidence["rule_id"])),
("co:evidenceBasis", _turtle_literal(evidence["basis"])),
("co:runtimeStatus", _turtle_literal(evidence["runtime_status"])),
]
if evidence.get("path"):
evidence_predicates.append(
("co:evidencePath", _turtle_literal(evidence["path"]))
)
if isinstance(evidence.get("line_start"), int):
evidence_predicates.extend(
[
("co:lineStart", f'"{evidence["line_start"]}"^^xsd:integer'),
("co:lineEnd", f'"{evidence["line_end"]}"^^xsd:integer'),
]
)
evidence_predicates.extend(
("co:limitationId", _turtle_literal(value))
for value in evidence.get("limitations", [])
)
lines.append(f"{evidence_uri} a co:RelationshipEvidence, rdf:Statement ;")
for index, (quality_predicate, value) in enumerate(evidence_predicates):
ending = " ." if index == len(evidence_predicates) - 1 else " ;"
lines.append(f" {quality_predicate} {value}{ending}")
lines.append("")
quality = document.get("quality", {})
relationship_quality = quality.get("relationship_evidence", {})
if isinstance(quality, dict) and isinstance(relationship_quality, dict):
quality_scope = json.dumps(
{
"generator": document.get("generator", {}),
"edges": [
(edge.get("source"), edge.get("target"), edge.get("type"))
for edge in document.get("edges", [])
if isinstance(edge, dict)
],
"quality": quality,
},
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
quality_digest = hashlib.sha256(quality_scope).hexdigest()
quality_uri = f"<urn:code-ontology:quality:{quality_digest}>"
lines.extend(
[
f"{quality_uri} a co:OntologyQuality ;",
f" co:contractVersion {_turtle_literal(quality.get('contract_version', 'unknown'))} ;",
" co:documentedEdges "
f'"{int(relationship_quality.get("documented_edges", 0))}"^^xsd:integer ;',
" co:missingEvidence "
f'"{int(relationship_quality.get("missing_evidence", 0))}"^^xsd:integer ;',
" co:coveragePercent "
f'"{float(relationship_quality.get("coverage_percent", 0.0))}"^^xsd:decimal .',
"",
]
)
for language, adapter in sorted(quality.get("adapters", {}).items()):
adapter_uri = (
f"<urn:code-ontology:adapter-quality:{quality_digest}:"
f"{quote(language, safe='')}>"
)
adapter_predicates = [
("co:language", _turtle_literal(language)),
("co:supportStatus", _turtle_literal(adapter.get("status", "unknown"))),
(
"co:detected",
f'"{str(adapter.get("detected") is True).lower()}"^^xsd:boolean',
),
]
adapter_predicates.extend(
(
"co:capability",
_turtle_literal(f"{name}={status}"),
)
for name, status in sorted(adapter.get("capabilities", {}).items())
)
adapter_predicates.extend(
("co:unsupportedRuntime", _turtle_literal(value))
for value in adapter.get("unsupported_runtime", [])
)
lines.append(f"{adapter_uri} a co:AdapterQuality ;")
for index, (quality_predicate, value) in enumerate(adapter_predicates):
ending = " ." if index == len(adapter_predicates) - 1 else " ;"
lines.append(f" {quality_predicate} {value}{ending}")
lines.append(f"{quality_uri} co:hasAdapterQuality {adapter_uri} .")
lines.append("")
lines.append("")
return "\n".join(lines)
def render_report(document: dict[str, Any]) -> str:
stats = document["statistics"]
quality = document.get("quality", {})
relationship_quality = quality.get("relationship_evidence", {})
lines = [
"# Code Ontology Report",
"",
f"- Repository label: `{document['repository']['name']}`",
f"- Source files analyzed: {sum(stats['source_files'].values())}",
f"- Nodes: {stats['nodes']}",
f"- Edges: {stats['edges']}",
f"- Parse warnings: {stats['warnings']}",
"- Relationship evidence coverage: "
f"{relationship_quality.get('coverage_percent', 'unknown')}% "
f"({relationship_quality.get('documented_edges', 0)}/"
f"{relationship_quality.get('total_edges', 0)})",
"- Relationship source-span coverage: "
f"{relationship_quality.get('source_span_coverage_percent', 'unknown')}%",
"",
"## Privacy boundary",
"",
"The ontology contains identifiers, relationships, language labels, and repository-relative paths.",
"It does not contain source bodies, comments, absolute paths, file hashes, credentials, or model prompts.",
"",
"## Node types",
"",
]
lines.extend(f"- {key}: {value}" for key, value in stats["node_types"].items())
lines.extend(["", "## Relationship types", ""])
lines.extend(f"- {key}: {value}" for key, value in stats["edge_types"].items())
lines.extend(["", "## Ontology quality", ""])
lines.append(
"Relationship evidence uses qualitative extraction bases, stable rule IDs, "
"and bounded relative source spans. It is not a probability score."
)
for basis, count in relationship_quality.get("basis_counts", {}).items():
lines.append(f"- Evidence basis `{basis}`: {count}")
for language, adapter in sorted(quality.get("adapters", {}).items()):
lines.append(
f"- {language} adapter: `{adapter.get('status', 'unknown')}` "
f"(detected: `{'yes' if adapter.get('detected') is True else 'no'}`)"
)
for capability, status in sorted(adapter.get("capabilities", {}).items()):
lines.append(f" - `{capability}`: `{status}`")
unsupported = ", ".join(adapter.get("unsupported_runtime", []))
if unsupported:
lines.append(f" - Unsupported runtime evidence: {unsupported}")
lines.extend(
[
"",
"## Interpretation note",
"",
"Static relationships are evidence for exploration, not proof of runtime behavior.",
"Reflection, generated code, framework configuration, and dynamic dispatch may be incomplete.",
"",
]
)
return "\n".join(lines)
def _write_text(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(
prefix=f".{path.name}.",
suffix=".tmp",
dir=path.parent,
)
temporary_path = Path(temporary_name)
try:
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary_path, path)
except Exception:
try:
temporary_path.unlink(missing_ok=True)
finally:
raise
def write_index(
repo: Path,
output: Path,
authorized: bool,
overwrite: bool = False,
) -> dict[str, Any]:
if not authorized:
raise OntologyError(
"Indexing requires --authorized after the repository owner has confirmed the scan."
)
raw_output = output.expanduser()
if raw_output.is_symlink():
raise OntologyError("The output directory may not be a symbolic link.")
output = raw_output.resolve()
if output == repo or _is_relative_to(output, repo) or _is_relative_to(repo, output):
raise OntologyError(
"The output directory must be separate from the target repository and its parent directories."
)
artifact_paths = [output / name for name in ("ontology.json", "ontology.ttl", "report.md")]
symlinks = [path.name for path in artifact_paths if path.is_symlink()]
if symlinks:
raise OntologyError("Refusing to write through symbolic-link artifacts: " + ", ".join(symlinks))
existing = [path.name for path in artifact_paths if path.exists()]
if existing and not overwrite:
raise OntologyError(
"Refusing to replace existing artifacts without --overwrite: " + ", ".join(existing)
)
document = build_document(repo)
output.mkdir(parents=True, exist_ok=True)
_write_text(output / "ontology.json", json.dumps(document, indent=2, ensure_ascii=False) + "\n")
_write_text(output / "ontology.ttl", render_turtle(document))
_write_text(output / "report.md", render_report(document))
return {
"status": "indexed",
"output_directory": str(output),
"artifacts": ["ontology.json", "ontology.ttl", "report.md"],
"statistics": document["statistics"],
}
def load_document(index_path: str) -> tuple[Path, dict[str, Any]]:
path = Path(index_path).expanduser().resolve()
if not path.is_file():
raise OntologyError(f"Ontology index does not exist: {path}")
try:
document = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise OntologyError(f"Could not read ontology JSON: {exc}") from exc
if document.get("schema_version") != SCHEMA_VERSION:
raise OntologyError(
f"Unsupported schema version: {document.get('schema_version', 'missing')}"
)
if not isinstance(document.get("nodes"), list) or not isinstance(document.get("edges"), list):
raise OntologyError("Ontology JSON is missing nodes or edges.")
return path, document
def _query_rank(node: dict[str, Any], needle: str) -> int:
for rank, field in enumerate(("id", "qualified_name", "name")):
if str(node.get(field, "")).casefold() == needle:
return rank
return 3
class SnapshotIndex:
"""Reusable indexes for one immutable snapshot; callers own its lifetime."""
def __init__(self, document: dict[str, Any]) -> None:
self.nodes = {node["id"]: node for node in document["nodes"]}
self.searchable = [
(node, " ".join(str(node.get(key, "")) for key in
("id", "type", "name", "language", "path", "qualified_name", "metadata")).casefold())
for node in document["nodes"]
]
self.adjacency: dict[str, list[tuple[str, dict[str, Any], str]]] = {}
for edge in document["edges"]:
self.adjacency.setdefault(edge["source"], []).append((edge["target"], edge, "outgoing"))
self.adjacency.setdefault(edge["target"], []).append((edge["source"], edge, "incoming"))
def query_document(
document: dict[str, Any], term: str, limit: int, *, offset: int = 0,
language: str | None = None, node_type: str | None = None,
path_prefix: str | None = None,
index: SnapshotIndex | None = None,
) -> dict[str, Any]:
if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= MAX_QUERY_RESULTS:
raise OntologyError("Query limit must be from 1 to 200.")
if isinstance(offset, bool) or not isinstance(offset, int) or not 0 <= offset <= MAX_GRAPH_NODES:
raise OntologyError("Query offset is out of range.")
if not isinstance(term, str) or not term.strip() or len(term) > 1000:
raise OntologyError("Query term must contain 1 to 1000 characters.")
if path_prefix is not None and not _portable_relative_path(path_prefix):
raise OntologyError("Query path prefix must be repository-relative.")
needle = term.casefold()
matches = []
indexed = index or SnapshotIndex(document)
for node, searchable in indexed.searchable:
if language is not None and str(node.get("language", "")).casefold() != language.casefold():
continue
if node_type is not None and str(node.get("type", "")).casefold() != node_type.casefold():
continue
if path_prefix is not None:
path = str(node.get("path", ""))
prefix = path_prefix.rstrip("/")
if path != prefix and not path.startswith(prefix + "/"):
continue
if needle in searchable:
matches.append(node)
matches.sort(key=lambda node: (_query_rank(node, needle), node["name"].casefold(), node["id"]))
page = matches[offset:offset + limit]
next_offset = offset + len(page)
return {
"term": term,
"match_count": len(matches),
"returned": len(page),
"matches": page,
"offset": offset,
"next_offset": next_offset if next_offset < len(matches) else None,
"truncated": next_offset < len(matches),
"scope": query_scope(document, page),
}
def query_scope(document: dict[str, Any], nodes: list[dict[str, Any]]) -> dict[str, Any]:
"""Report bounded static knowledge, never inferred accuracy or completeness."""
return {
"basis": "static_snapshot",
"completeness": "unknown",
"returned_nodes": len(nodes),
"external_or_unresolved_nodes": sum(
str(node.get("type", "")).startswith("External") for node in nodes
),
"snapshot_warning_count": len(document.get("warnings", [])),
"unresolved_call_count": document.get("statistics", {}).get("unresolved_calls"),
"interpretation": "External references are not proven runtime targets; missing results do not prove absence.",
}
def _portable_relationship_evidence(value: Any) -> list[dict[str, Any]]:
if not isinstance(value, list):
return []
portable: list[dict[str, Any]] = []
for raw in value[:MAX_EDGE_EVIDENCE_ITEMS]:
if not isinstance(raw, dict):
continue
rule_id = raw.get("rule_id")
basis = raw.get("basis")
runtime_status = raw.get("runtime_status")
if (
not isinstance(rule_id, str)
or not re.fullmatch(r"[a-z][a-z0-9_.-]{2,79}", rule_id)
or basis not in EVIDENCE_BASES
or runtime_status not in RUNTIME_STATUSES
):
continue
item: dict[str, Any] = {
"rule_id": rule_id,
"basis": basis,
"runtime_status": runtime_status,
}
path = raw.get("path")
if isinstance(path, str) and _portable_relative_path(path):
item["path"] = path.replace("\\", "/")
line_start = raw.get("line_start")
line_end = raw.get("line_end")
if isinstance(line_start, int) and line_start >= 1:
item["line_start"] = line_start
item["line_end"] = max(
line_start,
line_end if isinstance(line_end, int) else line_start,
)
limitations = raw.get("limitations")
if isinstance(limitations, list):
clean_limitations = sorted(
{
item
for item in limitations[:32]
if isinstance(item, str)
and re.fullmatch(r"[a-z][a-z0-9_.-]{2,79}", item)
}
)
if clean_limitations:
item["limitations"] = clean_limitations
portable.append(item)
return _bounded_edge_evidence(portable)
def relationship_evidence_id(
source: str, target: str, edge_type: str, evidence: dict[str, Any],
) -> str:
body = {key: value for key, value in evidence.items() if key not in {"id", "evidence_id"}}
encoded = json.dumps(
{"source": source, "target": target, "type": edge_type, "evidence": body},
sort_keys=True, ensure_ascii=False, separators=(",", ":"),
).encode("utf-8")
return "evidence:" + hashlib.sha256(encoded).hexdigest()[:24]
def relationship_evidence(edge: dict[str, Any]) -> list[dict[str, Any]]:
return [
{**item, "evidence_id": relationship_evidence_id(
str(edge.get("source", "")), str(edge.get("target", "")), str(edge.get("type", "")), item,
)}
for item in _portable_relationship_evidence(edge.get("evidence"))
]
def impact_document(
document: dict[str, Any], symbol: str, depth: int, *, direction: str = "both",
relationships: Iterable[str] | None = None, limit: int | None = None,
index: SnapshotIndex | None = None,
) -> dict[str, Any]:
if direction not in {"incoming", "outgoing", "both"}:
raise OntologyError("Impact direction must be incoming, outgoing, or both.")
if isinstance(depth, bool) or not isinstance(depth, int) or not 1 <= depth <= 5:
raise OntologyError("Impact depth must be from 1 to 5.")
selected_relations = set(DEPENDENCY_RELATIONSHIPS if relationships is None else relationships)
if not selected_relations or not selected_relations <= set(EDGE_EVIDENCE_DEFAULTS):
raise OntologyError("Unsupported impact relationship filter.")
result_limit = MAX_IMPACT_RESULTS if limit is None else limit
if isinstance(result_limit, bool) or not isinstance(result_limit, int) or not 1 <= result_limit <= MAX_IMPACT_RESULTS:
raise OntologyError("Impact result limit is out of range.")
# Resolve exact identities before any result cap, including in large graphs.
needle = symbol.casefold()
indexed = index or SnapshotIndex(document)
ranked = [(_query_rank(node, needle), node) for node in indexed.nodes.values()]
exact = [(rank, node) for rank, node in ranked if rank < 3]
best = min((rank for rank, _ in exact), default=3)
candidates = [node for rank, node in exact if rank == best]
if not candidates:
query = query_document(document, symbol, MAX_QUERY_RESULTS, index=indexed)
candidates = query["matches"]
if not candidates:
return {"symbol": symbol, "status": "not_found", "candidates": [], "impact": [],
"scope": query_scope(document, []),
"interpretation": "No match in this partial static snapshot; absence is not established."}
if len(candidates) > 1:
return {
"symbol": symbol,
"status": "ambiguous",
"candidates": [
{
"id": node["id"],
"name": node["name"],
"qualified_name": node.get("qualified_name"),
}
for node in candidates[:20]
],
"impact": [],
"scope": query_scope(document, candidates[:20]),
"interpretation": "Select an exact node id before following dependency paths.",
}
start = candidates[0]
nodes_by_id = indexed.nodes
excluded_edges: set[tuple[str, str, str]] = set()
visited = {start["id"]}
queue: deque[tuple[str, int]] = deque([(start["id"], 0)])
impact: list[dict[str, Any]] = []
paths: dict[str, list[dict[str, Any]]] = {start["id"]: []}
truncated = False
while queue:
current, current_depth = queue.popleft()
if current_depth >= depth:
continue
for neighbor, edge, step_direction in sorted(
indexed.adjacency.get(current, []),
key=lambda item: (item[0], item[1]["type"], item[2]),
):
endpoints = [nodes_by_id.get(edge[key], {}) for key in ("source", "target")]
if edge["type"] not in selected_relations or any(
node.get("type") in GENERIC_CONCEPT_TYPES or node.get("language") in {"Framework", "Concept"}
for node in endpoints
):
excluded_edges.add((edge["source"], edge["target"], edge["type"]))
continue
if direction != "both" and step_direction != direction:
continue
if neighbor in visited:
continue
visited.add(neighbor)
relationship = edge["type"]
evidence = relationship_evidence(edge)
next_depth = current_depth + 1
neighbor_node = nodes_by_id.get(
neighbor,
{"id": neighbor, "name": neighbor, "type": "Unknown", "language": "Unknown"},
)
step = {
"source": current if step_direction == "outgoing" else neighbor,
"target": neighbor if step_direction == "outgoing" else current,
"type": relationship, "direction": step_direction, "evidence": evidence,
}
paths[neighbor] = [*paths[current], step]
impact.append(
{
"depth": next_depth,
"relationship": relationship,
"direction": step_direction,
"evidence": evidence,
"node": neighbor_node,
"via": current,
"path": paths[neighbor],
}
)
if len(impact) > result_limit:
impact.pop()
truncated = True
queue.clear()
break
queue.append((neighbor, next_depth))
impact.sort(key=lambda item: (item["depth"], item["node"]["name"].casefold()))
return {
"symbol": symbol,
"status": "ok",
"root": start,
"depth": depth,
"direction": direction,
"relationships": sorted(selected_relations),
"excluded_edges": len(excluded_edges),
"excluded_edges_scope": "encountered_during_traversal",
"impact_count": len(impact),
"truncated": truncated,
"impact": impact,
"scope": query_scope(document, [start, *(item["node"] for item in impact)]),
"interpretation": "One shortest dependency path per reachable node; static evidence is not runtime proof.",
}
def _read_visualization_asset(relative_path: str, expected_sha256: str | None = None) -> str:
"""Read a bundled workbench asset and verify pinned third-party bytes."""
asset_root = VISUALIZATION_ASSET_DIR.resolve()
raw_asset = VISUALIZATION_ASSET_DIR / relative_path
if raw_asset.is_symlink():
raise OntologyError(f"Visualization asset may not be a symbolic link: {relative_path}")
asset = raw_asset.resolve()
if not _is_relative_to(asset, asset_root) or asset == asset_root:
raise OntologyError("Visualization asset escaped the bundled asset directory.")
if not asset.is_file():
raise OntologyError(f"Visualization asset is missing: {relative_path}")
try:
raw = asset.read_bytes()
except OSError as exc:
raise OntologyError(f"Visualization asset is unreadable: {relative_path}") from exc
if expected_sha256 and hashlib.sha256(raw).hexdigest() != expected_sha256:
raise OntologyError(f"Visualization asset failed integrity verification: {relative_path}")
try:
return raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise OntologyError(f"Visualization asset is not UTF-8: {relative_path}") from exc
def _portable_visualization_node(node: dict[str, Any]) -> dict[str, Any]:
"""Copy only ontology fields intended for the portable browser artifact."""
portable: dict[str, Any] = {
key: node[key]
for key in ("id", "type", "name", "language", "qualified_name")
if key in node and isinstance(node[key], (str, int, float, bool))
}
path = node.get("path")
if isinstance(path, str) and _is_portable_visualization_path(path):
portable["path"] = path
metadata = node.get("metadata")
if isinstance(metadata, dict):
allowed_metadata = {
key: metadata[key]
for key in (
"accessor",
"control_kind",
"ordinal",
"parameter_count",
"parameter_types",
"return_type",
"semantic_groups",
"line_start",
"line_end",
)
if key in metadata
and isinstance(metadata[key], (str, int, float, bool, list, type(None)))
}
if allowed_metadata:
portable["metadata"] = allowed_metadata
return portable
def _is_portable_visualization_path(value: str) -> bool:
if not value or Path(value).is_absolute() or re.match(r"^[A-Za-z]:[\\/]", value):
return False
normalized_parts = value.replace("\\", "/").split("/")
return ".." not in normalized_parts
def _visualization_statistics(document: dict[str, Any]) -> dict[str, Any]:
nodes = document.get("nodes", [])
edges = document.get("edges", [])
original = document.get("statistics", {})
source_files = original.get("source_files", {}) if isinstance(original, dict) else {}
skipped = original.get("skipped", {}) if isinstance(original, dict) else {}
return {
"sourceFiles": {
str(key): int(value)
for key, value in sorted(source_files.items())
if isinstance(value, int) and value >= 0
},
"nodes": len(nodes),
"edges": len(edges),
"nodeTypes": dict(sorted(Counter(str(node.get("type", "Unknown")) for node in nodes).items())),
"edgeTypes": dict(sorted(Counter(str(edge.get("type", "Unknown")) for edge in edges).items())),
"skipped": {
str(key): int(value)
for key, value in sorted(skipped.items())
if isinstance(value, int) and value >= 0
},
"warnings": len(document.get("warnings", [])),
}
def _portable_quality(value: Any) -> dict[str, Any]:
if not isinstance(value, dict):
return {
"contract_version": "legacy_unknown",
"relationship_evidence": {
"total_edges": 0,
"documented_edges": 0,
"missing_evidence": 0,
"coverage_percent": 0.0,
"basis_counts": {},
"runtime_status_counts": {},
},
"adapters": {},
"interpretation": "Legacy snapshot without ontology quality metadata.",
}
relationship = value.get("relationship_evidence", {})
if not isinstance(relationship, dict):
relationship = {}
counts = {}
for field in (
"total_edges",
"documented_edges",
"missing_evidence",
"source_span_edges",
):
raw = relationship.get(field, 0)
counts[field] = raw if isinstance(raw, int) and raw >= 0 else 0
for field in ("coverage_percent", "source_span_coverage_percent"):
raw = relationship.get(field, 0.0)
counts[field] = (
round(max(0.0, min(100.0, float(raw))), 3)
if isinstance(raw, (int, float)) and not isinstance(raw, bool)
else 0.0
)
for field in ("basis_counts", "runtime_status_counts"):
raw = relationship.get(field, {})
counts[field] = (
{
str(key): int(count)
for key, count in sorted(raw.items())
if isinstance(key, str) and isinstance(count, int) and count >= 0
}
if isinstance(raw, dict)
else {}
)
adapters: dict[str, Any] = {}
raw_adapters = value.get("adapters", {})
if isinstance(raw_adapters, dict):
for language, raw_adapter in sorted(raw_adapters.items()):
if not isinstance(language, str) or not isinstance(raw_adapter, dict):
continue
status = raw_adapter.get("status", "unsupported")
if status not in {"supported", "partial", "unsupported"}:
status = "unsupported"
capabilities = raw_adapter.get("capabilities", {})
clean_capabilities = (
{
str(name): capability_status
for name, capability_status in sorted(capabilities.items())
if isinstance(name, str)
and capability_status in {"supported", "partial", "unsupported"}
}
if isinstance(capabilities, dict)
else {}
)
unsupported_runtime = raw_adapter.get("unsupported_runtime", [])
adapters[language] = {
"status": status,
"detected": raw_adapter.get("detected") is True,
"capabilities": clean_capabilities,
"unsupported_runtime": [
item
for item in unsupported_runtime[:32]
if isinstance(item, str)
and re.fullmatch(r"[a-z][a-z0-9_.-]{2,79}", item)
]
if isinstance(unsupported_runtime, list)
else [],
}
interpretation = value.get("interpretation", "")
return {
"contract_version": str(value.get("contract_version", "legacy_unknown")),
"relationship_evidence": counts,
"adapters": adapters,
"interpretation": interpretation[:500] if isinstance(interpretation, str) else "",
}
def canonical_diff(
document: dict[str, Any],
previous_document: dict[str, Any] | None,
limit: int = VISUALIZATION_DIFF_ITEM_LIMIT,
) -> dict[str, Any]:
if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= VISUALIZATION_DIFF_ITEM_LIMIT:
raise OntologyError("Diff limit must be from 1 to 500.")
current_companion = document.get("companion", {})
after_snapshot = (
str(current_companion.get("snapshotId", "current"))
if isinstance(current_companion, dict)
else "current"
)
empty_counts = {
"nodesAdded": 0,
"nodesRemoved": 0,
"nodesModified": 0,
"edgesAdded": 0,
"edgesRemoved": 0,
"edgesModified": 0,
}
if previous_document is None:
return {
"available": False,
"basis": "no_previous_snapshot",
"beforeSnapshotId": "",
"afterSnapshotId": after_snapshot,
"counts": empty_counts,
"nodesAdded": [],
"nodesRemoved": [],
"nodesModified": [],
"edgesAdded": [],
"edgesRemoved": [],
"edgesModified": [],
"truncated": False,
}
current_nodes = {
str(node["id"]): _portable_visualization_node(node)
for node in document.get("nodes", [])
if isinstance(node, dict) and isinstance(node.get("id"), str)
}
previous_nodes = {
str(node["id"]): _portable_visualization_node(node)
for node in previous_document.get("nodes", [])
if isinstance(node, dict) and isinstance(node.get("id"), str)
}
added_ids = sorted(current_nodes.keys() - previous_nodes.keys())
removed_ids = sorted(previous_nodes.keys() - current_nodes.keys())
modified_ids = sorted(
node_id
for node_id in current_nodes.keys() & previous_nodes.keys()
if current_nodes[node_id] != previous_nodes[node_id]
)
def edge_keys(value: dict[str, Any]) -> set[tuple[str, str, str]]:
return {
(str(edge.get("source", "")), str(edge.get("target", "")), str(edge.get("type", "")))
for edge in value.get("edges", [])
if isinstance(edge, dict)
}
current_edges = edge_keys(document)
previous_edges = edge_keys(previous_document)
previous_companion = previous_document.get("companion", {})
current_fingerprint = (
current_companion.get("sourceFingerprint") if isinstance(current_companion, dict) else None
)
previous_fingerprint = (
previous_companion.get("sourceFingerprint") if isinstance(previous_companion, dict) else None
)
current_version = document.get("generator", {}).get("version")
previous_version = previous_document.get("generator", {}).get("version")
if not current_fingerprint or not previous_fingerprint:
basis = "legacy_unknown"
elif current_fingerprint != previous_fingerprint:
basis = "mixed" if current_version and previous_version and current_version != previous_version else "source_change"
elif current_version and previous_version and current_version != previous_version:
basis = "analyzer_reinterpretation"
elif current_version and previous_version:
before_companion_version = previous_companion.get("companionVersion")
after_companion_version = current_companion.get("companionVersion")
basis = "analysis_refresh" if before_companion_version and after_companion_version and before_companion_version != after_companion_version else "no_change"
else:
basis = "analysis_refresh"
added_edges = sorted(current_edges - previous_edges)
removed_edges = sorted(previous_edges - current_edges)
current_by_edge = {
(edge["source"], edge["target"], edge["type"]): relationship_evidence(edge)
for edge in document.get("edges", [])
}
previous_by_edge = {
(edge["source"], edge["target"], edge["type"]): relationship_evidence(edge)
for edge in previous_document.get("edges", [])
}
modified_edges = sorted(
key for key in current_edges & previous_edges
if current_by_edge[key] != previous_by_edge[key]
)
def portable_edges(values: list[tuple[str, str, str]]) -> list[dict[str, str]]:
return [
{"source": source, "target": target, "type": edge_type}
for source, target, edge_type in values[:limit]
]
before_snapshot = (
str(previous_companion.get("snapshotId", "previous"))
if isinstance(previous_companion, dict)
else "previous"
)
return {
"available": True,
"basis": basis,
"beforeSnapshotId": before_snapshot,
"afterSnapshotId": after_snapshot,
"counts": {
"nodesAdded": len(added_ids),
"nodesRemoved": len(removed_ids),
"nodesModified": len(modified_ids),
"edgesAdded": len(added_edges),
"edgesRemoved": len(removed_edges),
"edgesModified": len(modified_edges),
},
"nodesAdded": [current_nodes[node_id] for node_id in added_ids[:limit]],
"nodesRemoved": [previous_nodes[node_id] for node_id in removed_ids[:limit]],
"nodesModified": [current_nodes[node_id] for node_id in modified_ids[:limit]],
"edgesAdded": portable_edges(added_edges),
"edgesRemoved": portable_edges(removed_edges),
"edgesModified": [
{"source": source, "target": target, "type": edge_type,
"evidence": current_by_edge[(source, target, edge_type)],
"previousEvidence": previous_by_edge[(source, target, edge_type)]}
for source, target, edge_type in modified_edges[:limit]
],
"truncated": any(
len(values) > limit
for values in (added_ids, removed_ids, modified_ids, added_edges, removed_edges, modified_edges)
),
}
def _visualization_diff(document: dict[str, Any], previous_document: dict[str, Any] | None) -> dict[str, Any]:
return canonical_diff(document, previous_document)
def _visualization_payload(
document: dict[str, Any],
max_nodes: int,
previous_document: dict[str, Any] | None = None,
) -> dict[str, Any]:
nodes = sorted(
(
_portable_visualization_node(node)
for node in document.get("nodes", [])
if isinstance(node, dict) and isinstance(node.get("id"), str)
),
key=lambda node: node["id"],
)
node_ids = {node["id"] for node in nodes}
edges = sorted(
(
{
"source": str(edge["source"]),
"target": str(edge["target"]),
"type": str(edge["type"]),
"evidence": relationship_evidence(edge),
}
for edge in document.get("edges", [])
if isinstance(edge, dict)
and all(isinstance(edge.get(key), str) for key in ("source", "target", "type"))
and edge["source"] in node_ids
and edge["target"] in node_ids
),
key=lambda edge: (edge["source"], edge["target"], edge["type"]),
)
repository = document.get("repository", {})
generator = document.get("generator", {})
companion = document.get("companion", {})
warnings = []
for warning in document.get("warnings", []):
if not isinstance(warning, dict):
continue
warning_path = warning.get("path")
message = warning.get("message")
if (
isinstance(warning_path, str)
and _is_portable_visualization_path(warning_path)
and isinstance(message, str)
):
warnings.append({"path": warning_path, "message": message})
return {
"meta": {
"repositoryName": str(repository.get("name", "repository")),
"generatedAt": str(document.get("generated_at", "")),
"generatorVersion": str(generator.get("version", PLUGIN_VERSION)),
"snapshotId": str(companion.get("snapshotId", "standalone")),
"evidenceType": str(companion.get("evidenceType", "observed")),
},
"statistics": _visualization_statistics(document),
"quality": _portable_quality(document.get("quality")),
"warnings": warnings,
"nodes": nodes,
"edges": edges,
"changes": _visualization_diff(document, previous_document),
"limits": {
"maxVisibleNodes": min(
len(nodes), max(1, min(int(max_nodes), VISUALIZATION_MAX_VISIBLE_NODES))
),
"searchResultLimit": 80,
},
}
def render_visualization(
document: dict[str, Any],
max_nodes: int,
previous_document: dict[str, Any] | None = None,
) -> str:
"""Render a self-contained, offline, progressive-disclosure workbench."""
if max_nodes < 1:
raise OntologyError("Visualization max_nodes must be at least 1.")
template = _read_visualization_asset("workbench.html")
stylesheet = _read_visualization_asset("workbench.css")
application = _read_visualization_asset("workbench.js")
cytoscape = _read_visualization_asset(*VISUALIZATION_VENDOR_ASSETS["cytoscape"])
elk = _read_visualization_asset(*VISUALIZATION_VENDOR_ASSETS["elk"])
data = json.dumps(
_visualization_payload(document, max_nodes, previous_document),
ensure_ascii=True,
separators=(",", ":"),
)
# A JSON script element can still be terminated by source-derived HTML.
data = data.replace("&", "\\u0026").replace("<", "\\u003c").replace(">", "\\u003e")
title = html.escape(str(document.get("repository", {}).get("name", "repository")), quote=False)
replacements = {
"__CODE_ONTOLOGY_TITLE__": title,
"__CODE_ONTOLOGY_CSS__": stylesheet,
"__CODE_ONTOLOGY_CYTOSCAPE__": cytoscape,
"__CODE_ONTOLOGY_ELK__": elk,
"__CODE_ONTOLOGY_DATA__": data,
"__CODE_ONTOLOGY_APP__": application,
}
missing = [marker for marker in replacements if marker not in template]
if missing:
raise OntologyError("Visualization template is missing required placeholders.")
marker_pattern = re.compile("|".join(re.escape(marker) for marker in replacements))
return marker_pattern.sub(lambda match: replacements[match.group(0)], template)
def write_visualization(
index_path: str,
output_path: str,
max_nodes: int,
overwrite: bool = False,
previous_index_path: str | None = None,
) -> dict[str, Any]:
index, document = load_document(index_path)
previous_document = None
if previous_index_path:
_, previous_document = load_document(previous_index_path)
raw_output = Path(output_path).expanduser()
if raw_output.is_symlink():
raise OntologyError("The visualization output may not be a symbolic link.")
output = raw_output.resolve()
if output.suffix.lower() != ".html":
raise OntologyError("Visualization output must end in .html")
if output.parent != index.parent:
raise OntologyError("Visualization output must be in the ontology index directory.")
if output.exists() and not overwrite:
raise OntologyError("Refusing to replace an existing visualization without --overwrite")
_write_text(output, render_visualization(document, max_nodes, previous_document))
visible_limit = min(
len(document["nodes"]), max(1, min(max_nodes, VISUALIZATION_MAX_VISIBLE_NODES))
)
return {
"status": "visualized",
"index": str(index),
"output": str(output),
"nodes_rendered": visible_limit,
"nodes_indexed": len(document["nodes"]),
"max_visible_nodes": visible_limit,
"network_dependencies": 0,
}
def _json_print(value: Any) -> None:
print(json.dumps(value, indent=2, ensure_ascii=False))
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="code_ontology.py",
description="Build and explore a local static code ontology without executing target code.",
)
subparsers = parser.add_subparsers(dest="command", required=True)
preflight = subparsers.add_parser("preflight", help="Read-only scan summary; writes nothing.")
preflight.add_argument("--repo", required=True, help="Authorized repository directory.")
index = subparsers.add_parser("index", help="Build JSON, RDF/Turtle, and Markdown artifacts.")
index.add_argument("--repo", required=True, help="Authorized repository directory.")
index.add_argument("--output", required=True, help="Output directory outside the repository.")
index.add_argument(
"--authorized",
action="store_true",
help="Confirm the repository owner authorized static analysis.",
)
index.add_argument(
"--overwrite",
action="store_true",
help="Replace existing ontology artifacts after explicit confirmation.",
)
query = subparsers.add_parser("query", help="Search nodes in an ontology JSON index.")
query.add_argument("--index", required=True, help="Path to ontology.json.")
query.add_argument("--term", required=True, help="Case-insensitive symbol or concept search.")
query.add_argument("--limit", type=int, default=20, choices=range(1, 201), metavar="1..200")
query.add_argument("--offset", type=int, default=0)
query.add_argument("--language")
query.add_argument("--node-type")
query.add_argument("--path-prefix")
impact = subparsers.add_parser("impact", help="Explore the static relationship neighborhood.")
impact.add_argument("--index", required=True, help="Path to ontology.json.")
impact.add_argument("--symbol", required=True, help="Name, qualified name, or exact node id.")
impact.add_argument("--depth", type=int, default=2, choices=range(1, 6), metavar="1..5")
impact.add_argument("--direction", choices=("incoming", "outgoing", "both"), default="both")
impact.add_argument("--relationship", action="append", choices=sorted(EDGE_EVIDENCE_DEFAULTS))
impact.add_argument("--limit", type=int, default=200)
visualize = subparsers.add_parser(
"visualize", help="Create a self-contained offline HTML graph."
)
visualize.add_argument("--index", required=True, help="Path to ontology.json.")
visualize.add_argument("--output", required=True, help="Destination .html path.")
visualize.add_argument(
"--max-nodes", type=int, default=500, choices=range(1, 2001), metavar="1..2000"
)
visualize.add_argument(
"--overwrite",
action="store_true",
help="Replace an existing HTML file after explicit confirmation.",
)
visualize.add_argument(
"--previous-index",
help="Optional previous ontology.json used for a local snapshot diff.",
)
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
if args.command == "preflight":
repo = _resolve_dir(args.repo, "Repository")
_json_print(preflight_document(repo))
elif args.command == "index":
repo = _resolve_dir(args.repo, "Repository")
_json_print(
write_index(
repo,
Path(args.output),
args.authorized,
overwrite=args.overwrite,
)
)
elif args.command == "query":
_, document = load_document(args.index)
_json_print(query_document(document, args.term, args.limit, offset=args.offset,
language=args.language, node_type=args.node_type, path_prefix=args.path_prefix))
elif args.command == "impact":
_, document = load_document(args.index)
_json_print(impact_document(document, args.symbol, args.depth, direction=args.direction,
relationships=args.relationship, limit=args.limit))
elif args.command == "visualize":
_json_print(
write_visualization(
args.index,
args.output,
args.max_nodes,
overwrite=args.overwrite,
previous_index_path=args.previous_index,
)
)
else:
parser.error(f"Unknown command: {args.command}")
except OntologyError as exc:
print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 3318c12dd1f37e1a8402f83c5e03a385a793de15b9776b9c2f5e4bc49eaea267