feat(acp-nats): wire cancel/prompt/session handlers to Bridge infrastructure

- Wire cancel handler to mark sessions and resolve pending prompt waiters
- Add pre-flight cancellation check in prompt handler
- Replace inline publish_session_ready with bridge.spawn_session_ready()
  in new_session and load_session handlers

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

Author: yordis

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

@@ -1,7 +1,9 @@
 use super::Bridge;
 use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
 use crate::session_id::AcpSessionId;
-use agent_client_protocol::{CancelNotification, Error, ErrorCode, Result};
+use agent_client_protocol::{
+    CancelNotification, Error, ErrorCode, PromptResponse, Result, StopReason,
+};
 use tracing::{info, instrument, warn};
 use trogon_std::time::GetElapsed;
 
@@ -57,6 +59,16 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
             .record_error("cancel", "cancel_publish_failed");
     }
 
+    bridge
+        .cancelled_sessions
+        .mark_cancelled(args.session_id.clone(), &bridge.clock);
+    bridge
+        .pending_session_prompt_responses
+        .cancel_waiter_for_session(
+            &args.session_id,
+            Ok(PromptResponse::new(StopReason::Cancelled)),
+        );
+
     bridge.metrics.record_request(
         "cancel",
         bridge.clock.elapsed(start).as_secs_f64(),

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

@@ -1,13 +1,7 @@
 use super::Bridge;
-use crate::acp_prefix::AcpPrefix;
-use crate::config::SESSION_READY_DELAY;
 use crate::error::AGENT_UNAVAILABLE;
-use crate::nats::{
-    self, ExtSessionReady, FlushClient, FlushPolicy, PublishClient, PublishOptions, RequestClient,
-    RetryPolicy, agent,
-};
+use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
 use crate::session_id::AcpSessionId;
-use crate::telemetry::metrics::Metrics;
 use agent_client_protocol::{Error, ErrorCode, LoadSessionRequest, LoadSessionResponse, Result};
 use tracing::{info, instrument, warn};
 use trogon_nats::NatsError;
@@ -88,14 +82,7 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     .map_err(map_load_session_error);
 
     if result.is_ok() {
-        let nats = bridge.nats.clone();
-        let prefix = bridge.config.acp_prefix.clone();
-        let session_id = session_id.clone();
-        let metrics = bridge.metrics.clone();
-        // TODO: track the JoinHandle so we can drain in-flight publishes on graceful shutdown.
-        tokio::spawn(async move {
-            publish_session_ready(&nats, &prefix, &session_id, &metrics).await;
-        });
+        bridge.spawn_session_ready(&args.session_id);
     }
 
     bridge.metrics.record_request(
@@ -107,38 +94,6 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     result
 }
 
-async fn publish_session_ready<N: PublishClient + FlushClient>(
-    nats: &N,
-    prefix: &AcpPrefix,
-    session_id: &AcpSessionId,
-    metrics: &Metrics,
-) {
-    tokio::time::sleep(SESSION_READY_DELAY).await;
-
-    let subject = agent::ext_session_ready(prefix.as_str(), session_id.as_str());
-    info!(session_id = %session_id, subject = %subject, "Publishing session.ready");
-
-    let message = ExtSessionReady::new(agent_client_protocol::SessionId::from(
-        session_id.to_string(),
-    ));
-
-    let options = PublishOptions::builder()
-        .publish_retry_policy(RetryPolicy::standard())
-        .flush_policy(FlushPolicy::standard())
-        .build();
-
-    if let Err(e) = nats::publish(nats, &subject, &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");
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::{Bridge, map_load_session_error};

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

@@ -1,14 +1,7 @@
 use super::Bridge;
-use crate::config::SESSION_READY_DELAY;
 use crate::error::AGENT_UNAVAILABLE;
-use crate::nats::{
-    self, ExtSessionReady, FlushClient, FlushPolicy, PublishClient, PublishOptions, RequestClient,
-    RetryPolicy, agent,
-};
-use crate::telemetry::metrics::Metrics;
-use agent_client_protocol::{
-    Error, ErrorCode, NewSessionRequest, NewSessionResponse, Result, SessionId,
-};
+use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use agent_client_protocol::{Error, ErrorCode, NewSessionRequest, NewSessionResponse, Result};
 use tracing::{Span, info, instrument, warn};
 use trogon_nats::NatsError;
 use trogon_std::time::GetElapsed;
@@ -83,15 +76,7 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
         info!(session_id = %response.session_id, "Session created");
 
         bridge.metrics.record_session_created();
-
-        let nats = bridge.nats.clone();
-        let prefix = bridge.config.acp_prefix.clone();
-        let session_id = response.session_id.clone();
-        let metrics = bridge.metrics.clone();
-        // TODO: track the JoinHandle so we can drain in-flight publishes on graceful shutdown.
-        tokio::spawn(async move {
-            publish_session_ready(&nats, prefix.as_str(), &session_id, &metrics).await;
-        });
+        bridge.spawn_session_ready(&response.session_id);
     }
 
     bridge.metrics.record_request(
@@ -103,36 +88,6 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     result
 }
 
-async fn publish_session_ready<N: PublishClient + FlushClient>(
-    nats: &N,
-    prefix: &str,
-    session_id: &SessionId,
-    metrics: &Metrics,
-) {
-    tokio::time::sleep(SESSION_READY_DELAY).await;
-
-    let subject = agent::ext_session_ready(prefix, &session_id.to_string());
-    info!(session_id = %session_id, subject = %subject, "Publishing session.ready");
-
-    let message = ExtSessionReady::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, &subject, &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");
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::{Bridge, map_new_session_error};

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

@@ -84,6 +84,17 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
             Err(error) => return Err(invalid_session_id_error(bridge, error)),
         };
 
+        if bridge
+            .cancelled_sessions
+            .take_if_cancelled(&args.session_id, &bridge.clock)
+            .is_some()
+        {
+            info!(session_id = %args.session_id, "Prompt pre-flight: session already cancelled");
+            return Ok(PromptResponse::new(
+                agent_client_protocol::StopReason::Cancelled,
+            ));
+        }
+
         let nats = bridge.nats();
         let subject = agent::session_prompt(bridge.config.acp_prefix(), session_id.as_str());