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

zeph_core/
channel.rs
