Wire ACPX session selection through chat channels

Author: bglusman

BACKLOG.md

@@ -79,6 +79,10 @@
 - [ ] Explore a local web channel for desktop/LAN testing that uses the same
       identity, routing, message-envelope, artifact, and proxy policy paths as
       Telegram, Matrix, and text channels
+- [ ] Prototype a native ACP client adapter using the `agent-client-protocol`
+      Rust crate, with Zed's `codex-acp` as the first smoke target
+- [ ] Evaluate ACP orchestrators such as AgentPool, cagent, and fast-agent as
+      recipe-backed async work backends
 - [ ] Architecture decision records (ADRs)
 
 ### Observability

README.md

@@ -183,7 +183,6 @@ bash scripts/install-git-hooks.sh
 - [Roadmap](docs/roadmap/)
 - [Staging test matrix](docs/staging-test-matrix.md)
 - [Channel secret-input deprecation note](docs/roadmap/channel-secret-input-deprecation.md)
-- [Internal research and planning notes](research/)
 
 ## License
 

crates/calciforge/src/adapters/acpx.rs

@@ -14,6 +14,9 @@ use tracing::{debug, info};
 
 use crate::adapters::{AdapterError, AgentAdapter, DispatchContext};
 
+/// Shared cwd for ACPX sessions created through Calciforge.
+pub const ACPX_SESSION_DIR: &str = "/tmp/acpx-sessions";
+
 /// ACPX adapter — wraps acpx CLI for ACP agent communication
 pub struct AcpxAdapter {
     agent_name: String,
@@ -36,7 +39,7 @@ impl AcpxAdapter {
             _args: args.unwrap_or_default(),
             env: env.unwrap_or_default(),
             timeout_ms: timeout_ms.unwrap_or(300_000),
-            session_dir: PathBuf::from("/tmp/acpx-sessions"),
+            session_dir: PathBuf::from(ACPX_SESSION_DIR),
         }
     }
 
@@ -92,7 +95,9 @@ impl AcpxAdapter {
         let output = Command::new("acpx")
             .arg(&self.agent_name)
             .arg("sessions")
-            .arg("new")
+            .arg("ensure")
+            .arg("--name")
+            .arg(session_name)
             .current_dir(&self.session_dir)
             .envs(&self.env)
             .output()
@@ -142,6 +147,34 @@ impl AcpxAdapter {
             .to_string()
     }
 
+    async fn run_session_prompt(
+        &self,
+        message: &str,
+        session: Option<&str>,
+    ) -> Result<std::process::Output, AdapterError> {
+        let mut cmd = Command::new("acpx");
+        cmd.arg("--format").arg("text").arg(&self.agent_name);
+        if let Some(session) = session {
+            cmd.arg("--session").arg(session);
+        }
+        cmd.arg("prompt")
+            .arg(message)
+            .current_dir(&self.session_dir)
+            .envs(&self.env)
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped());
+
+        let timeout = std::time::Duration::from_millis(self.timeout_ms);
+        let child = cmd
+            .spawn()
+            .map_err(|e| AdapterError::Unavailable(format!("Failed to spawn acpx: {}", e)))?;
+
+        tokio::time::timeout(timeout, child.wait_with_output())
+            .await
+            .map_err(|_| AdapterError::Unavailable("acpx prompt timed out".to_string()))?
+            .map_err(|e| AdapterError::Unavailable(format!("Failed to run acpx: {}", e)))
+    }
+
     /// Execute one-shot prompt (no session persistence)
     async fn exec_prompt(&self, message: &str) -> Result<String, AdapterError> {
         self.ensure_session_dir().await?;
@@ -190,40 +223,38 @@ impl AcpxAdapter {
     }
 
     /// Send prompt to persistent session
-    async fn session_prompt(&self, message: &str) -> Result<String, AdapterError> {
+    async fn session_prompt(
+        &self,
+        message: &str,
+        session: Option<&str>,
+    ) -> Result<String, AdapterError> {
         self.ensure_session_dir().await?;
 
         // Use cwd session (default session name)
-        info!(agent = %self.agent_name, "Running acpx prompt with session");
-
-        let mut cmd = Command::new("acpx");
-        cmd.arg("--format")
-            .arg("text")
-            .arg(&self.agent_name)
-            .arg("prompt")
-            .arg(message)
-            .current_dir(&self.session_dir)
-            .envs(&self.env)
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped());
-
-        let timeout = std::time::Duration::from_millis(self.timeout_ms);
-        let child = cmd
-            .spawn()
-            .map_err(|e| AdapterError::Unavailable(format!("Failed to spawn acpx: {}", e)))?;
-
-        let result = tokio::time::timeout(timeout, child.wait_with_output()).await;
+        info!(
+            agent = %self.agent_name,
+            session = ?session,
+            "Running acpx prompt with session"
+        );
 
-        match result {
-            Ok(Ok(output)) => {
+        match self.run_session_prompt(message, session).await {
+            Ok(output) => {
                 if !output.status.success() {
                     let stderr = String::from_utf8_lossy(&output.stderr);
                     // Session might not exist — try creating it
                     if stderr.contains("session") || stderr.contains("not found") {
                         info!("Session not found, creating...");
-                        self.ensure_session("cwd").await?;
-                        // Retry
-                        return self.exec_prompt(message).await;
+                        self.ensure_session(session.unwrap_or("cwd")).await?;
+                        let retry = self.run_session_prompt(message, session).await?;
+                        if retry.status.success() {
+                            let stdout = String::from_utf8_lossy(&retry.stdout);
+                            return Ok(Self::strip_acpx_noise(&stdout));
+                        }
+                        let retry_stderr = String::from_utf8_lossy(&retry.stderr);
+                        return Err(AdapterError::Protocol(format!(
+                            "acpx prompt failed: {}",
+                            retry_stderr
+                        )));
                     }
                     return Err(AdapterError::Protocol(format!(
                         "acpx prompt failed: {}",
@@ -233,13 +264,7 @@ impl AcpxAdapter {
                 let stdout = String::from_utf8_lossy(&output.stdout);
                 Ok(Self::strip_acpx_noise(&stdout))
             }
-            Ok(Err(e)) => Err(AdapterError::Unavailable(format!(
-                "Failed to run acpx: {}",
-                e
-            ))),
-            Err(_) => Err(AdapterError::Unavailable(
-                "acpx prompt timed out".to_string(),
-            )),
+            Err(e) => Err(e),
         }
     }
 }
@@ -262,8 +287,9 @@ impl AgentAdapter for AcpxAdapter {
         };
 
         // Try session mode first, fall back to exec
-        match self.session_prompt(&message).await {
+        match self.session_prompt(&message, ctx.session).await {
             Ok(response) => Ok(response),
+            Err(e @ AdapterError::Protocol(_)) if ctx.session.is_some() => Err(e),
             Err(AdapterError::Protocol(_)) => {
                 // Session error — try one-shot exec
                 self.exec_prompt(&message).await

crates/calciforge/src/adapters/mod.rs

@@ -113,6 +113,8 @@ pub struct DispatchContext<'a> {
     /// Optional per-request model override (used by alloy routing).
     #[allow(dead_code)]
     pub model_override: Option<&'a str>,
+    /// Optional downstream session selected for session-aware adapters.
+    pub session: Option<&'a str>,
 }
 
 impl<'a> DispatchContext<'a> {
@@ -122,6 +124,7 @@ impl<'a> DispatchContext<'a> {
             message,
             sender: None,
             model_override: None,
+            session: None,
         }
     }
 }

crates/calciforge/src/adapters/openai_compat.rs

@@ -279,6 +279,7 @@ mod tests {
                 message: "hello",
                 sender: Some("brian"),
                 model_override: Some("override"),
+                session: None,
             })
             .await
             .unwrap();

crates/calciforge/src/adapters/openclaw_channel.rs

@@ -493,6 +493,7 @@ mod tests {
                 message: "hello from calciforge",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await
             .expect("dispatch should succeed");
@@ -536,6 +537,7 @@ mod tests {
                 message: "route this",
                 sender: Some("renee"),
                 model_override: None,
+                session: None,
             })
             .await
             .expect("dispatch should return reply callback");
@@ -566,6 +568,7 @@ mod tests {
                 message: "first",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await
             .expect("first dispatch should start reply server");
@@ -577,6 +580,7 @@ mod tests {
                 message: "second",
                 sender: Some("renee"),
                 model_override: None,
+                session: None,
             })
             .await
             .expect("rebuilt adapter should reuse reply server/router");
@@ -603,6 +607,7 @@ mod tests {
                 message: "will not send",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await
             .expect_err("conflicting reply auth token should fail");

crates/calciforge/src/adapters/openclaw_native.rs

@@ -463,6 +463,7 @@ mod tests {
             message: "hello",
             sender: Some("brian"),
             model_override: None,
+            session: None,
         };
         let result = a.dispatch_with_context(ctx).await;
 
@@ -597,6 +598,7 @@ mod tests {
                 message: msg,
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             };
             let _ = a.dispatch_with_context(ctx).await;
         }
@@ -667,6 +669,7 @@ mod tests {
             message: "status check",
             sender: Some("renee"),
             model_override: None,
+            session: None,
         };
         let _ = a.dispatch_with_context(ctx).await;
 

crates/calciforge/src/adapters/zeroclaw_native.rs

@@ -194,6 +194,7 @@ impl AgentAdapter for ZeroClawNativeAdapter {
             message: &full_message,
             sender: ctx.sender,
             model_override: ctx.model_override,
+            session: ctx.session,
         };
 
         match self.inner.dispatch_with_context(inner_ctx).await {
@@ -398,6 +399,7 @@ mod tests {
                 message: "what is 2+2?",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await;
         assert!(r1.is_ok(), "first dispatch failed: {:?}", r1);
@@ -408,6 +410,7 @@ mod tests {
                 message: "and 3+3?",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await;
         assert!(r2.is_ok(), "second dispatch failed: {:?}", r2);
@@ -496,6 +499,7 @@ mod tests {
                 message: "brian first message",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await;
 
@@ -505,6 +509,7 @@ mod tests {
                 message: "renee first message",
                 sender: Some("renee"),
                 model_override: None,
+                session: None,
             })
             .await;
 
@@ -514,6 +519,7 @@ mod tests {
                 message: "brian second message",
                 sender: Some("brian"),
                 model_override: None,
+                session: None,
             })
             .await;
 

crates/calciforge/src/channels/matrix.rs

@@ -832,14 +832,16 @@ async fn handle_message(
     );
     let dispatch_start = std::time::Instant::now();
     let model_override = cmd_handler.active_model_for_identity(identity_id);
+    let selected_session = cmd_handler.active_session_for(identity_id, &agent_id);
 
     match router
-        .dispatch_message_with_sender_and_model(
+        .dispatch_message_with_sender_model_and_session(
             &augmented,
             &agent,
             config,
             Some(identity_id),
             model_override.as_deref(),
+            selected_session.as_deref(),
         )
         .await
     {

crates/calciforge/src/channels/signal.rs

@@ -363,6 +363,9 @@ impl<C: Channel + ?Sized + 'static> SignalChannel<C> {
 
         let identity_id = identity.id.clone();
         let model_override = self.command_handler.active_model_for_identity(&identity_id);
+        let selected_session = self
+            .command_handler
+            .active_session_for(&identity_id, &agent_id);
         let preserve_native_commands = crate::adapters::agent_supports_native_commands(&agent);
 
         tokio::spawn(async move {
@@ -379,12 +382,13 @@ impl<C: Channel + ?Sized + 'static> SignalChannel<C> {
             let dispatch_start = std::time::Instant::now();
             match self
                 .router
-                .dispatch_message_with_sender_and_model(
+                .dispatch_message_with_sender_model_and_session(
                     &augmented,
                     &agent,
                     &self.config,
                     Some(&identity_id),
                     model_override.as_deref(),
+                    selected_session.as_deref(),
                 )
                 .await
             {