refactor(acp-nats): per-subject structs with marker traits

Each NATS subject is its own struct with typed parameters (AcpPrefix,
AcpSessionId). Marker traits (Requestable, Publishable, SessionCommand,
Subscribable, ClientRequestable) enforce correct transport at compile
time — passing a CancelSubject to request_with_timeout is a compile
error, not a runtime bug.

Also removes all production .unwrap() calls:
- publish_session_ready: gracefully handles invalid session IDs
- PendingSessionPromptResponseWaiters: returns Result for poison errors

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

Author: yordis

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

@@ -34,7 +34,7 @@ impl std::fmt::Display for AcpPrefixError {
 impl std::error::Error for AcpPrefixError {}
 
 /// NATS-safe ACP prefix. Guarantees validity at construction—invalid instances are unrepresentable.
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct AcpPrefix(Arc<str>);
 
 impl AcpPrefix {

rsworkspace/crates/acp-nats/src/acp_prefix.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_ref()),
         &args,
         bridge.config.operation_timeout,
     )

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

@@ -108,7 +108,7 @@ impl<N, C: GetElapsed, J> Bridge<N, C, J> {
 impl<N: PublishClient + FlushClient + Clone + Send + 'static, C: GetElapsed, J> Bridge<N, C, J> {
     pub(crate) fn schedule_session_ready(&self, session_id: SessionId) {
         let nats = self.nats.clone();
-        let prefix = self.config.acp_prefix().to_string();
+        let prefix = self.config.acp_prefix_ref().clone();
         let metrics = self.metrics.clone();
         let handle = tokio::spawn(async move {
             publish_session_ready(&nats, &prefix, &session_id, &metrics).await;
@@ -119,13 +119,21 @@ impl<N: PublishClient + FlushClient + Clone + Send + 'static, C: GetElapsed, J>
 
 async fn publish_session_ready<N: PublishClient + FlushClient>(
     nats: &N,
-    prefix: &str,
+    prefix: &crate::acp_prefix::AcpPrefix,
     session_id: &SessionId,
     metrics: &Metrics,
 ) {
     tokio::time::sleep(SESSION_READY_DELAY).await;
 
-    let subject = session::agent::ext_ready(prefix, &session_id.to_string());
+    let acp_session_id = match crate::session_id::AcpSessionId::new(session_id.to_string()) {
+        Ok(id) => id,
+        Err(e) => {
+            warn!(session_id = %session_id, error = %e, "Invalid session ID from backend, skipping session.ready");
+            metrics.record_error("session_ready", "invalid_session_id");
+            return;
+        }
+    };
+    let subject = session::agent::ExtReadySubject::new(prefix, &acp_session_id);
     info!(session_id = %session_id, subject = %subject, "Publishing session.ready");
 
     let message = ExtSessionReady::new(session_id.clone());
@@ -157,7 +165,7 @@ where
 {
     pub(crate) async fn session_request<Req, Res>(
         &self,
-        subject: &str,
+        subject: &impl crate::nats::markers::SessionCommand,
         args: &Req,
         session_id: &str,
     ) -> Result<Res>
@@ -167,12 +175,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 +191,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/bridge.rs

@@ -23,7 +23,7 @@ pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed, J>(
 
     info!(session_id = %args.session_id, "Cancel notification");
 
-    AcpSessionId::try_from(&args.session_id).map_err(|e| {
+    let session_id = AcpSessionId::try_from(&args.session_id).map_err(|e| {
         bridge
             .metrics
             .record_request("cancel", bridge.clock.elapsed(start).as_secs_f64(), false);
@@ -34,7 +34,8 @@ 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 prefix = bridge.config.acp_prefix_ref();
+    let subject = session::agent::CancelSubject::new(prefix, &session_id);
 
     let publish_result = nats::publish(
         bridge.nats(),
@@ -57,8 +58,7 @@ 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(prefix, &session_id);
     if let Err(e) = bridge
         .nats()
         .publish_with_headers(

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

@@ -35,8 +35,8 @@ where
             format!("Invalid session ID: {}", e),
         )
     })?;
-    let prefix = bridge.config.acp_prefix();
-    let subject = session::agent::close(prefix, session_id.as_str());
+    let prefix = bridge.config.acp_prefix_ref();
+    let subject = session::agent::CloseSubject::new(prefix, &session_id);
 
     let result = bridge
         .session_request::<CloseSessionRequest, CloseSessionResponse>(

rsworkspace/crates/acp-nats/src/agent/close_session.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_ref(), method_name.as_str());
 
     let result = nats::request_with_timeout::<N, ExtRequest, ExtResponse>(
         nats,

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

@@ -38,7 +38,8 @@ 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_ref(), method_name.as_str());
 
     let publish_result = nats::publish(
         bridge.nats(),

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

@@ -35,8 +35,8 @@ where
             format!("Invalid session ID: {}", e),
         )
     })?;
-    let prefix = bridge.config.acp_prefix();
-    let subject = session::agent::fork(prefix, session_id.as_str());
+    let prefix = bridge.config.acp_prefix_ref();
+    let subject = session::agent::ForkSubject::new(prefix, &session_id);
 
     let result = bridge
         .session_request::<ForkSessionRequest, ForkSessionResponse>(

rsworkspace/crates/acp-nats/src/agent/fork_session.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_ref());
 
     let result = nats::request_with_timeout::<N, InitializeRequest, InitializeResponse>(
         nats,