zeph_core/
channel.rs

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

zeph_core/
channel.rs
