| 92 | } |
| 93 | |
| 94 | pub(super) fn spawn_forwarder( |
| 95 | self, |
| 96 | mut runtime_rx: mpsc::Receiver<AgentEvent>, |
| 97 | tx: mpsc::Sender<AgentEvent>, |
| 98 | ) -> JoinHandle<()> { |
| 99 | tokio::spawn(async move { |
| 100 | while let Some(event) = runtime_rx.recv().await { |
| 101 | self.observe(&event).await; |
| 102 | let send_ok = tx.send(event.clone()).await.is_ok(); |
| 103 | if !send_ok { |
| 104 | // Receiver dropped or buffer full; preserve the existing stream contract |
| 105 | // by stopping instead of silently dropping later terminal events. |
| 106 | tracing::warn!("stream forwarder: receiver dropped, stopping event forward"); |
| 107 | break; |
| 108 | } |
| 109 | } |
| 110 | }) |
| 111 | } |
| 112 | |
| 113 | pub(super) async fn observe(&self, event: &AgentEvent) { |
| 114 | let _ = self |