feat(acp-nats): add prompt handler (#20)

- Add prompt handler with session validation, cancel pre-flight, backpressure
- Expand Bridge with CancelledSessions, PendingSessionPromptResponseWaiters
- Update cancel to mark sessions cancelled and resolve pending prompt waiters
- Add prompt_timeout and max_concurrent_client_tasks to config
- Add session_prompt NATS subject

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

Author: yordis

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

@@ -1,13 +1,16 @@
 use super::Bridge;
+use agent_client_protocol::{CancelNotification, PromptResponse, Result, StopReason};
 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::{Error, ErrorCode};
 use tracing::{info, instrument, warn};
 use trogon_std::time::GetElapsed;
 
-/// Publishes the cancel notification to the backend via NATS (fire-and-forget).
-/// The publish failure is logged and recorded as a metric but does not propagate
-/// to the caller, so the client always receives `Ok(())`.
+/// Handles cancel notification requests.
+///
+/// Marks the session as cancelled, resolves any pending prompt waiters, and publishes
+/// the cancellation to the backend. The backend publish is fire-and-forget - if it fails,
+/// the error is logged and recorded in metrics, but the method still returns `Ok(())`.
 #[instrument(
     name = "acp.session.cancel",
     skip(bridge, args),
@@ -22,18 +25,26 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     info!(session_id = %args.session_id, "Cancel notification");
 
     AcpSessionId::try_from(&args.session_id).map_err(|e| {
-        bridge
-            .metrics
-            .record_request("cancel", bridge.clock.elapsed(start).as_secs_f64(), false);
-        bridge
-            .metrics
-            .record_error("session_validate", "invalid_session_id");
+        bridge.metrics.record_request(
+            "cancel",
+            bridge.clock.elapsed(start).as_secs_f64(),
+            false,
+        );
+        bridge.metrics.record_error("cancel", "invalid_session_id");
         Error::new(
             ErrorCode::InvalidParams.into(),
             format!("Invalid session ID: {}", e),
         )
     })?;
 
+    bridge
+        .cancelled_sessions
+        .mark_cancelled(args.session_id.clone(), &bridge.clock);
+
+    bridge
+        .pending_session_prompt_responses
+        .resolve_waiter(&args.session_id, Ok(PromptResponse::new(StopReason::Cancelled)));
+
     let subject = agent::session_cancel(bridge.config.acp_prefix(), &args.session_id.to_string());
 
     let publish_result = nats::publish(
@@ -46,20 +57,24 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     )
     .await;
 
-    if let Err(error) = publish_result {
-        warn!(
-            session_id = %args.session_id,
-            error = %error,
-            "Failed to publish cancel notification to backend"
-        );
-        bridge
-            .metrics
-            .record_error("cancel", "cancel_publish_failed");
-    }
+    let publish_ok = match publish_result {
+        Ok(()) => true,
+        Err(error) => {
+            warn!(
+                session_id = %args.session_id,
+                error = %error,
+                "Failed to publish cancel notification to backend"
+            );
+            bridge.metrics.record_error("cancel", "cancel_publish_failed");
+            false
+        }
+    };
 
-    bridge
-        .metrics
-        .record_request("cancel", bridge.clock.elapsed(start).as_secs_f64(), true);
+    bridge.metrics.record_request(
+        "cancel",
+        bridge.clock.elapsed(start).as_secs_f64(),
+        publish_ok,
+    );
 
     Ok(())
 }
@@ -183,6 +198,20 @@ mod tests {
             .is_some()
     }
 
+    #[tokio::test]
+    async fn cancel_resolves_pending_prompt_waiter_with_cancelled() {
+        let (_mock, bridge) = mock_bridge();
+        let rx = bridge
+            .pending_session_prompt_responses
+            .register_waiter(agent_client_protocol::SessionId::from("s1"))
+            .unwrap();
+
+        let _ = bridge.cancel(CancelNotification::new("s1")).await;
+
+        let result = rx.await.unwrap().unwrap();
+        assert_eq!(result.stop_reason, agent_client_protocol::StopReason::Cancelled);
+    }
+
     #[tokio::test]
     async fn cancel_publishes_to_correct_subject() {
         let (mock, bridge) = mock_bridge();
@@ -223,8 +252,8 @@ mod tests {
             "expected acp.request.count with method=cancel, success=false on validation failure"
         );
         assert!(
-            has_error_metric(&finished_metrics, "session_validate", "invalid_session_id"),
-            "expected acp.errors.total with operation=session_validate, reason=invalid_session_id"
+            has_error_metric(&finished_metrics, "cancel", "invalid_session_id"),
+            "expected acp.errors.total with operation=cancel, reason=invalid_session_id"
         );
         provider.shutdown().unwrap();
     }
@@ -258,8 +287,8 @@ mod tests {
             "expected acp.errors.total with operation=cancel, reason=cancel_publish_failed"
         );
         assert!(
-            has_request_metric(&finished_metrics, "cancel", true),
-            "publish failure is fire-and-forget; caller still gets Ok, so success=true"
+            has_request_metric(&finished_metrics, "cancel", false),
+            "request metric records publish outcome; success=false when publish fails"
         );
         provider.shutdown().unwrap();
     }