From e5e412bdf213369bc8cddcdb93cc099c873c12b9 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sun, 21 Jun 2026 09:41:02 +0200 Subject: [PATCH] =?UTF-8?q?feat(diag):=20instrumente=20les=20blocages=20de?= =?UTF-8?q?=20d=C3=A9l=C3=A9gation=20idea=5Fask=5Fagent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ajoute des logs diag! sans changement sémantique le long du canal de délégation inter-agents (application/orchestrator, infrastructure input, mailbox, mcp server & tools) pour diagnostiquer les blocages récurrents de idea_ask_agent. Couvert par les tests existants étendus (mcp_server, mailbox, input, tools, orchestrator_service). Co-Authored-By: Claude Opus 4.8 --- .../application/src/orchestrator/service.rs | 93 ++++++++++++- crates/infrastructure/src/input/mod.rs | 45 +++++-- crates/infrastructure/src/mailbox/mod.rs | 124 +++++++++++++++--- .../src/orchestrator/mcp/server.rs | 72 +++++++++- .../src/orchestrator/mcp/tools.rs | 28 +++- crates/infrastructure/tests/mcp_server.rs | 107 +++++++++++++++ 6 files changed, 426 insertions(+), 43 deletions(-) diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index e267673..d5904e0 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -890,6 +890,10 @@ impl OrchestratorService { g.would_cycle(from, agent_id) }; if cycles { + crate::diag!( + "[rendezvous] cycle refused: requester={from} -> target={target} \ + (agent {agent_id}) (deadlock évité, aucun enqueue)", + ); return Err(AppError::Invalid(format!( "délégation ré-entrante refusée : demander à l'agent '{target}' créerait \ un cycle d'attente inter-agents (deadlock évité)" @@ -905,9 +909,31 @@ impl OrchestratorService { // Sérialisation FIFO **par agent** (A0) : verrou de tour de la **cible**, tenu // pour TOUT le tour (enqueue → réponse). RAII : tombe sur chaque early-return. let lock = self.ask_lock_for(&agent_id); + // Diagnostics : tracer l'attente du verrou de tour. Un « turn-lock waiting » sans + // « turn-lock acquired » qui suit signe un tour précédent jamais relâché (la cible + // reste Busy) — le verrou n'est libéré qu'à la fin du tour courant. + crate::diag!( + "[rendezvous] turn-lock waiting: requester={} -> target={target} (agent {agent_id}) \ + wait_cap_ms={}", + requester.map_or_else(|| "user".to_owned(), |a| a.to_string()), + ASK_QUEUE_WAIT_CAP.as_millis(), + ); + let lock_wait = Instant::now(); let _turn = match tokio::time::timeout(ASK_QUEUE_WAIT_CAP, lock.lock_owned()).await { - Ok(guard) => guard, + Ok(guard) => { + crate::diag!( + "[rendezvous] turn-lock acquired: target={target} (agent {agent_id}) \ + waited_ms={}", + lock_wait.elapsed().as_millis(), + ); + guard + } Err(_elapsed) => { + crate::diag!( + "[rendezvous] turn-lock TIMEOUT: target={target} (agent {agent_id}) \ + waited_ms={} (tour précédent jamais relâché ?)", + lock_wait.elapsed().as_millis(), + ); return Err(AppError::from(domain::ports::AgentSessionError::Timeout)); } }; @@ -967,6 +993,14 @@ impl OrchestratorService { // des deux ⇒ pas de gate (livraison immédiate, sinon blocage indéfini). let gate_cold_start = cold_launch && (prompt_pattern.as_ref().is_some_and(|p| !p.is_empty()) || has_mcp); + // Diagnostics : décision de gate du premier tour. `gate_cold_start=false` sur une + // cible froide SANS signal de libération (ni prompt-ready ni MCP) livrerait + // immédiatement ; un `true` diffère la livraison jusqu'au signal. + crate::diag!( + "[rendezvous] gate decision: target={target} (agent {agent_id}) cold_launch={cold_launch} \ + has_prompt_pattern={} has_mcp={has_mcp} gate_cold_start={gate_cold_start}", + prompt_pattern.as_ref().is_some_and(|p| !p.is_empty()), + ); if gate_cold_start { input.mark_starting(agent_id); } @@ -1011,8 +1045,8 @@ impl OrchestratorService { let started = Instant::now(); crate::diag!( "[rendezvous] ask started: requester={} -> target={target} (agent {agent_id}) \ - ticket={ticket_id} cold_launch={cold_launch} gate_cold_start={gate_cold_start} \ - turn_timeout_ms={}", + conversation={conversation_id} ticket={ticket_id} cold_launch={cold_launch} \ + gate_cold_start={gate_cold_start} turn_timeout_ms={}", requester.map_or_else(|| "user".to_owned(), |a| a.to_string()), turn_timeout.as_millis(), ); @@ -1155,6 +1189,18 @@ impl OrchestratorService { let busy_guard = BusyTurnGuard::new(Arc::clone(input), Arc::clone(mailbox), agent_id, ticket_id); + // Rendezvous beacon (chemin structuré) : équivalent du « ask started » du chemin + // PTY. La cible n'a pas de PTY ; le tour se débloque sur le `Final` de sa session + // OU sur un `idea_reply`. Sans l'un des deux avant `turn_timeout`, le drain expire. + let started = Instant::now(); + crate::diag!( + "[rendezvous] ask started (structured): requester={} -> target={target} \ + (agent {agent_id}) conversation={conversation_id} ticket={ticket_id} \ + turn_timeout_ms={}", + requester.map_or_else(|| "user".to_owned(), |a| a.to_string()), + turn_timeout.as_millis(), + ); + // Drainer le tour structuré (le `Final` ⇒ contenu + `mark_idle`), borné par le // **même** garde-fou que le chemin PTY (profil ou défaut). On attend la // **première** issue : le tour structuré OU un `idea_reply` explicite. @@ -1165,9 +1211,24 @@ impl OrchestratorService { biased; // Issue déterministe : la session a rendu son `Final`. drained = drain => match drained { - Ok(content) => content, + Ok(content) => { + crate::diag!( + "[rendezvous] ask resolved (structured/final): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}", + started.elapsed().as_millis(), + content.len(), + ); + content + } // Erreur de drain : le garde fait `cancel_head` + `mark_idle` au Drop. - Err(err) => return Err(AppError::from(err)), + Err(err) => { + crate::diag!( + "[rendezvous] ask drain-error (structured): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={} err={err}", + started.elapsed().as_millis(), + ); + return Err(AppError::from(err)); + } }, // Issue alternative : un `idea_reply` explicite a résolu le ticket d'abord. replied = pending => match replied { @@ -1175,10 +1236,21 @@ impl OrchestratorService { // La session draine encore en arrière-plan ; on s'assure que la FIFO // avance même si le `Final` n'est pas encore observé. input.mark_idle(agent_id); + crate::diag!( + "[rendezvous] ask resolved (structured/reply): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}", + started.elapsed().as_millis(), + content.len(), + ); content } // Canal fermé : le garde fait `cancel_head` + `mark_idle` au Drop. Err(_cancelled) => { + crate::diag!( + "[rendezvous] ask channel-closed (structured): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={}", + started.elapsed().as_millis(), + ); return Err(AppError::Process(format!( "agent {target} : canal de réponse fermé avant un résultat" ))); @@ -1413,7 +1485,18 @@ impl OrchestratorService { ticket: Option, result: String, ) -> Result { + // Diagnostics : un `idea_reply` est entré, AVANT toute corrélation. Longueur + // seulement (jamais le corps). Si ce beacon apparaît mais que `correlated`/ + // `UNMATCHED` ne suit pas, la file n'est pas câblée. + crate::diag!( + "[rendezvous] idea_reply received: from agent {from} ticket={} result_len={}", + ticket.map_or_else(|| "head".to_owned(), |t| t.to_string()), + result.len(), + ); let mailbox = self.mailbox.as_ref().ok_or_else(|| { + crate::diag!( + "[rendezvous] idea_reply DROPPED: from agent {from} (file inter-agents non câblée)", + ); AppError::Invalid( "idea_reply n'est pas disponible : file inter-agents non câblée".to_owned(), ) diff --git a/crates/infrastructure/src/input/mod.rs b/crates/infrastructure/src/input/mod.rs index 9c460ca..90b939f 100644 --- a/crates/infrastructure/src/input/mod.rs +++ b/crates/infrastructure/src/input/mod.rs @@ -147,7 +147,7 @@ impl BusyTracker { /// appel, l'`enqueue` publie la `DelegationReady` immédiatement (chemin chaud, /// zéro régression). fn mark_starting(&self, agent: AgentId) { - eprintln!("[input-mediator] cold-start gate armed agent={agent}"); + application::diag!("[input-mediator] mark_starting cold-start gate armed agent={agent}"); self.lock_starting().insert(agent); } @@ -235,6 +235,10 @@ impl BusyTracker { .collect() }; for agent in newly_stalled { + // Transition de vivacité Alive→Stalled : le tour de l'agent n'a pas battu + // depuis plus que son seuil. Beacon événementiel (une fois par transition, + // jamais en boucle) — utile pour repérer une cible silencieuse pendant un ask. + application::diag!("[input-mediator] liveness Alive->Stalled agent={agent}"); self.publish_liveness(agent, AgentLiveness::Stalled); } } @@ -264,10 +268,10 @@ impl BusyTracker { let mut busy = self.lock(); let entry = busy.entry(agent).or_insert(AgentBusyState::Idle); if entry.is_busy() { - eprintln!("[input-mediator] enqueue queued behind busy agent={agent}"); + application::diag!("[input-mediator] start_turn queued behind busy agent={agent}"); false } else { - eprintln!("[input-mediator] turn started agent={agent} state={state:?}"); + application::diag!("[input-mediator] start_turn started agent={agent} state={state:?}"); *entry = state; true } @@ -285,15 +289,17 @@ impl BusyTracker { Some(sink) => match sink(agent, d) { Some(d) => d, None => { - eprintln!("[input-mediator] delegation handled by headless sink agent={agent}"); + application::diag!( + "[input-mediator] delegationReady handled by headless sink agent={agent}" + ); return; } }, None => d, }; if let Some(events) = &self.events { - eprintln!( - "[input-mediator] publishing DelegationReady agent={agent} ticket={} text_bytes={} submit_sequence={} submit_delay_ms={:?}", + application::diag!( + "[input-mediator] delegationReady -> write-portal agent={agent} ticket={} text_bytes={} submit_sequence={} submit_delay_ms={:?}", d.ticket, d.text.as_bytes().len(), describe_submit_sequence(d.submit_sequence.as_deref()), @@ -323,13 +329,24 @@ impl BusyTracker { let deferred = self.lock_deferred().remove(&agent); match deferred { Some(d) => { - eprintln!( - "[input-mediator] prompt-ready released deferred delegation agent={agent} ticket={}", + application::diag!( + "[input-mediator] prompt_ready released deferred delegation agent={agent} ticket={}", d.ticket ); self.publish_deferred(agent, d); } - None => self.mark_idle(agent), + None => { + // Pas de tour différé : le prompt-ready ne fait que **libérer** le busy + // state / la FIFO (mark_idle). Il ne porte AUCUNE réponse pour réveiller + // un appelant en attente — d'où `no_reply_payload`. C'est exactement le + // scénario où une cible revient prompt-ready sans `idea_reply` : l'ask + // ne se résout pas par ce chemin, seulement par l'`idea_reply` ou le + // timeout côté appelant. + application::diag!( + "[input-mediator] prompt_ready agent={agent} no_reply_payload -> mark_idle" + ); + self.mark_idle(agent); + } } } @@ -341,8 +358,8 @@ impl BusyTracker { fn release_cold_start(&self, agent: AgentId) { self.lock_starting().remove(&agent); if let Some(d) = self.lock_deferred().remove(&agent) { - eprintln!( - "[input-mediator] mcp-ready released deferred delegation agent={agent} ticket={}", + application::diag!( + "[input-mediator] release_cold_start (mcp-ready) released deferred delegation agent={agent} ticket={}", d.ticket ); self.publish_deferred(agent, d); @@ -359,7 +376,7 @@ impl BusyTracker { .is_some_and(|s| s.is_busy()) }; if was_busy { - eprintln!("[input-mediator] turn marked idle agent={agent}"); + application::diag!("[input-mediator] mark_idle turn ended agent={agent}"); if let Some(events) = &self.events { events.publish(DomainEvent::AgentBusyChanged { agent_id: agent, @@ -837,6 +854,9 @@ impl InputMediator for MediatedInbox { submit_delay_ms: submit.delay_ms, }; if self.tracker.lock_starting().contains(&agent) { + application::diag!( + "[input-mediator] enqueue deferred (cold-start gate) agent={agent} ticket={ticket_id}" + ); self.tracker.lock_deferred().insert(agent, deferred); } else { self.tracker.publish_deferred(agent, deferred); @@ -918,6 +938,7 @@ impl InputMediator for MediatedInbox { // CLI agents honour. A missing handle/port is a no-op (the agent simply has no // live stream to interrupt). The busy state is left untouched: it returns to // Idle through `mark_idle` (prompt-ready / explicit signal), not here. + application::diag!("[input-mediator] preempt agent={agent} (interrupt byte, no ticket)"); if let Some(pty) = &self.pty { if let Some(handle) = self.handles().get(&agent).cloned() { let _ = pty.write(&handle, b"\x1b"); diff --git a/crates/infrastructure/src/mailbox/mod.rs b/crates/infrastructure/src/mailbox/mod.rs index 93b262b..ae12b1c 100644 --- a/crates/infrastructure/src/mailbox/mod.rs +++ b/crates/infrastructure/src/mailbox/mod.rs @@ -77,16 +77,21 @@ impl InMemoryMailbox { impl AgentMailbox for InMemoryMailbox { fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply { let (tx, rx) = oneshot::channel::(); - { + let ticket_id = ticket.id; + let (depth_before, depth_after) = { let mut queues = self .queues .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - queues - .entry(agent) - .or_default() - .push_back(Slot { ticket, reply: tx }); - } + let queue = queues.entry(agent).or_default(); + let before = queue.len(); + queue.push_back(Slot { ticket, reply: tx }); + (before, queue.len()) + }; + application::diag!( + "[mailbox] enqueue agent={agent} ticket={ticket_id} \ + queue_depth_before={depth_before} queue_depth_after={depth_after}" + ); // The await happens in the application layer, outside the map lock. A closed // channel (sender dropped without a value, e.g. the head was cancelled or the // session ended) maps to a typed `Cancelled` rather than a raw recv error. @@ -99,7 +104,7 @@ impl AgentMailbox for InMemoryMailbox { // Pop the head slot under the lock, then send **outside** any await (the send // is non-blocking). If the receiver already went away (caller timed out), the // ticket is still correctly retired from the head — the queue advances. - let slot = { + let (slot, depth_before, depth_after) = { let mut queues = self .queues .lock() @@ -107,9 +112,21 @@ impl AgentMailbox for InMemoryMailbox { let queue = queues .get_mut(&agent) .filter(|q| !q.is_empty()) - .ok_or(MailboxError::NoPendingRequest(agent))?; - queue.pop_front().expect("non-empty queue has a head") + .ok_or_else(|| { + application::diag!( + "[mailbox] resolve no-pending agent={agent} (réponse orpheline ?)" + ); + MailboxError::NoPendingRequest(agent) + })?; + let before = queue.len(); + let slot = queue.pop_front().expect("non-empty queue has a head"); + (slot, before, queue.len()) }; + application::diag!( + "[mailbox] resolve agent={agent} ticket={} \ + queue_depth_before={depth_before} queue_depth_after={depth_after}", + slot.ticket.id, + ); // A dropped receiver (the awaiting caller timed out and went away) is fine: // the reply is simply discarded, the head ticket is already retired. let _ = slot.reply.send(result); @@ -125,7 +142,7 @@ impl AgentMailbox for InMemoryMailbox { // Remove the slot whose ticket id matches, anywhere in the queue (multi-thread // correlation), then send outside any await. A missing match is a typed // NoPendingRequest — never a panic. - let slot = { + let (slot, depth_before, depth_after) = { let mut queues = self .queues .lock() @@ -133,33 +150,67 @@ impl AgentMailbox for InMemoryMailbox { let queue = queues .get_mut(&agent) .filter(|q| !q.is_empty()) - .ok_or(MailboxError::NoPendingRequest(agent))?; + .ok_or_else(|| { + application::diag!( + "[mailbox] resolve_ticket no-queue agent={agent} ticket={ticket_id}" + ); + MailboxError::NoPendingRequest(agent) + })?; + let before = queue.len(); let pos = queue .iter() .position(|s| s.ticket.id == ticket_id) - .ok_or(MailboxError::NoPendingRequest(agent))?; - queue.remove(pos).expect("position just found is in range") + .ok_or_else(|| { + application::diag!( + "[mailbox] resolve_ticket no-match agent={agent} ticket={ticket_id} \ + queue_depth={before}" + ); + MailboxError::NoPendingRequest(agent) + })?; + let slot = queue.remove(pos).expect("position just found is in range"); + (slot, before, queue.len()) }; + application::diag!( + "[mailbox] resolve_ticket agent={agent} ticket={ticket_id} \ + queue_depth_before={depth_before} queue_depth_after={depth_after}" + ); let _ = slot.reply.send(result); Ok(()) } fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) { - let mut queues = self - .queues - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(queue) = queues.get_mut(&agent) { + let (depth_before, depth_after, retired) = { + let mut queues = self + .queues + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(queue) = queues.get_mut(&agent) else { + application::diag!( + "[mailbox] cancel_head no-queue agent={agent} ticket={ticket_id} \ + queue_depth_before=0 queue_depth_after=0 retired=false" + ); + return; + }; + let before = queue.len(); // Only retire the head, and only if it is *this* ticket: a head that has // since changed (already resolved, or someone else's ticket now in front) // must not be dropped. Idempotent and positional. - if queue.front().map(|s| s.ticket.id) == Some(ticket_id) { + let retired = if queue.front().map(|s| s.ticket.id) == Some(ticket_id) { queue.pop_front(); - } + true + } else { + false + }; + let after = queue.len(); if queue.is_empty() { queues.remove(&agent); } - } + (before, after, retired) + }; + application::diag!( + "[mailbox] cancel_head agent={agent} ticket={ticket_id} \ + queue_depth_before={depth_before} queue_depth_after={depth_after} retired={retired}" + ); } } @@ -382,6 +433,37 @@ mod tests { assert_eq!(snap[0].position, 0, "remaining ticket becomes head"); } + /// Diagnostics capture (instrumentation) : `enqueue`/`resolve` émettent une ligne + /// `[mailbox]` avec les profondeurs avant/après dans le sink diag. On pointe le sink + /// (OnceLock du process) sur un fichier temp et on filtre par des id uniques à ce + /// test (immunisé contre le parallélisme et les ré-exécutions). + #[test] + fn diag_capture_enqueue_resolve_log_queue_depths() { + let path = std::env::temp_dir().join("idea-diag-mailbox-lib-test.log"); + application::diag::set_log_path(path.clone()); + + let mb = InMemoryMailbox::new(); + let a = agent(777_001); + let t = tid(777_010); + let _p = mb.enqueue(a, Ticket::new(t, "Main", "diag task")); + mb.resolve(a, "done".to_owned()).expect("resolve ok"); + + let body = std::fs::read_to_string(&path).unwrap_or_default(); + let (aid, tkt) = (a.to_string(), t.to_string()); + assert!( + body.contains(&format!( + "[mailbox] enqueue agent={aid} ticket={tkt} queue_depth_before=0 queue_depth_after=1" + )), + "enqueue beacon with depths missing: {body}" + ); + assert!( + body.contains(&format!( + "[mailbox] resolve agent={aid} ticket={tkt} queue_depth_before=1 queue_depth_after=0" + )), + "resolve beacon with depths missing: {body}" + ); + } + #[test] fn snapshot_updates_after_cancel_head() { let mb = InMemoryMailbox::new(); diff --git a/crates/infrastructure/src/orchestrator/mcp/server.rs b/crates/infrastructure/src/orchestrator/mcp/server.rs index af6e2ae..2e23925 100644 --- a/crates/infrastructure/src/orchestrator/mcp/server.rs +++ b/crates/infrastructure/src/orchestrator/mcp/server.rs @@ -18,7 +18,7 @@ //! duplicated here. use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use application::OrchestratorService; use domain::{DomainEvent, OrchestrationSource, Project}; @@ -345,10 +345,43 @@ impl McpServer { .to_owned(); let arguments = params.get("arguments").cloned().unwrap_or(json!({})); + // Diagnostics begin beacon (best-effort, jamais le corps task/result) : trace + // l'entrée d'un `tools/call`, son tool, le peer demandeur et la cible/longueurs. + // C'est l'amorce de la trace `[mcp]` begin → (armed) → (expired) → end qui + // permet de voir un `idea_ask_agent` entrer sans jamais ressortir (blocage). + let started = Instant::now(); + let requester_label = if self.requester.is_empty() { + "mcp".to_owned() + } else { + self.requester.clone() + }; + let arg_target = arguments + .get("target") + .and_then(Value::as_str) + .unwrap_or("-") + .to_owned(); + let task_len = arguments + .get("task") + .and_then(Value::as_str) + .map_or(0, str::len); + application::diag!( + "[mcp] tools_call begin tool={name} requester={requester_label} target={arg_target} \ + task_len={task_len}", + ); + // `idea_reply` needs the connected peer's identity as `from`; every other tool // ignores `requester`. The handshake-provided requester is the source of truth. - let command = - tools::map_tool_call(&name, &arguments, &self.requester).map_err(map_err_to_jsonrpc)?; + let command = match tools::map_tool_call(&name, &arguments, &self.requester) { + Ok(command) => command, + Err(e) => { + application::diag!( + "[mcp] tools_call end tool={name} requester={requester_label} mapped=err \ + err={e} elapsed_ms={}", + started.elapsed().as_millis(), + ); + return Err(map_err_to_jsonrpc(e)); + } + }; // `idea_ask_agent` is the only tool that blocks on a synchronous rendezvous // (the target's `idea_reply`). Bound *only* that path with a finite outer @@ -357,9 +390,23 @@ impl McpServer { // other tool dispatches unbounded — they never rendezvous. let dispatch = self.service.dispatch(&self.project, command); let result = if name == "idea_ask_agent" { + // Armed beacon : le rendezvous synchrone est désormais borné par le filet + // de sécurité serveur. Si « end » n'apparaît jamais après ce « armed », la + // cible n'a ni appelé `idea_reply` ni atteint son prompt-ready dans le délai. + application::diag!( + "[mcp] ask armed requester={requester_label} target={arg_target} \ + task_len={task_len} timeout_ms={}", + self.ask_rendezvous_timeout.as_millis(), + ); match tokio::time::timeout(self.ask_rendezvous_timeout, dispatch).await { Ok(result) => result, Err(_elapsed) => { + application::diag!( + "[mcp] ask EXPIRED requester={requester_label} target={arg_target} \ + timeout_ms={} elapsed_ms={}", + self.ask_rendezvous_timeout.as_millis(), + started.elapsed().as_millis(), + ); return Err(JsonRpcError::new( error_codes::INTERNAL_ERROR, "idea_ask_agent : aucune réponse de la cible avant le timeout du rendezvous", @@ -373,6 +420,8 @@ impl McpServer { // twin of the file watcher's publish. `ok` mirrors the dispatch outcome; the // action is the tool name. No-op when no sink is attached. self.publish_processed(&name, result.is_ok()); + // End beacon : longueurs uniquement (jamais le corps), ok/is_error et elapsed — + // ferme la trace ouverte par « begin ». match result { Ok(outcome) => { // The text the agent sees: the reply for an `ask`, else the detail. @@ -380,12 +429,27 @@ impl McpServer { .reply .clone() .unwrap_or_else(|| outcome.detail.clone()); + application::diag!( + "[mcp] tools_call end tool={name} 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)) } // A failed IdeA command is reported as a tool execution error // (`isError: true`) rather than a protocol error: the agent can read // it and decide, and the connection stays healthy. - Err(e) => Ok(tool_result_text(&e.to_string(), true)), + Err(e) => { + let detail = e.to_string(); + application::diag!( + "[mcp] tools_call end tool={name} requester={requester_label} \ + target={arg_target} ok=false is_error=true result_len={} elapsed_ms={}", + detail.len(), + started.elapsed().as_millis(), + ); + Ok(tool_result_text(&detail, true)) + } } } diff --git a/crates/infrastructure/src/orchestrator/mcp/tools.rs b/crates/infrastructure/src/orchestrator/mcp/tools.rs index 1bb6df2..5c5f7c4 100644 --- a/crates/infrastructure/src/orchestrator/mcp/tools.rs +++ b/crates/infrastructure/src/orchestrator/mcp/tools.rs @@ -330,7 +330,33 @@ pub fn map_tool_call( other => return Err(ToolMapError::UnknownTool(other.to_owned())), }; - Ok(request.validate()?) + let command = request.validate()?; + + // Diagnostics mapping beacon (best-effort) pour les deux tools du rendezvous : + // `idea_ask_agent` et `idea_reply`. Longueurs et présence de ticket uniquement — + // jamais le corps `task`/`result` ni de secret. Permet de corréler un `[mcp]` + // begin avec la commande effectivement construite (kind/target/requester). + match name { + "idea_ask_agent" => application::diag!( + "[mcp] mapped kind=ask_agent target={} task_len={} requester={}", + s("target").as_deref().unwrap_or("-"), + s("task").map_or(0, |t| t.len()), + if requester.is_empty() { "-" } else { requester }, + ), + "idea_reply" => application::diag!( + "[mcp] mapped kind=reply result_len={} ticket={} requester={}", + s("result").map_or(0, |r| r.len()), + if s("ticket").is_some() { + "present" + } else { + "absent" + }, + if requester.is_empty() { "-" } else { requester }, + ), + _ => {} + } + + Ok(command) } /// An all-`None` request to spread over; keeps each arm above to just its fields. diff --git a/crates/infrastructure/tests/mcp_server.rs b/crates/infrastructure/tests/mcp_server.rs index 708e8de..8415014 100644 --- a/crates/infrastructure/tests/mcp_server.rs +++ b/crates/infrastructure/tests/mcp_server.rs @@ -1249,3 +1249,110 @@ async fn ask_agent_rendezvous_times_out_with_a_jsonrpc_error() { ); assert!(response.result.is_none(), "error responses carry no result"); } + +// --------------------------------------------------------------------------- +// Diagnostics capture (instrumentation des blocages de délégation) +// +// Best-effort diag sink (`application::diag`) mirrored to a temp file. The sink's +// path is a process-wide OnceLock, so every capture test in this binary points it +// at the SAME canonical file and asserts on lines carrying tokens unique to the +// test (a distinctive target name / agent id), immune to parallelism and re-runs. +// --------------------------------------------------------------------------- + +/// Points the process-wide diag sink at one canonical temp file (idempotent) and +/// returns it. +fn diag_log_path() -> std::path::PathBuf { + let path = std::env::temp_dir().join("idea-diag-mcp-server-test.log"); + application::diag::set_log_path(path.clone()); + path +} + +fn read_diag(path: &std::path::Path) -> String { + std::fs::read_to_string(path).unwrap_or_default() +} + +/// Proves the `[mcp]` rendezvous trace: a wedged `idea_ask_agent` (target that +/// never replies) emits begin → armed → EXPIRED, plus the `[mcp] mapped` beacon. +#[tokio::test] +async fn diag_capture_mcp_ask_begin_armed_expired() { + let log = diag_log_path(); + // Unique target name ⇒ only THIS test's beacons match `target=diag-wedge`. + let contexts = FakeContexts::new(); + let agent_id = contexts.seed_agent("diag-wedge"); + let (service, _mailbox, sessions) = build_service_with_mailbox(contexts); + seed_live_pty( + &sessions, + agent_id, + SessionId::from_uuid(Uuid::from_u128(0xD1A6_0911)), + ); + let server = server(service).with_ask_rendezvous_timeout(std::time::Duration::from_millis(50)); + + let raw = tools_call( + 1, + "idea_ask_agent", + json!({ "target": "diag-wedge", "task": "never answered diag" }), + ); + let response = + tokio::time::timeout(std::time::Duration::from_secs(10), server.handle_raw(&raw)) + .await + .expect("must not hang past the injected timeout") + .expect("reply owed"); + assert!( + response.error.is_some(), + "wedged ask returns a JSON-RPC error" + ); + + let body = read_diag(&log); + assert!( + body.contains("[mcp] mapped kind=ask_agent target=diag-wedge"), + "mapped beacon missing: {body}" + ); + assert!( + body.contains("[mcp] tools_call begin tool=idea_ask_agent") + && body.contains("target=diag-wedge"), + "begin beacon missing: {body}" + ); + assert!( + body.contains("[mcp] ask armed") && body.contains("target=diag-wedge"), + "armed beacon missing: {body}" + ); + assert!( + body.contains("[mcp] ask EXPIRED") && body.contains("target=diag-wedge"), + "EXPIRED beacon missing: {body}" + ); +} + +/// Proves the `[rendezvous] idea_reply received` beacon fires BEFORE correlation and +/// is followed by `UNMATCHED` when no in-flight ask exists for the emitter. +#[tokio::test] +async fn diag_capture_idea_reply_received_then_unmatched() { + let log = diag_log_path(); + let contexts = FakeContexts::new(); + let (service, _mailbox, _sessions) = build_service_with_mailbox(contexts); + + // A distinctive emitter id ⇒ match only this test's reply beacons in the log. + let from = AgentId::from_uuid(Uuid::from_u128(0xD1A6_0000_0000_0000_0000_0000_0000_0042)); + let cmd = domain::OrchestratorCommand::Reply { + from, + ticket: None, + result: "orphan reply".to_owned(), + }; + // No ask is in flight for `from` ⇒ dispatch returns an Invalid error (unmatched). + let outcome = service.dispatch(&project(), cmd).await; + assert!(outcome.is_err(), "an orphan idea_reply is a typed error"); + + let body = read_diag(&log); + let from = from.to_string(); + assert!( + body.contains(&format!( + "[rendezvous] idea_reply received: from agent {from}" + )), + "received beacon missing: {body}" + ); + assert!( + body.contains(&format!( + "[rendezvous] idea_reply UNMATCHED: from agent {from}" + )), + "UNMATCHED beacon missing: {body}" + ); +}