agentd/runtime/
audit.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//! The **audit stream**: an append-only
3//! record of *who did what* — every A2A call, every principal-driven tool/command,
4//! config reloads, restores, store conflicts, and kills. Each event is
5//! `{ts, principal, role, action, target, outcome, request_id, trace, instance}`,
6//! emitted to the configured sinks: `log` (a closed-vocabulary `audit` log line)
7//! and/or `store` (a durable, append-only `Kind::Audit` record, ULID-keyed — never
8//! CAS'd, never listed, so it cannot be rewritten). Audit is security telemetry:
9//! it answers "why did the agent do that, and on whose authority?".
10
11use crate::config::v2::AuditSink;
12use crate::runtime::reactor::Runtime;
13use crate::state::{Kind, now_ms, ulid};
14use serde_json::{Value, json};
15
16/// One audit event to record.
17pub(crate) struct AuditEvent<'a> {
18    pub action: &'a str,
19    pub target: Value,
20    pub outcome: &'a str,
21    pub principal: Option<&'a str>,
22    pub role: Option<&'a str>,
23    pub request_id: Option<&'a str>,
24}
25
26impl Runtime {
27    /// Emit an audit event to the configured sinks. A no-op when no sink is
28    /// configured (`observability.audit.sink`). Cheap on the common path.
29    pub(crate) fn audit(&self, ev: AuditEvent<'_>) {
30        // Mirror onto the interface feed as operator-visible `audit` events
31        // when debug is on — independent of the sinks, which stay the
32        // durable/system record. The taskless interface READS are
33        // excluded: a display client polls them (debug.events at ~1 Hz), and
34        // mirroring their own audit back onto the feed would feed-loop the
35        // debug pane with its own plumbing. The durable sinks still record
36        // them.
37        #[cfg(feature = "a2a")]
38        if let Some(feed) = &self.a2a_feed
39            && feed.debug()
40            && !ev.action.ends_with(":interface.info")
41            && !ev.action.ends_with(":conversation.get")
42            && !ev.action.ends_with(":run.get")
43            && !ev.action.ends_with(":subagent.get")
44            && !ev.action.ends_with(":debug.events")
45            && !ev.action.ends_with(":pairing.code")
46        {
47            feed.push(
48                "audit",
49                super::a2a_server::FeedVis::Operator,
50                json!({
51                    "ts": now_ms(),
52                    "principal": ev.principal,
53                    "role": ev.role,
54                    "action": ev.action,
55                    "target": ev.target,
56                    "outcome": ev.outcome,
57                }),
58            );
59        }
60        let Some(sinks) = &self.settings.observability.audit.sink else {
61            return;
62        };
63        if sinks.is_empty() {
64            return;
65        }
66        let record = json!({
67            "ts": now_ms(),
68            "instance": self.instance,
69            "principal": ev.principal,
70            "role": ev.role,
71            "action": ev.action,
72            "target": ev.target,
73            "outcome": ev.outcome,
74            "request_id": ev.request_id,
75            "trace": self.trace_id,
76        });
77        if sinks.iter().any(|s| matches!(s, AuditSink::Log)) {
78            // A single closed-vocabulary `audit` event (never content-suppressed —
79            // an audit trail is metadata, not conversation content).
80            self.log.info("audit", record.clone());
81        }
82        if sinks.iter().any(|s| matches!(s, AuditSink::Stream))
83            && let Some(stream) = &self.settings.observability.audit.stream
84        {
85            // Queued, not appended: `audit` runs on `&self` from every
86            // authorization path, and the append needs the state owner. The
87            // tick drains it, which also puts these records behind the same
88            // pressure gate as every other admission.
89            crate::obs::log::tap_direct(stream, "audit", record.clone());
90        }
91        if sinks.iter().any(|s| matches!(s, AuditSink::Store)) {
92            // Append-only: a fresh ULID id per event (Kind::Audit is not indexed,
93            // so this never conflicts and is never overwritten).
94            let id = ulid::new();
95            if let Err(e) = self.durable.put(Kind::Audit, &id, record, None) {
96                // The store sink is best-effort telemetry — a failed audit write is
97                // logged but never fails the audited action.
98                self.log.warn(
99                    "audit.store.fail",
100                    json!({"action": ev.action, "err": e.to_string()}),
101                );
102            }
103        }
104    }
105
106    /// Audit an A2A request (the principal, the method/op, the outcome).
107    #[cfg(feature = "a2a")]
108    pub(crate) fn audit_a2a(
109        &self,
110        method: &str,
111        op: Option<&str>,
112        principal: &crate::a2a::Principal,
113        outcome: &str,
114        target: Value,
115        request_id: Option<&str>,
116    ) {
117        let action = match op {
118            Some(o) => format!("a2a.{method}:{o}"),
119            None => format!("a2a.{method}"),
120        };
121        let role = format!("{:?}", principal.role).to_lowercase();
122        self.audit(AuditEvent {
123            action: &action,
124            target,
125            outcome,
126            principal: Some(&principal.id),
127            role: Some(&role),
128            request_id,
129        });
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    // The emitter is exercised end-to-end by `runtime_v2_audit_e2e` (a real
136    // daemon with `observability.audit.sink: [log]`); a pure-unit test would only
137    // restate the JSON shape. The shape is asserted there against the log line.
138}
agentd/runtime/audit.rs

agentd/runtime/
audit.rs
