chemenu/tools/chemenu/commands/doctor.py

"""`wikitool doctor` - one deterministic health check for a wiki instance.

Read-only, never writes. Exists to back `instructions/setup-instance.md` (and
any other instance-setup procedure) with a single command instead of ten
individual checks spelled out in prose - the same reasoning that keeps
mechanical work in code everywhere else in this repo. Each check reports
`OK`, `WARN`, or `FAIL` plus, on anything but `OK`, the command to fix it.
Only a `FAIL` makes the overall exit code non-zero: a fresh instance with no
remote yet, or no `WIKITOOL_SESSION_ID` set, is a valid state, not a fault.
"""
from __future__ import annotations

import json as _json
import shutil
import subprocess
import sys
from dataclasses import dataclass
from typing import Optional

import typer
from rich.console import Console

from chemenu import config, conventions, kb_collections, version as version_mod
from chemenu.commands import git_publish, instructions_cmd
from chemenu.commands._util import rel_path
from chemenu.session import ENV_VAR as SESSION_ENV_VAR
from chemenu.session import session_id_source as _session_id_source

console = Console()


@dataclass
class Check:
    name: str
    status: str  # "OK" | "WARN" | "FAIL"
    detail: str
    fix: Optional[str] = None


def _git(args: list[str]) -> Optional[subprocess.CompletedProcess]:
    try:
        return subprocess.run(
            ["git", *args], cwd=config.ROOT, capture_output=True, text=True, timeout=5
        )
    except (OSError, subprocess.SubprocessError):
        return None


def check_python() -> Check:
    version = sys.version_info
    if version < (3, 11):
        return Check(
            "python", "FAIL", f"Python {version.major}.{version.minor} found, need >= 3.11",
            "Install Python 3.11+ and recreate tools/.venv",
        )
    return Check("python", "OK", f"Python {version.major}.{version.minor}.{version.micro}")


def check_ripgrep() -> Check:
    if shutil.which("rg"):
        return Check("ripgrep", "OK", "rg found on PATH")
    return Check(
        "ripgrep", "FAIL", "rg not found on PATH - `search` and `sources coverage` need it",
        "Install ripgrep (e.g. `apt install ripgrep` / `brew install ripgrep`)",
    )


def check_author() -> Check:
    author = config.default_author()
    if author is None:
        return Check(
            "author", "FAIL", "Neither $WIKI_AUTHOR nor `git config user.name` resolves",
            "Run `git config user.name \"<Your Name>\"`, or export WIKI_AUTHOR",
        )
    import os

    source = "WIKI_AUTHOR" if os.environ.get("WIKI_AUTHOR", "").strip() else "git config user.name"
    return Check("author", "OK", f"'{author}' (from {source})")


def check_git_repo() -> list[Check]:
    checks: list[Check] = []
    inside = _git(["rev-parse", "--is-inside-work-tree"])
    if inside is None or inside.returncode != 0 or inside.stdout.strip() != "true":
        checks.append(
            Check(
                "git-repo", "FAIL", "Not inside a git working tree",
                "Run `git init -b main`",
            )
        )
        return checks
    checks.append(Check("git-repo", "OK", "Inside a git working tree"))

    name = _git(["config", "user.name"])
    email = _git(["config", "user.email"])
    if not name or not name.stdout.strip():
        checks.append(
            Check("git-identity", "FAIL", "`git config user.name` is not set",
                  "Run `git config user.name \"<Your Name>\"`")
        )
    elif not email or not email.stdout.strip():
        checks.append(
            Check("git-identity", "FAIL", "`git config user.email` is not set",
                  "Run `git config user.email \"<you@example.com>\"`")
        )
    else:
        checks.append(Check("git-identity", "OK", f"{name.stdout.strip()} <{email.stdout.strip()}>"))

    branch = _git(["rev-parse", "--abbrev-ref", "HEAD"])
    branch_name = branch.stdout.strip() if branch and branch.returncode == 0 else ""
    if not branch_name or branch_name == "HEAD":
        checks.append(
            Check("git-branch", "WARN", "No commit yet, or detached HEAD",
                  "Make the first commit via `publish` once ready")
        )
    else:
        checks.append(Check("git-branch", "OK", f"On branch '{branch_name}'"))

    remote = _git(["remote", "get-url", "origin"])
    if remote and remote.returncode == 0 and remote.stdout.strip():
        checks.append(Check("git-remote", "OK", remote.stdout.strip()))
    else:
        checks.append(
            Check(
                "git-remote", "WARN", "No 'origin' remote configured",
                "A local-only instance is valid - `git remote add origin <url>` if you want one. "
                "Every `publish` needs --no-push until then",
            )
        )
    return checks


def check_skills() -> Check:
    sources = instructions_cmd.skill_dirs()
    if not sources:
        return Check("skills", "FAIL", "No skills found under instructions/", None)
    target_dirs = instructions_cmd.target_dirs()
    missing = 0
    drifted: list[str] = []
    for target_root in target_dirs:
        for source in sources:
            difference = instructions_cmd.drift(source, target_root / source.name)
            if difference == "missing":
                missing += 1
            elif difference:
                drifted.append(f"{rel_path(target_root / source.name)}: {difference}")
    expected = len(sources) * len(target_dirs)
    if missing == expected and not drifted:
        return Check(
            "skills", "FAIL", "No skills published yet",
            "Run `tools/wikitool instructions sync`",
        )
    if drifted:
        return Check(
            "skills", "FAIL", f"{len(drifted)} published copy/copies drifted from source",
            "Run `tools/wikitool instructions sync`",
        )
    return Check("skills", "OK", f"{expected} published copy/copies match their source")


def check_structure() -> Check:
    missing = []
    for relative_path in (
        "kb/CONTRACT.md", "raw/CONTRACT.md", "reports/CONTRACT.md",
        "work/CONTRACT.md", "instructions/CONTRACT.md", "types/type-spec.md",
    ):
        if not (config.ROOT / relative_path).exists():
            missing.append(relative_path)
    collections = kb_collections.iter_kb_collections()
    if not collections:
        missing.append("kb/*/COLLECTION.md")
    if missing:
        return Check(
            "structure", "FAIL", f"Missing: {', '.join(missing)}",
            "Re-run `dist export`, or restore the missing contract(s) from the source repo",
        )
    return Check(
        "structure", "OK", f"{len(collections)} collection(s), all stage contracts present"
    )


def check_personalization() -> Check:
    """Whether this instance knows who it works for, and how it sounds.

    `USER.md` and `SOUL.md` are read every session, so an instance without
    them runs a generic agent against a wiki built for one person - which is
    a fault, not a preference, hence `FAIL` rather than `WARN`. They are also
    the one pair of required files a distribution cannot ship filled: their
    content is personal, so `dist export` carries the templates and the
    Personalization step of `setup-instance.md` writes the real ones. That
    makes a still-templated file the second failure mode worth naming
    separately - it looks present and answers nothing.
    """
    missing: list[str] = []
    unfilled: list[str] = []
    for name in config.PERSONALIZATION_FILES:
        path = config.ROOT / name
        if not path.is_file():
            missing.append(name)
        elif config.TEMPLATE_SENTINEL in path.read_text(encoding="utf-8"):
            unfilled.append(name)

    fix = (
        "Run the Personalization step of instructions/setup-instance.md - it interviews you "
        f"along {' and '.join(config.PERSONALIZATION_TEMPLATES)} and writes your answers verbatim"
    )
    if missing:
        return Check("personalization", "FAIL", f"Missing: {', '.join(missing)}", fix)
    if unfilled:
        return Check(
            "personalization", "FAIL",
            f"Still the unfilled template: {', '.join(unfilled)}", fix,
        )
    return Check("personalization", "OK", f"{', '.join(config.PERSONALIZATION_FILES)} present and filled")


def check_conventions() -> Check:
    """Whether this instance has said how its own pages are written.

    `kb/CONVENTIONS.md` carries the decisions `kb/CONTRACT.md` deliberately no
    longer makes: the KB language and the headings its two generated regions
    render under, the tone examples, the hedging rule, the naming forms.

    `FAIL` rather than `WARN` because those decisions bind every page, and
    because it has the same two failure modes the personalization pair has: the
    distribution can ship the template but never the filled file, so a template
    renamed and left unanswered looks present and decides nothing.

    The headings themselves are only cosmetic now - the marker pair carries each
    region's identity, so a default renders wrong words rather than corrupting
    structure. That is why this check is about the *file*, not about rescuing a
    lookup the compiler can no longer get wrong.
    """
    path = conventions.conventions_file()
    fix = (
        "Copy kb/CONVENTIONS.md.template to kb/CONVENTIONS.md and answer it - the KB-language "
        "step of instructions/setup-instance.md walks it, and instructions/kb-profiles.md has "
        "the ready-made profiles to adopt"
    )
    if not path.is_file():
        return Check(
            "conventions", "FAIL",
            f"kb/{conventions.CONVENTIONS_FILENAME} is missing - this instance has not "
            "declared how its pages are written",
            fix,
        )
    issues = conventions.declaration_issues()
    if issues:
        return Check("conventions", "FAIL", "; ".join(issues), fix)
    declared = conventions.language() or "unspecified"
    from chemenu import blocks

    headings = ", ".join(conventions.heading(block) for block in blocks.BLOCKS)
    return Check(
        "conventions", "OK",
        f"kb/{conventions.CONVENTIONS_FILENAME} present, language {declared}, "
        f"sections {headings}",
    )


def check_environment() -> Check:
    """Whether this checkout records the environment it works through.

    `ENVIRONMENT.md` names the harness, the published skills, the MCP
    servers, the connectors and the git remotes this working copy actually
    uses. Missing it costs a session some questions, not correctness, so this
    check never FAILs - the whole point of the file is that it is optional,
    and a FAIL would make it mandatory by the back door.

    The one thing worth reporting is the failure mode the personalization
    check already knows: a template renamed but not filled in. That file is
    present, is loaded into every session, and answers nothing - worse than
    absence, because absence is honest.
    """
    path = config.ROOT / config.ENVIRONMENT_FILE
    if not path.is_file():
        return Check(
            "environment", "OK", f"{config.ENVIRONMENT_FILE} absent (optional)",
        )
    if config.TEMPLATE_SENTINEL in path.read_text(encoding="utf-8"):
        return Check(
            "environment", "WARN",
            f"{config.ENVIRONMENT_FILE} is still the unfilled template",
            f"Fill it in along {config.ENVIRONMENT_TEMPLATE}'s sections and drop the "
            f"`{config.TEMPLATE_SENTINEL}` line, or delete the file - it is optional",
        )
    return Check("environment", "OK", f"{config.ENVIRONMENT_FILE} present and filled")


def check_publish_remotes() -> Check:
    """Whether the Publish-Remote Gate is armed in this checkout.

    Absent is a legitimate state, not a fault: a checkout with a single remote
    and nothing private in it has nothing to protect, and making the file
    mandatory would turn a safeguard into paperwork. So this never FAILs - it
    reports, the way `environment` does.

    It does WARN for the case that actually bites: more than one remote
    configured and no allowlist. That is the shape a private instance has after
    it adds the public upstream, and it is exactly when a wrong `--remote`
    stops being a typo and starts being a disclosure.

    Both absent states say **armed** or **not armed** rather than only naming
    the file. AGENTS.md lists this among the three limits enforced in code, so a
    line that reports the file's absence and leaves the reader to infer what
    that means about the gate is how a checkout ends up trusting a safeguard
    that is not running - which is worse than having none.
    """
    urls = git_publish.read_allowed_push_urls()
    if urls is not None:
        return Check(
            "publish-remotes", "OK",
            f"Gate armed: {len(urls)} allowed push target(s) in "
            f"{config.PUBLISH_REMOTES_FILENAME}",
        )
    result = subprocess.run(
        ["git", "remote"], cwd=config.ROOT, capture_output=True, text=True
    )
    remotes = [r for r in result.stdout.split() if r]
    if len(remotes) > 1:
        return Check(
            "publish-remotes", "WARN",
            f"Gate not armed: {len(remotes)} remotes ({', '.join(remotes)}) and no "
            f"{config.PUBLISH_REMOTES_FILENAME} - every one of them is a legal publish target",
            f"Create {config.PUBLISH_REMOTES_FILENAME} naming the push URL this checkout "
            "may publish to - see instructions/gates.md",
        )
    return Check(
        "publish-remotes", "OK",
        f"Gate not armed: no {config.PUBLISH_REMOTES_FILENAME} - any push target passes "
        "(1 remote configured, nothing to confuse it with)",
    )


def check_generated_files() -> Check:
    missing = [
        rel_path(path)
        for path in (config.INDEX_FILE, config.LOG_FILE, config.PROVENANCE_FILE)
        if not path.exists()
    ]
    if missing:
        return Check(
            "generated-files", "FAIL", f"Missing: {', '.join(missing)}",
            "Run `index rebuild` and `sources rebuild-index`",
        )
    return Check("generated-files", "OK", "kb/index.md, kb/log.md, kb/provenance.md present")


def check_telemetry() -> Check:
    """Whether tracing is on for this checkout, why, and how full its two caps
    are.

    Never `FAIL`s, the same line `check_publish_remotes` and
    `check_environment` draw: both an enabled and a disabled tree are
    legitimate states, and a `FAIL` would make one of them mandatory by the
    back door. The tree walk below is fine here - `doctor` is not a hot path,
    unlike the `stat` `emit()` does on every append.
    """
    from chemenu.telemetry import policy as telemetry_policy
    from chemenu.telemetry import reader
    from chemenu.telemetry.writer import TRACE_FILE, trace_root

    pol = telemetry_policy.resolve(config.ROOT)
    root = trace_root()
    session_count = len(reader.sessions(root))
    total_bytes = (
        sum(f.stat().st_size for f in root.glob(f"*/{TRACE_FILE}") if f.is_file())
        if root.exists()
        else 0
    )
    state = "on" if pol.enabled else "off"
    return Check(
        "telemetry", "OK",
        f"{state} ({pol.reason}); {session_count}/{pol.keep_sessions} session(s), "
        f"{total_bytes:,} byte(s) under {rel_path(root)} "
        f"(cap {pol.max_session_bytes:,} byte(s)/session)",
    )


def check_upload_intake() -> Check:
    """Whether the MCP `submit` tool is armed for this checkout, and how full
    its quarantine is.

    Absent is the *safe* default here, unlike `check_publish_remotes`'s "any
    push target passes" absence: no `.wikitool-upload.json` means the write
    path does not exist at all, not that it is unrestricted - so this never
    `FAIL`s on a missing file. It does `FAIL` on one that parses to something
    invalid, because a broken opt-in must not silently disable the very
    limits it exists to enforce.
    """
    from chemenu import upload as upload_module
    from chemenu.errors import ValidationError

    try:
        cfg = upload_module.read_config(config.ROOT)
    except ValidationError as exc:
        return Check(
            "upload-intake", "FAIL", str(exc),
            f"Fix or delete {config.UPLOAD_CONFIG_FILENAME} - a broken one is not treated as "
            "'no limits'",
        )
    if cfg is None:
        return Check(
            "upload-intake", "OK",
            f"submit tool not registered - no {config.UPLOAD_CONFIG_FILENAME}",
        )
    pending = upload_module.list_submissions(config.ROOT)
    return Check(
        "upload-intake", "OK",
        f"submit tool armed (identity header {cfg.identity_header!r}, up to "
        f"{cfg.max_bytes:,} byte(s), {cfg.submissions_per_day}/day and "
        f"{cfg.bytes_per_day:,} byte(s)/day per submitter); "
        f"{len(pending)} submission(s) waiting in {rel_path(config.UPLOAD_DIR)}",
    )


def check_session_id() -> Check:
    """Three-valued, not two: an explicit `WIKITOOL_SESSION_ID` and a
    recognised harness variable (see `chemenu.session.HARNESS_ENV_VARS`) both
    keep a session's calls in one telemetry/budget bucket, so both are `OK`.
    Only the `getppid()` fallback - a fresh "session" on every call, on a
    harness that runs each tool call in its own shell - is a `WARN` (see
    Gitea #110)."""
    import os

    if os.environ.get(SESSION_ENV_VAR, "").strip():
        return Check("session-id", "OK", f"{SESSION_ENV_VAR}={os.environ[SESSION_ENV_VAR]}")
    source = _session_id_source()
    if source != "getppid() fallback":
        return Check("session-id", "OK", f"scoped by harness variable {source}")
    return Check(
        "session-id", "WARN",
        f"{SESSION_ENV_VAR} is not set and no harness session variable was found - "
        "budget falls back to the parent PID",
        "See instructions/session-setup.md",
    )


def check_stack_version() -> Check:
    """Which stack this instance runs, and where it came from.

    A missing `VERSION` is a WARN, not a FAIL: instances exported before the
    stack was versioned are still perfectly functional - they just cannot
    answer `version check`. A malformed one is a FAIL, because then something
    edited a generated fact by hand and every comparison built on it is wrong.
    """
    try:
        current = version_mod.read_version()
    except version_mod.VersionError as exc:
        if not version_mod.version_file().is_file():
            return Check(
                "stack-version", "WARN", "No VERSION file - this instance predates stack versioning",
                "Re-export from a current origin, or write the version this instance corresponds to",
            )
        return Check("stack-version", "FAIL", str(exc), f"Fix {version_mod.VERSION_FILENAME} by hand - it holds one semantic version, nothing else")

    try:
        stamp = version_mod.read_stamp()
    except version_mod.VersionError as exc:
        return Check(
            "stack-version", "FAIL", str(exc),
            f"Delete {version_mod.RELEASE_STAMP_FILENAME} or restore it from the release it came from",
        )

    origin = "development tree" if stamp is None else f"distribution, exported {stamp.get('exported_at', 'unknown')}"
    candidate = " - a running pre-release candidate, not yet fixed by `version release`" if current.is_prerelease else ""
    return Check("stack-version", "OK", f"{current} ({origin}){candidate}")


def check_kb_version() -> Check:
    """Whether the content is in the shape this machinery expects.

    A `WARN` when the content lags: that is the normal, transient state in the
    middle of an upgrade, not a fault - and `migrate status` names the chain
    that closes it. A missing declaration is also a `WARN` (an instance from
    before the file existed still works), an unreadable one a `FAIL`.
    """
    from chemenu import kb_state

    try:
        stack = version_mod.read_version()
    except version_mod.VersionError:
        return Check(
            "kb-version", "WARN", "No stack version to compare the content against",
            "See the stack-version check above",
        )
    try:
        kb_version = kb_state.read_kb_version()
    except version_mod.VersionError as exc:
        return Check(
            "kb-version", "FAIL", str(exc),
            f"Restore or delete {kb_state.KB_STATE_FILENAME}, then "
            "`tools/wikitool migrate baseline <version>`",
        )

    if kb_version is None:
        return Check(
            "kb-version", "WARN",
            f"{kb_state.KB_STATE_FILENAME} is missing - the content's shape is undeclared",
            f"Run `tools/wikitool migrate baseline {stack}` if this instance's content has "
            "never lagged behind its machinery",
        )
    if kb_version < stack:
        pending = kb_state.chain(kb_state.load_migrations(), kb_version, stack.base)
        if pending:
            return Check(
                "kb-version", "WARN",
                f"Content is at {kb_version}, machinery at {stack} - "
                f"{len(pending)} migration(s) outstanding",
                "Run `tools/wikitool migrate status`",
            )
        return Check("kb-version", "OK", f"{kb_version} (nothing outstanding up to {stack})")
    return Check("kb-version", "OK", f"{kb_version}")


def run_doctor() -> list[Check]:
    checks: list[Check] = [
        check_python(),
        check_ripgrep(),
        check_author(),
        check_stack_version(),
        check_kb_version(),
        *check_git_repo(),
        check_skills(),
        check_structure(),
        check_personalization(),
        check_conventions(),
        check_environment(),
        check_publish_remotes(),
        check_upload_intake(),
        check_generated_files(),
        check_session_id(),
        check_telemetry(),
    ]
    return checks


def doctor_command(
    json_out: bool = typer.Option(False, "--json", help="Print the checks as JSON"),
):
    """Check that this instance is correctly configured: dependencies, author,
    git identity/remote, published skills, structure, personalization, KB
    conventions, generated files, session scoping, telemetry state, and
    whether the MCP `submit` tool is armed. Read-only. Exits 1 only
    if a check FAILs."""
    checks = run_doctor()

    if json_out:
        typer.echo(_json.dumps([c.__dict__ for c in checks], indent=2))
    else:
        for check in checks:
            color = {"OK": "green", "WARN": "yellow", "FAIL": "bold red"}[check.status]
            line = f"[{color}]{check.status}[/{color}] {check.name}: {check.detail}"
            if check.fix and check.status != "OK":
                line += f"\n  fix: {check.fix}"
            typer.echo(line) if False else None
            from rich.console import Console

            Console().print(line)

    if any(check.status == "FAIL" for check in checks):
        raise typer.Exit(code=1)