use std::sync::Arc;
use rx4::provider::{
Message, Provider as Rx4Provider, ProviderError as Rx4ProviderError, Role, StreamEvent,
};
use crate::providers::{ChatMessage, ChatRequest, Provider as UnthinkclawProvider};
use crate::tools::{Tool as UnthinkclawTool, ToolSpec};
pub fn chat_message_to_rx4(msg: &ChatMessage) -> Message {
let role = match msg.role.as_str() {
"system" => Role::System,
"user" => Role::User,
"assistant" | "assistant_tool_use" => Role::Assistant,
"tool_result" => Role::Tool,
_ => Role::User,
};
Message {
role,
content: msg.content.clone(),
tool_call_id: msg.tool_use_id.clone(),
tool_calls: Vec::new(),
}
}
pub fn rx4_message_to_chat(msg: &Message) -> ChatMessage {
let role = match msg.role {
Role::System => "system",
Role::User => "user",
Role::Assistant => "assistant",
Role::Tool => "tool_result",
};
ChatMessage {
role: role.to_string(),
content: msg.content.clone(),
tool_use_id: msg.tool_call_id.clone(),
}
}
pub fn chat_messages_to_rx4(messages: &[ChatMessage]) -> Vec<Message> {
messages.iter().map(chat_message_to_rx4).collect()
}
pub fn tool_specs_to_rx4_json(specs: &[ToolSpec]) -> Vec<serde_json::Value> {
specs
.iter()
.map(|s| {
serde_json::json!({
"name": s.name,
"description": s.description,
"parameters": s.parameters,
})
})
.collect()
}
pub struct RotaryProviderAdapter {
inner: Arc<dyn UnthinkclawProvider>,
id: String,
name: String,
}
impl RotaryProviderAdapter {
pub fn new(provider: Arc<dyn UnthinkclawProvider>) -> Self {
let id = provider.name().to_string();
let name = format!("apollo-{}", provider.name());
Self {
inner: provider,
id,
name,
}
}
}
#[async_trait::async_trait]
impl Rx4Provider for RotaryProviderAdapter {
fn id(&self) -> &str {
&self.id
}
fn name(&self) -> &str {
&self.name
}
async fn stream(
&self,
messages: &[Message],
system: &Option<String>,
model: &str,
tools: &[serde_json::Value],
_reasoning_effort: Option<&str>,
) -> Result<rx4::provider::StreamResult, Rx4ProviderError> {
let mut chat_messages: Vec<ChatMessage> = Vec::new();
if let Some(sys) = system {
chat_messages.push(ChatMessage::system(sys));
}
for msg in messages {
chat_messages.push(rx4_message_to_chat(msg));
}
let tool_specs: Vec<ToolSpec> = tools
.iter()
.filter_map(|t| {
let name = t.get("name")?.as_str()?.to_string();
let description = t
.get("description")
.and_then(|d| d.as_str())
.unwrap_or("")
.to_string();
let parameters = t
.get("parameters")
.cloned()
.unwrap_or(serde_json::Value::Null);
Some(ToolSpec {
name,
description,
parameters,
})
})
.collect();
let tool_refs: &[ToolSpec] = if tool_specs.is_empty() {
&[]
} else {
&tool_specs
};
let request = ChatRequest {
messages: &chat_messages,
tools: if tool_refs.is_empty() {
None
} else {
Some(tool_refs)
},
model,
temperature: 0.7,
max_tokens: Some(8192),
};
let response = self
.inner
.chat(&request)
.await
.map_err(|e| Rx4ProviderError::Api(e.to_string()))?;
let text = response.text.unwrap_or_default();
let tool_calls = response.tool_calls;
let events: Vec<Result<StreamEvent, Rx4ProviderError>> = {
let mut evs = Vec::new();
if !text.is_empty() {
evs.push(Ok(StreamEvent::Delta(text)));
}
for tc in tool_calls {
evs.push(Ok(StreamEvent::ToolCall(rx4::ToolCall {
id: tc.id,
name: tc.name,
arguments: tc.arguments,
})));
}
evs.push(Ok(StreamEvent::Done));
evs
};
use futures_util::stream;
Ok(Box::new(Box::pin(stream::iter(events))))
}
}
pub fn register_apollo_tools(registry: &mut rx4::ToolRegistry, tools: &[Arc<dyn UnthinkclawTool>]) {
use rx4::guardrails::classify_tool;
use rx4::{ToolDefinition, ToolEffect, ToolExecuteBox};
for tool in tools {
let spec = tool.spec();
let name = spec.name.clone();
let description = spec.description.clone();
let parameters_json = serde_json::to_string(&spec.parameters).unwrap_or_default();
let tool_clone = Arc::clone(tool);
let execute: ToolExecuteBox = Box::new(move |_ctx, args| {
let tool = Arc::clone(&tool_clone);
Box::pin(async move {
match tool.execute(&args).await {
Ok(result) => rx4::ToolResult {
id: String::new(),
content: result.output,
is_error: result.is_error,
},
Err(e) => rx4::ToolResult {
id: String::new(),
content: crate::redaction::redact_text(&format!("Tool error: {e}")),
is_error: true,
},
}
})
});
let effect = match classify_tool(&name) {
rx4::guardrails::ToolClass::Idempotent => ToolEffect::Read,
rx4::guardrails::ToolClass::Mutating => ToolEffect::Write,
};
registry.register(
ToolDefinition::new_boxed(name, description, parameters_json, execute)
.with_effect(effect),
);
}
}
pub struct RotaryBridgeConfig {
pub provider: Arc<dyn UnthinkclawProvider>,
pub tools: Vec<Arc<dyn UnthinkclawTool>>,
pub system_prompt: String,
pub model: String,
pub workspace: std::path::PathBuf,
pub max_tool_iterations: usize,
}
pub struct RotaryAgentBridge {
agent: rx4::Agent,
messages: Vec<Message>,
}
impl RotaryAgentBridge {
pub fn new(config: RotaryBridgeConfig) -> Self {
let rx4_provider = Arc::new(RotaryProviderAdapter::new(config.provider));
let mut agent = rx4::Agent::new();
agent.set_model(&config.model);
agent.set_system_prompt(&config.system_prompt);
agent.set_provider(rx4_provider);
agent.set_workspace_root(&config.workspace);
agent.max_tool_iterations = config.max_tool_iterations;
let mut tool_registry = rx4::ToolRegistry::new();
register_apollo_tools(&mut tool_registry, &config.tools);
agent.tools = Arc::new(tool_registry);
Self {
agent,
messages: Vec::new(),
}
}
pub fn agent(&self) -> &rx4::Agent {
&self.agent
}
pub fn agent_mut(&mut self) -> &mut rx4::Agent {
&mut self.agent
}
pub fn clear_messages(&mut self) {
self.messages.clear();
self.agent.clear_messages();
}
pub fn message_count(&self) -> usize {
self.messages.len()
}
pub fn set_model(&mut self, model: &str) {
self.agent.set_model(model);
}
pub fn set_system_prompt(&mut self, prompt: &str) {
self.agent.set_system_prompt(prompt);
}
pub fn set_workspace_root(&mut self, path: &std::path::Path) {
self.agent.set_workspace_root(path);
}
pub fn set_scope(&mut self, scope: rx4::Scope) {
self.agent.set_scope(scope);
}
pub fn subscribe(&mut self, callback: impl Fn(&rx4::Event) + Send + Sync + 'static) {
self.agent.subscribe(callback);
}
pub async fn run_prompt(&mut self, prompt: &str) -> anyhow::Result<String> {
let last_response = Arc::new(parking_lot::RwLock::new(String::new()));
let last_response_clone = Arc::clone(&last_response);
self.agent.subscribe(move |event| {
if let rx4::Event::MessageEnd {
content,
role: Role::Assistant,
} = event
{
*last_response_clone.write() = content.clone();
}
});
self.agent.prompt(prompt).await?;
let response = last_response.read().clone();
Ok(response)
}
pub async fn run_prompt_with_history(
&mut self,
prompt: &str,
history: &[ChatMessage],
) -> anyhow::Result<String> {
self.agent.clear_messages();
for msg in history {
let rx4_msg = chat_message_to_rx4(msg);
self.agent.messages.write().push(rx4_msg);
}
self.run_prompt(prompt).await
}
pub fn register_tools(&mut self, tools: &[Arc<dyn UnthinkclawTool>]) {
if let Some(registry) = Arc::get_mut(&mut self.agent.tools) {
register_apollo_tools(registry, tools);
} else {
tracing::warn!("cannot register rx4 tools while the registry is shared");
}
}
pub fn list_tools(&self) -> Vec<String> {
self.agent
.tools
.definitions()
.iter()
.filter_map(|d| {
d.get("name")
.and_then(|n| n.as_str())
.map(|s| s.to_string())
})
.collect()
}
pub fn compact(&mut self, reason: &str) {
self.agent.compact(reason);
}
}
pub fn build_rx4_skill_engine(workspace: &std::path::Path) -> rx4::SkillEngine {
let home = dirs::home_dir().unwrap_or_default();
let managed_dir = workspace.join(".apollo/skills");
let mut engine = rx4::SkillEngine::new(managed_dir);
let openclaw_skills = home.join(".npm-global/lib/node_modules/openclaw/skills");
engine.add_extra_dir(openclaw_skills);
let shared_skills = home.join(".openclaw/workspace/skills");
engine.add_extra_dir(shared_skills);
engine
}
pub fn match_skill_via_rx4(
engine: &rx4::SkillEngine,
user_message: &str,
) -> Option<(String, String)> {
let results = engine.search(user_message);
if results.is_empty() {
return None;
}
let skill = results[0];
Some((skill.name.clone(), skill.instructions.clone()))
}
pub fn discover_skills_via_rx4(workspace: &std::path::Path) -> Vec<crate::skills::Skill> {
let mut engine = build_rx4_skill_engine(workspace);
if engine.load().is_err() {
tracing::warn!("rx4 SkillEngine load failed, returning empty skill list");
return Vec::new();
}
engine
.list()
.into_iter()
.map(|rx4_skill| {
let location = engine.skills_dir().join(format!("{}.json", rx4_skill.id));
crate::skills::Skill {
name: rx4_skill.name.clone(),
description: rx4_skill.description.clone(),
location,
}
})
.collect()
}
pub struct RotaryMemoryBridge {
graph: rx4::GraphMemory,
extractor: rx4::ConversationExtractor,
graph_path: Option<std::path::PathBuf>,
}
impl RotaryMemoryBridge {
pub fn new(workspace: &std::path::Path) -> Self {
let graph = rx4::GraphMemory::from_workspace(workspace);
let graph_path = workspace.join(".apollo/graph_memory.json");
Self {
graph,
extractor: rx4::ConversationExtractor::new(),
graph_path: Some(graph_path),
}
}
pub fn empty() -> Self {
Self {
graph: rx4::GraphMemory::new(),
extractor: rx4::ConversationExtractor::new(),
graph_path: None,
}
}
pub fn extract_conversation(
&mut self,
conversation: &[rx4::graph_memory::ConversationTurn],
) -> rx4::ExtractionResult {
let result = self.extractor.extract(conversation);
for node in &result.nodes {
self.graph.add_node(node.clone());
}
for edge in &result.edges {
let _ = self.graph.add_edge(edge.clone());
}
result
}
pub fn search(&self, query: &str) -> Vec<&rx4::GraphMemoryNode> {
self.graph.search(query)
}
pub fn pagerank(&self) -> Vec<(String, f64)> {
self.graph.pagerank()
}
pub fn stats(&self) -> rx4::graph_memory::GraphStats {
self.graph.stats()
}
pub fn save(&self) -> Result<(), rx4::GraphMemoryError> {
if let Some(path) = &self.graph_path {
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
self.graph.save(path)?;
tracing::debug!("graph memory saved to {:?}", self.graph_path);
}
Ok(())
}
pub fn load(&mut self) -> Result<(), rx4::GraphMemoryError> {
if let Some(path) = &self.graph_path {
if path.exists() {
self.graph = rx4::GraphMemory::load(path)?;
tracing::debug!("graph memory loaded from {:?}", self.graph_path);
}
}
Ok(())
}
pub fn graph(&self) -> &rx4::GraphMemory {
&self.graph
}
pub fn graph_mut(&mut self) -> &mut rx4::GraphMemory {
&mut self.graph
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_chat_message_to_rx4_system() {
let msg = ChatMessage::system("hello");
let rx4_msg = chat_message_to_rx4(&msg);
assert_eq!(rx4_msg.role, Role::System);
assert_eq!(rx4_msg.content, "hello");
}
#[test]
fn test_chat_message_to_rx4_user() {
let msg = ChatMessage::user("test");
let rx4_msg = chat_message_to_rx4(&msg);
assert_eq!(rx4_msg.role, Role::User);
assert_eq!(rx4_msg.content, "test");
}
#[test]
fn test_chat_message_to_rx4_tool_result() {
let msg = ChatMessage::tool_result("tc_123", "result text");
let rx4_msg = chat_message_to_rx4(&msg);
assert_eq!(rx4_msg.role, Role::Tool);
assert_eq!(rx4_msg.content, "result text");
assert_eq!(rx4_msg.tool_call_id.as_deref(), Some("tc_123"));
}
#[test]
fn test_rx4_message_to_chat() {
let msg = Message::assistant("hello back");
let chat_msg = rx4_message_to_chat(&msg);
assert_eq!(chat_msg.role, "assistant");
assert_eq!(chat_msg.content, "hello back");
}
#[test]
fn test_tool_specs_to_rx4_json() {
let specs = vec![ToolSpec {
name: "shell".to_string(),
description: "Run shell commands".to_string(),
parameters: serde_json::json!({"type": "object"}),
}];
let json = tool_specs_to_rx4_json(&specs);
assert_eq!(json.len(), 1);
assert_eq!(json[0]["name"], "shell");
}
#[test]
fn test_roundtrip_translation() {
let original = ChatMessage::user("roundtrip test");
let rx4_msg = chat_message_to_rx4(&original);
let back = rx4_message_to_chat(&rx4_msg);
assert_eq!(back.role, "user");
assert_eq!(back.content, "roundtrip test");
}
#[test]
fn test_build_rx4_skill_engine() {
let tmp = tempfile::tempdir().unwrap();
let engine = build_rx4_skill_engine(tmp.path());
assert!(
engine.skills_dir().exists()
|| engine.skills_dir() == tmp.path().join(".apollo/skills")
);
}
#[test]
fn test_match_skill_via_rx4_empty() {
let tmp = tempfile::tempdir().unwrap();
let mut engine = build_rx4_skill_engine(tmp.path());
let _ = engine.load();
let result = match_skill_via_rx4(&engine, "test query");
assert!(result.is_none());
}
#[test]
fn test_rotary_memory_bridge_empty() {
let bridge = RotaryMemoryBridge::empty();
let stats = bridge.stats();
assert_eq!(stats.node_count, 0);
}
#[test]
fn test_rotary_memory_bridge_search_empty() {
let bridge = RotaryMemoryBridge::empty();
let results = bridge.search("test");
assert!(results.is_empty());
}
#[test]
fn test_rotary_memory_bridge_save_load() {
let tmp = tempfile::tempdir().unwrap();
let mut bridge = RotaryMemoryBridge::new(tmp.path());
bridge.save().unwrap();
bridge.load().unwrap();
assert_eq!(bridge.stats().node_count, 0);
}
}