"""The inspector's query surface, as ordinary EDRI API events.
These are plain ``@api`` events: they reach the inspector through the same
broker, the same registration and the same middleware as every other event an
application exposes. There is no inspector-specific web server and no bypass
endpoint -- an application that authenticates its API authenticates these too.
Request fields stay scalar on purpose. The global ``@api`` validator admits
scalars, UUIDs, timestamps and lists of them, and that is exactly what a
filterable query needs; the nested trees the inspector returns live in the
*response*, which the serializer already carries.
"""
from datetime import datetime
from http import HTTPMethod
from typing import Any
from uuid import UUID
from dataclasses import field
from edri.api.dataclass.api_event import api
from edri.config.constant import ApiType
from edri.dataclass.event import Event, EventHandlingType
from edri.dataclass.response import Response, response
from edri.integrations.pydantic_ai.inspector_store import ChangeKind
from edri.integrations.pydantic_ai.observability import AgentActionPhase
__all__ = [
"AgentInspectorAction",
"AgentInspectorActionResponse",
"AgentInspectorRun",
"AgentInspectorRunResponse",
"AgentInspectorRuns",
"AgentInspectorRunsResponse",
"AgentInspectorStats",
"AgentInspectorStatsResponse",
"AgentInspectorUpdate",
"AgentInspectorUpdateResponse",
]
#: What a client gets when it asks for no page size, and the most it can ask
#: for. The cap is the point: a browser tab must not be able to request the
#: whole store in one message.
DEFAULT_PAGE_SIZE = 50
MAX_PAGE_SIZE = 200
[docs]
@response
class AgentInspectorRunsResponse(Response):
"""A page of run summaries, newest first."""
runs: list[dict[str, Any]] = field(default_factory=list)
#: Matching runs, not returned runs: a client paging needs to know what it
#: has not seen.
total: int = 0
limit: int = DEFAULT_PAGE_SIZE
offset: int = 0
error: str | None = None
[docs]
@api(url="/edri/agents", resource="agent-inspector-runs", template="agent_inspector.j2")
class AgentInspectorRuns(Event):
"""Recent runs, filtered and paged.
A browser asking for HTML gets the inspector dashboard instead of the
listing: one page that opens a WebSocket, subscribes to
:class:`AgentInspectorUpdate` and then answers itself out of the four query
events. It is the same event either way, so an application that authorizes
the listing authorizes the page, and one that never routes this event has
no page to reach.
"""
method: HTTPMethod = HTTPMethod.GET
agent: str | None = None
#: The upstream identifier. It may match several runs -- a caller can hand
#: one ``run_id`` to two invocations -- so it filters rather than selects.
run_id: str | None = None
errors_only: bool | None = None
since: datetime | None = None
until: datetime | None = None
limit: int | None = None
offset: int | None = None
response: AgentInspectorRunsResponse
[docs]
@response
class AgentInspectorRunResponse(Response):
"""One reconstructed run with its action tree."""
#: ``False`` means the store does not have this run -- never seen, or
#: evicted. It is not the same answer as a run that is present and empty.
found: bool = False
run: dict[str, Any] | None = None
generation_id: str | None = None
revision: int | None = None
error: str | None = None
[docs]
@api(url="/edri/agents/run", resource="agent-inspector-run")
class AgentInspectorRun(Event):
"""One run, addressed by the root action that identifies it."""
method: HTTPMethod = HTTPMethod.GET
root_action_id: UUID | None = None
#: An alternative for a caller that only knows the upstream id. Ambiguous
#: by nature, and answered as ambiguous rather than by picking one.
run_id: str | None = None
response: AgentInspectorRunResponse
[docs]
@response
class AgentInspectorActionResponse(Response):
"""One action in full retained detail."""
found: bool = False
action: dict[str, Any] | None = None
root_action_id: str | None = None
generation_id: str | None = None
revision: int | None = None
error: str | None = None
[docs]
@api(url="/edri/agents/action", resource="agent-inspector-action")
class AgentInspectorAction(Event):
"""One action, with the attempts and captured content the store kept."""
method: HTTPMethod = HTTPMethod.GET
action_id: UUID | None = None
response: AgentInspectorActionResponse
[docs]
@response
class AgentInspectorStatsResponse(Response):
"""What the store holds, and what it has already dropped."""
runs: int = 0
actions: int = 0
attempts: int = 0
bytes: int = 0
dedupe_entries: int = 0
incomplete_runs: int = 0
evicted_runs: int = 0
trimmed_runs: int = 0
limits: dict[str, Any] = field(default_factory=dict)
[docs]
@api(url="/edri/agents/stats", resource="agent-inspector-stats")
class AgentInspectorStats(Event):
"""Retention and store statistics."""
method: HTTPMethod = HTTPMethod.GET
response: AgentInspectorStatsResponse
[docs]
@response
class AgentInspectorUpdateResponse(Response):
"""What changed, in the detail a live view needs to redraw.
The rich body lives here rather than on the event because ``@api`` request
fields are restricted to API-safe scalars. That restriction is the reason
the filterable keys above are flat: a subscription predicate can only ever
compare a top-level field, so anything a client filters on has to be one.
"""
#: ``updated`` replaces; ``invalidated`` says the run is gone from the
#: store and a client holding it should stop trusting what it has.
kind: str = ChangeKind.UPDATED.value
#: The run summary as the listing would render it, or ``None`` for an
#: invalidation -- there is no longer a run to summarize.
run: dict[str, Any] | None = None
#: The one action this change touched, without captured content. A live
#: view redraws a node from it; the detail query is where content lives.
action: dict[str, Any] | None = None
[docs]
@api(url="/edri/agents/update", resource="agent-inspector-update",
handling=EventHandlingType.SUBSCRIBED, exclude={ApiType.REST, ApiType.HTML})
class AgentInspectorUpdate(Event):
"""One store change, delivered to whoever registered for it.
Every field here exists to be filtered on. The broker compares a
subscription's ``param_set`` against top-level event fields, so the keys a
client is expected to narrow by -- the run above all -- are carried flat
and typed, and the change itself travels in the response.
Only ``root_action_id``, ``generation_id``, ``revision`` and ``kind`` are
filled on every update. The rest describe the action that caused the
change and are absent on an invalidation, which no action caused. A client
that filters on one of them therefore receives a view of the run rather
than its complete history -- which is a legitimate feed, but not the
stream the snapshot handshake detects loss on.
"""
#: The run this change belongs to: the one filter a browsing client is
#: always expected to set.
root_action_id: UUID | None = None
#: A revision is only comparable within its generation. A run evicted and
#: later rebuilt starts counting again under a new ``generation_id``.
generation_id: UUID | None = None
revision: int | None = None
kind: ChangeKind | None = None
#: The action the change is about, absent on an invalidation.
action_id: UUID | None = None
run_id: str | None = None
agent: str | None = None
namespace: str | None = None
operation: str | None = None
phase: AgentActionPhase | None = None
#: The business event that started the run, when one did: an application
#: watching what its own request caused filters on this.
edri_event_id: UUID | None = None
response: AgentInspectorUpdateResponse