diff --git a/crates/buzz-acp/src/audit.rs b/crates/buzz-acp/src/audit.rs new file mode 100644 index 00000000000..3a39378805d --- /dev/null +++ b/crates/buzz-acp/src/audit.rs @@ -0,0 +1,480 @@ +//! 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)" + ); + } + } + } +} diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd3..642e132ae48 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 } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 146214197a8..86c8f04413e 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; @@ -1942,6 +1943,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 ─────────────────────────────────────────────── @@ -3737,11 +3744,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; @@ -3775,6 +3813,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 38749577398..11be0775738 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::{ @@ -1777,7 +1778,7 @@ pub async fn run_prompt_task( PromptSource::Channel(_) => "channel", PromptSource::Heartbeat => "heartbeat", }, - "triggeringEventIds": triggering_event_ids, + "triggeringEventIds": triggering_event_ids.clone(), }), ); @@ -1980,6 +1981,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 60866518bad..67a2285acdd 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 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");