administrator/DocForge2
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

Add strict reference project MCP binding

Browse source
This commit is contained in:
Andraxion 2026-07-29 14:54:10 -04:00
parent 7bee0220c2
commit efe4a443b7
5 changed files with 1010 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

428
src/docforge/reference_config.py Normal file
View file

@ -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

196
src/docforge/reference_mcp.py Normal file
View file

@ -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()