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

feat(agents): pont Codex inter-agents + readiness/heartbeat lot 1

Browse Source 
Deux chantiers livrés au vert (workspace entier : domain+application+
infrastructure 42 + app-tauri --lib 128, 0 échec).

## Codex inter-agents
- domaine: McpConfigStrategy::TomlConfigHome { target, home_env } +
  toml_config_home(...); AgentProfile::materializes_idea_bridge()
  (whitelist Claude/ConfigFile + Codex/TomlConfigHome); McpServerWiring
  + encodeur TOML.
- application: lifecycle apply_mcp_config bras TomlConfigHome (écrit
  {runDir}/<target>, pousse (home_env, parent) dans spec.env);
  guard_mcp_bridge_supported ré-exprimée via materializes_idea_bridge();
  catalogue Codex porte toml_config_home(".codex/config.toml","CODEX_HOME").
- app-tauri: is_codex_mcp_profile, migrate_codex_run_dir,
  mcp_server_entry_toml.
- tests: matrice domaine TomlConfigHome + round-trip dual Claude/Codex
  sur loopback réel (fakes, zéro token).

## Readiness/heartbeat lot 1
- domaine: readiness.rs — ReadinessPolicy::classify (Final => TurnEnded),
  variantes ReplyEvent::Heartbeat / ToolActivity.
- application: drain_with_readiness consulte la policy et appelle
  mark_idle sur le signal déterministe; branché dans ask_agent.
  Corrige la cause racine: une cible qui ne renvoie qu'un Final (sans
  idea_reply) débloque désormais sa file Busy.
- infrastructure: adapters de session émettent Heartbeat/ToolActivity.
- tests: drain_with_readiness_lot1 (points QA 5 & 6) verts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-06-14 09:28:44 +02:00
parent fdcf16c387
commit 0f8ba38d51
24 changed files with 2745 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
crates/application/src/agent/catalogue.rs
View File

@ -80,8 +80,11 @@ pub fn reference_profiles() -> Vec<AgentProfile> {
.expect("codex reference profile is valid")
.with_structured_adapter(StructuredAdapter::Codex)
.with_mcp(McpCapability::new(
McpConfigStrategy::config_file(".mcp.json")
.expect(".mcp.json is a valid relative MCP config target"),
// Codex lit ses serveurs MCP dans `$CODEX_HOME/config.toml`, pas `.mcp.json` :
// IdeA écrit ce TOML DANS le run dir et pointe `CODEX_HOME` dessus pour
// isoler l'agent du `~/.codex` global (miroir du `.mcp.json` de Claude).
McpConfigStrategy::toml_config_home(".codex/config.toml", "CODEX_HOME")
.expect(".codex/config.toml + CODEX_HOME is a valid MCP config target"),
McpTransport::Stdio,
)),
AgentProfile::new(
@ -157,18 +160,36 @@ mod mcp_tests {
}
#[test]
fn claude_and_codex_mcp_use_config_file_mcp_json() {
for slug in ["claude", "codex"] {
let mcp = profile(slug).mcp.expect("mcp present");
assert_eq!(
mcp.config,
McpConfigStrategy::ConfigFile {
target: ".mcp.json".to_owned()
},
"profile `{slug}` should declare `.mcp.json`"
);
assert_eq!(mcp.transport, McpTransport::Stdio);
}
fn claude_mcp_uses_config_file_mcp_json() {
let mcp = profile("claude").mcp.expect("mcp present");
assert_eq!(
mcp.config,
McpConfigStrategy::ConfigFile {
target: ".mcp.json".to_owned()
},
"Claude should declare `.mcp.json`"
);
assert_eq!(mcp.transport, McpTransport::Stdio);
}
#[test]
fn codex_mcp_uses_toml_config_home_codex() {
// Codex lit `$CODEX_HOME/config.toml`, pas `.mcp.json` : le seed doit déclarer
// la stratégie TOML isolée par `CODEX_HOME` (pendant Codex de Claude).
let mcp = profile("codex").mcp.expect("mcp present");
assert_eq!(
mcp.config,
McpConfigStrategy::TomlConfigHome {
target: ".codex/config.toml".to_owned(),
home_env: "CODEX_HOME".to_owned(),
},
"Codex should declare `.codex/config.toml` + CODEX_HOME"
);
assert_eq!(mcp.transport, McpTransport::Stdio);
assert!(
profile("codex").materializes_idea_bridge(),
"the Codex seed must materialise the idea bridge"
);
}
#[test]

124
crates/application/src/agent/lifecycle.rs
View File

@ -1746,6 +1746,36 @@ impl LaunchAgent {
},
}
}
domain::profile::McpConfigStrategy::TomlConfigHome { target, home_env } => {
// Pendant Codex de `ConfigFile` : on écrit un `config.toml` (table
// `[mcp_servers.idea]`, via l'encodeur TOML partagé D2) au chemin
// `<run_dir>/<target>`, et on pousse `home_env` (ex. `CODEX_HOME`) vers
// le **dossier parent** de ce fichier. Codex lit alors ses serveurs MCP
// dans ce `config.toml` ISOLÉ au run dir, jamais le `~/.codex` global.
let path = RemotePath::new(join(run_dir, target));
let declaration = mcp_server_wiring(mcp.transport, runtime).to_config_toml();
// Même régime clobber/non-clobber que `.mcp.json` : avec un runtime réel
// (lancement app-tauri) on régénère/clobber à chaque (re)lancement (l'exe
// `$APPIMAGE` et l'endpoint dérivent entre runs) ; sans runtime
// (orchestrateur / hot-swap / tests) on reste non-clobbering pour ne pas
// écraser une déclaration réelle par la minimale.
match runtime {
Some(_) => {
let _ = self.fs.write(&path, declaration.as_bytes()).await;
}
None => match self.fs.exists(&path).await {
Ok(true) => {}
Ok(false) => {
let _ = self.fs.write(&path, declaration.as_bytes()).await;
}
Err(_) => {}
},
}
// `home_env` pointe sur le DOSSIER PARENT de `target` (ex.
// `{runDir}/.codex`), pas sur le fichier — Codex y cherche `config.toml`.
let home_dir = parent_dir(run_dir, target);
spec.env.push((home_env.clone(), home_dir));
}
domain::profile::McpConfigStrategy::Flag { flag } => {
// Pass the server via a launch flag (e.g. `--mcp-config {path}`). The
// config path is the run dir itself (the CLI's cwd), where the server
@ -1858,58 +1888,40 @@ fn mcp_server_declaration(
transport: domain::profile::McpTransport,
runtime: Option<&McpRuntime>,
) -> String {
// Common `.mcp.json`-style shape (Claude Code et CLIs apparentées). The transport
// is surfaced so a socket-based CLI can be wired later without changing this seam.
let transport_label = match transport {
domain::profile::McpTransport::Stdio => "stdio",
domain::profile::McpTransport::Socket => "socket",
};
// `command` + extra args depend on whether OS/runtime facts were injected.
// Each `args` entry is emitted as an escaped JSON string so an exe path or
// endpoint with spaces/backslashes/quotes stays valid JSON.
let (command, extra_args) = match runtime {
Some(rt) => {
let arg = |label: &str, value: &str| {
format!(
",\n {},\n {}",
json_string(label),
json_string(value)
)
};
let extra = format!(
"{}{}{}",
arg("--endpoint", &rt.endpoint),
arg("--project", &rt.project_id),
arg("--requester", &rt.requester),
);
(rt.exe.as_str(), extra)
}
None => ("idea", String::new()),
};
let command = json_string(command);
format!(
r#"{{
"mcpServers": {{
"idea": {{
"command": {command},
"args": [
"mcp-server"{extra_args}
],
"transport": "{transport_label}"
}}
}}
}}
"#
)
mcp_server_wiring(transport, runtime).to_mcp_json()
}
/// Wraps a string as a JSON string literal (quotes + escaping), reusing the path
/// escaper so an exe path / endpoint with spaces, backslashes or quotes stays valid
/// JSON in the generated `.mcp.json`.
fn json_string(s: &str) -> String {
format!("\"{}\"", json_escape(s))
/// Builds the IdeA MCP server **wiring** (`command` + `args` + transport) shared by
/// every materialisation format (cadrage Codex D2). It is the **single source of
/// truth** for *what* the bridge is launched as; the concrete bytes (`.mcp.json`
/// JSON for Claude, `config.toml` TOML for Codex) are produced by the domain
/// encoders [`domain::McpServerWiring::to_mcp_json`] /
/// [`domain::McpServerWiring::to_config_toml`], so the two sites can never drift.
///
/// `Some(runtime)` ⇒ the **real** wiring (this IdeA exe + the project's loopback
/// endpoint/project/requester); `None` ⇒ the coherent **minimal** `idea mcp-server`
/// fallback (orchestrator / hot-swap / tests).
#[must_use]
fn mcp_server_wiring(
transport: domain::profile::McpTransport,
runtime: Option<&McpRuntime>,
) -> domain::McpServerWiring {
let (command, args) = match runtime {
Some(rt) => (
rt.exe.clone(),
vec![
"mcp-server".to_owned(),
"--endpoint".to_owned(),
rt.endpoint.clone(),
"--project".to_owned(),
rt.project_id.clone(),
"--requester".to_owned(),
rt.requester.clone(),
],
),
None => ("idea".to_owned(), vec!["mcp-server".to_owned()]),
};
domain::McpServerWiring::new(command, args, transport)
}
/// Builds an absolute path string by joining a [`ProjectPath`] with a relative
@ -1919,6 +1931,18 @@ fn join(base: &ProjectPath, rel: &str) -> String {
format!("{b}/{rel}")
}
/// Resolves the **parent directory** (absolute) of `<base>/<rel>` — used to point a
/// CLI's `home_env` (e.g. `CODEX_HOME`) at the directory *containing* the materialised
/// config file (`config.toml`), not the file itself. When `rel` has no separator
/// (file directly in the run dir), the parent is the run dir.
fn parent_dir(base: &ProjectPath, rel: &str) -> String {
let full = join(base, rel);
match full.rsplit_once(['/', '\\']) {
Some((parent, _)) if !parent.is_empty() => parent.to_owned(),
_ => base.as_str().trim_end_matches(['/', '\\']).to_owned(),
}
}
/// Computes an agent's isolated run directory `<root>/.ideai/run/<agent-id>/`
/// (ARCHITECTURE §14.1). This is the PTY cwd for the agent — never the project
/// root — guaranteeing that two distinct agents on the same project root get two

2
crates/application/src/agent/mod.rs
View File

@ -16,7 +16,7 @@ mod usecases;
pub(crate) use lifecycle::unique_md_path;
pub(crate) use lifecycle::ReattachDecision;
pub use structured::send_blocking;
pub use structured::{drain_with_readiness, send_blocking};
pub use catalogue::{reference_profile_id, reference_profiles, selectable_reference_profiles};
pub use inspect::{InspectConversation, InspectConversationInput, InspectConversationOutput};

184
crates/application/src/agent/structured.rs
View File

@ -14,7 +14,10 @@
use std::time::Duration;
use domain::input::InputMediator;
use domain::ids::AgentId;
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent};
use domain::readiness::{ReadinessPolicy, ReadinessSignal};
/// Envoie `prompt` à la session vivante puis **draine le flux de réponse jusqu'au
/// [`ReplyEvent::Final`]**, et retourne son contenu agrégé.
@ -39,14 +42,83 @@ pub async fn send_blocking(
session: &dyn AgentSession,
prompt: &str,
timeout: Option<Duration>,
) -> Result<String, AgentSessionError> {
drain_bounded_events(session, prompt, timeout, |_event| {}, |_signal| {}).await
}
/// Comme [`send_blocking`], mais **branche la readiness** : à chaque événement du
/// tour, [`ReadinessPolicy::classify`] est consulté et, dès qu'il renvoie
/// [`ReadinessSignal::TurnEnded`] (le `Final`), le médiateur d'entrée est notifié
/// (`mark_idle(agent)`) pour que la FIFO de l'agent avance — **sans** dépendre d'un
/// `idea_reply` explicite ni du sniff de prompt PTY (chantier readiness/heartbeat,
/// lot 1, fix de la cause racine du blocage `Busy`).
///
/// DRY : **un seul** chemin de lecture du flux (la boucle de [`drain_bounded`]) ;
/// cette fonction n'est que `send_blocking` muni d'un *sink* de readiness. Le `Final`
/// réveille donc à la fois le `pending` (via la valeur de retour) **et** la FIFO (via
/// `mark_idle`). `idea_reply` reste un signal alternatif (premier arrivé gagne) côté
/// orchestrateur.
///
/// # Errors
/// Identiques à [`send_blocking`] (échec `send`/décodage, flux clos sans `Final`,
/// timeout).
pub async fn drain_with_readiness(
session: &dyn AgentSession,
prompt: &str,
timeout: Option<Duration>,
mediator: &dyn InputMediator,
agent: AgentId,
) -> Result<String, AgentSessionError> {
// `on_signal` ne reçoit QUE les événements terminaux (le `Final` ⇒ `TurnEnded`) :
// la readiness ne classe pas les non-terminaux. Pour le **battement** de vivacité
// (lot 2) on a besoin de notifier le médiateur à CHAQUE événement non terminal
// (delta / activité / heartbeat) ⇒ on passe un sink d'événement bruts `on_event`.
drain_bounded_events(
session,
prompt,
timeout,
|event| {
// Tout événement **non terminal** prouve la vivacité ⇒ un battement.
if !matches!(event, ReplyEvent::Final { .. }) {
mediator.mark_alive(agent);
}
},
|signal| {
if signal == ReadinessSignal::TurnEnded {
mediator.mark_idle(agent);
}
},
)
.await
}
/// Ouvre le flux du tour (`send`) et le **draine jusqu'au `Final`**, en appliquant
/// la borne temporelle `timeout`, en notifiant `on_event` à **chaque** événement brut
/// (pour le battement de vivacité, lot 2) et `on_signal` à chaque [`ReadinessSignal`]
/// dérivé par [`ReadinessPolicy`] (le `Final` ⇒ `TurnEnded`).
///
/// **Chemin de lecture unique** (DRY) : `send_blocking` et `drain_with_readiness`
/// passent tous deux par ici, en différant seulement par leurs *sinks*. La session
/// **reste vivante** sur timeout (on ne `shutdown` rien ici, §17.1).
async fn drain_bounded_events(
session: &dyn AgentSession,
prompt: &str,
timeout: Option<Duration>,
on_event: impl FnMut(&ReplyEvent),
on_signal: impl FnMut(ReadinessSignal),
) -> Result<String, AgentSessionError> {
match timeout {
Some(dur) => match tokio::time::timeout(dur, drain_to_final(session, prompt)).await {
Some(dur) => match tokio::time::timeout(
dur,
drain_to_final(session, prompt, on_event, on_signal),
)
.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,
None => drain_to_final(session, prompt, on_event, on_signal).await,
}
}
@ -56,18 +128,124 @@ pub async fn send_blocking(
/// 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`].
///
/// Chaque événement est classé par [`ReadinessPolicy`] et le signal éventuel est
/// remonté à `on_signal` (le `Final` ⇒ [`ReadinessSignal::TurnEnded`]). Deltas,
/// activités et heartbeats sont non terminaux ⇒ ignorés par le rendez-vous synchrone.
async fn drain_to_final(
session: &dyn AgentSession,
prompt: &str,
mut on_event: impl FnMut(&ReplyEvent),
mut on_signal: impl FnMut(ReadinessSignal),
) -> Result<String, AgentSessionError> {
let stream = session.send(prompt).await?;
for event in stream {
// Battement de vivacité (lot 2) : notifié pour CHAQUE événement brut, avant le
// classement readiness. Le sink décide (les non-terminaux prouvent la vivacité).
on_event(&event);
if let Some(signal) = ReadinessPolicy::classify(&event) {
on_signal(signal);
}
if let ReplyEvent::Final { content } = event {
return Ok(content);
}
// TextDelta / ToolActivity : ignorés par le rendez-vous synchrone.
// TextDelta / ToolActivity / Heartbeat : non terminaux, ignorés ici.
}
Err(AgentSessionError::Io(
"le flux de réponse s'est terminé sans événement Final".to_string(),
))
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use domain::ids::SessionId;
use domain::input::{AgentBusyState, SubmitConfig};
use domain::mailbox::{PendingReply, Ticket};
use domain::ports::PtyHandle;
fn agent(n: u128) -> AgentId {
AgentId::from_uuid(uuid::Uuid::from_u128(n))
}
/// Session factice : `send` rejoue une liste fixe d'événements (terminée par un
/// `Final`).
struct FakeSession {
events: Vec<ReplyEvent>,
}
#[async_trait::async_trait]
impl AgentSession for FakeSession {
fn id(&self) -> SessionId {
SessionId::from_uuid(uuid::Uuid::from_u128(1))
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(
&self,
_prompt: &str,
) -> Result<domain::ports::ReplyStream, AgentSessionError> {
Ok(Box::new(self.events.clone().into_iter()))
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
/// Médiateur factice qui enregistre l'ordre des `mark_alive` / `mark_idle`.
#[derive(Default)]
struct RecordingMediator {
calls: Mutex<Vec<&'static str>>,
}
impl InputMediator for RecordingMediator {
fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply {
unreachable!("non utilisé par drain_with_readiness")
}
fn preempt(&self, _agent: AgentId) {}
fn mark_idle(&self, _agent: AgentId) {
self.calls.lock().unwrap().push("idle");
}
fn mark_alive(&self, _agent: AgentId) {
self.calls.lock().unwrap().push("alive");
}
fn busy_state(&self, _agent: AgentId) -> AgentBusyState {
AgentBusyState::Idle
}
fn bind_handle(&self, _agent: AgentId, _handle: PtyHandle) {}
fn bind_handle_with_prompt(
&self,
_agent: AgentId,
_handle: PtyHandle,
_pattern: Option<String>,
_submit: SubmitConfig,
) {
}
}
#[tokio::test]
async fn drain_marks_alive_on_each_non_terminal_then_idle_on_final() {
let session = FakeSession {
events: vec![
ReplyEvent::TextDelta { text: "a".into() },
ReplyEvent::ToolActivity { label: "lit".into() },
ReplyEvent::Heartbeat,
ReplyEvent::Final {
content: "fini".into(),
},
],
};
let mediator = RecordingMediator::default();
let out = drain_with_readiness(&session, "go", None, &mediator, agent(1))
.await
.expect("drain ok");
assert_eq!(out, "fini");
// Trois battements (delta, activité, heartbeat) PUIS l'idle sur le Final.
assert_eq!(
*mediator.calls.lock().unwrap(),
vec!["alive", "alive", "alive", "idle"],
"un battement par événement non terminal, idle au Final (pas de battement sur le Final)"
);
}
}

4
crates/application/src/lib.rs
View File

@ -28,8 +28,8 @@ pub mod terminal;
pub mod window;
pub use agent::{
reference_profile_id, reference_profiles, selectable_reference_profiles, send_blocking,
ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput, ConfigureProfiles,
drain_with_readiness, reference_profile_id, reference_profiles, selectable_reference_profiles,
send_blocking, ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput, ConfigureProfiles,
ConfigureProfilesInput, ConfigureProfilesOutput, CreateAgentFromScratch, CreateAgentInput,
CreateAgentOutput, DeleteAgent, DeleteAgentInput, DeleteProfile, DeleteProfileInput,
DetectProfiles, DetectProfilesInput, DetectProfilesOutput, FirstRunState, FirstRunStateOutput,

354
crates/application/src/orchestrator/service.rs
View File

@ -32,8 +32,9 @@ use domain::{
use crate::conversation::RecordTurn;
use crate::agent::{
CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents,
ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext, UpdateAgentContextInput,
drain_with_readiness, CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput,
ListAgents, ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext,
UpdateAgentContextInput,
};
use crate::error::AppError;
use crate::orchestrator::{
@ -41,7 +42,7 @@ use crate::orchestrator::{
ReadMemoryInput, WriteMemory, WriteMemoryInput,
};
use crate::skill::{CreateSkill, CreateSkillInput};
use crate::terminal::{CloseTerminal, CloseTerminalInput, TerminalSessions};
use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions};
/// Default terminal geometry for an orchestrator-launched agent cell. The UI
/// resizes the PTY to the real cell size on attach; these are sane starting rows
@ -57,7 +58,13 @@ const DEFAULT_COLS: u16 = 80;
/// **without killing the session**, so the requester can retry. Internal and
/// intentionally not yet config-exposed (it may become a per-project setting
/// without changing the contract).
const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(300);
///
/// TEMPORAIRE (demande utilisateur 2026-06-13) : la borne de 300 s coupait des tours
/// délégués légitimement très longs (gros refactors + tests). On la relève donc à 24 h
/// (≈ « pas de limite ») le temps de concevoir un vrai signal de vivacité (savoir si
/// l'agent travaille encore) qui remplacera ce timeout brut. NE PAS considérer comme
/// définitif : à reconvertir en borne courte + heartbeat once that liveness signal exists.
const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(24 * 60 * 60);
/// Borne d'attente **en file** pour acquérir le verrou de tour d'un agent (A0,
/// cadrage v5 §4).
@ -72,7 +79,20 @@ const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(300);
/// [`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);
///
/// TEMPORAIRE (cf. [`ASK_AGENT_TIMEOUT`]) : relevé à 48 h (≈ deux tours « sans limite »)
/// en cohérence avec la levée temporaire de la borne de tour, le temps d'un vrai
/// signal de vivacité.
const ASK_QUEUE_WAIT_CAP: Duration = Duration::from_secs(48 * 60 * 60);
/// Borne de tour effective (lot 2, timeouts pilotés par profil) : le
/// `turn_timeout_ms` du profil de la cible **quand il existe**, sinon le défaut
/// historique [`ASK_AGENT_TIMEOUT`] (zéro régression pour un profil sans `liveness` /
/// sans `turn_timeout_ms`). Pure et testable sans wiring de service.
#[must_use]
fn resolve_turn_timeout(turn_timeout_ms: Option<u32>) -> Duration {
turn_timeout_ms.map_or(ASK_AGENT_TIMEOUT, |ms| Duration::from_millis(u64::from(ms)))
}
/// Fournit les faits OS/runtime (exe + endpoint projet) pour écrire la déclaration MCP
/// réelle quand l'orchestrateur (re)lance une cible sur le chemin `ask`. Implémenté dans
@ -182,6 +202,15 @@ pub struct OrchestratorService {
/// injectée avec [`Self::record_turn`]. La couche `application` reste **pure** : pas
/// de `SystemTime::now()` brut ici, le temps vient du port injecté au composition root.
clock: Option<Arc<dyn Clock>>,
/// Registre des **sessions IA structurées** vivantes (§17.5), pour router un `ask`
/// vers une cible à `structured_adapter` directement sur sa session
/// ([`drain_with_readiness`](crate::agent::drain_with_readiness)) — le `Final`
/// déterministe réveille le `pending` **et** marque l'agent `Idle` (chantier
/// readiness/heartbeat, lot 1, fix du blocage `Busy`). Injecté via
/// [`Self::with_structured`] ; `None` ⇒ on conserve **uniquement** le chemin
/// PTY/mailbox legacy (zéro régression pour les call sites/tests qui ne le branchent
/// pas).
structured: Option<Arc<StructuredSessions>>,
}
/// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés
@ -244,9 +273,19 @@ impl OrchestratorService {
context_guard: None,
record_turn: None,
clock: None,
structured: None,
}
}
/// Branche le registre des [`StructuredSessions`] (§17.5) pour router un `ask`
/// vers une cible **structurée** sur sa propre session. Builder additif (signature
/// de [`Self::new`] inchangée) ; `None` ⇒ chemin PTY/mailbox legacy uniquement.
#[must_use]
pub fn with_structured(mut self, structured: Arc<StructuredSessions>) -> Self {
self.structured = Some(structured);
self
}
/// Branche les use cases C7 (`context.*`/`memory.*`) sous
/// [`domain::fileguard::FileGuard`]. Builder additif (signature de [`Self::new`]
/// inchangée).
@ -750,6 +789,42 @@ impl OrchestratorService {
// Poser l'arête d'attente A→B (retirée en fin de tour par le RAII `_edge`).
let _edge = requester.map(|from| WaitEdgeGuard::new(self, from, agent_id));
// ── Chemin **structuré** (readiness/heartbeat lot 1) ──────────────────────
// Une cible à `structured_adapter` n'a **pas** de PTY : `ensure_live_pty`
// échouerait, et le tour ne pourrait se débloquer que par un `idea_reply`
// explicite (cause racine du blocage `Busy`). Quand le registre structuré est
// câblé et que la cible a une session vivante, on draine son tour via
// `drain_with_readiness` : le `Final` déterministe réveille le `pending` (valeur
// de retour) **et** marque l'agent `Idle` (`mark_idle`). On enregistre tout de
// même un ticket dans la FIFO pour la comptabilité busy et pour préserver
// `idea_reply` comme signal **alternatif** (premier arrivé gagne).
if let Some(structured) = self.structured.as_ref() {
// Lot 1b — auto-lancement d'une cible **froide** structurée : si le registre
// est câblé, qu'aucune session ne vit encore pour la cible, mais que son
// profil porte un `structured_adapter`, on **démarre** sa session via le
// launcher (qui route §17.4 vers `launch_structured` et l'insère dans CE
// même registre, avec la conf MCP matérialisée) plutôt que de tomber dans
// `ensure_live_pty` — chemin PTY qui échouerait pour une cible sans PTY.
// Une cible SANS `structured_adapter` (agent PTY/TUI legacy) conserve le
// chemin `ensure_live_pty` ci-dessous (zéro régression).
if let Some(session) = self
.ensure_structured_session(project, agent_id, &agent.profile_id, structured)
.await?
{
return self
.ask_structured(
project,
agent_id,
&target,
conversation_id,
requester,
task,
session.as_ref(),
)
.await;
}
}
// 1. Garantir la cible vivante en PTY pour CE fil ; lier sa session à la
// conversation, et brancher son handle d'entrée sur le médiateur (livraison).
let handle = self
@ -786,6 +861,10 @@ impl OrchestratorService {
}
None => Ticket::from_human(ticket_id, conversation_id, requester_label, task),
};
// Timeout de tour piloté par profil (lot 2) : `turn_timeout_ms` de la cible si
// défini, sinon le défaut [`ASK_AGENT_TIMEOUT`]. Arme aussi le seuil de stall sur
// le médiateur AVANT l'enqueue (consommé au start_turn).
let turn_timeout = self.turn_timeout_for(project, agent_id).await;
let pending = input.enqueue(agent_id, ticket);
// Delivery is the mediator's responsibility (`InputMediator::enqueue` writes the
// turn into the bound handle). The service no longer writes the PTY directly —
@ -793,7 +872,7 @@ impl OrchestratorService {
// 3. Attendre la réponse, bornée. Timeout/canal fermé ⇒ retirer le ticket
// (cible laissée vivante) et renvoyer une erreur typée.
match tokio::time::timeout(ASK_AGENT_TIMEOUT, pending).await {
match tokio::time::timeout(turn_timeout, pending).await {
Ok(Ok(result)) => {
// Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de
// déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est
@ -821,6 +900,120 @@ impl OrchestratorService {
}
}
/// Chemin `ask` **structuré** (readiness/heartbeat lot 1) : la cible a un
/// `structured_adapter` et une session vivante ([`StructuredSessions`]). On pilote
/// son tour **directement** sur le port [`domain::ports::AgentSession`] et on le
/// draine via [`drain_with_readiness`] : le [`domain::ports::ReplyEvent::Final`]
/// déterministe rend le contenu (réveille le `pending`) **et** marque l'agent `Idle`
/// (`mark_idle`), sans dépendre d'un `idea_reply` explicite ni d'un PTY (que la cible
/// n'a pas). C'est le fix de la cause racine du blocage `Busy`.
///
/// On enregistre tout de même un ticket dans la FIFO (`enqueue`) pour la comptabilité
/// busy et pour **préserver `idea_reply` comme signal alternatif** : on attend la
/// **première** des deux issues (réponse de la session OU résolution du `pending`).
/// Le checkpoint Prompt/Response best-effort est conservé à l'identique du chemin PTY.
#[allow(clippy::too_many_arguments)]
async fn ask_structured(
&self,
project: &Project,
agent_id: AgentId,
target: &str,
conversation_id: domain::conversation::ConversationId,
requester: Option<AgentId>,
task: String,
session: &dyn domain::ports::AgentSession,
) -> Result<OrchestratorOutcome, AppError> {
let (input, mailbox) = match (&self.input, &self.mailbox) {
(Some(i), Some(m)) => (i, m),
_ => return Err(AppError::Invalid(
"la messagerie inter-agents (idea_ask_agent) n'est pas disponible : \
médiateur d'entrée non câblé"
.to_owned(),
)),
};
// Checkpoint Prompt (best-effort), AVANT de déplacer `task` dans le ticket.
let prompt_source = match requester {
Some(from) => InputSource::agent(from),
None => InputSource::Human,
};
self.record_turn_best_effort(
&project.root,
conversation_id,
prompt_source,
TurnRole::Prompt,
task.clone(),
)
.await;
// Ticket dans la FIFO : comptabilité busy + `idea_reply` comme signal alternatif.
let requester_label = self.requester_label(project, requester).await;
let ticket_id = TicketId::new_random();
let ticket = match requester {
Some(from) => Ticket::from_agent(
ticket_id,
from,
conversation_id,
requester_label,
task.clone(),
),
None => Ticket::from_human(ticket_id, conversation_id, requester_label, task.clone()),
};
// Timeout de tour piloté par profil (lot 2) + armement du seuil de stall, AVANT
// l'enqueue qui démarre le tour (le médiateur arme alors sa fenêtre de vivacité).
let turn_timeout = self.turn_timeout_for(project, agent_id).await;
let pending = input.enqueue(agent_id, ticket);
// Drainer le tour structuré (le `Final` ⇒ contenu + `mark_idle`), borné par le
// **même** garde-fou que le chemin PTY (profil ou défaut). On attend la
// **première** issue : le tour structuré OU un `idea_reply` explicite.
let drain = drain_with_readiness(
session,
&task,
Some(turn_timeout),
input.as_ref(),
agent_id,
);
let result = tokio::select! {
biased;
// Issue déterministe : la session a rendu son `Final`.
drained = drain => match drained {
Ok(content) => content,
Err(err) => {
mailbox.cancel_head(agent_id, ticket_id);
return Err(AppError::from(err));
}
},
// Issue alternative : un `idea_reply` explicite a résolu le ticket d'abord.
replied = pending => match replied {
Ok(content) => {
// La session draine encore en arrière-plan ; on s'assure que la FIFO
// avance même si le `Final` n'est pas encore observé.
input.mark_idle(agent_id);
content
}
Err(_cancelled) => {
mailbox.cancel_head(agent_id, ticket_id);
return Err(AppError::Process(format!(
"agent {target} : canal de réponse fermé avant un résultat"
)));
}
},
};
// Checkpoint Response (best-effort), AVANT de déplacer `result`.
self.record_turn_best_effort(
&project.root,
conversation_id,
InputSource::agent(agent_id),
TurnRole::Response,
result.clone(),
)
.await;
Ok(self.reply_outcome(agent_id, target, result))
}
/// `SubmitHumanInput` (cadrage C4 §5.3) — the **human** Envoyer path.
///
/// The operator types into IdeA's mediated input; this resolves the `User↔Agent`
@ -1083,6 +1276,74 @@ impl OrchestratorService {
})
}
/// Lot 1b — garantit une **session structurée vivante** pour la cible d'un `ask`.
///
/// Retourne :
/// - `Ok(Some(session))` si la cible a déjà une session vivante, **ou** si son
/// profil porte un `structured_adapter` et qu'on vient de la (re)lancer ;
/// - `Ok(None)` si la cible n'est **pas** structurée (profil sans `structured_adapter`,
/// ou profil introuvable) ⇒ l'appelant retombe sur le chemin PTY `ensure_live_pty`.
///
/// L'auto-lancement réutilise le **même** [`LaunchAgent`] que le chemin PTV
/// ([`Self::ensure_live_pty`]) avec le **même** `mcp_runtime` matérialisé : pour un
/// profil structuré, le launcher route §17.4 vers `launch_structured`, démarre la
/// session via la fabrique et l'insère dans le registre [`StructuredSessions`]
/// **partagé** (le même `Arc` que `self.structured`, câblé au composition root).
/// La conf MCP est donc matérialisée comme pour une cellule chat lancée à la main,
/// si bien que la cible voit les outils `idea_*` pour répondre.
async fn ensure_structured_session(
&self,
project: &Project,
agent_id: AgentId,
profile_id: &ProfileId,
structured: &Arc<StructuredSessions>,
) -> Result<Option<Arc<dyn domain::ports::AgentSession>>, AppError> {
// Cible déjà chaude : route directe (aucun lancement).
if let Some(session) = structured.session_for_agent(&agent_id) {
return Ok(Some(session));
}
// Cible froide : ne (re)lancer que si le profil sait être piloté en mode
// structuré. Sinon (agent PTY/TUI legacy, ou profil introuvable) ⇒ chemin PTY.
let is_structured = self
.profiles
.list()
.await?
.into_iter()
.find(|p| &p.id == profile_id)
.is_some_and(|p| p.is_selectable());
if !is_structured {
return Ok(None);
}
// Démarrer la session via le launcher (route §17.4 → `launch_structured`,
// insère dans CE registre). Mêmes faits MCP que le chemin PTY pour que le pont
// `idea_*` de la cible se branche. `conversation_id: None` ⇒ le launcher dérive
// l'id de paire (User↔agent) ou réutilise celui de la cellule (P8a), comme pour
// un lancement direct utilisateur.
self.launch_agent
.execute(LaunchAgentInput {
project: project.clone(),
agent_id,
rows: DEFAULT_ROWS,
cols: DEFAULT_COLS,
node_id: None,
conversation_id: None,
mcp_runtime: self
.mcp_runtime_provider
.as_ref()
.and_then(|p| p.runtime_for(project, agent_id)),
})
.await?;
// Le launcher a inséré la session dans le registre partagé : la relire.
structured.session_for_agent(&agent_id).map(Some).ok_or_else(|| {
AppError::Process(format!(
"agent {agent_id} : aucune session structurée vivante après lancement"
))
})
}
/// Binds `conversation` to `session` in both the terminal registry (fast
/// `session_for`) and the [`ConversationRegistry`] (domain thread state), when the
/// latter is wired. Idempotent.
@ -1315,8 +1576,6 @@ impl OrchestratorService {
profile_id: &ProfileId,
target: &str,
) -> Result<(), AppError> {
use domain::profile::{McpConfigStrategy, StructuredAdapter};
let Some(profile) = self
.profiles
.list()
@ -1327,21 +1586,20 @@ impl OrchestratorService {
return Ok(());
};
let honours_mcp_json = matches!(
profile.mcp.as_ref().map(|c| &c.config),
Some(McpConfigStrategy::ConfigFile { target }) if target == ".mcp.json"
);
let is_claude = profile.structured_adapter == Some(StructuredAdapter::Claude);
if honours_mcp_json && is_claude {
// Source de vérité UNIQUE (domaine) : un profil supporte le pont `idea_*` ssi
// IdeA matérialise réellement sa config MCP pour la CLI qu'il pilote — Claude
// via `.mcp.json`, Codex via `config.toml`/`CODEX_HOME`. Tout autre couple
// (y compris MCP absent) ⇒ repli fichier, pont non branché.
if profile.materializes_idea_bridge() {
return Ok(());
}
Err(AppError::Invalid(format!(
"la cible '{target}' (profil '{}', adaptateur {:?}) ne supporte pas encore le \
pont idea_* : la délégation inter-agents passe par un serveur MCP déclaré en \
.mcp.json, que seul un profil Claude consomme aujourd'hui. Cible un agent au \
profil Claude.",
pont idea_* : la délégation inter-agents passe par un serveur MCP qu'IdeA \
matérialise pour la CLI cible, et ce profil ne déclare pas un couple \
(adaptateur × stratégie MCP) pris en charge. Cible un agent dont le profil \
expose le pont MCP (Claude ou Codex).",
profile.name, profile.structured_adapter
)))
}
@ -1383,6 +1641,54 @@ impl OrchestratorService {
(profile.prompt_ready_pattern, submit)
}
/// Résout la [`domain::profile::LivenessStrategy`] du profil de la cible (lot 2) :
/// le seuil de stagnation (`stall_after_ms`) et le timeout de tour
/// (`turn_timeout_ms`). `None`/absent ⇒ `(None, None)` : pas de détection de stall et
/// repli sur les bornes par défaut codées en dur (zéro régression pour un profil sans
/// bloc `liveness`).
async fn liveness_for_agent(
&self,
project: &Project,
agent_id: AgentId,
) -> (Option<u32>, Option<u32>) {
let Some(agent) = self
.list_agents
.execute(ListAgentsInput {
project: project.clone(),
})
.await
.ok()
.and_then(|out| out.agents.into_iter().find(|a| a.id == agent_id))
else {
return (None, None);
};
let Some(profile) = self
.profiles
.list()
.await
.ok()
.and_then(|ps| ps.into_iter().find(|p| p.id == agent.profile_id))
else {
return (None, None);
};
match profile.liveness {
Some(l) => (l.stall_after_ms, l.turn_timeout_ms),
None => (None, None),
}
}
/// Borne de tour effective pour un `ask` : le `turn_timeout_ms` du profil de la cible
/// **quand il existe**, sinon le défaut historique [`ASK_AGENT_TIMEOUT`] (lot 2,
/// timeouts pilotés par profil). Arme aussi le seuil de stagnation sur le médiateur
/// avant l'enqueue (battement/`sweep_stalled`).
async fn turn_timeout_for(&self, project: &Project, agent_id: AgentId) -> Duration {
let (stall_after_ms, turn_timeout_ms) = self.liveness_for_agent(project, agent_id).await;
if let Some(input) = &self.input {
input.set_stall_threshold(agent_id, stall_after_ms);
}
resolve_turn_timeout(turn_timeout_ms)
}
/// Resolves a human-friendly profile reference (slug like `claude-code`,
/// command like `claude`, or display name like `Claude Code`) to a configured
/// [`ProfileId`]. Matching is universal — never hard-coded to one AI — by
@ -1459,6 +1765,18 @@ mod tests {
use domain::profile::{AgentProfile, ContextInjection};
use domain::ProfileId;
// (c) — timeouts pilotés par profil (lot 2) : `turn_timeout_ms` prime sur le défaut.
#[test]
fn profile_turn_timeout_overrides_default() {
// Seuil de profil défini ⇒ il prime sur ASK_AGENT_TIMEOUT.
assert_eq!(
resolve_turn_timeout(Some(120_000)),
Duration::from_millis(120_000)
);
// Absent (profil sans liveness / sans turn_timeout_ms) ⇒ repli sur le défaut.
assert_eq!(resolve_turn_timeout(None), ASK_AGENT_TIMEOUT);
}
fn profile(id: u128, name: &str, command: &str) -> AgentProfile {
AgentProfile::new(
ProfileId::from_uuid(uuid::Uuid::from_u128(id)),

338
crates/application/tests/drain_with_readiness_lot1.rs Normal file
View File

@ -0,0 +1,338 @@
//! Lot 1 (readiness/heartbeat model-agnostique) — tests unitaires QA **indépendants**
//! du helper applicatif `drain_with_readiness`, **100 % fakes**.
//!
//! Couvre les points 5 et 6 du périmètre QA :
//!
//! - **Point 5** : un flux se terminant par `Final` appelle `mark_idle(agent)`
//! **exactement une fois** ; les `Heartbeat`/deltas/activités intermédiaires
//! n'appellent **jamais** `mark_idle`.
//! - **Point 6 (le bug d'origine)** : un agent cible qui **ne fait que renvoyer un
//! `Final`** (sans jamais appeler `idea_reply`) débloque bien la file via
//! `mark_idle` — c'est exactement la cause racine du blocage `Busy`.
//!
//! On teste au niveau `drain_with_readiness` (le rendez-vous synchrone qu'`ask_agent`
//! utilise sur la session structurée d'une cible) avec :
//! - un fake `AgentSession` scriptable (calqué sur `send_blocking_d1.rs`) ;
//! - un fake `InputMediator` qui **compte** les `mark_idle` par agent et journalise
//! l'ordre des appels, sans seconde file ni I/O.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
use std::time::Duration;
use async_trait::async_trait;
use application::drain_with_readiness;
use domain::ids::AgentId;
use domain::input::{AgentBusyState, InputMediator};
use domain::mailbox::{PendingReply, Ticket};
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
use domain::SessionId;
use uuid::Uuid;
fn sid(n: u128) -> SessionId {
SessionId::from_uuid(Uuid::from_u128(n))
}
fn aid(n: u128) -> AgentId {
AgentId::from_uuid(Uuid::from_u128(n))
}
// ---------------------------------------------------------------------------
// Fake AgentSession scriptable (mono-usage), repris de send_blocking_d1.rs
// ---------------------------------------------------------------------------
enum Script {
Stream(Vec<ReplyEvent>),
Err(AgentSessionError),
}
struct ScriptedSession {
id: SessionId,
script: Mutex<Option<Script>>,
shutdowns: AtomicUsize,
}
impl ScriptedSession {
fn new(script: Script) -> Self {
Self {
id: sid(1),
script: Mutex::new(Some(script)),
shutdowns: AtomicUsize::new(0),
}
}
fn shutdown_count(&self) -> usize {
self.shutdowns.load(Ordering::SeqCst)
}
}
#[async_trait]
impl AgentSession for ScriptedSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
let script = self
.script
.lock()
.unwrap()
.take()
.expect("send scripté une seule fois");
match script {
Script::Stream(events) => Ok(Box::new(events.into_iter())),
Script::Err(e) => Err(e),
}
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
self.shutdowns.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
// ---------------------------------------------------------------------------
// Fake InputMediator : compte les mark_idle par agent + journalise les appels.
// ---------------------------------------------------------------------------
struct RecordingMediator {
/// (agent, "mark_idle") dans l'ordre des appels.
calls: Mutex<Vec<(AgentId, &'static str)>>,
}
impl RecordingMediator {
fn new() -> Self {
Self {
calls: Mutex::new(Vec::new()),
}
}
fn mark_idle_count(&self, agent: AgentId) -> usize {
self.calls
.lock()
.unwrap()
.iter()
.filter(|(a, kind)| *a == agent && *kind == "mark_idle")
.count()
}
fn total_calls(&self) -> usize {
self.calls.lock().unwrap().len()
}
}
impl InputMediator for RecordingMediator {
fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply {
// Jamais utilisé par drain_with_readiness ; un future qui ne résout pas.
PendingReply::new(Box::pin(std::future::pending()))
}
fn preempt(&self, agent: AgentId) {
self.calls.lock().unwrap().push((agent, "preempt"));
}
fn mark_idle(&self, agent: AgentId) {
self.calls.lock().unwrap().push((agent, "mark_idle"));
}
fn busy_state(&self, _agent: AgentId) -> AgentBusyState {
AgentBusyState::Idle
}
}
// ---------------------------------------------------------------------------
// Helpers d'événements
// ---------------------------------------------------------------------------
fn delta(t: &str) -> ReplyEvent {
ReplyEvent::TextDelta { text: t.to_owned() }
}
fn tool(l: &str) -> ReplyEvent {
ReplyEvent::ToolActivity {
label: l.to_owned(),
}
}
fn heartbeat() -> ReplyEvent {
ReplyEvent::Heartbeat
}
fn final_(c: &str) -> ReplyEvent {
ReplyEvent::Final {
content: c.to_owned(),
}
}
// ===========================================================================
// Point 6 (LE point qui compte) : un Final SEUL débloque la file via mark_idle,
// sans aucun idea_reply.
// ===========================================================================
#[tokio::test]
async fn final_only_stream_unblocks_queue_via_mark_idle() {
// L'agent cible ne fait QUE renvoyer un Final (jamais d'idea_reply).
let agent = aid(42);
let session = ScriptedSession::new(Script::Stream(vec![final_("done")]));
let mediator = RecordingMediator::new();
let out = drain_with_readiness(&session, "tâche", None, &mediator, agent).await;
assert_eq!(out, Ok("done".to_owned()), "le Final rend bien son contenu");
assert_eq!(
mediator.mark_idle_count(agent),
1,
"le Final déterministe DOIT marquer l'agent Idle (fix de la cause racine du blocage Busy)"
);
// Aucun autre appel parasite, et la session reste vivante (pas de shutdown).
assert_eq!(mediator.total_calls(), 1);
assert_eq!(session.shutdown_count(), 0);
}
// ===========================================================================
// Point 5 : exactement UN mark_idle ; les events intermédiaires n'en émettent pas.
// ===========================================================================
#[tokio::test]
async fn intermediate_events_do_not_mark_idle_only_final_does() {
let agent = aid(7);
// Heartbeats + deltas + activité PUIS le Final.
let session = ScriptedSession::new(Script::Stream(vec![
heartbeat(),
delta("hel"),
tool("lit un fichier"),
heartbeat(),
delta("lo"),
final_("hello"),
]));
let mediator = RecordingMediator::new();
let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
assert_eq!(out, Ok("hello".to_owned()));
assert_eq!(
mediator.mark_idle_count(agent),
1,
"mark_idle exactement UNE fois (au Final), jamais sur heartbeat/delta/activité"
);
assert_eq!(
mediator.total_calls(),
1,
"aucun appel mediator hormis l'unique mark_idle"
);
}
#[tokio::test]
async fn heartbeats_alone_never_mark_idle_before_final() {
// Que des heartbeats avant le Final : aucun ne doit marquer Idle.
let agent = aid(9);
let session = ScriptedSession::new(Script::Stream(vec![
heartbeat(),
heartbeat(),
heartbeat(),
final_("fini"),
]));
let mediator = RecordingMediator::new();
let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
assert_eq!(out, Ok("fini".to_owned()));
assert_eq!(mediator.mark_idle_count(agent), 1);
}
// ===========================================================================
// Bords : pas de Final ⇒ pas de mark_idle ; erreur de send propagée ; mauvais agent.
// ===========================================================================
#[tokio::test]
async fn stream_without_final_does_not_mark_idle_and_is_io_error() {
let agent = aid(3);
let session = ScriptedSession::new(Script::Stream(vec![heartbeat(), delta("a"), tool("b")]));
let mediator = RecordingMediator::new();
let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
assert!(
matches!(out, Err(AgentSessionError::Io(_))),
"flux épuisé sans Final ⇒ Io, obtenu {out:?}"
);
assert_eq!(
mediator.mark_idle_count(agent),
0,
"sans Final, on ne marque JAMAIS Idle (jamais de faux Idle)"
);
assert_eq!(mediator.total_calls(), 0);
}
#[tokio::test]
async fn send_error_is_propagated_and_no_mark_idle() {
let agent = aid(5);
let session =
ScriptedSession::new(Script::Err(AgentSessionError::Decode("bad json".to_owned())));
let mediator = RecordingMediator::new();
let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
assert_eq!(out, Err(AgentSessionError::Decode("bad json".to_owned())));
assert_eq!(mediator.mark_idle_count(agent), 0);
}
#[tokio::test]
async fn mark_idle_targets_the_drained_agent_only() {
// Le mark_idle doit porter sur l'agent passé à drain_with_readiness, pas un autre.
let drained = aid(100);
let other = aid(200);
let session = ScriptedSession::new(Script::Stream(vec![final_("ok")]));
let mediator = RecordingMediator::new();
let _ = drain_with_readiness(&session, "x", None, &mediator, drained).await;
assert_eq!(mediator.mark_idle_count(drained), 1);
assert_eq!(
mediator.mark_idle_count(other),
0,
"mark_idle ne doit cibler que l'agent drainé"
);
}
// ===========================================================================
// Timeout (garde-fou du tour) : Timeout renvoyé, pas de mark_idle, session vivante.
// (Le helper borne via tokio::time::timeout ; on force l'expiration avec un Final
// reporté — réutilise un fake retardé minimal local.)
// ===========================================================================
struct DelayedSession {
delay: Duration,
shutdowns: AtomicUsize,
}
#[async_trait]
impl AgentSession for DelayedSession {
fn id(&self) -> SessionId {
sid(2)
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
tokio::time::sleep(self.delay).await;
Ok(Box::new(vec![final_("too late")].into_iter()))
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
self.shutdowns.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
#[tokio::test]
async fn timeout_returns_timeout_no_mark_idle_session_alive() {
let agent = aid(11);
let session = DelayedSession {
delay: Duration::from_secs(1),
shutdowns: AtomicUsize::new(0),
};
let mediator = RecordingMediator::new();
let out =
drain_with_readiness(&session, "x", Some(Duration::from_millis(20)), &mediator, agent).await;
assert_eq!(out, Err(AgentSessionError::Timeout));
assert_eq!(
mediator.mark_idle_count(agent),
0,
"un timeout ne marque pas Idle (le tour n'a pas rendu son Final)"
);
assert_eq!(
session.shutdowns.load(Ordering::SeqCst),
0,
"timeout ne tue PAS la session (§17.1)"
);
}

254
crates/application/tests/orchestrator_service.rs
View File

@ -2825,3 +2825,257 @@ async fn p6b_failing_record_does_not_degrade_ask() {
"no turns stored when append fails"
);
}
// ===========================================================================
// Lot 1b — auto-lancement d'une cible **froide** structurée sur le chemin `ask`
// ===========================================================================
//
// Avant le Lot 1b, `ask_agent` ne routait vers `ask_structured` que si la cible avait
// **déjà** une session structurée vivante ; une cible structurée froide tombait dans
// `ensure_live_pty` (chemin PTY) — or une cible structurée n'a pas de PTY, donc le tour
// restait bloqué `Busy` (débloquable seulement par un `idea_reply` explicite). Le Lot 1b
// démarre la session structurée via le **même** launcher que le chemin PTY (qui route
// §17.4 vers `launch_structured` et l'insère dans le registre **partagé**), puis route
// vers `ask_structured` : le `Final` déterministe de la session débloque le tour et
// marque l'agent `Idle` (`mark_idle`), **sans** `idea_reply` ni PTY.
use std::sync::atomic::{AtomicUsize, Ordering};
use application::StructuredSessions;
use domain::ports::{
AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream,
};
/// Session structurée fake : `send()` rend un unique [`ReplyEvent::Final`] qui **écho**
/// le prompt reçu — c'est ce `Final` que `drain_with_readiness` transforme en réponse
/// du tour **et** en `mark_idle`. Aucun `idea_reply` n'est nécessaire.
struct EchoSession {
id: SessionId,
}
#[async_trait]
impl AgentSession for EchoSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
let stream: ReplyStream = Box::new(
vec![ReplyEvent::Final {
content: format!("structured: {prompt}"),
}]
.into_iter(),
);
Ok(stream)
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
/// Fabrique fake : `supports` = profil structuré, et `start` **compte** les démarrages
/// (preuve qu'une session froide a bien été auto-lancée) en rendant une [`EchoSession`].
#[derive(Clone, Default)]
struct CountingFactory {
starts: Arc<AtomicUsize>,
next_id: Arc<Mutex<u128>>,
}
impl CountingFactory {
fn new() -> Self {
Self {
starts: Arc::new(AtomicUsize::new(0)),
next_id: Arc::new(Mutex::new(9000)),
}
}
fn start_count(&self) -> usize {
self.starts.load(Ordering::SeqCst)
}
}
#[async_trait]
impl AgentSessionFactory for CountingFactory {
fn supports(&self, profile: &AgentProfile) -> bool {
profile.structured_adapter.is_some()
}
async fn start(
&self,
_profile: &AgentProfile,
_ctx: &PreparedContext,
_cwd: &ProjectPath,
_session: &SessionPlan,
) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
self.starts.fetch_add(1, Ordering::SeqCst);
let id = {
let mut n = self.next_id.lock().unwrap();
let id = SessionId::from_uuid(Uuid::from_u128(*n));
*n += 1;
id
};
Ok(Arc::new(EchoSession { id }))
}
}
/// Tout le câblage `ask` **structuré** : launcher + service voient le **même** registre
/// `StructuredSessions`, et le launcher porte la fabrique (routage §17.4).
struct StructuredAskFixture {
service: Arc<OrchestratorService>,
structured: Arc<StructuredSessions>,
factory: CountingFactory,
pty: FakePty,
}
fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
let structured = Arc::new(StructuredSessions::new());
let factory = CountingFactory::new();
let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()),
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
// Launcher câblé **structuré** : pour un profil à `structured_adapter`, `execute`
// route §17.4 → `launch_structured`, démarre la session via la fabrique et l'insère
// dans CE registre partagé (le même `Arc` que le service ci-dessous).
let launch = Arc::new(
LaunchAgent::new(
Arc::new(contexts.clone()),
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeRuntime),
Arc::new(FakeFs),
Arc::new(pty.clone()),
Arc::new(FakeSkills),
Arc::clone(&sessions),
Arc::new(bus.clone()),
Arc::new(SeqIds::new()),
Arc::new(FakeRecall),
None,
)
.with_structured(
Arc::new(factory.clone()) as Arc<dyn AgentSessionFactory>,
Arc::clone(&structured),
),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
));
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
let service = Arc::new(
OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()))
// Même registre que le launcher : `ask_agent` y relit la session fraîchement
// insérée par `launch_structured`.
.with_structured(Arc::clone(&structured)),
);
StructuredAskFixture {
service,
structured,
factory,
pty,
}
}
/// Cible **froide** structurée (adaptateur Claude, aucune session vivante) : `ask_agent`
/// auto-lance sa session structurée via le launcher (factory.start appelé **une** fois),
/// puis la draine — le `Final` débloque le round-trip **sans** `idea_reply` ni PTY.
#[tokio::test]
async fn ask_cold_structured_target_autolaunches_session_and_final_unblocks() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// Pré-condition : aucune session structurée vivante pour la cible (elle est froide).
assert!(
fx.structured.session_for_agent(&aid(1)).is_none(),
"cible structurée froide : aucune session avant l'ask"
);
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
// Le `Final` de la session a rendu le contenu — sans aucun `idea_reply` dispatché.
assert_eq!(
out.reply.as_deref(),
Some("structured: Analyse §17"),
"le Final de la session auto-lancée débloque le tour"
);
// Exactement **un** démarrage de session structurée (la cible froide a été lancée).
assert_eq!(
fx.factory.start_count(),
1,
"une session structurée a été auto-lancée pour la cible froide"
);
// La session est désormais vivante dans le registre partagé (insérée par le launcher).
assert!(
fx.structured.session_for_agent(&aid(1)).is_some(),
"la session auto-lancée est enregistrée dans StructuredSessions"
);
// Chemin structuré ⇒ **aucun** PTY spawné (la cible n'a pas de terminal).
assert!(
fx.pty.spawns().is_empty(),
"cible structurée : aucun PTY spawné (chemin chat, pas terminal)"
);
}
/// Réutilisation : une **deuxième** délégation vers la **même** cible (désormais chaude)
/// ne relance **pas** de session — elle réutilise celle insérée au premier `ask`. Prouve
/// que le `session_for_agent` court-circuite l'auto-lancement (idempotence du chemin).
#[tokio::test]
async fn ask_warm_structured_target_reuses_session_without_relaunch() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// 1er ask : auto-lance (start_count == 1).
let _ = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("first ask ok");
assert_eq!(fx.factory.start_count(), 1, "1er ask auto-lance la session");
// 2e ask : cible chaude ⇒ aucune relance.
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("second ask ok");
assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
assert_eq!(
fx.factory.start_count(),
1,
"cible chaude : la session est réutilisée, aucune relance"
);
}