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)

This commit is contained in:
2026-09-09 08:38:29 +02:00
parent 321ebdaab9
commit d03300a20f
16 changed files with 660 additions and 213 deletions

View File

@ -0,0 +1,13 @@
{
"id": "9a881769-d8e8-410d-85f1-7a3ec2afdde0",
"projectId": "97b49ac2-8376-4aa3-8ea9-bf3ac81d0023",
"agentId": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d",
"sessionId": "7d2abba7-ed93-405f-9fe5-857c8455e436",
"filename": "clipboard-image-1788934010151.png",
"mime": "image/png",
"sizeBytes": 101043,
"sourceKind": "clipboard",
"storagePath": "agent-chat/7d2abba7-ed93-405f-9fe5-857c8455e436/9a881769-d8e8-410d-85f1-7a3ec2afdde0-clipboard-image-1788934010151.png",
"readablePath": "/home/anthony/Documents/Projects/IdeA/.ideai/attachments/agent-chat/7d2abba7-ed93-405f-9fe5-857c8455e436/9a881769-d8e8-410d-85f1-7a3ec2afdde0-clipboard-image-1788934010151.png",
"createdAt": 1788934156106
}

View File

@ -1,8 +1,8 @@
--- ---
issueRef: "#289" issueRef: "#289"
version: 4 version: 5
updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"} updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"}
updatedAt: 1788900709736 updatedAt: 1788901112288
--- ---
## 2026-09-08 — Main ## 2026-09-08 — Main
@ -59,139 +59,29 @@ Tests QA ajoutés dans `crates/application/tests/orchestrator_service.rs` :
- `ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry` : succès same-project post-restart, cible revenue `Idle`, FIFO vide, live-state `Done`, ancienne session structurée absente, session de remplacement enregistrée sur le même agent logique. - `ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry` : succès same-project post-restart, cible revenue `Idle`, FIFO vide, live-state `Done`, ancienne session structurée absente, session de remplacement enregistrée sur le même agent logique.
- `same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session` : cancel/stop same-project pendant le gap de restart, ask réveillé en erreur explicite de cancel, busy/FIFO/live-state nettoyés, aucune session de remplacement enregistrée. - `same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session` : cancel/stop same-project pendant le gap de restart, ask réveillé en erreur explicite de cancel, busy/FIFO/live-state nettoyés, aucune session de remplacement enregistrée.
Commandes exécutées et sorties pertinentes : Commandes finales vertes :
- `cargo fmt --all -- --check`
- `cargo test -p application --test orchestrator_service` — 87 passés
- `cargo test -p application --lib` — 171 passés
- `cargo test -p backend --lib` — 89 passés, 7 ignorés
- `cargo check -p backend` — OK
```bash Couverture validée : same-project, cross-project, faux no-reply interdit, failure restart explicite, cancel pendant restart, nettoyage busy/workstate/registry, watchdog réarmé par activité restart.
git status --short --branch
```
```text Risques résiduels : pas d'E2E vrai provider relancé dans QA ; les 7 tests backend ignorés demandent une permission locale de bind socket.
## fix/289-ask-agent-during-target-restart
M .ideai/tickets/counter.json
M .ideai/tickets/index.json
M crates/application/src/lib.rs
M crates/application/src/orchestrator/mod.rs
M crates/application/src/orchestrator/service.rs
M crates/application/tests/orchestrator_service.rs
M crates/backend/src/lib.rs
?? .ideai/tickets/289/
?? scripts/
```
```bash ## 2026-09-08 — Git/Main clôture
cargo test -p application --test orchestrator_service ask_target_restart_keeps_logical_turn_and_returns_post_restart_final -- --nocapture
```
```text Git a committé le correctif sur la branche feature : `d6d9891`.
test ask_target_restart_keeps_logical_turn_and_returns_post_restart_final ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.02s
```
```bash Git a mergé localement dans `develop` avec merge commit : `321ebda`.
cargo test -p application --test orchestrator_service cross_project_ask_preserves_correlation_across_target_restart -- --nocapture
```
```text Aucun push distant effectué. État final après build : `develop...origin/develop [ahead 276]`, working tree propre sauf `scripts/` non suivi préexistant et hors périmètre.
test cross_project_ask_preserves_correlation_across_target_restart ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.02s
```
```bash Build AppImage exécuté via le skill `build-appimage` :
cargo test -p application --test orchestrator_service ask_target_restart_failure_is_explicit_not_no_reply -- --nocapture - `npm --prefix frontend run build` — OK.
``` - depuis `crates/app-tauri` : `CARGO_HOME=/tmp/idea-cargo-home APPIMAGE_EXTRACT_AND_RUN=1 NO_STRIP=1 ../../frontend/node_modules/.bin/tauri build --bundles appimage` — OK.
```text Artefact produit : `target/release/bundle/appimage/IdeA_0.3.0_amd64.AppImage`.
test ask_target_restart_failure_is_explicit_not_no_reply ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.01s
```
```bash Étape non automatisée volontairement : remplacement de `/home/anthony/Documents/IdeA_0.1.0_amd64.AppImage` et relance d'IdeA, car cela tue l'orchestrateur actif. À faire manuellement quand l'utilisateur est prêt pour activer le correctif dans l'application lancée.
cargo test -p application --test orchestrator_service cross_project_cancel_during_restart_releases_ask_and_correlation -- --nocapture
```
```text
test cross_project_cancel_during_restart_releases_ask_and_correlation ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.00s
```
```bash
cargo test -p application --lib restart_control_activity_rearms_inactivity_watchdog -- --nocapture
```
```text
test orchestrator::service::tests::restart_control_activity_rearms_inactivity_watchdog ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 170 filtered out; finished in 0.05s
```
```bash
cargo test -p application --test orchestrator_service ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry -- --nocapture
```
```text
test ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 86 filtered out; finished in 0.02s
```
```bash
cargo test -p application --test orchestrator_service same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session -- --nocapture
```
```text
test same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 86 filtered out; finished in 0.00s
```
```bash
cargo fmt --all -- --check
```
```text
OK (aucune sortie)
```
```bash
cargo test -p application --test orchestrator_service
```
```text
running 87 tests
test result: ok. 87 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.09s
```
```bash
cargo test -p application --lib
```
```text
running 171 tests
test result: ok. 171 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.15s
```
```bash
cargo test -p backend --lib
```
```text
running 96 tests
test result: ok. 89 passed; 0 failed; 7 ignored; 0 measured; 0 filtered out; finished in 0.04s
```
```bash
cargo check -p backend
```
```text
Finished `dev` profile [unoptimized + debuginfo] target(s) in 3.37s
```
Couverture validée :
- same-project : Final récupéré après restart et test QA de nettoyage busy/workstate/registre.
- cross-project : corrélation conservée jusqu'au Final post-restart puis supprimée.
- pas de faux no-reply/timeout pendant restart : failure restart reste `PROCESS` explicite, watchdog réarmé par activité de contrôle restart.
- failure restart : erreur explicite contenant la cause provider.
- cancel pendant restart : cross-project et same-project couverts.
- nettoyage busy/workstate/registry : FIFO vide, `AgentBusyState::Idle`, live-state `Done`, registre structuré remplacé après succès et non recréé après cancel.
Risques résiduels :
- Pas d'E2E AppImage / vrai provider relancé : la preuve porte sur les ports applicatifs, les sessions structurées de test et le câblage backend compilé.
- Les 7 tests backend ignorés demandent une permission de bind socket locale ; ils restent hors preuve dans cet environnement.

View File

@ -2,7 +2,7 @@
id: "ae7f4f57-087e-463e-9390-33727e751d4f" id: "ae7f4f57-087e-463e-9390-33727e751d4f"
number: 289 number: 289
title: "Garantir ask_agent pendant restart de la cible, same-project et cross-project" title: "Garantir ask_agent pendant restart de la cible, same-project et cross-project"
status: "qa" status: "closed"
priority: "high" priority: "high"
sprint: null sprint: null
links: [] links: []
@ -11,7 +11,7 @@ attachments: []
createdBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"} createdBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"}
updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"} updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"}
createdAt: 1788899102915 createdAt: 1788899102915
updatedAt: 1788900574986 updatedAt: 1788901112288
version: 3 version: 5
--- ---
Vérifier et corriger le comportement suivant : lorsqu'un agent cible a été appelé via ask_agent puis se restart, le demandeur doit attendre la fin du restart et le travail repris/après restart doit rester rattaché à cette même cible. Le contrat doit couvrir les appels intra-projet et cross-project, sans perte de résultat ni attribution à un mauvais agent. Vérifier et corriger le comportement suivant : lorsqu'un agent cible a été appelé via ask_agent puis se restart, le demandeur doit attendre la fin du restart et le travail repris/après restart doit rester rattaché à cette même cible. Le contrat doit couvrir les appels intra-projet et cross-project, sans perte de résultat ni attribution à un mauvais agent.

View File

@ -0,0 +1,87 @@
---
issueRef: "#290"
version: 5
updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"}
updatedAt: 1788935506924
---
# Ticket #290 - Redemarrage agent finalise le tour et bloque le chat
## Reproduction utilisateur
- Date: 2026-09-09.
- L'utilisateur demande a Main de se redemarrer puis d'indiquer le succes une fois fini.
- Dans la conversation, Main lance le redemarrage, envoie un message, le tour se finit et le message envoye est considere comme Final.
- Tentative Stop ensuite: message "Request already stopped".
- Le chat reste bloque en ecriture.
- Capture: `/home/anthony/Documents/Projects/IdeA/.ideai/attachments/agent-chat/7d2abba7-ed93-405f-9fe5-857c8455e436/9a881769-d8e8-410d-85f1-7a3ec2afdde0-clipboard-image-1788934010151.png`.
## Attendu
- Le restart agent doit provoquer une vraie relance/reprise, pas une simple fin de tour capturee comme Final.
- Le chat doit redevenir ecrivable apres redemarrage ou echec controle.
- Le message de succes doit correspondre a un redemarrage effectif, pas au tour qui initie le restart.
## Cadrage UX
- Etats visibles attendus: redemarrage demande, agent en reprise, reprise reussie ou echec controle.
- Stop ne doit etre cliquable que si une requete stoppable existe; ne pas exposer "Request already stopped" brut.
- Le message de succes ne doit etre affiche qu'apres preuve runtime d'un agent relance/reconnecte.
## Cadrage Architecture
- Bug de contrat entre restart autonome et chat humain `agent_send`.
- `ask_structured` etait deja restart-aware; `agent_send` devait ignorer le Final de l'ancien runtime pendant restart et attendre la session remplacee.
- Correctif backend-first, pas de filtre frontend sur le texte.
## Implementation DevBackend
- Branche: `fix/290-restart-agent-send-premature-final`.
- Le tour humain s'enregistre dans `RestartRendezvousCoordinator`.
- `Final`/`Error` de l'ancien provider ignore pendant restart actif.
- Le drain attend la session de remplacement, envoie le prompt de continuation, puis transmet uniquement le vrai terminal.
- Busy/FIFO reste actif jusqu'au vrai resultat, echec explicite ou annulation.
- Stop utilisateur annule aussi le rendez-vous restart.
- Pas de changement frontend/protocole signale.
## Fichiers modifies annonces
- `crates/application/src/orchestrator/service.rs`
- `crates/application/src/orchestrator/mod.rs`
- `crates/application/src/lib.rs`
- `crates/application/src/workstate/actions.rs`
- `crates/application/tests/orchestrator_service.rs`
- `crates/backend/src/lib.rs`
- `crates/app-tauri/src/commands.rs`
- `crates/web-server/src/lib.rs`
## Validations DevBackend annoncees vertes
- `cargo check --workspace`
- `cargo test -p backend self_restart_contract_tests -- --nocapture` — 3/3
- `cargo test -p application --test orchestrator_service restart -- --nocapture` — 8/8
- `cargo test -p application --test workstate_actions -- --test-threads=1` — 9/9
- `cargo test -p app-tauri --test chat_bridge` — 31/31
- `cargo test -p web-server websocket_chat_send -- --test-threads=1` — 3/3
- `cargo fmt --all -- --check`
- `git diff --check`
## Risques signales
- Suites globales avec echecs hors perimetre: 7 tests PTY `web-server` et 2 tests package plugin `application`.
- QA doit confirmer que ces echecs sont preexistants/hors #290 ou remonter le detail.
## Validation QA - 2026-09-09
Verdict: VERT sur le perimetre #290, avec reserves globales reproduites hors perimetre.
Commandes ciblees executees:
- `cargo check --workspace` -> vert.
- `cargo test -p backend self_restart_contract_tests -- --nocapture` -> vert, 3 passed.
- `cargo test -p application --test orchestrator_service restart -- --nocapture` -> vert, 8 passed.
- `cargo test -p application --test workstate_actions -- --test-threads=1` -> vert, 9 passed.
- `cargo test -p app-tauri --test chat_bridge` -> vert, 31 passed.
- `cargo test -p web-server websocket_chat_send -- --test-threads=1` -> vert, 3 passed.
- `cargo fmt --all -- --check` -> vert.
- `git diff --check` -> vert.
Couverture confirmee:
- `Final`/`Error` de l'ancien provider pendant restart ne finalise pas le tour humain: couvert par `self_restart_contract_tests::human_turn_ignores_old_terminal_and_accepts_replacement` et par les tests de registration directe du rendezvous restart.
- Busy/FIFO reste actif jusqu'au terminal post-restart/echec/annulation: couvert par `self_restart_contract_tests`, `orchestrator_service restart` et `workstate_actions`.
- Stop pendant restart annule proprement sans session de remplacement: couvert par `same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session`.
- Non-regression #289 `ask_structured` restart-aware: couverte par `orchestrator_service restart`, incluant ask same-project, cross-project, echec et annulation.
Suites globales rouges reproduites hors perimetre:
- `cargo test -p web-server -- --test-threads=1` -> 103 passed, 7 failed. Les 7 echecs sont les tests PTY `websocket_*terminal*`/`websocket_launch_agent_*` qui recoivent `error` au lieu de `terminal.attached`.
- `cargo test -p application -- --test-threads=1` -> rouge dans `tests/layout_usecases.rs`, 2 failed: `create_layout_plugin_validates_active_runtime_contribution_and_persists_root` et `mutate_set_plugin_layout_state_persists_opaque_state`, tous deux sur `NotFound("plugin package")`.

View File

@ -0,0 +1,17 @@
---
id: "00547de0-3061-4d45-ad9b-f798152f75ce"
number: 290
title: "Corriger le redemarrage agent qui finalise le tour et bloque le chat"
status: "qa"
priority: "high"
sprint: null
links: []
agentRefs: [{"agentId":"ab328d90-c307-4771-a3b6-6c56089c8506","role":"assigned"}]
attachments: []
createdBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"}
updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"}
createdAt: 1788934173642
updatedAt: 1788935506924
version: 5
---
Bug rapporte par l'utilisateur le 2026-09-09: apres une demande de redemarrage, l'agent lance le redemarrage puis envoie un message; ce message est considere comme Final, le tour se termine, Stop repond "Request already stopped", et le chat devient bloque en ecriture. Attachement de reproduction: /home/anthony/Documents/Projects/IdeA/.ideai/attachments/agent-chat/7d2abba7-ed93-405f-9fe5-857c8455e436/9a881769-d8e8-410d-85f1-7a3ec2afdde0-clipboard-image-1788934010151.png. Objectif: corriger le flux pour que le redemarrage aboutisse vraiment, que l'utilisateur recupere un chat utilisable, et que le message de succes ne soit pas capture comme final prematurant le rendez-vous.

View File

@ -1,3 +1,3 @@
{ {
"nextNumber": 290 "nextNumber": 291
} }

View File

@ -3954,7 +3954,7 @@
"issueRef": "#289", "issueRef": "#289",
"path": "289", "path": "289",
"title": "Garantir ask_agent pendant restart de la cible, same-project et cross-project", "title": "Garantir ask_agent pendant restart de la cible, same-project et cross-project",
"status": "qa", "status": "closed",
"priority": "high", "priority": "high",
"sprint": null, "sprint": null,
"assignedAgentIds": [], "assignedAgentIds": [],
@ -3962,7 +3962,23 @@
"kind": "agent", "kind": "agent",
"agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d" "agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d"
}, },
"updatedAt": 1788900574986 "updatedAt": 1788901112288
},
{
"issueRef": "#290",
"path": "290",
"title": "Corriger le redemarrage agent qui finalise le tour et bloque le chat",
"status": "qa",
"priority": "high",
"sprint": null,
"assignedAgentIds": [
"ab328d90-c307-4771-a3b6-6c56089c8506"
],
"createdBy": {
"kind": "agent",
"agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d"
},
"updatedAt": 1788935506924
} }
] ]
} }

View File

@ -2570,8 +2570,10 @@ pub async fn managed_temp_gc(
/// structured registry, open the turn stream (`session.send`), then drain that /// structured registry, open the turn stream (`session.send`), then drain that
/// blocking [`ReplyStream`] on a dedicated OS thread, mapping each /// blocking [`ReplyStream`] on a dedicated OS thread, mapping each
/// [`ReplyEvent`](domain::ports::ReplyEvent) to a [`ReplyChunk`] and forwarding it /// [`ReplyEvent`](domain::ports::ReplyEvent) to a [`ReplyChunk`] and forwarding it
/// through the [`ChatBridge`]. The turn ends deterministically at /// through the [`ChatBridge`]. The logical turn ends deterministically at the
/// [`ReplyChunk::Final`]; the stream is bounded and closes right after. /// 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 /// 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 /// 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( backend::StructuredHumanTurnGuard::start(
std::sync::Arc::clone(&state.turn_watch_input), std::sync::Arc::clone(&state.turn_watch_input),
std::sync::Arc::clone(&state.input_mailbox), std::sync::Arc::clone(&state.input_mailbox),
std::sync::Arc::clone(&state.restart_rendezvous),
*project_id, *project_id,
*agent_id, *agent_id,
&prompt_for_model, &prompt_for_model,
@ -2673,7 +2676,20 @@ pub async fn agent_send(
// still live, so we keep the attach and surface the error. // 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 stream_result = session.send_with_tap(&prompt_for_model, tap_tx).await;
let _ = tap_pump.join(); 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 // 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 // synchronous `Iterator`, exactly like the PTY byte stream). It runs to the
@ -2685,12 +2701,47 @@ pub async fn agent_send(
// event then feeds the service (détecter→planifier). Without a conversation id the // event then feeds the service (détecter→planifier). Without a conversation id the
// service must not pretend an automatic resume is armed. // service must not pretend an automatic resume is armed.
let service = std::sync::Arc::clone(&state.session_limit_service); 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; let meta = session_meta;
std::thread::spawn(move || { std::thread::spawn(move || {
let mut terminal_visible = false; let mut terminal_visible = false;
let mut settler = crate::chat::ProgressSettler::default(); let mut settler = crate::chat::ProgressSettler::default();
for event in stream { let mut current_stream = stream;
if !matches!(event, domain::ports::ReplyEvent::Final { .. }) { '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;
}
}
}
} else {
if let Some(guard) = &turn_guard { if let Some(guard) = &turn_guard {
guard.mark_alive(); guard.mark_alive();
} }
@ -2700,7 +2751,8 @@ pub async fn agent_send(
// continue à drainer comme pour un battement. // continue à drainer comme pour un battement.
if let domain::ports::ReplyEvent::RateLimited { resets_at_ms } = &event { if let domain::ports::ReplyEvent::RateLimited { resets_at_ms } = &event {
if let Some((project_id, agent_id, node_id, conversation_id)) = &meta { if let Some((project_id, agent_id, node_id, conversation_id)) = &meta {
let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms) { let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms)
{
(Some(conversation_id), resets_at_ms) => { (Some(conversation_id), resets_at_ms) => {
(Some(conversation_id.clone()), *resets_at_ms) (Some(conversation_id.clone()), *resets_at_ms)
} }
@ -2732,6 +2784,62 @@ pub async fn agent_send(
// completes and the scrollback stays whole for the next re-attach. // completes and the scrollback stays whole for the next re-attach.
let _ = bridge.send_output_settled(&sid, &mut settler, chunk); 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;
}
}
}
}
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;
}
}
}
if !terminal_visible { if !terminal_visible {
let chunk = crate::chat::unfinished_turn_chunk(); let chunk = crate::chat::unfinished_turn_chunk();
let _ = bridge.send_output_settled(&sid, &mut settler, chunk); let _ = bridge.send_output_settled(&sid, &mut settler, chunk);

View File

@ -168,9 +168,9 @@ pub use orchestrator::{
AgentRestartScheduled, AgentSelfRestartScheduler, AgentWakeService, AskLivenessProbe, AgentRestartScheduled, AgentSelfRestartScheduler, AgentWakeService, AskLivenessProbe,
ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider,
OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory, OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory,
RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousState, WakeSessionProvider, RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousRegistration,
WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, RestartRendezvousState, WakeSessionProvider, WatchdogOutcome, WriteMemory,
SELF_RESTART_CONTINUATION_PROMPT, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, SELF_RESTART_CONTINUATION_PROMPT,
}; };
pub use permission::{ pub use permission::{
GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput, GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput,

View File

@ -21,7 +21,7 @@ pub use rendezvous::{
pub use service::{ pub use service::{
AgentRestartScheduled, AgentSelfRestartScheduler, AskLivenessProbe, ContextGuardUseCases, AgentRestartScheduled, AgentSelfRestartScheduler, AskLivenessProbe, ContextGuardUseCases,
LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome,
OrchestratorService, RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousState, OrchestratorService, RecordTurnProvider, RestartRendezvousCoordinator,
SELF_RESTART_CONTINUATION_PROMPT, RestartRendezvousRegistration, RestartRendezvousState, SELF_RESTART_CONTINUATION_PROMPT,
}; };
pub use wake::{AgentWakeService, WakeSessionProvider}; pub use wake::{AgentWakeService, WakeSessionProvider};

View File

@ -138,7 +138,9 @@ impl RestartRendezvousCoordinator {
Self::default() Self::default()
} }
fn register(self: &Arc<Self>, key: RuntimeAgentKey) -> RestartRendezvousRegistration { /// Registers one logical turn that must survive replacement of its concrete
/// provider session.
pub fn register(self: &Arc<Self>, key: RuntimeAgentKey) -> RestartRendezvousRegistration {
let mut entries = self let mut entries = self
.entries .entries
.lock() .lock()
@ -359,20 +361,29 @@ impl RestartRendezvousCoordinator {
} }
} }
struct RestartRendezvousRegistration { /// Per-turn subscription to restart control-plane transitions.
pub struct RestartRendezvousRegistration {
coordinator: Arc<RestartRendezvousCoordinator>, coordinator: Arc<RestartRendezvousCoordinator>,
key: RuntimeAgentKey, key: RuntimeAgentKey,
rx: watch::Receiver<RestartRendezvousState>, rx: watch::Receiver<RestartRendezvousState>,
} }
impl RestartRendezvousRegistration { impl RestartRendezvousRegistration {
fn state(&mut self) -> RestartRendezvousState { /// Returns the latest transition, marking it observed by this turn.
pub fn state(&mut self) -> RestartRendezvousState {
self.rx.borrow_and_update().clone() self.rx.borrow_and_update().clone()
} }
async fn changed(&mut self) -> Result<(), watch::error::RecvError> { /// Waits asynchronously for a restart control transition.
pub async fn changed(&mut self) -> Result<(), watch::error::RecvError> {
self.rx.changed().await self.rx.changed().await
} }
/// Acknowledges ownership of the replacement session and releases the
/// detached restart scheduler.
pub fn accept_ready(&self, restart_id: &str) {
self.coordinator.resume_taken(self.key, restart_id);
}
} }
impl Drop for RestartRendezvousRegistration { impl Drop for RestartRendezvousRegistration {

View File

@ -149,6 +149,7 @@ pub struct StopLiveAgent {
input: Option<Arc<dyn InputMediator>>, input: Option<Arc<dyn InputMediator>>,
waits: Option<Arc<OrchestratorService>>, waits: Option<Arc<OrchestratorService>>,
cascade: Option<Arc<dyn StopAgentExecutionCascade>>, cascade: Option<Arc<dyn StopAgentExecutionCascade>>,
restart_rendezvous: Option<Arc<crate::RestartRendezvousCoordinator>>,
} }
impl StopLiveAgent { impl StopLiveAgent {
@ -161,6 +162,7 @@ impl StopLiveAgent {
input: None, input: None,
waits: None, waits: None,
cascade: None, cascade: None,
restart_rendezvous: None,
} }
} }
@ -187,6 +189,16 @@ impl StopLiveAgent {
self self
} }
/// Wires cancellation of a logical turn currently crossing a self-restart.
#[must_use]
pub fn with_restart_rendezvous(
mut self,
coordinator: Arc<crate::RestartRendezvousCoordinator>,
) -> Self {
self.restart_rendezvous = Some(coordinator);
self
}
/// Stops the agent's current work. /// Stops the agent's current work.
/// ///
/// # Errors /// # Errors
@ -198,6 +210,9 @@ impl StopLiveAgent {
) -> Result<StopLiveAgentOutput, AppError> { ) -> Result<StopLiveAgentOutput, AppError> {
self.stop_dependencies(&input.project, input.agent_id) self.stop_dependencies(&input.project, input.agent_id)
.await?; .await?;
if let Some(coordinator) = &self.restart_rendezvous {
coordinator.cancel(RuntimeAgentKey::new(input.project.id, input.agent_id));
}
if let Some(mediator) = &self.input { if let Some(mediator) = &self.input {
mediator.preempt(RuntimeAgentKey::new(input.project.id, input.agent_id)); mediator.preempt(RuntimeAgentKey::new(input.project.id, input.agent_id));
} }

View File

@ -5716,6 +5716,60 @@ async fn same_project_stop_during_restart_releases_ask_and_leaves_no_replacement
); );
} }
#[tokio::test]
async fn direct_human_turn_registration_holds_restart_until_replacement_handoff() {
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let key = rkey(1);
let restart_id = "restart-human-turn".to_owned();
let mut turn = coordinator.register(key);
assert!(
coordinator.begin(key, restart_id.clone()),
"a registered direct human turn owns self-restart continuation"
);
assert!(matches!(
turn.state(),
application::RestartRendezvousState::Restarting { .. }
));
coordinator.ready(key, &restart_id);
turn.changed().await.expect("ready transition remains live");
assert_eq!(
turn.state(),
application::RestartRendezvousState::Ready {
restart_id: restart_id.clone()
}
);
turn.accept_ready(&restart_id);
coordinator
.wait_for_handoff(key, &restart_id)
.await
.expect("scheduler is released only after the replacement is accepted");
assert_eq!(turn.state(), application::RestartRendezvousState::Idle);
}
#[tokio::test]
async fn direct_human_turn_registration_surfaces_restart_failure() {
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let key = rkey(1);
let restart_id = "restart-human-failure".to_owned();
let mut turn = coordinator.register(key);
assert!(coordinator.begin(key, restart_id.clone()));
coordinator.fail(key, &restart_id, "replacement launch failed".to_owned());
turn.changed()
.await
.expect("failure transition remains live");
assert_eq!(
turn.state(),
application::RestartRendezvousState::Failed {
restart_id,
message: "replacement launch failed".to_owned(),
}
);
}
/// Régression #287 : la sonde transcript peut être stable (notamment inexistante pour /// 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 /// 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. /// tour. Chaque événement non terminal doit réarmer la fenêtre jusqu'au `Final` tardif.

View File

@ -1072,6 +1072,7 @@ fn self_restart_launch_input(
#[cfg(test)] #[cfg(test)]
mod self_restart_contract_tests { mod self_restart_contract_tests {
use super::*; use super::*;
use domain::input::InputMediator;
#[test] #[test]
fn relaunch_preserves_conversation_and_never_requests_fresh_start() { fn relaunch_preserves_conversation_and_never_requests_fresh_start() {
@ -1104,6 +1105,77 @@ mod self_restart_contract_tests {
assert!(input.require_structured); assert!(input.require_structured);
assert!(!input.fresh_start); assert!(!input.fresh_start);
} }
#[tokio::test]
async fn human_turn_ignores_old_terminal_and_accepts_replacement() {
let mailbox = Arc::new(InMemoryMailbox::new());
let input = Arc::new(MediatedInbox::new(
Arc::clone(&mailbox),
Arc::new(SystemMillisClock),
));
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let project_id = ProjectId::from_uuid(Uuid::from_u128(290));
let agent_id = AgentId::from_uuid(Uuid::from_u128(1));
let key = domain::RuntimeAgentKey::new(project_id, agent_id);
let mut turn = StructuredHumanTurnGuard::start(
Arc::clone(&input) as Arc<dyn domain::input::InputMediator>,
Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
Arc::clone(&coordinator),
project_id,
agent_id,
"restart me",
);
let restart_id = "restart-human-290".to_owned();
assert!(coordinator.begin(key, restart_id.clone()));
coordinator.ready(key, &restart_id);
assert_eq!(
turn.transition_after_provider_terminal_async().await,
StructuredHumanTurnTransition::Resume,
"the old provider terminal is a control transition, not the human Final"
);
coordinator
.wait_for_handoff(key, &restart_id)
.await
.expect("replacement handoff is acknowledged");
turn.finish();
assert_eq!(input.busy_state(key), domain::input::AgentBusyState::Idle);
assert_eq!(mailbox.pending(&key), 0);
}
#[tokio::test]
async fn human_turn_surfaces_explicit_restart_failure_and_releases_busy() {
let mailbox = Arc::new(InMemoryMailbox::new());
let input = Arc::new(MediatedInbox::new(
Arc::clone(&mailbox),
Arc::new(SystemMillisClock),
));
let coordinator = Arc::new(application::RestartRendezvousCoordinator::new());
let project_id = ProjectId::from_uuid(Uuid::from_u128(290));
let agent_id = AgentId::from_uuid(Uuid::from_u128(2));
let key = domain::RuntimeAgentKey::new(project_id, agent_id);
let mut turn = StructuredHumanTurnGuard::start(
Arc::clone(&input) as Arc<dyn domain::input::InputMediator>,
Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
Arc::clone(&coordinator),
project_id,
agent_id,
"restart me",
);
let restart_id = "restart-human-failure-290".to_owned();
assert!(coordinator.begin(key, restart_id.clone()));
coordinator.fail(key, &restart_id, "replacement failed".to_owned());
assert_eq!(
turn.transition_after_provider_terminal_async().await,
StructuredHumanTurnTransition::Failed("replacement failed".to_owned())
);
turn.finish();
assert_eq!(input.busy_state(key), domain::input::AgentBusyState::Idle);
assert_eq!(mailbox.pending(&key), 0);
}
} }
impl AgentSelfRestartScheduler for AppAgentSelfRestarter { impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
@ -1306,6 +1378,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
let turn = StructuredHumanTurnGuard::start( let turn = StructuredHumanTurnGuard::start(
Arc::clone(&input), Arc::clone(&input),
Arc::clone(&mailbox), Arc::clone(&mailbox),
Arc::clone(&rendezvous),
project.id, project.id,
agent, agent,
application::SELF_RESTART_CONTINUATION_PROMPT, application::SELF_RESTART_CONTINUATION_PROMPT,
@ -1361,23 +1434,39 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
/// ///
/// Direct chat adapters (`agent_send`, web `chat.send`) still drive the provider /// Direct chat adapters (`agent_send`, web `chat.send`) still drive the provider
/// through `AgentSession::send`, but they must share the same busy/FIFO accounting /// through `AgentSession::send`, but they must share the same busy/FIFO accounting
/// as structured inter-agent asks. This guard starts a silent human turn and /// as structured inter-agent asks. This guard starts a silent human turn,
/// retires it on terminal outcome or drop. /// registers it for restart handoff, and retires it only on the logical terminal
/// outcome or drop.
pub struct StructuredHumanTurnGuard { pub struct StructuredHumanTurnGuard {
input: Arc<dyn domain::input::InputMediator>, input: Arc<dyn domain::input::InputMediator>,
mailbox: Arc<dyn domain::mailbox::AgentMailbox>, mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
agent: domain::RuntimeAgentKey, agent: domain::RuntimeAgentKey,
ticket_id: domain::mailbox::TicketId, ticket_id: domain::mailbox::TicketId,
armed: Arc<std::sync::atomic::AtomicBool>, armed: Arc<std::sync::atomic::AtomicBool>,
restart: Option<application::RestartRendezvousRegistration>,
finish_on_drop: bool, finish_on_drop: bool,
} }
/// Control-plane outcome observed when one concrete provider turn terminates.
#[derive(Debug, Clone, PartialEq, Eq)]
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,
/// Replacement failed explicitly.
Failed(String),
/// The user cancelled the logical turn while replacement was in flight.
Cancelled,
}
impl StructuredHumanTurnGuard { impl StructuredHumanTurnGuard {
/// Starts the mediated busy turn for a direct human structured chat message. /// Starts the mediated busy turn for a direct human structured chat message.
#[must_use] #[must_use]
pub fn start( pub fn start(
input: Arc<dyn domain::input::InputMediator>, input: Arc<dyn domain::input::InputMediator>,
mailbox: Arc<dyn domain::mailbox::AgentMailbox>, mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
restart_rendezvous: Arc<application::RestartRendezvousCoordinator>,
project_id: ProjectId, project_id: ProjectId,
agent_id: AgentId, agent_id: AgentId,
prompt: &str, prompt: &str,
@ -1397,6 +1486,7 @@ impl StructuredHumanTurnGuard {
agent, agent,
ticket_id, ticket_id,
armed: Arc::new(std::sync::atomic::AtomicBool::new(owns_turn)), armed: Arc::new(std::sync::atomic::AtomicBool::new(owns_turn)),
restart: owns_turn.then(|| restart_rendezvous.register(agent)),
finish_on_drop: true, finish_on_drop: true,
} }
} }
@ -1410,10 +1500,55 @@ impl StructuredHumanTurnGuard {
agent: self.agent, agent: self.agent,
ticket_id: self.ticket_id, ticket_id: self.ticket_id,
armed: Arc::clone(&self.armed), armed: Arc::clone(&self.armed),
restart: None,
finish_on_drop: false, finish_on_drop: false,
} }
} }
/// Resolves whether a concrete provider terminal event really ends the
/// logical human turn. During self-restart it waits for the replacement
/// session, then acknowledges the handoff so the scheduler can complete.
pub async fn transition_after_provider_terminal_async(
&mut self,
) -> StructuredHumanTurnTransition {
let Some(restart) = self.restart.as_mut() else {
return StructuredHumanTurnTransition::Complete;
};
loop {
match restart.state() {
application::RestartRendezvousState::Idle => {
return StructuredHumanTurnTransition::Complete;
}
application::RestartRendezvousState::Restarting { .. } => {
if restart.changed().await.is_err() {
return StructuredHumanTurnTransition::Failed(
"restart coordination closed".to_owned(),
);
}
}
application::RestartRendezvousState::Ready { restart_id } => {
restart.accept_ready(&restart_id);
return StructuredHumanTurnTransition::Resume;
}
application::RestartRendezvousState::Failed { message, .. } => {
return StructuredHumanTurnTransition::Failed(message);
}
application::RestartRendezvousState::Cancelled { .. } => {
return StructuredHumanTurnTransition::Cancelled;
}
}
}
}
/// Reports whether a self-restart already owns the current concrete
/// provider failure. The scheduler publishes `Restarting` synchronously
/// before the MCP tool returns, so this check does not inspect provider text.
pub fn restart_is_active(&mut self) -> bool {
self.restart.as_mut().is_some_and(|restart| {
!matches!(restart.state(), application::RestartRendezvousState::Idle)
})
}
/// Records proof of liveness for the running turn. /// Records proof of liveness for the running turn.
pub fn mark_alive(&self) { pub fn mark_alive(&self) {
if self.armed.load(std::sync::atomic::Ordering::Acquire) { if self.armed.load(std::sync::atomic::Ordering::Acquire) {
@ -2005,6 +2140,8 @@ pub struct BackendCore {
/// Mailbox paired with [`Self::turn_watch_input`], used by direct structured /// Mailbox paired with [`Self::turn_watch_input`], used by direct structured
/// human turns to retire the synthetic busy/FIFO ticket on terminal outcome. /// human turns to retire the synthetic busy/FIFO ticket on terminal outcome.
pub input_mailbox: Arc<dyn domain::mailbox::AgentMailbox>, pub input_mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
/// Stable restart control-plane shared by delegated and direct human turns.
pub restart_rendezvous: Arc<application::RestartRendezvousCoordinator>,
/// Handles des watches de fin-de-tour vivants, par clé runtime. (Re)lancer un agent /// Handles des watches de fin-de-tour vivants, par clé runtime. (Re)lancer un agent
/// dans un projet **remplace** son handle (l'ancien est droppé ⇒ polling arrêté) ; /// dans un projet **remplace** son handle (l'ancien est droppé ⇒ polling arrêté) ;
/// fermer/arrêter l'agent le retire. `Mutex` car launch/stop y accèdent concurremment. /// fermer/arrêter l'agent le retire. `Mutex` car launch/stop y accèdent concurremment.
@ -3639,7 +3776,7 @@ impl BackendCore {
.with_conversations(Arc::clone(&conversation_registry)) .with_conversations(Arc::clone(&conversation_registry))
.with_events(Arc::clone(&events_port)) .with_events(Arc::clone(&events_port))
.with_self_restart(self_restart) .with_self_restart(self_restart)
.with_restart_rendezvous(restart_rendezvous) .with_restart_rendezvous(Arc::clone(&restart_rendezvous))
// Faits OS/runtime (exe $APPIMAGE/current_exe + endpoint loopback) pour // Faits OS/runtime (exe $APPIMAGE/current_exe + endpoint loopback) pour
// que les (re)lancements issus du chemin `ask` écrivent la déclaration MCP // que les (re)lancements issus du chemin `ask` écrivent la déclaration MCP
// réelle ⇒ le pont MCP de la cible se spawne et `idea_reply` débloque le round-trip. // réelle ⇒ le pont MCP de la cible se spawne et `idea_reply` débloque le round-trip.
@ -3757,7 +3894,8 @@ impl BackendCore {
Arc::clone(&orchestrator_service), Arc::clone(&orchestrator_service),
) )
.with_execution_cascade(Arc::clone(&orchestrator_service) .with_execution_cascade(Arc::clone(&orchestrator_service)
as Arc<dyn application::StopAgentExecutionCascade>), as Arc<dyn application::StopAgentExecutionCascade>)
.with_restart_rendezvous(Arc::clone(&restart_rendezvous)),
); );
// --- Windows (L10) --- // --- Windows (L10) ---
@ -3952,6 +4090,7 @@ impl BackendCore {
turn_watcher, turn_watcher,
turn_watch_input: Arc::clone(&input_mediator), turn_watch_input: Arc::clone(&input_mediator),
input_mailbox: Arc::clone(&mailbox), input_mailbox: Arc::clone(&mailbox),
restart_rendezvous: Arc::clone(&restart_rendezvous),
turn_watch_handles: Mutex::new(HashMap::new()), turn_watch_handles: Mutex::new(HashMap::new()),
fs_port: Arc::clone(&fs_port), fs_port: Arc::clone(&fs_port),
home_dir, home_dir,

View File

@ -1882,6 +1882,7 @@ async fn ws_send_chat(
backend::StructuredHumanTurnGuard::start( backend::StructuredHumanTurnGuard::start(
Arc::clone(&state.app.turn_watch_input), Arc::clone(&state.app.turn_watch_input),
Arc::clone(&state.app.input_mailbox), Arc::clone(&state.app.input_mailbox),
Arc::clone(&state.app.restart_rendezvous),
*project_id, *project_id,
*agent_id, *agent_id,
&prompt, &prompt,
@ -1898,6 +1899,9 @@ async fn ws_send_chat(
guard.mark_alive(); guard.mark_alive();
} }
} }
if matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. }) {
continue;
}
let chunk = reply_event_chunk(event); let chunk = reply_event_chunk(event);
if tap_out if tap_out
.blocking_send(ServerFrame::chat_output(None, sid, chunk)) .blocking_send(ServerFrame::chat_output(None, sid, chunk))
@ -1909,14 +1913,61 @@ async fn ws_send_chat(
}); });
let stream_result = session.send_with_tap(&prompt, tap_tx).await; let stream_result = session.send_with_tap(&prompt, tap_tx).await;
let _ = tap_pump.join(); let _ = tap_pump.join();
let stream = stream_result.map_err(|err| ErrorDto { let mut turn_guard = turn_guard;
let mut stream = match stream_result {
Ok(stream) => stream,
Err(_err)
if turn_guard
.as_mut()
.is_some_and(|guard| guard.restart_is_active()) =>
{
Box::new(std::iter::empty()) as domain::ports::ReplyStream
}
Err(err) => {
return Err(ErrorDto {
code: "CHAT_SEND_FAILED".to_owned(), code: "CHAT_SEND_FAILED".to_owned(),
message: err.to_string(), message: err.to_string(),
})?; })
}
};
let mut sent_ack = false; let mut sent_ack = false;
'logical_turn: loop {
let mut concrete_terminal = false;
let mut resume = false;
for event in stream { for event in stream {
if !matches!(event, ReplyEvent::Final { .. }) { let provider_terminal =
matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. });
if provider_terminal {
concrete_terminal = true;
if let Some(guard) = turn_guard.as_mut() {
match guard.transition_after_provider_terminal_async().await {
backend::StructuredHumanTurnTransition::Complete => {}
backend::StructuredHumanTurnTransition::Resume => {
resume = true;
break;
}
backend::StructuredHumanTurnTransition::Failed(message) => {
let chunk = reply_event_chunk(ReplyEvent::Error { message });
tx.send(ServerFrame::chat_output(
(!sent_ack).then_some(frame.id.as_str()),
sid,
chunk,
))
.await
.map_err(|_| ErrorDto {
code: "WS_CLOSED".to_owned(),
message: "websocket output closed".to_owned(),
})?;
sent_ack = true;
break 'logical_turn;
}
backend::StructuredHumanTurnTransition::Cancelled => {
break 'logical_turn;
}
}
}
} else {
if let Some(guard) = &turn_guard { if let Some(guard) = &turn_guard {
guard.mark_alive(); guard.mark_alive();
} }
@ -1940,6 +1991,52 @@ async fn ws_send_chat(
} }
} }
if !resume && !concrete_terminal {
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::Failed(message) => {
let chunk = reply_event_chunk(ReplyEvent::Error { message });
tx.send(ServerFrame::chat_output(
(!sent_ack).then_some(frame.id.as_str()),
sid,
chunk,
))
.await
.map_err(|_| ErrorDto {
code: "WS_CLOSED".to_owned(),
message: "websocket output closed".to_owned(),
})?;
sent_ack = true;
}
backend::StructuredHumanTurnTransition::Cancelled => break,
}
}
}
if !resume {
break;
}
let Some((project_id, agent_id, _, _)) = session_meta.as_ref() else {
break;
};
let replacement = state
.app
.structured_sessions
.session_for_agent_in_project(*project_id, agent_id)
.ok_or_else(|| ErrorDto {
code: "CHAT_RESTART_FAILED".to_owned(),
message: "restart completed without a replacement structured session".to_owned(),
})?;
stream = replacement
.send(application::SELF_RESTART_CONTINUATION_PROMPT)
.await
.map_err(|err| ErrorDto {
code: "CHAT_RESTART_FAILED".to_owned(),
message: err.to_string(),
})?;
}
if !sent_ack { if !sent_ack {
tx.send(ServerFrame::chat_output( tx.send(ServerFrame::chat_output(
Some(&frame.id), Some(&frame.id),