From 646b9d9284a4934758be70c777949e8663b4b1be Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 12 Aug 2026 14:09:06 +0200 Subject: [PATCH] =?UTF-8?q?fix(orchestrator):=20la=20d=C3=A9l=C3=A9gation?= =?UTF-8?q?=20simple=20vers=20UX=20renvoie=20son=20Final=20structur=C3=A9?= =?UTF-8?q?=20(#207,=20QA=20verte)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Une demande idea_ask_agent vers UX échouait avec « finished without a structured Final response » alors qu'Architect répondait. Deux causes corrigées: - session/process: un tour CLI terminé en code non nul SANS stdout (ex. erreur 401/auth) renvoyait une sortie vide au lieu du stderr réel; l'orchestrateur concluait alors à un « no Final ». On remonte désormais le stderr effectif (drain sync + async + helper process_exit_message + test dédié). - orchestrator/service: la branche « ask waiting on background » masquait le premier Final structuré; la réponse synchrone redevient le premier Final (suppression de active_rendezvous_has_background_tasks / mark_rendezvous_task_waiting). Nettoyage runtime corollaire (#204): arrêt du live runtime stale (PTY + structuré) d'un agent recréé afin que la délégation route vers l'agent courant (LiveSessions, stop_live_agent_runtime_in_project, wiring backend via DeleteAgent::with_runtime_cleanup, lifecycle, workstate cascade). --- crates/application/src/agent/lifecycle.rs | 28 +- crates/application/src/error.rs | 15 +- .../application/src/orchestrator/service.rs | 134 ++---- crates/application/src/terminal/mod.rs | 5 +- crates/application/src/terminal/usecases.rs | 59 ++- crates/application/src/workstate/actions.rs | 51 +-- crates/application/tests/agent_lifecycle.rs | 65 ++- .../application/tests/orchestrator_service.rs | 383 +++++++++++++++++- crates/backend/src/lib.rs | 16 +- crates/infrastructure/src/session/mod.rs | 52 +++ crates/infrastructure/src/session/process.rs | 68 +++- 11 files changed, 673 insertions(+), 203 deletions(-) diff --git a/crates/application/src/agent/lifecycle.rs b/crates/application/src/agent/lifecycle.rs index 8205f926..70cabd52 100644 --- a/crates/application/src/agent/lifecycle.rs +++ b/crates/application/src/agent/lifecycle.rs @@ -43,7 +43,10 @@ use crate::model_server::{ }; use crate::project::project_context_path; use crate::skill::{AgentCapability, AssignedSkillResolver, ResolveAgentCapabilities}; -use crate::terminal::{StructuredSessions, TerminalSessions}; +use crate::terminal::{ + stop_live_agent_runtime_in_project, CloseTerminal, LiveSessions, StructuredSessions, + TerminalSessions, +}; use crate::workstate::GetLiveStateLean; /// Directory (relative to `.ideai/`) under which agent contexts are written. @@ -1033,13 +1036,29 @@ pub struct DeleteAgentInput { pub struct DeleteAgent { contexts: Arc, events: Arc, + runtime_cleanup: Option<(Arc, Arc)>, } impl DeleteAgent { /// Builds the use case. #[must_use] pub fn new(contexts: Arc, events: Arc) -> Self { - Self { contexts, events } + Self { + contexts, + events, + runtime_cleanup: None, + } + } + + /// Wires live-runtime cleanup performed before the manifest entry is deleted. + #[must_use] + pub fn with_runtime_cleanup( + mut self, + live: Arc, + close: Arc, + ) -> Self { + self.runtime_cleanup = Some((live, close)); + self } /// Drops the manifest entry for the agent. @@ -1058,6 +1077,11 @@ impl DeleteAgent { if entries.len() == before { return Err(AppError::NotFound(format!("agent {}", input.agent_id))); } + if let Some((live, close)) = &self.runtime_cleanup { + let _stopped = + stop_live_agent_runtime_in_project(&input.project, input.agent_id, live, close) + .await?; + } let manifest = AgentManifest::new(manifest.version, entries) .map_err(|e| AppError::Invalid(e.to_string()))?; self.contexts diff --git a/crates/application/src/error.rs b/crates/application/src/error.rs index 72246cad..7135fd56 100644 --- a/crates/application/src/error.rs +++ b/crates/application/src/error.rs @@ -73,15 +73,14 @@ pub enum AppError { node_id: NodeId, }, - /// The delegation target finished its turn and **returned to its prompt without - /// calling `idea_reply`**, so the rendezvous could produce no answer. Surfaced - /// promptly (after a short grace window) instead of blocking the caller until the - /// long safety timeout. **Retryable**: the caller may re-ask, ideally reminding the - /// target to render its result via `idea_reply` (never plain text). Carries the - /// target's display name. + /// The delegation target finished its structured/headless turn without producing + /// a [`ReplyEvent::Final`](domain::ports::ReplyEvent::Final), so the rendezvous + /// could not capture an inline answer. Surfaced promptly instead of blocking the + /// caller until the long safety timeout. **Retryable**: the caller may re-ask for a + /// concise final response. Carries the target's display name. #[error( - "agent {0} returned to its prompt without calling idea_reply (no answer was \ - rendered); retry the request and ensure the agent replies via idea_reply" + "agent {0} finished without a structured Final response (no inline answer was \ + captured); retry the request and ask for a concise final response" )] TargetReturnedNoReply(String), diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 721de71a..6059079d 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -6,7 +6,7 @@ //! that turns a request into the *same* calls the UI makes: //! //! - `spawn_agent` → [`CreateAgentFromScratch`] (if unknown) then [`LaunchAgent`], -//! - `stop_agent` → resolve the agent's live session, then [`CloseTerminal`], +//! - `stop_agent` → stop the agent's live PTY or structured runtime, //! - `update_agent_context` → [`UpdateAgentContext`]. //! //! It talks **only** to use cases and ports ([`ProfileStore`], [`TerminalSessions`]): @@ -53,7 +53,10 @@ use crate::orchestrator::{ WriteMemoryInput, }; use crate::skill::{CreateSkill, CreateSkillInput, ReadSkill, ReadSkillInput}; -use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions}; +use crate::terminal::{ + stop_live_agent_runtime_in_project, CloseTerminal, LiveSessions, StructuredSessions, + TerminalSessions, +}; use crate::workstate::{ GetLiveStateLean, StopAgentExecutionCascade, UpdateLiveState, UpdateLiveStateInput, }; @@ -431,9 +434,9 @@ pub struct OrchestratorService { /// Séparé de [`WaitForGraph`] qui reste un objet domaine minimal de détection de /// cycle, sans API de traversal. active_waits: StdMutex>, - /// Composite business rendezvous currently driven by a target agent. Used to - /// structurally attach `idea_run_in_background` tasks launched during - /// `idea_ask_agent` without parsing the model's textual `Final`. + /// Headless rendezvous currently driven by a target agent. Used to structurally + /// attach `idea_run_in_background` tasks for observability/cancellation while the + /// synchronous answer remains the first structured `Final`. active_rendezvous: StdMutex>, /// 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 @@ -751,19 +754,6 @@ impl OrchestratorService { } } - fn active_rendezvous_has_background_tasks( - &self, - agent: RuntimeAgentKey, - rendezvous_id: RendezvousId, - ) -> bool { - self.active_rendezvous - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(&agent) - .filter(|active| active.link.rendezvous_id == rendezvous_id) - .is_some_and(|active| !active.background_tasks.is_empty()) - } - /// 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 @@ -893,27 +883,6 @@ impl OrchestratorService { .await } - async fn mark_rendezvous_task_waiting(&self, task_id: Option) -> Result<(), AppError> { - let (Some(store), Some(task_id)) = (&self.background_tasks, task_id) else { - return Ok(()); - }; - let task = store - .get(task_id) - .await - .map_err(|err| AppError::Store(err.to_string()))? - .ok_or_else(|| AppError::Store(format!("background task {task_id} not found")))?; - if task.is_terminal() || task.state == BackgroundTaskState::Waiting { - return Ok(()); - } - let waiting = task - .transition(BackgroundTaskState::Waiting, self.now_ms()) - .map_err(|err| AppError::Internal(err.to_string()))?; - store - .save(&waiting) - .await - .map_err(|err| AppError::Store(err.to_string())) - } - async fn complete_rendezvous_task_failure( &self, project: &Project, @@ -1963,7 +1932,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_silent(agent_key, ticket); + let _pending = input.enqueue_silent(agent_key, ticket); let rendezvous_link = BackgroundTaskRendezvousLink { rendezvous_id: RendezvousId::new_random(), ticket_id, @@ -2071,7 +2040,7 @@ impl OrchestratorService { }; // Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu. - let initial_result = match self + let result = match self .run_ask_with_watchdog(wait, turn_timeout, &project.root, agent_id, target, started) .await { @@ -2163,64 +2132,6 @@ impl OrchestratorService { } }; - let result = if self - .active_rendezvous_has_background_tasks(agent_key, rendezvous_link.rendezvous_id) - { - self.mark_rendezvous_task_waiting(rendezvous_task).await?; - crate::diag!( - "[rendezvous] ask waiting on background: target={target} \ - (agent {agent_id}) ticket={ticket_id} rendezvous={} first_final_len={}", - rendezvous_link.rendezvous_id, - initial_result.len(), - ); - let remaining = self.ask_ceiling.saturating_sub(started.elapsed()); - match tokio::time::timeout(remaining, pending).await { - Ok(Ok(domain::mailbox::TurnResolution::Replied(content))) => content, - Ok(Ok(domain::mailbox::TurnResolution::ReturnedToPromptNoReply)) => { - let err = AppError::TargetReturnedNoReply(target.to_owned()); - self.complete_rendezvous_task_failure( - project, - agent_id, - rendezvous_task, - format!("NoReply: {err}"), - ) - .await?; - self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) - .await; - return Err(err); - } - Ok(Err(err)) => { - let err = AppError::Process(err.to_string()); - self.complete_rendezvous_task_failure( - project, - agent_id, - rendezvous_task, - format!("NoReply: {err}"), - ) - .await?; - self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) - .await; - return Err(err); - } - Err(_elapsed) => { - self.complete_rendezvous_task_failure( - project, - agent_id, - rendezvous_task, - format!( - "Timeout: composite rendezvous ceiling reached for target {target}" - ), - ) - .await?; - self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) - .await; - return Err(AppError::TargetCeilingActive(target.to_owned())); - } - } - } else { - initial_result - }; - self.complete_rendezvous_task_success(project, agent_id, rendezvous_task, &result) .await?; @@ -2665,7 +2576,7 @@ impl OrchestratorService { .as_ref() .and_then(|p| p.runtime_for(project, agent_id)), allow_structured_alongside_pty: true, - require_structured: false, + require_structured: true, }) .await?; @@ -2774,7 +2685,7 @@ impl OrchestratorService { }) } - /// `stop_agent`: translate the agent name → its live session → `CloseTerminal`. + /// `stop_agent`: translate the agent name → its live runtime → stop it. async fn stop_agent( &self, project: &Project, @@ -2785,14 +2696,19 @@ impl OrchestratorService { .await? .ok_or_else(|| AppError::NotFound(format!("agent {name}")))?; - let session_id = self - .sessions - .session_for_agent_in_project(project.id, &agent_id) - .ok_or_else(|| AppError::NotFound(format!("running session for agent {name}")))?; - - self.close_terminal - .execute(CloseTerminalInput { session_id }) - .await?; + let structured = self + .structured + .clone() + .unwrap_or_else(|| Arc::new(StructuredSessions::new())); + let live = LiveSessions::new(Arc::clone(&self.sessions), structured); + if stop_live_agent_runtime_in_project(project, agent_id, &live, &self.close_terminal) + .await? + .is_empty() + { + return Err(AppError::NotFound(format!( + "running session for agent {name}" + ))); + } Ok(OrchestratorOutcome { detail: format!("stopped agent {name}"), diff --git a/crates/application/src/terminal/mod.rs b/crates/application/src/terminal/mod.rs index 84f7ee0f..02e9f220 100644 --- a/crates/application/src/terminal/mod.rs +++ b/crates/application/src/terminal/mod.rs @@ -33,6 +33,7 @@ pub use registry::{ TerminalSessions, }; pub use usecases::{ - CloseTerminal, CloseTerminalInput, CloseTerminalOutput, OpenTerminal, OpenTerminalInput, - OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, WriteToTerminal, WriteToTerminalInput, + stop_live_agent_runtime_in_project, CloseTerminal, CloseTerminalInput, CloseTerminalOutput, + OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, + StoppedAgentRuntime, WriteToTerminal, WriteToTerminalInput, }; diff --git a/crates/application/src/terminal/usecases.rs b/crates/application/src/terminal/usecases.rs index 013281bd..357735e6 100644 --- a/crates/application/src/terminal/usecases.rs +++ b/crates/application/src/terminal/usecases.rs @@ -4,13 +4,13 @@ use std::sync::Arc; use domain::ports::{EventBus, PtyPort, SpawnSpec}; use domain::{ - DomainEvent, NodeId, ProjectPath, PtySize, SessionId, SessionKind, SessionStatus, - TerminalSession, + AgentId, DomainEvent, NodeId, Project, ProjectPath, PtySize, SessionId, SessionKind, + SessionStatus, TerminalSession, }; use crate::error::AppError; -use super::registry::TerminalSessions; +use super::registry::{LiveSessionKind, LiveSessions, TerminalSessions}; /// Input for [`OpenTerminal::execute`]. #[derive(Debug, Clone, PartialEq, Eq)] @@ -230,6 +230,59 @@ impl CloseTerminal { } } +/// Runtime session stopped for one agent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StoppedAgentRuntime { + /// Agent whose live runtime was stopped. + pub agent_id: AgentId, + /// Live session id removed from its registry. + pub session_id: SessionId, + /// Runtime registry that owned the session. + pub kind: LiveSessionKind, +} + +/// Stops one agent's live runtime across both registries. +/// +/// PTY sessions are closed through [`CloseTerminal`]. Structured sessions are +/// removed from their registry before `shutdown()` so relaunch guards stop seeing +/// them as live even while shutdown runs. +pub async fn stop_live_agent_runtime_in_project( + project: &Project, + agent_id: AgentId, + live: &LiveSessions, + close: &CloseTerminal, +) -> Result, AppError> { + let mut stopped = Vec::new(); + + if let Some(session_id) = live + .structured + .session_id_for_agent_in_project(project.id, &agent_id) + { + if let Some(session) = live.structured.remove(&session_id) { + session + .shutdown() + .await + .map_err(|e| AppError::Process(e.to_string()))?; + } + stopped.push(StoppedAgentRuntime { + agent_id, + session_id, + kind: LiveSessionKind::Structured, + }); + } + + if let Some(session_id) = live.pty.session_for_agent_in_project(project.id, &agent_id) { + close.execute(CloseTerminalInput { session_id }).await?; + stopped.push(StoppedAgentRuntime { + agent_id, + session_id, + kind: LiveSessionKind::Pty, + }); + } + + Ok(stopped) +} + /// The platform default interactive shell. /// /// Resolution policy lives in the application layer (a metier default), not the diff --git a/crates/application/src/workstate/actions.rs b/crates/application/src/workstate/actions.rs index 364e36a3..dcffcf7e 100644 --- a/crates/application/src/workstate/actions.rs +++ b/crates/application/src/workstate/actions.rs @@ -15,7 +15,9 @@ use domain::{AgentId, NodeId, Project, RuntimeAgentKey, SessionId}; use crate::error::AppError; use crate::orchestrator::OrchestratorService; -use crate::terminal::{CloseTerminal, CloseTerminalInput, LiveSessionKind, LiveSessions}; +use crate::terminal::{ + stop_live_agent_runtime_in_project, CloseTerminal, LiveSessionKind, LiveSessions, +}; /// Input for [`AttachLiveAgent::execute`]. #[derive(Debug, Clone, PartialEq, Eq)] @@ -223,42 +225,15 @@ impl StopLiveAgent { project: &Project, agent_id: AgentId, ) -> Result { - // PTY first: delegate to the existing close primitive (removes + kills). - if let Some(session_id) = self - .live - .pty - .session_for_agent_in_project(project.id, &agent_id) - { - self.close - .execute(CloseTerminalInput { session_id }) - .await?; - return Ok(StopLiveAgentOutput { - 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_in_project(project.id, &agent_id) - { - if let Some(session) = self.live.structured.remove(&session_id) { - session - .shutdown() - .await - .map_err(|e| AppError::Process(e.to_string()))?; - } - return Ok(StopLiveAgentOutput { - agent_id, - session_id, - kind: LiveSessionKind::Structured, - }); - } - Err(AppError::NotFound(format!( - "running session for agent {agent_id}" - ))) + stop_live_agent_runtime_in_project(project, agent_id, &self.live, &self.close) + .await? + .into_iter() + .next() + .map(|stopped| StopLiveAgentOutput { + agent_id: stopped.agent_id, + session_id: stopped.session_id, + kind: stopped.kind, + }) + .ok_or_else(|| AppError::NotFound(format!("running session for agent {agent_id}"))) } } diff --git a/crates/application/tests/agent_lifecycle.rs b/crates/application/tests/agent_lifecycle.rs index 6fcf3f79..d7097c68 100644 --- a/crates/application/tests/agent_lifecycle.rs +++ b/crates/application/tests/agent_lifecycle.rs @@ -47,10 +47,11 @@ use domain::{PtySize, SessionId, SkillId, SkillRef}; use uuid::Uuid; use application::{ - CreateAgentFromScratch, CreateAgentInput, DeleteAgent, DeleteAgentInput, LaunchAgent, - LaunchAgentInput, ListAgents, ListAgentsInput, PermissionProjectorRegistry, ReadAgentContext, - ReadAgentContextInput, StructuredRoutingMode, StructuredSessions, TerminalSessions, - UpdateAgentContext, UpdateAgentContextInput, UpdateAgentEffort, UpdateAgentEffortInput, + CloseTerminal, CreateAgentFromScratch, CreateAgentInput, DeleteAgent, DeleteAgentInput, + LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, LiveSessions, + PermissionProjectorRegistry, ReadAgentContext, ReadAgentContextInput, StructuredRoutingMode, + StructuredSessions, TerminalSessions, UpdateAgentContext, UpdateAgentContextInput, + UpdateAgentEffort, UpdateAgentEffortInput, }; // --------------------------------------------------------------------------- @@ -585,6 +586,7 @@ impl PtyPort for FakePty { struct FakeSession { id: SessionId, conversation_id: Option, + shutdowns: Arc>>, } #[async_trait] @@ -602,6 +604,7 @@ impl AgentSession for FakeSession { } async fn shutdown(&self) -> Result<(), AgentSessionError> { + self.shutdowns.lock().unwrap().push(self.id); Ok(()) } } @@ -613,6 +616,7 @@ struct FakeStructuredFactory { efforts: Arc>>>, envs: Arc>>>, policies: Arc>>>, + shutdowns: Arc>>, next_session: SessionId, } @@ -624,6 +628,7 @@ impl FakeStructuredFactory { efforts: Arc::new(Mutex::new(Vec::new())), envs: Arc::new(Mutex::new(Vec::new())), policies: Arc::new(Mutex::new(Vec::new())), + shutdowns: Arc::new(Mutex::new(Vec::new())), next_session, } } @@ -679,6 +684,7 @@ impl AgentSessionFactory for FakeStructuredFactory { Ok(Arc::new(FakeSession { id: self.next_session, conversation_id: Some("engine-session-1".to_owned()), + shutdowns: Arc::clone(&self.shutdowns), })) } } @@ -1048,6 +1054,56 @@ async fn delete_removes_entry_then_unknown_is_not_found() { assert_eq!(err.code(), "NOT_FOUND", "got {err:?}"); } +#[tokio::test] +async fn delete_agent_cleans_structured_then_pty_runtime_before_manifest_removal() { + let a = scratch_agent(aid(1), "Backend", "agents/backend.md", pid(9)); + let contexts = FakeContexts::with_agent(&a, "ctx"); + let sessions = Arc::new(TerminalSessions::new()); + let structured = Arc::new(StructuredSessions::new()); + let live = Arc::new(LiveSessions::new( + Arc::clone(&sessions), + Arc::clone(&structured), + )); + let pty = FakePty::new(trace(), sid(42)); + let close = Arc::new(CloseTerminal::new( + Arc::new(pty.clone()), + Arc::clone(&sessions), + )); + let shutdowns = Arc::new(Mutex::new(Vec::new())); + structured.insert_in_project( + project().id, + Arc::new(FakeSession { + id: sid(555), + conversation_id: Some("engine-session-stale".to_owned()), + shutdowns: Arc::clone(&shutdowns), + }), + a.id, + nid(7), + ); + seed_live_agent_session(&sessions, a.id, nid(8), sid(42)); + + let delete = DeleteAgent::new(Arc::new(contexts.clone()), Arc::new(SpyBus::default())) + .with_runtime_cleanup(live, close); + + delete + .execute(DeleteAgentInput { + project: project(), + agent_id: a.id, + }) + .await + .unwrap(); + + assert_eq!(*shutdowns.lock().unwrap(), vec![sid(555)]); + assert_eq!(pty.kills(), vec![sid(42)]); + assert!(structured + .session_for_agent_in_project(project().id, &a.id) + .is_none()); + assert!(sessions + .session_for_agent_in_project(project().id, &a.id) + .is_none()); + assert!(contexts.manifest().entries.is_empty()); +} + // --------------------------------------------------------------------------- // LaunchAgent // --------------------------------------------------------------------------- @@ -1369,6 +1425,7 @@ async fn chat_intent_reattaches_existing_structured_session_after_stale_pty_tran Arc::new(FakeSession { id: sid(555), conversation_id: Some("engine-session-stale".to_owned()), + shutdowns: Arc::new(Mutex::new(Vec::new())), }), agent.id, stale_custom_node, diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index f00f573f..573a9630 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -390,6 +390,10 @@ impl RecordingBackgroundStore { fn task(&self, id: TaskId) -> Option { self.tasks.lock().unwrap().get(&id).cloned() } + + fn tasks(&self) -> Vec { + self.tasks.lock().unwrap().values().cloned().collect() + } } #[async_trait] impl BackgroundTaskStore for RecordingBackgroundStore { @@ -1009,6 +1013,41 @@ async fn stop_agent_kills_the_right_session() { assert!(fx.sessions.session(&sid(777)).is_none()); } +#[tokio::test] +async fn stop_agent_stops_structured_session() { + let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); + let contexts = FakeContexts::with_agent(&agent, "# persona"); + let fx = ask_fixture(contexts); + let session = Arc::new(CompletionSession { + id: sid(9010), + agent: agent.id, + completions: fx.mailbox.completions(), + }) as Arc; + fx.structured + .insert_in_project(project_id(), session, agent.id, nid(9010)); + assert!(fx + .structured + .session_for_agent_in_project(project_id(), &agent.id) + .is_some()); + + fx.service + .dispatch( + &project(), + cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#), + ) + .await + .expect("stop ok"); + + assert!( + fx.pty.kills().is_empty(), + "structured stop must not kill PTY" + ); + assert!(fx + .structured + .session_for_agent_in_project(project_id(), &agent.id) + .is_none()); +} + #[tokio::test] async fn stop_agent_without_live_session_is_not_found() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); @@ -1025,6 +1064,47 @@ async fn stop_agent_without_live_session_is_not_found() { assert_eq!(err.code(), "NOT_FOUND", "got {err:?}"); } +#[tokio::test] +async fn ask_agent_after_delete_recreate_same_name_uses_new_agent_id() { + let old = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); + let new = scratch_agent(aid(2), "dev-backend", "agents/dev-backend-2.md"); + let contexts = FakeContexts::with_agent(&new, "# recreated persona"); + let fx = ask_fixture(contexts); + + let old_session = Arc::new(CompletionSession { + id: sid(9001), + agent: old.id, + completions: fx.mailbox.completions(), + }) as Arc; + fx.structured + .insert_in_project(project_id(), old_session, old.id, nid(9001)); + fx.mailbox + .completions() + .push(old.id, TestCompletion::Replied("old-agent".to_owned())); + fx.mailbox + .completions() + .push(new.id, TestCompletion::Replied("new-agent".to_owned())); + + let out = fx + .service + .dispatch( + &project(), + cmd(r#"{ "type":"agent.message", "targetAgent":"dev-backend", "task":"ping" }"#), + ) + .await + .unwrap(); + + assert_eq!(out.reply.as_deref(), Some("new-agent")); + assert!(fx + .structured + .session_for_agent_in_project(project_id(), &old.id) + .is_some()); + assert!(fx + .structured + .session_for_agent_in_project(project_id(), &new.id) + .is_some()); +} + #[tokio::test] async fn update_agent_context_overwrites_md() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); @@ -4198,6 +4278,96 @@ impl AgentSession for RateLimitedSession { } } +struct EmptyStructuredSession { + id: SessionId, +} + +#[async_trait] +impl AgentSession for EmptyStructuredSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(format!("engine-{}", self.id)) + } + + async fn send(&self, _prompt: &str) -> Result { + Ok(Box::new(Vec::::new().into_iter())) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +struct FailingStructuredSession { + id: SessionId, + message: String, +} + +#[async_trait] +impl AgentSession for FailingStructuredSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(format!("engine-{}", self.id)) + } + + async fn send(&self, _prompt: &str) -> Result { + Err(AgentSessionError::Io(self.message.clone())) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +struct BackgroundThenFinalSession { + id: SessionId, + service: Arc, +} + +#[async_trait] +impl AgentSession for BackgroundThenFinalSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(format!("engine-{}", self.id)) + } + + async fn send(&self, _prompt: &str) -> Result { + self.service + .dispatch( + &project(), + OrchestratorCommand::RunInBackground { + owner: aid(1), + label: "correlated child".to_owned(), + command: "bash".to_owned(), + args: vec!["-lc".to_owned(), "sleep 60".to_owned()], + cwd: None, + deadline_ms: None, + }, + ) + .await + .map_err(|err| AgentSessionError::Io(err.to_string()))?; + Ok(Box::new( + vec![ReplyEvent::Final { + content: "business final".to_owned(), + }] + .into_iter(), + )) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + #[derive(Default)] struct NoopScheduler { next: Mutex, @@ -4241,9 +4411,24 @@ struct StructuredAskFixture { pty: FakePty, bus: SpyBus, session_limits: Arc, + background_store: Arc, + background_runner: Arc, } fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { + structured_ask_fixture_with_background(contexts, false) +} + +fn structured_ask_fixture_with_observable_background( + contexts: FakeContexts, +) -> StructuredAskFixture { + structured_ask_fixture_with_background(contexts, true) +} + +fn structured_ask_fixture_with_background( + contexts: FakeContexts, + enable_background: bool, +) -> StructuredAskFixture { let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()])); let sessions = Arc::new(TerminalSessions::new()); let pty = FakePty::new(sid(777)); @@ -4251,6 +4436,15 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { let mailbox = Arc::new(TestMailbox::new()); let structured = Arc::new(StructuredSessions::new()); let factory = CountingFactory::new(); + let background_store = Arc::new(RecordingBackgroundStore::default()); + let background_runner = Arc::new(RecordingBackgroundRunner::default()); + let background_clock = Arc::new(FixedMillisClock(42)) as Arc; + let spawn_background = Arc::new(SpawnBackgroundCommand::new( + Arc::clone(&background_store) as Arc, + Arc::clone(&background_runner) as Arc, + Arc::clone(&background_clock), + Arc::new(SeqIds::new()) as Arc, + )); let create = Arc::new(CreateAgentFromScratch::new( Arc::new(contexts.clone()), @@ -4301,28 +4495,39 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { Arc::new(NoopResumer), )); - let service = Arc::new( - OrchestratorService::new( - create, - launch, - list, - close, - update, - create_skill, - Arc::clone(&profiles) as Arc, - Arc::clone(&sessions), - ) - .with_input_mediator( - Arc::clone(&mediator) as Arc, - Arc::clone(&mailbox) as Arc, - ) - .with_conversations(conversations) - .with_events(Arc::new(bus.clone())) - .with_session_limits(Arc::clone(&session_limits)) - // 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)), - ); + let mut service = OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + Arc::clone(&sessions), + ) + .with_input_mediator( + Arc::clone(&mediator) as Arc, + Arc::clone(&mailbox) as Arc, + ) + .with_conversations(conversations) + .with_events(Arc::new(bus.clone())) + .with_session_limits(Arc::clone(&session_limits)) + // 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)); + + if enable_background { + service = service + .with_background_tasks( + Arc::clone(&background_store) as Arc, + Arc::clone(&background_clock), + ) + .with_spawn_background_command(spawn_background) + .with_background_task_runner( + Arc::clone(&background_runner) as Arc + ); + } + let service = Arc::new(service); StructuredAskFixture { service, @@ -4332,6 +4537,8 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { pty, bus, session_limits, + background_store, + background_runner, } } @@ -4446,6 +4653,138 @@ async fn ask_structured_target_live_as_pty_keeps_visible_terminal() { ); } +/// Régression #201 : une tâche de fond lancée pendant le tour headless reste corrélée +/// au rendez-vous pour l'observabilité, mais elle ne prolonge plus la réponse +/// synchrone. Le premier `Final` structured est le succès métier. +#[tokio::test] +async fn ask_structured_final_wins_even_with_attached_background_task() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture_with_observable_background(FakeContexts::with_agent( + &agent, + "# persona", + )); + fx.structured.insert_in_project( + project_id(), + Arc::new(BackgroundThenFinalSession { + id: sid(9800), + service: Arc::clone(&fx.service), + }), + aid(1), + nid(700), + ); + + let out = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("Final must win even when a background task is attached"); + + assert_eq!(out.reply.as_deref(), Some("business final")); + assert_eq!( + fx.background_runner.specs().len(), + 1, + "the child background command is still spawned for observability" + ); + + let tasks = fx.background_store.tasks(); + let rendezvous_task = tasks + .iter() + .find(|task| { + matches!( + task.kind, + domain::BackgroundTaskKind::HeadlessRendezvous { .. } + ) + }) + .expect("headless rendezvous task persisted"); + assert_eq!( + rendezvous_task.state, + BackgroundTaskState::Completed, + "the rendezvous itself is marked successful inline" + ); + + let child_task = tasks + .iter() + .find(|task| matches!(task.kind, domain::BackgroundTaskKind::Command { .. })) + .expect("attached command task persisted"); + assert_eq!( + child_task.state, + BackgroundTaskState::Running, + "the attached background task remains observable without gating the reply" + ); + assert!( + child_task.rendezvous.is_some(), + "the child task keeps its rendezvous correlation" + ); +} + +/// Régression #201 : si le flux structuré se clôt sans `Final`, l'erreur publique +/// décrit l'absence de réponse inline exploitable, sans mentionner l'ancien protocole +/// `idea_reply`. +#[tokio::test] +async fn ask_structured_closed_stream_without_final_returns_inline_no_reply_error() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + fx.structured.insert_in_project( + project_id(), + Arc::new(EmptyStructuredSession { id: sid(9801) }), + aid(1), + nid(700), + ); + + let err = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect_err("closed stream without Final must be a typed no-reply"); + + assert_eq!(err.code(), "TARGET_RETURNED_NO_REPLY"); + let message = err.to_string(); + assert!( + message.contains("structured Final response"), + "message must explain the missing inline Final, got: {message}" + ); + assert!( + !message.contains("idea_reply"), + "message must not mention the legacy reply tool, got: {message}" + ); +} + +/// Régression #207 : si la CLI structurée échoue avant tout stdout exploitable +/// (ex. `401 Unauthorized`), l'orchestrateur doit remonter l'erreur réelle en +/// `PROCESS` au lieu de maquiller l'échec en faux no-reply. +#[tokio::test] +async fn ask_structured_session_error_surfaces_process_message_not_no_reply() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + fx.structured.insert_in_project( + project_id(), + Arc::new(FailingStructuredSession { + id: sid(9802), + message: "codex exec exited with code 1: 401 Unauthorized: Missing bearer" + .to_owned(), + }), + aid(1), + nid(700), + ); + + let err = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect_err("session startup failure must surface as a process error"); + + assert_eq!(err.code(), "PROCESS"); + let message = err.to_string(); + assert!( + message.contains("401 Unauthorized: Missing bearer"), + "public error must preserve stderr detail, got: {message}" + ); + assert!( + !message.contains("structured Final response"), + "public error must not fall back to no-reply wording, got: {message}" + ); +} + /// Sous-lot adjacent #7/F2 : une limite détectée pendant un rendez-vous structuré /// doit aussi passer par `SessionLimitService::on_rate_limited` pour la cible, afin /// de publier le flux mono-agent `AgentRateLimited` puis `AgentResumeScheduled`. diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index e5b0a5d2..1564e40f 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -1514,6 +1514,10 @@ impl BackendCore { // `profile.structured_adapter`. Injectés dans LaunchAgent (routage §17.4) et // ChangeAgentProfile (shutdown polymorphe au hot-swap). let structured_sessions = Arc::new(StructuredSessions::new()); + let live_sessions = Arc::new(LiveSessions::new( + Arc::clone(&terminal_sessions), + Arc::clone(&structured_sessions), + )); let openai_tool_invoker = Arc::new(LateBoundOpenAiToolInvoker::new()); let openai_tool_invoker_port = Arc::clone(&openai_tool_invoker) as Arc; let session_factory = Arc::new( @@ -2065,10 +2069,10 @@ impl BackendCore { )); let read_agent_context = Arc::new(ReadAgentContext::new(Arc::clone(&contexts_port))); let update_agent_context = Arc::new(UpdateAgentContext::new(Arc::clone(&contexts_port))); - let delete_agent = Arc::new(DeleteAgent::new( - Arc::clone(&contexts_port), - Arc::clone(&events_port), - )); + let delete_agent = Arc::new( + DeleteAgent::new(Arc::clone(&contexts_port), Arc::clone(&events_port)) + .with_runtime_cleanup(Arc::clone(&live_sessions), Arc::clone(&close_terminal)), + ); // LaunchAgent shares the SAME pty_port and terminal_sessions as the terminal // use cases — indispensable for transport bridges to work correctly. // @@ -2661,10 +2665,6 @@ impl BackendCore { } }); } - let live_sessions = Arc::new(LiveSessions::new( - Arc::clone(&terminal_sessions), - Arc::clone(&structured_sessions), - )); let delete_model_artifact = Arc::new(DeleteModelArtifact::new( Arc::clone(&model_server_registry_port), Arc::clone(&model_server_probe_port), diff --git a/crates/infrastructure/src/session/mod.rs b/crates/infrastructure/src/session/mod.rs index 33621049..fd75138f 100644 --- a/crates/infrastructure/src/session/mod.rs +++ b/crates/infrastructure/src/session/mod.rs @@ -186,6 +186,58 @@ mod tests { assert!(matches!(err, AgentSessionError::Start(_)), "vu: {err:?}"); } + #[tokio::test] + async fn run_turn_nonzero_without_stdout_surfaces_stderr() { + use std::io::Write as _; + use std::sync::atomic::{AtomicU64, Ordering}; + + static C: AtomicU64 = AtomicU64::new(0); + let mut path = std::env::temp_dir(); + path.push(format!( + "idea-empty-failing-cli-{}-{}", + std::process::id(), + C.fetch_add(1, Ordering::Relaxed) + )); + { + let mut file = std::fs::File::create(&path).expect("write fake cli"); + file.write_all( + b"#!/bin/sh\nprintf '%s\\n' '401 Unauthorized: Missing bearer' >&2\nexit 1\n", + ) + .expect("write fake cli"); + file.sync_all().expect("sync fake cli"); + } + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mut perms = std::fs::metadata(&path).unwrap().permissions(); + perms.set_mode(0o755); + std::fs::set_permissions(&path, perms).unwrap(); + } + + let spec = super::process::SpawnLine { + command: path.to_string_lossy().into_owned(), + args: Vec::new(), + cwd: "/".to_owned(), + env: Vec::new(), + stdin: None, + sandbox: None, + }; + + let err = run_turn(&spec, None, None, None) + .await + .expect_err("empty nonzero process must be a session error"); + match err { + AgentSessionError::Io(message) => { + assert!( + message.contains("401 Unauthorized: Missing bearer"), + "stderr must be preserved, got: {message}" + ); + } + other => panic!("expected Io, got {other:?}"), + } + let _ = std::fs::remove_file(&path); + } + // -- parse_event Claude (format RÉEL vérifié 2026-06-09) -------------- #[test] diff --git a/crates/infrastructure/src/session/process.rs b/crates/infrastructure/src/session/process.rs index 9a82fa68..84b21392 100644 --- a/crates/infrastructure/src/session/process.rs +++ b/crates/infrastructure/src/session/process.rs @@ -32,7 +32,7 @@ use std::sync::mpsc::Sender; use std::sync::Arc; use std::time::Duration; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::process::Command; use domain::ports::AgentSessionError; @@ -219,7 +219,7 @@ fn drain_sandboxed( killer_tx: tokio::sync::oneshot::Sender>>, line_tap: Option>, ) -> Result, AgentSessionError> { - use std::io::{BufRead, BufReader as StdBufReader, Write as _}; + use std::io::{BufRead, BufReader as StdBufReader, Read as _, Write as _}; use std::process::{Command as StdCommand, Stdio}; use std::sync::Mutex as StdMutex; @@ -266,6 +266,17 @@ fn drain_sandboxed( .stdout .take() .ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?; + let stderr = child + .stderr + .take() + .ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?; + let stderr_handle = std::thread::spawn(move || { + let mut bytes = Vec::new(); + StdBufReader::new(stderr) + .read_to_end(&mut bytes) + .map(|_| bytes) + .map_err(|e| AgentSessionError::Io(e.to_string())) + }); let child = Arc::new(StdMutex::new(child)); // Donne au côté async de quoi tuer l'enfant en cas de timeout. let _ = killer_tx.send(Arc::clone(&child)); @@ -285,8 +296,20 @@ fn drain_sandboxed( } // 6. Reap (pas de zombie). Lock tenu brièvement. - if let Ok(mut c) = child.lock() { - let _ = c.wait(); + let status = child + .lock() + .map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))? + .wait() + .map_err(|e| AgentSessionError::Io(e.to_string()))?; + let stderr_bytes = stderr_handle + .join() + .map_err(|_| AgentSessionError::Io("stderr thread terminé sans résultat".to_owned()))??; + if !status.success() && collected.is_empty() { + return Err(AgentSessionError::Io(process_exit_message( + &command, + status.code(), + &stderr_bytes, + ))); } Ok(collected) } @@ -333,6 +356,14 @@ async fn drain( .stdout .take() .ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?; + let mut stderr = child + .stderr + .take() + .ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?; + let stderr_task = tokio::spawn(async move { + let mut bytes = Vec::new(); + stderr.read_to_end(&mut bytes).await.map(|_| bytes) + }); let mut lines = BufReader::new(stdout).lines(); let mut collected = Vec::new(); @@ -350,9 +381,21 @@ async fn drain( } // Réifie le statut du process pour ne pas laisser de zombie ; un statut non nul - // n'est pas en soi une erreur de session (la CLI peut signaler une erreur - // *dans* sa sortie structurée, captée par le parser). - let _ = child.wait().await.map_err(map_io)?; + // n'est pas en soi une erreur de session si la CLI a imprimé une sortie + // structurée, captée ensuite par le parser. Sans stdout, il n'y a rien à parser : + // on remonte le stderr réel au lieu de laisser l'appelant conclure à un no-Final. + let status = child.wait().await.map_err(map_io)?; + let stderr_bytes = stderr_task + .await + .map_err(|e| AgentSessionError::Io(format!("lecture stderr: {e}")))? + .map_err(map_io)?; + if !status.success() && collected.is_empty() { + return Err(AgentSessionError::Io(process_exit_message( + &spec.command, + status.code(), + &stderr_bytes, + ))); + } Ok(collected) } @@ -360,3 +403,14 @@ async fn drain( fn map_io(e: io::Error) -> AgentSessionError { AgentSessionError::Io(e.to_string()) } + +fn process_exit_message(command: &str, code: Option, stderr: &[u8]) -> String { + let status = code.map_or_else(|| "signal".to_owned(), |code| format!("code {code}")); + let stderr = String::from_utf8_lossy(stderr); + let stderr = stderr.trim(); + if stderr.is_empty() { + format!("{command} exited with {status} without structured output") + } else { + format!("{command} exited with {status}: {stderr}") + } +}