refactor(trogon-nats): zero-wrapping JetStream trait alignment with async_nats

- publish_with_headers returns async_nats PublishAckFuture directly via IntoFuture
- get_or_create_stream returns async_nats Stream directly via associated type
- Both use async_nats error types directly — no mapping
- Renamed js_publish_with_headers to publish_with_headers
- Removed JsPublishAckFuture wrapper and ReadyAckFuture
- Mocks use std::future::Ready, no custom future types

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

rsworkspace/crates/acp-nats/src/agent/js_request.rs

@@ -62,9 +62,11 @@ where
     headers.insert(REQ_ID_HEADER, req_id);
     headers.insert(SESSION_ID_HEADER, session_id);
 
-    js.js_publish_with_headers(subject.to_string(), headers, Bytes::from(payload_bytes))
+    js.publish_with_headers(subject.to_string(), headers, Bytes::from(payload_bytes))
         .await
-        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js publish: {e}")))?;
+        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js publish: {e}")))?
+        .await
+        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js ack: {e}")))?;
 
     match timeout(operation_timeout, resp_messages.next()).await {
         Ok(Some(Ok(js_msg))) => {

rsworkspace/crates/acp-nats/src/agent/prompt.rs

@@ -262,9 +262,11 @@ where
     headers.insert(SESSION_ID_HEADER, sid);
 
     let prompt_subject = session::agent::prompt(prefix, sid);
-    js.js_publish_with_headers(prompt_subject, headers, Bytes::from(payload_bytes))
+    js.publish_with_headers(prompt_subject, headers, Bytes::from(payload_bytes))
         .await
-        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js publish: {e}")))?;
+        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js publish: {e}")))?
+        .await
+        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js ack: {e}")))?;
 
     let op_timeout = bridge.config.prompt_timeout();
 

rsworkspace/crates/acp-nats/src/agent/test_support.rs

@@ -41,15 +41,17 @@ impl MockJs {
 
 impl trogon_nats::jetstream::JetStreamPublisher for MockJs {
     type PublishError = trogon_nats::mocks::MockError;
+    type AckFuture =
+        std::future::Ready<Result<async_nats::jetstream::publish::PublishAck, Self::PublishError>>;
 
-    async fn js_publish_with_headers<S: async_nats::subject::ToSubject + Send>(
+    async fn publish_with_headers<S: async_nats::subject::ToSubject + Send>(
         &self,
         subject: S,
         headers: async_nats::HeaderMap,
         payload: bytes::Bytes,
-    ) -> Result<async_nats::jetstream::publish::PublishAck, Self::PublishError> {
+    ) -> Result<Self::AckFuture, Self::PublishError> {
         self.publisher
-            .js_publish_with_headers(subject, headers, payload)
+            .publish_with_headers(subject, headers, payload)
             .await
     }
 }

rsworkspace/crates/trogon-nats/src/jetstream/client.rs

@@ -2,7 +2,6 @@ use async_nats::HeaderMap;
 use async_nats::jetstream;
 use async_nats::jetstream::AckKind;
 use async_nats::jetstream::consumer::pull;
-use async_nats::jetstream::publish::PublishAck;
 use async_nats::jetstream::stream;
 use async_nats::subject::ToSubject;
 use bytes::Bytes;
@@ -40,35 +39,33 @@ impl std::fmt::Display for JetStreamError {
 impl std::error::Error for JetStreamError {}
 
 impl JetStreamContext for NatsJetStreamClient {
-    type Error = JetStreamError;
+    type Error = async_nats::jetstream::context::CreateStreamError;
+    type Stream = jetstream::stream::Stream;
 
     async fn get_or_create_stream<S: Into<stream::Config> + Send>(
         &self,
         config: S,
-    ) -> Result<(), JetStreamError> {
-        self.context
-            .get_or_create_stream(config)
-            .await
-            .map(|_| ())
-            .map_err(|e| JetStreamError(e.to_string()))
+    ) -> Result<jetstream::stream::Stream, async_nats::jetstream::context::CreateStreamError> {
+        self.context.get_or_create_stream(config).await
     }
 }
 
+pub type PublishError = async_nats::jetstream::context::PublishError;
+pub type PublishAckFuture = async_nats::jetstream::context::PublishAckFuture;
+
 impl JetStreamPublisher for NatsJetStreamClient {
-    type PublishError = JetStreamError;
+    type PublishError = PublishError;
+    type AckFuture = PublishAckFuture;
 
-    async fn js_publish_with_headers<S: ToSubject + Send>(
+    async fn publish_with_headers<S: ToSubject + Send>(
         &self,
         subject: S,
         headers: HeaderMap,
         payload: Bytes,
-    ) -> Result<PublishAck, JetStreamError> {
+    ) -> Result<PublishAckFuture, PublishError> {
         self.context
             .publish_with_headers(subject, headers, payload)
             .await
-            .map_err(|e| JetStreamError(e.to_string()))?
-            .await
-            .map_err(|e| JetStreamError(e.to_string()))
     }
 }
 

rsworkspace/crates/trogon-nats/src/jetstream/mocks.rs

@@ -171,6 +171,7 @@ impl Default for MockJetStreamContext {
 
 impl JetStreamContext for MockJetStreamContext {
     type Error = MockError;
+    type Stream = ();
 
     async fn get_or_create_stream<S: Into<stream::Config> + Send>(
         &self,
@@ -254,13 +255,14 @@ impl Default for MockJetStreamPublisher {
 
 impl JetStreamPublisher for MockJetStreamPublisher {
     type PublishError = MockError;
+    type AckFuture = std::future::Ready<Result<PublishAck, MockError>>;
 
-    async fn js_publish_with_headers<S: ToSubject + Send>(
+    async fn publish_with_headers<S: ToSubject + Send>(
         &self,
         subject: S,
         headers: HeaderMap,
         payload: Bytes,
-    ) -> Result<PublishAck, MockError> {
+    ) -> Result<Self::AckFuture, MockError> {
         let subject = subject.to_subject().to_string();
         let should_fail = {
             let mut count = self.publish_fail_count.lock().unwrap();
@@ -288,13 +290,13 @@ impl JetStreamPublisher for MockJetStreamPublisher {
             current
         };
 
-        Ok(PublishAck {
+        Ok(std::future::ready(Ok(PublishAck {
             stream: "mock-stream".to_string(),
             sequence: seq,
             domain: String::new(),
             duplicate: false,
             value: None,
-        })
+        })))
     }
 }
 
@@ -478,12 +480,14 @@ mod tests {
     async fn mock_publisher_records_publishes() {
         let pub_mock = MockJetStreamPublisher::new();
         let ack = pub_mock
-            .js_publish_with_headers(
+            .publish_with_headers(
                 "test.subject".to_string(),
                 HeaderMap::new(),
                 Bytes::from("hello"),
             )
             .await
+            .unwrap()
+            .await
             .unwrap();
         assert_eq!(ack.sequence, 1);
         assert_eq!(pub_mock.published_subjects(), vec!["test.subject"]);
@@ -494,11 +498,15 @@ mod tests {
     async fn mock_publisher_increments_sequence() {
         let pub_mock = MockJetStreamPublisher::new();
         let ack1 = pub_mock
-            .js_publish_with_headers("a".to_string(), HeaderMap::new(), Bytes::new())
+            .publish_with_headers("a".to_string(), HeaderMap::new(), Bytes::new())
+            .await
+            .unwrap()
             .await
             .unwrap();
         let ack2 = pub_mock
-            .js_publish_with_headers("b".to_string(), HeaderMap::new(), Bytes::new())
+            .publish_with_headers("b".to_string(), HeaderMap::new(), Bytes::new())
+            .await
+            .unwrap()
             .await
             .unwrap();
         assert_eq!(ack1.sequence, 1);
@@ -510,12 +518,12 @@ mod tests {
         let pub_mock = MockJetStreamPublisher::new();
         pub_mock.fail_next_js_publish();
         let result = pub_mock
-            .js_publish_with_headers("test".to_string(), HeaderMap::new(), Bytes::new())
+            .publish_with_headers("test".to_string(), HeaderMap::new(), Bytes::new())
             .await;
         assert!(result.is_err());
 
         let result = pub_mock
-            .js_publish_with_headers("test".to_string(), HeaderMap::new(), Bytes::new())
+            .publish_with_headers("test".to_string(), HeaderMap::new(), Bytes::new())
             .await;
         assert!(result.is_ok());
     }
@@ -589,19 +597,21 @@ mod tests {
         pub_mock.fail_js_publish_count(2);
         assert!(
             pub_mock
-                .js_publish_with_headers("a".to_string(), HeaderMap::new(), Bytes::new())
+                .publish_with_headers("a".to_string(), HeaderMap::new(), Bytes::new())
                 .await
                 .is_err()
         );
         assert!(
             pub_mock
-                .js_publish_with_headers("b".to_string(), HeaderMap::new(), Bytes::new())
+                .publish_with_headers("b".to_string(), HeaderMap::new(), Bytes::new())
                 .await
                 .is_err()
         );
         assert!(
             pub_mock
-                .js_publish_with_headers("c".to_string(), HeaderMap::new(), Bytes::new())
+                .publish_with_headers("c".to_string(), HeaderMap::new(), Bytes::new())
+                .await
+                .unwrap()
                 .await
                 .is_ok()
         );

rsworkspace/crates/trogon-nats/src/jetstream/mod.rs

@@ -7,7 +7,10 @@ pub mod traits;
 pub mod mocks;
 
 #[cfg(not(coverage))]
-pub use client::{JetStreamError, NatsJetStreamClient, NatsJetStreamConsumer, NatsJsMessage};
+pub use client::{
+    JetStreamError, NatsJetStreamClient, NatsJetStreamConsumer, NatsJsMessage, PublishAckFuture,
+    PublishError,
+};
 pub use message::{
     JsAck, JsAckWith, JsDispatchMessage, JsDoubleAck, JsDoubleAckWith, JsMessageRef,
     JsRequestMessage,

rsworkspace/crates/trogon-nats/src/jetstream/traits.rs

@@ -1,5 +1,5 @@
 use std::error::Error;
-use std::future::Future;
+use std::future::{Future, IntoFuture};
 
 use async_nats::HeaderMap;
 use async_nats::jetstream::consumer::pull;
@@ -11,22 +11,24 @@ use futures::Stream;
 
 pub trait JetStreamContext: Send + Sync + Clone + 'static {
     type Error: Error + Send + Sync;
+    type Stream: Send;
 
     fn get_or_create_stream<S: Into<stream::Config> + Send>(
         &self,
         config: S,
-    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
+    ) -> impl Future<Output = Result<Self::Stream, Self::Error>> + Send;
 }
 
 pub trait JetStreamPublisher: Send + Sync + Clone + 'static {
     type PublishError: Error + Send + Sync;
+    type AckFuture: IntoFuture<Output = Result<PublishAck, Self::PublishError>> + Send;
 
-    fn js_publish_with_headers<S: ToSubject + Send>(
+    fn publish_with_headers<S: ToSubject + Send>(
         &self,
         subject: S,
         headers: HeaderMap,
         payload: Bytes,
-    ) -> impl Future<Output = Result<PublishAck, Self::PublishError>> + Send;
+    ) -> impl Future<Output = Result<Self::AckFuture, Self::PublishError>> + Send;
 }
 
 pub trait JetStreamConsumerFactory: Send + Sync + Clone + 'static {
@@ -64,13 +66,14 @@ impl Error for NoJetStream {}
 
 impl JetStreamPublisher for () {
     type PublishError = NoJetStream;
+    type AckFuture = std::future::Ready<Result<PublishAck, NoJetStream>>;
 
-    async fn js_publish_with_headers<S: ToSubject + Send>(
+    async fn publish_with_headers<S: ToSubject + Send>(
         &self,
         _subject: S,
         _headers: HeaderMap,
         _payload: Bytes,
-    ) -> Result<PublishAck, NoJetStream> {
+    ) -> Result<Self::AckFuture, NoJetStream> {
         Err(NoJetStream)
     }
 }
@@ -194,7 +197,7 @@ mod tests {
 
     #[tokio::test]
     async fn unit_publisher_returns_err() {
-        let result = ().js_publish_with_headers("s", HeaderMap::new(), Bytes::new()).await;
+        let result = ().publish_with_headers("s", HeaderMap::new(), Bytes::new()).await;
         assert!(result.is_err());
     }