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

Every production trait impl is a direct passthrough to async_nats:
- No newtypes, no wrappers, no BoxStream, no error mapping
- Traits implemented directly on async_nats types (Stream, Consumer, Message)
- All associated types are async_nats types
- Removed NatsJsMessage, NatsJetStreamStream, NatsJetStreamConsumer newtypes
- Removed JetStreamError entirely
- Fixed swapped StreamError/MessagesError naming to match async_nats
- Split JetStreamConsumerFactory into JetStreamGetStream + JetStreamCreateConsumer
- Renamed js_publish_with_headers to publish_with_headers

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

Author: yordis

rsworkspace/Cargo.lock

@@ -112,8 +112,8 @@ where
         impl std::future::Future<Output = Result<(), ConnectionError>>,
     )
     where
-        J: JetStreamConsumerFactory + 'static,
-        <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsDispatchMessage,
+        J: JetStreamGetStream + 'static,
+        <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsDispatchMessage,
     {
         let nats_for_serve = nats.clone();
         let nats_for_js = nats.clone();
@@ -422,7 +422,8 @@ where
 }
 
 use trogon_nats::jetstream::{
-    JetStreamConsumer as _, JetStreamConsumerFactory, JsAckWith, JsDispatchMessage,
+    JetStreamConsumer as _, JetStreamCreateConsumer as _, JetStreamGetStream, JsAckWith,
+    JsDispatchMessage,
 };
 
 async fn handle_request_with_keepalive<N, Resp, ReqT, F, M>(
@@ -484,17 +485,22 @@ async fn serve_js<N, J, A>(
 ) -> Result<(), ConnectionError>
 where
     N: PublishClient + FlushClient + Clone + 'static,
-    J: JetStreamConsumerFactory + 'static,
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsDispatchMessage,
+    J: JetStreamGetStream + 'static,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsDispatchMessage,
     A: Agent + 'static,
 {
     let stream_name = acp_nats::jetstream::streams::commands_stream_name(prefix);
     let config = acp_nats::jetstream::consumers::commands_observer();
 
     info!(stream = %stream_name, "Starting JetStream consumer for COMMANDS stream");
 
-    let consumer = js
-        .create_consumer(&stream_name, config)
+    let stream = js
+        .get_stream(&stream_name)
+        .await
+        .map_err(|e| ConnectionError::JetStream(Box::new(e)))?;
+
+    let consumer = stream
+        .create_consumer(config)
         .await
         .map_err(|e| ConnectionError::JetStream(Box::new(e)))?;
 

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

@@ -21,7 +21,7 @@ use opentelemetry::metrics::Meter;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tracing::{info, warn};
-use trogon_nats::jetstream::{JetStreamConsumerFactory, JetStreamPublisher, JsRequestMessage};
+use trogon_nats::jetstream::{JetStreamGetStream, JetStreamPublisher, JsRequestMessage};
 use trogon_std::time::GetElapsed;
 
 use super::{
@@ -150,10 +150,10 @@ async fn publish_session_ready<N: PublishClient + FlushClient>(
 impl<
     N: RequestClient + PublishClient + FlushClient,
     C: GetElapsed,
-    J: JetStreamPublisher + JetStreamConsumerFactory,
+    J: JetStreamPublisher + JetStreamGetStream,
 > Bridge<N, C, J>
 where
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
 {
     pub(crate) async fn session_request<Req, Res>(
         &self,
@@ -198,10 +198,10 @@ where
 impl<
     N: RequestClient + PublishClient + SubscribeClient + FlushClient,
     C: GetElapsed,
-    J: JetStreamPublisher + JetStreamConsumerFactory,
+    J: JetStreamPublisher + JetStreamGetStream,
 > Agent for Bridge<N, C, J>
 where
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
 {
     async fn initialize(&self, args: InitializeRequest) -> Result<InitializeResponse> {
         initialize::handle(self, args).await

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

@@ -3,7 +3,7 @@ use crate::nats::{FlushClient, PublishClient, RequestClient, session};
 use crate::session_id::AcpSessionId;
 use agent_client_protocol::{CloseSessionRequest, CloseSessionResponse, Error, ErrorCode, Result};
 use tracing::{info, instrument};
-use trogon_nats::jetstream::{JetStreamConsumerFactory, JetStreamPublisher, JsRequestMessage};
+use trogon_nats::jetstream::{JetStreamGetStream, JetStreamPublisher, JsRequestMessage};
 use trogon_std::time::GetElapsed;
 
 #[instrument(
@@ -14,13 +14,13 @@ use trogon_std::time::GetElapsed;
 pub async fn handle<
     N: RequestClient + PublishClient + FlushClient,
     C: GetElapsed,
-    J: JetStreamPublisher + JetStreamConsumerFactory,
+    J: JetStreamPublisher + JetStreamGetStream,
 >(
     bridge: &Bridge<N, C, J>,
     args: CloseSessionRequest,
 ) -> Result<CloseSessionResponse>
 where
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
 {
     let start = bridge.clock.now();
 

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

@@ -3,7 +3,7 @@ use crate::nats::{FlushClient, PublishClient, RequestClient, session};
 use crate::session_id::AcpSessionId;
 use agent_client_protocol::{Error, ErrorCode, ForkSessionRequest, ForkSessionResponse, Result};
 use tracing::{Span, info, instrument};
-use trogon_nats::jetstream::{JetStreamConsumerFactory, JetStreamPublisher, JsRequestMessage};
+use trogon_nats::jetstream::{JetStreamGetStream, JetStreamPublisher, JsRequestMessage};
 use trogon_std::time::GetElapsed;
 
 #[instrument(
@@ -14,13 +14,13 @@ use trogon_std::time::GetElapsed;
 pub async fn handle<
     N: RequestClient + PublishClient + FlushClient,
     C: GetElapsed,
-    J: JetStreamPublisher + JetStreamConsumerFactory,
+    J: JetStreamPublisher + JetStreamGetStream,
 >(
     bridge: &Bridge<N, C, J>,
     args: ForkSessionRequest,
 ) -> Result<ForkSessionResponse>
 where
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
 {
     let start = bridge.clock.now();
 

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

@@ -7,8 +7,8 @@ use std::time::Duration;
 use tokio::time::timeout;
 use trogon_nats::REQ_ID_HEADER;
 use trogon_nats::jetstream::{
-    JetStreamConsumer as _, JetStreamConsumerFactory, JetStreamPublisher, JsAck as _,
-    JsAckWith as _, JsMessageRef as _, JsRequestMessage,
+    JetStreamConsumer as _, JetStreamCreateConsumer as _, JetStreamGetStream, JetStreamPublisher,
+    JsAck as _, JsAckWith as _, JsMessageRef as _, JsRequestMessage,
 };
 use trogon_std::JsonSerialize;
 
@@ -27,8 +27,8 @@ pub async fn js_request<J, Req, Res, S>(
     operation_timeout: Duration,
 ) -> agent_client_protocol::Result<Res>
 where
-    J: JetStreamPublisher + JetStreamConsumerFactory,
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
+    J: JetStreamPublisher + JetStreamGetStream,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
     Req: serde::Serialize,
     Res: DeserializeOwned,
     S: JsonSerialize,
@@ -37,22 +37,22 @@ where
     // runner responds before we start consuming. DeliverAll replays from stream start.
     let responses_stream = streams::responses_stream_name(prefix);
     let resp_config = consumers::response_consumer(prefix, session_id, req_id);
-    let resp_consumer: J::Consumer = js
-        .create_consumer(&responses_stream, resp_config)
+    let stream = js
+        .get_stream(&responses_stream)
         .await
-        .map_err(|e| {
-            Error::new(
-                ErrorCode::InternalError.into(),
-                format!("create response consumer: {e}"),
-            )
-        })?;
-    let mut resp_messages: <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Messages =
-        resp_consumer.messages().await.map_err(|e| {
-            Error::new(
-                ErrorCode::InternalError.into(),
-                format!("response messages: {e}"),
-            )
-        })?;
+        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("get stream: {e}")))?;
+    let resp_consumer = stream.create_consumer(resp_config).await.map_err(|e| {
+        Error::new(
+            ErrorCode::InternalError.into(),
+            format!("create response consumer: {e}"),
+        )
+    })?;
+    let mut resp_messages = resp_consumer.messages().await.map_err(|e| {
+        Error::new(
+            ErrorCode::InternalError.into(),
+            format!("response messages: {e}"),
+        )
+    })?;
 
     let payload_bytes = serializer
         .to_vec(request)
@@ -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}")))?
         .await
-        .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js publish: {e}")))?;
+        .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/js_request.rs

@@ -3,7 +3,7 @@ use crate::nats::{FlushClient, PublishClient, RequestClient, session};
 use crate::session_id::AcpSessionId;
 use agent_client_protocol::{Error, ErrorCode, LoadSessionRequest, LoadSessionResponse, Result};
 use tracing::{info, instrument};
-use trogon_nats::jetstream::{JetStreamConsumerFactory, JetStreamPublisher, JsRequestMessage};
+use trogon_nats::jetstream::{JetStreamGetStream, JetStreamPublisher, JsRequestMessage};
 use trogon_std::time::GetElapsed;
 
 #[instrument(
@@ -14,13 +14,13 @@ use trogon_std::time::GetElapsed;
 pub async fn handle<
     N: RequestClient + PublishClient + FlushClient,
     C: GetElapsed,
-    J: JetStreamPublisher + JetStreamConsumerFactory,
+    J: JetStreamPublisher + JetStreamGetStream,
 >(
     bridge: &Bridge<N, C, J>,
     args: LoadSessionRequest,
 ) -> Result<LoadSessionResponse>
 where
-    <J::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
+    <<J::Stream as trogon_nats::jetstream::JetStreamCreateConsumer>::Consumer as trogon_nats::jetstream::JetStreamConsumer>::Message: JsRequestMessage,
 {
     let start = bridge.clock.now();