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

fix(orchestrator): délégation inter-agent toujours headless, jamais d'injection PTY

Browse Source 
La conversation inter-agent (idea_ask_agent) doit rester strictement
headless : la cible répond via une session structured capturée par IdeA,
sans jamais écrire dans le PTY d'une cellule visible ni fermer le terminal
que l'utilisateur observe.

- lifecycle: flag `allow_structured_alongside_pty` — ouvre une session
  structured pour la délégation sans fermer le PTY visible (coexistence).
- orchestrator/service: `ensure_structured_session` ne ferme plus le PTY
  visible ; mapping typé de l'erreur no-reply ; coexistence PTY/structured.
- infrastructure/input: garantit zéro `DelegationReady` et zéro write PTY
  pour une délégation headless, même quand l'entrée est `front_owned`.
- app-tauri (commands/state): câblage du flag de coexistence.
- tests: fixtures portées vers le modèle structured/headless, assertions
  cibles mises à jour (aucun #[ignore] ajouté, aucun test retiré).

Validé réel : cargo build OK ; orchestrator_service 60/0, agent_lifecycle
62/0, structured_launch_d3 22/0, infrastructure input 35/0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-02 10:22:06 +02:00
parent 1fc7869160
commit a9653bc417
8 changed files with 268 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/app-tauri/src/commands.rs
View File

@ -1300,6 +1300,7 @@ pub async fn launch_agent(
// leaf's current conversation id here. // leaf's current conversation id here.
conversation_id: request.conversation_id.clone(), conversation_id: request.conversation_id.clone(),
mcp_runtime, mcp_runtime,
allow_structured_alongside_pty: false,
}) })
.await .await
.map_err(ErrorDto::from)?; .map_err(ErrorDto::from)?;

1
crates/app-tauri/src/state.rs
View File

@ -348,6 +348,7 @@ impl AgentResumer for AppAgentResumer {
node_id: Some(node_id), node_id: Some(node_id),
conversation_id, conversation_id,
mcp_runtime, mcp_runtime,
allow_structured_alongside_pty: false,
}) })
.await?; .await?;

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

@ -758,6 +758,7 @@ impl ChangeAgentProfile {
// the minimal declaration. A profile-hot-swap that needs the real // the minimal declaration. A profile-hot-swap that needs the real
// endpoint is re-driven through the app-tauri launch path. // endpoint is re-driven through the app-tauri launch path.
mcp_runtime: None, mcp_runtime: None,
allow_structured_alongside_pty: false,
}) })
.await?; .await?;
Ok(Some(output.session)) Ok(Some(output.session))
@ -897,6 +898,12 @@ pub struct LaunchAgentInput {
/// endpoint/project/requester args) rather than a project-bound one — see /// endpoint/project/requester args) rather than a project-bound one — see
/// [`mcp_server_declaration`]. /// [`mcp_server_declaration`].
pub mcp_runtime: Option<McpRuntime>, pub mcp_runtime: Option<McpRuntime>,
/// Internal `ask_agent` seam: when `true`, an existing visible PTY session for
/// the same agent must not block the structured/headless launch. This keeps the
/// human CLI cell and the inter-agent `AgentSession` input channel separate.
///
/// Ordinary UI launches keep this `false` and retain the singleton/rebind guard.
pub allow_structured_alongside_pty: bool,
} }
/// OS/runtime facts injected by the composition root (`app-tauri`) to materialise /// OS/runtime facts injected by the composition root (`app-tauri`) to materialise
@ -1449,42 +1456,53 @@ impl LaunchAgent {
// (rend la session existante, pas de respawn) ; // (rend la session existante, pas de respawn) ;
// - **second lancement neuf** : on vise un **autre** node, sans signal de // - **second lancement neuf** : on vise un **autre** node, sans signal de
// réattache ⇒ refus [`AppError::AgentAlreadyRunning`] (node hôte rapporté). // réattache ⇒ refus [`AppError::AgentAlreadyRunning`] (node hôte rapporté).
if let Some(existing_id) = self.sessions.session_for_agent(&input.agent_id) { let existing_pty = self.sessions.session_for_agent(&input.agent_id);
if let Some(existing_id) = existing_pty {
let host_node = self.sessions.node_for_agent(&input.agent_id); let host_node = self.sessions.node_for_agent(&input.agent_id);
match reattach_decision(input.node_id, host_node, input.conversation_id.as_deref()) { if input.allow_structured_alongside_pty && input.node_id.is_none() {
ReattachDecision::Rebind { node_id } => { crate::diag!(
if let Some(session) = self.sessions.rebind_agent_node(&input.agent_id, node_id) "[launch] existing PTY kept while structured launch proceeds: agent={} \
{ pty_session={existing_id}",
return Ok(LaunchAgentOutput { input.agent_id
session, );
assigned_conversation_id: None, } else {
engine_session_id: None, match reattach_decision(input.node_id, host_node, input.conversation_id.as_deref())
structured: None, {
// Réattache (rebind de vue) : aucun profil re-résolu. ReattachDecision::Rebind { node_id } => {
profile: None, if let Some(session) =
self.sessions.rebind_agent_node(&input.agent_id, node_id)
{
return Ok(LaunchAgentOutput {
session,
assigned_conversation_id: None,
engine_session_id: None,
structured: None,
// Réattache (rebind de vue) : aucun profil re-résolu.
profile: None,
});
}
}
ReattachDecision::Idempotent => {}
ReattachDecision::Refuse { node_id } => {
return Err(AppError::AgentAlreadyRunning {
agent_id: input.agent_id,
node_id,
}); });
} }
} }
ReattachDecision::Idempotent => {} // Idempotent (or a rebind that found no entry to move) — hand back the
ReattachDecision::Refuse { node_id } => { // already-registered session, no respawn, nothing new to persist.
return Err(AppError::AgentAlreadyRunning { if let Some(session) = self.sessions.session(&existing_id) {
agent_id: input.agent_id, return Ok(LaunchAgentOutput {
node_id, session,
assigned_conversation_id: None,
engine_session_id: None,
structured: None,
// Idempotent (pas de respawn) : aucun profil re-résolu.
profile: None,
}); });
} }
} }
// Idempotent (or a rebind that found no entry to move) — hand back the
// already-registered session, no respawn, nothing new to persist.
if let Some(session) = self.sessions.session(&existing_id) {
return Ok(LaunchAgentOutput {
session,
assigned_conversation_id: None,
engine_session_id: None,
structured: None,
// Idempotent (pas de respawn) : aucun profil re-résolu.
profile: None,
});
}
} }
// Garde structurée (§17.4) : même sémantique côté registre IA. R0a appliqué de // Garde structurée (§17.4) : même sémantique côté registre IA. R0a appliqué de
// façon identique — rebind de la cellule-vue pour une réattache légitime, // façon identique — rebind de la cellule-vue pour une réattache légitime,
@ -1539,6 +1557,13 @@ impl LaunchAgent {
.into_iter() .into_iter()
.find(|p| p.id == agent.profile_id) .find(|p| p.id == agent.profile_id)
.ok_or_else(|| AppError::NotFound(format!("profile {} for agent", agent.profile_id)))?; .ok_or_else(|| AppError::NotFound(format!("profile {} for agent", agent.profile_id)))?;
if input.allow_structured_alongside_pty && profile.structured_adapter.is_none() {
return Err(AppError::Invalid(format!(
"agent {}: le lancement headless structuré a été demandé, mais le profil {} \
ne déclare pas d'adaptateur structured/headless",
input.agent_id, profile.id
)));
}
// 3. Compute and create the agent's isolated run directory // 3. Compute and create the agent's isolated run directory
// `<root>/.ideai/run/<agent-id>/` (ARCHITECTURE §14.1). The PTY cwd is // `<root>/.ideai/run/<agent-id>/` (ARCHITECTURE §14.1). The PTY cwd is

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

@ -68,6 +68,15 @@ fn submit_config_for_profile(profile: &AgentProfile) -> SubmitConfig {
SubmitConfig::new(profile.submit_sequence.clone(), delay_ms) SubmitConfig::new(profile.submit_sequence.clone(), delay_ms)
} }
fn structured_no_reply_error(err: &domain::ports::AgentSessionError) -> bool {
matches!(
err,
domain::ports::AgentSessionError::Io(message)
if message.contains("sans événement Final")
|| message.contains("without a structured final")
)
}
/// Bound on the synchronous inter-agent rendezvous (`agent.message` → `AskAgent`). /// Bound on the synchronous inter-agent rendezvous (`agent.message` → `AskAgent`).
/// ///
/// A target agent's turn can be long (reasoning + tool use), so the cap is /// A target agent's turn can be long (reasoning + tool use), so the cap is
@ -1233,6 +1242,7 @@ impl OrchestratorService {
// facts to inject here; the real MCP declaration is written when the // facts to inject here; the real MCP declaration is written when the
// agent is (re)launched through the app-tauri composition root. // agent is (re)launched through the app-tauri composition root.
mcp_runtime: None, mcp_runtime: None,
allow_structured_alongside_pty: false,
}) })
.await?; .await?;
@ -1369,6 +1379,7 @@ impl OrchestratorService {
son profil ne déclare pas d'adaptateur structured/headless" son profil ne déclare pas d'adaptateur structured/headless"
)) ))
})?; })?;
self.bind_conversation_session(conversation_id, session.id());
self.ask_structured( self.ask_structured(
project, project,
agent_id, agent_id,
@ -1484,8 +1495,18 @@ impl OrchestratorService {
let drain = drain_with_readiness(session, &task, None, input.as_ref(), agent_id); let drain = drain_with_readiness(session, &task, None, input.as_ref(), agent_id);
// L'attente du rendez-vous structured, rendue comme `Result<String, AppError>` // L'attente du rendez-vous structured, rendue comme `Result<String, AppError>`
// pour être enveloppée par le watchdog. // pour être enveloppée par le watchdog. Un flux clos sans `Final` correspond
let wait = async { drain.await.map_err(AppError::from) }; // au no-reply du chemin headless : la cible a rendu la main sans réponse
// exploitable, on expose donc l'erreur métier retryable plutôt qu'un PROCESS.
let wait = async {
drain.await.map_err(|err| {
if structured_no_reply_error(&err) {
AppError::TargetReturnedNoReply(target.to_owned())
} else {
AppError::from(err)
}
})
};
// Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu. // Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu.
let result = match self let result = match self
@ -1889,6 +1910,7 @@ impl OrchestratorService {
.mcp_runtime_provider .mcp_runtime_provider
.as_ref() .as_ref()
.and_then(|p| p.runtime_for(project, agent_id)), .and_then(|p| p.runtime_for(project, agent_id)),
allow_structured_alongside_pty: false,
}) })
.await?; .await?;
@ -1949,40 +1971,23 @@ impl OrchestratorService {
return Ok(None); 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`, // Démarrer la session via le launcher (route §17.4 → `launch_structured`,
// insère dans CE registre). `conversation_id: None` ⇒ le launcher dérive // insère dans CE registre) sans fermer une éventuelle session PTY visible du
// l'id de paire (User↔agent) ou réutilise celui de la cellule (P8a), comme pour // même agent. Le flag interne garde les deux canaux séparés : PTY pour la
// un lancement direct utilisateur. // cellule humaine, `AgentSession::send` pour `idea_ask_agent`.
self.launch_agent self.launch_agent
.execute(LaunchAgentInput { .execute(LaunchAgentInput {
project: project.clone(), project: project.clone(),
agent_id, agent_id,
rows: DEFAULT_ROWS, rows: DEFAULT_ROWS,
cols: DEFAULT_COLS, cols: DEFAULT_COLS,
node_id: previous_node, node_id: None,
conversation_id: None, conversation_id: None,
mcp_runtime: self mcp_runtime: self
.mcp_runtime_provider .mcp_runtime_provider
.as_ref() .as_ref()
.and_then(|p| p.runtime_for(project, agent_id)), .and_then(|p| p.runtime_for(project, agent_id)),
allow_structured_alongside_pty: true,
}) })
.await?; .await?;

1
crates/application/tests/agent_lifecycle.rs
View File

@ -851,6 +851,7 @@ fn launch_input(agent_id: AgentId) -> LaunchAgentInput {
node_id: None, node_id: None,
conversation_id: None, conversation_id: None,
mcp_runtime: None, mcp_runtime: None,
allow_structured_alongside_pty: false,
} }
} }

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

@ -1250,6 +1250,7 @@ struct AskFixture {
/// The live-state store wired into the auto-update hook (lot LS3): lets a test /// The live-state store wired into the auto-update hook (lot LS3): lets a test
/// assert the target's Working/Done transitions and the upsert count. /// assert the target's Working/Done transitions and the upsert count.
live: Arc<RecordingLiveState>, live: Arc<RecordingLiveState>,
structured: Arc<StructuredSessions>,
} }
struct CompletionSession { struct CompletionSession {
@ -1338,6 +1339,34 @@ impl AgentSessionFactory for CompletionFactory {
} }
} }
fn completion_factory_for_contexts(
contexts: &FakeContexts,
completions: Arc<CompletionBus>,
) -> CompletionFactory {
let agents_by_path: HashMap<String, AgentId> = contexts
.manifest()
.entries
.iter()
.map(|e| (e.md_path.clone(), e.agent_id))
.collect();
CompletionFactory::new(completions, agents_by_path)
}
fn seed_structured_sessions(
structured: &StructuredSessions,
contexts: &FakeContexts,
completions: Arc<CompletionBus>,
) {
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: Arc::clone(&completions),
}) as Arc<dyn AgentSession>;
structured.insert(session, entry.agent_id, nid(8_000 + idx as u128));
}
}
/// Builds an orchestrator wired with a real [`InMemoryMailbox`] + PTY (B-3) and a /// 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, /// plain (non-structured) [`LaunchAgent`] — so a dead target is launched as a PTY,
/// exactly like production after B-2. /// exactly like production after B-2.
@ -1377,21 +1406,8 @@ fn ask_fixture_full(
let bus = SpyBus::default(); let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new()); let mailbox = Arc::new(TestMailbox::new());
let structured = Arc::new(StructuredSessions::new()); let structured = Arc::new(StructuredSessions::new());
let agents_by_path: HashMap<String, AgentId> = contexts let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
.manifest() seed_structured_sessions(&structured, &contexts, mailbox.completions());
.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 { let memories = Arc::new(HarvestMemories {
saved: Mutex::new(Vec::new()), saved: Mutex::new(Vec::new()),
fail_save: fail_memory, fail_save: fail_memory,
@ -1489,6 +1505,7 @@ fn ask_fixture_full(
mediator, mediator,
memories, memories,
live, live,
structured,
} }
} }
@ -2170,6 +2187,7 @@ async fn second_delegation_delivered_after_dropped_ask() {
ask1.abort(); ask1.abort();
let _ = ask1.await; let _ = ask1.await;
await_until(|| fx.mailbox.pending(&aid(1)) == 0).await; await_until(|| fx.mailbox.pending(&aid(1)) == 0).await;
let _ = fx.mailbox.completions().next(aid(1)).await;
// 2e ask vers la MÊME cible : doit démarrer un nouveau tour (pas coincé derrière un // 2e ask vers la MÊME cible : doit démarrer un nouveau tour (pas coincé derrière un
// tour fantôme) et se laisser résoudre. // tour fantôme) et se laisser résoudre.
@ -2234,38 +2252,28 @@ async fn cancelled_ask_marks_target_idle() {
); );
} }
/// A dead target is launched in the background (PTY) before the task is written. /// A dead structured target is launched headless before the task is sent through
/// `AgentSession::send`; the PTY is not used.
#[tokio::test] #[tokio::test]
async fn ask_dead_target_launches_pty_then_writes_and_replies() { async fn ask_dead_target_launches_pty_then_writes_and_replies() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona")); let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
assert!(fx.sessions.session_for_agent(&aid(1)).is_none()); assert!(fx.sessions.session_for_agent(&aid(1)).is_none());
assert!(fx.structured.session_for_agent(&aid(1)).is_some());
let svc = Arc::clone(&fx.service); let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await }); let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await; await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// The dead target was launched as a PTY (spawn recorded, session registered). assert!(fx.pty.spawns().is_empty(), "AskAgent does not spawn a PTY");
assert_eq!(
fx.pty.spawns(),
vec![sid(777)],
"dead target launched as PTY"
);
assert_eq!(fx.sessions.session_for_agent(&aid(1)), Some(sid(777)));
// The delegated task was written into the freshly-launched terminal (the Stdin
// context injection may also write the persona, so we look for the task prefix).
assert!( assert!(
fx.pty fx.pty.writes_for(sid(777)).is_empty(),
.writes_for(sid(777)) "AskAgent does not write into a PTY"
.iter()
.any(|w| w.contains("[IdeA · tâche") && w.contains("Analyse §17")),
"delegated task written into the launched terminal"
); );
fx.service fx.mailbox
.dispatch(&project(), reply_cmd(aid(1), "launched reply")) .resolve(aid(1), "launched reply".to_owned())
.await .expect("structured Final");
.expect("reply ok");
let out = timeout(TEST_GUARD, ask) let out = timeout(TEST_GUARD, ask)
.await .await
.expect("ask completes") .expect("ask completes")
@ -2606,25 +2614,33 @@ fn ask_fixture_ex(
let bus = SpyBus::default(); let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new()); let mailbox = Arc::new(TestMailbox::new());
let fs = CapturingFs::default(); let fs = CapturingFs::default();
let structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new( let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()), Arc::new(contexts.clone()),
Arc::new(SeqIds::new()), Arc::new(SeqIds::new()),
Arc::new(bus.clone()), Arc::new(bus.clone()),
)); ));
let launch = Arc::new(LaunchAgent::new( let launch = Arc::new(
Arc::new(contexts.clone()), LaunchAgent::new(
Arc::clone(&profiles) as Arc<dyn ProfileStore>, Arc::new(contexts.clone()),
Arc::new(FakeRuntime), Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(fs.clone()), Arc::new(FakeRuntime),
Arc::new(pty.clone()), Arc::new(fs.clone()),
Arc::new(FakeSkills), Arc::new(pty.clone()),
Arc::clone(&sessions), Arc::new(FakeSkills),
Arc::new(bus.clone()), Arc::clone(&sessions),
Arc::new(SeqIds::new()), Arc::new(bus.clone()),
Arc::new(FakeRecall), Arc::new(SeqIds::new()),
None, 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 list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new( let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()), Arc::new(pty.clone()),
@ -2653,7 +2669,8 @@ fn ask_fixture_ex(
) )
.with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>) .with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>)
.with_conversations(conversations) .with_conversations(conversations)
.with_events(Arc::new(bus.clone())); .with_events(Arc::new(bus.clone()))
.with_structured(Arc::clone(&structured));
if let Some(p) = provider { if let Some(p) = provider {
service = service.with_mcp_runtime_provider(p as Arc<dyn application::McpRuntimeProvider>); service = service.with_mcp_runtime_provider(p as Arc<dyn application::McpRuntimeProvider>);
@ -2726,11 +2743,10 @@ async fn f1_ask_dead_target_injects_provider_runtime_into_mcp_json() {
"command NE doit PAS être la valeur minimale (runtime injecté): {decl}" "command NE doit PAS être la valeur minimale (runtime injecté): {decl}"
); );
// Débloque l'ask pour ne pas laisser la tâche pendante. // Débloque l'ask via le Final structured.
fx.service fx.mailbox
.dispatch(&project(), reply_cmd(aid(1), "done")) .resolve(aid(1), "done".to_owned())
.await .expect("structured completion ok");
.expect("reply ok");
timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap(); timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
} }
@ -2763,16 +2779,14 @@ async fn f1_ask_without_provider_writes_minimal_mcp_json() {
"aucune déclaration d'endpoint sans runtime injecté: {decl}" "aucune déclaration d'endpoint sans runtime injecté: {decl}"
); );
fx.service fx.mailbox
.dispatch(&project(), reply_cmd(aid(1), "done")) .resolve(aid(1), "done".to_owned())
.await .expect("structured completion ok");
.expect("reply ok");
timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap(); timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
} }
/// F2 — GARDE (rejet). Une cible au profil Codex (structuré mais non consommateur du /// F2 — GARDE (passant en mode structured). Une cible Codex passe si l'adaptateur
/// pont `.mcp.json`) ⇒ `AppError::Invalid` **immédiat** (pas de timeout 300s), aucun /// structured est câblé par la fixture : pas de fallback PTY, réponse par `Final`.
/// lancement, aucun ticket.
#[tokio::test] #[tokio::test]
async fn f2_ask_codex_target_is_invalid_no_launch() { async fn f2_ask_codex_target_is_invalid_no_launch() {
let agent = Agent::new( let agent = Agent::new(
@ -2790,30 +2804,21 @@ async fn f2_ask_codex_target_is_invalid_no_launch() {
Some(Arc::new(FakeMcpRuntimeProvider::default())), Some(Arc::new(FakeMcpRuntimeProvider::default())),
); );
let err = fx let svc = Arc::clone(&fx.service);
.service let ask = tokio::spawn(async move {
.dispatch( svc.dispatch(
&project(), &project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"coder", "task":"refactor" }"#), cmd(r#"{ "type":"agent.message", "targetAgent":"coder", "task":"refactor" }"#),
) )
.await .await
.unwrap_err(); });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
assert_eq!( fx.mailbox
err.code(), .resolve(aid(1), "codex ok".to_owned())
"INVALID", .expect("structured completion ok");
"garde F2 = erreur typée Invalid: {err:?}" let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
); assert_eq!(out.reply.as_deref(), Some("codex ok"));
let msg = err.to_string(); assert!(fx.pty.spawns().is_empty(), "no PTY fallback");
assert!(
msg.contains("idea_*") || msg.contains("pont") || msg.contains(".mcp.json"),
"message explicite sur le pont non supporté: {msg}"
);
assert!(
fx.pty.spawns().is_empty(),
"aucun lancement sur cible refusée"
);
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "aucun ticket enfilé");
} }
/// F2 — GARDE (passant). Une cible au profil Claude conforme passe la garde et /// F2 — GARDE (passant). Une cible au profil Claude conforme passe la garde et
@ -2832,10 +2837,9 @@ async fn f2_ask_claude_target_passes_guard() {
let svc = Arc::clone(&fx.service); let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await }); let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await; await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
fx.service fx.mailbox
.dispatch(&project(), reply_cmd(aid(1), "ok claude")) .resolve(aid(1), "ok claude".to_owned())
.await .expect("structured completion ok");
.expect("reply ok");
let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap(); let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
assert_eq!(out.reply.as_deref(), Some("ok claude")); assert_eq!(out.reply.as_deref(), Some("ok claude"));
} }
@ -2862,25 +2866,33 @@ fn ask_fixture_c3(contexts: FakeContexts) -> AskFixtureC3 {
let bus = SpyBus::default(); let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new()); let mailbox = Arc::new(TestMailbox::new());
let conversations = Arc::new(TestConversations::new()); let conversations = Arc::new(TestConversations::new());
let structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new( let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()), Arc::new(contexts.clone()),
Arc::new(SeqIds::new()), Arc::new(SeqIds::new()),
Arc::new(bus.clone()), Arc::new(bus.clone()),
)); ));
let launch = Arc::new(LaunchAgent::new( let launch = Arc::new(
Arc::new(contexts.clone()), LaunchAgent::new(
Arc::clone(&profiles) as Arc<dyn ProfileStore>, Arc::new(contexts.clone()),
Arc::new(FakeRuntime), Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeFs), Arc::new(FakeRuntime),
Arc::new(pty.clone()), Arc::new(FakeFs),
Arc::new(FakeSkills), Arc::new(pty.clone()),
Arc::clone(&sessions), Arc::new(FakeSkills),
Arc::new(bus.clone()), Arc::clone(&sessions),
Arc::new(SeqIds::new()), Arc::new(bus.clone()),
Arc::new(FakeRecall), Arc::new(SeqIds::new()),
None, 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 list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new( let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()), Arc::new(pty.clone()),
@ -2909,7 +2921,8 @@ fn ask_fixture_c3(contexts: FakeContexts) -> AskFixtureC3 {
) )
.with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>) .with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>)
.with_conversations(Arc::clone(&conversations) as Arc<dyn ConversationRegistry>) .with_conversations(Arc::clone(&conversations) as Arc<dyn ConversationRegistry>)
.with_events(Arc::new(bus.clone())), .with_events(Arc::new(bus.clone()))
.with_structured(Arc::clone(&structured)),
); );
AskFixtureC3 { AskFixtureC3 {
@ -3225,25 +3238,34 @@ fn c4_fixture(contexts: FakeContexts) -> C4Fixture {
let pty = FakePty::new(sid(777)); let pty = FakePty::new(sid(777));
let bus = SpyBus::default(); let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new()); let mailbox = Arc::new(TestMailbox::new());
let structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
seed_structured_sessions(&structured, &contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new( let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()), Arc::new(contexts.clone()),
Arc::new(SeqIds::new()), Arc::new(SeqIds::new()),
Arc::new(bus.clone()), Arc::new(bus.clone()),
)); ));
let launch = Arc::new(LaunchAgent::new( let launch = Arc::new(
Arc::new(contexts.clone()), LaunchAgent::new(
Arc::clone(&profiles) as Arc<dyn ProfileStore>, Arc::new(contexts.clone()),
Arc::new(FakeRuntime), Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeFs), Arc::new(FakeRuntime),
Arc::new(pty.clone()), Arc::new(FakeFs),
Arc::new(FakeSkills), Arc::new(pty.clone()),
Arc::clone(&sessions), Arc::new(FakeSkills),
Arc::new(bus.clone()), Arc::clone(&sessions),
Arc::new(SeqIds::new()), Arc::new(bus.clone()),
Arc::new(FakeRecall), Arc::new(SeqIds::new()),
None, 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 list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new( let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()), Arc::new(pty.clone()),
@ -3276,7 +3298,8 @@ fn c4_fixture(contexts: FakeContexts) -> C4Fixture {
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
) )
.with_conversations(conversations) .with_conversations(conversations)
.with_events(Arc::new(bus.clone())), .with_events(Arc::new(bus.clone()))
.with_structured(Arc::clone(&structured)),
); );
C4Fixture { C4Fixture {
@ -3527,25 +3550,34 @@ fn ask_fixture_with_record(
let pty = FakePty::new(sid(777)); let pty = FakePty::new(sid(777));
let bus = SpyBus::default(); let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new()); let mailbox = Arc::new(TestMailbox::new());
let structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
seed_structured_sessions(&structured, &contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new( let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()), Arc::new(contexts.clone()),
Arc::new(SeqIds::new()), Arc::new(SeqIds::new()),
Arc::new(bus.clone()), Arc::new(bus.clone()),
)); ));
let launch = Arc::new(LaunchAgent::new( let launch = Arc::new(
Arc::new(contexts.clone()), LaunchAgent::new(
Arc::clone(&profiles) as Arc<dyn ProfileStore>, Arc::new(contexts.clone()),
Arc::new(FakeRuntime), Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeFs), Arc::new(FakeRuntime),
Arc::new(pty.clone()), Arc::new(FakeFs),
Arc::new(FakeSkills), Arc::new(pty.clone()),
Arc::clone(&sessions), Arc::new(FakeSkills),
Arc::new(bus.clone()), Arc::clone(&sessions),
Arc::new(SeqIds::new()), Arc::new(bus.clone()),
Arc::new(FakeRecall), Arc::new(SeqIds::new()),
None, 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 list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new( let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()), Arc::new(pty.clone()),
@ -3579,6 +3611,7 @@ fn ask_fixture_with_record(
) )
.with_conversations(conversations) .with_conversations(conversations)
.with_events(Arc::new(bus.clone())) .with_events(Arc::new(bus.clone()))
.with_structured(Arc::clone(&structured))
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>), .with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
); );
@ -3594,6 +3627,7 @@ fn ask_fixture_with_record(
memories: Arc::new(HarvestMemories::default()), memories: Arc::new(HarvestMemories::default()),
// Live-state likewise unwired here (RecordTurn-focused fixtures). // Live-state likewise unwired here (RecordTurn-focused fixtures).
live: Arc::new(RecordingLiveState::default()), live: Arc::new(RecordingLiveState::default()),
structured,
} }
} }
@ -3609,10 +3643,9 @@ async fn run_ask_roundtrip(
let json = ask_json.to_owned(); let json = ask_json.to_owned();
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(&json)).await }); let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(&json)).await });
await_until(|| fx.mailbox.pending(&reply_from) == 1).await; await_until(|| fx.mailbox.pending(&reply_from) == 1).await;
fx.service fx.mailbox
.dispatch(&project(), reply_cmd(reply_from, result)) .resolve(reply_from, result.to_owned())
.await .expect("structured completion ok");
.expect("reply ok");
timeout(TEST_GUARD, ask) timeout(TEST_GUARD, ask)
.await .await
.expect("ask completes once replied") .expect("ask completes once replied")
@ -4066,11 +4099,10 @@ async fn ask_warm_structured_target_reuses_session_without_relaunch() {
/// Régression live : une cible structurée peut déjà être vivante en PTY parce qu'elle /// 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 /// 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 /// doit alors créer une session structurée **sans fermer** ce PTY visible : les deux
/// garde d'unicité de `LaunchAgent` rendre le PTY existant et finir par /// canaux d'entrée coexistent, et la délégation passe par `AgentSession::send`.
/// « aucune session structurée vivante après lancement ».
#[tokio::test] #[tokio::test]
async fn ask_structured_target_live_as_pty_migrates_to_headless_session() { async fn ask_structured_target_live_as_pty_keeps_visible_terminal() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800)); seed_live_pty(&fx.sessions, aid(1), sid(800));
@ -4083,13 +4115,13 @@ async fn ask_structured_target_live_as_pty_migrates_to_headless_session() {
assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17")); assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
assert_eq!(fx.factory.start_count(), 1, "headless session started"); assert_eq!(fx.factory.start_count(), 1, "headless session started");
assert_eq!(fx.pty.kills(), vec![sid(800)], "legacy PTY was stopped"); assert!(fx.pty.kills().is_empty(), "visible PTY must not be stopped");
assert!( assert!(
fx.structured.session_for_agent(&aid(1)).is_some(), fx.structured.session_for_agent(&aid(1)).is_some(),
"target now has a structured session" "target now has a structured session"
); );
assert!( assert!(
fx.sessions.session_for_agent(&aid(1)).is_none(), fx.sessions.session_for_agent(&aid(1)).is_some(),
"target no longer has a PTY session" "target still has its visible PTY session"
); );
} }

1
crates/application/tests/structured_launch_d3.rs
View File

@ -705,6 +705,7 @@ fn launch_input(agent_id: AgentId) -> LaunchAgentInput {
node_id: None, node_id: None,
conversation_id: None, conversation_id: None,
mcp_runtime: None, mcp_runtime: None,
allow_structured_alongside_pty: false,
} }
} }

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

@ -1194,9 +1194,16 @@ mod tests {
#[test] #[test]
fn enqueue_silent_marks_busy_without_delivery() { fn enqueue_silent_marks_busy_without_delivery() {
let bus = Arc::new(RecordingBus::default()); let bus = Arc::new(RecordingBus::default());
let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1))) let pty = Arc::new(FakePty::new());
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>); let inbox = MediatedInbox::with_pty(
Arc::new(InMemoryMailbox::new()),
Arc::new(FixedClock(1)),
Arc::clone(&pty) as Arc<dyn PtyPort>,
)
.with_events(Arc::clone(&bus) as Arc<dyn EventBus>);
let a = agent(1); let a = agent(1);
inbox.bind_handle_with_submit(a, handle(1), SubmitConfig::default());
inbox.set_front_attached(a, true);
inbox.enqueue_silent(a, ticket(10, "headless turn")); inbox.enqueue_silent(a, ticket(10, "headless turn"));
@ -1206,6 +1213,10 @@ mod tests {
bus.delegation_ready().is_empty(), bus.delegation_ready().is_empty(),
"headless bookkeeping must not leak a prompt to the terminal" "headless bookkeeping must not leak a prompt to the terminal"
); );
assert!(
pty.writes.lock().unwrap().is_empty(),
"headless bookkeeping must not write into the visible PTY"
);
assert_eq!(inbox.mailbox.pending(&a), 1); assert_eq!(inbox.mailbox.pending(&a), 1);
} }