refactor(acp-nats): typed subjects enforce correct transport at compile time

yordis · yordis · commit 4f5a90e40bda · 2026-03-31T23:54:43.000-04:00
Introduce RequestSubject, PublishSubject, and JetStreamSubject newtypes.
Subject builders return the type matching their transport. The wrapper
functions in nats::mod only accept the correct type — passing a
JetStreamSubject to request_with_timeout is a compile error.

Signed-off-by: Yordis Prieto &lt;yordis.prieto@gmail.com&gt;
diff --git a/rsworkspace/crates/acp-nats/src/agent/bridge.rs b/rsworkspace/crates/acp-nats/src/agent/bridge.rs
@@ -157,7 +157,7 @@ where
 {
     pub(crate) async fn session_request<Req, Res>(
         &self,
-        subject: &str,
+        subject: &crate::nats::JetStreamSubject,
         args: &Req,
         session_id: &str,
     ) -> Result<Res>
@@ -172,7 +172,7 @@ where
                 let req_id = uuid::Uuid::new_v4().to_string();
                 js_request::js_request::<J, _, Res, _>(
                     js,
-                    subject,
+                    &subject.0,
                     args,
                     &trogon_std::StdJsonSerialize,
                     self.config.acp_prefix(),
@@ -182,9 +182,9 @@ where
                 )
                 .await
             }
-            None => nats::request_with_timeout::<N, Req, Res>(
+            None => trogon_nats::request_with_timeout::<N, Req, Res>(
                 self.nats(),
-                subject,
+                &subject.0,
                 args,
                 self.config.operation_timeout,
             )
diff --git a/rsworkspace/crates/acp-nats/src/agent/cancel.rs b/rsworkspace/crates/acp-nats/src/agent/cancel.rs
@@ -62,7 +62,7 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(
     if let Err(e) = bridge
         .nats()
         .publish_with_headers(
-            cancelled_subject,
+            cancelled_subject.0,
             async_nats::HeaderMap::new(),
             bytes::Bytes::new(),
         )
diff --git a/rsworkspace/crates/acp-nats/src/agent/ext_notification.rs b/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::ext_notify(bridge.config.acp_prefix(), method_name.as_str());
 
     let publish_result = nats::publish(
         bridge.nats(),
diff --git a/rsworkspace/crates/acp-nats/src/agent/prompt.rs b/rsworkspace/crates/acp-nats/src/agent/prompt.rs
@@ -90,7 +90,7 @@ where
 
     let mut cancel_sub = bridge
         .nats
-        .subscribe(session::agent::cancelled(prefix, sid))
+        .subscribe(session::agent::cancelled(prefix, sid).0)
         .await
         .map_err(|e| {
             Error::new(
@@ -110,7 +110,7 @@ where
     let prompt_subject = session::agent::prompt(prefix, sid);
     bridge
         .nats
-        .publish_with_headers(prompt_subject, headers, Bytes::from(payload_bytes))
+        .publish_with_headers(prompt_subject.0, headers, Bytes::from(payload_bytes))
         .await
         .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("publish: {e}")))?;
 
@@ -253,7 +253,7 @@ where
     // Cancel still uses core NATS — it's a fire-and-forget signal, not persisted.
     let mut cancel_sub = bridge
         .nats
-        .subscribe(session::agent::cancelled(prefix, sid))
+        .subscribe(session::agent::cancelled(prefix, sid).0)
         .await
         .map_err(|e| {
             Error::new(
@@ -272,7 +272,7 @@ where
     headers.insert(SESSION_ID_HEADER, sid);
 
     let prompt_subject = session::agent::prompt(prefix, sid);
-    js.publish_with_headers(prompt_subject, headers, Bytes::from(payload_bytes))
+    js.publish_with_headers(prompt_subject.0, headers, Bytes::from(payload_bytes))
         .await
         .map_err(|e| Error::new(ErrorCode::InternalError.into(), format!("js publish: {e}")))?
         .await
diff --git a/rsworkspace/crates/acp-nats/src/nats/mod.rs b/rsworkspace/crates/acp-nats/src/nats/mod.rs
@@ -3,14 +3,44 @@ pub mod parsing;
 mod subjects;
 pub(crate) mod token;
 
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use std::time::Duration;
+
 pub use extensions::ExtSessionReady;
 pub use parsing::{
     ClientMethod, GlobalAgentMethod, ParsedAgentSubject, ParsedClientSubject, SessionAgentMethod,
     parse_agent_subject, parse_client_subject,
 };
-pub use subjects::{agent, session};
+pub use subjects::{JetStreamSubject, PublishSubject, RequestSubject, agent, session};
 pub use trogon_nats::{
-    FlushClient, FlushPolicy, PublishClient, PublishOptions, RequestClient, RetryPolicy,
-    SubscribeClient, client, connect, headers_with_trace_context, inject_trace_context, publish,
-    request_with_timeout,
+    FlushClient, FlushPolicy, NatsError, PublishClient, PublishOptions, RequestClient, RetryPolicy,
+    SubscribeClient, client, connect, headers_with_trace_context, inject_trace_context,
 };
+
+/// Core NATS request/reply — only accepts [`RequestSubject`].
+pub async fn request_with_timeout<N: RequestClient, Req, Res>(
+    client: &N,
+    subject: &RequestSubject,
+    request: &Req,
+    timeout: Duration,
+) -> Result<Res, NatsError>
+where
+    Req: Serialize,
+    Res: DeserializeOwned,
+{
+    trogon_nats::request_with_timeout(client, &subject.0, request, timeout).await
+}
+
+/// Core NATS fire-and-forget publish — only accepts [`PublishSubject`].
+pub async fn publish<N: PublishClient + FlushClient, Req>(
+    client: &N,
+    subject: &PublishSubject,
+    request: &Req,
+    options: PublishOptions,
+) -> Result<(), NatsError>
+where
+    Req: Serialize,
+{
+    trogon_nats::publish(client, &subject.0, request, options).await
+}
diff --git a/rsworkspace/crates/acp-nats/src/nats/subjects.rs b/rsworkspace/crates/acp-nats/src/nats/subjects.rs
@@ -1,22 +1,106 @@
+/// Bridge → agent via Core NATS request/reply. Expects a response.
+#[derive(Debug)]
+#[repr(transparent)]
+pub struct RequestSubject(pub(crate) String);
+
+/// Bridge → agent via Core NATS publish. Fire-and-forget, no response.
+#[derive(Debug)]
+#[repr(transparent)]
+pub struct PublishSubject(pub(crate) String);
+
+/// Bridge → agent via JetStream. Captured by a stream, consumed by agent.
+#[derive(Debug)]
+#[repr(transparent)]
+pub struct JetStreamSubject(pub(crate) String);
+
+impl std::fmt::Display for RequestSubject {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl std::fmt::Display for PublishSubject {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl std::fmt::Display for JetStreamSubject {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl PartialEq<&str> for RequestSubject {
+    fn eq(&self, other: &&str) -> bool {
+        self.0 == *other
+    }
+}
+
+impl PartialEq<&str> for PublishSubject {
+    fn eq(&self, other: &&str) -> bool {
+        self.0 == *other
+    }
+}
+
+impl PartialEq<&str> for JetStreamSubject {
+    fn eq(&self, other: &&str) -> bool {
+        self.0 == *other
+    }
+}
+
+#[cfg(test)]
+impl RequestSubject {
+    pub fn starts_with(&self, prefix: &str) -> bool {
+        self.0.starts_with(prefix)
+    }
+}
+
+#[cfg(test)]
+impl PublishSubject {
+    pub fn starts_with(&self, prefix: &str) -> bool {
+        self.0.starts_with(prefix)
+    }
+}
+
+#[cfg(test)]
+impl JetStreamSubject {
+    pub fn starts_with(&self, prefix: &str) -> bool {
+        self.0.starts_with(prefix)
+    }
+}
+
 pub mod agent {
-    pub fn initialize(prefix: &str) -> String {
-        format!("{}.agent.initialize", prefix)
+    use super::{PublishSubject, RequestSubject};
+
+    /// Core NATS request/reply. Stream: GLOBAL (observability).
+    pub fn initialize(prefix: &str) -> RequestSubject {
+        RequestSubject(format!("{}.agent.initialize", prefix))
     }
 
-    pub fn authenticate(prefix: &str) -> String {
-        format!("{}.agent.authenticate", prefix)
+    /// Core NATS request/reply. Stream: GLOBAL (observability).
+    pub fn authenticate(prefix: &str) -> RequestSubject {
+        RequestSubject(format!("{}.agent.authenticate", prefix))
     }
 
-    pub fn session_new(prefix: &str) -> String {
-        format!("{}.agent.session.new", prefix)
+    /// Core NATS request/reply. Stream: GLOBAL (observability).
+    pub fn session_new(prefix: &str) -> RequestSubject {
+        RequestSubject(format!("{}.agent.session.new", prefix))
     }
 
-    pub fn session_list(prefix: &str) -> String {
-        format!("{}.agent.session.list", prefix)
+    /// Core NATS request/reply. No stream.
+    pub fn session_list(prefix: &str) -> RequestSubject {
+        RequestSubject(format!("{}.agent.session.list", prefix))
     }
 
-    pub fn ext(prefix: &str, method: &str) -> String {
-        format!("{}.agent.ext.{}", prefix, method)
+    /// Core NATS request/reply. Stream: GLOBAL_EXT (observability).
+    pub fn ext(prefix: &str, method: &str) -> RequestSubject {
+        RequestSubject(format!("{}.agent.ext.{}", prefix, method))
+    }
+
+    /// Core NATS publish (fire-and-forget notification). Stream: GLOBAL_EXT (observability).
+    pub fn ext_notify(prefix: &str, method: &str) -> PublishSubject {
+        PublishSubject(format!("{}.agent.ext.{}", prefix, method))
     }
 
     pub mod wildcards {
@@ -28,65 +112,84 @@ pub mod agent {
 
 pub mod session {
     pub mod agent {
-        pub fn load(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.load", prefix, session_id)
+        use super::super::{JetStreamSubject, PublishSubject};
+
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn load(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.load", prefix, session_id))
         }
 
-        pub fn prompt(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.prompt", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn prompt(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.prompt", prefix, session_id))
         }
 
         pub fn prompt_wildcard(prefix: &str) -> String {
             format!("{}.session.*.agent.prompt", prefix)
         }
 
-        pub fn cancel(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.cancel", prefix, session_id)
+        /// Bridge: Core NATS publish (fire-and-forget). Stream: COMMANDS.
+        pub fn cancel(prefix: &str, session_id: &str) -> PublishSubject {
+            PublishSubject(format!("{}.session.{}.agent.cancel", prefix, session_id))
         }
 
-        pub fn cancelled(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.cancelled", prefix, session_id)
+        /// Agent → bridge broadcast. Stream: RESPONSES.
+        pub fn cancelled(prefix: &str, session_id: &str) -> PublishSubject {
+            PublishSubject(format!("{}.session.{}.agent.cancelled", prefix, session_id))
         }
 
-        pub fn set_mode(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.set_mode", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn set_mode(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.set_mode", prefix, session_id))
         }
 
-        pub fn set_config_option(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.set_config_option", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn set_config_option(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!(
+                "{}.session.{}.agent.set_config_option",
+                prefix, session_id
+            ))
         }
 
-        pub fn set_model(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.set_model", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn set_model(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.set_model", prefix, session_id))
         }
 
-        pub fn fork(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.fork", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn fork(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.fork", prefix, session_id))
         }
 
-        pub fn resume(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.resume", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn resume(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.resume", prefix, session_id))
         }
 
-        pub fn close(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.close", prefix, session_id)
+        /// Bridge: JetStream publish. Stream: COMMANDS.
+        pub fn close(prefix: &str, session_id: &str) -> JetStreamSubject {
+            JetStreamSubject(format!("{}.session.{}.agent.close", prefix, session_id))
         }
 
-        pub fn ext_ready(prefix: &str, session_id: &str) -> String {
-            format!("{}.session.{}.agent.ext.ready", prefix, session_id)
+        /// Agent → bridge signal. Stream: RESPONSES.
+        pub fn ext_ready(prefix: &str, session_id: &str) -> PublishSubject {
+            PublishSubject(format!("{}.session.{}.agent.ext.ready", prefix, session_id))
         }
 
+        /// Agent → bridge async notification. Stream: NOTIFICATIONS.
         pub fn update(prefix: &str, session_id: &str, req_id: &str) -> String {
             format!("{}.session.{}.agent.update.{}", prefix, session_id, req_id)
         }
 
+        /// Agent → bridge streamed response. Stream: RESPONSES.
         pub fn prompt_response(prefix: &str, session_id: &str, req_id: &str) -> String {
             format!(
                 "{}.session.{}.agent.prompt.response.{}",
                 prefix, session_id, req_id
             )
         }
 
+        /// Agent → bridge one-shot response. Stream: RESPONSES.
         pub fn response(prefix: &str, session_id: &str, req_id: &str) -> String {
             format!(
                 "{}.session.{}.agent.response.{}",

Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(`
`62`	`62`	`if let Err(e) = bridge`
`63`	`63`	`.nats()`
`64`	`64`	`.publish_with_headers(`
`65`		`- cancelled_subject,`
	`65`	`+ cancelled_subject.0,`
`66`	`66`	`async_nats::HeaderMap::new(),`
`67`	`67`	`bytes::Bytes::new(),`
`68`	`68`	`)`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(`
`38`	`38`	`)`
`39`	`39`	`})?;`
`40`	`40`
`41`		`- let subject = agent::ext(bridge.config.acp_prefix(), method_name.as_str());`
	`41`	`+ let subject = agent::ext_notify(bridge.config.acp_prefix(), method_name.as_str());`
`42`	`42`
`43`	`43`	`let publish_result = nats::publish(`
`44`	`44`	`bridge.nats(),`