From 9cd7c4e424dab7cca98b1cc48c124a18252af9ea Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 14:52:05 -0400 Subject: [PATCH] Keep reference manifests parser-free --- src/docforge/adapters/javascript.py | 227 +++++++++++++++++---- src/docforge/adapters/python.py | 170 ++++++++++++--- tests/test_javascript_reference_adapter.py | 19 +- tests/test_python_reference_adapter.py | 19 +- 4 files changed, 368 insertions(+), 67 deletions(-) diff --git a/src/docforge/adapters/javascript.py b/src/docforge/adapters/javascript.py index 82f733f..37f9238 100644 --- a/src/docforge/adapters/javascript.py +++ b/src/docforge/adapters/javascript.py @@ -136,7 +136,7 @@ class _SourceRecord: fingerprint: str text: str raw: bytes - root: _TreeNode + root: _TreeNode | None @dataclass(frozen=True) @@ -229,7 +229,7 @@ class JavaScriptReferenceAdapter: return _UNSUPPORTED_FACTS def load_manifest(self) -> AdapterManifest: - records = self._inventory() + records = self._inventory(parse=False) path_sources = {record.source_path: record.source_id for record in records} sources = tuple( sorted( @@ -390,7 +390,7 @@ class JavaScriptReferenceAdapter: ) return result - def _inventory(self) -> tuple[_SourceRecord, ...]: + def _inventory(self, *, parse: bool = True) -> tuple[_SourceRecord, ...]: paths: list[str] = [] for source_root in self.source_roots: absolute_root = self.root.joinpath(*PurePosixPath(source_root).parts) @@ -439,7 +439,7 @@ class JavaScriptReferenceAdapter: "Script source inventory exceeds the configured limit", maximum=self.max_sources, ) - records = tuple(self._read_source(path) for path in sorted(paths)) + records = tuple(self._read_source(path, parse=parse) 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) @@ -450,7 +450,7 @@ class JavaScriptReferenceAdapter: ) return records - def _read_source(self, relative: str) -> _SourceRecord: + def _read_source(self, relative: str, *, parse: bool = True) -> _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") @@ -543,14 +543,17 @@ class JavaScriptReferenceAdapter: 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, - ) + parse_raw = text.encode("utf-8") + root: _TreeNode | None = None + if parse: + 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, @@ -682,31 +685,170 @@ class JavaScriptReferenceAdapter: @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) + result = { + value + for value in cls._lexical_static_specifiers(record.text) + if value.startswith(("./", "../")) + } 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] + def _script_tokens(source: str) -> tuple[tuple[str, str], ...]: + """Tokenize only enough JavaScript syntax to inventory static module specifiers.""" + + tokens: list[tuple[str, str]] = [] + position = 0 + length = len(source) + while position < length: + character = source[position] + if character in " \t\f\v": + position += 1 + continue + if character in "\r\n": + if character == "\r" and position + 1 < length and source[position + 1] == "\n": + position += 1 + tokens.append(("newline", "\n")) + position += 1 + continue + if source.startswith("//", position): + position += 2 + while position < length and source[position] not in "\r\n": + position += 1 + continue + if source.startswith("/*", position): + position += 2 + while position < length and not source.startswith("*/", position): + if source[position] in "\r\n": + tokens.append(("newline", "\n")) + if ( + source[position] == "\r" + and position + 1 < length + and source[position + 1] == "\n" + ): + position += 1 + position += 1 + position = min(length, position + 2) + continue + if character in {'"', "'"}: + quote = character + position += 1 + value: list[str] = [] + escaped = False + closed = False + while position < length: + current = source[position] + if current == "\\": + escaped = True + position += 2 + continue + if current == quote: + position += 1 + closed = True + break + if current in "\r\n": + break + value.append(current) + position += 1 + tokens.append(("string" if closed and not escaped else "opaque", "".join(value))) + continue + if character == "`": + position += 1 + while position < length: + current = source[position] + if current == "\\": + position += 2 + continue + position += 1 + if current == "`": + break + tokens.append(("opaque", "template")) + continue + if character.isalpha() or character in {"_", "$"}: + start = position + position += 1 + while position < length and ( + source[position].isalnum() or source[position] in {"_", "$"} + ): + position += 1 + tokens.append(("name", source[start:position])) + continue + tokens.append(("punct", character)) + position += 1 + return tuple(tokens) + + @classmethod + def _lexical_static_specifiers(cls, source: str) -> tuple[str, ...]: + tokens = cls._script_tokens(source) + result: set[str] = set() + for position, (kind, value) in enumerate(tokens): + if kind != "name" or value not in {"import", "export"}: + continue + if position and tokens[position - 1] == ("punct", "."): + continue + keyword = value + index = position + 1 + depth = 0 + clause_seen = False + line_break = False + while index < len(tokens): + token_kind, token_value = tokens[index] + if token_kind == "newline" and depth == 0: + line_break = True + if keyword == "export" and not clause_seen: + break + index += 1 + continue + if token_kind == "punct": + if token_value in "([{": + depth += 1 + clause_seen = True + elif token_value in ")]}": + depth = max(0, depth - 1) + clause_seen = True + elif token_value == ";" and depth == 0: + break + elif token_value in {"*", ","}: + clause_seen = True + elif token_kind == "string": + if keyword == "import" and not clause_seen: + result.add(token_value) + break + elif token_kind == "name": + if line_break and token_value in { + "class", + "const", + "export", + "function", + "import", + "let", + "return", + "throw", + "var", + }: + break + if keyword == "export" and token_value in { + "async", + "class", + "const", + "default", + "function", + "let", + "var", + }: + break + if token_value == "from" and clause_seen: + source_index = index + 1 + while source_index < len(tokens) and tokens[source_index][0] == "newline": + source_index += 1 + if source_index < len(tokens) and tokens[source_index][0] == "string": + result.add(tokens[source_index][1]) + break + if keyword == "import" and token_value not in {"type"}: + clause_seen = True + if token_kind != "newline": + line_break = False + index += 1 + return tuple(sorted(result)) @staticmethod def _resolve_relative_specifier( @@ -733,6 +875,12 @@ class JavaScriptReferenceAdapter: record: _SourceRecord, source: AdapterSource, ) -> AdapterSourceProjection: + if record.root is None: + raise DocForgeError( + "invalid_adapter", + "Script extraction requires a parsed source record", + source=record.source_path, + ) file_id = self._file_node_id(source.source_id) module_id = self._module_node_id(source.source_id) nodes: list[AdapterNode] = [ @@ -839,6 +987,13 @@ class JavaScriptReferenceAdapter: ) def _definitions(self, record: _SourceRecord) -> tuple[_Definition, ...]: + root = record.root + if root is None: + raise DocForgeError( + "invalid_adapter", + "Script definition extraction requires a parsed source record", + source=record.source_path, + ) definitions: list[_Definition] = [] seen: set[str] = set() @@ -859,7 +1014,7 @@ class JavaScriptReferenceAdapter: for child in node.named_children: visit(child, next_scopes) - visit(record.root, ()) + visit(root, ()) ordered = tuple(sorted(definitions, key=lambda item: item.node_id)) public_functions = { (function.qualified_name, function.line) diff --git a/src/docforge/adapters/python.py b/src/docforge/adapters/python.py index ef4349c..6a754d6 100644 --- a/src/docforge/adapters/python.py +++ b/src/docforge/adapters/python.py @@ -91,7 +91,7 @@ class _SourceRecord: fingerprint: str raw: bytes text: str - tree: ast.Module + tree: ast.Module | None dependencies: tuple[str, ...] = () @@ -154,7 +154,7 @@ class PythonReferenceAdapter: return _UNSUPPORTED_FACTS def load_manifest(self) -> AdapterManifest: - records = self._inventory() + records = self._inventory(parse=False) module_sources = {record.module_name: record.source_id for record in records} if len(module_sources) != len(records): raise DocForgeError( @@ -325,7 +325,7 @@ class PythonReferenceAdapter: ) return result - def _inventory(self) -> tuple[_SourceRecord, ...]: + def _inventory(self, *, parse: bool = True) -> tuple[_SourceRecord, ...]: paths: list[str] = [] for source_root in self.source_roots: absolute_root = self.root.joinpath(*PurePosixPath(source_root).parts) @@ -374,7 +374,7 @@ class PythonReferenceAdapter: "Python source inventory exceeds the configured limit", maximum=self.max_sources, ) - records = tuple(self._read_source(path) for path in sorted(paths)) + records = tuple(self._read_source(path, parse=parse) 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) @@ -385,7 +385,7 @@ class PythonReferenceAdapter: ) return records - def _read_source(self, relative: str) -> _SourceRecord: + def _read_source(self, relative: str, *, parse: bool = True) -> _SourceRecord: path = PurePosixPath(relative) if path.is_absolute() or ".." in path.parts or path.suffix != ".py": raise DocForgeError("path_escape", "Python source path is unsafe") @@ -478,15 +478,17 @@ class PythonReferenceAdapter: source=relative, ) text = self._decode_source(raw, relative) - try: - tree = ast.parse(text, filename=relative, type_comments=True) - except SyntaxError as error: - raise DocForgeError( - "invalid_python_source", - "Python source cannot be parsed by the reference adapter", - source=relative, - line=error.lineno, - ) from error + tree: ast.Module | None = None + if parse: + try: + tree = ast.parse(text, filename=relative, type_comments=True) + except SyntaxError as error: + raise DocForgeError( + "invalid_python_source", + "Python source cannot be parsed by the reference adapter", + source=relative, + line=error.lineno, + ) from error module_name = self._module_name(path, PurePosixPath(source_root)) return _SourceRecord( source_id=self._source_id(relative), @@ -585,24 +587,127 @@ class PythonReferenceAdapter: if record.source_path.endswith("/__init__.py") else record.module_name.rpartition(".")[0] ) - for node in ast.walk(record.tree): + for level, module, imported_names in self._lexical_imports(record): candidates: list[str] = [] - if isinstance(node, ast.Import): - candidates.extend(alias.name for alias in node.names) - elif isinstance(node, ast.ImportFrom): - base = self._import_from_base(package, node.level, node.module) + if not imported_names and level == 0 and module is not None: + candidates.append(module) + else: + base = self._import_from_base(package, level, module) if base: candidates.append(base) - if node.module is None and base: - candidates.extend( - f"{base}.{alias.name}" for alias in node.names if alias.name != "*" - ) + candidates.extend(f"{base}.{name}" for name in imported_names if name != "*") for module_name in candidates: target = module_sources.get(module_name) if target is not None and target != record.source_id: dependencies.add(target) return tuple(sorted(dependencies)) + @staticmethod + def _lexical_imports( + record: _SourceRecord, + ) -> tuple[tuple[int, str | None, tuple[str, ...]], ...]: + """Inventory import dependencies without constructing a Python AST.""" + + try: + tokens = tuple(tokenize.generate_tokens(io.StringIO(record.text).readline)) + except (IndentationError, tokenize.TokenError) as error: + line = error.args[1][0] if len(error.args) > 1 else None + raise DocForgeError( + "invalid_python_source", + "Python source cannot be tokenized by the reference adapter", + source=record.source_path, + line=line, + ) from error + + statements: list[list[tokenize.TokenInfo]] = [] + current: list[tokenize.TokenInfo] = [] + nesting = 0 + ignored = { + tokenize.ENCODING, + tokenize.INDENT, + tokenize.DEDENT, + tokenize.NL, + tokenize.COMMENT, + } + for token in tokens: + if token.type in ignored: + continue + if token.type == tokenize.OP: + if token.string in "([{": + nesting += 1 + elif token.string in ")]}": + nesting = max(0, nesting - 1) + elif token.string == ";" and nesting == 0: + if current: + statements.append(current) + current = [] + continue + if token.type in {tokenize.NEWLINE, tokenize.ENDMARKER} and nesting == 0: + if current: + statements.append(current) + current = [] + continue + current.append(token) + + imports: list[tuple[int, str | None, tuple[str, ...]]] = [] + for statement in statements: + for position, token in enumerate(statement): + if token.type != tokenize.NAME or token.string not in {"import", "from"}: + continue + if token.string == "import": + names = PythonReferenceAdapter._imported_names(statement[position + 1 :]) + imports.extend((0, name, ()) for name in names) + break + import_position = next( + ( + index + for index in range(position + 1, len(statement)) + if statement[index].type == tokenize.NAME + and statement[index].string == "import" + ), + None, + ) + if import_position is None: + break + prefix = statement[position + 1 : import_position] + level = 0 + for item in prefix: + if item.type == tokenize.NAME: + break + if item.type == tokenize.OP and set(item.string) == {"."}: + level += len(item.string) + module_parts = [item.string for item in prefix if item.type == tokenize.NAME] + module = ".".join(module_parts) or None + names = PythonReferenceAdapter._imported_names(statement[import_position + 1 :]) + imports.append((level, module, names)) + break + return tuple(imports) + + @staticmethod + def _imported_names(tokens: Sequence[tokenize.TokenInfo]) -> tuple[str, ...]: + names: list[str] = [] + current: list[str] = [] + skip_alias = False + for token in tokens: + if token.type == tokenize.NAME and token.string == "as": + skip_alias = True + continue + if token.type == tokenize.OP and token.string == ",": + if current: + names.append(".".join(current)) + current = [] + skip_alias = False + continue + if token.type == tokenize.OP and token.string == "*": + if not skip_alias: + current.append("*") + continue + if token.type == tokenize.NAME and not skip_alias: + current.append(token.string) + if current: + names.append(".".join(current)) + return tuple(name for name in names if name) + @staticmethod def _import_from_base(package: str, level: int, module: str | None) -> str: if level == 0: @@ -621,6 +726,13 @@ class PythonReferenceAdapter: record: _SourceRecord, source: AdapterSource, ) -> AdapterSourceProjection: + tree = record.tree + if tree is None: + raise DocForgeError( + "invalid_adapter", + "Python extraction requires a parsed source record", + source=record.source_path, + ) file_id = self._file_node_id(source.source_id) module_id = self._module_node_id(source.source_id) nodes: list[AdapterNode] = [ @@ -639,8 +751,7 @@ class PythonReferenceAdapter: kind="module", qualified_name=record.module_name, content=( - ast.get_docstring(record.tree, clean=False) - or f"Python module {record.module_name}." + ast.get_docstring(tree, clean=False) or f"Python module {record.module_name}." ), source_path=record.source_path, anchor="L1", @@ -721,6 +832,13 @@ class PythonReferenceAdapter: record: _SourceRecord, module_node_id: str, ) -> tuple[_Definition, ...]: + tree = record.tree + if tree is None: + raise DocForgeError( + "invalid_adapter", + "Python definition extraction requires a parsed source record", + source=record.source_path, + ) definitions: list[_Definition] = [] qualified_names: set[str] = set() adapter = self @@ -773,7 +891,7 @@ class PythonReferenceAdapter: def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: self._definition("function", node) - Collector().visit(record.tree) + Collector().visit(tree) return tuple(sorted(definitions, key=lambda item: item.node_id)) @staticmethod diff --git a/tests/test_javascript_reference_adapter.py b/tests/test_javascript_reference_adapter.py index 7ce3d60..14c5b96 100644 --- a/tests/test_javascript_reference_adapter.py +++ b/tests/test_javascript_reference_adapter.py @@ -132,10 +132,16 @@ class JavaScriptReferenceAdapterTests(unittest.TestCase): self.assertEqual(4, len(adapter.extracted_paths)) adapter.extracted_paths.clear() - warm = index.build() + with mock.patch.object( + adapter, + "_parse", + wraps=adapter._parse, # pyright: ignore[reportPrivateUsage] + ) as parsed: + 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) + parsed.assert_not_called() shared = root / shared_path shared.write_text( @@ -143,9 +149,15 @@ class JavaScriptReferenceAdapterTests(unittest.TestCase): encoding="utf-8", ) adapter.extracted_paths.clear() - changed = index.build() + with mock.patch.object( + adapter, + "_parse", + wraps=adapter._parse, # pyright: ignore[reportPrivateUsage] + ) as parsed: + changed = index.build() self.assertEqual(4, self.build_metrics(changed)["invalidated_sources"]) self.assertEqual(4, len(adapter.extracted_paths)) + self.assertEqual(4, parsed.call_count) equivalent = project.verify_incremental_equivalence() self.assertEqual("ok", equivalent["status"]) @@ -241,7 +253,8 @@ class JavaScriptReferenceAdapterTests(unittest.TestCase): ), self.assertRaises(DocForgeError) as captured, ): - adapter.load_manifest() + source = adapter.load_manifest().sources[0] + adapter.extract_source(source) self.assertEqual("optional_dependency_missing", captured.exception.code) self.assertEqual(language, captured.exception.details["extra"]) self.assertEqual( diff --git a/tests/test_python_reference_adapter.py b/tests/test_python_reference_adapter.py index 8887458..0755321 100644 --- a/tests/test_python_reference_adapter.py +++ b/tests/test_python_reference_adapter.py @@ -5,7 +5,9 @@ import tempfile import unittest from pathlib import Path from typing import cast +from unittest import mock +import docforge.adapters.python as python_adapter from docforge.adapter_sdk import ( AdapterProject, AdapterSource, @@ -137,10 +139,17 @@ class PythonReferenceAdapterTests(unittest.TestCase): ) adapter.extracted_paths.clear() - warm = index.build() + original_parse = python_adapter.ast.parse + with mock.patch.object( + python_adapter.ast, + "parse", + wraps=original_parse, + ) as parsed: + 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) + parsed.assert_not_called() shared = root / "src" / "sample" / "shared.py" shared.write_text( @@ -150,8 +159,14 @@ class PythonReferenceAdapterTests(unittest.TestCase): encoding="utf-8", ) adapter.extracted_paths.clear() - changed = index.build() + with mock.patch.object( + python_adapter.ast, + "parse", + wraps=original_parse, + ) as parsed: + changed = index.build() self.assertEqual(4, self.build_metrics(changed)["invalidated_sources"]) + self.assertGreater(parsed.call_count, 0) self.assertEqual( [ "src/sample/__init__.py",