feat(acp-nats): add Bridge infrastructure for session cancel and ready

- Add CancelledSessions with TTL eviction for pre-flight cancel tracking
- Add spawn_session_ready and session-ready task management to Bridge
- Consolidate ClientMethod/ParsedClientSubject into parsing.rs
- Add agent/client wildcard subjects
- Add cancel_waiter_for_session to PendingPromptWaiters
- Fix config min-concurrency clamping, add with_prefix alias
- Remove unused subject capacity validation (tracked in FOLLOWUP.md)

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

Author: yordis

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

@@ -1,3 +1,7 @@
+// Safety: `CancelledSessions` uses `Mutex` for interior mutability so the `Bridge` remains
+// `Send`/`Sync`. The fire-and-forget publish in `spawn_session_ready` uses `tokio::spawn`
+// and captures only cloned, `Send` values — it never touches shared state from the closure.
+
 mod authenticate;
 mod cancel;
 mod ext_method;
@@ -8,27 +12,78 @@ mod new_session;
 mod prompt;
 mod set_session_mode;
 
-use crate::config::Config;
-use crate::nats::{FlushClient, PublishClient, RequestClient};
+use crate::config::{Config, SESSION_READY_DELAY};
+use crate::nats::{self, ExtSessionReady, FlushClient, PublishClient, RequestClient, agent};
 use crate::pending_prompt_waiters::PendingSessionPromptResponseWaiters;
 use crate::prompt_slot_counter::PromptSlotCounter;
 use crate::telemetry::metrics::Metrics;
 use agent_client_protocol::{
     Agent, AuthenticateRequest, AuthenticateResponse, CancelNotification, ExtNotification,
     ExtRequest, ExtResponse, InitializeRequest, InitializeResponse, LoadSessionRequest,
     LoadSessionResponse, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse,
-    Result, SetSessionModeRequest, SetSessionModeResponse,
+    Result, SessionId, SetSessionModeRequest, SetSessionModeResponse,
 };
 use opentelemetry::metrics::Meter;
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::time::Duration;
+use tracing::{info, warn};
 use trogon_std::time::GetElapsed;
 
+const CANCELLED_SESSION_TTL: Duration = Duration::from_secs(300);
+const CLEANUP_EVERY: usize = 16;
+
+pub(crate) struct CancelledSessions<I: Copy> {
+    map: Mutex<HashMap<SessionId, I>>,
+    cleanup_counter: std::sync::atomic::AtomicUsize,
+}
+
+impl<I: Copy> CancelledSessions<I> {
+    pub fn new() -> Self {
+        Self {
+            map: Mutex::new(HashMap::new()),
+            cleanup_counter: std::sync::atomic::AtomicUsize::new(0),
+        }
+    }
+
+    pub fn mark_cancelled<C: GetElapsed<Instant = I>>(&self, session_id: SessionId, clock: &C) {
+        let mut map = self.map.lock().unwrap();
+        map.insert(session_id, clock.now());
+        let count = self.cleanup_counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        if count.is_multiple_of(CLEANUP_EVERY) {
+            map.retain(|_, ts| clock.elapsed(*ts) < CANCELLED_SESSION_TTL);
+        }
+    }
+
+    pub fn take_if_cancelled<C: GetElapsed<Instant = I>>(
+        &self,
+        session_id: &SessionId,
+        clock: &C,
+    ) -> Option<()> {
+        let mut map = self.map.lock().unwrap();
+        let is_valid = map
+            .get(session_id)
+            .is_some_and(|ts| clock.elapsed(*ts) < CANCELLED_SESSION_TTL);
+
+        if is_valid {
+            map.remove(session_id);
+            Some(())
+        } else {
+            map.remove(session_id);
+            None
+        }
+    }
+}
+
 pub struct Bridge<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> {
     pub(crate) nats: N,
     pub(crate) clock: C,
     pub(crate) metrics: Metrics,
+    pub(crate) cancelled_sessions: CancelledSessions<C::Instant>,
     pub(crate) pending_session_prompt_responses: PendingSessionPromptResponseWaiters<C::Instant>,
     pub(crate) prompt_slot_counter: PromptSlotCounter,
     pub(crate) config: Config,
+    pub(crate) session_ready_publish_tasks: Mutex<Vec<tokio::task::JoinHandle<()>>>,
 }
 
 impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Bridge<N, C> {
@@ -39,14 +94,69 @@ impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Bridge<N, C>
             clock,
             config,
             metrics: Metrics::new(meter),
+            cancelled_sessions: CancelledSessions::new(),
             pending_session_prompt_responses: PendingSessionPromptResponseWaiters::new(),
             prompt_slot_counter: PromptSlotCounter::new(max_concurrent),
+            session_ready_publish_tasks: Mutex::new(Vec::new()),
         }
     }
 
     pub(crate) fn nats(&self) -> &N {
         &self.nats
     }
+
+    pub(crate) fn register_session_ready_task(&self, task: tokio::task::JoinHandle<()>) {
+        let mut tasks = self.session_ready_publish_tasks.lock().unwrap();
+        tasks.retain(|task| !task.is_finished());
+        tasks.push(task);
+    }
+
+    pub fn has_pending_session_ready_tasks(&self) -> bool {
+        !self.session_ready_publish_tasks.lock().unwrap().is_empty()
+    }
+
+    pub async fn await_session_ready_tasks(&self) {
+        let tasks = std::mem::take(&mut *self.session_ready_publish_tasks.lock().unwrap());
+        for task in tasks {
+            if let Err(e) = task.await {
+                warn!(error = %e, "session_ready task panicked");
+            }
+        }
+    }
+
+    pub(crate) fn spawn_session_ready(&self, session_id: &SessionId) {
+        let nats_clone = self.nats.clone();
+        let prefix = self.config.acp_prefix().to_string();
+        let session_id = session_id.clone();
+        let metrics = self.metrics.clone();
+        let session_ready_task = tokio::spawn(async move {
+            tokio::time::sleep(SESSION_READY_DELAY).await;
+
+            let ready_subject = agent::ext_session_ready(&prefix, &session_id.to_string());
+            info!(session_id = %session_id, subject = %ready_subject, "Publishing session.ready");
+
+            let ready_message = ExtSessionReady::new(session_id.clone());
+
+            let options = nats::PublishOptions::builder()
+                .publish_retry_policy(nats::RetryPolicy::standard())
+                .flush_policy(nats::FlushPolicy::standard())
+                .build();
+
+            if let Err(e) =
+                nats::publish(&nats_clone, &ready_subject, &ready_message, options).await
+            {
+                warn!(
+                    error = %e,
+                    session_id = %session_id,
+                    "Failed to publish session.ready"
+                );
+                metrics.record_error("session_ready", "session_ready_publish_failed");
+            } else {
+                info!(session_id = %session_id, "Published session.ready");
+            }
+        });
+        self.register_session_ready_task(session_ready_task);
+    }
 }
 
 #[async_trait::async_trait(?Send)]

rsworkspace/crates/acp-nats/src/config.rs

@@ -6,6 +6,7 @@ use crate::acp_prefix::AcpPrefix;
 const DEFAULT_OPERATION_TIMEOUT: Duration = Duration::from_secs(30);
 const DEFAULT_PROMPT_TIMEOUT: Duration = Duration::from_secs(7200);
 const DEFAULT_MAX_CONCURRENT_CLIENT_TASKS: usize = 256;
+
 /// Above this value, prompt timeout errors are rendered in seconds instead of milliseconds.
 pub(crate) const PROMPT_TIMEOUT_MESSAGE_SECS_THRESHOLD: Duration = Duration::from_secs(60);
 /// Suppresses duplicate timeout-related warnings for a short late-response window.
@@ -36,6 +37,10 @@ impl Config {
         }
     }
 
+    pub fn with_prefix(acp_prefix: AcpPrefix, nats: NatsConfig) -> Self {
+        Self::new(acp_prefix, nats)
+    }
+
     pub fn with_operation_timeout(mut self, timeout: Duration) -> Self {
         self.operation_timeout = timeout;
         self
@@ -47,7 +52,7 @@ impl Config {
     }
 
     pub fn with_max_concurrent_client_tasks(mut self, max: usize) -> Self {
-        self.max_concurrent_client_tasks = max;
+        self.max_concurrent_client_tasks = max.max(1);
         self
     }
 
@@ -72,7 +77,7 @@ impl Config {
     ///
     /// A minimum of 1 avoids misconfiguration that would permanently reject all client messages.
     pub fn max_concurrent_client_tasks(&self) -> usize {
-        self.max_concurrent_client_tasks.max(1)
+        self.max_concurrent_client_tasks
     }
 
     #[cfg(test)]
@@ -81,7 +86,7 @@ impl Config {
             servers: vec!["localhost:4222".to_string()],
             auth: trogon_nats::NatsAuth::None,
         };
-        Self::new(AcpPrefix::new(acp_prefix).unwrap(), nats)
+        Self::new(AcpPrefix::new(acp_prefix.to_string()).unwrap(), nats)
     }
 }
 
@@ -132,9 +137,8 @@ mod tests {
     }
 
     #[test]
-    fn config_max_concurrent_client_tasks_zero_is_clamped_to_one() {
-        let config = Config::new(AcpPrefix::new("acp").unwrap(), default_nats())
-            .with_max_concurrent_client_tasks(0);
+    fn config_with_max_concurrent_client_tasks_enforces_minimum() {
+        let config = Config::for_test("acp").with_max_concurrent_client_tasks(0);
         assert_eq!(config.max_concurrent_client_tasks(), 1);
     }
 
@@ -144,4 +148,11 @@ mod tests {
         assert_eq!(config.nats().servers.len(), 1);
         assert_eq!(config.nats().servers[0], "localhost:4222");
     }
+
+    #[test]
+    fn config_with_prefix_accepts_validated_prefix() {
+        let config = Config::with_prefix(AcpPrefix::new("acp").unwrap(), default_nats());
+        assert_eq!(config.acp_prefix(), "acp");
+    }
+
 }

rsworkspace/crates/acp-nats/src/nats/client_method.rs

@@ -1,16 +1,13 @@
-mod client_method;
 mod extensions;
-mod parsed_client_subject;
 mod parsing;
 mod subjects;
 pub(crate) mod token;
 
-pub use client_method::ClientMethod;
-pub use extensions::ExtSessionReady;
-pub use parsed_client_subject::ParsedClientSubject;
-pub use parsing::parse_client_subject;
-pub use subjects::{agent, client};
 pub use trogon_nats::{
     FlushClient, FlushPolicy, PublishClient, PublishOptions, RequestClient, RetryPolicy,
     SubscribeClient, connect, headers_with_trace_context, publish, request_with_timeout,
 };
+
+pub use extensions::ExtSessionReady;
+pub use parsing::{ClientMethod, ParsedClientSubject, parse_client_subject};
+pub use subjects::{agent, client};