mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
refactor(agent): unify the three agent-turn loops into one TurnEngine (#3012)
This commit is contained in:
Steven Enamakel
@@ -92,6 +92,8 @@ loop {
|
||||
|
||||
Every iteration emits a real-time `AgentProgress` event so the UI can render token-by-token streaming, "calling tool X" status, and per-iteration cost updates.
|
||||
|
||||
**One engine, three entry points.** This loop lives in one place — `engine::run_turn_engine` (`harness/engine/`) — and every caller drives it: `Agent::turn` (web/desktop chat), `run_tool_call_loop` (the `agent.run_turn` bus handler for other channels + triage), and `run_subagent` (spawned sub-agents). What varies per caller is supplied through small seams the engine calls into: a `ToolSource` (which tools are advertised + how a call executes), a `ProgressReporter` (top-level `Turn*` events with streaming vs. nested `Subagent*` events), a `TurnObserver` (context management, transcript persistence, history shape), a `CheckpointStrategy` (error vs. summarize when the iteration cap is hit), and a `ResponseParser` (the `ToolDispatcher` dialect). The per-call executor (`run_one_tool`), the repeated-failure circuit breaker, and the `ProviderDelta → AgentProgress` stream forwarder are shared across all three, so they can't drift.
|
||||
|
||||
### Tool dispatch and tool-call dialects
|
||||
|
||||
Different LLMs speak different tool-calling dialects. The harness abstracts that with a `ToolDispatcher` trait, which has three concrete implementations:
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
//! Max-iteration checkpoint seam.
|
||||
//!
|
||||
//! When a turn exhausts its iteration budget the three callers diverge:
|
||||
//!
|
||||
//! * the channel/CLI loop returns the typed `AgentError::MaxIterationsExceeded`
|
||||
//! so `Agent::run_single` can downcast and suppress Sentry noise
|
||||
//! ([`ErrorCheckpoint`]);
|
||||
//! * the subagent and `Agent::turn` instead summarize the run-so-far into a
|
||||
//! resumable checkpoint string and return it as the turn's result (the
|
||||
//! `SummarizeCheckpoint`, landed with the subagent/Agent migrations).
|
||||
//!
|
||||
//! [`CheckpointStrategy::on_max_iter`] receives the accumulated tool digest so a
|
||||
//! summarizing strategy can produce a root-cause-aware checkpoint.
|
||||
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::openhuman::inference::provider::UsageInfo;
|
||||
|
||||
/// A checkpoint result. `usage`, when present, is the provider usage from a
|
||||
/// summarization call the strategy made — the engine folds it into the turn's
|
||||
/// cost and reports it to the observer so token accounting stays complete.
|
||||
pub(crate) struct CheckpointOutcome {
|
||||
pub text: String,
|
||||
pub usage: Option<UsageInfo>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub(crate) trait CheckpointStrategy: Send + Sync {
|
||||
/// Produce the turn's result after the iteration cap is hit, or return an
|
||||
/// error to surface the cap to the caller. `digest` is the accumulated
|
||||
/// `tool → outcome` summary of the run so far.
|
||||
async fn on_max_iter(&self, digest: &str, max_iterations: usize) -> Result<CheckpointOutcome>;
|
||||
}
|
||||
|
||||
/// Surface the cap as the typed [`AgentError::MaxIterationsExceeded`], boxed
|
||||
/// through `anyhow::Error`, so downstream wrappers — notably
|
||||
/// `Agent::run_single` — can downcast and suppress Sentry emission for this
|
||||
/// deterministic agent-state outcome (OPENHUMAN-TAURI-99 / -98).
|
||||
pub(crate) struct ErrorCheckpoint;
|
||||
|
||||
#[async_trait]
|
||||
impl CheckpointStrategy for ErrorCheckpoint {
|
||||
async fn on_max_iter(&self, _digest: &str, max_iterations: usize) -> Result<CheckpointOutcome> {
|
||||
Err(anyhow::Error::new(
|
||||
crate::openhuman::agent::error::AgentError::MaxIterationsExceeded {
|
||||
max: max_iterations,
|
||||
},
|
||||
))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,554 @@
|
||||
//! The unified turn loop.
|
||||
//!
|
||||
//! [`run_turn_engine`] is the single agentic loop the harness runs: announce the
|
||||
//! turn, then per iteration run the stop-hook + context guards, send the
|
||||
//! provider request (streaming deltas when the [`ProgressReporter`] supplies a
|
||||
//! sink), parse the response, either return the final text or execute every
|
||||
//! requested tool through the [`ToolSource`] and loop again — bailing early via
|
||||
//! the shared repeated-failure circuit breaker, or handing the iteration cap to
|
||||
//! the [`CheckpointStrategy`].
|
||||
//!
|
||||
//! Everything that varies per caller lives behind a seam: [`ToolSource`] (tool
|
||||
//! advertisement + per-call execution), [`ProgressReporter`] (Turn* vs
|
||||
//! Subagent* events + streaming), [`TurnObserver`] (context management,
|
||||
//! transcript persistence, worker-thread mirroring) and [`CheckpointStrategy`]
|
||||
//! (error vs summarize on cap). The universal concerns — stop hooks, the
|
||||
//! context guard, token-budget trimming, native/text parsing and the circuit
|
||||
//! breaker — stay inline.
|
||||
|
||||
use anyhow::Result;
|
||||
use std::fmt::Write as _;
|
||||
use std::io::Write as _;
|
||||
|
||||
use crate::openhuman::agent::cost::TurnCost;
|
||||
use crate::openhuman::agent::multimodal;
|
||||
use crate::openhuman::agent::stop_hooks::{current_stop_hooks, StopDecision, TurnState};
|
||||
use crate::openhuman::context::guard::{ContextCheckResult, ContextGuard};
|
||||
use crate::openhuman::inference::model_context::context_window_for_model;
|
||||
use crate::openhuman::inference::provider::{
|
||||
ChatMessage, ChatRequest, Provider, ProviderCapabilityError,
|
||||
};
|
||||
|
||||
use super::super::parse::build_native_assistant_history;
|
||||
use super::super::token_budget::trim_chat_messages_to_budget;
|
||||
use super::super::tool_loop::{RepeatFailureGuard, STREAM_CHUNK_MIN_CHARS};
|
||||
use super::checkpoint::CheckpointStrategy;
|
||||
use super::parser::ResponseParser;
|
||||
use super::progress::ProgressReporter;
|
||||
use super::state::TurnObserver;
|
||||
use super::tool_source::ToolSource;
|
||||
|
||||
/// What a completed turn yields. `text` is the final assistant text (or the
|
||||
/// circuit-breaker / checkpoint summary); `iterations` and `cost` let stateful
|
||||
/// callers attribute the run.
|
||||
pub(crate) struct TurnEngineOutcome {
|
||||
pub text: String,
|
||||
pub iterations: u32,
|
||||
pub cost: TurnCost,
|
||||
/// True when the turn stopped because it hit the iteration cap (the
|
||||
/// `CheckpointStrategy` produced `text`), false for a normal final response
|
||||
/// or an early circuit-breaker halt. `Agent::turn` keys its checkpoint-only
|
||||
/// history/transcript handling off this.
|
||||
pub hit_cap: bool,
|
||||
}
|
||||
|
||||
/// Truncate a digest entry's body so a huge tool result can't blow up the
|
||||
/// checkpoint summary. Mirrors the subagent's previous `truncate_with_ellipsis`.
|
||||
fn truncate_with_ellipsis(s: &str, max: usize) -> String {
|
||||
if s.chars().count() <= max {
|
||||
return s.to_string();
|
||||
}
|
||||
let head: String = s.chars().take(max).collect();
|
||||
format!("{head}…")
|
||||
}
|
||||
|
||||
/// Run the agent loop over `history` using `tools`. `max_iterations` must be
|
||||
/// pre-normalized (callers map `0` to a sane default). See the module docs for
|
||||
/// the per-iteration flow.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) async fn run_turn_engine(
|
||||
provider: &dyn Provider,
|
||||
history: &mut Vec<ChatMessage>,
|
||||
tools: &mut dyn ToolSource,
|
||||
progress: &dyn ProgressReporter,
|
||||
observer: &mut dyn TurnObserver,
|
||||
checkpoint: &dyn CheckpointStrategy,
|
||||
parser: &dyn ResponseParser,
|
||||
provider_name: &str,
|
||||
model: &str,
|
||||
temperature: f64,
|
||||
silent: bool,
|
||||
multimodal_config: &crate::openhuman::config::MultimodalConfig,
|
||||
max_iterations: usize,
|
||||
on_delta: Option<tokio::sync::mpsc::Sender<String>>,
|
||||
) -> Result<TurnEngineOutcome> {
|
||||
let mut context_guard = context_window_for_model(model)
|
||||
.map(ContextGuard::with_context_window)
|
||||
.unwrap_or_else(ContextGuard::new);
|
||||
let mut turn_cost = TurnCost::new();
|
||||
|
||||
// Compiled digest of this run's tool calls + results, for a graceful
|
||||
// checkpoint if the iteration cap is hit. Accumulated as the loop runs so
|
||||
// it survives history trimming.
|
||||
let mut run_tool_digest = String::new();
|
||||
|
||||
// Announce turn start. Lifecycle (turn/iteration) events are `.await`-ed so
|
||||
// they survive downstream backpressure — dropping one would desync the
|
||||
// web-channel progress bridge.
|
||||
progress.turn_started().await;
|
||||
|
||||
let stop_hooks = current_stop_hooks();
|
||||
// Repeated-failure circuit breaker — halts with a root cause rather than
|
||||
// grinding to `max_iterations`.
|
||||
let mut failure_guard = RepeatFailureGuard::new();
|
||||
let mut halt_reason: Option<String> = None;
|
||||
for iteration in 0..max_iterations {
|
||||
progress
|
||||
.iteration_started((iteration + 1) as u32, max_iterations as u32)
|
||||
.await;
|
||||
|
||||
// ── Stop hooks: policy check before the next LLM call ──
|
||||
if !stop_hooks.is_empty() {
|
||||
let state = TurnState {
|
||||
iteration: (iteration + 1) as u32,
|
||||
max_iterations: max_iterations as u32,
|
||||
cost: &turn_cost,
|
||||
model,
|
||||
};
|
||||
for hook in &stop_hooks {
|
||||
match hook.check(&state).await {
|
||||
StopDecision::Continue => {}
|
||||
StopDecision::Stop { reason } => {
|
||||
tracing::warn!(
|
||||
iteration = (iteration + 1),
|
||||
hook = hook.name(),
|
||||
reason = %reason,
|
||||
"[agent_loop] stop hook triggered — aborting turn"
|
||||
);
|
||||
anyhow::bail!("Agent turn stopped by hook '{}': {reason}", hook.name());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── 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)",
|
||||
crate::openhuman::context::guard::COMPACTION_TRIGGER_THRESHOLD * 100.0
|
||||
);
|
||||
}
|
||||
ContextCheckResult::ContextExhausted {
|
||||
utilization_pct,
|
||||
reason,
|
||||
} => {
|
||||
let msg = format!("Context window exhausted ({utilization_pct}% full): {reason}");
|
||||
crate::core::observability::report_error(
|
||||
msg.as_str(),
|
||||
"agent",
|
||||
"context_exhausted",
|
||||
&[
|
||||
("provider", provider_name),
|
||||
("model", model),
|
||||
("utilization_pct", &utilization_pct.to_string()),
|
||||
],
|
||||
);
|
||||
anyhow::bail!(msg);
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(context_window) = context_window_for_model(model) {
|
||||
let budget_outcome = trim_chat_messages_to_budget(history, context_window);
|
||||
if budget_outcome.trimmed {
|
||||
log::warn!(
|
||||
"[agent_loop] pre-dispatch history trimmed model={} context_window={} original_tokens={} final_tokens={} messages_removed={}",
|
||||
model,
|
||||
context_window,
|
||||
budget_outcome.original_tokens,
|
||||
budget_outcome.final_tokens,
|
||||
budget_outcome.messages_removed
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
model,
|
||||
context_window,
|
||||
estimated_tokens = budget_outcome.final_tokens,
|
||||
"[agent_loop] pre-dispatch token budget ok"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Caller-specific pre-dispatch work (e.g. Agent's ContextManager).
|
||||
observer.before_dispatch(history, iteration).await?;
|
||||
|
||||
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() {
|
||||
let cap_err = ProviderCapabilityError {
|
||||
provider: provider_name.to_string(),
|
||||
capability: "vision".to_string(),
|
||||
message: format!(
|
||||
"received {image_marker_count} image marker(s), but this provider does not support vision input"
|
||||
),
|
||||
};
|
||||
crate::core::observability::report_error(
|
||||
&cap_err,
|
||||
"agent",
|
||||
"provider_capability",
|
||||
&[
|
||||
("provider", provider_name),
|
||||
("capability", "vision"),
|
||||
("model", model),
|
||||
],
|
||||
);
|
||||
return Err(cap_err.into());
|
||||
}
|
||||
|
||||
let prepared_messages =
|
||||
multimodal::prepare_messages_for_provider(history, multimodal_config).await?;
|
||||
|
||||
// Recomputed each iteration: a `ToolSource` may register tools lazily
|
||||
// mid-turn, so native-tool enablement can flip from off to on.
|
||||
let request_tools = if provider.supports_native_tools() && !tools.request_specs().is_empty()
|
||||
{
|
||||
Some(tools.request_specs())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// ProviderDelta → progress forwarder for this iteration (no-op for
|
||||
// flavors that don't stream). Sender dropped after the chat call so the
|
||||
// forwarder exits cleanly.
|
||||
let (delta_tx_opt, delta_forwarder) = progress.make_stream_sink((iteration + 1) as u32);
|
||||
|
||||
let chat_result = provider
|
||||
.chat(
|
||||
ChatRequest {
|
||||
messages: &prepared_messages.messages,
|
||||
tools: request_tools,
|
||||
stream: delta_tx_opt.as_ref(),
|
||||
},
|
||||
model,
|
||||
temperature,
|
||||
)
|
||||
.await;
|
||||
|
||||
drop(delta_tx_opt);
|
||||
if let Some(handle) = delta_forwarder {
|
||||
let _ = handle.await;
|
||||
}
|
||||
|
||||
let (
|
||||
response_text,
|
||||
display_text,
|
||||
reasoning_content,
|
||||
tool_calls,
|
||||
assistant_history_content,
|
||||
native_tool_calls,
|
||||
) = match chat_result {
|
||||
Ok(resp) => {
|
||||
// Update context guard + cost with token usage from this response.
|
||||
if let Some(ref usage) = resp.usage {
|
||||
context_guard.update_usage(usage);
|
||||
turn_cost.add_call(model, usage);
|
||||
observer.record_usage(model, usage);
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
input_tokens = usage.input_tokens,
|
||||
output_tokens = usage.output_tokens,
|
||||
context_window = usage.context_window,
|
||||
cumulative_usd = turn_cost.total_usd(),
|
||||
"[agent_loop] LLM response received"
|
||||
);
|
||||
progress
|
||||
.cost_updated(model, (iteration + 1) as u32, &turn_cost)
|
||||
.await;
|
||||
} else {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
"[agent_loop] LLM response received (no usage info)"
|
||||
);
|
||||
}
|
||||
|
||||
let response_text = resp.text_or_empty().to_string();
|
||||
let (display_text, calls) = parser.parse(&resp);
|
||||
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
native_tool_calls = resp.tool_calls.len(),
|
||||
parsed_tool_calls = calls.len(),
|
||||
"[agent_loop] tool calls parsed"
|
||||
);
|
||||
|
||||
let assistant_history_content = if resp.tool_calls.is_empty() {
|
||||
response_text.clone()
|
||||
} else {
|
||||
build_native_assistant_history(
|
||||
&response_text,
|
||||
resp.reasoning_content.as_deref(),
|
||||
&resp.tool_calls,
|
||||
)
|
||||
};
|
||||
|
||||
let reasoning_content = resp.reasoning_content;
|
||||
let native_calls = resp.tool_calls;
|
||||
(
|
||||
response_text,
|
||||
display_text,
|
||||
reasoning_content,
|
||||
calls,
|
||||
assistant_history_content,
|
||||
native_calls,
|
||||
)
|
||||
}
|
||||
Err(e) => {
|
||||
// Transient upstream failures are already classified + retried by
|
||||
// reliable.rs and reported once when all providers are exhausted;
|
||||
// re-reporting per iteration floods Sentry (OPENHUMAN-TAURI-3Y/3Z).
|
||||
let transient =
|
||||
crate::openhuman::inference::provider::reliable::is_rate_limited(&e)
|
||||
|| crate::openhuman::inference::provider::reliable::is_upstream_unhealthy(
|
||||
&e,
|
||||
);
|
||||
if transient {
|
||||
tracing::warn!(
|
||||
domain = "agent",
|
||||
operation = "provider_chat",
|
||||
provider = provider_name,
|
||||
model = model,
|
||||
iteration = iteration + 1,
|
||||
error = %format!("{e:#}"),
|
||||
"[agent] transient provider_chat failure — retried upstream"
|
||||
);
|
||||
} else {
|
||||
crate::core::observability::report_error_or_expected(
|
||||
&e,
|
||||
"agent",
|
||||
"provider_chat",
|
||||
&[
|
||||
("provider", provider_name),
|
||||
("model", model),
|
||||
("iteration", &(iteration + 1).to_string()),
|
||||
],
|
||||
);
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
if tool_calls.is_empty() {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
"[agent_loop] no tool calls — returning final response"
|
||||
);
|
||||
// The final answer is the narrative text, falling back to the raw
|
||||
// response text when the parser stripped everything (mirrors the
|
||||
// legacy `Agent::turn` `final_text` logic).
|
||||
let final_out = if display_text.is_empty() {
|
||||
response_text.clone()
|
||||
} else {
|
||||
display_text.clone()
|
||||
};
|
||||
// A completion with no text *and* no tool calls is a degenerate
|
||||
// response. Callers that disallow it (Agent::turn) surface a typed
|
||||
// error instead of a silent blank reply; the channel/subagent loops
|
||||
// return it verbatim.
|
||||
if final_out.trim().is_empty() && !observer.allow_empty_final() {
|
||||
log::warn!(
|
||||
"[agent_loop] provider returned an empty final response (i={}, no text, no tool calls) — surfacing as error",
|
||||
iteration + 1
|
||||
);
|
||||
return Err(
|
||||
crate::openhuman::agent::error::AgentError::EmptyProviderResponse {
|
||||
iteration: iteration + 1,
|
||||
}
|
||||
.into(),
|
||||
);
|
||||
}
|
||||
// No tool calls — final response. Relay the text in small chunks
|
||||
// when a streaming draft sink exists.
|
||||
if let Some(ref tx) = on_delta {
|
||||
let mut chunk = String::new();
|
||||
for word in final_out.split_inclusive(char::is_whitespace) {
|
||||
chunk.push_str(word);
|
||||
if chunk.len() >= STREAM_CHUNK_MIN_CHARS
|
||||
&& tx.send(std::mem::take(&mut chunk)).await.is_err()
|
||||
{
|
||||
break; // receiver dropped
|
||||
}
|
||||
}
|
||||
if !chunk.is_empty() {
|
||||
let _ = tx.send(chunk).await;
|
||||
}
|
||||
}
|
||||
history.push(ChatMessage::assistant(response_text.clone()));
|
||||
observer.on_assistant(
|
||||
&final_out,
|
||||
&response_text,
|
||||
reasoning_content.as_deref(),
|
||||
&[],
|
||||
&[],
|
||||
iteration,
|
||||
true,
|
||||
);
|
||||
observer.after_iteration(history, iteration);
|
||||
log::info!(
|
||||
"[agent_loop] turn complete: iters={} provider_calls={} tokens_in={} tokens_out={} cached_in={} usd={:.4}",
|
||||
(iteration + 1),
|
||||
turn_cost.call_count,
|
||||
turn_cost.input_tokens,
|
||||
turn_cost.output_tokens,
|
||||
turn_cost.cached_input_tokens,
|
||||
turn_cost.total_usd(),
|
||||
);
|
||||
progress.turn_completed((iteration + 1) as u32).await;
|
||||
return Ok(TurnEngineOutcome {
|
||||
text: final_out,
|
||||
iterations: (iteration + 1) as u32,
|
||||
cost: turn_cost,
|
||||
hit_cap: false,
|
||||
});
|
||||
}
|
||||
|
||||
// Print any text the LLM produced alongside tool calls (unless silent)
|
||||
if !silent && !display_text.is_empty() {
|
||||
print!("{display_text}");
|
||||
let _ = std::io::stdout().flush();
|
||||
}
|
||||
|
||||
// Execute each tool call and build results. `individual_results` tracks
|
||||
// per-call output so native-mode history can emit one `role: tool`
|
||||
// message per call with the correct id.
|
||||
let mut tool_results = String::new();
|
||||
let mut individual_results: Vec<String> = Vec::new();
|
||||
for (call_idx, call) in tool_calls.iter().enumerate() {
|
||||
// Stable id threaded through the start/complete pair. The fallback
|
||||
// includes `call_idx` to stay unique when the same tool name
|
||||
// appears multiple times in one iteration.
|
||||
let progress_call_id = call
|
||||
.id
|
||||
.clone()
|
||||
.unwrap_or_else(|| format!("loop-{iteration}-{call_idx}-{}", call.name));
|
||||
|
||||
// Full per-call lifecycle is owned by the ToolSource.
|
||||
let outcome = tools
|
||||
.execute_call(call, iteration, progress, &progress_call_id)
|
||||
.await;
|
||||
|
||||
individual_results.push(outcome.text.clone());
|
||||
let _ = writeln!(
|
||||
tool_results,
|
||||
"<tool_result name=\"{}\">\n{}\n</tool_result>",
|
||||
call.name, outcome.text
|
||||
);
|
||||
|
||||
// Record this call in the run digest (output truncated) for a
|
||||
// possible max-iteration checkpoint.
|
||||
let _ = writeln!(
|
||||
run_tool_digest,
|
||||
"- {} [{}]: {}",
|
||||
call.name,
|
||||
if outcome.success { "ok" } else { "failed" },
|
||||
truncate_with_ellipsis(&outcome.text, 800)
|
||||
);
|
||||
|
||||
observer.on_tool_result(
|
||||
&progress_call_id,
|
||||
&call.name,
|
||||
&outcome.text,
|
||||
outcome.success,
|
||||
iteration,
|
||||
);
|
||||
|
||||
// Repeated-failure circuit breaker (shared guard).
|
||||
if let Some(reason) = failure_guard.record(
|
||||
&call.name,
|
||||
&call.arguments.to_string(),
|
||||
outcome.success,
|
||||
&outcome.text,
|
||||
) {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] circuit breaker tripped — halting with root cause"
|
||||
);
|
||||
halt_reason = Some(reason);
|
||||
}
|
||||
}
|
||||
|
||||
// Add assistant message with tool calls + tool results to history.
|
||||
// Native mode: JSON-structured messages so convert_messages() can
|
||||
// reconstruct OpenAI-format tool_calls + tool result messages. Prompt
|
||||
// mode: XML-based text format.
|
||||
history.push(ChatMessage::assistant(assistant_history_content));
|
||||
observer.on_assistant(
|
||||
&display_text,
|
||||
&response_text,
|
||||
reasoning_content.as_deref(),
|
||||
&native_tool_calls,
|
||||
&tool_calls,
|
||||
iteration,
|
||||
false,
|
||||
);
|
||||
if native_tool_calls.is_empty() {
|
||||
let content = format!("[Tool results]\n{tool_results}");
|
||||
observer.on_results_batch(&content, iteration);
|
||||
history.push(ChatMessage::user(content));
|
||||
} else {
|
||||
for (native_call, result) in native_tool_calls.iter().zip(individual_results.iter()) {
|
||||
let tool_msg = serde_json::json!({
|
||||
"tool_call_id": native_call.id,
|
||||
"content": result,
|
||||
});
|
||||
history.push(ChatMessage::tool(tool_msg.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
observer.after_iteration(history, iteration);
|
||||
|
||||
// Circuit breaker tripped this iteration: return the root-cause summary
|
||||
// instead of looping to `max_iterations`. Tool results are already in
|
||||
// `history`, so the caller still has full context.
|
||||
if let Some(reason) = halt_reason.take() {
|
||||
// Mirror the normal-completion path: emit turn-completed before the
|
||||
// early return so progress consumers don't stay in-flight.
|
||||
progress.turn_completed((iteration + 1) as u32).await;
|
||||
return Ok(TurnEngineOutcome {
|
||||
text: reason,
|
||||
iterations: (iteration + 1) as u32,
|
||||
cost: turn_cost,
|
||||
hit_cap: false,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Iteration cap reached — hand off to the checkpoint strategy (error vs
|
||||
// summarize). The accumulated digest lets a summarizing strategy produce a
|
||||
// resumable, root-cause-aware checkpoint.
|
||||
let digest = if run_tool_digest.is_empty() {
|
||||
"(no tool calls completed)"
|
||||
} else {
|
||||
run_tool_digest.as_str()
|
||||
};
|
||||
let co = checkpoint.on_max_iter(digest, max_iterations).await?;
|
||||
// Fold any summarization-call usage into the turn cost + observer so token
|
||||
// accounting stays complete.
|
||||
if let Some(ref u) = co.usage {
|
||||
turn_cost.add_call(model, u);
|
||||
observer.record_usage(model, u);
|
||||
}
|
||||
// Emit the terminal lifecycle event on this successful (checkpoint) exit
|
||||
// too, so consumers aren't left waiting — matching the final-response and
|
||||
// circuit-breaker paths.
|
||||
progress.turn_completed(max_iterations as u32).await;
|
||||
Ok(TurnEngineOutcome {
|
||||
text: co.text,
|
||||
iterations: max_iterations as u32,
|
||||
cost: turn_cost,
|
||||
hit_cap: true,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
//! Unified agent turn engine.
|
||||
//!
|
||||
//! Historically the harness carried THREE near-identical agentic loops — one
|
||||
//! per entry point (`Agent::turn` for web/desktop chat, `run_tool_call_loop`
|
||||
//! for non-web channels + triage, and the subagent `run_inner_loop`). They each
|
||||
//! re-implemented the same shape (call the LLM → parse tool calls → execute
|
||||
//! tools → append results → repeat until final text or the iteration cap) and
|
||||
//! had drifted in subtle ways.
|
||||
//!
|
||||
//! This module is the single home for the pieces those loops share, so they
|
||||
//! can't drift again. The extraction is incremental (see the unify-agent-turn
|
||||
//! plan): the first piece to land is [`tools::run_one_tool`] — the per-call
|
||||
//! tool executor (policy gate → scope guard → approval gate → execute with
|
||||
//! timeout → scrub/tokenjuice/cap/summarize → audit), which was previously
|
||||
//! duplicated verbatim across all three loops.
|
||||
|
||||
pub(crate) mod checkpoint;
|
||||
pub(crate) mod core;
|
||||
pub(crate) mod parser;
|
||||
pub(crate) mod progress;
|
||||
pub(crate) mod state;
|
||||
pub(crate) mod tool_source;
|
||||
pub(crate) mod tools;
|
||||
|
||||
pub(crate) use checkpoint::{CheckpointOutcome, CheckpointStrategy, ErrorCheckpoint};
|
||||
pub(crate) use core::run_turn_engine;
|
||||
pub(crate) use parser::{DefaultParser, DispatcherParser};
|
||||
pub(crate) use progress::{ProgressReporter, SubagentProgress, TurnProgress};
|
||||
pub(crate) use state::{NullObserver, TurnObserver};
|
||||
pub(crate) use tool_source::{RegistryToolSource, ToolSource};
|
||||
pub(crate) use tools::{run_one_tool, ToolRunResult};
|
||||
@@ -0,0 +1,70 @@
|
||||
//! Response-parsing seam.
|
||||
//!
|
||||
//! The channel loop and subagent extract tool calls from a provider response
|
||||
//! with the built-in native-first + XML-fallback logic ([`DefaultParser`]).
|
||||
//! `Agent::turn` instead uses its configured [`ToolDispatcher`] (native / XML /
|
||||
//! PFormat) — PFormat in particular parses positional `name[args]` calls the
|
||||
//! built-in path can't. [`DispatcherParser`] adapts a dispatcher to this seam so
|
||||
//! the engine stays parser-agnostic while preserving every dispatcher's grammar.
|
||||
//!
|
||||
//! `parse` returns `(display_text, calls)`: the narrative text to surface (tool
|
||||
//! markup stripped) and the parsed calls in the engine's internal
|
||||
//! [`ParsedToolCall`] shape. The engine keeps the *raw* response text
|
||||
//! separately for assistant-history serialization.
|
||||
|
||||
use crate::openhuman::agent::dispatcher::ToolDispatcher;
|
||||
use crate::openhuman::agent::harness::parse::{
|
||||
parse_structured_tool_calls, parse_tool_calls, ParsedToolCall,
|
||||
};
|
||||
use crate::openhuman::inference::provider::ChatResponse;
|
||||
|
||||
pub(crate) trait ResponseParser: Send + Sync {
|
||||
/// Returns `(display_text, calls)` for this provider response.
|
||||
fn parse(&self, resp: &ChatResponse) -> (String, Vec<ParsedToolCall>);
|
||||
}
|
||||
|
||||
/// Built-in parser: prefer native structured tool calls, fall back to the
|
||||
/// XML-tag parser over the response text. Used by the channel loop + subagent.
|
||||
pub(crate) struct DefaultParser;
|
||||
|
||||
impl ResponseParser for DefaultParser {
|
||||
fn parse(&self, resp: &ChatResponse) -> (String, Vec<ParsedToolCall>) {
|
||||
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();
|
||||
if calls.is_empty() {
|
||||
let (fallback_text, fallback_calls) = parse_tool_calls(&response_text);
|
||||
if !fallback_text.is_empty() {
|
||||
parsed_text = fallback_text;
|
||||
}
|
||||
calls = fallback_calls;
|
||||
}
|
||||
let display_text = if parsed_text.is_empty() {
|
||||
response_text
|
||||
} else {
|
||||
parsed_text
|
||||
};
|
||||
(display_text, calls)
|
||||
}
|
||||
}
|
||||
|
||||
/// Adapts an [`Agent`]'s configured [`ToolDispatcher`] to the parser seam,
|
||||
/// converting the dispatcher's `ParsedToolCall` shape into the engine's.
|
||||
pub(crate) struct DispatcherParser<'a> {
|
||||
pub dispatcher: &'a dyn ToolDispatcher,
|
||||
}
|
||||
|
||||
impl ResponseParser for DispatcherParser<'_> {
|
||||
fn parse(&self, resp: &ChatResponse) -> (String, Vec<ParsedToolCall>) {
|
||||
let (text, calls) = self.dispatcher.parse_response(resp);
|
||||
let calls = calls
|
||||
.into_iter()
|
||||
.map(|c| ParsedToolCall {
|
||||
name: c.name,
|
||||
arguments: c.arguments,
|
||||
id: c.tool_call_id,
|
||||
})
|
||||
.collect();
|
||||
(text, calls)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,356 @@
|
||||
//! Progress reporting seam + the shared streaming-delta forwarder.
|
||||
//!
|
||||
//! The engine never names a concrete [`AgentProgress`] variant. It talks to a
|
||||
//! [`ProgressReporter`], whose impls pick the event *flavor*:
|
||||
//!
|
||||
//! * [`TurnProgress`] — top-level chat (channel loop, `Agent::turn`): emits the
|
||||
//! `Turn*` / `ToolCall*` / `TurnCostUpdated` events and streams provider
|
||||
//! deltas as `TextDelta` / `ThinkingDelta` / `ToolCallArgsDelta`.
|
||||
//! * [`SubagentProgress`] — a spawned sub-agent: emits the `Subagent*` /
|
||||
//! `SubagentToolCall*` events (nested under the subagent row in the UI) and
|
||||
//! does not stream deltas. The `SubagentSpawned` / `SubagentCompleted` /
|
||||
//! `SubagentFailed` lifecycle events stay in the spawn tool, outside the loop.
|
||||
//! * [`NullProgress`] — triage / tests: every method is a no-op.
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::openhuman::agent::cost::TurnCost;
|
||||
use crate::openhuman::agent::progress::AgentProgress;
|
||||
use crate::openhuman::inference::provider::ProviderDelta;
|
||||
|
||||
/// What the engine emits as a turn progresses. All methods default to no-ops so
|
||||
/// an impl only overrides the events its flavor cares about.
|
||||
#[async_trait]
|
||||
pub(crate) trait ProgressReporter: Send + Sync {
|
||||
async fn turn_started(&self) {}
|
||||
async fn iteration_started(&self, _iteration: u32, _max_iterations: u32) {}
|
||||
async fn cost_updated(&self, _model: &str, _iteration: u32, _cost: &TurnCost) {}
|
||||
async fn turn_completed(&self, _iterations: u32) {}
|
||||
async fn tool_started(
|
||||
&self,
|
||||
_call_id: &str,
|
||||
_tool_name: &str,
|
||||
_arguments: &serde_json::Value,
|
||||
_iteration: u32,
|
||||
) {
|
||||
}
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn tool_completed(
|
||||
&self,
|
||||
_call_id: &str,
|
||||
_tool_name: &str,
|
||||
_success: bool,
|
||||
_output_chars: usize,
|
||||
_elapsed_ms: u64,
|
||||
_iteration: u32,
|
||||
) {
|
||||
}
|
||||
|
||||
/// Build the per-iteration `ProviderDelta` streaming sink + forwarder task,
|
||||
/// or `(None, None)` when this flavor doesn't stream. Default: no streaming.
|
||||
fn make_stream_sink(
|
||||
&self,
|
||||
_iteration: u32,
|
||||
) -> (
|
||||
Option<tokio::sync::mpsc::Sender<ProviderDelta>>,
|
||||
Option<tokio::task::JoinHandle<()>>,
|
||||
) {
|
||||
(None, None)
|
||||
}
|
||||
}
|
||||
|
||||
/// Top-level chat flavor: `Turn*` lifecycle + `ToolCall*` + streaming.
|
||||
pub(crate) struct TurnProgress {
|
||||
pub sink: Option<tokio::sync::mpsc::Sender<AgentProgress>>,
|
||||
}
|
||||
|
||||
impl TurnProgress {
|
||||
pub(crate) fn new(sink: Option<tokio::sync::mpsc::Sender<AgentProgress>>) -> Self {
|
||||
Self { sink }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ProgressReporter for TurnProgress {
|
||||
async fn turn_started(&self) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
if let Err(e) = sink.send(AgentProgress::TurnStarted).await {
|
||||
log::warn!("[agent_loop] progress sink closed at TurnStarted: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn iteration_started(&self, iteration: u32, max_iterations: u32) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::IterationStarted {
|
||||
iteration,
|
||||
max_iterations,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!("[agent_loop] progress sink closed at IterationStarted: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn cost_updated(&self, model: &str, iteration: u32, cost: &TurnCost) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
let event = AgentProgress::TurnCostUpdated {
|
||||
model: model.to_string(),
|
||||
iteration,
|
||||
input_tokens: cost.input_tokens,
|
||||
output_tokens: cost.output_tokens,
|
||||
cached_input_tokens: cost.cached_input_tokens,
|
||||
total_usd: cost.total_usd(),
|
||||
};
|
||||
if let Err(e) = sink.send(event).await {
|
||||
log::warn!("[agent_loop] progress sink closed at TurnCostUpdated: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn turn_completed(&self, iterations: u32) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
if let Err(e) = sink.send(AgentProgress::TurnCompleted { iterations }).await {
|
||||
log::warn!("[agent_loop] progress sink closed at TurnCompleted: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn tool_started(
|
||||
&self,
|
||||
call_id: &str,
|
||||
tool_name: &str,
|
||||
arguments: &serde_json::Value,
|
||||
iteration: u32,
|
||||
) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::ToolCallStarted {
|
||||
call_id: call_id.to_string(),
|
||||
tool_name: tool_name.to_string(),
|
||||
arguments: arguments.clone(),
|
||||
iteration,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!("[agent_loop] progress sink closed while emitting ToolCallStarted: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn tool_completed(
|
||||
&self,
|
||||
call_id: &str,
|
||||
tool_name: &str,
|
||||
success: bool,
|
||||
output_chars: usize,
|
||||
elapsed_ms: u64,
|
||||
iteration: u32,
|
||||
) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::ToolCallCompleted {
|
||||
call_id: call_id.to_string(),
|
||||
tool_name: tool_name.to_string(),
|
||||
success,
|
||||
output_chars,
|
||||
elapsed_ms,
|
||||
iteration,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!(
|
||||
"[agent_loop] progress sink closed while emitting ToolCallCompleted: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn make_stream_sink(
|
||||
&self,
|
||||
iteration: u32,
|
||||
) -> (
|
||||
Option<tokio::sync::mpsc::Sender<ProviderDelta>>,
|
||||
Option<tokio::task::JoinHandle<()>>,
|
||||
) {
|
||||
spawn_delta_forwarder(self.sink.clone(), iteration)
|
||||
}
|
||||
}
|
||||
|
||||
/// Sub-agent flavor: `Subagent*` lifecycle + `SubagentToolCall*`, no streaming.
|
||||
pub(crate) struct SubagentProgress {
|
||||
pub sink: Option<tokio::sync::mpsc::Sender<AgentProgress>>,
|
||||
pub agent_id: String,
|
||||
pub task_id: String,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ProgressReporter for SubagentProgress {
|
||||
async fn iteration_started(&self, iteration: u32, max_iterations: u32) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
let _ = sink
|
||||
.send(AgentProgress::SubagentIterationStarted {
|
||||
agent_id: self.agent_id.clone(),
|
||||
task_id: self.task_id.clone(),
|
||||
iteration,
|
||||
max_iterations,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn tool_started(
|
||||
&self,
|
||||
call_id: &str,
|
||||
tool_name: &str,
|
||||
_arguments: &serde_json::Value,
|
||||
iteration: u32,
|
||||
) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
let _ = sink
|
||||
.send(AgentProgress::SubagentToolCallStarted {
|
||||
agent_id: self.agent_id.clone(),
|
||||
task_id: self.task_id.clone(),
|
||||
call_id: call_id.to_string(),
|
||||
tool_name: tool_name.to_string(),
|
||||
iteration,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn tool_completed(
|
||||
&self,
|
||||
call_id: &str,
|
||||
tool_name: &str,
|
||||
success: bool,
|
||||
output_chars: usize,
|
||||
elapsed_ms: u64,
|
||||
iteration: u32,
|
||||
) {
|
||||
if let Some(ref sink) = self.sink {
|
||||
let _ = sink
|
||||
.send(AgentProgress::SubagentToolCallCompleted {
|
||||
agent_id: self.agent_id.clone(),
|
||||
task_id: self.task_id.clone(),
|
||||
call_id: call_id.to_string(),
|
||||
tool_name: tool_name.to_string(),
|
||||
success,
|
||||
output_chars,
|
||||
elapsed_ms,
|
||||
iteration,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Stream the child's visible text + reasoning deltas to the parent,
|
||||
/// attributed to this sub-agent's `task_id` so the UI renders them inside
|
||||
/// the live subagent row (PR #3007). Tool-call arg fragments are dropped
|
||||
/// here — they're already surfaced via the `SubagentToolCall*` lifecycle
|
||||
/// events, so forwarding them too would double-render.
|
||||
fn make_stream_sink(
|
||||
&self,
|
||||
iteration: u32,
|
||||
) -> (
|
||||
Option<tokio::sync::mpsc::Sender<ProviderDelta>>,
|
||||
Option<tokio::task::JoinHandle<()>>,
|
||||
) {
|
||||
let Some(sink) = self.sink.clone() else {
|
||||
return (None, None);
|
||||
};
|
||||
let agent_id = self.agent_id.clone();
|
||||
let task_id = self.task_id.clone();
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<ProviderDelta>(128);
|
||||
let forwarder = tokio::spawn(async move {
|
||||
while let Some(event) = rx.recv().await {
|
||||
let mapped = match event {
|
||||
ProviderDelta::TextDelta { delta } => AgentProgress::SubagentTextDelta {
|
||||
agent_id: agent_id.clone(),
|
||||
task_id: task_id.clone(),
|
||||
delta,
|
||||
iteration,
|
||||
},
|
||||
ProviderDelta::ThinkingDelta { delta } => {
|
||||
AgentProgress::SubagentThinkingDelta {
|
||||
agent_id: agent_id.clone(),
|
||||
task_id: task_id.clone(),
|
||||
delta,
|
||||
iteration,
|
||||
}
|
||||
}
|
||||
ProviderDelta::ToolCallStart { .. }
|
||||
| ProviderDelta::ToolCallArgsDelta { .. } => continue,
|
||||
};
|
||||
// Await backpressure so streamed deltas arrive in order.
|
||||
if sink.send(mapped).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
(Some(tx), Some(forwarder))
|
||||
}
|
||||
}
|
||||
|
||||
/// No-op reporter for triage / tests.
|
||||
pub(crate) struct NullProgress;
|
||||
|
||||
impl ProgressReporter for NullProgress {}
|
||||
|
||||
/// Spawn a task that forwards `ProviderDelta`s from the provider's streaming
|
||||
/// channel into `on_progress` as `AgentProgress` delta events, tagged with
|
||||
/// `iteration` (1-based). Returns the sender to hand to
|
||||
/// [`crate::openhuman::inference::provider::ChatRequest::stream`] and the task
|
||||
/// handle to await after the chat call.
|
||||
///
|
||||
/// Returns `(None, None)` when there is no progress sink — the caller then
|
||||
/// passes `stream: None` and the provider uses its non-streaming HTTP path.
|
||||
///
|
||||
/// Backpressure discipline: the forwarder `.await`s each `send`, so streamed
|
||||
/// deltas arrive in order and are never silently dropped when the downstream
|
||||
/// bridge is slow. It exits cleanly once the sender is dropped (after the chat
|
||||
/// call) or the downstream closes.
|
||||
pub(crate) fn spawn_delta_forwarder(
|
||||
on_progress: Option<tokio::sync::mpsc::Sender<AgentProgress>>,
|
||||
iteration: u32,
|
||||
) -> (
|
||||
Option<tokio::sync::mpsc::Sender<ProviderDelta>>,
|
||||
Option<tokio::task::JoinHandle<()>>,
|
||||
) {
|
||||
let Some(progress_sink) = on_progress else {
|
||||
return (None, None);
|
||||
};
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<ProviderDelta>(128);
|
||||
let forwarder = tokio::spawn(async move {
|
||||
while let Some(event) = rx.recv().await {
|
||||
let mapped = match event {
|
||||
ProviderDelta::TextDelta { delta } => AgentProgress::TextDelta { delta, iteration },
|
||||
ProviderDelta::ThinkingDelta { delta } => {
|
||||
AgentProgress::ThinkingDelta { delta, iteration }
|
||||
}
|
||||
ProviderDelta::ToolCallStart { call_id, tool_name } => {
|
||||
AgentProgress::ToolCallArgsDelta {
|
||||
call_id,
|
||||
tool_name,
|
||||
delta: String::new(),
|
||||
iteration,
|
||||
}
|
||||
}
|
||||
ProviderDelta::ToolCallArgsDelta { call_id, delta } => {
|
||||
AgentProgress::ToolCallArgsDelta {
|
||||
call_id,
|
||||
tool_name: String::new(),
|
||||
delta,
|
||||
iteration,
|
||||
}
|
||||
}
|
||||
};
|
||||
if progress_sink.send(mapped).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
(Some(tx), Some(forwarder))
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
//! Turn-state observer seam.
|
||||
//!
|
||||
//! The engine drives the loop over a `Vec<ChatMessage>` working buffer, but the
|
||||
//! three callers want to *do* different things around each step:
|
||||
//!
|
||||
//! * the channel loop wants nothing extra ([`NullObserver`]);
|
||||
//! * the subagent wants per-iteration transcript persistence, usage
|
||||
//! accumulation, and worker-thread mirroring (assistant intents, per-call
|
||||
//! results, batched text-mode results, final response);
|
||||
//! * `Agent::turn` wants its `ContextManager` reduction before each dispatch,
|
||||
//! transcript persistence, and per-turn usage/cost snapshots.
|
||||
//!
|
||||
//! [`TurnObserver`] is the seam: every method defaults to a no-op, so an impl
|
||||
//! only overrides the hooks its caller needs. The engine still owns the
|
||||
//! universal concerns (stop hooks, context guard, token-budget trim, the
|
||||
//! circuit breaker) inline — the observer is for caller-specific side effects.
|
||||
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::openhuman::agent::harness::parse::ParsedToolCall;
|
||||
use crate::openhuman::inference::provider::{ChatMessage, ToolCall, UsageInfo};
|
||||
|
||||
#[async_trait]
|
||||
pub(crate) trait TurnObserver: Send {
|
||||
/// Called before each provider dispatch, after the engine's own context
|
||||
/// guard + token-budget trim. `Agent::turn` runs its `ContextManager`
|
||||
/// reduction chain here. Default: no-op.
|
||||
async fn before_dispatch(
|
||||
&mut self,
|
||||
_history: &mut Vec<ChatMessage>,
|
||||
_iteration: usize,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Called once per provider response that carried a usage block, so the
|
||||
/// caller can accumulate its own token tally / transcript usage snapshot.
|
||||
fn record_usage(&mut self, _model: &str, _usage: &UsageInfo) {}
|
||||
|
||||
/// Called after the assistant message for this iteration is committed to
|
||||
/// the engine's working buffer. `response_text` is the raw provider text
|
||||
/// (pre native serialization); `reasoning_content` is the thinking-model
|
||||
/// content to round-trip; `native_tool_calls` are the provider's structured
|
||||
/// calls (empty in text/prompt mode); `parsed_calls` are the engine-parsed
|
||||
/// calls (empty when `is_final`). `Agent::turn` uses these to rebuild its
|
||||
/// typed `ConversationMessage` history; the subagent mirrors to its worker
|
||||
/// thread.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn on_assistant(
|
||||
&mut self,
|
||||
_display_text: &str,
|
||||
_response_text: &str,
|
||||
_reasoning_content: Option<&str>,
|
||||
_native_tool_calls: &[ToolCall],
|
||||
_parsed_calls: &[ParsedToolCall],
|
||||
_iteration: usize,
|
||||
_is_final: bool,
|
||||
) {
|
||||
}
|
||||
|
||||
/// Called after one tool's result is known, in native-tool mode (one
|
||||
/// `role:tool` message per call). Subagent mirrors per-call results to its
|
||||
/// worker thread; `Agent::turn` buffers them to rebuild typed history.
|
||||
fn on_tool_result(
|
||||
&mut self,
|
||||
_call_id: &str,
|
||||
_tool_name: &str,
|
||||
_result_text: &str,
|
||||
_success: bool,
|
||||
_iteration: usize,
|
||||
) {
|
||||
}
|
||||
|
||||
/// Called after a batched `[Tool results]` user message is committed
|
||||
/// (text/prompt mode, where there are no per-call `role:tool` messages).
|
||||
fn on_results_batch(&mut self, _content: &str, _iteration: usize) {}
|
||||
|
||||
/// Called after the iteration's history is finalized (the transcript
|
||||
/// persistence point) — both after the final response and after each tool
|
||||
/// round's results are appended.
|
||||
fn after_iteration(&mut self, _history: &[ChatMessage], _iteration: usize) {}
|
||||
|
||||
/// Whether an empty final response (no text, no tool calls) is acceptable.
|
||||
/// The channel/subagent loops return it as `Ok("")`; `Agent::turn` treats
|
||||
/// it as a degenerate/poisoned completion and surfaces an error instead of
|
||||
/// a silent blank reply (bug-report-2026-05-26 A1). Default: allowed.
|
||||
fn allow_empty_final(&self) -> bool {
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
/// No-op observer for the channel/CLI/triage loop, which keeps no extra state.
|
||||
pub(crate) struct NullObserver;
|
||||
|
||||
impl TurnObserver for NullObserver {}
|
||||
@@ -0,0 +1,131 @@
|
||||
//! Tool sourcing seam for the turn engine.
|
||||
//!
|
||||
//! The three former loops resolved "what tools can the model call this turn and
|
||||
//! how do I execute one" differently:
|
||||
//!
|
||||
//! * the channel loop advertised `registry + extra` filtered by a visibility
|
||||
//! whitelist, and executed via the shared [`run_one_tool`];
|
||||
//! * the subagent loop advertised a definition-filtered slice of the parent's
|
||||
//! tools (with lazy toolkit registration), and had its own per-call body;
|
||||
//! * `Agent::turn` advertised `Agent.visible_tool_specs` and executed via the
|
||||
//! richer `Agent::execute_tool_call` (session policy + per-call permission
|
||||
//! levels + `execute_with_options`).
|
||||
//!
|
||||
//! [`ToolSource`] is the single seam the engine talks to: it advertises the
|
||||
//! request specs and owns per-call execution (including the start/complete
|
||||
//! progress events). [`RegistryToolSource`] is the channel/CLI/triage impl; the
|
||||
//! subagent and `Agent` impls land in later phases.
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use super::super::payload_summarizer::PayloadSummarizer;
|
||||
use super::progress::ProgressReporter;
|
||||
use super::{run_one_tool, ToolRunResult};
|
||||
use crate::openhuman::agent::harness::parse::ParsedToolCall;
|
||||
use crate::openhuman::tools::policy::ToolPolicy;
|
||||
use crate::openhuman::tools::{Tool, ToolSpec};
|
||||
|
||||
/// What the engine needs from "the set of tools available this turn".
|
||||
#[async_trait]
|
||||
pub(crate) trait ToolSource: Send {
|
||||
/// The deduped, visibility-filtered specs to advertise to the provider
|
||||
/// this turn. Re-read each iteration so impls that register tools lazily
|
||||
/// (subagent toolkit resolution) can grow the advertised set over a turn.
|
||||
fn request_specs(&self) -> &[ToolSpec];
|
||||
|
||||
/// Execute one parsed tool call end-to-end, emitting its `ToolCallStarted`
|
||||
/// / `ToolCallCompleted` (or flavor-equivalent) progress events. Returns a
|
||||
/// [`ToolRunResult`] the engine folds into history + the circuit breaker.
|
||||
async fn execute_call(
|
||||
&mut self,
|
||||
call: &ParsedToolCall,
|
||||
iteration: usize,
|
||||
progress: &dyn ProgressReporter,
|
||||
progress_call_id: &str,
|
||||
) -> ToolRunResult;
|
||||
}
|
||||
|
||||
/// The channel/CLI/triage tool source: a persistent `registry`, optional
|
||||
/// per-turn synthesised `extra` tools, an optional visibility whitelist, and a
|
||||
/// pluggable [`ToolPolicy`]. Mirrors the original `run_tool_call_loop` tool
|
||||
/// plumbing exactly.
|
||||
pub(crate) struct RegistryToolSource<'a> {
|
||||
registry: &'a [Box<dyn Tool>],
|
||||
extra: &'a [Box<dyn Tool>],
|
||||
visible: Option<&'a HashSet<String>>,
|
||||
tool_policy: &'a dyn ToolPolicy,
|
||||
payload_summarizer: Option<&'a dyn PayloadSummarizer>,
|
||||
specs: Vec<ToolSpec>,
|
||||
}
|
||||
|
||||
impl<'a> RegistryToolSource<'a> {
|
||||
pub(crate) fn new(
|
||||
registry: &'a [Box<dyn Tool>],
|
||||
extra: &'a [Box<dyn Tool>],
|
||||
visible: Option<&'a HashSet<String>>,
|
||||
tool_policy: &'a dyn ToolPolicy,
|
||||
payload_summarizer: Option<&'a dyn PayloadSummarizer>,
|
||||
) -> Self {
|
||||
// Filter to visible tools, then dedup by name before sending to the
|
||||
// provider. Registry tools may collide with per-turn synthesised
|
||||
// extra_tools (e.g. an `ArchetypeDelegationTool` whose
|
||||
// `delegate_name = "research"` shadowing a same-named skill). Some
|
||||
// providers 400 on duplicate tool names — see TAURI-RUST-4.
|
||||
let filtered: Vec<ToolSpec> = registry
|
||||
.iter()
|
||||
.chain(extra.iter())
|
||||
.filter(|tool| visible.map(|s| s.contains(tool.name())).unwrap_or(true))
|
||||
.map(|tool| tool.spec())
|
||||
.collect();
|
||||
let specs = crate::openhuman::agent::harness::session::dedup_visible_tool_specs(filtered);
|
||||
Self {
|
||||
registry,
|
||||
extra,
|
||||
visible,
|
||||
tool_policy,
|
||||
payload_summarizer,
|
||||
specs,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_visible(&self, name: &str) -> bool {
|
||||
self.visible.map(|s| s.contains(name)).unwrap_or(true)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ToolSource for RegistryToolSource<'_> {
|
||||
fn request_specs(&self) -> &[ToolSpec] {
|
||||
&self.specs
|
||||
}
|
||||
|
||||
async fn execute_call(
|
||||
&mut self,
|
||||
call: &ParsedToolCall,
|
||||
iteration: usize,
|
||||
progress: &dyn ProgressReporter,
|
||||
progress_call_id: &str,
|
||||
) -> ToolRunResult {
|
||||
// Look up the tool by name in the combined registry + extras, subject
|
||||
// to the visibility whitelist. A hallucinated / filtered-out name
|
||||
// resolves to `None`, which `run_one_tool` reports as an unknown tool.
|
||||
let tool_opt: Option<&dyn Tool> = self
|
||||
.registry
|
||||
.iter()
|
||||
.chain(self.extra.iter())
|
||||
.find(|t| t.name() == call.name && self.is_visible(t.name()))
|
||||
.map(|b| b.as_ref());
|
||||
run_one_tool(
|
||||
tool_opt,
|
||||
call,
|
||||
iteration,
|
||||
progress,
|
||||
self.tool_policy,
|
||||
self.payload_summarizer,
|
||||
progress_call_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,381 @@
|
||||
//! Shared per-call tool executor.
|
||||
//!
|
||||
//! [`run_one_tool`] runs the full lifecycle of a single parsed tool call:
|
||||
//!
|
||||
//! 1. emit `ToolCallStarted` (for *every* call, including ones rejected below,
|
||||
//! so a client row created from streamed args always gets a terminal event);
|
||||
//! 2. evaluate the pluggable [`ToolPolicy`] (deny short-circuits everything,
|
||||
//! including approval side-effects);
|
||||
//! 3. guard `CliRpcOnly` scope (such tools can't run in the autonomous loop);
|
||||
//! 4. route external-effect tools through the process-global `ApprovalGate`;
|
||||
//! 5. execute with the configured timeout, then scrub credentials, apply
|
||||
//! tokenjuice, the per-tool size cap, and the optional payload summarizer;
|
||||
//! 6. stamp the approval audit "after" row (#2135);
|
||||
//! 7. emit `ToolCallCompleted`.
|
||||
//!
|
||||
//! It returns a [`ToolRunResult`] (`text` + `success`). The caller owns history
|
||||
//! shaping (native `role:tool` messages vs XML `<tool_result>` blocks) and the
|
||||
//! repeated-failure circuit breaker, both of which it drives uniformly from the
|
||||
//! returned `success`/`text` regardless of which branch produced them.
|
||||
//!
|
||||
//! This body was lifted verbatim (behavior-preserving) from the canonical
|
||||
//! `run_tool_call_loop` in `tool_loop.rs`; the three loops now call it instead
|
||||
//! of each carrying their own copy.
|
||||
|
||||
use super::super::payload_summarizer::PayloadSummarizer;
|
||||
use super::progress::ProgressReporter;
|
||||
use crate::openhuman::agent::harness::parse::ParsedToolCall;
|
||||
use crate::openhuman::tools::policy::{PolicyDecision, ToolPolicy};
|
||||
use crate::openhuman::tools::traits::ToolScope;
|
||||
use crate::openhuman::tools::Tool;
|
||||
|
||||
use super::super::credentials::scrub_credentials;
|
||||
|
||||
/// Outcome of a single tool call. `text` is what should be fed back to the
|
||||
/// model (a result body, an error, or a denial reason); `success` is `false`
|
||||
/// for any non-OK outcome (policy/approval denial, scope rejection, timeout,
|
||||
/// tool error, unknown tool) so the caller's circuit breaker and history
|
||||
/// formatting can treat every failure mode uniformly.
|
||||
pub(crate) struct ToolRunResult {
|
||||
pub text: String,
|
||||
pub success: bool,
|
||||
}
|
||||
|
||||
/// Execute one parsed tool call end-to-end. See the module docs for the full
|
||||
/// lifecycle. `tool_opt` is the (already visibility-filtered) tool the caller
|
||||
/// resolved by name — `None` means the model requested an unknown/filtered-out
|
||||
/// tool, which is reported as a structured error the LLM can correct next turn.
|
||||
///
|
||||
/// `progress_call_id` is the stable id threaded through the start/complete
|
||||
/// event pair (and any preceding args-delta events) so consumers can reconcile
|
||||
/// tool rows by id.
|
||||
pub(crate) async fn run_one_tool(
|
||||
tool_opt: Option<&dyn Tool>,
|
||||
call: &ParsedToolCall,
|
||||
iteration: usize,
|
||||
progress: &dyn ProgressReporter,
|
||||
tool_policy: &dyn ToolPolicy,
|
||||
payload_summarizer: Option<&dyn PayloadSummarizer>,
|
||||
progress_call_id: &str,
|
||||
) -> ToolRunResult {
|
||||
let iteration_u32 = (iteration + 1) as u32;
|
||||
|
||||
// Emit a "tool started" event for every parsed call, even ones that will be
|
||||
// rejected below (approval denied, CliRpcOnly, unknown) — the client-side
|
||||
// row was created from the streamed args and needs a terminal event.
|
||||
progress
|
||||
.tool_started(progress_call_id, &call.name, &call.arguments, iteration_u32)
|
||||
.await;
|
||||
|
||||
// Helper: emit a failed "tool completed" event for an early-exit path
|
||||
// (denied / CliRpcOnly / unknown) so the client row flips to `error`
|
||||
// instead of staying running.
|
||||
let emit_failed_completion = |message: &str| {
|
||||
let output_chars = message.chars().count();
|
||||
async move {
|
||||
progress
|
||||
.tool_completed(
|
||||
progress_call_id,
|
||||
&call.name,
|
||||
false,
|
||||
output_chars,
|
||||
0,
|
||||
iteration_u32,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
};
|
||||
|
||||
// ── Tool policy check (#2131) ─────────────────
|
||||
// Evaluate the pluggable ToolPolicy before any approval or execution. If
|
||||
// the policy denies the call, skip everything (including approval
|
||||
// side-effects) and return the denial reason as a tool error to the model.
|
||||
if let PolicyDecision::Deny(reason) = tool_policy.evaluate(&call.name, &call.arguments) {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
reason = %reason,
|
||||
"[agent_loop] tool policy denied tool call"
|
||||
);
|
||||
let denied = format!("Tool '{}' denied by policy: {reason}", call.name);
|
||||
emit_failed_completion(&denied).await;
|
||||
return ToolRunResult {
|
||||
text: denied,
|
||||
success: false,
|
||||
};
|
||||
}
|
||||
|
||||
let Some(tool) = tool_opt else {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] unknown tool requested"
|
||||
);
|
||||
let msg = format!("Unknown tool: {}", call.name);
|
||||
emit_failed_completion(&msg).await;
|
||||
return ToolRunResult {
|
||||
text: msg,
|
||||
success: false,
|
||||
};
|
||||
};
|
||||
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
found = true,
|
||||
"[agent_loop] executing tool"
|
||||
);
|
||||
|
||||
// Scope check: CliRpcOnly tools cannot run in the autonomous agent loop.
|
||||
if tool.scope() == ToolScope::CliRpcOnly {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] tool scope is CliRpcOnly — denied in agent loop"
|
||||
);
|
||||
let denied = format!(
|
||||
"Tool '{}' is only available via explicit CLI/RPC invocation, not in the autonomous agent loop.",
|
||||
call.name
|
||||
);
|
||||
emit_failed_completion(&denied).await;
|
||||
return ToolRunResult {
|
||||
text: denied,
|
||||
success: false,
|
||||
};
|
||||
}
|
||||
|
||||
// ── External-effect approval gate (#1339, #2135) ──
|
||||
// Tools whose `external_effect()` returns true route through the
|
||||
// process-global `ApprovalGate` so the UI can prompt the user before
|
||||
// `execute()` runs. The gate is `None` when supervised mode is disabled or
|
||||
// in test envs — behavior matches the pre-#1339 path.
|
||||
//
|
||||
// `approval_request_id` carries the persisted row id forward so we can
|
||||
// stamp the terminal execution outcome onto the same `pending_approvals`
|
||||
// row after the tool finishes (issue #2135). `None` means the tool was
|
||||
// either not gated, was session-allowlist-shortcutted, or was denied —
|
||||
// none of which produce an audit row that needs an "after" entry.
|
||||
let mut approval_request_id: Option<String> = None;
|
||||
let mut approval_gate_for_audit: Option<
|
||||
std::sync::Arc<crate::openhuman::approval::ApprovalGate>,
|
||||
> = None;
|
||||
if tool.external_effect_with_args(&call.arguments) {
|
||||
if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() {
|
||||
let summary = crate::openhuman::approval::summarize_action(&call.name, &call.arguments);
|
||||
let redacted = crate::openhuman::approval::redact_args(&call.arguments);
|
||||
let (outcome, request_id) =
|
||||
gate.intercept_audited(&call.name, &summary, redacted).await;
|
||||
match outcome {
|
||||
crate::openhuman::approval::GateOutcome::Allow => {
|
||||
approval_request_id = request_id;
|
||||
if approval_request_id.is_some() {
|
||||
approval_gate_for_audit = Some(gate);
|
||||
}
|
||||
}
|
||||
crate::openhuman::approval::GateOutcome::Deny { reason } => {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
reason = %reason,
|
||||
"[agent_loop] approval gate denied tool call"
|
||||
);
|
||||
emit_failed_completion(&reason).await;
|
||||
return ToolRunResult {
|
||||
text: reason,
|
||||
success: false,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let tool_deadline = crate::openhuman::tool_timeout::tool_execution_timeout_duration();
|
||||
let timeout_secs = crate::openhuman::tool_timeout::tool_execution_timeout_secs();
|
||||
let tool_started = std::time::Instant::now();
|
||||
let outcome = tokio::time::timeout(tool_deadline, tool.execute(call.arguments.clone())).await;
|
||||
let elapsed_ms = tool_started.elapsed().as_millis() as u64;
|
||||
let (result_text, success) = match outcome {
|
||||
Ok(Ok(r)) => {
|
||||
let output = r.output();
|
||||
let success = !r.is_error;
|
||||
if success {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
output_len = output.len(),
|
||||
"[agent_loop] tool succeeded"
|
||||
);
|
||||
let mut scrubbed = scrub_credentials(&output);
|
||||
let (compacted, tj_stats) = crate::openhuman::tokenjuice::compact_tool_output(
|
||||
&call.name,
|
||||
Some(&call.arguments),
|
||||
&scrubbed,
|
||||
Some(0),
|
||||
);
|
||||
if tj_stats.applied {
|
||||
log::debug!(
|
||||
"[agent_loop] tokenjuice applied tool={} rule={} {}->{} bytes",
|
||||
call.name,
|
||||
tj_stats.rule_id,
|
||||
tj_stats.original_bytes,
|
||||
tj_stats.compacted_bytes
|
||||
);
|
||||
scrubbed = compacted;
|
||||
}
|
||||
|
||||
// Per-tool max_result_size_chars cap. When a tool sets it and
|
||||
// the (post-tokenjuice) body still exceeds the cap, truncate
|
||||
// here and skip the global payload summarizer for this call —
|
||||
// the cap is fast and deterministic, the summarizer is the
|
||||
// fallback for tools that don't know their own size budget.
|
||||
let mut hit_per_tool_cap = false;
|
||||
if let Some(cap) = tool.max_result_size_chars() {
|
||||
let char_count = scrubbed.chars().count();
|
||||
if char_count > cap {
|
||||
let truncated: String = scrubbed.chars().take(cap).collect();
|
||||
let dropped = char_count - cap;
|
||||
log::info!(
|
||||
"[agent_loop] per-tool cap applied tool={} cap_chars={} original_chars={} dropped_chars={}",
|
||||
call.name,
|
||||
cap,
|
||||
char_count,
|
||||
dropped,
|
||||
);
|
||||
scrubbed = format!(
|
||||
"{truncated}\n\n[truncated by tool cap: {dropped} more chars not shown]"
|
||||
);
|
||||
hit_per_tool_cap = true;
|
||||
}
|
||||
}
|
||||
|
||||
if !hit_per_tool_cap {
|
||||
if let Some(summarizer) = payload_summarizer {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer intercepting tool={} bytes={}",
|
||||
call.name,
|
||||
scrubbed.len()
|
||||
);
|
||||
match summarizer
|
||||
.maybe_summarize(&call.name, None, &scrubbed)
|
||||
.await
|
||||
{
|
||||
Ok(Some(payload)) => {
|
||||
log::info!(
|
||||
"[agent_loop] payload_summarizer compressed tool={} {}->{} bytes",
|
||||
call.name,
|
||||
payload.original_bytes,
|
||||
payload.summary_bytes
|
||||
);
|
||||
scrubbed = payload.summary;
|
||||
}
|
||||
Ok(None) => {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer pass-through tool={} bytes={}",
|
||||
call.name,
|
||||
scrubbed.len()
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)",
|
||||
call.name,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
(scrubbed, true)
|
||||
} else {
|
||||
// Scrub before logging — a failing tool payload can carry
|
||||
// credentials / PII, so never log the raw output.
|
||||
let scrubbed = scrub_credentials(&output);
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] tool returned error: {scrubbed}"
|
||||
);
|
||||
let (compacted, _) = crate::openhuman::tokenjuice::compact_tool_output(
|
||||
&call.name,
|
||||
Some(&call.arguments),
|
||||
&scrubbed,
|
||||
Some(1),
|
||||
);
|
||||
(format!("Error: {compacted}"), false)
|
||||
}
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
crate::core::observability::report_error(
|
||||
&e,
|
||||
"tool",
|
||||
"execute",
|
||||
&[
|
||||
("tool", call.name.as_str()),
|
||||
("outcome", "failed"),
|
||||
("iteration", &(iteration + 1).to_string()),
|
||||
],
|
||||
);
|
||||
(format!("Error executing {}: {e}", call.name), false)
|
||||
}
|
||||
Err(_) => {
|
||||
let msg = format!(
|
||||
"tool '{}' timed out after {} seconds",
|
||||
call.name, timeout_secs
|
||||
);
|
||||
crate::core::observability::report_error(
|
||||
msg.as_str(),
|
||||
"tool",
|
||||
"execute",
|
||||
&[
|
||||
("tool", call.name.as_str()),
|
||||
("outcome", "timeout"),
|
||||
("timeout_secs", &timeout_secs.to_string()),
|
||||
("iteration", &(iteration + 1).to_string()),
|
||||
],
|
||||
);
|
||||
(
|
||||
format!(
|
||||
"Error: tool '{}' timed out after {} seconds",
|
||||
call.name, timeout_secs
|
||||
),
|
||||
false,
|
||||
)
|
||||
}
|
||||
};
|
||||
progress
|
||||
.tool_completed(
|
||||
progress_call_id,
|
||||
&call.name,
|
||||
success,
|
||||
result_text.chars().count(),
|
||||
elapsed_ms,
|
||||
iteration_u32,
|
||||
)
|
||||
.await;
|
||||
// ── Approval audit after-action row (#2135) ────
|
||||
// Stamp the terminal status onto the same `pending_approvals` row the gate
|
||||
// created before execution, so the audit trail carries both the before
|
||||
// (approval) and after (executed_at + outcome). Best-effort: a write
|
||||
// failure here is logged but not propagated to the agent.
|
||||
if let (Some(gate), Some(req_id)) = (
|
||||
approval_gate_for_audit.as_ref(),
|
||||
approval_request_id.as_ref(),
|
||||
) {
|
||||
let exec_outcome = if success {
|
||||
crate::openhuman::approval::ExecutionOutcome::Success
|
||||
} else {
|
||||
crate::openhuman::approval::ExecutionOutcome::Failure
|
||||
};
|
||||
let err_text = if success {
|
||||
None
|
||||
} else {
|
||||
Some(result_text.as_str())
|
||||
};
|
||||
gate.record_execution(req_id, exec_outcome, err_text);
|
||||
}
|
||||
|
||||
ToolRunResult {
|
||||
text: result_text,
|
||||
success,
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,7 @@ pub(crate) mod builtin_definitions;
|
||||
mod credentials;
|
||||
pub mod definition;
|
||||
pub(crate) mod definition_loader;
|
||||
pub(crate) mod engine;
|
||||
pub mod fork_context;
|
||||
mod instructions;
|
||||
pub mod interrupt;
|
||||
|
||||
@@ -0,0 +1,283 @@
|
||||
//! The Agent's per-call tool executor, extracted as a free function so both
|
||||
//! [`super::types::Agent::execute_tool_call`] and the turn engine's
|
||||
//! `AgentToolSource` run the exact same path (visibility gate → session policy
|
||||
//! → per-call permission → pluggable `ToolPolicy` → `execute_with_options` +
|
||||
//! payload summarizer → per-result byte budget), without one borrowing the
|
||||
//! `Agent` while the turn observer borrows it mutably.
|
||||
//!
|
||||
//! Progress is emitted through a [`ProgressReporter`] (the channel/web flavor),
|
||||
//! matching the `Agent::turn` events 1:1.
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
use crate::core::event_bus::{publish_global, DomainEvent};
|
||||
use crate::openhuman::agent::dispatcher::{ParsedToolCall, ToolExecutionResult};
|
||||
use crate::openhuman::agent::harness::engine::ProgressReporter;
|
||||
use crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer;
|
||||
use crate::openhuman::agent::hooks::{self, ToolCallRecord};
|
||||
use crate::openhuman::agent::tool_policy::{
|
||||
ToolCallContext, ToolPolicy, ToolPolicyDecision, ToolPolicyRequest,
|
||||
};
|
||||
use crate::openhuman::agent_tool_policy::ToolPolicySession;
|
||||
use crate::openhuman::tools::{Tool, ToolCallOptions};
|
||||
use crate::openhuman::util::truncate_with_ellipsis;
|
||||
|
||||
/// Read-only context the Agent tool executor needs, captured up front so it
|
||||
/// never borrows the `Agent` (whose history/context the turn observer mutates).
|
||||
pub(super) struct AgentToolExecCtx<'a> {
|
||||
pub tools: &'a [Box<dyn Tool>],
|
||||
pub visible_tool_names: &'a HashSet<String>,
|
||||
pub tool_policy_session: &'a ToolPolicySession,
|
||||
pub tool_policy: &'a dyn ToolPolicy,
|
||||
pub payload_summarizer: Option<&'a dyn PayloadSummarizer>,
|
||||
pub event_session_id: &'a str,
|
||||
pub event_channel: &'a str,
|
||||
pub agent_definition_id: &'a str,
|
||||
pub prefer_markdown: bool,
|
||||
pub budget_bytes: usize,
|
||||
}
|
||||
|
||||
/// Execute one parsed tool call end-to-end with the Agent's semantics, emitting
|
||||
/// `ToolCallStarted` / `ToolCallCompleted` through `progress`. Returns the
|
||||
/// result (for history formatting) + the call record (for post-turn hooks).
|
||||
pub(super) async fn run_agent_tool_call(
|
||||
ctx: &AgentToolExecCtx<'_>,
|
||||
progress: &dyn ProgressReporter,
|
||||
call: &ParsedToolCall,
|
||||
iteration: usize,
|
||||
) -> (ToolExecutionResult, ToolCallRecord) {
|
||||
let started = std::time::Instant::now();
|
||||
publish_global(DomainEvent::ToolExecutionStarted {
|
||||
tool_name: call.name.clone(),
|
||||
session_id: ctx.event_session_id.to_string(),
|
||||
});
|
||||
// Synthesise a fallback id for prompt-guided (non-native) tool calls so
|
||||
// downstream consumers always have a stable key to reconcile rows by.
|
||||
let call_id = call.tool_call_id.clone().unwrap_or_else(|| {
|
||||
format!(
|
||||
"turn-{iteration}-{}-{}",
|
||||
call.name,
|
||||
uuid::Uuid::new_v4().simple()
|
||||
)
|
||||
});
|
||||
progress
|
||||
.tool_started(
|
||||
&call_id,
|
||||
&call.name,
|
||||
&call.arguments,
|
||||
(iteration + 1) as u32,
|
||||
)
|
||||
.await;
|
||||
log::info!("[agent] executing tool: {}", call.name);
|
||||
|
||||
let (raw_result, success) = if !ctx.visible_tool_names.is_empty()
|
||||
&& !ctx.visible_tool_names.contains(&call.name)
|
||||
{
|
||||
log::warn!(
|
||||
"[agent] blocked tool call '{}' — not in visible tool set",
|
||||
call.name
|
||||
);
|
||||
(
|
||||
format!("Tool '{}' is not available to this agent", call.name),
|
||||
false,
|
||||
)
|
||||
} else if let Some(tool) = ctx.tools.iter().find(|t| t.name() == call.name) {
|
||||
let session_decision = ctx.tool_policy_session.decision_for(&call.name);
|
||||
if session_decision.is_denied() {
|
||||
let required = session_decision
|
||||
.required_permission
|
||||
.map(|permission| permission.to_string())
|
||||
.unwrap_or_else(|| "unknown".to_string());
|
||||
(
|
||||
format!(
|
||||
"Tool '{}' blocked by tool policy: requires {}, channel '{}' allows {}",
|
||||
call.name, required, ctx.event_channel, session_decision.allowed_permission
|
||||
),
|
||||
false,
|
||||
)
|
||||
} else {
|
||||
let call_required = tool.permission_level_with_args(&call.arguments);
|
||||
if call_required > session_decision.allowed_permission {
|
||||
tracing::debug!(
|
||||
tool = call.name.as_str(),
|
||||
call_required = %call_required,
|
||||
allowed = %session_decision.allowed_permission,
|
||||
"[agent_loop] tool action blocked by per-call permission check"
|
||||
);
|
||||
(
|
||||
format!(
|
||||
"Tool '{}' action requires {} permission, channel '{}' allows {}",
|
||||
call.name,
|
||||
call_required,
|
||||
ctx.event_channel,
|
||||
session_decision.allowed_permission
|
||||
),
|
||||
false,
|
||||
)
|
||||
} else {
|
||||
let context = ToolCallContext::session(
|
||||
ctx.event_session_id,
|
||||
ctx.event_channel,
|
||||
ctx.agent_definition_id.to_string(),
|
||||
call_id.clone(),
|
||||
(iteration + 1) as u32,
|
||||
);
|
||||
let mut policy_request =
|
||||
ToolPolicyRequest::new(call.name.clone(), call.arguments.clone(), context);
|
||||
if let Some(generated_context) = tool.generated_runtime_context(&call.arguments) {
|
||||
policy_request = policy_request.with_generated_tool_context(generated_context);
|
||||
}
|
||||
let policy_decision = ctx.tool_policy.check(&policy_request).await;
|
||||
if let Some(reason) = policy_decision.blocking_reason() {
|
||||
let blocked_action = match &policy_decision {
|
||||
ToolPolicyDecision::RequireApproval { .. } => "requires approval",
|
||||
ToolPolicyDecision::Deny { .. } => "denied",
|
||||
ToolPolicyDecision::Allow => "allowed",
|
||||
};
|
||||
crate::openhuman::tool_registry::denials::record(
|
||||
call.name.as_str(),
|
||||
ctx.tool_policy.name(),
|
||||
blocked_action,
|
||||
reason,
|
||||
);
|
||||
tracing::debug!(
|
||||
tool = call.name.as_str(),
|
||||
policy = ctx.tool_policy.name(),
|
||||
action = blocked_action,
|
||||
reason = %reason,
|
||||
"[agent_loop] tool blocked by policy"
|
||||
);
|
||||
(
|
||||
format!(
|
||||
"Tool '{}' {blocked_action} by policy '{}': {reason}",
|
||||
call.name,
|
||||
ctx.tool_policy.name()
|
||||
),
|
||||
false,
|
||||
)
|
||||
} else {
|
||||
let options = ToolCallOptions {
|
||||
prefer_markdown: ctx.prefer_markdown,
|
||||
};
|
||||
let outcome = tool
|
||||
.execute_with_options(call.arguments.clone(), options)
|
||||
.await;
|
||||
match outcome {
|
||||
Ok(r) => {
|
||||
if !r.is_error {
|
||||
let mut output = r.output_for_llm(ctx.prefer_markdown);
|
||||
if ctx.prefer_markdown && r.markdown_formatted.is_some() {
|
||||
log::debug!(
|
||||
"[agent_loop] tool={} returned markdown payload bytes={}",
|
||||
call.name,
|
||||
output.len()
|
||||
);
|
||||
}
|
||||
if let Some(ps) = ctx.payload_summarizer {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer intercepting tool={} bytes={}",
|
||||
call.name,
|
||||
output.len()
|
||||
);
|
||||
match ps.maybe_summarize(&call.name, None, &output).await {
|
||||
Ok(Some(payload)) => {
|
||||
log::info!(
|
||||
"[agent_loop] payload_summarizer compressed tool={} {}->{} bytes",
|
||||
call.name,
|
||||
payload.original_bytes,
|
||||
payload.summary_bytes
|
||||
);
|
||||
output = payload.summary;
|
||||
}
|
||||
Ok(None) => {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer pass-through tool={} bytes={}",
|
||||
call.name,
|
||||
output.len()
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)",
|
||||
call.name,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
(output, true)
|
||||
} else {
|
||||
(
|
||||
format!("Error: {}", r.output_for_llm(ctx.prefer_markdown)),
|
||||
false,
|
||||
)
|
||||
}
|
||||
}
|
||||
Err(e) => (format!("Error executing {}: {e}", call.name), false),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
(format!("Unknown tool: {}", call.name), false)
|
||||
};
|
||||
|
||||
// Per-result byte budget — the only cache-safe reduction stage (the
|
||||
// truncated body has never been sent to the backend).
|
||||
let (result, budget_outcome) =
|
||||
crate::openhuman::context::apply_tool_result_budget(raw_result, ctx.budget_bytes);
|
||||
if budget_outcome.truncated {
|
||||
log::info!(
|
||||
"[agent_loop] tool_result_budget applied name={} original_bytes={} final_bytes={} dropped_bytes={}",
|
||||
call.name,
|
||||
budget_outcome.original_bytes,
|
||||
budget_outcome.final_bytes,
|
||||
budget_outcome.original_bytes - budget_outcome.final_bytes
|
||||
);
|
||||
}
|
||||
|
||||
let elapsed_ms = started.elapsed().as_millis() as u64;
|
||||
publish_global(DomainEvent::ToolExecutionCompleted {
|
||||
tool_name: call.name.clone(),
|
||||
session_id: ctx.event_session_id.to_string(),
|
||||
success,
|
||||
elapsed_ms,
|
||||
});
|
||||
progress
|
||||
.tool_completed(
|
||||
&call_id,
|
||||
&call.name,
|
||||
success,
|
||||
result.chars().count(),
|
||||
elapsed_ms,
|
||||
(iteration + 1) as u32,
|
||||
)
|
||||
.await;
|
||||
log::info!(
|
||||
"[agent] tool completed: {} success={} elapsed_ms={}",
|
||||
call.name,
|
||||
success,
|
||||
elapsed_ms
|
||||
);
|
||||
log::debug!(
|
||||
"[agent] tool output for {}: {}",
|
||||
call.name,
|
||||
truncate_with_ellipsis(&result, 500)
|
||||
);
|
||||
|
||||
let output_summary = hooks::sanitize_tool_output(&result, &call.name, success);
|
||||
let record = ToolCallRecord {
|
||||
name: call.name.clone(),
|
||||
arguments: call.arguments.clone(),
|
||||
success,
|
||||
output_summary,
|
||||
duration_ms: elapsed_ms,
|
||||
};
|
||||
let exec_result = ToolExecutionResult {
|
||||
name: call.name.clone(),
|
||||
output: result,
|
||||
success,
|
||||
tool_call_id: call.tool_call_id.clone(),
|
||||
};
|
||||
(exec_result, record)
|
||||
}
|
||||
@@ -546,9 +546,10 @@ impl AgentBuilder {
|
||||
memory: self
|
||||
.memory
|
||||
.ok_or_else(|| anyhow::anyhow!("memory is required"))?,
|
||||
tool_dispatcher: self
|
||||
.tool_dispatcher
|
||||
.ok_or_else(|| anyhow::anyhow!("tool_dispatcher is required"))?,
|
||||
tool_dispatcher: std::sync::Arc::from(
|
||||
self.tool_dispatcher
|
||||
.ok_or_else(|| anyhow::anyhow!("tool_dispatcher is required"))?,
|
||||
),
|
||||
memory_loader: self
|
||||
.memory_loader
|
||||
.unwrap_or_else(|| Box::new(DefaultMemoryLoader::default())),
|
||||
|
||||
@@ -20,11 +20,14 @@
|
||||
//! `crate::openhuman::agent`, which re-exports them from this module.
|
||||
//! The child files are an implementation detail.
|
||||
|
||||
mod agent_tool_exec;
|
||||
mod builder;
|
||||
pub mod migration;
|
||||
mod runtime;
|
||||
pub(crate) mod transcript;
|
||||
mod turn;
|
||||
mod turn_checkpoint;
|
||||
mod turn_engine_adapter;
|
||||
mod types;
|
||||
|
||||
pub use migration::{migrate_session_layout_if_needed, MigrationOutcome};
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,449 @@
|
||||
//! Engine seams for the stateful `Agent::turn`.
|
||||
//!
|
||||
//! These adapt the `Agent` to the shared [`run_turn_engine`] so web/desktop
|
||||
//! chat runs the same loop as every other entry point, while preserving the
|
||||
//! Agent's richer state: typed `ConversationMessage` history (with structured
|
||||
//! tool calls + round-tripped `reasoning_content`), the `ContextManager`
|
||||
//! reduction chain, KV-cache transcript prefixes, transcript persistence, and
|
||||
//! the pluggable `ToolDispatcher` (incl. PFormat).
|
||||
//!
|
||||
//! * [`AgentToolSource`] owns `Arc`/value clones of the Agent's tool state
|
||||
//! (disjoint from the `&mut Agent` the observer holds) and runs each call
|
||||
//! through the shared [`run_agent_tool_call`], collecting `ToolCallRecord`s.
|
||||
//! * [`AgentObserver`] borrows the `Agent` mutably: it runs the context
|
||||
//! reduction + re-materializes the engine's `ChatMessage` buffer from the
|
||||
//! typed history each iteration, rebuilds the typed history from the engine's
|
||||
//! per-iteration callbacks, accumulates usage, and persists the transcript.
|
||||
//! * [`AgentCheckpoint`] summarizes the turn-so-far into a resumable checkpoint
|
||||
//! when the iteration cap is hit (mirrors `summarize_iteration_checkpoint`).
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
|
||||
use super::agent_tool_exec::{run_agent_tool_call, AgentToolExecCtx};
|
||||
use super::transcript;
|
||||
use super::turn_checkpoint::MAX_ITER_CHECKPOINT_INSTRUCTION;
|
||||
use super::types::Agent;
|
||||
use crate::openhuman::agent::dispatcher::{
|
||||
ParsedToolCall as DispatcherParsedToolCall, ToolDispatcher, ToolExecutionResult,
|
||||
};
|
||||
use crate::openhuman::agent::harness::engine::{
|
||||
CheckpointOutcome, CheckpointStrategy, ProgressReporter, ToolRunResult, ToolSource,
|
||||
TurnObserver,
|
||||
};
|
||||
use crate::openhuman::agent::harness::parse::ParsedToolCall;
|
||||
use crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer;
|
||||
use crate::openhuman::agent::hooks::ToolCallRecord;
|
||||
use crate::openhuman::agent::progress::AgentProgress;
|
||||
use crate::openhuman::agent::tool_policy::ToolPolicy;
|
||||
use crate::openhuman::agent_tool_policy::ToolPolicySession;
|
||||
use crate::openhuman::context::ReductionOutcome;
|
||||
use crate::openhuman::inference::model_context::context_window_for_model;
|
||||
use crate::openhuman::inference::provider::{
|
||||
ChatMessage, ChatRequest, ConversationMessage, Provider, ProviderDelta, ToolCall, UsageInfo,
|
||||
};
|
||||
use crate::openhuman::tools::{Tool, ToolSpec};
|
||||
|
||||
/// Rebuild the persisted `Vec<ToolCall>` for an assistant-with-tools history
|
||||
/// entry: prefer the provider's native calls, else synthesise from the parsed
|
||||
/// calls (mirrors `Agent::persisted_tool_calls_for_history`).
|
||||
fn persisted_tool_calls(
|
||||
native: &[ToolCall],
|
||||
parsed: &[ParsedToolCall],
|
||||
results: &[ToolExecutionResult],
|
||||
iteration: usize,
|
||||
) -> Vec<ToolCall> {
|
||||
if !native.is_empty() {
|
||||
return native.to_vec();
|
||||
}
|
||||
// Synthesise from the parsed calls, reusing the *exact* id each result was
|
||||
// recorded under (`results[i].tool_call_id`) so the persisted assistant
|
||||
// tool-call id matches its `ToolResults` entry — what the next provider
|
||||
// turn (and history-fidelity tests) rely on.
|
||||
parsed
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(idx, c)| {
|
||||
let id = results
|
||||
.get(idx)
|
||||
.and_then(|r| r.tool_call_id.clone())
|
||||
.or_else(|| c.id.clone())
|
||||
.unwrap_or_else(|| format!("parsed-{}-{}", iteration + 1, idx + 1));
|
||||
ToolCall {
|
||||
id,
|
||||
name: c.name.clone(),
|
||||
arguments: c.arguments.to_string(),
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Tool source for `Agent::turn`. Owns clones of the Agent's tool state so it
|
||||
/// doesn't borrow the `Agent` (which [`AgentObserver`] holds mutably).
|
||||
pub(super) struct AgentToolSource {
|
||||
pub tools: Arc<Vec<Box<dyn Tool>>>,
|
||||
pub visible_tool_names: HashSet<String>,
|
||||
pub tool_policy_session: ToolPolicySession,
|
||||
pub tool_policy: Arc<dyn ToolPolicy>,
|
||||
pub payload_summarizer: Option<Arc<dyn PayloadSummarizer>>,
|
||||
pub event_session_id: String,
|
||||
pub event_channel: String,
|
||||
pub agent_definition_id: String,
|
||||
pub prefer_markdown: bool,
|
||||
pub budget_bytes: usize,
|
||||
pub should_send_specs: bool,
|
||||
pub advertised_specs: Vec<ToolSpec>,
|
||||
/// Collected per-call records, drained by the post-loop epilogue for hooks.
|
||||
pub records: Vec<ToolCallRecord>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ToolSource for AgentToolSource {
|
||||
fn request_specs(&self) -> &[ToolSpec] {
|
||||
if self.should_send_specs {
|
||||
&self.advertised_specs
|
||||
} else {
|
||||
&[]
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_call(
|
||||
&mut self,
|
||||
call: &ParsedToolCall,
|
||||
iteration: usize,
|
||||
progress: &dyn ProgressReporter,
|
||||
_progress_call_id: &str,
|
||||
) -> ToolRunResult {
|
||||
// `run_agent_tool_call` takes the dispatcher's `ParsedToolCall` shape;
|
||||
// convert from the engine's internal one.
|
||||
let dispatcher_call = DispatcherParsedToolCall {
|
||||
name: call.name.clone(),
|
||||
arguments: call.arguments.clone(),
|
||||
tool_call_id: call.id.clone(),
|
||||
};
|
||||
let ctx = AgentToolExecCtx {
|
||||
tools: &self.tools,
|
||||
visible_tool_names: &self.visible_tool_names,
|
||||
tool_policy_session: &self.tool_policy_session,
|
||||
tool_policy: self.tool_policy.as_ref(),
|
||||
payload_summarizer: self.payload_summarizer.as_deref(),
|
||||
event_session_id: &self.event_session_id,
|
||||
event_channel: &self.event_channel,
|
||||
agent_definition_id: &self.agent_definition_id,
|
||||
prefer_markdown: self.prefer_markdown,
|
||||
budget_bytes: self.budget_bytes,
|
||||
};
|
||||
let (exec_result, record) =
|
||||
run_agent_tool_call(&ctx, progress, &dispatcher_call, iteration).await;
|
||||
self.records.push(record);
|
||||
ToolRunResult {
|
||||
text: exec_result.output,
|
||||
success: exec_result.success,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Turn observer for `Agent::turn`: owns the typed-history rebuild, context
|
||||
/// management, usage accounting, and transcript persistence.
|
||||
pub(super) struct AgentObserver<'a> {
|
||||
pub agent: &'a mut Agent,
|
||||
pub effective_model: String,
|
||||
pub cumulative_input: u64,
|
||||
pub cumulative_output: u64,
|
||||
pub cumulative_cached: u64,
|
||||
pub cumulative_charged: f64,
|
||||
pub last_turn_usage: Option<transcript::TurnUsage>,
|
||||
/// Cached transcript prefix for KV-cache reuse on a resumed session,
|
||||
/// consumed on the first iteration.
|
||||
pub cached_prefix: Option<Vec<ChatMessage>>,
|
||||
/// Tool results buffered during the per-call loop, flushed to typed history
|
||||
/// via the dispatcher's `format_results` once the assistant turn lands.
|
||||
pub pending_results: Vec<ToolExecutionResult>,
|
||||
/// Whether the engine reported a clean final response (so the post-loop
|
||||
/// epilogue knows not to push `outcome.text` itself).
|
||||
pub did_push_final: bool,
|
||||
}
|
||||
|
||||
impl AgentObserver<'_> {
|
||||
fn persist(&mut self) {
|
||||
let messages = self
|
||||
.agent
|
||||
.tool_dispatcher
|
||||
.to_provider_messages(&self.agent.history);
|
||||
self.agent.persist_session_transcript(
|
||||
&messages,
|
||||
self.cumulative_input,
|
||||
self.cumulative_output,
|
||||
self.cumulative_cached,
|
||||
self.cumulative_charged,
|
||||
self.last_turn_usage.as_ref(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl TurnObserver for AgentObserver<'_> {
|
||||
async fn before_dispatch(
|
||||
&mut self,
|
||||
buf: &mut Vec<ChatMessage>,
|
||||
_iteration: usize,
|
||||
) -> Result<()> {
|
||||
// Pre-dispatch token-budget trim on the typed history.
|
||||
if let Some(context_window) = context_window_for_model(&self.effective_model) {
|
||||
super::super::token_budget::trim_conversation_history_to_budget(
|
||||
&mut self.agent.history,
|
||||
context_window,
|
||||
);
|
||||
}
|
||||
// Global context-management reduction chain.
|
||||
let outcome = self
|
||||
.agent
|
||||
.context
|
||||
.reduce_before_call(&mut self.agent.history)
|
||||
.await?;
|
||||
if let ReductionOutcome::Exhausted {
|
||||
utilisation_pct,
|
||||
reason,
|
||||
} = &outcome
|
||||
{
|
||||
return Err(anyhow::anyhow!(
|
||||
"Context window exhausted ({utilisation_pct}% full): {reason}"
|
||||
));
|
||||
}
|
||||
|
||||
// Re-materialize the engine's ChatMessage buffer from the typed
|
||||
// history. On the first iteration of a resumed session, splice the
|
||||
// byte-identical cached prefix + the new user-message tail for KV-cache
|
||||
// reuse; otherwise rebuild from scratch.
|
||||
let messages = if let Some(mut cached) = self.cached_prefix.take() {
|
||||
let tail = self.agent.tool_dispatcher.to_provider_messages(
|
||||
&self.agent.history[self.agent.history.len().saturating_sub(1)..],
|
||||
);
|
||||
cached.extend(tail);
|
||||
cached
|
||||
} else {
|
||||
self.agent
|
||||
.tool_dispatcher
|
||||
.to_provider_messages(&self.agent.history)
|
||||
};
|
||||
*buf = messages;
|
||||
// Second-pass trim on the materialized provider messages (mirrors the
|
||||
// legacy `Agent::turn`, which trimmed both the typed history and the
|
||||
// built `ChatMessage` list).
|
||||
if let Some(context_window) = context_window_for_model(&self.effective_model) {
|
||||
super::super::token_budget::trim_chat_messages_to_budget(buf, context_window);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn allow_empty_final(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn record_usage(&mut self, model: &str, usage: &UsageInfo) {
|
||||
self.agent.context.record_usage(usage);
|
||||
crate::openhuman::cost::record_provider_usage(model, usage);
|
||||
self.cumulative_input += usage.input_tokens;
|
||||
self.cumulative_output += usage.output_tokens;
|
||||
self.cumulative_cached += usage.cached_input_tokens;
|
||||
self.cumulative_charged += usage.charged_amount_usd;
|
||||
self.last_turn_usage = Some(transcript::TurnUsage {
|
||||
model: model.to_string(),
|
||||
usage: transcript::MessageUsage {
|
||||
input: usage.input_tokens,
|
||||
output: usage.output_tokens,
|
||||
cached_input: usage.cached_input_tokens,
|
||||
cost_usd: usage.charged_amount_usd,
|
||||
},
|
||||
ts: chrono::Utc::now().to_rfc3339(),
|
||||
});
|
||||
}
|
||||
|
||||
fn on_assistant(
|
||||
&mut self,
|
||||
display_text: &str,
|
||||
_response_text: &str,
|
||||
reasoning_content: Option<&str>,
|
||||
native_tool_calls: &[ToolCall],
|
||||
parsed_calls: &[ParsedToolCall],
|
||||
iteration: usize,
|
||||
is_final: bool,
|
||||
) {
|
||||
if is_final {
|
||||
let mut assistant_msg = ChatMessage::assistant(display_text.to_string());
|
||||
if let Some(rc) = reasoning_content {
|
||||
assistant_msg.extra_metadata = Some(serde_json::json!({ "reasoning_content": rc }));
|
||||
}
|
||||
self.agent
|
||||
.history
|
||||
.push(ConversationMessage::Chat(assistant_msg));
|
||||
self.agent.trim_history();
|
||||
self.did_push_final = true;
|
||||
return;
|
||||
}
|
||||
|
||||
// Assistant turn with tool calls. Mirror `Agent::turn` exactly: push the
|
||||
// pre-tool narrative text (if any) as a standalone Chat message, then
|
||||
// the structured AssistantToolCalls, then the dispatcher-formatted
|
||||
// results buffered during the per-call loop.
|
||||
if !display_text.is_empty() {
|
||||
self.agent
|
||||
.history
|
||||
.push(ConversationMessage::Chat(ChatMessage::assistant(
|
||||
display_text.to_string(),
|
||||
)));
|
||||
}
|
||||
let tool_calls = persisted_tool_calls(
|
||||
native_tool_calls,
|
||||
parsed_calls,
|
||||
&self.pending_results,
|
||||
iteration,
|
||||
);
|
||||
self.agent
|
||||
.history
|
||||
.push(ConversationMessage::AssistantToolCalls {
|
||||
text: if display_text.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(display_text.to_string())
|
||||
},
|
||||
tool_calls,
|
||||
reasoning_content: reasoning_content
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(ToString::to_string),
|
||||
});
|
||||
let results = std::mem::take(&mut self.pending_results);
|
||||
let formatted = self.agent.tool_dispatcher.format_results(&results);
|
||||
self.agent.history.push(formatted);
|
||||
self.agent.trim_history();
|
||||
}
|
||||
|
||||
fn on_tool_result(
|
||||
&mut self,
|
||||
call_id: &str,
|
||||
tool_name: &str,
|
||||
result_text: &str,
|
||||
success: bool,
|
||||
_iteration: usize,
|
||||
) {
|
||||
self.pending_results.push(ToolExecutionResult {
|
||||
name: tool_name.to_string(),
|
||||
output: result_text.to_string(),
|
||||
success,
|
||||
tool_call_id: Some(call_id.to_string()),
|
||||
});
|
||||
}
|
||||
|
||||
fn after_iteration(&mut self, _buf: &[ChatMessage], _iteration: usize) {
|
||||
self.persist();
|
||||
}
|
||||
}
|
||||
|
||||
/// Max-iteration checkpoint for `Agent::turn`: summarize the turn's tool digest
|
||||
/// into a resumable checkpoint (streaming text deltas through the progress
|
||||
/// sink), with a deterministic fallback.
|
||||
pub(super) struct AgentCheckpoint {
|
||||
pub provider: Arc<dyn Provider>,
|
||||
pub dispatcher: Arc<dyn ToolDispatcher>,
|
||||
pub model: String,
|
||||
pub temperature: f64,
|
||||
pub on_progress: Option<tokio::sync::mpsc::Sender<AgentProgress>>,
|
||||
pub user_message: String,
|
||||
pub max_iterations: usize,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CheckpointStrategy for AgentCheckpoint {
|
||||
async fn on_max_iter(&self, digest: &str, max_iterations: usize) -> Result<CheckpointOutcome> {
|
||||
let deterministic = format!(
|
||||
"I reached the tool-call limit for this turn ({max_iterations} steps), so I paused here.\n\n\
|
||||
**Done so far:**\n{digest}\n\
|
||||
**Next steps:** I'll continue from here — just reply (e.g. \"continue\") and I'll pick up \
|
||||
where I left off."
|
||||
);
|
||||
let mut messages = vec![ChatMessage::user(format!(
|
||||
"You were working on this user request:\n{}\n\nHere are the tool calls you made this turn \
|
||||
and their results — compile your checkpoint from these:\n{}",
|
||||
self.user_message, digest
|
||||
))];
|
||||
messages.push(ChatMessage::user(MAX_ITER_CHECKPOINT_INSTRUCTION));
|
||||
|
||||
let checkpoint_iteration = (self.max_iterations + 1) as u32;
|
||||
// Stream the checkpoint prose as text deltas (tools disabled).
|
||||
let (delta_tx_opt, delta_forwarder) = if self.on_progress.is_some() {
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<ProviderDelta>(128);
|
||||
let progress_tx = self.on_progress.clone();
|
||||
let forwarder = tokio::spawn(async move {
|
||||
while let Some(event) = rx.recv().await {
|
||||
let Some(ref sink) = progress_tx else {
|
||||
continue;
|
||||
};
|
||||
if let ProviderDelta::TextDelta { delta } = event {
|
||||
if sink
|
||||
.send(AgentProgress::TextDelta {
|
||||
delta,
|
||||
iteration: checkpoint_iteration,
|
||||
})
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
(Some(tx), Some(forwarder))
|
||||
} else {
|
||||
(None, None)
|
||||
};
|
||||
|
||||
let result = self
|
||||
.provider
|
||||
.chat(
|
||||
ChatRequest {
|
||||
messages: &messages,
|
||||
tools: None,
|
||||
stream: delta_tx_opt.as_ref(),
|
||||
},
|
||||
&self.model,
|
||||
self.temperature,
|
||||
)
|
||||
.await;
|
||||
drop(delta_tx_opt);
|
||||
if let Some(handle) = delta_forwarder {
|
||||
let _ = handle.await;
|
||||
}
|
||||
|
||||
match result {
|
||||
Ok(resp) => {
|
||||
let usage = resp.usage.clone();
|
||||
// Strip any stray tool-call markup; keep only prose.
|
||||
let (text, calls) = self.dispatcher.parse_response(&resp);
|
||||
let checkpoint = if !text.trim().is_empty() {
|
||||
text
|
||||
} else if calls.is_empty() {
|
||||
resp.text.unwrap_or_default()
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
let text = if checkpoint.trim().is_empty() {
|
||||
deterministic
|
||||
} else {
|
||||
checkpoint
|
||||
};
|
||||
Ok(CheckpointOutcome { text, usage })
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!("[agent_loop] checkpoint summary call failed: {e:#}");
|
||||
Ok(CheckpointOutcome {
|
||||
text: deterministic,
|
||||
usage: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -45,7 +45,10 @@ pub struct Agent {
|
||||
pub(super) visible_tool_names: std::collections::HashSet<String>,
|
||||
pub(super) tool_policy_session: ToolPolicySession,
|
||||
pub(super) memory: Arc<dyn Memory>,
|
||||
pub(super) tool_dispatcher: Box<dyn ToolDispatcher>,
|
||||
// `Arc` (not `Box`) so the turn engine's parser seam can hold a cheap clone
|
||||
// of the dispatcher without borrowing the `Agent` (which the turn observer
|
||||
// borrows mutably) — see `engine::DispatcherParser`.
|
||||
pub(super) tool_dispatcher: Arc<dyn ToolDispatcher>,
|
||||
pub(super) memory_loader: Box<dyn MemoryLoader>,
|
||||
pub(super) config: crate::openhuman::config::AgentConfig,
|
||||
pub(super) model_name: String,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,28 +1,14 @@
|
||||
use crate::openhuman::agent::cost::TurnCost;
|
||||
use crate::openhuman::agent::multimodal;
|
||||
use crate::openhuman::agent::progress::AgentProgress;
|
||||
use crate::openhuman::agent::stop_hooks::{current_stop_hooks, StopDecision, TurnState};
|
||||
use crate::openhuman::inference::provider::{
|
||||
ChatMessage, ChatRequest, Provider, ProviderCapabilityError, ProviderDelta,
|
||||
};
|
||||
use crate::openhuman::tools::policy::{DefaultToolPolicy, PolicyDecision, ToolPolicy};
|
||||
use crate::openhuman::tools::traits::ToolScope;
|
||||
use crate::openhuman::inference::provider::{ChatMessage, Provider};
|
||||
use crate::openhuman::tools::policy::{DefaultToolPolicy, ToolPolicy};
|
||||
use crate::openhuman::tools::Tool;
|
||||
use anyhow::Result;
|
||||
use std::collections::HashSet;
|
||||
use std::fmt::Write as _;
|
||||
use std::io::Write as _;
|
||||
|
||||
use super::credentials::scrub_credentials;
|
||||
use super::parse::{build_native_assistant_history, parse_structured_tool_calls, parse_tool_calls};
|
||||
use super::payload_summarizer::PayloadSummarizer;
|
||||
use crate::openhuman::context::guard::{ContextCheckResult, ContextGuard};
|
||||
use crate::openhuman::inference::model_context::context_window_for_model;
|
||||
|
||||
use super::token_budget::trim_chat_messages_to_budget;
|
||||
|
||||
/// Minimum characters per chunk when relaying LLM text to a streaming draft.
|
||||
const STREAM_CHUNK_MIN_CHARS: usize = 80;
|
||||
pub(crate) const STREAM_CHUNK_MIN_CHARS: usize = 80;
|
||||
|
||||
/// Default maximum agentic tool-use iterations per user message to prevent runaway loops.
|
||||
/// Used as a safe fallback when `max_tool_iterations` is unset or configured as zero.
|
||||
@@ -272,932 +258,49 @@ pub(crate) async fn run_tool_call_loop(
|
||||
max_tool_iterations
|
||||
};
|
||||
|
||||
// Is a given tool name visible to the model this turn? `None`
|
||||
// means no filter (legacy behaviour = everything visible).
|
||||
let is_visible = |name: &str| -> bool {
|
||||
match visible_tool_names {
|
||||
Some(set) => set.contains(name),
|
||||
None => true,
|
||||
}
|
||||
};
|
||||
|
||||
// Filter to visible tools, then dedup by name before sending to the
|
||||
// provider. Registry tools may collide with per-turn synthesised
|
||||
// extra_tools (e.g. an `ArchetypeDelegationTool` whose
|
||||
// `delegate_name = "research"` shadowing a same-named skill). Some
|
||||
// providers (Anthropic, OpenHuman cloud after the uniqueness-enforcement
|
||||
// rollout) 400 on duplicate tool names — see TAURI-RUST-4.
|
||||
let filtered_specs: Vec<crate::openhuman::tools::ToolSpec> = tools_registry
|
||||
.iter()
|
||||
.chain(extra_tools.iter())
|
||||
.filter(|tool| is_visible(tool.name()))
|
||||
.map(|tool| tool.spec())
|
||||
.collect();
|
||||
let tool_specs =
|
||||
crate::openhuman::agent::harness::session::dedup_visible_tool_specs(filtered_specs);
|
||||
let use_native_tools = provider.supports_native_tools() && !tool_specs.is_empty();
|
||||
|
||||
// The agentic loop itself now lives in the shared turn engine; this
|
||||
// function is a thin adapter that builds the channel/CLI tool source
|
||||
// (registry + per-turn extras, visibility whitelist, pluggable policy)
|
||||
// and hands off. The signature is retained verbatim so existing callers
|
||||
// (the `agent.run_turn` bus handler, triage, the payload summarizer, and
|
||||
// the harness test suite) are unaffected.
|
||||
log::debug!(
|
||||
"[tool-loop] Registry has {} tool(s), extra {} tool(s), filter={} — {} visible in schema: [{}]",
|
||||
"[tool-loop] Registry has {} tool(s), extra {} tool(s), filter={}",
|
||||
tools_registry.len(),
|
||||
extra_tools.len(),
|
||||
visible_tool_names
|
||||
.map(|s| format!("whitelist({})", s.len()))
|
||||
.unwrap_or_else(|| "none".to_string()),
|
||||
tool_specs.len(),
|
||||
tool_specs
|
||||
.iter()
|
||||
.map(|s| s.name.as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ")
|
||||
);
|
||||
|
||||
let mut context_guard = context_window_for_model(model)
|
||||
.map(ContextGuard::with_context_window)
|
||||
.unwrap_or_else(ContextGuard::new);
|
||||
let mut turn_cost = TurnCost::new();
|
||||
|
||||
// Announce turn start to progress subscribers (if any). We use
|
||||
// `send().await` for lifecycle (turn/iteration) events so they
|
||||
// survive downstream backpressure — dropping one of these would
|
||||
// desync the web-channel progress bridge. High-volume delta events
|
||||
// use the same backpressure discipline (see below).
|
||||
if let Some(ref sink) = on_progress {
|
||||
if let Err(e) = sink.send(AgentProgress::TurnStarted).await {
|
||||
log::warn!("[agent_loop] progress sink closed at TurnStarted: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
let stop_hooks = current_stop_hooks();
|
||||
// Repeated-failure circuit breaker — halts with a root cause rather than
|
||||
// grinding to `max_iterations` (shared with the subagent loop).
|
||||
let mut failure_guard = RepeatFailureGuard::new();
|
||||
let mut halt_reason: Option<String> = None;
|
||||
for iteration in 0..max_iterations {
|
||||
if let Some(ref sink) = on_progress {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::IterationStarted {
|
||||
iteration: (iteration + 1) as u32,
|
||||
max_iterations: max_iterations as u32,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!("[agent_loop] progress sink closed at IterationStarted: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
// ── Stop hooks: policy check before the next LLM call ──
|
||||
if !stop_hooks.is_empty() {
|
||||
let state = TurnState {
|
||||
iteration: (iteration + 1) as u32,
|
||||
max_iterations: max_iterations as u32,
|
||||
cost: &turn_cost,
|
||||
model,
|
||||
};
|
||||
for hook in &stop_hooks {
|
||||
match hook.check(&state).await {
|
||||
StopDecision::Continue => {}
|
||||
StopDecision::Stop { reason } => {
|
||||
tracing::warn!(
|
||||
iteration = (iteration + 1),
|
||||
hook = hook.name(),
|
||||
reason = %reason,
|
||||
"[agent_loop] stop hook triggered — aborting turn"
|
||||
);
|
||||
anyhow::bail!("Agent turn stopped by hook '{}': {reason}", hook.name());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── 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)",
|
||||
crate::openhuman::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,
|
||||
} => {
|
||||
let msg = format!("Context window exhausted ({utilization_pct}% full): {reason}");
|
||||
crate::core::observability::report_error(
|
||||
msg.as_str(),
|
||||
"agent",
|
||||
"context_exhausted",
|
||||
&[
|
||||
("provider", provider_name),
|
||||
("model", model),
|
||||
("utilization_pct", &utilization_pct.to_string()),
|
||||
],
|
||||
);
|
||||
anyhow::bail!(msg);
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(context_window) = context_window_for_model(model) {
|
||||
let budget_outcome = trim_chat_messages_to_budget(history, context_window);
|
||||
if budget_outcome.trimmed {
|
||||
log::warn!(
|
||||
"[agent_loop] pre-dispatch history trimmed model={} context_window={} original_tokens={} final_tokens={} messages_removed={}",
|
||||
model,
|
||||
context_window,
|
||||
budget_outcome.original_tokens,
|
||||
budget_outcome.final_tokens,
|
||||
budget_outcome.messages_removed
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
model,
|
||||
context_window,
|
||||
estimated_tokens = budget_outcome.final_tokens,
|
||||
"[agent_loop] pre-dispatch token budget ok"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
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() {
|
||||
let cap_err = ProviderCapabilityError {
|
||||
provider: provider_name.to_string(),
|
||||
capability: "vision".to_string(),
|
||||
message: format!(
|
||||
"received {image_marker_count} image marker(s), but this provider does not support vision input"
|
||||
),
|
||||
};
|
||||
crate::core::observability::report_error(
|
||||
&cap_err,
|
||||
"agent",
|
||||
"provider_capability",
|
||||
&[
|
||||
("provider", provider_name),
|
||||
("capability", "vision"),
|
||||
("model", model),
|
||||
],
|
||||
);
|
||||
return Err(cap_err.into());
|
||||
}
|
||||
|
||||
let prepared_messages =
|
||||
multimodal::prepare_messages_for_provider(history, multimodal_config).await?;
|
||||
|
||||
// Unified path via Provider::chat so provider-specific native tool logic
|
||||
// (OpenAI/Anthropic/OpenRouter/compatible adapters) is honored.
|
||||
let request_tools = if use_native_tools {
|
||||
Some(tool_specs.as_slice())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Wire up a ProviderDelta → AgentProgress forwarder for this
|
||||
// iteration when a progress sink exists. Senders dropped after
|
||||
// the chat call so the forwarder task exits cleanly.
|
||||
let iteration_for_stream = (iteration + 1) as u32;
|
||||
let (delta_tx_opt, delta_forwarder) = if let Some(progress_sink) = on_progress.clone() {
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<ProviderDelta>(128);
|
||||
let forwarder = tokio::spawn(async move {
|
||||
while let Some(event) = rx.recv().await {
|
||||
let mapped = match event {
|
||||
ProviderDelta::TextDelta { delta } => AgentProgress::TextDelta {
|
||||
delta,
|
||||
iteration: iteration_for_stream,
|
||||
},
|
||||
ProviderDelta::ThinkingDelta { delta } => AgentProgress::ThinkingDelta {
|
||||
delta,
|
||||
iteration: iteration_for_stream,
|
||||
},
|
||||
ProviderDelta::ToolCallStart { call_id, tool_name } => {
|
||||
AgentProgress::ToolCallArgsDelta {
|
||||
call_id,
|
||||
tool_name,
|
||||
delta: String::new(),
|
||||
iteration: iteration_for_stream,
|
||||
}
|
||||
}
|
||||
ProviderDelta::ToolCallArgsDelta { call_id, delta } => {
|
||||
AgentProgress::ToolCallArgsDelta {
|
||||
call_id,
|
||||
tool_name: String::new(),
|
||||
delta,
|
||||
iteration: iteration_for_stream,
|
||||
}
|
||||
}
|
||||
};
|
||||
// Await backpressure rather than dropping deltas so
|
||||
// partial streamed text/args stays consistent with the
|
||||
// eventual ToolCallStarted / ToolCallCompleted events.
|
||||
if progress_sink.send(mapped).await.is_err() {
|
||||
// Downstream closed — abandon the forwarder.
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
(Some(tx), Some(forwarder))
|
||||
} else {
|
||||
(None, None)
|
||||
};
|
||||
|
||||
let chat_result = provider
|
||||
.chat(
|
||||
ChatRequest {
|
||||
messages: &prepared_messages.messages,
|
||||
tools: request_tools,
|
||||
stream: delta_tx_opt.as_ref(),
|
||||
},
|
||||
model,
|
||||
temperature,
|
||||
)
|
||||
.await;
|
||||
|
||||
drop(delta_tx_opt);
|
||||
if let Some(handle) = delta_forwarder {
|
||||
let _ = handle.await;
|
||||
}
|
||||
|
||||
let (response_text, parsed_text, tool_calls, assistant_history_content, native_tool_calls) =
|
||||
match chat_result {
|
||||
Ok(resp) => {
|
||||
// Update context guard with token usage from this response.
|
||||
if let Some(ref usage) = resp.usage {
|
||||
context_guard.update_usage(usage);
|
||||
turn_cost.add_call(model, usage);
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
input_tokens = usage.input_tokens,
|
||||
output_tokens = usage.output_tokens,
|
||||
context_window = usage.context_window,
|
||||
cumulative_usd = turn_cost.total_usd(),
|
||||
"[agent_loop] LLM response received"
|
||||
);
|
||||
if let Some(ref sink) = on_progress {
|
||||
let event = AgentProgress::TurnCostUpdated {
|
||||
model: model.to_string(),
|
||||
iteration: (iteration + 1) as u32,
|
||||
input_tokens: turn_cost.input_tokens,
|
||||
output_tokens: turn_cost.output_tokens,
|
||||
cached_input_tokens: turn_cost.cached_input_tokens,
|
||||
total_usd: turn_cost.total_usd(),
|
||||
};
|
||||
if let Err(e) = sink.send(event).await {
|
||||
log::warn!(
|
||||
"[agent_loop] progress sink closed at TurnCostUpdated: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
} 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();
|
||||
|
||||
if calls.is_empty() {
|
||||
let (fallback_text, fallback_calls) = parse_tool_calls(&response_text);
|
||||
if !fallback_text.is_empty() {
|
||||
parsed_text = fallback_text;
|
||||
}
|
||||
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() {
|
||||
response_text.clone()
|
||||
} else {
|
||||
build_native_assistant_history(
|
||||
&response_text,
|
||||
resp.reasoning_content.as_deref(),
|
||||
&resp.tool_calls,
|
||||
)
|
||||
};
|
||||
|
||||
let native_calls = resp.tool_calls;
|
||||
(
|
||||
response_text,
|
||||
parsed_text,
|
||||
calls,
|
||||
assistant_history_content,
|
||||
native_calls,
|
||||
)
|
||||
}
|
||||
Err(e) => {
|
||||
// Transient upstream failures (rate-limit, gateway 5xx, "no
|
||||
// healthy upstream", etc.) are already classified + retried
|
||||
// by reliable.rs and produce an aggregate Sentry event only
|
||||
// when every provider/model is exhausted. Reporting each
|
||||
// per-iteration provider_chat error here duplicates the
|
||||
// signal and floods Sentry — see OPENHUMAN-TAURI-3Y/3Z
|
||||
// (~46 events combined) and the underlying TAURI-2E/84/T
|
||||
// (~3300 events from raw per-attempt 429/503/504 reports).
|
||||
let transient = crate::openhuman::inference::provider::reliable::is_rate_limited(
|
||||
&e,
|
||||
)
|
||||
|| crate::openhuman::inference::provider::reliable::is_upstream_unhealthy(
|
||||
&e,
|
||||
);
|
||||
if transient {
|
||||
tracing::warn!(
|
||||
domain = "agent",
|
||||
operation = "provider_chat",
|
||||
provider = provider_name,
|
||||
model = model,
|
||||
iteration = iteration + 1,
|
||||
error = %format!("{e:#}"),
|
||||
"[agent] transient provider_chat failure — retried upstream; \
|
||||
aggregated all-providers-exhausted will report if applicable"
|
||||
);
|
||||
} else {
|
||||
crate::core::observability::report_error_or_expected(
|
||||
&e,
|
||||
"agent",
|
||||
"provider_chat",
|
||||
&[
|
||||
("provider", provider_name),
|
||||
("model", model),
|
||||
("iteration", &(iteration + 1).to_string()),
|
||||
],
|
||||
);
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
let display_text = if parsed_text.is_empty() {
|
||||
response_text.clone()
|
||||
} else {
|
||||
parsed_text
|
||||
};
|
||||
|
||||
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.
|
||||
if let Some(ref tx) = on_delta {
|
||||
// Split on whitespace boundaries, accumulating chunks of at least
|
||||
// STREAM_CHUNK_MIN_CHARS characters for progressive draft updates.
|
||||
let mut chunk = String::new();
|
||||
for word in display_text.split_inclusive(char::is_whitespace) {
|
||||
chunk.push_str(word);
|
||||
if chunk.len() >= STREAM_CHUNK_MIN_CHARS
|
||||
&& tx.send(std::mem::take(&mut chunk)).await.is_err()
|
||||
{
|
||||
break; // receiver dropped
|
||||
}
|
||||
}
|
||||
if !chunk.is_empty() {
|
||||
let _ = tx.send(chunk).await;
|
||||
}
|
||||
}
|
||||
history.push(ChatMessage::assistant(response_text.clone()));
|
||||
log::info!(
|
||||
"[agent_loop] turn complete: iters={} provider_calls={} tokens_in={} tokens_out={} cached_in={} usd={:.4}",
|
||||
(iteration + 1),
|
||||
turn_cost.call_count,
|
||||
turn_cost.input_tokens,
|
||||
turn_cost.output_tokens,
|
||||
turn_cost.cached_input_tokens,
|
||||
turn_cost.total_usd(),
|
||||
);
|
||||
if let Some(ref sink) = on_progress {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::TurnCompleted {
|
||||
iterations: (iteration + 1) as u32,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!("[agent_loop] progress sink closed at TurnCompleted: {e}");
|
||||
}
|
||||
}
|
||||
return Ok(display_text);
|
||||
}
|
||||
|
||||
// Print any text the LLM produced alongside tool calls (unless silent)
|
||||
if !silent && !display_text.is_empty() {
|
||||
print!("{display_text}");
|
||||
let _ = std::io::stdout().flush();
|
||||
}
|
||||
|
||||
// Execute each tool call and build results.
|
||||
// `individual_results` tracks per-call output so that native-mode history
|
||||
// can emit one `role: tool` message per tool call with the correct ID.
|
||||
let mut tool_results = String::new();
|
||||
let mut individual_results: Vec<String> = Vec::new();
|
||||
for (call_idx, call) in tool_calls.iter().enumerate() {
|
||||
// Stable id threaded through the start/complete pair (and
|
||||
// any preceding args-delta events) so consumers can
|
||||
// reconcile tool rows by id. The fallback includes
|
||||
// `call_idx` to stay unique when the same tool name
|
||||
// appears multiple times in one iteration.
|
||||
let progress_call_id = call
|
||||
.id
|
||||
.clone()
|
||||
.unwrap_or_else(|| format!("loop-{iteration}-{call_idx}-{}", call.name));
|
||||
// Emit `ToolCallStarted` for every parsed call, even ones
|
||||
// that will be rejected below (approval denied, CliRpcOnly,
|
||||
// unknown) — the client-side row was created from the
|
||||
// streamed args and needs a terminal event to resolve.
|
||||
if let Some(ref sink) = on_progress {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::ToolCallStarted {
|
||||
call_id: progress_call_id.clone(),
|
||||
tool_name: call.name.clone(),
|
||||
arguments: call.arguments.clone(),
|
||||
iteration: (iteration + 1) as u32,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!(
|
||||
"[agent_loop] progress sink closed while emitting ToolCallStarted: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Helper: emit a failed `ToolCallCompleted` for an
|
||||
// early-exit path (denied / CliRpcOnly / unknown) so the
|
||||
// client row flips to `error` instead of staying running.
|
||||
let emit_failed_completion = |message: &str| {
|
||||
let call_id = progress_call_id.clone();
|
||||
let tool_name = call.name.clone();
|
||||
let output_chars = message.chars().count();
|
||||
let iteration_u32 = (iteration + 1) as u32;
|
||||
let sink_opt = on_progress.clone();
|
||||
async move {
|
||||
if let Some(sink) = sink_opt {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::ToolCallCompleted {
|
||||
call_id,
|
||||
tool_name,
|
||||
success: false,
|
||||
output_chars,
|
||||
elapsed_ms: 0,
|
||||
iteration: iteration_u32,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!(
|
||||
"[agent_loop] progress sink closed while emitting early-exit ToolCallCompleted: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// ── Tool policy check (#2131) ─────────────────
|
||||
// Evaluate the pluggable ToolPolicy before any approval or
|
||||
// execution. If the policy denies the call, skip everything
|
||||
// (including approval side-effects) and return the denial
|
||||
// reason as a tool error to the model.
|
||||
if let PolicyDecision::Deny(reason) = tool_policy.evaluate(&call.name, &call.arguments)
|
||||
{
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
reason = %reason,
|
||||
"[agent_loop] tool policy denied tool call"
|
||||
);
|
||||
let denied = format!("Tool '{}' denied by policy: {reason}", call.name);
|
||||
emit_failed_completion(&denied).await;
|
||||
individual_results.push(denied.clone());
|
||||
let _ = writeln!(
|
||||
tool_results,
|
||||
"<tool_result name=\"{}\">\n{denied}\n</tool_result>",
|
||||
call.name
|
||||
);
|
||||
// Record so a re-issued identical call halts the turn rather than
|
||||
// repeating a deterministic policy denial to max_iterations.
|
||||
if let Some(halt) =
|
||||
failure_guard.record(&call.name, &call.arguments.to_string(), false, &denied)
|
||||
{
|
||||
halt_reason = Some(halt);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Look up the tool by name in the combined registry + extras,
|
||||
// subject to the visibility whitelist. If the model hallucinated
|
||||
// a filtered-out tool name we treat it as unknown — the error
|
||||
// path below produces a structured error message the LLM can
|
||||
// correct in the next iteration.
|
||||
let tool_opt: Option<&dyn Tool> = tools_registry
|
||||
.iter()
|
||||
.chain(extra_tools.iter())
|
||||
.find(|t| t.name() == call.name && is_visible(t.name()))
|
||||
.map(|b| b.as_ref());
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
found = tool_opt.is_some(),
|
||||
"[agent_loop] executing tool"
|
||||
);
|
||||
|
||||
// Scope check: CliRpcOnly tools cannot run in the autonomous agent loop.
|
||||
if let Some(tool) = tool_opt {
|
||||
if tool.scope() == ToolScope::CliRpcOnly {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] tool scope is CliRpcOnly — denied in agent loop"
|
||||
);
|
||||
let denied = format!(
|
||||
"Tool '{}' is only available via explicit CLI/RPC invocation, not in the autonomous agent loop.",
|
||||
call.name
|
||||
);
|
||||
emit_failed_completion(&denied).await;
|
||||
individual_results.push(denied.clone());
|
||||
let _ = writeln!(
|
||||
tool_results,
|
||||
"<tool_result name=\"{}\">\n{denied}\n</tool_result>",
|
||||
call.name
|
||||
);
|
||||
if let Some(halt) = failure_guard.record(
|
||||
&call.name,
|
||||
&call.arguments.to_string(),
|
||||
false,
|
||||
&denied,
|
||||
) {
|
||||
halt_reason = Some(halt);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// ── External-effect approval gate (#1339, #2135) ──
|
||||
// Tools whose `external_effect()` returns true route
|
||||
// through the process-global `ApprovalGate` so the UI
|
||||
// can prompt the user before `execute()` runs. The gate
|
||||
// is `None` when supervised mode is disabled or in test
|
||||
// envs — behavior matches the pre-#1339 path.
|
||||
//
|
||||
// `approval_request_id` carries the persisted row id
|
||||
// forward so we can stamp the terminal execution
|
||||
// outcome onto the same `pending_approvals` row after
|
||||
// the tool finishes (issue #2135). `None` means the
|
||||
// tool was either not gated (no supervised gate, not
|
||||
// external-effect), was session-allowlist-shortcutted,
|
||||
// or was denied — none of which produce an audit row
|
||||
// that needs an "after" entry.
|
||||
let mut approval_request_id: Option<String> = None;
|
||||
let mut approval_gate_for_audit: Option<
|
||||
std::sync::Arc<crate::openhuman::approval::ApprovalGate>,
|
||||
> = None;
|
||||
if let Some(tool) = tool_opt {
|
||||
if tool.external_effect_with_args(&call.arguments) {
|
||||
if let Some(gate) = crate::openhuman::approval::ApprovalGate::try_global() {
|
||||
let summary = crate::openhuman::approval::summarize_action(
|
||||
&call.name,
|
||||
&call.arguments,
|
||||
);
|
||||
let redacted = crate::openhuman::approval::redact_args(&call.arguments);
|
||||
let (outcome, request_id) =
|
||||
gate.intercept_audited(&call.name, &summary, redacted).await;
|
||||
match outcome {
|
||||
crate::openhuman::approval::GateOutcome::Allow => {
|
||||
approval_request_id = request_id;
|
||||
if approval_request_id.is_some() {
|
||||
approval_gate_for_audit = Some(gate);
|
||||
}
|
||||
}
|
||||
crate::openhuman::approval::GateOutcome::Deny { reason } => {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
reason = %reason,
|
||||
"[agent_loop] approval gate denied tool call"
|
||||
);
|
||||
emit_failed_completion(&reason).await;
|
||||
individual_results.push(reason.clone());
|
||||
let _ = writeln!(
|
||||
tool_results,
|
||||
"<tool_result name=\"{}\">\n{reason}\n</tool_result>",
|
||||
call.name
|
||||
);
|
||||
// Record the denial in the shared breaker (the
|
||||
// gate's `[policy-denied]` marker makes it a
|
||||
// hard reject) so a re-issued identical call
|
||||
// halts the turn instead of re-prompting
|
||||
// forever — the normal record path below is
|
||||
// skipped by this `continue`.
|
||||
if let Some(halt) = failure_guard.record(
|
||||
&call.name,
|
||||
&call.arguments.to_string(),
|
||||
false,
|
||||
&reason,
|
||||
) {
|
||||
halt_reason = Some(halt);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let (result, call_succeeded) = if let Some(tool) = tool_opt {
|
||||
let tool_deadline =
|
||||
crate::openhuman::tool_timeout::tool_execution_timeout_duration();
|
||||
let timeout_secs = crate::openhuman::tool_timeout::tool_execution_timeout_secs();
|
||||
let tool_started = std::time::Instant::now();
|
||||
let outcome =
|
||||
tokio::time::timeout(tool_deadline, tool.execute(call.arguments.clone())).await;
|
||||
let elapsed_ms = tool_started.elapsed().as_millis() as u64;
|
||||
let (result_text, success) = match outcome {
|
||||
Ok(Ok(r)) => {
|
||||
let output = r.output();
|
||||
let success = !r.is_error;
|
||||
if success {
|
||||
tracing::debug!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
output_len = output.len(),
|
||||
"[agent_loop] tool succeeded"
|
||||
);
|
||||
let mut scrubbed = scrub_credentials(&output);
|
||||
let (compacted, tj_stats) =
|
||||
crate::openhuman::tokenjuice::compact_tool_output(
|
||||
&call.name,
|
||||
Some(&call.arguments),
|
||||
&scrubbed,
|
||||
Some(0),
|
||||
);
|
||||
if tj_stats.applied {
|
||||
log::debug!(
|
||||
"[agent_loop] tokenjuice applied tool={} rule={} {}->{} bytes",
|
||||
call.name,
|
||||
tj_stats.rule_id,
|
||||
tj_stats.original_bytes,
|
||||
tj_stats.compacted_bytes
|
||||
);
|
||||
scrubbed = compacted;
|
||||
}
|
||||
|
||||
// Per-tool max_result_size_chars cap. When
|
||||
// a tool sets it and the (post-tokenjuice)
|
||||
// body still exceeds the cap, truncate
|
||||
// here and skip the global payload
|
||||
// summarizer for this call — the cap is
|
||||
// fast and deterministic, the summarizer
|
||||
// is the fallback for tools that don't
|
||||
// know their own size budget.
|
||||
let mut hit_per_tool_cap = false;
|
||||
if let Some(cap) = tool.max_result_size_chars() {
|
||||
let char_count = scrubbed.chars().count();
|
||||
if char_count > cap {
|
||||
let truncated: String = scrubbed.chars().take(cap).collect();
|
||||
let dropped = char_count - cap;
|
||||
log::info!(
|
||||
"[agent_loop] per-tool cap applied tool={} cap_chars={} original_chars={} dropped_chars={}",
|
||||
call.name,
|
||||
cap,
|
||||
char_count,
|
||||
dropped,
|
||||
);
|
||||
scrubbed = format!(
|
||||
"{truncated}\n\n[truncated by tool cap: {dropped} more chars not shown]"
|
||||
);
|
||||
hit_per_tool_cap = true;
|
||||
}
|
||||
}
|
||||
|
||||
if !hit_per_tool_cap {
|
||||
if let Some(summarizer) = payload_summarizer {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer intercepting tool={} bytes={}",
|
||||
call.name,
|
||||
scrubbed.len()
|
||||
);
|
||||
match summarizer
|
||||
.maybe_summarize(&call.name, None, &scrubbed)
|
||||
.await
|
||||
{
|
||||
Ok(Some(payload)) => {
|
||||
log::info!(
|
||||
"[agent_loop] payload_summarizer compressed tool={} {}->{} bytes",
|
||||
call.name,
|
||||
payload.original_bytes,
|
||||
payload.summary_bytes
|
||||
);
|
||||
scrubbed = payload.summary;
|
||||
}
|
||||
Ok(None) => {
|
||||
log::debug!(
|
||||
"[agent_loop] payload_summarizer pass-through tool={} bytes={}",
|
||||
call.name,
|
||||
scrubbed.len()
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)",
|
||||
call.name,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
(scrubbed, true)
|
||||
} else {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] tool returned error: {output}"
|
||||
);
|
||||
let scrubbed = scrub_credentials(&output);
|
||||
let (compacted, _) = crate::openhuman::tokenjuice::compact_tool_output(
|
||||
&call.name,
|
||||
Some(&call.arguments),
|
||||
&scrubbed,
|
||||
Some(1),
|
||||
);
|
||||
(format!("Error: {compacted}"), false)
|
||||
}
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
crate::core::observability::report_error(
|
||||
&e,
|
||||
"tool",
|
||||
"execute",
|
||||
&[
|
||||
("tool", call.name.as_str()),
|
||||
("outcome", "failed"),
|
||||
("iteration", &(iteration + 1).to_string()),
|
||||
],
|
||||
);
|
||||
(format!("Error executing {}: {e}", call.name), false)
|
||||
}
|
||||
Err(_) => {
|
||||
let msg = format!(
|
||||
"tool '{}' timed out after {} seconds",
|
||||
call.name, timeout_secs
|
||||
);
|
||||
crate::core::observability::report_error(
|
||||
msg.as_str(),
|
||||
"tool",
|
||||
"execute",
|
||||
&[
|
||||
("tool", call.name.as_str()),
|
||||
("outcome", "timeout"),
|
||||
("timeout_secs", &timeout_secs.to_string()),
|
||||
("iteration", &(iteration + 1).to_string()),
|
||||
],
|
||||
);
|
||||
(
|
||||
format!(
|
||||
"Error: tool '{}' timed out after {} seconds",
|
||||
call.name, timeout_secs
|
||||
),
|
||||
false,
|
||||
)
|
||||
}
|
||||
};
|
||||
if let Some(ref sink) = on_progress {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::ToolCallCompleted {
|
||||
call_id: progress_call_id.clone(),
|
||||
tool_name: call.name.clone(),
|
||||
success,
|
||||
output_chars: result_text.chars().count(),
|
||||
elapsed_ms,
|
||||
iteration: (iteration + 1) as u32,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!("[agent_loop] progress sink closed while emitting ToolCallCompleted: {e}");
|
||||
}
|
||||
}
|
||||
// ── Approval audit after-action row (#2135) ────
|
||||
// Stamp the terminal status onto the same
|
||||
// `pending_approvals` row the gate created before
|
||||
// execution, so the audit trail carries both the
|
||||
// before (approval) and after (executed_at +
|
||||
// outcome). Best-effort: a write failure here is
|
||||
// logged but not propagated to the agent.
|
||||
if let (Some(gate), Some(req_id)) = (
|
||||
approval_gate_for_audit.as_ref(),
|
||||
approval_request_id.as_ref(),
|
||||
) {
|
||||
let exec_outcome = if success {
|
||||
crate::openhuman::approval::ExecutionOutcome::Success
|
||||
} else {
|
||||
crate::openhuman::approval::ExecutionOutcome::Failure
|
||||
};
|
||||
let err_text = if success {
|
||||
None
|
||||
} else {
|
||||
Some(result_text.as_str())
|
||||
};
|
||||
gate.record_execution(req_id, exec_outcome, err_text);
|
||||
}
|
||||
(result_text, success)
|
||||
} else {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] unknown tool requested"
|
||||
);
|
||||
let msg = format!("Unknown tool: {}", call.name);
|
||||
emit_failed_completion(&msg).await;
|
||||
(msg, false)
|
||||
};
|
||||
|
||||
individual_results.push(result.clone());
|
||||
let _ = writeln!(
|
||||
tool_results,
|
||||
"<tool_result name=\"{}\">\n{}\n</tool_result>",
|
||||
call.name, result
|
||||
);
|
||||
|
||||
// Repeated-failure circuit breaker (shared guard) — halt with a root
|
||||
// cause instead of grinding to `max_iterations` on a doomed action.
|
||||
if let Some(reason) = failure_guard.record(
|
||||
&call.name,
|
||||
&call.arguments.to_string(),
|
||||
call_succeeded,
|
||||
&result,
|
||||
) {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
tool = call.name.as_str(),
|
||||
"[agent_loop] circuit breaker tripped — halting with root cause"
|
||||
);
|
||||
halt_reason = Some(reason);
|
||||
}
|
||||
}
|
||||
|
||||
// Add assistant message with tool calls + tool results to history.
|
||||
// Native mode: use JSON-structured messages so convert_messages() can
|
||||
// reconstruct proper OpenAI-format tool_calls and tool result messages.
|
||||
// Prompt mode: use XML-based text format as before.
|
||||
history.push(ChatMessage::assistant(assistant_history_content));
|
||||
if native_tool_calls.is_empty() {
|
||||
history.push(ChatMessage::user(format!("[Tool results]\n{tool_results}")));
|
||||
} else {
|
||||
for (native_call, result) in native_tool_calls.iter().zip(individual_results.iter()) {
|
||||
let tool_msg = serde_json::json!({
|
||||
"tool_call_id": native_call.id,
|
||||
"content": result,
|
||||
});
|
||||
history.push(ChatMessage::tool(tool_msg.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
// Circuit breaker tripped this iteration: return the root-cause summary
|
||||
// as the agent's result instead of looping to `max_iterations`. The
|
||||
// tool results are already in `history` above, so the caller still has
|
||||
// full context if it wants it.
|
||||
if let Some(reason) = halt_reason.take() {
|
||||
// Mirror the normal-completion path: emit TurnCompleted before the
|
||||
// early return, otherwise progress consumers stay "in-flight"
|
||||
// indefinitely when the circuit breaker trips.
|
||||
if let Some(ref sink) = on_progress {
|
||||
if let Err(e) = sink
|
||||
.send(AgentProgress::TurnCompleted {
|
||||
iterations: (iteration + 1) as u32,
|
||||
})
|
||||
.await
|
||||
{
|
||||
log::warn!("[agent_loop] progress sink closed at TurnCompleted: {e}");
|
||||
}
|
||||
}
|
||||
return Ok(reason);
|
||||
}
|
||||
}
|
||||
|
||||
// Return the typed `AgentError::MaxIterationsExceeded` variant (boxed
|
||||
// through `anyhow::Error`) so downstream wrappers — notably
|
||||
// `Agent::run_single` in `harness/session/runtime.rs` — can downcast and
|
||||
// suppress Sentry emission for this deterministic agent-state outcome
|
||||
// (OPENHUMAN-TAURI-99 / -98). The `Display` text is preserved verbatim so
|
||||
// any caller that already inspects the string (UI chat surface, tests)
|
||||
// continues to work.
|
||||
Err(anyhow::Error::new(
|
||||
crate::openhuman::agent::error::AgentError::MaxIterationsExceeded {
|
||||
max: max_iterations,
|
||||
},
|
||||
))
|
||||
let mut tool_source = super::engine::RegistryToolSource::new(
|
||||
tools_registry,
|
||||
extra_tools,
|
||||
visible_tool_names,
|
||||
tool_policy,
|
||||
payload_summarizer,
|
||||
);
|
||||
let progress = super::engine::TurnProgress::new(on_progress);
|
||||
let mut observer = super::engine::NullObserver;
|
||||
let checkpoint = super::engine::ErrorCheckpoint;
|
||||
let parser = super::engine::DefaultParser;
|
||||
super::engine::run_turn_engine(
|
||||
provider,
|
||||
history,
|
||||
&mut tool_source,
|
||||
&progress,
|
||||
&mut observer,
|
||||
&checkpoint,
|
||||
&parser,
|
||||
provider_name,
|
||||
model,
|
||||
temperature,
|
||||
silent,
|
||||
multimodal_config,
|
||||
max_iterations,
|
||||
on_delta,
|
||||
)
|
||||
.await
|
||||
.map(|outcome| outcome.text)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use super::*;
|
||||
use crate::openhuman::inference::provider::traits::ProviderCapabilities;
|
||||
use crate::openhuman::inference::provider::ChatResponse;
|
||||
use crate::openhuman::inference::provider::{ChatRequest, ChatResponse};
|
||||
use crate::openhuman::tools::{ToolResult, ToolScope};
|
||||
use async_trait::async_trait;
|
||||
use parking_lot::Mutex;
|
||||
|
||||
Reference in New Issue
Block a user