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

fix(domain,application,infrastructure,backend,app-tauri,frontend): Stop demande agent préserve la conversation ouverte — arrêt sans fermeture/clôture du chat, états cohérents après cancel sur les 3 adapters de session, cellule et workstate synchronisés — #343 (QA verte)

Browse Source
This commit is contained in:
Blomios
2026-09-13 10:03:36 +02:00
parent a1e8a6466f
commit c37638ff24
26 changed files with 1210 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/infrastructure/src/session/claude.rs
View File

@ -395,7 +395,9 @@ impl AgentSession for ClaudeSdkSession {
Ok(())
}
async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
async fn cancel_current_turn(
&self,
) -> Result<domain::ports::CancelCurrentTurnOutcome, AgentSessionError> {
self.turn_cancel.cancel().await
}
}

4
crates/infrastructure/src/session/codex.rs
View File

@ -628,7 +628,9 @@ impl AgentSession for CodexExecSession {
Ok(())
}
async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
async fn cancel_current_turn(
&self,
) -> Result<domain::ports::CancelCurrentTurnOutcome, AgentSessionError> {
self.turn_cancel.cancel().await
}
}

54
crates/infrastructure/src/session/opencode.rs
View File

@ -23,6 +23,8 @@ use domain::ports::{
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
use domain::SessionId;
use super::process::{TurnCancel, TurnLease};
/// Un événement OpenCode parsé depuis stdout JSONL.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ParsedEvent {
@ -553,25 +555,12 @@ pub struct OpenCodeSession {
project_root: String,
env: Vec<(String, String)>,
engine_session_id: Mutex<Option<String>>,
current_child: Mutex<Option<Arc<AsyncMutex<tokio::process::Child>>>>,
turn_cancel: TurnCancel,
sandbox: Option<SandboxPlan>,
sandbox_enforcer: Option<Arc<dyn SandboxEnforcer>>,
}
impl OpenCodeSession {
fn clear_current_child(&self, child: &Arc<AsyncMutex<tokio::process::Child>>) {
let mut current = self
.current_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if current
.as_ref()
.is_some_and(|stored| Arc::ptr_eq(stored, child))
{
*current = None;
}
}
/// Construit l'adapter. `command_prefix` peut être `opencode`, un chemin absolu,
/// ou un wrapper avec arguments; IdeA ajoute ensuite `run --format json`.
pub fn new(
@ -596,7 +585,7 @@ impl OpenCodeSession {
project_root: project_root.into(),
env,
engine_session_id: Mutex::new(seed),
current_child: Mutex::new(None),
turn_cancel: TurnCancel::new(),
sandbox,
sandbox_enforcer,
})
@ -652,21 +641,10 @@ impl AgentSession for OpenCodeSession {
Ok(())
}
async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
let child = self
.current_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
if let Some(child) = child {
child
.lock()
.await
.kill()
.await
.map_err(|e| AgentSessionError::Io(format!("annulation OpenCode: {e}")))?;
}
Ok(())
async fn cancel_current_turn(
&self,
) -> Result<domain::ports::CancelCurrentTurnOutcome, AgentSessionError> {
self.turn_cancel.cancel().await
}
}
@ -676,9 +654,10 @@ impl OpenCodeSession {
prompt: &str,
tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
) -> Result<ReplyStream, AgentSessionError> {
let turn = self.turn_cancel.begin();
let initial_session = self.conversation_id();
let first = self
.run_once(prompt, tap.clone(), initial_session.clone())
.run_once(prompt, tap.clone(), initial_session.clone(), &turn)
.await;
let events = match first {
Ok(events) => events,
@ -686,7 +665,7 @@ impl OpenCodeSession {
if initial_session.is_some()
&& matches!(err, AgentSessionError::Decode(ref message) if opencode_missing_final(message)) =>
{
self.run_once(prompt, tap, None).await?
self.run_once(prompt, tap, None, &turn).await?
}
Err(err) => return Err(err),
};
@ -698,6 +677,7 @@ impl OpenCodeSession {
prompt: &str,
tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
expected_session: Option<String>,
turn: &TurnLease,
) -> Result<Vec<ReplyEvent>, AgentSessionError> {
let mut cmd = Command::new(&self.command);
cmd.args(self.build_args(prompt, expected_session.as_deref()))
@ -718,13 +698,7 @@ impl OpenCodeSession {
.spawn()
.map_err(|e| AgentSessionError::Start(format!("{}: {e}", self.command)))?;
let child = Arc::new(AsyncMutex::new(child));
{
let mut current = self
.current_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
*current = Some(Arc::clone(&child));
}
turn.track_async(Arc::clone(&child)).await?;
let outcome = async {
let (stdout, mut stderr_pipe) =
{
@ -784,7 +758,7 @@ impl OpenCodeSession {
Ok((expected_session, collected, stderr_bytes, status))
}
.await;
self.clear_current_child(&child);
turn.clear_async(&child);
let (expected_session, collected, stderr_bytes, status) = outcome?;
let stderr = String::from_utf8_lossy(&stderr_bytes);
let parsed_session = expected_session.or_else(|| self.conversation_id());

310
crates/infrastructure/src/session/process.rs
View File

@ -36,7 +36,7 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;
use tokio::sync::Mutex as AsyncMutex;
use domain::ports::AgentSessionError;
use domain::ports::{AgentSessionError, CancelCurrentTurnOutcome};
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
/// Une invocation orientée lignes : binaire + arguments + cwd + env + prompt à
@ -64,10 +64,68 @@ pub struct SpawnLine {
/// Shared handle used by structured adapters to cancel the process backing the
/// current turn without shutting down the reusable session.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LastTurnState {
Completed,
Cancelled,
}
#[derive(Default)]
struct TurnCancelState {
next_generation: u64,
active: Option<(u64, bool)>,
last: Option<LastTurnState>,
}
type AsyncTurnChild = Arc<AsyncMutex<tokio::process::Child>>;
type StdTurnChild = Arc<StdMutex<std::process::Child>>;
type TrackedAsyncTurnChild = Option<(u64, AsyncTurnChild)>;
type TrackedStdTurnChild = Option<(u64, StdTurnChild)>;
/// Generation-aware handle for cancelling the process backing one current turn.
#[derive(Clone, Default)]
pub struct TurnCancel {
async_child: Arc<StdMutex<Option<Arc<AsyncMutex<tokio::process::Child>>>>>,
std_child: Arc<StdMutex<Option<Arc<StdMutex<std::process::Child>>>>>,
state: Arc<StdMutex<TurnCancelState>>,
async_child: Arc<StdMutex<TrackedAsyncTurnChild>>,
std_child: Arc<StdMutex<TrackedStdTurnChild>>,
}
#[derive(Clone)]
struct TurnToken {
cancel: TurnCancel,
generation: u64,
}
pub(super) struct TurnLease {
token: TurnToken,
}
impl TurnLease {
pub(super) async fn track_async(&self, child: AsyncTurnChild) -> Result<(), AgentSessionError> {
if !self
.token
.cancel
.set_async(self.token.generation, Arc::clone(&child))
{
child
.lock()
.await
.kill()
.await
.map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?;
}
Ok(())
}
pub(super) fn clear_async(&self, child: &AsyncTurnChild) {
self.token.cancel.clear_async(self.token.generation, child);
}
}
impl Drop for TurnLease {
fn drop(&mut self) {
self.token.cancel.finish(self.token.generation);
}
}
impl TurnCancel {
@ -77,76 +135,166 @@ impl TurnCancel {
Self::default()
}
fn set_async(&self, child: Arc<AsyncMutex<tokio::process::Child>>) {
pub(super) fn begin(&self) -> TurnLease {
let generation = {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.next_generation = state.next_generation.wrapping_add(1).max(1);
let generation = state.next_generation;
state.active = Some((generation, false));
generation
};
TurnLease {
token: TurnToken {
cancel: self.clone(),
generation,
},
}
}
fn finish(&self, generation: u64) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some((active_generation, cancel_requested)) = state.active {
if active_generation == generation {
state.active = None;
state.last = Some(if cancel_requested {
LastTurnState::Cancelled
} else {
LastTurnState::Completed
});
}
}
}
fn restore_after_cancel_failure(&self, generation: u64) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state.active == Some((generation, true)) {
state.active = Some((generation, false));
}
}
fn set_async(&self, generation: u64, child: AsyncTurnChild) -> bool {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !matches!(state.active, Some((active, false)) if active == generation) {
return false;
}
*self
.async_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child);
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some((generation, child));
true
}
fn clear_async(&self, child: &Arc<AsyncMutex<tokio::process::Child>>) {
fn clear_async(&self, generation: u64, child: &AsyncTurnChild) {
let mut current = self
.async_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if current
.as_ref()
.is_some_and(|stored| Arc::ptr_eq(stored, child))
{
if current.as_ref().is_some_and(|(stored_generation, stored)| {
*stored_generation == generation && Arc::ptr_eq(stored, child)
}) {
*current = None;
}
}
fn set_std(&self, child: Arc<StdMutex<std::process::Child>>) {
fn set_std(&self, generation: u64, child: StdTurnChild) -> bool {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !matches!(state.active, Some((active, false)) if active == generation) {
return false;
}
*self
.std_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child);
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some((generation, child));
true
}
fn clear_std(&self, child: &Arc<StdMutex<std::process::Child>>) {
fn clear_std(&self, generation: u64, child: &StdTurnChild) {
let mut current = self
.std_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if current
.as_ref()
.is_some_and(|stored| Arc::ptr_eq(stored, child))
{
if current.as_ref().is_some_and(|(stored_generation, stored)| {
*stored_generation == generation && Arc::ptr_eq(stored, child)
}) {
*current = None;
}
}
/// Kills the current turn process, if any. Missing/already-finished is a no-op.
pub async fn cancel(&self) -> Result<(), AgentSessionError> {
/// Kills the current turn process, if any, without shutting down the session.
pub async fn cancel(&self) -> Result<CancelCurrentTurnOutcome, AgentSessionError> {
let generation = {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
match state.active {
Some((_, true)) => return Ok(CancelCurrentTurnOutcome::AlreadyCancelled),
Some((generation, false)) => {
state.active = Some((generation, true));
generation
}
None => {
return Ok(match state.last {
Some(LastTurnState::Cancelled) => {
CancelCurrentTurnOutcome::AlreadyCancelled
}
Some(LastTurnState::Completed) | None => {
CancelCurrentTurnOutcome::AlreadyCompleted
}
});
}
}
};
let async_child = self
.async_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
.as_ref()
.filter(|(child_generation, _)| *child_generation == generation)
.map(|(_, child)| Arc::clone(child));
if let Some(child) = async_child {
child
.lock()
.await
.kill()
.await
.map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?;
if let Err(error) = child.lock().await.kill().await {
self.restore_after_cancel_failure(generation);
return Err(AgentSessionError::Io(format!(
"annulation du tour: {error}"
)));
}
}
let std_child = self
.std_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
.as_ref()
.filter(|(child_generation, _)| *child_generation == generation)
.map(|(_, child)| Arc::clone(child));
if let Some(child) = std_child {
let mut child = child
.lock()
.map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))?;
child
.kill()
.map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?;
if let Err(error) = child.kill() {
self.restore_after_cancel_failure(generation);
return Err(AgentSessionError::Io(format!(
"annulation du tour: {error}"
)));
}
}
Ok(())
Ok(CancelCurrentTurnOutcome::Cancelled)
}
}
@ -185,7 +333,8 @@ pub async fn run_turn_cancellable(
line_tap: Option<Sender<String>>,
cancel: &TurnCancel,
) -> Result<Vec<String>, AgentSessionError> {
run_turn_impl(spec, timeout, enforcer, line_tap, Some(cancel)).await
let lease = cancel.begin();
run_turn_impl(spec, timeout, enforcer, line_tap, Some(&lease.token)).await
}
async fn run_turn_impl(
@ -193,7 +342,7 @@ async fn run_turn_impl(
timeout: Option<Duration>,
enforcer: Option<&Arc<dyn SandboxEnforcer>>,
line_tap: Option<Sender<String>>,
cancel: Option<&TurnCancel>,
cancel: Option<&TurnToken>,
) -> Result<Vec<String>, AgentSessionError> {
// Chemin SANDBOXÉ (Linux + plan posé + enforcer câblé) : transpose la technique
// du PTY (`spawn_command_sandboxed`) — enforce sur un thread jetable AVANT le fork,
@ -220,7 +369,7 @@ async fn run_turn_impl(
Ok(result) => result,
Err(_elapsed) => {
if let Some(cancel) = cancel {
let _ = cancel.cancel().await;
let _ = cancel.cancel.cancel().await;
}
Err(AgentSessionError::Timeout)
}
@ -267,7 +416,7 @@ async fn run_turn_sandboxed(
enforcer: Arc<dyn SandboxEnforcer>,
timeout: Option<Duration>,
line_tap: Option<Sender<String>>,
cancel: Option<TurnCancel>,
cancel: Option<TurnToken>,
) -> Result<Vec<String>, AgentSessionError> {
// Données possédées : rien n'emprunte le thread jetable.
let command = spec.command.clone();
@ -339,7 +488,7 @@ fn drain_sandboxed(
plan: &SandboxPlan,
killer_tx: tokio::sync::oneshot::Sender<Arc<std::sync::Mutex<std::process::Child>>>,
line_tap: Option<Sender<String>>,
cancel: Option<TurnCancel>,
cancel: Option<TurnToken>,
) -> Result<Vec<String>, AgentSessionError> {
use std::io::{BufRead, BufReader as StdBufReader, Read as _, Write as _};
use std::process::{Command as StdCommand, Stdio};
@ -401,7 +550,13 @@ fn drain_sandboxed(
});
let child = Arc::new(StdMutex::new(child));
if let Some(cancel) = &cancel {
cancel.set_std(Arc::clone(&child));
if !cancel.cancel.set_std(cancel.generation, Arc::clone(&child)) {
child
.lock()
.map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))?
.kill()
.map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?;
}
}
// Donne au côté async de quoi tuer l'enfant en cas de timeout.
let _ = killer_tx.send(Arc::clone(&child));
@ -427,7 +582,7 @@ fn drain_sandboxed(
.wait()
.map_err(|e| AgentSessionError::Io(e.to_string()))?;
if let Some(cancel) = &cancel {
cancel.clear_std(&child);
cancel.cancel.clear_std(cancel.generation, &child);
}
let stderr_bytes = stderr_handle
.join()
@ -446,7 +601,7 @@ fn drain_sandboxed(
async fn drain(
spec: &SpawnLine,
line_tap: Option<Sender<String>>,
cancel: Option<&TurnCancel>,
cancel: Option<&TurnToken>,
) -> Result<Vec<String>, AgentSessionError> {
let mut cmd = Command::new(&spec.command);
cmd.args(&spec.args)
@ -465,7 +620,17 @@ async fn drain(
.map_err(|e| AgentSessionError::Start(format!("{}: {e}", spec.command)))?;
let child = Arc::new(AsyncMutex::new(child));
if let Some(cancel) = cancel {
cancel.set_async(Arc::clone(&child));
if !cancel
.cancel
.set_async(cancel.generation, Arc::clone(&child))
{
child
.lock()
.await
.kill()
.await
.map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?;
}
}
// Pousse le prompt sur stdin puis le ferme (EOF) — sans quoi un process en mode
@ -525,7 +690,7 @@ async fn drain(
// on remonte le stderr réel au lieu de laisser l'appelant conclure à un no-Final.
let status = child.lock().await.wait().await.map_err(map_io)?;
if let Some(cancel) = cancel {
cancel.clear_async(&child);
cancel.cancel.clear_async(cancel.generation, &child);
}
let stderr_bytes = stderr_task
.await
@ -568,7 +733,7 @@ mod tests {
command: "/bin/sh".to_owned(),
args: vec![
"-c".to_owned(),
"printf 'started\\n'; sleep 10; printf 'late\\n'".to_owned(),
"printf 'started\\n'; exec sleep 10".to_owned(),
],
cwd: "/".to_owned(),
env: Vec::new(),
@ -583,7 +748,14 @@ mod tests {
});
tokio::time::sleep(Duration::from_millis(100)).await;
cancel.cancel().await.expect("cancel current child");
assert_eq!(
cancel.cancel().await.expect("cancel current child"),
CancelCurrentTurnOutcome::Cancelled
);
assert_eq!(
cancel.cancel().await.expect("repeat cancel"),
CancelCurrentTurnOutcome::AlreadyCancelled
);
let lines = worker.await.expect("worker joins").expect("drain result");
assert_eq!(lines, vec!["started"]);
@ -591,5 +763,57 @@ mod tests {
started.elapsed() < Duration::from_secs(3),
"cancel should not wait for the child process natural exit"
);
assert_eq!(
cancel.cancel().await.expect("cancel after cancelled turn"),
CancelCurrentTurnOutcome::AlreadyCancelled
);
}
#[tokio::test]
async fn cancel_after_natural_completion_is_distinct_from_a_missing_session() {
let spec = SpawnLine {
command: "/bin/sh".to_owned(),
args: vec!["-c".to_owned(), "printf 'done\\n'".to_owned()],
cwd: "/".to_owned(),
env: Vec::new(),
stdin: None,
sandbox: None,
};
let cancel = TurnCancel::new();
let lines = run_turn_cancellable(&spec, None, None, None, &cancel)
.await
.expect("turn completes");
assert_eq!(lines, ["done"]);
assert_eq!(
cancel.cancel().await.expect("terminal cancel is harmless"),
CancelCurrentTurnOutcome::AlreadyCompleted
);
}
#[tokio::test]
async fn cancel_before_child_registration_kills_that_same_generation_on_registration() {
let cancel = TurnCancel::new();
let turn = cancel.begin();
assert_eq!(
cancel.cancel().await.expect("pre-spawn cancel"),
CancelCurrentTurnOutcome::Cancelled
);
let child = Command::new("/bin/sh")
.args(["-c", "sleep 10"])
.spawn()
.expect("spawn delayed child");
let child = Arc::new(AsyncMutex::new(child));
turn.track_async(Arc::clone(&child))
.await
.expect("cancelled generation kills child at registration");
let status = tokio::time::timeout(Duration::from_secs(2), child.lock().await.wait())
.await
.expect("child exits promptly")
.expect("child wait succeeds");
assert!(!status.success());
}
}