relay_knowledge/interfaces/agent/audit/
mod.rs

1use std::{
2    collections::VecDeque,
3    io,
4    path::{Path, PathBuf},
5    sync::{Arc, Mutex},
6};
7
8use serde::{Deserialize, Serialize};
9use tokio::{io::AsyncWriteExt, sync::mpsc};
10
11use crate::api::{AgentProtocolKind, RuntimeIdentity};
12
13const MAX_AUDIT_EVENTS: usize = 1024;
14const MAX_AUDIT_SINK_QUEUE_DEPTH: usize = 65_536;
15
16/// In-memory bounded audit log shared by resident agent adapters.
17#[derive(Clone)]
18pub struct AgentAuditLog {
19    inner: Arc<Mutex<AgentAuditState>>,
20    sink: Option<AgentAuditSink>,
21}
22
23#[derive(Default)]
24struct AgentAuditState {
25    entries: VecDeque<AgentAuditEvent>,
26    next_sequence: u64,
27}
28
29/// QoS decision captured before agent adapter work reaches application services.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
31#[serde(rename_all = "snake_case")]
32pub enum AgentAuditQosDecision {
33    Admitted,
34    Rejected,
35}
36
37/// Final status captured for an agent protocol operation.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
39#[serde(rename_all = "snake_case")]
40pub enum AgentAuditStatus {
41    Completed,
42    Failed,
43    Cancelled,
44}
45
46/// Redacted audit event for agent protocol requests.
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48pub struct AgentAuditEvent {
49    pub sequence: u64,
50    pub protocol: AgentProtocolKind,
51    pub operation: String,
52    pub request_id: String,
53    pub trace_id: String,
54    pub runtime_identity: RuntimeIdentity,
55    pub qos_decision: AgentAuditQosDecision,
56    pub status: AgentAuditStatus,
57    #[serde(skip_serializing_if = "Option::is_none")]
58    pub source_scope: Option<String>,
59    #[serde(skip_serializing_if = "Option::is_none")]
60    pub freshness: Option<String>,
61    #[serde(skip_serializing_if = "Option::is_none")]
62    pub limit: Option<usize>,
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub result_count: Option<usize>,
65    pub truncated: bool,
66    pub elapsed_ms: u64,
67    #[serde(skip_serializing_if = "Option::is_none")]
68    pub error_kind: Option<String>,
69}
70
71/// Optional async sink for durable resident-agent audit events.
72#[derive(Clone)]
73pub struct AgentAuditSink {
74    sender: mpsc::Sender<AgentAuditEvent>,
75}
76
77impl AgentAuditSink {
78    /// Spawns a bounded JSONL writer owned by the async runtime.
79    pub fn jsonl(path: PathBuf, queue_depth: usize) -> Option<Self> {
80        let handle = tokio::runtime::Handle::try_current().ok()?;
81        let queue_depth = queue_depth.clamp(1, MAX_AUDIT_SINK_QUEUE_DEPTH);
82        let (sender, mut receiver) = mpsc::channel(queue_depth);
83        handle.spawn(async move {
84            while let Some(event) = receiver.recv().await {
85                let _ = append_jsonl_event(&path, &event).await;
86            }
87        });
88
89        Some(Self { sender })
90    }
91
92    fn enqueue(&self, event: AgentAuditEvent) {
93        let _ = self.sender.try_send(event);
94    }
95}
96
97impl Default for AgentAuditLog {
98    fn default() -> Self {
99        Self {
100            inner: Arc::new(Mutex::new(AgentAuditState::default())),
101            sink: None,
102        }
103    }
104}
105
106impl AgentAuditLog {
107    /// Creates a bounded in-memory log with a durable async mirror.
108    pub fn with_sink(sink: AgentAuditSink) -> Self {
109        Self {
110            inner: Arc::new(Mutex::new(AgentAuditState::default())),
111            sink: Some(sink),
112        }
113    }
114
115    /// Records an event and returns its monotonic in-process sequence number.
116    pub fn record(&self, mut event: AgentAuditEvent) -> u64 {
117        let sequence = {
118            let mut state = self
119                .inner
120                .lock()
121                .unwrap_or_else(|poisoned| poisoned.into_inner());
122            state.next_sequence = state.next_sequence.wrapping_add(1).max(1);
123            event.sequence = state.next_sequence;
124            state.entries.push_back(event.clone());
125            while state.entries.len() > MAX_AUDIT_EVENTS {
126                state.entries.pop_front();
127            }
128
129            state.next_sequence
130        };
131        if let Some(sink) = &self.sink {
132            sink.enqueue(event);
133        }
134
135        sequence
136    }
137
138    /// Returns a stable snapshot for diagnostics and tests.
139    pub fn snapshot(&self) -> Vec<AgentAuditEvent> {
140        self.inner
141            .lock()
142            .unwrap_or_else(|poisoned| poisoned.into_inner())
143            .entries
144            .iter()
145            .cloned()
146            .collect()
147    }
148}
149
150async fn append_jsonl_event(path: &Path, event: &AgentAuditEvent) -> io::Result<()> {
151    if let Some(parent) = path.parent() {
152        tokio::fs::create_dir_all(parent).await?;
153    }
154    let mut file = tokio::fs::OpenOptions::new()
155        .create(true)
156        .append(true)
157        .open(path)
158        .await?;
159    let mut line = serde_json::to_vec(event).map_err(io::Error::other)?;
160    line.push(b'\n');
161    file.write_all(&line).await?;
162    file.flush().await
163}
164
165#[cfg(test)]
166#[path = "mod_tests.rs"]
167mod tests;
relay_knowledge/interfaces/agent/audit/mod.rs

relay_knowledge/interfaces/agent/audit/
mod.rs
