merge(diag): intègre l'instrumentation des blocages idea_ask_agent dans develop
Logs diag! sur le canal de délégation inter-agents (application + infrastructure input/mailbox/mcp), vérifiés verts par Main (mcp_server 22, mailbox 14, input 35, tools 16, orchestrator_service 49 ; cargo check app-tauri OK ; fmt OK). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -890,6 +890,10 @@ impl OrchestratorService {
|
|||||||
g.would_cycle(from, agent_id)
|
g.would_cycle(from, agent_id)
|
||||||
};
|
};
|
||||||
if cycles {
|
if cycles {
|
||||||
|
crate::diag!(
|
||||||
|
"[rendezvous] cycle refused: requester={from} -> target={target} \
|
||||||
|
(agent {agent_id}) (deadlock évité, aucun enqueue)",
|
||||||
|
);
|
||||||
return Err(AppError::Invalid(format!(
|
return Err(AppError::Invalid(format!(
|
||||||
"délégation ré-entrante refusée : demander à l'agent '{target}' créerait \
|
"délégation ré-entrante refusée : demander à l'agent '{target}' créerait \
|
||||||
un cycle d'attente inter-agents (deadlock évité)"
|
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
|
// 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.
|
// pour TOUT le tour (enqueue → réponse). RAII : tombe sur chaque early-return.
|
||||||
let lock = self.ask_lock_for(&agent_id);
|
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 {
|
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) => {
|
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));
|
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).
|
// des deux ⇒ pas de gate (livraison immédiate, sinon blocage indéfini).
|
||||||
let gate_cold_start =
|
let gate_cold_start =
|
||||||
cold_launch && (prompt_pattern.as_ref().is_some_and(|p| !p.is_empty()) || has_mcp);
|
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 {
|
if gate_cold_start {
|
||||||
input.mark_starting(agent_id);
|
input.mark_starting(agent_id);
|
||||||
}
|
}
|
||||||
@ -1011,8 +1045,8 @@ impl OrchestratorService {
|
|||||||
let started = Instant::now();
|
let started = Instant::now();
|
||||||
crate::diag!(
|
crate::diag!(
|
||||||
"[rendezvous] ask started: requester={} -> target={target} (agent {agent_id}) \
|
"[rendezvous] ask started: requester={} -> target={target} (agent {agent_id}) \
|
||||||
ticket={ticket_id} cold_launch={cold_launch} gate_cold_start={gate_cold_start} \
|
conversation={conversation_id} ticket={ticket_id} cold_launch={cold_launch} \
|
||||||
turn_timeout_ms={}",
|
gate_cold_start={gate_cold_start} turn_timeout_ms={}",
|
||||||
requester.map_or_else(|| "user".to_owned(), |a| a.to_string()),
|
requester.map_or_else(|| "user".to_owned(), |a| a.to_string()),
|
||||||
turn_timeout.as_millis(),
|
turn_timeout.as_millis(),
|
||||||
);
|
);
|
||||||
@ -1155,6 +1189,18 @@ impl OrchestratorService {
|
|||||||
let busy_guard =
|
let busy_guard =
|
||||||
BusyTurnGuard::new(Arc::clone(input), Arc::clone(mailbox), agent_id, ticket_id);
|
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
|
// 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
|
// **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.
|
// **première** issue : le tour structuré OU un `idea_reply` explicite.
|
||||||
@ -1165,9 +1211,24 @@ impl OrchestratorService {
|
|||||||
biased;
|
biased;
|
||||||
// Issue déterministe : la session a rendu son `Final`.
|
// Issue déterministe : la session a rendu son `Final`.
|
||||||
drained = drain => match drained {
|
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.
|
// 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.
|
// Issue alternative : un `idea_reply` explicite a résolu le ticket d'abord.
|
||||||
replied = pending => match replied {
|
replied = pending => match replied {
|
||||||
@ -1175,10 +1236,21 @@ impl OrchestratorService {
|
|||||||
// La session draine encore en arrière-plan ; on s'assure que la FIFO
|
// 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é.
|
// avance même si le `Final` n'est pas encore observé.
|
||||||
input.mark_idle(agent_id);
|
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
|
content
|
||||||
}
|
}
|
||||||
// Canal fermé : le garde fait `cancel_head` + `mark_idle` au Drop.
|
// Canal fermé : le garde fait `cancel_head` + `mark_idle` au Drop.
|
||||||
Err(_cancelled) => {
|
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!(
|
return Err(AppError::Process(format!(
|
||||||
"agent {target} : canal de réponse fermé avant un résultat"
|
"agent {target} : canal de réponse fermé avant un résultat"
|
||||||
)));
|
)));
|
||||||
@ -1413,7 +1485,18 @@ impl OrchestratorService {
|
|||||||
ticket: Option<TicketId>,
|
ticket: Option<TicketId>,
|
||||||
result: String,
|
result: String,
|
||||||
) -> Result<OrchestratorOutcome, AppError> {
|
) -> Result<OrchestratorOutcome, AppError> {
|
||||||
|
// 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(|| {
|
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(
|
AppError::Invalid(
|
||||||
"idea_reply n'est pas disponible : file inter-agents non câblée".to_owned(),
|
"idea_reply n'est pas disponible : file inter-agents non câblée".to_owned(),
|
||||||
)
|
)
|
||||||
|
|||||||
@ -147,7 +147,7 @@ impl BusyTracker {
|
|||||||
/// appel, l'`enqueue` publie la `DelegationReady` immédiatement (chemin chaud,
|
/// appel, l'`enqueue` publie la `DelegationReady` immédiatement (chemin chaud,
|
||||||
/// zéro régression).
|
/// zéro régression).
|
||||||
fn mark_starting(&self, agent: AgentId) {
|
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);
|
self.lock_starting().insert(agent);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -235,6 +235,10 @@ impl BusyTracker {
|
|||||||
.collect()
|
.collect()
|
||||||
};
|
};
|
||||||
for agent in newly_stalled {
|
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);
|
self.publish_liveness(agent, AgentLiveness::Stalled);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -264,10 +268,10 @@ impl BusyTracker {
|
|||||||
let mut busy = self.lock();
|
let mut busy = self.lock();
|
||||||
let entry = busy.entry(agent).or_insert(AgentBusyState::Idle);
|
let entry = busy.entry(agent).or_insert(AgentBusyState::Idle);
|
||||||
if entry.is_busy() {
|
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
|
false
|
||||||
} else {
|
} else {
|
||||||
eprintln!("[input-mediator] turn started agent={agent} state={state:?}");
|
application::diag!("[input-mediator] start_turn started agent={agent} state={state:?}");
|
||||||
*entry = state;
|
*entry = state;
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
@ -285,15 +289,17 @@ impl BusyTracker {
|
|||||||
Some(sink) => match sink(agent, d) {
|
Some(sink) => match sink(agent, d) {
|
||||||
Some(d) => d,
|
Some(d) => d,
|
||||||
None => {
|
None => {
|
||||||
eprintln!("[input-mediator] delegation handled by headless sink agent={agent}");
|
application::diag!(
|
||||||
|
"[input-mediator] delegationReady handled by headless sink agent={agent}"
|
||||||
|
);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
None => d,
|
None => d,
|
||||||
};
|
};
|
||||||
if let Some(events) = &self.events {
|
if let Some(events) = &self.events {
|
||||||
eprintln!(
|
application::diag!(
|
||||||
"[input-mediator] publishing DelegationReady agent={agent} ticket={} text_bytes={} submit_sequence={} submit_delay_ms={:?}",
|
"[input-mediator] delegationReady -> write-portal agent={agent} ticket={} text_bytes={} submit_sequence={} submit_delay_ms={:?}",
|
||||||
d.ticket,
|
d.ticket,
|
||||||
d.text.as_bytes().len(),
|
d.text.as_bytes().len(),
|
||||||
describe_submit_sequence(d.submit_sequence.as_deref()),
|
describe_submit_sequence(d.submit_sequence.as_deref()),
|
||||||
@ -323,13 +329,24 @@ impl BusyTracker {
|
|||||||
let deferred = self.lock_deferred().remove(&agent);
|
let deferred = self.lock_deferred().remove(&agent);
|
||||||
match deferred {
|
match deferred {
|
||||||
Some(d) => {
|
Some(d) => {
|
||||||
eprintln!(
|
application::diag!(
|
||||||
"[input-mediator] prompt-ready released deferred delegation agent={agent} ticket={}",
|
"[input-mediator] prompt_ready released deferred delegation agent={agent} ticket={}",
|
||||||
d.ticket
|
d.ticket
|
||||||
);
|
);
|
||||||
self.publish_deferred(agent, d);
|
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) {
|
fn release_cold_start(&self, agent: AgentId) {
|
||||||
self.lock_starting().remove(&agent);
|
self.lock_starting().remove(&agent);
|
||||||
if let Some(d) = self.lock_deferred().remove(&agent) {
|
if let Some(d) = self.lock_deferred().remove(&agent) {
|
||||||
eprintln!(
|
application::diag!(
|
||||||
"[input-mediator] mcp-ready released deferred delegation agent={agent} ticket={}",
|
"[input-mediator] release_cold_start (mcp-ready) released deferred delegation agent={agent} ticket={}",
|
||||||
d.ticket
|
d.ticket
|
||||||
);
|
);
|
||||||
self.publish_deferred(agent, d);
|
self.publish_deferred(agent, d);
|
||||||
@ -359,7 +376,7 @@ impl BusyTracker {
|
|||||||
.is_some_and(|s| s.is_busy())
|
.is_some_and(|s| s.is_busy())
|
||||||
};
|
};
|
||||||
if was_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 {
|
if let Some(events) = &self.events {
|
||||||
events.publish(DomainEvent::AgentBusyChanged {
|
events.publish(DomainEvent::AgentBusyChanged {
|
||||||
agent_id: agent,
|
agent_id: agent,
|
||||||
@ -837,6 +854,9 @@ impl InputMediator for MediatedInbox {
|
|||||||
submit_delay_ms: submit.delay_ms,
|
submit_delay_ms: submit.delay_ms,
|
||||||
};
|
};
|
||||||
if self.tracker.lock_starting().contains(&agent) {
|
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);
|
self.tracker.lock_deferred().insert(agent, deferred);
|
||||||
} else {
|
} else {
|
||||||
self.tracker.publish_deferred(agent, deferred);
|
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
|
// 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
|
// live stream to interrupt). The busy state is left untouched: it returns to
|
||||||
// Idle through `mark_idle` (prompt-ready / explicit signal), not here.
|
// 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(pty) = &self.pty {
|
||||||
if let Some(handle) = self.handles().get(&agent).cloned() {
|
if let Some(handle) = self.handles().get(&agent).cloned() {
|
||||||
let _ = pty.write(&handle, b"\x1b");
|
let _ = pty.write(&handle, b"\x1b");
|
||||||
|
|||||||
@ -77,16 +77,21 @@ impl InMemoryMailbox {
|
|||||||
impl AgentMailbox for InMemoryMailbox {
|
impl AgentMailbox for InMemoryMailbox {
|
||||||
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
|
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
|
||||||
let (tx, rx) = oneshot::channel::<String>();
|
let (tx, rx) = oneshot::channel::<String>();
|
||||||
{
|
let ticket_id = ticket.id;
|
||||||
|
let (depth_before, depth_after) = {
|
||||||
let mut queues = self
|
let mut queues = self
|
||||||
.queues
|
.queues
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||||
queues
|
let queue = queues.entry(agent).or_default();
|
||||||
.entry(agent)
|
let before = queue.len();
|
||||||
.or_default()
|
queue.push_back(Slot { ticket, reply: tx });
|
||||||
.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
|
// 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
|
// 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.
|
// 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
|
// 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
|
// is non-blocking). If the receiver already went away (caller timed out), the
|
||||||
// ticket is still correctly retired from the head — the queue advances.
|
// ticket is still correctly retired from the head — the queue advances.
|
||||||
let slot = {
|
let (slot, depth_before, depth_after) = {
|
||||||
let mut queues = self
|
let mut queues = self
|
||||||
.queues
|
.queues
|
||||||
.lock()
|
.lock()
|
||||||
@ -107,9 +112,21 @@ impl AgentMailbox for InMemoryMailbox {
|
|||||||
let queue = queues
|
let queue = queues
|
||||||
.get_mut(&agent)
|
.get_mut(&agent)
|
||||||
.filter(|q| !q.is_empty())
|
.filter(|q| !q.is_empty())
|
||||||
.ok_or(MailboxError::NoPendingRequest(agent))?;
|
.ok_or_else(|| {
|
||||||
queue.pop_front().expect("non-empty queue has a head")
|
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:
|
// A dropped receiver (the awaiting caller timed out and went away) is fine:
|
||||||
// the reply is simply discarded, the head ticket is already retired.
|
// the reply is simply discarded, the head ticket is already retired.
|
||||||
let _ = slot.reply.send(result);
|
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
|
// 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
|
// correlation), then send outside any await. A missing match is a typed
|
||||||
// NoPendingRequest — never a panic.
|
// NoPendingRequest — never a panic.
|
||||||
let slot = {
|
let (slot, depth_before, depth_after) = {
|
||||||
let mut queues = self
|
let mut queues = self
|
||||||
.queues
|
.queues
|
||||||
.lock()
|
.lock()
|
||||||
@ -133,33 +150,67 @@ impl AgentMailbox for InMemoryMailbox {
|
|||||||
let queue = queues
|
let queue = queues
|
||||||
.get_mut(&agent)
|
.get_mut(&agent)
|
||||||
.filter(|q| !q.is_empty())
|
.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
|
let pos = queue
|
||||||
.iter()
|
.iter()
|
||||||
.position(|s| s.ticket.id == ticket_id)
|
.position(|s| s.ticket.id == ticket_id)
|
||||||
.ok_or(MailboxError::NoPendingRequest(agent))?;
|
.ok_or_else(|| {
|
||||||
queue.remove(pos).expect("position just found is in range")
|
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);
|
let _ = slot.reply.send(result);
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) {
|
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) {
|
||||||
|
let (depth_before, depth_after, retired) = {
|
||||||
let mut queues = self
|
let mut queues = self
|
||||||
.queues
|
.queues
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||||
if let Some(queue) = queues.get_mut(&agent) {
|
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
|
// 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)
|
// since changed (already resolved, or someone else's ticket now in front)
|
||||||
// must not be dropped. Idempotent and positional.
|
// 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();
|
queue.pop_front();
|
||||||
}
|
true
|
||||||
|
} else {
|
||||||
|
false
|
||||||
|
};
|
||||||
|
let after = queue.len();
|
||||||
if queue.is_empty() {
|
if queue.is_empty() {
|
||||||
queues.remove(&agent);
|
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");
|
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]
|
#[test]
|
||||||
fn snapshot_updates_after_cancel_head() {
|
fn snapshot_updates_after_cancel_head() {
|
||||||
let mb = InMemoryMailbox::new();
|
let mb = InMemoryMailbox::new();
|
||||||
|
|||||||
@ -18,7 +18,7 @@
|
|||||||
//! duplicated here.
|
//! duplicated here.
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use application::OrchestratorService;
|
use application::OrchestratorService;
|
||||||
use domain::{DomainEvent, OrchestrationSource, Project};
|
use domain::{DomainEvent, OrchestrationSource, Project};
|
||||||
@ -345,10 +345,43 @@ impl McpServer {
|
|||||||
.to_owned();
|
.to_owned();
|
||||||
let arguments = params.get("arguments").cloned().unwrap_or(json!({}));
|
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
|
// `idea_reply` needs the connected peer's identity as `from`; every other tool
|
||||||
// ignores `requester`. The handshake-provided requester is the source of truth.
|
// ignores `requester`. The handshake-provided requester is the source of truth.
|
||||||
let command =
|
let command = match tools::map_tool_call(&name, &arguments, &self.requester) {
|
||||||
tools::map_tool_call(&name, &arguments, &self.requester).map_err(map_err_to_jsonrpc)?;
|
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
|
// `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
|
// (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.
|
// other tool dispatches unbounded — they never rendezvous.
|
||||||
let dispatch = self.service.dispatch(&self.project, command);
|
let dispatch = self.service.dispatch(&self.project, command);
|
||||||
let result = if name == "idea_ask_agent" {
|
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 {
|
match tokio::time::timeout(self.ask_rendezvous_timeout, dispatch).await {
|
||||||
Ok(result) => result,
|
Ok(result) => result,
|
||||||
Err(_elapsed) => {
|
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(
|
return Err(JsonRpcError::new(
|
||||||
error_codes::INTERNAL_ERROR,
|
error_codes::INTERNAL_ERROR,
|
||||||
"idea_ask_agent : aucune réponse de la cible avant le timeout du rendezvous",
|
"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
|
// 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.
|
// action is the tool name. No-op when no sink is attached.
|
||||||
self.publish_processed(&name, result.is_ok());
|
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 {
|
match result {
|
||||||
Ok(outcome) => {
|
Ok(outcome) => {
|
||||||
// The text the agent sees: the reply for an `ask`, else the detail.
|
// The text the agent sees: the reply for an `ask`, else the detail.
|
||||||
@ -380,12 +429,27 @@ impl McpServer {
|
|||||||
.reply
|
.reply
|
||||||
.clone()
|
.clone()
|
||||||
.unwrap_or_else(|| outcome.detail.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))
|
Ok(tool_result_text(&text, false))
|
||||||
}
|
}
|
||||||
// A failed IdeA command is reported as a tool execution error
|
// A failed IdeA command is reported as a tool execution error
|
||||||
// (`isError: true`) rather than a protocol error: the agent can read
|
// (`isError: true`) rather than a protocol error: the agent can read
|
||||||
// it and decide, and the connection stays healthy.
|
// 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))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -330,7 +330,33 @@ pub fn map_tool_call(
|
|||||||
other => return Err(ToolMapError::UnknownTool(other.to_owned())),
|
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.
|
/// An all-`None` request to spread over; keeps each arm above to just its fields.
|
||||||
|
|||||||
@ -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");
|
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}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user