diff --git a/crates/agentkit-loop/src/lib.rs b/crates/agentkit-loop/src/lib.rs
index 04db465..ec8aef7 100644
--- a/crates/agentkit-loop/src/lib.rs
+++ b/crates/agentkit-loop/src/lib.rs
@@ -90,6 +90,12 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use thiserror::Error;
 
+mod retry;
+pub use retry::{
+    ProviderClassification, ProviderFailure, ProviderFailureReason, ProviderRetryEvent,
+    ProviderRoute, RetryAccounting, RetryObserver, RetryProgress, UpstreamErrorKind,
+};
+
 const INTERRUPTED_METADATA_KEY: &str = "agentkit.interrupted";
 const INTERRUPT_REASON_METADATA_KEY: &str = "agentkit.interrupt_reason";
 const INTERRUPT_STAGE_METADATA_KEY: &str = "agentkit.interrupt_stage";
@@ -663,6 +669,10 @@ pub trait ModelAdapter: Send + Sync {
 /// [`ModelTurn`].
 #[async_trait]
 pub trait ModelSession: Send {
+    /// Install a per-session observer for retries inside both begin_turn and next_event.
+    /// The default is a no-op for adapters without retry observations.
+    fn set_retry_observer(&mut self, _observer: Option<Arc<dyn RetryObserver>>) {}
+
     /// The turn type produced by this session.
     type Turn: ModelTurn;
 
@@ -709,6 +719,11 @@ pub trait ModelSession: Send {
 /// `Ok(Some(ModelTurnEvent::Finished(_)))`.
 #[async_trait]
 pub trait ModelTurn: Send {
+    /// Notifies the turn before the driver drops it after explicit cancellation
+    /// observed between events. This synchronous hook must not block.
+    /// The default is a no-op; adapters can finalize retry accounting here.
+    fn on_cancelled(&mut self) {}
+
     /// Retrieve the next event from the model's response stream.
     ///
     /// Returns `Ok(None)` when the stream is exhausted.
@@ -908,6 +923,8 @@ pub trait LoopMutator: Send + Sync {
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[non_exhaustive]
 pub enum AgentEvent {
+    /// Sanitized provider retry lifecycle, separate from content and effects.
+    ProviderRetry(ProviderRetryEvent),
     /// The agent run has been initialised.
     RunStarted { session_id: SessionId },
     /// A new logical turn is starting.
@@ -1386,7 +1403,18 @@ where
     pub async fn start(&self, config: SessionConfig) -> Result<LoopDriver<M::Session>, LoopError> {
         let session_id = config.session_id.clone();
         let default_cache = config.cache.clone();
-        let session = self.model.start_session(config).await?;
+        let mut session = self.model.start_session(config).await?;
+        if !self.observers.is_empty() {
+            let observers = self.observers.clone();
+            let observed_session_id = Arc::new(session_id.clone());
+            session.set_retry_observer(Some(Arc::new(move |event| {
+                fan_out_observed_event(
+                    &observers,
+                    &observed_session_id,
+                    AgentEvent::ProviderRetry(event),
+                );
+            })));
+        }
         let provider_name = self.model.provider_name().map(str::to_owned);
         let tool_executor = self
             .tool_executor
@@ -2390,6 +2418,7 @@ where
                 .as_ref()
                 .is_some_and(TurnCancellation::is_cancelled)
             {
+                turn.on_cancelled();
                 self.task_manager
                     .on_turn_interrupted(&turn_id)
                     .await
@@ -4226,6 +4255,9 @@ fn tool_result_not_started(item: &Item) -> bool {
 /// Errors that can occur while driving the agent loop.
 #[derive(Debug, Error)]
 pub enum LoopError {
+    /// Typed, sanitized model failure with retry accounting.
+    #[error(transparent)]
+    ProviderFailure(Box<ProviderFailure>),
     /// The driver was in an unexpected state for the requested operation.
     #[error("invalid driver state: {0}")]
     InvalidState(String),
@@ -4246,6 +4278,16 @@ pub enum LoopError {
     Unsupported(String),
 }
 
+impl LoopError {
+    /// Returns structured provider metadata without parsing a rendered error.
+    pub fn provider_failure(&self) -> Option<&ProviderFailure> {
+        match self {
+            Self::ProviderFailure(failure) => Some(failure),
+            _ => None,
+        }
+    }
+}
+
 /// Internal [`EventEmitter`] backed by the driver's observer slice. Lives
 /// only for the duration of a [`LoopDriver::run_mutators`] call so the
 /// borrow against `self.observers` stays disjoint from the cursor's borrow
diff --git a/crates/agentkit-loop/src/retry.rs b/crates/agentkit-loop/src/retry.rs
new file mode 100644
index 0000000..d92a55a
--- /dev/null
+++ b/crates/agentkit-loop/src/retry.rs
@@ -0,0 +1,253 @@
+//! Sanitized, payload-free model retry observations. These are not effects provenance.
+
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+
+/// Static provider route; never an endpoint URL or account identifier.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum ProviderRoute {
+    #[default]
+    Unknown,
+    OpenAiResponses,
+    OpenAiChatGptResponses,
+}
+
+/// Allowlisted provider type/code values. Unknown strings are never retained.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum UpstreamErrorKind {
+    ServiceUnavailableError,
+    ServerIsOverloaded,
+    ServerError,
+    RateLimitError,
+    RateLimitExceeded,
+    TemporarilyUnavailable,
+    AuthenticationError,
+    InvalidApiKey,
+    InvalidAuthentication,
+    Unauthorized,
+    InvalidRequestError,
+    PermissionDenied,
+    InsufficientQuota,
+    ContentPolicyViolation,
+    #[default]
+    #[serde(other)]
+    Unknown,
+}
+
+/// Sanitized source classification, kept separate from the local stopping reason.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ProviderClassification {
+    pub error_type: UpstreamErrorKind,
+    pub code: UpstreamErrorKind,
+    /// Source HTTP status, if present. No headers or response body are retained.
+    pub http_status: Option<u16>,
+}
+
+/// Local reason for a failed attempt or logical request.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum ProviderFailureReason {
+    HttpStatus,
+    Transport,
+    ResponseFailed,
+    Protocol,
+    InvalidRequest,
+    Authentication,
+    AttemptTimeout,
+    IdleTimeout,
+    RetryExhausted,
+    RetryBudget,
+    RetryDisabled,
+    ReplayUnsafe,
+    Cancelled,
+}
+
+/// Per-logical-request accounting, independent of policy retry count.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RetryAccounting {
+    /// Actual HTTP sends started, including a resend after authentication refresh.
+    /// Preflight/authentication failures can have zero attempts.
+    pub attempts: u64,
+    /// Sum of requested durations of fully completed backoff waits. Interrupted
+    /// waits contribute zero, even when they consumed wall-clock time.
+    pub completed_backoff: Duration,
+    /// Monotonic elapsed time since before initial authentication/preflight.
+    pub elapsed: Duration,
+}
+
+/// A nonterminal snapshot emitted before a retry wait or reactive refresh.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RetryProgress {
+    pub route: ProviderRoute,
+    pub reason: ProviderFailureReason,
+    pub upstream: ProviderClassification,
+    /// Attempts already started. The planned next send is `attempts + 1`, but
+    /// cancellation/preflight failure can prevent it from ever starting.
+    pub accounting: RetryAccounting,
+    pub next_delay: Duration,
+}
+
+/// Payload-free terminal model failure. Display and Debug contain only typed data.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
+#[error("provider request failed ({reason:?}; attempts: {attempts})", attempts = .accounting.attempts)]
+pub struct ProviderFailure {
+    pub route: ProviderRoute,
+    pub reason: ProviderFailureReason,
+    /// Last failed request-attempt category, retained across local budget/limit stops.
+    /// None when no request attempt failed (for example initial authentication).
+    pub last_attempt_reason: Option<ProviderFailureReason>,
+    pub upstream: ProviderClassification,
+    pub accounting: RetryAccounting,
+}
+
+/// Observational lifecycle; never a second model result or a tool-effects record.
+///
+/// Correlate through the enclosing `ObservedEvent.session_id` and current
+/// `AgentEvent::TurnStarted`. Direct session consumers own that association.
+/// Stable fatal event IDs belong to the host, not to this payload.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[non_exhaustive]
+pub enum ProviderRetryEvent {
+    Scheduled(RetryProgress),
+    /// Emitted once on explicit failure/cancellation, including zero-send failures.
+    Stopped(ProviderFailure),
+    /// Clears retry activity without introducing another successful model result.
+    Succeeded {
+        route: ProviderRoute,
+        accounting: RetryAccounting,
+    },
+}
+
+/// Synchronous, queue-free observer installed before `begin_turn`.
+///
+/// Implementations must not block or re-enter the session. Panics propagate, just
+/// like loop observers; delivery cannot be guaranteed if observers panic/block.
+/// Dropping a future or turn is not explicit cancellation and does not promise a
+/// terminal observation. Implementations should rate-limit Scheduled snapshots
+/// per logical turn, retaining exact accounting and unsuppressed terminal events.
+pub trait RetryObserver: Send + Sync {
+    fn on_retry_event(&self, event: ProviderRetryEvent);
+}
+
+impl<F: Fn(ProviderRetryEvent) + Send + Sync> RetryObserver for F {
+    fn on_retry_event(&self, event: ProviderRetryEvent) {
+        self(event);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::AgentEvent;
+
+    #[test]
+    fn event_roundtrip_and_legacy_shape_remain_compatible() {
+        let old = r#"{"RunStarted":{"session_id":"session"}}"#;
+        let event: AgentEvent = serde_json::from_str(old).unwrap();
+        assert_eq!(serde_json::to_string(&event).unwrap(), old);
+        let current = AgentEvent::ProviderRetry(ProviderRetryEvent::Stopped(ProviderFailure {
+            route: ProviderRoute::OpenAiChatGptResponses,
+            reason: ProviderFailureReason::RetryExhausted,
+            last_attempt_reason: Some(ProviderFailureReason::ResponseFailed),
+            upstream: ProviderClassification {
+                error_type: UpstreamErrorKind::ServiceUnavailableError,
+                code: UpstreamErrorKind::ServerIsOverloaded,
+                http_status: None,
+            },
+            accounting: RetryAccounting {
+                attempts: 3,
+                completed_backoff: Duration::from_millis(125),
+                elapsed: Duration::from_millis(250),
+            },
+        }));
+        let encoded = serde_json::to_string(&current).unwrap();
+        assert_eq!(
+            serde_json::from_str::<AgentEvent>(&encoded).unwrap(),
+            current
+        );
+        assert!(encoded.contains("service_unavailable_error"));
+        assert!(encoded.contains("server_is_overloaded"));
+        assert!(serde_json::from_str::<AgentEvent>(r#"{"ProviderRetry":{"Stopped":{}}}"#).is_err());
+        assert_eq!(
+            serde_json::from_str::<UpstreamErrorKind>(r#""future-private-value""#).unwrap(),
+            UpstreamErrorKind::Unknown
+        );
+    }
+}
+
+#[cfg(test)]
+mod cancellation_hook_tests {
+    use crate::{
+        Agent, LoopError, ModelAdapter, ModelSession, ModelTurn, ModelTurnEvent, SessionConfig,
+        TurnRequest,
+    };
+    use agentkit_core::{CancellationController, Item, ItemKind, TurnCancellation, Usage};
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
+
+    #[derive(Clone)]
+    struct CancellingModel {
+        controller: Arc<CancellationController>,
+        notified: Arc<AtomicUsize>,
+    }
+
+    #[async_trait::async_trait]
+    impl ModelAdapter for CancellingModel {
+        type Session = Self;
+        async fn start_session(&self, _: SessionConfig) -> Result<Self, LoopError> {
+            Ok(self.clone())
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl ModelSession for CancellingModel {
+        type Turn = Self;
+        async fn begin_turn(
+            &mut self,
+            _: TurnRequest,
+            _: Option<TurnCancellation>,
+        ) -> Result<Self, LoopError> {
+            Ok(self.clone())
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl ModelTurn for CancellingModel {
+        fn on_cancelled(&mut self) {
+            self.notified.fetch_add(1, Ordering::SeqCst);
+        }
+        async fn next_event(
+            &mut self,
+            _: Option<TurnCancellation>,
+        ) -> Result<Option<ModelTurnEvent>, LoopError> {
+            self.controller.interrupt();
+            Ok(Some(ModelTurnEvent::Usage(Usage::default())))
+        }
+    }
+
+    #[tokio::test]
+    async fn driver_notifies_before_dropping_a_turn_cancelled_between_events() {
+        let controller = Arc::new(CancellationController::new());
+        let notified = Arc::new(AtomicUsize::new(0));
+        let agent = Agent::builder()
+            .model(CancellingModel {
+                controller: controller.clone(),
+                notified: notified.clone(),
+            })
+            .cancellation(controller.handle())
+            .input(vec![Item::text(ItemKind::User, "hello")])
+            .build()
+            .unwrap();
+        let mut driver = agent.start(SessionConfig::new("session")).await.unwrap();
+        driver.next().await.unwrap();
+        assert_eq!(notified.load(Ordering::SeqCst), 1);
+    }
+}
diff --git a/crates/agentkit-provider-openai/README.md b/crates/agentkit-provider-openai/README.md
index 62c6bc6..cfea30c 100644
--- a/crates/agentkit-provider-openai/README.md
+++ b/crates/agentkit-provider-openai/README.md
@@ -76,6 +76,72 @@ these with `OpenAIResponsesLimits` and
 Limits must be non-zero; the per-field bound must fit both request and attempt
 bounds, and the per-attempt bound must fit the aggregate wire bound.
 
+## Retry observations and typed failures
+
+Responses emits `agentkit_loop::ProviderRetryEvent` through
+`AgentEvent::ProviderRetry` to registered loop observers. Initial HTTP retries
+are visible while `begin_turn` is still pending; stream retries are visible
+before their deferred backoff, including before an attempt-supersession marker.
+Direct adapter consumers can call `ModelSession::set_retry_observer` before
+`begin_turn`. Other adapters retain the default no-op implementation.
+
+- `Scheduled(RetryProgress)` carries a canonical route, attempt reason,
+  allowlisted upstream type/code, HTTP status when available, next delay, and
+  cumulative accounting. The first snapshot is immediate; subsequent snapshots
+  are limited to one per 250 ms per model request (`begin_turn` invocation). Suppressed updates do not
+  discard accounting. There is no queue, heartbeat, or trailing-update timer.
+- `Stopped(ProviderFailure)` reports explicit failure or cancellation once.
+  `Succeeded { route, accounting }` clears activity once on successful model
+  completion. These observations are not additional model results or tool-effects
+  records. Polling a completed/failed turn again produces no duplicate terminal
+  observation. Dropping a future/turn does not promise terminal delivery.
+- Callbacks are synchronous, infallible at the interface, and run without
+  holding decoder/session locks. They must not block or re-enter the session.
+  As with loop observers, panics propagate; blocking/panicking observers prevent
+  delivery guarantees.
+
+`RetryAccounting::attempts` counts polled HTTP client executions, including a
+resend after reactive authentication refresh, not policy retries plus one.
+Initial authentication and preflight failures can have zero attempts. Malformed
+endpoints, non-HTTP(S) endpoints, and invalid local attribution headers are
+rejected as `InvalidRequest` before transport execution, without retrying.
+`completed_backoff` sums the requested durations of fully completed waits;
+interrupted waits contribute zero, while earlier completed waits remain counted.
+`elapsed` measures monotonic time from before authentication and includes
+preflight, callbacks, requests, and interrupted waits. Cancellation and logical
+deadlines are checked before accepting a ready completion, so a cancelled or
+budget-expired wait is not counted as completed.
+
+Responses model-session failures use `LoopError::ProviderFailure`;
+`LoopError::provider_failure()` exposes the typed payload without string parsing.
+`reason` distinguishes retry-count exhaustion, logical budget expiry, disabled
+retry, unsafe replay after output, authentication, and protocol/transport failure.
+`last_attempt_reason` retains the failed attempt category, such as attempt/idle
+timeout, across a local exhaustion/budget stop. The last source response's
+`upstream` classification survives local stops and authentication refresh failures.
+A later HTTP/SSE failure replaces it, even when its type/code are unknown.
+Unknown, missing, or malformed provider type/code values become `Unknown`;
+provider messages, bodies, headers, credentials, prompts, endpoint URLs, and
+customer identifiers never enter these payloads, their Debug, or their Display.
+Non-success HTTP bodies are not read to obtain classifications.
+
+`LoopError::Cancelled` remains unchanged for cancellation-aware callers; its
+accounting is available in the `Stopped` observation. The driver's explicit
+post-event cancellation boundary calls `ModelTurn::on_cancelled` before dropping
+the turn, so terminal accounting does not require another event poll. Direct
+consumers that explicitly abandon a turn can use the same synchronous hook.
+Correlate observations
+through `ObservedEvent.session_id` and the current `TurnStarted` event. Direct
+session consumers own that association. A host must add its own stable fatal
+event reference when projecting this contract into parent-visible failures.
+
+**Compatibility:** the new `LoopError` variant requires updating exhaustive
+error matches. Existing serialized `AgentEvent` variants are unchanged, but old
+readers can reject the new variant; coordinate reader updates and compatible
+crate releases before sending these events across a versioned boundary. This
+upstream API does not implement ACP parent delivery, fatal-record persistence,
+or stable fatal-record correlation.
+
 ## Responses API
 
 `OpenAIResponsesConfig::new(authentication, model)` deliberately matches
diff --git a/crates/agentkit-provider-openai/src/responses.rs b/crates/agentkit-provider-openai/src/responses.rs
index bc122b1..9c5dc57 100644
--- a/crates/agentkit-provider-openai/src/responses.rs
+++ b/crates/agentkit-provider-openai/src/responses.rs
@@ -21,7 +21,9 @@ use agentkit_http::{
 };
 use agentkit_loop::{
     LoopError, ModelAdapter, ModelSession, ModelTurn, ModelTurnEvent, ModelTurnResult,
-    PromptCacheMode, PromptCacheStrategy, SessionConfig, TurnRequest, set_provider_finish_reasons,
+    PromptCacheMode, PromptCacheStrategy, ProviderClassification, ProviderFailure,
+    ProviderFailureReason, ProviderRetryEvent, ProviderRoute, RetryAccounting, RetryObserver,
+    RetryProgress, SessionConfig, TurnRequest, UpstreamErrorKind, set_provider_finish_reasons,
 };
 use async_trait::async_trait;
 use futures_util::future::{Either, select};
@@ -29,6 +31,9 @@ use serde_json::{Map, Value, json};
 use thiserror::Error;
 use zeroize::{Zeroize, Zeroizing};
 
+mod retry;
+use retry::{RetryTracker, local_error, provider_error, stream_classification};
+
 const PUBLIC_ENDPOINT: &str = "https://api.openai.com/v1/responses";
 const PRIVATE_ENDPOINT: &str = "https://chatgpt.com/backend-api/codex/responses";
 const DEFAULT_MAX_REQUEST_BYTES: usize = 32 * 1024 * 1024;
@@ -346,6 +351,7 @@ impl ModelAdapter for OpenAIResponsesAdapter {
             client: self.client.clone(),
             config: self.config.clone(),
             session: config,
+            retry_observer: None,
         })
     }
 
@@ -359,61 +365,82 @@ pub struct OpenAIResponsesSession {
     client: Http,
     config: Arc<OpenAIResponsesConfig>,
     session: SessionConfig,
+    retry_observer: Option<Arc<dyn RetryObserver>>,
 }
 
 #[async_trait]
 impl ModelSession for OpenAIResponsesSession {
     type Turn = OpenAIResponsesTurn;
 
+    fn set_retry_observer(&mut self, observer: Option<Arc<dyn RetryObserver>>) {
+        self.retry_observer = observer;
+    }
+
     async fn begin_turn(
         &mut self,
         request: TurnRequest,
         cancellation: Option<TurnCancellation>,
     ) -> Result<Self::Turn, LoopError> {
-        if cancelled(cancellation.as_ref()) {
-            return Err(LoopError::Cancelled);
+        let mut tracker = RetryTracker::new(self.config.profile, self.retry_observer.clone());
+        let prepared = async {
+            if cancelled(cancellation.as_ref()) {
+                return Err(LoopError::Cancelled);
+            }
+            // One logical deadline starts before encoding and initial authentication.
+            let deadline = self
+                .config
+                .resilience
+                .as_ref()
+                .map(|config| LogicalDeadline {
+                    started_at: tracker.started_at,
+                    budget: config.retry_budget,
+                });
+            deadline_remaining(deadline.as_ref()).map_err(http_loop_error)?;
+            let auth_timeout = self
+                .config
+                .resilience
+                .as_ref()
+                .and_then(|config| config.attempt_timeout);
+            let auth = cancellable(
+                run_bounded_http(
+                    self.config.authentication.authenticate(None),
+                    auth_timeout,
+                    deadline.as_ref(),
+                    "OpenAI authentication",
+                ),
+                cancellation.as_ref(),
+            )
+            .await?
+            .map_err(|error| match error {
+                HttpError::Timeout {
+                    operation: "logical request retry budget",
+                    ..
+                } => http_loop_error(error),
+                _ => local_error(ProviderFailureReason::Authentication),
+            })?;
+            let mut value = encode_request_bound(&self.config, &request, auth.binding())
+                .map_err(|_| local_error(ProviderFailureReason::InvalidRequest))?;
+            // Serialize once. Every status, transport, and stream retry reuses these exact bytes.
+            let body = agentkit_http::Bytes::from_owner(Zeroizing::new(
+                serde_json::to_vec(&value)
+                    .map_err(OpenAIResponsesError::Serialize)
+                    .map_err(|_| local_error(ProviderFailureReason::InvalidRequest))?,
+            ));
+            zeroize_encrypted_content(&mut value);
+            let idempotency_key = stable_idempotency_key(
+                &self.session.session_id.to_string(),
+                &request.turn_id.to_string(),
+                &body,
+            );
+            let supersession_enabled = self
+                .session
+                .consumer_capabilities
+                .response_attempt_supersession;
+            Ok((body, idempotency_key, auth, deadline, supersession_enabled))
         }
-        // One logical deadline starts before encoding and initial authentication.
-        let deadline = self
-            .config
-            .resilience
-            .as_ref()
-            .map(|config| LogicalDeadline::new(config.retry_budget));
-        deadline_remaining(deadline.as_ref()).map_err(http_loop_error)?;
-        let auth_timeout = self
-            .config
-            .resilience
-            .as_ref()
-            .and_then(|config| config.attempt_timeout);
-        let auth = cancellable(
-            run_bounded_http(
-                self.config.authentication.authenticate(None),
-                auth_timeout,
-                deadline.as_ref(),
-                "OpenAI authentication",
-            ),
-            cancellation.as_ref(),
-        )
-        .await?
-        .map_err(http_loop_error)?;
-        let mut value = encode_request_bound(&self.config, &request, auth.binding())
-            .map_err(|error| LoopError::Provider(error.to_string()))?;
-        // Serialize once. Every status, transport, and stream retry reuses these exact bytes.
-        let body = agentkit_http::Bytes::from_owner(Zeroizing::new(
-            serde_json::to_vec(&value)
-                .map_err(OpenAIResponsesError::Serialize)
-                .map_err(|error| LoopError::Provider(error.to_string()))?,
-        ));
-        zeroize_encrypted_content(&mut value);
-        let idempotency_key = stable_idempotency_key(
-            &self.session.session_id.to_string(),
-            &request.turn_id.to_string(),
-            &body,
-        );
-        let supersession_enabled = self
-            .session
-            .consumer_capabilities
-            .response_attempt_supersession;
+        .await;
+        let (body, idempotency_key, auth, deadline, supersession_enabled) =
+            prepared.map_err(|error| tracker.finish(error))?;
         OpenAIResponsesTurn::open(
             ResponsesRequestContext {
                 client: self.client.clone(),
@@ -427,6 +454,7 @@ impl ModelSession for OpenAIResponsesSession {
                 retries: 0,
                 refreshed: false,
                 wire_bytes: 0,
+                tracker,
             },
             supersession_enabled,
             cancellation.as_ref(),
@@ -456,11 +484,34 @@ pub struct OpenAIResponsesTurn {
 
 #[async_trait]
 impl ModelTurn for OpenAIResponsesTurn {
+    fn on_cancelled(&mut self) {
+        if !self.finished {
+            self.finished = true;
+            self.attempt = None;
+            let _ = self.context.tracker.finish(LoopError::Cancelled);
+        }
+    }
+
     async fn next_event(
         &mut self,
         cancellation: Option<TurnCancellation>,
     ) -> Result<Option<ModelTurnEvent>, LoopError> {
-        self.next_event_inner(cancellation.as_ref()).await
+        if self.finished {
+            return Ok(None);
+        }
+        match self.next_event_inner(cancellation.as_ref()).await {
+            Ok(event) => {
+                if matches!(event, Some(ModelTurnEvent::Finished(_))) {
+                    self.context.tracker.succeed();
+                }
+                Ok(event)
+            }
+            Err(error) => {
+                self.finished = true;
+                self.attempt = None;
+                Err(self.context.tracker.finish(error))
+            }
+        }
     }
 }
 
@@ -479,7 +530,9 @@ impl OpenAIResponsesTurn {
             pending_delay: Duration::ZERO,
             finished: false,
         };
-        turn.reopen(cancellation).await?;
+        if let Err(error) = turn.reopen(cancellation).await {
+            return Err(turn.context.tracker.finish(error));
+        }
         Ok(turn)
     }
 
@@ -501,6 +554,7 @@ impl OpenAIResponsesTurn {
             )
             .await?
             .map_err(http_loop_error)?;
+            self.context.tracker.completed_wait(delay);
         }
         self.attempt = Some(open_live_attempt(&mut self.context, cancellation).await?);
         self.pending_reopen = false;
@@ -515,6 +569,8 @@ impl OpenAIResponsesTurn {
             if cancelled(cancellation) {
                 return Err(LoopError::Cancelled);
             }
+            // Buffered events and EOF must not bypass the logical request budget.
+            deadline_remaining(self.context.deadline.as_ref()).map_err(http_loop_error)?;
             if self.pending_reopen {
                 self.reopen(cancellation).await?;
             }
@@ -641,6 +697,7 @@ impl OpenAIResponsesTurn {
                 }
             };
             if let Err(failure) = result {
+                self.context.tracker.note_failure(&failure.error);
                 if !failure.retryable
                     || self.context.retries
                         >= self
@@ -650,10 +707,10 @@ impl OpenAIResponsesTurn {
                             .as_ref()
                             .map_or(0, |config| config.max_retries)
                 {
-                    return Err(*failure.error);
+                    return Err(stopped_attempt(&self.context, failure));
                 }
                 if self.attempt_output_emitted && !self.supersession_enabled {
-                    return Err(*failure.error);
+                    return Err(local_error(ProviderFailureReason::ReplayUnsafe));
                 }
                 let delay = self
                     .context
@@ -662,6 +719,12 @@ impl OpenAIResponsesTurn {
                     .as_ref()
                     .expect("retry requires resilience")
                     .retry_delay(self.context.retries, failure.headers.as_ref());
+                self.context.tracker.scheduled(&failure.error, delay);
+                // Observers may request cancellation without re-entering this turn.
+                // Finalize now: the driver can drop the turn after this event.
+                if cancelled(cancellation) {
+                    return Err(LoopError::Cancelled);
+                }
                 self.context.retries += 1;
                 self.attempt = None;
                 self.pending_reopen = true;
@@ -1517,20 +1580,38 @@ where
     futures_util::pin_mut!(future);
     let timer = sleep(timeout);
     futures_util::pin_mut!(timer);
-    match select(future, timer).await {
-        Either::Left((result, _)) => result,
-        Either::Right((_, _)) if budget_limited => Err(HttpError::Timeout {
+    match select(timer, future).await {
+        Either::Right((result, _)) => {
+            deadline_remaining(deadline)?;
+            result
+        }
+        Either::Left((_, _)) if budget_limited => Err(HttpError::Timeout {
             operation: "logical request retry budget",
             timeout: deadline
                 .expect("budget-limited operation has deadline")
                 .budget,
         }),
-        Either::Right((_, _)) => Err(HttpError::Timeout { operation, timeout }),
+        Either::Left((_, _)) => Err(HttpError::Timeout { operation, timeout }),
     }
 }
 
 fn http_loop_error(error: HttpError) -> LoopError {
-    LoopError::Provider(format!("OpenAI Responses {error}"))
+    let reason = match error {
+        HttpError::Timeout {
+            operation: "logical request retry budget",
+            ..
+        } => ProviderFailureReason::RetryBudget,
+        HttpError::Timeout {
+            operation: "response stream idle",
+            ..
+        } => ProviderFailureReason::IdleTimeout,
+        HttpError::Timeout { .. } => ProviderFailureReason::AttemptTimeout,
+        HttpError::InvalidUrl(_) | HttpError::InvalidHeader(_) | HttpError::Serialize(_) => {
+            ProviderFailureReason::InvalidRequest
+        }
+        _ => ProviderFailureReason::Transport,
+    };
+    local_error(reason)
 }
 
 #[derive(Debug)]
@@ -1552,6 +1633,7 @@ struct ResponsesRequestContext {
     retries: usize,
     refreshed: bool,
     wire_bytes: usize,
+    tracker: RetryTracker,
 }
 
 struct LiveAttempt {
@@ -1572,6 +1654,22 @@ impl LiveAttempt {
     }
 }
 
+fn stopped_attempt(context: &ResponsesRequestContext, failure: AttemptFailure) -> LoopError {
+    if failure
+        .error
+        .provider_failure()
+        .is_some_and(|failure| matches!(failure.upstream.http_status, Some(401 | 403)))
+    {
+        local_error(ProviderFailureReason::Authentication)
+    } else if !failure.retryable {
+        *failure.error
+    } else if context.config.resilience.is_none() {
+        local_error(ProviderFailureReason::RetryDisabled)
+    } else {
+        local_error(ProviderFailureReason::RetryExhausted)
+    }
+}
+
 async fn open_live_attempt(
     context: &mut ResponsesRequestContext,
     cancellation: Option<&TurnCancellation>,
@@ -1583,19 +1681,24 @@ async fn open_live_attempt(
             .as_ref()
             .and_then(|config| config.attempt_timeout);
         let attempt_deadline = attempt_timeout.map(LogicalDeadline::new);
+        let logical_deadline = context.deadline.clone();
         let result = attempt_with_timeout(
             send_live_attempt(context, cancellation),
             attempt_timeout,
-            context.deadline.as_ref(),
+            logical_deadline.as_ref(),
             cancellation,
         )
         .await;
+        if let Err(failure) = &result {
+            context.tracker.note_failure(&failure.error);
+        }
         match result {
             Ok(mut attempt) => {
                 attempt.deadline = attempt_deadline;
                 return Ok(attempt);
             }
             Err(failure) if is_unauthorized(&failure.error) && !context.refreshed => {
+                context.tracker.scheduled(&failure.error, Duration::ZERO);
                 let binding = context.auth.binding().map(str::to_owned);
                 let refreshed = cancellable(
                     run_bounded_http(
@@ -1614,11 +1717,15 @@ async fn open_live_attempt(
                     cancellation,
                 )
                 .await?
-                .map_err(http_loop_error)?;
+                .map_err(|error| match error {
+                    HttpError::Timeout {
+                        operation: "logical request retry budget",
+                        ..
+                    } => http_loop_error(error),
+                    _ => local_error(ProviderFailureReason::Authentication),
+                })?;
                 if refreshed.binding() != binding.as_deref() {
-                    return Err(LoopError::Provider(
-                        "OpenAI authentication binding changed during reactive refresh".into(),
-                    ));
+                    return Err(local_error(ProviderFailureReason::Authentication));
                 }
                 context.auth = refreshed;
                 context.refreshed = true;
@@ -1638,6 +1745,7 @@ async fn open_live_attempt(
                     .as_ref()
                     .expect("retry requires resilience")
                     .retry_delay(context.retries, failure.headers.as_ref());
+                context.tracker.scheduled(&failure.error, delay);
                 context.retries += 1;
                 cancellable(
                     run_bounded_http(
@@ -1653,18 +1761,25 @@ async fn open_live_attempt(
                 )
                 .await?
                 .map_err(http_loop_error)?;
+                context.tracker.completed_wait(delay);
             }
-            Err(failure) => return Err(*failure.error),
+            Err(failure) => return Err(stopped_attempt(context, failure)),
         }
     }
 }
 
 async fn send_live_attempt(
-    context: &ResponsesRequestContext,
+    context: &mut ResponsesRequestContext,
     cancellation: Option<&TurnCancellation>,
 ) -> Result<LiveAttempt, AttemptFailure> {
     deadline_remaining(context.deadline.as_ref())
         .map_err(|error| nonretryable(http_loop_error(error)))?;
+    // The transport-neutral request builder only stores the URL string. Reject
+    // local endpoint errors before they become counted, retryable transport errors.
+    reqwest::Url::parse(&context.config.endpoint)
+        .ok()
+        .filter(|url| matches!(url.scheme(), "http" | "https") && url.has_host())
+        .ok_or_else(|| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?;
     let mut headers = context.config.headers.clone();
     headers.insert("accept", HeaderValue::from_static("text/event-stream"));
     headers.insert("content-type", HeaderValue::from_static("application/json"));
@@ -1672,7 +1787,7 @@ async fn send_live_attempt(
         headers.insert(
             "user-agent",
             HeaderValue::from_str(user_agent)
-                .map_err(|_| protocol_failure("invalid user-agent header"))?,
+                .map_err(|_| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?,
         );
     } else {
         headers
@@ -1691,7 +1806,7 @@ async fn send_live_attempt(
         headers.insert(
             "originator",
             HeaderValue::from_str(originator)
-                .map_err(|_| protocol_failure("invalid originator header"))?,
+                .map_err(|_| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?,
         );
     }
     let sent_turn_state = if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate {
@@ -1701,7 +1816,7 @@ async fn send_live_attempt(
             .or_insert(HeaderValue::from_static("agentkit"));
         headers.entry("session-id").or_insert(
             HeaderValue::from_str(&context.session_id)
-                .map_err(|_| protocol_failure("invalid session ID header"))?,
+                .map_err(|_| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?,
         );
         let state = context
             .turn_state
@@ -1716,13 +1831,20 @@ async fn send_live_attempt(
         None
     };
     headers.extend(context.auth.headers().clone());
+    let request = context
+        .client
+        .post(&context.config.endpoint)
+        .headers(headers)
+        .body(context.body.clone())
+        .build()
+        .map_err(|error| nonretryable(http_loop_error(error)))?;
     let response = cancellable(
-        context
-            .client
-            .post(&context.config.endpoint)
-            .headers(headers)
-            .body(context.body.clone())
-            .send(),
+        async {
+            // Count only a polled send, after preflight and cancellation checks.
+            context.tracker.accounting.attempts =
+                context.tracker.accounting.attempts.saturating_add(1);
+            context.client.execute(request).await
+        },
         cancellation,
     )
     .await
@@ -1732,8 +1854,12 @@ async fn send_live_attempt(
     let status = response.status();
     if status == StatusCode::UNAUTHORIZED {
         return Err(AttemptFailure {
-            error: Box::new(LoopError::Provider(
-                "OpenAI Responses returned 401 Unauthorized".into(),
+            error: Box::new(provider_error(
+                ProviderFailureReason::HttpStatus,
+                ProviderClassification {
+                    http_status: Some(401),
+                    ..ProviderClassification::default()
+                },
             )),
             retryable: false,
             headers: None,
@@ -1741,9 +1867,13 @@ async fn send_live_attempt(
     }
     if !status.is_success() {
         return Err(AttemptFailure {
-            error: Box::new(LoopError::Provider(format!(
-                "OpenAI Responses returned HTTP {status}"
-            ))),
+            error: Box::new(provider_error(
+                ProviderFailureReason::HttpStatus,
+                ProviderClassification {
+                    http_status: Some(status.as_u16()),
+                    ..ProviderClassification::default()
+                },
+            )),
             retryable: retryable_response_status(status, context.config.profile),
             headers: retry_headers(response.headers()),
         });
@@ -1797,11 +1927,9 @@ fn retryable_response_status(status: StatusCode, profile: OpenAIResponsesProfile
             && matches!(status.as_u16(), 404 | 529))
 }
 
-fn attempt_timeout_failure(timeout: Duration) -> AttemptFailure {
+fn attempt_timeout_failure(_timeout: Duration) -> AttemptFailure {
     AttemptFailure {
-        error: Box::new(LoopError::Provider(format!(
-            "Responses attempt timed out after {timeout:?}"
-        ))),
+        error: Box::new(local_error(ProviderFailureReason::AttemptTimeout)),
         retryable: true,
         headers: None,
     }
@@ -1831,15 +1959,19 @@ where
         futures_util::pin_mut!(future);
         let timer = sleep(timeout);
         futures_util::pin_mut!(timer);
-        match select(future, timer).await {
-            Either::Left((result, _)) => result,
-            Either::Right((_, _)) if budget_limited => {
+        match select(timer, future).await {
+            Either::Right((result, _)) => {
+                deadline_remaining(deadline)
+                    .map_err(|error| nonretryable(http_loop_error(error)))?;
+                result
+            }
+            Either::Left((_, _)) if budget_limited => {
                 Err(nonretryable(http_loop_error(HttpError::Timeout {
                     operation: "logical request retry budget",
                     timeout: deadline.expect("budget timeout has deadline").budget,
                 })))
             }
-            Either::Right((_, _)) => Err(attempt_timeout_failure(timeout)),
+            Either::Left((_, _)) => Err(attempt_timeout_failure(timeout)),
         }
     };
     match cancellable(timed, cancellation).await {
@@ -2189,17 +2321,12 @@ impl ResponsesState {
                         }
                     }
                 }
-                let code = value
-                    .pointer("/response/error/code")
-                    .or_else(|| value.pointer("/error/code"))
-                    .or_else(|| value.get("code"))
-                    .and_then(Value::as_str)
-                    .unwrap_or("unknown");
                 let retryable = stream_failure_retryable(self.profile, value, kind);
                 Err(AttemptFailure {
-                    error: Box::new(LoopError::Provider(format!(
-                        "OpenAI Responses stream failed ({code})"
-                    ))),
+                    error: Box::new(provider_error(
+                        ProviderFailureReason::ResponseFailed,
+                        stream_classification(value, kind),
+                    )),
                     retryable,
                     headers: None,
                 })
@@ -3264,9 +3391,7 @@ fn stable_idempotency_key(session: &str, turn: &str, body: &[u8]) -> String {
 fn transport_failure(error: HttpError) -> AttemptFailure {
     let retryable = error.is_retryable_transport();
     AttemptFailure {
-        error: Box::new(LoopError::Provider(format!(
-            "OpenAI Responses transport failed: {error}"
-        ))),
+        error: Box::new(http_loop_error(error)),
         retryable,
         headers: None,
     }
@@ -3348,7 +3473,10 @@ fn nonretryable(error: LoopError) -> AttemptFailure {
 }
 
 fn is_unauthorized(error: &LoopError) -> bool {
-    matches!(error, LoopError::Provider(message) if message.contains("401 Unauthorized"))
+    error.provider_failure().is_some_and(|failure| {
+        failure.reason == ProviderFailureReason::HttpStatus
+            && failure.upstream.http_status == Some(401)
+    })
 }
 
 fn cancelled(cancellation: Option<&TurnCancellation>) -> bool {
@@ -3371,14 +3499,28 @@ where
     futures_util::pin_mut!(future);
     let cancelled = cancellation.cancelled();
     futures_util::pin_mut!(cancelled);
-    match select(future, cancelled).await {
-        Either::Left((result, _)) => Ok(result),
-        Either::Right((_, _)) => Err(LoopError::Cancelled),
-    }
+    let raced = select(cancelled, future);
+    futures_util::pin_mut!(raced);
+    futures_util::future::poll_fn(|cx| {
+        // TurnCancellation's notification future polls on a timer. Check the
+        // generation on every wake too, so a ready request cannot outrun it.
+        if cancellation.is_cancelled() {
+            return std::task::Poll::Ready(Err(LoopError::Cancelled));
+        }
+        match raced.as_mut().poll(cx) {
+            std::task::Poll::Ready(Either::Right((result, _))) if !cancellation.is_cancelled() => {
+                std::task::Poll::Ready(Ok(result))
+            }
+            std::task::Poll::Ready(_) => std::task::Poll::Ready(Err(LoopError::Cancelled)),
+            std::task::Poll::Pending => std::task::Poll::Pending,
+        }
+    })
+    .await
 }
 
 #[cfg(test)]
 mod tests {
+    mod retry_observability;
     use std::sync::Mutex;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
@@ -3733,7 +3875,10 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1
             Ok(_) => panic!("changed authentication binding unexpectedly succeeded"),
             Err(error) => error,
         };
-        assert!(error.to_string().contains("authentication binding changed"));
+        assert_eq!(
+            error.provider_failure().unwrap().reason,
+            ProviderFailureReason::Authentication
+        );
         assert_eq!(calls.load(Ordering::SeqCst), 2);
         assert_eq!(client.requests.lock().unwrap().len(), 1);
     }
@@ -4198,7 +4343,10 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1
             .await
             .unwrap();
         let error = session.begin_turn(request(), None).await.err().unwrap();
-        assert!(error.to_string().contains("logical request retry budget"));
+        assert_eq!(
+            error.provider_failure().unwrap().reason,
+            ProviderFailureReason::RetryBudget
+        );
         assert!(client.requests.lock().unwrap().is_empty());
     }
 
@@ -4249,7 +4397,10 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1
             .await
             .unwrap();
         let error = session.begin_turn(request(), None).await.err().unwrap();
-        assert!(error.to_string().contains("logical request retry budget"));
+        assert_eq!(
+            error.provider_failure().unwrap().reason,
+            ProviderFailureReason::RetryBudget
+        );
     }
 
     #[test]
@@ -4418,7 +4569,10 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1
             }
         };
         assert!(saw_visible);
-        assert!(matches!(error, LoopError::Provider(_)));
+        assert_eq!(
+            error.provider_failure().unwrap().reason,
+            ProviderFailureReason::ReplayUnsafe
+        );
         assert_eq!(client.requests.lock().unwrap().len(), 1);
     }
 
@@ -4867,7 +5021,10 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1
             .unwrap();
         let mut turn = session.begin_turn(request(), None).await.unwrap();
         let error = turn.next_event(None).await.unwrap_err();
-        assert!(error.to_string().contains("attempt timed out"));
+        assert_eq!(
+            error.provider_failure().unwrap().last_attempt_reason,
+            Some(ProviderFailureReason::AttemptTimeout)
+        );
         assert_eq!(client.requests.load(Ordering::SeqCst), 1);
     }
 
@@ -4911,7 +5068,10 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1
             .unwrap();
         let mut turn = session.begin_turn(request(), None).await.unwrap();
         let error = turn.next_event(None).await.unwrap_err();
-        assert!(error.to_string().contains("wire-byte limit"));
+        assert_eq!(
+            error.provider_failure().unwrap().reason,
+            ProviderFailureReason::Protocol
+        );
         assert_eq!(client.requests.lock().unwrap().len(), 2);
     }
 }
diff --git a/crates/agentkit-provider-openai/src/responses/retry.rs b/crates/agentkit-provider-openai/src/responses/retry.rs
new file mode 100644
index 0000000..61f7da1
--- /dev/null
+++ b/crates/agentkit-provider-openai/src/responses/retry.rs
@@ -0,0 +1,289 @@
+use super::*;
+
+/// One accumulator per logical request; no event queue or background task.
+pub(super) struct RetryTracker {
+    pub(super) started_at: Instant,
+    pub(super) accounting: RetryAccounting,
+    route: ProviderRoute,
+    upstream: ProviderClassification,
+    last_attempt_reason: Option<ProviderFailureReason>,
+    observer: Option<Arc<dyn RetryObserver>>,
+    last_progress: Option<Duration>,
+    finalized: bool,
+}
+
+impl RetryTracker {
+    pub(super) fn new(
+        profile: OpenAIResponsesProfile,
+        observer: Option<Arc<dyn RetryObserver>>,
+    ) -> Self {
+        Self {
+            started_at: Instant::now(),
+            accounting: RetryAccounting::default(),
+            route: match profile {
+                OpenAIResponsesProfile::Public => ProviderRoute::OpenAiResponses,
+                OpenAIResponsesProfile::ChatGptPrivate => ProviderRoute::OpenAiChatGptResponses,
+            },
+            upstream: ProviderClassification::default(),
+            last_attempt_reason: None,
+            observer,
+            last_progress: None,
+            finalized: false,
+        }
+    }
+
+    pub(super) fn snapshot(&self) -> RetryAccounting {
+        self.snapshot_at(self.started_at.elapsed())
+    }
+
+    fn snapshot_at(&self, elapsed: Duration) -> RetryAccounting {
+        RetryAccounting {
+            elapsed,
+            ..self.accounting
+        }
+    }
+
+    pub(super) fn note_failure(&mut self, error: &LoopError) {
+        let reason = failure_reason(error);
+        if self.accounting.attempts > 0
+            && matches!(
+                reason,
+                ProviderFailureReason::HttpStatus
+                    | ProviderFailureReason::Transport
+                    | ProviderFailureReason::ResponseFailed
+                    | ProviderFailureReason::Protocol
+                    | ProviderFailureReason::AttemptTimeout
+                    | ProviderFailureReason::IdleTimeout
+            )
+        {
+            self.last_attempt_reason = Some(reason);
+        }
+        if let Some(failure) = error.provider_failure() {
+            // Local timeout/budget/auth stops retain the last provider classification.
+            // A new source response (even Unknown) replaces it.
+            if matches!(
+                failure.reason,
+                ProviderFailureReason::HttpStatus | ProviderFailureReason::ResponseFailed
+            ) {
+                self.upstream = failure.upstream;
+            }
+        }
+    }
+
+    pub(super) fn scheduled(&mut self, error: &LoopError, delay: Duration) {
+        self.note_failure(error);
+        self.scheduled_at(error, delay, self.started_at.elapsed());
+    }
+
+    fn scheduled_at(&mut self, error: &LoopError, delay: Duration, elapsed: Duration) {
+        if self
+            .last_progress
+            .is_some_and(|last| elapsed.saturating_sub(last) < Duration::from_millis(250))
+        {
+            return;
+        }
+        self.last_progress = Some(elapsed);
+        self.emit(ProviderRetryEvent::Scheduled(RetryProgress {
+            route: self.route,
+            reason: failure_reason(error),
+            upstream: self.upstream,
+            accounting: self.snapshot_at(elapsed),
+            next_delay: delay,
+        }));
+    }
+
+    pub(super) fn completed_wait(&mut self, delay: Duration) {
+        self.accounting.completed_backoff = self.accounting.completed_backoff.saturating_add(delay);
+    }
+
+    pub(super) fn finish(&mut self, error: LoopError) -> LoopError {
+        self.note_failure(&error);
+        let failure = ProviderFailure {
+            route: self.route,
+            reason: failure_reason(&error),
+            last_attempt_reason: self.last_attempt_reason,
+            upstream: self.upstream,
+            accounting: self.snapshot(),
+        };
+        if !self.finalized {
+            self.finalized = true;
+            self.emit(ProviderRetryEvent::Stopped(failure));
+        }
+        if matches!(error, LoopError::Cancelled) {
+            error
+        } else {
+            LoopError::ProviderFailure(Box::new(failure))
+        }
+    }
+
+    pub(super) fn succeed(&mut self) {
+        if !self.finalized {
+            self.finalized = true;
+            self.emit(ProviderRetryEvent::Succeeded {
+                route: self.route,
+                accounting: self.snapshot(),
+            });
+        }
+    }
+
+    fn emit(&self, event: ProviderRetryEvent) {
+        if let Some(observer) = &self.observer {
+            observer.on_retry_event(event);
+        }
+    }
+}
+
+pub(super) fn failure_reason(error: &LoopError) -> ProviderFailureReason {
+    match error {
+        LoopError::Cancelled => ProviderFailureReason::Cancelled,
+        LoopError::ProviderFailure(failure) => failure.reason,
+        _ => ProviderFailureReason::Protocol,
+    }
+}
+
+pub(super) fn provider_error(
+    reason: ProviderFailureReason,
+    upstream: ProviderClassification,
+) -> LoopError {
+    LoopError::ProviderFailure(Box::new(ProviderFailure {
+        route: ProviderRoute::Unknown,
+        reason,
+        last_attempt_reason: None,
+        upstream,
+        accounting: RetryAccounting::default(),
+    }))
+}
+
+pub(super) fn local_error(reason: ProviderFailureReason) -> LoopError {
+    provider_error(reason, ProviderClassification::default())
+}
+
+pub(super) fn upstream_kind(value: Option<&Value>) -> UpstreamErrorKind {
+    match value.and_then(Value::as_str) {
+        Some("service_unavailable_error") => UpstreamErrorKind::ServiceUnavailableError,
+        Some("server_is_overloaded") => UpstreamErrorKind::ServerIsOverloaded,
+        Some("server_error") => UpstreamErrorKind::ServerError,
+        Some("rate_limit_error") => UpstreamErrorKind::RateLimitError,
+        Some("rate_limit_exceeded") => UpstreamErrorKind::RateLimitExceeded,
+        Some("temporarily_unavailable") => UpstreamErrorKind::TemporarilyUnavailable,
+        Some("authentication_error") => UpstreamErrorKind::AuthenticationError,
+        Some("invalid_api_key") => UpstreamErrorKind::InvalidApiKey,
+        Some("invalid_authentication") => UpstreamErrorKind::InvalidAuthentication,
+        Some("unauthorized") => UpstreamErrorKind::Unauthorized,
+        Some("invalid_request_error") => UpstreamErrorKind::InvalidRequestError,
+        Some("permission_denied") => UpstreamErrorKind::PermissionDenied,
+        Some("insufficient_quota") => UpstreamErrorKind::InsufficientQuota,
+        Some("content_policy_violation") => UpstreamErrorKind::ContentPolicyViolation,
+        _ => UpstreamErrorKind::Unknown,
+    }
+}
+
+pub(super) fn stream_classification(value: &Value, kind: &str) -> ProviderClassification {
+    let error = if kind == "response.failed" {
+        value.pointer("/response/error").unwrap_or(&Value::Null)
+    } else {
+        value.get("error").unwrap_or(value)
+    };
+    ProviderClassification {
+        error_type: upstream_kind(error.get("type")),
+        code: upstream_kind(error.get("code").or_else(|| value.get("code"))),
+        http_status: error
+            .get("status")
+            .or_else(|| value.get("status"))
+            .or_else(|| value.pointer("/response/status_code"))
+            .and_then(Value::as_u64)
+            .filter(|status| (100..=599).contains(status))
+            .map(|status| status as u16),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn throttling_keeps_exact_accounting_and_never_suppresses_terminal() {
+        let events = Arc::new(Mutex::new(Vec::new()));
+        let capture = events.clone();
+        let mut tracker = RetryTracker::new(
+            OpenAIResponsesProfile::Public,
+            Some(Arc::new(move |event| capture.lock().unwrap().push(event))),
+        );
+        let error = local_error(ProviderFailureReason::Transport);
+        for millis in [0, 249, 250, 499, 500] {
+            tracker.accounting.attempts += 1;
+            tracker.completed_wait(Duration::from_millis(15));
+            tracker.scheduled_at(
+                &error,
+                Duration::from_secs(2),
+                Duration::from_millis(millis),
+            );
+        }
+        assert_eq!(
+            tracker.snapshot_at(Duration::from_secs(1)),
+            RetryAccounting {
+                attempts: 5,
+                completed_backoff: Duration::from_millis(75),
+                elapsed: Duration::from_secs(1),
+            }
+        );
+        let error = tracker.finish(local_error(ProviderFailureReason::RetryBudget));
+        assert_eq!(error.provider_failure().unwrap().accounting.attempts, 5);
+        assert_eq!(
+            error
+                .provider_failure()
+                .unwrap()
+                .accounting
+                .completed_backoff,
+            Duration::from_millis(75)
+        );
+        tracker.succeed(); // finalization is idempotent, even across dispositions
+        let events = events.lock().unwrap();
+        assert_eq!(events.len(), 4);
+        for (index, millis) in [0, 250, 500].into_iter().enumerate() {
+            let ProviderRetryEvent::Scheduled(progress) = events[index] else {
+                panic!("missing schedule")
+            };
+            assert_eq!(progress.accounting.elapsed, Duration::from_millis(millis));
+        }
+        assert!(matches!(events[3], ProviderRetryEvent::Stopped(_)));
+    }
+
+    #[test]
+    fn unknown_classification_never_retains_payload_and_local_stops_retain_last_source() {
+        let secret = "secret-customer-credential-prompt".repeat(4096);
+        for value in [
+            Value::Null,
+            json!({"error": {"type": secret, "code": secret, "message": secret}}),
+            json!({"error": {"type": 42, "code": ["secret"], "status": 900}}),
+        ] {
+            let classification = stream_classification(&value, "error");
+            assert_eq!(classification, ProviderClassification::default());
+            assert!(serde_json::to_string(&classification).unwrap().len() < 100);
+        }
+        let source = stream_classification(
+            &json!({"response": {"error": {"type": "service_unavailable_error", "code": "server_is_overloaded"}}}),
+            "response.failed",
+        );
+        for reason in [
+            ProviderFailureReason::RetryBudget,
+            ProviderFailureReason::Authentication,
+            ProviderFailureReason::RetryExhausted,
+        ] {
+            let mut tracker = RetryTracker::new(OpenAIResponsesProfile::ChatGptPrivate, None);
+            tracker.accounting.attempts = 1;
+            tracker.note_failure(&provider_error(
+                ProviderFailureReason::ResponseFailed,
+                source,
+            ));
+            let error = tracker.finish(local_error(reason));
+            let failure = error.provider_failure().unwrap();
+            assert_eq!(failure.reason, reason);
+            assert_eq!(failure.upstream, source);
+            assert_eq!(
+                failure.last_attempt_reason,
+                Some(ProviderFailureReason::ResponseFailed)
+            );
+        }
+    }
+}
diff --git a/crates/agentkit-provider-openai/src/responses/tests/retry_observability.rs b/crates/agentkit-provider-openai/src/responses/tests/retry_observability.rs
new file mode 100644
index 0000000..9fb3756
--- /dev/null
+++ b/crates/agentkit-provider-openai/src/responses/tests/retry_observability.rs
@@ -0,0 +1,876 @@
+use super::*;
+use agentkit_core::CancellationController;
+
+const OVERLOADED: &str = r#"event: response.created
+data: {"type":"response.created","sequence_number":1,"response":{"id":"private-id","model":"private-model"}}
+
+event: response.failed
+data: {"type":"response.failed","sequence_number":2,"response":{"error":{"type":"service_unavailable_error","code":"server_is_overloaded","message":"SECRET-PROMPT-CREDENTIAL"}}}
+
+"#;
+
+fn policy(retries: usize) -> ResilienceConfig {
+    ResilienceConfig {
+        max_retries: retries,
+        retry_budget: Duration::from_secs(120),
+        attempt_timeout: None,
+        stream_idle_timeout: None,
+        initial_backoff: Duration::ZERO,
+        max_backoff: Duration::ZERO,
+    }
+}
+
+fn wire(status: StatusCode, body: &'static str) -> WireResponse {
+    WireResponse {
+        status,
+        headers: sse_headers(),
+        body,
+    }
+}
+
+async fn observed(
+    config: OpenAIResponsesConfig,
+    responses: Vec<WireResponse>,
+) -> (
+    OpenAIResponsesSession,
+    Arc<Mutex<Vec<ProviderRetryEvent>>>,
+    Arc<ScriptedClient>,
+) {
+    let client = Arc::new(ScriptedClient {
+        responses: Mutex::new(responses.into()),
+        requests: Mutex::new(Vec::new()),
+    });
+    let mut session = OpenAIResponsesAdapter::with_client(config, Http::from_arc(client.clone()))
+        .start_session(SessionConfig::new("session"))
+        .await
+        .unwrap();
+    let events = Arc::new(Mutex::new(Vec::new()));
+    let capture = events.clone();
+    session.set_retry_observer(Some(Arc::new(move |event| {
+        capture.lock().unwrap().push(event)
+    })));
+    (session, events, client)
+}
+
+async fn finish(turn: &mut OpenAIResponsesTurn) -> Result<usize, LoopError> {
+    let mut results = 0;
+    while let Some(event) = turn.next_event(None).await? {
+        if matches!(event, ModelTurnEvent::Finished(_)) {
+            results += 1;
+        }
+    }
+    Ok(results)
+}
+
+#[tokio::test]
+async fn initial_retries_are_live_before_begin_turn_returns_and_cancelled_wait_is_not_completed() {
+    let mut overloaded = wire(StatusCode::SERVICE_UNAVAILABLE, "SECRET-BODY");
+    overloaded
+        .headers
+        .insert("retry-after", HeaderValue::from_static("60"));
+    let (mut session, events, client) = observed(
+        OpenAIResponsesConfig::new("SECRET-CREDENTIAL", "private-model").with_resilience(policy(1)),
+        vec![overloaded],
+    )
+    .await;
+    let controller = CancellationController::new();
+    let cancellation = TurnCancellation::new(controller.handle());
+    let future = session.begin_turn(request(), Some(cancellation));
+    futures_util::pin_mut!(future);
+    assert!(futures_util::poll!(future.as_mut()).is_pending());
+    {
+        let events = events.lock().unwrap();
+        let ProviderRetryEvent::Scheduled(progress) = events[0] else {
+            panic!("missing live retry")
+        };
+        assert_eq!(progress.accounting.attempts, 1);
+        assert_eq!(progress.next_delay, Duration::from_secs(60));
+        assert_eq!(progress.accounting.completed_backoff, Duration::ZERO);
+        assert_eq!(progress.upstream.http_status, Some(503));
+    }
+    controller.interrupt();
+    assert!(matches!(future.await, Err(LoopError::Cancelled)));
+    let events = events.lock().unwrap();
+    let ProviderRetryEvent::Stopped(failure) = events[1] else {
+        panic!("missing cancellation summary")
+    };
+    assert_eq!(failure.reason, ProviderFailureReason::Cancelled);
+    assert_eq!(failure.accounting.attempts, 1);
+    assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+    assert_eq!(failure.upstream.http_status, Some(503));
+    assert_eq!(events.len(), 2);
+    assert_eq!(client.requests.lock().unwrap().len(), 1);
+    assert!(!serde_json::to_string(&*events).unwrap().contains("SECRET"));
+}
+
+#[tokio::test]
+async fn stream_exhaustion_preserves_two_classification_layers_and_is_finalized_once() {
+    let (mut session, events, client) = observed(
+        OpenAIResponsesConfig::chatgpt_private("private-model", "SECRET")
+            .with_resilience(policy(3)),
+        vec![wire(StatusCode::OK, OVERLOADED); 4],
+    )
+    .await;
+    let mut turn = session.begin_turn(request(), None).await.unwrap();
+    let error = finish(&mut turn).await.unwrap_err();
+    let failure = error.provider_failure().unwrap();
+    assert_eq!(failure.reason, ProviderFailureReason::RetryExhausted);
+    assert_eq!(
+        failure.last_attempt_reason,
+        Some(ProviderFailureReason::ResponseFailed)
+    );
+    assert_eq!(
+        failure.upstream.error_type,
+        UpstreamErrorKind::ServiceUnavailableError
+    );
+    assert_eq!(failure.upstream.code, UpstreamErrorKind::ServerIsOverloaded);
+    assert_eq!(failure.accounting.attempts, 4);
+    assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+    assert!(turn.next_event(None).await.unwrap().is_none());
+    let events = events.lock().unwrap();
+    assert_eq!(
+        events
+            .iter()
+            .filter(|event| matches!(event, ProviderRetryEvent::Stopped(_)))
+            .count(),
+        1
+    );
+    assert_eq!(events.last(), Some(&ProviderRetryEvent::Stopped(*failure)));
+    assert_eq!(client.requests.lock().unwrap().len(), 4);
+    let rendered = format!(
+        "{error:?} {error} {}",
+        serde_json::to_string(&*events).unwrap()
+    );
+    for secret in ["SECRET", "private-id", "private-model", "message"] {
+        assert!(!rendered.contains(secret));
+    }
+}
+
+#[tokio::test]
+async fn successful_retries_and_refresh_count_sends_and_clear_progress_once() {
+    for status in [StatusCode::SERVICE_UNAVAILABLE, StatusCode::UNAUTHORIZED] {
+        let (mut session, events, client) = observed(
+            OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(1)),
+            vec![wire(status, ""), wire(StatusCode::OK, SUCCESS)],
+        )
+        .await;
+        let mut turn = session.begin_turn(request(), None).await.unwrap();
+        assert_eq!(finish(&mut turn).await.unwrap(), 1);
+        assert!(turn.next_event(None).await.unwrap().is_none());
+        let events = events.lock().unwrap();
+        assert!(matches!(
+            events.first(),
+            Some(ProviderRetryEvent::Scheduled(_))
+        ));
+        let Some(ProviderRetryEvent::Succeeded { accounting, .. }) = events.last() else {
+            panic!("missing success")
+        };
+        assert_eq!(accounting.attempts, 2);
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| matches!(event, ProviderRetryEvent::Succeeded { .. }))
+                .count(),
+            1
+        );
+        assert_eq!(client.requests.lock().unwrap().len(), 2);
+    }
+}
+
+#[tokio::test]
+async fn already_cancelled_and_invalid_preflight_have_zero_sends() {
+    for cancel in [false, true] {
+        let (mut session, events, client) = observed(
+            OpenAIResponsesConfig::new("secret", "gpt-test")
+                .with_user_agent("invalid header value\n"),
+            vec![],
+        )
+        .await;
+        let controller = CancellationController::new();
+        let cancellation = TurnCancellation::new(controller.handle());
+        if cancel {
+            controller.interrupt();
+        }
+        let result = session.begin_turn(request(), Some(cancellation)).await;
+        assert!(result.is_err());
+        let events = events.lock().unwrap();
+        let ProviderRetryEvent::Stopped(failure) = events[0] else {
+            panic!("missing stop")
+        };
+        assert_eq!(
+            failure.reason,
+            if cancel {
+                ProviderFailureReason::Cancelled
+            } else {
+                ProviderFailureReason::InvalidRequest
+            }
+        );
+        assert_eq!(failure.accounting.attempts, 0);
+        assert_eq!(events.len(), 1);
+        assert!(client.requests.lock().unwrap().is_empty());
+    }
+}
+
+#[tokio::test]
+async fn ready_cancellation_wins_without_polling_request() {
+    let controller = CancellationController::new();
+    let cancellation = TurnCancellation::new(controller.handle());
+    controller.interrupt();
+    let polled = AtomicUsize::new(0);
+    let result = cancellable(
+        async {
+            polled.fetch_add(1, Ordering::SeqCst);
+        },
+        Some(&cancellation),
+    )
+    .await;
+    assert!(matches!(result, Err(LoopError::Cancelled)));
+    assert_eq!(polled.load(Ordering::SeqCst), 0);
+}
+
+#[tokio::test]
+async fn cancellation_of_deferred_stream_retry_preserves_supersession_order() {
+    let partial = SUCCESS
+        .split_once("event: response.output_text.done")
+        .unwrap()
+        .0;
+    let (mut session, _, client) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(1)),
+        vec![wire(StatusCode::OK, partial)],
+    )
+    .await;
+    session.session = SessionConfig::new("session").with_response_attempt_supersession();
+    let controller = Arc::new(CancellationController::new());
+    let cancellation = TurnCancellation::new(controller.handle());
+    let events = Arc::new(Mutex::new(Vec::new()));
+    let capture = events.clone();
+    let interrupt = controller.clone();
+    session.set_retry_observer(Some(Arc::new(move |event| {
+        capture.lock().unwrap().push(event);
+        if matches!(event, ProviderRetryEvent::Scheduled(_)) {
+            interrupt.interrupt();
+        }
+    })));
+    let mut turn = session
+        .begin_turn(request(), Some(cancellation.clone()))
+        .await
+        .unwrap();
+    loop {
+        match turn.next_event(Some(cancellation.clone())).await {
+            Err(LoopError::Cancelled) => break,
+            Ok(Some(ModelTurnEvent::ResponseAttemptSuperseded)) => {
+                panic!("cancelled retry must not escape as an event")
+            }
+            Ok(Some(_)) => {}
+            _ => panic!("expected cancellation"),
+        }
+    }
+    assert!(turn.next_event(None).await.unwrap().is_none());
+    let events = events.lock().unwrap();
+    let ProviderRetryEvent::Stopped(failure) = events[1] else {
+        panic!("missing stop")
+    };
+    assert_eq!(failure.accounting.attempts, 1);
+    assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+    assert_eq!(client.requests.lock().unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn cancellation_wins_when_request_and_cancellation_become_ready_together() {
+    use std::task::Poll;
+    let controller = CancellationController::new();
+    let cancellation = TurnCancellation::new(controller.handle());
+    let mut polls = 0;
+    let request = futures_util::future::poll_fn(|cx| {
+        polls += 1;
+        if polls == 1 {
+            controller.interrupt();
+            cx.waker().wake_by_ref();
+            Poll::Pending
+        } else {
+            Poll::Ready(())
+        }
+    });
+    assert!(matches!(
+        cancellable(request, Some(&cancellation)).await,
+        Err(LoopError::Cancelled)
+    ));
+    assert_eq!(polls, 1);
+}
+
+#[derive(Clone)]
+struct Capture(Arc<Mutex<Vec<agentkit_loop::ObservedEvent>>>);
+
+impl agentkit_loop::LoopObserver for Capture {
+    fn handle_event(&self, event: agentkit_loop::ObservedEvent) {
+        self.0.lock().unwrap().push(event);
+    }
+}
+
+#[tokio::test]
+async fn loop_fans_out_initial_retries_with_isolated_session_accounting() {
+    use agentkit_loop::{Agent, AgentEvent};
+    let client = Arc::new(ScriptedClient {
+        responses: Mutex::new(VecDeque::from([
+            wire(StatusCode::SERVICE_UNAVAILABLE, ""),
+            wire(StatusCode::BAD_REQUEST, ""),
+            wire(StatusCode::SERVICE_UNAVAILABLE, ""),
+            wire(StatusCode::BAD_REQUEST, ""),
+        ])),
+        requests: Mutex::new(Vec::new()),
+    });
+    let adapter = OpenAIResponsesAdapter::with_client(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(1)),
+        Http::from_arc(client),
+    );
+    let first = Arc::new(Mutex::new(Vec::new()));
+    let second = Arc::new(Mutex::new(Vec::new()));
+    let agent = Agent::builder()
+        .model(adapter)
+        .input(vec![Item::text(ItemKind::User, "SECRET-PROMPT")])
+        .observer(Capture(first.clone()))
+        .observer(Capture(second.clone()))
+        .build()
+        .unwrap();
+    for id in ["session-a", "session-b"] {
+        let mut driver = agent.start(SessionConfig::new(id)).await.unwrap();
+        let error = driver.next().await.unwrap_err();
+        assert_eq!(error.provider_failure().unwrap().accounting.attempts, 2);
+    }
+    let first = first.lock().unwrap();
+    assert_eq!(*first, *second.lock().unwrap());
+    for id in ["session-a", "session-b"] {
+        let retries: Vec<_> = first
+            .iter()
+            .filter(|event| event.session_id.0.as_str() == id)
+            .filter_map(|event| {
+                if let AgentEvent::ProviderRetry(event) = event.event {
+                    Some(event)
+                } else {
+                    None
+                }
+            })
+            .collect();
+        assert_eq!(retries.len(), 2);
+        let ProviderRetryEvent::Scheduled(progress) = retries[0] else {
+            panic!("missing schedule")
+        };
+        assert_eq!(progress.accounting.attempts, 1);
+        assert!(matches!(retries[1], ProviderRetryEvent::Stopped(_)));
+    }
+}
+
+struct FailingAuthentication;
+
+#[async_trait]
+impl AuthenticationProvider for FailingAuthentication {
+    async fn authenticate(
+        &self,
+        _: Option<&AuthenticationAttempt>,
+    ) -> Result<AuthenticationAttempt, HttpError> {
+        Err(HttpError::Other("SECRET-AUTH-DETAIL".into()))
+    }
+}
+
+#[tokio::test]
+async fn initial_auth_failure_has_typed_zero_send_summary_without_raw_source() {
+    let (mut session, events, client) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test")
+            .with_authentication_provider(FailingAuthentication),
+        vec![],
+    )
+    .await;
+    let error = match session.begin_turn(request(), None).await {
+        Err(error) => error,
+        Ok(_) => panic!("unexpected success"),
+    };
+    let failure = error.provider_failure().unwrap();
+    assert_eq!(failure.reason, ProviderFailureReason::Authentication);
+    assert_eq!(failure.accounting.attempts, 0);
+    assert_eq!(failure.last_attempt_reason, None);
+    assert_eq!(
+        *events.lock().unwrap(),
+        vec![ProviderRetryEvent::Stopped(*failure)]
+    );
+    assert!(client.requests.lock().unwrap().is_empty());
+    assert!(!format!("{error:?} {error}").contains("SECRET"));
+}
+
+#[tokio::test]
+async fn budget_interrupts_backoff_without_erasing_upstream_or_counting_planned_wait() {
+    let mut response = wire(StatusCode::SERVICE_UNAVAILABLE, "SECRET");
+    response
+        .headers
+        .insert("retry-after", HeaderValue::from_static("60"));
+    let mut resilience = policy(2);
+    resilience.retry_budget = Duration::from_millis(10);
+    let (mut session, events, _) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(resilience),
+        vec![response],
+    )
+    .await;
+    let error = match session.begin_turn(request(), None).await {
+        Err(error) => error,
+        Ok(_) => panic!("unexpected success"),
+    };
+    let failure = error.provider_failure().unwrap();
+    assert_eq!(failure.reason, ProviderFailureReason::RetryBudget);
+    assert_eq!(failure.upstream.http_status, Some(503));
+    assert_eq!(
+        failure.last_attempt_reason,
+        Some(ProviderFailureReason::HttpStatus)
+    );
+    assert_eq!(failure.accounting.attempts, 1);
+    assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+    assert!(failure.accounting.elapsed >= Duration::from_millis(10));
+    assert_eq!(
+        events.lock().unwrap().last(),
+        Some(&ProviderRetryEvent::Stopped(*failure))
+    );
+}
+
+#[tokio::test]
+async fn buffered_events_cannot_outrun_the_logical_budget() {
+    let (mut session, events, _) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(1)),
+        vec![wire(StatusCode::OK, SUCCESS)],
+    )
+    .await;
+    let mut turn = session.begin_turn(request(), None).await.unwrap();
+    assert!(turn.next_event(None).await.unwrap().is_some());
+    let deadline = turn.context.deadline.as_mut().unwrap();
+    deadline.started_at = Instant::now() - deadline.budget;
+    let error = turn.next_event(None).await.unwrap_err();
+    assert_eq!(
+        error.provider_failure().unwrap().reason,
+        ProviderFailureReason::RetryBudget
+    );
+    assert!(matches!(
+        events.lock().unwrap().as_slice(),
+        [ProviderRetryEvent::Stopped(_)]
+    ));
+}
+
+#[derive(Clone)]
+struct InterruptRetry {
+    events: Arc<Mutex<Vec<ProviderRetryEvent>>>,
+    controller: Arc<CancellationController>,
+}
+
+impl agentkit_loop::LoopObserver for InterruptRetry {
+    fn handle_event(&self, event: agentkit_loop::ObservedEvent) {
+        if let agentkit_loop::AgentEvent::ProviderRetry(event) = event.event {
+            self.events.lock().unwrap().push(event);
+            if matches!(event, ProviderRetryEvent::Scheduled(_)) {
+                self.controller.interrupt();
+            }
+        }
+    }
+}
+
+#[tokio::test]
+async fn driver_delivers_terminal_accounting_when_retry_observer_cancels() {
+    let partial = SUCCESS
+        .split_once("event: response.output_text.done")
+        .unwrap()
+        .0;
+    let client = Arc::new(ScriptedClient {
+        responses: Mutex::new(VecDeque::from([wire(StatusCode::OK, partial)])),
+        requests: Mutex::new(Vec::new()),
+    });
+    let controller = Arc::new(CancellationController::new());
+    let events = Arc::new(Mutex::new(Vec::new()));
+    let adapter = OpenAIResponsesAdapter::with_client(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(1)),
+        Http::from_arc(client.clone()),
+    );
+    let agent = agentkit_loop::Agent::builder()
+        .model(adapter)
+        .input(vec![Item::text(ItemKind::User, "hello")])
+        .cancellation(controller.handle())
+        .observer(InterruptRetry {
+            events: events.clone(),
+            controller,
+        })
+        .build()
+        .unwrap();
+    let mut driver = agent
+        .start(SessionConfig::new("session").with_response_attempt_supersession())
+        .await
+        .unwrap();
+    driver.next().await.unwrap();
+    let events = events.lock().unwrap();
+    assert_eq!(events.len(), 2);
+    assert!(matches!(events[0], ProviderRetryEvent::Scheduled(_)));
+    let ProviderRetryEvent::Stopped(failure) = events[1] else {
+        panic!("missing terminal accounting")
+    };
+    assert_eq!(failure.reason, ProviderFailureReason::Cancelled);
+    assert_eq!(failure.accounting.attempts, 1);
+    assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+    assert_eq!(client.requests.lock().unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn explicit_driver_drop_hook_finalizes_without_another_poll() {
+    let (mut session, events, _) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test"),
+        vec![wire(StatusCode::OK, SUCCESS)],
+    )
+    .await;
+    let mut turn = session.begin_turn(request(), None).await.unwrap();
+    turn.on_cancelled();
+    turn.on_cancelled();
+    assert!(turn.next_event(None).await.unwrap().is_none());
+    let events = events.lock().unwrap();
+    assert_eq!(events.len(), 1);
+    let ProviderRetryEvent::Stopped(failure) = events[0] else {
+        panic!("missing cancellation")
+    };
+    assert_eq!(failure.reason, ProviderFailureReason::Cancelled);
+    assert_eq!(failure.accounting.attempts, 1);
+}
+
+#[tokio::test]
+async fn completed_http_wait_survives_a_later_interrupted_wait() {
+    let mut first = wire(StatusCode::SERVICE_UNAVAILABLE, "");
+    first
+        .headers
+        .insert("retry-after", HeaderValue::from_static("0.001"));
+    let mut second = first.clone();
+    second
+        .headers
+        .insert("retry-after", HeaderValue::from_static("60"));
+    let (mut session, events, client) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(2)),
+        vec![first, second],
+    )
+    .await;
+    let controller = CancellationController::new();
+    let future = session.begin_turn(request(), Some(TurnCancellation::new(controller.handle())));
+    futures_util::pin_mut!(future);
+    // Poll the real HTTP retry path until the second wait starts. The only real
+    // completed sleep is a server-directed 1ms; no elapsed-time equality is used.
+    tokio::time::timeout(
+        Duration::from_secs(2),
+        futures_util::future::poll_fn(|cx| {
+            assert!(future.as_mut().poll(cx).is_pending());
+            if client.requests.lock().unwrap().len() == 2 {
+                std::task::Poll::Ready(())
+            } else {
+                std::task::Poll::Pending
+            }
+        }),
+    )
+    .await
+    .unwrap();
+    controller.interrupt();
+    assert!(matches!(future.await, Err(LoopError::Cancelled)));
+    let events = events.lock().unwrap();
+    let Some(ProviderRetryEvent::Stopped(failure)) = events.last() else {
+        panic!("missing stop")
+    };
+    assert_eq!(failure.accounting.attempts, 2);
+    assert_eq!(
+        failure.accounting.completed_backoff,
+        Duration::from_millis(1)
+    );
+    assert!(failure.accounting.elapsed >= Duration::from_millis(1));
+}
+
+async fn superseded(turn: &mut OpenAIResponsesTurn) {
+    loop {
+        match turn.next_event(None).await.unwrap() {
+            Some(ModelTurnEvent::ResponseAttemptSuperseded) => return,
+            Some(_) => {}
+            None => panic!("expected retry"),
+        }
+    }
+}
+
+#[tokio::test]
+async fn completed_deferred_stream_wait_survives_a_later_interrupted_wait() {
+    let partial = SUCCESS
+        .split_once("event: response.output_text.done")
+        .unwrap()
+        .0;
+    let (mut session, events, client) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(2)),
+        vec![wire(StatusCode::OK, partial); 2],
+    )
+    .await;
+    session.session = SessionConfig::new("session").with_response_attempt_supersession();
+    let mut turn = session.begin_turn(request(), None).await.unwrap();
+    superseded(&mut turn).await;
+    // Inject fixed delays at the existing deferred-wait seam rather than relying
+    // on random full jitter. Both waits still use the real reopen implementation.
+    turn.pending_delay = Duration::from_millis(1);
+    superseded(&mut turn).await;
+    turn.pending_delay = Duration::from_secs(60);
+    let controller = CancellationController::new();
+    let future = turn.next_event(Some(TurnCancellation::new(controller.handle())));
+    futures_util::pin_mut!(future);
+    assert!(futures_util::poll!(future.as_mut()).is_pending());
+    controller.interrupt();
+    assert!(matches!(future.await, Err(LoopError::Cancelled)));
+    let events = events.lock().unwrap();
+    let Some(ProviderRetryEvent::Stopped(failure)) = events.last() else {
+        panic!("missing stop")
+    };
+    assert_eq!(failure.accounting.attempts, 2);
+    assert_eq!(
+        failure.accounting.completed_backoff,
+        Duration::from_millis(1)
+    );
+    assert_eq!(client.requests.lock().unwrap().len(), 2);
+}
+
+struct PendingAuthentication {
+    initial: bool,
+}
+
+#[async_trait]
+impl AuthenticationProvider for PendingAuthentication {
+    async fn authenticate(
+        &self,
+        previous: Option<&AuthenticationAttempt>,
+    ) -> Result<AuthenticationAttempt, HttpError> {
+        if self.initial || previous.is_some() {
+            futures_util::future::pending().await
+        } else {
+            Ok(AuthenticationAttempt::stateless(HeaderMap::new()))
+        }
+    }
+}
+
+#[tokio::test]
+async fn cancellation_during_initial_auth_and_refresh_preserves_phase_accounting() {
+    for initial in [true, false] {
+        let (mut session, events, client) = observed(
+            OpenAIResponsesConfig::new("secret", "gpt-test")
+                .with_authentication_provider(PendingAuthentication { initial }),
+            if initial {
+                vec![]
+            } else {
+                vec![wire(StatusCode::UNAUTHORIZED, "")]
+            },
+        )
+        .await;
+        let controller = CancellationController::new();
+        let future =
+            session.begin_turn(request(), Some(TurnCancellation::new(controller.handle())));
+        futures_util::pin_mut!(future);
+        assert!(futures_util::poll!(future.as_mut()).is_pending());
+        controller.interrupt();
+        assert!(matches!(future.await, Err(LoopError::Cancelled)));
+        let events = events.lock().unwrap();
+        let Some(ProviderRetryEvent::Stopped(failure)) = events.last() else {
+            panic!("missing stop")
+        };
+        assert_eq!(failure.accounting.attempts, if initial { 0 } else { 1 });
+        assert_eq!(
+            failure.upstream.http_status,
+            if initial { None } else { Some(401) }
+        );
+        assert_eq!(
+            client.requests.lock().unwrap().len(),
+            if initial { 0 } else { 1 }
+        );
+    }
+}
+
+struct PendingClient(Arc<AtomicUsize>);
+
+#[async_trait]
+impl HttpClient for PendingClient {
+    async fn execute(&self, _: HttpRequest) -> Result<HttpResponse, HttpError> {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        futures_util::future::pending().await
+    }
+}
+
+#[tokio::test]
+async fn cancellation_during_send_counts_the_started_attempt() {
+    let sends = Arc::new(AtomicUsize::new(0));
+    let adapter = OpenAIResponsesAdapter::with_client(
+        OpenAIResponsesConfig::new("secret", "gpt-test"),
+        Http::new(PendingClient(sends.clone())),
+    );
+    let mut session = adapter
+        .start_session(SessionConfig::new("session"))
+        .await
+        .unwrap();
+    let events = Arc::new(Mutex::new(Vec::new()));
+    let capture = events.clone();
+    session.set_retry_observer(Some(Arc::new(move |event| {
+        capture.lock().unwrap().push(event)
+    })));
+    let controller = CancellationController::new();
+    let future = session.begin_turn(request(), Some(TurnCancellation::new(controller.handle())));
+    futures_util::pin_mut!(future);
+    assert!(futures_util::poll!(future.as_mut()).is_pending());
+    controller.interrupt();
+    assert!(matches!(future.await, Err(LoopError::Cancelled)));
+    let events = events.lock().unwrap();
+    assert_eq!(events.len(), 1);
+    let ProviderRetryEvent::Stopped(failure) = events[0] else {
+        panic!("missing stop")
+    };
+    assert_eq!(failure.accounting.attempts, 1);
+    assert_eq!(sends.load(Ordering::SeqCst), 1);
+}
+
+#[test]
+fn timeout_and_transport_source_classification_is_typed_and_sanitized() {
+    for (error, expected) in [
+        (
+            HttpError::Timeout {
+                operation: "response stream idle",
+                timeout: Duration::ZERO,
+            },
+            ProviderFailureReason::IdleTimeout,
+        ),
+        (
+            HttpError::Timeout {
+                operation: "secret-transport-operation",
+                timeout: Duration::ZERO,
+            },
+            ProviderFailureReason::AttemptTimeout,
+        ),
+        (
+            HttpError::Other("SECRET-TRANSPORT-BODY".into()),
+            ProviderFailureReason::Transport,
+        ),
+    ] {
+        let failure = transport_failure(error);
+        assert_eq!(failure.error.provider_failure().unwrap().reason, expected);
+        assert!(!format!("{failure:?}").to_lowercase().contains("secret"));
+    }
+}
+
+#[tokio::test]
+async fn ready_finished_event_loses_to_an_expired_logical_deadline() {
+    let (mut session, events, _) = observed(
+        OpenAIResponsesConfig::new("secret", "gpt-test").with_resilience(policy(1)),
+        vec![wire(StatusCode::OK, SUCCESS)],
+    )
+    .await;
+    let mut turn = session.begin_turn(request(), None).await.unwrap();
+    let attempt = turn.attempt.as_mut().unwrap();
+    attempt.decoder.push(SUCCESS.as_bytes()).unwrap();
+    attempt.decoder.process_all_pending().unwrap();
+    while !matches!(
+        attempt.decoder.peek_event(),
+        Some(ModelTurnEvent::Finished(_))
+    ) {
+        assert!(attempt.decoder.pop_event().is_some());
+    }
+    attempt.closed = true; // EOF and final result are already ready.
+    let deadline = turn.context.deadline.as_mut().unwrap();
+    deadline.started_at = Instant::now() - deadline.budget;
+    let error = turn.next_event(None).await.unwrap_err();
+    assert_eq!(
+        error.provider_failure().unwrap().reason,
+        ProviderFailureReason::RetryBudget
+    );
+    assert!(matches!(
+        events.lock().unwrap().as_slice(),
+        [ProviderRetryEvent::Stopped(_)]
+    ));
+}
+
+struct CountedReqwest {
+    client: reqwest::Client,
+    executions: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl HttpClient for CountedReqwest {
+    async fn execute(&self, request: HttpRequest) -> Result<HttpResponse, HttpError> {
+        self.executions.fetch_add(1, Ordering::SeqCst);
+        HttpClient::execute(&self.client, request).await
+    }
+}
+
+#[tokio::test]
+async fn invalid_endpoint_preflight_never_executes_or_retries_real_transport() {
+    // Each endpoint is also rejected locally by reqwest if adapter preflight
+    // regresses. The real transport cannot dispatch a network request here.
+    let client = reqwest::Client::builder().no_proxy().build().unwrap();
+    for endpoint in [
+        "SECRET-ENDPOINT not a URL",
+        "https://",
+        "http://127.0.0.1:SECRET-ENDPOINT",
+        "file:///SECRET-ENDPOINT",
+        "ftp://127.0.0.1/SECRET-ENDPOINT",
+    ] {
+        let executions = Arc::new(AtomicUsize::new(0));
+        let adapter = OpenAIResponsesAdapter::with_client(
+            OpenAIResponsesConfig::new("SECRET-CREDENTIAL", "gpt-test")
+                .with_endpoint(endpoint)
+                .with_resilience(policy(2)),
+            Http::new(CountedReqwest {
+                client: client.clone(),
+                executions: executions.clone(),
+            }),
+        );
+        let mut session = adapter
+            .start_session(SessionConfig::new("session"))
+            .await
+            .unwrap();
+        let events = Arc::new(Mutex::new(Vec::new()));
+        let capture = events.clone();
+        session.set_retry_observer(Some(Arc::new(move |event| {
+            capture.lock().unwrap().push(event)
+        })));
+        let error = match session.begin_turn(request(), None).await {
+            Err(error) => error,
+            Ok(_) => panic!("invalid endpoint accepted"),
+        };
+        let failure = error.provider_failure().unwrap();
+        assert_eq!(failure.reason, ProviderFailureReason::InvalidRequest);
+        assert_eq!(failure.last_attempt_reason, None);
+        assert_eq!(failure.upstream, ProviderClassification::default());
+        assert_eq!(failure.accounting.attempts, 0);
+        assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+        assert_eq!(executions.load(Ordering::SeqCst), 0);
+        let events = events.lock().unwrap();
+        assert_eq!(*events, vec![ProviderRetryEvent::Stopped(*failure)]);
+        let rendered = format!(
+            "{error:?} {error} {}",
+            serde_json::to_string(&*events).unwrap()
+        );
+        assert!(!rendered.contains("SECRET"));
+    }
+}
+
+#[tokio::test]
+async fn invalid_local_headers_preflight_have_zero_attempts_and_no_retry() {
+    for invalid_session in [false, true] {
+        let invalid = format!("SECRET-HEADER{}", char::from(10));
+        let mut config = OpenAIResponsesConfig::chatgpt_private("gpt-test", "SECRET-CREDENTIAL")
+            .with_resilience(policy(2));
+        let mut request = request();
+        if invalid_session {
+            request.session_id = SessionId::new(invalid);
+        } else {
+            config = config.with_originator(invalid);
+        }
+        let (mut session, events, client) = observed(config, vec![]).await;
+        let error = match session.begin_turn(request, None).await {
+            Err(error) => error,
+            Ok(_) => panic!("invalid local header accepted"),
+        };
+        let failure = error.provider_failure().unwrap();
+        assert_eq!(failure.reason, ProviderFailureReason::InvalidRequest);
+        assert_eq!(failure.accounting.attempts, 0);
+        assert_eq!(failure.accounting.completed_backoff, Duration::ZERO);
+        assert_eq!(failure.last_attempt_reason, None);
+        assert_eq!(
+            *events.lock().unwrap(),
+            vec![ProviderRetryEvent::Stopped(*failure)]
+        );
+        assert!(client.requests.lock().unwrap().is_empty());
+        assert!(!format!("{error:?} {error}").contains("SECRET"));
+    }
+}