"""Strict, data-only launch contracts for project-owned adapter MCP modules.""" from __future__ import annotations import re import stat from dataclasses import dataclass from pathlib import Path from typing import Literal from .changeset_contract import document_hash from .config_validation import ID_PATTERN from .errors import DocForgeError from .models import IncrementalStateProject, ProjectService, RuntimeValidatedProject ADAPTER_LAUNCHER_SCHEMA_VERSION = 1 _PYTHON_MODULE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*){0,31}") _ADAPTER_IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}@[A-Za-z0-9][A-Za-z0-9_.+-]{0,63}") _SHA256 = re.compile(r"[0-9a-f]{64}") __all__ = [ "ADAPTER_LAUNCHER_SCHEMA_VERSION", "AdapterLauncherV1", "AdapterSourceAvailabilityV1", "adapter_source_availability", "validate_adapter_launcher", ] @dataclass(frozen=True) class AdapterLauncherV1: """One immutable, project-bound Python module launch declaration. The contract intentionally has no command, shell string, working directory, environment, arbitrary arguments, discovery rule, or callable selector. """ schema_version: int entry_point: Literal["python-module"] project_id: str project_root: Path adapter: str descriptor_hash: str module: str def __post_init__(self) -> None: _validate_launcher_fields(self) @classmethod def for_project( cls, project: ProjectService, *, module: str, ) -> AdapterLauncherV1: """Bind a structural module entry point to one already constructed adapter project.""" descriptor = project.descriptor if descriptor.adapter == "generic": raise DocForgeError( "adapter_launcher_unavailable", "Project-owned launcher configuration requires a custom adapter", ) launcher = cls( schema_version=ADAPTER_LAUNCHER_SCHEMA_VERSION, entry_point="python-module", project_id=descriptor.project_id, project_root=descriptor.root, adapter=descriptor.adapter, descriptor_hash=descriptor.descriptor_hash, module=module, ) validate_adapter_launcher(project, launcher) return launcher def as_dict(self) -> dict[str, object]: return { "schema_version": self.schema_version, "entry_point": self.entry_point, "project_id": self.project_id, "project_root": str(self.project_root), "adapter": self.adapter, "descriptor_hash": self.descriptor_hash, "module": self.module, } @property def launcher_hash(self) -> str: return document_hash(self.as_dict()) @dataclass(frozen=True) class AdapterSourceAvailabilityV1: """One bounded source identity observed without executing the launcher module.""" schema_version: int status: Literal["available"] method: Literal["incremental-state", "complete-projection"] revision: str source_hash: str def __post_init__(self) -> None: revision = _runtime_value(self.revision) if ( self.schema_version != 1 or self.status != "available" or self.method not in {"incremental-state", "complete-projection"} or not isinstance(revision, str) or not revision or _SHA256.fullmatch(self.source_hash) is None ): raise DocForgeError( "invalid_adapter_launcher", "Adapter source availability evidence is invalid", ) def as_dict(self) -> dict[str, object]: return { "schema_version": self.schema_version, "status": self.status, "method": self.method, "revision": self.revision, "source_hash": self.source_hash, } @property def availability_hash(self) -> str: return document_hash(self.as_dict()) def validate_adapter_launcher( project: ProjectService, launcher: AdapterLauncherV1, ) -> None: """Require one launcher to match the exact current adapter project binding.""" _validate_launcher_fields(launcher) descriptor = project.descriptor _validate_project_root(descriptor.root) if descriptor.adapter == "generic": raise DocForgeError( "adapter_launcher_unavailable", "Project-owned launcher configuration requires a custom adapter", ) if ( launcher.project_id != descriptor.project_id or launcher.project_root != descriptor.root or launcher.adapter != descriptor.adapter or launcher.descriptor_hash != descriptor.descriptor_hash ): raise DocForgeError( "adapter_launcher_mismatch", "Adapter launcher does not match the selected project binding", project_id=descriptor.project_id, adapter=descriptor.adapter, ) if isinstance(project, RuntimeValidatedProject): project.validate_runtime() def adapter_source_availability( project: ProjectService, launcher: AdapterLauncherV1, ) -> AdapterSourceAvailabilityV1: """Capture current source identity without importing or executing the launcher module.""" validate_adapter_launcher(project, launcher) state = project.incremental_state() if isinstance(project, IncrementalStateProject) else None if state is not None: method: Literal["incremental-state", "complete-projection"] = "incremental-state" revision = state.revision source_hash = state.source_hash else: snapshot = project.load() descriptor = snapshot.descriptor if ( descriptor.project_id != launcher.project_id or descriptor.root != launcher.project_root or descriptor.adapter != launcher.adapter or descriptor.descriptor_hash != launcher.descriptor_hash ): raise DocForgeError( "adapter_launcher_mismatch", "Loaded adapter projection drifted from its launcher binding", ) method = "complete-projection" revision = snapshot.revision source_hash = snapshot.source_hash validate_adapter_launcher(project, launcher) return AdapterSourceAvailabilityV1( schema_version=1, status="available", method=method, revision=revision, source_hash=source_hash, ) def _validate_launcher_fields(launcher: AdapterLauncherV1) -> None: if type(launcher.schema_version) is not int or launcher.schema_version != 1: raise DocForgeError( "invalid_adapter_launcher", "Adapter launcher schema version is unsupported", supported=1, ) if launcher.entry_point != "python-module": raise DocForgeError( "invalid_adapter_launcher", "Adapter launcher entry point must be one isolated Python module", ) project_id = _runtime_value(launcher.project_id) project_root = _runtime_value(launcher.project_root) adapter = _runtime_value(launcher.adapter) descriptor_hash = _runtime_value(launcher.descriptor_hash) module = _runtime_value(launcher.module) if not isinstance(project_id, str) or ID_PATTERN.fullmatch(project_id) is None: raise DocForgeError("invalid_adapter_launcher", "Adapter launcher project ID is invalid") if not isinstance(project_root, Path): raise DocForgeError("invalid_adapter_launcher", "Adapter launcher project root is invalid") _validate_project_root(project_root) if not isinstance(adapter, str) or _ADAPTER_IDENTITY.fullmatch(adapter) is None: raise DocForgeError("invalid_adapter_launcher", "Adapter launcher identity is invalid") if not isinstance(descriptor_hash, str) or _SHA256.fullmatch(descriptor_hash) is None: raise DocForgeError("invalid_adapter_launcher", "Adapter descriptor hash is invalid") if ( not isinstance(module, str) or len(module) > 255 or _PYTHON_MODULE.fullmatch(module) is None or module == "docforge.mcp_server" ): raise DocForgeError( "invalid_adapter_launcher", "Adapter launcher module must be one constrained project-owned Python module", ) def _validate_project_root(root: Path) -> None: try: status = root.lstat() resolved = root.resolve(strict=True) except OSError as error: raise DocForgeError( "invalid_adapter_launcher", "Adapter launcher project root is unavailable", ) from error if ( not root.is_absolute() or stat.S_ISLNK(status.st_mode) or not stat.S_ISDIR(status.st_mode) or resolved != root ): raise DocForgeError( "invalid_adapter_launcher", "Adapter launcher project root must be one canonical real directory", ) def _runtime_value(value: object) -> object: """Keep runtime validation explicit even when static callers are typed.""" return value