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,182 @@
+use acp_nats::{Config, NatsAuth};
+use clap::Parser;
+use std::path::PathBuf;
+use trogon_std::env::ReadEnv;
+
+#[cfg(test)]
+use trogon_std::env::InMemoryEnv;
+
+const ENV_NATS_URL: &str = "NATS_URL";
+const ENV_NATS_CREDS: &str = "NATS_CREDS";
+const ENV_NATS_NKEY: &str = "NATS_NKEY";
+const ENV_NATS_USER: &str = "NATS_USER";
+const ENV_NATS_PASSWORD: &str = "NATS_PASSWORD";
+const ENV_NATS_TOKEN: &str = "NATS_TOKEN";
+const ENV_OTEL_SERVICE_NAME: &str = "OTEL_SERVICE_NAME";
+const ENV_ACP_PREFIX: &str = "ACP_PREFIX";
+
+const DEFAULT_NATS_URL: &str = "localhost:4222";
+const DEFAULT_SERVICE_NAME: &str = "acp-nats-stdio";
+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 {
+    /// Prefix for NATS subjects. Replaces "acp" as the root token to support multi-tenancy.
+    #[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(
+        env_provider
+            .var(ENV_OTEL_SERVICE_NAME)
+            .unwrap_or_else(|_| DEFAULT_SERVICE_NAME.to_string()),
+        args.acp_prefix
+            .unwrap_or_else(|| DEFAULT_ACP_PREFIX.to_string()),
+        nats_servers_from_provider(env_provider),
+        nats_auth_from_provider(env_provider),
+    )
+}
+
+fn nats_servers_from_provider<E: ReadEnv>(env_provider: &E) -> Vec<String> {
+    let servers_str = env_provider
+        .var(ENV_NATS_URL)
+        .unwrap_or_else(|_| DEFAULT_NATS_URL.to_string());
+    servers_str
+        .split(',')
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+        .collect()
+}
+
+fn nats_auth_from_provider<E: ReadEnv>(env_provider: &E) -> NatsAuth {
+    if let Ok(creds_path) = env_provider.var(ENV_NATS_CREDS) {
+        return NatsAuth::Credentials(PathBuf::from(creds_path));
+    }
+    if let Ok(nkey) = env_provider.var(ENV_NATS_NKEY) {
+        return NatsAuth::NKey(nkey);
+    }
+    if let (Ok(user), Ok(password)) = (
+        env_provider.var(ENV_NATS_USER),
+        env_provider.var(ENV_NATS_PASSWORD),
+    ) {
+        return NatsAuth::UserPassword { user, password };
+    }
+    if let Ok(token) = env_provider.var(ENV_NATS_TOKEN) {
+        return NatsAuth::Token(token);
+    }
+    NatsAuth::None
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_single_server() {
+        let env = InMemoryEnv::new();
+        env.set(ENV_NATS_URL, "localhost:4222");
+
+        let servers = nats_servers_from_provider(&env);
+        assert_eq!(servers, vec!["localhost:4222"]);
+    }
+
+    #[test]
+    fn test_parse_multiple_servers() {
+        let env = InMemoryEnv::new();
+        env.set(ENV_NATS_URL, "localhost:4222,localhost:4223,localhost:4224");
+
+        let servers = nats_servers_from_provider(&env);
+        assert_eq!(
+            servers,
+            vec!["localhost:4222", "localhost:4223", "localhost:4224"]
+        );
+    }
+
+    #[test]
+    fn test_parse_servers_with_spaces() {
+        let env = InMemoryEnv::new();
+        env.set(
+            ENV_NATS_URL,
+            "localhost:4222 , localhost:4223 , localhost:4224",
+        );
+
+        let servers = nats_servers_from_provider(&env);
+        assert_eq!(
+            servers,
+            vec!["localhost:4222", "localhost:4223", "localhost:4224"]
+        );
+    }
+
+    #[test]
+    fn test_default_server() {
+        let env = InMemoryEnv::new();
+
+        let servers = nats_servers_from_provider(&env);
+        assert_eq!(servers, vec!["localhost:4222"]);
+    }
+
+    #[test]
+    fn test_nats_auth_credentials() {
+        let env = InMemoryEnv::new();
+        env.set(ENV_NATS_CREDS, "/path/to/creds.file");
+
+        let auth = nats_auth_from_provider(&env);
+        assert!(matches!(auth, NatsAuth::Credentials(_)));
+    }
+
+    #[test]
+    fn test_nats_auth_nkey() {
+        let env = InMemoryEnv::new();
+        env.set(
+            ENV_NATS_NKEY,
+            "SUAIBDPBAUTWCWBKIO6XHQNINK5FWJW4OHLXC3HQ2KFE4PEJUA44CNHTC4",
+        );
+
+        let auth = nats_auth_from_provider(&env);
+        assert!(matches!(auth, NatsAuth::NKey(_)));
+    }
+
+    #[test]
+    fn test_nats_auth_user_password() {
+        let env = InMemoryEnv::new();
+        env.set(ENV_NATS_USER, "testuser");
+        env.set(ENV_NATS_PASSWORD, "testpass");
+
+        let auth = nats_auth_from_provider(&env);
+        assert!(matches!(auth, NatsAuth::UserPassword { .. }));
+    }
+
+    #[test]
+    fn test_nats_auth_token() {
+        let env = InMemoryEnv::new();
+        env.set(ENV_NATS_TOKEN, "test-token");
+
+        let auth = nats_auth_from_provider(&env);
+        assert!(matches!(auth, NatsAuth::Token(_)));
+    }
+
+    #[test]
+    fn test_nats_auth_none() {
+        let env = InMemoryEnv::new();
+
+        let auth = nats_auth_from_provider(&env);
+        assert!(matches!(auth, NatsAuth::None));
+    }
+
+    #[test]
+    fn test_nats_auth_priority_credentials_over_nkey() {
+        let env = InMemoryEnv::new();
+        env.set(ENV_NATS_CREDS, "/path/to/creds.file");
+        env.set(
+            ENV_NATS_NKEY,
+            "SUAIBDPBAUTWCWBKIO6XHQNINK5FWJW4OHLXC3HQ2KFE4PEJUA44CNHTC4",
+        );
+
+        let auth = nats_auth_from_provider(&env);
+        assert!(matches!(auth, NatsAuth::Credentials(_)));
+    }
+}

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

@@ -0,0 +1,80 @@
+mod config;
+mod runtime;
+
+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)");
+                }
+                _ = runtime::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/runtime.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"] }