hatchery_node/

provider_error.rs

//! Provider error classifier and retry tracker (SCOPE: "Provider errors: classify, retry, escalate").
//!
//! Inputs: the rendered PTY screen of a session and ACP turn errors. Four classes:
//! - `retryable`: 429 / 5xx / overloaded / timeout / dropped stream. The node retries the turn
//!   itself with exponential backoff, up to [`MAX_ATTEMPTS`] (PTY: types "continue"; ACP: the
//!   last prompt again).
//! - `limit`: a 5h / 7d window is spent. No retry; the existing `limit` session signal carries
//!   the reset hint and the task manager pauses or moves the work.
//! - `fatal`: 400 bad request, auth, context too long. No retry.
//! - `unknown`: an error-looking line we cannot place. No retry.
//! Only `fatal`, `unknown` and `retryable` with attempts spent escalate (a `provider_error`
//! session signal to core → one HQ-ERROR letter, the run's card to blocked). Retries themselves
//! go to the node log only (target `provider_retry`). Error text is clipped and never carries
//! prompt or screen content beyond the matching line.

use std::collections::HashMap;
use std::time::{Duration, Instant};

pub const MAX_ATTEMPTS: u32 = 3;
const BASE_DELAY: Duration = Duration::from_secs(5);
/// A session that shows no error for this long starts its attempt count over.
const QUIET_RESET: Duration = Duration::from_secs(120);

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Class {
    Retryable,
    Limit,
    Fatal,
    Unknown,
}

impl Class {
    pub fn as_str(self) -> &'static str {
        match self {
            Class::Retryable => "retryable",
            Class::Limit => "limit",
            Class::Fatal => "fatal",
            Class::Unknown => "unknown",
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Found {
    pub class: Class,
    pub code: Option<u16>,
    /// The matching line, trimmed and clipped.
    pub hint: String,
}

const 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"];
const 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"];
const 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 (", "model is not supported", "is not supported when using", "unsupported model", "model_not_found", "does not exist or you do not have access", "not supported when using", "issue with the selected model"];
const ERRORISH: &[&str] = &["error:", "error (", "request failed", "api error", "failed ("];

fn code_in(lower: &str) -> Option<u16> {
    // "(400)", "HTTP 400", "status 400", "code 400"
    let b = lower.as_bytes();
    for i in 0..b.len().saturating_sub(2) {
        if b[i].is_ascii_digit() && b[i + 1].is_ascii_digit() && b[i + 2].is_ascii_digit() {
            let before = if i > 0 { b[i - 1] } else { b' ' };
            let after = b.get(i + 3).copied().unwrap_or(b' ');
            if before.is_ascii_alphanumeric() || after.is_ascii_alphanumeric() {
                continue;
            }
            let n: u16 = lower[i..i + 3].parse().ok()?;
            // Byte window before the number (a char-boundary-safe slice of a multibyte line).
            let ctx = &b[i.saturating_sub(8)..i];
            let has = |w: &[u8]| ctx.windows(w.len()).any(|x| x == w);
            if (400..600).contains(&n) && (before == b'(' || has(b"http") || has(b"status") || has(b"code")) {
                return Some(n);
            }
        }
    }
    None
}

/// Classify one line of provider output.
pub fn classify_line(line: &str) -> Option<Found> {
    let lower = line.to_lowercase();
    let code = code_in(&lower);
    let has = |l: &[&str]| l.iter().any(|p| lower.contains(p));
    let class = if has(LIMIT) {
        Class::Limit
    } else if matches!(code, Some(429 | 500..=599)) || has(RETRYABLE) {
        Class::Retryable
    } else if matches!(code, Some(400..=499)) || has(FATAL) {
        Class::Fatal
    } else if has(ERRORISH) {
        Class::Unknown
    } else {
        return None;
    };
    let hint: String = line.trim().trim_matches(|c: char| !c.is_alphanumeric() && c != ')').chars().take(160).collect();
    Some(Found { class, code, hint })
}

/// Markers that tie a screen line to the provider's own API call (not to tool output such as a
/// compiler "error:" or a test's "timed out" the agent is showing).
const API_MARKERS: &[&str] = &["unexpected status", "is not supported when using", "model is not supported", "request failed", "api error", "http ", "overloaded", "stream disconnected", "too many requests", "usage limit", "you've hit your", "reached your limit", "quota exceeded", "not supported when using", "issue with the selected model"];

/// The newest provider-API error on a screen (bottom-up, last 12 non-empty lines) and its row
/// from the bottom. Stricter than [`classify_line`]: the line must carry an HTTP-like code in
/// parentheses or an API marker, and `unknown` is never taken from a screen.
pub fn classify_screen(screen: &str) -> Option<(Found, usize)> {
    screen.lines().rev().filter(|l| !l.trim().is_empty()).take(12).enumerate().find_map(|(row, l)| {
        let lower = l.to_lowercase();
        let api = API_MARKERS.iter().any(|m| lower.contains(m)) || code_in(&lower).is_some_and(|c| lower.contains(&format!("({c})")));
        if !api || l.len() > 300 {
            return None;
        }
        classify_line(l).filter(|f| f.class != Class::Unknown).map(|f| (f, row))
    })
}

/// Classify an ACP turn error (message plus the vendor code when present).
pub fn classify_message(message: &str, vendor_code: Option<&str>) -> Found {
    let joined = match vendor_code {
        Some(v) => format!("{v}: {message}"),
        None => message.to_owned(),
    };
    classify_line(&joined).unwrap_or(Found { class: Class::Unknown, code: None, hint: joined.chars().take(160).collect() })
}

/// Codex reports a spent account window as a bare JSON-RPC `-32603 Internal error`. When a
/// fresh limit reading (heartbeat or a `/status` screen, at most a few minutes old) puts the
/// 5h window at ~100%, such an otherwise unknown error is the limit.
pub fn limit_from_window(found: Found, vendor_code: Option<&str>, window_used: Option<f64>) -> Found {
    let internal = vendor_code == Some("-32603") || found.hint.to_lowercase().contains("internal error");
    match window_used {
        Some(u) if found.class == Class::Unknown && internal && u >= 98.0 => Found { class: Class::Limit, code: found.code, hint: format!("{} (5h window {u:.0}% used)", found.hint).chars().take(160).collect() },
        _ => found,
    }
}

pub fn backoff(attempt: u32) -> Duration {
    BASE_DELAY * 2u32.pow(attempt.saturating_sub(1).min(6))
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Retry {
    Pty,
    Acp { prompt: String },
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Action {
    /// Retry at `due` (attempt N of MAX_ATTEMPTS).
    Retry { attempt: u32, delay: Duration },
    /// Escalate to core (fatal/unknown, or retryable with attempts spent).
    Escalate { found: Found, attempts: u32 },
    /// Limit: no retry here; the `limit` signal path covers it.
    Limit { found: Found },
}

#[derive(Default)]
struct Track {
    attempts: u32,
    last_error: Option<Instant>,
    /// Screen fingerprint of the error already handled (line + row from bottom).
    seen: Option<(String, usize)>,
    last_prompt: Option<String>,
    escalated: Option<String>,
    /// Current ACP turn: tail of the agent text and the tool calls seen.
    turn_text: String,
    turn_tools: u32,
}

const TURN_TEXT_TAIL: usize = 4096;

#[derive(Default)]
pub struct Tracker {
    sessions: HashMap<u64, Track>,
    due: Vec<(u64, Instant, Retry, u32)>,
}

impl Tracker {
    fn decide(t: &mut Track, found: Found, now: Instant) -> Option<Action> {
        if t.last_error.is_some_and(|at| now.duration_since(at) > QUIET_RESET) {
            t.attempts = 0;
            t.escalated = None;
        }
        t.last_error = Some(now);
        match found.class {
            Class::Limit => Some(Action::Limit { found }),
            Class::Retryable if t.attempts < MAX_ATTEMPTS => {
                t.attempts += 1;
                Some(Action::Retry { attempt: t.attempts, delay: backoff(t.attempts) })
            }
            _ => {
                let key = format!("{}:{:?}", found.class.as_str(), found.code);
                if t.escalated.as_deref() == Some(key.as_str()) {
                    return None;
                }
                t.escalated = Some(key);
                Some(Action::Escalate { attempts: t.attempts, found })
            }
        }
    }

    /// Feed a PTY screen. Acts only when a new error line shows up.
    pub fn observe_screen(&mut self, instance: u64, screen: &str, now: Instant) -> Option<Action> {
        let t = self.sessions.entry(instance).or_default();
        let Some((found, row)) = classify_screen(screen) else {
            t.seen = None;
            return None;
        };
        let fp = (found.hint.clone(), row);
        if t.seen.as_ref() == Some(&fp) || t.seen.as_ref().is_some_and(|(h, r)| *h == fp.0 && *r < row) {
            // Same line still on screen (or scrolled up): already handled.
            t.seen = Some(fp);
            return None;
        }
        t.seen = Some(fp);
        let a = Self::decide(t, found, now);
        if let Some(Action::Retry { attempt, delay }) = &a {
            self.due.push((instance, now + *delay, Retry::Pty, *attempt));
        }
        a
    }

    pub fn turn_started(&mut self, instance: u64, prompt: Option<&str>) {
        let t = self.sessions.entry(instance).or_default();
        t.turn_text.clear();
        t.turn_tools = 0;
        if let Some(p) = prompt.filter(|p| !p.trim().is_empty()) {
            t.last_prompt = Some(p.to_owned());
        }
    }

    /// ACP agent text of the current turn (only the tail is kept).
    pub fn turn_text(&mut self, instance: u64, text: &str, is_delta: bool) {
        let t = self.sessions.entry(instance).or_default();
        if !is_delta {
            t.turn_text.clear();
        }
        t.turn_text.push_str(text);
        if t.turn_text.len() > TURN_TEXT_TAIL {
            let mut cut = t.turn_text.len() - TURN_TEXT_TAIL;
            while !t.turn_text.is_char_boundary(cut) {
                cut += 1;
            }
            t.turn_text.drain(..cut);
        }
    }

    pub fn turn_tool(&mut self, instance: u64) {
        self.sessions.entry(instance).or_default().turn_tools += 1;
    }

    /// An ACP turn that ended without an error flag. Some providers (codex over ACP) report a
    /// rejected request (e.g. 400 "model is not supported") as the turn's only text, or end the
    /// turn with nothing at all. A tool-less turn whose text (or result) is a provider-API error
    /// line is treated as that error; a turn with no text, no result and no tool calls after a
    /// real prompt escalates as `unknown`. Otherwise the count resets as before.
    pub fn turn_ended(&mut self, instance: u64, result: &str, now: Instant) -> Option<Action> {
        let t = self.sessions.entry(instance).or_default();
        let text = std::mem::take(&mut t.turn_text);
        let tools = std::mem::replace(&mut t.turn_tools, 0);
        if tools == 0 {
            let src = if text.trim().is_empty() { result } else { text.as_str() };
            if let Some((found, _)) = classify_screen(src) {
                return self.observe_acp_error(instance, found, now);
            }
            if src.trim().is_empty() && t.last_prompt.is_some() {
                let found = Found { class: Class::Unknown, code: None, hint: "turn ended with no output and no tool calls".into() };
                return Self::decide(t, found, now);
            }
        }
        self.turn_ok(instance);
        None
    }

    /// A clean ACP turn end resets the count.
    pub fn turn_ok(&mut self, instance: u64) {
        if let Some(t) = self.sessions.get_mut(&instance) {
            t.attempts = 0;
            t.escalated = None;
        }
    }

    pub fn observe_acp_error(&mut self, instance: u64, found: Found, now: Instant) -> Option<Action> {
        let t = self.sessions.entry(instance).or_default();
        let prompt = t.last_prompt.clone();
        let mut a = Self::decide(t, found.clone(), now);
        if let Some(Action::Retry { attempt, delay }) = &a {
            match prompt {
                Some(prompt) => self.due.push((instance, now + *delay, Retry::Acp { prompt }, *attempt)),
                // Nothing to resend: escalate instead of guessing.
                None => a = Some(Action::Escalate { found, attempts: *attempt - 1 }),
            }
        }
        a
    }

    /// Retries whose pause is over.
    pub fn take_due(&mut self, now: Instant) -> Vec<(u64, Retry, u32)> {
        let (ready, wait): (Vec<_>, Vec<_>) = std::mem::take(&mut self.due).into_iter().partition(|(_, at, _, _)| *at <= now);
        self.due = wait;
        ready.into_iter().map(|(i, _, r, n)| (i, r, n)).collect()
    }

    pub fn forget(&mut self, instance: u64) {
        self.sessions.remove(&instance);
        self.due.retain(|(i, ..)| *i != instance);
    }
}

#[cfg(test)]
mod tests {
    #[test]
    fn codex_internal_error_at_full_window_is_limit() {
        let f = classify_message("Internal error", Some("-32603"));
        assert_eq!(f.class, Class::Unknown);
        assert_eq!(limit_from_window(f.clone(), Some("-32603"), Some(100.0)).class, Class::Limit);
        assert_eq!(limit_from_window(f.clone(), Some("-32603"), Some(60.0)).class, Class::Unknown);
        assert_eq!(limit_from_window(f, Some("-32603"), None).class, Class::Unknown);
        let fatal = classify_message("model is not supported", None);
        assert_eq!(limit_from_window(fatal, Some("-32603"), Some(100.0)).class, Class::Fatal);
    }

    use super::*;

    #[test]
    fn classes() {
        let c = |s: &str| classify_line(s).map(|f| (f.class, f.code));
        assert_eq!(c("Bad request (400): Request failed (HTTP 400)"), Some((Class::Fatal, Some(400))));
        assert_eq!(c("Too Many Requests (429)"), Some((Class::Retryable, Some(429))));
        assert_eq!(c("API Error: 529 overloaded_error"), Some((Class::Retryable, None)));
        assert_eq!(c("Request failed (HTTP 503)"), Some((Class::Retryable, Some(503))));
        assert_eq!(c("stream disconnected before completion"), Some((Class::Retryable, None)));
        assert_eq!(c("You've hit your usage limit. Try again in 2 hours."), Some((Class::Limit, None)));
        assert_eq!(c("Unauthorized (401)"), Some((Class::Fatal, Some(401))));
        assert_eq!(c("prompt is too long: 210000 tokens"), Some((Class::Fatal, None)));
        assert_eq!(c("Error: something odd"), Some((Class::Unknown, None)));
        assert_eq!(c("fixed 400 tests in src/http"), None);
        assert_eq!(c("> continue"), None);
        // Screens: tool output is not a provider error.
        assert_eq!(classify_screen("error: could not compile `x`\ntest foo ... timed out\n> "), None);
        assert_eq!(classify_screen("Forbidden: path outside workspace"), None);
        assert!(classify_screen("Bad request (400): Request failed (HTTP 400)").is_some());
        // A model the account may not use is fatal (core keeps it out of routing).
        let codex = r#"■ unexpected status 400 Bad Request: {"detail":"The 'gpt-6.1-sol' model is not supported when using Codex with a ChatGPT account."}"#;
        assert_eq!(classify_screen(codex).map(|f| f.0.class), Some(Class::Fatal));
        assert!(classify_screen(codex).unwrap().0.hint.contains("gpt-6.1-sol"));
        assert_eq!(classify_message("There's an issue with the selected model (x-1). It may not exist or you may not have access to it.", None).class, Class::Fatal);
        let codex = r#"unexpected status 400 Bad Request: {"detail":"The 'gpt-x' model is not supported when using Codex with a ChatGPT account."}"#;
        assert_eq!(classify_line(codex).map(|f| (f.class, f.code)), Some((Class::Fatal, Some(400))));
        assert!(classify_screen(codex).is_some());
        assert_eq!(classify_message("You've hit your session limit · resets 1:40am (America/New_York)", Some("-32603")).class, Class::Limit);
    }

    #[test]
    fn acp_turn_text_error_and_empty_turn() {
        let mut t = Tracker::default();
        let t0 = Instant::now();
        // codex: the 400 arrives as the only text of a clean turn.
        t.turn_started(1, Some("task 27"));
        t.turn_text(1, "unexpected status 400 Bad Request: ", true);
        t.turn_text(1, "{\"detail\":\"The 'gpt-x' model is not supported when using Codex with a ChatGPT account.\"}", true);
        match t.turn_ended(1, "", t0) {
            Some(Action::Escalate { found, .. }) => assert_eq!((found.class, found.code), (Class::Fatal, Some(400))),
            other => panic!("expected fatal escalation, got {other:?}"),
        }
        // Same error next turn: escalated once only.
        t.turn_started(1, Some("again"));
        t.turn_text(1, "unexpected status 400 Bad Request: the model is not supported when using Codex", true);
        assert_eq!(t.turn_ended(1, "", t0), None);
        // Empty turn after a real prompt.
        t.turn_started(2, Some("hello"));
        assert!(matches!(t.turn_ended(2, "", t0), Some(Action::Escalate { found: Found { class: Class::Unknown, .. }, .. })));
        // A normal turn: tools ran, error-looking text is the agent quoting a log.
        t.turn_started(3, Some("fix"));
        t.turn_tool(3);
        t.turn_text(3, "the log says: Request failed (HTTP 503)", true);
        assert_eq!(t.turn_ended(3, "", t0), None);
        // A plain answer.
        t.turn_started(4, Some("hi"));
        t.turn_text(4, "Done, all tests pass.", false);
        assert_eq!(t.turn_ended(4, "", t0), None);
    }

    #[test]
    fn pty_retries_then_escalates() {
        let mut t = Tracker::default();
        let t0 = Instant::now();
        let s = |n: usize| format!("{}Request failed (HTTP 503)\n> ", "work\n".repeat(n));
        assert!(matches!(t.observe_screen(1, &s(1), t0), Some(Action::Retry { attempt: 1, .. })));
        assert_eq!(t.observe_screen(1, &s(1), t0), None, "same line on screen is not a new error");
        assert_eq!(t.take_due(t0).len(), 0);
        assert_eq!(t.take_due(t0 + Duration::from_secs(6)).len(), 1);
        assert_eq!(t.observe_screen(1, "> continue\nworking", t0), None);
        assert!(matches!(t.observe_screen(1, &s(2), t0), Some(Action::Retry { attempt: 2, .. })));
        t.observe_screen(1, "ok", t0);
        assert!(matches!(t.observe_screen(1, &s(1), t0), Some(Action::Retry { attempt: 3, .. })));
        t.observe_screen(1, "ok", t0);
        assert!(matches!(t.observe_screen(1, &s(1), t0), Some(Action::Escalate { attempts: 3, .. })));
        t.observe_screen(1, "ok", t0);
        assert_eq!(t.observe_screen(1, &s(1), t0), None, "one escalation per class+code");
    }

    #[test]
    fn fatal_escalates_at_once_and_acp_resends_prompt() {
        let mut t = Tracker::default();
        let t0 = Instant::now();
        assert!(matches!(t.observe_screen(2, "Bad request (400): Request failed (HTTP 400)", t0), Some(Action::Escalate { attempts: 0, .. })));
        t.turn_started(3, Some("do it"));
        let f = classify_message("Internal server error", None);
        assert!(matches!(t.observe_acp_error(3, f, t0), Some(Action::Retry { attempt: 1, .. })));
        let due = t.take_due(t0 + Duration::from_secs(5));
        assert_eq!(due, vec![(3, Retry::Acp { prompt: "do it".into() }, 1)]);
        t.turn_ok(3);
        assert!(matches!(t.observe_acp_error(4, classify_message("Internal server error", None), t0), Some(Action::Escalate { .. })), "no prompt to resend");
    }
}