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

agent conversation fix

Browse Source
This commit is contained in:
Blomios
2026-06-27 12:42:37 +02:00
parent c2d1a669c5
commit 1fc7869160
38 changed files with 1173 additions and 1549 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2732,30 +2732,25 @@ pub(crate) fn compose_convention_file(
out.push_str("---\n\n");
out.push_str("# Orchestration IdeA\n\n");
if mcp_enabled {
// Surface MCP (cadrage v3, D3) : l'alternative native aux subagents est
// exposée comme outils typés `idea_*`. L'interdiction des subagents natifs
// du fournisseur est **conservée** ; seule la voie de délégation change.
// Surface MCP : outils IdeA natifs. `idea_ask_agent` est une porte d'entrée
// ergonomique vers l'orchestrateur headless, pas un protocole ask/reply exposé
// aux modèles.
out.push_str(
"Pour déléguer une tâche à un autre agent, n'utilise jamais les subagents \
natifs du fournisseur IA. Utilise les outils IdeA natifs : \
`idea_ask_agent` (déléguer une tâche et recevoir la réponse), \
`idea_launch_agent` (lancer/réattacher un agent) et `idea_list_agents` \
(lister les agents du projet). IdeA lancera ou réattachera l'agent cible \
avec son propre AI Profile, son contexte et sa mémoire.\n\n",
"La conversation inter-agent est gérée par IdeA en headless. Pour contacter \
un autre agent, utilise l'outil MCP IdeA `idea_ask_agent(target, task)` quand \
il est disponible : IdeA lance ou réattache la cible via son profil \
structured/headless, capture sa réponse finale et te la renvoie inline. \
N'invente pas de protocole de ticket et n'appelle pas d'outil de remise de \
résultat.\n\n",
);
// Protocole de délégation (Option 1, B-5) : côté agent SOLLICITÉ. Une tâche
// déléguée arrive dans ton terminal préfixée `[IdeA · tâche de … · ticket …]`.
// Tu DOIS y répondre via l'outil `idea_reply`, jamais en texte libre — sinon
// l'agent qui t'a sollicité reste bloqué (sa réponse ne lui parviendra pas).
// Conversation inter-agent : le transport est géré par IdeA en headless. Les
// agents n'ont plus à connaître un protocole de reply ni à corréler des tickets.
out.push_str(
"Quand tu reçois une tâche déléguée par IdeA (un message préfixé \
`[IdeA · tâche de … · ticket …]`), traite-la puis appelle \
**impérativement** l'outil `idea_reply(ticket=…, result=…)` pour rendre ton \
résultat — **même pour une réponse triviale** (un simple `pong`). Ne réponds \
**JAMAIS** uniquement en texte/prose dans le terminal : tant que tu n'as pas \
appelé `idea_reply`, l'agent qui t'a sollicité reste **bloqué** et sa réponse \
ne lui parviendra pas. Terminer ton tour sans `idea_reply` est un échec du \
protocole, pas une réponse.\n\n",
"Quand IdeA te sollicite via une conversation inter-agent headless, traite \
simplement la demande et termine ton tour avec ta réponse normale. N'ajoute \
pas de protocole de ticket, n'appelle pas d'outil de remise de résultat : IdeA \
capture automatiquement la réponse finale du modèle et la transmet à l'agent \
demandeur.\n\n",
);
// Capacités IdeA (feature skill-awareness) : briefing à HAUTE ALTITUDE,
// télégraphique, injecté à CHAQUE lancement (coût token → pas d'exemples ni de
@ -3379,22 +3374,18 @@ mod tests {
#[test]
fn compose_convention_file_mcp_prose_points_to_idea_tools_and_keeps_subagent_ban() {
// mcp_enabled = true ⇒ prose exposes the native `idea_*` tools while keeping
// the ban on the provider's native subagents (cadrage v3, Décision 3).
// mcp_enabled = true ⇒ prose exposes the remaining native `idea_*` tools while
// keeping the ban on the provider's native subagents (cadrage v3, Décision 3).
let doc = compose_convention_file("/root", "", "# Persona", &[], &[], None, &[], true);
// Native IdeA orchestration tools surfaced.
assert!(
doc.contains("idea_ask_agent"),
"MCP prose must mention idea_ask_agent"
);
assert!(doc.contains("idea_launch_agent"));
assert!(doc.contains("idea_list_agents"));
// Native-subagent ban preserved.
assert!(
doc.contains("n'utilise jamais les subagents"),
"MCP prose must keep the native-subagent ban"
doc.contains("idea_context_read"),
"MCP prose must mention remaining IdeA tools"
);
assert!(doc.contains("idea_memory_read"));
assert!(doc.contains("idea_ask_agent"));
assert!(!doc.contains("idea_reply"));
// It does NOT fall back to the file protocol wording.
assert!(
!doc.contains(".ideai/requests"),
@ -3403,20 +3394,23 @@ mod tests {
}
#[test]
fn compose_convention_file_mcp_prose_carries_the_idea_reply_delegation_protocol() {
// B-5 — the solicited-agent side of the protocol: a delegated task arrives as
// `[IdeA · tâche …]` and MUST be answered via `idea_reply`, never plain text.
fn compose_convention_file_mcp_prose_does_not_carry_idea_reply_protocol() {
// Inter-agent conversation is headless: the model's final answer is captured by
// IdeA. Agents must not be instructed to call a reply tool or echo tickets.
let doc = compose_convention_file("/root", "", "# Persona", &[], &[], None, &[], true);
assert!(
doc.contains("idea_reply"),
"MCP prose must instruct answering via idea_reply"
!doc.contains("idea_reply"),
"MCP prose must not instruct answering via idea_reply"
);
assert!(
doc.contains("[IdeA · tâche"),
"MCP prose must describe the delegated-task prefix it answers to"
doc.contains("capture automatiquement la réponse finale")
|| doc.contains("capture sa réponse finale"),
"MCP prose must describe final-answer capture"
);
assert!(
!doc.contains("[IdeA · tâche"),
"MCP prose must not describe the old delegated-task prefix"
);
// Negative: the non-MCP (file-protocol) prose must NOT carry the idea_reply
// instruction (zero regression on the file path).
let file_doc =
compose_convention_file("/root", "", "# Persona", &[], &[], None, &[], false);
assert!(!file_doc.contains("idea_reply"));

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

@ -23,8 +23,7 @@ pub use structured::{
};
pub use catalogue::{
reference_profile_id, reference_profiles,
selectable_reference_profiles, CODEX_SUBMIT_DELAY_MS,
reference_profile_id, reference_profiles, selectable_reference_profiles, CODEX_SUBMIT_DELAY_MS,
};
pub use inspect::{InspectConversation, InspectConversationInput, InspectConversationOutput};
pub use lifecycle::{

20
crates/application/src/orchestrator/rendezvous.rs
View File

@ -284,11 +284,23 @@ mod tests {
/// `Some(0)` ⇒ "pas d'override" (jamais d'effondrement instantané).
#[test]
fn resolvers_apply_override_else_finite_default() {
assert_eq!(resolve_rendezvous_window(Some(1234)), Duration::from_millis(1234));
assert_eq!(resolve_rendezvous_window(Some(0)), DEFAULT_RENDEZVOUS_WINDOW);
assert_eq!(
resolve_rendezvous_window(Some(1234)),
Duration::from_millis(1234)
);
assert_eq!(
resolve_rendezvous_window(Some(0)),
DEFAULT_RENDEZVOUS_WINDOW
);
assert_eq!(resolve_rendezvous_window(None), DEFAULT_RENDEZVOUS_WINDOW);
assert_eq!(resolve_rendezvous_ceiling(Some(9999)), Duration::from_millis(9999));
assert_eq!(resolve_rendezvous_ceiling(Some(0)), DEFAULT_RENDEZVOUS_CEILING);
assert_eq!(
resolve_rendezvous_ceiling(Some(9999)),
Duration::from_millis(9999)
);
assert_eq!(
resolve_rendezvous_ceiling(Some(0)),
DEFAULT_RENDEZVOUS_CEILING
);
assert_eq!(resolve_rendezvous_ceiling(None), DEFAULT_RENDEZVOUS_CEILING);
}
}

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

@ -22,7 +22,7 @@ use tokio::sync::Mutex as AsyncMutex;
use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef, WaitForGraph};
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
use domain::input::{InputMediator, InputSource, SubmitConfig};
use domain::mailbox::{Ticket, TicketId, TurnResolution};
use domain::mailbox::{Ticket, TicketId};
use domain::ports::{Clock, EventBus, ProfileStore, PtyHandle};
use domain::project::ProjectPath;
use domain::{
@ -315,6 +315,11 @@ pub struct OrchestratorService {
/// `ask` A→B, retirée au reply/timeout (RAII via le garde de tour). Sert à
/// **refuser** une délégation ré-entrante (A→B→…→A) avant deadlock.
wait_for: StdMutex<WaitForGraph>,
/// Snapshot opérationnel des mêmes arêtes d'attente, exposé au chemin d'arrêt
/// utilisateur : si A est stoppé pendant qu'il attend B, IdeA stoppe aussi B.
/// Séparé de [`WaitForGraph`] qui reste un objet domaine minimal de détection de
/// cycle, sans API de traversal.
active_waits: StdMutex<Vec<(AgentId, AgentId)>>,
/// Bus d'événements pour publier [`DomainEvent::AgentReplied`] à l'issue d'un
/// `ask` réussi (§17.4). Injecté via [`Self::with_events`] ; `None` ⇒ pas de
/// publication (l'`ask` fonctionne quand même).
@ -459,6 +464,7 @@ impl OrchestratorService {
mailbox: None,
conversations: None,
wait_for: StdMutex::new(WaitForGraph::new()),
active_waits: StdMutex::new(Vec::new()),
events: None,
ask_locks: StdMutex::new(HashMap::new()),
mcp_runtime_provider: None,
@ -537,6 +543,46 @@ impl OrchestratorService {
Arc::clone(locks.entry(*agent_id).or_default())
}
/// Returns the transitive set of agents currently waited on by `agent`.
///
/// Used by user-driven cancellation: stopping A while A waits on B should also
/// stop B, and then any agent B itself waits on. The snapshot is best-effort and
/// lock-bounded; callers perform the actual interruption/stop outside the mutex.
#[must_use]
pub fn active_wait_dependencies(&self, agent: AgentId) -> Vec<AgentId> {
let edges = self
.active_waits
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
Self::active_wait_dependencies_from_edges(&edges, agent)
}
/// Returns the transitive dependencies of `agent` in an active wait-edge snapshot.
#[must_use]
fn active_wait_dependencies_from_edges(
edges: &[(AgentId, AgentId)],
agent: AgentId,
) -> Vec<AgentId> {
let mut out = Vec::new();
let mut stack: Vec<AgentId> = edges
.iter()
.filter_map(|(from, to)| (*from == agent).then_some(*to))
.collect();
while let Some(next) = stack.pop() {
if out.contains(&next) {
continue;
}
out.push(next);
stack.extend(
edges
.iter()
.filter_map(|(from, to)| (*from == next).then_some(*to)),
);
}
out
}
/// Branche le **médiateur d'entrée** (cadrage C3 §5.2) pour servir
/// `agent.message`/[`OrchestratorCommand::AskAgent`] et
/// `agent.reply`/[`OrchestratorCommand::Reply`]. Le `mailbox` est le moteur de
@ -1203,32 +1249,28 @@ impl OrchestratorService {
})
}
/// `agent.message` / `idea_ask_agent`: the **inter-agent delegation rendezvous**
/// (Option 1 « Terminal + MCP », lot B-3).
/// `agent.message` / `idea_ask_agent`: the **inter-agent delegation rendezvous**.
///
/// The target's human-facing view is now a **raw native terminal** (PTY REPL), and
/// delegation flows through the terminal's single FIFO input plus the MCP mailbox:
/// The MCP tool and the file watcher both enter here, but they are only entrypoints
/// into the same application orchestration path:
///
/// 1. Resolve the target by name and acquire its **per-agent turn lock** so two
/// `ask`s for the same target serialise FIFO (1 agent = 1 employee).
/// 2. Ensure the target is **live in the PTY registry** — reusing its terminal if
/// it is already running, otherwise launching it in the background (a normal
/// PTH launch: a live PTY *is* the channel now, not an error as before).
/// 3. **Enqueue a ticket** in the [`AgentMailbox`] (registering the reply slot)
/// **then write** the task into the target's terminal, prefixed with the asking
/// agent + ticket id so the target knows to answer via `idea_reply`.
/// 4. **Await** the [`domain::mailbox::PendingReply`] bounded by [`ASK_AGENT_TIMEOUT`]:
/// the target's later `idea_reply(result)` lands in [`Self::reply`] →
/// `mailbox.resolve`, waking this await. On timeout the ticket is retired from
/// the head ([`AgentMailbox::cancel_head`]) — **the target stays alive** — and a
/// typed timeout is returned (retry possible).
/// 5. Return the reply as [`OrchestratorOutcome::reply`] and publish
/// 2. Resolve the requester/target conversation and reject wait-for cycles before
/// enqueueing anything.
/// 3. Ensure the target has a structured/headless [`domain::ports::AgentSession`],
/// launching it through the profile adapter if it is cold.
/// 4. Drive the turn directly through [`domain::ports::AgentSession::send`] and
/// drain until the structured `Final` is captured. The target does **not** see
/// a ticket and does **not** call `idea_reply`.
/// 5. Return the captured final answer as [`OrchestratorOutcome::reply`] and publish
/// [`DomainEvent::AgentReplied`].
///
/// # Errors
/// - [`AppError::NotFound`] if the target agent is unknown;
/// - [`AppError::Invalid`] if the mailbox/PTY channel is not wired;
/// - [`AppError::Process`] on a launch/PTY-write failure, or on the await timeout
/// - [`AppError::Invalid`] if structured/headless orchestration is not wired, or if
/// the target profile cannot be driven as a structured session;
/// - [`AppError::Process`] on launch/session failure, or on the await timeout
/// (turn timeout *or* queue-wait timeout — same typed error).
async fn ask_agent(
&self,
@ -1237,29 +1279,12 @@ impl OrchestratorService {
task: String,
requester: Option<AgentId>,
) -> 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(),
))
}
};
let agent = self
.find_agent_by_name(project, &target)
.await?
.ok_or_else(|| AppError::NotFound(format!("agent {target}")))?;
let agent_id = agent.id;
// F2 — garde profil : refuser **immédiatement** une cible dont le profil ne
// sait pas consommer le pont `idea_*` matérialisé via `.mcp.json`, plutôt que
// de laisser le round-trip échouer en timeout muet (300s).
self.guard_mcp_bridge_supported(&agent.profile_id, &target)
.await?;
// Détection de cycle (cadrage C3 §6) : si l'ask vient d'un **agent** A vers la
// cible B, refuser AVANT tout enqueue si poser l'arête A→B fermerait un cycle
// d'attente (B attend déjà …→A). Pur, sans I/O ⇒ jamais de deadlock.
@ -1323,234 +1348,37 @@ 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, cold_launch) = self
.ensure_live_pty(project, agent_id, conversation_id, &target)
.await?;
// Résout la config de soumission du profil cible + s'il déclare un pont MCP.
let (submit, has_mcp) = self.submit_and_mcp_for_agent(project, agent_id).await;
// Gate cold-launch : un agent froid n'est pas encore prêt à recevoir son 1er tour.
// On le diffère s'il existe un signal pour le libérer — la connexion du pont MCP
// de l'agent (`InputMediator::release_cold_start`, déclenchée par l'McpServer sur
// `initialize`). C'est désormais le **seul** signal de readiness (le watcher
// prompt-ready PTY a été supprimé). Sans pont MCP ⇒ pas de gate (livraison
// immédiate, sinon blocage indéfini).
let gate_cold_start = cold_launch && has_mcp;
// Diagnostics : décision de gate du premier tour. `gate_cold_start=false` sur une
// cible froide SANS pont MCP livrerait immédiatement ; un `true` diffère la
// livraison jusqu'au signal MCP-initialize.
crate::diag!(
"[rendezvous] gate decision: target={target} (agent {agent_id}) cold_launch={cold_launch} \
has_mcp={has_mcp} gate_cold_start={gate_cold_start}",
);
if gate_cold_start {
input.mark_starting(agent_id);
}
input.bind_handle_with_submit(agent_id, handle.clone(), submit);
// 2. Enregistrer le ticket (slot de réponse) + livrer le tour via le médiateur
// (écriture sérialisée dans le PTY — plus d'écriture ad hoc ici). Le ticket
// porte la source (Human/Agent) et la conversation cible.
let requester_label = self.requester_label(project, requester).await;
let ticket_id = TicketId::new_random();
// Checkpoint Prompt (P6b, best-effort) : persister l'invite AVANT que `task` ne
// soit déplacé dans le `Ticket`. Source = origine de la requête (agent demandeur
// `Some(from)` ⇒ Agent, sinon Humain) — **même** `InputSource` que le ticket.
let prompt_source = match requester {
Some(from) => InputSource::agent(from),
None => InputSource::Human,
};
self.record_turn_best_effort(
&project.root,
// ── Chemin **structuré/headless uniquement** ──────────────────────────────
// La conversation inter-agent ne passe plus par le PTY ni par le rendez-vous
// `idea_reply` MCP. La cible doit être pilotable via `AgentSession::send`; son
// `Final` est la seule réponse normale du tour. Les profils legacy PTY/TUI sont
// refusés au lieu de retomber sur l'ancien chemin MCP fragile.
let structured = self.structured.as_ref().ok_or_else(|| {
AppError::Invalid(
"la conversation inter-agent headless n'est pas disponible : registre \
de sessions structurées non câblé"
.to_owned(),
)
})?;
let session = self
.ensure_structured_session(project, agent_id, &agent.profile_id, structured)
.await?
.ok_or_else(|| {
AppError::Invalid(format!(
"la cible '{target}' ne peut pas recevoir de conversation inter-agent : \
son profil ne déclare pas d'adaptateur structured/headless"
))
})?;
self.ask_structured(
project,
agent_id,
&target,
conversation_id,
prompt_source,
TurnRole::Prompt,
task.clone(),
requester,
task,
session.as_ref(),
)
.await;
// Live-state (lot LS3) : distiller l'intent AVANT que `task` ne soit déplacé
// dans le `Ticket`. Posé sur la cible après l'enqueue (délégation acceptée).
let working_intent = Self::distill_intent(&task);
let ticket = match requester {
Some(from) => {
Ticket::from_agent(ticket_id, from, conversation_id, requester_label, task)
}
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);
// Auto-update live-state (lot LS3), best-effort : la cible passe `Working` sur
// cette transition d'`ask` acceptée. N'altère jamais le succès de la délégation.
self.mark_target_working_best_effort(&project.root, agent_id, ticket_id, working_intent)
.await;
// Rendezvous beacon (diagnostics) : l'ask est désormais en attente du
// `idea_reply` (ou prompt-ready) de la cible. Si la cible termine son tour en
// texte SANS appeler `idea_reply`, ce beacon « ask started » n'aura pas de
// « ask resolved » correspondant avant l'expiration du `turn_timeout` — la
// signature exacte du blocage Main→cible.
let started = Instant::now();
crate::diag!(
"[rendezvous] ask started: requester={} -> target={target} (agent {agent_id}) \
conversation={conversation_id} ticket={ticket_id} cold_launch={cold_launch} \
gate_cold_start={gate_cold_start} turn_timeout_ms={}",
requester.map_or_else(|| "user".to_owned(), |a| a.to_string()),
turn_timeout.as_millis(),
);
// Garde RAII de fin de tour, armé JUSTE après l'enqueue (la cible est maintenant
// `Busy`). Quel que soit le chemin de sortie — erreur, timeout, ou **futur
// abandonné (drop)** — son `Drop` ramène la cible `Idle` et retire le ticket
// fantôme de la FIFO. C'est le fix de la cause racine (cf. [`BusyTurnGuard`]).
let busy_guard =
BusyTurnGuard::new(Arc::clone(input), Arc::clone(mailbox), agent_id, ticket_id);
// Delivery is the mediator's responsibility (`InputMediator::enqueue` writes the
// turn into the bound handle). The service no longer writes the PTY directly —
// no ad-hoc `[IdeA · tâche …]` line here, no `\r` band-aid (cadrage C3 §5.1).
// 3. Attendre la réponse, bornée par la **fenêtre d'inactivité réarmable**
// (signe de vie = sonde de transcript) au lieu d'un timeout plat : un long
// tour unique qui progresse n'est plus coupé à `turn_timeout`. Vrai silence
// ⇒ timeout typé (comme avant) ; plafond atteint malgré progrès ⇒ erreur
// typée distincte. Canal fermé ⇒ le garde retire le ticket au Drop.
let verdict = self
.run_ask_with_watchdog(pending, turn_timeout, &project.root, agent_id, &target, started)
.await;
match verdict {
WatchdogOutcome::Resolved(Ok(TurnResolution::Replied(result))) => {
// Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de
// déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est
// elle qui a rendu le tour) ; même conversation que le Prompt.
self.record_turn_best_effort(
&project.root,
conversation_id,
InputSource::agent(agent_id),
TurnRole::Response,
result.clone(),
)
.await;
// Auto-memory harvest (Lot E1), APRÈS l'append/handoff de la réponse :
// parse les blocs ` ```idea-memory ` et persiste les notes valides.
// Best-effort strict — n'altère jamais ce succès de délégation.
self.harvest_memory_best_effort(&project.root, &result)
.await;
// Auto-update live-state (lot LS3), best-effort : la cible vient de rendre
// son résultat ⇒ `Done` + `last_delegation` = ticket résolu.
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
// Succès : désarmer le garde AVANT de retourner. Le `mark_idle` propre
// sur cette branche est porté par le médiateur (prompt-ready / idea_reply
// qui a résolu le `pending`) ; on ne veut ni re-`cancel_head` un ticket
// déjà résolu, ni libérer un busy state qui ne nous appartient plus.
busy_guard.disarm();
crate::diag!(
"[rendezvous] ask resolved: target={target} (agent {agent_id}) \
ticket={ticket_id} after_ms={} reply_len={}",
started.elapsed().as_millis(),
result.len(),
);
Ok(self.reply_outcome(agent_id, &target, result))
}
// La cible est revenue à son prompt SANS `idea_reply` : la fenêtre de grâce a
// expiré et le médiateur a complété le tour « sans réponse ». Le ticket est
// déjà retiré (`complete_without_reply`) et la cible déjà `Idle` (prompt-ready
// `mark_idle`) ⇒ on **désarme** le garde (pas de `cancel_head`/`mark_idle`
// redondant ni de beacon « busy-guard freed » trompeur) et on renvoie une
// erreur typée claire/retryable, en ~G au lieu du timeout long. On repasse
// aussi la live-state de la cible à `Done` (best-effort) : le tour est conclu
// sans réponse, donc `idea_workstate_read` ne doit pas la voir `Working`
// (busy fantôme) comme sur la branche succès.
WatchdogOutcome::Resolved(Ok(TurnResolution::ReturnedToPromptNoReply)) => {
busy_guard.disarm();
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
crate::diag!(
"[rendezvous] ask returned-to-prompt-no-reply: target={target} \
(agent {agent_id}) ticket={ticket_id} after_ms={}",
started.elapsed().as_millis(),
);
Err(AppError::TargetReturnedNoReply(target))
}
// Plafond absolu atteint alors que la cible **progresse encore** (sonde de vie
// croissante) : verdict DISTINCT, non un faux timeout muet. Le garde fait
// `cancel_head` + `mark_idle` au Drop ; on réconcilie aussi la live-state à
// `Done` pour qu'un abandon ne laisse pas un busy fantôme.
WatchdogOutcome::CeilingActive => {
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
crate::diag!(
"[rendezvous] ask CEILING (still active): target={target} \
(agent {agent_id}) ticket={ticket_id} after_ms={}",
started.elapsed().as_millis(),
);
Err(AppError::TargetCeilingActive(target))
}
// Erreur / timeout : on laisse le garde faire `cancel_head` + `mark_idle` au
// Drop (retrait des `cancel_head` redondants — `cancel_head` reste idempotent).
WatchdogOutcome::Resolved(Err(_cancelled)) => {
crate::diag!(
"[rendezvous] ask channel-closed: target={target} (agent {agent_id}) \
ticket={ticket_id} after_ms={}",
started.elapsed().as_millis(),
);
Err(AppError::Process(format!(
"agent {target} : canal de réponse fermé avant un résultat"
)))
}
WatchdogOutcome::NoReply => {
// Vrai silence (aucun progrès sur une fenêtre) : sémantique identique à
// l'ancien timeout plat. On réconcilie la live-state à `Done` pour qu'un
// abandon ne laisse pas la cible `Working` (busy fantôme) ; le garde fait
// `cancel_head` + `mark_idle` au Drop.
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
crate::diag!(
"[rendezvous] ask TIMEOUT: target={target} (agent {agent_id}) \
ticket={ticket_id} after_ms={} (la cible n'a jamais appelé idea_reply \
ni atteint son prompt-ready, et aucun progrès observé sur la fenêtre)",
started.elapsed().as_millis(),
);
Err(AppError::from(domain::ports::AgentSessionError::Timeout))
}
}
.await
}
/// Chemin `ask` **structuré** (readiness/heartbeat lot 1) : la cible a un
@ -1562,8 +1390,8 @@ impl OrchestratorService {
/// 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`).
/// busy et l'exclusion mutuelle, mais il n'est plus une source de réponse : le ticket
/// est retiré quand le [`domain::ports::ReplyEvent::Final`] structured a été reçu.
/// Le checkpoint Prompt/Response best-effort est conservé à l'identique du chemin PTY.
#[allow(clippy::too_many_arguments)]
async fn ask_structured(
@ -1601,7 +1429,8 @@ impl OrchestratorService {
)
.await;
// Ticket dans la FIFO : comptabilité busy + `idea_reply` comme signal alternatif.
// Ticket dans la FIFO : comptabilité busy + exclusion mutuelle. La réponse ne
// viendra plus de cette mailbox, seulement du `Final` structured/headless.
let requester_label = self.requester_label(project, requester).await;
let ticket_id = TicketId::new_random();
let ticket = match requester {
@ -1617,7 +1446,7 @@ impl OrchestratorService {
// 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);
let _pending = input.enqueue_silent(agent_id, ticket);
// Auto-update live-state (lot LS3), best-effort : la cible passe `Working` sur
// cette transition d'`ask` acceptée (chemin structuré). `task` est encore vivant
// ici (utilisé par le drain plus bas), on le distille directement.
@ -1636,8 +1465,8 @@ impl OrchestratorService {
BusyTurnGuard::new(Arc::clone(input), Arc::clone(mailbox), agent_id, ticket_id);
// Rendezvous beacon (chemin structuré) : équivalent du « ask started » du chemin
// PTY. La cible n'a pas de PTY ; le tour se débloque sur le `Final` de sa session
// OU sur un `idea_reply`. Sans l'un des deux avant `turn_timeout`, le drain expire.
// PTY. La cible n'a pas de PTY ; le tour se débloque uniquement sur le `Final`
// de sa session structured/headless.
let started = Instant::now();
crate::diag!(
"[rendezvous] ask started (structured): requester={} -> target={target} \
@ -1651,35 +1480,12 @@ impl OrchestratorService {
// est **non borné** (`None`) : la borne de tour est désormais portée par la
// **fenêtre d'inactivité réarmable** autour de l'attente (signe de vie = sonde de
// transcript), pas par un timeout plat interne — un long tour unique qui progresse
// n'est plus coupé à `turn_timeout`. On attend la **première** issue : le tour
// structuré OU un `idea_reply` explicite.
// n'est plus coupé à `turn_timeout`. Aucun `idea_reply` ne peut résoudre ce tour.
let drain = drain_with_readiness(session, &task, None, input.as_ref(), agent_id);
// L'attente du rendez-vous (drain OU reply), rendue comme `Result<String, AppError>`
// pour être enveloppée par le watchdog (au lieu de `return` directs dans le `select!`).
let wait = async {
tokio::select! {
biased;
drained = drain => match drained {
Ok(content) => Ok(content),
Err(err) => Err(AppError::from(err)),
},
replied = pending => match replied {
Ok(TurnResolution::Replied(content)) => {
// La session draine encore en arrière-plan ; on fait avancer la FIFO.
input.mark_idle(agent_id);
Ok(content)
}
Ok(TurnResolution::ReturnedToPromptNoReply) => {
input.mark_idle(agent_id);
Err(AppError::TargetReturnedNoReply(target.to_owned()))
}
Err(_cancelled) => Err(AppError::Process(format!(
"agent {target} : canal de réponse fermé avant un résultat"
))),
},
}
};
// L'attente du rendez-vous structured, rendue comme `Result<String, AppError>`
// pour être enveloppée par le watchdog.
let wait = async { drain.await.map_err(AppError::from) };
// Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu.
let result = match self
@ -1733,8 +1539,10 @@ impl OrchestratorService {
}
};
// Succès : désarmer le garde (le `mark_idle` propre est déjà porté par le `Final`
// de la session ou par le bras `replied`) — pas de double `cancel_head`.
// Succès : le `Final` a rendu la réponse. On retire explicitement le ticket de
// comptabilité (aucun `idea_reply` ne le fera), puis on désarme le garde RAII.
mailbox.cancel_head(agent_id, ticket_id);
input.mark_idle(agent_id);
busy_guard.disarm();
// Checkpoint Response (best-effort), AVANT de déplacer `result`.
@ -2107,15 +1915,14 @@ impl OrchestratorService {
/// - `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`.
/// ou profil introuvable) ⇒ `AskAgent` refuse la conversation inter-agent.
///
/// 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.
/// L'auto-lancement réutilise le **même** [`LaunchAgent`] que les lancements UI :
/// 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 runtime MCP peut encore être matérialisé
/// pour les outils non conversationnels, mais il ne participe plus à la résolution
/// de la réponse inter-agent.
async fn ensure_structured_session(
&self,
project: &Project,
@ -2129,7 +1936,8 @@ impl OrchestratorService {
}
// 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.
// structuré. Sinon (agent PTY/TUI legacy, ou profil introuvable) ⇒ refus par
// l'appelant.
let is_structured = self
.profiles
.list()
@ -2141,9 +1949,26 @@ impl OrchestratorService {
return Ok(None);
}
// Une cible peut déjà être vivante dans le registre PTY parce qu'elle a été
// ouverte depuis la surface humaine historique (cellule/menu), alors que le
// chemin inter-agent actuel exige une session `AgentSession` headless. Si on
// laisse ce PTY en place, `LaunchAgent` applique correctement l'invariant
// « 1 session vivante/agent » et rend le PTY existant, donc aucune session
// structurée n'est insérée et l'ask échoue avec « aucune session structurée ».
//
// Le rendez-vous inter-agent est propriétaire du canal headless : on retire
// d'abord l'éventuelle session PTY de la cible, puis on relance via le launcher
// structuré partagé. Le node hôte est conservé best-effort pour que la surface
// puisse se rattacher au même emplacement si elle observe l'événement de relance.
let previous_node = self.sessions.node_for_agent(&agent_id);
if let Some(session_id) = self.sessions.session_for_agent(&agent_id) {
self.close_terminal
.execute(CloseTerminalInput { session_id })
.await?;
}
// 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
// insère dans CE registre). `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
@ -2152,7 +1977,7 @@ impl OrchestratorService {
agent_id,
rows: DEFAULT_ROWS,
cols: DEFAULT_COLS,
node_id: None,
node_id: previous_node,
conversation_id: None,
mcp_runtime: self
.mcp_runtime_provider
@ -2386,55 +2211,6 @@ impl OrchestratorService {
.find(|a| a.name.eq_ignore_ascii_case(name)))
}
/// Garde F2 : vérifie que le profil de la cible **sait consommer** le pont
/// `idea_*` matérialisé par IdeA, et renvoie sinon une [`AppError::Invalid`]
/// **immédiate** (au lieu d'un timeout 300s muet sur le round-trip).
///
/// **Critère retenu** (le plus robuste aujourd'hui) : le pont est honoré ssi le
/// profil porte une capacité MCP en stratégie `ConfigFile` ciblant `.mcp.json`
/// **ET** que son adaptateur structuré est `Claude`. En effet IdeA matérialise le
/// serveur MCP sous forme d'un fichier `.mcp.json` dans le run dir, ce que **seul**
/// Claude Code lit réellement ; Codex déclare pourtant la même stratégie
/// `ConfigFile(.mcp.json)` mais lit en pratique `~/.codex/config.toml` ⇒ le pont
/// n'est jamais branché et la cible ne peut pas appeler `idea_reply`. On exige donc
/// l'adaptateur `Claude` plutôt qu'une simple présence de capacité MCP, ce qui
/// exclut Codex de fait et reste valable pour tout futur profil non-Claude.
///
/// Profil introuvable ⇒ on **n'interdit pas** (laisse le flux suivre son cours
/// comme avant) : la garde ne fait que transformer un échec connu en erreur typée.
async fn guard_mcp_bridge_supported(
&self,
profile_id: &ProfileId,
target: &str,
) -> Result<(), AppError> {
let Some(profile) = self
.profiles
.list()
.await?
.into_iter()
.find(|p| &p.id == profile_id)
else {
return Ok(());
};
// 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 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
)))
}
/// Resolves the target agent profile's **submit config**
/// (`submit_sequence`/`submit_delay_ms`, ARCHITECTURE §20.3) **and** whether it
/// declares an **MCP bridge**, in a single profile lookup. The submit config is
@ -2623,6 +2399,7 @@ impl OrchestratorService {
/// service outlives the guard.
struct WaitEdgeGuard<'a> {
graph: &'a StdMutex<WaitForGraph>,
active_waits: &'a StdMutex<Vec<(AgentId, AgentId)>>,
from: AgentId,
to: AgentId,
}
@ -2636,8 +2413,18 @@ impl<'a> WaitEdgeGuard<'a> {
.unwrap_or_else(std::sync::PoisonError::into_inner);
g.add_edge(from, to);
}
{
let mut waits = service
.active_waits
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !waits.contains(&(from, to)) {
waits.push((from, to));
}
}
Self {
graph: &service.wait_for,
active_waits: &service.active_waits,
from,
to,
}
@ -2651,6 +2438,11 @@ impl Drop for WaitEdgeGuard<'_> {
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
g.remove_edge(self.from, self.to);
let mut waits = self
.active_waits
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
waits.retain(|edge| *edge != (self.from, self.to));
}
}
@ -2737,6 +2529,20 @@ mod tests {
assert_eq!(submit.delay_ms, Some(900));
}
#[test]
fn active_wait_dependencies_are_transitive_and_deduplicated() {
let edges = vec![
(aid(1), aid(2)),
(aid(2), aid(3)),
(aid(1), aid(3)),
(aid(9), aid(10)),
];
let deps = OrchestratorService::active_wait_dependencies_from_edges(&edges, aid(1));
assert_eq!(deps, vec![aid(3), aid(2)]);
}
// --- BusyTurnGuard (RAII de fin de tour) -------------------------------
//
// Fakes minimaux pour observer ce que le garde appelle à son Drop : un médiateur

58
crates/application/src/workstate/actions.rs
View File

@ -9,9 +9,11 @@
use std::sync::Arc;
use domain::input::InputMediator;
use domain::{AgentId, NodeId, Project, SessionId};
use crate::error::AppError;
use crate::orchestrator::OrchestratorService;
use crate::terminal::{CloseTerminal, CloseTerminalInput, LiveSessionKind, LiveSessions};
/// Input for [`AttachLiveAgent::execute`].
@ -125,13 +127,36 @@ pub struct StopLiveAgentOutput {
pub struct StopLiveAgent {
live: Arc<LiveSessions>,
close: Arc<CloseTerminal>,
input: Option<Arc<dyn InputMediator>>,
waits: Option<Arc<OrchestratorService>>,
}
impl StopLiveAgent {
/// Builds the use case from the live-session registry and the close primitive.
#[must_use]
pub fn new(live: Arc<LiveSessions>, close: Arc<CloseTerminal>) -> Self {
Self { live, close }
Self {
live,
close,
input: None,
waits: None,
}
}
/// Wires cancellation helpers used for user-driven stop cascades.
///
/// Stopping A while A waits on B should stop B too. The orchestrator service is
/// only used as a read-only provider of active wait edges; the input mediator is
/// used to preempt the running turn before the session is torn down.
#[must_use]
pub fn with_cascade(
mut self,
input: Arc<dyn InputMediator>,
waits: Arc<OrchestratorService>,
) -> Self {
self.input = Some(input);
self.waits = Some(waits);
self
}
/// Tears down the agent's live session.
@ -143,20 +168,40 @@ impl StopLiveAgent {
&self,
input: StopLiveAgentInput,
) -> Result<StopLiveAgentOutput, AppError> {
self.stop_dependencies(input.agent_id).await;
if let Some(mediator) = &self.input {
mediator.preempt(input.agent_id);
}
self.stop_one(input.agent_id).await
}
async fn stop_dependencies(&self, agent_id: AgentId) {
let Some(waits) = &self.waits else {
return;
};
for dep in waits.active_wait_dependencies(agent_id) {
if let Some(mediator) = &self.input {
mediator.preempt(dep);
}
let _ = self.stop_one(dep).await;
}
}
async fn stop_one(&self, agent_id: AgentId) -> Result<StopLiveAgentOutput, AppError> {
// PTY first: delegate to the existing close primitive (removes + kills).
if let Some(session_id) = self.live.pty.session_for_agent(&input.agent_id) {
if let Some(session_id) = self.live.pty.session_for_agent(&agent_id) {
self.close
.execute(CloseTerminalInput { session_id })
.await?;
return Ok(StopLiveAgentOutput {
agent_id: input.agent_id,
agent_id,
session_id,
kind: LiveSessionKind::Pty,
});
}
// Structured: remove from the registry first (so the uniqueness guard no
// longer sees a live session), then shut the session down out of the lock.
if let Some(session_id) = self.live.structured.session_id_for_agent(&input.agent_id) {
if let Some(session_id) = self.live.structured.session_id_for_agent(&agent_id) {
if let Some(session) = self.live.structured.remove(&session_id) {
session
.shutdown()
@ -164,14 +209,13 @@ impl StopLiveAgent {
.map_err(|e| AppError::Process(e.to_string()))?;
}
return Ok(StopLiveAgentOutput {
agent_id: input.agent_id,
agent_id,
session_id,
kind: LiveSessionKind::Structured,
});
}
Err(AppError::NotFound(format!(
"running session for agent {}",
input.agent_id
"running session for agent {agent_id}"
)))
}
}

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

@ -13,11 +13,11 @@ pub use actions::{
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
StopLiveAgentInput, StopLiveAgentOutput,
};
pub use reconcile::{ReconcileLiveState, ReconcileLiveStateInput};
pub use live::{
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
};
pub use reconcile::{ReconcileLiveState, ReconcileLiveStateInput};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

11
crates/application/src/workstate/reconcile.rs
View File

@ -173,7 +173,11 @@ mod tests {
run(store.clone(), HashSet::new(), 999).await;
let after = store.load().await.unwrap();
assert_eq!(after.entries.len(), 1, "still one keyed row (upsert, not append)");
assert_eq!(
after.entries.len(),
1,
"still one keyed row (upsert, not append)"
);
let row = &after.entries[0];
assert_eq!(row.status, WorkStatus::Idle, "orphan downgraded to idle");
assert_eq!(row.progress.as_deref(), Some(STALE_AT_RESTART_MARKER));
@ -225,6 +229,9 @@ mod tests {
let row = &store.load().await.unwrap().entries[0];
assert_eq!(row.status, WorkStatus::Idle);
assert_eq!(row.updated_at_ms, 3, "idle row never reconciled, not restamped");
assert_eq!(
row.updated_at_ms, 3,
"idle row never reconciled, not restamped"
);
}
}

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

@ -955,11 +955,13 @@ const TEST_GUARD: Duration = Duration::from_secs(10);
struct TestMailbox {
queues:
Mutex<HashMap<AgentId, VecDeque<(Ticket, tokio::sync::oneshot::Sender<TurnResolution>)>>>,
completions: Arc<CompletionBus>,
}
impl TestMailbox {
fn new() -> Self {
Self {
queues: Mutex::new(HashMap::new()),
completions: Arc::new(CompletionBus::default()),
}
}
fn pending(&self, agent: &AgentId) -> usize {
@ -978,7 +980,51 @@ impl TestMailbox {
.map(|q| q.iter().map(|(t, _)| t.id).collect())
.unwrap_or_default()
}
fn completions(&self) -> Arc<CompletionBus> {
Arc::clone(&self.completions)
}
}
#[derive(Clone)]
enum TestCompletion {
Replied(String),
NoReply,
Cancelled,
}
#[derive(Default)]
struct CompletionBus {
by_agent: Mutex<HashMap<AgentId, VecDeque<TestCompletion>>>,
notify: tokio::sync::Notify,
}
impl CompletionBus {
fn push(&self, agent: AgentId, completion: TestCompletion) {
self.by_agent
.lock()
.unwrap()
.entry(agent)
.or_default()
.push_back(completion);
self.notify.notify_waiters();
}
async fn next(&self, agent: AgentId) -> TestCompletion {
loop {
if let Some(item) = self
.by_agent
.lock()
.unwrap()
.get_mut(&agent)
.and_then(VecDeque::pop_front)
{
return item;
}
self.notify.notified().await;
}
}
}
impl AgentMailbox for TestMailbox {
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
let (tx, rx) = tokio::sync::oneshot::channel::<TurnResolution>();
@ -1001,6 +1047,8 @@ impl AgentMailbox for TestMailbox {
.ok_or(MailboxError::NoPendingRequest(agent))?;
queue.pop_front().expect("non-empty")
};
self.completions
.push(agent, TestCompletion::Replied(result.clone()));
let _ = slot.1.send(TurnResolution::Replied(result));
Ok(())
}
@ -1022,6 +1070,8 @@ impl AgentMailbox for TestMailbox {
.ok_or(MailboxError::NoPendingRequest(agent))?;
queue.remove(pos).expect("found position")
};
self.completions
.push(agent, TestCompletion::Replied(result.clone()));
let _ = slot.1.send(TurnResolution::Replied(result));
Ok(())
}
@ -1030,6 +1080,7 @@ impl AgentMailbox for TestMailbox {
if let Some(queue) = q.get_mut(&agent) {
if queue.front().map(|(t, _)| t.id) == Some(ticket_id) {
queue.pop_front();
self.completions.push(agent, TestCompletion::Cancelled);
}
}
}
@ -1044,6 +1095,7 @@ impl AgentMailbox for TestMailbox {
return; // receiver gone (human submit / timed-out caller): preserve head.
}
let (_, tx) = queue.pop_front().expect("head just matched");
self.completions.push(agent, TestCompletion::NoReply);
let _ = tx.send(TurnResolution::ReturnedToPromptNoReply);
}
}
@ -1200,6 +1252,92 @@ struct AskFixture {
live: Arc<RecordingLiveState>,
}
struct CompletionSession {
id: SessionId,
agent: AgentId,
completions: Arc<CompletionBus>,
}
#[async_trait]
impl AgentSession for CompletionSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
match self.completions.next(self.agent).await {
TestCompletion::Replied(content) => {
Ok(Box::new(vec![ReplyEvent::Final { content }].into_iter()))
}
TestCompletion::NoReply => Err(AgentSessionError::Io(
"target returned without a structured final".to_owned(),
)),
TestCompletion::Cancelled => Err(AgentSessionError::Io(
"structured test turn cancelled".to_owned(),
)),
}
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
#[derive(Clone)]
struct CompletionFactory {
completions: Arc<CompletionBus>,
agents_by_path: HashMap<String, AgentId>,
next_id: Arc<Mutex<u128>>,
}
impl CompletionFactory {
fn new(completions: Arc<CompletionBus>, agents_by_path: HashMap<String, AgentId>) -> Self {
Self {
completions,
agents_by_path,
next_id: Arc::new(Mutex::new(7000)),
}
}
}
#[async_trait]
impl AgentSessionFactory for CompletionFactory {
fn supports(&self, profile: &AgentProfile) -> bool {
profile.structured_adapter.is_some()
}
async fn start(
&self,
_profile: &AgentProfile,
ctx: &PreparedContext,
_cwd: &ProjectPath,
_session: &SessionPlan,
_sandbox: Option<&domain::sandbox::SandboxPlan>,
) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
let id = {
let mut n = self.next_id.lock().unwrap();
let id = SessionId::from_uuid(Uuid::from_u128(*n));
*n += 1;
id
};
let agent = self
.agents_by_path
.get(&ctx.relative_path)
.copied()
.ok_or_else(|| {
AgentSessionError::Start(format!(
"test completion session cannot resolve agent for {}",
ctx.relative_path
))
})?;
Ok(Arc::new(CompletionSession {
id,
agent,
completions: Arc::clone(&self.completions),
}))
}
}
/// Builds an orchestrator wired with a real [`InMemoryMailbox`] + PTY (B-3) and a
/// plain (non-structured) [`LaunchAgent`] — so a dead target is launched as a PTY,
/// exactly like production after B-2.
@ -1238,6 +1376,22 @@ fn ask_fixture_full(
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
let structured = Arc::new(StructuredSessions::new());
let agents_by_path: HashMap<String, AgentId> = contexts
.manifest()
.entries
.iter()
.map(|e| (e.md_path.clone(), e.agent_id))
.collect();
let completion_factory = CompletionFactory::new(mailbox.completions(), agents_by_path);
for (idx, entry) in contexts.manifest().entries.iter().enumerate() {
let session = Arc::new(CompletionSession {
id: SessionId::from_uuid(Uuid::from_u128(8_000 + idx as u128)),
agent: entry.agent_id,
completions: mailbox.completions(),
}) as Arc<dyn AgentSession>;
structured.insert(session, entry.agent_id, nid(8_000 + idx as u128));
}
let memories = Arc::new(HarvestMemories {
saved: Mutex::new(Vec::new()),
fail_save: fail_memory,
@ -1252,19 +1406,25 @@ fn ask_fixture_full(
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
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,
));
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(completion_factory) 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()),
@ -1297,6 +1457,7 @@ fn ask_fixture_full(
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()))
.with_structured(Arc::clone(&structured))
.with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new(
Arc::clone(&memories) as Arc<dyn MemoryStore>,
Arc::new(bus.clone()),
@ -1378,8 +1539,8 @@ fn reply_cmd_ticket(from: AgentId, ticket: TicketId, result: &str) -> Orchestrat
// --- B-3: ask blocks on the mailbox, reply unblocks it ---------------------
/// A live-PTY target: `ask` writes the prefixed task into its terminal and AWAITS;
/// a matching `idea_reply` resolves the head ticket and the ask returns its content.
/// A structured target: `ask` enqueues a bookkeeping ticket, sends the turn to the
/// headless session, and returns the session's `Final`. The PTY is not used.
#[tokio::test]
async fn ask_live_pty_target_writes_task_and_returns_reply() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
@ -1392,21 +1553,12 @@ async fn ask_live_pty_target_writes_task_and_returns_reply() {
// The ask enqueues a ticket and blocks awaiting the reply.
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// The task was written into the target's terminal, prefixed for idea_reply.
// The PTY is no longer part of the inter-agent conversation path.
let writes = fx.pty.writes_for(sid(800));
assert_eq!(
writes.len(),
1,
"exactly one task write to the live terminal"
);
assert!(writes[0].contains("Analyse §17"), "task body written");
assert!(
writes[0].contains("[IdeA · tâche"),
"delegated-task prefix present"
);
assert!(
writes[0].contains("idea_reply") || writes[0].contains("ticket"),
"carries ticket/idea_reply cue"
0,
"inter-agent conversation must not write to the PTY"
);
// No PTY spawned: the live terminal was reused.
assert!(
@ -3750,6 +3902,7 @@ impl AgentSessionFactory for CountingFactory {
struct StructuredAskFixture {
service: Arc<OrchestratorService>,
structured: Arc<StructuredSessions>,
sessions: Arc<TerminalSessions>,
factory: CountingFactory,
pty: FakePty,
}
@ -3831,6 +3984,7 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
StructuredAskFixture {
service,
structured,
sessions,
factory,
pty,
}
@ -3909,3 +4063,33 @@ async fn ask_warm_structured_target_reuses_session_without_relaunch() {
"cible chaude : la session est réutilisée, aucune relance"
);
}
/// Régression live : une cible structurée peut déjà être vivante en PTY parce qu'elle
/// a été ouverte depuis la surface humaine/menu historique. Le chemin `ask` headless
/// doit alors libérer ce PTY puis créer la session structurée, au lieu de laisser le
/// garde d'unicité de `LaunchAgent` rendre le PTY existant et finir par
/// « aucune session structurée vivante après lancement ».
#[tokio::test]
async fn ask_structured_target_live_as_pty_migrates_to_headless_session() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
assert_eq!(fx.factory.start_count(), 1, "headless session started");
assert_eq!(fx.pty.kills(), vec![sid(800)], "legacy PTY was stopped");
assert!(
fx.structured.session_for_agent(&aid(1)).is_some(),
"target now has a structured session"
);
assert!(
fx.sessions.session_for_agent(&aid(1)).is_none(),
"target no longer has a PTY session"
);
}