//! [`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 domain::events::DomainEvent;
use domain::ids::AgentId;
use domain::input::{AgentBusyState, InputMediator};
use domain::mailbox::{AgentMailbox, PendingReply, Ticket};
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>>,
    events: Option<Arc<dyn EventBus>>,
}

impl BusyTracker {
    fn new(events: Option<Arc<dyn EventBus>>) -> Self {
        Self {
            busy: Mutex::new(HashMap::new()),
            events,
        }
    }

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

    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() {
            false
        } else {
            *entry = state;
            true
        }
    }

    /// 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 {
            if let Some(events) = &self.events {
                events.publish(DomainEvent::AgentBusyChanged {
                    agent_id: agent,
                    busy: false,
                });
            }
        }
    }
}

/// 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.
///
/// When a [`PtyPort`] is wired (cadrage C3 §5.2), `enqueue` **delivers** the turn —
/// it writes the prefixed task line into the agent's bound [`PtyHandle`]. This is the
/// single, serialized write path that replaces the orchestrator's former ad-hoc PTY
/// write (no more `[IdeA · tâche …]` line emitted from `ask_agent`, no `\r` band-aid).
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`.
    handles: Mutex<HashMap<AgentId, PtyHandle>>,
    /// Agents whose prompt-ready watcher thread is already armed, so re-binding the same
    /// handle does not spawn a duplicate watcher (lot C5). Shared (`Arc`) because each
    /// watcher thread un-arms its own entry on exit.
    watched: Arc<Mutex<HashSet<AgentId>>>,
}

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 {
        Self {
            mailbox,
            tracker: Arc::new(BusyTracker::new(None)),
            clock,
            pty: None,
            handles: Mutex::new(HashMap::new()),
            watched: Arc::new(Mutex::new(HashSet::new())),
        }
    }

    /// 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 {
        self.tracker = Arc::new(BusyTracker::new(Some(events)));
        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 {
        Self {
            mailbox,
            tracker: Arc::new(BusyTracker::new(None)),
            clock,
            pty: Some(pty),
            handles: Mutex::new(HashMap::new()),
            watched: Arc::new(Mutex::new(HashSet::new())),
        }
    }

    /// 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))
    }

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

    /// The underlying mailbox (e.g. for `cancel_head` / `resolve` from the orchestrator).
    #[must_use]
    pub fn mailbox(&self) -> Arc<InMemoryMailbox> {
        Arc::clone(&self.mailbox)
    }

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

    /// Arms the **prompt-ready watcher** for `agent` (lot C5).
    ///
    /// Requires a wired [`PtyPort`] **and** a non-empty literal `pattern`. Spawns a
    /// detached thread that consumes the handle's output stream and calls
    /// [`BusyTracker::mark_idle`] the **first** time `pattern` appears as a substring of
    /// the cumulative output, then exits (one-shot per arming). A sliding buffer keeps
    /// the last `pattern.len()-1` bytes so a marker split across two chunks still
    /// matches. No watcher is armed when the port is absent or the pattern is empty
    /// (no detection ⇒ Idle only via explicit signal/timeout — the safe fallback).
    ///
    /// Re-arming the same agent is a no-op while a watcher is already live (tracked in
    /// `watched`), so re-binding a handle never spawns duplicate watchers.
    fn arm_prompt_watcher(&self, agent: AgentId, handle: &PtyHandle, pattern: String) {
        if pattern.is_empty() {
            return;
        }
        let Some(pty) = self.pty.clone() else {
            return;
        };
        // Already watching this agent ⇒ keep the live watcher (avoid duplicates).
        if !self.watched().insert(agent) {
            return;
        }
        let stream = match pty.subscribe_output(handle) {
            Ok(s) => s,
            Err(_) => {
                // Could not subscribe (unknown handle): un-arm so a later bind retries.
                self.watched().remove(&agent);
                return;
            }
        };
        let tracker = Arc::clone(&self.tracker);
        let watched = Arc::clone(&self.watched);
        let needle = pattern.into_bytes();
        std::thread::spawn(move || {
            // Cumulative tail kept small: just enough to catch a marker split across two
            // chunks (keep the last needle.len()-1 bytes between reads).
            let keep = needle.len().saturating_sub(1);
            let mut window: Vec<u8> = Vec::with_capacity(keep + 1);
            for chunk in stream {
                window.extend_from_slice(&chunk);
                if window
                    .windows(needle.len())
                    .any(|w| w == needle.as_slice())
                {
                    // Prompt-ready: first OR signal wins ⇒ Idle (advances the FIFO).
                    tracker.mark_idle(agent);
                    break;
                }
                if window.len() > keep {
                    let drop_to = window.len() - keep;
                    window.drain(..drop_to);
                }
            }
            // Watcher done (matched or stream closed at EOF): un-arm so a future bind
            // (e.g. after a relaunch) can re-arm a fresh watcher.
            watched
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .remove(&agent);
        });
    }
}

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 started_turn = self.tracker.start_turn(
            agent,
            AgentBusyState::Busy {
                ticket: ticket_id,
                since_ms: self.clock.now_ms(),
            },
        );
        // Publish Busy only on the enqueue that **starts** a turn (Idle→Busy); a
        // second enqueue while Busy queues behind without re-announcing (cadrage
        // C4 §4.2). Published outside the busy mutex (the tracker released it above).
        if started_turn {
            if let Some(events) = &self.tracker.events {
                events.publish(DomainEvent::AgentBusyChanged {
                    agent_id: agent,
                    busy: true,
                });
            }
        }
        // Delivery: write the prefixed task line into the agent's bound handle. The
        // prefix carries the requester + ticket id so the target replies via
        // `idea_reply(result, ticket)`. A best-effort write: a missing handle/port or
        // a write error never drops the ticket (the orchestrator's await + timeout
        // remain the safety net) — the reply slot is registered regardless.
        if let Some(pty) = &self.pty {
            if let Some(handle) = self.handles().get(&agent).cloned() {
                let line = format!(
                    "[IdeA · tâche de {} · ticket {}] {}\n",
                    ticket.requester, ticket_id, ticket.task
                );
                let _ = pty.write(&handle, line.as_bytes());
            }
        }
        self.mailbox.enqueue(agent, ticket)
    }

    fn bind_handle(&self, agent: AgentId, handle: PtyHandle) {
        self.handles().insert(agent, handle);
    }

    fn bind_handle_with_prompt(
        &self,
        agent: AgentId,
        handle: PtyHandle,
        prompt_ready_pattern: Option<String>,
    ) {
        // Register the input handle (delivery path) exactly like `bind_handle`, then arm
        // the prompt-ready watcher when the profile declares a literal marker (C5). A
        // `None`/empty pattern arms nothing: Idle then comes only from the explicit
        // signal or the per-turn timeout (safe fallback, never a false Idle).
        self.handles().insert(agent, handle.clone());
        if let Some(pattern) = prompt_ready_pattern {
            self.arm_prompt_watcher(agent, &handle, pattern);
        }
    }

    fn delivers_turn(&self, agent: AgentId) -> bool {
        self.pty.is_some() && self.handles().get(&agent).is_some()
    }

    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.
        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 (also used by the prompt-ready watcher): real Busy→Idle only,
        // publishing AgentBusyChanged{busy:false} once.
        self.tracker.mark_idle(agent);
    }

    fn busy_state(&self, agent: AgentId) -> AgentBusyState {
        self.tracker.busy_state(agent)
    }
}

#[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()
        }
    }

    #[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)]);
    }

    #[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(), "done");
    }

    #[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(), "real");
    }

    #[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);
    }

    // ====================================================================
    // Lot C5 — prompt-ready detection on the PTY output stream
    // ====================================================================

    use domain::ports::{ExitStatus, OutputStream, PtyError, PtyHandle as Handle, SpawnSpec};
    use domain::terminal::PtySize;
    use domain::ids::SessionId;

    /// A fake [`PtyPort`] whose `subscribe_output` replays a fixed list of chunks then
    /// ends (EOF), so the watcher thread sees a deterministic, finite stream. `write` is
    /// recorded; `spawn`/`resize`/`kill`/`scrollback` are unused stubs for these tests.
    struct FakePty {
        chunks: Mutex<HashMap<SessionId, Vec<Vec<u8>>>>,
        writes: Mutex<Vec<Vec<u8>>>,
    }
    impl FakePty {
        fn new() -> Self {
            Self {
                chunks: Mutex::new(HashMap::new()),
                writes: Mutex::new(Vec::new()),
            }
        }
        /// Seeds the chunks a later `subscribe_output(handle)` will replay.
        fn seed(&self, handle: &Handle, chunks: Vec<Vec<u8>>) {
            self.chunks
                .lock()
                .unwrap()
                .insert(handle.session_id.clone(), chunks);
        }
    }
    #[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> {
            let chunks = self
                .chunks
                .lock()
                .unwrap()
                .get(&handle.session_id)
                .cloned()
                .unwrap_or_default();
            Ok(Box::new(chunks.into_iter()))
        }
        fn scrollback(&self, _handle: &Handle) -> Result<Vec<u8>, PtyError> {
            Ok(Vec::new())
        }
        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 (the watcher runs on its own
    /// thread, so the transition 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()
    }

    fn inbox_with(pty: Arc<FakePty>) -> MediatedInbox {
        MediatedInbox::with_pty(
            Arc::new(InMemoryMailbox::new()),
            Arc::new(FixedClock(1)),
            pty as Arc<dyn PtyPort>,
        )
    }

    #[test]
    fn prompt_pattern_present_and_output_contains_it_flips_busy_to_idle() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(1);
        pty.seed(&h, vec![b"working...\n".to_vec(), b"done\n> ".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));

        inbox.enqueue(a, ticket(10, "task"));
        assert!(inbox.busy_state(a).is_busy(), "enqueue starts a turn");

        // Arm prompt detection with the literal marker "\n> " (a stable prompt sigil).
        inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()));

        assert!(
            wait_until(|| !inbox.busy_state(a).is_busy()),
            "prompt marker in output ⇒ Busy→Idle"
        );
    }

    #[test]
    fn prompt_marker_split_across_two_chunks_still_matches() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(2);
        // The marker "READY" straddles two chunks: "REA" | "DY done".
        pty.seed(&h, vec![b"out REA".to_vec(), b"DY done".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));
        inbox.enqueue(a, ticket(10, "t"));
        inbox.bind_handle_with_prompt(a, h, Some("READY".to_owned()));
        assert!(
            wait_until(|| !inbox.busy_state(a).is_busy()),
            "a marker split across chunks must still flip to Idle"
        );
    }

    #[test]
    fn no_pattern_in_profile_never_marks_idle_even_with_output() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(3);
        pty.seed(&h, vec![b"lots of output\n> $ done\n".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));
        inbox.enqueue(a, ticket(10, "t"));
        // No pattern: bind without arming a watcher (the safe fallback).
        inbox.bind_handle_with_prompt(a, h, None);
        // Give any (erroneously spawned) watcher a chance to fire — it must not.
        std::thread::sleep(std::time::Duration::from_millis(80));
        assert!(
            inbox.busy_state(a).is_busy(),
            "no pattern ⇒ never a false Idle; the agent stays Busy"
        );
        // The FIFO still accepts a second enqueue (forward, never reject).
        inbox.enqueue(a, ticket(11, "second"));
        assert_eq!(inbox.mailbox().pending(&a), 2, "enqueue accepted while Busy");
    }

    #[test]
    fn pattern_absent_from_output_keeps_agent_busy_but_queue_accepts() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(4);
        // Output never contains the marker ⇒ watcher runs to EOF without matching.
        pty.seed(&h, vec![b"still thinking, no prompt here\n".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));
        inbox.enqueue(a, ticket(10, "t"));
        inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()));
        // Even after the stream ends, no match ⇒ stays Busy (timeout is the ultimate
        // guard, exercised at the orchestrator layer).
        std::thread::sleep(std::time::Duration::from_millis(80));
        assert!(
            inbox.busy_state(a).is_busy(),
            "marker absent from output ⇒ stays Busy (no false Idle)"
        );
        inbox.enqueue(a, ticket(11, "queued"));
        assert_eq!(
            inbox.mailbox().pending(&a),
            2,
            "in doubt → keep accepting (forward, never reject)"
        );
    }

    #[test]
    fn empty_pattern_arms_nothing() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(5);
        pty.seed(&h, vec![b"anything\n> ".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));
        inbox.enqueue(a, ticket(10, "t"));
        inbox.bind_handle_with_prompt(a, h, Some(String::new()));
        std::thread::sleep(std::time::Duration::from_millis(60));
        assert!(
            inbox.busy_state(a).is_busy(),
            "an empty pattern must arm no watcher (treated as no detection)"
        );
    }

    #[test]
    fn prompt_match_advances_fifo_to_next_ticket() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(6);
        pty.seed(&h, vec![b"done\n> ".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));
        // Two tickets queued; the turn is on the first.
        inbox.enqueue(a, ticket(10, "first"));
        inbox.enqueue(a, ticket(11, "second"));
        assert_eq!(
            inbox.busy_state(a).ticket(),
            Some(domain::mailbox::TicketId::from_uuid(uuid::Uuid::from_u128(10)))
        );
        inbox.bind_handle_with_prompt(a, h, Some("\n> ".to_owned()));
        // Prompt-ready ⇒ Idle ⇒ the next enqueue can start a fresh turn.
        assert!(wait_until(|| !inbox.busy_state(a).is_busy()));
        inbox.enqueue(a, ticket(12, "third"));
        assert_eq!(
            inbox.busy_state(a).ticket(),
            Some(domain::mailbox::TicketId::from_uuid(uuid::Uuid::from_u128(12))),
            "after prompt-ready Idle, a new enqueue starts the next turn"
        );
    }

    #[test]
    fn rebinding_same_agent_does_not_spawn_duplicate_watcher() {
        let pty = Arc::new(FakePty::new());
        let a = agent(1);
        let h = handle(7);
        // A stream that never matches and never ends quickly: empty ⇒ immediate EOF.
        pty.seed(&h, vec![b"noise".to_vec()]);
        let inbox = inbox_with(Arc::clone(&pty));
        inbox.enqueue(a, ticket(10, "t"));
        // Arm twice in a row; the second must be a no-op while the first is live (no
        // panic, no double-subscribe). After EOF the agent is un-armed and stays Busy.
        inbox.bind_handle_with_prompt(a, h.clone(), Some("ZZZ".to_owned()));
        inbox.bind_handle_with_prompt(a, h, Some("ZZZ".to_owned()));
        std::thread::sleep(std::time::Duration::from_millis(60));
        assert!(inbox.busy_state(a).is_busy(), "no match ⇒ stays Busy");
    }
}