fix(orchestrator): le lancement d'une cible structurée froide lie la session au fil A-B résolu par ask_agent — #238

ensure_structured_session reçoit désormais le conversation_id déjà résolu et le
transmet au plan de spawn au lieu de retomber sur le fallback User-B. WorkState
et la session lancée pointent ainsi sur le même fil. Test de régression
ask_cold_structured_agent_requester_launches_on_resolved_pair_thread.
This commit is contained in:
2026-08-22 11:30:35 +02:00
parent 963b51a277
commit 5449526157
2 changed files with 77 additions and 5 deletions

View File

@ -1862,7 +1862,13 @@ impl OrchestratorService {
) )
})?; })?;
let session = self let session = self
.ensure_structured_session(project, agent_id, &agent.profile_id, structured) .ensure_structured_session(
project,
agent_id,
&agent.profile_id,
structured,
conversation_id,
)
.await? .await?
.ok_or_else(|| { .ok_or_else(|| {
AppError::Invalid(format!( AppError::Invalid(format!(
@ -2558,6 +2564,7 @@ impl OrchestratorService {
agent_id: AgentId, agent_id: AgentId,
profile_id: &ProfileId, profile_id: &ProfileId,
structured: &Arc<StructuredSessions>, structured: &Arc<StructuredSessions>,
conversation_id: domain::conversation::ConversationId,
) -> Result<Option<Arc<dyn domain::ports::AgentSession>>, AppError> { ) -> Result<Option<Arc<dyn domain::ports::AgentSession>>, AppError> {
// Cible déjà chaude : route directe (aucun lancement). // Cible déjà chaude : route directe (aucun lancement).
if let Some(session) = structured.session_for_agent_in_project(project.id, &agent_id) { if let Some(session) = structured.session_for_agent_in_project(project.id, &agent_id) {
@ -2589,7 +2596,7 @@ impl OrchestratorService {
rows: DEFAULT_ROWS, rows: DEFAULT_ROWS,
cols: DEFAULT_COLS, cols: DEFAULT_COLS,
node_id: None, node_id: None,
conversation_id: None, conversation_id: Some(conversation_id.to_string()),
mcp_runtime: self mcp_runtime: self
.mcp_runtime_provider .mcp_runtime_provider
.as_ref() .as_ref()

View File

@ -4293,17 +4293,22 @@ impl AgentSession for EchoSession {
struct CountingFactory { struct CountingFactory {
starts: Arc<AtomicUsize>, starts: Arc<AtomicUsize>,
next_id: Arc<Mutex<u128>>, next_id: Arc<Mutex<u128>>,
session_plans: Arc<Mutex<Vec<SessionPlan>>>,
} }
impl CountingFactory { impl CountingFactory {
fn new() -> Self { fn new() -> Self {
Self { Self {
starts: Arc::new(AtomicUsize::new(0)), starts: Arc::new(AtomicUsize::new(0)),
next_id: Arc::new(Mutex::new(9000)), next_id: Arc::new(Mutex::new(9000)),
session_plans: Arc::new(Mutex::new(Vec::new())),
} }
} }
fn start_count(&self) -> usize { fn start_count(&self) -> usize {
self.starts.load(Ordering::SeqCst) self.starts.load(Ordering::SeqCst)
} }
fn session_plans(&self) -> Vec<SessionPlan> {
self.session_plans.lock().unwrap().clone()
}
} }
#[async_trait] #[async_trait]
impl AgentSessionFactory for CountingFactory { impl AgentSessionFactory for CountingFactory {
@ -4315,13 +4320,14 @@ impl AgentSessionFactory for CountingFactory {
_profile: &AgentProfile, _profile: &AgentProfile,
_ctx: &PreparedContext, _ctx: &PreparedContext,
_cwd: &ProjectPath, _cwd: &ProjectPath,
_session: &SessionPlan, session: &SessionPlan,
_requester: Option<&str>, _requester: Option<&str>,
_env: &[(String, String)], _env: &[(String, String)],
_sandbox: Option<&domain::sandbox::SandboxPlan>, _sandbox: Option<&domain::sandbox::SandboxPlan>,
_structured_policy: Option<&domain::ports::StructuredProviderLaunchPolicy>, _structured_policy: Option<&domain::ports::StructuredProviderLaunchPolicy>,
) -> Result<Arc<dyn AgentSession>, AgentSessionError> { ) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
self.starts.fetch_add(1, Ordering::SeqCst); self.starts.fetch_add(1, Ordering::SeqCst);
self.session_plans.lock().unwrap().push(session.clone());
let id = { let id = {
let mut n = self.next_id.lock().unwrap(); let mut n = self.next_id.lock().unwrap();
let id = SessionId::from_uuid(Uuid::from_u128(*n)); let id = SessionId::from_uuid(Uuid::from_u128(*n));
@ -4500,6 +4506,7 @@ struct StructuredAskFixture {
session_limits: Arc<SessionLimitService>, session_limits: Arc<SessionLimitService>,
background_store: Arc<RecordingBackgroundStore>, background_store: Arc<RecordingBackgroundStore>,
background_runner: Arc<RecordingBackgroundRunner>, background_runner: Arc<RecordingBackgroundRunner>,
conversations: Arc<TestConversations>,
} }
fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
@ -4574,7 +4581,7 @@ fn structured_ask_fixture_with_background(
Arc::clone(&mailbox), Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>, Arc::new(pty.clone()) as Arc<dyn PtyPort>,
)); ));
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>; let conversations = Arc::new(TestConversations::new());
let session_limits = Arc::new(SessionLimitService::new( let session_limits = Arc::new(SessionLimitService::new(
Arc::new(FixedMillisClock(1_000)) as Arc<dyn Clock>, Arc::new(FixedMillisClock(1_000)) as Arc<dyn Clock>,
Arc::new(NoopScheduler::default()) as Arc<dyn Scheduler>, Arc::new(NoopScheduler::default()) as Arc<dyn Scheduler>,
@ -4596,7 +4603,7 @@ fn structured_ask_fixture_with_background(
Arc::clone(&mediator) as Arc<dyn InputMediator>, Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
) )
.with_conversations(conversations) .with_conversations(Arc::clone(&conversations) as Arc<dyn ConversationRegistry>)
.with_events(Arc::new(bus.clone())) .with_events(Arc::new(bus.clone()))
.with_session_limits(Arc::clone(&session_limits)) .with_session_limits(Arc::clone(&session_limits))
// Même registre que le launcher : `ask_agent` y relit la session fraîchement // Même registre que le launcher : `ask_agent` y relit la session fraîchement
@ -4626,6 +4633,7 @@ fn structured_ask_fixture_with_background(
session_limits, session_limits,
background_store, background_store,
background_runner, background_runner,
conversations,
} }
} }
@ -4677,6 +4685,63 @@ async fn ask_cold_structured_target_autolaunches_session_and_final_unblocks() {
); );
} }
/// Régression #238 : quand une cible structurée froide est lancée pour un ask A→B,
/// le launcher doit recevoir l'id du fil A↔B déjà résolu par `ask_agent`. Sans cela,
/// `launch_structured` retombe sur le fallback User↔B : WorkState pointe alors vers le
/// ticket/log A↔B tandis que la session/vue fraîchement lancée est liée à User↔B.
#[tokio::test]
async fn ask_cold_structured_agent_requester_launches_on_resolved_pair_thread() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
let requester = aid(7);
let requester_uuid = uuid::Uuid::from_u128(7);
let ask_json = format!(
r#"{{ "type":"agent.message", "requestedBy":"{requester_uuid}", "targetAgent":"architect", "task":"Analyse §17" }}"#
);
let out = fx
.service
.dispatch(&project(), cmd(&ask_json))
.await
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
assert_eq!(
fx.factory.start_count(),
1,
"the cold structured target is launched exactly once"
);
let expected = fx
.conversations
.resolve(
project_id(),
domain::ConversationParty::agent(requester),
domain::ConversationParty::agent(aid(1)),
)
.id;
let user_target = fx
.conversations
.resolve(
project_id(),
domain::ConversationParty::User,
domain::ConversationParty::agent(aid(1)),
)
.id;
assert_ne!(
expected, user_target,
"the regression guard must distinguish agent-requester and user threads"
);
assert_eq!(
fx.factory.session_plans(),
vec![SessionPlan::Resume {
conversation_id: expected.to_string()
}],
"cold launch must reuse the conversation id resolved by ask_agent"
);
}
/// Réutilisation : une **deuxième** délégation vers la **même** cible (désormais chaude) /// Réutilisation : une **deuxième** délégation vers la **même** cible (désormais chaude)
/// ne relance **pas** de session — elle réutilise celle insérée au premier `ask`. Prouve /// ne relance **pas** de session — elle réutilise celle insérée au premier `ask`. Prouve
/// que le `session_for_agent` court-circuite l'auto-lancement (idempotence du chemin). /// que le `session_for_agent` court-circuite l'auto-lancement (idempotence du chemin).