From ac4ced02ae27ca0d93299f1e4044a756eb8b0db7 Mon Sep 17 00:00:00 2001 From: "ilblackdragon@gmail.com" Date: Sun, 22 Mar 2026 00:22:13 -0700 Subject: [PATCH] =?UTF-8?q?feat(engine):=20Phase=206=20=E2=80=94=20bridge?= =?UTF-8?q?=20adapters=20for=20main=20crate=20integration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Strategy C parallel deployment: when ENGINE_V2=true env var is set, user messages route through the engine instead of the existing agentic loop. All existing behavior is unchanged when the flag is off. Bridge module (src/bridge/): - LlmBridgeAdapter: wraps LlmProvider as engine LlmBackend, converts ThreadMessage↔ChatMessage, ActionDef↔ToolDefinition, depth-based model routing (primary vs cheap_llm) - EffectBridgeAdapter: wraps ToolRegistry+SafetyLayer as EffectExecutor, routes tool calls through existing execute_tool_with_safety pipeline - InMemoryStore: HashMap-backed Store impl (no DB tables needed yet) - EngineRouter: is_engine_v2_enabled() + handle_with_engine() that builds engine from Agent deps and processes messages end-to-end Integration touchpoint (4 lines in agent_loop.rs): After hook processing, before session resolution, check ENGINE_V2 flag and route UserInput through the engine path. Accessor visibility widened: llm(), cheap_llm(), safety(), tools() changed from pub(super) to pub(crate) for bridge access. 85 engine tests + main crate clippy clean. Co-Authored-By: Claude Opus 4.6 (1M context) --- Cargo.lock | 1 + Cargo.toml | 1 + src/agent/agent_loop.rs | 15 ++- src/bridge/effect_adapter.rs | 86 ++++++++++++++++ src/bridge/llm_adapter.rs | 160 ++++++++++++++++++++++++++++++ src/bridge/mod.rs | 12 +++ src/bridge/router.rs | 163 ++++++++++++++++++++++++++++++ src/bridge/store_adapter.rs | 187 +++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + 9 files changed, 622 insertions(+), 4 deletions(-) create mode 100644 src/bridge/effect_adapter.rs create mode 100644 src/bridge/llm_adapter.rs create mode 100644 src/bridge/mod.rs create mode 100644 src/bridge/router.rs create mode 100644 src/bridge/store_adapter.rs diff --git a/Cargo.lock b/Cargo.lock index dda3b37e..4a58494b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3651,6 +3651,7 @@ dependencies = [ "hyper-util", "iana-time-zone", "insta", + "ironclaw_engine", "ironclaw_safety", "json5", "libsql", diff --git a/Cargo.toml b/Cargo.toml index 2895b7ea..fc132cf6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -101,6 +101,7 @@ tower-http = { version = "0.6", features = ["trace", "cors", "set-header"] } cron = "0.13" # Safety/sanitization +ironclaw_engine = { path = "crates/ironclaw_engine" } ironclaw_safety = { path = "crates/ironclaw_safety", version = "0.1.0" } regex = "1" aho-corasick = "1" diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index a0e8278f..b302cdcb 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -283,20 +283,20 @@ impl Agent { self.deps.store.as_ref() } - pub(super) fn llm(&self) -> &Arc { + pub(crate) fn llm(&self) -> &Arc { &self.deps.llm } /// Get the cheap/fast LLM provider, falling back to the main one. - pub(super) fn cheap_llm(&self) -> &Arc { + pub(crate) fn cheap_llm(&self) -> &Arc { self.deps.cheap_llm.as_ref().unwrap_or(&self.deps.llm) } - pub(super) fn safety(&self) -> &Arc { + pub(crate) fn safety(&self) -> &Arc { &self.deps.safety } - pub(super) fn tools(&self) -> &Arc { + pub(crate) fn tools(&self) -> &Arc { &self.deps.tools } @@ -1003,6 +1003,13 @@ impl Agent { } } + // Engine V2 routing (Strategy C: parallel deployment) + if let Submission::UserInput { ref content } = submission + && crate::bridge::is_engine_v2_enabled() + { + return crate::bridge::handle_with_engine(self, message, content).await; + } + // Hydrate thread from DB if it's a historical thread not in memory if let Some(external_thread_id) = message.conversation_scope() { tracing::trace!( diff --git a/src/bridge/effect_adapter.rs b/src/bridge/effect_adapter.rs new file mode 100644 index 00000000..972ecba9 --- /dev/null +++ b/src/bridge/effect_adapter.rs @@ -0,0 +1,86 @@ +//! Effect bridge adapter — wraps `ToolRegistry` + `SafetyLayer` as `ironclaw_engine::EffectExecutor`. + +use std::sync::Arc; +use std::time::Duration; + +use ironclaw_engine::{ + ActionDef, ActionResult, CapabilityLease, EffectExecutor, EngineError, ThreadExecutionContext, +}; + +use crate::context::JobContext; +use crate::safety::SafetyLayer; +use crate::tools::ToolRegistry; + +/// Wraps the existing tool pipeline to implement the engine's `EffectExecutor`. +pub struct EffectBridgeAdapter { + tools: Arc, + safety: Arc, +} + +impl EffectBridgeAdapter { + pub fn new(tools: Arc, safety: Arc) -> Self { + Self { tools, safety } + } +} + +#[async_trait::async_trait] +impl EffectExecutor for EffectBridgeAdapter { + async fn execute_action( + &self, + action_name: &str, + parameters: serde_json::Value, + _lease: &CapabilityLease, + context: &ThreadExecutionContext, + ) -> Result { + // Build a minimal JobContext for tool execution + let job_ctx = JobContext::with_user( + &context.user_id, + "engine_v2", + format!("Thread {}", context.thread_id), + ); + + // Execute through the existing tool pipeline + let result = crate::tools::execute::execute_tool_with_safety( + &self.tools, + &self.safety, + action_name, + ¶meters, + &job_ctx, + ) + .await; + + match result { + Ok(output) => Ok(ActionResult { + call_id: String::new(), // Caller fills this in + action_name: action_name.to_string(), + output: serde_json::json!(output), + is_error: false, + duration: Duration::from_millis(1), // TODO: measure actual duration + }), + Err(e) => Ok(ActionResult { + call_id: String::new(), + action_name: action_name.to_string(), + output: serde_json::json!({"error": e.to_string()}), + is_error: true, + duration: Duration::ZERO, + }), + } + } + + async fn available_actions( + &self, + _leases: &[CapabilityLease], + ) -> Result, EngineError> { + let tool_defs = self.tools.tool_definitions().await; + Ok(tool_defs + .into_iter() + .map(|td| ActionDef { + name: td.name, + description: td.description, + parameters_schema: td.parameters, + effects: vec![], // Effect classification happens at the engine level + requires_approval: false, + }) + .collect()) + } +} diff --git a/src/bridge/llm_adapter.rs b/src/bridge/llm_adapter.rs new file mode 100644 index 00000000..b190f0af --- /dev/null +++ b/src/bridge/llm_adapter.rs @@ -0,0 +1,160 @@ +//! LLM bridge adapter — wraps `LlmProvider` as `ironclaw_engine::LlmBackend`. + +use std::sync::Arc; + +use ironclaw_engine::{ + ActionDef, EngineError, LlmBackend, LlmCallConfig, LlmOutput, LlmResponse, ThreadMessage, + TokenUsage, +}; + +use crate::llm::{ + ChatMessage, LlmProvider, Role, ToolCall, ToolCompletionRequest, ToolDefinition, +}; + +/// Wraps an existing `LlmProvider` to implement the engine's `LlmBackend` trait. +pub struct LlmBridgeAdapter { + provider: Arc, + /// Optional cheaper provider for sub-calls (depth > 0). + cheap_provider: Option>, +} + +impl LlmBridgeAdapter { + pub fn new( + provider: Arc, + cheap_provider: Option>, + ) -> Self { + Self { + provider, + cheap_provider, + } + } + + fn provider_for_depth(&self, depth: u32) -> &Arc { + if depth > 0 { + self.cheap_provider.as_ref().unwrap_or(&self.provider) + } else { + &self.provider + } + } +} + +#[async_trait::async_trait] +impl LlmBackend for LlmBridgeAdapter { + async fn complete( + &self, + messages: &[ThreadMessage], + actions: &[ActionDef], + config: &LlmCallConfig, + ) -> Result { + let provider = self.provider_for_depth(config.depth); + + // Convert messages + let chat_messages: Vec = messages.iter().map(thread_msg_to_chat).collect(); + + // Convert actions to tool definitions + let tools: Vec = if config.force_text { + vec![] // No tools when forcing text + } else { + actions.iter().map(action_def_to_tool_def).collect() + }; + + // Build request + let mut request = ToolCompletionRequest::new(chat_messages, tools); + if let Some(max_tokens) = config.max_tokens { + request = request.with_max_tokens(max_tokens); + } + if let Some(temp) = config.temperature { + request = request.with_temperature(temp); + } + if config.force_text { + request = request.with_tool_choice("none"); + } + request.metadata = config.metadata.clone(); + + // Call provider + let response = provider + .complete_with_tools(request) + .await + .map_err(|e| EngineError::Llm { + reason: e.to_string(), + })?; + + // Convert response + let llm_response = if !response.tool_calls.is_empty() { + LlmResponse::ActionCalls { + calls: response + .tool_calls + .iter() + .map(|tc| ironclaw_engine::ActionCall { + id: tc.id.clone(), + action_name: tc.name.clone(), + parameters: tc.arguments.clone(), + }) + .collect(), + content: response.content.clone(), + } + } else { + LlmResponse::Text(response.content.unwrap_or_default()) + }; + + Ok(LlmOutput { + response: llm_response, + usage: TokenUsage { + input_tokens: u64::from(response.input_tokens), + output_tokens: u64::from(response.output_tokens), + cache_read_tokens: u64::from(response.cache_read_input_tokens), + cache_write_tokens: u64::from(response.cache_creation_input_tokens), + }, + }) + } + + fn model_name(&self) -> &str { + self.provider.model_name() + } +} + +// ── Conversion helpers ────────────────────────────────────── + +fn thread_msg_to_chat(msg: &ThreadMessage) -> ChatMessage { + use ironclaw_engine::MessageRole; + + let role = match msg.role { + MessageRole::System => Role::System, + MessageRole::User => Role::User, + MessageRole::Assistant => Role::Assistant, + MessageRole::ActionResult => Role::Tool, + }; + + let mut chat = ChatMessage { + role, + content: msg.content.clone(), + content_parts: Vec::new(), + tool_call_id: msg.action_call_id.clone(), + name: msg.action_name.clone(), + tool_calls: None, + }; + + // Convert action calls if present (assistant message with tool calls) + if let Some(ref calls) = msg.action_calls { + chat.tool_calls = Some( + calls + .iter() + .map(|c| ToolCall { + id: c.id.clone(), + name: c.action_name.clone(), + arguments: c.parameters.clone(), + }) + .collect(), + ); + } + + chat +} + +fn action_def_to_tool_def(action: &ActionDef) -> ToolDefinition { + ToolDefinition { + name: action.name.clone(), + description: action.description.clone(), + parameters: action.parameters_schema.clone(), + } +} diff --git a/src/bridge/mod.rs b/src/bridge/mod.rs new file mode 100644 index 00000000..f412b7b8 --- /dev/null +++ b/src/bridge/mod.rs @@ -0,0 +1,12 @@ +//! Engine v2 bridge — connects `ironclaw_engine` to existing infrastructure. +//! +//! Strategy C: parallel deployment. When `ENGINE_V2=true`, user messages +//! route through the engine instead of the existing agentic loop. All +//! existing behavior is unchanged when the flag is off. + +mod effect_adapter; +mod llm_adapter; +mod router; +mod store_adapter; + +pub use router::{handle_with_engine, is_engine_v2_enabled}; diff --git a/src/bridge/router.rs b/src/bridge/router.rs new file mode 100644 index 00000000..304139e5 --- /dev/null +++ b/src/bridge/router.rs @@ -0,0 +1,163 @@ +//! Engine v2 router — handles user messages via the engine when enabled. + +use std::sync::Arc; + +use tracing::{debug, info}; + +use ironclaw_engine::{ + Capability, CapabilityRegistry, LeaseManager, PolicyEngine, Project, Store, + ThreadConfig, ThreadManager, ThreadOutcome, ThreadType, +}; + +use crate::agent::Agent; +use crate::bridge::effect_adapter::EffectBridgeAdapter; +use crate::bridge::llm_adapter::LlmBridgeAdapter; +use crate::bridge::store_adapter::InMemoryStore; +use crate::channels::IncomingMessage; +use crate::error::Error; + +/// Check if the engine v2 is enabled via `ENGINE_V2=true` environment variable. +pub fn is_engine_v2_enabled() -> bool { + std::env::var("ENGINE_V2") + .map(|v| v == "true" || v == "1") + .unwrap_or(false) +} + +/// Handle a user message through the engine v2 pipeline. +/// +/// This is the engine-based equivalent of `Agent::process_user_input()`. +/// It builds bridge adapters from the agent's existing dependencies, +/// creates an engine `ThreadManager`, spawns a thread, and waits for +/// the result. +pub async fn handle_with_engine( + agent: &Agent, + message: &IncomingMessage, + content: &str, +) -> Result, Error> { + info!( + user_id = %message.user_id, + channel = %message.channel, + "engine v2: handling message" + ); + + // Build bridge adapters from agent's existing dependencies + let llm_adapter = Arc::new(LlmBridgeAdapter::new( + agent.llm().clone(), + Some(agent.cheap_llm().clone()), + )); + + let effect_adapter = Arc::new(EffectBridgeAdapter::new( + agent.tools().clone(), + agent.safety().clone(), + )); + + let store = Arc::new(InMemoryStore::new()); + + // Build capability registry from available tools + let mut capabilities = CapabilityRegistry::new(); + let tool_defs = agent.tools().tool_definitions().await; + if !tool_defs.is_empty() { + capabilities.register(Capability { + name: "tools".into(), + description: "Available tools".into(), + actions: tool_defs + .into_iter() + .map(|td| ironclaw_engine::ActionDef { + name: td.name, + description: td.description, + parameters_schema: td.parameters, + effects: vec![], + requires_approval: false, + }) + .collect(), + knowledge: vec![], + policies: vec![], + }); + } + + let leases = Arc::new(LeaseManager::new()); + let policy = Arc::new(PolicyEngine::new()); + + // Create thread manager + let thread_manager = Arc::new(ThreadManager::new( + llm_adapter, + effect_adapter, + store.clone(), + Arc::new(capabilities), + leases, + policy, + )); + + // Create a default project for this session + let project = Project::new( + format!("{}:{}", message.channel, message.user_id), + "Auto-created project", + ); + 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}"), + }) + })?; + + // Spawn a thread for this message + let config = ThreadConfig::default(); + let thread_id = thread_manager + .spawn_thread( + content, + ThreadType::Foreground, + project_id, + config, + None, + &message.user_id, + ) + .await + .map_err(|e| { + crate::error::Error::from(crate::error::JobError::ContextError { + id: uuid::Uuid::nil(), + reason: format!("engine v2 spawn error: {e}"), + }) + })?; + + debug!(thread_id = %thread_id, "engine v2: thread spawned, waiting for completion"); + + // Wait for the thread to complete + let 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}"), + }) + })?; + + // Convert outcome to response + match outcome { + ThreadOutcome::Completed { response } => { + debug!(thread_id = %thread_id, "engine v2: thread completed"); + Ok(response) + } + ThreadOutcome::Stopped => { + debug!(thread_id = %thread_id, "engine v2: thread stopped"); + Ok(Some("Thread was stopped.".into())) + } + ThreadOutcome::MaxIterations => { + debug!(thread_id = %thread_id, "engine v2: max iterations"); + Ok(Some("Reached maximum iterations without completing.".into())) + } + ThreadOutcome::Failed { error } => { + debug!(thread_id = %thread_id, error = %error, "engine v2: thread failed"); + Ok(Some(format!("Error: {error}"))) + } + ThreadOutcome::NeedApproval { + action_name, + call_id: _, + parameters: _, + } => { + // Phase 6: approval flow not yet wired — return as message + debug!(thread_id = %thread_id, action = %action_name, "engine v2: approval needed (not yet supported)"); + Ok(Some(format!( + "Action '{action_name}' requires approval (engine v2 approval flow not yet implemented)" + ))) + } + } +} diff --git a/src/bridge/store_adapter.rs b/src/bridge/store_adapter.rs new file mode 100644 index 00000000..ad6e3f39 --- /dev/null +++ b/src/bridge/store_adapter.rs @@ -0,0 +1,187 @@ +//! In-memory store adapter — implements `ironclaw_engine::Store` without database tables. +//! +//! Phase 6: threads and state live in memory during execution. Persistent +//! storage comes in Phase 7 when we add database migrations. + +use std::collections::HashMap; + +use tokio::sync::RwLock; + +use ironclaw_engine::{ + CapabilityLease, DocId, EngineError, LeaseId, MemoryDoc, Project, ProjectId, Step, Thread, + ThreadEvent, ThreadId, ThreadState, Store, +}; + +/// In-memory implementation of the engine's `Store` trait. +/// +/// All state is discarded when the agent process restarts. This is +/// sufficient for Phase 6 (proving the engine works end-to-end). +pub struct InMemoryStore { + threads: RwLock>, + steps: RwLock>>, + events: RwLock>>, + projects: RwLock>, + docs: RwLock>, + leases: RwLock>, +} + +impl InMemoryStore { + pub fn new() -> Self { + Self { + threads: RwLock::new(HashMap::new()), + steps: RwLock::new(HashMap::new()), + events: RwLock::new(HashMap::new()), + projects: RwLock::new(HashMap::new()), + docs: RwLock::new(HashMap::new()), + leases: RwLock::new(HashMap::new()), + } + } +} + +impl Default for InMemoryStore { + fn default() -> Self { + Self::new() + } +} + +#[async_trait::async_trait] +impl Store for InMemoryStore { + // ── Thread ────────────────────────────────────────────── + + async fn save_thread(&self, thread: &Thread) -> Result<(), EngineError> { + self.threads.write().await.insert(thread.id, thread.clone()); + Ok(()) + } + + async fn load_thread(&self, id: ThreadId) -> Result, EngineError> { + Ok(self.threads.read().await.get(&id).cloned()) + } + + async fn list_threads(&self, project_id: ProjectId) -> Result, EngineError> { + Ok(self + .threads + .read() + .await + .values() + .filter(|t| t.project_id == project_id) + .cloned() + .collect()) + } + + async fn update_thread_state( + &self, + id: ThreadId, + state: ThreadState, + ) -> Result<(), EngineError> { + if let Some(thread) = self.threads.write().await.get_mut(&id) { + thread.state = state; + } + Ok(()) + } + + // ── Step ──────────────────────────────────────────────── + + async fn save_step(&self, step: &Step) -> Result<(), EngineError> { + self.steps + .write() + .await + .entry(step.thread_id) + .or_default() + .push(step.clone()); + Ok(()) + } + + async fn load_steps(&self, thread_id: ThreadId) -> Result, EngineError> { + Ok(self + .steps + .read() + .await + .get(&thread_id) + .cloned() + .unwrap_or_default()) + } + + // ── Event ─────────────────────────────────────────────── + + async fn append_events(&self, events: &[ThreadEvent]) -> Result<(), EngineError> { + let mut store = self.events.write().await; + for event in events { + store.entry(event.thread_id).or_default().push(event.clone()); + } + Ok(()) + } + + async fn load_events(&self, thread_id: ThreadId) -> Result, EngineError> { + Ok(self + .events + .read() + .await + .get(&thread_id) + .cloned() + .unwrap_or_default()) + } + + // ── Project ───────────────────────────────────────────── + + async fn save_project(&self, project: &Project) -> Result<(), EngineError> { + self.projects + .write() + .await + .insert(project.id, project.clone()); + Ok(()) + } + + async fn load_project(&self, id: ProjectId) -> Result, EngineError> { + Ok(self.projects.read().await.get(&id).cloned()) + } + + // ── MemoryDoc ─────────────────────────────────────────── + + async fn save_memory_doc(&self, doc: &MemoryDoc) -> Result<(), EngineError> { + self.docs.write().await.insert(doc.id, doc.clone()); + Ok(()) + } + + async fn load_memory_doc(&self, id: DocId) -> Result, EngineError> { + Ok(self.docs.read().await.get(&id).cloned()) + } + + async fn list_memory_docs(&self, project_id: ProjectId) -> Result, EngineError> { + Ok(self + .docs + .read() + .await + .values() + .filter(|d| d.project_id == project_id) + .cloned() + .collect()) + } + + // ── Lease ─────────────────────────────────────────────── + + async fn save_lease(&self, lease: &CapabilityLease) -> Result<(), EngineError> { + self.leases.write().await.insert(lease.id, lease.clone()); + Ok(()) + } + + async fn load_active_leases( + &self, + thread_id: ThreadId, + ) -> Result, EngineError> { + Ok(self + .leases + .read() + .await + .values() + .filter(|l| l.thread_id == thread_id && l.is_valid()) + .cloned() + .collect()) + } + + async fn revoke_lease(&self, lease_id: LeaseId, _reason: &str) -> Result<(), EngineError> { + if let Some(lease) = self.leases.write().await.get_mut(&lease_id) { + lease.revoked = true; + } + Ok(()) + } +} diff --git a/src/lib.rs b/src/lib.rs index c87a31b2..656b51ae 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,6 +42,7 @@ pub mod agent; pub mod app; pub mod boot_screen; pub mod bootstrap; +pub mod bridge; pub mod channels; pub mod cli; pub mod config;