hatchery_node/
session_signals.rs

1//! Short session signals the node reads off its own PTY screens: a provider CLI hit a
2//! usage/rate limit, needs a login, or is ready again. Only state changes are sent to
3//! core (`task.session_signal` over the node-call channel); screens never leave the node
4//! for this. Detection is heuristic text matching on the rendered screen and is kept
5//! per provider-agnostic phrase lists so a CLI wording change degrades to "no signal".
6
7use std::collections::HashMap;
8use std::sync::{Mutex, OnceLock};
9use std::time::{Duration, Instant};
10
11#[derive(Clone, Debug, PartialEq, Eq)]
12pub enum Signal {
13    /// `hint` is the matching screen line (trimmed, short), e.g. "resets 3pm".
14    Limit { hint: String },
15    LoginRequired,
16    Ready,
17}
18
19impl Signal {
20    pub fn kind(&self) -> &'static str {
21        match self {
22            Signal::Limit { .. } => "limit",
23            Signal::LoginRequired => "login_required",
24            Signal::Ready => "ready",
25        }
26    }
27}
28
29const LIMIT_PHRASES: &[&str] = &[
30    "usage limit",
31    "rate limit",
32    "rate-limit",
33    "limit reached",
34    "reached your limit",
35    "you've hit your",
36    "you have hit your",
37    "quota exceeded",
38    "exceeded your current quota",
39    "too many requests",
40    "out of credits",
41    "weekly limit",
42    "5-hour limit",
43];
44const LOGIN_PHRASES: &[&str] = &[
45    "select login method",
46    "sign in with chatgpt",
47    "please log in",
48    "please login",
49    "run /login",
50    "not logged in",
51    "approve in your browser",
52    "how would you like to authenticate",
53    "invalid api key",
54    "authentication required",
55];
56
57/// Classify one screen. Lines are scanned bottom-up: the newest state wins.
58pub fn classify(screen: &str) -> Option<Signal> {
59    for line in screen.lines().rev().filter(|l| !l.trim().is_empty()).take(40) {
60        let lower = line.to_lowercase();
61        if LIMIT_PHRASES.iter().any(|p| lower.contains(p)) {
62            let hint: String = line.trim().trim_matches(|c: char| !c.is_alphanumeric()).chars().take(120).collect();
63            return Some(Signal::Limit { hint });
64        }
65        if LOGIN_PHRASES.iter().any(|p| lower.contains(p)) {
66            return Some(Signal::LoginRequired);
67        }
68    }
69    None
70}
71
72#[derive(Default)]
73struct Track {
74    last: Option<&'static str>,
75    scanned: Option<Instant>,
76}
77
78/// Per-session change detector with a scan budget (at most one scan per `every`).
79pub struct Detector {
80    every: Duration,
81    sessions: HashMap<(String, u64), Track>,
82}
83
84impl Detector {
85    pub fn new(every: Duration) -> Self {
86        Self { every, sessions: HashMap::new() }
87    }
88
89    /// Feed a screen. Returns a signal only when the state changed. A session leaving a
90    /// limit/login state reports `Ready`; a quiet session that never had one reports
91    /// nothing.
92    pub fn observe(&mut self, workspace: &str, instance: u64, screen: &str, now: Instant) -> Option<Signal> {
93        let t = self.sessions.entry((workspace.to_owned(), instance)).or_default();
94        if t.scanned.is_some_and(|at| now.duration_since(at) < self.every) {
95            return None;
96        }
97        t.scanned = Some(now);
98        let found = classify(screen);
99        let kind = found.as_ref().map(Signal::kind);
100        match (t.last, kind) {
101            (a, b) if a == b => None,
102            (Some(_), None) => {
103                t.last = None;
104                Some(Signal::Ready)
105            }
106            (_, Some(k)) => {
107                t.last = Some(k);
108                found
109            }
110            (None, None) => None,
111        }
112    }
113
114    pub fn forget(&mut self, workspace: &str, instance: u64) {
115        self.sessions.remove(&(workspace.to_owned(), instance));
116    }
117}
118
119/// One outgoing signal.
120#[derive(Clone, Debug)]
121pub struct Outgoing {
122    pub workspace: String,
123    pub instance: u64,
124    pub provider: String,
125    pub signal: Signal,
126}
127
128static SINK: OnceLock<tokio::sync::mpsc::UnboundedSender<Outgoing>> = OnceLock::new();
129static DETECTOR: OnceLock<Mutex<Detector>> = OnceLock::new();
130
131/// Install the sink (once, at startup). Without a sink, screens are not scanned.
132pub fn set_sink(tx: tokio::sync::mpsc::UnboundedSender<Outgoing>) {
133    let _ = SINK.set(tx);
134}
135
136/// Called from the frame publisher; cheap when no sink is installed.
137pub fn feed(workspace: &str, instance: u64, provider: &str, screen: &str) {
138    let Some(tx) = SINK.get() else { return };
139    let det = DETECTOR.get_or_init(|| Mutex::new(Detector::new(Duration::from_secs(1))));
140    let signal = det.lock().unwrap_or_else(|p| p.into_inner()).observe(workspace, instance, screen, Instant::now());
141    if let Some(signal) = signal {
142        let _ = tx.send(Outgoing { workspace: workspace.to_owned(), instance, provider: provider.to_owned(), signal });
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    #[test]
151    fn classifies_limits_and_logins() {
152        assert_eq!(
153            classify("● done\n\nClaude usage limit reached. Your limit will reset at 3pm.\n> "),
154            Some(Signal::Limit { hint: "Claude usage limit reached. Your limit will reset at 3pm".into() })
155        );
156        assert_eq!(classify("■ You've hit your usage limit. Try again in 2 hours."), Some(Signal::Limit { hint: "You've hit your usage limit. Try again in 2 hours".into() }));
157        assert_eq!(classify("Select login method:\n 1. Claude account"), Some(Signal::LoginRequired));
158        assert_eq!(classify("Welcome back!\n> fix the tests"), None);
159    }
160
161    #[test]
162    fn newest_line_wins() {
163        // A login prompt below an older limit line: login is the current state.
164        assert_eq!(classify("rate limit hit earlier\n...\nPlease log in again"), Some(Signal::LoginRequired));
165    }
166
167    #[test]
168    fn detector_reports_changes_only() {
169        let mut d = Detector::new(Duration::from_millis(0));
170        let t0 = Instant::now();
171        assert_eq!(d.observe("w", 1, "hello", t0), None);
172        assert!(matches!(d.observe("w", 1, "usage limit reached", t0), Some(Signal::Limit { .. })));
173        assert_eq!(d.observe("w", 1, "usage limit reached", t0), None);
174        assert_eq!(d.observe("w", 1, "Please log in", t0), Some(Signal::LoginRequired));
175        assert_eq!(d.observe("w", 1, "> ready", t0), Some(Signal::Ready));
176        assert_eq!(d.observe("w", 1, "> ready", t0), None);
177        // Separate sessions are independent.
178        assert!(d.observe("w", 2, "rate limit", t0).is_some());
179    }
180
181    #[test]
182    fn scan_budget_skips_frequent_frames() {
183        let mut d = Detector::new(Duration::from_secs(1));
184        let t0 = Instant::now();
185        assert_eq!(d.observe("w", 1, "x", t0), None);
186        assert_eq!(d.observe("w", 1, "usage limit", t0 + Duration::from_millis(10)), None);
187        assert!(d.observe("w", 1, "usage limit", t0 + Duration::from_millis(1100)).is_some());
188    }
189}
hatchery_node/session_signals.rs

hatchery_node/
session_signals.rs
