diff --git a/src/openhuman/agent/loop_/context_guard.rs b/src/openhuman/agent/loop_/context_guard.rs index c89569cc3..988cbfa30 100644 --- a/src/openhuman/agent/loop_/context_guard.rs +++ b/src/openhuman/agent/loop_/context_guard.rs @@ -7,7 +7,7 @@ use crate::openhuman::providers::UsageInfo; /// Threshold (0.0–1.0) at which auto-compaction is triggered. -const COMPACTION_TRIGGER_THRESHOLD: f64 = 0.90; +pub(crate) const COMPACTION_TRIGGER_THRESHOLD: f64 = 0.90; /// Threshold above which, if compaction is disabled, the guard returns an error. const HARD_LIMIT_THRESHOLD: f64 = 0.95; diff --git a/src/openhuman/agent/loop_/instructions.rs b/src/openhuman/agent/loop_/instructions.rs index 47df6c3a8..ea2f4d434 100644 --- a/src/openhuman/agent/loop_/instructions.rs +++ b/src/openhuman/agent/loop_/instructions.rs @@ -8,14 +8,35 @@ pub(crate) fn build_tool_instructions(tools_registry: &[Box]) -> Strin instructions.push_str("\n## Tool Use Protocol\n\n"); instructions.push_str("To use a tool, wrap a JSON object in tags:\n\n"); instructions.push_str("```\n\n{\"name\": \"tool_name\", \"arguments\": {\"param\": \"value\"}}\n\n```\n\n"); + + // Explicit anti-hallucination rules + instructions.push_str("### Rules (MUST follow)\n\n"); instructions.push_str( - "CRITICAL: Output actual tags—never describe steps or give examples.\n\n", + "1. **ALWAYS use tags** when a task requires a tool. \ + NEVER narrate what you would do — emit the actual tags.\n", ); - instructions.push_str("Example: User says \"what's the date?\". You MUST respond with:\n\n{\"name\":\"shell\",\"arguments\":{\"command\":\"date\"}}\n\n\n"); - instructions.push_str("You may use multiple tool calls in a single response. "); - instructions.push_str("After tool execution, results appear in tags. "); - instructions - .push_str("Continue reasoning with the results until you can give a final answer.\n\n"); + instructions.push_str( + "2. **NEVER describe a tool call in prose** (e.g. \"I'll run ls\" or \ + \"Let me check the file\") without also emitting the tags. \ + If you mention a tool, you must call it.\n", + ); + instructions.push_str( + "3. **Use the exact tool names** listed below. \ + Do not invent tool names that are not in the list.\n", + ); + instructions.push_str("4. You may use **multiple tool calls** in a single response.\n"); + instructions.push_str( + "5. After tool execution, results appear in tags. \ + Continue reasoning with the results until you can give a final answer.\n", + ); + instructions.push_str( + "6. Only respond **without** a tool call when the answer requires \ + no tool (e.g. general knowledge, math, or conversation).\n\n", + ); + + instructions.push_str("### Example\n\n"); + instructions.push_str("User: \"what's the date?\"\nCorrect response:\n\n{\"name\":\"shell\",\"arguments\":{\"command\":\"date\"}}\n\n\n"); + instructions.push_str("### Available Tools\n\n"); for tool in tools_registry { diff --git a/src/openhuman/agent/loop_/parse.rs b/src/openhuman/agent/loop_/parse.rs index 3b3625c7a..26183147a 100644 --- a/src/openhuman/agent/loop_/parse.rs +++ b/src/openhuman/agent/loop_/parse.rs @@ -82,7 +82,51 @@ pub(crate) fn parse_tool_calls_from_json_value(value: &serde_json::Value) -> Vec calls } -const TOOL_CALL_OPEN_TAGS: [&str; 4] = ["", "", "", ""]; +/// Parse bracket-style pseudo-syntax emitted by some models: +/// `{tool => "shell", args => { --command "ls" }}` +/// +/// Extracts tool name and converts `--key "value"` args into a JSON object. +fn parse_bracket_tool_call(inner: &str) -> Option { + static BRACKET_TOOL_RE: LazyLock = LazyLock::new(|| { + Regex::new(r#"(?s)\{?\s*tool\s*=>\s*"([^"]+)"\s*,\s*args\s*=>\s*\{(.*?)\}\s*\}?"#).unwrap() + }); + static ARG_RE: LazyLock = LazyLock::new(|| Regex::new(r#"--(\w+)\s+"([^"]*)"#).unwrap()); + + let cap = BRACKET_TOOL_RE.captures(inner.trim())?; + let name = cap.get(1)?.as_str().to_string(); + let args_body = cap.get(2)?.as_str(); + + let mut args = serde_json::Map::new(); + for arg_cap in ARG_RE.captures_iter(args_body) { + let key = arg_cap.get(1)?.as_str().to_string(); + let value = arg_cap.get(2)?.as_str().to_string(); + args.insert(key, serde_json::Value::String(value)); + } + + if name.is_empty() { + return None; + } + + tracing::debug!( + tool = name.as_str(), + parse_mode = "bracket_syntax", + "[parse] parsed bracket-style tool call" + ); + + Some(ParsedToolCall { + name, + arguments: serde_json::Value::Object(args), + }) +} + +const TOOL_CALL_OPEN_TAGS: [&str; 6] = [ + "", + "", + "", + "", + "[TOOL_CALL]", + "[tool_call]", +]; pub(crate) fn find_first_tag<'a>(haystack: &str, tags: &'a [&'a str]) -> Option<(usize, &'a str)> { tags.iter() @@ -96,6 +140,8 @@ pub(crate) fn matching_tool_call_close_tag(open_tag: &str) -> Option<&'static st "" => Some(""), "" => Some(""), "" => Some(""), + "[TOOL_CALL]" => Some("[/TOOL_CALL]"), + "[tool_call]" => Some("[/tool_call]"), _ => None, } } @@ -376,7 +422,16 @@ pub(crate) fn parse_tool_calls(response: &str) -> (String, Vec) } if !parsed_any { - tracing::warn!("Malformed JSON: expected tool-call object in tag body"); + // Try parsing bracket pseudo-syntax: + // {tool => "name", args => { --key "value" }} + if let Some(call) = parse_bracket_tool_call(inner) { + calls.push(call); + parsed_any = true; + } + } + + if !parsed_any { + tracing::warn!("Malformed tool_call body: could not parse tag content as JSON or bracket syntax"); } remaining = &after_open[close_idx + close_tag.len()..]; diff --git a/src/openhuman/agent/loop_/tool_loop.rs b/src/openhuman/agent/loop_/tool_loop.rs index e0461d220..e71e6b7f8 100644 --- a/src/openhuman/agent/loop_/tool_loop.rs +++ b/src/openhuman/agent/loop_/tool_loop.rs @@ -6,6 +6,7 @@ use anyhow::Result; use std::fmt::Write as _; use std::io::Write as _; +use super::context_guard::{ContextCheckResult, ContextGuard}; use super::credentials::scrub_credentials; use super::parse::{ build_native_assistant_history, find_tool, parse_structured_tool_calls, parse_tool_calls, @@ -77,7 +78,35 @@ pub(crate) async fn run_tool_call_loop( tools_registry.iter().map(|tool| tool.spec()).collect(); let use_native_tools = provider.supports_native_tools() && !tool_specs.is_empty(); - for _iteration in 0..max_iterations { + let mut context_guard = ContextGuard::new(); + + for iteration in 0..max_iterations { + // ── Context guard: check utilization before each LLM call ── + match context_guard.check() { + ContextCheckResult::Ok => {} + ContextCheckResult::CompactionNeeded => { + tracing::warn!( + iteration, + "[agent_loop] context guard: compaction needed (>{:.0}% full)", + super::context_guard::COMPACTION_TRIGGER_THRESHOLD * 100.0 + ); + // Compaction is handled by history management upstream; + // log and continue so the caller can act on it. + } + ContextCheckResult::ContextExhausted { + utilization_pct, + reason, + } => { + tracing::error!( + iteration, + utilization_pct, + "[agent_loop] context exhausted, aborting: {reason}" + ); + anyhow::bail!("Context window exhausted ({utilization_pct}% full): {reason}"); + } + } + + tracing::debug!(iteration, "[agent_loop] sending LLM request"); let image_marker_count = multimodal::count_image_markers(history); if image_marker_count > 0 && !provider.supports_vision() { return Err(ProviderCapabilityError { @@ -115,6 +144,23 @@ pub(crate) async fn run_tool_call_loop( .await { Ok(resp) => { + // Update context guard with token usage from this response. + if let Some(ref usage) = resp.usage { + context_guard.update_usage(usage); + tracing::debug!( + iteration, + input_tokens = usage.input_tokens, + output_tokens = usage.output_tokens, + context_window = usage.context_window, + "[agent_loop] LLM response received" + ); + } else { + tracing::debug!( + iteration, + "[agent_loop] LLM response received (no usage info)" + ); + } + let response_text = resp.text_or_empty().to_string(); let mut calls = parse_structured_tool_calls(&resp.tool_calls); let mut parsed_text = String::new(); @@ -127,6 +173,13 @@ pub(crate) async fn run_tool_call_loop( calls = fallback_calls; } + tracing::debug!( + iteration, + native_tool_calls = resp.tool_calls.len(), + parsed_tool_calls = calls.len(), + "[agent_loop] tool calls parsed" + ); + // Preserve native tool call IDs in assistant history so role=tool // follow-up messages can reference the exact call id. let assistant_history_content = if resp.tool_calls.is_empty() { @@ -156,6 +209,10 @@ pub(crate) async fn run_tool_call_loop( }; if tool_calls.is_empty() { + tracing::debug!( + iteration, + "[agent_loop] no tool calls — returning final response" + ); // No tool calls — this is the final response. // If a streaming sender is provided, relay the text in small chunks // so the channel can progressively update the draft message. @@ -221,20 +278,62 @@ pub(crate) async fn run_tool_call_loop( } } + tracing::debug!( + iteration, + tool = call.name.as_str(), + "[agent_loop] executing tool" + ); + let result = if let Some(tool) = find_tool(tools_registry, &call.name) { - match tool.execute(call.arguments.clone()).await { - Ok(r) => { + // Execute with a 120-second timeout to prevent hangs. + match tokio::time::timeout( + std::time::Duration::from_secs(120), + tool.execute(call.arguments.clone()), + ) + .await + { + Ok(Ok(r)) => { if r.success { + tracing::debug!( + iteration, + tool = call.name.as_str(), + output_len = r.output.len(), + "[agent_loop] tool succeeded" + ); scrub_credentials(&r.output) } else { - format!("Error: {}", r.error.unwrap_or(r.output)) + let err_msg = r.error.unwrap_or(r.output); + tracing::warn!( + iteration, + tool = call.name.as_str(), + "[agent_loop] tool returned error: {err_msg}" + ); + format!("Error: {err_msg}") } } - Err(e) => { + Ok(Err(e)) => { + tracing::error!( + iteration, + tool = call.name.as_str(), + "[agent_loop] tool execution failed: {e}" + ); format!("Error executing {}: {e}", call.name) } + Err(_) => { + tracing::error!( + iteration, + tool = call.name.as_str(), + "[agent_loop] tool execution timed out after 120s" + ); + format!("Error: tool '{}' timed out after 120 seconds", call.name) + } } } else { + tracing::warn!( + iteration, + tool = call.name.as_str(), + "[agent_loop] unknown tool requested" + ); format!("Unknown tool: {}", call.name) }; diff --git a/src/openhuman/providers/reliable.rs b/src/openhuman/providers/reliable.rs index 8e73d79bd..505bc34e9 100644 --- a/src/openhuman/providers/reliable.rs +++ b/src/openhuman/providers/reliable.rs @@ -1,4 +1,6 @@ -use super::traits::{ChatMessage, ChatResponse, StreamChunk, StreamOptions, StreamResult}; +use super::traits::{ + ChatMessage, ChatRequest, ChatResponse, StreamChunk, StreamOptions, StreamResult, +}; use super::Provider; use async_trait::async_trait; use futures_util::{stream, StreamExt}; @@ -517,6 +519,116 @@ impl Provider for ReliableProvider { .any(|(_, provider)| provider.supports_vision()) } + async fn chat( + &self, + request: ChatRequest<'_>, + model: &str, + temperature: f64, + ) -> anyhow::Result { + let models = self.model_chain(model); + let mut failures = Vec::new(); + + for current_model in &models { + for (provider_name, provider) in &self.providers { + let mut backoff_ms = self.base_backoff_ms; + + for attempt in 0..=self.max_retries { + let req = ChatRequest { + messages: request.messages, + tools: request.tools, + system_prompt_cache_boundary: request.system_prompt_cache_boundary, + }; + match provider.chat(req, current_model, temperature).await { + Ok(resp) => { + if attempt > 0 || *current_model != model { + tracing::info!( + provider = provider_name, + model = *current_model, + attempt, + original_model = model, + "Provider recovered (failover/retry)" + ); + } + return Ok(resp); + } + Err(e) => { + let non_retryable_rate_limit = is_non_retryable_rate_limit(&e); + let non_retryable = is_non_retryable(&e) || non_retryable_rate_limit; + let rate_limited = is_rate_limited(&e); + let failure_reason = failure_reason(rate_limited, non_retryable); + let error_detail = compact_error_detail(&e); + + push_failure( + &mut failures, + provider_name, + current_model, + attempt + 1, + self.max_retries + 1, + failure_reason, + &error_detail, + ); + + if rate_limited && !non_retryable_rate_limit { + if let Some(new_key) = self.rotate_key() { + tracing::info!( + provider = provider_name, + error = %error_detail, + "Rate limited, rotated API key (key ending ...{})", + &new_key[new_key.len().saturating_sub(4)..] + ); + } + } + + if non_retryable { + tracing::warn!( + provider = provider_name, + model = *current_model, + error = %error_detail, + "Non-retryable error, moving on" + ); + + if is_context_window_exceeded(&e) { + anyhow::bail!( + "Request exceeds model context window; retries and fallbacks were skipped. Attempts:\n{}", + failures.join("\n") + ); + } + + break; + } + + if attempt < self.max_retries { + let wait = self.compute_backoff(backoff_ms, &e); + tracing::warn!( + provider = provider_name, + model = *current_model, + attempt = attempt + 1, + backoff_ms = wait, + reason = failure_reason, + error = %error_detail, + "Provider call failed, retrying" + ); + tokio::time::sleep(Duration::from_millis(wait)).await; + backoff_ms = (backoff_ms.saturating_mul(2)).min(10_000); + } + } + } + } + + tracing::warn!( + provider = provider_name, + model = *current_model, + "Exhausted retries, trying next provider/model" + ); + } + } + + anyhow::bail!( + "All providers/models failed. Attempts:\n{}", + failures.join("\n") + ) + } + async fn chat_with_tools( &self, messages: &[ChatMessage],