docs: fix doctests for API refactoring

- Export JrConnection from lib.rs public API
- Rename ViaBytes to ByteStreams in all doctests
- Update JrConnection::new() to JrHandlerChain::new()
- Fix .with_client() examples to use .connect_to() first
- Add comprehensive docs and examples for spawn_connection()
- Update forward_to_request_cx doctest to show proper proxy pattern
- Add missing documentation for JrConnection struct
- Fix describe_chain() implementation in conductor (was todo!())

All workspace doctests now pass.

Co-authored-by: Claude <claude@anthropic.com>

Author: nikomatsakis

src/sacp-conductor/src/component.rs

@@ -1,11 +1,15 @@
 use sacp;
 use sacp_tokio::AcpAgent;
 use std::pin::Pin;
-use tokio_util::compat::{FuturesAsyncReadCompatExt as _, FuturesAsyncWriteCompatExt};
+use tokio_util::compat::{
+    FuturesAsyncReadCompatExt as _, FuturesAsyncWriteCompatExt, TokioAsyncReadCompatExt as _,
+    TokioAsyncWriteCompatExt as _,
+};
 
+use futures::channel::mpsc;
 use futures::{AsyncRead, AsyncWrite};
 
-use sacp::JrConnectionCx;
+use sacp::{IntoJrTransport, JrConnectionCx};
 use tokio::process::Child;
 use tracing::debug;
 
@@ -121,6 +125,35 @@ impl ComponentProvider for CommandComponentProvider {
     }
 }
 
+impl IntoJrTransport for Box<dyn ComponentProvider> {
+    fn into_jr_transport(
+        self: Box<Self>,
+        cx: &JrConnectionCx,
+        outgoing_rx: mpsc::UnboundedReceiver<jsonrpcmsg::Message>,
+        incoming_tx: mpsc::UnboundedSender<jsonrpcmsg::Message>,
+    ) -> Result<(), sacp::Error> {
+        use tokio::io::duplex;
+
+        // Create byte streams using tokio duplex channels
+        let (outgoing_write, outgoing_read) = duplex(8192);
+        let (incoming_write, incoming_read) = duplex(8192);
+
+        // Create the component with the byte streams
+        let _cleanup = self.create(
+            cx,
+            Box::pin(outgoing_write.compat_write()),
+            Box::pin(incoming_read.compat()),
+        )?;
+
+        // Delegate to ByteStreams for the transport layer
+        Box::new(sacp::ByteStreams::new(
+            incoming_write.compat_write(),
+            outgoing_read.compat(),
+        ))
+        .into_jr_transport(cx, outgoing_rx, incoming_tx)
+    }
+}
+
 impl ComponentProvider for AcpAgent {
     fn create(
         &self,

src/sacp-conductor/src/conductor.rs

@@ -117,7 +117,7 @@ impl Conductor {
         }
     }
 
-    pub fn into_handler_chain(self) -> JrHandlerChain<impl JrMessageHandler> {
+    pub fn into_handler_chain(self) -> JrHandlerChain<ConductorMessageHandler> {
         let (mut conductor_tx, mut conductor_rx) = mpsc::channel(128 /* chosen arbitrarily */);
 
         let conductor_command = self.conductor_command.unwrap_or_else(|| {
@@ -128,58 +128,33 @@ impl Conductor {
             vec![argv0]
         });
 
-        let mut handler = ConductorHandler {
+        let mut state = ConductorHandlerState {
             components: Default::default(),
             bridge_listeners: Default::default(),
             bridge_connections: Default::default(),
             conductor_command,
             proxy_mode: AtomicBool::new(false),
         };
 
-        let len_components = self.providers.len();
-
-        JrHandlerChain::new()
-            .name(self.name)
-            .on_receive_message_from_successor({
-                let mut conductor_tx = conductor_tx.clone();
-                async move |message: MessageAndCx| {
-                    // If we receive a message from our successor, we must be in proxy mode or else something odd is going on.
-                    conductor_tx
-                        .send(ConductorMessage::AgentToClient {
-                            source_component_index: len_components,
-                            message,
-                        })
-                        .await
-                        .map_err(sacp::util::internal_error)
-                }
-            })
-            // Any incoming messages from the client are client-to-agent messages targeting the first component.
-            .on_receive_message({
-                let mut conductor_tx = conductor_tx.clone();
-                async move |message: MessageAndCx| {
-                    conductor_tx
-                        .send(ConductorMessage::ClientToAgent {
-                            target_component_index: 0,
-                            message,
-                        })
-                        .await
-                        .map_err(sacp::util::internal_error)
-                }
-            })
-            .with_spawned(async move |cx| {
-                for provider in self.providers {
-                    handler.launch_proxy(&cx, &mut conductor_tx, provider)?;
-                }
+        JrHandlerChain::new_with(ConductorMessageHandler {
+            len_components: self.providers.len(),
+            conductor_tx: conductor_tx.clone(),
+        })
+        .name(self.name)
+        .with_spawned(async move |cx| {
+            for provider in self.providers {
+                state.launch_proxy(&cx, &mut conductor_tx, provider)?;
+            }
 
-                // This is the "central actor" of the conductor. Most other things forward messages
-                // via `conductor_tx` into this loop. This lets us serialize the conductor's activity.
-                while let Some(message) = conductor_rx.next().await {
-                    handler
-                        .handle_conductor_message(&cx, message, &mut conductor_tx)
-                        .await?;
-                }
-                Ok(())
-            })
+            // This is the "central actor" of the conductor. Most other things forward messages
+            // via `conductor_tx` into this loop. This lets us serialize the conductor's activity.
+            while let Some(message) = conductor_rx.next().await {
+                state
+                    .handle_conductor_message(&cx, message, &mut conductor_tx)
+                    .await?;
+            }
+            Ok(())
+        })
     }
 
     /// Convenience method to run the conductor with a transport.
@@ -199,12 +174,17 @@ impl Conductor {
     }
 }
 
+pub struct ConductorMessageHandler {
+    len_components: usize,
+    conductor_tx: mpsc::Sender<ConductorMessage>,
+}
+
 /// The conductor manages the proxy chain lifecycle and message routing.
 ///
 /// It maintains connections to all components in the chain and routes messages
 /// bidirectionally between the editor, components, and agent.
 ///
-pub struct ConductorHandler {
+struct ConductorHandlerState {
     /// Manages the TCP listeners for MCP connections that will be proxied over ACP.
     bridge_listeners: McpBridgeListeners,
 
@@ -225,7 +205,49 @@ pub struct ConductorHandler {
     proxy_mode: AtomicBool,
 }
 
-impl ConductorHandler {
+impl JrMessageHandler for ConductorMessageHandler {
+    async fn handle_message(
+        &mut self,
+        message: MessageAndCx,
+    ) -> Result<sacp::Handled<MessageAndCx>, sacp::Error> {
+        JrHandlerChain::new()
+            .on_receive_message_from_successor({
+                let len_components = self.len_components;
+                let mut conductor_tx = self.conductor_tx.clone();
+                async move |message: MessageAndCx| {
+                    // If we receive a message from our successor, we must be in proxy mode or else something odd is going on.
+                    conductor_tx
+                        .send(ConductorMessage::AgentToClient {
+                            source_component_index: len_components,
+                            message,
+                        })
+                        .await
+                        .map_err(sacp::util::internal_error)
+                }
+            })
+            // Any incoming messages from the client are client-to-agent messages targeting the first component.
+            .on_receive_message({
+                let mut conductor_tx = self.conductor_tx.clone();
+                async move |message: MessageAndCx| {
+                    conductor_tx
+                        .send(ConductorMessage::ClientToAgent {
+                            target_component_index: 0,
+                            message,
+                        })
+                        .await
+                        .map_err(sacp::util::internal_error)
+                }
+            })
+            .apply(message)
+            .await
+    }
+
+    fn describe_chain(&self) -> impl std::fmt::Debug {
+        "ConductorMessageHandler"
+    }
+}
+
+impl ConductorHandlerState {
     /// Recursively spawns components and builds the proxy chain.
     ///
     /// This function implements the recursive chain building pattern:

src/sacp-conductor/tests/arrow_proxy_eliza.rs

@@ -24,17 +24,20 @@ async fn test_conductor_with_arrow_proxy_and_eliza() -> Result<(), sacp::Error>
 
     // Spawn the conductor
     let conductor_handle = tokio::spawn(async move {
-        Conductor::run(
+        Conductor::new(
             "conductor".to_string(),
+            vec![arrow_proxy_agent, eliza_agent],
+            None,
+        )
+        .run(sacp::ByteStreams::new(
             conductor_write.compat_write(),
             conductor_read.compat(),
-            vec![Box::new(arrow_proxy_agent), Box::new(eliza_agent)],
-        )
+        ))
         .await
     });
 
-    // Editor side: connect and send a prompt using helper
-    let editor_handle = tokio::spawn(async move {
+    // Wait for editor to complete and get the result
+    let result = tokio::time::timeout(std::time::Duration::from_secs(30), async move {
         let result =
             yolo_prompt(editor_write.compat_write(), editor_read.compat(), "Hello").await?;
 
@@ -47,14 +50,10 @@ async fn test_conductor_with_arrow_proxy_and_eliza() -> Result<(), sacp::Error>
         );
 
         Ok::<String, sacp::Error>(result)
-    });
-
-    // Wait for editor to complete and get the result
-    let result = tokio::time::timeout(std::time::Duration::from_secs(30), editor_handle)
-        .await
-        .expect("Test timed out")
-        .expect("Editor task panicked")
-        .expect("Editor failed");
+    })
+    .await
+    .expect("Test timed out")
+    .expect("Editor failed");
 
     tracing::info!(
         ?result,

src/sacp-conductor/tests/initialization_sequence.rs

@@ -6,13 +6,11 @@
 //! 3. Proxy components must accept the capability or initialization fails
 //! 4. Last component (agent) never receives proxy capability offer
 
-use futures::{AsyncRead, AsyncWrite};
+use futures::channel::mpsc;
 use sacp::schema::{AgentCapabilities, InitializeRequest, InitializeResponse};
-use sacp::{JrConnectionCx, JrHandlerChain, MetaCapabilityExt, Proxy};
-use sacp_conductor::component::{Cleanup, ComponentProvider};
+use sacp::{IntoJrTransport, JrConnectionCx, JrHandlerChain, MetaCapabilityExt, Proxy};
 use sacp_conductor::conductor::Conductor;
 use sacp_proxy::JrCxExt;
-use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::Mutex;
 
@@ -64,23 +62,26 @@ struct InitComponentProvider {
 }
 
 impl InitComponentProvider {
-    fn new(config: &Arc<InitConfig>) -> Box<dyn ComponentProvider> {
+    fn new(config: &Arc<InitConfig>) -> Box<dyn IntoJrTransport> {
         Box::new(Self {
             config: config.clone(),
         })
     }
 }
 
-impl ComponentProvider for InitComponentProvider {
-    fn create(
-        &self,
+impl IntoJrTransport for InitComponentProvider {
+    fn into_jr_transport(
+        self: Box<Self>,
         cx: &JrConnectionCx,
-        outgoing_bytes: Pin<Box<dyn AsyncWrite + Send>>,
-        incoming_bytes: Pin<Box<dyn AsyncRead + Send>>,
-    ) -> Result<Cleanup, sacp::Error> {
+        outgoing_rx: mpsc::UnboundedReceiver<jsonrpcmsg::Message>,
+        incoming_tx: mpsc::UnboundedSender<jsonrpcmsg::Message>,
+    ) -> Result<(), sacp::Error> {
         let config = Arc::clone(&self.config);
+
         cx.spawn(async move {
-            let transport = sacp::ByteStreams::new(outgoing_bytes, incoming_bytes);
+            // Create the channel-based transport
+            let transport = sacp::Channels::new(outgoing_rx, incoming_tx);
+
             JrHandlerChain::new()
                 .name("init-component-provider")
                 .on_receive_request(async move |mut request: InitializeRequest, request_cx| {
@@ -117,12 +118,12 @@ impl ComponentProvider for InitComponentProvider {
                 .await
         })?;
 
-        Ok(Cleanup::None)
+        Ok(())
     }
 }
 
 async fn run_test_with_components(
-    components: Vec<Box<dyn ComponentProvider>>,
+    components: Vec<Box<dyn IntoJrTransport>>,
     editor_task: impl AsyncFnOnce(JrConnectionCx) -> Result<(), sacp::Error>,
 ) -> Result<(), sacp::Error> {
     // Set up editor <-> conductor communication
@@ -133,16 +134,15 @@ async fn run_test_with_components(
 
     JrHandlerChain::new()
         .name("editor-to-connector")
-        .with_spawned(async move {
-            Conductor::run(
-                "conductor".to_string(),
-                conductor_out.compat_write(),
-                conductor_in.compat(),
-                components,
-            )
-            .await
+        .with_spawned(|_cx| async move {
+            Conductor::new("conductor".to_string(), components, None)
+                .run(sacp::ByteStreams::new(
+                    conductor_out.compat_write(),
+                    conductor_in.compat(),
+                ))
+                .await
         })
-        .serve_with(transport, editor_task)
+        .with_client(transport, editor_task)
         .await
 }
 

src/sacp-conductor/tests/mcp-integration.rs

@@ -53,17 +53,19 @@ async fn run_test_with_components(
 
     JrHandlerChain::new()
         .name("editor-to-connector")
-        .with_spawned(async move {
-            Conductor::run_with_command(
+        .with_spawned(|_cx| async move {
+            Conductor::new(
                 "conductor".to_string(),
-                conductor_out.compat_write(),
-                conductor_in.compat(),
                 components,
                 Some(conductor_command()),
             )
+            .run(sacp::ByteStreams::new(
+                conductor_out.compat_write(),
+                conductor_in.compat(),
+            ))
             .await
         })
-        .serve_with(transport, editor_task)
+        .with_client(transport, editor_task)
         .await
 }
 
@@ -139,20 +141,22 @@ async fn test_agent_handles_prompt() -> Result<(), sacp::Error> {
                     .map_err(|_| sacp::Error::internal_error())
             }
         })
-        .with_spawned(async move {
-            Conductor::run_with_command(
+        .with_spawned(|_cx| async move {
+            Conductor::new(
                 "mcp-integration-conductor".to_string(),
-                conductor_out.compat_write(),
-                conductor_in.compat(),
                 vec![
                     mcp_integration::proxy::create(),
                     mcp_integration::agent::create(),
                 ],
                 Some(conductor_command()),
             )
+            .run(sacp::ByteStreams::new(
+                conductor_out.compat_write(),
+                conductor_in.compat(),
+            ))
             .await
         })
-        .serve_with(transport, async |editor_cx| {
+        .with_client(transport, async |editor_cx| {
             // Initialize
             recv(editor_cx.send_request(InitializeRequest {
                 protocol_version: Default::default(),