diff --git a/schemas/reference-adapter.schema.json b/schemas/reference-adapter.schema.json new file mode 100644 index 0000000..bc61277 --- /dev/null +++ b/schemas/reference-adapter.schema.json @@ -0,0 +1,78 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://andraxion.net/docforge/schemas/reference-adapter.schema.json", + "title": "DocForge Reference Adapter Configuration", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "project_id", + "title", + "language", + "source_roots" + ], + "properties": { + "schema_version": { + "const": 1 + }, + "project_id": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9._-]{1,127}$" + }, + "title": { + "type": "string", + "minLength": 1, + "maxLength": 256 + }, + "language": { + "enum": [ + "python", + "javascript", + "typescript", + "cpp" + ] + }, + "source_roots": { + "type": "array", + "minItems": 1, + "maxItems": 64, + "uniqueItems": true, + "items": { + "type": "string", + "minLength": 1, + "pattern": "^(?!/)(?!.*(?:^|/)\\.\\.(?:/|$)).+$" + } + }, + "compilation_database": { + "type": "string", + "minLength": 1, + "pattern": "^(?!/)(?!.*(?:^|/)\\.\\.(?:/|$)).+$" + } + }, + "allOf": [ + { + "if": { + "properties": { + "language": { + "const": "cpp" + } + }, + "required": [ + "language" + ] + }, + "then": { + "required": [ + "compilation_database" + ] + }, + "else": { + "not": { + "required": [ + "compilation_database" + ] + } + } + } + ] +} diff --git a/src/docforge/reference_config.py b/src/docforge/reference_config.py new file mode 100644 index 0000000..c6557a2 --- /dev/null +++ b/src/docforge/reference_config.py @@ -0,0 +1,428 @@ +"""Strict, data-only configuration for the runnable reference adapters.""" + +from __future__ import annotations + +import hashlib +import os +import stat +import tomllib +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Literal, cast + +from .config_validation import ID_PATTERN +from .errors import DocForgeError + +REFERENCE_ADAPTER_SCHEMA_VERSION = 1 +REFERENCE_ADAPTER_CONFIG = PurePosixPath(".docforge/reference-adapter.toml") +MAX_REFERENCE_CONFIG_BYTES = 65_536 +MAX_REFERENCE_INVENTORY_ENTRIES = 65_536 + +ReferenceLanguage = Literal["python", "javascript", "typescript", "cpp"] +REFERENCE_LANGUAGES: tuple[ReferenceLanguage, ...] = ( + "python", + "javascript", + "typescript", + "cpp", +) + +_CONFIG_FIELDS = frozenset( + { + "schema_version", + "project_id", + "title", + "language", + "source_roots", + "compilation_database", + } +) +_LANGUAGE_SUFFIXES: dict[ReferenceLanguage, frozenset[str]] = { + "python": frozenset({".py"}), + "javascript": frozenset({".cjs", ".js", ".mjs"}), + "typescript": frozenset({".cts", ".mts", ".ts"}), + "cpp": frozenset({".c", ".cc", ".cpp", ".cxx", ".h", ".hh", ".hpp", ".hxx"}), +} +_KNOWN_SOURCE_SUFFIXES = frozenset( + suffix for suffixes in _LANGUAGE_SUFFIXES.values() for suffix in suffixes +) + +__all__ = [ + "MAX_REFERENCE_CONFIG_BYTES", + "MAX_REFERENCE_INVENTORY_ENTRIES", + "REFERENCE_ADAPTER_CONFIG", + "REFERENCE_ADAPTER_SCHEMA_VERSION", + "REFERENCE_LANGUAGES", + "ReferenceAdapterConfigV1", + "ReferenceLanguage", + "load_reference_adapter_config", +] + + +@dataclass(frozen=True) +class ReferenceAdapterConfigV1: + """One immutable, project-confined reference adapter selection.""" + + schema_version: Literal[1] + project_id: str + title: str + language: ReferenceLanguage + source_roots: tuple[Path, ...] + compilation_database: Path | None + project_root: Path + config_path: Path + config_hash: str + + def as_dict(self) -> dict[str, object]: + document: dict[str, object] = { + "schema_version": self.schema_version, + "project_id": self.project_id, + "title": self.title, + "language": self.language, + "source_roots": [ + path.relative_to(self.project_root).as_posix() for path in self.source_roots + ], + } + if self.compilation_database is not None: + document["compilation_database"] = self.compilation_database.relative_to( + self.project_root + ).as_posix() + return document + + +def load_reference_adapter_config(project_root: str | Path) -> ReferenceAdapterConfigV1: + """Load the fixed reference-adapter descriptor without executing project content.""" + + root = _project_root(project_root) + config_path = root.joinpath(*REFERENCE_ADAPTER_CONFIG.parts) + try: + resolved_config = config_path.resolve(strict=True) + except (OSError, ValueError) as error: + raise DocForgeError( + "missing_path", + "Reference adapter configuration does not exist", + path=str(config_path), + ) from error + if resolved_config != config_path or not resolved_config.is_relative_to(root): + raise DocForgeError( + "path_escape", + "Reference adapter configuration must not traverse symbolic links", + ) + raw = _read_regular_file( + config_path, + field="reference adapter configuration", + maximum_bytes=MAX_REFERENCE_CONFIG_BYTES, + ) + try: + document = cast(dict[str, object], tomllib.loads(raw.decode("utf-8"))) + except UnicodeDecodeError as error: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter configuration must be UTF-8", + ) from error + except tomllib.TOMLDecodeError as error: + raise DocForgeError( + "invalid_reference_config", + f"Invalid reference adapter configuration: {error}", + ) from error + + unknown = sorted(set(document) - _CONFIG_FIELDS) + if unknown: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter configuration has unknown fields", + fields=unknown, + ) + if document.get("schema_version") != REFERENCE_ADAPTER_SCHEMA_VERSION: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter schema_version must be 1", + ) + project_id = _string(document, "project_id") + if ID_PATTERN.fullmatch(project_id) is None: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter project_id is not a stable ID", + project_id=project_id, + ) + title = _string(document, "title") + if len(title) > 256: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter title exceeds its size limit", + maximum_characters=256, + ) + raw_language = _string(document, "language") + if raw_language not in REFERENCE_LANGUAGES: + raise DocForgeError( + "unsupported_reference_language", + "Reference adapter language is not supported", + language=raw_language, + supported=list(REFERENCE_LANGUAGES), + ) + language = raw_language + source_roots = _source_roots(root, document.get("source_roots")) + + raw_compilation_database = document.get("compilation_database") + compilation_database: Path | None = None + if language == "cpp": + if raw_compilation_database is None: + raise DocForgeError( + "invalid_reference_config", + "C++ reference adapters require compilation_database", + ) + compilation_database = _confined_path( + root, + raw_compilation_database, + field="compilation_database", + expected="file", + ) + elif raw_compilation_database is not None: + raise DocForgeError( + "invalid_reference_config", + "compilation_database is allowed only for the C++ reference adapter", + language=language, + ) + + _validate_language_inventory(source_roots, language) + return ReferenceAdapterConfigV1( + schema_version=1, + project_id=project_id, + title=title, + language=language, + source_roots=source_roots, + compilation_database=compilation_database, + project_root=root, + config_path=config_path, + config_hash=hashlib.sha256(raw).hexdigest(), + ) + + +def _project_root(raw: str | Path) -> Path: + path = Path(raw) + if path.is_symlink(): + raise DocForgeError("path_escape", "Reference project root must not be a symlink") + try: + root = path.resolve(strict=True) + except (OSError, ValueError) as error: + raise DocForgeError( + "missing_path", + "Reference project root does not exist", + path=str(path), + ) from error + if not root.is_dir(): + raise DocForgeError("invalid_path", "Reference project root must be a directory") + return root + + +def _read_regular_file(path: Path, *, field: str, maximum_bytes: int) -> bytes: + try: + before = path.lstat() + except OSError as error: + raise DocForgeError("missing_path", f"{field} does not exist", path=str(path)) from error + if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode): + raise DocForgeError("invalid_path", f"{field} must be a regular non-symlink file") + if before.st_size > maximum_bytes: + raise DocForgeError( + "invalid_reference_config", + f"{field} exceeds its size limit", + maximum_bytes=maximum_bytes, + ) + + def identity(value: os.stat_result) -> tuple[int, int, int, int, int, int]: + return ( + value.st_dev, + value.st_ino, + value.st_mode, + value.st_size, + value.st_mtime_ns, + value.st_ctime_ns, + ) + + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as error: + raise DocForgeError( + "reference_config_changed", + f"{field} changed while it was opened", + ) from error + try: + opened = os.fstat(descriptor) + chunks: list[bytes] = [] + remaining = maximum_bytes + 1 + while remaining: + chunk = os.read(descriptor, min(remaining, 65_536)) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + raw = b"".join(chunks) + opened_after = os.fstat(descriptor) + except OSError as error: + raise DocForgeError( + "reference_config_changed", + f"{field} changed while it was read", + ) from error + finally: + os.close(descriptor) + try: + after = path.lstat() + except OSError as error: + raise DocForgeError( + "reference_config_changed", + f"{field} changed while it was read", + ) from error + + if ( + identity(before) != identity(opened) + or identity(opened) != identity(opened_after) + or identity(opened_after) != identity(after) + or len(raw) != opened_after.st_size + or len(raw) > maximum_bytes + ): + raise DocForgeError( + "reference_config_changed", + f"{field} changed while it was read", + ) + return raw + + +def _string(document: dict[str, object], field: str) -> str: + value = document.get(field) + if not isinstance(value, str) or not value or value != value.strip(): + raise DocForgeError( + "invalid_reference_config", + f"Reference adapter {field} must be a non-empty normalized string", + ) + return value + + +def _source_roots(root: Path, raw: object) -> tuple[Path, ...]: + if not isinstance(raw, list) or not raw: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter source_roots must be a non-empty string list", + ) + values = cast(list[object], raw) + if len(values) > 64: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter source_roots exceeds its item limit", + maximum_items=64, + ) + paths = tuple( + _confined_path(root, value, field="source_roots", expected="directory") for value in values + ) + if len(paths) != len(set(paths)): + raise DocForgeError( + "invalid_reference_config", + "Reference adapter source_roots contains duplicates", + ) + ordered = tuple(sorted(paths)) + if paths != ordered: + raise DocForgeError( + "invalid_reference_config", + "Reference adapter source_roots must be sorted", + ) + private_root = root / ".docforge" + if any( + path == private_root + or path.is_relative_to(private_root) + or any( + path.is_relative_to(other) or other.is_relative_to(path) + for other in ordered + if other != path + ) + for path in ordered + ): + raise DocForgeError( + "invalid_reference_config", + "Reference adapter source_roots must be non-overlapping and outside .docforge", + ) + return ordered + + +def _confined_path(root: Path, raw: object, *, field: str, expected: str) -> Path: + if not isinstance(raw, str) or not raw or raw != raw.strip(): + raise DocForgeError( + "invalid_reference_config", + f"{field} must be a non-empty normalized relative path", + ) + relative = Path(raw) + if relative.is_absolute() or ".." in relative.parts or relative == Path("."): + raise DocForgeError("path_escape", f"{field} must stay inside the project root", path=raw) + candidate = root / relative + if candidate.is_symlink(): + raise DocForgeError("path_escape", f"{field} must not be a symlink", path=raw) + try: + resolved = candidate.resolve(strict=True) + except (OSError, ValueError) as error: + raise DocForgeError("missing_path", f"{field} does not exist", path=raw) from error + if resolved != candidate or not resolved.is_relative_to(root): + raise DocForgeError( + "path_escape", + f"{field} must not traverse symlinks or leave the project root", + path=raw, + ) + if expected == "directory" and not resolved.is_dir(): + raise DocForgeError("invalid_path", f"{field} must identify a directory", path=raw) + if expected == "file" and (not resolved.is_file() or resolved.is_symlink()): + raise DocForgeError( + "invalid_path", + f"{field} must identify a regular non-symlink file", + path=raw, + ) + return resolved + + +def _validate_language_inventory( + source_roots: tuple[Path, ...], + language: ReferenceLanguage, +) -> None: + expected = _LANGUAGE_SUFFIXES[language] + selected = 0 + examined = 0 + other_languages: set[str] = set() + for source_root in source_roots: + for directory, names, files in os.walk( + source_root, + followlinks=False, + onerror=_inventory_error, + ): + base = Path(directory) + for name in (*names, *files): + examined += 1 + if examined > MAX_REFERENCE_INVENTORY_ENTRIES: + raise DocForgeError( + "reference_inventory_too_large", + "Reference source inventory exceeds its entry limit", + maximum_entries=MAX_REFERENCE_INVENTORY_ENTRIES, + ) + candidate = base / name + if candidate.is_symlink(): + raise DocForgeError( + "path_escape", + "Reference source roots must not contain symlinks", + path=candidate.as_posix(), + ) + for name in files: + suffix = Path(name).suffix + if suffix in expected: + selected += 1 + elif suffix in _KNOWN_SOURCE_SUFFIXES: + other_languages.add(suffix) + if selected == 0: + raise DocForgeError( + "reference_language_mismatch", + "Declared source roots contain no sources for the selected language", + language=language, + observed_suffixes=sorted(other_languages), + ) + + +def _inventory_error(error: OSError) -> None: + raise DocForgeError( + "invalid_path", + "Reference source inventory cannot be read safely", + path=error.filename, + ) from error diff --git a/src/docforge/reference_mcp.py b/src/docforge/reference_mcp.py new file mode 100644 index 0000000..bf4a76e --- /dev/null +++ b/src/docforge/reference_mcp.py @@ -0,0 +1,196 @@ +"""Runnable, read-only MCP binding for fixed in-repository reference adapters.""" + +from __future__ import annotations + +import argparse +import importlib +from pathlib import Path +from typing import cast + +from mcp.server.fastmcp import FastMCP + +from .adapter_sdk import ( + AdapterImplementation, + AdapterLoader, + AdapterProject, + AdapterProjectSettings, +) +from .errors import DocForgeError +from .mcp_server import create_read_only_server +from .reference_config import ReferenceAdapterConfigV1, load_reference_adapter_config + +REFERENCE_MCP_MODULE = "docforge.reference_mcp" +_FIXED_PROVIDERS = { + "python": ("docforge.adapters.python", "PythonReferenceAdapter"), + "javascript": ("docforge.adapters.javascript", "JavaScriptReferenceAdapter"), + "typescript": ("docforge.adapters.javascript", "JavaScriptReferenceAdapter"), + "cpp": ("docforge.adapters.cpp", "CppReferenceAdapter"), +} + +__all__ = [ + "REFERENCE_MCP_MODULE", + "create_reference_project", + "create_reference_server", + "main", +] + + +def create_reference_project(project_root: str | Path) -> AdapterProject: + """Construct one configured adapter project from fixed internal providers.""" + + config = load_reference_adapter_config(project_root) + return _project_from_config(config) + + +def create_reference_server( + project_root: str | Path, + *, + no_ast: bool = False, + diagnostics: bool = False, + capability_mode: str | None = None, + manual_projection_policy: str | None = None, + portable_graph_policy: str | None = None, + live_viewer_policy: str | None = None, +) -> FastMCP: + """Create the fixed read-only MCP surface for one reference project.""" + + config = load_reference_adapter_config(project_root) + project = _project_from_config(config) + return create_read_only_server( + project, + binding_metadata={ + "server_module": REFERENCE_MCP_MODULE, + "adapter_mode": "reference", + "reference_language": config.language, + "reference_config_hash": config.config_hash, + }, + no_ast=no_ast, + diagnostics=diagnostics, + capability_mode=capability_mode, + manual_projection_policy=manual_projection_policy, + portable_graph_policy=portable_graph_policy, + live_viewer_policy=live_viewer_policy, + ) + + +def _project_from_config(config: ReferenceAdapterConfigV1) -> AdapterProject: + loader = _adapter_loader(config) + cache_root = config.project_root / ".docforge" / "cache" / "reference-adapter" / config.language + return AdapterProject( + loader, + cache_root=cache_root, + settings=AdapterProjectSettings( + descriptor_path=config.config_path, + implementation=AdapterImplementation(files=(config.config_path,)), + ), + ) + + +def _adapter_loader(config: ReferenceAdapterConfigV1) -> AdapterLoader: + module_name, class_name = _FIXED_PROVIDERS[config.language] + try: + module = importlib.import_module(module_name) + except ModuleNotFoundError as error: + if error.name != module_name: + raise DocForgeError( + "optional_dependency_missing", + "The selected reference adapter dependency is not installed", + language=config.language, + missing_module=error.name, + install=_install_target(config.language), + ) from error + raise DocForgeError( + "reference_adapter_unavailable", + "The selected fixed reference adapter is not available", + language=config.language, + ) from error + constructor = getattr(module, class_name, None) + if not callable(constructor): + raise DocForgeError( + "reference_adapter_unavailable", + "The selected fixed reference adapter is not available", + language=config.language, + ) + arguments: dict[str, object] = { + "source_roots": tuple( + path.relative_to(config.project_root).as_posix() for path in config.source_roots + ), + "project_id": config.project_id, + "title": config.title, + } + if config.language == "cpp": + if config.compilation_database is None: + raise DocForgeError( + "invalid_reference_config", + "C++ reference adapters require compilation_database", + ) + arguments["compilation_database"] = config.compilation_database.relative_to( + config.project_root + ).as_posix() + loader = constructor(config.project_root, **arguments) + if not callable(getattr(loader, "load_projection", None)): + raise DocForgeError( + "invalid_adapter", + "The selected fixed provider does not implement the adapter contract", + language=config.language, + ) + return cast(AdapterLoader, loader) + + +def _install_target(language: str) -> str: + if language in {"javascript", "typescript"}: + return f"docforge[{language}]" + if language == "cpp": + return "docforge[cpp]" + return "docforge" + + +def main() -> None: + """Run the fixed reference binding over stdio.""" + + parser = argparse.ArgumentParser(prog="python -m docforge.reference_mcp") + parser.add_argument("--project-root", type=Path, required=True) + parser.add_argument( + "--no-ast", + action="store_true", + help=( + "Preserve the existing adapter and forbid AST, Tree-sitter, compiler-AST, " + "and function-Logic extraction changes" + ), + ) + parser.add_argument( + "--diagnostics", + action="store_true", + help="Attach bounded request-local stage timings and counters", + ) + parser.add_argument( + "--capability-mode", + choices=("read",), + help="Expose the fixed read-only project-bound capability surface", + ) + parser.add_argument( + "--manual-render-policy", + choices=("auto", "explicit", "disabled"), + ) + parser.add_argument( + "--portable-graph-policy", + choices=("explicit", "disabled"), + ) + parser.add_argument( + "--live-viewer-policy", + choices=("on-demand", "disabled"), + ) + arguments = parser.parse_args() + create_reference_server( + arguments.project_root, + no_ast=arguments.no_ast, + diagnostics=arguments.diagnostics, + capability_mode=arguments.capability_mode, + manual_projection_policy=arguments.manual_render_policy, + portable_graph_policy=arguments.portable_graph_policy, + live_viewer_policy=arguments.live_viewer_policy, + ).run(transport="stdio") + + +if __name__ == "__main__": + main() diff --git a/tests/test_public_contract.py b/tests/test_public_contract.py index f60fc83..403c1f6 100644 --- a/tests/test_public_contract.py +++ b/tests/test_public_contract.py @@ -135,6 +135,18 @@ PUBLIC_IMPORTS = { "projection_hash", ), "docforge.projection_worker": ("render_projection_in_worker",), + "docforge.reference_config": ( + "REFERENCE_ADAPTER_CONFIG", + "REFERENCE_ADAPTER_SCHEMA_VERSION", + "REFERENCE_LANGUAGES", + "ReferenceAdapterConfigV1", + "load_reference_adapter_config", + ), + "docforge.reference_mcp": ( + "REFERENCE_MCP_MODULE", + "create_reference_project", + "create_reference_server", + ), "docforge.retrieval": ( "ContextCapsuleV1", "RetrievalPlanV1", diff --git a/tests/test_reference_mcp.py b/tests/test_reference_mcp.py new file mode 100644 index 0000000..5b8ad06 --- /dev/null +++ b/tests/test_reference_mcp.py @@ -0,0 +1,296 @@ +from __future__ import annotations + +import io +import json +import shutil +import sys +import tempfile +import unittest +from contextlib import redirect_stderr +from pathlib import Path +from unittest import mock + +from jsonschema import Draft202012Validator +from mcp.shared.memory import create_connected_server_and_client_session + +from docforge.adapter_launcher import AdapterLauncherV1 +from docforge.client_config import generate_adapter_client_configuration +from docforge.errors import DocForgeError +from docforge.index import ProjectIndex +from docforge.mcp_server import READ_TOOLS +from docforge.reference_config import ( + REFERENCE_ADAPTER_CONFIG, + load_reference_adapter_config, +) +from docforge.reference_mcp import ( + REFERENCE_MCP_MODULE, + create_reference_project, + create_reference_server, + main, +) + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "tests" / "fixtures" / "reference-python" +CPP_FIXTURE = ROOT / "tests" / "fixtures" / "reference-cpp" +SCHEMA = json.loads( + (ROOT / "schemas" / "reference-adapter.schema.json").read_text(encoding="utf-8") +) + + +class ReferenceMcpTests(unittest.IsolatedAsyncioTestCase): + def copy_fixture(self, parent: Path) -> Path: + root = parent / "reference-python" + shutil.copytree(FIXTURE, root) + return root.resolve() + + def copy_cpp_fixture(self, parent: Path) -> Path: + root = parent / "reference-cpp" + shutil.copytree(CPP_FIXTURE, root) + return root.resolve() + + @staticmethod + def write_config( + root: Path, + *, + language: str = "python", + source_roots: tuple[str, ...] = ("src",), + extra: str = "", + ) -> Path: + config = root.joinpath(*REFERENCE_ADAPTER_CONFIG.parts) + config.parent.mkdir(parents=True, exist_ok=True) + quoted_roots = ", ".join(json.dumps(item) for item in source_roots) + config.write_text( + "\n".join( + ( + "schema_version = 1", + 'project_id = "reference-python"', + 'title = "Runnable Python reference"', + f"language = {json.dumps(language)}", + f"source_roots = [{quoted_roots}]", + extra, + "", + ) + ), + encoding="utf-8", + ) + return config + + async def test_python_config_build_check_and_real_mcp_retrieval(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + self.write_config(root) + config = load_reference_adapter_config(root) + + Draft202012Validator(SCHEMA).validate(config.as_dict()) + self.assertEqual("python", config.language) + self.assertEqual((root / "src",), config.source_roots) + self.assertIsNone(config.compilation_database) + self.assertEqual(64, len(config.config_hash)) + + project = create_reference_project(root) + index = ProjectIndex(project) + built = index.build() + checked = index.check() + self.assertEqual("ok", built["status"]) + self.assertEqual("ok", checked["status"]) + self.assertTrue(project.descriptor.index_path.is_relative_to(root / ".docforge")) + + async with create_connected_server_and_client_session( + create_reference_server(root, capability_mode="read"), + raise_exceptions=True, + ) as session: + tools = tuple(tool.name for tool in (await session.list_tools()).tools) + bootstrap = await session.call_tool("docforge_bootstrap", {}) + search = await session.call_tool( + "docforge_search", + {"query": "Service", "limit": 5}, + ) + node_id = search.structuredContent["results"][0]["node_id"] + node = await session.call_tool("docforge_get_node", {"node_id": node_id}) + + self.assertEqual(READ_TOOLS, tools) + self.assertEqual("ok", bootstrap.structuredContent["status"]) + self.assertEqual( + REFERENCE_MCP_MODULE, + bootstrap.structuredContent["binding"]["server_module"], + ) + self.assertEqual( + "python", + bootstrap.structuredContent["binding"]["reference_language"], + ) + self.assertGreater(search.structuredContent["count"], 0) + self.assertEqual(node_id, node.structuredContent["node"]["node_id"]) + + async def test_cpp_config_build_check_and_real_mcp_retrieval(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_cpp_fixture(Path(directory)) + self.write_config( + root, + language="cpp", + source_roots=("include", "src"), + extra='compilation_database = "compile_commands.json"', + ) + config = load_reference_adapter_config(root) + Draft202012Validator(SCHEMA).validate(config.as_dict()) + + project = create_reference_project(root) + index = ProjectIndex(project) + self.assertEqual("ok", index.build()["status"]) + self.assertEqual("ok", index.check()["status"]) + + async with create_connected_server_and_client_session( + create_reference_server(root, capability_mode="read"), + raise_exceptions=True, + ) as session: + bootstrap = await session.call_tool("docforge_bootstrap", {}) + search = await session.call_tool( + "docforge_search", + {"query": "worker", "limit": 5}, + ) + node_id = search.structuredContent["results"][0]["node_id"] + node = await session.call_tool("docforge_get_node", {"node_id": node_id}) + + self.assertEqual("cpp", config.language) + self.assertEqual(root / "compile_commands.json", config.compilation_database) + self.assertEqual( + "cpp", + bootstrap.structuredContent["binding"]["reference_language"], + ) + self.assertEqual(node_id, node.structuredContent["node"]["node_id"]) + + def test_launcher_generates_exact_isolated_module_invocation(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + self.write_config(root) + project = create_reference_project(root) + launcher = AdapterLauncherV1.for_project(project, module=REFERENCE_MCP_MODULE) + + with mock.patch("docforge.client_config.subprocess.run") as executed: + result = generate_adapter_client_configuration( + project, + launcher, + "codex", + capability_mode="read", + ) + + self.assertEqual( + [ + "-I", + "-m", + REFERENCE_MCP_MODULE, + "--project-root", + str(root), + "--capability-mode", + "read", + ], + result["binding"]["args"], + ) + self.assertEqual({}, result["binding"]["environment"]) + self.assertNotIn("cwd", result["binding"]) + executed.assert_not_called() + + def test_config_drift_requires_runtime_restart(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + config_path = self.write_config(root) + project = create_reference_project(root) + project.validate_runtime() + + config_path.write_text( + config_path.read_text(encoding="utf-8").replace( + "Runnable Python reference", + "Changed Python reference", + ), + encoding="utf-8", + ) + with self.assertRaises(DocForgeError) as captured: + project.validate_runtime() + self.assertEqual("adapter_restart_required", captured.exception.code) + self.assertEqual( + [REFERENCE_ADAPTER_CONFIG.as_posix()], + captured.exception.details["changed"], + ) + + def test_config_rejects_unknown_fields_languages_and_language_mismatch(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + for language, extra, code in ( + ("ruby", "", "unsupported_reference_language"), + ("javascript", "", "reference_language_mismatch"), + ("python", 'command = "python project.py"', "invalid_reference_config"), + ( + "python", + 'compilation_database = "compile_commands.json"', + "invalid_reference_config", + ), + ("cpp", "", "invalid_reference_config"), + ): + with self.subTest(language=language, extra=extra): + self.write_config(root, language=language, extra=extra) + with self.assertRaises(DocForgeError) as captured: + load_reference_adapter_config(root) + self.assertEqual(code, captured.exception.code) + + def test_config_rejects_path_escape_and_source_symlinks(self) -> None: + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + root = self.copy_fixture(parent) + outside = parent / "outside" + outside.mkdir() + + self.write_config(root, source_roots=("../outside",)) + with self.assertRaises(DocForgeError) as escaped: + load_reference_adapter_config(root) + self.assertEqual("path_escape", escaped.exception.code) + + linked = root / "linked" + linked.symlink_to(outside, target_is_directory=True) + self.write_config(root, source_roots=("linked",)) + with self.assertRaises(DocForgeError) as linked_root: + load_reference_adapter_config(root) + self.assertEqual("path_escape", linked_root.exception.code) + + self.write_config(root) + external_source = outside / "external.py" + external_source.write_text("def external(): pass\n", encoding="utf-8") + source_link = root / "src" / "external.py" + source_link.symlink_to(external_source) + with self.assertRaises(DocForgeError) as linked_source: + load_reference_adapter_config(root) + self.assertEqual("path_escape", linked_source.exception.code) + + def test_config_file_must_be_a_regular_non_symlink(self) -> None: + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + root = self.copy_fixture(parent) + config_path = self.write_config(root) + outside = parent / "reference-adapter.toml" + shutil.copyfile(config_path, outside) + config_path.unlink() + config_path.symlink_to(outside) + + with self.assertRaises(DocForgeError) as captured: + load_reference_adapter_config(root) + self.assertEqual("path_escape", captured.exception.code) + + def test_reference_module_cli_advertises_read_mode_only(self) -> None: + with ( + redirect_stderr(io.StringIO()), + mock.patch.object( + sys, + "argv", + [ + REFERENCE_MCP_MODULE, + "--project-root", + "/irrelevant", + "--capability-mode", + "proposal", + ], + ), + mock.patch("docforge.reference_mcp.create_reference_server") as create_server, + self.assertRaises(SystemExit) as captured, + ): + main() + + self.assertEqual(2, captured.exception.code) + create_server.assert_not_called()