feat(cp): land Agent Control Plane v1 on main

Cargo.toml

@@ -9,10 +9,17 @@ edition = "2021"
 license = "MIT"
 
 [dependencies]
-openab-core = { path = "crates/openab-core", default-features = false }
-openab-gateway = { path = "crates/openab-gateway", default-features = false, optional = true }
 # OAB MCP Facade: in-process loopback Streamable HTTP MCP server, enabled by
-# `[mcp]` in config.toml (OAB MCP Adapter ADR §6.2).
+# `[mcp]` in config.toml (OAB MCP Adapter ADR §6.2). The `acp-mcp` core feature
+# is enabled UNCONDITIONALLY here — not via the `acp` feature — because the
+# agent control-plane facade (`src/cp_tools.rs`) links `openab_core::acp_mcp`
+# and `openab_core::control_plane` in every build, including a default
+# Discord/Slack build without any gateway. Enabling it on the dependency once,
+# rather than re-adding `openab-core/acp-mcp` from the `acp` feature, keeps a
+# single source of truth (the `acp` feature adding it again was redundant and
+# could drift).
+openab-core = { path = "crates/openab-core", default-features = false, features = ["acp-mcp"] }
+openab-gateway = { path = "crates/openab-gateway", default-features = false, optional = true }
 openab-mcp = { path = "crates/openab-mcp" }
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1"
@@ -51,8 +58,22 @@ feishu = ["dep:openab-gateway", "dep:axum", "openab-gateway/feishu"]
 googlechat = ["dep:openab-gateway", "dep:axum", "openab-gateway/googlechat"]
 wecom = ["dep:openab-gateway", "dep:axum", "openab-gateway/wecom"]
 teams = ["dep:openab-gateway", "dep:axum", "openab-gateway/teams"]
-acp = ["dep:openab-gateway", "dep:axum", "openab-gateway/acp", "openab-core/acp-mcp"]
+# `openab-core/acp-mcp` is NOT re-listed here: it is enabled unconditionally on
+# the `openab-core` dependency above (the CP facade links it in every build), so
+# adding it to this feature would be redundant.
+acp = ["dep:openab-gateway", "dep:axum", "openab-gateway/acp"]
 lineworks = ["dep:openab-gateway", "dep:axum", "openab-gateway/lineworks"]
 
 [dev-dependencies]
 tempfile = "3.27.0"
+# `test-util` unlocks paused-time control (`tokio::time::pause`/`advance` and
+# `#[tokio::test(start_paused = true)]`) used by the shared spawn-timeout test,
+# which must not wait real seconds for the `deadline_secs + 5` bound.
+tokio = { version = "1", features = ["full", "test-util"] }
+# rmcp 1.7's Streamable-HTTP client transport is generic over a client that
+# implements its `StreamableHttpClient` trait, whose blanket impl is for
+# reqwest 0.13's `Client` (the runtime crate holds the same pin). The
+# cp_tools real-HTTP facade test drives that client, so the test build — and
+# only the test build — needs reqwest 0.13 to construct one with a custom
+# `Authorization` header.
+reqwest013 = { package = "reqwest", version = "0.13", default-features = false, features = ["rustls", "json"] }

README.md

@@ -58,7 +58,7 @@ webhook platforms and `WS/webhook` for Feishu/Lark.
 - **@mention trigger** — mention the bot in an allowed channel to start a conversation
 - **Thread-based multi-turn** — auto-creates threads; no @mention needed for follow-ups
 - **Multi-agent collaboration** — bot-to-bot messaging for coordinated workflows ([docs/multi-agent.md](docs/multi-agent.md))
-- **Agent control plane (preview)** — standalone `openab-cp` service for direct agent-to-agent delegation over WebSocket, with identity-bound registration and CP-authoritative policy; runtime client and facade land in follow-up releases ([docs/control-plane.md](docs/control-plane.md))
+- **Agent control plane (preview)** — standalone `openab-cp` routes direct agent-to-agent delegation; runtimes register outbound, workers serve through their ACP pool, and primaries get four direct MCP tools plus the `openab agent` CLI over an owner-only local socket ([docs/control-plane.md](docs/control-plane.md))
 - **Agent-controlled reply-to** — agents choose which message to reply to via `[[reply_to:id]]` directive, enabling clear conversation threads in multi-bot channels ([docs/output-directives.md](docs/output-directives.md))
 - **Edit-streaming** — live-updates the Discord message every 1.5s as tokens arrive
 - **Emoji status reactions** — 👀→🤔→🔥/👨‍💻/⚡→👍+random mood face

README.zh-TW.md

@@ -58,7 +58,7 @@ platforms 使用 `webhook/API`，Feishu/Lark 則使用 `WS/webhook`。
 - **@mention 觸發** — 在允許的頻道中 mention bot，即可開始對話
 - **以討論串進行多輪對話** — 自動建立討論串；後續訊息不需再次 @mention
 - **多 agent 協作** — 支援 bot-to-bot 訊息，實現協調式工作流程（[docs/multi-agent.md](docs/multi-agent.md)）
-- **Agent control plane（預覽）** — 獨立的 `openab-cp` 服務，讓 agent 之間可透過 WebSocket 直接委派任務，具備綁定身分的註冊機制與由 CP 主導的政策控管；runtime client 與 facade 將於後續版本推出（[docs/control-plane.md](docs/control-plane.md)）
+- **Agent control plane（預覽）** — 獨立的 `openab-cp` 負責 agent 間的直接委派；runtime 主動註冊、worker 透過 ACP session pool 執行工作，primary 則取得四個 MCP tools 與透過 owner-only 本機 socket 使用的 `openab agent` CLI（[docs/control-plane.md](docs/control-plane.md)）
 - **由 agent 控制回覆對象** — agent 可透過 `[[reply_to:id]]` 指令選擇要回覆的訊息，讓多 bot 頻道中的對話脈絡更清楚（[docs/output-directives.md](docs/output-directives.md)）
 - **編輯式串流輸出** — token 產生時每 1.5 秒即時更新 Discord 訊息
 - **Emoji 狀態反應** — 👀→🤔→🔥/👨‍💻/⚡→👍+隨機情緒表情

crates/openab-core/Cargo.toml

@@ -54,12 +54,23 @@ http = { version = "1", optional = true }
 # axum listener) was removed with the per-session proxy.
 rmcp = { version = "1.7", default-features = false, optional = true }
 tokio-util = { version = "0.7", optional = true }
+# Control-plane wire types ONLY (`openab_cp::proto`). `default-features = false`
+# keeps the CP's `server` feature — axum, its own tokio flavour, the registry —
+# out of the runtime build; the runtime is a WebSocket *client* of the CP.
+openab-cp = { path = "../openab-cp", default-features = false }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["test-util"] }
+# The control-plane client integration test drives the REAL CP server
+# in-process (ephemeral loopback port), so the test build — and only the test
+# build — needs the `server` feature.
+openab-cp = { path = "../openab-cp", features = ["server"] }
+# ... and `axum::serve` to actually bind it. Test-only: the runtime is a
+# WebSocket *client* of the CP and links no HTTP server.
+axum = "0.8"
 
 [features]
 default = ["discord", "slack", "secrets-aws", "agentcore", "config-s3"]

crates/openab-core/src/acp/pool.rs

@@ -179,13 +179,29 @@ fn purge_session_entries(state: &mut PoolState, key: &str) {
     state.pgids.remove(key);
     state.suspended.remove(key);
     state.persisted.remove(key);
-    // Do NOT remove the creating gate: it is concurrency control, not session
-    // state. Removing it while a holder still owns the old gate Arc would let
-    // a concurrent get_or_create mint a fresh gate and run two creations for
-    // the same key.
+    drop_idle_gate(&mut state.creating, key);
     state.session_workdirs.remove(key);
 }
 
+/// Remove `key`'s creating gate ONLY when the map holds the last reference.
+///
+/// The gate is concurrency control, not session state: removing it while a
+/// `get_or_create` still owns a clone would let a concurrent caller mint a
+/// fresh gate and run two creations for the same key. The strong count is
+/// authoritative here because every clone is taken under the same state
+/// write lock the caller holds (`get_or_insert_gate`), so a count of one
+/// means no creation is in progress or about to be.
+///
+/// Without this, keys that are used exactly once — the control-plane
+/// executor's admission-scoped `control-plane:<sha256>` sessions — would each
+/// leave a permanent `String` + `Arc<Mutex<()>>` behind for the process
+/// lifetime, a map `max_sessions` does not bound (F5).
+fn drop_idle_gate(map: &mut HashMap<String, Arc<Mutex<()>>>, key: &str) {
+    if map.get(key).is_some_and(|g| Arc::strong_count(g) == 1) {
+        map.remove(key);
+    }
+}
+
 /// Escalating kill for a hung agent's process group: wait 10s after the
 /// session/cancel attempt, SIGTERM, wait 2s, SIGKILL. Mirrors
 /// `AcpConnection::kill_process_group`, which cannot run here because the
@@ -856,6 +872,35 @@ impl SessionPool {
         }
     }
 
+    /// Drop a session and all its bookkeeping WITHOUT sending
+    /// `session/cancel` first. Returns `true` when there was an active
+    /// connection to drop.
+    ///
+    /// [`Self::reset_session`] is the same teardown *plus* a cancel and an
+    /// error when the session is unknown, which suits the interactive
+    /// `/reset` it serves. Non-interactive owners of single-use sessions —
+    /// the control-plane executor, which runs one fresh session per
+    /// delegation — need neither: after a completed turn there is nothing to
+    /// cancel, and after a cancelled one the cancel has already been sent.
+    /// Calling `reset_session` there would emit a spurious `session/cancel`
+    /// at the agent and log an error for the benign already-gone case.
+    ///
+    /// The ACP process exits once the last `Arc` to its connection drops, so
+    /// removing the map entry is what reclaims the pool slot.
+    pub async fn discard_session(&self, thread_id: &str) -> bool {
+        let mut state = self.state.write().await;
+        let had_active = state.active.remove(thread_id).is_some();
+        purge_session_entries(&mut state, thread_id);
+        #[cfg(feature = "acp-mcp")]
+        revoke_facade_token_for_key(&mut state, thread_id, self.session_registrar.as_ref());
+        self.save_mapping(&state.persisted);
+        self.save_meta(&state.session_workdirs);
+        if had_active {
+            info!(thread_id = %crate::redact::redact_session_ids(thread_id), "session discarded");
+        }
+        had_active
+    }
+
     pub async fn cleanup_idle(&self, ttl_secs: u64) {
         let cutoff = Instant::now() - std::time::Duration::from_secs(ttl_secs);
         let hung_threshold = std::time::Duration::from_secs(self.hung_threshold_secs);
@@ -1349,6 +1394,8 @@ mod tests {
             creating: HashMap::from([("hung".to_string(), Arc::new(Mutex::new(())))]),
             session_workdirs: HashMap::from([("hung".to_string(), "/tmp/ws".to_string())]),
         };
+        // A get_or_create in flight for "hung" holds a clone of its gate.
+        let held_gate = Arc::clone(state.creating.get("hung").unwrap());
 
         purge_session_entries(&mut state, "hung");
 
@@ -1359,9 +1406,11 @@ mod tests {
         assert!(!state.suspended.contains_key("hung"));
         assert!(!state.persisted.contains_key("hung"));
         assert!(!state.session_workdirs.contains_key("hung"));
-        // The creating gate is concurrency control, not session state: it must
-        // survive so an in-flight get_or_create holder stays serialized.
+        // The creating gate is concurrency control, not session state: while a
+        // get_or_create holder still owns a clone it must survive so that
+        // holder stays serialized against the next creation.
         assert!(state.creating.contains_key("hung"));
+        assert!(Arc::ptr_eq(&held_gate, state.creating.get("hung").unwrap()));
         assert_eq!(state.pgids.get("other"), Some(&5678));
         // Other keys survive untouched.
         assert_eq!(
@@ -1375,6 +1424,37 @@ mod tests {
         assert!(state.activity.contains_key("other"));
     }
 
+    /// F5: single-use keys (one fresh session per control-plane delegation)
+    /// must not leave a gate behind. Mirrors the pool's own sequence — the
+    /// gate is minted under the state lock and the clone released when the
+    /// creation finishes — then purges, and asserts the map does not grow.
+    #[test]
+    fn purge_drops_a_gate_nobody_holds_so_single_use_keys_stay_bounded() {
+        let mut state = empty_pool_state();
+        for i in 0..1_000 {
+            let key = format!("control-plane:{i:064x}");
+            let gate = get_or_insert_gate(&mut state.creating, &key);
+            // ... creation happens under `gate.lock()` and then the caller
+            // returns, dropping its clone ...
+            drop(gate);
+            purge_session_entries(&mut state, &key);
+            assert!(
+                !state.creating.contains_key(&key),
+                "an unheld gate must be dropped with the rest of the key's entries"
+            );
+        }
+        assert!(state.creating.is_empty(), "no gate may outlive its single-use key");
+
+        // The guard the purge relies on: a clone still held keeps the gate.
+        let key = "control-plane:held";
+        let held = get_or_insert_gate(&mut state.creating, key);
+        purge_session_entries(&mut state, key);
+        assert!(state.creating.contains_key(key));
+        drop(held);
+        purge_session_entries(&mut state, key);
+        assert!(state.creating.is_empty());
+    }
+
     #[test]
     fn persisted_mapping_can_include_active_and_suspended_sessions() {
         let persisted = HashMap::from([