1use std::time::Instant;
5
6use tokio::sync::mpsc;
7
8use crate::event::{AgentEvent, AppEvent};
9
10use super::{App, ChatMessage, ConfirmState, ElicitationState, MessageRole, debug};
11
12impl App {
13 pub fn handle_event(&mut self, event: AppEvent) {
17 match event {
18 AppEvent::Key(key) => self.handle_key(key),
19 AppEvent::Tick => {
20 self.throbber_state.calc_next();
21 self.wave_tick = self.wave_tick.saturating_add(1);
22 self.tick_delights();
23 }
24 AppEvent::Resize(_, _) => {
25 self.sessions.current_mut().render_cache.clear();
26 }
27 AppEvent::Agent(agent_event) => self.handle_agent_event(agent_event),
28 AppEvent::Paste(text) => self.handle_paste(&text),
29 AppEvent::Mouse(m) => self.handle_mouse(m),
30 }
31 }
32
33 pub fn poll_agent_event(&mut self) -> impl Future<Output = Option<AgentEvent>> + use<'_> {
38 self.agent_event_rx.recv()
39 }
40
41 pub fn try_recv_agent_event(&mut self) -> Result<AgentEvent, mpsc::error::TryRecvError> {
51 self.agent_event_rx.try_recv()
52 }
53
54 #[allow(clippy::too_many_lines)] pub fn handle_agent_event(&mut self, event: AgentEvent) {
61 match event {
62 AgentEvent::Chunk(text) => {
63 self.sessions.current_mut().status_label = None;
64 self.last_progress_at = Instant::now();
66 if let Some(last) = self.sessions.current_mut().messages.last_mut()
67 && last.role == MessageRole::Assistant
68 && last.streaming
69 {
70 last.content.push_str(&text);
71 } else {
72 self.sessions
73 .current_mut()
74 .messages
75 .push(ChatMessage::new(MessageRole::Assistant, text).streaming());
76 self.trim_messages();
77 }
78 let completion_tokens = self.metrics.completion_tokens;
80 self.stream_rate.on_token_chunk(completion_tokens);
81 self.auto_scroll();
84 }
85 AgentEvent::FullMessage(text) => {
86 self.sessions.current_mut().status_label = None;
87 if !text.starts_with("[tool output") {
88 self.sessions
89 .current_mut()
90 .messages
91 .push(ChatMessage::new(MessageRole::Assistant, text));
92 self.trim_messages();
93 }
94 self.auto_scroll();
95 }
96 AgentEvent::Flush => {
97 if let Some(last) = self.sessions.current_mut().messages.last_mut()
98 && last.streaming
99 {
100 last.streaming = false;
101 let last_idx = self.sessions.current().messages.len().saturating_sub(1);
102 self.sessions
103 .current_mut()
104 .render_cache
105 .invalidate(last_idx);
106 }
107 }
108 AgentEvent::Typing => {
109 self.pending_count = self.pending_count.saturating_sub(1);
110 self.sessions.current_mut().status_label = Some("thinking...".to_owned());
111 self.last_progress_at = Instant::now();
113 self.stream_rate.on_turn_start();
115 }
116 AgentEvent::Status(text) => {
117 self.sessions.current_mut().status_label =
118 if text.is_empty() { None } else { Some(text) };
119 if self.sessions.current().status_label.is_some() {
121 self.last_progress_at = Instant::now();
122 }
123 self.auto_scroll();
124 }
125 AgentEvent::ToolStart {
126 tool_name,
127 command,
128 tool_call_id,
129 is_mcp,
130 } => {
131 self.sessions.current_mut().status_label = None;
132 self.sessions.current_mut().messages.push(
133 ChatMessage::new(MessageRole::Tool, format!("$ {command}\n"))
134 .streaming()
135 .with_tool(tool_name)
136 .with_tool_call_id(tool_call_id)
137 .with_is_mcp(is_mcp),
138 );
139 self.trim_messages();
140 self.auto_scroll();
141 }
142 AgentEvent::ToolOutputChunk {
143 chunk,
144 tool_call_id,
145 ..
146 } => {
147 self.last_progress_at = Instant::now();
148 let pos = if tool_call_id.is_empty() {
149 self.sessions
152 .current()
153 .messages
154 .iter()
155 .rposition(|m| m.role == MessageRole::Tool && m.streaming)
156 } else {
157 let found =
158 self.sessions.current().messages.iter().rposition(|m| {
159 m.tool_call_id.as_deref() == Some(tool_call_id.as_str())
160 });
161 if found.is_none() {
162 tracing::warn!(
163 %tool_call_id,
164 "ToolOutputChunk: no message with matching tool_call_id — dropping chunk"
165 );
166 }
167 found
168 };
169 if let Some(pos) = pos {
170 self.sessions.current_mut().messages[pos]
171 .content
172 .push_str(&chunk);
173 self.sessions.current_mut().render_cache.invalidate(pos);
174 }
175 self.auto_scroll();
176 }
177 AgentEvent::ToolOutput {
178 tool_name,
179 output,
180 diff,
181 filter_stats,
182 kept_lines,
183 success,
184 tool_call_id,
185 ..
186 } => {
187 self.handle_tool_output_event(
188 tool_name,
189 output,
190 diff,
191 filter_stats,
192 kept_lines,
193 success,
194 tool_call_id,
195 );
196 }
197 AgentEvent::ConfirmRequest {
198 prompt,
199 response_tx,
200 } => {
201 self.confirm_state = Some(ConfirmState {
202 prompt,
203 response_tx: Some(response_tx),
204 });
205 }
206 AgentEvent::ElicitationRequest {
207 request,
208 response_tx,
209 } => {
210 let dialog = crate::widgets::elicitation::ElicitationDialogState::new(request);
211 self.elicitation_state = Some(ElicitationState {
212 dialog,
213 response_tx: Some(response_tx),
214 });
215 }
216 AgentEvent::QueueCount(count) => {
217 self.queued_count = count;
218 self.pending_count = count;
219 }
220 AgentEvent::DiffReady { diff, tool_call_id } => {
221 self.handle_diff_ready(diff, &tool_call_id);
222 }
223 AgentEvent::CommandResult { output, .. } => {
224 self.command_palette = None;
225 self.sessions
226 .current_mut()
227 .messages
228 .push(ChatMessage::new(MessageRole::System, output));
229 self.trim_messages();
230 self.auto_scroll();
231 }
232 AgentEvent::SetCancelSignal(signal) => {
233 self.set_cancel_signal(signal);
234 }
235 AgentEvent::SetMetricsRx(rx) => {
236 self.set_metrics_rx(rx);
237 }
238 AgentEvent::SetTaskSupervisor(supervisor) => {
239 self.set_task_supervisor(supervisor);
240 }
241 AgentEvent::ForegroundSubagentStarted { id, name } => {
242 self.sessions.current_mut().status_label =
243 Some(format!("Sub-agent '{name}' running..."));
244 self.last_progress_at = Instant::now();
246 self.set_view_target(super::AgentViewTarget::SubAgent { id, name });
247 }
248 AgentEvent::ForegroundSubagentCompleted { id, name, success } => {
249 if self.sessions.current().view_target.subagent_id() == Some(id.as_str()) {
252 self.set_view_target(super::AgentViewTarget::Main);
253 }
254 let label = if success {
255 format!("Sub-agent '{name}' completed")
256 } else {
257 format!("Sub-agent '{name}' failed")
258 };
259 self.sessions.current_mut().status_label = Some(label.clone());
260 self.last_progress_at = Instant::now();
262 self.sessions
263 .current_mut()
264 .messages
265 .push(ChatMessage::new(MessageRole::System, label));
266 self.trim_messages();
267 self.auto_scroll();
268 }
269 AgentEvent::ContextEstimate(tokens) => {
270 self.context_token_estimate = tokens;
271 }
272 AgentEvent::FleetSnapshot(snapshot) => {
273 self.fleet_snapshot = snapshot;
274 }
275 AgentEvent::DurableSnapshot(snapshot) => {
276 self.durable_snapshot = snapshot;
277 }
278 AgentEvent::ResumeBanner(text) => {
279 self.resume_banner = Some(text);
280 }
281 AgentEvent::HistoryBackfill(entries) => {
282 self.backfill_history_display_only(&entries);
283 }
284 }
285 }
286
287 fn handle_diff_ready(&mut self, diff: zeph_core::DiffData, tool_call_id: &str) {
288 if let Some(msg) = self
289 .sessions
290 .current_mut()
291 .messages
292 .iter_mut()
293 .rev()
294 .find(|m| {
295 m.role == MessageRole::Tool && m.tool_call_id.as_deref() == Some(tool_call_id)
296 })
297 {
298 msg.diff_data = Some(diff);
299 }
300 }
301
302 #[allow(clippy::too_many_arguments)]
303 fn handle_tool_output_event(
304 &mut self,
305 tool_name: zeph_common::ToolName,
306 output: String,
307 diff: Option<zeph_core::DiffData>,
308 filter_stats: Option<String>,
309 kept_lines: Option<Vec<usize>>,
310 success: bool,
311 tool_call_id: String,
312 ) {
313 debug!(
314 %tool_name,
315 has_diff = diff.is_some(),
316 has_filter_stats = filter_stats.is_some(),
317 output_len = output.len(),
318 "TUI ToolOutput event received"
319 );
320 let pos = if tool_call_id.is_empty() {
323 self.sessions
324 .current()
325 .messages
326 .iter()
327 .rposition(|m| m.role == MessageRole::Tool && m.streaming)
328 } else {
329 let found = self
330 .sessions
331 .current()
332 .messages
333 .iter()
334 .rposition(|m| {
335 m.role == MessageRole::Tool
336 && m.streaming
337 && m.tool_call_id.as_deref() == Some(tool_call_id.as_str())
338 })
339 .or_else(|| {
340 self.sessions
341 .current()
342 .messages
343 .iter()
344 .rposition(|m| m.role == MessageRole::Tool && m.streaming)
345 });
346 if found.is_none() {
347 tracing::warn!(
348 tool_call_id = %tool_call_id,
349 "ToolOutput: no streaming Tool message found — skipping finalization"
350 );
351 }
352 found
353 };
354
355 if let Some(pos) = pos {
356 debug!("finalizing existing streaming Tool message");
362 let header_end = self.sessions.current_mut().messages[pos]
363 .content
364 .find('\n')
365 .map_or(0, |i| i + 1);
366 self.sessions.current_mut().messages[pos]
367 .content
368 .truncate(header_end);
369 self.sessions.current_mut().messages[pos]
370 .content
371 .push_str(&output);
372 self.sessions.current_mut().messages[pos].streaming = false;
373 self.sessions.current_mut().messages[pos].diff_data = diff;
374 self.sessions.current_mut().messages[pos].filter_stats = filter_stats;
375 self.sessions.current_mut().messages[pos].kept_lines = kept_lines;
376 self.sessions.current_mut().messages[pos].success = Some(success);
377 self.sessions.current_mut().render_cache.invalidate(pos);
378 } else if diff.is_some() || filter_stats.is_some() || kept_lines.is_some() {
379 debug!("creating new Tool message with diff (no prior ToolStart)");
381 let mut msg = ChatMessage::new(MessageRole::Tool, output)
382 .with_tool(tool_name)
383 .with_tool_call_id(tool_call_id);
384 msg.diff_data = diff;
385 msg.filter_stats = filter_stats;
386 msg.kept_lines = kept_lines;
387 msg.success = Some(success);
388 self.sessions.current_mut().messages.push(msg);
389 self.trim_messages();
390 } else if let Some(msg) = self
391 .sessions
392 .current_mut()
393 .messages
394 .iter_mut()
395 .rev()
396 .find(|m| m.role == MessageRole::Tool)
397 {
398 msg.filter_stats = filter_stats;
399 }
400 self.auto_scroll();
401 self.maybe_flash_completed_group();
402 }
403
404 fn maybe_flash_completed_group(&mut self) {
411 if self.motion == zeph_config::Motion::Off || !self.delights.completion_flash {
412 return;
413 }
414
415 let messages = &self.sessions.current().messages;
416
417 let Some(last_tool_pos) = messages.iter().rposition(|m| m.role == MessageRole::Tool) else {
419 return;
420 };
421
422 let mut start_idx = last_tool_pos;
424 while start_idx > 0 && messages[start_idx - 1].role == MessageRole::Tool {
425 start_idx -= 1;
426 }
427
428 let all_done = messages[start_idx..=last_tool_pos]
430 .iter()
431 .all(|m| !m.streaming && m.success.is_some());
432 if !all_done {
433 return;
434 }
435
436 if self.sessions.current().flashed_groups.contains(&start_idx) {
438 return;
439 }
440
441 let group_size = last_tool_pos - start_idx + 1;
442 let now = self.anim_tick();
443 self.sessions.current_mut().flashed_groups.insert(start_idx);
444 self.sessions.current_mut().flash.insert(start_idx, now);
445
446 if self.delights.toasts {
448 let text = if group_size == 1 {
449 "Tool done".to_owned()
450 } else {
451 format!("{group_size} tools done")
452 };
453 self.push_toast(text, crate::delights::ToastKind::Success);
454 }
455 }
456
457 #[must_use]
458 pub fn confirm_state(&self) -> Option<&ConfirmState> {
459 self.confirm_state.as_ref()
460 }
461}
462
463#[cfg(test)]
464mod tests {
465 use tokio::sync::mpsc;
466
467 use crate::app::{AgentViewTarget, App};
468 use crate::event::AgentEvent;
469 use crate::types::{ChatMessage, MessageRole};
470 use zeph_core::DiffData;
471
472 fn make_app() -> App {
473 let (user_tx, agent_rx) = {
474 let (utx, _urx) = mpsc::channel(8);
475 let (_atx, arx) = mpsc::channel(8);
476 (utx, arx)
477 };
478 let mut app = App::new(user_tx, agent_rx);
479 app.sessions.current_mut().messages.clear();
480 app
481 }
482
483 fn push_tool_msg(app: &mut App, id: &str) {
485 let msg = ChatMessage::new(MessageRole::Tool, format!("$ cmd_{id}\n"))
486 .streaming()
487 .with_tool_call_id(id.to_owned());
488 app.sessions.current_mut().messages.push(msg);
489 }
490
491 fn tool_msg(id: &str) -> ChatMessage {
492 ChatMessage::new(MessageRole::Tool, "$ cmd\n".to_owned())
493 .with_tool("bash".into())
494 .with_tool_call_id(id.to_owned())
495 }
496
497 fn diff() -> DiffData {
498 DiffData {
499 file_path: "a.rs".into(),
500 old_content: "old".into(),
501 new_content: "new".into(),
502 }
503 }
504
505 #[test]
506 fn tool_output_chunk_routes_by_id_out_of_order() {
507 let mut app = make_app();
508 push_tool_msg(&mut app, "a");
509 push_tool_msg(&mut app, "b");
510 push_tool_msg(&mut app, "c");
511
512 for (id, chunk) in [
514 ("c", "c1"),
515 ("a", "a1"),
516 ("b", "b1"),
517 ("a", "a2"),
518 ("c", "c2"),
519 ] {
520 app.handle_agent_event(AgentEvent::ToolOutputChunk {
521 tool_name: "bash".into(),
522 command: String::new(),
523 chunk: chunk.to_owned(),
524 tool_call_id: id.to_owned(),
525 });
526 }
527
528 let msgs = app.messages();
529 assert_eq!(msgs.len(), 3);
530 assert_eq!(msgs[0].content, "$ cmd_a\na1a2");
532 assert_eq!(msgs[1].content, "$ cmd_b\nb1");
533 assert_eq!(msgs[2].content, "$ cmd_c\nc1c2");
534 }
535
536 #[test]
537 fn tool_output_chunk_with_unknown_id_is_dropped() {
538 let mut app = make_app();
539 push_tool_msg(&mut app, "known");
540
541 app.handle_agent_event(AgentEvent::ToolOutputChunk {
543 tool_name: "bash".into(),
544 command: String::new(),
545 chunk: "should-not-appear".to_owned(),
546 tool_call_id: "unknown-xyz".to_owned(),
547 });
548
549 assert_eq!(app.messages().len(), 1);
551 assert_eq!(app.messages()[0].content, "$ cmd_known\n");
552 }
553
554 #[test]
555 fn tool_output_finalizes_correct_message_by_id() {
556 let mut app = make_app();
557 push_tool_msg(&mut app, "t1");
558 push_tool_msg(&mut app, "t2");
559
560 app.handle_agent_event(AgentEvent::ToolOutput {
562 tool_name: "bash".into(),
563 command: "$ cmd_t1\n".into(),
564 output: "final-output-t1".to_owned(),
565 success: true,
566 diff: None,
567 filter_stats: None,
568 kept_lines: None,
569 tool_call_id: "t1".to_owned(),
570 });
571
572 let msgs = app.messages();
573 assert_eq!(msgs.len(), 2);
574 assert!(!msgs[0].streaming);
576 assert!(msgs[0].content.contains("final-output-t1"));
577 assert!(msgs[1].streaming);
579 assert_eq!(msgs[1].content, "$ cmd_t2\n");
580 }
581
582 #[test]
583 fn diff_ready_attaches_to_matching_id() {
584 let mut app = make_app();
585 app.sessions.current_mut().messages.push(tool_msg("call-1"));
586 app.sessions.current_mut().messages.push(tool_msg("call-2"));
587
588 app.handle_agent_event(AgentEvent::DiffReady {
589 diff: diff(),
590 tool_call_id: "call-2".into(),
591 });
592
593 assert!(app.sessions.current().messages[0].diff_data.is_none());
594 assert!(app.sessions.current().messages[1].diff_data.is_some());
595 }
596
597 #[test]
598 fn diff_ready_mismatched_id_does_not_attach() {
599 let mut app = make_app();
600 app.sessions.current_mut().messages.push(tool_msg("call-1"));
601
602 app.handle_agent_event(AgentEvent::DiffReady {
603 diff: diff(),
604 tool_call_id: "call-99".into(),
605 });
606
607 assert!(app.sessions.current().messages[0].diff_data.is_none());
608 }
609
610 #[test]
611 fn diff_ready_empty_id_does_not_attach() {
612 let mut app = make_app();
613 app.sessions.current_mut().messages.push(tool_msg("call-1"));
614
615 app.handle_agent_event(AgentEvent::DiffReady {
616 diff: diff(),
617 tool_call_id: String::new(),
618 });
619
620 assert!(app.sessions.current().messages[0].diff_data.is_none());
621 }
622
623 #[test]
624 fn diff_ready_two_concurrent_attach_to_correct_messages() {
625 let mut app = make_app();
626 app.sessions.current_mut().messages.push(tool_msg("call-A"));
627 app.sessions.current_mut().messages.push(tool_msg("call-B"));
628 app.sessions.current_mut().messages.push(tool_msg("call-C"));
629
630 let diff_a = DiffData {
631 file_path: "a.rs".into(),
632 old_content: "old_a".into(),
633 new_content: "new_a".into(),
634 };
635 let diff_b = DiffData {
636 file_path: "b.rs".into(),
637 old_content: "old_b".into(),
638 new_content: "new_b".into(),
639 };
640
641 app.handle_agent_event(AgentEvent::DiffReady {
643 diff: diff_b,
644 tool_call_id: "call-B".into(),
645 });
646 app.handle_agent_event(AgentEvent::DiffReady {
647 diff: diff_a,
648 tool_call_id: "call-A".into(),
649 });
650
651 let msgs = &app.sessions.current().messages;
652 assert_eq!(
653 msgs[0].diff_data.as_ref().map(|d| d.file_path.as_str()),
654 Some("a.rs"),
655 "call-A diff must attach to message 0"
656 );
657 assert_eq!(
658 msgs[1].diff_data.as_ref().map(|d| d.file_path.as_str()),
659 Some("b.rs"),
660 "call-B diff must attach to message 1"
661 );
662 assert!(
663 msgs[2].diff_data.is_none(),
664 "call-C must remain without diff"
665 );
666 }
667
668 #[test]
669 fn foreground_subagent_started_switches_view_to_subagent() {
670 let mut app = make_app();
671 assert!(app.sessions.current().view_target.is_main());
672
673 app.handle_agent_event(AgentEvent::ForegroundSubagentStarted {
674 id: "sa-001".into(),
675 name: "planner".into(),
676 });
677
678 assert_eq!(
679 app.sessions.current().view_target.subagent_id(),
680 Some("sa-001"),
681 "view must switch to the started subagent"
682 );
683 assert_eq!(
684 app.sessions.current().status_label.as_deref(),
685 Some("Sub-agent 'planner' running...")
686 );
687 }
688
689 #[test]
690 fn foreground_subagent_completed_switches_back_when_viewing_subagent() {
691 let mut app = make_app();
692
693 app.handle_agent_event(AgentEvent::ForegroundSubagentStarted {
694 id: "sa-002".into(),
695 name: "coder".into(),
696 });
697 assert_eq!(
698 app.sessions.current().view_target.subagent_id(),
699 Some("sa-002")
700 );
701
702 app.handle_agent_event(AgentEvent::ForegroundSubagentCompleted {
703 id: "sa-002".into(),
704 name: "coder".into(),
705 success: true,
706 });
707
708 assert!(
709 app.sessions.current().view_target.is_main(),
710 "view must return to Main after completion"
711 );
712 assert_eq!(
713 app.sessions.current().status_label.as_deref(),
714 Some("Sub-agent 'coder' completed")
715 );
716 let system_msg = app
717 .sessions
718 .current()
719 .messages
720 .iter()
721 .find(|m| m.role == MessageRole::System);
722 assert!(
723 system_msg.is_some(),
724 "completion system message must be pushed"
725 );
726 }
727
728 #[test]
729 fn foreground_subagent_completed_respects_manual_navigation() {
730 let mut app = make_app();
731
732 app.handle_agent_event(AgentEvent::ForegroundSubagentStarted {
733 id: "sa-003".into(),
734 name: "researcher".into(),
735 });
736
737 app.set_view_target(AgentViewTarget::SubAgent {
739 id: "sa-other".into(),
740 name: "other".into(),
741 });
742
743 app.handle_agent_event(AgentEvent::ForegroundSubagentCompleted {
744 id: "sa-003".into(),
745 name: "researcher".into(),
746 success: false,
747 });
748
749 assert_eq!(
751 app.sessions.current().view_target.subagent_id(),
752 Some("sa-other"),
753 "user's manual navigation must be preserved"
754 );
755 }
756
757 #[test]
758 fn foreground_subagent_failed_shows_failed_label() {
759 let mut app = make_app();
760
761 app.handle_agent_event(AgentEvent::ForegroundSubagentStarted {
762 id: "sa-004".into(),
763 name: "builder".into(),
764 });
765 app.handle_agent_event(AgentEvent::ForegroundSubagentCompleted {
766 id: "sa-004".into(),
767 name: "builder".into(),
768 success: false,
769 });
770
771 assert_eq!(
772 app.sessions.current().status_label.as_deref(),
773 Some("Sub-agent 'builder' failed")
774 );
775 }
776
777 #[test]
780 fn foreground_subagent_fast_complete_ends_on_main() {
781 let mut app = make_app();
782
783 app.handle_agent_event(AgentEvent::ForegroundSubagentStarted {
784 id: "sa-fast".into(),
785 name: "quick".into(),
786 });
787 app.handle_agent_event(AgentEvent::ForegroundSubagentCompleted {
788 id: "sa-fast".into(),
789 name: "quick".into(),
790 success: true,
791 });
792
793 assert!(app.sessions.current().view_target.is_main());
794 }
795
796 #[test]
797 fn context_estimate_updates_cached_value() {
798 let mut app = make_app();
799 assert_eq!(
800 app.context_token_estimate(),
801 0,
802 "initial estimate must be 0"
803 );
804
805 app.handle_agent_event(AgentEvent::ContextEstimate(14_200));
806 assert_eq!(app.context_token_estimate(), 14_200);
807
808 app.handle_agent_event(AgentEvent::ContextEstimate(512));
809 assert_eq!(
810 app.context_token_estimate(),
811 512,
812 "estimate must update on each event"
813 );
814 }
815}