1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
//! CLI sink — forwards EngineEvents to the TUI event loop.
//!
//! The TUI uses `CliSink::channel()` to forward all events to the
//! main event loop via `UiEvent`. The headless path uses
//! `HeadlessSink` (see `headless_sink.rs`).
use koda_core::engine::{EngineEvent, EngineSink};
// ── UiEvent ───────────────────────────────────────────────
/// Events forwarded from `CliSink` to the main event loop.
pub enum UiEvent {
Engine(EngineEvent),
}
// ── CliSink ───────────────────────────────────────────────
/// Channel-forwarding sink for the TUI event loop.
///
/// Uses an **unbounded** channel so engine events are never dropped.
///
/// Memory safety: the engine produces events sequentially (single
/// turn loop, I/O-bound on LLM streaming at ~50–100 tokens/sec).
/// The TUI drains events every frame (~16 ms). Even in the worst
/// case (large `ToolCallResult` output), only a handful of events
/// queue up — each a few KB — which is negligible. A bounded
/// channel with `try_send` silently dropped `TextDelta` events
/// when the TUI couldn’t keep up, truncating model output.
pub struct CliSink {
ui_tx: tokio::sync::mpsc::UnboundedSender<UiEvent>,
}
impl CliSink {
/// Create a channel-forwarding sink for the TUI event loop.
pub fn channel(ui_tx: tokio::sync::mpsc::UnboundedSender<UiEvent>) -> Self {
Self { ui_tx }
}
}
impl EngineSink for CliSink {
fn emit(&self, event: EngineEvent) {
if let Err(e) = self.ui_tx.send(UiEvent::Engine(event)) {
tracing::warn!("UI channel closed, event lost: {e}");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn emit_forwards_to_channel() {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let sink = CliSink::channel(tx);
sink.emit(EngineEvent::SpinnerStop);
let msg = rx.try_recv().expect("should receive event");
assert!(matches!(msg, UiEvent::Engine(EngineEvent::SpinnerStop)));
}
#[test]
fn emit_does_not_panic_on_closed_channel() {
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
let sink = CliSink::channel(tx);
drop(rx); // close the receiver
// Should not panic — just logs a warning
sink.emit(EngineEvent::SpinnerStop);
}
}