114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
"""Runtime hooks for Agent Observability.
|
|
|
|
OpenClaw/Hermes/sub-agents can call these hooks at start/end/delegation points.
|
|
The hooks store sanitized metadata only: no prompts, no secrets, no private message
|
|
content.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from .agent_observability import AgentExecutionRecord
|
|
from .agent_observability_store import AgentObservabilityStore
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def parse_time(value: str) -> datetime:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
|
|
|
|
@dataclass
|
|
class AgentRunContext:
|
|
execution_id: str
|
|
user_id: str
|
|
visible_scope: str
|
|
entry_agent: str
|
|
path: list[str] = field(default_factory=list)
|
|
used_hermes: bool = False
|
|
started_at: str = field(default_factory=utc_now)
|
|
input_tokens: int = 0
|
|
output_tokens: int = 0
|
|
status: str = "running"
|
|
|
|
def delegate(self, agent_name: str) -> None:
|
|
if agent_name not in self.path:
|
|
self.path.append(agent_name)
|
|
if agent_name.lower() == "hermes":
|
|
self.used_hermes = True
|
|
|
|
def add_tokens(self, input_tokens: int = 0, output_tokens: int = 0) -> None:
|
|
self.input_tokens += max(input_tokens, 0)
|
|
self.output_tokens += max(output_tokens, 0)
|
|
|
|
|
|
class AgentObservabilityHooks:
|
|
def __init__(self, store: AgentObservabilityStore | None = None):
|
|
self.store = store or AgentObservabilityStore()
|
|
self.active: dict[str, AgentRunContext] = {}
|
|
|
|
def start(
|
|
self,
|
|
user_id: str,
|
|
visible_scope: str,
|
|
entry_agent: str,
|
|
execution_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
run = AgentRunContext(
|
|
execution_id=execution_id or f"exec_{uuid4().hex[:12]}",
|
|
user_id=user_id,
|
|
visible_scope=visible_scope,
|
|
entry_agent=entry_agent,
|
|
path=[entry_agent],
|
|
used_hermes=entry_agent.lower() == "hermes",
|
|
)
|
|
self.active[run.execution_id] = run
|
|
return {"status": "started", "execution_id": run.execution_id, "run": run.__dict__}
|
|
|
|
def delegate(self, execution_id: str, agent_name: str) -> dict[str, Any]:
|
|
run = self.active[execution_id]
|
|
run.delegate(agent_name)
|
|
return {"status": "delegated", "execution_id": execution_id, "path": run.path}
|
|
|
|
def tokens(self, execution_id: str, input_tokens: int = 0, output_tokens: int = 0) -> dict[str, Any]:
|
|
run = self.active[execution_id]
|
|
run.add_tokens(input_tokens, output_tokens)
|
|
return {
|
|
"status": "tokens_recorded",
|
|
"execution_id": execution_id,
|
|
"input_tokens": run.input_tokens,
|
|
"output_tokens": run.output_tokens,
|
|
"total_tokens": run.input_tokens + run.output_tokens,
|
|
}
|
|
|
|
def finish(self, execution_id: str, closed_by: str | None = None, status: str = "completed") -> dict[str, Any]:
|
|
run = self.active.pop(execution_id)
|
|
ended_at = utc_now()
|
|
duration_ms = int((parse_time(ended_at) - parse_time(run.started_at)).total_seconds() * 1000)
|
|
record = AgentExecutionRecord(
|
|
execution_id=run.execution_id,
|
|
user_id=run.user_id,
|
|
visible_scope=run.visible_scope,
|
|
entry_agent=run.entry_agent,
|
|
path=run.path,
|
|
used_hermes=run.used_hermes,
|
|
closed_by=closed_by or run.path[-1],
|
|
duration_ms=duration_ms,
|
|
input_tokens=run.input_tokens,
|
|
output_tokens=run.output_tokens,
|
|
total_tokens=run.input_tokens + run.output_tokens,
|
|
status=status,
|
|
started_at=run.started_at,
|
|
ended_at=ended_at,
|
|
)
|
|
return {"status": "stored", "record": self.store.append(record)}
|
|
|
|
|
|
__all__ = ["AgentRunContext", "AgentObservabilityHooks"]
|