Gitea-Tools/webui/system_health.py

"""Read-only system-health model for the operator console API (#634).

`/health` answers liveness only. Operators automating readiness checks need a
structured view of *why* the control plane is or is not usable: which
dependencies answered, how long they took, what version of the code is running,
and whether the runtime is stale relative to its remote.

Three rules shape this module.

* **Read-only.** Every probe opens its subject read-only. The control-plane
  database is opened through a ``mode=ro`` URI so a health check can never
  create or migrate a schema, and no probe writes, restarts, or reloads
  anything — restart controls are Phase 2, and #630 forbids process-kill
  recovery outright.
* **Fail-soft.** A dependency that is unreachable is a *status*, not an
  exception. Probes catch their own failures and report them as a degraded or
  down entry carrying a reason.
* **Never claim more than was proven.** Readiness is derived only from probes
  that actually ran, ``mutation_safe`` stays false unless the parity commits are
  known and equal, and an MCP namespace is reported unproven because a web
  process cannot exercise the IDE-managed client path (#543).
"""

from __future__ import annotations

import os
import re
import sqlite3
import subprocess
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable
from urllib.parse import urlsplit, urlunsplit

import control_plane_db
import mcp_namespace_health
from gitea_auth import api_request, get_auth_header, gitea_url

from webui.project_registry import load_registry

SERVICE_NAME = "mcp-control-plane-webui"
API_PATH = "/api/v1/system/health"

STATUS_OK = "ok"
STATUS_DEGRADED = "degraded"
STATUS_DOWN = "down"
STATUS_SKIPPED = "skipped"
STATUS_UNPROVEN = "unproven"

# Statuses that count as a healthy answer from a probe.
_HEALTHY_STATUSES = frozenset({STATUS_OK})
# Statuses meaning "this probe did not run", as opposed to "it ran and failed".
_NOT_RUN_STATUSES = frozenset({STATUS_SKIPPED})

_DEEP_PROBE_TTL_ENV = "WEBUI_HEALTH_PROBE_TTL_SECONDS"
_DEFAULT_DEEP_PROBE_TTL = 15.0
_GITEA_PROBE_TIMEOUT_SECONDS = 5.0

_OFFLINE_ENV = "WEBUI_TEST_OFFLINE"

# Credential-shaped material that must never reach the browser, mirroring the
# forbidden client patterns in webui/deployment_boundary.py.
_SECRET_RE = re.compile(
    r"(?i)\b(token|password|passwd|secret|authorization|bearer)\b\s*[:=]?\s*\S+"
)
_LONG_OPAQUE_RE = re.compile(r"\b[A-Za-z0-9_\-]{32,}\b")

# Captured once at import so uptime measures this process, not the request.
_STARTED_AT = datetime.now(timezone.utc)
_STARTED_MONOTONIC = time.monotonic()

# TTL cache for the expensive (network) probe only.
_deep_cache: dict[str, tuple[float, "DependencyProbe"]] = {}


@dataclass(frozen=True)
class DependencyProbe:
    """One dependency check, fail-soft, with its own latency."""

    name: str
    kind: str
    status: str
    detail: str
    required: bool
    latency_ms: float | None = None
    metadata: dict[str, Any] | None = None

    @property
    def healthy(self) -> bool:
        return self.status in _HEALTHY_STATUSES

    @property
    def ran(self) -> bool:
        return self.status not in _NOT_RUN_STATUSES


@dataclass(frozen=True)
class VersionInfo:
    git_sha: str | None
    git_describe: str | None
    control_plane_schema_version: int | None
    python_version: str
    known: bool


@dataclass(frozen=True)
class StaleRuntime:
    """Parity between the running code, the checkout, and the remote.

    ``mutation_safe`` is deliberately conservative: unknown is not safe.
    """

    daemon_head: str | None
    checkout_head: str | None
    remote_head: str | None
    stale: bool
    determinable: bool
    mutation_safe: bool
    reasons: tuple[str, ...]


@dataclass(frozen=True)
class SystemHealthSnapshot:
    status: str
    ready: bool
    readiness_complete: bool
    readiness_reasons: tuple[str, ...]
    service: str
    mode: str
    version: VersionInfo
    started_at: str
    uptime_seconds: float
    timestamp: str
    deep_probes_requested: bool
    dependencies: tuple[DependencyProbe, ...]
    mcp_namespaces: tuple[dict[str, Any], ...]
    stale_runtime: StaleRuntime
    probe_errors: tuple[str, ...] = ()


def process_uptime() -> tuple[str, float]:
    """Process start timestamp and uptime — in-memory, safe for `/health`."""
    return _STARTED_AT.isoformat(), round(time.monotonic() - _STARTED_MONOTONIC, 3)


def _offline() -> bool:
    return (os.environ.get(_OFFLINE_ENV) or "").strip().lower() in {"1", "true", "yes"}


def _repo_root() -> Path:
    override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
    if override:
        return Path(override).resolve()
    return Path(__file__).resolve().parent.parent


def _deep_probe_ttl() -> float:
    raw = (os.environ.get(_DEEP_PROBE_TTL_ENV) or "").strip()
    if not raw:
        return _DEFAULT_DEEP_PROBE_TTL
    try:
        value = float(raw)
    except ValueError:
        return _DEFAULT_DEEP_PROBE_TTL
    return value if value >= 0 else _DEFAULT_DEEP_PROBE_TTL


def redact(text: str) -> str:
    """Strip credential-shaped material from operator-visible probe text.

    Probe details carry exception strings, and an exception raised by an HTTP
    client can quote the request that failed. Redaction happens here, at the
    boundary where those strings become part of a browser-bound payload.
    """
    if not text:
        return ""
    cleaned = _redact_urls(text)
    cleaned = _SECRET_RE.sub(lambda m: f"{m.group(1)}=[redacted]", cleaned)
    return _LONG_OPAQUE_RE.sub("[redacted]", cleaned)


def _redact_urls(text: str) -> str:
    return re.sub(r"https?://\S+", lambda m: redact_url(m.group(0)), text)


def redact_url(url: str) -> str:
    """Reduce a URL to scheme://host/path — no userinfo, no query, no fragment."""
    try:
        parts = urlsplit(url)
    except ValueError:
        return "[redacted-url]"
    if not parts.scheme or not parts.hostname:
        return "[redacted-url]"
    netloc = parts.hostname
    if parts.port:
        netloc = f"{netloc}:{parts.port}"
    return urlunsplit((parts.scheme, netloc, parts.path, "", ""))


def _git(repo: Path, *args: str) -> str | None:
    try:
        completed = subprocess.run(
            ["git", "-C", str(repo), *args],
            capture_output=True,
            text=True,
            check=False,
            timeout=10,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    if completed.returncode != 0:
        return None
    return (completed.stdout or "").strip() or None


def _load_version(repo: Path, *, schema_version: int | None) -> VersionInfo:
    import platform

    git_sha = None if _offline() else _git(repo, "rev-parse", "HEAD")
    describe = None if _offline() else _git(repo, "describe", "--tags", "--always")
    return VersionInfo(
        git_sha=git_sha,
        git_describe=describe,
        control_plane_schema_version=schema_version,
        python_version=platform.python_version(),
        known=bool(git_sha),
    )


# ---------------------------------------------------------------------------
# Dependency probes
# ---------------------------------------------------------------------------


def _elapsed_ms(started: float) -> float:
    return round((time.monotonic() - started) * 1000, 3)


def probe_control_plane_db(db_path: str | None = None) -> DependencyProbe:
    """Read-only reachability check for the control-plane SQLite substrate.

    Opened through a ``mode=ro`` URI on purpose: ``ControlPlaneDB.__init__``
    creates directories and runs schema migrations, which a health check must
    never do.
    """
    path = (db_path or control_plane_db.default_db_path()).strip()
    started = time.monotonic()
    metadata: dict[str, Any] = {"path": path}

    def _result(status: str, detail: str) -> DependencyProbe:
        return DependencyProbe(
            name="control_plane_db",
            kind="sqlite",
            status=status,
            detail=detail,
            required=True,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )

    if not path or not os.path.exists(path):
        return _result(STATUS_DOWN, "control-plane database file does not exist yet")
    try:
        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5)
        try:
            row = conn.execute(
                "SELECT value FROM schema_meta WHERE key = 'schema_version'"
            ).fetchone()
            leases = conn.execute(
                "SELECT COUNT(*) FROM leases WHERE status = 'active'"
            ).fetchone()
        finally:
            conn.close()
    except sqlite3.Error as exc:
        return _result(STATUS_DOWN, redact(f"control-plane database unreadable: {exc}"))

    schema_version = int(row[0]) if row and str(row[0]).isdigit() else None
    metadata["schema_version"] = schema_version
    metadata["active_leases"] = int(leases[0]) if leases else None
    if schema_version is None:
        return _result(
            STATUS_DEGRADED, "control-plane database has no recorded schema version"
        )
    if schema_version != control_plane_db.SCHEMA_VERSION:
        return _result(
            STATUS_DEGRADED,
            f"control-plane schema version {schema_version} does not match the "
            f"version this code expects ({control_plane_db.SCHEMA_VERSION})",
        )
    return _result(STATUS_OK, f"schema v{schema_version} readable")


def probe_repository(repo: Path) -> DependencyProbe:
    """Local checkout reachability — required, cheap, no network."""
    started = time.monotonic()
    metadata: dict[str, Any] = {"repo_root": str(repo)}
    if _offline():
        return DependencyProbe(
            name="repository",
            kind="git",
            status=STATUS_SKIPPED,
            detail=f"{_OFFLINE_ENV} is set; git probe skipped",
            required=True,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )
    head = _git(repo, "rev-parse", "HEAD")
    if not head:
        return DependencyProbe(
            name="repository",
            kind="git",
            status=STATUS_DOWN,
            detail=f"HEAD could not be read at {repo}",
            required=True,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )
    branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD")
    metadata["head"] = head
    metadata["branch"] = branch
    return DependencyProbe(
        name="repository",
        kind="git",
        status=STATUS_OK,
        detail=f"checkout readable at {branch or 'detached HEAD'}",
        required=True,
        latency_ms=_elapsed_ms(started),
        metadata=metadata,
    )


def probe_gitea(host: str) -> DependencyProbe:
    """Live Gitea reachability. Expensive (network), so opt-in via ``deep``.

    Optional by design: the console stays useful for local inventory when the
    remote is unreachable, so a failure here degrades status without claiming
    the process itself is unready.
    """
    started = time.monotonic()
    metadata: dict[str, Any] = {"host": host}

    def _failure(status: str, detail: str) -> DependencyProbe:
        return DependencyProbe(
            name="gitea",
            kind="http",
            status=status,
            detail=detail,
            required=False,
            latency_ms=_elapsed_ms(started),
            metadata=metadata,
        )

    if not host:
        return _failure(STATUS_DEGRADED, "no Gitea host is configured in the registry")
    try:
        auth = get_auth_header(host)
    except Exception as exc:  # noqa: BLE001 — credential guards are a status here
        return _failure(STATUS_DEGRADED, redact(f"credential lookup refused: {exc}"))
    if not auth:
        return _failure(STATUS_DEGRADED, f"no credentials available for {host}")

    url = gitea_url(host, "/api/v1/version")
    metadata["endpoint"] = redact_url(url)
    try:
        data = api_request("GET", url, auth, timeout=_GITEA_PROBE_TIMEOUT_SECONDS)
    except Exception as exc:  # noqa: BLE001 — a down dependency is a status
        return _failure(STATUS_DOWN, redact(f"Gitea probe failed: {exc}"))

    if isinstance(data, dict) and data.get("version"):
        metadata["gitea_version"] = str(data["version"])
    return DependencyProbe(
        name="gitea",
        kind="http",
        status=STATUS_OK,
        detail=f"{host} reachable",
        required=False,
        latency_ms=_elapsed_ms(started),
        metadata=metadata,
    )


def _skipped_gitea(host: str) -> DependencyProbe:
    return DependencyProbe(
        name="gitea",
        kind="http",
        status=STATUS_SKIPPED,
        detail="network probe not requested; call with ?deep=1 to run it",
        required=False,
        latency_ms=None,
        metadata={"host": host},
    )


def namespace_summaries() -> tuple[dict[str, Any], ...]:
    """Declared MCP namespaces, each honestly reported as unproven.

    The web process runs outside the IDE-managed MCP client, so it cannot
    invoke a namespace tool. Per #543 only a ``client_namespace`` probe proves
    that path, and inventing a healthy verdict here is exactly the false claim
    the mutation gates exist to prevent.
    """
    rows: list[dict[str, Any]] = []
    for namespace, required_tool in sorted(
        mcp_namespace_health.REQUIRED_NAMESPACE_TOOLS.items()
    ):
        classification = mcp_namespace_health.classify_namespace_probe(
            namespace,
            required_tool=required_tool,
            probe_result=None,
            probe_source=mcp_namespace_health.PROBE_SOURCE_UNKNOWN,
        )
        rows.append(
            {
                "namespace": namespace,
                "required_tool": required_tool,
                "status": STATUS_UNPROVEN,
                "ide_namespace_proven": bool(classification.get("ide_namespace_proven")),
                "reason": (
                    "the web console cannot invoke the IDE-managed MCP client; "
                    "namespace health must be proven with a client_namespace "
                    "probe (#543)"
                ),
                "error_type": classification.get("error_type"),
            }
        )
    return tuple(rows)


def assess_stale_runtime(
    repo: Path,
    *,
    daemon_head: str | None = None,
    git_reader: Callable[..., str | None] | None = None,
) -> StaleRuntime:
    """Three-way parity view: running code, local checkout, remote-tracking ref.

    ``mutation_safe`` requires all three to be known and equal. Anything less —
    including "the remote ref was never fetched" — is reported as not safe with
    a reason, so an operator never reads an unproven green.
    """
    reader = git_reader or (lambda *args: _git(repo, *args))
    reasons: list[str] = []
    # The offline switch suppresses real subprocess calls; an explicitly
    # injected reader is already a substitute for them and is always used.
    offline = _offline() and git_reader is None

    checkout_head = None if offline else reader("rev-parse", "HEAD")
    remote_head = None if offline else reader("rev-parse", "@{upstream}")
    if offline:
        reasons.append(f"{_OFFLINE_ENV} is set; parity commits were not read")
    else:
        if checkout_head is None:
            reasons.append("local checkout HEAD could not be read")
        if remote_head is None:
            reasons.append(
                "no remote-tracking commit is known for the current branch; "
                "remote staleness is indeterminate (no fetch is performed here)"
            )

    effective_daemon = daemon_head if daemon_head is not None else checkout_head
    if daemon_head is None:
        reasons.append(
            "the running MCP daemon's startup commit is not observable from the "
            "web process; the checkout commit is reported in its place"
        )

    determinable = bool(checkout_head and remote_head and effective_daemon)
    stale = bool(
        determinable and len({checkout_head, remote_head, effective_daemon}) > 1
    )
    if stale:
        reasons.append(
            "runtime, checkout, and remote commits disagree; restart the MCP "
            "server after updating the checkout before trusting capability gates"
        )

    return StaleRuntime(
        daemon_head=effective_daemon,
        checkout_head=checkout_head,
        remote_head=remote_head,
        stale=stale,
        determinable=determinable,
        mutation_safe=bool(determinable and not stale),
        reasons=tuple(reasons),
    )


# ---------------------------------------------------------------------------
# Snapshot assembly
# ---------------------------------------------------------------------------


def _default_host() -> str:
    registry = load_registry()
    if not registry.projects:
        return ""
    raw = registry.projects[0].remote_host
    parts = urlsplit(raw.strip())
    return parts.netloc or raw.strip().rstrip("/")


def _aggregate(
    probes: tuple[DependencyProbe, ...],
) -> tuple[str, bool, bool, tuple[str, ...]]:
    """Fold probe results into overall status and readiness.

    Required probes drive readiness; optional probes can only degrade status.
    A probe that did not run leaves readiness incomplete rather than passing.
    """
    reasons: list[str] = []
    required = [probe for probe in probes if probe.required]
    unrun_required = [probe for probe in required if not probe.ran]
    failed_required = [probe for probe in required if probe.ran and not probe.healthy]
    failed_optional = [
        probe
        for probe in probes
        if not probe.required and probe.ran and not probe.healthy
    ]

    for probe in unrun_required:
        reasons.append(
            f"required dependency '{probe.name}' was not probed: {probe.detail}"
        )
    for probe in failed_required:
        reasons.append(
            f"required dependency '{probe.name}' is {probe.status}: {probe.detail}"
        )
    for probe in failed_optional:
        reasons.append(
            f"optional dependency '{probe.name}' is {probe.status}: {probe.detail}"
        )

    readiness_complete = not unrun_required
    ready = readiness_complete and not failed_required

    if any(probe.status == STATUS_DOWN for probe in failed_required):
        status = STATUS_DOWN
    elif failed_required or failed_optional or unrun_required:
        status = STATUS_DEGRADED
    else:
        status = STATUS_OK
    return status, ready, readiness_complete, tuple(reasons)


def load_system_health(
    *,
    deep: bool = False,
    host: str | None = None,
    probes: tuple[DependencyProbe, ...] | None = None,
    daemon_head: str | None = None,
    use_cache: bool = True,
) -> SystemHealthSnapshot:
    """Assemble the read-only system-health snapshot.

    ``deep=True`` adds the network probe against Gitea; its result is cached for
    a short TTL so repeated dashboard polls do not amplify into remote load.
    """
    repo = _repo_root()
    probe_errors: list[str] = []

    if probes is None:
        collected: list[DependencyProbe] = []
        for probe_fn in (
            lambda: probe_control_plane_db(),
            lambda: probe_repository(repo),
        ):
            try:
                collected.append(probe_fn())
            except Exception as exc:  # noqa: BLE001 — a probe must not 500 the API
                probe_errors.append(redact(f"probe raised: {exc}"))
        resolved_host = host if host is not None else _default_host()
        if deep and not _offline():
            collected.append(_cached_gitea_probe(resolved_host, use_cache=use_cache))
        else:
            collected.append(_skipped_gitea(resolved_host))
        probes = tuple(collected)

    status, ready, readiness_complete, reasons = _aggregate(probes)
    stale = assess_stale_runtime(repo, daemon_head=daemon_head)
    if stale.stale:
        if status == STATUS_OK:
            status = STATUS_DEGRADED
        reasons = reasons + (
            "runtime is stale relative to its remote-tracking commit",
        )

    db_probe = next((p for p in probes if p.name == "control_plane_db"), None)
    schema_version = None
    if db_probe and db_probe.metadata:
        schema_version = db_probe.metadata.get("schema_version")

    return SystemHealthSnapshot(
        status=status,
        ready=ready,
        readiness_complete=readiness_complete,
        readiness_reasons=reasons,
        service=SERVICE_NAME,
        mode="read-only",
        version=_load_version(repo, schema_version=schema_version),
        started_at=_STARTED_AT.isoformat(),
        uptime_seconds=round(time.monotonic() - _STARTED_MONOTONIC, 3),
        timestamp=datetime.now(timezone.utc).isoformat(),
        deep_probes_requested=deep,
        dependencies=probes,
        mcp_namespaces=namespace_summaries(),
        stale_runtime=stale,
        probe_errors=tuple(probe_errors),
    )


def _cached_gitea_probe(host: str, *, use_cache: bool = True) -> DependencyProbe:
    ttl = _deep_probe_ttl()
    now = time.monotonic()
    if use_cache and ttl > 0:
        cached = _deep_cache.get(host)
        if cached and (now - cached[0]) < ttl:
            return cached[1]
    probe = probe_gitea(host)
    if use_cache and ttl > 0:
        _deep_cache[host] = (now, probe)
    return probe


def clear_probe_cache() -> None:
    """Drop cached deep-probe results (tests and operator-forced refresh)."""
    _deep_cache.clear()


def probe_to_dict(probe: DependencyProbe) -> dict[str, Any]:
    return {
        "name": probe.name,
        "kind": probe.kind,
        "status": probe.status,
        "detail": probe.detail,
        "required": probe.required,
        "healthy": probe.healthy,
        "latency_ms": probe.latency_ms,
        "metadata": dict(probe.metadata or {}),
    }


def snapshot_to_dict(snapshot: SystemHealthSnapshot) -> dict[str, Any]:
    return {
        "status": snapshot.status,
        "service": snapshot.service,
        "mode": snapshot.mode,
        "api": API_PATH,
        "timestamp": snapshot.timestamp,
        "readiness": {
            "ready": snapshot.ready,
            "complete": snapshot.readiness_complete,
            "reasons": list(snapshot.readiness_reasons),
        },
        "version": {
            "git_sha": snapshot.version.git_sha,
            "git_describe": snapshot.version.git_describe,
            "control_plane_schema_version": (
                snapshot.version.control_plane_schema_version
            ),
            "python_version": snapshot.version.python_version,
            "known": snapshot.version.known,
        },
        "process": {
            "started_at": snapshot.started_at,
            "uptime_seconds": snapshot.uptime_seconds,
        },
        "deep_probes_requested": snapshot.deep_probes_requested,
        "dependencies": [probe_to_dict(probe) for probe in snapshot.dependencies],
        "mcp_namespaces": [dict(row) for row in snapshot.mcp_namespaces],
        "stale_runtime": {
            "daemon_head": snapshot.stale_runtime.daemon_head,
            "checkout_head": snapshot.stale_runtime.checkout_head,
            "remote_head": snapshot.stale_runtime.remote_head,
            "stale": snapshot.stale_runtime.stale,
            "determinable": snapshot.stale_runtime.determinable,
            "mutation_safe": snapshot.stale_runtime.mutation_safe,
            "reasons": list(snapshot.stale_runtime.reasons),
        },
        "probe_errors": list(snapshot.probe_errors),
    }