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

fix(backend,orchestrator,chat): tour humain agent_send survit au self-restart — #290 (le tour humain direct agent_send/ws chat.send s'enregistre au RestartRendezvousCoordinator via StructuredHumanTurnGuard: Final/Error de l'ancien provider devient transition de contrôle et non le Final logique; transition_after_provider_terminal_async attend Ready puis accept_ready acquitte le handoff et libère le scheduler, Resume => drain de la session de remplacement avec SELF_RESTART_CONTINUATION_PROMPT, seul le terminal du remplacement est transmis au chat; Failed/Cancelled surfacés explicitement avec libération Busy/FIFO; send_with_tap en erreur pendant restart actif traité comme teardown attendu (stream vide puis attente du remplacement, plus de Final prématuré ni chat bloqué); StopLiveAgent câblé with_restart_rendezvous: stop utilisateur annule le rendez-vous restart; RestartRendezvousRegistration rendu public (state/changed/accept_ready); QA verte: backend self_restart_contract_tests 3/3 dont 2 nouveaux (old terminal ignoré + handoff accepté, failure explicite + busy libéré), application orchestrator_service restart 8/8 dont registration directe du rendez-vous, workstate_actions 9/9, app-tauri chat_bridge 31/31, web-server websocket_chat_send 3/3, cargo check --workspace, fmt --check, diff --check — reconfirmés par Git avant commit; réserves hors périmètre préexistantes: 7 échecs fixture PTY web-server + 2 échecs plugin application; carnet #290 + clôture store #289 inclus; scripts/ hors périmètre; aucun push)

Browse Source
This commit is contained in:
Blomios
2026-09-09 08:38:29 +02:00
parent 321ebdaab9
commit d03300a20f
16 changed files with 660 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
crates/app-tauri/src/commands.rs
View File

@ -2570,8 +2570,10 @@ pub async fn managed_temp_gc(
/// structured registry, open the turn stream (`session.send`), then drain that
/// blocking [`ReplyStream`] on a dedicated OS thread, mapping each
/// [`ReplyEvent`](domain::ports::ReplyEvent) to a [`ReplyChunk`] and forwarding it
/// through the [`ChatBridge`]. The turn ends deterministically at
/// [`ReplyChunk::Final`]; the stream is bounded and closes right after.
/// through the [`ChatBridge`]. The logical turn ends deterministically at the
/// authoritative [`ReplyChunk::Final`]. If the agent self-restarts, the old
/// provider's terminal event is a control transition: the pump waits for the
/// replacement session and only forwards its terminal outcome.
///
/// The channel is (re-)registered each call, bumping the bridge **generation** so
/// a previous attach's pump can no longer deliver (generation supersede): only the
@ -2621,6 +2623,7 @@ pub async fn agent_send(
backend::StructuredHumanTurnGuard::start(
std::sync::Arc::clone(&state.turn_watch_input),
std::sync::Arc::clone(&state.input_mailbox),
std::sync::Arc::clone(&state.restart_rendezvous),
*project_id,
*agent_id,
&prompt_for_model,
@ -2673,7 +2676,20 @@ pub async fn agent_send(
// still live, so we keep the attach and surface the error.
let stream_result = session.send_with_tap(&prompt_for_model, tap_tx).await;
let _ = tap_pump.join();
let stream = stream_result.map_err(|e| ErrorDto::from(AppError::from(e)))?;
let mut turn_guard = turn_guard;
let stream = match stream_result {
Ok(stream) => stream,
Err(_error)
if turn_guard
.as_mut()
.is_some_and(|guard| guard.restart_is_active()) =>
{
// Teardown of the old provider is expected once self-restart owns the
// logical turn. The detached drain below waits for the replacement.
Box::new(std::iter::empty()) as domain::ports::ReplyStream
}
Err(error) => return Err(ErrorDto::from(AppError::from(error))),
};
// Drain the blocking reply iterator on a dedicated OS thread (the stream is a
// synchronous `Iterator`, exactly like the PTY byte stream). It runs to the
@ -2685,52 +2701,144 @@ pub async fn agent_send(
// event then feeds the service (détecter→planifier). Without a conversation id the
// service must not pretend an automatic resume is armed.
let service = std::sync::Arc::clone(&state.session_limit_service);
let structured = std::sync::Arc::clone(&state.structured_sessions);
let runtime = tokio::runtime::Handle::current();
let meta = session_meta;
std::thread::spawn(move || {
let mut terminal_visible = false;
let mut settler = crate::chat::ProgressSettler::default();
for event in stream {
if !matches!(event, domain::ports::ReplyEvent::Final { .. }) {
if let Some(guard) = &turn_guard {
guard.mark_alive();
}
}
// Non-terminal, sans contenu chat : un signal de limite alimente le service
// (le badge UI vient du bus `AgentRateLimited`, pas du flux chat) puis on
// continue à drainer comme pour un battement.
if let domain::ports::ReplyEvent::RateLimited { resets_at_ms } = &event {
if let Some((project_id, agent_id, node_id, conversation_id)) = &meta {
let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms) {
(Some(conversation_id), resets_at_ms) => {
(Some(conversation_id.clone()), *resets_at_ms)
let mut current_stream = stream;
'logical_turn: loop {
let mut concrete_terminal = false;
let mut resume = false;
for event in current_stream {
let provider_terminal = matches!(
event,
domain::ports::ReplyEvent::Final { .. }
| domain::ports::ReplyEvent::Error { .. }
);
if provider_terminal {
concrete_terminal = true;
if let Some(guard) = turn_guard.as_mut() {
match runtime.block_on(guard.transition_after_provider_terminal_async()) {
backend::StructuredHumanTurnTransition::Complete => {}
backend::StructuredHumanTurnTransition::Resume => {
resume = true;
break;
}
backend::StructuredHumanTurnTransition::Failed(message) => {
terminal_visible = true;
let _ = bridge.send_output_settled(
&sid,
&mut settler,
ReplyChunk::Error { message },
);
break 'logical_turn;
}
backend::StructuredHumanTurnTransition::Cancelled => {
terminal_visible = true;
break 'logical_turn;
}
}
(None, None) => (None, None),
// Reset connu mais conversation moteur absente : ne pas armer
// une reprise non reprenable ; surfacer le fallback humain.
(None, Some(_)) => (None, None),
};
service.on_rate_limited(
*project_id,
*agent_id,
*node_id,
conversation_id,
resets_at_ms,
);
}
} else {
if let Some(guard) = &turn_guard {
guard.mark_alive();
}
}
// Non-terminal, sans contenu chat : un signal de limite alimente le service
// (le badge UI vient du bus `AgentRateLimited`, pas du flux chat) puis on
// continue à drainer comme pour un battement.
if let domain::ports::ReplyEvent::RateLimited { resets_at_ms } = &event {
if let Some((project_id, agent_id, node_id, conversation_id)) = &meta {
let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms)
{
(Some(conversation_id), resets_at_ms) => {
(Some(conversation_id.clone()), *resets_at_ms)
}
(None, None) => (None, None),
// Reset connu mais conversation moteur absente : ne pas armer
// une reprise non reprenable ; surfacer le fallback humain.
(None, Some(_)) => (None, None),
};
service.on_rate_limited(
*project_id,
*agent_id,
*node_id,
conversation_id,
resets_at_ms,
);
}
}
// Heartbeats carry no chat content (readiness/heartbeat lot 1) ⇒ no wire
// chunk; skip them while still draining so the turn runs to its `Final`.
let Some(chunk) = crate::chat::chunk_from_event(event) else {
continue;
};
if matches!(chunk, ReplyChunk::Final { .. } | ReplyChunk::Error { .. }) {
terminal_visible = true;
}
// `send_output` always records into the conversation scrollback; the
// boolean only reflects live delivery. If the view navigated away
// (no channel at this generation) we keep draining so the turn still
// completes and the scrollback stays whole for the next re-attach.
let _ = bridge.send_output_settled(&sid, &mut settler, chunk);
}
if !resume && !concrete_terminal {
if let Some(guard) = turn_guard.as_mut() {
match runtime.block_on(guard.transition_after_provider_terminal_async()) {
backend::StructuredHumanTurnTransition::Complete => {}
backend::StructuredHumanTurnTransition::Resume => resume = true,
backend::StructuredHumanTurnTransition::Failed(message) => {
terminal_visible = true;
let _ = bridge.send_output_settled(
&sid,
&mut settler,
ReplyChunk::Error { message },
);
}
backend::StructuredHumanTurnTransition::Cancelled => {
terminal_visible = true;
}
}
}
}
// Heartbeats carry no chat content (readiness/heartbeat lot 1) ⇒ no wire
// chunk; skip them while still draining so the turn runs to its `Final`.
let Some(chunk) = crate::chat::chunk_from_event(event) else {
continue;
};
if matches!(chunk, ReplyChunk::Final { .. } | ReplyChunk::Error { .. }) {
terminal_visible = true;
if !resume {
break;
}
let Some((project_id, agent_id, _, _)) = meta.as_ref() else {
break;
};
let Some(replacement) = structured.session_for_agent_in_project(*project_id, agent_id)
else {
terminal_visible = true;
let _ = bridge.send_output_settled(
&sid,
&mut settler,
ReplyChunk::Error {
message: "Le redémarrage n'a pas fourni de session de remplacement."
.to_owned(),
},
);
break;
};
match runtime.block_on(replacement.send(application::SELF_RESTART_CONTINUATION_PROMPT))
{
Ok(stream) => current_stream = stream,
Err(error) => {
terminal_visible = true;
let _ = bridge.send_output_settled(
&sid,
&mut settler,
ReplyChunk::Error {
message: error.to_string(),
},
);
break;
}
}
// `send_output` always records into the conversation scrollback; the
// boolean only reflects live delivery. If the view navigated away
// (no channel at this generation) we keep draining so the turn still
// completes and the scrollback stays whole for the next re-attach.
let _ = bridge.send_output_settled(&sid, &mut settler, chunk);
}
if !terminal_visible {
let chunk = crate::chat::unfinished_turn_chunk();