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

feat(application): réveil d'agent et rendez-vous comme tâche de fond (B5-B6)

Browse Source 
Orchestration du réveil (wake) d'un agent sur complétion/message et
traitement du rendez-vous inter-agent comme tâche de fond de 1re classe.
Couvert par agent_wake (vert).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-02 15:47:34 +02:00
parent c537da54ef
commit f94b54239b
5 changed files with 1092 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -23,11 +23,12 @@ use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef,
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
use domain::input::{InputMediator, InputSource, SubmitConfig};
use domain::mailbox::{Ticket, TicketId};
use domain::ports::{Clock, EventBus, ProfileStore, PtyHandle};
use domain::ports::{BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle};
use domain::project::ProjectPath;
use domain::{
AgentId, AgentProfile, DomainEvent, OrchestratorCommand, OrchestratorVisibility, ProfileId,
Project,
AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult,
BackgroundTaskState, BackgroundTaskWakePolicy, DomainEvent, OrchestratorCommand,
OrchestratorVisibility, ProfileId, Project, TaskId,
};
use crate::conversation::RecordTurn;
@ -77,6 +78,17 @@ fn structured_no_reply_error(err: &domain::ports::AgentSessionError) -> bool {
)
}
fn bound_task_text(value: &str, max_bytes: usize) -> String {
if value.len() <= max_bytes {
return value.to_owned();
}
let mut end = max_bytes;
while !value.is_char_boundary(end) {
end -= 1;
}
value[..end].to_owned()
}
/// 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
@ -417,6 +429,10 @@ pub struct OrchestratorService {
/// [`Self::with_ask_ceiling`] (la composition root résout l'override d'env) ; les
/// tests le rétrécissent à quelques ms. Défaut = 4 h.
ask_ceiling: Duration,
/// Store durable des tâches de fond de 1re classe. B6 l'utilise pour tracer un
/// `idea_ask_agent` comme [`BackgroundTaskKind::HeadlessRendezvous`]. `None` ⇒
/// comportement historique sans persistance de tâche (call sites non câblés).
background_tasks: Option<Arc<dyn BackgroundTaskStore>>,
}
/// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés
@ -487,9 +503,27 @@ impl OrchestratorService {
live_state_read: None,
ask_liveness_probe: None,
ask_ceiling: ASK_AGENT_CEILING,
background_tasks: None,
}
}
/// Branche le store des tâches de fond pour tracer les rendez-vous inter-agents
/// comme [`BackgroundTaskKind::HeadlessRendezvous`] (B6).
///
/// Le `clock` fournit les timestamps métier de la tâche. Il réutilise le champ
/// horloge applicatif déjà existant ; si un provider de conversation log l'avait
/// déjà posé, la même horloge reste partagée.
#[must_use]
pub fn with_background_tasks(
mut self,
store: Arc<dyn BackgroundTaskStore>,
clock: Arc<dyn Clock>,
) -> Self {
self.background_tasks = Some(store);
self.clock = Some(clock);
self
}
/// Branche la **sonde de vivacité** (signe de vie) du rendez-vous délégué : la borne
/// de tour devient une **fenêtre d'inactivité** réarmée à chaque progrès observé,
/// sous le plafond [`Self::with_ask_ceiling`]. Builder additif (signature de
@ -610,6 +644,162 @@ impl OrchestratorService {
self
}
fn now_ms(&self) -> u64 {
self.clock
.as_ref()
.map_or(0, |clock| clock.now_millis().max(0) as u64)
}
#[allow(clippy::too_many_arguments)]
async fn start_rendezvous_task(
&self,
project: &Project,
owner_agent_id: AgentId,
requester_agent_id: Option<AgentId>,
target_agent_id: AgentId,
ticket_id: TicketId,
conversation_id: domain::conversation::ConversationId,
) -> Result<Option<TaskId>, AppError> {
let Some(store) = &self.background_tasks else {
return Ok(None);
};
let task_id = TaskId::new_random();
let now = self.now_ms();
let task = BackgroundTask::new(
task_id,
project.id,
owner_agent_id,
BackgroundTaskKind::HeadlessRendezvous {
requester_agent_id,
target_agent_id,
ticket_id,
conversation_id,
},
BackgroundTaskWakePolicy::RecordOnly,
now,
None,
)
.map_err(|err| AppError::Internal(err.to_string()))?
.transition(BackgroundTaskState::Running, now)
.map_err(|err| AppError::Internal(err.to_string()))?;
store
.create(&task)
.await
.map_err(|err| AppError::Store(err.to_string()))?;
if let Some(events) = &self.events {
events.publish(DomainEvent::BackgroundTaskStarted {
project_id: project.id,
task_id,
owner_agent_id,
});
}
Ok(Some(task_id))
}
async fn complete_rendezvous_task_success(
&self,
project: &Project,
owner_agent_id: AgentId,
task_id: Option<TaskId>,
reply: &str,
) -> Result<(), AppError> {
let result = BackgroundTaskResult::Success {
finished_at_ms: self.now_ms(),
exit_code: None,
summary: "Headless rendezvous completed with Final".to_owned(),
stdout_tail: Some(bound_task_text(
reply,
domain::BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES,
)),
stderr_tail: None,
};
self.finish_rendezvous_task(project, owner_agent_id, task_id, result)
.await
}
async fn complete_rendezvous_task_failure(
&self,
project: &Project,
owner_agent_id: AgentId,
task_id: Option<TaskId>,
error: impl Into<String>,
) -> Result<(), AppError> {
let result = BackgroundTaskResult::Failure {
finished_at_ms: self.now_ms(),
exit_code: None,
error: bound_task_text(&error.into(), domain::BACKGROUND_TASK_TEXT_MAX_BYTES),
stdout_tail: None,
stderr_tail: None,
};
self.finish_rendezvous_task(project, owner_agent_id, task_id, result)
.await
}
async fn complete_rendezvous_task_cancelled(
&self,
project: &Project,
owner_agent_id: AgentId,
task_id: Option<TaskId>,
reason: impl Into<String>,
) -> Result<(), AppError> {
let result = BackgroundTaskResult::Cancelled {
finished_at_ms: self.now_ms(),
reason: bound_task_text(&reason.into(), domain::BACKGROUND_TASK_TEXT_MAX_BYTES),
};
self.finish_rendezvous_task(project, owner_agent_id, task_id, result)
.await
}
async fn finish_rendezvous_task(
&self,
project: &Project,
owner_agent_id: AgentId,
task_id: Option<TaskId>,
result: BackgroundTaskResult,
) -> 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() {
return Ok(());
}
let event = match &result {
BackgroundTaskResult::Success { .. } => DomainEvent::BackgroundTaskCompleted {
project_id: project.id,
task_id,
owner_agent_id,
},
BackgroundTaskResult::Failure { .. } => DomainEvent::BackgroundTaskFailed {
project_id: project.id,
task_id,
owner_agent_id,
},
BackgroundTaskResult::Cancelled { .. } | BackgroundTaskResult::Expired { .. } => {
DomainEvent::BackgroundTaskCancelled {
project_id: project.id,
task_id,
owner_agent_id,
}
}
};
let completed = task
.complete(result)
.map_err(|err| AppError::Internal(err.to_string()))?;
store
.save(&completed)
.await
.map_err(|err| AppError::Store(err.to_string()))?;
if let Some(events) = &self.events {
events.publish(event);
}
Ok(())
}
/// Branche le [`ConversationRegistry`] (cadrage C3 §5.2) pour résoudre
/// paresseusement le fil `A↔B` (ou `User↔B`) d'un `ask` et séparer les contextes.
/// Builder additif (signature de [`Self::new`] inchangée).
@ -1458,6 +1648,16 @@ impl OrchestratorService {
// 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_id, ticket);
let rendezvous_task = self
.start_rendezvous_task(
project,
agent_id,
requester,
agent_id,
ticket_id,
conversation_id,
)
.await?;
// 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.
@ -1514,6 +1714,8 @@ impl OrchestratorService {
.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} \
(agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}",
@ -1526,6 +1728,33 @@ impl OrchestratorService {
// `cancel_head` + `mark_idle` au Drop. On réconcilie la live-state à `Done`
// pour qu'un abandon ne laisse pas un busy fantôme.
WatchdogOutcome::Resolved(Err(err)) => {
let task_error = if matches!(
err,
AppError::TargetReturnedNoReply(_) | AppError::Process(_)
) {
format!("NoReply: {err}")
} else {
err.to_string()
};
if task_error.to_ascii_lowercase().contains("cancelled")
|| task_error.to_ascii_lowercase().contains("annul")
{
self.complete_rendezvous_task_cancelled(
project,
agent_id,
rendezvous_task,
task_error,
)
.await?;
} else {
self.complete_rendezvous_task_failure(
project,
agent_id,
rendezvous_task,
task_error,
)
.await?;
}
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
crate::diag!(
@ -1537,6 +1766,13 @@ impl OrchestratorService {
}
// Plafond absolu atteint alors que la cible progresse encore : verdict distinct.
WatchdogOutcome::CeilingActive => {
self.complete_rendezvous_task_failure(
project,
agent_id,
rendezvous_task,
format!("Timeout: rendezvous absolute ceiling reached for target {target}"),
)
.await?;
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
crate::diag!(
@ -1549,6 +1785,13 @@ impl OrchestratorService {
// Vrai silence (aucun progrès sur la fenêtre) : timeout typé, identique à
// l'ancienne borne plate. Live-state réconciliée à `Done`.
WatchdogOutcome::NoReply => {
self.complete_rendezvous_task_failure(
project,
agent_id,
rendezvous_task,
format!("Timeout: rendezvous inactivity window expired for target {target}"),
)
.await?;
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
.await;
crate::diag!(