Tara Menon
697c2c63b2
Adds [browser].node_path config option so camoufox-browser can use an explicit npx path (essential for nvm/fnm/asdf users). Resolution order (cross-platform): 1. config.browser.node_path — explicit user override 2. 'npx.cmd' on Windows, 'npx' on Linux/macOS — PATH lookup Changes: - BrowserSection config: added node_path: Option<String> - AgentConfig: now carries browser section (was dropped in From<GyandootConfig>) - browser_manager::init_browser(node_path): accepts optional path - All call sites updated (run(), init_session(), subagents, tests) - Error messages now hint at browser.node_path config
970 lines
39 KiB
Rust
970 lines
39 KiB
Rust
//! Production-grade agent inference loop.
|
|
//!
|
|
//! `AgentLoop` manages the full cycle: receive message → build context →
|
|
//! call LLM → execute tools → respond → compress context. Supports
|
|
//! interactive (stdin/stdout) and non-interactive (gateway/sub-agent) modes,
|
|
//! retry logic with exponential backoff, error classification, and streaming
|
|
//! response delivery.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use gyandoot_browser::BrowserClient;
|
|
use gyandoot_core::{
|
|
AgentConfig, AgentError, AgentResult, AgentStatus, ConversationContext, Message,
|
|
MessageRole,
|
|
};
|
|
use gyandoot_memory::palace::MemPalace;
|
|
use gyandoot_providers::{Provider, ProviderResponse};
|
|
use gyandoot_security::sanitizer::Sanitizer;
|
|
use gyandoot_skills::SkillsManager;
|
|
use tokio::sync::Mutex;
|
|
use tracing::{debug, error, info, span, warn, Level};
|
|
|
|
use crate::context::ContextCompressor;
|
|
use crate::tools::ToolExecutor;
|
|
|
|
// ─── CLI color constants ──────────────────────────────────────────────
|
|
#[allow(dead_code)]
|
|
mod cli_color {
|
|
pub const RESET: &str = "\x1b[0m";
|
|
pub const BOLD: &str = "\x1b[1m";
|
|
pub const DIM: &str = "\x1b[2m";
|
|
pub const CYAN: &str = "\x1b[36m";
|
|
pub const BRIGHT_CYAN: &str = "\x1b[96m";
|
|
pub const BRIGHT_MAGENTA: &str = "\x1b[95m";
|
|
pub const GREEN: &str = "\x1b[32m";
|
|
pub const YELLOW: &str = "\x1b[33m";
|
|
}
|
|
|
|
// ─── Error classification ───────────────────────────────────────────────
|
|
|
|
/// Categories of errors the agent loop can encounter during LLM calls.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum AgentErrorKind {
|
|
/// Provider rejected authentication credentials.
|
|
Auth,
|
|
/// API rate-limit exceeded.
|
|
RateLimit,
|
|
/// Request timed out (network or provider-side).
|
|
Timeout,
|
|
/// Server returned a 5xx or similar transient error.
|
|
ServerError,
|
|
/// Context window exceeded the model's maximum capacity.
|
|
ContextOverflow,
|
|
/// All other / unclassified errors.
|
|
Other,
|
|
}
|
|
|
|
impl AgentErrorKind {
|
|
/// Classify a provider- or agent-error string into a structured kind.
|
|
pub fn classify(error_text: &str) -> Self {
|
|
let lower = error_text.to_lowercase();
|
|
if lower.contains("rate limit")
|
|
|| lower.contains("rate_limit")
|
|
|| lower.contains("429")
|
|
{
|
|
Self::RateLimit
|
|
} else if lower.contains("auth")
|
|
|| lower.contains("unauthorized")
|
|
|| lower.contains("401")
|
|
|| lower.contains("api key")
|
|
{
|
|
Self::Auth
|
|
} else if lower.contains("timeout")
|
|
|| lower.contains("timed out")
|
|
|| lower.contains("connection reset")
|
|
{
|
|
Self::Timeout
|
|
} else if lower.contains("5")
|
|
|| lower.contains("server error")
|
|
|| lower.contains("internal server error")
|
|
{
|
|
Self::ServerError
|
|
} else if lower.contains("context length")
|
|
|| lower.contains("maximum context")
|
|
|| lower.contains("token limit")
|
|
|| lower.contains("context overflow")
|
|
{
|
|
Self::ContextOverflow
|
|
} else {
|
|
Self::Other
|
|
}
|
|
}
|
|
|
|
/// True for errors that are safe to retry.
|
|
pub fn is_retryable(&self) -> bool {
|
|
matches!(self, Self::RateLimit | Self::Timeout | Self::ServerError | Self::Other)
|
|
}
|
|
}
|
|
|
|
// ─── AgentLoop ──────────────────────────────────────────────────────────
|
|
|
|
/// Configuration for the agent-loop retry behaviour.
|
|
#[derive(Debug, Clone)]
|
|
pub struct RetryConfig {
|
|
/// Maximum number of retry attempts (including the initial call).
|
|
pub max_attempts: u32,
|
|
/// Base delay in milliseconds for exponential back-off.
|
|
pub base_delay_ms: u64,
|
|
/// Maximum delay cap in milliseconds.
|
|
pub max_delay_ms: u64,
|
|
}
|
|
|
|
impl Default for RetryConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
max_attempts: 3,
|
|
base_delay_ms: 1000,
|
|
max_delay_ms: 10000,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Structured output from a conversation turn (non-interactive callers).
|
|
#[derive(Debug, Clone)]
|
|
pub struct ConversationOutput {
|
|
pub text: String,
|
|
pub tool_results: Vec<String>,
|
|
}
|
|
|
|
// ─── AgentLoop ──────────────────────────────────────────────────────────
|
|
|
|
/// The core agent inference loop.
|
|
///
|
|
/// Reads messages, builds conversation context, calls the LLM provider,
|
|
/// executes tools, and manages context window limits.
|
|
pub struct AgentLoop {
|
|
config: AgentConfig,
|
|
provider: Box<dyn Provider + Send + Sync>,
|
|
tool_executor: Arc<Mutex<ToolExecutor>>,
|
|
context_compressor: ContextCompressor,
|
|
memory: Arc<Mutex<MemPalace>>,
|
|
_skills_manager: SkillsManager,
|
|
sanitizer: Sanitizer,
|
|
status: Arc<Mutex<AgentStatus>>,
|
|
retry_cfg: RetryConfig,
|
|
#[allow(dead_code)]
|
|
browser_client: Option<BrowserClient>,
|
|
/// Accumulated messages for stateful session mode (TUI / multi-turn).
|
|
session_messages: Mutex<Option<Vec<Message>>>,
|
|
}
|
|
|
|
impl AgentLoop {
|
|
/// Create a new agent loop from a fully-resolved configuration.
|
|
pub fn new(config: AgentConfig) -> Self {
|
|
let provider = create_provider(&config);
|
|
let tool_executor = Arc::new(Mutex::new(ToolExecutor::with_defaults()));
|
|
let context_compressor = ContextCompressor::new(config.max_tokens);
|
|
let memory = Arc::new(Mutex::new(MemPalace::new("default")));
|
|
let skills_manager = SkillsManager::new();
|
|
|
|
Self {
|
|
config,
|
|
provider,
|
|
tool_executor,
|
|
context_compressor,
|
|
memory,
|
|
_skills_manager: skills_manager,
|
|
sanitizer: Sanitizer::new(gyandoot_security::sanitizer::SanitizerConfig::default()),
|
|
status: Arc::new(Mutex::new(AgentStatus::Idle)),
|
|
retry_cfg: RetryConfig::default(),
|
|
browser_client: None,
|
|
session_messages: Mutex::new(None),
|
|
}
|
|
}
|
|
|
|
/// Create an agent loop with a custom retry policy.
|
|
pub fn with_retry(mut self, cfg: RetryConfig) -> Self {
|
|
self.retry_cfg = cfg;
|
|
self
|
|
}
|
|
|
|
// ── Interactive loop (stdin → stdout) ──────────────────────────────
|
|
|
|
/// Run the agent loop interactively, reading from stdin and writing to
|
|
/// stdout until the user types `exit` / `quit` or an unrecoverable error
|
|
/// occurs.
|
|
pub async fn run(&self) -> anyhow::Result<()> {
|
|
let span = span!(Level::INFO, "agent_loop", agent = %self.config.name);
|
|
let _guard = span.enter();
|
|
info!("Agent loop started (interactive)");
|
|
|
|
// ── Auto-init Camoufox and register browser tools ──────────────
|
|
{
|
|
let mut tool_executor = self.tool_executor.lock().await;
|
|
match crate::browser_manager::init_browser(self.config.browser.node_path.as_deref()).await {
|
|
crate::browser_manager::BrowserInitResult::Ready(client) => {
|
|
info!("Camoufox browser ready — registering web_search and web_browse tools");
|
|
crate::browser_tools::register_browser_tools(&mut tool_executor, client.clone());
|
|
// Store client for potential later use
|
|
// (browser_client field is set via unsafe trick — skip for now)
|
|
}
|
|
crate::browser_manager::BrowserInitResult::Unavailable(msg) => {
|
|
warn!("Camoufox not available: {}", msg);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Identity & memory injection on startup ─────────────────────
|
|
let (agent_name, agent_role, user_name, user_role) = {
|
|
let memory = self.memory.lock().await;
|
|
let (_l0, _) = memory.identity_get().unwrap_or_default();
|
|
|
|
// Priority: config identity > config.agent.name > defaults
|
|
let a_name = if !self.config.identity.agent_name.is_empty() {
|
|
self.config.identity.agent_name.clone()
|
|
} else {
|
|
self.config.name.clone()
|
|
};
|
|
|
|
let a_role = if !self.config.identity.agent_role.is_empty() {
|
|
self.config.identity.agent_role.clone()
|
|
} else {
|
|
"AI Assistant built by Gyannetra".to_string()
|
|
};
|
|
|
|
let u_name = if !self.config.identity.user_name.is_empty() {
|
|
self.config.identity.user_name.clone()
|
|
} else {
|
|
"the User".to_string()
|
|
};
|
|
|
|
let u_role = if !self.config.identity.user_role.is_empty() {
|
|
self.config.identity.user_role.clone()
|
|
} else {
|
|
"User".to_string()
|
|
};
|
|
|
|
(a_name, a_role, u_name, u_role)
|
|
};
|
|
|
|
// Build system prompt with identity + tool capabilities + dialogue memory
|
|
let mut system_prompt = self.config.system_prompt.clone();
|
|
if !agent_name.is_empty() {
|
|
system_prompt = format!(
|
|
"{}\n\n## Your Identity\n- Name: {}\n- Role: {}\n- You are speaking to: {} ({})\n\n## Your Capabilities\nYou have access to these tools:\n- **web_search** — Search the web using anti-detection browser. Use for any online query.\n- **web_browse** — Visit and interact with any website. Use for scraping pages.\n- **bash** — Execute shell commands (curl, python, git)\n- **read/write/edit** — File operations\n\n**For web tasks:** Always use `web_search` or `web_browse` first. If they fail, fall back to `bash` with Python requests.\n\nDo NOT say you can't access websites unless both `web_search`/`web_browse` AND `bash` with Python have failed.\n\n## Memory\nYou have a verbatim dialogue log. Use past conversations as context.\",",
|
|
system_prompt,
|
|
agent_name,
|
|
if agent_role.is_empty() { "AI Assistant" } else { &agent_role },
|
|
if user_name.is_empty() { "the User" } else { &user_name },
|
|
if user_role.is_empty() { "User" } else { &user_role },
|
|
);
|
|
}
|
|
|
|
// Inject recent dialogue from verbatim memory
|
|
{
|
|
let memory = self.memory.lock().await;
|
|
if let Ok(dialogue) = memory.dialogue_read(2000).await {
|
|
if !dialogue.is_empty() {
|
|
system_prompt = format!(
|
|
"{}\n\n## Past Conversations (Verbatim Memory)\n{}\n---\nContinue the conversation naturally.",
|
|
system_prompt, dialogue
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut messages: Vec<Message> =
|
|
vec![Message::new(MessageRole::System, &system_prompt)];
|
|
|
|
loop {
|
|
// 1. Read user input
|
|
let user_input = self.read_input().await?;
|
|
|
|
// 2. Handle built-in slash commands
|
|
if user_input.starts_with('/') {
|
|
let should_quit = self.handle_command(&user_input).await;
|
|
if should_quit {
|
|
break;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
// 3. Check exit conditions
|
|
if self.should_exit(&user_input) {
|
|
info!("User requested exit");
|
|
break;
|
|
}
|
|
|
|
// 3. Sanitize
|
|
let clean_input = self.sanitizer.sanitize_input(&user_input, &gyandoot_security::sanitizer::SanitizerConfig::default());
|
|
messages.push(Message::new(MessageRole::User, clean_input.clone()));
|
|
|
|
// 4. Run the core turn (LLM + tool execution)
|
|
let new_messages = self.execute_turn(messages).await?;
|
|
messages = new_messages;
|
|
|
|
// 5. Store conversation turn in memory (transcript + verbatim dialogue)
|
|
{
|
|
let memory = self.memory.lock().await;
|
|
memory
|
|
.transcript_append("user", &user_input)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store user message: {e}"));
|
|
memory
|
|
.dialogue_append("user", &user_input)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store user dialogue: {e}"));
|
|
|
|
if let Some(last) = messages.last() {
|
|
if last.role == MessageRole::Assistant {
|
|
memory
|
|
.transcript_append("assistant", &last.content)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store assistant message: {e}"));
|
|
memory
|
|
.dialogue_append("assistant", &last.content)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store assistant dialogue: {e}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
// 6. Compress context if token budget is tight
|
|
let total_tokens: usize =
|
|
messages.iter().map(|m| estimate_tokens(&m.content)).sum();
|
|
let limit = self.config.max_tokens;
|
|
if total_tokens > (limit * 3 / 4) {
|
|
debug!(
|
|
"Context at ~{:.0}% of limit ({total_tokens}/{limit}), compressing",
|
|
(total_tokens as f64 / limit as f64) * 100.0
|
|
);
|
|
messages = self.context_compressor.compress(messages, limit).await;
|
|
}
|
|
|
|
// 7. Update status
|
|
*self.status.lock().await = AgentStatus::Idle;
|
|
}
|
|
|
|
info!("Agent loop finished");
|
|
Ok(())
|
|
}
|
|
|
|
// ── Non-interactive (gateway / sub-agent mode) ─────────────────────
|
|
|
|
/// Run a single conversation turn given an incoming user message.
|
|
pub async fn run_conversation(
|
|
&self,
|
|
message: &str,
|
|
) -> AgentResult<ConversationOutput> {
|
|
let span = span!(Level::INFO, "conversation", agent = %self.config.name);
|
|
let _guard = span.enter();
|
|
debug!("run_conversation: processing message");
|
|
|
|
let mut messages = vec![
|
|
Message::new(MessageRole::System, &self.config.system_prompt),
|
|
Message::new(MessageRole::User, message),
|
|
];
|
|
|
|
let (result_messages, output) = self.execute_turn_with_output(messages).await?;
|
|
messages = result_messages;
|
|
|
|
// Store in memory
|
|
{
|
|
let memory = self.memory.lock().await;
|
|
memory
|
|
.transcript_append("user", message)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store user message: {e}"));
|
|
if let Some(last) = messages.last() {
|
|
if last.role == MessageRole::Assistant {
|
|
memory
|
|
.transcript_append("assistant", &last.content)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store assistant message: {e}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
*self.status.lock().await = AgentStatus::Idle;
|
|
Ok(output)
|
|
}
|
|
|
|
/// Get the current agent status.
|
|
pub async fn get_status(&self) -> AgentStatus {
|
|
self.status.lock().await.clone()
|
|
}
|
|
|
|
/// Read-only access to agent configuration.
|
|
pub fn config(&self) -> &AgentConfig {
|
|
&self.config
|
|
}
|
|
|
|
// ── Stateful session methods (for TUI / multi-turn callers) ─────────
|
|
|
|
/// Initialize a conversation session: browser init, identity injection,
|
|
/// memory recall, and system prompt construction. Must be called before
|
|
/// the first `session_turn()`.
|
|
pub async fn init_session(&self) -> AgentResult<()> {
|
|
// Browser init
|
|
{
|
|
let mut tool_executor = self.tool_executor.lock().await;
|
|
match crate::browser_manager::init_browser(self.config.browser.node_path.as_deref()).await {
|
|
crate::browser_manager::BrowserInitResult::Ready(client) => {
|
|
info!("Camoufox browser ready — registering web_search and web_browse tools");
|
|
crate::browser_tools::register_browser_tools(&mut tool_executor, client.clone());
|
|
}
|
|
crate::browser_manager::BrowserInitResult::Unavailable(msg) => {
|
|
warn!("Camoufox not available: {}", msg);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Identity & memory injection
|
|
let (agent_name, agent_role, user_name, user_role) = {
|
|
let memory = self.memory.lock().await;
|
|
let (_l0, _) = memory.identity_get().unwrap_or_default();
|
|
|
|
let a_name = if !self.config.identity.agent_name.is_empty() {
|
|
self.config.identity.agent_name.clone()
|
|
} else {
|
|
self.config.name.clone()
|
|
};
|
|
|
|
let a_role = if !self.config.identity.agent_role.is_empty() {
|
|
self.config.identity.agent_role.clone()
|
|
} else {
|
|
"AI Assistant built by Gyannetra".to_string()
|
|
};
|
|
|
|
let u_name = if !self.config.identity.user_name.is_empty() {
|
|
self.config.identity.user_name.clone()
|
|
} else {
|
|
"the User".to_string()
|
|
};
|
|
|
|
let u_role = if !self.config.identity.user_role.is_empty() {
|
|
self.config.identity.user_role.clone()
|
|
} else {
|
|
"User".to_string()
|
|
};
|
|
|
|
(a_name, a_role, u_name, u_role)
|
|
};
|
|
|
|
// Build system prompt with identity + capabilities
|
|
let mut system_prompt = self.config.system_prompt.clone();
|
|
if !agent_name.is_empty() {
|
|
system_prompt = format!(
|
|
"{}\n\n## Your Identity\n- Name: {}\n- Role: {}\n- You are speaking to: {} ({})\n\n## Your Capabilities\nYou have access to these tools:\n- **web_search** — Search the web using anti-detection browser. Use for any online query.\n- **web_browse** — Visit and interact with any website. Use for scraping pages.\n- **bash** — Execute shell commands (curl, python, git)\n- **read/write/edit** — File operations\n\n**For web tasks:** Always use `web_search` or `web_browse` first. If they fail, fall back to `bash` with Python requests.\n\nDo NOT say you can't access websites unless both `web_search`/`web_browse` AND `bash` with Python have failed.\n\n## Memory\nYou have a verbatim dialogue log. Use past conversations as context.\n",
|
|
system_prompt,
|
|
agent_name,
|
|
if agent_role.is_empty() { "AI Assistant" } else { &agent_role },
|
|
if user_name.is_empty() { "the User" } else { &user_name },
|
|
if user_role.is_empty() { "User" } else { &user_role },
|
|
);
|
|
}
|
|
|
|
// Inject recent dialogue from memory
|
|
{
|
|
let memory = self.memory.lock().await;
|
|
if let Ok(dialogue) = memory.dialogue_read(2000).await {
|
|
if !dialogue.is_empty() {
|
|
system_prompt = format!(
|
|
"{}\n\n## Past Conversations (Verbatim Memory)\n{}\n---\nContinue the conversation naturally.",
|
|
system_prompt, dialogue
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
let messages = vec![Message::new(MessageRole::System, &system_prompt)];
|
|
*self.session_messages.lock().await = Some(messages);
|
|
info!("Session initialised with identity and memory");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Run a stateful conversation turn. Accumulates messages across turns
|
|
/// (unlike `run_conversation` which is stateless). Returns the assistant
|
|
/// response text.
|
|
pub async fn session_turn(&self, user_message: &str) -> AgentResult<String> {
|
|
let span = span!(Level::INFO, "session_turn", agent = %self.config.name);
|
|
let _guard = span.enter();
|
|
|
|
// Ensure session is initialised
|
|
let mut session_guard = self.session_messages.lock().await;
|
|
if session_guard.is_none() {
|
|
drop(session_guard);
|
|
self.init_session().await?;
|
|
session_guard = self.session_messages.lock().await;
|
|
}
|
|
|
|
let mut messages = session_guard.take().unwrap_or_default();
|
|
|
|
// Sanitize and add user message
|
|
let clean_input = self.sanitizer.sanitize_input(user_message, &gyandoot_security::sanitizer::SanitizerConfig::default());
|
|
messages.push(Message::new(MessageRole::User, clean_input.clone()));
|
|
|
|
// Run the core turn
|
|
let new_messages = self.execute_turn(messages).await?;
|
|
messages = new_messages;
|
|
|
|
// Extract assistant's last text response
|
|
let assistant_text = messages
|
|
.iter()
|
|
.rev()
|
|
.find(|m| m.role == MessageRole::Assistant)
|
|
.map(|m| m.content.clone())
|
|
.unwrap_or_default();
|
|
|
|
// Store in memory
|
|
{
|
|
let memory = self.memory.lock().await;
|
|
memory
|
|
.transcript_append("user", user_message)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store user message: {e}"));
|
|
memory
|
|
.dialogue_append("user", user_message)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store user dialogue: {e}"));
|
|
if !assistant_text.is_empty() {
|
|
memory
|
|
.transcript_append("assistant", &assistant_text)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store assistant message: {e}"));
|
|
memory
|
|
.dialogue_append("assistant", &assistant_text)
|
|
.await
|
|
.unwrap_or_else(|e| warn!("Failed to store assistant dialogue: {e}"));
|
|
}
|
|
}
|
|
|
|
// Compress context if token budget is tight
|
|
let total_tokens: usize =
|
|
messages.iter().map(|m| estimate_tokens(&m.content)).sum();
|
|
let limit = self.config.max_tokens;
|
|
if total_tokens > (limit * 3 / 4) {
|
|
debug!(
|
|
"Context at ~{:.0}% of limit ({total_tokens}/{limit}), compressing",
|
|
(total_tokens as f64 / limit as f64) * 100.0
|
|
);
|
|
messages = self.context_compressor.compress(messages, limit).await;
|
|
}
|
|
|
|
// Save messages back for next turn
|
|
*session_guard = Some(messages);
|
|
|
|
*self.status.lock().await = AgentStatus::Idle;
|
|
Ok(assistant_text)
|
|
}
|
|
|
|
// ── Internal methods ───────────────────────────────────────────────
|
|
|
|
/// Execute a single turn: LLM call → tool calls → LLM follow-up.
|
|
async fn execute_turn(&self, mut messages: Vec<Message>) -> AgentResult<Vec<Message>> {
|
|
let context = self
|
|
.context_compressor
|
|
.build_context(&messages, &self.config.system_prompt, &self.tool_executor.lock().await.definitions(), self.config.max_tokens);
|
|
|
|
*self.status.lock().await = AgentStatus::Thinking;
|
|
|
|
// Show thinking animation while waiting for LLM
|
|
let spinner_handle = self.start_thinking_indicator();
|
|
let response = self.call_with_retry(context).await;
|
|
self.stop_thinking_indicator(spinner_handle).await;
|
|
let response = response?;
|
|
|
|
messages.push(Message::new(MessageRole::Assistant, &response.content)
|
|
.with_tool_calls(response.tool_calls.clone().unwrap_or_default()));
|
|
|
|
if let Some(calls) = response.tool_calls {
|
|
if !calls.is_empty() {
|
|
*self.status.lock().await = AgentStatus::WaitingForTool;
|
|
for call in calls {
|
|
debug!("Executing tool: {}", call.name);
|
|
let result = self.tool_executor.lock().await.execute(&call).await.map_err(|e| {
|
|
AgentError::ToolExecution(format!("{}: {}", call.name, e))
|
|
})?;
|
|
let output = if result.success {
|
|
result.output
|
|
} else {
|
|
format!("ERROR: {}", result.error.unwrap_or_default())
|
|
};
|
|
messages.push(
|
|
Message::new(MessageRole::Tool, &output)
|
|
.with_tool_result(output),
|
|
);
|
|
}
|
|
|
|
let follow_context = self
|
|
.context_compressor
|
|
.build_context(&messages, &self.config.system_prompt, &self.tool_executor.lock().await.definitions(), self.config.max_tokens);
|
|
|
|
*self.status.lock().await = AgentStatus::Thinking;
|
|
let follow_up = self.call_with_retry(follow_context).await?;
|
|
messages.push(
|
|
Message::new(MessageRole::Assistant, &follow_up.content)
|
|
.with_tool_calls(follow_up.tool_calls.unwrap_or_default()),
|
|
);
|
|
self.display_output(&follow_up.content);
|
|
} else {
|
|
self.display_output(&response.content);
|
|
}
|
|
} else {
|
|
self.display_output(&response.content);
|
|
}
|
|
|
|
Ok(messages)
|
|
}
|
|
|
|
/// Execute a single turn and return structured output.
|
|
async fn execute_turn_with_output(
|
|
&self,
|
|
mut messages: Vec<Message>,
|
|
) -> AgentResult<(Vec<Message>, ConversationOutput)> {
|
|
let context = self
|
|
.context_compressor
|
|
.build_context(&messages, &self.config.system_prompt, &self.tool_executor.lock().await.definitions(), self.config.max_tokens);
|
|
|
|
*self.status.lock().await = AgentStatus::Thinking;
|
|
let response = self.call_with_retry(context).await?;
|
|
|
|
let mut assistant_text = response.content.clone();
|
|
let mut tool_results: Vec<String> = Vec::new();
|
|
|
|
messages.push(Message::new(MessageRole::Assistant, &response.content)
|
|
.with_tool_calls(response.tool_calls.clone().unwrap_or_default()));
|
|
|
|
if let Some(calls) = response.tool_calls {
|
|
if !calls.is_empty() {
|
|
*self.status.lock().await = AgentStatus::WaitingForTool;
|
|
for call in calls {
|
|
let result = self.tool_executor.lock().await.execute(&call).await.map_err(|e| {
|
|
AgentError::ToolExecution(format!("{}: {}", call.name, e))
|
|
})?;
|
|
let output = if result.success {
|
|
result.output.clone()
|
|
} else {
|
|
format!("ERROR: {}", result.error.unwrap_or_default())
|
|
};
|
|
tool_results.push(output.clone());
|
|
messages.push(
|
|
Message::new(MessageRole::Tool, &output)
|
|
.with_tool_result(output),
|
|
);
|
|
}
|
|
|
|
let follow_context = self
|
|
.context_compressor
|
|
.build_context(&messages, &self.config.system_prompt, &self.tool_executor.lock().await.definitions(), self.config.max_tokens);
|
|
|
|
*self.status.lock().await = AgentStatus::Thinking;
|
|
let follow_up = self.call_with_retry(follow_context).await?;
|
|
assistant_text = follow_up.content.clone();
|
|
messages.push(
|
|
Message::new(MessageRole::Assistant, &follow_up.content)
|
|
.with_tool_calls(follow_up.tool_calls.unwrap_or_default()),
|
|
);
|
|
}
|
|
}
|
|
|
|
Ok((messages, ConversationOutput { text: assistant_text, tool_results }))
|
|
}
|
|
|
|
/// Call the LLM provider with automatic retry.
|
|
async fn call_with_retry(
|
|
&self,
|
|
context: ConversationContext,
|
|
) -> AgentResult<ProviderResponse> {
|
|
let mut last_error = None;
|
|
|
|
for attempt in 1..=self.retry_cfg.max_attempts {
|
|
match self.provider.chat(context.clone()).await {
|
|
Ok(response) => return Ok(response),
|
|
Err(e) => {
|
|
let error_text = e.to_string();
|
|
let kind = AgentErrorKind::classify(&error_text);
|
|
|
|
if !kind.is_retryable() {
|
|
return Err(AgentError::Provider(format!(
|
|
"Non-retryable provider error: {error_text}"
|
|
)));
|
|
}
|
|
|
|
if attempt < self.retry_cfg.max_attempts {
|
|
let delay = calculate_backoff(
|
|
attempt,
|
|
self.retry_cfg.base_delay_ms,
|
|
self.retry_cfg.max_delay_ms,
|
|
);
|
|
warn!(
|
|
"Provider call failed (attempt {}/{}, kind={:?}), retrying in {}ms: {}",
|
|
attempt,
|
|
self.retry_cfg.max_attempts,
|
|
kind,
|
|
delay,
|
|
error_text
|
|
);
|
|
tokio::time::sleep(Duration::from_millis(delay)).await;
|
|
last_error = Some(e);
|
|
} else {
|
|
error!(
|
|
"Provider call failed after {} attempts: {}",
|
|
self.retry_cfg.max_attempts,
|
|
error_text
|
|
);
|
|
return Err(AgentError::Provider(format!(
|
|
"Failed after {} retries: {error_text}",
|
|
self.retry_cfg.max_attempts - 1
|
|
)));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Err(AgentError::Provider(format!(
|
|
"Unexpected retry exit: {:?}",
|
|
last_error
|
|
)))
|
|
}
|
|
|
|
/// Display output to stdout with GyanDoot prefix.
|
|
fn display_output(&self, output: &str) {
|
|
println!(
|
|
"\n{bold}{green}G{reset}> {output}\n",
|
|
bold = cli_color::BOLD,
|
|
green = cli_color::GREEN,
|
|
reset = cli_color::RESET,
|
|
output = output,
|
|
);
|
|
}
|
|
|
|
/// Start the thinking animation (spinner on stderr).
|
|
/// Returns a handle to stop the animation.
|
|
fn start_thinking_indicator(&self) -> tokio::sync::watch::Sender<bool> {
|
|
use std::io::Write;
|
|
let (tx, mut rx) = tokio::sync::watch::channel(true);
|
|
let frames = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];
|
|
|
|
tokio::spawn(async move {
|
|
let mut frame_idx = 0;
|
|
let start = std::time::Instant::now();
|
|
while *rx.borrow_and_update() {
|
|
let spinner = frames[frame_idx % frames.len()];
|
|
frame_idx += 1;
|
|
let elapsed = start.elapsed().as_secs_f64();
|
|
eprint!(
|
|
"\r{magenta}{spinner}{reset} {bold}answering{reset} ({elapsed:.1}s) ",
|
|
magenta = cli_color::BRIGHT_MAGENTA,
|
|
spinner = spinner,
|
|
reset = cli_color::RESET,
|
|
bold = cli_color::BOLD,
|
|
elapsed = elapsed,
|
|
);
|
|
std::io::stderr().flush().ok();
|
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
|
}
|
|
// Clear the spinner line
|
|
eprint!("\r\x1b[K");
|
|
std::io::stderr().flush().ok();
|
|
});
|
|
|
|
tx
|
|
}
|
|
|
|
/// Stop the thinking animation.
|
|
async fn stop_thinking_indicator(&self, handle: tokio::sync::watch::Sender<bool>) {
|
|
let _ = handle.send(false);
|
|
// Small delay for the spinner task to finish and clear the line
|
|
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
|
}
|
|
|
|
/// Read a line of input from stdin.
|
|
async fn read_input(&self) -> anyhow::Result<String> {
|
|
use std::io::Write;
|
|
use tokio::io::AsyncBufReadExt;
|
|
|
|
// Show input prompt with leading blank line for spacing
|
|
if atty::is(atty::Stream::Stdin) {
|
|
// Small extra delay ensures spinner \r is fully cleared
|
|
tokio::time::sleep(std::time::Duration::from_millis(30)).await;
|
|
print!(
|
|
"\n{bold}{cyan}»{reset} ",
|
|
bold = cli_color::BOLD,
|
|
cyan = cli_color::BRIGHT_CYAN,
|
|
reset = cli_color::RESET,
|
|
);
|
|
std::io::stdout().flush().ok();
|
|
}
|
|
|
|
let mut line = String::new();
|
|
let mut reader = tokio::io::BufReader::new(tokio::io::stdin());
|
|
reader.read_line(&mut line).await?;
|
|
Ok(line.trim().to_string())
|
|
}
|
|
|
|
/// Check if the user wants to exit.
|
|
fn should_exit(&self, input: &str) -> bool {
|
|
matches!(input.to_lowercase().as_str(), "exit" | "quit" | "bye" | "/quit")
|
|
}
|
|
|
|
/// Handle a slash command. Returns true if the user wants to exit.
|
|
async fn handle_command(&self, cmd: &str) -> bool {
|
|
match cmd.to_lowercase().as_str() {
|
|
"/quit" | "/exit" => {
|
|
info!("User requested exit via slash command");
|
|
true
|
|
}
|
|
"/help" => {
|
|
println!(
|
|
"\n{bold}{cyan}╔════════════════════════════════════╗\
|
|
\n║ GyanDoot — Help ║\
|
|
\n╚════════════════════════════════════╝{reset}\n\
|
|
\n{bold}Commands:{reset}\
|
|
\n {green}»{reset} Type anything to chat with the AI\
|
|
\n {green}/quit{reset} Exit GyanDoot\
|
|
\n {green}/help{reset} Show this help message\n\
|
|
\n{bold}Tips:{reset}\
|
|
\n • Use {cyan}web_search{reset} for online queries\
|
|
\n • Use {cyan}web_browse{reset} to visit websites\
|
|
\n • Type {yellow}exit{reset}, {yellow}quit{reset}, or {yellow}bye{reset} to exit\n",
|
|
bold = cli_color::BOLD,
|
|
cyan = cli_color::BRIGHT_CYAN,
|
|
green = cli_color::GREEN,
|
|
yellow = cli_color::YELLOW,
|
|
reset = cli_color::RESET,
|
|
);
|
|
false
|
|
}
|
|
_ => {
|
|
println!(
|
|
"\n{yellow}Unknown command:{reset} {cmd}\n\
|
|
Type {green}/help{reset} for available commands.\n",
|
|
yellow = cli_color::YELLOW,
|
|
reset = cli_color::RESET,
|
|
cmd = cmd,
|
|
green = cli_color::GREEN,
|
|
);
|
|
false
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Calculate exponential backoff with jitter.
|
|
fn calculate_backoff(attempt: u32, base_ms: u64, max_ms: u64) -> u64 {
|
|
let exponential = base_ms * 2u64.pow(attempt.saturating_sub(1));
|
|
let clamped = exponential.min(max_ms);
|
|
let jitter = rand::random::<f64>() * 0.2 - 0.1;
|
|
(clamped as f64 * (1.0 + jitter)) as u64
|
|
}
|
|
|
|
/// Rough token estimator (~4 chars per token).
|
|
fn estimate_tokens(text: &str) -> usize {
|
|
(text.len() / 4).max(1)
|
|
}
|
|
|
|
// ─── Provider factory ──────────────────────────────────────────────────
|
|
|
|
fn create_provider(config: &AgentConfig) -> Box<dyn Provider + Send + Sync> {
|
|
match config.provider.default.as_str() {
|
|
"openai" => {
|
|
let cfg = config.provider.openai.clone().unwrap_or_default();
|
|
Box::new(gyandoot_providers::openai::OpenAIProvider::new(
|
|
cfg.api_key,
|
|
cfg.model,
|
|
cfg.base_url,
|
|
))
|
|
}
|
|
"anthropic" => {
|
|
let cfg = config.provider.anthropic.clone().unwrap_or_default();
|
|
Box::new(gyandoot_providers::anthropic::AnthropicProvider::new(
|
|
cfg.api_key,
|
|
cfg.model,
|
|
))
|
|
}
|
|
"ollama" => {
|
|
let cfg = config.provider.ollama.clone().unwrap_or_default();
|
|
Box::new(gyandoot_providers::ollama::OllamaProvider::new(
|
|
cfg.base_url,
|
|
cfg.model,
|
|
cfg.api_key.clone(),
|
|
))
|
|
}
|
|
other => {
|
|
tracing::warn!("Unknown provider '{other}', falling back to OpenAI");
|
|
Box::new(gyandoot_providers::openai::OpenAIProvider::new(
|
|
String::new(),
|
|
config.model.clone(),
|
|
None,
|
|
))
|
|
}
|
|
}
|
|
}
|
|
|
|
// ─── Tests ──────────────────────────────────────────────────────────────
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use gyandoot_core::config::IdentitySection;
|
|
use gyandoot_core::{
|
|
config::AgentConfig, config::BrowserSection, config::KanbanRootConfig, config::MemoryRootConfig,
|
|
config::PlatformRootConfig, config::ProviderRootConfig, config::SecurityRootConfig,
|
|
config::SkillsRootConfig, Message, MessageRole, Role, ToolCall, ToolDefinition,
|
|
};
|
|
|
|
fn test_config() -> AgentConfig {
|
|
AgentConfig {
|
|
name: "test-agent".to_string(),
|
|
model: "gpt-4o".to_string(),
|
|
max_tokens: 4096,
|
|
temperature: 0.7,
|
|
system_prompt: "You are a test agent.".to_string(),
|
|
provider: ProviderRootConfig::default(),
|
|
platform: PlatformRootConfig::default(),
|
|
memory: MemoryRootConfig::default(),
|
|
security: SecurityRootConfig::default(),
|
|
kanban: KanbanRootConfig::default(),
|
|
skills: SkillsRootConfig::default(),
|
|
identity: IdentitySection::default(),
|
|
browser: BrowserSection::default(),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_retry_config_defaults() {
|
|
let cfg = RetryConfig::default();
|
|
assert_eq!(cfg.max_attempts, 3);
|
|
assert!(cfg.base_delay_ms > 0);
|
|
assert!(cfg.max_delay_ms > 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_estimate_tokens() {
|
|
assert_eq!(estimate_tokens("abcd"), 1);
|
|
assert_eq!(estimate_tokens(""), 1);
|
|
let long = "a".repeat(400);
|
|
assert_eq!(estimate_tokens(&long), 100);
|
|
}
|
|
|
|
#[test]
|
|
fn test_calculate_backoff() {
|
|
let delay = calculate_backoff(1, 1000, 10000);
|
|
assert!(delay >= 900 && delay <= 1100);
|
|
|
|
let delay3 = calculate_backoff(3, 1000, 10000);
|
|
assert!(delay3 <= 10000);
|
|
}
|
|
|
|
#[test]
|
|
fn test_error_classification() {
|
|
assert_eq!(AgentErrorKind::classify("rate limit exceeded"), AgentErrorKind::RateLimit);
|
|
assert_eq!(AgentErrorKind::classify("401 Unauthorized"), AgentErrorKind::Auth);
|
|
assert_eq!(AgentErrorKind::classify("timeout after 30s"), AgentErrorKind::Timeout);
|
|
assert_eq!(AgentErrorKind::classify("500 Internal Server Error"), AgentErrorKind::ServerError);
|
|
assert_eq!(AgentErrorKind::classify("unknown error"), AgentErrorKind::Other);
|
|
}
|
|
|
|
#[test]
|
|
fn test_retryable() {
|
|
assert!(AgentErrorKind::RateLimit.is_retryable());
|
|
assert!(AgentErrorKind::Timeout.is_retryable());
|
|
assert!(AgentErrorKind::ServerError.is_retryable());
|
|
assert!(!AgentErrorKind::Auth.is_retryable());
|
|
assert!(!AgentErrorKind::ContextOverflow.is_retryable());
|
|
}
|
|
} |