gpu-rent/src/gpu_rent/debug_api.py

"""Localhost debug HTTP sidecar for gpu-rent.

Bind only 127.0.0.1. Started from `up` / `tunnel` / `debug`.
Mostly read-only; Assistent session / chat-eval endpoints are opt-in
(may load VRAM / persist chat) and are not part of `/snapshot`.
"""

from __future__ import annotations

import json
import threading
import time
import traceback
from collections import deque
from collections.abc import Callable
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
from urllib.parse import parse_qs, urlparse

from gpu_rent.config import Config
from gpu_rent import debug_checks

Log = Callable[[str], None]

DEFAULT_PORT = 17821
EVENT_CAPACITY = 500
CACHE_TTL_HEALTH = 5.0
CACHE_TTL_DIAG = 30.0
CACHE_TTL_LOGS = 10.0
CACHE_TTL_DEFAULT = 8.0

_CHAT_EVAL_PARAMS = [
    {
        "name": "message",
        "in": "query",
        "schema": {"type": "string"},
        "description": "User turn (default: short RU checkpoint/steps/cfg ask)",
    },
    {
        "name": "persona",
        "in": "query",
        "schema": {"type": "string"},
        "description": "Persona id (default neutral)",
    },
    {
        "name": "pack",
        "in": "query",
        "schema": {"type": "string"},
        "description": "Pack id (default ordinary)",
    },
    {
        "name": "model",
        "in": "query",
        "schema": {"type": "string"},
        "description": "Ollama chat model override (else preferred / default_chat)",
    },
    {
        "name": "timeout",
        "in": "query",
        "schema": {"type": "number", "default": 120},
        "description": "Seconds, capped (15–300)",
    },
]

_WARN_VRAM = (
    "WARNING: may load chat model into VRAM, may write assistent.sqlite chat, "
    "and incurs GPU billing if the VM is up. Not part of /snapshot. "
    "compactContext / Exact merge are reconstructed client-side "
    "(Assistent HTTP API does not return them)."
)

_OPENAPI: dict[str, Any] = {
    "openapi": "3.0.3",
    "info": {
        "title": "gpu-rent Debug API",
        "version": "1.3.0",
        "description": (
            "Localhost diagnostics for installer progress and "
            "SwarmUI / Comfy / Ollama / Assistent. Mostly read-only. "
            "Assistent multi-turn: POST /assistent/session + "
            "/assistent/session/{id}/chat (trace object). "
            "One-shot: POST/GET /assistent/chat-eval. "
            "Offline reply score: POST /assistent/analyze-reply "
            "(no VRAM). "
            + _WARN_VRAM
        ),
    },
    "servers": [{"url": "http://127.0.0.1:17821"}],
    "paths": {
        "/": {"get": {"summary": "HTML index of endpoints"}},
        "/openapi.json": {"get": {"summary": "This OpenAPI document"}},
        "/health": {"get": {"summary": "ok, phase, failed checks summary"}},
        "/progress": {"get": {"summary": "Current up/tunnel step + last log"}},
        "/events": {
            "get": {
                "summary": "Installer log ring buffer",
                "parameters": [
                    {
                        "name": "since",
                        "in": "query",
                        "schema": {"type": "integer"},
                        "description": "Return events with seq > since",
                    }
                ],
            }
        },
        "/snapshot": {
            "get": {
                "summary": "Cheap aggregate (no full diag, no Assistent chat/session)"
            }
        },
        "/status": {"get": {"summary": "JSON mirror of gpu-rent status"}},
        "/state": {"get": {"summary": "state.json without secrets"}},
        "/config": {"get": {"summary": "Config with tokens/passwords redacted"}},
        "/checks": {"get": {"summary": "Local + VM service checks"}},
        "/logs": {
            "get": {
                "summary": "journalctl via SSH",
                "parameters": [
                    {
                        "name": "unit",
                        "in": "query",
                        "schema": {"type": "string"},
                        "description": "swarm|ollama|killer|cloud-init|all",
                    },
                    {
                        "name": "lines",
                        "in": "query",
                        "schema": {"type": "integer", "default": 80},
                    },
                ],
            }
        },
        "/diag": {"get": {"summary": "Full swarm_diag (expensive, cached ~30s)"}},
        "/gpu": {"get": {"summary": "nvidia-smi / CUDA / torch probe"}},
        "/swarm": {"get": {"summary": "SwarmUI backend_status + ListBackends"}},
        "/ollama": {"get": {"summary": "Ollama tags + ps"}},
        "/assistent": {
            "get": {
                "summary": "Deep Assistent bundle (extension/roles/api/hints)",
                "parameters": [
                    {
                        "name": "chat_smoke",
                        "in": "query",
                        "schema": {"type": "boolean"},
                        "description": "1-token Ollama /api/chat with preferred model",
                    },
                    {
                        "name": "logs",
                        "in": "query",
                        "schema": {"type": "boolean"},
                        "description": "Include filtered swarmui journal lines",
                    },
                ],
            }
        },
        "/assistent/diagnose": {
            "get": {
                "summary": (
                    "Full static+live Assistent health (logs on by default). "
                    + _WARN_VRAM
                ),
                "parameters": [
                    {
                        "name": "chat_smoke",
                        "in": "query",
                        "schema": {"type": "boolean"},
                    },
                    {
                        "name": "logs",
                        "in": "query",
                        "schema": {"type": "boolean", "default": True},
                        "description": "Pass logs=0 to skip journal",
                    },
                ],
            }
        },
        "/assistent/extension": {
            "get": {"summary": "DLL/Sqlite deps/Tab assets under Extensions/swarm-assistent"}
        },
        "/assistent/overlay": {
            "get": {"summary": "Overlay personas + _base/assistant.json vs local seed"}
        },
        "/assistent/roles": {
            "get": {"summary": "ollama-roles.json vs /api/tags (default_chat)"}
        },
        "/assistent/memory": {
            "get": {"summary": "assistent.sqlite presence + counts + Sqlite DLL hint"}
        },
        "/assistent/api": {
            "get": {
                "summary": "Live AssistentListPersonas/Models/Memory/Chats smoke",
                "parameters": [
                    {
                        "name": "chat_smoke",
                        "in": "query",
                        "schema": {"type": "boolean"},
                        "description": "Also run 1-token Ollama chat",
                    }
                ],
            }
        },
        "/assistent/wanted": {"get": {"summary": "Wanted-models queue on data volume"}},
        "/assistent/logs": {
            "get": {"summary": "journalctl swarmui filtered for Assistent/Sqlite/build"}
        },
        "/assistent/session": {
            "post": {
                "summary": (
                    "Start multi-turn Assistent debug session "
                    "(GetNewSession + optional GetConfig). " + _WARN_VRAM
                ),
                "requestBody": {
                    "required": False,
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "object",
                                "properties": {
                                    "persona": {"type": "string"},
                                    "pack": {"type": "string"},
                                    "model": {"type": "string"},
                                    "skills": {
                                        "type": "array",
                                        "items": {"type": "string"},
                                    },
                                    "context": {
                                        "type": "object",
                                        "description": (
                                            "Synthetic compactContext fields "
                                            "(krea_profile, checkpoint, "
                                            "recommended_params, …)"
                                        ),
                                    },
                                    "probe_config": {
                                        "type": "boolean",
                                        "default": True,
                                    },
                                },
                            }
                        }
                    },
                },
            }
        },
        "/assistent/session/{id}": {
            "get": {"summary": "In-memory session summary + last trace"},
            "delete": {"summary": "Drop in-memory debug session"},
        },
        "/assistent/session/{id}/chat": {
            "post": {
                "summary": (
                    "Send user message; returns reply + trace "
                    "(patch, Exact merge, compact_context sizes, system_layers). "
                    + _WARN_VRAM
                ),
                "requestBody": {
                    "required": True,
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "object",
                                "required": ["message"],
                                "properties": {
                                    "message": {"type": "string"},
                                    "timeout": {"type": "number"},
                                    "persona": {"type": "string"},
                                    "pack": {"type": "string"},
                                    "model": {"type": "string"},
                                    "skills": {
                                        "type": "array",
                                        "items": {"type": "string"},
                                    },
                                    "context": {"type": "object"},
                                },
                            }
                        }
                    },
                },
            }
        },
        "/assistent/analyze-reply": {
            "post": {
                "summary": (
                    "Offline: extract patch + 0.15.3 client would_generate "
                    "from a pasted {message, reply}. No Swarm / VRAM."
                ),
                "requestBody": {
                    "required": True,
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "object",
                                "required": ["reply"],
                                "properties": {
                                    "message": {"type": "string"},
                                    "reply": {"type": "string"},
                                    "persona": {"type": "string"},
                                    "pack": {"type": "string"},
                                },
                            }
                        }
                    },
                },
            },
            "get": {
                "summary": "Same as POST; query message/reply (short only)",
            },
        },
        "/assistent/client-event": {
            "post": {
                "summary": (
                    "Browser beacon of last UI turn (intent, patch, "
                    "swarm_prompt). CORS localhost. No VRAM."
                ),
            }
        },
        "/assistent/client-events": {
            "get": {
                "summary": "In-memory UI beacons (query since=seq)",
                "parameters": [
                    {
                        "name": "since",
                        "in": "query",
                        "schema": {"type": "integer"},
                    }
                ],
            }
        },
        "/assistent/chat-eval": {
            "get": {
                "summary": (
                    "One-shot AssistentChat (session+chat+delete). " + _WARN_VRAM
                ),
                "parameters": _CHAT_EVAL_PARAMS,
            },
            "post": {
                "summary": (
                    "Same as GET; prefer JSON body "
                    "{message,persona,pack,model,timeout,context}"
                ),
                "requestBody": {
                    "required": False,
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "object",
                                "properties": {
                                    "message": {"type": "string"},
                                    "persona": {"type": "string"},
                                    "pack": {"type": "string"},
                                    "model": {"type": "string"},
                                    "timeout": {"type": "number"},
                                    "context": {"type": "object"},
                                },
                            }
                        }
                    },
                },
            },
        },
    },
}


def _is_assistent_session_chat(path: str) -> bool:
    parts = [p for p in path.split("/") if p]
    return (
        len(parts) == 4
        and parts[0] == "assistent"
        and parts[1] == "session"
        and parts[3] == "chat"
    )


def _is_assistent_session_id(path: str) -> bool:
    parts = [p for p in path.split("/") if p]
    return len(parts) == 3 and parts[0] == "assistent" and parts[1] == "session"


@dataclass
class DebugHub:
    """Shared state for the sidecar process."""

    cfg: Config
    port: int
    started_at: float = field(default_factory=time.time)
    step: str = "starting"
    step_started_at: float = field(default_factory=time.time)
    _events: deque[dict[str, Any]] = field(default_factory=lambda: deque(maxlen=EVENT_CAPACITY))
    _client_events: deque[dict[str, Any]] = field(
        default_factory=lambda: deque(maxlen=80)
    )
    _client_seq: int = 0
    _seq: int = 0
    _lock: threading.Lock = field(default_factory=threading.Lock)
    _cache: dict[str, tuple[float, Any]] = field(default_factory=dict)

    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self.port}"

    def set_step(self, step: str) -> None:
        with self._lock:
            if step != self.step:
                self.step = step
                self.step_started_at = time.time()

    def push_event(self, message: str) -> None:
        text = message
        if text.startswith("\r"):
            text = text[1:].rstrip()
            if not text:
                return
            # Progress lines: update step hint, keep one event per distinct text.
        text = text.rstrip()
        if not text:
            return
        with self._lock:
            self._seq += 1
            self._events.append(
                {
                    "seq": self._seq,
                    "ts": time.time(),
                    "msg": text[:2000],
                }
            )
            # Infer step from log line (best-effort).
            lower = text.lower()
            for needle, label in (
                ("жду ready backend", "wait_backend_idle"),
                ("backend ready", "backend_ready"),
                ("проверка gpu", "verify_gpu"),
                ("проверка на vm", "verify_stack"),
                ("проверка туннеля", "verify_tunnel"),
                ("bootstrap", "bootstrap"),
                ("civitai", "seed_models"),
                ("ollama", "ollama"),
                ("extensions", "extensions"),
                ("туннель", "tunnel"),
                ("создаю", "provisioning"),
                ("doctor", "doctor"),
            ):
                if needle in lower:
                    if self.step != label:
                        self.step = label
                        self.step_started_at = time.time()
                    break

    def events_since(self, since: int = 0) -> list[dict[str, Any]]:
        with self._lock:
            return [e for e in self._events if int(e["seq"]) > since]

    def events_tail(self, n: int = 30) -> list[dict[str, Any]]:
        with self._lock:
            items = list(self._events)
        return items[-n:]

    def push_client_event(self, event: dict[str, Any]) -> dict[str, Any]:
        with self._lock:
            self._client_seq += 1
            row = {
                "seq": self._client_seq,
                "ts": time.time(),
                **{k: event[k] for k in event if k != "seq"},
            }
            self._client_events.append(row)
        return {"ok": True, "seq": row["seq"], "stored": True}

    def client_events_since(self, since: int = 0) -> list[dict[str, Any]]:
        with self._lock:
            return [e for e in self._client_events if int(e.get("seq") or 0) > since]

    def progress(self) -> dict[str, Any]:
        with self._lock:
            last = self._events[-1] if self._events else None
            step = self.step
            step_age = int(time.time() - self.step_started_at)
            uptime = int(time.time() - self.started_at)
        return {
            "ok": True,
            "step": step,
            "step_age_sec": step_age,
            "uptime_sec": uptime,
            "last_log": last,
        }

    def cached(self, key: str, ttl: float, factory: Callable[[], Any]) -> Any:
        now = time.time()
        with self._lock:
            hit = self._cache.get(key)
            if hit and now - hit[0] < ttl:
                return hit[1]
        value = factory()
        with self._lock:
            self._cache[key] = (now, value)
        return value


@dataclass
class DebugServer:
    hub: DebugHub
    httpd: ThreadingHTTPServer
    thread: threading.Thread

    @property
    def base_url(self) -> str:
        return self.hub.base_url

    @property
    def port(self) -> int:
        return self.hub.port

    def set_step(self, step: str) -> None:
        self.hub.set_step(step)

    def push_event(self, message: str) -> None:
        self.hub.push_event(message)

    def wrap_log(self, log: Log) -> Log:
        def _tee(msg: str) -> None:
            try:
                self.hub.push_event(msg)
            except Exception:
                pass
            log(msg)

        return _tee

    def stop(self) -> None:
        try:
            self.httpd.shutdown()
        except Exception:
            pass
        try:
            self.httpd.server_close()
        except Exception:
            pass
        self.thread.join(timeout=5.0)


_active: DebugServer | None = None
_active_lock = threading.Lock()


def get_active() -> DebugServer | None:
    with _active_lock:
        return _active


def _make_handler(hub: DebugHub):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, fmt: str, *args: Any) -> None:
            return  # quiet — do not spam CLI

        def _send(self, code: int, body: bytes, content_type: str) -> None:
            self.send_response(code)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type")
            self.end_headers()
            self.wfile.write(body)

        def _json(self, code: int, payload: Any) -> None:
            body = json.dumps(payload, ensure_ascii=False, indent=2, default=str).encode(
                "utf-8"
            )
            self._send(code, body, "application/json; charset=utf-8")

        def _path_summary(self, path: str) -> str:
            ops = _OPENAPI["paths"].get(path) or {}
            for method in ("get", "post"):
                if method in ops and isinstance(ops[method], dict):
                    return str(ops[method].get("summary") or method)
            return ""

        def _html_index(self) -> None:
            links = sorted(p for p in _OPENAPI["paths"] if p != "/")
            rows = "\n".join(
                f'<li><a href="{p}">{p}</a> — {self._path_summary(p)}</li>'
                for p in links
            )
            html = (
                "<!DOCTYPE html><html><head><meta charset=utf-8>"
                "<title>gpu-rent debug</title></head><body>"
                f"<h1>gpu-rent Debug API</h1>"
                f"<p>base <code>{hub.base_url}</code> · mostly read-only · 127.0.0.1</p>"
                f"<p>Agent: start at <a href='/openapi.json'>/openapi.json</a> "
                f"then <a href='/snapshot'>/snapshot</a>. "
                f"Opt-in Assistent: <code>/assistent/session</code> (multi-turn) · "
                f"<code>/assistent/chat-eval</code> (one-shot) — not in snapshot; "
                f"may load VRAM. Offline: "
                f"<code>/assistent/analyze-reply</code>.</p>"
                f"<ul>{rows}</ul></body></html>"
            )
            self._send(200, html.encode("utf-8"), "text/html; charset=utf-8")

        def _read_json_body(self) -> dict[str, Any]:
            length = int(self.headers.get("Content-Length") or 0)
            if length <= 0:
                return {}
            raw = self.rfile.read(min(length, 1_000_000))
            if not raw:
                return {}
            try:
                data = json.loads(raw.decode("utf-8", "replace"))
            except json.JSONDecodeError as exc:
                raise ValueError(f"invalid JSON body: {exc}") from exc
            if data is None:
                return {}
            if not isinstance(data, dict):
                raise ValueError("JSON body must be an object")
            return data

        def do_OPTIONS(self) -> None:  # noqa: N802
            self.send_response(204)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type")
            self.send_header("Content-Length", "0")
            self.end_headers()

        def do_GET(self) -> None:  # noqa: N802
            try:
                self._dispatch("GET")
            except Exception as exc:
                self._json(
                    500,
                    {
                        "ok": False,
                        "error": str(exc)[:300],
                        "trace": traceback.format_exc()[-800:],
                    },
                )

        def do_POST(self) -> None:  # noqa: N802
            try:
                self._dispatch("POST")
            except Exception as exc:
                self._json(
                    500,
                    {
                        "ok": False,
                        "error": str(exc)[:300],
                        "trace": traceback.format_exc()[-800:],
                    },
                )

        def do_DELETE(self) -> None:  # noqa: N802
            try:
                self._dispatch("DELETE")
            except Exception as exc:
                self._json(
                    500,
                    {
                        "ok": False,
                        "error": str(exc)[:300],
                        "trace": traceback.format_exc()[-800:],
                    },
                )

        def _dispatch(self, method: str = "GET") -> None:
            parsed = urlparse(self.path)
            path = parsed.path.rstrip("/") or "/"
            qs = parse_qs(parsed.query)

            if path == "/":
                if method != "GET":
                    self._json(405, {"ok": False, "error": "GET only"})
                    return
                self._html_index()
                return
            if path == "/openapi.json":
                if method != "GET":
                    self._json(405, {"ok": False, "error": "GET only"})
                    return
                doc = dict(_OPENAPI)
                doc["servers"] = [{"url": hub.base_url}]
                self._json(200, doc)
                return
            if method == "POST" and path not in {
                "/assistent/chat-eval",
                "/assistent/analyze-reply",
                "/assistent/client-event",
                "/assistent/session",
            } and not _is_assistent_session_chat(path):
                self._json(
                    405,
                    {
                        "ok": False,
                        "error": (
                            "POST only for /assistent/chat-eval, "
                            "/assistent/analyze-reply, /assistent/client-event, "
                            "/assistent/session, /assistent/session/{id}/chat"
                        ),
                    },
                )
                return
            if method == "DELETE" and not _is_assistent_session_id(path):
                self._json(
                    405,
                    {
                        "ok": False,
                        "error": "DELETE only for /assistent/session/{id}",
                    },
                )
                return
            if path == "/progress":
                self._json(200, hub.progress())
                return
            if path == "/events":
                since = 0
                if qs.get("since"):
                    try:
                        since = int(qs["since"][0])
                    except ValueError:
                        self._json(400, {"ok": False, "error": "since must be int"})
                        return
                events = hub.events_since(since)
                self._json(
                    200,
                    {
                        "ok": True,
                        "since": since,
                        "next": events[-1]["seq"] if events else since,
                        "events": events,
                    },
                )
                return
            if path == "/state":
                self._json(200, {"ok": True, "state": debug_checks.redact_state()})
                return
            if path == "/config":
                self._json(200, {"ok": True, "config": debug_checks.redact_config(hub.cfg)})
                return
            if path == "/health":
                payload = hub.cached(
                    "health",
                    CACHE_TTL_HEALTH,
                    lambda: debug_checks.collect_health(
                        hub.cfg,
                        started_at=hub.started_at,
                        progress=hub.progress(),
                    ),
                )
                self._json(200, payload)
                return
            if path == "/snapshot":
                payload = hub.cached(
                    "snapshot",
                    CACHE_TTL_HEALTH,
                    lambda: debug_checks.collect_snapshot(
                        hub.cfg,
                        progress=hub.progress(),
                        events_tail=hub.events_tail(20),
                    ),
                )
                self._json(200, payload)
                return
            if path == "/status":
                payload = hub.cached(
                    "status",
                    CACHE_TTL_DEFAULT,
                    lambda: debug_checks.collect_status(hub.cfg),
                )
                self._json(200, payload)
                return
            if path == "/checks":
                payload = hub.cached(
                    "checks",
                    CACHE_TTL_DEFAULT,
                    lambda: debug_checks.collect_checks(hub.cfg),
                )
                self._json(200, payload)
                return
            if path == "/logs":
                unit = qs.get("unit", [None])[0]
                lines = 80
                if qs.get("lines"):
                    try:
                        lines = max(10, min(int(qs["lines"][0]), 500))
                    except ValueError:
                        self._json(400, {"ok": False, "error": "lines must be int"})
                        return
                cache_key = f"logs:{unit}:{lines}"
                payload = hub.cached(
                    cache_key,
                    CACHE_TTL_LOGS,
                    lambda: debug_checks.collect_logs(hub.cfg, unit=unit, lines=lines),
                )
                self._json(200, payload)
                return
            if path == "/diag":
                payload = hub.cached(
                    "diag",
                    CACHE_TTL_DIAG,
                    lambda: debug_checks.collect_diag(hub.cfg),
                )
                self._json(200, payload)
                return
            if path == "/gpu":
                payload = hub.cached(
                    "gpu",
                    CACHE_TTL_DEFAULT,
                    lambda: debug_checks.collect_gpu(hub.cfg),
                )
                self._json(200, payload)
                return
            if path == "/swarm":
                payload = hub.cached(
                    "swarm",
                    CACHE_TTL_DEFAULT,
                    lambda: debug_checks.collect_swarm(hub.cfg),
                )
                self._json(200, payload)
                return
            if path == "/ollama":
                payload = hub.cached(
                    "ollama",
                    CACHE_TTL_DEFAULT,
                    lambda: debug_checks.collect_ollama(hub.cfg),
                )
                self._json(200, payload)
                return
            if path == "/assistent" or path.startswith("/assistent/"):
                from gpu_rent import debug_assistent
                from gpu_rent import debug_assistent_session as das

                def _flag(name: str, *, default: bool = False) -> bool:
                    vals = qs.get(name)
                    if not vals:
                        return default
                    raw = vals[0].strip().lower()
                    return raw in {"1", "true", "yes", "on"}

                chat_smoke = _flag("chat_smoke")
                include_logs = _flag("logs")
                sub = path[len("/assistent") :].lstrip("/") or ""
                try_list = [
                    "/assistent",
                    "/assistent/diagnose",
                    "/assistent/extension",
                    "/assistent/overlay",
                    "/assistent/roles",
                    "/assistent/memory",
                    "/assistent/api",
                    "/assistent/wanted",
                    "/assistent/logs",
                    "/assistent/session",
                    "/assistent/session/{id}",
                    "/assistent/session/{id}/chat",
                    "/assistent/analyze-reply",
                    "/assistent/client-event",
                    "/assistent/client-events",
                    "/assistent/chat-eval",
                ]

                if sub == "":
                    if method != "GET":
                        self._json(405, {"ok": False, "error": "GET only"})
                        return
                    cache_key = f"assistent:{int(chat_smoke)}:{int(include_logs)}"
                    payload = hub.cached(
                        cache_key,
                        CACHE_TTL_DEFAULT,
                        lambda: debug_checks.collect_assistent(
                            hub.cfg,
                            chat_smoke=chat_smoke,
                            include_logs=include_logs,
                        ),
                    )
                elif sub == "diagnose":
                    if method != "GET":
                        self._json(405, {"ok": False, "error": "GET only"})
                        return
                    # logs default on for diagnose; logs=0 to skip
                    diag_logs = _flag("logs", default=True)
                    if qs.get("logs") and qs["logs"][0].strip().lower() in {
                        "0",
                        "false",
                        "no",
                        "off",
                    }:
                        diag_logs = False
                    cache_key = f"assistent/diagnose:{int(chat_smoke)}:{int(diag_logs)}"
                    payload = hub.cached(
                        cache_key,
                        CACHE_TTL_DEFAULT,
                        lambda: das.collect_assistent_diagnose(
                            hub.cfg,
                            chat_smoke=chat_smoke,
                            include_logs=diag_logs,
                        ),
                    )
                elif sub == "extension":
                    payload = hub.cached(
                        "assistent/extension",
                        CACHE_TTL_DEFAULT,
                        lambda: debug_assistent.collect_assistent_extension(hub.cfg),
                    )
                elif sub == "overlay":
                    payload = hub.cached(
                        "assistent/overlay",
                        CACHE_TTL_DEFAULT,
                        lambda: debug_assistent.collect_assistent_overlay(hub.cfg),
                    )
                elif sub == "roles":
                    payload = hub.cached(
                        "assistent/roles",
                        CACHE_TTL_DEFAULT,
                        lambda: debug_assistent.collect_assistent_roles(hub.cfg),
                    )
                elif sub == "memory":
                    payload = hub.cached(
                        "assistent/memory",
                        CACHE_TTL_DEFAULT,
                        lambda: debug_assistent.collect_assistent_memory(hub.cfg),
                    )
                elif sub == "api":
                    cache_key = f"assistent/api:{int(chat_smoke)}"
                    payload = hub.cached(
                        cache_key,
                        CACHE_TTL_DEFAULT,
                        lambda: debug_assistent.collect_assistent_api(
                            hub.cfg, chat_smoke=chat_smoke
                        ),
                    )
                elif sub == "wanted":
                    payload = hub.cached(
                        "assistent/wanted",
                        CACHE_TTL_DEFAULT,
                        lambda: debug_assistent.collect_assistent_wanted(hub.cfg),
                    )
                elif sub == "logs":
                    payload = hub.cached(
                        "assistent/logs",
                        CACHE_TTL_LOGS,
                        lambda: debug_assistent.collect_assistent_logs(hub.cfg),
                    )
                elif sub == "session" and method == "POST":
                    try:
                        body = self._read_json_body()
                    except ValueError as exc:
                        self._json(400, {"ok": False, "error": str(exc)})
                        return
                    skills = body.get("skills")
                    if skills is not None and not isinstance(skills, list):
                        self._json(
                            400, {"ok": False, "error": "skills must be an array"}
                        )
                        return
                    ctx = body.get("context")
                    if ctx is not None and not isinstance(ctx, dict):
                        self._json(
                            400, {"ok": False, "error": "context must be an object"}
                        )
                        return
                    probe = body.get("probe_config", True)
                    payload = das.create_debug_session(
                        hub.cfg,
                        persona=body.get("persona"),
                        pack=body.get("pack"),
                        model=body.get("model"),
                        context=ctx,
                        skills=skills,
                        probe_config=bool(probe),
                    )
                elif _is_assistent_session_id(path):
                    sid = sub.split("/", 1)[1]
                    if method == "GET":
                        payload = das.get_debug_session(sid)
                    elif method == "DELETE":
                        payload = das.delete_debug_session(sid)
                    else:
                        self._json(405, {"ok": False, "error": "GET or DELETE"})
                        return
                elif _is_assistent_session_chat(path):
                    if method != "POST":
                        self._json(405, {"ok": False, "error": "POST only"})
                        return
                    sid = sub.split("/")[1]
                    try:
                        body = self._read_json_body()
                    except ValueError as exc:
                        self._json(400, {"ok": False, "error": str(exc)})
                        return
                    skills = body.get("skills")
                    if skills is not None and not isinstance(skills, list):
                        self._json(
                            400, {"ok": False, "error": "skills must be an array"}
                        )
                        return
                    ctx = body.get("context")
                    if ctx is not None and not isinstance(ctx, dict):
                        self._json(
                            400, {"ok": False, "error": "context must be an object"}
                        )
                        return
                    payload = das.chat_debug_session(
                        hub.cfg,
                        sid,
                        message=body.get("message"),
                        timeout=body.get("timeout"),
                        context=ctx,
                        pack=body.get("pack"),
                        model=body.get("model"),
                        skills=skills,
                        persona=body.get("persona"),
                    )
                elif sub == "client-events":
                    if method != "GET":
                        self._json(405, {"ok": False, "error": "GET only"})
                        return
                    since = 0
                    try:
                        since = int((qs.get("since") or ["0"])[0])
                    except (TypeError, ValueError):
                        since = 0
                    payload = {
                        "ok": True,
                        "events": hub.client_events_since(since),
                    }
                elif sub == "client-event":
                    if method != "POST":
                        self._json(405, {"ok": False, "error": "POST only"})
                        return
                    try:
                        body = self._read_json_body()
                    except ValueError as exc:
                        self._json(400, {"ok": False, "error": str(exc)})
                        return
                    if not isinstance(body, dict):
                        self._json(400, {"ok": False, "error": "JSON object required"})
                        return
                    payload = hub.push_client_event(body)
                elif sub == "analyze-reply":
                    body = {}
                    if method == "POST":
                        try:
                            body = self._read_json_body()
                        except ValueError as exc:
                            self._json(400, {"ok": False, "error": str(exc)})
                            return
                    elif method != "GET":
                        self._json(405, {"ok": False, "error": "GET or POST"})
                        return

                    def _aq(name: str) -> Any:
                        if name in body and body[name] is not None:
                            return body[name]
                        vals = qs.get(name)
                        return vals[0] if vals else None

                    reply_text = _aq("reply")
                    if not (reply_text or "").strip():
                        self._json(
                            400,
                            {
                                "ok": False,
                                "error": "reply required",
                                "usage": (
                                    "POST {message, reply, persona?, pack?} "
                                    "— offline extract + client.would_generate"
                                ),
                            },
                        )
                        return
                    payload = debug_assistent.analyze_assistent_reply(
                        message=_aq("message"),
                        reply=str(reply_text),
                        persona=_aq("persona"),
                        pack=_aq("pack"),
                    )
                elif sub == "chat-eval":
                    body = {}
                    if method == "POST":
                        try:
                            body = self._read_json_body()
                        except ValueError as exc:
                            self._json(400, {"ok": False, "error": str(exc)})
                            return

                    def _q(name: str) -> Any:
                        if name in body and body[name] is not None:
                            return body[name]
                        vals = qs.get(name)
                        return vals[0] if vals else None

                    ctx = _q("context")
                    if isinstance(ctx, str):
                        try:
                            ctx = json.loads(ctx)
                        except json.JSONDecodeError:
                            ctx = None
                    if ctx is not None and not isinstance(ctx, dict):
                        ctx = None

                    # Never cache — live AssistentChat / VRAM side effects.
                    payload = debug_assistent.run_assistent_chat_eval(
                        hub.cfg,
                        message=_q("message"),
                        persona=_q("persona"),
                        pack=_q("pack"),
                        model=_q("model"),
                        timeout=_q("timeout"),
                        context=ctx,
                    )
                else:
                    self._json(
                        404,
                        {
                            "ok": False,
                            "error": f"unknown path {path}",
                            "try": try_list,
                        },
                    )
                    return
                self._json(200, payload)
                return

            self._json(404, {"ok": False, "error": f"unknown path {path}"})

    return Handler


def start_debug_server(
    cfg: Config,
    *,
    port: int | None = None,
    log: Log | None = None,
    print_urls: bool = True,
) -> DebugServer | None:
    """Bind 127.0.0.1 and serve in a daemon thread. Returns None on bind failure."""
    global _active
    wanted = int(port if port is not None else getattr(cfg, "debug_local_port", DEFAULT_PORT))
    hub = DebugHub(cfg=cfg, port=wanted)
    handler = _make_handler(hub)
    try:
        httpd = ThreadingHTTPServer(("127.0.0.1", wanted), handler)
    except OSError as exc:
        msg = (
            f"⚠ Debug API не стартовал на 127.0.0.1:{wanted} ({exc}). "
            f"up продолжается без sidecar — смени DEBUG_LOCAL_PORT или "
            f"`gpu-rent debug` в другом терминале."
        )
        if log:
            log(msg)
        else:
            print(msg)
        return None

    httpd.daemon_threads = True
    thread = threading.Thread(
        target=httpd.serve_forever,
        name="gpu-rent-debug-api",
        daemon=True,
    )
    thread.start()
    server = DebugServer(hub=hub, httpd=httpd, thread=thread)
    with _active_lock:
        old = _active
        _active = server
    if old is not None and old is not server:
        try:
            old.stop()
        except Exception:
            pass

    if print_urls:
        base = server.base_url
        lines = [
            f"Debug API  {base}/",
            f"           {base}/openapi.json",
            f"           {base}/snapshot",
            f"           {base}/assistent",
            f"           {base}/assistent/diagnose",
            f"           {base}/assistent/session  (multi-turn; not in snapshot)",
            f"           {base}/assistent/chat-eval  (one-shot; not in snapshot)",
            f"           {base}/assistent/analyze-reply  (offline; no VRAM)",
            f"           {base}/assistent/client-events  (UI beacons)",
        ]
        if log:
            for line in lines:
                log(line)
        else:
            for line in lines:
                print(line)
    return server


def stop_debug_server(server: DebugServer | None = None) -> None:
    global _active
    target = server
    with _active_lock:
        if target is None:
            target = _active
        if _active is target:
            _active = None
    if target is not None:
        target.stop()


def run_debug_blocking(
    cfg: Config,
    *,
    port: int | None = None,
    log: Log | None = None,
) -> None:
    """Foreground sidecar for `gpu-rent debug` (Ctrl+C to stop)."""
    server = start_debug_server(cfg, port=port, log=log, print_urls=True)
    if server is None:
        raise RuntimeError("Debug API bind failed")
    server.set_step("idle_debug")
    try:
        if log:
            log("Debug API слушает (Ctrl+C — выход). "
                "Мутаций конфига/GPU нет; "
                "/assistent/session + /chat-eval — opt-in AssistentChat.")
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        if log:
            log("Debug API остановлен.")
    finally:
        stop_debug_server(server)