sysops
141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
"""Agent handoffs and collaboration matrix."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from typing import Any, Optional
|
|
|
|
from app.db import execute, fetch_all, fetch_one
|
|
from app.services.agent_names import normalize_agent_key
|
|
|
|
|
|
def _serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if not row:
|
|
return None
|
|
out = dict(row)
|
|
for k, v in list(out.items()):
|
|
if hasattr(v, "isoformat"):
|
|
out[k] = v.isoformat()
|
|
elif k == "correlation_id" and v is not None:
|
|
out[k] = str(v)
|
|
return out
|
|
|
|
|
|
def list_collaboration() -> list[dict[str, Any]]:
|
|
rows = fetch_all(
|
|
"SELECT from_agent, to_agent, handoff_type, description FROM agent_collaboration ORDER BY from_agent, to_agent"
|
|
)
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def create_handoff(
|
|
from_agent: str,
|
|
to_agent: str,
|
|
*,
|
|
handoff_type: str = "partner",
|
|
payload: dict[str, Any] | None = None,
|
|
correlation_id: str | None = None,
|
|
status: str = "completed",
|
|
) -> dict[str, Any]:
|
|
src = normalize_agent_key(from_agent)
|
|
dst = normalize_agent_key(to_agent)
|
|
if not src or not dst:
|
|
raise ValueError("from_agent and to_agent required")
|
|
cid = correlation_id or str(uuid.uuid4())
|
|
try:
|
|
uuid.UUID(str(cid))
|
|
except ValueError:
|
|
cid = str(uuid.uuid4())
|
|
row = fetch_one(
|
|
"""
|
|
INSERT INTO agent_handoffs (correlation_id, from_agent, to_agent, handoff_type, payload, status, completed_at)
|
|
VALUES (%s::uuid, %s, %s, %s, %s::jsonb, %s, CASE WHEN %s = 'completed' THEN NOW() ELSE NULL END)
|
|
RETURNING *
|
|
""",
|
|
(cid, src, dst, handoff_type, json.dumps(payload or {}), status, status),
|
|
)
|
|
handoff = _serialize(row) or {}
|
|
meta = {
|
|
"correlation_id": cid,
|
|
"handoff_id": handoff.get("id"),
|
|
"target_agent": dst,
|
|
"source_agent": src,
|
|
"handoff_type": handoff_type,
|
|
}
|
|
_log_handoff_events(src, dst, handoff_type, payload or {}, meta, cid)
|
|
return handoff
|
|
|
|
|
|
def _log_handoff_events(
|
|
src: str,
|
|
dst: str,
|
|
handoff_type: str,
|
|
payload: dict[str, Any],
|
|
meta: dict[str, Any],
|
|
cid: str,
|
|
) -> None:
|
|
title_out = f"{src} → {dst}: {handoff_type}"
|
|
title_in = f"Handoff van {src}: {handoff_type}"
|
|
body = json.dumps(payload)[:2000] if payload else ""
|
|
for agent, etype, title, extra in (
|
|
(src, "handoff_out", title_out, {"target_agent": dst}),
|
|
(dst, "handoff_in", title_in, {"source_agent": src}),
|
|
):
|
|
try:
|
|
execute(
|
|
"""
|
|
INSERT INTO agent_events (agent_name, agent_type, event_type, title, body, status, channel, metadata)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb)
|
|
""",
|
|
(
|
|
agent,
|
|
"agent_handoff",
|
|
etype,
|
|
title[:255],
|
|
body,
|
|
"completed",
|
|
"agents",
|
|
json.dumps({**meta, **extra}),
|
|
),
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def recent_handoffs(hours: int = 6) -> list[dict[str, Any]]:
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT * FROM agent_handoffs
|
|
WHERE created_at >= NOW() - make_interval(hours => %s)
|
|
ORDER BY created_at DESC
|
|
LIMIT 200
|
|
""",
|
|
(max(1, min(hours, 168)),),
|
|
)
|
|
return [_serialize(r) for r in rows if r]
|
|
|
|
|
|
def peer_edges_live(hours: int = 6) -> list[dict[str, Any]]:
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT from_agent, to_agent, handoff_type, COUNT(*) AS weight,
|
|
MAX(correlation_id::text) AS correlation_id
|
|
FROM agent_handoffs
|
|
WHERE created_at >= NOW() - make_interval(hours => %s)
|
|
AND status = 'completed'
|
|
GROUP BY from_agent, to_agent, handoff_type
|
|
""",
|
|
(max(1, min(hours, 168)),),
|
|
)
|
|
return [
|
|
{
|
|
"source": str(r["from_agent"]),
|
|
"target": str(r["to_agent"]),
|
|
"type": "live",
|
|
"handoff_type": r.get("handoff_type"),
|
|
"weight": int(r["weight"] or 1),
|
|
"correlation_id": r.get("correlation_id"),
|
|
}
|
|
for r in rows
|
|
]
|