73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
|
"""Agent soul profiles and Herman permissions."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Optional
|
||
|
|
|
||
|
|
from app.db import execute, fetch_all, fetch_one
|
||
|
|
|
||
|
|
|
||
|
|
def list_souls() -> list[dict[str, Any]]:
|
||
|
|
rows = fetch_all(
|
||
|
|
"""SELECT s.*,
|
||
|
|
(SELECT COUNT(*) FROM agent_events e WHERE LOWER(e.agent_name) = s.agent_key) AS event_count,
|
||
|
|
(SELECT MAX(created_at) FROM agent_events e WHERE LOWER(e.agent_name) = s.agent_key) AS last_event_at
|
||
|
|
FROM agent_souls s ORDER BY s.display_name"""
|
||
|
|
)
|
||
|
|
return [dict(r) for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def get_soul(agent_key: str) -> Optional[dict[str, Any]]:
|
||
|
|
row = fetch_one(
|
||
|
|
"""SELECT s.*,
|
||
|
|
(SELECT COUNT(*) FROM agent_events e WHERE LOWER(e.agent_name) = s.agent_key) AS event_count
|
||
|
|
FROM agent_souls s WHERE agent_key = %s""",
|
||
|
|
(agent_key.lower(),),
|
||
|
|
)
|
||
|
|
if not row:
|
||
|
|
return None
|
||
|
|
events = fetch_all(
|
||
|
|
"""SELECT id, event_type, title, status, created_at FROM agent_events
|
||
|
|
WHERE LOWER(agent_name) = %s ORDER BY created_at DESC LIMIT 15""",
|
||
|
|
(agent_key.lower(),),
|
||
|
|
)
|
||
|
|
out = dict(row)
|
||
|
|
out["recent_events"] = [dict(e) for e in events]
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def update_soul(agent_key: str, **fields: Any) -> dict[str, Any]:
|
||
|
|
allowed = ("display_name", "role_title", "soul_md", "responsibilities", "permissions", "is_active")
|
||
|
|
sets, params = [], []
|
||
|
|
for k, v in fields.items():
|
||
|
|
if k in allowed and v is not None:
|
||
|
|
sets.append(f"{k} = %s")
|
||
|
|
params.append(v)
|
||
|
|
if not sets:
|
||
|
|
soul = get_soul(agent_key)
|
||
|
|
if not soul:
|
||
|
|
raise ValueError("Agent not found")
|
||
|
|
return soul
|
||
|
|
params.append(agent_key.lower())
|
||
|
|
execute(f"UPDATE agent_souls SET {', '.join(sets)}, updated_at = NOW() WHERE agent_key = %s", tuple(params))
|
||
|
|
return get_soul(agent_key) or {}
|
||
|
|
|
||
|
|
|
||
|
|
def list_permissions() -> list[dict[str, Any]]:
|
||
|
|
return [dict(r) for r in fetch_all("SELECT * FROM herman_permissions ORDER BY category, module_label")]
|
||
|
|
|
||
|
|
|
||
|
|
def update_permission(module_key: str, granted: bool) -> dict[str, Any]:
|
||
|
|
execute(
|
||
|
|
"""UPDATE herman_permissions SET granted = %s, granted_at = CASE WHEN %s THEN NOW() ELSE NULL END, updated_at = NOW()
|
||
|
|
WHERE module_key = %s""",
|
||
|
|
(granted, granted, module_key),
|
||
|
|
)
|
||
|
|
row = fetch_one("SELECT * FROM herman_permissions WHERE module_key = %s", (module_key,))
|
||
|
|
return dict(row or {})
|
||
|
|
|
||
|
|
|
||
|
|
def grant_all_permissions() -> int:
|
||
|
|
execute("UPDATE herman_permissions SET granted = TRUE, granted_at = NOW(), updated_at = NOW()")
|
||
|
|
row = fetch_one("SELECT COUNT(*) AS n FROM herman_permissions WHERE granted = TRUE")
|
||
|
|
return int((row or {}).get("n") or 0)
|