"""Read-only diagnostics for the local CyberBrain26 installation.

``cyberbrain doctor`` reports the effective configuration and the Bubblewrap
prerequisites of the supported local prototype. This module is strictly
read-only:

- it creates no directory and no database;
- it never migrates, prunes, or otherwise modifies existing local state;
- it never changes the policy mode and never offers to change it;
- it performs, requests, or approves no action of any kind;
- it does not evaluate or execute any agent request, and it does not start a
  sandbox in order to test it.

SECRETS RULE: values of environment variables are never printed, logged, or
serialized. Only *presence* flags (``True``/``False``) for a short list of
conventional credential variable names are reported, and those flags are
advisory only. No provider key and no endpoint value leaves this module.

Honesty rule: the quality gates executed here are *configuration self-reports*
produced by the installation itself, not verified security evidence. Several
gate checks (notably ``no_hardcoded_secrets`` in the security gate and
``default_deny_policy`` in the architecture gate) always report ``pass``
without inspecting anything, and the agent-related security checks are skipped
when no agents are supplied. The rendered output states this explicitly.

One unavoidable SQLite side effect is documented rather than hidden: opening an
existing WAL-mode store read-only can create or touch the ``-wal`` and ``-shm``
sidecar files. No schema, row, journal mode, or file permission is changed, and
a store that does not exist is never created.

The policy mode is only ever REPORTED here. SAFE remains the default and the
supported mode for a local installation; any other mode is reported as a hard
failure. No environment variable, configuration file, or CLI flag can switch
the mode, and this module deliberately offers no such switch.
"""

from __future__ import annotations

import json
import os
import platform
import shutil
import sqlite3
import subprocess
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import typer

from cyberbrain26 import __version__
from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase
from cyberbrain26.execution_plane.bubblewrap_provider import BubblewrapExecutionProvider
from cyberbrain26.product.local_store import (
    DEFAULT_RETENTION_DAYS,
    ENV_WORKSPACES_ROOT,
    resolve_data_dir,
    resolve_db_path,
    retention_days_from_env,
    workspaces_root,
)
from cyberbrain26.security_plane.approval import ApprovalManager
from cyberbrain26.security_plane.identity import AgentIdentity, TrustLevel
from cyberbrain26.security_plane.policy import PolicyEngine, PolicyMode
from cyberbrain26.security_plane.quality_gates import QualityGateRunner

# SECRETS: this module never prints environment variable VALUES. Only presence
# flags (True/False) for the names in _CREDENTIAL_ENV_NAMES are reported, and
# those flags are advisory hints only.
_CREDENTIAL_ENV_NAMES: tuple[str, ...] = (
    "OPENAI_API_KEY",
    "ANTHROPIC_API_KEY",
    "GEMINI_API_KEY",
    "MISTRAL_API_KEY",
    "COHERE_API_KEY",
    "GROQ_API_KEY",
    "OPENROUTER_API_KEY",
    "LITELLM_API_KEY",
    "LITELLM_MASTER_KEY",
)

MIN_PYTHON_VERSION: tuple[int, int] = (3, 11)
BUBBLEWRAP_BINARY: str = "bwrap"
EXPECTED_DB_FILE_MODE: int = 0o600
EXPECTED_DATA_DIR_MODE: int = 0o700
EXPECTED_WORKSPACES_ROOT_MODE: int = 0o700

SECRETS_NOTE: str = (
    "Environment variable values are never printed: this report contains only presence "
    "flags (true/false) for a short list of conventional credential variable names."
)

PROTOTYPE_NOTES: tuple[str, ...] = (
    (
        "CyberBrain26 is a tested local Python prototype, not a production-ready or "
        "security-certified system."
    ),
    (
        "Quality gate results are configuration self-reports by the installation, not "
        "verified security evidence: the security gate check 'no_hardcoded_secrets' "
        "always passes without inspecting any file and the architecture gate check "
        "'default_deny_policy' always passes."
    ),
    (
        "RLIMIT_AS, RLIMIT_CPU and RLIMIT_FSIZE are cooperative resource limits of the "
        "child process and are NOT enforcement against hostile code."
    ),
    (
        "The sandbox applies no cgroup or process-count limit and no disk or inode "
        "quota, so process forking inside the sandbox is not bounded by the sandbox."
    ),
    (
        "A writable workspace handed to the sandbox is confined to the sanctioned root "
        "<data dir>/workspaces (overridable with CYBERBRAIN_WORKSPACES_ROOT): a path that "
        "does not resolve inside that root is refused before any process is launched. "
        "Content written to a confined workspace does persist outside the disposable "
        "filesystem, and the sandbox does not clean it up."
    ),
    (
        "Bubblewrap execution is Linux-only; other platforms are reported as a warning "
        "and the supported execution path is unavailable there."
    ),
    (
        "Doctor never verifies a running end-to-end agent task, an LLM response, a "
        "network egress control, or a backup/restore cycle."
    ),
    (
        "The ``--scopes`` CLI option grants scopes only to the 'coder' role; "
        "it is not applied to security, evaluator, research, browser, or planner "
        "steps. This is a known limitation of the current identity_scopes schema, "
        "not a scope-gate bypass."
    ),
    (
        "Presence of a credential environment variable says nothing about whether a "
        "provider is configured, reachable, or permitted; doctor contacts no provider."
    ),
    (
        "Reading an existing WAL-mode store can create or touch its -wal and -shm "
        "sidecar files; no schema, row, journal mode, or file permission is changed, and "
        "a missing store is never created."
    ),
)


@dataclass
class CheckResult:
    """A single diagnostic check row produced by :func:`run_checks`."""

    name: str
    status: str
    detail: str
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-serializable representation of this row."""
        return {
            "name": self.name,
            "status": self.status,
            "detail": self.detail,
            "metadata": dict(self.metadata),
        }


@dataclass
class DoctorReport:
    """Aggregated diagnostic report returned by :func:`run_checks`."""

    checks: list[CheckResult] = field(default_factory=list)
    notes: list[str] = field(default_factory=lambda: list(PROTOTYPE_NOTES))

    @property
    def failed(self) -> int:
        """Number of checks reported as ``fail``."""
        return sum(1 for check in self.checks if check.status == "fail")

    @property
    def warnings(self) -> int:
        """Number of checks reported as ``warn``."""
        return sum(1 for check in self.checks if check.status == "warn")

    def overall_status(self, *, strict: bool = False) -> str:
        """Return the overall status of the report.

        Args:
            strict: When ``True``, a ``warn`` row counts as a failure.

        Returns:
            ``"fail"`` when any check failed (or, in strict mode, warned),
            otherwise ``"ok"``.
        """
        if self.failed > 0:
            return "fail"
        if strict and self.warnings > 0:
            return "fail"
        return "ok"

    def to_dict(self, *, strict: bool = False) -> dict[str, Any]:
        """Return a JSON-serializable representation of the whole report."""
        return {
            "command": "cyberbrain doctor",
            "read_only": True,
            "changes_nothing": True,
            "performs_or_approves_actions": False,
            "policy_mode_report_only": True,
            "secrets_policy": SECRETS_NOTE,
            "strict": strict,
            "overall_status": self.overall_status(strict=strict),
            "counts": {
                "checks": len(self.checks),
                "failed": self.failed,
                "warnings": self.warnings,
            },
            "checks": [check.to_dict() for check in self.checks],
            "notes": list(self.notes) + [SECRETS_NOTE],
        }


class _ReadOnlyDatabase(SQLiteDatabase):
    """A :class:`SQLiteDatabase` that can only read an existing store.

    Constructing the real database creates the parent directory, applies
    migrations, switches the journal to WAL and chmods the file. ``doctor``
    must not write, so this subclass skips initialization and opens every
    connection in SQLite read-only (``mode=ro``) mode.
    """

    def __init__(self, path: str | Path, *, retention_days: int = 90) -> None:
        """Record the location without creating or touching any file."""
        self.path = Path(path).expanduser().resolve()
        self.retention_days = retention_days

    def _initialize(self) -> None:
        """Do nothing: no migration, no WAL switch, no permission change."""

    @contextmanager
    def connection(self) -> Iterator[sqlite3.Connection]:
        """Yield a read-only SQLite connection and always close it afterward."""
        uri = f"file:{self.path}?mode=ro"
        connection = sqlite3.connect(uri, uri=True, timeout=5.0, isolation_level=None)
        connection.row_factory = sqlite3.Row
        try:
            yield connection
        finally:
            connection.close()


def _mode_string(path: Path) -> str:
    """Return the POSIX permission bits of ``path`` as an octal string."""
    try:
        return oct(path.stat().st_mode & 0o777)
    except OSError:
        return "unavailable"


def _is_world_or_group_readable(path: Path) -> bool:
    """Return whether the POSIX mode grants group or other read/write bits."""
    try:
        mode = path.stat().st_mode
    except OSError:
        return False
    return bool(mode & 0o077)


def _check_python_version() -> CheckResult:
    """Report the running interpreter against the declared ``>=3.11`` target."""
    info = sys.version_info
    current = f"{info.major}.{info.minor}.{info.micro}"
    required = f"{MIN_PYTHON_VERSION[0]}.{MIN_PYTHON_VERSION[1]}"
    metadata: dict[str, Any] = {"current": current, "required_minimum": required}
    if (info.major, info.minor) < MIN_PYTHON_VERSION:
        return CheckResult(
            name="python_version",
            status="fail",
            detail=(
                f"Python {current} is older than the required {required}+ declared in "
                'pyproject.toml (requires-python = ">=3.11").'
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="python_version",
        status="pass",
        detail=f"Python {current} (executable: {sys.executable}).",
        metadata=metadata,
    )


def _check_package_version() -> CheckResult:
    """Report the installed ``cyberbrain26`` package version."""
    return CheckResult(
        name="package_version",
        status="pass",
        detail=f"cyberbrain26 {__version__} imported from {Path(__file__).resolve().parent}.",
        metadata={"version": __version__},
    )


def _check_platform_supported() -> CheckResult:
    """Report whether the platform supports the Bubblewrap execution path."""
    system = platform.system()
    release = platform.release()
    machine = platform.machine()
    metadata: dict[str, Any] = {"system": system, "release": release, "machine": machine}
    if system != "Linux":
        return CheckResult(
            name="platform_supported",
            status="warn",
            detail=(
                f"Platform is {system} {release} ({machine}). The supported prototype "
                "target is Linux, so the Bubblewrap execution path is Linux-only and "
                "unavailable here; identity, policy, audit and the local store still "
                "apply."
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="platform_supported",
        status="pass",
        detail=f"Linux {release} ({machine}): the Bubblewrap execution path is supported.",
        metadata=metadata,
    )


def _check_data_dir(data_dir: Path) -> CheckResult:
    """Report the resolved data directory without creating it."""
    metadata: dict[str, Any] = {"path": str(data_dir), "exists": data_dir.exists()}
    expected_mode = oct(EXPECTED_DATA_DIR_MODE)[2:]
    if not data_dir.exists():
        return CheckResult(
            name="data_dir",
            status="warn",
            detail=(
                f"Data directory {data_dir} does not exist yet. Doctor does not create "
                f"it; the first command that writes local state creates it with mode "
                f"{expected_mode}."
            ),
            metadata=metadata,
        )
    if not data_dir.is_dir():
        return CheckResult(
            name="data_dir",
            status="fail",
            detail=f"Data directory path {data_dir} exists but is not a directory.",
            metadata=metadata,
        )
    writable = os.access(data_dir, os.W_OK | os.X_OK)
    mode = _mode_string(data_dir)
    metadata.update({"writable": writable, "mode": mode})
    if not writable:
        return CheckResult(
            name="data_dir",
            status="fail",
            detail=f"Data directory {data_dir} exists (mode {mode}) but is not writable.",
            metadata=metadata,
        )
    if _is_world_or_group_readable(data_dir):
        return CheckResult(
            name="data_dir",
            status="warn",
            detail=(
                f"Data directory {data_dir} is writable (mode {mode}) but readable by "
                f"group or others; local state should stay owner-only ({expected_mode})."
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="data_dir",
        status="pass",
        detail=f"Data directory {data_dir} exists, is writable, mode {mode}.",
        metadata=metadata,
    )


def _read_user_version(db_path: Path) -> int | None:
    """Read ``PRAGMA user_version`` from an existing database in read-only mode.

    Returns ``None`` when the file cannot be opened as a readable SQLite store.
    """
    try:
        uri = f"file:{db_path}?mode=ro"
        connection = sqlite3.connect(uri, uri=True, timeout=5.0)
    except sqlite3.Error:
        return None
    try:
        row = connection.execute("PRAGMA user_version").fetchone()
    except sqlite3.Error:
        return None
    finally:
        connection.close()
    if row is None or row[0] is None:
        return None
    try:
        return int(row[0])
    except (TypeError, ValueError):
        return None


def _check_database(db_path: Path) -> CheckResult:
    """Report the expected local store file without creating or migrating it."""
    expected = SQLiteDatabase.CURRENT_SCHEMA_VERSION
    expected_mode = oct(EXPECTED_DB_FILE_MODE)[2:]
    metadata: dict[str, Any] = {
        "path": str(db_path),
        "exists": db_path.exists(),
        "expected_schema_version": expected,
    }
    if not db_path.exists():
        return CheckResult(
            name="database",
            status="warn",
            detail=(
                f"Database {db_path} does not exist yet. Doctor never creates it; the "
                f"store is created on first use with schema version {expected} and file "
                f"mode {expected_mode}."
            ),
            metadata=metadata,
        )
    mode = _mode_string(db_path)
    user_version = _read_user_version(db_path)
    metadata.update({"mode": mode, "user_version": user_version})
    if user_version is None:
        return CheckResult(
            name="database",
            status="fail",
            detail=(
                f"Database {db_path} exists (mode {mode}) but could not be read as a "
                "SQLite store; PRAGMA user_version is unavailable."
            ),
            metadata=metadata,
        )
    if user_version != expected:
        return CheckResult(
            name="database",
            status="fail",
            detail=(
                f"Database {db_path} has schema version {user_version} but this code "
                f"expects {expected}; a migration is required before use."
            ),
            metadata=metadata,
        )
    if _is_world_or_group_readable(db_path):
        return CheckResult(
            name="database",
            status="warn",
            detail=(
                f"Database {db_path} is at schema version {user_version} (expected "
                f"{expected}) but its file mode {mode} is readable by group or others; "
                f"the expected mode is {expected_mode}."
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="database",
        status="pass",
        detail=(
            f"Database {db_path} exists, schema version {user_version} (expected "
            f"{expected}), file mode {mode}."
        ),
        metadata=metadata,
    )


def _check_retention(retention_days: Any) -> CheckResult:
    """Report the resolved retention window and reject a non-positive value."""
    metadata: dict[str, Any] = {
        "retention_days": retention_days,
        "env_var": "CYBERBRAIN_RETENTION_DAYS",
        "default": DEFAULT_RETENTION_DAYS,
    }
    if isinstance(retention_days, bool) or not isinstance(retention_days, int):
        return CheckResult(
            name="retention",
            status="fail",
            detail=(
                f"Resolved retention window {retention_days!r} is not an integer; the "
                "store requires an explicit positive number of days."
            ),
            metadata=metadata,
        )
    if retention_days <= 0:
        return CheckResult(
            name="retention",
            status="fail",
            detail=(
                f"Resolved retention window {retention_days} is not positive; the store "
                "requires an explicit positive number of days."
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="retention",
        status="pass",
        detail=(
            f"Retention window is {retention_days} days (CYBERBRAIN_RETENTION_DAYS, else "
            f"the default {DEFAULT_RETENTION_DAYS}). Retention is applied only when a "
            "caller explicitly runs pruning; doctor never prunes."
        ),
        metadata=metadata,
    )


def _check_bubblewrap() -> CheckResult:
    """Report Bubblewrap availability using only ``--version``.

    Bubblewrap is never asked to execute code; the sandbox is probed only by
    locating the binary and reading its version string.
    """
    resolved = shutil.which(BUBBLEWRAP_BINARY)
    metadata: dict[str, Any] = {"binary": BUBBLEWRAP_BINARY, "resolved_path": resolved}
    if resolved is None:
        return CheckResult(
            name="bubblewrap",
            status="fail",
            detail=(
                "bwrap was not found on PATH. The supported execution path fails closed "
                "without it (RuntimeError: Bubblewrap executable ... is unavailable; "
                "refusing to run code without a sandbox), so code execution is disabled "
                "on this machine. There is deliberately no unsandboxed fallback."
            ),
            metadata=metadata,
        )
    try:
        completed = subprocess.run(
            [resolved, "--version"],
            capture_output=True,
            text=True,
            timeout=10.0,
            check=False,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        return CheckResult(
            name="bubblewrap",
            status="fail",
            detail=f"bwrap found at {resolved} but 'bwrap --version' failed: {exc}.",
            metadata=metadata,
        )
    version_text = (completed.stdout or completed.stderr or "").strip()
    metadata["version_output"] = version_text
    metadata["returncode"] = completed.returncode
    if completed.returncode != 0:
        return CheckResult(
            name="bubblewrap",
            status="fail",
            detail=(
                f"bwrap found at {resolved} but 'bwrap --version' exited with "
                f"{completed.returncode}."
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="bubblewrap",
        status="pass",
        detail=f"bwrap found at {resolved}; 'bwrap --version' reports: {version_text}.",
        metadata=metadata,
    )


def _check_sandbox_limits() -> CheckResult:
    """Report the configured sandbox limits together with their honest caveats."""
    provider = BubblewrapExecutionProvider()
    limits: dict[str, Any] = {
        "max_timeout_seconds": provider.max_timeout_seconds,
        "max_memory_mb": provider.max_memory_mb,
        "max_cpu_seconds": provider.max_cpu_seconds,
        "max_file_size_mb": provider.max_file_size_mb,
        "max_output_bytes": provider.max_output_bytes,
    }
    detail = (
        f"Provider '{provider.provider_name}' is configured with: "
        f"max_timeout_seconds={limits['max_timeout_seconds']}, "
        f"max_memory_mb={limits['max_memory_mb']}, "
        f"max_cpu_seconds={limits['max_cpu_seconds']}, "
        f"max_file_size_mb={limits['max_file_size_mb']}, "
        f"max_output_bytes={limits['max_output_bytes']}. These are the DEFAULT values from "
        "the provider constructor; doctor does not verify them at runtime. Caveats: "
        "RLIMIT_AS, RLIMIT_CPU and RLIMIT_FSIZE are cooperative resource limits of the "
        "child process and are NOT enforcement against hostile code, because a program "
        "that ignores or resets them is not stopped by them. There is no cgroup or "
        "process-count limit, so forking inside the sandbox is not bounded by the "
        "sandbox, and there is no disk or inode quota on the sandbox filesystem. A "
        "writable workspace passed to the sandbox is confined to the sanctioned root "
        "<data dir>/workspaces (overridable with CYBERBRAIN_WORKSPACES_ROOT): a path that "
        "does not resolve inside that root is refused before any process is launched, so "
        "the read-write bind mount cannot expose an arbitrary host directory. Content "
        "written to a confined workspace does persist outside the disposable filesystem, "
        "and the sandbox does not clean it up."
    )
    return CheckResult(
        name="sandbox_limits",
        status="warn",
        detail=detail,
        metadata={"provider": provider.provider_name, **limits},
    )


def _configured_workspaces_root(data_dir: Path) -> tuple[Path, str]:
    """Return the sanctioned workspaces root and a label naming its source.

    This mirrors the precedence of
    :attr:`BubblewrapExecutionProvider.allowed_workspaces_root` but creates nothing:
    the provider's ``ensure_workspaces_root()`` is replaced by the non-creating
    :func:`workspaces_root`, because ``doctor`` must never make a directory.

    Args:
        data_dir: Resolved data directory used to derive the default root.

    Returns:
        The resolved root and a short label naming where it came from.
    """
    env_value = os.environ.get(ENV_WORKSPACES_ROOT)
    if env_value is not None and env_value.strip():
        return Path(env_value.strip()).expanduser().resolve(), ENV_WORKSPACES_ROOT
    return workspaces_root(data_dir), "default (<data dir>/workspaces)"


def _check_workspaces_root(data_dir: Path) -> CheckResult:
    """Report the sanctioned workspaces root without creating or modifying it.

    Only ``stat``-level inspection is performed: existence, directory-ness, and
    permission bits. No directory is created, no migration is run, and the root
    itself is left exactly as found.
    """
    root, source = _configured_workspaces_root(data_dir)
    expected_mode = oct(EXPECTED_WORKSPACES_ROOT_MODE)[2:]
    metadata: dict[str, Any] = {
        "path": str(root),
        "source": source,
        "env_var": ENV_WORKSPACES_ROOT,
        "exists": root.exists(),
        "expected_mode": expected_mode,
        "enforced_by": "local_store.resolve_workspace(), called before any process launch",
        "inspections_are_read_only": True,
    }
    confinement = (
        f"This root is the only tree a writable sandbox mount may live in: a workspace "
        f"that does not resolve inside {root} is refused before any process is launched."
    )
    if not root.exists():
        return CheckResult(
            name="workspaces_root",
            status="warn",
            detail=(
                f"Workspaces root {root} does not exist yet. Doctor never creates it; the "
                f"first command that needs a workspace creates it with mode "
                f"{expected_mode}. " + confinement
            ),
            metadata=metadata,
        )
    if not root.is_dir():
        return CheckResult(
            name="workspaces_root",
            status="fail",
            detail=f"Workspaces root path {root} exists but is not a directory.",
            metadata=metadata,
        )
    mode = _mode_string(root)
    metadata["mode"] = mode
    if _is_world_or_group_readable(root):
        return CheckResult(
            name="workspaces_root",
            status="warn",
            detail=(
                f"Workspaces root {root} exists (mode {mode}) but is readable by group or "
                f"others; sandboxed code writes there and the content survives the sandbox, "
                f"so the root should stay owner-only ({expected_mode}). " + confinement
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="workspaces_root",
        status="pass",
        detail=(
            f"Workspaces root {root} exists, is a directory, mode {mode}. " + confinement
        ),
        metadata=metadata,
    )


def _check_policy_mode() -> CheckResult:
    """Report the effective policy mode; anything but SAFE is a hard failure.

    The mode is only reported. ``doctor`` never changes it, and no environment
    variable, configuration file, or CLI flag switches it.
    """
    engine = PolicyEngine()
    configured = engine.mode
    rule_names = ", ".join(rule.name for rule in engine.list_rules())
    sensitive = sorted(PolicyEngine.SENSITIVE_ACTIONS)
    metadata: dict[str, Any] = {
        "configured_mode": configured.value,
        "rules_loaded_at_construction": rule_names,
        "rules_are_fixed_at_construction": True,
        "sensitive_actions": sensitive,
    }
    shared = (
        f"SAFE mode requires human approval for every action in SENSITIVE_ACTIONS "
        f"({len(sensitive)}): {', '.join(sensitive)}. No environment variable, "
        "configuration file, or CLI flag can switch the mode, and doctor reports the "
        "mode without offering to change it."
    )
    if configured is not PolicyMode.SAFE:
        metadata["mismatch"] = (
            "PolicyEngine.mode is a public attribute and can be reassigned without "
            f"reloading rules; the rules reported here were loaded for "
            f"{configured.value!r} at construction time."
        )
        return CheckResult(
            name="policy_mode",
            status="fail",
            detail=(
                f"Effective PolicyEngine mode is {configured.value!r}, not "
                f"{PolicyMode.SAFE.value!r}. A mode other than SAFE is not supported for a "
                f"local installation. Rules loaded at construction: [{rule_names}]. "
                "PolicyEngine.mode can be reassigned without reloading rules, so the "
                "configured value and the loaded rule set are reported separately. "
                + shared
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="policy_mode",
        status="pass",
        detail=(
            f"Effective PolicyEngine mode is {configured.value!r} (the SAFE default). "
            f"Rules loaded at construction: [{rule_names}]. PolicyEngine.mode is a public "
            "attribute that can be reassigned without reloading rules, so both the "
            "configured mode and the loaded rule set are reported here. " + shared
        ),
        metadata=metadata,
    )


def _check_agent_scopes() -> CheckResult:
    """Report how built-in agents receive scopes, using a scopeless probe identity.

    The probe identity below carries NO scopes: doctor grants itself no
    permission and reports no invented scopes.
    """
    probe = AgentIdentity(
        agent_id="cyberbrain-doctor",
        role="supervisor",
        trust_level=TrustLevel.VERIFIED,
    )
    probe_scopes = sorted(str(scope) for scope in probe.scopes)
    metadata: dict[str, Any] = {
        "probe_agent_id": probe.agent_id,
        "probe_role": probe.role,
        "probe_trust_level": probe.trust_level.value,
        "probe_scopes": probe_scopes,
    }
    if probe_scopes:
        return CheckResult(
            name="agent_scopes",
            status="fail",
            detail=(
                f"The doctor probe identity unexpectedly carries {len(probe_scopes)} "
                "scope(s); doctor must never hold execution scopes."
            ),
            metadata=metadata,
        )
    return CheckResult(
        name="agent_scopes",
        status="pass",
        detail=(
            "Built-in Ptah agents receive their permission scopes through "
            "AgentIdentity.scopes; this check infers no scope on their behalf. SAFE-mode "
            "identities have no execution scopes by default - a scope must be granted "
            "explicitly. The doctor probe identity 'cyberbrain-doctor' (role=supervisor, "
            f"trust_level={probe.trust_level.value}) was created with an empty scope set "
            f"({len(probe_scopes)} scopes) and holds no permission."
        ),
        metadata=metadata,
    )


def _check_pending_approvals(db_path: Path, retention_days: int) -> CheckResult:
    """Count pending human approval requests when an existing store is readable.

    The store is opened through a read-only connection, so no database is
    created, migrated, or modified. ``check_expired`` is deliberately not
    called because it marks requests as expired, which is a write.
    """
    metadata: dict[str, Any] = {"path": str(db_path), "exists": db_path.exists(), "read_only": True}
    if not db_path.exists():
        return CheckResult(
            name="pending_approvals",
            status="warn",
            detail=(
                f"No local store at {db_path}, so there is nothing to count. Doctor never "
                "creates the store to find out."
            ),
            metadata=metadata,
        )
    try:
        database = _ReadOnlyDatabase(db_path, retention_days=retention_days)
        pending = ApprovalManager(database=database).list_pending()
    except (sqlite3.Error, OSError, ValueError, RuntimeError) as exc:
        metadata["error"] = f"{type(exc).__name__}: {exc}"
        return CheckResult(
            name="pending_approvals",
            status="fail",
            detail=(
                f"The local store at {db_path} could not be read for pending approvals: "
                f"{type(exc).__name__}: {exc}"
            ),
            metadata=metadata,
        )
    metadata["pending_count"] = len(pending)
    metadata["approval_timeout_seconds"] = ApprovalManager.DEFAULT_TIMEOUT_SECONDS
    return CheckResult(
        name="pending_approvals",
        status="pass",
        detail=(
            f"{len(pending)} pending human approval request(s) in {db_path}, read through "
            "a read-only SQLite connection (default approval timeout "
            f"{ApprovalManager.DEFAULT_TIMEOUT_SECONDS:.0f}s). Doctor neither lists nor "
            "expires individual requests, and opening a WAL store read-only may create or "
            "touch its -wal and -shm sidecar files without changing any stored data."
        ),
        metadata=metadata,
    )


def _build_permission_gateway(policy_engine: PolicyEngine) -> Any:
    """Return a :class:`PermissionGateway` for gate wiring, or ``None``.

    The gateway exists only so the security gate can report that a permission
    gateway is configured. It grants nothing and evaluates nothing.
    """
    try:
        from cyberbrain26.security_plane.permission import PermissionGateway
    except ImportError:
        return None
    return PermissionGateway(policy_engine=policy_engine)


def _check_quality_gates(policy_engine: PolicyEngine, retention_days: int) -> CheckResult:
    """Run the quality gates and label the result as a configuration self-report."""
    gateway = _build_permission_gateway(policy_engine)
    runner = QualityGateRunner()
    try:
        results = runner.run_all(
            agents=None,
            permission_gateway=gateway,
            policy_engine=policy_engine,
            audit_log=None,
            firewall=None,
            retention_days=retention_days,
        )
    except Exception as exc:  # noqa: BLE001 - doctor must not crash on a gate
        return CheckResult(
            name="quality_gates",
            status="fail",
            detail=f"The quality gate runner raised {type(exc).__name__}: {exc}",
            metadata={},
        )
    summary: list[dict[str, Any]] = []
    failed_gates: list[str] = []
    warned_gates: list[str] = []
    for result in results:
        summary.append(
            {
                "gate_name": result.gate_name,
                "gate_id": result.gate_id,
                "status": result.status.value,
                "passed": result.passed,
                "failed": result.failed,
                "skipped": result.skipped,
                "warnings": result.warnings,
                "checks": [
                    {"name": check.name, "status": check.status.value} for check in result.checks
                ],
            }
        )
        if not result.overall_passed:
            failed_gates.append(f"{result.gate_name} ({result.gate_id})")
        elif result.warnings > 0:
            warned_gates.append(f"{result.gate_name} ({result.gate_id})")
    rendered = "; ".join(
        f"{item['gate_name']} [{item['gate_id']}] {item['status']} ({item['passed']} passed, "
        f"{item['failed']} failed, {item['skipped']} skipped, {item['warnings']} warnings)"
        for item in summary
    )
    detail = (
        "CONFIGURATION SELF-REPORT, NOT VERIFIED SECURITY EVIDENCE: these results are "
        f"produced by the installation itself. {rendered}. Limits of these gates: the "
        "security gate check 'no_hardcoded_secrets' always passes without inspecting any "
        "file, the architecture gate check 'default_deny_policy' always passes, and the "
        "agent checks of the security gate are skipped because doctor supplies no agent "
        "list. The permission gateway is wired only to show that an object is present, "
        "and no audit log or AI firewall instance is configured here, so these gates do "
        "not prove that audit logging or filtering works. A gate reported as passing is "
        "not evidence that the control it names is effective."
    )
    metadata: dict[str, Any] = {"gates": summary, "self_report": True}
    tail = (
        " A gate failure here reflects the configuration doctor wired, not a verified "
        "installation defect: doctor deliberately supplies no agent list, no audit log "
        "and no AI firewall instance, because opening an audit log would write to local "
        "state. A failing gate row is therefore reported as a warning, never as evidence "
        "that the named control is broken."
    )
    if failed_gates:
        metadata["failed_gates"] = failed_gates
        tail += f" Gates reported as failing: {', '.join(failed_gates)}."
    if warned_gates:
        metadata["warned_gates"] = warned_gates
        tail += f" Gates reported with warnings: {', '.join(warned_gates)}."
    return CheckResult(
        name="quality_gates",
        status="warn",
        detail=detail + tail,
        metadata=metadata,
    )


def _check_model_providers() -> CheckResult:
    """Report available provider adapters and credential presence flags only."""
    adapters: list[str] = []
    for module_name, label in (
        ("cyberbrain26.model_layer.litellm_client", "LiteLLM"),
        ("cyberbrain26.model_layer.ollama_client", "Ollama"),
        ("cyberbrain26.model_layer.llama_cpp_client", "llama.cpp"),
    ):
        try:
            __import__(module_name)
        except ImportError:
            continue
        adapters.append(label)
    presence = {name: bool(os.environ.get(name)) for name in _CREDENTIAL_ENV_NAMES}
    metadata: dict[str, Any] = {
        "adapters": adapters,
        "credential_env_presence": presence,
        "values_printed": False,
        "presence_is_advisory_only": True,
    }
    detail = (
        "Model provider adapters available in this installation: "
        f"{', '.join(adapters) if adapters else 'none importable'}. These are adapters "
        "only; doctor contacts no provider and cannot determine whether a provider is "
        "reachable, authorised, or correctly configured. The project defines no "
        "canonical provider-credential environment variable, so only presence "
        f"(true/false) is reported for the conventional variable names "
        f"{', '.join(_CREDENTIAL_ENV_NAMES)}, and no value is ever printed. A true "
        "presence flag means only that the variable is set and non-empty, not that a key "
        "is valid or authorised."
    )
    return CheckResult(
        name="model_providers",
        status="pass" if adapters else "warn",
        detail=detail,
        metadata=metadata,
    )


def run_checks(
    *,
    data_dir: str | Path | None = None,
    db: str | Path | None = None,
) -> list[CheckResult]:
    """Run every diagnostic check and return the list of results.

    This is the plain, importable entry point: it does no CLI work and can be
    called directly from tests. It creates no directory or database, never
    changes the policy mode, and never performs or approves any action.

    Args:
        data_dir: Optional explicit data directory, overriding resolution.
        db: Optional explicit database file path, overriding resolution.

    Returns:
        One :class:`CheckResult` per check, in reporting order.
    """
    resolved_data_dir = resolve_data_dir(data_dir)
    resolved_db_path = resolve_db_path(db, data_dir=resolved_data_dir)
    try:
        retention_days: Any = retention_days_from_env(DEFAULT_RETENTION_DAYS)
    except ValueError as exc:
        retention_check = CheckResult(
            name="retention",
            status="fail",
            detail=(
                "CYBERBRAIN_RETENTION_DAYS is malformed and was rejected fail-closed: "
                f"{exc} The store requires an explicit positive number of days."
            ),
            metadata={
                "retention_days": None,
                "env_var": "CYBERBRAIN_RETENTION_DAYS",
                "default": DEFAULT_RETENTION_DAYS,
            },
        )
        effective_retention = DEFAULT_RETENTION_DAYS
    else:
        retention_check = _check_retention(retention_days)
        effective_retention = (
            int(retention_days) if isinstance(retention_days, int) else DEFAULT_RETENTION_DAYS
        )
    policy_engine = PolicyEngine()
    checks: list[CheckResult] = [
        _check_python_version(),
        _check_package_version(),
        _check_platform_supported(),
        _check_data_dir(resolved_data_dir),
        _check_database(resolved_db_path),
        retention_check,
        _check_bubblewrap(),
        _check_sandbox_limits(),
        _check_workspaces_root(resolved_data_dir),
        _check_policy_mode(),
        _check_agent_scopes(),
        _check_pending_approvals(resolved_db_path, effective_retention),
        _check_quality_gates(policy_engine, effective_retention),
        _check_model_providers(),
    ]
    return checks


def build_report(
    *,
    data_dir: str | Path | None = None,
    db: str | Path | None = None,
) -> DoctorReport:
    """Run every diagnostic check and return the aggregated report.

    Args:
        data_dir: Optional explicit data directory, overriding resolution.
        db: Optional explicit database file path, overriding resolution.

    Returns:
        A :class:`DoctorReport` wrapping the rows from :func:`run_checks`.
    """
    return DoctorReport(checks=run_checks(data_dir=data_dir, db=db))


def render_text(report: DoctorReport, *, strict: bool = False) -> str:
    """Render a report as a plain-text table.

    Args:
        report: The report to render.
        strict: Whether a ``warn`` row counts as a failure.

    Returns:
        A newline-separated text block ready for printing.
    """
    lines: list[str] = [
        "CyberBrain26 doctor - read-only diagnostics",
        "(doctor changes nothing, performs no action, and never approves anything)",
        "",
    ]
    width = max((len(check.name) for check in report.checks), default=4)
    for check in report.checks:
        lines.append(f"{check.status.upper():<5} {check.name:<{width}}  {check.detail}")
    lines.append("")
    lines.append(
        f"Overall: {report.overall_status(strict=strict).upper()} "
        f"({report.failed} failed, {report.warnings} warnings, {len(report.checks)} checks, "
        f"strict={'on' if strict else 'off'})"
    )
    lines.append("")
    lines.append("Notes:")
    for note in list(report.notes) + [SECRETS_NOTE]:
        lines.append(f"- {note}")
    return "\n".join(lines)


doctor_app = typer.Typer(
    name="doctor",
    help="Check the local installation and report the effective security mode.",
    no_args_is_help=True,
)


@doctor_app.command("check")
def check(
    data_dir: str = typer.Option(None, "--data-dir", help="Explicit data directory to inspect."),
    db: str = typer.Option(None, "--db", help="Explicit SQLite database path to inspect."),
    strict: bool = typer.Option(False, "--strict", help="Treat WARN as FAIL."),
    as_json: bool = typer.Option(False, "--json", help="Emit the report as JSON."),
) -> None:
    """Run the read-only installation checks and report the effective security mode.

    Exits with code 1 when any check fails, or when ``--strict`` is given and any
    check warns. It never creates files and never changes the policy mode.
    """
    report = build_report(data_dir=data_dir, db=db)
    if as_json:
        typer.echo(json.dumps(report.to_dict(strict=strict), indent=2))
    else:
        typer.echo(render_text(report, strict=strict))
    if report.overall_status(strict=strict) == "fail":
        raise typer.Exit(code=1)


@doctor_app.command("main")
def main(
    data_dir: str = typer.Option(None, "--data-dir", help="Explicit data directory to inspect."),
    db: str = typer.Option(None, "--db", help="Explicit SQLite database path to inspect."),
    strict: bool = typer.Option(False, "--strict", help="Treat WARN as FAIL."),
    as_json: bool = typer.Option(False, "--json", help="Emit the report as JSON."),
) -> None:
    """Run the same read-only checks as ``check``; the alias used by the parent CLI."""
    check(data_dir=data_dir, db=db, strict=strict, as_json=as_json)


__all__ = [
    "BUBBLEWRAP_BINARY",
    "EXPECTED_DATA_DIR_MODE",
    "EXPECTED_DB_FILE_MODE",
    "EXPECTED_WORKSPACES_ROOT_MODE",
    "MIN_PYTHON_VERSION",
    "PROTOTYPE_NOTES",
    "SECRETS_NOTE",
    "CheckResult",
    "DoctorReport",
    "build_report",
    "check",
    "doctor_app",
    "main",
    "render_text",
    "run_checks",
]