"""Public observability contract for the Pydantic AI bridge.
This module holds the parts of the audit feature that are *not* Pydantic AI
specific: the action event that enters ordinary EDRI routing, the environment
switch that arms the feature, and the application policy that narrows what its
payloads may contain.
Nothing here imports ``pydantic_ai``. The action event is what application
managers and the inspector consume, and that side of the system should not need
the optional dependency to understand it.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Callable, Final, Mapping, NamedTuple
from uuid import UUID
from edri.dataclass.event import Event, event
__all__ = [
"AGENT_AUDIT_ENV_VAR",
"AgentAction",
"AgentActionNamespace",
"AgentActionOperation",
"AgentActionOutcome",
"AgentActionPhase",
"AgentActionPolicy",
"AgentAuditMode",
"NormalizationLimits",
"Redactor",
"resolve_audit_mode",
]
#: A redactor receives an already detached, bounded, built-in snapshot -- never
#: a live business object -- and returns a replacement snapshot. It runs on the
#: lifecycle hot path, so it must be synchronous, in-memory and cheap.
Redactor = Callable[[Any], Any]
[docs]
class AgentActionPhase(Enum):
"""Lifecycle point of one execution attempt.
Closed, unlike ``namespace``/``operation``: START/END/ERROR has stable
transport semantics that a new subsystem does not extend.
"""
START = "start"
END = "end"
ERROR = "error"
[docs]
class AgentActionOutcome(Enum):
"""Terminal result of one execution attempt.
Cancellation, timeout and deferral are not failures merely because an
upstream library uses an exception to unwind control flow, so they
terminate with :attr:`AgentActionPhase.END` and their own outcome.
"""
SUCCESS = "success"
ERROR = "error"
CANCELLED = "cancelled"
TIMEOUT = "timeout"
DEFERRED = "deferred"
#: Outcomes that terminate an attempt without it having failed.
_END_OUTCOMES: Final = frozenset(
{
AgentActionOutcome.SUCCESS,
AgentActionOutcome.CANCELLED,
AgentActionOutcome.TIMEOUT,
AgentActionOutcome.DEFERRED,
}
)
[docs]
class AgentActionNamespace:
"""Broad subsystems this integration owns.
Plain strings, deliberately: ``AgentAction.namespace`` is extensible, and
adding a future subsystem such as ``memory`` must not require changing the
event model. These constants only spare callers a literal.
"""
AGENT: Final = "agent"
MODEL: Final = "model"
TOOL: Final = "tool"
MCP: Final = "mcp"
EDRI: Final = "edri"
[docs]
class AgentActionOperation:
"""Operations this integration owns, by namespace.
The MCP operations are reserved by the event model from day one; which of
them the implementation can actually observe on the supported dependency
floor is a separate, measured question (see the spec's MCP coverage matrix).
"""
RUN: Final = "run"
REQUEST: Final = "request"
VALIDATE: Final = "validate"
EXECUTE: Final = "execute"
DISPATCH: Final = "dispatch"
MCP_INITIALIZE: Final = "initialize"
MCP_TOOLS_LIST: Final = "tools.list"
MCP_TOOLS_CALL: Final = "tools.call"
MCP_RESOURCES_LIST: Final = "resources.list"
MCP_RESOURCES_READ: Final = "resources.read"
MCP_PROMPTS_LIST: Final = "prompts.list"
MCP_PROMPTS_GET: Final = "prompts.get"
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
[docs]
@event
class AgentAction(Event):
"""One observed lifecycle point of one agent operation.
Identity comes in three layers. ``Event._id`` identifies this concrete routed
event; ``(action_id, attempt)`` identifies the logical operation attempt, so
a START and its terminal event pair up while retries stay distinguishable;
``root_action_id`` names the run the action belongs to. The run's root action
is the key because upstream ``run_id`` is not unique -- a caller may pass one
``run_id`` to two concurrent invocations -- and because a consumer must be
able to bucket an action whose parent it never saw or has already evicted.
The event is deliberately not frozen -- ``Event`` carries mutable framework
transport state -- but its payloads are detached and normalized *before*
publication, and the emitter never mutates a published instance. The router
queue serializes on a feeder thread, so a payload still aliased to a live
object could otherwise change mid-serialization.
``request``/``response_data`` are typed ``Any`` because they hold normalized
JSON-like values (built-in scalars and containers only). ``response_data``
rather than ``response`` because ``Event.response`` is framework state.
"""
action_id: UUID
namespace: str
operation: str
phase: AgentActionPhase
timestamp: datetime = field(default_factory=_utc_now)
attempt: int = 1
parent_action_id: UUID | None = None
root_action_id: UUID | None = None
duration: float | None = None
outcome: AgentActionOutcome | None = None
run_id: str | None = None
conversation_id: str | None = None
edri_event_id: UUID | None = None
agent: str | None = None
provider: str | None = None
framework: str = "pydantic_ai"
model: str | None = None
model_version: str | None = None
prompt_version: str | None = None
output_mode: str | None = None
name: str | None = None
usage: dict[str, Any] | None = None
metadata: dict[str, Any] = field(default_factory=dict)
request: Any = None
response_data: Any = None
error: dict[str, Any] | None = None
def __post_init__(self) -> None:
super().__post_init__()
if self.attempt < 1:
raise ValueError(f"attempt is one-based, got {self.attempt}")
if self.phase is AgentActionPhase.START:
if self.outcome is not None:
raise ValueError("a START action has no outcome yet")
elif self.outcome is None:
raise ValueError(f"a {self.phase.value.upper()} action must carry an outcome")
elif self.phase is AgentActionPhase.ERROR:
if self.outcome is not AgentActionOutcome.ERROR:
raise ValueError(
"ERROR is reserved for genuine failures; "
f"{self.outcome.value} terminates with END"
)
elif self.outcome not in _END_OUTCOMES:
raise ValueError(f"{self.outcome.value} terminates with ERROR, not END")
[docs]
@dataclass(frozen=True, slots=True)
class NormalizationLimits:
"""Bounds applied to every captured payload, before and after redaction.
``max_total_bytes`` is one shared budget across an action's request,
response, metadata, usage and error fields rather than a per-field
allowance, so a single action cannot multiply the limit by its field count.
"""
max_depth: int = 8
max_values: int = 1_024
max_string_characters: int = 4_096
max_total_bytes: int = 64 * 1_024
def __post_init__(self) -> None:
for name in (
"max_depth", "max_values", "max_string_characters", "max_total_bytes",
):
value = getattr(self, name)
if value < 1:
raise ValueError(f"{name} must be positive, got {value}")
[docs]
@dataclass(frozen=True, slots=True)
class AgentActionPolicy:
"""What an application would like captured, per content category.
This is the *application* half of the content decision. The deployment half
is :data:`AGENT_AUDIT_ENV_VAR`, and the effective policy is their
intersection: a policy can always remove content, never add content the
operator disallowed. Everything therefore defaults to off, error messages
included.
"""
include_model_requests: bool = False
include_model_responses: bool = False
include_tool_arguments: bool = False
include_tool_results: bool = False
include_mcp_requests: bool = False
include_mcp_responses: bool = False
include_dispatch_requests: bool = False
include_dispatch_responses: bool = False
include_error_messages: bool = False
redactor: Redactor | None = None
limits: NormalizationLimits = field(default_factory=NormalizationLimits)
#: Names of every independent content switch, in policy declaration order.
CONTENT_SWITCHES: Final[tuple[str, ...]] = (
"include_model_requests",
"include_model_responses",
"include_tool_arguments",
"include_tool_results",
"include_mcp_requests",
"include_mcp_responses",
"include_dispatch_requests",
"include_dispatch_responses",
"include_error_messages",
)
AGENT_AUDIT_ENV_VAR: Final = "EDRI_PYDANTIC_AI_AUDIT"
[docs]
class AgentAuditMode(Enum):
"""Deployment-level ceiling on what observability may capture."""
OFF = "off"
METADATA = "metadata"
CONTENT = "content"
@property
def enabled(self) -> bool:
return self is not AgentAuditMode.OFF
[docs]
def resolve_audit_mode(environ: Mapping[str, str] | None = None) -> AgentAuditMode:
"""Read :data:`AGENT_AUDIT_ENV_VAR`; missing means :attr:`AgentAuditMode.OFF`.
Read here rather than as an import-time constant in ``edri.config.setting``
because the mode is resolved once per worker during
``BaseAgentWorker.setup()``. An import-time constant would fix the value for
the whole process and could not be resolved per worker at all.
Unknown values are configuration errors, not silently reinterpreted: an
operator who typed ``EDRI_PYDANTIC_AI_AUDIT=metdata`` asked for less
exposure than a fallback to ``content`` would give them, and more than a
fallback to ``off``.
"""
if environ is None:
import os
environ = os.environ
raw = environ.get(AGENT_AUDIT_ENV_VAR)
if raw is None:
return AgentAuditMode.OFF
try:
return AgentAuditMode(raw.strip().lower())
except ValueError:
permitted = ", ".join(mode.value for mode in AgentAuditMode)
raise ValueError(
f"{AGENT_AUDIT_ENV_VAR}={raw!r} is not a supported audit mode; "
f"use one of: {permitted}"
) from None
@dataclass(frozen=True, slots=True)
class _EffectivePolicy:
"""The resolved intersection of the environment mode and the app policy.
Private on purpose: :class:`AgentActionPolicy` describes intent and belongs
to the application, while this is runtime state the emitter reads. Keeping
them separate is what makes "the environment is a ceiling" a property of one
resolution step instead of a rule every call site must remember.
"""
mode: AgentAuditMode
include_model_requests: bool
include_model_responses: bool
include_tool_arguments: bool
include_tool_results: bool
include_mcp_requests: bool
include_mcp_responses: bool
include_dispatch_requests: bool
include_dispatch_responses: bool
include_error_messages: bool
redactor: Redactor | None
limits: NormalizationLimits
@property
def enabled(self) -> bool:
return self.mode.enabled
@property
def captures_content(self) -> bool:
return any(getattr(self, name) for name in CONTENT_SWITCHES)
def _resolve_effective_policy(
mode: AgentAuditMode, policy: AgentActionPolicy
) -> _EffectivePolicy:
"""Intersect the deployment ceiling with the application policy.
``metadata`` forces every content category off, error messages included;
``content`` lets each category follow the policy. ``off`` produces a policy
whose only correct use is to be skipped entirely.
"""
permitted = mode is AgentAuditMode.CONTENT
switches = {
name: permitted and getattr(policy, name) for name in CONTENT_SWITCHES
}
return _EffectivePolicy(
mode=mode,
redactor=policy.redactor,
limits=policy.limits,
**switches,
)
@dataclass(frozen=True, slots=True)
class _ObservabilityContext:
"""Framework-internal plumbing handed to one built agent during setup.
Deliberately not part of :class:`~edri.integrations.pydantic_ai.AgentDeps`:
``AgentDeps`` is application-facing and applications routinely override
``agent_deps()``, so audit state must not be reachable through -- or
dependent on -- ``ctx.deps.resources``.
``edri_event_id`` is bound at setup time because a ``BaseAgentWorker``
handles its constructor event. If agent workers ever multiplex several
triggering events, this binding has to become run-scoped.
The identity fields are bound here rather than read off the agent at each
lifecycle point because a Pydantic AI hook receives a ``RunContext``, not
the ``AgentBase`` that owns it -- there is no agent to ask. They are
defaults: a call site that actually *observed* one of them (a provider
reporting the checkpoint it served) passes it to ``emit`` and wins.
"""
policy: _EffectivePolicy
router_queue: Any = None
edri_event_id: UUID | None = None
agent: str | None = None
model: str | None = None
provider: str | None = None
output_mode: str | None = None
model_version: str | None = None
prompt_version: str | None = None
@property
def enabled(self) -> bool:
return self.policy.enabled
@dataclass(slots=True)
class _RunState:
"""Correlation state for one ``AgentBase._run()`` invocation.
One instance per invocation, closed over by that run's hooks and by its
dispatch wrapper. Nothing is stored on the agent and ``_deps`` is never
temporarily mutated, so two runs overlapping on one built agent -- which
they routinely do, since a worker can drive several, and tool bodies run on
other threads -- cannot read each other's identity.
``root_action_id`` is minted before the run starts and is this run's
identity for the inspector. Upstream run identity is not knowable until the
run has entered its outermost hook, so it is written once there and only
read afterwards; an action emitted before that carries ``run_id=None``
rather than a guess, and the inspector still keeps it apart from every other
run because the root action id is unique regardless.
"""
root_action_id: UUID
run_id: str | None = None
conversation_id: str | None = None
def _describe_missing_queue(worker_name: str) -> str:
return (
f"{worker_name} has observability enabled ({AGENT_AUDIT_ENV_VAR} is not "
f"'{AgentAuditMode.OFF.value}') but no router queue is bound. Actions are "
"published straight to the router, so there is nowhere to publish them. "
"A worker gets its queue from WorkerController when its manager was "
"built with one and the worker runs as a thread: check that the manager "
"was constructed with a router_queue, that the worker is not a process "
"worker, and that it was started through the manager rather than "
f"constructed directly. Otherwise set {AGENT_AUDIT_ENV_VAR}="
f"{AgentAuditMode.OFF.value}."
)
def _unbound_observability() -> _ObservabilityContext:
"""The context an agent carries before -- or without -- a worker binding.
Disabled rather than ``None`` so every hot-path check is the same one
expression, and so an agent constructed directly in a test or a script is
unobserved instead of raising on attribute access.
"""
return _ObservabilityContext(
policy=_resolve_effective_policy(AgentAuditMode.OFF, AgentActionPolicy())
)
def _emit_safely(emitter: _ActionEmitter, /, **action: Any) -> None:
"""Emit one action, absorbing any failure of the emitter itself.
:meth:`_ActionEmitter.emit` already contains the failures it can foresee --
snapshot, redaction, construction, publication. This is the guard around
the *call*: a wrong keyword from a call site, or a future emitter bug,
would otherwise raise inside a lifecycle hook or a dispatch wrapper and
change what the observed operation returns. Containment is a property of
the observation points, not only of the emitter.
"""
try:
emitter.emit(**action)
except Exception:
return
class _Capture(NamedTuple):
"""A content value plus whether the effective policy admits it.
Pairing the two at the call site is what keeps the policy decision next to
the value it governs: the emitter cannot capture something the caller did
not explicitly mark, and a caller cannot hand over content without saying
which switch it consulted.
"""
value: Any
allowed: bool
class _ActionEmitter:
"""Publishes one action per lifecycle point, best effort.
There is no local queue, overflow policy, drain thread, flush or teardown:
the ordinary EDRI router queue is the transport, and one ``put_nowait()``
is the whole publication step. That call means only that the local queue
accepted the object for asynchronous transfer -- not that the router, an
inspector or any durable consumer has seen it.
Every failure here is contained. Snapshot, redaction, action construction
and publication can each fail without changing what the observed operation
returns or raises. That is a control-flow guarantee, not a promise of
isolation from shared CPU, memory or router load.
"""
__slots__ = ("_context",)
def __init__(self, context: _ObservabilityContext) -> None:
self._context = context
@property
def enabled(self) -> bool:
return self._context.enabled
def emit(
self,
*,
action_id: UUID,
namespace: str,
operation: str,
phase: AgentActionPhase,
parent_action_id: UUID | None = None,
root_action_id: UUID | None = None,
attempt: int = 1,
outcome: AgentActionOutcome | None = None,
timestamp: datetime | None = None,
duration: float | None = None,
run_id: str | None = None,
conversation_id: str | None = None,
provider: str | None = None,
output_mode: str | None = None,
model_version: str | None = None,
prompt_version: str | None = None,
name: str | None = None,
usage: Any = None,
usage_counters: tuple[str, ...] | None = None,
metadata: Any = None,
request: _Capture | None = None,
response_data: _Capture | None = None,
error: BaseException | None = None,
) -> None:
"""Build and publish one action, or drop it.
Returns early on the disabled path before allocating, normalizing or
touching the queue -- ``off`` is the default, and an agent whose
operator did not ask for observability should not pay for it.
``provider``, ``output_mode``, ``model_version`` and ``prompt_version``
default to the identity bound at setup; pass one only where the value
was actually observed for this operation.
``usage_counters`` names the scope of ``usage``: the caller passes the
field list matching what it observed, because a per-request counter
object and a run total overlap in field names while measuring different
things.
"""
if not self._context.enabled:
return
context = self._context
policy = context.policy
try:
# Imported here, not at module scope: normalization reads this
# module's policy types, so a top-level import either way is a
# cycle. The action event and the policy stay importable without
# the normalizer, which is what lets a consumer read actions
# without depending on how they were produced.
from edri.integrations.pydantic_ai.normalization import (
TRUNCATED,
USAGE_FIELDS,
NormalizationBudget,
describe_error,
normalize,
usage_snapshot,
)
# One budget for the whole action: request, response, metadata,
# usage and error share it, so an action cannot multiply the limit
# by its field count.
budget = NormalizationBudget(policy.limits)
metadata_snapshot = (
normalize(metadata, budget) if metadata is not None else {}
)
if type(metadata_snapshot) is not dict:
# ``metadata`` is a mapping the inspector indexes. A shared
# budget can exhaust before this field is reached, and the
# normalizer then degrades the whole value to a marker string;
# publishing that would make a consumer's ``metadata.get(...)``
# fail on the audit record rather than on anything the agent
# did. The marker is kept, the declared shape is restored.
metadata_snapshot = {TRUNCATED: True}
action = AgentAction(
action_id=action_id,
namespace=namespace,
operation=operation,
phase=phase,
timestamp=timestamp if timestamp is not None else _utc_now(),
attempt=attempt,
parent_action_id=parent_action_id,
root_action_id=root_action_id,
duration=duration,
outcome=outcome,
run_id=run_id,
conversation_id=conversation_id,
edri_event_id=self._context.edri_event_id,
agent=self._context.agent,
provider=provider if provider is not None else context.provider,
model=self._context.model,
model_version=(
model_version
if model_version is not None
else context.model_version
),
prompt_version=(
prompt_version
if prompt_version is not None
else context.prompt_version
),
output_mode=(
output_mode if output_mode is not None else context.output_mode
),
name=name,
usage=(
usage_snapshot(
usage,
budget,
usage_counters if usage_counters is not None else USAGE_FIELDS,
)
if usage is not None
else None
),
metadata=metadata_snapshot,
request=self._capture(request, budget),
response_data=self._capture(response_data, budget),
error=(
describe_error(
error,
include_message=policy.include_error_messages,
budget=budget,
redactor=policy.redactor,
)
if error is not None
else None
),
)
self._context.router_queue.put_nowait(action)
except Exception:
# Deliberately silent, and deliberately not BaseException: a
# cancellation or a process-control signal is not an observability
# failure to consume. No logger call either -- logging handlers can
# block, and this runs inside an agent lifecycle hook.
return
def _capture(self, content: _Capture | None, budget: Any) -> Any:
"""Normalize and redact one content field, or omit it.
A field the policy does not admit is never snapshotted at all, so
disallowed content is not merely dropped late -- it is never copied.
"""
from edri.integrations.pydantic_ai.normalization import capture_content
if content is None or not content.allowed:
return None
return capture_content(
content.value, budget=budget, redactor=self._context.policy.redactor
)