feat(background): déclencheur in-app des tâches de fond via MCP idea_run_in_background (BE-1+BE-2)
Ferme la boucle B8 côté surface agent : un agent peut lancer une commande en tâche de fond depuis l'app, sans dépendance à un déclencheur externe. - domain : variante OrchestratorCommand::RunInBackground. - infrastructure : outil MCP idea_run_in_background (déclaration + schéma + map_tool_call), tests de mapping et compteur de catalogue. - application : handler → SpawnBackgroundCommand (owner=requester, WakeOwner). - app-tauri : câblage .with_spawn_background_command + catalogue MCP 23→24. Tests verts : domain 452 / application 467 / infrastructure 489 / app-tauri 236, build --release vert. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -14,6 +14,7 @@
|
||||
//! infrastructure adapter's job. That keeps this fully unit-testable with fakes.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use std::sync::{Arc, Mutex as StdMutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
@ -39,6 +40,7 @@ use crate::agent::{
|
||||
ListAgents, ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext,
|
||||
UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS,
|
||||
};
|
||||
use crate::background::{SpawnBackgroundCommand, SpawnBackgroundCommandInput};
|
||||
use crate::error::AppError;
|
||||
use crate::orchestrator::rendezvous::{run_inactivity_watchdog, WatchdogOutcome};
|
||||
use crate::orchestrator::{
|
||||
@ -89,6 +91,55 @@ fn bound_task_text(value: &str, max_bytes: usize) -> String {
|
||||
value[..end].to_owned()
|
||||
}
|
||||
|
||||
fn resolve_background_cwd(
|
||||
project_root: &ProjectPath,
|
||||
cwd: Option<String>,
|
||||
) -> Result<ProjectPath, AppError> {
|
||||
let root = Path::new(project_root.as_str());
|
||||
let raw = cwd.as_deref().map(str::trim).filter(|s| !s.is_empty());
|
||||
let candidate = match raw {
|
||||
None => root.to_path_buf(),
|
||||
Some(value) => {
|
||||
let path = Path::new(value);
|
||||
if path.is_absolute() {
|
||||
path.to_path_buf()
|
||||
} else {
|
||||
root.join(path)
|
||||
}
|
||||
}
|
||||
};
|
||||
let normalized = normalize_path_no_parent(&candidate)?;
|
||||
if !normalized.starts_with(root) {
|
||||
return Err(AppError::Invalid(
|
||||
"background task cwd must stay under the project root".to_owned(),
|
||||
));
|
||||
}
|
||||
let Some(path) = normalized.to_str() else {
|
||||
return Err(AppError::Invalid(
|
||||
"background task cwd is not valid UTF-8".to_owned(),
|
||||
));
|
||||
};
|
||||
ProjectPath::new(path).map_err(|_| AppError::Invalid("invalid background task cwd".to_owned()))
|
||||
}
|
||||
|
||||
fn normalize_path_no_parent(path: &Path) -> Result<PathBuf, AppError> {
|
||||
let mut out = PathBuf::new();
|
||||
for component in path.components() {
|
||||
match component {
|
||||
Component::Prefix(prefix) => out.push(prefix.as_os_str()),
|
||||
Component::RootDir => out.push(component.as_os_str()),
|
||||
Component::CurDir => {}
|
||||
Component::Normal(part) => out.push(part),
|
||||
Component::ParentDir => {
|
||||
return Err(AppError::Invalid(
|
||||
"background task cwd must not contain `..`".to_owned(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Bound on the synchronous inter-agent rendezvous (`agent.message` → `AskAgent`).
|
||||
///
|
||||
/// A target agent's turn can be long (reasoning + tool use), so the cap is
|
||||
@ -433,6 +484,9 @@ pub struct OrchestratorService {
|
||||
/// `idea_ask_agent` comme [`BackgroundTaskKind::HeadlessRendezvous`]. `None` ⇒
|
||||
/// comportement historique sans persistance de tâche (call sites non câblés).
|
||||
background_tasks: Option<Arc<dyn BackgroundTaskStore>>,
|
||||
/// Use case B8 qui produit une tâche de fond command-backed depuis la surface
|
||||
/// agent MCP `idea_run_in_background`. `None` ⇒ outil non configuré.
|
||||
spawn_background_command: Option<Arc<SpawnBackgroundCommand>>,
|
||||
}
|
||||
|
||||
/// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés
|
||||
@ -504,9 +558,17 @@ impl OrchestratorService {
|
||||
ask_liveness_probe: None,
|
||||
ask_ceiling: ASK_AGENT_CEILING,
|
||||
background_tasks: None,
|
||||
spawn_background_command: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Branche le producteur B8 command-backed pour `idea_run_in_background`.
|
||||
#[must_use]
|
||||
pub fn with_spawn_background_command(mut self, spawn: Arc<SpawnBackgroundCommand>) -> Self {
|
||||
self.spawn_background_command = Some(spawn);
|
||||
self
|
||||
}
|
||||
|
||||
/// Branche le store des tâches de fond pour tracer les rendez-vous inter-agents
|
||||
/// comme [`BackgroundTaskKind::HeadlessRendezvous`] (B6).
|
||||
///
|
||||
@ -1096,9 +1158,70 @@ impl OrchestratorService {
|
||||
)
|
||||
.await
|
||||
}
|
||||
OrchestratorCommand::RunInBackground {
|
||||
owner,
|
||||
label,
|
||||
command,
|
||||
args,
|
||||
cwd,
|
||||
deadline_ms,
|
||||
} => {
|
||||
self.run_in_background(project, owner, label, command, args, cwd, deadline_ms)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_in_background(
|
||||
&self,
|
||||
project: &Project,
|
||||
owner: AgentId,
|
||||
label: String,
|
||||
command: String,
|
||||
args: Vec<String>,
|
||||
cwd: Option<String>,
|
||||
deadline_ms: Option<u64>,
|
||||
) -> Result<OrchestratorOutcome, AppError> {
|
||||
let spawn = self.spawn_background_command.as_ref().ok_or_else(|| {
|
||||
AppError::Invalid("background command runner is not configured".to_owned())
|
||||
})?;
|
||||
let cwd = resolve_background_cwd(&project.root, cwd)?;
|
||||
let output = spawn
|
||||
.execute(SpawnBackgroundCommandInput {
|
||||
project_id: project.id,
|
||||
owner_agent_id: owner,
|
||||
label,
|
||||
command: domain::ports::SpawnSpec {
|
||||
command,
|
||||
args,
|
||||
cwd,
|
||||
env: Vec::new(),
|
||||
context_plan: None,
|
||||
sandbox: None,
|
||||
},
|
||||
wake_policy: BackgroundTaskWakePolicy::WakeOwner,
|
||||
deadline_ms,
|
||||
})
|
||||
.await?;
|
||||
let state = match output.task.state {
|
||||
BackgroundTaskState::Queued => "Queued",
|
||||
BackgroundTaskState::Running => "Running",
|
||||
BackgroundTaskState::Waiting => "Waiting",
|
||||
BackgroundTaskState::Completed => "Completed",
|
||||
BackgroundTaskState::Failed => "Failed",
|
||||
BackgroundTaskState::Cancelled => "Cancelled",
|
||||
BackgroundTaskState::Expired => "Expired",
|
||||
};
|
||||
Ok(OrchestratorOutcome {
|
||||
detail: serde_json::json!({
|
||||
"taskId": output.task.id.to_string(),
|
||||
"state": state,
|
||||
})
|
||||
.to_string(),
|
||||
reply: None,
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns the injected C7 use cases, or a typed error when unwired.
|
||||
fn require_context_guard(&self) -> Result<&ContextGuardUseCases, AppError> {
|
||||
self.context_guard.as_deref().ok_or_else(|| {
|
||||
|
||||
@ -21,10 +21,11 @@ use domain::ids::SkillId;
|
||||
use domain::ids::{AgentId, NodeId, ProfileId, ProjectId};
|
||||
use domain::markdown::MarkdownDoc;
|
||||
use domain::ports::{
|
||||
AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream,
|
||||
ExitStatus, FileSystem, FsError, IdGenerator, MemoryError, MemoryStore, OutputStream,
|
||||
PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError,
|
||||
SessionPlan, SkillStore, SpawnSpec, StoreError,
|
||||
AgentContextStore, AgentRuntime, BackgroundCompletionStream, BackgroundTaskHandle,
|
||||
BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore,
|
||||
ContextInjectionPlan, DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError,
|
||||
IdGenerator, MemoryError, MemoryStore, OutputStream, PreparedContext, ProfileStore, PtyError,
|
||||
PtyHandle, PtyPort, RemotePath, RuntimeError, SessionPlan, SkillStore, SpawnSpec, StoreError,
|
||||
};
|
||||
use domain::profile::{
|
||||
AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport,
|
||||
@ -34,14 +35,17 @@ use domain::project::{Project, ProjectPath};
|
||||
use domain::remote::RemoteRef;
|
||||
use domain::skill::{Skill, SkillScope};
|
||||
use domain::terminal::{SessionKind, TerminalSession};
|
||||
use domain::{
|
||||
BackgroundTask, BackgroundTaskState, BackgroundTaskWakePolicy, OrchestratorCommand,
|
||||
OrchestratorRequest, PtySize, SessionId, TaskId,
|
||||
};
|
||||
use domain::{Memory, MemoryIndexEntry, MemoryLink, MemorySlug};
|
||||
use domain::{OrchestratorCommand, OrchestratorRequest, PtySize, SessionId};
|
||||
use uuid::Uuid;
|
||||
|
||||
use application::{
|
||||
CloseTerminal, CreateAgentFromScratch, CreateSkill, GetLiveStateLean, HarvestMemoryFromTurn,
|
||||
LaunchAgent, ListAgents, LiveStateProvider, LiveStateReadProvider, OrchestratorService,
|
||||
TerminalSessions, UpdateAgentContext, UpdateLiveState,
|
||||
SpawnBackgroundCommand, TerminalSessions, UpdateAgentContext, UpdateLiveState,
|
||||
};
|
||||
use domain::live_state::{LiveEntry, LiveState, WorkStatus};
|
||||
use domain::ports::LiveStateStore;
|
||||
@ -370,6 +374,104 @@ impl PtyPort for FakePty {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct RecordingBackgroundStore {
|
||||
tasks: Arc<Mutex<HashMap<TaskId, BackgroundTask>>>,
|
||||
}
|
||||
impl RecordingBackgroundStore {
|
||||
fn task(&self, id: TaskId) -> Option<BackgroundTask> {
|
||||
self.tasks.lock().unwrap().get(&id).cloned()
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl BackgroundTaskStore for RecordingBackgroundStore {
|
||||
async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
|
||||
let mut tasks = self.tasks.lock().unwrap();
|
||||
if tasks.contains_key(&task.id) {
|
||||
return Err(BackgroundTaskPortError::AlreadyExists);
|
||||
}
|
||||
tasks.insert(task.id, task.clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get(&self, id: TaskId) -> Result<Option<BackgroundTask>, BackgroundTaskPortError> {
|
||||
Ok(self.task(id))
|
||||
}
|
||||
|
||||
async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
|
||||
self.tasks.lock().unwrap().insert(task.id, 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(self
|
||||
.tasks
|
||||
.lock()
|
||||
.unwrap()
|
||||
.values()
|
||||
.filter(|task| task.has_pending_completion_delivery())
|
||||
.cloned()
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn mark_completion_delivered(
|
||||
&self,
|
||||
task_id: TaskId,
|
||||
) -> Result<(), BackgroundTaskPortError> {
|
||||
let task = self
|
||||
.task(task_id)
|
||||
.ok_or(BackgroundTaskPortError::NotFound)?
|
||||
.mark_completion_delivered()
|
||||
.map_err(|err| BackgroundTaskPortError::Invalid(err.to_string()))?;
|
||||
self.save(&task).await
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct RecordingBackgroundRunner {
|
||||
specs: Arc<Mutex<Vec<BackgroundTaskSpec>>>,
|
||||
}
|
||||
impl RecordingBackgroundRunner {
|
||||
fn specs(&self) -> Vec<BackgroundTaskSpec> {
|
||||
self.specs.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl BackgroundTaskRunner for RecordingBackgroundRunner {
|
||||
async fn spawn(
|
||||
&self,
|
||||
spec: BackgroundTaskSpec,
|
||||
) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
|
||||
let task_id = spec.task_id;
|
||||
self.specs.lock().unwrap().push(spec);
|
||||
Ok(BackgroundTaskHandle { task_id })
|
||||
}
|
||||
|
||||
async fn cancel(&self, _task_id: TaskId) -> Result<(), BackgroundTaskPortError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn subscribe_completions(&self) -> BackgroundCompletionStream {
|
||||
Box::new(std::iter::empty())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
struct SpyBus(Arc<Mutex<Vec<DomainEvent>>>);
|
||||
impl SpyBus {
|
||||
@ -683,6 +785,71 @@ async fn spawn_unknown_agent_creates_then_launches() {
|
||||
assert!(launched, "AgentLaunched must be published");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn run_in_background_dispatch_creates_running_command_for_requester() {
|
||||
let fx = fixture(FakeContexts::new());
|
||||
let store = Arc::new(RecordingBackgroundStore::default());
|
||||
let runner = Arc::new(RecordingBackgroundRunner::default());
|
||||
let spawn = Arc::new(SpawnBackgroundCommand::new(
|
||||
Arc::clone(&store) as Arc<dyn BackgroundTaskStore>,
|
||||
Arc::clone(&runner) as Arc<dyn BackgroundTaskRunner>,
|
||||
Arc::new(FixedMillisClock(42)) as Arc<dyn domain::ports::Clock>,
|
||||
Arc::new(SeqIds::new()) as Arc<dyn IdGenerator>,
|
||||
));
|
||||
let service = fx.service.with_spawn_background_command(spawn);
|
||||
let owner = aid(77);
|
||||
|
||||
let out = service
|
||||
.dispatch(
|
||||
&project(),
|
||||
OrchestratorCommand::RunInBackground {
|
||||
owner,
|
||||
label: "B8 T1".to_owned(),
|
||||
command: "bash".to_owned(),
|
||||
args: vec![
|
||||
"-lc".to_owned(),
|
||||
"sleep 45 && echo DONE_MARKER_DEVBACKEND_B8_T1".to_owned(),
|
||||
],
|
||||
cwd: Some("crates".to_owned()),
|
||||
deadline_ms: None,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("dispatch ok");
|
||||
|
||||
let payload: serde_json::Value = serde_json::from_str(&out.detail).unwrap();
|
||||
assert_eq!(payload["state"], "Running");
|
||||
let task_id =
|
||||
TaskId::from_uuid(uuid::Uuid::parse_str(payload["taskId"].as_str().unwrap()).unwrap());
|
||||
let task = store.task(task_id).expect("task persisted");
|
||||
assert_eq!(task.project_id, project().id);
|
||||
assert_eq!(task.owner_agent_id, owner);
|
||||
assert_eq!(task.state, BackgroundTaskState::Running);
|
||||
assert_eq!(task.wake_policy, BackgroundTaskWakePolicy::WakeOwner);
|
||||
assert_eq!(
|
||||
task.kind,
|
||||
domain::BackgroundTaskKind::Command {
|
||||
label: "B8 T1".to_owned(),
|
||||
}
|
||||
);
|
||||
|
||||
let specs = runner.specs();
|
||||
assert_eq!(specs.len(), 1);
|
||||
assert_eq!(specs[0].task_id, task_id);
|
||||
assert_eq!(specs[0].owner_agent_id, owner);
|
||||
assert_eq!(specs[0].wake_policy, BackgroundTaskWakePolicy::WakeOwner);
|
||||
let command = specs[0].command.as_ref().expect("command spec");
|
||||
assert_eq!(command.command, "bash");
|
||||
assert_eq!(
|
||||
command.args,
|
||||
vec![
|
||||
"-lc".to_owned(),
|
||||
"sleep 45 && echo DONE_MARKER_DEVBACKEND_B8_T1".to_owned(),
|
||||
]
|
||||
);
|
||||
assert_eq!(command.cwd.as_str(), "/home/me/proj/crates");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn spawn_known_agent_launches_without_recreating() {
|
||||
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
||||
|
||||
Reference in New Issue
Block a user