feat(agent): classify structured agent send errors (#356)

## Summary

- Add a public agent stream error taxonomy with ownership, code, retry,
feedback, and resolution metadata.
- Convert accepted-turn send failures and empty provider replies into
structured `message.stream` error events while preserving turn
completion semantics.
- Classify user agent, user LLM provider, AionUI, and unknown upstream
failures with priority-aware matching and redacted technical detail.
- Preserve mainline channel warmup/stream subscription behavior and
adapt affected mocks/tests to `AgentSendError`.

## Verification

- [x] `just push --force-with-lease origin agent-error-taxonomy` (cargo
fix, clippy fix, fmt, nextest, push)
- [x] `cargo nextest run --workspace` via `just push`: 5735 passed, 18
skipped
- [x] `cargo test -p aionui-channel`: 204 unit tests plus channel
integration tests passed

Author: TCP404

crates/aionui-ai-agent/src/agent_runtime.rs

@@ -14,6 +14,7 @@ use std::sync::{Arc, RwLock};
 
 use tokio::sync::broadcast;
 
+use aionui_api_types::AgentStreamErrorData;
 use aionui_common::{ConversationStatus, TimestampMs, now_ms};
 
 use crate::protocol::events::{AgentStreamEvent, ErrorEventData, FinishEventData};
@@ -127,6 +128,12 @@ impl AgentRuntime {
     /// Atomic: set status ← Finished AND broadcast `Error { message }`.
     /// Idempotent in the Finished absorbing state (no-op).
     pub fn emit_error(&self, message: impl Into<String>) {
+        self.emit_error_data(ErrorEventData::legacy(message, None));
+    }
+
+    /// Atomic: set status ← Finished AND broadcast the structured error payload.
+    /// Idempotent in the Finished absorbing state (no-op).
+    pub fn emit_error_data(&self, data: AgentStreamErrorData) {
         let already_finished = {
             let mut guard = self.status.write().unwrap_or_else(|e| e.into_inner());
             let was_finished = matches!(*guard, Some(ConversationStatus::Finished));
@@ -138,10 +145,7 @@ impl AgentRuntime {
         if already_finished {
             return;
         }
-        let _ = self.event_tx.send(AgentStreamEvent::Error(ErrorEventData {
-            message: message.into(),
-            code: None,
-        }));
+        let _ = self.event_tx.send(AgentStreamEvent::Error(data));
     }
 }
 

crates/aionui-ai-agent/src/agent_task.rs

@@ -22,6 +22,7 @@ use crate::manager::nanobot::NanobotAgentManager;
 use crate::manager::openclaw::OpenClawAgentManager;
 use crate::manager::remote::RemoteAgentManager;
 use crate::protocol::events::AgentStreamEvent;
+use crate::protocol::send_error::AgentSendError;
 use crate::types::SendMessageData;
 
 use aionui_api_types::{
@@ -62,7 +63,7 @@ pub trait IAgentTask: Send + Sync {
     /// Send a user message to the agent. Returns once the agent has
     /// accepted the turn; actual streaming proceeds on the broadcast
     /// channel returned by [`Self::subscribe`].
-    async fn send_message(&self, data: SendMessageData) -> Result<(), AppError>;
+    async fn send_message(&self, data: SendMessageData) -> Result<(), AgentSendError>;
 
     /// Stop the current streaming response without killing the agent.
     async fn cancel(&self) -> Result<(), AppError>;
@@ -220,7 +221,7 @@ impl AgentInstance {
     }
 
     /// Send a user message to the agent.
-    pub async fn send_message(&self, data: SendMessageData) -> Result<(), AppError> {
+    pub async fn send_message(&self, data: SendMessageData) -> Result<(), AgentSendError> {
         self.as_task().send_message(data).await
     }
 

crates/aionui-ai-agent/src/capability/backend_output_sink.rs

@@ -131,10 +131,9 @@ impl OutputSink for BackendOutputSink {
     }
 
     fn emit_error(&self, msg: &str) {
-        let _ = self.event_tx.send(AgentStreamEvent::Error(ErrorEventData {
-            message: msg.to_owned(),
-            code: None,
-        }));
+        let _ = self
+            .event_tx
+            .send(AgentStreamEvent::Error(ErrorEventData::legacy(msg, None)));
     }
 
     fn emit_info(&self, msg: &str) {

crates/aionui-ai-agent/src/lib.rs

@@ -32,6 +32,7 @@ pub use factory::{AgentFactoryDeps, build_agent_factory};
 pub use idle_scanner::start_idle_scanner;
 pub use persistence::AcpSessionSyncService;
 pub use protocol::events::AgentStreamEvent;
+pub use protocol::send_error::AgentSendError;
 pub use registry::{AgentRegistry, UnavailableReason};
 pub use routes::{AgentRouterState, RemoteAgentRouterState, agent_routes, remote_agent_routes};
 pub use services::AgentService;

crates/aionui-ai-agent/src/manager/acp/agent.rs

@@ -11,6 +11,7 @@ use crate::manager::process_registry::{register_session_process, unregister_agen
 use crate::protocol::acp::AcpProtocol;
 use crate::protocol::error::{AcpError, CloseReason};
 use crate::protocol::events::{AgentStreamEvent, FinishEventData};
+use crate::protocol::send_error::AgentSendError;
 use crate::registry::CatalogSender;
 use crate::shared_kernel::{ModeId, ModelId, SessionId as DomainSessionId};
 use crate::types::SendMessageData;
@@ -605,17 +606,17 @@ impl crate::agent_task::IAgentTask for AcpAgentManager {
     }
 
     #[tracing::instrument(skip_all, fields(conversation_id = %self.params.conversation_id, msg_id = %data.msg_id))]
-    async fn send_message(&self, data: SendMessageData) -> Result<(), AppError> {
+    async fn send_message(&self, data: SendMessageData) -> Result<(), AgentSendError> {
         self.runtime.bump_activity();
 
-        let result = self.ensure_session_and_send(&data).await;
-        match &result {
+        match self.ensure_session_and_send(&data).await {
             Ok(()) => {
                 info!("ACP send_message completed");
                 // ACP pattern: Finish with session_id = None (default).
                 // If ACP later wants to include the session_id in Finish,
                 // read it from `self.session.read().await.session_id()`.
                 self.runtime.emit_finish(None);
+                Ok(())
             }
             Err(err) => {
                 // Build a CloseReason that captures whatever context we still
@@ -628,22 +629,23 @@ impl crate::agent_task::IAgentTask for AcpAgentManager {
                 //      stderr-augmentation heuristic for the SDK's "default
                 //      Internal error" shape; otherwise the user-facing form
                 //      of the AppError is the best we can do.
-                let close_reason = self.build_close_reason_from_error(err).await;
+                let close_reason = self.build_close_reason_from_error(&err).await;
 
                 // Operator log: full error chain + the (raw, pre-redaction)
                 // stderr peek so on-call can correlate. The redacted summary
                 // is what reaches the UI.
                 let summary = close_reason.user_facing_message();
-                warn!(error = %ErrorChain(err), close_reason_summary = %summary, "ACP send_message failed");
+                error!(error = %ErrorChain(&err), close_reason_summary = %summary, "ACP send_message failed");
 
                 {
                     let mut session = self.session.write().await;
                     session.record_close_reason(Some(close_reason));
                 }
-                self.runtime.emit_error(summary);
+                let send_error = AgentSendError::from_app_error(err);
+                self.runtime.emit_error_data(send_error.stream_error().clone());
+                Err(send_error)
             }
         }
-        result
     }
 
     #[tracing::instrument(skip_all, fields(conversation_id = %self.params.conversation_id))]

crates/aionui-ai-agent/src/manager/acp/agent_session_flow.rs

@@ -7,6 +7,9 @@ use crate::protocol::events::{
 use crate::shared_kernel::SessionId as DomainSessionId;
 use crate::types::SendMessageData;
 use agent_client_protocol::schema::{ContentBlock, LoadSessionRequest, PromptRequest, SessionId, StopReason};
+use aionui_api_types::{
+    AgentErrorCode, AgentErrorOwnership, AgentErrorResolution, AgentErrorResolutionKind, AgentErrorResolutionTarget,
+};
 use aionui_common::AppError;
 use serde_json::Value;
 use tokio::sync::broadcast::error::TryRecvError;
@@ -259,13 +262,9 @@ impl AcpAgentManager {
         // user already initiated the cancel and doesn't need a second
         // notification.
         if !matches!(prompt_response.stop_reason, StopReason::Cancelled) && is_empty_turn(&mut probe_rx) {
-            self.runtime.emit(AgentStreamEvent::Error(ErrorEventData {
-                // TODO(i18n): wire to a frontend translation key once a
-                // pattern is established. For now this is the user-facing
-                // English string.
-                message: empty_finish_diagnostic_message(prompt_response.stop_reason),
-                code: Some("acp.empty_finish".into()),
-            }));
+            self.runtime.emit(AgentStreamEvent::Error(empty_finish_diagnostic_error(
+                prompt_response.stop_reason,
+            )));
         }
 
         // Emit Finish event
@@ -377,6 +376,24 @@ fn event_is_user_visible_output(event: &AgentStreamEvent) -> bool {
     )
 }
 
+fn empty_finish_diagnostic_error(stop_reason: StopReason) -> ErrorEventData {
+    ErrorEventData::classified(
+        // TODO(i18n): wire to a frontend translation key once a
+        // pattern is established. For now this is the user-facing
+        // English string.
+        empty_finish_diagnostic_message(stop_reason),
+        AgentErrorCode::UnknownUpstreamError,
+        AgentErrorOwnership::UnknownUpstream,
+        Some("Agent completed the turn without producing visible output.".into()),
+        true,
+        true,
+        Some(AgentErrorResolution::new(
+            AgentErrorResolutionKind::SendFeedback,
+            Some(AgentErrorResolutionTarget::Feedback),
+        )),
+    )
+}
+
 /// Build the user-facing message shown when the agent finished a turn
 /// without emitting any output. Wording is deliberately concrete so the
 /// user has something to act on (retry, reword, check provider).
@@ -580,6 +597,7 @@ mod tests {
         ToolCallStatus,
     };
     use agent_client_protocol::schema::StopReason;
+    use aionui_api_types::{AgentErrorResolutionKind, AgentErrorResolutionTarget};
     use tokio::sync::broadcast;
 
     /// Lifecycle-only events (`Start`/`Finish`) must NOT count as
@@ -671,4 +689,15 @@ mod tests {
         let refusal = super::empty_finish_diagnostic_message(StopReason::Refusal);
         assert!(refusal.to_lowercase().contains("refused"));
     }
+
+    #[test]
+    fn empty_finish_diagnostic_error_has_feedback_resolution() {
+        let error = super::empty_finish_diagnostic_error(StopReason::EndTurn);
+
+        let resolution = error
+            .resolution
+            .expect("empty-finish classified errors must include a resolution");
+        assert_eq!(resolution.kind, AgentErrorResolutionKind::SendFeedback);
+        assert_eq!(resolution.target, Some(AgentErrorResolutionTarget::Feedback));
+    }
 }

crates/aionui-ai-agent/src/manager/aionrs/agent.rs

@@ -21,6 +21,7 @@ use crate::agent_runtime::AgentRuntime;
 use crate::capability::backend_output_sink::BackendOutputSink;
 use crate::capability::backend_protocol_sink::BackendProtocolSink;
 use crate::protocol::events::AgentStreamEvent;
+use crate::protocol::send_error::AgentSendError;
 use crate::types::{AionrsResolvedConfig, SendMessageData};
 
 pub struct AionrsAgentManager {
@@ -185,7 +186,7 @@ impl crate::agent_task::IAgentTask for AionrsAgentManager {
         self.runtime.subscribe()
     }
 
-    async fn send_message(&self, data: SendMessageData) -> Result<(), AppError> {
+    async fn send_message(&self, data: SendMessageData) -> Result<(), AgentSendError> {
         let started_at = now_ms();
         info!(
             conversation_id = %self.runtime.conversation_id(),
@@ -230,9 +231,10 @@ impl crate::agent_task::IAgentTask for AionrsAgentManager {
                     error = %ErrorChain(&e),
                     "Aionrs engine.run() failed, emitting Error+Finish"
                 );
-                self.runtime.emit_error(error_msg.clone());
+                let send_error = aionrs_engine_error_to_send_error(error_msg);
+                self.runtime.emit_error_data(send_error.stream_error().clone());
                 self.runtime.emit_finish(None);
-                Err(AppError::Internal(error_msg))
+                Err(send_error)
             }
             None => {
                 self.runtime.emit_error("Stopped by user");
@@ -353,6 +355,14 @@ fn parse_session_mode(s: &str) -> SessionMode {
     }
 }
 
+fn aionrs_engine_error_to_send_error(error_msg: String) -> AgentSendError {
+    let lower = error_msg.to_ascii_lowercase();
+    if lower.contains("provider error") || lower.contains("provider:") {
+        return AgentSendError::from_app_error(AppError::BadGateway(error_msg));
+    }
+    AgentSendError::from_app_error(AppError::Internal(error_msg))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -488,4 +498,22 @@ mod tests {
             other => panic!("Expected Finish, got {:?}", other),
         }
     }
+
+    #[test]
+    fn aionrs_provider_connection_error_is_user_llm_provider_error() {
+        let send_error = aionrs_engine_error_to_send_error(
+            "Aionrs agent error: Provider error: Connection error: Signable request error: failed to create canonical request"
+                .to_owned(),
+        );
+
+        assert_eq!(
+            send_error.code(),
+            Some(aionui_api_types::AgentErrorCode::UserLlmProviderConfigError)
+        );
+        assert_eq!(
+            send_error.ownership(),
+            Some(aionui_api_types::AgentErrorOwnership::UserLlmProvider)
+        );
+        assert_eq!(send_error.stream_error().retryable, Some(false));
+    }
 }

crates/aionui-ai-agent/src/manager/nanobot/agent.rs

@@ -12,6 +12,7 @@ use crate::agent_runtime::AgentRuntime;
 use crate::capability::cli_process::CliAgentProcess;
 use crate::manager::process_registry::register_session_process;
 use crate::protocol::events::AgentStreamEvent;
+use crate::protocol::send_error::AgentSendError;
 use crate::types::SendMessageData;
 use std::path::PathBuf;
 
@@ -187,7 +188,7 @@ impl crate::agent_task::IAgentTask for NanobotAgentManager {
         self.runtime.subscribe()
     }
 
-    async fn send_message(&self, data: SendMessageData) -> Result<(), AppError> {
+    async fn send_message(&self, data: SendMessageData) -> Result<(), AgentSendError> {
         self.runtime.bump_activity();
 
         {
@@ -205,7 +206,19 @@ impl crate::agent_task::IAgentTask for NanobotAgentManager {
             }
         });
 
-        self.process.send(&payload).await
+        match self.process.send(&payload).await {
+            Ok(()) => Ok(()),
+            Err(err) => {
+                error!(
+                    conversation_id = %self.runtime.conversation_id(),
+                    error = %ErrorChain(&err),
+                    "Nanobot send_message failed, emitting Error"
+                );
+                let send_error = AgentSendError::from_app_error(err);
+                self.runtime.emit_error_data(send_error.stream_error().clone());
+                Err(send_error)
+            }
+        }
     }
 
     async fn cancel(&self) -> Result<(), AppError> {

crates/aionui-ai-agent/src/manager/openclaw/agent/mod.rs

@@ -11,6 +11,7 @@ use crate::agent_runtime::AgentRuntime;
 use crate::capability::cli_process::CliAgentProcess;
 use crate::manager::process_registry::register_session_process;
 use crate::protocol::events::AgentStreamEvent;
+use crate::protocol::send_error::AgentSendError;
 use crate::types::SendMessageData;
 use aionui_api_types::OpenClawBuildExtra;
 
@@ -403,7 +404,7 @@ impl crate::agent_task::IAgentTask for OpenClawAgentManager {
         self.runtime.subscribe()
     }
 
-    async fn send_message(&self, data: SendMessageData) -> Result<(), AppError> {
+    async fn send_message(&self, data: SendMessageData) -> Result<(), AgentSendError> {
         self.runtime.bump_activity();
 
         let is_first = {
@@ -419,17 +420,20 @@ impl crate::agent_task::IAgentTask for OpenClawAgentManager {
             text_state.reset_for_new_turn();
         }
 
-        let result = self.do_send_message(is_first, data).await;
-        if let Err(ref e) = result {
-            error!(
-                conversation_id = %self.runtime.conversation_id(),
-                error = %ErrorChain(e),
-                "OpenClaw send_message failed, emitting Error+Finish"
-            );
-            self.runtime.emit_error(format!("OpenClaw send failed: {e}"));
-            self.runtime.emit_finish(None);
+        match self.do_send_message(is_first, data).await {
+            Ok(()) => Ok(()),
+            Err(err) => {
+                error!(
+                    conversation_id = %self.runtime.conversation_id(),
+                    error = %ErrorChain(&err),
+                    "OpenClaw send_message failed, emitting Error+Finish"
+                );
+                let send_error = AgentSendError::from_app_error(err);
+                self.runtime.emit_error_data(send_error.stream_error().clone());
+                self.runtime.emit_finish(None);
+                Err(send_error)
+            }
         }
-        result
     }
 
     async fn cancel(&self) -> Result<(), AppError> {

crates/aionui-ai-agent/src/manager/openclaw/event_mapper.rs

@@ -133,10 +133,7 @@ fn map_chat_event(
         }
         ChatEventState::Error => {
             let msg = chat.error_message.unwrap_or_else(|| "Unknown chat error".into());
-            events.push(AgentStreamEvent::Error(ErrorEventData {
-                message: msg,
-                code: None,
-            }));
+            events.push(AgentStreamEvent::Error(ErrorEventData::legacy(msg, None)));
             text_state.turn_active = false;
         }
     }