feat(acp): dispatch dynamic tools through websocket

crates/aionui-ai-agent/src/factory/acp.rs

@@ -111,23 +111,23 @@ pub(super) async fn build(
         }
     }
 
-    let params = Arc::new(
-        assemble_acp_params(
-            ctx.conversation_id.clone(),
-            WorkspaceInfo {
-                path: ctx.workspace,
-                is_custom: ctx.is_custom_workspace,
-            },
-            meta,
-            command_spec,
-            config,
-            session_mcp_servers,
-            session_snapshot,
-            deps.data_dir.clone(),
-            deps.dump_prompts,
-        )
-        .await,
-    );
+    let mut params = assemble_acp_params(
+        ctx.conversation_id.clone(),
+        WorkspaceInfo {
+            path: ctx.workspace,
+            is_custom: ctx.is_custom_workspace,
+        },
+        meta,
+        command_spec,
+        config,
+        session_mcp_servers,
+        session_snapshot,
+        deps.data_dir.clone(),
+        deps.dump_prompts,
+    )
+    .await;
+    params.dynamic_tool_session = deps.dynamic_tool_registry.session_for(&ctx.conversation_id);
+    let params = Arc::new(params);
 
     let skill_mgr = deps.skill_manager.clone();
     let catalog_tx = deps.agent_registry.catalog_sender();

crates/aionui-ai-agent/src/factory/acp_assembler.rs

@@ -1,3 +1,4 @@
+use crate::protocol::dynamic_tools::DynamicToolSession;
 use crate::shared_kernel::PersistedSessionState;
 use agent_client_protocol::schema::{EnvVariable, McpServer, McpServerStdio, NewSessionRequest};
 use aionui_api_types::AgentMetadata;
@@ -34,16 +35,24 @@ pub struct AcpSessionParams {
     pub data_dir: PathBuf,
     /// Whether prompt diagnostics should be dumped under `data_dir/prompt-dumps`.
     pub dump_prompts: bool,
+    /// Registration captured before the ACP session starts. The handle binds
+    /// the CLI-assigned thread id after session/new or session/load.
+    pub dynamic_tool_session: Option<DynamicToolSession>,
 }
 
 impl AcpSessionParams {
     /// Build a `NewSessionRequest` using the pre-computed MCP servers.
     pub fn new_session_request(&self) -> NewSessionRequest {
         let req = NewSessionRequest::new(&self.workspace.path);
-        if self.mcp_servers.is_empty() {
+        let req = if self.mcp_servers.is_empty() {
             req
         } else {
             req.mcp_servers(self.mcp_servers.clone())
+        };
+        if let Some(session) = &self.dynamic_tool_session {
+            req.meta(session.metadata())
+        } else {
+            req
         }
     }
 }
@@ -83,6 +92,7 @@ pub async fn assemble_acp_params(
         session_snapshot,
         data_dir,
         dump_prompts,
+        dynamic_tool_session: None,
     }
 }
 

crates/aionui-ai-agent/src/factory/mod.rs

@@ -17,6 +17,7 @@ use crate::capability::skill_manager::AcpSkillManager;
 use crate::error::AgentError;
 use crate::factory::context::FactoryContext;
 use crate::persistence::AcpSessionSyncService;
+use crate::protocol::dynamic_tools::DynamicToolRegistry;
 use crate::registry::AgentRegistry;
 use crate::session_context::AgentSessionKind;
 use crate::task_manager::AgentFactory;
@@ -40,6 +41,8 @@ pub struct AgentFactoryDeps {
     /// inject enabled servers into `session/new` (ELECTRON-1JG fix).
     /// `None` for tests/composition paths that do not need MCP injection.
     pub mcp_server_repo: Option<Arc<dyn IMcpServerRepository>>,
+    /// WebSocket-owned dynamic tools available to ordinary ACP conversations.
+    pub dynamic_tool_registry: DynamicToolRegistry,
 }
 
 /// Build a production agent factory that dispatches to concrete agent types.

crates/aionui-ai-agent/src/lib.rs

@@ -37,6 +37,7 @@ pub use error::AgentError;
 pub use factory::{AgentFactoryDeps, build_agent_factory};
 pub use idle_scanner::{IdleCleanupCoordinator, start_idle_scanner, start_idle_scanner_with_coordinator};
 pub use persistence::AcpSessionSyncService;
+pub use protocol::dynamic_tools::{DynamicToolRegistry, DynamicToolSession};
 pub use protocol::error::AcpError;
 pub use protocol::events::AgentStreamEvent;
 pub use protocol::send_error::AgentSendError;

crates/aionui-ai-agent/src/manager/acp/agent.rs

@@ -438,7 +438,14 @@ impl AcpAgentManager {
         // 70ms in — ELECTRON-1BT), so we explicitly watch the child. If
         // it dies before init completes, surface a `StartupCrash` carrying
         // the buffered stderr instead of waiting out the timeout.
-        let connect_fut = AcpProtocol::connect(stdin, stdout, runtime.event_sender(), permission_tx, notification_tx);
+        let connect_fut = AcpProtocol::connect(
+            stdin,
+            stdout,
+            runtime.event_sender(),
+            permission_tx,
+            notification_tx,
+            params.dynamic_tool_session.clone(),
+        );
         tokio::pin!(connect_fut);
         let protocol = tokio::select! {
             biased;

crates/aionui-ai-agent/src/manager/acp/agent_session_flow.rs

@@ -39,6 +39,9 @@ impl AcpAgentManager {
         let session_response = self.protocol.new_session(req).await?;
 
         let sid = session_response.session_id.to_string();
+        if let Some(dynamic_tools) = &self.params.dynamic_tool_session {
+            dynamic_tools.bind_thread(&sid);
+        }
 
         {
             let mut session = self.session.write().await;
@@ -122,6 +125,9 @@ impl AcpAgentManager {
             options.insert("resume".into(), Value::String(session_id.to_owned()));
             claude_code.insert("options".into(), Value::Object(options));
             meta.insert("claudeCode".into(), Value::Object(claude_code));
+            if let Some(dynamic_tools) = &self.params.dynamic_tool_session {
+                meta.extend(dynamic_tools.metadata());
+            }
 
             let req = self.params.new_session_request().meta(meta);
             let new_response = match self.protocol.new_session(req).await {
@@ -132,6 +138,9 @@ impl AcpAgentManager {
                 Err(e) => return Err(e.into()),
             };
             let new_sid = new_response.session_id.to_string();
+            if let Some(dynamic_tools) = &self.params.dynamic_tool_session {
+                dynamic_tools.bind_thread(&new_sid);
+            }
 
             {
                 let mut session = self.session.write().await;
@@ -176,13 +185,19 @@ impl AcpAgentManager {
             if !self.params.mcp_servers.is_empty() {
                 load_req = load_req.mcp_servers(self.params.mcp_servers.clone());
             }
+            if let Some(dynamic_tools) = &self.params.dynamic_tool_session {
+                load_req = load_req.meta(dynamic_tools.metadata());
+            }
             let load_response = match self.protocol.load_session(load_req).await {
                 Ok(r) => r,
                 Err(e) if is_acp_session_not_found(&e) => {
                     return self.rebuild_after_acp_session_not_found(session_id, e).await;
                 }
                 Err(e) => return Err(e.into()),
             };
+            if let Some(dynamic_tools) = &self.params.dynamic_tool_session {
+                dynamic_tools.bind_thread(session_id);
+            }
 
             {
                 let mut session = self.session.write().await;
@@ -214,6 +229,9 @@ impl AcpAgentManager {
         // session/load. Seed the aggregate with the stored id and let the
         // caller prompt — matches pre-refactor behaviour.
         {
+            if let Some(dynamic_tools) = &self.params.dynamic_tool_session {
+                dynamic_tools.bind_thread(session_id);
+            }
             let mut session = self.session.write().await;
             session.set_session_id(DomainSessionId::new(session_id.to_owned()));
             self.commit_session_changes(&mut session).await;

crates/aionui-ai-agent/src/protocol/acp.rs

@@ -26,21 +26,22 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 
 use agent_client_protocol::schema::{
-    AGENT_METHOD_NAMES, AuthenticateResponse, ClientNotification, ClientRequest, CloseSessionResponse, ExtResponse,
-    ForkSessionResponse, Implementation, InitializeRequest, LoadSessionResponse, PromptResponse, ProtocolVersion,
-    RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse, ResumeSessionResponse,
-    SelectedPermissionOutcome, SessionNotification, SetSessionConfigOptionResponse, SetSessionModeResponse,
-    SetSessionModelResponse,
+    AGENT_METHOD_NAMES, AgentRequest, AuthenticateResponse, ClientNotification, ClientRequest, CloseSessionResponse,
+    ExtResponse, ForkSessionResponse, Implementation, InitializeRequest, LoadSessionResponse, PromptResponse,
+    ProtocolVersion, RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
+    ResumeSessionResponse, SelectedPermissionOutcome, SessionNotification, SetSessionConfigOptionResponse,
+    SetSessionModeResponse, SetSessionModelResponse,
 };
 use agent_client_protocol::{
-    Agent, ByteStreams, Client, ConnectionTo, Responder, on_receive_notification, on_receive_request,
+    Agent, ByteStreams, Client, ConnectionTo, Handled, Responder, on_receive_notification, on_receive_request,
 };
 use aionui_common::ErrorChain;
 use tokio::process::{ChildStdin, ChildStdout};
 use tokio::sync::{broadcast, mpsc, oneshot};
 use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
 use tracing::{debug, info, warn};
 
+use crate::protocol::dynamic_tools::{CODEX_DYNAMIC_TOOL_CALL_METHOD, DynamicToolSession, dynamic_tool_unavailable};
 use crate::protocol::error::AcpError;
 use crate::protocol::events::{self as stream_event, AgentStreamEvent};
 
@@ -137,6 +138,7 @@ impl AcpProtocol {
         event_tx: broadcast::Sender<AgentStreamEvent>,
         permission_tx: mpsc::Sender<PermissionRequest>,
         notification_tx: mpsc::Sender<SessionNotification>,
+        dynamic_tool_session: Option<DynamicToolSession>,
     ) -> Result<Self, AcpError> {
         let alive = Arc::new(AtomicBool::new(true));
         let replay_suppression = Arc::new(AtomicBool::new(false));
@@ -160,6 +162,7 @@ impl AcpProtocol {
             event_tx,
             permission_tx,
             notification_tx,
+            dynamic_tool_session,
             init_tx,
             ready_tx,
             shutdown_rx,
@@ -406,6 +409,7 @@ async fn run_sdk_background(
     event_tx: broadcast::Sender<AgentStreamEvent>,
     permission_tx: mpsc::Sender<PermissionRequest>,
     notification_tx: mpsc::Sender<SessionNotification>,
+    dynamic_tool_session: Option<DynamicToolSession>,
     init_tx: oneshot::Sender<Result<InitializeResponse, AcpError>>,
     ready_tx: oneshot::Sender<ConnectionTo<Agent>>,
     shutdown_rx: oneshot::Receiver<()>,
@@ -460,6 +464,37 @@ async fn run_sdk_background(
             },
             on_receive_request!(),
         )
+        .on_receive_request(
+            {
+                async move |request: AgentRequest, responder, _cx| {
+                    let AgentRequest::ExtMethodRequest(extension) = request else {
+                        return Ok(Handled::No {
+                            message: (request, responder),
+                            retry: false,
+                        });
+                    };
+                    if extension.method.as_ref() != CODEX_DYNAMIC_TOOL_CALL_METHOD {
+                        return Ok(Handled::No {
+                            message: (AgentRequest::ExtMethodRequest(extension), responder),
+                            retry: false,
+                        });
+                    }
+
+                    let response = match (
+                        dynamic_tool_session.as_ref(),
+                        serde_json::from_str(extension.params.get()),
+                    ) {
+                        (Some(session), Ok(params)) => session.dispatch(params).await,
+                        _ => dynamic_tool_unavailable(),
+                    };
+                    let response = serde_json::to_value(response)
+                        .unwrap_or_else(|_| serde_json::json!({"success": false, "contentItems": []}));
+                    responder.respond(response)?;
+                    Ok(Handled::Yes)
+                }
+            },
+            on_receive_request!(),
+        )
         .connect_with(transport, async move |connection: ConnectionTo<Agent>| {
             // Step 1 — initialize handshake. main_fn is the canonical place
             // to call `block_task` (see SDK `connect_with` doc example).

crates/aionui-ai-agent/src/protocol/custom_agent_probe.rs

@@ -206,7 +206,7 @@ async fn run_handshake(proc: &CliAgentProcess) -> ProbeOutcome {
     // immediately with a non-zero status; without this race the
     // `AcpProtocol::connect` call would block on its internal 30 s
     // timeout waiting for an `initialize` reply that will never arrive.
-    let connect = AcpProtocol::connect(stdin, stdout, event_tx, permission_tx, notification_tx);
+    let connect = AcpProtocol::connect(stdin, stdout, event_tx, permission_tx, notification_tx, None);
     let protocol = tokio::select! {
         biased;
         res = connect => match res {