Add ACP IO Nats Bridge

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

Author: yordis

rsworkspace/Cargo.lock

@@ -0,0 +1,23 @@
+[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"] }
+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"] }
+trogon-std = { path = "../trogon-std" }
+tokio = { version = "1.49.0", features = ["rt-multi-thread", "macros", "signal"] }
+tracing = "0.1.44"
+tracing-opentelemetry = "0.32.1"
+tracing-subscriber = { version = "0.3.22", features = ["env-filter", "fmt", "json"] }
+
+[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,81 @@
+mod config;
+mod otel;
+mod signal;
+
+use acp_nats::{agent::Bridge, client, nats};
+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,193 @@
+use acp_nats::Config;
+use opentelemetry::KeyValue;
+use opentelemetry::trace::TracerProvider;
+use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
+use opentelemetry_otlp::{LogExporter, MetricExporter, SpanExporter};
+use opentelemetry_sdk::logs::SdkLoggerProvider;
+use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
+use opentelemetry_sdk::propagation::TraceContextPropagator;
+use opentelemetry_sdk::{Resource, trace as sdktrace};
+use std::fs::File;
+use std::path::PathBuf;
+use std::sync::OnceLock;
+use std::time::Duration;
+use tracing_subscriber::EnvFilter;
+use tracing_subscriber::fmt::format::FmtSpan;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+
+static TRACER_PROVIDER: OnceLock<sdktrace::SdkTracerProvider> = OnceLock::new();
+static METER_PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();
+static LOGGER_PROVIDER: OnceLock<SdkLoggerProvider> = OnceLock::new();
+
+pub fn init_logger(config: &Config) -> Result<(), Box<dyn std::error::Error>> {
+    let log_dir = get_log_dir()?;
+    let log_file = log_dir.join("acp-io-bridge-nats.log");
+
+    let file = File::create(&log_file)?;
+
+    let file_layer = tracing_subscriber::fmt::layer()
+        .with_writer(file)
+        .with_thread_ids(true)
+        .with_span_events(FmtSpan::CLOSE)
+        .json();
+
+    let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+
+    match try_init_otel(config) {
+        Ok((tracer_provider, meter_provider, logger_provider)) => {
+            opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
+
+            let tracer = tracer_provider.tracer(DEFAULT_SERVICE_NAME);
+            let otel_trace_layer = tracing_opentelemetry::layer().with_tracer(tracer);
+
+            opentelemetry::global::set_meter_provider(meter_provider.clone());
+            METER_PROVIDER.set(meter_provider).ok();
+
+            let otel_logs_layer = OpenTelemetryTracingBridge::new(&logger_provider);
+            LOGGER_PROVIDER.set(logger_provider).ok();
+
+            tracing_subscriber::registry()
+                .with(env_filter)
+                .with(file_layer)
+                .with(otel_trace_layer)
+                .with(otel_logs_layer)
+                .init();
+
+            tracing::info!(
+                log_file = %log_file.display(),
+                "Logger initialized with OpenTelemetry"
+            );
+        }
+        Err(e) => {
+            tracing_subscriber::registry()
+                .with(env_filter)
+                .with(file_layer)
+                .init();
+
+            tracing::warn!(
+                log_file = %log_file.display(),
+                error = %e,
+                "Logger initialized without OpenTelemetry (init failed)"
+            );
+        }
+    }
+
+    Ok(())
+}
+
+const DEFAULT_SERVICE_NAME: &str = "acp-nats-stdio";
+
+fn try_init_otel(
+    config: &Config,
+) -> Result<
+    (
+        sdktrace::SdkTracerProvider,
+        SdkMeterProvider,
+        SdkLoggerProvider,
+    ),
+    Box<dyn std::error::Error>,
+> {
+    let mut builder = Resource::builder()
+        .with_attributes(vec![KeyValue::new("acp.prefix", config.acp_prefix.clone())]);
+
+    if std::env::var("OTEL_SERVICE_NAME").is_err() {
+        builder = builder.with_service_name(DEFAULT_SERVICE_NAME);
+    }
+
+    let resource = builder.build();
+
+    let tracer_provider = init_tracer_provider(&resource)?;
+    let meter_provider = init_meter_provider(&resource)?;
+    let logger_provider = init_logger_provider(&resource)?;
+
+    Ok((tracer_provider, meter_provider, logger_provider))
+}
+
+fn init_tracer_provider(
+    resource: &Resource,
+) -> Result<sdktrace::SdkTracerProvider, Box<dyn std::error::Error>> {
+    let exporter = SpanExporter::builder().with_http().build()?;
+
+    let provider = sdktrace::SdkTracerProvider::builder()
+        .with_batch_exporter(exporter)
+        .with_resource(resource.clone())
+        .build();
+
+    opentelemetry::global::set_tracer_provider(provider.clone());
+    TRACER_PROVIDER.set(provider.clone()).ok();
+
+    Ok(provider)
+}
+
+fn init_meter_provider(
+    resource: &Resource,
+) -> Result<SdkMeterProvider, Box<dyn std::error::Error>> {
+    let exporter = MetricExporter::builder().with_http().build()?;
+
+    let reader = PeriodicReader::builder(exporter)
+        .with_interval(Duration::from_secs(30))
+        .build();
+
+    let provider = SdkMeterProvider::builder()
+        .with_reader(reader)
+        .with_resource(resource.clone())
+        .build();
+
+    Ok(provider)
+}
+
+fn init_logger_provider(
+    resource: &Resource,
+) -> Result<SdkLoggerProvider, Box<dyn std::error::Error>> {
+    let exporter = LogExporter::builder().with_http().build()?;
+
+    let provider = SdkLoggerProvider::builder()
+        .with_batch_exporter(exporter)
+        .with_resource(resource.clone())
+        .build();
+
+    Ok(provider)
+}
+
+fn get_log_dir() -> Result<PathBuf, Box<dyn std::error::Error>> {
+    let base = std::env::var("XDG_CACHE_HOME")
+        .map(PathBuf::from)
+        .unwrap_or_else(|_| {
+            std::env::var("HOME")
+                .map(|h| PathBuf::from(h).join(".cache"))
+                .unwrap_or_else(|_| std::env::temp_dir())
+        });
+
+    let log_dir = base.join("acp-io-bridge-nats").join("logs");
+    std::fs::create_dir_all(&log_dir)?;
+    Ok(log_dir)
+}
+
+pub fn get_log_file_path() -> Result<PathBuf, Box<dyn std::error::Error>> {
+    let log_dir = get_log_dir()?;
+    Ok(log_dir.join("acp-io-bridge-nats.log"))
+}
+
+pub async fn shutdown_otel() {
+    tracing::info!("Shutting down OpenTelemetry providers");
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    if let Some(logger_provider) = LOGGER_PROVIDER.get() {
+        if let Err(e) = logger_provider.shutdown() {
+            eprintln!("Failed to shutdown logger provider: {e}");
+        }
+    }
+
+    if let Some(meter_provider) = METER_PROVIDER.get() {
+        if let Err(e) = meter_provider.shutdown() {
+            eprintln!("Failed to shutdown meter provider: {e}");
+        }
+    }
+
+    if let Some(tracer_provider) = TRACER_PROVIDER.get() {
+        if let Err(e) = tracer_provider.shutdown() {
+            eprintln!("Failed to shutdown tracer provider: {e}");
+        }
+    }
+}