av_harness/
otel_content.rs

1//! Full-content OpenTelemetry export to the tenant collector.
2//!
3//! [`TenantContentSink`] turns every journaled step into an OpenTelemetry
4//! span that follows the GenAI semantic conventions
5//! (<https://github.com/open-telemetry/semantic-conventions-genai>):
6//!
7//! - a chat call (request record plus terminal response record, joined by
8//!   their response-attempt id) becomes one `chat {model}` span of kind
9//!   `CLIENT` with `gen_ai.input.messages`, `gen_ai.output.messages`, token
10//!   usage, finish reasons, and `gen_ai.conversation.id` (the session);
11//! - a tool authorization becomes an `execute_tool {tool}` span with the
12//!   tool name, call id, and arguments, plus the gateway's verdict;
13//! - a tool completion becomes an `execute_tool.result` span carrying
14//!   `gen_ai.tool.call.result`;
15//! - every other audit event becomes an `agentvisor.{class}` span carrying
16//!   the redacted event.
17//!
18//! Every content field passes through the redaction engine (the built-in
19//! patterns plus the configured `redaction_patterns` and `redaction_paths`)
20//! before it becomes an attribute, even when journal redaction is off. The
21//! sink exports through its own tracer provider, whose only exporter is the
22//! tenant endpoint, so the operator's collector never receives content.
23
24use crate::content::{ContentRecord, ContentSink};
25use av_events::EventClass;
26use opentelemetry::trace::{Span as _, SpanKind, TracerProvider as _};
27use opentelemetry::{Array, KeyValue, StringValue, Value as OtelValue};
28use serde_json::{json, Value};
29use std::collections::{HashMap, VecDeque};
30use std::time::{Duration, SystemTime, UNIX_EPOCH};
31
32/// Chat requests waiting for their terminal response record.
33const MAX_PENDING_CHATS: usize = 4096;
34
35/// Longest content attribute exported; longer values are truncated with a
36/// marker that states how many bytes were cut.
37pub const MAX_ATTRIBUTE_BYTES: usize = 256 * 1024;
38
39/// Content sink exporting GenAI spans to the tenant collector.
40pub struct TenantContentSink {
41    tracer: opentelemetry_sdk::trace::SdkTracer,
42    redaction: av_redact::RedactionEngine,
43    provider_name: String,
44    pending: parking_lot::Mutex<(HashMap<String, ContentRecord>, VecDeque<String>)>,
45}
46
47impl TenantContentSink {
48    /// Build a sink on `provider` (which must export only to the tenant),
49    /// redacting with `redaction`. `provider_name` is the configured
50    /// upstream provider dialect (`gen_ai.provider.name`).
51    pub fn new(
52        provider: &opentelemetry_sdk::trace::SdkTracerProvider,
53        redaction: av_redact::RedactionEngine,
54        provider_name: &str,
55    ) -> Self {
56        Self {
57            tracer: provider.tracer("agentvisor-ai.content"),
58            redaction,
59            provider_name: provider_name.to_owned(),
60            pending: parking_lot::Mutex::new((HashMap::new(), VecDeque::new())),
61        }
62    }
63
64    fn redact(&self, record: &mut ContentRecord) {
65        let scrub = |value: &mut Option<Value>| {
66            if let Some(value) = value.as_mut() {
67                self.redaction.redact_value(value);
68            }
69        };
70        scrub(&mut record.message);
71        scrub(&mut record.tool_calls);
72        scrub(&mut record.observation);
73        self.redaction.redact_value(&mut record.event);
74        if let Some(reasoning) = record.reasoning.take() {
75            let mut value = Value::String(reasoning);
76            self.redaction.redact_value(&mut value);
77            record.reasoning = value.as_str().map(str::to_owned);
78        }
79    }
80
81    fn emit(&self, name: String, kind: SpanKind, start_ms: u64, end_ms: u64, attributes: Vec<KeyValue>) {
82        let at = |ms: u64| UNIX_EPOCH + Duration::from_millis(ms);
83        let builder = opentelemetry::trace::SpanBuilder::from_name(name)
84            .with_kind(kind)
85            .with_start_time(at(start_ms.min(end_ms)))
86            .with_attributes(attributes);
87        let mut span = builder.start_with_context(&self.tracer, &opentelemetry::Context::new());
88        span.end_with_timestamp(at(end_ms));
89    }
90
91    fn common(&self, record: &ContentRecord) -> Vec<KeyValue> {
92        let mut attributes = vec![
93            KeyValue::new("gen_ai.conversation.id", record.session_id.clone()),
94            KeyValue::new("gen_ai.agent.id", record.identity.instance_uid.clone()),
95            KeyValue::new("gen_ai.agent.name", record.identity.charter.name.clone()),
96            KeyValue::new("agentvisor.agent.version", record.identity.version.clone()),
97            KeyValue::new("agentvisor.event.uid", record.event_uid.clone()),
98            KeyValue::new("agentvisor.event.class", class_label(record.class)),
99            KeyValue::new("agentvisor.success", record.success),
100        ];
101        if !record.success {
102            attributes.push(KeyValue::new("error.type", "agentvisor.failure"));
103        }
104        attributes
105    }
106
107    fn chat(&self, request: Option<ContentRecord>, response: ContentRecord) {
108        let model = response.model.clone().unwrap_or_default();
109        let mut attributes = self.common(&response);
110        attributes.extend([
111            KeyValue::new("gen_ai.operation.name", "chat"),
112            KeyValue::new("gen_ai.provider.name", self.provider_name.clone()),
113        ]);
114        if !model.is_empty() {
115            attributes.push(KeyValue::new("gen_ai.request.model", model.clone()));
116            attributes.push(KeyValue::new("gen_ai.response.model", model.clone()));
117        }
118        let input_tokens = request
119            .as_ref()
120            .and_then(|request| request.prompt_tokens)
121            .or(response.prompt_tokens);
122        if let Some(tokens) = input_tokens.and_then(|tokens| i64::try_from(tokens).ok()) {
123            attributes.push(KeyValue::new("gen_ai.usage.input_tokens", tokens));
124        }
125        if let Some(tokens) = response
126            .completion_tokens
127            .and_then(|tokens| i64::try_from(tokens).ok())
128        {
129            attributes.push(KeyValue::new("gen_ai.usage.output_tokens", tokens));
130        }
131        if let Some(reason) = &response.finish_reason {
132            attributes.push(KeyValue::new(
133                "gen_ai.response.finish_reasons",
134                OtelValue::Array(Array::String(vec![StringValue::from(reason.clone())])),
135            ));
136        }
137        if let Some(request) = &request {
138            if let Some(message) = &request.message {
139                let input = json!([{
140                    "role": role(request.source.as_deref()),
141                    "parts": parts(message),
142                }]);
143                attributes.push(KeyValue::new(
144                    "gen_ai.input.messages",
145                    bounded(&input.to_string()),
146                ));
147            }
148            attributes.push(KeyValue::new(
149                "agentvisor.request.event.uid",
150                request.event_uid.clone(),
151            ));
152        }
153        let mut output_parts = response.message.as_ref().map(parts).unwrap_or_default();
154        if let Some(Value::Array(calls)) = &response.tool_calls {
155            for call in calls {
156                output_parts.push(json!({
157                    "type": "tool_call",
158                    "id": call.get("tool_call_id"),
159                    "name": call.get("function_name"),
160                    "arguments": call.get("arguments"),
161                }));
162            }
163        }
164        if let Some(reasoning) = &response.reasoning {
165            output_parts.insert(0, json!({"type": "reasoning", "content": reasoning}));
166        }
167        let output = json!([{
168            "role": "assistant",
169            "parts": output_parts,
170            "finish_reason": response.finish_reason,
171        }]);
172        attributes.push(KeyValue::new(
173            "gen_ai.output.messages",
174            bounded(&output.to_string()),
175        ));
176        let start = request
177            .as_ref()
178            .map_or(response.time_ms, |request| request.time_ms);
179        let name = if model.is_empty() {
180            "chat".to_owned()
181        } else {
182            format!("chat {model}")
183        };
184        self.emit(name, SpanKind::Client, start, response.time_ms, attributes);
185    }
186
187    fn tool(&self, record: &ContentRecord) {
188        let call = record
189            .tool_calls
190            .as_ref()
191            .and_then(Value::as_array)
192            .and_then(|calls| calls.first());
193        let tool = call
194            .and_then(|call| call.get("function_name"))
195            .and_then(Value::as_str)
196            .or_else(|| record.event.pointer("/payload/tool").and_then(Value::as_str))
197            .unwrap_or("unknown")
198            .to_owned();
199        let mut attributes = self.common(record);
200        attributes.extend([
201            KeyValue::new("gen_ai.operation.name", "execute_tool"),
202            KeyValue::new("gen_ai.tool.name", tool.clone()),
203            KeyValue::new("gen_ai.tool.type", "extension"),
204        ]);
205        if let Some(id) = call
206            .and_then(|call| call.get("tool_call_id"))
207            .and_then(Value::as_str)
208        {
209            attributes.push(KeyValue::new("gen_ai.tool.call.id", id.to_owned()));
210        }
211        if let Some(arguments) = call.and_then(|call| call.get("arguments")) {
212            attributes.push(KeyValue::new(
213                "gen_ai.tool.call.arguments",
214                bounded(&arguments.to_string()),
215            ));
216        }
217        let payload = record.event.get("payload").cloned().unwrap_or(Value::Null);
218        if let Some(allowed) = payload.get("allowed").and_then(Value::as_bool) {
219            attributes.push(KeyValue::new("agentvisor.tool.allowed", allowed));
220        }
221        for (key, attribute) in [
222            ("denial_code", "agentvisor.tool.denial_code"),
223            ("policy", "agentvisor.tool.policy"),
224            ("stage", "agentvisor.tool.stage"),
225            ("reason", "agentvisor.tool.reason"),
226        ] {
227            if let Some(value) = payload.get(key).and_then(Value::as_str) {
228                attributes.push(KeyValue::new(attribute, bounded(value)));
229            }
230        }
231        self.emit(
232            format!("execute_tool {tool}"),
233            SpanKind::Internal,
234            record.time_ms,
235            record.time_ms,
236            attributes,
237        );
238    }
239
240    fn tool_result(&self, record: &ContentRecord) {
241        let mut attributes = self.common(record);
242        attributes.push(KeyValue::new("gen_ai.operation.name", "execute_tool"));
243        if let Some(key) = record
244            .event
245            .pointer("/payload/execution_key")
246            .and_then(Value::as_str)
247        {
248            attributes.push(KeyValue::new("agentvisor.tool.execution_key", key.to_owned()));
249        }
250        if let Some(status) = record.event.pointer("/payload/status").and_then(Value::as_i64) {
251            attributes.push(KeyValue::new("agentvisor.tool.http_status", status));
252        }
253        if let Some(message) = &record.message {
254            let text = message
255                .as_str()
256                .map_or_else(|| message.to_string(), str::to_owned);
257            attributes.push(KeyValue::new("gen_ai.tool.call.result", bounded(&text)));
258        }
259        self.emit(
260            "execute_tool.result".to_owned(),
261            SpanKind::Internal,
262            record.time_ms,
263            record.time_ms,
264            attributes,
265        );
266    }
267
268    fn event(&self, record: &ContentRecord) {
269        let mut attributes = self.common(record);
270        attributes.push(KeyValue::new(
271            "agentvisor.event",
272            bounded(&record.event.to_string()),
273        ));
274        self.emit(
275            format!("agentvisor.{}", class_label(record.class)),
276            SpanKind::Internal,
277            record.time_ms,
278            record.time_ms,
279            attributes,
280        );
281    }
282}
283
284impl ContentSink for TenantContentSink {
285    fn record(&self, mut record: ContentRecord) {
286        self.redact(&mut record);
287        let attempt = record.attempt_id.clone();
288        match (attempt, record.terminal) {
289            // A chat request: hold it until its response completes.
290            (Some(attempt), false) => {
291                let evicted = {
292                    let mut pending = self.pending.lock();
293                    let (by_id, order) = &mut *pending;
294                    by_id.insert(attempt.clone(), record);
295                    order.push_back(attempt);
296                    let mut evicted = Vec::new();
297                    while order.len() > MAX_PENDING_CHATS {
298                        if let Some(stale) = order.pop_front().and_then(|id| by_id.remove(&id)) {
299                            evicted.push(stale);
300                        }
301                    }
302                    evicted
303                };
304                for stale in evicted {
305                    self.event(&stale);
306                }
307            }
308            (Some(attempt), true) => {
309                let request = {
310                    let mut pending = self.pending.lock();
311                    let (by_id, order) = &mut *pending;
312                    order.retain(|id| *id != attempt);
313                    by_id.remove(&attempt)
314                };
315                self.chat(request, record);
316            }
317            (None, _) if record.class == EventClass::ToolCall => self.tool(&record),
318            (None, _)
319                if record.class == EventClass::Session
320                    && record.event.pointer("/payload/action").and_then(Value::as_str)
321                        == Some("tool_completed") =>
322            {
323                self.tool_result(&record);
324            }
325            (None, _) => self.event(&record),
326        }
327    }
328}
329
330fn class_label(class: EventClass) -> &'static str {
331    match class {
332        EventClass::ToolCall => "tool_call",
333        EventClass::StopReason => "stop_reason",
334        EventClass::Receipt => "receipt",
335        EventClass::Compression => "compression",
336        EventClass::Identity => "identity",
337        EventClass::Session => "session",
338        _ => "event",
339    }
340}
341
342fn role(source: Option<&str>) -> &'static str {
343    match source {
344        Some("agent") => "assistant",
345        Some("system") => "system",
346        _ => "user",
347    }
348}
349
350/// GenAI message parts for a chat message's content: a string becomes one
351/// text part; OpenAI content-part arrays keep their text parts and carry
352/// any other part as JSON.
353fn parts(message: &Value) -> Vec<Value> {
354    match message {
355        Value::String(text) => vec![json!({"type": "text", "content": text})],
356        Value::Array(items) => items
357            .iter()
358            .map(|item| match item.get("text").and_then(Value::as_str) {
359                Some(text) => json!({"type": "text", "content": text}),
360                None => json!({"type": "text", "content": item.to_string()}),
361            })
362            .collect(),
363        other => vec![json!({"type": "text", "content": other.to_string()})],
364    }
365}
366
367/// Truncate at a character boundary so no attribute exceeds the limit.
368pub fn bounded(text: &str) -> String {
369    if text.len() <= MAX_ATTRIBUTE_BYTES {
370        return text.to_owned();
371    }
372    let mut end = MAX_ATTRIBUTE_BYTES;
373    while end > 0 && !text.is_char_boundary(end) {
374        end -= 1;
375    }
376    let cut = text.len().saturating_sub(end);
377    format!("{}…[truncated {cut} bytes]", text.get(..end).unwrap_or_default())
378}
379
380/// Wall-clock milliseconds, for records built outside the worker.
381pub fn now_ms() -> u64 {
382    SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |elapsed| {
383        u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)
384    })
385}
386
387#[cfg(test)]
388mod tests {
389    #![allow(
390        clippy::unwrap_used,
391        clippy::expect_used,
392        clippy::panic,
393        clippy::indexing_slicing
394    )]
395
396    use super::*;
397
398    #[test]
399    fn long_values_are_truncated_on_a_character_boundary() {
400        let text = "é".repeat(MAX_ATTRIBUTE_BYTES);
401        let cut = bounded(&text);
402        assert!(cut.len() < text.len());
403        assert!(cut.contains("[truncated"));
404        assert_eq!(bounded("short"), "short");
405    }
406
407    fn record(class: EventClass, attempt: Option<&str>, terminal: bool) -> ContentRecord {
408        ContentRecord {
409            session_id: "session-1".into(),
410            event_uid: av_core::ids::new_event_uid(),
411            class,
412            success: true,
413            identity: av_events::AgentIdentity {
414                version: "1".into(),
415                charter: "support".into(),
416                instance_uid: "inst-1".into(),
417                ttl_remaining_s: None,
418            },
419            time_ms: 1_000,
420            event: json!({"payload": {}}),
421            source: None,
422            message: None,
423            reasoning: None,
424            model: None,
425            tool_calls: None,
426            observation: None,
427            prompt_tokens: None,
428            completion_tokens: None,
429            finish_reason: None,
430            attempt_id: attempt.map(str::to_owned),
431            terminal,
432        }
433    }
434
435    fn attribute(span: &opentelemetry_sdk::trace::SpanData, key: &str) -> Option<String> {
436        span.attributes
437            .iter()
438            .find(|attribute| attribute.key.as_str() == key)
439            .map(|attribute| attribute.value.to_string())
440    }
441
442    #[test]
443    fn steps_become_redacted_genai_spans() {
444        let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default();
445        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
446            .with_simple_exporter(exporter.clone())
447            .build();
448        let redaction = av_redact::RedactionEngine::new(av_redact::RedactionConfig {
449            include_builtin_patterns: true,
450            ..av_redact::RedactionConfig::default()
451        })
452        .unwrap();
453        let sink = TenantContentSink::new(&provider, redaction, "openai");
454
455        let mut request = record(EventClass::Compression, Some("a1"), false);
456        request.source = Some("user".into());
457        request.message = Some(json!("mail me at bob@example.com"));
458        request.prompt_tokens = Some(12);
459        sink.record(request);
460        let mut response = record(EventClass::StopReason, Some("a1"), true);
461        response.time_ms = 1_500;
462        response.model = Some("gpt-x".into());
463        response.message = Some(json!("done"));
464        response.completion_tokens = Some(3);
465        response.finish_reason = Some("stop".into());
466        sink.record(response);
467
468        let mut tool = record(EventClass::ToolCall, None, false);
469        tool.tool_calls = Some(json!([{
470            "tool_call_id": "7", "function_name": "lookup",
471            "arguments": {"card": "4111 1111 1111 1111"}
472        }]));
473        tool.event = json!({"payload": {"tool": "lookup", "allowed": true}});
474        sink.record(tool);
475        let mut result = record(EventClass::Session, None, false);
476        result.event = json!({"payload": {"action": "tool_completed", "execution_key": "k", "status": 200}});
477        result.message = Some(json!("{\"answer\": 42}"));
478        sink.record(result);
479        sink.record(record(EventClass::StopReason, None, false));
480
481        provider.force_flush().unwrap();
482        let spans = exporter.get_finished_spans().unwrap();
483        let names: Vec<&str> = spans.iter().map(|span| span.name.as_ref()).collect();
484        assert_eq!(
485            names,
486            vec![
487                "chat gpt-x",
488                "execute_tool lookup",
489                "execute_tool.result",
490                "agentvisor.stop_reason"
491            ]
492        );
493        let chat = &spans[0];
494        assert_eq!(chat.span_kind, SpanKind::Client);
495        let input = attribute(chat, "gen_ai.input.messages").unwrap();
496        assert!(
497            input.contains("[REDACTED]") && !input.contains("bob@example.com"),
498            "{input}"
499        );
500        assert!(attribute(chat, "gen_ai.output.messages")
501            .unwrap()
502            .contains("done"));
503        assert_eq!(
504            attribute(chat, "gen_ai.usage.input_tokens").as_deref(),
505            Some("12")
506        );
507        assert_eq!(
508            attribute(chat, "gen_ai.usage.output_tokens").as_deref(),
509            Some("3")
510        );
511        assert_eq!(attribute(chat, "gen_ai.provider.name").as_deref(), Some("openai"));
512        assert_eq!(
513            attribute(chat, "gen_ai.conversation.id").as_deref(),
514            Some("session-1")
515        );
516        assert_eq!(
517            chat.end_time.duration_since(chat.start_time).unwrap(),
518            Duration::from_millis(500),
519            "the chat span runs from the request to the response"
520        );
521        let arguments = attribute(&spans[1], "gen_ai.tool.call.arguments").unwrap();
522        assert!(!arguments.contains("4111 1111 1111 1111"), "{arguments}");
523        assert_eq!(attribute(&spans[1], "gen_ai.tool.call.id").as_deref(), Some("7"));
524        assert_eq!(
525            attribute(&spans[1], "agentvisor.tool.allowed").as_deref(),
526            Some("true")
527        );
528        assert!(attribute(&spans[2], "gen_ai.tool.call.result")
529            .unwrap()
530            .contains("42"));
531    }
532
533    #[test]
534    fn parts_follow_the_genai_message_shape() {
535        assert_eq!(
536            parts(&json!("hi")),
537            vec![json!({"type": "text", "content": "hi"})]
538        );
539        assert_eq!(
540            parts(&json!([{"type": "text", "text": "a"}, {"type": "image_url", "image_url": {"url": "x"}}]))
541                [0],
542            json!({"type": "text", "content": "a"})
543        );
544        assert_eq!(role(Some("agent")), "assistant");
545        assert_eq!(role(Some("user")), "user");
546    }
547
548    #[test]
549    fn pending_chats_are_evicted_and_emitted_when_the_map_overflows() {
550        let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default();
551        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
552            .with_simple_exporter(exporter.clone())
553            .build();
554        let redaction = av_redact::RedactionEngine::new(av_redact::RedactionConfig {
555            include_builtin_patterns: true,
556            ..av_redact::RedactionConfig::default()
557        })
558        .unwrap();
559        let sink = TenantContentSink::new(&provider, redaction, "openai");
560
561        // Overfill the pending-chat map. Each request is non-terminal and
562        // carries an attempt id, so it is held rather than emitted.
563        for i in 0..(MAX_PENDING_CHATS + 10) {
564            let mut request = record(EventClass::Compression, Some(&format!("a{i}")), false);
565            request.message = Some(json!("hello"));
566            sink.record(request);
567        }
568
569        // The map must never exceed the limit.
570        {
571            let pending = sink.pending.lock();
572            assert!(pending.0.len() <= MAX_PENDING_CHATS);
573            assert_eq!(pending.1.len(), pending.0.len());
574        }
575
576        // Evicted records are emitted as event spans, not silently dropped.
577        provider.force_flush().unwrap();
578        let spans = exporter.get_finished_spans().unwrap();
579        assert!(
580            spans.len() >= 10,
581            "evicted requests must be emitted, got {}",
582            spans.len()
583        );
584    }
585
586    #[test]
587    fn terminal_responses_without_a_pending_request_still_emit() {
588        let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default();
589        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
590            .with_simple_exporter(exporter.clone())
591            .build();
592        let redaction = av_redact::RedactionEngine::new(av_redact::RedactionConfig {
593            include_builtin_patterns: true,
594            ..av_redact::RedactionConfig::default()
595        })
596        .unwrap();
597        let sink = TenantContentSink::new(&provider, redaction, "openai");
598
599        // A terminal record with an attempt id but no matching request.
600        let mut response = record(EventClass::StopReason, Some("orphan"), true);
601        response.message = Some(json!("orphan response"));
602        sink.record(response);
603
604        provider.force_flush().unwrap();
605        let spans = exporter.get_finished_spans().unwrap();
606        assert_eq!(spans.len(), 1);
607        assert_eq!(spans[0].name, "chat");
608        assert!(attribute(&spans[0], "gen_ai.output.messages")
609            .unwrap()
610            .contains("orphan response"));
611    }
612
613    #[test]
614    fn tool_records_without_tool_calls_fall_back_to_event_payload() {
615        let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default();
616        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
617            .with_simple_exporter(exporter.clone())
618            .build();
619        let redaction = av_redact::RedactionEngine::new(av_redact::RedactionConfig {
620            include_builtin_patterns: true,
621            ..av_redact::RedactionConfig::default()
622        })
623        .unwrap();
624        let sink = TenantContentSink::new(&provider, redaction, "openai");
625
626        // ToolCall with no tool_calls array: the tool name comes from the
627        // event payload.
628        let mut tool = record(EventClass::ToolCall, None, false);
629        tool.event = json!({"payload": {"tool": "from-payload", "allowed": false, "denial_code": "UNMAPPED_TOOL", "policy": "pdp.intent_map", "stage": "intent", "reason": "not mapped"}});
630        sink.record(tool);
631
632        provider.force_flush().unwrap();
633        let spans = exporter.get_finished_spans().unwrap();
634        assert_eq!(spans.len(), 1);
635        assert_eq!(spans[0].name, "execute_tool from-payload");
636        assert_eq!(
637            attribute(&spans[0], "agentvisor.tool.allowed").as_deref(),
638            Some("false")
639        );
640        assert_eq!(
641            attribute(&spans[0], "agentvisor.tool.denial_code").as_deref(),
642            Some("UNMAPPED_TOOL")
643        );
644        assert_eq!(
645            attribute(&spans[0], "agentvisor.tool.policy").as_deref(),
646            Some("pdp.intent_map")
647        );
648        assert_eq!(
649            attribute(&spans[0], "agentvisor.tool.stage").as_deref(),
650            Some("intent")
651        );
652    }
653
654    #[test]
655    fn tool_result_without_message_or_payload_still_emits() {
656        let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default();
657        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
658            .with_simple_exporter(exporter.clone())
659            .build();
660        let redaction = av_redact::RedactionEngine::new(av_redact::RedactionConfig {
661            include_builtin_patterns: true,
662            ..av_redact::RedactionConfig::default()
663        })
664        .unwrap();
665        let sink = TenantContentSink::new(&provider, redaction, "openai");
666
667        // tool_completed with no message and no execution_key.
668        let mut result = record(EventClass::Session, None, false);
669        result.event = json!({"payload": {"action": "tool_completed"}});
670        sink.record(result);
671
672        provider.force_flush().unwrap();
673        let spans = exporter.get_finished_spans().unwrap();
674        assert_eq!(spans.len(), 1);
675        assert_eq!(spans[0].name, "execute_tool.result");
676    }
677
678    #[test]
679    fn redaction_strips_reasoning_and_tool_calls_before_export() {
680        let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default();
681        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
682            .with_simple_exporter(exporter.clone())
683            .build();
684        let redaction = av_redact::RedactionEngine::new(av_redact::RedactionConfig {
685            include_builtin_patterns: true,
686            ..av_redact::RedactionConfig::default()
687        })
688        .unwrap();
689        let sink = TenantContentSink::new(&provider, redaction, "openai");
690
691        let mut request = record(EventClass::Compression, Some("r1"), false);
692        request.message = Some(json!("my SSN is 123-45-6789"));
693        request.reasoning = Some("thinking about 4111 1111 1111 1111".into());
694        sink.record(request);
695        let mut response = record(EventClass::StopReason, Some("r1"), true);
696        response.message = Some(json!("done"));
697        response.tool_calls = Some(json!([{
698            "tool_call_id": "1",
699            "function_name": "lookup",
700            "arguments": {"card": "4111 1111 1111 1111"}
701        }]));
702        sink.record(response);
703
704        provider.force_flush().unwrap();
705        let spans = exporter.get_finished_spans().unwrap();
706        assert_eq!(spans.len(), 1);
707        let chat = &spans[0];
708        let input = attribute(chat, "gen_ai.input.messages").unwrap();
709        assert!(!input.contains("123-45-6789"), "SSN leaked: {input}");
710        assert!(input.contains("[REDACTED]"), "{input}");
711        let output = attribute(chat, "gen_ai.output.messages").unwrap();
712        assert!(
713            !output.contains("4111 1111 1111 1111"),
714            "card leaked in output: {output}"
715        );
716    }
717
718    #[test]
719    fn class_labels_are_stable() {
720        assert_eq!(class_label(EventClass::ToolCall), "tool_call");
721        assert_eq!(class_label(EventClass::StopReason), "stop_reason");
722        assert_eq!(class_label(EventClass::Receipt), "receipt");
723        assert_eq!(class_label(EventClass::Compression), "compression");
724        assert_eq!(class_label(EventClass::Identity), "identity");
725        assert_eq!(class_label(EventClass::Session), "session");
726    }
727
728    #[test]
729    fn bounded_preserves_ascii_and_never_exceeds_limit() {
730        assert_eq!(bounded("short"), "short");
731        assert_eq!(bounded(""), "");
732        let text = "a".repeat(MAX_ATTRIBUTE_BYTES);
733        assert_eq!(bounded(&text), text);
734        let oversized = "b".repeat(MAX_ATTRIBUTE_BYTES + 1);
735        let cut = bounded(&oversized);
736        assert!(
737            cut.len() <= MAX_ATTRIBUTE_BYTES + 64,
738            "cut too large: {}",
739            cut.len()
740        );
741        assert!(cut.contains("[truncated"));
742        // "é" is two bytes, so this input is exactly at the limit and is
743        // returned unchanged.
744        let at_limit = "é".repeat(MAX_ATTRIBUTE_BYTES / 2);
745        assert_eq!(at_limit.len(), MAX_ATTRIBUTE_BYTES);
746        assert_eq!(bounded(&at_limit), at_limit);
747
748        // One more character puts the input over the limit, so it is cut —
749        // on a character boundary, never through the middle of "é".
750        let text = "é".repeat(MAX_ATTRIBUTE_BYTES / 2 + 1);
751        let cut = bounded(&text);
752        assert!(cut.contains("[truncated"), "not truncated: {cut}");
753        assert!(
754            cut.len() <= MAX_ATTRIBUTE_BYTES + 64,
755            "cut too large: {}",
756            cut.len()
757        );
758        let kept = cut.split('…').next().unwrap_or_default();
759        assert!(kept.ends_with('é'), "cut split a character: {kept}");
760        assert!(text.starts_with(kept), "kept prefix is not a prefix of the input");
761    }
762
763    #[test]
764    fn parts_handle_empty_and_nested_arrays() {
765        assert_eq!(parts(&json!([])), Vec::<Value>::new());
766        assert_eq!(
767            parts(&json!([{"type": "text", "text": "a"}, {"type": "text", "text": "b"}])),
768            vec![
769                json!({"type": "text", "content": "a"}),
770                json!({"type": "text", "content": "b"})
771            ]
772        );
773        assert_eq!(
774            parts(&json!(null)),
775            vec![json!({"type": "text", "content": "null"})]
776        );
777        assert_eq!(parts(&json!(42)), vec![json!({"type": "text", "content": "42"})]);
778    }
779}
av_harness/otel_content.rs

av_harness/
otel_content.rs
