fix(sacp): ensure NewSessionRequest flows through handler chain

McpServer was incorrectly forwarding NewSessionRequest directly to the
agent instead of returning Handled::No, which bypassed any downstream
.on_receive_request_from() handlers registered via the proxy builder.

This broke proactive embodiment in SparkleComponent, where the
NewSessionRequest handler intercepts sessions to inject embodiment
prompts before user prompts arrive.

Added regression test to prevent this from happening again.

Author: nikomatsakis

src/sacp-conductor/tests/mcp_server_handler_chain.rs

@@ -0,0 +1,215 @@
+//! Test that MCP server doesn't break the handler chain for NewSessionRequest.
+//!
+//! This is a regression test for a bug where `McpServer::handle_message` would
+//! forward `NewSessionRequest` directly to the agent instead of returning
+//! `Handled::No`, which prevented downstream `.on_receive_request_from()` handlers
+//! from being invoked.
+
+use sacp::mcp_server::McpServer;
+use sacp::schema::{
+    AgentCapabilities, InitializeRequest, InitializeResponse, NewSessionRequest,
+    NewSessionResponse, SessionId,
+};
+use sacp::{Agent, AgentToClient, Client, Component, ProxyToConductor};
+use sacp_conductor::Conductor;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use tokio::io::duplex;
+use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+
+/// Simple echo tool parameters
+#[derive(Debug, Serialize, Deserialize, JsonSchema)]
+struct EchoParams {
+    message: String,
+}
+
+/// Simple echo tool output
+#[derive(Debug, Serialize, Deserialize, JsonSchema)]
+struct EchoOutput {
+    result: String,
+}
+
+/// Test helper to receive a JSON-RPC response
+async fn recv<T: sacp::JrResponsePayload + Send>(
+    response: sacp::JrResponse<T>,
+) -> Result<T, sacp::Error> {
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    response.await_when_result_received(async move |result| {
+        tx.send(result).map_err(|_| sacp::Error::internal_error())
+    })?;
+    rx.await.map_err(|_| sacp::Error::internal_error())?
+}
+
+/// Tracks whether the NewSessionRequest handler was invoked
+struct HandlerConfig {
+    new_session_handler_called: AtomicBool,
+}
+
+impl HandlerConfig {
+    fn new() -> Arc<Self> {
+        Arc::new(Self {
+            new_session_handler_called: AtomicBool::new(false),
+        })
+    }
+
+    fn was_handler_called(&self) -> bool {
+        self.new_session_handler_called.load(Ordering::SeqCst)
+    }
+}
+
+/// A proxy component that has BOTH an MCP server AND a NewSessionRequest handler.
+/// The bug was that when both were present, the NewSessionRequest handler was never called.
+struct ProxyWithMcpAndHandler {
+    config: Arc<HandlerConfig>,
+}
+
+impl Component for ProxyWithMcpAndHandler {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
+        let config = Arc::clone(&self.config);
+
+        // Create an MCP server with a simple tool
+        let mcp_server = McpServer::builder("test-server".to_string())
+            .instructions("A test MCP server")
+            .tool_fn(
+                "echo",
+                "Echoes back the input",
+                async |params: EchoParams, _cx| {
+                    Ok(EchoOutput {
+                        result: format!("Echo: {}", params.message),
+                    })
+                },
+                |f, args, cx| Box::pin(f(args, cx)),
+            )
+            .build();
+
+        ProxyToConductor::builder()
+            .name("proxy-with-mcp-and-handler")
+            // Add the MCP server
+            .with_mcp_server(mcp_server)
+            // Add a NewSessionRequest handler - this should be invoked!
+            .on_receive_request_from(
+                Client,
+                async move |request: NewSessionRequest, request_cx, cx| {
+                    // Mark that we were called
+                    config
+                        .new_session_handler_called
+                        .store(true, Ordering::SeqCst);
+
+                    // Forward to agent and relay response
+                    cx.send_request_to(Agent, request)
+                        .await_when_result_received(async move |result| {
+                            let response: NewSessionResponse = result?;
+                            request_cx.respond(response)
+                        })
+                },
+            )
+            .serve(client)
+            .await
+    }
+}
+
+/// A simple agent that responds to initialization and session requests
+struct SimpleAgent;
+
+impl Component for SimpleAgent {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
+        AgentToClient::builder()
+            .name("simple-agent")
+            .on_receive_request(async |request: InitializeRequest, request_cx, _cx| {
+                request_cx.respond(InitializeResponse {
+                    protocol_version: request.protocol_version,
+                    agent_capabilities: AgentCapabilities::default(),
+                    auth_methods: vec![],
+                    meta: None,
+                    agent_info: None,
+                })
+            })
+            .on_receive_request(async |_request: NewSessionRequest, request_cx, _cx| {
+                request_cx.respond(NewSessionResponse {
+                    session_id: SessionId(Arc::from(uuid::Uuid::new_v4().to_string())),
+                    modes: None,
+                    meta: None,
+                })
+            })
+            .connect_to(client)?
+            .serve()
+            .await
+    }
+}
+
+async fn run_test(
+    components: Vec<sacp::DynComponent>,
+    editor_task: impl AsyncFnOnce(sacp::JrConnectionCx<sacp::ClientToAgent>) -> Result<(), sacp::Error>,
+) -> Result<(), sacp::Error> {
+    let (editor_out, conductor_in) = duplex(1024);
+    let (conductor_out, editor_in) = duplex(1024);
+
+    let transport = sacp::ByteStreams::new(editor_out.compat_write(), editor_in.compat());
+
+    sacp::ClientToAgent::builder()
+        .name("editor-to-conductor")
+        .with_spawned(|_cx| async move {
+            Conductor::new("conductor".to_string(), components, Default::default())
+                .run(sacp::ByteStreams::new(
+                    conductor_out.compat_write(),
+                    conductor_in.compat(),
+                ))
+                .await
+        })
+        .with_client(transport, editor_task)
+        .await
+}
+
+/// Regression test: NewSessionRequest handler should be invoked even when MCP server is present
+#[tokio::test]
+async fn test_new_session_handler_invoked_with_mcp_server() -> Result<(), sacp::Error> {
+    let handler_config = HandlerConfig::new();
+    let handler_config_clone = Arc::clone(&handler_config);
+
+    let proxy = sacp::DynComponent::new(ProxyWithMcpAndHandler {
+        config: handler_config,
+    });
+    let agent = sacp::DynComponent::new(SimpleAgent);
+
+    run_test(vec![proxy, agent], async |editor_cx| {
+        // Initialize first
+        let _init_response = recv(editor_cx.send_request(InitializeRequest {
+            protocol_version: Default::default(),
+            client_capabilities: Default::default(),
+            meta: None,
+            client_info: None,
+        }))
+        .await?;
+
+        // Create a new session - this should trigger the handler in the proxy
+        let session_response = recv(editor_cx.send_request(NewSessionRequest {
+            cwd: PathBuf::from("/tmp"),
+            mcp_servers: vec![],
+            meta: None,
+        }))
+        .await?;
+
+        // Verify we got a valid session ID
+        assert!(
+            !session_response.session_id.0.is_empty(),
+            "Should receive a valid session ID"
+        );
+
+        Ok::<(), sacp::Error>(())
+    })
+    .await?;
+
+    // THE KEY ASSERTION: verify the handler was actually called
+    assert!(
+        handler_config_clone.was_handler_called(),
+        "NewSessionRequest handler should be invoked even when MCP server is in the chain. \
+         This is a regression - the MCP server was incorrectly forwarding the request directly \
+         to the agent instead of letting it flow through the handler chain."
+    );
+
+    Ok(())
+}

src/sacp/src/mcp_server/server.rs

@@ -6,7 +6,7 @@ use agent_client_protocol_schema::NewSessionRequest;
 use uuid::Uuid;
 
 use crate::{
-    Agent, Client, HasEndpoint, JrConnectionCx, JrMessageHandler, JrRole,
+    Agent, Client, Handled, HasEndpoint, JrConnectionCx, JrMessageHandler, JrRole,
     jsonrpc::DynamicHandlerRegistration,
     mcp_server::{McpServerConnect, active_session::McpActiveSession, builder::McpServerBuilder},
     util::MatchMessageFrom,
@@ -103,8 +103,10 @@ where
                 async |mut request: NewSessionRequest, request_cx| {
                     self.add_to_new_session(&mut request, &cx)?
                         .run_indefinitely();
-                    cx.send_request_to(Agent, request)
-                        .forward_to_request_cx(request_cx)
+                    Ok(Handled::No {
+                        message: (request, request_cx),
+                        retry: false,
+                    })
                 },
             )
             .await