diff --git a/crates/infinity-agent-core/src/event_processor.rs b/crates/infinity-agent-core/src/event_processor.rs index d934fd83..0d4b9ae1 100644 --- a/crates/infinity-agent-core/src/event_processor.rs +++ b/crates/infinity-agent-core/src/event_processor.rs @@ -117,6 +117,20 @@ pub enum CompletionEvent { SyncToolResult(ToolResult), /// Some piece of information to log to the user. Info(String), + /// A completion attempt failed and will be retried; see + /// [`AgentEvent::CompletionRetrying`](crate::system::AgentEvent::CompletionRetrying). + Retrying { + error: String, + attempt: u32, + delay: Duration, + }, + /// The completion gave up; see + /// [`AgentEvent::CompletionFailed`](crate::system::AgentEvent::CompletionFailed). + Failed { + error: String, + retries: u32, + discarded_user_messages: usize, + }, } // ── HistoryManager (unchanged from before) ── @@ -517,6 +531,16 @@ impl HistoryManager { self.unvalidated_items.borrow().len() } + /// Number of user messages among the inputs still awaiting model + /// validation. + pub fn unvalidated_user_messages(&self) -> usize { + self.unvalidated_items + .borrow() + .iter() + .filter(|item| matches!(item.message, InfinityMessage::User { .. })) + .count() + } + /// Drop all unvalidated *user* inputs: items that are neither tool /// results nor subscription events (those are placeholder-replaceable, /// see [`Self::replace_unvalidated_tool_results`], and answering a tool @@ -1408,6 +1432,29 @@ const THROTTLED_RETRY_DELAY: Duration = Duration::from_secs(30); /// error. const TRANSIENT_RETRY_DELAY: Duration = Duration::from_secs(5); +/// A retry notice for the failure that ended attempt `retry_count`. +fn retrying(retry_count: u32, error: impl std::fmt::Display, delay: Duration) -> CompletionEvent { + CompletionEvent::Retrying { + error: error.to_string(), + attempt: retry_count + 1, + delay, + } +} + +/// The give-up notice. The round's unvalidated inputs are not persisted +/// (nothing shows they are safe). +fn give_up( + history: &HistoryManager, + retry_count: u32, + error: impl std::fmt::Display, +) -> CompletionEvent { + CompletionEvent::Failed { + error: error.to_string(), + retries: retry_count, + discarded_user_messages: history.unvalidated_user_messages(), + } +} + /// How [`run_completion`] recovered from a provider-declared context /// overflow (see [`ErrorClass::ContextOverflow`]). enum OverflowRecovery { @@ -1555,10 +1602,7 @@ where match e.class() { ErrorClass::Throttled if retry_count < MAX_COMPLETION_RETRIES => { tracing::warn!("Stream error (rate limit), retrying..."); - yield CompletionEvent::Info(format!( - "Stream error (rate limit), retrying after {} seconds...", - THROTTLED_RETRY_DELAY.as_secs() - )); + yield retrying(retry_count, &e, THROTTLED_RETRY_DELAY); tokio::select! { _ = tokio::time::sleep(THROTTLED_RETRY_DELAY) => {} _ = &mut cancel_rx => { @@ -1571,7 +1615,7 @@ where } ErrorClass::Transient if retry_count < MAX_COMPLETION_RETRIES => { tracing::warn!("Stream error ({e}), retrying..."); - yield CompletionEvent::Info(format!("Stream error ({e}), retrying...")); + yield retrying(retry_count, &e, TRANSIENT_RETRY_DELAY); tokio::select! { _ = tokio::time::sleep(TRANSIENT_RETRY_DELAY) => {} _ = &mut cancel_rx => { @@ -1601,6 +1645,7 @@ where } } _ => { + yield give_up(history, retry_count, &e); Err(Into::::into(e))?; unreachable!() } @@ -1656,7 +1701,9 @@ where } else { // Giving up: keep visible text, trim trailing reasoning // (the next appended message is a user turn). + let error = give_up(history, retry_count, "the model stream ended unexpectedly"); history.flush_turn_trimming_reasoning(); + yield error; Err(Into::::into("Stream ended unexpectedly"))?; unreachable!() } @@ -1681,18 +1728,16 @@ where ErrorClass::Transient if retry_count < MAX_COMPLETION_RETRIES => { // Retry rebuilds from committed history. history.discard_turn(); - yield CompletionEvent::Info(format!("Stream error ({e}), retrying...")); + let delay = Duration::from_secs(1); + yield retrying(retry_count, &e, delay); tracing::warn!("Stream error (transient), discarding partial turn and retrying..."); - tokio::time::sleep(Duration::from_secs(1)).await; + tokio::time::sleep(delay).await; retry_count += 1; continue 'outer; } ErrorClass::Throttled if retry_count < MAX_COMPLETION_RETRIES => { history.discard_turn(); - yield CompletionEvent::Info(format!( - "Stream error (rate limit), retrying after {} seconds...", - THROTTLED_RETRY_DELAY.as_secs() - )); + yield retrying(retry_count, &e, THROTTLED_RETRY_DELAY); tokio::select! { _ = tokio::time::sleep(THROTTLED_RETRY_DELAY) => {} _ = &mut cancel_rx => { @@ -1727,7 +1772,9 @@ where _ => { // Giving up: keep visible text, trim trailing reasoning // (the next appended message is a user turn). + let error = give_up(history, retry_count, &e); history.flush_turn_trimming_reasoning(); + yield error; Err(Into::::into(e))?; unreachable!() } @@ -3636,6 +3683,8 @@ mod tests { while let Some(ev) = stream.next().await { tags.push(match ev.expect("receive stream event") { CompletionEvent::Info(_) => "Info", + CompletionEvent::Retrying { .. } => "Retrying", + CompletionEvent::Failed { .. } => "Failed", CompletionEvent::TextChunk(_) => "Text", CompletionEvent::ThinkingStart => "ThinkingStart", CompletionEvent::ThinkingEnd => "ThinkingEnd", diff --git a/crates/infinity-agent-core/src/system/events.rs b/crates/infinity-agent-core/src/system/events.rs index bcd92b3d..6abf5379 100644 --- a/crates/infinity-agent-core/src/system/events.rs +++ b/crates/infinity-agent-core/src/system/events.rs @@ -66,6 +66,27 @@ pub enum AgentEvent { UserChoiceDismissed { choice_id: rap_protocol::ChoiceId }, /// Human-readable progress/diagnostic information (retries, warnings). Info { text: String }, + /// A completion attempt failed and will be retried after `delay`. + /// `attempt` counts retries from 1; `error` is the failure's message. + /// Clients should surface the first retry prominently, so a failing + /// completion is never a silent retry loop. + CompletionRetrying { + error: String, + attempt: u32, + delay: std::time::Duration, + }, + /// A completion gave up after `retries` retries; `error` is the failure's + /// message. The round's inputs had not been validated by model output, + /// so none were persisted; `discarded_user_messages` counts the user + /// messages among them, which are not re-sent on the next round either. + CompletionFailed { + error: String, + retries: u32, + discarded_user_messages: usize, + }, + /// A thread operation outside a completion failed (loading the thread, + /// a step, dispatching a tool call); `error` is the failure's message. + Error { error: String }, } /// A live view of a thread, used to bring a newly attached subscriber up to diff --git a/crates/infinity-agent-core/src/system/local/driver.rs b/crates/infinity-agent-core/src/system/local/driver.rs index 201a2afd..4fc10523 100644 --- a/crates/infinity-agent-core/src/system/local/driver.rs +++ b/crates/infinity-agent-core/src/system/local/driver.rs @@ -120,8 +120,8 @@ pub(crate) async fn drive_thread( Err(e) => { observer.on_event( &thread_id, - &AgentEvent::Info { - text: format!("Error: {e}"), + &AgentEvent::Error { + error: e.to_string(), }, ); return; @@ -298,8 +298,8 @@ pub(crate) async fn drive_thread( Err(e) => { observer.on_event( &thread_id, - &AgentEvent::Info { - text: format!("Error: {e}"), + &AgentEvent::Error { + error: e.to_string(), }, ); continue; @@ -416,8 +416,8 @@ fn handle_step_result( Err(e) => { observer.on_event( thread_id, - &AgentEvent::Info { - text: format!("Error: {e}"), + &AgentEvent::Error { + error: e.to_string(), }, ); } diff --git a/crates/infinity-agent-core/src/system/tests.rs b/crates/infinity-agent-core/src/system/tests.rs index 4bde4307..5a38d4bc 100644 --- a/crates/infinity-agent-core/src/system/tests.rs +++ b/crates/infinity-agent-core/src/system/tests.rs @@ -2197,3 +2197,90 @@ async fn parent_message_reaches_child_after_child_compaction() { }) .await; } + +/// Drain events until the round's terminal `CompletionFinished`, returning +/// every failure event (`CompletionRetrying`, `CompletionFailed`, `Error`) +/// seen. +async fn failures_until_finished(rx: &mut mpsc::UnboundedReceiver) -> Vec { + let mut failures = Vec::new(); + loop { + match next_evt(rx).await { + Evt::E( + e @ (AgentEvent::CompletionRetrying { .. } + | AgentEvent::CompletionFailed { .. } + | AgentEvent::Error { .. }), + ) => failures.push(e), + Evt::E(AgentEvent::CompletionFinished { .. }) => return failures, + _ => {} + } + } +} + +/// A completion that cannot start (e.g. expired credentials) reports one +/// `CompletionFailed` with the provider's message, counting the user message +/// it discarded. That message is not persisted. +#[tokio::test(flavor = "current_thread")] +async fn completion_start_failure_reports_discarded_user_message() { + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + let (mut running, mut rx, ctrl, conv) = start_system(vec![], None); + ctrl.fail_next_stream(infinity_provider_protocol::CompletionError::provider( + infinity_provider_protocol::ErrorClass::Fatal, + "credentials expired; sign in again", + )); + running + .send_user_text(ThreadId::from_ref("t-fail"), "important question") + .await; + + let failures = failures_until_finished(&mut rx).await; + assert!( + matches!( + failures.as_slice(), + [AgentEvent::CompletionFailed { + error, + retries: 0, + discarded_user_messages: 1, + }] if error.contains("sign in again") + ), + "{failures:?}" + ); + wait_idle(&mut running).await; + assert!(!persisted(&conv, "t-fail").contains("important question")); + }) + .await; +} + +/// The first failure of a retried completion is reported right away, before +/// the retry, rather than after a silent retry loop. +#[tokio::test(flavor = "current_thread", start_paused = true)] +async fn first_transient_failure_is_surfaced_before_retry() { + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + let (running, mut rx, ctrl, _conv) = start_system(vec![], None); + ctrl.fail_next_stream(infinity_provider_protocol::CompletionError::provider( + infinity_provider_protocol::ErrorClass::Transient, + "backend hiccup", + )); + running + .send_user_text(ThreadId::from_ref("t-retry"), "hello") + .await; + + // A missing event fails via next_evt's timeout. + loop { + match next_evt(&mut rx).await { + Evt::E(AgentEvent::CompletionRetrying { error, attempt, .. }) => { + assert!(error.contains("backend hiccup"), "{error}"); + assert_eq!(attempt, 1); + break; + } + Evt::E(AgentEvent::CompletionFinished { .. }) => { + panic!("round must not finish before the retry"); + } + _ => {} + } + } + }) + .await; +} diff --git a/crates/infinity-agent-core/src/system/thread.rs b/crates/infinity-agent-core/src/system/thread.rs index 68fe013f..a7d23cf3 100644 --- a/crates/infinity-agent-core/src/system/thread.rs +++ b/crates/infinity-agent-core/src/system/thread.rs @@ -547,6 +547,8 @@ where &mut usage, ), Err(e) => { + // Already surfaced via `CompletionEvent::Failed`; Info + // avoids double-reporting. observer.on_event( &thread_id, &AgentEvent::Info { @@ -580,8 +582,8 @@ where { observer.on_event( &thread_id, - &AgentEvent::Info { - text: format!("Error: {}", e), + &AgentEvent::Error { + error: e.to_string(), }, ); } @@ -604,6 +606,34 @@ where CompletionEvent::Info(text) => { observer.on_event(thread_id, &AgentEvent::Info { text }); } + CompletionEvent::Retrying { + error, + attempt, + delay, + } => { + observer.on_event( + thread_id, + &AgentEvent::CompletionRetrying { + error, + attempt, + delay, + }, + ); + } + CompletionEvent::Failed { + error, + retries, + discarded_user_messages, + } => { + observer.on_event( + thread_id, + &AgentEvent::CompletionFailed { + error, + retries, + discarded_user_messages, + }, + ); + } CompletionEvent::TextChunk(text) => { *self.current_thinking.borrow_mut() = None; observer.on_event(thread_id, &AgentEvent::TextChunk { text }); diff --git a/crates/infinity-daemon/src/session/display.rs b/crates/infinity-daemon/src/session/display.rs index f6ccc05d..07cac46c 100644 --- a/crates/infinity-daemon/src/session/display.rs +++ b/crates/infinity-daemon/src/session/display.rs @@ -30,6 +30,41 @@ pub(crate) fn agent_event_to_daemon(thread_id: &ThreadId, evt: &AgentEvent) thread_id: tid, text: text.clone(), }, + AgentEvent::CompletionRetrying { + error, + attempt, + delay, + } => { + let text = format!( + "Completion failed ({error}); retrying in {}s...", + delay.as_secs() + ); + // The first failure is an error so a failing completion is never + // a silent retry loop; later retries are progress. + if *attempt == 1 { + DaemonMessage::Error { + thread_id: tid, + text, + } + } else { + DaemonMessage::Info { + thread_id: tid, + text, + } + } + } + AgentEvent::CompletionFailed { + error, + retries, + discarded_user_messages, + } => DaemonMessage::Error { + thread_id: tid, + text: completion_failed_text(error, *retries, *discarded_user_messages), + }, + AgentEvent::Error { error } => DaemonMessage::Error { + thread_id: tid, + text: format!("Error: {error}"), + }, AgentEvent::CompletionFinished { usage } => { let token_usage = usage.map(|u| TokenUsage { input_tokens: Some(u.input_tokens), @@ -74,6 +109,22 @@ pub(crate) fn agent_event_to_daemon(thread_id: &ThreadId, evt: &AgentEvent) } } +/// The give-up message. User messages in a daemon session come from its +/// clients, so a discarded one is the reader's to resend. +fn completion_failed_text(error: &str, retries: u32, discarded_user_messages: usize) -> String { + let after = match retries { + 0 => String::new(), + 1 => " after 1 retry".to_owned(), + n => format!(" after {n} retries"), + }; + let outcome = match discarded_user_messages { + 0 => String::new(), + 1 => " Your message was not kept; please resend it.".to_owned(), + n => format!(" Your last {n} messages were not kept; please resend them."), + }; + format!("Completion failed{after}: {error}.{outcome}") +} + pub(crate) fn history_message_to_daemon( msg: &InfinityMessage, tid: &ThreadId, @@ -171,3 +222,52 @@ pub(crate) fn history_message_to_daemon( } } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + fn text_of(msg: DaemonMessage) -> (bool, String) { + match msg { + DaemonMessage::Error { text, .. } => (true, text), + DaemonMessage::Info { text, .. } => (false, text), + other => panic!("unexpected message: {other:?}"), + } + } + + /// Only the first retry is an error; later retries are progress. + #[test] + fn first_retry_is_an_error_later_retries_are_info() { + let tid = ThreadId::from_ref("t"); + let retry = |attempt| AgentEvent::CompletionRetrying { + error: "backend hiccup".to_owned(), + attempt, + delay: Duration::from_secs(5), + }; + let (is_error, text) = text_of(agent_event_to_daemon(tid, &retry(1))); + assert!(is_error && text.contains("backend hiccup"), "{text}"); + let (is_error, _) = text_of(agent_event_to_daemon(tid, &retry(2))); + assert!(!is_error); + } + + /// The give-up error asks for a resend only when a user message was + /// discarded. + #[test] + fn give_up_asks_for_resend_only_when_a_user_message_was_discarded() { + let tid = ThreadId::from_ref("t"); + let failed = |discarded_user_messages| AgentEvent::CompletionFailed { + error: "access denied".to_owned(), + retries: 0, + discarded_user_messages, + }; + let (is_error, text) = text_of(agent_event_to_daemon(tid, &failed(1))); + assert!( + is_error && text.contains("access denied") && text.contains("resend"), + "{text}" + ); + let (_, text) = text_of(agent_event_to_daemon(tid, &failed(0))); + assert!(!text.contains("resend"), "{text}"); + } +} diff --git a/crates/infinity-provider-bedrock/src/lib.rs b/crates/infinity-provider-bedrock/src/lib.rs index 2e1cd8dd..3f2dc67d 100644 --- a/crates/infinity-provider-bedrock/src/lib.rs +++ b/crates/infinity-provider-bedrock/src/lib.rs @@ -51,17 +51,42 @@ const FINE_GRAINED_TOOL_STREAMING_BETA: &str = "fine-grained-tool-streaming-2025 /// Anthropic beta flag enabling the 1M-token context window. const CONTEXT_1M_BETA: &str = "context-1m-2025-08-07"; -/// Extract a useful message from an AWS SDK error: the service error message -/// when present (the SDK's plain `Display` omits it), otherwise the full -/// error chain. +/// Whether an SDK error is a failure to load credentials (every AWS +/// `CredentialsError` mentions "credential"); retrying cannot fix it. +fn is_credentials_error(err: &SdkError) -> bool +where + E: ProvideErrorMetadata + std::error::Error + 'static, + R: std::fmt::Debug, +{ + err.message().is_none() + && DisplayErrorContext(err) + .to_string() + .to_ascii_lowercase() + .contains("credential") +} + +/// A concise, user-facing message for an AWS SDK error: the service error +/// message when present (the SDK's plain `Display` omits it), otherwise the +/// error kind and its root cause rather than the full error chain. pub(crate) fn sdk_error_message(err: &SdkError) -> String where E: ProvideErrorMetadata + std::error::Error + 'static, R: std::fmt::Debug, { - match err.message() { - Some(message) => message.to_owned(), - None => DisplayErrorContext(err).to_string(), + if let Some(message) = err.message() { + return message.to_owned(); + } + // Credential failures arrive this way too; their root cause carries the + // credential source's own explanation (e.g. a `credential_process`'s + // output), which is what the user needs to fix them. + let mut root: &dyn std::error::Error = err; + while let Some(source) = root.source() { + root = source; + } + if std::ptr::addr_eq(root, err) { + err.to_string() + } else { + format!("{err}: {root}") } } @@ -115,8 +140,8 @@ pub(crate) fn classify_bedrock_error(code: Option<&str>, message: &str) -> Error /// Classify a full [`SdkError`]: service errors go through /// [`classify_bedrock_error`]; transport-level failures (dispatch, timeout, -/// unparsable response) are transient; request construction failures are -/// ours and fatal. +/// unparsable response) are transient, except credential failures; request +/// construction failures are ours and fatal. pub(crate) fn classify_sdk_error(err: &SdkError) -> ErrorClass where E: ProvideErrorMetadata + std::error::Error + 'static, @@ -124,6 +149,7 @@ where { match err { SdkError::ConstructionFailure(_) => ErrorClass::Fatal, + _ if is_credentials_error(err) => ErrorClass::Fatal, SdkError::TimeoutError(_) | SdkError::DispatchFailure(_) | SdkError::ResponseError(_) => { ErrorClass::Transient } @@ -626,4 +652,33 @@ mod tests { // The timeout must be the initiation timeout, not some other layer. assert!(started.elapsed() >= REQUEST_INITIATION_TIMEOUT); } + + /// Credential failures fail fast and keep the credential source's own + /// explanation; other transport failures stay transient. Both report + /// just the root cause, not the whole error chain. + #[test] + fn transport_errors_are_concise() { + use aws_sdk_bedrockruntime::operation::converse_stream::ConverseStreamError; + use aws_smithy_runtime_api::client::result::ConnectorError; + type Err = SdkError; + + let creds = Err::dispatch_failure(ConnectorError::other( + "an error occurred while loading credentials: Error retrieving credentials: \ + external process exited with code exit status: 1. \ + Your session has expired; sign in again." + .into(), + None, + )); + assert_eq!(classify_sdk_error(&creds), ErrorClass::Fatal); + let message = sdk_error_message(&creds); + assert!( + message.ends_with("Your session has expired; sign in again."), + "{message}" + ); + + let net = Err::dispatch_failure(ConnectorError::io("connection reset by peer".into())); + assert_eq!(classify_sdk_error(&net), ErrorClass::Transient); + let message = sdk_error_message(&net); + assert!(message.ends_with("connection reset by peer"), "{message}"); + } } diff --git a/crates/infinity-provider-protocol/src/mock.rs b/crates/infinity-provider-protocol/src/mock.rs index 7ab7b7e6..1af134f6 100644 --- a/crates/infinity-provider-protocol/src/mock.rs +++ b/crates/infinity-provider-protocol/src/mock.rs @@ -6,6 +6,7 @@ //! drives it from the test. use async_trait::async_trait; +use std::sync::{Arc, Mutex}; use tokio::sync::mpsc; use crate::{ @@ -27,11 +28,21 @@ struct StreamRound { #[derive(Clone)] pub struct MockCompletionModel { round_tx: mpsc::UnboundedSender, + /// Set by [`MockModelController::fail_next_stream`]. + init_failure: Arc>>, } #[async_trait] impl CompletionModel for MockCompletionModel { async fn stream(&self, request: CompletionRequest) -> Result { + if let Some(err) = self + .init_failure + .lock() + .expect("bug: mock mutex poisoned") + .take() + { + return Err(err); + } let (chunk_tx, mut chunk_rx) = mpsc::unbounded_channel(); // Notify controller of this round. @@ -55,9 +66,17 @@ pub struct MockModelController { round_rx: mpsc::UnboundedReceiver, /// Sender for the current active round (set after `next_request`). current_tx: Option>>, + /// Shared with the model: the pending initiation failure. + init_failure: Arc>>, } impl MockModelController { + /// The next `stream()` call returns this error instead of a stream (no + /// round reaches the controller). + pub fn fail_next_stream(&self, err: CompletionError) { + *self.init_failure.lock().expect("bug: mock mutex poisoned") = Some(err); + } + /// Wait for the next `stream()` call and return the request. pub async fn next_request(&mut self) -> CompletionRequest { let round = self.round_rx.recv().await.expect("model dropped"); @@ -122,11 +141,16 @@ impl MockModelController { /// Create a mock model / controller pair. pub fn mock_model() -> (MockCompletionModel, MockModelController) { let (round_tx, round_rx) = mpsc::unbounded_channel(); + let init_failure = Arc::new(Mutex::new(None)); ( - MockCompletionModel { round_tx }, + MockCompletionModel { + round_tx, + init_failure: init_failure.clone(), + }, MockModelController { round_rx, current_tx: None, + init_failure, }, ) } diff --git a/docs/docs/infinity-runtime/agent-systems/observers.md b/docs/docs/infinity-runtime/agent-systems/observers.md index 5d558f2d..dc9adfa3 100644 --- a/docs/docs/infinity-runtime/agent-systems/observers.md +++ b/docs/docs/infinity-runtime/agent-systems/observers.md @@ -76,7 +76,7 @@ let running = system.start_with_observer(move |_thread_id| MyObserver { `start_with_observer` takes a factory rather than an observer because each thread's driver receives its own instance, created when the driver spawns. State that must outlive a driver, such as the client registry above, should live outside the observer and be cloned into each instance. For the trivial case there is a provided implementation: `EventCollector` buffers `(thread_id, event)` pairs in memory, which suits a [step-mode](./step-mode.md) handler that inspects them after the slice. -`AgentEvent` is the display-level record of a step, in order: the `UserInput` echo, `CompletionStarted`, streamed `TextChunk`s and `ThinkingStarted`/`ThinkingChunk`/`ThinkingEnded`, `ToolCall` (with a pretty-printed `display_as` when the tool has a display script), `ToolResult` (with RAP display segments), and `CompletionFinished` (with token usage), plus out-of-band `SubscriptionEvent`, `OAuthRequired`, `UserChoiceRequired`/`UserChoiceDismissed`, `CompactionApplied`, and `Info` diagnostics. The type is `Clone` and generic-free, so you can fan events out to any number of subscribers or buffer them freely. +`AgentEvent` is the display-level record of a step, in order: the `UserInput` echo, `CompletionStarted`, streamed `TextChunk`s and `ThinkingStarted`/`ThinkingChunk`/`ThinkingEnded`, `ToolCall` (with a pretty-printed `display_as` when the tool has a display script), `ToolResult` (with RAP display segments), and `CompletionFinished` (with token usage), plus out-of-band `SubscriptionEvent`, `OAuthRequired`, `UserChoiceRequired`/`UserChoiceDismissed`, `CompactionApplied`, and `Info` diagnostics. Failures are reported as data, so your application chooses the wording: `CompletionRetrying` (error, retry number, delay), `CompletionFailed` (error, retries, and how many user messages were discarded), and `Error` for failures outside a completion. The type is `Clone` and generic-free, so you can fan events out to any number of subscribers or buffer them freely. `on_event` is synchronous and called inline at the emission point, so it has to be fast: push to channels, append to buffers, or update in-memory counters. Work that needs to await should be moved to a task you spawn from it.