feat(engine): add missions, reliability tracker, reflection executor, and provenance-aware policy

- Add Mission type and MissionManager for recurring thread scheduling - Add ReliabilityTracker for per-capability success/failure/latency tracking - Add reflection executor that spawns CodeAct threads for post-completion reflection - Extend PolicyEngine with provenance-aware taint checking (LLM-generated data requires approval for financial/external-write effects) - Extend Store trait with mission CRUD methods - Add conversation surface tracking, compaction token fix, context memory injection - Wire new modules through lib.rs re-exports and bridge adapters Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-08-25 14:53:34 +00:00 · 2026-03-23 20:21:08 -07:00
parent 65a32c81b5
commit 10098e7958
20 changed files with 1305 additions and 103 deletions
@@ -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<PolicyRule>,
    /// Effect types that are always denied unless explicitly overridden.
-    denied_effects: Vec<EffectType>,
+    pub(crate) denied_effects: Vec<EffectType>,
 }

 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();
@@ -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,
@@ -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<Vec<CapabilityLease>, 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<Option<crate::types::mission::Mission>, EngineError> { Ok(None) }
+        async fn list_missions(&self, _: ProjectId) -> Result<Vec<crate::types::mission::Mission>, 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]
@@ -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;
@@ -169,6 +169,10 @@ mod tests {
        async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) }
        async fn load_active_leases(&self, _: ThreadId) -> Result<Vec<CapabilityLease>, 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<Option<crate::types::mission::Mission>, EngineError> { Ok(None) }
+        async fn list_missions(&self, _: ProjectId) -> Result<Vec<crate::types::mission::Mission>, EngineError> { Ok(vec![]) }
+        async fn update_mission_status(&self, _: crate::types::mission::MissionId, _: crate::types::mission::MissionStatus) -> Result<(), EngineError> { Ok(()) }
    }

    #[test]
@@ -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<dyn Store>,
+    capabilities: Arc<CapabilityRegistry>,
+    transcript: String,
+    project_id: ProjectId,
+}
+
+impl ReflectionExecutor {
+    pub fn new(
+        store: Arc<dyn Store>,
+        capabilities: Arc<CapabilityRegistry>,
+        transcript: String,
+        project_id: ProjectId,
+    ) -> Self {
+        Self {
+            store,
+            capabilities,
+            transcript,
+            project_id,
+        }
+    }
+
+    fn action_defs() -> Vec<ActionDef> {
+        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<ActionResult, EngineError> {
+        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<serde_json::Value> = 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<String> = 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<serde_json::Value> = 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<Vec<ActionDef>, 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(&params.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"#;
@@ -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};
@@ -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<dyn LlmBackend>,
+    store: &Arc<dyn Store>,
+    capabilities: &Arc<CapabilityRegistry>,
+) -> Result<ReflectionResult, EngineError> {
+    let transcript = build_transcript(thread);
+
+    // Build the reflection-specific effect executor
+    let executor: Arc<dyn crate::traits::effect::EffectExecutor> =
+        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<dyn LlmBackend>,
 ) -> Result<ReflectionResult, EngineError> {
    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<MemoryDoc> {
+    // Try parsing as JSON first (the expected format)
+    if let Ok(value) = serde_json::from_str::<serde_json::Value>(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::<serde_json::Value>(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<MemoryDoc> {
+    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<DocType> = 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());
    }
 }
@@ -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<String>,
+}
+
+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<RwLock<HashMap<String, ActionMetrics>>>,
+}
+
+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<ActionMetrics> {
+        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());
+    }
+}
@@ -327,6 +327,10 @@ mod tests {
        async fn save_lease(&self, _: &CapabilityLease) -> Result<(), EngineError> { Ok(()) }
        async fn load_active_leases(&self, _: ThreadId) -> Result<Vec<CapabilityLease>, 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<Option<crate::types::mission::Mission>, EngineError> { Ok(None) }
+        async fn list_missions(&self, _: ProjectId) -> Result<Vec<crate::types::mission::Mission>, 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<ThreadManager>, ConversationManager) {
@@ -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<String> = 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<Vec<CapabilityLease>, 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<Option<crate::types::mission::Mission>, EngineError> { Ok(None) }
+        async fn list_missions(&self, _: ProjectId) -> Result<Vec<crate::types::mission::Mission>, 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<dyn LlmBackend>) -> ThreadManager {
@@ -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<dyn Store>,
+    thread_manager: Arc<ThreadManager>,
+    /// Active missions indexed by ID for quick lookup.
+    active: RwLock<Vec<MissionId>>,
+}
+
+impl MissionManager {
+    pub fn new(store: Arc<dyn Store>, thread_manager: Arc<ThreadManager>) -> 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<String>,
+        goal: impl Into<String>,
+        cadence: MissionCadence,
+    ) -> Result<MissionId, EngineError> {
+        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<Option<ThreadId>, 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<Vec<Mission>, EngineError> {
+        self.store.list_missions(project_id).await
+    }
+
+    /// Get a mission by ID.
+    pub async fn get_mission(&self, id: MissionId) -> Result<Option<Mission>, 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<Vec<ThreadId>, 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)
+    }
+}
@@ -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;
@@ -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<Vec<CapabilityLease>, 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<Option<Mission>, EngineError>;
+    async fn list_missions(&self, project_id: ProjectId) -> Result<Vec<Mission>, EngineError>;
+    async fn update_mission_status(
+        &self,
+        id: MissionId,
+        status: MissionStatus,
+    ) -> Result<(), EngineError>;
 }
@@ -121,4 +121,15 @@ pub enum EventKind {
        call_id: String,
        approved: bool,
    },
+
+    // ── Reflection ───────────────────────────────────────────
+    ReflectionStarted,
+    ReflectionComplete {
+        docs_produced: usize,
+        doc_types: Vec<String>,
+        tokens_used: u64,
+    },
+    ReflectionFailed {
+        error: String,
+    },
 }
@@ -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<ThreadId>,
+    /// Optional criteria for declaring the mission complete.
+    pub success_criteria: Option<String>,
+    pub metadata: serde_json::Value,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+    /// When the next thread should be spawned (for Cron cadence).
+    pub next_fire_at: Option<DateTime<Utc>>,
+}
+
+impl Mission {
+    pub fn new(
+        project_id: ProjectId,
+        name: impl Into<String>,
+        goal: impl Into<String>,
+        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<String>) -> 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
+        )
+    }
+}
@@ -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;
@@ -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 {
@@ -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
            },
        })
    }
@@ -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<HashMap<ProjectId, Project>>,
    docs: RwLock<HashMap<DocId, MemoryDoc>>,
    leases: RwLock<HashMap<LeaseId, CapabilityLease>>,
+    missions: RwLock<HashMap<MissionId, Mission>>,
 }

 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<Option<Mission>, EngineError> {
+        Ok(self.missions.read().await.get(&id).cloned())
+    }
+
+    async fn list_missions(&self, project_id: ProjectId) -> Result<Vec<Mission>, 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(())
+    }
 }