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:
@ -390,6 +390,16 @@ pub fn chunk_from_event(event: ReplyEvent) -> Option<ReplyChunk> {
|
|||||||
chunk_from_event_at(event, None)
|
chunk_from_event_at(event, None)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Returns whether one provider event closes the concrete structured turn.
|
||||||
|
///
|
||||||
|
/// This is the control-plane boundary used by the `agent_send` pump. In
|
||||||
|
/// particular, the provider is not required to close its iterator after a
|
||||||
|
/// terminal event: keeping the iterator alive must not keep the agent Busy.
|
||||||
|
#[must_use]
|
||||||
|
pub fn reply_event_ends_turn(event: &ReplyEvent) -> bool {
|
||||||
|
matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. })
|
||||||
|
}
|
||||||
|
|
||||||
/// Maps a domain reply event while preserving the backend acceptance time.
|
/// Maps a domain reply event while preserving the backend acceptance time.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn chunk_from_event_at(event: ReplyEvent, at_ms: Option<u64>) -> Option<ReplyChunk> {
|
pub fn chunk_from_event_at(event: ReplyEvent, at_ms: Option<u64>) -> Option<ReplyChunk> {
|
||||||
|
|||||||
@ -3122,11 +3122,7 @@ pub async fn agent_send(
|
|||||||
terminal_visible = true;
|
terminal_visible = true;
|
||||||
break 'logical_turn;
|
break 'logical_turn;
|
||||||
}
|
}
|
||||||
let provider_terminal = matches!(
|
let provider_terminal = crate::chat::reply_event_ends_turn(&event);
|
||||||
event,
|
|
||||||
domain::ports::ReplyEvent::Final { .. }
|
|
||||||
| domain::ports::ReplyEvent::Error { .. }
|
|
||||||
);
|
|
||||||
if provider_terminal {
|
if provider_terminal {
|
||||||
concrete_terminal = true;
|
concrete_terminal = true;
|
||||||
if let Some(guard) = turn_guard.as_mut() {
|
if let Some(guard) = turn_guard.as_mut() {
|
||||||
@ -3199,6 +3195,13 @@ pub async fn agent_send(
|
|||||||
// (no channel at this generation) we keep draining so the turn still
|
// (no channel at this generation) we keep draining so the turn still
|
||||||
// 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 provider_terminal {
|
||||||
|
// Final/Error is the authoritative structured boundary. Some
|
||||||
|
// providers keep their iterator/channel alive after publishing it;
|
||||||
|
// waiting for EOF would leave the mediated ticket Busy forever even
|
||||||
|
// though the UI already displayed the terminal chunk.
|
||||||
|
break 'logical_turn;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if !resume && !concrete_terminal {
|
if !resume && !concrete_terminal {
|
||||||
|
|||||||
@ -14,7 +14,7 @@ use std::sync::{Arc, Mutex};
|
|||||||
use tauri::ipc::{Channel, InvokeResponseBody};
|
use tauri::ipc::{Channel, InvokeResponseBody};
|
||||||
|
|
||||||
use app_tauri_lib::chat::{
|
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,
|
MAX_CHAT_SCROLLBACK_BYTES, MAX_CHAT_SCROLLBACK_CHUNKS, MODEL_RESPONSE_NOT_FINALIZED_MESSAGE,
|
||||||
};
|
};
|
||||||
use app_tauri_lib::dto::{
|
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>) {
|
fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec<ReplyEvent>) {
|
||||||
let mut terminal_visible = false;
|
let mut terminal_visible = false;
|
||||||
for event in events {
|
for event in events {
|
||||||
|
let terminal = reply_event_ends_turn(&event);
|
||||||
// Mirror the real pump: heartbeats map to no chunk and are skipped.
|
// Mirror the real pump: heartbeats map to no chunk and are skipped.
|
||||||
if let Some(chunk) = chunk_from_event(event) {
|
if let Some(chunk) = chunk_from_event(event) {
|
||||||
if matches!(
|
if matches!(
|
||||||
@ -274,6 +275,9 @@ fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec<Rep
|
|||||||
}
|
}
|
||||||
let _ = bridge.send_output(session, chunk);
|
let _ = bridge.send_output(session, chunk);
|
||||||
}
|
}
|
||||||
|
if terminal {
|
||||||
|
break;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if !terminal_visible {
|
if !terminal_visible {
|
||||||
let _ = bridge.send_output(session, unfinished_turn_chunk());
|
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);
|
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]
|
#[test]
|
||||||
fn pump_delivers_deltas_then_exactly_one_final_in_order() {
|
fn pump_delivers_deltas_then_exactly_one_final_in_order() {
|
||||||
let bridge = ChatBridge::new();
|
let bridge = ChatBridge::new();
|
||||||
@ -423,9 +451,14 @@ fn pump_with_existing_error_does_not_emit_generic_error() {
|
|||||||
&bridge,
|
&bridge,
|
||||||
&session,
|
&session,
|
||||||
gen,
|
gen,
|
||||||
vec![ReplyEvent::Error {
|
vec![
|
||||||
|
ReplyEvent::Error {
|
||||||
message: "auth failed".into(),
|
message: "auth failed".into(),
|
||||||
}],
|
},
|
||||||
|
ReplyEvent::TextDelta {
|
||||||
|
text: "late provider output".into(),
|
||||||
|
},
|
||||||
|
],
|
||||||
);
|
);
|
||||||
|
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
|
|||||||
@ -275,11 +275,11 @@ pub use template::{
|
|||||||
SyncAgentWithTemplateOutput, UpdateTemplate, UpdateTemplateInput, UpdateTemplateOutput,
|
SyncAgentWithTemplateOutput, UpdateTemplate, UpdateTemplateInput, UpdateTemplateOutput,
|
||||||
};
|
};
|
||||||
pub use terminal::{
|
pub use terminal::{
|
||||||
CancelAgentChat, CancelAgentChatOutput, CloseTerminal, CloseTerminalInput, CloseTerminalOutput,
|
release_turn_accounting, CancelAgentChat, CancelAgentChatOutput, CloseTerminal,
|
||||||
LiveAgentRegistry, LiveRuntimeState, LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot,
|
CloseTerminalInput, CloseTerminalOutput, LiveAgentRegistry, LiveRuntimeState,
|
||||||
LiveSessions, OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal,
|
LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot, LiveSessions, OpenTerminal,
|
||||||
ResizeTerminalInput, StructuredSessions, TerminalSessions, WriteToTerminal,
|
OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, StructuredSessions,
|
||||||
WriteToTerminalInput,
|
TerminalSessions, WriteToTerminal, WriteToTerminalInput,
|
||||||
};
|
};
|
||||||
pub use ticket_assistant::{
|
pub use ticket_assistant::{
|
||||||
CloseTicketAssistant, CloseTicketAssistantInput, OpenTicketAssistant, OpenTicketAssistantInput,
|
CloseTicketAssistant, CloseTicketAssistantInput, OpenTicketAssistant, OpenTicketAssistantInput,
|
||||||
|
|||||||
@ -33,8 +33,8 @@ pub use registry::{
|
|||||||
LiveSessions, StructuredSessions, TerminalSessions,
|
LiveSessions, StructuredSessions, TerminalSessions,
|
||||||
};
|
};
|
||||||
pub use usecases::{
|
pub use usecases::{
|
||||||
stop_live_agent_runtime_in_project, CancelAgentChat, CancelAgentChatOutput, CloseTerminal,
|
release_turn_accounting, stop_live_agent_runtime_in_project, CancelAgentChat,
|
||||||
CloseTerminalInput, CloseTerminalOutput, OpenTerminal, OpenTerminalInput, OpenTerminalOutput,
|
CancelAgentChatOutput, CloseTerminal, CloseTerminalInput, CloseTerminalOutput, OpenTerminal,
|
||||||
ResizeTerminal, ResizeTerminalInput, StoppedAgentRuntime, WriteToTerminal,
|
OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput,
|
||||||
WriteToTerminalInput,
|
StoppedAgentRuntime, WriteToTerminal, WriteToTerminalInput,
|
||||||
};
|
};
|
||||||
|
|||||||
@ -26,6 +26,28 @@ pub struct CancelAgentChatOutput {
|
|||||||
pub outcome: CancelCurrentTurnOutcome,
|
pub outcome: CancelCurrentTurnOutcome,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Releases exactly one mediated turn without allowing late terminal cleanup to
|
||||||
|
/// affect a successor that already owns the agent.
|
||||||
|
///
|
||||||
|
/// `cancel_head` and `mark_idle` are individually idempotent, while the ticket
|
||||||
|
/// checks fence Final/close/timeout/cancel races across successive turns.
|
||||||
|
pub fn release_turn_accounting(
|
||||||
|
input: &dyn InputMediator,
|
||||||
|
mailbox: &dyn AgentMailbox,
|
||||||
|
agent: domain::RuntimeAgentKey,
|
||||||
|
ticket: domain::mailbox::TicketId,
|
||||||
|
) -> bool {
|
||||||
|
if input.busy_state(agent).ticket() != Some(ticket) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
mailbox.cancel_head(agent, ticket);
|
||||||
|
if input.busy_state(agent).ticket() != Some(ticket) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
input.mark_idle(agent);
|
||||||
|
true
|
||||||
|
}
|
||||||
|
|
||||||
/// Cancels one structured chat turn without stopping its reusable runtime.
|
/// Cancels one structured chat turn without stopping its reusable runtime.
|
||||||
pub struct CancelAgentChat {
|
pub struct CancelAgentChat {
|
||||||
sessions: Arc<StructuredSessions>,
|
sessions: Arc<StructuredSessions>,
|
||||||
@ -72,14 +94,7 @@ impl CancelAgentChat {
|
|||||||
if let (Some((agent, ticket)), Some((input, mailbox))) =
|
if let (Some((agent, ticket)), Some((input, mailbox))) =
|
||||||
(active_ticket, self.turn_accounting.as_ref())
|
(active_ticket, self.turn_accounting.as_ref())
|
||||||
{
|
{
|
||||||
// The provider kill can race the old pump's normal guard cleanup and
|
release_turn_accounting(input.as_ref(), mailbox.as_ref(), agent, ticket);
|
||||||
// a caller may already have started the next turn. Retire only the
|
|
||||||
// exact ticket observed before cancellation; never mark a successor
|
|
||||||
// idle when it has already replaced that ticket.
|
|
||||||
if input.busy_state(agent).ticket() == Some(ticket) {
|
|
||||||
mailbox.cancel_head(agent, ticket);
|
|
||||||
input.mark_idle(agent);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Ok(CancelAgentChatOutput {
|
Ok(CancelAgentChatOutput {
|
||||||
|
|||||||
@ -1165,6 +1165,7 @@ fn automatic_resume_launch_input(
|
|||||||
mod self_restart_contract_tests {
|
mod self_restart_contract_tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use domain::input::InputMediator;
|
use domain::input::InputMediator;
|
||||||
|
use domain::mailbox::AgentMailbox;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn relaunch_preserves_conversation_and_never_requests_fresh_start() {
|
fn relaunch_preserves_conversation_and_never_requests_fresh_start() {
|
||||||
@ -1342,6 +1343,70 @@ mod self_restart_contract_tests {
|
|||||||
assert_eq!(mailbox.pending(&key), 0);
|
assert_eq!(mailbox.pending(&key), 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn human_turn_finish_is_idempotent() {
|
||||||
|
let mailbox = Arc::new(InMemoryMailbox::new());
|
||||||
|
let input = Arc::new(MediatedInbox::new(
|
||||||
|
Arc::clone(&mailbox),
|
||||||
|
Arc::new(SystemMillisClock),
|
||||||
|
));
|
||||||
|
let project_id = ProjectId::from_uuid(Uuid::from_u128(362));
|
||||||
|
let agent_id = AgentId::from_uuid(Uuid::from_u128(1));
|
||||||
|
let key = domain::RuntimeAgentKey::new(project_id, agent_id);
|
||||||
|
let turn = StructuredHumanTurnGuard::start(
|
||||||
|
Arc::clone(&input) as Arc<dyn domain::input::InputMediator>,
|
||||||
|
Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
|
||||||
|
Arc::new(application::RestartRendezvousCoordinator::new()),
|
||||||
|
project_id,
|
||||||
|
agent_id,
|
||||||
|
"first",
|
||||||
|
);
|
||||||
|
|
||||||
|
turn.finish();
|
||||||
|
turn.finish();
|
||||||
|
|
||||||
|
assert_eq!(input.busy_state(key), domain::input::AgentBusyState::Idle);
|
||||||
|
assert_eq!(mailbox.pending(&key), 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn late_human_turn_finish_cannot_release_a_successor() {
|
||||||
|
let mailbox = Arc::new(InMemoryMailbox::new());
|
||||||
|
let input = Arc::new(MediatedInbox::new(
|
||||||
|
Arc::clone(&mailbox),
|
||||||
|
Arc::new(SystemMillisClock),
|
||||||
|
));
|
||||||
|
let project_id = ProjectId::from_uuid(Uuid::from_u128(363));
|
||||||
|
let agent_id = AgentId::from_uuid(Uuid::from_u128(1));
|
||||||
|
let key = domain::RuntimeAgentKey::new(project_id, agent_id);
|
||||||
|
let turn = StructuredHumanTurnGuard::start(
|
||||||
|
Arc::clone(&input) as Arc<dyn domain::input::InputMediator>,
|
||||||
|
Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
|
||||||
|
Arc::new(application::RestartRendezvousCoordinator::new()),
|
||||||
|
project_id,
|
||||||
|
agent_id,
|
||||||
|
"first",
|
||||||
|
);
|
||||||
|
let (_, first_ticket) = turn.accounting_identity().expect("first turn owns Busy");
|
||||||
|
|
||||||
|
// Stop owns the first release, then the FIFO starts its successor before a
|
||||||
|
// late Final/close/timeout path reaches the original guard.
|
||||||
|
mailbox.cancel_head(key, first_ticket);
|
||||||
|
input.mark_idle(key);
|
||||||
|
let successor = domain::mailbox::Ticket::new(
|
||||||
|
domain::mailbox::TicketId::new_random(),
|
||||||
|
"user",
|
||||||
|
"successor",
|
||||||
|
);
|
||||||
|
let successor_ticket = successor.id;
|
||||||
|
let _pending = input.enqueue_silent(key, successor);
|
||||||
|
|
||||||
|
turn.finish();
|
||||||
|
|
||||||
|
assert_eq!(input.busy_state(key).ticket(), Some(successor_ticket));
|
||||||
|
assert_eq!(mailbox.pending(&key), 1);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn cell_restart_stop_timeout_is_typed_and_never_reaches_the_next_step() {
|
async fn cell_restart_stop_timeout_is_typed_and_never_reaches_the_next_step() {
|
||||||
let agent = AgentId::from_uuid(Uuid::from_u128(317));
|
let agent = AgentId::from_uuid(Uuid::from_u128(317));
|
||||||
@ -1846,11 +1911,16 @@ impl StructuredHumanTurnGuard {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Retires the FIFO head and marks the agent idle. Idempotent across clones.
|
/// Retires the FIFO head and marks the agent idle. Idempotent across clones and
|
||||||
|
/// fenced against a late cleanup from an older turn.
|
||||||
pub fn finish(&self) {
|
pub fn finish(&self) {
|
||||||
if self.armed.swap(false, std::sync::atomic::Ordering::AcqRel) {
|
if self.armed.swap(false, std::sync::atomic::Ordering::AcqRel) {
|
||||||
self.mailbox.cancel_head(self.agent, self.ticket_id);
|
application::release_turn_accounting(
|
||||||
self.input.mark_idle(self.agent);
|
self.input.as_ref(),
|
||||||
|
self.mailbox.as_ref(),
|
||||||
|
self.agent,
|
||||||
|
self.ticket_id,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user