av_harness/
content.rs

1//! Content export for external telemetry (the tenant OpenTelemetry
2//! collector).
3//!
4//! The audit worker hands every step it has journaled and published to an
5//! optional [`ContentSink`] as a [`ContentRecord`]: the OCSF event plus the
6//! step's content (the new input message, the model output, tool calls and
7//! results, token usage, finish reason). `agentvisord` installs a sink that
8//! turns records into OpenTelemetry GenAI spans for the tenant collector
9//! only; the operational collector never receives content. The library has
10//! no telemetry dependency: without an installed sink nothing is exported.
11
12use av_events::{AgentIdentity, EventClass};
13use serde_json::Value;
14use std::sync::Arc;
15
16/// One journaled step in content form.
17#[derive(Debug, Clone)]
18pub struct ContentRecord {
19    /// AgentVisor session id (the conversation id).
20    pub session_id: String,
21    /// OCSF event uid of the step.
22    pub event_uid: String,
23    /// Event class of the step.
24    pub class: EventClass,
25    /// True when the step's status is success.
26    pub success: bool,
27    /// Agent identity bound to the step.
28    pub identity: AgentIdentity,
29    /// When the step was journaled, epoch milliseconds.
30    pub time_ms: u64,
31    /// The OCSF event as journaled (already redacted when redaction is on).
32    pub event: Value,
33    /// Who produced the message: `user`, `agent`, or `system`.
34    pub source: Option<String>,
35    /// The step's message (input message for a request, output for a response).
36    pub message: Option<Value>,
37    /// Explicit reasoning text, when the provider returned it.
38    pub reasoning: Option<String>,
39    /// Model that produced a response.
40    pub model: Option<String>,
41    /// Tool calls requested by the model, as JSON.
42    pub tool_calls: Option<Value>,
43    /// Tool results or other environment feedback, as JSON.
44    pub observation: Option<Value>,
45    /// Prompt tokens attributed to the step.
46    pub prompt_tokens: Option<u64>,
47    /// Completion tokens attributed to the step.
48    pub completion_tokens: Option<u64>,
49    /// Provider-native finish reason.
50    pub finish_reason: Option<String>,
51    /// Chat response attempt shared by a request and its response.
52    pub attempt_id: Option<String>,
53    /// True for the terminal (response) record of a chat attempt.
54    pub terminal: bool,
55}
56
57/// Receives journaled steps. Called on a dedicated thread, one record at a
58/// time, in journal order per session.
59pub trait ContentSink: Send + Sync {
60    /// Accept one record.
61    fn record(&self, record: ContentRecord);
62}
63
64/// Records waiting for the sink thread. A full queue drops new records
65/// (counted in `av_content_records_dropped_total`) instead of slowing the
66/// audit worker.
67pub const CONTENT_QUEUE_CAPACITY: usize = 8192;
68
69/// Hands records from the audit worker to a [`ContentSink`] on a dedicated
70/// thread. The worker only does a non-blocking `try_send`; redaction,
71/// serialization, and export all run on the sink thread, and a panicking
72/// sink cannot reach the worker.
73pub struct ContentDispatcher {
74    sender: std::sync::mpsc::SyncSender<ContentRecord>,
75    dropped: Arc<av_core::metrics::Counter>,
76}
77
78impl ContentDispatcher {
79    /// Start the sink thread.
80    pub fn spawn(sink: Arc<dyn ContentSink>, metrics: &av_core::metrics::Registry) -> std::io::Result<Self> {
81        let (sender, receiver) = std::sync::mpsc::sync_channel::<ContentRecord>(CONTENT_QUEUE_CAPACITY);
82        let dropped = metrics.counter(
83            "av_content_records_dropped_total",
84            "Content records dropped because the tenant telemetry queue was full or the sink failed",
85        );
86        let failures = Arc::clone(&dropped);
87        std::thread::Builder::new()
88            .name("agentvisor-content".to_owned())
89            .spawn(move || {
90                for record in receiver {
91                    let sink = Arc::clone(&sink);
92                    let outcome =
93                        std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || sink.record(record)));
94                    if outcome.is_err() {
95                        failures.inc();
96                        tracing::error!("content sink panicked; record dropped");
97                    }
98                }
99            })?;
100        Ok(Self { sender, dropped })
101    }
102
103    /// Queue one record without blocking; drop it when the queue is full.
104    pub fn submit(&self, record: ContentRecord) {
105        if self.sender.try_send(record).is_err() {
106            self.dropped.inc();
107        }
108    }
109}
110
111/// A dispatcher slot shared by the application state and the audit worker,
112/// so a sink can be installed after both exist.
113pub type SharedContentSink = Arc<parking_lot::RwLock<Option<Arc<ContentDispatcher>>>>;
114
115/// Everything the audit worker applies to a step on its way out: the
116/// redaction engine (before journal and broker writes) and the content
117/// dispatcher (after the step is durable and published).
118#[derive(Clone, Default)]
119pub struct AuditOutputs {
120    /// Redaction applied to the journaled event and trajectory step.
121    pub redaction: Option<Arc<av_redact::RedactionEngine>>,
122    /// Optional content dispatcher.
123    pub content: SharedContentSink,
124}
125
126impl AuditOutputs {
127    /// Outputs with redaction only.
128    pub fn with_redaction(redaction: Option<Arc<av_redact::RedactionEngine>>) -> Self {
129        Self {
130            redaction,
131            content: SharedContentSink::default(),
132        }
133    }
134
135    /// The installed dispatcher, if any.
136    pub fn dispatcher(&self) -> Option<Arc<ContentDispatcher>> {
137        self.content.read().clone()
138    }
139}
av_harness/content.rs

av_harness/
content.rs
