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

hatchery_node/
session_signals.rs
