feat: agent notes, OpenRouter onboarding, usage tracking, per-role tools docs
Agent notes tool (cortex/tools/agent_notes.py): - Private durable notepad for the orchestrator — not user-visible - agent_notes_read/write/append/clear with 3 rolling backups - Per-persona isolation via ContextVars; no TOOL_ROLES gating needed - PROTOCOLS.md updated to make this a core proactive tool OpenRouter guided onboarding: - Setup Step 3 (/setup/model) — OpenRouter quick-connect with curated model list - Amber banner in chat for users on server-default model - Settings quick-link card (/settings/models OpenRouter section) - POST /setup/model/skip for users who want to bypass Step 3 - Holly pre-configured: DeepSeek V4 Flash (OpenRouter) → Gemma Medium (local) → claude_cli Usage tracking: - cortex/routers/usage.py — GET /api/usage, /api/usage/summary, /api/usage/all (admin) Documentation: - HELP.md: Tools section rewritten — full tool table by category, per-role tool sets explained - TOOLS.md: Agent Notes section added; count corrected to 44 - ARCH__SYSTEM.md, ARCH__BACKENDS.md, MASTER.md, CLAUDE.md, README.md updated - TODO__Agents.md: onboarding task checked off with deviation notes Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
104
cortex/routers/usage.py
Normal file
104
cortex/routers/usage.py
Normal file
@@ -0,0 +1,104 @@
|
||||
"""
|
||||
Usage / token-tracking endpoints.
|
||||
|
||||
Self-service (any authenticated user, own data):
|
||||
GET /api/usage → full usage dict {date: {model_key: {calls, prompt_tokens, completion_tokens}}}
|
||||
GET /api/usage/summary → aggregate totals per model key, with friendly labels resolved from registry
|
||||
|
||||
Admin-only (cross-user aggregation):
|
||||
GET /api/usage/all → summary for every user {username: summary_dict}
|
||||
"""
|
||||
import jwt
|
||||
from fastapi import APIRouter, HTTPException, Request
|
||||
|
||||
from auth_utils import COOKIE_NAME, decode_token, get_user_role
|
||||
from persona import list_users
|
||||
import model_registry
|
||||
import usage_tracker
|
||||
|
||||
router = APIRouter(prefix="/api/usage")
|
||||
|
||||
|
||||
def _session_user(request: Request) -> str:
|
||||
token = request.cookies.get(COOKIE_NAME)
|
||||
if not token:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
try:
|
||||
return decode_token(token)
|
||||
except jwt.InvalidTokenError:
|
||||
raise HTTPException(status_code=401, detail="Invalid session")
|
||||
|
||||
|
||||
def _build_label_map(username: str) -> dict[str, str]:
|
||||
"""Build a map from usage key (backend/model_name) → registered label."""
|
||||
label_map: dict[str, str] = {}
|
||||
try:
|
||||
for m in model_registry.get_all_models(username):
|
||||
model_name = m.get("model_name", "")
|
||||
label = m.get("label", "")
|
||||
host_type = m.get("host_type", "")
|
||||
if not model_name or not label:
|
||||
continue
|
||||
# local models: key is "local/{model_name}"
|
||||
if host_type in ("openwebui", "ollama", "openai_compatible"):
|
||||
label_map[f"local/{model_name}"] = label
|
||||
# cloud Gemini: key is "gemini_api/{model_name}"
|
||||
elif host_type == "google":
|
||||
label_map[f"gemini_api/{model_name}"] = label
|
||||
except Exception:
|
||||
pass
|
||||
return label_map
|
||||
|
||||
|
||||
def _summarize(data: dict, label_map: dict[str, str] | None = None) -> list[dict]:
|
||||
"""Collapse date-keyed usage dict into per-model totals, sorted by total tokens desc."""
|
||||
totals: dict[str, dict] = {}
|
||||
for _date, models in data.items():
|
||||
for key, counts in models.items():
|
||||
t = totals.setdefault(key, {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0})
|
||||
t["calls"] += counts.get("calls", 0)
|
||||
t["prompt_tokens"] += counts.get("prompt_tokens", 0)
|
||||
t["completion_tokens"] += counts.get("completion_tokens", 0)
|
||||
|
||||
result = []
|
||||
for key, counts in totals.items():
|
||||
entry = {
|
||||
"key": key,
|
||||
"label": (label_map or {}).get(key) or key,
|
||||
"calls": counts["calls"],
|
||||
"prompt_tokens": counts["prompt_tokens"],
|
||||
"completion_tokens": counts["completion_tokens"],
|
||||
"total_tokens": counts["prompt_tokens"] + counts["completion_tokens"],
|
||||
}
|
||||
result.append(entry)
|
||||
|
||||
result.sort(key=lambda x: x["total_tokens"], reverse=True)
|
||||
return result
|
||||
|
||||
|
||||
@router.get("")
|
||||
async def get_usage(request: Request) -> dict:
|
||||
"""Return the raw daily usage log for the authenticated user."""
|
||||
username = _session_user(request)
|
||||
return usage_tracker.read_usage(username)
|
||||
|
||||
|
||||
@router.get("/summary")
|
||||
async def get_usage_summary(request: Request) -> list:
|
||||
"""Return per-model totals (all time) for the authenticated user, with friendly labels."""
|
||||
username = _session_user(request)
|
||||
label_map = _build_label_map(username)
|
||||
return _summarize(usage_tracker.read_usage(username), label_map)
|
||||
|
||||
|
||||
@router.get("/all")
|
||||
async def get_all_usage(request: Request) -> dict:
|
||||
"""Admin: return per-model summary for every user."""
|
||||
username = _session_user(request)
|
||||
if get_user_role(username) != "admin":
|
||||
raise HTTPException(status_code=403, detail="Admin access required")
|
||||
result = {}
|
||||
for user in list_users():
|
||||
label_map = _build_label_map(user)
|
||||
result[user] = _summarize(usage_tracker.read_usage(user), label_map)
|
||||
return result
|
||||
Reference in New Issue
Block a user