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

fix(application,backend,app-tauri,frontend): réconcilie l'état idle et les Finals après Stop — #357 (QA verte)

Browse Source 
Cause: après un cancel manuel de tour, le ticket médié busy/FIFO n'était pas
retiré (l'agent restait Busy, bouton Stop persistant) et un terminal tardif du
tour annulé pouvait bloquer le tour suivant (Final non affiché).

Backend: StructuredSessions lie chaque génération de tour à son ticket médié
(bind_turn_accounting) et request_turn_cancel restitue l'identité (agent,
ticket) exacte observée avant annulation; CancelAgentChat.with_turn_accounting
retire le ticket exact (cancel_head + mark_idle) seulement s'il est toujours
le ticket courant — jamais un successeur déjà remplacé;
StructuredHumanTurnGuard expose accounting_identity et agent_send (app-tauri)
lie la comptabilité au démarrage du tour humain.

Frontend: CustomAgentChatView réconcilie l'interaction active depuis le
work-state (agent idle sans délégation -> interaction terminée, fallback
« Tour terminé sans réponse finale. ») au lieu de laisser un agentBusyChanged
sans tour-id clore un tour plus récent; les terminaux tardifs (final/error) du
tour read-model-settled remplacent le fallback; tour suivant isolé des
callbacks tardifs de la génération précédente.

Tests: cancel_agent_chat (retrait du ticket exact avant le prompt suivant,
nettoyage tardif sans idle sur ticket successeur), CustomAgentChatView
(isolation tour suivant, réconciliation work-state + remplacement fallback par
Final tardif).

Réserve non bloquante: alignement fixture skills préexistant porté par le
commit test dédié 83659d9a.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-09-14 15:13:03 +02:00
parent 83659d9a09
commit 50e395b602
7 changed files with 414 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
crates/application/src/terminal/registry.rs
View File

@ -10,8 +10,11 @@ use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use domain::conversation::ConversationId;
use domain::mailbox::TicketId;
use domain::ports::{AgentSession, PtyHandle};
use domain::{AgentId, IssueRef, NodeId, ProjectId, SessionId, SessionKind, TerminalSession};
use domain::{
AgentId, IssueRef, NodeId, ProjectId, RuntimeAgentKey, SessionId, SessionKind, TerminalSession,
};
use crate::model_server::ModelServerUseGuard;
@ -541,6 +544,8 @@ struct StructuredEntry {
turn_generation: u64,
/// Current/last logical-turn state for idempotent request cancellation.
turn_state: StructuredTurnState,
/// Exact mediated ticket owned by the active generation, when it acquired one.
turn_accounting: Option<(u64, RuntimeAgentKey, TicketId)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@ -673,6 +678,7 @@ impl StructuredSessions {
_model_server_guard: model_server_guard,
turn_generation: 0,
turn_state: StructuredTurnState::Completed,
turn_accounting: None,
},
);
}
@ -712,10 +718,31 @@ impl StructuredSessions {
entry.turn_generation = entry.turn_generation.wrapping_add(1).max(1);
let generation = entry.turn_generation;
entry.turn_state = StructuredTurnState::Active(generation);
entry.turn_accounting = None;
Some(generation)
})
}
/// Correlates one logical provider generation with its mediated busy/FIFO ticket.
pub fn bind_turn_accounting(
&self,
id: &SessionId,
generation: u64,
agent: RuntimeAgentKey,
ticket: TicketId,
) -> bool {
self.entries.lock().ok().is_some_and(|mut entries| {
let Some(entry) = entries.get_mut(id) else {
return false;
};
if entry.turn_state != StructuredTurnState::Active(generation) {
return false;
}
entry.turn_accounting = Some((generation, agent, ticket));
true
})
}
/// Atomically records a cancellation request and returns the live session.
pub fn request_turn_cancel(
&self,
@ -723,6 +750,7 @@ impl StructuredSessions {
) -> Option<(
Arc<dyn AgentSession>,
domain::ports::CancelCurrentTurnOutcome,
Option<(RuntimeAgentKey, TicketId)>,
)> {
self.entries.lock().ok().and_then(|mut entries| {
let entry = entries.get_mut(id)?;
@ -738,7 +766,14 @@ impl StructuredSessions {
domain::ports::CancelCurrentTurnOutcome::AlreadyCompleted
}
};
Some((Arc::clone(&entry.session), outcome))
let accounting = match entry.turn_state {
StructuredTurnState::Cancelled(generation) => entry
.turn_accounting
.filter(|(bound, _, _)| *bound == generation)
.map(|(_, agent, ticket)| (agent, ticket)),
StructuredTurnState::Completed | StructuredTurnState::Active(_) => None,
};
Some((Arc::clone(&entry.session), outcome, accounting))
})
}
@ -758,6 +793,7 @@ impl StructuredSessions {
if let Some(entry) = entries.get_mut(id) {
if entry.turn_state == StructuredTurnState::Active(generation) {
entry.turn_state = StructuredTurnState::Completed;
entry.turn_accounting = None;
}
}
}

33
crates/application/src/terminal/usecases.rs
View File

@ -2,6 +2,8 @@
use std::sync::Arc;
use domain::input::InputMediator;
use domain::mailbox::AgentMailbox;
use domain::ports::{CancelCurrentTurnOutcome, EventBus, PtyPort, SpawnSpec};
use domain::{
AgentId, DomainEvent, NodeId, Project, ProjectPath, PtySize, SessionId, SessionKind,
@ -27,13 +29,28 @@ pub struct CancelAgentChatOutput {
/// Cancels one structured chat turn without stopping its reusable runtime.
pub struct CancelAgentChat {
sessions: Arc<StructuredSessions>,
turn_accounting: Option<(Arc<dyn InputMediator>, Arc<dyn AgentMailbox>)>,
}
impl CancelAgentChat {
/// Builds the use case from the structured runtime registry.
#[must_use]
pub fn new(sessions: Arc<StructuredSessions>) -> Self {
Self { sessions }
Self {
sessions,
turn_accounting: None,
}
}
/// Wires the mediated busy/FIFO state owned by direct human turns.
#[must_use]
pub fn with_turn_accounting(
mut self,
input: Arc<dyn InputMediator>,
mailbox: Arc<dyn AgentMailbox>,
) -> Self {
self.turn_accounting = Some((input, mailbox));
self
}
/// Requests cancellation from the live session and deliberately leaves its
@ -43,7 +60,7 @@ impl CancelAgentChat {
/// Returns [`AppError::NotFound`] only when the session itself is absent, and
/// a process error when the adapter could not interrupt its current child.
pub async fn execute(&self, session_id: SessionId) -> Result<CancelAgentChatOutput, AppError> {
let (session, outcome) = self
let (session, outcome, active_ticket) = self
.sessions
.request_turn_cancel(&session_id)
.ok_or_else(|| AppError::NotFound(format!("structured session {session_id}")))?;
@ -52,6 +69,18 @@ impl CancelAgentChat {
self.sessions.restore_turn_after_cancel_failure(&session_id);
return Err(AppError::from(error));
}
if let (Some((agent, ticket)), Some((input, mailbox))) =
(active_ticket, self.turn_accounting.as_ref())
{
// The provider kill can race the old pump's normal guard cleanup and
// 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 {
session_id,

157
crates/application/tests/cancel_agent_chat.rs
View File

@ -5,13 +5,16 @@ use application::{
CancelAgentChat, LiveRuntimeState, LiveSessions, StructuredSessions, TerminalSessions,
};
use async_trait::async_trait;
use domain::input::{AgentBusyState, InputMediator, SubmitConfig};
use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId};
use domain::ports::{AgentSession, AgentSessionError, CancelCurrentTurnOutcome, ReplyStream};
use domain::{AgentId, NodeId, ProjectId, SessionId};
use domain::{AgentId, NodeId, ProjectId, RuntimeAgentKey, SessionId};
use uuid::Uuid;
struct FakeSession {
id: SessionId,
cancellations: Mutex<VecDeque<CancelCurrentTurnOutcome>>,
cancel_hook: Mutex<Option<Box<dyn FnOnce() + Send>>>,
sends: Mutex<Vec<String>>,
shutdowns: Mutex<usize>,
}
@ -21,10 +24,16 @@ impl FakeSession {
Self {
id,
cancellations: Mutex::new(outcomes.into_iter().collect()),
cancel_hook: Mutex::new(None),
sends: Mutex::new(Vec::new()),
shutdowns: Mutex::new(0),
}
}
fn with_cancel_hook(self, hook: impl FnOnce() + Send + 'static) -> Self {
*self.cancel_hook.lock().unwrap() = Some(Box::new(hook));
self
}
}
#[async_trait]
@ -43,12 +52,16 @@ impl AgentSession for FakeSession {
}
async fn cancel_current_turn(&self) -> Result<CancelCurrentTurnOutcome, AgentSessionError> {
Ok(self
let outcome = self
.cancellations
.lock()
.unwrap()
.pop_front()
.unwrap_or(CancelCurrentTurnOutcome::AlreadyCompleted))
.unwrap_or(CancelCurrentTurnOutcome::AlreadyCompleted);
if let Some(hook) = self.cancel_hook.lock().unwrap().take() {
hook();
}
Ok(outcome)
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
@ -61,6 +74,58 @@ fn id<T>(value: u128, make: impl FnOnce(Uuid) -> T) -> T {
make(Uuid::from_u128(value))
}
struct TurnAccounting {
state: Mutex<AgentBusyState>,
cancelled: Mutex<Vec<(RuntimeAgentKey, TicketId)>>,
}
impl Default for TurnAccounting {
fn default() -> Self {
Self {
state: Mutex::new(AgentBusyState::Idle),
cancelled: Mutex::new(Vec::new()),
}
}
}
impl InputMediator for TurnAccounting {
fn enqueue(&self, _agent: RuntimeAgentKey, _ticket: Ticket) -> PendingReply {
panic!("not used")
}
fn bind_handle_with_submit(
&self,
_agent: RuntimeAgentKey,
_handle: domain::ports::PtyHandle,
_submit: SubmitConfig,
) {
}
fn preempt(&self, _agent: RuntimeAgentKey) {}
fn mark_idle(&self, _agent: RuntimeAgentKey) {
*self.state.lock().unwrap() = AgentBusyState::Idle;
}
fn busy_state(&self, _agent: RuntimeAgentKey) -> AgentBusyState {
*self.state.lock().unwrap()
}
}
impl AgentMailbox for TurnAccounting {
fn enqueue(&self, _agent: RuntimeAgentKey, _ticket: Ticket) -> PendingReply {
panic!("not used")
}
fn resolve(&self, agent: RuntimeAgentKey, _result: String) -> Result<(), MailboxError> {
Err(MailboxError::NoPendingRequest(agent.agent_id))
}
fn cancel_head(&self, agent: RuntimeAgentKey, ticket_id: TicketId) {
self.cancelled.lock().unwrap().push((agent, ticket_id));
}
}
#[tokio::test]
async fn cancel_keeps_session_and_accepts_a_second_prompt_without_shutdown() {
let session_id = id(1, SessionId::from_uuid);
@ -140,3 +205,89 @@ async fn double_cancel_is_typed_and_missing_session_remains_distinct() {
assert_eq!(error.code(), "NOT_FOUND");
assert!(sessions.session(&session_id).is_some());
}
#[tokio::test]
async fn cancel_retires_the_exact_busy_turn_before_the_next_prompt() {
let session_id = id(20, SessionId::from_uuid);
let project_id = id(21, ProjectId::from_uuid);
let agent_id = id(22, AgentId::from_uuid);
let ticket = TicketId::from_uuid(Uuid::from_u128(23));
let accounting = Arc::new(TurnAccounting::default());
*accounting.state.lock().unwrap() = AgentBusyState::Busy {
ticket,
since_ms: 1,
};
let session = Arc::new(FakeSession::new(
session_id,
[CancelCurrentTurnOutcome::Cancelled],
));
let sessions = Arc::new(StructuredSessions::new());
sessions.insert_in_project(project_id, session, agent_id, id(24, NodeId::from_uuid));
let generation = sessions.begin_turn(&session_id).unwrap();
assert!(sessions.bind_turn_accounting(
&session_id,
generation,
RuntimeAgentKey::new(project_id, agent_id),
ticket,
));
let input = Arc::clone(&accounting) as Arc<dyn InputMediator>;
let mailbox = Arc::clone(&accounting) as Arc<dyn AgentMailbox>;
let cancel = CancelAgentChat::new(Arc::clone(&sessions)).with_turn_accounting(input, mailbox);
let output = cancel.execute(session_id).await.unwrap();
assert_eq!(output.outcome, CancelCurrentTurnOutcome::Cancelled);
assert_eq!(*accounting.state.lock().unwrap(), AgentBusyState::Idle);
assert_eq!(
*accounting.cancelled.lock().unwrap(),
[(RuntimeAgentKey::new(project_id, agent_id), ticket)]
);
}
#[tokio::test]
async fn late_cancel_cleanup_does_not_mark_a_successor_ticket_idle() {
let session_id = id(30, SessionId::from_uuid);
let project_id = id(31, ProjectId::from_uuid);
let agent_id = id(32, AgentId::from_uuid);
let cancelled_ticket = TicketId::from_uuid(Uuid::from_u128(33));
let successor_ticket = TicketId::from_uuid(Uuid::from_u128(34));
let accounting = Arc::new(TurnAccounting::default());
*accounting.state.lock().unwrap() = AgentBusyState::Busy {
ticket: cancelled_ticket,
since_ms: 1,
};
let accounting_during_cancel = Arc::clone(&accounting);
let session = Arc::new(
FakeSession::new(session_id, [CancelCurrentTurnOutcome::Cancelled]).with_cancel_hook(
move || {
*accounting_during_cancel.state.lock().unwrap() = AgentBusyState::Busy {
ticket: successor_ticket,
since_ms: 2,
};
},
),
);
let sessions = Arc::new(StructuredSessions::new());
sessions.insert_in_project(project_id, session, agent_id, id(35, NodeId::from_uuid));
let generation = sessions.begin_turn(&session_id).unwrap();
assert!(sessions.bind_turn_accounting(
&session_id,
generation,
RuntimeAgentKey::new(project_id, agent_id),
cancelled_ticket,
));
let input = Arc::clone(&accounting) as Arc<dyn InputMediator>;
let mailbox = Arc::clone(&accounting) as Arc<dyn AgentMailbox>;
let cancel = CancelAgentChat::new(Arc::clone(&sessions)).with_turn_accounting(input, mailbox);
cancel.execute(session_id).await.unwrap();
assert_eq!(
*accounting.state.lock().unwrap(),
AgentBusyState::Busy {
ticket: successor_ticket,
since_ms: 2,
}
);
assert!(accounting.cancelled.lock().unwrap().is_empty());
}