From 098addce27c89b6cb0328b3dd8b4245096c527c6 Mon Sep 17 00:00:00 2001 From: "ilblackdragon@gmail.com" Date: Thu, 26 Mar 2026 18:19:10 -0700 Subject: [PATCH] feat(engine): complete v2 side-by-side integration with gateway API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire engine v2 into the full submission pipeline and expose threads, projects, and missions through the web gateway REST API. Bridge routing — route ExecApproval, Interrupt, NewThread, and Clear submissions to engine v2 when ENGINE_V2=true. Previously only UserInput and ApprovalResponse were handled; all other control commands fell through to disconnected v1 sessions. Bridge query layer — add 11 read-only query functions and 6 DTO types so gateway handlers can inspect engine state (threads, steps, events, projects, missions) without direct access to the EngineState singleton. Gateway endpoints — new /api/engine/* routes: GET /threads, /threads/{id}, /threads/{id}/steps, /threads/{id}/events GET /projects, /projects/{id} GET /missions, /missions/{id} POST /missions/{id}/fire, /missions/{id}/pause, /missions/{id}/resume SSE events — add ThreadStateChanged, ChildThreadSpawned, and MissionThreadSpawned AppEvent variants. Expand the bridge event mapper to forward StateChanged and ChildSpawned engine events to the browser. Engine crate — add ConversationManager::clear_conversation() for /new and /clear commands. Code quality — replace 10 .expect() calls with proper error returns, remove dead AgentConfig.engine_v2 field, log silent init errors, fix duplicate doc comment, improve fallthrough documentation. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/ironclaw_common/src/event.rs | 46 + .../src/executor/loop_engine.rs | 2 +- .../src/executor/orchestrator.rs | 148 ++-- .../src/runtime/conversation.rs | 56 ++ src/agent/agent_loop.rs | 29 +- src/bridge/effect_adapter.rs | 6 +- src/bridge/mod.rs | 29 +- src/bridge/router.rs | 827 +++++++++++++++--- src/channels/web/handlers/engine.rs | 143 +++ src/channels/web/handlers/mod.rs | 1 + src/channels/web/server.rs | 42 + src/channels/web/types.rs | 54 ++ 12 files changed, 1196 insertions(+), 187 deletions(-) create mode 100644 src/channels/web/handlers/engine.rs diff --git a/crates/ironclaw_common/src/event.rs b/crates/ironclaw_common/src/event.rs index 256aba3d..da801e11 100644 --- a/crates/ironclaw_common/src/event.rs +++ b/crates/ironclaw_common/src/event.rs @@ -206,6 +206,33 @@ pub enum AppEvent { narrative: String, decisions: Vec, }, + + // ── Engine v2 thread lifecycle events ── + /// Engine thread changed state (e.g. Running → Completed). + #[serde(rename = "thread_state_changed")] + ThreadStateChanged { + thread_id: String, + from_state: String, + to_state: String, + #[serde(skip_serializing_if = "Option::is_none")] + reason: Option, + }, + + /// A child thread was spawned by a parent thread. + #[serde(rename = "child_thread_spawned")] + ChildThreadSpawned { + parent_thread_id: String, + child_thread_id: String, + goal: String, + }, + + /// A mission spawned a new thread. + #[serde(rename = "mission_thread_spawned")] + MissionThreadSpawned { + mission_id: String, + thread_id: String, + mission_name: String, + }, } impl AppEvent { @@ -236,6 +263,9 @@ impl AppEvent { Self::ExtensionStatus { .. } => "extension_status", Self::ReasoningUpdate { .. } => "reasoning_update", Self::JobReasoning { .. } => "job_reasoning", + Self::ThreadStateChanged { .. } => "thread_state_changed", + Self::ChildThreadSpawned { .. } => "child_thread_spawned", + Self::MissionThreadSpawned { .. } => "mission_thread_spawned", } } } @@ -366,6 +396,22 @@ mod tests { narrative: String::new(), decisions: vec![], }, + AppEvent::ThreadStateChanged { + thread_id: String::new(), + from_state: String::new(), + to_state: String::new(), + reason: None, + }, + AppEvent::ChildThreadSpawned { + parent_thread_id: String::new(), + child_thread_id: String::new(), + goal: String::new(), + }, + AppEvent::MissionThreadSpawned { + mission_id: String::new(), + thread_id: String::new(), + mission_name: String::new(), + }, ]; for variant in &variants { diff --git a/crates/ironclaw_engine/src/executor/loop_engine.rs b/crates/ironclaw_engine/src/executor/loop_engine.rs index b1109e5a..2ac8c10a 100644 --- a/crates/ironclaw_engine/src/executor/loop_engine.rs +++ b/crates/ironclaw_engine/src/executor/loop_engine.rs @@ -444,8 +444,8 @@ mod tests { use crate::traits::effect::ThreadExecutionContext; use crate::traits::llm::{LlmCallConfig, LlmOutput}; use crate::types::capability::{ActionDef, CapabilityLease, EffectType}; - use crate::types::step::LlmResponse; use crate::types::project::ProjectId; + use crate::types::step::LlmResponse; use crate::types::step::{ActionResult, TokenUsage}; use crate::types::thread::{ThreadConfig, ThreadType}; diff --git a/crates/ironclaw_engine/src/executor/orchestrator.rs b/crates/ironclaw_engine/src/executor/orchestrator.rs index 48f0ea58..6b4987e0 100644 --- a/crates/ironclaw_engine/src/executor/orchestrator.rs +++ b/crates/ironclaw_engine/src/executor/orchestrator.rs @@ -99,13 +99,19 @@ pub async fn load_orchestrator( // Find all orchestrator versions, sorted by version number descending let mut versions: Vec<_> = docs .iter() - .filter(|d| { - d.title == ORCHESTRATOR_TITLE && d.tags.contains(&ORCHESTRATOR_TAG.to_string()) - }) + .filter(|d| d.title == ORCHESTRATOR_TITLE && d.tags.contains(&ORCHESTRATOR_TAG.to_string())) .collect(); versions.sort_by(|a, b| { - let va = a.metadata.get("version").and_then(|v| v.as_u64()).unwrap_or(0); - let vb = b.metadata.get("version").and_then(|v| v.as_u64()).unwrap_or(0); + let va = a + .metadata + .get("version") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + let vb = b + .metadata + .get("version") + .and_then(|v| v.as_u64()) + .unwrap_or(0); vb.cmp(&va) // descending }); @@ -125,7 +131,12 @@ pub async fn load_orchestrator( .unwrap_or(1); // Skip versions with too many failures (only check the latest) - if version == versions[0].metadata.get("version").and_then(|v| v.as_u64()).unwrap_or(1) + if version + == versions[0] + .metadata + .get("version") + .and_then(|v| v.as_u64()) + .unwrap_or(1) && failures >= MAX_FAILURES_BEFORE_ROLLBACK { warn!( @@ -189,10 +200,7 @@ pub async fn record_orchestrator_failure( } /// Reset the failure counter (called after successful execution). -pub async fn reset_orchestrator_failures( - store: &Arc, - project_id: ProjectId, -) { +pub async fn reset_orchestrator_failures(store: &Arc, project_id: ProjectId) { let docs = store.list_memory_docs(project_id).await.unwrap_or_default(); let existing = docs.iter().find(|d| d.title == FAILURE_TRACKER_TITLE); @@ -312,8 +320,16 @@ pub async fn execute_orchestrator( // __llm_complete__(messages, actions, config) "__llm_complete__" => { - handle_llm_complete(args, kwargs, thread, llm, effects, leases, &mut total_tokens) - .await + handle_llm_complete( + args, + kwargs, + thread, + llm, + effects, + leases, + &mut total_tokens, + ) + .await } // __execute_code_step__(code, state) @@ -512,7 +528,7 @@ async fn handle_execute_code_step( return ExtFunctionResult::Error(monty::MontyException::new( monty::ExcType::TypeError, Some("__execute_code_step__ requires a code string".into()), - )) + )); } }; @@ -531,7 +547,15 @@ async fn handle_execute_code_step( // Run user code in a nested Monty VM (same pattern as rlm_query) match Box::pin(execute_code( - &code, thread, llm, effects, leases, policy, &exec_ctx, &[], &state, + &code, + thread, + llm, + effects, + leases, + policy, + &exec_ctx, + &[], + &state, )) .await { @@ -593,7 +617,7 @@ async fn handle_execute_action( return ExtFunctionResult::Error(monty::MontyException::new( monty::ExcType::TypeError, Some("__execute_action__ requires a name argument".into()), - )) + )); } }; @@ -770,9 +794,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" | "assistant_actions" => thread.add_message(ThreadMessage::assistant(&content)), "system" => thread.add_message(ThreadMessage::system(&content)), "system_append" => { // Append to existing system message (for doc injection) @@ -847,7 +869,7 @@ fn handle_transition_to( return ExtFunctionResult::Error(monty::MontyException::new( monty::ExcType::ValueError, Some(format!("Unknown thread state: {other}")), - )) + )); } }; @@ -1029,7 +1051,10 @@ fn parse_outcome(result: &serde_json::Value) -> ThreadOutcome { match outcome { "completed" => ThreadOutcome::Completed { - response: result.get("response").and_then(|v| v.as_str()).map(String::from), + response: result + .get("response") + .and_then(|v| v.as_str()) + .map(String::from), }, "stopped" => ThreadOutcome::Stopped, "max_iterations" => ThreadOutcome::MaxIterations, @@ -1126,8 +1151,7 @@ mod tests { doc.metadata = serde_json::json!({"version": 1}); let store = Arc::new(crate::tests::InMemoryStore::with_docs(vec![doc])); - let (code, version) = - load_orchestrator(Some(&(store as Arc)), project_id).await; + let (code, version) = load_orchestrator(Some(&(store as Arc)), project_id).await; assert_eq!(version, 1); assert!(code.contains("custom_orchestrator_code")); } @@ -1135,37 +1159,22 @@ mod tests { #[tokio::test] async fn load_orchestrator_picks_highest_version() { let project_id = ProjectId::new(); - let mut doc_v1 = MemoryDoc::new( - project_id, - DocType::Note, - ORCHESTRATOR_TITLE, - "v1_code()", - ) - .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); + let mut doc_v1 = MemoryDoc::new(project_id, DocType::Note, ORCHESTRATOR_TITLE, "v1_code()") + .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); doc_v1.metadata = serde_json::json!({"version": 1}); - let mut doc_v3 = MemoryDoc::new( - project_id, - DocType::Note, - ORCHESTRATOR_TITLE, - "v3_code()", - ) - .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); + let mut doc_v3 = MemoryDoc::new(project_id, DocType::Note, ORCHESTRATOR_TITLE, "v3_code()") + .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); doc_v3.metadata = serde_json::json!({"version": 3}); - let mut doc_v2 = MemoryDoc::new( - project_id, - DocType::Note, - ORCHESTRATOR_TITLE, - "v2_code()", - ) - .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); + let mut doc_v2 = MemoryDoc::new(project_id, DocType::Note, ORCHESTRATOR_TITLE, "v2_code()") + .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); doc_v2.metadata = serde_json::json!({"version": 2}); - let store = - Arc::new(crate::tests::InMemoryStore::with_docs(vec![doc_v1, doc_v3, doc_v2])); - let (code, version) = - load_orchestrator(Some(&(store as Arc)), project_id).await; + let store = Arc::new(crate::tests::InMemoryStore::with_docs(vec![ + doc_v1, doc_v3, doc_v2, + ])); + let (code, version) = load_orchestrator(Some(&(store as Arc)), project_id).await; assert_eq!(version, 3); assert!(code.contains("v3_code")); } @@ -1175,23 +1184,15 @@ mod tests { let project_id = ProjectId::new(); // Create v2 orchestrator - let mut doc_v2 = MemoryDoc::new( - project_id, - DocType::Note, - ORCHESTRATOR_TITLE, - "v2_buggy()", - ) - .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); + let mut doc_v2 = + MemoryDoc::new(project_id, DocType::Note, ORCHESTRATOR_TITLE, "v2_buggy()") + .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); doc_v2.metadata = serde_json::json!({"version": 2}); // Create v1 orchestrator (fallback) - let mut doc_v1 = MemoryDoc::new( - project_id, - DocType::Note, - ORCHESTRATOR_TITLE, - "v1_stable()", - ) - .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); + let mut doc_v1 = + MemoryDoc::new(project_id, DocType::Note, ORCHESTRATOR_TITLE, "v1_stable()") + .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); doc_v1.metadata = serde_json::json!({"version": 1}); // Create failure tracker showing v2 has 3 failures @@ -1206,8 +1207,7 @@ mod tests { let store = Arc::new(crate::tests::InMemoryStore::with_docs(vec![ doc_v2, doc_v1, tracker, ])); - let (code, version) = - load_orchestrator(Some(&(store as Arc)), project_id).await; + let (code, version) = load_orchestrator(Some(&(store as Arc)), project_id).await; // Should skip v2 (too many failures) and load v1 assert_eq!(version, 1); @@ -1219,13 +1219,9 @@ mod tests { let project_id = ProjectId::new(); // Single version with 3 failures - let mut doc_v1 = MemoryDoc::new( - project_id, - DocType::Note, - ORCHESTRATOR_TITLE, - "v1_broken()", - ) - .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); + let mut doc_v1 = + MemoryDoc::new(project_id, DocType::Note, ORCHESTRATOR_TITLE, "v1_broken()") + .with_tags(vec![ORCHESTRATOR_TAG.to_string()]); doc_v1.metadata = serde_json::json!({"version": 1}); let tracker = MemoryDoc::new( @@ -1236,10 +1232,10 @@ mod tests { ) .with_tags(vec!["orchestrator_meta".to_string()]); - let store = - Arc::new(crate::tests::InMemoryStore::with_docs(vec![doc_v1, tracker])); - let (code, version) = - load_orchestrator(Some(&(store as Arc)), project_id).await; + let store = Arc::new(crate::tests::InMemoryStore::with_docs(vec![ + doc_v1, tracker, + ])); + let (code, version) = load_orchestrator(Some(&(store as Arc)), project_id).await; // Should fall back to compiled-in default (v0) assert_eq!(version, 0); @@ -1307,7 +1303,9 @@ mod tests { "parameters": {"cmd": "rm -rf /"} }); let outcome = parse_outcome(&result); - assert!(matches!(outcome, ThreadOutcome::NeedApproval { action_name, .. } if action_name == "shell")); + assert!( + matches!(outcome, ThreadOutcome::NeedApproval { action_name, .. } if action_name == "shell") + ); } #[test] diff --git a/crates/ironclaw_engine/src/runtime/conversation.rs b/crates/ironclaw_engine/src/runtime/conversation.rs index 0043e455..a90a7c63 100644 --- a/crates/ironclaw_engine/src/runtime/conversation.rs +++ b/crates/ironclaw_engine/src/runtime/conversation.rs @@ -255,6 +255,25 @@ impl ConversationManager { Ok(()) } + /// Clear a conversation's entries and active threads. + /// + /// Stops tracking all threads and removes conversation history so the next + /// user message spawns a fresh thread with no prior context. + pub async fn clear_conversation( + &self, + conversation_id: ConversationId, + ) -> Result<(), EngineError> { + let mut convs = self.conversations.write().await; + if let Some(conv) = convs.get_mut(&conversation_id) { + conv.active_threads.clear(); + conv.entries.clear(); + conv.updated_at = chrono::Utc::now(); + self.store.save_conversation(conv).await?; + debug!(conversation_id = %conversation_id, "cleared conversation"); + } + Ok(()) + } + /// Get a snapshot of a conversation. pub async fn get_conversation( &self, @@ -745,4 +764,41 @@ mod tests { assert_eq!(saved.entries.len(), 1); assert_eq!(saved.entries[0].content, "persisted"); } + + #[tokio::test] + async fn clear_conversation_resets_entries_and_threads() { + let (tm, cm) = make_conv_manager(); + let conv_id = cm.get_or_create_conversation("web", "user1").await.unwrap(); + let project = ProjectId::new(); + + // Spawn a thread so the conversation has entries and active threads + let tid = cm + .handle_user_message(conv_id, "Hello", project, "user1", ThreadConfig::default()) + .await + .unwrap(); + + // Wait for thread to finish + let _ = tm.join_thread(tid).await.unwrap(); + + // Record outcome so there's an agent entry + cm.record_thread_outcome( + conv_id, + tid, + &ThreadOutcome::Completed { + response: Some("Hi there".into()), + }, + ) + .await + .unwrap(); + + let conv = cm.get_conversation(conv_id).await.unwrap(); + assert!(!conv.entries.is_empty()); + + // Clear the conversation + cm.clear_conversation(conv_id).await.unwrap(); + + let conv = cm.get_conversation(conv_id).await.unwrap(); + assert!(conv.entries.is_empty()); + assert!(conv.active_threads.is_empty()); + } } diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 6ec9f69f..ae9ab041 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -1112,7 +1112,34 @@ impl Agent { Submission::ApprovalResponse { approved, always } => { return crate::bridge::handle_approval(self, message, *approved, *always).await; } - _ => {} // Other submissions fall through to v1 + Submission::ExecApproval { + request_id, + approved, + always, + } => { + return crate::bridge::handle_exec_approval( + self, + message, + *request_id, + *approved, + *always, + ) + .await; + } + Submission::Interrupt => { + return crate::bridge::handle_interrupt(self, message).await; + } + Submission::NewThread => { + return crate::bridge::handle_new_thread(self, message).await; + } + Submission::Clear => { + return crate::bridge::handle_clear(self, message).await; + } + // Undo/Redo/Resume/SwitchThread: v1-only (engine has no undo; + // thread switching is implicit via ConversationManager). + // Compact/Summarize/Suggest: orthogonal to engine (use workspace/LLM directly). + // Heartbeat/SystemCommand/JobStatus/JobCancel/Quit: v1 infrastructure. + _ => {} } } diff --git a/src/bridge/effect_adapter.rs b/src/bridge/effect_adapter.rs index d483a54c..da2e5e41 100644 --- a/src/bridge/effect_adapter.rs +++ b/src/bridge/effect_adapter.rs @@ -73,6 +73,11 @@ impl EffectBridgeAdapter { *self.mission_manager.write().await = Some(mgr); } + /// Get the mission manager if available. + pub async fn mission_manager(&self) -> Option> { + self.mission_manager.read().await.clone() + } + /// Handle mission_* function calls. Returns None if not a mission call. async fn handle_mission_call( &self, @@ -501,7 +506,6 @@ fn parse_cadence(s: &str) -> ironclaw_engine::types::mission::MissionCadence { /// Tools that depend on v1 runtime components (RoutineEngine, Scheduler, /// ContainerJobManager) and cannot work in engine v2's minimal JobContext. -/// Tools that depend on v1 runtime components and can't work in engine v2. /// Note: routine_* tools are NOT blocked — they map to mission operations. fn is_v1_only_tool(name: &str) -> bool { matches!( diff --git a/src/bridge/mod.rs b/src/bridge/mod.rs index 03671022..a2e94fde 100644 --- a/src/bridge/mod.rs +++ b/src/bridge/mod.rs @@ -10,5 +10,32 @@ mod router; mod store_adapter; pub use router::{ - handle_approval, handle_with_engine, is_engine_v2_enabled, pending_approval_for_user_thread, + // DTO types + EngineMissionDetail, + EngineMissionInfo, + EngineProjectInfo, + EngineStepInfo, + EngineThreadDetail, + EngineThreadInfo, + // Query functions + fire_engine_mission, + get_engine_mission, + get_engine_project, + get_engine_thread, + // Action handlers + handle_approval, + handle_clear, + handle_exec_approval, + handle_interrupt, + handle_new_thread, + handle_with_engine, + is_engine_v2_enabled, + list_engine_missions, + list_engine_projects, + list_engine_thread_events, + list_engine_thread_steps, + list_engine_threads, + pause_engine_mission, + pending_approval_for_user_thread, + resume_engine_mission, }; diff --git a/src/bridge/router.rs b/src/bridge/router.rs index 367addbc..b633f125 100644 --- a/src/bridge/router.rs +++ b/src/bridge/router.rs @@ -29,6 +29,14 @@ pub fn is_engine_v2_enabled() -> bool { .unwrap_or(false) } +/// Shorthand for building an `Error` from an engine-related failure. +fn engine_err(context: &str, e: impl std::fmt::Display) -> Error { + Error::from(crate::error::JobError::ContextError { + id: uuid::Uuid::nil(), + reason: format!("engine v2 {context}: {e}"), + }) +} + /// Pending approval info stored between the NeedApproval outcome and the user's response. #[derive(Clone)] struct PendingApproval { @@ -146,12 +154,7 @@ async fn get_or_init_engine(agent: &Agent) -> Result<(), Error> { let project_id = match store .list_projects() .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })? + .map_err(|e| engine_err("store error", e))? .into_iter() .find(|project| project.name == "default") { @@ -159,29 +162,39 @@ async fn get_or_init_engine(agent: &Agent) -> Result<(), Error> { None => { let project = Project::new("default", "Default project for engine v2"); let project_id = project.id; - store.save_project(&project).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })?; + store + .save_project(&project) + .await + .map_err(|e| engine_err("store error", e))?; project_id } }; let conversation_manager = ConversationManager::new(Arc::clone(&thread_manager), store.clone()); - let _ = conversation_manager + if let Err(e) = conversation_manager .bootstrap_user(&agent.deps.owner_id) - .await; + .await + { + debug!("engine v2: bootstrap_user failed: {e}"); + } // Create mission manager and start cron ticker let mission_manager = Arc::new(MissionManager::new(store_dyn, Arc::clone(&thread_manager))); - let _ = thread_manager.recover_project_threads(project_id).await; - let _ = mission_manager.bootstrap_project(project_id).await; - let _ = mission_manager + if let Err(e) = thread_manager.recover_project_threads(project_id).await { + debug!("engine v2: recover_project_threads failed: {e}"); + } + if let Err(e) = mission_manager.bootstrap_project(project_id).await { + debug!("engine v2: bootstrap_project failed: {e}"); + } + if let Err(e) = mission_manager .resume_recoverable_threads(&agent.deps.owner_id) - .await; - let _ = thread_manager.resume_background_threads(project_id).await; + .await + { + debug!("engine v2: resume_recoverable_threads failed: {e}"); + } + if let Err(e) = thread_manager.resume_background_threads(project_id).await { + debug!("engine v2: resume_background_threads failed: {e}"); + } mission_manager.start_cron_ticker(agent.deps.owner_id.clone()); mission_manager.start_event_listener(agent.deps.owner_id.clone()); @@ -219,25 +232,13 @@ async fn persist_pending_approval( let mut thread = store .load_thread(pending.thread_id) .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })? - .ok_or_else(|| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 thread {} not found", pending.thread_id), - }) - })?; + .map_err(|e| engine_err("store error", e))? + .ok_or_else(|| engine_err("thread not found", pending.thread_id))?; - let metadata = thread.metadata.as_object_mut().ok_or_else(|| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: "engine v2 thread metadata must be an object".into(), - }) - })?; + let metadata = thread + .metadata + .as_object_mut() + .ok_or_else(|| engine_err("thread metadata", "must be an object"))?; metadata.insert( PENDING_APPROVAL_METADATA_KEY.into(), serde_json::json!({ @@ -251,12 +252,10 @@ async fn persist_pending_approval( }), ); thread.updated_at = chrono::Utc::now(); - store.save_thread(&thread).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - }) + store + .save_thread(&thread) + .await + .map_err(|e| engine_err("store error", e)) } async fn load_pending_approval_from_thread( @@ -264,12 +263,10 @@ async fn load_pending_approval_from_thread( conversation_id: ironclaw_engine::ConversationId, thread_id: ironclaw_engine::ThreadId, ) -> Result, Error> { - let Some(thread) = store.load_thread(thread_id).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })? + let Some(thread) = store + .load_thread(thread_id) + .await + .map_err(|e| engine_err("store error", e))? else { return Ok(None); }; @@ -321,12 +318,10 @@ async fn clear_pending_approval_metadata( store: &Arc, thread_id: ironclaw_engine::ThreadId, ) -> Result<(), Error> { - let Some(mut thread) = store.load_thread(thread_id).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })? + let Some(mut thread) = store + .load_thread(thread_id) + .await + .map_err(|e| engine_err("store error", e))? else { return Ok(()); }; @@ -334,12 +329,10 @@ async fn clear_pending_approval_metadata( if let Some(metadata) = thread.metadata.as_object_mut() { metadata.remove(PENDING_APPROVAL_METADATA_KEY); thread.updated_at = chrono::Utc::now(); - store.save_thread(&thread).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })?; + store + .save_thread(&thread) + .await + .map_err(|e| engine_err("store error", e))?; } Ok(()) @@ -375,12 +368,10 @@ async fn resolve_pending_approval_for_thread( } } - let conversations = store.list_conversations(user_id).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })?; + let conversations = store + .list_conversations(user_id) + .await + .map_err(|e| engine_err("store error", e))?; let mut candidates = Vec::new(); for conversation in conversations { @@ -389,12 +380,10 @@ async fn resolve_pending_approval_for_thread( continue; } - let Some(thread) = store.load_thread(thread_id).await.map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 store error: {e}"), - }) - })? + let Some(thread) = store + .load_thread(thread_id) + .await + .map_err(|e| engine_err("store error", e))? else { continue; }; @@ -469,9 +458,13 @@ pub async fn handle_approval( ) -> Result, Error> { get_or_init_engine(agent).await?; - let lock = ENGINE_STATE.get().expect("engine initialized"); + let lock = ENGINE_STATE + .get() + .ok_or_else(|| engine_err("init", "engine state not initialized"))?; let guard = lock.read().await; - let state = guard.as_ref().expect("engine initialized"); + let state = guard + .as_ref() + .ok_or_else(|| engine_err("init", "engine state is empty"))?; let pending = match resolve_pending_approval_for_thread( &state.store, @@ -493,6 +486,75 @@ pub async fn handle_approval( } }; + process_resolved_approval(agent, state, message, pending, approved, always).await +} + +/// Handle an `ExecApproval` submission (web gateway JSON approval with explicit request_id). +pub async fn handle_exec_approval( + agent: &Agent, + message: &IncomingMessage, + request_id: uuid::Uuid, + approved: bool, + always: bool, +) -> Result, Error> { + get_or_init_engine(agent).await?; + + let lock = ENGINE_STATE + .get() + .ok_or_else(|| engine_err("init", "engine state not initialized"))?; + let guard = lock.read().await; + let state = guard + .as_ref() + .ok_or_else(|| engine_err("init", "engine state is empty"))?; + + let request_id_str = request_id.to_string(); + + // First try the in-memory cache (keyed by user_id, but we match on request_id). + let cached = state + .pending_approvals + .read() + .await + .get(&message.user_id) + .filter(|p| p.request_id == request_id_str) + .cloned(); + + if let Some(pending) = cached { + return process_resolved_approval(agent, state, message, pending, approved, always).await; + } + + // Fall back to scanning thread metadata for this user's conversations. + let resolution = resolve_pending_approval_for_thread( + &state.store, + &state.pending_approvals, + &message.user_id, + message.thread_id.as_deref(), + ) + .await?; + + match resolution { + PendingApprovalResolution::Resolved(pending) if pending.request_id == request_id_str => { + process_resolved_approval(agent, state, message, pending, approved, always).await + } + _ => { + debug!( + user_id = %message.user_id, + request_id = %request_id, + "engine v2: no matching pending approval for request_id" + ); + Ok(Some("No matching pending approval found.".into())) + } + } +} + +/// Shared logic for processing a resolved pending approval. +async fn process_resolved_approval( + agent: &Agent, + state: &EngineState, + message: &IncomingMessage, + pending: PendingApproval, + approved: bool, + always: bool, +) -> Result, Error> { if !approved { let _ = agent .channels @@ -504,7 +566,6 @@ pub async fn handle_approval( .await; } - // Approved — persist auto-approval when user chose "always" debug!( tool = %pending.action_name, always, @@ -556,12 +617,7 @@ pub async fn handle_approval( Some((pending.call_id.clone(), approved)), ) .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 resume error: {e}"), - }) - })?; + .map_err(|e| engine_err("resume error", e))?; clear_pending_approval_metadata(&state.store, pending.thread_id).await?; let mut approvals = state.pending_approvals.write().await; if approvals @@ -581,6 +637,120 @@ pub async fn handle_approval( .await } +/// Handle an interrupt submission — stop active engine threads. +pub async fn handle_interrupt( + agent: &Agent, + message: &IncomingMessage, +) -> Result, Error> { + get_or_init_engine(agent).await?; + + let lock = ENGINE_STATE + .get() + .ok_or_else(|| engine_err("init", "engine state not initialized"))?; + let guard = lock.read().await; + let state = guard + .as_ref() + .ok_or_else(|| engine_err("init", "engine state is empty"))?; + + let conv_id = state + .conversation_manager + .get_or_create_conversation(&message.channel, &message.user_id) + .await + .map_err(|e| engine_err("conversation error", e))?; + + let conv = state.conversation_manager.get_conversation(conv_id).await; + let active_threads = conv + .as_ref() + .map(|c| c.active_threads.clone()) + .unwrap_or_default(); + + let mut stopped = 0u32; + for tid in &active_threads { + if state.thread_manager.is_running(*tid).await { + if let Err(e) = state.thread_manager.stop_thread(*tid).await { + debug!(thread_id = %tid, error = %e, "engine v2: failed to stop thread"); + } else { + stopped += 1; + } + } + } + + if stopped > 0 { + debug!(stopped, "engine v2: interrupted running threads"); + Ok(Some("Interrupted.".into())) + } else { + Ok(Some("Nothing to interrupt.".into())) + } +} + +/// Handle a new-thread submission — clear conversation for a fresh start. +pub async fn handle_new_thread( + agent: &Agent, + message: &IncomingMessage, +) -> Result, Error> { + clear_engine_conversation(agent, message).await?; + Ok(Some("Started new conversation.".into())) +} + +/// Handle a clear submission — stop threads and reset conversation. +pub async fn handle_clear( + agent: &Agent, + message: &IncomingMessage, +) -> Result, Error> { + clear_engine_conversation(agent, message).await?; + Ok(Some("Conversation cleared.".into())) +} + +/// Stop all active threads and clear conversation entries. +async fn clear_engine_conversation(agent: &Agent, message: &IncomingMessage) -> Result<(), Error> { + get_or_init_engine(agent).await?; + + let lock = ENGINE_STATE + .get() + .ok_or_else(|| engine_err("init", "engine state not initialized"))?; + let guard = lock.read().await; + let state = guard + .as_ref() + .ok_or_else(|| engine_err("init", "engine state is empty"))?; + + let conv_id = state + .conversation_manager + .get_or_create_conversation(&message.channel, &message.user_id) + .await + .map_err(|e| engine_err("conversation error", e))?; + + // Stop all active threads first + if let Some(conv) = state.conversation_manager.get_conversation(conv_id).await { + for tid in &conv.active_threads { + if state.thread_manager.is_running(*tid).await { + let _ = state.thread_manager.stop_thread(*tid).await; + } + } + } + + // Clear the conversation entries and active thread list + state + .conversation_manager + .clear_conversation(conv_id) + .await + .map_err(|e| engine_err("clear conversation error", e))?; + + // Also clear any pending approvals for this user + state + .pending_approvals + .write() + .await + .remove(&message.user_id); + + debug!( + user_id = %message.user_id, + conversation_id = %conv_id, + "engine v2: conversation cleared" + ); + + Ok(()) +} + /// Handle a user message through the engine v2 pipeline. pub async fn handle_with_engine( agent: &Agent, @@ -590,9 +760,13 @@ pub async fn handle_with_engine( // Ensure engine is initialized get_or_init_engine(agent).await?; - let lock = ENGINE_STATE.get().expect("engine initialized"); + let lock = ENGINE_STATE + .get() + .ok_or_else(|| engine_err("init", "engine state not initialized"))?; let guard = lock.read().await; - let state = guard.as_ref().expect("engine initialized"); + let state = guard + .as_ref() + .ok_or_else(|| engine_err("init", "engine state is empty"))?; debug!( user_id = %message.user_id, @@ -618,12 +792,7 @@ pub async fn handle_with_engine( .conversation_manager .get_or_create_conversation(&message.channel, &message.user_id) .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 conversation error: {e}"), - }) - })?; + .map_err(|e| engine_err("conversation error", e))?; // Handle the message — spawns a new thread or injects into active one let thread_id = state @@ -639,12 +808,7 @@ pub async fn handle_with_engine( }, ) .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 error: {e}"), - }) - })?; + .map_err(|e| engine_err("thread error", e))?; if let Some(ref db) = state.db && let Ok(conv_id_v1) = db @@ -702,23 +866,13 @@ async fn await_thread_outcome( .thread_manager .join_thread(thread_id) .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 join error: {e}"), - }) - })?; + .map_err(|e| engine_err("join error", e))?; state .conversation_manager .record_thread_outcome(conv_id, thread_id, &outcome) .await - .map_err(|e| { - crate::error::Error::from(crate::error::JobError::ContextError { - id: uuid::Uuid::nil(), - reason: format!("engine v2 conversation error: {e}"), - }) - })?; + .map_err(|e| engine_err("conversation error", e))?; if let Some(ref db) = state.db && let Ok(conv_id_v1) = db @@ -898,10 +1052,467 @@ fn thread_event_to_app_event( message: "Processing results...".into(), thread_id: Some(thread_id.into()), }), + EventKind::StateChanged { from, to, reason } => Some(AppEvent::ThreadStateChanged { + thread_id: thread_id.into(), + from_state: format!("{from:?}"), + to_state: format!("{to:?}"), + reason: reason.clone(), + }), + EventKind::ChildSpawned { child_id, goal } => Some(AppEvent::ChildThreadSpawned { + parent_thread_id: thread_id.into(), + child_thread_id: child_id.to_string(), + goal: goal.clone(), + }), _ => None, } } +// ── Engine query DTOs ──────────────────────────────────────── + +/// Lightweight thread summary for list views. +#[derive(Debug, Clone, serde::Serialize)] +pub struct EngineThreadInfo { + pub id: String, + pub goal: String, + pub thread_type: String, + pub state: String, + pub project_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + pub step_count: usize, + pub total_tokens: u64, + pub created_at: String, + pub updated_at: String, +} + +/// Thread detail with messages and config. +#[derive(Debug, Clone, serde::Serialize)] +pub struct EngineThreadDetail { + #[serde(flatten)] + pub info: EngineThreadInfo, + pub messages: Vec, + pub max_iterations: usize, + #[serde(skip_serializing_if = "Option::is_none")] + pub completed_at: Option, + pub total_cost_usd: f64, +} + +/// Step summary for thread detail views. +#[derive(Debug, Clone, serde::Serialize)] +pub struct EngineStepInfo { + pub id: String, + pub sequence: usize, + pub status: String, + pub tier: String, + pub action_results_count: usize, + pub tokens_input: u64, + pub tokens_output: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub started_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub completed_at: Option, +} + +/// Project summary. +#[derive(Debug, Clone, serde::Serialize)] +pub struct EngineProjectInfo { + pub id: String, + pub name: String, + pub description: String, + pub created_at: String, +} + +/// Mission summary for list views. +#[derive(Debug, Clone, serde::Serialize)] +pub struct EngineMissionInfo { + pub id: String, + pub name: String, + pub goal: String, + pub status: String, + pub cadence_type: String, + pub thread_count: usize, + #[serde(skip_serializing_if = "Option::is_none")] + pub current_focus: Option, + pub created_at: String, + pub updated_at: String, +} + +/// Mission detail with full strategy and budget info. +#[derive(Debug, Clone, serde::Serialize)] +pub struct EngineMissionDetail { + #[serde(flatten)] + pub info: EngineMissionInfo, + pub cadence: serde_json::Value, + pub approach_history: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub success_criteria: Option, + pub threads_today: u32, + pub max_threads_per_day: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub next_fire_at: Option, + pub thread_ids: Vec, +} + +// ── Engine query functions ─────────────────────────────────── + +fn cadence_type_label(cadence: &ironclaw_engine::types::mission::MissionCadence) -> &'static str { + use ironclaw_engine::types::mission::MissionCadence; + match cadence { + MissionCadence::Cron { .. } => "cron", + MissionCadence::OnEvent { .. } => "event", + MissionCadence::OnSystemEvent { .. } => "system_event", + MissionCadence::Webhook { .. } => "webhook", + MissionCadence::Manual => "manual", + } +} + +fn thread_to_info(t: &ironclaw_engine::Thread) -> EngineThreadInfo { + EngineThreadInfo { + id: t.id.to_string(), + goal: t.goal.clone(), + thread_type: format!("{:?}", t.thread_type), + state: format!("{:?}", t.state), + project_id: t.project_id.to_string(), + parent_id: t.parent_id.map(|id| id.to_string()), + step_count: t.step_count, + total_tokens: t.total_tokens_used, + created_at: t.created_at.to_rfc3339(), + updated_at: t.updated_at.to_rfc3339(), + } +} + +/// List engine threads, optionally filtered by project. +pub async fn list_engine_threads(project_id: Option<&str>) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(Vec::new()); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(Vec::new()); + }; + + let pid = match project_id { + Some(id) => { + let uuid = uuid::Uuid::parse_str(id).map_err(|e| engine_err("parse project_id", e))?; + ironclaw_engine::ProjectId(uuid) + } + None => state.default_project_id, + }; + + let threads = state + .store + .list_threads(pid) + .await + .map_err(|e| engine_err("list threads", e))?; + + Ok(threads.iter().map(thread_to_info).collect()) +} + +/// Get a single engine thread by ID. +pub async fn get_engine_thread(thread_id: &str) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(None); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(None); + }; + + let tid = uuid::Uuid::parse_str(thread_id).map_err(|e| engine_err("parse thread_id", e))?; + let tid = ironclaw_engine::ThreadId(tid); + + let Some(thread) = state + .store + .load_thread(tid) + .await + .map_err(|e| engine_err("load thread", e))? + else { + return Ok(None); + }; + + let messages: Vec = thread + .messages + .iter() + .map(|m| { + serde_json::json!({ + "role": format!("{:?}", m.role), + "content": m.content, + "timestamp": m.timestamp.to_rfc3339(), + }) + }) + .collect(); + + Ok(Some(EngineThreadDetail { + info: thread_to_info(&thread), + messages, + max_iterations: thread.config.max_iterations, + completed_at: thread.completed_at.map(|dt| dt.to_rfc3339()), + total_cost_usd: thread.total_cost_usd, + })) +} + +/// List steps for a thread. +pub async fn list_engine_thread_steps(thread_id: &str) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(Vec::new()); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(Vec::new()); + }; + + let tid = uuid::Uuid::parse_str(thread_id).map_err(|e| engine_err("parse thread_id", e))?; + let steps = state + .store + .load_steps(ironclaw_engine::ThreadId(tid)) + .await + .map_err(|e| engine_err("load steps", e))?; + + Ok(steps + .iter() + .map(|s| EngineStepInfo { + id: s.id.0.to_string(), + sequence: s.sequence, + status: format!("{:?}", s.status), + tier: format!("{:?}", s.tier), + action_results_count: s.action_results.len(), + tokens_input: s.tokens_used.input_tokens, + tokens_output: s.tokens_used.output_tokens, + started_at: Some(s.started_at.to_rfc3339()), + completed_at: s.completed_at.map(|dt| dt.to_rfc3339()), + }) + .collect()) +} + +/// List events for a thread as raw JSON values. +pub async fn list_engine_thread_events(thread_id: &str) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(Vec::new()); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(Vec::new()); + }; + + let tid = uuid::Uuid::parse_str(thread_id).map_err(|e| engine_err("parse thread_id", e))?; + let events = state + .store + .load_events(ironclaw_engine::ThreadId(tid)) + .await + .map_err(|e| engine_err("load events", e))?; + + Ok(events + .iter() + .filter_map(|e| serde_json::to_value(e).ok()) + .collect()) +} + +/// List all projects. +pub async fn list_engine_projects() -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(Vec::new()); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(Vec::new()); + }; + + let projects = state + .store + .list_projects() + .await + .map_err(|e| engine_err("list projects", e))?; + + Ok(projects + .iter() + .map(|p| EngineProjectInfo { + id: p.id.to_string(), + name: p.name.clone(), + description: p.description.clone(), + created_at: p.created_at.to_rfc3339(), + }) + .collect()) +} + +/// Get a single project by ID. +pub async fn get_engine_project(project_id: &str) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(None); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(None); + }; + + let pid = uuid::Uuid::parse_str(project_id).map_err(|e| engine_err("parse project_id", e))?; + let project = state + .store + .load_project(ironclaw_engine::ProjectId(pid)) + .await + .map_err(|e| engine_err("load project", e))?; + + Ok(project.map(|p| EngineProjectInfo { + id: p.id.to_string(), + name: p.name, + description: p.description, + created_at: p.created_at.to_rfc3339(), + })) +} + +/// List missions, optionally filtered by project. +pub async fn list_engine_missions( + project_id: Option<&str>, +) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(Vec::new()); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(Vec::new()); + }; + + let pid = match project_id { + Some(id) => { + let uuid = uuid::Uuid::parse_str(id).map_err(|e| engine_err("parse project_id", e))?; + ironclaw_engine::ProjectId(uuid) + } + None => state.default_project_id, + }; + + let missions = state + .store + .list_missions(pid) + .await + .map_err(|e| engine_err("list missions", e))?; + + Ok(missions + .iter() + .map(|m| EngineMissionInfo { + id: m.id.to_string(), + name: m.name.clone(), + goal: m.goal.clone(), + status: format!("{:?}", m.status), + cadence_type: cadence_type_label(&m.cadence).to_string(), + thread_count: m.thread_history.len(), + current_focus: m.current_focus.clone(), + created_at: m.created_at.to_rfc3339(), + updated_at: m.updated_at.to_rfc3339(), + }) + .collect()) +} + +/// Get a single mission by ID. +pub async fn get_engine_mission(mission_id: &str) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Ok(None); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Ok(None); + }; + + let mid = uuid::Uuid::parse_str(mission_id).map_err(|e| engine_err("parse mission_id", e))?; + let mission = state + .store + .load_mission(ironclaw_engine::MissionId(mid)) + .await + .map_err(|e| engine_err("load mission", e))?; + + let Some(m) = mission else { + return Ok(None); + }; + + let cadence_json = serde_json::to_value(&m.cadence).unwrap_or(serde_json::Value::Null); + + Ok(Some(EngineMissionDetail { + info: EngineMissionInfo { + id: m.id.to_string(), + name: m.name.clone(), + goal: m.goal.clone(), + status: format!("{:?}", m.status), + cadence_type: cadence_type_label(&m.cadence).to_string(), + thread_count: m.thread_history.len(), + current_focus: m.current_focus.clone(), + created_at: m.created_at.to_rfc3339(), + updated_at: m.updated_at.to_rfc3339(), + }, + cadence: cadence_json, + approach_history: m.approach_history.clone(), + success_criteria: m.success_criteria.clone(), + threads_today: m.threads_today, + max_threads_per_day: m.max_threads_per_day, + next_fire_at: m.next_fire_at.map(|dt| dt.to_rfc3339()), + thread_ids: m.thread_history.iter().map(|t| t.to_string()).collect(), + })) +} + +/// Manually fire a mission (spawn a new thread). +pub async fn fire_engine_mission(mission_id: &str, user_id: &str) -> Result, Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Err(engine_err("not initialized", "engine v2 is not running")); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Err(engine_err("not initialized", "engine v2 is not running")); + }; + + let mid = uuid::Uuid::parse_str(mission_id).map_err(|e| engine_err("parse mission_id", e))?; + let mid = ironclaw_engine::MissionId(mid); + + let result = state + .effect_adapter + .mission_manager() + .await + .ok_or_else(|| engine_err("mission", "mission manager not available"))? + .fire_mission(mid, user_id, None) + .await + .map_err(|e| engine_err("fire mission", e))?; + + Ok(result.map(|tid| tid.to_string())) +} + +/// Pause a mission. +pub async fn pause_engine_mission(mission_id: &str) -> Result<(), Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Err(engine_err("not initialized", "engine v2 is not running")); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Err(engine_err("not initialized", "engine v2 is not running")); + }; + + let mid = uuid::Uuid::parse_str(mission_id).map_err(|e| engine_err("parse mission_id", e))?; + state + .effect_adapter + .mission_manager() + .await + .ok_or_else(|| engine_err("mission", "mission manager not available"))? + .pause_mission(ironclaw_engine::MissionId(mid)) + .await + .map_err(|e| engine_err("pause mission", e)) +} + +/// Resume a paused mission. +pub async fn resume_engine_mission(mission_id: &str) -> Result<(), Error> { + let Some(lock) = ENGINE_STATE.get() else { + return Err(engine_err("not initialized", "engine v2 is not running")); + }; + let guard = lock.read().await; + let Some(state) = guard.as_ref() else { + return Err(engine_err("not initialized", "engine v2 is not running")); + }; + + let mid = uuid::Uuid::parse_str(mission_id).map_err(|e| engine_err("parse mission_id", e))?; + state + .effect_adapter + .mission_manager() + .await + .ok_or_else(|| engine_err("mission", "mission manager not available"))? + .resume_mission(ironclaw_engine::MissionId(mid)) + .await + .map_err(|e| engine_err("resume mission", e)) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/channels/web/handlers/engine.rs b/src/channels/web/handlers/engine.rs new file mode 100644 index 00000000..99727f7e --- /dev/null +++ b/src/channels/web/handlers/engine.rs @@ -0,0 +1,143 @@ +//! Engine v2 API handlers — threads, projects, missions. + +use std::sync::Arc; + +use axum::{ + Json, + extract::{Path, State}, + http::StatusCode, +}; + +use crate::channels::web::auth::AuthenticatedUser; +use crate::channels::web::server::GatewayState; +use crate::channels::web::types::*; + +// ── Threads ───────────────────────────────────────────────── + +pub async fn engine_threads_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, +) -> Result, (StatusCode, String)> { + let threads = crate::bridge::list_engine_threads(None) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineThreadListResponse { threads })) +} + +pub async fn engine_thread_detail_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let thread = crate::bridge::get_engine_thread(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? + .ok_or((StatusCode::NOT_FOUND, "Thread not found".to_string()))?; + Ok(Json(EngineThreadDetailResponse { thread })) +} + +pub async fn engine_thread_steps_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let steps = crate::bridge::list_engine_thread_steps(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineStepListResponse { steps })) +} + +pub async fn engine_thread_events_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let events = crate::bridge::list_engine_thread_events(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineEventListResponse { events })) +} + +// ── Projects ──────────────────────────────────────────────── + +pub async fn engine_projects_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, +) -> Result, (StatusCode, String)> { + let projects = crate::bridge::list_engine_projects() + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineProjectListResponse { projects })) +} + +pub async fn engine_project_detail_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let project = crate::bridge::get_engine_project(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? + .ok_or((StatusCode::NOT_FOUND, "Project not found".to_string()))?; + Ok(Json(EngineProjectDetailResponse { project })) +} + +// ── Missions ──────────────────────────────────────────────── + +pub async fn engine_missions_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, +) -> Result, (StatusCode, String)> { + let missions = crate::bridge::list_engine_missions(None) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineMissionListResponse { missions })) +} + +pub async fn engine_mission_detail_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let mission = crate::bridge::get_engine_mission(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? + .ok_or((StatusCode::NOT_FOUND, "Mission not found".to_string()))?; + Ok(Json(EngineMissionDetailResponse { mission })) +} + +pub async fn engine_mission_fire_handler( + State(_state): State>, + AuthenticatedUser(user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let thread_id = crate::bridge::fire_engine_mission(&id, &user.user_id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineMissionFireResponse { + fired: thread_id.is_some(), + thread_id, + })) +} + +pub async fn engine_mission_pause_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + crate::bridge::pause_engine_mission(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineActionResponse { ok: true })) +} + +pub async fn engine_mission_resume_handler( + State(_state): State>, + AuthenticatedUser(_user): AuthenticatedUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + crate::bridge::resume_engine_mission(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok(Json(EngineActionResponse { ok: true })) +} diff --git a/src/channels/web/handlers/mod.rs b/src/channels/web/handlers/mod.rs index 50c7a0b9..a6bf6cbc 100644 --- a/src/channels/web/handlers/mod.rs +++ b/src/channels/web/handlers/mod.rs @@ -2,6 +2,7 @@ //! //! Each module groups related endpoint handlers by domain. +pub mod engine; pub mod jobs; pub mod memory; pub mod routines; diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 26c005d4..4d3b6e42 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -33,6 +33,12 @@ use crate::channels::relay::DEFAULT_RELAY_NAME; use crate::channels::web::auth::{ AuthenticatedUser, MultiAuthState, UserIdentity, auth_middleware, }; +use crate::channels::web::handlers::engine::{ + engine_mission_detail_handler, engine_mission_fire_handler, engine_mission_pause_handler, + engine_mission_resume_handler, engine_missions_handler, engine_project_detail_handler, + engine_projects_handler, engine_thread_detail_handler, engine_thread_events_handler, + engine_thread_steps_handler, engine_threads_handler, +}; use crate::channels::web::handlers::jobs::{ job_files_list_handler, job_files_read_handler, jobs_cancel_handler, jobs_detail_handler, jobs_events_handler, jobs_list_handler, jobs_prompt_handler, jobs_restart_handler, @@ -491,6 +497,42 @@ pub async fn start_server( axum::routing::delete(routines_delete_handler), ) .route("/api/routines/{id}/runs", get(routines_runs_handler)) + // Engine v2 + .route("/api/engine/threads", get(engine_threads_handler)) + .route( + "/api/engine/threads/{id}", + get(engine_thread_detail_handler), + ) + .route( + "/api/engine/threads/{id}/steps", + get(engine_thread_steps_handler), + ) + .route( + "/api/engine/threads/{id}/events", + get(engine_thread_events_handler), + ) + .route("/api/engine/projects", get(engine_projects_handler)) + .route( + "/api/engine/projects/{id}", + get(engine_project_detail_handler), + ) + .route("/api/engine/missions", get(engine_missions_handler)) + .route( + "/api/engine/missions/{id}", + get(engine_mission_detail_handler), + ) + .route( + "/api/engine/missions/{id}/fire", + post(engine_mission_fire_handler), + ) + .route( + "/api/engine/missions/{id}/pause", + post(engine_mission_pause_handler), + ) + .route( + "/api/engine/missions/{id}/resume", + post(engine_mission_resume_handler), + ) // Skills .route("/api/skills", get(skills_list_handler)) .route("/api/skills/search", post(skills_search_handler)) diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index 54288436..a878552d 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -820,6 +820,60 @@ pub struct HealthResponse { pub channel: &'static str, } +// ── Engine v2 response types ──────────────────────────────── + +#[derive(Debug, Serialize)] +pub struct EngineThreadListResponse { + pub threads: Vec, +} + +#[derive(Debug, Serialize)] +pub struct EngineThreadDetailResponse { + pub thread: crate::bridge::EngineThreadDetail, +} + +#[derive(Debug, Serialize)] +pub struct EngineStepListResponse { + pub steps: Vec, +} + +#[derive(Debug, Serialize)] +pub struct EngineEventListResponse { + pub events: Vec, +} + +#[derive(Debug, Serialize)] +pub struct EngineProjectListResponse { + pub projects: Vec, +} + +#[derive(Debug, Serialize)] +pub struct EngineProjectDetailResponse { + pub project: crate::bridge::EngineProjectInfo, +} + +#[derive(Debug, Serialize)] +pub struct EngineMissionListResponse { + pub missions: Vec, +} + +#[derive(Debug, Serialize)] +pub struct EngineMissionDetailResponse { + pub mission: crate::bridge::EngineMissionDetail, +} + +#[derive(Debug, Serialize)] +pub struct EngineMissionFireResponse { + #[serde(skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + pub fired: bool, +} + +#[derive(Debug, Serialize)] +pub struct EngineActionResponse { + pub ok: bool, +} + #[cfg(test)] mod tests { use super::*;