feat(client): local-key enterprise desktop and agent POC

crates/buzz-acp/Cargo.toml

@@ -37,6 +37,7 @@ futures-util = { workspace = true }
 
 # HTTP (channel discovery REST API)
 reqwest = { workspace = true }
+buzz-ws-client = { workspace = true }
 
 # Serialization
 serde = { workspace = true }

crates/buzz-acp/src/acp.rs

@@ -10,7 +10,9 @@
 
 use futures_util::StreamExt;
 use tokio::io::AsyncWriteExt;
-use tokio::process::{Child, ChildStdin, ChildStdout};
+#[cfg(not(test))]
+use tokio::process::ChildStdout;
+use tokio::process::{Child, ChildStdin};
 use tokio_util::codec::{FramedRead, LinesCodec, LinesCodecError};
 
 use crate::observer::{ObserverContext, ObserverHandle};
@@ -38,12 +40,21 @@ pub struct McpServer {
 }
 
 /// A single environment variable for an MCP server.
-#[derive(Debug, Clone, serde::Serialize)]
+#[derive(Clone, serde::Serialize)]
 pub struct EnvVar {
     pub name: String,
     pub value: String,
 }
 
+impl std::fmt::Debug for EnvVar {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("EnvVar")
+            .field("name", &self.name)
+            .field("value", &"[REDACTED]")
+            .finish()
+    }
+}
+
 /// Stop reason returned by `session/prompt` when the agent finishes a turn.
 ///
 /// Maps to the `stopReason` field in the `SessionPromptResponse`.
@@ -149,7 +160,10 @@ pub struct AcpClient {
     /// Framed reader over the agent's stdout pipe (line-oriented, bounded).
     /// Uses `LinesCodec::new_with_max_length` to enforce MAX_LINE_SIZE at the
     /// read level — prevents OOM from rogue agents writing infinite non-newline bytes.
+    #[cfg(not(test))]
     reader: FramedRead<ChildStdout, LinesCodec>,
+    #[cfg(test)]
+    reader: FramedRead<Box<dyn tokio::io::AsyncRead + Send + Unpin>, LinesCodec>,
     /// Monotonically increasing JSON-RPC request id counter.
     /// Harness-generated IDs are always numeric.
     next_id: u64,
@@ -560,7 +574,13 @@ impl AcpClient {
         Ok(Self {
             child,
             stdin,
+            #[cfg(not(test))]
             reader: FramedRead::new(stdout, LinesCodec::new_with_max_length(MAX_LINE_SIZE)),
+            #[cfg(test)]
+            reader: FramedRead::new(
+                Box::new(stdout),
+                LinesCodec::new_with_max_length(MAX_LINE_SIZE),
+            ),
             next_id: 0,
             pending_permission_id: None,
             permission_responded: false,
@@ -1125,7 +1145,9 @@ impl AcpClient {
             "params": params,
         });
 
-        tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default());
+        // session/new carries MCP credentials (including enterprise capabilities).
+        // Log only method/id; never serialize credential-bearing parameters.
+        tracing::debug!(target: "acp::wire", method, id, "→ request");
 
         // Wrap write + read in a single timeout so a hung agent can't block forever.
         // We cannot use an async block that borrows `self` mutably across two awaits
@@ -3206,31 +3228,54 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn idle_resets_on_stdout_activity() {
-        // Send valid JSON (session/update notifications) to reset the idle timer.
-        // Non-JSON lines no longer reset idle — only valid JSON notifications do.
-        let mut client = spawn_script(
-            r#"for i in $(seq 1 10); do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"agent_thought_chunk","content":{"text":"thinking"}}}}'; sleep 0.05; done; sleep 10"#,
-        )
-        .await;
+    // Drives the actual bounded reader/deadline implementation with in-memory
+    // bytes and virtual time. No dependency on OS scheduling a shell every 50ms.
+    async fn assert_activity_resets_idle(update: serde_json::Value, count: u32, idle_ms: u64) {
+        let mut client = spawn_script("read _done").await;
+        let (mut writer, reader) = tokio::io::duplex(4096);
+        client.reader = FramedRead::new(
+            Box::new(reader),
+            LinesCodec::new_with_max_length(MAX_LINE_SIZE),
+        );
+        tokio::time::pause();
         let max_dur = std::time::Duration::from_secs(10);
-        let hard_deadline = tokio::time::Instant::now() + max_dur;
-        let start = std::time::Instant::now();
-        let result = client
-            .read_until_response_with_idle_timeout(
-                "test",
-                999,
-                std::time::Duration::from_millis(200),
-                hard_deadline,
-                max_dur,
-            )
-            .await;
-        let elapsed = start.elapsed();
-        // 10 messages × 50ms = ~500ms of activity, then idle timeout fires after 200ms more
-        assert!(elapsed >= std::time::Duration::from_millis(400));
-        assert!(elapsed < std::time::Duration::from_secs(3));
+        let start = tokio::time::Instant::now();
+        let read = client.read_until_response_with_idle_timeout(
+            "test",
+            999,
+            std::time::Duration::from_millis(idle_ms),
+            start + max_dur,
+            max_dur,
+        );
+        let send = async {
+            for _ in 0..count {
+                let message = serde_json::json!({"jsonrpc":"2.0", "method":"session/update", "params":{"update":update}});
+                writer
+                    .write_all(format!("{message}\n").as_bytes())
+                    .await
+                    .unwrap();
+                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            }
+            // Keep the pipe open: the production result must be idle, not EOF.
+            std::future::pending::<()>().await;
+        };
+        tokio::pin!(send);
+        let result = tokio::select! { result = read => result, _ = &mut send => unreachable!() };
         assert!(matches!(result, Err(AcpError::IdleTimeout(_))));
+        let expected = std::time::Duration::from_millis(u64::from(count - 1) * 50 + idle_ms);
+        // Tokio's millisecond timer wheel rounds each scheduled wake. Keep the
+        // exact lower bound, allowing at most two ticks per scheduled timer.
+        assert!(start.elapsed() >= expected);
+        assert!(
+            start.elapsed()
+                <= expected + std::time::Duration::from_millis(u64::from(count + 1) * 2)
+        );
+        tokio::time::resume();
+    }
+
+    #[tokio::test]
+    async fn idle_resets_on_stdout_activity() {
+        assert_activity_resets_idle(serde_json::json!({"sessionUpdate":"agent_thought_chunk", "content":{"text":"thinking"}}), 10, 200).await;
     }
 
     #[tokio::test]
@@ -3405,33 +3450,8 @@ mod tests {
 
     #[tokio::test]
     async fn keepalive_resets_idle_past_deadline() {
-        // Keepalive session/update lines every 50ms against a 100ms idle deadline.
-        // The turn should survive well past the 100ms deadline (proves the fix).
-        let mut client = spawn_script(
-            r#"for i in $(seq 1 20); do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"keepalive"}}}'; sleep 0.05; done; sleep 10"#,
-        )
-        .await;
-        let max_dur = std::time::Duration::from_secs(10);
-        let hard_deadline = tokio::time::Instant::now() + max_dur;
-        let start = std::time::Instant::now();
-        let result = client
-            .read_until_response_with_idle_timeout(
-                "test",
-                999,
-                std::time::Duration::from_millis(100),
-                hard_deadline,
-                max_dur,
-            )
+        assert_activity_resets_idle(serde_json::json!({"sessionUpdate":"keepalive"}), 20, 100)
             .await;
-        let elapsed = start.elapsed();
-        // 20 keepalives × 50ms = ~1000ms of activity, then idle fires after 100ms more.
-        // Must survive well past the 100ms deadline.
-        assert!(
-            elapsed >= std::time::Duration::from_millis(500),
-            "keepalive should reset idle past the deadline; elapsed only {elapsed:?}"
-        );
-        assert!(elapsed < std::time::Duration::from_secs(5));
-        assert!(matches!(result, Err(AcpError::IdleTimeout(_))));
     }
 
     #[tokio::test]

crates/buzz-acp/src/lib.rs

@@ -5898,6 +5898,14 @@ fn build_mcp_servers(config: &Config) -> Vec<McpServer> {
                         .expect("secret key bech32 encoding should never fail"),
                 },
             ];
+            for name in buzz_ws_client::identity_adapter::ENV_KEYS {
+                if let Ok(value) = std::env::var(name) {
+                    env.push(EnvVar {
+                        name: name.into(),
+                        value,
+                    });
+                }
+            }
             // Forward BUZZ_AUTH_TAG (NIP-OA owner attestation credential)
             // so the MCP server can attach it to every signed event.
             if let Ok(auth_tag) = std::env::var("BUZZ_AUTH_TAG") {

crates/buzz-acp/src/relay.rs

@@ -125,7 +125,9 @@ use nostr::{Event, EventBuilder, Keys, Kind, RelayUrl, Tag};
 use serde_json::{json, Value};
 use tokio::sync::mpsc;
 use tokio::time::timeout;
-use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
+#[cfg(test)]
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream};
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
@@ -406,7 +408,7 @@ impl RestClient {
     ) -> Result<reqwest::Response, RelayError>
     where
         F: Fn() -> Fut,
-        Fut: std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
+        Fut: std::future::Future<Output = Result<reqwest::Response, RelayError>>,
     {
         let mut last_err = None;
 
@@ -439,11 +441,11 @@ impl RestClient {
                         resp.status()
                     )));
                 }
-                Err(e) if e.is_timeout() || e.is_connect() => {
+                Err(RelayError::HttpTransport(e)) if e.is_timeout() || e.is_connect() => {
                     tracing::warn!("{method} {path} network error: {e}");
                     last_err = Some(RelayError::Http(e.to_string()));
                 }
-                Err(e) => return Err(RelayError::Http(e.to_string())),
+                Err(e) => return Err(e),
             }
         }
 
@@ -460,12 +462,11 @@ impl RestClient {
         let url = format!("{}{}", self.base_url, path);
         let body_owned = body_bytes.to_vec();
         let auth_tag_header = self.auth_tag_json.clone();
-        self.request_with_retry("POST", path, || {
-            // NIP-98 is re-signed each attempt (fresh created_at).
-            // sign_nip98 is infallible in practice (key is always valid).
-            let auth = self
-                .nip98_header("POST", &url, Some(&body_owned))
-                .unwrap_or_default();
+        self.request_with_retry("POST", path, || async {
+            let identity = buzz_ws_client::identity_adapter::environment_header(&url, &self.keys)
+                .await
+                .map_err(|e| RelayError::Http(e.to_string()))?;
+            let auth = self.nip98_header("POST", &url, Some(&body_owned))?;
             let mut req = self
                 .http
                 .post(&url)
@@ -474,7 +475,13 @@ impl RestClient {
             if let Some(ref tag) = auth_tag_header {
                 req = req.header("x-auth-tag", tag);
             }
-            req.body(body_owned.clone()).send()
+            if let Some(header) = identity {
+                req = req.header(buzz_ws_client::federated_identity::IDENTITY_HEADER, header);
+            }
+            req.body(body_owned.clone())
+                .send()
+                .await
+                .map_err(RelayError::HttpTransport)
         })
         .await
     }
@@ -597,6 +604,9 @@ pub struct BuzzEvent {
 /// Errors from relay operations.
 #[derive(Debug, thiserror::Error)]
 pub enum RelayError {
+    /// HTTP transport failure, retaining retry classification.
+    #[error("HTTP transport error: {0}")]
+    HttpTransport(reqwest::Error),
     #[error("WebSocket error: {0}")]
     WebSocket(Box<tokio_tungstenite::tungstenite::Error>),
 
@@ -684,7 +694,8 @@ enum RelayCommand {
     SetStartupWatermark { ts: u64 },
 }
 
-type WsStream = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
+type WsStream =
+    buzz_ws_client::identity_socket::IdentitySocket<MaybeTlsStream<tokio::net::TcpStream>>;
 
 /// Harness-side relay client.
 ///
@@ -799,6 +810,7 @@ impl HarnessRelay {
             observer_control_rx: Some(observer_control_rx),
             cmd_tx,
             http: reqwest::Client::builder()
+                .redirect(reqwest::redirect::Policy::none())
                 .timeout(std::time::Duration::from_secs(10))
                 .connect_timeout(std::time::Duration::from_secs(5))
                 .build()
@@ -3838,6 +3850,7 @@ pub(crate) fn parse_relay_message(text: &str) -> Result<RelayMessage, RelayError
 /// - `NoAuthChallenge`, `ConnectionClosed`, `Timeout` — timing/link noise.
 fn is_terminal_connect_error(err: &RelayError) -> bool {
     match err {
+        RelayError::HttpTransport(_) => false,
         RelayError::Http(_) | RelayError::Json(_) | RelayError::UnexpectedMessage(_) => true,
         RelayError::WebSocket(e) => is_terminal_ws_error(e.as_ref()),
         RelayError::AuthFailed(message) => is_terminal_auth_failure(message),
@@ -4013,13 +4026,32 @@ async fn do_connect(
         .parse::<url::Url>()
         .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?;
 
-    let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str()))
+    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+    let mut request = parsed
+        .as_str()
+        .into_client_request()
+        .map_err(|e| RelayError::WebSocket(Box::new(e)))?;
+    let identity = buzz_ws_client::identity_adapter::environment_admission(relay_url, keys)
+        .await
+        .map_err(|e| RelayError::Http(e.to_string()))?;
+    if let Some((header, _)) = identity.clone() {
+        request
+            .headers_mut()
+            .insert(buzz_ws_client::federated_identity::IDENTITY_HEADER, header);
+    }
+    let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(request))
         .await
         .map_err(|_| RelayError::ConnectionClosed)? // timeout → treat as connection failure
         .map_err(|e| RelayError::WebSocket(Box::new(e)))?;
     debug!("connected to relay at {relay_url}");
 
-    let mut ws = ws;
+    let mut ws = buzz_ws_client::identity_socket::IdentitySocket::admitted(
+        ws,
+        identity.as_ref().map(|(_, deadline)| *deadline),
+        relay_url.into(),
+        keys.clone(),
+    )
+    .map_err(|e| RelayError::Http(e.to_string()))?;
     let mut buffer: VecDeque<RelayMessage> = VecDeque::new();
 
     let challenge = wait_for_auth_challenge(&mut ws, &mut buffer, AUTH_TIMEOUT).await?;
@@ -4719,7 +4751,10 @@ mod tests {
         let (client, _) = connect_async(format!("ws://{address}"))
             .await
             .expect("connect test websocket");
-        (client, server.await.expect("join test websocket server"))
+        (
+            client.into(),
+            server.await.expect("join test websocket server"),
+        )
     }
 
     pub(super) async fn next_test_frame(