feat(infinity-provider-protocol,infinity-provider-bedrock,infinity-agent-core)!: provider-classified errors, provider-owned timeouts, and oversized-input recovery

crates/infinity-agent-core/Cargo.toml

@@ -30,7 +30,7 @@ rhai = { workspace = true }
 insta = { version = "1", features = ["json", "redactions"] }
 libc = "0.2"
 infinity-provider-protocol = { path = "../infinity-provider-protocol", version = "^0.1.0", features = ["mock"] }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time", "test-util"] }
 futures-util = { workspace = true, features = ["sink"] }
 
 [lints]

crates/infinity-agent-core/src/system/local/driver.rs

@@ -44,7 +44,8 @@ struct InFlightStep<'a> {
 impl InFlightStep<'_> {
     /// Interrupt the step and wait for it to wind down. The cancellation
     /// path flushes whatever streamed so far to the store before returning,
-    /// so no partial turn is lost.
+    /// so no partial turn is lost. Inputs the model produced no output for
+    /// stay in memory (unpersisted) and are re-sent on the next round.
     async fn cancel(self) -> Result<StepOutcome, BoxError> {
         let _ = self.cancel_tx.send(());
         self.fut.await

crates/infinity-agent-core/src/system/thread.rs

@@ -470,8 +470,10 @@ where
         }
 
         if !any_ready {
-            // Commit anything prepare persisted (processed IDs, interruption
-            // results) even though no completion runs.
+            // Commit anything already known-safe (e.g. processed IDs from
+            // deduped inputs). Interruption results and other fresh inputs
+            // stay unvalidated — and unpersisted — until a completion
+            // produces model output for them.
             self.history.sync().await?;
             return Ok(StepOutcome::Skipped);
         }

crates/infinity-provider-bedrock/Cargo.toml

@@ -23,3 +23,13 @@ tracing-subscriber = { workspace = true }
 
 [lints]
 workspace = true
+
+[features]
+# Tests that talk to the real Bedrock service using your local AWS
+# credentials: `cargo test -p infinity-provider-bedrock --features live-tests`.
+live-tests = []
+
+[dev-dependencies]
+futures-util = { workspace = true }
+aws-smithy-runtime-api = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }

crates/infinity-provider-bedrock/src/convert.rs

@@ -6,8 +6,8 @@ use aws_smithy_types::{Document, Number};
 use base64::Engine;
 use base64::prelude::BASE64_STANDARD;
 use infinity_provider_protocol::{
-    AssistantContent, CompletionError, Image, ImageMediaType, ImageSource, Message, Reasoning,
-    ReasoningContent, ToolDefinition, ToolResultContent, UserContent,
+    AssistantContent, CompletionError, ErrorClass, Image, ImageMediaType, ImageSource, Message,
+    Reasoning, ReasoningContent, ToolDefinition, ToolResultContent, UserContent,
 };
 
 /// Convert a JSON value into a smithy [`Document`] (the representation
@@ -165,23 +165,24 @@ fn reasoning_block(
         })
         .count();
     if signed_text_count > 1 {
-        return Err(CompletionError::ProviderError(
-            "AWS Bedrock does not support multiple signed reasoning text blocks".to_owned(),
+        return Err(CompletionError::provider(
+            ErrorClass::Fatal,
+            "AWS Bedrock does not support multiple signed reasoning text blocks",
         ));
     }
     if signed_text_count == 1 && reasoning.content.len() > 1 {
-        return Err(CompletionError::ProviderError(
+        return Err(CompletionError::provider(
+            ErrorClass::Fatal,
             "AWS Bedrock requires a single signed reasoning text block without additional \
-             reasoning parts"
-                .to_owned(),
+             reasoning parts",
         ));
     }
 
     let text = reasoning.display_text();
     if text.is_empty() {
-        return Err(CompletionError::ProviderError(
-            "AWS Bedrock reasoning conversion requires at least one text or summary block"
-                .to_owned(),
+        return Err(CompletionError::provider(
+            ErrorClass::Fatal,
+            "AWS Bedrock reasoning conversion requires at least one text or summary block",
         ));
     }
 
@@ -200,14 +201,18 @@ fn image_block(image: Image) -> Result<bedrock::ImageBlock, CompletionError> {
         Some(ImageMediaType::GIF) => bedrock::ImageFormat::Gif,
         Some(ImageMediaType::WEBP) => bedrock::ImageFormat::Webp,
         Some(other) => {
-            return Err(CompletionError::ProviderError(format!(
-                "AWS Bedrock does not support {} images",
-                other.to_mime_type()
-            )));
+            return Err(CompletionError::provider(
+                ErrorClass::Fatal,
+                format!(
+                    "AWS Bedrock does not support {} images",
+                    other.to_mime_type()
+                ),
+            ));
         }
         None => {
-            return Err(CompletionError::ProviderError(
-                "image content requires a media type for AWS Bedrock".to_owned(),
+            return Err(CompletionError::provider(
+                ErrorClass::Fatal,
+                "image content requires a media type for AWS Bedrock",
             ));
         }
     };
@@ -217,9 +222,9 @@ fn image_block(image: Image) -> Result<bedrock::ImageBlock, CompletionError> {
             "only base64-encoded image data is supported by AWS Bedrock".into(),
         ));
     };
-    let bytes = BASE64_STANDARD
-        .decode(data)
-        .map_err(|e| CompletionError::ProviderError(format!("invalid base64 image data: {e}")))?;
+    let bytes = BASE64_STANDARD.decode(data).map_err(|e| {
+        CompletionError::provider(ErrorClass::Fatal, format!("invalid base64 image data: {e}"))
+    })?;
 
     bedrock::ImageBlock::builder()
         .format(format)

crates/infinity-provider-bedrock/src/lib.rs

@@ -14,12 +14,20 @@ mod stream;
 use async_trait::async_trait;
 use aws_sdk_bedrockruntime::error::{DisplayErrorContext, ProvideErrorMetadata, SdkError};
 use infinity_provider_protocol::{
-    CompletionError, CompletionRequest, ModelEntry, ModelProvider, ModelStream,
+    CompletionError, CompletionRequest, ErrorClass, ModelEntry, ModelProvider, ModelStream,
 };
 use tokio::sync::OnceCell;
 
 type BoxError = Box<dyn std::error::Error + Send + Sync>;
 
+/// How long to wait for Bedrock to accept a `ConverseStream` request before
+/// giving up. Bedrock occasionally black-holes a request (it neither
+/// responds nor fails); classifying the timeout as [`ErrorClass::Transient`]
+/// lets the caller retry. This deliberately only covers request
+/// *initiation* — once a response stream is live it is never artificially
+/// cut off by us.
+const REQUEST_INITIATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
+
 /// Extract a useful message from an AWS SDK error: the service error message
 /// when present (the SDK's plain `Display` omits it), otherwise the full
 /// error chain.
@@ -34,6 +42,79 @@ where
     }
 }
 
+/// Classify a Bedrock error into the retry classification declared to
+/// callers, from the service error code (when the failure reached the
+/// service) and the error message.
+///
+/// The message heuristics live *here*, in the provider — the agent runtime
+/// only ever sees the resulting [`ErrorClass`].
+pub(crate) fn classify_bedrock_error(code: Option<&str>, message: &str) -> ErrorClass {
+    let msg = message.to_ascii_lowercase();
+    if matches!(
+        code,
+        Some("ThrottlingException" | "ServiceQuotaExceededException")
+    ) || msg.contains("please wait before trying again")
+        || msg.contains("too many requests")
+        || msg.contains("please try again")
+    {
+        return ErrorClass::Throttled;
+    }
+    // Context overflow: Bedrock reports it as a ValidationException
+    // ("Input is too long for requested model."), anthropic models as
+    // "input length and `max_tokens` exceed context limit".
+    if msg.contains("too long")
+        || msg.contains("too large")
+        || msg.contains("input length")
+        || (msg.contains("exceed") && msg.contains("context"))
+    {
+        return ErrorClass::ContextOverflow;
+    }
+    if matches!(
+        code,
+        Some(
+            "InternalServerException"
+                | "ServiceUnavailableException"
+                | "ModelTimeoutException"
+                | "ModelNotReadyException"
+                | "ModelStreamErrorException"
+                | "ModelErrorException"
+        )
+    ) || msg.contains("unexpected end of stream")
+        || msg.contains("unexpected error when processing the request")
+        || msg.contains("is unable to process your request")
+    {
+        return ErrorClass::Transient;
+    }
+    ErrorClass::Fatal
+}
+
+/// Classify a full [`SdkError`]: service errors go through
+/// [`classify_bedrock_error`]; transport-level failures (dispatch, timeout,
+/// unparsable response) are transient; request construction failures are
+/// ours and fatal.
+pub(crate) fn classify_sdk_error<E, R>(err: &SdkError<E, R>) -> ErrorClass
+where
+    E: ProvideErrorMetadata + std::error::Error + 'static,
+    R: std::fmt::Debug,
+{
+    match err {
+        SdkError::ConstructionFailure(_) => ErrorClass::Fatal,
+        SdkError::TimeoutError(_) | SdkError::DispatchFailure(_) | SdkError::ResponseError(_) => {
+            ErrorClass::Transient
+        }
+        _ => classify_bedrock_error(err.code(), &sdk_error_message(err)),
+    }
+}
+
+/// Convert an [`SdkError`] into a classified [`CompletionError`].
+pub(crate) fn completion_error<E, R>(err: &SdkError<E, R>) -> CompletionError
+where
+    E: ProvideErrorMetadata + std::error::Error + 'static,
+    R: std::fmt::Debug,
+{
+    CompletionError::provider(classify_sdk_error(err), sdk_error_message(err))
+}
+
 /// A model offered by the Bedrock provider, along with the Bedrock-specific
 /// invocation configuration that stays internal to this crate.
 struct BedrockModel {
@@ -244,7 +325,7 @@ impl ModelProvider for BedrockProvider {
     ) -> Result<ModelStream, CompletionError> {
         let prepared = prepare_request(&self.models, model_id, request)?;
 
-        let response = self
+        let send = self
             .client()
             .await
             .converse_stream()
@@ -254,11 +335,28 @@ impl ModelProvider for BedrockProvider {
             .set_tool_config(prepared.tool_config)
             .set_inference_config(Some(prepared.inference_config))
             .set_additional_model_request_fields(prepared.additional_params)
-            .send()
+            .send();
+
+        // Guard request *initiation* only (see REQUEST_INITIATION_TIMEOUT);
+        // the returned stream itself is never timed out.
+        let response = tokio::time::timeout(REQUEST_INITIATION_TIMEOUT, send)
             .await
+            .map_err(|_| {
+                tracing::error!(
+                    "Bedrock ConverseStream request initiation timed out after {:?}",
+                    REQUEST_INITIATION_TIMEOUT
+                );
+                CompletionError::provider(
+                    ErrorClass::Transient,
+                    format!(
+                        "timed out waiting {}s for Bedrock to accept the request",
+                        REQUEST_INITIATION_TIMEOUT.as_secs()
+                    ),
+                )
+            })?
             .map_err(|e| {
                 tracing::error!(error = %DisplayErrorContext(&e), "Bedrock ConverseStream SDK error");
-                CompletionError::ProviderError(sdk_error_message(&e))
+                completion_error(&e)
             })?;
 
         Ok(stream::convert_stream(response))
@@ -382,4 +480,63 @@ mod tests {
         };
         assert!(obj.contains_key("anthropic_beta"));
     }
+
+    // ── Error classification ──
+    //
+    // Classification is mostly string/code matching against real Bedrock
+    // responses, so asserting the match table here would be tautological.
+    // The real assertions live in `tests/live.rs` (feature `live-tests`),
+    // which classifies actual Bedrock service errors using local AWS
+    // credentials.
+
+    /// A black-holed `ConverseStream` request must fail with a transient
+    /// (retryable) error after the initiation timeout instead of hanging
+    /// forever. Uses a Bedrock client whose HTTP connector never responds.
+    #[tokio::test(start_paused = true)]
+    async fn request_initiation_times_out_with_transient_error() {
+        #[derive(Debug)]
+        struct NeverRespond;
+        impl aws_smithy_runtime_api::client::http::HttpConnector for NeverRespond {
+            fn call(
+                &self,
+                _request: aws_smithy_runtime_api::client::orchestrator::HttpRequest,
+            ) -> aws_smithy_runtime_api::client::http::HttpConnectorFuture {
+                aws_smithy_runtime_api::client::http::HttpConnectorFuture::new(
+                    std::future::pending(),
+                )
+            }
+        }
+        impl aws_smithy_runtime_api::client::http::HttpClient for NeverRespond {
+            fn http_connector(
+                &self,
+                _settings: &aws_smithy_runtime_api::client::http::HttpConnectorSettings,
+                _components: &aws_smithy_runtime_api::client::runtime_components::RuntimeComponents,
+            ) -> aws_smithy_runtime_api::client::http::SharedHttpConnector {
+                aws_smithy_runtime_api::client::http::SharedHttpConnector::new(NeverRespond)
+            }
+        }
+
+        let config = aws_sdk_bedrockruntime::Config::builder()
+            .behavior_version(aws_config::BehaviorVersion::latest())
+            .region(aws_sdk_bedrockruntime::config::Region::new("us-east-1"))
+            .credentials_provider(aws_sdk_bedrockruntime::config::Credentials::for_tests())
+            .http_client(NeverRespond)
+            .build();
+        let provider = BedrockProvider::new(aws_sdk_bedrockruntime::Client::from_conf(config));
+
+        let started = tokio::time::Instant::now();
+        let Err(err) = provider
+            .invoke_model("global.anthropic.claude-sonnet-4-6", request("hi"))
+            .await
+        else {
+            panic!("black-holed request must time out")
+        };
+        assert_eq!(err.class(), ErrorClass::Transient);
+        assert!(
+            err.to_string().contains("timed out"),
+            "unexpected message: {err}"
+        );
+        // The timeout must be the initiation timeout, not some other layer.
+        assert!(started.elapsed() >= REQUEST_INITIATION_TIMEOUT);
+    }
 }