hatchery_node/
provider_error.rs

1//! Provider error classifier and retry tracker (SCOPE: "Provider errors: classify, retry, escalate").
2//!
3//! Inputs: the rendered PTY screen of a session and ACP turn errors. Four classes:
4//! - `retryable`: 429 / 5xx / overloaded / timeout / dropped stream. The node retries the turn
5//!   itself with exponential backoff, up to [`MAX_ATTEMPTS`] (PTY: types "continue"; ACP: the
6//!   last prompt again).
7//! - `limit`: a 5h / 7d window is spent. No retry; the existing `limit` session signal carries
8//!   the reset hint and the task manager pauses or moves the work.
9//! - `fatal`: 400 bad request, auth, context too long. No retry.
10//! - `unknown`: an error-looking line we cannot place. No retry.
11//! Only `fatal`, `unknown` and `retryable` with attempts spent escalate (a `provider_error`
12//! session signal to core → one HQ-ERROR letter, the run's card to blocked). Retries themselves
13//! go to the node log only (target `provider_retry`). Error text is clipped and never carries
14//! prompt or screen content beyond the matching line.
15
16use std::collections::HashMap;
17use std::time::{Duration, Instant};
18
19pub const MAX_ATTEMPTS: u32 = 3;
20const BASE_DELAY: Duration = Duration::from_secs(5);
21/// A session that shows no error for this long starts its attempt count over.
22const QUIET_RESET: Duration = Duration::from_secs(120);
23
24#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25pub enum Class {
26    Retryable,
27    Limit,
28    Fatal,
29    Unknown,
30}
31
32impl Class {
33    pub fn as_str(self) -> &'static str {
34        match self {
35            Class::Retryable => "retryable",
36            Class::Limit => "limit",
37            Class::Fatal => "fatal",
38            Class::Unknown => "unknown",
39        }
40    }
41}
42
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub struct Found {
45    pub class: Class,
46    pub code: Option<u16>,
47    /// The matching line, trimmed and clipped.
48    pub hint: String,
49}
50
51const LIMIT: &[&str] = &["usage limit", "limit reached", "reached your limit", "you've hit your", "you have hit your", "quota exceeded", "exceeded your current quota", "out of credits", "weekly limit", "5-hour limit", "usagelimitexceeded"];
52const RETRYABLE: &[&str] = &["too many requests", "rate limited", "rate limit", "overloaded", "internal server error", "bad gateway", "service unavailable", "gateway timeout", "request timed out", "timed out", "server error", "stream disconnected", "stream error", "connection reset", "connection closed", "error decoding response body", "unexpected eof"];
53const FATAL: &[&str] = &["bad request", "unauthorized", "forbidden", "invalid api key", "authentication", "not logged in", "please log in", "payment required", "request too large", "payload too large", "context length", "context window", "prompt is too long", "too many tokens", "maximum context", "is not in your available models", "not found ("];
54const ERRORISH: &[&str] = &["error:", "error (", "request failed", "api error", "failed ("];
55
56fn code_in(lower: &str) -> Option<u16> {
57    // "(400)", "HTTP 400", "status 400", "code 400"
58    let b = lower.as_bytes();
59    for i in 0..b.len().saturating_sub(2) {
60        if b[i].is_ascii_digit() && b[i + 1].is_ascii_digit() && b[i + 2].is_ascii_digit() {
61            let before = if i > 0 { b[i - 1] } else { b' ' };
62            let after = b.get(i + 3).copied().unwrap_or(b' ');
63            if before.is_ascii_alphanumeric() || after.is_ascii_alphanumeric() {
64                continue;
65            }
66            let n: u16 = lower[i..i + 3].parse().ok()?;
67            let ctx = &lower[i.saturating_sub(6)..i];
68            if (400..600).contains(&n) && (before == b'(' || ctx.contains("http") || ctx.contains("status") || ctx.contains("code")) {
69                return Some(n);
70            }
71        }
72    }
73    None
74}
75
76/// Classify one line of provider output.
77pub fn classify_line(line: &str) -> Option<Found> {
78    let lower = line.to_lowercase();
79    let code = code_in(&lower);
80    let has = |l: &[&str]| l.iter().any(|p| lower.contains(p));
81    let class = if has(LIMIT) {
82        Class::Limit
83    } else if matches!(code, Some(429 | 500..=599)) || has(RETRYABLE) {
84        Class::Retryable
85    } else if matches!(code, Some(400..=499)) || has(FATAL) {
86        Class::Fatal
87    } else if has(ERRORISH) {
88        Class::Unknown
89    } else {
90        return None;
91    };
92    let hint: String = line.trim().trim_matches(|c: char| !c.is_alphanumeric() && c != ')').chars().take(160).collect();
93    Some(Found { class, code, hint })
94}
95
96/// Markers that tie a screen line to the provider's own API call (not to tool output such as a
97/// compiler "error:" or a test's "timed out" the agent is showing).
98const API_MARKERS: &[&str] = &["request failed", "api error", "http ", "overloaded", "stream disconnected", "too many requests", "usage limit", "you've hit your", "reached your limit", "quota exceeded"];
99
100/// The newest provider-API error on a screen (bottom-up, last 12 non-empty lines) and its row
101/// from the bottom. Stricter than [`classify_line`]: the line must carry an HTTP-like code in
102/// parentheses or an API marker, and `unknown` is never taken from a screen.
103pub fn classify_screen(screen: &str) -> Option<(Found, usize)> {
104    screen.lines().rev().filter(|l| !l.trim().is_empty()).take(12).enumerate().find_map(|(row, l)| {
105        let lower = l.to_lowercase();
106        let api = API_MARKERS.iter().any(|m| lower.contains(m)) || code_in(&lower).is_some_and(|c| lower.contains(&format!("({c})")));
107        if !api || l.len() > 300 {
108            return None;
109        }
110        classify_line(l).filter(|f| f.class != Class::Unknown).map(|f| (f, row))
111    })
112}
113
114/// Classify an ACP turn error (message plus the vendor code when present).
115pub fn classify_message(message: &str, vendor_code: Option<&str>) -> Found {
116    let joined = match vendor_code {
117        Some(v) => format!("{v}: {message}"),
118        None => message.to_owned(),
119    };
120    classify_line(&joined).unwrap_or(Found { class: Class::Unknown, code: None, hint: joined.chars().take(160).collect() })
121}
122
123pub fn backoff(attempt: u32) -> Duration {
124    BASE_DELAY * 2u32.pow(attempt.saturating_sub(1).min(6))
125}
126
127#[derive(Clone, Debug, PartialEq, Eq)]
128pub enum Retry {
129    Pty,
130    Acp { prompt: String },
131}
132
133#[derive(Clone, Debug, PartialEq, Eq)]
134pub enum Action {
135    /// Retry at `due` (attempt N of MAX_ATTEMPTS).
136    Retry { attempt: u32, delay: Duration },
137    /// Escalate to core (fatal/unknown, or retryable with attempts spent).
138    Escalate { found: Found, attempts: u32 },
139    /// Limit: no retry here; the `limit` signal path covers it.
140    Limit { found: Found },
141}
142
143#[derive(Default)]
144struct Track {
145    attempts: u32,
146    last_error: Option<Instant>,
147    /// Screen fingerprint of the error already handled (line + row from bottom).
148    seen: Option<(String, usize)>,
149    last_prompt: Option<String>,
150    escalated: Option<String>,
151}
152
153#[derive(Default)]
154pub struct Tracker {
155    sessions: HashMap<u64, Track>,
156    due: Vec<(u64, Instant, Retry, u32)>,
157}
158
159impl Tracker {
160    fn decide(t: &mut Track, found: Found, now: Instant) -> Option<Action> {
161        if t.last_error.is_some_and(|at| now.duration_since(at) > QUIET_RESET) {
162            t.attempts = 0;
163            t.escalated = None;
164        }
165        t.last_error = Some(now);
166        match found.class {
167            Class::Limit => Some(Action::Limit { found }),
168            Class::Retryable if t.attempts < MAX_ATTEMPTS => {
169                t.attempts += 1;
170                Some(Action::Retry { attempt: t.attempts, delay: backoff(t.attempts) })
171            }
172            _ => {
173                let key = format!("{}:{:?}", found.class.as_str(), found.code);
174                if t.escalated.as_deref() == Some(key.as_str()) {
175                    return None;
176                }
177                t.escalated = Some(key);
178                Some(Action::Escalate { attempts: t.attempts, found })
179            }
180        }
181    }
182
183    /// Feed a PTY screen. Acts only when a new error line shows up.
184    pub fn observe_screen(&mut self, instance: u64, screen: &str, now: Instant) -> Option<Action> {
185        let t = self.sessions.entry(instance).or_default();
186        let Some((found, row)) = classify_screen(screen) else {
187            t.seen = None;
188            return None;
189        };
190        let fp = (found.hint.clone(), row);
191        if t.seen.as_ref() == Some(&fp) || t.seen.as_ref().is_some_and(|(h, r)| *h == fp.0 && *r < row) {
192            // Same line still on screen (or scrolled up): already handled.
193            t.seen = Some(fp);
194            return None;
195        }
196        t.seen = Some(fp);
197        let a = Self::decide(t, found, now);
198        if let Some(Action::Retry { attempt, delay }) = &a {
199            self.due.push((instance, now + *delay, Retry::Pty, *attempt));
200        }
201        a
202    }
203
204    pub fn turn_started(&mut self, instance: u64, prompt: Option<&str>) {
205        if let Some(p) = prompt.filter(|p| !p.trim().is_empty()) {
206            self.sessions.entry(instance).or_default().last_prompt = Some(p.to_owned());
207        }
208    }
209
210    /// A clean ACP turn end resets the count.
211    pub fn turn_ok(&mut self, instance: u64) {
212        if let Some(t) = self.sessions.get_mut(&instance) {
213            t.attempts = 0;
214            t.escalated = None;
215        }
216    }
217
218    pub fn observe_acp_error(&mut self, instance: u64, found: Found, now: Instant) -> Option<Action> {
219        let t = self.sessions.entry(instance).or_default();
220        let prompt = t.last_prompt.clone();
221        let mut a = Self::decide(t, found.clone(), now);
222        if let Some(Action::Retry { attempt, delay }) = &a {
223            match prompt {
224                Some(prompt) => self.due.push((instance, now + *delay, Retry::Acp { prompt }, *attempt)),
225                // Nothing to resend: escalate instead of guessing.
226                None => a = Some(Action::Escalate { found, attempts: *attempt - 1 }),
227            }
228        }
229        a
230    }
231
232    /// Retries whose pause is over.
233    pub fn take_due(&mut self, now: Instant) -> Vec<(u64, Retry, u32)> {
234        let (ready, wait): (Vec<_>, Vec<_>) = std::mem::take(&mut self.due).into_iter().partition(|(_, at, _, _)| *at <= now);
235        self.due = wait;
236        ready.into_iter().map(|(i, _, r, n)| (i, r, n)).collect()
237    }
238
239    pub fn forget(&mut self, instance: u64) {
240        self.sessions.remove(&instance);
241        self.due.retain(|(i, ..)| *i != instance);
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248
249    #[test]
250    fn classes() {
251        let c = |s: &str| classify_line(s).map(|f| (f.class, f.code));
252        assert_eq!(c("Bad request (400): Request failed (HTTP 400)"), Some((Class::Fatal, Some(400))));
253        assert_eq!(c("Too Many Requests (429)"), Some((Class::Retryable, Some(429))));
254        assert_eq!(c("API Error: 529 overloaded_error"), Some((Class::Retryable, None)));
255        assert_eq!(c("Request failed (HTTP 503)"), Some((Class::Retryable, Some(503))));
256        assert_eq!(c("stream disconnected before completion"), Some((Class::Retryable, None)));
257        assert_eq!(c("You've hit your usage limit. Try again in 2 hours."), Some((Class::Limit, None)));
258        assert_eq!(c("Unauthorized (401)"), Some((Class::Fatal, Some(401))));
259        assert_eq!(c("prompt is too long: 210000 tokens"), Some((Class::Fatal, None)));
260        assert_eq!(c("Error: something odd"), Some((Class::Unknown, None)));
261        assert_eq!(c("fixed 400 tests in src/http"), None);
262        assert_eq!(c("> continue"), None);
263        // Screens: tool output is not a provider error.
264        assert_eq!(classify_screen("error: could not compile `x`\ntest foo ... timed out\n> "), None);
265        assert_eq!(classify_screen("Forbidden: path outside workspace"), None);
266        assert!(classify_screen("Bad request (400): Request failed (HTTP 400)").is_some());
267    }
268
269    #[test]
270    fn pty_retries_then_escalates() {
271        let mut t = Tracker::default();
272        let t0 = Instant::now();
273        let s = |n: usize| format!("{}Request failed (HTTP 503)\n> ", "work\n".repeat(n));
274        assert!(matches!(t.observe_screen(1, &s(1), t0), Some(Action::Retry { attempt: 1, .. })));
275        assert_eq!(t.observe_screen(1, &s(1), t0), None, "same line on screen is not a new error");
276        assert_eq!(t.take_due(t0).len(), 0);
277        assert_eq!(t.take_due(t0 + Duration::from_secs(6)).len(), 1);
278        assert_eq!(t.observe_screen(1, "> continue\nworking", t0), None);
279        assert!(matches!(t.observe_screen(1, &s(2), t0), Some(Action::Retry { attempt: 2, .. })));
280        t.observe_screen(1, "ok", t0);
281        assert!(matches!(t.observe_screen(1, &s(1), t0), Some(Action::Retry { attempt: 3, .. })));
282        t.observe_screen(1, "ok", t0);
283        assert!(matches!(t.observe_screen(1, &s(1), t0), Some(Action::Escalate { attempts: 3, .. })));
284        t.observe_screen(1, "ok", t0);
285        assert_eq!(t.observe_screen(1, &s(1), t0), None, "one escalation per class+code");
286    }
287
288    #[test]
289    fn fatal_escalates_at_once_and_acp_resends_prompt() {
290        let mut t = Tracker::default();
291        let t0 = Instant::now();
292        assert!(matches!(t.observe_screen(2, "Bad request (400): Request failed (HTTP 400)", t0), Some(Action::Escalate { attempts: 0, .. })));
293        t.turn_started(3, Some("do it"));
294        let f = classify_message("Internal server error", None);
295        assert!(matches!(t.observe_acp_error(3, f, t0), Some(Action::Retry { attempt: 1, .. })));
296        let due = t.take_due(t0 + Duration::from_secs(5));
297        assert_eq!(due, vec![(3, Retry::Acp { prompt: "do it".into() }, 1)]);
298        t.turn_ok(3);
299        assert!(matches!(t.observe_acp_error(4, classify_message("Internal server error", None), t0), Some(Action::Escalate { .. })), "no prompt to resend");
300    }
301}
hatchery_node/provider_error.rs

hatchery_node/
provider_error.rs
