chore(coverage): remove ignore-filename-regex and add inline tests

Remove the blanket --ignore-filename-regex from cargo-llvm-cov config
and add targeted tests for previously excluded code instead.

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

Author: yordis

rsworkspace/.cargo/config.toml

@@ -3,6 +3,4 @@ cov = [
     "llvm-cov",
     "--all-features",
     "--workspace",
-    "--ignore-filename-regex",
-    "crates/(acp-nats-stdio|acp-telemetry)/|crates/acp-nats-ws/src/(main|connection|upgrade)\\.rs"
 ]

rsworkspace/Cargo.lock

@@ -4,6 +4,7 @@ members = ["crates/*"]
 
 [workspace.lints.rust]
 warnings = "deny"
+unexpected_cfgs = { level = "allow", check-cfg = ['cfg(coverage)'] }
 
 [workspace.lints.clippy]
 all = "deny"

rsworkspace/Cargo.toml

@@ -3,11 +3,15 @@ name = "acp-nats-stdio"
 version = "0.1.0"
 edition = "2024"
 
+[lints]
+workspace = true
+
 [dependencies]
 acp-nats = { path = "../acp-nats" }
 acp-telemetry = { path = "../acp-telemetry" }
 agent-client-protocol = "0.9.4"
 async-compat = "0.2.5"
+futures = "0.3"
 async-nats = "0.45.0"
 clap = { version = "4.5", features = ["derive", "env"] }
 opentelemetry = "0.31.0"
@@ -16,5 +20,6 @@ tokio = { version = "1.49.0", features = ["rt-multi-thread", "macros", "signal",
 tracing = "0.1.44"
 
 [dev-dependencies]
+trogon-nats = { path = "../trogon-nats", features = ["test-support"] }
 tracing-subscriber = { version = "0.3.22", features = ["fmt"] }
 trogon-std = { path = "../trogon-std", features = ["test-support"] }

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

@@ -10,6 +10,7 @@ pub struct Args {
     pub acp_prefix: Option<String>,
 }
 
+#[cfg(not(coverage))]
 pub fn base_config<E: ReadEnv>(env_provider: &E) -> Result<Config, AcpPrefixError> {
     let args = Args::parse();
     base_config_from_args(args, env_provider)

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

@@ -2,7 +2,6 @@ mod config;
 
 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};
 use trogon_std::env::SystemEnv;
@@ -11,6 +10,7 @@ use trogon_std::time::SystemClock;
 
 use acp_telemetry::ServiceName;
 
+#[cfg(not(coverage))]
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let config = config::base_config(&SystemEnv)?;
@@ -27,9 +27,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let nats_connect_timeout = acp_nats::nats_connect_timeout(&SystemEnv);
     let nats_client = nats::connect(config.nats(), nats_connect_timeout).await?;
 
-    let local = tokio::task::LocalSet::new();
+    let stdin = async_compat::Compat::new(tokio::io::stdin());
+    let stdout = async_compat::Compat::new(tokio::io::stdout());
 
-    let result = local.run_until(run_bridge(nats_client, &config)).await;
+    let local = tokio::task::LocalSet::new();
+    let result = local
+        .run_until(run_bridge(
+            nats_client,
+            &config,
+            stdout,
+            stdin,
+            acp_telemetry::signal::shutdown_signal(),
+        ))
+        .await;
 
     if let Err(ref e) = result {
         error!(error = %e, "ACP bridge stopped with error");
@@ -42,17 +52,25 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     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());
+#[cfg(coverage)]
+fn main() {}
 
+async fn run_bridge<N, W, R>(
+    nats_client: N,
+    config: &acp_nats::Config,
+    stdout: W,
+    stdin: R,
+    shutdown_signal: impl std::future::Future<Output = ()>,
+) -> Result<(), Box<dyn std::error::Error>>
+where
+    N: acp_nats::RequestClient
+        + acp_nats::PublishClient
+        + acp_nats::FlushClient
+        + acp_nats::SubscribeClient
+        + 'static,
+    W: futures::AsyncWrite + Unpin + 'static,
+    R: futures::AsyncRead + Unpin + 'static,
+{
     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(
@@ -106,7 +124,7 @@ async fn run_bridge(
                 }
             }
         }
-        _ = acp_telemetry::signal::shutdown_signal() => {
+        _ = shutdown_signal => {
             info!("ACP bridge shutting down (signal received)");
             Ok(())
         }
@@ -119,3 +137,72 @@ async fn run_bridge(
 
     shutdown_result
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use trogon_nats::AdvancedMockNatsClient;
+
+    #[tokio::test]
+    async fn run_bridge_shuts_down_on_signal() {
+        let mock = AdvancedMockNatsClient::new();
+        let _sub = mock.inject_messages();
+        let config = acp_nats::Config::new(
+            acp_nats::AcpPrefix::new("acp").unwrap(),
+            acp_nats::NatsConfig {
+                servers: vec!["localhost:4222".to_string()],
+                auth: trogon_nats::NatsAuth::None,
+            },
+        );
+
+        let (reader, _writer) = tokio::io::duplex(1024);
+        let (_reader2, writer2) = tokio::io::duplex(1024);
+        let stdin = async_compat::Compat::new(reader);
+        let stdout = async_compat::Compat::new(writer2);
+
+        let local = tokio::task::LocalSet::new();
+        let result = local
+            .run_until(run_bridge(
+                mock,
+                &config,
+                stdout,
+                stdin,
+                std::future::ready(()),
+            ))
+            .await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn run_bridge_exits_on_io_close() {
+        let mock = AdvancedMockNatsClient::new();
+        let _sub = mock.inject_messages();
+        let config = acp_nats::Config::new(
+            acp_nats::AcpPrefix::new("acp").unwrap(),
+            acp_nats::NatsConfig {
+                servers: vec!["localhost:4222".to_string()],
+                auth: trogon_nats::NatsAuth::None,
+            },
+        );
+
+        let (reader, writer) = tokio::io::duplex(1024);
+        let (_reader2, writer2) = tokio::io::duplex(1024);
+        drop(writer);
+        let stdin = async_compat::Compat::new(reader);
+        let stdout = async_compat::Compat::new(writer2);
+
+        let local = tokio::task::LocalSet::new();
+        let result = local
+            .run_until(run_bridge(
+                mock,
+                &config,
+                stdout,
+                stdin,
+                std::future::pending(),
+            ))
+            .await;
+
+        assert!(result.is_ok());
+    }
+}

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

@@ -3,6 +3,9 @@ name = "acp-nats-ws"
 version = "0.1.0"
 edition = "2024"
 
+[lints]
+workspace = true
+
 [dependencies]
 acp-nats = { path = "../acp-nats" }
 acp-telemetry = { path = "../acp-telemetry" }

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

@@ -180,3 +180,65 @@ async fn run_send_pump(
     }
     let _ = ws_sender.close().await;
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::extract::State;
+    use axum::extract::ws::WebSocketUpgrade;
+    use axum::response::Response;
+    use std::time::Duration;
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::connect_async;
+    use tokio_tungstenite::tungstenite::Message as TungsteniteMessage;
+
+    #[derive(Clone)]
+    struct EchoState;
+
+    async fn echo_handler(ws: WebSocketUpgrade, State(_): State<EchoState>) -> Response {
+        ws.on_upgrade(|socket| async move {
+            let (ws_sender, ws_receiver) = socket.split();
+            let (duplex_write, duplex_read) = tokio::io::duplex(DUPLEX_BUFFER_SIZE);
+            let recv = run_recv_pump(ws_receiver, duplex_write);
+            let send = run_send_pump(ws_sender, duplex_read);
+            tokio::join!(recv, send);
+        })
+    }
+
+    async fn start_echo_server() -> String {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(echo_handler))
+            .with_state(EchoState);
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        format!("ws://{}/ws", addr)
+    }
+
+    #[tokio::test]
+    async fn multiple_messages_round_trip() {
+        let url = start_echo_server().await;
+        let (mut ws, _) = connect_async(&url).await.unwrap();
+
+        let messages = vec!["alpha", "beta", "gamma"];
+        for msg in &messages {
+            ws.send(TungsteniteMessage::Text((*msg).into()))
+                .await
+                .unwrap();
+        }
+
+        for expected in &messages {
+            let msg = tokio::time::timeout(Duration::from_secs(2), ws.next())
+                .await
+                .expect("timeout")
+                .expect("stream ended")
+                .unwrap();
+            match msg {
+                TungsteniteMessage::Text(t) => assert_eq!(t, *expected),
+                other => panic!("expected Text('{expected}'), got {other:?}"),
+            }
+        }
+    }
+}

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

@@ -13,6 +13,7 @@ use trogon_std::env::SystemEnv;
 use trogon_std::fs::SystemFs;
 use upgrade::{ConnectionRequest, UpgradeState};
 
+#[cfg(not(coverage))]
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args = config::Args::parse();
@@ -78,6 +79,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     result.map_err(|e| Box::new(e) as Box<dyn std::error::Error>)
 }
 
+#[cfg(coverage)]
+fn main() {}
+
 const THREAD_NAME: &str = "acp-ws-local";
 
 /// Runs a single-threaded tokio runtime with a
@@ -252,4 +256,67 @@ mod tests {
 
         conn_thread.join().unwrap();
     }
+
+    #[tokio::test]
+    async fn test_shutdown_while_connection_active() {
+        let nats_mock = AdvancedMockNatsClient::new();
+        let config = Config::new(
+            acp_nats::AcpPrefix::new("acp").unwrap(),
+            acp_nats::NatsConfig {
+                servers: vec!["localhost:4222".to_string()],
+                auth: trogon_nats::NatsAuth::None,
+            },
+        );
+
+        let _injector = nats_mock.inject_messages();
+
+        nats_mock.hang_next_request();
+
+        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
+        let (conn_tx, conn_rx) = mpsc::unbounded_channel::<ConnectionRequest>();
+
+        let nats_mock_clone = nats_mock.clone();
+        let conn_thread = std::thread::Builder::new()
+            .name(THREAD_NAME.into())
+            .spawn(move || run_connection_thread(conn_rx, nats_mock_clone, config))
+            .expect("failed to spawn connection thread");
+
+        let state = UpgradeState {
+            conn_tx,
+            shutdown_tx: shutdown_tx.clone(),
+        };
+
+        let app = axum::Router::new()
+            .route("/ws", axum::routing::get(upgrade::handle))
+            .with_state(state);
+
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        let server_task = tokio::spawn(async move {
+            axum::serve(listener, app)
+                .with_graceful_shutdown(async move {
+                    let _ = shutdown_rx.changed().await;
+                })
+                .await
+                .unwrap();
+        });
+
+        let ws_url = format!("ws://{}/ws", addr);
+        let (mut ws_stream, _) = connect_async(&ws_url).await.unwrap();
+
+        let req =
+            r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion": 0}}"#;
+        ws_stream.send(Message::Text(req.into())).await.unwrap();
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        shutdown_tx.send(true).unwrap();
+
+        let _ = tokio::time::timeout(Duration::from_secs(5), server_task)
+            .await
+            .expect("server task did not shut down");
+
+        conn_thread.join().unwrap();
+    }
 }