rigdoctor/src/rigdoctor/core/diagnostic.py

"""Guided diagnostic session (SPEC §4 / ARCHITECTURE §7.1): orchestrate M3 + M4.

The seed use case, one flow: **pick a game** → **focused crash-capture** scoped to that
session (M3, tagged with the game) → on **finish**, **scan & analyze** (M4 health report)
over the captured window + system logs → return a prioritized result. This is not a new
module — it's a single shared callable so the CLI, GUI, and tray run the identical flow.

The capture is **manually bracketed** (start/finish) for now; auto start/stop on game launch
(the D12 wrapper/watcher) plugs in here later without changing the result shape.
"""

from __future__ import annotations

import json
import time
from dataclasses import dataclass

from .. import config
from . import reccontrol
from .crashlog import Summary, summarize
from .health import CRITICAL, OK, WARNING, Finding

_SEV_ORDER = {CRITICAL: 0, WARNING: 1, "info": 2, OK: 3}


@dataclass
class DiagnosticResult:
    game: str | None
    summary: Summary           # capture window: peak temps/power, events, last samples (M3)
    findings: list[Finding]    # health findings: Xid/SMART/driver/etc. (M4)
    dir: str | None = None     # storage directory when logging is on (M15); else None


@dataclass
class CrashInfo:
    game: str | None
    samples: int
    when: float | None         # ts of the last captured sample (≈ when the freeze hit)
    gpu_lost: bool


def _clear_diag_log() -> None:
    """Each diagnostic is a fresh focused capture — drop any previous session + segments."""
    base = config.DIAG_LOG
    for p in [base, *base.parent.glob(base.name + ".*")]:
        try:
            p.unlink()
        except OSError:
            pass


def start(game: str | None = None, interval: float | None = None) -> int | None:
    """Begin a focused capture, tagged with the game, into the dedicated diagnostic log.
    Returns the pid, or None if a capture is already running."""
    if reccontrol.running_pid():
        return None
    if _crash_from_log(config.DIAG_LOG):  # preserve an unanalyzed crash before overwriting it
        try:
            config.DIAG_LOG.replace(config.DIAG_CRASH)
        except OSError:
            pass
    _clear_diag_log()
    return reccontrol.start_background(interval=interval, out=str(config.DIAG_LOG), game=game)


def is_running() -> bool:
    return reccontrol.running_pid() is not None


def active() -> dict | None:
    """Status of the in-progress session (running flag, game, samples), or None if idle."""
    if not is_running():
        return None
    return reccontrol.read_status()


def _await_stopped(timeout: float = 6.0) -> None:
    deadline = time.monotonic() + timeout
    while reccontrol.running_pid() and time.monotonic() < deadline:
        time.sleep(0.1)


def _game_from_summary(summary: Summary) -> str | None:
    """Recover the focused game from the log's 'game' event (survives a crash + reboot)."""
    for _ts, kind, detail in reversed(summary.events):
        if kind == "game" and detail:
            return detail
    return None


def finish(last_n: int = 10, log_path=None) -> DiagnosticResult:
    """Stop the capture (if running), summarize the window, and run the health report."""
    from .health import run_health_checks

    reccontrol.stop_background()
    _await_stopped()
    path = log_path or config.DIAG_LOG
    summary = summarize(path, last_n=last_n)
    game = _game_from_summary(summary) or (reccontrol.read_status() or {}).get("game")
    findings = run_health_checks()
    result = DiagnosticResult(game=game, summary=summary, findings=findings)
    _store(result, path, summary)
    return result


def _store(result: DiagnosticResult, capture_path, summary: Summary) -> None:
    """Persist the diagnostic to its own directory when logging is enabled (M15)."""
    try:
        from . import diagstore

        since = (summary.start - 60) if summary.start else None
        directory = diagstore.store(result, capture_path, since=since)
        if directory:
            result.dir = str(directory)
    except Exception:  # storage must never break a diagnostic
        pass


# --- hard-crash detection & post-crash analysis -----------------------------------

def _crash_from_log(path) -> CrashInfo | None:
    """CrashInfo if `path` holds an abnormally-ended session (start, no stop, not acked)."""
    if not path.exists():
        return None
    summary = summarize(path)
    kinds = {kind for _ts, kind, _detail in summary.events}
    if "session-start" not in kinds:
        return None
    if "session-stop" in kinds or "diagnostic-acknowledged" in kinds:
        return None
    return CrashInfo(
        game=_game_from_summary(summary),
        samples=summary.samples,
        when=summary.end,
        gpu_lost="gpu-lost" in kinds,
    )


def _crash_path():
    """Where the pending crash lives: the preserved archive if present, else the live log."""
    return config.DIAG_CRASH if config.DIAG_CRASH.exists() else config.DIAG_LOG


def pending_crash() -> CrashInfo | None:
    """Detect a diagnostic that ended abnormally (no clean stop, no live recorder).

    A focused capture writes `session-start` (+ `game`) and, on a clean stop, `session-stop`.
    After a hard freeze that block never runs, so the log has a start with no stop and no
    live recorder — that's our hard-crash signal. A crash preserved across an auto-relaunch
    (`DIAG_CRASH`) is checked first. Returns None if a capture is running, none is recorded,
    it stopped cleanly, or the user already acknowledged it.
    """
    info = _crash_from_log(config.DIAG_CRASH)  # preserved across a relaunch (wrapper)
    if info is not None:
        return info
    if is_running():
        return None
    return _crash_from_log(config.DIAG_LOG)


def acknowledge_crash() -> None:
    """Mark the recorded crash as seen so it stops prompting."""
    try:
        config.DIAG_CRASH.unlink()  # drop the preserved archive, if any
    except OSError:
        pass
    try:
        config.DIAG_LOG.parent.mkdir(parents=True, exist_ok=True)
        with open(config.DIAG_LOG, "a", encoding="utf-8") as fh:
            fh.write(json.dumps({"ts": time.time(), "event": "diagnostic-acknowledged", "detail": ""}) + "\n")
    except OSError:
        pass


def _crash_headline(summary: Summary) -> Finding:
    gpu_lost = any(kind == "gpu-lost" for _ts, kind, _detail in summary.events)
    when = time.strftime("%H:%M:%S", time.localtime(summary.end)) if summary.end else "?"
    detail = (
        f"The capture stopped abruptly at {when} after {summary.samples} samples, with no clean "
        "shutdown recorded — consistent with a hard freeze or power loss."
    )
    if gpu_lost:
        detail += " A GPU-lost event was captured during the session."
    return Finding(
        CRITICAL if gpu_lost else WARNING,
        "Diagnostic",
        "Session ended without a clean stop (likely a hard crash)",
        detail,
        "Review the last readings (Capture, above) and the crash-boot findings below.",
    )


def analyze_crash(last_n: int = 15) -> DiagnosticResult:
    """Analyze a recorded hard crash: the captured window + the previous boot's kernel log
    + the rest of the health report (SMART/driver/persistence/temps)."""
    from .health import check_previous_boot, run_health_checks

    summary = summarize(_crash_path(), last_n=last_n)
    findings: list[Finding] = [_crash_headline(summary)]
    findings += check_previous_boot()                       # the crashed boot's kernel log
    findings += run_health_checks(include_journal=False)    # SMART/driver/persistence/temps
    findings.sort(key=lambda f: _SEV_ORDER.get(f.severity, 9))
    result = DiagnosticResult(game=_game_from_summary(summary), summary=summary, findings=findings)
    _store(result, _crash_path(), summary)
    return result