feat(tui/auto-drive): add dev fault injector and retry harness

DEVELOPING.md

@@ -0,0 +1,36 @@
+# Development Notes
+
+## Testing / Linting
+
+- `./build-fast.sh`
+- `pnpm install && pnpm lint`
+
+## Auto Drive Fault Injection (development only)
+
+Some retry/backoff paths are easier to exercise with injectable failures. Build
+the TUI with the `dev-faults` feature and set the appropriate environment
+variables before launching:
+
+```bash
+cd codex-rs
+cargo run --bin code-tui --features dev-faults
+```
+
+Available knobs (all optional):
+
+- `CODEX_FAULTS_SCOPE`: currently only `auto_drive` is supported. Faults are
+  ignored for other scopes.
+- `CODEX_FAULTS`: comma-delimited list of `kind:count`. Supported kinds:
+  - `disconnect` – simulate a transient “stream disconnected” error.
+  - `429` – simulate a rate-limit response.
+  Example: `CODEX_FAULTS="disconnect:2,429:1"` will fire two disconnects
+  followed by one rate-limit.
+- `CODEX_FAULTS_429_RESET`: optional reset hint for the 429 fault. Accepted
+  values:
+  - integer seconds (`120`)
+  - `now+90s`
+  - RFC3339 timestamp (`2025-09-29T12:00:00Z`)
+
+When a fault fires, the CLI logs a `[faults] …` warning including the remaining
+count. Without the feature flag (default builds), all hooks compile out and
+production behaviour is unaffected.

codex-rs/Cargo.lock

@@ -1341,7 +1341,9 @@ dependencies = [
  "syntect",
  "tempfile",
  "textwrap 0.16.2",
+ "thiserror 1.0.69",
  "tokio",
+ "tokio-util",
  "tracing",
  "tracing-appender",
  "tracing-subscriber",
@@ -6175,6 +6177,7 @@ dependencies = [
  "bytes",
  "futures-core",
  "futures-sink",
+ "futures-util",
  "pin-project-lite",
  "tokio",
 ]

codex-rs/tui/Cargo.toml

@@ -18,6 +18,7 @@ vt100-tests = []
 debug-logs = []
 # Disable legacy tests by default; enable with `--features legacy_tests` if needed.
 legacy_tests = []
+dev-faults = []
 
 [lints]
 workspace = true
@@ -78,6 +79,7 @@ tokio = { version = "1", features = [
     "rt-multi-thread",
     "signal",
 ] }
+tokio-util = { version = "0.7", features = ["rt"] }
 tracing = { version = "0.1.41", features = ["log"] }
 tracing-appender = "0.2.3"
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
@@ -94,6 +96,7 @@ which = "6"
 urlencoding = "2.1"
 pulldown-cmark = "0.13"
 portable-pty = "0.9.0"
+thiserror = "1"
 # vt100 0.16+ depends on unicode-width 0.2.1 which conflicts with ratatui 0.29.x.
 # Stick to 0.15.x until ratatui upgrades.
 vt100 = "0.15.0"

codex-rs/tui/src/chatwidget.rs

@@ -47,6 +47,9 @@ use codex_protocol::num_format::format_with_separators;
 
 
 mod auto_coordinator;
+#[cfg(feature = "dev-faults")]
+mod faults;
+mod retry;
 mod diff_handlers;
 mod agent_install;
 mod diff_ui;
@@ -65,13 +68,15 @@ mod streaming;
 mod terminal_handlers;
 mod terminal;
 mod tools;
+#[cfg(test)]
+mod tests_retry;
 use self::agent_install::{
     start_agent_install_session,
     start_direct_terminal_session,
     start_prompt_terminal_session,
     wrap_command,
 };
-use self::auto_coordinator::{start_auto_coordinator, AutoCoordinatorCommand};
+use self::auto_coordinator::{start_auto_coordinator, AutoCoordinatorCommand, AutoCoordinatorHandle};
 use self::limits_overlay::{LimitsOverlay, LimitsOverlayContent, LimitsTab};
 use self::rate_limit_refresh::start_rate_limit_refresh;
 use self::history_render::{CachedLayout, HistoryRenderState, LayoutRef};
@@ -632,7 +637,7 @@ pub(crate) struct ChatWidget<'a> {
     pending_snapshot_dispatches: VecDeque<PendingSnapshotDispatch>,
 
     auto_state: AutoCoordinatorUiState,
-    auto_tx: Option<Sender<AutoCoordinatorCommand>>,
+    auto_handle: Option<AutoCoordinatorHandle>,
 
     // Event sequencing to preserve original order across streaming/tool events
     // and stream-related flags moved into stream_state
@@ -2850,7 +2855,7 @@ impl ChatWidget<'_> {
             next_ghost_snapshot_id: 0,
             pending_snapshot_dispatches: VecDeque::new(),
             auto_state: AutoCoordinatorUiState::default(),
-            auto_tx: None,
+            auto_handle: None,
             browser_is_external: false,
             // Stable ordering & routing init
             cell_order_seq: vec![OrderKey {
@@ -3094,7 +3099,7 @@ impl ChatWidget<'_> {
             next_ghost_snapshot_id: 0,
             pending_snapshot_dispatches: VecDeque::new(),
             auto_state: AutoCoordinatorUiState::default(),
-            auto_tx: None,
+            auto_handle: None,
             browser_is_external: false,
             // Strict ordering init for forked widget
             cell_order_seq: vec![OrderKey {
@@ -3153,12 +3158,10 @@ impl ChatWidget<'_> {
                         .collect::<Vec<_>>()
                         .join("\n");
                     let prefixed = format!("Coordinator: {text}");
-                    let content = ContentItem::OutputText {
-                        text: prefixed.clone(),
-                    };
+                    let content = ContentItem::InputText { text: prefixed };
                     items.push(ResponseItem::Message {
                         id: None,
-                        role: "assistant".to_string(),
+                        role: "user".to_string(),
                         content: vec![content],
                     });
                 }
@@ -3175,12 +3178,10 @@ impl ChatWidget<'_> {
                         .collect::<Vec<_>>()
                         .join("\n");
                     let prefixed = format!("CLI: {text}");
-                    let content = ContentItem::InputText {
-                        text: prefixed.clone(),
-                    };
+                    let content = ContentItem::OutputText { text: prefixed };
                     items.push(ResponseItem::Message {
                         id: None,
-                        role: "user".to_string(),
+                        role: "assistant".to_string(),
                         content: vec![content],
                     });
                 }
@@ -9610,7 +9611,7 @@ impl ChatWidget<'_> {
             self.config.debug,
         ) {
             Ok(handle) => {
-                self.auto_tx = Some(handle.tx);
+                self.auto_handle = Some(handle);
                 self.auto_state.reset();
                 self.auto_state.active = true;
                 self.auto_state.goal = Some(goal_text.clone());
@@ -9638,11 +9639,11 @@ impl ChatWidget<'_> {
         if !self.auto_state.active || self.auto_state.waiting_for_response {
             return;
         }
-        let Some(tx) = self.auto_tx.as_ref() else {
+        let Some(handle) = self.auto_handle.as_ref() else {
             return;
         };
         let conversation = self.export_response_items();
-        if tx
+        if handle
             .send(AutoCoordinatorCommand::UpdateConversation(conversation))
             .is_err()
         {
@@ -9825,8 +9826,9 @@ impl ChatWidget<'_> {
     }
 
     fn auto_stop(&mut self, message: Option<String>) {
-        if let Some(tx) = self.auto_tx.take() {
-            let _ = tx.send(AutoCoordinatorCommand::Stop);
+        if let Some(handle) = self.auto_handle.take() {
+            handle.cancel();
+            let _ = handle.send(AutoCoordinatorCommand::Stop);
         }
         if let Some(msg) = message {
             self.push_background_tail(msg);

codex-rs/tui/src/chatwidget/auto_coordinator.rs

@@ -1,6 +1,7 @@
 use std::process::Command;
 use std::sync::mpsc::{self, Receiver, Sender};
 use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant, SystemTime};
 
 use anyhow::{anyhow, Context, Result};
 use codex_core::config::Config;
@@ -9,15 +10,32 @@ use codex_core::debug_logger::DebugLogger;
 use codex_core::model_family::{find_family_for_model, derive_default_model_family};
 use codex_core::protocol::SandboxPolicy;
 use codex_core::{AuthManager, ModelClient, Prompt, ResponseEvent, TextFormat};
+use codex_core::error::CodexErr;
 use codex_protocol::models::{ContentItem, ResponseItem};
 use futures::StreamExt;
+use reqwest::StatusCode;
 use serde::Deserialize;
 use serde_json::{self, json, Value};
+use tokio_util::sync::CancellationToken;
 use tracing::debug;
 use uuid::Uuid;
 
 use crate::app_event::{AppEvent, AutoCoordinatorStatus};
 use crate::app_event_sender::AppEventSender;
+use crate::chatwidget::retry::{retry_with_backoff, RetryDecision, RetryError, RetryOptions};
+#[cfg(feature = "dev-faults")]
+use crate::chatwidget::faults::{fault_to_error, next_fault, FaultScope, InjectedFault};
+use codex_common::elapsed::format_duration;
+use chrono::{DateTime, Local, Utc};
+use rand::Rng;
+
+const RATE_LIMIT_BUFFER: Duration = Duration::from_secs(120);
+const RATE_LIMIT_JITTER_MAX: Duration = Duration::from_secs(30);
+const MAX_RETRY_ELAPSED: Duration = Duration::from_secs(7 * 24 * 60 * 60);
+
+#[derive(Debug, thiserror::Error)]
+#[error("auto coordinator cancelled")]
+struct AutoCoordinatorCancelled;
 
 const MODEL_SLUG: &str = "gpt-5";
 const SCHEMA_NAME: &str = "auto_coordinator_flow";
@@ -25,6 +43,20 @@ const SCHEMA_NAME: &str = "auto_coordinator_flow";
 #[derive(Debug, Clone)]
 pub(super) struct AutoCoordinatorHandle {
     pub tx: Sender<AutoCoordinatorCommand>,
+    cancel_token: CancellationToken,
+}
+
+impl AutoCoordinatorHandle {
+    pub fn send(
+        &self,
+        command: AutoCoordinatorCommand,
+    ) -> std::result::Result<(), mpsc::SendError<AutoCoordinatorCommand>> {
+        self.tx.send(command)
+    }
+
+    pub fn cancel(&self) {
+        self.cancel_token.cancel();
+    }
 }
 
 #[derive(Debug)]
@@ -50,6 +82,8 @@ pub(super) fn start_auto_coordinator(
 ) -> Result<AutoCoordinatorHandle> {
     let (cmd_tx, cmd_rx) = mpsc::channel();
     let thread_tx = cmd_tx.clone();
+    let cancel_token = CancellationToken::new();
+    let thread_cancel = cancel_token.clone();
 
     std::thread::spawn(move || {
         if let Err(err) = run_auto_loop(
@@ -59,12 +93,16 @@ pub(super) fn start_auto_coordinator(
             config,
             cmd_rx,
             debug_enabled,
+            thread_cancel,
         ) {
             tracing::error!("auto coordinator loop error: {err:#}");
         }
     });
 
-    Ok(AutoCoordinatorHandle { tx: thread_tx })
+    Ok(AutoCoordinatorHandle {
+        tx: thread_tx,
+        cancel_token,
+    })
 }
 
 fn run_auto_loop(
@@ -74,6 +112,7 @@ fn run_auto_loop(
     config: Config,
     cmd_rx: Receiver<AutoCoordinatorCommand>,
     debug_enabled: bool,
+    cancel_token: CancellationToken,
 ) -> Result<()> {
     let preferred_auth = if config.using_chatgpt_auth {
         codex_protocol::mcp_protocol::AuthMode::ChatGPT
@@ -128,6 +167,11 @@ fn run_auto_loop(
         }
 
         if let Some(conv) = pending_conversation.take() {
+            if cancel_token.is_cancelled() {
+                stopped = true;
+                continue;
+            }
+
             match request_coordinator_decision(
                 &runtime,
                 &client,
@@ -136,6 +180,7 @@ fn run_auto_loop(
                 &schema,
                 conv,
                 &app_event_tx,
+                &cancel_token,
             ) {
                 Ok((status, thoughts, prompt_opt)) => {
                     let event = AppEvent::AutoCoordinatorDecision {
@@ -150,6 +195,10 @@ fn run_auto_loop(
                     }
                 }
                 Err(err) => {
+                    if err.downcast_ref::<AutoCoordinatorCancelled>().is_some() {
+                        stopped = true;
+                        continue;
+                    }
                     let event = AppEvent::AutoCoordinatorDecision {
                         status: AutoCoordinatorStatus::Failed,
                         thoughts: format!("Coordinator error: {err}"),
@@ -242,28 +291,20 @@ fn request_coordinator_decision(
     developer_intro: &str,
     primary_goal: &str,
     schema: &Value,
-    mut conversation: Vec<ResponseItem>,
+    conversation: Vec<ResponseItem>,
     app_event_tx: &AppEventSender,
+    cancel_token: &CancellationToken,
 ) -> Result<(AutoCoordinatorStatus, String, Option<String>)> {
-    let mut prompt = Prompt::default();
-    prompt.store = true;
-    prompt.input.push(make_message("developer", developer_intro.to_string()));
-    prompt
-        .input
-        .push(make_message("developer", primary_goal.to_string()));
-    prompt.input.append(&mut conversation);
-    prompt.text_format = Some(TextFormat {
-        r#type: "json_schema".to_string(),
-        name: Some(SCHEMA_NAME.to_string()),
-        strict: Some(true),
-        schema: Some(schema.clone()),
-    });
-    prompt.model_override = Some(MODEL_SLUG.to_string());
-    let family = find_family_for_model(MODEL_SLUG)
-        .unwrap_or_else(|| derive_default_model_family(MODEL_SLUG));
-    prompt.model_family_override = Some(family);
-
-    let raw = request_decision(runtime, client, &prompt, app_event_tx)?;
+    let raw = request_decision(
+        runtime,
+        client,
+        developer_intro,
+        primary_goal,
+        schema,
+        &conversation,
+        app_event_tx,
+        cancel_token,
+    )?;
     let (decision, value) = parse_decision(&raw)?;
     debug!("[Auto coordinator] model decision: {:?}", value);
 
@@ -296,50 +337,443 @@ fn request_coordinator_decision(
 fn request_decision(
     runtime: &tokio::runtime::Runtime,
     client: &ModelClient,
-    prompt: &Prompt,
+    developer_intro: &str,
+    primary_goal: &str,
+    schema: &Value,
+    conversation: &[ResponseItem],
     app_event_tx: &AppEventSender,
+    cancel_token: &CancellationToken,
 ) -> Result<String> {
+    let developer_intro = developer_intro.to_string();
+    let primary_goal = primary_goal.to_string();
+    let schema = schema.clone();
+    let conversation: Vec<ResponseItem> = conversation.to_vec();
     let tx = app_event_tx.clone();
-    runtime.block_on(async move {
-        let mut stream = client.stream(prompt).await?;
-        let mut out = String::new();
-        while let Some(ev) = stream.next().await {
-            match ev {
-                Ok(ResponseEvent::OutputTextDelta { delta, .. }) => out.push_str(&delta),
-                Ok(ResponseEvent::OutputItemDone { item, .. }) => {
-                    if let ResponseItem::Message { content, .. } = item {
-                        for c in content {
-                            if let ContentItem::OutputText { text } = c {
-                                out.push_str(&text);
+    let cancel = cancel_token.clone();
+    let classify = |error: &anyhow::Error| classify_model_error(error);
+    let options = RetryOptions::with_defaults(MAX_RETRY_ELAPSED);
+
+    let result = runtime.block_on(async move {
+        retry_with_backoff(
+            || {
+                let prompt = build_prompt(
+                    &developer_intro,
+                    &primary_goal,
+                    &schema,
+                    &conversation,
+                );
+                let tx_inner = tx.clone();
+                async move {
+                    #[cfg(feature = "dev-faults")]
+                    if let Some(fault) = next_fault(FaultScope::AutoDrive) {
+                        let err = fault_to_error(fault);
+                        return Err(err);
+                    }
+                    let mut stream = client.stream(&prompt).await?;
+                    let mut out = String::new();
+                    while let Some(ev) = stream.next().await {
+                        match ev {
+                            Ok(ResponseEvent::OutputTextDelta { delta, .. }) => out.push_str(&delta),
+                            Ok(ResponseEvent::OutputItemDone { item, .. }) => {
+                                if let ResponseItem::Message { content, .. } = item {
+                                    for c in content {
+                                        if let ContentItem::OutputText { text } = c {
+                                            out.push_str(&text);
+                                        }
+                                    }
+                                }
                             }
+                            Ok(ResponseEvent::ReasoningSummaryDelta {
+                                delta,
+                                summary_index,
+                                ..
+                            }) => {
+                                let message = strip_role_prefix(&delta).to_string();
+                                tx_inner.send(AppEvent::AutoCoordinatorThinking {
+                                    delta: message,
+                                    summary_index,
+                                });
+                            }
+                            Ok(ResponseEvent::ReasoningContentDelta { delta, .. }) => {
+                                let message = strip_role_prefix(&delta).to_string();
+                                tx_inner.send(AppEvent::AutoCoordinatorThinking {
+                                    delta: message,
+                                    summary_index: None,
+                                });
+                            }
+                            Ok(ResponseEvent::Completed { .. }) => break,
+                            Err(err) => return Err(err.into()),
+                            _ => {}
                         }
                     }
+                    Ok(out)
                 }
-                Ok(ResponseEvent::ReasoningSummaryDelta {
-                    delta,
-                    summary_index,
-                    ..
-                }) => {
-                    let message = strip_role_prefix(&delta).to_string();
-                    tx.send(AppEvent::AutoCoordinatorThinking {
-                        delta: message,
-                        summary_index,
-                    });
+            },
+            classify,
+            options,
+            &cancel,
+            |status| {
+                let human_delay = status
+                    .sleep
+                    .map(format_duration)
+                    .unwrap_or_else(|| "0s".to_string());
+                let elapsed = format_duration(status.elapsed);
+                let prefix = if status.is_rate_limit {
+                    "Rate limit"
+                } else {
+                    "Transient error"
+                };
+                let attempt = status.attempt;
+                let resume_str = status.resume_at.and_then(|resume| {
+                    let now = Instant::now();
+                    if resume <= now {
+                        Some("now".to_string())
+                    } else {
+                        let remaining = resume.duration_since(now);
+                        SystemTime::now()
+                            .checked_add(remaining)
+                            .map(|time| {
+                                let local: DateTime<Local> = time.into();
+                                local.format("%Y-%m-%d %H:%M:%S").to_string()
+                            })
+                    }
+                });
+                let message = format!(
+                    "{prefix} (attempt {attempt}): {}; retrying in {human_delay} (elapsed {elapsed}){}",
+                    status.reason,
+                    resume_str
+                        .map(|s| format!("; next attempt at {s}"))
+                        .unwrap_or_default()
+                );
+                let _ = tx.send(AppEvent::AutoCoordinatorThinking {
+                    delta: message,
+                    summary_index: None,
+                });
+            },
+        )
+        .await
+    });
+
+    match result {
+        Ok(output) => Ok(output),
+        Err(RetryError::Aborted) => Err(anyhow!(AutoCoordinatorCancelled)),
+        Err(RetryError::Fatal(err)) => Err(err),
+        Err(RetryError::Timeout { elapsed, last_error }) => Err(last_error.context(format!(
+            "auto coordinator retry window exceeded after {}",
+            format_duration(elapsed)
+        ))),
+    }
+}
+
+fn build_prompt(
+    developer_intro: &str,
+    primary_goal: &str,
+    schema: &Value,
+    conversation: &[ResponseItem],
+) -> Prompt {
+    let mut prompt = Prompt::default();
+    prompt.store = true;
+    prompt
+        .input
+        .push(make_message("developer", developer_intro.to_string()));
+    prompt
+        .input
+        .push(make_message("developer", primary_goal.to_string()));
+    prompt.input.extend(conversation.iter().cloned());
+    prompt.text_format = Some(TextFormat {
+        r#type: "json_schema".to_string(),
+        name: Some(SCHEMA_NAME.to_string()),
+        strict: Some(true),
+        schema: Some(schema.clone()),
+    });
+    prompt.model_override = Some(MODEL_SLUG.to_string());
+    let family = find_family_for_model(MODEL_SLUG)
+        .unwrap_or_else(|| derive_default_model_family(MODEL_SLUG));
+    prompt.model_family_override = Some(family);
+    prompt
+}
+
+fn classify_model_error(error: &anyhow::Error) -> RetryDecision {
+    if let Some(codex_err) = find_in_chain::<CodexErr>(error) {
+        match codex_err {
+            CodexErr::Stream(message, _) => {
+                return RetryDecision::RetryAfterBackoff {
+                    reason: format!("model stream error: {message}"),
+                };
+            }
+            CodexErr::Timeout => {
+                return RetryDecision::RetryAfterBackoff {
+                    reason: "model request timed out".to_string(),
+                };
+            }
+            CodexErr::UnexpectedStatus(status, body) => {
+                if *status == StatusCode::REQUEST_TIMEOUT || status.as_u16() == 408 {
+                    return RetryDecision::RetryAfterBackoff {
+                        reason: format!("provider returned {status}"),
+                    };
                 }
-                Ok(ResponseEvent::ReasoningContentDelta { delta, .. }) => {
-                    let message = strip_role_prefix(&delta).to_string();
-                    tx.send(AppEvent::AutoCoordinatorThinking {
-                        delta: message,
-                        summary_index: None,
-                    });
+                if status.as_u16() == 499 {
+                    return RetryDecision::RetryAfterBackoff {
+                        reason: "client closed request (499)".to_string(),
+                    };
+                }
+                if *status == StatusCode::TOO_MANY_REQUESTS {
+                    if let Some(wait_until) = parse_rate_limit_hint(body) {
+                        return RetryDecision::RateLimited {
+                            wait_until,
+                            reason: "rate limited; waiting for reset".to_string(),
+                        };
+                    }
+                    return RetryDecision::RetryAfterBackoff {
+                        reason: "rate limited (429)".to_string(),
+                    };
+                }
+                if status.is_client_error() {
+                    return RetryDecision::Fatal(anyhow!(error.to_string()));
+                }
+                if status.is_server_error() {
+                    return RetryDecision::RetryAfterBackoff {
+                        reason: format!("server error {status}"),
+                    };
+                }
+            }
+            CodexErr::UsageLimitReached(limit) => {
+                if let Some(seconds) = limit.resets_in_seconds {
+                    let wait_until = compute_rate_limit_wait(Duration::from_secs(seconds));
+                    return RetryDecision::RateLimited {
+                        wait_until,
+                        reason: "usage limit reached".to_string(),
+                    };
+                }
+                return RetryDecision::RetryAfterBackoff {
+                    reason: "usage limit reached".to_string(),
+                };
+            }
+            CodexErr::UsageNotIncluded => {
+                return RetryDecision::Fatal(anyhow!(error.to_string()));
+            }
+            CodexErr::ServerError(_) => {
+                return RetryDecision::RetryAfterBackoff {
+                    reason: error.to_string(),
+                };
+            }
+            CodexErr::RetryLimit(status) => {
+                return RetryDecision::Fatal(anyhow!("retry limit exceeded (status {status})"));
+            }
+            CodexErr::Reqwest(req_err) => {
+                return classify_reqwest_error(req_err);
+            }
+            CodexErr::Io(io_err) => {
+                if io_err.kind() == std::io::ErrorKind::TimedOut {
+                    return RetryDecision::RetryAfterBackoff {
+                        reason: "network timeout".to_string(),
+                    };
+                }
+            }
+            _ => {}
+        }
+    }
+
+    if let Some(req_err) = find_in_chain::<reqwest::Error>(error) {
+        return classify_reqwest_error(req_err);
+    }
+
+    if let Some(io_err) = find_in_chain::<std::io::Error>(error) {
+        if io_err.kind() == std::io::ErrorKind::TimedOut {
+            return RetryDecision::RetryAfterBackoff {
+                reason: "network timeout".to_string(),
+            };
+        }
+    }
+
+    RetryDecision::Fatal(anyhow!(error.to_string()))
+}
+
+fn classify_reqwest_error(err: &reqwest::Error) -> RetryDecision {
+    if err.is_timeout() || err.is_connect() || err.is_request() && err.status().is_none() {
+        return RetryDecision::RetryAfterBackoff {
+            reason: format!("network error: {err}"),
+        };
+    }
+
+    if let Some(status) = err.status() {
+        if status == StatusCode::TOO_MANY_REQUESTS {
+            return RetryDecision::RetryAfterBackoff {
+                reason: "rate limited (429)".to_string(),
+            };
+        }
+        if status == StatusCode::REQUEST_TIMEOUT || status.as_u16() == 408 {
+            return RetryDecision::RetryAfterBackoff {
+                reason: format!("provider returned {status}"),
+            };
+        }
+        if status.as_u16() == 499 {
+            return RetryDecision::RetryAfterBackoff {
+                reason: "client closed request (499)".to_string(),
+            };
+        }
+        if status.is_server_error() {
+            return RetryDecision::RetryAfterBackoff {
+                reason: format!("server error {status}"),
+            };
+        }
+        if status.is_client_error() {
+            return RetryDecision::Fatal(anyhow!(err.to_string()));
+        }
+    }
+
+    RetryDecision::Fatal(anyhow!(err.to_string()))
+}
+
+fn parse_rate_limit_hint(body: &str) -> Option<Instant> {
+    let value: serde_json::Value = serde_json::from_str(body).ok()?;
+    let error_obj = value.get("error").unwrap_or(&value);
+
+    if let Some(seconds) = extract_seconds(error_obj) {
+        return Some(compute_rate_limit_wait(seconds));
+    }
+
+    if let Some(reset_at) = extract_reset_at(error_obj) {
+        return Some(reset_at);
+    }
+
+    None
+}
+
+fn extract_seconds(value: &serde_json::Value) -> Option<Duration> {
+    let fields = [
+        "reset_seconds",
+        "reset_in_seconds",
+        "resets_in_seconds",
+        "x-ratelimit-reset",
+        "x-ratelimit-reset-requests",
+    ];
+    for key in fields {
+        if let Some(seconds) = value.get(key) {
+            if let Some(num) = seconds.as_f64() {
+                if num.is_sign_negative() {
+                    continue;
+                }
+                return Some(Duration::from_secs_f64(num));
+            }
+            if let Some(text) = seconds.as_str() {
+                if let Ok(num) = text.parse::<f64>() {
+                    if num.is_sign_negative() {
+                        continue;
+                    }
+                    return Some(Duration::from_secs_f64(num));
                 }
-                Ok(ResponseEvent::Completed { .. }) => break,
-                Err(err) => return Err(anyhow!("model stream error: {err}")),
-                _ => {}
             }
         }
-        Ok(out)
-    })
+    }
+    None
+}
+
+fn extract_reset_at(value: &serde_json::Value) -> Option<Instant> {
+    let reset_at = value.get("reset_at").and_then(|v| v.as_str())?;
+    let parsed = DateTime::parse_from_rfc3339(reset_at)
+        .or_else(|_| DateTime::parse_from_str(reset_at, "%+"))
+        .ok()?;
+    let reset_utc = parsed.with_timezone(&Utc);
+    let now = Utc::now();
+    let duration = reset_utc.signed_duration_since(now).to_std().unwrap_or_default();
+    Some(compute_rate_limit_wait(duration))
+}
+
+fn compute_rate_limit_wait(base: Duration) -> Instant {
+    let mut wait = if base > Duration::ZERO { base } else { Duration::ZERO };
+    wait += RATE_LIMIT_BUFFER;
+    wait += random_jitter(RATE_LIMIT_JITTER_MAX);
+    Instant::now() + wait
+}
+
+fn random_jitter(max: Duration) -> Duration {
+    if max.is_zero() {
+        return Duration::ZERO;
+    }
+    let mut rng = rand::rng();
+    let jitter = rng.random_range(0.0..max.as_secs_f64());
+    Duration::from_secs_f64(jitter)
+}
+
+fn find_in_chain<'a, T: std::error::Error + 'static>(error: &'a anyhow::Error) -> Option<&'a T> {
+    for cause in error.chain() {
+        if let Some(specific) = cause.downcast_ref::<T>() {
+            return Some(specific);
+        }
+    }
+    None
+}
+
+#[cfg(test)]
+pub(crate) use classify_model_error as test_classify_model_error;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use codex_core::error::UsageLimitReachedError;
+
+    #[test]
+    fn rate_limit_hint_uses_reset_seconds() {
+        let body = r#"{"error":{"reset_seconds":5}}"#;
+        let start = Instant::now();
+        let wait = parse_rate_limit_hint(body).expect("expected wait instant");
+        let remaining = wait
+            .checked_duration_since(start)
+            .expect("wait should be in the future");
+        let rem_secs = remaining.as_secs_f64();
+        let min_expected = (RATE_LIMIT_BUFFER + Duration::from_secs(5)).as_secs_f64() - 0.5;
+        let max_expected =
+            (RATE_LIMIT_BUFFER + Duration::from_secs(5) + RATE_LIMIT_JITTER_MAX).as_secs_f64() + 1.0;
+        assert!(rem_secs >= min_expected, "remaining {rem_secs}, min {min_expected}");
+        assert!(rem_secs <= max_expected, "remaining {rem_secs}, max {max_expected}");
+    }
+
+    #[test]
+    fn rate_limit_hint_uses_reset_at() {
+        let reset_at = (Utc::now() + chrono::Duration::seconds(10)).to_rfc3339();
+        let body = format!("{{\"error\":{{\"reset_at\":\"{reset_at}\"}}}}");
+        let wait = parse_rate_limit_hint(&body).expect("expected wait instant");
+        let remaining = wait
+            .checked_duration_since(Instant::now())
+            .expect("wait should be in the future");
+        assert!(
+            remaining.as_secs_f64()
+                >= (RATE_LIMIT_BUFFER + Duration::from_secs(10)).as_secs_f64() - 1.0
+        );
+    }
+
+    #[test]
+    fn classify_identifies_stream_disconnect() {
+        let err = anyhow!(CodexErr::Stream("disconnect".into(), None));
+        match classify_model_error(&err) {
+            RetryDecision::RetryAfterBackoff { reason } => {
+                assert!(reason.contains("disconnect"));
+            }
+            other => panic!("unexpected decision: {:?}", other),
+        }
+    }
+
+    #[test]
+    fn classify_usage_limit_returns_rate_limit() {
+        let err = anyhow!(CodexErr::UsageLimitReached(UsageLimitReachedError {
+            plan_type: None,
+            resets_in_seconds: Some(10),
+        }));
+        match classify_model_error(&err) {
+            RetryDecision::RateLimited { .. } => {}
+            other => panic!("expected rate limit, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn classify_fatal_on_bad_request() {
+        let err = anyhow!(CodexErr::UnexpectedStatus(StatusCode::BAD_REQUEST, "bad".to_string()));
+        match classify_model_error(&err) {
+            RetryDecision::Fatal(_) => {}
+            other => panic!("expected fatal, got {:?}", other),
+        }
+    }
 }
 
 fn parse_decision(raw: &str) -> Result<(CoordinatorDecision, Value)> {

codex-rs/tui/src/chatwidget/faults.rs

@@ -0,0 +1,176 @@
+#![cfg(feature = "dev-faults")]
+
+use anyhow::anyhow;
+use codex_core::error::{CodexErr, UsageLimitReachedError};
+use once_cell::sync::OnceCell;
+use rand::Rng;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+/// Scope flag – currently only `auto_drive` is recognised.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum FaultScope {
+    AutoDrive,
+}
+
+#[derive(Debug, Default)]
+struct FaultConfig {
+    disconnect: AtomicUsize,
+    rate_limit: AtomicUsize,
+    rate_limit_reset: Mutex<Option<FaultReset>>, // optional per-call reset hint
+}
+
+#[derive(Debug, Clone)]
+enum FaultReset {
+    Seconds(u64),
+    Timestamp(Instant),
+}
+
+static CONFIG: OnceCell<HashMap<FaultScope, FaultConfig>> = OnceCell::new();
+
+fn parse_fault_scope() -> Option<FaultScope> {
+    match std::env::var("CODEX_FAULTS_SCOPE").ok().as_deref() {
+        Some("auto_drive") => Some(FaultScope::AutoDrive),
+        _ => None,
+    }
+}
+
+fn parse_reset_hint() -> Option<FaultReset> {
+    if let Some(seconds) = std::env::var("CODEX_FAULTS_429_RESET").ok() {
+        if let Ok(value) = seconds.parse::<u64>() {
+            return Some(FaultReset::Seconds(value));
+        }
+        if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(&seconds) {
+            let instant = Instant::now()
+                + Duration::from_secs(parsed.signed_duration_since(chrono::Utc::now()).num_seconds().clamp(0, i64::MAX) as u64);
+            return Some(FaultReset::Timestamp(instant));
+        }
+        if let Some(stripped) = seconds.strip_prefix("now+") {
+            if let Ok(value) = stripped.trim_end_matches('s').parse::<u64>() {
+                return Some(FaultReset::Seconds(value));
+            }
+        }
+    }
+    None
+}
+
+fn init_config() -> HashMap<FaultScope, FaultConfig> {
+    let mut map = HashMap::new();
+    if let Some(scope) = parse_fault_scope() {
+        if let Ok(spec) = std::env::var("CODEX_FAULTS") {
+            let mut cfg = FaultConfig::default();
+            for entry in spec.split(',').map(str::trim).filter(|s| !s.is_empty()) {
+                if let Some((label, count)) = entry.split_once(':') {
+                    if let Ok(num) = count.parse::<usize>() {
+                        match label {
+                            "disconnect" => cfg.disconnect.store(num, Ordering::Relaxed),
+                            "429" => cfg.rate_limit.store(num, Ordering::Relaxed),
+                            _ => {}
+                        }
+                    }
+                }
+            }
+            *cfg.rate_limit_reset.lock().unwrap() = parse_reset_hint();
+            map.insert(scope, cfg);
+        }
+    }
+    map
+}
+
+fn config() -> &'static HashMap<FaultScope, FaultConfig> {
+    CONFIG.get_or_init(init_config)
+}
+
+fn jitter_seconds(max: Duration) -> f64 {
+    if max.is_zero() {
+        return 0.0;
+    }
+    rand::rng().random_range(0.0..max.as_secs_f64())
+}
+
+/// Represents a fault to inject.
+#[derive(Debug)]
+pub enum InjectedFault {
+    Disconnect,
+    RateLimit { reset_hint: Option<FaultReset> },
+}
+
+/// Determine whether a fault should fire for the given scope.
+pub fn next_fault(scope: FaultScope) -> Option<InjectedFault> {
+    let cfg = config().get(&scope)?;
+    if cfg.disconnect.load(Ordering::Relaxed) > 0 {
+        let remaining = cfg.disconnect.fetch_sub(1, Ordering::Relaxed);
+        if remaining > 0 {
+            tracing::warn!("[faults] inject transient disconnect (remaining {})", remaining - 1);
+            return Some(InjectedFault::Disconnect);
+        }
+    }
+    if cfg.rate_limit.load(Ordering::Relaxed) > 0 {
+        let remaining = cfg.rate_limit.fetch_sub(1, Ordering::Relaxed);
+        if remaining > 0 {
+            tracing::warn!("[faults] inject 429 rate limit (remaining {})", remaining - 1);
+            return Some(InjectedFault::RateLimit {
+                reset_hint: cfg.rate_limit_reset.lock().unwrap().clone(),
+            });
+        }
+    }
+    None
+}
+
+/// Convert a fault into an `anyhow::Error` matching production failures.
+pub fn fault_to_error(fault: InjectedFault) -> anyhow::Error {
+    match fault {
+        InjectedFault::Disconnect => anyhow!("model stream error: stream disconnected before completion"),
+        InjectedFault::RateLimit { reset_hint } => match reset_hint {
+            Some(FaultReset::Seconds(secs)) => anyhow!(CodexErr::UsageLimitReached(UsageLimitReachedError {
+                plan_type: None,
+                resets_in_seconds: Some(secs),
+            })),
+            Some(FaultReset::Timestamp(instant)) => {
+                let reset_at = chrono::Utc::now() + ChronoDuration::from_std(instant.saturating_duration_since(Instant::now())).unwrap_or_else(|_| ChronoDuration::seconds(0));
+                let body = json!({
+                    "error": {
+                        "reset_at": reset_at.to_rfc3339(),
+                    }
+                })
+                .to_string();
+                anyhow!(CodexErr::UnexpectedStatus( StatusCode::TOO_MANY_REQUESTS, body ))
+            }
+            None => anyhow!(CodexErr::UnexpectedStatus(
+                StatusCode::TOO_MANY_REQUESTS,
+                json!({ "error": { "message": "fault injector 429" } }).to_string()
+            )),
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use once_cell::sync::Lazy;
+    use std::env;
+
+    static _ENV_GUARD: Lazy<()> = Lazy::new(|| {
+        env::set_var("CODEX_FAULTS_SCOPE", "auto_drive");
+        env::set_var("CODEX_FAULTS", "disconnect:1,429:2");
+        env::set_var("CODEX_FAULTS_429_RESET", "30");
+    });
+
+    #[test]
+    fn injector_sequence_respects_scope_counts() {
+        let _guard = *_ENV_GUARD;
+        CONFIG.take(); // reset caching for test
+        assert!(config().contains_key(&FaultScope::AutoDrive));
+
+        let f1 = next_fault(FaultScope::AutoDrive);
+        matches!(f1, Some(InjectedFault::Disconnect));
+        let f2 = next_fault(FaultScope::AutoDrive);
+        matches!(f2, Some(InjectedFault::RateLimit { .. }));
+        let f3 = next_fault(FaultScope::AutoDrive);
+        matches!(f3, Some(InjectedFault::RateLimit { .. }));
+        let f4 = next_fault(FaultScope::AutoDrive);
+        assert!(f4.is_none());
+    }
+}

codex-rs/tui/src/chatwidget/retry.rs

@@ -0,0 +1,252 @@
+use std::time::{Duration, Instant};
+
+use anyhow::Error;
+use rand::{rngs::StdRng, Rng, SeedableRng};
+use tokio::time;
+use tokio_util::sync::CancellationToken;
+use tracing::warn;
+
+#[derive(Debug, Clone)]
+pub(crate) struct RetryOptions {
+    pub base_delay: Duration,
+    pub factor: f64,
+    pub max_delay: Duration,
+    pub max_elapsed: Duration,
+    pub jitter_seed: Option<u64>,
+}
+
+impl RetryOptions {
+    pub fn with_defaults(max_elapsed: Duration) -> Self {
+        Self {
+            base_delay: Duration::from_secs(1),
+            factor: 2.0,
+            max_delay: Duration::from_secs(15 * 60),
+            max_elapsed,
+            jitter_seed: None,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum RetryDecision {
+    RetryAfterBackoff { reason: String },
+    RateLimited { wait_until: Instant, reason: String },
+    Fatal(Error),
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct RetryStatus {
+    pub attempt: u32,
+    pub elapsed: Duration,
+    pub sleep: Option<Duration>,
+    pub resume_at: Option<Instant>,
+    pub reason: String,
+    pub is_rate_limit: bool,
+}
+
+#[derive(thiserror::Error, Debug)]
+pub(crate) enum RetryError {
+    #[error("retry aborted")]
+    Aborted,
+    #[error("retry timed out after {elapsed:?}")]
+    Timeout { elapsed: Duration, last_error: Error },
+    #[error(transparent)]
+    Fatal(Error),
+}
+
+pub(crate) async fn retry_with_backoff<F, Fut, T, Classify, StatusCb>(
+    mut run: F,
+    mut classify: Classify,
+    options: RetryOptions,
+    cancel: &CancellationToken,
+    mut status_cb: StatusCb,
+) -> Result<T, RetryError>
+where
+    F: FnMut() -> Fut + Send,
+    Fut: std::future::Future<Output = Result<T, Error>> + Send,
+    T: Send,
+    Classify: FnMut(&Error) -> RetryDecision + Send,
+    StatusCb: FnMut(RetryStatus) + Send,
+{
+    let start_time = Instant::now();
+    let mut attempt: u32 = 0;
+    let mut rng = if let Some(seed) = options.jitter_seed {
+        StdRng::seed_from_u64(seed)
+    } else {
+        let mut thread = rand::rng();
+        StdRng::from_rng(&mut thread)
+    };
+
+    loop {
+        if cancel.is_cancelled() {
+            return Err(RetryError::Aborted);
+        }
+
+        attempt = attempt.saturating_add(1);
+        let output = run().await;
+        match output {
+            Ok(value) => return Ok(value),
+            Err(error) => {
+                let elapsed = start_time.elapsed();
+                if elapsed >= options.max_elapsed {
+                    return Err(RetryError::Timeout {
+                        elapsed,
+                        last_error: error,
+                    });
+                }
+
+                match classify(&error) {
+                    RetryDecision::Fatal(fatal) => return Err(RetryError::Fatal(fatal)),
+                    RetryDecision::RateLimited { wait_until, reason } => {
+                        let now = Instant::now();
+                        if wait_until <= now {
+                            warn!(attempt, elapsed = ?elapsed, "{reason}; retrying immediately");
+                            continue;
+                        }
+                        let sleep = wait_until.duration_since(now);
+                        warn!(attempt, elapsed = ?elapsed, wait = ?sleep, resume_at = ?wait_until, "{reason}");
+                        status_cb(RetryStatus {
+                            attempt,
+                            elapsed,
+                            sleep: Some(sleep),
+                            resume_at: Some(wait_until),
+                            reason,
+                            is_rate_limit: true,
+                        });
+                        wait_with_cancel(cancel, sleep).await?;
+                    }
+                    RetryDecision::RetryAfterBackoff { reason } => {
+                        let sleep = compute_delay(&options, attempt, &mut rng);
+                        let resume_at = Instant::now() + sleep;
+                        warn!(attempt, elapsed = ?elapsed, wait = ?sleep, resume_at = ?resume_at, "{reason}");
+                        status_cb(RetryStatus {
+                            attempt,
+                            elapsed,
+                            sleep: Some(sleep),
+                            resume_at: Some(resume_at),
+                            reason,
+                            is_rate_limit: false,
+                        });
+                        wait_with_cancel(cancel, sleep).await?;
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn compute_delay(options: &RetryOptions, attempt: u32, rng: &mut StdRng) -> Duration {
+    let exponent = attempt.saturating_sub(1) as i32;
+    let factor = options.factor.powi(exponent);
+    let base = options.base_delay.as_secs_f64() * factor;
+    let capped = base.min(options.max_delay.as_secs_f64());
+    if capped <= f64::EPSILON {
+        return Duration::ZERO;
+    }
+
+    let jitter = rng.random_range(0.0..capped);
+    Duration::from_secs_f64(jitter)
+}
+
+async fn wait_with_cancel(cancel: &CancellationToken, duration: Duration) -> Result<(), RetryError> {
+    if duration.is_zero() {
+        return Ok(());
+    }
+
+    tokio::select! {
+        _ = time::sleep(duration) => Ok(()),
+        _ = cancel.cancelled() => Err(RetryError::Aborted),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use anyhow::anyhow;
+    use rand::{rngs::StdRng, SeedableRng};
+
+    #[tokio::test]
+    async fn aborts_immediately_when_cancelled() {
+        let cancel = CancellationToken::new();
+        cancel.cancel();
+        let options = RetryOptions::with_defaults(Duration::from_secs(10));
+        let result = retry_with_backoff(
+            || async { Err(anyhow!("boom")) },
+            |_| RetryDecision::RetryAfterBackoff {
+                reason: "retry".to_string(),
+            },
+            options,
+            &cancel,
+            |_status| {},
+        )
+        .await;
+        assert!(matches!(result, Err(RetryError::Aborted)));
+    }
+
+    #[tokio::test]
+    async fn backoff_uses_seeded_full_jitter() {
+        let cancel = CancellationToken::new();
+        let mut sleeps = Vec::new();
+        let options = RetryOptions {
+            base_delay: Duration::from_millis(100),
+            factor: 2.0,
+            max_delay: Duration::from_millis(800),
+            max_elapsed: Duration::from_secs(1),
+            jitter_seed: Some(7),
+        };
+        let expected_opts = options.clone();
+
+        let result = retry_with_backoff(
+            || async { Err(anyhow!("boom")) },
+            |_| RetryDecision::RetryAfterBackoff {
+                reason: "transient".to_string(),
+            },
+            options,
+            &cancel,
+            |status| {
+                if let Some(delay) = status.sleep {
+                    sleeps.push(delay);
+                }
+            },
+        )
+        .await;
+
+        assert!(matches!(result, Err(RetryError::Timeout { .. })));
+        assert!(!sleeps.is_empty());
+
+        let mut rng = StdRng::seed_from_u64(7);
+        for (idx, observed) in sleeps.iter().enumerate().take(4) {
+            let exponent = idx as i32;
+            let factor = expected_opts.factor.powi(exponent);
+            let cap = (expected_opts.base_delay.as_secs_f64() * factor)
+                .min(expected_opts.max_delay.as_secs_f64());
+            let expected = Duration::from_secs_f64(rng.random_range(0.0..cap));
+            let diff = (observed.as_secs_f64() - expected.as_secs_f64()).abs();
+            assert!(diff < 1e-6, "attempt {idx} diff {diff}");
+        }
+    }
+
+    #[tokio::test]
+    async fn stops_after_max_elapsed_budget() {
+        let cancel = CancellationToken::new();
+        let options = RetryOptions {
+            base_delay: Duration::from_millis(10),
+            factor: 2.0,
+            max_delay: Duration::from_millis(50),
+            max_elapsed: Duration::from_millis(60),
+            jitter_seed: Some(5),
+        };
+        let result = retry_with_backoff(
+            || async { Err(anyhow!("boom")) },
+            |_| RetryDecision::RetryAfterBackoff {
+                reason: "transient".to_string(),
+            },
+            options,
+            &cancel,
+            |_status| {},
+        )
+        .await;
+
+        assert!(matches!(result, Err(RetryError::Timeout { .. })));
+    }
+}

codex-rs/tui/src/chatwidget/tests.rs

@@ -2115,6 +2115,63 @@ fn two_final_answers_append_not_overwrite_when_no_deltas() {
     assert!(assistants.iter().any(|s| s.contains("Second message")), "missing second message");
 }
 
+#[test]
+fn export_preserves_user_role_and_content_type() {
+    let (mut chat, rx, _op_rx) = make_chatwidget_manual();
+
+    chat.submit_text_message("Run the integration tests".to_string());
+
+    // Drain any synthetic history updates generated during submission.
+    let _ = drain_insert_history(&rx);
+
+    let items = chat.export_response_items();
+    let mut user_items = items.iter().filter_map(|item| match item {
+        codex_protocol::models::ResponseItem::Message { role, content, .. }
+            if role == "user" =>
+        {
+            Some(content)
+        }
+        _ => None,
+    });
+
+    let content = user_items
+        .next()
+        .expect("expected at least one exported user item");
+
+    assert!(matches!(
+        content.first(),
+        Some(codex_protocol::models::ContentItem::InputText { .. })
+    ));
+}
+
+#[test]
+fn export_last_user_turn_detectable_after_assistant_reply() {
+    let (mut chat, rx, _op_rx) = make_chatwidget_manual();
+
+    chat.submit_text_message("Please summarize the diff".to_string());
+    chat.insert_final_answer_with_id(None, Vec::new(), "Sure, here's the summary.".to_string());
+
+    let _ = drain_insert_history(&rx);
+
+    let items = chat.export_response_items();
+    let last_user_index = items
+        .iter()
+        .enumerate()
+        .rev()
+        .find(|(_, item)| matches!(
+            item,
+            codex_protocol::models::ResponseItem::Message { role, content, .. }
+                if role == "user"
+                    && content.iter().any(|c| matches!(
+                        c,
+                        codex_protocol::models::ContentItem::InputText { .. }
+                    ))
+        ))
+        .map(|(idx, _)| idx);
+
+    assert_eq!(last_user_index, Some(0), "expected latest user prompt index");
+}
+
 #[test]
 fn second_final_that_is_superset_replaces_first() {
     let (mut chat, rx, _op_rx) = make_chatwidget_manual();

codex-rs/tui/src/chatwidget/tests_retry.rs

@@ -0,0 +1,349 @@
+use super::auto_coordinator::test_classify_model_error;
+use super::retry::{retry_with_backoff, RetryDecision, RetryError, RetryOptions, RetryStatus};
+use anyhow::{anyhow, Error};
+use chrono::{Duration as ChronoDuration, Utc};
+use codex_core::error::{CodexErr, UsageLimitReachedError};
+use rand::{rngs::StdRng, Rng, SeedableRng};
+use reqwest::StatusCode;
+use serde_json::json;
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::task::yield_now;
+use tokio::time::advance;
+use tokio_util::sync::CancellationToken;
+
+fn expected_backoffs(options: &RetryOptions, count: usize) -> Vec<Duration> {
+    let mut rng = StdRng::seed_from_u64(options.jitter_seed.expect("expected deterministic seed"));
+    (1..=count)
+        .map(|attempt| {
+            let exponent = attempt.saturating_sub(1) as i32;
+            let cap = (options.base_delay.as_secs_f64() * options.factor.powi(exponent))
+                .min(options.max_delay.as_secs_f64());
+            if cap <= f64::EPSILON {
+                Duration::ZERO
+            } else {
+                Duration::from_secs_f64(rng.random_range(0.0..cap))
+            }
+        })
+        .collect()
+}
+
+async fn wait_for_status_len(statuses: &Arc<Mutex<Vec<RetryStatus>>>, len: usize) {
+    loop {
+        if statuses.lock().unwrap().len() >= len {
+            break;
+        }
+        yield_now().await;
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn retry_retries_transient_disconnect_then_succeeds() {
+    let cancel = CancellationToken::new();
+    let options = RetryOptions {
+        base_delay: Duration::from_secs(4),
+        factor: 2.0,
+        max_delay: Duration::from_secs(60),
+        max_elapsed: Duration::from_secs(3600),
+        jitter_seed: Some(42),
+    };
+
+    let attempts = Arc::new(Mutex::new(VecDeque::from([
+        Err(anyhow!("model stream error: stream disconnected before completion")),
+        Err(anyhow!("model stream error: stream disconnected before completion")),
+        Ok::<(), Error>(()),
+    ])));
+
+    let statuses: Arc<Mutex<Vec<RetryStatus>>> = Arc::new(Mutex::new(Vec::new()));
+    let run_attempts = attempts.clone();
+    let status_log = statuses.clone();
+
+    let task = tokio::spawn(retry_with_backoff(
+        move || {
+            let run_attempts = run_attempts.clone();
+            async move {
+                let mut guard = run_attempts.lock().unwrap();
+                guard.pop_front().expect("attempt available")
+            }
+        },
+        |err| {
+            if err.to_string().contains("stream disconnected") {
+                RetryDecision::RetryAfterBackoff {
+                    reason: err.to_string(),
+                }
+            } else {
+                RetryDecision::Fatal(anyhow!(err.to_string()))
+            }
+        },
+        options.clone(),
+        &cancel,
+        move |status| {
+            status_log.lock().unwrap().push(status);
+        },
+    ));
+
+    wait_for_status_len(&statuses, 1).await;
+    let sleep1 = statuses.lock().unwrap()[0]
+        .sleep
+        .expect("sleep recorded");
+    advance(sleep1).await;
+    yield_now().await;
+
+    wait_for_status_len(&statuses, 2).await;
+    let sleep2 = statuses.lock().unwrap()[1]
+        .sleep
+        .expect("sleep recorded");
+    advance(sleep2).await;
+    yield_now().await;
+
+    let result = task.await.unwrap();
+    assert!(result.is_ok(), "retry should eventually succeed");
+
+    let recorded = statuses.lock().unwrap();
+    assert_eq!(recorded.len(), 2, "expected two backoff sleeps");
+
+    let expected = expected_backoffs(&options, 2);
+    for (observed, expected_delay) in recorded.iter().zip(expected) {
+        let sleep = observed.sleep.expect("sleep");
+        let delta = (sleep.as_secs_f64() - expected_delay.as_secs_f64()).abs();
+        assert!(delta < 1e-6, "sleep {:?} differed from expected {:?}", sleep, expected_delay);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn retry_respects_reset_seconds_rate_limit() {
+    let cancel = CancellationToken::new();
+    let options = RetryOptions {
+        base_delay: Duration::from_secs(2),
+        factor: 2.0,
+        max_delay: Duration::from_secs(30),
+        max_elapsed: Duration::from_secs(3600),
+        jitter_seed: Some(7),
+    };
+
+    let body = json!({
+        "error": {
+            "resets_in_seconds": 60
+        }
+    })
+    .to_string();
+    let rate_limit_err = anyhow!(CodexErr::UnexpectedStatus(
+        StatusCode::TOO_MANY_REQUESTS,
+        body
+    ));
+
+    let attempts = Arc::new(Mutex::new(VecDeque::from([
+        Err(rate_limit_err),
+        Ok::<(), Error>(()),
+    ])));
+
+    let statuses: Arc<Mutex<Vec<RetryStatus>>> = Arc::new(Mutex::new(Vec::new()));
+    let run_attempts = attempts.clone();
+    let status_log = statuses.clone();
+
+    let task = tokio::spawn(retry_with_backoff(
+        move || {
+            let run_attempts = run_attempts.clone();
+            async move {
+                let mut guard = run_attempts.lock().unwrap();
+                guard.pop_front().expect("attempt available")
+            }
+        },
+        |err| test_classify_model_error(err),
+        options,
+        &cancel,
+        move |status| {
+            status_log.lock().unwrap().push(status);
+        },
+    ));
+
+    wait_for_status_len(&statuses, 1).await;
+    let status = statuses.lock().unwrap()[0].clone();
+    assert!(status.is_rate_limit, "expected rate limit status");
+    assert!(status.reason.contains("rate") || status.reason.contains("usage"));
+    let sleep = status.sleep.expect("sleep duration");
+
+    // reset (60s) + buffer (120s) + jitter (<=30s)
+    let sleep_secs = sleep.as_secs_f64();
+    assert!(sleep_secs >= 180.0 - 0.5, "sleep too short: {sleep_secs}");
+    assert!(sleep_secs <= 210.0 + 1.0, "sleep too long: {sleep_secs}");
+
+    advance(sleep).await;
+    yield_now().await;
+
+    let result = task.await.unwrap();
+    assert!(result.is_ok());
+}
+
+#[tokio::test(start_paused = true)]
+async fn retry_respects_reset_at_rate_limit() {
+    let cancel = CancellationToken::new();
+    let options = RetryOptions {
+        base_delay: Duration::from_secs(1),
+        factor: 2.0,
+        max_delay: Duration::from_secs(30),
+        max_elapsed: Duration::from_secs(3600),
+        jitter_seed: Some(9),
+    };
+
+    let reset_at = (Utc::now() + ChronoDuration::seconds(45)).to_rfc3339();
+    let body = json!({
+        "error": {
+            "reset_at": reset_at
+        }
+    })
+    .to_string();
+    let rate_limit_err = anyhow!(CodexErr::UnexpectedStatus(
+        StatusCode::TOO_MANY_REQUESTS,
+        body
+    ));
+
+    let attempts = Arc::new(Mutex::new(VecDeque::from([
+        Err(rate_limit_err),
+        Ok::<(), Error>(()),
+    ])));
+
+    let statuses: Arc<Mutex<Vec<RetryStatus>>> = Arc::new(Mutex::new(Vec::new()));
+    let run_attempts = attempts.clone();
+    let status_log = statuses.clone();
+
+    let task = tokio::spawn(retry_with_backoff(
+        move || {
+            let run_attempts = run_attempts.clone();
+            async move {
+                let mut guard = run_attempts.lock().unwrap();
+                guard.pop_front().expect("attempt available")
+            }
+        },
+        |err| test_classify_model_error(err),
+        options,
+        &cancel,
+        move |status| {
+            status_log.lock().unwrap().push(status);
+        },
+    ));
+
+    wait_for_status_len(&statuses, 1).await;
+    let status = statuses.lock().unwrap()[0].clone();
+    assert!(status.is_rate_limit);
+    let sleep = status.sleep.expect("sleep duration");
+    let sleep_secs = sleep.as_secs_f64();
+    assert!(sleep_secs >= 120.0 - 1.0, "sleep too short: {sleep_secs}");
+    assert!(sleep_secs <= 150.0 + 1.5, "sleep too long: {sleep_secs}");
+
+    advance(sleep).await;
+    yield_now().await;
+
+    let result = task.await.unwrap();
+    assert!(result.is_ok());
+}
+
+#[tokio::test(start_paused = true)]
+async fn retry_rate_limit_without_reset_falls_back_to_backoff() {
+    let cancel = CancellationToken::new();
+    let options = RetryOptions {
+        base_delay: Duration::from_secs(3),
+        factor: 2.0,
+        max_delay: Duration::from_secs(30),
+        max_elapsed: Duration::from_secs(3600),
+        jitter_seed: Some(12),
+    };
+
+    let body = json!({ "error": { "message": "slow down" } }).to_string();
+    let rate_limit_err = anyhow!(CodexErr::UnexpectedStatus(
+        StatusCode::TOO_MANY_REQUESTS,
+        body
+    ));
+
+    let attempts = Arc::new(Mutex::new(VecDeque::from([
+        Err(rate_limit_err),
+        Ok::<(), Error>(()),
+    ])));
+
+    let statuses: Arc<Mutex<Vec<RetryStatus>>> = Arc::new(Mutex::new(Vec::new()));
+    let run_attempts = attempts.clone();
+    let status_log = statuses.clone();
+
+    let task = tokio::spawn(retry_with_backoff(
+        move || {
+            let run_attempts = run_attempts.clone();
+            async move {
+                let mut guard = run_attempts.lock().unwrap();
+                guard.pop_front().expect("attempt available")
+            }
+        },
+        |err| test_classify_model_error(err),
+        options.clone(),
+        &cancel,
+        move |status| {
+            status_log.lock().unwrap().push(status);
+        },
+    ));
+
+    wait_for_status_len(&statuses, 1).await;
+    let status = statuses.lock().unwrap()[0].clone();
+    assert!(
+        !status.is_rate_limit,
+        "should fall back to exponential backoff when no reset hints are present"
+    );
+    let expected = expected_backoffs(&options, 1)[0];
+    let sleep = status.sleep.expect("sleep duration");
+    let delta = (sleep.as_secs_f64() - expected.as_secs_f64()).abs();
+    assert!(delta < 1e-6);
+
+    advance(sleep).await;
+    yield_now().await;
+
+    let result = task.await.unwrap();
+    assert!(result.is_ok());
+}
+
+#[tokio::test(start_paused = true)]
+async fn retry_cancellation_interrupts_sleep() {
+    let cancel = CancellationToken::new();
+    let options = RetryOptions {
+        base_delay: Duration::from_secs(5),
+        factor: 2.0,
+        max_delay: Duration::from_secs(30),
+        max_elapsed: Duration::from_secs(3600),
+        jitter_seed: Some(21),
+    };
+
+    let rate_limit_err = anyhow!(CodexErr::UsageLimitReached(UsageLimitReachedError {
+        plan_type: None,
+        resets_in_seconds: Some(90),
+    }));
+
+    let attempts = Arc::new(Mutex::new(VecDeque::from([
+        Err(rate_limit_err),
+    ])));
+
+    let statuses: Arc<Mutex<Vec<RetryStatus>>> = Arc::new(Mutex::new(Vec::new()));
+    let run_attempts = attempts.clone();
+    let status_log = statuses.clone();
+
+    let cancel_clone = cancel.clone();
+    let task = tokio::spawn(retry_with_backoff(
+        move || {
+            let run_attempts = run_attempts.clone();
+            async move {
+                let mut guard = run_attempts.lock().unwrap();
+                guard.pop_front().unwrap()
+            }
+        },
+        |err| test_classify_model_error(err),
+        options,
+        &cancel_clone,
+        move |status| {
+            status_log.lock().unwrap().push(status);
+        },
+    ));
+
+    wait_for_status_len(&statuses, 1).await;
+    cancel.cancel();
+
+    let result = task.await.unwrap();
+    assert!(matches!(result, Err(RetryError::Aborted)));
+}
+