apollo/agent/

rotary_bridge.rs

//! Rotary (rx4) bridge — adapts apollo's types to rx4's agent harness.
//!
//! This module provides:
//! - `RotaryProviderAdapter`: wraps an apollo `Provider` as an `rx4::Provider`
//!   so rx4's `Agent` loop can use apollo's existing provider backends.
//! - `register_apollo_tools`: registers apollo's `Tool` trait objects
//!   into rx4's `ToolRegistry` via boxed closures.
//! - `chat_message_to_rx4` / `rx4_message_to_chat`: type translators between
//!   apollo's `ChatMessage` and rx4's `Message`.
//! - `RotaryAgentBridge`: wraps an `rx4::Agent`, wiring up provider, tools,
//!   system prompt, and providing a `run_prompt` method that the outer
//!   apollo shell (channels, swarm, cron, heartbeat) can call.
//!
//! The bridge is designed to be incremental. The existing `AgentRunner` loop
//! remains available; this bridge provides an alternative execution path that
//! delegates the core agent loop to rx4 while keeping apollo's unique
//! features (channels, swarm, cron, heartbeat, autonomous mode, plugins, MCP)
//! as the outer shell.

use std::sync::Arc;

use rx4::provider::{
    Message, Provider as Rx4Provider, ProviderError as Rx4ProviderError, Role, StreamEvent,
};

use crate::agent::hooks::{run_post_hooks, run_pre_hooks, HookDecision, ToolHook};
use crate::agent::stream::{emit, AgentStreamEvent, AgentStreamTx};
use crate::plugin::{HookManager, LifecycleEvent, PluginRegistry};
use crate::providers::{ChatMessage, ChatRequest, Provider as UnthinkclawProvider};
use crate::tools::{Tool as UnthinkclawTool, ToolResult as UnthinkclawToolResult, ToolSpec};

/// Everything a tool call must be wrapped in, for either engine.
///
/// Both engines run the same sequence around every tool: the
/// `BeforeToolCall` lifecycle event, a `ToolStart` stream event, plugin then
/// policy pre-checks, execution, the post hooks, the `AfterToolCall`
/// lifecycle event, plugin notification, and a `ToolEnd` stream event. That
/// sequence lives once, in `execute_tool_with_hooks`; this type carries the
/// collaborators it needs so a hook that fires under `agent.engine =
/// "legacy"` fires identically under `"rx4"`.
#[derive(Clone, Default)]
pub struct ToolHookContext {
    hooks: Vec<Arc<dyn ToolHook>>,
    plugins: Option<Arc<tokio::sync::RwLock<PluginRegistry>>>,
    hook_manager: Option<Arc<HookManager>>,
    stream: Option<AgentStreamTx>,
}

impl ToolHookContext {
    pub fn new(
        hooks: Vec<Arc<dyn ToolHook>>,
        plugins: Option<Arc<tokio::sync::RwLock<PluginRegistry>>>,
    ) -> Self {
        Self {
            hooks,
            plugins,
            hook_manager: None,
            stream: None,
        }
    }

    /// Attach the lifecycle hook manager, so plugins observing tool calls see
    /// them under either engine.
    pub fn with_hook_manager(mut self, hook_manager: Arc<HookManager>) -> Self {
        self.hook_manager = Some(hook_manager);
        self
    }

    /// Attach the turn's stream sink, so a WS client sees tool progress under
    /// either engine.
    pub fn with_stream(mut self, stream: Option<AgentStreamTx>) -> Self {
        self.stream = stream;
        self
    }

    async fn emit_lifecycle(&self, event: LifecycleEvent) {
        if let Some(manager) = &self.hook_manager {
            manager.emit(&event).await;
        }
    }

    /// Run the pre-tool checks. `Block` means the tool must not execute.
    pub async fn check_pre_tool(&self, name: &str, arguments: &str) -> HookDecision {
        if let Some(plugins) = &self.plugins {
            let registry = plugins.read().await;
            if let HookDecision::Block(reason) = registry.check_pre_tool(name, arguments).await {
                return HookDecision::Block(format!("Blocked by plugin: {reason}"));
            }
        }
        match run_pre_hooks(&self.hooks, name, arguments).await {
            HookDecision::Block(reason) => {
                HookDecision::Block(format!("Blocked by policy: {reason}"))
            }
            HookDecision::Allow => HookDecision::Allow,
        }
    }

    /// Notify the post-tool hooks and plugins.
    pub async fn notify_post_tool(
        &self,
        name: &str,
        arguments: &str,
        result: &UnthinkclawToolResult,
    ) {
        run_post_hooks(&self.hooks, name, arguments, result).await;
        self.emit_lifecycle(LifecycleEvent::AfterToolCall(
            name.to_string(),
            arguments.to_string(),
            result.clone(),
        ))
        .await;
        if let Some(plugins) = &self.plugins {
            let registry = plugins.read().await;
            registry.notify_post_tool(name, arguments, result).await;
        }
    }
}

/// Run one tool call with every hook and event both engines owe it.
///
/// This is the single place the ordering exists. `tool` is `None` when the
/// model named a tool that is not registered; the pre-checks still run, so a
/// policy that blocks an unknown name is honoured before that is reported.
pub async fn execute_tool_with_hooks(
    ctx: &ToolHookContext,
    name: &str,
    arguments: &str,
    tool: Option<&Arc<dyn UnthinkclawTool>>,
) -> UnthinkclawToolResult {
    ctx.emit_lifecycle(LifecycleEvent::BeforeToolCall(
        name.to_string(),
        arguments.to_string(),
    ))
    .await;
    emit(
        &ctx.stream,
        AgentStreamEvent::ToolStart {
            name: name.to_string(),
            hint: crate::agent::loop_runner::extract_tool_hint(name, arguments),
        },
    );

    let started = std::time::Instant::now();
    let result = match ctx.check_pre_tool(name, arguments).await {
        HookDecision::Block(reason) => {
            tracing::info!("blocked '{}': {}", name, reason);
            UnthinkclawToolResult::error(reason)
        }
        HookDecision::Allow => match tool {
            Some(tool) => match tool.execute(arguments).await {
                Ok(result) => result,
                Err(e) => UnthinkclawToolResult::error(crate::redaction::redact_text(&format!(
                    "Tool error: {e}"
                ))),
            },
            None => UnthinkclawToolResult::error(format!("Unknown tool: {name}")),
        },
    };

    ctx.notify_post_tool(name, arguments, &result).await;

    emit(
        &ctx.stream,
        AgentStreamEvent::ToolEnd {
            name: name.to_string(),
            ok: !result.is_error,
            elapsed_secs: started.elapsed().as_secs(),
        },
    );

    result
}

// ── Message translation ──────────────────────────────────────────────────

/// Convert an apollo `ChatMessage` to an rx4 `Message`.
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(),
    }
}

/// Convert an rx4 `Message` back to an apollo `ChatMessage`.
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(),
    }
}

/// Convert a slice of apollo `ChatMessage`s to rx4 `Message`s.
pub fn chat_messages_to_rx4(messages: &[ChatMessage]) -> Vec<Message> {
    messages.iter().map(chat_message_to_rx4).collect()
}

/// Convert apollo `ToolSpec`s to rx4 tool definitions (JSON array).
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()
}

// ── Provider adapter ─────────────────────────────────────────────────────

/// Adapter that wraps an apollo `Provider` and implements rx4's `Provider`
/// trait. This lets rx4's `Agent` loop use apollo's existing provider
/// backends (Anthropic, OpenAI-compat, Ollama, Copilot) without modification.
///
/// rx4's `Provider` trait is streaming-based (`stream()`), while apollo's
/// is request-response (`chat()`). This adapter bridges the gap by calling
/// apollo's `chat()` and wrapping the result in a single-element stream.
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> {
        // Translate rx4 messages to apollo ChatMessages
        let mut chat_messages: Vec<ChatMessage> = Vec::new();

        // rx4 passes system prompt separately; apollo includes it in messages
        if let Some(sys) = system {
            chat_messages.push(ChatMessage::system(sys));
        }

        for msg in messages {
            chat_messages.push(rx4_message_to_chat(msg));
        }

        // Convert rx4 tool definitions to apollo ToolSpecs
        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 {
            // Safety: tool_specs lives for the duration of this call
            // This is a workaround for the lifetime constraint in ChatRequest
            &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()))?;

        // Build a stream that emits the response as events
        let text = response.text.unwrap_or_default();
        let tool_calls = response.tool_calls;

        // Create a single-shot stream
        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
        };

        // Return a stream that yields the pre-computed events
        use futures_util::stream;
        Ok(Box::new(Box::pin(stream::iter(events))))
    }
}

// ── Tool registration ────────────────────────────────────────────────────

/// Register apollo's `Tool` trait objects into rx4's `ToolRegistry`.
///
/// Each apollo tool is wrapped in a boxed closure that captures the
/// `Arc<dyn Tool>` and calls its `execute()` method. The closure is registered
/// via `ToolDefinition::new_boxed()`, which uses `ToolExecutor::Boxed`.
///
/// Tool effects are classified based on the tool name using rx4's
/// `classify_tool()` guardrail function — idempotent tools get `ToolEffect::Read`,
/// mutating tools get `ToolEffect::Write`.
pub fn register_apollo_tools(
    registry: &mut rx4::ToolRegistry,
    tools: &[Arc<dyn UnthinkclawTool>],
    hook_ctx: &ToolHookContext,
) {
    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 hook_ctx = hook_ctx.clone();
        let tool_name = name.clone();
        let execute: ToolExecuteBox = Box::new(move |_ctx, args| {
            let tool = Arc::clone(&tool_clone);
            let hook_ctx = hook_ctx.clone();
            let tool_name = tool_name.clone();
            Box::pin(async move {
                let result =
                    execute_tool_with_hooks(&hook_ctx, &tool_name, &args, Some(&tool)).await;

                rx4::ToolResult {
                    id: String::new(),
                    content: result.output,
                    is_error: result.is_error,
                }
            })
        });

        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),
        );
    }
}

// ── Agent bridge ─────────────────────────────────────────────────────────

/// Configuration for building a `RotaryAgentBridge`.
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,
    /// Pre/post tool hooks, so rx4 enforces the same permissions as the
    /// legacy loop.
    pub hook_ctx: ToolHookContext,
}

/// Bridge that wraps an `rx4::Agent` and provides a simplified interface for
/// apollo's outer shell to use.
///
/// The bridge handles:
/// - Creating and configuring the rx4::Agent (provider, tools, system prompt)
/// - Translating messages between apollo and rx4 types
/// - Running prompts through rx4's agent loop
///
/// Unthinkclaw's unique features (channels, swarm, cron, heartbeat, autonomous
/// mode, plugins) remain in the outer shell and call `run_prompt()` on this
/// bridge to execute agent turns.
pub struct RotaryAgentBridge {
    agent: rx4::Agent,
    hook_ctx: ToolHookContext,
    /// Conversation messages maintained in rx4 format (per-session)
    messages: Vec<Message>,
}

impl RotaryAgentBridge {
    /// Build a new bridge from the given configuration.
    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;

        // Register apollo's tools into rx4's tool registry
        let mut tool_registry = rx4::ToolRegistry::new();
        register_apollo_tools(&mut tool_registry, &config.tools, &config.hook_ctx);
        agent.tools = Arc::new(tool_registry);

        // Use rx4's guardrails for loop detection (replaces apollo's
        // ToolGuardrails in the main loop path)
        // rx4's Agent already has built-in tool caching and effect classification

        Self {
            agent,
            hook_ctx: config.hook_ctx,
            messages: Vec::new(),
        }
    }

    /// Get a reference to the inner rx4::Agent (for advanced configuration).
    pub fn agent(&self) -> &rx4::Agent {
        &self.agent
    }

    /// Get a mutable reference to the inner rx4::Agent.
    pub fn agent_mut(&mut self) -> &mut rx4::Agent {
        &mut self.agent
    }

    /// Clear the conversation history.
    pub fn clear_messages(&mut self) {
        self.messages.clear();
        self.agent.clear_messages();
    }

    /// Get the number of messages in the conversation.
    pub fn message_count(&self) -> usize {
        self.messages.len()
    }

    /// Set the model for the agent.
    pub fn set_model(&mut self, model: &str) {
        self.agent.set_model(model);
    }

    /// Set the system prompt.
    pub fn set_system_prompt(&mut self, prompt: &str) {
        self.agent.set_system_prompt(prompt);
    }

    /// Set the workspace root.
    pub fn set_workspace_root(&mut self, path: &std::path::Path) {
        self.agent.set_workspace_root(path);
    }

    /// Set the scope (e.g., Coding, Research, Ask).
    pub fn set_scope(&mut self, scope: rx4::Scope) {
        self.agent.set_scope(scope);
    }

    /// Add a subscriber to receive agent events (tool calls, deltas, etc.).
    pub fn subscribe(&mut self, callback: impl Fn(&rx4::Event) + Send + Sync + 'static) {
        self.agent.subscribe(callback);
    }

    /// Run a single user prompt through the rx4 agent loop.
    ///
    /// This delegates the core agent loop (LLM calls, tool execution, turn
    /// cycling) to rx4::Agent. The caller (apollo's channel/swarm/cron
    /// shell) is responsible for:
    /// - Receiving the user message from a channel
    /// - Calling this method with the prompt text
    /// - Sending the final response back through the channel
    ///
    /// Returns the final assistant response text.
    pub async fn run_prompt(&mut self, prompt: &str) -> anyhow::Result<String> {
        // Track the last assistant message for the return value
        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)
    }

    /// Run a prompt with pre-loaded conversation history.
    ///
    /// The history is loaded into rx4's message buffer before running the
    /// prompt. This is used when apollo's memory backend provides
    /// conversation history for a chat session.
    pub async fn run_prompt_with_history(
        &mut self,
        prompt: &str,
        history: &[ChatMessage],
    ) -> anyhow::Result<String> {
        // Load history into rx4's message buffer
        self.agent.clear_messages();
        for msg in history {
            let rx4_msg = chat_message_to_rx4(msg);
            // rx4's messages are stored internally; we push them via the
            // messages RwLock
            self.agent.messages.write().push(rx4_msg);
        }

        self.run_prompt(prompt).await
    }

    /// Register additional tools at runtime.
    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, &self.hook_ctx);
        } else {
            tracing::warn!("cannot register rx4 tools while the registry is shared");
        }
    }

    /// Get the list of registered tool names.
    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()
    }

    /// Compact the conversation context (delegates to rx4's compact).
    pub fn compact(&mut self, reason: &str) {
        self.agent.compact(reason);
    }
}

// ── Skill bridge ─────────────────────────────────────────────────────────

/// Build an `rx4::SkillEngine` configured with apollo's skill directories.
///
/// Unthinkclaw discovers skills from 3 directories:
/// 1. `~/.npm-global/lib/node_modules/openclaw/skills` (legacy)
/// 2. `~/.openclaw/workspace/skills` (shared workspace skills)
/// 3. `{workspace}/.apollo/skills` (project-local managed skills)
///
/// This maps to rx4's `SkillEngine` with the primary dir set to the managed
/// skills directory and the other two as `extra_dirs`.
///
/// After calling this, use `engine.load()` to populate skills from disk,
/// then `engine.search()` for keyword matching (replaces apollo's
/// `match_skill()`).
///
/// Note: apollo's template variable substitution and inline shell
/// preprocessing (`preprocess_skill_content`) are not part of rx4's
/// SkillEngine and remain in apollo's `skills` module. Use
/// `skills::preprocess_skill_content()` on the matched skill's instructions
/// before injecting into the system prompt.
pub fn build_rx4_skill_engine(workspace: &std::path::Path) -> rx4::SkillEngine {
    let home = dirs::home_dir().unwrap_or_default();

    // Primary dir: managed skills in the workspace
    let managed_dir = workspace.join(".apollo/skills");

    let mut engine = rx4::SkillEngine::new(managed_dir);

    // Extra dirs: legacy openclaw skills and shared workspace skills
    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
}

/// Match a skill using rx4's SkillEngine keyword search.
///
/// This replaces apollo's `skills::match_skill()` when using the rx4
/// bridge path. Returns the best-matching skill's name and instructions
/// (raw, unpreprocessed).
///
/// The caller should preprocess the instructions using
/// `apollo::skills::preprocess_skill_content()` before injecting
/// into the system prompt, as rx4's SkillEngine does not perform template
/// variable substitution or inline shell expansion.
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;
    }

    // Pick the first result (rx4's search returns matches sorted by relevance)
    let skill = results[0];
    Some((skill.name.clone(), skill.instructions.clone()))
}

/// Discover skills using rx4's SkillEngine, returning apollo-compatible
/// Skill structs for backward compatibility with existing code that expects
/// the `Vec<skills::Skill>` type.
///
/// This loads skills from disk via rx4's SkillEngine (which handles both
/// JSON and SKILL.md formats with YAML frontmatter), then converts them to
/// apollo's Skill type.
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()
}

// ── Memory bridge ────────────────────────────────────────────────────────

/// Bridge that wraps rx4's GraphMemory for agent memory (concepts, decisions,
/// patterns, bugs) while keeping apollo's SurrealDB for channel state,
/// swarm coordination, and cron scheduling.
///
/// rx4's GraphMemory is an in-memory knowledge graph with PageRank, community
/// detection, and JSON persistence. It does not require SQLite (unlike rx4's
/// MemoryStore which uses rusqlite 0.37).
///
/// This bridge provides:
/// - Graph-based agent memory via rx4::GraphMemory (concepts, decisions, patterns)
/// - Conversation extraction via rx4::ConversationExtractor
/// - JSON persistence for the graph
///
/// TODO: Enable rx4's `memory` feature to also get SQLite FTS5 full-text search
/// via rx4::MemoryStore. For now, apollo's SurrealDB backend remains
/// the primary persistent memory for conversation history and key-value store.
pub struct RotaryMemoryBridge {
    graph: rx4::GraphMemory,
    extractor: rx4::ConversationExtractor,
    /// Path for JSON persistence of the graph
    graph_path: Option<std::path::PathBuf>,
}

impl RotaryMemoryBridge {
    /// Create a new memory bridge rooted at the given workspace.
    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),
        }
    }

    /// Create a new memory bridge with an empty graph (no workspace).
    pub fn empty() -> Self {
        Self {
            graph: rx4::GraphMemory::new(),
            extractor: rx4::ConversationExtractor::new(),
            graph_path: None,
        }
    }

    /// Extract memory nodes and edges from a conversation and add them to
    /// the graph.
    ///
    /// This uses rx4's ConversationExtractor to identify concepts, decisions,
    /// and patterns from the conversation, then adds them as nodes in the
    /// GraphMemory.
    pub fn extract_conversation(
        &mut self,
        conversation: &[rx4::graph_memory::ConversationTurn],
    ) -> rx4::ExtractionResult {
        let result = self.extractor.extract(conversation);

        // Add extracted nodes to the graph
        for node in &result.nodes {
            self.graph.add_node(node.clone());
        }
        for edge in &result.edges {
            let _ = self.graph.add_edge(edge.clone());
        }

        result
    }

    /// Search the graph memory for nodes matching the query.
    pub fn search(&self, query: &str) -> Vec<&rx4::GraphMemoryNode> {
        self.graph.search(query)
    }

    /// Get PageRank scores for all nodes (identifies the most important
    /// concepts/decisions in the agent's memory).
    pub fn pagerank(&self) -> Vec<(String, f64)> {
        self.graph.pagerank()
    }

    /// Get graph statistics (node count, edge count, etc.)
    pub fn stats(&self) -> rx4::graph_memory::GraphStats {
        self.graph.stats()
    }

    /// Save the graph to disk (JSON format).
    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(())
    }

    /// Load the graph from disk (JSON format).
    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(())
    }

    /// Get a reference to the inner GraphMemory.
    pub fn graph(&self) -> &rx4::GraphMemory {
        &self.graph
    }

    /// Get a mutable reference to the inner GraphMemory.
    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() {
        // Just verify it doesn't panic with a temp dir
        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();
        // No skills in empty dir, should return None
        let result = match_skill_via_rx4(&engine, "test query");
        assert!(result.is_none());
    }

    struct RecordingTool {
        ran: Arc<std::sync::atomic::AtomicBool>,
    }

    #[async_trait::async_trait]
    impl UnthinkclawTool for RecordingTool {
        fn name(&self) -> &str {
            "exec"
        }

        fn spec(&self) -> ToolSpec {
            ToolSpec {
                name: "exec".to_string(),
                description: "test tool".to_string(),
                parameters: serde_json::json!({"type": "object"}),
            }
        }

        async fn execute(&self, _arguments: &str) -> anyhow::Result<UnthinkclawToolResult> {
            self.ran.store(true, std::sync::atomic::Ordering::SeqCst);
            Ok(UnthinkclawToolResult::success("ran"))
        }
    }

    async fn run_exec_through_rx4(hook_ctx: ToolHookContext) -> (rx4::ToolResult, bool) {
        let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let tool: Arc<dyn UnthinkclawTool> = Arc::new(RecordingTool {
            ran: Arc::clone(&ran),
        });
        let mut registry = rx4::ToolRegistry::new();
        register_apollo_tools(&mut registry, &[tool], &hook_ctx);

        let ctx = Arc::new(rx4::ToolContext::new("."));
        let result = registry
            .execute("exec", &ctx, r#"{"command":"rm -rf /"}"#)
            .await
            .expect("tool registered");
        (result, ran.load(std::sync::atomic::Ordering::SeqCst))
    }

    #[tokio::test]
    async fn rx4_bridge_enforces_blocking_hooks() {
        let hook: Arc<dyn ToolHook> = Arc::new(crate::agent::hooks::PermissionHook::new(
            vec!["exec".to_string()],
            vec![],
        ));
        let (result, ran) = run_exec_through_rx4(ToolHookContext::new(vec![hook], None)).await;
        assert!(result.is_error, "blocked tool must report an error");
        assert!(
            result.content.contains("Blocked by policy"),
            "unexpected content: {}",
            result.content
        );
        assert!(!ran, "a blocked tool must not execute under rx4");
    }

    #[tokio::test]
    async fn rx4_bridge_allows_unblocked_tools() {
        let (result, ran) = run_exec_through_rx4(ToolHookContext::default()).await;
        assert!(!result.is_error);
        assert_eq!(result.content, "ran");
        assert!(ran);
    }

    #[tokio::test]
    async fn rx4_bridge_enforces_plugin_pre_tool_block() {
        let mut registry = PluginRegistry::new();
        registry.register_pre_tool_hook(Arc::new(BlockingPluginHook));
        let ctx = ToolHookContext::new(
            Vec::new(),
            Some(Arc::new(tokio::sync::RwLock::new(registry))),
        );
        let (result, ran) = run_exec_through_rx4(ctx).await;
        assert!(result.is_error);
        assert!(
            result.content.contains("Blocked by plugin"),
            "unexpected content: {}",
            result.content
        );
        assert!(!ran);
    }

    struct BlockingPluginHook;

    #[async_trait::async_trait]
    impl crate::plugin::PreToolHook for BlockingPluginHook {
        fn name(&self) -> &str {
            "blocking-test-hook"
        }

        async fn before_tool_call(&self, _name: &str, _arguments: &str) -> HookDecision {
            HookDecision::Block("plugin says no".to_string())
        }
    }

    /// Records the lifecycle events a plugin would see.
    struct RecordingLifecycleHook {
        seen: Arc<std::sync::Mutex<Vec<String>>>,
    }

    #[async_trait::async_trait]
    impl crate::plugin::LifecycleHook for RecordingLifecycleHook {
        fn name(&self) -> &str {
            "recording-lifecycle"
        }

        async fn on_event(&self, event: &LifecycleEvent) -> anyhow::Result<()> {
            let label = match event {
                LifecycleEvent::BeforeToolCall(name, _) => format!("before:{name}"),
                LifecycleEvent::AfterToolCall(name, _, _) => format!("after:{name}"),
                other => format!("other:{other:?}"),
            };
            self.seen.lock().unwrap().push(label);
            Ok(())
        }
    }

    fn stream_labels(
        rx: &mut tokio::sync::mpsc::UnboundedReceiver<AgentStreamEvent>,
    ) -> Vec<String> {
        let mut labels = Vec::new();
        while let Ok(event) = rx.try_recv() {
            labels.push(match event {
                AgentStreamEvent::ToolStart { name, .. } => format!("tool_start:{name}"),
                AgentStreamEvent::ToolEnd { name, ok, .. } => format!("tool_end:{name}:{ok}"),
                other => format!("other:{other:?}"),
            });
        }
        labels
    }

    /// Build a context that records everything a plugin or WS client sees.
    fn recording_context() -> (
        ToolHookContext,
        Arc<std::sync::Mutex<Vec<String>>>,
        tokio::sync::mpsc::UnboundedReceiver<AgentStreamEvent>,
    ) {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let mut manager = HookManager::new();
        manager.register_lifecycle(Arc::new(RecordingLifecycleHook {
            seen: Arc::clone(&seen),
        }));
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let ctx = ToolHookContext::default()
            .with_hook_manager(Arc::new(manager))
            .with_stream(Some(tx));
        (ctx, seen, rx)
    }

    /// Both engines must produce the same hooks and stream events for a tool
    /// call. The legacy loop and the rx4 registry now reach the tool through
    /// the same `execute_tool_with_hooks`; this fails if either side stops
    /// doing so, which is how rx4 previously lost `BeforeToolCall` and the
    /// `ToolStart`/`ToolEnd` progress events.
    #[tokio::test]
    async fn both_engines_emit_the_same_hooks_and_events() {
        let args = r#"{"command":"ls"}"#;

        // rx4: the tool runs inside the registry closure.
        let (ctx, rx4_seen, mut rx4_stream) = recording_context();
        let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let tool: Arc<dyn UnthinkclawTool> = Arc::new(RecordingTool {
            ran: Arc::clone(&ran),
        });
        let mut registry = rx4::ToolRegistry::new();
        register_apollo_tools(&mut registry, &[Arc::clone(&tool)], &ctx);
        let tool_ctx = Arc::new(rx4::ToolContext::new("."));
        registry
            .execute("exec", &tool_ctx, args)
            .await
            .expect("tool registered");
        let rx4_events = rx4_seen.lock().unwrap().clone();
        let rx4_stream_events = stream_labels(&mut rx4_stream);

        // legacy: the loop calls the shared path directly.
        let (ctx, legacy_seen, mut legacy_stream) = recording_context();
        execute_tool_with_hooks(&ctx, "exec", args, Some(&tool)).await;
        let legacy_events = legacy_seen.lock().unwrap().clone();
        let legacy_stream_events = stream_labels(&mut legacy_stream);

        assert_eq!(
            rx4_events, legacy_events,
            "the engines disagree on lifecycle hooks"
        );
        assert_eq!(
            rx4_stream_events, legacy_stream_events,
            "the engines disagree on stream events"
        );
        assert_eq!(legacy_events, vec!["before:exec", "after:exec"]);
        assert_eq!(
            legacy_stream_events,
            vec!["tool_start:exec", "tool_end:exec:true"]
        );
    }

    #[tokio::test]
    async fn a_blocked_tool_still_reports_start_and_end() {
        let hook: Arc<dyn ToolHook> = Arc::new(crate::agent::hooks::PermissionHook::new(
            vec!["exec".to_string()],
            vec![],
        ));
        let (ctx, seen, mut stream) = recording_context();
        let ctx = ToolHookContext::new(vec![hook], None)
            .with_hook_manager(Arc::new({
                let mut manager = HookManager::new();
                manager.register_lifecycle(Arc::new(RecordingLifecycleHook {
                    seen: Arc::clone(&seen),
                }));
                manager
            }))
            .with_stream(ctx.stream.clone());
        let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let tool: Arc<dyn UnthinkclawTool> = Arc::new(RecordingTool {
            ran: Arc::clone(&ran),
        });
        let result = execute_tool_with_hooks(&ctx, "exec", "{}", Some(&tool)).await;
        assert!(result.is_error);
        assert!(!ran.load(std::sync::atomic::Ordering::SeqCst));
        assert_eq!(
            stream_labels(&mut stream),
            vec!["tool_start:exec", "tool_end:exec:false"],
            "a blocked call must still open and close its progress line"
        );
        assert_eq!(
            seen.lock().unwrap().clone(),
            vec!["before:exec", "after:exec"]
        );
    }

    #[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());
        // Save and load should work even with empty graph
        bridge.save().unwrap();
        bridge.load().unwrap();
        assert_eq!(bridge.stats().node_count, 0);
    }
}