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

ticket117: synchronie métier idea_ask_agent + corrélation task ↔ rendezvous

Browse Source
This commit is contained in:
Blomios
2026-07-31 15:04:54 +02:00
parent 045e0984cc
commit 10c0bef5e5
10 changed files with 435 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -27,9 +27,10 @@ use domain::mailbox::{Ticket, TicketId};
use domain::ports::{BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle};
use domain::project::ProjectPath;
use domain::{
AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult,
BackgroundTaskState, BackgroundTaskWakePolicy, DomainEvent, OrchestratorCommand,
OrchestratorVisibility, ProfileId, Project, RuntimeAgentKey, TaskId,
AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskRendezvousLink,
BackgroundTaskResult, BackgroundTaskState, BackgroundTaskWakePolicy, DomainEvent,
OrchestratorCommand, OrchestratorVisibility, ProfileId, Project, RendezvousId, RuntimeAgentKey,
TaskId,
};
use crate::conversation::RecordTurn;
@ -128,6 +129,13 @@ fn resolve_background_cwd(
}
fn rendezvous_context_for_task(task: &BackgroundTask) -> Option<domain::RendezvousContext> {
if let Some(link) = &task.rendezvous {
return Some(domain::RendezvousContext {
requester_agent_id: link.requester_agent_id,
target_agent_id: link.target_agent_id,
conversation_id: link.conversation_id,
});
}
match &task.kind {
BackgroundTaskKind::HeadlessRendezvous {
requester_agent_id,
@ -413,6 +421,10 @@ 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<Vec<(AgentId, AgentId)>>,
/// 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`.
active_rendezvous: StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
/// Bus d'événements pour publier [`DomainEvent::AgentReplied`] à l'issue d'un
/// `ask` réussi (§17.4). Injecté via [`Self::with_events`] ; `None` ⇒ pas de
/// publication (l'`ask` fonctionne quand même).
@ -541,6 +553,12 @@ pub struct OrchestratorOutcome {
pub reply: Option<String>,
}
#[derive(Debug, Clone)]
struct ActiveRendezvous {
link: BackgroundTaskRendezvousLink,
background_tasks: Vec<TaskId>,
}
impl OrchestratorService {
/// Builds the service from the use cases and ports it dispatches to.
#[must_use]
@ -569,6 +587,7 @@ impl OrchestratorService {
conversations: None,
wait_for: StdMutex::new(WaitForGraph::new()),
active_waits: StdMutex::new(Vec::new()),
active_rendezvous: StdMutex::new(HashMap::new()),
events: None,
ask_locks: StdMutex::new(HashMap::new()),
mcp_runtime_provider: None,
@ -681,6 +700,47 @@ impl OrchestratorService {
Arc::clone(locks.entry(*agent_id).or_default())
}
fn active_rendezvous_link(
&self,
agent: RuntimeAgentKey,
) -> Option<BackgroundTaskRendezvousLink> {
self.active_rendezvous
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(&agent)
.map(|r| r.link.clone())
}
fn attach_background_to_active_rendezvous(
&self,
agent: RuntimeAgentKey,
rendezvous_id: RendezvousId,
task_id: TaskId,
) {
if let Some(active) = self
.active_rendezvous
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get_mut(&agent)
.filter(|active| active.link.rendezvous_id == rendezvous_id)
{
active.background_tasks.push(task_id);
}
}
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
@ -750,10 +810,7 @@ impl OrchestratorService {
&self,
project: &Project,
owner_agent_id: AgentId,
requester_agent_id: Option<AgentId>,
target_agent_id: AgentId,
ticket_id: TicketId,
conversation_id: domain::conversation::ConversationId,
link: BackgroundTaskRendezvousLink,
) -> Result<Option<TaskId>, AppError> {
let Some(store) = &self.background_tasks else {
return Ok(None);
@ -765,16 +822,17 @@ impl OrchestratorService {
project.id,
owner_agent_id,
BackgroundTaskKind::HeadlessRendezvous {
requester_agent_id,
target_agent_id,
ticket_id,
conversation_id,
requester_agent_id: link.requester_agent_id,
target_agent_id: link.target_agent_id,
ticket_id: link.ticket_id,
conversation_id: link.conversation_id,
},
BackgroundTaskWakePolicy::RecordOnly,
now,
None,
)
.map_err(|err| AppError::Internal(err.to_string()))?
.with_rendezvous(link)
.transition(BackgroundTaskState::Running, now)
.map_err(|err| AppError::Internal(err.to_string()))?;
store
@ -812,6 +870,27 @@ impl OrchestratorService {
.await
}
async fn mark_rendezvous_task_waiting(&self, task_id: Option<TaskId>) -> 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,
@ -1223,6 +1302,8 @@ impl OrchestratorService {
AppError::Invalid("background command runner is not configured".to_owned())
})?;
let cwd = resolve_background_cwd(&project.root, cwd)?;
let agent_key = runtime_key(project, owner);
let rendezvous = self.active_rendezvous_link(agent_key);
let output = spawn
.execute(SpawnBackgroundCommandInput {
project_id: project.id,
@ -1237,9 +1318,17 @@ impl OrchestratorService {
sandbox: None,
},
wake_policy: BackgroundTaskWakePolicy::WakeOwner,
rendezvous: rendezvous.clone(),
deadline_ms,
})
.await?;
if let Some(link) = rendezvous {
self.attach_background_to_active_rendezvous(
agent_key,
link.rendezvous_id,
output.task.id,
);
}
let state = match output.task.state {
BackgroundTaskState::Queued => "Queued",
BackgroundTaskState::Running => "Running",
@ -1812,17 +1901,19 @@ 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,
requester_agent_id: requester,
target_agent_id: agent_id,
conversation_id,
};
let rendezvous_task = self
.start_rendezvous_task(
project,
agent_id,
requester,
agent_id,
ticket_id,
conversation_id,
)
.start_rendezvous_task(project, agent_id, rendezvous_link.clone())
.await?;
let _active_rendezvous =
ActiveRendezvousGuard::new(self, agent_key, rendezvous_link.clone());
// Auto-update live-state (lot LS3), best-effort : la cible passe `Working` sur
// cette transition d'`ask` acceptée (chemin structuré). `task` est encore vivant
// ici (utilisé par le drain plus bas), on le distille directement.
@ -1918,15 +2009,13 @@ impl OrchestratorService {
};
// Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu.
let result = match self
let initial_result = match self
.run_ask_with_watchdog(wait, turn_timeout, &project.root, agent_id, target, started)
.await
{
WatchdogOutcome::Resolved(Ok(content)) => {
self.complete_rendezvous_task_success(project, agent_id, rendezvous_task, &content)
.await?;
crate::diag!(
"[rendezvous] ask resolved (structured): target={target} \
"[rendezvous] ask turn finalized (structured): target={target} \
(agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}",
started.elapsed().as_millis(),
content.len(),
@ -2012,6 +2101,67 @@ 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?;
// Succès : le `Final` a rendu la réponse. On retire explicitement le ticket de
// comptabilité (aucun `idea_reply` ne le fera), puis on désarme le garde RAII.
mailbox.cancel_head(agent_key, ticket_id);
@ -2915,6 +3065,53 @@ impl Drop for WaitEdgeGuard<'_> {
}
}
struct ActiveRendezvousGuard<'a> {
active: &'a StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
agent: RuntimeAgentKey,
rendezvous_id: RendezvousId,
}
impl<'a> ActiveRendezvousGuard<'a> {
fn new(
service: &'a OrchestratorService,
agent: RuntimeAgentKey,
link: BackgroundTaskRendezvousLink,
) -> Self {
let rendezvous_id = link.rendezvous_id;
service
.active_rendezvous
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(
agent,
ActiveRendezvous {
link,
background_tasks: Vec::new(),
},
);
Self {
active: &service.active_rendezvous,
agent,
rendezvous_id,
}
}
}
impl Drop for ActiveRendezvousGuard<'_> {
fn drop(&mut self) {
let mut active = self
.active
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if active
.get(&self.agent)
.is_some_and(|entry| entry.link.rendezvous_id == self.rendezvous_id)
{
active.remove(&self.agent);
}
}
}
/// Normalises a profile reference for tolerant matching: lowercased, with spaces,
/// dashes and underscores stripped (`"Claude Code"`, `"claude-code"`, `"claude"`
/// → comparable forms; `claude` ⊂ ... handled by the command match above).

25
crates/application/src/orchestrator/wake.rs
View File

@ -134,10 +134,33 @@ impl AgentWakeService {
owner_agent_id: agent,
});
}
drain_reply_stream_with_readiness(stream, self.input.as_ref(), key)
let final_content = drain_reply_stream_with_readiness(stream, self.input.as_ref(), key)
.await
.map_err(|err| WakeError::Session(err.to_string()))?;
if let Some(task_id) = delivery.delivered_task_id {
let task = self.load_task(task_id).await?;
if let Some(link) = task.rendezvous.as_ref() {
let still_waiting = self
.tasks
.list_open_for_agent(agent)
.await
.map_err(|err| WakeError::Store(err.to_string()))?
.into_iter()
.any(|open| {
open.id != task_id
&& open.rendezvous.as_ref().is_some_and(|open_link| {
open_link.rendezvous_id == link.rendezvous_id
})
});
if !still_waiting {
self.mailbox
.resolve_ticket(key, link.ticket_id, final_content)
.map_err(|err| WakeError::Task(err.to_string()))?;
}
}
}
self.mailbox.cancel_head(key, delivery.ticket_id);
self.input.mark_idle(key);
guard.disarm();