fix(delegation): résout idea_ask_agent quand la cible revient au prompt sans idea_reply
Introduit le seam TurnResolution : sur prompt-ready, après un délai de grâce, la délégation en attente est résolue de façon typée plutôt que de rester bloquée indéfiniment (jusquà 24h). Couvre domain (mailbox), infrastructure (mailbox/input) et application (orchestrator/error), avec tests étendus. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -57,6 +57,18 @@ pub enum AppError {
|
|||||||
node_id: NodeId,
|
node_id: NodeId,
|
||||||
},
|
},
|
||||||
|
|
||||||
|
/// The delegation target finished its turn and **returned to its prompt without
|
||||||
|
/// calling `idea_reply`**, so the rendezvous could produce no answer. Surfaced
|
||||||
|
/// promptly (after a short grace window) instead of blocking the caller until the
|
||||||
|
/// long safety timeout. **Retryable**: the caller may re-ask, ideally reminding the
|
||||||
|
/// target to render its result via `idea_reply` (never plain text). Carries the
|
||||||
|
/// target's display name.
|
||||||
|
#[error(
|
||||||
|
"agent {0} returned to its prompt without calling idea_reply (no answer was \
|
||||||
|
rendered); retry the request and ensure the agent replies via idea_reply"
|
||||||
|
)]
|
||||||
|
TargetReturnedNoReply(String),
|
||||||
|
|
||||||
/// An unexpected internal error.
|
/// An unexpected internal error.
|
||||||
#[error("internal error: {0}")]
|
#[error("internal error: {0}")]
|
||||||
Internal(String),
|
Internal(String),
|
||||||
@ -76,6 +88,7 @@ impl AppError {
|
|||||||
Self::Git(_) => "GIT",
|
Self::Git(_) => "GIT",
|
||||||
Self::Remote(_) => "REMOTE",
|
Self::Remote(_) => "REMOTE",
|
||||||
Self::AgentAlreadyRunning { .. } => "AGENT_ALREADY_RUNNING",
|
Self::AgentAlreadyRunning { .. } => "AGENT_ALREADY_RUNNING",
|
||||||
|
Self::TargetReturnedNoReply(_) => "TARGET_RETURNED_NO_REPLY",
|
||||||
Self::Internal(_) => "INTERNAL",
|
Self::Internal(_) => "INTERNAL",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -22,7 +22,7 @@ use tokio::sync::Mutex as AsyncMutex;
|
|||||||
use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef, WaitForGraph};
|
use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef, WaitForGraph};
|
||||||
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
|
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
|
||||||
use domain::input::{InputMediator, InputSource, SubmitConfig};
|
use domain::input::{InputMediator, InputSource, SubmitConfig};
|
||||||
use domain::mailbox::{Ticket, TicketId};
|
use domain::mailbox::{Ticket, TicketId, TurnResolution};
|
||||||
use domain::ports::{Clock, EventBus, ProfileStore, PtyHandle};
|
use domain::ports::{Clock, EventBus, ProfileStore, PtyHandle};
|
||||||
use domain::project::ProjectPath;
|
use domain::project::ProjectPath;
|
||||||
use domain::{
|
use domain::{
|
||||||
@ -1063,7 +1063,7 @@ impl OrchestratorService {
|
|||||||
// 3. Attendre la réponse, bornée. Timeout/canal fermé ⇒ le garde retire le ticket
|
// 3. Attendre la réponse, bornée. Timeout/canal fermé ⇒ le garde retire le ticket
|
||||||
// (cible laissée vivante) et la ramène `Idle` au Drop ; renvoie une erreur typée.
|
// (cible laissée vivante) et la ramène `Idle` au Drop ; renvoie une erreur typée.
|
||||||
match tokio::time::timeout(turn_timeout, pending).await {
|
match tokio::time::timeout(turn_timeout, pending).await {
|
||||||
Ok(Ok(result)) => {
|
Ok(Ok(TurnResolution::Replied(result))) => {
|
||||||
// Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de
|
// Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de
|
||||||
// déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est
|
// déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est
|
||||||
// elle qui a rendu le tour) ; même conversation que le Prompt.
|
// elle qui a rendu le tour) ; même conversation que le Prompt.
|
||||||
@ -1093,6 +1093,21 @@ impl OrchestratorService {
|
|||||||
);
|
);
|
||||||
Ok(self.reply_outcome(agent_id, &target, result))
|
Ok(self.reply_outcome(agent_id, &target, result))
|
||||||
}
|
}
|
||||||
|
// La cible est revenue à son prompt SANS `idea_reply` : la fenêtre de grâce a
|
||||||
|
// expiré et le médiateur a complété le tour « sans réponse ». Le ticket est
|
||||||
|
// déjà retiré (`complete_without_reply`) et la cible déjà `Idle` (prompt-ready
|
||||||
|
// `mark_idle`) ⇒ on **désarme** le garde (pas de `cancel_head`/`mark_idle`
|
||||||
|
// redondant ni de beacon « busy-guard freed » trompeur) et on renvoie une
|
||||||
|
// erreur typée claire/retryable, en ~G au lieu du timeout long.
|
||||||
|
Ok(Ok(TurnResolution::ReturnedToPromptNoReply)) => {
|
||||||
|
busy_guard.disarm();
|
||||||
|
crate::diag!(
|
||||||
|
"[rendezvous] ask returned-to-prompt-no-reply: target={target} \
|
||||||
|
(agent {agent_id}) ticket={ticket_id} after_ms={}",
|
||||||
|
started.elapsed().as_millis(),
|
||||||
|
);
|
||||||
|
Err(AppError::TargetReturnedNoReply(target))
|
||||||
|
}
|
||||||
// Erreur / timeout : on laisse le garde faire `cancel_head` + `mark_idle` au
|
// Erreur / timeout : on laisse le garde faire `cancel_head` + `mark_idle` au
|
||||||
// Drop (retrait des `cancel_head` redondants — `cancel_head` reste idempotent).
|
// Drop (retrait des `cancel_head` redondants — `cancel_head` reste idempotent).
|
||||||
Ok(Err(_cancelled)) => {
|
Ok(Err(_cancelled)) => {
|
||||||
@ -1232,7 +1247,7 @@ impl OrchestratorService {
|
|||||||
},
|
},
|
||||||
// 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 {
|
||||||
Ok(content) => {
|
Ok(TurnResolution::Replied(content)) => {
|
||||||
// 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);
|
||||||
@ -1244,6 +1259,19 @@ impl OrchestratorService {
|
|||||||
);
|
);
|
||||||
content
|
content
|
||||||
}
|
}
|
||||||
|
// Retour au prompt sans `idea_reply` : très rare sur le chemin structuré
|
||||||
|
// (la cible n'a pas de PTY/watcher), mais on traite la résolution comme
|
||||||
|
// sur le chemin PTY — erreur typée claire/retryable, garde désarmé.
|
||||||
|
Ok(TurnResolution::ReturnedToPromptNoReply) => {
|
||||||
|
input.mark_idle(agent_id);
|
||||||
|
busy_guard.disarm();
|
||||||
|
crate::diag!(
|
||||||
|
"[rendezvous] ask returned-to-prompt-no-reply (structured): \
|
||||||
|
target={target} (agent {agent_id}) ticket={ticket_id} after_ms={}",
|
||||||
|
started.elapsed().as_millis(),
|
||||||
|
);
|
||||||
|
return Err(AppError::TargetReturnedNoReply(target.to_owned()));
|
||||||
|
}
|
||||||
// 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!(
|
crate::diag!(
|
||||||
|
|||||||
@ -861,7 +861,7 @@ use domain::conversation::{
|
|||||||
SessionRef,
|
SessionRef,
|
||||||
};
|
};
|
||||||
use domain::input::{AgentBusyState, InputMediator};
|
use domain::input::{AgentBusyState, InputMediator};
|
||||||
use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId};
|
use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId, TurnResolution};
|
||||||
use tokio::time::{timeout, Duration};
|
use tokio::time::{timeout, Duration};
|
||||||
|
|
||||||
/// Safety guard: no awaited rendezvous in these tests should ever exceed this.
|
/// Safety guard: no awaited rendezvous in these tests should ever exceed this.
|
||||||
@ -872,7 +872,8 @@ const TEST_GUARD: Duration = Duration::from_secs(10);
|
|||||||
/// would be a dependency cycle). Per-agent FIFO of `(Ticket, oneshot::Sender)`.
|
/// would be a dependency cycle). Per-agent FIFO of `(Ticket, oneshot::Sender)`.
|
||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
struct TestMailbox {
|
struct TestMailbox {
|
||||||
queues: Mutex<HashMap<AgentId, VecDeque<(Ticket, tokio::sync::oneshot::Sender<String>)>>>,
|
queues:
|
||||||
|
Mutex<HashMap<AgentId, VecDeque<(Ticket, tokio::sync::oneshot::Sender<TurnResolution>)>>>,
|
||||||
}
|
}
|
||||||
impl TestMailbox {
|
impl TestMailbox {
|
||||||
fn new() -> Self {
|
fn new() -> Self {
|
||||||
@ -899,7 +900,7 @@ impl TestMailbox {
|
|||||||
}
|
}
|
||||||
impl AgentMailbox for TestMailbox {
|
impl AgentMailbox for TestMailbox {
|
||||||
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
|
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
|
||||||
let (tx, rx) = tokio::sync::oneshot::channel::<String>();
|
let (tx, rx) = tokio::sync::oneshot::channel::<TurnResolution>();
|
||||||
self.queues
|
self.queues
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
@ -919,7 +920,7 @@ impl AgentMailbox for TestMailbox {
|
|||||||
.ok_or(MailboxError::NoPendingRequest(agent))?;
|
.ok_or(MailboxError::NoPendingRequest(agent))?;
|
||||||
queue.pop_front().expect("non-empty")
|
queue.pop_front().expect("non-empty")
|
||||||
};
|
};
|
||||||
let _ = slot.1.send(result);
|
let _ = slot.1.send(TurnResolution::Replied(result));
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
fn resolve_ticket(
|
fn resolve_ticket(
|
||||||
@ -940,7 +941,7 @@ impl AgentMailbox for TestMailbox {
|
|||||||
.ok_or(MailboxError::NoPendingRequest(agent))?;
|
.ok_or(MailboxError::NoPendingRequest(agent))?;
|
||||||
queue.remove(pos).expect("found position")
|
queue.remove(pos).expect("found position")
|
||||||
};
|
};
|
||||||
let _ = slot.1.send(result);
|
let _ = slot.1.send(TurnResolution::Replied(result));
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) {
|
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) {
|
||||||
@ -951,6 +952,20 @@ impl AgentMailbox for TestMailbox {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
fn complete_without_reply(&self, agent: AgentId, ticket_id: TicketId) {
|
||||||
|
// Mirror the production adapter: head-only, idempotent, fire-and-forget-safe.
|
||||||
|
let mut q = self.queues.lock().unwrap();
|
||||||
|
if let Some(queue) = q.get_mut(&agent) {
|
||||||
|
if queue.front().map(|(t, _)| t.id) != Some(ticket_id) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if queue.front().is_some_and(|(_, tx)| tx.is_closed()) {
|
||||||
|
return; // receiver gone (human submit / timed-out caller): preserve head.
|
||||||
|
}
|
||||||
|
let (_, tx) = queue.pop_front().expect("head just matched");
|
||||||
|
let _ = tx.send(TurnResolution::ReturnedToPromptNoReply);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A delivering [`InputMediator`] for the application tests: it composes the
|
/// A delivering [`InputMediator`] for the application tests: it composes the
|
||||||
@ -1287,6 +1302,63 @@ async fn ask_live_pty_target_writes_task_and_returns_reply() {
|
|||||||
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "ticket drained on reply");
|
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "ticket drained on reply");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The target returned to its prompt **without** `idea_reply`: the mediator's grace
|
||||||
|
/// window expired and completed the turn "no reply". The awaiting ask must error out
|
||||||
|
/// promptly with the typed, retryable [`AppError::TargetReturnedNoReply`] — never hang
|
||||||
|
/// until the long safety timeout. (The grace **timer** lives in the infrastructure
|
||||||
|
/// mediator; here we drive the resulting mailbox completion directly.)
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ask_target_returns_to_prompt_without_reply_is_a_typed_error() {
|
||||||
|
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||||
|
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
|
||||||
|
seed_live_pty(&fx.sessions, aid(1), sid(800));
|
||||||
|
|
||||||
|
let svc = Arc::clone(&fx.service);
|
||||||
|
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
|
||||||
|
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
|
||||||
|
|
||||||
|
// Simulate the grace-window completion (target back at prompt, no idea_reply).
|
||||||
|
let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
|
||||||
|
fx.mailbox.complete_without_reply(aid(1), ticket);
|
||||||
|
|
||||||
|
let err = timeout(TEST_GUARD, ask)
|
||||||
|
.await
|
||||||
|
.expect("ask completes promptly, not after the long timeout")
|
||||||
|
.expect("join ok")
|
||||||
|
.expect_err("a no-reply turn is a typed error");
|
||||||
|
assert_eq!(err.code(), "TARGET_RETURNED_NO_REPLY", "got {err:?}");
|
||||||
|
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "head retired by completion");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An `idea_reply` that lands **first** still wins over a (later) no-reply completion:
|
||||||
|
/// the completion then finds the head gone and is a no-op (idempotent race).
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ask_reply_wins_then_late_completion_is_noop() {
|
||||||
|
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||||
|
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
|
||||||
|
seed_live_pty(&fx.sessions, aid(1), sid(800));
|
||||||
|
|
||||||
|
let svc = Arc::clone(&fx.service);
|
||||||
|
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
|
||||||
|
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
|
||||||
|
|
||||||
|
let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
|
||||||
|
// idea_reply resolves the head first…
|
||||||
|
fx.service
|
||||||
|
.dispatch(&project(), reply_cmd(aid(1), "the real answer"))
|
||||||
|
.await
|
||||||
|
.expect("reply ok");
|
||||||
|
// …then a late grace completion for the same ticket is a no-op.
|
||||||
|
fx.mailbox.complete_without_reply(aid(1), ticket);
|
||||||
|
|
||||||
|
let out = timeout(TEST_GUARD, ask)
|
||||||
|
.await
|
||||||
|
.expect("ask completes")
|
||||||
|
.expect("join ok")
|
||||||
|
.expect("ask ok");
|
||||||
|
assert_eq!(out.reply.as_deref(), Some("the real answer"));
|
||||||
|
}
|
||||||
|
|
||||||
// --- Lot E1: auto-memory harvest on a successful ask reply -----------------
|
// --- Lot E1: auto-memory harvest on a successful ask reply -----------------
|
||||||
|
|
||||||
/// A reply carrying an `idea-memory` block is harvested **after** the reply
|
/// A reply carrying an `idea-memory` block is harvested **after** the reply
|
||||||
|
|||||||
@ -13,7 +13,7 @@ use application::{
|
|||||||
TerminalSessions, TicketWorkSource, TicketWorkStatus,
|
TerminalSessions, TicketWorkSource, TicketWorkStatus,
|
||||||
};
|
};
|
||||||
use domain::mailbox::{
|
use domain::mailbox::{
|
||||||
AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket,
|
AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, TurnResolution,
|
||||||
};
|
};
|
||||||
use domain::ports::{
|
use domain::ports::{
|
||||||
AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError,
|
AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError,
|
||||||
@ -123,7 +123,7 @@ impl FakeInput {
|
|||||||
|
|
||||||
impl InputMediator for FakeInput {
|
impl InputMediator for FakeInput {
|
||||||
fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply {
|
fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply {
|
||||||
let fut: Pin<Box<dyn Future<Output = Result<String, MailboxError>> + Send>> =
|
let fut: Pin<Box<dyn Future<Output = Result<TurnResolution, MailboxError>> + Send>> =
|
||||||
Box::pin(async { Err(MailboxError::Cancelled) });
|
Box::pin(async { Err(MailboxError::Cancelled) });
|
||||||
PendingReply::new(fut)
|
PendingReply::new(fut)
|
||||||
}
|
}
|
||||||
|
|||||||
@ -191,6 +191,28 @@ impl Ticket {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The outcome of a delegated turn, delivered to the awaiting [`PendingReply`].
|
||||||
|
///
|
||||||
|
/// A turn ends in **exactly one** of two ways (besides the channel closing, which is
|
||||||
|
/// [`MailboxError::Cancelled`]):
|
||||||
|
/// - [`TurnResolution::Replied`] — the target rendered a result via `idea_reply`
|
||||||
|
/// ([`AgentMailbox::resolve`] / [`AgentMailbox::resolve_ticket`]). The normal
|
||||||
|
/// success: the caller receives the answer inline.
|
||||||
|
/// - [`TurnResolution::ReturnedToPromptNoReply`] — the target **finished its turn and
|
||||||
|
/// returned to its prompt without calling `idea_reply`**. The rendezvous can produce
|
||||||
|
/// no payload, so instead of leaving the caller blocked until the long safety
|
||||||
|
/// timeout, the mailbox wakes it with this variant
|
||||||
|
/// ([`AgentMailbox::complete_without_reply`]) so the application layer can surface a
|
||||||
|
/// clear, retryable error promptly.
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
|
pub enum TurnResolution {
|
||||||
|
/// The target rendered a result via `idea_reply`. Carries the answer.
|
||||||
|
Replied(String),
|
||||||
|
/// The target returned to its prompt (turn ended) **without** an `idea_reply`.
|
||||||
|
/// No payload exists; the caller must surface a typed, retryable error.
|
||||||
|
ReturnedToPromptNoReply,
|
||||||
|
}
|
||||||
|
|
||||||
/// Errors raised by the [`AgentMailbox`] port.
|
/// Errors raised by the [`AgentMailbox`] port.
|
||||||
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
|
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
|
||||||
pub enum MailboxError {
|
pub enum MailboxError {
|
||||||
@ -206,28 +228,32 @@ pub enum MailboxError {
|
|||||||
Cancelled,
|
Cancelled,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A handle the caller awaits to receive a target agent's reply.
|
/// A handle the caller awaits to receive a target agent's turn outcome.
|
||||||
///
|
///
|
||||||
/// Returned by [`AgentMailbox::enqueue`]. Awaiting it yields the `String` result a
|
/// Returned by [`AgentMailbox::enqueue`]. Awaiting it yields a [`TurnResolution`] (a
|
||||||
/// later [`AgentMailbox::resolve`] feeds into this ticket's slot, or
|
/// [`TurnResolution::Replied`] result fed by a later [`AgentMailbox::resolve`], or a
|
||||||
/// [`MailboxError::Cancelled`] if the reply channel closes first. The future is
|
/// [`TurnResolution::ReturnedToPromptNoReply`] fed by
|
||||||
/// **opaque** (the adapter builds it from its one-shot receiver): the domain stays
|
/// [`AgentMailbox::complete_without_reply`] when the target returns to its prompt
|
||||||
/// free of any concrete channel type, and the await point lives in the application
|
/// without replying), or [`MailboxError::Cancelled`] if the reply channel closes
|
||||||
/// layer's `ask_agent`.
|
/// first. The future is **opaque** (the adapter builds it from its one-shot
|
||||||
|
/// receiver): the domain stays free of any concrete channel type, and the await point
|
||||||
|
/// lives in the application layer's `ask_agent`.
|
||||||
pub struct PendingReply {
|
pub struct PendingReply {
|
||||||
inner: Pin<Box<dyn Future<Output = Result<String, MailboxError>> + Send>>,
|
inner: Pin<Box<dyn Future<Output = Result<TurnResolution, MailboxError>> + Send>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl PendingReply {
|
impl PendingReply {
|
||||||
/// Wraps a reply future built by the adapter (e.g. over a one-shot receiver).
|
/// Wraps a reply future built by the adapter (e.g. over a one-shot receiver).
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn new(inner: Pin<Box<dyn Future<Output = Result<String, MailboxError>> + Send>>) -> Self {
|
pub fn new(
|
||||||
|
inner: Pin<Box<dyn Future<Output = Result<TurnResolution, MailboxError>> + Send>>,
|
||||||
|
) -> Self {
|
||||||
Self { inner }
|
Self { inner }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Future for PendingReply {
|
impl Future for PendingReply {
|
||||||
type Output = Result<String, MailboxError>;
|
type Output = Result<TurnResolution, MailboxError>;
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
self.inner.as_mut().poll(cx)
|
self.inner.as_mut().poll(cx)
|
||||||
@ -301,6 +327,28 @@ pub trait AgentMailbox: Send + Sync {
|
|||||||
/// A no-op when the head is a different ticket (the timed-out one was already
|
/// A no-op when the head is a different ticket (the timed-out one was already
|
||||||
/// resolved, or another caller's ticket is now in front) — idempotent and safe.
|
/// resolved, or another caller's ticket is now in front) — idempotent and safe.
|
||||||
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId);
|
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId);
|
||||||
|
|
||||||
|
/// Completes the turn of the ticket `ticket_id` **without** a reply, waking its
|
||||||
|
/// awaiting [`PendingReply`] with [`TurnResolution::ReturnedToPromptNoReply`].
|
||||||
|
///
|
||||||
|
/// Called when the target **returned to its prompt** (its turn ended) without
|
||||||
|
/// rendering an `idea_reply`: the rendezvous can produce no payload, so rather than
|
||||||
|
/// leaving the caller blocked until its long safety timeout, the head ticket is
|
||||||
|
/// resolved with the "no reply" outcome so the caller errors out promptly.
|
||||||
|
///
|
||||||
|
/// Contract for implementations:
|
||||||
|
/// - **head-only & idempotent**: act only if `ticket_id` is currently at the head
|
||||||
|
/// (a no-op once it has been resolved/cancelled, or if another ticket is in
|
||||||
|
/// front) — exactly like [`AgentMailbox::cancel_head`];
|
||||||
|
/// - **preserve fire-and-forget**: if the awaiting receiver already went away
|
||||||
|
/// (a human submit that is not awaited, or a caller that already timed out),
|
||||||
|
/// skip without retiring the head — the existing cancel/timeout paths own it.
|
||||||
|
///
|
||||||
|
/// The default is a no-op: a mailbox that cannot wake a pending caller early simply
|
||||||
|
/// falls back to the caller's timeout net (no behavioural change for it).
|
||||||
|
fn complete_without_reply(&self, agent: AgentId, ticket_id: TicketId) {
|
||||||
|
let _ = (agent, ticket_id);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|||||||
@ -22,6 +22,7 @@
|
|||||||
|
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use domain::events::DomainEvent;
|
use domain::events::DomainEvent;
|
||||||
use domain::ids::AgentId;
|
use domain::ids::AgentId;
|
||||||
@ -62,8 +63,24 @@ struct BusyTracker {
|
|||||||
/// présent ; `None` ⇒ toute livraison passe par l'événement (médiateur sans PTY,
|
/// présent ; `None` ⇒ toute livraison passe par l'événement (médiateur sans PTY,
|
||||||
/// utilisé par les tests événementiels — comportement historique).
|
/// utilisé par les tests événementiels — comportement historique).
|
||||||
headless_sink: Option<HeadlessSink>,
|
headless_sink: Option<HeadlessSink>,
|
||||||
|
/// Mailbox sous-jacent, pour armer la **fenêtre de grâce** au `prompt_ready` sans
|
||||||
|
/// délégation différée : si la cible revient à son prompt sans `idea_reply`, on
|
||||||
|
/// réveille l'appelant (via [`AgentMailbox::complete_without_reply`]) au lieu de le
|
||||||
|
/// laisser bloqué jusqu'au timeout long. `None` ⇒ pas de grâce (repli sur le filet
|
||||||
|
/// timeout — comportement historique pour les médiateurs sans mailbox câblé).
|
||||||
|
mailbox: Option<Arc<InMemoryMailbox>>,
|
||||||
|
/// Durée de la **fenêtre de grâce** (G) accordée à la cible après son retour au
|
||||||
|
/// prompt pour qu'un `idea_reply` arrive avant qu'on complète le tour « sans
|
||||||
|
/// réponse ». Prod = [`PROMPT_READY_GRACE`] (2 s) ; un test peut la raccourcir via
|
||||||
|
/// [`MediatedInbox::with_grace`].
|
||||||
|
grace: Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Fenêtre de grâce (G) par défaut entre le retour au prompt de la cible et la
|
||||||
|
/// complétion « sans réponse » du tour : laisse à un `idea_reply` tardif le temps
|
||||||
|
/// d'arriver d'abord. Décision produit (Main) : **2 secondes**.
|
||||||
|
const PROMPT_READY_GRACE: Duration = Duration::from_secs(2);
|
||||||
|
|
||||||
/// Payload d'une [`DomainEvent::DelegationReady`] **différée** le temps qu'un agent
|
/// Payload d'une [`DomainEvent::DelegationReady`] **différée** le temps qu'un agent
|
||||||
/// lancé à froid atteigne son prompt (fix race cold-launch). Reconstruite à l'identique
|
/// lancé à froid atteigne son prompt (fix race cold-launch). Reconstruite à l'identique
|
||||||
/// quand le prompt-ready watcher déclenche, de sorte que le premier tour soit livré au
|
/// quand le prompt-ready watcher déclenche, de sorte que le premier tour soit livré au
|
||||||
@ -124,6 +141,8 @@ impl BusyTracker {
|
|||||||
deferred: Mutex::new(HashMap::new()),
|
deferred: Mutex::new(HashMap::new()),
|
||||||
events,
|
events,
|
||||||
headless_sink: None,
|
headless_sink: None,
|
||||||
|
mailbox: None,
|
||||||
|
grace: PROMPT_READY_GRACE,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -336,16 +355,34 @@ impl BusyTracker {
|
|||||||
self.publish_deferred(agent, d);
|
self.publish_deferred(agent, d);
|
||||||
}
|
}
|
||||||
None => {
|
None => {
|
||||||
// Pas de tour différé : le prompt-ready ne fait que **libérer** le busy
|
// Pas de tour différé : la cible est revenue à son prompt sans
|
||||||
// state / la FIFO (mark_idle). Il ne porte AUCUNE réponse pour réveiller
|
// `idea_reply`. On capture le ticket actif AVANT `mark_idle` (qui efface
|
||||||
// un appelant en attente — d'où `no_reply_payload`. C'est exactement le
|
// l'état Busy), puis on libère le busy state / la FIFO comme avant. Le
|
||||||
// scénario où une cible revient prompt-ready sans `idea_reply` : l'ask
|
// prompt-ready ne porte AUCUNE réponse (`no_reply_payload`) : pour ne pas
|
||||||
// ne se résout pas par ce chemin, seulement par l'`idea_reply` ou le
|
// laisser l'appelant bloqué jusqu'au timeout long, on **arme une fenêtre
|
||||||
// timeout côté appelant.
|
// de grâce** G — si aucun `idea_reply` n'a corrélé le ticket d'ici là, on
|
||||||
|
// complète le tour « sans réponse » (réveille l'appelant avec une erreur
|
||||||
|
// typée). Un `idea_reply` arrivé pendant G gagne (la complétion devient
|
||||||
|
// un no-op, tête déjà retirée).
|
||||||
|
let active = self.lock().get(&agent).and_then(AgentBusyState::ticket);
|
||||||
application::diag!(
|
application::diag!(
|
||||||
"[input-mediator] prompt_ready agent={agent} no_reply_payload -> mark_idle"
|
"[input-mediator] prompt_ready agent={agent} no_reply_payload -> mark_idle + grace"
|
||||||
);
|
);
|
||||||
self.mark_idle(agent);
|
self.mark_idle(agent);
|
||||||
|
if let (Some(ticket), Some(mailbox)) = (active, self.mailbox.clone()) {
|
||||||
|
let grace = self.grace;
|
||||||
|
application::diag!(
|
||||||
|
"[input-mediator] prompt_ready arming grace agent={agent} ticket={ticket} grace_ms={}",
|
||||||
|
grace.as_millis(),
|
||||||
|
);
|
||||||
|
// Thread détaché (jamais de hot loop, jamais de blocage de l'appelant) :
|
||||||
|
// attend G puis tente la complétion « sans réponse ». `complete_without_reply`
|
||||||
|
// est head-only/idempotent et préserve le fire-and-forget humain.
|
||||||
|
std::thread::spawn(move || {
|
||||||
|
std::thread::sleep(grace);
|
||||||
|
mailbox.complete_without_reply(agent, ticket);
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -452,6 +489,9 @@ pub struct MediatedInbox {
|
|||||||
/// handle does not spawn a duplicate watcher (lot C5). Shared (`Arc`) because each
|
/// handle does not spawn a duplicate watcher (lot C5). Shared (`Arc`) because each
|
||||||
/// watcher thread un-arms its own entry on exit.
|
/// watcher thread un-arms its own entry on exit.
|
||||||
watched: Arc<Mutex<HashSet<AgentId>>>,
|
watched: Arc<Mutex<HashSet<AgentId>>>,
|
||||||
|
/// Fenêtre de grâce (G) propagée au [`BusyTracker`] à chaque (re)construction.
|
||||||
|
/// Prod = [`PROMPT_READY_GRACE`] (2 s) ; surchargée par [`Self::with_grace`] en test.
|
||||||
|
grace: Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl MediatedInbox {
|
impl MediatedInbox {
|
||||||
@ -461,9 +501,17 @@ impl MediatedInbox {
|
|||||||
pub fn new(mailbox: Arc<InMemoryMailbox>, clock: Arc<dyn MillisClock>) -> Self {
|
pub fn new(mailbox: Arc<InMemoryMailbox>, clock: Arc<dyn MillisClock>) -> Self {
|
||||||
let handles = Arc::new(Mutex::new(HashMap::new()));
|
let handles = Arc::new(Mutex::new(HashMap::new()));
|
||||||
let front_owned = Arc::new(Mutex::new(HashSet::new()));
|
let front_owned = Arc::new(Mutex::new(HashSet::new()));
|
||||||
|
let tracker = Self::build_tracker(
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
&handles,
|
||||||
|
&front_owned,
|
||||||
|
&mailbox,
|
||||||
|
PROMPT_READY_GRACE,
|
||||||
|
);
|
||||||
Self {
|
Self {
|
||||||
mailbox,
|
mailbox,
|
||||||
tracker: Self::build_tracker(None, None, &handles, &front_owned),
|
tracker,
|
||||||
clock,
|
clock,
|
||||||
pty: None,
|
pty: None,
|
||||||
handles,
|
handles,
|
||||||
@ -471,6 +519,7 @@ impl MediatedInbox {
|
|||||||
submit: Mutex::new(HashMap::new()),
|
submit: Mutex::new(HashMap::new()),
|
||||||
stall: Mutex::new(HashMap::new()),
|
stall: Mutex::new(HashMap::new()),
|
||||||
watched: Arc::new(Mutex::new(HashSet::new())),
|
watched: Arc::new(Mutex::new(HashSet::new())),
|
||||||
|
grace: PROMPT_READY_GRACE,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -481,8 +530,14 @@ impl MediatedInbox {
|
|||||||
pty: Option<&Arc<dyn PtyPort>>,
|
pty: Option<&Arc<dyn PtyPort>>,
|
||||||
handles: &Arc<Mutex<HashMap<AgentId, PtyHandle>>>,
|
handles: &Arc<Mutex<HashMap<AgentId, PtyHandle>>>,
|
||||||
front_owned: &Arc<Mutex<HashSet<AgentId>>>,
|
front_owned: &Arc<Mutex<HashSet<AgentId>>>,
|
||||||
|
mailbox: &Arc<InMemoryMailbox>,
|
||||||
|
grace: Duration,
|
||||||
) -> Arc<BusyTracker> {
|
) -> Arc<BusyTracker> {
|
||||||
let mut tracker = BusyTracker::new(events);
|
let mut tracker = BusyTracker::new(events);
|
||||||
|
// Le tracker porte le mailbox + la grâce pour pouvoir compléter « sans réponse »
|
||||||
|
// un tour dont la cible est revenue au prompt sans `idea_reply` (cf. prompt_ready).
|
||||||
|
tracker.mailbox = Some(Arc::clone(mailbox));
|
||||||
|
tracker.grace = grace;
|
||||||
if let Some(pty) = pty {
|
if let Some(pty) = pty {
|
||||||
tracker.headless_sink = Some(Self::make_headless_sink(
|
tracker.headless_sink = Some(Self::make_headless_sink(
|
||||||
Arc::clone(pty),
|
Arc::clone(pty),
|
||||||
@ -601,6 +656,27 @@ impl MediatedInbox {
|
|||||||
self.pty.as_ref(),
|
self.pty.as_ref(),
|
||||||
&self.handles,
|
&self.handles,
|
||||||
&self.front_owned,
|
&self.front_owned,
|
||||||
|
&self.mailbox,
|
||||||
|
self.grace,
|
||||||
|
);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// **Test seam**: shrinks the prompt-ready **grace window** (G, see
|
||||||
|
/// [`PROMPT_READY_GRACE`]) so a no-reply test need not wait the full 2 s. Rebuilds
|
||||||
|
/// the tracker with the new grace; production code keeps the default. Call at
|
||||||
|
/// construction time (before any turn) — it resets the freshly built tracker.
|
||||||
|
#[doc(hidden)]
|
||||||
|
#[must_use]
|
||||||
|
pub fn with_grace(mut self, grace: Duration) -> Self {
|
||||||
|
self.grace = grace;
|
||||||
|
self.tracker = Self::build_tracker(
|
||||||
|
self.tracker.events.clone(),
|
||||||
|
self.pty.as_ref(),
|
||||||
|
&self.handles,
|
||||||
|
&self.front_owned,
|
||||||
|
&self.mailbox,
|
||||||
|
grace,
|
||||||
);
|
);
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
@ -617,9 +693,17 @@ impl MediatedInbox {
|
|||||||
let handles = Arc::new(Mutex::new(HashMap::new()));
|
let handles = Arc::new(Mutex::new(HashMap::new()));
|
||||||
let front_owned = Arc::new(Mutex::new(HashSet::new()));
|
let front_owned = Arc::new(Mutex::new(HashSet::new()));
|
||||||
let pty_opt = Some(pty);
|
let pty_opt = Some(pty);
|
||||||
|
let tracker = Self::build_tracker(
|
||||||
|
None,
|
||||||
|
pty_opt.as_ref(),
|
||||||
|
&handles,
|
||||||
|
&front_owned,
|
||||||
|
&mailbox,
|
||||||
|
PROMPT_READY_GRACE,
|
||||||
|
);
|
||||||
Self {
|
Self {
|
||||||
mailbox,
|
mailbox,
|
||||||
tracker: Self::build_tracker(None, pty_opt.as_ref(), &handles, &front_owned),
|
tracker,
|
||||||
clock,
|
clock,
|
||||||
pty: pty_opt,
|
pty: pty_opt,
|
||||||
handles,
|
handles,
|
||||||
@ -627,6 +711,7 @@ impl MediatedInbox {
|
|||||||
submit: Mutex::new(HashMap::new()),
|
submit: Mutex::new(HashMap::new()),
|
||||||
stall: Mutex::new(HashMap::new()),
|
stall: Mutex::new(HashMap::new()),
|
||||||
watched: Arc::new(Mutex::new(HashSet::new())),
|
watched: Arc::new(Mutex::new(HashSet::new())),
|
||||||
|
grace: PROMPT_READY_GRACE,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1309,7 +1394,10 @@ mod tests {
|
|||||||
let pending = inbox.enqueue(a, ticket(10, "do X"));
|
let pending = inbox.enqueue(a, ticket(10, "do X"));
|
||||||
// Resolve through the shared mailbox (the orchestrator's path).
|
// Resolve through the shared mailbox (the orchestrator's path).
|
||||||
inbox.mailbox().resolve(a, "done".to_owned()).unwrap();
|
inbox.mailbox().resolve(a, "done".to_owned()).unwrap();
|
||||||
assert_eq!(pending.await.unwrap(), "done");
|
assert_eq!(
|
||||||
|
pending.await.unwrap(),
|
||||||
|
domain::mailbox::TurnResolution::Replied("done".to_owned())
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@ -1367,7 +1455,10 @@ mod tests {
|
|||||||
assert_eq!(inbox.mailbox().pending(&a), 1);
|
assert_eq!(inbox.mailbox().pending(&a), 1);
|
||||||
// Nothing answered the caller via preempt.
|
// Nothing answered the caller via preempt.
|
||||||
inbox.mailbox().resolve(a, "real".to_owned()).unwrap();
|
inbox.mailbox().resolve(a, "real".to_owned()).unwrap();
|
||||||
assert_eq!(pending.await.unwrap(), "real");
|
assert_eq!(
|
||||||
|
pending.await.unwrap(),
|
||||||
|
domain::mailbox::TurnResolution::Replied("real".to_owned())
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@ -1490,6 +1581,155 @@ mod tests {
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Builds a PTY-backed inbox with a **short grace** so the no-reply path is
|
||||||
|
/// observable in milliseconds (prod grace is 2 s).
|
||||||
|
fn inbox_with_grace(pty: Arc<FakePty>, grace: Duration) -> MediatedInbox {
|
||||||
|
MediatedInbox::with_pty(
|
||||||
|
Arc::new(InMemoryMailbox::new()),
|
||||||
|
Arc::new(FixedClock(1)),
|
||||||
|
pty as Arc<dyn PtyPort>,
|
||||||
|
)
|
||||||
|
.with_grace(grace)
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- fix: prompt-ready sans idea_reply ⇒ réveil de l'appelant après G ----------
|
||||||
|
|
||||||
|
/// La cible revient à son prompt SANS `idea_reply` : après la fenêtre de grâce,
|
||||||
|
/// l'appelant en attente est réveillé avec `ReturnedToPromptNoReply` — pas de
|
||||||
|
/// blocage jusqu'au timeout long (le bug corrigé).
|
||||||
|
#[tokio::test]
|
||||||
|
async fn prompt_ready_without_reply_wakes_caller_after_grace() {
|
||||||
|
let pty = Arc::new(FakePty::new());
|
||||||
|
let a = agent(1);
|
||||||
|
let h = handle(50);
|
||||||
|
pty.seed(&h, vec![b"done\n> ".to_vec()]);
|
||||||
|
let inbox = inbox_with_grace(Arc::clone(&pty), Duration::from_millis(40));
|
||||||
|
|
||||||
|
let pending = inbox.enqueue(a, ticket(10, "task"));
|
||||||
|
inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()), SubmitConfig::default());
|
||||||
|
|
||||||
|
// Bounded await: the grace (40 ms) must resolve the caller well under this.
|
||||||
|
let res = tokio::time::timeout(Duration::from_secs(2), pending)
|
||||||
|
.await
|
||||||
|
.expect("must NOT hang until the long timeout")
|
||||||
|
.expect("a turn outcome, not a closed channel");
|
||||||
|
assert_eq!(
|
||||||
|
res,
|
||||||
|
domain::mailbox::TurnResolution::ReturnedToPromptNoReply
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
inbox.mailbox().pending(&a),
|
||||||
|
0,
|
||||||
|
"head retired by the no-reply completion"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `idea_reply` arrivé AVANT le prompt-ready gagne : l'appelant reçoit le `Replied`
|
||||||
|
/// et la complétion de grâce (tête déjà retirée) est un no-op.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reply_before_prompt_ready_wins() {
|
||||||
|
let pty = Arc::new(FakePty::new());
|
||||||
|
let a = agent(1);
|
||||||
|
let h = handle(51);
|
||||||
|
pty.seed(&h, vec![b"done\n> ".to_vec()]);
|
||||||
|
let inbox = inbox_with_grace(Arc::clone(&pty), Duration::from_millis(40));
|
||||||
|
|
||||||
|
let pending = inbox.enqueue(a, ticket(10, "task"));
|
||||||
|
// idea_reply resolves the head BEFORE the watcher fires prompt-ready.
|
||||||
|
inbox.mailbox().resolve(a, "answer".to_owned()).unwrap();
|
||||||
|
inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()), SubmitConfig::default());
|
||||||
|
|
||||||
|
let res = tokio::time::timeout(Duration::from_secs(2), pending)
|
||||||
|
.await
|
||||||
|
.expect("no hang")
|
||||||
|
.expect("outcome");
|
||||||
|
assert_eq!(
|
||||||
|
res,
|
||||||
|
domain::mailbox::TurnResolution::Replied("answer".to_owned())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `idea_reply` arrivé APRÈS le prompt-ready mais DANS la fenêtre de grâce gagne :
|
||||||
|
/// avec une grâce longue, on résout via la mailbox avant son expiration.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reply_within_grace_after_prompt_ready_wins() {
|
||||||
|
let pty = Arc::new(FakePty::new());
|
||||||
|
let a = agent(1);
|
||||||
|
let h = handle(52);
|
||||||
|
pty.seed(&h, vec![b"done\n> ".to_vec()]);
|
||||||
|
// Long grace ⇒ the reply lands well within it.
|
||||||
|
let inbox = inbox_with_grace(Arc::clone(&pty), Duration::from_secs(30));
|
||||||
|
|
||||||
|
let pending = inbox.enqueue(a, ticket(10, "task"));
|
||||||
|
inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()), SubmitConfig::default());
|
||||||
|
// Let the watcher fire prompt-ready (arms the grace), then reply within it.
|
||||||
|
assert!(
|
||||||
|
wait_until(|| !inbox.busy_state(a).is_busy()),
|
||||||
|
"prompt-ready marks idle (grace armed)"
|
||||||
|
);
|
||||||
|
inbox.mailbox().resolve(a, "in-time".to_owned()).unwrap();
|
||||||
|
|
||||||
|
let res = tokio::time::timeout(Duration::from_secs(2), pending)
|
||||||
|
.await
|
||||||
|
.expect("no hang")
|
||||||
|
.expect("outcome");
|
||||||
|
assert_eq!(
|
||||||
|
res,
|
||||||
|
domain::mailbox::TurnResolution::Replied("in-time".to_owned())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `idea_reply` arrivé APRÈS l'expiration de la grâce devient *unmatched* (la tête a
|
||||||
|
/// été retirée par la complétion) — typé, idempotent, jamais un panic.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reply_after_grace_is_unmatched() {
|
||||||
|
let pty = Arc::new(FakePty::new());
|
||||||
|
let a = agent(1);
|
||||||
|
let h = handle(53);
|
||||||
|
pty.seed(&h, vec![b"done\n> ".to_vec()]);
|
||||||
|
let inbox = inbox_with_grace(Arc::clone(&pty), Duration::from_millis(30));
|
||||||
|
|
||||||
|
let pending = inbox.enqueue(a, ticket(10, "task"));
|
||||||
|
inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()), SubmitConfig::default());
|
||||||
|
// The caller is woken by the grace completion (head retired).
|
||||||
|
let res = tokio::time::timeout(Duration::from_secs(2), pending)
|
||||||
|
.await
|
||||||
|
.expect("no hang")
|
||||||
|
.expect("outcome");
|
||||||
|
assert_eq!(
|
||||||
|
res,
|
||||||
|
domain::mailbox::TurnResolution::ReturnedToPromptNoReply
|
||||||
|
);
|
||||||
|
|
||||||
|
// A late idea_reply now has no head to resolve ⇒ typed NoPendingRequest.
|
||||||
|
assert!(
|
||||||
|
inbox.mailbox().resolve(a, "too late".to_owned()).is_err(),
|
||||||
|
"late reply is unmatched, not a panic"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Submit humain *fire-and-forget* (PendingReply non attendu) : la grâce ne doit PAS
|
||||||
|
/// retirer la tête — `complete_without_reply` skip quand le receiver est fermé.
|
||||||
|
#[test]
|
||||||
|
fn fire_and_forget_head_is_preserved_through_grace() {
|
||||||
|
let pty = Arc::new(FakePty::new());
|
||||||
|
let a = agent(1);
|
||||||
|
let h = handle(54);
|
||||||
|
pty.seed(&h, vec![b"done\n> ".to_vec()]);
|
||||||
|
let inbox = inbox_with_grace(Arc::clone(&pty), Duration::from_millis(20));
|
||||||
|
|
||||||
|
// Human submit: the reply handle is dropped immediately (not awaited).
|
||||||
|
drop(inbox.enqueue(a, ticket(10, "human submit")));
|
||||||
|
inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()), SubmitConfig::default());
|
||||||
|
// Give the grace thread time to run; the head must survive (skip-closed).
|
||||||
|
std::thread::sleep(Duration::from_millis(120));
|
||||||
|
assert_eq!(
|
||||||
|
inbox.mailbox().pending(&a),
|
||||||
|
1,
|
||||||
|
"fire-and-forget head preserved (grace is a no-op on a closed receiver)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn prompt_pattern_present_and_output_contains_it_flips_busy_to_idle() {
|
fn prompt_pattern_present_and_output_contains_it_flips_busy_to_idle() {
|
||||||
let pty = Arc::new(FakePty::new());
|
let pty = Arc::new(FakePty::new());
|
||||||
|
|||||||
@ -27,13 +27,18 @@ use tokio::sync::oneshot;
|
|||||||
use domain::ids::AgentId;
|
use domain::ids::AgentId;
|
||||||
use domain::mailbox::{
|
use domain::mailbox::{
|
||||||
AgentMailbox, AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket,
|
AgentMailbox, AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket,
|
||||||
TicketId,
|
TicketId, TurnResolution,
|
||||||
};
|
};
|
||||||
|
|
||||||
/// One queued request plus the sender that resolves its awaiting [`PendingReply`].
|
/// One queued request plus the sender that resolves its awaiting [`PendingReply`].
|
||||||
|
///
|
||||||
|
/// The one-shot carries a [`TurnResolution`] so the awaiting caller learns **how** the
|
||||||
|
/// turn ended: a [`TurnResolution::Replied`] (an `idea_reply`) or a
|
||||||
|
/// [`TurnResolution::ReturnedToPromptNoReply`] (the target returned to its prompt
|
||||||
|
/// without replying — see [`AgentMailbox::complete_without_reply`]).
|
||||||
struct Slot {
|
struct Slot {
|
||||||
ticket: Ticket,
|
ticket: Ticket,
|
||||||
reply: oneshot::Sender<String>,
|
reply: oneshot::Sender<TurnResolution>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// In-memory, per-agent FIFO mailbox (the production [`AgentMailbox`]).
|
/// In-memory, per-agent FIFO mailbox (the production [`AgentMailbox`]).
|
||||||
@ -76,7 +81,7 @@ 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::<TurnResolution>();
|
||||||
let ticket_id = ticket.id;
|
let ticket_id = ticket.id;
|
||||||
let (depth_before, depth_after) = {
|
let (depth_before, depth_after) = {
|
||||||
let mut queues = self
|
let mut queues = self
|
||||||
@ -129,7 +134,7 @@ impl AgentMailbox for InMemoryMailbox {
|
|||||||
);
|
);
|
||||||
// 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(TurnResolution::Replied(result));
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -174,7 +179,7 @@ impl AgentMailbox for InMemoryMailbox {
|
|||||||
"[mailbox] resolve_ticket agent={agent} ticket={ticket_id} \
|
"[mailbox] resolve_ticket agent={agent} ticket={ticket_id} \
|
||||||
queue_depth_before={depth_before} queue_depth_after={depth_after}"
|
queue_depth_before={depth_before} queue_depth_after={depth_after}"
|
||||||
);
|
);
|
||||||
let _ = slot.reply.send(result);
|
let _ = slot.reply.send(TurnResolution::Replied(result));
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -212,6 +217,59 @@ impl AgentMailbox for InMemoryMailbox {
|
|||||||
queue_depth_before={depth_before} queue_depth_after={depth_after} retired={retired}"
|
queue_depth_before={depth_before} queue_depth_after={depth_after} retired={retired}"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn complete_without_reply(&self, agent: AgentId, ticket_id: TicketId) {
|
||||||
|
// Head-only, idempotent, and fire-and-forget-safe (see the port contract).
|
||||||
|
// `outcome` records what happened for the diag beacon: "sent" (head retired +
|
||||||
|
// caller woken with ReturnedToPromptNoReply), "skip-closed" (receiver already
|
||||||
|
// gone — human submit / timed-out caller; head left for cancel/timeout to own),
|
||||||
|
// or "skip-not-head" (already resolved/cancelled or another ticket in front).
|
||||||
|
let (outcome, depth_before, depth_after) = {
|
||||||
|
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] complete_without_reply agent={agent} ticket={ticket_id} \
|
||||||
|
queue_depth_before=0 queue_depth_after=0 outcome=skip-no-queue"
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let before = queue.len();
|
||||||
|
// Only act on the head, and only if it is *this* ticket (idempotent: a head
|
||||||
|
// that has since changed must never be retired here).
|
||||||
|
if queue.front().map(|s| s.ticket.id) != Some(ticket_id) {
|
||||||
|
application::diag!(
|
||||||
|
"[mailbox] complete_without_reply agent={agent} ticket={ticket_id} \
|
||||||
|
queue_depth_before={before} queue_depth_after={before} outcome=skip-not-head"
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// Preserve fire-and-forget: if the awaiting receiver already went away (a
|
||||||
|
// human submit that is never awaited, or a caller that already timed out),
|
||||||
|
// do NOT pop — the existing cancel_head/timeout path owns that head.
|
||||||
|
if queue.front().is_some_and(|s| s.reply.is_closed()) {
|
||||||
|
application::diag!(
|
||||||
|
"[mailbox] complete_without_reply agent={agent} ticket={ticket_id} \
|
||||||
|
queue_depth_before={before} queue_depth_after={before} outcome=skip-closed"
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let slot = queue.pop_front().expect("head just matched is present");
|
||||||
|
let after = queue.len();
|
||||||
|
if queue.is_empty() {
|
||||||
|
queues.remove(&agent);
|
||||||
|
}
|
||||||
|
// Wake the awaiting caller with the "returned to prompt, no reply" outcome.
|
||||||
|
let _ = slot.reply.send(TurnResolution::ReturnedToPromptNoReply);
|
||||||
|
("sent", before, after)
|
||||||
|
};
|
||||||
|
application::diag!(
|
||||||
|
"[mailbox] complete_without_reply agent={agent} ticket={ticket_id} \
|
||||||
|
queue_depth_before={depth_before} queue_depth_after={depth_after} outcome={outcome}"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AgentQueueSnapshot for InMemoryMailbox {
|
impl AgentQueueSnapshot for InMemoryMailbox {
|
||||||
@ -264,7 +322,7 @@ mod tests {
|
|||||||
mb.resolve(a, "done X".to_owned()).expect("resolve ok");
|
mb.resolve(a, "done X".to_owned()).expect("resolve ok");
|
||||||
|
|
||||||
let reply = pending.await.expect("reply received");
|
let reply = pending.await.expect("reply received");
|
||||||
assert_eq!(reply, "done X");
|
assert_eq!(reply, TurnResolution::Replied("done X".to_owned()));
|
||||||
assert_eq!(mb.pending(&a), 0, "queue drained after resolve");
|
assert_eq!(mb.pending(&a), 0, "queue drained after resolve");
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -282,14 +340,14 @@ mod tests {
|
|||||||
|
|
||||||
// First resolve goes to the FIRST (head) ticket.
|
// First resolve goes to the FIRST (head) ticket.
|
||||||
mb.resolve(a, "r1".to_owned()).unwrap();
|
mb.resolve(a, "r1".to_owned()).unwrap();
|
||||||
assert_eq!(p1.await.unwrap(), "r1");
|
assert_eq!(p1.await.unwrap(), TurnResolution::Replied("r1".to_owned()));
|
||||||
// Now the second is at the head.
|
// Now the second is at the head.
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
mb.head_ticket(&a),
|
mb.head_ticket(&a),
|
||||||
Some(TicketId::from_uuid(uuid::Uuid::from_u128(11)))
|
Some(TicketId::from_uuid(uuid::Uuid::from_u128(11)))
|
||||||
);
|
);
|
||||||
mb.resolve(a, "r2".to_owned()).unwrap();
|
mb.resolve(a, "r2".to_owned()).unwrap();
|
||||||
assert_eq!(p2.await.unwrap(), "r2");
|
assert_eq!(p2.await.unwrap(), TurnResolution::Replied("r2".to_owned()));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
@ -304,7 +362,7 @@ mod tests {
|
|||||||
mb.resolve(b, "rb".to_owned()).unwrap();
|
mb.resolve(b, "rb".to_owned()).unwrap();
|
||||||
assert_eq!(mb.pending(&a), 1, "A's queue is independent of B's");
|
assert_eq!(mb.pending(&a), 1, "A's queue is independent of B's");
|
||||||
mb.resolve(a, "ra".to_owned()).unwrap();
|
mb.resolve(a, "ra".to_owned()).unwrap();
|
||||||
assert_eq!(pa.await.unwrap(), "ra");
|
assert_eq!(pa.await.unwrap(), TurnResolution::Replied("ra".to_owned()));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@ -337,7 +395,78 @@ mod tests {
|
|||||||
|
|
||||||
// The next ticket is now resolvable normally.
|
// The next ticket is now resolvable normally.
|
||||||
mb.resolve(a, "r2".to_owned()).unwrap();
|
mb.resolve(a, "r2".to_owned()).unwrap();
|
||||||
assert_eq!(p2.await.unwrap(), "r2");
|
assert_eq!(p2.await.unwrap(), TurnResolution::Replied("r2".to_owned()));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn complete_without_reply_wakes_head_caller_with_returned_to_prompt() {
|
||||||
|
let mb = InMemoryMailbox::new();
|
||||||
|
let a = agent(1);
|
||||||
|
let pending = mb.enqueue(a, ticket(10, "awaited"));
|
||||||
|
|
||||||
|
// The target returned to its prompt without idea_reply ⇒ the awaiting caller
|
||||||
|
// is woken with ReturnedToPromptNoReply (not a String), and the head is retired.
|
||||||
|
mb.complete_without_reply(a, TicketId::from_uuid(uuid::Uuid::from_u128(10)));
|
||||||
|
assert_eq!(
|
||||||
|
pending.await.unwrap(),
|
||||||
|
TurnResolution::ReturnedToPromptNoReply
|
||||||
|
);
|
||||||
|
assert_eq!(mb.pending(&a), 0, "head retired after completion");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn complete_without_reply_is_noop_when_receiver_already_gone() {
|
||||||
|
let mb = InMemoryMailbox::new();
|
||||||
|
let a = agent(1);
|
||||||
|
// Human fire-and-forget: the PendingReply is dropped immediately (not awaited).
|
||||||
|
drop(mb.enqueue(a, ticket(10, "human submit")));
|
||||||
|
|
||||||
|
// The receiver is closed ⇒ completion must SKIP without retiring the head, so
|
||||||
|
// the existing cancel/timeout path still owns it (no behavioural change).
|
||||||
|
mb.complete_without_reply(a, TicketId::from_uuid(uuid::Uuid::from_u128(10)));
|
||||||
|
assert_eq!(mb.pending(&a), 1, "fire-and-forget head preserved");
|
||||||
|
assert_eq!(
|
||||||
|
mb.head_ticket(&a),
|
||||||
|
Some(TicketId::from_uuid(uuid::Uuid::from_u128(10)))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn complete_without_reply_is_noop_when_not_head_and_idempotent() {
|
||||||
|
let mb = InMemoryMailbox::new();
|
||||||
|
let a = agent(1);
|
||||||
|
let p1 = mb.enqueue(a, ticket(10, "head"));
|
||||||
|
let _p2 = mb.enqueue(a, ticket(11, "behind"));
|
||||||
|
|
||||||
|
// Targeting a non-head ticket ⇒ no-op.
|
||||||
|
mb.complete_without_reply(a, TicketId::from_uuid(uuid::Uuid::from_u128(11)));
|
||||||
|
assert_eq!(mb.pending(&a), 2, "non-head completion is a no-op");
|
||||||
|
|
||||||
|
// Complete the head, then a second call for the same ticket is idempotent.
|
||||||
|
mb.complete_without_reply(a, TicketId::from_uuid(uuid::Uuid::from_u128(10)));
|
||||||
|
assert_eq!(p1.await.unwrap(), TurnResolution::ReturnedToPromptNoReply);
|
||||||
|
assert_eq!(mb.pending(&a), 1, "only the head was retired");
|
||||||
|
mb.complete_without_reply(a, TicketId::from_uuid(uuid::Uuid::from_u128(10)));
|
||||||
|
assert_eq!(mb.pending(&a), 1, "idempotent: already retired");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `idea_reply` wins a race against a pending completion: once the head is resolved
|
||||||
|
/// with a `Replied`, a later `complete_without_reply` for the same ticket is a no-op.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reply_before_completion_wins_then_completion_is_noop() {
|
||||||
|
let mb = InMemoryMailbox::new();
|
||||||
|
let a = agent(1);
|
||||||
|
let pending = mb.enqueue(a, ticket(10, "raced"));
|
||||||
|
|
||||||
|
mb.resolve(a, "real answer".to_owned()).unwrap();
|
||||||
|
// The (late) completion finds the head already gone ⇒ no-op.
|
||||||
|
mb.complete_without_reply(a, TicketId::from_uuid(uuid::Uuid::from_u128(10)));
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
pending.await.unwrap(),
|
||||||
|
TurnResolution::Replied("real answer".to_owned()),
|
||||||
|
"the idea_reply result wins the race"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
Reference in New Issue
Block a user