1use std::sync::Arc;
21
22use rx4::provider::{
23 Message, Provider as Rx4Provider, ProviderError as Rx4ProviderError, Role, StreamEvent,
24};
25
26use crate::providers::{ChatMessage, ChatRequest, Provider as UnthinkclawProvider};
27use crate::tools::{Tool as UnthinkclawTool, ToolSpec};
28
29pub fn chat_message_to_rx4(msg: &ChatMessage) -> Message {
33 let role = match msg.role.as_str() {
34 "system" => Role::System,
35 "user" => Role::User,
36 "assistant" | "assistant_tool_use" => Role::Assistant,
37 "tool_result" => Role::Tool,
38 _ => Role::User,
39 };
40 Message {
41 role,
42 content: msg.content.clone(),
43 tool_call_id: msg.tool_use_id.clone(),
44 }
45}
46
47pub fn rx4_message_to_chat(msg: &Message) -> ChatMessage {
49 let role = match msg.role {
50 Role::System => "system",
51 Role::User => "user",
52 Role::Assistant => "assistant",
53 Role::Tool => "tool_result",
54 };
55 ChatMessage {
56 role: role.to_string(),
57 content: msg.content.clone(),
58 tool_use_id: msg.tool_call_id.clone(),
59 }
60}
61
62pub fn chat_messages_to_rx4(messages: &[ChatMessage]) -> Vec<Message> {
64 messages.iter().map(chat_message_to_rx4).collect()
65}
66
67pub fn tool_specs_to_rx4_json(specs: &[ToolSpec]) -> Vec<serde_json::Value> {
69 specs
70 .iter()
71 .map(|s| {
72 serde_json::json!({
73 "name": s.name,
74 "description": s.description,
75 "parameters": s.parameters,
76 })
77 })
78 .collect()
79}
80
81pub struct RotaryProviderAdapter {
91 inner: Arc<dyn UnthinkclawProvider>,
92 id: String,
93 name: String,
94}
95
96impl RotaryProviderAdapter {
97 pub fn new(provider: Arc<dyn UnthinkclawProvider>) -> Self {
98 let id = provider.name().to_string();
99 let name = format!("apollo-{}", provider.name());
100 Self {
101 inner: provider,
102 id,
103 name,
104 }
105 }
106}
107
108#[async_trait::async_trait]
109impl Rx4Provider for RotaryProviderAdapter {
110 fn id(&self) -> &str {
111 &self.id
112 }
113
114 fn name(&self) -> &str {
115 &self.name
116 }
117
118 async fn stream(
119 &self,
120 messages: &[Message],
121 system: &Option<String>,
122 model: &str,
123 tools: &[serde_json::Value],
124 ) -> Result<rx4::provider::StreamResult, Rx4ProviderError> {
125 let mut chat_messages: Vec<ChatMessage> = Vec::new();
127
128 if let Some(sys) = system {
130 chat_messages.push(ChatMessage::system(sys));
131 }
132
133 for msg in messages {
134 chat_messages.push(rx4_message_to_chat(msg));
135 }
136
137 let tool_specs: Vec<ToolSpec> = tools
139 .iter()
140 .filter_map(|t| {
141 let name = t.get("name")?.as_str()?.to_string();
142 let description = t
143 .get("description")
144 .and_then(|d| d.as_str())
145 .unwrap_or("")
146 .to_string();
147 let parameters = t
148 .get("parameters")
149 .cloned()
150 .unwrap_or(serde_json::Value::Null);
151 Some(ToolSpec {
152 name,
153 description,
154 parameters,
155 })
156 })
157 .collect();
158
159 let tool_refs: &[ToolSpec] = if tool_specs.is_empty() {
160 &[]
161 } else {
162 &tool_specs
165 };
166
167 let request = ChatRequest {
168 messages: &chat_messages,
169 tools: if tool_refs.is_empty() {
170 None
171 } else {
172 Some(tool_refs)
173 },
174 model,
175 temperature: 0.7,
176 max_tokens: Some(8192),
177 };
178
179 let response = self
180 .inner
181 .chat(&request)
182 .await
183 .map_err(|e| Rx4ProviderError::Api(e.to_string()))?;
184
185 let text = response.text.unwrap_or_default();
187 let tool_calls = response.tool_calls;
188
189 let events: Vec<Result<StreamEvent, Rx4ProviderError>> = {
191 let mut evs = Vec::new();
192 if !text.is_empty() {
193 evs.push(Ok(StreamEvent::Delta(text)));
194 }
195 for tc in tool_calls {
196 evs.push(Ok(StreamEvent::ToolCall(rx4::ToolCall {
197 id: tc.id,
198 name: tc.name,
199 arguments: tc.arguments,
200 })));
201 }
202 evs.push(Ok(StreamEvent::Done));
203 evs
204 };
205
206 use futures_util::stream;
208 Ok(Box::new(Box::pin(stream::iter(events))))
209 }
210}
211
212pub fn register_apollo_tools(registry: &mut rx4::ToolRegistry, tools: &[Arc<dyn UnthinkclawTool>]) {
224 use rx4::guardrails::classify_tool;
225 use rx4::{ToolDefinition, ToolEffect, ToolExecuteBox};
226
227 for tool in tools {
228 let spec = tool.spec();
229 let name = spec.name.clone();
230 let description = spec.description.clone();
231 let parameters_json = serde_json::to_string(&spec.parameters).unwrap_or_default();
232
233 let tool_clone = Arc::clone(tool);
234 let execute: ToolExecuteBox = Box::new(move |_ctx, args| {
235 let tool = Arc::clone(&tool_clone);
236 Box::pin(async move {
237 match tool.execute(&args).await {
238 Ok(result) => rx4::ToolResult {
239 id: String::new(),
240 content: result.output,
241 is_error: result.is_error,
242 },
243 Err(e) => rx4::ToolResult {
244 id: String::new(),
245 content: crate::redaction::redact_text(&format!("Tool error: {e}")),
246 is_error: true,
247 },
248 }
249 })
250 });
251
252 let effect = match classify_tool(&name) {
253 rx4::guardrails::ToolClass::Idempotent => ToolEffect::Read,
254 rx4::guardrails::ToolClass::Mutating => ToolEffect::Write,
255 };
256
257 registry.register(
258 ToolDefinition::new_boxed(name, description, parameters_json, execute)
259 .with_effect(effect),
260 );
261 }
262}
263
264pub struct RotaryBridgeConfig {
268 pub provider: Arc<dyn UnthinkclawProvider>,
269 pub tools: Vec<Arc<dyn UnthinkclawTool>>,
270 pub system_prompt: String,
271 pub model: String,
272 pub workspace: std::path::PathBuf,
273 pub max_tool_iterations: usize,
274}
275
276pub struct RotaryAgentBridge {
288 agent: rx4::Agent,
289 messages: Vec<Message>,
291}
292
293impl RotaryAgentBridge {
294 pub fn new(config: RotaryBridgeConfig) -> Self {
296 let rx4_provider = Arc::new(RotaryProviderAdapter::new(config.provider));
297
298 let mut agent = rx4::Agent::new();
299 agent.set_model(&config.model);
300 agent.set_system_prompt(&config.system_prompt);
301 agent.set_provider(rx4_provider);
302 agent.set_workspace_root(&config.workspace);
303 agent.max_tool_iterations = config.max_tool_iterations;
304
305 let mut tool_registry = rx4::ToolRegistry::new();
307 register_apollo_tools(&mut tool_registry, &config.tools);
308 agent.tools = Arc::new(tool_registry);
309
310 Self {
315 agent,
316 messages: Vec::new(),
317 }
318 }
319
320 pub fn agent(&self) -> &rx4::Agent {
322 &self.agent
323 }
324
325 pub fn agent_mut(&mut self) -> &mut rx4::Agent {
327 &mut self.agent
328 }
329
330 pub fn clear_messages(&mut self) {
332 self.messages.clear();
333 self.agent.clear_messages();
334 }
335
336 pub fn message_count(&self) -> usize {
338 self.messages.len()
339 }
340
341 pub fn set_model(&mut self, model: &str) {
343 self.agent.set_model(model);
344 }
345
346 pub fn set_system_prompt(&mut self, prompt: &str) {
348 self.agent.set_system_prompt(prompt);
349 }
350
351 pub fn set_workspace_root(&mut self, path: &std::path::Path) {
353 self.agent.set_workspace_root(path);
354 }
355
356 pub fn set_scope(&mut self, scope: rx4::Scope) {
358 self.agent.set_scope(scope);
359 }
360
361 pub fn subscribe(&mut self, callback: impl Fn(&rx4::Event) + Send + Sync + 'static) {
363 self.agent.subscribe(callback);
364 }
365
366 pub async fn run_prompt(&mut self, prompt: &str) -> anyhow::Result<String> {
377 let last_response = Arc::new(parking_lot::RwLock::new(String::new()));
379 let last_response_clone = Arc::clone(&last_response);
380
381 self.agent.subscribe(move |event| {
382 if let rx4::Event::MessageEnd {
383 content,
384 role: Role::Assistant,
385 } = event
386 {
387 *last_response_clone.write() = content.clone();
388 }
389 });
390
391 self.agent.prompt(prompt).await?;
392
393 let response = last_response.read().clone();
394 Ok(response)
395 }
396
397 pub async fn run_prompt_with_history(
403 &mut self,
404 prompt: &str,
405 history: &[ChatMessage],
406 ) -> anyhow::Result<String> {
407 self.agent.clear_messages();
409 for msg in history {
410 let rx4_msg = chat_message_to_rx4(msg);
411 self.agent.messages.write().push(rx4_msg);
414 }
415
416 self.run_prompt(prompt).await
417 }
418
419 pub fn register_tools(&mut self, tools: &[Arc<dyn UnthinkclawTool>]) {
421 if let Some(registry) = Arc::get_mut(&mut self.agent.tools) {
422 register_apollo_tools(registry, tools);
423 } else {
424 tracing::warn!("cannot register rx4 tools while the registry is shared");
425 }
426 }
427
428 pub fn list_tools(&self) -> Vec<String> {
430 self.agent
431 .tools
432 .definitions()
433 .iter()
434 .filter_map(|d| {
435 d.get("name")
436 .and_then(|n| n.as_str())
437 .map(|s| s.to_string())
438 })
439 .collect()
440 }
441
442 pub fn compact(&mut self, reason: &str) {
444 self.agent.compact(reason);
445 }
446}
447
448pub fn build_rx4_skill_engine(workspace: &std::path::Path) -> rx4::SkillEngine {
470 let home = dirs::home_dir().unwrap_or_default();
471
472 let managed_dir = workspace.join(".apollo/skills");
474
475 let mut engine = rx4::SkillEngine::new(managed_dir);
476
477 let openclaw_skills = home.join(".npm-global/lib/node_modules/openclaw/skills");
479 engine.add_extra_dir(openclaw_skills);
480
481 let shared_skills = home.join(".openclaw/workspace/skills");
482 engine.add_extra_dir(shared_skills);
483
484 engine
485}
486
487pub fn match_skill_via_rx4(
498 engine: &rx4::SkillEngine,
499 user_message: &str,
500) -> Option<(String, String)> {
501 let results = engine.search(user_message);
502 if results.is_empty() {
503 return None;
504 }
505
506 let skill = results[0];
508 Some((skill.name.clone(), skill.instructions.clone()))
509}
510
511pub fn discover_skills_via_rx4(workspace: &std::path::Path) -> Vec<crate::skills::Skill> {
519 let mut engine = build_rx4_skill_engine(workspace);
520 if engine.load().is_err() {
521 tracing::warn!("rx4 SkillEngine load failed, returning empty skill list");
522 return Vec::new();
523 }
524
525 engine
526 .list()
527 .into_iter()
528 .map(|rx4_skill| {
529 let location = engine.skills_dir().join(format!("{}.json", rx4_skill.id));
530 crate::skills::Skill {
531 name: rx4_skill.name.clone(),
532 description: rx4_skill.description.clone(),
533 location,
534 }
535 })
536 .collect()
537}
538
539pub struct RotaryMemoryBridge {
558 graph: rx4::GraphMemory,
559 extractor: rx4::ConversationExtractor,
560 graph_path: Option<std::path::PathBuf>,
562}
563
564impl RotaryMemoryBridge {
565 pub fn new(workspace: &std::path::Path) -> Self {
567 let graph = rx4::GraphMemory::from_workspace(workspace);
568 let graph_path = workspace.join(".apollo/graph_memory.json");
569 Self {
570 graph,
571 extractor: rx4::ConversationExtractor::new(),
572 graph_path: Some(graph_path),
573 }
574 }
575
576 pub fn empty() -> Self {
578 Self {
579 graph: rx4::GraphMemory::new(),
580 extractor: rx4::ConversationExtractor::new(),
581 graph_path: None,
582 }
583 }
584
585 pub fn extract_conversation(
592 &mut self,
593 conversation: &[rx4::graph_memory::ConversationTurn],
594 ) -> rx4::ExtractionResult {
595 let result = self.extractor.extract(conversation);
596
597 for node in &result.nodes {
599 self.graph.add_node(node.clone());
600 }
601 for edge in &result.edges {
602 let _ = self.graph.add_edge(edge.clone());
603 }
604
605 result
606 }
607
608 pub fn search(&self, query: &str) -> Vec<&rx4::GraphMemoryNode> {
610 self.graph.search(query)
611 }
612
613 pub fn pagerank(&self) -> Vec<(String, f64)> {
616 self.graph.pagerank()
617 }
618
619 pub fn stats(&self) -> rx4::graph_memory::GraphStats {
621 self.graph.stats()
622 }
623
624 pub fn save(&self) -> Result<(), rx4::GraphMemoryError> {
626 if let Some(path) = &self.graph_path {
627 if let Some(parent) = path.parent() {
628 let _ = std::fs::create_dir_all(parent);
629 }
630 self.graph.save(path)?;
631 tracing::debug!("graph memory saved to {:?}", self.graph_path);
632 }
633 Ok(())
634 }
635
636 pub fn load(&mut self) -> Result<(), rx4::GraphMemoryError> {
638 if let Some(path) = &self.graph_path {
639 if path.exists() {
640 self.graph = rx4::GraphMemory::load(path)?;
641 tracing::debug!("graph memory loaded from {:?}", self.graph_path);
642 }
643 }
644 Ok(())
645 }
646
647 pub fn graph(&self) -> &rx4::GraphMemory {
649 &self.graph
650 }
651
652 pub fn graph_mut(&mut self) -> &mut rx4::GraphMemory {
654 &mut self.graph
655 }
656}
657
658#[cfg(test)]
659mod tests {
660 use super::*;
661
662 #[test]
663 fn test_chat_message_to_rx4_system() {
664 let msg = ChatMessage::system("hello");
665 let rx4_msg = chat_message_to_rx4(&msg);
666 assert_eq!(rx4_msg.role, Role::System);
667 assert_eq!(rx4_msg.content, "hello");
668 }
669
670 #[test]
671 fn test_chat_message_to_rx4_user() {
672 let msg = ChatMessage::user("test");
673 let rx4_msg = chat_message_to_rx4(&msg);
674 assert_eq!(rx4_msg.role, Role::User);
675 assert_eq!(rx4_msg.content, "test");
676 }
677
678 #[test]
679 fn test_chat_message_to_rx4_tool_result() {
680 let msg = ChatMessage::tool_result("tc_123", "result text");
681 let rx4_msg = chat_message_to_rx4(&msg);
682 assert_eq!(rx4_msg.role, Role::Tool);
683 assert_eq!(rx4_msg.content, "result text");
684 assert_eq!(rx4_msg.tool_call_id.as_deref(), Some("tc_123"));
685 }
686
687 #[test]
688 fn test_rx4_message_to_chat() {
689 let msg = Message::assistant("hello back");
690 let chat_msg = rx4_message_to_chat(&msg);
691 assert_eq!(chat_msg.role, "assistant");
692 assert_eq!(chat_msg.content, "hello back");
693 }
694
695 #[test]
696 fn test_tool_specs_to_rx4_json() {
697 let specs = vec![ToolSpec {
698 name: "shell".to_string(),
699 description: "Run shell commands".to_string(),
700 parameters: serde_json::json!({"type": "object"}),
701 }];
702 let json = tool_specs_to_rx4_json(&specs);
703 assert_eq!(json.len(), 1);
704 assert_eq!(json[0]["name"], "shell");
705 }
706
707 #[test]
708 fn test_roundtrip_translation() {
709 let original = ChatMessage::user("roundtrip test");
710 let rx4_msg = chat_message_to_rx4(&original);
711 let back = rx4_message_to_chat(&rx4_msg);
712 assert_eq!(back.role, "user");
713 assert_eq!(back.content, "roundtrip test");
714 }
715
716 #[test]
717 fn test_build_rx4_skill_engine() {
718 let tmp = tempfile::tempdir().unwrap();
720 let engine = build_rx4_skill_engine(tmp.path());
721 assert!(
722 engine.skills_dir().exists()
723 || engine.skills_dir() == tmp.path().join(".apollo/skills")
724 );
725 }
726
727 #[test]
728 fn test_match_skill_via_rx4_empty() {
729 let tmp = tempfile::tempdir().unwrap();
730 let mut engine = build_rx4_skill_engine(tmp.path());
731 let _ = engine.load();
732 let result = match_skill_via_rx4(&engine, "test query");
734 assert!(result.is_none());
735 }
736
737 #[test]
738 fn test_rotary_memory_bridge_empty() {
739 let bridge = RotaryMemoryBridge::empty();
740 let stats = bridge.stats();
741 assert_eq!(stats.node_count, 0);
742 }
743
744 #[test]
745 fn test_rotary_memory_bridge_search_empty() {
746 let bridge = RotaryMemoryBridge::empty();
747 let results = bridge.search("test");
748 assert!(results.is_empty());
749 }
750
751 #[test]
752 fn test_rotary_memory_bridge_save_load() {
753 let tmp = tempfile::tempdir().unwrap();
754 let mut bridge = RotaryMemoryBridge::new(tmp.path());
755 bridge.save().unwrap();
757 bridge.load().unwrap();
758 assert_eq!(bridge.stats().node_count, 0);
759 }
760}