gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-25 14:53:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(engine): complete v2 side-by-side integration with gateway API

Browse Source 
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) <[email protected]>
This commit is contained in:
[email protected]
2026-03-26 18:19:10 -07:00
co-authored by Claude Opus 4.6
parent b07cc5e144
commit 098addce27
12 changed files with 1196 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ironclaw_common/src/event.rs
+46
View File
@@ -206,6 +206,33 @@ pub enum AppEvent {
narrative: String,
decisions: Vec<ToolDecisionDto>,
},
// ── 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<String>,
},
/// 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 {
crates/ironclaw_engine/src/executor/loop_engine.rs
+1 -1
View File
@@ -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};
crates/ironclaw_engine/src/executor/orchestrator.rs
+73 -75
View File
@@ -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<dyn Store>,
project_id: ProjectId,
) {
pub async fn reset_orchestrator_failures(store: &Arc<dyn Store>, 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<dyn Store>)), project_id).await;
let (code, version) = load_orchestrator(Some(&(store as Arc<dyn Store>)), 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<dyn Store>)), 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<dyn Store>)), 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<dyn Store>)), project_id).await;
let (code, version) = load_orchestrator(Some(&(store as Arc<dyn Store>)), 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<dyn Store>)), 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<dyn Store>)), 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]
crates/ironclaw_engine/src/runtime/conversation.rs
+56
View File
@@ -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());
}
}
src/agent/agent_loop.rs
+28 -1
View File
@@ -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.
_ => {}
}
}
src/bridge/effect_adapter.rs
+5 -1
View File
@@ -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<Arc<ironclaw_engine::MissionManager>> {
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!(
src/bridge/mod.rs
+28 -1
View File
@@ -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,
};
src/bridge/router.rs
+719 -108
View File
@@ -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<Option<PendingApproval>, 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<dyn Store>,
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<Option<String>, 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<Option<String>, 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<Option<String>, 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<Option<String>, 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<Option<String>, 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<Option<String>, 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<String>,
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<serde_json::Value>,
pub max_iterations: usize,
#[serde(skip_serializing_if = "Option::is_none")]
pub completed_at: Option<String>,
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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub completed_at: Option<String>,
}
/// 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<String>,
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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub success_criteria: Option<String>,
pub threads_today: u32,
pub max_threads_per_day: u32,
#[serde(skip_serializing_if = "Option::is_none")]
pub next_fire_at: Option<String>,
pub thread_ids: Vec<String>,
}
// ── 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<Vec<EngineThreadInfo>, 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<Option<EngineThreadDetail>, 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<serde_json::Value> = 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<Vec<EngineStepInfo>, 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<Vec<serde_json::Value>, 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<Vec<EngineProjectInfo>, 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<Option<EngineProjectInfo>, 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<Vec<EngineMissionInfo>, 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<Option<EngineMissionDetail>, 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<Option<String>, 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::*;
src/channels/web/handlers/engine.rs
+143
View File
@@ -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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
) -> Result<Json<EngineThreadListResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineThreadDetailResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineStepListResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineEventListResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
) -> Result<Json<EngineProjectListResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineProjectDetailResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
) -> Result<Json<EngineMissionListResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineMissionDetailResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineMissionFireResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineActionResponse>, (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<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<EngineActionResponse>, (StatusCode, String)> {
crate::bridge::resume_engine_mission(&id)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
Ok(Json(EngineActionResponse { ok: true }))
}
src/channels/web/handlers/mod.rs
+1
View File
@@ -2,6 +2,7 @@
//!
//! Each module groups related endpoint handlers by domain.
pub mod engine;
pub mod jobs;
pub mod memory;
pub mod routines;
src/channels/web/server.rs
+42
View File
@@ -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))
src/channels/web/types.rs
+54
View File
@@ -820,6 +820,60 @@ pub struct HealthResponse {
pub channel: &'static str,
}
// ── Engine v2 response types ────────────────────────────────
#[derive(Debug, Serialize)]
pub struct EngineThreadListResponse {
pub threads: Vec<crate::bridge::EngineThreadInfo>,
}
#[derive(Debug, Serialize)]
pub struct EngineThreadDetailResponse {
pub thread: crate::bridge::EngineThreadDetail,
}
#[derive(Debug, Serialize)]
pub struct EngineStepListResponse {
pub steps: Vec<crate::bridge::EngineStepInfo>,
}
#[derive(Debug, Serialize)]
pub struct EngineEventListResponse {
pub events: Vec<serde_json::Value>,
}
#[derive(Debug, Serialize)]
pub struct EngineProjectListResponse {
pub projects: Vec<crate::bridge::EngineProjectInfo>,
}
#[derive(Debug, Serialize)]
pub struct EngineProjectDetailResponse {
pub project: crate::bridge::EngineProjectInfo,
}
#[derive(Debug, Serialize)]
pub struct EngineMissionListResponse {
pub missions: Vec<crate::bridge::EngineMissionInfo>,
}
#[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<String>,
pub fired: bool,
}
#[derive(Debug, Serialize)]
pub struct EngineActionResponse {
pub ok: bool,
}
#[cfg(test)]
mod tests {
use super::*;