Bolt-Console/bolt_console/doctor.py

"""`python -m bolt_console --doctor` — is this install actually going to work?

Carried over from Bolt-Pet, where a missing port publish, a wrong reverse-proxy
header and a venv whose python was a zero-byte file all presented as "it's
being weird" and each took a conversation to find.

Each check is independent and reports one of three things — ok, a warning
(works, but degraded), or a failure (this will not do what you expect) — and
says *what to do about it* rather than just what is wrong. Nothing here raises:
a doctor that crashes on a broken install is diagnosing the wrong patient.

Deliberately does not touch the network by default — it's the first thing you
reach for when the network is what's broken. `--deep` contacts the server.
"""

from __future__ import annotations

import sys
from dataclasses import dataclass
from typing import Callable, Optional

from . import config

OK, WARN, FAIL = "ok", "warn", "fail"

_MARKS = {OK: "  ok  ", WARN: " warn ", FAIL: " FAIL "}


@dataclass
class Check:
    name: str
    status: str
    detail: str = ""
    fix: str = ""

    def line(self) -> str:
        text = f"[{_MARKS[self.status]}] {self.name:22} {self.detail}"
        if self.fix and self.status != OK:
            text += f"\n{'':32}→ {self.fix}"
        return text


# ── the checks ──────────────────────────────────────────────────────────────

def check_config() -> Check:
    missing = config.missing_config()
    if missing:
        return Check("server config", FAIL, f"missing {', '.join(missing)}",
                     "set them in .env — without these the console exits at startup")
    return Check("server config", OK, f"{config.SERVER_URL} as {config.SESSION_ID}")


def check_server(deep: bool = False) -> Check:
    if not config.SERVER_URL:
        return Check("server", FAIL, "no BOLT_SERVER_URL",
                     "cp .env.example .env and set BOLT_SERVER_URL to your Bolt server")
    if not deep:
        return Check("server", OK, f"{config.SERVER_URL} (not contacted; --deep to try)")
    try:
        from . import server_client

        health = server_client.check_health(timeout=8)
        return Check("server", OK, f"reachable — {health}")
    except Exception as exc:
        return Check("server", FAIL, f"unreachable: {exc}",
                     "check the URL, the key, and that the container is up")


def check_streaming_endpoint(deep: bool = False) -> Check:
    """The streamed-reply endpoint is newer than some deployed servers."""
    if not config.STREAMING_REPLIES:
        return Check("streamed replies", WARN, "disabled (STREAMING_REPLIES=false)")
    if not deep:
        return Check("streamed replies", OK, "enabled (endpoint not probed)")
    try:
        import requests

        response = requests.post(
            f"{config.SERVER_URL}/desk/converse_stream",
            json={"session_id": config.SESSION_ID, "text": ""},
            headers={"X-Desk-Api-Key": config.API_KEY}, timeout=8, stream=True,
        )
        if response.status_code == 404:
            return Check("streamed replies", WARN, "server has no /desk/converse_stream",
                         "update the server, or set STREAMING_REPLIES=false to skip the probe")
        return Check("streamed replies", OK, f"endpoint answered {response.status_code}")
    except Exception as exc:
        return Check("streamed replies", WARN, f"probe failed: {exc}")


def check_python() -> Check:
    version = ".".join(str(n) for n in sys.version_info[:3])
    if sys.version_info < (3, 9):
        return Check("python", FAIL, version, "Python 3.9 or newer is required")
    return Check("python", OK, f"{version} at {sys.executable}")


def check_approval() -> Check:
    mode = config.APPROVAL_MODE
    if mode == "auto":
        return Check("approval", WARN, "auto — Bolt's commands, writes and edits run without asking",
                     "APPROVAL_MODE=ask to confirm each one (or /ask for one session)")
    if mode != "ask":
        return Check("approval", WARN, f"unknown APPROVAL_MODE={mode!r}, treated as ask",
                     "use ask or auto")
    return Check("approval", OK, "ask before commands, writes and edits")


def check_deliveries() -> Check:
    if not config.RECEIVE_FILES:
        return Check("file delivery", WARN, "disabled (RECEIVE_FILES=false)")
    directory = config.DELIVERED_FILES_DIR
    probe = directory
    while not probe.exists() and probe.parent != probe:
        probe = probe.parent
    import os

    if not os.access(str(probe), os.W_OK):
        return Check("file delivery", FAIL, f"can't write under {probe}",
                     "set DELIVERED_FILES_DIR to a folder you can write to")
    return Check("file delivery", OK, str(directory))


CHECKS = (
    ("python", check_python),
    ("config", check_config),
    ("server", check_server),
    ("stream", check_streaming_endpoint),
    ("approval", check_approval),
    ("deliveries", check_deliveries),
)


def run(deep: bool = False) -> list:
    results = []
    for _name, check in CHECKS:
        try:
            try:
                results.append(check(deep))
            except TypeError:
                results.append(check())
        except Exception as exc:  # a broken check must not hide the others
            results.append(Check(_name, FAIL, f"the check itself failed: {exc}"))
    return results


def main(argv: Optional[list] = None) -> int:
    argv = list(argv if argv is not None else sys.argv[1:])
    deep = "--deep" in argv
    print(f"Bolt Console preflight{' (deep: contacting the server)' if deep else ''}\n")
    results = run(deep=deep)
    for check in results:
        print(check.line())
    failures = [c for c in results if c.status == FAIL]
    warnings = [c for c in results if c.status == WARN]
    print()
    if failures:
        print(f"{len(failures)} problem(s) will stop this working. Fix those first.")
    elif warnings:
        print(f"Ready. {len(warnings)} thing(s) degraded but working.")
    else:
        print("Everything checks out.")
    return 1 if failures else 0