fix(application,backend,app-tauri): libère l'état busy après Final et draine la queue des messages agents — Final/Error terminent immédiatement le pump structuré via reply_event_ends_turn (le provider n'est pas tenu de fermer son itérateur), release busy/FIFO idempotente bornée au ticket observé (release_turn_accounting: cancel_head + mark_idle seulement si le ticket appartient toujours au tour), cleanup tardif Final/close/timeout/cancel sans libérer un tour successeur, factoring CancelAgentChat sur la même primitive, régression #357 préservée — #362 (QA verte: cargo fmt --check, chat_bridge 36 tests, cancel_agent_chat 4 tests, backend 106 tests, drain_with_readiness_lot1 7 tests)
This commit is contained in:
@ -14,7 +14,7 @@ use std::sync::{Arc, Mutex};
|
||||
use tauri::ipc::{Channel, InvokeResponseBody};
|
||||
|
||||
use app_tauri_lib::chat::{
|
||||
chunk_from_event, unfinished_turn_chunk, ChatBridge, ProgressSettler,
|
||||
chunk_from_event, reply_event_ends_turn, unfinished_turn_chunk, ChatBridge, ProgressSettler,
|
||||
MAX_CHAT_SCROLLBACK_BYTES, MAX_CHAT_SCROLLBACK_CHUNKS, MODEL_RESPONSE_NOT_FINALIZED_MESSAGE,
|
||||
};
|
||||
use app_tauri_lib::dto::{
|
||||
@ -264,6 +264,7 @@ fn chunk_from_event_drops_heartbeat() {
|
||||
fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec<ReplyEvent>) {
|
||||
let mut terminal_visible = false;
|
||||
for event in events {
|
||||
let terminal = reply_event_ends_turn(&event);
|
||||
// Mirror the real pump: heartbeats map to no chunk and are skipped.
|
||||
if let Some(chunk) = chunk_from_event(event) {
|
||||
if matches!(
|
||||
@ -274,6 +275,9 @@ fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec<Rep
|
||||
}
|
||||
let _ = bridge.send_output(session, chunk);
|
||||
}
|
||||
if terminal {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if !terminal_visible {
|
||||
let _ = bridge.send_output(session, unfinished_turn_chunk());
|
||||
@ -281,6 +285,30 @@ fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec<Rep
|
||||
bridge.detach_if(session, gen);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pump_stops_at_final_without_waiting_for_provider_eof() {
|
||||
let bridge = ChatBridge::new();
|
||||
let session = sid();
|
||||
let sink = Arc::new(Mutex::new(Vec::new()));
|
||||
let gen = bridge.register(session, capturing_channel(Arc::clone(&sink)));
|
||||
|
||||
pump_turn(
|
||||
&bridge,
|
||||
&session,
|
||||
gen,
|
||||
vec![
|
||||
ReplyEvent::Final {
|
||||
content: "done".into(),
|
||||
},
|
||||
ReplyEvent::TextDelta {
|
||||
text: "late provider output".into(),
|
||||
},
|
||||
],
|
||||
);
|
||||
|
||||
assert_eq!(sink.lock().unwrap().as_slice(), &[final_chunk("done")]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pump_delivers_deltas_then_exactly_one_final_in_order() {
|
||||
let bridge = ChatBridge::new();
|
||||
@ -423,9 +451,14 @@ fn pump_with_existing_error_does_not_emit_generic_error() {
|
||||
&bridge,
|
||||
&session,
|
||||
gen,
|
||||
vec![ReplyEvent::Error {
|
||||
message: "auth failed".into(),
|
||||
}],
|
||||
vec![
|
||||
ReplyEvent::Error {
|
||||
message: "auth failed".into(),
|
||||
},
|
||||
ReplyEvent::TextDelta {
|
||||
text: "late provider output".into(),
|
||||
},
|
||||
],
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
|
||||
Reference in New Issue
Block a user