"""diagnostics.py — Synchronous diagnostic checks for the Mugi backend.

All functions are blocking (Docker, MCP, filesystem) and should be called
via ``run_in_executor`` from async FastAPI handlers.
"""

from __future__ import annotations

import logging
import os
import shutil
import subprocess
import sys
import time
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from typing import Any

_log = logging.getLogger(__name__)


def check_python() -> dict:
    """Python interpreter health."""
    return {
        "name": "python",
        "ok": os.path.isfile(sys.executable),
        "detail": {
            "executable": sys.executable,
            "version": sys.version.split()[0],
        },
    }


def check_venv() -> dict:
    """Virtual environment reachability.

    The app launches the venv's python directly (no ``activate``), so
    ``VIRTUAL_ENV`` is often unset.  Fall back to inferring from
    ``sys.executable`` (e.g. ``~/.mugi/venv/bin/python3``).
    """
    venv = os.environ.get("VIRTUAL_ENV", "")
    if not venv:
        exe = os.path.abspath(sys.executable)
        parts = exe.split(os.sep)
        if "bin" in parts:
            venv = os.sep.join(parts[: parts.index("bin")])
    ok = bool(venv) and os.path.isdir(venv)
    return {
        "name": "venv",
        "ok": ok,
        "detail": {"path": venv or "(none)"},
    }


def check_workspace() -> dict:
    """Workspace directory existence and writability."""
    try:
        from core import get_workspace_path

        ws = get_workspace_path() or os.path.expanduser("~/.mugi/workspace")
    except Exception:
        ws = os.path.expanduser("~/.mugi/workspace")
    exists = os.path.isdir(ws)
    writable = os.access(ws, os.W_OK) if exists else False
    return {
        "name": "workspace",
        "ok": exists and writable,
        "detail": {"path": ws, "exists": exists, "writable": writable},
    }


def check_database() -> dict:
    """SQLite database readability (WAL mode, user_version)."""
    try:
        import sqlite3

        try:
            from chat_storage import DB_PATH as db_path
        except Exception:
            db_path = os.path.join(os.path.expanduser("~/.mugi"), "chats.db")
        if not os.path.isfile(db_path):
            return {"name": "database", "ok": False, "detail": {"error": "not found", "path": db_path}}
        conn = sqlite3.connect(db_path, timeout=5)
        journal = conn.execute("PRAGMA journal_mode").fetchone()[0]
        version = conn.execute("PRAGMA user_version").fetchone()[0]
        conn.close()
        return {
            "name": "database",
            "ok": True,
            "detail": {"path": db_path, "journal_mode": journal, "schema_version": version},
        }
    except Exception as e:
        return {"name": "database", "ok": False, "detail": {"error": str(e)}}


def check_docker() -> dict:
    """Docker daemon reachability."""
    docker = shutil.which("docker")
    if not docker:
        return {"name": "docker", "ok": False, "detail": {"error": "docker not found in PATH"}}
    try:
        result = subprocess.run(
            [docker, "info", "--format", "{{.ServerVersion}}"],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode == 0:
            return {"name": "docker", "ok": True, "detail": {"version": result.stdout.strip()}}
        return {"name": "docker", "ok": False, "detail": {"error": result.stderr.strip()[:200]}}
    except subprocess.TimeoutExpired:
        return {"name": "docker", "ok": False, "detail": {"error": "timeout"}}
    except Exception as e:
        return {"name": "docker", "ok": False, "detail": {"error": str(e)}}


def check_searxng() -> dict:
    """SearXNG container status."""
    try:
        from tools.searxng_docker import searxng_container_status

        status = searxng_container_status()
        return {
            "name": "searxng",
            "ok": status == "running",
            "detail": {"status": status},
        }
    except Exception as e:
        return {"name": "searxng", "ok": False, "detail": {"error": str(e)}}


def check_mcp() -> dict:
    """MCP bridge configuration summary."""
    try:
        from mcp_bridge.config import MCP_CONFIG_PATH, summarize_servers

        has_config = os.path.isfile(MCP_CONFIG_PATH)
        if not has_config:
            return {"name": "mcp", "ok": True, "detail": {"configured": False, "servers": 0}}
        summary = summarize_servers()
        server_count = len(summary) if isinstance(summary, (list, dict)) else 0
        return {
            "name": "mcp",
            "ok": True,
            "detail": {"configured": True, "servers": server_count, "config_path": MCP_CONFIG_PATH},
        }
    except Exception as e:
        return {"name": "mcp", "ok": False, "detail": {"error": str(e)}}


def check_encryption() -> dict:
    """Encryption key availability (Keychain)."""
    try:
        import keyring

        key = keyring.get_password("Mugi", "chat_encryption_key")
        return {
            "name": "encryption",
            "ok": key is not None and len(key) > 0,
            "detail": {"keychain_entry": "present" if key else "missing"},
        }
    except Exception as e:
        return {"name": "encryption", "ok": False, "detail": {"error": str(e)}}


def check_drive_discovery_config() -> dict:
    """Drive discovery readiness (model configured or main LLM fallback enabled)."""
    try:
        from core.config import get_drives_config

        cfg = get_drives_config() or {}
        enabled = bool(cfg.get("enabled", True))
        if not enabled:
            return {"name": "drive_discovery", "ok": True, "detail": {"enabled": False, "skipped": "drives disabled"}}

        has_model = bool(str(cfg.get("discovery_model") or "").strip())
        has_base_url = bool(str(cfg.get("base_url") or "").strip())
        use_main = bool(cfg.get("discovery_use_main_llm", False))
        ok = has_model or has_base_url or use_main
        return {
            "name": "drive_discovery",
            "ok": ok,
            "detail": {
                "enabled": True,
                "discovery_model_configured": has_model,
                "base_url_configured": has_base_url,
                "discovery_use_main_llm": use_main,
            },
        }
    except Exception as e:
        return {"name": "drive_discovery", "ok": False, "detail": {"error": str(e)}}


def _classify_bridge_error(error_text: str, status_code: int | None = None) -> str:
    if status_code == 401:
        return "401"
    text = (error_text or "").lower()
    if "connection refused" in text or "errno 61" in text:
        return "refused"
    if "timed out" in text or "timeout" in text:
        return "timeout"
    return "other"


def check_macos_bridge() -> dict:
    if sys.platform != "darwin":
        return {"name": "macos_bridge", "ok": True, "detail": {"skipped": "non-darwin"}}

    base_url = os.environ.get("MUGI_CALENDAR_URL", "").rstrip("/")
    if not base_url:
        return {"name": "macos_bridge", "ok": True, "detail": {"skipped": "MUGI_CALENDAR_URL not set"}}

    token = os.environ.get("MUGI_AUTH_TOKEN", "").strip()
    if not token:
        return {"name": "macos_bridge", "ok": True, "detail": {"skipped": "MUGI_AUTH_TOKEN not set"}}

    req = Request(
        f"{base_url}/health",
        headers={"X-Auth-Token": token},
        method="GET",
    )
    try:
        with urlopen(req, timeout=5) as resp:
            ok = int(resp.getcode() or 0) == 200
            return {
                "name": "macos_bridge",
                "ok": ok,
                "detail": {"status_code": int(resp.getcode() or 0)},
            }
    except HTTPError as e:
        cls = _classify_bridge_error("", status_code=int(getattr(e, "code", 0) or 0))
        return {
            "name": "macos_bridge",
            "ok": False,
            "detail": {"status_code": int(getattr(e, "code", 0) or 0), "class": cls},
        }
    except (URLError, OSError) as e:
        cls = _classify_bridge_error(str(e))
        return {"name": "macos_bridge", "ok": False, "detail": {"error": str(e), "class": cls}}
    except Exception as e:
        cls = _classify_bridge_error(str(e))
        return {"name": "macos_bridge", "ok": False, "detail": {"error": str(e), "class": cls}}


def check_kanban_salvage_slo() -> dict:
    """Salvage SLO: harness-synthesized/salvaged handoffs should be a rare backup, not the
    norm. Flags when the salvage rate over recent terminal kanban runs exceeds the configured
    ceiling. Requires a minimum sample so a fresh install (e.g. 1/1 salvaged) does not
    false-alarm. This probe measures over-reliance on the fallbacks; it never disables them.
    """
    try:
        from core.config import get_kanban_config
        import kanban_storage as ks

        cfg = get_kanban_config() or {}
        if not bool(cfg.get("enabled", True)):
            return {"name": "kanban_salvage_slo", "ok": True, "detail": {"skipped": "kanban disabled"}}

        def _num(value: object, default: float, cast):
            # Non-lossy coercion: a legitimate 0 (e.g. salvage_slo_max=0 -> "flag any salvage")
            # must survive. ``x or default`` would silently clobber it, defeating operator intent.
            try:
                return cast(value)
            except (TypeError, ValueError):
                return cast(default)

        window = _num(cfg.get("salvage_stats_window", 200), 200, int)
        max_rate = _num(cfg.get("salvage_slo_max", 0.25), 0.25, float)
        min_sample = _num(cfg.get("salvage_slo_min_sample", 20), 20, int)
        stats = ks.salvage_stats(window=window)
        total = int(stats.get("terminal_runs") or 0)
        rate = float(stats.get("salvage_rate") or 0.0)
        detail = {
            "salvage_rate": rate,
            "salvage_slo_max": max_rate,
            "terminal_runs": total,
            "min_sample": min_sample,
            "by_source": stats.get("by_source") or {},
        }
        if total < min_sample:
            detail["skipped"] = "insufficient_sample"
            return {"name": "kanban_salvage_slo", "ok": True, "detail": detail}
        if rate > max_rate:
            detail["error"] = (
                f"salvage rate {rate:.0%} exceeds SLO {max_rate:.0%} over last {total} runs "
                "(harness is reconstructing handoffs too often — investigate the root cause, "
                "do not raise the ceiling)"
            )
            return {"name": "kanban_salvage_slo", "ok": False, "detail": detail}
        return {"name": "kanban_salvage_slo", "ok": True, "detail": detail}
    except Exception as e:
        return {"name": "kanban_salvage_slo", "ok": False, "detail": {"error": str(e)}}


_ALL_CHECKS = [
    check_python,
    check_venv,
    check_workspace,
    check_database,
    check_docker,
    check_searxng,
    check_mcp,
    check_encryption,
    check_drive_discovery_config,
    check_macos_bridge,
    check_kanban_salvage_slo,
]


def run_diagnostics() -> dict:
    """Run all diagnostic checks. Returns a summary dict.

    This function performs blocking I/O and must NOT be called directly
    from an async handler — use ``asyncio.get_event_loop().run_in_executor``.
    """
    t0 = time.monotonic()
    results = []
    all_ok = True
    for fn in _ALL_CHECKS:
        try:
            result = fn()
        except Exception as e:
            result = {"name": fn.__name__.replace("check_", ""), "ok": False, "detail": {"error": str(e)}}
        results.append(result)
        if not result.get("ok"):
            all_ok = False
    elapsed_ms = round((time.monotonic() - t0) * 1000)
    return {
        "ok": all_ok,
        "checks": results,
        "elapsed_ms": elapsed_ms,
    }


# ---------------------------------------------------------------------------
# Doctor report — stable, versioned, redacted JSON for external consumers
# ---------------------------------------------------------------------------
_DOCTOR_SCHEMA_VERSION = 1

_REDACT_KEYS = frozenset({
    "api_key", "token", "password", "secret", "chat_encryption_key",
    "discord_token", "imap_password", "auth_token",
})


def _redact_value(key: str, value: object) -> object:
    """Redact sensitive values by key name."""
    if isinstance(value, str) and any(s in key.lower() for s in _REDACT_KEYS):
        return "***" if value else ""
    if isinstance(value, dict):
        return {k: _redact_value(k, v) for k, v in value.items()}
    if isinstance(value, list):
        return [_redact_value(key, v) for v in value]
    return value


def _build_autonomy_posture(app_prefs: dict) -> dict:
    """Summarize the solo-dev autonomy surface for the doctor report.

    Reads webhook source capabilities, file-watcher rules, outbound hook
    enablement, shell/plan-mode prefs, and best-effort background-runner
    status.  Returns ``{"level": "ok"|"warn", "notes": [...], "signals": {...}}``
    with short, actionable notes so Diagnostics can flag risky combinations
    (e.g. ``full_turn`` webhooks + shell enabled, or webhooks on but no
    background agent running to wake them).  Never raises — every probe is
    wrapped so a missing module does not break the report.
    """
    notes: list[str] = []
    signals: dict = {
        "plan_mode_enabled": bool(app_prefs.get("plan_mode_enabled", False)),
        "allow_shell_execute": bool(app_prefs.get("allow_shell_execute", False)),
        "webhooks_enabled": bool(app_prefs.get("webhooks_enabled", False)),
    }

    webhook_sources: list[dict] = []
    try:
        from persistence import webhook_sources_list
        webhook_sources = [
            {"source": str(r.get("source") or ""), "capability": str(r.get("capability") or "full_turn")}
            for r in (webhook_sources_list() or [])
            if r.get("source")
        ]
    except Exception:
        webhook_sources = []
    full_turn_sources = [r["source"] for r in webhook_sources if r.get("capability") == "full_turn"]
    notify_only_sources = [r["source"] for r in webhook_sources if r.get("capability") == "notify_only"]
    signals["webhook_sources"] = {
        "full_turn": full_turn_sources,
        "notify_only": notify_only_sources,
    }

    watcher_rules_full_turn: list[str] = []
    watcher_enabled = False
    try:
        from utils.file_watcher import load_rules_from_config, is_enabled as _fw_enabled
        watcher_enabled = bool(_fw_enabled())
        for rule in (load_rules_from_config() or []):
            if getattr(rule, "capability", "notify_only") == "full_turn":
                watcher_rules_full_turn.append(getattr(rule, "path", ""))
    except Exception:
        pass
    signals["file_watcher"] = {
        "enabled": watcher_enabled,
        "full_turn_rule_paths": watcher_rules_full_turn,
    }

    outbound_slugs: list[str] = []
    outbound_enabled = False
    try:
        from core.config import get_outbound_webhooks_config
        outbound_enabled = bool((get_outbound_webhooks_config() or {}).get("enabled", False))
    except Exception:
        outbound_enabled = False
    try:
        from persistence import outbound_webhooks_list
        outbound_slugs = [str(r.get("slug") or "") for r in (outbound_webhooks_list() or []) if r.get("slug")]
    except Exception:
        outbound_slugs = []
    signals["outbound_hooks"] = {
        "enabled": outbound_enabled,
        "registered_slugs": outbound_slugs,
    }

    # Background runner — best-effort socket ping; None means "could not detect".
    runner_installed: bool | None = None
    runner_running: bool | None = None
    try:
        from app_background_agent import (
            _background_agent_is_installed as _installed,
            _background_agent_ping as _ping,
        )
        runner_installed = bool(_installed())
        runner_running = bool(_ping())
    except Exception:
        runner_installed = None
        runner_running = None
    signals["background_runner"] = {
        "installed": runner_installed,
        "running": runner_running,
    }

    level = "ok"

    def _warn(msg: str) -> None:
        nonlocal level
        notes.append(msg)
        level = "warn"

    if full_turn_sources and signals["allow_shell_execute"]:
        _warn(
            "Inbound webhook source(s) "
            + ", ".join(f"`{s}`" for s in full_turn_sources)
            + " are `full_turn` and shell is enabled — a signed request can run shell commands. "
            "Consider switching these sources to `notify_only`."
        )
    if full_turn_sources and signals["plan_mode_enabled"]:
        _warn(
            "Plan mode is enabled but `full_turn` webhook sources can still wake the runner; "
            "the awoken turn will run under plan-mode restrictions. Confirm this is intended."
        )
    if signals["webhooks_enabled"] and full_turn_sources:
        if runner_running is False:
            _warn(
                "`full_turn` webhook sources are configured but the background runner is not responding. "
                "Start the background agent (Preferences → Advanced) or verify `com.mugi.launchd.runner` LaunchAgent is loaded."
            )
        elif runner_running is None and runner_installed is False:
            _warn(
                "`full_turn` webhook sources are configured but the background agent LaunchAgent is not installed. "
                "Install it from Preferences → Advanced so wakes can actually run."
            )
    if watcher_rules_full_turn and signals["allow_shell_execute"]:
        _warn(
            "File watcher has `full_turn` rules and shell is enabled — any matched file change can execute shell. "
            "Prefer `notify_only` rules unless you explicitly need autonomous response."
        )
    if outbound_enabled and outbound_slugs:
        notes.append(
            "Outbound webhooks are enabled with registered slugs: "
            + ", ".join(f"`{s}`" for s in outbound_slugs)
            + ". Review the targets in Preferences → Advanced."
        )
    if signals["allow_shell_execute"] and signals["plan_mode_enabled"]:
        notes.append(
            "Shell is enabled while plan mode is active — plan mode still blocks shell-like tools "
            "during planning turns."
        )

    drives_autonomy_mode = "bounded"
    try:
        from drive_engine import _get_drives_config

        drives_autonomy_mode = str(_get_drives_config().get("autonomy_mode", "bounded")).strip().lower()
    except Exception:
        drives_autonomy_mode = "bounded"
    signals["drives_autonomy_mode"] = drives_autonomy_mode
    if drives_autonomy_mode == "aggressive":
        _warn(
            "[drives].autonomy_mode is 'aggressive': drive_action_cycle may execute tool-using "
            "anticipatory work without prior observation approval. Prefer 'bounded' (default) "
            "for consent-gated autonomy."
        )

    return {"level": level, "notes": notes, "signals": signals}


def _build_profile_health() -> dict:
    """Lightweight checks on workspace profile markdown files."""
    out: dict = {"ok": True, "issues": []}
    try:
        from core import get_config, get_workspace_path
        from core.config import get_agent_display_name, parse_identity_name_from_markdown
        from persistence import _identity_file_has_real_content

        ws = get_workspace_path()
        cfg = get_config() or {}
        onboarding_done = bool((cfg.get("onboarding") or {}).get("done"))
        for name in ("PERSONA.md", "IDENTITY.md", "USER.md"):
            path = os.path.join(ws, name)
            if not os.path.isfile(path):
                continue
            with open(path, "r", encoding="utf-8", errors="replace") as f:
                head = f.read(256)
            first = head.lstrip().split("\n", 1)[0].strip().lower()
            if name.upper() == "IDENTITY.MD" and first.startswith("# user.md"):
                out["issues"].append("IDENTITY.md cross-contaminated with USER header")
            if name.upper() == "PERSONA.MD" and first.startswith("# user.md"):
                out["issues"].append("PERSONA.md cross-contaminated with USER header")
            if name.upper() == "USER.MD" and (
                first.startswith("# identity") or first.startswith("# persona")
            ):
                out["issues"].append("USER.md cross-contaminated with persona header")
        display = (get_agent_display_name() or "").strip()
        if onboarding_done and display.lower() == "mugi":
            out["issues"].append("Assistant display name is 'Mugi' (product name) after onboarding")
        persona_path = os.path.join(ws, "PERSONA.md")
        if os.path.isfile(persona_path):
            with open(persona_path, "r", encoding="utf-8", errors="replace") as f:
                pname = parse_identity_name_from_markdown(f.read(4096))
            out["persona_name_set"] = bool(pname)
        user_path = os.path.join(ws, "USER.md")
        out["user_profile_populated"] = (
            os.path.isfile(user_path) and _identity_file_has_real_content(user_path)
        )
        inject_files = (cfg.get("personality_files") or {}).get("files") or []
        if isinstance(inject_files, list) and any(
            str(f or "").strip().upper() == "MEMORY.MD" for f in inject_files
        ):
            out["issues"].append(
                "MEMORY.md is still in personality_files.files — remove it; use Active Memory and search_memory"
            )
        agents_path = os.path.join(ws, "AGENTS.md")
        if os.path.isfile(agents_path):
            try:
                with open(agents_path, "r", encoding="utf-8", errors="replace") as f:
                    agents_text = f.read(8192)
            except OSError:
                agents_text = ""
            if agents_text.strip():
                from utils.profile_markdown import extract_markdown_section

                standing = extract_markdown_section(agents_text, "Standing Orders", level=2)
                if not standing.strip():
                    out["issues"].append(
                        "AGENTS.md has no ## Standing Orders section — lazy inject falls back to capped preamble"
                    )
        if out["issues"]:
            out["ok"] = False
    except Exception as exc:
        out = {"ok": False, "issues": [str(exc)[:200]]}
    return out


def build_doctor_report() -> dict:
    """Build a comprehensive doctor report for Mugi.

    Returns a stable-schema JSON dict with ``schema_version`` at root.
    All sensitive values are redacted.  Safe for HTTP responses and MCP tools.
    """
    diag = run_diagnostics()
    warnings: list[str] = []

    # Collect failing checks into warnings.
    for check in diag.get("checks", []):
        if not check.get("ok"):
            detail = check.get("detail", {})
            err = detail.get("error", "unknown")
            warnings.append(f"{check.get('name', '?')}: {err}")

    # App prefs summary (redacted).
    app_prefs_summary: dict = {}
    try:
        from utils.app_prefs_store import load_app_prefs_cached
        raw_prefs = load_app_prefs_cached()
        app_prefs_summary = {k: _redact_value(k, v) for k, v in raw_prefs.items()}
    except Exception:
        app_prefs_summary = {"error": "could not load"}

    # Plan mode status
    plan_mode = app_prefs_summary.get("plan_mode_enabled", False)

    # Bridge info (loopback + token present only).
    bridge_url = os.environ.get("MUGI_CALENDAR_URL", "")
    bridge_info = {
        "url_set": bool(bridge_url),
        "token_present": bool(os.environ.get("MUGI_AUTH_TOKEN", "")),
    }
    if bridge_url:
        from urllib.parse import urlparse
        parsed = urlparse(bridge_url)
        bridge_info["loopback"] = parsed.hostname in ("localhost", "127.0.0.1", "::1")

    # MLX status — only check if already imported (avoids heavy native deps).
    mlx_info: dict = {"available": False}
    if "app_mlx" in sys.modules:
        try:
            _fn = getattr(sys.modules["app_mlx"], "mlx_server_running", None)
            if _fn:
                mlx_info = {"server_running": _fn(), "available": True}
        except Exception:
            pass

    # Mesh status — only check if already imported (avoids native dep segfaults).
    mesh_info: dict = {"available": False}
    if "mesh" in sys.modules:
        try:
            _fn = getattr(sys.modules["mesh"], "get_mesh_manager", None)
            if _fn:
                mgr = _fn()
                mesh_info = {"listener_bound": mgr is not None, "available": True}
        except Exception:
            pass

    # Emergency trim note.
    warnings.append(
        "Context emergency trim is lossy by design; aggressive trim can drop conversation history."
    )

    autonomy_posture = _build_autonomy_posture(app_prefs_summary)

    # Catalog of human-readable hints for ``skip_reason`` codes recorded in the
    # task ledger.  Static strings; safe to ship with doctor.
    try:
        from utils.skip_reason_catalog import full_catalog as _skip_catalog
        skip_reason_catalog = _skip_catalog()
    except Exception:
        skip_reason_catalog = {}

    memory_surface: dict[str, Any] = {}
    try:
        from core import get_workspace_path
        from utils.memory_shard_layout import list_shard_abs_paths
        from utils.memory_scopes import list_mesh_safe_memory_files

        _ws = get_workspace_path()

        memory_surface["workspace_configured"] = bool((_ws or "").strip())
        _insights = os.path.join(_ws, "memory", "insights.md")
        memory_surface["insights_present"] = os.path.isfile(_insights)
        try:
            memory_surface["shard_file_count"] = len(list_shard_abs_paths(_ws))
        except Exception:
            memory_surface["shard_file_count"] = -1
        memory_surface["mesh_safe_note_count"] = len(list_mesh_safe_memory_files(_ws))
    except Exception as e:
        memory_surface = {"error": str(e)[:200]}

    knowledge_surface: dict[str, Any] = {}
    try:
        from tools.knowledge_tools import list_knowledge_corpora

        corpora = list_knowledge_corpora()
        failed = [c["slug"] for c in corpora if str(c.get("digest_status") or "").lower() == "failed"]
        running = [c["slug"] for c in corpora if str(c.get("digest_status") or "").lower() == "running"]
        partial = [c["slug"] for c in corpora if str(c.get("digest_status") or "").lower() == "partial"]
        stale_slugs: list[str] = []
        for c in corpora:
            slug = str(c.get("slug") or "")
            if not slug:
                continue
            try:
                from tools.knowledge_tools import check_corpus_staleness

                st = check_corpus_staleness(slug)
                if st.get("ok") and int(st.get("stale_count") or 0) > 0:
                    stale_slugs.append(slug)
            except Exception:
                continue
        search_index_hint = ""
        try:
            from tools.knowledge_tools import list_knowledge_corpora, search_knowledge_hits

            if corpora:
                probe = search_knowledge_hits("knowledge index health", max_results=1, semantic=False)
                if not probe and any(
                    str(c.get("digest_status") or "").lower() == "complete" for c in corpora
                ):
                    search_index_hint = (
                        "knowledge search index may be empty — re-digest a corpus or check session_search_db"
                    )
        except Exception as probe_exc:
            search_index_hint = f"knowledge search probe failed: {str(probe_exc)[:120]}"
        gitnexus_operator_note = (
            "Repo call graphs and impact analysis are via GitNexus MCP (not Knowledge Base). "
            "Run `npx gitnexus analyze` on indexed repos; configure servers in ~/.mugi/mcp_servers.json."
        )
        knowledge_surface = {
            "corpus_count": len(corpora),
            "failed_corpora": failed,
            "running_corpora": running,
            "partial_corpora": partial,
            "stale_corpora": stale_slugs,
            "search_index_hint": search_index_hint,
            "gitnexus_operator_note": gitnexus_operator_note,
        }
        if failed:
            warnings.append(f"knowledge: {len(failed)} corpus(es) in failed state")
        if stale_slugs:
            warnings.append(f"knowledge: {len(stale_slugs)} corpus(es) have changed source files (re-digest recommended)")
    except Exception as e:
        knowledge_surface = {"error": str(e)[:200]}

    web_act_surface: dict[str, Any] = {}
    try:
        from core.config import get_web_act_config
        from utils.playwright_runner import playwright_available, playwright_browsers_ready

        cfg = get_web_act_config()
        pref_on = bool(app_prefs_summary.get("allow_web_act_scripts"))
        pw_pkg = playwright_available()
        pw_browsers, pw_browsers_detail = playwright_browsers_ready() if pw_pkg else (False, "")
        cfg_on = bool(cfg.get("enabled", True))
        require_pw = bool(cfg.get("require_playwright"))
        pw_ready = pw_pkg and pw_browsers if require_pw else pw_pkg
        web_act_surface = {
            "pref_allow_web_act_scripts": pref_on,
            "config_enabled": cfg_on,
            "playwright_installed": pw_pkg,
            "playwright_browsers_ready": pw_browsers,
            "tools_available": pref_on and cfg_on,
            "ready": pref_on and cfg_on and (pw_ready or not require_pw),
        }
        try:
            from utils.web_act_retention import measure_web_act_disk_usage, prune_web_act_scopes

            disk = measure_web_act_disk_usage()
            web_act_surface.update(disk)
            dry = prune_web_act_scopes(dry_run=True)
            web_act_surface["retention_would_reclaim_bytes"] = int(dry.get("deleted_bytes") or 0)
            max_scope = int(cfg.get("max_scope_bytes") or 0)
            largest_bytes = int(disk.get("web_runs_largest_scope_bytes") or 0)
            if max_scope > 0 and largest_bytes > max_scope:
                largest_name = disk.get("web_runs_largest_scope") or "?"
                warnings.append(
                    f"web_act: largest scope {largest_name!r} ({largest_bytes} bytes) exceeds "
                    f"per-scope max_scope_bytes ({max_scope})"
                )
        except Exception:
            _log.debug("web_act disk measure failed", exc_info=True)
        if pw_pkg and not pw_browsers and pw_browsers_detail:
            web_act_surface["playwright_browsers_detail"] = pw_browsers_detail[:300]
        if pref_on and cfg_on and require_pw and not pw_pkg:
            warnings.append("web_act: pref enabled but Playwright venv not installed")
        elif pref_on and cfg_on and require_pw and pw_pkg and not pw_browsers:
            warnings.append("web_act: Playwright package installed but browser binaries missing")
    except Exception as e:
        web_act_surface = {"error": str(e)[:200]}

    profile_health = _build_profile_health()

    report = {
        "schema_version": _DOCTOR_SCHEMA_VERSION,
        "ok": diag["ok"],
        "plan_mode_enabled": plan_mode,
        "checks": diag["checks"],
        "bridge": bridge_info,
        "mlx": mlx_info,
        "mesh": mesh_info,
        "app_prefs": app_prefs_summary,
        "autonomy_posture": autonomy_posture,
        "skip_reason_catalog": skip_reason_catalog,
        "memory_surface": memory_surface,
        "knowledge_surface": knowledge_surface,
        "web_act_surface": web_act_surface,
        "profile_health": profile_health,
        "warnings": warnings,
        "elapsed_ms": diag["elapsed_ms"],
    }
    return {k: _redact_value(k, v) for k, v in report.items()}
