""" ThinkStorm Admin API Router Handles Prompts Management, Versioning, Profiles, Services Config, Job Observability, Moderation, and Token Accounting. """ import json import time from typing import Optional, List, Dict, Any from fastapi import APIRouter, Request, HTTPException, status, Depends from pydantic import BaseModel from ..config import config from ..database import get_db, get_utc_now from ..models import User, UserRole, ProcessorStatus from ..auth import require_admin from ..prompts.catalog import ( get_all_prompts, get_prompt_version, update_prompt, duplicate_prompt, get_all_profiles ) from ..services.omniroute import OmniRouteAdapter from ..services.searxng import SearXNGAdapter from ..services.perplexica import PerplexicaAdapter from ..services.opengist import OpenGistAdapter from ..services.gitea import GiteaAdapter from ..services.virustotal import VirusTotalAdapter from ..services.signal_gateway import SignalGatewayAdapter from ..queue.worker import job_queue router = APIRouter(prefix="/api/admin", tags=["admin"], dependencies=[Depends(require_admin)]) omniroute_svc = OmniRouteAdapter() searxng_svc = SearXNGAdapter() perplexica_svc = PerplexicaAdapter() opengist_svc = OpenGistAdapter() gitea_svc = GiteaAdapter() virustotal_svc = VirusTotalAdapter() signal_gateway_svc = SignalGatewayAdapter() class PromptUpdateRequest(BaseModel): name: Optional[str] = None description: Optional[str] = None system_prompt: str user_prompt_template: str model_policy: str = "reasoning" expected_outputs: Optional[List[str]] = None class PromptDuplicateRequest(BaseModel): new_id: str new_name: str class ServiceUpdateRequest(BaseModel): endpoint: str api_key_raw: Optional[str] = None enabled: bool = True config_json: Optional[Dict[str, Any]] = None class QuarantineDecisionRequest(BaseModel): decision: str # APPROVE or REJECT notes: Optional[str] = "" # ----------------- Prompts & Profiles ----------------- @router.get("/prompts") async def list_admin_prompts(): """Returns full prompt definitions with templates (Admin only).""" return get_all_prompts(is_admin=True) @router.get("/prompts/{prompt_id}") async def get_admin_prompt(prompt_id: str, version: Optional[int] = None): p = get_prompt_version(prompt_id, version, is_admin=True) if not p: raise HTTPException(status_code=404, detail="Prompt not found.") return p @router.put("/prompts/{prompt_id}") async def update_admin_prompt(prompt_id: str, payload: PromptUpdateRequest, current_user: User = Depends(require_admin)): """ Creates a new immutable prompt version. Historical execution records remain linked to their original version. """ new_version = update_prompt( prompt_id=prompt_id, system_prompt=payload.system_prompt, user_prompt_template=payload.user_prompt_template, model_policy=payload.model_policy, expected_outputs=payload.expected_outputs, updated_by=current_user.username, name=payload.name, description=payload.description ) # Log Audit Event with get_db() as conn: conn.execute( """ INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at) VALUES (?, 'UPDATE_PROMPT_VERSION', 'PROMPT', ?, ?, ?) """, (current_user.username, prompt_id, json.dumps({"new_version": new_version}), get_utc_now()) ) return {"message": f"Created new version {new_version} for prompt {prompt_id}.", "version": new_version} @router.post("/prompts/{prompt_id}/duplicate") async def duplicate_admin_prompt(prompt_id: str, payload: PromptDuplicateRequest, current_user: User = Depends(require_admin)): duplicate_prompt(prompt_id, payload.new_id, payload.new_name, created_by=current_user.username) return {"message": f"Prompt {prompt_id} duplicated to {payload.new_id}."} @router.post("/prompts/{prompt_id}/toggle") async def toggle_admin_prompt(prompt_id: str): with get_db() as conn: p = conn.execute("SELECT enabled FROM prompt_definitions WHERE id = ?", (prompt_id,)).fetchone() if not p: raise HTTPException(status_code=404, detail="Prompt not found.") new_state = 0 if p["enabled"] else 1 conn.execute("UPDATE prompt_definitions SET enabled = ? WHERE id = ?", (new_state, prompt_id)) return {"message": f"Prompt {prompt_id} enabled set to {bool(new_state)}."} @router.get("/profiles") async def list_admin_profiles(): return get_all_profiles() # ----------------- Services Configuration ----------------- @router.get("/services") async def list_services(): with get_db() as conn: rows = conn.execute("SELECT id, name, endpoint, api_key_masked, enabled, config_json, last_tested_at, health_status, last_error FROM service_configurations").fetchall() return [ { "id": r["id"], "name": r["name"], "endpoint": r["endpoint"], "api_key_masked": r["api_key_masked"], "enabled": bool(r["enabled"]), "config": json.loads(r["config_json"] or "{}"), "last_tested_at": r["last_tested_at"], "health_status": r["health_status"], "last_error": r["last_error"] } for r in rows ] @router.post("/services/{service_id}/test") async def test_service_connection(service_id: str): """Performs live health probe to external service.""" now = get_utc_now() adapter = None if service_id == "searxng": adapter = searxng_svc elif service_id == "omniroute": adapter = omniroute_svc elif service_id == "opengist": adapter = opengist_svc elif service_id == "gitea": adapter = gitea_svc elif service_id == "perplexica": adapter = perplexica_svc elif service_id == "virustotal": adapter = virustotal_svc elif service_id == "signal_gateway": adapter = signal_gateway_svc else: raise HTTPException(status_code=404, detail=f"Service '{service_id}' not found.") health = await adapter.check_health() status_str = "HEALTHY" if health.healthy else "UNHEALTHY" with get_db() as conn: conn.execute( """ UPDATE service_configurations SET last_tested_at = ?, health_status = ?, last_error = ? WHERE id = ? """, (now, status_str, None if health.healthy else health.message, service_id) ) return { "service_id": service_id, "healthy": health.healthy, "status": status_str, "message": health.message, "response_time_ms": health.response_time_ms } @router.put("/services/{service_id}") async def update_service(service_id: str, payload: ServiceUpdateRequest, current_user: User = Depends(require_admin)): now = get_utc_now() with get_db() as conn: s = conn.execute("SELECT * FROM service_configurations WHERE id = ?", (service_id,)).fetchone() if not s: raise HTTPException(status_code=404, detail="Service not found.") masked = s["api_key_masked"] raw = s["api_key_raw"] if payload.api_key_raw: raw = payload.api_key_raw masked = f"{raw[:4]}...{raw[-4:]}" if len(raw) > 8 else "****" conf_json = json.dumps(payload.config_json or json.loads(s["config_json"] or "{}")) conn.execute( """ UPDATE service_configurations SET endpoint = ?, api_key_masked = ?, api_key_raw = ?, enabled = ?, config_json = ? WHERE id = ? """, (payload.endpoint, masked, raw, 1 if payload.enabled else 0, conf_json, service_id) ) conn.execute( """ INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at) VALUES (?, 'UPDATE_SERVICE_CONFIG', 'SERVICE', ?, ?, ?) """, (current_user.username, service_id, json.dumps({"endpoint": payload.endpoint, "enabled": payload.enabled}), now) ) return {"message": f"Service '{service_id}' updated successfully."} # ----------------- Observability & Jobs ----------------- @router.get("/jobs") async def list_jobs(): queue_status = job_queue.get_status() with get_db() as conn: recent_runs = conn.execute( """ SELECT * FROM processor_runs ORDER BY started_at DESC LIMIT 30 """ ).fetchall() runs = [] for r in recent_runs: runs.append({ "id": r["id"], "idea_id": r["idea_id"], "work_track_id": r["work_track_id"], "processor_name": r["processor_name"], "stage": r["stage"], "prompt_id": r["prompt_id"], "prompt_version": r["prompt_version"], "resolved_model": r["resolved_model"], "total_tokens": r["total_tokens"], "duration_ms": r["duration_ms"], "started_at": r["started_at"], "status": r["status"], "error_message": r["error_message"] }) return { "queue": queue_status, "recent_runs": runs } @router.post("/jobs/retry/{idea_id}") async def retry_idea_job(idea_id: str): with get_db() as conn: conn.execute("UPDATE ideas SET processing_state = 'QUEUED' WHERE id = ?", (idea_id,)) await job_queue.enqueue_foreground("intake", idea_id, {"bypass_duplicate_check": True}) return {"message": f"Idea {idea_id} enqueued for processing retry."} # ----------------- Moderation & Quarantine ----------------- @router.get("/quarantine") async def list_quarantined_ideas(): with get_db() as conn: ideas = conn.execute( """ SELECT i.*, (SELECT COUNT(*) FROM idea_urls u WHERE u.idea_id = i.id AND u.safety_state = 'MALICIOUS') AS malicious_url_count, (SELECT COUNT(*) FROM idea_urls u WHERE u.idea_id = i.id AND u.safety_state = 'SUSPICIOUS') AS suspicious_url_count FROM ideas i WHERE i.lifecycle_state = 'QUARANTINED' ORDER BY i.submitted_at DESC """ ).fetchall() results = [] for r in ideas: urls = conn.execute("SELECT * FROM idea_urls WHERE idea_id = ?", (r["id"],)).fetchall() results.append({ "id": r["id"], "title": r["title"], "original_text": r["original_text"], "submitted_at": r["submitted_at"], "malicious_urls": r["malicious_url_count"], "suspicious_urls": r["suspicious_url_count"], "urls": [ { "url": u["url"], "safety_state": u["safety_state"], "automation_policy": u["automation_policy"], "virustotal": json.loads(u["virustotal_data"] or "{}") } for u in urls ] }) return results @router.post("/quarantine/{idea_id}/decision") async def quarantine_decision(idea_id: str, payload: QuarantineDecisionRequest, current_user: User = Depends(require_admin)): now = get_utc_now() with get_db() as conn: idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone() if not idea: raise HTTPException(status_code=404, detail="Idea not found.") if payload.decision.upper() == "APPROVE": conn.execute("UPDATE ideas SET lifecycle_state = 'AVAILABLE', updated_at = ? WHERE id = ?", (now, idea_id)) conn.execute("UPDATE idea_urls SET automation_policy = 'APPROVED', decision = 'APPROVED_BY_ADMIN', reviewed_by = ?, reviewed_at = ? WHERE idea_id = ?", (current_user.username, now, idea_id)) else: conn.execute("UPDATE ideas SET lifecycle_state = 'REJECTED', updated_at = ? WHERE id = ?", (now, idea_id)) conn.execute("UPDATE idea_urls SET automation_policy = 'BLOCKED', decision = 'REJECTED_BY_ADMIN', reviewed_by = ?, reviewed_at = ? WHERE idea_id = ?", (current_user.username, now, idea_id)) conn.execute( """ INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at) VALUES (?, 'QUARANTINE_DECISION', 'IDEA', ?, ?, ?) """, (current_user.username, idea_id, json.dumps({"decision": payload.decision, "notes": payload.notes}), now) ) return {"message": f"Idea {idea_id} marked as {payload.decision.upper()}."} # ----------------- Token Accounting & Metrics ----------------- @router.get("/token-metrics") async def get_token_metrics(): with get_db() as conn: totals = conn.execute("SELECT SUM(input_tokens) AS in_tok, SUM(output_tokens) AS out_tok, SUM(total_tokens) AS tot_tok, COUNT(*) AS run_count FROM processor_runs").fetchone() by_stage = conn.execute("SELECT stage, SUM(total_tokens) as tokens, COUNT(*) as count FROM processor_runs GROUP BY stage").fetchall() by_model = conn.execute("SELECT resolved_model, SUM(total_tokens) as tokens, COUNT(*) as count FROM processor_runs GROUP BY resolved_model").fetchall() by_idea = conn.execute("SELECT idea_id, SUM(total_tokens) as tokens, COUNT(*) as count FROM processor_runs GROUP BY idea_id ORDER BY tokens DESC LIMIT 10").fetchall() return { "overall": { "input_tokens": totals["in_tok"] or 0, "output_tokens": totals["out_tok"] or 0, "total_tokens": totals["tot_tok"] or 0, "runs_count": totals["run_count"] or 0 }, "by_stage": [dict(r) for r in by_stage], "by_model": [dict(r) for r in by_model], "top_ideas": [dict(r) for r in by_idea] } @router.get("/audit-logs") async def get_audit_logs(): with get_db() as conn: rows = conn.execute("SELECT * FROM audit_events ORDER BY created_at DESC LIMIT 50").fetchall() return [ { "id": r["id"], "user_id": r["user_id"], "action": r["action"], "entity_type": r["entity_type"], "entity_id": r["entity_id"], "details": json.loads(r["details"] or "{}"), "created_at": r["created_at"] } for r in rows ]