opencrabs 0.3.30

The autonomous, self-improving AI agent. Single Rust binary. Every channel. Install with: cargo install opencrabs
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

//! Channel Factory
//!
//! Shared factory for creating channel agent services at runtime.
//! Used by both static startup (ui.rs) and dynamic connection (whatsapp_connect tool).

use crate::brain::agent::AgentService;
use crate::brain::provider::Provider;
use crate::brain::tools::ToolRegistry;
use crate::config::{Config, VoiceConfig};
use crate::services::ServiceContext;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
use tokio::sync::Mutex;
use uuid::Uuid;

/// Factory for creating channel-specific AgentService instances.
///
/// Holds all shared state needed to spin up channel agents (Telegram, WhatsApp, etc.)
/// both at startup and dynamically at runtime via tools.
///
/// The `tool_registry` is set lazily via [`set_tool_registry`] to break the circular
/// dependency between tool registration and factory creation.
pub struct ChannelFactory {
    provider: Arc<dyn Provider>,
    service_context: ServiceContext,
    shared_brain: String,
    tool_registry: OnceLock<Arc<ToolRegistry>>,
    working_directory: PathBuf,
    brain_path: PathBuf,
    shared_session_id: Arc<Mutex<Option<Uuid>>>,
    config_rx: tokio::sync::watch::Receiver<Config>,
    session_updated_tx:
        OnceLock<tokio::sync::mpsc::UnboundedSender<crate::brain::agent::ChannelSessionEvent>>,
}

impl ChannelFactory {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        provider: Arc<dyn Provider>,
        service_context: ServiceContext,
        shared_brain: String,
        working_directory: PathBuf,
        brain_path: PathBuf,
        shared_session_id: Arc<Mutex<Option<Uuid>>>,
        config_rx: tokio::sync::watch::Receiver<Config>,
    ) -> Self {
        Self {
            provider,
            service_context,
            shared_brain,
            tool_registry: OnceLock::new(),
            working_directory,
            brain_path,
            shared_session_id,
            config_rx,
            session_updated_tx: OnceLock::new(),
        }
    }

    /// Wire in the TUI session-updated sender so channel agents trigger live TUI refresh.
    pub fn set_session_updated_tx(
        &self,
        tx: tokio::sync::mpsc::UnboundedSender<crate::brain::agent::ChannelSessionEvent>,
    ) {
        let _ = self.session_updated_tx.set(tx);
    }

    /// Set the tool registry (call once, after Arc<ToolRegistry> is created).
    pub fn set_tool_registry(&self, registry: Arc<ToolRegistry>) {
        let _ = self.tool_registry.set(registry);
    }

    /// Create a new AgentService configured for channel use.
    ///
    /// Channels that implement their own approval flow (WhatsApp, Telegram, Discord, Slack)
    /// pass an `override_approval_callback` per call — they must NOT have `auto_approve_tools`
    /// set, otherwise the override is ignored. A2A and headless tools that have no interactive
    /// user can set their own auto-approval via session context.
    pub async fn create_agent_service(&self) -> Arc<AgentService> {
        let config = self.config_rx.borrow().clone();
        let mut builder =
            AgentService::new(self.provider.clone(), self.service_context.clone(), &config)
                .await
                .with_system_brain(self.shared_brain.clone())
                .with_working_directory(self.working_directory.clone())
                .with_brain_path(self.brain_path.clone());

        if let Some(registry) = self.tool_registry.get() {
            builder = builder.with_tool_registry(registry.clone());
        }

        if let Some(tx) = self.session_updated_tx.get() {
            builder = builder.with_session_updated_tx(tx.clone());
        }

        Arc::new(builder)
    }

    pub fn shared_session_id(&self) -> Arc<Mutex<Option<Uuid>>> {
        self.shared_session_id.clone()
    }

    pub fn service_context(&self) -> ServiceContext {
        self.service_context.clone()
    }

    /// Get a clone of the config watch receiver for channels to subscribe to.
    pub fn config_rx(&self) -> tokio::sync::watch::Receiver<Config> {
        self.config_rx.clone()
    }

    /// Read the current voice config from the watch channel (always latest).
    pub fn voice_config(&self) -> VoiceConfig {
        self.config_rx.borrow().voice_config()
    }

    /// Read the current OpenAI TTS key from the watch channel (always latest).
    pub fn openai_tts_key(&self) -> Option<String> {
        let cfg = self.config_rx.borrow();
        cfg.providers
            .tts
            .as_ref()
            .and_then(|t| t.openai.as_ref())
            .and_then(|p| p.api_key.clone())
    }
}