1. 1//! `tasks` — model-visible schedule management (docs/design/29-personal-os.md
2. 2//! P2). Lets the running agent create, list, edit, and delete this
3. 3//! workspace's scheduled routines from a plain-language request ("remind me
4. 4//! every weekday at 9am to check the deploy") or a spoken one relayed
5. 5//! through the same turn — from the CLI, the desktop app, or any bound chat
6. 6//! surface (Telegram/Discord/Slack) alike, since it's just another tool on
7. 7//! the same agent loop those all share.
8. 8//!
9. 9//! Deliberately thin: all validation, cron parsing, and persistence are
10. 10//! [`crate::tasks`] (`TaskStore`/`TaskDef`), the exact store the CLI
11. 11//! (`vak tasks`) and the server's `/tasks` REST endpoints already share —
12. 12//! a task created by any of the three is visible and editable from the
13. 13//! other two.
14. 14
15. 15use std::path::PathBuf;
16. 16
17. 17use serde_json::Value;
18. 18
19. 19use crate::tasks::{TaskDef, TaskError, TaskStore};
20. 20
21. 21pub struct TasksTool {
22. 22 pub sessions_home: PathBuf,
23. 23 pub cwd: PathBuf,
24. 24 /// `<surface>:<chat>` for the conversation this turn is running in, if
25. 25 /// any (see [`crate::Core::with_default_deliver_to`]). Used only when
26. 26 /// the model's `add` call omits `deliver_to` — the common case, since a
27. 27 /// user asking for a reminder from inside a chat means "tell me here,"
28. 28 /// not "tell nobody."
29. 29 pub default_deliver_to: Option<String>,
30. 30}
31. 31
32. 32fn task_error_message(e: TaskError) -> String {
33. 33 e.to_string()
34. 34}
35. 35
36. 36fn nanos_id() -> String {
37. 37 std::time::SystemTime::now()
38. 38 .duration_since(std::time::UNIX_EPOCH)
39. 39 .unwrap_or_default()
40. 40 .as_nanos()
41. 41 .to_string()
42. 42}
43. 43
44. 44fn render_task(t: &TaskDef) -> Value {
45. 45 serde_json::json!({
46. 46 "id": t.id,
47. 47 "name": t.name,
48. 48 "enabled": t.enabled,
49. 49 "prompt": if t.prompt.is_empty() { None } else { Some(t.prompt.clone()) },
50. 50 "script": t.script,
51. 51 "schedule": t.schedule,
52. 52 "timezone": t.timezone,
53. 53 "due_at": t.due_at,
54. 54 "interval_secs": if t.schedule.is_none() { Some(t.interval_secs) } else { None },
55. 55 "deliver_to": t.deliver_to,
56. 56 "model_pin": t.model_pin,
57. 57 "agent_id": t.agent_id,
58. 58 "agent_revision": t.agent_revision,
59. 59 "last_run_at": t.last_run_at,
60. 60 "last_summary": t.last_summary,
61. 61 "last_result_id": t.last_result_id,
62. 62 "last_run_status": t.last_run_status,
63. 63 "last_delivery_state": t.last_delivery_state,
64. 64 })
65. 65}
66. 66
67. 67#[async_trait::async_trait]
68. 68impl vak_tools::Tool for TasksTool {
69. 69 fn name(&self) -> &str {
70. 70 "tasks"
71. 71 }
72. 72
73. 73 fn serves(&self) -> &'static [&'static str] {
74. 74 &["orchestration"]
75. 75 }
76. 76
77. 77 fn description(&self) -> &str {
78. 78 "Manage this workspace's scheduled routines — recurring prompts or \
79. 79 shell checks that fire on their own later, without you being asked \
80. 80 again. Use for any request to be reminded, checked in on, or have \
81. 81 something run on a schedule ('remind me every morning at 8', \
82. 82 'check disk space hourly and tell me if it's low', 'run the nightly \
83. 83 report'). `action: \"add\"` creates one; give `cron` for a specific \
84. 84 schedule (5-field, local time: min hour dom mon dow, e.g. '0 9 * * \
85. 85 1-5' for weekdays at 9am) or `every_secs` for a plain interval — \
86. 86 exactly one of the two. Give exactly one of `prompt` (a task for \
87. 87 you, running as a fresh turn later) or `script` (a one-line shell \
88. 88 check; only its output is reported, so a silent/zero-exit check \
89. 89 costs nothing). Omit `deliver_to` to have results come back to \
90. 90 this same conversation when one is bound; pass it explicitly \
91. 91 ('<surface>:<chat>') to route elsewhere. `list` shows every task \
92. 92 with its id, schedule, and last result; `enable`/`disable`/`remove` \
93. 93 take that id."
94. 94 }
95. 95
96. 96 fn schema(&self) -> Value {
97. 97 serde_json::json!({
98. 98 "type": "object",
99. 99 "properties": {
100. 100 "action": {
101. 101 "type": "string",
102. 102 "enum": ["list", "add", "enable", "disable", "remove"],
103. 103 "description": "What to do"
104. 104 },
105. 105 "id": {
106. 106 "type": "string",
107. 107 "description": "Task id — required for enable/disable/remove, from a prior 'list' or 'add'"
108. 108 },
109. 109 "name": {
110. 110 "type": "string",
111. 111 "description": "Short label for the task (required for add)"
112. 112 },
113. 113 "prompt": {
114. 114 "type": "string",
115. 115 "description": "What you should do when this fires, in your own words (XOR with 'script')"
116. 116 },
117. 117 "script": {
118. 118 "type": "string",
119. 119 "description": "A one-line shell command to run instead of a prompt turn (XOR with 'prompt')"
120. 120 },
121. 121 "cron": {
122. 122 "type": "string",
123. 123 "description": "5-field cron expression, local time (XOR with 'every_secs')"
124. 124 },
125. 125 "every_secs": {
126. 126 "type": "integer",
127. 127 "description": "Plain interval in seconds since last run (XOR with 'cron'; default 3600 if neither is given)"
128. 128 },
129. 129 "timezone": {
130. 130 "type": "string",
131. 131 "description": "Optional IANA timezone name for cron interpretation, such as America/New_York"
132. 132 },
133. 133 "due_at": {
134. 134 "type": "string",
135. 135 "description": "Optional one-shot UTC timestamp in RFC3339; cannot be combined with cron"
136. 136 },
137. 137 "deliver_to": {
138. 138 "type": "string",
139. 139 "description": "Where to send the result, as '<surface>:<chat>'. Omit to use the current conversation, when one is bound."
140. 140 },
141. 141 "model_pin": {
142. 142 "type": "string",
143. 143 "description": "Optional: pin this task to one model id instead of the workspace default"
144. 144 },
145. 145 "agent": {
146. 146 "type": "string",
147. 147 "description": "Optional saved Agent name or id; records Agent provenance for future runs"
148. 148 }
149. 149 },
150. 150 "required": ["action"]
151. 151 })
152. 152 }
153. 153
154. 154 async fn execute(&self, args: &Value, _ctx: &vak_tools::ToolContext) -> vak_tools::ToolOutput {
155. 155 let mut store = match TaskStore::load(&self.sessions_home) {
156. 156 Ok(s) => s,
157. 157 Err(e) => return vak_tools::ToolOutput::error(task_error_message(e)),
158. 158 };
159. 159 let action = args.get("action").and_then(Value::as_str).unwrap_or("");
160. 160 let str_arg = |key: &str| -> Option<String> {
161. 161 args.get(key)
162. 162 .and_then(Value::as_str)
163. 163 .map(str::trim)
164. 164 .filter(|s| !s.is_empty())
165. 165 .map(str::to_string)
166. 166 };
167. 167
168. 168 match action {
169. 169 "list" => {
170. 170 let mut tasks = store.for_cwd(&self.cwd);
171. 171 tasks.sort_by_key(|t| t.created_at);
172. 172 if tasks.is_empty() {
173. 173 return vak_tools::ToolOutput::ok(
174. 174 "no scheduled tasks in this workspace yet".to_string(),
175. 175 );
176. 176 }
177. 177 let rendered: Vec<Value> = tasks.iter().map(render_task).collect();
178. 178 vak_tools::ToolOutput::ok(
179. 179 serde_json::to_string_pretty(&rendered).unwrap_or_default(),
180. 180 )
181. 181 }
182. 182 "add" => {
183. 183 let Some(name) = str_arg("name") else {
184. 184 return vak_tools::ToolOutput::error("'name' is required for action 'add'");
185. 185 };
186. 186 let prompt = str_arg("prompt");
187. 187 let script = str_arg("script");
188. 188 let cron = str_arg("cron");
189. 189 let every_secs = args.get("every_secs").and_then(Value::as_u64);
190. 190 if cron.is_some() && every_secs.is_some() {
191. 191 return vak_tools::ToolOutput::error("give 'cron' or 'every_secs', not both");
192. 192 }
193. 193 let (prompt, script) = match (prompt, script) {
194. 194 (Some(p), None) => (p, None),
195. 195 (None, Some(s)) => (String::new(), Some(s)),
196. 196 (None, None) => {
197. 197 return vak_tools::ToolOutput::error(
198. 198 "exactly one of 'prompt' or 'script' is required",
199. 199 );
200. 200 }
201. 201 (Some(_), Some(_)) => {
202. 202 return vak_tools::ToolOutput::error("give 'prompt' or 'script', not both");
203. 203 }
204. 204 };
205. 205 let deliver_to = str_arg("deliver_to").or_else(|| self.default_deliver_to.clone());
206. 206 let agent_id = str_arg("agent");
207. 207 let agent_revision = agent_id.as_ref().and_then(|id| {
208. 208 std::fs::read_to_string(self.cwd.join(".vak/agents.json"))
209. 209 .ok()
210. 210 .and_then(|raw| serde_json::from_str::<Vec<Value>>(&raw).ok())
211. 211 .and_then(|profiles| {
212. 212 profiles.into_iter().find(|profile| {
213. 213 profile.get("id").and_then(Value::as_str) == Some(id)
214. 214 || profile
215. 215 .get("name")
216. 216 .and_then(Value::as_str)
217. 217 .is_some_and(|name| name.eq_ignore_ascii_case(id))
218. 218 })
219. 219 })
220. 220 .and_then(|profile| profile.get("revision").and_then(Value::as_u64))
221. 221 });
222. 222 if let Some(d) = &deliver_to
223. 223 && !d.contains(':')
224. 224 {
225. 225 return vak_tools::ToolOutput::error(
226. 226 "'deliver_to' must be '<surface>:<chat>', e.g. 'telegram:12345'",
227. 227 );
228. 228 }
229. 229 let task = TaskDef {
230. 230 id: format!("agent-{}", nanos_id()),
231. 231 name,
232. 232 prompt,
233. 233 interval_secs: every_secs.unwrap_or(3600),
234. 234 enabled: true,
235. 235 cwd: self.cwd.clone(),
236. 236 created_at: chrono::Utc::now(),
237. 237 last_run_at: None,
238. 238 last_session_id: None,
239. 239 last_summary: None,
240. 240 last_result_id: None,
241. 241 last_run_status: None,
242. 242 last_delivery_state: None,
243. 243 last_wt: None,
244. 244 deliver_to,
245. 245 schedule: cron,
246. 246 timezone: args
247. 247 .get("timezone")
248. 248 .and_then(Value::as_str)
249. 249 .map(str::to_string),
250. 250 due_at: args
251. 251 .get("due_at")
252. 252 .and_then(Value::as_str)
253. 253 .and_then(|value| value.parse().ok()),
254. 254 script,
255. 255 model_pin: str_arg("model_pin"),
256. 256 agent_id,
257. 257 agent_revision,
258. 258 };
259. 259 if let Err(e) = task.validate() {
260. 260 return vak_tools::ToolOutput::error(task_error_message(e));
261. 261 }
262. 262 let rendered = render_task(&task);
263. 263 store.put(task);
264. 264 if let Err(e) = store.save() {
265. 265 return vak_tools::ToolOutput::error(task_error_message(e));
266. 266 }
267. 267 vak_tools::ToolOutput::ok(format!(
268. 268 "created task {}\n{}",
269. 269 rendered["id"].as_str().unwrap_or_default(),
270. 270 serde_json::to_string_pretty(&rendered).unwrap_or_default()
271. 271 ))
272. 272 }
273. 273 "enable" | "disable" => {
274. 274 let Some(id) = str_arg("id") else {
275. 275 return vak_tools::ToolOutput::error(format!(
276. 276 "'id' is required for action '{action}'"
277. 277 ));
278. 278 };
279. 279 let Some(mut task) = store.get(&id).cloned() else {
280. 280 return vak_tools::ToolOutput::error(format!("no task '{id}'"));
281. 281 };
282. 282 task.enabled = action == "enable";
283. 283 store.put(task);
284. 284 if let Err(e) = store.save() {
285. 285 return vak_tools::ToolOutput::error(task_error_message(e));
286. 286 }
287. 287 vak_tools::ToolOutput::ok(format!(
288. 288 "{} task {id}",
289. 289 if action == "enable" {
290. 290 "enabled"
291. 291 } else {
292. 292 "disabled"
293. 293 }
294. 294 ))
295. 295 }
296. 296 "remove" => {
297. 297 let Some(id) = str_arg("id") else {
298. 298 return vak_tools::ToolOutput::error("'id' is required for action 'remove'");
299. 299 };
300. 300 if !store.remove(&id) {
301. 301 return vak_tools::ToolOutput::error(format!("no task '{id}'"));
302. 302 }
303. 303 if let Err(e) = store.save() {
304. 304 return vak_tools::ToolOutput::error(task_error_message(e));
305. 305 }
306. 306 vak_tools::ToolOutput::ok(format!("removed task {id}"))
307. 307 }
308. 308 other => vak_tools::ToolOutput::error(format!(
309. 309 "unknown action '{other}'; expected list, add, enable, disable, or remove"
310. 310 )),
311. 311 }
312. 312 }
313. 313}
314. 314
315. 315#[cfg(test)]
316. 316mod tests {
317. 317 #![allow(clippy::unwrap_used, clippy::expect_used)]
318. 318 use super::*;
319. 319 use vak_tools::Tool;
320. 320
321. 321 fn tool(dir: &std::path::Path) -> TasksTool {
322. 322 TasksTool {
323. 323 sessions_home: dir.to_path_buf(),
324. 324 cwd: PathBuf::from("/ws"),
325. 325 default_deliver_to: None,
326. 326 }
327. 327 }
328. 328
329. 329 fn ctx() -> vak_tools::ToolContext {
330. 330 vak_tools::ToolContext::new(PathBuf::from("/ws"))
331. 331 }
332. 332
333. 333 #[tokio::test]
334. 334 async fn add_then_list_round_trips_through_the_shared_store() {
335. 335 let dir = tempfile::tempdir().unwrap();
336. 336 let t = tool(dir.path());
337. 337 let out = t
338. 338 .execute(
339. 339 &serde_json::json!({
340. 340 "action": "add",
341. 341 "name": "nightly digest",
342. 342 "prompt": "summarize today",
343. 343 "cron": "0 21 * * *"
344. 344 }),
345. 345 &ctx(),
346. 346 )
347. 347 .await;
348. 348 assert!(!out.is_error, "{}", out.content);
349. 349 assert!(out.content.contains("created task agent-"));
350. 350
351. 351 let out = t
352. 352 .execute(&serde_json::json!({"action": "list"}), &ctx())
353. 353 .await;
354. 354 assert!(!out.is_error);
355. 355 assert!(out.content.contains("nightly digest"));
356. 356 assert!(out.content.contains("0 21 * * *"));
357. 357
358. 358 // Same file the CLI/server store reads — round-trips unchanged.
359. 359 let reloaded = TaskStore::load(dir.path()).unwrap();
360. 360 assert_eq!(reloaded.for_cwd(&PathBuf::from("/ws")).len(), 1);
361. 361 }
362. 362
363. 363 #[tokio::test]
364. 364 async fn add_defaults_deliver_to_from_the_bound_chat() {
365. 365 let dir = tempfile::tempdir().unwrap();
366. 366 let mut t = tool(dir.path());
367. 367 t.default_deliver_to = Some("telegram:12345".to_string());
368. 368 let out = t
369. 369 .execute(
370. 370 &serde_json::json!({
371. 371 "action": "add",
372. 372 "name": "reminder",
373. 373 "prompt": "check in",
374. 374 "every_secs": 3600
375. 375 }),
376. 376 &ctx(),
377. 377 )
378. 378 .await;
379. 379 assert!(!out.is_error, "{}", out.content);
380. 380 assert!(out.content.contains("telegram:12345"));
381. 381 }
382. 382
383. 383 #[tokio::test]
384. 384 async fn explicit_deliver_to_overrides_the_chat_default() {
385. 385 let dir = tempfile::tempdir().unwrap();
386. 386 let mut t = tool(dir.path());
387. 387 t.default_deliver_to = Some("telegram:12345".to_string());
388. 388 let out = t
389. 389 .execute(
390. 390 &serde_json::json!({
391. 391 "action": "add",
392. 392 "name": "reminder",
393. 393 "prompt": "check in",
394. 394 "every_secs": 3600,
395. 395 "deliver_to": "log:audit"
396. 396 }),
397. 397 &ctx(),
398. 398 )
399. 399 .await;
400. 400 assert!(!out.is_error, "{}", out.content);
401. 401 assert!(out.content.contains("log:audit"));
402. 402 assert!(!out.content.contains("telegram:12345"));
403. 403 }
404. 404
405. 405 #[tokio::test]
406. 406 async fn cron_and_interval_together_is_rejected() {
407. 407 let dir = tempfile::tempdir().unwrap();
408. 408 let t = tool(dir.path());
409. 409 let out = t
410. 410 .execute(
411. 411 &serde_json::json!({
412. 412 "action": "add",
413. 413 "name": "x",
414. 414 "prompt": "p",
415. 415 "cron": "0 9 * * *",
416. 416 "every_secs": 60
417. 417 }),
418. 418 &ctx(),
419. 419 )
420. 420 .await;
421. 421 assert!(out.is_error);
422. 422 assert!(out.content.contains("not both"));
423. 423 }
424. 424
425. 425 #[tokio::test]
426. 426 async fn prompt_and_script_together_is_rejected() {
427. 427 let dir = tempfile::tempdir().unwrap();
428. 428 let t = tool(dir.path());
429. 429 let out = t
430. 430 .execute(
431. 431 &serde_json::json!({
432. 432 "action": "add",
433. 433 "name": "x",
434. 434 "prompt": "p",
435. 435 "script": "echo hi"
436. 436 }),
437. 437 &ctx(),
438. 438 )
439. 439 .await;
440. 440 assert!(out.is_error);
441. 441 assert!(out.content.contains("not both"));
442. 442 }
443. 443
444. 444 #[tokio::test]
445. 445 async fn bad_cron_is_rejected_before_touching_the_store() {
446. 446 let dir = tempfile::tempdir().unwrap();
447. 447 let t = tool(dir.path());
448. 448 let out = t
449. 449 .execute(
450. 450 &serde_json::json!({
451. 451 "action": "add",
452. 452 "name": "x",
453. 453 "prompt": "p",
454. 454 "cron": "not a cron"
455. 455 }),
456. 456 &ctx(),
457. 457 )
458. 458 .await;
459. 459 assert!(out.is_error);
460. 460 assert!(!dir.path().join("tasks.json").exists());
461. 461 }
462. 462
463. 463 #[tokio::test]
464. 464 async fn enable_disable_remove_round_trip() {
465. 465 let dir = tempfile::tempdir().unwrap();
466. 466 let t = tool(dir.path());
467. 467 t.execute(
468. 468 &serde_json::json!({"action": "add", "name": "x", "script": "true"}),
469. 469 &ctx(),
470. 470 )
471. 471 .await;
472. 472 let store = TaskStore::load(dir.path()).unwrap();
473. 473 let id = store.for_cwd(&PathBuf::from("/ws"))[0].id.clone();
474. 474
475. 475 let out = t
476. 476 .execute(&serde_json::json!({"action": "disable", "id": id}), &ctx())
477. 477 .await;
478. 478 assert!(!out.is_error);
479. 479 let store = TaskStore::load(dir.path()).unwrap();
480. 480 assert!(!store.get(&id).unwrap().enabled);
481. 481
482. 482 let out = t
483. 483 .execute(&serde_json::json!({"action": "enable", "id": id}), &ctx())
484. 484 .await;
485. 485 assert!(!out.is_error);
486. 486 let store = TaskStore::load(dir.path()).unwrap();
487. 487 assert!(store.get(&id).unwrap().enabled);
488. 488
489. 489 let out = t
490. 490 .execute(&serde_json::json!({"action": "remove", "id": id}), &ctx())
491. 491 .await;
492. 492 assert!(!out.is_error);
493. 493 let store = TaskStore::load(dir.path()).unwrap();
494. 494 assert!(store.get(&id).is_none());
495. 495 }
496. 496
497. 497 #[tokio::test]
498. 498 async fn unknown_id_is_a_typed_error_not_a_panic() {
499. 499 let dir = tempfile::tempdir().unwrap();
500. 500 let t = tool(dir.path());
501. 501 let out = t
502. 502 .execute(
503. 503 &serde_json::json!({"action": "remove", "id": "nope"}),
504. 504 &ctx(),
505. 505 )
506. 506 .await;
507. 507 assert!(out.is_error);
508. 508 assert!(out.content.contains("no task 'nope'"));
509. 509 }
510. 510}
511. 511