feat(agent): fondation exécution structurée des agents IA (D0+D1) — §17

Pivot orchestration : agents IA pilotés via leur mode programmatique/JSON
(capture déterministe), au lieu du TUI brut + self-report. §16 (idea/MCP)
marquée remplacée comme voie principale.

- D0 (domaine) : port AgentSession + AgentSessionFactory, types ReplyEvent
  /ReplyStream/AgentSessionError, champ AgentProfile.structured_adapter
  (Option<StructuredAdapter{Claude,Codex}>, skip si None ⇒ zéro régression),
  catalogue Claude/Codex annotés.
- D1 (application) : registre StructuredSessions (jumeau de TerminalSessions),
  agrégateur LiveSessions{pty,structured} derrière LiveAgentRegistry (vivant si
  PTY OU structuré, surface du trait inchangée), helper send_blocking (draine le
  ReplyStream jusqu'au Final, Timeout sans tuer la session).

Tests : domaine 16+2 ; application registre 11 + send_blocking 9 ; workspace 0 échec.
A/B intacts. Aucun adapter concret (D2), pas de Tauri/front.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-09 17:54:48 +02:00
parent 7375f706da
commit 5e10b5eb42
17 changed files with 2084 additions and 11 deletions

View File

@ -18,7 +18,7 @@
//! making the reference profiles addressable across runs without a registry.
use domain::ids::ProfileId;
use domain::profile::{AgentProfile, ContextInjection};
use domain::profile::{AgentProfile, ContextInjection, StructuredAdapter};
/// A fixed UUID namespace used to derive stable ids for reference profiles.
/// (Random-looking but constant; only its stability matters.)
@ -56,7 +56,8 @@ pub fn reference_profiles() -> Vec<AgentProfile> {
"{agentRunDir}",
None,
)
.expect("claude reference profile is valid"),
.expect("claude reference profile is valid")
.with_structured_adapter(StructuredAdapter::Claude),
AgentProfile::new(
reference_id("codex"),
"OpenAI Codex CLI",
@ -68,7 +69,8 @@ pub fn reference_profiles() -> Vec<AgentProfile> {
"{agentRunDir}",
None,
)
.expect("codex reference profile is valid"),
.expect("codex reference profile is valid")
.with_structured_adapter(StructuredAdapter::Codex),
AgentProfile::new(
reference_id("gemini"),
"Gemini CLI",

View File

@ -10,10 +10,13 @@ mod catalogue;
mod inspect;
mod lifecycle;
mod resume;
mod structured;
mod usecases;
pub(crate) use lifecycle::unique_md_path;
pub use structured::send_blocking;
pub use catalogue::{reference_profile_id, reference_profiles};
pub use inspect::{InspectConversation, InspectConversationInput, InspectConversationOutput};
pub use resume::{

View File

@ -0,0 +1,73 @@
//! Helper applicatif `send_blocking` — le rendez-vous synchrone inter-agents
//! au-dessus du port [`AgentSession`] (ARCHITECTURE §17.1 / §17.4).
//!
//! `AgentSession::send` retourne un **flux** d'événements (`ReplyStream`), à la
//! manière de `PtyPort::subscribe_output`, mais **typé** : deltas de texte →
//! activités d'outil → **un** événement terminal déterministe
//! [`ReplyEvent::Final`]. Le rendez-vous synchrone dont l'orchestrateur a besoin
//! (§17.4) s'obtient en **drainant ce flux jusqu'au `Final`** : c'est *la* primitive
//! de la messagerie inter-agents, **sans outbox, sans corrélation fichier** (le
//! `Final` *est* la fin de tour).
//!
//! DRY : un seul chemin de lecture (le flux). `send_blocking` n'est qu'un *consom-
//! mateur* du même flux que la cellule chat utilise pour le rendu incrémental.
use std::time::Duration;
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent};
/// Envoie `prompt` à la session vivante puis **draine le flux de réponse jusqu'au
/// [`ReplyEvent::Final`]**, et retourne son contenu agrégé.
///
/// C'est le rendez-vous synchrone (§17.4) : on attend que le tour soit
/// déterministiquement terminé (`Final`) avant de rendre la main. Les deltas de
/// texte et les activités d'outil traversés en chemin sont **ignorés** ici (ils
/// servent le rendu incrémental côté UI, pas l'appelant synchrone).
///
/// `timeout`, lorsqu'il est fourni, borne l'attente : si aucun `Final` n'est observé
/// dans le délai, on retourne [`AgentSessionError::Timeout`] **sans tuer la
/// session** (elle reste vivante dans le registre ; l'appelant décide de la suite).
/// `None` ⇒ pas de borne temporelle (on attend la fin du tour).
///
/// # Errors
/// - [`AgentSessionError::Io`]/[`AgentSessionError::Decode`] remontées par `send`
/// (échec de communication / décodage de la sortie structurée) ;
/// - [`AgentSessionError::Io`] si le flux se termine **sans** `Final` (tour
/// interrompu) ;
/// - [`AgentSessionError::Timeout`] si `timeout` expire avant le `Final`.
pub async fn send_blocking(
session: &dyn AgentSession,
prompt: &str,
timeout: Option<Duration>,
) -> Result<String, AgentSessionError> {
match timeout {
Some(dur) => match tokio::time::timeout(dur, drain_to_final(session, prompt)).await {
Ok(result) => result,
// La session **reste vivante** : on ne `shutdown` rien ici (§17.1).
Err(_elapsed) => Err(AgentSessionError::Timeout),
},
None => drain_to_final(session, prompt).await,
}
}
/// Ouvre le flux du tour (`send`) et le **draine jusqu'au `Final`**.
///
/// Le flux ([`domain::ports::ReplyStream`]) est un itérateur synchrone et borné :
/// après le `Final` il ne produit plus rien. On le parcourt donc simplement
/// jusqu'à rencontrer le `Final` (et on retourne son contenu) ; si le flux
/// s'épuise avant, c'est un tour interrompu → erreur [`AgentSessionError::Io`].
async fn drain_to_final(
session: &dyn AgentSession,
prompt: &str,
) -> Result<String, AgentSessionError> {
let stream = session.send(prompt).await?;
for event in stream {
if let ReplyEvent::Final { content } = event {
return Ok(content);
}
// TextDelta / ToolActivity : ignorés par le rendez-vous synchrone.
}
Err(AgentSessionError::Io(
"le flux de réponse s'est terminé sans événement Final".to_string(),
))
}

View File

@ -38,7 +38,7 @@ pub use agent::{
ListResumableAgentsOutput, ProfileAvailability, ReadAgentContext, ReadAgentContextInput,
ReadAgentContextOutput, ReferenceProfiles, ReferenceProfilesOutput, ResumableAgent, SaveProfile,
SaveProfileInput, SaveProfileOutput, UpdateAgentContext, UpdateAgentContextInput,
AGENT_MEMORY_RECALL_BUDGET,
send_blocking, AGENT_MEMORY_RECALL_BUDGET,
};
pub use embedder::{
CheckEmbedderSuggestion, CheckEmbedderSuggestionInput, CheckEmbedderSuggestionOutput,
@ -94,8 +94,8 @@ pub use template::{
UpdateTemplateOutput,
};
pub use terminal::{
CloseTerminal, CloseTerminalInput, CloseTerminalOutput, LiveAgentRegistry, OpenTerminal,
OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, TerminalSessions,
WriteToTerminal, WriteToTerminalInput,
CloseTerminal, CloseTerminalInput, CloseTerminalOutput, LiveAgentRegistry, LiveSessions,
OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput,
StructuredSessions, TerminalSessions, WriteToTerminal, WriteToTerminalInput,
};
pub use window::{MoveTabToNewWindow, MoveTabToNewWindowInput, MoveTabToNewWindowOutput};

View File

@ -28,7 +28,7 @@
mod registry;
mod usecases;
pub use registry::{LiveAgentRegistry, TerminalSessions};
pub use registry::{LiveAgentRegistry, LiveSessions, StructuredSessions, TerminalSessions};
pub use usecases::{
CloseTerminal, CloseTerminalInput, CloseTerminalOutput, OpenTerminal, OpenTerminalInput,
OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, WriteToTerminal, WriteToTerminalInput,

View File

@ -7,9 +7,9 @@
//! application layer rather than the domain or the adapter.
use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::{Arc, Mutex};
use domain::ports::PtyHandle;
use domain::ports::{AgentSession, PtyHandle};
use domain::{AgentId, NodeId, SessionId, SessionKind, TerminalSession};
/// A registered, live terminal: its PTY handle plus the domain snapshot.
@ -204,3 +204,254 @@ impl TerminalSessions {
self.len() == 0
}
}
// ---------------------------------------------------------------------------
// StructuredSessions — le jumeau de TerminalSessions pour les sessions IA (§17.5)
// ---------------------------------------------------------------------------
/// Une session structurée enregistrée : la session vivante plus les coordonnées
/// (agent + cellule hôte) que [`AgentSession`] ne porte pas lui-même.
///
/// `AgentSession` n'expose que `id()`/`conversation_id()` ; comme le snapshot
/// [`TerminalSession`] côté PTY, on associe ici l'`agent_id` (clé de liveness) et
/// le `node_id` (cellule-vue, rebindable) pour offrir la **même** surface que
/// [`TerminalSessions`].
struct StructuredEntry {
/// La session vivante (ressource process/SDK), derrière le port domaine.
session: Arc<dyn AgentSession>,
/// L'agent IA pilotant cette session (invariant « 1 session vivante/agent »).
agent_id: AgentId,
/// La cellule (feuille de layout) qui héberge actuellement la vue.
node_id: NodeId,
}
/// Registre en mémoire des sessions IA structurées vivantes (ARCHITECTURE §17.5).
///
/// **Jumeau de [`TerminalSessions`]** : même rôle (état d'exécution applicatif, pas
/// du modèle métier — cf. les docs de [`TerminalSessions`]), même surface côté
/// liveness/agent (`session_for_agent`, `node_for_agent`, `live_agents`,
/// `rebind_agent_node`, `insert`/`remove`/`session`). La seule différence : il
/// stocke des `Arc<dyn AgentSession>` (sessions programmatiques) au lieu de
/// [`PtyHandle`]/[`TerminalSession`].
///
/// Respecte l'invariant produit **« 1 session vivante par agent »** : la garde
/// d'unicité (généralisée sur les deux registres via [`LiveSessions`]) interroge
/// `session_for_agent` avant tout lancement.
#[derive(Default)]
pub struct StructuredSessions {
entries: Mutex<HashMap<SessionId, StructuredEntry>>,
}
impl LiveAgentRegistry for StructuredSessions {
fn is_agent_live(&self, agent_id: &AgentId) -> bool {
self.session_for_agent(agent_id).is_some()
}
fn is_node_live(&self, node_id: &NodeId) -> bool {
self.entries
.lock()
.map(|m| m.values().any(|e| e.node_id == *node_id))
.unwrap_or(false)
}
}
impl StructuredSessions {
/// Crée un registre vide.
#[must_use]
pub fn new() -> Self {
Self {
entries: Mutex::new(HashMap::new()),
}
}
/// Enregistre une session fraîchement démarrée pour `agent_id`, hébergée par
/// la cellule `node_id`. Clé par l'id de session ([`AgentSession::id`]).
pub fn insert(&self, session: Arc<dyn AgentSession>, agent_id: AgentId, node_id: NodeId) {
if let Ok(mut map) = self.entries.lock() {
let id = session.id();
map.insert(
id,
StructuredEntry {
session,
agent_id,
node_id,
},
);
}
}
/// Retourne la session enregistrée pour un id, si présente.
#[must_use]
pub fn session(&self, id: &SessionId) -> Option<Arc<dyn AgentSession>> {
self.entries
.lock()
.ok()
.and_then(|m| m.get(id).map(|e| Arc::clone(&e.session)))
}
/// Retourne la session vivante hébergeant `agent_id`, si elle existe.
///
/// Jumeau de [`TerminalSessions::session_for_agent`] : **non ambigu** par
/// l'invariant « 1 session vivante/agent » — le premier match est *le* match.
#[must_use]
pub fn session_for_agent(&self, agent_id: &AgentId) -> Option<Arc<dyn AgentSession>> {
self.entries.lock().ok().and_then(|m| {
m.values()
.find(|e| &e.agent_id == agent_id)
.map(|e| Arc::clone(&e.session))
})
}
/// Retourne l'[`SessionId`] de la session vivante hébergeant `agent_id`, si any.
#[must_use]
pub fn session_id_for_agent(&self, agent_id: &AgentId) -> Option<SessionId> {
self.entries.lock().ok().and_then(|m| {
m.values()
.find(|e| &e.agent_id == agent_id)
.map(|e| e.session.id())
})
}
/// Retourne le [`NodeId`] de la cellule vivante hébergeant `agent_id`, si any.
///
/// Jumeau de [`TerminalSessions::node_for_agent`].
#[must_use]
pub fn node_for_agent(&self, agent_id: &AgentId) -> Option<NodeId> {
self.entries.lock().ok().and_then(|m| {
m.values()
.find(|e| &e.agent_id == agent_id)
.map(|e| e.node_id)
})
}
/// Liste chaque agent IA vivant, sa cellule hôte et son id de session.
///
/// Jumeau de [`TerminalSessions::live_agents`] : un tuple
/// `(AgentId, NodeId, SessionId)` par session structurée vivante.
#[must_use]
pub fn live_agents(&self) -> Vec<(AgentId, NodeId, SessionId)> {
self.entries
.lock()
.map(|m| {
m.values()
.map(|e| (e.agent_id, e.node_id, e.session.id()))
.collect()
})
.unwrap_or_default()
}
/// Rebinde la session vivante d'un agent vers une nouvelle cellule-vue sans
/// redémarrer la conversation (« la cellule est une vue », §17.6).
///
/// Jumeau de [`TerminalSessions::rebind_agent_node`] : seul le `node_id`
/// change ; la session, son id et sa conversation restent intacts. Retourne la
/// session rebindée, ou `None` si l'agent n'a pas de session vivante.
#[must_use]
pub fn rebind_agent_node(
&self,
agent_id: &AgentId,
node_id: NodeId,
) -> Option<Arc<dyn AgentSession>> {
self.entries.lock().ok().and_then(|mut m| {
let entry = m.values_mut().find(|e| &e.agent_id == agent_id)?;
entry.node_id = node_id;
Some(Arc::clone(&entry.session))
})
}
/// Retire une session du registre, retournant la session si présente (pour que
/// l'appelant la `shutdown` hors du verrou).
pub fn remove(&self, id: &SessionId) -> Option<Arc<dyn AgentSession>> {
self.entries
.lock()
.ok()
.and_then(|mut m| m.remove(id).map(|e| e.session))
}
/// Retourne toutes les sessions vivantes (pour un arrêt global propre au
/// shutdown applicatif, jumeau de [`TerminalSessions::handles`]).
#[must_use]
pub fn sessions(&self) -> Vec<Arc<dyn AgentSession>> {
self.entries
.lock()
.map(|m| m.values().map(|e| Arc::clone(&e.session)).collect())
.unwrap_or_default()
}
/// Nombre de sessions structurées vivantes.
#[must_use]
pub fn len(&self) -> usize {
self.entries.lock().map(|m| m.len()).unwrap_or(0)
}
/// Si le registre est vide.
#[must_use]
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
// ---------------------------------------------------------------------------
// LiveSessions — agrégateur des deux registres derrière LiveAgentRegistry (§17.5)
// ---------------------------------------------------------------------------
/// Agrégateur de liveness sur **les deux** registres (PTY + structuré).
///
/// Un agent terminal brut vit dans [`TerminalSessions`] ; un agent IA structuré vit
/// dans [`StructuredSessions`]. La garde d'unicité et l'orchestrateur (§17.4)
/// dépendent de cet agrégateur (ISP : ils ne voient que la capacité « liveness +
/// résolution »), et une requête de liveness/agent voit donc les deux registres :
/// un agent est vivant s'il a une session vivante dans **l'un OU l'autre**.
///
/// Implémente [`LiveAgentRegistry`] : le trait existant n'est **pas modifié** (ses
/// implémenteurs et appelants actuels — `TerminalSessions`, le snapshot — restent
/// inchangés), il est simplement **réalisé par un troisième implémenteur** qui
/// agrège, ce qui *généralise* sa portée à l'ensemble PTY+structuré sans régression.
pub struct LiveSessions {
/// Registre des sessions terminal brut (PTY).
pub pty: Arc<TerminalSessions>,
/// Registre des sessions IA structurées.
pub structured: Arc<StructuredSessions>,
}
impl LiveSessions {
/// Construit l'agrégateur à partir des deux registres partagés.
#[must_use]
pub fn new(pty: Arc<TerminalSessions>, structured: Arc<StructuredSessions>) -> Self {
Self { pty, structured }
}
/// L'[`SessionId`] de la session vivante d'un agent, PTY **ou** structurée.
#[must_use]
pub fn session_id_for_agent(&self, agent_id: &AgentId) -> Option<SessionId> {
self.pty
.session_for_agent(agent_id)
.or_else(|| self.structured.session_id_for_agent(agent_id))
}
/// La cellule hôte de la session vivante d'un agent, PTY **ou** structurée.
#[must_use]
pub fn node_for_agent(&self, agent_id: &AgentId) -> Option<NodeId> {
self.pty
.node_for_agent(agent_id)
.or_else(|| self.structured.node_for_agent(agent_id))
}
/// Tous les agents vivants des deux registres (PTY puis structurés).
#[must_use]
pub fn live_agents(&self) -> Vec<(AgentId, NodeId, SessionId)> {
let mut all = self.pty.live_agents();
all.extend(self.structured.live_agents());
all
}
}
impl LiveAgentRegistry for LiveSessions {
fn is_agent_live(&self, agent_id: &AgentId) -> bool {
self.pty.is_agent_live(agent_id) || self.structured.is_agent_live(agent_id)
}
fn is_node_live(&self, node_id: &NodeId) -> bool {
self.pty.is_node_live(node_id) || self.structured.is_node_live(node_id)
}
}