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

fix(custom-cli): le bouton Stop annule le tour courant sans fermer la session et cascade l'annulation aux agents délégués — stop_agent suit l'arbre des dépendances actives (cancel_current_turn + préemption runtime + enfants d'exécution), TurnCancel tue le process du tour en cours, autorisation requester/sous-arbre sur idea_stop_agent, fixtures DTO runtime_state/node_id alignées — #257 (QA verte)

Browse Source
This commit is contained in:
Blomios
2026-08-28 22:57:32 +02:00
parent 9505a3305d
commit 1aabc3d702
12 changed files with 534 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,7 +25,7 @@ use domain::ports::{
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
use domain::SessionId;
use super::process::{run_turn, SpawnLine};
use super::process::{run_turn_cancellable, SpawnLine, TurnCancel};
/// Résultat du parsing d'une ligne : **zéro ou plusieurs** événements à émettre et/ou
/// un `session_id` capté (init/result). Permet à [`parse_event`] de rester **pure**
@ -306,6 +306,8 @@ pub struct ClaudeSdkSession {
/// Enforcer OS **par instance** (lot LP4-4), passé à [`run_turn`]. `None` ⇒ pas
/// de sandboxing même si un plan est présent (cohérent avec le chemin PTY).
sandbox_enforcer: Option<Arc<dyn SandboxEnforcer>>,
/// Cancel handle for the process backing the current `claude` turn.
turn_cancel: TurnCancel,
}
impl ClaudeSdkSession {
@ -330,6 +332,7 @@ impl ClaudeSdkSession {
conversation_id: Mutex::new(seed_conversation_id),
sandbox,
sandbox_enforcer,
turn_cancel: TurnCancel::new(),
}
}
@ -391,6 +394,10 @@ impl AgentSession for ClaudeSdkSession {
// tours, donc `shutdown` est intrinsèquement idempotent et sans effet.
Ok(())
}
async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
self.turn_cancel.cancel().await
}
}
impl ClaudeSdkSession {
@ -415,7 +422,14 @@ impl ClaudeSdkSession {
});
(Some(line_tx), Some(parser))
});
let raw_result = run_turn(&spec, None, self.sandbox_enforcer.as_ref(), line_tap).await;
let raw_result = run_turn_cancellable(
&spec,
None,
self.sandbox_enforcer.as_ref(),
line_tap,
&self.turn_cancel,
)
.await;
if let Some(parser) = parser_thread {
let _ = parser.join();
}

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

@ -22,7 +22,7 @@ use domain::ports::{
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
use domain::SessionId;
use super::process::{run_turn, SpawnLine};
use super::process::{run_turn_cancellable, SpawnLine, TurnCancel};
/// Résultat du parsing d'une ligne Codex : **zéro ou plusieurs** événements et/ou un
/// id de conversation Codex capté. Miroir de `claude::ParsedLine` (vecteur d'events
@ -377,6 +377,8 @@ pub struct CodexExecSession {
/// Enforcer OS **par instance** (lot LP4-4), passé à [`run_turn`]. `None` ⇒ pas
/// de sandboxing même si un plan est présent (cohérent avec le chemin PTY).
sandbox_enforcer: Option<Arc<dyn SandboxEnforcer>>,
/// Cancel handle for the process backing the current `codex exec` turn.
turn_cancel: TurnCancel,
}
impl CodexExecSession {
@ -473,6 +475,7 @@ impl CodexExecSession {
conversation_id: Mutex::new(seed_conversation_id),
sandbox,
sandbox_enforcer,
turn_cancel: TurnCancel::new(),
}
}
@ -594,6 +597,10 @@ impl AgentSession for CodexExecSession {
// « un run par tour » ⇒ pas de process long survivant : idempotent, no-op.
Ok(())
}
async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
self.turn_cancel.cancel().await
}
}
impl CodexExecSession {
@ -631,7 +638,14 @@ impl CodexExecSession {
});
(Some(line_tx), Some(parser))
});
let raw_result = run_turn(&spec, None, self.sandbox_enforcer.as_ref(), line_tap).await;
let raw_result = run_turn_cancellable(
&spec,
None,
self.sandbox_enforcer.as_ref(),
line_tap,
&self.turn_cancel,
)
.await;
if let Some(parser) = parser_thread {
let _ = parser.join();
}

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

@ -29,11 +29,12 @@
use std::io;
use std::process::Stdio;
use std::sync::mpsc::Sender;
use std::sync::Arc;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;
use tokio::sync::Mutex as AsyncMutex;
use domain::ports::AgentSessionError;
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
@ -61,6 +62,94 @@ pub struct SpawnLine {
pub sandbox: Option<SandboxPlan>,
}
/// Shared handle used by structured adapters to cancel the process backing the
/// current turn without shutting down the reusable session.
#[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>>>>>,
}
impl TurnCancel {
/// Creates an empty cancel handle with no active child.
#[must_use]
pub fn new() -> Self {
Self::default()
}
fn set_async(&self, child: Arc<AsyncMutex<tokio::process::Child>>) {
*self
.async_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child);
}
fn clear_async(&self, child: &Arc<AsyncMutex<tokio::process::Child>>) {
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))
{
*current = None;
}
}
fn set_std(&self, child: Arc<StdMutex<std::process::Child>>) {
*self
.std_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child);
}
fn clear_std(&self, child: &Arc<StdMutex<std::process::Child>>) {
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))
{
*current = None;
}
}
/// Kills the current turn process, if any. Missing/already-finished is a no-op.
pub async fn cancel(&self) -> Result<(), AgentSessionError> {
let async_child = self
.async_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
if let Some(child) = async_child {
child
.lock()
.await
.kill()
.await
.map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?;
}
let std_child = self
.std_child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
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}")))?;
}
Ok(())
}
}
/// Lance `spec`, pousse `spec.stdin` sur l'entrée standard, **draine toutes les
/// lignes de stdout jusqu'à EOF**, puis attend la fin du process. Retourne les
/// lignes brutes (sans `\n`) dans l'ordre d'émission.
@ -84,14 +173,42 @@ pub async fn run_turn(
timeout: Option<Duration>,
enforcer: Option<&Arc<dyn SandboxEnforcer>>,
line_tap: Option<Sender<String>>,
) -> Result<Vec<String>, AgentSessionError> {
run_turn_impl(spec, timeout, enforcer, line_tap, None).await
}
/// Same as [`run_turn`], but tracks the live child in `cancel` while the turn runs.
pub async fn run_turn_cancellable(
spec: &SpawnLine,
timeout: Option<Duration>,
enforcer: Option<&Arc<dyn SandboxEnforcer>>,
line_tap: Option<Sender<String>>,
cancel: &TurnCancel,
) -> Result<Vec<String>, AgentSessionError> {
run_turn_impl(spec, timeout, enforcer, line_tap, Some(cancel)).await
}
async fn run_turn_impl(
spec: &SpawnLine,
timeout: Option<Duration>,
enforcer: Option<&Arc<dyn SandboxEnforcer>>,
line_tap: Option<Sender<String>>,
cancel: Option<&TurnCancel>,
) -> 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,
// l'enfant hérite le domaine via fork+exec.
#[cfg(target_os = "linux")]
if let (Some(plan), Some(enforcer)) = (spec.sandbox.as_ref(), enforcer) {
return run_turn_sandboxed(spec, plan.clone(), Arc::clone(enforcer), timeout, line_tap)
.await;
return run_turn_sandboxed(
spec,
plan.clone(),
Arc::clone(enforcer),
timeout,
line_tap,
cancel.cloned(),
)
.await;
}
// Hors Linux : aucun sandboxing OS ⇒ on ignore l'enforcer (Noop de toute façon) et
// on garde le drain async historique. `let _` évite l'avertissement « unused ».
@ -99,11 +216,16 @@ pub async fn run_turn(
let _ = enforcer;
match timeout {
Some(dur) => match tokio::time::timeout(dur, drain(spec, line_tap)).await {
Some(dur) => match tokio::time::timeout(dur, drain(spec, line_tap, cancel)).await {
Ok(result) => result,
Err(_elapsed) => Err(AgentSessionError::Timeout),
Err(_elapsed) => {
if let Some(cancel) = cancel {
let _ = cancel.cancel().await;
}
Err(AgentSessionError::Timeout)
}
},
None => drain(spec, line_tap).await,
None => drain(spec, line_tap, cancel).await,
}
}
@ -145,9 +267,8 @@ async fn run_turn_sandboxed(
enforcer: Arc<dyn SandboxEnforcer>,
timeout: Option<Duration>,
line_tap: Option<Sender<String>>,
cancel: Option<TurnCancel>,
) -> Result<Vec<String>, AgentSessionError> {
use std::sync::Mutex as StdMutex;
// Données possédées : rien n'emprunte le thread jetable.
let command = spec.command.clone();
let args = spec.args.clone();
@ -163,7 +284,7 @@ async fn run_turn_sandboxed(
// Thread JETABLE : sa restriction Landlock meurt avec lui.
std::thread::spawn(move || {
let result = drain_sandboxed(
command, args, cwd, env, stdin, &enforcer, &plan, killer_tx, line_tap,
command, args, cwd, env, stdin, &enforcer, &plan, killer_tx, line_tap, cancel,
);
// Le récepteur peut avoir abandonné (timeout) : on ignore l'erreur d'envoi.
let _ = done_tx.send(result);
@ -218,6 +339,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>,
) -> Result<Vec<String>, AgentSessionError> {
use std::io::{BufRead, BufReader as StdBufReader, Read as _, Write as _};
use std::process::{Command as StdCommand, Stdio};
@ -278,6 +400,9 @@ fn drain_sandboxed(
.map_err(|e| AgentSessionError::Io(e.to_string()))
});
let child = Arc::new(StdMutex::new(child));
if let Some(cancel) = &cancel {
cancel.set_std(Arc::clone(&child));
}
// Donne au côté async de quoi tuer l'enfant en cas de timeout.
let _ = killer_tx.send(Arc::clone(&child));
@ -301,6 +426,9 @@ fn drain_sandboxed(
.map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))?
.wait()
.map_err(|e| AgentSessionError::Io(e.to_string()))?;
if let Some(cancel) = &cancel {
cancel.clear_std(&child);
}
let stderr_bytes = stderr_handle
.join()
.map_err(|_| AgentSessionError::Io("stderr thread terminé sans résultat".to_owned()))??;
@ -318,6 +446,7 @@ fn drain_sandboxed(
async fn drain(
spec: &SpawnLine,
line_tap: Option<Sender<String>>,
cancel: Option<&TurnCancel>,
) -> Result<Vec<String>, AgentSessionError> {
let mut cmd = Command::new(&spec.command);
cmd.args(&spec.args)
@ -331,14 +460,20 @@ async fn drain(
cmd.env(key, value);
}
let mut child = cmd
let child = cmd
.spawn()
.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));
}
// Pousse le prompt sur stdin puis le ferme (EOF) — sans quoi un process en mode
// `stream-json` attend indéfiniment. Erreur d'I/O ⇒ `Io` (pas `Start`).
if let Some(input) = &spec.stdin {
let mut stdin = child
.lock()
.await
.stdin
.take()
.ok_or_else(|| AgentSessionError::Io("stdin pipe indisponible".to_owned()))?;
@ -349,14 +484,18 @@ async fn drain(
// Drop explicite ⇒ EOF côté process.
drop(stdin);
} else {
drop(child.stdin.take());
drop(child.lock().await.stdin.take());
}
let stdout = child
.lock()
.await
.stdout
.take()
.ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?;
let mut stderr = child
.lock()
.await
.stderr
.take()
.ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?;
@ -384,7 +523,10 @@ async fn drain(
// n'est pas en soi une erreur de session si la CLI a imprimé une sortie
// structurée, captée ensuite par le parser. Sans stdout, il n'y a rien à parser :
// on remonte le stderr réel au lieu de laisser l'appelant conclure à un no-Final.
let status = child.wait().await.map_err(map_io)?;
let status = child.lock().await.wait().await.map_err(map_io)?;
if let Some(cancel) = cancel {
cancel.clear_async(&child);
}
let stderr_bytes = stderr_task
.await
.map_err(|e| AgentSessionError::Io(format!("lecture stderr: {e}")))?
@ -414,3 +556,40 @@ fn process_exit_message(command: &str, code: Option<i32>, stderr: &[u8]) -> Stri
format!("{command} exited with {status}: {stderr}")
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Instant;
#[tokio::test]
async fn run_turn_cancellable_kills_current_child_without_waiting_for_natural_exit() {
let spec = SpawnLine {
command: "/bin/sh".to_owned(),
args: vec![
"-c".to_owned(),
"printf 'started\\n'; sleep 10; printf 'late\\n'".to_owned(),
],
cwd: "/".to_owned(),
env: Vec::new(),
stdin: None,
sandbox: None,
};
let cancel = TurnCancel::new();
let started = Instant::now();
let worker = tokio::spawn({
let cancel = cancel.clone();
async move { run_turn_cancellable(&spec, None, None, None, &cancel).await }
});
tokio::time::sleep(Duration::from_millis(100)).await;
cancel.cancel().await.expect("cancel current child");
let lines = worker.await.expect("worker joins").expect("drain result");
assert_eq!(lines, vec!["started"]);
assert!(
started.elapsed() < Duration::from_secs(3),
"cancel should not wait for the child process natural exit"
);
}
}