feldera
diff --git a/‎crates/adapters/src/controller.rs‎
Lines changed: 12 additions & 11 deletions b/‎crates/adapters/src/controller.rs‎
Lines changed: 12 additions & 11 deletions
@@ -5701,7 +5701,7 @@ impl ControllerInner {
         let encoder = if let Some(mut endpoint) = endpoint {
             endpoint
                 .connect(Box::new(
-                    move |fatal: bool, e: AnyError, error_tag: Option<&'static str>| {
+                    move |fatal: bool, e: AnyError, error_tag: Option<&str>| {
                         if let Some(controller) = self_weak.upgrade() {
                             controller.output_transport_error(
                                 endpoint_id,
@@ -6028,15 +6028,15 @@ impl ControllerInner {
         endpoint_name: &str,
     ) -> Result<ExternalInputEndpointStatus, ControllerError> {
         let endpoint_id = self.input_endpoint_id_by_name(endpoint_name)?;
-        Ok(self.status.input_status()[&endpoint_id].to_api_type())
+        Ok(self.status.input_status()[&endpoint_id].to_api_type(true))
     }
 
     fn output_endpoint_status(
         &self,
         endpoint_name: &str,
     ) -> Result<ExternalOutputEndpointStatus, ControllerError> {
         let endpoint_id = self.output_endpoint_id_by_name(endpoint_name)?;
-        Ok(self.status.output_status()[&endpoint_id].to_api_type())
+        Ok(self.status.output_status()[&endpoint_id].to_api_type(true))
     }
 
     fn send_command(&self, command: Command) {
@@ -6059,10 +6059,10 @@ impl ControllerInner {
         endpoint_name: &str,
         fatal: bool,
         error: AnyError,
-        tag: Option<&'static str>,
+        tag: Option<&str>,
     ) {
         self.status
-            .input_transport_error(endpoint_id, fatal, &error);
+            .input_transport_error(endpoint_id, fatal, tag, &error);
         let tag = tag.map(|tag| format!("{endpoint_name}-{tag}"));
         self.error(
             Arc::new(ControllerError::input_transport_error(
@@ -6075,7 +6075,8 @@ impl ControllerInner {
     }
 
     pub fn parse_error(&self, endpoint_id: EndpointId, endpoint_name: &str, error: ParseError) {
-        self.status.parse_error(endpoint_id);
+        let tag = error.get_error_tag();
+        self.status.parse_error(endpoint_id, tag.as_deref(), &error);
 
         let tag = error
             .get_error_tag()
@@ -6092,9 +6093,9 @@ impl ControllerInner {
         endpoint_id: EndpointId,
         endpoint_name: &str,
         error: AnyError,
-        tag: Option<&'static str>,
+        tag: Option<&str>,
     ) {
-        self.status.encode_error(endpoint_id);
+        self.status.encode_error(endpoint_id, tag, &error);
         let tag = tag.map(|tag| format!("{endpoint_name}-{tag}"));
         self.error(
             Arc::new(ControllerError::encode_error(endpoint_name, error)),
@@ -6111,10 +6112,10 @@ impl ControllerInner {
         endpoint_name: &str,
         fatal: bool,
         error: AnyError,
-        tag: Option<&'static str>,
+        tag: Option<&str>,
     ) {
         self.status
-            .output_transport_error(endpoint_id, fatal, &error);
+            .output_transport_error(endpoint_id, fatal, tag, &error);
         let tag = tag.map(|tag| format!("{endpoint_name}-{tag}"));
         self.error(
             Arc::new(ControllerError::output_transport_error(
@@ -6682,7 +6683,7 @@ impl InputConsumer for InputProbe {
         self.transaction_in_progress.store(false, Ordering::Release);
     }
 
-    fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>) {
+    fn error(&self, fatal: bool, error: AnyError, tag: Option<&str>) {
         self.controller.input_transport_error(
             self.endpoint_id,
             &self.endpoint_name,