fix(orchestrator,agent): ask_agent survit au restart de la cible same-project et cross-project — #289 (RestartRendezvousCoordinator stable par RuntimeAgentKey (projectId, agentId) partagé entre le rendez-vous ask_structured et le scheduler AppAgentSelfRestarter: l'ask actif garde verrou de tour, ticket logique et conversation_id, abandonne le drain de l'ancienne session sans no-reply puis capture le Final de continuation de la session relancée (SELF_RESTART_CONTINUATION_PROMPT partagé, un seul tour de continuation); branche delegated du scheduler: pas de preempt/cancel_head/mark_idle locaux, ready + wait_for_handoff, échecs restart propagés explicitement et cancel en cascade sans relance de continuation; watchdog d'inactivité réarmé par l'activité de contrôle restart; corrélation cross-project (requesterProjectId, requesterAgentId, correlationId) conservée jusqu'au résultat post-restart puis supprimée; stop_agent d'une cible en restart annule le rendez-vous au lieu d'un NotFound; le scheduler standalone conserve son comportement historique sans ask actif; QA verte: orchestrator_service 87/87 dont 6 tests restart (final post-restart, nettoyage busy/workstate/registre, corrélation cross-project, failure explicite, cancel cross-project + stop same-project pendant restart), application lib 171/171 dont watchdog réarmé, backend 89+7 ignored, cargo check -p backend, fmt --check; réserves: pas d'E2E AppImage/vrai provider, 7 tests backend ignorés bind socket local; scripts/ hors périmètre; aucun push)
This commit is contained in:
@ -2022,6 +2022,7 @@ struct AskFixture {
|
||||
/// assert the target's Working/Done transitions and the upsert count.
|
||||
live: Arc<RecordingLiveState>,
|
||||
structured: Arc<StructuredSessions>,
|
||||
restart_rendezvous: Arc<application::RestartRendezvousCoordinator>,
|
||||
}
|
||||
|
||||
struct CompletionSession {
|
||||
@ -2198,6 +2199,7 @@ fn ask_fixture_full(
|
||||
let bus = SpyBus::default();
|
||||
let mailbox = Arc::new(TestMailbox::new());
|
||||
let structured = Arc::new(StructuredSessions::new());
|
||||
let restart_rendezvous = Arc::new(application::RestartRendezvousCoordinator::new());
|
||||
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
|
||||
seed_structured_sessions(&structured, &contexts, mailbox.completions());
|
||||
let memories = Arc::new(HarvestMemories {
|
||||
@ -2270,6 +2272,7 @@ fn ask_fixture_full(
|
||||
.with_conversations(conversations)
|
||||
.with_events(Arc::new(bus.clone()))
|
||||
.with_structured(Arc::clone(&structured))
|
||||
.with_restart_rendezvous(Arc::clone(&restart_rendezvous))
|
||||
.with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new(
|
||||
Arc::clone(&memories) as Arc<dyn MemoryStore>,
|
||||
Arc::new(bus.clone()),
|
||||
@ -2302,6 +2305,7 @@ fn ask_fixture_full(
|
||||
memories,
|
||||
live,
|
||||
structured,
|
||||
restart_rendezvous,
|
||||
}
|
||||
}
|
||||
|
||||
@ -4412,6 +4416,7 @@ fn ask_fixture_with_record(
|
||||
let bus = SpyBus::default();
|
||||
let mailbox = Arc::new(TestMailbox::new());
|
||||
let structured = Arc::new(StructuredSessions::new());
|
||||
let restart_rendezvous = Arc::new(application::RestartRendezvousCoordinator::new());
|
||||
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
|
||||
seed_structured_sessions(&structured, &contexts, mailbox.completions());
|
||||
|
||||
@ -4473,6 +4478,7 @@ fn ask_fixture_with_record(
|
||||
.with_conversations(conversations)
|
||||
.with_events(Arc::new(bus.clone()))
|
||||
.with_structured(Arc::clone(&structured))
|
||||
.with_restart_rendezvous(Arc::clone(&restart_rendezvous))
|
||||
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
|
||||
);
|
||||
|
||||
@ -4489,6 +4495,7 @@ fn ask_fixture_with_record(
|
||||
// Live-state likewise unwired here (RecordTurn-focused fixtures).
|
||||
live: Arc::new(RecordingLiveState::default()),
|
||||
structured,
|
||||
restart_rendezvous,
|
||||
}
|
||||
}
|
||||
|
||||
@ -4868,6 +4875,106 @@ struct FailingStructuredSession {
|
||||
message: String,
|
||||
}
|
||||
|
||||
struct RestartingStructuredSession {
|
||||
id: SessionId,
|
||||
project_id: ProjectId,
|
||||
agent_id: AgentId,
|
||||
coordinator: Arc<application::RestartRendezvousCoordinator>,
|
||||
structured: Arc<StructuredSessions>,
|
||||
replacement: Arc<dyn AgentSession>,
|
||||
delay: Duration,
|
||||
failure: Option<String>,
|
||||
began: Arc<std::sync::atomic::AtomicBool>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AgentSession for RestartingStructuredSession {
|
||||
fn id(&self) -> SessionId {
|
||||
self.id
|
||||
}
|
||||
|
||||
fn conversation_id(&self) -> Option<String> {
|
||||
Some("stable-engine-conversation".to_owned())
|
||||
}
|
||||
|
||||
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||
let key = domain::RuntimeAgentKey::new(self.project_id, self.agent_id);
|
||||
let restart_id = format!("restart-{}", self.id);
|
||||
assert!(
|
||||
self.coordinator.begin(key, restart_id.clone()),
|
||||
"the logical rendezvous must be registered before provider send"
|
||||
);
|
||||
self.began.store(true, Ordering::SeqCst);
|
||||
let coordinator = Arc::clone(&self.coordinator);
|
||||
let structured = Arc::clone(&self.structured);
|
||||
let replacement = Arc::clone(&self.replacement);
|
||||
let replacement_project_id = self.project_id;
|
||||
let old_id = self.id;
|
||||
let delay = self.delay;
|
||||
let failure = self.failure.clone();
|
||||
structured.remove(&old_id);
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(delay).await;
|
||||
if let Some(message) = failure {
|
||||
coordinator.fail(key, &restart_id, message);
|
||||
return;
|
||||
}
|
||||
if coordinator.is_cancelled(key, &restart_id) {
|
||||
return;
|
||||
}
|
||||
structured.insert_in_project(
|
||||
replacement_project_id,
|
||||
replacement,
|
||||
key.agent_id,
|
||||
nid(799),
|
||||
);
|
||||
coordinator.ready(key, &restart_id);
|
||||
});
|
||||
Err(AgentSessionError::Io(
|
||||
"old provider session closed during restart".to_owned(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct CapturingFinalSession {
|
||||
id: SessionId,
|
||||
prompts: Arc<Mutex<Vec<String>>>,
|
||||
final_text: String,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AgentSession for CapturingFinalSession {
|
||||
fn id(&self) -> SessionId {
|
||||
self.id
|
||||
}
|
||||
|
||||
fn conversation_id(&self) -> Option<String> {
|
||||
Some("stable-engine-conversation".to_owned())
|
||||
}
|
||||
|
||||
async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||
self.prompts.lock().unwrap().push(prompt.to_owned());
|
||||
Ok(Box::new(
|
||||
vec![ReplyEvent::Final {
|
||||
content: self.final_text.clone(),
|
||||
}]
|
||||
.into_iter(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AgentSession for FailingStructuredSession {
|
||||
fn id(&self) -> SessionId {
|
||||
@ -5329,6 +5436,286 @@ async fn ask_warm_structured_target_reuses_session_without_relaunch() {
|
||||
);
|
||||
}
|
||||
|
||||
fn install_restarting_session(
|
||||
fx: &AskFixture,
|
||||
project_id: ProjectId,
|
||||
agent_id: AgentId,
|
||||
delay: Duration,
|
||||
failure: Option<&str>,
|
||||
) -> (Arc<Mutex<Vec<String>>>, Arc<std::sync::atomic::AtomicBool>) {
|
||||
if let Some(existing) = fx
|
||||
.structured
|
||||
.session_for_agent_in_project(project_id, &agent_id)
|
||||
{
|
||||
fx.structured.remove(&existing.id());
|
||||
}
|
||||
let prompts = Arc::new(Mutex::new(Vec::new()));
|
||||
let began = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
let replacement = Arc::new(CapturingFinalSession {
|
||||
id: sid(9951),
|
||||
prompts: Arc::clone(&prompts),
|
||||
final_text: "final after restart".to_owned(),
|
||||
}) as Arc<dyn AgentSession>;
|
||||
fx.structured.insert_in_project(
|
||||
project_id,
|
||||
Arc::new(RestartingStructuredSession {
|
||||
id: sid(9950),
|
||||
project_id,
|
||||
agent_id,
|
||||
coordinator: Arc::clone(&fx.restart_rendezvous),
|
||||
structured: Arc::clone(&fx.structured),
|
||||
replacement,
|
||||
delay,
|
||||
failure: failure.map(str::to_owned),
|
||||
began: Arc::clone(&began),
|
||||
}),
|
||||
agent_id,
|
||||
nid(798),
|
||||
);
|
||||
(prompts, began)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ask_target_restart_keeps_logical_turn_and_returns_post_restart_final() {
|
||||
let target = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||
let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona"));
|
||||
let (prompts, _) = install_restarting_session(
|
||||
&fx,
|
||||
project_id(),
|
||||
target.id,
|
||||
Duration::from_millis(20),
|
||||
None,
|
||||
);
|
||||
|
||||
let out = fx
|
||||
.service
|
||||
.dispatch(&project(), cmd(ASK_JSON))
|
||||
.await
|
||||
.expect("restart-aware ask must return the replacement session Final");
|
||||
|
||||
assert_eq!(out.reply.as_deref(), Some("final after restart"));
|
||||
assert_eq!(
|
||||
prompts.lock().unwrap().as_slice(),
|
||||
[application::SELF_RESTART_CONTINUATION_PROMPT],
|
||||
"the replacement runtime receives one continuation turn, not the original task twice"
|
||||
);
|
||||
assert_eq!(
|
||||
fx.mailbox.pending(&target.id),
|
||||
0,
|
||||
"logical ticket is retired"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry() {
|
||||
let target = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||
let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona"));
|
||||
install_restarting_session(
|
||||
&fx,
|
||||
project_id(),
|
||||
target.id,
|
||||
Duration::from_millis(20),
|
||||
None,
|
||||
);
|
||||
|
||||
let out = fx
|
||||
.service
|
||||
.dispatch(&project(), cmd(ASK_JSON))
|
||||
.await
|
||||
.expect("restart-aware ask must complete");
|
||||
|
||||
assert_eq!(out.reply.as_deref(), Some("final after restart"));
|
||||
assert_eq!(
|
||||
fx.mediator.busy_state(rkey(1)),
|
||||
AgentBusyState::Idle,
|
||||
"the logical target is not left busy after the post-restart Final"
|
||||
);
|
||||
assert_eq!(fx.mailbox.pending(&target.id), 0, "FIFO is empty");
|
||||
let entry = fx
|
||||
.live
|
||||
.entry_for(target.id)
|
||||
.expect("target live-state entry is written");
|
||||
assert_eq!(entry.status, WorkStatus::Done);
|
||||
assert_eq!(entry.last_delegation, entry.ticket);
|
||||
assert!(
|
||||
fx.structured.remove(&sid(9950)).is_none(),
|
||||
"old provider session was removed during restart"
|
||||
);
|
||||
let current = fx
|
||||
.structured
|
||||
.session_for_agent_in_project(project_id(), &target.id)
|
||||
.expect("replacement session remains registered for the same logical agent");
|
||||
assert_eq!(current.id(), sid(9951));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cross_project_ask_preserves_correlation_across_target_restart() {
|
||||
let target = scratch_agent(aid(1), "architect", "agents/architect.md").with_contactable(true);
|
||||
let requester = scratch_agent(aid(99), "requester", "agents/requester.md");
|
||||
let contexts = FakeContexts::with_agent(&target, "# target");
|
||||
contexts.add_agent(&requester, "# requester");
|
||||
let fx = ask_fixture(contexts);
|
||||
let target_project = other_project();
|
||||
let (prompts, _) = install_restarting_session(
|
||||
&fx,
|
||||
target_project.id,
|
||||
target.id,
|
||||
Duration::from_millis(20),
|
||||
None,
|
||||
);
|
||||
|
||||
let out = fx
|
||||
.service
|
||||
.ask_project_agent_for_mcp(
|
||||
target_project.id,
|
||||
target.id,
|
||||
"cross-project work".to_owned(),
|
||||
project_id(),
|
||||
requester.id,
|
||||
"corr-restart".to_owned(),
|
||||
)
|
||||
.await
|
||||
.expect("cross-project rendezvous survives target restart");
|
||||
|
||||
assert_eq!(out.reply.as_deref(), Some("final after restart"));
|
||||
assert_eq!(
|
||||
prompts.lock().unwrap().as_slice(),
|
||||
[application::SELF_RESTART_CONTINUATION_PROMPT]
|
||||
);
|
||||
let cleaned = fx
|
||||
.service
|
||||
.cancel_project_agent_for_mcp(
|
||||
target_project.id,
|
||||
target.id,
|
||||
project_id(),
|
||||
requester.id,
|
||||
"corr-restart".to_owned(),
|
||||
)
|
||||
.await
|
||||
.expect_err("completed cross-project correlation must be removed");
|
||||
assert_eq!(cleaned.code(), "NOT_FOUND");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ask_target_restart_failure_is_explicit_not_no_reply() {
|
||||
let target = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||
let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona"));
|
||||
install_restarting_session(
|
||||
&fx,
|
||||
project_id(),
|
||||
target.id,
|
||||
Duration::from_millis(10),
|
||||
Some("replacement launch exploded"),
|
||||
);
|
||||
|
||||
let err = fx
|
||||
.service
|
||||
.dispatch(&project(), cmd(ASK_JSON))
|
||||
.await
|
||||
.expect_err("restart failure must terminate the awaiting ask explicitly");
|
||||
|
||||
assert_eq!(err.code(), "PROCESS");
|
||||
assert!(err.to_string().contains("replacement launch exploded"));
|
||||
assert!(!err.to_string().contains("structured Final response"));
|
||||
assert_eq!(fx.mailbox.pending(&target.id), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cross_project_cancel_during_restart_releases_ask_and_correlation() {
|
||||
let target = scratch_agent(aid(1), "architect", "agents/architect.md").with_contactable(true);
|
||||
let requester = scratch_agent(aid(99), "requester", "agents/requester.md");
|
||||
let contexts = FakeContexts::with_agent(&target, "# target");
|
||||
contexts.add_agent(&requester, "# requester");
|
||||
let fx = ask_fixture(contexts);
|
||||
let target_project = other_project();
|
||||
let (_, began) = install_restarting_session(
|
||||
&fx,
|
||||
target_project.id,
|
||||
target.id,
|
||||
Duration::from_secs(5),
|
||||
None,
|
||||
);
|
||||
let service = Arc::clone(&fx.service);
|
||||
let ask = tokio::spawn(async move {
|
||||
service
|
||||
.ask_project_agent_for_mcp(
|
||||
target_project.id,
|
||||
target.id,
|
||||
"cross-project work".to_owned(),
|
||||
project_id(),
|
||||
requester.id,
|
||||
"corr-cancel-restart".to_owned(),
|
||||
)
|
||||
.await
|
||||
});
|
||||
await_until(|| began.load(Ordering::SeqCst)).await;
|
||||
|
||||
fx.service
|
||||
.cancel_project_agent_for_mcp(
|
||||
other_project().id,
|
||||
aid(1),
|
||||
project_id(),
|
||||
aid(99),
|
||||
"corr-cancel-restart".to_owned(),
|
||||
)
|
||||
.await
|
||||
.expect("cancel routes through the stable cross-project correlation");
|
||||
let err = ask
|
||||
.await
|
||||
.expect("ask task joins")
|
||||
.expect_err("cancelled restart-aware ask must fail");
|
||||
assert!(err.to_string().to_ascii_lowercase().contains("cancel"));
|
||||
assert_eq!(fx.mailbox.pending(&aid(1)), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session() {
|
||||
let target = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||
let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona"));
|
||||
let (_, began) =
|
||||
install_restarting_session(&fx, project_id(), target.id, Duration::from_secs(5), None);
|
||||
let service = Arc::clone(&fx.service);
|
||||
let ask = tokio::spawn(async move { service.dispatch(&project(), cmd(ASK_JSON)).await });
|
||||
await_until(|| began.load(Ordering::SeqCst)).await;
|
||||
|
||||
let stopped = fx
|
||||
.service
|
||||
.dispatch(
|
||||
&project(),
|
||||
cmd(r#"{ "action":"stop_agent", "name":"architect" }"#),
|
||||
)
|
||||
.await
|
||||
.expect("same-project stop cancels the restart-aware logical turn");
|
||||
|
||||
assert_eq!(
|
||||
stopped.detail,
|
||||
"cancelled restarting work for agent architect"
|
||||
);
|
||||
let err = ask
|
||||
.await
|
||||
.expect("ask task joins")
|
||||
.expect_err("cancelled restart-aware ask must fail");
|
||||
assert!(err.to_string().to_ascii_lowercase().contains("cancel"));
|
||||
assert_eq!(
|
||||
fx.mediator.busy_state(rkey(1)),
|
||||
AgentBusyState::Idle,
|
||||
"cancel cleanup frees the target busy state"
|
||||
);
|
||||
assert_eq!(fx.mailbox.pending(&target.id), 0, "FIFO is empty");
|
||||
await_until(|| {
|
||||
fx.live
|
||||
.entry_for(target.id)
|
||||
.is_some_and(|entry| entry.status == WorkStatus::Done)
|
||||
})
|
||||
.await;
|
||||
assert!(
|
||||
fx.structured
|
||||
.session_for_agent_in_project(project_id(), &target.id)
|
||||
.is_none(),
|
||||
"cancel during restart prevents a replacement runtime from being registered"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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.
|
||||
|
||||
Reference in New Issue
Block a user