foodlinkk-command-center/cockpit/app/routes/agents_api.py

"""Agents API — souls & activity."""
from __future__ import annotations

import asyncio
import json
from typing import Any, Optional

from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from app.db import fetch_all, fetch_one
from app.services import agent_approvals, agent_integration, agent_souls
from app.services.agent_names import normalize_agent_key
from app.services.agent_terminal import execute_terminal_command

router = APIRouter(prefix="/api/agents", tags=["agents-api"])


class SoulUpdate(BaseModel):
    display_name: Optional[str] = None
    role_title: Optional[str] = None
    soul_md: Optional[str] = None
    responsibilities: Optional[str] = None
    is_active: Optional[bool] = None


class ActionRequestBody(BaseModel):
    agent_key: str
    title: str = Field(..., min_length=1, max_length=255)
    action_type: str = "query"
    query_payload: dict[str, Any] = Field(default_factory=dict)


class RejectBody(BaseModel):
    reason: str = ""


class ExecuteBody(BaseModel):
    result: dict[str, Any] = Field(default_factory=dict)


class HandoffBody(BaseModel):
    from_agent: str
    to_agent: str
    handoff_type: str = "partner"
    payload: dict[str, Any] = Field(default_factory=dict)
    correlation_id: Optional[str] = None


class TerminalCommandBody(BaseModel):
    command: str = Field(..., min_length=1, max_length=2000)


def _aggregate_agent_stats() -> dict[str, dict[str, Any]]:
    from datetime import datetime, timedelta, timezone

    rows = fetch_all(
        """
        SELECT agent_name, title, status, created_at
        FROM agent_events
        WHERE created_at >= NOW() - INTERVAL '7 days'
        ORDER BY created_at DESC
        """
    )
    now = datetime.now(timezone.utc)
    by_key: dict[str, dict[str, Any]] = {}
    for r in rows:
        key = normalize_agent_key(str(r.get("agent_name") or ""))
        if not key:
            continue
        bucket = by_key.setdefault(
            key,
            {"events_6h": 0, "errors_24h": 0, "last_event_at": None, "last_event_title": None},
        )
        created = r.get("created_at")
        if bucket["last_event_at"] is None and created is not None:
            bucket["last_event_at"] = created
            bucket["last_event_title"] = r.get("title")
        if created is not None:
            if hasattr(created, "tzinfo") and created.tzinfo is None:
                created = created.replace(tzinfo=timezone.utc)
            if created >= now - timedelta(hours=6):
                bucket["events_6h"] += 1
            if created >= now - timedelta(hours=24) and str(r.get("status") or "") in ("error", "rejected"):
                bucket["errors_24h"] += 1
    return by_key


def _node_health(events_6h: int, errors_24h: int, event_count: int) -> str:
    if events_6h > 0 and errors_24h == 0:
        return "healthy"
    if events_6h > 0:
        return "warn"
    if event_count > 0:
        return "idle"
    return "offline"


@router.get("/collaboration")
def api_collaboration() -> dict[str, Any]:
    items = agent_integration.list_collaboration()
    return {"items": items, "count": len(items)}


@router.post("/handoff")
def api_create_handoff(body: HandoffBody) -> dict[str, Any]:
    try:
        handoff = agent_integration.create_handoff(
            body.from_agent,
            body.to_agent,
            handoff_type=body.handoff_type,
            payload=body.payload,
            correlation_id=body.correlation_id,
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    return {"ok": True, "handoff": handoff}


def _terminal_payload(ev: dict[str, Any]) -> dict[str, Any]:
    if ev.get("created_at") and hasattr(ev["created_at"], "isoformat"):
        ev["created_at"] = ev["created_at"].isoformat()
    meta = ev.get("metadata")
    if isinstance(meta, str):
        try:
            meta = json.loads(meta)
        except Exception:
            meta = {}
    ev["metadata"] = meta or {}
    line_type = "handoff" if "handoff" in str(ev.get("event_type") or "") else "action"
    if str(ev.get("event_type") or "").startswith("terminal_"):
        line_type = "command" if ev.get("event_type") == "terminal_in" else "output"
    if ev["metadata"].get("target_agent"):
        line_type = "handoff_out"
    if ev["metadata"].get("source_agent"):
        line_type = "handoff_in"
    if str(ev.get("status") or "") in ("error", "rejected"):
        line_type = "error"
    return {
        "type": line_type,
        "id": ev.get("id"),
        "agent": normalize_agent_key(ev.get("agent_name")),
        "message": ev.get("title") or ev.get("event_type"),
        "detail": (ev.get("body") or "")[:500],
        "status": ev.get("status"),
        "correlation_id": ev["metadata"].get("correlation_id"),
        "at": ev.get("created_at"),
    }


@router.get("/souls/{agent_key}/terminal/stream")
async def api_agent_terminal_stream(agent_key: str, correlation_id: Optional[str] = None):
    key = normalize_agent_key(agent_key)
    soul = agent_souls.get_soul(key)
    if not soul:
        raise HTTPException(404, "Agent not found")

    async def event_gen():
        last_id = 0
        try:
            hist = fetch_all(
                """
                SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
                FROM agent_events
                WHERE LOWER(agent_name) = %s
                ORDER BY id DESC
                LIMIT 30
                """,
                (key,),
            )
            for row in reversed(hist):
                last_id = max(last_id, int(row["id"]))
                payload = _terminal_payload(dict(row))
                yield f"data: {json.dumps(payload, default=str)}\n\n"
        except Exception:
            pass

        while True:
            try:
                if correlation_id:
                    rows = fetch_all(
                        """
                        SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
                        FROM agent_events
                        WHERE metadata->>'correlation_id' = %s
                          AND id > %s
                        ORDER BY id ASC
                        LIMIT 50
                        """,
                        (correlation_id, last_id),
                    )
                else:
                    rows = fetch_all(
                        """
                        SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
                        FROM agent_events
                        WHERE LOWER(agent_name) = %s AND id > %s
                        ORDER BY id ASC
                        LIMIT 50
                        """,
                        (key, last_id),
                    )
                for row in rows:
                    last_id = max(last_id, int(row["id"]))
                    payload = _terminal_payload(dict(row))
                    yield f"data: {json.dumps(payload, default=str)}\n\n"
            except Exception as exc:
                err = {"type": "error", "message": str(exc)}
                yield f"data: {json.dumps(err)}\n\n"
            await asyncio.sleep(0.4)

    return StreamingResponse(
        event_gen(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@router.get("/live/stream")
async def api_agents_live_stream():
    """Single realtime stream for all agent terminals."""

    async def event_gen():
        last_id = 0
        try:
            recent = fetch_all(
                """
                SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
                FROM agent_events
                WHERE created_at >= NOW() - INTERVAL '30 minutes'
                ORDER BY id ASC
                LIMIT 80
                """
            )
            for row in recent:
                last_id = max(last_id, int(row["id"]))
                payload = _terminal_payload(dict(row))
                payload["replay"] = True
                yield f"data: {json.dumps(payload, default=str)}\n\n"
        except Exception:
            pass
        yield f"data: {json.dumps({'type': 'connected', 'last_id': last_id})}\n\n"

        while True:
            try:
                rows = fetch_all(
                    """
                    SELECT id, agent_name, event_type, title, body, status, channel, metadata, created_at
                    FROM agent_events
                    WHERE id > %s
                    ORDER BY id ASC
                    LIMIT 100
                    """,
                    (last_id,),
                )
                for row in rows:
                    last_id = max(last_id, int(row["id"]))
                    payload = _terminal_payload(dict(row))
                    yield f"data: {json.dumps(payload, default=str)}\n\n"
            except Exception as exc:
                err = {"type": "error", "message": str(exc)}
                yield f"data: {json.dumps(err)}\n\n"
            await asyncio.sleep(0.25)

    return StreamingResponse(
        event_gen(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@router.get("/souls")
def api_list_souls() -> dict[str, Any]:
    return {"items": agent_souls.list_souls(), "count": len(agent_souls.list_souls())}


@router.get("/souls/{agent_key}")
def api_get_soul(agent_key: str) -> dict[str, Any]:
    soul = agent_souls.get_soul(agent_key)
    if not soul:
        raise HTTPException(404, "Agent not found")
    return {"soul": soul}


@router.get("/souls/{agent_key}/events")
def api_list_agent_events(agent_key: str, limit: int = 50) -> dict[str, Any]:
    soul = agent_souls.get_soul(agent_key)
    if not soul:
        raise HTTPException(404, "Agent not found")
    items = agent_souls.list_agent_events(agent_key, limit=min(limit, 100))
    return {"agent_key": agent_key.lower(), "items": items, "count": len(items)}


@router.post("/souls/{agent_key}/terminal/command")
async def api_terminal_command(agent_key: str, body: TerminalCommandBody) -> dict[str, Any]:
    try:
        return await execute_terminal_command(agent_key, body.command.strip())
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@router.put("/souls/{agent_key}")
def api_update_soul(agent_key: str, body: SoulUpdate) -> dict[str, Any]:
    try:
        soul = agent_souls.update_soul(agent_key, **body.model_dump(exclude_none=True))
    except ValueError as exc:
        raise HTTPException(404, str(exc)) from exc
    return {"soul": soul}


@router.get("/mesh")
def api_agents_mesh() -> dict[str, Any]:
    souls = agent_souls.list_souls()
    by_key = _aggregate_agent_stats()

    nodes: list[dict[str, Any]] = []
    for soul in souls:
        key = str(soul.get("agent_key") or "").lower()
        row = by_key.get(key, {})
        events_6h = int(row.get("events_6h") or 0)
        errors_24h = int(row.get("errors_24h") or 0)
        health = _node_health(events_6h, errors_24h, int(soul.get("event_count") or 0))
        is_active = health == "healthy" and events_6h > 0
        node = dict(soul)
        node["health"] = health
        node["events_6h"] = events_6h
        node["errors_24h"] = errors_24h
        node["is_active"] = is_active
        node["last_event_title"] = row.get("last_event_title")
        lat = row.get("last_event_at")
        if lat is not None and hasattr(lat, "isoformat"):
            node["last_event_at"] = lat.isoformat()
        nodes.append(node)

    report_edges: list[dict[str, Any]] = []
    for node in nodes:
        key = str(node.get("agent_key") or "").lower()
        if key == "herman" or not node.get("is_active"):
            continue
        report_edges.append(
            {
                "source": key,
                "target": "herman",
                "type": "report",
                "active": True,
                "weight": int(node.get("events_6h") or 1),
            }
        )

    delegate_rows = fetch_all(
        """
        SELECT metadata, created_at
        FROM agent_events
        WHERE agent_name = 'herman'
          AND created_at >= NOW() - INTERVAL '6 hours'
          AND (
            event_type IN ('openswarm_delegation', 'delegate', 'packaging_delivered', 'telegram_delegation')
            OR metadata ? 'delegated'
          )
        ORDER BY created_at DESC
        LIMIT 100
        """
    )
    delegate_edges: list[dict[str, Any]] = []
    seen_delegate: set[tuple[str, str]] = set()
    for r in delegate_rows:
        meta = r.get("metadata") or {}
        if isinstance(meta, str):
            try:
                meta = json.loads(meta)
            except Exception:
                meta = {}
        delegated = meta.get("delegated") or meta.get("delegated_agents") or []
        if isinstance(delegated, str):
            delegated = [delegated]
        channel = meta.get("channel") or "herman"
        for agent in delegated:
            tgt = normalize_agent_key(str(agent))
            if not tgt or tgt == "herman":
                continue
            pair = ("herman", tgt)
            if pair in seen_delegate:
                continue
            seen_delegate.add(pair)
            delegate_edges.append(
                {
                    "source": "herman",
                    "target": tgt,
                    "type": "delegate",
                    "active": True,
                    "channel": channel,
                }
            )

    peer_static = [
        {
            "source": str(r["from_agent"]),
            "target": str(r["to_agent"]),
            "type": "static",
            "handoff_type": r.get("handoff_type"),
        }
        for r in agent_integration.list_collaboration()
    ]
    peer_live = agent_integration.peer_edges_live(hours=6)
    peer_edges = peer_static + peer_live

    herman_active = any(n.get("agent_key") == "herman" and n.get("is_active") for n in nodes)
    herman_active = herman_active or bool(delegate_edges) or bool(report_edges)
    executive_edges = []
    if herman_active:
        executive_edges = [
            {"source": "herman", "target": "ceo", "type": "executive", "active": True},
            {"source": "herman", "target": "cto", "type": "executive", "active": True},
        ]

    return {
        "nodes": nodes,
        "report_edges": report_edges,
        "delegate_edges": delegate_edges,
        "peer_edges": peer_edges,
        "executive_edges": executive_edges,
        "edges": report_edges,
        "executives": [
            {"id": "ceo", "label": "CEO", "role": "Aissa"},
            {"id": "cto", "label": "CTO", "role": "Platform"},
        ],
    }


@router.get("/approvals")
def api_list_approvals(status: Optional[str] = None, limit: int = 50) -> dict[str, Any]:
    try:
        items = agent_approvals.list_requests(status=status, limit=limit)
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    return {"items": items, "count": len(items)}


@router.post("/requests")
def api_create_action_request(body: ActionRequestBody) -> dict[str, Any]:
    try:
        req = agent_approvals.create_request(
            body.agent_key,
            body.title,
            action_type=body.action_type,
            query_payload=body.query_payload,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    auto = None
    rid = (req or {}).get("id")
    if rid:
        try:
            auto = agent_approvals.try_auto_approve_and_execute(int(rid))
        except Exception:
            auto = None
    msg = "Wacht op goedkeuring voordat query uitgevoerd mag worden"
    if auto:
        msg = "Auto-goedgekeurd en uitgevoerd (SysOps policy)"
    return {"ok": True, "request": req, "auto_executed": bool(auto), "message": msg}


@router.post("/approvals/{request_id}/approve")
def api_approve_request(request_id: int) -> dict[str, Any]:
    import os

    import httpx

    try:
        req = agent_approvals.approve_request(request_id, approved_by="ceo")
        action = req.get("action_type") or ""
        auto_result = None

        if action in ("config_backup", "maintenance_scan"):
            tools_url = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
            path = "/ops/backup/run" if action == "config_backup" else "/ops/maintenance/scan"
            payload = {"approval_request_id": request_id} if action == "config_backup" else None
            with httpx.Client(timeout=180.0) as client:
                resp = client.post(f"{tools_url}{path}", json=payload or {})
            try:
                auto_result = resp.json()
            except Exception:
                auto_result = {"ok": False, "detail": resp.text}
            req = agent_approvals.mark_executed(request_id, auto_result or {})

    except ValueError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    return {"ok": True, "request": req, "executed": auto_result is not None, "result": auto_result}


@router.post("/approvals/{request_id}/reject")
def api_reject_request(request_id: int, body: RejectBody) -> dict[str, Any]:
    try:
        req = agent_approvals.reject_request(request_id, reason=body.reason)
    except ValueError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return {"ok": True, "request": req}


@router.post("/approvals/{request_id}/execute")
def api_execute_approved_request(request_id: int, body: ExecuteBody) -> dict[str, Any]:
    import os

    import httpx

    try:
        req = agent_approvals.require_approved(request_id)
        result_payload = body.result or {}
        action = req.get("action_type") or ""

        if action == "config_backup":
            tools_url = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
            with httpx.Client(timeout=180.0) as client:
                resp = client.post(
                    f"{tools_url}/ops/backup/run",
                    json={"approval_request_id": request_id},
                )
            try:
                result_payload = resp.json()
            except Exception:
                result_payload = {"ok": False, "detail": resp.text}
            if resp.status_code >= 400:
                raise HTTPException(status_code=502, detail=result_payload.get("detail") or resp.text)

        elif action == "maintenance_scan":
            tools_url = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
            with httpx.Client(timeout=120.0) as client:
                resp = client.post(f"{tools_url}/ops/maintenance/scan")
            result_payload = resp.json() if resp.status_code < 400 else {"ok": False, "detail": resp.text}

        req = agent_approvals.mark_executed(request_id, result_payload)
    except HTTPException:
        raise
    except PermissionError as exc:
        raise HTTPException(status_code=403, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    return {"ok": True, "request": req}