diff --git a/crates/ironclaw_engine/src/capability/policy.rs b/crates/ironclaw_engine/src/capability/policy.rs index ce6b610a..3fb8e52c 100644 --- a/crates/ironclaw_engine/src/capability/policy.rs +++ b/crates/ironclaw_engine/src/capability/policy.rs @@ -7,6 +7,7 @@ use crate::types::capability::{ ActionDef, CapabilityLease, EffectType, PolicyCondition, PolicyEffect, PolicyRule, }; +use crate::types::provenance::Provenance; /// The result of a policy evaluation. #[derive(Debug, Clone, PartialEq, Eq)] @@ -25,7 +26,7 @@ pub enum PolicyDecision { pub struct PolicyEngine { global_policies: Vec, /// Effect types that are always denied unless explicitly overridden. - denied_effects: Vec, + pub(crate) denied_effects: Vec, } impl PolicyEngine { @@ -105,6 +106,57 @@ impl PolicyEngine { decision } + + /// Evaluate with provenance-aware taint checking. + /// + /// Extends the base evaluation with provenance-based rules: + /// - `LlmGenerated` data + `Financial` effect → RequireApproval + /// - `LlmGenerated` data + `WriteExternal` effect → RequireApproval + /// - `ToolOutput` data + `Financial` effect → RequireApproval + pub fn evaluate_with_provenance( + &self, + action: &ActionDef, + lease: &CapabilityLease, + capability_policies: &[PolicyRule], + provenance: &Provenance, + ) -> PolicyDecision { + let mut decision = self.evaluate(action, lease, capability_policies); + + // Provenance-based taint rules + match provenance { + Provenance::LlmGenerated => { + if action.effects.contains(&EffectType::Financial) { + decision = merge_decision( + decision, + PolicyEffect::RequireApproval, + "LLM-generated data cannot trigger financial effects without approval", + ); + } + if action.effects.contains(&EffectType::WriteExternal) { + decision = merge_decision( + decision, + PolicyEffect::RequireApproval, + "LLM-generated data requires approval for external writes", + ); + } + } + Provenance::ToolOutput { .. } => { + if action.effects.contains(&EffectType::Financial) { + decision = merge_decision( + decision, + PolicyEffect::RequireApproval, + "tool output data requires approval for financial effects", + ); + } + } + // User and System provenance are trusted + Provenance::User | Provenance::System => {} + // Reflection and MemoryRetrieval are internal, treat as trusted + Provenance::Reflection { .. } | Provenance::MemoryRetrieval { .. } => {} + } + + decision + } } impl Default for PolicyEngine { @@ -257,6 +309,62 @@ mod tests { )); } + #[test] + fn llm_generated_financial_requires_approval() { + let engine = PolicyEngine::new(); + let action = make_action("transfer_funds", vec![EffectType::Financial], false); + let lease = make_lease(); + let decision = engine.evaluate_with_provenance( + &action, + &lease, + &[], + &Provenance::LlmGenerated, + ); + assert!(matches!(decision, PolicyDecision::RequireApproval { .. })); + } + + #[test] + fn llm_generated_write_external_requires_approval() { + let engine = PolicyEngine::new(); + let action = make_action("post_message", vec![EffectType::WriteExternal], false); + let lease = make_lease(); + let decision = engine.evaluate_with_provenance( + &action, + &lease, + &[], + &Provenance::LlmGenerated, + ); + assert!(matches!(decision, PolicyDecision::RequireApproval { .. })); + } + + #[test] + fn user_provenance_allows_financial() { + let engine = PolicyEngine::new(); + let action = make_action("transfer_funds", vec![EffectType::Financial], false); + let lease = make_lease(); + let decision = engine.evaluate_with_provenance( + &action, + &lease, + &[], + &Provenance::User, + ); + assert_eq!(decision, PolicyDecision::Allow); + } + + #[test] + fn tool_output_financial_requires_approval() { + let engine = PolicyEngine::new(); + let action = make_action("pay_invoice", vec![EffectType::Financial], false); + let lease = make_lease(); + let decision = engine.evaluate_with_provenance( + &action, + &lease, + &[], + &Provenance::ToolOutput { action_name: "scrape_invoices".into() }, + ); + assert!(matches!(decision, PolicyDecision::RequireApproval { .. })); + } + #[test] fn action_matches_pattern() { let mut engine = PolicyEngine::new(); diff --git a/crates/ironclaw_engine/src/executor/compaction.rs b/crates/ironclaw_engine/src/executor/compaction.rs index 1985bc06..aa598194 100644 --- a/crates/ironclaw_engine/src/executor/compaction.rs +++ b/crates/ironclaw_engine/src/executor/compaction.rs @@ -101,7 +101,7 @@ pub async fn compact_messages( compacted.push(sys); } compacted.push(ThreadMessage::assistant(summary_text.clone())); - compacted.push(ThreadMessage::system(format!( + compacted.push(ThreadMessage::user(format!( "Your conversation has been compacted {n} time(s). \ The summary above captures your progress. Continue working on the task.", n = compaction_count + 1, diff --git a/crates/ironclaw_engine/src/executor/context.rs b/crates/ironclaw_engine/src/executor/context.rs index b29ba0f9..52a000a7 100644 --- a/crates/ironclaw_engine/src/executor/context.rs +++ b/crates/ironclaw_engine/src/executor/context.rs @@ -33,23 +33,26 @@ pub async fn build_step_context( let mut ctx_messages = messages.to_vec(); - // Inject retrieved memory docs as context + // Inject retrieved memory docs into the existing system prompt. + // Many providers require all system messages at the beginning (or a single + // system message), so we append to the first system message rather than + // inserting a separate one. if let Some(engine) = retrieval { let docs = engine .retrieve_context(project_id, goal, MAX_CONTEXT_DOCS) .await?; if !docs.is_empty() { - let context_msg = format_docs_as_context(&docs); - // Insert after the system prompt (index 1) if one exists, - // otherwise prepend. - let insert_pos = if !ctx_messages.is_empty() + let context_section = format_docs_as_context(&docs); + if !ctx_messages.is_empty() && ctx_messages[0].role == crate::types::message::MessageRole::System { - 1 + // Append to existing system prompt + ctx_messages[0].content.push_str("\n\n"); + ctx_messages[0].content.push_str(&context_section); } else { - 0 - }; - ctx_messages.insert(insert_pos, ThreadMessage::system(context_msg)); + // No system message — prepend as one + ctx_messages.insert(0, ThreadMessage::system(context_section)); + } } } @@ -141,6 +144,10 @@ mod tests { async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) } async fn load_active_leases(&self, _: ThreadId) -> Result, EngineError> { Ok(vec![]) } async fn revoke_lease(&self, _: LeaseId, _: &str) -> Result<(), EngineError> { Ok(()) } + async fn save_mission(&self, _: &crate::types::mission::Mission) -> Result<(), EngineError> { Ok(()) } + async fn load_mission(&self, _: crate::types::mission::MissionId) -> Result, EngineError> { Ok(None) } + async fn list_missions(&self, _: ProjectId) -> Result, EngineError> { Ok(vec![]) } + async fn update_mission_status(&self, _: crate::types::mission::MissionId, _: crate::types::mission::MissionStatus) -> Result<(), EngineError> { Ok(()) } } #[tokio::test] @@ -168,14 +175,14 @@ mod tests { .await .unwrap(); - // Should have 3 messages: system prompt, injected context, user message - assert_eq!(ctx_msgs.len(), 3); + // Should have 2 messages: system prompt (with docs appended), user message + assert_eq!(ctx_msgs.len(), 2); assert_eq!(ctx_msgs[0].role, crate::types::message::MessageRole::System); - assert_eq!(ctx_msgs[1].role, crate::types::message::MessageRole::System); - assert!(ctx_msgs[1].content.contains("Prior Knowledge")); - assert!(ctx_msgs[1].content.contains("LESSON")); - assert!(ctx_msgs[1].content.contains("web-search")); - assert_eq!(ctx_msgs[2].role, crate::types::message::MessageRole::User); + assert!(ctx_msgs[0].content.contains("You are an assistant.")); + assert!(ctx_msgs[0].content.contains("Prior Knowledge")); + assert!(ctx_msgs[0].content.contains("LESSON")); + assert!(ctx_msgs[0].content.contains("web-search")); + assert_eq!(ctx_msgs[1].role, crate::types::message::MessageRole::User); } #[tokio::test] diff --git a/crates/ironclaw_engine/src/lib.rs b/crates/ironclaw_engine/src/lib.rs index 152cae72..31664bc7 100644 --- a/crates/ironclaw_engine/src/lib.rs +++ b/crates/ironclaw_engine/src/lib.rs @@ -18,6 +18,7 @@ pub mod capability; pub mod executor; pub mod memory; pub mod reflection; +pub mod reliability; pub mod runtime; pub mod traits; pub mod types; @@ -37,6 +38,7 @@ pub use types::provenance::Provenance; pub use types::step::{ ActionCall, ActionResult, ExecutionTier, LlmResponse, Step, StepId, StepStatus, TokenUsage, }; +pub use types::mission::{Mission, MissionCadence, MissionId, MissionStatus}; pub use types::thread::{Thread, ThreadConfig, ThreadId, ThreadState, ThreadType}; // ── Re-exports: traits ────────────────────────────────────── @@ -56,6 +58,7 @@ pub use capability::policy::{PolicyDecision, PolicyEngine}; pub use runtime::conversation::ConversationManager; pub use runtime::manager::ThreadManager; pub use runtime::messaging::ThreadOutcome; +pub use runtime::mission::MissionManager; pub use runtime::tree::ThreadTree; pub use types::conversation::{ @@ -74,3 +77,7 @@ pub use memory::RetrievalEngine; // ── Re-exports: reflection ──────────────────────────────────── pub use reflection::ReflectionResult; + +// ── Re-exports: reliability ────────────────────────────────── + +pub use reliability::ReliabilityTracker; diff --git a/crates/ironclaw_engine/src/memory/retrieval.rs b/crates/ironclaw_engine/src/memory/retrieval.rs index 31d13dc2..f1082cfb 100644 --- a/crates/ironclaw_engine/src/memory/retrieval.rs +++ b/crates/ironclaw_engine/src/memory/retrieval.rs @@ -169,6 +169,10 @@ mod tests { async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) } async fn load_active_leases(&self, _: ThreadId) -> Result, EngineError> { Ok(vec![]) } async fn revoke_lease(&self, _: LeaseId, _: &str) -> Result<(), EngineError> { Ok(()) } + async fn save_mission(&self, _: &crate::types::mission::Mission) -> Result<(), EngineError> { Ok(()) } + async fn load_mission(&self, _: crate::types::mission::MissionId) -> Result, EngineError> { Ok(None) } + async fn list_missions(&self, _: ProjectId) -> Result, EngineError> { Ok(vec![]) } + async fn update_mission_status(&self, _: crate::types::mission::MissionId, _: crate::types::mission::MissionStatus) -> Result<(), EngineError> { Ok(()) } } #[test] diff --git a/crates/ironclaw_engine/src/reflection/executor.rs b/crates/ironclaw_engine/src/reflection/executor.rs new file mode 100644 index 00000000..30f5a77d --- /dev/null +++ b/crates/ironclaw_engine/src/reflection/executor.rs @@ -0,0 +1,254 @@ +//! Effect executor for reflection threads. +//! +//! Provides read-only tools that let the reflection CodeAct thread +//! introspect the completed thread, query existing knowledge, and +//! verify tool names against the capability registry. + +use std::sync::Arc; + +use crate::capability::registry::CapabilityRegistry; +use crate::memory::RetrievalEngine; +use crate::traits::effect::{EffectExecutor, ThreadExecutionContext}; +use crate::traits::store::Store; +use crate::types::capability::{ActionDef, CapabilityLease, EffectType}; +use crate::types::error::EngineError; +use crate::types::project::ProjectId; +use crate::types::step::ActionResult; + +/// EffectExecutor that provides reflection-specific read-only tools. +pub struct ReflectionExecutor { + store: Arc, + capabilities: Arc, + transcript: String, + project_id: ProjectId, +} + +impl ReflectionExecutor { + pub fn new( + store: Arc, + capabilities: Arc, + transcript: String, + project_id: ProjectId, + ) -> Self { + Self { + store, + capabilities, + transcript, + project_id, + } + } + + fn action_defs() -> Vec { + vec![ + ActionDef { + name: "get_transcript".into(), + description: "Get the full execution transcript of the completed thread, \ + including messages, tool calls, errors, and outcomes." + .into(), + parameters_schema: serde_json::json!({"type": "object", "properties": {}}), + effects: vec![EffectType::ReadLocal], + requires_approval: false, + }, + ActionDef { + name: "query_memory".into(), + description: "Search existing memory docs in this project for prior knowledge. \ + Use to check if a lesson or issue has already been recorded." + .into(), + parameters_schema: serde_json::json!({ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + "max_docs": {"type": "integer", "description": "Max results (default 5)"} + }, + "required": ["query"] + }), + effects: vec![EffectType::ReadLocal], + requires_approval: false, + }, + ActionDef { + name: "check_tool_exists".into(), + description: "Check if a tool/action exists in the capability registry. \ + Returns whether it exists and lists similar tool names if not found." + .into(), + parameters_schema: serde_json::json!({ + "type": "object", + "properties": { + "name": {"type": "string", "description": "Tool name to check"} + }, + "required": ["name"] + }), + effects: vec![EffectType::ReadLocal], + requires_approval: false, + }, + ActionDef { + name: "list_tools".into(), + description: "List all available tools/actions in the capability registry.".into(), + parameters_schema: serde_json::json!({"type": "object", "properties": {}}), + effects: vec![EffectType::ReadLocal], + requires_approval: false, + }, + ] + } +} + +#[async_trait::async_trait] +impl EffectExecutor for ReflectionExecutor { + async fn execute_action( + &self, + action_name: &str, + parameters: serde_json::Value, + _lease: &CapabilityLease, + _context: &ThreadExecutionContext, + ) -> Result { + let start = std::time::Instant::now(); + let output = match action_name { + "get_transcript" => serde_json::json!({ "transcript": self.transcript }), + + "query_memory" => { + let query = parameters["query"].as_str().unwrap_or(""); + let max_docs = parameters["max_docs"].as_u64().unwrap_or(5) as usize; + let retrieval = RetrievalEngine::new(Arc::clone(&self.store)); + let docs = retrieval + .retrieve_context(self.project_id, query, max_docs) + .await?; + let results: Vec = docs + .iter() + .map(|d| { + serde_json::json!({ + "type": format!("{:?}", d.doc_type), + "title": &d.title, + "content": &d.content, + }) + }) + .collect(); + serde_json::json!({ "docs": results, "count": results.len() }) + } + + "check_tool_exists" => { + let name = parameters["name"].as_str().unwrap_or(""); + let exists = self.capabilities.find_action(name).is_some(); + let similar: Vec = if exists { + vec![] + } else { + // Find tools with similar names (substring or edit-distance-like match) + let name_lower = name.to_lowercase(); + // Normalize: replace hyphens with underscores and vice versa for matching + let alt_name = if name.contains('_') { + name.replace('_', "-") + } else { + name.replace('-', "_") + }; + self.capabilities + .all_actions() + .iter() + .filter(|a| { + let a_lower = a.name.to_lowercase(); + a_lower.contains(&name_lower) + || name_lower.contains(&a_lower) + || a.name == alt_name + }) + .map(|a| a.name.clone()) + .collect() + }; + serde_json::json!({ "exists": exists, "similar": similar }) + } + + "list_tools" => { + let tools: Vec = self + .capabilities + .all_actions() + .iter() + .map(|a| { + serde_json::json!({ + "name": &a.name, + "description": &a.description, + }) + }) + .collect(); + serde_json::json!({ "tools": tools, "count": tools.len() }) + } + + _ => { + return Err(EngineError::Effect { + reason: format!("unknown reflection action: {action_name}"), + }); + } + }; + + Ok(ActionResult { + call_id: String::new(), + action_name: action_name.into(), + output, + is_error: false, + duration: start.elapsed(), + }) + } + + async fn available_actions( + &self, + _leases: &[CapabilityLease], + ) -> Result, EngineError> { + // Reflection tools are always available regardless of leases + Ok(Self::action_defs()) + } +} + +/// Build the system prompt for a reflection CodeAct thread. +pub fn build_reflection_prompt(actions: &[ActionDef], thread_goal: &str) -> String { + let mut prompt = String::from(REFLECTION_PREAMBLE); + + prompt.push_str("\n## Available tools (call as Python functions)\n\n"); + for action in actions { + prompt.push_str(&format!("- `{}(", action.name)); + if let Some(props) = action.parameters_schema.get("properties") + && let Some(obj) = props.as_object() + { + let params: Vec<&str> = obj.keys().map(String::as_str).collect(); + prompt.push_str(¶ms.join(", ")); + } + prompt.push_str(&format!(")` — {}\n", action.description)); + } + + prompt.push_str(&format!( + "\n## Thread Under Analysis\n\nGoal: {thread_goal}\n" + )); + + prompt.push_str(REFLECTION_POSTAMBLE); + prompt +} + +const REFLECTION_PREAMBLE: &str = "\ +You are analyzing a completed agent thread to extract structured knowledge. \ +You have tools to inspect the thread's execution, check existing knowledge, \ +and verify tool names. + +Write Python code in ```repl blocks to analyze the thread."; + +const REFLECTION_POSTAMBLE: &str = r#" + +## Your Task + +1. Call `get_transcript()` to read the thread's execution history +2. Analyze the transcript for: successes, failures, tool errors, lessons learned +3. Call `query_memory(query)` to check if similar knowledge already exists +4. For any tool errors with "not found", call `check_tool_exists(name)` to find the correct name +5. Call `FINAL()` with a JSON object containing a `docs` array: + +```repl +FINAL({ + "docs": [ + {"type": "summary", "title": "...", "content": "2-4 sentence summary"}, + {"type": "lesson", "title": "...", "content": "what was learned"}, + {"type": "spec", "title": "...", "content": "ALIAS: wrong_name -> correct_name"}, + {"type": "playbook", "title": "...", "content": "1. step one\n2. step two"} + ] +}) +``` + +Rules: +- Always include a "summary" doc +- Include "lesson" only if there were errors or workarounds +- Include "spec" only if tool-not-found errors occurred (verify with check_tool_exists) +- Include "playbook" only if the thread completed successfully with 2+ tool calls +- Skip docs that duplicate existing knowledge (check with query_memory first) +- Keep content concise — each doc should be a few sentences, not paragraphs"#; diff --git a/crates/ironclaw_engine/src/reflection/mod.rs b/crates/ironclaw_engine/src/reflection/mod.rs index 2ffbb664..f71e7660 100644 --- a/crates/ironclaw_engine/src/reflection/mod.rs +++ b/crates/ironclaw_engine/src/reflection/mod.rs @@ -1,13 +1,18 @@ //! Post-thread reflection pipeline. //! -//! After a thread completes, [`reflect()`] uses the LLM to produce structured -//! knowledge (MemoryDocs) from the thread's execution trace: +//! After a thread completes, [`reflect()`] spawns a CodeAct thread with +//! reflection-specific tools to produce structured knowledge (MemoryDocs): //! - Summary — what the thread accomplished //! - Lesson — what was learned from errors/workarounds //! - Issue — unresolved problems for follow-up //! - Spec — missing capabilities / tool alias suggestions //! - Playbook — reusable multi-step procedures from successful threads +//! +//! The reflection thread can introspect the completed thread's transcript, +//! query existing knowledge, and verify tool names against the capability +//! registry. [`reflect_simple()`] is a fallback using direct LLM calls. +pub mod executor; pub mod pipeline; -pub use pipeline::{reflect, ReflectionResult}; +pub use pipeline::{reflect, reflect_simple, ReflectionResult}; diff --git a/crates/ironclaw_engine/src/reflection/pipeline.rs b/crates/ironclaw_engine/src/reflection/pipeline.rs index fc7b338f..3fe7e201 100644 --- a/crates/ironclaw_engine/src/reflection/pipeline.rs +++ b/crates/ironclaw_engine/src/reflection/pipeline.rs @@ -1,24 +1,30 @@ //! Reflection pipeline — produces structured knowledge from completed threads. //! -//! After a thread completes, the reflection pipeline uses the LLM to: -//! 1. Summarize what the thread accomplished -//! 2. Extract lessons from failures and workarounds -//! 3. Detect unresolved issues -//! 4. Identify missing capabilities +//! After a thread completes, the reflection pipeline spawns a CodeAct thread +//! that uses reflection-specific tools (transcript inspection, memory queries, +//! tool registry checks) to produce structured MemoryDocs. //! -//! Each produces a MemoryDoc stored in the thread's project scope. +//! The reflection thread runs with [`ThreadType::Reflection`] and its own +//! [`ExecutionLoop`], making it a fully recursive CodeAct agent. use std::sync::Arc; -use tracing::debug; +use tracing::{debug, warn}; -use crate::traits::llm::{LlmBackend, LlmCallConfig}; +use crate::capability::lease::LeaseManager; +use crate::capability::policy::PolicyEngine; +use crate::capability::registry::CapabilityRegistry; +use crate::executor::ExecutionLoop; +use crate::reflection::executor::{build_reflection_prompt, ReflectionExecutor}; +use crate::runtime::messaging::{self, ThreadOutcome}; +use crate::traits::llm::LlmBackend; +use crate::traits::store::Store; use crate::types::error::EngineError; use crate::types::event::EventKind; use crate::types::memory::{DocType, MemoryDoc}; use crate::types::message::ThreadMessage; -use crate::types::step::{LlmResponse, TokenUsage}; -use crate::types::thread::Thread; +use crate::types::step::TokenUsage; +use crate::types::thread::{Thread, ThreadConfig, ThreadType}; /// Result of running the reflection pipeline on a completed thread. pub struct ReflectionResult { @@ -30,16 +36,118 @@ pub struct ReflectionResult { /// Run the reflection pipeline on a completed thread. /// -/// Produces structured knowledge (MemoryDocs) from the thread's messages -/// and events. Uses the LLM for summarization and analysis. +/// Spawns a CodeAct thread with reflection-specific tools that can: +/// - Read the completed thread's execution transcript +/// - Query existing knowledge in the project +/// - Verify tool names against the capability registry +/// +/// The reflection thread produces structured findings via `FINAL()` which +/// are parsed into MemoryDocs. pub async fn reflect( thread: &Thread, llm: &Arc, + store: &Arc, + capabilities: &Arc, +) -> Result { + let transcript = build_transcript(thread); + + // Build the reflection-specific effect executor + let executor: Arc = + Arc::new(ReflectionExecutor::new( + Arc::clone(store), + Arc::clone(capabilities), + transcript, + thread.project_id, + )); + + // Create a reflection thread + let mut refl_thread = Thread::new( + format!("Reflect on: {}", thread.goal), + ThreadType::Reflection, + thread.project_id, + ThreadConfig { + max_iterations: 10, + enable_reflection: false, // no recursive reflection + ..ThreadConfig::default() + }, + ); + + // Build and inject the reflection system prompt + let actions = executor.available_actions(&[]).await?; + let system_prompt = build_reflection_prompt(&actions, &thread.goal); + refl_thread + .messages + .insert(0, ThreadMessage::system(system_prompt)); + refl_thread.add_message(ThreadMessage::user(format!( + "Analyze the completed thread '{}' and produce structured findings.", + thread.goal + ))); + + // Set up infrastructure for the reflection loop + let lease_manager = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let (_signal_tx, signal_rx) = messaging::signal_channel(32); + + // Grant a blanket lease (empty granted_actions = all actions allowed) + let lease = lease_manager + .grant(refl_thread.id, "reflection_tools", vec![], None, None) + .await; + refl_thread.capability_leases.push(lease.id); + + // Run the execution loop + let mut exec_loop = ExecutionLoop::new( + refl_thread, + Arc::clone(llm), + executor, + lease_manager, + policy, + signal_rx, + "system".to_string(), + ); + + let outcome = exec_loop.run().await?; + + // Parse the outcome into MemoryDocs + let response = match outcome { + ThreadOutcome::Completed { response: Some(r) } => r, + ThreadOutcome::Completed { response: None } => String::new(), + ThreadOutcome::Failed { error } => { + warn!( + thread_id = %thread.id, + "reflection thread failed: {error}" + ); + String::new() + } + _ => String::new(), + }; + + let docs = parse_reflection_output(&response, thread); + let tokens_used = TokenUsage { + input_tokens: exec_loop.thread.total_tokens_used, + output_tokens: 0, // total already tracked + ..TokenUsage::default() + }; + + debug!( + thread_id = %thread.id, + docs_produced = docs.len(), + total_tokens = tokens_used.total(), + "reflection complete (CodeAct)" + ); + + Ok(ReflectionResult { docs, tokens_used }) +} + +/// Run a simplified reflection pipeline using direct LLM calls. +/// +/// This is a fallback for when CodeAct execution is not available or when +/// the reflection thread overhead is not desired (e.g., in tests). +pub async fn reflect_simple( + thread: &Thread, + llm: &Arc, ) -> Result { let mut docs = Vec::new(); let mut total_tokens = TokenUsage::default(); - - // Build a transcript of the thread's work for the LLM to analyze let transcript = build_transcript(thread); // 1. Summary doc @@ -69,7 +177,6 @@ pub async fn reflect( if thread_failed || had_errors { let (issue_doc, tokens) = produce_doc(thread, llm, DocType::Issue, &transcript, ISSUE_PROMPT).await?; - // Only add if the LLM produced non-trivial content if issue_doc.content.len() > 20 { docs.push(issue_doc); } @@ -77,7 +184,7 @@ pub async fn reflect( total_tokens.output_tokens += tokens.output_tokens; } - // 4. Missing capabilities (if tool-not-found errors detected) + // 4. Missing capabilities let has_missing_tools = thread.events.iter().any(|e| { if let EventKind::ActionFailed { error, .. } = &e.kind { error.contains("not found") || error.contains("not available") @@ -95,7 +202,7 @@ pub async fn reflect( total_tokens.output_tokens += tokens.output_tokens; } - // 5. Playbook (successful threads with multiple tool-using steps) + // 5. Playbook let action_count = thread .events .iter() @@ -117,7 +224,7 @@ pub async fn reflect( thread_id = %thread.id, docs_produced = docs.len(), total_tokens = total_tokens.total(), - "reflection complete" + "reflection complete (simple)" ); Ok(ReflectionResult { @@ -126,7 +233,76 @@ pub async fn reflect( }) } -// ── Prompts ───────────────────────────────────────────────── +// ── Output parsing ──────────────────────────────────────────── + +/// Parse the FINAL() output from a reflection CodeAct thread into MemoryDocs. +fn parse_reflection_output(response: &str, source_thread: &Thread) -> Vec { + // Try parsing as JSON first (the expected format) + if let Ok(value) = serde_json::from_str::(response) + && let Some(docs_arr) = value.get("docs").and_then(|d| d.as_array()) + { + return docs_arr + .iter() + .filter_map(|doc_val| parse_doc_entry(doc_val, source_thread)) + .collect(); + } + + // If the response is not valid JSON, try to find JSON in the response + if let Some(start) = response.find('{') + && let Some(end) = response.rfind('}') + { + let json_str = &response[start..=end]; + if let Ok(value) = serde_json::from_str::(json_str) + && let Some(docs_arr) = value.get("docs").and_then(|d| d.as_array()) + { + return docs_arr + .iter() + .filter_map(|doc_val| parse_doc_entry(doc_val, source_thread)) + .collect(); + } + } + + // Fallback: treat the entire response as a summary + if response.len() > 20 { + vec![MemoryDoc::new( + source_thread.project_id, + DocType::Summary, + format!("Summary: {}", source_thread.goal), + response, + ) + .with_source_thread(source_thread.id)] + } else { + vec![] + } +} + +/// Parse a single doc entry from the JSON output. +fn parse_doc_entry(value: &serde_json::Value, source_thread: &Thread) -> Option { + let doc_type_str = value.get("type")?.as_str()?; + let title = value.get("title")?.as_str()?; + let content = value.get("content")?.as_str()?; + + if content.len() <= 20 { + return None; + } + + let doc_type = match doc_type_str.to_lowercase().as_str() { + "summary" => DocType::Summary, + "lesson" => DocType::Lesson, + "issue" => DocType::Issue, + "spec" => DocType::Spec, + "playbook" => DocType::Playbook, + "note" => DocType::Note, + _ => return None, + }; + + Some( + MemoryDoc::new(source_thread.project_id, doc_type, title, content) + .with_source_thread(source_thread.id), + ) +} + +// ── Prompts (for reflect_simple fallback) ───────────────────── const SUMMARY_PROMPT: &str = "\ Summarize what this thread accomplished in 2-4 sentences. Include: @@ -166,10 +342,10 @@ This thread successfully completed a multi-step task. Extract a reusable playboo - Describe the pattern so it can be reused for similar tasks Write the playbook as a numbered list of steps. Be specific about tool names and parameters used."; -// ── Helpers ───────────────────────────────────────────────── +// ── Helpers ─────────────────────────────────────────────────── /// Build a concise transcript of the thread's work. -fn build_transcript(thread: &Thread) -> String { +pub(crate) fn build_transcript(thread: &Thread) -> String { let mut parts = Vec::new(); parts.push(format!("Goal: {}", thread.goal)); @@ -198,9 +374,9 @@ fn build_transcript(thread: &Thread) -> String { .events .iter() .filter_map(|e| match &e.kind { - EventKind::ActionFailed { action_name, error, .. } => { - Some(format!("Action '{action_name}' failed: {error}")) - } + EventKind::ActionFailed { + action_name, error, .. + } => Some(format!("Action '{action_name}' failed: {error}")), EventKind::StepFailed { error, .. } => Some(format!("Step failed: {error}")), _ => None, }) @@ -231,18 +407,17 @@ async fn produce_doc( ThreadMessage::user(prompt.to_string()), ]; - let config = LlmCallConfig { + let config = crate::traits::llm::LlmCallConfig { force_text: true, - ..LlmCallConfig::default() + ..crate::traits::llm::LlmCallConfig::default() }; let output = llm.complete(&messages, &[], &config).await?; let content = match output.response { - LlmResponse::Text(t) => t, - LlmResponse::ActionCalls { content, .. } | LlmResponse::Code { content, .. } => { - content.unwrap_or_default() - } + crate::types::step::LlmResponse::Text(t) => t, + crate::types::step::LlmResponse::ActionCalls { content, .. } + | crate::types::step::LlmResponse::Code { content, .. } => content.unwrap_or_default(), }; let title = match doc_type { @@ -268,7 +443,7 @@ mod tests { use crate::types::event::ThreadEvent; use crate::types::project::ProjectId; use crate::types::step::TokenUsage; - use crate::types::thread::{ThreadConfig, ThreadType}; + use crate::types::thread::ThreadConfig; use std::sync::Mutex; struct MockLlm { @@ -298,7 +473,7 @@ mod tests { r.remove(0) }; Ok(LlmOutput { - response: LlmResponse::Text(text), + response: crate::types::step::LlmResponse::Text(text), usage: TokenUsage { input_tokens: 100, output_tokens: 50, @@ -323,18 +498,21 @@ mod tests { thread } - #[tokio::test] - async fn reflect_produces_summary_for_clean_thread() { - let thread = make_completed_thread(); - let llm = MockLlm::with_responses(vec!["Thread accomplished the test task successfully."]); + // ── reflect_simple tests (direct LLM calls) ──────────────── - let result = reflect(&thread, &llm).await.unwrap(); + #[tokio::test] + async fn reflect_simple_produces_summary() { + let thread = make_completed_thread(); + let llm = + MockLlm::with_responses(vec!["Thread accomplished the test task successfully."]); + + let result = reflect_simple(&thread, &llm).await.unwrap(); assert_eq!(result.docs.len(), 1); assert_eq!(result.docs[0].doc_type, DocType::Summary); } #[tokio::test] - async fn reflect_produces_lesson_on_errors() { + async fn reflect_simple_produces_lesson_on_errors() { let mut thread = make_completed_thread(); thread.events.push(ThreadEvent::new( thread.id, @@ -353,7 +531,7 @@ mod tests { "ALIAS: web_search -> web-search", ]); - let result = reflect(&thread, &llm).await.unwrap(); + let result = reflect_simple(&thread, &llm).await.unwrap(); let types: Vec = result.docs.iter().map(|d| d.doc_type).collect(); assert!(types.contains(&DocType::Summary)); assert!(types.contains(&DocType::Lesson)); @@ -362,7 +540,7 @@ mod tests { } #[tokio::test] - async fn reflect_produces_spec_on_tool_not_found() { + async fn reflect_simple_produces_spec_on_tool_not_found() { let mut thread = make_completed_thread(); thread.events.push(ThreadEvent::new( thread.id, @@ -381,7 +559,7 @@ mod tests { "MISSING: missing_tool -> needs implementation", ]); - let result = reflect(&thread, &llm).await.unwrap(); + let result = reflect_simple(&thread, &llm).await.unwrap(); let spec_docs: Vec<&MemoryDoc> = result .docs .iter() @@ -392,9 +570,8 @@ mod tests { } #[tokio::test] - async fn reflect_produces_playbook_on_successful_multi_step() { + async fn reflect_simple_produces_playbook_on_multi_step() { let mut thread = make_completed_thread(); - // Add 2+ action executed events to trigger playbook thread.events.push(ThreadEvent::new( thread.id, EventKind::ActionExecuted { @@ -416,23 +593,19 @@ mod tests { let llm = MockLlm::with_responses(vec![ "Summary of successful thread.", - "1. Search web for topic\n2. Analyze results with llm_query\n3. Return summary", + "1. Search web\n2. Analyze results\n3. Return summary", ]); - let result = reflect(&thread, &llm).await.unwrap(); - let playbook_docs: Vec<&MemoryDoc> = result + let result = reflect_simple(&thread, &llm).await.unwrap(); + assert!(result .docs .iter() - .filter(|d| d.doc_type == DocType::Playbook) - .collect(); - assert_eq!(playbook_docs.len(), 1); - assert!(playbook_docs[0].title.starts_with("Playbook:")); + .any(|d| d.doc_type == DocType::Playbook)); } #[tokio::test] - async fn reflect_skips_playbook_for_single_action() { + async fn reflect_simple_skips_playbook_for_single_action() { let mut thread = make_completed_thread(); - // Only 1 action — not enough for a playbook thread.events.push(ThreadEvent::new( thread.id, EventKind::ActionExecuted { @@ -445,12 +618,64 @@ mod tests { let llm = MockLlm::with_responses(vec!["Simple summary."]); - let result = reflect(&thread, &llm).await.unwrap(); - let playbook_docs: Vec<&MemoryDoc> = result + let result = reflect_simple(&thread, &llm).await.unwrap(); + assert!(!result .docs .iter() - .filter(|d| d.doc_type == DocType::Playbook) - .collect(); - assert!(playbook_docs.is_empty()); + .any(|d| d.doc_type == DocType::Playbook)); + } + + // ── parse_reflection_output tests ────────────────────────── + + #[test] + fn parse_valid_json_output() { + let thread = make_completed_thread(); + let json = r#"{"docs": [ + {"type": "summary", "title": "Summary: test", "content": "The thread completed successfully with good results."}, + {"type": "lesson", "title": "Lesson: test", "content": "Always check tool names before calling them."} + ]}"#; + + let docs = parse_reflection_output(json, &thread); + assert_eq!(docs.len(), 2); + assert_eq!(docs[0].doc_type, DocType::Summary); + assert_eq!(docs[1].doc_type, DocType::Lesson); + } + + #[test] + fn parse_json_embedded_in_text() { + let thread = make_completed_thread(); + let text = r#"Here are my findings: {"docs": [{"type": "summary", "title": "test", "content": "The thread did something interesting and useful."}]} end"#; + + let docs = parse_reflection_output(text, &thread); + assert_eq!(docs.len(), 1); + } + + #[test] + fn parse_fallback_to_summary() { + let thread = make_completed_thread(); + let text = "This is a plain text response with enough content to be a valid summary doc."; + + let docs = parse_reflection_output(text, &thread); + assert_eq!(docs.len(), 1); + assert_eq!(docs[0].doc_type, DocType::Summary); + } + + #[test] + fn parse_skips_short_content() { + let thread = make_completed_thread(); + let json = + r#"{"docs": [{"type": "summary", "title": "test", "content": "too short"}]}"#; + + let docs = parse_reflection_output(json, &thread); + assert!(docs.is_empty()); + } + + #[test] + fn parse_skips_unknown_doc_type() { + let thread = make_completed_thread(); + let json = r#"{"docs": [{"type": "unknown_type", "title": "test", "content": "This has enough content but unknown type so it gets skipped."}]}"#; + + let docs = parse_reflection_output(json, &thread); + assert!(docs.is_empty()); } } diff --git a/crates/ironclaw_engine/src/reliability.rs b/crates/ironclaw_engine/src/reliability.rs new file mode 100644 index 00000000..63b4b5e8 --- /dev/null +++ b/crates/ironclaw_engine/src/reliability.rs @@ -0,0 +1,198 @@ +//! Tool reliability tracking with exponential moving averages. +//! +//! Tracks per-action success rate and latency using EMA (exponential moving +//! average) to smooth out noise. This data can be injected into the context +//! builder to inform the LLM about unreliable tools. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::RwLock; + +/// EMA smoothing factor. Higher = more weight on recent observations. +const EMA_ALPHA: f64 = 0.3; + +/// Per-action reliability metrics. +#[derive(Debug, Clone)] +pub struct ActionMetrics { + /// EMA of success rate (0.0 to 1.0). + pub success_rate: f64, + /// EMA of latency in milliseconds. + pub avg_latency_ms: f64, + /// Total number of calls recorded. + pub call_count: u64, + /// Last error message (if any). + pub last_error: Option, +} + +impl Default for ActionMetrics { + fn default() -> Self { + Self { + success_rate: 1.0, // assume success until proven otherwise + avg_latency_ms: 0.0, + call_count: 0, + last_error: None, + } + } +} + +/// Thread-safe registry of per-action reliability metrics. +#[derive(Clone)] +pub struct ReliabilityTracker { + metrics: Arc>>, +} + +impl ReliabilityTracker { + pub fn new() -> Self { + Self { + metrics: Arc::new(RwLock::new(HashMap::new())), + } + } + + /// Record a successful action execution. + pub async fn record_success(&self, action_name: &str, latency: Duration) { + let mut metrics = self.metrics.write().await; + let entry = metrics.entry(action_name.to_string()).or_default(); + entry.call_count += 1; + let latency_ms = latency.as_millis() as f64; + + if entry.call_count == 1 { + // First observation — use raw values + entry.avg_latency_ms = latency_ms; + // success_rate stays at 1.0 + } else { + entry.success_rate = ema(entry.success_rate, 1.0); + entry.avg_latency_ms = ema(entry.avg_latency_ms, latency_ms); + } + } + + /// Record a failed action execution. + pub async fn record_failure(&self, action_name: &str, error: &str) { + let mut metrics = self.metrics.write().await; + let entry = metrics.entry(action_name.to_string()).or_default(); + entry.call_count += 1; + entry.last_error = Some(error.to_string()); + + if entry.call_count == 1 { + entry.success_rate = 0.0; + } else { + entry.success_rate = ema(entry.success_rate, 0.0); + } + } + + /// Get metrics for a specific action. + pub async fn get_metrics(&self, action_name: &str) -> Option { + let metrics = self.metrics.read().await; + metrics.get(action_name).cloned() + } + + /// Get all metrics, sorted by success rate (worst first). + pub async fn all_metrics(&self) -> Vec<(String, ActionMetrics)> { + let metrics = self.metrics.read().await; + let mut entries: Vec<(String, ActionMetrics)> = metrics + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + entries.sort_by(|a, b| { + a.1.success_rate + .partial_cmp(&b.1.success_rate) + .unwrap_or(std::cmp::Ordering::Equal) + }); + entries + } + + /// Get actions with reliability below a threshold. + pub async fn unreliable_actions(&self, threshold: f64) -> Vec<(String, ActionMetrics)> { + let all = self.all_metrics().await; + all.into_iter() + .filter(|(_, m)| m.success_rate < threshold) + .collect() + } +} + +impl Default for ReliabilityTracker { + fn default() -> Self { + Self::new() + } +} + +/// Compute exponential moving average. +fn ema(prev: f64, new: f64) -> f64 { + EMA_ALPHA * new + (1.0 - EMA_ALPHA) * prev +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ema_moves_toward_new() { + let result = ema(1.0, 0.0); + // 0.3 * 0.0 + 0.7 * 1.0 = 0.7 + assert!((result - 0.7).abs() < f64::EPSILON); + } + + #[test] + fn ema_converges_on_repeated() { + let mut val = 1.0; + for _ in 0..20 { + val = ema(val, 0.0); + } + // Should converge toward 0.0 + assert!(val < 0.01); + } + + #[tokio::test] + async fn track_success() { + let tracker = ReliabilityTracker::new(); + tracker + .record_success("tool_a", Duration::from_millis(100)) + .await; + tracker + .record_success("tool_a", Duration::from_millis(200)) + .await; + + let m = tracker.get_metrics("tool_a").await.unwrap(); + assert_eq!(m.call_count, 2); + assert!((m.success_rate - 1.0).abs() < f64::EPSILON); + assert!(m.avg_latency_ms > 100.0); // EMA of 100 and 200 + } + + #[tokio::test] + async fn track_failure_lowers_success_rate() { + let tracker = ReliabilityTracker::new(); + tracker + .record_success("tool_b", Duration::from_millis(50)) + .await; + tracker + .record_failure("tool_b", "not found") + .await; + + let m = tracker.get_metrics("tool_b").await.unwrap(); + assert_eq!(m.call_count, 2); + assert!(m.success_rate < 1.0); + assert_eq!(m.last_error, Some("not found".into())); + } + + #[tokio::test] + async fn unreliable_actions_filters() { + let tracker = ReliabilityTracker::new(); + tracker + .record_success("good_tool", Duration::from_millis(10)) + .await; + tracker + .record_failure("bad_tool", "always fails") + .await; + + let unreliable = tracker.unreliable_actions(0.5).await; + assert_eq!(unreliable.len(), 1); + assert_eq!(unreliable[0].0, "bad_tool"); + } + + #[tokio::test] + async fn unknown_action_returns_none() { + let tracker = ReliabilityTracker::new(); + assert!(tracker.get_metrics("nonexistent").await.is_none()); + } +} diff --git a/crates/ironclaw_engine/src/runtime/conversation.rs b/crates/ironclaw_engine/src/runtime/conversation.rs index 156b2265..012a22c1 100644 --- a/crates/ironclaw_engine/src/runtime/conversation.rs +++ b/crates/ironclaw_engine/src/runtime/conversation.rs @@ -327,6 +327,10 @@ mod tests { async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) } async fn load_active_leases(&self, _: ThreadId) -> Result, EngineError> { Ok(vec![]) } async fn revoke_lease(&self, _: crate::types::capability::LeaseId, _: &str) -> Result<(), EngineError> { Ok(()) } + async fn save_mission(&self, _: &crate::types::mission::Mission) -> Result<(), EngineError> { Ok(()) } + async fn load_mission(&self, _: crate::types::mission::MissionId) -> Result, EngineError> { Ok(None) } + async fn list_missions(&self, _: ProjectId) -> Result, EngineError> { Ok(vec![]) } + async fn update_mission_status(&self, _: crate::types::mission::MissionId, _: crate::types::mission::MissionStatus) -> Result<(), EngineError> { Ok(()) } } fn make_conv_manager() -> (Arc, ConversationManager) { diff --git a/crates/ironclaw_engine/src/runtime/manager.rs b/crates/ironclaw_engine/src/runtime/manager.rs index 4262cc37..a498e5d5 100644 --- a/crates/ironclaw_engine/src/runtime/manager.rs +++ b/crates/ironclaw_engine/src/runtime/manager.rs @@ -161,49 +161,82 @@ impl ThreadManager { // Spawn background task let store_for_task = Arc::clone(&self.store); let llm_for_reflection = Arc::clone(&self.llm); + let caps_for_reflection = Arc::clone(&self.capabilities); + let event_tx = self.event_tx.clone(); let handle = tokio::spawn(async move { let mut exec = exec_loop; let result = exec.run().await; debug!(thread_id = %thread_id, "thread execution finished"); + // Helper to emit events on both the thread and broadcast channel + let emit = |thread: &mut crate::types::thread::Thread, kind: crate::types::event::EventKind| { + let event = crate::types::event::ThreadEvent::new(thread.id, kind); + let _ = event_tx.send(event.clone()); + thread.events.push(event); + thread.updated_at = chrono::Utc::now(); + }; + // Run retrospective trace analysis (non-LLM, always runs) - let trace = crate::executor::trace::build_trace(&exec.thread); + let mut trace = crate::executor::trace::build_trace(&exec.thread); if !trace.issues.is_empty() { crate::executor::trace::log_trace_summary(&trace); } - // Write trace file if enabled - if crate::executor::trace::is_trace_enabled() { - crate::executor::trace::write_trace(&trace); - } - // Run LLM reflection if enabled and thread completed if exec.thread.config.enable_reflection - && (exec.thread.state == crate::types::thread::ThreadState::Completed - || exec.thread.state == crate::types::thread::ThreadState::Done) + && exec.thread.state == crate::types::thread::ThreadState::Completed { - debug!(thread_id = %thread_id, "running reflection pipeline"); - match crate::reflection::reflect(&exec.thread, &llm_for_reflection).await { - Ok(reflection) => { - debug!( - thread_id = %thread_id, - docs = reflection.docs.len(), - tokens = reflection.tokens_used.total(), - "reflection complete" - ); - for doc in &reflection.docs { - let _ = store_for_task.save_memory_doc(doc).await; + // Transition: Completed → Reflecting + if let Err(e) = exec.thread.transition_to( + crate::types::thread::ThreadState::Reflecting, + Some("starting reflection".into()), + ) { + tracing::warn!(thread_id = %thread_id, "failed to transition to Reflecting: {e}"); + } else { + emit(&mut exec.thread, crate::types::event::EventKind::ReflectionStarted); + + match crate::reflection::reflect(&exec.thread, &llm_for_reflection, &store_for_task, &caps_for_reflection).await { + Ok(reflection) => { + let doc_types: Vec = reflection + .docs + .iter() + .map(|d| format!("{:?}", d.doc_type)) + .collect(); + + emit(&mut exec.thread, crate::types::event::EventKind::ReflectionComplete { + docs_produced: reflection.docs.len(), + doc_types, + tokens_used: reflection.tokens_used.total(), + }); + + // Attach reflection results to the trace + crate::executor::trace::attach_reflection(&mut trace, &reflection); + + for doc in &reflection.docs { + let _ = store_for_task.save_memory_doc(doc).await; + } + } + Err(e) => { + emit(&mut exec.thread, crate::types::event::EventKind::ReflectionFailed { + error: e.to_string(), + }); } } - Err(e) => { - tracing::warn!( - thread_id = %thread_id, - "reflection failed: {e}" - ); - } + + // Transition: Reflecting → Done + let _ = exec.thread.transition_to( + crate::types::thread::ThreadState::Done, + Some("reflection finished".into()), + ); } } + // Write trace file if enabled (after reflection, so it's included) + if crate::executor::trace::is_trace_enabled() { + crate::executor::trace::log_trace_summary(&trace); + crate::executor::trace::write_trace(&trace); + } + // Save final thread state to store let _ = store_for_task.save_thread(&exec.thread).await; result @@ -411,6 +444,10 @@ mod tests { async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) } async fn load_active_leases(&self, _: ThreadId) -> Result, EngineError> { Ok(vec![]) } async fn revoke_lease(&self, _: crate::types::capability::LeaseId, _: &str) -> Result<(), EngineError> { Ok(()) } + async fn save_mission(&self, _: &crate::types::mission::Mission) -> Result<(), EngineError> { Ok(()) } + async fn load_mission(&self, _: crate::types::mission::MissionId) -> Result, EngineError> { Ok(None) } + async fn list_missions(&self, _: ProjectId) -> Result, EngineError> { Ok(vec![]) } + async fn update_mission_status(&self, _: crate::types::mission::MissionId, _: crate::types::mission::MissionStatus) -> Result<(), EngineError> { Ok(()) } } fn make_manager(llm: Arc) -> ThreadManager { diff --git a/crates/ironclaw_engine/src/runtime/mission.rs b/crates/ironclaw_engine/src/runtime/mission.rs new file mode 100644 index 00000000..4466c6d9 --- /dev/null +++ b/crates/ironclaw_engine/src/runtime/mission.rs @@ -0,0 +1,174 @@ +//! Mission manager — orchestrates long-running goals that spawn threads over time. +//! +//! Missions track ongoing objectives and periodically spawn threads to make +//! progress. The manager handles lifecycle (create, pause, resume, complete) +//! and delegates thread spawning to [`ThreadManager`]. + +use std::sync::Arc; + +use tokio::sync::RwLock; +use tracing::{debug, warn}; + +use crate::runtime::manager::ThreadManager; +use crate::traits::store::Store; +use crate::types::error::EngineError; +use crate::types::mission::{Mission, MissionCadence, MissionId, MissionStatus}; +use crate::types::project::ProjectId; +use crate::types::thread::{ThreadConfig, ThreadId, ThreadType}; + +/// Manages mission lifecycle and thread spawning. +pub struct MissionManager { + store: Arc, + thread_manager: Arc, + /// Active missions indexed by ID for quick lookup. + active: RwLock>, +} + +impl MissionManager { + pub fn new(store: Arc, thread_manager: Arc) -> Self { + Self { + store, + thread_manager, + active: RwLock::new(Vec::new()), + } + } + + /// Create and persist a new mission. Returns the mission ID. + pub async fn create_mission( + &self, + project_id: ProjectId, + name: impl Into, + goal: impl Into, + cadence: MissionCadence, + ) -> Result { + let mission = Mission::new(project_id, name, goal, cadence); + let id = mission.id; + self.store.save_mission(&mission).await?; + self.active.write().await.push(id); + debug!(mission_id = %id, "mission created"); + Ok(id) + } + + /// Pause an active mission. No new threads will be spawned. + pub async fn pause_mission(&self, id: MissionId) -> Result<(), EngineError> { + self.store + .update_mission_status(id, MissionStatus::Paused) + .await?; + debug!(mission_id = %id, "mission paused"); + Ok(()) + } + + /// Resume a paused mission. + pub async fn resume_mission(&self, id: MissionId) -> Result<(), EngineError> { + self.store + .update_mission_status(id, MissionStatus::Active) + .await?; + debug!(mission_id = %id, "mission resumed"); + Ok(()) + } + + /// Mark a mission as completed. + pub async fn complete_mission(&self, id: MissionId) -> Result<(), EngineError> { + self.store + .update_mission_status(id, MissionStatus::Completed) + .await?; + self.active.write().await.retain(|mid| *mid != id); + debug!(mission_id = %id, "mission completed"); + Ok(()) + } + + /// Manually fire a mission — spawn a thread for it right now. + pub async fn fire_mission( + &self, + id: MissionId, + user_id: &str, + ) -> Result, EngineError> { + let mission = self.store.load_mission(id).await?; + let mission = match mission { + Some(m) => m, + None => { + return Err(EngineError::Store { + reason: format!("mission {id} not found"), + }); + } + }; + + if mission.is_terminal() { + warn!(mission_id = %id, status = ?mission.status, "cannot fire terminal mission"); + return Ok(None); + } + + let thread_id = self + .thread_manager + .spawn_thread( + &mission.goal, + ThreadType::Mission, + mission.project_id, + ThreadConfig { + enable_reflection: true, + ..ThreadConfig::default() + }, + None, + user_id, + ) + .await?; + + // Record the thread in mission history + let mut updated = mission; + updated.record_thread(thread_id); + self.store.save_mission(&updated).await?; + + debug!(mission_id = %id, thread_id = %thread_id, "mission fired"); + Ok(Some(thread_id)) + } + + /// List all missions in a project. + pub async fn list_missions( + &self, + project_id: ProjectId, + ) -> Result, EngineError> { + self.store.list_missions(project_id).await + } + + /// Get a mission by ID. + pub async fn get_mission(&self, id: MissionId) -> Result, EngineError> { + self.store.load_mission(id).await + } + + /// Tick — check all active missions and fire any that are due. + /// + /// For `Cron` cadence missions, checks `next_fire_at` against current time. + /// For `Manual` missions, this is a no-op. + /// Returns the IDs of threads spawned. + pub async fn tick(&self, user_id: &str) -> Result, EngineError> { + let active_ids = self.active.read().await.clone(); + let mut spawned = Vec::new(); + let now = chrono::Utc::now(); + + for mid in active_ids { + let mission = match self.store.load_mission(mid).await? { + Some(m) if m.status == MissionStatus::Active => m, + _ => continue, + }; + + let should_fire = match &mission.cadence { + MissionCadence::Cron { .. } => { + // Fire if next_fire_at has passed + mission + .next_fire_at + .is_some_and(|next| next <= now) + } + MissionCadence::Manual => false, + MissionCadence::OnEvent { .. } | MissionCadence::OnPush => false, + }; + + if should_fire + && let Some(tid) = self.fire_mission(mid, user_id).await? + { + spawned.push(tid); + } + } + + Ok(spawned) + } +} diff --git a/crates/ironclaw_engine/src/runtime/mod.rs b/crates/ironclaw_engine/src/runtime/mod.rs index ed341e35..6d757a7b 100644 --- a/crates/ironclaw_engine/src/runtime/mod.rs +++ b/crates/ironclaw_engine/src/runtime/mod.rs @@ -7,9 +7,11 @@ pub mod conversation; pub mod manager; pub mod messaging; +pub mod mission; pub mod tree; pub use conversation::ConversationManager; pub use manager::ThreadManager; pub use messaging::ThreadOutcome; +pub use mission::MissionManager; pub use tree::ThreadTree; diff --git a/crates/ironclaw_engine/src/traits/store.rs b/crates/ironclaw_engine/src/traits/store.rs index 2ed3f4df..6e858dd5 100644 --- a/crates/ironclaw_engine/src/traits/store.rs +++ b/crates/ironclaw_engine/src/traits/store.rs @@ -7,6 +7,7 @@ use crate::types::capability::{CapabilityLease, LeaseId}; use crate::types::error::EngineError; use crate::types::event::ThreadEvent; use crate::types::memory::{DocId, MemoryDoc}; +use crate::types::mission::{Mission, MissionId, MissionStatus}; use crate::types::project::{Project, ProjectId}; use crate::types::step::Step; use crate::types::thread::{Thread, ThreadId, ThreadState}; @@ -54,4 +55,15 @@ pub trait Store: Send + Sync { thread_id: ThreadId, ) -> Result, EngineError>; async fn revoke_lease(&self, lease_id: LeaseId, reason: &str) -> Result<(), EngineError>; + + // ── Mission operations ─────────────────────────────────── + + async fn save_mission(&self, mission: &Mission) -> Result<(), EngineError>; + async fn load_mission(&self, id: MissionId) -> Result, EngineError>; + async fn list_missions(&self, project_id: ProjectId) -> Result, EngineError>; + async fn update_mission_status( + &self, + id: MissionId, + status: MissionStatus, + ) -> Result<(), EngineError>; } diff --git a/crates/ironclaw_engine/src/types/event.rs b/crates/ironclaw_engine/src/types/event.rs index 661a5a14..61d4d3f4 100644 --- a/crates/ironclaw_engine/src/types/event.rs +++ b/crates/ironclaw_engine/src/types/event.rs @@ -121,4 +121,15 @@ pub enum EventKind { call_id: String, approved: bool, }, + + // ── Reflection ─────────────────────────────────────────── + ReflectionStarted, + ReflectionComplete { + docs_produced: usize, + doc_types: Vec, + tokens_used: u64, + }, + ReflectionFailed { + error: String, + }, } diff --git a/crates/ironclaw_engine/src/types/mission.rs b/crates/ironclaw_engine/src/types/mission.rs new file mode 100644 index 00000000..94870e41 --- /dev/null +++ b/crates/ironclaw_engine/src/types/mission.rs @@ -0,0 +1,124 @@ +//! Missions — long-running goals that spawn threads over time. +//! +//! A mission represents an ongoing objective that periodically spawns +//! threads to make progress. Missions can run on a schedule (cron), +//! in response to events, or be triggered manually. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::types::project::ProjectId; +use crate::types::thread::ThreadId; + +/// Strongly-typed mission identifier. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct MissionId(pub Uuid); + +impl MissionId { + pub fn new() -> Self { + Self(Uuid::new_v4()) + } +} + +impl Default for MissionId { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Display for MissionId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Lifecycle status of a mission. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum MissionStatus { + /// Mission is actively spawning threads on cadence. + Active, + /// Mission is paused — no new threads will be spawned. + Paused, + /// Mission has achieved its goal. + Completed, + /// Mission has been abandoned or failed irrecoverably. + Failed, +} + +/// How a mission triggers new threads. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum MissionCadence { + /// Spawn on a cron schedule (e.g., "0 */6 * * *" for every 6 hours). + Cron { expression: String }, + /// Spawn in response to a named event. + OnEvent { event_pattern: String }, + /// Spawn when code is pushed (webhook-driven). + OnPush, + /// Only spawn when manually triggered. + Manual, +} + +/// A mission — a long-running goal that spawns threads over time. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Mission { + pub id: MissionId, + pub project_id: ProjectId, + pub name: String, + pub goal: String, + pub status: MissionStatus, + pub cadence: MissionCadence, + /// History of threads spawned by this mission. + pub thread_history: Vec, + /// Optional criteria for declaring the mission complete. + pub success_criteria: Option, + pub metadata: serde_json::Value, + pub created_at: DateTime, + pub updated_at: DateTime, + /// When the next thread should be spawned (for Cron cadence). + pub next_fire_at: Option>, +} + +impl Mission { + pub fn new( + project_id: ProjectId, + name: impl Into, + goal: impl Into, + cadence: MissionCadence, + ) -> Self { + let now = Utc::now(); + Self { + id: MissionId::new(), + project_id, + name: name.into(), + goal: goal.into(), + status: MissionStatus::Active, + cadence, + thread_history: Vec::new(), + success_criteria: None, + metadata: serde_json::Value::Object(serde_json::Map::new()), + created_at: now, + updated_at: now, + next_fire_at: None, + } + } + + pub fn with_success_criteria(mut self, criteria: impl Into) -> Self { + self.success_criteria = Some(criteria.into()); + self + } + + /// Record that a thread was spawned for this mission. + pub fn record_thread(&mut self, thread_id: ThreadId) { + self.thread_history.push(thread_id); + self.updated_at = Utc::now(); + } + + /// Whether the mission is in a terminal state. + pub fn is_terminal(&self) -> bool { + matches!( + self.status, + MissionStatus::Completed | MissionStatus::Failed + ) + } +} diff --git a/crates/ironclaw_engine/src/types/mod.rs b/crates/ironclaw_engine/src/types/mod.rs index 73a46a63..3d42860b 100644 --- a/crates/ironclaw_engine/src/types/mod.rs +++ b/crates/ironclaw_engine/src/types/mod.rs @@ -9,6 +9,7 @@ pub mod error; pub mod event; pub mod memory; pub mod message; +pub mod mission; pub mod project; pub mod provenance; pub mod step; diff --git a/crates/ironclaw_engine/src/types/step.rs b/crates/ironclaw_engine/src/types/step.rs index e4d3d504..f09bcb0e 100644 --- a/crates/ironclaw_engine/src/types/step.rs +++ b/crates/ironclaw_engine/src/types/step.rs @@ -139,6 +139,8 @@ pub struct TokenUsage { pub output_tokens: u64, pub cache_read_tokens: u64, pub cache_write_tokens: u64, + /// USD cost for this call (populated by LlmBackend if cost data is available). + pub cost_usd: f64, } impl TokenUsage { diff --git a/src/bridge/llm_adapter.rs b/src/bridge/llm_adapter.rs index 525cbb7a..eda5597d 100644 --- a/src/bridge/llm_adapter.rs +++ b/src/bridge/llm_adapter.rs @@ -92,6 +92,7 @@ impl LlmBackend for LlmBridgeAdapter { output_tokens: u64::from(response.output_tokens), cache_read_tokens: u64::from(response.cache_read_input_tokens), cache_write_tokens: u64::from(response.cache_creation_input_tokens), + cost_usd: 0.0, }, }); } @@ -144,6 +145,7 @@ impl LlmBackend for LlmBridgeAdapter { output_tokens: u64::from(response.output_tokens), cache_read_tokens: u64::from(response.cache_read_input_tokens), cache_write_tokens: u64::from(response.cache_creation_input_tokens), + cost_usd: 0.0, // TODO: populate from provider cost data when available }, }) } diff --git a/src/bridge/store_adapter.rs b/src/bridge/store_adapter.rs index ad6e3f39..23de837e 100644 --- a/src/bridge/store_adapter.rs +++ b/src/bridge/store_adapter.rs @@ -10,6 +10,7 @@ use tokio::sync::RwLock; use ironclaw_engine::{ CapabilityLease, DocId, EngineError, LeaseId, MemoryDoc, Project, ProjectId, Step, Thread, ThreadEvent, ThreadId, ThreadState, Store, + types::mission::{Mission, MissionId, MissionStatus}, }; /// In-memory implementation of the engine's `Store` trait. @@ -23,6 +24,7 @@ pub struct InMemoryStore { projects: RwLock>, docs: RwLock>, leases: RwLock>, + missions: RwLock>, } impl InMemoryStore { @@ -34,6 +36,7 @@ impl InMemoryStore { projects: RwLock::new(HashMap::new()), docs: RwLock::new(HashMap::new()), leases: RwLock::new(HashMap::new()), + missions: RwLock::new(HashMap::new()), } } } @@ -184,4 +187,26 @@ impl Store for InMemoryStore { } Ok(()) } + + // ── Mission ────────────────────────────────────────────── + + async fn save_mission(&self, mission: &Mission) -> Result<(), EngineError> { + self.missions.write().await.insert(mission.id, mission.clone()); + Ok(()) + } + + async fn load_mission(&self, id: MissionId) -> Result, EngineError> { + Ok(self.missions.read().await.get(&id).cloned()) + } + + async fn list_missions(&self, project_id: ProjectId) -> Result, EngineError> { + Ok(self.missions.read().await.values().filter(|m| m.project_id == project_id).cloned().collect()) + } + + async fn update_mission_status(&self, id: MissionId, status: MissionStatus) -> Result<(), EngineError> { + if let Some(mission) = self.missions.write().await.get_mut(&id) { + mission.status = status; + } + Ok(()) + } }