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

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

Author: yordis

rsworkspace/crates/AGENTS.md

@@ -1,6 +1,6 @@
 Prefer domain-specific value objects over primitives (e.g. `AcpPrefix` not `String`). Each type's factory must guarantee correctness at construction—invalid instances should be unrepresentable. Validate per-type, not per-aggregate: avoid validating unrelated fields together in a single constructor.
 
-Every value object lives in its own file named after the type (e.g. `acp_prefix.rs`, `ext_method_name.rs`, `session_id.rs`). Never inline a value object into a config, aggregate, or service file.
+Every value object lives in its own file named after the type (e.g. `acp_prefix.rs`, `ext_method_name.rs`, `session_id.rs`). Never inline a value object into a config, aggregate, or service file. File layout: `src/{type_snake_case}.rs`; export in `lib.rs` as `pub use {module}::{Type, TypeError}`.
 
 You must use the `test-support` feature to share test helpers between crates.
 Prefer one trait per operation over a single trait with multiple operations.

rsworkspace/crates/acp-nats/Cargo.toml

@@ -21,5 +21,6 @@ trogon-std = { path = "../trogon-std" }
 
 [dev-dependencies]
 opentelemetry_sdk = { version = "0.31.0", features = ["rt-tokio", "metrics", "testing"] }
+tokio = { version = "1.49.0", features = ["test-util"] }
 trogon-nats = { path = "../trogon-nats", features = ["test-support"] }
 trogon-std = { path = "../trogon-std", features = ["test-support"] }

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

@@ -5,9 +5,11 @@ use agent_client_protocol::{CancelNotification, Error, ErrorCode, Result};
 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.
+///
+/// Validates the session ID and publishes the cancellation to the backend (fire-and-forget).
+/// The backend owns session state and will respond to the in-flight prompt with `stopReason: cancelled`.
+/// Publish failure is logged and recorded in metrics but does not propagate to the caller.
 #[instrument(
     name = "acp.session.cancel",
     skip(bridge, args),
@@ -25,9 +27,7 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
         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_error("cancel", "invalid_session_id");
         Error::new(
             ErrorCode::InvalidParams.into(),
             format!("Invalid session ID: {}", e),
@@ -46,7 +46,7 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     )
     .await;
 
-    if let Err(error) = publish_result {
+    if let Err(error) = &publish_result {
         warn!(
             session_id = %args.session_id,
             error = %error,
@@ -57,9 +57,11 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
             .record_error("cancel", "cancel_publish_failed");
     }
 
-    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_result.is_ok(),
+    );
 
     Ok(())
 }
@@ -223,8 +225,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 +260,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();
     }

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

@@ -1,5 +1,6 @@
 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,
@@ -8,13 +9,10 @@ use crate::nats::{
 use crate::session_id::AcpSessionId;
 use crate::telemetry::metrics::Metrics;
 use agent_client_protocol::{Error, ErrorCode, LoadSessionRequest, LoadSessionResponse, Result};
-use std::time::Duration;
 use tracing::{info, instrument, warn};
 use trogon_nats::NatsError;
 use trogon_std::time::GetElapsed;
 
-const SESSION_READY_DELAY: Duration = Duration::from_millis(100);
-
 fn map_load_session_error(e: NatsError) -> Error {
     match &e {
         NatsError::Timeout { subject } => {

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

@@ -5,14 +5,16 @@ mod ext_notification;
 mod initialize;
 mod load_session;
 mod new_session;
+mod prompt;
 mod set_session_mode;
 
 use crate::config::Config;
 use crate::nats::{FlushClient, PublishClient, RequestClient};
+use crate::pending_prompt_waiters::PendingSessionPromptResponseWaiters;
+use crate::prompt_slot_counter::PromptSlotCounter;
 use crate::telemetry::metrics::Metrics;
-use agent_client_protocol::ErrorCode;
 use agent_client_protocol::{
-    Agent, AuthenticateRequest, AuthenticateResponse, CancelNotification, Error, ExtNotification,
+    Agent, AuthenticateRequest, AuthenticateResponse, CancelNotification, ExtNotification,
     ExtRequest, ExtResponse, InitializeRequest, InitializeResponse, LoadSessionRequest,
     LoadSessionResponse, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse,
     Result, SetSessionModeRequest, SetSessionModeResponse,
@@ -23,17 +25,22 @@ use trogon_std::time::GetElapsed;
 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) pending_session_prompt_responses: PendingSessionPromptResponseWaiters<C::Instant>,
+    pub(crate) prompt_slot_counter: PromptSlotCounter,
+    pub(crate) config: Config,
 }
 
 impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Bridge<N, C> {
     pub fn new(nats: N, clock: C, meter: &Meter, config: Config) -> Self {
+        let max_concurrent = config.max_concurrent_client_tasks();
         Self {
             nats,
             clock,
             config,
             metrics: Metrics::new(meter),
+            pending_session_prompt_responses: PendingSessionPromptResponseWaiters::new(),
+            prompt_slot_counter: PromptSlotCounter::new(max_concurrent),
         }
     }
 
@@ -67,11 +74,8 @@ impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Agent for Br
         set_session_mode::handle(self, args).await
     }
 
-    async fn prompt(&self, _args: PromptRequest) -> Result<PromptResponse> {
-        Err(Error::new(
-            ErrorCode::InternalError.into(),
-            "not yet implemented",
-        ))
+    async fn prompt(&self, args: PromptRequest) -> Result<PromptResponse> {
+        prompt::handle(self, args).await
     }
 
     async fn cancel(&self, args: CancelNotification) -> Result<()> {
@@ -88,37 +92,14 @@ impl<N: RequestClient + PublishClient + FlushClient, C: GetElapsed> Agent for Br
 }
 
 #[cfg(test)]
-mod tests {
+mod send_sync_tests {
     use super::Bridge;
-    use crate::config::Config;
-    use agent_client_protocol::{Agent, PromptRequest};
     use trogon_nats::AdvancedMockNatsClient;
+    use trogon_std::time::SystemClock;
 
-    fn mock_bridge() -> Bridge<AdvancedMockNatsClient, trogon_std::time::SystemClock> {
-        Bridge::new(
-            AdvancedMockNatsClient::new(),
-            trogon_std::time::SystemClock,
-            &opentelemetry::global::meter("acp-nats-test"),
-            Config::for_test("acp"),
-        )
-    }
-
-    #[tokio::test]
-    async fn stub_methods_return_not_implemented() {
-        let bridge = mock_bridge();
-        let msg = "not yet implemented";
-
-        assert!(
-            bridge
-                .prompt(PromptRequest::new("s1", vec![]))
-                .await
-                .is_err()
-        );
-
-        let err = bridge
-            .prompt(PromptRequest::new("s1", vec![]))
-            .await
-            .unwrap_err();
-        assert!(err.to_string().contains(msg));
+    #[test]
+    fn bridge_is_send_and_sync() {
+        fn assert_send_sync<T: Send + Sync>() {}
+        assert_send_sync::<Bridge<AdvancedMockNatsClient, SystemClock>>();
     }
 }

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

@@ -1,4 +1,5 @@
 use super::Bridge;
+use crate::config::SESSION_READY_DELAY;
 use crate::error::AGENT_UNAVAILABLE;
 use crate::nats::{
     self, ExtSessionReady, FlushClient, FlushPolicy, PublishClient, PublishOptions, RequestClient,
@@ -8,25 +9,10 @@ use crate::telemetry::metrics::Metrics;
 use agent_client_protocol::{
     Error, ErrorCode, NewSessionRequest, NewSessionResponse, Result, SessionId,
 };
-use std::time::Duration;
 use tracing::{Span, info, instrument, warn};
 use trogon_nats::NatsError;
 use trogon_std::time::GetElapsed;
 
-/// Delay before publishing `session.ready` to NATS.
-///
-/// The `Agent` trait returns the response value *before* the transport layer
-/// serializes and writes it to the client. Without a delay the spawned task
-/// could publish `session.ready` to NATS before the client has received the
-/// `session/new` response, violating the ordering guarantee documented on
-/// [`ExtSessionReady`].
-///
-/// A post-send callback from the transport would be the ideal fix, but the
-/// external `agent_client_protocol` crate does not expose one. This constant
-/// delay provides a practical safety margin (serialization + write is typically
-/// sub-millisecond).
-const SESSION_READY_DELAY: Duration = Duration::from_millis(100);
-
 fn map_new_session_error(e: NatsError) -> Error {
     match &e {
         NatsError::Timeout { subject } => {