refactor(acp-nats): per-subject structs enforce correct NATS usage at compile time

Each NATS subject is now its own struct with typed parameters.
Passing the wrong subject to the wrong function is a compile error.

Organized into directories by role:
- commands/   — bridge→agent JetStream commands
- global/     — global Core NATS request/reply
- responses/  — agent→bridge responses
- client_ops/ — agent→bridge client proxy
- subscriptions/ — wildcard patterns for .subscribe()

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

Author: yordis

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

@@ -12,6 +12,7 @@ use agent_client_protocol::{
 };
 use async_nats::Message;
 use async_nats::jetstream::AckKind;
+use async_nats::subject::ToSubject;
 #[cfg(test)]
 use bytes::Bytes;
 use futures::StreamExt;
@@ -160,8 +161,8 @@ where
     N: SubscribeClient + PublishClient + FlushClient + Clone + 'static,
     A: Agent + 'static,
 {
-    let global_wildcard = acp_nats::nats::agent::wildcards::all(prefix);
-    let session_wildcard = acp_nats::nats::session::wildcards::all_agent(prefix);
+    let global_wildcard = acp_nats::nats::agent::wildcards::GlobalAllSubject::new(prefix);
+    let session_wildcard = acp_nats::nats::session::wildcards::AllAgentSubject::new(prefix);
 
     info!(
         global = %global_wildcard,
@@ -206,8 +207,8 @@ where
     N: SubscribeClient + PublishClient + FlushClient + Clone + 'static,
     A: Agent + 'static,
 {
-    let global_wildcard = acp_nats::nats::agent::wildcards::all(prefix);
-    let ext_wildcard = acp_nats::nats::session::wildcards::all_agent_ext(prefix);
+    let global_wildcard = acp_nats::nats::agent::wildcards::GlobalAllSubject::new(prefix);
+    let ext_wildcard = acp_nats::nats::session::wildcards::AllAgentExtSubject::new(prefix);
 
     info!(
         global = %global_wildcard,
@@ -568,16 +569,20 @@ async fn dispatch_js_message<N: PublishClient + FlushClient, A: Agent, M: JsDisp
         .and_then(|h| h.get(trogon_nats::REQ_ID_HEADER))
         .map(|v| v.as_str().to_string());
 
-    let reply_subject = match (&req_id, &method) {
+    let reply_subject: Option<async_nats::subject::Subject> = match (&req_id, &method) {
         (Some(rid), SessionAgentMethod::Prompt) => Some(
-            acp_nats::nats::session::agent::prompt_response(prefix, session_id.as_str(), rid),
+            acp_nats::nats::session::agent::PromptResponseSubject::new(
+                prefix,
+                session_id.as_str(),
+                rid,
+            )
+            .to_subject(),
         ),
         (_, SessionAgentMethod::Cancel) => None,
-        (Some(rid), _) => Some(acp_nats::nats::session::agent::response(
-            prefix,
-            session_id.as_str(),
-            rid,
-        )),
+        (Some(rid), _) => Some(
+            acp_nats::nats::session::agent::ResponseSubject::new(prefix, session_id.as_str(), rid)
+                .to_subject(),
+        ),
         (None, _) => {
             warn!(subject, "JetStream message missing X-Req-Id header");
             None
@@ -587,7 +592,7 @@ async fn dispatch_js_message<N: PublishClient + FlushClient, A: Agent, M: JsDisp
     let inner = js_msg.message();
     let msg = Message {
         subject: subject.as_str().into(),
-        reply: reply_subject.as_deref().map(|s| s.into()),
+        reply: reply_subject,
         payload: inner.payload.clone(),
         headers: inner.headers.clone(),
         status: None,

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

@@ -21,7 +21,7 @@ pub async fn handle<N: RequestClient, C: GetElapsed, J>(
 
     let result = nats::request_with_timeout::<N, AuthenticateRequest, AuthenticateResponse>(
         nats,
-        &agent::authenticate(bridge.config.acp_prefix()),
+        &agent::AuthenticateSubject::new(bridge.config.acp_prefix()),
         &args,
         bridge.config.operation_timeout,
     )

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

@@ -125,7 +125,7 @@ async fn publish_session_ready<N: PublishClient + FlushClient>(
 ) {
     tokio::time::sleep(SESSION_READY_DELAY).await;
 
-    let subject = session::agent::ext_ready(prefix, &session_id.to_string());
+    let subject = session::agent::ExtReadySubject::new(prefix, &session_id.to_string());
     info!(session_id = %session_id, subject = %subject, "Publishing session.ready");
 
     let message = ExtSessionReady::new(session_id.clone());
@@ -157,7 +157,7 @@ where
 {
     pub(crate) async fn session_request<Req, Res>(
         &self,
-        subject: &str,
+        subject: &impl std::fmt::Display,
         args: &Req,
         session_id: &str,
     ) -> Result<Res>
@@ -167,12 +167,13 @@ where
     {
         use crate::error::map_nats_error;
 
+        let subject_str = subject.to_string();
         match self.js() {
             Some(js) => {
                 let req_id = uuid::Uuid::new_v4().to_string();
                 js_request::js_request::<J, _, Res, _>(
                     js,
-                    subject,
+                    &subject_str,
                     args,
                     &trogon_std::StdJsonSerialize,
                     self.config.acp_prefix(),
@@ -182,9 +183,9 @@ where
                 )
                 .await
             }
-            None => nats::request_with_timeout::<N, Req, Res>(
+            None => trogon_nats::request_with_timeout::<N, Req, Res>(
                 self.nats(),
-                subject,
+                &subject_str,
                 args,
                 self.config.operation_timeout,
             )

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

@@ -34,7 +34,10 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(
         )
     })?;
 
-    let subject = session::agent::cancel(bridge.config.acp_prefix(), &args.session_id.to_string());
+    let subject = session::agent::CancelSubject::new(
+        bridge.config.acp_prefix(),
+        &args.session_id.to_string(),
+    );
 
     let publish_result = nats::publish(
         bridge.nats(),
@@ -57,8 +60,10 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(
             .record_error("cancel", "cancel_publish_failed");
     }
 
-    let cancelled_subject =
-        session::agent::cancelled(bridge.config.acp_prefix(), &args.session_id.to_string());
+    let cancelled_subject = session::agent::CancelledSubject::new(
+        bridge.config.acp_prefix(),
+        &args.session_id.to_string(),
+    );
     if let Err(e) = bridge
         .nats()
         .publish_with_headers(

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

@@ -36,7 +36,7 @@ where
         )
     })?;
     let prefix = bridge.config.acp_prefix();
-    let subject = session::agent::close(prefix, session_id.as_str());
+    let subject = session::agent::CloseSubject::new(prefix, session_id.as_str());
 
     let result = bridge
         .session_request::<CloseSessionRequest, CloseSessionResponse>(

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

@@ -35,7 +35,7 @@ pub async fn handle<N: RequestClient, C: GetElapsed, J>(
     })?;
 
     let nats = bridge.nats();
-    let subject = agent::ext(bridge.config.acp_prefix(), method_name.as_str());
+    let subject = agent::ExtSubject::new(bridge.config.acp_prefix(), method_name.as_str());
 
     let result = nats::request_with_timeout::<N, ExtRequest, ExtResponse>(
         nats,

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

@@ -38,7 +38,7 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(
         )
     })?;
 
-    let subject = agent::ext(bridge.config.acp_prefix(), method_name.as_str());
+    let subject = agent::ExtNotifySubject::new(bridge.config.acp_prefix(), method_name.as_str());
 
     let publish_result = nats::publish(
         bridge.nats(),

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

@@ -36,7 +36,7 @@ where
         )
     })?;
     let prefix = bridge.config.acp_prefix();
-    let subject = session::agent::fork(prefix, session_id.as_str());
+    let subject = session::agent::ForkSubject::new(prefix, session_id.as_str());
 
     let result = bridge
         .session_request::<ForkSessionRequest, ForkSessionResponse>(

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

@@ -25,7 +25,7 @@ pub async fn handle<N: RequestClient, C: GetElapsed, J>(
     info!(client = %client_name, "Initialize request");
 
     let nats = bridge.nats();
-    let subject = agent::initialize(bridge.config.acp_prefix());
+    let subject = agent::InitializeSubject::new(bridge.config.acp_prefix());
 
     let result = nats::request_with_timeout::<N, InitializeRequest, InitializeResponse>(
         nats,

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

@@ -15,7 +15,7 @@ pub async fn handle<N: RequestClient, C: GetElapsed, J>(
     info!("List sessions request");
 
     let nats = bridge.nats();
-    let subject = agent::session_list(bridge.config.acp_prefix());
+    let subject = agent::SessionListSubject::new(bridge.config.acp_prefix());
 
     let result = nats::request_with_timeout::<N, ListSessionsRequest, ListSessionsResponse>(
         nats,