diff --git a/src/docforge/adapters/__init__.py b/src/docforge/adapters/__init__.py index 489a70b..0ae5694 100644 --- a/src/docforge/adapters/__init__.py +++ b/src/docforge/adapters/__init__.py @@ -1,5 +1,11 @@ """Repository-owned reference adapters built on the public adapter SDK.""" +from .javascript import ( + JAVASCRIPT_ADAPTER_VERSION, + JAVASCRIPT_EXTRACTOR_VERSION, + JavaScriptReferenceAdapter, + JavaScriptUnsupportedFact, +) from .python import ( PYTHON_ADAPTER_VERSION, PYTHON_EXTRACTOR_VERSION, @@ -8,8 +14,12 @@ from .python import ( ) __all__ = [ + "JAVASCRIPT_ADAPTER_VERSION", + "JAVASCRIPT_EXTRACTOR_VERSION", "PYTHON_ADAPTER_VERSION", "PYTHON_EXTRACTOR_VERSION", + "JavaScriptReferenceAdapter", + "JavaScriptUnsupportedFact", "PythonReferenceAdapter", "PythonUnsupportedFact", ] diff --git a/src/docforge/adapters/javascript.py b/src/docforge/adapters/javascript.py new file mode 100644 index 0000000..82f733f --- /dev/null +++ b/src/docforge/adapters/javascript.py @@ -0,0 +1,1002 @@ +"""Deterministic JavaScript and TypeScript reference adapter. + +The adapter parses explicitly declared source roots as untrusted data with the +optional pinned Tree-sitter frontends. It never imports, compiles, transpiles, +or executes project code. Only relative static imports and re-exports that +resolve to another inventoried source file become dependency evidence. +""" + +from __future__ import annotations + +import hashlib +import importlib +import os +import posixpath +import stat +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Protocol, cast + +from ..adapter_sdk import ( + AdapterAssembly, + AdapterEdge, + AdapterManifest, + AdapterNode, + AdapterProjection, + AdapterSource, + AdapterSourceProjection, + Edge, +) +from ..adapter_sdk import ( + Node as GraphNode, +) +from ..errors import DocForgeError +from ..treesitter_logic import ( + DiscoveredFunction, + TreeSitterLogicOwner, + analyze_javascript_source, + analyze_typescript_source, + discover_javascript_functions, + discover_typescript_functions, +) + +if TYPE_CHECKING: + from tree_sitter import Language + +JAVASCRIPT_ADAPTER_ID = "docforge.reference.javascript" +JAVASCRIPT_ADAPTER_VERSION = "1" +JAVASCRIPT_EXTRACTOR_VERSION = "tree-sitter-script@1" +JAVASCRIPT_IDENTITY_VERSION = "javascript-reference-id@1" +JAVASCRIPT_SUPPORT_SCHEMA_VERSION = 1 + +_JAVASCRIPT_SUFFIXES = frozenset({".js", ".mjs", ".cjs"}) +_TYPESCRIPT_SUFFIXES = frozenset({".ts", ".mts", ".cts"}) +_SOURCE_SUFFIXES = (*sorted(_JAVASCRIPT_SUFFIXES), *sorted(_TYPESCRIPT_SUFFIXES)) +_FUNCTION_TYPES = frozenset( + {"function_declaration", "generator_function_declaration", "method_definition"} +) + +_SUPPORTED_FACTS = ( + "script_file", + "script_module", + "script_class", + "script_function", + "lexical_containment", + "project_local_static_import_dependency", + "function_logic", +) + + +@dataclass(frozen=True) +class JavaScriptUnsupportedFact: + """One semantic fact this syntax-only reference adapter does not claim.""" + + code: str + description: str + + def as_dict(self) -> dict[str, str]: + return {"code": self.code, "description": self.description} + + +_UNSUPPORTED_FACTS = ( + JavaScriptUnsupportedFact( + "call_resolution", + "Calls are represented only inside function Logic and are not resolved to symbols.", + ), + JavaScriptUnsupportedFact( + "dynamic_module_resolution", + "Dynamic import(), require(), package exports, and loader hooks are not dependencies.", + ), + JavaScriptUnsupportedFact( + "inheritance_resolution", + "Class extends and implements clauses are not resolved.", + ), + JavaScriptUnsupportedFact( + "module_configuration_resolution", + "Bare specifiers, aliases, tsconfig paths, and JavaScript-to-TypeScript " + "remapping are omitted.", + ), + JavaScriptUnsupportedFact( + "runtime_generated_facts", + "Decorators, prototypes, proxies, eval, and executed module code are never evaluated.", + ), + JavaScriptUnsupportedFact( + "type_and_symbol_resolution", + "Types, interfaces, overloads, variables, references, and re-exported " + "symbols are not resolved.", + ), +) + + +class _Point(Protocol): + row: int + + +class _TreeNode(Protocol): + type: str + has_error: bool + is_error: bool + is_missing: bool + start_byte: int + end_byte: int + start_point: _Point + named_children: list[_TreeNode] + + def child_by_field_name(self, name: str, /) -> _TreeNode | None: ... + + +@dataclass(frozen=True) +class _SourceRecord: + source_id: str + source_path: str + module_name: str + language: str + fingerprint: str + text: str + raw: bytes + root: _TreeNode + + +@dataclass(frozen=True) +class _Definition: + kind: str + qualified_name: str + line: int + parent_qualified_name: str | None + node: _TreeNode + node_id: str + + +@lru_cache(maxsize=2) +def _language(language: str) -> Language: + if language == "javascript": + grammar_module = "tree_sitter_javascript" + grammar_function = "language" + extra = "javascript" + elif language == "typescript": + grammar_module = "tree_sitter_typescript" + grammar_function = "language_typescript" + extra = "typescript" + else: + raise ValueError(f"Unsupported script language: {language}") + try: + tree_sitter = importlib.import_module("tree_sitter") + grammar = importlib.import_module(grammar_module) + except ModuleNotFoundError as error: + raise DocForgeError( + "optional_dependency_missing", + "The requested language frontend is not installed", + extra=extra, + install=f"docforge[{extra}]", + missing_module=error.name, + ) from error + language_type = tree_sitter.Language + language_factory = getattr(grammar, grammar_function) + return cast("Language", language_type(language_factory())) + + +class JavaScriptReferenceAdapter: + """Reference JavaScript/TypeScript adapter over explicit source roots.""" + + def __init__( + self, + root: Path, + *, + source_roots: Iterable[str | Path], + project_id: str = "javascript-reference", + title: str = "JavaScript and TypeScript reference project", + max_sources: int = 4_096, + max_source_bytes: int = 1_000_000, + max_logic_nodes_per_function: int = 2_000, + ) -> None: + resolved_root = root.resolve(strict=True) + if not resolved_root.is_dir(): + raise DocForgeError("invalid_adapter", "Script adapter root must be a directory") + if max_sources < 1 or max_source_bytes < 1 or max_logic_nodes_per_function < 2: + raise ValueError("Script adapter limits must be positive") + self.root = resolved_root + self.project_id = project_id + self.title = title + self.max_sources = max_sources + self.max_source_bytes = max_source_bytes + self.max_logic_nodes_per_function = max_logic_nodes_per_function + self.source_roots = self._normalize_source_roots(source_roots) + + def support_report(self) -> dict[str, object]: + """Return deterministic scope, frontend, and limitation evidence.""" + + return { + "schema_version": JAVASCRIPT_SUPPORT_SCHEMA_VERSION, + "adapter_id": JAVASCRIPT_ADAPTER_ID, + "adapter_version": JAVASCRIPT_ADAPTER_VERSION, + "extractor_version": JAVASCRIPT_EXTRACTOR_VERSION, + "identity_version": JAVASCRIPT_IDENTITY_VERSION, + "frontends": { + "javascript": "tree-sitter-javascript", + "typescript": "tree-sitter-typescript", + }, + "imports_project_code": False, + "executes_project_code": False, + "source_roots": list(self.source_roots), + "source_suffixes": list(_SOURCE_SUFFIXES), + "supported_facts": list(_SUPPORTED_FACTS), + "unsupported_facts": [fact.as_dict() for fact in _UNSUPPORTED_FACTS], + } + + def unsupported_facts(self) -> tuple[JavaScriptUnsupportedFact, ...]: + return _UNSUPPORTED_FACTS + + def load_manifest(self) -> AdapterManifest: + records = self._inventory() + path_sources = {record.source_path: record.source_id for record in records} + sources = tuple( + sorted( + ( + AdapterSource( + source_id=record.source_id, + source_path=record.source_path, + fingerprint=record.fingerprint, + extractor_version=(f"{JAVASCRIPT_EXTRACTOR_VERSION}:{record.language}"), + dependencies=self._local_dependencies(record, path_sources), + ) + for record in records + ), + key=lambda source: source.source_id, + ) + ) + source_hash = self._source_hash(sources) + return AdapterManifest( + project_id=self.project_id, + title=self.title, + adapter_id=JAVASCRIPT_ADAPTER_ID, + adapter_version=JAVASCRIPT_ADAPTER_VERSION, + root=self.root, + revision=source_hash[:12], + source_hash=source_hash, + families=("code",), + allowed_relations=("contains", "depends_on"), + sources=sources, + estimated_nodes=max(1, len(sources) * 8), + ) + + def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: + record = self._read_source(source.source_path) + if record.source_id != source.source_id or record.fingerprint != source.fingerprint: + raise DocForgeError( + "stale_adapter_source", + "Script source changed after its manifest was captured", + source=source.source_path, + ) + return self._extract_record(record, source) + + def assemble_projection( + self, + manifest: AdapterManifest, + contributions: tuple[AdapterSourceProjection, ...], + ) -> AdapterAssembly: + expected = tuple(source.source_id for source in manifest.sources) + actual = tuple(sorted(contribution.source_id for contribution in contributions)) + if actual != expected: + raise DocForgeError( + "invalid_adapter", + "Script assembly contributions do not match the current manifest", + ) + nodes = tuple( + sorted( + (node for contribution in contributions for node in contribution.nodes), + key=lambda item: item.node.node_id, + ) + ) + edges = tuple( + sorted( + (edge for contribution in contributions for edge in contribution.edges), + key=lambda item: ( + item.edge.source_id, + item.edge.relation, + item.edge.target_id, + ), + ) + ) + logic = tuple( + sorted( + (projection for contribution in contributions for projection in contribution.logic), + key=lambda projection: projection.owner_node_id, + ) + ) + return AdapterAssembly( + projection=AdapterProjection( + project_id=manifest.project_id, + title=manifest.title, + adapter_id=manifest.adapter_id, + adapter_version=manifest.adapter_version, + root=manifest.root, + revision=manifest.revision, + source_hash=manifest.source_hash, + nodes=nodes, + edges=edges, + ), + logic=logic, + ) + + def load_assembly(self) -> AdapterAssembly: + """Load one cache-independent complete graph and Logic assembly.""" + + manifest = self.load_manifest() + contributions = tuple(self.extract_source(source) for source in manifest.sources) + if self.load_manifest() != manifest: + raise DocForgeError( + "source_changed", + "Script sources changed during complete adapter extraction", + ) + return self.assemble_projection(manifest, contributions) + + def load_complete_assembly(self) -> AdapterAssembly: + return self.load_assembly() + + def load_projection(self) -> AdapterProjection: + return self.load_assembly().projection + + def _normalize_source_roots(self, source_roots: Iterable[str | Path]) -> tuple[str, ...]: + normalized: list[str] = [] + for value in source_roots: + candidate = PurePosixPath(Path(value).as_posix()) + if ( + candidate.is_absolute() + or not candidate.parts + or ".." in candidate.parts + or str(candidate) in {"", "."} + ): + raise DocForgeError( + "path_escape", + "Script source roots must be explicit project-relative directories", + ) + relative = candidate.as_posix() + absolute = self.root.joinpath(*candidate.parts) + if absolute.is_symlink(): + raise DocForgeError("path_escape", "Script source root cannot be a symbolic link") + try: + resolved = absolute.resolve(strict=True) + except OSError as error: + raise DocForgeError( + "invalid_adapter", + "Script source root does not exist", + source_root=relative, + ) from error + if ( + resolved != absolute + or not resolved.is_dir() + or not resolved.is_relative_to(self.root) + ): + raise DocForgeError( + "path_escape", + "Script source root must be a confined real directory", + source_root=relative, + ) + normalized.append(relative) + result = tuple(sorted(set(normalized))) + if not result: + raise DocForgeError("invalid_adapter", "At least one script source root is required") + if len(result) != len(normalized): + raise DocForgeError("invalid_adapter", "Script source roots must be unique") + paths = [PurePosixPath(value) for value in result] + for index, left in enumerate(paths): + for right in paths[index + 1 :]: + if left in right.parents or right in left.parents: + raise DocForgeError( + "invalid_adapter", + "Script source roots must not overlap", + ) + return result + + def _inventory(self) -> tuple[_SourceRecord, ...]: + paths: list[str] = [] + for source_root in self.source_roots: + absolute_root = self.root.joinpath(*PurePosixPath(source_root).parts) + if ( + absolute_root.is_symlink() + or not absolute_root.is_dir() + or absolute_root.resolve(strict=True) != absolute_root + ): + raise DocForgeError( + "path_escape", + "Script source root changed after adapter binding", + source_root=source_root, + ) + for directory, directory_names, file_names in os.walk( + absolute_root, + topdown=True, + followlinks=False, + ): + current = Path(directory) + for name in tuple(directory_names): + child = current / name + if child.is_symlink(): + raise DocForgeError( + "path_escape", + "Script source inventory contains a symbolic-link directory", + source=child.relative_to(self.root).as_posix(), + ) + directory_names[:] = sorted( + name for name in directory_names if name != "node_modules" + ) + for name in sorted(file_names): + if Path(name).suffix not in _SOURCE_SUFFIXES: + continue + child = current / name + relative = child.relative_to(self.root).as_posix() + if child.is_symlink(): + raise DocForgeError( + "path_escape", + "Script source inventory contains a symbolic-link file", + source=relative, + ) + paths.append(relative) + if len(paths) > self.max_sources: + raise DocForgeError( + "adapter_too_large", + "Script source inventory exceeds the configured limit", + maximum=self.max_sources, + ) + records = tuple(self._read_source(path) for path in sorted(paths)) + modules = [record.module_name for record in records] + if len(modules) != len(set(modules)): + duplicates = sorted(module for module in set(modules) if modules.count(module) > 1) + raise DocForgeError( + "ambiguous_script_module", + "Declared script source roots produce duplicate module names", + modules=duplicates, + ) + return records + + def _read_source(self, relative: str) -> _SourceRecord: + path = PurePosixPath(relative) + if path.is_absolute() or ".." in path.parts or path.suffix not in _SOURCE_SUFFIXES: + raise DocForgeError("path_escape", "Script source path is unsafe") + source_root = self._source_root_for(path) + candidate = self.root.joinpath(*path.parts) + try: + before = candidate.lstat() + except OSError as error: + raise DocForgeError( + "stale_adapter_source", + "Script source is unavailable", + source=relative, + ) from error + if ( + stat.S_ISLNK(before.st_mode) + or not stat.S_ISREG(before.st_mode) + or candidate.resolve(strict=True) != candidate + or not candidate.is_relative_to(self.root) + ): + raise DocForgeError( + "path_escape", + "Script source must be a confined regular file", + source=relative, + ) + if before.st_size > self.max_source_bytes: + raise DocForgeError( + "source_too_large", + "Script source exceeds the configured adapter limit", + source=relative, + maximum=self.max_source_bytes, + ) + descriptor = -1 + try: + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(candidate, flags) + opened = os.fstat(descriptor) + if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != ( + before.st_dev, + before.st_ino, + ): + raise DocForgeError( + "path_escape", + "Script source identity changed before it was opened", + source=relative, + ) + with os.fdopen(descriptor, "rb") as handle: + descriptor = -1 + raw = handle.read(self.max_source_bytes + 1) + after_open = os.fstat(handle.fileno()) + after = candidate.lstat() + except DocForgeError: + raise + except OSError as error: + raise DocForgeError( + "stale_adapter_source", + "Script source changed while being read", + source=relative, + ) from error + finally: + if descriptor >= 0: + os.close(descriptor) + identities = ( + (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns), + ( + after_open.st_dev, + after_open.st_ino, + after_open.st_size, + after_open.st_mtime_ns, + ), + (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns), + ) + if ( + len(raw) > self.max_source_bytes + or identities[0] != identities[1] + or identities[0] != identities[2] + or len(raw) != after.st_size + ): + raise DocForgeError( + "stale_adapter_source", + "Script source changed while being read", + source=relative, + ) + try: + text = raw.decode("utf-8-sig") + except UnicodeDecodeError as error: + raise DocForgeError( + "invalid_script_source", + "Script source must use UTF-8", + source=relative, + ) from error + language = self._language_for_suffix(path.suffix) + parse_raw, root = self._parse(text, language) + if root.has_error: + raise DocForgeError( + "invalid_script_source", + f"{language.title()} source cannot be parsed by the reference adapter", + source=relative, + line=self._first_error_line(root) or 1, + ) + return _SourceRecord( + source_id=self._source_id(relative), + source_path=relative, + module_name=self._module_name(path, PurePosixPath(source_root)), + language=language, + fingerprint=hashlib.sha256(raw).hexdigest(), + text=text, + raw=parse_raw, + root=root, + ) + + @staticmethod + def _parse(source: str, language: str) -> tuple[bytes, _TreeNode]: + try: + tree_sitter = importlib.import_module("tree_sitter") + except ModuleNotFoundError as error: + raise DocForgeError( + "optional_dependency_missing", + "The requested language frontend is not installed", + extra=language, + install=f"docforge[{language}]", + missing_module=error.name, + ) from error + raw = source.encode("utf-8") + parser = tree_sitter.Parser(_language(language)) + tree = parser.parse(raw) + return raw, cast("_TreeNode", tree.root_node) + + @classmethod + def _first_error_line(cls, node: _TreeNode) -> int | None: + if node.is_error or node.is_missing: + return node.start_point.row + 1 + for child in node.named_children: + line = cls._first_error_line(child) + if line is not None: + return line + return None + + def _source_root_for(self, path: PurePosixPath) -> str: + matches = [ + source_root + for source_root in self.source_roots + if path == PurePosixPath(source_root) or PurePosixPath(source_root) in path.parents + ] + if len(matches) != 1: + raise DocForgeError( + "path_escape", + "Script source is outside the declared source roots", + source=path.as_posix(), + ) + return matches[0] + + @staticmethod + def _language_for_suffix(suffix: str) -> str: + if suffix in _JAVASCRIPT_SUFFIXES: + return "javascript" + if suffix in _TYPESCRIPT_SUFFIXES: + return "typescript" + raise DocForgeError("invalid_adapter", "Unsupported script source suffix") + + @staticmethod + def _module_name(path: PurePosixPath, source_root: PurePosixPath) -> str: + local = path.relative_to(source_root) + parts = list(local.with_suffix("").parts) + if parts and parts[-1] == "index": + parts.pop() + if not parts: + parts = list(source_root.parts) + return ".".join(parts) + + @staticmethod + def _source_id(relative: str) -> str: + digest = hashlib.sha256( + f"{JAVASCRIPT_IDENTITY_VERSION}\0source\0{relative}".encode() + ).hexdigest()[:24] + return f"javascript.source.{digest}" + + @staticmethod + def _file_node_id(source_id: str) -> str: + return source_id.replace("javascript.source.", "javascript.file.", 1) + + @staticmethod + def _module_node_id(source_id: str) -> str: + return source_id.replace("javascript.source.", "javascript.module.", 1) + + @staticmethod + def _symbol_node_id(kind: str, source_id: str, qualified_name: str) -> str: + digest = hashlib.sha256( + (f"{JAVASCRIPT_IDENTITY_VERSION}\0{kind}\0{source_id}\0{qualified_name}").encode() + ).hexdigest()[:24] + return f"javascript.{kind}.{digest}" + + @staticmethod + def _source_hash(sources: Sequence[AdapterSource]) -> str: + digest = hashlib.sha256() + digest.update(JAVASCRIPT_ADAPTER_VERSION.encode()) + digest.update(JAVASCRIPT_EXTRACTOR_VERSION.encode()) + digest.update(JAVASCRIPT_IDENTITY_VERSION.encode()) + for source in sources: + digest.update(b"\0source\0") + digest.update(source.source_id.encode()) + digest.update(b"\0path\0") + digest.update(source.source_path.encode()) + digest.update(b"\0fingerprint\0") + digest.update(source.fingerprint.encode()) + digest.update(b"\0extractor\0") + digest.update(source.extractor_version.encode()) + for dependency in source.dependencies: + digest.update(b"\0dependency\0") + digest.update(dependency.encode()) + return digest.hexdigest() + + def _local_dependencies( + self, + record: _SourceRecord, + path_sources: dict[str, str], + ) -> tuple[str, ...]: + dependencies: set[str] = set() + for specifier in self._static_specifiers(record): + target_path = self._resolve_relative_specifier( + record.source_path, specifier, path_sources + ) + if target_path is None: + continue + target = path_sources[target_path] + if target != record.source_id: + dependencies.add(target) + return tuple(sorted(dependencies)) + + @classmethod + def _static_specifiers(cls, record: _SourceRecord) -> tuple[str, ...]: + result: set[str] = set() + stack = [record.root] + while stack: + node = stack.pop() + if node.type in {"import_statement", "export_statement"}: + source = node.child_by_field_name("source") + if source is not None: + value = cls._string_literal(source, record.raw) + if value is not None and value.startswith(("./", "../")): + result.add(value) + stack.extend(node.named_children) + return tuple(sorted(result)) + + @staticmethod + def _string_literal(node: _TreeNode, raw: bytes) -> str | None: + value = raw[node.start_byte : node.end_byte].decode("utf-8") + if ( + len(value) < 2 + or value[0] not in {'"', "'"} + or value[-1] != value[0] + or "\\" in value + or "\x00" in value + ): + return None + return value[1:-1] + + @staticmethod + def _resolve_relative_specifier( + source_path: str, + specifier: str, + path_sources: dict[str, str], + ) -> str | None: + parent = PurePosixPath(source_path).parent.as_posix() + normalized = posixpath.normpath(posixpath.join(parent, specifier)) + if normalized == ".." or normalized.startswith("../") or normalized.startswith("/"): + return None + candidate = PurePosixPath(normalized) + if candidate.suffix in _SOURCE_SUFFIXES: + return normalized if normalized in path_sources else None + candidates = [ + *(f"{normalized}{suffix}" for suffix in _SOURCE_SUFFIXES), + *(f"{normalized}/index{suffix}" for suffix in _SOURCE_SUFFIXES), + ] + matches = [value for value in candidates if value in path_sources] + return matches[0] if len(matches) == 1 else None + + def _extract_record( + self, + record: _SourceRecord, + source: AdapterSource, + ) -> AdapterSourceProjection: + file_id = self._file_node_id(source.source_id) + module_id = self._module_node_id(source.source_id) + nodes: list[AdapterNode] = [ + self._node( + node_id=file_id, + title=record.source_path, + kind="file", + qualified_name=record.source_path, + language=record.language, + content=record.text, + source_path=record.source_path, + anchor="L1", + ), + self._node( + node_id=module_id, + title=record.module_name, + kind="module", + qualified_name=record.module_name, + language=record.language, + content=f"{record.language.title()} module {record.module_name}.", + source_path=record.source_path, + anchor="L1", + ), + ] + edges: list[AdapterEdge] = [self._edge(file_id, "contains", module_id, "syntax")] + definitions = self._definitions(record) + definition_ids = { + definition.qualified_name: definition.node_id for definition in definitions + } + for definition in definitions: + parent_id = ( + definition_ids.get(definition.parent_qualified_name, module_id) + if definition.parent_qualified_name is not None + else module_id + ) + anchor = f"L{definition.line}" + content = self._text(definition.node, record.raw).strip() + nodes.append( + self._node( + node_id=definition.node_id, + title=definition.qualified_name, + kind=definition.kind, + qualified_name=f"{record.module_name}.{definition.qualified_name}", + language=record.language, + content=content, + source_path=record.source_path, + anchor=anchor, + ) + ) + edges.append( + self._edge( + parent_id, + "contains", + definition.node_id, + "tree_sitter_syntax", + anchor=anchor, + ) + ) + for dependency in source.dependencies: + edges.append( + self._edge( + module_id, + "depends_on", + self._module_node_id(dependency), + "static_relative_module", + ) + ) + functions = self._discovered_functions(record) + owners = tuple( + TreeSitterLogicOwner( + owner_node_id=self._function_definition(definitions, function).node_id, + qualified_name=function.qualified_name, + line=function.line, + ) + for function in functions + ) + analyzer = ( + analyze_javascript_source + if record.language == "javascript" + else analyze_typescript_source + ) + logic = analyzer( + record.text, + source_id=source.source_id, + owners=owners, + filename=record.source_path, + max_nodes_per_function=self.max_logic_nodes_per_function, + ) + return AdapterSourceProjection( + source_id=source.source_id, + fingerprint=source.fingerprint, + nodes=tuple(sorted(nodes, key=lambda item: item.node.node_id)), + edges=tuple( + sorted( + edges, + key=lambda item: ( + item.edge.source_id, + item.edge.relation, + item.edge.target_id, + ), + ) + ), + logic=logic, + ) + + def _definitions(self, record: _SourceRecord) -> tuple[_Definition, ...]: + definitions: list[_Definition] = [] + seen: set[str] = set() + + def visit(node: _TreeNode, scopes: tuple[str, ...]) -> None: + definition = self._definition(record, node, scopes) + next_scopes = scopes + if definition is not None: + if definition.qualified_name in seen: + raise DocForgeError( + "ambiguous_script_symbol", + "Script source repeats a class or function identity", + source=record.source_path, + qualified_name=definition.qualified_name, + ) + seen.add(definition.qualified_name) + definitions.append(definition) + next_scopes = (*scopes, definition.qualified_name.rsplit(".", 1)[-1]) + for child in node.named_children: + visit(child, next_scopes) + + visit(record.root, ()) + ordered = tuple(sorted(definitions, key=lambda item: item.node_id)) + public_functions = { + (function.qualified_name, function.line) + for function in self._discovered_functions(record) + } + extracted_functions = { + (definition.qualified_name, definition.line) + for definition in ordered + if definition.kind == "function" + } + if extracted_functions != public_functions: + raise DocForgeError( + "invalid_adapter", + "Reference function discovery drifted from the Logic frontend", + source=record.source_path, + ) + return ordered + + def _definition( + self, + record: _SourceRecord, + node: _TreeNode, + scopes: tuple[str, ...], + ) -> _Definition | None: + kind: str | None = None + name: str | None = None + definition_node = node + if node.type in {"class_declaration", "class"}: + name = self._field_text(node, "name", record.raw) + kind = "class" if name is not None else None + elif node.type in _FUNCTION_TYPES: + name = self._field_text(node, "name", record.raw) + kind = "function" if name is not None else None + elif node.type == "variable_declarator": + value = node.child_by_field_name("value") + if value is not None and value.type in {"arrow_function", "function_expression"}: + name = self._field_text(node, "name", record.raw) + definition_node = value + kind = "function" if name is not None else None + if kind is None or name is None: + return None + qualified_name = ".".join((*scopes, name)) + parent = ".".join(scopes) or None + return _Definition( + kind=kind, + qualified_name=qualified_name, + line=node.start_point.row + 1, + parent_qualified_name=parent, + node=definition_node, + node_id=self._symbol_node_id(kind, record.source_id, qualified_name), + ) + + @staticmethod + def _discovered_functions(record: _SourceRecord) -> tuple[DiscoveredFunction, ...]: + if record.language == "javascript": + return discover_javascript_functions(record.text) + return discover_typescript_functions(record.text) + + @staticmethod + def _function_definition( + definitions: tuple[_Definition, ...], + function: DiscoveredFunction, + ) -> _Definition: + matches = [ + definition + for definition in definitions + if definition.kind == "function" + and definition.qualified_name == function.qualified_name + and definition.line == function.line + ] + if len(matches) != 1: + raise DocForgeError( + "invalid_adapter", + "Logic owner does not match one extracted script function", + qualified_name=function.qualified_name, + line=function.line, + ) + return matches[0] + + @staticmethod + def _text(node: _TreeNode, raw: bytes) -> str: + return raw[node.start_byte : node.end_byte].decode("utf-8") + + @classmethod + def _field_text(cls, node: _TreeNode, field: str, raw: bytes) -> str | None: + child = node.child_by_field_name(field) + return cls._text(child, raw) if child is not None else None + + @staticmethod + def _node( + *, + node_id: str, + title: str, + kind: str, + qualified_name: str, + language: str, + content: str, + source_path: str, + anchor: str, + ) -> AdapterNode: + normalized = content.strip() or f"{language.title()} {kind} {qualified_name}." + return AdapterNode( + node=GraphNode( + node_id=node_id, + title=title, + family="code", + authority="derived", + status="active", + tags=tuple(sorted({language, kind})), + summary=f"{language.title()} {kind} fact for {qualified_name}.", + content=normalized, + source_path=source_path, + source_anchor=anchor, + content_hash=hashlib.sha256(normalized.encode()).hexdigest(), + ), + metadata=( + ("extractor", JAVASCRIPT_EXTRACTOR_VERSION), + ("identity", JAVASCRIPT_IDENTITY_VERSION), + ("kind", kind), + ("language", language), + ("qualified_name", qualified_name), + ), + ) + + @staticmethod + def _edge( + source_id: str, + relation: str, + target_id: str, + evidence: str, + *, + anchor: str | None = None, + ) -> AdapterEdge: + metadata = [("evidence", evidence)] + if anchor is not None: + metadata.append(("source_anchor", anchor)) + return AdapterEdge( + edge=Edge(source_id, relation, target_id), + metadata=tuple(metadata), + ) diff --git a/tests/fixtures/reference-javascript/src/app/index.js b/tests/fixtures/reference-javascript/src/app/index.js new file mode 100644 index 0000000..4721b77 --- /dev/null +++ b/tests/fixtures/reference-javascript/src/app/index.js @@ -0,0 +1,8 @@ +import { Service } from "./service.mjs"; +export { clamp } from "./shared.js"; + +throw new Error("The reference adapter must never execute project code"); + +export function applicationName() { + return Service.name; +} diff --git a/tests/fixtures/reference-javascript/src/app/service.mjs b/tests/fixtures/reference-javascript/src/app/service.mjs new file mode 100644 index 0000000..80a7882 --- /dev/null +++ b/tests/fixtures/reference-javascript/src/app/service.mjs @@ -0,0 +1,14 @@ +import { clamp } from "./shared.js"; + +export class Service { + run(value) { + return clamp(value); + } +} + +export const buildService = (enabled) => { + if (enabled) { + return new Service(); + } + return null; +}; diff --git a/tests/fixtures/reference-javascript/src/app/shared.js b/tests/fixtures/reference-javascript/src/app/shared.js new file mode 100644 index 0000000..e502fbe --- /dev/null +++ b/tests/fixtures/reference-javascript/src/app/shared.js @@ -0,0 +1,11 @@ +export const DEFAULT_LIMIT = 3; + +export function clamp(value, limit = DEFAULT_LIMIT) { + if (value < 0) { + return 0; + } + if (value > limit) { + return limit; + } + return value; +} diff --git a/tests/fixtures/reference-javascript/src/app/worker.cjs b/tests/fixtures/reference-javascript/src/app/worker.cjs new file mode 100644 index 0000000..30592c9 --- /dev/null +++ b/tests/fixtures/reference-javascript/src/app/worker.cjs @@ -0,0 +1,5 @@ +export { Service } from "./service.mjs"; + +export function execute(worker, value) { + return worker.run(value); +} diff --git a/tests/fixtures/reference-typescript/src/app/index.ts b/tests/fixtures/reference-typescript/src/app/index.ts new file mode 100644 index 0000000..30e3fb4 --- /dev/null +++ b/tests/fixtures/reference-typescript/src/app/index.ts @@ -0,0 +1,8 @@ +import { Service } from "./service"; +export type { Choice } from "./types"; + +throw new Error("The reference adapter must never execute project code"); + +export function applicationName(): string { + return Service.name; +} diff --git a/tests/fixtures/reference-typescript/src/app/service.ts b/tests/fixtures/reference-typescript/src/app/service.ts new file mode 100644 index 0000000..aec4304 --- /dev/null +++ b/tests/fixtures/reference-typescript/src/app/service.ts @@ -0,0 +1,17 @@ +import type { Choice } from "./types"; + +export class Service { + run(choice: Choice): number { + if (choice.enabled) { + return choice.value; + } + return 0; + } +} + +export const buildService = (choice: Choice): Service | null => { + if (choice.enabled) { + return new Service(); + } + return null; +}; diff --git a/tests/fixtures/reference-typescript/src/app/types.mts b/tests/fixtures/reference-typescript/src/app/types.mts new file mode 100644 index 0000000..9d09da8 --- /dev/null +++ b/tests/fixtures/reference-typescript/src/app/types.mts @@ -0,0 +1,6 @@ +export interface Choice { + enabled: boolean; + value: number; +} + +export const DEFAULT_LIMIT: number = 3; diff --git a/tests/fixtures/reference-typescript/src/app/worker.cts b/tests/fixtures/reference-typescript/src/app/worker.cts new file mode 100644 index 0000000..4fc33e1 --- /dev/null +++ b/tests/fixtures/reference-typescript/src/app/worker.cts @@ -0,0 +1,6 @@ +import { Service } from "./service"; +import type { Choice } from "./types"; + +export function execute(worker: Service, choice: Choice): number { + return worker.run(choice); +} diff --git a/tests/test_javascript_reference_adapter.py b/tests/test_javascript_reference_adapter.py new file mode 100644 index 0000000..7ce3d60 --- /dev/null +++ b/tests/test_javascript_reference_adapter.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import shutil +import tempfile +import unittest +from pathlib import Path +from typing import cast +from unittest import mock + +import docforge.adapters.javascript as javascript_adapter +from docforge.adapter_sdk import ( + AdapterProject, + AdapterSource, + AdapterSourceProjection, + verify_adapter_conformance, +) +from docforge.adapters.javascript import ( + JAVASCRIPT_ADAPTER_ID, + JAVASCRIPT_ADAPTER_VERSION, + JAVASCRIPT_EXTRACTOR_VERSION, + JavaScriptReferenceAdapter, +) +from docforge.errors import DocForgeError +from docforge.index import ProjectIndex + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" + + +class RecordingScriptAdapter(JavaScriptReferenceAdapter): + def __init__(self, root: Path, project_id: str) -> None: + super().__init__( + root, + source_roots=("src",), + project_id=project_id, + title=f"{project_id} fixture", + ) + self.extracted_paths: list[str] = [] + + def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: + self.extracted_paths.append(source.source_path) + return super().extract_source(source) + + +class JavaScriptReferenceAdapterTests(unittest.TestCase): + CASES = ( + ("reference-javascript", "javascript", 14, 14, 5, "src/app/shared.js"), + ("reference-typescript", "typescript", 13, 14, 4, "src/app/types.mts"), + ) + + def copy_fixture(self, parent: Path, fixture: str) -> Path: + root = parent / fixture + shutil.copytree(FIXTURES / fixture, root) + return root.resolve() + + @staticmethod + def build_metrics(result: dict[str, object]) -> dict[str, object]: + return cast(dict[str, object], result["build"]) + + def test_both_grammars_produce_deterministic_complete_logic_assemblies(self) -> None: + for fixture, language, nodes, edges, logic, _shared in self.CASES: + with self.subTest(language=language), tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory), fixture) + adapter = RecordingScriptAdapter(root, fixture) + + first = adapter.load_assembly() + second = adapter.load_complete_assembly() + projection = adapter.load_projection() + + self.assertEqual(first, second) + self.assertEqual(first.projection, projection) + self.assertEqual(JAVASCRIPT_ADAPTER_ID, projection.adapter_id) + self.assertEqual(JAVASCRIPT_ADAPTER_VERSION, projection.adapter_version) + self.assertEqual(nodes, len(projection.nodes)) + self.assertEqual(edges, len(projection.edges)) + self.assertEqual(logic, len(first.logic)) + self.assertEqual( + {language}, + {dict(item.metadata)["language"] for item in projection.nodes}, + ) + self.assertEqual( + {"contains", "depends_on"}, + {item.edge.relation for item in projection.edges}, + ) + self.assertTrue( + any( + item.node.title == "Service.run" + and dict(item.metadata)["kind"] == "function" + for item in projection.nodes + ) + ) + self.assertFalse((root / ".cache").exists()) + + report = adapter.support_report() + self.assertEqual(1, report["schema_version"]) + self.assertEqual(JAVASCRIPT_EXTRACTOR_VERSION, report["extractor_version"]) + self.assertFalse(report["imports_project_code"]) + self.assertFalse(report["executes_project_code"]) + self.assertEqual( + [ + "call_resolution", + "dynamic_module_resolution", + "inheritance_resolution", + "module_configuration_resolution", + "runtime_generated_facts", + "type_and_symbol_resolution", + ], + [item.code for item in adapter.unsupported_facts()], + ) + + conformance = verify_adapter_conformance( + adapter, + cache_root=root / ".cache" / "conformance", + ) + self.assertEqual(nodes, conformance.node_count) + self.assertEqual(edges, conformance.edge_count) + self.assertEqual(logic, conformance.logic_projection_count) + self.assertTrue(conformance.incremental) + + def test_cold_warm_reverse_dependency_and_exact_equivalence_for_both_grammars( + self, + ) -> None: + for fixture, language, nodes, edges, logic, shared_path in self.CASES: + with self.subTest(language=language), tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory), fixture) + adapter = RecordingScriptAdapter(root, fixture) + project = AdapterProject(adapter, cache_root=root / ".cache" / "incremental") + index = ProjectIndex(project) + + cold = index.build() + self.assertEqual(4, self.build_metrics(cold)["reparsed_sources"]) + self.assertEqual(4, len(adapter.extracted_paths)) + + adapter.extracted_paths.clear() + warm = index.build() + self.assertEqual(4, self.build_metrics(warm)["cache_hits"]) + self.assertEqual(0, self.build_metrics(warm)["reparsed_sources"]) + self.assertEqual([], adapter.extracted_paths) + + shared = root / shared_path + shared.write_text( + shared.read_text(encoding="utf-8") + "\n// changed dependency evidence\n", + encoding="utf-8", + ) + adapter.extracted_paths.clear() + changed = index.build() + self.assertEqual(4, self.build_metrics(changed)["invalidated_sources"]) + self.assertEqual(4, len(adapter.extracted_paths)) + + equivalent = project.verify_incremental_equivalence() + self.assertEqual("ok", equivalent["status"]) + self.assertEqual(nodes, equivalent["node_count"]) + self.assertEqual(edges, equivalent["edge_count"]) + self.assertEqual(logic, equivalent["logic_projection_count"]) + + def test_add_delete_and_corrupt_cache_recovery_for_both_grammars(self) -> None: + additions = { + "javascript": ( + "src/app/extra.js", + "export function extra() { return 'extra'; }\n", + ), + "typescript": ( + "src/app/extra.ts", + "export function extra(): string { return 'extra'; }\n", + ), + } + for fixture, language, _nodes, _edges, _logic, _shared_path in self.CASES: + with self.subTest(language=language), tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory), fixture) + adapter = RecordingScriptAdapter(root, fixture) + cache_root = root / ".cache" / "incremental" + project = AdapterProject(adapter, cache_root=cache_root) + index = ProjectIndex(project) + index.build() + + relative, content = additions[language] + added_path = root / relative + added_path.write_text(content, encoding="utf-8") + adapter.extracted_paths.clear() + added = index.build() + self.assertEqual(1, self.build_metrics(added)["reparsed_sources"]) + self.assertEqual([relative], adapter.extracted_paths) + self.assertIn(relative, {node.source_path for node in project.load().nodes}) + + added_path.unlink() + adapter.extracted_paths.clear() + deleted = index.build() + self.assertEqual(1, self.build_metrics(deleted)["deleted_sources"]) + self.assertNotIn(relative, {node.source_path for node in project.load().nodes}) + + (cache_root / "extractions.json").write_text("{broken", encoding="utf-8") + adapter.extracted_paths.clear() + recovered = index.build() + self.assertEqual(4, self.build_metrics(recovered)["reparsed_sources"]) + self.assertEqual(4, len(adapter.extracted_paths)) + self.assertEqual("ok", project.verify_incremental_equivalence()["status"]) + + def test_confinement_and_no_ast_reject_logic_for_both_grammars(self) -> None: + for fixture, language, _nodes, _edges, _logic, _shared_path in self.CASES: + with self.subTest(language=language), tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + root = self.copy_fixture(parent, fixture) + + for source_root in (root / "src", Path("../outside")): + with self.assertRaises(DocForgeError) as captured: + JavaScriptReferenceAdapter(root, source_roots=(source_root,)) + self.assertEqual("path_escape", captured.exception.code) + + outside = parent / "outside" + outside.mkdir() + linked_root = root / "linked" + linked_root.symlink_to(outside, target_is_directory=True) + with self.assertRaises(DocForgeError) as linked: + JavaScriptReferenceAdapter(root, source_roots=("linked",)) + self.assertEqual("path_escape", linked.exception.code) + + adapter = RecordingScriptAdapter(root, fixture) + project = AdapterProject(adapter, cache_root=root / ".cache" / "no-ast") + with self.assertRaises(DocForgeError) as no_ast: + ProjectIndex(project, allow_logic=False).build() + self.assertEqual("adapter_policy_forbids_logic", no_ast.exception.code) + self.assertFalse(project.descriptor.index_path.exists()) + + def test_missing_optional_frontends_report_exact_install_remediation(self) -> None: + for fixture, language, _nodes, _edges, _logic, _shared_path in self.CASES: + grammar_module = ( + "tree_sitter_javascript" if language == "javascript" else "tree_sitter_typescript" + ) + with self.subTest(language=language), tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory), fixture) + adapter = RecordingScriptAdapter(root, fixture) + missing = ModuleNotFoundError( + f"No module named {grammar_module!r}", + name=grammar_module, + ) + with ( + mock.patch.object( + javascript_adapter.importlib, + "import_module", + side_effect=missing, + ), + self.assertRaises(DocForgeError) as captured, + ): + adapter.load_manifest() + self.assertEqual("optional_dependency_missing", captured.exception.code) + self.assertEqual(language, captured.exception.details["extra"]) + self.assertEqual( + f"docforge[{language}]", + captured.exception.details["install"], + ) + self.assertEqual( + grammar_module, + captured.exception.details["missing_module"], + )