diff --git a/crates/application/src/error.rs b/crates/application/src/error.rs index 055ccb0..5d1be6c 100644 --- a/crates/application/src/error.rs +++ b/crates/application/src/error.rs @@ -57,6 +57,18 @@ pub enum AppError { 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. #[error("internal error: {0}")] Internal(String), @@ -76,6 +88,7 @@ impl AppError { Self::Git(_) => "GIT", Self::Remote(_) => "REMOTE", Self::AgentAlreadyRunning { .. } => "AGENT_ALREADY_RUNNING", + Self::TargetReturnedNoReply(_) => "TARGET_RETURNED_NO_REPLY", Self::Internal(_) => "INTERNAL", } } diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index d5904e0..61fdec6 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -22,7 +22,7 @@ use tokio::sync::Mutex as AsyncMutex; use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef, WaitForGraph}; use domain::conversation_log::{ConversationTurn, TurnId, TurnRole}; 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::project::ProjectPath; use domain::{ @@ -1063,7 +1063,7 @@ impl OrchestratorService { // 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. 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 // déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est // 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)) } + // 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 // Drop (retrait des `cancel_head` redondants — `cancel_head` reste idempotent). Ok(Err(_cancelled)) => { @@ -1232,7 +1247,7 @@ impl OrchestratorService { }, // Issue alternative : un `idea_reply` explicite a résolu le ticket d'abord. replied = pending => match replied { - Ok(content) => { + Ok(TurnResolution::Replied(content)) => { // La session draine encore en arrière-plan ; on s'assure que la FIFO // avance même si le `Final` n'est pas encore observé. input.mark_idle(agent_id); @@ -1244,6 +1259,19 @@ impl OrchestratorService { ); 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. Err(_cancelled) => { crate::diag!( diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index a36c3c2..6e8fb0f 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -861,7 +861,7 @@ use domain::conversation::{ SessionRef, }; 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}; /// 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)`. #[derive(Default)] struct TestMailbox { - queues: Mutex)>>>, + queues: + Mutex)>>>, } impl TestMailbox { fn new() -> Self { @@ -899,7 +900,7 @@ impl TestMailbox { } impl AgentMailbox for TestMailbox { fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply { - let (tx, rx) = tokio::sync::oneshot::channel::(); + let (tx, rx) = tokio::sync::oneshot::channel::(); self.queues .lock() .unwrap() @@ -919,7 +920,7 @@ impl AgentMailbox for TestMailbox { .ok_or(MailboxError::NoPendingRequest(agent))?; queue.pop_front().expect("non-empty") }; - let _ = slot.1.send(result); + let _ = slot.1.send(TurnResolution::Replied(result)); Ok(()) } fn resolve_ticket( @@ -940,7 +941,7 @@ impl AgentMailbox for TestMailbox { .ok_or(MailboxError::NoPendingRequest(agent))?; queue.remove(pos).expect("found position") }; - let _ = slot.1.send(result); + let _ = slot.1.send(TurnResolution::Replied(result)); Ok(()) } 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 @@ -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"); } +/// 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 ----------------- /// A reply carrying an `idea-memory` block is harvested **after** the reply diff --git a/crates/application/tests/workstate.rs b/crates/application/tests/workstate.rs index b2aaf2b..2ddf00b 100644 --- a/crates/application/tests/workstate.rs +++ b/crates/application/tests/workstate.rs @@ -13,7 +13,7 @@ use application::{ TerminalSessions, TicketWorkSource, TicketWorkStatus, }; use domain::mailbox::{ - AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, + AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, TurnResolution, }; use domain::ports::{ AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError, @@ -123,7 +123,7 @@ impl FakeInput { impl InputMediator for FakeInput { fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply { - let fut: Pin> + Send>> = + let fut: Pin> + Send>> = Box::pin(async { Err(MailboxError::Cancelled) }); PendingReply::new(fut) } diff --git a/crates/domain/src/mailbox.rs b/crates/domain/src/mailbox.rs index 92f813f..e852b86 100644 --- a/crates/domain/src/mailbox.rs +++ b/crates/domain/src/mailbox.rs @@ -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. #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] pub enum MailboxError { @@ -206,28 +228,32 @@ pub enum MailboxError { 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 -/// later [`AgentMailbox::resolve`] feeds into this ticket's slot, or -/// [`MailboxError::Cancelled`] if the reply channel closes 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`. +/// Returned by [`AgentMailbox::enqueue`]. Awaiting it yields a [`TurnResolution`] (a +/// [`TurnResolution::Replied`] result fed by a later [`AgentMailbox::resolve`], or a +/// [`TurnResolution::ReturnedToPromptNoReply`] fed by +/// [`AgentMailbox::complete_without_reply`] when the target returns to its prompt +/// without replying), or [`MailboxError::Cancelled`] if the reply channel closes +/// 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 { - inner: Pin> + Send>>, + inner: Pin> + Send>>, } impl PendingReply { /// Wraps a reply future built by the adapter (e.g. over a one-shot receiver). #[must_use] - pub fn new(inner: Pin> + Send>>) -> Self { + pub fn new( + inner: Pin> + Send>>, + ) -> Self { Self { inner } } } impl Future for PendingReply { - type Output = Result; + type Output = Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { 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 /// resolved, or another caller's ticket is now in front) — idempotent and safe. 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)] diff --git a/crates/infrastructure/src/input/mod.rs b/crates/infrastructure/src/input/mod.rs index 90b939f..8a9d40c 100644 --- a/crates/infrastructure/src/input/mod.rs +++ b/crates/infrastructure/src/input/mod.rs @@ -22,6 +22,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; +use std::time::Duration; use domain::events::DomainEvent; use domain::ids::AgentId; @@ -62,8 +63,24 @@ struct BusyTracker { /// présent ; `None` ⇒ toute livraison passe par l'événement (médiateur sans PTY, /// utilisé par les tests événementiels — comportement historique). headless_sink: Option, + /// 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>, + /// 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 /// 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 @@ -124,6 +141,8 @@ impl BusyTracker { deferred: Mutex::new(HashMap::new()), events, headless_sink: None, + mailbox: None, + grace: PROMPT_READY_GRACE, } } @@ -336,16 +355,34 @@ impl BusyTracker { self.publish_deferred(agent, d); } 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. + // Pas de tour différé : la cible est revenue à son prompt sans + // `idea_reply`. On capture le ticket actif AVANT `mark_idle` (qui efface + // l'état Busy), puis on libère le busy state / la FIFO comme avant. Le + // prompt-ready ne porte AUCUNE réponse (`no_reply_payload`) : pour ne pas + // laisser l'appelant bloqué jusqu'au timeout long, on **arme une fenêtre + // 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!( - "[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); + 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 /// watcher thread un-arms its own entry on exit. watched: Arc>>, + /// 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 { @@ -461,9 +501,17 @@ impl MediatedInbox { pub fn new(mailbox: Arc, clock: Arc) -> Self { let handles = Arc::new(Mutex::new(HashMap::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 { mailbox, - tracker: Self::build_tracker(None, None, &handles, &front_owned), + tracker, clock, pty: None, handles, @@ -471,6 +519,7 @@ impl MediatedInbox { submit: Mutex::new(HashMap::new()), stall: Mutex::new(HashMap::new()), watched: Arc::new(Mutex::new(HashSet::new())), + grace: PROMPT_READY_GRACE, } } @@ -481,8 +530,14 @@ impl MediatedInbox { pty: Option<&Arc>, handles: &Arc>>, front_owned: &Arc>>, + mailbox: &Arc, + grace: Duration, ) -> Arc { 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 { tracker.headless_sink = Some(Self::make_headless_sink( Arc::clone(pty), @@ -601,6 +656,27 @@ impl MediatedInbox { self.pty.as_ref(), &self.handles, &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 } @@ -617,9 +693,17 @@ impl MediatedInbox { let handles = Arc::new(Mutex::new(HashMap::new())); let front_owned = Arc::new(Mutex::new(HashSet::new())); let pty_opt = Some(pty); + let tracker = Self::build_tracker( + None, + pty_opt.as_ref(), + &handles, + &front_owned, + &mailbox, + PROMPT_READY_GRACE, + ); Self { mailbox, - tracker: Self::build_tracker(None, pty_opt.as_ref(), &handles, &front_owned), + tracker, clock, pty: pty_opt, handles, @@ -627,6 +711,7 @@ impl MediatedInbox { submit: Mutex::new(HashMap::new()), stall: Mutex::new(HashMap::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")); // Resolve through the shared mailbox (the orchestrator's path). 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] @@ -1367,7 +1455,10 @@ mod tests { assert_eq!(inbox.mailbox().pending(&a), 1); // Nothing answered the caller via preempt. 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] @@ -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, grace: Duration) -> MediatedInbox { + MediatedInbox::with_pty( + Arc::new(InMemoryMailbox::new()), + Arc::new(FixedClock(1)), + pty as Arc, + ) + .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] fn prompt_pattern_present_and_output_contains_it_flips_busy_to_idle() { let pty = Arc::new(FakePty::new()); diff --git a/crates/infrastructure/src/mailbox/mod.rs b/crates/infrastructure/src/mailbox/mod.rs index ae12b1c..b0c3f6b 100644 --- a/crates/infrastructure/src/mailbox/mod.rs +++ b/crates/infrastructure/src/mailbox/mod.rs @@ -27,13 +27,18 @@ use tokio::sync::oneshot; use domain::ids::AgentId; use domain::mailbox::{ AgentMailbox, AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, - TicketId, + TicketId, TurnResolution, }; /// 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 { ticket: Ticket, - reply: oneshot::Sender, + reply: oneshot::Sender, } /// In-memory, per-agent FIFO mailbox (the production [`AgentMailbox`]). @@ -76,7 +81,7 @@ impl InMemoryMailbox { impl AgentMailbox for InMemoryMailbox { fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply { - let (tx, rx) = oneshot::channel::(); + let (tx, rx) = oneshot::channel::(); let ticket_id = ticket.id; let (depth_before, depth_after) = { 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: // the reply is simply discarded, the head ticket is already retired. - let _ = slot.reply.send(result); + let _ = slot.reply.send(TurnResolution::Replied(result)); Ok(()) } @@ -174,7 +179,7 @@ impl AgentMailbox for InMemoryMailbox { "[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(TurnResolution::Replied(result)); Ok(()) } @@ -212,6 +217,59 @@ impl AgentMailbox for InMemoryMailbox { 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 { @@ -264,7 +322,7 @@ mod tests { mb.resolve(a, "done X".to_owned()).expect("resolve ok"); 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"); } @@ -282,14 +340,14 @@ mod tests { // First resolve goes to the FIRST (head) ticket. 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. assert_eq!( mb.head_ticket(&a), Some(TicketId::from_uuid(uuid::Uuid::from_u128(11))) ); 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] @@ -304,7 +362,7 @@ mod tests { mb.resolve(b, "rb".to_owned()).unwrap(); assert_eq!(mb.pending(&a), 1, "A's queue is independent of B's"); mb.resolve(a, "ra".to_owned()).unwrap(); - assert_eq!(pa.await.unwrap(), "ra"); + assert_eq!(pa.await.unwrap(), TurnResolution::Replied("ra".to_owned())); } #[test] @@ -337,7 +395,78 @@ mod tests { // The next ticket is now resolvable normally. 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]