From cb52bf8ae6f170130ba58c81906432eb61533bf4 Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 14:33:42 -0400 Subject: [PATCH] Add project-owned adapter client launchers --- .../adapter-client-configuration.schema.json | 358 ++++++++++++++ schemas/adapter-launcher.schema.json | 44 ++ src/docforge/adapter_launcher.py | 260 ++++++++++ src/docforge/client_config.py | 453 ++++++++++++++++++ tests/test_adapter_launcher.py | 326 +++++++++++++ 5 files changed, 1441 insertions(+) create mode 100644 schemas/adapter-client-configuration.schema.json create mode 100644 schemas/adapter-launcher.schema.json create mode 100644 src/docforge/adapter_launcher.py create mode 100644 tests/test_adapter_launcher.py diff --git a/schemas/adapter-client-configuration.schema.json b/schemas/adapter-client-configuration.schema.json new file mode 100644 index 0000000..e316c93 --- /dev/null +++ b/schemas/adapter-client-configuration.schema.json @@ -0,0 +1,358 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://docforge.local/schema/adapter-client-configuration-v1.json", + "title": "DocForge project-owned adapter client configuration", + "$defs": { + "sha256": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, + "project": { + "type": "object", + "required": [ + "project_id", + "project_root", + "project_root_fingerprint", + "adapter", + "descriptor_hash" + ], + "properties": { + "project_id": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9._-]{1,127}$" + }, + "project_root": { + "type": "string", + "minLength": 1, + "maxLength": 4096 + }, + "project_root_fingerprint": { + "type": "string", + "pattern": "^[0-9a-f]{16}$" + }, + "adapter": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}@[A-Za-z0-9][A-Za-z0-9_.+-]{0,63}$" + }, + "descriptor_hash": { "$ref": "#/$defs/sha256" } + }, + "additionalProperties": false + }, + "source_availability": { + "type": "object", + "required": [ + "schema_version", + "status", + "method", + "revision", + "source_hash" + ], + "properties": { + "schema_version": { "const": 1 }, + "status": { "const": "available" }, + "method": { + "enum": ["incremental-state", "complete-projection"] + }, + "revision": { + "type": "string", + "minLength": 1, + "maxLength": 4096 + }, + "source_hash": { "$ref": "#/$defs/sha256" } + }, + "additionalProperties": false + }, + "binding": { + "type": "object", + "required": [ + "transport", + "capability_mode", + "adapter_policy", + "render_policy", + "command", + "args", + "environment", + "timeouts", + "launcher_hash" + ], + "properties": { + "transport": { "const": "stdio" }, + "capability_mode": { + "enum": ["read", "proposal", "application"] + }, + "adapter_policy": { + "$ref": "https://docforge.local/schema/client-configuration-v1.json#/$defs/adapter_policy" + }, + "render_policy": { + "type": "object", + "required": ["manual", "graph", "live_viewer"], + "properties": { + "manual": { "enum": ["auto", "explicit", "disabled"] }, + "graph": { "const": "disabled" }, + "live_viewer": { "const": "on-demand" } + }, + "additionalProperties": false + }, + "command": { + "type": "string", + "minLength": 1, + "maxLength": 4096 + }, + "args": { + "type": "array", + "minItems": 7, + "maxItems": 64, + "prefixItems": [ + { "const": "-I" }, + { "const": "-m" }, + { + "type": "string", + "minLength": 1, + "maxLength": 255, + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*){0,31}$", + "not": { "const": "docforge.mcp_server" } + }, + { "const": "--project-root" }, + { + "type": "string", + "minLength": 1, + "maxLength": 4096 + }, + { "const": "--capability-mode" }, + { "enum": ["read", "proposal", "application"] } + ], + "items": { + "type": "string", + "minLength": 1, + "maxLength": 4096 + } + }, + "environment": { + "type": "object", + "maxProperties": 0 + }, + "timeouts": { + "type": "object", + "required": ["startup_seconds", "tool_seconds"], + "properties": { + "startup_seconds": { + "type": "integer", + "minimum": 1, + "maximum": 3600 + }, + "tool_seconds": { + "type": "integer", + "minimum": 1, + "maximum": 86400 + } + }, + "additionalProperties": false + }, + "launcher_hash": { "$ref": "#/$defs/sha256" } + }, + "additionalProperties": false + }, + "projection_availability": { + "type": "object", + "required": [ + "manual_configured", + "portable_graph_configured", + "application_enabled", + "live_viewer_available" + ], + "properties": { + "manual_configured": { "type": "boolean" }, + "portable_graph_configured": { "type": "boolean" }, + "application_enabled": { "type": "boolean" }, + "live_viewer_available": { "const": true } + }, + "additionalProperties": false + } + }, + "type": "object", + "required": [ + "status", + "schema_version", + "operation", + "action", + "client", + "server_name", + "project", + "launcher", + "launcher_hash", + "source_availability", + "source_availability_hash", + "binding", + "effective_policy", + "projection_policy", + "projection_policy_hash", + "projection_availability", + "artifact", + "configuration_hash", + "warnings" + ], + "properties": { + "status": { "const": "ok" }, + "schema_version": { "const": 1 }, + "operation": { "const": "adapter_client.configure" }, + "action": { "enum": ["preview", "write"] }, + "client": { "enum": ["codex", "claude", "openclaw"] }, + "server_name": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + }, + "project": { "$ref": "#/$defs/project" }, + "launcher": { + "$ref": "https://docforge.local/schema/adapter-launcher-v1.json" + }, + "launcher_hash": { "$ref": "#/$defs/sha256" }, + "source_availability": { + "$ref": "#/$defs/source_availability" + }, + "source_availability_hash": { "$ref": "#/$defs/sha256" }, + "binding": { "$ref": "#/$defs/binding" }, + "effective_policy": { + "$ref": "https://docforge.local/schema/client-configuration-v1.json#/$defs/effective_policy" + }, + "projection_policy": { + "$ref": "https://docforge.local/schema/client-configuration-v1.json#/$defs/projection_policy" + }, + "projection_policy_hash": { "$ref": "#/$defs/sha256" }, + "projection_availability": { + "$ref": "#/$defs/projection_availability" + }, + "artifact": { + "$ref": "https://docforge.local/schema/client-configuration-v1.json#/properties/artifact" + }, + "configuration_hash": { "$ref": "#/$defs/sha256" }, + "warnings": { + "$ref": "https://docforge.local/schema/client-configuration-v1.json#/properties/warnings" + } + }, + "allOf": [ + { + "if": { + "properties": { "client": { "const": "codex" } }, + "required": ["client"] + }, + "then": { + "properties": { + "artifact": { + "properties": { + "format": { "const": "codex-toml-fragment-v1" } + } + } + } + } + }, + { + "if": { + "properties": { "client": { "const": "claude" } }, + "required": ["client"] + }, + "then": { + "properties": { + "artifact": { + "properties": { + "format": { "const": "claude-json-fragment-v1" } + } + }, + "warnings": { + "contains": { + "properties": { + "code": { "const": "timeout_format_unverified" } + }, + "required": ["code"] + } + } + } + } + }, + { + "if": { + "properties": { "client": { "const": "openclaw" } }, + "required": ["client"] + }, + "then": { + "properties": { + "artifact": { + "properties": { + "format": { "const": "openclaw-json-fragment-v1" } + } + } + } + } + }, + { + "if": { + "properties": { "action": { "const": "preview" } }, + "required": ["action"] + }, + "then": { + "properties": { + "artifact": { + "properties": { + "output_path": { "type": "null" }, + "write_state": { "const": "not_requested" }, + "durability": { "const": "not_applicable" } + } + } + } + }, + "else": { + "properties": { + "artifact": { + "properties": { + "write_state": { "enum": ["created", "unchanged"] } + } + } + } + } + }, + { + "if": { + "properties": { + "binding": { + "properties": { + "adapter_policy": { + "properties": { + "mode": { "const": "preserve-no-ast" } + }, + "required": ["mode"] + } + }, + "required": ["adapter_policy"] + } + }, + "required": ["binding"] + }, + "then": { + "properties": { + "binding": { + "properties": { + "args": { + "contains": { "const": "--no-ast" }, + "minContains": 1, + "maxContains": 1 + } + } + } + } + }, + "else": { + "properties": { + "binding": { + "properties": { + "args": { + "not": { + "contains": { "const": "--no-ast" } + } + } + } + } + } + } + } + ], + "additionalProperties": false +} diff --git a/schemas/adapter-launcher.schema.json b/schemas/adapter-launcher.schema.json new file mode 100644 index 0000000..04197a5 --- /dev/null +++ b/schemas/adapter-launcher.schema.json @@ -0,0 +1,44 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://docforge.local/schema/adapter-launcher-v1.json", + "title": "DocForge project-owned adapter launcher", + "type": "object", + "required": [ + "schema_version", + "entry_point", + "project_id", + "project_root", + "adapter", + "descriptor_hash", + "module" + ], + "properties": { + "schema_version": { "const": 1 }, + "entry_point": { "const": "python-module" }, + "project_id": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9._-]{1,127}$" + }, + "project_root": { + "type": "string", + "minLength": 1, + "maxLength": 4096 + }, + "adapter": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}@[A-Za-z0-9][A-Za-z0-9_.+-]{0,63}$" + }, + "descriptor_hash": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, + "module": { + "type": "string", + "minLength": 1, + "maxLength": 255, + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*){0,31}$", + "not": { "const": "docforge.mcp_server" } + } + }, + "additionalProperties": false +} diff --git a/src/docforge/adapter_launcher.py b/src/docforge/adapter_launcher.py new file mode 100644 index 0000000..6cd57cc --- /dev/null +++ b/src/docforge/adapter_launcher.py @@ -0,0 +1,260 @@ +"""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 diff --git a/src/docforge/client_config.py b/src/docforge/client_config.py index 18c9de6..d02e493 100644 --- a/src/docforge/client_config.py +++ b/src/docforge/client_config.py @@ -16,6 +16,12 @@ from dataclasses import dataclass from pathlib import Path from typing import Literal, cast +from .adapter_launcher import ( + AdapterLauncherV1, + AdapterSourceAvailabilityV1, + adapter_source_availability, + validate_adapter_launcher, +) from .changeset_contract import document_hash from .errors import DocForgeError from .models import ProjectDescriptor, ProjectService @@ -1079,3 +1085,450 @@ def generate_client_configuration( } _validate_configuration_result(result, trusted_descriptor=descriptor) return result + + +def generate_adapter_client_configuration( + project: ProjectService, + launcher: AdapterLauncherV1, + client: str, + *, + server_name: str | None = None, + capability_mode: str = "read", + proposal_writer: str | None = None, + canonical_applier: str | None = None, + no_ast: bool = False, + manual_render_policy: str | None = None, + portable_graph_policy: str | None = None, + live_viewer_policy: str | None = None, + startup_timeout: int = 30, + tool_timeout: int = 300, + output: Path | None = None, +) -> dict[str, object]: + """Generate one client fragment for an explicitly constructed project-owned adapter.""" + + validate_adapter_launcher(project, launcher) + selected_client = _client_name(client) + selected_mode = _capability_mode(capability_mode) + startup_seconds = _bounded_seconds( + startup_timeout, + field="startup_timeout", + maximum=3_600, + ) + tool_seconds = _bounded_seconds( + tool_timeout, + field="tool_timeout", + maximum=86_400, + ) + descriptor = project.descriptor + _validate_adapter_authority( + descriptor, + selected_mode=selected_mode, + proposal_writer=proposal_writer, + canonical_applier=canonical_applier, + ) + source_availability = adapter_source_availability(project, launcher) + fingerprint = project_root_fingerprint(descriptor.root) + selected_name = _validated_server_name( + server_name, + project_id=descriptor.project_id, + fingerprint=fingerprint, + ) + executable = Path(os.path.abspath(sys.executable)) + try: + executable_status = executable.stat() + except OSError as error: + raise DocForgeError( + "client_configuration_unavailable", + "Current Python executable cannot be inspected", + ) from error + if not stat.S_ISREG(executable_status.st_mode) or not os.access(executable, os.X_OK): + raise DocForgeError( + "client_configuration_unavailable", + "Current Python executable is not a runnable regular file", + ) + + policy = compose_effective_policy( + selected_mode=selected_mode, + capability_source="explicit", + no_ast=no_ast, + diagnostics=False, + render_configured=descriptor.render is not None, + application_enabled=canonical_applier is not None, + ) + projection_policy = compose_projection_policy( + manual=manual_render_policy, + portable_graph=portable_graph_policy, + live_viewer=live_viewer_policy, + manual_configured=descriptor.render is not None, + portable_graph_configured=descriptor.graph_render is not None, + application_enabled=canonical_applier is not None, + ) + default_projection_policy = compose_projection_policy( + manual_configured=descriptor.render is not None, + portable_graph_configured=descriptor.graph_render is not None, + application_enabled=canonical_applier is not None, + ) + arguments = [ + "-I", + "-m", + launcher.module, + "--project-root", + str(descriptor.root), + "--capability-mode", + selected_mode, + ] + if proposal_writer is not None: + arguments.extend(("--proposal-writer", proposal_writer)) + if canonical_applier is not None: + arguments.extend(("--canonical-applier", canonical_applier)) + for option, selected, default in ( + ( + "--manual-render-policy", + projection_policy.manual, + default_projection_policy.manual, + ), + ( + "--portable-graph-policy", + projection_policy.portable_graph, + default_projection_policy.portable_graph, + ), + ( + "--live-viewer-policy", + projection_policy.live_viewer, + default_projection_policy.live_viewer, + ), + ): + if selected != default: + arguments.extend((option, selected)) + if no_ast: + arguments.append("--no-ast") + + artifact_format, content, warning = _artifact( + selected_client, + server_name=selected_name, + command=str(executable), + arguments=arguments, + startup_timeout=startup_seconds, + tool_timeout=tool_seconds, + ) + + def validate_current_binding() -> None: + validate_adapter_launcher(project, launcher) + current = adapter_source_availability(project, launcher) + if current != source_availability: + raise DocForgeError( + "source_changed", + "Adapter sources changed during client configuration generation", + ) + + if output is None: + validate_current_binding() + write_state = "not_requested" + durability = "not_applicable" + publication_warning = None + output_path = None + else: + validate_current_binding() + write_state, durability, publication_warning, published_path = _atomic_write( + output, + content, + validate_binding=validate_current_binding, + ) + output_path = str(published_path) if published_path is not None else None + artifact = { + "format": artifact_format, + "content": content, + "content_sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), + "output_path": output_path, + "write_state": write_state, + "durability": durability, + } + binding = { + "transport": "stdio", + "capability_mode": selected_mode, + "adapter_policy": policy.adapter_policy(), + "render_policy": { + "manual": policy.manual_render, + "graph": policy.graph_render, + "live_viewer": policy.live_viewer, + }, + "command": str(executable), + "args": arguments, + "environment": {}, + "timeouts": { + "startup_seconds": startup_seconds, + "tool_seconds": tool_seconds, + }, + "launcher_hash": launcher.launcher_hash, + } + project_binding = { + "project_id": descriptor.project_id, + "project_root": str(descriptor.root), + "project_root_fingerprint": fingerprint, + "adapter": descriptor.adapter, + "descriptor_hash": descriptor.descriptor_hash, + } + projection_availability = { + "manual_configured": descriptor.render is not None, + "portable_graph_configured": descriptor.graph_render is not None, + "application_enabled": canonical_applier is not None, + "live_viewer_available": True, + } + policy_payload = policy.as_dict() + result: dict[str, object] = { + "status": "ok", + "schema_version": 1, + "operation": "adapter_client.configure", + "action": "write" if output is not None else "preview", + "client": selected_client, + "server_name": selected_name, + "project": project_binding, + "launcher": launcher.as_dict(), + "launcher_hash": launcher.launcher_hash, + "source_availability": source_availability.as_dict(), + "source_availability_hash": source_availability.availability_hash, + "binding": binding, + "effective_policy": policy_payload, + "projection_policy": projection_policy.as_dict(), + "projection_policy_hash": projection_policy.policy_hash, + "projection_availability": projection_availability, + "artifact": artifact, + "warnings": [ + *([] if warning is None else [{"code": "timeout_format_unverified"}]), + *([] if publication_warning is None else [{"code": publication_warning}]), + ], + } + result["configuration_hash"] = document_hash(_adapter_configuration_hash_payload(result)) + _validate_adapter_configuration_result( + result, + project=project, + launcher=launcher, + source_availability=source_availability, + ) + return result + + +def _validate_adapter_authority( + descriptor: ProjectDescriptor, + *, + selected_mode: CapabilityMode, + proposal_writer: str | None, + canonical_applier: str | None, +) -> None: + writer_ids = {writer.writer_id for writer in descriptor.proposal_writers} + if selected_mode == "read": + if proposal_writer is not None or canonical_applier is not None: + raise DocForgeError( + "invalid_capability_binding", + "Read configuration cannot bind proposal or application authority", + ) + elif selected_mode == "proposal": + if proposal_writer is None or proposal_writer not in writer_ids: + raise DocForgeError( + "capability_unavailable", + "Proposal configuration requires a descriptor-declared writer", + required="proposal_writer", + ) + if canonical_applier is not None: + raise DocForgeError( + "invalid_capability_binding", + "Proposal configuration cannot bind a canonical applier", + ) + elif ( + proposal_writer is None + or canonical_applier is None + or proposal_writer != canonical_applier + or proposal_writer not in writer_ids + ): + raise DocForgeError( + "capability_unavailable", + "Application configuration requires one declared writer/applier identity", + required="matching_declared_writer_and_applier", + ) + + +def _adapter_configuration_hash_payload(result: dict[str, object]) -> dict[str, object]: + artifact = cast(dict[str, object], result["artifact"]) + return { + "schema_version": 1, + "client": result["client"], + "server_name": result["server_name"], + "project": result["project"], + "launcher": result["launcher"], + "launcher_hash": result["launcher_hash"], + "source_availability": result["source_availability"], + "source_availability_hash": result["source_availability_hash"], + "binding": result["binding"], + "effective_policy": result["effective_policy"], + "projection_policy": result["projection_policy"], + "projection_policy_hash": result["projection_policy_hash"], + "projection_availability": result["projection_availability"], + "artifact_format": artifact["format"], + "artifact_content_sha256": artifact["content_sha256"], + } + + +def _validate_adapter_configuration_result( + result: dict[str, object], + *, + project: ProjectService, + launcher: AdapterLauncherV1, + source_availability: AdapterSourceAvailabilityV1, +) -> None: + validate_adapter_launcher(project, launcher) + descriptor = project.descriptor + project_binding = cast(dict[str, object], result["project"]) + if ( + result["launcher"] != launcher.as_dict() + or result["launcher_hash"] != launcher.launcher_hash + or result["source_availability"] != source_availability.as_dict() + or result["source_availability_hash"] != source_availability.availability_hash + or project_binding + != { + "project_id": descriptor.project_id, + "project_root": str(descriptor.root), + "project_root_fingerprint": project_root_fingerprint(descriptor.root), + "adapter": descriptor.adapter, + "descriptor_hash": descriptor.descriptor_hash, + } + ): + raise AssertionError("Generated adapter client identity drifted") + + artifact = cast(dict[str, object], result["artifact"]) + binding = cast(dict[str, object], result["binding"]) + content = cast(str, artifact["content"]) + if artifact["content_sha256"] != hashlib.sha256(content.encode("utf-8")).hexdigest(): + raise AssertionError("Generated adapter client content hash drifted") + timeouts = cast(dict[str, object], binding["timeouts"]) + expected_format, expected_content, _ = _artifact( + cast(ClientName, result["client"]), + server_name=cast(str, result["server_name"]), + command=cast(str, binding["command"]), + arguments=cast(list[str], binding["args"]), + startup_timeout=cast(int, timeouts["startup_seconds"]), + tool_timeout=cast(int, timeouts["tool_seconds"]), + ) + if artifact["format"] != expected_format or content != expected_content: + raise AssertionError("Generated adapter client artifact drifted from its binding") + if binding["environment"] != {} or binding["launcher_hash"] != launcher.launcher_hash: + raise AssertionError("Generated adapter launch boundary drifted") + + adapter_policy = cast(dict[str, object], binding["adapter_policy"]) + render_policy = cast(dict[str, object], binding["render_policy"]) + projection_policy = cast(dict[str, object], result["projection_policy"]) + projection_availability = cast(dict[str, object], result["projection_availability"]) + arguments = cast(list[str], binding["args"]) + prefix = [ + "-I", + "-m", + launcher.module, + "--project-root", + str(descriptor.root), + "--capability-mode", + cast(str, binding["capability_mode"]), + ] + if arguments[:7] != prefix: + raise AssertionError("Generated adapter client arguments drifted from their binding") + remaining = arguments[7:] + no_ast_argument = "--no-ast" in arguments + if no_ast_argument: + if remaining[-1:] != ["--no-ast"] or arguments.count("--no-ast") != 1: + raise AssertionError("Generated adapter no-AST argument layout drifted") + remaining = remaining[:-1] + projection_arguments: dict[str, str] = {} + authority_arguments: list[str] = [] + projection_options = { + "--manual-render-policy": "manual", + "--portable-graph-policy": "portable_graph", + "--live-viewer-policy": "live_viewer", + } + position = 0 + while position < len(remaining): + option = remaining[position] + field = projection_options.get(option) + if field is None: + authority_arguments.append(option) + position += 1 + continue + if position + 1 >= len(remaining) or option in projection_arguments: + raise AssertionError("Generated adapter projection argument layout drifted") + value = remaining[position + 1] + projection_arguments[option] = value + if projection_policy[field] != value: + raise AssertionError("Generated adapter projection policy argument drifted") + position += 2 + mode = binding["capability_mode"] + if ( + (mode == "read" and authority_arguments) + or ( + mode == "proposal" + and ( + len(authority_arguments) != 2 + or authority_arguments[0] != "--proposal-writer" + or not authority_arguments[1] + ) + ) + or ( + mode == "application" + and ( + len(authority_arguments) != 4 + or authority_arguments[0] != "--proposal-writer" + or authority_arguments[2] != "--canonical-applier" + or not authority_arguments[1] + or authority_arguments[1] != authority_arguments[3] + ) + ) + ): + raise AssertionError("Generated adapter authority argument layout drifted") + + expected_projection_policy = compose_projection_policy( + manual=projection_arguments.get("--manual-render-policy"), + portable_graph=projection_arguments.get("--portable-graph-policy"), + live_viewer=projection_arguments.get("--live-viewer-policy"), + manual_configured=cast(bool, projection_availability["manual_configured"]), + portable_graph_configured=cast( + bool, + projection_availability["portable_graph_configured"], + ), + application_enabled=cast( + bool, + projection_availability["application_enabled"], + ), + live_viewer_available=cast( + bool, + projection_availability["live_viewer_available"], + ), + ) + policy = cast(dict[str, object], result["effective_policy"]) + composed_policy = compose_effective_policy( + selected_mode=cast(CapabilityMode, mode), + capability_source="explicit", + no_ast=adapter_policy["mode"] == "preserve-no-ast", + diagnostics=False, + render_configured=render_policy["manual"] != "disabled", + application_enabled=mode == "application", + ) + if ( + projection_policy != expected_projection_policy.as_dict() + or projection_availability["manual_configured"] != (descriptor.render is not None) + or projection_availability["portable_graph_configured"] + != (descriptor.graph_render is not None) + or projection_availability["application_enabled"] != (mode == "application") + or projection_availability["live_viewer_available"] is not True + or policy != composed_policy.as_dict() + or adapter_policy != composed_policy.adapter_policy() + or no_ast_argument != (adapter_policy["mode"] == "preserve-no-ast") + or binding["launcher_hash"] != result["launcher_hash"] + ): + raise AssertionError("Generated adapter client policy drifted") + if result["projection_policy_hash"] != hashlib.sha256( + json.dumps( + projection_policy, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + ).hexdigest() or result["configuration_hash"] != document_hash( + _adapter_configuration_hash_payload(result) + ): + raise AssertionError("Generated adapter client hash drifted") diff --git a/tests/test_adapter_launcher.py b/tests/test_adapter_launcher.py new file mode 100644 index 0000000..2afa1e4 --- /dev/null +++ b/tests/test_adapter_launcher.py @@ -0,0 +1,326 @@ +from __future__ import annotations + +import json +import os +import shutil +import tempfile +import tomllib +import unittest +from dataclasses import replace +from pathlib import Path +from unittest import mock + +from jsonschema import Draft202012Validator +from referencing import Registry, Resource + +from docforge.adapter_contract import AdapterNode, AdapterProject, AdapterProjection +from docforge.adapter_launcher import AdapterLauncherV1 +from docforge.client_config import ( + _validate_adapter_configuration_result, + generate_adapter_client_configuration, + generate_client_configuration, +) +from docforge.errors import DocForgeError +from docforge.models import Node +from docforge.project import Project + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" +SCHEMAS = ROOT / "schemas" +LAUNCHER_SCHEMA = json.loads((SCHEMAS / "adapter-launcher.schema.json").read_text(encoding="utf-8")) +ADAPTER_CONFIGURATION_SCHEMA = json.loads( + (SCHEMAS / "adapter-client-configuration.schema.json").read_text(encoding="utf-8") +) +GENERIC_CONFIGURATION_SCHEMA = json.loads( + (SCHEMAS / "client-configuration.schema.json").read_text(encoding="utf-8") +) +SCHEMA_REGISTRY = Registry().with_resources( + ( + (LAUNCHER_SCHEMA["$id"], Resource.from_contents(LAUNCHER_SCHEMA)), + ( + GENERIC_CONFIGURATION_SCHEMA["$id"], + Resource.from_contents(GENERIC_CONFIGURATION_SCHEMA), + ), + ( + ADAPTER_CONFIGURATION_SCHEMA["$id"], + Resource.from_contents(ADAPTER_CONFIGURATION_SCHEMA), + ), + ) +) +ADAPTER_CONFIGURATION_VALIDATOR = Draft202012Validator( + ADAPTER_CONFIGURATION_SCHEMA, + registry=SCHEMA_REGISTRY, +) + + +class Loader: + def __init__(self, projection: AdapterProjection) -> None: + self.projection = projection + self.load_calls = 0 + + def load_projection(self) -> AdapterProjection: + self.load_calls += 1 + return self.projection + + +class DriftingLoader(Loader): + def __init__( + self, + projection: AdapterProjection, + changed: AdapterProjection, + ) -> None: + super().__init__(projection) + self.changed = changed + + def load_projection(self) -> AdapterProjection: + self.load_calls += 1 + return self.projection if self.load_calls < 3 else self.changed + + +class AdapterLauncherTests(unittest.TestCase): + def copy_fixture(self, destination: Path) -> Path: + root = destination / "alpha" + shutil.copytree(FIXTURES / "alpha", root) + return root + + def projection( + self, + root: Path, + *, + source_hash: str = "3" * 64, + ) -> AdapterProjection: + node = Node( + node_id="source.entry", + title="Entry", + family="source", + authority="derived", + status="active", + tags=("python",), + summary="Reference adapter entry.", + content="Reference adapter content.", + source_path="src/entry.py", + source_anchor="L1", + content_hash="2" * 64, + ) + return AdapterProjection( + project_id="adapter-client-fixture", + title="Adapter client fixture", + adapter_id="fixture-client", + adapter_version="1", + root=root, + revision="fixture-revision", + source_hash=source_hash, + nodes=(AdapterNode(node),), + edges=(), + ) + + def project( + self, + root: Path, + *, + loader: Loader | None = None, + ) -> tuple[AdapterProject, Loader]: + effective_loader = loader or Loader(self.projection(root)) + return ( + AdapterProject( + effective_loader, + cache_root=root / ".docforge-cache" / "adapter-client", + ), + effective_loader, + ) + + def test_launcher_and_all_client_fragments_are_deterministic_and_schema_valid(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + project, _ = self.project(root) + launcher = AdapterLauncherV1.for_project( + project, + module="fixture_adapter.mcp_server", + ) + Draft202012Validator(LAUNCHER_SCHEMA).validate(launcher.as_dict()) + previous = os.environ.get("DOCFORGE_ADAPTER_LAUNCHER_SECRET") + os.environ["DOCFORGE_ADAPTER_LAUNCHER_SECRET"] = "must-not-appear" + try: + with mock.patch("docforge.client_config.subprocess.run") as executed: + results: dict[str, dict[str, object]] = {} + for client in ("codex", "claude", "openclaw"): + first = generate_adapter_client_configuration( + project, + launcher, + client, + no_ast=True, + ) + second = generate_adapter_client_configuration( + project, + launcher, + client, + no_ast=True, + ) + self.assertEqual(first, second) + ADAPTER_CONFIGURATION_VALIDATOR.validate(first) + self.assertEqual(launcher.launcher_hash, first["launcher_hash"]) + self.assertEqual( + launcher.launcher_hash, + first["binding"]["launcher_hash"], + ) + self.assertEqual( + [ + "-I", + "-m", + "fixture_adapter.mcp_server", + "--project-root", + str(root), + "--capability-mode", + "read", + "--no-ast", + ], + first["binding"]["args"], + ) + self.assertEqual({}, first["binding"]["environment"]) + self.assertNotIn("cwd", first["binding"]) + self.assertNotIn( + "must-not-appear", + json.dumps(first, sort_keys=True), + ) + results[client] = first + executed.assert_not_called() + finally: + if previous is None: + os.environ.pop("DOCFORGE_ADAPTER_LAUNCHER_SECRET", None) + else: + os.environ["DOCFORGE_ADAPTER_LAUNCHER_SECRET"] = previous + + self.assertIn( + "mcp_servers", + tomllib.loads(results["codex"]["artifact"]["content"]), + ) + self.assertIn( + "mcpServers", + json.loads(results["claude"]["artifact"]["content"]), + ) + self.assertIn( + "mcp", + json.loads(results["openclaw"]["artifact"]["content"]), + ) + + def test_launcher_rejects_module_path_entry_point_and_argument_injection(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + project, _ = self.project(root) + launcher = AdapterLauncherV1.for_project( + project, + module="fixture_adapter.mcp_server", + ) + for module in ( + "-m", + "fixture_adapter.mcp_server --debug", + "fixture_adapter:mcp_server", + "fixture_adapter/mcp_server", + ".fixture_adapter", + "docforge.mcp_server", + "fixture_adapter.$server", + ): + with self.subTest(module=module), self.assertRaises(DocForgeError) as captured: + replace(launcher, module=module) + self.assertEqual("invalid_adapter_launcher", captured.exception.code) + + with self.assertRaises(DocForgeError): + replace(launcher, project_root=Path("relative/project")) + with self.assertRaises(DocForgeError): + replace(launcher, entry_point="console-script") # type: ignore[arg-type] + payload = launcher.as_dict() + payload["arguments"] = ["--shell", "command"] + with self.assertRaises(TypeError): + AdapterLauncherV1(**payload) # type: ignore[arg-type] + + result = generate_adapter_client_configuration(project, launcher, "codex") + result["binding"]["args"].append("--arbitrary") + with self.assertRaisesRegex(AssertionError, "Generated adapter client"): + _validate_adapter_configuration_result( + result, + project=project, + launcher=launcher, + source_availability=project_source_availability(result), + ) + + def test_wrong_project_descriptor_and_source_drift_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + first_root = (parent / "first").resolve() + second_root = (parent / "second").resolve() + first_root.mkdir() + second_root.mkdir() + first, _ = self.project(first_root) + second, _ = self.project(second_root) + launcher = AdapterLauncherV1.for_project( + first, + module="fixture_adapter.mcp_server", + ) + + with self.assertRaises(DocForgeError) as wrong_project: + generate_adapter_client_configuration(second, launcher, "codex") + self.assertEqual("adapter_launcher_mismatch", wrong_project.exception.code) + with self.assertRaises(DocForgeError) as descriptor_drift: + generate_adapter_client_configuration( + first, + replace(launcher, descriptor_hash="0" * 64), + "codex", + ) + self.assertEqual("adapter_launcher_mismatch", descriptor_drift.exception.code) + + base = self.projection(first_root) + changed = replace(base, source_hash="4" * 64) + drifting, _ = self.project( + first_root, + loader=DriftingLoader(base, changed), + ) + drifting_launcher = AdapterLauncherV1.for_project( + drifting, + module="fixture_adapter.mcp_server", + ) + with self.assertRaises(DocForgeError) as source_drift: + generate_adapter_client_configuration( + drifting, + drifting_launcher, + "openclaw", + ) + self.assertEqual("source_changed", source_drift.exception.code) + + def test_generic_configuration_remains_separate_and_compatible(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + generic = Project.open(root) + result = generate_client_configuration(generic, "codex", no_ast=True) + Draft202012Validator(GENERIC_CONFIGURATION_SCHEMA).validate(result) + self.assertEqual("client.configure", result["operation"]) + self.assertNotIn("launcher", result) + with self.assertRaises(DocForgeError) as generic_launcher: + AdapterLauncherV1.for_project( + generic, + module="fixture_adapter.mcp_server", + ) + self.assertEqual("adapter_launcher_unavailable", generic_launcher.exception.code) + + custom_root = (Path(directory) / "custom").resolve() + custom_root.mkdir() + custom, _ = self.project(custom_root) + with self.assertRaises(DocForgeError) as generic_api: + generate_client_configuration(custom, "codex") + self.assertEqual("missing_config", generic_api.exception.code) + + +def project_source_availability(result: dict[str, object]): + from docforge.adapter_launcher import AdapterSourceAvailabilityV1 + + payload = result["source_availability"] + return AdapterSourceAvailabilityV1( + schema_version=payload["schema_version"], + status=payload["status"], + method=payload["method"], + revision=payload["revision"], + source_hash=payload["source_hash"], + ) + + +if __name__ == "__main__": + unittest.main()