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

hatchery_node/
session_signals.rs
