"""The inspector manager: a standard EDRI manager that watches agent runs.
It consumes :class:`~edri.integrations.pydantic_ai.observability.AgentAction`
through ordinary routing and answers queries about what it has seen. An
application includes it like any other component; the environment switch
decides whether workers *emit* actions, this manager decides whether EDRI also
offers the standard browsing experience over them.
What it deliberately does not do is own anything on an agent's path. It holds
no agent workers, no Pydantic AI objects and no queue a run waits on, so an
inspector that is slow, wedged or entirely absent cannot slow a run down or
fail one. The worst it can do to an agent is forget what the agent did.
"""
# No ``from __future__ import annotations`` here, unlike its siblings: the
# manager resolver reads handler annotations at runtime through ``signature()``
# and would see the event parameter as a string. Nothing in this module needs
# deferred evaluation -- it imports no optional dependency, so every annotation
# below is resolvable as written.
from dataclasses import asdict
from datetime import datetime, timezone
from logging import Logger
from typing import Any, Callable
from edri.abstract import ManagerBase, request
from edri.dataclass.event import Event
from edri.dataclass.response import ResponseStatus
from edri.integrations.pydantic_ai.inspector_events import (
DEFAULT_PAGE_SIZE,
MAX_PAGE_SIZE,
AgentInspectorAction,
AgentInspectorRun,
AgentInspectorRuns,
AgentInspectorStats,
AgentInspectorUpdate,
)
from edri.integrations.pydantic_ai.inspector_store import (
ActionState,
AttemptState,
ChangeKind,
InspectorStore,
RunState,
StoreChange,
StoreLimits,
)
from edri.integrations.pydantic_ai.observability import AgentAction
__all__ = ["AgentInspectorManager"]
def _stamp(value: datetime | None) -> str | None:
return value.isoformat() if value is not None else None
def _attempt_view(attempt: AttemptState, *, content: bool) -> dict[str, Any]:
"""One attempt, projected.
``content`` is what separates a listing from a detail view: captured
prompts and results travel only where a client asked for that one action,
never in a page of every run the store holds.
"""
view: dict[str, Any] = {
"attempt": attempt.attempt,
"started_at": _stamp(attempt.started_at),
"ended_at": _stamp(attempt.ended_at),
"phase": attempt.phase.value if attempt.phase is not None else None,
"outcome": attempt.outcome.value if attempt.outcome is not None else None,
"duration": attempt.duration,
"usage": attempt.usage,
"start_missing": attempt.start_missing,
"terminal_missing": attempt.terminal_missing,
}
if content:
view["request"] = attempt.request
view["response_data"] = attempt.response_data
view["error"] = attempt.error
return view
def _action_view(action: ActionState, *, content: bool) -> dict[str, Any]:
return {
"action_id": str(action.action_id),
"root_action_id": str(action.root_action_id),
"parent_action_id": (
str(action.parent_action_id)
if action.parent_action_id is not None
else None
),
"namespace": action.namespace,
"operation": action.operation,
"name": action.name,
"agent": action.agent,
"model": action.model,
"provider": action.provider,
"metadata": action.metadata,
"truncated": action.truncated,
"start_missing": action.start_missing,
"terminal_missing": action.terminal_missing,
"attempts": [
_attempt_view(attempt, content=content)
for attempt in sorted(action.attempts.values(), key=lambda a: a.attempt)
],
}
def _run_view(run: RunState) -> dict[str, Any]:
"""A run without its actions: what a listing row needs and no more."""
return {
"root_action_id": str(run.root_action_id),
"generation_id": str(run.generation_id),
"revision": run.revision,
"run_id": run.run_id,
"conversation_id": run.conversation_id,
"agent": run.agent,
"created_at": _stamp(run.created_at),
"updated_at": _stamp(run.updated_at),
"completed": run.completed,
"outcome": run.outcome.value if run.outcome is not None else None,
"failed": run.failed,
"partial": run.partial,
"truncated": run.truncated,
"actions": len(run.actions),
}
def _run_detail(run: RunState) -> dict[str, Any]:
"""The run plus its action tree, each action's parent named in it."""
detail = _run_view(run)
detail["actions"] = [
_action_view(action, content=False) for action in run.actions.values()
]
return detail
[docs]
class AgentInspectorManager(ManagerBase):
"""Bounded, in-memory history of observed agent activity.
``limits`` is the application's retention choice; ``log_actions`` turns on
a concise human-readable line per action, off by default because an
application that wanted the query API did not necessarily ask for its log
to double in size.
``publish_updates`` is on because live inspection is what the manager is
for, and the broker sends an update only to clients whose subscription
filter matches it -- no browser is fanned the whole stream. What it does
cost is one router message per observed action. An application that wants
the query API and nothing else turns it off and pays nothing.
"""
def __init__(
self,
router_queue: Any | None = None,
logger: Logger | None = None,
from_time: datetime | None = None,
max_workers: int = -1,
*,
limits: StoreLimits | None = None,
log_actions: bool = False,
publish_updates: bool = True,
clock: Callable[[], datetime] | None = None,
) -> None:
super().__init__(
router_queue=router_queue, logger=logger, from_time=from_time, max_workers=max_workers
)
self.store = (
InspectorStore(limits=limits, clock=clock)
if clock is not None
else InspectorStore(limits=limits)
)
self._log_actions = log_actions
self._publish_updates = publish_updates
[docs]
def resolve(self, event: Event) -> None:
"""Resolve one event, then publish what the store noticed doing it.
Retention by age has no arrival to hang off: nothing reaches an idle
store, so a run that timed out is discovered by whichever read ran
next -- a query, or even a refused one. Draining here rather than in
each handler means no read can quietly swallow an invalidation a
subscriber is waiting for.
"""
super().resolve(event)
self._drain()
@request
def observe(self, event: AgentAction) -> None:
"""Take one observed action into the store.
The action itself is never sent on -- publishing the observed event
back would feed the stream into its own input. What does go back is a
subscribed update per state change, which is a different event.
"""
changes = self.store.ingest(event)
if self._log_actions:
self._summarize(event)
self._publish(changes, event)
# -- queries -----------------------------------------------------------
@request
def list_runs(self, event: AgentInspectorRuns) -> None:
"""A page of run summaries, newest first."""
limit = event.limit if event.limit is not None else DEFAULT_PAGE_SIZE
offset = event.offset if event.offset is not None else 0
if limit < 1 or offset < 0:
self._refuse(event, "limit must be positive and offset must not be negative")
return
# Clamped rather than refused: asking for too much is a reasonable
# request answered with what the store is willing to send.
limit = min(limit, MAX_PAGE_SIZE)
for bound in (event.since, event.until):
if bound is not None and bound.tzinfo is None:
self._refuse(event, "time bounds must carry a timezone")
return
runs, total = self.store.runs(
agent=event.agent,
run_id=event.run_id,
errors_only=bool(event.errors_only),
since=event.since,
until=event.until,
offset=offset,
limit=limit,
)
event.response.runs = [_run_view(run) for run in runs]
event.response.total = total
event.response.limit = limit
event.response.offset = offset
event.response.set_status(ResponseStatus.OK)
@request
def get_run(self, event: AgentInspectorRun) -> None:
"""One run and its action tree."""
if event.root_action_id is None and event.run_id is None:
self._refuse(event, "a run is addressed by root_action_id or run_id")
return
if event.root_action_id is not None:
run = self.store.run(event.root_action_id)
else:
matches, _ = self.store.runs(run_id=event.run_id)
if len(matches) > 1:
# One upstream id can belong to two concurrent invocations.
# Choosing between them would show an operator a run they did
# not ask for.
self._refuse(
event,
f"run_id {event.run_id!r} is ambiguous: "
f"{len(matches)} runs carry it",
)
return
run = matches[0] if matches else None
if run is None:
# Present and empty is a different answer from never seen or
# evicted, and a client draws them differently.
event.response.found = False
event.response.set_status(ResponseStatus.OK)
return
event.response.found = True
event.response.run = _run_detail(run)
event.response.generation_id = str(run.generation_id)
event.response.revision = run.revision
event.response.set_status(ResponseStatus.OK)
@request
def get_action(self, event: AgentInspectorAction) -> None:
"""One action in the detail the store still holds."""
if event.action_id is None:
self._refuse(event, "action_id is required")
return
found = self.store.action(event.action_id)
if found is None:
event.response.found = False
event.response.set_status(ResponseStatus.OK)
return
run, action = found
event.response.found = True
event.response.action = _action_view(action, content=True)
event.response.root_action_id = str(run.root_action_id)
event.response.generation_id = str(run.generation_id)
event.response.revision = run.revision
event.response.set_status(ResponseStatus.OK)
@request
def get_stats(self, event: AgentInspectorStats) -> None:
"""What the store holds and what retention has already removed."""
stats = self.store.stats()
event.response.runs = stats.runs
event.response.actions = stats.actions
event.response.attempts = stats.attempts
event.response.bytes = stats.bytes
event.response.dedupe_entries = stats.dedupe_entries
event.response.incomplete_runs = stats.incomplete_runs
event.response.evicted_runs = stats.evicted_runs
event.response.trimmed_runs = stats.trimmed_runs
event.response.limits = {
key: (value.total_seconds() if hasattr(value, "total_seconds") else value)
for key, value in asdict(stats.limits).items()
}
event.response.set_status(ResponseStatus.OK)
# -- live updates ------------------------------------------------------
def _publish(
self, changes: tuple[StoreChange, ...], action: AgentAction | None = None
) -> None:
"""Turn store changes into subscribed updates.
``action`` is the event that caused them, when one did. It supplies the
descriptive fields a client may filter on -- namespace, operation,
phase -- which belong to the action rather than to the run and so have
no meaning on an invalidation.
"""
if not self._publish_updates:
return
for change in changes:
self.send_to_router(self._update(change, action))
def _update(
self, change: StoreChange, action: AgentAction | None
) -> AgentInspectorUpdate:
# ``response`` is non-init on an @api event (the decorator gives it a
# default_factory), which mypy cannot see through.
update = AgentInspectorUpdate( # type: ignore[call-arg]
root_action_id=change.root_action_id,
generation_id=change.generation_id,
revision=change.revision,
kind=change.kind,
action_id=change.action_id,
run_id=change.run_id,
agent=change.agent,
)
update.response.kind = change.kind.value
if change.kind is ChangeKind.UPDATED:
if action is not None and action.action_id == change.action_id:
update.namespace = action.namespace
update.operation = action.operation
update.phase = action.phase
update.edri_event_id = action.edri_event_id
run = self.store.run(change.root_action_id)
if run is not None:
update.response.run = _run_view(run)
state = (
run.actions.get(change.action_id)
if change.action_id is not None
else None
)
if state is not None:
# Without content: an update is delivered to every client
# watching the run, and captured prompts and results are
# for the client that asked for that one action.
update.response.action = _action_view(state, content=False)
# A subscribed event reaches the broker only as a resolved response --
# the router forwards it to response subscribers, and an event without
# a status is not one.
update.response.set_status(ResponseStatus.OK)
return update
def _drain(self) -> None:
"""Publish what a query noticed.
Expiry has no other moment: nothing arrives on an idle store, so the
run that timed out is discovered by whichever read happened next. The
client holding it still has to be told.
"""
self._publish(self.store.drain_changes())
@staticmethod
def _refuse(event: Any, reason: str) -> None:
"""Answer a request the inspector will not guess at."""
event.response.error = reason
event.response.set_status(ResponseStatus.FAILED)
def _summarize(self, action: AgentAction) -> None:
"""Log one line about an action.
Identity and outcome only. The summary is a view of the stream, not a
second route by which prompts, results or error text reach a log that
the content policy never admitted them to.
"""
try:
self.logger.info(
"agent action %s.%s %s%s attempt=%s action=%s run=%s",
action.namespace,
action.operation,
action.phase.value,
f" {action.outcome.value}" if action.outcome is not None else "",
action.attempt,
action.action_id,
action.root_action_id,
)
except Exception:
# A broken log handler is not a reason to lose history: the action
# is already in the store by the time this runs.
pass