feat(acp): AgentsMesh _loopal/* control-panel extension (#195)

Backend support for the AgentsMesh Loopal graphical control console over ACP.

- translate/panel.rs: bgTask/cron/task/mcp/topology/goal/model/thinking → _loopal/*
- adapter/control_command.rs + control.rs: session/control_request dispatch
  (18 loopal.* controls + set_model); unknown subtype → METHOD_NOT_FOUND
- adapter/snapshot.rs: cold-start replay of view/snapshot as _loopal/* events
- adapter/lifecycle.rs: advertise agentsmeshExtensions.controlRequest on initialize
- adapter/events.rs: mode dual-emit (standard ACP CurrentModeUpdate + _loopal/mode)
- translate/mod.rs: route non-ACP events through the panel translator
- tests: translate golden-fixture contract + control round-trip e2e + unit

Author: yishuiliunian

crates/loopal-acp/BUILD.bazel

@@ -29,6 +29,7 @@ rust_test(
 rust_test(
     name = "loopal-acp_test",
     srcs = glob(["tests/**/*.rs"]),
+    compile_data = glob(["tests/fixtures/*.json"]),
     crate_root = "tests/suite.rs",
     edition = "2024",
     local = True,

crates/loopal-acp/src/adapter/control.rs

@@ -4,6 +4,7 @@ use loopal_protocol::ControlCommand;
 use serde_json::Value;
 
 use crate::adapter::AcpAdapter;
+use crate::adapter::control_command::parse_loopal_control;
 use crate::jsonrpc;
 
 impl AcpAdapter {
@@ -77,4 +78,31 @@ impl AcpAdapter {
             }
         }
     }
+
+    /// Handle `session/control_request` — AgentsMesh ACP extension carrying
+    /// Loopal control-panel actions (bgTaskKill / cronDelete / compact / clear)
+    /// and the runner's `set_model`. Unknown subtypes are rejected so the
+    /// runner degrades gracefully.
+    pub(crate) async fn handle_control_request(&self, id: i64, params: Value) {
+        let subtype = params["subtype"].as_str().unwrap_or("");
+        match parse_loopal_control(subtype, &params["params"]) {
+            Some(c) => match self.client.send_control(&c).await {
+                Ok(_) => self.acp_out.respond(id, serde_json::json!({})).await,
+                Err(e) => {
+                    self.acp_out
+                        .respond_error(id, jsonrpc::INTERNAL_ERROR, &e.to_string())
+                        .await
+                }
+            },
+            None => {
+                self.acp_out
+                    .respond_error(
+                        id,
+                        jsonrpc::METHOD_NOT_FOUND,
+                        &format!("unsupported control subtype: {subtype}"),
+                    )
+                    .await
+            }
+        }
+    }
 }

crates/loopal-acp/src/adapter/control_command.rs

@@ -0,0 +1,109 @@
+//! Pure `subtype` → `ControlCommand` mapping for `session/control_request`.
+//! Kept separate from the IO-bound handler so the full dispatch table is
+//! unit-testable and `handle_control_request` stays thin. Every Loopal
+//! ControlCommand the runtime implements (`input_control.rs`) is reachable here.
+
+use loopal_protocol::{AgentMode, ControlCommand};
+use serde_json::Value;
+
+pub(crate) fn parse_loopal_control(subtype: &str, p: &Value) -> Option<ControlCommand> {
+    use ControlCommand as C;
+    match subtype {
+        "loopal.bgTaskKill" => p["id"]
+            .as_str()
+            .map(|s| C::BgTaskKill { id: s.to_string() }),
+        "loopal.cronDelete" => p["id"]
+            .as_str()
+            .map(|s| C::CronDelete { id: s.to_string() }),
+        "loopal.compact" => Some(C::Compact {
+            instructions: p["instructions"].as_str().map(String::from),
+        }),
+        "loopal.clear" => Some(C::Clear),
+        "loopal.suspend" => Some(C::Suspend),
+        "loopal.unsuspend" => Some(C::Unsuspend),
+        "loopal.mode" => match p["mode"].as_str()? {
+            "plan" => Some(C::ModeSwitch(AgentMode::Plan)),
+            "act" => Some(C::ModeSwitch(AgentMode::Act)),
+            _ => None,
+        },
+        "loopal.mcpStatus" => Some(C::QueryMcpStatus),
+        "loopal.mcpReconnect" => p["server"].as_str().map(|s| C::McpReconnect {
+            server: s.to_string(),
+        }),
+        "loopal.mcpDisconnect" => p["server"].as_str().map(|s| C::McpDisconnect {
+            server: s.to_string(),
+        }),
+        "loopal.rewind" => p["turn_index"].as_u64().map(|n| C::Rewind {
+            turn_index: n as usize,
+        }),
+        "loopal.resumeSession" => p["session_id"]
+            .as_str()
+            .map(|s| C::ResumeSession(s.to_string())),
+        "loopal.thinking" => p["config"]
+            .as_str()
+            .map(|s| C::ThinkingSwitch(s.to_string())),
+        "loopal.goalCreate" => p["objective"].as_str().map(|s| C::GoalCreate {
+            objective: s.to_string(),
+        }),
+        "loopal.goalPause" => Some(C::GoalUserPause),
+        "loopal.goalResume" => Some(C::GoalUserResume),
+        "loopal.goalComplete" => Some(C::GoalUserComplete),
+        "loopal.goalReopen" => Some(C::GoalUserReopen),
+        "loopal.goalClear" => Some(C::GoalClear),
+        "set_model" => p["model"].as_str().map(|s| C::ModelSwitch(s.to_string())),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn maps_object_and_session_controls() {
+        assert!(matches!(
+            parse_loopal_control("loopal.bgTaskKill", &json!({"id":"bg1"})),
+            Some(ControlCommand::BgTaskKill { id }) if id == "bg1"
+        ));
+        assert!(matches!(
+            parse_loopal_control("loopal.compact", &json!({})),
+            Some(ControlCommand::Compact { instructions: None })
+        ));
+        assert!(matches!(
+            parse_loopal_control("loopal.clear", &json!({})),
+            Some(ControlCommand::Clear)
+        ));
+        assert!(matches!(
+            parse_loopal_control("loopal.mode", &json!({"mode":"plan"})),
+            Some(ControlCommand::ModeSwitch(AgentMode::Plan))
+        ));
+    }
+
+    #[test]
+    fn maps_mcp_suspend_goal() {
+        assert!(matches!(
+            parse_loopal_control("loopal.mcpReconnect", &json!({"server":"fs"})),
+            Some(ControlCommand::McpReconnect { server }) if server == "fs"
+        ));
+        assert!(matches!(
+            parse_loopal_control("loopal.suspend", &json!({})),
+            Some(ControlCommand::Suspend)
+        ));
+        assert!(matches!(
+            parse_loopal_control("loopal.goalCreate", &json!({"objective":"ship"})),
+            Some(ControlCommand::GoalCreate { objective }) if objective == "ship"
+        ));
+        assert!(matches!(
+            parse_loopal_control("loopal.goalClear", &json!({})),
+            Some(ControlCommand::GoalClear)
+        ));
+    }
+
+    #[test]
+    fn rejects_unknown_and_missing_fields() {
+        assert!(parse_loopal_control("loopal.bogus", &json!({})).is_none());
+        assert!(parse_loopal_control("loopal.bgTaskKill", &json!({})).is_none());
+        assert!(parse_loopal_control("loopal.mode", &json!({"mode":"bogus"})).is_none());
+    }
+}

crates/loopal-acp/src/adapter/events.rs

@@ -75,6 +75,19 @@ impl AcpAdapter {
                 }
             }
         }
+        // Mode dual-emit: translate_event already sent the standard ACP
+        // CurrentModeUpdate (for generic clients). Additionally surface
+        // `_loopal/mode` so the Loopal console status bar reads mode from the
+        // same loopalSession mirror as thinking/model — AgentsMesh does not
+        // consume the ACP session-mode channel.
+        if let AgentEventPayload::ModeChanged { mode } = &event.payload {
+            let (method, params) = crate::translate::ext::ext_notification(
+                session_id,
+                "mode",
+                serde_json::json!({ "mode": mode }),
+            );
+            self.acp_out.notify(&method, params).await;
+        }
         None
     }
 

crates/loopal-acp/src/adapter/lifecycle.rs

@@ -7,12 +7,19 @@ use crate::adapter::AcpAdapter;
 use crate::types::make_init_response;
 
 impl AcpAdapter {
-    /// Handle `initialize` — return agent capabilities and info.
+    /// Handle `initialize` — return agent capabilities and info. Advertises
+    /// the AgentsMesh `controlRequest` extension so the runner routes Loopal
+    /// control-panel actions (bg-task kill / cron delete) via
+    /// `session/control_request`.
     pub(crate) async fn handle_initialize(&self, id: i64, _params: Value) {
-        let result = make_init_response();
-        self.acp_out
-            .respond(id, serde_json::to_value(result).unwrap_or_default())
-            .await;
+        let mut result = serde_json::to_value(make_init_response()).unwrap_or_default();
+        if let Some(obj) = result.as_object_mut() {
+            obj.insert(
+                "agentsmeshExtensions".into(),
+                serde_json::json!({ "controlRequest": true }),
+            );
+        }
+        self.acp_out.respond(id, result).await;
         info!("ACP initialized");
     }
 

crates/loopal-acp/src/adapter/mod.rs

@@ -1,11 +1,13 @@
 //! ACP adapter — bridges ACP (session/*) with Hub via UiSession.
 
 mod control;
+mod control_command;
 mod events;
 mod lifecycle;
 mod permission;
 mod prompt;
 mod session;
+mod snapshot;
 
 use std::sync::Arc;
 
@@ -74,6 +76,7 @@ impl AcpAdapter {
             "session/close" => self.handle_close(id, params).await,
             "session/set_mode" => self.handle_set_mode(id, params).await,
             "session/set_config_option" => self.handle_set_config_option(id, params).await,
+            "session/control_request" => self.handle_control_request(id, params).await,
             "session/load" => {
                 self.acp_out
                     .respond_error(

crates/loopal-acp/src/adapter/session.rs

@@ -8,18 +8,21 @@ use crate::jsonrpc;
 use crate::types::make_new_session_response;
 
 impl AcpAdapter {
-    /// Handle `session/new` — assign session ID, drain bootstrap events.
+    /// Handle `session/new` — assign session ID, drain bootstrap events, then
+    /// replay the agent's current view/snapshot as `_loopal/*` events so the
+    /// control panel cold-starts with pre-existing cron / task / bg-shell state.
     pub(crate) async fn handle_new_session(&self, id: i64, _params: Value) {
         let sid = uuid::Uuid::new_v4().to_string();
         *self.session_id.lock().await = Some(sid.clone());
         self.acp_out
             .respond(
                 id,
-                serde_json::to_value(make_new_session_response(sid)).unwrap_or_default(),
+                serde_json::to_value(make_new_session_response(sid.clone())).unwrap_or_default(),
             )
             .await;
         info!("ACP session created");
         self.drain_bootstrap_events().await;
+        self.replay_loopal_snapshot(&sid).await;
     }
 
     /// Handle `session/list` → HubClient.list_agents().