From 16970bc06293b3dc422923ad77b254e18b3ea87f Mon Sep 17 00:00:00 2001
From: Christophe <cvauclair@protonmail.com>
Date: Fri, 6 Dec 2024 13:31:10 -0500
Subject: [PATCH] fix(pipelines): Type errors

---
 rig-core/src/pipeline/agent_ops.rs |  34 ++++-----
 rig-core/src/pipeline/mod.rs       |  44 ++++++------
 rig-core/src/pipeline/op.rs        | 106 ++++++++++++++++++-----------
 rig-core/src/pipeline/try_op.rs    |  65 +++++++++---------
 4 files changed, 137 insertions(+), 112 deletions(-)

diff --git a/rig-core/src/pipeline/agent_ops.rs b/rig-core/src/pipeline/agent_ops.rs
index f51ce713..c7562737 100644
--- a/rig-core/src/pipeline/agent_ops.rs
+++ b/rig-core/src/pipeline/agent_ops.rs
@@ -17,7 +17,7 @@ impl<I, In, T> Lookup<I, In, T>
 where
     I: vector_store::VectorStoreIndex,
 {
-    pub fn new(index: I, n: usize) -> Self {
+    pub(crate) fn new(index: I, n: usize) -> Self {
         Self {
             index,
             n,
@@ -66,7 +66,7 @@ pub struct Prompt<P, In> {
 }
 
 impl<P, In> Prompt<P, In> {
-    pub fn new(prompt: P) -> Self {
+    pub(crate) fn new(prompt: P) -> Self {
         Self {
             prompt,
             _in: std::marker::PhantomData,
@@ -96,21 +96,21 @@ where
     Prompt::new(prompt)
 }
 
-pub struct Extract<M, T, In>
+pub struct Extract<M, Input, Output>
 where
     M: CompletionModel,
-    T: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
+    Output: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
 {
-    extractor: Extractor<M, T>,
-    _in: std::marker::PhantomData<In>,
+    extractor: Extractor<M, Output>,
+    _in: std::marker::PhantomData<Input>,
 }
 
-impl<M, T, In> Extract<M, T, In>
+impl<M, Input, Output> Extract<M, Input, Output>
 where
     M: CompletionModel,
-    T: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
+    Output: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
 {
-    pub fn new(extractor: Extractor<M, T>) -> Self {
+    pub(crate) fn new(extractor: Extractor<M, Output>) -> Self {
         Self {
             extractor,
             _in: std::marker::PhantomData,
@@ -118,25 +118,25 @@ where
     }
 }
 
-impl<M, T, In> Op for Extract<M, T, In>
+impl<M, Input, Output> Op for Extract<M, Input, Output>
 where
     M: CompletionModel,
-    T: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
-    In: Into<String> + Send + Sync,
+    Output: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
+    Input: Into<String> + Send + Sync,
 {
-    type Input = In;
-    type Output = Result<T, ExtractionError>;
+    type Input = Input;
+    type Output = Result<Output, ExtractionError>;
 
     async fn call(&self, input: Self::Input) -> Self::Output {
         self.extractor.extract(&input.into()).await
     }
 }
 
-pub fn extract<M, T, In>(extractor: Extractor<M, T>) -> Extract<M, T, In>
+pub fn extract<M, Input, Output>(extractor: Extractor<M, Output>) -> Extract<M, Input, Output>
 where
     M: CompletionModel,
-    T: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
-    In: Into<String> + Send + Sync,
+    Output: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
+    Input: Into<String> + Send + Sync,
 {
     Extract::new(extractor)
 }
diff --git a/rig-core/src/pipeline/mod.rs b/rig-core/src/pipeline/mod.rs
index 417ae762..27ea5b1f 100644
--- a/rig-core/src/pipeline/mod.rs
+++ b/rig-core/src/pipeline/mod.rs
@@ -118,14 +118,14 @@ impl<E> PipelineBuilder<E> {
     /// let result = pipeline.call((1, 2)).await;
     /// assert_eq!(result, "Result: 3!");
     /// ```
-    pub fn map<F, In, T>(self, f: F) -> impl Op<Input = In, Output = T>
+    pub fn map<F, Input, Output>(self, f: F) -> op::Map<F, Input>
     where
-        F: Fn(In) -> T + Send + Sync,
-        In: Send + Sync,
-        T: Send + Sync,
+        F: Fn(Input) -> Output + Send + Sync,
+        Input: Send + Sync,
+        Output: Send + Sync,
         Self: Sized,
     {
-        map(f)
+        op::Map::new(f)
     }
 
     /// Same as `map` but for asynchronous functions
@@ -145,15 +145,15 @@ impl<E> PipelineBuilder<E> {
     /// let result = pipeline.call("bob@gmail.com".to_string()).await;
     /// assert_eq!(result, "Hello, bob!");
     /// ```
-    pub fn then<F, In, Fut>(self, f: F) -> impl Op<Input = In, Output = Fut::Output>
+    pub fn then<F, Input, Fut>(self, f: F) -> op::Then<F, Input>
     where
-        F: Fn(In) -> Fut + Send + Sync,
-        In: Send + Sync,
+        F: Fn(Input) -> Fut + Send + Sync,
+        Input: Send + Sync,
         Fut: Future + Send + Sync,
         Fut::Output: Send + Sync,
         Self: Sized,
     {
-        then(f)
+        op::Then::new(f)
     }
 
     /// Add an arbitrary operation to the current pipeline.
@@ -179,7 +179,7 @@ impl<E> PipelineBuilder<E> {
     /// let result = pipeline.call(1).await;
     /// assert_eq!(result, 2);
     /// ```
-    pub fn chain<T>(self, op: T) -> impl Op<Input = T::Input, Output = T::Output>
+    pub fn chain<T>(self, op: T) -> T
     where
         T: Op,
         Self: Sized,
@@ -203,19 +203,19 @@ impl<E> PipelineBuilder<E> {
     ///
     /// let result = pipeline.call("What is a flurbo?".to_string()).await;
     /// ```
-    pub fn lookup<I, In, T>(
+    pub fn lookup<I, Input, Output>(
         self,
         index: I,
         n: usize,
-    ) -> impl Op<Input = In, Output = Result<Vec<T>, vector_store::VectorStoreError>>
+    ) -> agent_ops::Lookup<I, Input, Output>
     where
         I: vector_store::VectorStoreIndex,
-        T: Send + Sync + for<'a> serde::Deserialize<'a>,
-        In: Into<String> + Send + Sync,
+        Output: Send + Sync + for<'a> serde::Deserialize<'a>,
+        Input: Into<String> + Send + Sync,
         // E: From<vector_store::VectorStoreError> + Send + Sync,
         Self: Sized,
     {
-        agent_ops::lookup(index, n)
+        agent_ops::Lookup::new(index, n)
     }
 
     /// Add a prompt operation to the current pipeline/op. The prompt operation expects the
@@ -235,14 +235,14 @@ impl<E> PipelineBuilder<E> {
     ///
     /// let result = pipeline.call("Alice".to_string()).await;
     /// ```
-    pub fn prompt<P, In>(self, agent: P) -> agent_ops::Prompt<P, In>
+    pub fn prompt<P, Input>(self, agent: P) -> agent_ops::Prompt<P, Input>
     where
         P: completion::Prompt,
-        In: Into<String> + Send + Sync,
+        Input: Into<String> + Send + Sync,
         // E: From<completion::PromptError> + Send + Sync,
         Self: Sized,
     {
-        agent_ops::prompt(agent)
+        agent_ops::Prompt::new(agent)
     }
 
     /// Add an extract operation to the current pipeline/op. The extract operation expects the
@@ -268,13 +268,13 @@ impl<E> PipelineBuilder<E> {
     /// let result: Sentiment = pipeline.call("I love ice cream!".to_string()).await?;
     /// assert!(result.score > 0.5);
     /// ```
-    pub fn extract<M, T, In>(self, extractor: Extractor<M, T>) -> agent_ops::Extract<M, T, In>
+    pub fn extract<M, Input, Output>(self, extractor: Extractor<M, Output>) -> agent_ops::Extract<M, Input, Output>
     where
         M: completion::CompletionModel,
-        T: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
-        In: Into<String> + Send + Sync,
+        Output: schemars::JsonSchema + for<'a> serde::Deserialize<'a> + Send + Sync,
+        Input: Into<String> + Send + Sync,
     {
-        agent_ops::extract(extractor)
+        agent_ops::Extract::new(extractor)
     }
 }
 
diff --git a/rig-core/src/pipeline/op.rs b/rig-core/src/pipeline/op.rs
index 0ddf32bf..96ee67e8 100644
--- a/rig-core/src/pipeline/op.rs
+++ b/rig-core/src/pipeline/op.rs
@@ -2,7 +2,7 @@ use std::future::Future;
 
 #[allow(unused_imports)] // Needed since this is used in a macro rule
 use futures::join;
-use futures::{stream, StreamExt};
+use futures::stream;
 
 // ================================================================
 // Core Op trait
@@ -21,6 +21,8 @@ pub trait Op: Send + Sync {
         I::IntoIter: Send,
         Self: Sized,
     {
+        use futures::stream::StreamExt;
+        
         async move {
             stream::iter(input)
                 .map(|input| self.call(input))
@@ -43,13 +45,13 @@ pub trait Op: Send + Sync {
     /// let result = chain.call((1, 2)).await;
     /// assert_eq!(result, "Result: 3!");
     /// ```
-    fn map<F, T>(self, f: F) -> impl Op<Input = Self::Input, Output = T>
+    fn map<F, Input>(self, f: F) -> Sequential<Self, Map<F, Self::Output>>
     where
-        F: Fn(Self::Output) -> T + Send + Sync,
-        T: Send + Sync,
+        F: Fn(Self::Output) -> Input + Send + Sync,
+        Input: Send + Sync,
         Self: Sized,
     {
-        Sequential::new(self, map(f))
+        Sequential::new(self, Map::new(f))
     }
 
     /// Same as `map` but for asynchronous functions
@@ -69,14 +71,14 @@ pub trait Op: Send + Sync {
     /// let result = chain.call("bob@gmail.com".to_string()).await;
     /// assert_eq!(result, "Hello, bob!");
     /// ```
-    fn then<F, Fut>(self, f: F) -> impl Op<Input = Self::Input, Output = Fut::Output>
+    fn then<F, Fut>(self, f: F) -> Sequential<Self, Then<F, Fut::Output>>
     where
         F: Fn(Self::Output) -> Fut + Send + Sync,
         Fut: Future + Send + Sync,
         Fut::Output: Send + Sync,
         Self: Sized,
     {
-        Sequential::new(self, then(f))
+        Sequential::new(self, Then::new(f))
     }
 
     /// Chain an arbitrary operation to the current op.
@@ -102,7 +104,7 @@ pub trait Op: Send + Sync {
     /// let result = chain.call(1).await;
     /// assert_eq!(result, 2);
     /// ```
-    fn chain<T>(self, op: T) -> impl Op<Input = Self::Input, Output = T::Output>
+    fn chain<T>(self, op: T) -> Sequential<Self, T>
     where
         T: Op<Input = Self::Output>,
         Self: Sized,
@@ -126,14 +128,14 @@ pub trait Op: Send + Sync {
     ///
     /// let result = chain.call("What is a flurbo?".to_string()).await;
     /// ```
-    fn lookup<I, T>(
+    fn lookup<I, Input>(
         self,
         index: I,
         n: usize,
-    ) -> impl Op<Input = Self::Input, Output = Result<Vec<T>, vector_store::VectorStoreError>>
+    ) -> Sequential<Self, Lookup<I, Self::Output, Input>>
     where
         I: vector_store::VectorStoreIndex,
-        T: Send + Sync + for<'a> serde::Deserialize<'a>,
+        Input: Send + Sync + for<'a> serde::Deserialize<'a>,
         Self::Output: Into<String>,
         Self: Sized,
     {
@@ -160,7 +162,7 @@ pub trait Op: Send + Sync {
     fn prompt<P>(
         self,
         prompt: P,
-    ) -> impl Op<Input = Self::Input, Output = Result<String, completion::PromptError>>
+    ) -> Sequential<Self, Prompt<P, Self::Output>>
     where
         P: completion::Prompt,
         Self::Output: Into<String>,
@@ -189,7 +191,7 @@ pub struct Sequential<Op1, Op2> {
 }
 
 impl<Op1, Op2> Sequential<Op1, Op2> {
-    pub fn new(prev: Op1, op: Op2) -> Self {
+    pub(crate) fn new(prev: Op1, op: Op2) -> Self {
         Self { prev, op }
     }
 }
@@ -216,13 +218,13 @@ use super::agent_ops::{Lookup, Prompt};
 // ================================================================
 // Core Op implementations
 // ================================================================
-pub struct Map<F, T> {
+pub struct Map<F, Input> {
     f: F,
-    _t: std::marker::PhantomData<T>,
+    _t: std::marker::PhantomData<Input>,
 }
 
-impl<F, T> Map<F, T> {
-    pub fn new(f: F) -> Self {
+impl<F, Input> Map<F, Input> {
+    pub(crate) fn new(f: F) -> Self {
         Self {
             f,
             _t: std::marker::PhantomData,
@@ -230,14 +232,14 @@ impl<F, T> Map<F, T> {
     }
 }
 
-impl<F, T, Out> Op for Map<F, T>
+impl<F, Input, Output> Op for Map<F, Input>
 where
-    F: Fn(T) -> Out + Send + Sync,
-    T: Send + Sync,
-    Out: Send + Sync,
+    F: Fn(Input) -> Output + Send + Sync,
+    Input: Send + Sync,
+    Output: Send + Sync,
 {
-    type Input = T;
-    type Output = Out;
+    type Input = Input;
+    type Output = Output;
 
     #[inline]
     async fn call(&self, input: Self::Input) -> Self::Output {
@@ -245,29 +247,53 @@ where
     }
 }
 
-pub fn map<F, T, Out>(f: F) -> impl Op<Input = T, Output = Out>
+pub fn map<F, Input, Output>(f: F) -> Map<F, Input>
 where
-    F: Fn(T) -> Out + Send + Sync,
-    T: Send + Sync,
-    Out: Send + Sync,
+    F: Fn(Input) -> Output + Send + Sync,
+    Input: Send + Sync,
+    Output: Send + Sync,
 {
     Map::new(f)
 }
 
-pub fn passthrough<T>() -> impl Op<Input = T, Output = T>
+pub struct Passthrough<T> {
+    _t: std::marker::PhantomData<T>,
+}
+
+impl<T> Passthrough<T> {
+    pub(crate) fn new() -> Self {
+        Self {
+            _t: std::marker::PhantomData,
+        }
+    }
+}
+
+impl<T> Op for Passthrough<T> 
+where 
+    T: Send + Sync,
+{
+    type Input = T;
+    type Output = T;
+
+    async fn call(&self, input: Self::Input) -> Self::Output {
+        input
+    }
+}
+
+pub fn passthrough<T>() -> Passthrough<T>
 where
     T: Send + Sync,
 {
-    Map::new(|x| x)
+    Passthrough::new()
 }
 
-pub struct Then<F, T> {
+pub struct Then<F, Input> {
     f: F,
-    _t: std::marker::PhantomData<T>,
+    _t: std::marker::PhantomData<Input>,
 }
 
-impl<F, T> Then<F, T> {
-    fn new(f: F) -> Self {
+impl<F, Input> Then<F, Input> {
+    pub(crate) fn new(f: F) -> Self {
         Self {
             f,
             _t: std::marker::PhantomData,
@@ -275,14 +301,14 @@ impl<F, T> Then<F, T> {
     }
 }
 
-impl<F, T, Fut> Op for Then<F, T>
+impl<F, Input, Fut> Op for Then<F, Input>
 where
-    F: Fn(T) -> Fut + Send + Sync,
-    T: Send + Sync,
+    F: Fn(Input) -> Fut + Send + Sync,
+    Input: Send + Sync,
     Fut: Future + Send,
     Fut::Output: Send + Sync,
 {
-    type Input = T;
+    type Input = Input;
     type Output = Fut::Output;
 
     #[inline]
@@ -291,10 +317,10 @@ where
     }
 }
 
-pub fn then<F, T, Fut>(f: F) -> impl Op<Input = T, Output = Fut::Output>
+pub fn then<F, Input, Fut>(f: F) -> Then<F, Input>
 where
-    F: Fn(T) -> Fut + Send + Sync,
-    T: Send + Sync,
+    F: Fn(Input) -> Fut + Send + Sync,
+    Input: Send + Sync,
     Fut: Future + Send,
     Fut::Output: Send + Sync,
 {
diff --git a/rig-core/src/pipeline/try_op.rs b/rig-core/src/pipeline/try_op.rs
index e8d276e5..76b41e91 100644
--- a/rig-core/src/pipeline/try_op.rs
+++ b/rig-core/src/pipeline/try_op.rs
@@ -2,9 +2,9 @@ use std::future::Future;
 
 #[allow(unused_imports)] // Needed since this is used in a macro rule
 use futures::try_join;
-use futures::{stream, StreamExt, TryStreamExt};
+use futures::stream;
 
-use super::op::{self, map, then};
+use super::op::{self};
 
 // ================================================================
 // Core TryOp trait
@@ -46,6 +46,8 @@ pub trait TryOp: Send + Sync {
         I::IntoIter: Send,
         Self: Sized,
     {
+        use stream::{StreamExt, TryStreamExt};
+        
         async move {
             stream::iter(input)
                 .map(|input| self.try_call(input))
@@ -69,13 +71,13 @@ pub trait TryOp: Send + Sync {
     /// let result = op.try_call(2).await;
     /// assert_eq!(result, Ok(4));
     /// ```
-    fn map_ok<F, T>(self, f: F) -> impl op::Op<Input = Self::Input, Output = Result<T, Self::Error>>
+    fn map_ok<F, Output>(self, f: F) -> MapOk<Self, op::Map<F, Self::Output>>
     where
-        F: Fn(Self::Output) -> T + Send + Sync,
-        T: Send + Sync,
+        F: Fn(Self::Output) -> Output + Send + Sync,
+        Output: Send + Sync,
         Self: Sized,
     {
-        MapOk::new(self, map(f))
+        MapOk::new(self, op::Map::new(f))
     }
 
     /// Map the error return value (i.e., `Err`) of the current op to a different value
@@ -95,13 +97,13 @@ pub trait TryOp: Send + Sync {
     fn map_err<F, E>(
         self,
         f: F,
-    ) -> impl op::Op<Input = Self::Input, Output = Result<Self::Output, E>>
+    ) -> MapErr<Self, op::Map<F, Self::Error>>
     where
         F: Fn(Self::Error) -> E + Send + Sync,
         E: Send + Sync,
         Self: Sized,
     {
-        MapErr::new(self, map(f))
+        MapErr::new(self, op::Map::new(f))
     }
 
     /// Chain a function to the current op. The function will only be called
@@ -119,17 +121,17 @@ pub trait TryOp: Send + Sync {
     /// let result = op.try_call(2).await;
     /// assert_eq!(result, Ok(4));
     /// ```
-    fn and_then<F, Fut, T>(
+    fn and_then<F, Fut, Output>(
         self,
         f: F,
-    ) -> impl TryOp<Input = Self::Input, Output = T, Error = Self::Error>
+    ) -> AndThen<Self, op::Then<F, Self::Output>>
     where
         F: Fn(Self::Output) -> Fut + Send + Sync,
-        Fut: Future<Output = Result<T, Self::Error>> + Send + Sync,
-        T: Send + Sync,
+        Fut: Future<Output = Result<Output, Self::Error>> + Send + Sync,
+        Output: Send + Sync,
         Self: Sized,
     {
-        AndThen::new(self, then(f))
+        AndThen::new(self, op::Then::new(f))
     }
 
     /// Chain a function `f` to the current op. The function `f` will only be called
@@ -150,14 +152,14 @@ pub trait TryOp: Send + Sync {
     fn or_else<F, Fut, E>(
         self,
         f: F,
-    ) -> impl TryOp<Input = Self::Input, Output = Self::Output, Error = E>
+    ) -> OrElse<Self, op::Then<F, Self::Error>>
     where
         F: Fn(Self::Error) -> Fut + Send + Sync,
         Fut: Future<Output = Result<Self::Output, E>> + Send + Sync,
         E: Send + Sync,
         Self: Sized,
     {
-        OrElse::new(self, then(f))
+        OrElse::new(self, op::Then::new(f))
     }
 
     /// Chain a new op `op` to the current op. The new op will be called with the success
@@ -189,7 +191,7 @@ pub trait TryOp: Send + Sync {
     fn chain_ok<T>(
         self,
         op: T,
-    ) -> impl TryOp<Input = Self::Input, Output = T::Output, Error = Self::Error>
+    ) -> TrySequential<Self, T>
     where
         T: op::Op<Input = Self::Output>,
         Self: Sized,
@@ -222,7 +224,7 @@ pub struct MapOk<Op1, Op2> {
 }
 
 impl<Op1, Op2> MapOk<Op1, Op2> {
-    pub fn new(prev: Op1, op: Op2) -> Self {
+    pub(crate) fn new(prev: Op1, op: Op2) -> Self {
         Self { prev, op }
     }
 }
@@ -250,7 +252,7 @@ pub struct MapErr<Op1, Op2> {
 }
 
 impl<Op1, Op2> MapErr<Op1, Op2> {
-    pub fn new(prev: Op1, op: Op2) -> Self {
+    pub(crate) fn new(prev: Op1, op: Op2) -> Self {
         Self { prev, op }
     }
 }
@@ -279,22 +281,21 @@ pub struct AndThen<Op1, Op2> {
 }
 
 impl<Op1, Op2> AndThen<Op1, Op2> {
-    pub fn new(prev: Op1, op: Op2) -> Self {
+    pub(crate) fn new(prev: Op1, op: Op2) -> Self {
         Self { prev, op }
     }
 }
 
-impl<Op1, Op2> TryOp for AndThen<Op1, Op2>
+impl<Op1, Op2> op::Op for AndThen<Op1, Op2>
 where
     Op1: TryOp,
     Op2: TryOp<Input = Op1::Output, Error = Op1::Error>,
 {
     type Input = Op1::Input;
-    type Output = Op2::Output;
-    type Error = Op1::Error;
+    type Output = Result<Op2::Output, Op1::Error>;
 
     #[inline]
-    async fn try_call(&self, input: Self::Input) -> Result<Self::Output, Self::Error> {
+    async fn call(&self, input: Self::Input) -> Self::Output {
         let output = self.prev.try_call(input).await?;
         self.op.try_call(output).await
     }
@@ -306,22 +307,21 @@ pub struct OrElse<Op1, Op2> {
 }
 
 impl<Op1, Op2> OrElse<Op1, Op2> {
-    pub fn new(prev: Op1, op: Op2) -> Self {
+    pub(crate) fn new(prev: Op1, op: Op2) -> Self {
         Self { prev, op }
     }
 }
 
-impl<Op1, Op2> TryOp for OrElse<Op1, Op2>
+impl<Op1, Op2> op::Op for OrElse<Op1, Op2>
 where
     Op1: TryOp,
     Op2: TryOp<Input = Op1::Error, Output = Op1::Output>,
 {
     type Input = Op1::Input;
-    type Output = Op1::Output;
-    type Error = Op2::Error;
+    type Output = Result<Op1::Output, Op2::Error>;
 
     #[inline]
-    async fn try_call(&self, input: Self::Input) -> Result<Self::Output, Self::Error> {
+    async fn call(&self, input: Self::Input) -> Self::Output {
         match self.prev.try_call(input).await {
             Ok(output) => Ok(output),
             Err(err) => self.op.try_call(err).await,
@@ -335,22 +335,21 @@ pub struct TrySequential<Op1, Op2> {
 }
 
 impl<Op1, Op2> TrySequential<Op1, Op2> {
-    pub fn new(prev: Op1, op: Op2) -> Self {
+    pub(crate) fn new(prev: Op1, op: Op2) -> Self {
         Self { prev, op }
     }
 }
 
-impl<Op1, Op2> TryOp for TrySequential<Op1, Op2>
+impl<Op1, Op2> op::Op for TrySequential<Op1, Op2>
 where
     Op1: TryOp,
     Op2: op::Op<Input = Op1::Output>,
 {
     type Input = Op1::Input;
-    type Output = Op2::Output;
-    type Error = Op1::Error;
+    type Output = Result<Op2::Output, Op1::Error>;
 
     #[inline]
-    async fn try_call(&self, input: Self::Input) -> Result<Self::Output, Self::Error> {
+    async fn call(&self, input: Self::Input) -> Self::Output {
         match self.prev.try_call(input).await {
             Ok(output) => Ok(self.op.call(output).await),
             Err(err) => Err(err),