Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ fastembed = "4"

# Encoding
base64 = "0.22"
hex = "0.4"

# Logging and tracing
tracing = "0.1"
Expand Down
20 changes: 10 additions & 10 deletions src/agent/cortex_chat.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,16 +52,16 @@ pub enum CortexChatEvent {
/// Prompt hook that forwards tool events to an mpsc channel for SSE streaming.
#[derive(Clone)]
struct CortexChatHook {
event_tx: mpsc::UnboundedSender<CortexChatEvent>,
event_tx: mpsc::Sender<CortexChatEvent>,
}

impl CortexChatHook {
fn new(event_tx: mpsc::UnboundedSender<CortexChatEvent>) -> Self {
fn new(event_tx: mpsc::Sender<CortexChatEvent>) -> Self {
Self { event_tx }
}

fn send(&self, event: CortexChatEvent) {
let _ = self.event_tx.send(event);
async fn send(&self, event: CortexChatEvent) {
let _ = self.event_tx.send(event).await;
}
}

Expand All @@ -75,7 +75,7 @@ impl<M: CompletionModel> PromptHook<M> for CortexChatHook {
) -> ToolCallHookAction {
self.send(CortexChatEvent::ToolStarted {
tool: tool_name.to_string(),
});
}).await;
ToolCallHookAction::Continue
}

Expand All @@ -95,7 +95,7 @@ impl<M: CompletionModel> PromptHook<M> for CortexChatHook {
self.send(CortexChatEvent::ToolCompleted {
tool: tool_name.to_string(),
result_preview: preview,
});
}).await;
HookAction::Continue
}

Expand Down Expand Up @@ -232,7 +232,7 @@ impl CortexChatSession {
thread_id: &str,
user_text: &str,
channel_context_id: Option<&str>,
) -> Result<mpsc::UnboundedReceiver<CortexChatEvent>, anyhow::Error> {
) -> Result<mpsc::Receiver<CortexChatEvent>, anyhow::Error> {
let _guard = self.send_lock.lock().await;

// Save the user message
Expand Down Expand Up @@ -271,7 +271,7 @@ impl CortexChatSession {
.tool_server_handle(self.tool_server.clone())
.build();

let (event_tx, event_rx) = mpsc::unbounded_channel();
let (event_tx, event_rx) = mpsc::channel(256);
let hook = CortexChatHook::new(event_tx.clone());

// Clone what the spawned task needs
Expand All @@ -296,7 +296,7 @@ impl CortexChatSession {
.await;
let _ = event_tx.send(CortexChatEvent::Done {
full_text: response,
});
}).await;
}
Err(error) => {
let error_text = format!("Cortex chat error: {error}");
Expand All @@ -305,7 +305,7 @@ impl CortexChatSession {
.await;
let _ = event_tx.send(CortexChatEvent::Error {
message: error_text,
});
}).await;
}
}
});
Expand Down
20 changes: 16 additions & 4 deletions src/api/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,12 @@ use super::{
};

use axum::Router;
use axum::extract::DefaultBodyLimit;
use axum::http::{StatusCode, Uri, header};
use axum::response::{Html, IntoResponse, Response};
use axum::routing::{delete, get, post, put};
use rust_embed::Embed;
use tower_http::cors::{Any, CorsLayer};
use tower_http::cors::CorsLayer;

use std::net::SocketAddr;
use std::sync::Arc;
Expand All @@ -32,9 +33,19 @@ pub async fn start_http_server(
shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> anyhow::Result<tokio::task::JoinHandle<()>> {
let cors = CorsLayer::new()
.allow_origin(Any)
.allow_methods(Any)
.allow_headers(Any);
.allow_origin(tower_http::cors::AllowOrigin::mirror_request())
.allow_methods([
axum::http::Method::GET,
axum::http::Method::POST,
axum::http::Method::PUT,
axum::http::Method::DELETE,
axum::http::Method::OPTIONS,
])
.allow_headers([
header::CONTENT_TYPE,
header::AUTHORIZATION,
header::ACCEPT,
]);

let api_routes = Router::new()
.route("/health", get(system::health))
Expand Down Expand Up @@ -137,6 +148,7 @@ pub async fn start_http_server(
.nest("/api", api_routes)
.fallback(static_handler)
.layer(cors)
.layer(DefaultBodyLimit::max(10 * 1024 * 1024)) // 10 MiB
.with_state(state);

let listener = tokio::net::TcpListener::bind(bind).await?;
Expand Down
136 changes: 124 additions & 12 deletions src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -125,16 +125,27 @@ impl<'de> serde::Deserialize<'de> for ApiType {
}

/// Configuration for a single LLM provider.
#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct ProviderConfig {
pub api_type: ApiType,
pub base_url: String,
pub api_key: String,
pub name: Option<String>,
}

impl std::fmt::Debug for ProviderConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ProviderConfig")
.field("api_type", &self.api_type)
.field("base_url", &self.base_url)
.field("api_key", &"[REDACTED]")
.field("name", &self.name)
.finish()
}
}

/// LLM provider credentials (instance-level).
#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct LlmConfig {
pub anthropic_key: Option<String>,
pub openai_key: Option<String>,
Expand All @@ -157,6 +168,32 @@ pub struct LlmConfig {
pub providers: HashMap<String, ProviderConfig>,
}

impl std::fmt::Debug for LlmConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LlmConfig")
.field("anthropic_key", &self.anthropic_key.as_ref().map(|_| "[REDACTED]"))
.field("openai_key", &self.openai_key.as_ref().map(|_| "[REDACTED]"))
.field("openrouter_key", &self.openrouter_key.as_ref().map(|_| "[REDACTED]"))
.field("zhipu_key", &self.zhipu_key.as_ref().map(|_| "[REDACTED]"))
.field("groq_key", &self.groq_key.as_ref().map(|_| "[REDACTED]"))
.field("together_key", &self.together_key.as_ref().map(|_| "[REDACTED]"))
.field("fireworks_key", &self.fireworks_key.as_ref().map(|_| "[REDACTED]"))
.field("deepseek_key", &self.deepseek_key.as_ref().map(|_| "[REDACTED]"))
.field("xai_key", &self.xai_key.as_ref().map(|_| "[REDACTED]"))
.field("mistral_key", &self.mistral_key.as_ref().map(|_| "[REDACTED]"))
.field("gemini_key", &self.gemini_key.as_ref().map(|_| "[REDACTED]"))
.field("ollama_key", &self.ollama_key.as_ref().map(|_| "[REDACTED]"))
.field("ollama_base_url", &self.ollama_base_url)
.field("opencode_zen_key", &self.opencode_zen_key.as_ref().map(|_| "[REDACTED]"))
.field("nvidia_key", &self.nvidia_key.as_ref().map(|_| "[REDACTED]"))
.field("minimax_key", &self.minimax_key.as_ref().map(|_| "[REDACTED]"))
.field("moonshot_key", &self.moonshot_key.as_ref().map(|_| "[REDACTED]"))
.field("zai_coding_plan_key", &self.zai_coding_plan_key.as_ref().map(|_| "[REDACTED]"))
.field("providers", &self.providers)
.finish()
}
}

impl LlmConfig {
/// Check if any provider configuration is set.
pub fn has_any_key(&self) -> bool {
Expand Down Expand Up @@ -195,7 +232,7 @@ const NVIDIA_PROVIDER_BASE_URL: &str = "https://integrate.api.nvidia.com";
pub(crate) const GEMINI_PROVIDER_BASE_URL: &str = "https://generativelanguage.googleapis.com/v1beta/openai";

/// Defaults inherited by all agents. Individual agents can override any field.
#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct DefaultsConfig {
pub routing: RoutingConfig,
pub max_concurrent_branches: usize,
Expand All @@ -219,6 +256,31 @@ pub struct DefaultsConfig {
pub worker_log_mode: crate::settings::WorkerLogMode,
}

impl std::fmt::Debug for DefaultsConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DefaultsConfig")
.field("routing", &self.routing)
.field("max_concurrent_branches", &self.max_concurrent_branches)
.field("max_concurrent_workers", &self.max_concurrent_workers)
.field("max_turns", &self.max_turns)
.field("branch_max_turns", &self.branch_max_turns)
.field("context_window", &self.context_window)
.field("compaction", &self.compaction)
.field("memory_persistence", &self.memory_persistence)
.field("coalesce", &self.coalesce)
.field("ingestion", &self.ingestion)
.field("cortex", &self.cortex)
.field("browser", &self.browser)
.field("mcp", &self.mcp)
.field("brave_search_key", &self.brave_search_key.as_ref().map(|_| "[REDACTED]"))
.field("history_backfill_count", &self.history_backfill_count)
.field("cron", &self.cron)
.field("opencode", &self.opencode)
.field("worker_log_mode", &self.worker_log_mode)
.finish()
}
}

/// MCP server configuration.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct McpServerConfig {
Expand Down Expand Up @@ -782,7 +844,7 @@ pub struct MessagingConfig {
pub twitch: Option<TwitchConfig>,
}

#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct DiscordConfig {
pub enabled: bool,
pub token: String,
Expand All @@ -792,6 +854,17 @@ pub struct DiscordConfig {
pub allow_bot_messages: bool,
}

impl std::fmt::Debug for DiscordConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DiscordConfig")
.field("enabled", &self.enabled)
.field("token", &"[REDACTED]")
.field("dm_allowed_users", &self.dm_allowed_users)
.field("allow_bot_messages", &self.allow_bot_messages)
.finish()
}
}

/// A single slash command definition for the Slack adapter.
///
/// Maps a Slack slash command (e.g. `/ask`) to a target agent.
Expand All @@ -806,7 +879,7 @@ pub struct SlackCommandConfig {
pub description: Option<String>,
}

#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct SlackConfig {
pub enabled: bool,
pub bot_token: String,
Expand All @@ -817,6 +890,18 @@ pub struct SlackConfig {
pub commands: Vec<SlackCommandConfig>,
}

impl std::fmt::Debug for SlackConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SlackConfig")
.field("enabled", &self.enabled)
.field("bot_token", &"[REDACTED]")
.field("app_token", &"[REDACTED]")
.field("dm_allowed_users", &self.dm_allowed_users)
.field("commands", &self.commands)
.finish()
}
}

/// Hot-reloadable Discord permission filters.
///
/// Derived from bindings + discord config. Shared with the Discord adapter
Expand Down Expand Up @@ -949,14 +1034,24 @@ impl DiscordPermissions {
}
}

#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct TelegramConfig {
pub enabled: bool,
pub token: String,
/// User IDs allowed to DM the bot. If empty, DMs are ignored entirely.
pub dm_allowed_users: Vec<String>,
}

impl std::fmt::Debug for TelegramConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("TelegramConfig")
.field("enabled", &self.enabled)
.field("token", &"[REDACTED]")
.field("dm_allowed_users", &self.dm_allowed_users)
.finish()
}
}

/// Hot-reloadable Telegram permission filters.
///
/// Shared with the Telegram adapter via `Arc<ArcSwap<..>>` for hot-reloading.
Expand Down Expand Up @@ -1011,7 +1106,7 @@ impl TelegramPermissions {
}
}

#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct TwitchConfig {
pub enabled: bool,
pub username: String,
Expand All @@ -1022,6 +1117,18 @@ pub struct TwitchConfig {
pub trigger_prefix: Option<String>,
}

impl std::fmt::Debug for TwitchConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("TwitchConfig")
.field("enabled", &self.enabled)
.field("username", &self.username)
.field("oauth_token", &"[REDACTED]")
.field("channels", &self.channels)
.field("trigger_prefix", &self.trigger_prefix)
.finish()
}
}

/// Hot-reloadable Twitch permission filters.
///
/// Shared with the Twitch adapter via `Arc<ArcSwap<..>>` for hot-reloading.
Expand Down Expand Up @@ -2139,18 +2246,23 @@ impl Config {
.providers
.into_iter()
.map(|(provider_id, config)| {
(
let api_key = resolve_env_value(&config.api_key).ok_or_else(|| {
anyhow::anyhow!(
"failed to resolve API key for provider '{}'",
provider_id
)
})?;
Ok((
provider_id.to_lowercase(),
ProviderConfig {
api_type: config.api_type,
base_url: config.base_url,
api_key: resolve_env_value(&config.api_key)
.expect("Failed to resolve API key for provider"),
api_key,
name: config.name,
},
)
))
})
.collect(),
.collect::<anyhow::Result<_>>()?,
};

if let Some(anthropic_key) = llm.anthropic_key.clone() {
Expand Down
Loading