hatchery-node 0.2.74

//! 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 ("];
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()?;
            let ctx = &lower[i.saturating_sub(6)..i];
            if (400..600).contains(&n) && (before == b'(' || ctx.contains("http") || ctx.contains("status") || ctx.contains("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] = &["request failed", "api error", "http ", "overloaded", "stream disconnected", "too many requests", "usage limit", "you've hit your", "reached your limit", "quota exceeded"];

/// 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() })
}

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>,
}

#[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>) {
        if let Some(p) = prompt.filter(|p| !p.trim().is_empty()) {
            self.sessions.entry(instance).or_default().last_prompt = Some(p.to_owned());
        }
    }

    /// 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 {
    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());
    }

    #[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");
    }
}