diff --git a/objectstore-server/src/endpoints/batch.rs b/objectstore-server/src/endpoints/batch.rs
index 20174881..2d733667 100644
--- a/objectstore-server/src/endpoints/batch.rs
+++ b/objectstore-server/src/endpoints/batch.rs
@@ -93,11 +93,7 @@ async fn batch(
         let usecase = context.usecase.clone();
         move |op| {
             if let Operation::Insert(ins) = op {
-                state
-                    .config
-                    .usecases
-                    .validate(&usecase, &ins.metadata)
-                    .map_err(|e| ApiError::Client(e.to_string()))?;
+                state.config.usecases.validate(&usecase, &ins.metadata)?;
             }
             Ok(())
         }
diff --git a/objectstore-server/src/endpoints/common.rs b/objectstore-server/src/endpoints/common.rs
index 96fcde0e..02cbdcd6 100644
--- a/objectstore-server/src/endpoints/common.rs
+++ b/objectstore-server/src/endpoints/common.rs
@@ -1,45 +1,19 @@
 //! Common types and utilities for API endpoints.
 
+use std::borrow::Cow;
 use std::error::Error;
 
 use axum::Json;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use http::HeaderValue;
-use objectstore_service::error::Error as ServiceError;
+use objectstore_service::error::{Error as ServiceError, ErrorKind as ServiceErrorKind};
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
 use crate::auth::AuthError;
 use crate::extractors::batch::BatchError;
 
-/// Error type for API operations.
-#[derive(Debug, Error)]
-pub enum ApiError {
-    /// Errors indicating malformed or illegal requests.
-    #[error("client error: {0}")]
-    Client(String),
-
-    /// Authorization/authentication errors.
-    #[error("auth error: {0}")]
-    Auth(#[from] AuthError),
-
-    /// Service errors, indicating that something went wrong when receiving or executing a request.
-    #[error("service error: {0}")]
-    Service(#[from] ServiceError),
-
-    /// Errors encountered when parsing or executing a batch request.
-    #[error("batch error: {0}")]
-    Batch(#[from] BatchError),
-
-    /// Internal server errors.
-    #[error("internal error: {0}")]
-    Internal(String),
-}
-
-/// Result type for API operations.
-pub type ApiResult<T> = Result<T, ApiError>;
-
 /// A JSON error response returned by the API.
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ApiErrorResponse {
@@ -67,11 +41,77 @@ impl ApiErrorResponse {
     }
 }
 
+/// Error type for API operations.
+#[derive(Debug, Error)]
+pub enum ApiError {
+    /// Errors indicating malformed or illegal requests.
+    #[error("client error: {context}")]
+    Client {
+        /// Context describing the operation that failed.
+        context: Cow<'static, str>,
+        /// The underlying error, if available.
+        #[source]
+        cause: Option<Box<dyn Error + Send + Sync>>,
+    },
+
+    /// Authorization/authentication errors.
+    #[error("auth error: {0}")]
+    Auth(#[from] AuthError),
+
+    /// Service errors, indicating that something went wrong when receiving or executing a request.
+    #[error("service error: {0}")]
+    Service(#[from] ServiceError),
+
+    /// Errors encountered when parsing or executing a batch request.
+    #[error("batch error: {0}")]
+    Batch(#[from] BatchError),
+
+    /// Internal server errors.
+    #[error("internal error: {context}")]
+    Internal {
+        /// Context describing the operation that failed.
+        context: Cow<'static, str>,
+        /// The underlying error, if available.
+        #[source]
+        cause: Option<Box<dyn Error + Send + Sync>>,
+    },
+}
+
 impl ApiError {
+    /// Creates a client error with context and an underlying cause.
+    pub fn map_client<E>(context: impl Into<Cow<'static, str>>, cause: E) -> Self
+    where
+        E: Error + Send + Sync + 'static,
+    {
+        Self::Client {
+            context: context.into(),
+            cause: Some(Box::new(cause)),
+        }
+    }
+
+    /// Creates a client error with context and no underlying cause.
+    pub fn client(context: impl Into<Cow<'static, str>>) -> Self {
+        Self::Client {
+            context: context.into(),
+            cause: None,
+        }
+    }
+
+    /// Creates an internal server error with context and an underlying cause.
+    pub fn internal<E>(context: impl Into<Cow<'static, str>>, cause: E) -> Self
+    where
+        E: Error + Send + Sync + 'static,
+    {
+        Self::Internal {
+            context: context.into(),
+            cause: Some(Box::new(cause)),
+        }
+    }
+
     /// Returns the HTTP status code appropriate for this error variant.
     pub fn status(&self) -> StatusCode {
         match &self {
-            ApiError::Client(_) => StatusCode::BAD_REQUEST,
+            ApiError::Client { .. } => StatusCode::BAD_REQUEST,
 
             ApiError::Batch(BatchError::BadRequest(_))
             | ApiError::Batch(BatchError::Metadata(_))
@@ -90,23 +130,28 @@ impl ApiError {
             ApiError::Auth(AuthError::NotPermitted) => StatusCode::FORBIDDEN,
             ApiError::Auth(AuthError::InternalError(_)) => StatusCode::INTERNAL_SERVER_ERROR,
 
-            ApiError::Service(ServiceError::Client(_)) => StatusCode::BAD_REQUEST,
-            ApiError::Service(ServiceError::Metadata(_)) => StatusCode::BAD_REQUEST,
-            ApiError::Service(ServiceError::RangeNotSatisfiable { .. }) => {
-                StatusCode::RANGE_NOT_SATISFIABLE
-            }
-            ApiError::Service(ServiceError::InvalidUploadId(_)) => StatusCode::BAD_REQUEST,
-            ApiError::Service(ServiceError::UnknownUploadSession) => StatusCode::BAD_REQUEST,
-            ApiError::Service(ServiceError::ChunkExceedsUploadLength { .. }) => {
-                StatusCode::BAD_REQUEST
-            }
-            ApiError::Service(ServiceError::UploadOffsetMismatch { .. }) => StatusCode::CONFLICT,
-            ApiError::Service(ServiceError::UploadSessionGone) => StatusCode::GONE,
-            ApiError::Service(ServiceError::AtCapacity) => StatusCode::TOO_MANY_REQUESTS,
-            ApiError::Service(ServiceError::NotImplemented) => StatusCode::NOT_IMPLEMENTED,
-            ApiError::Service(_) => StatusCode::INTERNAL_SERVER_ERROR,
-
-            ApiError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            ApiError::Service(error) => match error.kind() {
+                ServiceErrorKind::InvalidMetadata
+                | ServiceErrorKind::InvalidUploadId
+                | ServiceErrorKind::ClientStream
+                | ServiceErrorKind::UnknownUploadSession
+                | ServiceErrorKind::ChunkExceedsUploadLength { .. } => StatusCode::BAD_REQUEST,
+                ServiceErrorKind::RangeNotSatisfiable { .. } => StatusCode::RANGE_NOT_SATISFIABLE,
+                ServiceErrorKind::UploadOffsetMismatch { .. } => StatusCode::CONFLICT,
+                ServiceErrorKind::UploadSessionGone => StatusCode::GONE,
+                ServiceErrorKind::AtCapacity => StatusCode::TOO_MANY_REQUESTS,
+                ServiceErrorKind::Unsupported => StatusCode::NOT_IMPLEMENTED,
+                ServiceErrorKind::BackendRateLimited => StatusCode::TOO_MANY_REQUESTS,
+                ServiceErrorKind::BackendTimeout | ServiceErrorKind::BackendUnavailable => {
+                    StatusCode::SERVICE_UNAVAILABLE
+                }
+                ServiceErrorKind::BackendFailure
+                | ServiceErrorKind::CorruptData
+                | ServiceErrorKind::Panic
+                | ServiceErrorKind::Internal => StatusCode::INTERNAL_SERVER_ERROR,
+            },
+
+            ApiError::Internal { .. } => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 
@@ -134,6 +179,21 @@ impl IntoResponse for ApiError {
     }
 }
 
+impl From<crate::usecases::UseCaseError> for ApiError {
+    fn from(error: crate::usecases::UseCaseError) -> Self {
+        ApiError::map_client("use case policy violation", error)
+    }
+}
+
+impl From<objectstore_types::metadata::Error> for ApiError {
+    fn from(error: objectstore_types::metadata::Error) -> Self {
+        ApiError::map_client("invalid metadata", error)
+    }
+}
+
+/// Result type for API operations.
+pub type ApiResult<T> = Result<T, ApiError>;
+
 /// Inserts `Accept-Ranges: bytes` into the response headers.
 pub fn insert_accept_ranges(response: &mut Response) {
     response.headers_mut().insert(
diff --git a/objectstore-server/src/endpoints/multipart.rs b/objectstore-server/src/endpoints/multipart.rs
index f1e231d2..cb8824e5 100644
--- a/objectstore-server/src/endpoints/multipart.rs
+++ b/objectstore-server/src/endpoints/multipart.rs
@@ -16,7 +16,6 @@ use bytes::Bytes;
 use futures::StreamExt;
 use http::HeaderValue;
 use http::header;
-use objectstore_service::error::Error as ServiceError;
 use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_service::multipart::{CompletedPart, PartNumber, UploadId};
 use objectstore_types::metadata::Metadata;
@@ -96,13 +95,12 @@ async fn initiate_inner(
     headers: HeaderMap,
 ) -> ApiResult<Response> {
     // TODO: Update time_created in `complete`, when we have a Service API to mutate metadata.
-    let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?;
+    let metadata = Metadata::from_insert_headers(&headers, "")?;
 
     state
         .config
         .usecases
-        .validate(&id.context().usecase, &metadata)
-        .map_err(|e| ApiError::Client(e.to_string()))?;
+        .validate(&id.context().usecase, &metadata)?;
 
     let upload_id = service.initiate_multipart(id.clone(), metadata).await?;
 
@@ -125,7 +123,7 @@ async fn upload_part(
         .get(header::CONTENT_LENGTH)
         .and_then(|v| v.to_str().ok())
         .and_then(|v| v.parse::<u64>().ok())
-        .ok_or_else(|| ApiError::Client("Content-Length header is required".into()))?;
+        .ok_or_else(|| ApiError::client("content-length header is required"))?;
 
     let content_md5 = headers
         .get("content-md5")
diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs
index 2504eb51..2cab0545 100644
--- a/objectstore-server/src/endpoints/objects.rs
+++ b/objectstore-server/src/endpoints/objects.rs
@@ -7,7 +7,7 @@ use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use axum::routing;
 use axum::{Json, Router};
-use objectstore_service::error::Error as ServiceError;
+use objectstore_service::error::ErrorKind;
 use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_types::headers::ExtValue;
 use objectstore_types::metadata::Metadata;
@@ -89,13 +89,12 @@ async fn create_object(
     headers: HeaderMap,
     MeteredBody(body): MeteredBody,
 ) -> ApiResult<Response> {
-    let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?;
+    let metadata = Metadata::from_insert_headers(&headers, "")?;
 
     state
         .config
         .usecases
-        .validate(&context.usecase, &metadata)
-        .map_err(|e| ApiError::Client(e.to_string()))?;
+        .validate(&context.usecase, &metadata)?;
 
     let response_id = service.insert_object(context, None, metadata, body).await?;
     let response = Json(InsertObjectResponse {
@@ -118,23 +117,28 @@ async fn object_get(
     let (metadata, content_range, stream) = match result {
         Ok(Some(result)) => result,
         Ok(None) => return Ok(StatusCode::NOT_FOUND.into_response()),
-        Err(ApiError::Service(ServiceError::RangeNotSatisfiable { total })) => {
-            let mut response = (
-                StatusCode::RANGE_NOT_SATISFIABLE,
-                [(
-                    http::header::CONTENT_RANGE,
-                    ContentRange::unsatisfiable_total_to_header_value(total),
-                )],
-            )
-                .into_response();
-            insert_accept_ranges(&mut response);
-            return Ok(response);
-        }
+        Err(ApiError::Service(e)) => match e.kind() {
+            ErrorKind::RangeNotSatisfiable { total } => {
+                let mut response = (
+                    StatusCode::RANGE_NOT_SATISFIABLE,
+                    [(
+                        http::header::CONTENT_RANGE,
+                        ContentRange::unsatisfiable_total_to_header_value(total),
+                    )],
+                )
+                    .into_response();
+                insert_accept_ranges(&mut response);
+                return Ok(response);
+            }
+            _ => return Err(e.into()),
+        },
         Err(e) => return Err(e),
     };
 
     let stream = state.meter_stream(stream, &context);
-    let mut metadata_headers = metadata.to_headers("").map_err(ServiceError::from)?;
+    let mut metadata_headers = metadata
+        .to_headers("")
+        .map_err(|error| ApiError::internal("encoding object response metadata", error))?;
 
     let mut response = match content_range {
         Some(ref content_range) => {
@@ -168,7 +172,9 @@ async fn object_head(service: AuthAwareService, Xt(id): Xt<ObjectId>) -> ApiResu
         return Ok(StatusCode::NOT_FOUND.into_response());
     };
 
-    let mut headers = metadata.to_headers("").map_err(ServiceError::from)?;
+    let mut headers = metadata
+        .to_headers("")
+        .map_err(|error| ApiError::internal("encoding object response metadata", error))?;
     insert_content_length(&mut headers, &metadata);
 
     let mut response = (StatusCode::OK, headers).into_response();
@@ -245,15 +251,14 @@ async fn insert_object(
     headers: HeaderMap,
     MeteredBody(body): MeteredBody,
 ) -> ApiResult<Response> {
-    let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?;
+    let metadata = Metadata::from_insert_headers(&headers, "")?;
 
     let ObjectId { context, key } = id;
 
     state
         .config
         .usecases
-        .validate(&context.usecase, &metadata)
-        .map_err(|e| ApiError::Client(e.to_string()))?;
+        .validate(&context.usecase, &metadata)?;
 
     let response_id = service
         .insert_object(context, Some(key), metadata, body)
diff --git a/objectstore-server/src/endpoints/resumable.rs b/objectstore-server/src/endpoints/resumable.rs
index 9fdae841..0c9a7059 100644
--- a/objectstore-server/src/endpoints/resumable.rs
+++ b/objectstore-server/src/endpoints/resumable.rs
@@ -18,7 +18,7 @@ use axum::response::{IntoResponse, Response};
 use axum::{Json, http};
 use axum_extra::TypedHeader;
 use axum_extra::headers::ContentLength;
-use objectstore_service::error::Error as ServiceError;
+use objectstore_service::error::{Error as ServiceError, ErrorKind};
 use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_service::stream::ClientStream;
 use objectstore_types::metadata::Metadata;
@@ -90,18 +90,17 @@ async fn create_session_for_id(
     body: ClientStream,
 ) -> ApiResult<Response> {
     require_empty_body(content_length, body, "resumable session creation").await?;
-    let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?;
+    let metadata = Metadata::from_insert_headers(&headers, "")?;
 
     state
         .config
         .usecases
-        .validate(&id.context().usecase, &metadata)
-        .map_err(|e| ApiError::Client(e.to_string()))?;
+        .validate(&id.context().usecase, &metadata)?;
 
     let session = service
         .create_upload_session(id.clone(), metadata, total_length)
         .await?
-        .ok_or(ServiceError::NotImplemented)?;
+        .ok_or_else(|| ServiceError::from(ErrorKind::Unsupported))?;
 
     let body = Json(CreateSessionResponse {
         key: id.key().to_owned(),
@@ -134,7 +133,7 @@ pub(super) async fn continue_session(
         UploadOffset::At(offset) => {
             let content_length = content_length
                 .map(|TypedHeader(ContentLength(length))| length)
-                .ok_or_else(|| ApiError::Client("Content-Length header is required".into()))?;
+                .ok_or_else(|| ApiError::client("content-length header is required"))?;
             service
                 .put_chunk(id, session, offset, content_length, body)
                 .await
@@ -170,13 +169,17 @@ pub(super) async fn cancel_session(
 fn progress_response(progress: ApiResult<UploadProgress>, key: String) -> ApiResult<Response> {
     let progress = match progress {
         Ok(progress) => progress,
-        Err(error @ ApiError::Service(ServiceError::UploadOffsetMismatch { offset })) => {
-            let mut response = error.into_response();
-            response
-                .headers_mut()
-                .insert(HEADER_UPLOAD_OFFSET, http::HeaderValue::from(offset));
-            return Ok(response);
-        }
+        Err(ApiError::Service(error)) => match error.kind() {
+            ErrorKind::UploadOffsetMismatch { offset } => {
+                let error = ApiError::Service(error);
+                let mut response = error.into_response();
+                response
+                    .headers_mut()
+                    .insert(HEADER_UPLOAD_OFFSET, http::HeaderValue::from(offset));
+                return Ok(response);
+            }
+            _ => return Err(ApiError::Service(error)),
+        },
         Err(error) => return Err(error),
     };
 
@@ -236,7 +239,7 @@ mod tests {
 
     #[tokio::test]
     async fn offset_mismatch_answers_conflict_with_the_authoritative_offset() {
-        let mismatch = ServiceError::UploadOffsetMismatch { offset: 786_432 };
+        let mismatch = ErrorKind::UploadOffsetMismatch { offset: 786_432 }.into();
         let response =
             progress_response(Err(ApiError::Service(mismatch)), "my-key".into()).unwrap();
 
@@ -252,15 +255,16 @@ mod tests {
 
     #[tokio::test]
     async fn other_errors_propagate_unchanged() {
-        let gone = ApiError::Service(ServiceError::UploadSessionGone);
+        let gone = ApiError::Service(ErrorKind::UploadSessionGone.into());
         let error = progress_response(Err(gone), "my-key".into()).unwrap_err();
         assert_eq!(error.status(), StatusCode::GONE);
 
-        let oversized = ApiError::Service(ServiceError::ChunkExceedsUploadLength {
-            offset: 8,
-            content_length: 4,
-            upload_length: 10,
-        });
+        let oversized =
+            ApiError::Service(ServiceError::from(ErrorKind::ChunkExceedsUploadLength {
+                offset: 8,
+                content_length: 4,
+                upload_length: 10,
+            }));
         let error = progress_response(Err(oversized), "my-key".into()).unwrap_err();
         assert_eq!(error.status(), StatusCode::BAD_REQUEST);
     }
diff --git a/objectstore-server/src/extractors/byte_range.rs b/objectstore-server/src/extractors/byte_range.rs
index 9f39dc77..0ef5f156 100644
--- a/objectstore-server/src/extractors/byte_range.rs
+++ b/objectstore-server/src/extractors/byte_range.rs
@@ -23,7 +23,7 @@ impl FromRequestParts<ServiceState> for OptionalByteRange {
         };
         let range = range
             .to_str()
-            .map_err(|_| ApiError::Client("invalid Range header".into()))?;
+            .map_err(|_| ApiError::client("invalid range header"))?;
 
         match range.parse::<ByteRange>() {
             Ok(range) => Ok(Self(Some(range))),
@@ -43,7 +43,7 @@ impl FromRequestParts<ServiceState> for OptionalByteRange {
             // The client requested an invalid unit or sent a malformed header.
             // We could fall back, but better fail hard and let them know they sent something
             // invalid.
-            Err(err) => Err(ApiError::Client(format!("invalid Range header: {err}"))),
+            Err(err) => Err(ApiError::map_client("invalid range header", err)),
         }
     }
 }
diff --git a/objectstore-server/src/resumable.rs b/objectstore-server/src/resumable.rs
index 3ad06687..a43f9f94 100644
--- a/objectstore-server/src/resumable.rs
+++ b/objectstore-server/src/resumable.rs
@@ -71,7 +71,7 @@ where
         }
 
         let Query(query) = Query::<ResumableQuery>::try_from_uri(&parts.uri)
-            .map_err(|error| ApiError::Client(error.to_string()))?;
+            .map_err(|error| ApiError::map_client("invalid query parameters", error))?;
 
         Ok(query.classify())
     }
@@ -94,7 +94,7 @@ where
 
     async fn from_request_parts(parts: &mut Parts, _state: &S) -> ApiResult<Session> {
         let Query(SessionQuery { session }) = Query::<SessionQuery>::try_from_uri(&parts.uri)
-            .map_err(|error| ApiError::Client(error.to_string()))?;
+            .map_err(|error| ApiError::map_client("invalid query parameters", error))?;
         Ok(Session(decode_session_token(&session)?))
     }
 }
@@ -179,17 +179,17 @@ where
 fn decode_session_token(encoded: &str) -> ApiResult<SessionToken> {
     let bytes = URL_SAFE_NO_PAD
         .decode(encoded)
-        .map_err(|error| ApiError::Client(error.to_string()))?;
+        .map_err(|error| ApiError::map_client("invalid session token", error))?;
 
     if URL_SAFE_NO_PAD.encode(&bytes) != encoded {
-        return Err(ApiError::Client(
-            "session token must use unpadded base64url encoding".into(),
+        return Err(ApiError::client(
+            "session token must use unpadded base64 URL encoding",
         ));
     }
 
     String::from_utf8(bytes)
         .map(SessionToken::from)
-        .map_err(|error| ApiError::Client(error.to_string()))
+        .map_err(|error| ApiError::map_client("session token is not valid UTF-8", error))
 }
 
 /// Confirms that a request neither declares nor streams a non-empty body.
@@ -199,14 +199,14 @@ pub(crate) async fn require_empty_body(
     request: &str,
 ) -> ApiResult<()> {
     if content_length.is_some_and(|ContentLength(length)| length > 0) {
-        return Err(ApiError::Client(format!(
+        return Err(ApiError::client(format!(
             "{request} must be sent with an empty body"
         )));
     }
 
     while let Some(chunk) = body.try_next().await.map_err(ServiceError::from)? {
         if !chunk.is_empty() {
-            return Err(ApiError::Client(format!(
+            return Err(ApiError::client(format!(
                 "{request} must be sent with an empty body"
             )));
         }
diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md
index ba397773..7cdee09f 100644
--- a/objectstore-service/docs/architecture.md
+++ b/objectstore-service/docs/architecture.md
@@ -237,7 +237,7 @@ A concurrency limiter caps in-flight
 backend operations. When all execution permits are held, new operations are
 queued — adding latency instead of rejecting immediately. The queue itself is
 bounded in both depth and time: operations that cannot be served within those
-limits fail with [`Error::AtCapacity`](error::Error::AtCapacity).
+limits fail with [`ErrorKind::AtCapacity`](error::ErrorKind::AtCapacity).
 
 The default execution limit is
 [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See
diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs
index 3fa64593..f8e377d3 100644
--- a/objectstore-service/src/backend/bigtable.rs
+++ b/objectstore-service/src/backend/bigtable.rs
@@ -50,7 +50,7 @@ use crate::backend::common::{
 use crate::change_stream::{
     ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream,
 };
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Result, ResultExt as _};
 use crate::gcp_auth::PrefetchingTokenProvider;
 use crate::id::ObjectId;
 use crate::stream::{ChunkedBytes, ClientStream};
@@ -519,7 +519,7 @@ fn object_mutations(
     metadata.size = Some(payload.len());
 
     let metadata_bytes = serde_json::to_vec(&metadata)
-        .map_err(|cause| Error::serde("failed to serialize metadata", cause))?;
+        .context(ErrorKind::Internal, "encoding Bigtable object metadata")?;
 
     let mutations = [
         // NB: We explicitly delete the row to clear metadata on overwrite.
@@ -607,7 +607,7 @@ fn tombstone_mutations(tombstone: &Tombstone, now: SystemTime) -> Result<[v2::Mu
             column_qualifier: COLUMN_TOMBSTONE_META.to_owned(),
             timestamp_micros,
             value: serde_json::to_vec(&tombstone_meta)
-                .map_err(|cause| Error::serde("failed to serialize tombstone", cause))?,
+                .context(ErrorKind::Internal, "encoding Bigtable tombstone metadata")?,
         })),
     ])
 }
@@ -679,10 +679,10 @@ impl RowData {
                     payload = cell.value;
                 }
                 COLUMN_TOMBSTONE_META => {
-                    tombstone_meta_opt =
-                        Some(serde_json::from_slice(&cell.value).map_err(|cause| {
-                            Error::serde("failed to deserialize tombstone meta", cause)
-                        })?);
+                    tombstone_meta_opt = Some(serde_json::from_slice(&cell.value).context(
+                        ErrorKind::CorruptData,
+                        "decoding Bigtable tombstone metadata",
+                    )?);
                 }
                 COLUMN_METADATA => {
                     if let Ok(legacy_meta) =
@@ -695,10 +695,10 @@ impl RowData {
                             expiration_policy: legacy_meta.expiration_policy,
                         });
                     } else {
-                        metadata_opt =
-                            Some(serde_json::from_slice(&cell.value).map_err(|cause| {
-                                Error::serde("failed to deserialize metadata", cause)
-                            })?);
+                        metadata_opt = Some(serde_json::from_slice(&cell.value).context(
+                            ErrorKind::CorruptData,
+                            "decoding Bigtable object metadata",
+                        )?);
                     }
                 }
                 _ => {}
@@ -763,9 +763,9 @@ fn parse_redirect_target(redirect_path: &[u8], tombstone_id: &ObjectId) -> Resul
         Ok(tombstone_id.clone())
     } else {
         let redirect_str = std::str::from_utf8(redirect_path)
-            .map_err(|_| Error::generic("invalid UTF-8 in redirect path"))?;
+            .context(ErrorKind::CorruptData, "decoding Bigtable redirect target")?;
         ObjectId::from_storage_path(redirect_str)
-            .ok_or_else(|| Error::generic("corrupt redirect path"))
+            .ok_or_else(|| Error::new(ErrorKind::CorruptData, "parsing Bigtable redirect target"))
     }
 }
 
@@ -1041,7 +1041,10 @@ impl Backend for BigTableBackend {
             TieredGet::Object(metadata, content_range, payload) => {
                 Ok(Some((metadata, content_range, payload)))
             }
-            TieredGet::Tombstone(_) => Err(Error::UnexpectedTombstone),
+            TieredGet::Tombstone(_) => Err(Error::new(
+                ErrorKind::Internal,
+                "unexpected Bigtable tombstone",
+            )),
             TieredGet::NotFound => Ok(None),
         }
     }
@@ -1050,7 +1053,10 @@ impl Backend for BigTableBackend {
     async fn get_metadata(&self, id: &ObjectId) -> Result<MetadataResponse> {
         match self.get_tiered_metadata(id).await? {
             TieredMetadata::Object(metadata) => Ok(Some(metadata)),
-            TieredMetadata::Tombstone(_) => Err(Error::UnexpectedTombstone),
+            TieredMetadata::Tombstone(_) => Err(Error::new(
+                ErrorKind::Internal,
+                "unexpected Bigtable tombstone",
+            )),
             TieredMetadata::NotFound => Ok(None),
         }
     }
@@ -1119,7 +1125,10 @@ impl HighVolumeBackend for BigTableBackend {
             }
         }
 
-        Err(Error::generic("BigTable: race loop in put_non_tombstone"))
+        Err(Error::new(
+            ErrorKind::Internal,
+            "Bigtable put race exhausted",
+        ))
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
@@ -1229,8 +1238,9 @@ impl HighVolumeBackend for BigTableBackend {
             }
         }
 
-        Err(Error::generic(
-            "BigTable: race loop in delete_non_tombstone",
+        Err(Error::new(
+            ErrorKind::Internal,
+            "Bigtable delete race exhausted",
         ))
     }
 
@@ -1300,14 +1310,9 @@ impl HighVolumeBackend for BigTableBackend {
 /// required by BigTable, the resulting timestamp has millisecond precision, with the last digits at
 /// 0.
 fn ttl_to_micros(ttl: Duration, from: SystemTime) -> Result<i64> {
-    let deadline = from.checked_add(ttl).ok_or_else(|| Error::Generic {
-        context: format!(
-            "TTL duration overflow: {} plus {}s cannot be represented as SystemTime",
-            humantime::format_rfc3339_seconds(from),
-            ttl.as_secs()
-        ),
-        cause: None,
-    })?;
+    let deadline = from
+        .checked_add(ttl)
+        .ok_or_else(|| Error::new(ErrorKind::Internal, "calculating Bigtable expiration"))?;
 
     system_time_to_micros(deadline)
 }
@@ -1319,19 +1324,12 @@ fn ttl_to_micros(ttl: Duration, from: SystemTime) -> Result<i64> {
 fn system_time_to_micros(deadline: SystemTime) -> Result<i64> {
     let millis = deadline
         .duration_since(SystemTime::UNIX_EPOCH)
-        .map_err(|e| Error::Generic {
-            context: format!(
-                "unable to get duration since UNIX_EPOCH for SystemTime {}",
-                humantime::format_rfc3339_seconds(deadline)
-            ),
-            cause: Some(Box::new(e)),
-        })?
+        .context(ErrorKind::Internal, "converting Bigtable timestamp")?
         .as_millis();
 
-    (millis * 1000).try_into().map_err(|e| Error::Generic {
-        context: format!("failed to convert {millis}ms to i64 microseconds"),
-        cause: Some(Box::new(e)),
-    })
+    (millis * 1000)
+        .try_into()
+        .context(ErrorKind::Internal, "converting Bigtable timestamp")
 }
 
 /// Converts a wall-clock time to Bigtable's microsecond timestamp, saturating at `i64::MAX`
@@ -1382,10 +1380,10 @@ where
             Ok(res) => return Ok(res),
             Err(e) if retry_count >= REQUEST_RETRY_COUNT || !is_retryable(&e) => {
                 objectstore_metrics::count!("bigtable.failures", action = context);
-                return Err(Error::Generic {
-                    context: format!("Bigtable: `{context}` failed"),
-                    cause: Some(Box::new(e)),
-                });
+                return Err(e).context(
+                    ErrorKind::BackendFailure,
+                    format!("running Bigtable {context}"),
+                );
             }
             Err(e) => {
                 retry_count += 1;
@@ -1439,7 +1437,7 @@ fn apply_range(payload: Bytes, range: Option<ByteRange>) -> Result<(Option<Conte
     let total = payload.len() as u64;
     let content_range = byte_range
         .resolve(total)
-        .ok_or(Error::RangeNotSatisfiable { total })?;
+        .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
 
     let sliced = payload.slice(content_range.start as usize..content_range.end as usize + 1);
     Ok((Some(content_range), sliced))
@@ -2031,14 +2029,18 @@ mod tests {
         }
 
         // Legacy reads must error rather than leak tombstone data.
-        assert!(matches!(
-            backend.get_object(&hv_id, None).await,
-            Err(Error::UnexpectedTombstone)
-        ));
-        assert!(matches!(
-            backend.get_metadata(&hv_id).await,
-            Err(Error::UnexpectedTombstone)
-        ));
+        assert!(
+            backend
+                .get_object(&hv_id, None)
+                .await
+                .is_err_and(|error| error.kind() == ErrorKind::Internal)
+        );
+        assert!(
+            backend
+                .get_metadata(&hv_id)
+                .await
+                .is_err_and(|error| error.kind() == ErrorKind::Internal)
+        );
 
         // Idempotent retry: retry with the same target succeeds
         let second = backend
@@ -2512,7 +2514,7 @@ mod tests {
         let id = put_range_test_object(&backend).await?;
 
         match backend.get_object(&id, Some(ByteRange::From(100))).await {
-            Err(Error::RangeNotSatisfiable { total }) => assert_eq!(total, 22),
+            Err(error) if matches!(error.kind(), ErrorKind::RangeNotSatisfiable { total: 22 }) => {}
             Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"),
             Err(e) => panic!("expected RangeNotSatisfiable, got {e:?}"),
         }
diff --git a/objectstore-service/src/backend/common.rs b/objectstore-service/src/backend/common.rs
index 5e36469a..b4ce87b4 100644
--- a/objectstore-service/src/backend/common.rs
+++ b/objectstore-service/src/backend/common.rs
@@ -8,7 +8,7 @@ use objectstore_types::resumable::{SessionToken, UploadProgress};
 
 use bytes::Bytes;
 
-use crate::error::{Error, Result};
+use crate::error::{ErrorKind, Result};
 use crate::id::ObjectId;
 use crate::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
@@ -68,10 +68,10 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static {
 
     /// Borrows this backend as a [`MultipartUploadBackend`] if supported.
     ///
-    /// The default returns [`Error::NotImplemented`]. Backends that implement
+    /// The default returns an [`ErrorKind::Unsupported`]. Backends that implement
     /// [`MultipartUploadBackend`] should override this to return `Ok(self)`.
     fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> {
-        Err(Error::NotImplemented)
+        Err(ErrorKind::Unsupported.into())
     }
 
     /// Creates a resumable upload session for the object at `id`.
@@ -99,8 +99,8 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static {
     /// Writes a chunk of `content_length` bytes at `offset` into an open session.
     ///
     /// `offset` must equal the offset the backend currently holds.
-    /// Returns [`Error::UnknownUploadSession`] when `session` does not identify an open session,
-    /// and [`Error::ChunkExceedsUploadLength`] when the chunk would exceed the total length
+    /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify an open session,
+    /// and [`ErrorKind::ChunkExceedsUploadLength`] when the chunk would exceed the total length
     /// declared when the session was created.
     async fn put_chunk(
         &self,
@@ -111,23 +111,23 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static {
         stream: ClientStream,
     ) -> Result<UploadProgress> {
         let _ = (id, session, offset, content_length, stream);
-        Err(Error::NotImplemented)
+        Err(ErrorKind::Unsupported.into())
     }
 
     /// Reports how far the session has progressed.
     ///
-    /// Returns [`Error::UnknownUploadSession`] when `session` does not identify an open session.
+    /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify an open session.
     async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result<UploadProgress> {
         let _ = (id, session);
-        Err(Error::NotImplemented)
+        Err(ErrorKind::Unsupported.into())
     }
 
     /// Cancels an upload session, discarding whatever was uploaded.
     ///
-    /// Returns [`Error::UnknownUploadSession`] when `session` does not identify an open session.
+    /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify an open session.
     async fn cancel_upload(&self, id: &ObjectId, session: &SessionToken) -> Result<()> {
         let _ = (id, session);
-        Err(Error::NotImplemented)
+        Err(ErrorKind::Unsupported.into())
     }
 }
 
diff --git a/objectstore-service/src/backend/extensions.rs b/objectstore-service/src/backend/extensions.rs
index 31503126..040c75ec 100644
--- a/objectstore-service/src/backend/extensions.rs
+++ b/objectstore-service/src/backend/extensions.rs
@@ -6,13 +6,29 @@
 //! structured error code and message from it (JSON for GCS JSON API, XML for GCS
 //! XML API and S3).
 
-use reqwest::{Response, header};
+use std::borrow::Cow;
+use std::error::Error as StdError;
+use std::fmt;
+
+use reqwest::{Response, StatusCode, header};
 use serde::Deserialize;
 use tracing::Instrument;
 
-use crate::error::{BackendDetail, Error, Result};
+use crate::error::{Error, ErrorKind, Result};
 use crate::stream;
 
+/// Classifies a backend HTTP error status into a semantic [`ErrorKind`].
+fn status_to_kind(status: StatusCode) -> ErrorKind {
+    match status {
+        StatusCode::TOO_MANY_REQUESTS => ErrorKind::BackendRateLimited,
+        StatusCode::REQUEST_TIMEOUT | StatusCode::GATEWAY_TIMEOUT => ErrorKind::BackendTimeout,
+        StatusCode::INTERNAL_SERVER_ERROR
+        | StatusCode::BAD_GATEWAY
+        | StatusCode::SERVICE_UNAVAILABLE => ErrorKind::BackendUnavailable,
+        _ => ErrorKind::BackendFailure,
+    }
+}
+
 /// Extension trait that sends a request inside a tracing span.
 pub trait SendTraced {
     /// Sends the request, wrapping it in a span that covers the full request
@@ -77,8 +93,8 @@ struct XmlApiError {
 ///
 /// Use [`check_error`](Self::check_error) instead of
 /// [`error_for_status`](reqwest::Response::error_for_status) to avoid losing the response body on
-/// 4xx/5xx errors. The method parses the structured error body (JSON or XML) and returns an
-/// [`Error::BackendResponse`] with the extracted error code and message.
+/// 4xx/5xx errors. The method parses the structured error body (JSON or XML) and returns a
+/// semantic backend service error with the extracted error code and message.
 ///
 /// Implemented for both [`reqwest::Response`] and `Result<Response, reqwest::Error>` so it can be
 /// chained directly.
@@ -90,8 +106,8 @@ pub trait ResponseExt {
     /// For other error statuses (e.g., redirects), falls back to
     /// [`reqwest::Response::error_for_status`].
     ///
-    /// When called on `Result<Response, reqwest::Error>`, transport errors are
-    /// wrapped as [`Error::Reqwest`] with the same context string.
+    /// When called on `Result<Response, reqwest::Error>`, transport errors are semantically
+    /// classified and given the same context string.
     async fn check_error(self, context: &'static str) -> Result<Response>;
 
     /// Drains the response body of a response we are otherwise done with.
@@ -124,14 +140,10 @@ impl ResponseExt for Response {
                 return Ok(self);
             };
             self.drain_body().await;
-            return Err(Error::reqwest(context, e));
+            return Err(Error::with_context(status_to_kind(status), context, e));
         };
 
-        Err(Error::BackendResponse {
-            context,
-            status,
-            detail,
-        })
+        Err(BackendResponseError::new(context, status, detail).into())
     }
 
     async fn drain_body(mut self) {
@@ -141,13 +153,7 @@ impl ResponseExt for Response {
 
 impl ResponseExt for Result<Response, reqwest::Error> {
     async fn check_error(self, context: &'static str) -> Result<Response> {
-        match self {
-            Ok(resp) => resp.check_error(context).await,
-            Err(e) => Err(match stream::unpack_client_error(&e) {
-                Some(ce) => Error::Client(ce),
-                None => Error::reqwest(context, e),
-            }),
-        }
+        self.reqwest_context(context)?.check_error(context).await
     }
 
     async fn drain_body(self) {
@@ -157,6 +163,32 @@ impl ResponseExt for Result<Response, reqwest::Error> {
     }
 }
 
+pub trait ReqwestResultExt<T> {
+    fn reqwest_context(self, context: &'static str) -> Result<T>;
+}
+
+impl<T> ReqwestResultExt<T> for Result<T, reqwest::Error> {
+    fn reqwest_context(self, context: &'static str) -> Result<T> {
+        self.map_err(|error| {
+            if let Some(ce) = stream::unpack_client_error(&error) {
+                return Error::with_context(ErrorKind::ClientStream, context, ce);
+            }
+
+            let kind = if error.is_timeout() {
+                ErrorKind::BackendTimeout
+            } else if error.is_connect() || error.is_request() {
+                ErrorKind::BackendUnavailable
+            } else if error.is_decode() {
+                ErrorKind::CorruptData
+            } else {
+                ErrorKind::BackendFailure
+            };
+
+            Error::with_context(kind, context, error)
+        })
+    }
+}
+
 async fn parse_json_error(resp: Response) -> BackendDetail {
     match resp.json().await {
         Ok(JsonApiError { error }) => {
@@ -185,3 +217,202 @@ async fn parse_xml_error(resp: Response) -> BackendDetail {
         BackendDetail::none()
     }
 }
+
+/// Structured error detail parsed from a backend HTTP error response.
+///
+/// Formats conditionally: includes only the fields that are non-empty.
+#[derive(Debug)]
+struct BackendDetail {
+    /// Machine-readable error code (e.g., "InvalidArgument", "NoSuchKey").
+    code: String,
+    /// Human-readable error message from the response body.
+    message: String,
+}
+
+impl BackendDetail {
+    /// Creates a new [`BackendDetail`] with empty code and message.
+    fn none() -> Self {
+        Self {
+            code: String::new(),
+            message: String::new(),
+        }
+    }
+
+    fn is_empty(&self) -> bool {
+        self.code.is_empty() && self.message.is_empty()
+    }
+}
+
+impl fmt::Display for BackendDetail {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match (self.code.is_empty(), self.message.is_empty()) {
+            (false, false) => write!(f, "{} (backend code {})", self.message, self.code),
+            (true, false) => f.write_str(&self.message),
+            (false, true) => write!(f, "backend code {}", self.code),
+            (true, true) => Ok(()),
+        }
+    }
+}
+
+/// An HTTP error response received from a storage backend such as GCS or S3.
+///
+/// Unlike [`reqwest::Error`], which covers transport-level failures, this type captures an
+/// application-level error response where the backend returned a 4xx or 5xx status together with
+/// a structured response body. It retains the request context, HTTP status, and parsed backend
+/// error code and message.
+#[derive(Debug)]
+struct BackendResponseError {
+    context: Cow<'static, str>,
+    status: StatusCode,
+    detail: BackendDetail,
+}
+
+impl BackendResponseError {
+    /// Creates a backend response error from its operation context, status, and detail.
+    pub fn new(
+        context: impl Into<Cow<'static, str>>,
+        status: StatusCode,
+        detail: BackendDetail,
+    ) -> Self {
+        Self {
+            context: context.into(),
+            status,
+            detail,
+        }
+    }
+}
+
+impl fmt::Display for BackendResponseError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{} ({})", self.context, self.status)?;
+        if !self.detail.is_empty() {
+            write!(f, ". {}", self.detail)?;
+        }
+        Ok(())
+    }
+}
+
+impl StdError for BackendResponseError {}
+
+impl From<BackendResponseError> for Error {
+    fn from(source: BackendResponseError) -> Self {
+        let kind = status_to_kind(source.status);
+        let context = source.context.clone();
+        Self::with_context(kind, context, source)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::error::Error as _;
+    use std::io::{Read, Write};
+    use std::net::TcpListener;
+    use std::sync::mpsc;
+    use std::thread;
+    use std::time::Duration;
+
+    use reqwest::StatusCode;
+
+    use super::{BackendDetail, BackendResponseError, status_to_kind};
+    use crate::backend::extensions::ReqwestResultExt;
+    use crate::error::{Error, ErrorKind};
+
+    #[test]
+    fn backend_response_preserves_status_and_structured_source() {
+        let error: Error = BackendResponseError::new(
+            "getting a GCS object",
+            StatusCode::TOO_MANY_REQUESTS,
+            BackendDetail {
+                code: "rateLimitExceeded".to_owned(),
+                message: "too many requests".to_owned(),
+            },
+        )
+        .into();
+
+        assert_eq!(error.kind(), ErrorKind::BackendRateLimited);
+        assert_eq!(
+            error.to_string(),
+            "backend rate limited: getting a GCS object"
+        );
+        assert_eq!(
+            error.source().unwrap().to_string(),
+            "getting a GCS object (429 Too Many Requests). too many requests (backend code rateLimitExceeded)"
+        );
+    }
+
+    #[test]
+    fn backend_response_omits_separator_without_detail() {
+        let error = BackendResponseError::new(
+            "getting a GCS object",
+            StatusCode::INTERNAL_SERVER_ERROR,
+            BackendDetail::none(),
+        );
+
+        assert_eq!(
+            error.to_string(),
+            "getting a GCS object (500 Internal Server Error)"
+        );
+    }
+
+    #[test]
+    fn retryable_http_statuses_have_transient_semantic_kinds() {
+        for status in [
+            StatusCode::REQUEST_TIMEOUT,
+            StatusCode::TOO_MANY_REQUESTS,
+            StatusCode::INTERNAL_SERVER_ERROR,
+            StatusCode::BAD_GATEWAY,
+            StatusCode::SERVICE_UNAVAILABLE,
+            StatusCode::GATEWAY_TIMEOUT,
+        ] {
+            assert!(matches!(
+                status_to_kind(status),
+                ErrorKind::BackendRateLimited
+                    | ErrorKind::BackendTimeout
+                    | ErrorKind::BackendUnavailable
+            ));
+        }
+    }
+
+    #[tokio::test]
+    async fn response_body_timeout_is_classified_as_backend_timeout() {
+        let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
+        let address = listener.local_addr().unwrap();
+        let (release_tx, release_rx) = mpsc::channel();
+        let server = thread::spawn(move || {
+            let (mut connection, _) = listener.accept().unwrap();
+            let mut request = [0];
+            connection.read_exact(&mut request).unwrap();
+            connection
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\nConnection: close\r\n\r\n")
+                .unwrap();
+
+            // Keep the body open without sending its promised byte until the client times out.
+            let _ = release_rx.recv_timeout(Duration::from_secs(5));
+        });
+
+        let client = reqwest::Client::builder()
+            .read_timeout(Duration::from_millis(100))
+            .build()
+            .unwrap();
+        let response = client
+            .get(format!("http://{address}"))
+            .send()
+            .await
+            .unwrap();
+
+        let result = response.bytes().await;
+        assert!(result.as_ref().unwrap_err().is_timeout());
+
+        let error = result
+            .reqwest_context("reading backend response body")
+            .unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::BackendTimeout);
+        assert_eq!(
+            error.to_string(),
+            "backend timed out: reading backend response body"
+        );
+
+        release_tx.send(()).unwrap();
+        server.join().unwrap();
+    }
+}
diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs
index 5204dca0..565c888b 100644
--- a/objectstore-service/src/backend/gcs.rs
+++ b/objectstore-service/src/backend/gcs.rs
@@ -7,7 +7,6 @@ use std::sync::Arc;
 use std::time::SystemTime;
 use std::{fmt, io};
 
-use anyhow::Context;
 use futures_util::{StreamExt, TryStreamExt};
 use gcp_auth::TokenProvider;
 use objectstore_types::headers;
@@ -17,15 +16,15 @@ use reqwest::header::HeaderName;
 use reqwest::{Body, IntoUrl, Method, RequestBuilder, StatusCode, Url, header, multipart};
 use serde::{Deserialize, Serialize};
 
-use super::extensions::{ResponseExt, SendTraced};
 use crate::backend::common::{
     self, Backend, DeleteResponse, GetResponse, MetadataResponse, MultipartUploadBackend,
     PutResponse,
 };
+use crate::backend::extensions::{ReqwestResultExt, ResponseExt, SendTraced};
 use crate::change_stream::{
     ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream,
 };
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Result, ResultExt as _};
 use crate::gcp_auth::PrefetchingTokenProvider;
 use crate::id::ObjectId;
 use crate::multipart::{
@@ -258,44 +257,47 @@ impl GcsObject {
             .metadata
             .remove(&GcsMetaKey::Expiration)
             .map(|s| s.parse())
-            .transpose()?
+            .transpose()
+            .context(ErrorKind::CorruptData, "decoding GCS expiration policy")?
             .unwrap_or_default();
 
         let origin = self
             .metadata
             .remove(&GcsMetaKey::Origin)
-            .map(|value| decode_gcs_meta_value(&value))
+            .map(|value| decode_gcs_meta_value(&value, "decoding GCS origin metadata"))
             .transpose()?;
         let filename = self
             .metadata
             .remove(&GcsMetaKey::Filename)
-            .map(|value| decode_gcs_meta_value(&value))
+            .map(|value| decode_gcs_meta_value(&value, "decoding GCS filename metadata"))
             .transpose()?;
 
         let content_type = self.content_type;
-        let compression = self.content_encoding.map(|s| s.parse()).transpose()?;
+        let compression = self
+            .content_encoding
+            .map(|s| s.parse())
+            .transpose()
+            .context(ErrorKind::CorruptData, "decoding GCS compression")?;
         let size = self
             .size
             .map(|size| size.parse())
             .transpose()
-            .map_err(|e| Error::Generic {
-                context: "GCS: failed to parse size from object metadata".to_string(),
-                cause: Some(Box::new(e)),
-            })?;
+            .context(ErrorKind::CorruptData, "decoding GCS object size")?;
         let time_created = self.time_created;
 
         // At this point, all built-in metadata should have been removed from self.metadata.
         let mut custom = BTreeMap::new();
         for (key, value) in self.metadata {
             if let GcsMetaKey::Custom(custom_key) = key {
-                custom.insert(custom_key, decode_gcs_meta_value(&value)?);
+                custom.insert(
+                    custom_key,
+                    decode_gcs_meta_value(&value, "decoding GCS custom metadata")?,
+                );
             } else {
-                return Err(Error::Generic {
-                    context: format!(
-                        "GCS: unexpected built-in metadata key in object metadata: {key}"
-                    ),
-                    cause: None,
-                });
+                return Err(Error::new(
+                    ErrorKind::CorruptData,
+                    format!("unexpected GCS metadata key: {key}"),
+                ));
             }
         }
 
@@ -388,10 +390,10 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result<header::HeaderMap> {
         let formatted = humantime::format_rfc3339_seconds(custom_time);
         headers.insert(
             HeaderName::from_static("x-goog-custom-time"),
-            formatted.to_string().parse().map_err(|e| Error::Generic {
-                context: "GCS: invalid custom-time header value".into(),
-                cause: Some(Box::new(e)),
-            })?,
+            formatted
+                .to_string()
+                .parse()
+                .context(ErrorKind::Internal, "encoding GCS custom-time header")?,
         );
     }
 
@@ -401,10 +403,7 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result<header::HeaderMap> {
             compression
                 .to_string()
                 .parse()
-                .map_err(|e| Error::Generic {
-                    context: "GCS: invalid content-encoding header value".into(),
-                    cause: Some(Box::new(e)),
-                })?,
+                .context(ErrorKind::Internal, "encoding GCS content-encoding header")?,
         );
     }
 
@@ -432,11 +431,8 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result<header::HeaderMap> {
 }
 
 /// Decodes a stored GCS metadata value into its logical string.
-fn decode_gcs_meta_value(value: &str) -> Result<String> {
-    headers::decode_header_str(value).map_err(|cause| Error::Generic {
-        context: "GCS: invalid percent-encoded UTF-8 in object metadata".to_owned(),
-        cause: Some(Box::new(cause)),
-    })
+fn decode_gcs_meta_value(value: &str, context: &'static str) -> Result<String> {
+    headers::decode_header_str(value).context(ErrorKind::CorruptData, context)
 }
 
 /// Inserts a single `x-goog-meta-*` header, escaping the value for transport.
@@ -456,39 +452,20 @@ fn insert_gcs_meta_header(
 ) -> Result<()> {
     let header_name = format!("x-goog-meta-{key}");
     headers.insert(
-        HeaderName::try_from(&header_name).map_err(|e| Error::Generic {
-            context: format!("GCS: invalid header name: {header_name}"),
-            cause: Some(Box::new(e)),
-        })?,
+        HeaderName::try_from(&header_name).context(
+            ErrorKind::Internal,
+            format!("encoding GCS metadata header {header_name}"),
+        )?,
         headers::encode_header_value(value),
     );
     Ok(())
 }
 
-/// Returns `true` if the error is a transient reqwest failure worth retrying.
+/// Returns `true` if the error is a transient backend failure worth retrying.
 fn error_is_retryable(error: &Error) -> bool {
-    match error {
-        Error::Reqwest { cause, .. } => {
-            cause.is_timeout()
-                || cause.is_connect()
-                || cause.is_request()
-                || cause.status().is_some_and(status_is_retryable)
-        }
-        Error::BackendResponse { status, .. } => status_is_retryable(*status),
-        _ => false,
-    }
-}
-
-fn status_is_retryable(status: StatusCode) -> bool {
-    // https://docs.cloud.google.com/storage/docs/json_api/v1/status-codes
     matches!(
-        status,
-        StatusCode::REQUEST_TIMEOUT
-            | StatusCode::TOO_MANY_REQUESTS
-            | StatusCode::INTERNAL_SERVER_ERROR
-            | StatusCode::BAD_GATEWAY
-            | StatusCode::SERVICE_UNAVAILABLE
-            | StatusCode::GATEWAY_TIMEOUT
+        error.kind(),
+        ErrorKind::BackendRateLimited | ErrorKind::BackendTimeout | ErrorKind::BackendUnavailable
     )
 }
 
@@ -522,7 +499,9 @@ impl GcsBackend {
 
         Ok(Self {
             client: common::reqwest_client(),
-            endpoint: endpoint_str.parse().context("invalid GCS endpoint URL")?,
+            endpoint: endpoint_str
+                .parse()
+                .map_err(|e| anyhow::Error::new(e).context("invalid GCS endpoint URL"))?,
             bucket,
             token_provider,
             change_stream,
@@ -535,12 +514,11 @@ impl GcsBackend {
 
         let path = id.as_storage_path().to_string();
         url.path_segments_mut()
-            .map_err(|()| Error::Generic {
-                context: format!(
-                    "GCS: invalid endpoint URL, {} cannot be a base",
-                    self.endpoint
-                ),
-                cause: None,
+            .map_err(|()| {
+                Error::new(
+                    ErrorKind::Internal,
+                    format!("building GCS object URL from {}", self.endpoint),
+                )
             })?
             .extend(&["storage", "v1", "b", &self.bucket, "o", &path]);
 
@@ -552,12 +530,11 @@ impl GcsBackend {
         let mut url = self.endpoint.clone();
 
         url.path_segments_mut()
-            .map_err(|()| Error::Generic {
-                context: format!(
-                    "GCS: invalid endpoint URL, {} cannot be a base",
-                    self.endpoint
-                ),
-                cause: None,
+            .map_err(|()| {
+                Error::new(
+                    ErrorKind::Internal,
+                    format!("building GCS object URL from {}", self.endpoint),
+                )
             })?
             .extend(&["upload", "storage", "v1", "b", &self.bucket, "o"]);
 
@@ -577,12 +554,11 @@ impl GcsBackend {
     fn xml_object_url(&self, id: &ObjectId) -> Result<Url> {
         let mut url = self.endpoint.clone();
         {
-            let mut segments = url.path_segments_mut().map_err(|()| Error::Generic {
-                context: format!(
-                    "GCS: invalid endpoint URL, {} cannot be a base",
-                    self.endpoint
-                ),
-                cause: None,
+            let mut segments = url.path_segments_mut().map_err(|()| {
+                Error::new(
+                    ErrorKind::Internal,
+                    format!("building GCS object URL from {}", self.endpoint),
+                )
             })?;
             segments.push(&self.bucket);
             for part in id.as_storage_path().to_string().split('/') {
@@ -596,7 +572,10 @@ impl GcsBackend {
     async fn request(&self, method: Method, url: impl IntoUrl) -> Result<RequestBuilder> {
         let mut builder = self.client.request(method, url);
         if let Some(provider) = &self.token_provider {
-            let token = provider.token(TOKEN_SCOPES).await?;
+            let token = provider.token(TOKEN_SCOPES).await.context(
+                ErrorKind::BackendFailure,
+                "getting GCS authentication token",
+            )?;
             builder = builder.bearer_auth(token.as_str());
         }
         Ok(builder)
@@ -642,7 +621,7 @@ impl GcsBackend {
                     .await?
                     .send_traced()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: get metadata request", e))?;
+                    .reqwest_context("getting GCS object metadata")?;
 
                 if resp.status() == StatusCode::NOT_FOUND {
                     resp.drain_body().await;
@@ -650,11 +629,11 @@ impl GcsBackend {
                 }
 
                 let metadata: GcsObject = resp
-                    .check_error("GCS: get metadata status")
+                    .check_error("getting GCS object metadata")
                     .await?
                     .json()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: get metadata parse", e))?;
+                    .reqwest_context("getting GCS object metadata")?;
 
                 Ok(Some(metadata))
             })
@@ -725,27 +704,28 @@ impl GcsBackend {
             .append_pair("ifMetagenerationMatch", metageneration);
 
         self.with_retry("update_custom_time", || async {
-            match self
+            let response = self
                 .request(Method::PATCH, object_url.clone())
                 .await?
                 .json(&CustomTimeRequest { custom_time })
                 .send_traced()
                 .await
-                .check_error("GCS: update custom time")
-                .await
-            {
-                Ok(response) => {
-                    response.drain_body().await;
-                    Ok(true)
-                }
-                // Bumping TTI is opportunistic. A concurrent metadata writer won the CAS race,
-                // so leave its update intact and let a future read evaluate the TTI again.
-                Err(Error::BackendResponse {
-                    status: StatusCode::PRECONDITION_FAILED,
-                    ..
-                }) => Ok(false),
-                Err(error) => Err(error),
+                .reqwest_context("updating GCS custom time")?;
+
+            // Bumping TTI is opportunistic. A concurrent metadata writer won the CAS race, so
+            // leave its update intact and let a future read evaluate the TTI again.
+            if response.status() == StatusCode::PRECONDITION_FAILED {
+                response.drain_body().await;
+                return Ok(false);
             }
+
+            response
+                .check_error("updating GCS custom time")
+                .await?
+                .drain_body()
+                .await;
+
+            Ok(true)
         })
         .await
     }
@@ -782,10 +762,8 @@ impl Backend for GcsBackend {
 
         // NB: Ensure the order of these fields and that a content-type is attached to them. Both
         // are required by the GCS API.
-        let metadata_json = serde_json::to_string(&gcs_metadata).map_err(|cause| Error::Serde {
-            context: "failed to serialize metadata for GCS upload".to_string(),
-            cause,
-        })?;
+        let metadata_json = serde_json::to_string(&gcs_metadata)
+            .context(ErrorKind::Internal, "encoding GCS upload metadata")?;
 
         let multipart = multipart::Form::new()
             .part(
@@ -798,10 +776,7 @@ impl Backend for GcsBackend {
                 "media",
                 multipart::Part::stream(Body::wrap_stream(stream.boxed()))
                     .mime_str(&metadata.content_type)
-                    .map_err(|e| Error::Generic {
-                        context: format!("invalid mime type: {}", metadata.content_type),
-                        cause: Some(Box::new(e)),
-                    })?,
+                    .context(ErrorKind::InvalidMetadata, "encoding GCS content type")?,
             );
 
         // GCS requires a multipart/related request. Its body looks identical to
@@ -816,7 +791,7 @@ impl Backend for GcsBackend {
             .header(header::CONTENT_TYPE, content_type)
             .send_traced()
             .await
-            .check_error("GCS: upload object")
+            .check_error("uploading a GCS object")
             .await?;
 
         let stored_size = read_stored_content_length(response).await;
@@ -852,10 +827,11 @@ impl Backend for GcsBackend {
                 if let Some(r) = range {
                     req = req.header(header::RANGE, r.to_header_value());
                 }
+
                 let resp = req
                     .send_traced()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: get payload", e))?;
+                    .reqwest_context("getting a GCS object payload")?;
 
                 if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE {
                     let raw = resp
@@ -864,16 +840,16 @@ impl Backend for GcsBackend {
                         .and_then(|v| v.to_str().ok());
                     let total = raw.and_then(ContentRange::parse_unsatisfiable_total);
                     let err = match total {
-                        Some(total) => Error::RangeNotSatisfiable { total },
-                        None => Error::generic(format!(
-                            "GCS: 416 response with invalid Content-Range: {raw:?}"
-                        )),
+                        Some(total) => ErrorKind::RangeNotSatisfiable { total }.into(),
+                        None => {
+                            Error::new(ErrorKind::BackendFailure, "invalid GCS 416 Content-Range")
+                        }
                     };
                     resp.drain_body().await;
                     return Err(err);
                 }
 
-                resp.check_error("GCS: get payload").await
+                resp.check_error("getting a GCS object payload").await
             })
             .await?;
 
@@ -884,9 +860,8 @@ impl Backend for GcsBackend {
                     .get(header::CONTENT_RANGE)
                     .and_then(|v| v.to_str().ok())
                     .and_then(|s| s.parse::<ContentRange>().ok())
-                    .ok_or_else(|| Error::Generic {
-                        context: "GCS: 206 response missing valid Content-Range header".to_owned(),
-                        cause: None,
+                    .ok_or_else(|| {
+                        Error::new(ErrorKind::BackendFailure, "missing GCS 206 Content-Range")
                     })?,
             )
         } else {
@@ -920,7 +895,7 @@ impl Backend for GcsBackend {
                     .await?
                     .send_traced()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: delete object", e))?;
+                    .reqwest_context("deleting a GCS object")?;
 
                 // Do not error for objects that do not exist
                 if resp.status() == StatusCode::NOT_FOUND {
@@ -928,7 +903,7 @@ impl Backend for GcsBackend {
                     return Ok(false);
                 }
 
-                resp.check_error("GCS: delete object")
+                resp.check_error("deleting a GCS object")
                     .await?
                     .drain_body()
                     .await;
@@ -1071,10 +1046,10 @@ impl MultipartUploadBackend for GcsBackend {
         let mut headers = metadata_to_gcs_headers(metadata)?;
         headers.insert(
             header::CONTENT_TYPE,
-            metadata.content_type.parse().map_err(|e| Error::Generic {
-                context: "GCS: invalid content-type header value".into(),
-                cause: Some(Box::new(e)),
-            })?,
+            metadata
+                .content_type
+                .parse()
+                .context(ErrorKind::InvalidMetadata, "encoding GCS content type")?,
         );
         headers.insert(
             header::CONTENT_LENGTH,
@@ -1091,19 +1066,19 @@ impl MultipartUploadBackend for GcsBackend {
                     .headers(headers)
                     .send_traced()
                     .await
-                    .check_error("GCS: initiate multipart upload")
+                    .check_error("initiating a GCS multipart upload")
                     .await?;
 
                 let body = resp
                     .bytes()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?;
+                    .reqwest_context("reading GCS initiate-multipart response")?;
 
                 let xml: XmlInitiateMultipartUploadResponse =
-                    quick_xml::de::from_reader(body.as_ref()).map_err(|e| Error::Generic {
-                        context: "GCS: failed to parse initiate multipart response".to_owned(),
-                        cause: Some(Box::new(e)),
-                    })?;
+                    quick_xml::de::from_reader(body.as_ref()).context(
+                        ErrorKind::CorruptData,
+                        "decoding GCS initiate-multipart response",
+                    )?;
 
                 xml.try_into()
             }
@@ -1140,7 +1115,7 @@ impl MultipartUploadBackend for GcsBackend {
         let resp = builder
             .send_traced()
             .await
-            .check_error("GCS: upload part")
+            .check_error("uploading a GCS multipart part")
             .await?;
 
         let etag = resp
@@ -1148,7 +1123,12 @@ impl MultipartUploadBackend for GcsBackend {
             .get(header::ETAG)
             .and_then(|v| v.to_str().ok())
             .map(|s| s.to_owned())
-            .ok_or_else(|| Error::generic("GCS: upload part response missing ETag header"))?;
+            .ok_or_else(|| {
+                Error::new(
+                    ErrorKind::BackendFailure,
+                    "GCS upload-part response missing ETag",
+                )
+            })?;
 
         resp.drain_body().await;
 
@@ -1184,19 +1164,16 @@ impl MultipartUploadBackend for GcsBackend {
                     .await?
                     .send_traced()
                     .await
-                    .check_error("GCS: list parts")
+                    .check_error("listing GCS multipart parts")
                     .await?;
 
                 let body = resp
                     .bytes()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: read list parts body", e))?;
+                    .reqwest_context("reading GCS list-parts response")?;
 
-                let xml: XmlListPartsResponse =
-                    quick_xml::de::from_reader(body.as_ref()).map_err(|e| Error::Generic {
-                        context: "GCS: failed to parse list parts response".to_owned(),
-                        cause: Some(Box::new(e)),
-                    })?;
+                let xml: XmlListPartsResponse = quick_xml::de::from_reader(body.as_ref())
+                    .context(ErrorKind::CorruptData, "decoding GCS list-parts response")?;
 
                 Ok(xml.into())
             }
@@ -1222,13 +1199,13 @@ impl MultipartUploadBackend for GcsBackend {
                     .await?
                     .send_traced()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: abort multipart upload", e))?;
+                    .reqwest_context("aborting a GCS multipart upload")?;
 
                 // XXX: real S3 would return 404 here if the upload has been recently completed and we
                 // would have to handle it. It turns out GCS returns 204 instead, so we don't need to
                 // handle that case.
 
-                resp.check_error("GCS: abort multipart upload")
+                resp.check_error("aborting a GCS multipart upload")
                     .await?
                     .drain_body()
                     .await;
@@ -1251,10 +1228,10 @@ impl MultipartUploadBackend for GcsBackend {
         url.query_pairs_mut().append_pair("uploadId", upload_id);
 
         let body = XmlCompleteMultipartUpload::from(parts);
-        let xml = quick_xml::se::to_string(&body).map_err(|e| Error::Generic {
-            context: "GCS: failed to serialize complete multipart request".into(),
-            cause: Some(Box::new(e)),
-        })?;
+        let xml = quick_xml::se::to_string(&body).context(
+            ErrorKind::Internal,
+            "encoding GCS complete-multipart request",
+        )?;
 
         self.with_retry("complete_multipart", || {
             let url = url.clone();
@@ -1267,7 +1244,7 @@ impl MultipartUploadBackend for GcsBackend {
                     .body(xml)
                     .send_traced()
                     .await
-                    .check_error("GCS: complete multipart upload")
+                    .check_error("completing a GCS multipart upload")
                     .await?;
 
                 // XXX: real S3 would return 404 here if the upload has been recently completed and we
@@ -1277,7 +1254,7 @@ impl MultipartUploadBackend for GcsBackend {
                 let body = resp
                     .bytes()
                     .await
-                    .map_err(|e| Error::reqwest("GCS: read complete multipart body", e))?;
+                    .reqwest_context("reading GCS complete-multipart response")?;
 
                 let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref())
                     .ok()
@@ -1359,11 +1336,14 @@ mod tests {
             .await?
             .send_traced()
             .await
-            .check_error("GCS: get metadata request")
+            .check_error("getting GCS object metadata")
             .await?
             .json::<GcsObject>()
             .await
-            .map_err(|e| Error::reqwest("GCS: get metadata parse", e))
+            .context(
+                ErrorKind::BackendFailure,
+                "decoding GCS object metadata response",
+            )
             .map(|object| (object.generation, object.metageneration))?)
     }
 
diff --git a/objectstore-service/src/backend/in_memory.rs b/objectstore-service/src/backend/in_memory.rs
index 93e74cfc..e3635886 100644
--- a/objectstore-service/src/backend/in_memory.rs
+++ b/objectstore-service/src/backend/in_memory.rs
@@ -19,7 +19,7 @@ use super::common::{
     DeleteResponse, GetResponse, HighVolumeBackend, MultipartUploadBackend, PutResponse, TieredGet,
     TieredMetadata, TieredWrite, Tombstone,
 };
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Result};
 use crate::id::ObjectId;
 use crate::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
@@ -128,7 +128,10 @@ impl super::common::Backend for InMemoryBackend {
         let entry = self.store.lock().unwrap().get(id).cloned();
         match entry {
             None => Ok(None),
-            Some(StoreEntry::Tombstone(_)) => Err(Error::UnexpectedTombstone),
+            Some(StoreEntry::Tombstone(_)) => Err(Error::new(
+                ErrorKind::Internal,
+                "unexpected in-memory tombstone",
+            )),
             Some(StoreEntry::Object(mut metadata, bytes)) => {
                 let total = bytes.len() as u64;
                 metadata.size = Some(bytes.len());
@@ -136,7 +139,7 @@ impl super::common::Backend for InMemoryBackend {
                     Some(range) => {
                         let content_range = range
                             .resolve(total)
-                            .ok_or(Error::RangeNotSatisfiable { total })?;
+                            .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
                         let sliced =
                             bytes.slice(content_range.start as usize..=content_range.end as usize);
                         (Some(content_range), sliced)
@@ -193,7 +196,7 @@ impl HighVolumeBackend for InMemoryBackend {
                     Some(range) => {
                         let content_range = range
                             .resolve(total)
-                            .ok_or(Error::RangeNotSatisfiable { total })?;
+                            .ok_or(ErrorKind::RangeNotSatisfiable { total })?;
                         let sliced =
                             bytes.slice(content_range.start as usize..=content_range.end as usize);
                         (Some(content_range), sliced)
@@ -289,7 +292,12 @@ impl MultipartUploadBackend for InMemoryBackend {
         let mut store = self.multipart_store.lock().unwrap();
         let upload = store
             .get_mut(&(id.clone(), upload_id.clone()))
-            .ok_or_else(|| Error::generic("multipart upload not found"))?;
+            .ok_or_else(|| {
+                Error::new(
+                    ErrorKind::BackendFailure,
+                    "in-memory multipart upload not found",
+                )
+            })?;
 
         upload.parts.insert(
             part_number,
@@ -311,9 +319,12 @@ impl MultipartUploadBackend for InMemoryBackend {
         part_number_marker: Option<PartNumber>,
     ) -> Result<ListPartsResponse> {
         let store = self.multipart_store.lock().unwrap();
-        let upload = store
-            .get(&(id.clone(), upload_id.clone()))
-            .ok_or_else(|| Error::generic("multipart upload not found"))?;
+        let upload = store.get(&(id.clone(), upload_id.clone())).ok_or_else(|| {
+            Error::new(
+                ErrorKind::BackendFailure,
+                "in-memory multipart upload not found",
+            )
+        })?;
 
         let iter = upload
             .parts
@@ -376,9 +387,12 @@ impl MultipartUploadBackend for InMemoryBackend {
         // the client can retry.
         let assembled = {
             let store = self.multipart_store.lock().unwrap();
-            let upload = store
-                .get(&key)
-                .ok_or_else(|| Error::generic("multipart upload not found"))?;
+            let upload = store.get(&key).ok_or_else(|| {
+                Error::new(
+                    ErrorKind::BackendFailure,
+                    "in-memory multipart upload not found",
+                )
+            })?;
 
             for completed in &parts {
                 match upload.parts.get(&completed.part_number) {
diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs
index af0b233d..f24bcad2 100644
--- a/objectstore-service/src/backend/local_fs.rs
+++ b/objectstore-service/src/backend/local_fs.rs
@@ -1,6 +1,6 @@
 //! Local filesystem backend for development and testing.
 
-use std::io::ErrorKind;
+use std::io;
 use std::path::PathBuf;
 use std::pin::pin;
 use std::time::SystemTime;
@@ -15,7 +15,7 @@ use tokio_util::io::{ReaderStream, StreamReader};
 use crate::backend::common::{
     Backend, DeleteResponse, GetResponse, MultipartUploadBackend, PutResponse,
 };
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Result, ResultExt as _};
 use crate::id::ObjectId;
 use crate::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
@@ -85,34 +85,56 @@ impl Backend for LocalFsBackend {
     ) -> Result<PutResponse> {
         let path = self.path.join(id.as_storage_path().to_string());
         objectstore_log::debug!(path=%path.display(), "Writing to local_fs backend");
-        tokio::fs::create_dir_all(path.parent().unwrap()).await?;
+        tokio::fs::create_dir_all(path.parent().unwrap())
+            .await
+            .context(
+                ErrorKind::BackendFailure,
+                "creating local-fs object directory",
+            )?;
         let file = OpenOptions::new()
             .create(true)
             .write(true)
             .truncate(true)
             .open(path)
-            .await?;
+            .await
+            .context(
+                ErrorKind::BackendFailure,
+                "opening local-fs object for writing",
+            )?;
 
         let mut reader = pin!(StreamReader::new(stream));
         let mut writer = BufWriter::new(file);
 
-        let metadata_json = serde_json::to_string(metadata).map_err(|cause| Error::Serde {
-            context: "failed to serialize metadata".to_string(),
-            cause,
-        })?;
-        writer.write_all(metadata_json.as_bytes()).await?;
-        writer.write_all(b"\n").await?;
+        let metadata_json = serde_json::to_string(metadata)
+            .context(ErrorKind::Internal, "encoding local-fs object metadata")?;
+        writer.write_all(metadata_json.as_bytes()).await.context(
+            ErrorKind::BackendFailure,
+            "writing local-fs object metadata",
+        )?;
+        writer.write_all(b"\n").await.context(
+            ErrorKind::BackendFailure,
+            "writing local-fs object metadata",
+        )?;
 
         tokio::io::copy(&mut reader, &mut writer)
             .await
             .map_err(|e| match stream::unpack_client_error(&e) {
-                Some(ce) => Error::Client(ce),
-                None => e.into(),
+                Some(ce) => Error::from(ce),
+                None => Error::with_context(
+                    ErrorKind::BackendFailure,
+                    "writing local-fs object payload",
+                    e,
+                ),
             })?;
 
-        writer.flush().await?;
+        writer
+            .flush()
+            .await
+            .context(ErrorKind::BackendFailure, "flushing local-fs object")?;
         let file = writer.into_inner();
-        file.sync_data().await?;
+        file.sync_data()
+            .await
+            .context(ErrorKind::BackendFailure, "syncing local-fs object")?;
         drop(file);
 
         Ok(())
@@ -125,25 +147,35 @@ impl Backend for LocalFsBackend {
         let path = self.path.join(id.as_storage_path().to_string());
         let file = match OpenOptions::new().read(true).open(path).await {
             Ok(file) => file,
-            Err(err) if err.kind() == ErrorKind::NotFound => {
+            Err(err) if err.kind() == io::ErrorKind::NotFound => {
                 objectstore_log::debug!("Object not found");
                 return Ok(None);
             }
-            err => err?,
+            err => err.context(
+                ErrorKind::BackendFailure,
+                "opening local-fs object for reading",
+            )?,
         };
 
         let mut reader = BufReader::new(file);
         let mut metadata_line = String::new();
-        reader.read_line(&mut metadata_line).await?;
-        let file_len = reader.get_ref().metadata().await?.len();
-        let mut metadata: Metadata =
-            serde_json::from_str(metadata_line.trim_end()).map_err(|cause| Error::Serde {
-                context: "failed to deserialize metadata".to_string(),
-                cause,
-            })?;
+        reader.read_line(&mut metadata_line).await.context(
+            ErrorKind::BackendFailure,
+            "reading local-fs object metadata",
+        )?;
+        let file_len = reader
+            .get_ref()
+            .metadata()
+            .await
+            .context(ErrorKind::BackendFailure, "reading local-fs object size")?
+            .len();
+        let mut metadata: Metadata = serde_json::from_str(metadata_line.trim_end())
+            .context(ErrorKind::CorruptData, "decoding local-fs object metadata")?;
         let payload_size = file_len
             .checked_sub(metadata_line.len() as u64)
-            .ok_or_else(|| Error::generic("local-fs file corrupted: shorter than header"))?;
+            .ok_or_else(|| {
+                Error::new(ErrorKind::CorruptData, "reading truncated local-fs object")
+            })?;
         metadata.size = Some(payload_size as usize);
 
         let (content_range, stream) = match range {
@@ -151,11 +183,14 @@ impl Backend for LocalFsBackend {
                 let content_range =
                     byte_range
                         .resolve(payload_size)
-                        .ok_or(Error::RangeNotSatisfiable {
+                        .ok_or(ErrorKind::RangeNotSatisfiable {
                             total: payload_size,
                         })?;
                 let payload_start = metadata_line.len() as u64 + content_range.start;
-                reader.seek(std::io::SeekFrom::Start(payload_start)).await?;
+                reader
+                    .seek(std::io::SeekFrom::Start(payload_start))
+                    .await
+                    .context(ErrorKind::BackendFailure, "seeking local-fs object payload")?;
                 let limited = reader.take(content_range.len());
                 (Some(content_range), ReaderStream::new(limited).boxed())
             }
@@ -170,11 +205,12 @@ impl Backend for LocalFsBackend {
         let path = self.path.join(id.as_storage_path().to_string());
         let result = tokio::fs::remove_file(path).await;
         if let Err(e) = &result
-            && e.kind() == ErrorKind::NotFound
+            && e.kind() == io::ErrorKind::NotFound
         {
             objectstore_log::debug!("Object not found");
         }
-        Ok(result?)
+        result.context(ErrorKind::BackendFailure, "deleting local-fs object")?;
+        Ok(())
     }
 }
 
@@ -196,14 +232,18 @@ impl MultipartUploadBackend for LocalFsBackend {
     ) -> Result<InitiateMultipartResponse> {
         let upload_id = UploadId::new(uuid::Uuid::now_v7().to_string())?;
         let dir = self.multipart_dir(id, &upload_id);
-        tokio::fs::create_dir_all(&dir).await?;
+        tokio::fs::create_dir_all(&dir).await.context(
+            ErrorKind::BackendFailure,
+            "creating local-fs multipart upload",
+        )?;
 
         let meta_path = dir.join("metadata.json");
-        let metadata_json = serde_json::to_string(metadata).map_err(|cause| Error::Serde {
-            context: "failed to serialize multipart metadata".to_string(),
-            cause,
-        })?;
-        tokio::fs::write(meta_path, metadata_json).await?;
+        let metadata_json = serde_json::to_string(metadata)
+            .context(ErrorKind::Internal, "encoding local-fs multipart metadata")?;
+        tokio::fs::write(meta_path, metadata_json).await.context(
+            ErrorKind::BackendFailure,
+            "writing local-fs multipart metadata",
+        )?;
 
         Ok(upload_id)
     }
@@ -218,8 +258,14 @@ impl MultipartUploadBackend for LocalFsBackend {
         body: ClientStream,
     ) -> Result<UploadPartResponse> {
         let dir = self.multipart_dir(id, upload_id);
-        if !tokio::fs::try_exists(&dir).await? {
-            return Err(Error::generic("multipart upload not found"));
+        if !tokio::fs::try_exists(&dir).await.context(
+            ErrorKind::BackendFailure,
+            "checking local-fs multipart upload",
+        )? {
+            return Err(Error::new(
+                ErrorKind::BackendFailure,
+                "local-fs multipart upload not found",
+            ));
         }
 
         let etag = format!("\"etag-{part_number}-{content_length}\"");
@@ -229,10 +275,8 @@ impl MultipartUploadBackend for LocalFsBackend {
             "uploaded_at": SystemTime::now(),
             "size": content_length,
         });
-        let header_line = serde_json::to_string(&header).map_err(|cause| Error::Serde {
-            context: "failed to serialize part header".to_string(),
-            cause,
-        })?;
+        let header_line = serde_json::to_string(&header)
+            .context(ErrorKind::Internal, "encoding local-fs part header")?;
 
         let part_path = dir.join(format!("{part_number}.part"));
         let file = OpenOptions::new()
@@ -240,27 +284,46 @@ impl MultipartUploadBackend for LocalFsBackend {
             .write(true)
             .truncate(true)
             .open(part_path)
-            .await?;
+            .await
+            .context(
+                ErrorKind::BackendFailure,
+                "opening local-fs multipart part for writing",
+            )?;
 
         let mut reader = pin!(StreamReader::new(body));
         let mut writer = BufWriter::new(file);
-        writer.write_all(header_line.as_bytes()).await?;
-        writer.write_all(b"\n").await?;
+        writer
+            .write_all(header_line.as_bytes())
+            .await
+            .context(ErrorKind::BackendFailure, "writing local-fs part header")?;
+        writer
+            .write_all(b"\n")
+            .await
+            .context(ErrorKind::BackendFailure, "writing local-fs part header")?;
 
         let _bytes_copied = tokio::io::copy(&mut reader, &mut writer)
             .await
             .map_err(|e| match stream::unpack_client_error(&e) {
-                Some(ce) => Error::Client(ce),
-                None => e.into(),
+                Some(ce) => Error::from(ce),
+                None => Error::with_context(
+                    ErrorKind::BackendFailure,
+                    "writing local-fs multipart part payload",
+                    e,
+                ),
             })?;
 
         // TODO: validate bytes_copied against content_length and return a BadRequest-style
         // error. Needs a service-layer error variant that maps to HTTP 400 without abusing
         // ClientError (which is meant for stream errors).
 
-        writer.flush().await?;
+        writer.flush().await.context(
+            ErrorKind::BackendFailure,
+            "flushing local-fs multipart part",
+        )?;
         let file = writer.into_inner();
-        file.sync_data().await?;
+        file.sync_data()
+            .await
+            .context(ErrorKind::BackendFailure, "syncing local-fs multipart part")?;
         drop(file);
 
         Ok(etag)
@@ -274,14 +337,26 @@ impl MultipartUploadBackend for LocalFsBackend {
         part_number_marker: Option<PartNumber>,
     ) -> Result<ListPartsResponse> {
         let dir = self.multipart_dir(id, upload_id);
-        if !tokio::fs::try_exists(&dir).await? {
-            return Err(Error::generic("multipart upload not found"));
+        if !tokio::fs::try_exists(&dir).await.context(
+            ErrorKind::BackendFailure,
+            "checking local-fs multipart upload",
+        )? {
+            return Err(Error::new(
+                ErrorKind::BackendFailure,
+                "local-fs multipart upload not found",
+            ));
         }
 
-        let mut entries = tokio::fs::read_dir(&dir).await?;
+        let mut entries = tokio::fs::read_dir(&dir).await.context(
+            ErrorKind::BackendFailure,
+            "listing local-fs multipart parts",
+        )?;
         let mut parts = Vec::new();
 
-        while let Some(entry) = entries.next_entry().await? {
+        while let Some(entry) = entries.next_entry().await.context(
+            ErrorKind::BackendFailure,
+            "listing local-fs multipart parts",
+        )? {
             let name = entry.file_name();
             let name_str = name.to_string_lossy();
             let Some(pn_str) = name_str.strip_suffix(".part") else {
@@ -295,15 +370,17 @@ impl MultipartUploadBackend for LocalFsBackend {
                 continue;
             }
 
-            let file = tokio::fs::File::open(entry.path()).await?;
+            let file = tokio::fs::File::open(entry.path())
+                .await
+                .context(ErrorKind::BackendFailure, "opening local-fs multipart part")?;
             let mut reader = BufReader::new(file);
             let mut header_line = String::new();
-            reader.read_line(&mut header_line).await?;
-            let header: serde_json::Value =
-                serde_json::from_str(header_line.trim_end()).map_err(|cause| Error::Serde {
-                    context: "failed to deserialize part header".to_string(),
-                    cause,
-                })?;
+            reader
+                .read_line(&mut header_line)
+                .await
+                .context(ErrorKind::BackendFailure, "reading local-fs part header")?;
+            let header: serde_json::Value = serde_json::from_str(header_line.trim_end())
+                .context(ErrorKind::CorruptData, "decoding local-fs part header")?;
 
             parts.push(Part {
                 part_number: pn,
@@ -339,8 +416,14 @@ impl MultipartUploadBackend for LocalFsBackend {
         upload_id: &UploadId,
     ) -> Result<AbortMultipartResponse> {
         let dir = self.multipart_dir(id, upload_id);
-        if tokio::fs::try_exists(&dir).await? {
-            tokio::fs::remove_dir_all(dir).await?;
+        if tokio::fs::try_exists(&dir).await.context(
+            ErrorKind::BackendFailure,
+            "checking local-fs multipart upload",
+        )? {
+            tokio::fs::remove_dir_all(dir).await.context(
+                ErrorKind::BackendFailure,
+                "removing local-fs multipart upload",
+            )?;
         }
         Ok(())
     }
@@ -352,18 +435,26 @@ impl MultipartUploadBackend for LocalFsBackend {
         parts: Vec<CompletedPart>,
     ) -> Result<CompleteMultipartResponse> {
         let dir = self.multipart_dir(id, upload_id);
-        if !tokio::fs::try_exists(&dir).await? {
-            return Err(Error::generic("multipart upload not found"));
+        if !tokio::fs::try_exists(&dir).await.context(
+            ErrorKind::BackendFailure,
+            "checking local-fs multipart upload",
+        )? {
+            return Err(Error::new(
+                ErrorKind::BackendFailure,
+                "local-fs multipart upload not found",
+            ));
         }
 
         // Read metadata
         let meta_path = dir.join("metadata.json");
-        let meta_bytes = tokio::fs::read(&meta_path).await?;
-        let metadata: Metadata =
-            serde_json::from_slice(&meta_bytes).map_err(|cause| Error::Serde {
-                context: "failed to deserialize multipart metadata".to_string(),
-                cause,
-            })?;
+        let meta_bytes = tokio::fs::read(&meta_path).await.context(
+            ErrorKind::BackendFailure,
+            "reading local-fs multipart metadata",
+        )?;
+        let metadata: Metadata = serde_json::from_slice(&meta_bytes).context(
+            ErrorKind::CorruptData,
+            "decoding local-fs multipart metadata",
+        )?;
 
         // TODO: validate that parts are in ascending part_number order and reject with
         // InvalidPartOrder if not (matches S3/GCS behavior). Needs a proper client error variant.
@@ -371,22 +462,27 @@ impl MultipartUploadBackend for LocalFsBackend {
         // Validate all parts (headers only) before writing anything
         for completed in &parts {
             let part_path = dir.join(format!("{}.part", completed.part_number));
-            if !tokio::fs::try_exists(&part_path).await? {
+            if !tokio::fs::try_exists(&part_path).await.context(
+                ErrorKind::BackendFailure,
+                "checking local-fs multipart part",
+            )? {
                 return Ok(Some(crate::multipart::CompleteMultipartError {
                     code: "InvalidPart".into(),
                     message: format!("part number {} was not uploaded", completed.part_number),
                 }));
             }
 
-            let file = tokio::fs::File::open(&part_path).await?;
+            let file = tokio::fs::File::open(&part_path)
+                .await
+                .context(ErrorKind::BackendFailure, "opening local-fs multipart part")?;
             let mut reader = BufReader::new(file);
             let mut header_line = String::new();
-            reader.read_line(&mut header_line).await?;
-            let header: serde_json::Value =
-                serde_json::from_str(header_line.trim_end()).map_err(|cause| Error::Serde {
-                    context: "failed to deserialize part header".to_string(),
-                    cause,
-                })?;
+            reader
+                .read_line(&mut header_line)
+                .await
+                .context(ErrorKind::BackendFailure, "reading local-fs part header")?;
+            let header: serde_json::Value = serde_json::from_str(header_line.trim_end())
+                .context(ErrorKind::CorruptData, "decoding local-fs part header")?;
 
             let stored_etag = header["etag"].as_str().unwrap_or("");
             if stored_etag != completed.etag {
@@ -402,38 +498,67 @@ impl MultipartUploadBackend for LocalFsBackend {
 
         // Stream parts directly to the final object file
         let path = self.path.join(id.as_storage_path().to_string());
-        tokio::fs::create_dir_all(path.parent().unwrap()).await?;
+        tokio::fs::create_dir_all(path.parent().unwrap())
+            .await
+            .context(
+                ErrorKind::BackendFailure,
+                "creating local-fs object directory",
+            )?;
         let file = OpenOptions::new()
             .create(true)
             .write(true)
             .truncate(true)
             .open(path)
-            .await?;
+            .await
+            .context(
+                ErrorKind::BackendFailure,
+                "opening local-fs object for writing",
+            )?;
         let mut writer = BufWriter::new(file);
 
-        let metadata_json = serde_json::to_string(&metadata).map_err(|cause| Error::Serde {
-            context: "failed to serialize metadata".to_string(),
-            cause,
-        })?;
-        writer.write_all(metadata_json.as_bytes()).await?;
-        writer.write_all(b"\n").await?;
+        let metadata_json = serde_json::to_string(&metadata)
+            .context(ErrorKind::Internal, "encoding local-fs object metadata")?;
+        writer.write_all(metadata_json.as_bytes()).await.context(
+            ErrorKind::BackendFailure,
+            "writing local-fs object metadata",
+        )?;
+        writer.write_all(b"\n").await.context(
+            ErrorKind::BackendFailure,
+            "writing local-fs object metadata",
+        )?;
 
         for completed in &parts {
             let part_path = dir.join(format!("{}.part", completed.part_number));
-            let file = tokio::fs::File::open(&part_path).await?;
+            let file = tokio::fs::File::open(&part_path)
+                .await
+                .context(ErrorKind::BackendFailure, "opening local-fs multipart part")?;
             let mut reader = BufReader::new(file);
             let mut header_line = String::new();
-            reader.read_line(&mut header_line).await?;
-            tokio::io::copy(&mut reader, &mut writer).await?;
+            reader
+                .read_line(&mut header_line)
+                .await
+                .context(ErrorKind::BackendFailure, "reading local-fs part header")?;
+            tokio::io::copy(&mut reader, &mut writer).await.context(
+                ErrorKind::BackendFailure,
+                "assembling local-fs object payload",
+            )?;
         }
 
-        writer.flush().await?;
+        writer
+            .flush()
+            .await
+            .context(ErrorKind::BackendFailure, "flushing local-fs object")?;
         let file = writer.into_inner();
-        file.sync_data().await?;
+        file.sync_data()
+            .await
+            .context(ErrorKind::BackendFailure, "syncing local-fs object")?;
         drop(file);
 
         // Clean up multipart state
-        tokio::fs::remove_dir_all(dir).await?;
+        tokio::fs::remove_dir_all(dir).await.context(
+            ErrorKind::BackendFailure,
+            "removing local-fs multipart upload",
+        )?;
 
         Ok(None)
     }
@@ -841,7 +966,7 @@ mod tests {
             .unwrap();
 
         match backend.get_object(&id, Some(ByteRange::From(100))).await {
-            Err(Error::RangeNotSatisfiable { total: 5 }) => {}
+            Err(error) if matches!(error.kind(), ErrorKind::RangeNotSatisfiable { total: 5 }) => {}
             Err(other) => panic!("expected RangeNotSatisfiable, got: {other:?}"),
             Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"),
         }
diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs
index f34beec0..0c11f924 100644
--- a/objectstore-service/src/backend/s3_compatible.rs
+++ b/objectstore-service/src/backend/s3_compatible.rs
@@ -1,5 +1,7 @@
 //! S3-compatible backend with generic protocol support.
 
+use std::convert::Infallible;
+use std::error::Error as StdError;
 use std::time::SystemTime;
 use std::{fmt, io};
 
@@ -13,7 +15,8 @@ use super::extensions::{ResponseExt, SendTraced};
 use crate::backend::common::{
     self, Backend, DeleteResponse, GetResponse, MetadataResponse, PutResponse,
 };
-use crate::error::{Error, Result};
+use crate::backend::extensions::ReqwestResultExt;
+use crate::error::{Error, ErrorKind, Result, ResultExt as _};
 use crate::id::ObjectId;
 use crate::stream::ClientStream;
 
@@ -72,8 +75,13 @@ pub trait Token: Send + Sync {
 
 /// Provides authentication tokens for S3-compatible requests.
 pub trait TokenProvider: Send + Sync + 'static {
+    /// Error returned when a token cannot be provided.
+    type Error: StdError + Send + Sync + 'static;
+
     /// Returns a fresh token, fetching or refreshing it as needed.
-    fn get_token(&self) -> impl Future<Output = anyhow::Result<impl Token>> + Send;
+    fn get_token(
+        &self,
+    ) -> impl Future<Output = std::result::Result<impl Token, Self::Error>> + Send;
 }
 
 /// Placeholder [`TokenProvider`] for unauthenticated backends.
@@ -81,8 +89,10 @@ pub trait TokenProvider: Send + Sync + 'static {
 pub struct NoToken;
 
 impl TokenProvider for NoToken {
+    type Error = Infallible;
+
     #[allow(refining_impl_trait)]
-    async fn get_token(&self) -> anyhow::Result<NoToken> {
+    async fn get_token(&self) -> std::result::Result<NoToken, Infallible> {
         unimplemented!()
     }
 }
@@ -152,10 +162,7 @@ where
                 provider
                     .get_token()
                     .await
-                    .map_err(|err| Error::Generic {
-                        context: "S3: failed to get authentication token".to_owned(),
-                        cause: Some(err.into()),
-                    })?
+                    .context(ErrorKind::BackendFailure, "getting S3 authentication token")?
                     .as_str(),
             );
         }
@@ -180,10 +187,7 @@ where
         let response = builder
             .send_traced()
             .await
-            .map_err(|cause| Error::Reqwest {
-                context: "S3: failed to send request".to_string(),
-                cause,
-            })?;
+            .reqwest_context("sending an S3 object request")?;
 
         if response.status() == StatusCode::NOT_FOUND {
             objectstore_log::debug!("Object not found");
@@ -198,28 +202,26 @@ where
                 .and_then(|v| v.to_str().ok());
             let total = raw.and_then(ContentRange::parse_unsatisfiable_total);
             let err = match total {
-                Some(total) => Error::RangeNotSatisfiable { total },
-                None => Error::generic(format!(
-                    "S3: 416 response with invalid Content-Range: {raw:?}"
-                )),
+                Some(total) => ErrorKind::RangeNotSatisfiable { total }.into(),
+                None => Error::new(ErrorKind::BackendFailure, "invalid S3 416 Content-Range"),
             };
             response.drain_body().await;
             return Err(err);
         }
 
-        let response = response.check_error("S3: failed to get object").await?;
+        let response = response.check_error("getting an S3 object").await?;
 
         let headers = response.headers();
-        let mut metadata = Metadata::from_headers(headers, GCS_CUSTOM_PREFIX)?;
+        let mut metadata = Metadata::from_headers(headers, GCS_CUSTOM_PREFIX)
+            .context(ErrorKind::CorruptData, "decoding S3 object metadata")?;
 
         let content_range = if response.status() == StatusCode::PARTIAL_CONTENT {
             let range = headers
                 .get(reqwest::header::CONTENT_RANGE)
                 .and_then(|v| v.to_str().ok())
                 .and_then(|s| s.parse::<ContentRange>().ok())
-                .ok_or_else(|| Error::Generic {
-                    context: "S3: 206 response missing valid Content-Range header".to_owned(),
-                    cause: None,
+                .ok_or_else(|| {
+                    Error::new(ErrorKind::BackendFailure, "missing S3 206 Content-Range")
                 })?;
             metadata.size = Some(range.total as usize);
             Some(range)
@@ -231,10 +233,7 @@ where
                 .and_then(|value| value.to_str().ok())
                 .map(|value| value.parse::<usize>())
                 .transpose()
-                .map_err(|cause| Error::Generic {
-                    context: "S3: failed to parse Content-Length from object response".to_string(),
-                    cause: Some(Box::new(cause)),
-                })?;
+                .context(ErrorKind::CorruptData, "decoding S3 Content-Length")?;
 
             if let Some(size) = size {
                 metadata.size = Some(size);
@@ -278,10 +277,13 @@ where
                 format!("/{}/{}", self.bucket, id.as_storage_path()),
             )
             .header("x-goog-metadata-directive", "REPLACE")
-            .headers(metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)?)
+            .headers(
+                metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)
+                    .context(ErrorKind::InvalidMetadata, "encoding S3 object metadata")?,
+            )
             .send_traced()
             .await
-            .check_error("S3: update expiration time")
+            .check_error("updating S3 expiration")
             .await?
             .drain_body()
             .await;
@@ -328,11 +330,14 @@ impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
         objectstore_log::debug!("Writing to s3_compatible backend");
         self.request(Method::PUT, self.object_url(id))
             .await?
-            .headers(metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)?)
+            .headers(
+                metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)
+                    .context(ErrorKind::InvalidMetadata, "encoding S3 object metadata")?,
+            )
             .body(Body::wrap_stream(stream))
             .send_traced()
             .await
-            .check_error("S3: failed to put object")
+            .check_error("uploading an S3 object")
             .await?
             .drain_body()
             .await;
@@ -369,10 +374,7 @@ impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
             .await?
             .send_traced()
             .await
-            .map_err(|cause| Error::Reqwest {
-                context: "S3: failed to send delete request".to_string(),
-                cause,
-            })?;
+            .reqwest_context("sending an S3 delete request")?;
 
         // Do not error for objects that do not exist.
         if response.status() == StatusCode::NOT_FOUND {
@@ -381,7 +383,7 @@ impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
         }
 
         response
-            .check_error("S3: failed to delete object")
+            .check_error("deleting an S3 object")
             .await?
             .drain_body()
             .await;
diff --git a/objectstore-service/src/backend/testing.rs b/objectstore-service/src/backend/testing.rs
index cc062ec0..4b4730e4 100644
--- a/objectstore-service/src/backend/testing.rs
+++ b/objectstore-service/src/backend/testing.rs
@@ -24,7 +24,7 @@
 //!         _inner: &InMemoryBackend,
 //!         _id: &ObjectId,
 //!     ) -> Result<DeleteResponse> {
-//!         Err(crate::error::Error::Io(std::io::Error::new(
+//!         Err(crate::error::Error::with_source(crate::error::ErrorKind::BackendFailure, std::io::Error::new(
 //!             std::io::ErrorKind::ConnectionRefused,
 //!             "simulated delete failure",
 //!         )))
diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs
index 8af2fbfb..d0f245b7 100644
--- a/objectstore-service/src/backend/tiered.rs
+++ b/objectstore-service/src/backend/tiered.rs
@@ -102,11 +102,11 @@
 //! TODO: Update this section when tiered storage implements resumable uploads.
 //!
 //! Not implemented here yet, so [`TieredStorage`] inherits the unsupported defaults from
-//! [`Backend`] and every session creation returns [`Error::NotImplemented`]. A resumable upload
+//! [`Backend`] and every session creation returns [`ErrorKind::Unsupported`]. A resumable upload
 //! will be a regular
 //! long-term write whose payload arrives across several requests, reusing the revision keys,
 //! changelog phases and compare-and-write commit described above: session creation decides
-//! the tier from the declared total length and returns [`Error::NotImplemented`] if that tier
+//! the tier from the declared total length and returns [`ErrorKind::Unsupported`] if that tier
 //! cannot support it,
 //! non-final chunks pass straight through to the upstream session, and the final chunk runs
 //! the long-term write sequence.
@@ -129,7 +129,7 @@ use crate::backend::common::{
     MultipartUploadBackend, PutResponse, TieredGet, TieredMetadata, TieredWrite, Tombstone,
 };
 use crate::backend::{HighVolumeStorageConfig, MultipartUploadStorageConfig};
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Result, ResultExt as _};
 use crate::id::ObjectId;
 use crate::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
@@ -602,7 +602,7 @@ impl TryInto<UploadId> for TieredUploadId {
 
     fn try_into(self) -> Result<UploadId, Self::Error> {
         let json =
-            serde_json::to_vec(&self).map_err(|e| Error::serde("encoding multipart token", e))?;
+            serde_json::to_vec(&self).context(ErrorKind::Internal, "encoding tiered upload ID")?;
         Ok(UploadId::new(
             base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json),
         )?)
@@ -615,8 +615,8 @@ impl TryFrom<&UploadId> for TieredUploadId {
     fn try_from(value: &UploadId) -> Result<Self, Self::Error> {
         let json = base64::engine::general_purpose::URL_SAFE_NO_PAD
             .decode(value.as_bytes())
-            .map_err(|e| Error::generic(format!("invalid multipart upload ID: {e}")))?;
-        serde_json::from_slice(&json).map_err(|e| Error::serde("decoding multipart token", e))
+            .kind(ErrorKind::InvalidUploadId)?;
+        serde_json::from_slice(&json).kind(ErrorKind::InvalidUploadId)
     }
 }
 
@@ -835,8 +835,9 @@ impl MultipartUploadBackend for TieredStorage {
                     physical = ?physical,
                     "complete_multipart call succeeded on long_term backend, but subsequent get_metadata found no object"
                 );
-                return Err(Error::generic(
-                    "completed multipart object not found in long-term storage",
+                return Err(Error::new(
+                    ErrorKind::BackendFailure,
+                    "tiered multipart object missing from long-term storage",
                 ));
             }
             // Failed to `get_metadata`, cannot proceed.
@@ -1165,10 +1166,13 @@ mod tests {
             _inner: &InMemoryBackend,
             _id: &ObjectId,
         ) -> Result<DeleteResponse> {
-            Err(Error::Io(std::io::Error::new(
-                std::io::ErrorKind::ConnectionRefused,
-                "simulated long-term delete failure",
-            )))
+            Err(Error::with_source(
+                ErrorKind::BackendFailure,
+                std::io::Error::new(
+                    std::io::ErrorKind::ConnectionRefused,
+                    "simulated long-term delete failure",
+                ),
+            ))
         }
     }
 
@@ -1302,10 +1306,13 @@ mod tests {
                 // simulate a network error _after_ commit went through
                 inner.compare_and_write(id, current, write).await?;
             }
-            Err(Error::Io(std::io::Error::new(
-                std::io::ErrorKind::TimedOut,
-                "simulated compare_and_write failure",
-            )))
+            Err(Error::with_source(
+                ErrorKind::BackendFailure,
+                std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    "simulated compare_and_write failure",
+                ),
+            ))
         }
     }
 
@@ -1649,6 +1656,17 @@ mod tests {
         assert_eq!(decoded, id);
     }
 
+    #[test]
+    fn malformed_multipart_upload_ids_are_invalid_upload_ids() {
+        let invalid_base64 = UploadId::new("%%%".into()).unwrap();
+        let malformed_json = UploadId::new("bm90IGpzb24".into()).unwrap();
+
+        for upload_id in [&invalid_base64, &malformed_json] {
+            let error = TieredUploadId::try_from(upload_id).unwrap_err();
+            assert_eq!(error.kind(), ErrorKind::InvalidUploadId);
+        }
+    }
+
     #[tokio::test]
     async fn multipart_single_part_roundtrip() {
         let (storage, hv, lt, _) = make_tiered_storage();
@@ -1953,10 +1971,13 @@ mod tests {
                 .complete_multipart(id, upload_id, parts)
                 .await
                 .unwrap();
-            Err(Error::Io(std::io::Error::new(
-                std::io::ErrorKind::TimedOut,
-                "simulated network error on complete_multipart",
-            )))
+            Err(Error::with_source(
+                ErrorKind::BackendFailure,
+                std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    "simulated network error on complete_multipart",
+                ),
+            ))
         }
 
         async fn get_metadata(
@@ -1964,10 +1985,13 @@ mod tests {
             _inner: &InMemoryBackend,
             _id: &ObjectId,
         ) -> Result<MetadataResponse> {
-            Err(Error::Io(std::io::Error::new(
-                std::io::ErrorKind::TimedOut,
-                "simulated network error on get_metadata",
-            )))
+            Err(Error::with_source(
+                ErrorKind::BackendFailure,
+                std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    "simulated network error on get_metadata",
+                ),
+            ))
         }
     }
 
@@ -2072,10 +2096,10 @@ mod tests {
             let mut attempt = self.attempt.lock().await;
             *attempt += 1;
             if *attempt == 1 {
-                Err(Error::Io(std::io::Error::new(
-                    std::io::ErrorKind::TimedOut,
-                    "simulated network error",
-                )))
+                Err(Error::with_source(
+                    ErrorKind::BackendFailure,
+                    std::io::Error::new(std::io::ErrorKind::TimedOut, "simulated network error"),
+                ))
             } else {
                 Ok(inner
                     .complete_multipart(id, upload_id, parts)
@@ -2208,10 +2232,10 @@ mod tests {
             let mut attempt = self.attempt.lock().await;
             *attempt += 1;
             if *attempt == 1 {
-                Err(Error::Io(std::io::Error::new(
-                    std::io::ErrorKind::TimedOut,
-                    "simulated network error",
-                )))
+                Err(Error::with_source(
+                    ErrorKind::BackendFailure,
+                    std::io::Error::new(std::io::ErrorKind::TimedOut, "simulated network error"),
+                ))
             } else {
                 inner.get_metadata(id).await
             }
diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs
index 7715568e..96e5e4a7 100644
--- a/objectstore-service/src/concurrency.rs
+++ b/objectstore-service/src/concurrency.rs
@@ -16,7 +16,7 @@ use futures_util::FutureExt;
 use sentry::{Hub, SentryFutureExt, TransactionContext};
 use tokio::sync::{AcquireError, Notify, OwnedSemaphorePermit, Semaphore};
 
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Panic, Result};
 
 /// Interval for the periodic metrics emitter.
 const EMITTER_INTERVAL: Duration = Duration::from_secs(1);
@@ -120,10 +120,10 @@ impl ConcurrencyLimiter {
     /// If a permit is free, returns immediately without touching the
     /// queue. Otherwise, acquires a queue ticket (bounded by the queue
     /// depth) and waits up to the configured timeout. Returns
-    /// [`Error::AtCapacity`] if the queue is full or on timeout.
+    /// [`ErrorKind::AtCapacity`] if the queue is full or on timeout.
     pub async fn acquire(&self) -> Result<ConcurrencyPermit> {
         if self.tasks_total == 0 {
-            return Err(Error::AtCapacity);
+            return Err(ErrorKind::AtCapacity.into());
         }
 
         // Fast path: Instantly grab a free permit without parking.
@@ -141,13 +141,13 @@ impl ConcurrencyLimiter {
             .queue
             .clone()
             .try_acquire_owned()
-            .map_err(|_| Error::AtCapacity)?;
+            .map_err(|_| ErrorKind::AtCapacity)?;
 
         let acquire = self.tasks.clone().acquire_owned();
         let task_permit = tokio::time::timeout(self.timeout, acquire)
             .await
-            .map_err(|_| Error::AtCapacity)?
-            .map_err(|_| Error::AtCapacity)?;
+            .map_err(|_| ErrorKind::AtCapacity)?
+            .map_err(|_| ErrorKind::AtCapacity)?;
 
         Ok(ConcurrencyPermit {
             task_permit: Some(task_permit),
@@ -158,13 +158,13 @@ impl ConcurrencyLimiter {
 
     /// Tries to acquire a single permit without waiting.
     ///
-    /// Returns [`Error::AtCapacity`] when no permits are available.
+    /// Returns [`ErrorKind::AtCapacity`] when no permits are available.
     pub fn try_acquire(&self) -> Result<ConcurrencyPermit> {
         let task_permit = self
             .tasks
             .clone()
             .try_acquire_owned()
-            .map_err(|_| Error::AtCapacity)?;
+            .map_err(|_| ErrorKind::AtCapacity)?;
 
         Ok(ConcurrencyPermit {
             task_permit: Some(task_permit),
@@ -181,10 +181,10 @@ impl ConcurrencyLimiter {
     /// acquired under a single timeout deadline configured via
     /// [`with_timeout`](Self::with_timeout).
     ///
-    /// Returns [`Error::AtCapacity`] on timeout or when `max` is zero.
+    /// Returns [`ErrorKind::AtCapacity`] on timeout or when `max` is zero.
     pub async fn acquire_bulk(&self) -> Result<ConcurrencyPermit> {
         if self.tasks_total == 0 {
-            return Err(Error::AtCapacity);
+            return Err(ErrorKind::AtCapacity.into());
         }
 
         let bulk_sem = self.bulk.clone();
@@ -198,8 +198,8 @@ impl ConcurrencyLimiter {
 
         let (task_permit, bulk_permit) = tokio::time::timeout(self.timeout, acquire)
             .await
-            .map_err(|_| Error::AtCapacity)?
-            .map_err(|_: AcquireError| Error::AtCapacity)?;
+            .map_err(|_| ErrorKind::AtCapacity)?
+            .map_err(|_: AcquireError| ErrorKind::AtCapacity)?;
 
         Ok(ConcurrencyPermit {
             task_permit: Some(task_permit),
@@ -348,7 +348,7 @@ where
             let result = std::panic::AssertUnwindSafe(f)
                 .catch_unwind()
                 .await
-                .unwrap_or_else(|payload| Err(Error::panic(payload)));
+                .unwrap_or_else(|payload| Err(Panic::new(payload).into()));
 
             if let Err(ref e) = result {
                 let error = e as &dyn std::error::Error;
@@ -370,8 +370,9 @@ where
     );
 
     rx.await.map_err(|_| {
-        objectstore_log::error!(!!&Error::Dropped, operation, "Task failed");
-        Error::Dropped
+        let error = Error::new(ErrorKind::Internal, "service task dropped");
+        objectstore_log::error!(!!&error, operation, "Task failed");
+        error
     })?
 }
 
@@ -380,7 +381,6 @@ mod tests {
     use std::sync::atomic::{AtomicU32, Ordering};
 
     use super::*;
-    use crate::error::Error;
 
     #[test]
     fn available_permits_tracks_held() {
@@ -430,7 +430,7 @@ mod tests {
         let _permit = limiter.try_acquire().unwrap();
 
         let result = limiter.try_acquire();
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
     }
 
     #[test]
@@ -505,7 +505,11 @@ mod tests {
 
         let p1 = limiter.try_acquire().unwrap();
         let p2 = limiter.try_acquire().unwrap();
-        assert!(matches!(limiter.try_acquire(), Err(Error::AtCapacity)));
+        assert!(
+            limiter
+                .try_acquire()
+                .is_err_and(|error| error.kind() == ErrorKind::AtCapacity)
+        );
 
         drop(p1);
         assert!(limiter.try_acquire().is_ok());
@@ -520,7 +524,7 @@ mod tests {
 
         let start = tokio::time::Instant::now();
         let result = limiter.acquire().await;
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         assert_eq!(start.elapsed(), Duration::ZERO);
         drop(bulk_permits);
     }
@@ -568,7 +572,7 @@ mod tests {
         tokio::time::sleep(Duration::from_secs(2)).await;
 
         let result = waiter.await.unwrap();
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         assert_eq!(limiter.queued_permits(), 0);
     }
 
@@ -584,7 +588,7 @@ mod tests {
 
         assert_eq!(limiter.queued_permits(), 1);
         let result = limiter.acquire().await;
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
     }
 
     #[tokio::test(start_paused = true)]
@@ -640,7 +644,7 @@ mod tests {
 
         let start = tokio::time::Instant::now();
         let result = limiter.acquire().await;
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         assert_eq!(start.elapsed(), Duration::ZERO);
     }
 
@@ -768,7 +772,7 @@ mod tests {
         tokio::time::sleep(Duration::from_secs(2)).await;
 
         let result = waiter.await.unwrap();
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         assert_eq!(limiter.used_bulk_permits(), 0);
     }
 
@@ -797,7 +801,7 @@ mod tests {
 
         let start = tokio::time::Instant::now();
         let result = limiter.acquire_bulk().await;
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         assert_eq!(start.elapsed(), Duration::ZERO);
     }
 
@@ -815,7 +819,7 @@ mod tests {
         tokio::time::sleep(Duration::from_secs(2)).await;
 
         let result = waiter.await.unwrap();
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         drop(permit);
     }
 
@@ -842,7 +846,7 @@ mod tests {
         // Third waiter exceeds queue depth — rejected instantly.
         let start = tokio::time::Instant::now();
         let result = limiter.acquire().await;
-        assert!(matches!(result, Err(Error::AtCapacity)));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity));
         assert_eq!(start.elapsed(), Duration::ZERO);
 
         drop(bulk_permits);
diff --git a/objectstore-service/src/error.rs b/objectstore-service/src/error.rs
index 3f8dd71d..022d0828 100644
--- a/objectstore-service/src/error.rs
+++ b/objectstore-service/src/error.rs
@@ -1,177 +1,67 @@
-//! Error types for service and backend operations.
+//! Semantic errors for service and backend operations.
 //!
-//! [`Error`] covers I/O, serialization, HTTP, metadata, authentication,
-//! and backend-specific failures. [`Result`] is the corresponding alias.
+//! [`Error`] deliberately exposes only a stable semantic [`ErrorKind`]. Human-readable context and
+//! the source chain retain diagnostic detail without making backend implementation details part of
+//! the service API.
 
 use std::any::Any;
+use std::borrow::Cow;
+use std::error::Error as StdError;
 use std::fmt;
 
 use objectstore_log::Level;
-use reqwest::StatusCode;
-use thiserror::Error as ThisError;
-
-use crate::stream::ClientError;
-
-/// Structured error detail parsed from a backend HTTP error response.
-///
-/// Formats conditionally: includes only the fields that are non-empty.
+/// A panic captured from a service task.
 #[derive(Debug)]
-pub struct BackendDetail {
-    /// Machine-readable error code (e.g., "InvalidArgument", "NoSuchKey").
-    pub code: String,
-    /// Human-readable error message from the response body.
-    pub message: String,
+pub struct Panic {
+    message: Cow<'static, str>,
 }
 
-impl BackendDetail {
-    /// Creates a new [`BackendDetail`] with empty code and message.
-    pub fn none() -> Self {
-        Self {
-            code: String::new(),
-            message: String::new(),
-        }
+impl Panic {
+    /// Extracts a message from a panic payload.
+    pub fn new(payload: Box<dyn Any + Send>) -> Self {
+        let message = if let Some(s) = payload.downcast_ref::<&str>() {
+            Cow::Borrowed(*s)
+        } else if let Some(s) = payload.downcast_ref::<String>() {
+            Cow::Owned(s.clone())
+        } else {
+            Cow::Borrowed("unknown panic")
+        };
+        Self { message }
     }
 }
 
-impl fmt::Display for BackendDetail {
+impl fmt::Display for Panic {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match (self.code.is_empty(), self.message.is_empty()) {
-            (false, false) => write!(f, "{} (backend code {})", self.message, self.code),
-            (true, false) => write!(f, "{}", self.message),
-            (false, true) => write!(f, "backend code {}", self.code),
-            (true, true) => Ok(()),
-        }
+        f.write_str(&self.message)
     }
 }
 
-/// Error type for service operations.
-#[derive(Debug, ThisError)]
-pub enum Error {
-    /// IO errors related to payload streaming or file operations.
-    #[error("i/o error: {0}")]
-    Io(#[from] std::io::Error),
-
-    /// Error originating from a client-supplied input stream.
-    ///
-    /// Indicates the client is at fault (e.g. dropped connection mid-upload) and should
-    /// map to a 4xx response rather than a 5xx.
-    #[error("error reading client stream: {0}")]
-    Client(#[from] ClientError),
-
-    /// Errors related to de/serialization.
-    #[error("serde error: {context}")]
-    Serde {
-        /// Context describing what was being serialized/deserialized.
-        context: String,
-        /// The underlying serde error.
-        #[source]
-        cause: serde_json::Error,
-    },
-
-    /// All errors stemming from the reqwest client, used in multiple backends to send requests to
-    /// e.g. GCP APIs.
-    /// These can be network errors encountered when sending the requests, but can also indicate
-    /// errors returned by the API itself.
-    #[error("reqwest error: {context}")]
-    Reqwest {
-        /// Context describing the request that failed.
-        context: String,
-        /// The underlying reqwest error.
-        #[source]
-        cause: reqwest::Error,
-    },
-
-    /// An HTTP error response from a storage backend (e.g., GCS, S3).
-    ///
-    /// Unlike [`Reqwest`](Self::Reqwest), which covers transport-level failures, this variant
-    /// captures application-level error responses where the server returned a 4xx/5xx status code
-    /// along with a structured error body.
-    #[error("{context} ({status}). {detail}")]
-    BackendResponse {
-        /// Context describing the request that failed.
-        context: &'static str,
-        /// The HTTP status code returned by the backend.
-        status: StatusCode,
-        /// Parsed error code and message from the response body.
-        detail: BackendDetail,
-    },
-
-    /// Errors related to de/serialization and parsing of object metadata.
-    #[error("metadata error: {0}")]
-    Metadata(#[from] objectstore_types::metadata::Error),
-
-    /// Errors encountered when attempting to authenticate with GCP.
-    #[error("GCP authentication error: {0}")]
-    GcpAuth(#[from] gcp_auth::Error),
-
-    /// A spawned service task panicked.
-    #[error("service task failed: {0}")]
-    Panic(String),
-
-    /// A spawned service task was dropped before it could deliver its result.
-    ///
-    /// This is an unexpected condition that can occur when the runtime drops the task for unknown
-    /// reasons.
-    #[error("task dropped")]
-    Dropped,
-
-    /// A redirect tombstone was encountered at a place where it is not supported.
-    ///
-    /// This indicates a caller bug — tombstone-aware reads must go through the
-    /// [`HighVolumeBackend`](crate::backend::common::HighVolumeBackend) methods.
-    #[error("unexpected tombstone")]
-    UnexpectedTombstone,
+impl StdError for Panic {}
 
-    /// The requested byte range is not satisfiable for the object's size.
-    #[error("range not satisfiable (object size: {total} bytes)")]
+/// The client-visible semantic classification of a service error.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ErrorKind {
+    /// Object metadata supplied by a client is invalid.
+    InvalidMetadata,
+    /// A multipart upload identifier is invalid.
+    InvalidUploadId,
+    /// A client-provided request stream failed.
+    ClientStream,
+    /// A requested byte range cannot be resolved against the object size.
     RangeNotSatisfiable {
-        /// Total size of the object in bytes.
+        /// Total object length in bytes.
         total: u64,
     },
-
-    /// The service has reached its concurrency limit and cannot accept more operations.
-    #[error("concurrency limit reached")]
-    AtCapacity,
-
-    /// Any other error stemming from one of the storage backends, which might be specific to that
-    /// backend or to a certain operation.
-    #[error("storage backend error: {context}")]
-    Generic {
-        /// Context describing the operation that failed.
-        context: String,
-        /// The underlying error, if available.
-        #[source]
-        cause: Option<Box<dyn std::error::Error + Send + Sync>>,
-    },
-
-    /// The functionality is not implemented by this instance of the service.
-    #[error("not implemented")]
-    NotImplemented,
-
-    /// Invalid upload ID (e.g. path traversal attempt).
-    #[error(transparent)]
-    InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId),
-
-    /// A resumable chunk was submitted at an offset that's different from the one held by the
-    /// backend.
-    #[error("upload offset mismatch (server holds {offset} bytes)")]
+    /// A resumable chunk starts at a different offset than the backend currently holds.
     UploadOffsetMismatch {
         /// The offset the backend currently holds.
         offset: u64,
     },
-
-    /// The resumable upload session expired or was canceled, retaining nothing.
-    #[error("upload session gone")]
+    /// A resumable upload session expired or was canceled.
     UploadSessionGone,
-
-    /// The backend does not recognize the addressed resumable upload session.
-    #[error("unknown upload session")]
+    /// The backend does not recognize a resumable upload session.
     UnknownUploadSession,
-
-    /// A resumable upload chunk would exceed the length declared for the session.
-    #[error(
-        "chunk at offset {offset} with length {content_length} exceeds upload length {upload_length}"
-    )]
+    /// A resumable chunk would exceed the total upload length.
     ChunkExceedsUploadLength {
         /// The offset at which the chunk would be written.
         offset: u64,
@@ -180,76 +70,296 @@ pub enum Error {
         /// The total upload length declared when the session was created.
         upload_length: u64,
     },
+    /// The service cannot accept more work.
+    AtCapacity,
+    /// The requested operation is unsupported.
+    Unsupported,
+    /// A storage backend operation failed.
+    BackendFailure,
+    /// A storage backend rejected the operation because it is rate limited.
+    BackendRateLimited,
+    /// A storage backend operation timed out.
+    BackendTimeout,
+    /// A storage backend is temporarily unavailable.
+    BackendUnavailable,
+    /// A service task panicked.
+    Panic,
+    /// Persisted or remote data is corrupt.
+    CorruptData,
+    /// An unexpected internal service failure occurred.
+    Internal,
+}
+
+impl fmt::Display for ErrorKind {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::InvalidMetadata => f.write_str("invalid object metadata"),
+            Self::InvalidUploadId => f.write_str("invalid upload id"),
+            Self::ClientStream => f.write_str("invalid client stream"),
+            Self::RangeNotSatisfiable { total } => {
+                write!(f, "range not satisfiable (object size: {total} bytes)")
+            }
+            Self::UploadOffsetMismatch { offset } => {
+                write!(f, "upload offset mismatch (server holds {offset} bytes)")
+            }
+            Self::UploadSessionGone => f.write_str("upload session gone"),
+            Self::UnknownUploadSession => f.write_str("unknown upload session"),
+            Self::ChunkExceedsUploadLength {
+                offset,
+                content_length,
+                upload_length,
+            } => write!(
+                f,
+                "chunk at offset {offset} with length {content_length} exceeds upload length {upload_length}"
+            ),
+            Self::AtCapacity => f.write_str("service at capacity"),
+            Self::Unsupported => f.write_str("unsupported operation"),
+            Self::BackendFailure => f.write_str("backend operation failed"),
+            Self::BackendRateLimited => f.write_str("backend rate limited"),
+            Self::BackendTimeout => f.write_str("backend timed out"),
+            Self::BackendUnavailable => f.write_str("backend unavailable"),
+            Self::CorruptData => f.write_str("corrupt stored data"),
+            Self::Panic => f.write_str("service task panicked"),
+            Self::Internal => f.write_str("internal service error"),
+        }
+    }
+}
+
+/// Opaque service error with a stable semantic kind.
+///
+/// Its string representation is the kind followed by `: ` and human-readable context when context
+/// is present. The underlying source is retained separately through [`StdError::source`].
+pub struct Error {
+    kind: ErrorKind,
+    context: Option<Cow<'static, str>>,
+    source: Option<Box<dyn StdError + Send + Sync>>,
 }
 
 impl Error {
-    /// Creates an [`Error::Panic`] from a panic payload, extracting the message.
-    pub fn panic(payload: Box<dyn Any + Send>) -> Self {
-        let msg = if let Some(s) = payload.downcast_ref::<&str>() {
-            (*s).to_owned()
-        } else if let Some(s) = payload.downcast_ref::<String>() {
-            s.clone()
-        } else {
-            "unknown panic".to_owned()
-        };
-        Self::Panic(msg)
+    /// Returns this error's semantic kind.
+    pub fn kind(&self) -> ErrorKind {
+        self.kind
     }
 
-    /// Creates an [`Error::Reqwest`] from a reqwest error with context.
-    pub fn reqwest(context: impl Into<String>, cause: reqwest::Error) -> Self {
-        Self::Reqwest {
-            context: context.into(),
-            cause,
-        }
+    /// Creates an error without an underlying source and with human-readable context.
+    pub fn new(kind: ErrorKind, context: impl Into<Cow<'static, str>>) -> Self {
+        Self::build(kind, Some(context.into()), None)
     }
 
-    /// Creates an [`Error::Serde`] from a serde error with context.
-    pub fn serde(context: impl Into<String>, cause: serde_json::Error) -> Self {
-        Self::Serde {
-            context: context.into(),
-            cause,
-        }
+    /// Creates an error with an underlying source.
+    pub fn with_source<E>(kind: ErrorKind, source: E) -> Self
+    where
+        E: StdError + Send + Sync + 'static,
+    {
+        Self::build(kind, None, Some(Box::new(source)))
     }
 
-    /// Creates an [`Error::Generic`] with a context string and no cause.
-    pub fn generic(context: impl Into<String>) -> Self {
-        Self::Generic {
-            context: context.into(),
-            cause: None,
+    pub(crate) fn with_context<E>(
+        kind: ErrorKind,
+        context: impl Into<Cow<'static, str>>,
+        source: E,
+    ) -> Self
+    where
+        E: StdError + Send + Sync + 'static,
+    {
+        Self::build(kind, Some(context.into()), Some(Box::new(source)))
+    }
+
+    fn build(
+        kind: ErrorKind,
+        context: Option<Cow<'static, str>>,
+        source: Option<Box<dyn StdError + Send + Sync>>,
+    ) -> Self {
+        Self {
+            kind,
+            context,
+            source,
         }
     }
 
     /// Returns the appropriate log level for this error.
     pub fn level(&self) -> Level {
-        match self {
+        match self.kind {
             // Malformed client input at DEBUG level
-            Self::Client(_) => Level::DEBUG,
-            Self::Metadata(_) => Level::DEBUG,
-            Self::RangeNotSatisfiable { .. } => Level::DEBUG,
-            Self::UploadOffsetMismatch { .. } => Level::DEBUG,
-            Self::UploadSessionGone => Level::DEBUG,
-            Self::UnknownUploadSession => Level::DEBUG,
-            Self::ChunkExceedsUploadLength { .. } => Level::DEBUG,
+            ErrorKind::InvalidMetadata => Level::DEBUG,
+            ErrorKind::InvalidUploadId => Level::DEBUG,
+            ErrorKind::ClientStream => Level::DEBUG,
+            ErrorKind::RangeNotSatisfiable { .. } => Level::DEBUG,
+            ErrorKind::UploadOffsetMismatch { .. } => Level::DEBUG,
+            ErrorKind::UploadSessionGone => Level::DEBUG,
+            ErrorKind::UnknownUploadSession => Level::DEBUG,
+            ErrorKind::ChunkExceedsUploadLength { .. } => Level::DEBUG,
             // Indicates that optional functionality is not supported.
             // We don't want a rogue client spamming us with Sentry errors just by calling an API
             // that the server doesn't support, so we just log it.
-            Self::NotImplemented => Level::INFO,
-            // Like rate limits, we treat capacity errors as warnings
-            Self::AtCapacity => Level::WARN,
-            // All other errors are service or backend failures
-            Self::Io(_) => Level::ERROR,
-            Self::Serde { .. } => Level::ERROR,
-            Self::Reqwest { .. } => Level::ERROR,
-            Self::BackendResponse { .. } => Level::ERROR,
-            Self::GcpAuth(_) => Level::ERROR,
-            Self::Panic(_) => Level::ERROR,
-            Self::Dropped => Level::ERROR,
-            Self::UnexpectedTombstone => Level::ERROR,
-            Self::InvalidUploadId(_) => Level::DEBUG,
-            Self::Generic { .. } => Level::ERROR,
+            ErrorKind::Unsupported => Level::INFO,
+            // Capacity, rate-limit, and transient backend errors are warnings.
+            ErrorKind::AtCapacity => Level::WARN,
+            ErrorKind::BackendRateLimited => Level::WARN,
+            ErrorKind::BackendTimeout => Level::WARN,
+            ErrorKind::BackendUnavailable => Level::WARN,
+            // All other errors are service or backend failures. These become Sentry errors.
+            ErrorKind::BackendFailure => Level::ERROR,
+            ErrorKind::Panic => Level::ERROR,
+            ErrorKind::CorruptData => Level::ERROR,
+            ErrorKind::Internal => Level::ERROR,
         }
     }
 }
 
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.kind.fmt(f)?;
+        if let Some(context) = &self.context {
+            write!(f, ": {context}")?;
+        }
+        Ok(())
+    }
+}
+
+impl fmt::Debug for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Error")
+            .field("kind", &self.kind)
+            .field("context", &self.context)
+            .field("source", &self.source)
+            .finish()
+    }
+}
+
+impl StdError for Error {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        self.source.as_deref().map(|source| source as _)
+    }
+}
+
+impl From<ErrorKind> for Error {
+    fn from(kind: ErrorKind) -> Self {
+        Self::build(kind, None, None)
+    }
+}
+
+impl From<Panic> for Error {
+    fn from(source: Panic) -> Self {
+        Self::with_source(ErrorKind::Panic, source)
+    }
+}
+
+/// Adds a semantic kind and optional context when converting an external error.
+pub trait ResultExt<T> {
+    /// Converts an external error into a service error with `kind` and human-readable context.
+    ///
+    /// The source error is retained, while the rendered service error contains the semantic kind
+    /// and context.
+    ///
+    /// ```
+    /// use objectstore_service::error::{ErrorKind, ResultExt as _};
+    ///
+    /// let result = std::fs::read("missing")
+    ///     .context(ErrorKind::BackendFailure, "reading local object");
+    /// let error = result.unwrap_err();
+    /// assert_eq!(
+    ///     error.to_string(),
+    ///     "backend operation failed: reading local object"
+    /// );
+    /// ```
+    fn context(self, kind: ErrorKind, context: impl Into<Cow<'static, str>>) -> Result<T>;
+
+    /// Converts an external error into a service error with only `kind`.
+    ///
+    /// Use this when the source already identifies the failure or when the operation is expected to
+    /// be infallible. The source error is still retained.
+    ///
+    /// ```
+    /// use objectstore_service::error::{ErrorKind, ResultExt as _};
+    ///
+    /// let result = "invalid".parse::<u64>().kind(ErrorKind::InvalidMetadata);
+    /// assert_eq!(result.unwrap_err().to_string(), "invalid object metadata");
+    /// ```
+    fn kind(self, kind: ErrorKind) -> Result<T>;
+}
+
+impl<T, E> ResultExt<T> for std::result::Result<T, E>
+where
+    E: StdError + Send + Sync + 'static,
+{
+    fn context(self, kind: ErrorKind, context: impl Into<Cow<'static, str>>) -> Result<T> {
+        self.map_err(|source| Error::with_context(kind, context, source))
+    }
+
+    fn kind(self, kind: ErrorKind) -> Result<T> {
+        self.map_err(|source| Error::with_source(kind, source))
+    }
+}
+
+impl From<std::io::Error> for Error {
+    fn from(source: std::io::Error) -> Self {
+        Self::with_source(ErrorKind::BackendFailure, source)
+    }
+}
+
+impl From<crate::stream::ClientError> for Error {
+    fn from(source: crate::stream::ClientError) -> Self {
+        Self::with_source(ErrorKind::ClientStream, source)
+    }
+}
+
+impl From<objectstore_types::multipart::InvalidUploadId> for Error {
+    fn from(source: objectstore_types::multipart::InvalidUploadId) -> Self {
+        Self::with_source(ErrorKind::InvalidUploadId, source)
+    }
+}
+
 /// Result type for service operations.
 pub type Result<T, E = Error> = std::result::Result<T, E>;
+
+#[cfg(test)]
+mod tests {
+    use std::error::Error as _;
+    use std::io;
+
+    use super::{Error, ErrorKind, Panic};
+
+    #[test]
+    fn opaque_error_preserves_source() {
+        let error = Error::with_source(ErrorKind::BackendFailure, io::Error::other("backend down"));
+        let standard_error: &dyn std::error::Error = &error;
+
+        assert_eq!(error.kind(), ErrorKind::BackendFailure);
+        assert_eq!(standard_error.source().unwrap().to_string(), "backend down");
+    }
+
+    #[test]
+    fn context_renders_after_kind() {
+        let error = Error::with_context(
+            ErrorKind::BackendFailure,
+            "reading local object",
+            io::Error::other("backend down"),
+        );
+
+        assert_eq!(
+            error.to_string(),
+            "backend operation failed: reading local object"
+        );
+    }
+
+    #[test]
+    fn error_kind_default_message_includes_range_size() {
+        let error: Error = ErrorKind::RangeNotSatisfiable { total: 42 }.into();
+
+        assert_eq!(
+            error.to_string(),
+            "range not satisfiable (object size: 42 bytes)"
+        );
+    }
+
+    #[test]
+    fn panic_uses_the_payload_message() {
+        let panic = Panic::new(Box::new("task panicked"));
+        let error: Error = panic.into();
+
+        assert_eq!(error.kind(), ErrorKind::Panic);
+        assert_eq!(error.to_string(), "service task panicked");
+        assert_eq!(error.source().unwrap().to_string(), "task panicked");
+    }
+}
diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs
index f410607e..9cd48288 100644
--- a/objectstore-service/src/service.rs
+++ b/objectstore-service/src/service.rs
@@ -15,7 +15,7 @@ use objectstore_types::resumable::{SessionToken, UploadProgress};
 use crate::backend::common::Backend;
 use crate::backend::counting::CountingBackend;
 use crate::concurrency::ConcurrencyLimiter;
-use crate::error::Result;
+use crate::error::{ErrorKind, Result, ResultExt as _};
 use crate::id::{ObjectContext, ObjectId};
 use crate::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
@@ -206,7 +206,7 @@ impl StorageService {
         metadata: Metadata,
         stream: ClientStream,
     ) -> Result<InsertResponse> {
-        metadata.validate()?;
+        metadata.validate().kind(ErrorKind::InvalidMetadata)?;
         let id = ObjectId::optional(context, key);
         let inner = Arc::clone(&self.inner);
         self.spawn("insert", async move {
@@ -260,7 +260,7 @@ impl StorageService {
         id: ObjectId,
         metadata: Metadata,
     ) -> Result<InitiateMultipartResponse> {
-        metadata.validate()?;
+        metadata.validate().kind(ErrorKind::InvalidMetadata)?;
         self.inner.as_multipart_upload_backend()?; // Fail before clone/spawn if unsupported
         let inner = self.inner.clone();
         self.spawn("initiate_multipart", async move {
@@ -373,7 +373,7 @@ impl StorageService {
         metadata: Metadata,
         total_length: u64,
     ) -> Result<Option<SessionToken>> {
-        metadata.validate()?;
+        metadata.validate().kind(ErrorKind::InvalidMetadata)?;
         let inner = Arc::clone(&self.inner);
         self.spawn("create_upload_session", async move {
             inner
@@ -453,7 +453,6 @@ mod tests {
     use crate::backend::testing::{Hooks, TestBackend};
     use crate::backend::tiered::TieredStorage;
     use crate::change_stream::ChangeStreamFactory;
-    use crate::error::Error;
     use crate::stream::{self, ClientStream};
 
     fn make_context() -> ObjectContext {
@@ -664,10 +663,15 @@ mod tests {
         let id = ObjectId::new(make_context(), "panic-test".into());
         let result = service.get_object(id, None).await;
 
-        let Err(Error::Panic(msg)) = result else {
+        let Err(error) = result else {
             panic!("expected Panic error");
         };
-        assert!(msg.contains("intentional panic in get_object"), "{msg}");
+        assert_eq!(error.kind(), ErrorKind::Panic);
+        assert_eq!(error.to_string(), "service task panicked");
+        assert_eq!(
+            std::error::Error::source(&error).unwrap().to_string(),
+            "intentional panic in get_object"
+        );
     }
 
     /// In-memory backend with optional synchronization for `put_object`.
@@ -806,7 +810,9 @@ mod tests {
             .await;
 
         assert!(
-            matches!(result, Err(Error::AtCapacity)),
+            result
+                .as_ref()
+                .is_err_and(|error| error.kind() == ErrorKind::AtCapacity),
             "expected AtCapacity, got {result:?}"
         );
 
@@ -860,12 +866,12 @@ mod tests {
         // First operation panics — the permit must still be released.
         let id = ObjectId::new(make_context(), "panic-permit".into());
         let result = service.get_object(id.clone(), None).await;
-        assert!(matches!(result, Err(Error::Panic(_))));
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::Panic));
 
         // Second operation should succeed in acquiring the permit (not AtCapacity).
         let result = service.get_object(id, None).await;
         assert!(
-            !matches!(result, Err(Error::AtCapacity)),
+            !result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity),
             "permit was not released after panic"
         );
     }
@@ -897,6 +903,6 @@ mod tests {
         };
 
         let result = service.create_upload_session(id, metadata, 1024).await;
-        assert!(matches!(result, Err(Error::Metadata(_))), "{result:?}");
+        assert!(result.is_err_and(|error| error.kind() == ErrorKind::InvalidMetadata));
     }
 }
diff --git a/objectstore-service/src/stream.rs b/objectstore-service/src/stream.rs
index efaf9c43..bc2a0301 100644
--- a/objectstore-service/src/stream.rs
+++ b/objectstore-service/src/stream.rs
@@ -66,8 +66,8 @@ impl From<ClientError> for io::Error {
 /// Uses [`ClientError`] as the error type so that a dropped or interrupted
 /// client connection is distinguishable from a backend I/O failure. Backends
 /// that detect a [`ClientError`] (via [`unpack_client_error`]) can surface it
-/// as [`crate::error::Error::Client`], which the server maps to HTTP 400 rather
-/// than 500.
+/// as [`ClientStream`](crate::error::ErrorKind::ClientStream), which the server
+/// maps to HTTP 400 rather than 500.
 ///
 /// Use [`single`] to construct a single-chunk `ClientStream` from an owned value.
 pub type ClientStream = BoxStream<'static, Result<Bytes, ClientError>>;
@@ -81,7 +81,8 @@ pub type ClientStream = BoxStream<'static, Result<Bytes, ClientError>>;
 ///   value is a `ClientError`.
 ///
 /// Use this in `put_object` implementations to reclassify body-stream errors
-/// as [`crate::error::Error::Client`] instead of an opaque server error.
+/// as [`ClientStream`](crate::error::ErrorKind::ClientStream) instead of an
+/// opaque server error.
 pub fn unpack_client_error<E>(err: &E) -> Option<ClientError>
 where
     E: Error + 'static,
diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs
index 27eed920..5ce07088 100644
--- a/objectstore-service/src/streaming.rs
+++ b/objectstore-service/src/streaming.rs
@@ -13,14 +13,15 @@
 //! regular requests.
 //!
 //! The regular acquire timeout applies: Operations that cannot acquire a permit within the
-//! configured queue timeout fail with [`Error::AtCapacity`].
+//! configured queue timeout fail with [`AtCapacity`](crate::error::ErrorKind::AtCapacity).
 //!
 //! ## Concurrency Model
 //!
 //! [`StreamExecutor::execute`] uses `buffer_unordered` with the bulk budget as the concurrency
 //! bound. The input stream is pulled lazily and results are yielded in completion order. Each
 //! operation is wrapped in a [`tokio::spawn`] for panic isolation: a panic in one operation
-//! surfaces as [`Error::Panic`] for that item and does not affect the others.
+//! surfaces as a [`Panic`](crate::error::ErrorKind::Panic) for that item and does not affect
+//! the others.
 
 use std::sync::Arc;
 
@@ -229,8 +230,8 @@ impl StreamExecutor {
     /// that already hold a permit proceeds concurrently.
     ///
     /// Operations that cannot acquire a permit within the configured queue
-    /// timeout fail with [`Error::AtCapacity`]. Results are yielded in
-    /// completion order (not submission order).
+    /// timeout fail with [`AtCapacity`](crate::error::ErrorKind::AtCapacity).
+    /// Results are yielded in completion order (not submission order).
     pub fn execute<E>(
         self,
         context: ObjectContext,
@@ -342,7 +343,7 @@ mod tests {
     use crate::backend::in_memory::InMemoryBackend;
     use crate::backend::testing::{Hooks, TestBackend};
     use crate::concurrency::ConcurrencyLimiter;
-    use crate::error::Error;
+    use crate::error::{Error, ErrorKind};
     use crate::service::StorageService;
     use crate::stream::{self, ClientStream};
 
@@ -586,7 +587,10 @@ mod tests {
 
         assert_eq!(outcomes.len(), 1);
         assert!(
-            matches!(&outcomes[0].1, Err(Error::AtCapacity)),
+            outcomes[0]
+                .1
+                .as_ref()
+                .is_err_and(|error| error.kind() == ErrorKind::AtCapacity),
             "expected AtCapacity, got {:?}",
             outcomes[0].1,
         );