feat(diag): instrumente les blocages de délégation idea_ask_agent
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 <noreply@anthropic.com>
This commit is contained in:
@ -77,16 +77,21 @@ impl InMemoryMailbox {
|
||||
impl AgentMailbox for InMemoryMailbox {
|
||||
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
|
||||
let (tx, rx) = oneshot::channel::<String>();
|
||||
{
|
||||
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();
|
||||
|
||||
Reference in New Issue
Block a user