From 1ef1fd9e40fe4be0b96673a2d0ee550952a4f531 Mon Sep 17 00:00:00 2001 From: Blomios Date: Tue, 28 Jul 2026 16:56:22 +0200 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20ajoute=20l'outil=20idea=5Fask=5Fag?= =?UTF-8?q?ents=20pour=20la=20d=C3=A9l=C3=A9gation=20parall=C3=A8le=20(#2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrapper MCP backend pur : dispatche N requêtes idea_ask_agent en parallèle et retourne un résultat par requête, dans l'ordre, en rejetant les cibles dupliquées avant dispatch. Aucun changement de domaine/application — pose la primitive nécessaire au SDK (chantier feature/sdk-integration). QA vert (ticket #2). Co-Authored-By: Claude Sonnet 5 --- crates/backend/src/lib.rs | 5 +- .../src/orchestrator/mcp/server.rs | 192 ++++++++++++++- .../src/orchestrator/mcp/tools.rs | 40 +++ crates/infrastructure/tests/mcp_server.rs | 229 +++++++++++++++++- 4 files changed, 456 insertions(+), 10 deletions(-) diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index f12d4cc..ccc27eb 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -4923,6 +4923,7 @@ mod mcp_serve_peer_tests { for expected in [ "idea_list_agents", "idea_ask_agent", + "idea_ask_agents", "idea_run_in_background", "idea_launch_agent", "idea_stop_agent", @@ -4967,8 +4968,8 @@ mod mcp_serve_peer_tests { assert!(!names.contains(&"idea_reply")); assert_eq!( tools.len(), - 30, - "exactly the thirty exposed idea_* tools; got {names:?}" + 31, + "exactly the thirty-one exposed idea_* tools; got {names:?}" ); drop(client); // EOF ⇒ serve loop ends diff --git a/crates/infrastructure/src/orchestrator/mcp/server.rs b/crates/infrastructure/src/orchestrator/mcp/server.rs index 14bae66..ef791b7 100644 --- a/crates/infrastructure/src/orchestrator/mcp/server.rs +++ b/crates/infrastructure/src/orchestrator/mcp/server.rs @@ -23,8 +23,10 @@ use std::time::Instant; use application::OrchestratorService; use domain::ports::McpToolPermissionStore; use domain::{ - AgentId, AgentToolPolicy, DomainEvent, IssueRef, McpToolPolicy, OrchestrationSource, Project, + AgentId, AgentToolPolicy, DomainEvent, IssueRef, McpToolPolicy, OrchestrationSource, + OrchestratorCommand, Project, }; +use futures_util::future::join_all; use serde_json::{json, Value}; use tokio::sync::mpsc; @@ -482,6 +484,12 @@ impl McpServer { task_len={task_len}", ); + if name == "idea_ask_agents" { + return self + .tools_call_ask_agents(arguments, requester_label, arg_target, started) + .await; + } + if tools::is_ticket_tool(&name) { let result = match &self.ticket_tools { Some(provider) => { @@ -611,6 +619,65 @@ impl McpServer { } } + /// Comfort wrapper over several parallel `idea_ask_agent` dispatches. + /// + /// This remains an MCP-adapter-only fan-out/fan-in: each item is validated and + /// translated to the existing [`OrchestratorCommand::AskAgent`] shape, then sent to + /// [`OrchestratorService::dispatch`]. Per-item dispatch failures are folded into + /// the result array so one failed target never cancels the others. + async fn tools_call_ask_agents( + &self, + arguments: Value, + requester_label: String, + arg_target: String, + started: Instant, + ) -> Result { + let requests = parse_ask_agents_requests(&arguments, &self.requester)?; + let project = self.project.clone(); + let service = Arc::clone(&self.service); + + let futures = requests.iter().cloned().map(|request| { + let project = project.clone(); + let service = Arc::clone(&service); + async move { + let command = OrchestratorCommand::AskAgent { + target: request.target.clone(), + task: request.task, + requester: request.requester, + }; + match service.dispatch(&project, command).await { + Ok(outcome) => AskAgentsItemResult { + target: request.target, + ok: true, + text: outcome.reply.unwrap_or(outcome.detail), + }, + Err(err) => AskAgentsItemResult { + target: request.target, + ok: false, + text: err.to_string(), + }, + } + } + }); + + let results = join_all(futures).await; + self.publish_processed("idea_ask_agents", true); + let text = serde_json::to_string( + &results + .iter() + .map(AskAgentsItemResult::to_value) + .collect::>(), + ) + .unwrap_or_else(|_| "[]".to_owned()); + application::diag!( + "[mcp] tools_call end tool=idea_ask_agents requester={requester_label} \ + target={arg_target} ok=true is_error=false result_len={} elapsed_ms={}", + text.len(), + started.elapsed().as_millis(), + ); + Ok(tool_result_text(&text, false)) + } + fn enforce_tool_policy( &self, policy: &AgentToolPolicy, @@ -690,6 +757,129 @@ fn tool_result_text(text: &str, is_error: bool) -> Value { }) } +#[derive(Clone)] +struct AskAgentsRequest { + target: String, + task: String, + requester: Option, +} + +struct AskAgentsItemResult { + target: String, + ok: bool, + text: String, +} + +impl AskAgentsItemResult { + fn to_value(&self) -> Value { + json!({ + "target": self.target, + "ok": self.ok, + "isError": !self.ok, + "text": self.text, + }) + } +} + +fn parse_ask_agents_requests( + arguments: &Value, + requester: &str, +) -> Result, JsonRpcError> { + let args = arguments.as_object().ok_or_else(|| { + JsonRpcError::new( + error_codes::INVALID_PARAMS, + "tool `idea_ask_agents` arguments must be a JSON object", + ) + })?; + if args.len() != 1 || !args.contains_key("requests") { + return Err(JsonRpcError::new( + error_codes::INVALID_PARAMS, + "tool `idea_ask_agents` expects exactly one field: requests", + )); + } + let requests = args + .get("requests") + .and_then(Value::as_array) + .ok_or_else(|| { + JsonRpcError::new( + error_codes::INVALID_PARAMS, + "tool `idea_ask_agents` field `requests` must be a non-empty array", + ) + })?; + if requests.is_empty() { + return Err(JsonRpcError::new( + error_codes::INVALID_PARAMS, + "tool `idea_ask_agents` field `requests` must be a non-empty array", + )); + } + + let mut seen = std::collections::HashSet::new(); + let mut out = Vec::with_capacity(requests.len()); + for (index, request) in requests.iter().enumerate() { + let obj = request.as_object().ok_or_else(|| { + JsonRpcError::new( + error_codes::INVALID_PARAMS, + format!("tool `idea_ask_agents` requests[{index}] must be an object"), + ) + })?; + if obj.len() != 2 || !obj.contains_key("target") || !obj.contains_key("task") { + return Err(JsonRpcError::new( + error_codes::INVALID_PARAMS, + format!("tool `idea_ask_agents` requests[{index}] expects exactly target and task"), + )); + } + let target = obj + .get("target") + .and_then(Value::as_str) + .filter(|v| !v.trim().is_empty()) + .ok_or_else(|| { + JsonRpcError::new( + error_codes::INVALID_PARAMS, + format!("tool `idea_ask_agents` requests[{index}].target must be a string"), + ) + })? + .to_owned(); + let task = obj + .get("task") + .and_then(Value::as_str) + .filter(|v| !v.trim().is_empty()) + .ok_or_else(|| { + JsonRpcError::new( + error_codes::INVALID_PARAMS, + format!("tool `idea_ask_agents` requests[{index}].task must be a string"), + ) + })? + .to_owned(); + let duplicate_key = target.to_lowercase(); + if !seen.insert(duplicate_key) { + return Err(JsonRpcError::new( + error_codes::INVALID_PARAMS, + format!("tool `idea_ask_agents` duplicate target `{target}` is not supported"), + )); + } + let command = tools::map_tool_call( + "idea_ask_agent", + &json!({ "target": target, "task": task }), + requester, + ) + .map_err(map_err_to_jsonrpc)?; + let OrchestratorCommand::AskAgent { + target, + task, + requester, + } = command + else { + unreachable!("idea_ask_agent must map to AskAgent"); + }; + out.push(AskAgentsRequest { + target, + task, + requester, + }); + } + Ok(out) +} + /// Maps a [`ToolMapError`] to the right JSON-RPC error code. fn map_err_to_jsonrpc(err: ToolMapError) -> JsonRpcError { match err { diff --git a/crates/infrastructure/src/orchestrator/mcp/tools.rs b/crates/infrastructure/src/orchestrator/mcp/tools.rs index 9235569..600b401 100644 --- a/crates/infrastructure/src/orchestrator/mcp/tools.rs +++ b/crates/infrastructure/src/orchestrator/mcp/tools.rs @@ -61,6 +61,7 @@ pub const READ_ONLY_TOOLS: &[&str] = &[ /// Canonical write/action MCP tools denied by default. pub const WRITE_ACTION_TOOLS: &[&str] = &[ "idea_ask_agent", + "idea_ask_agents", "idea_run_in_background", "idea_launch_agent", "idea_stop_agent", @@ -125,6 +126,7 @@ pub fn tool_returns_reply(tool: &str) -> bool { matches!( tool, "idea_ask_agent" + | "idea_ask_agents" | "idea_list_agents" | "idea_context_read" | "idea_memory_read" @@ -180,6 +182,32 @@ pub fn catalogue() -> Vec { "additionalProperties": false }), }, + ToolDef { + name: "idea_ask_agents", + description: "Ask several IdeA agents in parallel and return one result per request, \ + in the same order. Each request is equivalent to one idea_ask_agent \ + call; duplicate targets are rejected before dispatch.", + input_schema: json!({ + "type": "object", + "properties": { + "requests": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "properties": { + "target": { "type": "string", "description": "Target agent display name." }, + "task": { "type": "string", "description": "Task or question to send to the target agent." } + }, + "required": ["target", "task"], + "additionalProperties": false + } + } + }, + "required": ["requests"], + "additionalProperties": false + }), + }, ToolDef { name: "idea_run_in_background", description: "Run a command as a first-class IdeA background task. Returns the task id \ @@ -659,6 +687,18 @@ mod tests { ); } + #[test] + fn ask_agents_is_classified_and_returns_inline_reply() { + assert_eq!( + tool_access("idea_ask_agents"), + Some(McpToolAccess::WriteAction) + ); + assert!(catalogue() + .into_iter() + .any(|tool| tool.name == "idea_ask_agents")); + assert!(tool_returns_reply("idea_ask_agents")); + } + #[test] fn run_in_background_maps_to_background_command_with_handshake_owner() { let cmd = map_tool_call( diff --git a/crates/infrastructure/tests/mcp_server.rs b/crates/infrastructure/tests/mcp_server.rs index 7bbe397..2c30df1 100644 --- a/crates/infrastructure/tests/mcp_server.rs +++ b/crates/infrastructure/tests/mcp_server.rs @@ -32,10 +32,11 @@ use domain::ids::SkillId; use domain::ids::{AgentId, ProfileId, ProjectId}; use domain::markdown::MarkdownDoc; use domain::ports::{ - AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream, - ExitStatus, FileSystem, FsError, IdGenerator, McpToolPermissionStore, OutputStream, - PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError, - SessionPlan, SkillStore, SpawnSpec, StoreError, + AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, ContextInjectionPlan, + DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError, IdGenerator, + McpToolPermissionStore, OutputStream, PreparedContext, ProfileStore, PtyError, PtyHandle, + PtyPort, RemotePath, ReplyEvent, ReplyStream, RuntimeError, SessionPlan, SkillStore, SpawnSpec, + StoreError, }; use domain::profile::{ AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport, @@ -53,7 +54,7 @@ use uuid::Uuid; use application::{ CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, - OrchestratorService, TerminalSessions, UpdateAgentContext, + OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext, }; use infrastructure::orchestrator::mcp::jsonrpc::error_codes; use infrastructure::orchestrator::mcp::tools::classified_tool_names; @@ -467,6 +468,66 @@ fn build_service_with_mailbox( (Arc::new(service), mailbox, sessions) } +#[derive(Clone)] +struct FakeStructuredSession { + id: SessionId, + reply: String, + delay_ms: u64, +} + +#[async_trait] +impl AgentSession for FakeStructuredSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(format!("conv-{}", self.id)) + } + + async fn send(&self, _prompt: &str) -> Result { + if self.delay_ms > 0 { + tokio::time::sleep(std::time::Duration::from_millis(self.delay_ms)).await; + } + Ok(Box::new(std::iter::once(ReplyEvent::Final { + content: self.reply.clone(), + }))) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +fn build_service_with_structured( + contexts: FakeContexts, +) -> (Arc, Arc) { + let (base, _mailbox, _sessions) = build_service_with_mailbox(contexts); + let structured = Arc::new(StructuredSessions::new()); + let service = Arc::try_unwrap(base) + .unwrap_or_else(|_| panic!("freshly built service must be uniquely owned")) + .with_structured(Arc::clone(&structured)); + (Arc::new(service), structured) +} + +fn insert_structured_reply( + structured: &StructuredSessions, + agent_id: AgentId, + reply: &str, + delay_ms: u64, +) { + structured.insert_in_project( + project().id, + Arc::new(FakeStructuredSession { + id: SessionId::from_uuid(Uuid::new_v4()), + reply: reply.to_owned(), + delay_ms, + }), + agent_id, + NodeId::from_uuid(Uuid::new_v4()), + ); +} + fn server(service: Arc) -> McpServer { McpServer::new(service, project()) } @@ -665,6 +726,7 @@ async fn tools_list_advertises_the_idea_tools_with_schemas() { for expected in [ "idea_list_agents", "idea_ask_agent", + "idea_ask_agents", "idea_run_in_background", "idea_launch_agent", "idea_stop_agent", @@ -707,8 +769,8 @@ async fn tools_list_advertises_the_idea_tools_with_schemas() { assert!(!names.contains(&"idea_reply")); assert_eq!( tools.len(), - 30, - "exactly the thirty exposed idea_* tools; got {names:?}" + 31, + "exactly the thirty-one exposed idea_* tools; got {names:?}" ); // Every tool advertises an object input schema. @@ -1446,6 +1508,159 @@ async fn reply_tool_is_not_exposed_over_mcp() { assert!(resp.result.is_none()); } +#[tokio::test] +async fn ask_agents_fans_out_in_parallel_and_preserves_input_order() { + let contexts = FakeContexts::new(); + let slow_id = contexts.seed_agent("slow"); + let fast_id = contexts.seed_agent("fast"); + let (service, structured) = build_service_with_structured(contexts); + insert_structured_reply(&structured, slow_id, "slow reply", 60); + insert_structured_reply(&structured, fast_id, "fast reply", 0); + let server = server(service); + + let raw = tools_call( + 21, + "idea_ask_agents", + json!({ + "requests": [ + { "target": "slow", "task": "first" }, + { "target": "fast", "task": "second" } + ] + }), + ); + let response = server.handle_raw(&raw).await.expect("reply owed"); + assert!( + response.error.is_none(), + "transport error: {:?}", + response.error + ); + let result = response.result.expect("result"); + assert_eq!(result["isError"], json!(false), "got {result}"); + let payload: Value = + serde_json::from_str(result_text(&result)).expect("ask_agents result is JSON"); + let items = payload.as_array().expect("result array"); + assert_eq!(items.len(), 2); + assert_eq!(items[0]["target"], json!("slow")); + assert_eq!(items[0]["ok"], json!(true)); + assert_eq!(items[0]["isError"], json!(false)); + assert_eq!(items[0]["text"], json!("slow reply")); + assert_eq!(items[1]["target"], json!("fast")); + assert_eq!(items[1]["ok"], json!(true)); + assert_eq!(items[1]["text"], json!("fast reply")); +} + +#[tokio::test] +async fn ask_agents_allows_partial_success() { + let contexts = FakeContexts::new(); + let ok_id = contexts.seed_agent("ok-agent"); + let (service, structured) = build_service_with_structured(contexts); + insert_structured_reply(&structured, ok_id, "ok reply", 0); + let server = server(service); + + let raw = tools_call( + 22, + "idea_ask_agents", + json!({ + "requests": [ + { "target": "ok-agent", "task": "works" }, + { "target": "missing-agent", "task": "fails" } + ] + }), + ); + let response = server.handle_raw(&raw).await.expect("reply owed"); + assert!( + response.error.is_none(), + "wrapper should not fail transport" + ); + let result = response.result.expect("result"); + assert_eq!( + result["isError"], + json!(false), + "partial success is allowed" + ); + + let payload: Value = serde_json::from_str(result_text(&result)).unwrap(); + let items = payload.as_array().unwrap(); + assert_eq!(items[0]["target"], json!("ok-agent")); + assert_eq!(items[0]["ok"], json!(true)); + assert_eq!(items[0]["text"], json!("ok reply")); + assert_eq!(items[1]["target"], json!("missing-agent")); + assert_eq!(items[1]["ok"], json!(false)); + assert_eq!(items[1]["isError"], json!(true)); + assert!( + items[1]["text"].as_str().unwrap().contains("missing-agent"), + "per-target error should be surfaced; got {payload}" + ); +} + +#[tokio::test] +async fn ask_agents_rejects_duplicate_targets_before_dispatch() { + let contexts = FakeContexts::new(); + let agent_id = contexts.seed_agent("Architect"); + let (service, structured) = build_service_with_structured(contexts); + insert_structured_reply(&structured, agent_id, "must not run", 0); + let server = server(service); + + let raw = tools_call( + 23, + "idea_ask_agents", + json!({ + "requests": [ + { "target": "Architect", "task": "one" }, + { "target": "architect", "task": "two" } + ] + }), + ); + let response = server.handle_raw(&raw).await.expect("reply owed"); + let error = response.error.expect("duplicate validation error expected"); + assert_eq!(error.code, error_codes::INVALID_PARAMS); + assert!( + error.message.contains("duplicate target"), + "got {}", + error.message + ); + assert!(response.result.is_none()); +} + +#[tokio::test] +async fn ask_agents_rejects_non_strict_shapes_before_dispatch() { + let (service, _structured) = build_service_with_structured(FakeContexts::new()); + let server = server(service); + + for (id, arguments, expected) in [ + (24, json!({}), "requests"), + (25, json!({ "requests": [] }), "non-empty"), + ( + 26, + json!({ "requests": [{ "target": "a", "task": "x", "extra": true }] }), + "exactly target and task", + ), + ( + 27, + json!({ "requests": [{ "target": "a" }] }), + "exactly target and task", + ), + ( + 28, + json!({ "requests": [{ "target": "", "task": "x" }] }), + "target", + ), + ] { + let response = server + .handle_raw(&tools_call(id, "idea_ask_agents", arguments)) + .await + .expect("reply owed"); + let error = response.error.expect("validation error expected"); + assert_eq!(error.code, error_codes::INVALID_PARAMS); + assert!( + error.message.contains(expected), + "expected message to contain `{expected}`, got {}", + error.message + ); + assert!(response.result.is_none()); + } +} + // --------------------------------------------------------------------------- // 4. idea_list_agents returns the agent list inline (JSON array) // ---------------------------------------------------------------------------