diff --git a/crates/buzz-acp/src/audit.rs b/crates/buzz-acp/src/audit.rs new file mode 100644 index 00000000000..c208041f4a1 --- /dev/null +++ b/crates/buzz-acp/src/audit.rs @@ -0,0 +1,867 @@ +//! Phase B causal audit trail — local, append-only, metadata-only, fail-open. +//! +//! Single sink for the nine MVP causal boundaries defined in +//! `HIVE_PHASE_B_EMIT_POINT_SPEC_FIZZ_HONEY.md`: `EVENT_PUBLISHED`, +//! `RELAY_ACCEPTED`, `EVENT_RECEIVED`, `FILTER_DECISION`, `QUEUE_DECISION`, +//! `WAKE_DECISION`, `ACP_SESSION_STARTED`, `RESPONSE_PUBLISHED`, +//! `RECIPIENT_EVIDENCE_OBSERVED`. +//! +//! No hook lives in this file — it defines the schema and the sink only. +//! Call sites are added elsewhere, one per boundary, each an additive +//! `audit::record(...)` statement next to an existing decision. +//! +//! # Design constraints (Schema V3 + sink architecture, approved after +//! causality and hot-path review) +//! +//! - **Local only by default.** `$HOME/.buzz/audit/buzz-acp-audit.jsonl` — +//! no network call, no relay publish, no database. This is deliberately +//! *not* [`crate::observer`], which exists to publish encrypted telemetry +//! frames to the relay. `BUZZ_ACP_AUDIT_PATH`, when set, is used verbatim +//! and is entirely operator-controlled: nothing here asserts an explicit +//! override can't point at a mounted or network filesystem — only the +//! *default* (no override) is guaranteed strictly local. +//! - **Metadata only, redaction by construction.** [`AuditFields`] and +//! [`EventDetail`] have no field that can hold raw event content, prompts, +//! secrets, or arbitrary relay-supplied text. +//! - **Global, opt-in, default OFF.** Controlled by `BUZZ_ACP_AUDIT`, read +//! once at startup — outside [`crate::config::Config`]'s clap/env surface +//! (per-agent business configuration), and unrelated to the Phase A +//! per-agent `RUST_LOG` experiment. +//! - **Non-blocking producer, fail-open sink.** `record()` never performs +//! filesystem I/O and never awaits: it builds a JSON line in memory and +//! hands it to a bounded `std::sync::mpsc::sync_channel` via `try_send` +//! only. A dedicated named OS thread is the sole owner of the `File` and +//! does all actual disk I/O off the causal hot path. Backpressure +//! (channel full or the writer thread gone) drops the record, counts it, +//! and logs a rate-limited warning — it never blocks or retries. A write +//! error inside the writer thread is likewise counted, rate-limited, and +//! never propagates or panics. +//! - **No invented business state.** There is no `handoff_id` and no +//! `source_event_id`. Two separately-signed events cannot be proven to be +//! "the same logical handoff, retried" from transport data alone. +//! `workflow_id` is a verbatim tag pass-through, populated only when the +//! Hive protocol layer actually sets it. `attempt` is populated only +//! where [`crate::queue::EventQueue`] has an authoritative retry counter +//! (`QUEUE_DECISION` / `WAKE_DECISION`), `None` everywhere else. +//! - **Honest batch cardinality.** `WAKE_DECISION` and `ACP_SESSION_STARTED` +//! can each concern more than one physical event (a `FlushBatch` may hold +//! several). The scalar `AuditFields::event_id` is `None` at both — the +//! full triggering set lives in `EventDetail::WakeDecision` / +//! `EventDetail::AcpSession` instead. The two are paired by `turn_id` +//! (reused verbatim from `dispatch_pending`'s existing `Uuid::new_v4()` +//! local, not invented here) and independently verifiable by equal +//! `triggering_event_ids`. + +use std::fs::OpenOptions; +use std::io::Write as _; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{sync_channel, SyncSender, TrySendError}; +use std::sync::OnceLock; + +use serde::Serialize; + +/// Environment variable that globally enables the audit sink for this +/// process. Accepts `1` / `true` (case-insensitive); anything else — +/// including unset — is disabled. The only switch: no per-agent override, +/// no config-file key, no relationship to `RUST_LOG`. +const AUDIT_ENABLE_ENV: &str = "BUZZ_ACP_AUDIT"; + +/// Optional override for the JSONL sink path. When unset, defaults to +/// `$HOME/.buzz/audit/` + [`DEFAULT_AUDIT_FILENAME`]. +const AUDIT_PATH_ENV: &str = "BUZZ_ACP_AUDIT_PATH"; + +const DEFAULT_AUDIT_FILENAME: &str = "buzz-acp-audit.jsonl"; + +/// Bounded producer→writer channel capacity. Fixed — guarantees bounded +/// memory regardless of audit volume; excess records are dropped, not +/// queued without limit. +const CHANNEL_CAPACITY: usize = 4096; + +/// Schema iteration. Bump on any wire-format-breaking change to +/// [`AuditFields`] / [`EventDetail`]. +const SCHEMA_VERSION: u32 = 1; + +/// Log a backpressure/write-error warning only every Nth occurrence, so a +/// sustained overload can't turn the warning itself into a second source of +/// load. The first occurrence (n == 1) always logs immediately. +const WARN_LOG_MODULUS: u64 = 100; + +/// The nine MVP causal boundaries. Serializes as the `event_type` field. +/// +/// `TransportPublished` / `TransportAccepted` stand in for the spec's +/// `EVENT_PUBLISHED` / `RELAY_ACCEPTED` slots, but are deliberately NOT +/// named that: source inspection confirmed genuine agent handoff/chat +/// content (kind:9, sent via `buzz messages send`) is published by the +/// separate `buzz-cli` binary over its own REST connection, never through +/// buzz-acp's WS `HarnessRelay`. buzz-acp's WS publish path only ever +/// carries its own internal traffic (typing indicators, presence, +/// membership, observer frames) — real, useful transport-health telemetry, +/// but not the causal handoff/response fact `EVENT_PUBLISHED` / +/// `RELAY_ACCEPTED` are meant to prove. Those two spec-canonical names are +/// reserved for a future buzz-cli-side implementation and must not be +/// reused here. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum EventType { + /// buzz-acp's own WS publish path (`relay.rs::execute_connected_command`, + /// `RelayCommand::PublishEvent`). Does NOT represent real agent + /// handoff/chat publication and does NOT cover kind:9 messages sent via + /// `buzz messages send` — those go out through buzz-cli's independent + /// REST client, never through this process's relay connection. + TransportPublished, + /// Relay OK acknowledgment for buzz-acp's own WS-published traffic + /// (`relay.rs::handle_ws_message`, `RelayMessage::Ok`). Does NOT + /// represent acceptance of real agent chat messages — those are + /// published through buzz-cli's REST path and acknowledged via that + /// HTTP response, not this WS OK frame. + TransportAccepted, + EventReceived, + FilterDecision, + QueueDecision, + WakeDecision, + AcpSessionStarted, + ResponsePublished, + RecipientEvidenceObserved, +} + +/// Outcome of a `QUEUE_DECISION`. Sourced only in +/// `queue.rs::EventQueue::push`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum QueueOutcome { + Accepted, + /// [`DedupMode::Drop`](crate::config::DedupMode) discarded the event for + /// an in-flight channel. + DroppedInFlight, + /// Per-channel depth cap evicted the oldest queued event to make room. + CapEvicted, +} + +/// Outcome of a `WAKE_DECISION`. Sourced only in `lib.rs::dispatch_pending`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum WakeOutcome { + Claimed, + PoolExhausted, +} + +/// Boundary-specific extras. Only the fields a boundary can actually source +/// appear on its variant — e.g. `rule_index` cannot be set on a +/// `WAKE_DECISION` record because that variant has no such field. +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "kind", rename_all = "SCREAMING_SNAKE_CASE")] +pub enum EventDetail { + /// `TransportPublished`, `EVENT_RECEIVED`, `RESPONSE_PUBLISHED`, + /// `RECIPIENT_EVIDENCE_OBSERVED` — nothing beyond the shared envelope. + Empty, + FilterDecision { + rule_index: Option, + fail_closed: bool, + }, + QueueDecision { + outcome: QueueOutcome, + }, + /// `TransportAccepted` — accept/reject only. Deliberately no + /// relay-supplied message text: that string is arbitrary, + /// externally-controlled content and has no place in a metadata-only, + /// redaction-by-construction schema. + RelayAck { + accepted: bool, + }, + WakeDecision { + /// Exact event set from the `FlushBatch` this decision concerns — + /// same computation `pool.rs::run_prompt_task` already performs + /// (`batch.events.iter().map(|be| be.event.id.to_hex())`), not a + /// new derivation. May hold more than one id. + triggering_event_ids: Vec, + outcome: WakeOutcome, + }, + /// `ACP_SESSION_STARTED`. Same event set as the `WakeDecision` sharing + /// this record's `turn_id`, by construction (same moved `FlushBatch`). + AcpSession { + triggering_event_ids: Vec, + }, +} + +/// Shared envelope fields, common to all nine boundaries. +/// +/// No `content`, `prompt`, `handoff_id`, or `source_event_id` field exists +/// on this type — the first two because a hook site has nothing to pass if +/// it wanted to leak a raw message body; the latter two because neither is +/// derivable from transport data without either fabricating a value or +/// conflating two distinct concepts (see the causality-review addenda). +#[derive(Debug, Default, Clone, Serialize)] +pub struct AuditFields { + /// The single physical Nostr event id (hex) this record concerns. + /// Populated at `TransportPublished`, `TransportAccepted` (bare acked + /// id), `EVENT_RECEIVED`, `FILTER_DECISION`, `QUEUE_DECISION`, + /// `RESPONSE_PUBLISHED`, `RECIPIENT_EVIDENCE_OBSERVED` — all + /// structurally one-event-at-a-time boundaries. Always `None` at + /// `WAKE_DECISION` / `ACP_SESSION_STARTED`, which are batch-cardinality + /// (see `EventDetail::WakeDecision` / `EventDetail::AcpSession`). + pub event_id: Option, + /// NIP-10 direct-reply parent event id. + pub direct_parent_event_id: Option, + /// NIP-10 thread-root event id. + pub thread_root_event_id: Option, + /// `thread_root_event_id`, falling back to the event's own id when it + /// has no parent (it IS the root). Thread-level grouping only — NOT a + /// unique per-handoff identity. + pub correlation_id: Option, + /// Verbatim pass-through of a `workflow_id` tag, ONLY when the Hive + /// protocol layer above buzz-acp actually set one. Never interpreted + /// here, never derived, never defaulted. + pub workflow_id: Option, + pub sender_pubkey: Option, + /// Provenance is boundary-dependent — this field is NOT one consistent + /// kind of evidence across `event_type`s, and must always be read + /// together with it: + /// - At `FILTER_DECISION` (`filter.rs`), the value is the authoritative + /// local agent pubkey supplied to `match_event` as `agent_pubkey_hex`: + /// this process genuinely IS the recipient there. + /// - At `EVENT_RECEIVED` (`relay.rs`), the value is the authoritative + /// local recipient process pubkey already in scope as + /// `agent_pubkey_hex` — same reasoning, different file. + /// - At `QUEUE_DECISION` (`queue.rs`), and any other boundary populated + /// via [`first_mentioned_pubkey`], the value is merely the first + /// signed `p` tag found on the event — mention/target evidence only. + /// A `p` tag records who the event's author addressed; it is NOT a + /// confirmation that any particular process received or claimed the + /// role of recipient. + /// Consumers MUST NOT interpret `target_pubkey` generically as proof of + /// process-recipient identity — its evidentiary weight depends entirely + /// on which boundary (`event_type`) produced the record. + pub target_pubkey: Option, + pub channel_id: Option, + pub agent_session_id: Option, + /// Dispatch/turn identity spanning exactly one `WAKE_DECISION` and its + /// paired `ACP_SESSION_STARTED`. `Some(_)` ONLY at `WAKE_DECISION` + /// (`Claimed` outcome — `PoolExhausted` never reaches the point a + /// turn_id is assigned) and `ACP_SESSION_STARTED`. Sourced verbatim + /// from `dispatch_pending`'s existing `turn_id` local, not invented. + pub turn_id: Option, + /// `EventQueue`'s per-channel dispatch-retry counter. `Some(_)` ONLY at + /// `QUEUE_DECISION` (`queue.rs::push`, `&self.retry_counts`) and + /// `WAKE_DECISION` (`lib.rs::dispatch_pending`, via a new + /// `EventQueue::retry_count_for` accessor). This is NOT a count of + /// business-level (Nostr republish) retries, which buzz-acp cannot + /// observe — no default-0 fallback exists anywhere in this module. + pub attempt: Option, +} + +/// Thread-level correlation key for an event: its NIP-10 thread-root id, or +/// its own id when it has no parent (it IS the root). Reuses +/// [`crate::queue::parse_thread_tags`] rather than re-implementing NIP-10 +/// parsing. Metadata only — reads `.id` / `.tags`, never `.content`. +pub fn correlation_id_for(event: &nostr::Event) -> String { + crate::queue::parse_thread_tags(event) + .root_event_id + .unwrap_or_else(|| event.id.to_hex()) +} + +/// First mentioned pubkey (`p` tag) on an event — used as `target_pubkey` +/// where no more authoritative source (e.g. this process's own identity via +/// an `agent_pubkey_hex` parameter already in scope) is available. Metadata +/// only. +pub fn first_mentioned_pubkey(event: &nostr::Event) -> Option { + event.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.len() >= 2 && parts[0] == "p").then(|| parts[1].clone()) + }) +} + +/// Best-effort, verbatim pass-through of a `workflow_id` tag. buzz-acp does +/// not define, require, or interpret this tag — Hive workflow semantics are +/// out of scope here; this only forwards what's already signed on the wire. +pub fn workflow_id_for(event: &nostr::Event) -> Option { + event.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.len() >= 2 && parts[0] == "workflow_id").then(|| parts[1].clone()) + }) +} + +/// Handle to the background writer. The producer side only ever holds a +/// cheap, cloneable `SyncSender` — the `File` itself belongs exclusively to +/// the writer thread. +struct AuditSink { + tx: SyncSender, + /// Records dropped due to a full or disconnected channel. + dropped: AtomicU64, + seq: AtomicU64, +} + +static SINK: OnceLock> = OnceLock::new(); + +/// Initialize the global audit sink from the environment. Idempotent — call +/// once at process startup; later calls are no-ops against the +/// already-initialized [`OnceLock`]. +/// +/// Fail-open: any error resolving `HOME`, creating the audit directory, +/// opening the file, or spawning the writer thread disables auditing for +/// this process. Never panics, never blocks startup. +pub fn init() { + SINK.get_or_init(build_sink); +} + +fn enabled_from_env() -> bool { + std::env::var(AUDIT_ENABLE_ENV) + .map(|v| matches!(v.trim(), "1" | "true" | "TRUE" | "True")) + .unwrap_or(false) +} + +/// `$HOME/.buzz/audit` — the default sink directory. `None` when `HOME` is +/// unset or empty, in which case the location would no longer be +/// deterministic and the caller fails open (disables auditing) rather than +/// guessing a CWD-relative fallback. +fn default_audit_dir() -> Option { + let home = std::env::var("HOME").ok().filter(|h| !h.is_empty())?; + Some(PathBuf::from(home).join(".buzz").join("audit")) +} + +/// Resolve the JSONL sink path. `BUZZ_ACP_AUDIT_PATH`, when set to a +/// non-empty value, is used verbatim — it is operator-controlled and may +/// point anywhere the operator chooses, local or otherwise; this function +/// makes no claim about it. Only the no-override default is guaranteed +/// local, and only after successfully creating `$HOME/.buzz/audit/` (itself +/// fail-open on failure). +fn resolve_audit_path() -> Option { + if let Ok(p) = std::env::var(AUDIT_PATH_ENV) { + if !p.is_empty() { + return Some(PathBuf::from(p)); + } + } + let dir = default_audit_dir()?; + if let Err(error) = std::fs::create_dir_all(&dir) { + tracing::warn!( + target: "audit", + path = %dir.display(), + %error, + "failed to create default audit directory; auditing disabled for this process" + ); + return None; + } + Some(dir.join(DEFAULT_AUDIT_FILENAME)) +} + +fn build_sink() -> Option { + if !enabled_from_env() { + return None; + } + let path = resolve_audit_path()?; + let file = match OpenOptions::new().create(true).append(true).open(&path) { + Ok(file) => file, + Err(error) => { + tracing::warn!( + target: "audit", + path = %path.display(), + %error, + "failed to open audit sink; auditing disabled for this process" + ); + return None; + } + }; + + let (tx, rx) = sync_channel::(CHANNEL_CAPACITY); + let spawned = std::thread::Builder::new() + .name("buzz-acp-audit-writer".to_string()) + .spawn(move || { + // Single-threaded by construction — a plain local counter is + // enough, no atomics needed on this side. + let mut file = file; + let mut write_errors: u64 = 0; + while let Ok(line) = rx.recv() { + if let Err(error) = writeln!(file, "{line}") { + write_errors += 1; + if write_errors % WARN_LOG_MODULUS == 1 { + tracing::warn!( + target: "audit", + %error, + count = write_errors, + "audit write failed (rate-limited log)" + ); + } + // Never panic on a transient I/O error — keep draining + // so a temporarily-unwritable sink doesn't wedge the + // channel; the caller already moved on either way. + } + } + }); + + match spawned { + Ok(_handle) => { + // Detached deliberately: the writer runs for the process + // lifetime, fed only by `SINK`'s `SyncSender`. No explicit + // shutdown/flush is required — fail-open allows dropped + // records, and each successful write is its own syscall (no + // extra application-level buffering to flush). + tracing::info!(target: "audit", path = %path.display(), "buzz-acp audit sink enabled"); + Some(AuditSink { + tx, + dropped: AtomicU64::new(0), + seq: AtomicU64::new(0), + }) + } + Err(error) => { + tracing::warn!( + target: "audit", + %error, + "failed to spawn audit writer thread; auditing disabled for this process" + ); + None + } + } +} + +/// True when the audit sink is enabled for this process. Cheap — callers on +/// hot paths may use this to skip building [`AuditFields`] entirely when +/// disabled, though [`record`] itself is already a no-op in that case. +pub fn is_enabled() -> bool { + SINK.get().is_some_and(Option::is_some) +} + +/// One JSONL line. Assembled only inside [`record`] — callers never +/// construct this directly, so `schema_version` / `ts` / `seq` can never be +/// forgotten or spoofed by a call site. +#[derive(Serialize)] +struct AuditRecord<'a> { + schema_version: u32, + ts: String, + seq: u64, + event_type: EventType, + #[serde(flatten)] + fields: &'a AuditFields, + detail: &'a EventDetail, +} + +/// Record one causal-boundary event. +/// +/// No-op when auditing is disabled or uninitialized. Never awaits, never +/// performs filesystem I/O itself, never panics: this function only builds +/// a JSON string in memory and hands it to the writer thread via +/// [`SyncSender::try_send`]. A serialization failure, a full channel, or a +/// disconnected writer are all handled the same way — the record is +/// dropped, counted, and (rate-limited) logged; the caller is never blocked +/// and never sees an error. +pub fn record(event_type: EventType, fields: AuditFields, detail: EventDetail) { + let Some(sink) = SINK.get().and_then(|s| s.as_ref()) else { + return; + }; + let seq = sink.seq.fetch_add(1, Ordering::Relaxed); + let record = AuditRecord { + schema_version: SCHEMA_VERSION, + ts: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Nanos, true), + seq, + event_type, + fields: &fields, + detail: &detail, + }; + let line = match serde_json::to_string(&record) { + Ok(line) => line, + Err(error) => { + tracing::warn!(target: "audit", %error, "failed to serialize audit record; dropping"); + return; + } + }; + match sink.tx.try_send(line) { + Ok(()) => {} + Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => { + let n = sink.dropped.fetch_add(1, Ordering::Relaxed) + 1; + if n % WARN_LOG_MODULUS == 1 { + tracing::warn!( + target: "audit", + dropped = n, + "audit channel backpressure — dropping records (rate-limited log)" + ); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + /// Env-var-touching tests must run serially — env vars are process- + /// global. Mirrors `lib.rs`'s `build_mcp_servers_tests::ENV_LOCK` idiom + /// exactly. No other file in this crate reads `BUZZ_ACP_AUDIT`, + /// `BUZZ_ACP_AUDIT_PATH`, or `HOME` (confirmed by grep before writing + /// these tests), so this lock only needs to protect tests within this + /// module from each other. + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + /// RAII guard: sets (or unsets) an env var for the duration of a test + /// and restores its exact prior value on drop — `Some(v)` restores `v`, + /// `None` removes the var entirely. Runs on the panic/unwind path too + /// (a failed `assert!` still drops locals), so a test failure can never + /// leak mutated env state to a later test in the same binary. Must be + /// held only while `ENV_LOCK` is also held. + struct EnvVarGuard { + key: &'static str, + previous: Option, + } + + impl EnvVarGuard { + fn set(key: &'static str, value: &str) -> Self { + let previous = std::env::var(key).ok(); + std::env::set_var(key, value); + Self { key, previous } + } + + fn unset(key: &'static str) -> Self { + let previous = std::env::var(key).ok(); + std::env::remove_var(key); + Self { key, previous } + } + } + + impl Drop for EnvVarGuard { + fn drop(&mut self) { + match &self.previous { + Some(v) => std::env::set_var(self.key, v), + None => std::env::remove_var(self.key), + } + } + } + + /// No test in this module ever calls [`init`]. `SINK` therefore stays + /// uninitialized for the entire test binary (nothing else in the crate + /// calls `audit::init()` either — it's only reached from + /// `lib.rs::run()`'s production entry point). This is deliberate: `SINK` + /// is a `OnceLock` and can only be populated once per process, so tests + /// must not race to be "the one" that initializes it. Every test below + /// instead calls the pure, stateless functions `build_sink()` / + /// `resolve_audit_path()` / `enabled_from_env()` directly, or exercises + /// `record()`/`is_enabled()` only in the (safe, permanent-for-this- + /// binary) uninitialized state. + + // ---- T1: schema / envelope shape ---------------------------------- + + const EXPECTED_ENVELOPE_KEYS: &[&str] = &[ + "schema_version", + "ts", + "seq", + "event_type", + "event_id", + "direct_parent_event_id", + "thread_root_event_id", + "correlation_id", + "workflow_id", + "sender_pubkey", + "target_pubkey", + "channel_id", + "agent_session_id", + "turn_id", + "attempt", + "detail", + ]; + + const OBSOLETE_FIELD_NAMES: &[&str] = &[ + "handoff_id", + "source_event_id", + "runtime_id", + "pid", + "buzz_acp_version", + "stage", + "latency_ms", + ]; + + #[test] + fn schema_version_is_one() { + assert_eq!(SCHEMA_VERSION, 1); + } + + #[test] + fn envelope_serializes_expected_keys_and_omits_obsolete_fields() { + let fields = AuditFields { + event_id: Some("deadbeef".into()), + ..Default::default() + }; + let record = AuditRecord { + schema_version: SCHEMA_VERSION, + ts: "2026-01-01T00:00:00.000000000Z".into(), + seq: 0, + event_type: EventType::EventReceived, + fields: &fields, + detail: &EventDetail::Empty, + }; + let value = serde_json::to_value(&record).expect("serialize"); + let obj = value.as_object().expect("record must be a JSON object"); + + let mut keys: Vec<&str> = obj.keys().map(String::as_str).collect(); + keys.sort_unstable(); + let mut expected: Vec<&str> = EXPECTED_ENVELOPE_KEYS.to_vec(); + expected.sort_unstable(); + assert_eq!(keys, expected, "envelope key set changed unexpectedly"); + + for obsolete in OBSOLETE_FIELD_NAMES { + assert!( + !obj.contains_key(*obsolete), + "obsolete field {obsolete:?} must never reappear in the schema" + ); + } + + assert_eq!(value["schema_version"], 1); + assert_eq!(value["event_type"], "EVENT_RECEIVED"); + assert_eq!(value["event_id"], "deadbeef"); + } + + #[test] + fn event_type_variants_serialize_screaming_snake_case() { + let cases = [ + (EventType::TransportPublished, "TRANSPORT_PUBLISHED"), + (EventType::TransportAccepted, "TRANSPORT_ACCEPTED"), + (EventType::EventReceived, "EVENT_RECEIVED"), + (EventType::FilterDecision, "FILTER_DECISION"), + (EventType::QueueDecision, "QUEUE_DECISION"), + (EventType::WakeDecision, "WAKE_DECISION"), + (EventType::AcpSessionStarted, "ACP_SESSION_STARTED"), + (EventType::ResponsePublished, "RESPONSE_PUBLISHED"), + ( + EventType::RecipientEvidenceObserved, + "RECIPIENT_EVIDENCE_OBSERVED", + ), + ]; + for (variant, expected) in cases { + let value = serde_json::to_value(variant).expect("serialize"); + assert_eq!(value, expected); + } + } + + #[test] + fn filter_decision_detail_shape() { + let detail = EventDetail::FilterDecision { + rule_index: Some(2), + fail_closed: true, + }; + let value = serde_json::to_value(&detail).expect("serialize"); + assert_eq!(value["kind"], "FILTER_DECISION"); + assert_eq!(value["rule_index"], 2); + assert_eq!(value["fail_closed"], true); + } + + #[test] + fn queue_decision_detail_shape() { + let detail = EventDetail::QueueDecision { + outcome: QueueOutcome::CapEvicted, + }; + let value = serde_json::to_value(&detail).expect("serialize"); + assert_eq!(value["kind"], "QUEUE_DECISION"); + assert_eq!(value["outcome"], "CAP_EVICTED"); + } + + #[test] + fn relay_ack_detail_has_no_message_field() { + let detail = EventDetail::RelayAck { accepted: false }; + let value = serde_json::to_value(&detail).expect("serialize"); + let obj = value.as_object().expect("must be an object"); + assert_eq!(value["kind"], "RELAY_ACK"); + assert_eq!(value["accepted"], false); + assert_eq!( + obj.len(), + 2, + "RelayAck must carry only kind+accepted, never a relay message string" + ); + assert!(!obj.contains_key("message")); + } + + #[test] + fn wake_decision_detail_shape() { + let detail = EventDetail::WakeDecision { + triggering_event_ids: vec!["ev1".into(), "ev2".into()], + outcome: WakeOutcome::PoolExhausted, + }; + let value = serde_json::to_value(&detail).expect("serialize"); + assert_eq!(value["kind"], "WAKE_DECISION"); + assert_eq!(value["outcome"], "POOL_EXHAUSTED"); + assert_eq!(value["triggering_event_ids"], serde_json::json!(["ev1", "ev2"])); + } + + #[test] + fn acp_session_detail_shape() { + let detail = EventDetail::AcpSession { + triggering_event_ids: vec!["ev1".into()], + }; + let value = serde_json::to_value(&detail).expect("serialize"); + assert_eq!(value["kind"], "ACP_SESSION"); + assert_eq!(value["triggering_event_ids"], serde_json::json!(["ev1"])); + } + + // ---- T2: metadata / redaction-by-construction ---------------------- + + #[test] + fn sentinel_populated_fields_serialize_as_plain_strings_no_extra_keys() { + // Synthetic, obviously-fake sentinel values only — never real + // secrets, keys, or content. + let fields = AuditFields { + event_id: Some("SENTINEL_EVENT_ID".into()), + direct_parent_event_id: Some("SENTINEL_PARENT_ID".into()), + thread_root_event_id: Some("SENTINEL_ROOT_ID".into()), + correlation_id: Some("SENTINEL_CORRELATION_ID".into()), + workflow_id: Some("SENTINEL_WORKFLOW_ID".into()), + sender_pubkey: Some("SENTINEL_SENDER_PUBKEY".into()), + target_pubkey: Some("SENTINEL_TARGET_PUBKEY".into()), + channel_id: Some("SENTINEL_CHANNEL_ID".into()), + agent_session_id: Some("SENTINEL_SESSION_ID".into()), + turn_id: Some("SENTINEL_TURN_ID".into()), + attempt: Some(9), + }; + let record = AuditRecord { + schema_version: SCHEMA_VERSION, + ts: "2026-01-01T00:00:00.000000000Z".into(), + seq: 42, + event_type: EventType::FilterDecision, + fields: &fields, + detail: &EventDetail::FilterDecision { + rule_index: Some(0), + fail_closed: false, + }, + }; + let value = serde_json::to_value(&record).expect("serialize"); + let obj = value.as_object().expect("must be an object"); + + let mut keys: Vec<&str> = obj.keys().map(String::as_str).collect(); + keys.sort_unstable(); + let mut expected: Vec<&str> = EXPECTED_ENVELOPE_KEYS.to_vec(); + expected.sort_unstable(); + assert_eq!( + keys, expected, + "fully-populated record must not gain or lose keys" + ); + + // Every sentinel must round-trip unchanged as a plain string — no + // hidden expansion, parsing, or restructuring of caller-supplied + // metadata values. + assert_eq!(value["event_id"], "SENTINEL_EVENT_ID"); + assert_eq!(value["sender_pubkey"], "SENTINEL_SENDER_PUBKEY"); + assert_eq!(value["target_pubkey"], "SENTINEL_TARGET_PUBKEY"); + assert_eq!(value["turn_id"], "SENTINEL_TURN_ID"); + assert_eq!(value["attempt"], 9); + + // No field anywhere in this type is named/shaped for raw content — + // this is a structural guarantee, re-asserted here as a regression + // check on the field list itself. + for forbidden in ["content", "prompt", "body", "message", "text"] { + assert!( + !obj.contains_key(forbidden), + "AuditFields must never gain a field named {forbidden:?}" + ); + } + } + + // ---- T7: audit OFF --------------------------------------------------- + + #[test] + fn enabled_from_env_true_variants() { + let _guard = ENV_LOCK.lock().unwrap(); + for v in ["1", "true", "True", "TRUE"] { + let _env = EnvVarGuard::set(AUDIT_ENABLE_ENV, v); + assert!(enabled_from_env(), "expected enabled for {v:?}"); + } + } + + #[test] + fn enabled_from_env_false_for_unset_and_other_values() { + let _guard = ENV_LOCK.lock().unwrap(); + { + let _env = EnvVarGuard::unset(AUDIT_ENABLE_ENV); + assert!(!enabled_from_env(), "unset must be disabled"); + } + for v in ["0", "false", "yes", "enabled", ""] { + let _env = EnvVarGuard::set(AUDIT_ENABLE_ENV, v); + assert!(!enabled_from_env(), "expected disabled for {v:?}"); + } + } + + #[test] + fn record_is_noop_and_does_not_panic_without_init() { + // Deliberately never calls audit::init() anywhere in this module — + // see the module-level doc comment above. + let fields = AuditFields { + event_id: Some("noop-test-event".into()), + ..Default::default() + }; + record(EventType::EventReceived, fields, EventDetail::Empty); + // No panic = pass. There is no observable side channel from outside + // this module without touching the real filesystem sink, which this + // test deliberately avoids. + } + + #[test] + fn is_enabled_false_without_init() { + assert!(!is_enabled()); + } + + // ---- T8: fail-open --------------------------------------------------- + + #[test] + fn build_sink_returns_none_when_disabled() { + let _guard = ENV_LOCK.lock().unwrap(); + let _env = EnvVarGuard::unset(AUDIT_ENABLE_ENV); + assert!(build_sink().is_none()); + } + + #[test] + fn resolve_audit_path_uses_explicit_override_verbatim() { + let _guard = ENV_LOCK.lock().unwrap(); + let _env = EnvVarGuard::set(AUDIT_PATH_ENV, "/tmp/some-explicit-audit-path.jsonl"); + let resolved = resolve_audit_path(); + assert_eq!( + resolved, + Some(PathBuf::from("/tmp/some-explicit-audit-path.jsonl")) + ); + } + + /// Builds a fresh, unique temp directory containing a plain FILE named + /// `.buzz` (not a directory). Any attempt to `create_dir_all(".buzz/ + /// audit")` underneath it must fail deterministically on every OS — no + /// chmod/permission trick required. + fn make_home_with_blocked_dot_buzz() -> PathBuf { + let base = std::env::temp_dir().join(format!( + "buzz-acp-audit-test-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&base).expect("create temp base dir"); + std::fs::write(base.join(".buzz"), b"not a directory").expect("write blocking file"); + base + } + + #[test] + fn resolve_audit_path_fails_open_when_home_dot_buzz_is_a_regular_file() { + let _guard = ENV_LOCK.lock().unwrap(); + let base = make_home_with_blocked_dot_buzz(); + + let _path_env = EnvVarGuard::unset(AUDIT_PATH_ENV); + let _home_env = EnvVarGuard::set("HOME", base.to_str().expect("temp path is valid UTF-8")); + let resolved = resolve_audit_path(); + + let _ = std::fs::remove_dir_all(&base); + assert!( + resolved.is_none(), + "resolve_audit_path must fail open (None) when $HOME/.buzz is a regular file" + ); + } + + #[test] + fn build_sink_fails_open_when_default_dir_blocked() { + let _guard = ENV_LOCK.lock().unwrap(); + let base = make_home_with_blocked_dot_buzz(); + + let _enable_env = EnvVarGuard::set(AUDIT_ENABLE_ENV, "1"); + let _path_env = EnvVarGuard::unset(AUDIT_PATH_ENV); + let _home_env = EnvVarGuard::set("HOME", base.to_str().expect("temp path is valid UTF-8")); + let sink = build_sink(); + + let _ = std::fs::remove_dir_all(&base); + assert!( + sink.is_none(), + "build_sink must fail open (None), never panic, when the default \ + audit directory cannot be created" + ); + } +} diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd3..5b31ea53d75 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -11,6 +11,8 @@ use std::time::Duration; use tracing::{error, warn}; +use crate::audit; + /// Errors that can occur during filter expression evaluation. #[derive(Debug, thiserror::Error)] pub enum FilterError { @@ -373,6 +375,29 @@ pub async fn match_event( ) -> Option { let filter_ctx = FilterContext::from_event(event, channel_id); + // Phase B B5 FILTER_DECISION — computed once per call (not per rule) to + // avoid redundant NIP-10 parsing across the loop below. `None` when + // auditing is disabled, so no metadata is built at all in that case. + let audit_fields = audit::is_enabled().then(|| { + let tags = crate::queue::parse_thread_tags(event); + let event_id = event.id.to_hex(); + let correlation_id = tags + .root_event_id + .clone() + .unwrap_or_else(|| event_id.clone()); + audit::AuditFields { + event_id: Some(event_id), + direct_parent_event_id: tags.parent_event_id, + thread_root_event_id: tags.root_event_id, + correlation_id: Some(correlation_id), + workflow_id: audit::workflow_id_for(event), + sender_pubkey: Some(event.pubkey.to_hex()), + target_pubkey: Some(agent_pubkey_hex.to_string()), + channel_id: Some(channel_id.to_string()), + ..Default::default() + } + }); + for (index, rule) in rules.iter().enumerate() { // 1. Channel scope check. if !rule.channels.matches(&channel_id) { @@ -411,6 +436,16 @@ pub async fn match_event( failing closed (no match for any rule)" ); // Fail-closed: disabled rule → no match for this event. + if let Some(fields) = &audit_fields { + audit::record( + audit::EventType::FilterDecision, + fields.clone(), + audit::EventDetail::FilterDecision { + rule_index: Some(index), + fail_closed: true, + }, + ); + } return None; } @@ -432,6 +467,16 @@ pub async fn match_event( "filter expression timed out; failing closed (no match for any rule)" ); // Fail-closed: timeout → no match, not next rule. + if let Some(fields) = &audit_fields { + audit::record( + audit::EventType::FilterDecision, + fields.clone(), + audit::EventDetail::FilterDecision { + rule_index: Some(index), + fail_closed: true, + }, + ); + } return None; } Err(e) => { @@ -442,6 +487,16 @@ pub async fn match_event( "filter expression error; failing closed (no match for any rule)" ); // Fail-closed: any error → no match, not next rule. + if let Some(fields) = &audit_fields { + audit::record( + audit::EventType::FilterDecision, + fields.clone(), + audit::EventDetail::FilterDecision { + rule_index: Some(index), + fail_closed: true, + }, + ); + } return None; } } @@ -450,12 +505,34 @@ pub async fn match_event( // All checks passed — this rule wins. let prompt_tag = rule.prompt_tag.clone().unwrap_or_else(|| rule.name.clone()); + if let Some(fields) = &audit_fields { + audit::record( + audit::EventType::FilterDecision, + fields.clone(), + audit::EventDetail::FilterDecision { + rule_index: Some(index), + fail_closed: false, + }, + ); + } return Some(MatchedRule { rule_index: index, prompt_tag, }); } + // No rule matched (and no fail-closed return above fired) — a normal, + // non-error non-match. + if let Some(fields) = &audit_fields { + audit::record( + audit::EventType::FilterDecision, + fields.clone(), + audit::EventDetail::FilterDecision { + rule_index: None, + fail_closed: false, + }, + ); + } None } @@ -784,4 +861,44 @@ mod tests { let result = match_event(&event, channel_id, &rules, "").await; assert!(result.is_none(), "disabled rule must return None"); } + + // ---- Phase B: match_event rule_index attribution (T4, matched/no-match + // only — fail-closed audit-payload specifics deferred; see the + // Phase 5A/5B test plan) ---------------------------------------------- + + #[tokio::test] + async fn match_event_skips_out_of_scope_rule_and_matches_next() { + // Not already covered by test_match_event_kind_filter (which skips + // via the kind guard) — this exercises the channel-scope `continue` + // branch specifically reaching a non-zero rule_index. + let event = make_event(9, "hello"); + let channel_id = any_channel(); + let other_channel = Uuid::new_v4(); + + let rules = vec![ + make_rule( + "out-of-scope", + ChannelScope::List(vec![other_channel.to_string()]), + vec![], + false, + None, + Some("should-not-win"), + ), + make_rule( + "in-scope", + ChannelScope::All("all".into()), + vec![], + false, + None, + Some("should-win"), + ), + ]; + + let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + assert_eq!( + matched.rule_index, 1, + "channel-scope skip (not kind-filter) must reach the second rule" + ); + assert_eq!(matched.prompt_tag, "should-win"); + } } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 1352b31cad8..9f543c7b8d2 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1,6 +1,7 @@ #![deny(unsafe_code)] mod acp; +mod audit; mod config; mod engram_fetch; mod filter; @@ -1926,6 +1927,12 @@ async fn tokio_main() -> Result<()> { .compact() .init(); + // Phase B causal audit sink — global, opt-in, default OFF via + // `BUZZ_ACP_AUDIT`. Must run after the tracing subscriber above (so the + // sink's own enable/fail-open logging is captured) and before any relay + // activity begins. A no-op when the env var is unset — see audit.rs. + audit::init(); + let mut config = Config::from_cli().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?; // ── Setup-mode early branch ─────────────────────────────────────────────── @@ -3723,11 +3730,42 @@ fn dispatch_pending( .map(|event| queue::parse_thread_tags(&event.event)) .unwrap_or_default(); let affinity_hit = pool.has_session_for(channel_id); + // Phase B B7 WAKE_DECISION — `triggering_event_ids` is the same + // computation `pool.rs::run_prompt_task` performs on this same + // `batch` once it's moved there (see B8); computed here too since + // `batch` is consumed below in either branch of the claim match. + // `event_id`/`direct_parent_event_id`/`thread_root_event_id`/ + // `correlation_id`/`workflow_id`/`sender_pubkey`/`target_pubkey` + // stay `None` in the shared envelope: a batch can hold more than + // one physical event, so no single one of those per-event fields + // can honestly stand in for the whole batch — see + // `EventDetail::WakeDecision` for the full, honest set instead. + // `attempt` also stays `None` here: `EventQueue` has no public + // accessor for its per-channel retry counter today, and adding one + // is out of scope for this lib.rs-only pass. + let triggering_event_ids: Vec = if audit::is_enabled() { + batch.events.iter().map(|be| be.event.id.to_hex()).collect() + } else { + Vec::new() + }; let mut agent = match pool.try_claim(Some(channel_id)) { Some(a) => a, None => { let pending = queue.pending_channels(); tracing::debug!(pending_channels = pending, "pool_exhausted"); + if audit::is_enabled() { + audit::record( + audit::EventType::WakeDecision, + audit::AuditFields { + channel_id: Some(channel_id.to_string()), + ..Default::default() + }, + audit::EventDetail::WakeDecision { + triggering_event_ids, + outcome: audit::WakeOutcome::PoolExhausted, + }, + ); + } queue.requeue_preserve_timestamps(batch); queue.mark_complete(channel_id); break; @@ -3761,6 +3799,23 @@ fn dispatch_pending( // context fetching). Pass None for prompt_text; batch carries the data. let (control_tx, control_rx) = tokio::sync::oneshot::channel::(); let turn_id = Uuid::new_v4().to_string(); + // Phase B B7 WAKE_DECISION (Claimed) — `turn_id` is the same value + // threaded into `run_prompt_task` below via `task_turn_id`, giving a + // deterministic B7→B8 pairing (see pool.rs::run_prompt_task). + if audit::is_enabled() { + audit::record( + audit::EventType::WakeDecision, + audit::AuditFields { + channel_id: Some(channel_id.to_string()), + turn_id: Some(turn_id.clone()), + ..Default::default() + }, + audit::EventDetail::WakeDecision { + triggering_event_ids, + outcome: audit::WakeOutcome::Claimed, + }, + ); + } let task_turn_id = turn_id.clone(); let abort_handle = pool.join_set.spawn(async move { diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index f3916b5fdab..982f7cdab11 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -34,6 +34,7 @@ use crate::acp::{ model_in_catalog, resolve_model_switch_method, AcpClient, AcpError, EnvVar, McpServer, ModelSwitchMethod, StopReason, SystemPromptTransport, }; +use crate::audit; use crate::config::{compose_session_title, DedupMode, PermissionMode}; use crate::observer; use crate::queue::{ @@ -1802,7 +1803,7 @@ pub async fn run_prompt_task( PromptSource::Channel(_) => "channel", PromptSource::Heartbeat => "heartbeat", }, - "triggeringEventIds": triggering_event_ids, + "triggeringEventIds": triggering_event_ids.clone(), }), ); @@ -2005,6 +2006,36 @@ pub async fn run_prompt_task( target: "pool::session", "created session {sid} for channel {cid}" ); + // Phase B B8 ACP_SESSION_STARTED — emitted only on + // this success path, using the exact B7 turn_id + // (the `turn_id` parameter, unchanged — no new one + // generated) and the same `triggering_event_ids` + // already computed once at function entry from + // this same `batch` (explicitly cloned at its + // earlier use above so this canonical value + // survives unmoved to here), giving a + // deterministic B7→B8 pairing (equal turn_id, + // equal event set). Batch-cardinality boundary: + // per-event fields stay None — no single event + // from the batch is picked to stand in for the + // whole batch (see EventDetail::AcpSession for + // the honest full set). `attempt` stays None: no + // authoritative source is directly available + // here, and none is being added to obtain one. + if audit::is_enabled() { + audit::record( + audit::EventType::AcpSessionStarted, + audit::AuditFields { + channel_id: Some(cid.to_string()), + agent_session_id: Some(sid.clone()), + turn_id: Some(turn_id.clone()), + ..Default::default() + }, + audit::EventDetail::AcpSession { + triggering_event_ids: triggering_event_ids.clone(), + }, + ); + } agent.state.sessions.insert(*cid, sid.clone()); agent .state diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index b50f926d8b7..2714fffce1f 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -18,6 +18,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::time::{Duration, Instant}; use uuid::Uuid; +use crate::audit; use crate::config::DedupMode; /// Maximum events queued per channel before oldest events are dropped. @@ -227,7 +228,27 @@ impl EventQueue { /// silently discarded (debug-logged). /// /// Returns `true` if the event was accepted, `false` if dropped. + /// + /// Phase B B6 QUEUE_DECISION: emits one record per actual queue-state + /// transition — `DroppedInFlight`, `CapEvicted` (for the evicted event, + /// not the incoming one), and/or `Accepted`. A single call can + /// legitimately emit both `CapEvicted` and `Accepted`. `attempt` is + /// read from `self.retry_counts` once, up front, before `self.queues` + /// is mutably borrowed — the two borrows would otherwise conflict. pub fn push(&mut self, event: QueuedEvent) -> bool { + let audit_enabled = audit::is_enabled(); + // `retry_counts` is keyed by channel_id, so this single read is + // equally authoritative for the incoming event and, on cap + // eviction, the evicted event below: both necessarily share the + // same channel_id, since `EventQueue` is strictly per-channel. + // `None` means "no retry recorded for this channel" — a real fact + // from the map, never fabricated as 0. + let attempt: Option = if audit_enabled { + self.retry_counts.get(&event.channel_id).copied() + } else { + None + }; + if matches!(self.dedup_mode, DedupMode::Drop) && self.in_flight_channels.contains(&event.channel_id) { @@ -235,19 +256,54 @@ impl EventQueue { channel_id = %event.channel_id, "dropping event for in-flight channel (drop mode)" ); + if audit_enabled { + audit::record( + audit::EventType::QueueDecision, + build_queue_audit_fields(&event, attempt), + audit::EventDetail::QueueDecision { + outcome: audit::QueueOutcome::DroppedInFlight, + }, + ); + } return false; } + + // Precompute Accepted-outcome fields now, while `event` is still + // owned here — `push_back` below consumes it, and the Accepted + // record must only be emitted once the enqueue has actually + // happened (see below). + let accepted_fields = audit_enabled.then(|| build_queue_audit_fields(&event, attempt)); + let queue = self.queues.entry(event.channel_id).or_default(); // Enforce per-channel depth cap: drop oldest to make room. if queue.len() >= MAX_PENDING_PER_CHANNEL { - queue.pop_front(); - tracing::warn!( - channel_id = %event.channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "queue depth cap reached — dropped oldest event" - ); + if let Some(evicted) = queue.pop_front() { + tracing::warn!( + channel_id = %event.channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "queue depth cap reached — dropped oldest event" + ); + if audit_enabled { + audit::record( + audit::EventType::QueueDecision, + build_queue_audit_fields(&evicted, attempt), + audit::EventDetail::QueueDecision { + outcome: audit::QueueOutcome::CapEvicted, + }, + ); + } + } } queue.push_back(event); + if let Some(fields) = accepted_fields { + audit::record( + audit::EventType::QueueDecision, + fields, + audit::EventDetail::QueueDecision { + outcome: audit::QueueOutcome::Accepted, + }, + ); + } true } @@ -907,6 +963,31 @@ pub fn parse_thread_tags(event: &Event) -> ThreadTags { } } +/// Phase B B6 QUEUE_DECISION metadata for one queued event — reused for +/// both the incoming event and, on cap eviction, the evicted event. +/// `attempt` is passed in verbatim from the caller's `retry_counts` read; +/// this function never fabricates or defaults it. +fn build_queue_audit_fields(qe: &QueuedEvent, attempt: Option) -> audit::AuditFields { + let tags = parse_thread_tags(&qe.event); + let event_id = qe.event.id.to_hex(); + let correlation_id = tags + .root_event_id + .clone() + .unwrap_or_else(|| event_id.clone()); + audit::AuditFields { + event_id: Some(event_id), + direct_parent_event_id: tags.parent_event_id, + thread_root_event_id: tags.root_event_id, + correlation_id: Some(correlation_id), + workflow_id: audit::workflow_id_for(&qe.event), + sender_pubkey: Some(qe.event.pubkey.to_hex()), + target_pubkey: audit::first_mentioned_pubkey(&qe.event), + channel_id: Some(qe.channel_id.to_string()), + attempt, + ..Default::default() + } +} + /// Extract a leading slash command from message content. /// /// ACP connectors (claude-agent-acp, codex-acp) detect slash commands by @@ -5402,4 +5483,106 @@ mod tests { "unresolved metadata must not render a Description field; got: {prompt}" ); } + + // ---- Phase B: build_queue_audit_fields (T3) ------------------------- + + #[test] + fn build_queue_audit_fields_preserves_some_attempt() { + let channel_id = Uuid::new_v4(); + let qe = make_queued(channel_id, "hello"); + let fields = build_queue_audit_fields(&qe, Some(3)); + assert_eq!(fields.attempt, Some(3)); + } + + #[test] + fn build_queue_audit_fields_preserves_none_attempt_without_defaulting_to_zero() { + let channel_id = Uuid::new_v4(); + let qe = make_queued(channel_id, "hello"); + let fields = build_queue_audit_fields(&qe, None); + assert_eq!( + fields.attempt, None, + "must stay None — never fabricated as Some(0)" + ); + } + + #[test] + fn build_queue_audit_fields_extracts_correct_event_identity() { + let channel_id = Uuid::new_v4(); + let qe = make_queued(channel_id, "hello"); + let expected_event_id = qe.event.id.to_hex(); + let expected_sender = qe.event.pubkey.to_hex(); + let fields = build_queue_audit_fields(&qe, None); + assert_eq!(fields.event_id, Some(expected_event_id)); + assert_eq!(fields.sender_pubkey, Some(expected_sender)); + assert_eq!(fields.channel_id, Some(channel_id.to_string())); + } + + // ---- Phase B: push() QueueDecision outcomes (T5) --------------------- + + #[test] + fn push_accepted_returns_true_and_enqueues_incoming_event() { + let mut queue = EventQueue::new(DedupMode::Queue); + let channel_id = Uuid::new_v4(); + let qe = make_queued(channel_id, "audit-t5-accepted"); + assert!(queue.push(qe)); + assert_eq!(pending_count(&queue), 1); + let stored = queue.queues.get(&channel_id).expect("channel present"); + assert_eq!(stored.len(), 1); + assert_eq!(stored[0].event.content.to_string(), "audit-t5-accepted"); + } + + #[test] + fn push_dropped_in_flight_returns_false_in_drop_mode_when_channel_busy() { + let mut queue = EventQueue::new(DedupMode::Drop); + let channel_id = Uuid::new_v4(); + queue.in_flight_channels.insert(channel_id); + let qe = make_queued(channel_id, "should-be-dropped"); + assert!(!queue.push(qe)); + assert_eq!(pending_count(&queue), 0); + } + + #[test] + fn push_cap_eviction_removes_oldest_event_not_incoming() { + let mut queue = EventQueue::new(DedupMode::Queue); + let channel_id = Uuid::new_v4(); + + assert!(queue.push(make_queued(channel_id, "oldest"))); + for i in 1..MAX_PENDING_PER_CHANNEL { + assert!(queue.push(make_queued(channel_id, &format!("filler-{i}")))); + } + assert_eq!(pending_count(&queue), MAX_PENDING_PER_CHANNEL); + + assert!(queue.push(make_queued(channel_id, "overflow"))); + + let stored = queue.queues.get(&channel_id).expect("channel present"); + assert_eq!( + stored.len(), + MAX_PENDING_PER_CHANNEL, + "cap must still hold after eviction" + ); + assert!( + stored.iter().all(|e| e.event.content.to_string() != "oldest"), + "the oldest (first-pushed) event must have been evicted" + ); + assert!( + stored + .iter() + .any(|e| e.event.content.to_string() == "overflow"), + "the incoming event must have been accepted, not evicted" + ); + } + + #[test] + fn retry_counts_seeded_value_flows_into_build_queue_audit_fields() { + let mut queue = EventQueue::new(DedupMode::Queue); + let channel_id = Uuid::new_v4(); + queue.set_retry_count_for_test(channel_id, 7); + + let attempt = queue.retry_counts.get(&channel_id).copied(); + assert_eq!(attempt, Some(7)); + + let qe = make_queued(channel_id, "retry-flow"); + let fields = build_queue_audit_fields(&qe, attempt); + assert_eq!(fields.attempt, Some(7)); + } } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 17a818867dd..4a6be66af59 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -129,6 +129,7 @@ use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, Web use tracing::{debug, info, warn}; use uuid::Uuid; +use crate::audit; use crate::config::ChannelFilter; /// Metadata about a channel, populated at discovery time. @@ -1530,6 +1531,40 @@ async fn execute_connected_command( // frame is parked so the post-reconnect drain redelivers it. let is_observer = event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME; if send_publish_event_frame(ws, &event).await { + // Phase B TRANSPORT_PUBLISHED — proves the frame was + // actually written to the socket, not merely queued as an + // intention to publish. This is buzz-acp's own WS publish + // path ONLY: it does NOT represent real agent handoff/chat + // publication, and does NOT cover kind:9 messages sent via + // `buzz messages send` — those go out through buzz-cli's + // independent REST client (see audit::EventType docs). + // Skips observer telemetry frames and typing indicators: + // neither is even buzz-acp's own causal traffic. + if audit::is_enabled() { + let kind = event.kind.as_u16() as u32; + if kind != KIND_AGENT_OBSERVER_FRAME && kind != KIND_TYPING_INDICATOR { + let tags = crate::queue::parse_thread_tags(&event); + let event_id = event.id.to_hex(); + let correlation_id = tags + .root_event_id + .clone() + .unwrap_or_else(|| event_id.clone()); + audit::record( + audit::EventType::TransportPublished, + audit::AuditFields { + event_id: Some(event_id), + direct_parent_event_id: tags.parent_event_id, + thread_root_event_id: tags.root_event_id, + correlation_id: Some(correlation_id), + workflow_id: audit::workflow_id_for(&event), + sender_pubkey: Some(event.pubkey.to_hex()), + target_pubkey: audit::first_mentioned_pubkey(&event), + ..Default::default() + }, + audit::EventDetail::Empty, + ); + } + } if is_observer { state.track_observer_in_flight(event); } @@ -2175,6 +2210,36 @@ async fn handle_ws_message( channel_id, event: *event, }; + // Phase B B4 EVENT_RECEIVED — fires once the + // inbound relay event has been deduped + // (`record_event`) and parsed into `BuzzEvent`, + // before it's forwarded to the harness main + // loop. `target_pubkey` uses this process's own + // confirmed identity (`agent_pubkey_hex`), not a + // tag guess — it IS the recipient here. + if audit::is_enabled() { + let tags = crate::queue::parse_thread_tags(&buzz_event.event); + let event_id = buzz_event.event.id.to_hex(); + let correlation_id = tags + .root_event_id + .clone() + .unwrap_or_else(|| event_id.clone()); + audit::record( + audit::EventType::EventReceived, + audit::AuditFields { + event_id: Some(event_id), + direct_parent_event_id: tags.parent_event_id, + thread_root_event_id: tags.root_event_id, + correlation_id: Some(correlation_id), + workflow_id: audit::workflow_id_for(&buzz_event.event), + sender_pubkey: Some(buzz_event.event.pubkey.to_hex()), + target_pubkey: Some(agent_pubkey_hex.to_string()), + channel_id: Some(channel_id.to_string()), + ..Default::default() + }, + audit::EventDetail::Empty, + ); + } // Warn at 80% capacity. let cap = event_tx.max_capacity(); let used = cap - event_tx.capacity(); @@ -2387,6 +2452,25 @@ async fn handle_ws_message( accepted, message, } => { + // Phase B TRANSPORT_ACCEPTED — the relay's NIP-01 OK + // for event_id. Acknowledges buzz-acp's own WS-published + // traffic ONLY — it does NOT represent acceptance of + // real agent chat messages, which are published through + // buzz-cli's REST path and acknowledged via that HTTP + // response, not this WS OK frame (see audit::EventType + // docs). Deliberately no `message` text in the audit + // record (arbitrary relay-supplied content); `accepted` + // alone is the fact this boundary exists to prove. + if audit::is_enabled() { + audit::record( + audit::EventType::TransportAccepted, + audit::AuditFields { + event_id: Some(event_id.clone()), + ..Default::default() + }, + audit::EventDetail::RelayAck { accepted }, + ); + } if !accepted && message.starts_with("auth") { // AUTH OK with accepted=false means auth was rejected. warn!("mid-session AUTH rejected (event {event_id}): {message} — triggering reconnect");