zeph_core/
channel.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4/// Typed error for channel operations.
5#[derive(Debug, thiserror::Error)]
6pub enum ChannelError {
7    /// Underlying I/O failure.
8    #[error("I/O error: {0}")]
9    Io(#[from] std::io::Error),
10
11    /// Channel closed (mpsc send/recv failure).
12    #[error("channel closed")]
13    ChannelClosed,
14
15    /// Confirmation dialog cancelled.
16    #[error("confirmation cancelled")]
17    ConfirmCancelled,
18
19    /// Catch-all for provider-specific errors.
20    #[error("{0}")]
21    Other(String),
22}
23
24/// All fields that describe a tool-start event sent to a channel.
25#[derive(Debug)]
26pub struct ToolStartEvent<'a> {
27    pub tool_name: &'a str,
28    pub tool_call_id: &'a str,
29    pub params: Option<serde_json::Value>,
30    pub parent_tool_use_id: Option<String>,
31}
32
33/// All fields that describe a completed tool-output event sent to a channel.
34#[derive(Debug)]
35pub struct ToolOutputEvent<'a> {
36    pub tool_name: &'a str,
37    pub body: &'a str,
38    pub diff: Option<crate::DiffData>,
39    pub filter_stats: Option<String>,
40    pub kept_lines: Option<Vec<usize>>,
41    pub locations: Option<Vec<String>>,
42    pub tool_call_id: &'a str,
43    pub is_error: bool,
44    pub parent_tool_use_id: Option<String>,
45    pub raw_response: Option<serde_json::Value>,
46    pub started_at: Option<std::time::Instant>,
47}
48
49/// Kind of binary attachment on an incoming message.
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub enum AttachmentKind {
52    Audio,
53    Image,
54    Video,
55    File,
56}
57
58/// Binary attachment carried by a [`ChannelMessage`].
59#[derive(Debug, Clone)]
60pub struct Attachment {
61    pub kind: AttachmentKind,
62    pub data: Vec<u8>,
63    pub filename: Option<String>,
64}
65
66/// Incoming message from a channel.
67#[derive(Debug, Clone)]
68pub struct ChannelMessage {
69    pub text: String,
70    pub attachments: Vec<Attachment>,
71}
72
73/// Bidirectional communication channel for the agent.
74pub trait Channel: Send {
75    /// Receive the next message. Returns `None` on EOF or shutdown.
76    ///
77    /// # Errors
78    ///
79    /// Returns an error if the underlying I/O fails.
80    fn recv(&mut self)
81    -> impl Future<Output = Result<Option<ChannelMessage>, ChannelError>> + Send;
82
83    /// Non-blocking receive. Returns `None` if no message is immediately available.
84    fn try_recv(&mut self) -> Option<ChannelMessage> {
85        None
86    }
87
88    /// Whether `/exit` and `/quit` commands should terminate the agent loop.
89    ///
90    /// Returns `false` for persistent server-side channels (e.g. Telegram) where
91    /// breaking the loop would not meaningfully exit from the user's perspective.
92    fn supports_exit(&self) -> bool {
93        true
94    }
95
96    /// Send a text response.
97    ///
98    /// # Errors
99    ///
100    /// Returns an error if the underlying I/O fails.
101    fn send(&mut self, text: &str) -> impl Future<Output = Result<(), ChannelError>> + Send;
102
103    /// Send a partial chunk of streaming response.
104    ///
105    /// # Errors
106    ///
107    /// Returns an error if the underlying I/O fails.
108    fn send_chunk(&mut self, chunk: &str) -> impl Future<Output = Result<(), ChannelError>> + Send;
109
110    /// Flush any buffered chunks.
111    ///
112    /// # Errors
113    ///
114    /// Returns an error if the underlying I/O fails.
115    fn flush_chunks(&mut self) -> impl Future<Output = Result<(), ChannelError>> + Send;
116
117    /// Send a typing indicator. No-op by default.
118    ///
119    /// # Errors
120    ///
121    /// Returns an error if the underlying I/O fails.
122    fn send_typing(&mut self) -> impl Future<Output = Result<(), ChannelError>> + Send {
123        async { Ok(()) }
124    }
125
126    /// Send a status label (shown as spinner text in TUI). No-op by default.
127    ///
128    /// # Errors
129    ///
130    /// Returns an error if the underlying I/O fails.
131    fn send_status(
132        &mut self,
133        _text: &str,
134    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
135        async { Ok(()) }
136    }
137
138    /// Send a thinking/reasoning token chunk. No-op by default.
139    ///
140    /// # Errors
141    ///
142    /// Returns an error if the underlying I/O fails.
143    fn send_thinking_chunk(
144        &mut self,
145        _chunk: &str,
146    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
147        async { Ok(()) }
148    }
149
150    /// Notify channel of queued message count. No-op by default.
151    ///
152    /// # Errors
153    ///
154    /// Returns an error if the underlying I/O fails.
155    fn send_queue_count(
156        &mut self,
157        _count: usize,
158    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
159        async { Ok(()) }
160    }
161
162    /// Send token usage after an LLM call. No-op by default.
163    ///
164    /// # Errors
165    ///
166    /// Returns an error if the underlying I/O fails.
167    fn send_usage(
168        &mut self,
169        _input_tokens: u64,
170        _output_tokens: u64,
171        _context_window: u64,
172    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
173        async { Ok(()) }
174    }
175
176    /// Send diff data for a tool result. No-op by default (TUI overrides).
177    ///
178    /// # Errors
179    ///
180    /// Returns an error if the underlying I/O fails.
181    fn send_diff(
182        &mut self,
183        _diff: crate::DiffData,
184    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
185        async { Ok(()) }
186    }
187
188    /// Announce that a tool call is starting.
189    ///
190    /// Emitted before execution begins so the transport layer can send an
191    /// `InProgress` status to the peer before the result arrives.
192    /// No-op by default.
193    ///
194    /// # Errors
195    ///
196    /// Returns an error if the underlying I/O fails.
197    fn send_tool_start(
198        &mut self,
199        _event: ToolStartEvent<'_>,
200    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
201        async { Ok(()) }
202    }
203
204    /// Send a complete tool output with optional diff and filter stats atomically.
205    ///
206    /// `body` is the raw tool output content (no header). The default implementation
207    /// formats it with `[tool output: <name>]` prefix for human-readable channels.
208    /// Structured channels (e.g. `LoopbackChannel`) override this to emit a typed event
209    /// so consumers can access `tool_name` and `body` as separate fields.
210    ///
211    /// # Errors
212    ///
213    /// Returns an error if the underlying I/O fails.
214    fn send_tool_output(
215        &mut self,
216        event: ToolOutputEvent<'_>,
217    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
218        let formatted = crate::agent::format_tool_output(event.tool_name, event.body);
219        async move { self.send(&formatted).await }
220    }
221
222    /// Request user confirmation for a destructive action. Returns `true` if confirmed.
223    /// Default: auto-confirm (for headless/test scenarios).
224    ///
225    /// # Errors
226    ///
227    /// Returns an error if the underlying I/O fails.
228    fn confirm(
229        &mut self,
230        _prompt: &str,
231    ) -> impl Future<Output = Result<bool, ChannelError>> + Send {
232        async { Ok(true) }
233    }
234
235    /// Signal the non-default stop reason to the consumer before flushing.
236    ///
237    /// Called by the agent loop immediately before `flush_chunks()` when a
238    /// truncation or turn-limit condition is detected. No-op by default.
239    ///
240    /// # Errors
241    ///
242    /// Returns an error if the underlying I/O fails.
243    fn send_stop_hint(
244        &mut self,
245        _hint: StopHint,
246    ) -> impl Future<Output = Result<(), ChannelError>> + Send {
247        async { Ok(()) }
248    }
249}
250
251/// Reason why the agent turn ended — carried by [`LoopbackEvent::Stop`].
252///
253/// Emitted by the agent loop immediately before `Flush` when a non-default
254/// terminal condition is detected. Consumers (e.g. the ACP layer) map this to
255/// the protocol-level `StopReason`.
256#[derive(Debug, Clone, Copy, PartialEq, Eq)]
257pub enum StopHint {
258    /// The LLM response was cut off by the token limit.
259    MaxTokens,
260    /// The turn loop exhausted `max_turns` without a final text response.
261    MaxTurnRequests,
262}
263
264/// Data carried by a [`LoopbackEvent::ToolStart`] variant.
265#[derive(Debug, Clone)]
266pub struct ToolStartData {
267    pub tool_name: String,
268    pub tool_call_id: String,
269    /// Raw input parameters passed to the tool (e.g. `{"command": "..."}` for bash).
270    pub params: Option<serde_json::Value>,
271    /// Set when this tool call is made by a subagent; identifies the parent's `tool_call_id`.
272    pub parent_tool_use_id: Option<String>,
273    /// Wall-clock instant when the tool call was initiated; used to compute elapsed time.
274    pub started_at: std::time::Instant,
275}
276
277/// Data carried by a [`LoopbackEvent::ToolOutput`] variant.
278#[derive(Debug, Clone)]
279pub struct ToolOutputData {
280    pub tool_name: String,
281    pub display: String,
282    pub diff: Option<crate::DiffData>,
283    pub filter_stats: Option<String>,
284    pub kept_lines: Option<Vec<usize>>,
285    pub locations: Option<Vec<String>>,
286    pub tool_call_id: String,
287    pub is_error: bool,
288    /// Terminal ID for shell tool calls routed through the IDE terminal.
289    pub terminal_id: Option<String>,
290    /// Set when this tool output belongs to a subagent; identifies the parent's `tool_call_id`.
291    pub parent_tool_use_id: Option<String>,
292    /// Structured tool response payload for ACP intermediate `tool_call_update` notifications.
293    pub raw_response: Option<serde_json::Value>,
294    /// Wall-clock instant when the corresponding `ToolStart` was emitted; used for elapsed time.
295    pub started_at: Option<std::time::Instant>,
296}
297
298/// Events emitted by the agent side toward the A2A caller.
299#[derive(Debug, Clone)]
300pub enum LoopbackEvent {
301    Chunk(String),
302    Flush,
303    FullMessage(String),
304    Status(String),
305    /// Emitted immediately before tool execution begins.
306    ToolStart(Box<ToolStartData>),
307    ToolOutput(Box<ToolOutputData>),
308    /// Token usage from the last LLM turn.
309    Usage {
310        input_tokens: u64,
311        output_tokens: u64,
312        context_window: u64,
313    },
314    /// Generated session title (emitted after the first agent response).
315    SessionTitle(String),
316    /// Execution plan update.
317    Plan(Vec<(String, PlanItemStatus)>),
318    /// Thinking/reasoning token chunk from the LLM.
319    ThinkingChunk(String),
320    /// Non-default stop condition detected by the agent loop.
321    ///
322    /// Emitted immediately before `Flush`. When absent, the stop reason is `EndTurn`.
323    Stop(StopHint),
324}
325
326/// Status of a plan item, mirroring `acp::PlanEntryStatus`.
327#[derive(Debug, Clone)]
328pub enum PlanItemStatus {
329    Pending,
330    InProgress,
331    Completed,
332}
333
334/// Caller-side handle for sending input and receiving agent output.
335pub struct LoopbackHandle {
336    pub input_tx: tokio::sync::mpsc::Sender<ChannelMessage>,
337    pub output_rx: tokio::sync::mpsc::Receiver<LoopbackEvent>,
338    /// Shared cancel signal: notify to interrupt the agent's current operation.
339    pub cancel_signal: std::sync::Arc<tokio::sync::Notify>,
340}
341
342/// Headless channel bridging an A2A `TaskProcessor` with the agent loop.
343pub struct LoopbackChannel {
344    input_rx: tokio::sync::mpsc::Receiver<ChannelMessage>,
345    output_tx: tokio::sync::mpsc::Sender<LoopbackEvent>,
346}
347
348impl LoopbackChannel {
349    /// Create a linked `(LoopbackChannel, LoopbackHandle)` pair.
350    #[must_use]
351    pub fn pair(buffer: usize) -> (Self, LoopbackHandle) {
352        let (input_tx, input_rx) = tokio::sync::mpsc::channel(buffer);
353        let (output_tx, output_rx) = tokio::sync::mpsc::channel(buffer);
354        let cancel_signal = std::sync::Arc::new(tokio::sync::Notify::new());
355        (
356            Self {
357                input_rx,
358                output_tx,
359            },
360            LoopbackHandle {
361                input_tx,
362                output_rx,
363                cancel_signal,
364            },
365        )
366    }
367}
368
369impl Channel for LoopbackChannel {
370    fn supports_exit(&self) -> bool {
371        false
372    }
373
374    async fn recv(&mut self) -> Result<Option<ChannelMessage>, ChannelError> {
375        Ok(self.input_rx.recv().await)
376    }
377
378    async fn send(&mut self, text: &str) -> Result<(), ChannelError> {
379        self.output_tx
380            .send(LoopbackEvent::FullMessage(text.to_owned()))
381            .await
382            .map_err(|_| ChannelError::ChannelClosed)
383    }
384
385    async fn send_chunk(&mut self, chunk: &str) -> Result<(), ChannelError> {
386        self.output_tx
387            .send(LoopbackEvent::Chunk(chunk.to_owned()))
388            .await
389            .map_err(|_| ChannelError::ChannelClosed)
390    }
391
392    async fn flush_chunks(&mut self) -> Result<(), ChannelError> {
393        self.output_tx
394            .send(LoopbackEvent::Flush)
395            .await
396            .map_err(|_| ChannelError::ChannelClosed)
397    }
398
399    async fn send_status(&mut self, text: &str) -> Result<(), ChannelError> {
400        self.output_tx
401            .send(LoopbackEvent::Status(text.to_owned()))
402            .await
403            .map_err(|_| ChannelError::ChannelClosed)
404    }
405
406    async fn send_thinking_chunk(&mut self, chunk: &str) -> Result<(), ChannelError> {
407        self.output_tx
408            .send(LoopbackEvent::ThinkingChunk(chunk.to_owned()))
409            .await
410            .map_err(|_| ChannelError::ChannelClosed)
411    }
412
413    async fn send_tool_start(&mut self, event: ToolStartEvent<'_>) -> Result<(), ChannelError> {
414        self.output_tx
415            .send(LoopbackEvent::ToolStart(Box::new(ToolStartData {
416                tool_name: event.tool_name.to_owned(),
417                tool_call_id: event.tool_call_id.to_owned(),
418                params: event.params,
419                parent_tool_use_id: event.parent_tool_use_id,
420                started_at: std::time::Instant::now(),
421            })))
422            .await
423            .map_err(|_| ChannelError::ChannelClosed)
424    }
425
426    async fn send_tool_output(&mut self, event: ToolOutputEvent<'_>) -> Result<(), ChannelError> {
427        self.output_tx
428            .send(LoopbackEvent::ToolOutput(Box::new(ToolOutputData {
429                tool_name: event.tool_name.to_owned(),
430                display: event.body.to_owned(),
431                diff: event.diff,
432                filter_stats: event.filter_stats,
433                kept_lines: event.kept_lines,
434                locations: event.locations,
435                tool_call_id: event.tool_call_id.to_owned(),
436                is_error: event.is_error,
437                terminal_id: None,
438                parent_tool_use_id: event.parent_tool_use_id,
439                raw_response: event.raw_response,
440                started_at: event.started_at,
441            })))
442            .await
443            .map_err(|_| ChannelError::ChannelClosed)
444    }
445
446    async fn confirm(&mut self, _prompt: &str) -> Result<bool, ChannelError> {
447        Ok(true)
448    }
449
450    async fn send_stop_hint(&mut self, hint: StopHint) -> Result<(), ChannelError> {
451        self.output_tx
452            .send(LoopbackEvent::Stop(hint))
453            .await
454            .map_err(|_| ChannelError::ChannelClosed)
455    }
456
457    async fn send_usage(
458        &mut self,
459        input_tokens: u64,
460        output_tokens: u64,
461        context_window: u64,
462    ) -> Result<(), ChannelError> {
463        self.output_tx
464            .send(LoopbackEvent::Usage {
465                input_tokens,
466                output_tokens,
467                context_window,
468            })
469            .await
470            .map_err(|_| ChannelError::ChannelClosed)
471    }
472}
473
474#[cfg(test)]
475mod tests {
476    use super::*;
477
478    #[test]
479    fn channel_message_creation() {
480        let msg = ChannelMessage {
481            text: "hello".to_string(),
482            attachments: vec![],
483        };
484        assert_eq!(msg.text, "hello");
485        assert!(msg.attachments.is_empty());
486    }
487
488    struct StubChannel;
489
490    impl Channel for StubChannel {
491        async fn recv(&mut self) -> Result<Option<ChannelMessage>, ChannelError> {
492            Ok(None)
493        }
494
495        async fn send(&mut self, _text: &str) -> Result<(), ChannelError> {
496            Ok(())
497        }
498
499        async fn send_chunk(&mut self, _chunk: &str) -> Result<(), ChannelError> {
500            Ok(())
501        }
502
503        async fn flush_chunks(&mut self) -> Result<(), ChannelError> {
504            Ok(())
505        }
506    }
507
508    #[tokio::test]
509    async fn send_chunk_default_is_noop() {
510        let mut ch = StubChannel;
511        ch.send_chunk("partial").await.unwrap();
512    }
513
514    #[tokio::test]
515    async fn flush_chunks_default_is_noop() {
516        let mut ch = StubChannel;
517        ch.flush_chunks().await.unwrap();
518    }
519
520    #[tokio::test]
521    async fn stub_channel_confirm_auto_approves() {
522        let mut ch = StubChannel;
523        let result = ch.confirm("Delete everything?").await.unwrap();
524        assert!(result);
525    }
526
527    #[tokio::test]
528    async fn stub_channel_send_typing_default() {
529        let mut ch = StubChannel;
530        ch.send_typing().await.unwrap();
531    }
532
533    #[tokio::test]
534    async fn stub_channel_recv_returns_none() {
535        let mut ch = StubChannel;
536        let msg = ch.recv().await.unwrap();
537        assert!(msg.is_none());
538    }
539
540    #[tokio::test]
541    async fn stub_channel_send_ok() {
542        let mut ch = StubChannel;
543        ch.send("hello").await.unwrap();
544    }
545
546    #[test]
547    fn channel_message_clone() {
548        let msg = ChannelMessage {
549            text: "test".to_string(),
550            attachments: vec![],
551        };
552        let cloned = msg.clone();
553        assert_eq!(cloned.text, "test");
554    }
555
556    #[test]
557    fn channel_message_debug() {
558        let msg = ChannelMessage {
559            text: "debug".to_string(),
560            attachments: vec![],
561        };
562        let debug = format!("{msg:?}");
563        assert!(debug.contains("debug"));
564    }
565
566    #[test]
567    fn attachment_kind_equality() {
568        assert_eq!(AttachmentKind::Audio, AttachmentKind::Audio);
569        assert_ne!(AttachmentKind::Audio, AttachmentKind::Image);
570    }
571
572    #[test]
573    fn attachment_construction() {
574        let a = Attachment {
575            kind: AttachmentKind::Audio,
576            data: vec![0, 1, 2],
577            filename: Some("test.wav".into()),
578        };
579        assert_eq!(a.kind, AttachmentKind::Audio);
580        assert_eq!(a.data.len(), 3);
581        assert_eq!(a.filename.as_deref(), Some("test.wav"));
582    }
583
584    #[test]
585    fn channel_message_with_attachments() {
586        let msg = ChannelMessage {
587            text: String::new(),
588            attachments: vec![Attachment {
589                kind: AttachmentKind::Audio,
590                data: vec![42],
591                filename: None,
592            }],
593        };
594        assert_eq!(msg.attachments.len(), 1);
595        assert_eq!(msg.attachments[0].kind, AttachmentKind::Audio);
596    }
597
598    #[test]
599    fn stub_channel_try_recv_returns_none() {
600        let mut ch = StubChannel;
601        assert!(ch.try_recv().is_none());
602    }
603
604    #[tokio::test]
605    async fn stub_channel_send_queue_count_noop() {
606        let mut ch = StubChannel;
607        ch.send_queue_count(5).await.unwrap();
608    }
609
610    // LoopbackChannel tests
611
612    #[test]
613    fn loopback_pair_returns_linked_handles() {
614        let (channel, handle) = LoopbackChannel::pair(8);
615        // Both sides exist and channels are connected via their sender capacity
616        drop(channel);
617        drop(handle);
618    }
619
620    #[tokio::test]
621    async fn loopback_cancel_signal_can_be_notified_and_awaited() {
622        let (_channel, handle) = LoopbackChannel::pair(8);
623        let signal = std::sync::Arc::clone(&handle.cancel_signal);
624        // Notify from one side, await on the other.
625        let notified = signal.notified();
626        handle.cancel_signal.notify_one();
627        notified.await; // resolves immediately after notify_one()
628    }
629
630    #[tokio::test]
631    async fn loopback_cancel_signal_shared_across_clones() {
632        let (_channel, handle) = LoopbackChannel::pair(8);
633        let signal_a = std::sync::Arc::clone(&handle.cancel_signal);
634        let signal_b = std::sync::Arc::clone(&handle.cancel_signal);
635        let notified = signal_b.notified();
636        signal_a.notify_one();
637        notified.await;
638    }
639
640    #[tokio::test]
641    async fn loopback_send_recv_round_trip() {
642        let (mut channel, handle) = LoopbackChannel::pair(8);
643        handle
644            .input_tx
645            .send(ChannelMessage {
646                text: "hello".to_owned(),
647                attachments: vec![],
648            })
649            .await
650            .unwrap();
651        let msg = channel.recv().await.unwrap().unwrap();
652        assert_eq!(msg.text, "hello");
653    }
654
655    #[tokio::test]
656    async fn loopback_recv_returns_none_when_handle_dropped() {
657        let (mut channel, handle) = LoopbackChannel::pair(8);
658        drop(handle);
659        let result = channel.recv().await.unwrap();
660        assert!(result.is_none());
661    }
662
663    #[tokio::test]
664    async fn loopback_send_produces_full_message_event() {
665        let (mut channel, mut handle) = LoopbackChannel::pair(8);
666        channel.send("world").await.unwrap();
667        let event = handle.output_rx.recv().await.unwrap();
668        assert!(matches!(event, LoopbackEvent::FullMessage(t) if t == "world"));
669    }
670
671    #[tokio::test]
672    async fn loopback_send_chunk_then_flush() {
673        let (mut channel, mut handle) = LoopbackChannel::pair(8);
674        channel.send_chunk("part1").await.unwrap();
675        channel.flush_chunks().await.unwrap();
676        let ev1 = handle.output_rx.recv().await.unwrap();
677        let ev2 = handle.output_rx.recv().await.unwrap();
678        assert!(matches!(ev1, LoopbackEvent::Chunk(t) if t == "part1"));
679        assert!(matches!(ev2, LoopbackEvent::Flush));
680    }
681
682    #[tokio::test]
683    async fn loopback_send_tool_output() {
684        let (mut channel, mut handle) = LoopbackChannel::pair(8);
685        channel
686            .send_tool_output(ToolOutputEvent {
687                tool_name: "bash",
688                body: "exit 0",
689                diff: None,
690                filter_stats: None,
691                kept_lines: None,
692                locations: None,
693                tool_call_id: "",
694                is_error: false,
695                parent_tool_use_id: None,
696                raw_response: None,
697                started_at: None,
698            })
699            .await
700            .unwrap();
701        let event = handle.output_rx.recv().await.unwrap();
702        match event {
703            LoopbackEvent::ToolOutput(data) => {
704                assert_eq!(data.tool_name, "bash");
705                assert_eq!(data.display, "exit 0");
706                assert!(data.diff.is_none());
707                assert!(data.filter_stats.is_none());
708                assert!(data.kept_lines.is_none());
709                assert!(data.locations.is_none());
710                assert_eq!(data.tool_call_id, "");
711                assert!(!data.is_error);
712                assert!(data.terminal_id.is_none());
713                assert!(data.parent_tool_use_id.is_none());
714                assert!(data.raw_response.is_none());
715            }
716            _ => panic!("expected ToolOutput event"),
717        }
718    }
719
720    #[tokio::test]
721    async fn loopback_confirm_auto_approves() {
722        let (mut channel, _handle) = LoopbackChannel::pair(8);
723        let result = channel.confirm("are you sure?").await.unwrap();
724        assert!(result);
725    }
726
727    #[tokio::test]
728    async fn loopback_send_error_when_output_closed() {
729        let (mut channel, handle) = LoopbackChannel::pair(8);
730        // Drop only the output_rx side by dropping the handle
731        drop(handle);
732        let result = channel.send("too late").await;
733        assert!(matches!(result, Err(ChannelError::ChannelClosed)));
734    }
735
736    #[tokio::test]
737    async fn loopback_send_chunk_error_when_output_closed() {
738        let (mut channel, handle) = LoopbackChannel::pair(8);
739        drop(handle);
740        let result = channel.send_chunk("chunk").await;
741        assert!(matches!(result, Err(ChannelError::ChannelClosed)));
742    }
743
744    #[tokio::test]
745    async fn loopback_flush_error_when_output_closed() {
746        let (mut channel, handle) = LoopbackChannel::pair(8);
747        drop(handle);
748        let result = channel.flush_chunks().await;
749        assert!(matches!(result, Err(ChannelError::ChannelClosed)));
750    }
751
752    #[tokio::test]
753    async fn loopback_send_status_event() {
754        let (mut channel, mut handle) = LoopbackChannel::pair(8);
755        channel.send_status("working...").await.unwrap();
756        let event = handle.output_rx.recv().await.unwrap();
757        assert!(matches!(event, LoopbackEvent::Status(s) if s == "working..."));
758    }
759
760    #[tokio::test]
761    async fn loopback_send_usage_produces_usage_event() {
762        let (mut channel, mut handle) = LoopbackChannel::pair(8);
763        channel.send_usage(100, 50, 200_000).await.unwrap();
764        let event = handle.output_rx.recv().await.unwrap();
765        match event {
766            LoopbackEvent::Usage {
767                input_tokens,
768                output_tokens,
769                context_window,
770            } => {
771                assert_eq!(input_tokens, 100);
772                assert_eq!(output_tokens, 50);
773                assert_eq!(context_window, 200_000);
774            }
775            _ => panic!("expected Usage event"),
776        }
777    }
778
779    #[tokio::test]
780    async fn loopback_send_usage_error_when_closed() {
781        let (mut channel, handle) = LoopbackChannel::pair(8);
782        drop(handle);
783        let result = channel.send_usage(1, 2, 3).await;
784        assert!(matches!(result, Err(ChannelError::ChannelClosed)));
785    }
786
787    #[test]
788    fn plan_item_status_variants_are_distinct() {
789        assert!(!matches!(
790            PlanItemStatus::Pending,
791            PlanItemStatus::InProgress
792        ));
793        assert!(!matches!(
794            PlanItemStatus::InProgress,
795            PlanItemStatus::Completed
796        ));
797        assert!(!matches!(
798            PlanItemStatus::Completed,
799            PlanItemStatus::Pending
800        ));
801    }
802
803    #[test]
804    fn loopback_event_session_title_carries_string() {
805        let event = LoopbackEvent::SessionTitle("hello".to_owned());
806        assert!(matches!(event, LoopbackEvent::SessionTitle(s) if s == "hello"));
807    }
808
809    #[test]
810    fn loopback_event_plan_carries_entries() {
811        let entries = vec![
812            ("step 1".to_owned(), PlanItemStatus::Pending),
813            ("step 2".to_owned(), PlanItemStatus::InProgress),
814        ];
815        let event = LoopbackEvent::Plan(entries);
816        match event {
817            LoopbackEvent::Plan(e) => {
818                assert_eq!(e.len(), 2);
819                assert!(matches!(e[0].1, PlanItemStatus::Pending));
820                assert!(matches!(e[1].1, PlanItemStatus::InProgress));
821            }
822            _ => panic!("expected Plan event"),
823        }
824    }
825
826    #[tokio::test]
827    async fn loopback_send_tool_start_produces_tool_start_event() {
828        let (mut channel, mut handle) = LoopbackChannel::pair(8);
829        channel
830            .send_tool_start(ToolStartEvent {
831                tool_name: "shell",
832                tool_call_id: "tc-001",
833                params: Some(serde_json::json!({"command": "ls"})),
834                parent_tool_use_id: None,
835            })
836            .await
837            .unwrap();
838        let event = handle.output_rx.recv().await.unwrap();
839        match event {
840            LoopbackEvent::ToolStart(data) => {
841                assert_eq!(data.tool_name, "shell");
842                assert_eq!(data.tool_call_id, "tc-001");
843                assert!(data.params.is_some());
844                assert!(data.parent_tool_use_id.is_none());
845            }
846            _ => panic!("expected ToolStart event"),
847        }
848    }
849
850    #[tokio::test]
851    async fn loopback_send_tool_start_with_parent_id() {
852        let (mut channel, mut handle) = LoopbackChannel::pair(8);
853        channel
854            .send_tool_start(ToolStartEvent {
855                tool_name: "web",
856                tool_call_id: "tc-002",
857                params: None,
858                parent_tool_use_id: Some("parent-123".into()),
859            })
860            .await
861            .unwrap();
862        let event = handle.output_rx.recv().await.unwrap();
863        assert!(matches!(
864            event,
865            LoopbackEvent::ToolStart(ref data) if data.parent_tool_use_id.as_deref() == Some("parent-123")
866        ));
867    }
868
869    #[tokio::test]
870    async fn loopback_send_tool_start_error_when_output_closed() {
871        let (mut channel, handle) = LoopbackChannel::pair(8);
872        drop(handle);
873        let result = channel
874            .send_tool_start(ToolStartEvent {
875                tool_name: "shell",
876                tool_call_id: "tc-003",
877                params: None,
878                parent_tool_use_id: None,
879            })
880            .await;
881        assert!(matches!(result, Err(ChannelError::ChannelClosed)));
882    }
883
884    #[tokio::test]
885    async fn default_send_tool_output_formats_message() {
886        let mut ch = StubChannel;
887        // Default impl calls self.send() which is a no-op in StubChannel — just verify it doesn't panic.
888        ch.send_tool_output(ToolOutputEvent {
889            tool_name: "bash",
890            body: "hello",
891            diff: None,
892            filter_stats: None,
893            kept_lines: None,
894            locations: None,
895            tool_call_id: "id",
896            is_error: false,
897            parent_tool_use_id: None,
898            raw_response: None,
899            started_at: None,
900        })
901        .await
902        .unwrap();
903    }
904}
zeph_core/channel.rs

zeph_core/
channel.rs
