admin/openfang
1
0
Fork 0
mirror of https://github.com/RightNow-AI/openfang.git synced 2026-07-30 23:05:08 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
ad10aa5e805772d5e66439fd96aca9beecdc7ccd
openfang/crates/openfang-runtime/src/agent_loop.rs
T
jaberjaber23 ad10aa5e80 community fixes 
Fix 12 GitHub issues: SSE streaming token counts (#stream_options), UTF-8 boundary panics (#472), cron timezone scheduling (#473), TOML multiline system_prompt (#463), dashboard 401 auth interceptor (#468), custom provider env var convention (#471), cron stale agent_id reassignment (#461), concurrent provider probing with cache (#474), model switch provider sync (#466/#387), OpenRouter real models (#385), embedding URL normalization (#395), ZHIPU content format (#384), Fish shell PATH detection (#372). 1915 tests pass, 0 clippy warnings.
2026-03-09 21:19:50 +03:00

3194 lines
124 KiB
Rust
Raw Blame History

//! Core agent execution loop.
//!
//! The agent loop handles receiving a user message, recalling relevant memories,
//! calling the LLM, executing tool calls, and saving the conversation.
use crate::auth_cooldown::{CooldownVerdict, ProviderCooldown};
use crate::context_budget::{apply_context_guard, truncate_tool_result_dynamic, ContextBudget};
use crate::context_overflow::{recover_from_overflow, RecoveryStage};
use crate::embedding::EmbeddingDriver;
use crate::kernel_handle::KernelHandle;
use crate::llm_driver::{CompletionRequest, LlmDriver, LlmError, StreamEvent};
use crate::llm_errors;
use crate::loop_guard::{LoopGuard, LoopGuardConfig, LoopGuardVerdict};
use crate::mcp::McpConnection;
use crate::tool_runner;
use crate::web_search::WebToolsContext;
use openfang_memory::session::Session;
use openfang_memory::MemorySubstrate;
use openfang_skills::registry::SkillRegistry;
use openfang_types::agent::AgentManifest;
use openfang_types::error::{OpenFangError, OpenFangResult};
use openfang_types::memory::{Memory, MemoryFilter, MemorySource};
use openfang_types::message::{
ContentBlock, Message, MessageContent, Role, StopReason, TokenUsage,
};
use openfang_types::tool::{ToolCall, ToolDefinition};
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tracing::{debug, info, warn};
/// Maximum iterations in the agent loop before giving up.
const MAX_ITERATIONS: u32 = 50;
/// Maximum retries for rate-limited or overloaded API calls.
const MAX_RETRIES: u32 = 3;
/// Base delay for exponential backoff (milliseconds).
const BASE_RETRY_DELAY_MS: u64 = 1000;
/// Timeout for individual tool executions (seconds).
/// Raised from 60s to 120s for browser automation and long-running builds.
const TOOL_TIMEOUT_SECS: u64 = 120;
/// Maximum consecutive MaxTokens continuations before returning partial response.
/// Raised from 3 to 5 to allow longer-form generation.
const MAX_CONTINUATIONS: u32 = 5;
/// Maximum message history size before auto-trimming to prevent context overflow.
const MAX_HISTORY_MESSAGES: usize = 20;
/// Strip a provider prefix from a model ID before sending to the API.
///
/// Many models are stored as `provider/org/model` (e.g. `openrouter/google/gemini-2.5-flash`)
/// but the upstream API expects just `org/model` (e.g. `google/gemini-2.5-flash`).
pub fn strip_provider_prefix(model: &str, provider: &str) -> String {
let slash_prefix = format!("{}/", provider);
let colon_prefix = format!("{}:", provider);
if model.starts_with(&slash_prefix) {
model[slash_prefix.len()..].to_string()
} else if model.starts_with(&colon_prefix) {
model[colon_prefix.len()..].to_string()
} else {
model.to_string()
}
}
/// Default context window size (tokens) for token-based trimming.
const DEFAULT_CONTEXT_WINDOW: usize = 200_000;
/// Agent lifecycle phase within the execution loop.
/// Used for UX indicators (typing, reactions) without coupling to channel types.
#[derive(Debug, Clone, PartialEq)]
pub enum LoopPhase {
/// Agent is calling the LLM.
Thinking,
/// Agent is executing a tool.
ToolUse { tool_name: String },
/// Agent is streaming tokens.
Streaming,
/// Agent finished successfully.
Done,
/// Agent encountered an error.
Error,
}
/// Callback for agent lifecycle phase changes.
/// Implementations should be non-blocking (fire-and-forget) to avoid slowing the loop.
pub type PhaseCallback = Arc<dyn Fn(LoopPhase) + Send + Sync>;
/// Result of an agent loop execution.
#[derive(Debug)]
pub struct AgentLoopResult {
/// The final text response from the agent.
pub response: String,
/// Total token usage across all LLM calls.
pub total_usage: TokenUsage,
/// Number of iterations the loop ran.
pub iterations: u32,
/// Estimated cost in USD (populated by the kernel after the loop returns).
pub cost_usd: Option<f64>,
/// True when the agent intentionally chose not to reply (NO_REPLY token or [[silent]]).
pub silent: bool,
/// Reply directives extracted from the agent's response.
pub directives: openfang_types::message::ReplyDirectives,
}
/// Run the agent execution loop for a single user message.
///
/// This is the core of OpenFang: it loads session context, recalls memories,
/// runs the LLM in a tool-use loop, and saves the updated session.
#[allow(clippy::too_many_arguments)]
pub async fn run_agent_loop(
manifest: &AgentManifest,
user_message: &str,
session: &mut Session,
memory: &MemorySubstrate,
driver: Arc<dyn LlmDriver>,
available_tools: &[ToolDefinition],
kernel: Option<Arc<dyn KernelHandle>>,
skill_registry: Option<&SkillRegistry>,
mcp_connections: Option<&tokio::sync::Mutex<Vec<McpConnection>>>,
web_ctx: Option<&WebToolsContext>,
browser_ctx: Option<&crate::browser::BrowserManager>,
embedding_driver: Option<&(dyn EmbeddingDriver + Send + Sync)>,
workspace_root: Option<&Path>,
on_phase: Option<&PhaseCallback>,
media_engine: Option<&crate::media_understanding::MediaEngine>,
tts_engine: Option<&crate::tts::TtsEngine>,
docker_config: Option<&openfang_types::config::DockerSandboxConfig>,
hooks: Option<&crate::hooks::HookRegistry>,
context_window_tokens: Option<usize>,
process_manager: Option<&crate::process_manager::ProcessManager>,
) -> OpenFangResult<AgentLoopResult> {
info!(agent = %manifest.name, "Starting agent loop");
// Extract hand-allowed env vars from manifest metadata (set by kernel for hand settings)
let hand_allowed_env: Vec<String> = manifest
.metadata
.get("hand_allowed_env")
.and_then(|v| serde_json::from_value(v.clone()).ok())
.unwrap_or_default();
// Recall relevant memories — prefer vector similarity search when embedding driver is available
let memories = if let Some(emb) = embedding_driver {
match emb.embed_one(user_message).await {
Ok(query_vec) => {
debug!("Using vector recall (dims={})", query_vec.len());
memory
.recall_with_embedding_async(
user_message,
5,
Some(MemoryFilter {
agent_id: Some(session.agent_id),
..Default::default()
}),
Some(&query_vec),
)
.await
.unwrap_or_default()
}
Err(e) => {
warn!("Embedding recall failed, falling back to text search: {e}");
memory
.recall(
user_message,
5,
Some(MemoryFilter {
agent_id: Some(session.agent_id),
..Default::default()
}),
)
.await
.unwrap_or_default()
}
}
} else {
memory
.recall(
user_message,
5,
Some(MemoryFilter {
agent_id: Some(session.agent_id),
..Default::default()
}),
)
.await
.unwrap_or_default()
};
// Fire BeforePromptBuild hook
let agent_id_str = session.agent_id.0.to_string();
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::BeforePromptBuild,
data: serde_json::json!({
"system_prompt": &manifest.model.system_prompt,
"user_message": user_message,
}),
};
let _ = hook_reg.fire(&ctx);
}
// Build the system prompt — base prompt comes from kernel (prompt_builder),
// we append recalled memories here since they are resolved at loop time.
let mut system_prompt = manifest.model.system_prompt.clone();
if !memories.is_empty() {
let mem_pairs: Vec<(String, String)> = memories
.iter()
.map(|m| (String::new(), m.content.clone()))
.collect();
system_prompt.push_str("\n\n");
system_prompt.push_str(&crate::prompt_builder::build_memory_section(&mem_pairs));
}
// Add the user message to session history
session.messages.push(Message::user(user_message));
// Build the messages for the LLM, filtering system messages
// System prompt goes into the separate `system` field
let llm_messages: Vec<Message> = session
.messages
.iter()
.filter(|m| m.role != Role::System)
.cloned()
.collect();
// Validate and repair session history (drop orphans, merge consecutive)
let mut messages = crate::session_repair::validate_and_repair(&llm_messages);
// Inject canonical context as the first user message (not in system prompt)
// to keep the system prompt stable across turns for provider prompt caching.
if let Some(cc_msg) = manifest
.metadata
.get("canonical_context_msg")
.and_then(|v| v.as_str())
{
if !cc_msg.is_empty() {
messages.insert(0, Message::user(cc_msg));
}
}
let mut total_usage = TokenUsage::default();
let final_response;
// Safety valve: trim excessively long message histories to prevent context overflow.
// The full compaction system handles sophisticated summarization, but this prevents
// the catastrophic case where 200+ messages cause instant context overflow.
if messages.len() > MAX_HISTORY_MESSAGES {
let trim_count = messages.len() - MAX_HISTORY_MESSAGES;
warn!(
agent = %manifest.name,
total_messages = messages.len(),
trimming = trim_count,
"Trimming old messages to prevent context overflow"
);
messages.drain(..trim_count);
}
// Use autonomous config max_iterations if set, else default
let max_iterations = manifest
.autonomous
.as_ref()
.map(|a| a.max_iterations)
.unwrap_or(MAX_ITERATIONS);
// Initialize loop guard — scale circuit breaker for autonomous agents
let loop_guard_config = {
let mut cfg = LoopGuardConfig::default();
if max_iterations > cfg.global_circuit_breaker {
cfg.global_circuit_breaker = max_iterations * 3;
}
cfg
};
let mut loop_guard = LoopGuard::new(loop_guard_config);
let mut consecutive_max_tokens: u32 = 0;
// Build context budget from model's actual context window (or fallback to default)
let ctx_window = context_window_tokens.unwrap_or(DEFAULT_CONTEXT_WINDOW);
let context_budget = ContextBudget::new(ctx_window);
let mut any_tools_executed = false;
for iteration in 0..max_iterations {
debug!(iteration, "Agent loop iteration");
// Context overflow recovery pipeline (replaces emergency_trim_messages)
let recovery =
recover_from_overflow(&mut messages, &system_prompt, available_tools, ctx_window);
if recovery == RecoveryStage::FinalError {
warn!("Context overflow unrecoverable — suggest /reset or /compact");
}
// Re-validate tool_call/tool_result pairing after overflow drains
// which may have broken assistant→tool ordering invariants.
if recovery != RecoveryStage::None {
messages = crate::session_repair::validate_and_repair(&messages);
}
// Context guard: compact oversized tool results before LLM call
apply_context_guard(&mut messages, &context_budget, available_tools);
// Strip provider prefix: "openrouter/google/gemini-2.5-flash" → "google/gemini-2.5-flash"
let api_model = strip_provider_prefix(&manifest.model.model, &manifest.model.provider);
let request = CompletionRequest {
model: api_model,
messages: messages.clone(),
tools: available_tools.to_vec(),
max_tokens: manifest.model.max_tokens,
temperature: manifest.model.temperature,
system: Some(system_prompt.clone()),
thinking: None,
};
// Notify phase: Thinking
if let Some(cb) = on_phase {
cb(LoopPhase::Thinking);
}
// Call LLM with retry, error classification, and circuit breaker
let provider_name = manifest.model.provider.as_str();
let mut response = call_with_retry(&*driver, request, Some(provider_name), None).await?;
total_usage.input_tokens += response.usage.input_tokens;
total_usage.output_tokens += response.usage.output_tokens;
// Recover tool calls output as text by models that don't use the tool_calls API field
// (e.g. Groq/Llama, DeepSeek emit `<function=name>{json}</function>` in text)
if matches!(
response.stop_reason,
StopReason::EndTurn | StopReason::StopSequence
) && response.tool_calls.is_empty()
{
let recovered = recover_text_tool_calls(&response.text(), available_tools);
if !recovered.is_empty() {
info!(
count = recovered.len(),
"Recovered text-based tool calls → promoting to ToolUse"
);
response.tool_calls = recovered;
response.stop_reason = StopReason::ToolUse;
// Build ToolUse content blocks from recovered calls
let mut new_blocks: Vec<ContentBlock> = Vec::new();
for tc in &response.tool_calls {
new_blocks.push(ContentBlock::ToolUse {
id: tc.id.clone(),
name: tc.name.clone(),
input: tc.input.clone(),
});
}
response.content = new_blocks;
}
}
match response.stop_reason {
StopReason::EndTurn | StopReason::StopSequence => {
// LLM is done — extract text and save
let text = response.text();
// Parse reply directives from the response text
let (cleaned_text, parsed_directives) =
crate::reply_directives::parse_directives(&text);
let text = cleaned_text;
// NO_REPLY: agent intentionally chose not to reply
if text.trim() == "NO_REPLY" || parsed_directives.silent {
debug!(agent = %manifest.name, "Agent chose NO_REPLY/silent — silent completion");
session
.messages
.push(Message::assistant("[no reply needed]".to_string()));
memory
.save_session(session)
.map_err(|e| OpenFangError::Memory(e.to_string()))?;
return Ok(AgentLoopResult {
response: String::new(),
total_usage,
iterations: iteration + 1,
cost_usd: None,
silent: true,
directives: openfang_types::message::ReplyDirectives {
reply_to: parsed_directives.reply_to,
current_thread: parsed_directives.current_thread,
silent: true,
},
});
}
// One-shot retry: if the very first LLM call returns empty text
// with no tool use, try once more before accepting the empty result.
// This catches transient LLM hiccups (overload, empty stream, etc.).
if text.trim().is_empty() && iteration == 0 && response.tool_calls.is_empty() {
warn!(agent = %manifest.name, "Empty response on first call, retrying once");
messages.push(Message::assistant("[no response]".to_string()));
messages.push(Message::user("Please provide your response.".to_string()));
continue;
}
// Guard against empty response — covers both iteration 0 and post-tool cycles
let text = if text.trim().is_empty() {
warn!(
agent = %manifest.name,
iteration,
input_tokens = total_usage.input_tokens,
output_tokens = total_usage.output_tokens,
messages_count = messages.len(),
"Empty response from LLM — guard activated"
);
if any_tools_executed {
"[Task completed — the agent executed tools but did not produce a text summary.]".to_string()
} else {
"[The model returned an empty response. This usually means the model is overloaded, the context is too large, or the API key lacks credits. Try again or check /status.]".to_string()
}
} else {
text
};
final_response = text.clone();
session.messages.push(Message::assistant(text));
// Prune NO_REPLY heartbeat turns to save context budget
crate::session_repair::prune_heartbeat_turns(&mut session.messages, 10);
// Save session
memory
.save_session(session)
.map_err(|e| OpenFangError::Memory(e.to_string()))?;
// Remember this interaction (with embedding if available)
let interaction_text = format!(
"User asked: {}\nI responded: {}",
user_message, final_response
);
if let Some(emb) = embedding_driver {
match emb.embed_one(&interaction_text).await {
Ok(vec) => {
let _ = memory
.remember_with_embedding_async(
session.agent_id,
&interaction_text,
MemorySource::Conversation,
"episodic",
HashMap::new(),
Some(&vec),
)
.await;
}
Err(e) => {
warn!("Embedding for remember failed: {e}");
let _ = memory
.remember(
session.agent_id,
&interaction_text,
MemorySource::Conversation,
"episodic",
HashMap::new(),
)
.await;
}
}
} else {
let _ = memory
.remember(
session.agent_id,
&interaction_text,
MemorySource::Conversation,
"episodic",
HashMap::new(),
)
.await;
}
// Notify phase: Done
if let Some(cb) = on_phase {
cb(LoopPhase::Done);
}
info!(
agent = %manifest.name,
iterations = iteration + 1,
tokens = total_usage.total(),
"Agent loop completed"
);
// Fire AgentLoopEnd hook
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"iterations": iteration + 1,
"response_length": final_response.len(),
}),
};
let _ = hook_reg.fire(&ctx);
}
return Ok(AgentLoopResult {
response: final_response,
total_usage,
iterations: iteration + 1,
cost_usd: None,
silent: false,
directives: Default::default(),
});
}
StopReason::ToolUse => {
// Reset MaxTokens continuation counter on tool use
consecutive_max_tokens = 0;
any_tools_executed = true;
// Execute tool calls
let assistant_blocks = response.content.clone();
// Add assistant message with tool use blocks
session.messages.push(Message {
role: Role::Assistant,
content: MessageContent::Blocks(assistant_blocks.clone()),
});
messages.push(Message {
role: Role::Assistant,
content: MessageContent::Blocks(assistant_blocks),
});
// Build allowed tool names list for capability enforcement
let allowed_tool_names: Vec<String> =
available_tools.iter().map(|t| t.name.clone()).collect();
let caller_id_str = session.agent_id.to_string();
// Execute each tool call with loop guard, timeout, and truncation
let mut tool_result_blocks = Vec::new();
for tool_call in &response.tool_calls {
// Loop guard check
let verdict = loop_guard.check(&tool_call.name, &tool_call.input);
match &verdict {
LoopGuardVerdict::CircuitBreak(msg) => {
warn!(tool = %tool_call.name, "Circuit breaker triggered");
// Save session before bailing
if let Err(e) = memory.save_session(session) {
warn!("Failed to save session on circuit break: {e}");
}
// Fire AgentLoopEnd hook on circuit break
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"reason": "circuit_break",
"error": msg.as_str(),
}),
};
let _ = hook_reg.fire(&ctx);
}
return Err(OpenFangError::Internal(msg.clone()));
}
LoopGuardVerdict::Block(msg) => {
warn!(tool = %tool_call.name, "Tool call blocked by loop guard");
tool_result_blocks.push(ContentBlock::ToolResult {
tool_use_id: tool_call.id.clone(),
tool_name: tool_call.name.clone(),
content: msg.clone(),
is_error: true,
});
continue;
}
_ => {} // Allow or Warn — proceed with execution
}
debug!(tool = %tool_call.name, id = %tool_call.id, "Executing tool");
// Notify phase: ToolUse
if let Some(cb) = on_phase {
let sanitized: String = tool_call
.name
.chars()
.filter(|c| !c.is_control())
.take(64)
.collect();
cb(LoopPhase::ToolUse {
tool_name: sanitized,
});
}
// Fire BeforeToolCall hook (can block execution)
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: &caller_id_str,
event: openfang_types::agent::HookEvent::BeforeToolCall,
data: serde_json::json!({
"tool_name": &tool_call.name,
"input": &tool_call.input,
}),
};
if let Err(reason) = hook_reg.fire(&ctx) {
tool_result_blocks.push(ContentBlock::ToolResult {
tool_use_id: tool_call.id.clone(),
tool_name: tool_call.name.clone(),
content: format!(
"Hook blocked tool '{}': {}",
tool_call.name, reason
),
is_error: true,
});
continue;
}
}
// Resolve effective exec policy (per-agent override or global)
let effective_exec_policy = manifest.exec_policy.as_ref();
// Timeout-wrapped execution
let result = match tokio::time::timeout(
Duration::from_secs(TOOL_TIMEOUT_SECS),
tool_runner::execute_tool(
&tool_call.id,
&tool_call.name,
&tool_call.input,
kernel.as_ref(),
Some(&allowed_tool_names),
Some(&caller_id_str),
skill_registry,
mcp_connections,
web_ctx,
browser_ctx,
if hand_allowed_env.is_empty() {
None
} else {
Some(&hand_allowed_env)
},
workspace_root,
media_engine,
effective_exec_policy,
tts_engine,
docker_config,
process_manager,
),
)
.await
{
Ok(result) => result,
Err(_) => {
warn!(tool = %tool_call.name, "Tool execution timed out after {}s", TOOL_TIMEOUT_SECS);
openfang_types::tool::ToolResult {
tool_use_id: tool_call.id.clone(),
content: format!(
"Tool '{}' timed out after {}s.",
tool_call.name, TOOL_TIMEOUT_SECS
),
is_error: true,
}
}
};
// Fire AfterToolCall hook
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: caller_id_str.as_str(),
event: openfang_types::agent::HookEvent::AfterToolCall,
data: serde_json::json!({
"tool_name": &tool_call.name,
"result": &result.content,
"is_error": result.is_error,
}),
};
let _ = hook_reg.fire(&ctx);
}
// Dynamic truncation based on context budget (replaces flat MAX_TOOL_RESULT_CHARS)
let content = truncate_tool_result_dynamic(&result.content, &context_budget);
// Append warning if verdict was Warn
let final_content = if let LoopGuardVerdict::Warn(ref warn_msg) = verdict {
format!("{content}\n\n[LOOP GUARD] {warn_msg}")
} else {
content
};
tool_result_blocks.push(ContentBlock::ToolResult {
tool_use_id: result.tool_use_id,
tool_name: tool_call.name.clone(),
content: final_content,
is_error: result.is_error,
});
}
// Detect approval denials and inject guidance to prevent infinite retry loops
let denial_count = tool_result_blocks.iter().filter(|b| {
matches!(b, ContentBlock::ToolResult { content, is_error: true, .. }
if content.contains("requires human approval and was denied"))
}).count();
if denial_count > 0 {
tool_result_blocks.push(ContentBlock::Text {
text: format!(
"[System: {} tool call(s) were denied by approval policy. \
Do NOT retry denied tools. Explain to the user what you \
wanted to do and that it requires their approval.]",
denial_count
),
});
}
// Detect tool errors and inject guidance to prevent fabrication
let error_count = tool_result_blocks.iter().filter(|b| {
matches!(b, ContentBlock::ToolResult { is_error: true, .. })
}).count();
let non_denial_errors = error_count.saturating_sub(denial_count);
if non_denial_errors > 0 {
tool_result_blocks.push(ContentBlock::Text {
text: format!(
"[System: {} tool(s) returned errors. Report the error honestly \
to the user. Do NOT fabricate results or pretend the tool succeeded. \
If a search or fetch failed, tell the user it failed and suggest \
alternatives instead of making up data.]",
non_denial_errors
),
});
}
// Add tool results as a user message (Anthropic API requirement)
let tool_results_msg = Message {
role: Role::User,
content: MessageContent::Blocks(tool_result_blocks.clone()),
};
session.messages.push(tool_results_msg.clone());
messages.push(tool_results_msg);
// Interim save after tool execution to prevent data loss on crash
if let Err(e) = memory.save_session(session) {
warn!("Failed to interim-save session: {e}");
}
}
StopReason::MaxTokens => {
consecutive_max_tokens += 1;
if consecutive_max_tokens >= MAX_CONTINUATIONS {
// Return partial response instead of continuing forever
let text = response.text();
let text = if text.trim().is_empty() {
"[Partial response — token limit reached with no text output.]".to_string()
} else {
text
};
session.messages.push(Message::assistant(&text));
if let Err(e) = memory.save_session(session) {
warn!("Failed to save session on max continuations: {e}");
}
warn!(
iteration,
consecutive_max_tokens,
"Max continuations reached, returning partial response"
);
// Fire AgentLoopEnd hook
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"iterations": iteration + 1,
"reason": "max_continuations",
}),
};
let _ = hook_reg.fire(&ctx);
}
return Ok(AgentLoopResult {
response: text,
total_usage,
iterations: iteration + 1,
cost_usd: None,
silent: false,
directives: Default::default(),
});
}
// Model hit token limit — add partial response and continue
let text = response.text();
session.messages.push(Message::assistant(&text));
messages.push(Message::assistant(&text));
session.messages.push(Message::user("Please continue."));
messages.push(Message::user("Please continue."));
warn!(iteration, "Max tokens hit, continuing");
}
}
}
// Save session before failing so conversation history is preserved
if let Err(e) = memory.save_session(session) {
warn!("Failed to save session on max iterations: {e}");
}
// Fire AgentLoopEnd hook on max iterations exceeded
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"reason": "max_iterations_exceeded",
"iterations": max_iterations,
}),
};
let _ = hook_reg.fire(&ctx);
}
Err(OpenFangError::MaxIterationsExceeded(max_iterations))
}
/// Call an LLM driver with automatic retry on rate-limit and overload errors.
///
/// Uses the `llm_errors` classifier for smart error handling and the
/// `ProviderCooldown` circuit breaker to prevent request storms.
async fn call_with_retry(
driver: &dyn LlmDriver,
request: CompletionRequest,
provider: Option<&str>,
cooldown: Option<&ProviderCooldown>,
) -> OpenFangResult<crate::llm_driver::CompletionResponse> {
// Check circuit breaker before calling
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
match cooldown.check(provider) {
CooldownVerdict::Reject {
reason,
retry_after_secs,
} => {
return Err(OpenFangError::LlmDriver(format!(
"Provider '{provider}' is in cooldown ({reason}). Retry in {retry_after_secs}s."
)));
}
CooldownVerdict::AllowProbe => {
debug!(provider, "Allowing probe request through circuit breaker");
}
CooldownVerdict::Allow => {}
}
}
let mut last_error = None;
for attempt in 0..=MAX_RETRIES {
match driver.complete(request.clone()).await {
Ok(response) => {
// Record success with circuit breaker
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_success(provider);
}
return Ok(response);
}
Err(LlmError::RateLimited { retry_after_ms }) => {
if attempt == MAX_RETRIES {
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_failure(provider, false);
}
return Err(OpenFangError::LlmDriver(format!(
"Rate limited after {} retries",
MAX_RETRIES
)));
}
let delay = std::cmp::max(retry_after_ms, BASE_RETRY_DELAY_MS * 2u64.pow(attempt));
warn!(
attempt,
delay_ms = delay,
"Rate limited, retrying after delay"
);
tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
last_error = Some("Rate limited".to_string());
}
Err(LlmError::Overloaded { retry_after_ms }) => {
if attempt == MAX_RETRIES {
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_failure(provider, false);
}
return Err(OpenFangError::LlmDriver(format!(
"Model overloaded after {} retries",
MAX_RETRIES
)));
}
let delay = std::cmp::max(retry_after_ms, BASE_RETRY_DELAY_MS * 2u64.pow(attempt));
warn!(
attempt,
delay_ms = delay,
"Model overloaded, retrying after delay"
);
tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
last_error = Some("Overloaded".to_string());
}
Err(e) => {
// Use classifier for smarter error handling
let raw_error = e.to_string();
let classified = llm_errors::classify_error(&raw_error, None);
warn!(
category = ?classified.category,
retryable = classified.is_retryable,
raw = %raw_error,
"LLM error classified: {}",
classified.sanitized_message
);
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_failure(provider, classified.is_billing);
}
// Include raw error detail so dashboard users can debug
let user_msg = if classified.category == llm_errors::LlmErrorCategory::Format {
format!("{} — raw: {}", classified.sanitized_message, raw_error)
} else {
classified.sanitized_message
};
return Err(OpenFangError::LlmDriver(user_msg));
}
}
}
Err(OpenFangError::LlmDriver(
last_error.unwrap_or_else(|| "Unknown error".to_string()),
))
}
/// Call an LLM driver in streaming mode with automatic retry on rate-limit and overload errors.
///
/// Uses the `llm_errors` classifier and `ProviderCooldown` circuit breaker.
async fn stream_with_retry(
driver: &dyn LlmDriver,
request: CompletionRequest,
tx: mpsc::Sender<StreamEvent>,
provider: Option<&str>,
cooldown: Option<&ProviderCooldown>,
) -> OpenFangResult<crate::llm_driver::CompletionResponse> {
// Check circuit breaker before calling
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
match cooldown.check(provider) {
CooldownVerdict::Reject {
reason,
retry_after_secs,
} => {
return Err(OpenFangError::LlmDriver(format!(
"Provider '{provider}' is in cooldown ({reason}). Retry in {retry_after_secs}s."
)));
}
CooldownVerdict::AllowProbe => {
debug!(
provider,
"Allowing probe request through circuit breaker (stream)"
);
}
CooldownVerdict::Allow => {}
}
}
let mut last_error = None;
for attempt in 0..=MAX_RETRIES {
match driver.stream(request.clone(), tx.clone()).await {
Ok(response) => {
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_success(provider);
}
return Ok(response);
}
Err(LlmError::RateLimited { retry_after_ms }) => {
if attempt == MAX_RETRIES {
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_failure(provider, false);
}
return Err(OpenFangError::LlmDriver(format!(
"Rate limited after {} retries",
MAX_RETRIES
)));
}
let delay = std::cmp::max(retry_after_ms, BASE_RETRY_DELAY_MS * 2u64.pow(attempt));
warn!(
attempt,
delay_ms = delay,
"Rate limited (stream), retrying after delay"
);
tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
last_error = Some("Rate limited".to_string());
}
Err(LlmError::Overloaded { retry_after_ms }) => {
if attempt == MAX_RETRIES {
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_failure(provider, false);
}
return Err(OpenFangError::LlmDriver(format!(
"Model overloaded after {} retries",
MAX_RETRIES
)));
}
let delay = std::cmp::max(retry_after_ms, BASE_RETRY_DELAY_MS * 2u64.pow(attempt));
warn!(
attempt,
delay_ms = delay,
"Model overloaded (stream), retrying after delay"
);
tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
last_error = Some("Overloaded".to_string());
}
Err(e) => {
let raw_error = e.to_string();
let classified = llm_errors::classify_error(&raw_error, None);
warn!(
category = ?classified.category,
retryable = classified.is_retryable,
raw = %raw_error,
"LLM stream error classified: {}",
classified.sanitized_message
);
if let (Some(provider), Some(cooldown)) = (provider, cooldown) {
cooldown.record_failure(provider, classified.is_billing);
}
let user_msg = if classified.category == llm_errors::LlmErrorCategory::Format {
format!("{} — raw: {}", classified.sanitized_message, raw_error)
} else {
classified.sanitized_message
};
return Err(OpenFangError::LlmDriver(user_msg));
}
}
}
Err(OpenFangError::LlmDriver(
last_error.unwrap_or_else(|| "Unknown error".to_string()),
))
}
/// Run the agent execution loop with streaming support.
///
/// Like `run_agent_loop`, but sends `StreamEvent`s to the provided channel
/// as tokens arrive from the LLM. Tool execution happens between LLM calls
/// and is not streamed.
#[allow(clippy::too_many_arguments)]
pub async fn run_agent_loop_streaming(
manifest: &AgentManifest,
user_message: &str,
session: &mut Session,
memory: &MemorySubstrate,
driver: Arc<dyn LlmDriver>,
available_tools: &[ToolDefinition],
kernel: Option<Arc<dyn KernelHandle>>,
stream_tx: mpsc::Sender<StreamEvent>,
skill_registry: Option<&SkillRegistry>,
mcp_connections: Option<&tokio::sync::Mutex<Vec<McpConnection>>>,
web_ctx: Option<&WebToolsContext>,
browser_ctx: Option<&crate::browser::BrowserManager>,
embedding_driver: Option<&(dyn EmbeddingDriver + Send + Sync)>,
workspace_root: Option<&Path>,
on_phase: Option<&PhaseCallback>,
media_engine: Option<&crate::media_understanding::MediaEngine>,
tts_engine: Option<&crate::tts::TtsEngine>,
docker_config: Option<&openfang_types::config::DockerSandboxConfig>,
hooks: Option<&crate::hooks::HookRegistry>,
context_window_tokens: Option<usize>,
process_manager: Option<&crate::process_manager::ProcessManager>,
) -> OpenFangResult<AgentLoopResult> {
info!(agent = %manifest.name, "Starting streaming agent loop");
// Extract hand-allowed env vars from manifest metadata (set by kernel for hand settings)
let hand_allowed_env: Vec<String> = manifest
.metadata
.get("hand_allowed_env")
.and_then(|v| serde_json::from_value(v.clone()).ok())
.unwrap_or_default();
// Recall relevant memories — prefer vector similarity search when embedding driver is available
let memories = if let Some(emb) = embedding_driver {
match emb.embed_one(user_message).await {
Ok(query_vec) => {
debug!("Using vector recall (streaming, dims={})", query_vec.len());
memory
.recall_with_embedding_async(
user_message,
5,
Some(MemoryFilter {
agent_id: Some(session.agent_id),
..Default::default()
}),
Some(&query_vec),
)
.await
.unwrap_or_default()
}
Err(e) => {
warn!("Embedding recall failed (streaming), falling back to text search: {e}");
memory
.recall(
user_message,
5,
Some(MemoryFilter {
agent_id: Some(session.agent_id),
..Default::default()
}),
)
.await
.unwrap_or_default()
}
}
} else {
memory
.recall(
user_message,
5,
Some(MemoryFilter {
agent_id: Some(session.agent_id),
..Default::default()
}),
)
.await
.unwrap_or_default()
};
// Fire BeforePromptBuild hook
let agent_id_str = session.agent_id.0.to_string();
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::BeforePromptBuild,
data: serde_json::json!({
"system_prompt": &manifest.model.system_prompt,
"user_message": user_message,
}),
};
let _ = hook_reg.fire(&ctx);
}
// Build the system prompt — base prompt comes from kernel (prompt_builder),
// we append recalled memories here since they are resolved at loop time.
let mut system_prompt = manifest.model.system_prompt.clone();
if !memories.is_empty() {
let mem_pairs: Vec<(String, String)> = memories
.iter()
.map(|m| (String::new(), m.content.clone()))
.collect();
system_prompt.push_str("\n\n");
system_prompt.push_str(&crate::prompt_builder::build_memory_section(&mem_pairs));
}
// Add the user message to session history
session.messages.push(Message::user(user_message));
let llm_messages: Vec<Message> = session
.messages
.iter()
.filter(|m| m.role != Role::System)
.cloned()
.collect();
// Validate and repair session history (drop orphans, merge consecutive)
let mut messages = crate::session_repair::validate_and_repair(&llm_messages);
// Inject canonical context as the first user message (not in system prompt)
// to keep the system prompt stable across turns for provider prompt caching.
if let Some(cc_msg) = manifest
.metadata
.get("canonical_context_msg")
.and_then(|v| v.as_str())
{
if !cc_msg.is_empty() {
messages.insert(0, Message::user(cc_msg));
}
}
let mut total_usage = TokenUsage::default();
let final_response;
// Safety valve: trim excessively long message histories to prevent context overflow.
if messages.len() > MAX_HISTORY_MESSAGES {
let trim_count = messages.len() - MAX_HISTORY_MESSAGES;
warn!(
agent = %manifest.name,
total_messages = messages.len(),
trimming = trim_count,
"Trimming old messages to prevent context overflow (streaming)"
);
messages.drain(..trim_count);
}
// Use autonomous config max_iterations if set, else default
let max_iterations = manifest
.autonomous
.as_ref()
.map(|a| a.max_iterations)
.unwrap_or(MAX_ITERATIONS);
// Initialize loop guard — scale circuit breaker for autonomous agents
let loop_guard_config = {
let mut cfg = LoopGuardConfig::default();
if max_iterations > cfg.global_circuit_breaker {
cfg.global_circuit_breaker = max_iterations * 3;
}
cfg
};
let mut loop_guard = LoopGuard::new(loop_guard_config);
let mut consecutive_max_tokens: u32 = 0;
// Build context budget from model's actual context window (or fallback to default)
let ctx_window = context_window_tokens.unwrap_or(DEFAULT_CONTEXT_WINDOW);
let context_budget = ContextBudget::new(ctx_window);
let mut any_tools_executed = false;
for iteration in 0..max_iterations {
debug!(iteration, "Streaming agent loop iteration");
// Context overflow recovery pipeline (replaces emergency_trim_messages)
let recovery =
recover_from_overflow(&mut messages, &system_prompt, available_tools, ctx_window);
match &recovery {
RecoveryStage::None => {}
RecoveryStage::FinalError => {
if stream_tx.send(StreamEvent::PhaseChange {
phase: "context_warning".to_string(),
detail: Some("Context overflow unrecoverable. Use /reset or /compact.".to_string()),
}).await.is_err() {
warn!("Stream consumer disconnected while sending context overflow warning");
}
}
_ => {
if stream_tx.send(StreamEvent::PhaseChange {
phase: "context_warning".to_string(),
detail: Some("Older messages trimmed to stay within context limits. Use /compact for smarter summarization.".to_string()),
}).await.is_err() {
warn!("Stream consumer disconnected while sending context trim warning");
}
}
}
// Context guard: compact oversized tool results before LLM call
apply_context_guard(&mut messages, &context_budget, available_tools);
// Strip provider prefix: "openrouter/google/gemini-2.5-flash" → "google/gemini-2.5-flash"
let api_model = strip_provider_prefix(&manifest.model.model, &manifest.model.provider);
let request = CompletionRequest {
model: api_model,
messages: messages.clone(),
tools: available_tools.to_vec(),
max_tokens: manifest.model.max_tokens,
temperature: manifest.model.temperature,
system: Some(system_prompt.clone()),
thinking: None,
};
// Notify phase: Streaming (streaming variant always streams)
if let Some(cb) = on_phase {
cb(LoopPhase::Streaming);
}
// Stream LLM call with retry, error classification, and circuit breaker
let provider_name = manifest.model.provider.as_str();
let mut response = stream_with_retry(
&*driver,
request,
stream_tx.clone(),
Some(provider_name),
None,
)
.await?;
total_usage.input_tokens += response.usage.input_tokens;
total_usage.output_tokens += response.usage.output_tokens;
// Recover tool calls output as text (streaming path)
if matches!(
response.stop_reason,
StopReason::EndTurn | StopReason::StopSequence
) && response.tool_calls.is_empty()
{
let recovered = recover_text_tool_calls(&response.text(), available_tools);
if !recovered.is_empty() {
info!(
count = recovered.len(),
"Recovered text-based tool calls (streaming) → promoting to ToolUse"
);
response.tool_calls = recovered;
response.stop_reason = StopReason::ToolUse;
let mut new_blocks: Vec<ContentBlock> = Vec::new();
for tc in &response.tool_calls {
new_blocks.push(ContentBlock::ToolUse {
id: tc.id.clone(),
name: tc.name.clone(),
input: tc.input.clone(),
});
}
response.content = new_blocks;
}
}
match response.stop_reason {
StopReason::EndTurn | StopReason::StopSequence => {
let text = response.text();
// Parse reply directives from the streaming response text
let (cleaned_text_s, parsed_directives_s) =
crate::reply_directives::parse_directives(&text);
let text = cleaned_text_s;
// NO_REPLY: agent intentionally chose not to reply
if text.trim() == "NO_REPLY" || parsed_directives_s.silent {
debug!(agent = %manifest.name, "Agent chose NO_REPLY/silent (streaming) — silent completion");
session
.messages
.push(Message::assistant("[no reply needed]".to_string()));
memory
.save_session(session)
.map_err(|e| OpenFangError::Memory(e.to_string()))?;
return Ok(AgentLoopResult {
response: String::new(),
total_usage,
iterations: iteration + 1,
cost_usd: None,
silent: true,
directives: openfang_types::message::ReplyDirectives {
reply_to: parsed_directives_s.reply_to,
current_thread: parsed_directives_s.current_thread,
silent: true,
},
});
}
// One-shot retry: if the very first LLM call returns empty text
// with no tool use, try once more before accepting the empty result.
if text.trim().is_empty() && iteration == 0 && response.tool_calls.is_empty() {
warn!(agent = %manifest.name, "Empty response on first call (streaming), retrying once");
messages.push(Message::assistant("[no response]".to_string()));
messages.push(Message::user("Please provide your response.".to_string()));
continue;
}
// Guard against empty response — covers both iteration 0 and post-tool cycles
let text = if text.trim().is_empty() {
warn!(
agent = %manifest.name,
iteration,
input_tokens = total_usage.input_tokens,
output_tokens = total_usage.output_tokens,
messages_count = messages.len(),
"Empty response from LLM (streaming) — guard activated"
);
if any_tools_executed {
"[Task completed — the agent executed tools but did not produce a text summary.]".to_string()
} else {
"[The model returned an empty response. This usually means the model is overloaded, the context is too large, or the API key lacks credits. Try again or check /status.]".to_string()
}
} else {
text
};
final_response = text.clone();
session.messages.push(Message::assistant(text));
// Prune NO_REPLY heartbeat turns to save context budget
crate::session_repair::prune_heartbeat_turns(&mut session.messages, 10);
memory
.save_session(session)
.map_err(|e| OpenFangError::Memory(e.to_string()))?;
// Remember this interaction (with embedding if available)
let interaction_text = format!(
"User asked: {}\nI responded: {}",
user_message, final_response
);
if let Some(emb) = embedding_driver {
match emb.embed_one(&interaction_text).await {
Ok(vec) => {
let _ = memory
.remember_with_embedding_async(
session.agent_id,
&interaction_text,
MemorySource::Conversation,
"episodic",
HashMap::new(),
Some(&vec),
)
.await;
}
Err(e) => {
warn!("Embedding for remember failed (streaming): {e}");
let _ = memory
.remember(
session.agent_id,
&interaction_text,
MemorySource::Conversation,
"episodic",
HashMap::new(),
)
.await;
}
}
} else {
let _ = memory
.remember(
session.agent_id,
&interaction_text,
MemorySource::Conversation,
"episodic",
HashMap::new(),
)
.await;
}
// Notify phase: Done
if let Some(cb) = on_phase {
cb(LoopPhase::Done);
}
info!(
agent = %manifest.name,
iterations = iteration + 1,
tokens = total_usage.total(),
"Streaming agent loop completed"
);
// Fire AgentLoopEnd hook
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"iterations": iteration + 1,
"response_length": final_response.len(),
}),
};
let _ = hook_reg.fire(&ctx);
}
return Ok(AgentLoopResult {
response: final_response,
total_usage,
iterations: iteration + 1,
cost_usd: None,
silent: false,
directives: Default::default(),
});
}
StopReason::ToolUse => {
// Reset MaxTokens continuation counter on tool use
consecutive_max_tokens = 0;
any_tools_executed = true;
let assistant_blocks = response.content.clone();
session.messages.push(Message {
role: Role::Assistant,
content: MessageContent::Blocks(assistant_blocks.clone()),
});
messages.push(Message {
role: Role::Assistant,
content: MessageContent::Blocks(assistant_blocks),
});
let allowed_tool_names: Vec<String> =
available_tools.iter().map(|t| t.name.clone()).collect();
let caller_id_str = session.agent_id.to_string();
// Execute each tool call with loop guard, timeout, and truncation
let mut tool_result_blocks = Vec::new();
for tool_call in &response.tool_calls {
// Loop guard check
let verdict = loop_guard.check(&tool_call.name, &tool_call.input);
match &verdict {
LoopGuardVerdict::CircuitBreak(msg) => {
warn!(tool = %tool_call.name, "Circuit breaker triggered (streaming)");
if let Err(e) = memory.save_session(session) {
warn!("Failed to save session on circuit break: {e}");
}
// Fire AgentLoopEnd hook on circuit break
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"reason": "circuit_break",
"error": msg.as_str(),
}),
};
let _ = hook_reg.fire(&ctx);
}
return Err(OpenFangError::Internal(msg.clone()));
}
LoopGuardVerdict::Block(msg) => {
warn!(tool = %tool_call.name, "Tool call blocked by loop guard (streaming)");
tool_result_blocks.push(ContentBlock::ToolResult {
tool_use_id: tool_call.id.clone(),
tool_name: tool_call.name.clone(),
content: msg.clone(),
is_error: true,
});
continue;
}
_ => {} // Allow or Warn — proceed with execution
}
debug!(tool = %tool_call.name, id = %tool_call.id, "Executing tool (streaming)");
// Notify phase: ToolUse
if let Some(cb) = on_phase {
let sanitized: String = tool_call
.name
.chars()
.filter(|c| !c.is_control())
.take(64)
.collect();
cb(LoopPhase::ToolUse {
tool_name: sanitized,
});
}
// Fire BeforeToolCall hook (can block execution)
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: &caller_id_str,
event: openfang_types::agent::HookEvent::BeforeToolCall,
data: serde_json::json!({
"tool_name": &tool_call.name,
"input": &tool_call.input,
}),
};
if let Err(reason) = hook_reg.fire(&ctx) {
tool_result_blocks.push(ContentBlock::ToolResult {
tool_use_id: tool_call.id.clone(),
tool_name: tool_call.name.clone(),
content: format!(
"Hook blocked tool '{}': {}",
tool_call.name, reason
),
is_error: true,
});
continue;
}
}
// Resolve effective exec policy (per-agent override or global)
let effective_exec_policy = manifest.exec_policy.as_ref();
// Timeout-wrapped execution
let result = match tokio::time::timeout(
Duration::from_secs(TOOL_TIMEOUT_SECS),
tool_runner::execute_tool(
&tool_call.id,
&tool_call.name,
&tool_call.input,
kernel.as_ref(),
Some(&allowed_tool_names),
Some(&caller_id_str),
skill_registry,
mcp_connections,
web_ctx,
browser_ctx,
if hand_allowed_env.is_empty() {
None
} else {
Some(&hand_allowed_env)
},
workspace_root,
media_engine,
effective_exec_policy,
tts_engine,
docker_config,
process_manager,
),
)
.await
{
Ok(result) => result,
Err(_) => {
warn!(tool = %tool_call.name, "Tool execution timed out after {}s (streaming)", TOOL_TIMEOUT_SECS);
openfang_types::tool::ToolResult {
tool_use_id: tool_call.id.clone(),
content: format!(
"Tool '{}' timed out after {}s.",
tool_call.name, TOOL_TIMEOUT_SECS
),
is_error: true,
}
}
};
// Fire AfterToolCall hook
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: caller_id_str.as_str(),
event: openfang_types::agent::HookEvent::AfterToolCall,
data: serde_json::json!({
"tool_name": &tool_call.name,
"result": &result.content,
"is_error": result.is_error,
}),
};
let _ = hook_reg.fire(&ctx);
}
// Dynamic truncation based on context budget (replaces flat MAX_TOOL_RESULT_CHARS)
let content = truncate_tool_result_dynamic(&result.content, &context_budget);
// Append warning if verdict was Warn
let final_content = if let LoopGuardVerdict::Warn(ref warn_msg) = verdict {
format!("{content}\n\n[LOOP GUARD] {warn_msg}")
} else {
content
};
// Notify client of tool execution result (detect dead consumer)
let preview: String = final_content.chars().take(300).collect();
if stream_tx
.send(StreamEvent::ToolExecutionResult {
name: tool_call.name.clone(),
result_preview: preview,
is_error: result.is_error,
})
.await
.is_err()
{
warn!(agent = %manifest.name, "Stream consumer disconnected — continuing tool loop but will not stream further");
}
tool_result_blocks.push(ContentBlock::ToolResult {
tool_use_id: result.tool_use_id,
tool_name: tool_call.name.clone(),
content: final_content,
is_error: result.is_error,
});
}
// Detect approval denials and inject guidance to prevent infinite retry loops
let denial_count = tool_result_blocks.iter().filter(|b| {
matches!(b, ContentBlock::ToolResult { content, is_error: true, .. }
if content.contains("requires human approval and was denied"))
}).count();
if denial_count > 0 {
tool_result_blocks.push(ContentBlock::Text {
text: format!(
"[System: {} tool call(s) were denied by approval policy. \
Do NOT retry denied tools. Explain to the user what you \
wanted to do and that it requires their approval.]",
denial_count
),
});
}
// Detect tool errors and inject guidance to prevent fabrication
let error_count = tool_result_blocks.iter().filter(|b| {
matches!(b, ContentBlock::ToolResult { is_error: true, .. })
}).count();
let non_denial_errors = error_count.saturating_sub(denial_count);
if non_denial_errors > 0 {
tool_result_blocks.push(ContentBlock::Text {
text: format!(
"[System: {} tool(s) returned errors. Report the error honestly \
to the user. Do NOT fabricate results or pretend the tool succeeded. \
If a search or fetch failed, tell the user it failed and suggest \
alternatives instead of making up data.]",
non_denial_errors
),
});
}
let tool_results_msg = Message {
role: Role::User,
content: MessageContent::Blocks(tool_result_blocks.clone()),
};
session.messages.push(tool_results_msg.clone());
messages.push(tool_results_msg);
if let Err(e) = memory.save_session(session) {
warn!("Failed to interim-save session: {e}");
}
}
StopReason::MaxTokens => {
consecutive_max_tokens += 1;
if consecutive_max_tokens >= MAX_CONTINUATIONS {
let text = response.text();
let text = if text.trim().is_empty() {
"[Partial response — token limit reached with no text output.]".to_string()
} else {
text
};
session.messages.push(Message::assistant(&text));
if let Err(e) = memory.save_session(session) {
warn!("Failed to save session on max continuations: {e}");
}
warn!(
iteration,
consecutive_max_tokens,
"Max continuations reached (streaming), returning partial response"
);
// Fire AgentLoopEnd hook
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"iterations": iteration + 1,
"reason": "max_continuations",
}),
};
let _ = hook_reg.fire(&ctx);
}
return Ok(AgentLoopResult {
response: text,
total_usage,
iterations: iteration + 1,
cost_usd: None,
silent: false,
directives: Default::default(),
});
}
let text = response.text();
session.messages.push(Message::assistant(&text));
messages.push(Message::assistant(&text));
session.messages.push(Message::user("Please continue."));
messages.push(Message::user("Please continue."));
warn!(iteration, "Max tokens hit (streaming), continuing");
}
}
}
if let Err(e) = memory.save_session(session) {
warn!("Failed to save session on max iterations: {e}");
}
// Fire AgentLoopEnd hook on max iterations exceeded
if let Some(hook_reg) = hooks {
let ctx = crate::hooks::HookContext {
agent_name: &manifest.name,
agent_id: agent_id_str.as_str(),
event: openfang_types::agent::HookEvent::AgentLoopEnd,
data: serde_json::json!({
"reason": "max_iterations_exceeded",
"iterations": max_iterations,
}),
};
let _ = hook_reg.fire(&ctx);
}
Err(OpenFangError::MaxIterationsExceeded(max_iterations))
}
/// Recover tool calls that LLMs (Groq/Llama, DeepSeek) output as plain text
/// instead of the proper `tool_calls` API field.
///
/// Parses patterns like `<function=tool_name>{"key":"value"}</function>` from
/// the model's text output, validates tool names against the available tools,
/// and returns synthetic `ToolCall` entries.
fn recover_text_tool_calls(text: &str, available_tools: &[ToolDefinition]) -> Vec<ToolCall> {
let mut calls = Vec::new();
let tool_names: Vec<&str> = available_tools.iter().map(|t| t.name.as_str()).collect();
// Pattern 1: <function=TOOL_NAME>JSON_BODY</function>
let mut search_from = 0;
while let Some(start) = text[search_from..].find("<function=") {
let abs_start = search_from + start;
let after_prefix = abs_start + "<function=".len();
// Extract tool name (ends at '>')
let Some(name_end) = text[after_prefix..].find('>') else {
search_from = after_prefix;
continue;
};
let tool_name = &text[after_prefix..after_prefix + name_end];
let json_start = after_prefix + name_end + 1;
// Find closing </function>
let Some(close_offset) = text[json_start..].find("</function>") else {
search_from = json_start;
continue;
};
let json_body = text[json_start..json_start + close_offset].trim();
search_from = json_start + close_offset + "</function>".len();
// Validate: tool name must be in available_tools
if !tool_names.contains(&tool_name) {
warn!(
tool = tool_name,
"Text-based tool call for unknown tool — skipping"
);
continue;
}
// Parse JSON input
let input: serde_json::Value = match serde_json::from_str(json_body) {
Ok(v) => v,
Err(e) => {
warn!(tool = tool_name, error = %e, "Failed to parse text-based tool call JSON — skipping");
continue;
}
};
info!(
tool = tool_name,
"Recovered text-based tool call → synthetic ToolUse"
);
calls.push(ToolCall {
id: format!("recovered_{}", uuid::Uuid::new_v4()),
name: tool_name.to_string(),
input,
});
}
// Pattern 2: <function>TOOL_NAME{JSON_BODY}</function>
// (Groq/Llama variant — tool name immediately followed by JSON object)
search_from = 0;
while let Some(start) = text[search_from..].find("<function>") {
let abs_start = search_from + start;
let after_tag = abs_start + "<function>".len();
// Find closing </function>
let Some(close_offset) = text[after_tag..].find("</function>") else {
search_from = after_tag;
continue;
};
let inner = &text[after_tag..after_tag + close_offset];
search_from = after_tag + close_offset + "</function>".len();
// The inner content is "tool_name{json}" — find the first '{' to split
let Some(brace_pos) = inner.find('{') else {
continue;
};
let tool_name = inner[..brace_pos].trim();
let json_body = inner[brace_pos..].trim();
if tool_name.is_empty() {
continue;
}
// Validate: tool name must be in available_tools
if !tool_names.contains(&tool_name) {
warn!(
tool = tool_name,
"Text-based tool call (variant 2) for unknown tool — skipping"
);
continue;
}
// Parse JSON input
let input: serde_json::Value = match serde_json::from_str(json_body) {
Ok(v) => v,
Err(e) => {
warn!(tool = tool_name, error = %e, "Failed to parse text-based tool call JSON (variant 2) — skipping");
continue;
}
};
// Avoid duplicates if pattern 1 already captured this call
if calls
.iter()
.any(|c| c.name == tool_name && c.input == input)
{
continue;
}
info!(
tool = tool_name,
"Recovered text-based tool call (variant 2) → synthetic ToolUse"
);
calls.push(ToolCall {
id: format!("recovered_{}", uuid::Uuid::new_v4()),
name: tool_name.to_string(),
input,
});
}
// Pattern 3: <tool>TOOL_NAME{JSON}</tool> (Qwen / DeepSeek variant)
search_from = 0;
while let Some(start) = text[search_from..].find("<tool>") {
let abs_start = search_from + start;
let after_tag = abs_start + "<tool>".len();
let Some(close_offset) = text[after_tag..].find("</tool>") else {
search_from = after_tag;
continue;
};
let inner = &text[after_tag..after_tag + close_offset];
search_from = after_tag + close_offset + "</tool>".len();
let Some(brace_pos) = inner.find('{') else {
continue;
};
let tool_name = inner[..brace_pos].trim();
let json_body = inner[brace_pos..].trim();
if tool_name.is_empty() || !tool_names.contains(&tool_name) {
continue;
}
let input: serde_json::Value = match serde_json::from_str(json_body) {
Ok(v) => v,
Err(_) => continue,
};
if calls
.iter()
.any(|c| c.name == tool_name && c.input == input)
{
continue;
}
info!(
tool = tool_name,
"Recovered text-based tool call (<tool> variant) → synthetic ToolUse"
);
calls.push(ToolCall {
id: format!("recovered_{}", uuid::Uuid::new_v4()),
name: tool_name.to_string(),
input,
});
}
// Pattern 4: Markdown code blocks containing tool_name {JSON}
// Matches: ```\nexec {"command":"ls"}\n``` or ```bash\nexec {"command":"ls"}\n```
{
let mut in_block = false;
let mut block_content = String::new();
for line in text.lines() {
let trimmed = line.trim();
if trimmed.starts_with("```") {
if in_block {
// End of block — try to extract tool call from content
let content = block_content.trim();
if let Some(brace_pos) = content.find('{') {
let potential_tool = content[..brace_pos].trim();
if tool_names.contains(&potential_tool) {
if let Ok(input) = serde_json::from_str::<serde_json::Value>(
content[brace_pos..].trim(),
) {
if !calls
.iter()
.any(|c| c.name == potential_tool && c.input == input)
{
info!(
tool = potential_tool,
"Recovered tool call from markdown code block"
);
calls.push(ToolCall {
id: format!("recovered_{}", uuid::Uuid::new_v4()),
name: potential_tool.to_string(),
input,
});
}
}
}
}
block_content.clear();
in_block = false;
} else {
in_block = true;
block_content.clear();
}
} else if in_block {
if !block_content.is_empty() {
block_content.push('\n');
}
block_content.push_str(trimmed);
}
}
}
// Pattern 5: Backtick-wrapped tool call: `tool_name {"key":"value"}`
{
let parts: Vec<&str> = text.split('`').collect();
// Every odd-indexed element is inside backticks
for chunk in parts.iter().skip(1).step_by(2) {
let trimmed = chunk.trim();
if let Some(brace_pos) = trimmed.find('{') {
let potential_tool = trimmed[..brace_pos].trim();
if !potential_tool.is_empty()
&& !potential_tool.contains(' ')
&& tool_names.contains(&potential_tool)
{
if let Ok(input) =
serde_json::from_str::<serde_json::Value>(trimmed[brace_pos..].trim())
{
if !calls
.iter()
.any(|c| c.name == potential_tool && c.input == input)
{
info!(
tool = potential_tool,
"Recovered tool call from backtick-wrapped text"
);
calls.push(ToolCall {
id: format!("recovered_{}", uuid::Uuid::new_v4()),
name: potential_tool.to_string(),
input,
});
}
}
}
}
}
}
calls
}
#[cfg(test)]
mod tests {
use super::*;
use crate::llm_driver::{CompletionResponse, LlmError};
use async_trait::async_trait;
use openfang_types::tool::ToolCall;
use std::sync::atomic::{AtomicU32, Ordering};
#[test]
fn test_max_iterations_constant() {
assert_eq!(MAX_ITERATIONS, 50);
}
#[test]
fn test_retry_constants() {
assert_eq!(MAX_RETRIES, 3);
assert_eq!(BASE_RETRY_DELAY_MS, 1000);
}
#[test]
fn test_dynamic_truncate_short_unchanged() {
use crate::context_budget::{truncate_tool_result_dynamic, ContextBudget};
let budget = ContextBudget::new(200_000);
let short = "Hello, world!";
assert_eq!(truncate_tool_result_dynamic(short, &budget), short);
}
#[test]
fn test_dynamic_truncate_over_limit() {
use crate::context_budget::{truncate_tool_result_dynamic, ContextBudget};
let budget = ContextBudget::new(200_000);
let long = "x".repeat(budget.per_result_cap() + 10_000);
let result = truncate_tool_result_dynamic(&long, &budget);
assert!(result.len() <= budget.per_result_cap() + 200);
assert!(result.contains("[TRUNCATED:"));
}
#[test]
fn test_dynamic_truncate_newline_boundary() {
use crate::context_budget::{truncate_tool_result_dynamic, ContextBudget};
// Small budget to force truncation
let budget = ContextBudget::new(1_000);
let content = (0..200)
.map(|i| format!("line {i}"))
.collect::<Vec<_>>()
.join("\n");
let result = truncate_tool_result_dynamic(&content, &budget);
// Should break at a newline, not mid-line
let before_marker = result.split("[TRUNCATED:").next().unwrap();
let trimmed = before_marker.trim_end();
assert!(!trimmed.is_empty());
}
#[test]
fn test_max_continuations_constant() {
assert_eq!(MAX_CONTINUATIONS, 5);
}
#[test]
fn test_tool_timeout_constant() {
assert_eq!(TOOL_TIMEOUT_SECS, 120);
}
#[test]
fn test_max_history_messages() {
assert_eq!(MAX_HISTORY_MESSAGES, 20);
}
// --- Integration tests for empty response guards ---
fn test_manifest() -> AgentManifest {
AgentManifest {
name: "test-agent".to_string(),
model: openfang_types::agent::ModelConfig {
system_prompt: "You are a test agent.".to_string(),
..Default::default()
},
..Default::default()
}
}
/// Mock driver that simulates: first call returns ToolUse with no text,
/// second call returns EndTurn with empty text. This reproduces the bug
/// where the LLM ends with no text after a tool-use cycle.
struct EmptyAfterToolUseDriver {
call_count: AtomicU32,
}
impl EmptyAfterToolUseDriver {
fn new() -> Self {
Self {
call_count: AtomicU32::new(0),
}
}
}
#[async_trait]
impl LlmDriver for EmptyAfterToolUseDriver {
async fn complete(
&self,
_request: CompletionRequest,
) -> Result<CompletionResponse, LlmError> {
let call = self.call_count.fetch_add(1, Ordering::Relaxed);
if call == 0 {
// First call: LLM wants to use a tool (with no text block)
Ok(CompletionResponse {
content: vec![ContentBlock::ToolUse {
id: "tool_1".to_string(),
name: "fake_tool".to_string(),
input: serde_json::json!({"query": "test"}),
}],
stop_reason: StopReason::ToolUse,
tool_calls: vec![ToolCall {
id: "tool_1".to_string(),
name: "fake_tool".to_string(),
input: serde_json::json!({"query": "test"}),
}],
usage: TokenUsage {
input_tokens: 10,
output_tokens: 5,
},
})
} else {
// Second call: LLM returns EndTurn with EMPTY text (the bug)
Ok(CompletionResponse {
content: vec![],
stop_reason: StopReason::EndTurn,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 10,
output_tokens: 0,
},
})
}
}
}
/// Mock driver that returns empty text with MaxTokens stop reason,
/// repeated MAX_CONTINUATIONS times to trigger the max continuations path.
struct EmptyMaxTokensDriver;
#[async_trait]
impl LlmDriver for EmptyMaxTokensDriver {
async fn complete(
&self,
_request: CompletionRequest,
) -> Result<CompletionResponse, LlmError> {
Ok(CompletionResponse {
content: vec![],
stop_reason: StopReason::MaxTokens,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 10,
output_tokens: 0,
},
})
}
}
/// Mock driver that returns normal text (sanity check).
struct NormalDriver;
#[async_trait]
impl LlmDriver for NormalDriver {
async fn complete(
&self,
_request: CompletionRequest,
) -> Result<CompletionResponse, LlmError> {
Ok(CompletionResponse {
content: vec![ContentBlock::Text {
text: "Hello from the agent!".to_string(),
}],
stop_reason: StopReason::EndTurn,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 10,
output_tokens: 8,
},
})
}
}
#[tokio::test]
async fn test_empty_response_after_tool_use_returns_fallback() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(EmptyAfterToolUseDriver::new());
let result = run_agent_loop(
&manifest,
"Do something with tools",
&mut session,
&memory,
driver,
&[], // no tools registered — the tool call will fail, which is fine
None,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Loop should complete without error");
// The response MUST NOT be empty — it should contain our fallback text
assert!(
!result.response.trim().is_empty(),
"Response should not be empty after tool use, got: {:?}",
result.response
);
assert!(
result.response.contains("Task completed"),
"Expected fallback message, got: {:?}",
result.response
);
}
#[tokio::test]
async fn test_empty_response_max_tokens_returns_fallback() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(EmptyMaxTokensDriver);
let result = run_agent_loop(
&manifest,
"Tell me something long",
&mut session,
&memory,
driver,
&[],
None,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Loop should complete without error");
// Should hit MAX_CONTINUATIONS and return fallback instead of empty
assert!(
!result.response.trim().is_empty(),
"Response should not be empty on max tokens, got: {:?}",
result.response
);
assert!(
result.response.contains("token limit"),
"Expected max-tokens fallback message, got: {:?}",
result.response
);
}
#[tokio::test]
async fn test_normal_response_not_replaced_by_fallback() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(NormalDriver);
let result = run_agent_loop(
&manifest,
"Say hello",
&mut session,
&memory,
driver,
&[],
None,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Loop should complete without error");
// Normal response should pass through unchanged
assert_eq!(result.response, "Hello from the agent!");
}
#[tokio::test]
async fn test_streaming_empty_response_after_tool_use_returns_fallback() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(EmptyAfterToolUseDriver::new());
let (tx, _rx) = mpsc::channel(64);
let result = run_agent_loop_streaming(
&manifest,
"Do something with tools",
&mut session,
&memory,
driver,
&[],
None,
tx,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Streaming loop should complete without error");
assert!(
!result.response.trim().is_empty(),
"Streaming response should not be empty after tool use, got: {:?}",
result.response
);
assert!(
result.response.contains("Task completed"),
"Expected fallback message in streaming, got: {:?}",
result.response
);
}
/// Mock driver that returns empty text on first call (EndTurn), then normal text on second.
/// This tests the one-shot retry logic for iteration 0 empty responses.
struct EmptyThenNormalDriver {
call_count: AtomicU32,
}
impl EmptyThenNormalDriver {
fn new() -> Self {
Self {
call_count: AtomicU32::new(0),
}
}
}
#[async_trait]
impl LlmDriver for EmptyThenNormalDriver {
async fn complete(
&self,
_request: CompletionRequest,
) -> Result<CompletionResponse, LlmError> {
let call = self.call_count.fetch_add(1, Ordering::Relaxed);
if call == 0 {
// First call: empty EndTurn (triggers retry)
Ok(CompletionResponse {
content: vec![],
stop_reason: StopReason::EndTurn,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 10,
output_tokens: 0,
},
})
} else {
// Second call (retry): normal response
Ok(CompletionResponse {
content: vec![ContentBlock::Text {
text: "Recovered after retry!".to_string(),
}],
stop_reason: StopReason::EndTurn,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 15,
output_tokens: 8,
},
})
}
}
}
/// Mock driver that always returns empty EndTurn (no recovery on retry).
/// Tests that the fallback message appears when retry also fails.
struct AlwaysEmptyDriver;
#[async_trait]
impl LlmDriver for AlwaysEmptyDriver {
async fn complete(
&self,
_request: CompletionRequest,
) -> Result<CompletionResponse, LlmError> {
Ok(CompletionResponse {
content: vec![],
stop_reason: StopReason::EndTurn,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 10,
output_tokens: 0,
},
})
}
}
#[tokio::test]
async fn test_empty_first_response_retries_and_recovers() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(EmptyThenNormalDriver::new());
let result = run_agent_loop(
&manifest,
"Hello",
&mut session,
&memory,
driver,
&[],
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Loop should recover via retry");
assert_eq!(result.response, "Recovered after retry!");
assert_eq!(
result.iterations, 2,
"Should have taken 2 iterations (retry)"
);
}
#[tokio::test]
async fn test_empty_first_response_fallback_when_retry_also_empty() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(AlwaysEmptyDriver);
let result = run_agent_loop(
&manifest,
"Hello",
&mut session,
&memory,
driver,
&[],
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Loop should complete with fallback");
// No tools were executed, so should get the empty response message
assert!(
result.response.contains("empty response"),
"Expected empty response fallback (no tools executed), got: {:?}",
result.response
);
}
#[tokio::test]
async fn test_max_history_messages_constant() {
assert_eq!(MAX_HISTORY_MESSAGES, 20);
}
#[tokio::test]
async fn test_streaming_empty_response_max_tokens_returns_fallback() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(EmptyMaxTokensDriver);
let (tx, _rx) = mpsc::channel(64);
let result = run_agent_loop_streaming(
&manifest,
"Tell me something long",
&mut session,
&memory,
driver,
&[],
None,
tx,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Streaming loop should complete without error");
assert!(
!result.response.trim().is_empty(),
"Streaming response should not be empty on max tokens, got: {:?}",
result.response
);
assert!(
result.response.contains("token limit"),
"Expected max-tokens fallback in streaming, got: {:?}",
result.response
);
}
#[test]
fn test_recover_text_tool_calls_basic() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search the web".into(),
input_schema: serde_json::json!({}),
}];
let text =
r#"Let me search for that. <function=web_search>{"query":"rust async"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "web_search");
assert_eq!(calls[0].input["query"], "rust async");
assert!(calls[0].id.starts_with("recovered_"));
}
#[test]
fn test_recover_text_tool_calls_unknown_tool() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search the web".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"<function=hack_system>{"cmd":"rm -rf /"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert!(calls.is_empty(), "Unknown tools should be rejected");
}
#[test]
fn test_recover_text_tool_calls_invalid_json() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search the web".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"<function=web_search>not valid json</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert!(calls.is_empty(), "Invalid JSON should be skipped");
}
#[test]
fn test_recover_text_tool_calls_multiple() {
let tools = vec![
ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
},
ToolDefinition {
name: "read_file".into(),
description: "Read a file".into(),
input_schema: serde_json::json!({}),
},
];
let text = r#"<function=web_search>{"query":"hello"}</function> then <function=read_file>{"path":"a.txt"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 2);
assert_eq!(calls[0].name, "web_search");
assert_eq!(calls[1].name, "read_file");
}
#[test]
fn test_recover_text_tool_calls_no_pattern() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
let text = "Just a normal response with no tool calls.";
let calls = recover_text_tool_calls(text, &tools);
assert!(calls.is_empty());
}
#[test]
fn test_recover_text_tool_calls_empty_tools() {
let text = r#"<function=web_search>{"query":"hello"}</function>"#;
let calls = recover_text_tool_calls(text, &[]);
assert!(calls.is_empty(), "No tools = no recovery");
}
// --- Deep edge-case tests for text-to-tool recovery ---
#[test]
fn test_recover_text_tool_calls_nested_json() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"<function=web_search>{"query":"rust","filters":{"lang":"en","year":2024}}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].input["filters"]["lang"], "en");
}
#[test]
fn test_recover_text_tool_calls_with_surrounding_text() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
let text = "Sure, let me search that for you.\n\n<function=web_search>{\"query\":\"rust async programming\"}</function>\n\nI'll get back to you with results.";
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].input["query"], "rust async programming");
}
#[test]
fn test_recover_text_tool_calls_whitespace_in_json() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
// Some models emit pretty-printed JSON
let text = "<function=web_search>\n {\"query\": \"hello world\"}\n</function>";
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].input["query"], "hello world");
}
#[test]
fn test_recover_text_tool_calls_unclosed_tag() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
// Missing </function> — should gracefully skip
let text = r#"<function=web_search>{"query":"test"}"#;
let calls = recover_text_tool_calls(text, &tools);
assert!(calls.is_empty(), "Unclosed tag should be skipped");
}
#[test]
fn test_recover_text_tool_calls_missing_closing_bracket() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
// Missing > after tool name
let text = r#"<function=web_search{"query":"test"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
// The parser finds > inside JSON, will likely produce invalid tool name
// or invalid JSON — either way, should not panic
// (just verifying no panic / no bad behavior)
let _ = calls;
}
#[test]
fn test_recover_text_tool_calls_empty_json_object() {
let tools = vec![ToolDefinition {
name: "list_files".into(),
description: "List".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"<function=list_files>{}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "list_files");
assert_eq!(calls[0].input, serde_json::json!({}));
}
#[test]
fn test_recover_text_tool_calls_mixed_valid_invalid() {
let tools = vec![
ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
},
ToolDefinition {
name: "read_file".into(),
description: "Read".into(),
input_schema: serde_json::json!({}),
},
];
// First: valid, second: unknown tool, third: valid
let text = r#"<function=web_search>{"q":"a"}</function> <function=unknown>{"x":1}</function> <function=read_file>{"path":"b"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 2, "Should recover 2 valid, skip 1 unknown");
assert_eq!(calls[0].name, "web_search");
assert_eq!(calls[1].name, "read_file");
}
// --- Variant 2 pattern tests: <function>NAME{JSON}</function> ---
#[test]
fn test_recover_variant2_basic() {
let tools = vec![ToolDefinition {
name: "web_fetch".into(),
description: "Fetch".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"<function>web_fetch{"url":"https://example.com"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "web_fetch");
assert_eq!(calls[0].input["url"], "https://example.com");
}
#[test]
fn test_recover_variant2_unknown_tool() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"<function>unknown_tool{"q":"test"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 0);
}
#[test]
fn test_recover_variant2_with_surrounding_text() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"Let me search for that. <function>web_search{"query":"rust lang"}</function> I'll find the answer."#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "web_search");
}
#[test]
fn test_recover_both_variants_mixed() {
let tools = vec![
ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
},
ToolDefinition {
name: "web_fetch".into(),
description: "Fetch".into(),
input_schema: serde_json::json!({}),
},
];
// Mix of variant 1 and variant 2
let text = r#"<function=web_search>{"q":"a"}</function> <function>web_fetch{"url":"https://x.com"}</function>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 2);
assert_eq!(calls[0].name, "web_search");
assert_eq!(calls[1].name, "web_fetch");
}
#[test]
fn test_recover_tool_tag_variant() {
let tools = vec![ToolDefinition {
name: "exec".into(),
description: "Execute".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"I'll run that for you. <tool>exec{"command":"ls -la"}</tool>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "exec");
assert_eq!(calls[0].input["command"], "ls -la");
}
#[test]
fn test_recover_markdown_code_block() {
let tools = vec![ToolDefinition {
name: "exec".into(),
description: "Execute".into(),
input_schema: serde_json::json!({}),
}];
let text = "I'll execute that command:\n```\nexec {\"command\": \"ls -la\"}\n```";
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "exec");
assert_eq!(calls[0].input["command"], "ls -la");
}
#[test]
fn test_recover_markdown_code_block_with_lang() {
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search".into(),
input_schema: serde_json::json!({}),
}];
let text = "```json\nweb_search {\"query\": \"rust\"}\n```";
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "web_search");
}
#[test]
fn test_recover_backtick_wrapped() {
let tools = vec![ToolDefinition {
name: "exec".into(),
description: "Execute".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"Let me run `exec {"command":"pwd"}` for you."#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "exec");
assert_eq!(calls[0].input["command"], "pwd");
}
#[test]
fn test_recover_backtick_ignores_unknown_tool() {
let tools = vec![ToolDefinition {
name: "exec".into(),
description: "Execute".into(),
input_schema: serde_json::json!({}),
}];
let text = r#"Try `unknown_tool {"key":"val"}` instead."#;
let calls = recover_text_tool_calls(text, &tools);
assert!(calls.is_empty());
}
#[test]
fn test_recover_no_duplicates_across_patterns() {
let tools = vec![ToolDefinition {
name: "exec".into(),
description: "Execute".into(),
input_schema: serde_json::json!({}),
}];
// Same call in both function tag and tool tag — should only appear once
let text = r#"<function=exec>{"command":"ls"}</function> <tool>exec{"command":"ls"}</tool>"#;
let calls = recover_text_tool_calls(text, &tools);
assert_eq!(calls.len(), 1);
}
// --- End-to-end integration test: text-as-tool-call recovery through agent loop ---
/// Mock driver that simulates a Groq/Llama model outputting tool calls as text.
/// Call 1: Returns text with `<function=web_search>...</function>` (EndTurn, no tool_calls)
/// Call 2: Returns a normal text response (after tool result is provided)
struct TextToolCallDriver {
call_count: AtomicU32,
}
impl TextToolCallDriver {
fn new() -> Self {
Self {
call_count: AtomicU32::new(0),
}
}
}
#[async_trait]
impl LlmDriver for TextToolCallDriver {
async fn complete(
&self,
_request: CompletionRequest,
) -> Result<CompletionResponse, LlmError> {
let call = self.call_count.fetch_add(1, Ordering::Relaxed);
if call == 0 {
// Simulate Groq/Llama: tool call as text, not in tool_calls field
Ok(CompletionResponse {
content: vec![ContentBlock::Text {
text: r#"Let me search for that. <function=web_search>{"query":"rust async"}</function>"#.to_string(),
}],
stop_reason: StopReason::EndTurn,
tool_calls: vec![], // BUG: no tool_calls!
usage: TokenUsage {
input_tokens: 20,
output_tokens: 15,
},
})
} else {
// After tool result, return normal response
Ok(CompletionResponse {
content: vec![ContentBlock::Text {
text: "Based on the search results, Rust async is great!".to_string(),
}],
stop_reason: StopReason::EndTurn,
tool_calls: vec![],
usage: TokenUsage {
input_tokens: 30,
output_tokens: 12,
},
})
}
}
}
#[tokio::test]
async fn test_text_tool_call_recovery_e2e() {
// This is THE critical test: a model outputs a tool call as text,
// the recovery code detects it, promotes it to ToolUse, executes the tool,
// and the agent loop continues to produce a final response.
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(TextToolCallDriver::new());
// Provide web_search as an available tool so recovery can match it
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search the web".into(),
input_schema: serde_json::json!({
"type": "object",
"properties": {
"query": {"type": "string"}
}
}),
}];
let result = run_agent_loop(
&manifest,
"Search for rust async programming",
&mut session,
&memory,
driver,
&tools,
None,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Agent loop should complete");
// The response should contain the second call's output, NOT the raw function tag
assert!(
!result.response.contains("<function="),
"Response should not contain raw function tags, got: {:?}",
result.response
);
assert!(
result.iterations >= 2,
"Should have at least 2 iterations (tool call + final response), got: {}",
result.iterations
);
// Verify the final text response came through
assert!(
result.response.contains("search results") || result.response.contains("Rust async"),
"Expected final response text, got: {:?}",
result.response
);
}
/// Mock driver that returns NO text-based tool calls — just normal text.
/// Verifies recovery does NOT interfere with normal flow.
#[tokio::test]
async fn test_normal_flow_unaffected_by_recovery() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(NormalDriver);
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search the web".into(),
input_schema: serde_json::json!({}),
}];
let result = run_agent_loop(
&manifest,
"Say hello",
&mut session,
&memory,
driver,
&tools, // tools available but not used
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
)
.await
.expect("Normal loop should complete");
assert_eq!(result.response, "Hello from the agent!");
assert_eq!(
result.iterations, 1,
"Normal response should complete in 1 iteration"
);
}
// --- Streaming path: text-as-tool-call recovery ---
#[tokio::test]
async fn test_text_tool_call_recovery_streaming_e2e() {
let memory = openfang_memory::MemorySubstrate::open_in_memory(0.01).unwrap();
let agent_id = openfang_types::agent::AgentId::new();
let mut session = openfang_memory::session::Session {
id: openfang_types::agent::SessionId::new(),
agent_id,
messages: Vec::new(),
context_window_tokens: 0,
label: None,
};
let manifest = test_manifest();
let driver: Arc<dyn LlmDriver> = Arc::new(TextToolCallDriver::new());
let tools = vec![ToolDefinition {
name: "web_search".into(),
description: "Search the web".into(),
input_schema: serde_json::json!({
"type": "object",
"properties": {
"query": {"type": "string"}
}
}),
}];
let (tx, mut rx) = mpsc::channel(64);
let result = run_agent_loop_streaming(
&manifest,
"Search for rust async programming",
&mut session,
&memory,
driver,
&tools,
None,
tx,
None,
None,
None,
None,
None,
None,
None, // on_phase
None, // media_engine
None, // tts_engine
None, // docker_config
None, // hooks
None, // context_window_tokens
None, // process_manager
)
.await
.expect("Streaming loop should complete");
// Same assertions as non-streaming
assert!(
!result.response.contains("<function="),
"Streaming: response should not contain raw function tags, got: {:?}",
result.response
);
assert!(
result.iterations >= 2,
"Streaming: should have at least 2 iterations, got: {}",
result.iterations
);
// Drain the stream channel to verify events were sent
let mut events = Vec::new();
while let Ok(ev) = rx.try_recv() {
events.push(ev);
}
assert!(!events.is_empty(), "Should have received stream events");
}
}
Reference in New Issue View Git Blame Copy Permalink