Merge pull request #30 from nikomatsakis/main

nikomatsakis · web-flow · commit c4030d0cc0e0 · 2025-11-12T06:20:17.000-05:00
cleanup component trait, access connection_cx explicitly
diff --git a/src/elizacp/src/lib.rs b/src/elizacp/src/lib.rs
@@ -109,14 +109,16 @@ impl ElizaAgent {
 
         tracing::debug!("Eliza response: {}", response_text);
 
-        request_cx.send_notification(SessionNotification {
-            session_id: session_id.clone(),
-            update: SessionUpdate::AgentMessageChunk(ContentChunk {
-                content: response_text.into(),
+        request_cx
+            .connection_cx()
+            .send_notification(SessionNotification {
+                session_id: session_id.clone(),
+                update: SessionUpdate::AgentMessageChunk(ContentChunk {
+                    content: response_text.into(),
+                    meta: None,
+                }),
                 meta: None,
-            }),
-            meta: None,
-        })?;
+            })?;
 
         // Complete the request
         request_cx.respond(PromptResponse {
diff --git a/src/sacp-conductor/src/conductor.rs b/src/sacp-conductor/src/conductor.rs
@@ -223,8 +223,8 @@ impl Conductor {
 }
 
 impl sacp::Component for Conductor {
-    async fn serve(self, channels: sacp::Channels) -> Result<(), sacp::Error> {
-        self.run(channels).await
+    async fn serve(self, client: impl sacp::Component) -> Result<(), sacp::Error> {
+        self.run(client).await
     }
 }
 
@@ -824,7 +824,7 @@ impl ConductorHandlerState {
             for mcp_server in &mut request.mcp_servers {
                 self.bridge_listeners
                     .transform_mcp_servers(
-                        &request_cx,
+                        &request_cx.connection_cx(),
                         mcp_server,
                         conductor_tx,
                         &self.conductor_command,
diff --git a/src/sacp-conductor/tests/initialization_sequence.rs b/src/sacp-conductor/tests/initialization_sequence.rs
@@ -7,7 +7,7 @@
 //! 4. Last component (agent) never receives proxy capability offer
 
 use sacp::schema::{AgentCapabilities, InitializeRequest, InitializeResponse};
-use sacp::{Channels, Component, JrHandlerChain, MetaCapabilityExt, Proxy};
+use sacp::{Component, JrHandlerChain, MetaCapabilityExt, Proxy};
 use sacp_conductor::conductor::Conductor;
 use sacp_proxy::JrCxExt;
 use std::sync::Arc;
@@ -69,7 +69,7 @@ impl InitComponent {
 }
 
 impl Component for InitComponent {
-    async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
         let config = Arc::clone(&self.config);
 
         {
@@ -86,6 +86,7 @@ impl Component for InitComponent {
 
                     if config.respond_with_proxy {
                         request_cx
+                            .connection_cx()
                             .send_request_to_successor(request)
                             .await_when_result_received(async move |response| {
                                 let mut response = response?;
@@ -105,7 +106,7 @@ impl Component for InitComponent {
                         request_cx.respond(response)
                     }
                 })
-                .serve(channels)
+                .serve(client)
                 .await
         }
     }
diff --git a/src/sacp-conductor/tests/mcp-integration.rs b/src/sacp-conductor/tests/mcp-integration.rs
@@ -9,7 +9,6 @@ mod mcp_integration;
 
 use expect_test::expect;
 use futures::{SinkExt, StreamExt, channel::mpsc};
-use sacp::Component;
 use sacp::JrHandlerChain;
 use sacp::schema::{
     ContentBlock, InitializeRequest, NewSessionRequest, PromptRequest, SessionNotification,
diff --git a/src/sacp-conductor/tests/mcp_integration/agent.rs b/src/sacp-conductor/tests/mcp_integration/agent.rs
@@ -6,7 +6,7 @@ use sacp::schema::{
     McpServer, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse,
     SessionNotification, SessionUpdate, StopReason, TextContent,
 };
-use sacp::{BoxFuture, Channels, Component, JrHandlerChain, JrRequestCx};
+use sacp::{Component, JrHandlerChain, JrRequestCx};
 use std::sync::Arc;
 use tokio::sync::Mutex;
 
@@ -22,7 +22,7 @@ struct AgentState {
 }
 
 impl Component for AgentComponent {
-    async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
         let state = AgentState {
             mcp_servers: Arc::new(Mutex::new(Vec::new())),
         };
@@ -95,7 +95,7 @@ impl Component for AgentComponent {
                     connection_cx.spawn(Self::respond_to_prompt(state, request, request_cx))
                 }
             })
-            .serve(channels)
+            .serve(client)
             .await
     }
 }
diff --git a/src/sacp-conductor/tests/mcp_integration/proxy.rs b/src/sacp-conductor/tests/mcp_integration/proxy.rs
@@ -1,21 +1,21 @@
 //! Proxy component that provides MCP tools
 
-use sacp::{BoxFuture, Channels, Component, JrHandlerChain};
+use sacp::{Component, JrHandlerChain};
 use sacp_proxy::{AcpProxyExt, McpServiceRegistry};
 
 use crate::mcp_integration::mcp_server::TestMcpServer;
 
 pub struct ProxyComponent;
 
 impl Component for ProxyComponent {
-    async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
         JrHandlerChain::new()
             .name("proxy-component")
             .provide_mcp(
                 McpServiceRegistry::default().with_rmcp_server("test", TestMcpServer::new)?,
             )
             .proxy()
-            .serve(channels)
+            .serve(client)
             .await
     }
 }
diff --git a/src/sacp-conductor/tests/nested_conductor.rs b/src/sacp-conductor/tests/nested_conductor.rs
@@ -17,7 +17,7 @@
 //! - Inner conductor operates in proxy mode, forwarding to eliza
 //! - Outer conductor receives the ">>" prefixed response
 
-use sacp::{BoxFuture, Channels, Component};
+use sacp::Component;
 use sacp_conductor::conductor::Conductor;
 use sacp_test::arrow_proxy::run_arrow_proxy;
 use sacp_test::test_client::yolo_prompt;
@@ -31,8 +31,8 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
 struct MockArrowProxy;
 
 impl Component for MockArrowProxy {
-    async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {
-        run_arrow_proxy(channels).await
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
+        run_arrow_proxy(client).await
     }
 }
 
@@ -41,8 +41,8 @@ impl Component for MockArrowProxy {
 struct MockEliza;
 
 impl Component for MockEliza {
-    async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {
-        elizacp::run_elizacp(channels).await
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
+        elizacp::run_elizacp(client).await
     }
 }
 
@@ -59,7 +59,7 @@ impl MockInnerConductor {
 }
 
 impl Component for MockInnerConductor {
-    async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
         // Create mock arrow proxy components for the inner conductor
         // This conductor is ONLY proxies - no actual agent
         let mut components: Vec<sacp::DynComponent> = Vec::new();
@@ -68,7 +68,7 @@ impl Component for MockInnerConductor {
         }
 
         Conductor::new("inner-conductor".to_string(), components, None)
-            .run(channels)
+            .run(client)
             .await
     }
 }
diff --git a/src/sacp-proxy/src/mcp_server.rs b/src/sacp-proxy/src/mcp_server.rs
@@ -232,9 +232,10 @@ impl McpServiceRegistry {
         // Every request/notification that is sent over `mcp_server_tx` we will
         // send to the MCP server.
         let spawn_results = request_cx
+            .connection_cx()
             .spawn({
                 let connection_id = connection_id.clone();
-                let outer_cx = request_cx.clone();
+                let outer_cx = request_cx.connection_cx();
 
                 let transport = sacp::ByteStreams::new(
                     mcp_client_write.compat_write(),
@@ -277,7 +278,7 @@ impl McpServiceRegistry {
             })
             .and_then(|()| {
                 // Spawn MCP server task
-                request_cx.spawn(async move {
+                request_cx.connection_cx().spawn(async move {
                     registered_server
                         .spawn
                         .spawn(Box::pin(mcp_server_write), Box::pin(mcp_server_read))
@@ -386,7 +387,7 @@ impl McpServiceRegistry {
         let mut request = match result {
             Ok(request) => request,
             Err(err) => {
-                request_cx.send_error_notification(err)?;
+                request_cx.connection_cx().send_error_notification(err)?;
                 return Ok(Handled::Yes);
             }
         };
@@ -403,6 +404,7 @@ impl McpServiceRegistry {
 
         // Forward it to the successor.
         request_cx
+            .connection_cx()
             .send_request_to_successor(request)
             .forward_to_request_cx(request_cx.cast())?;
 
diff --git a/src/sacp-proxy/src/to_from_successor.rs b/src/sacp-proxy/src/to_from_successor.rs
@@ -525,6 +525,7 @@ impl JrMessageHandler for ProxyHandler {
                 ) {
                     let request = result?;
                     request_cx
+                        .connection_cx()
                         .send_request(request.request)
                         .forward_to_request_cx(request_cx)?;
                     return Ok(Handled::Yes);
@@ -543,6 +544,7 @@ impl JrMessageHandler for ProxyHandler {
 
                 // If we receive any other request, send it to our successor.
                 request_cx
+                    .connection_cx()
                     .send_request_to_successor(request)
                     .forward_to_request_cx(request_cx)?;
                 Ok(Handled::Yes)
@@ -596,6 +598,7 @@ impl ProxyHandler {
 
         request = request.remove_meta_capability(Proxy);
         request_cx
+            .connection_cx()
             .send_request_to_successor(request)
             .await_when_result_received(async move |mut result| {
                 result = result.map(|r| r.add_meta_capability(Proxy));
diff --git a/src/sacp-test/src/lib.rs b/src/sacp-test/src/lib.rs
@@ -10,7 +10,7 @@ pub mod test_client;
 pub struct MockTransport;
 
 impl Component for MockTransport {
-    async fn serve(self, _channels: Channels) -> Result<(), Error> {
+    async fn serve(self, _client: impl Component) -> Result<(), Error> {
         panic!("MockTransport should never be used in running code - it's only for doctests")
     }
 }
diff --git a/src/sacp-tokio/src/acp_agent.rs b/src/sacp-tokio/src/acp_agent.rs
@@ -162,15 +162,15 @@ impl Drop for ChildHolder {
 }
 
 impl sacp::Component for AcpAgent {
-    async fn serve(self, channels: sacp::Channels) -> Result<(), sacp::Error> {
+    async fn serve(self, client: impl sacp::Component) -> Result<(), sacp::Error> {
         let (child_stdin, child_stdout, child) = self.spawn_process()?;
 
         // Hold the child process - it will be killed when this future completes
         let _child_holder = ChildHolder { _child: child };
 
         // Create the ByteStreams component and serve it
         sacp::ByteStreams::new(child_stdin.compat_write(), child_stdout.compat())
-            .serve(channels)
+            .serve(client)
             .await
     }
 }
diff --git a/src/sacp-tokio/src/lib.rs b/src/sacp-tokio/src/lib.rs
@@ -13,12 +13,12 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
 pub struct Stdio;
 
 impl Component for Stdio {
-    async fn serve(self, channels: sacp::Channels) -> Result<(), sacp::Error> {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
         ByteStreams::new(
             tokio::io::stdout().compat_write(),
             tokio::io::stdin().compat(),
         )
-        .serve(channels)
+        .serve(client)
         .await
     }
 }
diff --git a/src/sacp/src/component.rs b/src/sacp/src/component.rs
diff --git a/src/sacp/src/jsonrpc.rs b/src/sacp/src/jsonrpc.rs
diff --git a/src/sacp/src/jsonrpc/actors.rs b/src/sacp/src/jsonrpc/actors.rs
diff --git a/src/sacp/src/lib.rs b/src/sacp/src/lib.rs
diff --git a/src/sacp/tests/match_message.rs b/src/sacp/tests/match_message.rs

Original file line number	Diff line number	Diff line change
`@@ -223,8 +223,8 @@ impl Conductor {`
`223`	`223`	`}`
`224`	`224`
`225`	`225`	`impl sacp::Component for Conductor {`
`226`		`- async fn serve(self, channels: sacp::Channels) -> Result<(), sacp::Error> {`
`227`		`- self.run(channels).await`
	`226`	`+ async fn serve(self, client: impl sacp::Component) -> Result<(), sacp::Error> {`
	`227`	`+ self.run(client).await`
`228`	`228`	`}`
`229`	`229`	`}`
`230`	`230`
`@@ -824,7 +824,7 @@ impl ConductorHandlerState {`
`824`	`824`	`for mcp_server in &mut request.mcp_servers {`
`825`	`825`	`self.bridge_listeners`
`826`	`826`	`.transform_mcp_servers(`
`827`		`- &request_cx,`
	`827`	`+ &request_cx.connection_cx(),`
`828`	`828`	`mcp_server,`
`829`	`829`	`conductor_tx,`
`830`	`830`	`&self.conductor_command,`
Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,7 @@`
`7`	`7`	`//! 4. Last component (agent) never receives proxy capability offer`
`8`	`8`
`9`	`9`	`use sacp::schema::{AgentCapabilities, InitializeRequest, InitializeResponse};`
`10`		`-use sacp::{Channels, Component, JrHandlerChain, MetaCapabilityExt, Proxy};`
	`10`	`+use sacp::{Component, JrHandlerChain, MetaCapabilityExt, Proxy};`
`11`	`11`	`use sacp_conductor::conductor::Conductor;`
`12`	`12`	`use sacp_proxy::JrCxExt;`
`13`	`13`	`use std::sync::Arc;`
`@@ -69,7 +69,7 @@ impl InitComponent {`
`69`	`69`	`}`
`70`	`70`
`71`	`71`	`impl Component for InitComponent {`
`72`		`- async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {`
	`72`	`+ async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {`
`73`	`73`	`let config = Arc::clone(&self.config);`
`74`	`74`
`75`	`75`	`{`
`@@ -86,6 +86,7 @@ impl Component for InitComponent {`
`86`	`86`
`87`	`87`	`if config.respond_with_proxy {`
`88`	`88`	`request_cx`
	`89`	`+ .connection_cx()`
`89`	`90`	`.send_request_to_successor(request)`
`90`	`91`	`.await_when_result_received(async move \|response\| {`
`91`	`92`	`let mut response = response?;`
`@@ -105,7 +106,7 @@ impl Component for InitComponent {`
`105`	`106`	`request_cx.respond(response)`
`106`	`107`	`}`
`107`	`108`	`})`
`108`		`- .serve(channels)`
	`109`	`+ .serve(client)`
`109`	`110`	`.await`
`110`	`111`	`}`
`111`	`112`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,7 +6,7 @@ use sacp::schema::{`
`6`	`6`	`McpServer, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse,`
`7`	`7`	`SessionNotification, SessionUpdate, StopReason, TextContent,`
`8`	`8`	`};`
`9`		`-use sacp::{BoxFuture, Channels, Component, JrHandlerChain, JrRequestCx};`
	`9`	`+use sacp::{Component, JrHandlerChain, JrRequestCx};`
`10`	`10`	`use std::sync::Arc;`
`11`	`11`	`use tokio::sync::Mutex;`
`12`	`12`
`@@ -22,7 +22,7 @@ struct AgentState {`
`22`	`22`	`}`
`23`	`23`
`24`	`24`	`impl Component for AgentComponent {`
`25`		`- async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {`
	`25`	`+ async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {`
`26`	`26`	`let state = AgentState {`
`27`	`27`	`mcp_servers: Arc::new(Mutex::new(Vec::new())),`
`28`	`28`	`};`
`@@ -95,7 +95,7 @@ impl Component for AgentComponent {`
`95`	`95`	`connection_cx.spawn(Self::respond_to_prompt(state, request, request_cx))`
`96`	`96`	`}`
`97`	`97`	`})`
`98`		`- .serve(channels)`
	`98`	`+ .serve(client)`
`99`	`99`	`.await`
`100`	`100`	`}`
`101`	`101`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@`
`17`	`17`	`//! - Inner conductor operates in proxy mode, forwarding to eliza`
`18`	`18`	`//! - Outer conductor receives the ">>" prefixed response`
`19`	`19`
`20`		`-use sacp::{BoxFuture, Channels, Component};`
	`20`	`+use sacp::Component;`
`21`	`21`	`use sacp_conductor::conductor::Conductor;`
`22`	`22`	`use sacp_test::arrow_proxy::run_arrow_proxy;`
`23`	`23`	`use sacp_test::test_client::yolo_prompt;`
`@@ -31,8 +31,8 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};`
`31`	`31`	`struct MockArrowProxy;`
`32`	`32`
`33`	`33`	`impl Component for MockArrowProxy {`
`34`		`- async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {`
`35`		`- run_arrow_proxy(channels).await`
	`34`	`+ async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {`
	`35`	`+ run_arrow_proxy(client).await`
`36`	`36`	`}`
`37`	`37`	`}`
`38`	`38`
`@@ -41,8 +41,8 @@ impl Component for MockArrowProxy {`
`41`	`41`	`struct MockEliza;`
`42`	`42`
`43`	`43`	`impl Component for MockEliza {`
`44`		`- async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {`
`45`		`- elizacp::run_elizacp(channels).await`
	`44`	`+ async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {`
	`45`	`+ elizacp::run_elizacp(client).await`
`46`	`46`	`}`
`47`	`47`	`}`
`48`	`48`
`@@ -59,7 +59,7 @@ impl MockInnerConductor {`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`impl Component for MockInnerConductor {`
`62`		`- async fn serve(self, channels: Channels) -> Result<(), sacp::Error> {`
	`62`	`+ async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {`
`63`	`63`	`// Create mock arrow proxy components for the inner conductor`
`64`	`64`	`// This conductor is ONLY proxies - no actual agent`
`65`	`65`	`let mut components: Vec<sacp::DynComponent> = Vec::new();`
`@@ -68,7 +68,7 @@ impl Component for MockInnerConductor {`
`68`	`68`	`}`
`69`	`69`
`70`	`70`	`Conductor::new("inner-conductor".to_string(), components, None)`
`71`		`- .run(channels)`
	`71`	`+ .run(client)`
`72`	`72`	`.await`
`73`	`73`	`}`
`74`	`74`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ pub mod test_client;`
`10`	`10`	`pub struct MockTransport;`
`11`	`11`
`12`	`12`	`impl Component for MockTransport {`
`13`		`- async fn serve(self, _channels: Channels) -> Result<(), Error> {`
	`13`	`+ async fn serve(self, _client: impl Component) -> Result<(), Error> {`
`14`	`14`	`panic!("MockTransport should never be used in running code - it's only for doctests")`
`15`	`15`	`}`
`16`	`16`	`}`