owner
1f4066b86c
- aios_gitops_model_api (adapters+app): drop X-AIOS-Role/X-AIOS-User trust; role/user resolved from server-side token->identity mapping in TOKEN_FILE. - task_action_api: ignore body role; role from injected authenticator (default user). - agent_observability_api: tenant/principal decides scope; client role=owner ignored.
94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
"""API for agent orchestration observability."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
try:
|
|
from fastapi import APIRouter
|
|
except Exception:
|
|
APIRouter = None
|
|
|
|
from .agent_observability import AgentExecutionRecord
|
|
from .agent_observability_dashboard import AgentObservabilityDashboard
|
|
from .agent_observability_store import AgentObservabilityStore
|
|
|
|
store = AgentObservabilityStore()
|
|
dashboard = AgentObservabilityDashboard(store.load())
|
|
|
|
_ROLE_WHITELIST = {"user", "group_admin", "owner"}
|
|
_authenticator = None # callable() -> {"user_id":..., "role":...} | None, injected by host
|
|
|
|
|
|
def set_authenticator(fn):
|
|
"""Install a server-side authenticator used to derive the querying tenant.
|
|
|
|
The callable returns the authenticated principal (dict with "user_id" and
|
|
a whitelisted "role") or None for anonymous. The client-supplied role and
|
|
user/group ids are NOT trusted to escalate scope.
|
|
"""
|
|
global _authenticator
|
|
_authenticator = fn
|
|
|
|
|
|
def _principal():
|
|
if _authenticator is not None:
|
|
p = _authenticator()
|
|
if isinstance(p, dict) and p.get("role") in _ROLE_WHITELIST:
|
|
return p
|
|
return None
|
|
|
|
|
|
def record_agent_execution(payload: dict[str, Any]):
|
|
record = AgentExecutionRecord(**payload)
|
|
return {"status": "ok", "record": store.append(record)}
|
|
|
|
|
|
def get_agent_observability(user_id: str | None = None, group_id: str | None = None, role: str = "user"):
|
|
# The authenticated principal (tenant) is authoritative; the request params
|
|
# role/user_id/group_id can never choose owner or another tenant.
|
|
principal = _principal()
|
|
pr_user = principal.get("user_id") if principal else None
|
|
pr_role = principal.get("role", "user") if principal else "user"
|
|
|
|
if pr_role == "owner":
|
|
# Server-verified owner may browse a requested scope.
|
|
q_user, q_group, q_role = user_id, group_id, "owner"
|
|
summary_scope = None
|
|
elif pr_user:
|
|
# Authenticated non-owner: strictly scoped to own tenant.
|
|
q_user, q_group, q_role = pr_user, None, "user"
|
|
summary_scope = pr_user
|
|
else:
|
|
# Anonymous: nothing scoped to them.
|
|
q_user, q_group, q_role = None, None, "user"
|
|
summary_scope = None
|
|
|
|
records = store.query(q_user, q_group, q_role)
|
|
return {
|
|
"status": "ok",
|
|
"records": records,
|
|
"summary": store.load().summary(summary_scope),
|
|
}
|
|
|
|
|
|
def build_router():
|
|
if APIRouter is None:
|
|
return None
|
|
router = APIRouter(prefix="/admin/ai-evolution/agents", tags=["agent-observability"])
|
|
|
|
@router.post("/execution", operation_id="recordAgentExecution")
|
|
def record(payload: dict[str, Any]):
|
|
return record_agent_execution(payload)
|
|
|
|
@router.get("/overview", operation_id="getAgentObservabilityOverview")
|
|
def overview(user_id: str | None = None, group_id: str | None = None, role: str = "user"):
|
|
return get_agent_observability(user_id, group_id, role)
|
|
|
|
return router
|
|
|
|
|
|
router = build_router()
|
|
|
|
__all__ = ["router", "record_agent_execution", "get_agent_observability"]
|