feat(agent): robustesse routage ask — fix registre session + concurrence (R0+A0) — §14.3/§15
Stabilise le routage de `ask` avant le bind transport MCP (v5). Invariant « 1 agent = 1 employé » durci ; un agent traite un tour à la fois. - R0a garde LaunchAgent : lève AgentAlreadyRunning pour un lancement neuf ciblant un agent déjà vivant sur un autre node (PTY + structuré) ; rebind seulement même-node ou réattache explicite (conversation_id). Idem spawn_agent. - R0b list_live_agents agrège PTY + structuré (LiveSessions) + dédup. - R0c réconciliation des layouts.json à doublons à l'ouverture (host déterministe, idempotent) — corrige « une cellule reset au retour d'onglet ». - R0d UI : option agent désactivée si vivant ailleurs + « aller à la cellule », mapping AGENT_ALREADY_RUNNING. - A0 sérialisation FIFO des tours par agent_id dans ask_agent (verrou tokio par agent ; agents différents en parallèle ; timeout tour 300s, cap attente 600s). Cadrage : .ideai/briefs/orchestration-v5-transport-bind-cadrage.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -13,15 +13,18 @@
|
||||
//! no filesystem watching, no JSON, no process spawning here — those are the
|
||||
//! infrastructure adapter's job. That keeps this fully unit-testable with fakes.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex as StdMutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::sync::Mutex as AsyncMutex;
|
||||
|
||||
use domain::ports::{EventBus, ProfileStore};
|
||||
use domain::{DomainEvent, OrchestratorCommand, OrchestratorVisibility, ProfileId, Project};
|
||||
use domain::{AgentId, DomainEvent, OrchestratorCommand, OrchestratorVisibility, ProfileId, Project};
|
||||
|
||||
use crate::agent::{
|
||||
send_blocking, CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput,
|
||||
ListAgents, ListAgentsInput, UpdateAgentContext, UpdateAgentContextInput,
|
||||
ListAgents, ListAgentsInput, ReattachDecision, UpdateAgentContext, UpdateAgentContextInput,
|
||||
};
|
||||
use crate::error::AppError;
|
||||
use crate::skill::{CreateSkill, CreateSkillInput};
|
||||
@ -43,6 +46,21 @@ const DEFAULT_COLS: u16 = 80;
|
||||
/// without changing the contract).
|
||||
const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(300);
|
||||
|
||||
/// Borne d'attente **en file** pour acquérir le verrou de tour d'un agent (A0,
|
||||
/// cadrage v5 §4).
|
||||
///
|
||||
/// La sérialisation FIFO par agent (« 1 agent = 1 employé : un tour à la fois »)
|
||||
/// fait qu'un `ask` concurrent vers la **même** cible patiente derrière le tour en
|
||||
/// cours. Le timeout de tour ([`ASK_AGENT_TIMEOUT`]) borne **le tour lui-même**,
|
||||
/// **pas** cette attente : on lui donne donc son propre plafond, généreux mais fini,
|
||||
/// pour qu'un `ask` ne reste jamais bloqué indéfiniment si la file est longue
|
||||
/// (inanition). À l'expiration, l'`ask` renvoie un timeout typé (réutilise le
|
||||
/// **même** type que le timeout de tour, cf. [`AppError::Process`] via
|
||||
/// [`domain::ports::AgentSessionError::Timeout`]) ; le tour en cours n'est pas
|
||||
/// affecté. Largeur = un tour complet + sa propre file ⇒ on autorise deux tours
|
||||
/// pleins d'attente.
|
||||
const ASK_QUEUE_WAIT_CAP: Duration = Duration::from_secs(600);
|
||||
|
||||
/// Dispatches validated orchestrator commands to the agent/terminal use cases.
|
||||
pub struct OrchestratorService {
|
||||
create_agent: Arc<CreateAgentFromScratch>,
|
||||
@ -62,6 +80,25 @@ pub struct OrchestratorService {
|
||||
/// `ask` réussi (§17.4). Injecté via [`Self::with_events`] ; `None` ⇒ pas de
|
||||
/// publication (l'`ask` fonctionne quand même).
|
||||
events: Option<Arc<dyn EventBus>>,
|
||||
/// Verrous de **tour par agent** (A0, cadrage v5 §4) — sérialisation FIFO des
|
||||
/// `ask` vers une **même** cible : « 1 agent = 1 employé, un tour à la fois ».
|
||||
///
|
||||
/// Clé = `AgentId` de la **cible** ; valeur = un `tokio::Mutex` d'unité dont le
|
||||
/// garde est tenu **à travers** le `.await` de `send_blocking` (d'où le mutex
|
||||
/// **async**, pas `std`). La `HashMap` elle-même n'est tenue que le temps du
|
||||
/// get-or-create (jamais à travers un `.await`), donc protégée par un mutex
|
||||
/// **synchrone** `std` — pas de garde gardé en travers d'un point de suspension.
|
||||
///
|
||||
/// Le registre vit **ici** (règle applicative d'orchestration, frontière
|
||||
/// hexagonale, cadrage v5 §5) et **non** dans [`StructuredSessions`] : sérialiser
|
||||
/// les tours est une responsabilité de l'orchestrateur, pas du stockage de
|
||||
/// sessions (SRP) ; le couplage reste minimal. Verrou **par agent** ⇒ deux `ask`
|
||||
/// vers des agents **différents** n'entrent jamais en contention (un map d'`Arc`
|
||||
/// distincts).
|
||||
///
|
||||
/// Croissance bornée en pratique au nombre d'agents du projet ; une entrée
|
||||
/// morte ne coûte qu'un `Arc<Mutex<()>>` vide (pas de session, pas de process).
|
||||
ask_locks: StdMutex<HashMap<AgentId, Arc<AsyncMutex<()>>>>,
|
||||
}
|
||||
|
||||
/// Outcome of dispatching a command — a short, human-readable success summary the
|
||||
@ -102,9 +139,24 @@ impl OrchestratorService {
|
||||
sessions,
|
||||
structured: None,
|
||||
events: None,
|
||||
ask_locks: StdMutex::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the per-agent **turn lock**, creating it on first use.
|
||||
///
|
||||
/// Get-or-create under the synchronous map mutex (held only for this lookup,
|
||||
/// never across an `.await`); the returned `Arc<AsyncMutex<()>>` is the lock the
|
||||
/// caller acquires (and holds across `send_blocking`) to serialise turns for
|
||||
/// `agent_id`. Two different agents get two distinct `Arc`s ⇒ no contention.
|
||||
fn ask_lock_for(&self, agent_id: &AgentId) -> Arc<AsyncMutex<()>> {
|
||||
let mut locks = self
|
||||
.ask_locks
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
Arc::clone(locks.entry(*agent_id).or_default())
|
||||
}
|
||||
|
||||
/// Branche le registre des sessions **structurées** (§17.5) pour servir
|
||||
/// `agent.message`/[`OrchestratorCommand::AskAgent`]. Builder additif façon D3 :
|
||||
/// signature de [`Self::new`] **inchangée** (les tests/call sites legacy restent
|
||||
@ -202,18 +254,43 @@ impl OrchestratorService {
|
||||
});
|
||||
}
|
||||
OrchestratorVisibility::Visible { node_id } => {
|
||||
let session = self
|
||||
.sessions
|
||||
.rebind_agent_node(&agent_id, node_id)
|
||||
.ok_or_else(|| {
|
||||
AppError::NotFound(format!(
|
||||
"running session {session_id} for agent {name}"
|
||||
))
|
||||
})?;
|
||||
return Ok(OrchestratorOutcome {
|
||||
detail: format!("attached agent {name} to cell {}", session.node_id),
|
||||
reply: None,
|
||||
});
|
||||
// R0a — même règle que le garde de `LaunchAgent` (cadrage v5 §3.2,
|
||||
// Trou A) : un `spawn` est un **lancement neuf** (pas de
|
||||
// conversation portée), donc ré-attacher à la **même** cellule
|
||||
// hôte est légitime (rebind de vue), mais viser un **autre** node
|
||||
// pour un agent singleton déjà vivant est un second lancement ⇒
|
||||
// refus `AgentAlreadyRunning`.
|
||||
let host_node = self.sessions.node_for_agent(&agent_id);
|
||||
match ReattachDecision::resolve(Some(node_id), host_node, None) {
|
||||
ReattachDecision::Rebind { node_id } => {
|
||||
let session = self
|
||||
.sessions
|
||||
.rebind_agent_node(&agent_id, node_id)
|
||||
.ok_or_else(|| {
|
||||
AppError::NotFound(format!(
|
||||
"running session {session_id} for agent {name}"
|
||||
))
|
||||
})?;
|
||||
return Ok(OrchestratorOutcome {
|
||||
detail: format!(
|
||||
"attached agent {name} to cell {}",
|
||||
session.node_id
|
||||
),
|
||||
reply: None,
|
||||
});
|
||||
}
|
||||
ReattachDecision::Refuse { node_id } => {
|
||||
return Err(AppError::AgentAlreadyRunning { agent_id, node_id });
|
||||
}
|
||||
// A `Visible` spawn always carries a node, so the decision is
|
||||
// never `Idempotent`; treat it as a same-node rebind for safety.
|
||||
ReattachDecision::Idempotent => {
|
||||
return Ok(OrchestratorOutcome {
|
||||
detail: format!("agent {name} already running"),
|
||||
reply: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -287,6 +364,29 @@ impl OrchestratorService {
|
||||
.await?
|
||||
.ok_or_else(|| AppError::NotFound(format!("agent {target}")))?;
|
||||
|
||||
// Sérialisation FIFO **par agent** (A0, cadrage v5 §4) : on acquiert le verrou
|
||||
// de tour de la **cible** avant tout `send_blocking`, et on le tient pour TOUT
|
||||
// le tour réel (rendez-vous direct *ou* lancement-puis-envoi sur cible morte).
|
||||
// Le garde `_turn` est RAII : il tombe en fin de portée, y compris sur chaque
|
||||
// early-return d'erreur ci-dessous ⇒ le tour suivant en file démarre alors.
|
||||
//
|
||||
// Verrou par `agent_id` ⇒ un `ask` vers A et un vers B ne se bloquent jamais ;
|
||||
// un seul verrou tenu (celui de la cible) ⇒ pas d'inter-verrouillage/deadlock.
|
||||
// Le timeout de TOUR ([`ASK_AGENT_TIMEOUT`]) borne `send_blocking`, **pas**
|
||||
// l'attente du verrou ; cette attente a son **propre** plafond
|
||||
// ([`ASK_QUEUE_WAIT_CAP`]) pour éviter l'inanition.
|
||||
let lock = self.ask_lock_for(&agent_id);
|
||||
let _turn = match tokio::time::timeout(ASK_QUEUE_WAIT_CAP, lock.lock_owned()).await {
|
||||
Ok(guard) => guard,
|
||||
Err(_elapsed) => {
|
||||
// Réutilise le **même** type de timeout que le tour (cadrage v5 §4) :
|
||||
// `AgentSessionError::Timeout` ⇒ `AppError::Process`.
|
||||
return Err(AppError::from(
|
||||
domain::ports::AgentSessionError::Timeout,
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
// 1. Session structurée déjà vivante ? ⇒ rendez-vous direct.
|
||||
if let Some(session) = structured.session_for_agent(&agent_id) {
|
||||
let content = send_blocking(session.as_ref(), &task, Some(ASK_AGENT_TIMEOUT)).await?;
|
||||
|
||||
Reference in New Issue
Block a user