feat: add Ollama chat and live memory telemetry

dashboard/plugin_api.py

@@ -1,8 +1,16 @@
 """Native Hermes dashboard API for managing a local Ollama instance."""
 from __future__ import annotations
 
+import base64
+import binascii
+import io
+import ipaddress
 import json
+import mimetypes
+import os
 import re
+import socket
+import subprocess
 import threading
 import time
 import uuid
@@ -12,12 +20,12 @@ from html.parser import HTMLParser
 from pathlib import Path
 from typing import Any
 from urllib.error import HTTPError, URLError
-from urllib.parse import unquote, urlencode
-from urllib.request import Request, urlopen
+from urllib.parse import unquote, urlencode, urlparse
+from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen
 from zoneinfo import ZoneInfo
 
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from hermes_constants import get_hermes_home
 
 router = APIRouter()
@@ -27,6 +35,10 @@ CATALOG_FILE = "catalog.json"
 MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,190}$")
 MELBOURNE = ZoneInfo("Australia/Melbourne")
 POPULAR_RAM_LIMIT_GIB = 30.0
+MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
+MAX_ATTACHMENT_TEXT = 80_000
+MAX_URL_BYTES = 15 * 1024 * 1024
+CHAT_KEEP_ALIVE = "10m"
 
 _jobs: dict[str, dict[str, Any]] = {}
 _jobs_lock = threading.Lock()
@@ -99,6 +111,204 @@ def _local_ps() -> list[dict[str, Any]]:
         return []
 
 
+def _local_ps() -> list[dict[str, Any]]:
+    try:
+        payload = _json_request(LOCAL_OLLAMA + "/api/ps", timeout=10)
+        models = payload.get("models", [])
+        return [item for item in models if isinstance(item, dict)]
+    except (HTTPError, URLError, OSError, ValueError):
+        return []
+
+
+def _read_meminfo() -> dict[str, int]:
+    values: dict[str, int] = {}
+    try:
+        for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():
+            key, _, raw = line.partition(":")
+            match = re.search(r"([0-9]+)", raw)
+            if match:
+                values[key] = int(match.group(1)) * 1024
+    except OSError:
+        return {}
+    return values
+
+
+def _gpu_snapshot() -> dict[str, Any]:
+    """Return NVIDIA GPU telemetry when available, without requiring CUDA."""
+    query = "name,memory.total,memory.used,memory.free"
+    try:
+        result = subprocess.run(
+            ["nvidia-smi", f"--query-gpu={query}", "--format=csv,noheader,nounits"],
+            capture_output=True,
+            text=True,
+            timeout=4,
+            check=False,
+        )
+    except (OSError, subprocess.SubprocessError):
+        result = None
+    if result and result.returncode == 0:
+        gpus = []
+        for line in result.stdout.splitlines():
+            parts = [part.strip() for part in line.split(",")]
+            if len(parts) != 4:
+                continue
+            try:
+                total, used, free = (int(float(value)) * 1024 * 1024 for value in parts[1:])
+            except ValueError:
+                continue
+            gpus.append({"name": parts[0], "total_bytes": total, "used_bytes": used, "free_bytes": free})
+        if gpus:
+            return {"detected": True, "telemetry_available": True, "gpus": gpus}
+    nvidia_present = False
+    for vendor in Path("/sys/class/drm").glob("card*/device/vendor"):
+        try:
+            nvidia_present = nvidia_present or vendor.read_text().strip().lower() == "0x10de"
+        except OSError:
+            continue
+    return {"detected": nvidia_present, "telemetry_available": False, "gpus": []}
+
+
+def _runtime_snapshot() -> dict[str, Any]:
+    mem = _read_meminfo()
+    total = mem.get("MemTotal", 0)
+    available = mem.get("MemAvailable", mem.get("MemFree", 0))
+    swap_total = mem.get("SwapTotal", 0)
+    swap_free = mem.get("SwapFree", 0)
+    ps_rows = _local_ps()
+    model_memory = []
+    for row in ps_rows:
+        name = str(row.get("name") or row.get("model") or "")
+        total_bytes = int(row.get("size") or 0)
+        gpu_bytes = int(row.get("size_vram") or 0)
+        model_memory.append({
+            "name": name,
+            "total_bytes": total_bytes,
+            "gpu_bytes": gpu_bytes,
+            "ram_bytes": max(0, total_bytes - gpu_bytes),
+            "gpu_offload_percent": round(gpu_bytes * 100 / total_bytes, 1) if total_bytes else 0,
+        })
+    return {
+        "captured_at": time.time(),
+        "memory_total_bytes": total,
+        "memory_used_bytes": max(0, total - available),
+        "memory_available_bytes": available,
+        "swap_total_bytes": swap_total,
+        "swap_used_bytes": max(0, swap_total - swap_free),
+        "model_memory": model_memory,
+        "gpu": _gpu_snapshot(),
+    }
+
+
+def _validate_public_url(value: str) -> str:
+    parsed = urlparse(value.strip())
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+        raise HTTPException(400, "URL attachments must use http:// or https://")
+    host = parsed.hostname
+    try:
+        addresses = {info[4][0] for info in socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM)}
+    except (OSError, ValueError) as exc:
+        raise HTTPException(400, f"Could not resolve URL host: {exc}") from exc
+    for address in addresses:
+        ip = ipaddress.ip_address(address)
+        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified:
+            raise HTTPException(400, "Private or local URL targets are not allowed")
+    return value.strip()
+
+
+class _SafeRedirectHandler(HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        _validate_public_url(newurl)
+        return super().redirect_request(req, fp, code, msg, headers, newurl)
+
+
+_SAFE_URL_OPENER = build_opener(_SafeRedirectHandler)
+
+
+def _fetch_attachment_url(value: str) -> tuple[bytes, str, str]:
+    value = _validate_public_url(value)
+    request = Request(value, headers={"Accept": "text/html, text/plain, application/pdf, image/*", "User-Agent": "Hermes-Ollama-Manager/1.3"})
+    try:
+        with _SAFE_URL_OPENER.open(request, timeout=30) as response:
+            final_url = _validate_public_url(response.geturl())
+            content_type = response.headers.get_content_type() if response.headers else "application/octet-stream"
+            data = response.read(MAX_URL_BYTES + 1)
+    except (HTTPError, URLError, OSError, ValueError) as exc:
+        raise HTTPException(400, f"Could not fetch URL: {exc}") from exc
+    if len(data) > MAX_URL_BYTES:
+        raise HTTPException(413, "URL attachment is larger than 15 MiB")
+    return data, content_type, final_url
+
+
+def _extract_pdf_text(data: bytes, label: str) -> str:
+    try:
+        from pypdf import PdfReader
+    except ImportError as exc:
+        raise HTTPException(500, "PDF support requires the pypdf package") from exc
+    try:
+        reader = PdfReader(io.BytesIO(data))
+        text = "\n\n".join(page.extract_text() or "" for page in reader.pages)
+    except Exception as exc:
+        raise HTTPException(400, f"Could not extract text from PDF {label}: {exc}") from exc
+    return text[:MAX_ATTACHMENT_TEXT]
+
+
+class _PageTextParser(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__()
+        self.parts: list[str] = []
+        self._skip = 0
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        if tag.lower() in {"script", "style", "noscript", "svg"}:
+            self._skip += 1
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag.lower() in {"script", "style", "noscript", "svg"} and self._skip:
+            self._skip -= 1
+
+    def handle_data(self, data: str) -> None:
+        if not self._skip and data.strip():
+            self.parts.append(data.strip())
+
+
+def _extract_page_text(data: bytes, content_type: str) -> str:
+    decoded = data.decode("utf-8", errors="replace")
+    if "html" in content_type.lower() or re.search(r"<html|<body|<article", decoded, re.I):
+        parser = _PageTextParser()
+        parser.feed(decoded)
+        decoded = "\n".join(parser.parts)
+    return unescape(decoded)[:MAX_ATTACHMENT_TEXT]
+
+
+def _decode_data_url(data_url: str, fallback_mime: str, label: str) -> tuple[bytes, str]:
+    match = re.match(r"data:([^;,]+)?;base64,(.*)", data_url or "", re.S)
+    if not match:
+        raise HTTPException(400, f"Attachment {label} is not a valid base64 data URL")
+    mime = (match.group(1) or fallback_mime or "application/octet-stream").lower()
+    try:
+        data = base64.b64decode(match.group(2), validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise HTTPException(400, f"Attachment {label} has invalid base64 data") from exc
+    if len(data) > MAX_ATTACHMENT_BYTES:
+        raise HTTPException(413, f"Attachment {label} is larger than 20 MiB")
+    return data, mime
+
+
+def _attachment_parts(attachment: "ChatAttachment") -> tuple[str | None, str | None]:
+    label = attachment.name or attachment.url or "attachment"
+    if attachment.url:
+        data, mime, _ = _fetch_attachment_url(attachment.url)
+    elif attachment.data_url:
+        data, mime = _decode_data_url(attachment.data_url, attachment.mime_type, label)
+    else:
+        raise HTTPException(400, f"Attachment {label} has no data or URL")
+    if mime == "application/pdf" or label.lower().endswith(".pdf"):
+        return f"[PDF: {label}]\n{_extract_pdf_text(data, label)}", None
+    if mime.startswith("image/"):
+        return f"[Image attached: {label}]", base64.b64encode(data).decode("ascii")
+    return f"[Text attachment: {label}]\n{_extract_page_text(data, mime)}", None
+
+
 def _is_mlx(raw: dict[str, Any] | str) -> bool:
     text = str(raw if isinstance(raw, str) else {
         "name": raw.get("name") or raw.get("model"),
@@ -523,6 +733,111 @@ class ModelRequest(BaseModel):
     name: str
 
 
+class ChatAttachment(BaseModel):
+    name: str = ""
+    mime_type: str = ""
+    data_url: str | None = None
+    url: str | None = None
+
+
+class ChatRequest(BaseModel):
+    model: str
+    message: str = ""
+    history: list[dict[str, Any]] = Field(default_factory=list)
+    attachments: list[ChatAttachment] = Field(default_factory=list)
+
+
+def _installed_model_names() -> set[str]:
+    return {
+        str(row.get("name") or row.get("model"))
+        for row in _local_tags()
+        if not _is_mlx(row)
+    }
+
+
+def _require_installed_model(name: str) -> str:
+    name = _valid_name(name)
+    if name not in _installed_model_names():
+        raise HTTPException(400, f"Model '{name}' is not installed locally")
+    return name
+
+
+def _ollama_error(exc: HTTPError) -> HTTPException:
+    try:
+        detail = exc.read().decode("utf-8", errors="replace")[:1000]
+        payload = json.loads(detail)
+        detail = str(payload.get("error") or detail)
+    except (OSError, ValueError):
+        detail = str(exc)
+    return HTTPException(502, f"Ollama request failed: {detail}")
+
+
+def _load_model(name: str) -> dict[str, Any]:
+    name = _require_installed_model(name)
+    try:
+        result = _json_request(
+            LOCAL_OLLAMA + "/api/generate",
+            method="POST",
+            payload={"model": name, "prompt": "", "stream": False, "keep_alive": CHAT_KEEP_ALIVE, "options": {"num_predict": 1}},
+            timeout=900,
+        )
+    except HTTPError as exc:
+        raise _ollama_error(exc) from exc
+    return {"ok": True, "model": name, "response": result.get("response", ""), "runtime": _runtime_snapshot()}
+
+
+def _chat_payload(body: ChatRequest) -> dict[str, Any]:
+    model = _require_installed_model(body.model)
+    messages: list[dict[str, Any]] = []
+    for item in body.history[-24:]:
+        role = str(item.get("role") or "")
+        content = str(item.get("content") or "").strip()
+        if role in {"user", "assistant"} and content:
+            messages.append({"role": role, "content": content[:MAX_ATTACHMENT_TEXT]})
+    text_parts = [body.message.strip()] if body.message.strip() else []
+    images: list[str] = []
+    for attachment in body.attachments[:12]:
+        text, image = _attachment_parts(attachment)
+        if text:
+            text_parts.append(text)
+        if image:
+            images.append(image)
+    if not text_parts and not images:
+        raise HTTPException(400, "Enter a message or attach a file/URL")
+    user_message: dict[str, Any] = {"role": "user", "content": "\n\n".join(text_parts)[:MAX_ATTACHMENT_TEXT]}
+    if images:
+        user_message["images"] = images
+    messages.append(user_message)
+    return {"model": model, "messages": messages, "stream": False, "keep_alive": CHAT_KEEP_ALIVE}
+
+
+@router.get("/runtime")
+def runtime() -> dict[str, Any]:
+    return _runtime_snapshot()
+
+
+@router.post("/chat/load")
+def chat_load(body: ModelRequest) -> dict[str, Any]:
+    return _load_model(body.name)
+
+
+@router.post("/chat")
+def chat(body: ChatRequest) -> dict[str, Any]:
+    payload = _chat_payload(body)
+    try:
+        result = _json_request(LOCAL_OLLAMA + "/api/chat", method="POST", payload=payload, timeout=1800)
+    except HTTPError as exc:
+        raise _ollama_error(exc) from exc
+    message = result.get("message") if isinstance(result.get("message"), dict) else {}
+    return {
+        "ok": True,
+        "model": payload["model"],
+        "message": {"role": "assistant", "content": str(message.get("content") or "")},
+        "done": bool(result.get("done", True)),
+        "runtime": _runtime_snapshot(),
+    }
+
+
 @router.get("/status")
 def status() -> dict[str, Any]:
     tags = _local_tags()