feat(acp): browser control via MCP-over-ACP + stdio bridge (Phase 2)

.dockerignore

@@ -0,0 +1,4 @@
+target/
+.git/
+data/
+*.tgz

.github/workflows/ci.yml

@@ -56,6 +56,25 @@ jobs:
       # `-p openab-gateway` to avoid the workspace hooks::tests parallel flake.
       - name: cargo test (acp gateway)
         run: cargo test -p openab-gateway --features acp
+      # Same default-features gap on the core side: `mcp_proxy` is the only `acp-mcp`-gated module,
+      # so its tunnel-trait, tools-cache and bridge-dispatch tests never compile above either.
+      # Filtered to `mcp_proxy::` for the same reason the gateway step is scoped to one package —
+      # it keeps the flaky hooks::tests out of this job.
+      - name: cargo test (acp-mcp core)
+        # Quoted: the trailing `::` of the filter would otherwise read as a YAML mapping
+        # indicator and make the whole workflow file unparseable.
+        run: "cargo test -p openab-core --features acp-mcp mcp_proxy::"
+      # The pool's facade-session tests are `acp-mcp`-gated too but live outside `mcp_proxy::`, so
+      # the filter above compiled them and then selected them away. Naming the module runs them
+      # while still leaving the flaky hooks::tests out.
+      - name: cargo test (acp-mcp pool)
+        # Quoted for the same reason as above — a trailing `::` reads as a YAML mapping indicator.
+        run: "cargo test -p openab-core --features acp-mcp acp::pool::"
+      # And the root package's own tests — the ACP-tunnel capability source (browser_source.rs) and
+      # are `acp`-gated, so `--workspace` above skips
+      # them too. This is where the source's routing and trust-gate coverage lives.
+      - name: cargo test (acp root)
+        run: cargo test --features acp
       - name: cargo build (unified)
         run: cargo build --features unified
 

Cargo.toml

@@ -51,7 +51,7 @@ 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"]
+acp = ["dep:openab-gateway", "dep:axum", "openab-gateway/acp", "openab-core/acp-mcp"]
 
 [dev-dependencies]
 tempfile = "3.27.0"

crates/openab-core/Cargo.toml

@@ -49,6 +49,11 @@ aws-credential-types = { version = "1", optional = true }
 urlencoding = { version = "2", optional = true }
 hex = { version = "0.4", optional = true }
 http = { version = "1", optional = true }
+# Browser tool definitions for the facade's capability source. Only `rmcp::model` is used now —
+# the loopback MCP server that needed the server + streamable-http transport features (and its own
+# 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 }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
@@ -65,3 +70,6 @@ config-s3 = ["dep:aws-sdk-s3", "dep:aws-config"]
 pre-seed = ["dep:aws-sdk-s3", "dep:aws-config", "dep:zip", "dep:hex", "dep:flate2", "dep:tar"]
 filestore = ["dep:aws-sdk-s3", "dep:aws-config"]
 agentcore = ["dep:aws-config", "dep:aws-sigv4", "dep:aws-credential-types", "dep:urlencoding", "dep:hex", "dep:http", "dep:rustls", "dep:tokio-rustls", "dep:webpki-roots"]
+# Core-side wiring for MCP-over-ACP browser control (enabled by the root `acp`): the browser tool
+# definitions and the agent's facade MCP config. Core hosts no MCP server of its own any more.
+acp-mcp = ["dep:rmcp", "dep:tokio-util"]

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

@@ -189,6 +189,14 @@ pub struct AcpConnection {
     pub session_reset: bool,
     _reader_handle: JoinHandle<()>,
     _stderr_handle: Option<JoinHandle<()>>,
+    /// Revokes this session's facade token when the connection is dropped, on any evict path.
+    /// Held only for its `Drop` side effect (never read).
+    ///
+    /// It used to cancel a per-session MCP proxy server; that server is gone and the guard now
+    /// carries the minted token instead.
+    #[cfg(feature = "acp-mcp")]
+    #[allow(dead_code)]
+    facade_token_guard: Option<tokio_util::sync::DropGuard>,
 }
 
 /// Build the final set of env vars for the agent subprocess.
@@ -485,9 +493,17 @@ impl AcpConnection {
             session_reset: false,
             _reader_handle: reader_handle,
             _stderr_handle: stderr_handle,
+            #[cfg(feature = "acp-mcp")]
+            facade_token_guard: None,
         })
     }
 
+    /// Attach the guard that revokes this session's facade token when the connection drops.
+    #[cfg(feature = "acp-mcp")]
+    pub fn set_facade_token_guard(&mut self, guard: Option<tokio_util::sync::DropGuard>) {
+        self.facade_token_guard = guard;
+    }
+
     fn next_id(&self) -> u64 {
         self.next_id.fetch_add(1, Ordering::Relaxed)
     }

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

@@ -53,6 +53,10 @@ pub struct SessionPool {
     mapping_path: PathBuf,
     meta_path: PathBuf,
     default_config_options: HashMap<String, String>,
+    #[cfg(feature = "acp-mcp")]
+    session_registrar: Option<Arc<dyn crate::mcp_proxy::SessionTokenRegistrar>>,
+    #[cfg(feature = "acp-mcp")]
+    facade_url: Option<String>,
 }
 
 type CancelHandle = (Arc<tokio::sync::Mutex<tokio::process::ChildStdin>>, String);
@@ -102,13 +106,46 @@ fn better_candidate(current_oldest: Option<Instant>, candidate_last_active: Inst
     }
 }
 
-/// Remove every non-`active` pool entry for `key`, reset-style.
+/// Prepare facade browser capabilities for one session: write the agent's facade MCP entry, and
+/// mint its session token **only if that write succeeded**.
 ///
-/// Hung eviction must NOT leave the session resumable: the old streaming task
-/// still holds an Arc clone of the connection, so the agent process may be
-/// alive and mid-turn. If the session id stayed in `suspended`/`persisted`,
-/// the next message would `session/load` the same session while the old
-/// process still owns an in-flight turn. Mirror `reset_session` instead.
+/// The token is useless without the config. The entry is what points the agent at the facade and
+/// carries `Authorization: Bearer ${OPENAB_SESSION_TOKEN}`; with no entry the agent never reaches
+/// the facade and never presents the token. Minting regardless would register a live credential
+/// for a session that cannot use it and leave it valid until eviction, while the failure showed up
+/// only as a warning. Returning `None` keeps the session running without browser capabilities,
+/// which is the honest description of what actually happened.
+#[cfg(feature = "acp-mcp")]
+async fn setup_facade_session(
+    workdir: &str,
+    facade_url: &str,
+    channel_id: &str,
+    registrar: &Arc<dyn crate::mcp_proxy::SessionTokenRegistrar>,
+) -> Option<String> {
+    match crate::mcp_proxy::write_facade_mcp_config(workdir, facade_url).await {
+        Ok(()) => Some(registrar.mint(channel_id)),
+        Err(e) => {
+            tracing::error!(
+                workdir, error = %e,
+                "facade mcp config write failed — starting this session WITHOUT browser \
+                 capabilities and not minting a session token that could never be presented"
+            );
+            None
+        }
+    }
+}
+
+/// Remove every non-`active` pool entry for `key`.
+///
+/// The single implementation for both hung eviction and [`SessionPool::reset_session`]; the latter
+/// removes `active` itself and then calls this. It used to be a second copy of the same list, which
+/// is how the two could drift — and the line most likely to be lost from a copy is the one below
+/// about the creating gate, because it says *not* to remove something.
+///
+/// Hung eviction must NOT leave the session resumable: the old streaming task still holds an Arc
+/// clone of the connection, so the agent process may be alive and mid-turn. If the session id
+/// stayed in `suspended`/`persisted`, the next message would `session/load` the same session while
+/// the old process still owns an in-flight turn.
 fn purge_session_entries(state: &mut PoolState, key: &str) {
     state.cancel_handles.remove(key);
     state.activity.remove(key);
@@ -197,9 +234,29 @@ impl SessionPool {
             mapping_path,
             meta_path,
             default_config_options,
+            #[cfg(feature = "acp-mcp")]
+            session_registrar: None,
+            #[cfg(feature = "acp-mcp")]
+            facade_url: None,
         }
     }
 
+    /// Wire the facade session-token registrar + facade URL (Facade mode,
+    /// set by the root when `[mcp]` is running). With both present, browser
+    /// capabilities route through the facade: the pool mints one token per
+    /// session, injects it as `OPENAB_SESSION_TOKEN` in the agent process
+    /// env, and writes the static facade MCP entry once per workdir.
+    #[cfg(feature = "acp-mcp")]
+    pub fn with_facade_sessions(
+        mut self,
+        registrar: Option<Arc<dyn crate::mcp_proxy::SessionTokenRegistrar>>,
+        facade_url: Option<String>,
+    ) -> Self {
+        self.session_registrar = registrar;
+        self.facade_url = facade_url;
+        self
+    }
+
     fn load_mapping(path: &Path) -> HashMap<String, String> {
         match std::fs::read_to_string(path) {
             Ok(data) => serde_json::from_str(&data).unwrap_or_else(|e| {
@@ -347,13 +404,70 @@ impl SessionPool {
             self.config.working_dir.clone()
         };
 
+        // Browser capabilities for an `acp:` session come from the OAB MCP Facade and nowhere
+        // else: mint a per-session token (it rides the agent spawn below as OPENAB_SESSION_TOKEN)
+        // and write the static facade entry before the agent boots. The returned guard revokes
+        // that token when this connection is dropped, on any evict path.
+        //
+        // There is no transport fallback. Without `[mcp]` the root wires no registrar, and the
+        // session simply starts without browser capabilities — which is the honest outcome and is
+        // reported once at startup rather than being silently substituted per session.
+        #[cfg(feature = "acp-mcp")]
+        let mut session_token: Option<String> = None;
+        #[cfg(feature = "acp-mcp")]
+        let facade_token_guard: Option<tokio_util::sync::DropGuard> = match (
+            thread_id.strip_prefix("acp:"),
+            self.session_registrar.as_ref(),
+            self.facade_url.as_ref(),
+        ) {
+            (Some(channel_id), Some(registrar), Some(facade_url)) => {
+                match setup_facade_session(&effective_workdir, facade_url, channel_id, registrar)
+                    .await
+                {
+                    Some(token) => {
+                        session_token = Some(token.clone());
+                        info!(thread_id, "session token minted for facade browser capabilities");
+                        // The guard carries the TOKEN it minted, not the channel. A replaced
+                        // session's teardown runs after its successor has already re-minted for
+                        // the same channel, so revoking by channel would strip the live token and
+                        // silently cut the new agent off from the facade; revoking this exact
+                        // token is a no-op by then (R1).
+                        let ct = tokio_util::sync::CancellationToken::new();
+                        let child = ct.child_token();
+                        let registrar = registrar.clone();
+                        tokio::spawn(async move {
+                            child.cancelled().await;
+                            registrar.revoke(&token);
+                        });
+                        Some(ct.drop_guard())
+                    }
+                    // No config, so no token and no revoke guard to arm. The session still
+                    // starts — it simply has no browser capabilities.
+                    None => None,
+                }
+            }
+            _ => None,
+        };
+
         // Build the replacement connection outside the state lock so one stuck
         // initialization does not block all unrelated sessions.
+        #[cfg(feature = "acp-mcp")]
+        let spawn_env: std::collections::HashMap<String, String> = {
+            let mut env = self.config.env.clone();
+            if let Some(tok) = &session_token {
+                // The static facade MCP entry references ${OPENAB_SESSION_TOKEN};
+                // the value lives only in this agent process's environment.
+                env.insert("OPENAB_SESSION_TOKEN".to_string(), tok.clone());
+            }
+            env
+        };
+        #[cfg(not(feature = "acp-mcp"))]
+        let spawn_env = self.config.env.clone();
         let mut new_conn = AcpConnection::spawn(
             &self.config.command,
             &self.config.args,
             &effective_workdir,
-            &self.config.env,
+            &spawn_env,
             &self.config.inherit_env,
         )
         .await?;
@@ -423,6 +537,8 @@ impl SessionPool {
         let activity_handle = new_conn.activity_handle();
         let child_pgid = new_conn.child_pgid();
         let cancel_session_id = new_conn.acp_session_id.clone().unwrap_or_default();
+        #[cfg(feature = "acp-mcp")]
+        new_conn.set_facade_token_guard(facade_token_guard);
         let new_conn = Arc::new(Mutex::new(new_conn));
 
         let mut state = self.state.write().await;
@@ -636,13 +752,11 @@ impl SessionPool {
 
         let mut state = self.state.write().await;
         let had_active = state.active.remove(thread_id).is_some();
-        state.cancel_handles.remove(thread_id);
-        state.activity.remove(thread_id);
-        state.pgids.remove(thread_id);
-        state.suspended.remove(thread_id);
-        state.persisted.remove(thread_id);
-        state.creating.remove(thread_id);
-        state.session_workdirs.remove(thread_id);
+        // Everything else a reset clears is exactly what hung eviction clears, including the rule
+        // that the creating gate survives. Call the one implementation rather than keeping a second
+        // copy of the list: the copies are what let the two drift, and the gate rule is precisely
+        // the kind of line that gets dropped from a duplicate without anyone noticing.
+        purge_session_entries(&mut state, thread_id);
         self.save_mapping(&state.persisted);
         self.save_meta(&state.session_workdirs);
         if had_active {
@@ -818,6 +932,68 @@ mod tests {
     use tokio::sync::Mutex;
     use tokio::time::Instant;
 
+    /// Registrar double that records every mint, so a test can assert one never happened.
+    #[cfg(feature = "acp-mcp")]
+    #[derive(Default)]
+    struct CountingRegistrar {
+        minted: std::sync::Mutex<Vec<String>>,
+    }
+
+    #[cfg(feature = "acp-mcp")]
+    impl crate::mcp_proxy::SessionTokenRegistrar for CountingRegistrar {
+        fn mint(&self, channel_id: &str) -> String {
+            self.minted.lock().unwrap().push(channel_id.to_string());
+            "token-xyz".to_string()
+        }
+        fn revoke(&self, _token: &str) {}
+    }
+
+    /// A failed facade config write must not mint a token. The agent has no `openab` entry, so it
+    /// can never present one; minting anyway would leave a live credential registered for a
+    /// session that cannot use it until eviction.
+    #[cfg(feature = "acp-mcp")]
+    #[tokio::test]
+    async fn no_token_is_minted_when_the_facade_config_write_fails() {
+        let dir = tempfile::tempdir().unwrap();
+        // Make `<workdir>/.cursor` a FILE, so create_dir_all inside the writer fails.
+        std::fs::write(dir.path().join(".cursor"), b"not a directory").unwrap();
+
+        let counting = Arc::new(CountingRegistrar::default());
+        let registrar: Arc<dyn crate::mcp_proxy::SessionTokenRegistrar> = counting.clone();
+        let token = super::setup_facade_session(
+            dir.path().to_str().unwrap(),
+            "http://127.0.0.1:8848/mcp",
+            "acp_x",
+            &registrar,
+        )
+        .await;
+
+        assert!(token.is_none(), "a failed config write must yield no token");
+        assert!(
+            counting.minted.lock().unwrap().is_empty(),
+            "the registrar must never be asked to mint when the config could not be written"
+        );
+    }
+
+    /// The happy path still mints exactly once, for the right channel.
+    #[cfg(feature = "acp-mcp")]
+    #[tokio::test]
+    async fn a_successful_facade_config_write_mints_one_token() {
+        let dir = tempfile::tempdir().unwrap();
+        let counting = Arc::new(CountingRegistrar::default());
+        let registrar: Arc<dyn crate::mcp_proxy::SessionTokenRegistrar> = counting.clone();
+        let token = super::setup_facade_session(
+            dir.path().to_str().unwrap(),
+            "http://127.0.0.1:8848/mcp",
+            "acp_x",
+            &registrar,
+        )
+        .await;
+
+        assert_eq!(token.as_deref(), Some("token-xyz"));
+        assert_eq!(counting.minted.lock().unwrap().as_slice(), ["acp_x"]);
+    }
+
     #[test]
     fn remove_if_same_handle_removes_matching_entry() {
         let expected = Arc::new(Mutex::new(1_u8));