mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
feat: Sandbox jobs (#4)
* Orchestrating jobs and running them in sandboxes * Fix heartbeat: dynamic max_tokens, empty content guard, notification fallback - Query /v1/models API for context_length and set max_tokens to half (floor 4096) instead of hardcoded 1024; reasoning models like GLM-4.7 need much larger budgets - Guard against empty LLM content (reasoning models can burn all tokens on chain-of-thought and return content: null) - Simplify notification routing: try configured channel first, fall back to broadcast_all so heartbeat alerts always reach someone - Add ModelMetadata struct and model_metadata() to LlmProvider trait - Refactor NearAiChatProvider::list_models into shared fetch_models() - Add standalone test_heartbeat example for isolated debugging Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add job detail view with drill-down from jobs list Click a job row to see full details across four sub-tabs: Overview (metadata grid, description, state transitions timeline), Actions (expandable tool call cards with input/output JSON), Thinking (conversation messages styled by role), and Files (embedded workspace tree browser). Co-Authored-By: Claude Opus 4.6 <[email protected]> * Strip model-internal XML tags from LLM responses, fix Telegram parse_mode 400 Some models (GLM-4.7, etc.) emit <tool_call>tool_list</tool_call> in the content field instead of using the OpenAI tool_calls array. This XML leaks through to channels as text, and Telegram's Markdown parser chokes on the underscores, returning 400 "can't parse entities". Two fixes: - Generalize clean_response() to strip <tool_call>, <function_call>, <tool_calls>, and pipe-delimited variants (<|tool_call|>) alongside the existing <thinking> tag stripping - Add Telegram send_message helper with parse_mode fallback: try Markdown first, retry as plain text on "can't parse entities" 400 errors Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add SystemCommand submission type for thread-state-independent commands System commands (/help, /model, /version, /tools, /ping, /debug) now bypass thread-state checks and safety validation via a dedicated Submission::SystemCommand variant. Previously these flowed through process_user_input() which blocked them during Processing/AwaitingApproval /Completed states. - Add /model [name] for runtime model switching with provider validation - Add active_model_name()/set_model() to LlmProvider trait with RwLock hot-swap in both NEAR AI providers - Rewrite /help with aligned columns grouped by category - Expand REPL tab-completion from 10 to 23 slash commands - Remove REPL-local /help interception (now handled by agent) Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add per-tool execution timeouts, auto-create sandbox project dirs, serve built files The sandbox e2e pipeline (agent -> container -> built website -> browsable URL) was broken by three gaps: hardcoded 60s timeouts killed sandbox jobs that need minutes, no auto-created project directory meant container output vanished, and no HTTP route to browse the built files. - Add `execution_timeout()` to the `Tool` trait (default 60s), replace all four hardcoded `Duration::from_secs(60)` call sites (agent_loop, worker, scheduler, worker/runtime) with the per-tool value - Override to 660s in `RunInSandboxTool` (10 min polling + 60s buffer) - Auto-create `~/.ironclaw/projects/{uuid}/` when no `project_dir` is specified, so every sandbox job gets a persistent bind mount - Include `project_dir` and `browse_url` in sandbox tool output JSON - Add `/projects/{id}` and `/projects/{id}/{path}` static file serving routes to the web gateway with path traversal protection and MIME type detection - Add `mime_guess` dependency for content-type detection Co-Authored-By: Claude Opus 4.6 <[email protected]> * Apply cargo fmt to wizard.rs after merge Co-Authored-By: Claude Opus 4.6 <[email protected]> * Persist sandbox jobs in DB, fix web UI, unify job model Sandbox container jobs were invisible to the web UI because they lived only in ContainerJobManager's in-memory HashMap while the API queried ContextManager. This persists them to the agent_jobs table and fixes all six front-end bugs (empty job list, broken back button, empty actions/thinking tabs, wrong files tab, stuck status, no persistence). Key changes: - V4 migration adds project_dir and user_id columns to agent_jobs - Embedded migrations via refinery (no external CLI needed) - SandboxJobRecord CRUD in Store with fire-and-forget DB writes - Unified job_id: sandbox tool generates UUID, passes to ContainerJobManager - Web API queries DB for sandbox jobs, merges with ContextManager direct jobs - New endpoints: restart, project file list/read with path traversal protection - Front-end: rebuild DOM on back navigation, sandbox-aware tabs, job cards in chat stream, source badges, restart button for failed/interrupted jobs - Gateway defaults to enabled, prints Web UI URL on startup - Stale jobs marked "interrupted" on restart for visibility and restartability Co-Authored-By: Claude Opus 4.6 <[email protected]> * Secure in-chat auth: tokens never touch the LLM or chat history Remove the token parameter from tool_auth so the LLM cannot pass raw API keys. Add dedicated REST (POST /api/chat/auth-token) and WebSocket (auth_token) endpoints that route tokens directly to ext_mgr.auth(), completely bypassing the message pipeline, turns, history, and compaction. Web UI shows an auth card (password input + OAuth button) when the agent enters auth mode, submitted via the dedicated endpoint. CLI auth mode interception is unchanged (already secure). New StatusUpdate::AuthRequired/AuthCompleted variants propagate through all channels (SSE, WebSocket, REPL, WASM). Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Add Claude Code mode for sandbox jobs Run Claude Code CLI inside Docker containers as an alternative to the standard worker mode. The bridge spawns `claude -p` with stream-json output, posts events to the orchestrator, and supports follow-up prompts via `--resume`. Key additions: - `claude-bridge` CLI subcommand and ClaudeBridgeRuntime - JobMode enum (Worker vs ClaudeCode) with per-mode container config - Orchestrator endpoints for Claude events and prompt polling - SSE event variants for real-time Claude Code streaming to frontend - Claude Code sub-tab in web UI with terminal-style output and input bar - Database migration for job_mode column and claude_code_events table - ClaudeCodeConfig with env var support (CLAUDE_CODE_ENABLED, etc.) - Mode parameter on run_in_sandbox tool schema Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Skip create_job tool when sandbox is enabled to prevent duplicate jobs When sandbox mode is on, the LLM would call create_job (creating a pending "direct" entry) then run_in_sandbox (creating a second "sandbox" entry), producing two jobs in the list for a single user request. Now register_job_tools() skips create_job when sandbox is enabled since run_in_sandbox already creates tracked jobs. Also improved the run_in_sandbox description to guide the LLM to use it directly and to mention wait=false for async execution. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Web gateway UI quality-of-life improvements Phase 1: Send button disabled state to prevent double-sends, copy button on code blocks, confirm() guards on destructive actions, SSE-driven job list auto-refresh, log filters re-applied on tab switch, jobEvents memory leak fix (cap at 500, cleanup after 60s). Phase 2: Toast notification system replacing chat-based system messages, memory search highlighting with centered snippets, keyboard shortcuts (Ctrl+1-5 tabs, Ctrl+K focus, Ctrl+N new thread, Escape close/blur), activity tab toolbar with event type filter and auto-scroll toggle. Phase 3: Thread sidebar with load/switch/create, thread_id passed with messages, collapsible to hamburger. Memory inline editing with textarea, Save/Cancel, POST to /api/memory/write. Phase 4: Gateway status popover on hover (polls every 30s), extension install form (name/URL/kind), markdown rendering in memory viewer for .md files, mobile responsive layout at 768px breakpoint. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Add routines system, remove non-sandbox job mode from web UI Routines: scheduled & reactive job system with cron and event triggers, lightweight (single LLM call) and full-job execution modes, guardrails (cooldown, max concurrent, dedup), and LLM-facing tools for CRUD. Web UI: remove ContextManager-backed "direct" job mode entirely. Jobs are now exclusively sandbox-backed (DB + container). Simplify job detail response, drop dead types (ActionInfo, MessageInfo, MessageToolCallInfo), fix Browse Files CSS loading (trailing-slash redirect), fix Activity tab event rendering. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Re-enable chat input on agent completion, auto-auth on tool_activate, recover tool calls from content XML Three fixes: 1. Chat input stays disabled after agent finishes: the "Done" status SSE event now calls enableChatInput() as a safety net when the response event is empty or lost. Same for auth_completed and cancelAuth(). 2. tool_activate never triggers auth: when activation fails due to missing authentication, it now auto-initiates the auth flow (same pattern as the web API handler). detect_auth_awaiting() also matches tool_activate results now. 3. Models like GLM-4.7 emit tool calls as XML tags in content (<tool_call>tool_list</tool_call>) instead of using the structured tool_calls array. recover_tool_calls_from_content() extracts and validates these before falling back to plain text. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Add routines web UI tab, update docs for sandbox-jobs branch Add full routines management to the web gateway (list, detail, trigger, toggle, delete) with 7 new API endpoints, response types, and frontend (HTML, JS, CSS). Update FEATURE_PARITY.md (~23 rows), CLAUDE.md (new subsystems, config, TODOs), and README.md (architecture diagram, features, components, fix onboard command). Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Bind Telegram bot to owner account during setup Without owner binding, anyone who discovers the bot can send it messages. The setup wizard now prompts the user to message their bot, captures their Telegram user ID via getUpdates, and persists it as telegram_owner_id in settings. On startup, the owner_id is injected into the WASM channel config so the existing owner restriction logic drops messages from non-owners. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Move settings from disk to PostgreSQL database Settings previously lived in three JSON files on disk (settings.json, mcp-servers.json, session.json). This made them inaccessible from the web UI and caused redundant disk reads (Settings::load() called 8+ times during startup). Now all settings live in a `settings` table (user_id + key -> JSONB) with only 4 bootstrap fields remaining on disk (database_url, pool size, secrets key source, onboard_completed) since they're needed before the DB connection exists. - Add V8 migration for settings table - Add BootstrapConfig (thin disk file) and Settings DB round-trip - Add Store CRUD methods for settings (get/set/delete/list/bulk) - Refactor Config to load from DB (env > DB > default cascade) - Add SessionManager DB persistence for session tokens - Add DB-backed MCP server config load/save functions - Add 6 settings web API endpoints (list/get/set/delete/export/import) - Add one-time disk-to-DB migration on first boot - Make CLI config commands async with DB access (disk fallback) Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Seed workspace on boot, fix gateway duplicate logs and URL auto-auth - Add Workspace::seed_if_empty() to create core identity files (README, MEMORY, IDENTITY, SOUL, AGENTS, USER, HEARTBEAT) when missing, called on every boot without overwriting existing user edits - Remove duplicate gateway log lines from web/mod.rs (main.rs has the useful clickable ?token= URL) - Auto-authenticate from ?token= URL parameter in the web UI and strip the token from the address bar after successful auth Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Harden sandbox security (path traversal + orchestrator auth) Two vulnerabilities fixed: 1. project_dir path traversal: The create_job tool let the LLM specify arbitrary host paths for Docker bind mounts. Removed project_dir from the tool schema entirely, and added canonicalization + prefix validation at both resolve_project_dir() and the job_manager bind mount point. 2. Orchestrator API auth bypass: worker_auth_middleware was defined but never applied. Each handler manually called validate_token(), so any new endpoint that forgot would be publicly accessible. Applied the middleware as route_layer on all /worker/ routes, removed manual auth from all 7 handlers. Bind to 127.0.0.1 on macOS/Windows (Linux keeps 0.0.0.0 since containers reach host via docker bridge, not loopback). Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Rework gateway chat with pinned assistant, pagination, and NEAR AI response chaining Implements the 4-phase plan for overhauling the web gateway chat: - Phase 1: Pinned "Assistant" thread at top of sidebar, regular threads below - Phase 2: Cursor-based history pagination with infinite scroll - Phase 3: NEAR AI previous_response_id chaining (delta-only messages), with fallback to full history on chain errors, and DB persistence of chain state across restarts - Phase 4: SSE thread isolation (events filtered by thread_id) Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Add per-request HTTP timeout to WASM host, redact credentials in errors Three fixes for WASM channel reliability: 1. Per-request timeout: Add optional timeout-ms parameter to http-request in both channel and tool WIT interfaces. Telegram long-poll now specifies 35s (outliving the 30s server-side hold), while regular API calls use the 30s default. Fixes the triple-30s timeout race that caused polling failures. 2. Credential redaction: reqwest::Error includes the full URL (with injected bot tokens) in its Display output. Scrub credential values from error messages before logging or returning to WASM. 3. Webhook route registration: Remove tunnel URL gate so webhook routes are always available when webhook channels exist, not only when TUNNEL_URL is configured. Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: Fix clippy warnings in WASM tools and channels - slack channel: allow dead_code on signing_secret_name (forward compat field) - gmail tool: use div_ceil() instead of manual (n+2)/3 - google-calendar tool: extract CreateEventParams/UpdateEventParams structs to fix too-many-arguments warnings Co-Authored-By: Claude Opus 4.6 <[email protected]> * Fix approval flow * fix: Rebuild bundled telegram.wasm with updated WIT interface The bundled WASM binary must match the host's WIT definition. Previous binary was compiled against the old 4-arg http-request; this rebuild includes the new timeout-ms parameter. Co-Authored-By: Claude Opus 4.6 <[email protected]> * refactor: Load WASM channels from disk instead of bundling in binary Remove include_bytes! embedding of telegram.wasm. Channels are now loaded from their build output directories (channels-src/<name>/target/) during onboarding, then from ~/.ironclaw/channels/ at runtime. - bundled.rs: locate_channel_artifacts() finds WASM + capabilities from build output; IRONCLAW_CHANNELS_SRC env var overrides the default path - available_channel_names(): only lists channels with build artifacts - bundled_channel_names(): lists all known channels (manifest) - Setup wizard uses available_channel_names() to offer installable channels - Add *.wasm to .gitignore, remove tracked telegram.wasm Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Persist gateway auth token, fix thread hydration race, polish auth screen Three web gateway UX fixes: 1. Token persistence: Store auth token in sessionStorage so refreshing the page doesn't force re-authentication. Hide the auth screen immediately when a saved token exists to prevent flash. 2. Thread hydration: Remove the !msgs.is_empty() bail-out in maybe_hydrate_thread so that even brand-new (empty) assistant threads get hydrated with their correct DB UUID. Previously resolve_thread would mint a fresh UUID, causing messages to land in the wrong conversation and duplicate threads to appear. 3. Auth screen: Redesign as a centered card with brand, tagline, labeled input, and hint text. Also adds 34 new tests covering session/thread lifecycle, thread resolution isolation (user, channel, external ID), hydration edge cases, serialization round-trips, approval flows, and stale mapping recovery. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Use bindgen! for WASM tool wrapper, add dev tool loading Three changes: 1. Rewrite src/tools/wasm/wrapper.rs to use wasmtime::component::bindgen! instead of manual linker.root().func_wrap(). This fixes the "component imports instance 'near:agent/host', but a matching implementation was not found in the linker" error. All 6 host functions (log, now-millis, workspace-read, http-request, secret-exists, tool-invoke) are now properly registered under the near:agent/host namespace. Also adds WASI support, credential injection, and leak detection for HTTP requests made by WASM tools. 2. Add dev tool loading to src/tools/wasm/loader.rs. During startup, the loader now also scans tools-src/*/target/wasm32-wasip2/release/ for build artifacts that are newer than installed copies. This means during development you just rebuild the WASM and restart the host; no manual copy step needed. Set IRONCLAW_TOOLS_SRC to override the source dir. 3. Wire up load_dev_tools() in main.rs alongside the existing load_from_dir() call. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: Wire main startup and CLI to use DB-backed settings main.rs now reloads Config from the database after connecting, attaches the store to the session manager for dual-write tokens, and loads MCP servers from DB instead of disk. ExtensionManager and MCP CLI commands use DB when available with disk fallback. 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
202665a55c
commit
ced83d5b4d
+640
-116
@@ -9,13 +9,14 @@ use uuid::Uuid;
|
||||
use crate::agent::compaction::ContextCompactor;
|
||||
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::{PendingApproval, Session, ThreadState};
|
||||
use crate::agent::session_manager::SessionManager;
|
||||
use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult};
|
||||
use crate::agent::{HeartbeatConfig as AgentHeartbeatConfig, MessageIntent, Router, Scheduler};
|
||||
use crate::channels::{ChannelManager, IncomingMessage, OutgoingResponse, StatusUpdate};
|
||||
use crate::config::{AgentConfig, HeartbeatConfig};
|
||||
use crate::config::{AgentConfig, HeartbeatConfig, RoutineConfig};
|
||||
use crate::context::ContextManager;
|
||||
use crate::context::JobContext;
|
||||
use crate::error::Error;
|
||||
@@ -77,6 +78,7 @@ pub struct Agent {
|
||||
session_manager: Arc<SessionManager>,
|
||||
context_monitor: ContextMonitor,
|
||||
heartbeat_config: Option<HeartbeatConfig>,
|
||||
routine_config: Option<RoutineConfig>,
|
||||
}
|
||||
|
||||
impl Agent {
|
||||
@@ -89,6 +91,7 @@ impl Agent {
|
||||
deps: AgentDeps,
|
||||
channels: ChannelManager,
|
||||
heartbeat_config: Option<HeartbeatConfig>,
|
||||
routine_config: Option<RoutineConfig>,
|
||||
context_manager: Option<Arc<ContextManager>>,
|
||||
session_manager: Option<Arc<SessionManager>>,
|
||||
) -> Self {
|
||||
@@ -116,6 +119,7 @@ impl Agent {
|
||||
session_manager,
|
||||
context_monitor: ContextMonitor::new(),
|
||||
heartbeat_config,
|
||||
routine_config,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -256,53 +260,28 @@ impl Agent {
|
||||
let channels = self.channels.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(response) = notify_rx.recv().await {
|
||||
// Route notification to configured channel/user, or broadcast to all
|
||||
match (¬ify_channel, ¬ify_user) {
|
||||
(Some(channel), Some(user)) => {
|
||||
// Send to specific channel and user
|
||||
if let Err(e) =
|
||||
channels.broadcast(channel, user, response.clone()).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 send heartbeat to {}/{}: {}",
|
||||
channel,
|
||||
user,
|
||||
"Failed to broadcast heartbeat to {}: {}",
|
||||
ch,
|
||||
e
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
"Heartbeat notification sent to {}/{}",
|
||||
channel,
|
||||
user
|
||||
);
|
||||
}
|
||||
}
|
||||
(None, Some(user)) => {
|
||||
// Broadcast to all channels for this user
|
||||
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
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// No explicit target, broadcast to all channels
|
||||
// for the default user so notifications actually
|
||||
// reach someone instead of vanishing into logs.
|
||||
let results = channels.broadcast_all("default", response).await;
|
||||
for (ch, result) in results {
|
||||
if let Err(e) = result {
|
||||
tracing::warn!(
|
||||
"Failed to broadcast heartbeat to {}: {}",
|
||||
ch,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -330,6 +309,85 @@ impl Agent {
|
||||
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,
|
||||
));
|
||||
|
||||
// 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
|
||||
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 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.
|
||||
|
||||
tracing::info!(
|
||||
"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::info!("Agent {} ready and listening", self.config.name);
|
||||
|
||||
@@ -374,6 +432,14 @@ impl Agent {
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -383,6 +449,9 @@ impl Agent {
|
||||
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?;
|
||||
|
||||
@@ -393,6 +462,11 @@ impl Agent {
|
||||
// Parse submission type first
|
||||
let submission = SubmissionParser::parse(&message.content);
|
||||
|
||||
// Hydrate thread from DB if it's a historical thread not in memory
|
||||
if let Some(ref external_thread_id) = message.thread_id {
|
||||
self.maybe_hydrate_thread(message, external_thread_id).await;
|
||||
}
|
||||
|
||||
// Resolve session and thread
|
||||
let (session, thread_id) = self
|
||||
.session_manager
|
||||
@@ -444,6 +518,9 @@ impl Agent {
|
||||
self.process_user_input(message, session, thread_id, &content)
|
||||
.await
|
||||
}
|
||||
Submission::SystemCommand { command, args } => {
|
||||
self.handle_system_command(&command, &args).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,
|
||||
@@ -515,6 +592,107 @@ impl Agent {
|
||||
}
|
||||
}
|
||||
|
||||
/// Hydrate a historical thread from DB into memory if not already present.
|
||||
///
|
||||
/// Called before `resolve_thread` so that the session manager finds the
|
||||
/// thread on lookup instead of creating a new one.
|
||||
///
|
||||
/// Creates an in-memory thread with the exact UUID the frontend sent,
|
||||
/// even when the conversation has zero messages (e.g. a brand-new
|
||||
/// assistant thread). Without this, `resolve_thread` would mint a
|
||||
/// fresh UUID and all messages would land in the wrong conversation.
|
||||
async fn maybe_hydrate_thread(&self, message: &IncomingMessage, external_thread_id: &str) {
|
||||
// Only hydrate UUID-shaped thread IDs (web gateway uses UUIDs)
|
||||
let thread_uuid = match Uuid::parse_str(external_thread_id) {
|
||||
Ok(id) => id,
|
||||
Err(_) => return,
|
||||
};
|
||||
|
||||
// Check if already in memory
|
||||
let session = self
|
||||
.session_manager
|
||||
.get_or_create_session(&message.user_id)
|
||||
.await;
|
||||
{
|
||||
let sess = session.lock().await;
|
||||
if sess.threads.contains_key(&thread_uuid) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Load history from DB (may be empty for a newly created thread).
|
||||
let mut chat_messages: Vec<ChatMessage> = Vec::new();
|
||||
let msg_count;
|
||||
|
||||
if let Some(store) = self.store() {
|
||||
let db_messages = store
|
||||
.list_conversation_messages(thread_uuid)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
msg_count = db_messages.len();
|
||||
chat_messages = db_messages
|
||||
.iter()
|
||||
.filter_map(|m| match m.role.as_str() {
|
||||
"user" => Some(ChatMessage::user(&m.content)),
|
||||
"assistant" => Some(ChatMessage::assistant(&m.content)),
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
} else {
|
||||
msg_count = 0;
|
||||
}
|
||||
|
||||
// Create thread with the historical ID and restore messages
|
||||
let session_id = {
|
||||
let sess = session.lock().await;
|
||||
sess.id
|
||||
};
|
||||
|
||||
let mut thread = crate::agent::session::Thread::with_id(thread_uuid, session_id);
|
||||
if !chat_messages.is_empty() {
|
||||
thread.restore_from_messages(chat_messages);
|
||||
}
|
||||
|
||||
// Restore response chain from conversation metadata
|
||||
if let Some(store) = self.store() {
|
||||
if let Ok(Some(metadata)) = store.get_conversation_metadata(thread_uuid).await {
|
||||
if let Some(rid) = metadata
|
||||
.get("last_response_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(String::from)
|
||||
{
|
||||
thread.last_response_id = Some(rid.clone());
|
||||
self.llm()
|
||||
.seed_response_chain(&thread_uuid.to_string(), rid);
|
||||
tracing::debug!("Restored response chain for thread {}", thread_uuid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Insert into session and register with session manager
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
sess.threads.insert(thread_uuid, thread);
|
||||
sess.active_thread = Some(thread_uuid);
|
||||
sess.last_active_at = chrono::Utc::now();
|
||||
}
|
||||
|
||||
self.session_manager
|
||||
.register_thread(
|
||||
&message.user_id,
|
||||
&message.channel,
|
||||
thread_uuid,
|
||||
Arc::clone(&session),
|
||||
)
|
||||
.await;
|
||||
|
||||
tracing::debug!(
|
||||
"Hydrated thread {} from DB ({} messages)",
|
||||
thread_uuid,
|
||||
msg_count
|
||||
);
|
||||
}
|
||||
|
||||
async fn process_user_input(
|
||||
&self,
|
||||
message: &IncomingMessage,
|
||||
@@ -694,6 +872,7 @@ impl Agent {
|
||||
match result {
|
||||
Ok(AgenticLoopResult::Response(response)) => {
|
||||
thread.complete_turn(&response);
|
||||
self.persist_response_chain(thread);
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
@@ -702,6 +881,10 @@ impl Agent {
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
|
||||
// Fire-and-forget: persist turn to DB
|
||||
self.persist_turn(thread_id, &message.user_id, content, Some(&response));
|
||||
|
||||
Ok(SubmissionResult::response(response))
|
||||
}
|
||||
Ok(AgenticLoopResult::NeedApproval { pending }) => {
|
||||
@@ -728,11 +911,95 @@ impl Agent {
|
||||
}
|
||||
Err(e) => {
|
||||
thread.fail_turn(e.to_string());
|
||||
|
||||
// Persist the user message even on failure
|
||||
self.persist_turn(thread_id, &message.user_id, content, None);
|
||||
|
||||
Ok(SubmissionResult::error(e.to_string()))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Fire-and-forget: persist a turn (user message + optional assistant response) to the DB.
|
||||
fn persist_turn(
|
||||
&self,
|
||||
thread_id: Uuid,
|
||||
user_id: &str,
|
||||
user_input: &str,
|
||||
response: Option<&str>,
|
||||
) {
|
||||
let store = match self.store() {
|
||||
Some(s) => Arc::clone(s),
|
||||
None => return,
|
||||
};
|
||||
|
||||
let user_id = user_id.to_string();
|
||||
let user_input = user_input.to_string();
|
||||
let response = response.map(String::from);
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = store
|
||||
.ensure_conversation(thread_id, "gateway", &user_id, None)
|
||||
.await
|
||||
{
|
||||
tracing::warn!("Failed to ensure conversation {}: {}", thread_id, e);
|
||||
return;
|
||||
}
|
||||
|
||||
if let Err(e) = store
|
||||
.add_conversation_message(thread_id, "user", &user_input)
|
||||
.await
|
||||
{
|
||||
tracing::warn!("Failed to persist user message: {}", e);
|
||||
return;
|
||||
}
|
||||
|
||||
if let Some(ref resp) = response {
|
||||
if let Err(e) = store
|
||||
.add_conversation_message(thread_id, "assistant", resp)
|
||||
.await
|
||||
{
|
||||
tracing::warn!("Failed to persist assistant message: {}", e);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Sync the provider's response chain ID to the thread and DB metadata.
|
||||
///
|
||||
/// Call after a successful agentic loop to persist the latest
|
||||
/// `previous_response_id` so chaining survives restarts.
|
||||
fn persist_response_chain(&self, thread: &mut crate::agent::session::Thread) {
|
||||
let tid = thread.id.to_string();
|
||||
let response_id = match self.llm().get_response_chain_id(&tid) {
|
||||
Some(rid) => rid,
|
||||
None => return,
|
||||
};
|
||||
|
||||
// Update in-memory thread
|
||||
thread.last_response_id = Some(response_id.clone());
|
||||
|
||||
// Fire-and-forget DB write
|
||||
let store = match self.store() {
|
||||
Some(s) => Arc::clone(s),
|
||||
None => return,
|
||||
};
|
||||
let thread_id = thread.id;
|
||||
tokio::spawn(async move {
|
||||
let val = serde_json::json!(response_id);
|
||||
if let Err(e) = store
|
||||
.update_conversation_metadata_field(thread_id, "last_response_id", &val)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(
|
||||
"Failed to persist response chain for thread {}: {}",
|
||||
thread_id,
|
||||
e
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Run the agentic loop: call LLM, execute tools, repeat until text response.
|
||||
///
|
||||
/// Returns `AgenticLoopResult::Response` on completion, or
|
||||
@@ -808,7 +1075,12 @@ impl Agent {
|
||||
// Call LLM with current context
|
||||
let context = ReasoningContext::new()
|
||||
.with_messages(context_messages.clone())
|
||||
.with_tools(tool_defs);
|
||||
.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 result = reasoning.respond_with_tools(&context).await?;
|
||||
|
||||
@@ -832,13 +1104,16 @@ impl Agent {
|
||||
// Tools have been executed or we've tried multiple times, return response
|
||||
return Ok(AgenticLoopResult::Response(text));
|
||||
}
|
||||
RespondResult::ToolCalls(tool_calls) => {
|
||||
RespondResult::ToolCalls {
|
||||
tool_calls,
|
||||
content,
|
||||
} => {
|
||||
tools_executed = true;
|
||||
|
||||
// Add the assistant message with tool_calls to context.
|
||||
// OpenAI-compatible APIs require this before tool-result messages.
|
||||
// OpenAI protocol requires this before tool-result messages.
|
||||
context_messages.push(ChatMessage::assistant_with_tool_calls(
|
||||
"",
|
||||
content,
|
||||
tool_calls.clone(),
|
||||
));
|
||||
|
||||
@@ -960,10 +1235,26 @@ impl Agent {
|
||||
if let Some((ext_name, instructions)) =
|
||||
detect_auth_awaiting(&tc.name, &tool_result)
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
||||
thread.enter_auth_mode(ext_name);
|
||||
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));
|
||||
}
|
||||
|
||||
@@ -1024,19 +1315,59 @@ impl Agent {
|
||||
.into());
|
||||
}
|
||||
|
||||
// Execute with timeout
|
||||
let result = tokio::time::timeout(std::time::Duration::from_secs(60), async {
|
||||
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
|
||||
.map_err(|_| crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout: std::time::Duration::from_secs(60),
|
||||
})?
|
||||
.map_err(|e| crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
.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| {
|
||||
@@ -1380,10 +1711,11 @@ impl Agent {
|
||||
if let Some((ext_name, instructions)) =
|
||||
detect_auth_awaiting(&pending.tool_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);
|
||||
thread.enter_auth_mode(ext_name.clone());
|
||||
thread.complete_turn(&instructions);
|
||||
}
|
||||
}
|
||||
@@ -1391,7 +1723,12 @@ impl Agent {
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::Status("Awaiting token".into()),
|
||||
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;
|
||||
@@ -1434,6 +1771,7 @@ impl Agent {
|
||||
match result {
|
||||
Ok(AgenticLoopResult::Response(response)) => {
|
||||
thread.complete_turn(&response);
|
||||
self.persist_response_chain(thread);
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
@@ -1532,16 +1870,6 @@ impl Agent {
|
||||
pending.extension_name
|
||||
);
|
||||
|
||||
// Notify via channel status so the response doesn't echo the token
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::Status("Authenticated, loading tools...".into()),
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
|
||||
// Auto-activate so tools are available immediately after auth
|
||||
match ext_mgr.activate(&pending.extension_name).await {
|
||||
Ok(activate_result) => {
|
||||
@@ -1551,10 +1879,23 @@ impl Agent {
|
||||
} else {
|
||||
format!("\n\nTools: {}", activate_result.tools_loaded.join(", "))
|
||||
};
|
||||
Ok(Some(format!(
|
||||
let msg = format!(
|
||||
"{} authenticated and activated ({} tools loaded).{}",
|
||||
pending.extension_name, tool_count, tool_list
|
||||
)))
|
||||
);
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::AuthCompleted {
|
||||
extension_name: pending.extension_name.clone(),
|
||||
success: true,
|
||||
message: msg.clone(),
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
Ok(Some(msg))
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
@@ -1562,16 +1903,29 @@ impl Agent {
|
||||
pending.extension_name,
|
||||
e
|
||||
);
|
||||
Ok(Some(format!(
|
||||
let msg = format!(
|
||||
"{} authenticated successfully, but activation failed: {}. \
|
||||
Try activating manually.",
|
||||
pending.extension_name, e
|
||||
)))
|
||||
);
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::AuthCompleted {
|
||||
extension_name: pending.extension_name.clone(),
|
||||
success: true,
|
||||
message: msg.clone(),
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
Ok(Some(msg))
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(result) => {
|
||||
// Unexpected state, re-enter auth mode
|
||||
// Invalid token, re-enter auth mode
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
||||
@@ -1580,13 +1934,43 @@ impl Agent {
|
||||
}
|
||||
let msg = result
|
||||
.instructions
|
||||
.clone()
|
||||
.unwrap_or_else(|| "Invalid token. Please try again.".to_string());
|
||||
// Re-emit AuthRequired so web UI re-shows the card
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::AuthRequired {
|
||||
extension_name: pending.extension_name.clone(),
|
||||
instructions: Some(msg.clone()),
|
||||
auth_url: result.auth_url,
|
||||
setup_url: result.setup_url,
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
Ok(Some(msg))
|
||||
}
|
||||
Err(e) => {
|
||||
let msg = format!(
|
||||
"Authentication failed for {}: {}",
|
||||
pending.extension_name, e
|
||||
);
|
||||
let _ = self
|
||||
.channels
|
||||
.send_status(
|
||||
&message.channel,
|
||||
StatusUpdate::AuthCompleted {
|
||||
extension_name: pending.extension_name.clone(),
|
||||
success: false,
|
||||
message: msg.clone(),
|
||||
},
|
||||
&message.metadata,
|
||||
)
|
||||
.await;
|
||||
Ok(Some(msg))
|
||||
}
|
||||
Err(e) => Ok(Some(format!(
|
||||
"Authentication failed for {}: {}",
|
||||
pending.extension_name, e
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1937,40 +2321,49 @@ impl Agent {
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_command(
|
||||
/// Handle system commands that bypass thread-state checks entirely.
|
||||
async fn handle_system_command(
|
||||
&self,
|
||||
command: &str,
|
||||
_args: &[String],
|
||||
) -> Result<Option<String>, Error> {
|
||||
args: &[String],
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
match command {
|
||||
"help" => Ok(Some(
|
||||
r#"Commands:
|
||||
/job <desc> - Create a job
|
||||
/status [id] - Check job status
|
||||
/cancel <id> - Cancel a job
|
||||
/list - List all jobs
|
||||
/help <job_id> - Help a stuck job
|
||||
"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",
|
||||
))),
|
||||
|
||||
/undo - Undo last turn
|
||||
/redo - Redo undone turn
|
||||
/compact - Compress context
|
||||
/clear - Clear thread
|
||||
/interrupt - Stop current turn
|
||||
/thread new - New thread
|
||||
/thread <id> - Switch thread
|
||||
/resume <id> - Resume checkpoint
|
||||
"ping" => Ok(SubmissionResult::response("pong!")),
|
||||
|
||||
/heartbeat - Run heartbeat check now
|
||||
/summarize - Summarize current thread
|
||||
/suggest - Suggest next steps
|
||||
|
||||
/quit - Exit"#
|
||||
.to_string(),
|
||||
)),
|
||||
|
||||
"ping" => Ok(Some("pong!".to_string())),
|
||||
|
||||
"version" => Ok(Some(format!(
|
||||
"version" => Ok(SubmissionResult::response(format!(
|
||||
"{} v{}",
|
||||
env!("CARGO_PKG_NAME"),
|
||||
env!("CARGO_PKG_VERSION")
|
||||
@@ -1978,12 +2371,113 @@ impl Agent {
|
||||
|
||||
"tools" => {
|
||||
let tools = self.tools().list().await;
|
||||
Ok(Some(format!("Available tools: {}", tools.join(", "))))
|
||||
Ok(SubmissionResult::response(format!(
|
||||
"Available tools: {}",
|
||||
tools.join(", ")
|
||||
)))
|
||||
}
|
||||
|
||||
_ => Ok(Some(format!("Unknown command: {}. Try /help", command))),
|
||||
"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).
|
||||
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),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parsed auth result fields for emitting StatusUpdate::AuthRequired.
|
||||
struct ParsedAuthData {
|
||||
auth_url: Option<String>,
|
||||
setup_url: Option<String>,
|
||||
}
|
||||
|
||||
/// Extract auth_url and setup_url from a tool_auth result JSON string.
|
||||
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.
|
||||
@@ -1994,7 +2488,7 @@ fn detect_auth_awaiting(
|
||||
tool_name: &str,
|
||||
result: &Result<String, Error>,
|
||||
) -> Option<(String, String)> {
|
||||
if tool_name != "tool_auth" {
|
||||
if tool_name != "tool_auth" && tool_name != "tool_activate" {
|
||||
return None;
|
||||
}
|
||||
let output = result.as_ref().ok()?;
|
||||
@@ -2078,4 +2572,34 @@ mod tests {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
+34
-3
@@ -29,7 +29,7 @@ use std::time::Duration;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::channels::OutgoingResponse;
|
||||
use crate::llm::{ChatMessage, CompletionRequest, LlmProvider};
|
||||
use crate::llm::{ChatMessage, CompletionRequest, FinishReason, LlmProvider};
|
||||
use crate::workspace::Workspace;
|
||||
|
||||
/// Configuration for the heartbeat runner.
|
||||
@@ -217,9 +217,26 @@ impl HeartbeatRunner {
|
||||
]
|
||||
};
|
||||
|
||||
// Use the model's context_length to set max_tokens. The API returns
|
||||
// the total context window; we cap output at half of that (the rest is
|
||||
// the prompt) with a floor of 4096.
|
||||
let max_tokens = match self.llm.model_metadata().await {
|
||||
Ok(meta) => {
|
||||
let from_api = meta.context_length.map(|ctx| ctx / 2).unwrap_or(4096);
|
||||
from_api.max(4096)
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"Could not fetch model metadata, using default max_tokens: {}",
|
||||
e
|
||||
);
|
||||
4096
|
||||
}
|
||||
};
|
||||
|
||||
let request = CompletionRequest::new(messages)
|
||||
.with_max_tokens(1024)
|
||||
.with_temperature(0.3); // Lower temperature for more focused responses
|
||||
.with_max_tokens(max_tokens)
|
||||
.with_temperature(0.3);
|
||||
|
||||
let response = match self.llm.complete(request).await {
|
||||
Ok(r) => r,
|
||||
@@ -228,6 +245,20 @@ impl HeartbeatRunner {
|
||||
|
||||
let content = response.content.trim();
|
||||
|
||||
// Guard against empty content. Reasoning models (e.g. GLM-4.7) may
|
||||
// burn all output tokens on chain-of-thought and return content: null.
|
||||
if content.is_empty() {
|
||||
return if response.finish_reason == FinishReason::Length {
|
||||
HeartbeatResult::Failed(
|
||||
"LLM response was truncated (finish_reason=length) with no content. \
|
||||
The model may have exhausted its token budget on reasoning."
|
||||
.to_string(),
|
||||
)
|
||||
} else {
|
||||
HeartbeatResult::Failed("LLM returned empty content.".to_string())
|
||||
};
|
||||
}
|
||||
|
||||
// Check if nothing needs attention
|
||||
if content == "HEARTBEAT_OK" || content.contains("HEARTBEAT_OK") {
|
||||
return HeartbeatResult::Ok;
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
//! - Tool invocation with safety
|
||||
//! - Self-repair for stuck jobs
|
||||
//! - Proactive heartbeat execution
|
||||
//! - Routine-based scheduled and reactive jobs
|
||||
//! - Turn-based session management with undo
|
||||
//! - Context compaction for long conversations
|
||||
|
||||
@@ -14,6 +15,8 @@ pub mod compaction;
|
||||
pub mod context_monitor;
|
||||
mod heartbeat;
|
||||
mod router;
|
||||
pub mod routine;
|
||||
pub mod routine_engine;
|
||||
mod scheduler;
|
||||
mod self_repair;
|
||||
pub mod session;
|
||||
@@ -28,6 +31,8 @@ pub use compaction::{CompactionResult, ContextCompactor};
|
||||
pub use context_monitor::{CompactionStrategy, ContextBreakdown, ContextMonitor};
|
||||
pub use heartbeat::{HeartbeatConfig, HeartbeatResult, HeartbeatRunner, spawn_heartbeat};
|
||||
pub use router::{MessageIntent, Router};
|
||||
pub use routine::{Routine, RoutineAction, RoutineRun, Trigger};
|
||||
pub use routine_engine::RoutineEngine;
|
||||
pub use scheduler::Scheduler;
|
||||
pub use self_repair::{BrokenTool, RepairResult, RepairTask, SelfRepair, StuckJob};
|
||||
pub use session::{PendingApproval, PendingAuth, Session, Thread, ThreadState, Turn, TurnState};
|
||||
|
||||
@@ -0,0 +1,509 @@
|
||||
//! Core types for the routines system.
|
||||
//!
|
||||
//! A routine is a named, persistent, user-owned task with a trigger and an action.
|
||||
//! Each routine fires independently when its trigger condition is met, with only
|
||||
//! that routine's prompt and context sent to the LLM.
|
||||
//!
|
||||
//! ```text
|
||||
//! ┌──────────┐ ┌─────────┐ ┌──────────────────┐
|
||||
//! │ Trigger │────▶│ Engine │────▶│ Execution Mode │
|
||||
//! │ cron/event│ │guardrail│ │lightweight│full_job│
|
||||
//! │ webhook │ │ check │ └──────────────────┘
|
||||
//! │ manual │ └─────────┘ │
|
||||
//! └──────────┘ ▼
|
||||
//! ┌──────────────┐
|
||||
//! │ Notify user │
|
||||
//! │ if needed │
|
||||
//! └──────────────┘
|
||||
//! ```
|
||||
|
||||
use std::collections::hash_map::DefaultHasher;
|
||||
use std::hash::{Hash, Hasher};
|
||||
use std::str::FromStr;
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
/// A routine is a named, persistent, user-owned task with a trigger and an action.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Routine {
|
||||
pub id: Uuid,
|
||||
pub name: String,
|
||||
pub description: String,
|
||||
pub user_id: String,
|
||||
pub enabled: bool,
|
||||
pub trigger: Trigger,
|
||||
pub action: RoutineAction,
|
||||
pub guardrails: RoutineGuardrails,
|
||||
pub notify: NotifyConfig,
|
||||
|
||||
// Runtime state (DB-managed)
|
||||
pub last_run_at: Option<DateTime<Utc>>,
|
||||
pub next_fire_at: Option<DateTime<Utc>>,
|
||||
pub run_count: u64,
|
||||
pub consecutive_failures: u32,
|
||||
pub state: serde_json::Value,
|
||||
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// When a routine should fire.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "type", rename_all = "snake_case")]
|
||||
pub enum Trigger {
|
||||
/// Fire on a cron schedule (e.g. "0 9 * * MON-FRI" or "every 2h").
|
||||
Cron { schedule: String },
|
||||
/// Fire when a channel message matches a pattern.
|
||||
Event {
|
||||
/// Optional channel filter (e.g. "telegram", "slack").
|
||||
channel: Option<String>,
|
||||
/// Regex pattern to match against message content.
|
||||
pattern: String,
|
||||
},
|
||||
/// Fire on incoming webhook POST to /hooks/routine/{id}.
|
||||
Webhook {
|
||||
/// Optional webhook path suffix (defaults to routine id).
|
||||
path: Option<String>,
|
||||
/// Optional shared secret for HMAC validation.
|
||||
secret: Option<String>,
|
||||
},
|
||||
/// Only fires via tool call or CLI.
|
||||
Manual,
|
||||
}
|
||||
|
||||
impl Trigger {
|
||||
/// The string tag stored in the DB trigger_type column.
|
||||
pub fn type_tag(&self) -> &'static str {
|
||||
match self {
|
||||
Trigger::Cron { .. } => "cron",
|
||||
Trigger::Event { .. } => "event",
|
||||
Trigger::Webhook { .. } => "webhook",
|
||||
Trigger::Manual => "manual",
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse a trigger from its DB representation.
|
||||
pub fn from_db(trigger_type: &str, config: serde_json::Value) -> Result<Self, String> {
|
||||
match trigger_type {
|
||||
"cron" => {
|
||||
let schedule = config
|
||||
.get("schedule")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("cron trigger missing 'schedule'")?
|
||||
.to_string();
|
||||
Ok(Trigger::Cron { schedule })
|
||||
}
|
||||
"event" => {
|
||||
let pattern = config
|
||||
.get("pattern")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("event trigger missing 'pattern'")?
|
||||
.to_string();
|
||||
let channel = config
|
||||
.get("channel")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(String::from);
|
||||
Ok(Trigger::Event { channel, pattern })
|
||||
}
|
||||
"webhook" => {
|
||||
let path = config
|
||||
.get("path")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(String::from);
|
||||
let secret = config
|
||||
.get("secret")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(String::from);
|
||||
Ok(Trigger::Webhook { path, secret })
|
||||
}
|
||||
"manual" => Ok(Trigger::Manual),
|
||||
other => Err(format!("unknown trigger type: {other}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// Serialize trigger-specific config to JSON for DB storage.
|
||||
pub fn to_config_json(&self) -> serde_json::Value {
|
||||
match self {
|
||||
Trigger::Cron { schedule } => serde_json::json!({ "schedule": schedule }),
|
||||
Trigger::Event { channel, pattern } => serde_json::json!({
|
||||
"pattern": pattern,
|
||||
"channel": channel,
|
||||
}),
|
||||
Trigger::Webhook { path, secret } => serde_json::json!({
|
||||
"path": path,
|
||||
"secret": secret,
|
||||
}),
|
||||
Trigger::Manual => serde_json::json!({}),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// What happens when a routine fires.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "type", rename_all = "snake_case")]
|
||||
pub enum RoutineAction {
|
||||
/// Single LLM call, no tools. Cheap and fast.
|
||||
Lightweight {
|
||||
/// The prompt sent to the LLM.
|
||||
prompt: String,
|
||||
/// Workspace paths to load as context (e.g. ["context/priorities.md"]).
|
||||
#[serde(default)]
|
||||
context_paths: Vec<String>,
|
||||
/// Max output tokens (default: 4096).
|
||||
#[serde(default = "default_max_tokens")]
|
||||
max_tokens: u32,
|
||||
},
|
||||
/// Full multi-turn worker job with tool access.
|
||||
FullJob {
|
||||
/// Job title for the scheduler.
|
||||
title: String,
|
||||
/// Job description / initial prompt.
|
||||
description: String,
|
||||
/// Max reasoning iterations (default: 10).
|
||||
#[serde(default = "default_max_iterations")]
|
||||
max_iterations: u32,
|
||||
},
|
||||
}
|
||||
|
||||
fn default_max_tokens() -> u32 {
|
||||
4096
|
||||
}
|
||||
|
||||
fn default_max_iterations() -> u32 {
|
||||
10
|
||||
}
|
||||
|
||||
impl RoutineAction {
|
||||
/// The string tag stored in the DB action_type column.
|
||||
pub fn type_tag(&self) -> &'static str {
|
||||
match self {
|
||||
RoutineAction::Lightweight { .. } => "lightweight",
|
||||
RoutineAction::FullJob { .. } => "full_job",
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse an action from its DB representation.
|
||||
pub fn from_db(action_type: &str, config: serde_json::Value) -> Result<Self, String> {
|
||||
match action_type {
|
||||
"lightweight" => {
|
||||
let prompt = config
|
||||
.get("prompt")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("lightweight action missing 'prompt'")?
|
||||
.to_string();
|
||||
let context_paths = config
|
||||
.get("context_paths")
|
||||
.and_then(|v| v.as_array())
|
||||
.map(|arr| {
|
||||
arr.iter()
|
||||
.filter_map(|v| v.as_str().map(String::from))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
let max_tokens = config
|
||||
.get("max_tokens")
|
||||
.and_then(|v| v.as_u64())
|
||||
.unwrap_or(default_max_tokens() as u64) as u32;
|
||||
Ok(RoutineAction::Lightweight {
|
||||
prompt,
|
||||
context_paths,
|
||||
max_tokens,
|
||||
})
|
||||
}
|
||||
"full_job" => {
|
||||
let title = config
|
||||
.get("title")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("full_job action missing 'title'")?
|
||||
.to_string();
|
||||
let description = config
|
||||
.get("description")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("full_job action missing 'description'")?
|
||||
.to_string();
|
||||
let max_iterations = config
|
||||
.get("max_iterations")
|
||||
.and_then(|v| v.as_u64())
|
||||
.unwrap_or(default_max_iterations() as u64)
|
||||
as u32;
|
||||
Ok(RoutineAction::FullJob {
|
||||
title,
|
||||
description,
|
||||
max_iterations,
|
||||
})
|
||||
}
|
||||
other => Err(format!("unknown action type: {other}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// Serialize action config to JSON for DB storage.
|
||||
pub fn to_config_json(&self) -> serde_json::Value {
|
||||
match self {
|
||||
RoutineAction::Lightweight {
|
||||
prompt,
|
||||
context_paths,
|
||||
max_tokens,
|
||||
} => serde_json::json!({
|
||||
"prompt": prompt,
|
||||
"context_paths": context_paths,
|
||||
"max_tokens": max_tokens,
|
||||
}),
|
||||
RoutineAction::FullJob {
|
||||
title,
|
||||
description,
|
||||
max_iterations,
|
||||
} => serde_json::json!({
|
||||
"title": title,
|
||||
"description": description,
|
||||
"max_iterations": max_iterations,
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Guardrails to prevent runaway execution.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RoutineGuardrails {
|
||||
/// Minimum time between fires.
|
||||
pub cooldown: Duration,
|
||||
/// Max simultaneous runs of this routine.
|
||||
pub max_concurrent: u32,
|
||||
/// Window for content-hash dedup (event triggers). None = no dedup.
|
||||
pub dedup_window: Option<Duration>,
|
||||
}
|
||||
|
||||
impl Default for RoutineGuardrails {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
cooldown: Duration::from_secs(300),
|
||||
max_concurrent: 1,
|
||||
dedup_window: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Notification preferences for a routine.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct NotifyConfig {
|
||||
/// Channel to notify on (None = default/broadcast all).
|
||||
pub channel: Option<String>,
|
||||
/// User to notify.
|
||||
pub user: String,
|
||||
/// Notify when routine produces actionable output.
|
||||
pub on_attention: bool,
|
||||
/// Notify when routine errors.
|
||||
pub on_failure: bool,
|
||||
/// Notify when routine runs with no findings.
|
||||
pub on_success: bool,
|
||||
}
|
||||
|
||||
impl Default for NotifyConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
channel: None,
|
||||
user: "default".to_string(),
|
||||
on_attention: true,
|
||||
on_failure: true,
|
||||
on_success: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Status of a routine run.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum RunStatus {
|
||||
Running,
|
||||
Ok,
|
||||
Attention,
|
||||
Failed,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for RunStatus {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
RunStatus::Running => write!(f, "running"),
|
||||
RunStatus::Ok => write!(f, "ok"),
|
||||
RunStatus::Attention => write!(f, "attention"),
|
||||
RunStatus::Failed => write!(f, "failed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl FromStr for RunStatus {
|
||||
type Err = String;
|
||||
fn from_str(s: &str) -> Result<Self, Self::Err> {
|
||||
match s {
|
||||
"running" => Ok(RunStatus::Running),
|
||||
"ok" => Ok(RunStatus::Ok),
|
||||
"attention" => Ok(RunStatus::Attention),
|
||||
"failed" => Ok(RunStatus::Failed),
|
||||
other => Err(format!("unknown run status: {other}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A single execution of a routine.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RoutineRun {
|
||||
pub id: Uuid,
|
||||
pub routine_id: Uuid,
|
||||
pub trigger_type: String,
|
||||
pub trigger_detail: Option<String>,
|
||||
pub started_at: DateTime<Utc>,
|
||||
pub completed_at: Option<DateTime<Utc>>,
|
||||
pub status: RunStatus,
|
||||
pub result_summary: Option<String>,
|
||||
pub tokens_used: Option<i32>,
|
||||
pub job_id: Option<Uuid>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// Compute a content hash for event dedup.
|
||||
pub fn content_hash(content: &str) -> u64 {
|
||||
let mut hasher = DefaultHasher::new();
|
||||
content.hash(&mut hasher);
|
||||
hasher.finish()
|
||||
}
|
||||
|
||||
/// Parse a cron expression and compute the next fire time from now.
|
||||
pub fn next_cron_fire(schedule: &str) -> Result<Option<DateTime<Utc>>, String> {
|
||||
let cron_schedule =
|
||||
cron::Schedule::from_str(schedule).map_err(|e| format!("invalid cron: {e}"))?;
|
||||
Ok(cron_schedule.upcoming(Utc).next())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::agent::routine::{
|
||||
RoutineAction, RoutineGuardrails, RunStatus, Trigger, content_hash, next_cron_fire,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn test_trigger_roundtrip() {
|
||||
let trigger = Trigger::Cron {
|
||||
schedule: "0 9 * * MON-FRI".to_string(),
|
||||
};
|
||||
let json = trigger.to_config_json();
|
||||
let parsed = Trigger::from_db("cron", json).expect("parse cron");
|
||||
assert!(matches!(parsed, Trigger::Cron { schedule } if schedule == "0 9 * * MON-FRI"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_event_trigger_roundtrip() {
|
||||
let trigger = Trigger::Event {
|
||||
channel: Some("telegram".to_string()),
|
||||
pattern: r"deploy\s+\w+".to_string(),
|
||||
};
|
||||
let json = trigger.to_config_json();
|
||||
let parsed = Trigger::from_db("event", json).expect("parse event");
|
||||
assert!(matches!(parsed, Trigger::Event { channel, pattern }
|
||||
if channel == Some("telegram".to_string()) && pattern == r"deploy\s+\w+"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_action_lightweight_roundtrip() {
|
||||
let action = RoutineAction::Lightweight {
|
||||
prompt: "Check PRs".to_string(),
|
||||
context_paths: vec!["context/priorities.md".to_string()],
|
||||
max_tokens: 2048,
|
||||
};
|
||||
let json = action.to_config_json();
|
||||
let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight");
|
||||
assert!(
|
||||
matches!(parsed, RoutineAction::Lightweight { prompt, context_paths, max_tokens }
|
||||
if prompt == "Check PRs" && context_paths.len() == 1 && max_tokens == 2048)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_action_full_job_roundtrip() {
|
||||
let action = RoutineAction::FullJob {
|
||||
title: "Deploy review".to_string(),
|
||||
description: "Review and deploy pending changes".to_string(),
|
||||
max_iterations: 5,
|
||||
};
|
||||
let json = action.to_config_json();
|
||||
let parsed = RoutineAction::from_db("full_job", json).expect("parse full_job");
|
||||
assert!(
|
||||
matches!(parsed, RoutineAction::FullJob { title, max_iterations, .. }
|
||||
if title == "Deploy review" && max_iterations == 5)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_run_status_display_parse() {
|
||||
for status in [
|
||||
RunStatus::Running,
|
||||
RunStatus::Ok,
|
||||
RunStatus::Attention,
|
||||
RunStatus::Failed,
|
||||
] {
|
||||
let s = status.to_string();
|
||||
let parsed: RunStatus = s.parse().expect("parse status");
|
||||
assert_eq!(parsed, status);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_content_hash_deterministic() {
|
||||
let h1 = content_hash("deploy production");
|
||||
let h2 = content_hash("deploy production");
|
||||
assert_eq!(h1, h2);
|
||||
|
||||
let h3 = content_hash("deploy staging");
|
||||
assert_ne!(h1, h3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_next_cron_fire_valid() {
|
||||
// Every minute should always have a next fire
|
||||
let next = next_cron_fire("* * * * * *").expect("valid cron");
|
||||
assert!(next.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_next_cron_fire_invalid() {
|
||||
let result = next_cron_fire("not a cron");
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_guardrails_default() {
|
||||
let g = RoutineGuardrails::default();
|
||||
assert_eq!(g.cooldown.as_secs(), 300);
|
||||
assert_eq!(g.max_concurrent, 1);
|
||||
assert!(g.dedup_window.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_trigger_type_tag() {
|
||||
assert_eq!(
|
||||
Trigger::Cron {
|
||||
schedule: String::new()
|
||||
}
|
||||
.type_tag(),
|
||||
"cron"
|
||||
);
|
||||
assert_eq!(
|
||||
Trigger::Event {
|
||||
channel: None,
|
||||
pattern: String::new()
|
||||
}
|
||||
.type_tag(),
|
||||
"event"
|
||||
);
|
||||
assert_eq!(
|
||||
Trigger::Webhook {
|
||||
path: None,
|
||||
secret: None
|
||||
}
|
||||
.type_tag(),
|
||||
"webhook"
|
||||
);
|
||||
assert_eq!(Trigger::Manual.type_tag(), "manual");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,606 @@
|
||||
//! Routine execution engine.
|
||||
//!
|
||||
//! Handles loading routines, checking triggers, enforcing guardrails,
|
||||
//! and executing both lightweight (single LLM call) and full-job routines.
|
||||
//!
|
||||
//! The engine runs two independent loops:
|
||||
//! - A **cron ticker** that polls the DB every N seconds for due cron routines
|
||||
//! - An **event matcher** called synchronously from the agent main loop
|
||||
//!
|
||||
//! Lightweight routines execute inline (single LLM call, no scheduler slot).
|
||||
//! Full-job routines are delegated to the existing `Scheduler`.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::Utc;
|
||||
use regex::Regex;
|
||||
use tokio::sync::{RwLock, mpsc};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::agent::routine::{
|
||||
NotifyConfig, Routine, RoutineAction, RoutineRun, RunStatus, Trigger, next_cron_fire,
|
||||
};
|
||||
use crate::channels::{IncomingMessage, OutgoingResponse};
|
||||
use crate::config::RoutineConfig;
|
||||
use crate::history::Store;
|
||||
use crate::llm::{ChatMessage, CompletionRequest, FinishReason, LlmProvider};
|
||||
use crate::workspace::Workspace;
|
||||
|
||||
/// The routine execution engine.
|
||||
pub struct RoutineEngine {
|
||||
config: RoutineConfig,
|
||||
store: Arc<Store>,
|
||||
llm: Arc<dyn LlmProvider>,
|
||||
workspace: Arc<Workspace>,
|
||||
/// Sender for notifications (routed to channel manager).
|
||||
notify_tx: mpsc::Sender<OutgoingResponse>,
|
||||
/// Currently running routine count (across all routines).
|
||||
running_count: Arc<RwLock<usize>>,
|
||||
/// Compiled event regex cache: routine_id -> compiled regex.
|
||||
event_cache: Arc<RwLock<Vec<(Uuid, Routine, Regex)>>>,
|
||||
}
|
||||
|
||||
impl RoutineEngine {
|
||||
pub fn new(
|
||||
config: RoutineConfig,
|
||||
store: Arc<Store>,
|
||||
llm: Arc<dyn LlmProvider>,
|
||||
workspace: Arc<Workspace>,
|
||||
notify_tx: mpsc::Sender<OutgoingResponse>,
|
||||
) -> Self {
|
||||
Self {
|
||||
config,
|
||||
store,
|
||||
llm,
|
||||
workspace,
|
||||
notify_tx,
|
||||
running_count: Arc::new(RwLock::new(0)),
|
||||
event_cache: Arc::new(RwLock::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
|
||||
/// Refresh the in-memory event trigger cache from DB.
|
||||
pub async fn refresh_event_cache(&self) {
|
||||
match self.store.list_event_routines().await {
|
||||
Ok(routines) => {
|
||||
let mut cache = Vec::new();
|
||||
for routine in routines {
|
||||
if let Trigger::Event { ref pattern, .. } = routine.trigger {
|
||||
match Regex::new(pattern) {
|
||||
Ok(re) => cache.push((routine.id, routine.clone(), re)),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
routine = %routine.name,
|
||||
"Invalid event regex '{}': {}",
|
||||
pattern, e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let count = cache.len();
|
||||
*self.event_cache.write().await = cache;
|
||||
tracing::debug!("Refreshed event cache: {} routines", count);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to refresh event cache: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Check incoming message against event triggers. Returns number of routines fired.
|
||||
///
|
||||
/// Called synchronously from the main loop after handle_message(). The actual
|
||||
/// execution is spawned async so this returns quickly.
|
||||
pub async fn check_event_triggers(&self, message: &IncomingMessage) -> usize {
|
||||
let cache = self.event_cache.read().await;
|
||||
let mut fired = 0;
|
||||
|
||||
for (_, routine, re) in cache.iter() {
|
||||
// Channel filter
|
||||
if let Trigger::Event {
|
||||
channel: Some(ch), ..
|
||||
} = &routine.trigger
|
||||
{
|
||||
if ch != &message.channel {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// Regex match
|
||||
if !re.is_match(&message.content) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Cooldown check
|
||||
if !self.check_cooldown(routine) {
|
||||
tracing::debug!(routine = %routine.name, "Skipped: cooldown active");
|
||||
continue;
|
||||
}
|
||||
|
||||
// Concurrent run check
|
||||
if !self.check_concurrent(routine).await {
|
||||
tracing::debug!(routine = %routine.name, "Skipped: max concurrent reached");
|
||||
continue;
|
||||
}
|
||||
|
||||
// Global capacity check
|
||||
if *self.running_count.read().await >= self.config.max_concurrent_routines {
|
||||
tracing::warn!(routine = %routine.name, "Skipped: global max concurrent reached");
|
||||
continue;
|
||||
}
|
||||
|
||||
let detail = truncate(&message.content, 200);
|
||||
self.spawn_fire(routine.clone(), "event", Some(detail));
|
||||
fired += 1;
|
||||
}
|
||||
|
||||
fired
|
||||
}
|
||||
|
||||
/// Check all due cron routines and fire them. Called by the cron ticker.
|
||||
pub async fn check_cron_triggers(&self) {
|
||||
let routines = match self.store.list_due_cron_routines().await {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to load due cron routines: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
for routine in routines {
|
||||
if *self.running_count.read().await >= self.config.max_concurrent_routines {
|
||||
tracing::warn!("Global max concurrent routines reached, skipping remaining");
|
||||
break;
|
||||
}
|
||||
|
||||
if !self.check_cooldown(&routine) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if !self.check_concurrent(&routine).await {
|
||||
continue;
|
||||
}
|
||||
|
||||
let detail = if let Trigger::Cron { ref schedule } = routine.trigger {
|
||||
Some(schedule.clone())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
self.spawn_fire(routine, "cron", detail);
|
||||
}
|
||||
}
|
||||
|
||||
/// Fire a routine manually (from tool call or CLI).
|
||||
pub async fn fire_manual(&self, routine_id: Uuid) -> Result<Uuid, String> {
|
||||
let routine = self
|
||||
.store
|
||||
.get_routine(routine_id)
|
||||
.await
|
||||
.map_err(|e| format!("DB error: {e}"))?
|
||||
.ok_or_else(|| format!("routine {routine_id} not found"))?;
|
||||
|
||||
if !routine.enabled {
|
||||
return Err(format!("routine '{}' is disabled", routine.name));
|
||||
}
|
||||
|
||||
if !self.check_concurrent(&routine).await {
|
||||
return Err(format!(
|
||||
"routine '{}' already at max concurrent runs",
|
||||
routine.name
|
||||
));
|
||||
}
|
||||
|
||||
let run_id = Uuid::new_v4();
|
||||
let run = RoutineRun {
|
||||
id: run_id,
|
||||
routine_id: routine.id,
|
||||
trigger_type: "manual".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
|
||||
if let Err(e) = self.store.create_routine_run(&run).await {
|
||||
return Err(format!("failed to create run record: {e}"));
|
||||
}
|
||||
|
||||
// Execute inline for manual triggers (caller wants to wait)
|
||||
let engine = EngineContext {
|
||||
store: self.store.clone(),
|
||||
llm: self.llm.clone(),
|
||||
workspace: self.workspace.clone(),
|
||||
notify_tx: self.notify_tx.clone(),
|
||||
running_count: self.running_count.clone(),
|
||||
max_lightweight_tokens: self.config.max_lightweight_tokens,
|
||||
};
|
||||
|
||||
tokio::spawn(async move {
|
||||
execute_routine(engine, routine, run).await;
|
||||
});
|
||||
|
||||
Ok(run_id)
|
||||
}
|
||||
|
||||
/// Spawn a fire in a background task.
|
||||
fn spawn_fire(&self, routine: Routine, trigger_type: &str, trigger_detail: Option<String>) {
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: routine.id,
|
||||
trigger_type: trigger_type.to_string(),
|
||||
trigger_detail,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
|
||||
let engine = EngineContext {
|
||||
store: self.store.clone(),
|
||||
llm: self.llm.clone(),
|
||||
workspace: self.workspace.clone(),
|
||||
notify_tx: self.notify_tx.clone(),
|
||||
running_count: self.running_count.clone(),
|
||||
max_lightweight_tokens: self.config.max_lightweight_tokens,
|
||||
};
|
||||
|
||||
// Record the run in DB, then spawn execution
|
||||
let store = self.store.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = store.create_routine_run(&run).await {
|
||||
tracing::error!(routine = %routine.name, "Failed to record run: {}", e);
|
||||
return;
|
||||
}
|
||||
execute_routine(engine, routine, run).await;
|
||||
});
|
||||
}
|
||||
|
||||
fn check_cooldown(&self, routine: &Routine) -> bool {
|
||||
if let Some(last_run) = routine.last_run_at {
|
||||
let elapsed = Utc::now().signed_duration_since(last_run);
|
||||
let cooldown = chrono::Duration::from_std(routine.guardrails.cooldown)
|
||||
.unwrap_or(chrono::Duration::seconds(300));
|
||||
if elapsed < cooldown {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
async fn check_concurrent(&self, routine: &Routine) -> bool {
|
||||
match self.store.count_running_routine_runs(routine.id).await {
|
||||
Ok(count) => count < routine.guardrails.max_concurrent as i64,
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
routine = %routine.name,
|
||||
"Failed to check concurrent runs: {}", e
|
||||
);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared context passed to the execution function.
|
||||
struct EngineContext {
|
||||
store: Arc<Store>,
|
||||
llm: Arc<dyn LlmProvider>,
|
||||
workspace: Arc<Workspace>,
|
||||
notify_tx: mpsc::Sender<OutgoingResponse>,
|
||||
running_count: Arc<RwLock<usize>>,
|
||||
max_lightweight_tokens: u32,
|
||||
}
|
||||
|
||||
/// Execute a routine run. Handles both lightweight and full_job modes.
|
||||
async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun) {
|
||||
// Increment running count
|
||||
{
|
||||
let mut count = ctx.running_count.write().await;
|
||||
*count += 1;
|
||||
}
|
||||
|
||||
let result = match &routine.action {
|
||||
RoutineAction::Lightweight {
|
||||
prompt,
|
||||
context_paths,
|
||||
max_tokens,
|
||||
} => execute_lightweight(&ctx, &routine, prompt, context_paths, *max_tokens).await,
|
||||
RoutineAction::FullJob { description, .. } => {
|
||||
// Full job mode: for now, execute as lightweight with the description
|
||||
// as prompt. Full scheduler integration will come as a follow-up.
|
||||
tracing::info!(
|
||||
routine = %routine.name,
|
||||
"FullJob mode executing as lightweight (scheduler integration pending)"
|
||||
);
|
||||
execute_lightweight(&ctx, &routine, description, &[], ctx.max_lightweight_tokens).await
|
||||
}
|
||||
};
|
||||
|
||||
// Decrement running count
|
||||
{
|
||||
let mut count = ctx.running_count.write().await;
|
||||
*count = count.saturating_sub(1);
|
||||
}
|
||||
|
||||
// Process result
|
||||
let (status, summary, tokens) = match result {
|
||||
Ok(execution) => execution,
|
||||
Err(e) => {
|
||||
tracing::error!(routine = %routine.name, "Execution failed: {}", e);
|
||||
(RunStatus::Failed, Some(e), None)
|
||||
}
|
||||
};
|
||||
|
||||
// Complete the run record
|
||||
if let Err(e) = ctx
|
||||
.store
|
||||
.complete_routine_run(run.id, status, summary.as_deref(), tokens)
|
||||
.await
|
||||
{
|
||||
tracing::error!(routine = %routine.name, "Failed to complete run record: {}", e);
|
||||
}
|
||||
|
||||
// Update routine runtime state
|
||||
let now = Utc::now();
|
||||
let next_fire = if let Trigger::Cron { ref schedule } = routine.trigger {
|
||||
next_cron_fire(schedule).unwrap_or(None)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let new_failures = if status == RunStatus::Failed {
|
||||
routine.consecutive_failures + 1
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
if let Err(e) = ctx
|
||||
.store
|
||||
.update_routine_runtime(
|
||||
routine.id,
|
||||
now,
|
||||
next_fire,
|
||||
routine.run_count + 1,
|
||||
new_failures,
|
||||
&routine.state,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::error!(routine = %routine.name, "Failed to update runtime state: {}", e);
|
||||
}
|
||||
|
||||
// Send notifications based on config
|
||||
send_notification(
|
||||
&ctx.notify_tx,
|
||||
&routine.notify,
|
||||
&routine.name,
|
||||
status,
|
||||
summary.as_deref(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Execute a lightweight routine (single LLM call).
|
||||
async fn execute_lightweight(
|
||||
ctx: &EngineContext,
|
||||
routine: &Routine,
|
||||
prompt: &str,
|
||||
context_paths: &[String],
|
||||
max_tokens: u32,
|
||||
) -> Result<(RunStatus, Option<String>, Option<i32>), String> {
|
||||
// Load context from workspace
|
||||
let mut context_parts = Vec::new();
|
||||
for path in context_paths {
|
||||
match ctx.workspace.read(path).await {
|
||||
Ok(doc) => {
|
||||
context_parts.push(format!("## {}\n\n{}", path, doc.content));
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::debug!(
|
||||
routine = %routine.name,
|
||||
"Failed to read context path {}: {}", path, e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Load routine state from workspace
|
||||
let state_path = format!("routines/{}/state.md", routine.name);
|
||||
let state_content = match ctx.workspace.read(&state_path).await {
|
||||
Ok(doc) => Some(doc.content),
|
||||
Err(_) => None,
|
||||
};
|
||||
|
||||
// Build the prompt
|
||||
let mut full_prompt = String::new();
|
||||
full_prompt.push_str(prompt);
|
||||
|
||||
if !context_parts.is_empty() {
|
||||
full_prompt.push_str("\n\n---\n\n# Context\n\n");
|
||||
full_prompt.push_str(&context_parts.join("\n\n"));
|
||||
}
|
||||
|
||||
if let Some(state) = &state_content {
|
||||
full_prompt.push_str("\n\n---\n\n# Previous State\n\n");
|
||||
full_prompt.push_str(state);
|
||||
}
|
||||
|
||||
full_prompt.push_str(
|
||||
"\n\n---\n\nIf nothing needs attention, reply EXACTLY with: ROUTINE_OK\n\
|
||||
If something needs attention, provide a concise summary.",
|
||||
);
|
||||
|
||||
// Get system prompt
|
||||
let system_prompt = match ctx.workspace.system_prompt().await {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::warn!(routine = %routine.name, "Failed to get system prompt: {}", e);
|
||||
String::new()
|
||||
}
|
||||
};
|
||||
|
||||
let messages = if system_prompt.is_empty() {
|
||||
vec![ChatMessage::user(&full_prompt)]
|
||||
} else {
|
||||
vec![
|
||||
ChatMessage::system(&system_prompt),
|
||||
ChatMessage::user(&full_prompt),
|
||||
]
|
||||
};
|
||||
|
||||
// Determine max_tokens from model metadata with fallback
|
||||
let effective_max_tokens = match ctx.llm.model_metadata().await {
|
||||
Ok(meta) => {
|
||||
let from_api = meta.context_length.map(|ctx| ctx / 2).unwrap_or(max_tokens);
|
||||
from_api.max(max_tokens)
|
||||
}
|
||||
Err(_) => max_tokens,
|
||||
};
|
||||
|
||||
let request = CompletionRequest::new(messages)
|
||||
.with_max_tokens(effective_max_tokens)
|
||||
.with_temperature(0.3);
|
||||
|
||||
let response = ctx
|
||||
.llm
|
||||
.complete(request)
|
||||
.await
|
||||
.map_err(|e| format!("LLM call failed: {e}"))?;
|
||||
|
||||
let content = response.content.trim();
|
||||
let tokens_used = Some((response.input_tokens + response.output_tokens) as i32);
|
||||
|
||||
// Empty content guard (same as heartbeat)
|
||||
if content.is_empty() {
|
||||
return if response.finish_reason == FinishReason::Length {
|
||||
Err(
|
||||
"LLM response truncated (finish_reason=length) with no content. \
|
||||
Model may have exhausted token budget on reasoning."
|
||||
.to_string(),
|
||||
)
|
||||
} else {
|
||||
Err("LLM returned empty content.".to_string())
|
||||
};
|
||||
}
|
||||
|
||||
// Check for the "nothing to do" sentinel
|
||||
if content == "ROUTINE_OK" || content.contains("ROUTINE_OK") {
|
||||
return Ok((RunStatus::Ok, None, tokens_used));
|
||||
}
|
||||
|
||||
Ok((RunStatus::Attention, Some(content.to_string()), tokens_used))
|
||||
}
|
||||
|
||||
/// Send a notification based on the routine's notify config and run status.
|
||||
async fn send_notification(
|
||||
tx: &mpsc::Sender<OutgoingResponse>,
|
||||
notify: &NotifyConfig,
|
||||
routine_name: &str,
|
||||
status: RunStatus,
|
||||
summary: Option<&str>,
|
||||
) {
|
||||
let should_notify = match status {
|
||||
RunStatus::Ok => notify.on_success,
|
||||
RunStatus::Attention => notify.on_attention,
|
||||
RunStatus::Failed => notify.on_failure,
|
||||
RunStatus::Running => false,
|
||||
};
|
||||
|
||||
if !should_notify {
|
||||
return;
|
||||
}
|
||||
|
||||
let icon = match status {
|
||||
RunStatus::Ok => "✅",
|
||||
RunStatus::Attention => "🔔",
|
||||
RunStatus::Failed => "❌",
|
||||
RunStatus::Running => "⏳",
|
||||
};
|
||||
|
||||
let message = match summary {
|
||||
Some(s) => format!("{} *Routine '{}'*: {}\n\n{}", icon, routine_name, status, s),
|
||||
None => format!("{} *Routine '{}'*: {}", icon, routine_name, status),
|
||||
};
|
||||
|
||||
let response = OutgoingResponse {
|
||||
content: message,
|
||||
thread_id: None,
|
||||
metadata: serde_json::json!({
|
||||
"source": "routine",
|
||||
"routine_name": routine_name,
|
||||
"status": status.to_string(),
|
||||
}),
|
||||
};
|
||||
|
||||
if let Err(e) = tx.send(response).await {
|
||||
tracing::error!(routine = %routine_name, "Failed to send notification: {}", e);
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the cron ticker background task.
|
||||
pub fn spawn_cron_ticker(
|
||||
engine: Arc<RoutineEngine>,
|
||||
interval: Duration,
|
||||
) -> tokio::task::JoinHandle<()> {
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = tokio::time::interval(interval);
|
||||
// Skip immediate first tick
|
||||
ticker.tick().await;
|
||||
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
engine.check_cron_triggers().await;
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn truncate(s: &str, max: usize) -> String {
|
||||
if s.len() <= max {
|
||||
s.to_string()
|
||||
} else {
|
||||
format!("{}...", &s[..max])
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::agent::routine::{NotifyConfig, RunStatus};
|
||||
|
||||
#[test]
|
||||
fn test_notification_gating() {
|
||||
let config = NotifyConfig {
|
||||
on_success: false,
|
||||
on_failure: true,
|
||||
on_attention: true,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// on_success = false means Ok status should not notify
|
||||
assert!(!config.on_success);
|
||||
assert!(config.on_failure);
|
||||
assert!(config.on_attention);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_run_status_icons() {
|
||||
// Just verify the mapping doesn't panic
|
||||
for status in [
|
||||
RunStatus::Ok,
|
||||
RunStatus::Attention,
|
||||
RunStatus::Failed,
|
||||
RunStatus::Running,
|
||||
] {
|
||||
let _ = status.to_string();
|
||||
}
|
||||
}
|
||||
}
|
||||
+17
-17
@@ -373,23 +373,23 @@ impl Scheduler {
|
||||
.into());
|
||||
}
|
||||
|
||||
// Execute with timeout
|
||||
let result = tokio::time::timeout(Duration::from_secs(60), async {
|
||||
tool.execute(params, &job_ctx).await
|
||||
})
|
||||
.await
|
||||
.map_err(|_| {
|
||||
Error::Tool(crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout: Duration::from_secs(60),
|
||||
})
|
||||
})?
|
||||
.map_err(|e| {
|
||||
Error::Tool(crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: e.to_string(),
|
||||
})
|
||||
})?;
|
||||
// Execute with per-tool timeout
|
||||
let tool_timeout = tool.execution_timeout();
|
||||
let result =
|
||||
tokio::time::timeout(tool_timeout, async { tool.execute(params, &job_ctx).await })
|
||||
.await
|
||||
.map_err(|_| {
|
||||
Error::Tool(crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout: tool_timeout,
|
||||
})
|
||||
})?
|
||||
.map_err(|e| {
|
||||
Error::Tool(crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: e.to_string(),
|
||||
})
|
||||
})?;
|
||||
|
||||
Ok(TaskOutput::new(result.result, start.elapsed()))
|
||||
}
|
||||
|
||||
@@ -173,6 +173,10 @@ pub struct Thread {
|
||||
/// Pending auth token request (thread is in auth mode).
|
||||
#[serde(default)]
|
||||
pub pending_auth: Option<PendingAuth>,
|
||||
/// Last NEAR AI response ID for response chaining. Persisted to DB
|
||||
/// metadata so we can resume chaining across restarts.
|
||||
#[serde(default)]
|
||||
pub last_response_id: Option<String>,
|
||||
}
|
||||
|
||||
impl Thread {
|
||||
@@ -189,6 +193,24 @@ impl Thread {
|
||||
metadata: serde_json::Value::Null,
|
||||
pending_approval: None,
|
||||
pending_auth: None,
|
||||
last_response_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a thread with a specific ID (for DB hydration).
|
||||
pub fn with_id(id: Uuid, session_id: Uuid) -> Self {
|
||||
let now = Utc::now();
|
||||
Self {
|
||||
id,
|
||||
session_id,
|
||||
state: ThreadState::Idle,
|
||||
turns: Vec::new(),
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
metadata: serde_json::Value::Null,
|
||||
pending_approval: None,
|
||||
pending_auth: None,
|
||||
last_response_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -593,4 +615,386 @@ mod tests {
|
||||
let restored: Thread = serde_json::from_str(&json).expect("should deserialize");
|
||||
assert!(restored.pending_auth.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_with_id() {
|
||||
let specific_id = Uuid::new_v4();
|
||||
let session_id = Uuid::new_v4();
|
||||
let thread = Thread::with_id(specific_id, session_id);
|
||||
|
||||
assert_eq!(thread.id, specific_id);
|
||||
assert_eq!(thread.session_id, session_id);
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
assert!(thread.turns.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_with_id_restore_messages() {
|
||||
let thread_id = Uuid::new_v4();
|
||||
let session_id = Uuid::new_v4();
|
||||
let mut thread = Thread::with_id(thread_id, session_id);
|
||||
|
||||
let messages = vec![
|
||||
ChatMessage::user("Hello from DB"),
|
||||
ChatMessage::assistant("Restored response"),
|
||||
];
|
||||
thread.restore_from_messages(messages);
|
||||
|
||||
assert_eq!(thread.id, thread_id);
|
||||
assert_eq!(thread.turns.len(), 1);
|
||||
assert_eq!(thread.turns[0].user_input, "Hello from DB");
|
||||
assert_eq!(
|
||||
thread.turns[0].response,
|
||||
Some("Restored response".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_empty() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Add a turn first, then restore with empty vec
|
||||
thread.start_turn("hello");
|
||||
thread.complete_turn("hi");
|
||||
assert_eq!(thread.turns.len(), 1);
|
||||
|
||||
thread.restore_from_messages(Vec::new());
|
||||
|
||||
// Should clear all turns and stay idle
|
||||
assert!(thread.turns.is_empty());
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_only_assistant_messages() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Only assistant messages (no user messages to anchor turns)
|
||||
let messages = vec![
|
||||
ChatMessage::assistant("I'm here"),
|
||||
ChatMessage::assistant("Still here"),
|
||||
];
|
||||
|
||||
thread.restore_from_messages(messages);
|
||||
|
||||
// Assistant-only messages have no user turn to attach to, so
|
||||
// they should be skipped entirely.
|
||||
assert!(thread.turns.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_multiple_user_messages_in_a_row() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Two user messages with no assistant response between them
|
||||
let messages = vec![
|
||||
ChatMessage::user("first"),
|
||||
ChatMessage::user("second"),
|
||||
ChatMessage::assistant("reply to second"),
|
||||
];
|
||||
|
||||
thread.restore_from_messages(messages);
|
||||
|
||||
// First user message becomes a turn with no response,
|
||||
// second user message pairs with the assistant response.
|
||||
assert_eq!(thread.turns.len(), 2);
|
||||
assert_eq!(thread.turns[0].user_input, "first");
|
||||
assert!(thread.turns[0].response.is_none());
|
||||
assert_eq!(thread.turns[1].user_input, "second");
|
||||
assert_eq!(
|
||||
thread.turns[1].response,
|
||||
Some("reply to second".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_switch() {
|
||||
let mut session = Session::new("user-1");
|
||||
|
||||
let t1_id = session.create_thread().id;
|
||||
let t2_id = session.create_thread().id;
|
||||
|
||||
// After creating two threads, active should be the last one
|
||||
assert_eq!(session.active_thread, Some(t2_id));
|
||||
|
||||
// Switch back to the first
|
||||
assert!(session.switch_thread(t1_id));
|
||||
assert_eq!(session.active_thread, Some(t1_id));
|
||||
|
||||
// Switching to a nonexistent thread should fail
|
||||
let fake_id = Uuid::new_v4();
|
||||
assert!(!session.switch_thread(fake_id));
|
||||
// Active thread should remain unchanged
|
||||
assert_eq!(session.active_thread, Some(t1_id));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_get_or_create_thread_idempotent() {
|
||||
let mut session = Session::new("user-1");
|
||||
|
||||
let tid1 = session.get_or_create_thread().id;
|
||||
let tid2 = session.get_or_create_thread().id;
|
||||
|
||||
// Should return the same thread (not create a new one each time)
|
||||
assert_eq!(tid1, tid2);
|
||||
assert_eq!(session.threads.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_turns() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
for i in 0..5 {
|
||||
thread.start_turn(format!("msg-{}", i));
|
||||
thread.complete_turn(format!("resp-{}", i));
|
||||
}
|
||||
assert_eq!(thread.turns.len(), 5);
|
||||
|
||||
thread.truncate_turns(3);
|
||||
assert_eq!(thread.turns.len(), 3);
|
||||
|
||||
// Should keep the most recent turns
|
||||
assert_eq!(thread.turns[0].user_input, "msg-2");
|
||||
assert_eq!(thread.turns[1].user_input, "msg-3");
|
||||
assert_eq!(thread.turns[2].user_input, "msg-4");
|
||||
|
||||
// Turn numbers should be re-indexed
|
||||
assert_eq!(thread.turns[0].turn_number, 0);
|
||||
assert_eq!(thread.turns[1].turn_number, 1);
|
||||
assert_eq!(thread.turns[2].turn_number, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_turns_noop_when_fewer() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("only one");
|
||||
thread.complete_turn("response");
|
||||
|
||||
thread.truncate_turns(10);
|
||||
assert_eq!(thread.turns.len(), 1);
|
||||
assert_eq!(thread.turns[0].user_input, "only one");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_interrupt_and_resume() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("do something");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
|
||||
thread.interrupt();
|
||||
assert_eq!(thread.state, ThreadState::Interrupted);
|
||||
|
||||
let last_turn = thread.last_turn().unwrap();
|
||||
assert_eq!(last_turn.state, TurnState::Interrupted);
|
||||
assert!(last_turn.completed_at.is_some());
|
||||
|
||||
thread.resume();
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resume_only_from_interrupted() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Idle thread: resume should be a no-op
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
thread.resume();
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
|
||||
// Processing thread: resume should not change state
|
||||
thread.start_turn("work");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
thread.resume();
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_turn_fail() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("risky operation");
|
||||
thread.fail_turn("connection timed out");
|
||||
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
|
||||
let turn = thread.last_turn().unwrap();
|
||||
assert_eq!(turn.state, TurnState::Failed);
|
||||
assert_eq!(turn.error, Some("connection timed out".to_string()));
|
||||
assert!(turn.response.is_none());
|
||||
assert!(turn.completed_at.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_messages_with_incomplete_last_turn() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("first");
|
||||
thread.complete_turn("first reply");
|
||||
thread.start_turn("second (in progress)");
|
||||
|
||||
let messages = thread.messages();
|
||||
// Should have 3 messages: user, assistant, user (no assistant for in-progress)
|
||||
assert_eq!(messages.len(), 3);
|
||||
assert_eq!(messages[0].content, "first");
|
||||
assert_eq!(messages[1].content, "first reply");
|
||||
assert_eq!(messages[2].content, "second (in progress)");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_serialization_round_trip() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("hello");
|
||||
thread.complete_turn("world");
|
||||
thread.last_response_id = Some("resp_abc123".to_string());
|
||||
|
||||
let json = serde_json::to_string(&thread).unwrap();
|
||||
let restored: Thread = serde_json::from_str(&json).unwrap();
|
||||
|
||||
assert_eq!(restored.id, thread.id);
|
||||
assert_eq!(restored.session_id, thread.session_id);
|
||||
assert_eq!(restored.turns.len(), 1);
|
||||
assert_eq!(restored.turns[0].user_input, "hello");
|
||||
assert_eq!(restored.turns[0].response, Some("world".to_string()));
|
||||
assert_eq!(restored.last_response_id, Some("resp_abc123".to_string()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_session_serialization_round_trip() {
|
||||
let mut session = Session::new("user-ser");
|
||||
session.create_thread();
|
||||
session.auto_approve_tool("echo");
|
||||
|
||||
let json = serde_json::to_string(&session).unwrap();
|
||||
let restored: Session = serde_json::from_str(&json).unwrap();
|
||||
|
||||
assert_eq!(restored.user_id, "user-ser");
|
||||
assert_eq!(restored.threads.len(), 1);
|
||||
assert!(restored.is_tool_auto_approved("echo"));
|
||||
assert!(!restored.is_tool_auto_approved("shell"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_auto_approved_tools() {
|
||||
let mut session = Session::new("user-1");
|
||||
|
||||
assert!(!session.is_tool_auto_approved("shell"));
|
||||
session.auto_approve_tool("shell");
|
||||
assert!(session.is_tool_auto_approved("shell"));
|
||||
|
||||
// Idempotent
|
||||
session.auto_approve_tool("shell");
|
||||
assert_eq!(session.auto_approved_tools.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_turn_tool_call_error() {
|
||||
let mut turn = Turn::new(0, "test");
|
||||
turn.record_tool_call("http", serde_json::json!({"url": "example.com"}));
|
||||
turn.record_tool_error("timeout");
|
||||
|
||||
assert_eq!(turn.tool_calls.len(), 1);
|
||||
assert_eq!(turn.tool_calls[0].error, Some("timeout".to_string()));
|
||||
assert!(turn.tool_calls[0].result.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_turn_number_increments() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Before any turns, turn_number() is 1 (1-indexed for display)
|
||||
assert_eq!(thread.turn_number(), 1);
|
||||
|
||||
thread.start_turn("first");
|
||||
thread.complete_turn("done");
|
||||
assert_eq!(thread.turn_number(), 2);
|
||||
|
||||
thread.start_turn("second");
|
||||
assert_eq!(thread.turn_number(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_complete_turn_on_empty_thread() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Completing a turn when there are no turns should be a safe no-op
|
||||
thread.complete_turn("phantom response");
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
assert!(thread.turns.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fail_turn_on_empty_thread() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Failing a turn when there are no turns should be a safe no-op
|
||||
thread.fail_turn("phantom error");
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
assert!(thread.turns.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_pending_approval_flow() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
let approval = PendingApproval {
|
||||
request_id: Uuid::new_v4(),
|
||||
tool_name: "shell".to_string(),
|
||||
parameters: serde_json::json!({"command": "rm -rf /"}),
|
||||
description: "dangerous command".to_string(),
|
||||
tool_call_id: "call_123".to_string(),
|
||||
context_messages: vec![ChatMessage::user("do it")],
|
||||
};
|
||||
|
||||
thread.await_approval(approval);
|
||||
assert_eq!(thread.state, ThreadState::AwaitingApproval);
|
||||
assert!(thread.pending_approval.is_some());
|
||||
|
||||
let taken = thread.take_pending_approval();
|
||||
assert!(taken.is_some());
|
||||
assert_eq!(taken.unwrap().tool_name, "shell");
|
||||
assert!(thread.pending_approval.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_clear_pending_approval() {
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
let approval = PendingApproval {
|
||||
request_id: Uuid::new_v4(),
|
||||
tool_name: "http".to_string(),
|
||||
parameters: serde_json::json!({}),
|
||||
description: "test".to_string(),
|
||||
tool_call_id: "call_456".to_string(),
|
||||
context_messages: vec![],
|
||||
};
|
||||
|
||||
thread.await_approval(approval);
|
||||
thread.clear_pending_approval();
|
||||
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
assert!(thread.pending_approval.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_active_thread_accessors() {
|
||||
let mut session = Session::new("user-1");
|
||||
|
||||
assert!(session.active_thread().is_none());
|
||||
assert!(session.active_thread_mut().is_none());
|
||||
|
||||
let tid = session.create_thread().id;
|
||||
|
||||
assert!(session.active_thread().is_some());
|
||||
assert_eq!(session.active_thread().unwrap().id, tid);
|
||||
|
||||
// Mutably modify through accessor
|
||||
session.active_thread_mut().unwrap().start_turn("test");
|
||||
assert_eq!(
|
||||
session.active_thread().unwrap().state,
|
||||
ThreadState::Processing
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -110,6 +110,41 @@ impl SessionManager {
|
||||
(session, thread_id)
|
||||
}
|
||||
|
||||
/// Register a hydrated thread so subsequent `resolve_thread` calls find it.
|
||||
///
|
||||
/// Inserts into the thread_map and creates an undo manager for the thread.
|
||||
pub async fn register_thread(
|
||||
&self,
|
||||
user_id: &str,
|
||||
channel: &str,
|
||||
thread_id: Uuid,
|
||||
session: Arc<Mutex<Session>>,
|
||||
) {
|
||||
let key = ThreadKey {
|
||||
user_id: user_id.to_string(),
|
||||
channel: channel.to_string(),
|
||||
external_thread_id: Some(thread_id.to_string()),
|
||||
};
|
||||
|
||||
{
|
||||
let mut thread_map = self.thread_map.write().await;
|
||||
thread_map.insert(key, thread_id);
|
||||
}
|
||||
|
||||
{
|
||||
let mut undo_managers = self.undo_managers.write().await;
|
||||
undo_managers
|
||||
.entry(thread_id)
|
||||
.or_insert_with(|| Arc::new(Mutex::new(UndoManager::new())));
|
||||
}
|
||||
|
||||
// Ensure the session is tracked
|
||||
{
|
||||
let mut sessions = self.sessions.write().await;
|
||||
sessions.entry(user_id.to_string()).or_insert(session);
|
||||
}
|
||||
}
|
||||
|
||||
/// Get undo manager for a thread.
|
||||
pub async fn get_undo_manager(&self, thread_id: Uuid) -> Arc<Mutex<UndoManager>> {
|
||||
// Fast path
|
||||
@@ -296,4 +331,344 @@ mod tests {
|
||||
.await;
|
||||
assert_eq!(pruned, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_register_thread() {
|
||||
use crate::agent::session::{Session, Thread};
|
||||
|
||||
let manager = SessionManager::new();
|
||||
let thread_id = Uuid::new_v4();
|
||||
|
||||
// Create a session with a hydrated thread
|
||||
let session = Arc::new(Mutex::new(Session::new("user-hydrate")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(thread_id, sess.id);
|
||||
sess.threads.insert(thread_id, thread);
|
||||
sess.active_thread = Some(thread_id);
|
||||
}
|
||||
|
||||
// Register the thread
|
||||
manager
|
||||
.register_thread("user-hydrate", "gateway", thread_id, Arc::clone(&session))
|
||||
.await;
|
||||
|
||||
// resolve_thread should find it (using the UUID as external_thread_id)
|
||||
let (resolved_session, resolved_tid) = manager
|
||||
.resolve_thread("user-hydrate", "gateway", Some(&thread_id.to_string()))
|
||||
.await;
|
||||
assert_eq!(resolved_tid, thread_id);
|
||||
|
||||
// Should be the same session object
|
||||
let sess = resolved_session.lock().await;
|
||||
assert!(sess.threads.contains_key(&thread_id));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_thread_with_explicit_external_id() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
// Two calls with the same explicit external thread ID should resolve
|
||||
// to the same internal thread.
|
||||
let (_, t1) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("ext-abc"))
|
||||
.await;
|
||||
let (_, t2) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("ext-abc"))
|
||||
.await;
|
||||
assert_eq!(t1, t2);
|
||||
|
||||
// A different external ID on the same channel/user gets a new thread.
|
||||
let (_, t3) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("ext-xyz"))
|
||||
.await;
|
||||
assert_ne!(t1, t3);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_thread_none_vs_some_external_id() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
// None external_thread_id is a distinct key from Some("ext-1").
|
||||
let (_, t_none) = manager.resolve_thread("user-1", "cli", None).await;
|
||||
let (_, t_some) = manager.resolve_thread("user-1", "cli", Some("ext-1")).await;
|
||||
assert_ne!(t_none, t_some);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_thread_different_users_isolated() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
let (_, t1) = manager
|
||||
.resolve_thread("user-a", "gateway", Some("same-ext"))
|
||||
.await;
|
||||
let (_, t2) = manager
|
||||
.resolve_thread("user-b", "gateway", Some("same-ext"))
|
||||
.await;
|
||||
|
||||
// Same channel + same external ID but different users = different threads
|
||||
assert_ne!(t1, t2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_thread_different_channels_isolated() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
let (_, t1) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("thread-x"))
|
||||
.await;
|
||||
let (_, t2) = manager
|
||||
.resolve_thread("user-1", "telegram", Some("thread-x"))
|
||||
.await;
|
||||
|
||||
// Same user + same external ID but different channels = different threads
|
||||
assert_ne!(t1, t2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_thread_stale_mapping_creates_new_thread() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
// Create a thread normally
|
||||
let (session, original_tid) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("ext-1"))
|
||||
.await;
|
||||
|
||||
// Simulate the thread being removed from the session (e.g. pruned)
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
sess.threads.remove(&original_tid);
|
||||
}
|
||||
|
||||
// Next resolve should detect the stale mapping and create a fresh thread
|
||||
let (_, new_tid) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("ext-1"))
|
||||
.await;
|
||||
assert_ne!(original_tid, new_tid);
|
||||
|
||||
// The new thread should actually exist in the session
|
||||
let sess = session.lock().await;
|
||||
assert!(sess.threads.contains_key(&new_tid));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_register_thread_preserves_uuid_on_resolve() {
|
||||
use crate::agent::session::{Session, Thread};
|
||||
|
||||
let manager = SessionManager::new();
|
||||
let known_uuid = Uuid::new_v4();
|
||||
|
||||
let session = Arc::new(Mutex::new(Session::new("user-web")));
|
||||
let session_id = {
|
||||
let sess = session.lock().await;
|
||||
sess.id
|
||||
};
|
||||
|
||||
// Simulate hydration: create thread with a known UUID
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(known_uuid, session_id);
|
||||
sess.threads.insert(known_uuid, thread);
|
||||
}
|
||||
|
||||
// Register it
|
||||
manager
|
||||
.register_thread("user-web", "gateway", known_uuid, Arc::clone(&session))
|
||||
.await;
|
||||
|
||||
// resolve_thread with UUID as external_thread_id MUST return the same UUID,
|
||||
// not mint a new one (this was the root cause of the "wrong conversation" bug)
|
||||
let (_, resolved) = manager
|
||||
.resolve_thread("user-web", "gateway", Some(&known_uuid.to_string()))
|
||||
.await;
|
||||
assert_eq!(resolved, known_uuid);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_register_thread_idempotent() {
|
||||
use crate::agent::session::{Session, Thread};
|
||||
|
||||
let manager = SessionManager::new();
|
||||
let tid = Uuid::new_v4();
|
||||
|
||||
let session = Arc::new(Mutex::new(Session::new("user-idem")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
// Register twice
|
||||
manager
|
||||
.register_thread("user-idem", "gateway", tid, Arc::clone(&session))
|
||||
.await;
|
||||
manager
|
||||
.register_thread("user-idem", "gateway", tid, Arc::clone(&session))
|
||||
.await;
|
||||
|
||||
// Should still resolve to the same thread
|
||||
let (_, resolved) = manager
|
||||
.resolve_thread("user-idem", "gateway", Some(&tid.to_string()))
|
||||
.await;
|
||||
assert_eq!(resolved, tid);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_register_thread_creates_undo_manager() {
|
||||
use crate::agent::session::{Session, Thread};
|
||||
|
||||
let manager = SessionManager::new();
|
||||
let tid = Uuid::new_v4();
|
||||
|
||||
let session = Arc::new(Mutex::new(Session::new("user-undo")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
manager
|
||||
.register_thread("user-undo", "gateway", tid, Arc::clone(&session))
|
||||
.await;
|
||||
|
||||
// Undo manager should exist for the registered thread
|
||||
let undo = manager.get_undo_manager(tid).await;
|
||||
let undo2 = manager.get_undo_manager(tid).await;
|
||||
assert!(Arc::ptr_eq(&undo, &undo2));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_register_thread_stores_session() {
|
||||
use crate::agent::session::{Session, Thread};
|
||||
|
||||
let manager = SessionManager::new();
|
||||
let tid = Uuid::new_v4();
|
||||
|
||||
let session = Arc::new(Mutex::new(Session::new("user-new")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
// The user has no session yet in the manager
|
||||
{
|
||||
let sessions = manager.sessions.read().await;
|
||||
assert!(!sessions.contains_key("user-new"));
|
||||
}
|
||||
|
||||
manager
|
||||
.register_thread("user-new", "gateway", tid, Arc::clone(&session))
|
||||
.await;
|
||||
|
||||
// Now the session should be tracked
|
||||
{
|
||||
let sessions = manager.sessions.read().await;
|
||||
assert!(sessions.contains_key("user-new"));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multiple_threads_per_user() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
let (_, t1) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("thread-a"))
|
||||
.await;
|
||||
let (_, t2) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("thread-b"))
|
||||
.await;
|
||||
let (session, t3) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("thread-c"))
|
||||
.await;
|
||||
|
||||
// All three should be distinct
|
||||
assert_ne!(t1, t2);
|
||||
assert_ne!(t2, t3);
|
||||
assert_ne!(t1, t3);
|
||||
|
||||
// All three should exist in the same session
|
||||
let sess = session.lock().await;
|
||||
assert!(sess.threads.contains_key(&t1));
|
||||
assert!(sess.threads.contains_key(&t2));
|
||||
assert!(sess.threads.contains_key(&t3));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_prune_cleans_thread_map_and_undo_managers() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
let (stale_session, stale_tid) = manager.resolve_thread("user-stale", "cli", None).await;
|
||||
|
||||
// Backdate the session
|
||||
{
|
||||
let mut sess = stale_session.lock().await;
|
||||
sess.last_active_at = chrono::Utc::now() - chrono::TimeDelta::seconds(86400 * 30);
|
||||
}
|
||||
|
||||
// Verify thread_map and undo_managers have entries
|
||||
{
|
||||
let tm = manager.thread_map.read().await;
|
||||
assert!(!tm.is_empty());
|
||||
}
|
||||
{
|
||||
let um = manager.undo_managers.read().await;
|
||||
assert!(um.contains_key(&stale_tid));
|
||||
}
|
||||
|
||||
let pruned = manager
|
||||
.prune_stale_sessions(std::time::Duration::from_secs(86400 * 7))
|
||||
.await;
|
||||
assert_eq!(pruned, 1);
|
||||
|
||||
// Thread map and undo managers should be cleaned up
|
||||
{
|
||||
let tm = manager.thread_map.read().await;
|
||||
assert!(tm.is_empty());
|
||||
}
|
||||
{
|
||||
let um = manager.undo_managers.read().await;
|
||||
assert!(!um.contains_key(&stale_tid));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_thread_active_thread_set() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
let (session, thread_id) = manager
|
||||
.resolve_thread("user-1", "gateway", Some("ext-1"))
|
||||
.await;
|
||||
|
||||
// The resolved thread should be set as the active thread
|
||||
let sess = session.lock().await;
|
||||
assert_eq!(sess.active_thread, Some(thread_id));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_register_then_resolve_different_channel_creates_new() {
|
||||
use crate::agent::session::{Session, Thread};
|
||||
|
||||
let manager = SessionManager::new();
|
||||
let tid = Uuid::new_v4();
|
||||
|
||||
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
// Register on "gateway" channel
|
||||
manager
|
||||
.register_thread("user-cross", "gateway", tid, Arc::clone(&session))
|
||||
.await;
|
||||
|
||||
// Resolve on a different channel with the same UUID string should NOT
|
||||
// find the registered thread (channel is part of the key)
|
||||
let (_, resolved) = manager
|
||||
.resolve_thread("user-cross", "telegram", Some(&tid.to_string()))
|
||||
.await;
|
||||
assert_ne!(resolved, tid);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,6 +43,49 @@ impl SubmissionParser {
|
||||
if lower == "/thread new" || lower == "/new" {
|
||||
return Submission::NewThread;
|
||||
}
|
||||
// System commands (bypass thread-state checks)
|
||||
if lower == "/help" || lower == "/?" {
|
||||
return Submission::SystemCommand {
|
||||
command: "help".to_string(),
|
||||
args: vec![],
|
||||
};
|
||||
}
|
||||
if lower == "/version" {
|
||||
return Submission::SystemCommand {
|
||||
command: "version".to_string(),
|
||||
args: vec![],
|
||||
};
|
||||
}
|
||||
if lower == "/tools" {
|
||||
return Submission::SystemCommand {
|
||||
command: "tools".to_string(),
|
||||
args: vec![],
|
||||
};
|
||||
}
|
||||
if lower == "/ping" {
|
||||
return Submission::SystemCommand {
|
||||
command: "ping".to_string(),
|
||||
args: vec![],
|
||||
};
|
||||
}
|
||||
if lower == "/debug" {
|
||||
return Submission::SystemCommand {
|
||||
command: "debug".to_string(),
|
||||
args: vec![],
|
||||
};
|
||||
}
|
||||
if lower.starts_with("/model") {
|
||||
let args: Vec<String> = trimmed
|
||||
.split_whitespace()
|
||||
.skip(1)
|
||||
.map(|s| s.to_string())
|
||||
.collect();
|
||||
return Submission::SystemCommand {
|
||||
command: "model".to_string(),
|
||||
args,
|
||||
};
|
||||
}
|
||||
|
||||
if lower == "/quit" || lower == "/exit" || lower == "/shutdown" {
|
||||
return Submission::Quit;
|
||||
}
|
||||
@@ -172,6 +215,15 @@ pub enum Submission {
|
||||
|
||||
/// Quit the agent. Bypasses thread-state checks.
|
||||
Quit,
|
||||
|
||||
/// System command (help, model, version, tools, ping, debug).
|
||||
/// Bypasses thread-state checks and safety validation.
|
||||
SystemCommand {
|
||||
/// The command name (e.g. "help", "model", "version").
|
||||
command: String,
|
||||
/// Arguments to the command.
|
||||
args: Vec<String>,
|
||||
},
|
||||
}
|
||||
|
||||
impl Submission {
|
||||
@@ -238,6 +290,7 @@ impl Submission {
|
||||
| Self::Heartbeat
|
||||
| Self::Summarize
|
||||
| Self::Suggest
|
||||
| Self::SystemCommand { .. }
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -504,6 +557,84 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_help() {
|
||||
let submission = SubmissionParser::parse("/help");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "help" && args.is_empty())
|
||||
);
|
||||
|
||||
let submission = SubmissionParser::parse("/?");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, .. } if command == "help")
|
||||
);
|
||||
|
||||
let submission = SubmissionParser::parse("/HELP");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, .. } if command == "help")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_model() {
|
||||
// No args: show current model
|
||||
let submission = SubmissionParser::parse("/model");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "model" && args.is_empty())
|
||||
);
|
||||
|
||||
// With args: switch model
|
||||
let submission = SubmissionParser::parse("/model gpt-4o");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "model" && args == vec!["gpt-4o"])
|
||||
);
|
||||
|
||||
// Case insensitive command, preserves arg case
|
||||
let submission = SubmissionParser::parse("/MODEL Claude-3.5");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "model" && args == vec!["Claude-3.5"])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_version() {
|
||||
let submission = SubmissionParser::parse("/version");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "version" && args.is_empty())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_tools() {
|
||||
let submission = SubmissionParser::parse("/tools");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "tools" && args.is_empty())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_ping() {
|
||||
let submission = SubmissionParser::parse("/ping");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "ping" && args.is_empty())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_debug() {
|
||||
let submission = SubmissionParser::parse("/debug");
|
||||
assert!(
|
||||
matches!(submission, Submission::SystemCommand { command, args } if command == "debug" && args.is_empty())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_system_command_is_control() {
|
||||
let submission = SubmissionParser::parse("/help");
|
||||
assert!(submission.is_control());
|
||||
assert!(!submission.starts_turn());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parser_quit() {
|
||||
assert!(matches!(SubmissionParser::parse("/quit"), Submission::Quit));
|
||||
|
||||
+52
-4
@@ -272,7 +272,10 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
));
|
||||
}
|
||||
}
|
||||
RespondResult::ToolCalls(tool_calls) => {
|
||||
RespondResult::ToolCalls {
|
||||
tool_calls,
|
||||
content,
|
||||
} => {
|
||||
// Model returned tool calls - execute them
|
||||
tracing::debug!(
|
||||
"Job {} respond_with_tools returned {} tool calls",
|
||||
@@ -280,6 +283,14 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
tool_calls.len()
|
||||
);
|
||||
|
||||
// Add assistant message with tool_calls (OpenAI protocol)
|
||||
reason_ctx
|
||||
.messages
|
||||
.push(ChatMessage::assistant_with_tool_calls(
|
||||
content,
|
||||
tool_calls.clone(),
|
||||
));
|
||||
|
||||
for tc in tool_calls {
|
||||
let result = self.execute_tool(&tc.name, &tc.arguments).await;
|
||||
|
||||
@@ -417,14 +428,51 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
.into());
|
||||
}
|
||||
|
||||
// Execute with timeout and timing
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
params = %params,
|
||||
job = %job_id,
|
||||
"Tool call started"
|
||||
);
|
||||
|
||||
// Execute with per-tool timeout and timing
|
||||
let tool_timeout = tool.execution_timeout();
|
||||
let start = std::time::Instant::now();
|
||||
let result = tokio::time::timeout(Duration::from_secs(60), async {
|
||||
let result = tokio::time::timeout(tool_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 = tool_timeout.as_secs(),
|
||||
"Tool call timed out"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Record action in memory and get the ActionRecord for persistence
|
||||
let action = match &result {
|
||||
Ok(Ok(output)) => {
|
||||
@@ -479,7 +527,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
let output = result
|
||||
.map_err(|_| crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout: Duration::from_secs(60),
|
||||
timeout: tool_timeout,
|
||||
})?
|
||||
.map_err(|e| crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
|
||||
Reference in New Issue
Block a user