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

fix(orchestrator): la délégation simple vers UX renvoie son Final structuré (#207, QA verte)

Browse Source 
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).
This commit is contained in:
Blomios
2026-08-12 14:09:06 +02:00
parent ba176e3cf2
commit 646b9d9284
11 changed files with 673 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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<String>,
shutdowns: Arc<Mutex<Vec<SessionId>>>,
}
#[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<Mutex<Vec<Option<String>>>>,
envs: Arc<Mutex<Vec<Vec<(String, String)>>>>,
policies: Arc<Mutex<Vec<Option<domain::ports::StructuredProviderLaunchPolicy>>>>,
shutdowns: Arc<Mutex<Vec<SessionId>>>,
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,

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

@ -390,6 +390,10 @@ impl RecordingBackgroundStore {
fn task(&self, id: TaskId) -> Option<BackgroundTask> {
self.tasks.lock().unwrap().get(&id).cloned()
}
fn tasks(&self) -> Vec<BackgroundTask> {
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<dyn AgentSession>;
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<dyn AgentSession>;
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<String> {
Some(format!("engine-{}", self.id))
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
Ok(Box::new(Vec::<ReplyEvent>::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<String> {
Some(format!("engine-{}", self.id))
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
Err(AgentSessionError::Io(self.message.clone()))
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
struct BackgroundThenFinalSession {
id: SessionId,
service: Arc<OrchestratorService>,
}
#[async_trait]
impl AgentSession for BackgroundThenFinalSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
Some(format!("engine-{}", self.id))
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
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<u128>,
@ -4241,9 +4411,24 @@ struct StructuredAskFixture {
pty: FakePty,
bus: SpyBus,
session_limits: Arc<SessionLimitService>,
background_store: Arc<RecordingBackgroundStore>,
background_runner: Arc<RecordingBackgroundRunner>,
}
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<dyn domain::ports::Clock>;
let spawn_background = Arc::new(SpawnBackgroundCommand::new(
Arc::clone(&background_store) as Arc<dyn BackgroundTaskStore>,
Arc::clone(&background_runner) as Arc<dyn BackgroundTaskRunner>,
Arc::clone(&background_clock),
Arc::new(SeqIds::new()) as Arc<dyn IdGenerator>,
));
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<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()))
.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<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()))
.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<dyn BackgroundTaskStore>,
Arc::clone(&background_clock),
)
.with_spawn_background_command(spawn_background)
.with_background_task_runner(
Arc::clone(&background_runner) as Arc<dyn BackgroundTaskRunner>
);
}
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`.