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