From e05edc686365f0ff3353abf238dfbabcf5e36549 Mon Sep 17 00:00:00 2001 From: Blomios Date: Thu, 2 Jul 2026 21:33:45 +0200 Subject: [PATCH] =?UTF-8?q?feat(app-tauri):=20c=C3=A2blage=20runtime=20des?= =?UTF-8?q?=20t=C3=A2ches=20de=20fond=20(B7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Branche le store de tâches de fond dans le runtime : store per-root routé par project_id, AppReconcileBackgroundTasks au boot, AppWakeSessionProvider et boucle ready→MediatedInbox→AgentWakeService, avec .with_background_tasks(store, clock) au builder OrchestratorService. open_project appelle reconcile_background_tasks au démarrage. Aligne aussi 4 tests MCP périmés de l'ancien protocole : idea_reply retiré (erreur JSON-RPC -32601), idea_ask_agent exposé (13 tools), rendez-vous capture le Final inline. cargo test -p app-tauri --lib = 52 passed. Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/commands.rs | 4 + crates/app-tauri/src/state.rs | 860 +++++++++++++++++++++++++------ 2 files changed, 716 insertions(+), 148 deletions(-) diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 3c36ce6..ca8c828 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -134,6 +134,10 @@ pub async fn open_project( .reconcile_live_state .execute(ReconcileLiveStateInput { project_id: id }) .await; + // Réconcilie les tâches de fond persistées au même point de boot que le + // live-state : les Running sans handle vivant deviennent terminales, les + // complétions WakeOwner non livrées repassent dans l'inbox unifiée. + let _ = state.reconcile_background_tasks.execute(id).await; // (Re)start the orchestrator watcher for this project (idempotent, §14.3). state.ensure_orchestrator_watch(&output.project); state.reconcile_claude_run_dirs(&output.project).await; diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 46b18ba..e9004aa 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -12,11 +12,11 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex}; use application::{ - AgentResumer, AppError, AssignSkillToAgent, AttachLiveAgent, ChangeAgentProfile, - CheckEmbedderSuggestion, CloseProject, CloseTab, CloseTerminal, ConfigureProfiles, - ContextGuardUseCases, CreateAgentFromScratch, CreateAgentFromTemplate, CreateLayout, - CreateMemory, CreateProject, CreateSkill, CreateTemplate, DeleteAgent, DeleteEmbedderProfile, - DeleteLayout, DeleteMemory, DeleteProfile, DeleteSkill, DeleteTemplate, + AgentResumer, AgentWakeService, AppError, AssignSkillToAgent, AttachLiveAgent, + ChangeAgentProfile, CheckEmbedderSuggestion, CloseProject, CloseTab, CloseTerminal, + ConfigureProfiles, ContextGuardUseCases, CreateAgentFromScratch, CreateAgentFromTemplate, + CreateLayout, CreateMemory, CreateProject, CreateSkill, CreateTemplate, DeleteAgent, + DeleteEmbedderProfile, DeleteLayout, DeleteMemory, DeleteProfile, DeleteSkill, DeleteTemplate, DescribeEmbedderEngines, DetectAgentDrift, DetectProfiles, DismissEmbedderSuggestion, FirstRunState, GetLiveStateLean, GetMemory, GetProjectPermissions, GetProjectWorkState, GitBranches, GitCheckout, GitCommit, GitGraph, GitInit, GitLog, GitStage, GitStatus, @@ -33,29 +33,36 @@ use application::{ SessionLimitService, SetActiveLayout, SnapshotRunningAgents, StopLiveAgent, StructuredSessions, SuggestedThisSession, SyncAgentWithTemplate, TerminalSessions, UnassignSkillFromAgent, UpdateAgentContext, UpdateAgentPermissions, UpdateLiveState, UpdateMemory, - UpdateProjectContext, UpdateProjectPermissions, UpdateSkill, UpdateTemplate, WriteMemory, - WriteToTerminal, AGENT_MEMORY_RECALL_BUDGET, + UpdateProjectContext, UpdateProjectPermissions, UpdateSkill, UpdateTemplate, + WakeSessionProvider, WriteMemory, WriteToTerminal, AGENT_MEMORY_RECALL_BUDGET, }; +use async_trait::async_trait; use domain::ports::{ - AgentContextStore, AgentRuntime, AgentSessionFactory, Clock, Embedder, EmbedderEnvInspector, + AgentContextStore, AgentRuntime, AgentSession, AgentSessionFactory, AgentWakePort, + BackgroundTaskPortError, BackgroundTaskStore, Clock, Embedder, EmbedderEnvInspector, EmbedderProfileStore, EmbedderPromptStore, EventBus, FileSystem, GitPort, IdGenerator, MemoryRecall, MemoryStore, PermissionStore, ProcessSpawner, ProfileStore, ProjectStore, - PtyPort, ScheduledTask, Scheduler, SkillStore, TemplateStore, + PtyPort, ScheduledTask, Scheduler, SkillStore, TemplateStore, WakeError, WakeReason, }; use domain::profile::{ AgentProfile, ContextInjection, McpConfigStrategy, McpTransport, StructuredAdapter, }; use domain::remote::RemoteKind; -use domain::{AgentId, DomainEvent, EmbedderProfile, Project, ProjectId}; +use domain::{ + AgentId, AgentInbox, BackgroundTask, BackgroundTaskWakePolicy, DomainEvent, EmbedderProfile, + InboxError, InboxItem, InboxItemKind, InboxReceiptStatus, InboxSource, Project, ProjectId, + TaskId, TicketId, +}; use serde_json::{json, Map, Value}; use uuid::Uuid; use infrastructure::{ - embedder_from_profile, AdaptiveMemoryRecall, ClaudePermissionProjector, - ClaudeTranscriptInspector, CliAgentRuntime, CodexPermissionProjector, EmbedderEnvProbe, - FsConversationLog, FsEmbedderProfileStore, FsEmbedderPromptStore, FsHandoffStore, - FsLiveStateStore, FsMemoryStore, FsOrchestratorWatcher, FsPermissionStore, FsProfileStore, - FsProjectStore, FsProviderSessionStore, FsSkillStore, FsTemplateStore, Git2Repository, + embedder_from_profile, AdaptiveMemoryRecall, BackgroundTaskReadyToDeliver, + ClaudePermissionProjector, ClaudeTranscriptInspector, CliAgentRuntime, + CodexPermissionProjector, EmbedderEnvProbe, FsBackgroundTaskStore, FsConversationLog, + FsEmbedderProfileStore, FsEmbedderPromptStore, FsHandoffStore, FsLiveStateStore, FsMemoryStore, + FsOrchestratorWatcher, FsPermissionStore, FsProfileStore, FsProjectStore, + FsProviderSessionStore, FsSkillStore, FsTemplateStore, Git2Repository, HeuristicHandoffSummarizer, IdeaiContextStore, InMemoryConversationRegistry, InMemoryMailbox, LocalFileSystem, LocalProcessSpawner, McpServer, MediatedInbox, NaiveMemoryRecall, OrchestratorWatchHandle, PortablePtyAdapter, RwFileGuard, StructuredSessionFactory, @@ -245,6 +252,292 @@ impl AppReconcileLiveState { } } +/// Store global injecté à l'[`OrchestratorService`] qui route chaque opération vers +/// un [`FsBackgroundTaskStore`] lié au root du projet concerné. +/// +/// Le service orchestrateur est unique et multi-projets, alors que +/// `FsBackgroundTaskStore` fixe `/.ideai/background-tasks` à la construction. +/// Ce routeur garde donc la frontière per-root sans instancier un store global sur +/// une racine arbitraire. +struct AppBackgroundTaskStore { + projects: Arc, + stores: Mutex>>, + task_projects: Mutex>, +} + +impl AppBackgroundTaskStore { + fn new(projects: Arc) -> Self { + Self { + projects, + stores: Mutex::new(HashMap::new()), + task_projects: Mutex::new(HashMap::new()), + } + } + + async fn store_for_project( + &self, + project_id: ProjectId, + ) -> Result, BackgroundTaskPortError> { + if let Some(store) = self + .stores + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&project_id) + .cloned() + { + return Ok(store); + } + + let project = self + .projects + .load_project(project_id) + .await + .map_err(|err| BackgroundTaskPortError::Store(err.to_string()))?; + let store = Arc::new(FsBackgroundTaskStore::new(&project.root)); + self.stores + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(project_id, Arc::clone(&store)); + Ok(store) + } + + fn remember(&self, task_id: TaskId, project_id: ProjectId) { + self.task_projects + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(task_id, project_id); + } + + async fn project_for_task( + &self, + task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + if let Some(project_id) = self + .task_projects + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&task_id) + .copied() + { + return Ok(Some(project_id)); + } + + for project in self + .projects + .list_projects() + .await + .map_err(|err| BackgroundTaskPortError::Store(err.to_string()))? + { + let store = self.store_for_project(project.id).await?; + if store.get(task_id).await?.is_some() { + self.remember(task_id, project.id); + return Ok(Some(project.id)); + } + } + Ok(None) + } + + async fn list_undelivered_for_project( + &self, + project_id: ProjectId, + ) -> Result, BackgroundTaskPortError> { + let store = self.store_for_project(project_id).await?; + let tasks = store.list_undelivered_completions().await?; + for task in &tasks { + self.remember(task.id, task.project_id); + } + Ok(tasks) + } + + async fn reconcile_project_boot( + &self, + project_id: ProjectId, + live_task_ids: &[TaskId], + now_ms: u64, + ) -> Result<(), BackgroundTaskPortError> { + let store = self.store_for_project(project_id).await?; + let report = store.reconcile_boot(live_task_ids, now_ms).await?; + for task_id in report + .failed_task_ids + .into_iter() + .chain(report.delivery_pending_task_ids.into_iter()) + { + self.remember(task_id, project_id); + } + Ok(()) + } +} + +#[async_trait] +impl BackgroundTaskStore for AppBackgroundTaskStore { + async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + let store = self.store_for_project(task.project_id).await?; + store.create(task).await?; + self.remember(task.id, task.project_id); + Ok(()) + } + + async fn get(&self, id: TaskId) -> Result, BackgroundTaskPortError> { + let Some(project_id) = self.project_for_task(id).await? else { + return Ok(None); + }; + let task = self.store_for_project(project_id).await?.get(id).await?; + if let Some(task) = &task { + self.remember(task.id, task.project_id); + } + Ok(task) + } + + async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + let store = self.store_for_project(task.project_id).await?; + store.save(task).await?; + self.remember(task.id, task.project_id); + Ok(()) + } + + async fn list_open_for_agent( + &self, + agent_id: AgentId, + ) -> Result, BackgroundTaskPortError> { + let mut tasks = Vec::new(); + for project in self + .projects + .list_projects() + .await + .map_err(|err| BackgroundTaskPortError::Store(err.to_string()))? + { + let store = self.store_for_project(project.id).await?; + tasks.extend(store.list_open_for_agent(agent_id).await?); + } + for task in &tasks { + self.remember(task.id, task.project_id); + } + tasks.sort_by_key(|task| (task.created_at_ms, task.id)); + Ok(tasks) + } + + async fn list_undelivered_completions( + &self, + ) -> Result, BackgroundTaskPortError> { + let mut tasks = Vec::new(); + for project in self + .projects + .list_projects() + .await + .map_err(|err| BackgroundTaskPortError::Store(err.to_string()))? + { + tasks.extend(self.list_undelivered_for_project(project.id).await?); + } + tasks.sort_by_key(|task| (task.updated_at_ms, task.id)); + Ok(tasks) + } + + async fn mark_completion_delivered( + &self, + task_id: TaskId, + ) -> Result<(), BackgroundTaskPortError> { + let Some(project_id) = self.project_for_task(task_id).await? else { + return Err(BackgroundTaskPortError::NotFound); + }; + self.store_for_project(project_id) + .await? + .mark_completion_delivered(task_id) + .await + } +} + +pub(crate) struct AppReconcileBackgroundTasks { + projects: Arc, + store: Arc, + clock: Arc, + ready: tokio::sync::mpsc::UnboundedSender, +} + +impl AppReconcileBackgroundTasks { + pub(crate) async fn execute(&self, project_id: ProjectId) -> Result<(), AppError> { + let project = self.projects.load_project(project_id).await?; + self.store + .reconcile_project_boot(project_id, &[], self.clock.now_millis().max(0) as u64) + .await + .map_err(|err| AppError::Store(err.to_string()))?; + + let tasks = self + .store + .list_undelivered_for_project(project_id) + .await + .map_err(|err| AppError::Store(err.to_string()))?; + for task in tasks { + match task.wake_policy { + BackgroundTaskWakePolicy::WakeOwner => { + let ready = BackgroundTaskReadyToDeliver { + task_id: task.id, + project_id: task.project_id, + owner_agent_id: task.owner_agent_id, + }; + if self.ready.send(ready).is_err() { + return Err(AppError::Internal( + "background task delivery channel is closed".to_owned(), + )); + } + } + BackgroundTaskWakePolicy::RecordOnly => { + self.store + .mark_completion_delivered(task.id) + .await + .map_err(|err| AppError::Store(err.to_string()))?; + } + } + } + + application::diag!( + "[background-task] boot reconcile complete: project={} root={}", + project.id, + project.root.as_str() + ); + Ok(()) + } +} + +struct AppWakeSessionProvider { + launch_agent: Arc, + structured_sessions: Arc, +} + +#[async_trait] +impl WakeSessionProvider for AppWakeSessionProvider { + async fn session_for_wake( + &self, + project: &Project, + agent: AgentId, + ) -> Result, WakeError> { + if let Some(session) = self.structured_sessions.session_for_agent(&agent) { + return Ok(session); + } + + self.launch_agent + .execute(LaunchAgentInput { + project: project.clone(), + agent_id: agent, + rows: 24, + cols: 80, + node_id: None, + conversation_id: None, + mcp_runtime: None, + allow_structured_alongside_pty: true, + }) + .await + .map_err(|err| WakeError::Session(err.to_string()))?; + + self.structured_sessions + .session_for_agent(&agent) + .ok_or_else(|| { + WakeError::Session(format!( + "agent {agent} has no structured session after background wake launch" + )) + }) + } +} + /// Implémente [`ProviderSessionProvider`](application::ProviderSessionProvider) (lot /// P8b) en matérialisant un [`FsProviderSessionStore`] ciblant le **project root** du /// lancement en cours. @@ -421,6 +714,10 @@ pub struct AppState { /// `working`/`waiting`/`blocked` dont la session n'est plus vivante) : les /// repasse en `idle`. Acte système best-effort, jamais via la surface MCP. pub(crate) reconcile_live_state: Arc, + /// Réconcilie, à l'ouverture, les tâches de fond persistées : marque les + /// `Running` sans handle vivant comme perdues, ré-enqueue les complétions + /// `WakeOwner` non livrées et ferme les `RecordOnly`. + pub(crate) reconcile_background_tasks: Arc, /// Detect which candidate profiles' CLIs are installed (first-run). pub detect_profiles: Arc, /// List configured profiles. @@ -1240,6 +1537,103 @@ impl AppState { }); } let input_mediator = Arc::clone(&mediated_inbox) as Arc; + let background_tasks = Arc::new(AppBackgroundTaskStore::new(Arc::clone(&store_port))); + let background_tasks_port = Arc::clone(&background_tasks) as Arc; + let (background_ready_tx, mut background_ready_rx) = + tokio::sync::mpsc::unbounded_channel::(); + let background_wake = Arc::new(AgentWakeService::new( + Arc::clone(&mediated_inbox) as Arc, + Arc::clone(&input_mediator), + Arc::clone(&mailbox), + Arc::clone(&background_tasks_port), + Arc::new(AppWakeSessionProvider { + launch_agent: Arc::clone(&launch_agent), + structured_sessions: Arc::clone(&structured_sessions), + }), + Some(Arc::clone(&events_port)), + )) as Arc; + { + let inbox = Arc::clone(&mediated_inbox); + let wake = Arc::clone(&background_wake); + let projects = Arc::clone(&store_port); + let clock_for_items = Arc::clone(&clock) as Arc; + tauri::async_runtime::spawn(async move { + while let Some(ready) = background_ready_rx.recv().await { + let item = InboxItem { + id: TicketId::new_random(), + agent_id: ready.owner_agent_id, + source: InboxSource::BackgroundTask { + task_id: ready.task_id, + }, + kind: InboxItemKind::BackgroundCompletion, + body: format!("Background task {} completed.", ready.task_id), + created_at_ms: clock_for_items.now_millis().max(0) as u64, + correlation_id: Some(ready.task_id.to_string()), + }; + match inbox.enqueue_message(ready.owner_agent_id, item) { + Ok(receipt) if receipt.status == InboxReceiptStatus::Deferred => { + application::diag!( + "[background-task] completion deferred: task={} owner={} \ + queue_depth={}", + ready.task_id, + ready.owner_agent_id, + receipt.depth + ); + continue; + } + Ok(_) => {} + Err(InboxError::InboxFull { .. }) => { + application::diag!( + "[background-task] completion inbox full but durable: task={} \ + owner={}", + ready.task_id, + ready.owner_agent_id + ); + continue; + } + Err(err) => { + application::diag!( + "[background-task] completion inbox enqueue failed: task={} \ + owner={} err={err}", + ready.task_id, + ready.owner_agent_id + ); + continue; + } + } + + let project = match projects.load_project(ready.project_id).await { + Ok(project) => project, + Err(err) => { + application::diag!( + "[background-task] completion wake skipped: project={} task={} \ + owner={} err={err}", + ready.project_id, + ready.task_id, + ready.owner_agent_id + ); + continue; + } + }; + if let Err(err) = wake + .wake_agent( + &project, + ready.owner_agent_id, + WakeReason::BackgroundCompletion { + task_id: ready.task_id, + }, + ) + .await + { + application::diag!( + "[background-task] completion wake failed: task={} owner={} err={err}", + ready.task_id, + ready.owner_agent_id + ); + } + } + }); + } let live_sessions = Arc::new(LiveSessions::new( Arc::clone(&terminal_sessions), Arc::clone(&structured_sessions), @@ -1254,6 +1648,12 @@ impl AppState { registry: Arc::clone(&live_sessions), clock: Arc::clone(&clock) as Arc, }); + let reconcile_background_tasks = Arc::new(AppReconcileBackgroundTasks { + projects: Arc::clone(&store_port), + store: Arc::clone(&background_tasks), + clock: Arc::clone(&clock) as Arc, + ready: background_ready_tx, + }); // Lot C — résumés de conversation best-effort : on câble les sources par // project root (handoff = primaire, log = repli `last(_, 3)`). Lecture seule, // aucune persistance ; un échec de preview ne bloque ni live/busy ni tickets. @@ -1450,7 +1850,15 @@ impl AppState { // launcher orchestrateur ci-dessus. Une cible à `structured_adapter` est donc // démarrée/drainée via `AgentSession::send` et son `Final`, sans dépendre de // `idea_reply`; les autres outils MCP restent câblés par ailleurs. - .with_structured(Arc::clone(&structured_sessions)), + .with_structured(Arc::clone(&structured_sessions)) + // Tâches de fond de 1re classe (B6/B7) : le port injecté est global mais + // route chaque opération vers un `FsBackgroundTaskStore` lié au root du + // projet concerné. Active notamment le rendez-vous headless tracé comme + // BackgroundTask, sans figer un store unique sur un root arbitraire. + .with_background_tasks( + Arc::clone(&background_tasks_port), + Arc::clone(&clock) as Arc, + ), ); let stop_live_agent = Arc::new( @@ -1490,6 +1898,7 @@ impl AppState { snapshot_running_agents, reconcile_layouts, reconcile_live_state, + reconcile_background_tasks, detect_profiles, list_profiles, save_profile, @@ -2845,6 +3254,208 @@ pub(crate) fn build_memory_recall( } } +#[cfg(test)] +mod background_tasks_b7_tests { + use std::collections::HashMap; + use std::path::PathBuf; + use std::sync::{Arc, Mutex}; + + use async_trait::async_trait; + use domain::ports::{BackgroundTaskStore, ProjectStore, StoreError}; + use domain::{ + AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState, + BackgroundTaskWakePolicy, Project, ProjectId, ProjectPath, RemoteRef, TaskId, + }; + use tokio::sync::mpsc::unbounded_channel; + use uuid::Uuid; + + use super::{AppBackgroundTaskStore, AppReconcileBackgroundTasks}; + + struct TempDir(PathBuf); + + impl TempDir { + fn new() -> Self { + let path = std::env::temp_dir().join(format!("idea-b7-reconcile-{}", Uuid::new_v4())); + std::fs::create_dir_all(&path).unwrap(); + Self(path) + } + + fn project_path(&self) -> ProjectPath { + ProjectPath::new(self.0.to_string_lossy().into_owned()).unwrap() + } + } + + impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + #[derive(Default)] + struct FakeProjects { + projects: Mutex>, + } + + impl FakeProjects { + fn insert(&self, project: Project) { + self.projects.lock().unwrap().insert(project.id, project); + } + } + + #[async_trait] + impl ProjectStore for FakeProjects { + async fn list_projects(&self) -> Result, StoreError> { + Ok(self.projects.lock().unwrap().values().cloned().collect()) + } + + async fn load_project(&self, id: ProjectId) -> Result { + self.projects + .lock() + .unwrap() + .get(&id) + .cloned() + .ok_or(StoreError::NotFound) + } + + async fn save_project(&self, project: &Project) -> Result<(), StoreError> { + self.insert(project.clone()); + Ok(()) + } + + async fn save_workspace(&self, _workspace: &domain::Workspace) -> Result<(), StoreError> { + Ok(()) + } + + async fn load_workspace(&self) -> Result { + Ok(domain::Workspace::default()) + } + } + + struct FixedClock(i64); + + impl domain::ports::Clock for FixedClock { + fn now_millis(&self) -> i64 { + self.0 + } + } + + fn project_id(n: u128) -> ProjectId { + ProjectId::from_uuid(Uuid::from_u128(n)) + } + + fn agent_id(n: u128) -> AgentId { + AgentId::from_uuid(Uuid::from_u128(n)) + } + + fn task_id(n: u128) -> TaskId { + TaskId::from_uuid(Uuid::from_u128(n)) + } + + fn project(id: ProjectId, root: ProjectPath) -> Project { + Project::new(id, "demo", root, RemoteRef::local(), 1_000).unwrap() + } + + fn completed_task( + id: u128, + project_id: ProjectId, + owner: AgentId, + wake_policy: BackgroundTaskWakePolicy, + ) -> BackgroundTask { + BackgroundTask::new( + task_id(id), + project_id, + owner, + BackgroundTaskKind::Command { + label: format!("task-{id}"), + }, + wake_policy, + 1_000, + None, + ) + .unwrap() + .transition(BackgroundTaskState::Running, 1_010) + .unwrap() + .complete(BackgroundTaskResult::Success { + finished_at_ms: 1_020, + exit_code: Some(0), + summary: "ok".to_owned(), + stdout_tail: None, + stderr_tail: None, + }) + .unwrap() + } + + fn running_task(id: u128, project_id: ProjectId, owner: AgentId) -> BackgroundTask { + BackgroundTask::new( + task_id(id), + project_id, + owner, + BackgroundTaskKind::Command { + label: format!("task-{id}"), + }, + BackgroundTaskWakePolicy::WakeOwner, + 1_000, + None, + ) + .unwrap() + .transition(BackgroundTaskState::Running, 1_010) + .unwrap() + } + + #[tokio::test] + async fn app_reconcile_requeues_wake_owner_marks_record_only_and_repairs_orphan_running() { + let tmp = TempDir::new(); + let project_id = project_id(10); + let owner = agent_id(100); + let projects = Arc::new(FakeProjects::default()); + projects.insert(project(project_id, tmp.project_path())); + + let store = Arc::new(AppBackgroundTaskStore::new( + Arc::clone(&projects) as Arc + )); + let wake_owner = completed_task(1, project_id, owner, BackgroundTaskWakePolicy::WakeOwner); + let record_only = + completed_task(2, project_id, owner, BackgroundTaskWakePolicy::RecordOnly); + let orphan = running_task(3, project_id, owner); + store.create(&wake_owner).await.unwrap(); + store.create(&record_only).await.unwrap(); + store.create(&orphan).await.unwrap(); + + let (ready_tx, mut ready_rx) = unbounded_channel(); + let reconcile = AppReconcileBackgroundTasks { + projects: Arc::clone(&projects) as Arc, + store: Arc::clone(&store), + clock: Arc::new(FixedClock(2_000)), + ready: ready_tx, + }; + + reconcile.execute(project_id).await.unwrap(); + + let mut ready = Vec::new(); + while let Ok(item) = ready_rx.try_recv() { + ready.push(item.task_id); + } + assert_eq!(ready, vec![wake_owner.id, orphan.id]); + + let wake_owner_after = store.get(wake_owner.id).await.unwrap().unwrap(); + assert!(!wake_owner_after.completion_delivered); + + let record_only_after = store.get(record_only.id).await.unwrap().unwrap(); + assert!(record_only_after.completion_delivered); + + let orphan_after = store.get(orphan.id).await.unwrap().unwrap(); + assert_eq!(orphan_after.state, BackgroundTaskState::Failed); + assert!(orphan_after.has_pending_completion_delivery()); + assert!(matches!( + orphan_after.result, + Some(BackgroundTaskResult::Failure { + finished_at_ms: 2_000, + .. + }) + )); + } +} + #[cfg(test)] mod mcp_serve_peer_tests { //! M5c — server side of the bind transport: [`serve_peer`] + [`parse_handshake`]. @@ -3354,6 +3965,8 @@ mod mcp_serve_peer_tests { "idea_memory_write", // Skill-awareness : lecture à la demande du corps d'un skill. "idea_skill_read", + // Conversation inter-agent headless : réponse inline capturée depuis le Final. + "idea_ask_agent", // Live-state (programme live-state, lot LS4). "idea_workstate_read", "idea_workstate_set", @@ -3363,12 +3976,11 @@ mod mcp_serve_peer_tests { "missing tool {expected}; got {names:?}" ); } - assert!(!names.contains(&"idea_ask_agent")); assert!(!names.contains(&"idea_reply")); assert_eq!( tools.len(), - 12, - "exactly the twelve non-conversation idea_* tools; got {names:?}" + 13, + "exactly the thirteen current idea_* tools; got {names:?}" ); drop(client); // EOF ⇒ serve loop ends @@ -4208,17 +4820,17 @@ mod mcp_e2e_loopback_tests { use application::{ CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, - OrchestratorService, TerminalSessions, UpdateAgentContext, + OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext, }; use domain::agent::{AgentManifest, ManifestEntry}; use domain::events::{DomainEvent, OrchestrationSource}; use domain::ids::{AgentId, NodeId, ProfileId, ProjectId, SkillId}; use domain::markdown::MarkdownDoc; use domain::ports::{ - AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream, - ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext, ProfileStore, - PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError, SessionPlan, SkillStore, SpawnSpec, - StoreError, + AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, ContextInjectionPlan, + DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError, IdGenerator, + OutputStream, PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, + ReplyEvent, ReplyStream, RuntimeError, SessionPlan, SkillStore, SpawnSpec, StoreError, }; use domain::profile::{ AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport, @@ -4227,7 +4839,6 @@ mod mcp_e2e_loopback_tests { use domain::project::{Project, ProjectPath}; use domain::remote::RemoteRef; use domain::skill::{Skill, SkillScope}; - use domain::terminal::{SessionKind, TerminalSession}; use domain::{PtySize, SessionId}; use serde_json::{json, Value}; @@ -4401,6 +5012,32 @@ mod mcp_e2e_loopback_tests { } } + struct FakeSession { + id: SessionId, + reply: String, + } + + #[async_trait] + impl AgentSession for FakeSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(self.id.to_string()) + } + + async fn send(&self, _prompt: &str) -> Result { + Ok(Box::new(std::iter::once(ReplyEvent::Final { + content: self.reply.clone(), + }))) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } + } + struct FakeRuntime; impl AgentRuntime for FakeRuntime { fn detect(&self, _p: &AgentProfile) -> Result { @@ -4552,6 +5189,8 @@ mod mcp_e2e_loopback_tests { McpTransport::Stdio, ))]))); let sessions = Arc::new(TerminalSessions::new()); + let structured = Arc::new(StructuredSessions::new()); + seed_structured_sessions(&structured, &contexts); let mailbox = Arc::new(InMemoryMailbox::new()); let bus = Arc::new(NoopBus); let create = Arc::new(CreateAgentFromScratch::new( @@ -4600,7 +5239,8 @@ mod mcp_e2e_loopback_tests { input, Arc::clone(&mailbox) as Arc, ) - .with_conversations(conversations); + .with_conversations(conversations) + .with_structured(structured); (Arc::new(service), sessions, mailbox) } @@ -4634,6 +5274,8 @@ mod mcp_e2e_loopback_tests { McpTransport::Stdio, ))]))); let sessions = Arc::new(TerminalSessions::new()); + let structured = Arc::new(StructuredSessions::new()); + seed_structured_sessions(&structured, &contexts); let mailbox = Arc::new(InMemoryMailbox::new()); let bus = Arc::new(NoopBus); let create = Arc::new(CreateAgentFromScratch::new( @@ -4682,22 +5324,24 @@ mod mcp_e2e_loopback_tests { input, Arc::clone(&mailbox) as Arc, ) - .with_conversations(conversations); + .with_conversations(conversations) + .with_structured(structured); (Arc::new(service), sessions, mailbox) } - /// Pre-seeds a live PTY terminal session for `agent_id` so `ask_agent` reuses it. - fn seed_live_pty(sessions: &TerminalSessions, agent_id: AgentId, session_id: SessionId) { - sessions.insert( - PtyHandle { session_id }, - TerminalSession::starting( - session_id, - NodeId::from_uuid(Uuid::from_u128(7)), - ProjectPath::new("/home/me/proj").unwrap(), - SessionKind::Agent { agent_id }, - PtySize { rows: 24, cols: 80 }, - ), - ); + fn seed_structured_sessions(sessions: &StructuredSessions, contexts: &FakeContexts) { + let entries = contexts.0.lock().unwrap().manifest.entries.clone(); + for (index, entry) in entries.into_iter().enumerate() { + let session = Arc::new(FakeSession { + id: SessionId::from_uuid(Uuid::from_u128(10_000 + index as u128)), + reply: "the answer is 42".to_owned(), + }); + sessions.insert( + session, + entry.agent_id, + NodeId::from_uuid(Uuid::from_u128(20_000 + index as u128)), + ); + } } // --- real loopback harness --------------------------------------------- @@ -4826,23 +5470,17 @@ mod mcp_e2e_loopback_tests { } // ----------------------------------------------------------------------- - // 2. idea_ask_agent → idea_reply e2e over the real loopback (Option 1, B-3/B-4): - // the asker's `idea_ask_agent` blocks on the mailbox; the target's `idea_reply` - // (on a second connection, with its agent id as the handshake requester) - // resolves it and the asker gets the result INLINE. + // 2. `idea_ask_agent` e2e over the real loopback under the current headless + // protocol: the orchestrator drives the target's structured session directly + // and returns the captured `Final` inline. No target-side `idea_reply` tool is + // involved. // ----------------------------------------------------------------------- #[tokio::test] async fn ask_then_reply_round_trips_inline_over_real_loopback() { let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("architect"); - let (service, sessions, mailbox) = build_service(contexts); - // Target already live in the PTY registry, so the ask reuses its terminal. - seed_live_pty( - &sessions, - agent_id, - SessionId::from_uuid(Uuid::from_u128(4242)), - ); + let (service, _sessions, mailbox) = build_service(contexts); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); @@ -4867,42 +5505,7 @@ mod mcp_e2e_loopback_tests { .unwrap(); write_a.flush().await.unwrap(); - // The ask is now blocked awaiting the reply — observe the pending ticket. - tokio::time::timeout(TIMEOUT, async { - while mailbox.pending(&agent_id) == 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("ask must enqueue a ticket"); - - // Connection B: the target rendering its result. Its handshake requester is - // the target agent's id, which the server injects as the Reply `from`. - let conn_b = connect_client(&endpoint).await; - let (read_b, mut write_b) = tokio::io::split(conn_b); - let mut reader_b = BufReader::new(read_b); - write_b - .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) - .await - .unwrap(); - write_b - .write_all( - tools_call_line(8, "idea_reply", json!({ "result": "the answer is 42" })) - .as_bytes(), - ) - .await - .unwrap(); - write_b.flush().await.unwrap(); - let reply_resp = read_one_response(&mut reader_b) - .await - .expect("a reply ack line"); - assert_eq!( - reply_resp["result"]["isError"], - json!(false), - "got {reply_resp}" - ); - - // The asker now receives its inline result. + // The asker receives the structured target's Final inline. let resp = read_one_response(&mut reader_a) .await .expect("an ask response line"); @@ -4915,9 +5518,13 @@ mod mcp_e2e_loopback_tests { "the answer is 42", "ask reply must be returned inline over the real loopback; got {result}" ); + assert_eq!( + mailbox.pending(&agent_id), + 0, + "structured ask must drain its accounting ticket" + ); drop(write_a); - drop(write_b); handle.stop(); } @@ -4931,13 +5538,7 @@ mod mcp_e2e_loopback_tests { async fn ask_then_reply_round_trips_inline_over_real_loopback_codex() { let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("architect"); - let (service, sessions, mailbox) = build_service_codex(contexts); - // Target already live in the PTY registry, so the ask reuses its terminal. - seed_live_pty( - &sessions, - agent_id, - SessionId::from_uuid(Uuid::from_u128(4242)), - ); + let (service, _sessions, mailbox) = build_service_codex(contexts); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); @@ -4962,42 +5563,7 @@ mod mcp_e2e_loopback_tests { .unwrap(); write_a.flush().await.unwrap(); - // The ask is now blocked awaiting the reply — observe the pending ticket. - tokio::time::timeout(TIMEOUT, async { - while mailbox.pending(&agent_id) == 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("ask must enqueue a ticket"); - - // Connection B: the target rendering its result. Its handshake requester is - // the target agent's id, which the server injects as the Reply `from`. - let conn_b = connect_client(&endpoint).await; - let (read_b, mut write_b) = tokio::io::split(conn_b); - let mut reader_b = BufReader::new(read_b); - write_b - .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) - .await - .unwrap(); - write_b - .write_all( - tools_call_line(8, "idea_reply", json!({ "result": "the answer is 42" })) - .as_bytes(), - ) - .await - .unwrap(); - write_b.flush().await.unwrap(); - let reply_resp = read_one_response(&mut reader_b) - .await - .expect("a reply ack line"); - assert_eq!( - reply_resp["result"]["isError"], - json!(false), - "got {reply_resp}" - ); - - // The asker now receives its inline result. + // The asker receives the structured target's Final inline. let resp = read_one_response(&mut reader_a) .await .expect("an ask response line"); @@ -5010,19 +5576,24 @@ mod mcp_e2e_loopback_tests { "the answer is 42", "ask reply must be returned inline over the real loopback (Codex target); got {result}" ); + assert_eq!( + mailbox.pending(&agent_id), + 0, + "structured ask must drain its accounting ticket" + ); drop(write_a); - drop(write_b); handle.stop(); } // ----------------------------------------------------------------------- - // 3. idea_reply with no in-flight ask ⇒ typed tool error (isError:true), no - // panic, connection stays healthy for a follow-up call. + // 3. `idea_reply` is no longer part of the MCP surface. Calling it returns a + // JSON-RPC unknown-tool error and the connection stays healthy for a follow-up + // call. // ----------------------------------------------------------------------- #[tokio::test] - async fn orphan_reply_is_typed_error_over_real_loopback() { + async fn removed_reply_tool_is_unknown_over_real_loopback() { let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("dev"); let (service, _sessions, _mailbox) = build_service(contexts); @@ -5033,8 +5604,7 @@ mod mcp_e2e_loopback_tests { let (read_half, mut write_half) = tokio::io::split(conn); let mut reader = BufReader::new(read_half); - // The peer identifies as `agent_id`; an idea_reply with no matching ask in - // flight ⇒ the IdeA command fails ⇒ tool result isError:true (not transport). + // The peer identifies as `agent_id`; `idea_reply` itself is no longer mapped. write_half .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) .await @@ -5049,19 +5619,13 @@ mod mcp_e2e_loopback_tests { .await .expect("a reply response line"); assert_eq!(resp["id"], json!(3)); - assert!( - resp["error"].is_null(), - "an orphan reply must be a tool error, not a transport error: {resp}" - ); - let result = &resp["result"]; - assert_eq!(result["isError"], json!(true), "got {result}"); - assert!( - !result["content"][0]["text"] - .as_str() - .unwrap_or_default() - .is_empty(), - "error text should describe the failure; got {result}" + assert_eq!(resp["error"]["code"], json!(-32601), "got {resp}"); + assert_eq!( + resp["error"]["message"], + json!("unknown tool: idea_reply"), + "got {resp}" ); + assert!(resp.get("result").is_none(), "got {resp}"); // The connection is still healthy: a follow-up tools/list still answers. write_half