diff --git a/crates/ironclaw_engine/CLAUDE.md b/crates/ironclaw_engine/CLAUDE.md index a446458f..8f02c03e 100644 --- a/crates/ironclaw_engine/CLAUDE.md +++ b/crates/ironclaw_engine/CLAUDE.md @@ -67,9 +67,6 @@ src/ ├── memory/ # Memory document system │ ├── store.rs # MemoryStore — project-scoped doc CRUD │ └── retrieval.rs # RetrievalEngine — keyword-based context retrieval from project docs -├── reflection/ # Post-thread reflection pipeline -│ ├── pipeline.rs # reflect() (CodeAct) + reflect_simple() (direct LLM) + output parsing -│ └── executor.rs # ReflectionExecutor — read-only tools for reflection threads └── reliability.rs # ReliabilityTracker — per-action success rate and latency via EMA ``` @@ -78,12 +75,22 @@ src/ ``` Created → Running → Waiting → Running (resume) → Suspended → Running (resume) - → Completed → Reflecting → Done + → Completed → Done → Failed ``` Validated by `ThreadState::can_transition_to()`. Terminal states: `Done`, `Failed`. +## Learning Missions + +Three event-driven missions fire automatically after thread completion: + +1. **Error diagnosis** (`self-improvement`) — fires when a thread completes with trace issues. Diagnoses root cause and applies prompt overlays or orchestrator patches. +2. **Playbook extraction** (`playbook-extraction`) — fires when a thread succeeds with 5+ steps and 3+ tool actions. Extracts reusable step-by-step procedures. +3. **Conversation insights** (`conversation-insights`) — fires every 5 completed threads in a project. Extracts user preferences, domain knowledge, and workflow patterns. + +Created by `MissionManager::ensure_learning_missions()` at project bootstrap. + ## External Trait Boundaries The engine defines three traits that the host crate implements: diff --git a/crates/ironclaw_engine/orchestrator/default.py b/crates/ironclaw_engine/orchestrator/default.py index c54c06fd..c2ca5bf6 100644 --- a/crates/ironclaw_engine/orchestrator/default.py +++ b/crates/ironclaw_engine/orchestrator/default.py @@ -5,7 +5,7 @@ # by the self-improvement Mission. # # Host functions (provided by Rust via Monty suspension): -# __llm_complete__(messages, actions, config) -> response dict +# __llm_complete__(messages, actions, config) -> response dict (args ignored; Rust builds context from thread) # __execute_code_step__(code, state) -> result dict # __execute_action__(name, params) -> result dict # __check_signals__() -> None | "stop" | {"inject": msg} @@ -243,21 +243,20 @@ def run_loop(context, goal, actions, state, config): }) elif resp_type == "actions": - # Tier 0: structured tool calls + # Tier 0: structured tool calls. + # The assistant message with structured action_calls is added by + # __llm_complete__ in Rust — do NOT add it here. nudge_count = 0 calls = response.get("calls", []) - __add_message__("assistant_actions", str(calls)) for call in calls: name = call.get("name", "") params = call.get("params", {}) call_id = call.get("call_id", "") - r = __execute_action__(name, params) - - __emit_event__("action_executed" if not r.get("is_error") else "action_failed", - action_name=name, call_id=call_id) - __add_message__("action_result", str(r.get("output", {}))) + # __execute_action__ handles event emission, message addition, + # and lease consumption in Rust — no duplicate logic needed here. + r = __execute_action__(name, params, call_id=call_id) if r.get("need_approval"): __save_checkpoint__(state, { diff --git a/crates/ironclaw_engine/src/capability/planner.rs b/crates/ironclaw_engine/src/capability/planner.rs index 5db0def6..3fbbe3eb 100644 --- a/crates/ironclaw_engine/src/capability/planner.rs +++ b/crates/ironclaw_engine/src/capability/planner.rs @@ -23,18 +23,11 @@ impl LeasePlanner { } /// Build the capability grants for a new thread. - /// - /// Reflection threads are handled by the reflection pipeline's dedicated - /// executor, so the default planner grants no host capabilities to them. pub fn plan_for_thread( &self, - thread_type: ThreadType, + _thread_type: ThreadType, capabilities: &CapabilityRegistry, ) -> Vec { - if thread_type == ThreadType::Reflection { - return Vec::new(); - } - capabilities .list() .into_iter() @@ -89,10 +82,4 @@ mod tests { assert_eq!(plans[0].granted_actions, vec!["read_file"]); } - #[test] - fn reflection_threads_do_not_get_default_capabilities() { - let planner = LeasePlanner::new(); - let plans = planner.plan_for_thread(ThreadType::Reflection, ®istry()); - assert!(plans.is_empty()); - } } diff --git a/crates/ironclaw_engine/src/capability/policy.rs b/crates/ironclaw_engine/src/capability/policy.rs index 0c1dc5bf..6f3690bc 100644 --- a/crates/ironclaw_engine/src/capability/policy.rs +++ b/crates/ironclaw_engine/src/capability/policy.rs @@ -151,8 +151,8 @@ impl PolicyEngine { } // User and System provenance are trusted Provenance::User | Provenance::System => {} - // Reflection and MemoryRetrieval are internal, treat as trusted - Provenance::Reflection { .. } | Provenance::MemoryRetrieval { .. } => {} + // MemoryRetrieval is internal, treat as trusted + Provenance::MemoryRetrieval { .. } => {} } decision diff --git a/crates/ironclaw_engine/src/executor/loop_engine.rs b/crates/ironclaw_engine/src/executor/loop_engine.rs index 2ac8c10a..059acc13 100644 --- a/crates/ironclaw_engine/src/executor/loop_engine.rs +++ b/crates/ironclaw_engine/src/executor/loop_engine.rs @@ -1112,4 +1112,225 @@ mod tests { let text = "A very long explanation...\n\n🔚 Final Thought\n\nFINAL(\"the conclusion\")"; assert_eq!(extract_final_from_text(text).unwrap(), "the conclusion"); } + + // ── call_id propagation through orchestrator pipeline ──── + // + // These tests verify the end-to-end flow: LLM returns ActionCalls with + // call_ids → orchestrator executes them → ActionResult messages on the + // thread have correct call_ids (not empty). This catches the class of + // bugs that caused OpenAI/Codex HTTP 400 rejections. + + #[tokio::test] + async fn action_result_messages_have_correct_call_id() { + // LLM returns a tool call, then a text response + let (mut exec, _tx) = make_loop( + vec![ + action_response("test_tool", "call_xK9mZq123"), + text_response("Done!"), + ], + vec![Ok(ActionResult { + call_id: String::new(), // EffectExecutor returns empty + action_name: "test_tool".into(), + output: serde_json::json!({"data": "result"}), + is_error: false, + duration: Duration::from_millis(5), + })], + ThreadConfig::default(), + ) + .await; + + exec.run().await.unwrap(); + + // Find the ActionResult message on the thread + let action_results: Vec<_> = exec + .thread + .messages + .iter() + .filter(|m| m.role == crate::types::message::MessageRole::ActionResult) + .collect(); + + assert!( + !action_results.is_empty(), + "thread should have at least one ActionResult message" + ); + + for msg in &action_results { + let call_id = msg.action_call_id.as_deref().unwrap_or(""); + assert!( + !call_id.is_empty(), + "ActionResult message must have non-empty call_id, got empty for tool '{}'", + msg.action_name.as_deref().unwrap_or("?") + ); + } + } + + /// Verify that the ActionExecuted event carries the call_id from the LLM. + #[tokio::test] + async fn action_executed_events_carry_call_id() { + let (mut exec, _tx) = make_loop( + vec![ + action_response("test_tool", "call_evt_id_42"), + text_response("ok"), + ], + vec![Ok(ActionResult { + call_id: String::new(), + action_name: "test_tool".into(), + output: serde_json::json!({}), + is_error: false, + duration: Duration::from_millis(1), + })], + ThreadConfig::default(), + ) + .await; + + exec.run().await.unwrap(); + + let exec_events: Vec<_> = exec + .thread + .events + .iter() + .filter_map(|e| match &e.kind { + EventKind::ActionExecuted { call_id, .. } => Some(call_id.clone()), + _ => None, + }) + .collect(); + + assert!(!exec_events.is_empty(), "should have ActionExecuted events"); + for call_id in &exec_events { + assert!(!call_id.is_empty(), "ActionExecuted event must have non-empty call_id"); + } + } + + /// When a tool call fails (no lease), the ActionResult message and + /// ActionFailed event must still carry the original call_id. + #[tokio::test] + async fn failed_action_preserves_call_id_in_message_and_event() { + let project_id = ProjectId::new(); + let thread = Thread::new( + "test", + ThreadType::Foreground, + project_id, + ThreadConfig::default(), + ); + let tid = thread.id; + + // Create a tool that requires a separate capability + let missing_action = ActionDef { + name: "restricted_tool".into(), + description: "A tool with no lease".into(), + parameters_schema: serde_json::json!({"type": "object"}), + effects: vec![EffectType::WriteExternal], + requires_approval: false, + }; + + let llm = Arc::new(MockLlm::new(vec![ + // LLM calls a tool the thread has no lease for + LlmOutput { + response: LlmResponse::ActionCalls { + calls: vec![crate::types::step::ActionCall { + id: "call_nolease_xyz".into(), + action_name: "restricted_tool".into(), + parameters: serde_json::json!({}), + }], + content: None, + }, + usage: TokenUsage::default(), + }, + text_response("I couldn't access that tool"), + ])); + let effects = Arc::new(MockEffects::new(vec![missing_action], vec![])); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + + // Grant a lease that does NOT cover "restricted_tool" + leases + .grant(tid, "basic_cap", vec![], None, None) + .await; + + let (_tx, rx) = crate::runtime::messaging::signal_channel(16); + let mut exec = ExecutionLoop::new( + thread, + llm, + effects, + leases, + policy, + rx, + "test-user".into(), + ); + + exec.run().await.unwrap(); + + // Check ActionResult messages + let action_results: Vec<_> = exec + .thread + .messages + .iter() + .filter(|m| m.role == crate::types::message::MessageRole::ActionResult) + .collect(); + + for msg in &action_results { + let call_id = msg.action_call_id.as_deref().unwrap_or(""); + assert!( + !call_id.is_empty(), + "even failed ActionResult must have call_id" + ); + } + + // Check ActionFailed events + let fail_events: Vec<_> = exec + .thread + .events + .iter() + .filter_map(|e| match &e.kind { + EventKind::ActionFailed { + call_id, + action_name, + .. + } => Some((call_id.clone(), action_name.clone())), + _ => None, + }) + .collect(); + + for (call_id, _name) in &fail_events { + assert!( + !call_id.is_empty(), + "ActionFailed event must have call_id" + ); + } + } + + /// Verify the trace analyzer does NOT flag any issues on a clean + /// action execution (no empty call_ids). + #[tokio::test] + async fn trace_analysis_clean_after_successful_tool_use() { + let (mut exec, _tx) = make_loop( + vec![ + action_response("test_tool", "call_clean_id"), + text_response("All done"), + ], + vec![Ok(ActionResult { + call_id: String::new(), + action_name: "test_tool".into(), + output: serde_json::json!({"status": "ok"}), + is_error: false, + duration: Duration::from_millis(3), + })], + ThreadConfig::default(), + ) + .await; + + exec.run().await.unwrap(); + + let trace = crate::executor::trace::build_trace(&exec.thread); + let empty_id_issues: Vec<_> = trace + .issues + .iter() + .filter(|i| i.category == "empty_call_id") + .collect(); + + assert!( + empty_id_issues.is_empty(), + "clean execution should have no empty_call_id issues, got: {empty_id_issues:?}" + ); + } } diff --git a/crates/ironclaw_engine/src/executor/orchestrator.rs b/crates/ironclaw_engine/src/executor/orchestrator.rs index 6b4987e0..4cd207cb 100644 --- a/crates/ironclaw_engine/src/executor/orchestrator.rs +++ b/crates/ironclaw_engine/src/executor/orchestrator.rs @@ -338,9 +338,12 @@ pub async fn execute_orchestrator( .await } - // __execute_action__(name, params) + // __execute_action__(name, params, call_id=...) "__execute_action__" => { - handle_execute_action(args, kwargs, thread, effects, leases, policy).await + handle_execute_action( + args, kwargs, thread, effects, leases, policy, event_tx, + ) + .await } // __check_signals__() @@ -441,10 +444,14 @@ pub async fn execute_orchestrator( /// /// Calls the LLM and returns the response as a dict: /// `{type: "text"|"code"|"actions", content/code/calls: ..., usage: {...}}` +/// +/// For `ActionCalls` responses, the assistant message with structured action_calls +/// is added directly to the thread (not by Python) so the LLM backend can convert +/// them to the provider-specific tool_calls format on the next call. async fn handle_llm_complete( _args: &[MontyObject], _kwargs: &[(MontyObject, MontyObject)], - thread: &Thread, + thread: &mut Thread, llm: &Arc, effects: &Arc, leases: &Arc, @@ -485,7 +492,16 @@ async fn handle_llm_complete( LlmResponse::Code { code, .. } => { serde_json::json!({"type": "code", "code": code, "usage": usage}) } - LlmResponse::ActionCalls { calls, .. } => { + LlmResponse::ActionCalls { calls, content } => { + // Add the assistant message with structured action_calls so the + // LLM backend sees proper tool_calls on the next round-trip. + // Python must NOT call __add_message__("assistant_actions", ...) — + // the message is already on the thread. + thread.add_message(ThreadMessage::assistant_with_actions( + content, + calls.clone(), + )); + let calls_json: Vec = calls .iter() .map(|c| { @@ -602,14 +618,25 @@ async fn handle_execute_code_step( } } -/// Handle `__execute_action__(name, params)`. +/// Handle `__execute_action__(name, params, call_id=...)`. +/// +/// Single source of truth for action execution. Performs: +/// 1. Lease lookup +/// 2. Policy check +/// 3. Lease consumption +/// 4. Action execution via EffectExecutor +/// 5. Event emission (ActionExecuted/ActionFailed) +/// 6. Message addition (ActionResult with correct call_id) +/// +/// Python only needs to check the returned `need_approval` flag. async fn handle_execute_action( args: &[MontyObject], kwargs: &[(MontyObject, MontyObject)], - thread: &Thread, + thread: &mut Thread, effects: &Arc, leases: &Arc, policy: &Arc, + event_tx: Option<&tokio::sync::broadcast::Sender>, ) -> ExtFunctionResult { let name = match extract_string_arg(args, kwargs, "name", 0) { Some(n) => n, @@ -626,6 +653,8 @@ async fn handle_execute_action( .map(monty_to_json) .unwrap_or(serde_json::json!({})); + let call_id = extract_string_kwarg(kwargs, "call_id").unwrap_or_default(); + let exec_ctx = ThreadExecutionContext { thread_id: thread.id, thread_type: thread.thread_type, @@ -634,19 +663,50 @@ async fn handle_execute_action( step_id: StepId::new(), }; - // Find lease for this action + // Helper: emit event and add ActionResult message to thread + let emit_and_record = |thread: &mut Thread, + event_tx: Option<&tokio::sync::broadcast::Sender>, + event_kind: EventKind, + call_id: &str, + action_name: &str, + output: &serde_json::Value| { + let event = ThreadEvent::new(thread.id, event_kind); + if let Some(tx) = event_tx { + let _ = tx.send(event.clone()); + } + thread.events.push(event); + thread.updated_at = chrono::Utc::now(); + thread.add_message(ThreadMessage::action_result(call_id, action_name, output.to_string())); + }; + + // 1. Find lease for this action let lease = match leases.find_lease_for_action(thread.id, &name).await { Some(l) => l, None => { + let error = format!("No lease for action '{name}'"); + let output = serde_json::json!({"error": &error}); + emit_and_record( + thread, + event_tx, + EventKind::ActionFailed { + step_id: exec_ctx.step_id, + action_name: name.clone(), + call_id: call_id.clone(), + error, + }, + &call_id, + &name, + &output, + ); let result = serde_json::json!({ - "output": {"error": format!("No lease for action '{name}'")}, + "output": output, "is_error": true, }); return ExtFunctionResult::Return(json_to_monty(&result)); } }; - // Check policy + // 2. Check policy let action_def = effects .available_actions(std::slice::from_ref(&lease)) .await @@ -656,8 +716,22 @@ async fn handle_execute_action( if let Some(ref ad) = action_def { match policy.evaluate(ad, &lease, &[]) { crate::capability::policy::PolicyDecision::Deny { reason } => { + let output = serde_json::json!({"error": format!("Denied: {reason}")}); + emit_and_record( + thread, + event_tx, + EventKind::ActionFailed { + step_id: exec_ctx.step_id, + action_name: name.clone(), + call_id: call_id.clone(), + error: reason, + }, + &call_id, + &name, + &output, + ); let result = serde_json::json!({ - "output": {"error": format!("Denied: {reason}")}, + "output": output, "is_error": true, }); return ExtFunctionResult::Return(json_to_monty(&result)); @@ -673,12 +747,30 @@ async fn handle_execute_action( } } - // Execute + // 3. Consume a lease use + if let Err(e) = leases.consume_use(lease.id).await { + debug!(error = %e, "lease consumption failed (non-fatal)"); + } + + // 4. Execute match effects .execute_action(&name, params, &lease, &exec_ctx) .await { Ok(r) => { + emit_and_record( + thread, + event_tx, + EventKind::ActionExecuted { + step_id: exec_ctx.step_id, + action_name: name.clone(), + call_id: call_id.clone(), + duration_ms: r.duration.as_millis() as u64, + }, + &call_id, + &name, + &r.output, + ); let result = serde_json::json!({ "action_name": r.action_name, "output": r.output, @@ -688,8 +780,22 @@ async fn handle_execute_action( ExtFunctionResult::Return(json_to_monty(&result)) } Err(e) => { + let output = serde_json::json!({"error": e.to_string()}); + emit_and_record( + thread, + event_tx, + EventKind::ActionFailed { + step_id: exec_ctx.step_id, + action_name: name.clone(), + call_id: call_id.clone(), + error: e.to_string(), + }, + &call_id, + &name, + &output, + ); let result = serde_json::json!({ - "output": {"error": e.to_string()}, + "output": output, "is_error": true, }); ExtFunctionResult::Return(json_to_monty(&result)) @@ -784,6 +890,10 @@ fn handle_emit_event( } /// Handle `__add_message__(role, content)`. +/// +/// ActionResult messages are NOT added here — they are handled by +/// `__execute_action__` which is the single source of truth for +/// action execution, event emission, and message recording. fn handle_add_message( args: &[MontyObject], _kwargs: &[(MontyObject, MontyObject)], @@ -794,7 +904,7 @@ fn handle_add_message( match role.as_str() { "user" => thread.add_message(ThreadMessage::user(&content)), - "assistant" | "assistant_actions" => thread.add_message(ThreadMessage::assistant(&content)), + "assistant" => thread.add_message(ThreadMessage::assistant(&content)), "system" => thread.add_message(ThreadMessage::system(&content)), "system_append" => { // Append to existing system message (for doc injection) @@ -807,15 +917,11 @@ fn handle_add_message( msg.content.push_str(&content); } } - "action_result" => { - thread.add_message(ThreadMessage::action_result("", "", &content)); - } _ => { thread.add_message(ThreadMessage::user(&content)); } } - thread.step_count += 0; // Message addition tracked by thread itself ExtFunctionResult::Return(MontyObject::None) } diff --git a/crates/ironclaw_engine/src/executor/scripting.rs b/crates/ironclaw_engine/src/executor/scripting.rs index 72dccee1..5c89d96e 100644 --- a/crates/ironclaw_engine/src/executor/scripting.rs +++ b/crates/ironclaw_engine/src/executor/scripting.rs @@ -770,7 +770,6 @@ async fn handle_rlm_query( // Build child thread with inherited budget let child_config = crate::types::thread::ThreadConfig { max_iterations: parent_thread.config.max_iterations.min(20), // cap child iterations - enable_reflection: false, enable_tool_intent_nudge: false, max_tokens_total: parent_thread .config diff --git a/crates/ironclaw_engine/src/executor/structured.rs b/crates/ironclaw_engine/src/executor/structured.rs index f5e99eaa..0e554624 100644 --- a/crates/ironclaw_engine/src/executor/structured.rs +++ b/crates/ironclaw_engine/src/executor/structured.rs @@ -166,3 +166,322 @@ pub async fn execute_action_calls( need_approval: None, }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::traits::effect::ThreadExecutionContext; + use crate::types::capability::{ActionDef, CapabilityLease, EffectType}; + use crate::types::project::ProjectId; + use crate::types::step::StepId; + use crate::types::thread::{Thread, ThreadConfig, ThreadType}; + + use std::sync::Mutex; + use std::time::Duration; + + struct MockEffects { + results: Mutex>>, + actions: Vec, + } + + impl MockEffects { + fn new(actions: Vec, results: Vec>) -> Self { + Self { + results: Mutex::new(results), + actions, + } + } + } + + #[async_trait::async_trait] + impl EffectExecutor for MockEffects { + async fn execute_action( + &self, + _name: &str, + _params: serde_json::Value, + _lease: &CapabilityLease, + _ctx: &ThreadExecutionContext, + ) -> Result { + let mut results = self.results.lock().unwrap(); + if results.is_empty() { + Ok(ActionResult { + call_id: String::new(), // EffectExecutor doesn't set call_id + action_name: String::new(), + output: serde_json::json!({"result": "ok"}), + is_error: false, + duration: Duration::from_millis(1), + }) + } else { + results.remove(0) + } + } + + async fn available_actions( + &self, + _leases: &[CapabilityLease], + ) -> Result, EngineError> { + Ok(self.actions.clone()) + } + } + + fn test_action(name: &str) -> ActionDef { + ActionDef { + name: name.into(), + description: "Test tool".into(), + parameters_schema: serde_json::json!({"type": "object"}), + effects: vec![EffectType::ReadLocal], + requires_approval: false, + } + } + + fn make_exec_context(thread: &Thread) -> ThreadExecutionContext { + ThreadExecutionContext { + thread_id: thread.id, + thread_type: thread.thread_type, + project_id: thread.project_id, + user_id: "test".into(), + step_id: StepId::new(), + } + } + + // ── call_id propagation tests ──────────────────────────── + + #[tokio::test] + async fn call_id_preserved_on_successful_execution() { + let thread = Thread::new("test", ThreadType::Foreground, ProjectId::new(), ThreadConfig::default()); + let effects: Arc = Arc::new(MockEffects::new( + vec![test_action("web_search")], + vec![Ok(ActionResult { + call_id: String::new(), // EffectExecutor returns empty + action_name: "web_search".into(), + output: serde_json::json!({"results": []}), + is_error: false, + duration: Duration::from_millis(42), + })], + )); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let ctx = make_exec_context(&thread); + + leases.grant(thread.id, "search", vec![], None, None).await; + + let calls = vec![ActionCall { + id: "call_r2o5mqBgdNUlH8KzskncUGaX".into(), + action_name: "web_search".into(), + parameters: serde_json::json!({"query": "test"}), + }]; + + let result = execute_action_calls(&calls, &thread, &effects, &leases, &policy, &ctx, &[]) + .await + .unwrap(); + + // call_id must be stamped from ActionCall, not the empty EffectExecutor return + assert_eq!(result.results.len(), 1); + assert_eq!(result.results[0].call_id, "call_r2o5mqBgdNUlH8KzskncUGaX"); + assert_eq!(result.results[0].action_name, "web_search"); + assert!(!result.results[0].is_error); + + // Event should carry the same call_id + let exec_event = result.events.iter().find(|e| matches!(e, EventKind::ActionExecuted { .. })); + assert!(exec_event.is_some()); + if let Some(EventKind::ActionExecuted { call_id, action_name, .. }) = exec_event { + assert_eq!(call_id, "call_r2o5mqBgdNUlH8KzskncUGaX"); + assert_eq!(action_name, "web_search"); + } + } + + #[tokio::test] + async fn call_id_preserved_on_execution_error() { + let thread = Thread::new("test", ThreadType::Foreground, ProjectId::new(), ThreadConfig::default()); + let effects: Arc = Arc::new(MockEffects::new( + vec![test_action("shell")], + vec![Err(EngineError::Effect { + reason: "permission denied".into(), + })], + )); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let ctx = make_exec_context(&thread); + + leases.grant(thread.id, "exec", vec![], None, None).await; + + let calls = vec![ActionCall { + id: "call_abc123def".into(), + action_name: "shell".into(), + parameters: serde_json::json!({"cmd": "ls"}), + }]; + + let result = execute_action_calls(&calls, &thread, &effects, &leases, &policy, &ctx, &[]) + .await + .unwrap(); + + assert_eq!(result.results.len(), 1); + assert_eq!(result.results[0].call_id, "call_abc123def"); + assert!(result.results[0].is_error); + + let fail_event = result.events.iter().find(|e| matches!(e, EventKind::ActionFailed { .. })); + assert!(fail_event.is_some()); + if let Some(EventKind::ActionFailed { call_id, .. }) = fail_event { + assert_eq!(call_id, "call_abc123def"); + } + } + + #[tokio::test] + async fn call_id_preserved_when_no_lease() { + let thread = Thread::new("test", ThreadType::Foreground, ProjectId::new(), ThreadConfig::default()); + let effects: Arc = Arc::new(MockEffects::new(vec![], vec![])); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let ctx = make_exec_context(&thread); + + // No lease granted — action should fail with correct call_id + let calls = vec![ActionCall { + id: "call_no_lease_123".into(), + action_name: "web_search".into(), + parameters: serde_json::json!({}), + }]; + + let result = execute_action_calls(&calls, &thread, &effects, &leases, &policy, &ctx, &[]) + .await + .unwrap(); + + assert_eq!(result.results.len(), 1); + assert_eq!(result.results[0].call_id, "call_no_lease_123"); + assert!(result.results[0].is_error); + + if let Some(EventKind::ActionFailed { call_id, error, .. }) = result.events.first() { + assert_eq!(call_id, "call_no_lease_123"); + assert!(error.contains("no lease")); + } else { + panic!("expected ActionFailed event"); + } + } + + #[tokio::test] + async fn multiple_calls_each_get_correct_call_id() { + let thread = Thread::new("test", ThreadType::Foreground, ProjectId::new(), ThreadConfig::default()); + let effects: Arc = Arc::new(MockEffects::new( + vec![test_action("tool_a"), test_action("tool_b")], + vec![ + Ok(ActionResult { + call_id: String::new(), + action_name: "tool_a".into(), + output: serde_json::json!("a_result"), + is_error: false, + duration: Duration::from_millis(1), + }), + Ok(ActionResult { + call_id: String::new(), + action_name: "tool_b".into(), + output: serde_json::json!("b_result"), + is_error: false, + duration: Duration::from_millis(2), + }), + ], + )); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let ctx = make_exec_context(&thread); + + leases.grant(thread.id, "cap", vec![], None, None).await; + + let calls = vec![ + ActionCall { + id: "id_aaaa".into(), + action_name: "tool_a".into(), + parameters: serde_json::json!({}), + }, + ActionCall { + id: "id_bbbb".into(), + action_name: "tool_b".into(), + parameters: serde_json::json!({}), + }, + ]; + + let result = execute_action_calls(&calls, &thread, &effects, &leases, &policy, &ctx, &[]) + .await + .unwrap(); + + assert_eq!(result.results.len(), 2); + assert_eq!(result.results[0].call_id, "id_aaaa"); + assert_eq!(result.results[1].call_id, "id_bbbb"); + } + + /// Provider-specific: OpenAI rejects empty string call_id. Verify no result + /// ever has an empty call_id when the ActionCall provided one. + #[tokio::test] + async fn openai_empty_call_id_never_produced() { + let thread = Thread::new("test", ThreadType::Foreground, ProjectId::new(), ThreadConfig::default()); + let effects: Arc = Arc::new(MockEffects::new( + vec![test_action("echo")], + vec![Ok(ActionResult { + call_id: String::new(), // EffectExecutor always returns empty + action_name: String::new(), + output: serde_json::json!("hello"), + is_error: false, + duration: Duration::from_millis(1), + })], + )); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let ctx = make_exec_context(&thread); + + leases.grant(thread.id, "cap", vec![], None, None).await; + + let calls = vec![ActionCall { + id: "aB3xK9mZq".into(), // Mistral-compatible 9-char ID + action_name: "echo".into(), + parameters: serde_json::json!({}), + }]; + + let result = execute_action_calls(&calls, &thread, &effects, &leases, &policy, &ctx, &[]) + .await + .unwrap(); + + // Must NOT be empty — must be stamped from the ActionCall + assert!(!result.results[0].call_id.is_empty()); + assert_eq!(result.results[0].call_id, "aB3xK9mZq"); + } + + /// Mistral requires call_id matching [a-zA-Z0-9]{9}. + /// Verify the ID passes through unmodified (normalization is LLM-layer concern, + /// but engine must never lose it). + #[tokio::test] + async fn mistral_format_call_id_preserved() { + let thread = Thread::new("test", ThreadType::Foreground, ProjectId::new(), ThreadConfig::default()); + let effects: Arc = Arc::new(MockEffects::new( + vec![test_action("web_search")], + vec![Ok(ActionResult { + call_id: String::new(), + action_name: "web_search".into(), + output: serde_json::json!({}), + is_error: false, + duration: Duration::from_millis(1), + })], + )); + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + let ctx = make_exec_context(&thread); + + leases.grant(thread.id, "cap", vec![], None, None).await; + + // Mistral format: exactly 9 alphanumeric chars + let mistral_id = "xK3mR9bZq"; + let calls = vec![ActionCall { + id: mistral_id.into(), + action_name: "web_search".into(), + parameters: serde_json::json!({}), + }]; + + let result = execute_action_calls(&calls, &thread, &effects, &leases, &policy, &ctx, &[]) + .await + .unwrap(); + + assert_eq!(result.results[0].call_id, mistral_id); + + // Event also preserves the exact format + if let Some(EventKind::ActionExecuted { call_id, .. }) = result.events.first() { + assert_eq!(call_id, mistral_id); + } + } +} diff --git a/crates/ironclaw_engine/src/executor/trace.rs b/crates/ironclaw_engine/src/executor/trace.rs index c2d06310..01830c97 100644 --- a/crates/ironclaw_engine/src/executor/trace.rs +++ b/crates/ironclaw_engine/src/executor/trace.rs @@ -33,20 +33,12 @@ pub struct ExecutionTrace { pub messages: Vec, pub events: Vec, pub issues: Vec, - pub reflection: Option, pub timestamp: chrono::DateTime, } -/// Reflection results captured in the trace. +/// A single doc record, for the trace. #[derive(Debug, Serialize)] -pub struct ReflectionTrace { - pub docs: Vec, - pub tokens_used: u64, -} - -/// A single doc produced by reflection, for the trace. -#[derive(Debug, Serialize)] -pub struct ReflectionDocRecord { +pub struct DocRecord { pub doc_type: String, pub title: String, pub content: String, @@ -72,7 +64,7 @@ pub struct TraceIssue { pub step: Option, } -#[derive(Debug, Serialize)] +#[derive(Debug, PartialEq, Serialize)] pub enum IssueSeverity { Error, Warning, @@ -112,7 +104,6 @@ pub fn build_trace(thread: &Thread) -> ExecutionTrace { messages, events: thread.events.clone(), issues, - reflection: None, timestamp: Utc::now(), } } @@ -140,22 +131,6 @@ pub fn write_trace(trace: &ExecutionTrace) -> Option { } } -/// Attach reflection results to a trace. -pub fn attach_reflection(trace: &mut ExecutionTrace, result: &crate::reflection::ReflectionResult) { - trace.reflection = Some(ReflectionTrace { - docs: result - .docs - .iter() - .map(|d| ReflectionDocRecord { - doc_type: format!("{:?}", d.doc_type), - title: d.title.clone(), - content: d.content.clone(), - }) - .collect(), - tokens_used: result.tokens_used.total(), - }); -} - /// Print a summary of the trace to the log. pub fn log_trace_summary(trace: &ExecutionTrace) { debug!( @@ -192,28 +167,6 @@ pub fn log_trace_summary(trace: &ExecutionTrace) { ), } } - - if let Some(ref refl) = trace.reflection { - debug!( - thread_id = %trace.thread_id, - docs = refl.docs.len(), - tokens = refl.tokens_used, - "=== Reflection ===" - ); - for doc in &refl.docs { - let preview: String = doc.content.chars().take(200).collect(); - let truncated = if doc.content.chars().count() > 200 { - "..." - } else { - "" - }; - debug!( - doc_type = %doc.doc_type, - title = %doc.title, - " {preview}{truncated}" - ); - } - } } // ── Retrospective analysis ────────────────────────────────── @@ -429,3 +382,177 @@ fn truncate(s: &str, max_chars: usize) -> String { chars } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::types::event::EventKind; + use crate::types::message::ThreadMessage; + use crate::types::project::ProjectId; + use crate::types::step::StepId; + use crate::types::thread::{ThreadConfig, ThreadType}; + + fn make_thread() -> Thread { + Thread::new( + "test goal", + ThreadType::Foreground, + ProjectId::new(), + ThreadConfig::default(), + ) + } + + // ── empty_call_id detection (OpenAI / Codex rejection) ─── + + /// OpenAI and Codex reject ActionResult messages with empty call_id. + /// The trace analyzer must flag these as errors. + #[test] + fn detects_empty_call_id_on_action_result() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("calling tool")); + // Simulate the bug: empty call_id + thread.add_message(ThreadMessage::action_result("", "web_search", "result")); + + let issues = analyze_trace(&thread); + let empty_id_issues: Vec<_> = issues + .iter() + .filter(|i| i.category == "empty_call_id") + .collect(); + + assert_eq!(empty_id_issues.len(), 1); + assert_eq!(empty_id_issues[0].severity, IssueSeverity::Error); + assert!(empty_id_issues[0].description.contains("web_search")); + } + + /// ActionResult with None call_id should also be flagged. + #[test] + fn detects_none_call_id_on_action_result() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("calling tool")); + // Manually construct a message with None call_id + thread.add_message(ThreadMessage { + role: crate::types::message::MessageRole::ActionResult, + content: "result".into(), + provenance: crate::types::provenance::Provenance::ToolOutput { + action_name: "shell".into(), + }, + action_call_id: None, + action_name: Some("shell".into()), + action_calls: None, + timestamp: chrono::Utc::now(), + }); + + let issues = analyze_trace(&thread); + assert!(issues.iter().any(|i| i.category == "empty_call_id")); + } + + /// No false positive: valid call_id should not be flagged. + #[test] + fn no_false_positive_for_valid_call_id() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("calling tool")); + thread.add_message(ThreadMessage::action_result( + "call_abc123", + "web_search", + "result", + )); + + let issues = analyze_trace(&thread); + assert!( + !issues.iter().any(|i| i.category == "empty_call_id"), + "valid call_id should not be flagged" + ); + } + + // ── tool_error detection ───────────────────────────────── + + /// ActionFailed events should produce tool_error warnings. + #[test] + fn detects_tool_failures_in_events() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("ok")); + thread.events.push(ThreadEvent::new( + thread.id, + EventKind::ActionFailed { + step_id: StepId::new(), + action_name: "web_search".into(), + call_id: "call_123".into(), + error: "No lease for action 'web_search'".into(), + }, + )); + + let issues = analyze_trace(&thread); + let tool_errors: Vec<_> = issues + .iter() + .filter(|i| i.category == "tool_error") + .collect(); + assert_eq!(tool_errors.len(), 1); + assert!(tool_errors[0].description.contains("web_search")); + } + + // ── thread_failure detection ───────────────────────────── + + #[test] + fn detects_failed_thread_state() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("trying")); + thread.state = ThreadState::Failed; + + let issues = analyze_trace(&thread); + assert!(issues.iter().any(|i| i.category == "thread_failure")); + } + + // ── LLM error detection from StateChanged events ───────── + + /// Reproduces the exact pattern from the trace: OpenAI rejects empty call_id. + #[test] + fn detects_llm_error_from_state_changed() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("ok")); + thread.state = ThreadState::Failed; + thread.events.push(ThreadEvent::new( + thread.id, + EventKind::StateChanged { + from: ThreadState::Running, + to: ThreadState::Failed, + reason: Some( + "LLM error: Provider openai_codex request failed: HTTP 400 Bad Request: \ + Invalid 'input[5].call_id': empty string" + .into(), + ), + }, + )); + + let issues = analyze_trace(&thread); + assert!( + issues.iter().any(|i| i.category == "llm_error"), + "should detect LLM provider error in StateChanged reason" + ); + } + + // ── Multiple empty call_ids ────────────────────────────── + + /// Anthropic sends consecutive tool results merged into one User message. + /// If multiple ActionResults have empty call_ids, each must be flagged. + #[test] + fn flags_each_empty_call_id_separately() { + let mut thread = make_thread(); + thread.add_message(ThreadMessage::system("sys")); + thread.add_message(ThreadMessage::assistant("parallel calls")); + thread.add_message(ThreadMessage::action_result("", "tool_a", "result_a")); + thread.add_message(ThreadMessage::action_result("", "tool_b", "result_b")); + thread.add_message(ThreadMessage::action_result("call_ok", "tool_c", "result_c")); + + let issues = analyze_trace(&thread); + let empty_issues: Vec<_> = issues + .iter() + .filter(|i| i.category == "empty_call_id") + .collect(); + assert_eq!(empty_issues.len(), 2, "should flag exactly the 2 empty call_ids"); + } +} diff --git a/crates/ironclaw_engine/src/lib.rs b/crates/ironclaw_engine/src/lib.rs index d0297bd2..1c17d281 100644 --- a/crates/ironclaw_engine/src/lib.rs +++ b/crates/ironclaw_engine/src/lib.rs @@ -17,7 +17,6 @@ pub mod capability; pub mod executor; pub mod memory; -pub mod reflection; pub mod reliability; pub mod runtime; pub mod traits; @@ -75,10 +74,6 @@ pub use executor::ExecutionLoop; pub use memory::MemoryStore; 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/reflection/executor.rs b/crates/ironclaw_engine/src/reflection/executor.rs deleted file mode 100644 index c4ab5951..00000000 --- a/crates/ironclaw_engine/src/reflection/executor.rs +++ /dev/null @@ -1,651 +0,0 @@ -//! 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"#; - -#[cfg(test)] -mod tests { - use std::sync::Arc; - - use chrono::Utc; - - use crate::capability::registry::CapabilityRegistry; - use crate::traits::effect::{EffectExecutor, ThreadExecutionContext}; - use crate::traits::store::Store; - use crate::types::capability::{ActionDef, Capability, CapabilityLease, EffectType, LeaseId}; - use crate::types::error::EngineError; - use crate::types::event::ThreadEvent; - use crate::types::memory::{DocId, DocType, MemoryDoc}; - use crate::types::mission::{Mission, MissionId, MissionStatus}; - use crate::types::project::{Project, ProjectId}; - use crate::types::step::{Step, StepId}; - use crate::types::thread::{Thread, ThreadId, ThreadState, ThreadType}; - - use super::{ReflectionExecutor, build_reflection_prompt}; - - // ── MockStore ────────────────────────────────────────────── - - struct MockStore { - docs: tokio::sync::Mutex>, - } - - impl MockStore { - fn new(docs: Vec) -> Arc { - Arc::new(Self { - docs: tokio::sync::Mutex::new(docs), - }) - } - - fn empty() -> Arc { - Self::new(vec![]) - } - } - - #[async_trait::async_trait] - impl Store for MockStore { - async fn save_thread(&self, _: &Thread) -> Result<(), EngineError> { - Ok(()) - } - async fn load_thread(&self, _: ThreadId) -> Result, EngineError> { - Ok(None) - } - async fn list_threads(&self, _: ProjectId) -> Result, EngineError> { - Ok(vec![]) - } - async fn update_thread_state( - &self, - _: ThreadId, - _: ThreadState, - ) -> Result<(), EngineError> { - Ok(()) - } - async fn save_step(&self, _: &Step) -> Result<(), EngineError> { - Ok(()) - } - async fn load_steps(&self, _: ThreadId) -> Result, EngineError> { - Ok(vec![]) - } - async fn append_events(&self, _: &[ThreadEvent]) -> Result<(), EngineError> { - Ok(()) - } - async fn load_events(&self, _: ThreadId) -> Result, EngineError> { - Ok(vec![]) - } - async fn save_project(&self, _: &Project) -> Result<(), EngineError> { - Ok(()) - } - async fn load_project(&self, _: ProjectId) -> Result, EngineError> { - Ok(None) - } - async fn save_memory_doc(&self, _: &MemoryDoc) -> Result<(), EngineError> { - Ok(()) - } - async fn load_memory_doc(&self, _: DocId) -> Result, EngineError> { - Ok(None) - } - async fn list_memory_docs( - &self, - project_id: ProjectId, - ) -> Result, EngineError> { - let docs = self.docs.lock().await; - Ok(docs - .iter() - .filter(|d| d.project_id == project_id) - .cloned() - .collect()) - } - 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, _: &Mission) -> Result<(), EngineError> { - Ok(()) - } - async fn load_mission(&self, _: MissionId) -> Result, EngineError> { - Ok(None) - } - async fn list_missions(&self, _: ProjectId) -> Result, EngineError> { - Ok(vec![]) - } - async fn update_mission_status( - &self, - _: MissionId, - _: MissionStatus, - ) -> Result<(), EngineError> { - Ok(()) - } - } - - // ── Helpers ──────────────────────────────────────────────── - - fn make_lease() -> CapabilityLease { - CapabilityLease { - id: LeaseId::new(), - thread_id: ThreadId::new(), - capability_name: "test".into(), - granted_actions: vec![], - granted_at: Utc::now(), - expires_at: None, - max_uses: None, - uses_remaining: None, - revoked: false, - } - } - - fn make_ctx() -> ThreadExecutionContext { - ThreadExecutionContext { - thread_id: ThreadId::new(), - thread_type: ThreadType::Reflection, - project_id: ProjectId::new(), - user_id: "test".into(), - step_id: StepId::new(), - } - } - - fn make_capability(name: &str, actions: Vec) -> Capability { - Capability { - name: name.into(), - description: format!("{name} capability"), - actions, - knowledge: vec![], - policies: vec![], - } - } - - fn make_action_def(name: &str) -> ActionDef { - ActionDef { - name: name.into(), - description: format!("{name} action"), - parameters_schema: serde_json::json!({"type": "object", "properties": {}}), - effects: vec![EffectType::ReadLocal], - requires_approval: false, - } - } - - // ── Tests ────────────────────────────────────────────────── - - #[tokio::test] - async fn get_transcript_returns_content() { - let transcript = "Step 1: called web_search\nStep 2: got results\nDone."; - let project_id = ProjectId::new(); - let executor = ReflectionExecutor::new( - MockStore::empty(), - Arc::new(CapabilityRegistry::new()), - transcript.to_string(), - project_id, - ); - - let lease = make_lease(); - let ctx = make_ctx(); - let result = executor - .execute_action("get_transcript", serde_json::json!({}), &lease, &ctx) - .await - .unwrap(); - - assert_eq!(result.action_name, "get_transcript"); - assert!(!result.is_error); - assert_eq!(result.output["transcript"].as_str().unwrap(), transcript); - } - - #[tokio::test] - async fn query_memory_finds_docs() { - let project_id = ProjectId::new(); - let docs = vec![ - MemoryDoc::new( - project_id, - DocType::Lesson, - "deployment error", - "Fix: restart the service", - ), - MemoryDoc::new( - project_id, - DocType::Summary, - "weather check", - "Fetched weather data", - ), - ]; - let store = MockStore::new(docs); - let executor = ReflectionExecutor::new( - store, - Arc::new(CapabilityRegistry::new()), - String::new(), - project_id, - ); - - let lease = make_lease(); - let ctx = make_ctx(); - let result = executor - .execute_action( - "query_memory", - serde_json::json!({"query": "deployment error", "max_docs": 5}), - &lease, - &ctx, - ) - .await - .unwrap(); - - assert!(!result.is_error); - let count = result.output["count"].as_u64().unwrap(); - assert!(count >= 1, "expected at least 1 doc, got {count}"); - - let docs_arr = result.output["docs"].as_array().unwrap(); - // The deployment error doc should be present - let has_deployment = docs_arr - .iter() - .any(|d| d["title"].as_str().unwrap().contains("deployment")); - assert!(has_deployment, "expected deployment doc in results"); - } - - #[tokio::test] - async fn check_tool_exists_found() { - let mut registry = CapabilityRegistry::new(); - registry.register(make_capability( - "search", - vec![make_action_def("web-search")], - )); - - let project_id = ProjectId::new(); - let executor = ReflectionExecutor::new( - MockStore::empty(), - Arc::new(registry), - String::new(), - project_id, - ); - - let lease = make_lease(); - let ctx = make_ctx(); - let result = executor - .execute_action( - "check_tool_exists", - serde_json::json!({"name": "web-search"}), - &lease, - &ctx, - ) - .await - .unwrap(); - - assert!(!result.is_error); - assert_eq!(result.output["exists"], true); - assert!(result.output["similar"].as_array().unwrap().is_empty()); - } - - #[tokio::test] - async fn check_tool_exists_not_found_suggests_similar() { - let mut registry = CapabilityRegistry::new(); - registry.register(make_capability( - "search", - vec![make_action_def("web-search")], - )); - - let project_id = ProjectId::new(); - let executor = ReflectionExecutor::new( - MockStore::empty(), - Arc::new(registry), - String::new(), - project_id, - ); - - let lease = make_lease(); - let ctx = make_ctx(); - let result = executor - .execute_action( - "check_tool_exists", - serde_json::json!({"name": "web_search"}), - &lease, - &ctx, - ) - .await - .unwrap(); - - assert!(!result.is_error); - assert_eq!(result.output["exists"], false); - let similar: Vec = result.output["similar"] - .as_array() - .unwrap() - .iter() - .map(|v| v.as_str().unwrap().to_string()) - .collect(); - assert!( - similar.contains(&"web-search".to_string()), - "expected 'web-search' in similar list, got: {similar:?}" - ); - } - - #[tokio::test] - async fn list_tools_returns_all() { - let mut registry = CapabilityRegistry::new(); - registry.register(make_capability( - "search", - vec![ - make_action_def("web-search"), - make_action_def("memory-search"), - ], - )); - registry.register(make_capability("files", vec![make_action_def("read-file")])); - - let project_id = ProjectId::new(); - let executor = ReflectionExecutor::new( - MockStore::empty(), - Arc::new(registry), - String::new(), - project_id, - ); - - let lease = make_lease(); - let ctx = make_ctx(); - let result = executor - .execute_action("list_tools", serde_json::json!({}), &lease, &ctx) - .await - .unwrap(); - - assert!(!result.is_error); - assert_eq!(result.output["count"].as_u64().unwrap(), 3); - let tools = result.output["tools"].as_array().unwrap(); - let names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect(); - assert!(names.contains(&"web-search")); - assert!(names.contains(&"memory-search")); - assert!(names.contains(&"read-file")); - } - - #[test] - fn build_reflection_prompt_includes_tools() { - let actions = vec![ - ActionDef { - name: "get_transcript".into(), - description: "Get the execution transcript".into(), - parameters_schema: serde_json::json!({"type": "object", "properties": {}}), - effects: vec![EffectType::ReadLocal], - requires_approval: false, - }, - ActionDef { - name: "query_memory".into(), - description: "Search memory docs".into(), - parameters_schema: serde_json::json!({ - "type": "object", - "properties": { - "query": {"type": "string"}, - "max_docs": {"type": "integer"} - } - }), - effects: vec![EffectType::ReadLocal], - requires_approval: false, - }, - ]; - - let prompt = build_reflection_prompt(&actions, "analyze deployment failure"); - assert!( - prompt.contains("get_transcript"), - "prompt should contain get_transcript tool name" - ); - assert!( - prompt.contains("query_memory"), - "prompt should contain query_memory tool name" - ); - assert!( - prompt.contains("analyze deployment failure"), - "prompt should contain the thread goal" - ); - assert!( - prompt.contains("Available tools"), - "prompt should contain the tools section header" - ); - } -} diff --git a/crates/ironclaw_engine/src/reflection/mod.rs b/crates/ironclaw_engine/src/reflection/mod.rs deleted file mode 100644 index e32114c5..00000000 --- a/crates/ironclaw_engine/src/reflection/mod.rs +++ /dev/null @@ -1,18 +0,0 @@ -//! Post-thread reflection pipeline. -//! -//! 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::{ReflectionResult, reflect, reflect_simple}; diff --git a/crates/ironclaw_engine/src/reflection/pipeline.rs b/crates/ironclaw_engine/src/reflection/pipeline.rs deleted file mode 100644 index 8866010a..00000000 --- a/crates/ironclaw_engine/src/reflection/pipeline.rs +++ /dev/null @@ -1,682 +0,0 @@ -//! Reflection pipeline — produces structured knowledge from completed threads. -//! -//! 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. -//! -//! 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, warn}; - -use crate::capability::lease::LeaseManager; -use crate::capability::policy::PolicyEngine; -use crate::capability::registry::CapabilityRegistry; -use crate::executor::ExecutionLoop; -use crate::reflection::executor::{ReflectionExecutor, build_reflection_prompt}; -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::TokenUsage; -use crate::types::thread::{Thread, ThreadConfig, ThreadType}; - -/// Result of running the reflection pipeline on a completed thread. -pub struct ReflectionResult { - /// Memory docs produced by reflection. - pub docs: Vec, - /// Total tokens used by reflection LLM calls. - pub tokens_used: TokenUsage, -} - -/// Run the reflection pipeline on a completed thread. -/// -/// 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); - store.save_thread(&refl_thread).await?; - store.save_lease(&lease).await?; - - // Run the execution loop - let mut exec_loop = ExecutionLoop::new( - refl_thread, - Arc::clone(llm), - executor, - lease_manager, - policy, - signal_rx, - "system".to_string(), - ) - .with_store(Arc::clone(store)); - - 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(); - let transcript = build_transcript(thread); - - // 1. Summary doc - let (summary_doc, tokens) = - produce_doc(thread, llm, DocType::Summary, &transcript, SUMMARY_PROMPT).await?; - docs.push(summary_doc); - total_tokens.input_tokens += tokens.input_tokens; - total_tokens.output_tokens += tokens.output_tokens; - - // 2. Lessons (only if there were errors) - let had_errors = thread.events.iter().any(|e| { - matches!( - e.kind, - EventKind::ActionFailed { .. } | EventKind::StepFailed { .. } - ) - }); - if had_errors { - let (lesson_doc, tokens) = - produce_doc(thread, llm, DocType::Lesson, &transcript, LESSON_PROMPT).await?; - docs.push(lesson_doc); - total_tokens.input_tokens += tokens.input_tokens; - total_tokens.output_tokens += tokens.output_tokens; - } - - // 3. Issues (if thread failed or had unresolved problems) - let thread_failed = thread.state == crate::types::thread::ThreadState::Failed; - if thread_failed || had_errors { - let (issue_doc, tokens) = - produce_doc(thread, llm, DocType::Issue, &transcript, ISSUE_PROMPT).await?; - if issue_doc.content.chars().count() > 20 { - docs.push(issue_doc); - } - total_tokens.input_tokens += tokens.input_tokens; - total_tokens.output_tokens += tokens.output_tokens; - } - - // 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") - } else { - false - } - }); - if has_missing_tools { - let (spec_doc, tokens) = - produce_doc(thread, llm, DocType::Spec, &transcript, SPEC_PROMPT).await?; - if spec_doc.content.chars().count() > 20 { - docs.push(spec_doc); - } - total_tokens.input_tokens += tokens.input_tokens; - total_tokens.output_tokens += tokens.output_tokens; - } - - // 5. Playbook - let action_count = thread - .events - .iter() - .filter(|e| matches!(e.kind, EventKind::ActionExecuted { .. })) - .count(); - let thread_succeeded = - thread.state == crate::types::thread::ThreadState::Completed && !thread_failed; - if thread_succeeded && action_count >= 2 { - let (playbook_doc, tokens) = - produce_doc(thread, llm, DocType::Playbook, &transcript, PLAYBOOK_PROMPT).await?; - if playbook_doc.content.chars().count() > 20 { - docs.push(playbook_doc); - } - total_tokens.input_tokens += tokens.input_tokens; - total_tokens.output_tokens += tokens.output_tokens; - } - - debug!( - thread_id = %thread.id, - docs_produced = docs.len(), - total_tokens = total_tokens.total(), - "reflection complete (simple)" - ); - - Ok(ReflectionResult { - docs, - tokens_used: total_tokens, - }) -} - -// ── 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.chars().count() > 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.chars().count() <= 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: -- The goal and whether it was achieved -- Key results or outputs -- Tools/actions that were used -Be factual and concise."; - -const LESSON_PROMPT: &str = "\ -Extract lessons learned from this thread's execution. Focus on: -- Errors encountered and how they were resolved (or not) -- Workarounds that were discovered -- Surprising findings about tool behavior -- Patterns that could be reused in similar tasks -Write each lesson as a single clear sentence. If there are no meaningful lessons, write 'No lessons.'."; - -const ISSUE_PROMPT: &str = "\ -Identify any unresolved issues from this thread. Focus on: -- Errors that were not resolved -- Tasks that could not be completed -- Missing tools or capabilities that were needed -- Data quality issues encountered -If there are no unresolved issues, write 'No issues.'."; - -const SPEC_PROMPT: &str = "\ -This thread encountered missing tools or capabilities. Analyze the errors and identify: -- Which tool names were attempted but not found -- What the correct tool name might be (if a similar tool exists under a different name) -- What capabilities would need to be added to handle this task -For each missing capability, write one line: MISSING: -> . -If the tool exists under a different name, write: ALIAS: -> ."; - -const PLAYBOOK_PROMPT: &str = "\ -This thread successfully completed a multi-step task. Extract a reusable playbook: -- List the steps taken in order (tool calls, queries, transformations) -- Note which tools were used and in what sequence -- 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 ─────────────────────────────────────────────────── - -/// Build a concise transcript of the thread's work. -pub(crate) fn build_transcript(thread: &Thread) -> String { - let mut parts = Vec::new(); - - parts.push(format!("Goal: {}", thread.goal)); - parts.push(format!("Steps: {}", thread.step_count)); - parts.push(format!("Tokens used: {}", thread.total_tokens_used)); - parts.push(format!("State: {:?}", thread.state)); - - // Include messages (truncated for very long threads) - let max_messages = 30; - let messages = if thread.messages.len() > max_messages { - &thread.messages[thread.messages.len() - max_messages..] - } else { - &thread.messages - }; - - parts.push("\n--- Messages ---".into()); - for msg in messages { - let role = format!("{:?}", msg.role); - let content_preview: String = msg.content.chars().take(500).collect(); - let truncated = if msg.content.chars().count() > 500 { - "..." - } else { - "" - }; - parts.push(format!("[{role}] {content_preview}{truncated}")); - } - - // Include notable events - let error_events: Vec = thread - .events - .iter() - .filter_map(|e| match &e.kind { - EventKind::ActionFailed { - action_name, error, .. - } => Some(format!("Action '{action_name}' failed: {error}")), - EventKind::StepFailed { error, .. } => Some(format!("Step failed: {error}")), - _ => None, - }) - .collect(); - - if !error_events.is_empty() { - parts.push("\n--- Errors ---".into()); - for err in error_events { - parts.push(err); - } - } - - parts.join("\n") -} - -/// Produce a single MemoryDoc by asking the LLM to analyze the transcript. -async fn produce_doc( - thread: &Thread, - llm: &Arc, - doc_type: DocType, - transcript: &str, - prompt: &str, -) -> Result<(MemoryDoc, TokenUsage), EngineError> { - let messages = vec![ - ThreadMessage::system(format!( - "You are analyzing a completed agent thread. Here is the transcript:\n\n{transcript}" - )), - ThreadMessage::user(prompt.to_string()), - ]; - - let config = crate::traits::llm::LlmCallConfig { - force_text: true, - ..crate::traits::llm::LlmCallConfig::default() - }; - - let output = llm.complete(&messages, &[], &config).await?; - - let content = match output.response { - 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 { - DocType::Summary => format!("Summary: {}", thread.goal), - DocType::Lesson => format!("Lessons: {}", thread.goal), - DocType::Issue => format!("Issues: {}", thread.goal), - DocType::Playbook => format!("Playbook: {}", thread.goal), - DocType::Spec => format!("Spec: {}", thread.goal), - DocType::Note => format!("Note: {}", thread.goal), - }; - - let doc = - MemoryDoc::new(thread.project_id, doc_type, title, content).with_source_thread(thread.id); - - Ok((doc, output.usage)) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::traits::llm::{LlmCallConfig, LlmOutput}; - use crate::types::capability::ActionDef; - use crate::types::event::ThreadEvent; - use crate::types::project::ProjectId; - use crate::types::step::TokenUsage; - use crate::types::thread::ThreadConfig; - use std::sync::Mutex; - - struct MockLlm { - responses: Mutex>, - } - - impl MockLlm { - fn with_responses(responses: Vec<&str>) -> Arc { - Arc::new(Self { - responses: Mutex::new(responses.into_iter().map(String::from).collect()), - }) - } - } - - #[async_trait::async_trait] - impl crate::traits::llm::LlmBackend for MockLlm { - async fn complete( - &self, - _: &[ThreadMessage], - _: &[ActionDef], - _: &LlmCallConfig, - ) -> Result { - let mut r = self.responses.lock().unwrap(); - let text = if r.is_empty() { - "mock response".to_string() - } else { - r.remove(0) - }; - Ok(LlmOutput { - response: crate::types::step::LlmResponse::Text(text), - usage: TokenUsage { - input_tokens: 100, - output_tokens: 50, - ..TokenUsage::default() - }, - }) - } - - fn model_name(&self) -> &str { - "mock" - } - } - - fn make_completed_thread() -> Thread { - let mut thread = Thread::new( - "test task", - ThreadType::Foreground, - ProjectId::new(), - ThreadConfig::default(), - ); - thread.state = crate::types::thread::ThreadState::Completed; - thread - } - - // ── reflect_simple tests (direct LLM calls) ──────────────── - - #[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_simple_produces_lesson_on_errors() { - let mut thread = make_completed_thread(); - thread.events.push(ThreadEvent::new( - thread.id, - EventKind::ActionFailed { - step_id: crate::types::step::StepId::new(), - action_name: "web_search".into(), - call_id: String::new(), - error: "Tool web_search not found".into(), - }, - )); - - let llm = MockLlm::with_responses(vec![ - "Summary of thread with errors.", - "Lesson: use web-search instead of web_search.", - "Issue: web_search tool is missing.", - "ALIAS: web_search -> web-search", - ]); - - 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)); - assert!(types.contains(&DocType::Issue)); - assert!(types.contains(&DocType::Spec)); - } - - #[tokio::test] - async fn reflect_simple_produces_spec_on_tool_not_found() { - let mut thread = make_completed_thread(); - thread.events.push(ThreadEvent::new( - thread.id, - EventKind::ActionFailed { - step_id: crate::types::step::StepId::new(), - action_name: "missing_tool".into(), - call_id: String::new(), - error: "Tool missing_tool not found".into(), - }, - )); - - let llm = MockLlm::with_responses(vec![ - "Summary.", - "Lesson learned.", - "Issues found.", - "MISSING: missing_tool -> needs implementation", - ]); - - let result = reflect_simple(&thread, &llm).await.unwrap(); - let spec_docs: Vec<&MemoryDoc> = result - .docs - .iter() - .filter(|d| d.doc_type == DocType::Spec) - .collect(); - assert_eq!(spec_docs.len(), 1); - assert!(spec_docs[0].content.contains("MISSING")); - } - - #[tokio::test] - async fn reflect_simple_produces_playbook_on_multi_step() { - let mut thread = make_completed_thread(); - thread.events.push(ThreadEvent::new( - thread.id, - EventKind::ActionExecuted { - step_id: crate::types::step::StepId::new(), - action_name: "web-search".into(), - call_id: String::new(), - duration_ms: 100, - }, - )); - thread.events.push(ThreadEvent::new( - thread.id, - EventKind::ActionExecuted { - step_id: crate::types::step::StepId::new(), - action_name: "llm_query".into(), - call_id: String::new(), - duration_ms: 200, - }, - )); - - let llm = MockLlm::with_responses(vec![ - "Summary of successful thread.", - "1. Search web\n2. Analyze results\n3. Return summary", - ]); - - let result = reflect_simple(&thread, &llm).await.unwrap(); - assert!(result.docs.iter().any(|d| d.doc_type == DocType::Playbook)); - } - - #[tokio::test] - async fn reflect_simple_skips_playbook_for_single_action() { - let mut thread = make_completed_thread(); - thread.events.push(ThreadEvent::new( - thread.id, - EventKind::ActionExecuted { - step_id: crate::types::step::StepId::new(), - action_name: "echo".into(), - call_id: String::new(), - duration_ms: 5, - }, - )); - - let llm = MockLlm::with_responses(vec!["Simple summary."]); - - let result = reflect_simple(&thread, &llm).await.unwrap(); - assert!(!result.docs.iter().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/runtime/manager.rs b/crates/ironclaw_engine/src/runtime/manager.rs index 89f56e91..4386bcd9 100644 --- a/crates/ironclaw_engine/src/runtime/manager.rs +++ b/crates/ironclaw_engine/src/runtime/manager.rs @@ -245,9 +245,6 @@ 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 running = Arc::clone(&self.running); let completed = Arc::clone(&self.completed); let handle = tokio::spawn(async move { @@ -255,98 +252,24 @@ impl ThreadManager { 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 mut trace = crate::executor::trace::build_trace(&exec.thread); + // Run retrospective trace analysis (non-LLM, always runs). + // Issues are picked up by the self-improvement mission via event listener. + let trace = crate::executor::trace::build_trace(&exec.thread); if !trace.issues.is_empty() { crate::executor::trace::log_trace_summary(&trace); } - // Run LLM reflection if enabled and thread completed - if exec.thread.config.enable_reflection - && exec.thread.state == crate::types::thread::ThreadState::Completed + // Transition Completed → Done + if exec.thread.state == crate::types::thread::ThreadState::Completed + && let Err(e) = exec.thread.transition_to( + crate::types::thread::ThreadState::Done, + None, + ) { - // 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 { - if let Err(e) = store_for_task.save_memory_doc(doc).await { - tracing::warn!( - thread_id = %thread_id, - doc_title = %doc.title, - "failed to save reflection doc: {e}" - ); - } - } - } - Err(e) => { - emit( - &mut exec.thread, - crate::types::event::EventKind::ReflectionFailed { - error: e.to_string(), - }, - ); - } - } - - // Transition: Reflecting → Done - if let Err(e) = exec.thread.transition_to( - crate::types::thread::ThreadState::Done, - Some("reflection finished".into()), - ) { - tracing::warn!( - thread_id = %thread_id, - "failed to transition to Done after reflection: {e}" - ); - } - } + tracing::warn!(thread_id = %thread_id, "failed to transition to Done: {e}"); } - // Write trace file if enabled (after reflection, so it's included) + // Write trace file if enabled if crate::executor::trace::is_trace_enabled() { crate::executor::trace::log_trace_summary(&trace); crate::executor::trace::write_trace(&trace); diff --git a/crates/ironclaw_engine/src/runtime/mission.rs b/crates/ironclaw_engine/src/runtime/mission.rs index c9bd1899..36bbf725 100644 --- a/crates/ironclaw_engine/src/runtime/mission.rs +++ b/crates/ironclaw_engine/src/runtime/mission.rs @@ -146,10 +146,7 @@ impl MissionManager { &meta_prompt, ThreadType::Mission, mission.project_id, - ThreadConfig { - enable_reflection: true, - ..ThreadConfig::default() - }, + ThreadConfig::default(), None, user_id, ) @@ -270,51 +267,62 @@ impl MissionManager { Ok(spawned) } - /// Start a background event listener that fires `OnSystemEvent` missions - /// when threads complete with issues. + /// Start a background event listener that fires learning missions when + /// threads complete. /// /// Subscribes to the ThreadManager's event broadcast channel and watches - /// for thread completion events. When a non-Mission, non-Reflection thread - /// completes and its trace has issues, fires matching OnSystemEvent missions - /// with trace data as the trigger payload. + /// for `StateChanged { to: Done }`. For each completed non-Mission thread: + /// + /// 1. **Error diagnosis** — if trace has issues, fires `thread_completed_with_issues` + /// 2. **Playbook extraction** — if thread succeeded with many steps/actions, + /// fires `thread_completed_with_learnings` + /// 3. **Conversation insights** — after every N threads in a conversation, + /// fires `conversation_insights_due` pub fn start_event_listener(self: &Arc, user_id: String) { let mgr = Arc::clone(self); let mut rx = mgr.thread_manager.subscribe_events(); + + /// Minimum steps for a thread to be a playbook candidate. + const PLAYBOOK_MIN_STEPS: usize = 5; + /// Minimum distinct action executions for playbook extraction. + const PLAYBOOK_MIN_ACTIONS: usize = 3; + /// Completed thread interval for conversation insights. + const CONVERSATION_INSIGHTS_INTERVAL: u32 = 5; + tokio::spawn(async move { + // Track completed thread count per conversation for insights trigger. + let mut conv_thread_counts: std::collections::HashMap = + std::collections::HashMap::new(); + loop { match rx.recv().await { Ok(event) => { - // React to ReflectionComplete — the thread is done and - // we have reflection doc info for the trigger payload. - if let crate::types::event::EventKind::ReflectionComplete { - docs_produced, - ref doc_types, - .. - } = event.kind - { - // Load the thread to check its type and build the payload - let thread = mgr.store.load_thread(event.thread_id).await; - let thread = match thread { - Ok(Some(t)) => t, - _ => continue, - }; - - // Skip Mission and Reflection threads (no recursive self-improvement) - if matches!( - thread.thread_type, - ThreadType::Mission | ThreadType::Reflection - ) { - continue; + // Only react to threads transitioning to Done + let is_done = matches!( + event.kind, + crate::types::event::EventKind::StateChanged { + to: crate::types::thread::ThreadState::Done, + .. } + ); + if !is_done { + continue; + } - // Build trace to check for issues - let trace = crate::executor::trace::build_trace(&thread); - if trace.issues.is_empty() { - continue; - } + // Load the completed thread + let thread = match mgr.store.load_thread(event.thread_id).await { + Ok(Some(t)) => t, + _ => continue, + }; - // Build trigger payload with trace issues, error messages, - // and reflection summary + // Skip Mission threads (no recursive self-improvement) + if thread.thread_type == ThreadType::Mission { + continue; + } + + // ── Trigger 1: Error diagnosis ────────────────── + let trace = crate::executor::trace::build_trace(&thread); + if !trace.issues.is_empty() { let issues: Vec = trace .issues .iter() @@ -328,8 +336,6 @@ impl MissionManager { }) .collect(); - // Extract actual error text from ActionFailed events - // and system messages (these contain the real diagnostics) let error_messages: Vec = thread .events .iter() @@ -345,7 +351,7 @@ impl MissionManager { None } }) - .take(10) // cap to avoid bloating payload + .take(10) .collect(); let payload = serde_json::json!({ @@ -353,10 +359,6 @@ impl MissionManager { "goal": thread.goal, "issues": issues, "error_messages": error_messages, - "reflection": { - "docs_produced": docs_produced, - "doc_types": doc_types, - }, }); if let Err(e) = mgr @@ -368,7 +370,121 @@ impl MissionManager { ) .await { - warn!("event listener: failed to fire self-improvement: {e}"); + warn!("event listener: failed to fire error diagnosis: {e}"); + } + } + + // ── Trigger 2: Playbook extraction ────────────── + let action_count = thread + .events + .iter() + .filter(|e| { + matches!( + e.kind, + crate::types::event::EventKind::ActionExecuted { .. } + ) + }) + .count(); + + if thread.state == crate::types::thread::ThreadState::Done + && trace.issues.iter().all(|i| { + i.severity != crate::executor::trace::IssueSeverity::Error + }) + && thread.step_count >= PLAYBOOK_MIN_STEPS + && action_count >= PLAYBOOK_MIN_ACTIONS + { + let actions_used: Vec = thread + .events + .iter() + .filter_map(|e| { + if let crate::types::event::EventKind::ActionExecuted { + action_name, + .. + } = &e.kind + { + Some(action_name.clone()) + } else { + None + } + }) + .collect(); + + let payload = serde_json::json!({ + "source_thread_id": event.thread_id.0.to_string(), + "goal": thread.goal, + "step_count": thread.step_count, + "action_count": action_count, + "actions_used": actions_used, + "total_tokens": thread.total_tokens_used, + }); + + if let Err(e) = mgr + .fire_on_system_event( + "engine", + "thread_completed_with_learnings", + &user_id, + Some(payload), + ) + .await + { + warn!("event listener: failed to fire playbook extraction: {e}"); + } + } + + // ── Trigger 3: Conversation insights ──────────── + // Use the thread's project_id as a proxy for conversation scope. + let conv_key = thread.project_id.0.to_string(); + let count = conv_thread_counts + .entry(conv_key.clone()) + .or_insert(0); + *count += 1; + + if (*count).is_multiple_of(CONVERSATION_INSIGHTS_INTERVAL) { + // Collect recent thread goals for context + let thread_goals: Vec = match mgr + .store + .list_threads(thread.project_id) + .await + { + Ok(threads) => threads + .iter() + .rev() + .take(CONVERSATION_INSIGHTS_INTERVAL as usize) + .map(|t| t.goal.clone()) + .collect(), + Err(_) => vec![thread.goal.clone()], + }; + + // Collect sample user messages from recent threads + let sample_messages: Vec = thread + .messages + .iter() + .filter(|m| { + m.role == crate::types::message::MessageRole::User + }) + .map(|m| { + m.content.chars().take(200).collect::() + }) + .take(10) + .collect(); + + let payload = serde_json::json!({ + "project_id": thread.project_id.0.to_string(), + "completed_thread_count": *count, + "thread_goals": thread_goals, + "sample_user_messages": sample_messages, + }); + + if let Err(e) = mgr + .fire_on_system_event( + "engine", + "conversation_insights_due", + &user_id, + Some(payload), + ) + .await + { + warn!("event listener: failed to fire conversation insights: {e}"); } } } @@ -446,6 +562,88 @@ impl MissionManager { Ok(id) } + /// Ensure all three learning missions exist for the given project. + /// + /// Creates (if missing) the self-improvement, playbook extraction, and + /// conversation insights missions. This is the preferred entry point — + /// call once at project bootstrap. + pub async fn ensure_learning_missions( + &self, + project_id: ProjectId, + ) -> Result<(), EngineError> { + // 1. Error diagnosis (self-improvement) — existing + self.ensure_self_improvement_mission(project_id).await?; + + // 2. Playbook extraction + self.ensure_mission_by_metadata( + project_id, + "playbook_extraction", + "playbook-extraction", + PLAYBOOK_EXTRACTION_GOAL, + MissionCadence::OnSystemEvent { + source: "engine".into(), + event_type: "thread_completed_with_learnings".into(), + }, + "Extract reusable playbooks from successful multi-step threads", + 3, // max 3/day + ) + .await?; + + // 3. Conversation insights + self.ensure_mission_by_metadata( + project_id, + "conversation_insights", + "conversation-insights", + CONVERSATION_INSIGHTS_GOAL, + MissionCadence::OnSystemEvent { + source: "engine".into(), + event_type: "conversation_insights_due".into(), + }, + "Extract user preferences, domain knowledge, and workflow patterns from conversations", + 2, // max 2/day + ) + .await?; + + Ok(()) + } + + /// Ensure a mission with a specific metadata tag exists, creating it if not. + #[allow(clippy::too_many_arguments)] + async fn ensure_mission_by_metadata( + &self, + project_id: ProjectId, + metadata_key: &str, + name: &str, + goal: &str, + cadence: MissionCadence, + success_criteria: &str, + max_per_day: u32, + ) -> Result { + let missions = self.store.list_missions(project_id).await?; + if let Some(existing) = missions + .iter() + .find(|m| m.metadata.get(metadata_key).is_some()) + { + let mut active = self.active.write().await; + if !active.contains(&existing.id) { + active.push(existing.id); + } + return Ok(existing.id); + } + + let mut mission = Mission::new(project_id, name, goal, cadence); + mission.success_criteria = Some(success_criteria.into()); + mission.metadata = serde_json::json!({metadata_key: true}); + mission.max_threads_per_day = max_per_day; + + let id = mission.id; + self.store.save_mission(&mission).await?; + self.active.write().await.push(id); + + debug!(mission_id = %id, name, "created learning mission"); + Ok(id) + } + /// Tick — check all active missions and fire any that are due. /// /// For `Cron` cadence missions, checks `next_fire_at` against current time. @@ -901,6 +1099,91 @@ pub const FIX_PATTERN_DB_TITLE: &str = "fix_pattern_database"; /// Well-known tag for the fix pattern database. pub const FIX_PATTERN_DB_TAG: &str = "fix_patterns"; +/// The goal for the playbook extraction mission. +const PLAYBOOK_EXTRACTION_GOAL: &str = "\ +You extract reusable playbooks from successfully completed multi-step threads. + +## Input + +`state[\"trigger_payload\"]` contains: +- `source_thread_id` — the thread that completed successfully +- `goal` — what the thread accomplished +- `step_count` — number of execution steps +- `action_count` — number of tool actions executed +- `actions_used` — list of tool names used +- `total_tokens` — tokens consumed + +## Process + +1. Search for the source thread's messages in memory: `memory_search(query=goal)` +2. Check for existing playbooks that cover this procedure: `memory_search(query=\"playbook\")` +3. If a similar playbook already exists, decide whether this thread adds new detail worth updating +4. Extract the step-by-step procedure, noting specific tool names and parameter patterns +5. Save as a Playbook memory doc via `memory_write(target=\"memory\", content=playbook_text)` \ + with title format \"playbook:\" + +## Output (FINAL) + +Report what you did: +- The playbook title and a one-line summary +- Whether it is new or an update to an existing playbook +- Next focus: what patterns to watch for + +## Rules + +- Only extract playbooks from threads with 3+ distinct tool calls +- Be specific about tool names and parameters — vague playbooks are useless +- If the thread was a trivial query-response, call FINAL(\"No playbook needed — simple interaction\") \ + and stop immediately +- One playbook per FINAL — do not combine unrelated procedures +"; + +/// The goal for the conversation insights mission. +const CONVERSATION_INSIGHTS_GOAL: &str = "\ +You extract user preferences, patterns, and domain knowledge from a batch of recent \ +conversation threads. + +## Input + +`state[\"trigger_payload\"]` contains: +- `project_id` — the project scope +- `completed_thread_count` — total threads completed in this conversation +- `thread_goals` — list of recent thread goals (what the user asked for) +- `sample_user_messages` — sample of actual user messages (truncated to 200 chars) + +## Process + +1. Analyze the thread goals and user messages for patterns +2. Search existing insights: `memory_search(query=\"user preferences\")` and \ + `memory_search(query=\"domain knowledge\")` +3. Extract NEW insights not already recorded in memory +4. Write each insight to memory via `memory_write(target=\"memory\", content=insight_text)` \ + with title format \"insight::\" + +## Categories to look for + +- **Preferences**: communication style, format choices, tool preferences +- **Domain**: project names, API patterns, data formats, technology stack +- **Workflow**: recurring task sequences, common follow-up questions +- **Corrections**: things the user corrected or repeated — these signal unmet expectations + +## Output (FINAL) + +Report: +- Number of new insights extracted (0 is fine) +- Brief list of what was found +- Next focus + +## Rules + +- Only record actionable, specific insights — not vague observations +- Do not record personal information, only work patterns +- If no meaningful new insights after analysis, call FINAL(\"No new insights — \ + conversation patterns already captured\") immediately +- Merge with existing insight docs rather than creating duplicates +- Max 5 insights per run to keep quality high +"; + /// Seed content for the fix pattern database. const SEED_FIX_PATTERNS: &str = "\ | Trace pattern | Fix strategy | Location pattern | diff --git a/crates/ironclaw_engine/src/types/event.rs b/crates/ironclaw_engine/src/types/event.rs index 159105a8..b858e57a 100644 --- a/crates/ironclaw_engine/src/types/event.rs +++ b/crates/ironclaw_engine/src/types/event.rs @@ -122,17 +122,6 @@ pub enum EventKind { approved: bool, }, - // ── Reflection ─────────────────────────────────────────── - ReflectionStarted, - ReflectionComplete { - docs_produced: usize, - doc_types: Vec, - tokens_used: u64, - }, - ReflectionFailed { - error: String, - }, - // ── Self-improvement ────────────────────────────────────── SelfImprovementStarted, SelfImprovementComplete { diff --git a/crates/ironclaw_engine/src/types/provenance.rs b/crates/ironclaw_engine/src/types/provenance.rs index e0f3e603..0b9c47bf 100644 --- a/crates/ironclaw_engine/src/types/provenance.rs +++ b/crates/ironclaw_engine/src/types/provenance.rs @@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize}; use crate::types::memory::DocId; -use crate::types::thread::ThreadId; /// The origin of a piece of data. #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -21,8 +20,6 @@ pub enum Provenance { ToolOutput { action_name: String }, /// Generated by the LLM. LlmGenerated, - /// Produced by the reflection pipeline. - Reflection { source_thread_id: ThreadId }, /// Retrieved from project memory. MemoryRetrieval { doc_id: DocId }, } diff --git a/crates/ironclaw_engine/src/types/thread.rs b/crates/ironclaw_engine/src/types/thread.rs index e1d7c904..793d9c2a 100644 --- a/crates/ironclaw_engine/src/types/thread.rs +++ b/crates/ironclaw_engine/src/types/thread.rs @@ -38,7 +38,7 @@ impl Default for ThreadId { /// ```text /// Created → Running → Waiting → Running (resume) /// → Suspended → Running (resume) -/// → Completed → Reflecting → Done +/// → Completed → Done /// → Failed /// ``` #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] @@ -51,10 +51,8 @@ pub enum ThreadState { Waiting, /// Paused by system (resource pressure, priority preemption). Suspended, - /// Execution finished successfully, may undergo reflection. + /// Execution finished successfully. Completed, - /// Post-completion reflection is running. - Reflecting, /// Fully finished (terminal). Done, /// Terminal failure. @@ -81,11 +79,7 @@ impl ThreadState { | (Self::Suspended, Self::Running) | (Self::Suspended, Self::Failed) // From Completed - | (Self::Completed, Self::Reflecting) | (Self::Completed, Self::Done) - // From Reflecting - | (Self::Reflecting, Self::Done) - | (Self::Reflecting, Self::Failed) ) } @@ -111,8 +105,6 @@ pub enum ThreadType { Research, /// Long-running goal that spawns threads over time. Mission, - /// Post-completion analysis of another thread. - Reflection, } // ── Thread configuration ──────────────────────────────────── @@ -124,8 +116,6 @@ pub struct ThreadConfig { pub max_iterations: usize, /// Maximum wall-clock duration for the thread. pub max_duration: Option, - /// Whether to run reflection after completion. - pub enable_reflection: bool, /// Whether to detect and nudge on tool intent without action calls. pub enable_tool_intent_nudge: bool, /// Maximum number of tool intent nudges per thread. @@ -160,7 +150,6 @@ impl Default for ThreadConfig { Self { max_iterations: 50, max_duration: None, - enable_reflection: false, enable_tool_intent_nudge: true, max_tool_intent_nudges: 2, max_tokens_total: None, @@ -350,21 +339,11 @@ mod tests { assert!(ThreadState::Suspended.can_transition_to(ThreadState::Running)); } - #[test] - fn completed_can_transition_to_reflecting() { - assert!(ThreadState::Completed.can_transition_to(ThreadState::Reflecting)); - } - #[test] fn completed_can_transition_to_done() { assert!(ThreadState::Completed.can_transition_to(ThreadState::Done)); } - #[test] - fn reflecting_can_transition_to_done() { - assert!(ThreadState::Reflecting.can_transition_to(ThreadState::Done)); - } - #[test] fn done_is_terminal() { assert!(ThreadState::Done.is_terminal()); @@ -430,16 +409,6 @@ mod tests { assert!(t.completed_at.is_some()); } - #[test] - fn full_lifecycle_with_reflection() { - let mut t = make_thread(); - t.transition_to(ThreadState::Running, None).unwrap(); - t.transition_to(ThreadState::Completed, None).unwrap(); - t.transition_to(ThreadState::Reflecting, None).unwrap(); - t.transition_to(ThreadState::Done, None).unwrap(); - assert_eq!(t.events.len(), 4); - } - #[test] fn add_message_records_event() { let mut t = make_thread(); diff --git a/src/bridge/router.rs b/src/bridge/router.rs index 4125bc10..c02db51d 100644 --- a/src/bridge/router.rs +++ b/src/bridge/router.rs @@ -201,12 +201,12 @@ pub async fn init_engine(agent: &Agent) -> Result<(), Error> { mission_manager.start_cron_ticker(agent.deps.owner_id.clone()); mission_manager.start_event_listener(agent.deps.owner_id.clone()); - // Ensure self-improvement mission exists for this project + // Ensure all learning missions exist for this project if let Err(e) = mission_manager - .ensure_self_improvement_mission(project_id) + .ensure_learning_missions(project_id) .await { - debug!("engine v2: failed to create self-improvement mission: {e}"); + debug!("engine v2: failed to create learning missions: {e}"); } // Wire mission manager into effect adapter for mission_* function calls @@ -816,10 +816,7 @@ pub async fn handle_with_engine( content, state.default_project_id, &message.user_id, - ThreadConfig { - enable_reflection: true, - ..ThreadConfig::default() - }, + ThreadConfig::default(), ) .await .map_err(|e| engine_err("thread error", e))?; @@ -1108,26 +1105,6 @@ async fn forward_event_to_channel( .await; } } - EventKind::ReflectionStarted => { - let _ = channels - .send_status( - channel_name, - StatusUpdate::Thinking("Reflecting on execution...".into()), - metadata, - ) - .await; - } - EventKind::ReflectionComplete { docs_produced, .. } => { - let _ = channels - .send_status( - channel_name, - StatusUpdate::Thinking(format!( - "Reflection complete — {docs_produced} insight(s) saved" - )), - metadata, - ) - .await; - } _ => {} } } @@ -1211,14 +1188,6 @@ fn thread_event_to_app_events( .into_iter() .collect() } - EventKind::ReflectionStarted => vec![AppEvent::Thinking { - message: "Reflecting on execution...".into(), - thread_id: Some(thread_id.into()), - }], - EventKind::ReflectionComplete { docs_produced, .. } => vec![AppEvent::Status { - message: format!("Reflection complete — {docs_produced} insight(s) saved"), - thread_id: Some(thread_id.into()), - }], EventKind::StateChanged { from, to, reason } => { vec![AppEvent::ThreadStateChanged { thread_id: thread_id.into(), diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 63aedaa0..2b8203af 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -534,22 +534,17 @@ impl Channel for GatewayChannel { user_id: &str, response: OutgoingResponse, ) -> Result<(), ChannelError> { - let thread_id = match response.thread_id { - Some(tid) => tid, - None => { - tracing::warn!( - "Gateway broadcast with no thread_id — skipping (clients would drop it)" - ); - return Ok(()); - } - }; - self.state.sse.broadcast_for_user( - user_id, - AppEvent::Response { + let event = match response.thread_id { + Some(thread_id) => AppEvent::Response { content: response.content, thread_id, }, - ); + None => AppEvent::Status { + message: response.content, + thread_id: None, + }, + }; + self.state.sse.broadcast_for_user(user_id, event); Ok(()) }