warn/errorのTUIへの通知ルート

crates/pod/src/agents_md.rs

@@ -18,21 +18,43 @@ pub(crate) const AGENTS_MD_LIMIT: usize = 64 * 1024;
 
 const TRUNCATION_NOTICE: &str = "\n\n[truncated: AGENTS.md exceeded 64KB limit]";
 
-/// Read `AGENTS.md` from `cwd` if present. Returns `None` for "absent or
-/// unreadable"; all non-fatal problems are logged via `tracing::warn!`.
+/// Outcome of an `AGENTS.md` ingestion attempt.
 ///
-/// - Absent: `None`, no warn.
-/// - Over limit: first 64KB (UTF-8 char boundary) + truncation notice, warn.
-/// - Non-UTF-8 or I/O error: `None`, warn.
-pub(crate) fn read_agents_md(cwd: &Path) -> Option<String> {
+/// `body` carries the text that should be handed to the template
+/// engine (if any); `warnings` are short human-readable messages that
+/// Pod forwards to the user-facing notification channel. The caller
+/// also gets `tracing::warn!` lines for the developer log.
+pub(crate) struct AgentsMdResult {
+    pub body: Option<String>,
+    pub warnings: Vec<String>,
+}
+
+/// Read `AGENTS.md` from `cwd` if present. All non-fatal problems are
+/// both logged via `tracing::warn!` (developer-facing) and surfaced
+/// via `AgentsMdResult::warnings` (user-facing).
+///
+/// - Absent: `body = None`, no warning.
+/// - Over limit: first 64KB (UTF-8 char boundary) + truncation notice, warning.
+/// - Non-UTF-8 or I/O error: `body = None`, warning.
+pub(crate) fn read_agents_md(cwd: &Path) -> AgentsMdResult {
     let path = cwd.join("AGENTS.md");
+    let mut warnings = Vec::new();
 
     let file = match File::open(&path) {
         Ok(f) => f,
-        Err(e) if e.kind() == ErrorKind::NotFound => return None,
+        Err(e) if e.kind() == ErrorKind::NotFound => {
+            return AgentsMdResult {
+                body: None,
+                warnings,
+            };
+        }
         Err(e) => {
             warn!(path = %path.display(), error = %e, "failed to open AGENTS.md");
-            return None;
+            warnings.push(format!("failed to open AGENTS.md ({}): {}", path.display(), e));
+            return AgentsMdResult {
+                body: None,
+                warnings,
+            };
         }
     };
 
@@ -42,7 +64,11 @@ pub(crate) fn read_agents_md(cwd: &Path) -> Option<String> {
     let read_limit = (AGENTS_MD_LIMIT as u64) + 1;
     if let Err(e) = file.take(read_limit).read_to_end(&mut buf) {
         warn!(path = %path.display(), error = %e, "failed to read AGENTS.md");
-        return None;
+        warnings.push(format!("failed to read AGENTS.md ({}): {}", path.display(), e));
+        return AgentsMdResult {
+            body: None,
+            warnings,
+        };
     }
 
     let truncated = buf.len() > AGENTS_MD_LIMIT;
@@ -69,7 +95,15 @@ pub(crate) fn read_agents_md(cwd: &Path) -> Option<String> {
         }
         Err(e) => {
             warn!(path = %path.display(), error = %e, "AGENTS.md is not valid UTF-8");
-            return None;
+            warnings.push(format!(
+                "AGENTS.md ({}) is not valid UTF-8: {}",
+                path.display(),
+                e
+            ));
+            return AgentsMdResult {
+                body: None,
+                warnings,
+            };
         }
     };
 
@@ -80,10 +114,18 @@ pub(crate) fn read_agents_md(cwd: &Path) -> Option<String> {
             limit = AGENTS_MD_LIMIT,
             "AGENTS.md exceeded size limit; truncating"
         );
+        warnings.push(format!(
+            "AGENTS.md ({}) exceeded {} bytes; the tail was truncated",
+            path.display(),
+            AGENTS_MD_LIMIT
+        ));
         text.push_str(TRUNCATION_NOTICE);
     }
 
-    Some(text)
+    AgentsMdResult {
+        body: Some(text),
+        warnings,
+    }
 }
 
 #[cfg(test)]
@@ -95,17 +137,16 @@ mod tests {
     #[test]
     fn absent_file_returns_none() {
         let dir = TempDir::new().unwrap();
-        assert!(read_agents_md(dir.path()).is_none());
+        assert!(read_agents_md(dir.path()).body.is_none());
     }
 
     #[test]
     fn reads_small_file_verbatim() {
         let dir = TempDir::new().unwrap();
         fs::write(dir.path().join("AGENTS.md"), "# hello\nworld").unwrap();
-        assert_eq!(
-            read_agents_md(dir.path()).as_deref(),
-            Some("# hello\nworld"),
-        );
+        let result = read_agents_md(dir.path());
+        assert_eq!(result.body.as_deref(), Some("# hello\nworld"));
+        assert!(result.warnings.is_empty());
     }
 
     #[test]
@@ -114,11 +155,13 @@ mod tests {
         let body = "a".repeat(AGENTS_MD_LIMIT + 1024);
         fs::write(dir.path().join("AGENTS.md"), &body).unwrap();
 
-        let got = read_agents_md(dir.path()).expect("some");
+        let result = read_agents_md(dir.path());
+        let got = result.body.expect("some");
         assert!(got.ends_with(TRUNCATION_NOTICE));
         let prefix = got.strip_suffix(TRUNCATION_NOTICE).unwrap();
         assert_eq!(prefix.len(), AGENTS_MD_LIMIT);
         assert!(prefix.chars().all(|c| c == 'a'));
+        assert_eq!(result.warnings.len(), 1);
     }
 
     #[test]
@@ -127,9 +170,11 @@ mod tests {
         let body = "a".repeat(AGENTS_MD_LIMIT);
         fs::write(dir.path().join("AGENTS.md"), &body).unwrap();
 
-        let got = read_agents_md(dir.path()).expect("some");
+        let result = read_agents_md(dir.path());
+        let got = result.body.expect("some");
         assert_eq!(got.len(), AGENTS_MD_LIMIT);
         assert!(!got.contains("truncated"));
+        assert!(result.warnings.is_empty());
     }
 
     #[test]
@@ -142,12 +187,23 @@ mod tests {
         body.push_str(&"b".repeat(128));
         fs::write(dir.path().join("AGENTS.md"), &body).unwrap();
 
-        let got = read_agents_md(dir.path()).expect("some");
+        let result = read_agents_md(dir.path());
+        let got = result.body.expect("some");
         assert!(got.ends_with(TRUNCATION_NOTICE));
         let prefix = got.strip_suffix(TRUNCATION_NOTICE).unwrap();
         // The partial 'あ' must have been dropped, leaving only the ASCII prefix.
         assert_eq!(prefix.len(), AGENTS_MD_LIMIT - 1);
         assert!(prefix.chars().all(|c| c == 'a'));
+        assert_eq!(result.warnings.len(), 1);
+    }
+
+    #[test]
+    fn non_utf8_surfaces_warning() {
+        let dir = TempDir::new().unwrap();
+        fs::write(dir.path().join("AGENTS.md"), [0xff, 0xfe, 0xfd]).unwrap();
+        let result = read_agents_md(dir.path());
+        assert!(result.body.is_none());
+        assert_eq!(result.warnings.len(), 1);
     }
 
     #[test]
@@ -159,7 +215,7 @@ mod tests {
         let dir = TempDir::new().unwrap();
         let body = vec![0xffu8; AGENTS_MD_LIMIT + 1024];
         fs::write(dir.path().join("AGENTS.md"), body).unwrap();
-        assert!(read_agents_md(dir.path()).is_none());
+        assert!(read_agents_md(dir.path()).body.is_none());
     }
 
     #[test]
@@ -167,6 +223,6 @@ mod tests {
         let dir = TempDir::new().unwrap();
         // Invalid UTF-8 start byte.
         fs::write(dir.path().join("AGENTS.md"), [0xff, 0xfe, 0xfd]).unwrap();
-        assert!(read_agents_md(dir.path()).is_none());
+        assert!(read_agents_md(dir.path()).body.is_none());
     }
 }

crates/pod/src/controller.rs

@@ -6,11 +6,12 @@ use llm_worker::llm_client::client::LlmClient;
 use session_store::Store;
 use tokio::sync::{broadcast, mpsc};
 
+use crate::notifier::Notifier;
 use crate::pod::{Pod, PodError, PodRunResult};
 use crate::runtime_dir::RuntimeDir;
 use crate::shared_state::{PodSharedState, PodStatus};
 use crate::socket_server::SocketServer;
-use protocol::{ErrorCode, Event, Method, RunResult, TurnResult};
+use protocol::{ErrorCode, Event, Method, NotificationLevel, NotificationSource, RunResult, TurnResult};
 
 // ---------------------------------------------------------------------------
 // PodHandle — client-facing, Clone-able
@@ -22,6 +23,7 @@ pub struct PodHandle {
     event_tx: broadcast::Sender<Event>,
     pub shared_state: Arc<PodSharedState>,
     pub runtime_dir: Arc<RuntimeDir>,
+    pub notifier: Notifier,
 }
 
 impl PodHandle {
@@ -37,6 +39,11 @@ impl PodHandle {
     pub fn send_event(&self, event: Event) -> Result<usize, broadcast::error::SendError<Event>> {
         self.event_tx.send(event)
     }
+
+    /// Emit a user-facing notification. Thin wrapper over `Notifier::notify`.
+    pub fn notify(&self, level: NotificationLevel, source: NotificationSource, message: String) {
+        self.notifier.notify(level, source, message);
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -56,6 +63,7 @@ impl PodController {
     {
         let (method_tx, mut method_rx) = mpsc::channel::<Method>(32);
         let (event_tx, _) = broadcast::channel::<Event>(256);
+        let notifier = Notifier::new(event_tx.clone());
 
         let manifest_toml = toml::to_string_pretty(pod.manifest()).unwrap_or_default();
         let greeting = build_greeting(&pod);
@@ -78,8 +86,14 @@ impl PodController {
             event_tx: event_tx.clone(),
             shared_state: shared_state.clone(),
             runtime_dir: runtime_dir.clone(),
+            notifier: notifier.clone(),
         };
 
+        // Hand the notifier to the Pod so internal operations (compaction,
+        // AGENTS.md ingestion during the first turn) can emit user-facing
+        // notifications on the same channel.
+        pod.attach_notifier(notifier.clone());
+
         // Start socket server (lives as a background task, cleaned up on drop via RuntimeDir)
         let _socket_server = SocketServer::start(&handle).await?;
         // Keep the server alive by moving it into the controller task
@@ -163,6 +177,15 @@ impl PodController {
                 });
             });
 
+            let notifier_for_worker = notifier.clone();
+            worker.on_warning(move |message| {
+                notifier_for_worker.notify(
+                    NotificationLevel::Warn,
+                    NotificationSource::Worker,
+                    message.to_owned(),
+                );
+            });
+
             // Register the builtin file-manipulation tools (Read / Write /
             // Edit / Glob / Grep). `ScopedFs` carries the pod-lifetime
             // scope/pwd; `Tracker` is session-scoped — a fresh instance per
@@ -215,6 +238,11 @@ impl PodController {
                         if new_status == PodStatus::Idle {
                             if let Err(e) = pod.try_post_run_compact().await {
                                 tracing::warn!(error = %e, "Post-run compaction error");
+                                notifier.notify(
+                                    NotificationLevel::Warn,
+                                    NotificationSource::Compactor,
+                                    format!("post-run compaction error: {e}"),
+                                );
                             }
                         }
 
@@ -249,6 +277,11 @@ impl PodController {
                         if new_status == PodStatus::Idle {
                             if let Err(e) = pod.try_post_run_compact().await {
                                 tracing::warn!(error = %e, "Post-run compaction error");
+                                notifier.notify(
+                                    NotificationLevel::Warn,
+                                    NotificationSource::Compactor,
+                                    format!("post-run compaction error: {e}"),
+                                );
                             }
                         }
 

crates/pod/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod controller;
 pub mod hook;
+pub mod notifier;
 pub mod runtime_dir;
 pub mod shared_state;
 pub mod socket_server;
@@ -17,6 +18,7 @@ mod usage_tracker;
 pub use token_counter::{EstimateSource, SplitPoint, TokenEstimate};
 
 pub use controller::{PodController, PodHandle};
+pub use notifier::Notifier;
 pub use hook::{Hook, HookEventKind, HookRegistryBuilder};
 pub use manifest::{PodManifest, ProviderConfig, ProviderKind, Scope};
 pub use pod::{Pod, PodError, PodRunResult, apply_worker_manifest};

crates/pod/src/notifier.rs

@@ -0,0 +1,191 @@
+//! User-facing notification channel for Pod → client.
+//!
+//! Separate from `tracing` (which is for developer logs). Notifications
+//! are short human-readable messages the Pod layer wants a client to
+//! see — for example "compaction failed", "tool output truncated".
+//!
+//! Each notification is broadcast on the shared `Event` channel and
+//! also appended to an in-memory buffer so that clients connecting
+//! after the fact still see everything emitted during the session.
+
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use tokio::sync::broadcast;
+
+use protocol::{Event, Notification, NotificationLevel, NotificationSource};
+
+/// Upper bound on buffered notifications. When exceeded, the oldest
+/// entries are discarded so a long-running session cannot leak
+/// memory through a pathological loop of recurring notifications
+/// (e.g. compaction failing every turn).
+const MAX_BUFFERED_NOTIFICATIONS: usize = 512;
+
+#[derive(Clone)]
+pub struct Notifier {
+    inner: Arc<Inner>,
+}
+
+struct Inner {
+    event_tx: broadcast::Sender<Event>,
+    buffer: Mutex<VecDeque<Notification>>,
+}
+
+impl Notifier {
+    pub fn new(event_tx: broadcast::Sender<Event>) -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                event_tx,
+                buffer: Mutex::new(VecDeque::with_capacity(MAX_BUFFERED_NOTIFICATIONS)),
+            }),
+        }
+    }
+
+    /// Record and broadcast a notification.
+    ///
+    /// The broadcast may have no subscribers (e.g. during Pod
+    /// construction before any client has connected); the buffer
+    /// guarantees the message is still delivered once a client
+    /// attaches.
+    ///
+    /// The buffer mutex is held across `broadcast::send` to make
+    /// `subscribe_with_snapshot` race-free — a client that snapshots
+    /// the buffer while holding the same lock sees every notification
+    /// exactly once: older ones from the snapshot, newer ones from
+    /// the freshly-subscribed receiver.
+    pub fn notify(&self, level: NotificationLevel, source: NotificationSource, message: String) {
+        let notification = Notification {
+            level,
+            source,
+            message,
+            timestamp_ms: now_ms(),
+        };
+        if let Ok(mut buf) = self.inner.buffer.lock() {
+            if buf.len() >= MAX_BUFFERED_NOTIFICATIONS {
+                buf.pop_front();
+            }
+            buf.push_back(notification.clone());
+            let _ = self
+                .inner
+                .event_tx
+                .send(Event::Notification(notification));
+        }
+    }
+
+    /// Subscribe and atomically snapshot the current buffer.
+    ///
+    /// The returned snapshot contains notifications emitted before
+    /// this call; the receiver will deliver notifications emitted
+    /// after. A notification cannot appear in both.
+    pub fn subscribe_with_snapshot(&self) -> (Vec<Notification>, broadcast::Receiver<Event>) {
+        let buf = self
+            .inner
+            .buffer
+            .lock()
+            .expect("notifier buffer mutex poisoned");
+        let rx = self.inner.event_tx.subscribe();
+        let snapshot: Vec<Notification> = buf.iter().cloned().collect();
+        (snapshot, rx)
+    }
+}
+
+fn now_ms() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn notify_broadcasts_to_existing_subscriber() {
+        let (tx, _keep) = broadcast::channel::<Event>(8);
+        let notifier = Notifier::new(tx);
+        let (_snapshot, mut rx) = notifier.subscribe_with_snapshot();
+
+        notifier.notify(
+            NotificationLevel::Warn,
+            NotificationSource::Compactor,
+            "test message".into(),
+        );
+
+        match rx.try_recv() {
+            Ok(Event::Notification(n)) => assert_eq!(n.message, "test message"),
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn late_subscriber_sees_earlier_notifications_via_snapshot() {
+        let (tx, _keep) = broadcast::channel::<Event>(8);
+        let notifier = Notifier::new(tx);
+
+        notifier.notify(
+            NotificationLevel::Error,
+            NotificationSource::Pod,
+            "first".into(),
+        );
+        notifier.notify(
+            NotificationLevel::Warn,
+            NotificationSource::AgentsMd,
+            "second".into(),
+        );
+
+        let (snapshot, mut rx) = notifier.subscribe_with_snapshot();
+        assert_eq!(snapshot.len(), 2);
+        assert_eq!(snapshot[0].message, "first");
+        assert_eq!(snapshot[1].message, "second");
+        assert!(rx.try_recv().is_err()); // nothing pending on the receiver
+    }
+
+    #[test]
+    fn buffer_discards_oldest_past_cap() {
+        let (tx, _keep) = broadcast::channel::<Event>(1024);
+        let notifier = Notifier::new(tx);
+
+        for i in 0..(MAX_BUFFERED_NOTIFICATIONS + 50) {
+            notifier.notify(
+                NotificationLevel::Warn,
+                NotificationSource::Worker,
+                format!("msg-{i}"),
+            );
+        }
+
+        let (snapshot, _rx) = notifier.subscribe_with_snapshot();
+        assert_eq!(snapshot.len(), MAX_BUFFERED_NOTIFICATIONS);
+        // First 50 were evicted; the oldest remaining is msg-50.
+        assert_eq!(snapshot.first().unwrap().message, "msg-50");
+        let last = format!("msg-{}", MAX_BUFFERED_NOTIFICATIONS + 49);
+        assert_eq!(snapshot.last().unwrap().message, last);
+    }
+
+    #[test]
+    fn subscribe_snapshot_and_live_do_not_overlap() {
+        let (tx, _keep) = broadcast::channel::<Event>(8);
+        let notifier = Notifier::new(tx);
+
+        notifier.notify(
+            NotificationLevel::Warn,
+            NotificationSource::Worker,
+            "historic".into(),
+        );
+        let (snapshot, mut rx) = notifier.subscribe_with_snapshot();
+        notifier.notify(
+            NotificationLevel::Error,
+            NotificationSource::Worker,
+            "live".into(),
+        );
+
+        assert_eq!(snapshot.len(), 1);
+        assert_eq!(snapshot[0].message, "historic");
+        match rx.try_recv() {
+            Ok(Event::Notification(n)) => assert_eq!(n.message, "live"),
+            other => panic!("unexpected: {other:?}"),
+        }
+        assert!(rx.try_recv().is_err());
+    }
+}

crates/pod/src/pod.rs

@@ -21,8 +21,10 @@ use crate::hook::{
     PreToolCall,
 };
 use crate::hook_interceptor::HookInterceptor;
+use crate::notifier::Notifier;
 use crate::system_prompt::{SystemPromptContext, SystemPromptError, SystemPromptTemplate};
 use crate::usage_tracker::UsageTracker;
+use protocol::{NotificationLevel, NotificationSource};
 use async_trait::async_trait;
 use llm_worker::interceptor::PreRequestAction;
 
@@ -97,6 +99,9 @@ pub struct Pod<C: LlmClient, St: Store> {
     /// `Some` until `ensure_system_prompt_materialized` renders it once,
     /// then `None` forever — including after compaction.
     system_prompt_template: Option<SystemPromptTemplate>,
+    /// User-facing notification sink attached by the Controller at
+    /// spawn time. `None` in tests / direct `Pod::new` usage.
+    notifier: Option<Notifier>,
 }
 
 impl<C: LlmClient, St: Store> Pod<C, St> {
@@ -137,6 +142,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             usage_history: Arc::new(Mutex::new(Vec::<UsageRecord>::new())),
             tracker: None,
             system_prompt_template: None,
+            notifier: None,
         };
         pod.apply_prune_from_manifest();
         Ok(pod)
@@ -185,6 +191,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             usage_history: Arc::new(Mutex::new(state.usage_history)),
             tracker: None,
             system_prompt_template: None,
+            notifier: None,
         };
         pod.apply_prune_from_manifest();
         Ok(pod)
@@ -275,6 +282,21 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         self.tracker.as_ref()
     }
 
+    /// Attach a user-facing notification sink.
+    ///
+    /// Called by the Controller immediately after spawning so that
+    /// Pod-internal operations (compaction failures, AGENTS.md
+    /// ingestion warnings) can surface messages to connected clients.
+    pub fn attach_notifier(&mut self, notifier: Notifier) {
+        self.notifier = Some(notifier);
+    }
+
+    fn notify(&self, level: NotificationLevel, source: NotificationSource, message: String) {
+        if let Some(n) = self.notifier.as_ref() {
+            n.notify(level, source, message);
+        }
+    }
+
     // --- Hook registration ---
 
     fn assert_hooks_open(&self) {
@@ -392,6 +414,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
         let Some(template) = self.system_prompt_template.take() else {
             return Ok(());
         };
+        let notifier = self.notifier.clone();
         let worker = self.worker.as_mut().expect("worker present");
         // Materialise any pending tool factories so the template sees the
         // full list of tool names. Redundant with the flush inside
@@ -404,7 +427,17 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             .map(|d| d.name)
             .collect();
         let mut files = std::collections::BTreeMap::new();
-        if let Some(body) = read_agents_md(&self.pwd) {
+        let agents_md = read_agents_md(&self.pwd);
+        for warning in agents_md.warnings {
+            if let Some(n) = notifier.as_ref() {
+                n.notify(
+                    NotificationLevel::Warn,
+                    NotificationSource::AgentsMd,
+                    warning,
+                );
+            }
+        }
+        if let Some(body) = agents_md.body {
             files.insert("agents_md".to_string(), body);
         }
         let ctx = SystemPromptContext {
@@ -553,6 +586,11 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
                 }
                 Err(e) => {
                     warn!(error = %e, "Compaction failed during run");
+                    self.notify(
+                        NotificationLevel::Error,
+                        NotificationSource::Compactor,
+                        format!("mid-run compaction failed: {e}"),
+                    );
                     if let Some(ref state) = self.compact_state {
                         state.record_compact_failure();
                     }
@@ -583,6 +621,11 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             }
             Err(e) => {
                 warn!(error = %e, "Proactive post-run compaction failed");
+                self.notify(
+                    NotificationLevel::Warn,
+                    NotificationSource::Compactor,
+                    format!("post-run compaction failed: {e}"),
+                );
                 state.record_compact_failure();
                 Ok(())
             }
@@ -830,6 +873,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             usage_history: Arc::new(Mutex::new(Vec::new())),
             tracker: None,
             system_prompt_template,
+            notifier: None,
         };
         pod.apply_prune_from_manifest();
         Ok(pod)

crates/pod/src/socket_server.rs

@@ -61,7 +61,21 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
     let (reader, writer) = stream.into_split();
     let mut reader = JsonLineReader::new(reader);
     let mut writer = JsonLineWriter::new(writer);
-    let mut rx = handle.subscribe();
+
+    // Atomically subscribe and snapshot buffered notifications so that
+    // warnings emitted before this client connected are replayed
+    // exactly once — they appear in the snapshot, and any notification
+    // arriving afterwards reaches us through `rx`.
+    let (notification_snapshot, mut rx) = handle.notifier.subscribe_with_snapshot();
+    for notification in notification_snapshot {
+        if writer
+            .write(&Event::Notification(notification))
+            .await
+            .is_err()
+        {
+            return;
+        }
+    }
 
     loop {
         tokio::select! {