claw/src/compat/runtime.rs

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use async_trait::async_trait;
use futures_util::{stream, StreamExt};
use zeroclaw::agent::TurnEvent;
use zeroclaw::config::Config as ZeroClawConfig;
use zeroclaw::providers::traits::{ProviderCapabilities, StreamEvent, StreamOptions, StreamResult};
use zeroclaw::providers::{self, ChatMessage, ChatRequest, ChatResponse, Provider};

use crate::browser::{BrowserBackend, PipeBrowserBackend};
use crate::compat::browser_script_skill_tool::build_browser_script_skill_tools;
use crate::compat::browser_tool_adapter::ZeroClawBrowserTool;
use crate::compat::config_adapter::{
    build_zeroclaw_config_from_sgclaw_settings, resolve_skills_dir_from_sgclaw_settings,
};
use crate::compat::event_bridge::log_entry_for_turn_event;
use crate::compat::workflow_executor::parse_generated_article_draft;
use crate::compat::openxml_office_tool::OpenXmlOfficeTool;
use crate::compat::screen_html_export_tool::ScreenHtmlExportTool;
use crate::config::{DeepSeekSettings, OfficeBackend, SgClawSettings};
use crate::pipe::{BrowserPipeTool, ConversationMessage, PipeError, Transport};
use crate::runtime::RuntimeEngine;

#[derive(Debug, Clone, Default)]
pub struct CompatTaskContext {
    pub conversation_id: Option<String>,
    pub messages: Vec<ConversationMessage>,
    pub page_url: Option<String>,
    pub page_title: Option<String>,
}

pub fn execute_task<T: Transport + 'static>(
    transport: &T,
    browser_tool: BrowserPipeTool<T>,
    instruction: &str,
    task_context: &CompatTaskContext,
    workspace_root: &Path,
    settings: &DeepSeekSettings,
) -> Result<String, PipeError> {
    let sgclaw_settings = SgClawSettings::from(settings);
    execute_task_with_sgclaw_settings(
        transport,
        browser_tool,
        instruction,
        task_context,
        workspace_root,
        &sgclaw_settings,
    )
}

pub fn execute_task_with_browser_backend(
    transport: &dyn crate::agent::AgentEventSink,
    browser_backend: Arc<dyn BrowserBackend>,
    instruction: &str,
    task_context: &CompatTaskContext,
    workspace_root: &Path,
    settings: &SgClawSettings,
) -> Result<String, PipeError> {
    let config = build_zeroclaw_config_from_sgclaw_settings(workspace_root, settings);
    let skills_dir = resolve_skills_dir_from_sgclaw_settings(workspace_root, settings);
    let provider = build_provider(&config)?;
    let runtime = tokio::runtime::Runtime::new()
        .map_err(|err| PipeError::Protocol(format!("failed to create tokio runtime: {err}")))?;

    runtime.block_on(execute_task_with_provider(
        transport,
        browser_backend,
        provider,
        instruction,
        task_context,
        config,
        skills_dir,
        settings.clone(),
    ))
}

pub fn execute_task_with_sgclaw_settings<T: Transport + 'static>(
    transport: &T,
    browser_tool: BrowserPipeTool<T>,
    instruction: &str,
    task_context: &CompatTaskContext,
    workspace_root: &Path,
    settings: &SgClawSettings,
) -> Result<String, PipeError> {
    let config = build_zeroclaw_config_from_sgclaw_settings(workspace_root, settings);
    let skills_dir = resolve_skills_dir_from_sgclaw_settings(workspace_root, settings);
    let provider = build_provider(&config)?;
    let runtime = tokio::runtime::Runtime::new()
        .map_err(|err| PipeError::Protocol(format!("failed to create tokio runtime: {err}")))?;

    runtime.block_on(execute_task_with_provider(
        transport,
        Arc::new(PipeBrowserBackend::from_inner(browser_tool)),
        provider,
        instruction,
        task_context,
        config,
        skills_dir,
        settings.clone(),
    ))
}

pub(crate) fn generate_zhihu_article_draft(
    instruction: &str,
    topic: &str,
    _task_context: &CompatTaskContext,
    workspace_root: &Path,
    settings: &SgClawSettings,
) -> Result<crate::compat::workflow_executor::ArticleDraft, PipeError> {
    let mut generation_settings = settings.clone();
    generation_settings.runtime_profile = crate::runtime::RuntimeProfile::GeneralAssistant;

    let config = build_zeroclaw_config_from_sgclaw_settings(workspace_root, &generation_settings);
    let provider = build_provider(&config)?;
    let runtime = tokio::runtime::Runtime::new()
        .map_err(|err| PipeError::Protocol(format!("failed to create tokio runtime: {err}")))?;
    let generation_prompt = format!(
        "为知乎文章生成可直接发布的草稿。用户原始请求：{instruction}\n\n主题：{topic}\n\n请严格只输出以下格式，不要添加解释、前言、代码块或其他内容：\n标题：<简洁具体的中文标题>\n正文：<适合知乎发布的中文正文，使用自然段>"
    );

    let generated = runtime.block_on(async move {
        provider
            .chat_with_system(
                Some("You write concise Chinese Zhihu article drafts. Return only the requested title/body format."),
                &generation_prompt,
                config.default_model.as_deref().unwrap_or("deepseek-chat"),
                config.default_temperature,
            )
            .await
            .map_err(map_anyhow_to_pipe_error)
    })?;

    parse_generated_article_draft(&generated).ok_or_else(|| {
        PipeError::Protocol(format!(
            "generated Zhihu article draft did not match 标题/正文 format: {generated}"
        ))
    })
}

pub async fn execute_task_with_provider(
    transport: &dyn crate::agent::AgentEventSink,
    browser_backend: Arc<dyn BrowserBackend>,
    provider: Box<dyn Provider>,
    instruction: &str,
    task_context: &CompatTaskContext,
    config: ZeroClawConfig,
    skills_dir: Vec<PathBuf>,
    settings: SgClawSettings,
) -> Result<String, PipeError> {
    let engine = RuntimeEngine::new(settings.runtime_profile);
    let browser_surface_present = engine.browser_surface_enabled();
    let loaded_skills = engine.loaded_skills(&config, &skills_dir);
    let loaded_skill_versions = loaded_skills
        .iter()
        .map(|skill| (skill.name.clone(), skill.version.clone()))
        .collect::<HashMap<_, _>>();
    let loaded_skill_labels = loaded_skills
        .iter()
        .map(|skill| format!("{}@{}", skill.name, skill.version))
        .collect::<Vec<_>>();
    if !loaded_skill_labels.is_empty() {
        transport.send(&crate::pipe::AgentMessage::LogEntry {
            level: "info".to_string(),
            message: format!("loaded skills: {}", loaded_skill_labels.join(", ")),
        })?;
    }
    let browser_tool_for_scripts = browser_backend.clone();
    let browser_tool_for_superrpa = browser_backend.clone();
    let browser_tool_for_browser_action = browser_backend;
    let mut tools: Vec<Box<dyn zeroclaw::tools::Tool>> = if browser_surface_present {
        vec![
            Box::new(ZeroClawBrowserTool::new_superrpa(browser_tool_for_superrpa)),
            Box::new(ZeroClawBrowserTool::new(browser_tool_for_browser_action)),
        ]
    } else {
        Vec::new()
    };
    if browser_surface_present {
        tools.extend(
            build_browser_script_skill_tools(&loaded_skills, browser_tool_for_scripts)
                .map_err(map_anyhow_to_pipe_error)?,
        );
    }
    if matches!(settings.office_backend, OfficeBackend::OpenXml)
        && engine.should_attach_openxml_office_tool(instruction)
    {
        tools.push(Box::new(OpenXmlOfficeTool::new(
            config.workspace_dir.clone(),
        )));
    }
    if engine.should_attach_screen_html_export_tool(instruction) {
        tools.push(Box::new(ScreenHtmlExportTool::new(
            config.workspace_dir.clone(),
        )));
    }
    let mut agent = engine.build_agent(
        provider,
        &config,
        &skills_dir,
        tools,
        browser_surface_present,
        instruction,
    )?;
    if let Some(conversation_id) = task_context
        .conversation_id
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        agent.set_memory_session_id(Some(conversation_id.to_string()));
    }

    let mut seed_messages = Vec::new();
    seed_messages.extend(build_seed_history(task_context));
    if !seed_messages.is_empty() {
        agent.seed_history(&seed_messages);
    }

    let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<TurnEvent>(32);
    let instruction = engine.build_instruction(
        instruction,
        task_context.page_url.as_deref(),
        task_context.page_title.as_deref(),
        browser_surface_present,
    );

    let task = tokio::spawn(async move { agent.turn_streamed(&instruction, event_tx).await });

    while let Some(event) = event_rx.recv().await {
        if let Some(log_entry) = log_entry_for_turn_event(&event, &loaded_skill_versions) {
            transport.send(&log_entry)?;
        }
    }

    task.await
        .map_err(|err| PipeError::Protocol(format!("zeroclaw task join failed: {err}")))?
        .map_err(|err| PipeError::Protocol(err.to_string()))
}

fn build_provider(config: &ZeroClawConfig) -> Result<Box<dyn Provider>, PipeError> {
    let provider_name = config.default_provider.as_deref().unwrap_or("deepseek");
    let model_name = config.default_model.as_deref().unwrap_or("deepseek-chat");
    let runtime_options = providers::provider_runtime_options_from_config(config);
    let resolved_provider_name = if provider_name == "deepseek" {
        config
            .api_url
            .as_deref()
            .map(str::trim)
            .filter(|url| !url.is_empty())
            .map(|url| format!("custom:{url}"))
            .unwrap_or_else(|| provider_name.to_string())
    } else {
        provider_name.to_string()
    };
    let provider = providers::create_routed_provider_with_options(
        &resolved_provider_name,
        config.api_key.as_deref(),
        config.api_url.as_deref(),
        &config.reliability,
        &config.model_routes,
        model_name,
        &runtime_options,
    )
    .map_err(map_anyhow_to_pipe_error)?;

    Ok(Box::new(NonStreamingProvider::new(provider)))
}

fn map_anyhow_to_pipe_error(err: anyhow::Error) -> PipeError {
    PipeError::Protocol(err.to_string())
}

struct NonStreamingProvider {
    inner: Box<dyn Provider>,
}

impl NonStreamingProvider {
    fn new(inner: Box<dyn Provider>) -> Self {
        Self { inner }
    }
}

#[async_trait]
impl Provider for NonStreamingProvider {
    fn capabilities(&self) -> ProviderCapabilities {
        self.inner.capabilities()
    }

    async fn chat_with_system(
        &self,
        system_prompt: Option<&str>,
        message: &str,
        model: &str,
        temperature: f64,
    ) -> anyhow::Result<String> {
        self.inner
            .chat_with_system(system_prompt, message, model, temperature)
            .await
    }

    async fn chat_with_history(
        &self,
        messages: &[ChatMessage],
        model: &str,
        temperature: f64,
    ) -> anyhow::Result<String> {
        self.inner
            .chat_with_history(messages, model, temperature)
            .await
    }

    async fn chat(
        &self,
        request: ChatRequest<'_>,
        model: &str,
        temperature: f64,
    ) -> anyhow::Result<ChatResponse> {
        self.inner.chat(request, model, temperature).await
    }

    fn supports_streaming(&self) -> bool {
        false
    }

    fn supports_streaming_tool_events(&self) -> bool {
        false
    }

    fn stream_chat(
        &self,
        _request: ChatRequest<'_>,
        _model: &str,
        _temperature: f64,
        _options: StreamOptions,
    ) -> stream::BoxStream<'static, StreamResult<StreamEvent>> {
        stream::empty().boxed()
    }
}

fn build_seed_history(task_context: &CompatTaskContext) -> Vec<ChatMessage> {
    task_context
        .messages
        .iter()
        .filter_map(to_chat_message)
        .collect()
}

fn to_chat_message(message: &ConversationMessage) -> Option<ChatMessage> {
    let content = message.content.trim();
    if content.is_empty() {
        return None;
    }

    match message.role.as_str() {
        "user" => Some(ChatMessage::user(content)),
        "assistant" => Some(ChatMessage::assistant(content)),
        "system" => Some(ChatMessage::system(content)),
        _ => None,
    }
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::path::PathBuf;

    #[test]
    fn compat_runtime_source_no_longer_references_legacy_planner_preview() {
        let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let source = fs::read_to_string(manifest_dir.join("src/compat/runtime.rs")).unwrap();
        let preview_prefix = ["if let Some(preview) = crate::agent::", "planner::build_execution_preview("].concat();
        let plan_level_expr = ["level: ", "\"plan\".to_string(),"].concat();

        assert!(!source
            .lines()
            .any(|line| line.trim_start().starts_with(&preview_prefix)));
        assert!(!source.lines().any(|line| line.trim() == plan_level_expr));
    }
}