| 1 | //! Model-callable one-shot delayed continuation tool (`send_later`). |
| 2 | //! |
| 3 | //! Lets the agent schedule a future message into the current workspace. When |
| 4 | //! the trigger fires the scheduler enqueues a normal durable task with the |
| 5 | //! specified message, just like an automation run. |
| 6 | //! |
| 7 | //! # Actions |
| 8 | //! |
| 9 | //! | action | description | |
| 10 | //! |------------|-----------------------------------------------------------| |
| 11 | //! | `schedule` | Create a pending trigger; returns `trigger_id`+`fire_at` | |
| 12 | //! | `list` | List recent triggers (default: 50 newest) | |
| 13 | //! | `read` | Read a single trigger by `trigger_id` | |
| 14 | //! | `cancel` | Cancel a pending trigger before it fires | |
| 15 | |
| 16 | use std::path::PathBuf; |
| 17 | |
| 18 | use async_trait::async_trait; |
| 19 | use chrono::{DateTime, Duration, Utc}; |
| 20 | use serde_json::{Value, json}; |
| 21 | |
| 22 | use crate::automation_manager::{ |
| 23 | CreateDelayedTriggerRequest, DelayedTriggerStatus, SharedAutomationManager, |
| 24 | }; |
| 25 | use crate::tools::spec::{ |
| 26 | ApprovalRequirement, ToolCapability, ToolContext, ToolError, ToolResult, ToolSpec, |
| 27 | optional_str, optional_u64, |
| 28 | }; |
| 29 | |
| 30 | const ALL_ACTIONS: &[&str] = &["schedule", "list", "read", "cancel"]; |
| 31 | const READ_ACTIONS: &[&str] = &["list", "read"]; |
| 32 | |
| 33 | /// One-shot delayed-continuation tool. |
| 34 | pub struct SendLaterTool { |
| 35 | name: &'static str, |
| 36 | read_only: bool, |
| 37 | } |
| 38 | |
| 39 | impl SendLaterTool { |
| 40 | pub const fn new(name: &'static str) -> Self { |
| 41 | Self { |
| 42 | name, |
| 43 | read_only: false, |
| 44 | } |
| 45 | } |
| 46 | |
| 47 | /// Plan-mode variant: only the read-only actions are advertised. |
| 48 | pub const fn read_only(name: &'static str) -> Self { |
| 49 | Self { |
| 50 | name, |
| 51 | read_only: true, |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | fn allowed_actions(&self) -> &'static [&'static str] { |
| 56 | if self.read_only { |
| 57 | READ_ACTIONS |
| 58 | } else { |
| 59 | ALL_ACTIONS |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | fn resolve_action<'a>(&self, input: &'a Value) -> Result<&'a str, ToolError> { |
| 64 | let action = input.get("action").and_then(Value::as_str).ok_or_else(|| { |
| 65 | ToolError::invalid_input(format!( |
| 66 | "send_later: missing `action` (one of: {})", |
| 67 | self.allowed_actions().join(", ") |
| 68 | )) |
| 69 | })?; |
| 70 | if self.allowed_actions().contains(&action) { |
| 71 | Ok(action) |
| 72 | } else { |
| 73 | Err(ToolError::invalid_input(format!( |
| 74 | "send_later: invalid action `{action}` (one of: {})", |
| 75 | self.allowed_actions().join(", ") |
| 76 | ))) |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | fn automations_from_context( |
| 81 | context: &ToolContext, |
| 82 | ) -> Result<SharedAutomationManager, ToolError> { |
| 83 | context |
| 84 | .runtime |
| 85 | .automations |
| 86 | .as_ref() |
| 87 | .cloned() |
| 88 | .ok_or_else(|| ToolError::not_available("send_later: automation manager not available")) |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | #[async_trait] |
| 93 | impl ToolSpec for SendLaterTool { |
| 94 | fn name(&self) -> &'static str { |
| 95 | self.name |
| 96 | } |
| 97 | |
| 98 | fn description(&self) -> &'static str { |
| 99 | if self.read_only { |
| 100 | "Inspect pending one-shot delayed continuations. Actions: \"list\" (newest triggers), \"read\" (one trigger by trigger_id)." |
| 101 | } else { |
| 102 | "Schedule a one-shot delayed message into the current workspace. \ |
| 103 | Actions: \"schedule\" (create a pending trigger; requires approval), \ |
| 104 | \"list\" (recent triggers), \"read\" (one trigger by trigger_id), \ |
| 105 | \"cancel\" (cancel a pending trigger before it fires; requires approval). \ |
| 106 | Use delay_minutes or fire_at (ISO 8601 UTC) — not both. \ |
| 107 | Returns trigger_id and resolved fire_at." |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | fn input_schema(&self) -> Value { |
| 112 | let actions: Vec<&str> = self.allowed_actions().to_vec(); |
| 113 | let mut properties = serde_json::Map::new(); |
| 114 | properties.insert( |
| 115 | "action".to_string(), |
| 116 | json!({ |
| 117 | "type": "string", |
| 118 | "enum": actions, |
| 119 | "description": "Action to perform." |
| 120 | }), |
| 121 | ); |
| 122 | if !self.read_only { |
| 123 | properties.insert( |
| 124 | "delay_minutes".to_string(), |
| 125 | json!({ |
| 126 | "type": "integer", |
| 127 | "minimum": 1, |
| 128 | "description": "Minutes from now to fire. Mutually exclusive with fire_at. (action=schedule)" |
| 129 | }), |
| 130 | ); |
| 131 | properties.insert( |
| 132 | "fire_at".to_string(), |
| 133 | json!({ |
| 134 | "type": "string", |
| 135 | "description": "Absolute UTC fire time as an ISO 8601 string, e.g. \"2026-07-08T00:43:00Z\". Mutually exclusive with delay_minutes. (action=schedule)" |
| 136 | }), |
| 137 | ); |
| 138 | properties.insert( |
| 139 | "message".to_string(), |
| 140 | json!({ |
| 141 | "type": "string", |
| 142 | "description": "The message to inject as a new task when the trigger fires. (action=schedule)" |
| 143 | }), |
| 144 | ); |
| 145 | properties.insert( |
| 146 | "workspace".to_string(), |
| 147 | json!({ |
| 148 | "type": "string", |
| 149 | "description": "Optional working directory for the fired task; defaults to the current workspace. (action=schedule)" |
| 150 | }), |
| 151 | ); |
| 152 | properties.insert( |
| 153 | "parent_trigger_id".to_string(), |
| 154 | json!({ |
| 155 | "type": "string", |
| 156 | "description": "Optional id of the trigger that re-armed this one, for lineage tracking. (action=schedule)" |
| 157 | }), |
| 158 | ); |
| 159 | } |
| 160 | properties.insert( |
| 161 | "trigger_id".to_string(), |
| 162 | json!({ |
| 163 | "type": "string", |
| 164 | "description": "Target trigger id. (action=read/cancel)" |
| 165 | }), |
| 166 | ); |
| 167 | properties.insert( |
| 168 | "limit".to_string(), |
| 169 | json!({ |
| 170 | "type": "integer", |
| 171 | "minimum": 1, |
| 172 | "maximum": 200, |
| 173 | "default": 50, |
| 174 | "description": "Maximum number of results to return. (action=list)" |
| 175 | }), |
| 176 | ); |
| 177 | properties.insert( |
| 178 | "status".to_string(), |
| 179 | json!({ |
| 180 | "type": "string", |
| 181 | "enum": ["pending", "fired", "canceled", "failed"], |
| 182 | "description": "Filter by trigger status. (action=list)" |
| 183 | }), |
| 184 | ); |
| 185 | json!({ |
| 186 | "type": "object", |
| 187 | "properties": properties, |
| 188 | "required": ["action"] |
| 189 | }) |
| 190 | } |
| 191 | |
| 192 | fn capabilities(&self) -> Vec<ToolCapability> { |
| 193 | if self.read_only { |
| 194 | vec![ToolCapability::ReadOnly] |
| 195 | } else { |
| 196 | vec![ |
| 197 | ToolCapability::ExecutesCode, |
| 198 | ToolCapability::RequiresApproval, |
| 199 | ] |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | fn approval_requirement(&self) -> ApprovalRequirement { |
| 204 | if self.read_only { |
| 205 | ApprovalRequirement::Auto |
| 206 | } else { |
| 207 | ApprovalRequirement::Required |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | fn approval_requirement_for(&self, input: &Value) -> ApprovalRequirement { |
| 212 | match input.get("action").and_then(Value::as_str) { |
| 213 | Some("list") | Some("read") => ApprovalRequirement::Auto, |
| 214 | _ if self.read_only => ApprovalRequirement::Auto, |
| 215 | _ => ApprovalRequirement::Required, |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | fn is_read_only_for(&self, input: &Value) -> bool { |
| 220 | match input.get("action").and_then(Value::as_str) { |
| 221 | Some("list") | Some("read") => true, |
| 222 | _ => self.read_only, |
| 223 | } |
| 224 | } |
| 225 | |
| 226 | async fn execute(&self, input: Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 227 | let action = self.resolve_action(&input)?; |
| 228 | |
| 229 | match action { |
| 230 | "schedule" => execute_schedule(&input, context).await, |
| 231 | "list" => execute_list(&input, context).await, |
| 232 | "read" => execute_read(&input, context).await, |
| 233 | "cancel" => execute_cancel(&input, context).await, |
| 234 | _ => Err(ToolError::invalid_input(format!( |
| 235 | "send_later: unhandled action `{action}`" |
| 236 | ))), |
| 237 | } |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | // ── Action handlers ──────────────────────────────────────────────────────── |
| 242 | |
| 243 | async fn execute_schedule(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 244 | let automations = SendLaterTool::automations_from_context(context)?; |
| 245 | |
| 246 | let delay_minutes = input.get("delay_minutes").and_then(Value::as_u64); |
| 247 | let fire_at_str = optional_str(input, "fire_at")?; |
| 248 | let message = input |
| 249 | .get("message") |
| 250 | .and_then(Value::as_str) |
| 251 | .ok_or_else(|| ToolError::invalid_input("send_later schedule: `message` is required"))?; |
| 252 | |
| 253 | if message.trim().is_empty() { |
| 254 | return Err(ToolError::invalid_input( |
| 255 | "send_later schedule: `message` must not be empty", |
| 256 | )); |
| 257 | } |
| 258 | |
| 259 | // Validate mutual exclusivity and resolve fire_at. |
| 260 | let fire_at: DateTime<Utc> = match (delay_minutes, fire_at_str) { |
| 261 | (Some(_), Some(_)) => { |
| 262 | return Err(ToolError::invalid_input( |
| 263 | "send_later schedule: `delay_minutes` and `fire_at` are mutually exclusive", |
| 264 | )); |
| 265 | } |
| 266 | (None, None) => { |
| 267 | return Err(ToolError::invalid_input( |
| 268 | "send_later schedule: one of `delay_minutes` or `fire_at` is required", |
| 269 | )); |
| 270 | } |
| 271 | (Some(minutes), None) => { |
| 272 | if minutes == 0 { |
| 273 | return Err(ToolError::invalid_input( |
| 274 | "send_later schedule: `delay_minutes` must be >= 1", |
| 275 | )); |
| 276 | } |
| 277 | let minutes_i64 = i64::try_from(minutes).map_err(|_| { |
| 278 | ToolError::invalid_input("send_later schedule: `delay_minutes` is too large") |
| 279 | })?; |
| 280 | Utc::now() + Duration::minutes(minutes_i64) |
| 281 | } |
| 282 | (None, Some(fire_at_s)) => fire_at_s.parse::<DateTime<Utc>>().map_err(|err| { |
| 283 | ToolError::invalid_input(format!( |
| 284 | "send_later schedule: invalid `fire_at` — expected ISO 8601 UTC, e.g. \ |
| 285 | \"2026-07-08T00:43:00Z\": {err}" |
| 286 | )) |
| 287 | })?, |
| 288 | }; |
| 289 | |
| 290 | let workspace: Option<PathBuf> = optional_str(input, "workspace")? |
| 291 | .map(PathBuf::from) |
| 292 | .or_else(|| { |
| 293 | let ws = &context.workspace; |
| 294 | if ws == std::path::Path::new(".") { |
| 295 | None |
| 296 | } else { |
| 297 | Some(ws.clone()) |
| 298 | } |
| 299 | }); |
| 300 | |
| 301 | let parent_trigger_id = optional_str(input, "parent_trigger_id")?.map(String::from); |
| 302 | |
| 303 | let req = CreateDelayedTriggerRequest { |
| 304 | fire_at, |
| 305 | message: message.to_string(), |
| 306 | workspace, |
| 307 | parent_trigger_id, |
| 308 | }; |
| 309 | |
| 310 | let record = { |
| 311 | let manager = automations.lock().await; |
| 312 | manager.create_trigger(req).map_err(|err| { |
| 313 | ToolError::execution_failed(format!("send_later schedule failed: {err}")) |
| 314 | })? |
| 315 | }; |
| 316 | |
| 317 | Ok(ToolResult::success( |
| 318 | serde_json::to_string_pretty(&json!({ |
| 319 | "trigger_id": record.trigger_id, |
| 320 | "fire_at": record.fire_at.to_rfc3339(), |
| 321 | "status": "pending", |
| 322 | })) |
| 323 | .unwrap_or_default(), |
| 324 | )) |
| 325 | } |
| 326 | |
| 327 | async fn execute_list(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 328 | let automations = SendLaterTool::automations_from_context(context)?; |
| 329 | |
| 330 | let limit = Some(optional_u64(input, "limit", 50)? as usize); |
| 331 | let status_filter = optional_str(input, "status")? |
| 332 | .map(parse_trigger_status) |
| 333 | .transpose() |
| 334 | .map_err(ToolError::invalid_input)?; |
| 335 | |
| 336 | let records = { |
| 337 | let manager = automations.lock().await; |
| 338 | manager |
| 339 | .list_triggers(status_filter, limit) |
| 340 | .map_err(|err| ToolError::execution_failed(format!("send_later list failed: {err}")))? |
| 341 | }; |
| 342 | |
| 343 | let items: Vec<Value> = records |
| 344 | .iter() |
| 345 | .map(|r| { |
| 346 | json!({ |
| 347 | "trigger_id": r.trigger_id, |
| 348 | "fire_at": r.fire_at.to_rfc3339(), |
| 349 | "status": trigger_status_str(r.status), |
| 350 | "created_at": r.created_at.to_rfc3339(), |
| 351 | "message_preview": r.message.chars().take(120).collect::<String>(), |
| 352 | "workspace": r.workspace, |
| 353 | "parent_trigger_id": r.parent_trigger_id, |
| 354 | "task_id": r.task_id, |
| 355 | "error": r.error, |
| 356 | }) |
| 357 | }) |
| 358 | .collect(); |
| 359 | |
| 360 | let count = items.len(); |
| 361 | Ok(ToolResult::success( |
| 362 | serde_json::to_string_pretty(&json!({ "triggers": items, "count": count })) |
| 363 | .unwrap_or_default(), |
| 364 | )) |
| 365 | } |
| 366 | |
| 367 | async fn execute_read(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 368 | let automations = SendLaterTool::automations_from_context(context)?; |
| 369 | |
| 370 | let trigger_id = input |
| 371 | .get("trigger_id") |
| 372 | .and_then(Value::as_str) |
| 373 | .ok_or_else(|| ToolError::invalid_input("send_later read: `trigger_id` is required"))?; |
| 374 | |
| 375 | let record = { |
| 376 | let manager = automations.lock().await; |
| 377 | manager |
| 378 | .get_trigger(trigger_id) |
| 379 | .map_err(|err| ToolError::execution_failed(format!("send_later read failed: {err}")))? |
| 380 | }; |
| 381 | |
| 382 | Ok(ToolResult::success( |
| 383 | serde_json::to_string_pretty(&json!({ |
| 384 | "trigger_id": record.trigger_id, |
| 385 | "fire_at": record.fire_at.to_rfc3339(), |
| 386 | "status": trigger_status_str(record.status), |
| 387 | "created_at": record.created_at.to_rfc3339(), |
| 388 | "fired_at": record.fired_at.map(|t| t.to_rfc3339()), |
| 389 | "message": record.message, |
| 390 | "workspace": record.workspace, |
| 391 | "parent_trigger_id": record.parent_trigger_id, |
| 392 | "task_id": record.task_id, |
| 393 | "thread_id": record.thread_id, |
| 394 | "error": record.error, |
| 395 | })) |
| 396 | .unwrap_or_default(), |
| 397 | )) |
| 398 | } |
| 399 | |
| 400 | async fn execute_cancel(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 401 | let automations = SendLaterTool::automations_from_context(context)?; |
| 402 | |
| 403 | let trigger_id = input |
| 404 | .get("trigger_id") |
| 405 | .and_then(Value::as_str) |
| 406 | .ok_or_else(|| ToolError::invalid_input("send_later cancel: `trigger_id` is required"))?; |
| 407 | |
| 408 | let record = { |
| 409 | let manager = automations.lock().await; |
| 410 | manager.cancel_trigger(trigger_id).map_err(|err| { |
| 411 | ToolError::execution_failed(format!("send_later cancel failed: {err}")) |
| 412 | })? |
| 413 | }; |
| 414 | |
| 415 | Ok(ToolResult::success( |
| 416 | serde_json::to_string_pretty(&json!({ |
| 417 | "trigger_id": record.trigger_id, |
| 418 | "status": "canceled", |
| 419 | "fire_at": record.fire_at.to_rfc3339(), |
| 420 | })) |
| 421 | .unwrap_or_default(), |
| 422 | )) |
| 423 | } |
| 424 | |
| 425 | fn parse_trigger_status(s: &str) -> Result<DelayedTriggerStatus, String> { |
| 426 | match s { |
| 427 | "pending" => Ok(DelayedTriggerStatus::Pending), |
| 428 | "fired" => Ok(DelayedTriggerStatus::Fired), |
| 429 | "canceled" => Ok(DelayedTriggerStatus::Canceled), |
| 430 | "failed" => Ok(DelayedTriggerStatus::Failed), |
| 431 | other => Err(format!( |
| 432 | "unknown trigger status '{other}'; expected one of: pending, fired, canceled, failed" |
| 433 | )), |
| 434 | } |
| 435 | } |
| 436 | |
| 437 | fn trigger_status_str(status: DelayedTriggerStatus) -> &'static str { |
| 438 | match status { |
| 439 | DelayedTriggerStatus::Pending => "pending", |
| 440 | DelayedTriggerStatus::Fired => "fired", |
| 441 | DelayedTriggerStatus::Canceled => "canceled", |
| 442 | DelayedTriggerStatus::Failed => "failed", |
| 443 | } |
| 444 | } |
| 445 | |
| 446 | // ── Tests ────────────────────────────────────────────────────────────────── |
| 447 | |
| 448 | #[cfg(test)] |
| 449 | mod tests { |
| 450 | use std::sync::Arc; |
| 451 | |
| 452 | use chrono::Duration; |
| 453 | use serde_json::json; |
| 454 | use tempfile::TempDir; |
| 455 | use tokio::sync::Mutex; |
| 456 | |
| 457 | use crate::automation_manager::AutomationManager; |
| 458 | use crate::tools::spec::{RuntimeToolServices, ToolContext, ToolSpec}; |
| 459 | |
| 460 | use super::SendLaterTool; |
| 461 | |
| 462 | fn make_context(tmp: &TempDir) -> ToolContext { |
| 463 | let manager = AutomationManager::open(tmp.path().to_path_buf()).unwrap(); |
| 464 | let shared = Arc::new(Mutex::new(manager)); |
| 465 | ToolContext::new(".").with_runtime_services(RuntimeToolServices { |
| 466 | automations: Some(shared), |
| 467 | ..Default::default() |
| 468 | }) |
| 469 | } |
| 470 | |
| 471 | #[tokio::test] |
| 472 | async fn schedule_with_delay_minutes_succeeds() { |
| 473 | let tmp = TempDir::new().unwrap(); |
| 474 | let ctx = make_context(&tmp); |
| 475 | let tool = SendLaterTool::new("send_later"); |
| 476 | |
| 477 | let result = tool |
| 478 | .execute( |
| 479 | json!({ |
| 480 | "action": "schedule", |
| 481 | "delay_minutes": 60, |
| 482 | "message": "Check CI status on open PRs." |
| 483 | }), |
| 484 | &ctx, |
| 485 | ) |
| 486 | .await |
| 487 | .unwrap(); |
| 488 | |
| 489 | assert!(result.content.contains("trigger_id")); |
| 490 | assert!(result.content.contains("fire_at")); |
| 491 | assert!(result.content.contains("pending")); |
| 492 | } |
| 493 | |
| 494 | #[tokio::test] |
| 495 | async fn schedule_with_absolute_fire_at_succeeds() { |
| 496 | let tmp = TempDir::new().unwrap(); |
| 497 | let ctx = make_context(&tmp); |
| 498 | let tool = SendLaterTool::new("send_later"); |
| 499 | |
| 500 | let fire_at = (chrono::Utc::now() + Duration::minutes(30)).to_rfc3339(); |
| 501 | let result = tool |
| 502 | .execute( |
| 503 | json!({ |
| 504 | "action": "schedule", |
| 505 | "fire_at": fire_at, |
| 506 | "message": "Check PR mergeability." |
| 507 | }), |
| 508 | &ctx, |
| 509 | ) |
| 510 | .await |
| 511 | .unwrap(); |
| 512 | |
| 513 | assert!(result.content.contains("trig_")); |
| 514 | } |
| 515 | |
| 516 | #[tokio::test] |
| 517 | async fn schedule_rejects_mutually_exclusive_inputs() { |
| 518 | let tmp = TempDir::new().unwrap(); |
| 519 | let ctx = make_context(&tmp); |
| 520 | let tool = SendLaterTool::new("send_later"); |
| 521 | |
| 522 | let fire_at = (chrono::Utc::now() + Duration::minutes(30)).to_rfc3339(); |
| 523 | let err = tool |
| 524 | .execute( |
| 525 | json!({ |
| 526 | "action": "schedule", |
| 527 | "delay_minutes": 60, |
| 528 | "fire_at": fire_at, |
| 529 | "message": "Should fail." |
| 530 | }), |
| 531 | &ctx, |
| 532 | ) |
| 533 | .await |
| 534 | .unwrap_err(); |
| 535 | |
| 536 | assert!(err.to_string().contains("mutually exclusive")); |
| 537 | } |
| 538 | |
| 539 | #[tokio::test] |
| 540 | async fn schedule_rejects_missing_timing_input() { |
| 541 | let tmp = TempDir::new().unwrap(); |
| 542 | let ctx = make_context(&tmp); |
| 543 | let tool = SendLaterTool::new("send_later"); |
| 544 | |
| 545 | let err = tool |
| 546 | .execute( |
| 547 | json!({ |
| 548 | "action": "schedule", |
| 549 | "message": "No timing provided." |
| 550 | }), |
| 551 | &ctx, |
| 552 | ) |
| 553 | .await |
| 554 | .unwrap_err(); |
| 555 | |
| 556 | assert!( |
| 557 | err.to_string().contains("required"), |
| 558 | "expected required-timing error, got: {err}" |
| 559 | ); |
| 560 | } |
| 561 | |
| 562 | #[tokio::test] |
| 563 | async fn schedule_rejects_past_fire_at() { |
| 564 | let tmp = TempDir::new().unwrap(); |
| 565 | let ctx = make_context(&tmp); |
| 566 | let tool = SendLaterTool::new("send_later"); |
| 567 | |
| 568 | let fire_at = (chrono::Utc::now() - Duration::minutes(5)).to_rfc3339(); |
| 569 | let err = tool |
| 570 | .execute( |
| 571 | json!({ |
| 572 | "action": "schedule", |
| 573 | "fire_at": fire_at, |
| 574 | "message": "This is in the past." |
| 575 | }), |
| 576 | &ctx, |
| 577 | ) |
| 578 | .await |
| 579 | .unwrap_err(); |
| 580 | |
| 581 | assert!( |
| 582 | err.to_string().contains("future"), |
| 583 | "expected future-time error, got: {err}" |
| 584 | ); |
| 585 | } |
| 586 | |
| 587 | #[tokio::test] |
| 588 | async fn schedule_rejects_malformed_fire_at() { |
| 589 | let tmp = TempDir::new().unwrap(); |
| 590 | let ctx = make_context(&tmp); |
| 591 | let tool = SendLaterTool::new("send_later"); |
| 592 | |
| 593 | let err = tool |
| 594 | .execute( |
| 595 | json!({ |
| 596 | "action": "schedule", |
| 597 | "fire_at": "not-a-timestamp", |
| 598 | "message": "Bad time." |
| 599 | }), |
| 600 | &ctx, |
| 601 | ) |
| 602 | .await |
| 603 | .unwrap_err(); |
| 604 | |
| 605 | assert!( |
| 606 | err.to_string().contains("invalid `fire_at`"), |
| 607 | "expected parse error, got: {err}" |
| 608 | ); |
| 609 | } |
| 610 | |
| 611 | #[tokio::test] |
| 612 | async fn list_and_read_round_trip() { |
| 613 | let tmp = TempDir::new().unwrap(); |
| 614 | let ctx = make_context(&tmp); |
| 615 | let tool = SendLaterTool::new("send_later"); |
| 616 | |
| 617 | // Schedule a trigger. |
| 618 | let sched_result = tool |
| 619 | .execute( |
| 620 | json!({ |
| 621 | "action": "schedule", |
| 622 | "delay_minutes": 15, |
| 623 | "message": "PR watcher check-in." |
| 624 | }), |
| 625 | &ctx, |
| 626 | ) |
| 627 | .await |
| 628 | .unwrap(); |
| 629 | |
| 630 | let sched_val: serde_json::Value = serde_json::from_str(&sched_result.content).unwrap(); |
| 631 | let trigger_id = sched_val["trigger_id"].as_str().unwrap(); |
| 632 | |
| 633 | // List should include it. |
| 634 | let list_result = tool |
| 635 | .execute(json!({ "action": "list" }), &ctx) |
| 636 | .await |
| 637 | .unwrap(); |
| 638 | assert!(list_result.content.contains(trigger_id)); |
| 639 | |
| 640 | // Read should return full detail. |
| 641 | let read_result = tool |
| 642 | .execute(json!({ "action": "read", "trigger_id": trigger_id }), &ctx) |
| 643 | .await |
| 644 | .unwrap(); |
| 645 | assert!(read_result.content.contains("PR watcher check-in.")); |
| 646 | } |
| 647 | |
| 648 | #[tokio::test] |
| 649 | async fn cancel_pending_trigger() { |
| 650 | let tmp = TempDir::new().unwrap(); |
| 651 | let ctx = make_context(&tmp); |
| 652 | let tool = SendLaterTool::new("send_later"); |
| 653 | |
| 654 | let sched_result = tool |
| 655 | .execute( |
| 656 | json!({ |
| 657 | "action": "schedule", |
| 658 | "delay_minutes": 30, |
| 659 | "message": "Will be canceled." |
| 660 | }), |
| 661 | &ctx, |
| 662 | ) |
| 663 | .await |
| 664 | .unwrap(); |
| 665 | |
| 666 | let trigger_id = serde_json::from_str::<serde_json::Value>(&sched_result.content).unwrap() |
| 667 | ["trigger_id"] |
| 668 | .as_str() |
| 669 | .unwrap() |
| 670 | .to_string(); |
| 671 | |
| 672 | let cancel_result = tool |
| 673 | .execute( |
| 674 | json!({ "action": "cancel", "trigger_id": &trigger_id }), |
| 675 | &ctx, |
| 676 | ) |
| 677 | .await |
| 678 | .unwrap(); |
| 679 | |
| 680 | assert!(cancel_result.content.contains("canceled")); |
| 681 | |
| 682 | // Canceling again should fail. |
| 683 | let err = tool |
| 684 | .execute( |
| 685 | json!({ "action": "cancel", "trigger_id": &trigger_id }), |
| 686 | &ctx, |
| 687 | ) |
| 688 | .await |
| 689 | .unwrap_err(); |
| 690 | assert!(err.to_string().contains("canceled")); |
| 691 | } |
| 692 | |
| 693 | #[tokio::test] |
| 694 | async fn list_with_status_filter() { |
| 695 | let tmp = TempDir::new().unwrap(); |
| 696 | let ctx = make_context(&tmp); |
| 697 | let tool = SendLaterTool::new("send_later"); |
| 698 | |
| 699 | // Schedule one trigger. |
| 700 | let sched_result = tool |
| 701 | .execute( |
| 702 | json!({ |
| 703 | "action": "schedule", |
| 704 | "delay_minutes": 10, |
| 705 | "message": "Pending trigger." |
| 706 | }), |
| 707 | &ctx, |
| 708 | ) |
| 709 | .await |
| 710 | .unwrap(); |
| 711 | let trigger_id = serde_json::from_str::<serde_json::Value>(&sched_result.content).unwrap() |
| 712 | ["trigger_id"] |
| 713 | .as_str() |
| 714 | .unwrap() |
| 715 | .to_string(); |
| 716 | |
| 717 | // Cancel it. |
| 718 | tool.execute( |
| 719 | json!({ "action": "cancel", "trigger_id": &trigger_id }), |
| 720 | &ctx, |
| 721 | ) |
| 722 | .await |
| 723 | .unwrap(); |
| 724 | |
| 725 | // List pending: should be empty. |
| 726 | let pending_list = tool |
| 727 | .execute(json!({ "action": "list", "status": "pending" }), &ctx) |
| 728 | .await |
| 729 | .unwrap(); |
| 730 | let pending_val: serde_json::Value = serde_json::from_str(&pending_list.content).unwrap(); |
| 731 | assert_eq!(pending_val["count"].as_u64().unwrap(), 0); |
| 732 | |
| 733 | // List canceled: should contain our trigger. |
| 734 | let canceled_list = tool |
| 735 | .execute(json!({ "action": "list", "status": "canceled" }), &ctx) |
| 736 | .await |
| 737 | .unwrap(); |
| 738 | assert!(canceled_list.content.contains(&trigger_id)); |
| 739 | } |
| 740 | |
| 741 | #[tokio::test] |
| 742 | async fn parent_trigger_id_is_preserved() { |
| 743 | let tmp = TempDir::new().unwrap(); |
| 744 | let ctx = make_context(&tmp); |
| 745 | let tool = SendLaterTool::new("send_later"); |
| 746 | |
| 747 | // First trigger (the "parent"). |
| 748 | let parent_result = tool |
| 749 | .execute( |
| 750 | json!({ |
| 751 | "action": "schedule", |
| 752 | "delay_minutes": 60, |
| 753 | "message": "First check-in." |
| 754 | }), |
| 755 | &ctx, |
| 756 | ) |
| 757 | .await |
| 758 | .unwrap(); |
| 759 | let parent_id = serde_json::from_str::<serde_json::Value>(&parent_result.content).unwrap() |
| 760 | ["trigger_id"] |
| 761 | .as_str() |
| 762 | .unwrap() |
| 763 | .to_string(); |
| 764 | |
| 765 | // Re-armed trigger referencing the parent. |
| 766 | let child_result = tool |
| 767 | .execute( |
| 768 | json!({ |
| 769 | "action": "schedule", |
| 770 | "delay_minutes": 60, |
| 771 | "message": "Second check-in (re-arm).", |
| 772 | "parent_trigger_id": &parent_id, |
| 773 | }), |
| 774 | &ctx, |
| 775 | ) |
| 776 | .await |
| 777 | .unwrap(); |
| 778 | let child_id = |
| 779 | serde_json::from_str::<serde_json::Value>(&child_result.content).unwrap()["trigger_id"] |
| 780 | .as_str() |
| 781 | .unwrap() |
| 782 | .to_string(); |
| 783 | |
| 784 | let read_result = tool |
| 785 | .execute(json!({ "action": "read", "trigger_id": &child_id }), &ctx) |
| 786 | .await |
| 787 | .unwrap(); |
| 788 | assert!(read_result.content.contains(&parent_id)); |
| 789 | } |
| 790 | |
| 791 | #[tokio::test] |
| 792 | async fn persistence_survives_manager_reload() { |
| 793 | let tmp = TempDir::new().unwrap(); |
| 794 | let ctx = make_context(&tmp); |
| 795 | let tool = SendLaterTool::new("send_later"); |
| 796 | |
| 797 | let sched_result = tool |
| 798 | .execute( |
| 799 | json!({ |
| 800 | "action": "schedule", |
| 801 | "delay_minutes": 45, |
| 802 | "message": "Persisted trigger." |
| 803 | }), |
| 804 | &ctx, |
| 805 | ) |
| 806 | .await |
| 807 | .unwrap(); |
| 808 | let trigger_id = serde_json::from_str::<serde_json::Value>(&sched_result.content).unwrap() |
| 809 | ["trigger_id"] |
| 810 | .as_str() |
| 811 | .unwrap() |
| 812 | .to_string(); |
| 813 | |
| 814 | // Open a fresh manager over the same directory to simulate restart. |
| 815 | let ctx2 = make_context(&tmp); |
| 816 | let read_result = tool |
| 817 | .execute( |
| 818 | json!({ "action": "read", "trigger_id": &trigger_id }), |
| 819 | &ctx2, |
| 820 | ) |
| 821 | .await |
| 822 | .unwrap(); |
| 823 | assert!(read_result.content.contains("Persisted trigger.")); |
| 824 | } |
| 825 | |
| 826 | #[tokio::test] |
| 827 | async fn collect_due_triggers_returns_past_pending() { |
| 828 | let tmp = TempDir::new().unwrap(); |
| 829 | let manager = AutomationManager::open(tmp.path().to_path_buf()).unwrap(); |
| 830 | |
| 831 | // Create a trigger with fire_at one hour from now — not due yet. |
| 832 | let req = crate::automation_manager::CreateDelayedTriggerRequest { |
| 833 | fire_at: chrono::Utc::now() + Duration::hours(1), |
| 834 | message: "Not due yet.".to_string(), |
| 835 | workspace: None, |
| 836 | parent_trigger_id: None, |
| 837 | }; |
| 838 | let record = manager.create_trigger(req).unwrap(); |
| 839 | let due = manager.collect_due_triggers(chrono::Utc::now()).unwrap(); |
| 840 | assert!(due.is_empty(), "should not fire a future trigger"); |
| 841 | |
| 842 | // Back-date the fire_at to the past and re-save. |
| 843 | let mut past_record = record; |
| 844 | past_record.fire_at = chrono::Utc::now() - Duration::minutes(5); |
| 845 | manager.save_trigger(&past_record).unwrap(); |
| 846 | |
| 847 | let due = manager.collect_due_triggers(chrono::Utc::now()).unwrap(); |
| 848 | assert_eq!(due.len(), 1, "should fire a past-due trigger"); |
| 849 | } |
| 850 | } |
| 851 |