blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(runtime): isolate agent state by project (#101)

Browse Source
This commit is contained in:
Blomios
2026-07-25 22:14:02 +02:00
parent 6a87c4635f
commit 6e98fd89f7
52 changed files with 1894 additions and 805 deletions
Download Patch File Download Diff File Expand all files Collapse all files

286
crates/infrastructure/src/input/mod.rs
View File

@ -25,7 +25,7 @@ use std::sync::{Arc, Mutex};
use std::time::Duration;
use domain::events::DomainEvent;
use domain::ids::AgentId;
use domain::ids::RuntimeAgentKey;
use domain::inbox::{
AgentInbox, AgentInboxSnapshot, InboxError, InboxItem, InboxReceipt, InboxReceiptStatus,
InboxSource, DEFAULT_AGENT_INBOX_CAPACITY,
@ -44,11 +44,11 @@ use crate::mailbox::InMemoryMailbox;
/// 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>>,
busy: Mutex<HashMap<RuntimeAgentKey, AgentBusyState>>,
/// Per-agent **liveness** bookkeeping (lot 2) : dernier battement observé, seuil de
/// stagnation issu du profil, et état de vivacité courant pour n'émettre
/// `AgentLivenessChanged` qu'**une fois par transition** (pas de spam).
liveness: Mutex<HashMap<AgentId, LivenessState>>,
liveness: Mutex<HashMap<RuntimeAgentKey, LivenessState>>,
/// **Démarrage à froid** (fix race cold-launch) : ensemble des agents fraîchement
/// lancés à froid pour lesquels la livraison du **premier** tour doit être *gatée*
/// sur la readiness MCP. Un agent y est inscrit par
@ -56,11 +56,11 @@ struct BusyTracker {
/// 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>>,
starting: Mutex<HashSet<RuntimeAgentKey>>,
/// **Tour différé** (fix race cold-launch) : payload de la `DelegationReady` retenue
/// pour un démarrage à froid, publiée par [`BusyTracker::release_cold_start`] à la
/// connexion du pont MCP (jamais avant). Absent ⇒ aucun tour en attente de gate.
deferred: Mutex<HashMap<AgentId, DeferredDelegation>>,
deferred: Mutex<HashMap<RuntimeAgentKey, DeferredDelegation>>,
/// **Latch « déjà libéré »** (fix race cold-start, ordre inverse) : ensemble des
/// agents pour lesquels le signal de readiness (`release_cold_start`, pont MCP) est
/// arrivé **avant** que l'`enqueue` n'ait parqué son tour dans
@ -71,7 +71,7 @@ struct BusyTracker {
/// 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>>,
released: Mutex<HashSet<RuntimeAgentKey>>,
events: Option<Arc<dyn EventBus>>,
/// Sink de livraison headless (cf. [`HeadlessSink`]). Câblé par
/// [`MediatedInbox::with_pty`]/[`MediatedInbox::with_events`] quand un PTY est
@ -118,7 +118,7 @@ struct DeferredDelegation {
/// 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>;
Arc<dyn Fn(RuntimeAgentKey, DeferredDelegation) -> Option<DeferredDelegation> + Send + Sync>;
/// Délai (ms) entre l'écriture du texte de la tâche et celle de la séquence de
/// soumission lors d'une livraison **headless** (le médiateur écrit lui-même le PTY).
@ -162,19 +162,21 @@ impl BusyTracker {
}
}
fn lock_starting(&self) -> std::sync::MutexGuard<'_, HashSet<AgentId>> {
fn lock_starting(&self) -> std::sync::MutexGuard<'_, HashSet<RuntimeAgentKey>> {
self.starting
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_deferred(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, DeferredDelegation>> {
fn lock_deferred(
&self,
) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, DeferredDelegation>> {
self.deferred
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_released(&self) -> std::sync::MutexGuard<'_, HashSet<AgentId>> {
fn lock_released(&self) -> std::sync::MutexGuard<'_, HashSet<RuntimeAgentKey>> {
self.released
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
@ -186,28 +188,28 @@ impl BusyTracker {
/// (`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) {
fn mark_starting(&self, agent: RuntimeAgentKey) {
application::diag!("[input-mediator] mark_starting cold-start gate armed agent={agent}");
self.lock_starting().insert(agent);
}
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, AgentBusyState>> {
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, AgentBusyState>> {
self.busy
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_liveness(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, LivenessState>> {
fn lock_liveness(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, LivenessState>> {
self.liveness
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// Publie un `AgentLivenessChanged` (si un bus est câblé).
fn publish_liveness(&self, agent: AgentId, liveness: AgentLiveness) {
fn publish_liveness(&self, agent: RuntimeAgentKey, liveness: AgentLiveness) {
if let Some(events) = &self.events {
events.publish(DomainEvent::AgentLivenessChanged {
agent_id: agent,
agent_id: agent.agent_id,
liveness,
});
}
@ -217,7 +219,7 @@ impl BusyTracker {
/// 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) {
fn arm_liveness(&self, agent: RuntimeAgentKey, stall_after_ms: Option<u32>, now_ms: u64) {
self.lock_liveness().insert(
agent,
LivenessState {
@ -231,7 +233,7 @@ impl BusyTracker {
/// 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) {
fn touch(&self, agent: RuntimeAgentKey, now_ms: u64) {
let recovered = {
let mut map = self.lock_liveness();
let Some(state) = map.get_mut(&agent) else {
@ -256,7 +258,7 @@ impl BusyTracker {
/// `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 newly_stalled: Vec<RuntimeAgentKey> = {
let mut map = self.lock_liveness();
map.iter_mut()
.filter_map(|(agent, state)| {
@ -285,7 +287,7 @@ impl BusyTracker {
/// 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) {
fn clear_liveness(&self, agent: RuntimeAgentKey) {
let was_stalled = self
.lock_liveness()
.remove(&agent)
@ -295,7 +297,7 @@ impl BusyTracker {
}
}
fn busy_state(&self, agent: AgentId) -> AgentBusyState {
fn busy_state(&self, agent: RuntimeAgentKey) -> AgentBusyState {
self.lock()
.get(&agent)
.copied()
@ -304,7 +306,7 @@ impl BusyTracker {
/// 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 {
fn start_turn(&self, agent: RuntimeAgentKey, state: AgentBusyState) -> bool {
let mut busy = self.lock();
let entry = busy.entry(agent).or_insert(AgentBusyState::Idle);
if entry.is_busy() {
@ -320,7 +322,7 @@ impl BusyTracker {
/// 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) {
fn publish_deferred(&self, agent: RuntimeAgentKey, d: DeferredDelegation) {
// Point de livraison unique (tours chauds immédiats ET drains à froid). Si un
// sink headless est câblé, il a la priorité : pour un agent sans cellule
// frontend il écrit lui-même la tâche dans le PTY et renvoie `None` (pris en
@ -346,7 +348,7 @@ impl BusyTracker {
d.submit_delay_ms,
);
events.publish(DomainEvent::DelegationReady {
agent_id: agent,
agent_id: agent.agent_id,
ticket: d.ticket,
text: d.text,
submit_sequence: d.submit_sequence,
@ -368,7 +370,7 @@ impl BusyTracker {
/// 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) {
fn turn_ended(&self, agent: RuntimeAgentKey) {
let active = self.lock().get(&agent).and_then(AgentBusyState::ticket);
application::diag!(
"[input-mediator] turn_ended agent={agent} no_reply_payload -> mark_idle + grace"
@ -395,7 +397,7 @@ impl BusyTracker {
/// 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) {
fn release_cold_start(&self, agent: RuntimeAgentKey) {
let was_starting = self.lock_starting().remove(&agent);
if let Some(d) = self.lock_deferred().remove(&agent) {
// Cas nominal : l'`enqueue` a déjà parqué le tour ⇒ on le draine.
@ -421,7 +423,7 @@ impl BusyTracker {
/// 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) {
fn mark_idle(&self, agent: RuntimeAgentKey) {
let was_busy = {
let mut busy = self.lock();
busy.insert(agent, AgentBusyState::Idle)
@ -431,7 +433,7 @@ impl BusyTracker {
application::diag!("[input-mediator] mark_idle turn ended agent={agent}");
if let Some(events) = &self.events {
events.publish(DomainEvent::AgentBusyChanged {
agent_id: agent,
agent_id: agent.agent_id,
busy: false,
});
}
@ -485,21 +487,21 @@ pub struct MediatedInbox {
/// 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>>>,
handles: Arc<Mutex<HashMap<RuntimeAgentKey, PtyHandle>>>,
/// Agents qui ont une **cellule terminal frontend montée** (write-portal actif),
/// tenu à jour par [`InputMediator::set_front_attached`]. Quand un agent y figure,
/// la livraison passe par l'événement `DelegationReady` (le front écrit) ; sinon
/// (agent headless / délégué en arrière-plan) le médiateur écrit lui-même le tour
/// dans le PTY. `Arc` car le sink headless du tracker le consulte.
front_owned: Arc<Mutex<HashSet<AgentId>>>,
front_owned: Arc<Mutex<HashSet<RuntimeAgentKey>>>,
/// Per-agent submit config (target profile's `submit_sequence`/`submit_delay_ms`),
/// stashed at bind time (§20.3) and echoed on the `DelegationReady` event when a
/// turn starts. Absent ⇒ both `None` (the front applies its defaults).
submit: Mutex<HashMap<AgentId, SubmitConfig>>,
submit: Mutex<HashMap<RuntimeAgentKey, SubmitConfig>>,
/// Per-agent stall threshold (`LivenessStrategy::stall_after_ms`, lot 2), stashed by
/// `set_stall_threshold` and consumed to arm a fresh liveness window on the enqueue
/// that starts a turn. Absent ⇒ `None` (no stall detection — legacy behaviour).
stall: Mutex<HashMap<AgentId, Option<u32>>>,
stall: Mutex<HashMap<RuntimeAgentKey, Option<u32>>>,
/// Rich inbox payloads keyed by the mailbox ticket id. The mailbox remains the
/// only FIFO; this map is metadata only.
inbox_items: Mutex<HashMap<TicketId, InboxItem>>,
@ -545,8 +547,8 @@ impl MediatedInbox {
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>>>,
handles: &Arc<Mutex<HashMap<RuntimeAgentKey, PtyHandle>>>,
front_owned: &Arc<Mutex<HashSet<RuntimeAgentKey>>>,
mailbox: &Arc<InMemoryMailbox>,
grace: Duration,
) -> Arc<BusyTracker> {
@ -572,10 +574,10 @@ impl MediatedInbox {
/// `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>>>,
handles: Arc<Mutex<HashMap<RuntimeAgentKey, PtyHandle>>>,
front_owned: Arc<Mutex<HashSet<RuntimeAgentKey>>>,
) -> HeadlessSink {
Arc::new(move |agent: AgentId, d: DeferredDelegation| {
Arc::new(move |agent: RuntimeAgentKey, d: DeferredDelegation| {
// Cellule frontend montée ⇒ c'est le write-portal qui écrit (et qui sait
// composer avec une saisie humaine en cours). On rend la main.
if front_owned
@ -749,19 +751,19 @@ impl MediatedInbox {
self
}
fn handles(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, PtyHandle>> {
fn handles(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, PtyHandle>> {
self.handles
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn submit(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, SubmitConfig>> {
fn submit(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, SubmitConfig>> {
self.submit
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn stall(&self) -> std::sync::MutexGuard<'_, HashMap<AgentId, Option<u32>>> {
fn stall(&self) -> std::sync::MutexGuard<'_, HashMap<RuntimeAgentKey, Option<u32>>> {
self.stall
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
@ -843,7 +845,7 @@ fn describe_submit_sequence(value: Option<&str>) -> String {
}
impl InputMediator for MediatedInbox {
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
fn enqueue(&self, agent: RuntimeAgentKey, ticket: Ticket) -> PendingReply {
let ticket_id = ticket.id;
// If the agent is Idle, this enqueue starts its turn ⇒ go Busy. If already
// Busy, we still accept (queue grows; the turn advances on mark_idle) — never
@ -875,7 +877,7 @@ impl InputMediator for MediatedInbox {
if started_turn {
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentBusyChanged {
agent_id: agent,
agent_id: agent.agent_id,
busy: true,
});
let submit = self.submit().get(&agent).cloned().unwrap_or_default();
@ -921,7 +923,7 @@ impl InputMediator for MediatedInbox {
self.mailbox.enqueue(agent, ticket)
}
fn enqueue_silent(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
fn enqueue_silent(&self, agent: RuntimeAgentKey, ticket: Ticket) -> PendingReply {
let ticket_id = ticket.id;
// Headless/system turns share the same FIFO and busy/liveness accounting as
// terminal-delivered turns, but their prompt is sent through AgentSession::send.
@ -940,7 +942,7 @@ impl InputMediator for MediatedInbox {
self.tracker.arm_liveness(agent, stall_after_ms, now_ms);
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentBusyChanged {
agent_id: agent,
agent_id: agent.agent_id,
busy: true,
});
}
@ -948,7 +950,7 @@ impl InputMediator for MediatedInbox {
self.mailbox.enqueue(agent, ticket)
}
fn bind_handle(&self, agent: AgentId, handle: PtyHandle) {
fn bind_handle(&self, agent: RuntimeAgentKey, handle: PtyHandle) {
eprintln!(
"[input-mediator] bind handle agent={agent} handle={}",
handle.session_id
@ -956,7 +958,12 @@ impl InputMediator for MediatedInbox {
self.handles().insert(agent, handle);
}
fn bind_handle_with_submit(&self, agent: AgentId, handle: PtyHandle, submit: SubmitConfig) {
fn bind_handle_with_submit(
&self,
agent: RuntimeAgentKey,
handle: PtyHandle,
submit: SubmitConfig,
) {
// Register the input handle exactly like `bind_handle` and stash the target's
// submit config (echoed on `DelegationReady` at the next turn start, §20.3).
// Turn-end detection is NO LONGER armed here: the dead PTY prompt-ready watcher
@ -972,11 +979,11 @@ impl InputMediator for MediatedInbox {
self.submit().insert(agent, submit);
}
fn delivers_turn(&self, agent: AgentId) -> bool {
fn delivers_turn(&self, agent: RuntimeAgentKey) -> bool {
self.pty.is_some() && self.handles().get(&agent).is_some()
}
fn mark_starting(&self, agent: AgentId) {
fn mark_starting(&self, agent: RuntimeAgentKey) {
// Gate du premier tour d'un agent froid : appelé par l'orchestrateur juste après
// un (re)lancement à froid, AVANT le bind/enqueue, et uniquement si un signal de
// readiness le libérera (pont MCP). Consommé par l'`enqueue` qui démarre le tour
@ -984,11 +991,11 @@ impl InputMediator for MediatedInbox {
self.tracker.mark_starting(agent);
}
fn release_cold_start(&self, agent: AgentId) {
fn release_cold_start(&self, agent: RuntimeAgentKey) {
self.tracker.release_cold_start(agent);
}
fn set_front_attached(&self, agent: AgentId, attached: bool) {
fn set_front_attached(&self, agent: RuntimeAgentKey, attached: bool) {
let mut front = self
.front_owned
.lock()
@ -1002,7 +1009,7 @@ impl InputMediator for MediatedInbox {
}
}
fn preempt(&self, agent: AgentId) {
fn preempt(&self, agent: RuntimeAgentKey) {
// Interrompre: signals the running turn to stop. It is NOT an enqueue and
// correlates **no** ticket (we never pop/resolve a pending caller — preempt
// must never silently answer one). The only effect is a best-effort interrupt
@ -1018,42 +1025,46 @@ impl InputMediator for MediatedInbox {
}
}
fn mark_idle(&self, agent: AgentId) {
fn mark_idle(&self, agent: RuntimeAgentKey) {
// Single authority: real Busy→Idle only, publishing AgentBusyChanged{busy:false}
// once. Also clears the liveness entry (fin de tour) — émet la reprise si l'agent
// était `Stalled`.
self.tracker.mark_idle(agent);
}
fn turn_ended(&self, agent: AgentId) {
fn turn_ended(&self, agent: RuntimeAgentKey) {
// Déclenché par le `TurnWatcher` transcript (Claude `turn_duration`) : fin de tour
// sans `idea_reply` ⇒ `mark_idle` + fenêtre de grâce pour réveiller l'appelant
// parqué (cf. [`BusyTracker::turn_ended`]). Backstop no-reply.
self.tracker.turn_ended(agent);
}
fn mark_alive(&self, agent: AgentId) {
fn mark_alive(&self, agent: RuntimeAgentKey) {
// Un battement (delta / activité / heartbeat) rafraîchit `last_seen` et ramène
// l'agent à `Alive` s'il était `Stalled` (lot 2).
self.tracker.touch(agent, self.clock.now_ms());
}
fn set_stall_threshold(&self, agent: AgentId, stall_after_ms: Option<u32>) {
fn set_stall_threshold(&self, agent: RuntimeAgentKey, stall_after_ms: Option<u32>) {
// Stashé par l'orchestrateur depuis le profil de la cible AVANT l'enqueue qui
// démarre le tour ; consommé par `arm_liveness` au start_turn (lot 2).
self.stall().insert(agent, stall_after_ms);
}
fn busy_state(&self, agent: AgentId) -> AgentBusyState {
fn busy_state(&self, agent: RuntimeAgentKey) -> 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 {
fn enqueue_message(
&self,
agent: RuntimeAgentKey,
item: InboxItem,
) -> Result<InboxReceipt, InboxError> {
if item.agent_id != agent.agent_id {
return Err(InboxError::AgentMismatch {
agent_id: agent,
agent_id: agent.agent_id,
item_agent_id: item.agent_id,
});
}
@ -1063,13 +1074,14 @@ impl AgentInbox for MediatedInbox {
if item.is_lossless_system() {
return Ok(InboxReceipt {
item_id: item.id,
agent_id: agent,
agent_id: agent.agent_id,
runtime_key: agent,
depth: depth_before,
status: InboxReceiptStatus::Deferred,
});
}
return Err(InboxError::InboxFull {
agent_id: agent,
agent_id: agent.agent_id,
capacity: self.inbox_capacity,
});
}
@ -1081,32 +1093,33 @@ impl AgentInbox for MediatedInbox {
let depth = self.mailbox.pending(&agent);
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentInboxQueued {
agent_id: agent,
agent_id: agent.agent_id,
depth,
});
}
Ok(InboxReceipt {
item_id,
agent_id: agent,
agent_id: agent.agent_id,
runtime_key: agent,
depth,
status: InboxReceiptStatus::Queued,
})
}
fn dequeue_next(&self, agent: AgentId) -> Option<InboxItem> {
fn dequeue_next(&self, agent: RuntimeAgentKey) -> Option<InboxItem> {
let head = self.mailbox.head_ticket(&agent)?;
let item = self.inbox_items().remove(&head)?;
self.mailbox.cancel_head(agent, head);
if let Some(events) = &self.tracker.events {
events.publish(DomainEvent::AgentInboxDrained {
agent_id: agent,
agent_id: agent.agent_id,
depth: self.snapshot(agent).depth,
});
}
Some(item)
}
fn snapshot(&self, agent: AgentId) -> AgentInboxSnapshot {
fn snapshot(&self, agent: RuntimeAgentKey) -> AgentInboxSnapshot {
let items_by_id = self.inbox_items();
let items: Vec<InboxItem> = self
.mailbox
@ -1115,7 +1128,8 @@ impl AgentInbox for MediatedInbox {
.filter_map(|ticket| items_by_id.get(&ticket.id).cloned())
.collect();
AgentInboxSnapshot {
agent_id: agent,
agent_id: agent.agent_id,
runtime_key: agent,
depth: self.mailbox.pending(&agent),
items,
}
@ -1152,6 +1166,7 @@ mod tests {
use super::*;
use domain::conversation::ConversationId;
use domain::mailbox::TicketId;
use domain::AgentId;
/// Deterministic clock for assertions on `since_ms`.
struct FixedClock(u64);
@ -1165,6 +1180,13 @@ mod tests {
AgentId::from_uuid(uuid::Uuid::from_u128(n))
}
fn key(n: u128) -> RuntimeAgentKey {
RuntimeAgentKey::new(
domain::ProjectId::from_uuid(uuid::Uuid::from_u128(1000 + n)),
agent(n),
)
}
fn ticket(n: u128, task: &str) -> Ticket {
Ticket::from_human(
TicketId::from_uuid(uuid::Uuid::from_u128(n)),
@ -1240,27 +1262,33 @@ mod tests {
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 a = key(1);
// First enqueue starts a turn ⇒ exactly one Busy(true) event.
inbox.enqueue(a, ticket(10, "first"));
assert_eq!(bus.busy_events(), vec![(a, true)]);
assert_eq!(bus.busy_events(), vec![(a.agent_id, true)]);
// Second enqueue while Busy queues behind ⇒ NO new busy event.
inbox.enqueue(a, ticket(11, "second"));
assert_eq!(
bus.busy_events(),
vec![(a, true)],
vec![(a.agent_id, true)],
"no re-announce while busy"
);
// mark_idle on a busy agent ⇒ exactly one Idle(false) event.
inbox.mark_idle(a);
assert_eq!(bus.busy_events(), vec![(a, true), (a, false)]);
assert_eq!(
bus.busy_events(),
vec![(a.agent_id, true), (a.agent_id, false)]
);
// mark_idle on an already-idle agent ⇒ no spurious event.
inbox.mark_idle(a);
assert_eq!(bus.busy_events(), vec![(a, true), (a, false)]);
assert_eq!(
bus.busy_events(),
vec![(a.agent_id, true), (a.agent_id, false)]
);
}
#[test]
@ -1268,11 +1296,11 @@ mod tests {
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 a = key(1);
inbox.enqueue(a, ticket(10, "t"));
inbox.preempt(a);
// Only the enqueue's Busy(true); preempt does not toggle busy state.
assert_eq!(bus.busy_events(), vec![(a, true)]);
assert_eq!(bus.busy_events(), vec![(a.agent_id, true)]);
}
// ====================================================================
@ -1284,7 +1312,7 @@ mod tests {
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 a = key(1);
// Idle→Busy ⇒ exactly one DelegationReady carrying the task text + ticket.
inbox.enqueue(a, ticket(10, "do the thing"));
@ -1325,14 +1353,14 @@ mod tests {
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = agent(1);
let a = key(1);
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.set_front_attached(a, true);
inbox.enqueue_silent(a, ticket(10, "headless turn"));
assert!(inbox.busy_state(a).is_busy());
assert_eq!(bus.busy_events(), vec![(a, true)]);
assert_eq!(bus.busy_events(), vec![(a.agent_id, true)]);
assert!(
bus.delegation_ready().is_empty(),
"headless bookkeeping must not leak a prompt to the terminal"
@ -1349,11 +1377,11 @@ mod tests {
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 a = key(1);
let task_id = domain::TaskId::from_uuid(uuid::Uuid::from_u128(42));
let item = domain::InboxItem {
id: TicketId::from_uuid(uuid::Uuid::from_u128(10)),
agent_id: a,
agent_id: a.agent_id,
source: InboxSource::BackgroundTask { task_id },
kind: domain::InboxItemKind::BackgroundCompletion,
body: "Background task completed.".to_owned(),
@ -1384,7 +1412,7 @@ mod tests {
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = agent(1);
let a = key(1);
let h = handle(1);
// Bind the target's submit config (resolved from its profile by the service).
@ -1405,7 +1433,7 @@ mod tests {
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 a = key(1);
inbox.enqueue(a, ticket(10, "task"));
let ready = bus.delegation_ready();
assert_eq!(ready.len(), 1);
@ -1428,7 +1456,7 @@ mod tests {
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = agent(1);
let a = key(1);
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.set_front_attached(a, true);
@ -1460,7 +1488,7 @@ mod tests {
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = agent(1);
let a = key(1);
// No set_front_attached ⇒ headless.
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
@ -1495,7 +1523,7 @@ mod tests {
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = agent(1);
let a = key(1);
let task = format!("début {} fin", "x".repeat(1200));
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
@ -1531,7 +1559,7 @@ mod tests {
#[tokio::test]
async fn enqueue_returns_pending_reply_resolved_via_mailbox() {
let inbox = inbox_at(5);
let a = agent(1);
let a = key(1);
let pending = inbox.enqueue(a, ticket(10, "do X"));
// Resolve through the shared mailbox (the orchestrator's path).
inbox.mailbox().resolve(a, "done".to_owned()).unwrap();
@ -1544,7 +1572,7 @@ mod tests {
#[test]
fn first_enqueue_marks_busy_with_ticket_and_stamp() {
let inbox = inbox_at(1234);
let a = agent(1);
let a = key(1);
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
inbox.enqueue(a, ticket(10, "t"));
assert_eq!(
@ -1559,7 +1587,7 @@ mod tests {
#[test]
fn second_enqueue_while_busy_keeps_first_ticket_and_does_not_reject() {
let inbox = inbox_at(1);
let a = agent(1);
let a = key(1);
inbox.enqueue(a, ticket(10, "first"));
inbox.enqueue(a, ticket(11, "second")); // accepted, queues behind
// Still busy on the FIRST ticket (turn unchanged), both queued in the mailbox.
@ -1573,7 +1601,7 @@ mod tests {
#[test]
fn mark_idle_returns_to_idle_so_next_turn_can_start() {
let inbox = inbox_at(1);
let a = agent(1);
let a = key(1);
inbox.enqueue(a, ticket(10, "t"));
assert!(inbox.busy_state(a).is_busy());
inbox.mark_idle(a);
@ -1589,7 +1617,7 @@ mod tests {
#[tokio::test]
async fn preempt_is_distinct_from_enqueue_and_resolves_no_ticket() {
let inbox = inbox_at(1);
let a = agent(1);
let a = key(1);
let pending = inbox.enqueue(a, ticket(10, "t"));
inbox.preempt(a);
// preempt did not pop/resolve the ticket: still pending in the mailbox.
@ -1605,7 +1633,7 @@ mod tests {
#[test]
fn two_enqueues_same_agent_serialise_in_one_fifo() {
let inbox = inbox_at(1);
let a = agent(1);
let a = key(1);
inbox.enqueue(a, ticket(10, "first"));
inbox.enqueue(a, ticket(11, "second"));
assert_eq!(inbox.mailbox().pending(&a), 2);
@ -1619,8 +1647,8 @@ mod tests {
#[test]
fn different_agents_are_independent_not_blocking() {
let inbox = inbox_at(1);
let a = agent(1);
let b = agent(2);
let a = key(1);
let b = key(2);
inbox.enqueue(a, ticket(10, "a"));
inbox.enqueue(b, ticket(20, "b"));
assert!(inbox.busy_state(a).is_busy());
@ -1721,7 +1749,7 @@ mod tests {
/// 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 a = key(1);
let inbox = inbox_grace(Duration::from_millis(40));
let pending = inbox.enqueue(a, ticket(10, "task"));
@ -1747,7 +1775,7 @@ mod tests {
/// 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 a = key(1);
let inbox = inbox_grace(Duration::from_millis(40));
let pending = inbox.enqueue(a, ticket(10, "task"));
@ -1769,7 +1797,7 @@ mod tests {
/// 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);
let a = key(1);
// Long grace ⇒ the reply lands well within it.
let inbox = inbox_grace(Duration::from_secs(30));
@ -1796,7 +1824,7 @@ mod tests {
/// é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 a = key(1);
let inbox = inbox_grace(Duration::from_millis(30));
let pending = inbox.enqueue(a, ticket(10, "task"));
@ -1822,7 +1850,7 @@ mod tests {
/// 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 a = key(1);
let inbox = inbox_grace(Duration::from_millis(20));
// Human submit: the reply handle is dropped immediately (not awaited).
@ -1841,7 +1869,7 @@ mod tests {
/// idempotent, aucune grâce armée, aucun panic.
#[test]
fn turn_ended_on_idle_agent_is_noop() {
let a = agent(1);
let a = key(1);
let inbox = inbox_grace(Duration::from_millis(20));
inbox.turn_ended(a); // jamais de tour démarré.
assert_eq!(inbox.busy_state(a), AgentBusyState::Idle);
@ -1859,7 +1887,7 @@ mod tests {
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 a = key(1);
// Démarrage à froid : gate armé AVANT l'enqueue (ordre de l'orchestrateur).
inbox.mark_starting(a);
@ -1893,7 +1921,7 @@ mod tests {
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 a = key(1);
// Démarrage à froid : gate armé AVANT l'enqueue (ordre de l'orchestrateur).
inbox.mark_starting(a);
@ -1927,7 +1955,7 @@ mod tests {
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 a = key(1);
// Démarre un tour normal (Busy), SANS gate cold-launch (pas de mark_starting).
inbox.enqueue(a, ticket(10, "task"));
@ -1952,7 +1980,7 @@ mod tests {
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 a = key(1);
inbox.mark_starting(a);
inbox.enqueue(a, ticket(10, "cold task"));
@ -1973,7 +2001,7 @@ mod tests {
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 a = key(1);
// Pas de mark_starting ⇒ agent chaud.
inbox.enqueue(a, ticket(10, "warm task"));
@ -1994,7 +2022,7 @@ mod tests {
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 a = key(1);
// Cold launch SANS pont MCP ⇒ l'orchestrateur N'APPELLE PAS mark_starting.
inbox.enqueue(a, ticket(10, "task"));
@ -2017,7 +2045,7 @@ mod tests {
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 a = key(1);
inbox.mark_starting(a);
inbox.enqueue(a, ticket(10, "first"));
@ -2091,7 +2119,7 @@ mod tests {
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);
let a = key(1);
// Profil : seuil de stagnation à 30_000 ms. Armé avant le tour.
inbox.set_stall_threshold(a, Some(30_000));
@ -2109,14 +2137,17 @@ mod tests {
// 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)]);
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)]
);
// Idempotent : un second sweep ne ré-émet pas.
clock.set(1_000 + 60_000);
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![(a, AgentLiveness::Stalled)],
vec![(a.agent_id, AgentLiveness::Stalled)],
"déjà stalled ⇒ pas de spam"
);
}
@ -2126,7 +2157,7 @@ mod tests {
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);
let a = key(1);
inbox.set_stall_threshold(a, Some(30_000));
inbox.enqueue(a, ticket(10, "task")); // last_seen=1_000
@ -2149,7 +2180,7 @@ mod tests {
fn liveness_event_once_per_transition() {
let clock = MutClock::new(0);
let (inbox, bus) = inbox_with_clock(Arc::clone(&clock));
let a = agent(1);
let a = key(1);
inbox.set_stall_threshold(a, Some(10_000));
inbox.enqueue(a, ticket(10, "t")); // last_seen=0
@ -2159,7 +2190,10 @@ mod tests {
// Sweeps répétés : pas de ré-émission.
inbox.sweep_stalled();
inbox.sweep_stalled();
assert_eq!(bus.liveness_events(), vec![(a, AgentLiveness::Stalled)]);
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)]
);
// Battement tardif ⇒ une seule reprise Alive.
clock.set(20_000);
@ -2167,7 +2201,10 @@ mod tests {
inbox.mark_alive(a); // second battement : déjà Alive ⇒ rien.
assert_eq!(
bus.liveness_events(),
vec![(a, AgentLiveness::Stalled), (a, AgentLiveness::Alive)]
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive)
]
);
// Re-stall possible après reprise (nouvelle transition).
@ -2176,9 +2213,9 @@ mod tests {
assert_eq!(
bus.liveness_events(),
vec![
(a, AgentLiveness::Stalled),
(a, AgentLiveness::Alive),
(a, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive),
(a.agent_id, AgentLiveness::Stalled),
]
);
}
@ -2188,17 +2225,23 @@ mod tests {
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);
let a = key(1);
inbox.set_stall_threshold(a, Some(5_000));
inbox.enqueue(a, ticket(10, "t"));
clock.set(5_001);
inbox.sweep_stalled();
assert_eq!(bus.liveness_events(), vec![(a, AgentLiveness::Stalled)]);
assert_eq!(
bus.liveness_events(),
vec![(a.agent_id, AgentLiveness::Stalled)]
);
inbox.mark_idle(a); // fin de tour ⇒ reprise Alive + entrée retirée.
assert_eq!(
bus.liveness_events(),
vec![(a, AgentLiveness::Stalled), (a, AgentLiveness::Alive)]
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive)
]
);
// Plus d'entrée : un sweep ultérieur ne ré-émet rien (même très tard).
@ -2206,7 +2249,10 @@ mod tests {
inbox.sweep_stalled();
assert_eq!(
bus.liveness_events(),
vec![(a, AgentLiveness::Stalled), (a, AgentLiveness::Alive)]
vec![
(a.agent_id, AgentLiveness::Stalled),
(a.agent_id, AgentLiveness::Alive)
]
);
}
@ -2216,7 +2262,7 @@ mod tests {
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);
let a = key(1);
// Pas de set_stall_threshold (ou None) : armé sans seuil au start_turn.
inbox.enqueue(a, ticket(10, "t"));
clock.set(10_000_000); // très loin dans le futur.
@ -2236,7 +2282,7 @@ mod tests {
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);
let a = key(1);
inbox.mark_alive(a); // jamais de tour démarré.
inbox.sweep_stalled();
assert_eq!(bus.liveness_events(), vec![]);