mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
feat: 10 infrastructure improvements from zeroclaw (#126)
* refactor: break up agent_loop.rs into four focused modules Split the monolithic 2835-line agent_loop.rs into: - agent_loop.rs (722L): Agent struct, event loop, message dispatch - dispatcher.rs (635L): Agentic tool loop, tool execution, auth detection - commands.rs (484L): System commands, job handlers, heartbeat, summarize - thread_ops.rs (1059L): Thread lifecycle, approval, undo/redo, persistence Each module gets its own impl Agent block. Agent fields changed to pub(super) so sibling modules in the agent package can access them. All 16 existing tests pass in their new locations. Inspired by ZeroClaw's agent module split (agent.rs, loop_.rs, dispatcher.rs, prompt.rs, memory_loader.rs). Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add cost caps and guardrails for autonomous agent spending Daily budget (MAX_COST_PER_DAY_CENTS) and hourly action rate (MAX_ACTIONS_PER_HOUR) limits prevent runaway agents from burning through API credits, especially in daemon/heartbeat modes. - CostGuard with pre-flight check and post-call recording - Sliding window for hourly rate, midnight-UTC daily reset - 80% threshold warning, atomic fast-path for exceeded budget - Wired into dispatcher loop (check before LLM call, record after) Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add circuit breaker on LLM providers Wraps LlmProvider with a Closed/Open/HalfOpen state machine that trips after consecutive transient failures, preventing request storms against a degraded backend. Automatically probes for recovery. - CircuitBreakerProvider implements LlmProvider (drop-in wrapper) - Transient error classification (server, rate-limit, network, auth infra) - Client errors (wrong model, context overflow) don't trip the breaker - Configurable via CIRCUIT_BREAKER_THRESHOLD and CIRCUIT_BREAKER_RECOVERY_SECS - Composes with existing FailoverProvider (circuit breaker wraps failover) - 12 tests covering full state machine and error classification Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add tunnel abstraction for remote access Trait-based tunnel system with lifecycle management (start/stop/health) for exposing the agent to the internet through external tunnel binaries. Five providers: - Cloudflare Tunnel (cloudflared, Zero Trust token auth) - Tailscale (serve for tailnet, funnel for public) - ngrok (with optional custom domain) - Custom (arbitrary command with {host}/{port} placeholders) - None (local-only, no external exposure) Config via TUNNEL_PROVIDER + provider-specific env vars. Extends existing TunnelConfig with optional managed provider alongside the static TUNNEL_URL path. Factory, shared process management, and 37 tests covering all providers and edge cases. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add OS service management (launchd/systemd) Adds `ironclaw service {install,start,stop,status,uninstall}` for running the agent as a background daemon. macOS uses launchd plists under ~/Library/LaunchAgents, Linux uses systemd user units. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add observability trait system with noop, log, and multi backends Introduces an Observer trait for recording agent lifecycle events and metrics, with pluggable backends. The noop backend compiles to zero overhead, log backend uses tracing, and multi fans out to multiple observers. Configured via OBSERVABILITY_BACKEND env var. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add in-memory LLM response cache with TTL and LRU eviction CachedProvider wraps any LlmProvider and caches complete() responses keyed by SHA-256(model + messages). Tool-calling requests are never cached since they trigger side effects. Configurable via RESPONSE_CACHE_ENABLED, RESPONSE_CACHE_TTL_SECS, and RESPONSE_CACHE_MAX_ENTRIES env vars. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add memory hygiene with cadence-gated daily log cleanup Adds workspace::hygiene module that automatically deletes daily log documents older than a configurable retention period (default 30 days). Runs on a 12-hour cadence tracked via a local state file to avoid redundant passes. Best-effort design: failures are logged, never fatal. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add doctor diagnostics command for active health probing Probes external dependencies (Docker, cloudflared, ngrok, tailscale), validates NEAR AI session, checks database connectivity, and verifies workspace directory. Complements the passive `status` command. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: add structured TOML config file support Adds ~/.ironclaw/config.toml as a configuration layer between env vars and database settings. Priority: env var > TOML file > DB > defaults. - `ironclaw config init` generates a commented config.toml from current settings - `ironclaw --config path/to/config.toml` loads a custom config file - Settings.merge_from() only overlays non-default values from the TOML file - `ironclaw config path` now shows TOML file status Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address codex review findings - apply_toml_overlay now returns Result and errors on explicit missing or invalid config paths (was log-only, violating the documented contract that explicit paths are fatal) - custom tunnel url_pattern is now used to filter extracted URLs, not just as a gate for scanning stdout - systemd ExecStart path is now quoted to handle spaces in paths Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review feedback - Cache key now includes max_tokens, temperature, and stop_sequences so different request parameters produce distinct keys - to_cents() uses .trunc() + parse::<u64> instead of f64 intermediary, avoiding precision loss for large values - Tailscale public URL no longer includes local port (serve/funnel expose on standard HTTPS port 443) Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: wire up tunnel lifecycle and fix audit findings Connect the tunnel module to the rest of the application so that setting TUNNEL_PROVIDER actually starts a managed tunnel at boot and stops it on shutdown. Previously create_tunnel() was never called outside tests. Changes: - Expand TunnelSettings with provider credential fields (settings.rs) - TunnelConfig::resolve() falls back to DB settings when env vars unset - Start tunnel at boot, stop on shutdown, show URL in boot screen - Setup wizard collects provider-specific credentials (ngrok, cloudflare, tailscale, custom, static URL) - Fix public_url() returning None under lock contention (SharedUrl) - Fix local_host parameter ignored by cloudflare/ngrok/tailscale - Fix tailscale silent fallback to "localhost" on bad JSON - Fix ngrok globally mutating config via add-authtoken (use env var) - Add 10s timeout to tailscale status --json Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review comments - Document split_whitespace limitation in CustomTunnel doc comment - Remove unnecessary quotes from systemd ExecStart directive Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review feedback (round 3) - doctor: missing libSQL DB on fresh install is Pass, not Fail - service: quote ExecStart path for systemd space handling Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: correct cost guard doc comment (LLM calls, not LLM/tool) Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Claude Opus 4.6 <[email protected]>
This commit is contained in:
Illia Polosukhin
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
436dda0f2f
commit
a158eee1b0
+33
-2127
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,484 @@
|
||||
//! System commands and job handlers for the agent.
|
||||
//!
|
||||
//! Extracted from `agent_loop.rs` to isolate the /help, /model, /status,
|
||||
//! and other command processing from the core agent loop.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use tokio::sync::Mutex;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::agent::session::Session;
|
||||
use crate::agent::submission::SubmissionResult;
|
||||
use crate::agent::{Agent, MessageIntent};
|
||||
use crate::channels::{IncomingMessage, StatusUpdate};
|
||||
use crate::error::Error;
|
||||
use crate::llm::ChatMessage;
|
||||
|
||||
impl Agent {
|
||||
/// Handle job-related intents without turn tracking.
|
||||
pub(super) async fn handle_job_or_command(
|
||||
&self,
|
||||
intent: MessageIntent,
|
||||
message: &IncomingMessage,
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
// Send thinking status for non-trivial operations
|
||||
if let MessageIntent::CreateJob { .. } = &intent {
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::Thinking("Processing...".into()),
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
let response = match intent {
|
||||
MessageIntent::CreateJob {
|
||||
title,
|
||||
description,
|
||||
category,
|
||||
} => {
|
||||
self.handle_create_job(&message.user_id, title, description, category)
|
||||
.await?
|
||||
}
|
||||
MessageIntent::CheckJobStatus { job_id } => {
|
||||
self.handle_check_status(&message.user_id, job_id).await?
|
||||
}
|
||||
MessageIntent::CancelJob { job_id } => {
|
||||
self.handle_cancel_job(&message.user_id, &job_id).await?
|
||||
}
|
||||
MessageIntent::ListJobs { filter } => {
|
||||
self.handle_list_jobs(&message.user_id, filter).await?
|
||||
}
|
||||
MessageIntent::HelpJob { job_id } => {
|
||||
self.handle_help_job(&message.user_id, &job_id).await?
|
||||
}
|
||||
MessageIntent::Command { command, args } => {
|
||||
match self.handle_command(&command, &args).await? {
|
||||
Some(s) => s,
|
||||
None => return Ok(SubmissionResult::Ok { message: None }), // Shutdown signal
|
||||
}
|
||||
}
|
||||
_ => "Unknown intent".to_string(),
|
||||
};
|
||||
Ok(SubmissionResult::response(response))
|
||||
}
|
||||
|
||||
async fn handle_create_job(
|
||||
&self,
|
||||
user_id: &str,
|
||||
title: String,
|
||||
description: String,
|
||||
category: Option<String>,
|
||||
) -> Result<String, Error> {
|
||||
// Create job context
|
||||
let job_id = self
|
||||
.context_manager
|
||||
.create_job_for_user(user_id, &title, &description)
|
||||
.await?;
|
||||
|
||||
// Update category if provided
|
||||
if let Some(cat) = category {
|
||||
self.context_manager
|
||||
.update_context(job_id, |ctx| {
|
||||
ctx.category = Some(cat);
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
// Persist new job to database (fire-and-forget)
|
||||
if let Some(store) = self.store()
|
||||
&& let Ok(ctx) = self.context_manager.get_context(job_id).await
|
||||
{
|
||||
let store = store.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = store.save_job(&ctx).await {
|
||||
tracing::warn!("Failed to persist new job {}: {}", job_id, e);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Schedule for execution
|
||||
self.scheduler.schedule(job_id).await?;
|
||||
|
||||
Ok(format!(
|
||||
"Created job: {}\nID: {}\n\nThe job has been scheduled and is now running.",
|
||||
title, job_id
|
||||
))
|
||||
}
|
||||
|
||||
async fn handle_check_status(
|
||||
&self,
|
||||
user_id: &str,
|
||||
job_id: Option<String>,
|
||||
) -> Result<String, Error> {
|
||||
match job_id {
|
||||
Some(id) => {
|
||||
let uuid = Uuid::parse_str(&id)
|
||||
.map_err(|_| crate::error::JobError::NotFound { id: Uuid::nil() })?;
|
||||
|
||||
let ctx = self.context_manager.get_context(uuid).await?;
|
||||
if ctx.user_id != user_id {
|
||||
return Err(crate::error::JobError::NotFound { id: uuid }.into());
|
||||
}
|
||||
|
||||
Ok(format!(
|
||||
"Job: {}\nStatus: {:?}\nCreated: {}\nStarted: {}\nActual cost: {}",
|
||||
ctx.title,
|
||||
ctx.state,
|
||||
ctx.created_at.format("%Y-%m-%d %H:%M:%S"),
|
||||
ctx.started_at
|
||||
.map(|t| t.format("%Y-%m-%d %H:%M:%S").to_string())
|
||||
.unwrap_or_else(|| "Not started".to_string()),
|
||||
ctx.actual_cost
|
||||
))
|
||||
}
|
||||
None => {
|
||||
// Show summary of all jobs
|
||||
let summary = self.context_manager.summary_for(user_id).await;
|
||||
Ok(format!(
|
||||
"Jobs summary:\n Total: {}\n In Progress: {}\n Completed: {}\n Failed: {}\n Stuck: {}",
|
||||
summary.total,
|
||||
summary.in_progress,
|
||||
summary.completed,
|
||||
summary.failed,
|
||||
summary.stuck
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_cancel_job(&self, user_id: &str, job_id: &str) -> Result<String, Error> {
|
||||
let uuid = Uuid::parse_str(job_id)
|
||||
.map_err(|_| crate::error::JobError::NotFound { id: Uuid::nil() })?;
|
||||
|
||||
let ctx = self.context_manager.get_context(uuid).await?;
|
||||
if ctx.user_id != user_id {
|
||||
return Err(crate::error::JobError::NotFound { id: uuid }.into());
|
||||
}
|
||||
|
||||
self.scheduler.stop(uuid).await?;
|
||||
|
||||
Ok(format!("Job {} has been cancelled.", job_id))
|
||||
}
|
||||
|
||||
async fn handle_list_jobs(
|
||||
&self,
|
||||
user_id: &str,
|
||||
_filter: Option<String>,
|
||||
) -> Result<String, Error> {
|
||||
let jobs = self.context_manager.all_jobs_for(user_id).await;
|
||||
|
||||
if jobs.is_empty() {
|
||||
return Ok("No jobs found.".to_string());
|
||||
}
|
||||
|
||||
let mut output = String::from("Jobs:\n");
|
||||
for job_id in jobs {
|
||||
if let Ok(ctx) = self.context_manager.get_context(job_id).await
|
||||
&& ctx.user_id == user_id
|
||||
{
|
||||
output.push_str(&format!(" {} - {} ({:?})\n", job_id, ctx.title, ctx.state));
|
||||
}
|
||||
}
|
||||
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
async fn handle_help_job(&self, user_id: &str, job_id: &str) -> Result<String, Error> {
|
||||
let uuid = Uuid::parse_str(job_id)
|
||||
.map_err(|_| crate::error::JobError::NotFound { id: Uuid::nil() })?;
|
||||
|
||||
let ctx = self.context_manager.get_context(uuid).await?;
|
||||
if ctx.user_id != user_id {
|
||||
return Err(crate::error::JobError::NotFound { id: uuid }.into());
|
||||
}
|
||||
|
||||
if ctx.state == crate::context::JobState::Stuck {
|
||||
// Attempt recovery
|
||||
self.context_manager
|
||||
.update_context(uuid, |ctx| ctx.attempt_recovery())
|
||||
.await?
|
||||
.map_err(|s| crate::error::JobError::ContextError {
|
||||
id: uuid,
|
||||
reason: s,
|
||||
})?;
|
||||
|
||||
// Reschedule
|
||||
self.scheduler.schedule(uuid).await?;
|
||||
|
||||
Ok(format!(
|
||||
"Job {} was stuck. Attempting recovery (attempt #{}).",
|
||||
job_id,
|
||||
ctx.repair_attempts + 1
|
||||
))
|
||||
} else {
|
||||
Ok(format!(
|
||||
"Job {} is not stuck (current state: {:?}). No help needed.",
|
||||
job_id, ctx.state
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// Trigger a manual heartbeat check.
|
||||
pub(super) async fn process_heartbeat(&self) -> Result<SubmissionResult, Error> {
|
||||
let Some(workspace) = self.workspace() else {
|
||||
return Ok(SubmissionResult::error(
|
||||
"Heartbeat requires a workspace (database must be connected).",
|
||||
));
|
||||
};
|
||||
|
||||
let runner = crate::agent::HeartbeatRunner::new(
|
||||
crate::agent::HeartbeatConfig::default(),
|
||||
workspace.clone(),
|
||||
self.llm().clone(),
|
||||
);
|
||||
|
||||
match runner.check_heartbeat().await {
|
||||
crate::agent::HeartbeatResult::Ok => Ok(SubmissionResult::ok_with_message(
|
||||
"Heartbeat: all clear, nothing needs attention.",
|
||||
)),
|
||||
crate::agent::HeartbeatResult::NeedsAttention(msg) => Ok(SubmissionResult::response(
|
||||
format!("Heartbeat findings:\n\n{}", msg),
|
||||
)),
|
||||
crate::agent::HeartbeatResult::Skipped => Ok(SubmissionResult::ok_with_message(
|
||||
"Heartbeat skipped: no HEARTBEAT.md checklist found in workspace.",
|
||||
)),
|
||||
crate::agent::HeartbeatResult::Failed(err) => Ok(SubmissionResult::error(format!(
|
||||
"Heartbeat failed: {}",
|
||||
err
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Summarize the current thread's conversation.
|
||||
pub(super) async fn process_summarize(
|
||||
&self,
|
||||
session: Arc<Mutex<Session>>,
|
||||
thread_id: Uuid,
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
let messages = {
|
||||
let sess = session.lock().await;
|
||||
let thread = sess
|
||||
.threads
|
||||
.get(&thread_id)
|
||||
.ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
|
||||
thread.messages()
|
||||
};
|
||||
|
||||
if messages.is_empty() {
|
||||
return Ok(SubmissionResult::ok_with_message(
|
||||
"Nothing to summarize (empty thread).",
|
||||
));
|
||||
}
|
||||
|
||||
// Build a summary prompt with the conversation
|
||||
let mut context = Vec::new();
|
||||
context.push(ChatMessage::system(
|
||||
"Summarize the conversation so far in 3-5 concise bullet points. \
|
||||
Focus on decisions made, actions taken, and key outcomes. \
|
||||
Be brief and factual.",
|
||||
));
|
||||
// Include the conversation messages (truncate to last 20 to avoid context overflow)
|
||||
let start = if messages.len() > 20 {
|
||||
messages.len() - 20
|
||||
} else {
|
||||
0
|
||||
};
|
||||
context.extend_from_slice(&messages[start..]);
|
||||
context.push(ChatMessage::user("Summarize this conversation."));
|
||||
|
||||
let request = crate::llm::CompletionRequest::new(context)
|
||||
.with_max_tokens(512)
|
||||
.with_temperature(0.3);
|
||||
|
||||
match self.llm().complete(request).await {
|
||||
Ok(response) => Ok(SubmissionResult::response(format!(
|
||||
"Thread Summary:\n\n{}",
|
||||
response.content.trim()
|
||||
))),
|
||||
Err(e) => Ok(SubmissionResult::error(format!("Summarize failed: {}", e))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Suggest next steps based on the current thread.
|
||||
pub(super) async fn process_suggest(
|
||||
&self,
|
||||
session: Arc<Mutex<Session>>,
|
||||
thread_id: Uuid,
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
let messages = {
|
||||
let sess = session.lock().await;
|
||||
let thread = sess
|
||||
.threads
|
||||
.get(&thread_id)
|
||||
.ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
|
||||
thread.messages()
|
||||
};
|
||||
|
||||
if messages.is_empty() {
|
||||
return Ok(SubmissionResult::ok_with_message(
|
||||
"Nothing to suggest from (empty thread).",
|
||||
));
|
||||
}
|
||||
|
||||
let mut context = Vec::new();
|
||||
context.push(ChatMessage::system(
|
||||
"Based on the conversation so far, suggest 2-4 concrete next steps the user could take. \
|
||||
Be actionable and specific. Format as a numbered list.",
|
||||
));
|
||||
let start = if messages.len() > 20 {
|
||||
messages.len() - 20
|
||||
} else {
|
||||
0
|
||||
};
|
||||
context.extend_from_slice(&messages[start..]);
|
||||
context.push(ChatMessage::user("What should I do next?"));
|
||||
|
||||
let request = crate::llm::CompletionRequest::new(context)
|
||||
.with_max_tokens(512)
|
||||
.with_temperature(0.5);
|
||||
|
||||
match self.llm().complete(request).await {
|
||||
Ok(response) => Ok(SubmissionResult::response(format!(
|
||||
"Suggested Next Steps:\n\n{}",
|
||||
response.content.trim()
|
||||
))),
|
||||
Err(e) => Ok(SubmissionResult::error(format!("Suggest failed: {}", e))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle system commands that bypass thread-state checks entirely.
|
||||
pub(super) async fn handle_system_command(
|
||||
&self,
|
||||
command: &str,
|
||||
args: &[String],
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
match command {
|
||||
"help" => Ok(SubmissionResult::response(concat!(
|
||||
"System:\n",
|
||||
" /help Show this help\n",
|
||||
" /model [name] Show or switch the active model\n",
|
||||
" /version Show version info\n",
|
||||
" /tools List available tools\n",
|
||||
" /debug Toggle debug mode\n",
|
||||
" /ping Connectivity check\n",
|
||||
"\n",
|
||||
"Jobs:\n",
|
||||
" /job <desc> Create a new job\n",
|
||||
" /status [id] Check job status\n",
|
||||
" /cancel <id> Cancel a job\n",
|
||||
" /list List all jobs\n",
|
||||
"\n",
|
||||
"Session:\n",
|
||||
" /undo Undo last turn\n",
|
||||
" /redo Redo undone turn\n",
|
||||
" /compact Compress context window\n",
|
||||
" /clear Clear current thread\n",
|
||||
" /interrupt Stop current operation\n",
|
||||
" /new New conversation thread\n",
|
||||
" /thread <id> Switch to thread\n",
|
||||
" /resume <id> Resume from checkpoint\n",
|
||||
"\n",
|
||||
"Agent:\n",
|
||||
" /heartbeat Run heartbeat check\n",
|
||||
" /summarize Summarize current thread\n",
|
||||
" /suggest Suggest next steps\n",
|
||||
"\n",
|
||||
" /quit Exit",
|
||||
))),
|
||||
|
||||
"ping" => Ok(SubmissionResult::response("pong!")),
|
||||
|
||||
"version" => Ok(SubmissionResult::response(format!(
|
||||
"{} v{}",
|
||||
env!("CARGO_PKG_NAME"),
|
||||
env!("CARGO_PKG_VERSION")
|
||||
))),
|
||||
|
||||
"tools" => {
|
||||
let tools = self.tools().list().await;
|
||||
Ok(SubmissionResult::response(format!(
|
||||
"Available tools: {}",
|
||||
tools.join(", ")
|
||||
)))
|
||||
}
|
||||
|
||||
"debug" => {
|
||||
// Debug toggle is handled client-side in the REPL.
|
||||
// For non-REPL channels, just acknowledge.
|
||||
Ok(SubmissionResult::ok_with_message(
|
||||
"Debug toggle is handled by your client.",
|
||||
))
|
||||
}
|
||||
|
||||
"model" => {
|
||||
if args.is_empty() {
|
||||
// Show current model
|
||||
let name = self.llm().active_model_name();
|
||||
Ok(SubmissionResult::response(format!(
|
||||
"Active model: {}",
|
||||
name
|
||||
)))
|
||||
} else {
|
||||
let requested = &args[0];
|
||||
|
||||
// Validate the model exists
|
||||
match self.llm().list_models().await {
|
||||
Ok(models) if !models.is_empty() => {
|
||||
if !models.iter().any(|m| m == requested) {
|
||||
return Ok(SubmissionResult::error(format!(
|
||||
"Unknown model: {}. Available models:\n {}",
|
||||
requested,
|
||||
models.join("\n ")
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(_) => {
|
||||
// Empty model list, can't validate but try anyway
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Could not fetch model list for validation: {}", e);
|
||||
// Proceed anyway, the provider will error on the next call if invalid
|
||||
}
|
||||
}
|
||||
|
||||
match self.llm().set_model(requested) {
|
||||
Ok(()) => Ok(SubmissionResult::response(format!(
|
||||
"Switched model to: {}",
|
||||
requested
|
||||
))),
|
||||
Err(e) => Ok(SubmissionResult::error(format!(
|
||||
"Failed to switch model: {}",
|
||||
e
|
||||
))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_ => Ok(SubmissionResult::error(format!(
|
||||
"Unknown command: {}. Try /help",
|
||||
command
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle legacy command routing from the Router (job commands that go through
|
||||
/// process_user_input -> router -> handle_job_or_command -> here).
|
||||
pub(super) async fn handle_command(
|
||||
&self,
|
||||
command: &str,
|
||||
args: &[String],
|
||||
) -> Result<Option<String>, Error> {
|
||||
// System commands are now handled directly via Submission::SystemCommand,
|
||||
// but the router may still send us unknown /commands.
|
||||
match self.handle_system_command(command, args).await? {
|
||||
SubmissionResult::Response { content } => Ok(Some(content)),
|
||||
SubmissionResult::Ok { message } => Ok(message),
|
||||
SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
|
||||
_ => Ok(None),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,339 @@
|
||||
//! Cost enforcement guardrails for the agent.
|
||||
//!
|
||||
//! Tracks LLM spending and action rates, enforcing configurable limits
|
||||
//! to prevent runaway agents from burning through API credits. Especially
|
||||
//! important for daemon/heartbeat modes where the agent acts autonomously.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Instant;
|
||||
|
||||
use rust_decimal::Decimal;
|
||||
use rust_decimal_macros::dec;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use crate::llm::costs;
|
||||
|
||||
/// Configuration for cost guardrails.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct CostGuardConfig {
|
||||
/// Maximum spend per day in cents (e.g. 10000 = $100). None = unlimited.
|
||||
pub max_cost_per_day_cents: Option<u64>,
|
||||
/// Maximum LLM calls per hour. None = unlimited.
|
||||
pub max_actions_per_hour: Option<u64>,
|
||||
}
|
||||
|
||||
/// Error returned when a cost limit is exceeded.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum CostLimitExceeded {
|
||||
/// Daily spending cap reached.
|
||||
DailyBudget { spent_cents: u64, limit_cents: u64 },
|
||||
/// Hourly action rate limit reached.
|
||||
HourlyRate { actions: u64, limit: u64 },
|
||||
}
|
||||
|
||||
impl std::fmt::Display for CostLimitExceeded {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::DailyBudget {
|
||||
spent_cents,
|
||||
limit_cents,
|
||||
} => write!(
|
||||
f,
|
||||
"Daily cost limit exceeded: spent ${:.2} of ${:.2} allowed",
|
||||
*spent_cents as f64 / 100.0,
|
||||
*limit_cents as f64 / 100.0
|
||||
),
|
||||
Self::HourlyRate { actions, limit } => write!(
|
||||
f,
|
||||
"Hourly action limit exceeded: {} actions of {} allowed per hour",
|
||||
actions, limit
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Tracks costs and action rates, enforcing configurable limits.
|
||||
///
|
||||
/// Thread-safe; designed to be shared via `Arc<CostGuard>`.
|
||||
pub struct CostGuard {
|
||||
config: CostGuardConfig,
|
||||
|
||||
/// Running cost total for the current day (in USD, not cents).
|
||||
daily_cost: Mutex<DailyCost>,
|
||||
|
||||
/// Sliding window of action timestamps for rate limiting.
|
||||
action_window: Mutex<VecDeque<Instant>>,
|
||||
|
||||
/// Flag set when daily budget is exceeded to short-circuit checks.
|
||||
budget_exceeded: AtomicBool,
|
||||
}
|
||||
|
||||
struct DailyCost {
|
||||
total: Decimal,
|
||||
/// Day boundary (midnight UTC) for resetting the counter.
|
||||
reset_date: chrono::NaiveDate,
|
||||
}
|
||||
|
||||
impl CostGuard {
|
||||
pub fn new(config: CostGuardConfig) -> Self {
|
||||
Self {
|
||||
config,
|
||||
daily_cost: Mutex::new(DailyCost {
|
||||
total: Decimal::ZERO,
|
||||
reset_date: chrono::Utc::now().date_naive(),
|
||||
}),
|
||||
action_window: Mutex::new(VecDeque::new()),
|
||||
budget_exceeded: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
/// Check whether the next action is allowed under the configured limits.
|
||||
///
|
||||
/// Call this BEFORE making an LLM call. Does NOT record the action yet,
|
||||
/// call `record_action` after the action completes.
|
||||
pub async fn check_allowed(&self) -> Result<(), CostLimitExceeded> {
|
||||
// Fast path: if budget already blown, skip the lock
|
||||
if self.budget_exceeded.load(Ordering::Relaxed) {
|
||||
let daily = self.daily_cost.lock().await;
|
||||
let spent_cents = to_cents(daily.total);
|
||||
return Err(CostLimitExceeded::DailyBudget {
|
||||
spent_cents,
|
||||
limit_cents: self.config.max_cost_per_day_cents.unwrap_or(0),
|
||||
});
|
||||
}
|
||||
|
||||
// Check daily budget
|
||||
if let Some(limit_cents) = self.config.max_cost_per_day_cents {
|
||||
let daily = self.daily_cost.lock().await;
|
||||
let spent_cents = to_cents(daily.total);
|
||||
if spent_cents >= limit_cents {
|
||||
self.budget_exceeded.store(true, Ordering::Relaxed);
|
||||
return Err(CostLimitExceeded::DailyBudget {
|
||||
spent_cents,
|
||||
limit_cents,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Check hourly rate
|
||||
if let Some(limit) = self.config.max_actions_per_hour {
|
||||
let mut window = self.action_window.lock().await;
|
||||
let cutoff = Instant::now() - std::time::Duration::from_secs(3600);
|
||||
// Drain expired entries
|
||||
while window.front().is_some_and(|t| *t < cutoff) {
|
||||
window.pop_front();
|
||||
}
|
||||
let count = window.len() as u64;
|
||||
if count >= limit {
|
||||
return Err(CostLimitExceeded::HourlyRate {
|
||||
actions: count,
|
||||
limit,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Record a completed LLM action: its token costs and the action timestamp.
|
||||
///
|
||||
/// Call this AFTER an LLM call completes so that costs are tracked.
|
||||
pub async fn record_llm_call(
|
||||
&self,
|
||||
model: &str,
|
||||
input_tokens: u32,
|
||||
output_tokens: u32,
|
||||
) -> Decimal {
|
||||
let (input_rate, output_rate) =
|
||||
costs::model_cost(model).unwrap_or_else(costs::default_cost);
|
||||
let cost =
|
||||
input_rate * Decimal::from(input_tokens) + output_rate * Decimal::from(output_tokens);
|
||||
|
||||
// Update daily cost (reset if new day)
|
||||
{
|
||||
let mut daily = self.daily_cost.lock().await;
|
||||
let today = chrono::Utc::now().date_naive();
|
||||
if today != daily.reset_date {
|
||||
daily.total = Decimal::ZERO;
|
||||
daily.reset_date = today;
|
||||
self.budget_exceeded.store(false, Ordering::Relaxed);
|
||||
tracing::info!("Cost guard: daily counter reset for {}", today);
|
||||
}
|
||||
daily.total += cost;
|
||||
|
||||
// Check if we just crossed the threshold
|
||||
if let Some(limit_cents) = self.config.max_cost_per_day_cents {
|
||||
let spent_cents = to_cents(daily.total);
|
||||
if spent_cents >= limit_cents {
|
||||
self.budget_exceeded.store(true, Ordering::Relaxed);
|
||||
tracing::warn!(
|
||||
"Daily cost limit reached: ${:.2} of ${:.2}",
|
||||
daily.total,
|
||||
Decimal::from(limit_cents) / dec!(100)
|
||||
);
|
||||
}
|
||||
// Warn at 80% threshold
|
||||
let warn_threshold = limit_cents * 80 / 100;
|
||||
if spent_cents >= warn_threshold && spent_cents < limit_cents {
|
||||
tracing::warn!(
|
||||
"Approaching daily cost limit: ${:.2} of ${:.2} ({}%)",
|
||||
daily.total,
|
||||
Decimal::from(limit_cents) / dec!(100),
|
||||
spent_cents * 100 / limit_cents
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Record action in sliding window
|
||||
{
|
||||
let mut window = self.action_window.lock().await;
|
||||
window.push_back(Instant::now());
|
||||
}
|
||||
|
||||
cost
|
||||
}
|
||||
|
||||
/// Current daily spend in USD (as Decimal).
|
||||
pub async fn daily_spend(&self) -> Decimal {
|
||||
let daily = self.daily_cost.lock().await;
|
||||
let today = chrono::Utc::now().date_naive();
|
||||
if today != daily.reset_date {
|
||||
Decimal::ZERO
|
||||
} else {
|
||||
daily.total
|
||||
}
|
||||
}
|
||||
|
||||
/// Number of actions in the current hourly window.
|
||||
pub async fn actions_this_hour(&self) -> u64 {
|
||||
let mut window = self.action_window.lock().await;
|
||||
let cutoff = Instant::now() - std::time::Duration::from_secs(3600);
|
||||
while window.front().is_some_and(|t| *t < cutoff) {
|
||||
window.pop_front();
|
||||
}
|
||||
window.len() as u64
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a Decimal USD amount to whole cents (truncated).
|
||||
fn to_cents(usd: Decimal) -> u64 {
|
||||
let cents = (usd * dec!(100)).trunc();
|
||||
cents.to_string().parse::<u64>().unwrap_or(0)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_unlimited_allows_everything() {
|
||||
let guard = CostGuard::new(CostGuardConfig::default());
|
||||
|
||||
// No limits set, should always be allowed
|
||||
assert!(guard.check_allowed().await.is_ok());
|
||||
|
||||
// Record a big call, still allowed
|
||||
guard.record_llm_call("gpt-4o", 100_000, 100_000).await;
|
||||
assert!(guard.check_allowed().await.is_ok());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_daily_budget_enforcement() {
|
||||
let guard = CostGuard::new(CostGuardConfig {
|
||||
max_cost_per_day_cents: Some(1), // $0.01 limit
|
||||
max_actions_per_hour: None,
|
||||
});
|
||||
|
||||
// First call allowed
|
||||
assert!(guard.check_allowed().await.is_ok());
|
||||
|
||||
// Record a call that costs more than $0.01
|
||||
// gpt-4o: input=$0.0000025/tok, output=$0.00001/tok
|
||||
// 10000 input + 10000 output = $0.025 + $0.10 = $0.125
|
||||
guard.record_llm_call("gpt-4o", 10_000, 10_000).await;
|
||||
|
||||
// Now should be blocked
|
||||
let result = guard.check_allowed().await;
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
CostLimitExceeded::DailyBudget { limit_cents, .. } => {
|
||||
assert_eq!(limit_cents, 1);
|
||||
}
|
||||
other => panic!("Expected DailyBudget, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hourly_rate_enforcement() {
|
||||
let guard = CostGuard::new(CostGuardConfig {
|
||||
max_cost_per_day_cents: None,
|
||||
max_actions_per_hour: Some(3),
|
||||
});
|
||||
|
||||
// First 3 actions allowed
|
||||
for _ in 0..3 {
|
||||
assert!(guard.check_allowed().await.is_ok());
|
||||
guard.record_llm_call("gpt-4o", 10, 10).await;
|
||||
}
|
||||
|
||||
// 4th should be blocked
|
||||
let result = guard.check_allowed().await;
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
CostLimitExceeded::HourlyRate { actions, limit } => {
|
||||
assert_eq!(actions, 3);
|
||||
assert_eq!(limit, 3);
|
||||
}
|
||||
other => panic!("Expected HourlyRate, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_daily_spend_tracking() {
|
||||
let guard = CostGuard::new(CostGuardConfig::default());
|
||||
|
||||
assert_eq!(guard.daily_spend().await, Decimal::ZERO);
|
||||
|
||||
let cost = guard.record_llm_call("gpt-4o", 1000, 500).await;
|
||||
assert!(cost > Decimal::ZERO);
|
||||
assert_eq!(guard.daily_spend().await, cost);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_actions_this_hour() {
|
||||
let guard = CostGuard::new(CostGuardConfig::default());
|
||||
|
||||
assert_eq!(guard.actions_this_hour().await, 0);
|
||||
|
||||
guard.record_llm_call("gpt-4o", 10, 10).await;
|
||||
guard.record_llm_call("gpt-4o", 10, 10).await;
|
||||
|
||||
assert_eq!(guard.actions_this_hour().await, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_to_cents() {
|
||||
assert_eq!(to_cents(dec!(1.50)), 150);
|
||||
assert_eq!(to_cents(dec!(0.01)), 1);
|
||||
assert_eq!(to_cents(Decimal::ZERO), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cost_limit_display() {
|
||||
let budget = CostLimitExceeded::DailyBudget {
|
||||
spent_cents: 1050,
|
||||
limit_cents: 1000,
|
||||
};
|
||||
assert!(budget.to_string().contains("$10.50"));
|
||||
assert!(budget.to_string().contains("$10.00"));
|
||||
|
||||
let rate = CostLimitExceeded::HourlyRate {
|
||||
actions: 101,
|
||||
limit: 100,
|
||||
};
|
||||
assert!(rate.to_string().contains("101 actions"));
|
||||
assert!(rate.to_string().contains("100 allowed"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,635 @@
|
||||
//! Tool dispatch logic for the agent.
|
||||
//!
|
||||
//! Extracted from `agent_loop.rs` to keep the core agentic tool execution
|
||||
//! loop (LLM call -> tool calls -> repeat) in its own focused module.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use tokio::sync::Mutex;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::agent::Agent;
|
||||
use crate::agent::session::{PendingApproval, Session, ThreadState};
|
||||
use crate::channels::{IncomingMessage, StatusUpdate};
|
||||
use crate::context::JobContext;
|
||||
use crate::error::Error;
|
||||
use crate::llm::{ChatMessage, Reasoning, ReasoningContext, RespondResult};
|
||||
|
||||
/// Result of the agentic loop execution.
|
||||
pub(super) enum AgenticLoopResult {
|
||||
/// Completed with a response.
|
||||
Response(String),
|
||||
/// A tool requires approval before continuing.
|
||||
NeedApproval {
|
||||
/// The pending approval request to store.
|
||||
pending: PendingApproval,
|
||||
},
|
||||
}
|
||||
|
||||
impl Agent {
|
||||
/// Run the agentic loop: call LLM, execute tools, repeat until text response.
|
||||
///
|
||||
/// Returns `AgenticLoopResult::Response` on completion, or
|
||||
/// `AgenticLoopResult::NeedApproval` if a tool requires user approval.
|
||||
///
|
||||
/// When `resume_after_tool` is true the loop already knows a tool was
|
||||
/// executed earlier in this turn (e.g. an approved tool), so it won't
|
||||
/// force the LLM to use tools if it responds with text.
|
||||
pub(super) async fn run_agentic_loop(
|
||||
&self,
|
||||
message: &IncomingMessage,
|
||||
session: Arc<Mutex<Session>>,
|
||||
thread_id: Uuid,
|
||||
initial_messages: Vec<ChatMessage>,
|
||||
resume_after_tool: bool,
|
||||
) -> Result<AgenticLoopResult, Error> {
|
||||
// Load workspace system prompt (identity files: AGENTS.md, SOUL.md, etc.)
|
||||
let system_prompt = if let Some(ws) = self.workspace() {
|
||||
match ws.system_prompt().await {
|
||||
Ok(prompt) if !prompt.is_empty() => Some(prompt),
|
||||
Ok(_) => None,
|
||||
Err(e) => {
|
||||
tracing::debug!("Could not load workspace system prompt: {}", e);
|
||||
None
|
||||
}
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let mut reasoning = Reasoning::new(self.llm().clone(), self.safety().clone());
|
||||
if let Some(prompt) = system_prompt {
|
||||
reasoning = reasoning.with_system_prompt(prompt);
|
||||
}
|
||||
|
||||
// Build context with messages that we'll mutate during the loop
|
||||
let mut context_messages = initial_messages;
|
||||
|
||||
// Create a JobContext for tool execution (chat doesn't have a real job)
|
||||
let job_ctx = JobContext::with_user(&message.user_id, "chat", "Interactive chat session");
|
||||
|
||||
const MAX_TOOL_ITERATIONS: usize = 10;
|
||||
let mut iteration = 0;
|
||||
let mut tools_executed = resume_after_tool;
|
||||
|
||||
loop {
|
||||
iteration += 1;
|
||||
if iteration > MAX_TOOL_ITERATIONS {
|
||||
return Err(crate::error::LlmError::InvalidResponse {
|
||||
provider: "agent".to_string(),
|
||||
reason: format!("Exceeded maximum tool iterations ({})", MAX_TOOL_ITERATIONS),
|
||||
}
|
||||
.into());
|
||||
}
|
||||
|
||||
// Check if interrupted
|
||||
{
|
||||
let sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get(&thread_id)
|
||||
&& thread.state == ThreadState::Interrupted
|
||||
{
|
||||
return Err(crate::error::JobError::ContextError {
|
||||
id: thread_id,
|
||||
reason: "Interrupted".to_string(),
|
||||
}
|
||||
.into());
|
||||
}
|
||||
}
|
||||
|
||||
// Enforce cost guardrails before the LLM call
|
||||
if let Err(limit) = self.cost_guard().check_allowed().await {
|
||||
return Err(crate::error::LlmError::InvalidResponse {
|
||||
provider: "agent".to_string(),
|
||||
reason: limit.to_string(),
|
||||
}
|
||||
.into());
|
||||
}
|
||||
|
||||
// Refresh tool definitions each iteration so newly built tools become visible
|
||||
let tool_defs = self.tools().tool_definitions().await;
|
||||
|
||||
// Call LLM with current context
|
||||
let context = ReasoningContext::new()
|
||||
.with_messages(context_messages.clone())
|
||||
.with_tools(tool_defs)
|
||||
.with_metadata({
|
||||
let mut m = std::collections::HashMap::new();
|
||||
m.insert("thread_id".to_string(), thread_id.to_string());
|
||||
m
|
||||
});
|
||||
|
||||
let output = reasoning.respond_with_tools(&context).await?;
|
||||
|
||||
// Record cost and track token usage
|
||||
let model_name = self.llm().active_model_name();
|
||||
let call_cost = self
|
||||
.cost_guard()
|
||||
.record_llm_call(
|
||||
&model_name,
|
||||
output.usage.input_tokens,
|
||||
output.usage.output_tokens,
|
||||
)
|
||||
.await;
|
||||
tracing::debug!(
|
||||
"LLM call used {} input + {} output tokens (${:.6})",
|
||||
output.usage.input_tokens,
|
||||
output.usage.output_tokens,
|
||||
call_cost,
|
||||
);
|
||||
|
||||
match output.result {
|
||||
RespondResult::Text(text) => {
|
||||
// If no tools have been executed yet, prompt the LLM to use tools
|
||||
// This handles the case where the model explains what it will do
|
||||
// instead of actually calling tools
|
||||
if !tools_executed && iteration < 3 {
|
||||
tracing::debug!(
|
||||
"No tools executed yet (iteration {}), prompting for tool use",
|
||||
iteration
|
||||
);
|
||||
context_messages.push(ChatMessage::assistant(&text));
|
||||
context_messages.push(ChatMessage::user(
|
||||
"Please proceed and use the available tools to complete this task.",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
|
||||
// Tools have been executed or we've tried multiple times, return response
|
||||
return Ok(AgenticLoopResult::Response(text));
|
||||
}
|
||||
RespondResult::ToolCalls {
|
||||
tool_calls,
|
||||
content,
|
||||
} => {
|
||||
tools_executed = true;
|
||||
|
||||
// Add the assistant message with tool_calls to context.
|
||||
// OpenAI protocol requires this before tool-result messages.
|
||||
context_messages.push(ChatMessage::assistant_with_tool_calls(
|
||||
content,
|
||||
tool_calls.clone(),
|
||||
));
|
||||
|
||||
// Execute tools and add results to context
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::Thinking(format!(
|
||||
"Executing {} tool(s)...",
|
||||
tool_calls.len()
|
||||
)),
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
|
||||
// Record tool calls in the thread
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id)
|
||||
&& let Some(turn) = thread.last_turn_mut()
|
||||
{
|
||||
for tc in &tool_calls {
|
||||
turn.record_tool_call(&tc.name, tc.arguments.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Execute each tool (with approval checking and hook interception)
|
||||
for mut tc in tool_calls {
|
||||
// Check if tool requires approval
|
||||
if let Some(tool) = self.tools().get(&tc.name).await
|
||||
&& tool.requires_approval()
|
||||
{
|
||||
// Check if auto-approved for this session
|
||||
let mut is_auto_approved = {
|
||||
let sess = session.lock().await;
|
||||
sess.is_tool_auto_approved(&tc.name)
|
||||
};
|
||||
|
||||
// Override auto-approval for destructive parameters
|
||||
// (e.g. `rm -rf`, `git push --force` in shell commands).
|
||||
if is_auto_approved && tool.requires_approval_for(&tc.arguments) {
|
||||
tracing::info!(
|
||||
tool = %tc.name,
|
||||
"Parameters require explicit approval despite auto-approve"
|
||||
);
|
||||
is_auto_approved = false;
|
||||
}
|
||||
|
||||
if !is_auto_approved {
|
||||
// Need approval - store pending request and return
|
||||
let pending = PendingApproval {
|
||||
request_id: Uuid::new_v4(),
|
||||
tool_name: tc.name.clone(),
|
||||
parameters: tc.arguments.clone(),
|
||||
description: tool.description().to_string(),
|
||||
tool_call_id: tc.id.clone(),
|
||||
context_messages: context_messages.clone(),
|
||||
};
|
||||
|
||||
return Ok(AgenticLoopResult::NeedApproval { pending });
|
||||
}
|
||||
}
|
||||
|
||||
// Hook: BeforeToolCall — allow hooks to modify or reject tool calls
|
||||
{
|
||||
let event = crate::hooks::HookEvent::ToolCall {
|
||||
tool_name: tc.name.clone(),
|
||||
parameters: tc.arguments.clone(),
|
||||
user_id: message.user_id.clone(),
|
||||
context: "chat".to_string(),
|
||||
};
|
||||
match self.hooks().run(&event).await {
|
||||
Err(crate::hooks::HookError::Rejected { reason }) => {
|
||||
context_messages.push(ChatMessage::tool_result(
|
||||
&tc.id,
|
||||
&tc.name,
|
||||
format!("Tool call rejected by hook: {}", reason),
|
||||
));
|
||||
continue;
|
||||
}
|
||||
Err(err) => {
|
||||
context_messages.push(ChatMessage::tool_result(
|
||||
&tc.id,
|
||||
&tc.name,
|
||||
format!("Tool call blocked by hook policy: {}", err),
|
||||
));
|
||||
continue;
|
||||
}
|
||||
Ok(crate::hooks::HookOutcome::Continue {
|
||||
modified: Some(new_params),
|
||||
}) => match serde_json::from_str(&new_params) {
|
||||
Ok(parsed) => tc.arguments = parsed,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
tool = %tc.name,
|
||||
"Hook returned non-JSON modification for ToolCall, ignoring: {}",
|
||||
e
|
||||
);
|
||||
}
|
||||
},
|
||||
_ => {} // Continue, fail-open errors already logged
|
||||
}
|
||||
}
|
||||
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::ToolStarted {
|
||||
name: tc.name.clone(),
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
|
||||
let tool_result = self
|
||||
.execute_chat_tool(&tc.name, &tc.arguments, &job_ctx)
|
||||
.await;
|
||||
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::ToolCompleted {
|
||||
name: tc.name.clone(),
|
||||
success: tool_result.is_ok(),
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
|
||||
if let Ok(ref output) = tool_result
|
||||
&& !output.is_empty()
|
||||
{
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::ToolResult {
|
||||
name: tc.name.clone(),
|
||||
preview: output.clone(),
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Record result in thread
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id)
|
||||
&& let Some(turn) = thread.last_turn_mut()
|
||||
{
|
||||
match &tool_result {
|
||||
Ok(output) => {
|
||||
turn.record_tool_result(serde_json::json!(output));
|
||||
}
|
||||
Err(e) => {
|
||||
turn.record_tool_error(e.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If tool_auth returned awaiting_token, enter auth mode
|
||||
// and short-circuit: return the instructions directly so
|
||||
// the LLM doesn't get a chance to hallucinate tool calls.
|
||||
if let Some((ext_name, instructions)) =
|
||||
detect_auth_awaiting(&tc.name, &tool_result)
|
||||
{
|
||||
let auth_data = parse_auth_result(&tool_result);
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
||||
thread.enter_auth_mode(ext_name.clone());
|
||||
}
|
||||
}
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::AuthRequired {
|
||||
extension_name: ext_name,
|
||||
instructions: Some(instructions.clone()),
|
||||
auth_url: auth_data.auth_url,
|
||||
setup_url: auth_data.setup_url,
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
return Ok(AgenticLoopResult::Response(instructions));
|
||||
}
|
||||
|
||||
// Add tool result to context for next LLM call
|
||||
let result_content = match tool_result {
|
||||
Ok(output) => {
|
||||
// Sanitize output before showing to LLM
|
||||
let sanitized =
|
||||
self.safety().sanitize_tool_output(&tc.name, &output);
|
||||
self.safety().wrap_for_llm(
|
||||
&tc.name,
|
||||
&sanitized.content,
|
||||
sanitized.was_modified,
|
||||
)
|
||||
}
|
||||
Err(e) => format!("Error: {}", e),
|
||||
};
|
||||
|
||||
context_messages.push(ChatMessage::tool_result(
|
||||
&tc.id,
|
||||
&tc.name,
|
||||
result_content,
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Execute a tool for chat (without full job context).
|
||||
pub(super) async fn execute_chat_tool(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
params: &serde_json::Value,
|
||||
job_ctx: &JobContext,
|
||||
) -> Result<String, Error> {
|
||||
let tool =
|
||||
self.tools()
|
||||
.get(tool_name)
|
||||
.await
|
||||
.ok_or_else(|| crate::error::ToolError::NotFound {
|
||||
name: tool_name.to_string(),
|
||||
})?;
|
||||
|
||||
// Validate tool parameters
|
||||
let validation = self.safety().validator().validate_tool_params(params);
|
||||
if !validation.is_valid {
|
||||
let details = validation
|
||||
.errors
|
||||
.iter()
|
||||
.map(|e| format!("{}: {}", e.field, e.message))
|
||||
.collect::<Vec<_>>()
|
||||
.join("; ");
|
||||
return Err(crate::error::ToolError::InvalidParameters {
|
||||
name: tool_name.to_string(),
|
||||
reason: format!("Invalid tool parameters: {}", details),
|
||||
}
|
||||
.into());
|
||||
}
|
||||
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
params = %params,
|
||||
"Tool call started"
|
||||
);
|
||||
|
||||
// Execute with per-tool timeout
|
||||
let timeout = tool.execution_timeout();
|
||||
let start = std::time::Instant::now();
|
||||
let result = tokio::time::timeout(timeout, async {
|
||||
tool.execute(params.clone(), job_ctx).await
|
||||
})
|
||||
.await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
match &result {
|
||||
Ok(Ok(output)) => {
|
||||
let result_str = serde_json::to_string(&output.result)
|
||||
.unwrap_or_else(|_| "<serialize error>".to_string());
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
elapsed_ms = elapsed.as_millis() as u64,
|
||||
result = %result_str,
|
||||
"Tool call succeeded"
|
||||
);
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
elapsed_ms = elapsed.as_millis() as u64,
|
||||
error = %e,
|
||||
"Tool call failed"
|
||||
);
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
elapsed_ms = elapsed.as_millis() as u64,
|
||||
timeout_secs = timeout.as_secs(),
|
||||
"Tool call timed out"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let result = result
|
||||
.map_err(|_| crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout,
|
||||
})?
|
||||
.map_err(|e| crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
|
||||
// Convert result to string
|
||||
serde_json::to_string_pretty(&result.result).map_err(|e| {
|
||||
crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: format!("Failed to serialize result: {}", e),
|
||||
}
|
||||
.into()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Parsed auth result fields for emitting StatusUpdate::AuthRequired.
|
||||
pub(super) struct ParsedAuthData {
|
||||
pub(super) auth_url: Option<String>,
|
||||
pub(super) setup_url: Option<String>,
|
||||
}
|
||||
|
||||
/// Extract auth_url and setup_url from a tool_auth result JSON string.
|
||||
pub(super) fn parse_auth_result(result: &Result<String, Error>) -> ParsedAuthData {
|
||||
let parsed = result
|
||||
.as_ref()
|
||||
.ok()
|
||||
.and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok());
|
||||
ParsedAuthData {
|
||||
auth_url: parsed
|
||||
.as_ref()
|
||||
.and_then(|v| v.get("auth_url"))
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|s| s.to_string()),
|
||||
setup_url: parsed
|
||||
.as_ref()
|
||||
.and_then(|v| v.get("setup_url"))
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|s| s.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if a tool_auth result indicates the extension is awaiting a token.
|
||||
///
|
||||
/// Returns `Some((extension_name, instructions))` if the tool result contains
|
||||
/// `awaiting_token: true`, meaning the thread should enter auth mode.
|
||||
pub(super) fn detect_auth_awaiting(
|
||||
tool_name: &str,
|
||||
result: &Result<String, Error>,
|
||||
) -> Option<(String, String)> {
|
||||
if tool_name != "tool_auth" && tool_name != "tool_activate" {
|
||||
return None;
|
||||
}
|
||||
let output = result.as_ref().ok()?;
|
||||
let parsed: serde_json::Value = serde_json::from_str(output).ok()?;
|
||||
if parsed.get("awaiting_token") != Some(&serde_json::Value::Bool(true)) {
|
||||
return None;
|
||||
}
|
||||
let name = parsed.get("name")?.as_str()?.to_string();
|
||||
let instructions = parsed
|
||||
.get("instructions")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("Please provide your API token/key.")
|
||||
.to_string();
|
||||
Some((name, instructions))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::error::Error;
|
||||
|
||||
use super::detect_auth_awaiting;
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_positive() {
|
||||
let result: Result<String, Error> = Ok(serde_json::json!({
|
||||
"name": "telegram",
|
||||
"kind": "WasmTool",
|
||||
"awaiting_token": true,
|
||||
"status": "awaiting_token",
|
||||
"instructions": "Please provide your Telegram Bot API token."
|
||||
})
|
||||
.to_string());
|
||||
|
||||
let detected = detect_auth_awaiting("tool_auth", &result);
|
||||
assert!(detected.is_some());
|
||||
let (name, instructions) = detected.unwrap();
|
||||
assert_eq!(name, "telegram");
|
||||
assert!(instructions.contains("Telegram Bot API"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_not_awaiting() {
|
||||
let result: Result<String, Error> = Ok(serde_json::json!({
|
||||
"name": "telegram",
|
||||
"kind": "WasmTool",
|
||||
"awaiting_token": false,
|
||||
"status": "authenticated"
|
||||
})
|
||||
.to_string());
|
||||
|
||||
assert!(detect_auth_awaiting("tool_auth", &result).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_wrong_tool() {
|
||||
let result: Result<String, Error> = Ok(serde_json::json!({
|
||||
"name": "telegram",
|
||||
"awaiting_token": true,
|
||||
})
|
||||
.to_string());
|
||||
|
||||
assert!(detect_auth_awaiting("tool_list", &result).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_error_result() {
|
||||
let result: Result<String, Error> =
|
||||
Err(crate::error::ToolError::NotFound { name: "x".into() }.into());
|
||||
assert!(detect_auth_awaiting("tool_auth", &result).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_default_instructions() {
|
||||
let result: Result<String, Error> = Ok(serde_json::json!({
|
||||
"name": "custom_tool",
|
||||
"awaiting_token": true,
|
||||
"status": "awaiting_token"
|
||||
})
|
||||
.to_string());
|
||||
|
||||
let (_, instructions) = detect_auth_awaiting("tool_auth", &result).unwrap();
|
||||
assert_eq!(instructions, "Please provide your API token/key.");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_tool_activate() {
|
||||
let result: Result<String, Error> = Ok(serde_json::json!({
|
||||
"name": "slack",
|
||||
"kind": "McpServer",
|
||||
"awaiting_token": true,
|
||||
"status": "awaiting_token",
|
||||
"instructions": "Provide your Slack Bot token."
|
||||
})
|
||||
.to_string());
|
||||
|
||||
let detected = detect_auth_awaiting("tool_activate", &result);
|
||||
assert!(detected.is_some());
|
||||
let (name, instructions) = detected.unwrap();
|
||||
assert_eq!(name, "slack");
|
||||
assert!(instructions.contains("Slack Bot"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_detect_auth_awaiting_tool_activate_not_awaiting() {
|
||||
let result: Result<String, Error> = Ok(serde_json::json!({
|
||||
"name": "slack",
|
||||
"tools_loaded": ["slack_post_message"],
|
||||
"message": "Activated"
|
||||
})
|
||||
.to_string());
|
||||
|
||||
assert!(detect_auth_awaiting("tool_activate", &result).is_none());
|
||||
}
|
||||
}
|
||||
@@ -11,8 +11,11 @@
|
||||
//! - Context compaction for long conversations
|
||||
|
||||
mod agent_loop;
|
||||
mod commands;
|
||||
pub mod compaction;
|
||||
pub mod context_monitor;
|
||||
pub mod cost_guard;
|
||||
mod dispatcher;
|
||||
mod heartbeat;
|
||||
mod router;
|
||||
pub mod routine;
|
||||
@@ -23,6 +26,7 @@ pub mod session;
|
||||
mod session_manager;
|
||||
pub mod submission;
|
||||
pub mod task;
|
||||
mod thread_ops;
|
||||
pub mod undo;
|
||||
pub mod worker;
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user