mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
1f5b582c5f
* fix: optimize agent logging to reduce DataDog bill * fix: log permanent repair failures as ERROR not WARN RepairResult::Failed is permanent failure requiring attention (ERROR level) not a temporary/retryable condition (WARN level). [skip-regression-check] Co-Authored-By: Claude Haiku 4.5 <[email protected]> * security: remove user message content from trace logs Never log user message content at any log level (includes TRACE). Log only safe metadata: content length, message ID, image count. This prevents accidental exposure of sensitive user data in logs even at the most verbose logging level. [skip-regression-check] Co-Authored-By: Claude Haiku 4.5 <[email protected]> * security: move LLM response body logging to TRACE level Response bodies can contain user-generated content, tool outputs, and leaked secrets. Moving to TRACE (not enabled in production) prevents exposure in DEBUG logs. Status log remains at DEBUG. [skip-regression-check] Co-Authored-By: Claude Haiku 4.5 <[email protected]> * refactor: simplify URL sanitization using url::Url API Use set_query, set_fragment, set_username, set_password methods instead of manual string reconstruction. Cleaner, handles edge cases, eliminates port branching complexity. [skip-regression-check] Co-Authored-By: Claude Haiku 4.5 <[email protected]> * test: add comprehensive unit tests for sanitize_url_for_logging Add 9 test cases covering: - URL with query parameters - URL with credentials (user:pass@host) - URL with fragment - URL with port - URL with all components combined - Malformed URL fallback behavior - Short strings (pass-through) - Non-URL-like strings - Path preservation Tests verify that sanitization correctly removes sensitive components while preserving safe components like host, port, and path. Co-Authored-By: Claude Haiku 4.5 <[email protected]> * fix: libsql per-migration logs should be DEBUG, not TRACE Individual migration logs are now visible with standard debug logging (RUST_LOG=ironclaw=debug), improving debuggability when troubleshooting migration issues. Summary log remains at INFO level. Fixes behavioral change that made it harder to identify which specific migration ran or failed without enabling full TRACE logging. [skip-regression-check] --------- Co-authored-by: Claude Haiku 4.5 <[email protected]>
1032 lines
41 KiB
Rust
1032 lines
41 KiB
Rust
//! Main agent loop.
|
|
//!
|
|
//! Contains the `Agent` struct, `AgentDeps`, and the core event loop (`run`).
|
|
//! The heavy lifting is delegated to sibling modules:
|
|
//!
|
|
//! - `dispatcher` - Tool dispatch (agentic loop, tool execution)
|
|
//! - `commands` - System commands and job handlers
|
|
//! - `thread_ops` - Thread/session operations (user input, undo, approval, persistence)
|
|
|
|
use std::sync::Arc;
|
|
|
|
use futures::StreamExt;
|
|
|
|
use crate::agent::context_monitor::ContextMonitor;
|
|
use crate::agent::heartbeat::spawn_heartbeat;
|
|
use crate::agent::routine_engine::{RoutineEngine, spawn_cron_ticker};
|
|
use crate::agent::self_repair::{DefaultSelfRepair, RepairResult, SelfRepair};
|
|
use crate::agent::session_manager::SessionManager;
|
|
use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult};
|
|
use crate::agent::{HeartbeatConfig as AgentHeartbeatConfig, Router, Scheduler};
|
|
use crate::channels::{ChannelManager, IncomingMessage, OutgoingResponse, StatusUpdate};
|
|
use crate::config::{AgentConfig, HeartbeatConfig, RoutineConfig, SkillsConfig};
|
|
use crate::context::ContextManager;
|
|
use crate::db::Database;
|
|
use crate::error::Error;
|
|
use crate::extensions::ExtensionManager;
|
|
use crate::hooks::HookRegistry;
|
|
use crate::llm::LlmProvider;
|
|
use crate::safety::SafetyLayer;
|
|
use crate::skills::SkillRegistry;
|
|
use crate::tools::ToolRegistry;
|
|
use crate::workspace::Workspace;
|
|
|
|
/// Collapse a tool output string into a single-line preview for display.
|
|
pub(crate) fn truncate_for_preview(output: &str, max_chars: usize) -> String {
|
|
let collapsed: String = output
|
|
.chars()
|
|
.take(max_chars + 50)
|
|
.map(|c| if c == '\n' { ' ' } else { c })
|
|
.collect::<String>()
|
|
.split_whitespace()
|
|
.collect::<Vec<_>>()
|
|
.join(" ");
|
|
// char_indices gives us byte offsets at char boundaries, so the slice is always valid UTF-8.
|
|
if collapsed.chars().count() > max_chars {
|
|
let byte_offset = collapsed
|
|
.char_indices()
|
|
.nth(max_chars)
|
|
.map(|(i, _)| i)
|
|
.unwrap_or(collapsed.len());
|
|
format!("{}...", &collapsed[..byte_offset])
|
|
} else {
|
|
collapsed
|
|
}
|
|
}
|
|
|
|
/// Core dependencies for the agent.
|
|
///
|
|
/// Bundles the shared components to reduce argument count.
|
|
pub struct AgentDeps {
|
|
pub store: Option<Arc<dyn Database>>,
|
|
pub llm: Arc<dyn LlmProvider>,
|
|
/// Cheap/fast LLM for lightweight tasks (heartbeat, routing, evaluation).
|
|
/// Falls back to the main `llm` if None.
|
|
pub cheap_llm: Option<Arc<dyn LlmProvider>>,
|
|
pub safety: Arc<SafetyLayer>,
|
|
pub tools: Arc<ToolRegistry>,
|
|
pub workspace: Option<Arc<Workspace>>,
|
|
pub extension_manager: Option<Arc<ExtensionManager>>,
|
|
pub skill_registry: Option<Arc<std::sync::RwLock<SkillRegistry>>>,
|
|
pub skill_catalog: Option<Arc<crate::skills::catalog::SkillCatalog>>,
|
|
pub skills_config: SkillsConfig,
|
|
pub hooks: Arc<HookRegistry>,
|
|
/// Cost enforcement guardrails (daily budget, hourly rate limits).
|
|
pub cost_guard: Arc<crate::agent::cost_guard::CostGuard>,
|
|
/// SSE broadcast sender for live job event streaming to the web gateway.
|
|
pub sse_tx: Option<tokio::sync::broadcast::Sender<crate::channels::web::types::SseEvent>>,
|
|
/// HTTP interceptor for trace recording/replay.
|
|
pub http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
|
|
/// Audio transcription middleware for voice messages.
|
|
pub transcription: Option<Arc<crate::transcription::TranscriptionMiddleware>>,
|
|
/// Document text extraction middleware for PDF, DOCX, PPTX, etc.
|
|
pub document_extraction: Option<Arc<crate::document_extraction::DocumentExtractionMiddleware>>,
|
|
}
|
|
|
|
/// The main agent that coordinates all components.
|
|
pub struct Agent {
|
|
pub(super) config: AgentConfig,
|
|
pub(super) deps: AgentDeps,
|
|
pub(super) channels: Arc<ChannelManager>,
|
|
pub(super) context_manager: Arc<ContextManager>,
|
|
pub(super) scheduler: Arc<Scheduler>,
|
|
pub(super) router: Router,
|
|
pub(super) session_manager: Arc<SessionManager>,
|
|
pub(super) context_monitor: ContextMonitor,
|
|
pub(super) heartbeat_config: Option<HeartbeatConfig>,
|
|
pub(super) hygiene_config: Option<crate::config::HygieneConfig>,
|
|
pub(super) routine_config: Option<RoutineConfig>,
|
|
/// Optional slot to expose the routine engine to the gateway for manual triggering.
|
|
pub(super) routine_engine_slot:
|
|
Option<Arc<tokio::sync::RwLock<Option<Arc<crate::agent::routine_engine::RoutineEngine>>>>>,
|
|
}
|
|
|
|
impl Agent {
|
|
/// Create a new agent.
|
|
///
|
|
/// Optionally accepts pre-created `ContextManager` and `SessionManager` for sharing
|
|
/// with external components (job tools, web gateway). Creates new ones if not provided.
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub fn new(
|
|
config: AgentConfig,
|
|
deps: AgentDeps,
|
|
channels: Arc<ChannelManager>,
|
|
heartbeat_config: Option<HeartbeatConfig>,
|
|
hygiene_config: Option<crate::config::HygieneConfig>,
|
|
routine_config: Option<RoutineConfig>,
|
|
context_manager: Option<Arc<ContextManager>>,
|
|
session_manager: Option<Arc<SessionManager>>,
|
|
) -> Self {
|
|
let context_manager = context_manager
|
|
.unwrap_or_else(|| Arc::new(ContextManager::new(config.max_parallel_jobs)));
|
|
|
|
let session_manager = session_manager.unwrap_or_else(|| Arc::new(SessionManager::new()));
|
|
|
|
let mut scheduler = Scheduler::new(
|
|
config.clone(),
|
|
context_manager.clone(),
|
|
deps.llm.clone(),
|
|
deps.safety.clone(),
|
|
deps.tools.clone(),
|
|
deps.store.clone(),
|
|
deps.hooks.clone(),
|
|
);
|
|
if let Some(ref tx) = deps.sse_tx {
|
|
scheduler.set_sse_sender(tx.clone());
|
|
}
|
|
if let Some(ref interceptor) = deps.http_interceptor {
|
|
scheduler.set_http_interceptor(Arc::clone(interceptor));
|
|
}
|
|
let scheduler = Arc::new(scheduler);
|
|
|
|
Self {
|
|
config,
|
|
deps,
|
|
channels,
|
|
context_manager,
|
|
scheduler,
|
|
router: Router::new(),
|
|
session_manager,
|
|
context_monitor: ContextMonitor::new(),
|
|
heartbeat_config,
|
|
hygiene_config,
|
|
routine_config,
|
|
routine_engine_slot: None,
|
|
}
|
|
}
|
|
|
|
/// Set the routine engine slot for exposing the engine to the gateway.
|
|
pub fn set_routine_engine_slot(
|
|
&mut self,
|
|
slot: Arc<tokio::sync::RwLock<Option<Arc<crate::agent::routine_engine::RoutineEngine>>>>,
|
|
) {
|
|
self.routine_engine_slot = Some(slot);
|
|
}
|
|
|
|
// Convenience accessors
|
|
|
|
/// Get the scheduler (for external wiring, e.g. CreateJobTool).
|
|
pub fn scheduler(&self) -> Arc<Scheduler> {
|
|
Arc::clone(&self.scheduler)
|
|
}
|
|
|
|
pub(super) fn store(&self) -> Option<&Arc<dyn Database>> {
|
|
self.deps.store.as_ref()
|
|
}
|
|
|
|
pub(super) fn llm(&self) -> &Arc<dyn LlmProvider> {
|
|
&self.deps.llm
|
|
}
|
|
|
|
/// Get the cheap/fast LLM provider, falling back to the main one.
|
|
pub(super) fn cheap_llm(&self) -> &Arc<dyn LlmProvider> {
|
|
self.deps.cheap_llm.as_ref().unwrap_or(&self.deps.llm)
|
|
}
|
|
|
|
pub(super) fn safety(&self) -> &Arc<SafetyLayer> {
|
|
&self.deps.safety
|
|
}
|
|
|
|
pub(super) fn tools(&self) -> &Arc<ToolRegistry> {
|
|
&self.deps.tools
|
|
}
|
|
|
|
pub(super) fn workspace(&self) -> Option<&Arc<Workspace>> {
|
|
self.deps.workspace.as_ref()
|
|
}
|
|
|
|
pub(super) fn hooks(&self) -> &Arc<HookRegistry> {
|
|
&self.deps.hooks
|
|
}
|
|
|
|
pub(super) fn cost_guard(&self) -> &Arc<crate::agent::cost_guard::CostGuard> {
|
|
&self.deps.cost_guard
|
|
}
|
|
|
|
pub(super) fn skill_registry(&self) -> Option<&Arc<std::sync::RwLock<SkillRegistry>>> {
|
|
self.deps.skill_registry.as_ref()
|
|
}
|
|
|
|
pub(super) fn skill_catalog(&self) -> Option<&Arc<crate::skills::catalog::SkillCatalog>> {
|
|
self.deps.skill_catalog.as_ref()
|
|
}
|
|
|
|
/// Select active skills for a message using deterministic prefiltering.
|
|
pub(super) fn select_active_skills(
|
|
&self,
|
|
message_content: &str,
|
|
) -> Vec<crate::skills::LoadedSkill> {
|
|
if let Some(registry) = self.skill_registry() {
|
|
let guard = match registry.read() {
|
|
Ok(g) => g,
|
|
Err(e) => {
|
|
tracing::error!("Skill registry lock poisoned: {}", e);
|
|
return vec![];
|
|
}
|
|
};
|
|
let available = guard.skills();
|
|
let skills_cfg = &self.deps.skills_config;
|
|
let selected = crate::skills::prefilter_skills(
|
|
message_content,
|
|
available,
|
|
skills_cfg.max_active_skills,
|
|
skills_cfg.max_context_tokens,
|
|
);
|
|
|
|
if !selected.is_empty() {
|
|
tracing::debug!(
|
|
"Selected {} skill(s) for message: {}",
|
|
selected.len(),
|
|
selected
|
|
.iter()
|
|
.map(|s| s.name())
|
|
.collect::<Vec<_>>()
|
|
.join(", ")
|
|
);
|
|
}
|
|
|
|
selected.into_iter().cloned().collect()
|
|
} else {
|
|
vec![]
|
|
}
|
|
}
|
|
|
|
/// Run the agent main loop.
|
|
pub async fn run(self) -> Result<(), Error> {
|
|
// Start channels
|
|
let mut message_stream = self.channels.start_all().await?;
|
|
|
|
// Start self-repair task with notification forwarding
|
|
let repair = Arc::new(DefaultSelfRepair::new(
|
|
self.context_manager.clone(),
|
|
self.config.stuck_threshold,
|
|
self.config.max_repair_attempts,
|
|
));
|
|
let repair_interval = self.config.repair_check_interval;
|
|
let repair_channels = self.channels.clone();
|
|
let repair_handle = tokio::spawn(async move {
|
|
loop {
|
|
tokio::time::sleep(repair_interval).await;
|
|
|
|
// Check stuck jobs
|
|
let stuck_jobs = repair.detect_stuck_jobs().await;
|
|
for job in stuck_jobs {
|
|
tracing::info!("Attempting to repair stuck job {}", job.job_id);
|
|
let result = repair.repair_stuck_job(&job).await;
|
|
let notification = match &result {
|
|
Ok(RepairResult::Success { message }) => {
|
|
tracing::info!("Repair succeeded: {}", message);
|
|
Some(format!(
|
|
"Job {} was stuck for {}s, recovery succeeded: {}",
|
|
job.job_id,
|
|
job.stuck_duration.as_secs(),
|
|
message
|
|
))
|
|
}
|
|
Ok(RepairResult::Failed { message }) => {
|
|
tracing::error!("Repair failed: {}", message);
|
|
Some(format!(
|
|
"Job {} was stuck for {}s, recovery failed permanently: {}",
|
|
job.job_id,
|
|
job.stuck_duration.as_secs(),
|
|
message
|
|
))
|
|
}
|
|
Ok(RepairResult::ManualRequired { message }) => {
|
|
tracing::warn!("Manual intervention needed: {}", message);
|
|
Some(format!(
|
|
"Job {} needs manual intervention: {}",
|
|
job.job_id, message
|
|
))
|
|
}
|
|
Ok(RepairResult::Retry { message }) => {
|
|
tracing::warn!("Repair needs retry: {}", message);
|
|
None // Don't spam the user on retries
|
|
}
|
|
Err(e) => {
|
|
tracing::error!("Repair error: {}", e);
|
|
None
|
|
}
|
|
};
|
|
|
|
if let Some(msg) = notification {
|
|
let response = OutgoingResponse::text(format!("Self-Repair: {}", msg));
|
|
let _ = repair_channels.broadcast_all("default", response).await;
|
|
}
|
|
}
|
|
|
|
// Check broken tools
|
|
let broken_tools = repair.detect_broken_tools().await;
|
|
for tool in broken_tools {
|
|
tracing::info!("Attempting to repair broken tool: {}", tool.name);
|
|
match repair.repair_broken_tool(&tool).await {
|
|
Ok(RepairResult::Success { message }) => {
|
|
let response = OutgoingResponse::text(format!(
|
|
"Self-Repair: Tool '{}' repaired: {}",
|
|
tool.name, message
|
|
));
|
|
let _ = repair_channels.broadcast_all("default", response).await;
|
|
}
|
|
Ok(result) => {
|
|
tracing::info!("Tool repair result: {:?}", result);
|
|
}
|
|
Err(e) => {
|
|
tracing::error!("Tool repair error: {}", e);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// Spawn session pruning task
|
|
let session_mgr = self.session_manager.clone();
|
|
let session_idle_timeout = self.config.session_idle_timeout;
|
|
let pruning_handle = tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(std::time::Duration::from_secs(600)); // Every 10 min
|
|
interval.tick().await; // Skip immediate first tick
|
|
loop {
|
|
interval.tick().await;
|
|
session_mgr.prune_stale_sessions(session_idle_timeout).await;
|
|
}
|
|
});
|
|
|
|
// Spawn heartbeat if enabled
|
|
let heartbeat_handle = if let Some(ref hb_config) = self.heartbeat_config {
|
|
if hb_config.enabled {
|
|
if let Some(workspace) = self.workspace() {
|
|
let mut config = AgentHeartbeatConfig::default()
|
|
.with_interval(std::time::Duration::from_secs(hb_config.interval_secs));
|
|
config.quiet_hours_start = hb_config.quiet_hours_start;
|
|
config.quiet_hours_end = hb_config.quiet_hours_end;
|
|
config.timezone = hb_config
|
|
.timezone
|
|
.clone()
|
|
.or_else(|| Some(self.config.default_timezone.clone()));
|
|
if let (Some(user), Some(channel)) =
|
|
(&hb_config.notify_user, &hb_config.notify_channel)
|
|
{
|
|
config = config.with_notify(user, channel);
|
|
}
|
|
|
|
// Set up notification channel
|
|
let (notify_tx, mut notify_rx) =
|
|
tokio::sync::mpsc::channel::<OutgoingResponse>(16);
|
|
|
|
// Spawn notification forwarder that routes through channel manager
|
|
let notify_channel = hb_config.notify_channel.clone();
|
|
let notify_user = hb_config.notify_user.clone();
|
|
let channels = self.channels.clone();
|
|
tokio::spawn(async move {
|
|
while let Some(response) = notify_rx.recv().await {
|
|
let user = notify_user.as_deref().unwrap_or("default");
|
|
|
|
// Try the configured channel first, fall back to
|
|
// broadcasting on all channels.
|
|
let targeted_ok = if let Some(ref channel) = notify_channel {
|
|
channels
|
|
.broadcast(channel, user, response.clone())
|
|
.await
|
|
.is_ok()
|
|
} else {
|
|
false
|
|
};
|
|
|
|
if !targeted_ok {
|
|
let results = channels.broadcast_all(user, response).await;
|
|
for (ch, result) in results {
|
|
if let Err(e) = result {
|
|
tracing::warn!(
|
|
"Failed to broadcast heartbeat to {}: {}",
|
|
ch,
|
|
e
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
let hygiene = self
|
|
.hygiene_config
|
|
.as_ref()
|
|
.map(|h| h.to_workspace_config())
|
|
.unwrap_or_default();
|
|
|
|
Some(spawn_heartbeat(
|
|
config,
|
|
hygiene,
|
|
workspace.clone(),
|
|
self.cheap_llm().clone(),
|
|
Some(notify_tx),
|
|
self.store().map(Arc::clone),
|
|
))
|
|
} else {
|
|
tracing::warn!("Heartbeat enabled but no workspace available");
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
};
|
|
|
|
// Spawn routine engine if enabled
|
|
let routine_handle = if let Some(ref rt_config) = self.routine_config {
|
|
if rt_config.enabled {
|
|
if let (Some(store), Some(workspace)) = (self.store(), self.workspace()) {
|
|
// Set up notification channel (same pattern as heartbeat)
|
|
let (notify_tx, mut notify_rx) =
|
|
tokio::sync::mpsc::channel::<OutgoingResponse>(32);
|
|
|
|
let engine = Arc::new(RoutineEngine::new(
|
|
rt_config.clone(),
|
|
Arc::clone(store),
|
|
self.llm().clone(),
|
|
Arc::clone(workspace),
|
|
notify_tx,
|
|
Some(self.scheduler.clone()),
|
|
self.tools().clone(),
|
|
self.safety().clone(),
|
|
));
|
|
|
|
// Register routine tools
|
|
self.deps
|
|
.tools
|
|
.register_routine_tools(Arc::clone(store), Arc::clone(&engine));
|
|
|
|
// Load initial event cache
|
|
engine.refresh_event_cache().await;
|
|
|
|
// Spawn notification forwarder (mirrors heartbeat pattern)
|
|
let channels = self.channels.clone();
|
|
tokio::spawn(async move {
|
|
while let Some(response) = notify_rx.recv().await {
|
|
let user = response
|
|
.metadata
|
|
.get("notify_user")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("default")
|
|
.to_string();
|
|
let notify_channel = response
|
|
.metadata
|
|
.get("notify_channel")
|
|
.and_then(|v| v.as_str())
|
|
.map(|s| s.to_string());
|
|
|
|
// Try the configured channel first, fall back to
|
|
// broadcasting on all channels.
|
|
let targeted_ok = if let Some(ref channel) = notify_channel {
|
|
channels
|
|
.broadcast(channel, &user, response.clone())
|
|
.await
|
|
.is_ok()
|
|
} else {
|
|
false
|
|
};
|
|
|
|
if !targeted_ok {
|
|
let results = channels.broadcast_all(&user, response).await;
|
|
for (ch, result) in results {
|
|
if let Err(e) = result {
|
|
tracing::warn!(
|
|
"Failed to broadcast routine notification to {}: {}",
|
|
ch,
|
|
e
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// Spawn cron ticker
|
|
let cron_interval =
|
|
std::time::Duration::from_secs(rt_config.cron_check_interval_secs);
|
|
let cron_handle = spawn_cron_ticker(Arc::clone(&engine), cron_interval);
|
|
|
|
// Store engine reference for event trigger checking
|
|
// Safety: we're in run() which takes self, no other reference exists
|
|
let engine_ref = Arc::clone(&engine);
|
|
// SAFETY: self is consumed by run(), we can smuggle the engine in
|
|
// via a local to use in the message loop below.
|
|
|
|
// Expose engine to gateway for manual triggering
|
|
if let Some(ref slot) = self.routine_engine_slot {
|
|
*slot.write().await = Some(Arc::clone(&engine));
|
|
}
|
|
|
|
tracing::debug!(
|
|
"Routines enabled: cron ticker every {}s, max {} concurrent",
|
|
rt_config.cron_check_interval_secs,
|
|
rt_config.max_concurrent_routines
|
|
);
|
|
|
|
Some((cron_handle, engine_ref))
|
|
} else {
|
|
tracing::warn!("Routines enabled but store/workspace not available");
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
}
|
|
} else {
|
|
None
|
|
};
|
|
|
|
// Extract engine ref for use in message loop
|
|
let routine_engine_for_loop = routine_handle.as_ref().map(|(_, e)| Arc::clone(e));
|
|
|
|
// Main message loop
|
|
tracing::debug!("Agent {} ready and listening", self.config.name);
|
|
|
|
loop {
|
|
let message = tokio::select! {
|
|
biased;
|
|
_ = tokio::signal::ctrl_c() => {
|
|
tracing::debug!("Ctrl+C received, shutting down...");
|
|
break;
|
|
}
|
|
msg = message_stream.next() => {
|
|
match msg {
|
|
Some(m) => m,
|
|
None => {
|
|
tracing::debug!("All channel streams ended, shutting down...");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
// Apply transcription middleware to audio attachments
|
|
let mut message = message;
|
|
if let Some(ref transcription) = self.deps.transcription {
|
|
transcription.process(&mut message).await;
|
|
}
|
|
|
|
// Apply document extraction middleware to document attachments
|
|
if let Some(ref doc_extraction) = self.deps.document_extraction {
|
|
doc_extraction.process(&mut message).await;
|
|
}
|
|
|
|
// Store successfully extracted document text in workspace for indexing
|
|
self.store_extracted_documents(&message).await;
|
|
|
|
match self.handle_message(&message).await {
|
|
Ok(Some(response)) if !response.is_empty() => {
|
|
// Hook: BeforeOutbound — allow hooks to modify or suppress outbound
|
|
let event = crate::hooks::HookEvent::Outbound {
|
|
user_id: message.user_id.clone(),
|
|
channel: message.channel.clone(),
|
|
content: response.clone(),
|
|
thread_id: message.thread_id.clone(),
|
|
};
|
|
match self.hooks().run(&event).await {
|
|
Err(err) => {
|
|
tracing::warn!("BeforeOutbound hook blocked response: {}", err);
|
|
}
|
|
Ok(crate::hooks::HookOutcome::Continue {
|
|
modified: Some(new_content),
|
|
}) => {
|
|
if let Err(e) = self
|
|
.channels
|
|
.respond(&message, OutgoingResponse::text(new_content))
|
|
.await
|
|
{
|
|
tracing::error!(
|
|
channel = %message.channel,
|
|
error = %e,
|
|
"Failed to send response to channel"
|
|
);
|
|
}
|
|
}
|
|
_ => {
|
|
if let Err(e) = self
|
|
.channels
|
|
.respond(&message, OutgoingResponse::text(response))
|
|
.await
|
|
{
|
|
tracing::error!(
|
|
channel = %message.channel,
|
|
error = %e,
|
|
"Failed to send response to channel"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Ok(Some(empty)) => {
|
|
// Empty response, nothing to send (e.g. approval handled via send_status)
|
|
tracing::debug!(
|
|
channel = %message.channel,
|
|
user = %message.user_id,
|
|
empty_len = empty.len(),
|
|
"Suppressed empty response (not sent to channel)"
|
|
);
|
|
}
|
|
Ok(None) => {
|
|
// Shutdown signal received (/quit, /exit, /shutdown)
|
|
tracing::debug!("Shutdown command received, exiting...");
|
|
break;
|
|
}
|
|
Err(e) => {
|
|
tracing::error!("Error handling message: {}", e);
|
|
if let Err(send_err) = self
|
|
.channels
|
|
.respond(&message, OutgoingResponse::text(format!("Error: {}", e)))
|
|
.await
|
|
{
|
|
tracing::error!(
|
|
channel = %message.channel,
|
|
error = %send_err,
|
|
"Failed to send error response to channel"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check event triggers (cheap in-memory regex, fires async if matched)
|
|
if let Some(ref engine) = routine_engine_for_loop {
|
|
let fired = engine.check_event_triggers(&message).await;
|
|
if fired > 0 {
|
|
tracing::debug!("Fired {} event-triggered routines", fired);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Cleanup
|
|
tracing::debug!("Agent shutting down...");
|
|
repair_handle.abort();
|
|
pruning_handle.abort();
|
|
if let Some(handle) = heartbeat_handle {
|
|
handle.abort();
|
|
}
|
|
if let Some((cron_handle, _)) = routine_handle {
|
|
cron_handle.abort();
|
|
}
|
|
self.scheduler.stop_all().await;
|
|
self.channels.shutdown_all().await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Store extracted document text in workspace memory for future search/recall.
|
|
async fn store_extracted_documents(&self, message: &IncomingMessage) {
|
|
let workspace = match self.workspace() {
|
|
Some(ws) => ws,
|
|
None => return,
|
|
};
|
|
|
|
for attachment in &message.attachments {
|
|
if attachment.kind != crate::channels::AttachmentKind::Document {
|
|
continue;
|
|
}
|
|
let text = match &attachment.extracted_text {
|
|
Some(t) if !t.starts_with('[') => t, // skip error messages like "[Failed to..."
|
|
_ => continue,
|
|
};
|
|
|
|
// Sanitize filename: strip path separators to prevent directory traversal
|
|
let raw_name = attachment.filename.as_deref().unwrap_or("unnamed_document");
|
|
let filename: String = raw_name
|
|
.chars()
|
|
.map(|c| {
|
|
if c == '/' || c == '\\' || c == '\0' {
|
|
'_'
|
|
} else {
|
|
c
|
|
}
|
|
})
|
|
.collect();
|
|
let filename = filename.trim_start_matches('.');
|
|
let filename = if filename.is_empty() {
|
|
"unnamed_document"
|
|
} else {
|
|
filename
|
|
};
|
|
let date = chrono::Utc::now().format("%Y-%m-%d");
|
|
let path = format!("documents/{date}/{filename}");
|
|
|
|
let header = format!(
|
|
"# {filename}\n\n\
|
|
> Uploaded by **{}** via **{}** on {date}\n\
|
|
> MIME: {} | Size: {} bytes\n\n---\n\n",
|
|
message.user_id,
|
|
message.channel,
|
|
attachment.mime_type,
|
|
attachment.size_bytes.unwrap_or(0),
|
|
);
|
|
let content = format!("{header}{text}");
|
|
|
|
match workspace.write(&path, &content).await {
|
|
Ok(_) => {
|
|
tracing::info!(
|
|
path = %path,
|
|
text_len = text.len(),
|
|
"Stored extracted document in workspace memory"
|
|
);
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
path = %path,
|
|
error = %e,
|
|
"Failed to store extracted document in workspace"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn handle_message(&self, message: &IncomingMessage) -> Result<Option<String>, Error> {
|
|
// Log at info level only for tracking without exposing PII (user_id can be a phone number)
|
|
tracing::info!(message_id = %message.id, "Processing message");
|
|
|
|
// Log sensitive details at debug level for troubleshooting
|
|
tracing::debug!(
|
|
message_id = %message.id,
|
|
user_id = %message.user_id,
|
|
channel = %message.channel,
|
|
thread_id = ?message.thread_id,
|
|
"Message details"
|
|
);
|
|
|
|
// Set message tool context for this turn (current channel and target)
|
|
// For Signal, use signal_target from metadata (group:ID or phone number),
|
|
// otherwise fall back to user_id
|
|
let target = message
|
|
.metadata
|
|
.get("signal_target")
|
|
.and_then(|v| v.as_str())
|
|
.map(|s| s.to_string())
|
|
.unwrap_or_else(|| message.user_id.clone());
|
|
self.tools()
|
|
.set_message_tool_context(Some(message.channel.clone()), Some(target))
|
|
.await;
|
|
|
|
// Parse submission type first
|
|
let mut submission = SubmissionParser::parse(&message.content);
|
|
tracing::trace!(
|
|
"[agent_loop] Parsed submission: {:?}",
|
|
std::any::type_name_of_val(&submission)
|
|
);
|
|
|
|
// Hook: BeforeInbound — allow hooks to modify or reject user input
|
|
if let Submission::UserInput { ref content } = submission {
|
|
let event = crate::hooks::HookEvent::Inbound {
|
|
user_id: message.user_id.clone(),
|
|
channel: message.channel.clone(),
|
|
content: content.clone(),
|
|
thread_id: message.thread_id.clone(),
|
|
};
|
|
match self.hooks().run(&event).await {
|
|
Err(crate::hooks::HookError::Rejected { reason }) => {
|
|
return Ok(Some(format!("[Message rejected: {}]", reason)));
|
|
}
|
|
Err(err) => {
|
|
return Ok(Some(format!("[Message blocked by hook policy: {}]", err)));
|
|
}
|
|
Ok(crate::hooks::HookOutcome::Continue {
|
|
modified: Some(new_content),
|
|
}) => {
|
|
submission = Submission::UserInput {
|
|
content: new_content,
|
|
};
|
|
}
|
|
_ => {} // Continue, fail-open errors already logged in registry
|
|
}
|
|
}
|
|
|
|
// Hydrate thread from DB if it's a historical thread not in memory
|
|
if let Some(ref external_thread_id) = message.thread_id {
|
|
tracing::trace!(
|
|
message_id = %message.id,
|
|
thread_id = %external_thread_id,
|
|
"Hydrating thread from DB"
|
|
);
|
|
self.maybe_hydrate_thread(message, external_thread_id).await;
|
|
}
|
|
|
|
// Resolve session and thread
|
|
tracing::debug!(
|
|
message_id = %message.id,
|
|
"Resolving session and thread"
|
|
);
|
|
let (session, thread_id) = self
|
|
.session_manager
|
|
.resolve_thread(
|
|
&message.user_id,
|
|
&message.channel,
|
|
message.thread_id.as_deref(),
|
|
)
|
|
.await;
|
|
tracing::debug!(
|
|
message_id = %message.id,
|
|
thread_id = %thread_id,
|
|
"Resolved session and thread"
|
|
);
|
|
|
|
// Auth mode interception: if the thread is awaiting a token, route
|
|
// the message directly to the credential store. Nothing touches
|
|
// logs, turns, history, or compaction.
|
|
let pending_auth = {
|
|
let sess = session.lock().await;
|
|
sess.threads
|
|
.get(&thread_id)
|
|
.and_then(|t| t.pending_auth.clone())
|
|
};
|
|
|
|
if let Some(pending) = pending_auth {
|
|
match &submission {
|
|
Submission::UserInput { content } => {
|
|
return self
|
|
.process_auth_token(message, &pending, content, session, thread_id)
|
|
.await;
|
|
}
|
|
_ => {
|
|
// Any control submission (interrupt, undo, etc.) cancels auth mode
|
|
let mut sess = session.lock().await;
|
|
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
|
thread.pending_auth = None;
|
|
}
|
|
// Fall through to normal handling
|
|
}
|
|
}
|
|
}
|
|
|
|
tracing::trace!(
|
|
"Received message from {} on {} ({} chars)",
|
|
message.user_id,
|
|
message.channel,
|
|
message.content.len()
|
|
);
|
|
|
|
// Process based on submission type
|
|
let result = match submission {
|
|
Submission::UserInput { content } => {
|
|
self.process_user_input(message, session, thread_id, &content)
|
|
.await
|
|
}
|
|
Submission::SystemCommand { command, args } => {
|
|
tracing::debug!(
|
|
"[agent_loop] SystemCommand: command={}, channel={}",
|
|
command,
|
|
message.channel
|
|
);
|
|
// Authorization checks (including restart channel check) are enforced in handle_system_command
|
|
self.handle_system_command(&command, &args, &message.channel)
|
|
.await
|
|
}
|
|
Submission::Undo => self.process_undo(session, thread_id).await,
|
|
Submission::Redo => self.process_redo(session, thread_id).await,
|
|
Submission::Interrupt => self.process_interrupt(session, thread_id).await,
|
|
Submission::Compact => self.process_compact(session, thread_id).await,
|
|
Submission::Clear => self.process_clear(session, thread_id).await,
|
|
Submission::NewThread => self.process_new_thread(message).await,
|
|
Submission::Heartbeat => self.process_heartbeat().await,
|
|
Submission::Summarize => self.process_summarize(session, thread_id).await,
|
|
Submission::Suggest => self.process_suggest(session, thread_id).await,
|
|
Submission::JobStatus { job_id } => {
|
|
self.process_job_status(&message.user_id, job_id.as_deref())
|
|
.await
|
|
}
|
|
Submission::JobCancel { job_id } => {
|
|
self.process_job_cancel(&message.user_id, &job_id).await
|
|
}
|
|
Submission::Quit => return Ok(None),
|
|
Submission::SwitchThread { thread_id: target } => {
|
|
self.process_switch_thread(message, target).await
|
|
}
|
|
Submission::Resume { checkpoint_id } => {
|
|
self.process_resume(session, thread_id, checkpoint_id).await
|
|
}
|
|
Submission::ExecApproval {
|
|
request_id,
|
|
approved,
|
|
always,
|
|
} => {
|
|
self.process_approval(
|
|
message,
|
|
session,
|
|
thread_id,
|
|
Some(request_id),
|
|
approved,
|
|
always,
|
|
)
|
|
.await
|
|
}
|
|
Submission::ApprovalResponse { approved, always } => {
|
|
self.process_approval(message, session, thread_id, None, approved, always)
|
|
.await
|
|
}
|
|
};
|
|
|
|
// Convert SubmissionResult to response string
|
|
match result? {
|
|
SubmissionResult::Response { content } => {
|
|
// Suppress silent replies (e.g. from group chat "nothing to say" responses)
|
|
if crate::llm::is_silent_reply(&content) {
|
|
tracing::debug!("Suppressing silent reply token");
|
|
Ok(None)
|
|
} else {
|
|
Ok(Some(content))
|
|
}
|
|
}
|
|
SubmissionResult::Ok { message } => Ok(message),
|
|
SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
|
|
SubmissionResult::Interrupted => Ok(Some("Interrupted.".into())),
|
|
SubmissionResult::NeedApproval {
|
|
request_id,
|
|
tool_name,
|
|
description,
|
|
parameters,
|
|
} => {
|
|
// Each channel renders the approval prompt via send_status.
|
|
// Web gateway shows an inline card, REPL prints a formatted prompt, etc.
|
|
let _ = self
|
|
.channels
|
|
.send_status(
|
|
&message.channel,
|
|
StatusUpdate::ApprovalNeeded {
|
|
request_id: request_id.to_string(),
|
|
tool_name,
|
|
description,
|
|
parameters,
|
|
},
|
|
&message.metadata,
|
|
)
|
|
.await;
|
|
|
|
// Empty string signals the caller to skip respond() (no duplicate text)
|
|
Ok(Some(String::new()))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::truncate_for_preview;
|
|
|
|
#[test]
|
|
fn test_truncate_short_input() {
|
|
assert_eq!(truncate_for_preview("hello", 10), "hello");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_empty_input() {
|
|
assert_eq!(truncate_for_preview("", 10), "");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_exact_length() {
|
|
assert_eq!(truncate_for_preview("hello", 5), "hello");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_over_limit() {
|
|
let result = truncate_for_preview("hello world, this is long", 10);
|
|
assert!(result.ends_with("..."));
|
|
// "hello worl" = 10 chars + "..."
|
|
assert_eq!(result, "hello worl...");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_collapses_newlines() {
|
|
let result = truncate_for_preview("line1\nline2\nline3", 100);
|
|
assert!(!result.contains('\n'));
|
|
assert_eq!(result, "line1 line2 line3");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_collapses_whitespace() {
|
|
let result = truncate_for_preview("hello world", 100);
|
|
assert_eq!(result, "hello world");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_multibyte_utf8() {
|
|
// Each emoji is 4 bytes. Truncating at char boundary must not panic.
|
|
let input = "😀😁😂🤣😃😄😅😆😉😊";
|
|
let result = truncate_for_preview(input, 5);
|
|
assert!(result.ends_with("..."));
|
|
// First 5 chars = 5 emoji
|
|
assert_eq!(result, "😀😁😂🤣😃...");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_cjk_characters() {
|
|
// CJK chars are 3 bytes each in UTF-8.
|
|
let input = "你好世界测试数据很长的字符串";
|
|
let result = truncate_for_preview(input, 4);
|
|
assert_eq!(result, "你好世界...");
|
|
}
|
|
|
|
#[test]
|
|
fn test_truncate_mixed_multibyte_and_ascii() {
|
|
let input = "hello 世界 foo";
|
|
let result = truncate_for_preview(input, 8);
|
|
// 'h','e','l','l','o',' ','世','界' = 8 chars
|
|
assert_eq!(result, "hello 世界...");
|
|
}
|
|
}
|