Spaces:
Running
Running
File size: 8,608 Bytes
a31f556 9afc3bb a31f556 f6cfbb3 a940caa fe987b7 a940caa f6cfbb3 a940caa f6cfbb3 a940caa f6cfbb3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | from fastapi import HTTPException, APIRouter
router = APIRouter()
@router.get("/stats")
async def get_stats():
import psutil
return {
"cpu": psutil.cpu_percent(),
"ram": psutil.virtual_memory().percent,
"disk": psutil.disk_usage('/').percent,
"gpu": 0
}
@router.get("/info")
async def get_info():
# Read from the one place that owns it. This used to hardcode "1.0.0-OMEGA" while
# /health reported "1.1.0" from the same process — two answers to the same question.
from backend.version import VERSION, BUILD
return {"os": "windows", "version": VERSION, "build": BUILD}
@router.get("/pip_freeze")
async def pip_freeze():
"""Return the RUNNING environment's exact installed versions.
Dependency pinning has been deferred across sessions because the authoritative
pins must come from a `pip freeze` inside the real Linux container, not from the
Windows dev box. This route IS that mechanism: called against the deployed Space
it returns the container's real resolved versions, which then become
requirements.lock.txt (scripts/apply_pins.py consumes this output). It reads the
live environment via importlib.metadata — no shell-out needed inside the frozen
sidecar either."""
try:
from importlib.metadata import distributions
pkgs = sorted(
(f"{d.metadata['Name']}=={d.version}" for d in distributions()
if d.metadata and d.metadata.get('Name')),
key=str.lower)
import platform
return {"platform": platform.platform(), "python": platform.python_version(),
"count": len(pkgs), "freeze": pkgs}
except Exception as e:
raise HTTPException(status_code=500, detail=f"freeze failed: {e}")
@router.get("/health/mongo")
async def health_mongo():
"""Check MongoDB Atlas connectivity."""
try:
from backend.db.mongodb import MongoDBClient
await MongoDBClient.get_db().command("ping")
return {"status": "online", "service": "MongoDB Atlas"}
except Exception as e:
raise HTTPException(status_code=503, detail=f"MongoDB offline: {e}")
@router.get("/health/chroma")
async def health_chroma():
"""Check ChromaDB local vector store connectivity."""
try:
import chromadb
from chromadb.config import Settings
client = chromadb.PersistentClient(path="chroma_db", settings=Settings(anonymized_telemetry=False))
collections = client.list_collections()
return {"status": "online", "service": "ChromaDB", "collections": len(collections)}
except Exception as e:
raise HTTPException(status_code=503, detail=f"ChromaDB offline: {e}")
@router.get("/health/supervisor")
async def health_supervisor():
"""Check AI Self-Heal supervisor status."""
import os
supervisor_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "backend", "supervisor.py")
exists = os.path.exists(supervisor_path)
return {
"status": "online" if exists else "error",
"service": "AI Self-Heal Supervisor",
"supervisor_present": exists,
"max_restarts_per_minute": 5,
"heal_engine": "Gemini API"
}
from pydantic import BaseModel
class ActivationEventPayload(BaseModel):
persona: str
trigger_source: str
@router.post("/activation_event")
async def handle_activation_event(payload: ActivationEventPayload):
from backend.voice.activation import on_jarvis_friday_activated
import asyncio
asyncio.create_task(on_jarvis_friday_activated(payload.persona, payload.trigger_source))
return {"status": "ok", "message": "Activation sequence initiated"}
@router.get("/key_diagnostics")
async def key_diagnostics(probe: bool = False):
"""Report, per key domain, whether this runtime can actually resolve its key.
Answers "does every key work *here*" rather than "is the key string valid" —
those are different questions, because get_secret() prefers the environment
(HF Space Secrets) and only falls back to the encrypted vault. A key can be
perfectly valid at the provider and still be unreachable from the process
that needs it.
Never returns a key value: only presence, length and a fingerprint. With
?probe=true it also makes one real upstream call per distinct key so the
result reflects live provider state, not just local resolution.
"""
import hashlib, os, json, urllib.request, urllib.error, urllib.parse
from backend.services.usb_vault import KeyDomain, KEY_DOMAIN_ENV_MAP, get_secret
def fingerprint(v: str) -> str:
return hashlib.sha256(v.encode()).hexdigest()[:10]
def probe_google(key: str):
"""Do REAL work, not just an auth ping.
Listing models only proves the key authenticates. This runs an actual
generateContent call, which exercises the whole chain the domain depends
on: key resolution -> model availability -> request shape -> response
parsing. A key can list models fine and still fail to generate (quota
exhausted, model retired, safety block), which an auth ping hides.
"""
# Probe whatever the system is ACTUALLY configured to use, so this
# diagnostic can never drift from production behaviour.
try:
from backend.services.token_manager import MODEL as model
except Exception:
model = "gemini-3.1-flash-lite"
url = (f"https://generativelanguage.googleapis.com/v1beta/models/{model}"
f":generateContent?key=" + urllib.parse.quote(key))
body = json.dumps({
"contents": [{"parts": [{"text": "Reply with the single word: WORKING"}]}],
"generationConfig": {"maxOutputTokens": 2000},
}).encode()
req = urllib.request.Request(url, data=body,
headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=45) as r:
d = json.loads(r.read().decode())
cand = (d.get("candidates") or [{}])[0]
text = "".join(p.get("text", "") for p in
(cand.get("content", {}).get("parts") or []))
return {"ok": bool(text.strip()), "reply": text.strip()[:40],
"finish": cand.get("finishReason"), "model": model}
except urllib.error.HTTPError as e:
detail = ""
try:
detail = json.loads(e.read().decode())["error"]["message"][:80]
except Exception:
pass
return {"ok": False, "http": e.code, "why": detail}
except Exception as e:
return {"ok": False, "err": type(e).__name__}
domains, seen = [], {}
for domain, env_name in KEY_DOMAIN_ENV_MAP.items():
value = None
try:
value = get_secret(env_name)
except Exception:
value = None
entry = {
"domain": domain.value,
"env_var": env_name,
"resolved": bool(value),
"source": "env" if os.environ.get(env_name) else ("vault" if value else None),
"length": len(value) if value else 0,
"fingerprint": fingerprint(value) if value else None,
}
if probe and value:
fp = entry["fingerprint"]
if fp not in seen:
seen[fp] = probe_google(value)
entry["live"] = seen[fp]
domains.append(entry)
nvidia = []
for i in range(1, 16):
name = f"NVIDIA_API_KEY_{i}"
try:
v = get_secret(name)
except Exception:
v = None
nvidia.append({"env_var": name, "resolved": bool(v),
"source": "env" if os.environ.get(name) else ("vault" if v else None),
"fingerprint": fingerprint(v) if v else None})
others = {}
for name in ("GEMINI_API_KEY", "HF_API_TOKEN", "SKETCHFAB_TOKEN",
"BRAVE_SEARCH_API_KEY", "SERPAPI_KEY", "VAULT_MASTER_PASSWORD"):
try:
v = get_secret(name)
except Exception:
v = None
others[name] = {"resolved": bool(v),
"source": "env" if os.environ.get(name) else ("vault" if v else None)}
return {
"domains": domains,
"domains_resolved": sum(1 for d in domains if d["resolved"]),
"domains_total": len(domains),
"nvidia": nvidia,
"nvidia_resolved": sum(1 for n in nvidia if n["resolved"]),
"other": others,
}
|