feat(acp-nats-stdio): add stdio-to-NATS bridge binary crate

- New acp-nats-stdio crate bridging ACP agent-client-protocol over stdio to NATS
- CLI config with --acp-prefix arg and env var fallback
- Graceful shutdown on SIGINT/SIGTERM
- Full OpenTelemetry telemetry (traces, metrics, logs) with file logging fallback

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.4"
+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", "io-std"] }
+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,212 @@
+use acp_nats::{AcpPrefix, AcpPrefixError, Config, NatsConfig};
+use clap::Parser;
+use std::time::Duration;
+use tracing::warn;
+use trogon_std::env::ReadEnv;
+
+const MIN_TIMEOUT_SECS: u64 = 1;
+const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 10;
+
+const ENV_ACP_PREFIX: &str = "ACP_PREFIX";
+const ENV_OPERATION_TIMEOUT_SECS: &str = "ACP_OPERATION_TIMEOUT_SECS";
+const ENV_PROMPT_TIMEOUT_SECS: &str = "ACP_PROMPT_TIMEOUT_SECS";
+const ENV_CONNECT_TIMEOUT_SECS: &str = "ACP_NATS_CONNECT_TIMEOUT_SECS";
+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")]
+    pub acp_prefix: Option<String>,
+}
+
+pub fn from_env_with_provider<E: ReadEnv>(
+    env_provider: &E,
+) -> Result<Config, AcpPrefixError> {
+    let args = Args::parse();
+    from_args(args, env_provider)
+}
+
+pub fn from_args<E: ReadEnv>(
+    args: Args,
+    env_provider: &E,
+) -> Result<Config, AcpPrefixError> {
+    let raw_prefix = args
+        .acp_prefix
+        .or_else(|| env_provider.var(ENV_ACP_PREFIX).ok())
+        .unwrap_or_else(|| DEFAULT_ACP_PREFIX.to_string());
+    let prefix = AcpPrefix::new(raw_prefix)?;
+    let mut config = Config::with_prefix(prefix, NatsConfig::from_env(env_provider));
+
+    if let Ok(raw) = env_provider.var(ENV_OPERATION_TIMEOUT_SECS) {
+        match raw.parse::<u64>() {
+            Ok(secs) if secs >= MIN_TIMEOUT_SECS => {
+                config = config.with_operation_timeout(Duration::from_secs(secs));
+            }
+            Ok(secs) => {
+                warn!(
+                    "{ENV_OPERATION_TIMEOUT_SECS}={secs} is below minimum ({MIN_TIMEOUT_SECS}), using default"
+                );
+            }
+            Err(_) => {
+                warn!("{ENV_OPERATION_TIMEOUT_SECS}={raw:?} is not a valid integer, using default");
+            }
+        }
+    }
+
+    if let Ok(raw) = env_provider.var(ENV_PROMPT_TIMEOUT_SECS) {
+        match raw.parse::<u64>() {
+            Ok(secs) if secs >= MIN_TIMEOUT_SECS => {
+                config = config.with_prompt_timeout(Duration::from_secs(secs));
+            }
+            Ok(secs) => {
+                warn!(
+                    "{ENV_PROMPT_TIMEOUT_SECS}={secs} is below minimum ({MIN_TIMEOUT_SECS}), using default"
+                );
+            }
+            Err(_) => {
+                warn!("{ENV_PROMPT_TIMEOUT_SECS}={raw:?} is not a valid integer, using default");
+            }
+        }
+    }
+
+    Ok(config)
+}
+
+pub fn nats_connect_timeout<E: ReadEnv>(env_provider: &E) -> Duration {
+    let default = Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS);
+
+    match env_provider.var(ENV_CONNECT_TIMEOUT_SECS) {
+        Ok(raw) => match raw.parse::<u64>() {
+            Ok(secs) if secs >= MIN_TIMEOUT_SECS => Duration::from_secs(secs),
+            Ok(secs) => {
+                warn!(
+                    "{ENV_CONNECT_TIMEOUT_SECS}={secs} is below minimum ({MIN_TIMEOUT_SECS}), using default"
+                );
+                default
+            }
+            Err(_) => {
+                warn!("{ENV_CONNECT_TIMEOUT_SECS}={raw:?} is not a valid integer, using default");
+                default
+            }
+        },
+        Err(_) => default,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use trogon_std::env::InMemoryEnv;
+
+    fn config_from_env(env: &InMemoryEnv) -> Config {
+        let args = Args { acp_prefix: None };
+        from_args(args, env).unwrap()
+    }
+
+    #[test]
+    fn test_default_config() {
+        let env = InMemoryEnv::new();
+        let config = config_from_env(&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_acp_prefix_from_env_provider() {
+        let env = InMemoryEnv::new();
+        env.set("ACP_PREFIX", "custom-prefix");
+        let config = config_from_env(&env);
+        assert_eq!(config.acp_prefix(), "custom-prefix");
+    }
+
+    #[test]
+    fn test_acp_prefix_from_args() {
+        let env = InMemoryEnv::new();
+        let args = Args {
+            acp_prefix: Some("cli-prefix".to_string()),
+        };
+        let config = from_args(args, &env).unwrap();
+        assert_eq!(config.acp_prefix(), "cli-prefix");
+    }
+
+    #[test]
+    fn test_args_override_env() {
+        let env = InMemoryEnv::new();
+        env.set("ACP_PREFIX", "env-prefix");
+        let args = Args {
+            acp_prefix: Some("cli-prefix".to_string()),
+        };
+        let config = from_args(args, &env).unwrap();
+        assert_eq!(config.acp_prefix(), "cli-prefix");
+    }
+
+    #[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 = config_from_env(&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"));
+    }
+
+    #[test]
+    fn test_nats_connect_timeout_from_env() {
+        let env = InMemoryEnv::new();
+        assert_eq!(nats_connect_timeout(&env), Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS));
+
+        env.set(ENV_CONNECT_TIMEOUT_SECS, "15");
+        assert_eq!(nats_connect_timeout(&env), Duration::from_secs(15));
+        env.set(ENV_CONNECT_TIMEOUT_SECS, "0");
+        assert_eq!(nats_connect_timeout(&env), Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS));
+        env.set(ENV_CONNECT_TIMEOUT_SECS, "not-a-number");
+        assert_eq!(nats_connect_timeout(&env), Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS));
+    }
+
+    #[test]
+    fn test_operation_timeout_invalid_env_is_ignored() {
+        let env = InMemoryEnv::new();
+        let default = config_from_env(&env);
+        let default_timeout = default.operation_timeout();
+
+        env.set("ACP_OPERATION_TIMEOUT_SECS", "not-a-number");
+        let invalid = config_from_env(&env);
+        assert_eq!(invalid.operation_timeout(), default_timeout);
+    }
+
+    #[test]
+    fn test_operation_timeout_under_min_is_ignored() {
+        let env = InMemoryEnv::new();
+        let default = config_from_env(&env);
+        let default_timeout = default.operation_timeout();
+
+        env.set("ACP_OPERATION_TIMEOUT_SECS", "0");
+        let ignored = config_from_env(&env);
+        assert_eq!(ignored.operation_timeout(), default_timeout);
+    }
+
+    #[test]
+    fn test_prompt_timeout_invalid_env_is_ignored() {
+        let env = InMemoryEnv::new();
+        let default = config_from_env(&env);
+        let default_timeout = default.prompt_timeout();
+
+        env.set("ACP_PROMPT_TIMEOUT_SECS", "not-a-number");
+        let invalid = config_from_env(&env);
+        assert_eq!(invalid.prompt_timeout(), default_timeout);
+    }
+
+    #[test]
+    fn test_prompt_timeout_under_min_is_ignored() {
+        let env = InMemoryEnv::new();
+        let default = config_from_env(&env);
+        let default_timeout = default.prompt_timeout();
+
+        env.set("ACP_PROMPT_TIMEOUT_SECS", "0");
+        let ignored = config_from_env(&env);
+        assert_eq!(ignored.prompt_timeout(), default_timeout);
+    }
+}

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

@@ -0,0 +1,108 @@
+mod config;
+mod signal;
+mod telemetry;
+
+use acp_nats::{agent::Bridge, client, nats, StdJsonSerialize};
+use agent_client_protocol::AgentSideConnection;
+use async_nats::Client as NatsAsyncClient;
+use std::rc::Rc;
+use tracing::{error, info};
+use trogon_std::env::SystemEnv;
+use trogon_std::fs::SystemFs;
+use trogon_std::time::SystemClock;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let config = config::from_env_with_provider(&SystemEnv)?;
+    telemetry::init_logger(config.acp_prefix(), &SystemEnv, &SystemFs);
+
+    info!("ACP bridge starting");
+
+    let nats_connect_timeout = config::nats_connect_timeout(&SystemEnv);
+    let nats_client = nats::connect(config.nats(), nats_connect_timeout).await?;
+
+    let local = tokio::task::LocalSet::new();
+
+    let result = local.run_until(run_bridge(nats_client, &config)).await;
+
+    telemetry::shutdown_otel();
+
+    if let Err(ref e) = result {
+        error!(error = %e, "ACP bridge stopped with error");
+    } else {
+        info!("ACP bridge stopped");
+    }
+
+    result
+}
+
+// `Rc` is intentional throughout this function: the ACP `Agent` trait is
+// `?Send`, so the entire bridge runs on a `LocalSet` with `spawn_local`.
+// Do not replace with `Arc` or move tasks to `tokio::spawn` — that would
+// violate the `!Send` constraint.
+async fn run_bridge(
+    nats_client: NatsAsyncClient,
+    config: &acp_nats::Config,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let stdin = async_compat::Compat::new(tokio::io::stdin());
+    let stdout = async_compat::Compat::new(tokio::io::stdout());
+
+    let meter = opentelemetry::global::meter("acp-io-bridge-nats");
+    let bridge = Rc::new(Bridge::new(
+        nats_client.clone(),
+        SystemClock,
+        &meter,
+        config.clone(),
+    ));
+
+    let (connection, io_task) = AgentSideConnection::new(bridge.clone(), stdout, stdin, |fut| {
+        tokio::task::spawn_local(fut);
+    });
+
+    let connection = Rc::new(connection);
+
+    let client_connection = connection.clone();
+    let bridge_for_client = bridge.clone();
+    let mut client_task = tokio::task::spawn_local(async move {
+        client::run(nats_client, client_connection, bridge_for_client, StdJsonSerialize).await;
+    });
+    info!("ACP bridge running on stdio with NATS client proxy");
+
+    let shutdown_result = tokio::select! {
+        result = &mut client_task => {
+            match result {
+                Ok(()) => {
+                    info!("ACP bridge client task completed");
+                    Ok(())
+                }
+                Err(e) => {
+                    error!(error = %e, "Client task ended with error");
+                    Err(Box::new(e) as Box<dyn std::error::Error>)
+                }
+            }
+        }
+        result = io_task => {
+            match result {
+                Err(e) => {
+                    error!(error = %e, "IO task error");
+                    Err(Box::new(e) as Box<dyn std::error::Error>)
+                }
+                Ok(()) => {
+                    info!("ACP bridge shutting down (IO closed)");
+                    Ok(())
+                }
+            }
+        }
+        _ = signal::shutdown_signal() => {
+            info!("ACP bridge shutting down (signal received)");
+            Ok(())
+        }
+    };
+
+    if !client_task.is_finished() {
+        client_task.abort();
+    }
+    let _ = client_task.await;
+
+    shutdown_result
+}

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

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