IdeA/crates/infrastructure/src/input/mod.rs

//! [`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::AgentId;
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<AgentId, 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<AgentId, 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<AgentId>>,
    /// **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<AgentId, 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<AgentId>>,
    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(AgentId, 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<AgentId>> {
        self.starting
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn lock_deferred(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, DeferredDelegation>> {
        self.deferred
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn lock_released(&self) -> std::sync::MutexGuard<'_, HashSet<AgentId>> {
        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: AgentId) {
        application::diag!("[input-mediator] mark_starting cold-start gate armed agent={agent}");
        self.lock_starting().insert(agent);
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, AgentBusyState>> {
        self.busy
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn lock_liveness(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, 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: AgentId, liveness: AgentLiveness) {
        if let Some(events) = &self.events {
            events.publish(DomainEvent::AgentLivenessChanged {
                agent_id: agent,
                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: AgentId, 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: AgentId, 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<AgentId> = {
            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: AgentId) {
        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: AgentId) -> 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: AgentId, 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: AgentId, 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,
                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: AgentId) {
        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: AgentId) {
        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: AgentId) {
        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,
                    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<AgentId, 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<AgentId>>>,
    /// 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<AgentId, 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<AgentId, 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<AgentId, PtyHandle>>>,
        front_owned: &Arc<Mutex<HashSet<AgentId>>>,
        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<AgentId, PtyHandle>>>,
        front_owned: Arc<Mutex<HashSet<AgentId>>>,
    ) -> HeadlessSink {
        Arc::new(move |agent: AgentId, 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<AgentId, PtyHandle>> {
        self.handles
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn submit(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, SubmitConfig>> {
        self.submit
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn stall(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, 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: AgentId, 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,
                    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: AgentId, 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,
                    busy: true,
                });
            }
        }
        self.mailbox.enqueue(agent, ticket)
    }

    fn bind_handle(&self, agent: AgentId, 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: AgentId, 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: AgentId) -> bool {
        self.pty.is_some() && self.handles().get(&agent).is_some()
    }

    fn mark_starting(&self, agent: AgentId) {
        // 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: AgentId) {
        self.tracker.release_cold_start(agent);
    }

    fn set_front_attached(&self, agent: AgentId, 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: AgentId) {
        // 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: AgentId) {
        // 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: AgentId) {
        // 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: AgentId) {
        // 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: AgentId, 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: AgentId) -> AgentBusyState {
        self.tracker.busy_state(agent)
    }
}

impl AgentInbox for MediatedInbox {
    fn enqueue_message(&self, agent: AgentId, item: InboxItem) -> Result<InboxReceipt, InboxError> {
        if item.agent_id != agent {
            return Err(InboxError::AgentMismatch {
                agent_id: agent,
                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,
                    depth: depth_before,
                    status: InboxReceiptStatus::Deferred,
                });
            }
            return Err(InboxError::InboxFull {
                agent_id: agent,
                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,
                depth,
            });
        }
        Ok(InboxReceipt {
            item_id,
            agent_id: agent,
            depth,
            status: InboxReceiptStatus::Queued,
        })
    }

    fn dequeue_next(&self, agent: AgentId) -> 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,
                depth: self.snapshot(agent).depth,
            });
        }
        Some(item)
    }

    fn snapshot(&self, agent: AgentId) -> 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,
            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;

    /// 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 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 = agent(1);

        // First enqueue starts a turn ⇒ exactly one Busy(true) event.
        inbox.enqueue(a, ticket(10, "first"));
        assert_eq!(bus.busy_events(), vec![(a, true)]);

        // Second enqueue while Busy queues behind ⇒ NO new busy event.
        inbox.enqueue(a, ticket(11, "second"));
        assert_eq!(
            bus.busy_events(),
            vec![(a, 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, true), (a, false)]);

        // mark_idle on an already-idle agent ⇒ no spurious event.
        inbox.mark_idle(a);
        assert_eq!(bus.busy_events(), vec![(a, true), (a, 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 = agent(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, 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 = agent(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 = agent(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, 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 = agent(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,
            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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(1);
        let b = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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 = agent(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, 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, 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 = agent(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 = agent(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, 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, AgentLiveness::Stalled), (a, 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, AgentLiveness::Stalled),
                (a, AgentLiveness::Alive),
                (a, 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 = agent(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, AgentLiveness::Stalled)]);

        inbox.mark_idle(a); // fin de tour ⇒ reprise Alive + entrée retirée.
        assert_eq!(
            bus.liveness_events(),
            vec![(a, AgentLiveness::Stalled), (a, 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, AgentLiveness::Stalled), (a, 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 = agent(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 = agent(1);
        inbox.mark_alive(a); // jamais de tour démarré.
        inbox.sweep_stalled();
        assert_eq!(bus.liveness_events(), vec![]);
    }
}