"""Bounded, non-mutating checks for one project-bound client integration.""" from __future__ import annotations import hashlib import json import os import stat import sys import tomllib from contextlib import suppress from pathlib import Path from typing import Literal, cast from .client_config import CLIENT_NAMES, ClientName from .errors import DocForgeError from .models import ProjectService from .policy import CapabilityMode, compose_effective_policy from .project import ( project_root_fingerprint, validate_descriptor_binding, ) from .projection_policy import compose_projection_policy MAX_CLIENT_CONFIG_BYTES = 1_000_000 MAX_CLIENT_SERVERS = 256 MAX_CLIENT_ARGUMENTS = 64 MAX_CLIENT_ARGUMENT_CHARS = 4_096 MAX_CLIENT_ENVIRONMENT_KEYS = 64 RISKY_ENVIRONMENT_KEYS = frozenset({"LD_PRELOAD", "PYTHONHOME", "PYTHONPATH", "PYTHONSTARTUP"}) CheckState = Literal["passed", "warning", "failed", "skipped"] DOCTOR_RESULT_LIMIT_BYTES = 32_768 CHECK_IDS = ( "project.binding", "project.canonical_validation", "client.driver", "client.config", "client.entry", "server.executable", "server.arguments", "server.project_binding", "policy.effective", "policy.no_ast", "client.timeouts", "client.environment", "client.tool_filter", "derived.index", ) def _check( check_id: str, state: CheckState, code: str, message: str, **details: object, ) -> dict[str, object]: bounded_details: dict[str, object] = {} for key, value in sorted(details.items())[:16]: if isinstance(value, str): bounded_details[key] = value[:512] elif isinstance(value, (bool, int)) or value is None: bounded_details[key] = value elif isinstance(value, list): items = cast(list[object], value) bounded_details[key] = [ item[:256] if isinstance(item, str) else item for item in items[:16] if isinstance(item, (str, bool, int)) or item is None ] return { "check_id": check_id, "state": state, "code": code, "message": message, "details": bounded_details, } def _replace_check( checks: list[dict[str, object]], replacement: dict[str, object], ) -> None: check_id = replacement["check_id"] for index, check in enumerate(checks): if check["check_id"] == check_id: checks[index] = replacement return raise AssertionError(f"Doctor check {check_id!r} was not initialized") def _skipped(check_id: str, code: str, message: str) -> dict[str, object]: return _check(check_id, "skipped", code, message) def _validate_doctor_result(result: dict[str, object]) -> None: checks = cast(list[dict[str, object]], result["checks"]) check_ids = [cast(str, check["check_id"]) for check in checks] if tuple(check_ids) != CHECK_IDS: raise AssertionError("Doctor did not emit its exact ordered check inventory") expected = { state: sum(check["state"] == state for check in checks) for state in ("passed", "warning", "failed", "skipped") } if result["summary"] != expected: raise AssertionError("Doctor summary does not match its checks") expected_state = ( "unhealthy" if expected["failed"] else ("degraded" if expected["warning"] else "healthy") ) if result["doctor_state"] != expected_state: raise AssertionError("Doctor health does not match its checks") if len(json.dumps(result, sort_keys=True, separators=(",", ":")).encode("utf-8")) > ( DOCTOR_RESULT_LIMIT_BYTES ): raise AssertionError("Doctor result exceeded its bounded response contract") def _default_config(client: ClientName) -> Path: if client == "codex": codex_home = os.environ.get("CODEX_HOME") if codex_home: return Path(codex_home).expanduser() / "config.toml" return Path.home() / ".codex" / "config.toml" if client == "openclaw": explicit = os.environ.get("OPENCLAW_CONFIG_PATH") if explicit: return Path(explicit).expanduser() state_root = os.environ.get("OPENCLAW_STATE_DIR") if state_root: return Path(state_root).expanduser() / "openclaw.json" return Path.home() / ".openclaw" / "openclaw.json" claude_root = os.environ.get("CLAUDE_CONFIG_DIR") if claude_root: return Path(claude_root).expanduser() / ".claude.json" return Path.home() / ".claude.json" def _parent_binding_current(path: Path, directory_fd: int) -> bool: try: before = path.lstat() resolved = path.resolve(strict=True) after = path.lstat() opened = os.fstat(directory_fd) return ( not stat.S_ISLNK(before.st_mode) and stat.S_ISDIR(before.st_mode) and resolved == path and (before.st_dev, before.st_ino, before.st_mode) == (after.st_dev, after.st_ino, after.st_mode) == (opened.st_dev, opened.st_ino, opened.st_mode) ) except OSError: return False def _read_stable_regular( path: Path, ) -> tuple[bytes, tuple[int, int, int, int, int]]: absolute = Path(os.path.abspath(path.expanduser())) try: parent = absolute.parent parent_status = parent.lstat() if ( stat.S_ISLNK(parent_status.st_mode) or not stat.S_ISDIR(parent_status.st_mode) or parent.resolve(strict=True) != parent ): raise DocForgeError( "client_config_unsafe", "Client configuration parent is not one real directory", ) directory_fd = os.open( parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, ) except FileNotFoundError as error: raise DocForgeError( "client_config_missing", "Client configuration does not exist", ) from error except OSError as error: raise DocForgeError( "client_config_unsafe", "Client configuration cannot be inspected safely", ) from error opened_parent = os.fstat(directory_fd) if opened_parent.st_dev != parent_status.st_dev or opened_parent.st_ino != parent_status.st_ino: with suppress(OSError): os.close(directory_fd) raise DocForgeError( "client_config_changed", "Client configuration parent changed while it was opened", ) try: if not _parent_binding_current(parent, directory_fd): raise DocForgeError( "client_config_changed", "Client configuration parent changed before it was read", ) try: before = os.stat( absolute.name, dir_fd=directory_fd, follow_symlinks=False, ) except FileNotFoundError as error: raise DocForgeError( "client_config_missing", "Client configuration does not exist", ) from error except OSError as error: raise DocForgeError( "client_config_unsafe", "Client configuration cannot be inspected safely", ) from error if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode): raise DocForgeError( "client_config_unsafe", "Client configuration must be a regular file and not a symbolic link", ) if before.st_size > MAX_CLIENT_CONFIG_BYTES: raise DocForgeError( "client_config_oversized", "Client configuration exceeds the bounded doctor limit", maximum_bytes=MAX_CLIENT_CONFIG_BYTES, ) try: descriptor = os.open( absolute.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd, ) except OSError as error: raise DocForgeError( "client_config_unsafe", "Client configuration cannot be opened safely", ) from error try: opened = os.fstat(descriptor) if opened.st_dev != before.st_dev or opened.st_ino != before.st_ino: raise DocForgeError( "client_config_changed", "Client configuration changed while it was opened", ) chunks: list[bytes] = [] remaining = MAX_CLIENT_CONFIG_BYTES + 1 while remaining: chunk = os.read(descriptor, min(65_536, remaining)) if not chunk: break chunks.append(chunk) remaining -= len(chunk) raw = b"".join(chunks) finally: with suppress(OSError): os.close(descriptor) if len(raw) > MAX_CLIENT_CONFIG_BYTES: raise DocForgeError( "client_config_oversized", "Client configuration exceeds the bounded doctor limit", maximum_bytes=MAX_CLIENT_CONFIG_BYTES, ) try: after = os.stat( absolute.name, dir_fd=directory_fd, follow_symlinks=False, ) except OSError as error: raise DocForgeError( "client_config_changed", "Client configuration changed while it was read", ) from error before_identity = ( before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns, ) after_identity = ( after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns, ) if before_identity != after_identity: raise DocForgeError( "client_config_changed", "Client configuration changed while it was read", ) if not _parent_binding_current(parent, directory_fd): raise DocForgeError( "client_config_changed", "Client configuration parent changed while it was read", ) return raw, before_identity finally: with suppress(OSError): os.close(directory_fd) def _safe_regular_state(path: Path) -> Literal["missing", "present", "unsafe"]: absolute = Path(os.path.abspath(path)) parent = absolute.parent try: parent_status = parent.lstat() if ( stat.S_ISLNK(parent_status.st_mode) or not stat.S_ISDIR(parent_status.st_mode) or parent.resolve(strict=True) != parent ): return "unsafe" directory_fd = os.open( parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, ) except FileNotFoundError: return "missing" except OSError: return "unsafe" try: opened = os.fstat(directory_fd) if opened.st_dev != parent_status.st_dev or opened.st_ino != parent_status.st_ino: return "unsafe" if not _parent_binding_current(parent, directory_fd): return "unsafe" try: before = os.stat( absolute.name, dir_fd=directory_fd, follow_symlinks=False, ) except FileNotFoundError: return "missing" if _parent_binding_current(parent, directory_fd) else "unsafe" except OSError: return "unsafe" state: Literal["present", "unsafe"] = ( "present" if stat.S_ISREG(before.st_mode) and not stat.S_ISLNK(before.st_mode) else "unsafe" ) try: after = os.stat( absolute.name, dir_fd=directory_fd, follow_symlinks=False, ) except OSError: return "unsafe" if ( before.st_dev, before.st_ino, before.st_mode, before.st_size, before.st_mtime_ns, before.st_ctime_ns, ) != ( after.st_dev, after.st_ino, after.st_mode, after.st_size, after.st_mtime_ns, after.st_ctime_ns, ): return "unsafe" return state if _parent_binding_current(parent, directory_fd) else "unsafe" finally: with suppress(OSError): os.close(directory_fd) def _safe_directory(path: Path) -> bool: if not path.is_absolute(): return False try: before = path.lstat() if stat.S_ISLNK(before.st_mode) or not stat.S_ISDIR(before.st_mode): return False directory_fd = os.open( path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, ) except OSError: return False try: resolved = path.resolve(strict=True) after = path.lstat() opened = os.fstat(directory_fd) return resolved == path and (before.st_dev, before.st_ino, before.st_mode) == ( after.st_dev, after.st_ino, after.st_mode, ) == (opened.st_dev, opened.st_ino, opened.st_mode) except OSError: return False finally: with suppress(OSError): os.close(directory_fd) def _server_documents(client: ClientName, raw: bytes) -> dict[str, object]: try: text = raw.decode("utf-8") except UnicodeDecodeError as error: raise DocForgeError( "client_config_invalid", "Client configuration is not UTF-8", ) from error try: if client == "codex": document = cast(dict[str, object], tomllib.loads(text)) servers = document.get("mcp_servers", {}) else: loaded = cast(object, json.loads(text)) if not isinstance(loaded, dict): raise TypeError document = cast(dict[str, object], loaded) if client == "openclaw": mcp_value = document.get("mcp", {}) if not isinstance(mcp_value, dict): raise TypeError mcp = cast(dict[str, object], mcp_value) servers = mcp.get("servers", {}) else: servers = document.get("mcpServers", {}) except ( json.JSONDecodeError, tomllib.TOMLDecodeError, RecursionError, TypeError, ValueError, ) as error: raise DocForgeError( "client_config_invalid", "Client configuration has invalid syntax or structure", ) from error if not isinstance(servers, dict): raise DocForgeError( "client_config_invalid", "Client MCP server collection must be an object", ) server_map = cast(dict[object, object], servers) if len(server_map) > MAX_CLIENT_SERVERS: raise DocForgeError( "client_config_oversized", "Client configuration declares too many MCP servers", maximum_servers=MAX_CLIENT_SERVERS, ) if any(not isinstance(name, str) or len(name) > 256 for name in server_map): raise DocForgeError( "client_config_invalid", "Client MCP server names are invalid", ) return cast(dict[str, object], server_map) def _normalize_entry(name: str, value: object, *, client: ClientName) -> dict[str, object]: if not isinstance(value, dict): raise DocForgeError( "client_entry_invalid", "Client MCP server entry must be an object", server_name=name, ) entry = cast(dict[str, object], value) allowed_fields = { "codex": { "command", "args", "env", "startup_timeout_sec", "tool_timeout_sec", "tools", }, "openclaw": { "command", "args", "env", "connectTimeout", "timeout", "supportsParallelToolCalls", "cwd", "toolFilter", }, "claude": {"command", "args", "env"}, }[client] unknown_fields = sorted(set(entry) - allowed_fields) if unknown_fields: raise DocForgeError( "client_entry_invalid", "Client MCP server entry contains unsupported fields", server_name=name, field_count=len(unknown_fields), fields_sha256=hashlib.sha256("\0".join(unknown_fields).encode("utf-8")).hexdigest(), ) command = entry.get("command") arguments_value = entry.get("args", []) environment_value = entry.get("env", {}) if ( not isinstance(command, str) or not command or len(command) > MAX_CLIENT_ARGUMENT_CHARS or "\0" in command or not isinstance(arguments_value, list) or not isinstance(environment_value, dict) ): raise DocForgeError( "client_entry_invalid", "Client MCP server command, arguments, or environment is invalid", server_name=name, ) arguments = cast(list[object], arguments_value) if len(arguments) > MAX_CLIENT_ARGUMENTS or any( not isinstance(argument, str) or len(argument) > MAX_CLIENT_ARGUMENT_CHARS or "\0" in argument for argument in arguments ): raise DocForgeError( "client_entry_invalid", "Client MCP server arguments exceed the bounded contract", server_name=name, ) environment = cast(dict[object, object], environment_value) if len(environment) > MAX_CLIENT_ENVIRONMENT_KEYS or any( not isinstance(value, str) or len(value) > MAX_CLIENT_ARGUMENT_CHARS or "\0" in value for value in environment.values() ): raise DocForgeError( "client_entry_invalid", "Client MCP server environment values exceed the bounded contract", server_name=name, ) raw_environment_keys = list(environment) if any(not isinstance(key, str) or len(key) > 256 for key in raw_environment_keys): raise DocForgeError( "client_entry_invalid", "Client MCP server environment keys are invalid", server_name=name, ) environment_keys = sorted(cast(list[str], raw_environment_keys)) startup: object = None tool: object = None if client == "codex": startup = entry.get("startup_timeout_sec") tool = entry.get("tool_timeout_sec") elif client == "openclaw": startup = entry.get("connectTimeout") tool = entry.get("timeout") cwd = entry.get("cwd") if cwd is not None and ( not isinstance(cwd, str) or not cwd or len(cwd) > MAX_CLIENT_ARGUMENT_CHARS or "\0" in cwd ): raise DocForgeError( "client_entry_invalid", "Client MCP working directory is invalid", server_name=name, ) parallel_calls = entry.get("supportsParallelToolCalls") if parallel_calls is not None and type(parallel_calls) is not bool: raise DocForgeError( "client_entry_invalid", "Client parallel-call setting must be Boolean", server_name=name, ) filter_value = entry.get("toolFilter", entry.get("tools")) filter_valid = True if filter_value is not None: if not isinstance(filter_value, dict): filter_valid = False elif client == "openclaw": filter_document = cast(dict[object, object], filter_value) filter_valid = set(filter_document).issubset({"include", "exclude"}) and all( isinstance(values, list) and len(cast(list[object], values)) <= 256 and all( isinstance(item, str) and 0 < len(item) <= 256 and "\0" not in item for item in cast(list[object], values) ) for values in filter_document.values() ) if not filter_valid: raise DocForgeError( "client_entry_invalid", "Client tool-filter setting is malformed", server_name=name, ) return { "server_name": name, "command": command, "args": cast(list[str], arguments), "environment_keys": environment_keys, "startup_timeout": startup, "tool_timeout": tool, "cwd": cwd, "tool_filter_present": filter_value is not None, "parallel_calls": parallel_calls, } def _parse_binding(arguments: list[str]) -> dict[str, object]: remaining = list(arguments) if remaining[:3] != ["-I", "-m", "docforge.mcp_server"]: raise DocForgeError( "server_arguments_invalid", "Python launch arguments must begin with the isolated DocForge module prefix", ) remaining = remaining[3:] values: dict[str, str] = {} flags: set[str] = set() value_options = { "--project-root", "--proposal-writer", "--canonical-applier", "--capability-mode", "--manual-render-policy", "--portable-graph-policy", "--live-viewer-policy", } flag_options = {"--no-ast", "--diagnostics"} position = 0 while position < len(remaining): option = remaining[position] if option in value_options: if option in values or position + 1 >= len(remaining): raise DocForgeError( "server_arguments_invalid", "Server arguments contain a duplicate or missing option value", option=option, ) values[option] = remaining[position + 1] position += 2 continue if option in flag_options: if option in flags: raise DocForgeError( "server_arguments_invalid", "Server arguments contain a duplicate flag", option=option, ) flags.add(option) position += 1 continue raise DocForgeError( "server_arguments_invalid", "Server arguments contain an unsupported option", argument_index=position + 3, argument_sha256=hashlib.sha256(option.encode("utf-8")).hexdigest(), ) selected_mode = values.get("--capability-mode") implicit = selected_mode is None if selected_mode is None: selected_mode = "application" if "--canonical-applier" in values else "proposal" if selected_mode not in {"read", "proposal", "application", "operator"}: raise DocForgeError( "server_arguments_invalid", "Configured capability mode is unsupported", ) return { "project_root": values.get("--project-root"), "proposal_writer": values.get("--proposal-writer"), "canonical_applier": values.get("--canonical-applier"), "capability_mode": selected_mode, "capability_mode_implicit": implicit, "no_ast": "--no-ast" in flags, "diagnostics": "--diagnostics" in flags, "manual_render_policy": values.get("--manual-render-policy"), "portable_graph_policy": values.get("--portable-graph-policy"), "live_viewer_policy": values.get("--live-viewer-policy"), } def _partial_project_roots(value: object) -> tuple[str, ...]: if not isinstance(value, dict): return () entry = cast(dict[str, object], value) arguments_value = entry.get("args") if not isinstance(arguments_value, list): return () arguments = cast(list[object], arguments_value) roots: list[str] = [] for position, argument in enumerate(arguments[:-1]): if argument == "--project-root" and isinstance(arguments[position + 1], str): roots.append(cast(str, arguments[position + 1])) return tuple(roots) def _entry_for_project( servers: dict[str, object], *, client: ClientName, project_root: Path, server_name: str | None, ) -> dict[str, object] | None: if server_name is not None: value = servers.get(server_name) return None if value is None else _normalize_entry(server_name, value, client=client) candidates: list[tuple[str, object]] = [] for name, value in sorted(servers.items()): for configured_root in _partial_project_roots(value): try: configured_path = Path(os.path.normpath(configured_root)) matches = configured_path.is_absolute() and configured_path == project_root except ValueError: matches = False if matches: candidates.append((name, value)) break if len(candidates) > 1: raise DocForgeError( "client_entry_ambiguous", "More than one client entry binds the project; select --server-name", count=len(candidates), ) if not candidates: return None name, value = candidates[0] return _normalize_entry(name, value, client=client) def _runtime_policy_check( project: ProjectService, binding: dict[str, object], ) -> tuple[CheckState, str, str]: mode = cast(str, binding["capability_mode"]) selected_mode = cast(CapabilityMode, mode) proposal_writer = cast(str | None, binding["proposal_writer"]) canonical_applier = cast(str | None, binding["canonical_applier"]) writer_ids = {writer.writer_id for writer in project.descriptor.proposal_writers} if proposal_writer is not None and proposal_writer not in writer_ids: raise DocForgeError( "effective_policy_invalid", "Configured proposal writer is not declared by the project", ) if canonical_applier is not None and canonical_applier not in writer_ids: raise DocForgeError( "effective_policy_invalid", "Configured canonical applier is not declared by the project", ) if selected_mode == "application" and canonical_applier is None: raise DocForgeError( "effective_policy_invalid", "Application capability requires a canonical applier", ) compose_effective_policy( selected_mode=selected_mode, capability_source=( "factory_default" if cast(bool, binding["capability_mode_implicit"]) else "explicit" ), no_ast=cast(bool, binding["no_ast"]), diagnostics=cast(bool, binding["diagnostics"]), render_configured=project.descriptor.render is not None, application_enabled=( canonical_applier is not None and selected_mode in {"application", "operator"} ), ) compose_projection_policy( manual=cast(str | None, binding["manual_render_policy"]), portable_graph=cast(str | None, binding["portable_graph_policy"]), live_viewer=cast(str | None, binding["live_viewer_policy"]), manual_configured=project.descriptor.render is not None, portable_graph_configured=project.descriptor.graph_render is not None, application_enabled=( canonical_applier is not None and selected_mode in {"application", "operator"} ), ) if cast(bool, binding["capability_mode_implicit"]): return ( "warning", "capability_mode_implicit", "Legacy configuration infers capability mode.", ) if selected_mode == "read" and (proposal_writer is not None or canonical_applier is not None): return ( "warning", "read_authority_shadowed", "Read mode shadows configured proposal or application authority.", ) if selected_mode == "proposal" and proposal_writer is None: return ( "warning", "proposal_access_disabled", "Proposal surface is present, but mutation access has no writer.", ) if selected_mode == "operator": return ( "warning", "operator_mode_reserved", "Operator mode is valid but currently adds no tools.", ) return ( "passed", "effective_policy_valid", "Configured capability and authority are valid.", ) def run_doctor( project: ProjectService, client: str, *, config_path: Path | None = None, server_name: str | None = None, ) -> dict[str, object]: """Inspect one client binding without loading project sources or mutating state.""" if client not in CLIENT_NAMES: raise DocForgeError( "unsupported_client", "Doctor client is unsupported", client=client, allowed=list(CLIENT_NAMES), ) selected_client = client descriptor = project.descriptor validate_descriptor_binding(descriptor) checks: list[dict[str, object]] = [ _check( "project.binding", "passed", "project_binding_valid", "Project descriptor and root binding are valid.", ), _check( "project.canonical_validation", "skipped", "canonical_validation_not_run", "Doctor does not parse canonical project sources.", ), _check( "client.driver", "passed" if selected_client != "claude" else "warning", ( "client_driver_valid" if selected_client != "claude" else "client_driver_format_partially_verified" ), ( "Client configuration format is supported." if selected_client != "claude" else "Claude fragment syntax is supported, but timeout fields are unverified." ), ), ] candidate_path = str((config_path or _default_config(selected_client)).expanduser()) path_error: DocForgeError | None = None if "\0" in candidate_path or len(candidate_path) > 4_096: path_hash = hashlib.sha256(candidate_path.encode("utf-8", errors="replace")).hexdigest() selected_path = Path.cwd() / ".invalid-docforge-client-config" displayed_path = f"" path_error = DocForgeError( "client_config_invalid", "Client configuration path is empty, oversized, or contains NUL", path_sha256=path_hash, ) else: selected_path = Path(os.path.abspath(candidate_path)) displayed_path = str(selected_path) selected_server_name = server_name server_name_error: DocForgeError | None = None if selected_server_name is not None and ( not selected_server_name or len(selected_server_name) > 256 or "\0" in selected_server_name ): name_hash = hashlib.sha256( selected_server_name.encode("utf-8", errors="replace") ).hexdigest() selected_server_name = None server_name_error = DocForgeError( "client_entry_invalid", "Explicit server name is empty, oversized, or contains NUL", server_name_sha256=name_hash, ) entry: dict[str, object] | None = None binding: dict[str, object] | None = None servers: dict[str, object] | None = None initial_config: bytes | None = None initial_config_identity: tuple[int, int, int, int, int] | None = None try: if path_error is not None: raise path_error initial_config, initial_config_identity = _read_stable_regular(selected_path) servers = _server_documents(selected_client, initial_config) checks.append( _check( "client.config", "passed", "client_config_valid", "Client configuration is bounded, stable, and parseable.", server_count=len(servers), ) ) except DocForgeError as error: checks.append( _check( "client.config", "failed", error.code, error.message, **error.details, ) ) if servers is None: checks.append( _skipped( "client.entry", "client_config_unavailable", "Client entry selection requires one valid configuration.", ) ) elif server_name_error is not None: checks.append( _check( "client.entry", "failed", server_name_error.code, server_name_error.message, **server_name_error.details, ) ) else: try: entry = _entry_for_project( servers, client=selected_client, project_root=descriptor.root, server_name=selected_server_name, ) checks.append( _check( "client.entry", "passed" if entry is not None else "failed", "client_entry_valid" if entry is not None else "client_entry_missing", ( "One client entry uniquely binds this project." if entry is not None else "No client entry binds this project." ), **({"server_name": entry["server_name"]} if entry is not None else {}), ) ) except DocForgeError as error: checks.append( _check( "client.entry", "failed", error.code, error.message, **error.details, ) ) if entry is None: checks.extend( ( _skipped( "server.executable", "client_entry_unavailable", "Executable validation requires one selected client entry.", ), _skipped( "server.arguments", "client_entry_unavailable", "Argument validation requires one selected client entry.", ), ) ) else: command = cast(str, entry["command"]) expected = os.path.abspath(sys.executable) executable_valid = False if command == expected: try: command_path = Path(expected) command_status = command_path.stat() executable_valid = ( command_path.is_absolute() and stat.S_ISREG(command_status.st_mode) and os.access(command_path, os.X_OK) ) except (OSError, ValueError): pass checks.append( _check( "server.executable", "passed" if executable_valid and command == expected else "failed", ( "server_executable_valid" if executable_valid and command == expected else "server_executable_unexpected" ), ( "Configured server uses the current absolute Python executable." if executable_valid and command == expected else "Configured server executable is missing, unsafe, or unexpected." ), ) ) try: binding = _parse_binding(cast(list[str], entry["args"])) checks.append( _check( "server.arguments", "passed", "server_arguments_valid", "Server arguments use the closed DocForge option set.", ) ) except DocForgeError as error: checks.append( _check( "server.arguments", "failed", error.code, error.message, **error.details, ) ) if entry is None or binding is None: prerequisite = ( "client_entry_unavailable" if entry is None else "server_arguments_unavailable" ) checks.extend( _skipped( check_id, prerequisite, "Check requires one selected entry with valid server arguments.", ) for check_id in ( "server.project_binding", "policy.effective", "policy.no_ast", "client.timeouts", "client.environment", "client.tool_filter", ) ) else: configured_root = binding["project_root"] root_matches = False if isinstance(configured_root, str): try: configured_path = Path(os.path.normpath(configured_root)) root_matches = configured_path.is_absolute() and configured_path == descriptor.root except ValueError: root_matches = False checks.append( _check( "server.project_binding", "passed" if root_matches else "failed", "project_binding_matches" if root_matches else "project_binding_mismatch", ( "Configured project root matches the inspected project." if root_matches else "Configured project root does not match the inspected project." ), ) ) try: policy_state, policy_code, policy_message = _runtime_policy_check( project, binding, ) checks.append( _check( "policy.effective", policy_state, policy_code, policy_message, ) ) except DocForgeError as error: checks.append( _check( "policy.effective", "failed", error.code, error.message, **error.details, ) ) no_ast = cast(bool, binding["no_ast"]) checks.append( _check( "policy.no_ast", "passed", "no_ast_policy_valid" if no_ast else "no_ast_disabled", ( "No-AST binding blocks Logic publication and retrieval surfaces." if no_ast else "No-AST compatibility shorthand is not enabled." ), adapter_internals="unverifiable" if no_ast else "not_applicable", ) ) startup = entry["startup_timeout"] tool = entry["tool_timeout"] timeouts_valid = ( type(startup) is int and 1 <= startup <= 3_600 and type(tool) is int and 1 <= tool <= 86_400 ) timeouts_invalid = ( startup is not None and not (type(startup) is int and 1 <= startup <= 3_600) ) or (tool is not None and not (type(tool) is int and 1 <= tool <= 86_400)) checks.append( _check( "client.timeouts", ("passed" if timeouts_valid else ("failed" if timeouts_invalid else "warning")), ( "client_timeouts_valid" if timeouts_valid else ( "client_timeouts_invalid" if timeouts_invalid else "client_timeouts_unverified" ) ), ( "Client timeouts are explicit and bounded." if timeouts_valid else ( "Client timeout representation is invalid." if timeouts_invalid else "Client timeout representation is missing or unverified." ) ), ) ) environment_keys = cast(list[str], entry["environment_keys"]) risky_environment = sorted(set(environment_keys).intersection(RISKY_ENVIRONMENT_KEYS)) checks.append( _check( "client.environment", "passed" if not environment_keys else "warning", ( "client_environment_valid" if not environment_keys else ( "client_environment_risky" if risky_environment else "client_environment_present" ) ), ( "Client entry inherits no secret-bearing environment values." if not environment_keys else ( "Client entry declares environment keys; " "values were not returned or logged." ) ), environment_keys=environment_keys, risky_environment_keys=risky_environment, ) ) tool_filter_present = cast(bool, entry["tool_filter_present"]) parallel_calls = entry["parallel_calls"] cwd = entry["cwd"] cwd_valid = True if cwd is not None: try: cwd_valid = _safe_directory(Path(cast(str, cwd))) except ValueError: cwd_valid = False runtime_controls_present = tool_filter_present or parallel_calls is True checks.append( _check( "client.tool_filter", ( "failed" if not cwd_valid else ("warning" if runtime_controls_present else "passed") ), ( "client_cwd_invalid" if not cwd_valid else ( "client_tool_filter_unverified" if tool_filter_present else ( "client_parallel_calls_unverified" if parallel_calls is True else "client_tool_filter_not_configured" ) ) ), ( "Client working directory is missing, unsafe, or not absolute." if not cwd_valid else ( "Client runtime filtering or parallel-call controls are not interpreted." if runtime_controls_present else "Server-side capability registration is authoritative." ) ), parallel_calls=parallel_calls, tool_filter_present=tool_filter_present, ) ) index_state = _safe_regular_state(descriptor.index_path) checks.append( _check( "derived.index", "warning" if index_state != "present" else "passed", { "present": "index_present_unverified", "missing": "index_missing", "unsafe": "index_unsafe", }[index_state], ( "Index exists but was not opened or validated." if index_state == "present" else ( "Index is absent and may be built by an explicit bootstrap." if index_state == "missing" else "Index path is not a safe regular file." ) ), ) ) if initial_config is not None and initial_config_identity is not None: try: final_config, final_config_identity = _read_stable_regular(selected_path) if final_config != initial_config or final_config_identity != initial_config_identity: raise DocForgeError( "client_config_changed", "Client configuration changed during doctor inspection", ) except DocForgeError as error: _replace_check( checks, _check( "client.config", "failed", error.code, error.message, **error.details, ), ) counts = { state: sum(check["state"] == state for check in checks) for state in ("passed", "warning", "failed", "skipped") } doctor_state = ( "unhealthy" if counts["failed"] else ("degraded" if counts["warning"] else "healthy") ) validate_descriptor_binding(descriptor) result: dict[str, object] = { "status": "ok", "schema_version": 1, "doctor_state": doctor_state, "client": selected_client, "project": { "project_id": descriptor.project_id, "project_root": str(descriptor.root), "project_root_fingerprint": project_root_fingerprint(descriptor.root), "adapter": descriptor.adapter, }, "config": { "path": displayed_path, "server_name": ( entry["server_name"] if entry is not None else ( selected_server_name if selected_server_name is not None else ( None if server_name is None else ( "" ) ) ) ), }, "summary": counts, "guarantees": { "read_only": True, "project_loads": 0, "adapter_projection_loads": 0, "adapter_source_extractions": 0, "sqlite_opens": 0, "index_checks": 0, "index_synchronizations": 0, "index_builds": 0, "renders": 0, "viewer_operations": 0, "client_config_writes": 0, "configured_command_executions": 0, }, "checks": checks, } _validate_doctor_result(result) return result