diff --git a/crates/ironclaw_engine/prompts/codeact_preamble.md b/crates/ironclaw_engine/prompts/codeact_preamble.md index 1f40637b..7fcc49f5 100644 --- a/crates/ironclaw_engine/prompts/codeact_preamble.md +++ b/crates/ironclaw_engine/prompts/codeact_preamble.md @@ -17,6 +17,10 @@ You can write multiple code blocks across turns. Variables persist between block - `llm_query_batched(prompts, context=None)` — Same but for multiple prompts in parallel. Returns a list of strings. - `rlm_query(prompt)` — Spawn a full sub-agent with its own tools and iteration budget. Use for complex sub-tasks that need tool access. Returns the sub-agent's final answer as a string. More powerful but more expensive than llm_query. - `FINAL(answer)` — Call this when you have the final answer. The argument is returned to the user. +- `mission_create(name, goal, cadence="manual", success_criteria=None)` — Create a long-running mission that spawns threads over time. Cadence: "manual", cron expression (e.g. "0 9 * * *"), "event:pattern", or "webhook:path". Returns {"mission_id": "...", "status": "created"}. +- `mission_list()` — List all missions with their status, goal, and current focus. +- `mission_fire(id)` — Manually trigger a mission to spawn a thread now. +- `mission_pause(id)` / `mission_resume(id)` — Pause or resume a mission. ## Context variables diff --git a/crates/ironclaw_engine/src/runtime/mission.rs b/crates/ironclaw_engine/src/runtime/mission.rs index 992a4e05..bcb58a6b 100644 --- a/crates/ironclaw_engine/src/runtime/mission.rs +++ b/crates/ironclaw_engine/src/runtime/mission.rs @@ -326,34 +326,35 @@ async fn process_mission_outcome( }; match outcome { - ThreadOutcome::Completed { response } => { - if let Some(text) = response { - // Try to extract next focus and goal status from the response - let lower = text.to_lowercase(); + ThreadOutcome::Completed { + response: Some(text), + } => { + // Try to extract next focus and goal status from the response + let lower = text.to_lowercase(); - // Check if goal achieved - if lower.contains("goal has been achieved: yes") - || lower.contains("goal achieved: yes") - || lower.contains("mission complete") - { - debug!(mission_id = %mission_id, "goal achieved — completing mission"); - mission.status = MissionStatus::Completed; - } - - // Extract next focus (look for "next focus:" pattern) - if let Some(focus_start) = lower.find("next focus:") { - let after = &text[focus_start + "next focus:".len()..]; - let next_focus: String = after.lines().next().unwrap_or("").trim().to_string(); - if !next_focus.is_empty() { - mission.current_focus = Some(next_focus); - } - } - - // Record approach - let accomplishment: String = text.chars().take(200).collect(); - mission.approach_history.push(accomplishment); + // Check if goal achieved + if lower.contains("goal has been achieved: yes") + || lower.contains("goal achieved: yes") + || lower.contains("mission complete") + { + debug!(mission_id = %mission_id, "goal achieved — completing mission"); + mission.status = MissionStatus::Completed; } + + // Extract next focus (look for "next focus:" pattern) + if let Some(focus_start) = lower.find("next focus:") { + let after = &text[focus_start + "next focus:".len()..]; + let next_focus: String = after.lines().next().unwrap_or("").trim().to_string(); + if !next_focus.is_empty() { + mission.current_focus = Some(next_focus); + } + } + + // Record approach + let accomplishment: String = text.chars().take(200).collect(); + mission.approach_history.push(accomplishment); } + ThreadOutcome::Completed { response: None } => {} ThreadOutcome::Failed { error } => { mission.approach_history.push(format!("FAILED: {error}")); } diff --git a/src/bridge/effect_adapter.rs b/src/bridge/effect_adapter.rs index 1855559a..e7b318e9 100644 --- a/src/bridge/effect_adapter.rs +++ b/src/bridge/effect_adapter.rs @@ -39,6 +39,8 @@ pub struct EffectBridgeAdapter { call_count: std::sync::atomic::AtomicU32, /// Per-user per-tool sliding window rate limiter. rate_limiter: RateLimiter, + /// Mission manager for handling mission_* function calls. + mission_manager: RwLock>>, } impl EffectBridgeAdapter { @@ -54,6 +56,7 @@ impl EffectBridgeAdapter { auto_approved: RwLock::new(HashSet::new()), call_count: std::sync::atomic::AtomicU32::new(0), rate_limiter: RateLimiter::new(), + mission_manager: RwLock::new(None), } } @@ -65,6 +68,138 @@ impl EffectBridgeAdapter { .insert(tool_name.to_string()); } + /// Set the mission manager (called after engine init). + pub async fn set_mission_manager(&self, mgr: Arc) { + *self.mission_manager.write().await = Some(mgr); + } + + /// Handle mission_* function calls. Returns None if not a mission call. + async fn handle_mission_call( + &self, + action_name: &str, + params: &serde_json::Value, + context: &ThreadExecutionContext, + ) -> Option> { + let mgr = self.mission_manager.read().await; + let mgr = mgr.as_ref()?; + + let result = match action_name { + "mission_create" => { + let name = params + .get("name") + .or_else(|| params.get("_args").and_then(|a| a.get(0))) + .and_then(|v| v.as_str()) + .unwrap_or("unnamed mission"); + let goal = params + .get("goal") + .or_else(|| params.get("_args").and_then(|a| a.get(1))) + .and_then(|v| v.as_str()) + .unwrap_or(""); + let cadence_str = params + .get("cadence") + .or_else(|| params.get("_args").and_then(|a| a.get(2))) + .and_then(|v| v.as_str()) + .unwrap_or("manual"); + match mgr + .create_mission(context.project_id, name, goal, parse_cadence(cadence_str)) + .await + { + Ok(id) => { + Ok(serde_json::json!({"mission_id": id.to_string(), "status": "created"})) + } + Err(e) => Err(e), + } + } + "mission_list" => match mgr.list_missions(context.project_id).await { + Ok(missions) => { + let list: Vec = missions + .iter() + .map(|m| { + serde_json::json!({ + "id": m.id.to_string(), + "name": m.name, + "goal": m.goal, + "status": format!("{:?}", m.status), + "threads": m.thread_history.len(), + "current_focus": m.current_focus, + }) + }) + .collect(); + Ok(serde_json::json!(list)) + } + Err(e) => Err(e), + }, + "mission_fire" => { + let id_str = params + .get("id") + .or_else(|| params.get("_args").and_then(|a| a.get(0))) + .and_then(|v| v.as_str()) + .unwrap_or(""); + let id = uuid::Uuid::parse_str(id_str) + .map(ironclaw_engine::MissionId) + .map_err(|e| EngineError::Effect { + reason: format!("invalid mission id: {e}"), + }); + match id { + Ok(id) => match mgr.fire_mission(id, &context.user_id, None).await { + Ok(Some(tid)) => { + Ok(serde_json::json!({"thread_id": tid.to_string(), "status": "fired"})) + } + Ok(None) => Ok( + serde_json::json!({"status": "not_fired", "reason": "mission is terminal or budget exhausted"}), + ), + Err(e) => Err(e), + }, + Err(e) => Err(e), + } + } + "mission_pause" | "mission_resume" => { + let id_str = params + .get("id") + .or_else(|| params.get("_args").and_then(|a| a.get(0))) + .and_then(|v| v.as_str()) + .unwrap_or(""); + let id = uuid::Uuid::parse_str(id_str) + .map(ironclaw_engine::MissionId) + .map_err(|e| EngineError::Effect { + reason: format!("invalid mission id: {e}"), + }); + match id { + Ok(id) => { + let res = if action_name == "mission_pause" { + mgr.pause_mission(id).await + } else { + mgr.resume_mission(id).await + }; + match res { + Ok(()) => Ok(serde_json::json!({"status": "ok"})), + Err(e) => Err(e), + } + } + Err(e) => Err(e), + } + } + _ => return None, // Not a mission call + }; + + Some(match result { + Ok(output) => Ok(ActionResult { + call_id: String::new(), + action_name: action_name.to_string(), + output, + is_error: false, + duration: std::time::Duration::ZERO, + }), + Err(e) => Ok(ActionResult { + call_id: String::new(), + action_name: action_name.to_string(), + output: serde_json::json!({"error": e.to_string()}), + is_error: true, + duration: std::time::Duration::ZERO, + }), + }) + } + /// Reset the per-step call counter (called between code steps). #[allow(dead_code)] pub fn reset_call_count(&self) { @@ -106,7 +241,18 @@ impl EffectExecutor for EffectBridgeAdapter { }); } - // ── 0. Block tools that need v1 runtime deps (RoutineEngine, Scheduler) ── + // ── 0a. Handle mission_* functions via MissionManager ── + if let Some(result) = self + .handle_mission_call(action_name, ¶meters, context) + .await + { + return result.map(|mut r| { + r.duration = start.elapsed(); + r + }); + } + + // ── 0b. Block tools that need v1 runtime deps (RoutineEngine, Scheduler) ── if is_v1_only_tool(lookup_name) { return Err(EngineError::Effect { reason: format!( @@ -296,6 +442,41 @@ impl EffectExecutor for EffectBridgeAdapter { } } +/// Parse a cadence string into a MissionCadence. +fn parse_cadence(s: &str) -> ironclaw_engine::types::mission::MissionCadence { + use ironclaw_engine::types::mission::MissionCadence; + let trimmed = s.trim().to_lowercase(); + if trimmed == "manual" { + MissionCadence::Manual + } else if trimmed.contains(' ') && trimmed.split_whitespace().count() >= 5 { + // Looks like a cron expression + MissionCadence::Cron { + expression: s.trim().to_string(), + timezone: None, + } + } else if trimmed.starts_with("event:") { + MissionCadence::OnEvent { + event_pattern: trimmed + .strip_prefix("event:") + .unwrap_or("") + .trim() + .to_string(), + } + } else if trimmed.starts_with("webhook:") { + MissionCadence::Webhook { + path: trimmed + .strip_prefix("webhook:") + .unwrap_or("") + .trim() + .to_string(), + secret: None, + } + } else { + // Default to manual if unrecognized + MissionCadence::Manual + } +} + /// Tools that depend on v1 runtime components (RoutineEngine, Scheduler, /// ContainerJobManager) and cannot work in engine v2's minimal JobContext. fn is_v1_only_tool(name: &str) -> bool { diff --git a/src/bridge/router.rs b/src/bridge/router.rs index 6187051c..2803df1d 100644 --- a/src/bridge/router.rs +++ b/src/bridge/router.rs @@ -6,8 +6,8 @@ use tokio::sync::RwLock; use tracing::debug; use ironclaw_engine::{ - Capability, CapabilityRegistry, ConversationManager, LeaseManager, PolicyEngine, Project, - Store, ThreadConfig, ThreadManager, ThreadOutcome, + Capability, CapabilityRegistry, ConversationManager, LeaseManager, MissionManager, + PolicyEngine, Project, Store, ThreadConfig, ThreadManager, ThreadOutcome, }; use ironclaw_common::AppEvent; @@ -49,6 +49,8 @@ struct EngineState { sse: Option>, /// V1 database for writing conversation messages (gateway reads from here). db: Option>, + /// Mission manager for long-running goals. + mission_manager: Arc, } /// Global engine state, initialized on first use. @@ -133,6 +135,18 @@ async fn get_or_init_engine(agent: &Agent) -> Result<(), Error> { let conversation_manager = ConversationManager::new(Arc::clone(&thread_manager)); + // Create mission manager and start cron ticker + let mission_manager = Arc::new(MissionManager::new( + store.clone() as Arc, + Arc::clone(&thread_manager), + )); + mission_manager.start_cron_ticker(agent.deps.owner_id.clone()); + + // Wire mission manager into effect adapter for mission_* function calls + effect_adapter + .set_mission_manager(Arc::clone(&mission_manager)) + .await; + *guard = Some(EngineState { thread_manager, conversation_manager, @@ -142,6 +156,7 @@ async fn get_or_init_engine(agent: &Agent) -> Result<(), Error> { pending_approval: RwLock::new(None), sse: agent.deps.sse_tx.clone(), db: agent.deps.store.clone(), + mission_manager, }); Ok(())