hatchery_session_events/
engine.rs

1use crate::event::*;
2use crate::providers::{parser_for, Item, Parsed, ProviderParser};
3use std::collections::{hash_map::DefaultHasher, HashMap};
4use std::hash::{Hash, Hasher};
5
6/// An item must be unchanged this long before it is emitted (suppresses character-by-character flapping).
7const STABLE_MS: u64 = 350;
8/// No screen change for this long = idle (unless the provider shows a working marker).
9const QUIET_MS: u64 = 1200;
10
11struct Slot {
12    id: u64,
13    emitted_hash: Option<u64>,
14    rev: u32,
15    cand_hash: u64,
16    cand_since: u64,
17}
18
19/// Feeds screens (or raw bytes) of one session through its provider parser and emits typed events.
20pub struct EventStream {
21    provider: String,
22    parser: Box<dyn ProviderParser>,
23    slots: HashMap<String, Slot>,
24    next_id: u64,
25    last_items: Vec<Item>,
26    last_parsed: Parsed,
27    screen_hash: u64,
28    last_change: u64,
29    state: Option<(SessionState, String)>,
30    state_id: u64,
31    state_rev: u32,
32    emu: Option<vt100::Parser>,
33    size: (u16, u16),
34}
35
36fn h<T: Hash + ?Sized>(t: &T) -> u64 {
37    let mut s = DefaultHasher::new();
38    t.hash(&mut s);
39    s.finish()
40}
41
42impl EventStream {
43    pub fn new(provider: &str) -> Self {
44        Self::with_parser(provider, parser_for(provider))
45    }
46
47    pub fn with_parser(provider: &str, parser: Box<dyn ProviderParser>) -> Self {
48        Self {
49            provider: provider.to_owned(), parser, slots: HashMap::new(), next_id: 1, last_items: Vec::new(), last_parsed: Parsed::default(),
50            screen_hash: 0, last_change: 0, state: None, state_id: 0, state_rev: 0, emu: None, size: (30, 100),
51        }
52    }
53
54    pub fn set_size(&mut self, rows: u16, cols: u16) {
55        self.size = (rows, cols);
56        if let Some(e) = self.emu.as_mut() { e.screen_mut().set_size(rows, cols); }
57    }
58
59    /// Raw PTY bytes: emulate them, then observe the resulting screen.
60    pub fn feed_bytes(&mut self, bytes: &[u8], now_ms: u64) -> Vec<SessionEvent> {
61        let (r, c) = self.size;
62        let emu = self.emu.get_or_insert_with(|| vt100::Parser::new(r, c, 0));
63        emu.process(bytes);
64        let text = emu.screen().contents();
65        self.observe_screen(&text, now_ms)
66    }
67
68    /// A rendered screen (what C2 relays as a terminal frame).
69    pub fn observe_screen(&mut self, text: &str, now_ms: u64) -> Vec<SessionEvent> {
70        let rows: Vec<String> = text.lines().map(|l| l.trim_end().to_owned()).collect();
71        let hash = h(&rows);
72        if hash != self.screen_hash {
73            self.screen_hash = hash;
74            self.last_change = now_ms;
75        }
76        let parsed = self.parser.parse(&rows);
77        self.last_items = parsed.items.clone();
78        self.last_parsed = parsed;
79        self.settle(now_ms)
80    }
81
82    /// Call periodically so quiet screens settle (items become stable, working turns idle).
83    pub fn tick(&mut self, now_ms: u64) -> Vec<SessionEvent> {
84        self.settle(now_ms)
85    }
86
87    fn settle(&mut self, now_ms: u64) -> Vec<SessionEvent> {
88        let mut out = Vec::new();
89        let items = std::mem::take(&mut self.last_items);
90        for it in &items {
91            let hash = h(&it.kind);
92            let next_id = &mut self.next_id;
93            let slot = self.slots.entry(it.key.clone()).or_insert_with(|| {
94                let id = *next_id;
95                *next_id += 1;
96                Slot { id, emitted_hash: None, rev: 0, cand_hash: hash, cand_since: now_ms }
97            });
98            if slot.cand_hash != hash {
99                slot.cand_hash = hash;
100                slot.cand_since = now_ms;
101            }
102            if now_ms.saturating_sub(slot.cand_since) >= STABLE_MS && slot.emitted_hash != Some(hash) {
103                slot.emitted_hash = Some(hash);
104                slot.rev += 1;
105                out.push(SessionEvent { id: slot.id, rev: slot.rev, provider: self.provider.clone(), source: format!("pty:{}", self.provider), confidence: Confidence::Heuristic, ts_ms: now_ms, event: it.kind.clone() });
106            }
107        }
108        self.last_items = items;
109        // session state
110        let p = &self.last_parsed;
111        let quiet = now_ms.saturating_sub(self.last_change) >= QUIET_MS;
112        let (state, detail) = if p.needs_input {
113            (SessionState::NeedsInput, p.prompt_hint.clone())
114        } else if let Some(d) = &p.working {
115            (SessionState::Working, d.clone())
116        } else if self.state.is_none() && quiet && self.last_items.is_empty() {
117            (SessionState::Starting, String::new())
118        } else if !quiet {
119            (SessionState::Working, "output changing".to_owned())
120        } else {
121            (SessionState::Idle, String::new())
122        };
123        if self.state.as_ref().map(|s| (s.0, &s.1)) != Some((state, &detail)) {
124            if self.state.is_none() { self.state_id = self.next_id; self.next_id += 1; }
125            self.state_rev += 1;
126            self.state = Some((state, detail.clone()));
127            out.push(SessionEvent { id: self.state_id, rev: self.state_rev, provider: self.provider.clone(), source: format!("pty:{}", self.provider), confidence: Confidence::Heuristic, ts_ms: now_ms, event: EventKind::State { state, detail } });
128        }
129        out
130    }
131
132    pub fn state(&self) -> Option<SessionState> { self.state.as_ref().map(|s| s.0) }
133}
hatchery_session_events/engine.rs

hatchery_session_events/
engine.rs
