"""
Vincent Dashboard — AI assistant operational state monitor.
Shows Vincent's own state: memory, skills, sessions, calibration, grounding, cron jobs, services.
"""
import json
import os
import re
import sqlite3
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
from flask import Flask, jsonify, render_template, make_response, request
app = Flask(__name__)
@app.after_request
def no_cache(response):
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
response.headers["Pragma"] = "no-cache"
response.headers["Expires"] = "0"
return response
HERMES_HOME = Path(os.path.expanduser("~/.hermes"))
STATE_DB = HERMES_HOME / "state.db"
SESSION_STATE = HERMES_HOME / "session_state.json"
MEMORY_FILE = HERMES_HOME / "memories" / "MEMORY.md"
USER_FILE = HERMES_HOME / "memories" / "USER.md"
CALIBRATION_FILE = HERMES_HOME / "calibration" / "predictions.json"
CONFIG_FILE = HERMES_HOME / "config.yaml"
CRON_JOBS_FILE = HERMES_HOME / "cron" / "jobs.json"
SKILLS_DIR = HERMES_HOME / "skills"
def get_session_state():
"""Read session continuity state."""
try:
with open(SESSION_STATE) as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {"focus": "None", "next": "None", "context": "Idle"}
def get_memory_stats():
"""Read memory file sizes and estimate usage."""
stats = {"memory": {}, "user": {}}
for name, path in [("memory", MEMORY_FILE), ("user", USER_FILE)]:
try:
with open(path) as f:
content = f.read()
stats[name] = {
"exists": True,
"size_bytes": os.path.getsize(path),
"char_count": len(content),
"entries": content.count("§"),
}
except FileNotFoundError:
stats[name] = {"exists": False, "size_bytes": 0, "char_count": 0, "entries": 0}
return stats
def get_session_stats():
"""Query state.db for session statistics."""
try:
conn = sqlite3.connect(str(STATE_DB))
conn.row_factory = sqlite3.Row
cur = conn.cursor()
# Total sessions
cur.execute("SELECT COUNT(*) as total FROM sessions WHERE archived = 0")
total = cur.fetchone()["total"]
# Total tokens
cur.execute(
"SELECT SUM(input_tokens) as inp, SUM(output_tokens) as out, "
"SUM(cache_read_tokens) as cr, SUM(cache_write_tokens) as cw, "
"SUM(reasoning_tokens) as reason FROM sessions WHERE archived = 0"
)
tokens = cur.fetchone()
# Total cost
cur.execute(
"SELECT SUM(estimated_cost_usd) as cost FROM sessions "
"WHERE archived = 0 AND estimated_cost_usd IS NOT NULL"
)
cost_row = cur.fetchone()
# Recent sessions (last 10)
cur.execute(
"SELECT id, title, source, model, started_at, message_count, "
"input_tokens, output_tokens, estimated_cost_usd, end_reason "
"FROM sessions WHERE archived = 0 "
"ORDER BY started_at DESC LIMIT 10"
)
recent = []
for row in cur.fetchall():
ts = datetime.fromtimestamp(row["started_at"], tz=timezone.utc)
recent.append({
"id": row["id"][:16],
"title": row["title"] or "Untitled",
"source": row["source"],
"model": (row["model"] or "").replace("-UD-Q4_K_XL.gguf", "").replace(".gguf", ""),
"started": ts.strftime("%Y-%m-%d %H:%M"),
"messages": row["message_count"] or 0,
"input_tokens": row["input_tokens"] or 0,
"output_tokens": row["output_tokens"] or 0,
"cost": row["estimated_cost_usd"] or 0,
"end_reason": row["end_reason"] or "",
})
# Sessions today
today_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0).timestamp()
cur.execute(
"SELECT COUNT(*) as cnt FROM sessions "
"WHERE started_at >= ? AND archived = 0", (today_start,)
)
today_count = cur.fetchone()["cnt"]
conn.close()
return {
"total": total,
"today": today_count,
"tokens": {
"input": tokens["inp"] or 0,
"output": tokens["out"] or 0,
"cache_read": tokens["cr"] or 0,
"cache_write": tokens["cw"] or 0,
"reasoning": tokens["reason"] or 0,
},
"cost_usd": cost_row["cost"] or 0,
"recent": recent,
}
except Exception as e:
return {"error": str(e)}
def get_calibration_stats():
"""Read calibration prediction tracking."""
try:
with open(CALIBRATION_FILE) as f:
predictions = json.load(f)
total = len(predictions)
correct = sum(1 for p in predictions if p.get("status") == "correct")
incorrect = sum(1 for p in predictions if p.get("status") == "incorrect")
pending = total - correct - incorrect
accuracy = (correct / total * 100) if total > 0 else 0
return {
"total": total,
"correct": correct,
"incorrect": incorrect,
"pending": pending,
"accuracy": accuracy,
"recent": predictions[-5:] if predictions else [],
}
except (FileNotFoundError, json.JSONDecodeError) as e:
return {"error": str(e)}
def get_system_health():
"""Read system health from /proc."""
health = {}
# CPU load
try:
with open("/proc/loadavg") as f:
parts = f.read().split()
health["load"] = {
"1m": float(parts[0]),
"5m": float(parts[1]),
"15m": float(parts[2]),
}
except Exception:
health["load"] = {}
# Memory
try:
with open("/proc/meminfo") as f:
meminfo = {}
for line in f:
parts = line.split()
key = parts[0].rstrip(":")
val = int(parts[1]) # kB
meminfo[key] = val
total_mb = meminfo.get("MemTotal", 0) / 1024
avail_mb = meminfo.get("MemAvailable", 0) / 1024
used_mb = total_mb - avail_mb
health["memory"] = {
"total_gb": round(total_mb / 1024, 1),
"used_gb": round(used_mb / 1024, 1),
"percent": round(used_mb / total_mb * 100, 1) if total_mb > 0 else 0,
}
except Exception:
health["memory"] = {}
# Disk — check both SSD and HDD
disks = []
for mount, label in [("/", "SSD"), ("/mnt/storage", "HDD")]:
try:
stat = os.statvfs(mount)
total_gb = stat.f_blocks * stat.f_frsize / (1024**3)
free_gb = stat.f_bavail * stat.f_frsize / (1024**3)
used_gb = total_gb - free_gb
disks.append({
"label": label,
"mount": mount,
"total_gb": round(total_gb, 0),
"used_gb": round(used_gb, 0),
"free_gb": round(free_gb, 0),
"percent": round(used_gb / total_gb * 100, 1) if total_gb > 0 else 0,
})
except Exception:
pass
health["disk"] = disks
# CPU cores
try:
health["cpu_cores"] = os.cpu_count() or 0
except Exception:
health["cpu_cores"] = 0
# CPU usage % (sample /proc/stat twice)
try:
def read_cpu_stat():
with open("/proc/stat") as f:
parts = f.readline().split()
vals = [int(v) for v in parts[1:]]
idle = vals[3] + (vals[4] if len(vals) > 4 else 0)
total = sum(vals[:8])
return idle, total
idle1, total1 = read_cpu_stat()
time.sleep(0.3)
idle2, total2 = read_cpu_stat()
d_total = total2 - total1
d_idle = idle2 - idle1
cpu_pct = ((d_total - d_idle) / d_total * 100) if d_total > 0 else 0
health["cpu_usage_pct"] = round(cpu_pct, 1)
except Exception:
health["cpu_usage_pct"] = 0
return health
def get_gpu_stats():
"""Read GPU stats from nvidia-smi."""
gpu = {}
try:
result = subprocess.run(
[
"nvidia-smi",
"--query-gpu=temperature.gpu,utilization.gpu,utilization.memory,memory.total,memory.used,memory.free,name",
"--format=csv,noheader",
],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0:
parts = [p.strip().replace(" MiB", "").replace(" %", "").strip() for p in result.stdout.strip().split(",")]
gpu = {
"name": parts[6] if len(parts) > 6 else "Unknown",
"temp_c": int(parts[0]) if parts[0].isdigit() else 0,
"util_pct": int(parts[1]) if parts[1].isdigit() else 0,
"mem_util_pct": int(parts[2]) if parts[2].isdigit() else 0,
"mem_total_mb": int(parts[3]) if parts[3].isdigit() else 0,
"mem_used_mb": int(parts[4]) if parts[4].isdigit() else 0,
"mem_free_mb": int(parts[5]) if parts[5].isdigit() else 0,
# Calculate actual VRAM usage ratio (model-loaded memory)
"mem_used_pct": round(int(parts[4]) / int(parts[3]) * 100) if (len(parts) > 4 and parts[3].isdigit() and int(parts[3]) > 0) else 0,
}
except Exception:
pass
return gpu
def get_cron_jobs():
"""Read cron jobs dynamically from jobs.json."""
try:
with open(CRON_JOBS_FILE) as f:
data = json.load(f)
# Handle both dict wrapper and bare list formats
if isinstance(data, dict):
jobs = data.get("jobs", [])
elif isinstance(data, list):
jobs = data
else:
return {"jobs": [], "total": 0}
enabled = []
for job in jobs:
if isinstance(job, dict) and not job.get("enabled", True):
continue
sched = job.get("schedule", "unknown")
if isinstance(sched, dict):
sched = sched.get("display", str(sched))
enabled.append({
"name": job.get("name", "Unnamed"),
"id": job.get("id", "")[:8],
"schedule": sched,
"next_run": job.get("next_run_at", "")[:19].replace("T", " ").replace("-04:00", "") if job.get("next_run_at") else "—",
"last_run": job.get("last_run_at", "")[:19].replace("T", " ").replace("-04:00", "") if job.get("last_run_at") else "—",
"last_status": job.get("last_status", "unknown"),
})
return {"jobs": enabled, "total": len(enabled)}
except (FileNotFoundError, json.JSONDecodeError) as e:
return {"jobs": [], "total": 0, "error": str(e)}
def get_skills_count():
"""Count skills dynamically."""
try:
count = 0
for item in SKILLS_DIR.rglob("SKILL.md"):
count += 1
return count
except Exception:
return 0
def get_services():
"""Check known services."""
services = {
"Qwen3.6-27B (Main)": {"port": 8080, "pid": None, "running": False},
"Qwen2.5-3B Supervisor": {"port": 8085, "pid": None, "running": False},
"Qwen3.5-4B Builder": {"port": 8086, "pid": None, "running": False},
"Freqtrade": {"port": 8081, "pid": None, "running": False},
"AgentForms Relay": {"port": 5060, "pid": None, "running": False},
"AgentForms Worker": {"port": 5070, "pid": None, "running": False},
"Cloudflare Tunnel": {"port": 5051, "pid": None, "running": False},
"Home Assistant": {"port": 8123, "pid": None, "running": False},
"SSH": {"port": 2222, "pid": None, "running": False},
}
try:
import socket
for name, svc in services.items():
port = svc["port"]
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
try:
s.connect(("127.0.0.1", port))
svc["running"] = True
except (socket.timeout, socket.error, ConnectionRefusedError):
svc["running"] = False
finally:
s.close()
except:
pass
return services
def get_config():
"""Read basic config info."""
config = {}
try:
with open(CONFIG_FILE) as f:
content = f.read()
# Extract model info
for line in content.split("\n"):
line = line.strip()
if line.startswith("model:"):
config["model"] = line.split(":", 1)[1].strip().strip("'\"")
elif line.startswith("provider:"):
config["provider"] = line.split(":", 1)[1].strip().strip("'\"")
except FileNotFoundError:
pass
return config
def fmt_size(size_bytes):
"""Format bytes to human readable."""
if size_bytes < 1024:
return f"{size_bytes}B"
elif size_bytes < 1024**2:
return f"{size_bytes / 1024:.1f}K"
elif size_bytes < 1024**3:
return f"{size_bytes / 1024**2:.1f}M"
elif size_bytes < 1024**4:
return f"{size_bytes / 1024**3:.1f}G"
else:
return f"{size_bytes / 1024**4:.1f}T"
def list_directory(path_str):
"""List directory contents with metadata. Read-only browsing only."""
# Normalize path
path = Path(path_str if path_str else "/").expanduser().resolve()
# Security: only allow paths that exist
if not path.is_dir():
return {"error": f"Not a directory or does not exist: {path_str}", "path": str(path)}
entries = []
try:
for item in sorted(path.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
try:
stat = item.stat()
is_dir = item.is_dir()
size = stat.st_size if not is_dir else 0
# Get total size for directories (top-level only, no recursion)
mod_time = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
entry = {
"name": item.name,
"path": str(item),
"is_dir": is_dir,
"size": size,
"size_str": fmt_size(size) if not is_dir else "—",
"modified": mod_time.strftime("%Y-%m-%d %H:%M"),
"permissions": oct(stat.st_mode)[-3:],
}
entries.append(entry)
except (PermissionError, OSError):
# Include entry with limited info
entries.append({
"name": item.name,
"path": str(item),
"is_dir": item.is_dir() if not item.is_symlink() else False,
"size": 0,
"size_str": "—",
"modified": "—",
"permissions": "---",
"error": "Permission denied",
})
except PermissionError:
return {"error": f"Permission denied: {path_str}", "path": str(path), "entries": []}
# Count subdirectories and files
dirs = sum(1 for e in entries if e["is_dir"])
files = len(entries) - dirs
return {
"path": str(path),
"display_path": path_str if path_str and path_str != "/" else "/",
"dirs": dirs,
"files": files,
"entries": entries,
}
@app.route("/")
def index():
return render_template("index.html")
@app.route("/api/fs")
def api_fs():
"""File system explorer — list directory contents."""
path = request.args.get("path", "/")
return jsonify(list_directory(path))
@app.route("/api/state")
def api_state():
"""Aggregate all dashboard data."""
return jsonify({
"identity": {
"name": "Vincent",
"role": "AI Assistant for Grepples",
"model": "Qwen3.6-27B-UD-Q4_K_XL.gguf",
"provider": "custom (llama.cpp)",
"runtime": "Hermes Agent",
"connected_platforms": ["telegram", "local"],
},
"session": get_session_state(),
"memory": get_memory_stats(),
"sessions": get_session_stats(),
"calibration": get_calibration_stats(),
"system": get_system_health(),
"gpu": get_gpu_stats(),
"services": get_services(),
"config": get_config(),
"grounding": {
"components": [
{"name": "background-calibration", "status": "active"},
{"name": "causal-models", "status": "active"},
{"name": "counterfactual-testing", "status": "active"},
{"name": "failure-mode-library", "status": "active"},
{"name": "grounding-integration", "status": "active"},
{"name": "multimodal-consistency", "status": "active"},
],
"total": 6,
},
"cron_jobs": get_cron_jobs(),
"skills": {
"total": get_skills_count(),
},
"timestamp": datetime.now(timezone.utc).isoformat(),
})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5090, debug=False)