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

fix(backend,orchestrator,chat): l'objectif suspendu guide le premier tour post-restart — #290 regression partielle (register_suspended_turn capture la demande utilisateur exacte au point d'entrée chat agent_send/chat.send avec source diagnostique, RestartRendezvousCoordinator conserve tour/objectif/restart_attempt_id, take_continuation consommable exactement une fois et seulement par la tentative courante: suspended_restart_continuation_prompt place l'objectif utilisateur avant toute consigne generique et StructuredHumanTurnTransition::Resume porte le prompt host-provided injecte au runtime de remplacement; nettoyage du tour suspendu sur succes, echec, annulation et abandon/expiration; diagnostics projet/agent/turn_id/restart_attempt_id/source; rendez-vous inter-agents #289 inchanges; QA verte reconfirmee par Git: cargo check --workspace, backend self_restart_contract_tests 3/3, orchestrator_service restart 10/10, workstate_actions 9/9, app-tauri chat_bridge 31/31, web-server websocket_chat_send 3/3, fmt --check, diff --check; restants: rebuild AppImage + validation live reprise objectif apres restart; carnet #290 iteration regression + passage store qa inclus; scripts/ hors périmètre; aucun push)

Browse Source
This commit is contained in:
Blomios
2026-09-09 09:17:36 +02:00
parent 93a1c23a29
commit 18c1b01afb
10 changed files with 426 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2620,13 +2620,14 @@ pub async fn agent_send(
.collect();
let session_meta = state.structured_sessions.meta_for_session(&sid);
let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| {
backend::StructuredHumanTurnGuard::start(
backend::StructuredHumanTurnGuard::start_with_source(
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,
&prompt,
"agent_send",
)
});
@ -2711,6 +2712,7 @@ pub async fn agent_send(
'logical_turn: loop {
let mut concrete_terminal = false;
let mut resume = false;
let mut resume_prompt = None;
for event in current_stream {
let provider_terminal = matches!(
event,
@ -2722,8 +2724,9 @@ pub async fn agent_send(
if let Some(guard) = turn_guard.as_mut() {
match runtime.block_on(guard.transition_after_provider_terminal_async()) {
backend::StructuredHumanTurnTransition::Complete => {}
backend::StructuredHumanTurnTransition::Resume => {
backend::StructuredHumanTurnTransition::Resume { prompt } => {
resume = true;
resume_prompt = Some(prompt);
break;
}
backend::StructuredHumanTurnTransition::Failed(message) => {
@ -2789,7 +2792,10 @@ pub async fn agent_send(
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::Resume { prompt } => {
resume = true;
resume_prompt = Some(prompt);
}
backend::StructuredHumanTurnTransition::Failed(message) => {
terminal_visible = true;
let _ = bridge.send_output_settled(
@ -2824,8 +2830,9 @@ pub async fn agent_send(
);
break;
};
match runtime.block_on(replacement.send(application::SELF_RESTART_CONTINUATION_PROMPT))
{
let resume_prompt = resume_prompt
.unwrap_or_else(|| application::SELF_RESTART_CONTINUATION_PROMPT.to_owned());
match runtime.block_on(replacement.send(&resume_prompt)) {
Ok(stream) => current_stream = stream,
Err(error) => {
terminal_visible = true;

13
crates/application/src/lib.rs
View File

@ -165,12 +165,13 @@ pub use model_server::{
};
pub use orchestrator::{
resolve_rendezvous_ceiling, resolve_rendezvous_window, run_inactivity_watchdog,
AgentRestartScheduled, AgentSelfRestartScheduler, AgentWakeService, AskLivenessProbe,
ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider,
OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory,
RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousRegistration,
RestartRendezvousState, WakeSessionProvider, WatchdogOutcome, WriteMemory,
DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, SELF_RESTART_CONTINUATION_PROMPT,
suspended_restart_continuation_prompt, AgentRestartScheduled, AgentSelfRestartScheduler,
AgentWakeService, AskLivenessProbe, ContextGuardUseCases, LiveStateProvider,
LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome, OrchestratorService,
ProposeContext, ReadContext, ReadMemory, RecordTurnProvider, RestartContinuation,
RestartRendezvousCoordinator, RestartRendezvousRegistration, RestartRendezvousState,
WakeSessionProvider, WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING,
DEFAULT_RENDEZVOUS_WINDOW, SELF_RESTART_CONTINUATION_PROMPT,
};
pub use permission::{
GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput,

9
crates/application/src/orchestrator/mod.rs
View File

@ -19,9 +19,10 @@ pub use rendezvous::{
WatchdogOutcome, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW,
};
pub use service::{
AgentRestartScheduled, AgentSelfRestartScheduler, AskLivenessProbe, ContextGuardUseCases,
LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome,
OrchestratorService, RecordTurnProvider, RestartRendezvousCoordinator,
RestartRendezvousRegistration, RestartRendezvousState, SELF_RESTART_CONTINUATION_PROMPT,
suspended_restart_continuation_prompt, AgentRestartScheduled, AgentSelfRestartScheduler,
AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider,
McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, RecordTurnProvider,
RestartContinuation, RestartRendezvousCoordinator, RestartRendezvousRegistration,
RestartRendezvousState, SELF_RESTART_CONTINUATION_PROMPT,
};
pub use wake::{AgentWakeService, WakeSessionProvider};

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

@ -87,6 +87,18 @@ pub trait AgentSelfRestartScheduler: Send + Sync {
pub const SELF_RESTART_CONTINUATION_PROMPT: &str =
"IdeA vient de te redémarrer. Reprends immédiatement le travail en cours là où tu l'avais laissé.";
/// Builds the host-provided first instruction for a replacement runtime. The
/// suspended objective is deliberately placed before any generic continuation
/// guidance so the resumed agent cannot substitute unrelated project status.
#[must_use]
pub fn suspended_restart_continuation_prompt(objective_text: &str) -> String {
format!(
"Objectif utilisateur suspendu avant ton redémarrage :\n\n{objective_text}\n\n\
IdeA vient de te redémarrer. Reprends cet objectif immédiatement et réponds à \
l'utilisateur dans ce même tour."
)
}
/// Control-plane state of a self-restart affecting an active logical rendezvous.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RestartRendezvousState {
@ -120,9 +132,30 @@ struct RestartRendezvousEntry {
active_waiters: usize,
activity: u64,
state: RestartRendezvousState,
suspended_turn: Option<SuspendedRestartTurn>,
tx: watch::Sender<RestartRendezvousState>,
}
#[derive(Debug, Clone)]
struct SuspendedRestartTurn {
turn_id: String,
objective_text: String,
source: String,
restart_attempt_id: Option<String>,
injected: bool,
}
/// Host-provided continuation payload for a logical turn suspended by self-restart.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RestartContinuation {
/// Stable id of the logical turn that originally received the user request.
pub turn_id: String,
/// Exact user objective captured at the normal chat entry point.
pub objective_text: String,
/// Entry-point label used for restart diagnostics.
pub source: String,
}
/// Stable, runtime-keyed seam shared by the orchestrator and the detached restart
/// adapter. It keeps a logical ask alive while its concrete provider session is
/// replaced.
@ -152,6 +185,7 @@ impl RestartRendezvousCoordinator {
active_waiters: 0,
activity: 0,
state,
suspended_turn: None,
tx,
}
});
@ -163,6 +197,39 @@ impl RestartRendezvousCoordinator {
}
}
/// Registers a logical human turn and retains its original objective as the
/// application-owned restart handoff. The payload is never reconstructed from
/// transcript or generic project context.
pub fn register_suspended_turn(
self: &Arc<Self>,
key: RuntimeAgentKey,
turn_id: String,
objective_text: String,
source: String,
) -> RestartRendezvousRegistration {
let registration = self.register(key);
let mut entries = self
.entries
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(entry) = entries.get_mut(&key) {
crate::diag!(
"[restart-rendezvous] suspended objective registered: project={} agent={} \
turn_id={turn_id} source={source}",
key.project_id,
key.agent_id,
);
entry.suspended_turn = Some(SuspendedRestartTurn {
turn_id,
objective_text,
source,
restart_attempt_id: None,
injected: false,
});
}
registration
}
/// Announces a restart synchronously. Returns whether an active ask owns the
/// continuation and therefore must receive the post-restart `Final`.
pub fn begin(&self, key: RuntimeAgentKey, restart_id: String) -> bool {
@ -176,6 +243,21 @@ impl RestartRendezvousCoordinator {
if entry.active_waiters == 0 {
return false;
}
if let Some(turn) = entry.suspended_turn.as_mut() {
if turn.restart_attempt_id.as_deref() != Some(restart_id.as_str()) {
turn.restart_attempt_id = Some(restart_id.clone());
turn.injected = false;
}
crate::diag!(
"[restart-rendezvous] restart attached to suspended objective: project={} \
agent={} turn_id={} restart_attempt_id={} source={}",
key.project_id,
key.agent_id,
turn.turn_id,
restart_id,
turn.source,
);
}
entry.activity = entry.activity.saturating_add(1);
entry.state = RestartRendezvousState::Restarting { restart_id };
entry.tx.send_replace(entry.state.clone());
@ -204,10 +286,50 @@ impl RestartRendezvousCoordinator {
return;
}
entry.state = RestartRendezvousState::Idle;
entry.suspended_turn = None;
entry.activity = entry.activity.saturating_add(1);
entry.tx.send_replace(entry.state.clone());
}
/// Takes the suspended objective exactly once for the matching restart attempt.
/// Stale attempts and duplicate consumers cannot obtain a payload.
pub fn take_continuation(
&self,
key: RuntimeAgentKey,
restart_id: &str,
) -> Option<RestartContinuation> {
let mut entries = self
.entries
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let entry = entries.get_mut(&key)?;
if !matches!(
&entry.state,
RestartRendezvousState::Ready { restart_id: current } if current == restart_id
) {
return None;
}
let turn = entry.suspended_turn.as_mut()?;
if turn.restart_attempt_id.as_deref() != Some(restart_id) || turn.injected {
return None;
}
turn.injected = true;
crate::diag!(
"[restart-rendezvous] suspended objective injected: project={} agent={} \
turn_id={} restart_attempt_id={} source={}",
key.project_id,
key.agent_id,
turn.turn_id,
restart_id,
turn.source,
);
Some(RestartContinuation {
turn_id: turn.turn_id.clone(),
objective_text: turn.objective_text.clone(),
source: turn.source.clone(),
})
}
/// Publishes an explicit restart failure to the awaiting logical rendezvous.
pub fn fail(&self, key: RuntimeAgentKey, restart_id: &str, message: String) {
self.transition(key, restart_id, |restart_id| {
@ -233,7 +355,19 @@ impl RestartRendezvousCoordinator {
| RestartRendezvousState::Ready { restart_id } => restart_id.clone(),
_ => return false,
};
if let Some(turn) = entry.suspended_turn.as_ref() {
crate::diag!(
"[restart-rendezvous] suspended objective cancelled: project={} agent={} \
turn_id={} restart_attempt_id={} source={}",
key.project_id,
key.agent_id,
turn.turn_id,
restart_id,
turn.source,
);
}
entry.state = RestartRendezvousState::Cancelled { restart_id };
entry.suspended_turn = None;
entry.activity = entry.activity.saturating_add(1);
entry.tx.send_replace(entry.state.clone());
true
@ -264,8 +398,23 @@ impl RestartRendezvousCoordinator {
if !matches {
return;
}
let next_state = next(restart_id.to_owned());
if matches!(next_state, RestartRendezvousState::Failed { .. }) {
if let Some(turn) = entry.suspended_turn.as_ref() {
crate::diag!(
"[restart-rendezvous] suspended objective failed: project={} agent={} \
turn_id={} restart_attempt_id={} source={}",
key.project_id,
key.agent_id,
turn.turn_id,
restart_id,
turn.source,
);
}
entry.suspended_turn = None;
}
entry.activity = entry.activity.saturating_add(1);
entry.state = next(restart_id.to_owned());
entry.state = next_state;
entry.tx.send_replace(entry.state.clone());
}
@ -379,6 +528,11 @@ impl RestartRendezvousRegistration {
self.rx.changed().await
}
/// Takes the application-owned suspended objective for this exact attempt.
pub fn take_continuation(&self, restart_id: &str) -> Option<RestartContinuation> {
self.coordinator.take_continuation(self.key, restart_id)
}
/// Acknowledges ownership of the replacement session and releases the
/// detached restart scheduler.
pub fn accept_ready(&self, restart_id: &str) {
@ -404,7 +558,19 @@ impl Drop for RestartRendezvousRegistration {
_ => None,
};
if let Some(restart_id) = restart_id {
if let Some(turn) = entry.suspended_turn.as_ref() {
crate::diag!(
"[restart-rendezvous] suspended objective abandoned: project={} agent={} \
turn_id={} restart_attempt_id={} source={}",
self.key.project_id,
self.key.agent_id,
turn.turn_id,
restart_id,
turn.source,
);
}
entry.state = RestartRendezvousState::Cancelled { restart_id };
entry.suspended_turn = None;
entry.activity = entry.activity.saturating_add(1);
entry.tx.send_replace(entry.state.clone());
} else if matches!(

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

@ -5770,6 +5770,96 @@ async fn direct_human_turn_registration_surfaces_restart_failure() {
);
}
#[tokio::test]
async fn suspended_human_objective_is_bound_to_latest_attempt_and_taken_once() {
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let key = rkey(1);
let turn = coordinator.register_suspended_turn(
key,
"human-turn-290".to_owned(),
"redémarre puis confirme que tu as redémarré".to_owned(),
"agent_send".to_owned(),
);
assert!(coordinator.begin(key, "attempt-old".to_owned()));
assert!(coordinator.begin(key, "attempt-current".to_owned()));
coordinator.ready(key, "attempt-old");
assert!(
turn.take_continuation("attempt-old").is_none(),
"an older restart attempt cannot inject its handoff"
);
coordinator.ready(key, "attempt-current");
let continuation = turn
.take_continuation("attempt-current")
.expect("the current attempt owns the suspended objective");
assert_eq!(continuation.turn_id, "human-turn-290");
assert_eq!(
continuation.objective_text,
"redémarre puis confirme que tu as redémarré"
);
assert_eq!(continuation.source, "agent_send");
assert!(
turn.take_continuation("attempt-current").is_none(),
"the same restart attempt cannot inject twice"
);
}
#[tokio::test]
async fn dropping_suspended_turn_during_restart_cleans_handoff_and_unblocks_scheduler() {
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let key = rkey(1);
let turn = coordinator.register_suspended_turn(
key,
"human-turn-expired".to_owned(),
"objectif expiré".to_owned(),
"agent_send".to_owned(),
);
let restart_id = "attempt-expired";
assert!(coordinator.begin(key, restart_id.to_owned()));
drop(turn);
let error = coordinator
.wait_for_handoff(key, restart_id)
.await
.expect_err("dropping the bounded turn cancels and unblocks restart");
assert!(error.contains("cancelled"));
coordinator.ready(key, restart_id);
assert!(
coordinator.take_continuation(key, restart_id).is_none(),
"an expired objective is cleaned and cannot be injected"
);
coordinator.finish(key, restart_id);
}
#[tokio::test]
async fn cancelling_suspended_human_restart_cleans_handoff_and_unblocks_scheduler() {
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let key = rkey(1);
let turn = coordinator.register_suspended_turn(
key,
"human-turn-cancelled".to_owned(),
"objectif annulé".to_owned(),
"agent_send".to_owned(),
);
let restart_id = "attempt-cancelled";
assert!(coordinator.begin(key, restart_id.to_owned()));
assert!(coordinator.cancel(key));
let error = coordinator
.wait_for_handoff(key, restart_id)
.await
.expect_err("cancel must unblock the detached restart scheduler");
assert!(error.contains("cancelled"));
assert!(
turn.take_continuation(restart_id).is_none(),
"cancelled objective cannot reach a replacement runtime"
);
drop(turn);
coordinator.finish(key, restart_id);
}
/// Régression #287 : la sonde transcript peut être stable (notamment inexistante pour
/// un provider non-Claude), tandis que `send_with_tap` publie réellement l'activité du
/// tour. Chaque événement non terminal doit réarmer la fenêtre jusqu'au `Final` tardif.

51
crates/backend/src/lib.rs
View File

@ -1131,7 +1131,9 @@ mod self_restart_contract_tests {
coordinator.ready(key, &restart_id);
assert_eq!(
turn.transition_after_provider_terminal_async().await,
StructuredHumanTurnTransition::Resume,
StructuredHumanTurnTransition::Resume {
prompt: application::suspended_restart_continuation_prompt("restart me"),
},
"the old provider terminal is a control transition, not the human Final"
);
coordinator
@ -1453,7 +1455,10 @@ pub enum StructuredHumanTurnTransition {
/// No self-restart owns this terminal event; it is authoritative for the UI.
Complete,
/// The old terminal event is superseded; continue on the replacement runtime.
Resume,
Resume {
/// Host-provided first prompt for the replacement runtime.
prompt: String,
},
/// Replacement failed explicitly.
Failed(String),
/// The user cancelled the logical turn while replacement was in flight.
@ -1470,6 +1475,29 @@ impl StructuredHumanTurnGuard {
project_id: ProjectId,
agent_id: AgentId,
prompt: &str,
) -> Self {
Self::start_with_source(
input,
mailbox,
restart_rendezvous,
project_id,
agent_id,
prompt,
"direct_human_chat",
)
}
/// Starts a direct human turn while retaining its exact restart objective and
/// diagnostic entry-point source in application state.
#[must_use]
pub fn start_with_source(
input: Arc<dyn domain::input::InputMediator>,
mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
restart_rendezvous: Arc<application::RestartRendezvousCoordinator>,
project_id: ProjectId,
agent_id: AgentId,
prompt: &str,
source: &str,
) -> Self {
let agent = domain::RuntimeAgentKey::new(project_id, agent_id);
let ticket_id = domain::mailbox::TicketId::new_random();
@ -1486,7 +1514,14 @@ impl StructuredHumanTurnGuard {
agent,
ticket_id,
armed: Arc::new(std::sync::atomic::AtomicBool::new(owns_turn)),
restart: owns_turn.then(|| restart_rendezvous.register(agent)),
restart: owns_turn.then(|| {
restart_rendezvous.register_suspended_turn(
agent,
ticket_id.to_string(),
prompt.to_owned(),
source.to_owned(),
)
}),
finish_on_drop: true,
}
}
@ -1527,8 +1562,16 @@ impl StructuredHumanTurnGuard {
}
}
application::RestartRendezvousState::Ready { restart_id } => {
let prompt = restart.take_continuation(&restart_id).map_or_else(
|| application::SELF_RESTART_CONTINUATION_PROMPT.to_owned(),
|continuation| {
application::suspended_restart_continuation_prompt(
&continuation.objective_text,
)
},
);
restart.accept_ready(&restart_id);
return StructuredHumanTurnTransition::Resume;
return StructuredHumanTurnTransition::Resume { prompt };
}
application::RestartRendezvousState::Failed { message, .. } => {
return StructuredHumanTurnTransition::Failed(message);

16
crates/web-server/src/lib.rs
View File

@ -1879,13 +1879,14 @@ async fn ws_send_chat(
})?;
let session_meta = state.app.structured_sessions.meta_for_session(&sid);
let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| {
backend::StructuredHumanTurnGuard::start(
backend::StructuredHumanTurnGuard::start_with_source(
Arc::clone(&state.app.turn_watch_input),
Arc::clone(&state.app.input_mailbox),
Arc::clone(&state.app.restart_rendezvous),
*project_id,
*agent_id,
&prompt,
"websocket_chat_send",
)
});
@ -1935,6 +1936,7 @@ async fn ws_send_chat(
'logical_turn: loop {
let mut concrete_terminal = false;
let mut resume = false;
let mut resume_prompt = None;
for event in stream {
let provider_terminal =
matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. });
@ -1943,8 +1945,9 @@ async fn ws_send_chat(
if let Some(guard) = turn_guard.as_mut() {
match guard.transition_after_provider_terminal_async().await {
backend::StructuredHumanTurnTransition::Complete => {}
backend::StructuredHumanTurnTransition::Resume => {
backend::StructuredHumanTurnTransition::Resume { prompt } => {
resume = true;
resume_prompt = Some(prompt);
break;
}
backend::StructuredHumanTurnTransition::Failed(message) => {
@ -1995,7 +1998,10 @@ async fn ws_send_chat(
if let Some(guard) = turn_guard.as_mut() {
match guard.transition_after_provider_terminal_async().await {
backend::StructuredHumanTurnTransition::Complete => {}
backend::StructuredHumanTurnTransition::Resume => resume = true,
backend::StructuredHumanTurnTransition::Resume { prompt } => {
resume = true;
resume_prompt = Some(prompt);
}
backend::StructuredHumanTurnTransition::Failed(message) => {
let chunk = reply_event_chunk(ReplyEvent::Error { message });
tx.send(ServerFrame::chat_output(
@ -2028,8 +2034,10 @@ async fn ws_send_chat(
code: "CHAT_RESTART_FAILED".to_owned(),
message: "restart completed without a replacement structured session".to_owned(),
})?;
let resume_prompt = resume_prompt
.unwrap_or_else(|| application::SELF_RESTART_CONTINUATION_PROMPT.to_owned());
stream = replacement
.send(application::SELF_RESTART_CONTINUATION_PROMPT)
.send(&resume_prompt)
.await
.map_err(|err| ErrorDto {
code: "CHAT_RESTART_FAILED".to_owned(),