diff --git a/src/lib.rs b/src/lib.rs index 1b65f0c..d667a03 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,6 +13,7 @@ //! - **[`builtin`]** — Reference implementations of core traits ([`builtin::memory::InMemoryStore`], [`builtin::observer::LoggingObserver`], etc.). //! - **[`loop_control`]** — Detection and intervention modules for agent loops. //! - **[`engine`]** — The core agentic loop that orchestrates the full agent lifecycle. +//! - **[`observability`]** — Structured event streaming (`EventSink`, `ObserveEvent`, metrics). //! - **[`stream`]** — Streaming event types for LLM API responses. //! - **[`tool`]** — Tool trait, registry, and supporting types. @@ -26,6 +27,7 @@ pub mod core; pub mod engine; pub mod loop_control; pub mod message; +pub mod observability; pub mod stream; #[cfg(feature = "testing")] pub mod testing; diff --git a/src/observability.rs b/src/observability.rs new file mode 100644 index 0000000..181cc81 --- /dev/null +++ b/src/observability.rs @@ -0,0 +1,72 @@ +//! Structured event streaming for agent observability. +//! +//! This module provides the [`EventSink`] trait and [`ObserveEvent`] enum — +//! the primary observability abstractions in `loopctl`. Every consumer — +//! console logging, JSONL files, metrics, custom monitors — implements +//! [`EventSink`]. +//! +//! # Architecture +//! +//! ```text +//! ┌───────────────────────────┐ +//! │ Agent Loop │ +//! │ │ +//! │ emits ObserveEvent ├──▶ dyn EventSink::on_event() +//! │ │ │ +//! └───────────────────────────┘ ├─▶ ConsoleSink +//! ├─▶ NullSink +//! ├─▶ CompositeSink ─┬─▶ Sink 1 +//! │ ├─▶ Sink 2 +//! │ └─▶ Sink 3 +//! └─▶ MetricsSink (feature-gated) +//! ``` +//! +//! # Provided Sinks +//! +//! | Sink | Purpose | +//! |---------------------|------------------------------------------------------| +//! | [`NullSink`] | Discards all events. Useful as a default. | +//! | [`ConsoleSink`] | Prints human-readable summaries to stdout. | +//! | [`CompositeSink`] | Fans out to multiple sinks with panic isolation. | +//! +//! # Event Types +//! +//! [`ObserveEvent`] covers the full agent lifecycle: +//! +//! - Session lifecycle (`SessionStart`, `SessionStop`) +//! - Turn lifecycle (`TurnStart`, `TurnComplete`, `TurnFailed`) +//! - Tool execution (`ToolStart`, `ToolComplete`) +//! - Context management (`ContextWarning`, `ContextCompacted`) +//! - Errors (`Error`) +//! +//! # Quick Start +//! +//! ```rust +//! use loopctl::observability::{EventSink, NullSink, ObserveEvent}; +//! +//! let sink = NullSink; +//! sink.on_event(&ObserveEvent::SessionStart { +//! session_id: uuid::Uuid::nil(), +//! }); +//! ``` +//! +//! # Composing Sinks +//! +//! Use [`CompositeSink`] to fan out to multiple sinks: +//! +//! ```rust +//! use loopctl::observability::{CompositeSink, ConsoleSink, NullSink}; +//! +//! let sink = CompositeSink::new(vec![ +//! Box::new(ConsoleSink), +//! Box::new(NullSink), +//! ]); +//! ``` + +pub mod console; +pub mod event; +pub mod sink; + +pub use console::ConsoleSink; +pub use event::ObserveEvent; +pub use sink::{CompositeSink, EventSink, NullSink}; diff --git a/src/observability/console.rs b/src/observability/console.rs new file mode 100644 index 0000000..1172119 --- /dev/null +++ b/src/observability/console.rs @@ -0,0 +1,225 @@ +//! A console-based [`EventSink`] that prints human-readable event summaries. +//! +//! [`ConsoleSink`] provides lightweight, readable output suitable for +//! development and debugging. Not intended for production logging — use +//! a structured sink (JSONL, metrics) for that. + +use super::event::ObserveEvent; +use super::sink::EventSink; + +/// Prints a human-readable summary of each event to stdout. +/// +/// Output is designed for developer ergonomics, not machine parsing. +/// Use this during development to see what the agent loop is doing. +/// Each event type gets a formatted one-line summary with the relevant +/// context (turn number, duration, token counts, etc.). +/// +/// # Example +/// +/// ```rust +/// use loopctl::observability::{ConsoleSink, EventSink, ObserveEvent}; +/// +/// let sink = ConsoleSink; +/// sink.on_event(&ObserveEvent::TurnComplete { +/// turn: 3, +/// duration_ms: 1200, +/// input_tokens: 450, +/// output_tokens: 200, +/// }); +/// // Prints: [turn 3] complete in 1200ms (450 in / 200 out tokens) +/// ``` +#[derive(Debug, Clone, Copy, Default)] +pub struct ConsoleSink; + +impl EventSink for ConsoleSink { + fn on_event(&self, event: &ObserveEvent) { + match event { + ObserveEvent::SessionStart { session_id } => { + println!("[session] started {session_id}"); + } + ObserveEvent::SessionStop { + session_id, + success, + total_turns, + duration_ms, + reason, + } => { + let status = if *success { "ok" } else { "failed" }; + println!( + "[session] {status} {session_id} \u{2014} {total_turns} turns in {duration_ms}ms ({reason})" + ); + } + ObserveEvent::TurnStart { turn, query } => { + let preview = truncate(query, 60); + println!("[turn {turn}] start: {preview}"); + } + ObserveEvent::TurnComplete { + turn, + duration_ms, + input_tokens, + output_tokens, + } => { + println!( + "[turn {turn}] complete in {duration_ms}ms ({input_tokens} in / {output_tokens} out tokens)" + ); + } + ObserveEvent::TurnFailed { + turn, + duration_ms, + error, + } => { + println!("[turn {turn}] failed after {duration_ms}ms: {error}"); + } + ObserveEvent::ToolStart { name, .. } => { + println!("[tool] {name} started"); + } + ObserveEvent::ToolComplete { + name, + is_error: false, + duration_ms, + .. + } => { + println!("[tool] {name} completed in {duration_ms}ms"); + } + ObserveEvent::ToolComplete { + name, + is_error: true, + duration_ms, + .. + } => { + println!("[tool] {name} failed after {duration_ms}ms"); + } + ObserveEvent::ContextWarning { + tokens_used, + tokens_remaining, + } => { + println!("[context] warning: {tokens_used} used, {tokens_remaining} remaining"); + } + ObserveEvent::ContextCompacted { + messages_before, + messages_after, + tokens_saved, + } => { + println!( + "[context] compacted {messages_before} \u{2192} {messages_after} messages ({tokens_saved} tokens saved)" + ); + } + ObserveEvent::Error { message, source } => { + println!("[error] {source}: {message}"); + } + } + } +} + +/// Truncate a string to approximately `max_len` characters, appending an +/// ellipsis (`\u{2026}`) if truncated. +/// +/// Handles multi-byte characters correctly by operating on char boundaries. +fn truncate(s: &str, max_len: usize) -> String { + if s.chars().count() <= max_len { + return s.to_string(); + } + let truncated: String = s.chars().take(max_len).collect(); + format!("{truncated}\u{2026}") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn console_sink_handles_all_event_types() { + let sink = ConsoleSink; + + sink.on_event(&ObserveEvent::SessionStart { + session_id: uuid::Uuid::nil(), + }); + sink.on_event(&ObserveEvent::TurnStart { + turn: 0, + query: "hello".to_string(), + }); + sink.on_event(&ObserveEvent::ToolStart { + name: "read_file".to_string(), + input: "{}".to_string(), + }); + sink.on_event(&ObserveEvent::ToolComplete { + name: "read_file".to_string(), + output: "contents".to_string(), + is_error: false, + duration_ms: 50, + }); + sink.on_event(&ObserveEvent::ToolComplete { + name: "read_file".to_string(), + output: "not found".to_string(), + is_error: true, + duration_ms: 10, + }); + sink.on_event(&ObserveEvent::TurnComplete { + turn: 0, + duration_ms: 100, + input_tokens: 10, + output_tokens: 5, + }); + sink.on_event(&ObserveEvent::TurnFailed { + turn: 1, + duration_ms: 200, + error: "timeout".to_string(), + }); + sink.on_event(&ObserveEvent::ContextWarning { + tokens_used: 180_000, + tokens_remaining: 20_000, + }); + sink.on_event(&ObserveEvent::ContextCompacted { + messages_before: 50, + messages_after: 20, + tokens_saved: 10_000, + }); + sink.on_event(&ObserveEvent::Error { + message: "something broke".to_string(), + source: "api".to_string(), + }); + sink.on_event(&ObserveEvent::SessionStop { + session_id: uuid::Uuid::nil(), + success: true, + reason: "done".to_string(), + total_turns: 5, + duration_ms: 10_000, + }); + } + + #[test] + fn truncate_short_string_unchanged() { + assert_eq!(truncate("hello", 10), "hello"); + } + + #[test] + fn truncate_long_string() { + let long = "a".repeat(100); + let result = truncate(&long, 10); + assert_eq!(result.chars().count(), 11); // 10 chars + ellipsis + assert!(result.ends_with('\u{2026}')); + } + + #[test] + fn truncate_exact_length() { + assert_eq!(truncate("exactly!", 8), "exactly!"); + } + + #[test] + fn truncate_empty_string() { + assert_eq!(truncate("", 10), ""); + } + + #[test] + fn truncate_multibyte_chars() { + let input = "héllo😊world"; + let result = truncate(input, 5); + assert_eq!(result, "héllo…"); + } + + #[test] + fn truncate_zero_max_len() { + let result = truncate("hello", 0); + assert_eq!(result, "\u{2026}"); + } +} diff --git a/src/observability/event.rs b/src/observability/event.rs new file mode 100644 index 0000000..7471480 --- /dev/null +++ b/src/observability/event.rs @@ -0,0 +1,355 @@ +//! Observable events emitted during the agent lifecycle. +//! +//! [`ObserveEvent`] is the central enum for all structured events in the +//! framework. The agent loop emits these at key lifecycle points, and +//! [`EventSink`](super::EventSink) implementations consume them. +//! +//! # Serialization +//! +//! Uses `#[serde(tag = "type")]` with `rename_all = "snake_case"` to produce +//! JSON objects like `{"type": "session_start", "session_id": "..."}`. +//! +//! Agents can wrap `ObserveEvent` in their own enum that adds agent-specific +//! variants. The framework's `EventSink` accepts `&ObserveEvent`; agent sinks +//! can extend with their own event types. + +use serde::{Deserialize, Serialize}; + +/// All observable events in the agent lifecycle. +/// +/// Each variant captures the relevant context for its lifecycle point. +/// Events are ordered chronologically within a session: +/// +/// ```text +/// SessionStart +/// └─ TurnStart +/// ├─ ToolStart ── ToolComplete [per tool call] +/// └─ ContextWarning? [if context is low] +/// └─ TurnComplete | TurnFailed +/// └─ ... +/// └─ ContextCompacted? [if compaction ran] +/// SessionStop +/// ``` +/// +/// # Serialization +/// +/// Each variant serializes as a JSON object with a `"type"` tag: +/// +/// ```rust +/// use loopctl::observability::ObserveEvent; +/// +/// let event = ObserveEvent::SessionStart { +/// session_id: uuid::Uuid::nil(), +/// }; +/// let json = serde_json::to_string(&event).unwrap(); +/// assert!(json.contains("\"type\":\"session_start\"")); +/// ``` +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ObserveEvent { + /// Session has started. + /// + /// Emitted once at the beginning of an agent session, before any turns. + SessionStart { + /// Unique session identifier. + session_id: uuid::Uuid, + }, + + /// Session has ended. + /// + /// Emitted once after the agent session completes (or fails). + /// Contains aggregate statistics about the entire session. + SessionStop { + /// Unique session identifier. + session_id: uuid::Uuid, + /// Whether the session completed successfully. + success: bool, + /// Why the session stopped (`"max_turns"`, `"cancelled"`, `"error"`, etc.). + reason: String, + /// Total turns completed. + total_turns: usize, + /// Total session duration in milliseconds. + duration_ms: u64, + }, + + /// A turn has started. + /// + /// Emitted at the beginning of each turn, before the LLM is called. + TurnStart { + /// Current turn number (0-indexed). + turn: usize, + /// The user query that initiated this turn. + query: String, + }, + + /// A turn completed successfully. + /// + /// Emitted after the LLM response has been fully processed, + /// including any tool calls made during the turn. + TurnComplete { + /// Turn number. + turn: usize, + /// Wall-clock duration of the turn in milliseconds. + duration_ms: u64, + /// Input tokens consumed this turn. + input_tokens: u64, + /// Output tokens generated this turn. + output_tokens: u64, + }, + + /// A turn failed. + /// + /// Emitted when a turn encounters an unrecoverable error. + TurnFailed { + /// Turn number. + turn: usize, + /// Wall-clock duration before failure. + duration_ms: u64, + /// Error description. + error: String, + }, + + /// A tool execution has started. + /// + /// Emitted just before a tool is invoked. + ToolStart { + /// Tool name. + name: String, + /// Tool input as a JSON string. + input: String, + }, + + /// A tool execution has completed. + /// + /// Emitted after the tool returns, whether successfully or with an error. + ToolComplete { + /// Tool name. + name: String, + /// Tool output as a string. + output: String, + /// Whether the tool reported an error. + is_error: bool, + /// Wall-clock duration in milliseconds. + duration_ms: u64, + }, + + /// Context window is running low on capacity. + /// + /// Emitted when token usage exceeds a warning threshold, + /// before compaction is triggered. + ContextWarning { + /// Estimated tokens currently used. + tokens_used: u64, + /// Estimated tokens remaining. + tokens_remaining: u64, + }, + + /// A context compaction occurred. + /// + /// Emitted after the conversation history has been compacted + /// to free up context window space. + ContextCompacted { + /// Messages before compaction. + messages_before: usize, + /// Messages after compaction. + messages_after: usize, + /// Estimated tokens saved by compaction. + tokens_saved: u64, + }, + + /// A generic error event. + /// + /// Emitted for errors that don't fit a more specific category. + Error { + /// Error description. + message: String, + /// Error source or category. + source: String, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn session_start_round_trip() { + let event = ObserveEvent::SessionStart { + session_id: uuid::Uuid::nil(), + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"session_start\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!(de, ObserveEvent::SessionStart { .. })); + } + + #[test] + fn session_stop_round_trip() { + let event = ObserveEvent::SessionStop { + session_id: uuid::Uuid::nil(), + success: true, + reason: "max_turns".to_string(), + total_turns: 5, + duration_ms: 10_000, + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"session_stop\"")); + assert!(json.contains("\"success\":true")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!(de, ObserveEvent::SessionStop { .. })); + } + + #[test] + fn turn_complete_round_trip() { + let event = ObserveEvent::TurnComplete { + turn: 3, + duration_ms: 1200, + input_tokens: 450, + output_tokens: 200, + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"turn_complete\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + if let ObserveEvent::TurnComplete { + turn, + input_tokens, + output_tokens, + .. + } = de + { + assert_eq!(turn, 3); + assert_eq!(input_tokens, 450); + assert_eq!(output_tokens, 200); + } else { + panic!("expected TurnComplete"); + } + } + + #[test] + fn tool_complete_round_trip() { + let success = ObserveEvent::ToolComplete { + name: "read_file".to_string(), + output: "file contents".to_string(), + is_error: false, + duration_ms: 50, + }; + let json = serde_json::to_string(&success).unwrap(); + assert!(json.contains("\"type\":\"tool_complete\"")); + assert!(json.contains("\"is_error\":false")); + + let failure = ObserveEvent::ToolComplete { + name: "read_file".to_string(), + output: "not found".to_string(), + is_error: true, + duration_ms: 10, + }; + let json = serde_json::to_string(&failure).unwrap(); + assert!(json.contains("\"is_error\":true")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + if let ObserveEvent::ToolComplete { is_error, .. } = de { + assert!(is_error); + } else { + panic!("expected ToolComplete"); + } + } + + #[test] + fn context_compacted_round_trip() { + let event = ObserveEvent::ContextCompacted { + messages_before: 50, + messages_after: 20, + tokens_saved: 10_000, + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"context_compacted\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + if let ObserveEvent::ContextCompacted { + messages_before, + messages_after, + tokens_saved, + } = de + { + assert_eq!(messages_before, 50); + assert_eq!(messages_after, 20); + assert_eq!(tokens_saved, 10_000); + } else { + panic!("expected ContextCompacted"); + } + } + + #[test] + fn error_round_trip() { + let event = ObserveEvent::Error { + message: "something went wrong".to_string(), + source: "api".to_string(), + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"error\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + if let ObserveEvent::Error { message, source } = de { + assert_eq!(message, "something went wrong"); + assert_eq!(source, "api"); + } else { + panic!("expected Error"); + } + } + + #[test] + fn context_warning_round_trip() { + let event = ObserveEvent::ContextWarning { + tokens_used: 180_000, + tokens_remaining: 20_000, + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"context_warning\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!(de, ObserveEvent::ContextWarning { .. })); + } + + #[test] + fn turn_start_round_trip() { + let event = ObserveEvent::TurnStart { + turn: 0, + query: "hello".to_string(), + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"turn_start\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!(de, ObserveEvent::TurnStart { .. })); + } + + #[test] + fn turn_failed_round_trip() { + let event = ObserveEvent::TurnFailed { + turn: 2, + duration_ms: 500, + error: "timeout".to_string(), + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"turn_failed\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!(de, ObserveEvent::TurnFailed { .. })); + } + + #[test] + fn tool_start_round_trip() { + let event = ObserveEvent::ToolStart { + name: "read_file".to_string(), + input: r#"{"path":"/tmp/x"}"#.to_string(), + }; + let json = serde_json::to_string(&event).unwrap(); + assert!(json.contains("\"type\":\"tool_start\"")); + + let de: ObserveEvent = serde_json::from_str(&json).unwrap(); + assert!(matches!(de, ObserveEvent::ToolStart { .. })); + } +} diff --git a/src/observability/sink.rs b/src/observability/sink.rs new file mode 100644 index 0000000..f72913c --- /dev/null +++ b/src/observability/sink.rs @@ -0,0 +1,410 @@ +//! The [`EventSink`] trait and provided implementations. +//! +//! [`EventSink`] is the primary observability abstraction in `loopctl`. +//! Every consumer — console logging, JSONL files, metrics, custom monitors — +//! implements this trait. +//! +//! # Implementations +//! +//! - [`NullSink`] — Discards all events. Useful as a default. +//! - [`CompositeSink`] — Fans out to multiple sinks with panic isolation. + +use super::event::ObserveEvent; +use std::panic::catch_unwind; +use std::sync::Arc; + +// =================================================== +// EventSink trait +// =================================================== + +/// The primary observability abstraction in `loopctl`. +/// +/// Every consumer — console logging, JSONL files, metrics, custom monitors — +/// implements `EventSink`. The agent loop calls [`on_event`](EventSink::on_event) +/// at key lifecycle points with an [`ObserveEvent`]. +/// +/// # Object Safety +/// +/// The trait is object-safe: `Box` and `Arc` work. +/// +/// # Example +/// +/// ```rust +/// use loopctl::observability::{EventSink, ObserveEvent}; +/// +/// struct PrintSink; +/// +/// impl EventSink for PrintSink { +/// fn on_event(&self, event: &ObserveEvent) { +/// match event { +/// ObserveEvent::ToolStart { name, .. } => { +/// println!("tool started: {name}"); +/// } +/// ObserveEvent::TurnComplete { turn, duration_ms, .. } => { +/// println!("turn {turn} done in {duration_ms}ms"); +/// } +/// _ => {} +/// } +/// } +/// } +/// ``` +pub trait EventSink: Send + Sync { + /// Handle an observed event. + /// + /// Called by the agent loop at each lifecycle point. Implementations + /// should be fast and non-blocking — heavy work should be offloaded + /// to a channel or background task. + fn on_event(&self, event: &ObserveEvent); +} + +// =================================================== +// NullSink +// =================================================== + +/// A sink that discards all events. +/// +/// Useful as a default when no observability is needed, or as a +/// placeholder during testing. Equivalent to `/dev/null` for events. +/// +/// # Example +/// +/// ```rust +/// use loopctl::observability::{EventSink, NullSink, ObserveEvent}; +/// +/// let sink = NullSink; +/// sink.on_event(&ObserveEvent::SessionStart { +/// session_id: uuid::Uuid::nil(), +/// }); +/// // Event is silently discarded. +/// ``` +#[derive(Debug, Clone, Copy, Default)] +pub struct NullSink; + +impl EventSink for NullSink { + fn on_event(&self, _event: &ObserveEvent) {} +} + +// =================================================== +// CompositeSink +// =================================================== + +/// A composite sink that fans out every event to multiple inner sinks. +/// +/// Holds an ordered list of [`EventSink`] trait objects behind [`Arc`] +/// and forwards each event to every inner sink in sequence. This +/// implements the classic **Composite** pattern, allowing you to combine +/// console logging, JSONL output, and custom sinks without writing a +/// new struct. +/// +/// Sinks are called in insertion order. If any individual sink panics, +/// the remaining sinks in the list are still called — this is achieved +/// internally via [`std::panic::catch_unwind`]. This ensures one +/// misbehaving sink does not prevent others from receiving events. +/// +/// # Architecture +/// +/// ```text +/// CompositeSink +/// ┌──────────────┐ +/// on_event() ─►│ sinks[] │──▶ sink[0].on_event() +/// │ │──▶ sink[1].on_event() +/// │ │──▶ sink[2].on_event() +/// └──────────────┘ +/// ``` +/// +/// # Thread Safety +/// +/// Each inner sink is stored as `Arc`, so the same sink +/// can be shared across multiple [`CompositeSink`] instances or other +/// parts of the system. The fan-out loop borrows `&self`, meaning all +/// callbacks must be `&self`-safe (no `&mut self`). +/// +/// # Example +/// +/// ```rust +/// use loopctl::observability::{CompositeSink, ConsoleSink, NullSink, EventSink}; +/// use std::sync::Arc; +/// +/// // Build with owned sinks: +/// let sink = CompositeSink::new(vec![ +/// Box::new(ConsoleSink), +/// Box::new(NullSink), +/// ]); +/// assert_eq!(sink.len(), 2); +/// +/// // Build with the builder pattern: +/// let sink = CompositeSink::new(vec![]) +/// .with(ConsoleSink) +/// .with(NullSink); +/// assert_eq!(sink.len(), 2); +/// +/// // Build with pre-Arc'd sinks: +/// let shared = Arc::new(NullSink); +/// let sink = CompositeSink::new(vec![]) +/// .with_arc(shared.clone()) +/// .with_arc(shared); // same sink twice +/// assert_eq!(sink.len(), 2); +/// ``` +pub struct CompositeSink { + /// The ordered list of inner sinks to fan out to. + /// + /// Each sink is stored as `Arc` so it can be shared + /// across threads. Sinks are called in insertion order — first added + /// is first notified. + sinks: Vec>, +} + +impl std::fmt::Debug for CompositeSink { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CompositeSink") + .field("sink_count", &self.sinks.len()) + .finish() + } +} + +impl Default for CompositeSink { + /// Returns an empty composite sink with no inner sinks. + /// + /// The default instance contains zero inner sinks, so all events + /// are silently discarded until sinks are added. + fn default() -> Self { + Self { sinks: Vec::new() } + } +} + +impl CompositeSink { + /// Create a new composite sink that fans out to the given sinks. + /// + /// Each sink is wrapped in `Arc` and stored for + /// ordered fan-out. Returns a ready-to-use composite. + /// + /// # Example + /// + /// ```rust + /// use loopctl::observability::{CompositeSink, ConsoleSink, NullSink}; + /// + /// let sink = CompositeSink::new(vec![ + /// Box::new(ConsoleSink), + /// Box::new(NullSink), + /// ]); + /// assert_eq!(sink.len(), 2); + /// ``` + #[must_use] + pub fn new(sinks: Vec>) -> Self { + let sinks: Vec> = sinks.into_iter().map(Arc::from).collect(); + Self { sinks } + } + + /// Add an owned sink to the fan-out list. + /// + /// The sink is wrapped as `Arc` and appended to the + /// end of the list. Returns `self` for chaining. + /// + /// # Example + /// + /// ```rust + /// use loopctl::observability::{CompositeSink, ConsoleSink, NullSink}; + /// + /// let sink = CompositeSink::new(vec![]) + /// .with(ConsoleSink) + /// .with(NullSink); + /// assert_eq!(sink.len(), 2); + /// ``` + #[must_use] + pub fn with(mut self, sink: S) -> Self { + self.sinks.push(Arc::new(sink)); + self + } + + /// Add a sink that is already behind an `Arc`. + /// + /// Useful when multiple [`CompositeSink`] instances need to share the + /// same inner sink, or when the sink is constructed externally and + /// already wrapped in an `Arc`. + /// + /// # Example + /// + /// ```rust + /// use loopctl::observability::{CompositeSink, NullSink}; + /// use std::sync::Arc; + /// + /// let shared = Arc::new(NullSink); + /// let sink = CompositeSink::new(vec![]) + /// .with_arc(shared.clone()) + /// .with_arc(shared); + /// assert_eq!(sink.len(), 2); + /// ``` + #[must_use] + pub fn with_arc(mut self, sink: Arc) -> Self { + self.sinks.push(sink); + self + } + + /// Number of inner sinks in the fan-out list. + /// + /// Returns the count of sinks that will receive events. + #[must_use] + pub fn len(&self) -> usize { + self.sinks.len() + } + + /// Whether there are no inner sinks in the fan-out list. + /// + /// Returns `true` when [`len`](CompositeSink::len) is zero. When + /// empty, [`on_event`](EventSink::on_event) is effectively a no-op + /// (the internal loop body never executes). + #[must_use] + pub fn is_empty(&self) -> bool { + self.sinks.is_empty() + } + + /// Append a sink after construction. + /// + /// The sink is wrapped as `Arc` and appended to the + /// end of the fan-out list. + pub fn add(&mut self, sink: Box) { + self.sinks.push(Arc::from(sink)); + } +} + +impl EventSink for CompositeSink { + fn on_event(&self, event: &ObserveEvent) { + for sink in &self.sinks { + // Panic isolation: a failing sink must not break other sinks. + let result = catch_unwind(std::panic::AssertUnwindSafe(|| { + sink.on_event(event); + })); + if let Err(panic_payload) = result { + let msg = if let Some(s) = panic_payload.downcast_ref::<&str>() { + (*s).to_string() + } else if let Some(s) = panic_payload.downcast_ref::() { + s.clone() + } else { + "unknown panic".to_string() + }; + tracing::error!("EventSink panicked: {msg}"); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[test] + fn null_sink_accepts_events() { + let sink = NullSink; + sink.on_event(&ObserveEvent::SessionStart { + session_id: uuid::Uuid::nil(), + }); + } + + #[test] + fn composite_fans_out() { + static COUNT_A: AtomicUsize = AtomicUsize::new(0); + static COUNT_B: AtomicUsize = AtomicUsize::new(0); + + struct SinkA; + struct SinkB; + + impl EventSink for SinkA { + fn on_event(&self, _event: &ObserveEvent) { + COUNT_A.fetch_add(1, Ordering::Relaxed); + } + } + + impl EventSink for SinkB { + fn on_event(&self, _event: &ObserveEvent) { + COUNT_B.fetch_add(1, Ordering::Relaxed); + } + } + + let composite = CompositeSink::new(vec![Box::new(SinkA), Box::new(SinkB)]); + composite.on_event(&ObserveEvent::SessionStart { + session_id: uuid::Uuid::nil(), + }); + + assert_eq!(COUNT_A.load(Ordering::Relaxed), 1); + assert_eq!(COUNT_B.load(Ordering::Relaxed), 1); + } + + #[test] + fn composite_isolates_panics() { + static COUNT: AtomicUsize = AtomicUsize::new(0); + + struct PanicSink; + struct CountSink; + + impl EventSink for PanicSink { + fn on_event(&self, _event: &ObserveEvent) { + panic!("boom"); + } + } + + impl EventSink for CountSink { + fn on_event(&self, _event: &ObserveEvent) { + COUNT.fetch_add(1, Ordering::Relaxed); + } + } + + // PanicSink is first; CountSink should still receive the event. + let composite = CompositeSink::new(vec![Box::new(PanicSink), Box::new(CountSink)]); + composite.on_event(&ObserveEvent::SessionStart { + session_id: uuid::Uuid::nil(), + }); + + assert_eq!(COUNT.load(Ordering::Relaxed), 1); + } + + #[test] + fn composite_with_builder() { + let composite = CompositeSink::new(vec![]) + .with(NullSink) + .with(NullSink) + .with(NullSink); + assert_eq!(composite.len(), 3); + } + + #[test] + fn composite_with_arc_builder() { + let shared = Arc::new(NullSink); + let composite = CompositeSink::new(vec![]) + .with_arc(shared.clone()) + .with_arc(shared); + assert_eq!(composite.len(), 2); + } + + #[test] + fn composite_add_and_len() { + let mut composite = CompositeSink::new(vec![Box::new(NullSink)]); + assert_eq!(composite.len(), 1); + assert!(!composite.is_empty()); + + composite.add(Box::new(NullSink)); + assert_eq!(composite.len(), 2); + } + + #[test] + fn composite_empty() { + let composite = CompositeSink::new(vec![]); + assert!(composite.is_empty()); + composite.on_event(&ObserveEvent::SessionStart { + session_id: uuid::Uuid::nil(), + }); + } + + #[test] + fn composite_default_is_empty() { + let composite = CompositeSink::default(); + assert!(composite.is_empty()); + assert_eq!(composite.len(), 0); + } + + #[test] + fn event_sink_is_object_safe() { + let _boxed: Box = Box::new(NullSink); + let _arc: Arc = Arc::new(NullSink); + } +}