atman_runtime/
watch.rs

1use std::collections::HashMap;
2use std::sync::{Arc, Mutex};
3use std::time::Duration;
4
5use tokio::sync::Notify;
6use tokio_util::sync::CancellationToken;
7
8use crate::error::RuntimeError;
9use crate::message::Message;
10use crate::tool::{BoxFut, Tier, Tool, ToolArgs, ToolCtx, ToolResult};
11use crate::value::Value;
12
13pub type WatcherId = String;
14
15#[derive(Debug, Clone)]
16pub enum WatchSource {
17    Terminal { handle: String },
18    Bash { handle: String },
19    Agent { handle: String },
20}
21
22impl WatchSource {
23    pub fn kind_str(&self) -> &'static str {
24        match self {
25            Self::Terminal { .. } => "terminal",
26            Self::Bash { .. } => "bash",
27            Self::Agent { .. } => "agent",
28        }
29    }
30
31    pub fn handle(&self) -> &str {
32        match self {
33            Self::Terminal { handle } | Self::Bash { handle } | Self::Agent { handle } => handle,
34        }
35    }
36}
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum WatchMode {
40    Once,
41    Persist,
42}
43
44#[derive(Debug, Clone)]
45pub struct WatchEvent {
46    pub watcher_id: WatcherId,
47    pub source: WatchSource,
48    pub pattern: String,
49    pub row: Option<u16>,
50    pub col: Option<u16>,
51    pub text: String,
52    pub timestamp: chrono::DateTime<chrono::Utc>,
53    pub timed_out: bool,
54    pub exited: bool,
55    pub timeout: Duration,
56}
57
58#[derive(Debug, Clone)]
59pub enum WatchResult {
60    Matched {
61        row: Option<u16>,
62        col: Option<u16>,
63        text: String,
64    },
65    SourceExited,
66    Cancelled,
67}
68
69struct Watcher {
70    id: WatcherId,
71    source: WatchSource,
72    pattern: String,
73    mode: WatchMode,
74    timeout: Duration,
75    created_at: chrono::DateTime<chrono::Utc>,
76    cancel: CancellationToken,
77}
78
79#[derive(Debug, Clone)]
80pub struct WatcherInfo {
81    pub id: WatcherId,
82    pub source: WatchSource,
83    pub pattern: String,
84    pub mode: WatchMode,
85    pub age: Duration,
86    pub timeout: Duration,
87}
88
89pub struct WatchHub {
90    watchers: Mutex<HashMap<WatcherId, Watcher>>,
91    pending_events: Mutex<Vec<WatchEvent>>,
92    notify: Notify,
93}
94
95impl Default for WatchHub {
96    fn default() -> Self {
97        Self::new()
98    }
99}
100
101impl WatchHub {
102    pub fn new() -> Self {
103        Self {
104            watchers: Mutex::new(HashMap::new()),
105            pending_events: Mutex::new(Vec::new()),
106            notify: Notify::new(),
107        }
108    }
109
110    pub fn register(
111        &self,
112        source: WatchSource,
113        pattern: String,
114        mode: WatchMode,
115        timeout: Duration,
116    ) -> WatcherId {
117        let id = format!("w_{}", uuid::Uuid::now_v7().simple());
118        let cancel = CancellationToken::new();
119        let watcher = Watcher {
120            id: id.clone(),
121            source,
122            pattern,
123            mode,
124            timeout,
125            created_at: chrono::Utc::now(),
126            cancel,
127        };
128        self.watchers.lock().unwrap().insert(id.clone(), watcher);
129        id
130    }
131
132    pub fn unregister(&self, id: &WatcherId) -> bool {
133        if let Some(w) = self.watchers.lock().unwrap().remove(id) {
134            w.cancel.cancel();
135            true
136        } else {
137            false
138        }
139    }
140
141    pub fn has_active_watchers(&self) -> bool {
142        !self.watchers.lock().unwrap().is_empty()
143    }
144
145    pub fn list_active(&self) -> Vec<WatcherInfo> {
146        let now = chrono::Utc::now();
147        self.watchers
148            .lock()
149            .unwrap()
150            .values()
151            .map(|w| WatcherInfo {
152                id: w.id.clone(),
153                source: w.source.clone(),
154                pattern: w.pattern.clone(),
155                mode: w.mode,
156                age: (now - w.created_at).to_std().unwrap_or(Duration::ZERO),
157                timeout: w.timeout,
158            })
159            .collect()
160    }
161
162    pub fn list_watchers_for_handle(&self, handle: &str) -> Vec<WatcherInfo> {
163        self.list_active()
164            .into_iter()
165            .filter(|w| w.source.handle() == handle)
166            .collect()
167    }
168
169    pub fn get_cancel(&self, id: &WatcherId) -> CancellationToken {
170        self.watchers
171            .lock()
172            .unwrap()
173            .get(id)
174            .map(|w| w.cancel.clone())
175            .unwrap_or_default()
176    }
177
178    pub fn enqueue_event(&self, event: WatchEvent) {
179        let should_remove = {
180            let watchers = self.watchers.lock().unwrap();
181            watchers
182                .get(&event.watcher_id)
183                .map(|w| matches!(w.mode, WatchMode::Once))
184                .unwrap_or(false)
185        };
186        if should_remove {
187            if let Some(w) = self.watchers.lock().unwrap().remove(&event.watcher_id) {
188                w.cancel.cancel();
189            }
190        }
191        self.pending_events.lock().unwrap().push(event);
192        self.notify.notify_one();
193    }
194
195    pub async fn wait_for_event(&self, timeout: Duration) -> Option<WatchEvent> {
196        if let Some(evt) = self.pending_events.lock().unwrap().pop() {
197            return Some(evt);
198        }
199        if !self.has_active_watchers() {
200            return None;
201        }
202        let _ = tokio::time::timeout(timeout, self.notify.notified()).await;
203        self.pending_events.lock().unwrap().pop()
204    }
205}
206
207pub trait Watchable: Send + Sync {
208    fn watch_output(
209        self: Arc<Self>,
210        pattern: String,
211        cancel: CancellationToken,
212    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = WatchResult> + Send>>;
213}
214
215pub async fn handle_watch_result(
216    hub: Arc<WatchHub>,
217    wid: WatcherId,
218    source: WatchSource,
219    pattern: String,
220    timeout: Duration,
221    result: Result<WatchResult, tokio::time::error::Elapsed>,
222) {
223    match result {
224        Ok(WatchResult::Matched { row, col, text }) => {
225            hub.enqueue_event(WatchEvent {
226                watcher_id: wid,
227                source,
228                pattern,
229                row,
230                col,
231                text,
232                timestamp: chrono::Utc::now(),
233                timed_out: false,
234                exited: false,
235                timeout,
236            });
237        }
238        Ok(WatchResult::SourceExited) => {
239            hub.enqueue_event(WatchEvent {
240                watcher_id: wid.clone(),
241                source,
242                pattern,
243                row: None,
244                col: None,
245                text: String::new(),
246                timestamp: chrono::Utc::now(),
247                timed_out: false,
248                exited: true,
249                timeout,
250            });
251            hub.unregister(&wid);
252        }
253        Ok(WatchResult::Cancelled) => {
254            hub.unregister(&wid);
255        }
256        Err(_) => {
257            hub.enqueue_event(WatchEvent {
258                watcher_id: wid,
259                source,
260                pattern,
261                row: None,
262                col: None,
263                text: String::new(),
264                timestamp: chrono::Utc::now(),
265                timed_out: true,
266                exited: false,
267                timeout,
268            });
269        }
270    }
271}
272
273pub fn format_watch_event_text(evt: &WatchEvent) -> String {
274    let kind = evt.source.kind_str();
275    let handle = evt.source.handle();
276    if evt.exited {
277        format!(
278            "[watcher {}] {} '{}' has already exited. Pattern '{}' will never match.",
279            evt.watcher_id, kind, handle, evt.pattern
280        )
281    } else if evt.timed_out {
282        format!(
283            "[watcher {}] {} '{}' pattern '{}' not detected in {}s. \
284             Consider using {}.capture or {}.output to check current state.",
285            evt.watcher_id,
286            kind,
287            handle,
288            evt.pattern,
289            evt.timeout.as_secs(),
290            kind,
291            kind
292        )
293    } else {
294        format!(
295            "[watcher {}] {} '{}' matched '{}' at row {:?}, col {:?}: {}",
296            evt.watcher_id, kind, handle, evt.pattern, evt.row, evt.col, evt.text
297        )
298    }
299}
300
301fn watch_source_to_value(src: &WatchSource) -> Value {
302    Value::Struct(vec![
303        ("kind".into(), Value::Str(src.kind_str().into())),
304        ("handle".into(), Value::Str(src.handle().into())),
305    ])
306}
307
308fn extract_string(args: &ToolArgs, name: &str, pos: usize) -> Result<String, RuntimeError> {
309    let value = match args.named(name) {
310        Some(v) => v,
311        None => args.positional(pos)?,
312    };
313    match value {
314        Value::Str(s) => Ok(s.clone()),
315        other => Err(RuntimeError::TypeMismatch {
316            expected: "string".into(),
317            actual: other.kind_name().into(),
318        }),
319    }
320}
321
322fn extract_optional_string(args: &ToolArgs, name: &str) -> Option<String> {
323    args.named(name).and_then(|v| {
324        if let Value::Str(s) = v {
325            Some(s.clone())
326        } else {
327            None
328        }
329    })
330}
331
332fn extract_optional_int(args: &ToolArgs, name: &str) -> Option<i64> {
333    args.named(name).and_then(|v| {
334        if let Value::Int(i) = v {
335            Some(*i)
336        } else {
337            None
338        }
339    })
340}
341
342pub struct Watch;
343impl Tool for Watch {
344    fn name(&self) -> &str {
345        "watch"
346    }
347    fn tier(&self) -> Tier {
348        Tier::Four
349    }
350    fn description(&self) -> Option<&str> {
351        Some(
352            "Register a background watcher on any running task (terminal, bash, or agent).\n\
353             When pattern appears in the task's output, the agent is woken up.\n\n\
354             Non-blocking: returns watcher_id immediately.\n\
355             mode: \"once\" (default) or \"persist\".\n\
356             timeout_ms: default 120000 (120s), max 600000 (10min).",
357        )
358    }
359    fn input_schema(&self) -> serde_json::Value {
360        serde_json::json!({
361            "type": "object",
362            "properties": {
363                "handle": {"type": "string"},
364                "pattern": {"type": "string"},
365                "mode": {"type": "string", "enum": ["once", "persist"], "default": "once"},
366                "timeout_ms": {"type": "integer", "default": 120000}
367            },
368            "required": ["handle", "pattern"]
369        })
370    }
371    fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
372        Box::pin(async move {
373            let handle = extract_string(&args, "handle", 0)?;
374            let pattern = extract_string(&args, "pattern", 1)?;
375            let mode = match extract_optional_string(&args, "mode").as_deref() {
376                Some("persist") => WatchMode::Persist,
377                _ => WatchMode::Once,
378            };
379            let timeout_ms = extract_optional_int(&args, "timeout_ms")
380                .unwrap_or(120_000)
381                .clamp(1_000, 600_000) as u64;
382            let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".into());
383
384            let task_registry = ctx.task_registry.clone().ok_or_else(|| {
385                RuntimeError::ToolFailed("watch: task registry not available".into())
386            })?;
387            let snapshot = task_registry.lookup_by_handle(&handle).ok_or_else(|| {
388                RuntimeError::ToolFailed(format!("watch: handle '{handle}' not found"))
389            })?;
390
391            let watch_hub = ctx
392                .watch_hub
393                .clone()
394                .ok_or_else(|| RuntimeError::ToolFailed("watch: watch hub not available".into()))?;
395
396            let source = match snapshot.kind {
397                crate::task_registry::TaskKind::Terminal => WatchSource::Terminal {
398                    handle: handle.clone(),
399                },
400                crate::task_registry::TaskKind::Bash => WatchSource::Bash {
401                    handle: handle.clone(),
402                },
403                crate::task_registry::TaskKind::Flow => WatchSource::Agent {
404                    handle: handle.clone(),
405                },
406            };
407
408            let watcher_id = watch_hub.register(
409                source.clone(),
410                pattern.clone(),
411                mode,
412                Duration::from_millis(timeout_ms),
413            );
414
415            match snapshot.kind {
416                crate::task_registry::TaskKind::Terminal => {
417                    let reg = ctx.term_registry.clone().ok_or_else(|| {
418                        RuntimeError::ToolFailed("watch: terminal registry not available".into())
419                    })?;
420                    let entry = reg.lookup(&handle, &session_id)?;
421                    let watchable: Arc<dyn Watchable> = Arc::clone(&entry) as Arc<dyn Watchable>;
422                    spawn_watcher(
423                        watch_hub,
424                        watcher_id.clone(),
425                        source,
426                        pattern,
427                        mode,
428                        Duration::from_millis(timeout_ms),
429                        watchable,
430                    );
431                }
432                crate::task_registry::TaskKind::Bash => {
433                    let reg = ctx.bg_registry.clone().ok_or_else(|| {
434                        RuntimeError::ToolFailed("watch: bash registry not available".into())
435                    })?;
436                    let entry = reg.lookup(&handle, &session_id)?;
437                    let watchable: Arc<dyn Watchable> = Arc::clone(&entry) as Arc<dyn Watchable>;
438                    spawn_watcher(
439                        watch_hub,
440                        watcher_id.clone(),
441                        source,
442                        pattern,
443                        mode,
444                        Duration::from_millis(timeout_ms),
445                        watchable,
446                    );
447                }
448                crate::task_registry::TaskKind::Flow => {
449                    let reg = ctx.flow_registry.clone().ok_or_else(|| {
450                        RuntimeError::ToolFailed("watch: agent registry not available".into())
451                    })?;
452                    let entry = reg.lookup(&handle)?;
453                    let watchable: Arc<dyn Watchable> = Arc::clone(&entry) as Arc<dyn Watchable>;
454                    spawn_watcher(
455                        watch_hub,
456                        watcher_id.clone(),
457                        source,
458                        pattern,
459                        mode,
460                        Duration::from_millis(timeout_ms),
461                        watchable,
462                    );
463                }
464            }
465
466            Ok(Value::Struct(vec![(
467                "watcher_id".into(),
468                Value::Str(watcher_id),
469            )]))
470        })
471    }
472}
473
474fn spawn_watcher(
475    hub: Arc<WatchHub>,
476    wid: WatcherId,
477    source: WatchSource,
478    pattern: String,
479    mode: WatchMode,
480    timeout: Duration,
481    watchable: Arc<dyn Watchable>,
482) {
483    let cancel = hub.get_cancel(&wid);
484    let w = Arc::clone(&watchable);
485    let pat = pattern.clone();
486    tokio::spawn(async move {
487        loop {
488            let result = tokio::time::timeout(
489                timeout,
490                Arc::clone(&w).watch_output(pat.clone(), cancel.clone()),
491            )
492            .await;
493            let matched = matches!(result, Ok(WatchResult::Matched { .. }));
494            handle_watch_result(
495                Arc::clone(&hub),
496                wid.clone(),
497                source.clone(),
498                pattern.clone(),
499                timeout,
500                result,
501            )
502            .await;
503            if !matched || mode == WatchMode::Once {
504                break;
505            }
506        }
507        if mode == WatchMode::Persist {
508            hub.unregister(&wid);
509        }
510    });
511}
512
513pub struct WatcherList;
514impl Tool for WatcherList {
515    fn name(&self) -> &str {
516        "watcher.list"
517    }
518    fn tier(&self) -> Tier {
519        Tier::Zero
520    }
521    fn description(&self) -> Option<&str> {
522        Some("List all active background watchers with their source, pattern, mode, and age.")
523    }
524    fn input_schema(&self) -> serde_json::Value {
525        serde_json::json!({"type": "object", "properties": {}})
526    }
527    fn call<'a>(&'a self, _args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
528        Box::pin(async move {
529            let hub = ctx.watch_hub.clone().ok_or_else(|| {
530                RuntimeError::ToolFailed("watcher.list: watch hub not available".into())
531            })?;
532            let list = hub.list_active();
533            let items: Vec<Value> = list
534                .iter()
535                .map(|w| {
536                    Value::Struct(vec![
537                        ("watcher_id".into(), Value::Str(w.id.clone())),
538                        ("source".into(), watch_source_to_value(&w.source)),
539                        ("pattern".into(), Value::Str(w.pattern.clone())),
540                        (
541                            "mode".into(),
542                            Value::Str(
543                                match w.mode {
544                                    WatchMode::Once => "once",
545                                    WatchMode::Persist => "persist",
546                                }
547                                .into(),
548                            ),
549                        ),
550                        ("age_ms".into(), Value::Int(w.age.as_millis() as i64)),
551                        (
552                            "timeout_ms".into(),
553                            Value::Int(w.timeout.as_millis() as i64),
554                        ),
555                    ])
556                })
557                .collect();
558            Ok(Value::List(items))
559        })
560    }
561}
562
563pub struct WatcherUnwatch;
564impl Tool for WatcherUnwatch {
565    fn name(&self) -> &str {
566        "watcher.unwatch"
567    }
568    fn tier(&self) -> Tier {
569        Tier::Zero
570    }
571    fn description(&self) -> Option<&str> {
572        Some("Cancel a background watcher by id. Works for any source (terminal/bash/agent).")
573    }
574    fn input_schema(&self) -> serde_json::Value {
575        serde_json::json!({
576            "type": "object",
577            "properties": {
578                "watcher_id": {"type": "string"}
579            },
580            "required": ["watcher_id"]
581        })
582    }
583    fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
584        Box::pin(async move {
585            let id = extract_string(&args, "watcher_id", 0)?;
586            let hub = ctx.watch_hub.clone().ok_or_else(|| {
587                RuntimeError::ToolFailed("watcher.unwatch: watch hub not available".into())
588            })?;
589            if hub.unregister(&id) {
590                Ok(Value::Unit)
591            } else {
592                Err(RuntimeError::ToolFailed(format!(
593                    "watcher.unwatch: '{id}' not found"
594                )))
595            }
596        })
597    }
598}
599
600pub struct WaitForWatcher;
601impl Tool for WaitForWatcher {
602    fn name(&self) -> &str {
603        "wait_for_watcher"
604    }
605    fn tier(&self) -> Tier {
606        Tier::Zero
607    }
608    fn description(&self) -> Option<&str> {
609        Some(
610            "Block until a registered watcher fires, or timeout.\n\
611             If no active watchers exist, returns immediately (Unit).\n\
612             Returns Str with event details on match, or Unit on timeout/no watchers.",
613        )
614    }
615    fn input_schema(&self) -> serde_json::Value {
616        serde_json::json!({
617            "type": "object",
618            "properties": {
619                "timeout_ms": {"type": "integer", "default": 30000}
620            }
621        })
622    }
623    fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
624        Box::pin(async move {
625            let timeout_ms = extract_optional_int(&args, "timeout_ms")
626                .unwrap_or(30_000)
627                .clamp(100, 300_000) as u64;
628            let hub = ctx.watch_hub.clone().ok_or_else(|| {
629                RuntimeError::ToolFailed("wait_for_watcher: watch hub not available".into())
630            })?;
631            match hub.wait_for_event(Duration::from_millis(timeout_ms)).await {
632                Some(evt) => Ok(Value::Message(Message::system_text(
633                    ctx.turn_id
634                        .clone()
635                        .unwrap_or_else(crate::event::TurnId::now),
636                    format_watch_event_text(&evt),
637                ))),
638                None => Ok(Value::Unit),
639            }
640        })
641    }
642}
643
644pub struct HasPendingInjections;
645impl Tool for HasPendingInjections {
646    fn name(&self) -> &str {
647        "has_pending_injections"
648    }
649    fn tier(&self) -> Tier {
650        Tier::Zero
651    }
652    fn description(&self) -> Option<&str> {
653        Some(
654            "Check if there are pending user injections for the current turn.\n\
655             Returns true if any exist — the agent loop should continue so they get drained.",
656        )
657    }
658    fn input_schema(&self) -> serde_json::Value {
659        serde_json::json!({"type": "object", "properties": {}})
660    }
661    fn call<'a>(&'a self, _args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
662        Box::pin(async move {
663            let session = ctx.session_runtime.as_ref().ok_or_else(|| {
664                RuntimeError::ToolFailed("has_pending_injections: no session".into())
665            })?;
666            let turn_id = ctx.turn_id.clone().ok_or_else(|| {
667                RuntimeError::ToolFailed("has_pending_injections: no turn id".into())
668            })?;
669            let has_pending = session.list_pending_injections().iter().any(|inj| {
670                inj.turn_id == turn_id && inj.state == crate::injection::InjectionState::Pending
671            });
672            Ok(Value::Bool(has_pending))
673        })
674    }
675}
676
677#[cfg(test)]
678mod tests {
679    use super::*;
680
681    #[test]
682    fn watch_hub_register_and_unregister() {
683        let hub = WatchHub::new();
684        let id = hub.register(
685            WatchSource::Terminal {
686                handle: "t1".into(),
687            },
688            "$ ".into(),
689            WatchMode::Once,
690            Duration::from_secs(10),
691        );
692        assert!(hub.has_active_watchers());
693        assert!(hub.unregister(&id));
694        assert!(!hub.has_active_watchers());
695    }
696
697    #[test]
698    fn watch_hub_list_watchers_for_handle() {
699        let hub = WatchHub::new();
700        hub.register(
701            WatchSource::Terminal {
702                handle: "t1".into(),
703            },
704            "$ ".into(),
705            WatchMode::Once,
706            Duration::from_secs(10),
707        );
708        hub.register(
709            WatchSource::Terminal {
710                handle: "t1".into(),
711            },
712            "error".into(),
713            WatchMode::Persist,
714            Duration::from_secs(30),
715        );
716        hub.register(
717            WatchSource::Bash {
718                handle: "b1".into(),
719            },
720            "done".into(),
721            WatchMode::Once,
722            Duration::from_secs(10),
723        );
724        let t1_watchers = hub.list_watchers_for_handle("t1");
725        assert_eq!(t1_watchers.len(), 2);
726        let b1_watchers = hub.list_watchers_for_handle("b1");
727        assert_eq!(b1_watchers.len(), 1);
728        assert!(hub.list_watchers_for_handle("nonexistent").is_empty());
729    }
730
731    #[test]
732    fn watch_hub_enqueue_event_removes_once_mode() {
733        let hub = WatchHub::new();
734        let id = hub.register(
735            WatchSource::Bash {
736                handle: "b1".into(),
737            },
738            "done".into(),
739            WatchMode::Once,
740            Duration::from_secs(10),
741        );
742        hub.enqueue_event(WatchEvent {
743            watcher_id: id.clone(),
744            source: WatchSource::Bash {
745                handle: "b1".into(),
746            },
747            pattern: "done".into(),
748            row: Some(0),
749            col: Some(0),
750            text: "done".into(),
751            timestamp: chrono::Utc::now(),
752            timed_out: false,
753            exited: false,
754            timeout: Duration::from_secs(10),
755        });
756        assert!(
757            !hub.has_active_watchers(),
758            "once-mode watcher should be auto-removed after event"
759        );
760    }
761
762    #[test]
763    fn watch_hub_enqueue_event_keeps_persist_mode() {
764        let hub = WatchHub::new();
765        let id = hub.register(
766            WatchSource::Bash {
767                handle: "b1".into(),
768            },
769            "done".into(),
770            WatchMode::Persist,
771            Duration::from_secs(10),
772        );
773        hub.enqueue_event(WatchEvent {
774            watcher_id: id,
775            source: WatchSource::Bash {
776                handle: "b1".into(),
777            },
778            pattern: "done".into(),
779            row: Some(0),
780            col: Some(0),
781            text: "done".into(),
782            timestamp: chrono::Utc::now(),
783            timed_out: false,
784            exited: false,
785            timeout: Duration::from_secs(10),
786        });
787        assert!(
788            hub.has_active_watchers(),
789            "persist-mode watcher should remain after event"
790        );
791    }
792
793    #[test]
794    fn format_event_text_includes_watcher_id_and_pattern() {
795        let evt = WatchEvent {
796            watcher_id: "w_abc123".into(),
797            source: WatchSource::Terminal {
798                handle: "term_x".into(),
799            },
800            pattern: "$ ".into(),
801            row: Some(10),
802            col: Some(0),
803            text: "$ ls".into(),
804            timestamp: chrono::Utc::now(),
805            timed_out: false,
806            exited: false,
807            timeout: Duration::from_secs(120),
808        };
809        let text = format_watch_event_text(&evt);
810        assert!(text.contains("w_abc123"));
811        assert!(text.contains("terminal"));
812        assert!(text.contains("term_x"));
813        assert!(text.contains("$ "));
814    }
815
816    #[test]
817    fn format_event_text_timeout_includes_suggestion() {
818        let evt = WatchEvent {
819            watcher_id: "w_abc".into(),
820            source: WatchSource::Bash {
821                handle: "b1".into(),
822            },
823            pattern: "done".into(),
824            row: None,
825            col: None,
826            text: String::new(),
827            timestamp: chrono::Utc::now(),
828            timed_out: true,
829            exited: false,
830            timeout: Duration::from_secs(120),
831        };
832        let text = format_watch_event_text(&evt);
833        assert!(text.contains("not detected"));
834        assert!(text.contains("capture or"));
835    }
836
837    #[tokio::test]
838    async fn wait_for_event_returns_none_when_no_watchers() {
839        let hub = WatchHub::new();
840        let result = hub.wait_for_event(Duration::from_millis(100)).await;
841        assert!(result.is_none());
842    }
843
844    #[tokio::test]
845    async fn wait_for_event_returns_pending_event() {
846        let hub = Arc::new(WatchHub::new());
847        hub.register(
848            WatchSource::Terminal {
849                handle: "t1".into(),
850            },
851            "$ ".into(),
852            WatchMode::Once,
853            Duration::from_secs(10),
854        );
855        hub.enqueue_event(WatchEvent {
856            watcher_id: "w_test".into(),
857            source: WatchSource::Terminal {
858                handle: "t1".into(),
859            },
860            pattern: "$ ".into(),
861            row: Some(0),
862            col: Some(0),
863            text: "$ ".into(),
864            timestamp: chrono::Utc::now(),
865            timed_out: false,
866            exited: false,
867            timeout: Duration::from_secs(10),
868        });
869        let evt = hub.wait_for_event(Duration::from_millis(100)).await;
870        assert!(evt.is_some());
871        assert_eq!(evt.unwrap().pattern, "$ ");
872    }
873
874    #[tokio::test]
875    async fn wait_for_event_returns_source_exited_event() {
876        // When a watcher's source exits, handle_watch_result enqueues a WatchEvent
877        // with exited=true and unregisters the watcher. wait_for_event must still
878        // return that event even though no watchers remain active.
879        let hub = Arc::new(WatchHub::new());
880        let wid = hub.register(
881            WatchSource::Agent {
882                handle: "a1".into(),
883            },
884            "done".into(),
885            WatchMode::Once,
886            Duration::from_secs(10),
887        );
888        // Simulate source-exited: enqueue event then unregister (mirrors handle_watch_result)
889        hub.enqueue_event(WatchEvent {
890            watcher_id: wid.clone(),
891            source: WatchSource::Agent {
892                handle: "a1".into(),
893            },
894            pattern: "done".into(),
895            row: None,
896            col: None,
897            text: String::new(),
898            timestamp: chrono::Utc::now(),
899            timed_out: false,
900            exited: true,
901            timeout: Duration::from_secs(10),
902        });
903        hub.unregister(&wid);
904        assert!(!hub.has_active_watchers());
905        // Must still return the pending exited event despite no active watchers
906        let evt = hub.wait_for_event(Duration::from_millis(100)).await;
907        assert!(
908            evt.is_some(),
909            "exited event must be consumable via wait_for_event"
910        );
911        let evt = evt.unwrap();
912        assert!(evt.exited);
913        assert_eq!(evt.pattern, "done");
914    }
915
916    #[test]
917    fn format_event_text_exited_includes_already_exited() {
918        let evt = WatchEvent {
919            watcher_id: "w_xyz".into(),
920            source: WatchSource::Agent {
921                handle: "a1".into(),
922            },
923            pattern: "done".into(),
924            row: None,
925            col: None,
926            text: String::new(),
927            timestamp: chrono::Utc::now(),
928            timed_out: false,
929            exited: true,
930            timeout: Duration::from_secs(120),
931        };
932        let text = format_watch_event_text(&evt);
933        assert!(text.contains("already exited"));
934        assert!(text.contains("a1"));
935        assert!(text.contains("done"));
936    }
937
938    struct ScriptedWatchable {
939        results: Vec<WatchResult>,
940        idx: Mutex<usize>,
941    }
942
943    impl Watchable for ScriptedWatchable {
944        fn watch_output(
945            self: Arc<Self>,
946            _pattern: String,
947            _cancel: CancellationToken,
948        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = WatchResult> + Send>> {
949            let mut idx = self.idx.lock().unwrap();
950            let i = *idx;
951            *idx += 1;
952            let result = self
953                .results
954                .get(i)
955                .cloned()
956                .unwrap_or(WatchResult::SourceExited);
957            Box::pin(async move {
958                tokio::time::sleep(Duration::from_millis(10)).await;
959                result
960            })
961        }
962    }
963
964    #[tokio::test]
965    async fn persist_watcher_fires_on_every_match_then_cleans_up() {
966        let hub = Arc::new(WatchHub::new());
967        let wid = hub.register(
968            WatchSource::Bash {
969                handle: "b1".into(),
970            },
971            "done".into(),
972            WatchMode::Persist,
973            Duration::from_secs(10),
974        );
975        let watchable: Arc<dyn Watchable> = Arc::new(ScriptedWatchable {
976            results: vec![
977                WatchResult::Matched {
978                    row: None,
979                    col: Some(0),
980                    text: "done".into(),
981                },
982                WatchResult::Matched {
983                    row: None,
984                    col: Some(4),
985                    text: "done".into(),
986                },
987                WatchResult::SourceExited,
988            ],
989            idx: Mutex::new(0),
990        });
991        spawn_watcher(
992            Arc::clone(&hub),
993            wid,
994            WatchSource::Bash {
995                handle: "b1".into(),
996            },
997            "done".into(),
998            WatchMode::Persist,
999            Duration::from_secs(10),
1000            watchable,
1001        );
1002
1003        let e1 = hub
1004            .wait_for_event(Duration::from_secs(2))
1005            .await
1006            .expect("first match event");
1007        assert!(!e1.exited && !e1.timed_out, "first event should be a match");
1008        let e2 = hub
1009            .wait_for_event(Duration::from_secs(2))
1010            .await
1011            .expect("second match event");
1012        assert!(
1013            !e2.exited && !e2.timed_out,
1014            "second event should be a match"
1015        );
1016        let e3 = hub
1017            .wait_for_event(Duration::from_secs(2))
1018            .await
1019            .expect("exited event");
1020        assert!(e3.exited, "third event should be source-exited");
1021
1022        tokio::time::sleep(Duration::from_millis(50)).await;
1023        assert!(
1024            !hub.has_active_watchers(),
1025            "persist watcher must be removed after source exits"
1026        );
1027    }
1028
1029    #[tokio::test]
1030    async fn persist_watcher_timeout_removes_watcher() {
1031        struct NeverMatch;
1032        impl Watchable for NeverMatch {
1033            fn watch_output(
1034                self: Arc<Self>,
1035                _pattern: String,
1036                cancel: CancellationToken,
1037            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = WatchResult> + Send>>
1038            {
1039                Box::pin(async move {
1040                    cancel.cancelled().await;
1041                    WatchResult::Cancelled
1042                })
1043            }
1044        }
1045
1046        let hub = Arc::new(WatchHub::new());
1047        let wid = hub.register(
1048            WatchSource::Bash {
1049                handle: "b1".into(),
1050            },
1051            "done".into(),
1052            WatchMode::Persist,
1053            Duration::from_millis(50),
1054        );
1055        spawn_watcher(
1056            Arc::clone(&hub),
1057            wid,
1058            WatchSource::Bash {
1059                handle: "b1".into(),
1060            },
1061            "done".into(),
1062            WatchMode::Persist,
1063            Duration::from_millis(50),
1064            Arc::new(NeverMatch) as Arc<dyn Watchable>,
1065        );
1066
1067        let evt = hub
1068            .wait_for_event(Duration::from_secs(2))
1069            .await
1070            .expect("timeout event");
1071        assert!(evt.timed_out, "should receive a timeout event");
1072
1073        tokio::time::sleep(Duration::from_millis(50)).await;
1074        assert!(
1075            !hub.has_active_watchers(),
1076            "persist watcher must be removed after timeout, not linger forever"
1077        );
1078    }
1079
1080    #[tokio::test]
1081    async fn persist_watcher_stops_on_cancel() {
1082        struct NeverMatch;
1083        impl Watchable for NeverMatch {
1084            fn watch_output(
1085                self: Arc<Self>,
1086                _pattern: String,
1087                cancel: CancellationToken,
1088            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = WatchResult> + Send>>
1089            {
1090                Box::pin(async move {
1091                    cancel.cancelled().await;
1092                    WatchResult::Cancelled
1093                })
1094            }
1095        }
1096
1097        let hub = Arc::new(WatchHub::new());
1098        let wid = hub.register(
1099            WatchSource::Bash {
1100                handle: "b1".into(),
1101            },
1102            "done".into(),
1103            WatchMode::Persist,
1104            Duration::from_secs(10),
1105        );
1106        spawn_watcher(
1107            Arc::clone(&hub),
1108            wid.clone(),
1109            WatchSource::Bash {
1110                handle: "b1".into(),
1111            },
1112            "done".into(),
1113            WatchMode::Persist,
1114            Duration::from_secs(10),
1115            Arc::new(NeverMatch) as Arc<dyn Watchable>,
1116        );
1117
1118        hub.unregister(&wid);
1119
1120        tokio::time::sleep(Duration::from_millis(100)).await;
1121        assert!(
1122            !hub.has_active_watchers(),
1123            "persist watcher must stop and be removed when cancelled"
1124        );
1125    }
1126}
atman_runtime/watch.rs

atman_runtime/
watch.rs
