fix(infinity-agent-core): surface completion failures to clients

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<C: ConversationStore, S: StateStore> HistoryManager<C, S> {
         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<C: ConversationStore, S: StateStore>(
+    history: &HistoryManager<C, S>,
+    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::<BoxError>::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::<BoxError>::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::<BoxError>::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",

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

crates/infinity-agent-core/src/system/local/driver.rs

@@ -120,8 +120,8 @@ pub(crate) async fn drive_thread<C, S, H, O>(
         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<C, S, H, O>(
             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(),
                 },
             );
         }

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<Evt>) -> Vec<AgentEvent> {
+    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;
+}

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 });