"""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'{p} — {self._path_summary(p)}'
for p in links
)
html = (
""
"gpu-rent debug"
f"gpu-rent Debug API
"
f"base {hub.base_url} · mostly read-only · 127.0.0.1
"
f"Agent: start at /openapi.json "
f"then /snapshot. "
f"Opt-in Assistent: /assistent/session (multi-turn) · "
f"/assistent/chat-eval (one-shot) — not in snapshot; "
f"may load VRAM. Offline: "
f"/assistent/analyze-reply.
"
f""
)
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)