Add ACP IO Nats Bridge

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

Author: yordis

rsworkspace/Cargo.lock

@@ -0,0 +1,17 @@
+[package]
+name = "acp-nats-stdio"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+acp-nats = { path = "../acp-nats" }
+agent-client-protocol = "0.9.3"
+async-compat = "0.2.5"
+async-nats = "0.45.0"
+clap = { version = "4.5", features = ["derive", "env"] }
+trogon-std = { path = "../trogon-std" }
+tokio = { version = "1.49.0", features = ["rt-multi-thread", "macros"] }
+tracing = "0.1.44"
+
+[dev-dependencies]
+trogon-std = { path = "../trogon-std", features = ["test-support"] }

rsworkspace/crates/acp-nats-stdio/Cargo.toml

@@ -0,0 +1,47 @@
+# ACP NATS Bridge
+
+A Rust bridge that connects ACP (Agent Client Protocol) clients to NATS-based backends.
+
+```mermaid
+graph LR
+    A[IDE] <-->|stdio| B[acp-nats-bridge]
+    B <-->|NATS| C[Backend]
+    
+    style A fill:#e1f5ff,stroke:#0288d1
+    style B fill:#fff3e0,stroke:#f57c00
+    style C fill:#f3e5f5,stroke:#7b1fa2
+```
+
+## Features
+
+- Bidirectional ACP Bridge with request forwarding
+- Auto-reconnect with exponential backoff
+- OpenTelemetry integration (logs, metrics, traces)
+- Graceful shutdown (SIGINT/SIGTERM)
+- Custom prefix support for multi-tenancy
+
+## Quick Start
+
+```bash
+# Prerequisites: NATS server running
+docker run -p 4222:4222 nats:latest
+
+# Build
+cargo build --release
+
+# Run
+./target/release/acp-nats-stdio
+```
+
+## Configuration
+
+Configure via environment variables:
+
+- `NATS_URL` - NATS server URL(s). Single server: `localhost:4222` or multiple for failover: `localhost:4222,localhost:4223,localhost:4224` (default: `localhost:4222`)
+- `NATS_USER` - NATS username (optional)
+- `NATS_PASSWORD` - NATS password (optional)
+- `NATS_TOKEN` - NATS token (optional)
+- `CUSTOM_PREFIX` - Custom subject prefix for multi-tenancy (optional)
+- `OTEL_EXPORTER_OTLP_ENDPOINT` - OpenTelemetry collector endpoint (optional)
+
+See code documentation for additional configuration options.

rsworkspace/crates/acp-nats-stdio/README.md

@@ -0,0 +1,48 @@
+use acp_nats::{Config, NatsConfig};
+use clap::Parser;
+use trogon_std::env::ReadEnv;
+
+const ENV_ACP_PREFIX: &str = "ACP_PREFIX";
+const DEFAULT_ACP_PREFIX: &str = "acp";
+
+#[derive(Parser, Debug)]
+#[command(name = "acp-nats-stdio")]
+#[command(about = "ACP stdio to NATS bridge for agent-client protocol", long_about = None)]
+pub struct Args {
+    #[arg(long = "acp-prefix", env = ENV_ACP_PREFIX)]
+    pub acp_prefix: Option<String>,
+}
+
+pub fn from_env_with_provider<E: ReadEnv>(env_provider: &E) -> Config {
+    let args = Args::parse();
+    Config::new(
+        args.acp_prefix
+            .unwrap_or_else(|| DEFAULT_ACP_PREFIX.to_string()),
+        NatsConfig::from_env(env_provider),
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use trogon_std::env::InMemoryEnv;
+
+    #[test]
+    fn test_default_config() {
+        let env = InMemoryEnv::new();
+        let config = from_env_with_provider(&env);
+        assert_eq!(config.acp_prefix, DEFAULT_ACP_PREFIX);
+        assert_eq!(config.nats.servers, vec!["localhost:4222"]);
+        assert!(matches!(config.nats.auth, acp_nats::NatsAuth::None));
+    }
+
+    #[test]
+    fn test_nats_config_from_env() {
+        let env = InMemoryEnv::new();
+        env.set("NATS_URL", "host1:4222,host2:4222");
+        env.set("NATS_TOKEN", "my-token");
+        let config = from_env_with_provider(&env);
+        assert_eq!(config.nats.servers, vec!["host1:4222", "host2:4222"]);
+        assert!(matches!(config.nats.auth, acp_nats::NatsAuth::Token(t) if t == "my-token"));
+    }
+}

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

@@ -0,0 +1,80 @@
+mod config;
+mod signal;
+
+use acp_nats::{agent::Bridge, client, nats, otel};
+use agent_client_protocol::AgentSideConnection;
+use async_nats::Client as NatsAsyncClient;
+use std::sync::Arc;
+use tracing::{error, info};
+use trogon_std::env::SystemEnv;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let config = config::from_env_with_provider(&SystemEnv);
+    otel::init_logger(&config)?;
+
+    let log_file = otel::get_log_file_path()?;
+    info!(log_file = %log_file.display(), "ACP bridge starting");
+
+    let nats_client = match nats::connect(&config.nats).await {
+        Ok(client) => Some(client),
+        Err(e) => {
+            error!(error = %e, "Failed to connect to NATS");
+            None
+        }
+    };
+    let local = tokio::task::LocalSet::new();
+
+    local
+        .run_until(async {
+            let stdin = async_compat::Compat::new(tokio::io::stdin());
+            let stdout = async_compat::Compat::new(tokio::io::stdout());
+
+            let bridge = Arc::new(Bridge::<NatsAsyncClient>::new(
+                nats_client.clone(),
+                config.acp_prefix.clone(),
+            ));
+
+            let (connection, io_task) =
+                AgentSideConnection::new(bridge.clone(), stdout, stdin, |fut| {
+                    tokio::task::spawn_local(fut);
+                });
+
+            let connection = Arc::new(connection);
+
+            if let Some(nats_instance) = nats_client {
+                let client_connection = connection.clone();
+                let bridge_for_client = bridge.clone();
+                tokio::task::spawn_local(async move {
+                    client::run::<NatsAsyncClient, _>(
+                        nats_instance,
+                        client_connection,
+                        bridge_for_client,
+                    )
+                    .await;
+                });
+                info!("ACP bridge running on stdio with NATS client proxy");
+            } else {
+                info!("ACP bridge running on stdio (no NATS)");
+            }
+
+            tokio::select! {
+                result = io_task => {
+                    if let Err(e) = result {
+                        error!(error = %e, "IO task error");
+                    }
+                    info!("ACP bridge shutting down (IO closed)");
+                }
+                _ = signal::shutdown_signal() => {
+                    info!("ACP bridge shutting down (signal received)");
+                }
+            }
+        })
+        .await;
+
+    info!("Flushing telemetry...");
+    otel::shutdown_otel().await;
+    info!("ACP bridge stopped");
+
+    Ok(())
+}

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

@@ -0,0 +1,30 @@
+use tokio::signal;
+use tracing::warn;
+
+pub async fn shutdown_signal() {
+    let ctrl_c = async {
+        signal::ctrl_c()
+            .await
+            .expect("Failed to install Ctrl+C handler");
+    };
+
+    #[cfg(unix)]
+    let terminate = async {
+        signal::unix::signal(signal::unix::SignalKind::terminate())
+            .expect("Failed to install SIGTERM handler")
+            .recv()
+            .await;
+    };
+
+    #[cfg(not(unix))]
+    let terminate = std::future::pending::<()>();
+
+    tokio::select! {
+        _ = ctrl_c => {
+            warn!("Received SIGINT (Ctrl+C)");
+        }
+        _ = terminate => {
+            warn!("Received SIGTERM");
+        }
+    }
+}

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

@@ -0,0 +1,28 @@
+[package]
+name = "acp-nats"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+agent-client-protocol = "0.9.3"
+async-nats = "0.45.0"
+async-trait = "0.1.89"
+bytes = "1"
+futures = "0.3.31"
+opentelemetry = "0.31.0"
+opentelemetry-appender-tracing = "0.31.0"
+opentelemetry-otlp = { version = "0.31.0", features = ["http-json", "logs", "metrics", "reqwest-rustls"] }
+opentelemetry_sdk = { version = "0.31.0", features = ["rt-tokio", "logs", "metrics", "testing"] }
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = "1.0.149"
+tokio = { version = "1.49.0", features = ["process", "io-util", "io-std", "rt-multi-thread", "macros", "sync", "signal"] }
+tracing = "0.1.44"
+tracing-opentelemetry = "0.32.1"
+tracing-subscriber = { version = "0.3.22", features = ["env-filter", "fmt", "json"] }
+uuid = { version = "1.16", features = ["v4"] }
+
+# Shared NATS infrastructure
+trogon-nats = { path = "../trogon-nats" }
+
+[dev-dependencies]
+trogon-nats = { path = "../trogon-nats", features = ["test-support"] }

rsworkspace/crates/acp-nats/Cargo.toml

@@ -0,0 +1,37 @@
+use super::Bridge;
+use crate::nats::{self, agent, FlushClient, PublishClient, RequestClient, SubscribeClient};
+use agent_client_protocol::{AuthenticateRequest, AuthenticateResponse, Error, Result};
+use std::time::Instant;
+use tracing::{info, instrument};
+
+#[instrument(
+    name = "acp.authenticate",
+    skip(bridge, args),
+    fields(method_id = %args.method_id)
+)]
+pub async fn handle<N: SubscribeClient + RequestClient + PublishClient + FlushClient>(
+    bridge: &Bridge<N>,
+    args: AuthenticateRequest,
+) -> Result<AuthenticateResponse> {
+    let start = Instant::now();
+
+    info!(method_id = %args.method_id, "Authenticate request");
+
+    let nats = bridge.require_nats()?;
+
+    let result = nats::request::<N, AuthenticateRequest, AuthenticateResponse>(
+        nats,
+        &agent::authenticate(&bridge.acp_prefix),
+        &args,
+    )
+    .await
+    .map_err(|e| Error::new(-32603, e.to_string()));
+
+    bridge.metrics.record_request(
+        "authenticate",
+        start.elapsed().as_secs_f64(),
+        result.is_ok(),
+    );
+
+    result
+}

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

@@ -0,0 +1,34 @@
+use super::Bridge;
+use crate::nats::{self, agent, FlushClient, PublishClient, RequestClient, SubscribeClient};
+use agent_client_protocol::{CancelNotification, Result};
+use std::time::Instant;
+use tracing::{info, instrument, warn};
+
+#[instrument(
+    name = "acp.session.cancel",
+    skip(bridge, args),
+    fields(session_id = %args.session_id)
+)]
+pub async fn handle<N: SubscribeClient + RequestClient + PublishClient + FlushClient>(bridge: &Bridge<N>, args: CancelNotification) -> Result<()> {
+    let start = Instant::now();
+
+    info!(session_id = %args.session_id, "Cancel notification");
+
+    bridge
+        .cancelled_sessions
+        .mark_cancelled(args.session_id.clone());
+
+    if let Some(nats) = &bridge.nats {
+        let subject = agent::session_cancel(&bridge.acp_prefix, &args.session_id.to_string());
+
+        if let Err(e) = nats::publish(nats, &subject, &args, nats::PublishOptions::simple()).await {
+            warn!(error = %e, session_id = %args.session_id, "Failed to publish cancel notification");
+        }
+    }
+
+    bridge
+        .metrics
+        .record_request("cancel", start.elapsed().as_secs_f64(), true);
+
+    Ok(())
+}

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

@@ -0,0 +1,29 @@
+use super::Bridge;
+use crate::nats::{self, agent, FlushClient, PublishClient, RequestClient, SubscribeClient};
+use agent_client_protocol::{Error, ExtRequest, ExtResponse, Result};
+use std::time::Instant;
+use tracing::{info, instrument};
+
+#[instrument(
+    name = "acp.ext",
+    skip(bridge, args),
+    fields(method = %args.method)
+)]
+pub async fn handle<N: SubscribeClient + RequestClient + PublishClient + FlushClient>(bridge: &Bridge<N>, args: ExtRequest) -> Result<ExtResponse> {
+    let start = Instant::now();
+
+    info!(method = %args.method, "Extension method request");
+
+    let nats = bridge.require_nats()?;
+    let subject = agent::ext(&bridge.acp_prefix, &args.method);
+
+    let result = nats::request::<N, ExtRequest, ExtResponse>(nats, &subject, &args)
+        .await
+        .map_err(|e| Error::new(-32603, e.to_string()));
+
+    bridge
+        .metrics
+        .record_request("ext_method", start.elapsed().as_secs_f64(), result.is_ok());
+
+    result
+}