| 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", "dispatching", "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 | owner_session_id: Some(context.state_namespace.clone()), |
| 308 | parent_trigger_id, |
| 309 | }; |
| 310 | |
| 311 | let record = { |
| 312 | let manager = automations.lock().await; |
| 313 | // A pending trigger is fired only by the scope that owns it |
| 314 | // (`fire_due_triggers`), and a trigger has no paused state to fall |
| 315 | // back to, so an unbound host must refuse rather than store a delayed |
| 316 | // message that never arrives. |
| 317 | crate::tools::automation::require_dispatch_owner(&manager, "schedule a delayed message")?; |
| 318 | manager.create_trigger(req).map_err(|err| { |
| 319 | ToolError::execution_failed(format!("send_later schedule failed: {err}")) |
| 320 | })? |
| 321 | }; |
| 322 | |
| 323 | Ok(ToolResult::success( |
| 324 | serde_json::to_string_pretty(&json!({ |
| 325 | "trigger_id": record.trigger_id, |
| 326 | "fire_at": record.fire_at.to_rfc3339(), |
| 327 | "status": "pending", |
| 328 | })) |
| 329 | .unwrap_or_default(), |
| 330 | )) |
| 331 | } |
| 332 | |
| 333 | async fn execute_list(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 334 | let automations = SendLaterTool::automations_from_context(context)?; |
| 335 | |
| 336 | let limit = Some(optional_u64(input, "limit", 50)? as usize); |
| 337 | let status_filter = optional_str(input, "status")? |
| 338 | .map(parse_trigger_status) |
| 339 | .transpose() |
| 340 | .map_err(ToolError::invalid_input)?; |
| 341 | |
| 342 | let records = { |
| 343 | let manager = automations.lock().await; |
| 344 | manager |
| 345 | .list_triggers_for_owner(status_filter, limit, &context.state_namespace) |
| 346 | .map_err(|err| ToolError::execution_failed(format!("send_later list failed: {err}")))? |
| 347 | }; |
| 348 | |
| 349 | let items: Vec<Value> = records |
| 350 | .iter() |
| 351 | .map(|r| { |
| 352 | json!({ |
| 353 | "trigger_id": r.trigger_id, |
| 354 | "fire_at": r.fire_at.to_rfc3339(), |
| 355 | "status": trigger_status_str(r.status), |
| 356 | "created_at": r.created_at.to_rfc3339(), |
| 357 | "message_preview": r.message.chars().take(120).collect::<String>(), |
| 358 | "workspace": r.workspace, |
| 359 | "parent_trigger_id": r.parent_trigger_id, |
| 360 | "task_id": r.task_id, |
| 361 | "error": r.error, |
| 362 | }) |
| 363 | }) |
| 364 | .collect(); |
| 365 | |
| 366 | let count = items.len(); |
| 367 | Ok(ToolResult::success( |
| 368 | serde_json::to_string_pretty(&json!({ "triggers": items, "count": count })) |
| 369 | .unwrap_or_default(), |
| 370 | )) |
| 371 | } |
| 372 | |
| 373 | async fn execute_read(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 374 | let automations = SendLaterTool::automations_from_context(context)?; |
| 375 | |
| 376 | let trigger_id = input |
| 377 | .get("trigger_id") |
| 378 | .and_then(Value::as_str) |
| 379 | .ok_or_else(|| ToolError::invalid_input("send_later read: `trigger_id` is required"))?; |
| 380 | |
| 381 | let record = { |
| 382 | let manager = automations.lock().await; |
| 383 | manager |
| 384 | .get_trigger_for_owner(trigger_id, &context.state_namespace) |
| 385 | .map_err(|err| ToolError::execution_failed(format!("send_later read failed: {err}")))? |
| 386 | }; |
| 387 | |
| 388 | Ok(ToolResult::success( |
| 389 | serde_json::to_string_pretty(&json!({ |
| 390 | "trigger_id": record.trigger_id, |
| 391 | "fire_at": record.fire_at.to_rfc3339(), |
| 392 | "status": trigger_status_str(record.status), |
| 393 | "created_at": record.created_at.to_rfc3339(), |
| 394 | "fired_at": record.fired_at.map(|t| t.to_rfc3339()), |
| 395 | "message": record.message, |
| 396 | "workspace": record.workspace, |
| 397 | "parent_trigger_id": record.parent_trigger_id, |
| 398 | "task_id": record.task_id, |
| 399 | "thread_id": record.thread_id, |
| 400 | "error": record.error, |
| 401 | })) |
| 402 | .unwrap_or_default(), |
| 403 | )) |
| 404 | } |
| 405 | |
| 406 | async fn execute_cancel(input: &Value, context: &ToolContext) -> Result<ToolResult, ToolError> { |
| 407 | let automations = SendLaterTool::automations_from_context(context)?; |
| 408 | |
| 409 | let trigger_id = input |
| 410 | .get("trigger_id") |
| 411 | .and_then(Value::as_str) |
| 412 | .ok_or_else(|| ToolError::invalid_input("send_later cancel: `trigger_id` is required"))?; |
| 413 | |
| 414 | let record = { |
| 415 | let manager = automations.lock().await; |
| 416 | manager |
| 417 | .cancel_trigger_for_owner(trigger_id, &context.state_namespace) |
| 418 | .map_err(|err| { |
| 419 | ToolError::execution_failed(format!("send_later cancel failed: {err}")) |
| 420 | })? |
| 421 | }; |
| 422 | |
| 423 | Ok(ToolResult::success( |
| 424 | serde_json::to_string_pretty(&json!({ |
| 425 | "trigger_id": record.trigger_id, |
| 426 | "status": "canceled", |
| 427 | "fire_at": record.fire_at.to_rfc3339(), |
| 428 | })) |
| 429 | .unwrap_or_default(), |
| 430 | )) |
| 431 | } |
| 432 | |
| 433 | fn parse_trigger_status(s: &str) -> Result<DelayedTriggerStatus, String> { |
| 434 | match s { |
| 435 | "pending" => Ok(DelayedTriggerStatus::Pending), |
| 436 | "dispatching" => Ok(DelayedTriggerStatus::Dispatching), |
| 437 | "fired" => Ok(DelayedTriggerStatus::Fired), |
| 438 | "canceled" => Ok(DelayedTriggerStatus::Canceled), |
| 439 | "failed" => Ok(DelayedTriggerStatus::Failed), |
| 440 | other => Err(format!( |
| 441 | "unknown trigger status '{other}'; expected one of: pending, dispatching, fired, canceled, failed" |
| 442 | )), |
| 443 | } |
| 444 | } |
| 445 | |
| 446 | fn trigger_status_str(status: DelayedTriggerStatus) -> &'static str { |
| 447 | match status { |
| 448 | DelayedTriggerStatus::Pending => "pending", |
| 449 | DelayedTriggerStatus::Dispatching => "dispatching", |
| 450 | DelayedTriggerStatus::Fired => "fired", |
| 451 | DelayedTriggerStatus::Canceled => "canceled", |
| 452 | DelayedTriggerStatus::Failed => "failed", |
| 453 | } |
| 454 | } |
| 455 | |
| 456 | // ── Tests ────────────────────────────────────────────────────────────────── |
| 457 | |
| 458 | #[cfg(test)] |
| 459 | mod tests { |
| 460 | use std::sync::Arc; |
| 461 | |
| 462 | use chrono::Duration; |
| 463 | use serde_json::json; |
| 464 | use tempfile::TempDir; |
| 465 | use tokio::sync::Mutex; |
| 466 | |
| 467 | use crate::automation_manager::{AutomationManager, DelayedTriggerStatus}; |
| 468 | use crate::tools::spec::{RuntimeToolServices, ToolContext, ToolSpec}; |
| 469 | |
| 470 | use super::SendLaterTool; |
| 471 | |
| 472 | fn make_context(tmp: &TempDir) -> ToolContext { |
| 473 | make_context_for_session(tmp, "test-session") |
| 474 | } |
| 475 | |
| 476 | fn make_context_for_session(tmp: &TempDir, session_id: &str) -> ToolContext { |
| 477 | let manager = AutomationManager::open_for_test(tmp.path().to_path_buf()).unwrap(); |
| 478 | let shared = Arc::new(Mutex::new(manager)); |
| 479 | ToolContext::new(".") |
| 480 | .with_state_namespace(session_id) |
| 481 | .with_runtime_services(RuntimeToolServices { |
| 482 | automations: Some(shared), |
| 483 | ..Default::default() |
| 484 | }) |
| 485 | } |
| 486 | |
| 487 | #[tokio::test] |
| 488 | async fn schedule_with_delay_minutes_succeeds() { |
| 489 | let tmp = TempDir::new().unwrap(); |
| 490 | let ctx = make_context(&tmp); |
| 491 | let tool = SendLaterTool::new("send_later"); |
| 492 | |
| 493 | let result = tool |
| 494 | .execute( |
| 495 | json!({ |
| 496 | "action": "schedule", |
| 497 | "delay_minutes": 60, |
| 498 | "message": "Check CI status on open PRs." |
| 499 | }), |
| 500 | &ctx, |
| 501 | ) |
| 502 | .await |
| 503 | .unwrap(); |
| 504 | |
| 505 | assert!(result.content.contains("trigger_id")); |
| 506 | assert!(result.content.contains("fire_at")); |
| 507 | assert!(result.content.contains("pending")); |
| 508 | } |
| 509 | |
| 510 | #[tokio::test] |
| 511 | async fn schedule_with_absolute_fire_at_succeeds() { |
| 512 | let tmp = TempDir::new().unwrap(); |
| 513 | let ctx = make_context(&tmp); |
| 514 | let tool = SendLaterTool::new("send_later"); |
| 515 | |
| 516 | let fire_at = (chrono::Utc::now() + Duration::minutes(30)).to_rfc3339(); |
| 517 | let result = tool |
| 518 | .execute( |
| 519 | json!({ |
| 520 | "action": "schedule", |
| 521 | "fire_at": fire_at, |
| 522 | "message": "Check PR mergeability." |
| 523 | }), |
| 524 | &ctx, |
| 525 | ) |
| 526 | .await |
| 527 | .unwrap(); |
| 528 | |
| 529 | assert!(result.content.contains("trig_")); |
| 530 | } |
| 531 | |
| 532 | #[tokio::test] |
| 533 | async fn schedule_rejects_mutually_exclusive_inputs() { |
| 534 | let tmp = TempDir::new().unwrap(); |
| 535 | let ctx = make_context(&tmp); |
| 536 | let tool = SendLaterTool::new("send_later"); |
| 537 | |
| 538 | let fire_at = (chrono::Utc::now() + Duration::minutes(30)).to_rfc3339(); |
| 539 | let err = tool |
| 540 | .execute( |
| 541 | json!({ |
| 542 | "action": "schedule", |
| 543 | "delay_minutes": 60, |
| 544 | "fire_at": fire_at, |
| 545 | "message": "Should fail." |
| 546 | }), |
| 547 | &ctx, |
| 548 | ) |
| 549 | .await |
| 550 | .unwrap_err(); |
| 551 | |
| 552 | assert!(err.to_string().contains("mutually exclusive")); |
| 553 | } |
| 554 | |
| 555 | #[tokio::test] |
| 556 | async fn schedule_rejects_missing_timing_input() { |
| 557 | let tmp = TempDir::new().unwrap(); |
| 558 | let ctx = make_context(&tmp); |
| 559 | let tool = SendLaterTool::new("send_later"); |
| 560 | |
| 561 | let err = tool |
| 562 | .execute( |
| 563 | json!({ |
| 564 | "action": "schedule", |
| 565 | "message": "No timing provided." |
| 566 | }), |
| 567 | &ctx, |
| 568 | ) |
| 569 | .await |
| 570 | .unwrap_err(); |
| 571 | |
| 572 | assert!( |
| 573 | err.to_string().contains("required"), |
| 574 | "expected required-timing error, got: {err}" |
| 575 | ); |
| 576 | } |
| 577 | |
| 578 | #[tokio::test] |
| 579 | async fn schedule_rejects_past_fire_at() { |
| 580 | let tmp = TempDir::new().unwrap(); |
| 581 | let ctx = make_context(&tmp); |
| 582 | let tool = SendLaterTool::new("send_later"); |
| 583 | |
| 584 | let fire_at = (chrono::Utc::now() - Duration::minutes(5)).to_rfc3339(); |
| 585 | let err = tool |
| 586 | .execute( |
| 587 | json!({ |
| 588 | "action": "schedule", |
| 589 | "fire_at": fire_at, |
| 590 | "message": "This is in the past." |
| 591 | }), |
| 592 | &ctx, |
| 593 | ) |
| 594 | .await |
| 595 | .unwrap_err(); |
| 596 | |
| 597 | assert!( |
| 598 | err.to_string().contains("future"), |
| 599 | "expected future-time error, got: {err}" |
| 600 | ); |
| 601 | } |
| 602 | |
| 603 | #[tokio::test] |
| 604 | async fn schedule_rejects_malformed_fire_at() { |
| 605 | let tmp = TempDir::new().unwrap(); |
| 606 | let ctx = make_context(&tmp); |
| 607 | let tool = SendLaterTool::new("send_later"); |
| 608 | |
| 609 | let err = tool |
| 610 | .execute( |
| 611 | json!({ |
| 612 | "action": "schedule", |
| 613 | "fire_at": "not-a-timestamp", |
| 614 | "message": "Bad time." |
| 615 | }), |
| 616 | &ctx, |
| 617 | ) |
| 618 | .await |
| 619 | .unwrap_err(); |
| 620 | |
| 621 | assert!( |
| 622 | err.to_string().contains("invalid `fire_at`"), |
| 623 | "expected parse error, got: {err}" |
| 624 | ); |
| 625 | } |
| 626 | |
| 627 | #[tokio::test] |
| 628 | async fn list_and_read_round_trip() { |
| 629 | let tmp = TempDir::new().unwrap(); |
| 630 | let ctx = make_context(&tmp); |
| 631 | let tool = SendLaterTool::new("send_later"); |
| 632 | |
| 633 | // Schedule a trigger. |
| 634 | let sched_result = tool |
| 635 | .execute( |
| 636 | json!({ |
| 637 | "action": "schedule", |
| 638 | "delay_minutes": 15, |
| 639 | "message": "PR watcher check-in." |
| 640 | }), |
| 641 | &ctx, |
| 642 | ) |
| 643 | .await |
| 644 | .unwrap(); |
| 645 | |
| 646 | let sched_val: serde_json::Value = serde_json::from_str(&sched_result.content).unwrap(); |
| 647 | let trigger_id = sched_val["trigger_id"].as_str().unwrap(); |
| 648 | |
| 649 | // List should include it. |
| 650 | let list_result = tool |
| 651 | .execute(json!({ "action": "list" }), &ctx) |
| 652 | .await |
| 653 | .unwrap(); |
| 654 | assert!(list_result.content.contains(trigger_id)); |
| 655 | |
| 656 | // Read should return full detail. |
| 657 | let read_result = tool |
| 658 | .execute(json!({ "action": "read", "trigger_id": trigger_id }), &ctx) |
| 659 | .await |
| 660 | .unwrap(); |
| 661 | assert!(read_result.content.contains("PR watcher check-in.")); |
| 662 | } |
| 663 | |
| 664 | #[tokio::test] |
| 665 | async fn cancel_pending_trigger() { |
| 666 | let tmp = TempDir::new().unwrap(); |
| 667 | let ctx = make_context(&tmp); |
| 668 | let tool = SendLaterTool::new("send_later"); |
| 669 | |
| 670 | let sched_result = tool |
| 671 | .execute( |
| 672 | json!({ |
| 673 | "action": "schedule", |
| 674 | "delay_minutes": 30, |
| 675 | "message": "Will be canceled." |
| 676 | }), |
| 677 | &ctx, |
| 678 | ) |
| 679 | .await |
| 680 | .unwrap(); |
| 681 | |
| 682 | let trigger_id = serde_json::from_str::<serde_json::Value>(&sched_result.content).unwrap() |
| 683 | ["trigger_id"] |
| 684 | .as_str() |
| 685 | .unwrap() |
| 686 | .to_string(); |
| 687 | |
| 688 | let cancel_result = tool |
| 689 | .execute( |
| 690 | json!({ "action": "cancel", "trigger_id": &trigger_id }), |
| 691 | &ctx, |
| 692 | ) |
| 693 | .await |
| 694 | .unwrap(); |
| 695 | |
| 696 | assert!(cancel_result.content.contains("canceled")); |
| 697 | |
| 698 | // Canceling again should fail. |
| 699 | let err = tool |
| 700 | .execute( |
| 701 | json!({ "action": "cancel", "trigger_id": &trigger_id }), |
| 702 | &ctx, |
| 703 | ) |
| 704 | .await |
| 705 | .unwrap_err(); |
| 706 | assert!(err.to_string().contains("canceled")); |
| 707 | } |
| 708 | |
| 709 | #[tokio::test] |
| 710 | async fn list_with_status_filter() { |
| 711 | let tmp = TempDir::new().unwrap(); |
| 712 | let ctx = make_context(&tmp); |
| 713 | let tool = SendLaterTool::new("send_later"); |
| 714 | |
| 715 | // Schedule one trigger. |
| 716 | let sched_result = tool |
| 717 | .execute( |
| 718 | json!({ |
| 719 | "action": "schedule", |
| 720 | "delay_minutes": 10, |
| 721 | "message": "Pending trigger." |
| 722 | }), |
| 723 | &ctx, |
| 724 | ) |
| 725 | .await |
| 726 | .unwrap(); |
| 727 | let trigger_id = serde_json::from_str::<serde_json::Value>(&sched_result.content).unwrap() |
| 728 | ["trigger_id"] |
| 729 | .as_str() |
| 730 | .unwrap() |
| 731 | .to_string(); |
| 732 | |
| 733 | // Cancel it. |
| 734 | tool.execute( |
| 735 | json!({ "action": "cancel", "trigger_id": &trigger_id }), |
| 736 | &ctx, |
| 737 | ) |
| 738 | .await |
| 739 | .unwrap(); |
| 740 | |
| 741 | // List pending: should be empty. |
| 742 | let pending_list = tool |
| 743 | .execute(json!({ "action": "list", "status": "pending" }), &ctx) |
| 744 | .await |
| 745 | .unwrap(); |
| 746 | let pending_val: serde_json::Value = serde_json::from_str(&pending_list.content).unwrap(); |
| 747 | assert_eq!(pending_val["count"].as_u64().unwrap(), 0); |
| 748 | |
| 749 | // List canceled: should contain our trigger. |
| 750 | let canceled_list = tool |
| 751 | .execute(json!({ "action": "list", "status": "canceled" }), &ctx) |
| 752 | .await |
| 753 | .unwrap(); |
| 754 | assert!(canceled_list.content.contains(&trigger_id)); |
| 755 | } |
| 756 | |
| 757 | #[tokio::test] |
| 758 | async fn parent_trigger_id_is_preserved() { |
| 759 | let tmp = TempDir::new().unwrap(); |
| 760 | let ctx = make_context(&tmp); |
| 761 | let tool = SendLaterTool::new("send_later"); |
| 762 | |
| 763 | // First trigger (the "parent"). |
| 764 | let parent_result = tool |
| 765 | .execute( |
| 766 | json!({ |
| 767 | "action": "schedule", |
| 768 | "delay_minutes": 60, |
| 769 | "message": "First check-in." |
| 770 | }), |
| 771 | &ctx, |
| 772 | ) |
| 773 | .await |
| 774 | .unwrap(); |
| 775 | let parent_id = serde_json::from_str::<serde_json::Value>(&parent_result.content).unwrap() |
| 776 | ["trigger_id"] |
| 777 | .as_str() |
| 778 | .unwrap() |
| 779 | .to_string(); |
| 780 | |
| 781 | // Re-armed trigger referencing the parent. |
| 782 | let child_result = tool |
| 783 | .execute( |
| 784 | json!({ |
| 785 | "action": "schedule", |
| 786 | "delay_minutes": 60, |
| 787 | "message": "Second check-in (re-arm).", |
| 788 | "parent_trigger_id": &parent_id, |
| 789 | }), |
| 790 | &ctx, |
| 791 | ) |
| 792 | .await |
| 793 | .unwrap(); |
| 794 | let child_id = |
| 795 | serde_json::from_str::<serde_json::Value>(&child_result.content).unwrap()["trigger_id"] |
| 796 | .as_str() |
| 797 | .unwrap() |
| 798 | .to_string(); |
| 799 | |
| 800 | let read_result = tool |
| 801 | .execute(json!({ "action": "read", "trigger_id": &child_id }), &ctx) |
| 802 | .await |
| 803 | .unwrap(); |
| 804 | assert!(read_result.content.contains(&parent_id)); |
| 805 | } |
| 806 | |
| 807 | #[tokio::test] |
| 808 | async fn persistence_survives_manager_reload() { |
| 809 | let tmp = TempDir::new().unwrap(); |
| 810 | let ctx = make_context(&tmp); |
| 811 | let tool = SendLaterTool::new("send_later"); |
| 812 | |
| 813 | let sched_result = tool |
| 814 | .execute( |
| 815 | json!({ |
| 816 | "action": "schedule", |
| 817 | "delay_minutes": 45, |
| 818 | "message": "Persisted trigger." |
| 819 | }), |
| 820 | &ctx, |
| 821 | ) |
| 822 | .await |
| 823 | .unwrap(); |
| 824 | let trigger_id = serde_json::from_str::<serde_json::Value>(&sched_result.content).unwrap() |
| 825 | ["trigger_id"] |
| 826 | .as_str() |
| 827 | .unwrap() |
| 828 | .to_string(); |
| 829 | |
| 830 | // Open a fresh manager over the same directory to simulate restart. |
| 831 | let ctx2 = make_context(&tmp); |
| 832 | let read_result = tool |
| 833 | .execute( |
| 834 | json!({ "action": "read", "trigger_id": &trigger_id }), |
| 835 | &ctx2, |
| 836 | ) |
| 837 | .await |
| 838 | .unwrap(); |
| 839 | assert!(read_result.content.contains("Persisted trigger.")); |
| 840 | } |
| 841 | |
| 842 | #[tokio::test] |
| 843 | async fn trigger_controls_are_session_owned_and_legacy_records_fail_closed() { |
| 844 | let tmp = TempDir::new().unwrap(); |
| 845 | let session_a = make_context_for_session(&tmp, "session-a"); |
| 846 | let session_b = make_context_for_session(&tmp, "session-b"); |
| 847 | let tool = SendLaterTool::new("send_later"); |
| 848 | |
| 849 | let session_b_result = tool |
| 850 | .execute( |
| 851 | json!({ |
| 852 | "action": "schedule", |
| 853 | "delay_minutes": 30, |
| 854 | "message": "session B continuation" |
| 855 | }), |
| 856 | &session_b, |
| 857 | ) |
| 858 | .await |
| 859 | .unwrap(); |
| 860 | let session_b_id = serde_json::from_str::<serde_json::Value>(&session_b_result.content) |
| 861 | .unwrap()["trigger_id"] |
| 862 | .as_str() |
| 863 | .unwrap() |
| 864 | .to_string(); |
| 865 | |
| 866 | let session_a_result = tool |
| 867 | .execute( |
| 868 | json!({ |
| 869 | "action": "schedule", |
| 870 | "delay_minutes": 30, |
| 871 | "message": "session A continuation" |
| 872 | }), |
| 873 | &session_a, |
| 874 | ) |
| 875 | .await |
| 876 | .unwrap(); |
| 877 | let session_a_id = serde_json::from_str::<serde_json::Value>(&session_a_result.content) |
| 878 | .unwrap()["trigger_id"] |
| 879 | .as_str() |
| 880 | .unwrap() |
| 881 | .to_string(); |
| 882 | |
| 883 | let manager = AutomationManager::open_for_test(tmp.path().to_path_buf()).unwrap(); |
| 884 | let mut legacy = manager |
| 885 | .create_trigger(crate::automation_manager::CreateDelayedTriggerRequest { |
| 886 | fire_at: chrono::Utc::now() + Duration::hours(1), |
| 887 | message: "ownerless legacy continuation".to_string(), |
| 888 | workspace: None, |
| 889 | owner_session_id: None, |
| 890 | parent_trigger_id: None, |
| 891 | }) |
| 892 | .unwrap(); |
| 893 | |
| 894 | let session_b_list = tool |
| 895 | .execute(json!({ "action": "list", "limit": 1 }), &session_b) |
| 896 | .await |
| 897 | .unwrap(); |
| 898 | assert!(session_b_list.content.contains(&session_b_id)); |
| 899 | assert!(!session_b_list.content.contains(&session_a_id)); |
| 900 | assert!(!session_b_list.content.contains(&legacy.trigger_id)); |
| 901 | |
| 902 | for action in ["read", "cancel"] { |
| 903 | let error = tool |
| 904 | .execute( |
| 905 | json!({ "action": action, "trigger_id": &session_a_id }), |
| 906 | &session_b, |
| 907 | ) |
| 908 | .await |
| 909 | .unwrap_err(); |
| 910 | assert!(error.to_string().contains("not found"), "{error}"); |
| 911 | } |
| 912 | let still_pending = manager.get_trigger(&session_a_id).unwrap(); |
| 913 | assert_eq!(still_pending.status, DelayedTriggerStatus::Pending); |
| 914 | |
| 915 | let legacy_error = tool |
| 916 | .execute( |
| 917 | json!({ "action": "read", "trigger_id": &legacy.trigger_id }), |
| 918 | &session_a, |
| 919 | ) |
| 920 | .await |
| 921 | .unwrap_err(); |
| 922 | assert!(legacy_error.to_string().contains("not found")); |
| 923 | |
| 924 | let restored_a = tool |
| 925 | .execute( |
| 926 | json!({ "action": "read", "trigger_id": &session_a_id }), |
| 927 | &session_a, |
| 928 | ) |
| 929 | .await |
| 930 | .unwrap(); |
| 931 | assert!( |
| 932 | restored_a.content.contains("session A continuation"), |
| 933 | "switching A to B and back must restore A's controls" |
| 934 | ); |
| 935 | |
| 936 | let own_cancel = tool |
| 937 | .execute( |
| 938 | json!({ "action": "cancel", "trigger_id": &session_b_id }), |
| 939 | &session_b, |
| 940 | ) |
| 941 | .await |
| 942 | .unwrap(); |
| 943 | assert!(own_cancel.content.contains("canceled")); |
| 944 | |
| 945 | legacy.fire_at = chrono::Utc::now() - Duration::minutes(1); |
| 946 | manager.save_trigger(&legacy).unwrap(); |
| 947 | assert!( |
| 948 | manager |
| 949 | .collect_due_triggers(chrono::Utc::now()) |
| 950 | .unwrap() |
| 951 | .iter() |
| 952 | .all(|record| record.trigger_id != legacy.trigger_id), |
| 953 | "ownerless legacy triggers must never fire" |
| 954 | ); |
| 955 | } |
| 956 | |
| 957 | #[tokio::test] |
| 958 | async fn collect_due_triggers_returns_past_pending() { |
| 959 | let tmp = TempDir::new().unwrap(); |
| 960 | let manager = AutomationManager::open_for_test(tmp.path().to_path_buf()).unwrap(); |
| 961 | |
| 962 | // Create a trigger with fire_at one hour from now — not due yet. |
| 963 | let req = crate::automation_manager::CreateDelayedTriggerRequest { |
| 964 | fire_at: chrono::Utc::now() + Duration::hours(1), |
| 965 | message: "Not due yet.".to_string(), |
| 966 | workspace: None, |
| 967 | owner_session_id: Some("session-a".to_string()), |
| 968 | parent_trigger_id: None, |
| 969 | }; |
| 970 | let record = manager.create_trigger(req).unwrap(); |
| 971 | let due = manager.collect_due_triggers(chrono::Utc::now()).unwrap(); |
| 972 | assert!(due.is_empty(), "should not fire a future trigger"); |
| 973 | |
| 974 | // Back-date the fire_at to the past and re-save. |
| 975 | let mut past_record = record; |
| 976 | past_record.fire_at = chrono::Utc::now() - Duration::minutes(5); |
| 977 | manager.save_trigger(&past_record).unwrap(); |
| 978 | |
| 979 | let due = manager.collect_due_triggers(chrono::Utc::now()).unwrap(); |
| 980 | assert_eq!(due.len(), 1, "should fire a past-due trigger"); |
| 981 | } |
| 982 | |
| 983 | /// A pending trigger is fired only by the scope that owns it, and a |
| 984 | /// trigger has no paused state, so a one-shot host that attaches the |
| 985 | /// store for inspection must refuse to schedule rather than store a |
| 986 | /// delayed message that never arrives. |
| 987 | #[tokio::test] |
| 988 | async fn scheduling_requires_a_dispatch_owner() { |
| 989 | let tmp = TempDir::new().unwrap(); |
| 990 | let manager = AutomationManager::open(tmp.path().to_path_buf()).unwrap(); |
| 991 | assert!( |
| 992 | manager.execution_scope().is_none(), |
| 993 | "fixture must model the unbound one-shot host" |
| 994 | ); |
| 995 | let ctx = ToolContext::new(".").with_runtime_services(RuntimeToolServices { |
| 996 | automations: Some(Arc::new(Mutex::new(manager))), |
| 997 | ..Default::default() |
| 998 | }); |
| 999 | let tool = SendLaterTool::new("send_later"); |
| 1000 | |
| 1001 | let err = tool |
| 1002 | .execute( |
| 1003 | json!({"action": "schedule", "delay_minutes": 60, "message": "Check CI."}), |
| 1004 | &ctx, |
| 1005 | ) |
| 1006 | .await |
| 1007 | .expect_err("must refuse without a dispatch owner"); |
| 1008 | assert!( |
| 1009 | err.to_string().contains("persistent execution owner"), |
| 1010 | "{err}" |
| 1011 | ); |
| 1012 | |
| 1013 | // Inspection still works against the attached store, and nothing was |
| 1014 | // persisted by the refused schedule. |
| 1015 | let listed = tool |
| 1016 | .execute(json!({"action": "list"}), &ctx) |
| 1017 | .await |
| 1018 | .expect("list must serve an attached store"); |
| 1019 | assert!( |
| 1020 | AutomationManager::open(tmp.path().to_path_buf()) |
| 1021 | .unwrap() |
| 1022 | .list_triggers(None, None) |
| 1023 | .unwrap() |
| 1024 | .is_empty(), |
| 1025 | "a refused schedule must not leave a trigger behind: {}", |
| 1026 | listed.content |
| 1027 | ); |
| 1028 | } |
| 1029 | } |
| 1030 |