mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-30 16:19:21 +00:00
* feat: unified thread model for web gateway Every piece of activity (user chat, routine run, heartbeat alert, external channel message) now lives in its own thread, properly isolated, with meaningful titles and visual distinction. Key changes: - Add `channel` field to ConversationSummary and ThreadInfo so the gateway can distinguish thread origins (gateway, telegram, routine, heartbeat). - Add `list_conversations_all_channels` to Database trait (both postgres and libsql) so chat_threads_handler shows cross-channel threads. - Routine runs get a persistent conversation per routine via `get_or_create_routine_conversation`; notifications carry thread_id. - Heartbeat gets a persistent conversation via `get_or_create_heartbeat_conversation`; HeartbeatRunner accepts an optional Database store and binds notifications to the thread. - Fix broadcast() in web gateway to propagate response.thread_id instead of hardcoding empty string. - Fix isCurrentThread(null) returning true (the core notification leak bug) — now returns false so events without a thread_id don't leak into the active thread. - Rewrite frontend thread sidebar: meaningful titles with channel-specific fallbacks, relative timestamps instead of turn counts, channel badges for non-gateway threads, unread notification dots, read-only indicator for external channel threads. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review — TOCTOU races, stale comment, debounce, broadcast warning - Fix TOCTOU race in get_or_create_routine_conversation (postgres): use INSERT ON CONFLICT on new uq_conv_routine unique index + SELECT-back. - Fix TOCTOU race in get_or_create_heartbeat_conversation (postgres): use INSERT ON CONFLICT on new uq_conv_heartbeat unique index + SELECT-back. - Fix TOCTOU race in get_or_create_routine_conversation (libsql): use BEGIN IMMEDIATE transaction to serialize concurrent writers. - Fix TOCTOU race in get_or_create_heartbeat_conversation (libsql): use BEGIN IMMEDIATE transaction to serialize concurrent writers. - Add V11 migration with partial unique indexes for postgres. - Add matching unique indexes to libsql schema. - Update stale comment on isCurrentThread (said "always shown" but logic now returns false for missing thread_id). - Debounce loadThreads() on off-thread SSE events to prevent request storms. - Log warning in broadcast() when thread_id is None (clients will drop it). Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: sort in-memory thread fallback by updated_at descending The in-memory thread list fallback (when no DB is available) used HashMap::values() which has no guaranteed ordering. Sort by updated_at descending to match the SQL query ordering. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: retry libsql connect() on transient "unable to open database file" The cron ticker's background task occasionally fails with "unable to open database file" when creating a new SQLite connection concurrently with the main thread. Add retry with exponential backoff (50ms, 100ms, 200ms) to handle transient VFS/locking issues in libsql's local mode. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: use ON CONFLICT with index expressions instead of named constraints PostgreSQL ON CONFLICT ON CONSTRAINT requires a named table constraint, but V11 migration creates unique indexes. Switch to the expression form (ON CONFLICT (columns) WHERE condition) which works with unique indexes. Also fix dead code in threadTitle() where thread.title was already checked on the previous line. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * style: fix rustfmt chain collapse in heartbeat.rs [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: skip broadcast when thread_id is None instead of sending empty Clients drop SSE events with empty thread_id anyway, so avoid the unnecessary network traffic by returning early. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add libsql routine/heartbeat conversation idempotency tests Add tests proving get_or_create_routine_conversation returns the same conversation ID across multiple invocations with the same routine_id. Add debug logging to routine engine to track conversation resolution. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: show "New chat" title for empty threads - threadTitle() returns "New chat" when turn_count is 0 - Assistant thread label updates dynamically from API data - Default HTML label changed from "Assistant" to "New chat" - New threads naturally sort to top via last_activity DESC [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: thread sorting, routine isolation, and UI polish - Fix libsql timestamp format mismatch causing broken thread sort order. SQLite defaults used `datetime('now')` (space-separated) while Rust code used RFC3339 (T-separated), breaking string-based ORDER BY. All INSERTs now use RFC3339, and queries use `datetime()` to normalize comparison. - Route manual routine triggers through RoutineEngine.fire_manual() instead of injecting as regular chat messages, so routines always run in their dedicated conversation thread. - Add RoutineEngineSlot to GatewayState for gateway<->engine communication. - Derive routine thread titles from conversation metadata (routine_name) instead of showing truncated UUID hashes. - Make chat_new_thread_handler persist to DB synchronously so loadThreads() sees newly created threads immediately. - Fix enableChatInput() no-op and wrong element ID in disableChatInputReadOnly(). - Fix handlers/chat.rs stale gateway-only query (use list_conversations_all_channels). - Sort in-memory threads by DateTime before converting to RFC3339 strings. - Trigger debouncedLoadThreads() on thinking/status SSE events for non-current threads so routine/heartbeat threads appear in sidebar promptly. - Remove "Threads" text from sidebar header. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: routine history display, orphaned tool_results, duplicate system messages Three independent fixes with regression tests: 1. Routine conversations now display in the web UI. build_turns_from_db_messages() handles standalone assistant messages (no preceding user message) by creating turns with empty user_input. Frontend skips empty user bubbles. 2. Worker select_tools and execute_plan paths now push an assistant_with_tool_calls message before tool execution, preventing sanitize_tool_messages from rewriting tool_results as orphaned user messages. 3. Reasoning::plan() and respond_with_tools() merge system messages from context into a single system prompt instead of creating [system, system, ...] sequences that strict LLM providers (Qwen) reject. Also: sidebar padding/spacing improvements, wider thread panel (240px). Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR #607 review — RwLock held across await, missing ownership check, heartbeat config - Clone Arc<RoutineEngine> out of RwLock before .await in trigger handler - Add user_id ownership check to fire_manual() with NotAuthorized error - Wire heartbeat notify_user/notify_channel from config to AgentHeartbeatConfig Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: gitignore trace_*.json files and remove stale traces Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: remove trace JSON files from repo Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: proper HTTP status codes for routine errors, read-only input guard, respond thread_id - Map RoutineError::NotFound → 404, NotAuthorized → 403, Disabled → 409 - Guard enableChatInput() against re-enabling on read-only threads - Skip respond() when thread_id is None (matches broadcast() behavior) [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Claude Opus 4.6 <[email protected]>
746 lines
23 KiB
Rust
746 lines
23 KiB
Rust
//! LLM-facing tools for managing routines.
|
|
//!
|
|
//! Six tools let the agent manage routines conversationally:
|
|
//! - `routine_create` - Create a new routine
|
|
//! - `routine_list` - List all routines with status
|
|
//! - `routine_update` - Modify or toggle a routine
|
|
//! - `routine_delete` - Remove a routine
|
|
//! - `routine_fire` - Manually trigger a routine
|
|
//! - `routine_history` - View past runs
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use async_trait::async_trait;
|
|
use chrono::Utc;
|
|
use uuid::Uuid;
|
|
|
|
use crate::agent::routine::{
|
|
NotifyConfig, Routine, RoutineAction, RoutineGuardrails, Trigger, next_cron_fire,
|
|
};
|
|
use crate::agent::routine_engine::RoutineEngine;
|
|
use crate::context::JobContext;
|
|
use crate::db::Database;
|
|
use crate::tools::tool::{ApprovalRequirement, Tool, ToolError, ToolOutput, require_str};
|
|
|
|
// ==================== routine_create ====================
|
|
|
|
pub struct RoutineCreateTool {
|
|
store: Arc<dyn Database>,
|
|
engine: Arc<RoutineEngine>,
|
|
}
|
|
|
|
impl RoutineCreateTool {
|
|
pub fn new(store: Arc<dyn Database>, engine: Arc<RoutineEngine>) -> Self {
|
|
Self { store, engine }
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Tool for RoutineCreateTool {
|
|
fn name(&self) -> &str {
|
|
"routine_create"
|
|
}
|
|
|
|
fn description(&self) -> &str {
|
|
"Create a new routine (scheduled or event-driven task). \
|
|
Supports cron schedules, event pattern matching, webhooks, and manual triggers. \
|
|
Use this when the user wants something to happen periodically or reactively."
|
|
}
|
|
|
|
fn parameters_schema(&self) -> serde_json::Value {
|
|
serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {
|
|
"type": "string",
|
|
"description": "Unique name for the routine (e.g. 'daily-pr-review')"
|
|
},
|
|
"description": {
|
|
"type": "string",
|
|
"description": "What this routine does"
|
|
},
|
|
"trigger_type": {
|
|
"type": "string",
|
|
"enum": ["cron", "event", "webhook", "manual"],
|
|
"description": "When the routine fires"
|
|
},
|
|
"schedule": {
|
|
"type": "string",
|
|
"description": "Cron expression (for cron trigger). E.g. '0 9 * * MON-FRI' for weekdays at 9am. Uses 6-field cron (sec min hour day month weekday)."
|
|
},
|
|
"event_pattern": {
|
|
"type": "string",
|
|
"description": "Regex pattern to match messages (for event trigger)"
|
|
},
|
|
"event_channel": {
|
|
"type": "string",
|
|
"description": "Optional channel filter for event trigger (e.g. 'telegram')"
|
|
},
|
|
"prompt": {
|
|
"type": "string",
|
|
"description": "The prompt/instructions for the routine"
|
|
},
|
|
"context_paths": {
|
|
"type": "array",
|
|
"items": { "type": "string" },
|
|
"description": "Workspace paths to load as context (e.g. ['context/priorities.md'])"
|
|
},
|
|
"action_type": {
|
|
"type": "string",
|
|
"enum": ["lightweight", "full_job"],
|
|
"description": "Execution mode: 'lightweight' (single LLM call, default) or 'full_job' (multi-turn with tools)"
|
|
},
|
|
"cooldown_secs": {
|
|
"type": "integer",
|
|
"description": "Minimum seconds between fires (default: 300)"
|
|
},
|
|
"tool_permissions": {
|
|
"type": "array",
|
|
"items": { "type": "string" },
|
|
"description": "Tool names pre-authorized for Always-approval tools in full_job mode (e.g. ['shell']). UnlessAutoApproved tools are automatically permitted in routines."
|
|
},
|
|
"notify_channel": {
|
|
"type": "string",
|
|
"description": "Channel to send results to (e.g. 'telegram', 'slack', 'tui'). Sets the default channel for message tool calls in routine jobs."
|
|
},
|
|
"notify_user": {
|
|
"type": "string",
|
|
"description": "User/target to notify (e.g. username, chat ID). Defaults to 'default'."
|
|
}
|
|
},
|
|
"required": ["name", "trigger_type", "prompt"]
|
|
})
|
|
}
|
|
|
|
async fn execute(
|
|
&self,
|
|
params: serde_json::Value,
|
|
ctx: &JobContext,
|
|
) -> Result<ToolOutput, ToolError> {
|
|
let start = std::time::Instant::now();
|
|
|
|
let name = require_str(¶ms, "name")?;
|
|
|
|
let description = params
|
|
.get("description")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("");
|
|
|
|
let trigger_type = require_str(¶ms, "trigger_type")?;
|
|
|
|
let prompt = require_str(¶ms, "prompt")?;
|
|
|
|
// Build trigger
|
|
let trigger = match trigger_type {
|
|
"cron" => {
|
|
let schedule =
|
|
params
|
|
.get("schedule")
|
|
.and_then(|v| v.as_str())
|
|
.ok_or_else(|| {
|
|
ToolError::InvalidParameters(
|
|
"cron trigger requires 'schedule'".to_string(),
|
|
)
|
|
})?;
|
|
// Validate cron expression
|
|
next_cron_fire(schedule).map_err(|e| {
|
|
ToolError::InvalidParameters(format!("invalid cron schedule: {e}"))
|
|
})?;
|
|
Trigger::Cron {
|
|
schedule: schedule.to_string(),
|
|
}
|
|
}
|
|
"event" => {
|
|
let pattern = params
|
|
.get("event_pattern")
|
|
.and_then(|v| v.as_str())
|
|
.ok_or_else(|| {
|
|
ToolError::InvalidParameters(
|
|
"event trigger requires 'event_pattern'".to_string(),
|
|
)
|
|
})?;
|
|
// Validate regex
|
|
regex::Regex::new(pattern)
|
|
.map_err(|e| ToolError::InvalidParameters(format!("invalid regex: {e}")))?;
|
|
let channel = params
|
|
.get("event_channel")
|
|
.and_then(|v| v.as_str())
|
|
.map(String::from);
|
|
Trigger::Event {
|
|
channel,
|
|
pattern: pattern.to_string(),
|
|
}
|
|
}
|
|
"webhook" => Trigger::Webhook {
|
|
path: None,
|
|
secret: None,
|
|
},
|
|
"manual" => Trigger::Manual,
|
|
other => {
|
|
return Err(ToolError::InvalidParameters(format!(
|
|
"unknown trigger_type: {other}"
|
|
)));
|
|
}
|
|
};
|
|
|
|
// Build action
|
|
let action_type = params
|
|
.get("action_type")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("lightweight");
|
|
|
|
let context_paths: Vec<String> = params
|
|
.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 action = match action_type {
|
|
"lightweight" => RoutineAction::Lightweight {
|
|
prompt: prompt.to_string(),
|
|
context_paths,
|
|
max_tokens: 4096,
|
|
},
|
|
"full_job" => {
|
|
let tool_permissions = crate::agent::routine::parse_tool_permissions(¶ms);
|
|
RoutineAction::FullJob {
|
|
title: name.to_string(),
|
|
description: prompt.to_string(),
|
|
max_iterations: 10,
|
|
tool_permissions,
|
|
}
|
|
}
|
|
other => {
|
|
return Err(ToolError::InvalidParameters(format!(
|
|
"unknown action_type: {other}"
|
|
)));
|
|
}
|
|
};
|
|
|
|
let cooldown_secs = params
|
|
.get("cooldown_secs")
|
|
.and_then(|v| v.as_u64())
|
|
.unwrap_or(300);
|
|
|
|
// Compute next fire time for cron
|
|
let next_fire = if let Trigger::Cron { ref schedule } = trigger {
|
|
next_cron_fire(schedule).unwrap_or(None)
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let routine = Routine {
|
|
id: Uuid::new_v4(),
|
|
name: name.to_string(),
|
|
description: description.to_string(),
|
|
user_id: ctx.user_id.clone(),
|
|
enabled: true,
|
|
trigger,
|
|
action,
|
|
guardrails: RoutineGuardrails {
|
|
cooldown: Duration::from_secs(cooldown_secs),
|
|
max_concurrent: 1,
|
|
dedup_window: None,
|
|
},
|
|
notify: NotifyConfig {
|
|
channel: params
|
|
.get("notify_channel")
|
|
.and_then(|v| v.as_str())
|
|
.map(String::from),
|
|
user: params
|
|
.get("notify_user")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("default")
|
|
.to_string(),
|
|
..NotifyConfig::default()
|
|
},
|
|
last_run_at: None,
|
|
next_fire_at: next_fire,
|
|
run_count: 0,
|
|
consecutive_failures: 0,
|
|
state: serde_json::json!({}),
|
|
created_at: Utc::now(),
|
|
updated_at: Utc::now(),
|
|
};
|
|
|
|
self.store
|
|
.create_routine(&routine)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("failed to create routine: {e}")))?;
|
|
|
|
// Refresh event cache if this is an event trigger
|
|
if routine.trigger.type_tag() == "event" {
|
|
self.engine.refresh_event_cache().await;
|
|
}
|
|
|
|
let result = serde_json::json!({
|
|
"id": routine.id.to_string(),
|
|
"name": routine.name,
|
|
"trigger_type": routine.trigger.type_tag(),
|
|
"next_fire_at": routine.next_fire_at.map(|t| t.to_rfc3339()),
|
|
"status": "created",
|
|
});
|
|
|
|
Ok(ToolOutput::success(result, start.elapsed()))
|
|
}
|
|
|
|
fn requires_sanitization(&self) -> bool {
|
|
false
|
|
}
|
|
}
|
|
|
|
// ==================== routine_list ====================
|
|
|
|
pub struct RoutineListTool {
|
|
store: Arc<dyn Database>,
|
|
}
|
|
|
|
impl RoutineListTool {
|
|
pub fn new(store: Arc<dyn Database>) -> Self {
|
|
Self { store }
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Tool for RoutineListTool {
|
|
fn name(&self) -> &str {
|
|
"routine_list"
|
|
}
|
|
|
|
fn description(&self) -> &str {
|
|
"List all routines with their status, trigger info, and next fire time."
|
|
}
|
|
|
|
fn parameters_schema(&self) -> serde_json::Value {
|
|
serde_json::json!({
|
|
"type": "object",
|
|
"properties": {},
|
|
"required": []
|
|
})
|
|
}
|
|
|
|
async fn execute(
|
|
&self,
|
|
_params: serde_json::Value,
|
|
ctx: &JobContext,
|
|
) -> Result<ToolOutput, ToolError> {
|
|
let start = std::time::Instant::now();
|
|
|
|
let routines = self
|
|
.store
|
|
.list_routines(&ctx.user_id)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("failed to list routines: {e}")))?;
|
|
|
|
let list: Vec<serde_json::Value> = routines
|
|
.iter()
|
|
.map(|r| {
|
|
serde_json::json!({
|
|
"id": r.id.to_string(),
|
|
"name": r.name,
|
|
"description": r.description,
|
|
"enabled": r.enabled,
|
|
"trigger_type": r.trigger.type_tag(),
|
|
"action_type": r.action.type_tag(),
|
|
"last_run_at": r.last_run_at.map(|t| t.to_rfc3339()),
|
|
"next_fire_at": r.next_fire_at.map(|t| t.to_rfc3339()),
|
|
"run_count": r.run_count,
|
|
"consecutive_failures": r.consecutive_failures,
|
|
})
|
|
})
|
|
.collect();
|
|
|
|
let result = serde_json::json!({
|
|
"count": list.len(),
|
|
"routines": list,
|
|
});
|
|
|
|
Ok(ToolOutput::success(result, start.elapsed()))
|
|
}
|
|
|
|
fn requires_sanitization(&self) -> bool {
|
|
false
|
|
}
|
|
}
|
|
|
|
// ==================== routine_update ====================
|
|
|
|
pub struct RoutineUpdateTool {
|
|
store: Arc<dyn Database>,
|
|
engine: Arc<RoutineEngine>,
|
|
}
|
|
|
|
impl RoutineUpdateTool {
|
|
pub fn new(store: Arc<dyn Database>, engine: Arc<RoutineEngine>) -> Self {
|
|
Self { store, engine }
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Tool for RoutineUpdateTool {
|
|
fn name(&self) -> &str {
|
|
"routine_update"
|
|
}
|
|
|
|
fn description(&self) -> &str {
|
|
"Update an existing routine. Can modify trigger, prompt, schedule, or toggle enabled state. \
|
|
Pass the routine name and only the fields you want to change."
|
|
}
|
|
|
|
fn parameters_schema(&self) -> serde_json::Value {
|
|
serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {
|
|
"type": "string",
|
|
"description": "Name of the routine to update"
|
|
},
|
|
"enabled": {
|
|
"type": "boolean",
|
|
"description": "Enable or disable the routine"
|
|
},
|
|
"prompt": {
|
|
"type": "string",
|
|
"description": "New prompt/instructions"
|
|
},
|
|
"schedule": {
|
|
"type": "string",
|
|
"description": "New cron schedule (for cron triggers)"
|
|
},
|
|
"description": {
|
|
"type": "string",
|
|
"description": "New description"
|
|
}
|
|
},
|
|
"required": ["name"]
|
|
})
|
|
}
|
|
|
|
async fn execute(
|
|
&self,
|
|
params: serde_json::Value,
|
|
ctx: &JobContext,
|
|
) -> Result<ToolOutput, ToolError> {
|
|
let start = std::time::Instant::now();
|
|
|
|
let name = require_str(¶ms, "name")?;
|
|
|
|
let mut routine = self
|
|
.store
|
|
.get_routine_by_name(&ctx.user_id, name)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("DB error: {e}")))?
|
|
.ok_or_else(|| ToolError::ExecutionFailed(format!("routine '{}' not found", name)))?;
|
|
|
|
// Apply updates
|
|
if let Some(enabled) = params.get("enabled").and_then(|v| v.as_bool()) {
|
|
routine.enabled = enabled;
|
|
}
|
|
|
|
if let Some(desc) = params.get("description").and_then(|v| v.as_str()) {
|
|
routine.description = desc.to_string();
|
|
}
|
|
|
|
if let Some(prompt) = params.get("prompt").and_then(|v| v.as_str()) {
|
|
match &mut routine.action {
|
|
RoutineAction::Lightweight { prompt: p, .. } => *p = prompt.to_string(),
|
|
RoutineAction::FullJob { description: d, .. } => *d = prompt.to_string(),
|
|
}
|
|
}
|
|
|
|
if let Some(schedule) = params.get("schedule").and_then(|v| v.as_str()) {
|
|
// Validate
|
|
next_cron_fire(schedule)
|
|
.map_err(|e| ToolError::InvalidParameters(format!("invalid cron schedule: {e}")))?;
|
|
|
|
routine.trigger = Trigger::Cron {
|
|
schedule: schedule.to_string(),
|
|
};
|
|
routine.next_fire_at = next_cron_fire(schedule).unwrap_or(None);
|
|
}
|
|
|
|
self.store
|
|
.update_routine(&routine)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("failed to update: {e}")))?;
|
|
|
|
// Refresh event cache in case trigger changed
|
|
self.engine.refresh_event_cache().await;
|
|
|
|
let result = serde_json::json!({
|
|
"name": routine.name,
|
|
"enabled": routine.enabled,
|
|
"trigger_type": routine.trigger.type_tag(),
|
|
"next_fire_at": routine.next_fire_at.map(|t| t.to_rfc3339()),
|
|
"status": "updated",
|
|
});
|
|
|
|
Ok(ToolOutput::success(result, start.elapsed()))
|
|
}
|
|
|
|
fn requires_sanitization(&self) -> bool {
|
|
false
|
|
}
|
|
}
|
|
|
|
// ==================== routine_delete ====================
|
|
|
|
pub struct RoutineDeleteTool {
|
|
store: Arc<dyn Database>,
|
|
engine: Arc<RoutineEngine>,
|
|
}
|
|
|
|
impl RoutineDeleteTool {
|
|
pub fn new(store: Arc<dyn Database>, engine: Arc<RoutineEngine>) -> Self {
|
|
Self { store, engine }
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Tool for RoutineDeleteTool {
|
|
fn name(&self) -> &str {
|
|
"routine_delete"
|
|
}
|
|
|
|
fn description(&self) -> &str {
|
|
"Delete a routine permanently. This also removes all run history."
|
|
}
|
|
|
|
fn parameters_schema(&self) -> serde_json::Value {
|
|
serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {
|
|
"type": "string",
|
|
"description": "Name of the routine to delete"
|
|
}
|
|
},
|
|
"required": ["name"]
|
|
})
|
|
}
|
|
|
|
async fn execute(
|
|
&self,
|
|
params: serde_json::Value,
|
|
ctx: &JobContext,
|
|
) -> Result<ToolOutput, ToolError> {
|
|
let start = std::time::Instant::now();
|
|
|
|
let name = require_str(¶ms, "name")?;
|
|
|
|
let routine = self
|
|
.store
|
|
.get_routine_by_name(&ctx.user_id, name)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("DB error: {e}")))?
|
|
.ok_or_else(|| ToolError::ExecutionFailed(format!("routine '{}' not found", name)))?;
|
|
|
|
let deleted = self
|
|
.store
|
|
.delete_routine(routine.id)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("failed to delete: {e}")))?;
|
|
|
|
// Refresh event cache
|
|
self.engine.refresh_event_cache().await;
|
|
|
|
let result = serde_json::json!({
|
|
"name": name,
|
|
"deleted": deleted,
|
|
});
|
|
|
|
Ok(ToolOutput::success(result, start.elapsed()))
|
|
}
|
|
|
|
fn requires_sanitization(&self) -> bool {
|
|
false
|
|
}
|
|
}
|
|
|
|
// ==================== routine_fire ====================
|
|
|
|
pub struct RoutineFireTool {
|
|
store: Arc<dyn Database>,
|
|
engine: Arc<RoutineEngine>,
|
|
}
|
|
|
|
impl RoutineFireTool {
|
|
pub fn new(store: Arc<dyn Database>, engine: Arc<RoutineEngine>) -> Self {
|
|
Self { store, engine }
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Tool for RoutineFireTool {
|
|
fn name(&self) -> &str {
|
|
"routine_fire"
|
|
}
|
|
|
|
fn description(&self) -> &str {
|
|
"Manually trigger a routine to run immediately, bypassing schedule, trigger type, and cooldown."
|
|
}
|
|
|
|
fn requires_approval(&self, _params: &serde_json::Value) -> ApprovalRequirement {
|
|
// Firing a routine can dispatch a full_job with pre-authorized Always-gated tools,
|
|
// so this is a meaningful escalation that warrants auto-approval gating.
|
|
ApprovalRequirement::UnlessAutoApproved
|
|
}
|
|
|
|
fn parameters_schema(&self) -> serde_json::Value {
|
|
serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {
|
|
"type": "string",
|
|
"description": "Name of the routine to fire"
|
|
}
|
|
},
|
|
"required": ["name"]
|
|
})
|
|
}
|
|
|
|
async fn execute(
|
|
&self,
|
|
params: serde_json::Value,
|
|
ctx: &JobContext,
|
|
) -> Result<ToolOutput, ToolError> {
|
|
let start = std::time::Instant::now();
|
|
|
|
let name = require_str(¶ms, "name")?;
|
|
|
|
let routine = self
|
|
.store
|
|
.get_routine_by_name(&ctx.user_id, name)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("DB error: {e}")))?
|
|
.ok_or_else(|| ToolError::ExecutionFailed(format!("routine '{}' not found", name)))?;
|
|
|
|
let run_id = self
|
|
.engine
|
|
.fire_manual(routine.id, None)
|
|
.await
|
|
.map_err(|e| {
|
|
ToolError::ExecutionFailed(format!("failed to fire routine '{}': {e}", name))
|
|
})?;
|
|
|
|
let result = serde_json::json!({
|
|
"name": name,
|
|
"run_id": run_id.to_string(),
|
|
"status": "fired",
|
|
});
|
|
|
|
Ok(ToolOutput::success(result, start.elapsed()))
|
|
}
|
|
|
|
fn requires_sanitization(&self) -> bool {
|
|
false
|
|
}
|
|
}
|
|
|
|
// ==================== routine_history ====================
|
|
|
|
pub struct RoutineHistoryTool {
|
|
store: Arc<dyn Database>,
|
|
}
|
|
|
|
impl RoutineHistoryTool {
|
|
pub fn new(store: Arc<dyn Database>) -> Self {
|
|
Self { store }
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Tool for RoutineHistoryTool {
|
|
fn name(&self) -> &str {
|
|
"routine_history"
|
|
}
|
|
|
|
fn description(&self) -> &str {
|
|
"View the execution history of a routine. Shows recent runs with status, duration, and results."
|
|
}
|
|
|
|
fn parameters_schema(&self) -> serde_json::Value {
|
|
serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {
|
|
"type": "string",
|
|
"description": "Name of the routine"
|
|
},
|
|
"limit": {
|
|
"type": "integer",
|
|
"description": "Max runs to return (default: 10)",
|
|
"default": 10
|
|
}
|
|
},
|
|
"required": ["name"]
|
|
})
|
|
}
|
|
|
|
async fn execute(
|
|
&self,
|
|
params: serde_json::Value,
|
|
ctx: &JobContext,
|
|
) -> Result<ToolOutput, ToolError> {
|
|
let start = std::time::Instant::now();
|
|
|
|
let name = require_str(¶ms, "name")?;
|
|
|
|
let limit = params
|
|
.get("limit")
|
|
.and_then(|v| v.as_i64())
|
|
.unwrap_or(10)
|
|
.min(50);
|
|
|
|
let routine = self
|
|
.store
|
|
.get_routine_by_name(&ctx.user_id, name)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("DB error: {e}")))?
|
|
.ok_or_else(|| ToolError::ExecutionFailed(format!("routine '{}' not found", name)))?;
|
|
|
|
let runs = self
|
|
.store
|
|
.list_routine_runs(routine.id, limit)
|
|
.await
|
|
.map_err(|e| ToolError::ExecutionFailed(format!("failed to list runs: {e}")))?;
|
|
|
|
let run_list: Vec<serde_json::Value> = runs
|
|
.iter()
|
|
.map(|r| {
|
|
let duration_secs = r
|
|
.completed_at
|
|
.map(|c| c.signed_duration_since(r.started_at).num_seconds());
|
|
serde_json::json!({
|
|
"id": r.id.to_string(),
|
|
"trigger_type": r.trigger_type,
|
|
"trigger_detail": r.trigger_detail,
|
|
"started_at": r.started_at.to_rfc3339(),
|
|
"completed_at": r.completed_at.map(|t| t.to_rfc3339()),
|
|
"duration_secs": duration_secs,
|
|
"status": r.status.to_string(),
|
|
"result_summary": r.result_summary,
|
|
"tokens_used": r.tokens_used,
|
|
})
|
|
})
|
|
.collect();
|
|
|
|
let result = serde_json::json!({
|
|
"routine": name,
|
|
"total_runs": routine.run_count,
|
|
"runs": run_list,
|
|
});
|
|
|
|
Ok(ToolOutput::success(result, start.elapsed()))
|
|
}
|
|
|
|
fn requires_sanitization(&self) -> bool {
|
|
false
|
|
}
|
|
}
|