feat: preserve task executor error types with associated errors

examples/provisioned_executor.rs

@@ -70,6 +70,8 @@ impl SandboxExecutor {
 }
 
 impl TaskExecutor<AgentTurnInput, AgentTurnOutput> for SandboxExecutor {
+    type Error = Error;
+
     fn execute(
         &self,
         input: AgentTurnInput,

src/error.rs

@@ -4,6 +4,9 @@ use thiserror::Error;
 
 use crate::types::TaskId;
 
+/// Boxed error returned by task execution code.
+pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
 /// Result type used by Steda.
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -86,12 +89,21 @@ pub enum Error {
     #[error("serialization error: {0}")]
     Serialization(#[from] serde_json::Error),
 
+    /// Error returned by task execution code outside Steda's own error domain.
+    #[error("{0}")]
+    Task(#[source] BoxError),
+
     /// Catch-all error for cases that do not deserve a dedicated variant.
     #[error("{0}")]
     Other(String),
 }
 
 impl Error {
+    /// Wrap an external task execution error while preserving its source chain.
+    pub fn task(error: impl std::error::Error + Send + Sync + 'static) -> Self {
+        Self::Task(Box::new(error))
+    }
+
     /// Returns true when the error represents a deliberate task suspension.
     pub const fn is_suspended(&self) -> bool {
         matches!(self, Self::Suspended)

src/lib.rs

@@ -332,7 +332,7 @@ pub mod middleware {
 pub const SCHEMA_SQL: &str = include_str!("../sql/steda.sql");
 
 pub use context::{TaskContext, TaskWait};
-pub use error::{Error, Result};
+pub use error::{BoxError, Error, Result};
 pub use queue::Queue;
 pub use steda::{Steda, StedaBuilder};
 pub use task::{Spawn, SpawnedTask, Task, TaskHandle, TaskRef, TaskSnapshot};

src/worker.rs

@@ -269,6 +269,8 @@ fn is_transient_sqlstate(code: &str) -> bool {
 /// struct ProvisionedExecutor;
 ///
 /// impl TaskExecutor<i64, i64> for ProvisionedExecutor {
+///     type Error = steda::Error;
+///
 ///     fn execute(
 ///         &self,
 ///         input: i64,
@@ -300,6 +302,12 @@ fn is_transient_sqlstate(code: &str) -> bool {
 /// must ensure dropping its future also terminates or fences that work so it cannot keep
 /// acting as the no-longer-owned attempt.
 pub trait TaskExecutor<Input, Output>: Send + Sync + 'static {
+    /// Error returned by this executor.
+    ///
+    /// Steda converts this once at the worker boundary, allowing executor
+    /// implementations to use their own domain error type internally.
+    type Error: Into<Error> + Send + 'static;
+
     /// Execute one typed task attempt.
     ///
     /// # Errors
@@ -310,7 +318,7 @@ pub trait TaskExecutor<Input, Output>: Send + Sync + 'static {
         &self,
         input: Input,
         context: TaskContext,
-    ) -> impl Future<Output = Result<Output>> + Send;
+    ) -> impl Future<Output = std::result::Result<Output, Self::Error>> + Send;
 }
 
 /// Reusable in-process async handler accepted by [`WorkerBuilder::task`].
@@ -337,6 +345,8 @@ impl<Input, Output, F> TaskExecutor<Input, Output> for F
 where
     F: TaskHandler<Input, Output>,
 {
+    type Error = Error;
+
     fn execute(
         &self,
         input: Input,
@@ -488,7 +498,7 @@ impl WorkerBuilder {
             let executor = Arc::clone(&executor);
             Box::pin(async move {
                 let input = serde_json::from_value::<Input>(raw)?;
-                let output = executor.execute(input, context).await?;
+                let output = executor.execute(input, context).await.map_err(Into::into)?;
                 Ok(serde_json::to_value(output)?)
             })
         });

tests/task_executors.rs

@@ -48,6 +48,8 @@ mod tests {
     }
 
     impl TaskExecutor<(), ()> for CancellableExecutor {
+        type Error = Error;
+
         fn execute(
             &self,
             (): (),
@@ -72,6 +74,8 @@ mod tests {
     }
 
     impl TaskExecutor<i64, i64> for ProvisionedExecutor {
+        type Error = Error;
+
         fn execute(
             &self,
             input: i64,