buzz/crates/buzz-acp/src/observer.rs

//! In-process observer bus for ACP session activity.
//!
//! This is intentionally process-local infrastructure: it lets the harness
//! collect raw ACP JSON-RPC activity and publish owner-scoped encrypted relay
//! frames without exposing a local HTTP port.

use std::{
    collections::VecDeque,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc, Mutex,
    },
};

use serde::Serialize;
use tokio::sync::broadcast;

#[cfg(test)]
mod packing_tests;
#[cfg(test)]
mod perf_tests;
#[cfg(test)]
mod tests;

const OBSERVER_BUFFER_CAP: usize = 1_000;
const OBSERVER_REPLAY_MAX_BYTES: usize = 4 * 1024 * 1024;

/// Best-effort metadata attached to observer events.
#[derive(Clone, Debug, Default)]
pub struct ObserverContext {
    /// Buzz channel UUID for the current turn, when channel-scoped.
    pub channel_id: Option<String>,
    /// ACP session ID associated with the current turn, once known.
    pub session_id: Option<String>,
    /// Local UUID for one prompt turn.
    pub turn_id: Option<String>,
    /// RFC3339 timestamp at which the current turn began, when known.
    pub started_at: Option<String>,
}

/// Handle used by the harness to publish local observer events.
#[derive(Clone)]
pub struct ObserverHandle {
    inner: Arc<ObserverInner>,
}

struct ObserverInner {
    // Fit events before either buffer retains them. Tokio rounds 1,000 slots
    // up to 1,024, so the per-event cap bounds live JSON below 64 MiB.
    tx: broadcast::Sender<ObserverEvent>,
    buffer: Mutex<ReplayBuffer>,
    seq: AtomicU64,
}

#[derive(Default)]
struct ReplayBuffer {
    events: VecDeque<(usize, Arc<ObserverEvent>)>,
    bytes: usize,
}

fn new_observer_handle() -> ObserverHandle {
    let (tx, _) = broadcast::channel(OBSERVER_BUFFER_CAP);
    ObserverHandle {
        inner: Arc::new(ObserverInner {
            tx,
            buffer: Mutex::new(ReplayBuffer::default()),
            seq: AtomicU64::new(1),
        }),
    }
}

/// Event delivered through the in-process observer bus.
#[derive(Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ObserverEvent {
    /// Monotonic process-local sequence number.
    pub seq: u64,
    /// RFC3339 UTC timestamp.
    pub timestamp: String,
    /// Observer event kind, for example `acp_read` or `turn_started`.
    pub kind: String,
    /// Pool slot index for the agent process that emitted the event.
    pub agent_index: Option<usize>,
    /// Buzz channel UUID for channel-scoped events.
    pub channel_id: Option<String>,
    /// ACP session ID when known.
    pub session_id: Option<String>,
    /// Local UUID for one prompt turn.
    pub turn_id: Option<String>,
    /// RFC3339 timestamp at which the current turn began, when known.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub started_at: Option<String>,
    /// Raw or semantic event payload.
    pub payload: serde_json::Value,
}

impl ObserverHandle {
    /// Create an in-process observer feed.
    pub fn in_process() -> Self {
        new_observer_handle()
    }

    /// Subscribe to live observer events.
    pub fn subscribe(&self) -> broadcast::Receiver<ObserverEvent> {
        self.inner.tx.subscribe()
    }

    /// Return the newest events within the 1,000-event and 4 MiB JSON budgets.
    pub fn snapshot(&self) -> Vec<ObserverEvent> {
        // Pin the replay window under the lock, then copy JSON after releasing it.
        let events: Vec<_> = match self.inner.buffer.lock() {
            Ok(buffer) => buffer
                .events
                .iter()
                .map(|(_, event)| event.clone())
                .collect(),
            Err(error) => {
                tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}");
                Vec::new()
            }
        };
        events.into_iter().map(|event| (*event).clone()).collect()
    }

    /// Emit a local observer event, fitting telemetry before retaining it.
    pub fn emit(
        &self,
        kind: impl Into<String>,
        agent_index: Option<usize>,
        context: &ObserverContext,
        payload: serde_json::Value,
    ) {
        let mut event = ObserverEvent {
            seq: self.inner.seq.load(Ordering::Relaxed),
            timestamp: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Nanos, false),
            kind: kind.into(),
            agent_index,
            channel_id: context.channel_id.clone(),
            session_id: context.session_id.clone(),
            turn_id: context.turn_id.clone(),
            started_at: context.started_at.clone(),
            payload,
        };

        // Fit and copy before locking so large inputs do not stall other
        // emitters. Fixed timestamp precision keeps metadata width stable;
        // account for the final publication metadata below without a JSON walk.
        let prepared_metadata_bytes = event.seq.to_string().len() + event.timestamp.len();
        let mut bytes = super::fit_observer_event_to_budget(&mut event);
        let prepared_within_budget = bytes <= super::OBSERVER_MAX_PLAINTEXT_LEN;
        let mut replay = event.clone();
        let mut buffer = match self.inner.buffer.lock() {
            Ok(buffer) => buffer,
            Err(error) => {
                tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}");
                return;
            }
        };
        // Assign both fields with publication: desktop sorts by timestamp,
        // and the publisher deduplicates snapshots by sequence high-water mark.
        event.seq = self.inner.seq.fetch_add(1, Ordering::Relaxed);
        event.timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Nanos, false);
        bytes =
            bytes - prepared_metadata_bytes + event.seq.to_string().len() + event.timestamp.len();
        if prepared_within_budget && bytes > super::OBSERVER_MAX_PLAINTEXT_LEN {
            // A sequence-width boundary can grow a frame prepared just below
            // the cap. Only an already-bounded frame can need this final fit.
            bytes = super::fit_observer_event_to_budget(&mut event);
            replay = event.clone();
        }
        if bytes > super::OBSERVER_MAX_PLAINTEXT_LEN {
            tracing::warn!(target: "observer", seq = event.seq, bytes, dropped = 1,
                "observer envelope exceeds byte budget; dropped event");
            return;
        }
        replay.seq = event.seq;
        replay.timestamp.clone_from(&event.timestamp);
        while buffer.events.len() >= OBSERVER_BUFFER_CAP
            || buffer.bytes + bytes > OBSERVER_REPLAY_MAX_BYTES
        {
            let Some((old_bytes, _)) = buffer.events.pop_front() else {
                break;
            };
            buffer.bytes -= old_bytes;
        }
        buffer.bytes += bytes;
        buffer.events.push_back((bytes, Arc::new(replay)));

        let _ = self.inner.tx.send(event);
    }
}

/// Build observer context values from optional channel/session/turn IDs.
pub fn context_for(
    channel_id: Option<uuid::Uuid>,
    session_id: Option<String>,
    turn_id: Option<String>,
) -> ObserverContext {
    ObserverContext {
        channel_id: channel_id.map(|id| id.to_string()),
        session_id,
        turn_id,
        started_at: None,
    }
}

/// Attach the authoritative start timestamp to every observer frame for a turn.
pub fn context_for_turn(
    channel_id: Option<uuid::Uuid>,
    session_id: Option<String>,
    turn_id: String,
    started_at: String,
) -> ObserverContext {
    ObserverContext {
        channel_id: channel_id.map(|id| id.to_string()),
        session_id,
        turn_id: Some(turn_id),
        started_at: Some(started_at),
    }
}