blomios/IdeaSDK
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6bd5e64dd1a12ff5e4269fa16af3201d7072d16a
IdeaSDK/crates/infrastructure/src/input/mod.rs

2291 lines
96 KiB
Rust
Raw Blame History

//! [`MediatedInbox`] — the [`InputMediator`] adapter (lot C2).
//!
//! The single convergence point of an agent's input. It **composes** the existing
//! [`InMemoryMailbox`] (FIFO + one-shot reply — the correlation engine) and adds the
//! two things the mailbox alone does not express:
//!
//! - a **busy/turn** bookkeeping per agent ([`AgentBusyState`]), so the front can be
//! told when an agent is processing;
//! - a **preempt** signal distinct from `enqueue` (Interrompre ≠ Envoyer): it does
//! **not** queue anything and correlates no ticket.
//!
//! It does **not** spawn a second queue: the FIFO is the mailbox's. The first
//! enqueue while `Idle` starts a turn (agent → `Busy`); `mark_idle` ends it and lets
//! the next ticket start. In doubt we stay `Busy` but **keep accepting** enqueues
//! (forward, never reject — cf. cadrage §6 fallback).
//!
//! ## Concurrency
//!
//! Busy state lives behind a **synchronous** [`Mutex`], held only for O(1) reads and
//! mutations and **never across an `.await`** (the await is the caller's, on the
//! returned [`PendingReply`]). The mailbox owns its own locking.
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use domain::events::DomainEvent;
use domain::ids::RuntimeAgentKey;
use domain::inbox::{
AgentInbox, AgentInboxSnapshot, InboxError, InboxItem, InboxReceipt, InboxReceiptStatus,
InboxSource, DEFAULT_AGENT_INBOX_CAPACITY,
};
use domain::input::{AgentBusyState, AgentLiveness, InputMediator, SubmitConfig};
use domain::mailbox::{AgentMailbox, AgentQueueSnapshot, PendingReply, Ticket, TicketId};
use domain::ports::{EventBus, PtyHandle, PtyPort};
use crate::mailbox::InMemoryMailbox;
/// Shared busy/idle bookkeeping for one set of agents.
///
/// Extracted so the **prompt-ready watcher** (a detached thread observing an agent's
/// PTY output, lot C5) can flip an agent back to `Idle` without holding the whole
/// [`MediatedInbox`]: it only needs the busy map + the event bus. This is the single
/// authority for the `Busy→Idle` transition and its `AgentBusyChanged` event, so
/// every path (explicit `mark_idle`, prompt-ready match) stays consistent.
struct BusyTracker {
busy: Mutex<HashMap<RuntimeAgentKey, AgentBusyState>>,
/// Per-agent **liveness** bookkeeping (lot 2) : dernier battement observé, seuil de
/// stagnation issu du profil, et état de vivacité courant pour n'émettre
/// `AgentLivenessChanged` qu'**une fois par transition** (pas de spam).
liveness: Mutex<HashMap<RuntimeAgentKey, LivenessState>>,
/// **Démarrage à froid** (fix race cold-launch) : ensemble des agents fraîchement
/// lancés à froid pour lesquels la livraison du **premier** tour doit être *gatée*
/// sur la readiness MCP. Un agent y est inscrit par
/// [`BusyTracker::mark_starting`] (uniquement quand un pont MCP le libérera), puis
/// consommé par l'`enqueue` qui démarre le tour : la `DelegationReady`
/// est alors **différée** dans `deferred` au lieu d'être publiée immédiatement (le
/// CLI n'a pas encore chargé ses outils MCP). Vide ⇒ comportement chaud inchangé.
starting: Mutex<HashSet<RuntimeAgentKey>>,
/// **Tour différé** (fix race cold-launch) : payload de la `DelegationReady` retenue
/// pour un démarrage à froid, publiée par [`BusyTracker::release_cold_start`] à la
/// connexion du pont MCP (jamais avant). Absent ⇒ aucun tour en attente de gate.
deferred: Mutex<HashMap<RuntimeAgentKey, DeferredDelegation>>,
/// **Latch « déjà libéré »** (fix race cold-start, ordre inverse) : ensemble des
/// agents pour lesquels le signal de readiness (`release_cold_start`, pont MCP) est
/// arrivé **avant** que l'`enqueue` n'ait parqué son tour dans
/// `deferred`. Sans ce latch, ce signal trouverait `deferred` vide, ne ferait rien
/// d'utile (l'agent encore `starting` ⇒ pas de `mark_idle`), puis l'`enqueue`
/// parquerait le tour dans `deferred` — où il resterait à jamais (le watcher
/// prompt-ready PTY ayant été supprimé, aucun second signal ne viendrait le drainer). Le
/// latch enregistre « cet agent en démarrage est déjà prêt » : l'`enqueue` qui suit
/// livre alors **immédiatement** au lieu de parquer. Consommé (retiré) à la livraison
/// ⇒ exactement-une-fois, quel que soit l'ordre. Vide ⇒ comportement inchangé.
released: Mutex<HashSet<RuntimeAgentKey>>,
events: Option<Arc<dyn EventBus>>,
/// Sink de livraison headless (cf. [`HeadlessSink`]). Câblé par
/// [`MediatedInbox::with_pty`]/[`MediatedInbox::with_events`] quand un PTY est
/// 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<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
/// 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
/// bon moment (et un seul `DelegationReady`, comme pour un agent chaud).
#[derive(Debug, Clone)]
struct DeferredDelegation {
ticket: TicketId,
text: String,
submit_sequence: Option<String>,
submit_delay_ms: Option<u32>,
}
/// Sink de livraison « headless » optionnel branché sur le [`BusyTracker`].
///
/// Appelé pour **chaque** tour à livrer (chemin chaud immédiat comme drains à froid).
/// Reçoit l'agent + le payload de délégation et décide :
/// - `None` ⇒ il a **pris en charge** la livraison (écriture directe dans le PTY d'un
/// agent headless qui n'a pas de cellule frontend) ;
/// - `Some(d)` ⇒ il **rend la main** : le tracker publie alors le `DelegationReady`
/// normal (cellule frontend présente ⇒ c'est le write-portal qui écrira, ou pas de
/// PTY/handle disponible ⇒ repli sur l'événement, comportement historique).
type HeadlessSink =
Arc<dyn Fn(RuntimeAgentKey, DeferredDelegation) -> Option<DeferredDelegation> + Send + Sync>;
/// Délai (ms) entre l'écriture du texte de la tâche et celle de la séquence de
/// soumission lors d'une livraison **headless** (le médiateur écrit lui-même le PTY).
/// Aligné sur le défaut du write-portal frontend (`DEFAULT_SUBMIT_DELAY_MS`) : sépare
/// la soumission du collage pour esquiver la paste-detection des CLI. Utilisé seulement
/// quand le profil de la cible ne fournit pas de `submit_delay_ms`.
const DEFAULT_SUBMIT_DELAY_MS: u32 = 350;
/// Taille maximale d'un fragment de consigne écrit dans le PTY en mode headless.
const DELEGATION_WRITE_CHUNK_BYTES: usize = 512;
/// Pause entre deux fragments de consigne en mode headless.
const DELEGATION_WRITE_CHUNK_DELAY_MS: u64 = 8;
/// État de vivacité d'un agent maintenu par le [`BusyTracker`] (lot 2).
///
/// Pur (aucune I/O) : un timestamp `last_seen_ms` rafraîchi à chaque battement, le
/// seuil `stall_after_ms` du profil (cf. [`domain::profile::LivenessStrategy`]) et
/// l'état courant `current` pour décider d'une **transition** (et donc d'un seul
/// événement). `stall_after_ms = None` ⇒ pas de détection (comportement legacy).
#[derive(Debug, Clone, Copy)]
struct LivenessState {
/// Epoch-millis du dernier battement (ou du démarrage du tour).
last_seen_ms: u64,
/// Seuil de stagnation du profil de l'agent. `None` ⇒ détection désactivée.
stall_after_ms: Option<u32>,
/// État de vivacité courant (pour n'émettre qu'à la transition).
current: AgentLiveness,
}
impl BusyTracker {
fn new(events: Option<Arc<dyn EventBus>>) -> Self {
Self {
busy: Mutex::new(HashMap::new()),
liveness: Mutex::new(HashMap::new()),
starting: Mutex::new(HashSet::new()),
deferred: Mutex::new(HashMap::new()),
released: Mutex::new(HashSet::new()),
events,
headless_sink: None,
mailbox: None,
grace: PROMPT_READY_GRACE,
}
}
fn lock_starting(&self) -> std::sync::MutexGuard<'_, HashSet<RuntimeAgentKey>> {
self.starting
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_deferred(
&self,
) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, DeferredDelegation>> {
self.deferred
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_released(&self) -> std::sync::MutexGuard<'_, HashSet<RuntimeAgentKey>> {
self.released
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// Marque un agent **en démarrage à froid** : son tout premier tour devra être
/// *gaté* sur la readiness MCP (la `DelegationReady` sera différée jusqu'à la
/// connexion du pont MCP). À n'appeler **que** lorsqu'un pont MCP libérera le tour
/// (`release_cold_start`), sinon le premier tour resterait bloqué indéfiniment
/// (aucun signal ne viendrait le libérer). Sans cet appel, l'`enqueue` publie la
/// `DelegationReady` immédiatement (chemin chaud, zéro régression).
fn mark_starting(&self, agent: RuntimeAgentKey) {
application::diag!("[input-mediator] mark_starting cold-start gate armed agent={agent}");
self.lock_starting().insert(agent);
}
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, AgentBusyState>> {
self.busy
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_liveness(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, LivenessState>> {
self.liveness
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// Publie un `AgentLivenessChanged` (si un bus est câblé).
fn publish_liveness(&self, agent: RuntimeAgentKey, liveness: AgentLiveness) {
if let Some(events) = &self.events {
events.publish(DomainEvent::AgentLivenessChanged {
agent_id: agent.agent_id,
liveness,
});
}
}
/// Enregistre le seuil de stagnation du profil d'un agent au moment où son tour
/// démarre (depuis l'enqueue) : (re)initialise `last_seen` à `now` et repart d'un
/// état `Alive`. Sans seuil (`None`), l'entrée existe quand même mais le sweep ne
/// la déclarera jamais `Stalled` (zéro régression pour un profil sans liveness).
fn arm_liveness(&self, agent: RuntimeAgentKey, stall_after_ms: Option<u32>, now_ms: u64) {
self.lock_liveness().insert(
agent,
LivenessState {
last_seen_ms: now_ms,
stall_after_ms,
current: AgentLiveness::Alive,
},
);
}
/// Rafraîchit le `last_seen` d'un agent (un **battement**) et, s'il était
/// `Stalled`, le ramène à `Alive` en émettant l'unique transition de reprise. No-op
/// si l'agent n'a pas d'entrée de vivacité armée (tour non structuré / legacy).
fn touch(&self, agent: RuntimeAgentKey, now_ms: u64) {
let recovered = {
let mut map = self.lock_liveness();
let Some(state) = map.get_mut(&agent) else {
return;
};
state.last_seen_ms = now_ms;
if state.current == AgentLiveness::Stalled {
state.current = AgentLiveness::Alive;
true
} else {
false
}
};
if recovered {
self.publish_liveness(agent, AgentLiveness::Alive);
}
}
/// Balaye tous les agents et, pour chacun dont le tour stagne
/// (`now - last_seen > stall_after_ms` et seuil défini), bascule `Alive→Stalled`
/// en émettant **une seule** transition. **Pure et testable sans horloge réelle** :
/// `now_ms` est passé en paramètre. Idempotente : un agent déjà `Stalled` ne ré-émet
/// pas. Un agent sans seuil (`None`) ou redevenu `Idle` n'est jamais déclaré stalled.
fn sweep_stalled(&self, now_ms: u64) {
let newly_stalled: Vec<RuntimeAgentKey> = {
let mut map = self.lock_liveness();
map.iter_mut()
.filter_map(|(agent, state)| {
let threshold = state.stall_after_ms?;
if state.current != AgentLiveness::Alive {
return None; // déjà Stalled : pas de ré-émission.
}
let elapsed = now_ms.saturating_sub(state.last_seen_ms);
if elapsed > u64::from(threshold) {
state.current = AgentLiveness::Stalled;
Some(*agent)
} else {
None
}
})
.collect()
};
for agent in newly_stalled {
// Transition de vivacité Alive→Stalled : le tour de l'agent n'a pas battu
// depuis plus que son seuil. Beacon événementiel (une fois par transition,
// jamais en boucle) — utile pour repérer une cible silencieuse pendant un ask.
application::diag!("[input-mediator] liveness Alive->Stalled agent={agent}");
self.publish_liveness(agent, AgentLiveness::Stalled);
}
}
/// Retire l'entrée de vivacité d'un agent (fin de tour). Si l'agent était `Stalled`,
/// la fin de tour est en soi un retour à `Alive` ⇒ on émet la transition de reprise.
fn clear_liveness(&self, agent: RuntimeAgentKey) {
let was_stalled = self
.lock_liveness()
.remove(&agent)
.is_some_and(|s| s.current == AgentLiveness::Stalled);
if was_stalled {
self.publish_liveness(agent, AgentLiveness::Alive);
}
}
fn busy_state(&self, agent: RuntimeAgentKey) -> AgentBusyState {
self.lock()
.get(&agent)
.copied()
.unwrap_or(AgentBusyState::Idle)
}
/// Marks `agent` `Busy` if it was `Idle`, returning whether a turn actually
/// started (so the caller publishes `AgentBusyChanged{busy:true}` only once).
fn start_turn(&self, agent: RuntimeAgentKey, state: AgentBusyState) -> bool {
let mut busy = self.lock();
let entry = busy.entry(agent).or_insert(AgentBusyState::Idle);
if entry.is_busy() {
application::diag!("[input-mediator] start_turn queued behind busy agent={agent}");
false
} else {
application::diag!("[input-mediator] start_turn started agent={agent} state={state:?}");
*entry = state;
true
}
}
/// Publie une [`DomainEvent::DelegationReady`] depuis un payload différé (si un bus
/// est câblé). Utilisée pour livrer le **premier** tour d'un agent froid au moment
/// où son prompt apparaît.
fn publish_deferred(&self, agent: RuntimeAgentKey, d: DeferredDelegation) {
// Point de livraison unique (tours chauds immédiats ET drains à froid). Si un
// sink headless est câblé, il a la priorité : pour un agent sans cellule
// frontend il écrit lui-même la tâche dans le PTY et renvoie `None` (pris en
// charge) ; sinon il renvoie `Some(d)` et on retombe sur l'événement.
let d = match &self.headless_sink {
Some(sink) => match sink(agent, d) {
Some(d) => d,
None => {
application::diag!(
"[input-mediator] delegationReady handled by headless sink agent={agent}"
);
return;
}
},
None => d,
};
if let Some(events) = &self.events {
application::diag!(
"[input-mediator] delegationReady -> write-portal agent={agent} ticket={} text_bytes={} submit_sequence={} submit_delay_ms={:?}",
d.ticket,
d.text.as_bytes().len(),
describe_submit_sequence(d.submit_sequence.as_deref()),
d.submit_delay_ms,
);
events.publish(DomainEvent::DelegationReady {
agent_id: agent.agent_id,
ticket: d.ticket,
text: d.text,
submit_sequence: d.submit_sequence,
submit_delay_ms: d.submit_delay_ms,
});
}
}
/// Signal de **fin de tour** émis par le [`domain::ports::TurnWatcher`] à l'apparition
/// d'une fin de tour dans le transcript (Claude `turn_duration`). Remplace **verbatim**
/// l'ancienne branche `None` du watcher prompt-ready PTY supprimé : seul le déclencheur
/// change (transcript au lieu d'un sigil PTY).
///
/// La cible a terminé son tour 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. La
/// fin de tour ne porte AUCUNE réponse : 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). Sur un agent déjà `Idle` (aucun ticket actif) ⇒ simple
/// `mark_idle` idempotent, pas de grâce.
fn turn_ended(&self, agent: RuntimeAgentKey) {
let active = self.lock().get(&agent).and_then(AgentBusyState::ticket);
application::diag!(
"[input-mediator] turn_ended 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] turn_ended 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);
});
}
}
/// Libère un premier tour différé sur **readiness de démarrage** (connexion du pont
/// MCP de l'agent). Draine le `DeferredDelegation` retenu s'il existe (et retire
/// l'agent de `starting`) ; sinon no-op. **Pas de `mark_idle`** (signal de démarrage,
/// pas de fin de tour). Idempotent : c'est le seul drain du tour différé (le `remove`
/// ne rend `Some` qu'une fois).
fn release_cold_start(&self, agent: RuntimeAgentKey) {
let was_starting = self.lock_starting().remove(&agent);
if let Some(d) = self.lock_deferred().remove(&agent) {
// Cas nominal : l'`enqueue` a déjà parqué le tour ⇒ on le draine.
application::diag!(
"[input-mediator] release_cold_start (mcp-ready) released deferred delegation agent={agent} ticket={}",
d.ticket
);
self.publish_deferred(agent, d);
} else if was_starting {
// **Ordre inverse de la race** : la readiness MCP est arrivée AVANT que
// l'`enqueue` ait parqué son tour. On pose le latch « déjà libéré » : le
// prochain `enqueue` (agent encore vu comme démarrant) livrera immédiatement au
// lieu de parquer dans `deferred` — où le tour resterait à jamais (le watcher
// prompt-ready PTY ayant été supprimé, aucun second signal ne le drainerait).
// Hors démarrage à froid (`was_starting == false`), aucun latch : no-op.
application::diag!(
"[input-mediator] release_cold_start before deferred enqueue (cold-start) -> released latch agent={agent}"
);
self.lock_released().insert(agent);
}
}
/// Marks `agent` `Idle`, publishing `AgentBusyChanged{busy:false}` only on a real
/// `Busy→Idle` transition. Idempotent: a `mark_idle` on an already-idle agent is a
/// no-op and emits nothing.
fn mark_idle(&self, agent: RuntimeAgentKey) {
let was_busy = {
let mut busy = self.lock();
busy.insert(agent, AgentBusyState::Idle)
.is_some_and(|s| s.is_busy())
};
if was_busy {
application::diag!("[input-mediator] mark_idle turn ended agent={agent}");
if let Some(events) = &self.events {
events.publish(DomainEvent::AgentBusyChanged {
agent_id: agent.agent_id,
busy: false,
});
}
}
// Fin de tour : retirer l'entrée de vivacité (émet la reprise si l'agent
// était `Stalled`). Indépendant de `was_busy` pour rester idempotent.
self.clear_liveness(agent);
}
}
/// Supplies the epoch-millis stamp used for `AgentBusyState::Busy { since_ms }`.
///
/// Injectable so tests are deterministic and the adapter stays decoupled from the
/// wall clock (the composition root wires the real clock).
pub trait MillisClock: Send + Sync {
/// Current time as milliseconds since the Unix epoch.
fn now_ms(&self) -> u64;
}
/// Wall-clock implementation of [`MillisClock`] (composition-root default).
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemMillisClock;
impl MillisClock for SystemMillisClock {
fn now_ms(&self) -> u64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
.unwrap_or(0)
}
}
/// In-memory mediated inbox: one FIFO per agent (the mailbox) plus busy state.
///
/// Since ARCHITECTURE §20, `enqueue` **no longer writes the turn into the PTY** (the
/// `\n` band-aid is gone — it never submitted in raw mode and produced a "double chat").
/// On the enqueue that starts a turn (Idle→Busy) it publishes a
/// [`DomainEvent::DelegationReady`] carrying the task text + ticket + the target's
/// submit config; the **frontend** write-portal owns the single physical PTY write. A
/// wired [`PtyPort`] is kept only to observe the output stream for prompt-ready
/// detection (lot C5) and to deliver the `preempt` interrupt byte.
pub struct MediatedInbox {
mailbox: Arc<InMemoryMailbox>,
/// Shared busy/idle authority (also handed to prompt-ready watcher threads, C5).
tracker: Arc<BusyTracker>,
clock: Arc<dyn MillisClock>,
/// Optional PTY port: present ⇒ the inbox owns the turn-delivery write **and** can
/// observe an agent's output stream for prompt-ready detection (lot C5).
pty: Option<Arc<dyn PtyPort>>,
/// Per-agent live input handle (one stream per agent), fed by `bind_handle`.
/// `Arc` so the headless delivery sink (wired into the [`BusyTracker`]) can read it
/// to resolve an agent's PTY handle when it must write the turn itself.
handles: Arc<Mutex<HashMap<RuntimeAgentKey, PtyHandle>>>,
/// Agents qui ont une **cellule terminal frontend montée** (write-portal actif),
/// tenu à jour par [`InputMediator::set_front_attached`]. Quand un agent y figure,
/// la livraison passe par l'événement `DelegationReady` (le front écrit) ; sinon
/// (agent headless / délégué en arrière-plan) le médiateur écrit lui-même le tour
/// dans le PTY. `Arc` car le sink headless du tracker le consulte.
front_owned: Arc<Mutex<HashSet<RuntimeAgentKey>>>,
/// Per-agent submit config (target profile's `submit_sequence`/`submit_delay_ms`),
/// stashed at bind time (§20.3) and echoed on the `DelegationReady` event when a
/// turn starts. Absent ⇒ both `None` (the front applies its defaults).
submit: Mutex<HashMap<RuntimeAgentKey, SubmitConfig>>,
/// Per-agent stall threshold (`LivenessStrategy::stall_after_ms`, lot 2), stashed by
/// `set_stall_threshold` and consumed to arm a fresh liveness window on the enqueue
/// that starts a turn. Absent ⇒ `None` (no stall detection — legacy behaviour).
stall: Mutex<HashMap<RuntimeAgentKey, Option<u32>>>,
/// Rich inbox payloads keyed by the mailbox ticket id. The mailbox remains the
/// only FIFO; this map is metadata only.
inbox_items: Mutex<HashMap<TicketId, InboxItem>>,
/// Bounded inbox capacity per agent.
inbox_capacity: usize,
/// 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 {
/// Builds an inbox over a shared [`InMemoryMailbox`] with the given clock, without
/// turn delivery (the orchestrator writes the turn itself).
#[must_use]
pub fn new(mailbox: Arc<InMemoryMailbox>, clock: Arc<dyn MillisClock>) -> 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,
clock,
pty: None,
handles,
front_owned,
submit: Mutex::new(HashMap::new()),
stall: Mutex::new(HashMap::new()),
inbox_items: Mutex::new(HashMap::new()),
inbox_capacity: configured_inbox_capacity(),
grace: PROMPT_READY_GRACE,
}
}
/// Construit le [`BusyTracker`] avec, quand un PTY est présent, le **sink headless**
/// branché (cf. [`HeadlessSink`]). Sans PTY ⇒ aucun sink (livraison par événement).
fn build_tracker(
events: Option<Arc<dyn EventBus>>,
pty: Option<&Arc<dyn PtyPort>>,
handles: &Arc<Mutex<HashMap<RuntimeAgentKey, PtyHandle>>>,
front_owned: &Arc<Mutex<HashSet<RuntimeAgentKey>>>,
mailbox: &Arc<InMemoryMailbox>,
grace: Duration,
) -> Arc<BusyTracker> {
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),
Arc::clone(handles),
Arc::clone(front_owned),
));
}
Arc::new(tracker)
}
/// Fabrique le sink de livraison headless : pour un agent **sans cellule frontend**
/// (absent de `front_owned`), écrit la tâche dans son PTY (texte, puis — après le
/// délai anti-paste-detection — la séquence de soumission) et renvoie `None` (pris
/// en charge). Pour un agent avec cellule, ou sans handle PTY connu, renvoie
/// `Some(d)` ⇒ le tracker publie l'événement `DelegationReady` comme avant.
fn make_headless_sink(
pty: Arc<dyn PtyPort>,
handles: Arc<Mutex<HashMap<RuntimeAgentKey, PtyHandle>>>,
front_owned: Arc<Mutex<HashSet<RuntimeAgentKey>>>,
) -> HeadlessSink {
Arc::new(move |agent: RuntimeAgentKey, d: DeferredDelegation| {
// Cellule frontend montée ⇒ c'est le write-portal qui écrit (et qui sait
// composer avec une saisie humaine en cours). On rend la main.
if front_owned
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.contains(&agent)
{
eprintln!(
"[input-mediator] front-owned delivery selected agent={agent} ticket={}",
d.ticket
);
return Some(d);
}
// Agent headless : récupérer son handle PTY. Absent ⇒ repli sur l'événement
// (best-effort, ne devrait pas arriver pour un agent qui livre un tour).
let handle = handles
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(&agent)
.cloned();
let Some(handle) = handle else {
eprintln!(
"[input-mediator] no PTY handle for headless delivery agent={agent} ticket={}",
d.ticket
);
return Some(d);
};
let pty = Arc::clone(&pty);
let text = d.text;
let ticket = d.ticket;
let submit = d
.submit_sequence
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "\r".to_owned());
let delay = u64::from(d.submit_delay_ms.unwrap_or(DEFAULT_SUBMIT_DELAY_MS));
// Écriture sur un thread détaché : ne JAMAIS bloquer l'appelant (thread du
// watcher prompt-ready, tâche tokio du serveur MCP, ou le fil de l'enqueue).
// Texte d'abord, puis la séquence de soumission après le délai — comme le
// write-portal frontend — pour esquiver la détection de coller des CLI.
// La consigne est fragmentée : les TUI (Codex surtout) peuvent traiter un
// gros write unique comme un paste et garder/perdre la fin au démarrage.
std::thread::spawn(move || {
eprintln!(
"[input-mediator] headless write text start agent={agent} ticket={ticket} handle={} text_bytes={} submit_sequence={} submit_delay_ms={delay}",
handle.session_id,
text.as_bytes().len(),
describe_submit_sequence(Some(&submit)),
);
match write_delegation_chunks(pty.as_ref(), &handle, &text) {
Ok(()) => eprintln!(
"[input-mediator] headless write text ok agent={agent} ticket={ticket} handle={}",
handle.session_id
),
Err(e) => {
eprintln!(
"[input-mediator] headless write text failed agent={agent} ticket={ticket} handle={} error={e}",
handle.session_id
);
return;
}
}
if delay > 0 {
std::thread::sleep(std::time::Duration::from_millis(delay));
}
eprintln!(
"[input-mediator] headless submit start agent={agent} ticket={ticket} handle={} submit_sequence={}",
handle.session_id,
describe_submit_sequence(Some(&submit)),
);
match pty.write(&handle, submit.as_bytes()) {
Ok(()) => eprintln!(
"[input-mediator] headless submit ok agent={agent} ticket={ticket} handle={}",
handle.session_id
),
Err(e) => eprintln!(
"[input-mediator] headless submit failed agent={agent} ticket={ticket} handle={} error={e}",
handle.session_id
),
}
});
None
})
}
/// Wires an [`EventBus`] so busy/idle transitions publish
/// [`DomainEvent::AgentBusyChanged`] at their source (cadrage C4 §4.2). Builder
/// additive: callers that do not wire a bus stay silent.
#[must_use]
pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
// Reconstruit le tracker avec le bus ET — si un PTY est déjà câblé (cas
// `with_pty().with_events()` du composition root) — le sink headless, sinon
// l'ordre des builders perdrait la livraison headless.
self.tracker = Self::build_tracker(
Some(events),
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
}
/// Builds an inbox that **delivers** the turn through `pty` to each agent's bound
/// handle (cadrage C3 §5.2). Use [`MediatedInbox::bind_handle`] to register the
/// agent's live handle before/at enqueue time.
#[must_use]
pub fn with_pty(
mailbox: Arc<InMemoryMailbox>,
clock: Arc<dyn MillisClock>,
pty: Arc<dyn PtyPort>,
) -> Self {
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,
clock,
pty: pty_opt,
handles,
front_owned,
submit: Mutex::new(HashMap::new()),
stall: Mutex::new(HashMap::new()),
inbox_items: Mutex::new(HashMap::new()),
inbox_capacity: configured_inbox_capacity(),
grace: PROMPT_READY_GRACE,
}
}
/// Convenience constructor over a fresh mailbox and the wall clock.
#[must_use]
pub fn in_memory() -> Self {
Self::new(
Arc::new(InMemoryMailbox::new()),
Arc::new(SystemMillisClock),
)
}
/// Overrides bounded inbox capacity. Intended for tests and explicit wiring.
#[must_use]
pub fn with_inbox_capacity(mut self, capacity: usize) -> Self {
self.inbox_capacity = capacity;
self
}
fn handles(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, PtyHandle>> {
self.handles
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn submit(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, SubmitConfig>> {
self.submit
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn stall(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, Option<u32>>> {
self.stall
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn inbox_items(&self) -> std::sync::MutexGuard<'_, HashMap<TicketId, InboxItem>> {
self.inbox_items
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// **Détection de stagnation** (lot 2) : balaie les agents `Busy` et bascule
/// `Alive→Stalled` ceux dont le dernier battement remonte à plus de
/// `stall_after_ms`. Émet `AgentLivenessChanged` **une fois par transition**. La
/// logique de décision est **pure** ([`BusyTracker::sweep_stalled`]) : `now_ms` est
/// fourni par l'horloge injectée, donc testable sans horloge réelle. Destinée à
/// être appelée périodiquement par une tâche détenue au composition root.
pub fn sweep_stalled(&self) {
self.tracker.sweep_stalled(self.clock.now_ms());
}
/// The underlying mailbox (e.g. for `cancel_head` / `resolve` from the orchestrator).
#[must_use]
pub fn mailbox(&self) -> Arc<InMemoryMailbox> {
Arc::clone(&self.mailbox)
}
}
/// Composes the text delivered into a human-facing terminal turn.
///
/// Inter-agent conversation no longer uses this path: structured/headless asks send the
/// raw task through `AgentSession::send` and capture the model `Final`. Therefore this
/// terminal delivery must not inject any inter-agent ticket/reply protocol.
fn delegation_preamble(requester: &str, ticket: TicketId, task: &str) -> String {
let _ = (requester, ticket);
task.to_owned()
}
fn write_delegation_chunks(
pty: &dyn PtyPort,
handle: &PtyHandle,
text: &str,
) -> Result<(), domain::ports::PtyError> {
let mut start = 0;
let mut current_len = 0;
for (idx, ch) in text.char_indices() {
let len = ch.len_utf8();
if current_len > 0 && current_len + len > DELEGATION_WRITE_CHUNK_BYTES {
pty.write(handle, text[start..idx].as_bytes())?;
std::thread::sleep(std::time::Duration::from_millis(
DELEGATION_WRITE_CHUNK_DELAY_MS,
));
start = idx;
current_len = 0;
}
current_len += len;
}
pty.write(handle, text[start..].as_bytes())
}
fn describe_submit_sequence(value: Option<&str>) -> String {
match value {
None => "<default>".to_owned(),
Some("") => "<empty>".to_owned(),
Some(value) => value
.chars()
.map(|ch| match ch {
'\r' => "\\r".to_owned(),
'\n' => "\\n".to_owned(),
'\u{7f}' => "\\x7f".to_owned(),
ch if ch.is_control() => format!("\\x{:02x}", ch as u32),
ch => ch.to_string(),
})
.collect::<Vec<_>>()
.join(""),
}
}
impl InputMediator for MediatedInbox {
fn enqueue(&self, agent: RuntimeAgentKey, ticket: Ticket) -> PendingReply {
let ticket_id = ticket.id;
// If the agent is Idle, this enqueue starts its turn ⇒ go Busy. If already
// Busy, we still accept (queue grows; the turn advances on mark_idle) — never
// reject the sender (forward fallback).
let now_ms = self.clock.now_ms();
let started_turn = self.tracker.start_turn(
agent,
AgentBusyState::Busy {
ticket: ticket_id,
since_ms: now_ms,
},
);
// Sur l'enqueue qui **démarre** un tour : armer une fenêtre de vivacité fraîche
// avec le seuil de stagnation stashé du profil (lot 2). `last_seen = now`, état
// `Alive`. Un profil sans seuil (`None`) arme une entrée jamais déclarée stalled
// (zéro régression). Un re-enqueue pendant `Busy` ne ré-arme pas (le tour court).
if started_turn {
let stall_after_ms = self.stall().get(&agent).copied().flatten();
self.tracker.arm_liveness(agent, stall_after_ms, now_ms);
}
// Publish only on the enqueue that **starts** a turn (Idle→Busy); a second
// enqueue while Busy queues behind without re-announcing (cadrage C4 §4.2,
// ARCHITECTURE §20). Published outside the busy mutex (the tracker released it
// above). On a started turn we emit BOTH the advisory busy beacon AND the new
// `DelegationReady` carrying the task text + ticket + the target's submit
// config: the backend NO LONGER writes the turn into the PTY (the `\n` band-aid
// is gone — it never submitted in raw mode). The frontend write-portal owns the
// physical write (text + submit_sequence) through the single PTY writer.
if started_turn {
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentBusyChanged {
agent_id: agent.agent_id,
busy: true,
});
let submit = self.submit().get(&agent).cloned().unwrap_or_default();
let text = delegation_preamble(&ticket.requester, ticket_id, &ticket.task);
// **Fix race cold-launch** : si l'agent est en démarrage à froid (inscrit
// par `mark_starting` quand un pont MCP le libérera), ON DIFFÈRE la
// `DelegationReady` jusqu'à la readiness MCP (le CLI n'a pas encore chargé
// ses outils — la livrer maintenant perdrait le premier tour).
// `release_cold_start` la publiera. Agent chaud (pas dans `starting`) ⇒
// publication immédiate, comportement inchangé.
let deferred = DeferredDelegation {
ticket: ticket_id,
text,
submit_sequence: submit.sequence,
submit_delay_ms: submit.delay_ms,
};
if self.tracker.lock_starting().contains(&agent) {
// Agent en démarrage à froid. Si un signal de readiness est DÉJÀ arrivé
// (latch « déjà libéré » posé par `release_cold_start`
// AVANT cet enqueue — l'ordre inverse de la race), on consomme le latch,
// on retire l'agent du gate et on livre TOUT DE SUITE : sinon ce tour
// serait parqué dans `deferred` sans plus aucun signal pour le drainer
// (blocage éternel). Sinon (readiness pas encore arrivée), on parque
// normalement : le prochain signal le drainera. Exactement-une-fois dans
// les deux ordres (release-avant-enqueue ET enqueue-avant-release).
if self.tracker.lock_released().remove(&agent) {
application::diag!(
"[input-mediator] enqueue sees released latch (cold-start, reverse order) -> deliver now agent={agent} ticket={ticket_id}"
);
self.tracker.lock_starting().remove(&agent);
self.tracker.publish_deferred(agent, deferred);
} else {
application::diag!(
"[input-mediator] enqueue deferred (cold-start gate) agent={agent} ticket={ticket_id}"
);
self.tracker.lock_deferred().insert(agent, deferred);
}
} else {
self.tracker.publish_deferred(agent, deferred);
}
}
}
self.mailbox.enqueue(agent, ticket)
}
fn enqueue_silent(&self, agent: RuntimeAgentKey, ticket: Ticket) -> PendingReply {
let ticket_id = ticket.id;
// Headless/system turns share the same FIFO and busy/liveness accounting as
// terminal-delivered turns, but their prompt is sent through AgentSession::send.
// Do not publish DelegationReady here: that would leak the task into the
// human-facing terminal/write portal while the headless turn is already running.
let now_ms = self.clock.now_ms();
let started_turn = self.tracker.start_turn(
agent,
AgentBusyState::Busy {
ticket: ticket_id,
since_ms: now_ms,
},
);
if started_turn {
let stall_after_ms = self.stall().get(&agent).copied().flatten();
self.tracker.arm_liveness(agent, stall_after_ms, now_ms);
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentBusyChanged {
agent_id: agent.agent_id,
busy: true,
});
}
}
self.mailbox.enqueue(agent, ticket)
}
fn bind_handle(&self, agent: RuntimeAgentKey, handle: PtyHandle) {
eprintln!(
"[input-mediator] bind handle agent={agent} handle={}",
handle.session_id
);
self.handles().insert(agent, handle);
}
fn bind_handle_with_submit(
&self,
agent: RuntimeAgentKey,
handle: PtyHandle,
submit: SubmitConfig,
) {
// Register the input handle exactly like `bind_handle` and stash the target's
// submit config (echoed on `DelegationReady` at the next turn start, §20.3).
// Turn-end detection is NO LONGER armed here: the dead PTY prompt-ready watcher
// was replaced by the transcript `TurnWatcher` (armed at the composition root,
// once per live session) which calls `turn_ended`.
eprintln!(
"[input-mediator] bind handle with submit agent={agent} handle={} submit_sequence={} submit_delay_ms={:?}",
handle.session_id,
describe_submit_sequence(submit.sequence.as_deref()),
submit.delay_ms,
);
self.handles().insert(agent, handle);
self.submit().insert(agent, submit);
}
fn delivers_turn(&self, agent: RuntimeAgentKey) -> bool {
self.pty.is_some() && self.handles().get(&agent).is_some()
}
fn mark_starting(&self, agent: RuntimeAgentKey) {
// Gate du premier tour d'un agent froid : appelé par l'orchestrateur juste après
// un (re)lancement à froid, AVANT le bind/enqueue, et uniquement si un signal de
// readiness le libérera (pont MCP). Consommé par l'`enqueue` qui démarre le tour
// (diffère la `DelegationReady`) puis par `release_cold_start` (la libère).
self.tracker.mark_starting(agent);
}
fn release_cold_start(&self, agent: RuntimeAgentKey) {
self.tracker.release_cold_start(agent);
}
fn set_front_attached(&self, agent: RuntimeAgentKey, attached: bool) {
let mut front = self
.front_owned
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if attached {
eprintln!("[input-mediator] front attached agent={agent}");
front.insert(agent);
} else {
eprintln!("[input-mediator] front detached agent={agent}");
front.remove(&agent);
}
}
fn preempt(&self, agent: RuntimeAgentKey) {
// Interrompre: signals the running turn to stop. It is NOT an enqueue and
// correlates **no** ticket (we never pop/resolve a pending caller — preempt
// must never silently answer one). The only effect is a best-effort interrupt
// byte written into the agent's bound PTY handle: ESC (`\x1b`), the stop key
// CLI agents honour. A missing handle/port is a no-op (the agent simply has no
// live stream to interrupt). The busy state is left untouched: it returns to
// Idle through `mark_idle` (prompt-ready / explicit signal), not here.
application::diag!("[input-mediator] preempt agent={agent} (interrupt byte, no ticket)");
if let Some(pty) = &self.pty {
if let Some(handle) = self.handles().get(&agent).cloned() {
let _ = pty.write(&handle, b"\x1b");
}
}
}
fn mark_idle(&self, agent: RuntimeAgentKey) {
// Single authority: real Busy→Idle only, publishing AgentBusyChanged{busy:false}
// once. Also clears the liveness entry (fin de tour) — émet la reprise si l'agent
// était `Stalled`.
self.tracker.mark_idle(agent);
}
fn turn_ended(&self, agent: RuntimeAgentKey) {
// Déclenché par le `TurnWatcher` transcript (Claude `turn_duration`) : fin de tour
// sans `idea_reply` ⇒ `mark_idle` + fenêtre de grâce pour réveiller l'appelant
// parqué (cf. [`BusyTracker::turn_ended`]). Backstop no-reply.
self.tracker.turn_ended(agent);
}
fn mark_alive(&self, agent: RuntimeAgentKey) {
// Un battement (delta / activité / heartbeat) rafraîchit `last_seen` et ramène
// l'agent à `Alive` s'il était `Stalled` (lot 2).
self.tracker.touch(agent, self.clock.now_ms());
}
fn set_stall_threshold(&self, agent: RuntimeAgentKey, stall_after_ms: Option<u32>) {
// Stashé par l'orchestrateur depuis le profil de la cible AVANT l'enqueue qui
// démarre le tour ; consommé par `arm_liveness` au start_turn (lot 2).
self.stall().insert(agent, stall_after_ms);
}
fn busy_state(&self, agent: RuntimeAgentKey) -> AgentBusyState {
self.tracker.busy_state(agent)
}
}
impl AgentInbox for MediatedInbox {
fn enqueue_message(
&self,
agent: RuntimeAgentKey,
item: InboxItem,
) -> Result<InboxReceipt, InboxError> {
if item.agent_id != agent.agent_id {
return Err(InboxError::AgentMismatch {
agent_id: agent.agent_id,
item_agent_id: item.agent_id,
});
}
let depth_before = self.mailbox.pending(&agent);
if depth_before >= self.inbox_capacity {
if item.is_lossless_system() {
return Ok(InboxReceipt {
item_id: item.id,
agent_id: agent.agent_id,
runtime_key: agent,
depth: depth_before,
status: InboxReceiptStatus::Deferred,
});
}
return Err(InboxError::InboxFull {
agent_id: agent.agent_id,
capacity: self.inbox_capacity,
});
}
let item_id = item.id;
let ticket = inbox_item_to_ticket(&item);
self.inbox_items().insert(item_id, item);
let _pending = self.mailbox.enqueue(agent, ticket);
let depth = self.mailbox.pending(&agent);
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentInboxQueued {
agent_id: agent.agent_id,
depth,
});
}
Ok(InboxReceipt {
item_id,
agent_id: agent.agent_id,
runtime_key: agent,
depth,
status: InboxReceiptStatus::Queued,
})
}
fn dequeue_next(&self, agent: RuntimeAgentKey) -> Option<InboxItem> {
let head = self.mailbox.head_ticket(&agent)?;
let item = self.inbox_items().remove(&head)?;
self.mailbox.cancel_head(agent, head);
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentInboxDrained {
agent_id: agent.agent_id,
depth: self.snapshot(agent).depth,
});
}
Some(item)
}
fn snapshot(&self, agent: RuntimeAgentKey) -> AgentInboxSnapshot {
let items_by_id = self.inbox_items();
let items: Vec<InboxItem> = self
.mailbox
.queue_for(agent)
.into_iter()
.filter_map(|ticket| items_by_id.get(&ticket.id).cloned())
.collect();
AgentInboxSnapshot {
agent_id: agent.agent_id,
runtime_key: agent,
depth: self.mailbox.pending(&agent),
items,
}
}
}
fn configured_inbox_capacity() -> usize {
std::env::var("IDEA_AGENT_INBOX_CAPACITY")
.ok()
.and_then(|raw| raw.parse::<usize>().ok())
.filter(|capacity| *capacity > 0)
.unwrap_or(DEFAULT_AGENT_INBOX_CAPACITY)
}
fn inbox_item_to_ticket(item: &InboxItem) -> Ticket {
let conversation = domain::conversation::ConversationId::from_uuid(uuid::Uuid::nil());
match item.source {
InboxSource::Agent { agent_id } => Ticket::from_agent(
item.id,
agent_id,
conversation,
agent_id.to_string(),
item.body.clone(),
),
InboxSource::Human => Ticket::from_human(item.id, conversation, "vous", item.body.clone()),
InboxSource::BackgroundTask { .. } | InboxSource::System => {
Ticket::from_human(item.id, conversation, "IdeA", item.body.clone())
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use domain::conversation::ConversationId;
use domain::mailbox::TicketId;
use domain::AgentId;
/// Deterministic clock for assertions on `since_ms`.
struct FixedClock(u64);
impl MillisClock for FixedClock {
fn now_ms(&self) -> u64 {
self.0
}
}
fn agent(n: u128) -> AgentId {
AgentId::from_uuid(uuid::Uuid::from_u128(n))
}
fn key(n: u128) -> RuntimeAgentKey {
RuntimeAgentKey::new(
domain::ProjectId::from_uuid(uuid::Uuid::from_u128(1000 + n)),
agent(n),
)
}
fn ticket(n: u128, task: &str) -> Ticket {
Ticket::from_human(
TicketId::from_uuid(uuid::Uuid::from_u128(n)),
ConversationId::from_uuid(uuid::Uuid::from_u128(1000 + n)),
"User",
task,
)
}
fn inbox_at(now_ms: u64) -> MediatedInbox {
MediatedInbox::new(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(now_ms)),
)
}
/// Records every [`DomainEvent`] published, for busy/idle assertions.
#[derive(Default)]
struct RecordingBus(Mutex<Vec<DomainEvent>>);
impl EventBus for RecordingBus {
fn publish(&self, event: DomainEvent) {
self.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(event);
}
fn subscribe(&self) -> domain::ports::EventStream {
unreachable!("RecordingBus is publish-only for these tests")
}
}
impl RecordingBus {
fn busy_events(&self) -> Vec<(AgentId, bool)> {
self.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.filter_map(|e| match e {
DomainEvent::AgentBusyChanged { agent_id, busy } => Some((*agent_id, *busy)),
_ => None,
})
.collect()
}
/// The `(ticket, text, submit_sequence, submit_delay_ms)` of every
/// `DelegationReady` published, in order.
#[allow(clippy::type_complexity)]
fn delegation_ready(&self) -> Vec<(TicketId, String, Option<String>, Option<u32>)> {
self.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.filter_map(|e| match e {
DomainEvent::DelegationReady {
ticket,
text,
submit_sequence,
submit_delay_ms,
..
} => Some((
*ticket,
text.clone(),
submit_sequence.clone(),
*submit_delay_ms,
)),
_ => None,
})
.collect()
}
}
#[test]
fn busy_event_fires_on_turn_start_and_idle_on_mark_idle() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// First enqueue starts a turn ⇒ exactly one Busy(true) event.
inbox.enqueue(a, ticket(10, "first"));
assert_eq!(bus.busy_events(), vec![(a.agent_id, true)]);
// Second enqueue while Busy queues behind ⇒ NO new busy event.
inbox.enqueue(a, ticket(11, "second"));
assert_eq!(
bus.busy_events(),
vec![(a.agent_id, true)],
"no re-announce while busy"
);
// mark_idle on a busy agent ⇒ exactly one Idle(false) event.
inbox.mark_idle(a);
assert_eq!(
bus.busy_events(),
vec![(a.agent_id, true), (a.agent_id, false)]
);
// mark_idle on an already-idle agent ⇒ no spurious event.
inbox.mark_idle(a);
assert_eq!(
bus.busy_events(),
vec![(a.agent_id, true), (a.agent_id, false)]
);
}
#[test]
fn preempt_emits_no_busy_event() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
inbox.enqueue(a, ticket(10, "t"));
inbox.preempt(a);
// Only the enqueue's Busy(true); preempt does not toggle busy state.
assert_eq!(bus.busy_events(), vec![(a.agent_id, true)]);
}
// ====================================================================
// §20 — enqueue no longer PTY-writes; it publishes DelegationReady
// ====================================================================
#[test]
fn enqueue_publishes_exactly_one_delegation_ready_on_idle_to_busy() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// Idle→Busy ⇒ exactly one DelegationReady carrying the task text + ticket.
inbox.enqueue(a, ticket(10, "do the thing"));
let ready = bus.delegation_ready();
assert_eq!(ready.len(), 1, "exactly one DelegationReady on turn start");
let tid = TicketId::from_uuid(uuid::Uuid::from_u128(10));
assert_eq!(ready[0].0, tid);
assert_eq!(ready[0].1, delegation_preamble("User", tid, "do the thing"));
assert!(
ready[0].1 == "do the thing",
"delivered text must be the raw task: {:?}",
ready[0].1
);
// A second enqueue while Busy must NOT re-publish (consistent with started_turn).
inbox.enqueue(a, ticket(11, "queued"));
assert_eq!(
bus.delegation_ready().len(),
1,
"no DelegationReady on a re-enqueue while Busy"
);
// After mark_idle, a fresh enqueue starts a new turn ⇒ a second DelegationReady.
inbox.mark_idle(a);
inbox.enqueue(a, ticket(12, "next turn"));
let ready = bus.delegation_ready();
assert_eq!(ready.len(), 2, "new turn ⇒ a new DelegationReady");
assert_eq!(ready[1].1, "next turn");
}
#[test]
fn enqueue_silent_marks_busy_without_delivery() {
let bus = Arc::new(RecordingBus::default());
let pty = Arc::new(FakePty::new());
let inbox = MediatedInbox::with_pty(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(1)),
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.set_front_attached(a, true);
inbox.enqueue_silent(a, ticket(10, "headless turn"));
assert!(inbox.busy_state(a).is_busy());
assert_eq!(bus.busy_events(), vec![(a.agent_id, true)]);
assert!(
bus.delegation_ready().is_empty(),
"headless bookkeeping must not leak a prompt to the terminal"
);
assert!(
pty.writes.lock().unwrap().is_empty(),
"headless bookkeeping must not write into the visible PTY"
);
assert_eq!(inbox.mailbox.pending(&a), 1);
}
#[test]
fn enqueue_message_queues_inbox_item_without_starting_turn() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
let task_id = domain::TaskId::from_uuid(uuid::Uuid::from_u128(42));
let item = domain::InboxItem {
id: TicketId::from_uuid(uuid::Uuid::from_u128(10)),
agent_id: a.agent_id,
source: InboxSource::BackgroundTask { task_id },
kind: domain::InboxItemKind::BackgroundCompletion,
body: "Background task completed.".to_owned(),
created_at_ms: 1,
correlation_id: Some(task_id.to_string()),
};
let receipt = inbox.enqueue_message(a, item.clone()).unwrap();
assert_eq!(receipt.status, InboxReceiptStatus::Queued);
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
assert_eq!(inbox.mailbox.pending(&a), 1);
assert!(
bus.busy_events().is_empty(),
"inbox enqueue must not start a mediated turn"
);
assert_eq!(inbox.dequeue_next(a), Some(item));
assert_eq!(inbox.mailbox.pending(&a), 0);
}
#[test]
fn delegation_ready_carries_bound_submit_config() {
let bus = Arc::new(RecordingBus::default());
let pty = Arc::new(FakePty::new());
let inbox = MediatedInbox::with_pty(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(1)),
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
let h = handle(1);
// Bind the target's submit config (resolved from its profile by the service).
inbox.bind_handle_with_submit(a, h, SubmitConfig::new(Some("\r".to_owned()), Some(42)));
// A frontend cell is mounted ⇒ delivery flows through the event (the front
// writes). This test asserts the event carries the bound submit config.
inbox.set_front_attached(a, true);
inbox.enqueue(a, ticket(10, "task"));
let ready = bus.delegation_ready();
assert_eq!(ready.len(), 1);
assert_eq!(ready[0].2.as_deref(), Some("\r"), "submit_sequence echoed");
assert_eq!(ready[0].3, Some(42), "submit_delay_ms echoed");
}
#[test]
fn enqueue_without_bound_submit_yields_none_fields() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
inbox.enqueue(a, ticket(10, "task"));
let ready = bus.delegation_ready();
assert_eq!(ready.len(), 1);
assert_eq!(
ready[0].2, None,
"no bound submit ⇒ None (front applies default)"
);
assert_eq!(ready[0].3, None);
}
#[test]
fn front_attached_agent_is_delivered_via_event_not_backend_write() {
// §20 nominal: a mounted frontend cell owns the physical write. The backend
// publishes DelegationReady and writes NOTHING into the PTY itself.
let pty = Arc::new(FakePty::new());
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::with_pty(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(1)),
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.set_front_attached(a, true);
inbox.enqueue(a, ticket(10, "do X"));
assert_eq!(
bus.delegation_ready().len(),
1,
"front-attached ⇒ exactly one DelegationReady event for the write-portal"
);
// The headless path is not taken ⇒ no thread is spawned ⇒ no PTY write, ever.
assert!(
pty.writes.lock().unwrap().is_empty(),
"front-attached ⇒ the backend must NOT write the PTY (the front does)"
);
}
#[test]
fn headless_agent_without_front_cell_is_written_by_the_backend() {
// Cold/background delegation target: no frontend cell is mounted, so NOBODY
// consumes DelegationReady. The mediator must therefore write the turn into the
// PTY itself (text + submit), else the agent never receives its task — the
// root cause of "a delegated agent never replies".
let pty = Arc::new(FakePty::new());
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::with_pty(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(1)),
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// No set_front_attached ⇒ headless.
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.enqueue(a, ticket(10, "do X"));
assert!(
wait_until(|| pty.writes.lock().unwrap().len() >= 2),
"headless delivery writes the task text + submit sequence into the PTY"
);
let writes = pty.writes.lock().unwrap().clone();
assert!(
String::from_utf8_lossy(&writes[0]).contains("do X"),
"first write carries the task text"
);
assert_eq!(
writes[1], b"\r",
"second write is the default submit sequence"
);
assert!(
bus.delegation_ready().is_empty(),
"headless delivery takes over ⇒ no DelegationReady event is published"
);
}
#[test]
fn headless_agent_chunks_long_delegation_before_submit() {
let pty = Arc::new(FakePty::new());
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::with_pty(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(1)),
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
let task = format!("début {} fin", "x".repeat(1200));
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.enqueue(a, ticket(10, &task));
// Attendre le SUBMIT final (`\r`, écrit après le délai), pas un simple seuil de
// writes : le nombre de chunks de contenu dépend de la taille du payload (préambule
// + tâche), donc seul le `\r` terminal est un point d'arrêt stable.
assert!(
wait_until(|| {
let w = pty.writes.lock().unwrap();
w.len() >= 4 && w.last().map(Vec::as_slice) == Some(b"\r".as_slice())
}),
"long headless delivery writes multiple chunks followed by submit"
);
let writes = pty.writes.lock().unwrap().clone();
assert_eq!(writes.last().map(Vec::as_slice), Some(b"\r".as_slice()));
let delivered = writes[..writes.len() - 1]
.iter()
.flat_map(|chunk| chunk.iter().copied())
.collect::<Vec<_>>();
let delivered = String::from_utf8(delivered).unwrap();
assert!(
delivered.contains(&task),
"chunked writes reconstruct the full task"
);
assert!(
bus.delegation_ready().is_empty(),
"headless delivery does not also publish DelegationReady"
);
}
#[tokio::test]
async fn enqueue_returns_pending_reply_resolved_via_mailbox() {
let inbox = inbox_at(5);
let a = key(1);
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(),
domain::mailbox::TurnResolution::Replied("done".to_owned())
);
}
#[test]
fn first_enqueue_marks_busy_with_ticket_and_stamp() {
let inbox = inbox_at(1234);
let a = key(1);
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
inbox.enqueue(a, ticket(10, "t"));
assert_eq!(
inbox.busy_state(a),
AgentBusyState::Busy {
ticket: TicketId::from_uuid(uuid::Uuid::from_u128(10)),
since_ms: 1234,
}
);
}
#[test]
fn second_enqueue_while_busy_keeps_first_ticket_and_does_not_reject() {
let inbox = inbox_at(1);
let a = key(1);
inbox.enqueue(a, ticket(10, "first"));
inbox.enqueue(a, ticket(11, "second")); // accepted, queues behind
// Still busy on the FIRST ticket (turn unchanged), both queued in the mailbox.
assert_eq!(
inbox.busy_state(a).ticket(),
Some(TicketId::from_uuid(uuid::Uuid::from_u128(10)))
);
assert_eq!(inbox.mailbox().pending(&a), 2, "forward, never reject");
}
#[test]
fn mark_idle_returns_to_idle_so_next_turn_can_start() {
let inbox = inbox_at(1);
let a = key(1);
inbox.enqueue(a, ticket(10, "t"));
assert!(inbox.busy_state(a).is_busy());
inbox.mark_idle(a);
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
// A subsequent enqueue starts a fresh turn.
inbox.enqueue(a, ticket(11, "t2"));
assert_eq!(
inbox.busy_state(a).ticket(),
Some(TicketId::from_uuid(uuid::Uuid::from_u128(11)))
);
}
#[tokio::test]
async fn preempt_is_distinct_from_enqueue_and_resolves_no_ticket() {
let inbox = inbox_at(1);
let a = key(1);
let pending = inbox.enqueue(a, ticket(10, "t"));
inbox.preempt(a);
// preempt did not pop/resolve the ticket: still pending in the mailbox.
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(),
domain::mailbox::TurnResolution::Replied("real".to_owned())
);
}
#[test]
fn two_enqueues_same_agent_serialise_in_one_fifo() {
let inbox = inbox_at(1);
let a = key(1);
inbox.enqueue(a, ticket(10, "first"));
inbox.enqueue(a, ticket(11, "second"));
assert_eq!(inbox.mailbox().pending(&a), 2);
assert_eq!(
inbox.mailbox().head_ticket(&a),
Some(TicketId::from_uuid(uuid::Uuid::from_u128(10))),
"FIFO order preserved"
);
}
#[test]
fn different_agents_are_independent_not_blocking() {
let inbox = inbox_at(1);
let a = key(1);
let b = key(2);
inbox.enqueue(a, ticket(10, "a"));
inbox.enqueue(b, ticket(20, "b"));
assert!(inbox.busy_state(a).is_busy());
assert!(inbox.busy_state(b).is_busy());
// Marking A idle leaves B untouched.
inbox.mark_idle(a);
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
assert!(inbox.busy_state(b).is_busy());
assert_eq!(inbox.mailbox().pending(&a), 1);
assert_eq!(inbox.mailbox().pending(&b), 1);
}
// ====================================================================
// Headless delivery — the backend writes the PTY when no front cell is mounted
// ====================================================================
use domain::ids::SessionId;
use domain::ports::{ExitStatus, OutputStream, PtyError, PtyHandle as Handle, SpawnSpec};
use domain::terminal::PtySize;
/// A fake [`PtyPort`] that records `write`s (for the headless-delivery tests).
/// `subscribe_output` yields an empty stream; `spawn`/`resize`/`kill`/`scrollback`
/// are unused stubs.
struct FakePty {
writes: Mutex<Vec<Vec<u8>>>,
}
impl FakePty {
fn new() -> Self {
Self {
writes: Mutex::new(Vec::new()),
}
}
}
#[async_trait::async_trait]
impl PtyPort for FakePty {
async fn spawn(&self, _spec: SpawnSpec, _size: PtySize) -> Result<Handle, PtyError> {
Ok(Handle {
session_id: SessionId::new_random(),
})
}
fn write(&self, _handle: &Handle, data: &[u8]) -> Result<(), PtyError> {
self.writes.lock().unwrap().push(data.to_vec());
Ok(())
}
fn resize(&self, _handle: &Handle, _size: PtySize) -> Result<(), PtyError> {
Ok(())
}
fn subscribe_output(&self, _handle: &Handle) -> Result<OutputStream, PtyError> {
Ok(Box::new(std::iter::empty()))
}
fn scrollback(&self, _handle: &Handle) -> Result<Vec<u8>, PtyError> {
Ok(Vec::new())
}
async fn wait(&self, _handle: &Handle) -> Result<ExitStatus, PtyError> {
Ok(ExitStatus { code: Some(0) })
}
fn try_wait(&self, _handle: &Handle) -> Result<Option<ExitStatus>, PtyError> {
Ok(Some(ExitStatus { code: Some(0) }))
}
async fn kill(&self, _handle: &Handle) -> Result<ExitStatus, PtyError> {
Ok(ExitStatus { code: Some(0) })
}
}
fn handle(n: u128) -> Handle {
Handle {
session_id: SessionId::from_uuid(uuid::Uuid::from_u128(n)),
}
}
/// Spins until `cond` holds or the deadline passes (headless writes happen on a
/// detached thread, so the effect is observed asynchronously).
fn wait_until(mut cond: impl FnMut() -> bool) -> bool {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
while std::time::Instant::now() < deadline {
if cond() {
return true;
}
std::thread::sleep(std::time::Duration::from_millis(5));
}
cond()
}
// --- backstop no-reply : `turn_ended` (fin de tour transcript) ⇒ réveil après G ---
//
// Le déclencheur n'est plus un watcher PTY mais le `TurnWatcher` transcript ; on
// appelle donc `turn_ended` directement (le watcher est testé à part dans
// `inspector::claude_turn_watcher`). La logique grâce/complétion est inchangée.
/// Builds a grace-shortened inbox (no PTY needed: `turn_ended` is invoked directly).
fn inbox_grace(grace: Duration) -> MediatedInbox {
MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_grace(grace)
}
/// La cible termine son tour 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 turn_ended_without_reply_wakes_caller_after_grace() {
let a = key(1);
let inbox = inbox_grace(Duration::from_millis(40));
let pending = inbox.enqueue(a, ticket(10, "task"));
inbox.turn_ended(a);
// 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 la fin de tour 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_turn_ended_wins() {
let a = key(1);
let inbox = inbox_grace(Duration::from_millis(40));
let pending = inbox.enqueue(a, ticket(10, "task"));
// idea_reply resolves the head BEFORE the turn-end signal.
inbox.mailbox().resolve(a, "answer".to_owned()).unwrap();
inbox.turn_ended(a);
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 la fin de tour 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_turn_ended_wins() {
let a = key(1);
// Long grace ⇒ the reply lands well within it.
let inbox = inbox_grace(Duration::from_secs(30));
let pending = inbox.enqueue(a, ticket(10, "task"));
inbox.turn_ended(a);
// turn_ended a marqué Idle et armé la grâce ; on répond pendant la grâce.
assert!(
!inbox.busy_state(a).is_busy(),
"turn_ended 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 a = key(1);
let inbox = inbox_grace(Duration::from_millis(30));
let pending = inbox.enqueue(a, ticket(10, "task"));
inbox.turn_ended(a);
// 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 a = key(1);
let inbox = inbox_grace(Duration::from_millis(20));
// Human submit: the reply handle is dropped immediately (not awaited).
drop(inbox.enqueue(a, ticket(10, "human submit")));
inbox.turn_ended(a);
// 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)"
);
}
/// `turn_ended` sur un agent déjà `Idle` (aucun ticket actif) ⇒ simple `mark_idle`
/// idempotent, aucune grâce armée, aucun panic.
#[test]
fn turn_ended_on_idle_agent_is_noop() {
let a = key(1);
let inbox = inbox_grace(Duration::from_millis(20));
inbox.turn_ended(a); // jamais de tour démarré.
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
}
// ====================================================================
// Fix race cold-launch — `mark_starting` gate le PREMIER tour sur la readiness MCP
// ====================================================================
/// Readiness de démarrage MCP : un 1er tour différé (`mark_starting` + enqueue) reste
/// retenu tant que rien ne le libère ; `release_cold_start` (signal connexion MCP) le
/// draine ⇒ exactement UNE `DelegationReady`. C'est le seul drain du tour différé.
#[test]
fn release_cold_start_delivers_deferred_first_turn() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// Démarrage à froid : gate armé AVANT l'enqueue (ordre de l'orchestrateur).
inbox.mark_starting(a);
inbox.enqueue(a, ticket(10, "cold task"));
assert!(inbox.busy_state(a).is_busy(), "le tour démarre (Busy)");
assert!(
bus.delegation_ready().is_empty(),
"cold-launch ⇒ pas de DelegationReady tant que rien ne libère"
);
// Connexion du pont MCP ⇒ libère le 1er tour différé.
inbox.release_cold_start(a);
let ready = bus.delegation_ready();
assert_eq!(
ready.len(),
1,
"release_cold_start ⇒ exactement une DelegationReady"
);
assert_eq!(ready[0].1, "cold task");
}
/// **Régression de la race cold-start (ordre INVERSE)** : `release_cold_start`
/// (signal de connexion du pont MCP) arrive AVANT que l'`enqueue` n'ait parqué son
/// tour différé. Avant le fix, le signal trouvait `deferred` vide (no-op), puis
/// l'`enqueue` parquait le tour dans `deferred` — où il restait à JAMAIS (aucun second
/// signal ne venait le drainer ⇒ `idea_ask_agent` bloquait jusqu'au timeout, le watcher
/// prompt-ready PTY ayant été supprimé). Après le fix : exactement UNE
/// `DelegationReady` est livrée (le latch « déjà libéré » garantit la livraison).
#[test]
fn release_cold_start_before_deferred_enqueue_still_delivers_once() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// Démarrage à froid : gate armé AVANT l'enqueue (ordre de l'orchestrateur).
inbox.mark_starting(a);
// **La race** : la readiness MCP arrive AVANT l'enqueue qui parquera le tour.
inbox.release_cold_start(a);
assert!(
bus.delegation_ready().is_empty(),
"rien à livrer tant que l'enqueue n'a pas fourni le tour"
);
// Puis l'enqueue fournit le tour : grâce au latch « déjà libéré », il livre
// immédiatement au lieu de le perdre dans `deferred`.
inbox.enqueue(a, ticket(10, "cold task"));
let ready = bus.delegation_ready();
assert_eq!(
ready.len(),
1,
"release AVANT enqueue ⇒ exactement une DelegationReady (jamais zéro, jamais deux)"
);
assert_eq!(ready[0].1, "cold task");
assert!(
inbox.busy_state(a).is_busy(),
"le tour froid livré ⇒ l'agent reste Busy (idle viendra d'idea_reply)"
);
}
/// `release_cold_start` sans tour différé ⇒ no-op : aucune `DelegationReady` et,
/// surtout, PAS de transition Idle (signal de DÉMARRAGE, pas de fin de tour).
#[test]
fn release_cold_start_is_noop_without_deferred() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// Démarre un tour normal (Busy), SANS gate cold-launch (pas de mark_starting).
inbox.enqueue(a, ticket(10, "task"));
let busy_before = bus.busy_events();
inbox.release_cold_start(a);
assert!(
bus.delegation_ready().len() == 1,
"pas de DelegationReady supplémentaire (la seule est celle de l'enqueue)"
);
assert_eq!(
bus.busy_events(),
busy_before,
"release_cold_start ne marque PAS Idle (aucun AgentBusyChanged{{busy:false}})"
);
}
/// `release_cold_start` est idempotent : un second appel ne re-draine rien ⇒ une seule
/// `DelegationReady` au total.
#[test]
fn release_cold_start_is_idempotent() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
inbox.mark_starting(a);
inbox.enqueue(a, ticket(10, "cold task"));
inbox.release_cold_start(a);
inbox.release_cold_start(a);
assert_eq!(
bus.delegation_ready().len(),
1,
"deux release_cold_start ⇒ une seule DelegationReady"
);
}
/// (b) Agent chaud (non `mark_starting`) : la `DelegationReady` est publiée
/// IMMÉDIATEMENT à l'enqueue — non-régression du chemin existant.
#[test]
fn warm_agent_delivers_first_turn_immediately() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// Pas de mark_starting ⇒ agent chaud.
inbox.enqueue(a, ticket(10, "warm task"));
let ready = bus.delegation_ready();
assert_eq!(
ready.len(),
1,
"agent chaud ⇒ DelegationReady immédiate (zéro régression)"
);
assert_eq!(ready[0].1, "warm task");
}
/// (c) Cas limite : si on ne marque PAS `starting` (l'orchestrateur n'arme pas le
/// gate quand la cible n'a pas de pont MCP), l'enqueue livre la `DelegationReady`
/// immédiatement — donc PAS de blocage indéfini du premier tour.
#[test]
fn no_mcp_no_starting_gate_delivers_immediately() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
// Cold launch SANS pont MCP ⇒ l'orchestrateur N'APPELLE PAS mark_starting.
inbox.enqueue(a, ticket(10, "task"));
// Premier tour livré tout de suite : aucun signal de readiness ne viendrait le
// débloquer, donc surtout pas de gate.
assert_eq!(
bus.delegation_ready().len(),
1,
"sans gate ⇒ DelegationReady immédiate (pas de blocage indéfini)"
);
// Reste Busy (idle viendra d'idea_reply / turn_ended / timeout), tour bien livré.
assert!(inbox.busy_state(a).is_busy());
}
/// Après livraison du 1er tour froid (drainé par `release_cold_start`), un `mark_idle`
/// (idea_reply) puis un nouvel enqueue (agent désormais chaud) repart en livraison
/// immédiate.
#[test]
fn after_cold_first_turn_subsequent_enqueue_is_immediate() {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1)))
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = key(1);
inbox.mark_starting(a);
inbox.enqueue(a, ticket(10, "first"));
// Readiness MCP ⇒ draine le 1er tour différé.
inbox.release_cold_start(a);
assert_eq!(bus.delegation_ready().len(), 1, "1er tour froid livré");
// Fin du 1er tour (idea_reply) puis nouvel enqueue ⇒ livraison immédiate.
inbox.mark_idle(a);
inbox.enqueue(a, ticket(11, "second"));
let ready = bus.delegation_ready();
assert_eq!(
ready.len(),
2,
"second tour livré immédiatement (plus de gate)"
);
assert_eq!(ready[1].1, "second");
}
// ====================================================================
// Lot 2 — détection de stagnation (last_seen / sweep_stalled / liveness)
// ====================================================================
use domain::input::AgentLiveness;
use std::sync::atomic::{AtomicU64, Ordering};
/// Horloge millis **mutable** (atomique) pour piloter le temps dans les tests de
/// stagnation sans horloge réelle.
struct MutClock(AtomicU64);
impl MutClock {
fn new(start: u64) -> Arc<Self> {
Arc::new(Self(AtomicU64::new(start)))
}
fn set(&self, now: u64) {
self.0.store(now, Ordering::SeqCst);
}
}
impl MillisClock for MutClock {
fn now_ms(&self) -> u64 {
self.0.load(Ordering::SeqCst)
}
}
/// Construit un inbox sur l'horloge mutable + un bus enregistreur, renvoie les deux.
fn inbox_with_clock(clock: Arc<MutClock>) -> (MediatedInbox, Arc<RecordingBus>) {
let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), clock)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
(inbox, bus)
}
impl RecordingBus {
/// Toutes les transitions de vivacité publiées, dans l'ordre.
fn liveness_events(&self) -> Vec<(AgentId, AgentLiveness)> {
self.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.filter_map(|e| match e {
DomainEvent::AgentLivenessChanged { agent_id, liveness } => {
Some((*agent_id, *liveness))
}
_ => None,
})
.collect()
}
}
// (a) absence de battement > stall_after_ms ⇒ Stalled
#[test]
fn no_heartbeat_past_threshold_marks_stalled() {
let clock = MutClock::new(1_000);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = key(1);
// Profil : seuil de stagnation à 30_000 ms. Armé avant le tour.
inbox.set_stall_threshold(a, Some(30_000));
inbox.enqueue(a, ticket(10, "task")); // démarre le tour à t=1_000 ⇒ last_seen=1_000
// Dans la fenêtre : pas de transition.
clock.set(1_000 + 30_000);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![],
"à la limite exacte, pas encore stalled"
);
// Au-delà du seuil : exactement une transition Stalled.
clock.set(1_000 + 30_001);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)]
);
// Idempotent : un second sweep ne ré-émet pas.
clock.set(1_000 + 60_000);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)],
"déjà stalled ⇒ pas de spam"
);
}
// (b) un battement dans la fenêtre réinitialise last_seen (pas de stall)
#[test]
fn heartbeat_within_window_resets_last_seen() {
let clock = MutClock::new(1_000);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = key(1);
inbox.set_stall_threshold(a, Some(30_000));
inbox.enqueue(a, ticket(10, "task")); // last_seen=1_000
// Battement à t=20_000 ⇒ last_seen=20_000.
clock.set(20_000);
inbox.mark_alive(a);
// À t=45_000 : 45_000 - 20_000 = 25_000 < 30_000 ⇒ toujours Alive.
clock.set(45_000);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![],
"un battement a réinitialisé la fenêtre ⇒ pas de stall"
);
}
// (d) AgentLivenessChanged émis une seule fois par transition (Alive→Stalled→Alive)
#[test]
fn liveness_event_once_per_transition() {
let clock = MutClock::new(0);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = key(1);
inbox.set_stall_threshold(a, Some(10_000));
inbox.enqueue(a, ticket(10, "t")); // last_seen=0
// Stall.
clock.set(10_001);
inbox.sweep_stalled();
// Sweeps répétés : pas de ré-émission.
inbox.sweep_stalled();
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)]
);
// Battement tardif ⇒ une seule reprise Alive.
clock.set(20_000);
inbox.mark_alive(a);
inbox.mark_alive(a); // second battement : déjà Alive ⇒ rien.
assert_eq!(
bus.liveness_events(),
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive)
]
);
// Re-stall possible après reprise (nouvelle transition).
clock.set(20_000 + 10_001);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive),
(a.agent_id, AgentLiveness::Stalled),
]
);
}
// mark_idle (fin de tour) sur un agent Stalled émet la reprise et retire l'entrée.
#[test]
fn mark_idle_on_stalled_emits_recovery_and_clears() {
let clock = MutClock::new(0);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = key(1);
inbox.set_stall_threshold(a, Some(5_000));
inbox.enqueue(a, ticket(10, "t"));
clock.set(5_001);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)]
);
inbox.mark_idle(a); // fin de tour ⇒ reprise Alive + entrée retirée.
assert_eq!(
bus.liveness_events(),
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive)
]
);
// Plus d'entrée : un sweep ultérieur ne ré-émet rien (même très tard).
clock.set(1_000_000);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive)
]
);
}
// (e) agent sans LivenessStrategy (stall_after_ms = None) ⇒ comportement legacy :
// jamais Stalled, aucun AgentLivenessChanged.
#[test]
fn agent_without_threshold_is_never_stalled() {
let clock = MutClock::new(0);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = key(1);
// Pas de set_stall_threshold (ou None) : armé sans seuil au start_turn.
inbox.enqueue(a, ticket(10, "t"));
clock.set(10_000_000); // très loin dans le futur.
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![],
"sans seuil ⇒ jamais stalled (zéro régression)"
);
// Et un mark_alive sur un agent jamais stalled n'émet rien non plus.
inbox.mark_alive(a);
assert_eq!(bus.liveness_events(), vec![]);
}
// Un battement n'a aucun effet sur un agent dont aucune entrée n'est armée.
#[test]
fn mark_alive_on_unarmed_agent_is_noop() {
let clock = MutClock::new(0);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = key(1);
inbox.mark_alive(a); // jamais de tour démarré.
inbox.sweep_stalled();
assert_eq!(bus.liveness_events(), vec![]);
}
}
Reference in New Issue View Git Blame Copy Permalink