feat(acp-nats): add new_session handler with session.ready lifecycle

- Add new_session handler aligned with initialize.rs patterns
  (handler-specific error mapping, single-line generics, local subject variable)
- Add spawn_session_ready to publish ext.session.ready after successful session creation
- Add SessionReady extension type for bridge-backend coordination
- Add session_new and ext_session_ready NATS subject functions
- Add record_session_created() and record_error() to Metrics
- Add Bridge task tracking for session_ready publish tasks
- Add comprehensive test suite mirroring initialize handler tests

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

Author: yordis

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

@@ -1,23 +1,42 @@
 mod initialize;
+mod new_session;
 
 use crate::config::Config;
-use crate::nats::{FlushClient, PublishClient, RequestClient};
+use crate::nats::{
+    self, FlushClient, FlushPolicy, PublishClient, PublishOptions, RequestClient, RetryPolicy,
+    SessionReady, agent,
+};
 use crate::telemetry::metrics::Metrics;
 use agent_client_protocol::ErrorCode;
 use agent_client_protocol::{
     Agent, AuthenticateRequest, AuthenticateResponse, CancelNotification, Error, ExtNotification,
     ExtRequest, ExtResponse, InitializeRequest, InitializeResponse, LoadSessionRequest,
     LoadSessionResponse, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse,
-    Result, SetSessionModeRequest, SetSessionModeResponse,
+    Result, SessionId, SetSessionModeRequest, SetSessionModeResponse,
 };
 use opentelemetry::metrics::Meter;
+use std::cell::RefCell;
+use std::marker::PhantomData;
+use tracing::{info, warn};
 use trogon_std::time::GetElapsed;
 
+/// NATS-backed implementation of the Agent Client Protocol.
+///
+/// # Thread Safety
+///
+/// `session_ready_publish_tasks` uses `RefCell` for interior mutability. The `Bridge`
+/// must be driven from a **single task** (or a single-threaded `LocalSet`). A
+/// `PhantomData<Rc<()>>` marker prevents accidental `Send`/`Sync` usage.
+///
+/// Background work (`spawn_session_ready`) is dispatched via `tokio::spawn`
+/// and captures only cloned, `Send` values -- it never touches the `RefCell`.
 pub struct Bridge<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> {
     pub(crate) nats: N,
     pub(crate) clock: C,
     pub(crate) config: Config,
     pub(crate) metrics: Metrics,
+    pub(crate) session_ready_publish_tasks: RefCell<Vec<tokio::task::JoinHandle<()>>>,
+    _not_send_sync: PhantomData<std::rc::Rc<()>>,
 }
 
 impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Bridge<N, C> {
@@ -27,12 +46,51 @@ impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Bridge<N, C>
             clock,
             config,
             metrics: Metrics::new(meter),
+            session_ready_publish_tasks: RefCell::new(Vec::new()),
+            _not_send_sync: PhantomData,
         }
     }
 
     pub(crate) fn nats(&self) -> &N {
         &self.nats
     }
+
+    pub(crate) fn spawn_session_ready(&self, session_id: &SessionId) {
+        let nats_clone = self.nats.clone();
+        let prefix = self.config.acp_prefix().to_owned();
+        let session_id = session_id.clone();
+        let metrics = self.metrics.clone();
+
+        let task = tokio::spawn(async move {
+            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 = SessionReady::new(session_id.clone());
+
+            let options = PublishOptions::builder()
+                .publish_retry_policy(RetryPolicy::standard())
+                .flush_policy(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");
+            }
+        });
+
+        let mut tasks = self.session_ready_publish_tasks.borrow_mut();
+        tasks.retain(|t| !t.is_finished());
+        tasks.push(task);
+    }
 }
 
 #[async_trait::async_trait(?Send)]
@@ -48,11 +106,8 @@ impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Agent for Br
         ))
     }
 
-    async fn new_session(&self, _args: NewSessionRequest) -> Result<NewSessionResponse> {
-        Err(Error::new(
-            ErrorCode::InternalError.into(),
-            "not yet implemented",
-        ))
+    async fn new_session(&self, args: NewSessionRequest) -> Result<NewSessionResponse> {
+        new_session::handle(self, args).await
     }
 
     async fn load_session(&self, _args: LoadSessionRequest) -> Result<LoadSessionResponse> {
@@ -109,7 +164,7 @@ mod tests {
     use crate::config::Config;
     use agent_client_protocol::{
         Agent, AuthenticateRequest, CancelNotification, ExtNotification, ExtRequest,
-        LoadSessionRequest, NewSessionRequest, PromptRequest, SetSessionModeRequest,
+        LoadSessionRequest, PromptRequest, SetSessionModeRequest,
     };
     use trogon_nats::AdvancedMockNatsClient;
 
@@ -137,12 +192,6 @@ mod tests {
                 .await
                 .is_err()
         );
-        assert!(
-            bridge
-                .new_session(NewSessionRequest::new("."))
-                .await
-                .is_err()
-        );
         assert!(
             bridge
                 .load_session(LoadSessionRequest::new("s1", "."))