feat(acp-nats): turn Bridge into protocol translator (ACP ↔ NATS)

Bridge no longer executes prompts — it publishes to a runner over NATS
and streams PromptEvents back to the editor.

Author: yordis

rsworkspace/Cargo.lock

@@ -1,7 +1,7 @@
 mod config;
 
-use acp_nats::{StdJsonSerialize, agent::Bridge, client, nats};
-use agent_client_protocol::AgentSideConnection;
+use acp_nats::{StdJsonSerialize, agent::Bridge, client, nats, spawn_notification_forwarder};
+use agent_client_protocol::{AgentSideConnection, SessionNotification};
 use async_nats::Client as NatsAsyncClient;
 use std::rc::Rc;
 use tracing::{error, info};
@@ -54,11 +54,13 @@ async fn run_bridge(
     let stdout = async_compat::Compat::new(tokio::io::stdout());
 
     let meter = acp_telemetry::meter("acp-io-bridge-nats");
+    let (notification_tx, notification_rx) = tokio::sync::mpsc::channel::<SessionNotification>(64);
     let bridge = Rc::new(Bridge::new(
         nats_client.clone(),
         SystemClock,
         &meter,
         config.clone(),
+        notification_tx,
     ));
 
     let (connection, io_task) = AgentSideConnection::new(bridge.clone(), stdout, stdin, |fut| {
@@ -67,6 +69,8 @@ async fn run_bridge(
 
     let connection = Rc::new(connection);
 
+    spawn_notification_forwarder(connection.clone(), notification_rx);
+
     let client_connection = connection.clone();
     let bridge_for_client = bridge.clone();
     let mut client_task = tokio::task::spawn_local(client::run(

rsworkspace/crates/acp-nats-stdio/src/main.rs

@@ -1,5 +1,5 @@
-use acp_nats::{StdJsonSerialize, agent::Bridge, client};
-use agent_client_protocol::AgentSideConnection;
+use acp_nats::{StdJsonSerialize, agent::Bridge, client, spawn_notification_forwarder};
+use agent_client_protocol::{AgentSideConnection, SessionNotification};
 use axum::extract::ws::{Message, WebSocket};
 use futures_util::stream::{SplitSink, SplitStream};
 use futures_util::{SinkExt, StreamExt};
@@ -34,11 +34,13 @@ pub async fn handle<N>(
     let outgoing = async_compat::Compat::new(agent_write);
 
     let meter = acp_telemetry::meter("acp-nats-ws");
+    let (notification_tx, notification_rx) = tokio::sync::mpsc::channel::<SessionNotification>(64);
     let bridge = Rc::new(Bridge::new(
         nats_client.clone(),
         SystemClock,
         &meter,
         config,
+        notification_tx,
     ));
 
     let (connection, io_task) =
@@ -48,6 +50,8 @@ pub async fn handle<N>(
 
     let connection = Rc::new(connection);
 
+    spawn_notification_forwarder(connection.clone(), notification_rx);
+
     let recv_pump = tokio::task::spawn_local(run_recv_pump(ws_receiver, ws_recv_write));
     let send_pump = tokio::task::spawn_local(run_send_pump(ws_sender, ws_send_read));
 

rsworkspace/crates/acp-nats-ws/src/connection.rs

@@ -7,16 +7,17 @@ edition = "2024"
 workspace = true
 
 [dependencies]
-agent-client-protocol = "0.9.4"
+agent-client-protocol = { version = "0.9.4", features = ["unstable_session_list", "unstable_session_model", "unstable_session_fork", "unstable_session_resume", "unstable_session_usage"] }
+opentelemetry = "0.31.0"
 async-nats = "0.45.0"
 async-trait = "0.1.89"
 bytes = "1.9"
 futures = "0.3"
-opentelemetry = "0.31.0"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.149"
 tokio = { version = "1.49.0", features = ["rt", "macros", "sync", "time"] }
 tracing = "0.1.44"
+uuid = { version = "1", features = ["v4"] }
 
 trogon-nats = { path = "../trogon-nats" }
 trogon-std = { path = "../trogon-std" }

rsworkspace/crates/acp-nats/Cargo.toml

@@ -1,6 +1,6 @@
 use super::Bridge;
 use crate::error::AGENT_UNAVAILABLE;
-use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use crate::nats::{self, RequestClient, agent};
 use agent_client_protocol::{AuthenticateRequest, AuthenticateResponse, Error, ErrorCode, Result};
 use tracing::{info, instrument, warn};
 use trogon_nats::NatsError;
@@ -51,7 +51,7 @@ fn map_authenticate_error(e: NatsError) -> Error {
     skip(bridge, args),
     fields(method_id = %args.method_id)
 )]
-pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapsed>(
+pub async fn handle<N: RequestClient, C: GetElapsed>(
     bridge: &Bridge<N, C>,
     args: AuthenticateRequest,
 ) -> Result<AuthenticateResponse> {
@@ -150,6 +150,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &meter,
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge, exporter, provider)
     }
@@ -164,6 +165,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &opentelemetry::global::meter("acp-nats-test"),
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge)
     }

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

@@ -1,9 +1,7 @@
 use super::Bridge;
-use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use crate::nats::{self, FlushClient, PublishClient, agent};
 use crate::session_id::AcpSessionId;
-use agent_client_protocol::{
-    CancelNotification, Error, ErrorCode, PromptResponse, Result, StopReason,
-};
+use agent_client_protocol::{CancelNotification, Error, ErrorCode, Result};
 use tracing::{info, instrument, warn};
 use trogon_std::time::GetElapsed;
 
@@ -17,7 +15,7 @@ use trogon_std::time::GetElapsed;
     skip(bridge, args),
     fields(session_id = %args.session_id)
 )]
-pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapsed>(
+pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed>(
     bridge: &Bridge<N, C>,
     args: CancelNotification,
 ) -> Result<()> {
@@ -59,15 +57,19 @@ 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)),
-        );
+    let cancelled_subject =
+        agent::session_cancelled(bridge.config.acp_prefix(), &args.session_id.to_string());
+    if let Err(e) = bridge
+        .nats()
+        .publish_with_headers(
+            cancelled_subject,
+            async_nats::HeaderMap::new(),
+            bytes::Bytes::new(),
+        )
+        .await
+    {
+        warn!(session_id = %args.session_id, error = %e, "Failed to publish session_cancelled broadcast");
+    }
 
     bridge.metrics.record_request(
         "cancel",
@@ -82,7 +84,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, SessionId, StopReason};
+    use agent_client_protocol::{Agent, CancelNotification, ErrorCode};
     use opentelemetry::Value;
     use opentelemetry::metrics::MeterProvider;
     use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
@@ -103,6 +105,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &opentelemetry::global::meter("acp-nats-test"),
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge)
     }
@@ -126,6 +129,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &meter,
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge, exporter, provider)
     }
@@ -142,6 +146,7 @@ mod tests {
             clock.clone(),
             &opentelemetry::global::meter("acp-nats-test"),
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, clock, bridge)
     }
@@ -228,6 +233,20 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn cancel_also_publishes_session_cancelled_broadcast() {
+        let (mock, bridge) = mock_bridge();
+
+        let _ = bridge.cancel(CancelNotification::new("s1")).await;
+
+        let published = mock.published_messages();
+        assert!(
+            published.contains(&"acp.s1.agent.session.cancelled".to_string()),
+            "expected publish to acp.s1.agent.session.cancelled (prompt broadcast), got: {:?}",
+            published
+        );
+    }
+
     #[tokio::test]
     async fn cancel_validates_session_id() {
         let (_mock, bridge) = mock_bridge();
@@ -337,49 +356,6 @@ 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();
@@ -395,43 +371,4 @@ mod tests {
             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/cancel.rs

@@ -1,7 +1,7 @@
 use super::Bridge;
 use crate::error::AGENT_UNAVAILABLE;
 use crate::ext_method_name::ExtMethodName;
-use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use crate::nats::{self, RequestClient, agent};
 use agent_client_protocol::{Error, ErrorCode, ExtRequest, ExtResponse, Result};
 use tracing::{info, instrument, warn};
 use trogon_nats::NatsError;
@@ -52,7 +52,7 @@ fn map_ext_method_error(e: NatsError) -> Error {
     skip(bridge, args),
     fields(method = %args.method)
 )]
-pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapsed>(
+pub async fn handle<N: RequestClient, C: GetElapsed>(
     bridge: &Bridge<N, C>,
     args: ExtRequest,
 ) -> Result<ExtResponse> {
@@ -121,6 +121,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &opentelemetry::global::meter("acp-nats-test"),
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge)
     }
@@ -144,6 +145,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &meter,
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge, exporter, provider)
     }

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

@@ -1,6 +1,6 @@
 use super::Bridge;
 use crate::ext_method_name::ExtMethodName;
-use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use crate::nats::{self, FlushClient, PublishClient, agent};
 use agent_client_protocol::{Error, ErrorCode, ExtNotification, Result};
 use tracing::{info, instrument, warn};
 use trogon_std::time::GetElapsed;
@@ -15,7 +15,7 @@ use trogon_std::time::GetElapsed;
     skip(bridge, args),
     fields(method = %args.method)
 )]
-pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapsed>(
+pub async fn handle<N: PublishClient + FlushClient, C: GetElapsed>(
     bridge: &Bridge<N, C>,
     args: ExtNotification,
 ) -> Result<()> {
@@ -95,6 +95,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &opentelemetry::global::meter("acp-nats-test"),
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge)
     }
@@ -118,6 +119,7 @@ mod tests {
             trogon_std::time::SystemClock,
             &meter,
             Config::for_test("acp"),
+            tokio::sync::mpsc::channel(1).0,
         );
         (mock, bridge, exporter, provider)
     }