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

feat(background): livraison auto au propriétaire + projection des tâches de fond dans le work-state (T1+T3)

Browse Source 
T1 — Wake automatique du propriétaire à la complétion.
La complétion d'une tâche de fond est désormais livrée à l'agent
propriétaire dès que la session accepte l'envoi (wake.rs :
mark_completion_delivered au send accepté), via un drain de flux dédié
(structured.rs : drain_reply_stream_with_readiness). L'inbox médiée
enfile l'item sans démarrer de tour ni marquer l'agent busy
(input/mod.rs : enqueue FIFO silencieux). Régression couverte
(tests/agent_wake.rs, tests input/mod.rs).

T3 — Tâches de fond projetées dans le read-model du panneau Work.
AgentWorkState porte désormais background_tasks
(VO AgentBackgroundTaskState), alimenté par un builder best-effort
with_background_tasks(store) : union list_open_for_agent + dispatch des
completions non livrées par owner_agent_id, erreur store => Vec vide
(aucune régression live/busy/tickets). DTO Tauri backgroundTasks et
wiring du BackgroundTaskStore côté state.rs. Le frontend, déjà câblé,
affiche Cancel/Retry (mapping queued/waiting -> pending, tri sur
updatedAtMs). Borne V1 : une tâche terminale déjà livrée n'est plus
énumérable (Retry limité à la fenêtre non livrée).

Tests : cargo build --workspace OK ; cargo test -p application /
-p app-tauri / -p infrastructure verts ; frontend build + vitest verts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-04 00:21:33 +02:00
parent f08dae62eb
commit eb9cc16181
21 changed files with 946 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
crates/application/tests/agent_wake.rs
View File

@ -298,6 +298,16 @@ impl BackgroundTaskStore for FakeTaskStore {
#[derive(Default)]
struct FakeSession {
prompts: Mutex<Vec<String>>,
events: Mutex<Option<Vec<ReplyEvent>>>,
}
impl FakeSession {
fn with_events(events: Vec<ReplyEvent>) -> Self {
Self {
prompts: Mutex::new(Vec::new()),
events: Mutex::new(Some(events)),
}
}
}
#[async_trait]
@ -312,12 +322,12 @@ impl AgentSession for FakeSession {
async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
self.prompts.lock().unwrap().push(prompt.to_owned());
Ok(Box::new(
let events = self.events.lock().unwrap().take().unwrap_or_else(|| {
vec![ReplyEvent::Final {
content: "ack".to_owned(),
}]
.into_iter(),
))
});
Ok(Box::new(events.into_iter()))
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
@ -421,15 +431,16 @@ async fn owner_busy_does_not_start_concurrent_wake_and_keeps_item_queued() {
.unwrap();
turns.force_busy(owner, ticket(99));
service(inbox.clone(), turns, tasks, sessions)
let err = service(inbox.clone(), turns, tasks, sessions)
.wake_agent(
&project,
owner,
WakeReason::BackgroundCompletion { task_id },
)
.await
.unwrap();
.unwrap_err();
assert_eq!(err, WakeError::AgentBusy { agent_id: owner });
assert!(session.prompts.lock().unwrap().is_empty());
assert_eq!(inbox.snapshot(owner).depth, 1);
}
@ -489,6 +500,35 @@ async fn completion_is_marked_delivered_after_successful_wake() {
assert_eq!(tasks.delivered(), vec![task_id]);
}
#[tokio::test]
async fn completion_is_marked_delivered_once_send_is_accepted_even_if_drain_fails() {
let project = project();
let owner = agent(1);
let task_id = task_id(10);
let inbox = Arc::new(FakeInbox::default());
let turns = Arc::new(SharedTurnState::default());
let tasks = Arc::new(FakeTaskStore::default());
let session = Arc::new(FakeSession::with_events(Vec::new()));
let sessions = Arc::new(FakeSessionProvider::with_session(session.clone()));
tasks.insert(completed_task(&project, owner, task_id, "done"));
inbox
.enqueue_message(owner, completion_item(owner, task_id, ticket(20)))
.unwrap();
let err = service(inbox, turns, tasks.clone(), sessions)
.wake_agent(
&project,
owner,
WakeReason::BackgroundCompletion { task_id },
)
.await
.unwrap_err();
assert!(matches!(err, WakeError::Session(_)));
assert_eq!(session.prompts.lock().unwrap().len(), 1);
assert_eq!(tasks.delivered(), vec![task_id]);
}
#[tokio::test]
async fn wake_drains_exactly_one_item_per_turn() {
let project = project();

333
crates/application/tests/workstate.rs
View File

@ -16,13 +16,15 @@ use domain::mailbox::{
AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, TurnResolution,
};
use domain::ports::{
AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError,
AgentContextStore, AgentSession, AgentSessionError, BackgroundTaskPortError,
BackgroundTaskStore, PtyHandle, ReplyStream, StoreError,
};
use domain::{
Agent, AgentBusyState, AgentId, AgentManifest, AgentOrigin, ConversationId, ConversationLog,
ConversationTurn, Handoff, HandoffStore, InputMediator, InputSource, ManifestEntry,
MarkdownDoc, NodeId, ProfileId, Project, ProjectId, ProjectPath, PtySize, RemoteRef, SessionId,
SessionKind, TerminalSession, TicketId, TurnId, TurnRole,
Agent, AgentBusyState, AgentId, AgentManifest, AgentOrigin, BackgroundTask, BackgroundTaskKind,
BackgroundTaskResult, BackgroundTaskState, BackgroundTaskWakePolicy, ConversationId,
ConversationLog, ConversationTurn, Handoff, HandoffStore, InputMediator, InputSource,
ManifestEntry, MarkdownDoc, NodeId, ProfileId, Project, ProjectId, ProjectPath, PtySize,
RemoteRef, SessionId, SessionKind, TaskId, TerminalSession, TicketId, TurnId, TurnRole,
};
use uuid::Uuid;
@ -46,6 +48,10 @@ fn ticket_id(n: u128) -> domain::TicketId {
domain::TicketId::from_uuid(Uuid::from_u128(n))
}
fn task_id(n: u128) -> TaskId {
TaskId::from_uuid(Uuid::from_u128(n))
}
fn project() -> Project {
Project::new(
ProjectId::from_uuid(Uuid::from_u128(1)),
@ -147,6 +153,106 @@ struct FakeQueue {
queues: Mutex<HashMap<AgentId, Vec<QueuedTicketSnapshot>>>,
}
#[derive(Default)]
struct FakeBackgroundTaskStore {
tasks: Mutex<Vec<BackgroundTask>>,
fail_open: Mutex<bool>,
fail_undelivered: Mutex<bool>,
}
impl FakeBackgroundTaskStore {
fn set_tasks(&self, tasks: Vec<BackgroundTask>) {
*self.tasks.lock().unwrap() = tasks;
}
fn fail_open(&self) {
*self.fail_open.lock().unwrap() = true;
}
fn fail_undelivered(&self) {
*self.fail_undelivered.lock().unwrap() = true;
}
}
#[async_trait]
impl BackgroundTaskStore for FakeBackgroundTaskStore {
async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
self.tasks.lock().unwrap().push(task.clone());
Ok(())
}
async fn get(&self, id: TaskId) -> Result<Option<BackgroundTask>, BackgroundTaskPortError> {
Ok(self
.tasks
.lock()
.unwrap()
.iter()
.find(|task| task.id == id)
.cloned())
}
async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
let mut tasks = self.tasks.lock().unwrap();
if let Some(existing) = tasks.iter_mut().find(|existing| existing.id == task.id) {
*existing = task.clone();
} else {
tasks.push(task.clone());
}
Ok(())
}
async fn list_open_for_agent(
&self,
agent_id: AgentId,
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
if *self.fail_open.lock().unwrap() {
return Err(BackgroundTaskPortError::Store("open failed".to_owned()));
}
Ok(self
.tasks
.lock()
.unwrap()
.iter()
.filter(|task| task.owner_agent_id == agent_id && !task.is_terminal())
.cloned()
.collect())
}
async fn list_undelivered_completions(
&self,
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
if *self.fail_undelivered.lock().unwrap() {
return Err(BackgroundTaskPortError::Store(
"undelivered failed".to_owned(),
));
}
Ok(self
.tasks
.lock()
.unwrap()
.iter()
.filter(|task| task.has_pending_completion_delivery())
.cloned()
.collect())
}
async fn mark_completion_delivered(
&self,
task_id: TaskId,
) -> Result<(), BackgroundTaskPortError> {
if let Some(task) = self
.tasks
.lock()
.unwrap()
.iter_mut()
.find(|task| task.id == task_id)
{
task.completion_delivered = true;
}
Ok(())
}
}
impl FakeQueue {
fn set(&self, agent: AgentId, tickets: Vec<QueuedTicketSnapshot>) {
self.queues.lock().unwrap().insert(agent, tickets);
@ -400,6 +506,106 @@ struct ConvFixture {
project: Project,
}
struct BackgroundFixture {
usecase: GetProjectWorkState,
queue: Arc<FakeQueue>,
store: Arc<FakeBackgroundTaskStore>,
project: Project,
}
fn background_fixture(agents: &[Agent]) -> BackgroundFixture {
let pty = Arc::new(TerminalSessions::new());
let structured = Arc::new(StructuredSessions::new());
let live = Arc::new(LiveSessions::new(pty, structured));
let input = Arc::new(FakeInput::default());
let queue = Arc::new(FakeQueue::default());
let store = Arc::new(FakeBackgroundTaskStore::default());
let usecase = GetProjectWorkState::new(
Arc::new(FakeContexts {
manifest: manifest(agents),
}),
live,
input as Arc<dyn InputMediator>,
Arc::clone(&queue) as Arc<dyn AgentQueueSnapshot>,
)
.with_background_tasks(Arc::clone(&store) as Arc<dyn BackgroundTaskStore>);
BackgroundFixture {
usecase,
queue,
store,
project: project(),
}
}
fn background_task(
id: u128,
project_id: ProjectId,
owner: AgentId,
created_at_ms: u64,
state: BackgroundTaskState,
delivered: bool,
) -> BackgroundTask {
let base = BackgroundTask::new(
task_id(id),
project_id,
owner,
BackgroundTaskKind::Command {
label: format!("task-{id}"),
},
BackgroundTaskWakePolicy::WakeOwner,
created_at_ms,
None,
)
.unwrap();
match state {
BackgroundTaskState::Queued => base,
BackgroundTaskState::Running | BackgroundTaskState::Waiting => {
base.transition(state, created_at_ms + 10).unwrap()
}
BackgroundTaskState::Completed
| BackgroundTaskState::Failed
| BackgroundTaskState::Cancelled
| BackgroundTaskState::Expired => {
let result = match state {
BackgroundTaskState::Completed => BackgroundTaskResult::Success {
finished_at_ms: created_at_ms + 20,
exit_code: Some(0),
summary: "done".to_owned(),
stdout_tail: Some("stdout".to_owned()),
stderr_tail: None,
},
BackgroundTaskState::Failed => BackgroundTaskResult::Failure {
finished_at_ms: created_at_ms + 20,
exit_code: Some(2),
error: "failed".to_owned(),
stdout_tail: Some("out".to_owned()),
stderr_tail: Some("err".to_owned()),
},
BackgroundTaskState::Cancelled => BackgroundTaskResult::Cancelled {
finished_at_ms: created_at_ms + 20,
reason: "cancelled".to_owned(),
},
BackgroundTaskState::Expired => BackgroundTaskResult::Expired {
finished_at_ms: created_at_ms + 20,
reason: "expired".to_owned(),
},
BackgroundTaskState::Queued
| BackgroundTaskState::Running
| BackgroundTaskState::Waiting => unreachable!(),
};
let running = base
.transition(BackgroundTaskState::Running, created_at_ms + 10)
.unwrap();
let completed = running.complete(result).unwrap();
if delivered {
completed.mark_completion_delivered().unwrap()
} else {
completed
}
}
}
}
/// Fixture wiring the best-effort conversation sources (handoff + log) onto the
/// read model, exposing the fake stores so each test configures their outcomes.
fn conv_fixture(agents: &[Agent]) -> ConvFixture {
@ -539,6 +745,123 @@ async fn workstate_agent_without_queue_has_no_tickets() {
assert!(out.agents[0].tickets.is_empty());
}
#[tokio::test]
async fn workstate_projects_open_and_undelivered_background_tasks_by_agent() {
let a = agent(10, "alpha");
let b = agent(20, "beta");
let f = background_fixture(&[a.clone(), b.clone()]);
let other_project = ProjectId::from_uuid(Uuid::from_u128(999));
f.store.set_tasks(vec![
background_task(
2,
f.project.id,
a.id,
2_000,
BackgroundTaskState::Running,
false,
),
background_task(
1,
f.project.id,
a.id,
1_000,
BackgroundTaskState::Failed,
false,
),
background_task(
3,
f.project.id,
a.id,
3_000,
BackgroundTaskState::Completed,
true,
),
background_task(
4,
other_project,
a.id,
4_000,
BackgroundTaskState::Running,
false,
),
]);
let out = f
.usecase
.execute(GetProjectWorkStateInput { project: f.project })
.await
.unwrap();
let alpha_tasks = &out.agents[0].background_tasks;
assert_eq!(alpha_tasks.len(), 2);
assert_eq!(alpha_tasks[0].task_id, task_id(1));
assert_eq!(alpha_tasks[0].state, BackgroundTaskState::Failed);
assert_eq!(alpha_tasks[0].exit_code, Some(2));
assert_eq!(alpha_tasks[0].summary.as_deref(), Some("failed"));
assert_eq!(alpha_tasks[0].stdout_tail.as_deref(), Some("out"));
assert_eq!(alpha_tasks[0].stderr_tail.as_deref(), Some("err"));
assert_eq!(alpha_tasks[1].task_id, task_id(2));
assert_eq!(alpha_tasks[1].state, BackgroundTaskState::Running);
assert!(out.agents[1].background_tasks.is_empty());
}
#[tokio::test]
async fn workstate_background_task_store_error_degrades_without_dropping_tickets() {
let a = agent(10, "alpha");
let f = background_fixture(std::slice::from_ref(&a));
f.queue.set(
a.id,
vec![snapshot(1, 0, InputSource::Human, "Human", "queued")],
);
f.store.set_tasks(vec![background_task(
1,
f.project.id,
a.id,
1_000,
BackgroundTaskState::Running,
false,
)]);
f.store.fail_open();
let out = f
.usecase
.execute(GetProjectWorkStateInput { project: f.project })
.await
.unwrap();
assert!(out.agents[0].background_tasks.is_empty());
assert_eq!(out.agents[0].tickets.len(), 1);
assert_eq!(out.agents[0].tickets[0].ticket_id, ticket_id(1));
}
#[tokio::test]
async fn workstate_undelivered_completion_error_degrades_background_tasks_only() {
let a = agent(10, "alpha");
let f = background_fixture(std::slice::from_ref(&a));
f.queue.set(
a.id,
vec![snapshot(1, 0, InputSource::Human, "Human", "queued")],
);
f.store.set_tasks(vec![background_task(
1,
f.project.id,
a.id,
1_000,
BackgroundTaskState::Running,
false,
)]);
f.store.fail_undelivered();
let out = f
.usecase
.execute(GetProjectWorkStateInput { project: f.project })
.await
.unwrap();
assert!(out.agents[0].background_tasks.is_empty());
assert_eq!(out.agents[0].tickets.len(), 1);
}
#[tokio::test]
async fn workstate_lists_two_tickets_in_fifo_order() {
let a = agent(10, "alpha");