feat(acp-nats): add integration test suite for Bridge, handlers, and workflows

Add tests/ module with agent_handlers, builders, prefix, and workflows
integration tests covering session lifecycle, cancellation, and ext methods.

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

Author: yordis

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

@@ -187,12 +187,11 @@ mod tests {
     #[tokio::test]
     async fn authenticate_forwards_request_and_returns_response() {
         let (mock, bridge) = mock_bridge();
-        let expected = AuthenticateResponse::default();
+        let expected = AuthenticateResponse::new();
         set_json_response(&mock, "acp.agent.authenticate", &expected);
 
-        let request = AuthenticateRequest::new("test");
+        let request = AuthenticateRequest::new("api-key");
         let result = bridge.authenticate(request).await;
-
         assert!(result.is_ok());
     }
 

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

@@ -82,7 +82,7 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
 mod tests {
     use super::Bridge;
     use crate::config::Config;
-    use agent_client_protocol::{Agent, CancelNotification, ErrorCode};
+    use agent_client_protocol::{Agent, CancelNotification, ErrorCode, SessionId, StopReason};
     use opentelemetry::Value;
     use opentelemetry::metrics::MeterProvider;
     use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
@@ -91,6 +91,7 @@ mod tests {
     };
     use std::time::Duration;
     use trogon_nats::AdvancedMockNatsClient;
+    use trogon_std::time::MockClock;
 
     fn mock_bridge() -> (
         AdvancedMockNatsClient,
@@ -129,6 +130,22 @@ mod tests {
         (mock, bridge, exporter, provider)
     }
 
+    fn mock_bridge_with_clock() -> (
+        AdvancedMockNatsClient,
+        MockClock,
+        Bridge<AdvancedMockNatsClient, MockClock>,
+    ) {
+        let mock = AdvancedMockNatsClient::new();
+        let clock = MockClock::new();
+        let bridge = Bridge::new(
+            mock.clone(),
+            clock.clone(),
+            &opentelemetry::global::meter("acp-nats-test"),
+            Config::for_test("acp"),
+        );
+        (mock, clock, bridge)
+    }
+
     fn has_request_metric(
         finished_metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics],
         method: &str,
@@ -319,4 +336,102 @@ mod tests {
         ));
         provider.shutdown().unwrap();
     }
+
+    #[tokio::test]
+    async fn cancel_marks_session_as_cancelled() {
+        let (_mock, _clock, bridge) = mock_bridge_with_clock();
+        let session_id = "cancel-session-001";
+
+        assert!(
+            bridge
+                .cancelled_sessions
+                .take_if_cancelled(&session_id.into(), &bridge.clock)
+                .is_none()
+        );
+
+        let notification = CancelNotification::new(session_id);
+        bridge.cancel(notification).await.unwrap();
+
+        assert!(
+            bridge
+                .cancelled_sessions
+                .take_if_cancelled(&session_id.into(), &bridge.clock)
+                .is_some()
+        );
+    }
+
+    #[tokio::test]
+    async fn cancel_resolves_pending_prompt_waiter_with_cancelled() {
+        let (_mock, _clock, bridge) = mock_bridge_with_clock();
+        let session_id: SessionId = "cancel-session-002".into();
+
+        let (rx, _guard, _token) = bridge
+            .pending_session_prompt_responses
+            .register_waiter(session_id.clone())
+            .unwrap();
+
+        let notification = CancelNotification::new(session_id.clone());
+        bridge.cancel(notification).await.unwrap();
+
+        let response = rx
+            .await
+            .expect("Should receive cancelled response")
+            .expect("Prompt waiter should receive success response");
+        assert_eq!(response.stop_reason, StopReason::Cancelled);
+    }
+
+    #[tokio::test]
+    async fn cancel_publishes_to_nats() {
+        let (mock, _clock, bridge) = mock_bridge_with_clock();
+        let session_id = "cancel-session-003";
+
+        let notification = CancelNotification::new(session_id);
+        bridge.cancel(notification).await.unwrap();
+
+        let published = mock.published_messages();
+        assert!(
+            published.iter().any(|s| s.contains("session.cancel")),
+            "Expected cancel publish, got: {:?}",
+            published
+        );
+    }
+
+    #[tokio::test]
+    async fn cancel_session_evicts_expired_on_mark() {
+        let (_mock, clock, bridge) = mock_bridge_with_clock();
+
+        let session_old: SessionId = "old-session".into();
+        let session_new: SessionId = "new-session".into();
+
+        bridge
+            .cancelled_sessions
+            .mark_cancelled(session_old.clone(), &bridge.clock);
+
+        clock.advance(Duration::from_secs(301));
+
+        for idx in 0..15 {
+            let filler_session: SessionId = format!("filler-{idx}").into();
+            bridge
+                .cancelled_sessions
+                .mark_cancelled(filler_session, &bridge.clock);
+        }
+
+        bridge
+            .cancelled_sessions
+            .mark_cancelled(session_new.clone(), &bridge.clock);
+
+        assert!(
+            bridge
+                .cancelled_sessions
+                .take_if_cancelled(&session_old, &bridge.clock)
+                .is_none()
+        );
+
+        assert!(
+            bridge
+                .cancelled_sessions
+                .take_if_cancelled(&session_new, &bridge.clock)
+                .is_some()
+        );
+    }
 }

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

@@ -317,4 +317,21 @@ mod tests {
         assert!(err.to_string().contains("Initialize request failed"));
         assert_eq!(err.code, ErrorCode::InternalError);
     }
+
+    #[tokio::test]
+    async fn handlers_use_custom_prefix() {
+        let mock = AdvancedMockNatsClient::new();
+        let bridge = Bridge::new(
+            mock.clone(),
+            trogon_std::time::SystemClock,
+            &opentelemetry::global::meter("acp-nats-test"),
+            Config::for_test("myorg.prod"),
+        );
+        let expected = InitializeResponse::new(ProtocolVersion::LATEST);
+        set_json_response(&mock, "myorg.prod.agent.initialize", &expected);
+
+        let request = InitializeRequest::new(ProtocolVersion::LATEST);
+        let result = bridge.initialize(request).await;
+        assert!(result.is_ok());
+    }
 }

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

@@ -250,14 +250,24 @@ mod tests {
 
     #[tokio::test]
     async fn load_session_forwards_request_and_returns_response() {
-        let (mock, bridge) = mock_bridge();
-        let expected = LoadSessionResponse::new();
-        set_json_response(&mock, "acp.s1.agent.session.load", &expected);
-
-        let request = LoadSessionRequest::new("s1", ".");
-        let result = bridge.load_session(request).await;
-
-        assert!(result.is_ok());
+        let local = tokio::task::LocalSet::new();
+        local
+            .run_until(async {
+                let mock = AdvancedMockNatsClient::new();
+                let bridge = Bridge::new(
+                    mock.clone(),
+                    trogon_std::time::MockClock::new(),
+                    &opentelemetry::global::meter("acp-nats-test"),
+                    Config::for_test("acp"),
+                );
+                let expected = LoadSessionResponse::new();
+                set_json_response(&mock, "acp.session-load-001.agent.session.load", &expected);
+
+                let request = LoadSessionRequest::new("session-load-001", "/tmp");
+                let result = bridge.load_session(request).await;
+                assert!(result.is_ok());
+            })
+            .await;
     }
 
     #[tokio::test]

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

@@ -239,17 +239,61 @@ mod tests {
 
     #[tokio::test]
     async fn new_session_forwards_request_and_returns_response() {
-        let (mock, bridge) = mock_bridge();
-        let session_id = SessionId::from("test-session-1");
-        let expected = NewSessionResponse::new(session_id.clone());
-        set_json_response(&mock, "acp.agent.session.new", &expected);
+        let local = tokio::task::LocalSet::new();
+        local
+            .run_until(async {
+                let mock = AdvancedMockNatsClient::new();
+                let bridge = Bridge::new(
+                    mock.clone(),
+                    trogon_std::time::MockClock::new(),
+                    &opentelemetry::global::meter("acp-nats-test"),
+                    Config::for_test("acp"),
+                );
+                let expected = NewSessionResponse::new("session-001");
+                set_json_response(&mock, "acp.agent.session.new", &expected);
+
+                let request = NewSessionRequest::new("/tmp");
+                let result = bridge.new_session(request).await;
+
+                assert!(result.is_ok());
+                let response = result.unwrap();
+                assert_eq!(response.session_id.to_string(), "session-001");
+            })
+            .await;
+    }
 
-        let request = NewSessionRequest::new(".");
-        let result = bridge.new_session(request).await;
+    #[tokio::test]
+    async fn session_ready_publish_tasks_are_tracked_and_awaited() {
+        let mock = AdvancedMockNatsClient::new();
+        let expected = NewSessionResponse::new("session-ready-track");
+        set_json_response(&mock, "acp.agent.session.new", &expected);
 
-        assert!(result.is_ok());
-        let response = result.unwrap();
-        assert_eq!(response.session_id, session_id);
+        let local = tokio::task::LocalSet::new();
+        local
+            .run_until(async {
+                let bridge = Bridge::new(
+                    mock.clone(),
+                    trogon_std::time::MockClock::new(),
+                    &opentelemetry::global::meter("acp-nats-test"),
+                    Config::for_test("acp"),
+                );
+                let request = NewSessionRequest::new("/tmp");
+                let result = bridge.new_session(request).await;
+                assert!(result.is_ok());
+
+                assert!(
+                    bridge.has_pending_session_ready_tasks(),
+                    "session.ready task should be tracked"
+                );
+
+                bridge.await_session_ready_tasks().await;
+
+                assert!(
+                    !bridge.has_pending_session_ready_tasks(),
+                    "session.ready tasks should be fully awaited and cleared"
+                );
+            })
+            .await;
     }
 
     #[tokio::test]

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

@@ -191,11 +191,14 @@ mod tests {
     async fn set_session_mode_forwards_request_and_returns_response() {
         let (mock, bridge) = mock_bridge();
         let expected = SetSessionModeResponse::new();
-        set_json_response(&mock, "acp.s1.agent.session.set_mode", &expected);
+        set_json_response(
+            &mock,
+            "acp.session-mode-001.agent.session.set_mode",
+            &expected,
+        );
 
-        let request = SetSessionModeRequest::new("s1", "mode-1");
+        let request = SetSessionModeRequest::new("session-mode-001", "auto");
         let result = bridge.set_session_mode(request).await;
-
         assert!(result.is_ok());
     }
 

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

@@ -12,6 +12,8 @@ pub mod prompt_slot_counter;
 pub mod session_id;
 pub mod subject_token_violation;
 pub(crate) mod telemetry;
+#[cfg(test)]
+mod tests;
 
 pub use acp_prefix::{AcpPrefix, AcpPrefixError};
 pub use agent::Bridge;

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

@@ -306,4 +306,34 @@ mod tests {
         assert!(agent::session_set_mode(prefix, sid).starts_with(&expected_prefix));
         assert!(agent::ext_session_ready(prefix, sid).starts_with(&expected_prefix));
     }
+
+    #[test]
+    fn multi_tenant_prefixes_are_isolated() {
+        let sid = "sess123";
+        assert_ne!(agent::initialize("tenant1"), agent::initialize("tenant2"));
+        assert_ne!(
+            agent::session_prompt("tenant1", sid),
+            agent::session_prompt("tenant2", sid),
+        );
+    }
+
+    #[test]
+    fn environment_based_prefixes() {
+        assert_eq!(agent::initialize("dev"), "dev.agent.initialize");
+        assert_eq!(agent::initialize("prod"), "prod.agent.initialize");
+        assert_eq!(agent::initialize("staging"), "staging.agent.initialize");
+    }
+
+    #[test]
+    fn special_characters_in_prefix() {
+        assert_eq!(agent::initialize("my_app"), "my_app.agent.initialize");
+        assert_eq!(agent::initialize("my-app"), "my-app.agent.initialize");
+        assert_eq!(agent::initialize("app123"), "app123.agent.initialize");
+    }
+
+    #[test]
+    fn prefix_does_not_add_extra_namespace() {
+        assert_eq!(agent::initialize("myapp"), "myapp.agent.initialize");
+        assert_ne!(agent::initialize("myapp"), "myapp.acp.agent.initialize");
+    }
 }