fix(runtime): cancel de background task immédiat et fiable depuis le workstate (#199, QA verte)
Transforme l'annulation d'une background task en contrat immédiat, observable et non-best-effort: - domain/background_task: états Cancelling/Cancelled/CancelFailed et transitions idempotentes; un kill PTY qui échoue est surfacé en CancelFailed plutôt qu'en succès silencieux. - infrastructure/runner + store: cancel effectif de la task ciblée, reconcile marque Cancelling même sans live handle, et départage CancelFailed des tâches pending au boot. - application/background + workstate + orchestrator/wake: chemin de cancel initié depuis le workstate, projection synchronisée de l'état (pas d'attente d'un cycle ambigu). - frontend: projection workstate de l'état cancelled/cancelling/cancelFailed (cancellingRow/cancelFailedRow/cancelButtons), normalization et services. Tests QA verts: background_task_cancel, background_task_runner cancel_wins_even_if_output_is_still_available, background_task_store reconcile_marks_cancelling_without_live_handle_cancel_failed_and_pending, workstate.test.tsx.
This commit is contained in:
@ -209,33 +209,103 @@ impl SpawnBackgroundCommand {
|
||||
/// Output of [`CancelBackgroundTask`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CancelBackgroundTaskOutput {
|
||||
/// The task as currently persisted, if it still exists. The terminal
|
||||
/// `Cancelled` state is written by the completion sink, not here.
|
||||
pub task: Option<BackgroundTask>,
|
||||
/// The task as currently persisted after the cancel request.
|
||||
pub task: BackgroundTask,
|
||||
}
|
||||
|
||||
/// Requests cancellation of a running background task.
|
||||
pub struct CancelBackgroundTask {
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
runner: Arc<dyn BackgroundTaskRunner>,
|
||||
clock: Arc<dyn Clock>,
|
||||
events: Option<Arc<dyn EventBus>>,
|
||||
}
|
||||
|
||||
impl CancelBackgroundTask {
|
||||
/// Wires the use case to its ports.
|
||||
#[must_use]
|
||||
pub fn new(store: Arc<dyn BackgroundTaskStore>, runner: Arc<dyn BackgroundTaskRunner>) -> Self {
|
||||
Self { store, runner }
|
||||
pub fn new(
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
runner: Arc<dyn BackgroundTaskRunner>,
|
||||
clock: Arc<dyn Clock>,
|
||||
) -> Self {
|
||||
Self {
|
||||
store,
|
||||
runner,
|
||||
clock,
|
||||
events: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Signals the runner to cancel `task_id` (idempotent).
|
||||
/// Attaches the public event stream publisher.
|
||||
#[must_use]
|
||||
pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
|
||||
self.events = Some(events);
|
||||
self
|
||||
}
|
||||
|
||||
/// Projects `Cancelling` immediately, then signals the runner to cancel `task_id`.
|
||||
///
|
||||
/// # Errors
|
||||
/// [`AppError`] if the runner or store fails.
|
||||
/// [`AppError`] if the task is missing, terminal, or the runner/store fails.
|
||||
pub async fn execute(&self, task_id: TaskId) -> Result<CancelBackgroundTaskOutput, AppError> {
|
||||
self.runner.cancel(task_id).await.map_err(map_port_err)?;
|
||||
let task = self.store.get(task_id).await.map_err(map_port_err)?;
|
||||
let task = self
|
||||
.store
|
||||
.get(task_id)
|
||||
.await
|
||||
.map_err(map_port_err)?
|
||||
.ok_or_else(|| AppError::NotFound(format!("background task {task_id}")))?;
|
||||
if task.is_terminal() {
|
||||
return Err(AppError::Invalid(format!(
|
||||
"background task {task_id} is already terminal ({:?})",
|
||||
task.state
|
||||
)));
|
||||
}
|
||||
|
||||
let cancelling = if task.state == BackgroundTaskState::Cancelling {
|
||||
task
|
||||
} else {
|
||||
let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);
|
||||
let cancelling = task
|
||||
.transition(BackgroundTaskState::Cancelling, now)
|
||||
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
||||
self.store.save(&cancelling).await.map_err(map_port_err)?;
|
||||
self.publish_state_changed(&cancelling);
|
||||
cancelling
|
||||
};
|
||||
|
||||
if let Err(err) = self.runner.cancel(task_id).await {
|
||||
let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);
|
||||
let failed = cancelling
|
||||
.complete(BackgroundTaskResult::CancelFailed {
|
||||
finished_at_ms: now,
|
||||
error: format!("failed to cancel background task: {err}"),
|
||||
})
|
||||
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
||||
self.store.save(&failed).await.map_err(map_port_err)?;
|
||||
self.publish_state_changed(&failed);
|
||||
return Err(map_port_err(err));
|
||||
}
|
||||
|
||||
let task = self
|
||||
.store
|
||||
.get(task_id)
|
||||
.await
|
||||
.map_err(map_port_err)?
|
||||
.unwrap_or(cancelling);
|
||||
Ok(CancelBackgroundTaskOutput { task })
|
||||
}
|
||||
|
||||
fn publish_state_changed(&self, task: &BackgroundTask) {
|
||||
if let Some(events) = &self.events {
|
||||
events.publish(DomainEvent::BackgroundTaskStateChanged {
|
||||
project_id: task.project_id,
|
||||
task_id: task.id,
|
||||
owner_agent_id: task.owner_agent_id,
|
||||
state: task.state,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Read-only recall of the command invocation a background task was spawned with.
|
||||
|
||||
@ -942,12 +942,14 @@ impl OrchestratorService {
|
||||
owner_agent_id,
|
||||
rendezvous,
|
||||
},
|
||||
BackgroundTaskResult::Failure { .. } => DomainEvent::BackgroundTaskFailed {
|
||||
project_id: project.id,
|
||||
task_id,
|
||||
owner_agent_id,
|
||||
rendezvous,
|
||||
},
|
||||
BackgroundTaskResult::Failure { .. } | BackgroundTaskResult::CancelFailed { .. } => {
|
||||
DomainEvent::BackgroundTaskFailed {
|
||||
project_id: project.id,
|
||||
task_id,
|
||||
owner_agent_id,
|
||||
rendezvous,
|
||||
}
|
||||
}
|
||||
BackgroundTaskResult::Cancelled { .. } | BackgroundTaskResult::Expired { .. } => {
|
||||
DomainEvent::BackgroundTaskCancelled {
|
||||
project_id: project.id,
|
||||
@ -1333,9 +1335,11 @@ impl OrchestratorService {
|
||||
BackgroundTaskState::Queued => "Queued",
|
||||
BackgroundTaskState::Running => "Running",
|
||||
BackgroundTaskState::Waiting => "Waiting",
|
||||
BackgroundTaskState::Cancelling => "Cancelling",
|
||||
BackgroundTaskState::Completed => "Completed",
|
||||
BackgroundTaskState::Failed => "Failed",
|
||||
BackgroundTaskState::Cancelled => "Cancelled",
|
||||
BackgroundTaskState::CancelFailed => "CancelFailed",
|
||||
BackgroundTaskState::Expired => "Expired",
|
||||
};
|
||||
Ok(OrchestratorOutcome {
|
||||
|
||||
@ -317,6 +317,10 @@ fn background_completion_prompt(task: &BackgroundTask) -> String {
|
||||
"La tâche de fond {} est terminée : annulée.\nRaison : {}",
|
||||
task.id, reason
|
||||
),
|
||||
Some(BackgroundTaskResult::CancelFailed { error, .. }) => format!(
|
||||
"La tâche de fond {} est terminée : échec de l'annulation.\nErreur : {}",
|
||||
task.id, error
|
||||
),
|
||||
Some(BackgroundTaskResult::Expired { reason, .. }) => format!(
|
||||
"La tâche de fond {} est terminée : expirée.\nRaison : {}",
|
||||
task.id, reason
|
||||
|
||||
@ -1940,9 +1940,11 @@ fn background_task_state_label(state: BackgroundTaskState) -> &'static str {
|
||||
BackgroundTaskState::Queued => "queued",
|
||||
BackgroundTaskState::Running => "running",
|
||||
BackgroundTaskState::Waiting => "waiting",
|
||||
BackgroundTaskState::Cancelling => "cancelling",
|
||||
BackgroundTaskState::Completed => "completed",
|
||||
BackgroundTaskState::Failed => "failed",
|
||||
BackgroundTaskState::Cancelled => "cancelled",
|
||||
BackgroundTaskState::CancelFailed => "cancelFailed",
|
||||
BackgroundTaskState::Expired => "expired",
|
||||
}
|
||||
}
|
||||
|
||||
@ -689,6 +689,9 @@ fn flatten_background_result(
|
||||
| Some(BackgroundTaskResult::Expired { reason, .. }) => {
|
||||
(None, Some(reason.clone()), None, None)
|
||||
}
|
||||
Some(BackgroundTaskResult::CancelFailed { error, .. }) => {
|
||||
(None, Some(error.clone()), None, None)
|
||||
}
|
||||
None => (None, None, None, None),
|
||||
}
|
||||
}
|
||||
|
||||
306
crates/application/tests/background_task_cancel.rs
Normal file
306
crates/application/tests/background_task_cancel.rs
Normal file
@ -0,0 +1,306 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use application::CancelBackgroundTask;
|
||||
use domain::ports::{
|
||||
BackgroundCompletionStream, BackgroundTaskHandle, BackgroundTaskPortError,
|
||||
BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore, Clock, EventBus, EventStream,
|
||||
};
|
||||
use domain::{
|
||||
AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState,
|
||||
BackgroundTaskWakePolicy, DomainEvent, ProjectId, TaskId,
|
||||
};
|
||||
use uuid::Uuid;
|
||||
|
||||
fn tid(n: u128) -> TaskId {
|
||||
TaskId::from_uuid(Uuid::from_u128(n))
|
||||
}
|
||||
|
||||
fn pid(n: u128) -> ProjectId {
|
||||
ProjectId::from_uuid(Uuid::from_u128(n))
|
||||
}
|
||||
|
||||
fn aid(n: u128) -> AgentId {
|
||||
AgentId::from_uuid(Uuid::from_u128(n))
|
||||
}
|
||||
|
||||
fn running_task(id: TaskId) -> BackgroundTask {
|
||||
BackgroundTask::new(
|
||||
id,
|
||||
pid(1),
|
||||
aid(2),
|
||||
BackgroundTaskKind::Command {
|
||||
label: "build".to_owned(),
|
||||
},
|
||||
BackgroundTaskWakePolicy::WakeOwner,
|
||||
100,
|
||||
None,
|
||||
)
|
||||
.unwrap()
|
||||
.transition(BackgroundTaskState::Running, 110)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FakeStore {
|
||||
tasks: Mutex<HashMap<TaskId, BackgroundTask>>,
|
||||
saves: Mutex<Vec<BackgroundTaskState>>,
|
||||
}
|
||||
|
||||
impl FakeStore {
|
||||
fn insert(&self, task: BackgroundTask) {
|
||||
self.tasks.lock().unwrap().insert(task.id, task);
|
||||
}
|
||||
|
||||
fn saved_states(&self) -> Vec<BackgroundTaskState> {
|
||||
self.saves.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl BackgroundTaskStore for FakeStore {
|
||||
async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
|
||||
self.insert(task.clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get(&self, id: TaskId) -> Result<Option<BackgroundTask>, BackgroundTaskPortError> {
|
||||
Ok(self.tasks.lock().unwrap().get(&id).cloned())
|
||||
}
|
||||
|
||||
async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
|
||||
self.saves.lock().unwrap().push(task.state);
|
||||
self.insert(task.clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_open_for_agent(
|
||||
&self,
|
||||
agent_id: AgentId,
|
||||
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
|
||||
Ok(self
|
||||
.tasks
|
||||
.lock()
|
||||
.unwrap()
|
||||
.values()
|
||||
.filter(|task| task.owner_agent_id == agent_id && !task.is_terminal())
|
||||
.cloned()
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn list_undelivered_completions(
|
||||
&self,
|
||||
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
|
||||
async fn mark_completion_delivered(
|
||||
&self,
|
||||
_task_id: TaskId,
|
||||
) -> Result<(), BackgroundTaskPortError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct FakeRunner {
|
||||
result: Mutex<Result<(), BackgroundTaskPortError>>,
|
||||
cancels: Mutex<Vec<TaskId>>,
|
||||
}
|
||||
|
||||
impl FakeRunner {
|
||||
fn ok() -> Self {
|
||||
Self {
|
||||
result: Mutex::new(Ok(())),
|
||||
cancels: Mutex::new(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
fn failing(message: &str) -> Self {
|
||||
Self {
|
||||
result: Mutex::new(Err(BackgroundTaskPortError::Runner(message.to_owned()))),
|
||||
cancels: Mutex::new(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
fn cancels(&self) -> Vec<TaskId> {
|
||||
self.cancels.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl BackgroundTaskRunner for FakeRunner {
|
||||
async fn spawn(
|
||||
&self,
|
||||
spec: BackgroundTaskSpec,
|
||||
) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
|
||||
Ok(BackgroundTaskHandle {
|
||||
task_id: spec.task_id,
|
||||
})
|
||||
}
|
||||
|
||||
async fn cancel(&self, task_id: TaskId) -> Result<(), BackgroundTaskPortError> {
|
||||
self.cancels.lock().unwrap().push(task_id);
|
||||
self.result.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
fn subscribe_completions(&self) -> BackgroundCompletionStream {
|
||||
Box::new(std::iter::empty())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct SpyEvents {
|
||||
events: Mutex<Vec<DomainEvent>>,
|
||||
}
|
||||
|
||||
impl SpyEvents {
|
||||
fn events(&self) -> Vec<DomainEvent> {
|
||||
self.events.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl EventBus for SpyEvents {
|
||||
fn publish(&self, event: DomainEvent) {
|
||||
self.events.lock().unwrap().push(event);
|
||||
}
|
||||
|
||||
fn subscribe(&self) -> EventStream {
|
||||
Box::new(std::iter::empty())
|
||||
}
|
||||
}
|
||||
|
||||
struct FixedClock(i64);
|
||||
|
||||
impl Clock for FixedClock {
|
||||
fn now_millis(&self) -> i64 {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
fn usecase(
|
||||
store: Arc<FakeStore>,
|
||||
runner: Arc<FakeRunner>,
|
||||
events: Arc<SpyEvents>,
|
||||
) -> CancelBackgroundTask {
|
||||
CancelBackgroundTask::new(store, runner, Arc::new(FixedClock(200))).with_events(events)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancel_running_projects_cancelling_immediately_then_calls_runner() {
|
||||
let store = Arc::new(FakeStore::default());
|
||||
let runner = Arc::new(FakeRunner::ok());
|
||||
let events = Arc::new(SpyEvents::default());
|
||||
store.insert(running_task(tid(10)));
|
||||
|
||||
let out = usecase(Arc::clone(&store), Arc::clone(&runner), Arc::clone(&events))
|
||||
.execute(tid(10))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(out.task.state, BackgroundTaskState::Cancelling);
|
||||
assert_eq!(store.saved_states(), vec![BackgroundTaskState::Cancelling]);
|
||||
assert_eq!(runner.cancels(), vec![tid(10)]);
|
||||
assert_eq!(
|
||||
events.events(),
|
||||
vec![DomainEvent::BackgroundTaskStateChanged {
|
||||
project_id: pid(1),
|
||||
task_id: tid(10),
|
||||
owner_agent_id: aid(2),
|
||||
state: BackgroundTaskState::Cancelling,
|
||||
}]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancel_terminal_task_is_explicit_error_without_runner_call() {
|
||||
let store = Arc::new(FakeStore::default());
|
||||
let runner = Arc::new(FakeRunner::ok());
|
||||
let events = Arc::new(SpyEvents::default());
|
||||
let done = running_task(tid(10))
|
||||
.complete(BackgroundTaskResult::Success {
|
||||
finished_at_ms: 120,
|
||||
exit_code: Some(0),
|
||||
summary: "ok".to_owned(),
|
||||
stdout_tail: None,
|
||||
stderr_tail: None,
|
||||
})
|
||||
.unwrap();
|
||||
store.insert(done);
|
||||
|
||||
let err = usecase(Arc::clone(&store), Arc::clone(&runner), events)
|
||||
.execute(tid(10))
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(err, application::AppError::Invalid(_)));
|
||||
assert!(runner.cancels().is_empty());
|
||||
assert!(store.saved_states().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancel_already_cancelling_is_idempotent_and_retries_runner_cancel() {
|
||||
let store = Arc::new(FakeStore::default());
|
||||
let runner = Arc::new(FakeRunner::ok());
|
||||
let events = Arc::new(SpyEvents::default());
|
||||
store.insert(
|
||||
running_task(tid(10))
|
||||
.transition(BackgroundTaskState::Cancelling, 120)
|
||||
.unwrap(),
|
||||
);
|
||||
|
||||
let out = usecase(Arc::clone(&store), Arc::clone(&runner), events)
|
||||
.execute(tid(10))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(out.task.state, BackgroundTaskState::Cancelling);
|
||||
assert_eq!(runner.cancels(), vec![tid(10)]);
|
||||
assert!(store.saved_states().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancel_kill_failure_persists_cancel_failed_and_returns_error() {
|
||||
let store = Arc::new(FakeStore::default());
|
||||
let runner = Arc::new(FakeRunner::failing("kill denied"));
|
||||
let events = Arc::new(SpyEvents::default());
|
||||
store.insert(running_task(tid(10)));
|
||||
|
||||
let err = usecase(Arc::clone(&store), Arc::clone(&runner), Arc::clone(&events))
|
||||
.execute(tid(10))
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(err, application::AppError::Process(_)));
|
||||
assert_eq!(
|
||||
store.saved_states(),
|
||||
vec![
|
||||
BackgroundTaskState::Cancelling,
|
||||
BackgroundTaskState::CancelFailed,
|
||||
]
|
||||
);
|
||||
let persisted = store.get(tid(10)).await.unwrap().unwrap();
|
||||
assert_eq!(persisted.state, BackgroundTaskState::CancelFailed);
|
||||
assert!(matches!(
|
||||
persisted.result,
|
||||
Some(BackgroundTaskResult::CancelFailed { .. })
|
||||
));
|
||||
assert_eq!(
|
||||
events.events(),
|
||||
vec![
|
||||
DomainEvent::BackgroundTaskStateChanged {
|
||||
project_id: pid(1),
|
||||
task_id: tid(10),
|
||||
owner_agent_id: aid(2),
|
||||
state: BackgroundTaskState::Cancelling,
|
||||
},
|
||||
DomainEvent::BackgroundTaskStateChanged {
|
||||
project_id: pid(1),
|
||||
task_id: tid(10),
|
||||
owner_agent_id: aid(2),
|
||||
state: BackgroundTaskState::CancelFailed,
|
||||
},
|
||||
]
|
||||
);
|
||||
}
|
||||
@ -568,9 +568,15 @@ fn background_task(
|
||||
.unwrap()
|
||||
.transition(BackgroundTaskState::Waiting, created_at_ms + 20)
|
||||
.unwrap(),
|
||||
BackgroundTaskState::Cancelling => base
|
||||
.transition(BackgroundTaskState::Running, created_at_ms + 10)
|
||||
.unwrap()
|
||||
.transition(BackgroundTaskState::Cancelling, created_at_ms + 20)
|
||||
.unwrap(),
|
||||
BackgroundTaskState::Completed
|
||||
| BackgroundTaskState::Failed
|
||||
| BackgroundTaskState::Cancelled
|
||||
| BackgroundTaskState::CancelFailed
|
||||
| BackgroundTaskState::Expired => {
|
||||
let result = match state {
|
||||
BackgroundTaskState::Completed => BackgroundTaskResult::Success {
|
||||
@ -591,13 +597,18 @@ fn background_task(
|
||||
finished_at_ms: created_at_ms + 20,
|
||||
reason: "cancelled".to_owned(),
|
||||
},
|
||||
BackgroundTaskState::CancelFailed => BackgroundTaskResult::CancelFailed {
|
||||
finished_at_ms: created_at_ms + 20,
|
||||
error: "cancel failed".to_owned(),
|
||||
},
|
||||
BackgroundTaskState::Expired => BackgroundTaskResult::Expired {
|
||||
finished_at_ms: created_at_ms + 20,
|
||||
reason: "expired".to_owned(),
|
||||
},
|
||||
BackgroundTaskState::Queued
|
||||
| BackgroundTaskState::Running
|
||||
| BackgroundTaskState::Waiting => unreachable!(),
|
||||
| BackgroundTaskState::Waiting
|
||||
| BackgroundTaskState::Cancelling => unreachable!(),
|
||||
};
|
||||
let running = base
|
||||
.transition(BackgroundTaskState::Running, created_at_ms + 10)
|
||||
|
||||
Reference in New Issue
Block a user