automation_manager.rs - CodeWhale

automation_manager.rs

1	//! Durable automation records and scheduler support.
2	//!
3	//! Automations are local-first recurring jobs that enqueue standard background
4	//! tasks. This module stores automation definitions and run history under
5	//! `~/.codewhale/automations` (or `DEEPSEEK_AUTOMATIONS_DIR` override).
6
7	use std::collections::BTreeMap;
8	use std::fs;
9	use std::future::Future;
10	use std::path::{Path, PathBuf};
11	use std::sync::Arc;
12
13	use anyhow::{Context, Result, bail};
14	use chrono::{
15	DateTime, Datelike, Duration, Local, NaiveDateTime, TimeZone, Timelike, Utc, Weekday,
16	};
17	use serde::{Deserialize, Serialize};
18	use tokio::sync::Mutex;
19	use tokio::time::sleep;
20	use tokio_util::sync::CancellationToken;
21	use uuid::Uuid;
22
23	use crate::task_manager::{NewTaskRequest, SharedTaskManager, TaskStatus};
24	use crate::utils::spawn_supervised;
25
26	/// Current automation record schema. `pub(crate)` so the Operate keepalive
27	/// can build a fixed-id record directly (no create/delete id swap).
28	// v2 pins provider identity. Older runtimes must reject a pinned definition
29	// instead of silently sending its model through their current provider.
30	pub(crate) const CURRENT_AUTOMATION_SCHEMA_VERSION: u32 = 3;
31	const CURRENT_RUN_SCHEMA_VERSION: u32 = 3;
32	const CURRENT_TRIGGER_SCHEMA_VERSION: u32 = 3;
33	const DEFAULT_AUTOMATION_MODE: &str = "agent";
34	const DEFAULT_AUTOMATION_ALLOW_SHELL: bool = false;
35	const DEFAULT_AUTOMATION_TRUST_MODE: bool = false;
36	const DEFAULT_AUTOMATION_AUTO_APPROVE: bool = false;
37	const DEFAULT_AUTOMATION_DELIVERY_MODE: AutomationDeliveryMode = AutomationDeliveryMode::Task;
38	pub const AUTOMATION_WATCHER_NO_REPORT_SENTINEL: &str = "NOTHING_TO_REPORT";
39	const MAX_HOURLY_SEARCH_STEPS: usize = 24 * 21;
40	const MAX_CRON_SEARCH_MINUTES: usize = 60 * 24 * 366 * 5;
41	const fn default_automation_schema_version() -> u32 {
42	CURRENT_AUTOMATION_SCHEMA_VERSION
43	}
44
45	const fn default_run_schema_version() -> u32 {
46	CURRENT_RUN_SCHEMA_VERSION
47	}
48
49	const fn default_trigger_schema_version() -> u32 {
50	CURRENT_TRIGGER_SCHEMA_VERSION
51	}
52
53	// ── Delayed-trigger types ──────────────────────────────────────────────────
54
55	/// Status of a one-shot delayed trigger.
56	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
57	#[serde(rename_all = "snake_case")]
58	pub enum DelayedTriggerStatus {
59	/// Waiting to fire.
60	Pending,
61	/// Durable admission owns this trigger; task acceptance is being recovered.
62	Dispatching,
63	/// The trigger was fired and a task was enqueued.
64	Fired,
65	/// The trigger was explicitly canceled before it fired.
66	Canceled,
67	/// The trigger fired but failed to enqueue a task.
68	Failed,
69	}
70
71	/// A durable one-shot delayed continuation record.
72	///
73	/// Stored under `~/.codewhale/automations/triggers/{trigger_id}.json`.
74	#[derive(Debug, Clone, Serialize, Deserialize)]
75	pub struct DelayedTriggerRecord {
76	#[serde(default = "default_trigger_schema_version")]
77	pub schema_version: u32,
78	pub trigger_id: String,
79	/// Absolute UTC time at which the trigger should fire.
80	pub fire_at: DateTime<Utc>,
81	/// The message that will be submitted as a new task when the trigger fires.
82	pub message: String,
83	/// Working directory for the task that fires when the trigger trips.
84	#[serde(skip_serializing_if = "Option::is_none")]
85	pub workspace: Option<PathBuf>,
86	/// Session that scheduled this trigger. Missing legacy ownership fails closed.
87	#[serde(default, skip_serializing_if = "Option::is_none")]
88	pub owner_session_id: Option<String>,
89	pub status: DelayedTriggerStatus,
90	pub created_at: DateTime<Utc>,
91	#[serde(skip_serializing_if = "Option::is_none")]
92	pub fired_at: Option<DateTime<Utc>>,
93	#[serde(skip_serializing_if = "Option::is_none")]
94	pub task_id: Option<String>,
95	#[serde(skip_serializing_if = "Option::is_none")]
96	pub thread_id: Option<String>,
97	#[serde(skip_serializing_if = "Option::is_none")]
98	pub error: Option<String>,
99	/// Optional lineage: the trigger id that scheduled this one (for re-arm chains).
100	#[serde(skip_serializing_if = "Option::is_none")]
101	pub parent_trigger_id: Option<String>,
102	#[serde(default, skip_serializing_if = "Option::is_none")]
103	pub dispatch: Option<AutomationDispatch>,
104	/// Bound by the trusted service, independently of visibility ownership.
105	#[serde(default, skip_serializing_if = "Option::is_none")]
106	pub execution_scope: Option<String>,
107	}
108
109	/// Input for creating a new delayed trigger.
110	#[derive(Debug, Clone)]
111	pub struct CreateDelayedTriggerRequest {
112	/// Absolute fire time. Callers must resolve `delay_minutes` → `fire_at`
113	/// before calling this function.
114	pub fire_at: DateTime<Utc>,
115	/// Message to submit as a new task when the trigger fires.
116	pub message: String,
117	/// Optional workspace directory for the fired task.
118	pub workspace: Option<PathBuf>,
119	/// Session that owns controls and the task created when this trigger fires.
120	pub owner_session_id: Option<String>,
121	/// Optional parent trigger id for re-arm lineage tracking.
122	pub parent_trigger_id: Option<String>,
123	}
124
125	// ── End delayed-trigger types ──────────────────────────────────────────────
126
127	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
128	#[serde(rename_all = "snake_case")]
129	pub enum AutomationStatus {
130	Active,
131	Paused,
132	}
133
134	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
135	#[serde(rename_all = "snake_case")]
136	pub enum AutomationRunStatus {
137	Queued,
138	Running,
139	Completed,
140	Failed,
141	Canceled,
142	}
143
144	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
145	#[serde(rename_all = "snake_case")]
146	pub enum AutomationDeliveryMode {
147	#[default]
148	Task,
149	Watcher,
150	}
151
152	#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
153	pub struct AutomationRecord {
154	#[serde(default = "default_automation_schema_version")]
155	pub schema_version: u32,
156	pub id: String,
157	pub name: String,
158	pub prompt: String,
159	pub rrule: String,
160	#[serde(default)]
161	pub cwds: Vec<PathBuf>,
162	#[serde(default, skip_serializing_if = "Option::is_none")]
163	pub model: Option<String>,
164	/// Exact provider provenance for a pinned model; absent on legacy records.
165	#[serde(default, skip_serializing_if = "Option::is_none")]
166	pub model_provider: Option<String>,
167	#[serde(default, skip_serializing_if = "Option::is_none")]
168	pub model_provider_id: Option<String>,
169	#[serde(default, skip_serializing_if = "Option::is_none")]
170	pub mode: Option<String>,
171	#[serde(default, skip_serializing_if = "Option::is_none")]
172	pub allow_shell: Option<bool>,
173	#[serde(default, skip_serializing_if = "Option::is_none")]
174	pub trust_mode: Option<bool>,
175	#[serde(default, skip_serializing_if = "Option::is_none")]
176	pub auto_approve: Option<bool>,
177	#[serde(default, skip_serializing_if = "Option::is_none")]
178	pub delivery_mode: Option<AutomationDeliveryMode>,
179	pub status: AutomationStatus,
180	pub created_at: DateTime<Utc>,
181	pub updated_at: DateTime<Utc>,
182	#[serde(skip_serializing_if = "Option::is_none")]
183	pub next_run_at: Option<DateTime<Utc>>,
184	#[serde(skip_serializing_if = "Option::is_none")]
185	pub last_run_at: Option<DateTime<Utc>>,
186	/// Bound by the trusted service, independently of visibility ownership.
187	#[serde(default, skip_serializing_if = "Option::is_none")]
188	pub execution_scope: Option<String>,
189	}
190
191	impl AutomationRecord {
192	fn task_mode(&self) -> String {
193	self.mode
194	.as_deref()
195	.map(str::trim)
196	.filter(\|mode\| !mode.is_empty())
197	.unwrap_or(DEFAULT_AUTOMATION_MODE)
198	.to_string()
199	}
200
201	fn task_allow_shell(&self) -> bool {
202	self.allow_shell.unwrap_or(DEFAULT_AUTOMATION_ALLOW_SHELL)
203	}
204
205	fn task_trust_mode(&self) -> bool {
206	self.trust_mode.unwrap_or(DEFAULT_AUTOMATION_TRUST_MODE)
207	}
208
209	fn task_auto_approve(&self) -> bool {
210	self.auto_approve.unwrap_or(DEFAULT_AUTOMATION_AUTO_APPROVE)
211	}
212
213	fn delivery_mode(&self) -> AutomationDeliveryMode {
214	self.delivery_mode
215	.unwrap_or(DEFAULT_AUTOMATION_DELIVERY_MODE)
216	}
217	}
218
219	#[derive(Debug, Clone, Serialize, Deserialize)]
220	pub struct AutomationRunRecord {
221	#[serde(default = "default_run_schema_version")]
222	pub schema_version: u32,
223	pub id: String,
224	pub automation_id: String,
225	pub scheduled_for: DateTime<Utc>,
226	pub status: AutomationRunStatus,
227	pub created_at: DateTime<Utc>,
228	#[serde(skip_serializing_if = "Option::is_none")]
229	pub started_at: Option<DateTime<Utc>>,
230	#[serde(skip_serializing_if = "Option::is_none")]
231	pub ended_at: Option<DateTime<Utc>>,
232	#[serde(skip_serializing_if = "Option::is_none")]
233	pub task_id: Option<String>,
234	#[serde(skip_serializing_if = "Option::is_none")]
235	pub thread_id: Option<String>,
236	#[serde(skip_serializing_if = "Option::is_none")]
237	pub turn_id: Option<String>,
238	#[serde(skip_serializing_if = "Option::is_none")]
239	pub error: Option<String>,
240	#[serde(default, skip_serializing_if = "Option::is_none")]
241	pub dispatch: Option<AutomationDispatch>,
242	}
243
244	/// Immutable request and store bound to a durable occurrence before enqueue.
245	/// `accepted` records task promotion, not provider execution or completion.
246	#[derive(Debug, Clone, Serialize, Deserialize)]
247	pub struct AutomationDispatch {
248	#[serde(default, skip_serializing_if = "Option::is_none")]
249	execution_scope: Option<String>,
250	request: NewTaskRequest,
251	task_data_dir: PathBuf,
252	#[serde(default)]
253	accepted: bool,
254	#[serde(default)]
255	delivery_mode: AutomationDeliveryMode,
256	#[serde(default)]
257	suppress_report: bool,
258	#[serde(default, skip_serializing_if = "Option::is_none")]
259	schedule: Option<AdmittedSchedule>,
260	}
261
262	#[derive(Debug, Clone, Serialize, Deserialize)]
263	struct AdmittedSchedule {
264	updated_at: DateTime<Utc>,
265	rrule: String,
266	}
267
268	#[derive(Debug, Clone, Serialize, Deserialize)]
269	pub struct CreateAutomationRequest {
270	pub name: String,
271	pub prompt: String,
272	pub rrule: String,
273	#[serde(default)]
274	pub cwds: Vec<PathBuf>,
275	#[serde(default)]
276	pub model: Option<String>,
277	#[serde(default)]
278	pub model_provider: Option<String>,
279	#[serde(default)]
280	pub model_provider_id: Option<String>,
281	#[serde(default)]
282	pub mode: Option<String>,
283	#[serde(default)]
284	pub allow_shell: Option<bool>,
285	#[serde(default)]
286	pub trust_mode: Option<bool>,
287	#[serde(default)]
288	pub auto_approve: Option<bool>,
289	#[serde(default)]
290	pub delivery_mode: Option<AutomationDeliveryMode>,
291	#[serde(default)]
292	pub status: Option<AutomationStatus>,
293	}
294
295	#[derive(Debug, Clone, Serialize, Deserialize, Default)]
296	pub struct UpdateAutomationRequest {
297	pub name: Option<String>,
298	pub prompt: Option<String>,
299	pub rrule: Option<String>,
300	pub cwds: Option<Vec<PathBuf>>,
301	pub model: Option<String>,
302	pub model_provider: Option<String>,
303	pub model_provider_id: Option<String>,
304	pub mode: Option<String>,
305	pub allow_shell: Option<bool>,
306	pub trust_mode: Option<bool>,
307	pub auto_approve: Option<bool>,
308	pub delivery_mode: Option<AutomationDeliveryMode>,
309	pub status: Option<AutomationStatus>,
310	}
311
312	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
313	enum AutomationFrequency {
314	Hourly,
315	Weekly,
316	}
317
318	#[derive(Debug, Clone)]
319	pub enum AutomationSchedule {
320	Once {
321	at: DateTime<Utc>,
322	},
323	Hourly {
324	interval_hours: u32,
325	byday: Option<Vec<Weekday>>,
326	anchor_hour: Option<u32>,
327	anchor_minute: Option<u32>,
328	},
329	Weekly {
330	byday: Vec<Weekday>,
331	byhour: u32,
332	byminute: u32,
333	},
334	Cron {
335	expr: String,
336	},
337	}
338
339	impl AutomationSchedule {
340	pub fn parse_rrule(rrule: &str) -> Result<Self> {
341	let mut parts: BTreeMap<String, String> = BTreeMap::new();
342	for raw in rrule.split(';') {
343	let item = raw.trim();
344	if item.is_empty() {
345	continue;
346	}
347	let Some((k, v)) = item.split_once('=') else {
348	bail!("Invalid RRULE segment '{item}'");
349	};
350	parts.insert(k.trim().to_ascii_uppercase(), v.trim().to_string());
351	}
352
353	let freq = match parts
354	.get("FREQ")
355	.map(\|value\| value.trim().to_ascii_uppercase())
356	.as_deref()
357	{
358	Some("ONCE") => return parse_once_schedule(&parts),
359	Some("HOURLY") => AutomationFrequency::Hourly,
360	Some("WEEKLY") => AutomationFrequency::Weekly,
361	Some("CRON") => return parse_cron_schedule(&parts),
362	Some(other) => {
363	bail!("Unsupported RRULE FREQ '{other}'. Supported: ONCE, HOURLY, WEEKLY, and CRON")
364	}
365	None => bail!("RRULE must include FREQ"),
366	};
367
368	match freq {
369	AutomationFrequency::Hourly => {
370	for key in parts.keys() {
371	if key != "FREQ"
372	&& key != "INTERVAL"
373	&& key != "BYDAY"
374	&& key != "BYHOUR"
375	&& key != "BYMINUTE"
376	{
377	bail!(
378	"Unsupported RRULE field '{key}' for HOURLY. Allowed: FREQ,INTERVAL,BYDAY,BYHOUR,BYMINUTE"
379	);
380	}
381	}
382	let interval_hours = parts
383	.get("INTERVAL")
384	.map(\|v\| v.parse::<u32>())
385	.transpose()
386	.context("Failed to parse INTERVAL")?
387	.unwrap_or(1);
388	if interval_hours == 0 {
389	bail!("INTERVAL must be >= 1 for HOURLY schedules");
390	}
391	let byday = parts
392	.get("BYDAY")
393	.map(\|value\| parse_byday(&value.to_ascii_uppercase()))
394	.transpose()?;
395	let anchor_hour = parts
396	.get("BYHOUR")
397	.map(\|value\| value.parse::<u32>())
398	.transpose()
399	.context("Failed to parse BYHOUR")?;
400	let anchor_minute = parts
401	.get("BYMINUTE")
402	.map(\|value\| value.parse::<u32>())
403	.transpose()
404	.context("Failed to parse BYMINUTE")?;
405	if anchor_hour.is_some_and(\|hour\| hour > 23) {
406	bail!("BYHOUR must be between 0 and 23");
407	}
408	if anchor_minute.is_some_and(\|minute\| minute > 59) {
409	bail!("BYMINUTE must be between 0 and 59");
410	}
411	Ok(Self::Hourly {
412	interval_hours,
413	byday,
414	anchor_hour,
415	anchor_minute,
416	})
417	}
418	AutomationFrequency::Weekly => {
419	for key in parts.keys() {
420	if key != "FREQ" && key != "BYDAY" && key != "BYHOUR" && key != "BYMINUTE" {
421	bail!(
422	"Unsupported RRULE field '{key}' for WEEKLY. Allowed: FREQ,BYDAY,BYHOUR,BYMINUTE"
423	);
424	}
425	}
426	let byday_raw = parts
427	.get("BYDAY")
428	.ok_or_else(\|\| anyhow::anyhow!("WEEKLY schedules require BYDAY"))?;
429	let byday = parse_byday(&byday_raw.to_ascii_uppercase())?;
430	if byday.is_empty() {
431	bail!("BYDAY cannot be empty for WEEKLY schedules");
432	}
433	let byhour = parts
434	.get("BYHOUR")
435	.ok_or_else(\|\| anyhow::anyhow!("WEEKLY schedules require BYHOUR"))?
436	.parse::<u32>()
437	.context("Failed to parse BYHOUR")?;
438	let byminute = parts
439	.get("BYMINUTE")
440	.ok_or_else(\|\| anyhow::anyhow!("WEEKLY schedules require BYMINUTE"))?
441	.parse::<u32>()
442	.context("Failed to parse BYMINUTE")?;
443
444	if byhour > 23 {
445	bail!("BYHOUR must be between 0 and 23");
446	}
447	if byminute > 59 {
448	bail!("BYMINUTE must be between 0 and 59");
449	}
450
451	Ok(Self::Weekly {
452	byday,
453	byhour,
454	byminute,
455	})
456	}
457	}
458	}
459
460	pub(crate) fn next_after_with_anchor(
461	&self,
462	after: DateTime<Utc>,
463	anchor_reference: DateTime<Utc>,
464	) -> Result<DateTime<Utc>> {
465	self.next_after_in_timezone(after, anchor_reference, &Local)
466	}
467
468	fn next_after_in_timezone<Tz: TimeZone>(
469	&self,
470	after: DateTime<Utc>,
471	anchor_reference: DateTime<Utc>,
472	timezone: &Tz,
473	) -> Result<DateTime<Utc>> {
474	let local_after = after.with_timezone(timezone);
475	match self {
476	Self::Once { at } => {
477	if *at > after {
478	Ok(*at)
479	} else {
480	bail!(
481	"Once schedule has no future run after {}",
482	after.to_rfc3339()
483	)
484	}
485	}
486	Self::Hourly {
487	interval_hours,
488	byday,
489	anchor_hour,
490	anchor_minute,
491	} => {
492	if anchor_hour.is_some() \|\| anchor_minute.is_some() {
493	let local_anchor_reference = anchor_reference.with_timezone(timezone);
494	let hour = anchor_hour.unwrap_or(local_anchor_reference.hour());
495	let minute = anchor_minute.unwrap_or(0);
496	let anchor_naive = local_anchor_reference
497	.date_naive()
498	.and_hms_opt(hour, minute, 0)
499	.ok_or_else(\|\| anyhow::anyhow!("Unable to construct HOURLY anchor"))?;
500	let interval_seconds = i64::from(interval_hours) 60 * 60;
501	let elapsed_seconds = local_after
502	.naive_local()
503	.signed_duration_since(anchor_naive)
504	.num_seconds();
505	let mut steps = if elapsed_seconds < 0 {
506	0
507	} else {
508	elapsed_seconds / interval_seconds + 1
509	};
510
511	for _ in 0..MAX_HOURLY_SEARCH_STEPS {
512	let hours = i64::from(*interval_hours)
513	.checked_mul(steps)
514	.ok_or_else(\|\| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
515	let delta = Duration::try_hours(hours)
516	.ok_or_else(\|\| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
517	let candidate_naive = anchor_naive
518	.checked_add_signed(delta)
519	.ok_or_else(\|\| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
520
521	if byday
522	.as_ref()
523	.is_none_or(\|days\| days.contains(&candidate_naive.weekday()))
524	&& let Some(candidate) =
525	resolve_local_datetime(timezone, candidate_naive)
526	{
527	let candidate = candidate.with_timezone(&Utc);
528	if candidate > after {
529	return Ok(candidate);
530	}
531	}
532
533	steps = steps
534	.checked_add(1)
535	.ok_or_else(\|\| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
536	}
537	bail!("Unable to compute next anchored HOURLY run");
538	}
539
540	let after_second = local_after.second();
541	let after_nanosecond = local_after.nanosecond();
542	let mut candidate = local_after + Duration::hours(i64::from(*interval_hours))
543	- Duration::seconds(i64::from(after_second))
544	- Duration::nanoseconds(i64::from(after_nanosecond));
545
546	if let Some(days) = byday {
547	for _ in 0..(24 * 21) {
548	if days.contains(&candidate.weekday()) {
549	return Ok(candidate.with_timezone(&Utc));
550	}
551	candidate += Duration::hours(i64::from(*interval_hours));
552	}
553	bail!("Unable to compute next HOURLY run for BYDAY filter");
554	}
555
556	Ok(candidate.with_timezone(&Utc))
557	}
558	Self::Weekly {
559	byday,
560	byhour,
561	byminute,
562	} => {
563	for day_offset in 0..15 {
564	let date = local_after.date_naive() + Duration::days(i64::from(day_offset));
565	if !byday.contains(&date.weekday()) {
566	continue;
567	}
568	let Some(candidate_naive) = date.and_hms_opt(byhour, byminute, 0) else {
569	continue;
570	};
571	if let Some(candidate) = resolve_local_datetime(timezone, candidate_naive)
572	&& candidate.with_timezone(&Utc) > after
573	{
574	return Ok(candidate.with_timezone(&Utc));
575	}
576	}
577	bail!("Unable to compute next WEEKLY run");
578	}
579	Self::Cron { expr } => {
580	let cron = ParsedCronExpr::parse(expr)?;
581	let mut candidate_naive = local_after
582	.naive_local()
583	.with_second(0)
584	.and_then(\|dt\| dt.with_nanosecond(0))
585	.ok_or_else(\|\| anyhow::anyhow!("Unable to round CRON search start"))?
586	.checked_add_signed(Duration::minutes(1))
587	.ok_or_else(\|\| anyhow::anyhow!("CRON schedule exceeded its range"))?;
588
589	for _ in 0..MAX_CRON_SEARCH_MINUTES {
590	if cron.matches(candidate_naive)
591	&& let Some(candidate) = resolve_local_datetime(timezone, candidate_naive)
592	{
593	let candidate = candidate.with_timezone(&Utc);
594	if candidate > after {
595	return Ok(candidate);
596	}
597	}
598	candidate_naive = candidate_naive
599	.checked_add_signed(Duration::minutes(1))
600	.ok_or_else(\|\| anyhow::anyhow!("CRON schedule exceeded its range"))?;
601	}
602	bail!("Unable to compute next CRON run within 5 years");
603	}
604	}
605	}
606
607	fn next_after_slot(
608	&self,
609	slot: DateTime<Utc>,
610	anchor_reference: DateTime<Utc>,
611	) -> Result<Option<DateTime<Utc>>> {
612	match self {
613	Self::Once { .. } => Ok(None),
614	_ => self
615	.next_after_with_anchor(slot, anchor_reference)
616	.map(Some),
617	}
618	}
619
620	/// First slot after `slot` that is still in the future at `now`.
621	///
622	/// Missed slots coalesce: downtime, a paused window, or an in-flight
623	/// occurrence earn one receipt for the oldest owed slot, then the
624	/// schedule resumes on its own grid instead of replaying one stale slot
625	/// per tick. Calendar-anchored schedules (anchored HOURLY, WEEKLY, CRON)
626	/// live on a fixed wall-clock grid, so the first slot after `now` is
627	/// exactly the slot plain chaining would converge to; computing from
628	/// `now` directly skips the whole backlog in one step. Unanchored HOURLY
629	/// is a relative cadence with no calendar grid — hop along its
630	/// established `slot + k * interval` chain so a late recovery does not
631	/// re-phase the schedule to the recovery instant.
632	fn next_unskipped_slot(
633	&self,
634	slot: DateTime<Utc>,
635	now: DateTime<Utc>,
636	anchor_reference: DateTime<Utc>,
637	) -> Result<Option<DateTime<Utc>>> {
638	if let Self::Hourly {
639	interval_hours,
640	anchor_hour: None,
641	anchor_minute: None,
642	..
643	} = self
644	{
645	let first = self.next_after_with_anchor(slot, anchor_reference)?;
646	if first > now {
647	return Ok(Some(first));
648	}
649	// Jump whole intervals on the established UTC grid, then reuse
650	// the schedule's weekday filter for the next eligible slot.
651	// Minute normalization happens in the first advance above.
652	let interval_seconds = i64::from(interval_hours) 60 * 60;
653	let elapsed = (now - first).num_seconds();
654	let delta = Duration::seconds(elapsed / interval_seconds * interval_seconds);
655	let previous = first
656	.checked_add_signed(delta)
657	.context("HOURLY catch-up exceeded its range")?;
658	self.next_after_slot(previous, anchor_reference)
659	} else {
660	self.next_after_slot(slot.max(now), anchor_reference)
661	}
662	}
663	}
664
665	/// Resolve one calendar-local schedule slot.
666	///
667	/// Nonexistent wall times in a forward clock change are skipped rather than
668	/// shifted to a different clock time. Ambiguous wall times in a backward clock
669	/// change use the first occurrence only, preventing a recurring automation from
670	/// running twice for one calendar slot.
671	fn resolve_local_datetime<Tz: TimeZone>(
672	timezone: &Tz,
673	naive: NaiveDateTime,
674	) -> Option<DateTime<Tz>> {
675	timezone.from_local_datetime(&naive).earliest()
676	}
677
678	fn parse_byday(value: &str) -> Result<Vec<Weekday>> {
679	let mut days = Vec::new();
680	for token in value.split(',') {
681	let day = match token.trim().to_ascii_uppercase().as_str() {
682	"MO" => Weekday::Mon,
683	"TU" => Weekday::Tue,
684	"WE" => Weekday::Wed,
685	"TH" => Weekday::Thu,
686	"FR" => Weekday::Fri,
687	"SA" => Weekday::Sat,
688	"SU" => Weekday::Sun,
689	other => bail!("Invalid BYDAY value '{other}'"),
690	};
691	if !days.contains(&day) {
692	days.push(day);
693	}
694	}
695	Ok(days)
696	}
697
698	fn parse_once_schedule(parts: &BTreeMap<String, String>) -> Result<AutomationSchedule> {
699	for key in parts.keys() {
700	if key != "FREQ" && key != "AT" {
701	bail!("Unsupported RRULE field '{key}' for ONCE. Allowed: FREQ,AT");
702	}
703	}
704	let raw_at = parts
705	.get("AT")
706	.ok_or_else(\|\| anyhow::anyhow!("ONCE schedules require AT"))?;
707	let at = parse_once_at(raw_at)?;
708	Ok(AutomationSchedule::Once { at })
709	}
710
711	fn parse_cron_schedule(parts: &BTreeMap<String, String>) -> Result<AutomationSchedule> {
712	for key in parts.keys() {
713	if key != "FREQ" && key != "EXPR" {
714	bail!("Unsupported RRULE field '{key}' for CRON. Allowed: FREQ,EXPR");
715	}
716	}
717	let expr = parts
718	.get("EXPR")
719	.map(\|value\| value.trim().to_string())
720	.filter(\|value\| !value.is_empty())
721	.ok_or_else(\|\| anyhow::anyhow!("CRON schedules require EXPR"))?;
722	ParsedCronExpr::parse(&expr)?;
723	Ok(AutomationSchedule::Cron { expr })
724	}
725
726	fn parse_once_at(raw: &str) -> Result<DateTime<Utc>> {
727	let trimmed = raw.trim();
728	if let Ok(at) = DateTime::parse_from_rfc3339(trimmed) {
729	return Ok(at.with_timezone(&Utc));
730	}
731	for format in ["%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"] {
732	if let Ok(naive) = NaiveDateTime::parse_from_str(trimmed, format) {
733	return resolve_local_datetime(&Local, naive)
734	.map(\|value\| value.with_timezone(&Utc))
735	.ok_or_else(\|\| anyhow::anyhow!("ONCE local time does not exist: {trimmed}"));
736	}
737	}
738	bail!("Failed to parse ONCE AT '{trimmed}'. Use local YYYY-MM-DDTHH:MM[:SS] or RFC3339")
739	}
740
741	#[derive(Debug, Clone)]
742	struct ParsedCronExpr {
743	minute: CronField,
744	hour: CronField,
745	day_of_month: CronField,
746	month: CronField,
747	day_of_week: CronField,
748	}
749
750	impl ParsedCronExpr {
751	fn parse(expr: &str) -> Result<Self> {
752	let fields: Vec<&str> = expr.split_whitespace().collect();
753	if fields.len() != 5 {
754	bail!(
755	"CRON EXPR must have exactly 5 fields: minute hour day-of-month month day-of-week"
756	);
757	}
758	let parsed = Self {
759	minute: CronField::parse(fields[0], 0, 59, CronNameMap::none(), "minute")?,
760	hour: CronField::parse(fields[1], 0, 23, CronNameMap::none(), "hour")?,
761	day_of_month: CronField::parse(fields[2], 1, 31, CronNameMap::none(), "day-of-month")?,
762	month: CronField::parse(fields[3], 1, 12, CronNameMap::month(), "month")?,
763	day_of_week: CronField::parse(fields[4], 0, 7, CronNameMap::weekday(), "day-of-week")?
764	.normalized_day_of_week(),
765	};
766	parsed.validate_date_space()?;
767	Ok(parsed)
768	}
769
770	fn matches(&self, candidate: NaiveDateTime) -> bool {
771	if !self.minute.contains(candidate.minute())
772	\|\| !self.hour.contains(candidate.hour())
773	\|\| !self.month.contains(candidate.month())
774	{
775	return false;
776	}
777
778	let day_of_month = self.day_of_month.contains(candidate.day());
779	let weekday = self
780	.day_of_week
781	.contains(weekday_to_cron(candidate.weekday()));
782	if self.day_of_month.is_wildcard && self.day_of_week.is_wildcard {
783	true
784	} else if self.day_of_month.is_wildcard {
785	weekday
786	} else if self.day_of_week.is_wildcard {
787	day_of_month
788	} else {
789	day_of_month \|\| weekday
790	}
791	}
792
793	fn validate_date_space(&self) -> Result<()> {
794	if self.day_of_month.is_wildcard {
795	return Ok(());
796	}
797	let months = self.month.values();
798	let days = self.day_of_month.values();
799	let valid = months.iter().copied().any(\|month\| {
800	let common = days_in_month(2025, month);
801	let leap = days_in_month(2024, month);
802	days.iter().copied().any(\|day\| day <= common \|\| day <= leap)
803	});
804	if valid {
805	Ok(())
806	} else {
807	bail!("CRON EXPR day-of-month/month combination can never occur")
808	}
809	}
810	}
811
812	#[derive(Debug, Clone)]
813	struct CronField {
814	values: Vec<u32>,
815	is_wildcard: bool,
816	}
817
818	impl CronField {
819	fn parse(raw: &str, min: u32, max: u32, names: CronNameMap, field_name: &str) -> Result<Self> {
820	let trimmed = raw.trim();
821	if trimmed.is_empty() {
822	bail!("CRON {field_name} field must not be empty");
823	}
824	let mut values = Vec::new();
825	let is_wildcard = trimmed == "*";
826	for part in trimmed.split(',') {
827	let part = part.trim();
828	if part.is_empty() {
829	bail!("CRON {field_name} field contains an empty list item");
830	}
831	let (base, step) = if let Some((base, step)) = part.split_once('/') {
832	let step = step
833	.trim()
834	.parse::<u32>()
835	.with_context(\|\| format!("Failed to parse CRON {field_name} step"))?;
836	if step == 0 {
837	bail!("CRON {field_name} step must be >= 1");
838	}
839	(base.trim(), step)
840	} else {
841	(part, 1)
842	};
843
844	let range = if base == "*" {
845	(min, max)
846	} else if let Some((start, end)) = base.split_once('-') {
847	let start = parse_cron_atom(start.trim(), min, max, names, field_name)?;
848	let end = parse_cron_atom(end.trim(), min, max, names, field_name)?;
849	if start > end {
850	bail!("CRON {field_name} range start must be <= end");
851	}
852	(start, end)
853	} else {
854	let start = parse_cron_atom(base, min, max, names, field_name)?;
855	if part.contains('/') {
856	(start, max)
857	} else {
858	(start, start)
859	}
860	};
861
862	let mut current = range.0;
863	while current <= range.1 {
864	if !values.contains(&current) {
865	values.push(current);
866	}
867	let Some(next) = current.checked_add(step) else {
868	break;
869	};
870	if next <= current {
871	break;
872	}
873	current = next;
874	}
875	}
876	values.sort_unstable();
877	Ok(Self {
878	values,
879	is_wildcard,
880	})
881	}
882
883	fn normalized_day_of_week(mut self) -> Self {
884	for value in &mut self.values {
885	if *value == 7 {
886	*value = 0;
887	}
888	}
889	self.values.sort_unstable();
890	self.values.dedup();
891	self
892	}
893
894	fn contains(&self, value: u32) -> bool {
895	self.values.binary_search(&value).is_ok()
896	}
897
898	fn values(&self) -> &[u32] {
899	&self.values
900	}
901	}
902
903	#[derive(Debug, Clone, Copy)]
904	struct CronNameMap(&'static [(&'static str, u32)]);
905
906	impl CronNameMap {
907	const fn none() -> Self {
908	Self(&[])
909	}
910
911	const fn month() -> Self {
912	Self(&[
913	("JAN", 1),
914	("FEB", 2),
915	("MAR", 3),
916	("APR", 4),
917	("MAY", 5),
918	("JUN", 6),
919	("JUL", 7),
920	("AUG", 8),
921	("SEP", 9),
922	("OCT", 10),
923	("NOV", 11),
924	("DEC", 12),
925	])
926	}
927
928	const fn weekday() -> Self {
929	Self(&[
930	("SUN", 0),
931	("MON", 1),
932	("TUE", 2),
933	("WED", 3),
934	("THU", 4),
935	("FRI", 5),
936	("SAT", 6),
937	])
938	}
939
940	fn lookup(self, token: &str) -> Option<u32> {
941	let needle = token.trim().to_ascii_uppercase();
942	self.0
943	.iter()
944	.find_map(\|(name, value)\| (name == needle).then_some(value))
945	}
946	}
947
948	fn parse_cron_atom(
949	raw: &str,
950	min: u32,
951	max: u32,
952	names: CronNameMap,
953	field_name: &str,
954	) -> Result<u32> {
955	let value = names
956	.lookup(raw)
957	.or_else(\|\| raw.parse::<u32>().ok())
958	.ok_or_else(\|\| anyhow::anyhow!("Invalid CRON {field_name} value '{raw}'"))?;
959	if !(min..=max).contains(&value) {
960	bail!("CRON {field_name} value {value} is out of range {min}-{max}");
961	}
962	Ok(value)
963	}
964
965	fn weekday_to_cron(day: Weekday) -> u32 {
966	match day {
967	Weekday::Sun => 0,
968	Weekday::Mon => 1,
969	Weekday::Tue => 2,
970	Weekday::Wed => 3,
971	Weekday::Thu => 4,
972	Weekday::Fri => 5,
973	Weekday::Sat => 6,
974	}
975	}
976
977	fn days_in_month(year: i32, month: u32) -> u32 {
978	match month {
979	1 \| 3 \| 5 \| 7 \| 8 \| 10 \| 12 => 31,
980	4 \| 6 \| 9 \| 11 => 30,
981	2 => {
982	let leap = (year % 4 == 0 && year % 100 != 0) \|\| year % 400 == 0;
983	if leap { 29 } else { 28 }
984	}
985	_ => 0,
986	}
987	}
988
989	#[derive(Debug, Clone)]
990	pub struct AutomationManager {
991	execution_scope: Option<String>,
992	automations_dir: PathBuf,
993	runs_dir: PathBuf,
994	triggers_dir: PathBuf,
995	}
996
997	impl AutomationManager {
998	fn open_lock(&self, name: &str) -> Result<fd_lock::RwLock<fs::File>> {
999	let path = self
1000	.automations_dir
1001	.parent()
1002	.context("automation root")?
1003	.join(name);
1004	let mut options = fs::OpenOptions::new();
1005	options.create(true).truncate(false).read(true).write(true);
1006	#[cfg(unix)]
1007	{
1008	use std::os::unix::fs::OpenOptionsExt as _;
1009	options
1010	.mode(0o600)
1011	.custom_flags(libc::O_NOFOLLOW \| libc::O_CLOEXEC);
1012	}
1013	#[cfg(windows)]
1014	{
1015	use std::os::windows::fs::OpenOptionsExt as _;
1016	options.custom_flags(0x0020_0000); // FILE_FLAG_OPEN_REPARSE_POINT
1017	}
1018	let file = options
1019	.open(&path)
1020	.with_context(\|\| format!("open {}", path.display()))?;
1021	let metadata = file.metadata()?;
1022	if !metadata.is_file() {
1023	bail!("Automation lock must be a regular file");
1024	}
1025	#[cfg(unix)]
1026	{
1027	use std::os::unix::fs::MetadataExt as _;
1028	if metadata.nlink() != 1 {
1029	bail!("Automation lock must not have hard links");
1030	}
1031	}
1032	#[cfg(windows)]
1033	{
1034	use std::os::windows::fs::MetadataExt as _;
1035	if metadata.file_attributes() & 0x400 != 0 {
1036	bail!("Automation lock must not be a reparse point");
1037	}
1038	}
1039	Ok(fd_lock::RwLock::new(file))
1040	}
1041
1042	fn with_transaction<T>(&self, operation: impl FnOnce() -> Result<T>) -> Result<T> {
1043	let mut lock = self.open_lock("state.lock")?;
1044	let _guard = lock.write().context("lock automation state")?;
1045	operation()
1046	}
1047
1048	/// Short read/modify/write transaction shared with scheduler admission.
1049	/// Returning None leaves an absent record absent; it does not delete one.
1050	pub(crate) fn edit_automation(
1051	&self,
1052	id: &str,
1053	edit: impl FnOnce(Option<AutomationRecord>) -> Result<Option<AutomationRecord>>,
1054	) -> Result<Option<AutomationRecord>> {
1055	self.with_transaction(\|\| {
1056	let current = if self.automation_path(id)?.try_exists()? {
1057	Some(self.get_automation(id)?)
1058	} else {
1059	None
1060	};
1061	let edited = edit(current)?;
1062	if let Some(record) = &edited {
1063	if record.id != id {
1064	bail!("Automation transaction cannot replace its identity");
1065	}
1066	self.save_automation_unlocked(record)?;
1067	}
1068	Ok(edited)
1069	})
1070	}
1071
1072	pub fn open(root: PathBuf) -> Result<Self> {
1073	let automations_dir = root.join("automations");
1074	let runs_dir = root.join("runs");
1075	let triggers_dir = root.join("triggers");
1076	fs::create_dir_all(&automations_dir)
1077	.with_context(\|\| format!("Failed to create {}", automations_dir.display()))?;
1078	fs::create_dir_all(&runs_dir)
1079	.with_context(\|\| format!("Failed to create {}", runs_dir.display()))?;
1080	fs::create_dir_all(&triggers_dir)
1081	.with_context(\|\| format!("Failed to create {}", triggers_dir.display()))?;
1082	Ok(Self {
1083	execution_scope: None,
1084	automations_dir,
1085	runs_dir,
1086	triggers_dir,
1087	})
1088	}
1089
1090	#[cfg(test)]
1091	pub(crate) fn open_for_test(root: PathBuf) -> Result<Self> {
1092	let mut manager = Self::open(root)?;
1093	manager.execution_scope = Some(crate::task_manager::test_execution_scope("test"));
1094	Ok(manager)
1095	}
1096
1097	pub(crate) fn bind_task_manager(
1098	&mut self,
1099	tasks: &crate::task_manager::TaskManager,
1100	) -> Result<()> {
1101	if self
1102	.execution_scope
1103	.as_deref()
1104	.is_some_and(\|scope\| scope != tasks.execution_scope())
1105	{
1106	bail!("Automation service belongs to another Runtime scope");
1107	}
1108	self.execution_scope = Some(tasks.execution_scope().to_string());
1109	Ok(())
1110	}
1111
1112	pub(crate) fn execution_scope(&self) -> Option<&str> {
1113	self.execution_scope.as_deref()
1114	}
1115
1116	fn eligible_scope(&self, scope: Option<&str>) -> bool {
1117	scope.is_some() && scope == self.execution_scope()
1118	}
1119
1120	/// Explicit control may bind an unbound definition; saved admissions never
1121	/// read this field back from the definition during recovery.
1122	fn adopt_for_run(&self, automation: &mut AutomationRecord) -> Result<()> {
1123	let scope = self
1124	.execution_scope()
1125	.context("Automation execution ownership is unverified")?;
1126	if let Some(bound) = &automation.execution_scope {
1127	if bound != scope {
1128	bail!("Automation belongs to another Runtime execution scope");
1129	}
1130	} else {
1131	automation.execution_scope = Some(scope.to_string());
1132	automation.schema_version = CURRENT_AUTOMATION_SCHEMA_VERSION;
1133	automation.updated_at = Utc::now();
1134	if automation.status == AutomationStatus::Active {
1135	let schedule = AutomationSchedule::parse_rrule(&automation.rrule)?;
1136	automation.next_run_at =
1137	match schedule.next_after_with_anchor(Utc::now(), automation.created_at) {
1138	Ok(next) => Some(next),
1139	Err(_) if matches!(schedule, AutomationSchedule::Once { .. }) => {
1140	automation.status = AutomationStatus::Paused;
1141	None
1142	}
1143	Err(error) => return Err(error),
1144	};
1145	}
1146	self.save_automation_unlocked(automation)?;
1147	}
1148	Ok(())
1149	}
1150
1151	pub fn default_location() -> Result<Self> {
1152	Self::open(default_automations_dir())
1153	}
1154
1155	fn automation_path(&self, id: &str) -> Result<PathBuf> {
1156	ensure_safe_storage_id("automation id", id)?;
1157	Ok(self.automations_dir.join(format!("{id}.json")))
1158	}
1159
1160	fn runs_dir_for(&self, automation_id: &str) -> Result<PathBuf> {
1161	ensure_safe_storage_id("automation id", automation_id)?;
1162	Ok(self.runs_dir.join(automation_id))
1163	}
1164
1165	fn trigger_path(&self, trigger_id: &str) -> Result<PathBuf> {
1166	ensure_safe_storage_id("trigger id", trigger_id)?;
1167	Ok(self.triggers_dir.join(format!("{trigger_id}.json")))
1168	}
1169
1170	/// Current run file name: `{sortable-created-at}-{run_id}.json`. The
1171	/// fixed-width timestamp prefix makes directory listings sort
1172	/// chronologically without reading file contents (see [`Self::list_runs`]).
1173	fn run_path(&self, run: &AutomationRunRecord) -> Result<PathBuf> {
1174	ensure_safe_storage_id("run id", &run.id)?;
1175	Ok(self.runs_dir_for(&run.automation_id)?.join(format!(
1176	"{}-{}.json",
1177	run_file_stamp(run.created_at),
1178	run.id
1179	)))
1180	}
1181
1182	/// Pre-sortable-name run file: `{run_id}.json` (run ids are UUIDs, so
1183	/// these carry no ordering hint and must be read to learn `created_at`).
1184	fn legacy_run_path(&self, automation_id: &str, run_id: &str) -> Result<PathBuf> {
1185	ensure_safe_storage_id("run id", run_id)?;
1186	Ok(self
1187	.runs_dir_for(automation_id)?
1188	.join(format!("{run_id}.json")))
1189	}
1190
1191	pub fn create_automation(&self, req: CreateAutomationRequest) -> Result<AutomationRecord> {
1192	validate_name_and_prompt(&req.name, &req.prompt)?;
1193	let schedule = AutomationSchedule::parse_rrule(&req.rrule)?;
1194	let now = Utc::now();
1195	let status = req.status.unwrap_or(AutomationStatus::Active);
1196	let next_run_at = if matches!(status, AutomationStatus::Active) {
1197	Some(schedule.next_after_with_anchor(now, now)?)
1198	} else {
1199	None
1200	};
1201
1202	let record = AutomationRecord {
1203	schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION,
1204	execution_scope: self.execution_scope.clone(),
1205	id: Uuid::new_v4().to_string(),
1206	name: req.name.trim().to_string(),
1207	prompt: req.prompt.trim().to_string(),
1208	rrule: req.rrule.trim().to_ascii_uppercase(),
1209	cwds: req.cwds,
1210	model: normalize_optional_string(req.model),
1211	model_provider: normalize_optional_string(req.model_provider),
1212	model_provider_id: normalize_optional_string(req.model_provider_id),
1213	mode: normalize_optional_string(req.mode),
1214	allow_shell: req.allow_shell,
1215	trust_mode: req.trust_mode,
1216	auto_approve: req.auto_approve,
1217	delivery_mode: req.delivery_mode,
1218	status,
1219	created_at: now,
1220	updated_at: now,
1221	next_run_at,
1222	last_run_at: None,
1223	};
1224
1225	self.save_automation(&record)?;
1226	Ok(record)
1227	}
1228
1229	pub fn get_automation(&self, id: &str) -> Result<AutomationRecord> {
1230	let path = self.automation_path(id)?;
1231	read_automation_file(&path)
1232	}
1233
1234	pub fn save_automation(&self, record: &AutomationRecord) -> Result<()> {
1235	self.with_transaction(\|\| self.save_automation_unlocked(record))
1236	}
1237
1238	fn save_automation_unlocked(&self, record: &AutomationRecord) -> Result<()> {
1239	if record.model_provider.is_some() \|\| record.model_provider_id.is_some() {
1240	if record
1241	.model
1242	.as_deref()
1243	.is_none_or(\|model\| model.trim().is_empty())
1244	{
1245	bail!("A pinned automation provider requires an explicit model");
1246	}
1247	let mut record = record.clone();
1248	record.schema_version = record.schema_version.max(CURRENT_AUTOMATION_SCHEMA_VERSION);
1249	return write_json_atomic(&self.automation_path(&record.id)?, &record);
1250	}
1251	write_json_atomic(&self.automation_path(&record.id)?, record)
1252	}
1253
1254	pub fn list_automations(&self) -> Result<Vec<AutomationRecord>> {
1255	let mut out = Vec::new();
1256	for entry in fs::read_dir(&self.automations_dir)
1257	.with_context(\|\| format!("Failed to read {}", self.automations_dir.display()))?
1258	{
1259	let entry = entry?;
1260	let path = entry.path();
1261	if path.extension().is_none_or(\|ext\| ext != "json") {
1262	continue;
1263	}
1264	let record = read_automation_file(&path)?;
1265	out.push(record);
1266	}
1267	out.sort_by_key(\|r\| std::cmp::Reverse(r.updated_at));
1268	Ok(out)
1269	}
1270
1271	pub fn update_automation(
1272	&self,
1273	id: &str,
1274	req: UpdateAutomationRequest,
1275	) -> Result<AutomationRecord> {
1276	self.with_transaction(\|\| self.update_automation_unlocked(id, req))
1277	}
1278
1279	fn update_automation_unlocked(
1280	&self,
1281	id: &str,
1282	req: UpdateAutomationRequest,
1283	) -> Result<AutomationRecord> {
1284	let mut existing = self.get_automation(id)?;
1285	let adopting = existing.execution_scope.is_none()
1286	&& self.execution_scope.is_some()
1287	&& req.status != Some(AutomationStatus::Paused);
1288	if adopting {
1289	existing.execution_scope = self.execution_scope.clone();
1290	}
1291	let schedule_changed = adopting \|\| req.rrule.is_some() \|\| req.status.is_some();
1292
1293	if let Some(name) = req.name {
1294	if name.trim().is_empty() {
1295	bail!("Automation name cannot be empty");
1296	}
1297	existing.name = name.trim().to_string();
1298	}
1299	if let Some(prompt) = req.prompt {
1300	if prompt.trim().is_empty() {
1301	bail!("Automation prompt cannot be empty");
1302	}
1303	existing.prompt = prompt.trim().to_string();
1304	}
1305	if let Some(rrule) = req.rrule {
1306	let normalized = rrule.trim().to_ascii_uppercase();
1307	AutomationSchedule::parse_rrule(&normalized)?;
1308	existing.rrule = normalized;
1309	}
1310	if let Some(cwds) = req.cwds {
1311	existing.cwds = cwds;
1312	}
1313	if let Some(model) = req.model {
1314	existing.model = normalize_optional_string(Some(model));
1315	}
1316	if let Some(provider) = req.model_provider {
1317	existing.model_provider = normalize_optional_string(Some(provider));
1318	}
1319	if let Some(provider_id) = req.model_provider_id {
1320	existing.model_provider_id = normalize_optional_string(Some(provider_id));
1321	}
1322	if let Some(mode) = req.mode {
1323	existing.mode = normalize_optional_string(Some(mode));
1324	}
1325	if let Some(allow_shell) = req.allow_shell {
1326	existing.allow_shell = Some(allow_shell);
1327	}
1328	if let Some(trust_mode) = req.trust_mode {
1329	existing.trust_mode = Some(trust_mode);
1330	}
1331	if let Some(auto_approve) = req.auto_approve {
1332	existing.auto_approve = Some(auto_approve);
1333	}
1334	if let Some(delivery_mode) = req.delivery_mode {
1335	existing.delivery_mode = Some(delivery_mode);
1336	}
1337	if let Some(status) = req.status {
1338	existing.status = status;
1339	}
1340	// Evaluate the final status once: editing a schedule and pausing it is
1341	// one atomic update, and must not first schedule an active run.
1342	if schedule_changed {
1343	if matches!(existing.status, AutomationStatus::Paused) {
1344	existing.next_run_at = None;
1345	} else {
1346	let schedule = AutomationSchedule::parse_rrule(&existing.rrule)?;
1347	existing.next_run_at =
1348	Some(schedule.next_after_with_anchor(Utc::now(), existing.created_at)?);
1349	}
1350	}
1351
1352	if existing.execution_scope.is_some()
1353	\|\| existing.model_provider.is_some()
1354	\|\| existing.model_provider_id.is_some()
1355	{
1356	existing.schema_version = CURRENT_AUTOMATION_SCHEMA_VERSION;
1357	}
1358
1359	existing.updated_at = Utc::now();
1360	self.save_automation_unlocked(&existing)?;
1361	Ok(existing)
1362	}
1363
1364	pub fn pause_automation(&self, id: &str) -> Result<AutomationRecord> {
1365	self.update_automation(
1366	id,
1367	UpdateAutomationRequest {
1368	status: Some(AutomationStatus::Paused),
1369	..UpdateAutomationRequest::default()
1370	},
1371	)
1372	}
1373
1374	pub fn resume_automation(&self, id: &str) -> Result<AutomationRecord> {
1375	self.update_automation(
1376	id,
1377	UpdateAutomationRequest {
1378	status: Some(AutomationStatus::Active),
1379	..UpdateAutomationRequest::default()
1380	},
1381	)
1382	}
1383
1384	pub fn delete_automation(&self, id: &str) -> Result<AutomationRecord> {
1385	self.with_transaction(\|\| {
1386	let existing = self.get_automation(id)?;
1387	let path = self.automation_path(id)?;
1388	fs::remove_file(&path)
1389	.with_context(\|\| format!("Failed to delete automation {}", path.display()))?;
1390	// A claimed occurrence has already crossed the admission boundary.
1391	// Keep its binding through deletion so recovery cannot lose or repeat it.
1392	for run in self.list_runs_with_visibility(id, None, true)? {
1393	if !matches!(
1394	run.status,
1395	AutomationRunStatus::Queued \| AutomationRunStatus::Running
1396	) {
1397	self.delete_run(&run)?;
1398	}
1399	}
1400	let runs_dir = self.runs_dir_for(id)?;
1401	if runs_dir.try_exists()? && fs::read_dir(&runs_dir)?.next().is_none() {
1402	fs::remove_dir(&runs_dir).with_context(\|\| {
1403	format!(
1404	"Failed to remove empty run directory {}",
1405	runs_dir.display()
1406	)
1407	})?;
1408	}
1409	Ok(existing)
1410	})
1411	}
1412
1413	pub fn list_runs(
1414	&self,
1415	automation_id: &str,
1416	limit: Option<usize>,
1417	) -> Result<Vec<AutomationRunRecord>> {
1418	self.list_runs_with_visibility(automation_id, limit, false)
1419	}
1420
1421	fn list_runs_with_visibility(
1422	&self,
1423	automation_id: &str,
1424	limit: Option<usize>,
1425	include_suppressed: bool,
1426	) -> Result<Vec<AutomationRunRecord>> {
1427	let dir = self.runs_dir_for(automation_id)?;
1428	if !dir.exists() {
1429	return Ok(Vec::new());
1430	}
1431
1432	// Split the listing into sortable-name files (newest-first by file
1433	// name alone, so reads stop after the newest `limit`) and legacy
1434	// `{uuid}.json` files, which must all be read to learn `created_at`.
1435	let mut sortable = Vec::new();
1436	let mut legacy = Vec::new();
1437	for entry in
1438	fs::read_dir(&dir).with_context(\|\| format!("Failed to read {}", dir.display()))?
1439	{
1440	let entry = entry?;
1441	let path = entry.path();
1442	if path.extension().is_none_or(\|ext\| ext != "json") {
1443	continue;
1444	}
1445	if path
1446	.file_stem()
1447	.and_then(\|stem\| stem.to_str())
1448	.is_some_and(has_sortable_run_stem)
1449	{
1450	sortable.push(path);
1451	} else {
1452	legacy.push(path);
1453	}
1454	}
1455
1456	// A sortable receipt supersedes its legacy copy even when hidden or
1457	// older than the requested window. Fence identity before visibility.
1458	let sortable_ids: std::collections::BTreeSet<_> = sortable
1459	.iter()
1460	.filter_map(\|path\| path.file_stem()?.to_str()?.get(RUN_STAMP_LEN + 1..))
1461	.collect();
1462	legacy.retain(\|path\| {
1463	!path
1464	.file_stem()
1465	.and_then(\|stem\| stem.to_str())
1466	.is_some_and(\|id\| sortable_ids.contains(id))
1467	});
1468	sortable.sort_by(\|a, b\| b.file_name().cmp(&a.file_name()));
1469	let visible = \|run: &AutomationRunRecord\| {
1470	include_suppressed
1471	\|\| !run
1472	.dispatch
1473	.as_ref()
1474	.is_some_and(\|dispatch\| dispatch.suppress_report)
1475	};
1476	let mut out = Vec::new();
1477	for path in sortable {
1478	if limit.is_some_and(\|limit\| out.len() >= limit) {
1479	break;
1480	}
1481	let run = read_run_file(&path)?;
1482	if visible(&run) {
1483	out.push(run);
1484	}
1485	}
1486	for path in legacy {
1487	let run = read_run_file(&path)?;
1488	if visible(&run) {
1489	out.push(run);
1490	}
1491	}
1492
1493	out.sort_by_key(\|r\| std::cmp::Reverse(r.created_at));
1494	// A crash between the sortable-name write and the legacy-file removal
1495	// in `save_run` can leave one run under both names; keep the sortable
1496	// copy (chained first above, so it survives the stable sort).
1497	out.dedup_by(\|a, b\| a.id == b.id);
1498	if let Some(limit) = limit {
1499	out.truncate(limit);
1500	}
1501	Ok(out)
1502	}
1503
1504	/// Re-read specific runs by id without a full history pass. Run file
1505	/// names end in `-{run_id}.json` (or are legacy `{run_id}.json`), so a
1506	/// directory listing locates them and only those files are read. Used
1507	/// by the activity-band scan to keep watching runs this session saw go
1508	/// live even after newer runs push them past the newest-run window.
1509	pub fn get_runs_by_ids(
1510	&self,
1511	automation_id: &str,
1512	run_ids: &std::collections::BTreeSet<String>,
1513	) -> Result<Vec<AutomationRunRecord>> {
1514	if run_ids.is_empty() {
1515	return Ok(Vec::new());
1516	}
1517	let dir = self.runs_dir_for(automation_id)?;
1518	if !dir.exists() {
1519	return Ok(Vec::new());
1520	}
1521	let mut paths = BTreeMap::<String, (bool, PathBuf)>::new();
1522	for entry in
1523	fs::read_dir(&dir).with_context(\|\| format!("Failed to read {}", dir.display()))?
1524	{
1525	let path = entry?.path();
1526	if path.extension().and_then(\|ext\| ext.to_str()) != Some("json") {
1527	continue;
1528	}
1529	let Some(stem) = path.file_stem().and_then(\|stem\| stem.to_str()) else {
1530	continue;
1531	};
1532	let sortable = has_sortable_run_stem(stem);
1533	let id = if sortable {
1534	&stem[RUN_STAMP_LEN + 1..]
1535	} else {
1536	stem
1537	};
1538	if run_ids.contains(id)
1539	&& paths
1540	.get(id)
1541	.is_none_or(\|(current, _)\| !current && sortable)
1542	{
1543	paths.insert(id.to_string(), (sortable, path));
1544	}
1545	}
1546	let mut out = Vec::new();
1547	for (_, path) in paths.into_values() {
1548	let run = read_run_file(&path)?;
1549	if !run
1550	.dispatch
1551	.as_ref()
1552	.is_some_and(\|dispatch\| dispatch.suppress_report)
1553	{
1554	out.push(run);
1555	}
1556	}
1557	Ok(out)
1558	}
1559
1560	fn save_run(&self, run: &AutomationRunRecord) -> Result<()> {
1561	let dir = self.runs_dir_for(&run.automation_id)?;
1562	fs::create_dir_all(&dir).with_context(\|\| format!("Failed to create {}", dir.display()))?;
1563	let path = self.run_path(run)?;
1564	write_json_atomic(&path, run)?;
1565	// Rewrites of a legacy-named run migrate it to the sortable name; drop
1566	// the old file so the run never exists twice.
1567	let legacy = self.legacy_run_path(&run.automation_id, &run.id)?;
1568	if legacy != path && legacy.exists() {
1569	fs::remove_file(&legacy)
1570	.with_context(\|\| format!("Failed to remove legacy run {}", legacy.display()))?;
1571	}
1572	Ok(())
1573	}
1574
1575	fn delete_run(&self, run: &AutomationRunRecord) -> Result<()> {
1576	let sortable = self.run_path(run)?;
1577	if sortable.exists() {
1578	fs::remove_file(&sortable)
1579	.with_context(\|\| format!("Failed to delete run {}", sortable.display()))?;
1580	}
1581	let legacy = self.legacy_run_path(&run.automation_id, &run.id)?;
1582	if legacy.exists() {
1583	fs::remove_file(&legacy)
1584	.with_context(\|\| format!("Failed to delete run {}", legacy.display()))?;
1585	}
1586	Ok(())
1587	}
1588
1589	/// Definitions this build can read, in `list_automations` order.
1590	///
1591	/// One corrupt, unreadable, or newer-schema file is quarantined in place:
1592	/// its bytes stay on disk and every pass logs the path, but it cannot
1593	/// starve collection of the healthy definitions behind it, and it is never
1594	/// rewritten or adopted by a runtime that does not understand it.
1595	fn readable_automations(&self) -> Result<Vec<AutomationRecord>> {
1596	let mut out = Vec::new();
1597	for entry in fs::read_dir(&self.automations_dir)
1598	.with_context(\|\| format!("Failed to read {}", self.automations_dir.display()))?
1599	{
1600	let entry = entry?;
1601	let path = entry.path();
1602	if path.extension().is_none_or(\|ext\| ext != "json") {
1603	continue;
1604	}
1605	match read_automation_file(&path) {
1606	Ok(record) => out.push(record),
1607	Err(error) => {
1608	tracing::warn!("Skipping damaged automation file: {error:#}");
1609	}
1610	}
1611	}
1612	out.sort_by_key(\|r\| std::cmp::Reverse(r.updated_at));
1613	Ok(out)
1614	}
1615
1616	/// List proposals only. Every proposal is revalidated and durably claimed
1617	/// immediately before dispatch, not when an earlier batch item is awaited.
1618	fn collect_due_runs(
1619	&self,
1620	now: DateTime<Utc>,
1621	) -> Result<Vec<(AutomationRecord, AutomationRunRecord)>> {
1622	self.with_transaction(\|\| {
1623	let mut due = Vec::new();
1624	for mut automation in self.readable_automations()? {
1625	if automation.status != AutomationStatus::Active
1626	\|\| !self.eligible_scope(automation.execution_scope.as_deref())
1627	{
1628	continue;
1629	}
1630	// An owned definition whose schedule cannot be evaluated is
1631	// quarantined like a damaged file: left untouched (a newer
1632	// build may understand it), diagnosed every pass, and never
1633	// allowed to take down the rest of the collection.
1634	let schedule = match AutomationSchedule::parse_rrule(&automation.rrule) {
1635	Ok(schedule) => schedule,
1636	Err(error) => {
1637	tracing::warn!(
1638	"Skipping automation {} with unevaluable schedule {:?}: {error:#}",
1639	automation.id,
1640	automation.rrule
1641	);
1642	continue;
1643	}
1644	};
1645	let Some(due_at) = automation.next_run_at else {
1646	automation.next_run_at =
1647	match schedule.next_after_with_anchor(now, automation.created_at) {
1648	Ok(next) => Some(next),
1649	Err(error)
1650	if matches!(schedule, AutomationSchedule::Once { .. })
1651	&& error
1652	.to_string()
1653	.contains("Once schedule has no future run") =>
1654	{
1655	automation.status = AutomationStatus::Paused;
1656	None
1657	}
1658	Err(error) => {
1659	tracing::warn!(
1660	"Skipping automation {} whose schedule cannot produce a slot: {error:#}",
1661	automation.id
1662	);
1663	continue;
1664	}
1665	};
1666	automation.updated_at = now;
1667	self.save_automation_unlocked(&automation)?;
1668	continue;
1669	};
1670	if due_at <= now {
1671	due.push((
1672	automation.clone(),
1673	new_run_record(&automation.id, due_at, now),
1674	));
1675	}
1676	}
1677	Ok(due)
1678	})
1679	}
1680
1681	fn claim_scheduled_run(
1682	&self,
1683	observed: &AutomationRecord,
1684	mut run: AutomationRunRecord,
1685	task_data_dir: &Path,
1686	) -> Result<Option<AutomationRunRecord>> {
1687	self.with_transaction(\|\| {
1688	if !self.automation_path(&observed.id)?.try_exists()? {
1689	return Ok(None);
1690	}
1691	let mut current = self.get_automation(&observed.id)?;
1692	if !self.eligible_scope(current.execution_scope.as_deref())
1693	\|\| current != *observed
1694	\|\| current.status != AutomationStatus::Active
1695	\|\| current.next_run_at != Some(run.scheduled_for)
1696	{
1697	return Ok(None);
1698	}
1699	let schedule = AutomationSchedule::parse_rrule(&current.rrule)?;
1700	// Include the complete history, including legacy occurrence ids.
1701	let history = self.list_runs_with_visibility(&current.id, None, true)?;
1702	if history
1703	.iter()
1704	.any(\|existing\| existing.scheduled_for == run.scheduled_for)
1705	{
1706	self.advance_automation_after_slot(
1707	&mut current,
1708	&schedule,
1709	run.scheduled_for,
1710	Utc::now(),
1711	)?;
1712	return Ok(None);
1713	}
1714	// Keep the owed slot until the earlier run settles. Its eventual
1715	// catch-up coalesces the backlog without overlapping executions
1716	// or inventing cancellation receipts. Explicit run-now requests
1717	// remain operator intent and stay ungated.
1718	if history.iter().any(\|existing\| {
1719	matches!(
1720	existing.status,
1721	AutomationRunStatus::Queued \| AutomationRunStatus::Running
1722	)
1723	}) {
1724	return Ok(None);
1725	}
1726	bind_run_dispatch(&mut run, &current, task_data_dir, true)?;
1727	self.save_run(&run)?;
1728	// The durable claim is the point of no return. Pause/delete after
1729	// this point affects future occurrences, not this admitted work.
1730	// No task can start before the binding above is durable.
1731	self.advance_automation_after_slot(
1732	&mut current,
1733	&schedule,
1734	run.scheduled_for,
1735	Utc::now(),
1736	)?;
1737	Ok(Some(run))
1738	})
1739	}
1740
1741	/// Repair only a torn claim/advance transaction. A replacement definition,
1742	/// pause, or Operate kick has a different generation or slot and wins.
1743	fn recover_schedule_advance(&self, run: &AutomationRunRecord) -> Result<()> {
1744	let Some(admitted) = run
1745	.dispatch
1746	.as_ref()
1747	.and_then(\|dispatch\| dispatch.schedule.as_ref())
1748	else {
1749	return Ok(());
1750	};
1751	self.with_transaction(\|\| {
1752	if !self.automation_path(&run.automation_id)?.try_exists()? {
1753	return Ok(());
1754	}
1755	let mut current = self.get_automation(&run.automation_id)?;
1756	if current.status == AutomationStatus::Active
1757	&& current.updated_at == admitted.updated_at
1758	&& current.rrule == admitted.rrule
1759	&& current.next_run_at == Some(run.scheduled_for)
1760	{
1761	let schedule = AutomationSchedule::parse_rrule(&current.rrule)?;
1762	self.advance_automation_after_slot(
1763	&mut current,
1764	&schedule,
1765	run.scheduled_for,
1766	Utc::now(),
1767	)?;
1768	}
1769	Ok(())
1770	})
1771	}
1772
1773	/// Completion publishes only this occurrence's receipt. It never advances
1774	/// a definition that may have been edited during the enqueue await.
1775	fn finish_scheduled_run(&self, run: &AutomationRunRecord, now: DateTime<Utc>) -> Result<()> {
1776	self.with_transaction(\|\| {
1777	self.save_run(run)?;
1778	if matches!(
1779	run.status,
1780	AutomationRunStatus::Completed
1781	\| AutomationRunStatus::Failed
1782	\| AutomationRunStatus::Canceled
1783	) && !run
1784	.dispatch
1785	.as_ref()
1786	.is_some_and(\|dispatch\| dispatch.suppress_report)
1787	&& self.automation_path(&run.automation_id)?.try_exists()?
1788	{
1789	let mut current = self.get_automation(&run.automation_id)?;
1790	let ended = run.ended_at.unwrap_or(now);
1791	current.last_run_at = Some(
1792	current
1793	.last_run_at
1794	.map_or(ended, \|previous\| previous.max(ended)),
1795	);
1796	// Receipt metadata is not a new schedule generation. Never
1797	// recompute the current definition from this run's old slot.
1798	self.save_automation_unlocked(&current)?;
1799	}
1800	Ok(())
1801	})
1802	}
1803
1804	fn advance_automation_after_slot(
1805	&self,
1806	automation: &mut AutomationRecord,
1807	schedule: &AutomationSchedule,
1808	slot: DateTime<Utc>,
1809	now: DateTime<Utc>,
1810	) -> Result<()> {
1811	automation.updated_at = now;
1812	automation.next_run_at = schedule.next_unskipped_slot(slot, now, automation.created_at)?;
1813	if automation.next_run_at.is_none() {
1814	automation.status = AutomationStatus::Paused;
1815	}
1816	self.save_automation_unlocked(automation)
1817	}
1818
1819	/// Active receipts are independent of the definition's lifetime and of
1820	/// presentation limits on recent history.
1821	fn collect_pending_runs(&self) -> Result<Vec<AutomationRunRecord>> {
1822	let mut pending = Vec::new();
1823	for entry in fs::read_dir(&self.runs_dir)? {
1824	let entry = entry?;
1825	if !entry.file_type()?.is_dir() {
1826	continue;
1827	}
1828	let Some(id) = entry.file_name().to_str().map(str::to_owned) else {
1829	continue;
1830	};
1831	// A damaged receipt quarantines only its own automation: the bytes
1832	// stay on disk and the diagnostic is logged every pass, but one
1833	// corrupt file must not block recovery of every other pending run.
1834	let runs = match self.list_runs_with_visibility(&id, None, true) {
1835	Ok(runs) => runs,
1836	Err(error) => {
1837	tracing::warn!("Skipping damaged run history for automation {id}: {error:#}");
1838	continue;
1839	}
1840	};
1841	for run in runs {
1842	if matches!(
1843	run.status,
1844	AutomationRunStatus::Queued \| AutomationRunStatus::Running
1845	) && run.task_id.is_some()
1846	{
1847	pending.push(run);
1848	}
1849	}
1850	}
1851	Ok(pending)
1852	}
1853
1854	// ── Delayed-trigger storage methods ──────────────────────────────────
1855
1856	/// Persist a new delayed trigger and return the record.
1857	pub fn create_trigger(&self, req: CreateDelayedTriggerRequest) -> Result<DelayedTriggerRecord> {
1858	let now = Utc::now();
1859	if req.fire_at <= now {
1860	bail!(
1861	"fire_at must be in the future (got {}, now is {})",
1862	req.fire_at.to_rfc3339(),
1863	now.to_rfc3339()
1864	);
1865	}
1866	if req.message.trim().is_empty() {
1867	bail!("Trigger message must not be empty");
1868	}
1869	let record = DelayedTriggerRecord {
1870	schema_version: CURRENT_TRIGGER_SCHEMA_VERSION,
1871	execution_scope: self.execution_scope.clone(),
1872	trigger_id: format!("trig_{}", Uuid::new_v4().simple()),
1873	fire_at: req.fire_at,
1874	message: req.message.trim().to_string(),
1875	workspace: req.workspace,
1876	owner_session_id: req.owner_session_id,
1877	status: DelayedTriggerStatus::Pending,
1878	created_at: now,
1879	fired_at: None,
1880	task_id: None,
1881	thread_id: None,
1882	error: None,
1883	parent_trigger_id: req.parent_trigger_id,
1884	dispatch: None,
1885	};
1886	self.save_trigger(&record)?;
1887	Ok(record)
1888	}
1889
1890	/// Load a trigger by id.
1891	pub fn get_trigger(&self, trigger_id: &str) -> Result<DelayedTriggerRecord> {
1892	let path = self.trigger_path(trigger_id)?;
1893	let raw = fs::read_to_string(&path)
1894	.with_context(\|\| format!("Trigger '{trigger_id}' not found"))?;
1895	let record: DelayedTriggerRecord = serde_json::from_str(&raw)
1896	.with_context(\|\| format!("Failed to parse trigger '{trigger_id}'"))?;
1897	if record.schema_version > CURRENT_TRIGGER_SCHEMA_VERSION {
1898	bail!(
1899	"Trigger schema v{} is newer than supported v{}",
1900	record.schema_version,
1901	CURRENT_TRIGGER_SCHEMA_VERSION
1902	);
1903	}
1904	Ok(record)
1905	}
1906
1907	/// Load a trigger only when it belongs to the given session.
1908	///
1909	/// Foreign, ownerless legacy, unreadable, and absent records share the same
1910	/// result so trigger existence cannot be disclosed across sessions.
1911	pub fn get_trigger_for_owner(
1912	&self,
1913	trigger_id: &str,
1914	owner_session_id: &str,
1915	) -> Result<DelayedTriggerRecord> {
1916	self.get_trigger(trigger_id)
1917	.ok()
1918	.filter(\|record\| record.owner_session_id.as_deref() == Some(owner_session_id))
1919	.ok_or_else(\|\| anyhow::anyhow!("Trigger '{trigger_id}' not found"))
1920	}
1921
1922	/// Atomically persist a trigger record.
1923	pub fn save_trigger(&self, record: &DelayedTriggerRecord) -> Result<()> {
1924	self.with_transaction(\|\| self.save_trigger_unlocked(record))
1925	}
1926
1927	fn save_trigger_unlocked(&self, record: &DelayedTriggerRecord) -> Result<()> {
1928	let path = self.trigger_path(&record.trigger_id)?;
1929	write_json_atomic(&path, record)
1930	}
1931
1932	/// List triggers, newest first. Pass `status_filter` to restrict results.
1933	pub fn list_triggers(
1934	&self,
1935	status_filter: Option<DelayedTriggerStatus>,
1936	limit: Option<usize>,
1937	) -> Result<Vec<DelayedTriggerRecord>> {
1938	let mut out = Vec::new();
1939	if !self.triggers_dir.exists() {
1940	return Ok(out);
1941	}
1942	for entry in fs::read_dir(&self.triggers_dir)
1943	.with_context(\|\| format!("Failed to read {}", self.triggers_dir.display()))?
1944	{
1945	let entry = entry?;
1946	let path = entry.path();
1947	if path.extension().is_none_or(\|ext\| ext != "json") {
1948	continue;
1949	}
1950	match fs::read_to_string(&path)
1951	.ok()
1952	.and_then(\|raw\| serde_json::from_str::<DelayedTriggerRecord>(&raw).ok())
1953	{
1954	Some(record) => {
1955	if let Some(filter) = status_filter
1956	&& record.status != filter
1957	{
1958	continue;
1959	}
1960	out.push(record);
1961	}
1962	None => {
1963	tracing::warn!("Skipping unreadable trigger file {}", path.display());
1964	}
1965	}
1966	}
1967	out.sort_by_key(\|r\| std::cmp::Reverse(r.created_at));
1968	if let Some(limit) = limit {
1969	out.truncate(limit);
1970	}
1971	Ok(out)
1972	}
1973
1974	/// List session-owned triggers, applying ownership before sorting and limit.
1975	pub fn list_triggers_for_owner(
1976	&self,
1977	status_filter: Option<DelayedTriggerStatus>,
1978	limit: Option<usize>,
1979	owner_session_id: &str,
1980	) -> Result<Vec<DelayedTriggerRecord>> {
1981	let mut records = self.list_triggers(status_filter, None)?;
1982	records.retain(\|record\| record.owner_session_id.as_deref() == Some(owner_session_id));
1983	if let Some(limit) = limit {
1984	records.truncate(limit);
1985	}
1986	Ok(records)
1987	}
1988
1989	/// Cancel a pending trigger owned by the given session.
1990	pub fn cancel_trigger_for_owner(
1991	&self,
1992	trigger_id: &str,
1993	owner_session_id: &str,
1994	) -> Result<DelayedTriggerRecord> {
1995	self.with_transaction(\|\| {
1996	let mut record = self.get_trigger_for_owner(trigger_id, owner_session_id)?;
1997	if record.status != DelayedTriggerStatus::Pending \|\| record.dispatch.is_some() {
1998	bail!(
1999	"Trigger '{trigger_id}' cannot be canceled after admission (status: {:?})",
2000	record.status
2001	);
2002	}
2003	record.status = DelayedTriggerStatus::Canceled;
2004	self.save_trigger_unlocked(&record)?;
2005	Ok(record)
2006	})
2007	}
2008
2009	/// Return due proposals and unfinished durable trigger admissions.
2010	pub fn collect_due_triggers(&self, now: DateTime<Utc>) -> Result<Vec<DelayedTriggerRecord>> {
2011	Ok(self
2012	.list_triggers(None, None)?
2013	.into_iter()
2014	.filter(\|trigger\| {
2015	trigger.status == DelayedTriggerStatus::Dispatching
2016	\|\| (trigger.status == DelayedTriggerStatus::Pending
2017	&& trigger.owner_session_id.is_some()
2018	&& trigger.fire_at <= now)
2019	})
2020	.collect())
2021	}
2022	}
2023
2024	fn new_run_record(
2025	automation_id: &str,
2026	scheduled_for: DateTime<Utc>,
2027	created_at: DateTime<Utc>,
2028	) -> AutomationRunRecord {
2029	AutomationRunRecord {
2030	schema_version: CURRENT_RUN_SCHEMA_VERSION,
2031	id: Uuid::new_v4().to_string(),
2032	automation_id: automation_id.to_string(),
2033	scheduled_for,
2034	status: AutomationRunStatus::Queued,
2035	created_at,
2036	started_at: None,
2037	ended_at: None,
2038	task_id: None,
2039	thread_id: None,
2040	turn_id: None,
2041	error: None,
2042	dispatch: None,
2043	}
2044	}
2045
2046	fn automation_task_request(automation: &AutomationRecord) -> NewTaskRequest {
2047	NewTaskRequest {
2048	prompt: automation.prompt.clone(),
2049	name: Some(automation.name.clone()),
2050	model: automation.model.clone(),
2051	model_provider: automation.model_provider.clone(),
2052	model_provider_id: automation.model_provider_id.clone(),
2053	workspace: automation.cwds.first().cloned(),
2054	mode: Some(automation.task_mode()),
2055	allow_shell: Some(automation.task_allow_shell()),
2056	trust_mode: Some(automation.task_trust_mode()),
2057	auto_approve: Some(automation.task_auto_approve()),
2058	owner_session_id: None,
2059	}
2060	}
2061
2062	fn bind_run_dispatch(
2063	run: &mut AutomationRunRecord,
2064	automation: &AutomationRecord,
2065	task_data_dir: &Path,
2066	scheduled: bool,
2067	) -> Result<()> {
2068	if automation.execution_scope.is_none() {
2069	bail!("Automation execution ownership is unverified");
2070	}
2071	run.schema_version = CURRENT_RUN_SCHEMA_VERSION;
2072	run.task_id = Some(crate::task_manager::TaskManager::new_task_id());
2073	run.dispatch = Some(AutomationDispatch {
2074	execution_scope: automation.execution_scope.clone(),
2075	request: automation_task_request(automation),
2076	task_data_dir: task_data_dir
2077	.canonicalize()
2078	.context("resolve task store before automation admission")?,
2079	accepted: false,
2080	delivery_mode: automation.delivery_mode(),
2081	suppress_report: false,
2082	schedule: scheduled.then(\|\| AdmittedSchedule {
2083	updated_at: automation.updated_at,
2084	rrule: automation.rrule.clone(),
2085	}),
2086	});
2087	Ok(())
2088	}
2089
2090	fn check_dispatch_store(dispatch: &AutomationDispatch, tasks: &SharedTaskManager) -> Result<()> {
2091	if tasks.data_dir().canonicalize()? != dispatch.task_data_dir.canonicalize()? {
2092	bail!("Automation admission belongs to a different task store; it cannot be replayed here");
2093	}
2094	Ok(())
2095	}
2096
2097	async fn dispatch_bound_task(
2098	dispatch: &mut AutomationDispatch,
2099	task_id: &str,
2100	tasks: &SharedTaskManager,
2101	) -> Result<crate::task_manager::TaskRecord> {
2102	check_dispatch_store(dispatch, tasks)?;
2103	if dispatch.execution_scope.as_deref() != Some(tasks.execution_scope()) {
2104	bail!(
2105	"Automation admission execution ownership is unverified or belongs to another Runtime"
2106	);
2107	}
2108	let task = if dispatch.accepted {
2109	tasks
2110	.read_bound_task(task_id)?
2111	.context("Accepted automation task is missing; refusing to replay it")?
2112	} else {
2113	tasks
2114	.recover_task_admission(dispatch.request.clone(), task_id.to_owned())
2115	.await?
2116	};
2117	crate::task_manager::validate_bound_task_request(&task, &dispatch.request)?;
2118	dispatch.accepted = true;
2119	dispatch.suppress_report = dispatch.delivery_mode == AutomationDeliveryMode::Watcher
2120	&& task.status == TaskStatus::Completed
2121	&& task
2122	.result_summary
2123	.as_deref()
2124	.is_some_and(\|summary\| summary.trim() == AUTOMATION_WATCHER_NO_REPORT_SENTINEL);
2125	Ok(task)
2126	}
2127
2128	/// Caller owns dispatch.lock and has already persisted this exact binding.
2129	async fn enqueue_run_task(run: &mut AutomationRunRecord, tasks: &SharedTaskManager) {
2130	let result = match (&mut run.dispatch, &run.task_id) {
2131	(Some(dispatch), Some(task_id)) => dispatch_bound_task(dispatch, task_id, tasks).await,
2132	_ => Err(anyhow::anyhow!(
2133	"Automation run has no durable task binding"
2134	)),
2135	};
2136	match result {
2137	Ok(task) => {
2138	run.error = None;
2139	apply_task_status(run, &task);
2140	}
2141	Err(error) => {
2142	// Keep the same pending identity after uncertain admission. A later
2143	// tick first looks for its canonical task; no new id is allocated.
2144	run.error = Some(format!(
2145	"Automation task admission needs recovery: {error:#}"
2146	));
2147	}
2148	}
2149	}
2150
2151	fn dispatch_lock_busy(error: &std::io::Error) -> bool {
2152	error.kind() == std::io::ErrorKind::WouldBlock \|\| matches!(error.raw_os_error(), Some(32 \| 33))
2153	}
2154
2155	pub async fn run_now_shared(
2156	automations: &SharedAutomationManager,
2157	automation_id: &str,
2158	task_manager: &SharedTaskManager,
2159	) -> Result<AutomationRunRecord> {
2160	automations.lock().await.bind_task_manager(task_manager)?;
2161	let task_manager = Arc::clone(task_manager);
2162	let task_data_dir = task_manager.data_dir();
2163	run_now_with(
2164	automations,
2165	automation_id,
2166	&task_data_dir,
2167	move \|_, mut run\| async move {
2168	enqueue_run_task(&mut run, &task_manager).await;
2169	run
2170	},
2171	)
2172	.await
2173	}
2174
2175	/// Keep the process-wide dispatch claim over the await, never the manager mutex.
2176	async fn run_now_with<F, Fut>(
2177	automations: &SharedAutomationManager,
2178	automation_id: &str,
2179	task_data_dir: &Path,
2180	enqueue: F,
2181	) -> Result<AutomationRunRecord>
2182	where
2183	F: FnOnce(AutomationRecord, AutomationRunRecord) -> Fut,
2184	Fut: Future<Output = AutomationRunRecord>,
2185	{
2186	let mut lock = automations.lock().await.open_lock("dispatch.lock")?;
2187	let _dispatch = lock
2188	.try_write()
2189	.context("Automation dispatcher is busy; retry the run")?;
2190	let (automation, run) = {
2191	let manager = automations.lock().await;
2192	manager.with_transaction(\|\| {
2193	let mut automation = manager.get_automation(automation_id)?;
2194	manager.adopt_for_run(&mut automation)?;
2195	let now = Utc::now();
2196	let mut run = new_run_record(&automation.id, now, now);
2197	bind_run_dispatch(&mut run, &automation, task_data_dir, false)?;
2198	manager.save_run(&run)?;
2199	Ok((automation, run))
2200	})?
2201	};
2202	let run = enqueue(automation, run).await;
2203	automations
2204	.lock()
2205	.await
2206	.finish_scheduled_run(&run, Utc::now())?;
2207	Ok(run)
2208	}
2209
2210	async fn scheduler_tick_shared(
2211	automations: &SharedAutomationManager,
2212	task_manager: &SharedTaskManager,
2213	) -> Result<()> {
2214	automations.lock().await.bind_task_manager(task_manager)?;
2215	let tasks = Arc::clone(task_manager);
2216	scheduler_tick_with(automations, &tasks.data_dir(), move \|mut run\| {
2217	let tasks = Arc::clone(&tasks);
2218	async move {
2219	enqueue_run_task(&mut run, &tasks).await;
2220	run
2221	}
2222	})
2223	.await
2224	}
2225
2226	async fn scheduler_tick_with<F, Fut>(
2227	automations: &SharedAutomationManager,
2228	task_data_dir: &Path,
2229	mut enqueue: F,
2230	) -> Result<()>
2231	where
2232	F: FnMut(AutomationRunRecord) -> Fut,
2233	Fut: Future<Output = AutomationRunRecord>,
2234	{
2235	let mut lock = automations.lock().await.open_lock("dispatch.lock")?;
2236	let _dispatch = match lock.try_write() {
2237	Ok(guard) => guard,
2238	Err(error) if dispatch_lock_busy(&error) => return Ok(()),
2239	Err(error) => return Err(error).context("claim automation dispatch"),
2240	};
2241	// Repair admitted work before collecting a new occurrence, including claims
2242	// whose definitions were edited/deleted or whose final enqueue save tore.
2243	let pending = automations.lock().await.collect_pending_runs()?;
2244	for run in pending.into_iter().filter(\|run\| {
2245	run.dispatch
2246	.as_ref()
2247	.is_some_and(\|dispatch\| !dispatch.accepted)
2248	}) {
2249	if !automations.lock().await.eligible_scope(
2250	run.dispatch
2251	.as_ref()
2252	.and_then(\|dispatch\| dispatch.execution_scope.as_deref()),
2253	) {
2254	continue;
2255	}
2256	// A single damaged admission is quarantined to its diagnostic; it must
2257	// not take down recovery of every pending run behind it.
2258	if let Err(error) = automations.lock().await.recover_schedule_advance(&run) {
2259	tracing::warn!(
2260	"automation schedule recovery failed for run {}: {error:#}",
2261	run.id
2262	);
2263	continue;
2264	}
2265	let run = enqueue(run).await;
2266	if let Err(error) = automations
2267	.lock()
2268	.await
2269	.finish_scheduled_run(&run, Utc::now())
2270	{
2271	tracing::warn!(
2272	"automation run {} receipt could not be persisted: {error:#}",
2273	run.id
2274	);
2275	}
2276	}
2277	let now = Utc::now();
2278	let due = automations.lock().await.collect_due_runs(now)?;
2279	for (observed, proposed) in due {
2280	if !automations
2281	.lock()
2282	.await
2283	.eligible_scope(observed.execution_scope.as_deref())
2284	{
2285	continue;
2286	}
2287	let run =
2288	match automations
2289	.lock()
2290	.await
2291	.claim_scheduled_run(&observed, proposed, task_data_dir)
2292	{
2293	Ok(run) => run,
2294	Err(error) => {
2295	// One automation's claim failure (for example a corrupt
2296	// receipt in its own dedup history) quarantines that
2297	// automation, not the tick: later due work still dispatches.
2298	tracing::warn!(
2299	"automation {} occurrence claim failed: {error:#}",
2300	observed.id
2301	);
2302	continue;
2303	}
2304	};
2305	let Some(run) = run else {
2306	continue;
2307	};
2308	let run = enqueue(run).await;
2309	if let Err(error) = automations.lock().await.finish_scheduled_run(&run, now) {
2310	tracing::warn!(
2311	"automation run {} receipt could not be persisted: {error:#}",
2312	run.id
2313	);
2314	}
2315	}
2316	Ok(())
2317	}
2318
2319	async fn fire_due_triggers_shared(
2320	automations: &SharedAutomationManager,
2321	task_manager: &SharedTaskManager,
2322	) -> Result<()> {
2323	automations.lock().await.bind_task_manager(task_manager)?;
2324	let tasks = Arc::clone(task_manager);
2325	fire_due_triggers_with(automations, &tasks.data_dir(), move \|trigger\| {
2326	let tasks = Arc::clone(&tasks);
2327	async move { enqueue_trigger_task(trigger, &tasks).await }
2328	})
2329	.await
2330	}
2331
2332	async fn enqueue_trigger_task(
2333	mut trigger: DelayedTriggerRecord,
2334	tasks: &SharedTaskManager,
2335	) -> Result<DelayedTriggerRecord> {
2336	let result = dispatch_bound_task(
2337	trigger.dispatch.as_mut().context("trigger dispatch")?,
2338	trigger.task_id.as_deref().context("trigger task binding")?,
2339	tasks,
2340	)
2341	.await;
2342	match result {
2343	Ok(task) => {
2344	trigger.status = DelayedTriggerStatus::Fired;
2345	trigger.fired_at = Some(Utc::now());
2346	trigger.thread_id = task.thread_id.clone();
2347	trigger.error = None;
2348	}
2349	Err(error) => {
2350	trigger.error = Some(format!(
2351	"Delayed trigger admission needs recovery: {error:#}"
2352	))
2353	}
2354	}
2355	Ok(trigger)
2356	}
2357
2358	async fn fire_due_triggers_with<F, Fut>(
2359	automations: &SharedAutomationManager,
2360	task_data_dir: &Path,
2361	mut enqueue: F,
2362	) -> Result<()>
2363	where
2364	F: FnMut(DelayedTriggerRecord) -> Fut,
2365	Fut: Future<Output = Result<DelayedTriggerRecord>>,
2366	{
2367	let mut lock = automations.lock().await.open_lock("dispatch.lock")?;
2368	let _dispatch = match lock.try_write() {
2369	Ok(guard) => guard,
2370	Err(error) if dispatch_lock_busy(&error) => return Ok(()),
2371	Err(error) => return Err(error).context("claim delayed-trigger dispatch"),
2372	};
2373	let now = Utc::now();
2374	let candidates = automations.lock().await.collect_due_triggers(now)?;
2375	for candidate in candidates {
2376	let scope = if candidate.status == DelayedTriggerStatus::Dispatching {
2377	candidate
2378	.dispatch
2379	.as_ref()
2380	.and_then(\|d\| d.execution_scope.as_deref())
2381	} else {
2382	candidate.execution_scope.as_deref()
2383	};
2384	if !automations.lock().await.eligible_scope(scope) {
2385	continue;
2386	}
2387	if !(candidate.status == DelayedTriggerStatus::Dispatching
2388	\|\| (candidate.status == DelayedTriggerStatus::Pending && candidate.fire_at <= now))
2389	{
2390	continue;
2391	}
2392	let claimed = {
2393	let manager = automations.lock().await;
2394	match manager.with_transaction(\|\| {
2395	let mut current = manager.get_trigger(&candidate.trigger_id)?;
2396	if current.status == DelayedTriggerStatus::Dispatching {
2397	if !manager.eligible_scope(
2398	current
2399	.dispatch
2400	.as_ref()
2401	.and_then(\|d\| d.execution_scope.as_deref()),
2402	) {
2403	return Ok(None);
2404	}
2405	if current.dispatch.is_none() \|\| current.task_id.is_none() {
2406	bail!("Claimed delayed trigger has no durable task binding");
2407	}
2408	return Ok(Some(current));
2409	}
2410	if !manager.eligible_scope(current.execution_scope.as_deref())
2411	\|\| current.status != DelayedTriggerStatus::Pending
2412	\|\| current.fire_at > now
2413	\|\| current.owner_session_id.is_none()
2414	{
2415	return Ok(None);
2416	}
2417	current.schema_version = CURRENT_TRIGGER_SCHEMA_VERSION;
2418	current.status = DelayedTriggerStatus::Dispatching;
2419	current.task_id = Some(crate::task_manager::TaskManager::new_task_id());
2420	current.dispatch = Some(AutomationDispatch {
2421	execution_scope: current.execution_scope.clone(),
2422	request: NewTaskRequest {
2423	prompt: current.message.clone(),
2424	name: None,
2425	model: None,
2426	model_provider: None,
2427	model_provider_id: None,
2428	workspace: current.workspace.clone(),
2429	mode: Some("agent".into()),
2430	allow_shell: Some(false),
2431	trust_mode: Some(false),
2432	auto_approve: Some(false),
2433	owner_session_id: current.owner_session_id.clone(),
2434	},
2435	task_data_dir: task_data_dir.canonicalize()?,
2436	accepted: false,
2437	delivery_mode: AutomationDeliveryMode::Task,
2438	suppress_report: false,
2439	schedule: None,
2440	});
2441	manager.save_trigger_unlocked(&current)?;
2442	Ok(Some(current))
2443	}) {
2444	Ok(claimed) => claimed,
2445	Err(error) => {
2446	// One damaged trigger record quarantines to a diagnostic;
2447	// the remaining due triggers still fire this pass.
2448	tracing::warn!(
2449	"delayed trigger {} claim failed: {error:#}",
2450	candidate.trigger_id
2451	);
2452	continue;
2453	}
2454	}
2455	};
2456	let Some(trigger) = claimed else {
2457	continue;
2458	};
2459	let trigger = match enqueue(trigger).await {
2460	Ok(trigger) => trigger,
2461	Err(error) => {
2462	tracing::warn!(
2463	"delayed trigger {} enqueue failed: {error:#}",
2464	candidate.trigger_id
2465	);
2466	continue;
2467	}
2468	};
2469	if let Err(error) = automations.lock().await.save_trigger(&trigger) {
2470	tracing::warn!(
2471	"delayed trigger {} receipt could not be persisted: {error:#}",
2472	trigger.trigger_id
2473	);
2474	}
2475	}
2476	Ok(())
2477	}
2478
2479	/// Fold a durable task's state back into its automation run. Returns whether
2480	/// the run changed and needs persisting.
2481	fn apply_task_status(
2482	run: &mut AutomationRunRecord,
2483	task: &crate::task_manager::TaskRecord,
2484	) -> bool {
2485	let mut changed = run.thread_id != task.thread_id \|\| run.turn_id != task.turn_id;
2486	run.thread_id = task.thread_id.clone();
2487	run.turn_id = task.turn_id.clone();
2488	match task.status {
2489	TaskStatus::Queued => {
2490	if !matches!(run.status, AutomationRunStatus::Queued) {
2491	run.status = AutomationRunStatus::Queued;
2492	changed = true;
2493	}
2494	}
2495	TaskStatus::Running => {
2496	if !matches!(run.status, AutomationRunStatus::Running) {
2497	run.status = AutomationRunStatus::Running;
2498	changed = true;
2499	}
2500	if run.started_at.is_none() {
2501	run.started_at = Some(task.started_at.unwrap_or_else(Utc::now));
2502	changed = true;
2503	}
2504	}
2505	TaskStatus::Completed => {
2506	run.status = AutomationRunStatus::Completed;
2507	run.started_at = run.started_at.or(task.started_at);
2508	run.ended_at = task.ended_at.or(Some(Utc::now()));
2509	run.error = None;
2510	changed = true;
2511	}
2512	TaskStatus::Failed => {
2513	run.status = AutomationRunStatus::Failed;
2514	run.started_at = run.started_at.or(task.started_at);
2515	run.ended_at = task.ended_at.or(Some(Utc::now()));
2516	run.error = task.error.clone();
2517	changed = true;
2518	}
2519	TaskStatus::Canceled => {
2520	run.status = AutomationRunStatus::Canceled;
2521	run.started_at = run.started_at.or(task.started_at);
2522	run.ended_at = task.ended_at.or(Some(Utc::now()));
2523	// #6162: a cancellation is not silent. Keep the task's own error
2524	// when it recorded one, otherwise name the terminal reason so the
2525	// settled receipt can say who or what canceled the run.
2526	run.error = task.error.clone().or_else(\|\| {
2527	task.terminal_reason
2528	.as_deref()
2529	.map(cancellation_reason_text)
2530	});
2531	changed = true;
2532	}
2533	}
2534	changed
2535	}
2536
2537	/// Human-readable cancellation detail for a run whose task ended without an
2538	/// error of its own. The task manager's terminal reasons are stable strings
2539	/// (`TaskTerminalReason::as_str`); anything unknown is passed through.
2540	fn cancellation_reason_text(terminal_reason: &str) -> String {
2541	// A cooperative cancel is the one path that arrives without an error of
2542	// its own; cancel-timeout and shutdown already carry the task manager's
2543	// receipt message, so those arms are a fallback for records that lost it.
2544	match terminal_reason {
2545	"canceled" => "canceled by request".to_string(),
2546	"cancel_timeout" => "canceled; the task did not stop within the cancel timeout".to_string(),
2547	"shutdown" => "canceled by shutdown".to_string(),
2548	other => format!("canceled ({other})"),
2549	}
2550	}
2551
2552	async fn reconcile_run_statuses_shared(
2553	automations: &SharedAutomationManager,
2554	task_manager: &SharedTaskManager,
2555	) -> Result<()> {
2556	automations.lock().await.bind_task_manager(task_manager)?;
2557	let mut lock = automations.lock().await.open_lock("dispatch.lock")?;
2558	let _dispatch = match lock.try_write() {
2559	Ok(guard) => guard,
2560	Err(error) if dispatch_lock_busy(&error) => return Ok(()),
2561	Err(error) => return Err(error).context("claim automation reconciliation"),
2562	};
2563	let pending = automations.lock().await.collect_pending_runs()?;
2564	for mut run in pending {
2565	// Shared storage is not shared ownership. A receipt admitted under
2566	// another execution scope is reconciled by that scope's owner; this
2567	// process must not stamp errors onto it or rewrite it from a bound
2568	// task record it cannot see.
2569	if run
2570	.dispatch
2571	.as_ref()
2572	.and_then(\|dispatch\| dispatch.execution_scope.as_deref())
2573	!= Some(task_manager.execution_scope())
2574	{
2575	continue;
2576	}
2577	let Some(task_id) = run.task_id.clone() else {
2578	continue;
2579	};
2580	let lookup = (\|\| {
2581	if let Some(dispatch) = &run.dispatch {
2582	check_dispatch_store(dispatch, task_manager)?;
2583	}
2584	let task = task_manager.read_bound_task(&task_id)?;
2585	if let Some(task) = &task
2586	&& let Some(dispatch) = &run.dispatch
2587	{
2588	crate::task_manager::validate_bound_task_request(task, &dispatch.request)?;
2589	}
2590	Ok::<_, anyhow::Error>(task)
2591	})();
2592	let task = match lookup {
2593	Ok(Some(task)) => task,
2594	Ok(None) => {
2595	if run
2596	.dispatch
2597	.as_ref()
2598	.is_some_and(\|dispatch\| dispatch.accepted)
2599	{
2600	// The admission was durably accepted but the bound task
2601	// record is gone: this occurrence can never be replayed
2602	// or reconciled. Settle it terminally instead of retrying
2603	// the same lookup every pass and starving the schedule
2604	// behind it.
2605	run.status = AutomationRunStatus::Failed;
2606	run.ended_at = Some(run.ended_at.unwrap_or_else(Utc::now));
2607	run.error = Some(format!(
2608	"Bound automation task {task_id} is missing after durable acceptance"
2609	));
2610	} else {
2611	// Unaccepted admissions are the scheduler's recovery path:
2612	// the next tick reuses the durable binding or recreates
2613	// the task; reconcile only records the uncertainty.
2614	run.error = Some(format!(
2615	"Automation reconciliation unavailable: bound task {task_id} is missing"
2616	));
2617	}
2618	if let Err(error) = automations
2619	.lock()
2620	.await
2621	.finish_scheduled_run(&run, Utc::now())
2622	{
2623	tracing::warn!(
2624	"automation run {} receipt could not be persisted: {error:#}",
2625	run.id
2626	);
2627	}
2628	continue;
2629	}
2630	Err(error) => {
2631	run.error = Some(format!("Automation reconciliation unavailable: {error:#}"));
2632	if let Err(error) = automations
2633	.lock()
2634	.await
2635	.finish_scheduled_run(&run, Utc::now())
2636	{
2637	tracing::warn!(
2638	"automation run {} receipt could not be persisted: {error:#}",
2639	run.id
2640	);
2641	}
2642	continue;
2643	}
2644	};
2645	// The bound task itself belongs to another Runtime — hands off.
2646	if task.execution_scope.as_deref() != Some(task_manager.execution_scope()) {
2647	continue;
2648	}
2649	let watcher = run
2650	.dispatch
2651	.as_ref()
2652	.map(\|dispatch\| dispatch.delivery_mode)
2653	.or_else(\|\| {
2654	automations.try_lock().ok().and_then(\|manager\| {
2655	manager
2656	.get_automation(&run.automation_id)
2657	.ok()
2658	.map(\|automation\| automation.delivery_mode())
2659	})
2660	})
2661	== Some(AutomationDeliveryMode::Watcher);
2662	let watcher_noop = watcher
2663	&& task.status == TaskStatus::Completed
2664	&& task
2665	.result_summary
2666	.as_deref()
2667	.is_some_and(\|summary\| summary.trim() == AUTOMATION_WATCHER_NO_REPORT_SENTINEL);
2668	if !apply_task_status(&mut run, &task) {
2669	continue;
2670	}
2671	let dispatch = run.dispatch.get_or_insert_with(\|\| AutomationDispatch {
2672	execution_scope: task.execution_scope.clone(),
2673	request: NewTaskRequest::from_task(&task),
2674	task_data_dir: task_manager.data_dir(),
2675	accepted: true,
2676	delivery_mode: if watcher {
2677	AutomationDeliveryMode::Watcher
2678	} else {
2679	AutomationDeliveryMode::Task
2680	},
2681	suppress_report: false,
2682	schedule: None,
2683	});
2684	run.schema_version = CURRENT_RUN_SCHEMA_VERSION;
2685	dispatch.accepted = true;
2686	dispatch.suppress_report = watcher_noop;
2687	if let Err(error) = automations
2688	.lock()
2689	.await
2690	.finish_scheduled_run(&run, Utc::now())
2691	{
2692	tracing::warn!(
2693	"automation run {} receipt could not be persisted: {error:#}",
2694	run.id
2695	);
2696	}
2697	}
2698	Ok(())
2699	}
2700
2701	/// Fixed-width, lexically-sortable UTC stamp for run file names, e.g.
2702	/// `20260705T142530123Z` (millisecond precision; the run id suffix breaks
2703	/// same-millisecond ties deterministically).
2704	const RUN_STAMP_FORMAT: &str = "%Y%m%dT%H%M%S%3fZ";
2705	const RUN_STAMP_LEN: usize = "20260705T142530123Z".len();
2706
2707	fn run_file_stamp(created_at: DateTime<Utc>) -> String {
2708	created_at.format(RUN_STAMP_FORMAT).to_string()
2709	}
2710
2711	/// Shape check for `{stamp}-{run_id}` file stems. Ordering trusts the file
2712	/// name only for pruning; the parsed record's `created_at` stays
2713	/// authoritative for the final sort.
2714	fn has_sortable_run_stem(stem: &str) -> bool {
2715	let Some((stamp, rest)) = stem.split_at_checked(RUN_STAMP_LEN) else {
2716	return false;
2717	};
2718	if !rest.starts_with('-') \|\| rest.len() < 2 {
2719	return false;
2720	}
2721	stamp.char_indices().all(\|(idx, ch)\| match idx {
2722	8 => ch == 'T',
2723	18 => ch == 'Z',
2724	_ => ch.is_ascii_digit(),
2725	})
2726	}
2727
2728	fn read_automation_file(path: &Path) -> Result<AutomationRecord> {
2729	let raw = fs::read_to_string(path)
2730	.with_context(\|\| format!("Failed to read automation {}", path.display()))?;
2731	let record: AutomationRecord = serde_json::from_str(&raw)
2732	.with_context(\|\| format!("Failed to parse automation {}", path.display()))?;
2733	if record.schema_version > CURRENT_AUTOMATION_SCHEMA_VERSION {
2734	bail!(
2735	"Automation schema v{} is newer than supported v{}",
2736	record.schema_version,
2737	CURRENT_AUTOMATION_SCHEMA_VERSION
2738	);
2739	}
2740	Ok(record)
2741	}
2742
2743	fn read_run_file(path: &Path) -> Result<AutomationRunRecord> {
2744	let raw =
2745	fs::read_to_string(path).with_context(\|\| format!("Failed to read {}", path.display()))?;
2746	let run: AutomationRunRecord = serde_json::from_str(&raw)
2747	.with_context(\|\| format!("Failed to parse {}", path.display()))?;
2748	if run.schema_version > CURRENT_RUN_SCHEMA_VERSION {
2749	bail!(
2750	"Automation run schema v{} is newer than supported v{}",
2751	run.schema_version,
2752	CURRENT_RUN_SCHEMA_VERSION
2753	);
2754	}
2755	Ok(run)
2756	}
2757
2758	fn ensure_safe_storage_id(kind: &str, value: &str) -> Result<()> {
2759	let mut components = Path::new(value).components();
2760	let Some(component) = components.next() else {
2761	bail!("{kind} must not be empty");
2762	};
2763	if components.next().is_some() \|\| !matches!(component, std::path::Component::Normal(_)) {
2764	bail!("{kind} must be a single path component");
2765	}
2766	Ok(())
2767	}
2768
2769	fn validate_name_and_prompt(name: &str, prompt: &str) -> Result<()> {
2770	if name.trim().is_empty() {
2771	bail!("Automation name is required");
2772	}
2773	if prompt.trim().is_empty() {
2774	bail!("Automation prompt is required");
2775	}
2776	Ok(())
2777	}
2778
2779	fn normalize_optional_string(value: Option<String>) -> Option<String> {
2780	value
2781	.map(\|value\| value.trim().to_string())
2782	.filter(\|value\| !value.is_empty())
2783	}
2784
2785	fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
2786	if let Some(parent) = path.parent() {
2787	fs::create_dir_all(parent).with_context(\|\| format!("create {}", parent.display()))?;
2788	}
2789	crate::utils::write_atomic(path, &serde_json::to_vec_pretty(value)?)
2790	.with_context(\|\| format!("write {}", path.display()))
2791	}
2792
2793	pub fn default_automations_dir() -> PathBuf {
2794	// Most-specific override: an explicit automations dir.
2795	for var in ["CODEWHALE_AUTOMATIONS_DIR", "DEEPSEEK_AUTOMATIONS_DIR"] {
2796	if let Ok(path) = std::env::var(var) {
2797	let trimmed = path.trim();
2798	if !trimmed.is_empty() {
2799	return PathBuf::from(trimmed);
2800	}
2801	}
2802	}
2803	// $CODEWHALE_HOME is a hard override of the base data directory
2804	// (docs/CONFIGURATION.md): when SET, automations live under it and we do
2805	// NOT fall back to the legacy ~/.deepseek path — silent fallback would
2806	// defeat the isolation the override promises. Check the env var directly
2807	// (not codewhale_home()'s Ok/Err, which succeeds for the default home too).
2808	if let Some(home) = codewhale_paths::codewhale_home_override().ok().flatten() {
2809	return home.join("automations");
2810	}
2811	codewhale_paths::user_home()
2812	.map(\|home\| {
2813	let primary = home.join(".codewhale").join("automations");
2814	let legacy = home.join(".deepseek").join("automations");
2815	if primary.exists() \|\| !legacy.exists() {
2816	return primary;
2817	}
2818	legacy
2819	})
2820	.unwrap_or_else(\|\| PathBuf::from(".codewhale").join("automations"))
2821	}
2822
2823	pub type SharedAutomationManager = Arc<Mutex<AutomationManager>>;
2824
2825	#[derive(Debug, Clone)]
2826	pub struct AutomationSchedulerConfig {
2827	pub tick_interval_secs: u64,
2828	}
2829
2830	impl Default for AutomationSchedulerConfig {
2831	fn default() -> Self {
2832	Self {
2833	tick_interval_secs: 15,
2834	}
2835	}
2836	}
2837
2838	pub fn spawn_scheduler(
2839	automations: SharedAutomationManager,
2840	task_manager: SharedTaskManager,
2841	cancel: CancellationToken,
2842	config: AutomationSchedulerConfig,
2843	) -> tokio::task::JoinHandle<()> {
2844	spawn_supervised(
2845	"automation-scheduler",
2846	std::panic::Location::caller(),
2847	async move {
2848	let interval = config.tick_interval_secs.max(5);
2849	loop {
2850	if cancel.is_cancelled() {
2851	break;
2852	}
2853
2854	// Lock scope lives inside the shared helpers: the manager
2855	// mutex is dropped across every task-manager await so API and
2856	// tool callers are never queued behind enqueue/status latency.
2857	if let Err(err) = scheduler_tick_shared(&automations, &task_manager).await {
2858	tracing::warn!("automation scheduler tick failed: {err}");
2859	}
2860	if let Err(err) = reconcile_run_statuses_shared(&automations, &task_manager).await {
2861	tracing::warn!("automation reconcile failed: {err}");
2862	}
2863	if let Err(err) = fire_due_triggers_shared(&automations, &task_manager).await {
2864	tracing::warn!("delayed trigger tick failed: {err}");
2865	}
2866
2867	tokio::select! {
2868	_ = cancel.cancelled() => break,
2869	_ = sleep(std::time::Duration::from_secs(interval)) => {}
2870	}
2871	}
2872	},
2873	)
2874	}
2875
2876	#[cfg(test)]
2877	mod tests {
2878	use super::*;
2879	use async_trait::async_trait;
2880	use chrono::{FixedOffset, LocalResult, NaiveDate};
2881	use tokio::sync::mpsc;
2882
2883	use crate::task_manager::{
2884	ExecutionTask, TaskExecutionEvent, TaskExecutionResult, TaskExecutor, TaskManager,
2885	TaskManagerConfig,
2886	};
2887
2888	struct AutomationRecordingExecutor(PathBuf);
2889
2890	#[async_trait]
2891	impl TaskExecutor for AutomationRecordingExecutor {
2892	async fn execute(
2893	&self,
2894	task: ExecutionTask,
2895	_events: mpsc::Sender<TaskExecutionEvent>,
2896	_cancel: CancellationToken,
2897	) -> TaskExecutionResult {
2898	use std::io::Write as _;
2899	let recorded = (\|\| -> Result<()> {
2900	let id = task
2901	.thread_request()
2902	.task_id
2903	.context("fixture task identity")?;
2904	let mut file = fs::OpenOptions::new()
2905	.create(true)
2906	.append(true)
2907	.open(&self.0)?;
2908	writeln!(file, "{id}")?;
2909	file.sync_all()?;
2910	Ok(())
2911	})();
2912	match recorded {
2913	Ok(()) => TaskExecutionResult {
2914	status: TaskStatus::Completed,
2915	result_text: Some("automation fixture completed".into()),
2916	error: None,
2917	terminal_reason: crate::task_manager::TaskTerminalReason::Completed,
2918	},
2919	Err(error) => TaskExecutionResult {
2920	status: TaskStatus::Failed,
2921	result_text: None,
2922	error: Some(error.to_string()),
2923	terminal_reason: crate::task_manager::TaskTerminalReason::Failed,
2924	},
2925	}
2926	}
2927	}
2928
2929	fn fixture_executions(path: &Path) -> Vec<String> {
2930	match fs::read_to_string(path) {
2931	Ok(text) => text.lines().map(str::to_owned).collect(),
2932	Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
2933	Err(error) => panic!("read independent execution receipts: {error}"),
2934	}
2935	}
2936
2937	async fn fixture_tasks(root: &Path, receipts: &Path) -> Result<SharedTaskManager> {
2938	TaskManager::start_with_executor(
2939	automation_task_config(root.to_path_buf()),
2940	Arc::new(AutomationRecordingExecutor(receipts.to_path_buf())),
2941	)
2942	.await
2943	}
2944
2945	fn fixture_due_automation(
2946	manager: &AutomationManager,
2947	id: &str,
2948	order: i64,
2949	) -> AutomationRecord {
2950	let mut automation = automation_record_with_settings(None, None, None, None);
2951	automation.id = id.to_owned();
2952	automation.prompt = format!("automation fixture {id}");
2953	automation.created_at = Utc::now() - Duration::hours(2);
2954	automation.updated_at = Utc::now() - Duration::minutes(order);
2955	automation.next_run_at = Some(Utc::now() - Duration::minutes(1));
2956	manager
2957	.save_automation(&automation)
2958	.expect("save due fixture");
2959	automation
2960	}
2961
2962	#[tokio::test]
2963	async fn interrupted_dispatch_reuses_binding_and_preserves_replacement_schedule() -> Result<()>
2964	{
2965	for accepted_before_cut in [false, true] {
2966	let root = tempfile::tempdir()?;
2967	let receipts = root.path().join("executions");
2968	let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
2969	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
2970	let automation = fixture_due_automation(&manager, "recover", 1);
2971	let shared = Arc::new(Mutex::new(manager));
2972	let (at_cut, reached_cut) = tokio::sync::oneshot::channel();
2973	let tick = tokio::spawn({
2974	let shared = shared.clone();
2975	let tasks = tasks.clone();
2976	let mut at_cut = Some(at_cut);
2977	async move {
2978	scheduler_tick_with(&shared, &tasks.data_dir(), move \|mut run\| {
2979	let tasks = tasks.clone();
2980	let at_cut = at_cut.take();
2981	async move {
2982	if accepted_before_cut {
2983	enqueue_run_task(&mut run, &tasks).await;
2984	}
2985	if let Some(at_cut) = at_cut {
2986	let _ = at_cut.send(run.clone());
2987	}
2988	std::future::pending::<()>().await;
2989	run
2990	}
2991	})
2992	.await
2993	}
2994	});
2995	let observed =
2996	tokio::time::timeout(std::time::Duration::from_secs(5), reached_cut).await??;
2997	let bound_id = observed.task_id.clone().context("claimed task id")?;
2998	let persisted = shared.lock().await.list_runs(&automation.id, None)?;
2999	assert_eq!(
3000	persisted.len(),
3001	1,
3002	"intent exists before either interruption boundary"
3003	);
3004	assert_eq!(persisted[0].task_id.as_deref(), Some(bound_id.as_str()));
3005	assert!(
3006	!persisted[0].dispatch.as_ref().unwrap().accepted,
3007	"final acceptance save has not run"
3008	);
3009	if accepted_before_cut {
3010	let task = crate::task_manager::wait_for_terminal_state(
3011	&tasks,
3012	&bound_id,
3013	std::time::Duration::from_secs(5),
3014	)
3015	.await?;
3016	assert_eq!(task.status, TaskStatus::Completed);
3017	assert!(
3018	task.result_summary
3019	.as_deref()
3020	.unwrap_or_default()
3021	.contains("automation fixture completed")
3022	);
3023	assert_eq!(fixture_executions(&receipts), vec![bound_id.clone()]);
3024	} else {
3025	assert!(tasks.read_bound_task(&bound_id)?.is_none());
3026	assert!(fixture_executions(&receipts).is_empty());
3027	}
3028	// Simulate a dropped request/process before the final run receipt.
3029	tick.abort();
3030	assert!(tick.await.unwrap_err().is_cancelled());
3031	let reopened = AutomationManager::open_for_test(root.path().join("automations"))?;
3032	let future = Utc::now() + Duration::hours(4);
3033	let edited = reopened.update_automation(
3034	&automation.id,
3035	UpdateAutomationRequest {
3036	rrule: Some(format!("FREQ=ONCE;AT={}", future.to_rfc3339())),
3037	prompt: Some("future revised prompt".into()),
3038	..Default::default()
3039	},
3040	)?;
3041	let restarted = Arc::new(Mutex::new(reopened));
3042	scheduler_tick_shared(&restarted, &tasks).await?;
3043	let task = crate::task_manager::wait_for_terminal_state(
3044	&tasks,
3045	&bound_id,
3046	std::time::Duration::from_secs(5),
3047	)
3048	.await?;
3049	assert_eq!(task.status, TaskStatus::Completed);
3050	assert_eq!(
3051	task.prompt, automation.prompt,
3052	"claim keeps its original request"
3053	);
3054	reconcile_run_statuses_shared(&restarted, &tasks).await?;
3055	scheduler_tick_shared(&restarted, &tasks).await?;
3056	assert_eq!(
3057	fixture_executions(&receipts),
3058	vec![bound_id.clone()],
3059	"same occurrence executes exactly once"
3060	);
3061	let manager = restarted.lock().await;
3062	let runs = manager.list_runs(&automation.id, None)?;
3063	assert_eq!(runs.len(), 1);
3064	assert_eq!(runs[0].task_id.as_deref(), Some(bound_id.as_str()));
3065	assert_eq!(runs[0].status, AutomationRunStatus::Completed);
3066	assert!(runs[0].dispatch.as_ref().unwrap().accepted);
3067	let current = manager.get_automation(&automation.id)?;
3068	assert_eq!(current.rrule, edited.rrule);
3069	assert_eq!(
3070	current.next_run_at, edited.next_run_at,
3071	"old completion must not clear future ONCE"
3072	);
3073	assert_eq!(current.status, AutomationStatus::Active);
3074	assert_eq!(current.updated_at, edited.updated_at);
3075	assert_eq!(current.prompt, edited.prompt);
3076	assert!(current.last_run_at.is_some());
3077	tasks.shutdown();
3078	}
3079	Ok(())
3080	}
3081
3082	#[tokio::test]
3083	async fn pause_and_delete_before_claim_stop_collected_work_but_preserve_admitted_work()
3084	-> Result<()> {
3085	let root = tempfile::tempdir()?;
3086	let receipts = root.path().join("executions");
3087	let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
3088	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
3089	let first = fixture_due_automation(&manager, "first", 1);
3090	let paused = fixture_due_automation(&manager, "paused", 2);
3091	let deleted = fixture_due_automation(&manager, "deleted", 3);
3092	let shared = Arc::new(Mutex::new(manager));
3093	let (entered, reached) = tokio::sync::oneshot::channel();
3094	let (release, released) = tokio::sync::oneshot::channel();
3095	let tick = tokio::spawn({
3096	let shared = shared.clone();
3097	let tasks = tasks.clone();
3098	let mut barrier = Some((entered, released));
3099	async move {
3100	scheduler_tick_with(&shared, &tasks.data_dir(), move \|mut run\| {
3101	let barrier = barrier.take();
3102	let tasks = tasks.clone();
3103	async move {
3104	if let Some((entered, released)) = barrier {
3105	let _ = entered.send(run.clone());
3106	let _ = released.await;
3107	}
3108	enqueue_run_task(&mut run, &tasks).await;
3109	run
3110	}
3111	})
3112	.await
3113	}
3114	});
3115	let admitted = tokio::time::timeout(std::time::Duration::from_secs(5), reached).await??;
3116	assert_eq!(
3117	admitted.automation_id, first.id,
3118	"ordered first claim really entered"
3119	);
3120	let other = AutomationManager::open_for_test(root.path().join("automations"))?;
3121	other.pause_automation(&paused.id)?;
3122	other.delete_automation(&deleted.id)?;
3123	// Pause after durable admission affects future runs; this run remains owned.
3124	other.pause_automation(&first.id)?;
3125	release
3126	.send(())
3127	.map_err(\|_\| anyhow::anyhow!("release fixture"))?;
3128	tokio::time::timeout(std::time::Duration::from_secs(5), tick).await???;
3129	let id = admitted.task_id.context("admitted task identity")?;
3130	let task = crate::task_manager::wait_for_terminal_state(
3131	&tasks,
3132	&id,
3133	std::time::Duration::from_secs(5),
3134	)
3135	.await?;
3136	assert_eq!(task.status, TaskStatus::Completed);
3137	reconcile_run_statuses_shared(&shared, &tasks).await?;
3138	assert_eq!(fixture_executions(&receipts), vec![id]);
3139	assert!(other.list_runs(&paused.id, None)?.is_empty());
3140	assert!(other.list_runs(&deleted.id, None)?.is_empty());
3141	assert_eq!(
3142	other.get_automation(&first.id)?.status,
3143	AutomationStatus::Paused
3144	);
3145	assert!(other.get_automation(&first.id)?.next_run_at.is_none());
3146	tasks.shutdown();
3147	Ok(())
3148	}
3149
3150	#[tokio::test]
3151	async fn reconciliation_reaches_old_runs_and_runs_of_deleted_definitions() -> Result<()> {
3152	for delete_definition in [false, true] {
3153	let root = tempfile::tempdir()?;
3154	let receipts = root.path().join("executions");
3155	let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
3156	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
3157	let automation = fixture_due_automation(&manager, "history", 1);
3158	let shared = Arc::new(Mutex::new(manager));
3159	let run = run_now_shared(&shared, &automation.id, &tasks).await?;
3160	let id = run.task_id.clone().context("real task id")?;
3161	let task = crate::task_manager::wait_for_terminal_state(
3162	&tasks,
3163	&id,
3164	std::time::Duration::from_secs(5),
3165	)
3166	.await?;
3167	assert_eq!(task.status, TaskStatus::Completed);
3168	{
3169	let manager = shared.lock().await;
3170	for offset in 1..=110 {
3171	let mut newer = new_run_record(
3172	&automation.id,
3173	Utc::now(),
3174	run.created_at + Duration::seconds(offset),
3175	);
3176	newer.status = AutomationRunStatus::Completed;
3177	newer.ended_at = Some(newer.created_at);
3178	manager.save_run(&newer)?;
3179	}
3180	assert!(
3181	manager
3182	.list_runs(&automation.id, Some(100))?
3183	.iter()
3184	.all(\|candidate\| candidate.id != run.id)
3185	);
3186	if delete_definition {
3187	manager.delete_automation(&automation.id)?;
3188	}
3189	}
3190	reconcile_run_statuses_shared(&shared, &tasks).await?;
3191	let manager = shared.lock().await;
3192	let found =
3193	manager.get_runs_by_ids(&automation.id, &[run.id.clone()].into_iter().collect())?;
3194	assert_eq!(found.len(), 1, "unfinished receipt survives deletion");
3195	assert_eq!(found[0].status, AutomationRunStatus::Completed);
3196	assert_eq!(found[0].task_id.as_deref(), Some(id.as_str()));
3197	assert_eq!(fixture_executions(&receipts), vec![id]);
3198	if delete_definition {
3199	assert!(manager.get_automation(&automation.id).is_err());
3200	} else {
3201	assert!(
3202	manager
3203	.get_automation(&automation.id)?
3204	.last_run_at
3205	.is_some()
3206	);
3207	}
3208	tasks.shutdown();
3209	}
3210	Ok(())
3211	}
3212
3213	#[tokio::test]
3214	async fn foreign_store_and_missing_accepted_task_preserve_binding_without_fallback()
3215	-> Result<()> {
3216	let root = tempfile::tempdir()?;
3217	let receipts = root.path().join("executions");
3218	let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
3219	let other_receipts = root.path().join("foreign-executions");
3220	let other_tasks =
3221	fixture_tasks(&root.path().join("foreign-tasks"), &other_receipts).await?;
3222	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
3223	let automation = fixture_due_automation(&manager, "bound-store", 1);
3224	let proposed = manager.collect_due_runs(Utc::now())?.remove(0);
3225	let mut run = manager
3226	.claim_scheduled_run(&proposed.0, proposed.1, &tasks.data_dir())?
3227	.context("claim")?;
3228	let id = run.task_id.clone().context("bound task")?;
3229	let shared = Arc::new(Mutex::new(manager));
3230	scheduler_tick_shared(&shared, &other_tasks).await?;
3231	let rows = shared.lock().await.list_runs(&automation.id, None)?;
3232	assert_eq!(rows[0].task_id.as_deref(), Some(id.as_str()));
3233	assert!(
3234	rows[0]
3235	.error
3236	.as_deref()
3237	.unwrap_or_default()
3238	.contains("different task store")
3239	);
3240	assert!(fixture_executions(&other_receipts).is_empty());
3241	enqueue_run_task(&mut run, &tasks).await;
3242	let task = crate::task_manager::wait_for_terminal_state(
3243	&tasks,
3244	&id,
3245	std::time::Duration::from_secs(5),
3246	)
3247	.await?;
3248	assert_eq!(task.status, TaskStatus::Completed);
3249	assert!(run.dispatch.as_ref().unwrap().accepted);
3250	shared.lock().await.finish_scheduled_run(&run, Utc::now())?;
3251	fs::remove_file(tasks.data_dir().join("tasks").join(format!("{id}.json")))?;
3252	scheduler_tick_shared(&shared, &tasks).await?;
3253	reconcile_run_statuses_shared(&shared, &tasks).await?;
3254	let rows = shared.lock().await.list_runs(&automation.id, None)?;
3255	assert_eq!(rows[0].task_id.as_deref(), Some(id.as_str()));
3256	assert!(rows[0].dispatch.as_ref().unwrap().accepted);
3257	assert!(
3258	rows[0]
3259	.error
3260	.as_deref()
3261	.unwrap_or_default()
3262	.contains("missing")
3263	);
3264	assert_eq!(fixture_executions(&receipts), vec![id]);
3265	assert!(fixture_executions(&other_receipts).is_empty());
3266	tasks.shutdown();
3267	other_tasks.shutdown();
3268	Ok(())
3269	}
3270
3271	#[tokio::test]
3272	async fn canceled_collected_trigger_is_not_admitted_after_an_earlier_enqueue_wait() -> Result<()>
3273	{
3274	let root = tempfile::tempdir()?;
3275	let receipts = root.path().join("executions");
3276	let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
3277	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
3278	let mut triggers = Vec::new();
3279	for index in 0..2 {
3280	let mut trigger = manager.create_trigger(CreateDelayedTriggerRequest {
3281	fire_at: Utc::now() + Duration::hours(1),
3282	message: format!("trigger fixture {index}"),
3283	workspace: None,
3284	owner_session_id: Some("fixture-owner".into()),
3285	parent_trigger_id: None,
3286	})?;
3287	trigger.fire_at = Utc::now() - Duration::minutes(1);
3288	trigger.created_at = Utc::now() - Duration::minutes(index);
3289	manager.save_trigger(&trigger)?;
3290	triggers.push(trigger);
3291	}
3292	let shared = Arc::new(Mutex::new(manager));
3293	let (entered, reached) = tokio::sync::oneshot::channel();
3294	let (release, released) = tokio::sync::oneshot::channel();
3295	let tick = tokio::spawn({
3296	let shared = shared.clone();
3297	let tasks = tasks.clone();
3298	let mut barrier = Some((entered, released));
3299	async move {
3300	fire_due_triggers_with(&shared, &tasks.data_dir(), move \|trigger\| {
3301	let tasks = tasks.clone();
3302	let barrier = barrier.take();
3303	async move {
3304	if let Some((entered, released)) = barrier {
3305	let _ = entered.send(trigger.clone());
3306	let _ = released.await;
3307	}
3308	enqueue_trigger_task(trigger, &tasks).await
3309	}
3310	})
3311	.await
3312	}
3313	});
3314	let admitted = tokio::time::timeout(std::time::Duration::from_secs(5), reached).await??;
3315	assert_eq!(admitted.trigger_id, triggers[0].trigger_id);
3316	assert_eq!(admitted.status, DelayedTriggerStatus::Dispatching);
3317	let other = AutomationManager::open_for_test(root.path().join("automations"))?;
3318	other.cancel_trigger_for_owner(&triggers[1].trigger_id, "fixture-owner")?;
3319	assert!(
3320	other
3321	.cancel_trigger_for_owner(&admitted.trigger_id, "fixture-owner")
3322	.is_err(),
3323	"claimed work has crossed the admission boundary"
3324	);
3325	release
3326	.send(())
3327	.map_err(\|_\| anyhow::anyhow!("release trigger fixture"))?;
3328	tokio::time::timeout(std::time::Duration::from_secs(5), tick).await???;
3329	let id = admitted.task_id.context("trigger task identity")?;
3330	let task = crate::task_manager::wait_for_terminal_state(
3331	&tasks,
3332	&id,
3333	std::time::Duration::from_secs(5),
3334	)
3335	.await?;
3336	assert_eq!(task.status, TaskStatus::Completed);
3337	assert_eq!(task.owner_session_id.as_deref(), Some("fixture-owner"));
3338	assert!(!task.allow_shell && !task.trust_mode && !task.auto_approve);
3339	assert_eq!(fixture_executions(&receipts), vec![id]);
3340	assert_eq!(
3341	other.get_trigger(&triggers[0].trigger_id)?.status,
3342	DelayedTriggerStatus::Fired
3343	);
3344	let canceled = other.get_trigger(&triggers[1].trigger_id)?;
3345	assert_eq!(canceled.status, DelayedTriggerStatus::Canceled);
3346	assert!(canceled.task_id.is_none());
3347	tasks.shutdown();
3348	Ok(())
3349	}
3350
3351	async fn wait_fixture_path(path: &Path) -> Result<()> {
3352	let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
3353	while !path.try_exists()? {
3354	if std::time::Instant::now() >= deadline {
3355	bail!("fixture barrier timed out: {}", path.display());
3356	}
3357	tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3358	}
3359	Ok(())
3360	}
3361
3362	struct SchedulerFixtureChild(std::process::Child);
3363
3364	impl Drop for SchedulerFixtureChild {
3365	fn drop(&mut self) {
3366	let _ = self.0.kill();
3367	let _ = self.0.wait();
3368	}
3369	}
3370
3371	fn spawn_scheduler_fixture(root: &Path, role: &str) -> Result<SchedulerFixtureChild> {
3372	let log = fs::File::create(root.join(format!("{role}.log")))?;
3373	let home = root.join(format!("{role}-home"));
3374	fs::create_dir_all(&home)?;
3375	Ok(SchedulerFixtureChild(
3376	std::process::Command::new(std::env::current_exe()?)
3377	.args([
3378	"--ignored",
3379	"--exact",
3380	"automation_manager::tests::scheduler_process_fixture",
3381	"--nocapture",
3382	"--test-threads=1",
3383	])
3384	.env("CW_AUTOMATION_PROCESS_FIXTURE", root)
3385	.env("CW_AUTOMATION_PROCESS_ROLE", role)
3386	.env("CODEWHALE_HOME", &home)
3387	.env("HOME", &home)
3388	.env("USERPROFILE", &home)
3389	.stdin(std::process::Stdio::null())
3390	.stdout(std::process::Stdio::from(log.try_clone()?))
3391	.stderr(std::process::Stdio::from(log))
3392	.spawn()?,
3393	))
3394	}
3395
3396	async fn finish_scheduler_fixture(child: &mut SchedulerFixtureChild, log: &Path) -> Result<()> {
3397	let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
3398	loop {
3399	if let Some(status) = child.0.try_wait()? {
3400	if !status.success() {
3401	bail!("fixture process failed: {}", fs::read_to_string(log)?);
3402	}
3403	return Ok(());
3404	}
3405	if std::time::Instant::now() >= deadline {
3406	bail!("fixture process timed out: {}", log.display());
3407	}
3408	tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3409	}
3410	}
3411
3412	/// Only the parent test launches this entry point with explicit temporary
3413	/// stores. Missing fixture parameters fail; this helper cannot pass by skip.
3414	#[tokio::test]
3415	#[ignore = "subprocess entry point for independent scheduler contention fixture"]
3416	async fn scheduler_process_fixture() -> Result<()> {
3417	let root = PathBuf::from(
3418	std::env::var_os("CW_AUTOMATION_PROCESS_FIXTURE").context("required fixture root")?,
3419	);
3420	let role = std::env::var("CW_AUTOMATION_PROCESS_ROLE")?;
3421	if !matches!(role.as_str(), "first" \| "second") {
3422	bail!("invalid fixture role");
3423	}
3424	let scope = if role == "first" {
3425	"test"
3426	} else {
3427	"foreign-scheduler"
3428	};
3429	let tasks = TaskManager::start_with_executor_in_scope(
3430	automation_task_config(root.join("tasks")),
3431	Arc::new(AutomationRecordingExecutor(root.join("executions"))),
3432	scope,
3433	)
3434	.await?;
3435	let mut service = AutomationManager::open(root.join("automations"))?;
3436	service.bind_task_manager(&tasks)?;
3437	let shared = Arc::new(Mutex::new(service));
3438	crate::utils::write_atomic(&root.join(format!("{role}-ready")), b"ready")?;
3439	wait_fixture_path(&root.join(format!("{role}-go"))).await?;
3440	let observations = Arc::new(std::sync::Mutex::new(Vec::new()));
3441	scheduler_tick_with(&shared, &tasks.data_dir(), {
3442	let tasks = tasks.clone();
3443	let root = root.clone();
3444	let role = role.clone();
3445	let observations = observations.clone();
3446	move \|mut run\| {
3447	let tasks = tasks.clone();
3448	let root = root.clone();
3449	let role = role.clone();
3450	let observations = observations.clone();
3451	async move {
3452	let id = run.task_id.clone().expect("durably claimed fixture id");
3453	observations.lock().unwrap().push(id.clone());
3454	crate::utils::write_atomic(
3455	&root.join(format!("{role}-entered")),
3456	id.as_bytes(),
3457	)
3458	.expect("record dispatch entry");
3459	if role == "first" {
3460	crate::utils::write_atomic(&root.join("first-held"), id.as_bytes())
3461	.expect("record claim barrier");
3462	wait_fixture_path(&root.join("release-first"))
3463	.await
3464	.expect("release owner");
3465	}
3466	enqueue_run_task(&mut run, &tasks).await;
3467	run
3468	}
3469	}
3470	})
3471	.await?;
3472	let ids = observations.lock().unwrap().clone();
3473	for id in &ids {
3474	let task = crate::task_manager::wait_for_terminal_state(
3475	&tasks,
3476	id,
3477	std::time::Duration::from_secs(5),
3478	)
3479	.await?;
3480	assert_eq!(task.status, TaskStatus::Completed);
3481	}
3482	reconcile_run_statuses_shared(&shared, &tasks).await?;
3483	crate::utils::write_atomic(
3484	&root.join(format!("{role}-done")),
3485	serde_json::to_vec(&ids)?.as_slice(),
3486	)?;
3487	tasks.shutdown();
3488	Ok(())
3489	}
3490
3491	#[tokio::test]
3492	async fn two_scheduler_processes_preserve_the_bound_scope_and_execute_one_task() -> Result<()> {
3493	let root = tempfile::tempdir()?;
3494	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
3495	let automation = fixture_due_automation(&manager, "cross-process", 1);
3496	let mut first = spawn_scheduler_fixture(root.path(), "first")?;
3497	let mut second = spawn_scheduler_fixture(root.path(), "second")?;
3498	// Separate Runtime scopes share storage. The foreign scheduler cannot
3499	// adopt the first scope's immutable dispatch even before acceptance.
3500	wait_fixture_path(&root.path().join("first-ready")).await?;
3501	wait_fixture_path(&root.path().join("second-ready")).await?;
3502	crate::utils::write_atomic(&root.path().join("first-go"), b"go")?;
3503	wait_fixture_path(&root.path().join("first-held")).await?;
3504	let id = fs::read_to_string(root.path().join("first-held"))?;
3505	crate::utils::write_atomic(&root.path().join("second-go"), b"go")?;
3506	wait_fixture_path(&root.path().join("second-done")).await?;
3507	finish_scheduler_fixture(&mut second, &root.path().join("second.log")).await?;
3508	assert!(
3509	!root.path().join("second-entered").exists(),
3510	"a live dispatcher still owns the unaccepted intent"
3511	);
3512	assert!(
3513	fixture_executions(&root.path().join("executions")).is_empty(),
3514	"owner is blocked before actual enqueue"
3515	);
3516	crate::utils::write_atomic(&root.path().join("release-first"), b"release")?;
3517	finish_scheduler_fixture(&mut first, &root.path().join("first.log")).await?;
3518	assert_eq!(
3519	fixture_executions(&root.path().join("executions")),
3520	vec![id.clone()]
3521	);
3522	let entered: Vec<String> =
3523	serde_json::from_slice(&fs::read(root.path().join("first-done"))?)?;
3524	assert_eq!(
3525	entered,
3526	vec![id.clone()],
3527	"first process traversed the real enqueue path"
3528	);
3529	let runs = manager.list_runs(&automation.id, None)?;
3530	assert_eq!(runs.len(), 1);
3531	assert_eq!(runs[0].task_id.as_deref(), Some(id.as_str()));
3532	assert_eq!(runs[0].status, AutomationRunStatus::Completed);
3533	assert!(runs[0].dispatch.as_ref().unwrap().accepted);
3534	let task: crate::task_manager::TaskRecord = serde_json::from_slice(&fs::read(
3535	root.path().join("tasks/tasks").join(format!("{id}.json")),
3536	)?)?;
3537	assert_eq!(task.id, id);
3538	assert_eq!(task.status, TaskStatus::Completed);
3539	assert!(
3540	task.result_summary
3541	.as_deref()
3542	.unwrap_or_default()
3543	.contains("automation fixture completed")
3544	);
3545	Ok(())
3546	}
3547
3548	struct AutomationNoopExecutor;
3549	struct AutomationWatcherNoopExecutor;
3550
3551	/// A deterministic America/New_York-compatible zone for the 2026 DST
3552	/// boundary tests. Keeping the transition table local avoids mutating the
3553	/// process-wide `TZ` setting while the test binary runs in parallel.
3554	#[derive(Debug, Clone, Copy)]
3555	struct Eastern2026;
3556
3557	impl Eastern2026 {
3558	fn standard_offset() -> FixedOffset {
3559	FixedOffset::west_opt(5 * 60 * 60).expect("valid standard offset")
3560	}
3561
3562	fn daylight_offset() -> FixedOffset {
3563	FixedOffset::west_opt(4 * 60 * 60).expect("valid daylight offset")
3564	}
3565
3566	fn time(month: u32, day: u32, hour: u32) -> NaiveDateTime {
3567	NaiveDate::from_ymd_opt(2026, month, day)
3568	.expect("valid transition date")
3569	.and_hms_opt(hour, 0, 0)
3570	.expect("valid transition time")
3571	}
3572	}
3573
3574	impl TimeZone for Eastern2026 {
3575	type Offset = FixedOffset;
3576
3577	fn from_offset(_offset: &Self::Offset) -> Self {
3578	Self
3579	}
3580
3581	fn offset_from_local_date(&self, local: &NaiveDate) -> LocalResult<Self::Offset> {
3582	self.offset_from_local_datetime(
3583	&local
3584	.and_hms_opt(12, 0, 0)
3585	.expect("valid local date midpoint"),
3586	)
3587	}
3588
3589	fn offset_from_local_datetime(&self, local: &NaiveDateTime) -> LocalResult<Self::Offset> {
3590	let gap_start = Self::time(3, 8, 2);
3591	let gap_end = Self::time(3, 8, 3);
3592	let fold_start = Self::time(11, 1, 1);
3593	let fold_end = Self::time(11, 1, 2);
3594
3595	if local >= gap_start && local < gap_end {
3596	LocalResult::None
3597	} else if local >= fold_start && local < fold_end {
3598	LocalResult::Ambiguous(Self::daylight_offset(), Self::standard_offset())
3599	} else if local >= gap_end && local < fold_start {
3600	LocalResult::Single(Self::daylight_offset())
3601	} else {
3602	LocalResult::Single(Self::standard_offset())
3603	}
3604	}
3605
3606	fn offset_from_utc_date(&self, utc: &NaiveDate) -> Self::Offset {
3607	self.offset_from_utc_datetime(
3608	&utc.and_hms_opt(12, 0, 0).expect("valid UTC date midpoint"),
3609	)
3610	}
3611
3612	fn offset_from_utc_datetime(&self, utc: &NaiveDateTime) -> Self::Offset {
3613	let daylight_start = Self::time(3, 8, 7);
3614	let daylight_end = Self::time(11, 1, 6);
3615	if utc >= daylight_start && utc < daylight_end {
3616	Self::daylight_offset()
3617	} else {
3618	Self::standard_offset()
3619	}
3620	}
3621	}
3622
3623	#[async_trait]
3624	impl TaskExecutor for AutomationNoopExecutor {
3625	async fn execute(
3626	&self,
3627	_task: ExecutionTask,
3628	_events: mpsc::Sender<TaskExecutionEvent>,
3629	_cancel: CancellationToken,
3630	) -> TaskExecutionResult {
3631	TaskExecutionResult {
3632	status: TaskStatus::Completed,
3633	result_text: Some("done".to_string()),
3634	error: None,
3635	terminal_reason: crate::task_manager::TaskTerminalReason::Completed,
3636	}
3637	}
3638	}
3639
3640	#[async_trait]
3641	impl TaskExecutor for AutomationWatcherNoopExecutor {
3642	async fn execute(
3643	&self,
3644	_task: ExecutionTask,
3645	_events: mpsc::Sender<TaskExecutionEvent>,
3646	_cancel: CancellationToken,
3647	) -> TaskExecutionResult {
3648	TaskExecutionResult {
3649	status: TaskStatus::Completed,
3650	result_text: Some(AUTOMATION_WATCHER_NO_REPORT_SENTINEL.to_string()),
3651	error: None,
3652	terminal_reason: crate::task_manager::TaskTerminalReason::Completed,
3653	}
3654	}
3655	}
3656
3657	fn automation_task_config(root: PathBuf) -> TaskManagerConfig {
3658	TaskManagerConfig {
3659	data_dir: root,
3660	worker_count: 1,
3661	default_workspace: PathBuf::from("."),
3662	default_model: "deepseek-v4-flash".to_string(),
3663	default_mode: "plan".to_string(),
3664	allow_shell: true,
3665	trust_mode: true,
3666	execution_limits: crate::task_manager::TaskExecutionLimits::default(),
3667	}
3668	}
3669
3670	fn automation_record_with_settings(
3671	mode: Option<&str>,
3672	allow_shell: Option<bool>,
3673	trust_mode: Option<bool>,
3674	auto_approve: Option<bool>,
3675	) -> AutomationRecord {
3676	let now = Utc::now();
3677	AutomationRecord {
3678	schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION,
3679	execution_scope: Some(crate::task_manager::test_execution_scope("test")),
3680	id: Uuid::new_v4().to_string(),
3681	name: "Test automation".to_string(),
3682	prompt: "Run the automation".to_string(),
3683	rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
3684	cwds: Vec::new(),
3685	model: None,
3686	model_provider: None,
3687	model_provider_id: None,
3688	mode: mode.map(ToString::to_string),
3689	allow_shell,
3690	trust_mode,
3691	auto_approve,
3692	delivery_mode: None,
3693	status: AutomationStatus::Active,
3694	created_at: now,
3695	updated_at: now,
3696	next_run_at: None,
3697	last_run_at: None,
3698	}
3699	}
3700
3701	fn queued_run_for(automation: &AutomationRecord) -> AutomationRunRecord {
3702	let now = Utc::now();
3703	AutomationRunRecord {
3704	schema_version: CURRENT_RUN_SCHEMA_VERSION,
3705	id: Uuid::new_v4().to_string(),
3706	automation_id: automation.id.clone(),
3707	scheduled_for: now,
3708	status: AutomationRunStatus::Queued,
3709	created_at: now,
3710	started_at: None,
3711	ended_at: None,
3712	task_id: None,
3713	thread_id: None,
3714	turn_id: None,
3715	error: None,
3716	dispatch: None,
3717	}
3718	}
3719
3720	fn eastern_datetime(year: i32, month: u32, day: u32, hour: u32, minute: u32) -> DateTime<Utc> {
3721	Eastern2026
3722	.with_ymd_and_hms(year, month, day, hour, minute, 0)
3723	.single()
3724	.expect("unambiguous Eastern wall time")
3725	.with_timezone(&Utc)
3726	}
3727
3728	fn anchored_automation(
3729	created_at: DateTime<Utc>,
3730	status: AutomationStatus,
3731	) -> AutomationRecord {
3732	let mut record = automation_record_with_settings(None, None, None, None);
3733	record.rrule = "FREQ=HOURLY;INTERVAL=7;BYMINUTE=17".to_string();
3734	record.status = status;
3735	record.created_at = created_at;
3736	record.updated_at = created_at;
3737	record.next_run_at = None;
3738	record
3739	}
3740
3741	fn local_naive_to_utc(naive: NaiveDateTime) -> DateTime<Utc> {
3742	Local
3743	.from_local_datetime(&naive)
3744	.earliest()
3745	.expect("valid unambiguous local time")
3746	.with_timezone(&Utc)
3747	}
3748
3749	#[test]
3750	fn parses_hourly_rrule() {
3751	let parsed =
3752	AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=2;BYDAY=MO,TU").expect("parse");
3753	match parsed {
3754	AutomationSchedule::Hourly {
3755	interval_hours,
3756	byday,
3757	..
3758	} => {
3759	assert_eq!(interval_hours, 2);
3760	assert_eq!(byday.expect("byday").len(), 2);
3761	}
3762	_ => panic!("expected hourly"),
3763	}
3764	}
3765
3766	#[test]
3767	fn parses_once_rrule() {
3768	let parsed =
3769	AutomationSchedule::parse_rrule("FREQ=ONCE;AT=2026-08-03T14:30").expect("parse");
3770	match parsed {
3771	AutomationSchedule::Once { at } => {
3772	assert_eq!(
3773	at,
3774	local_naive_to_utc(
3775	NaiveDateTime::parse_from_str("2026-08-03T14:30", "%Y-%m-%dT%H:%M")
3776	.expect("naive")
3777	)
3778	);
3779	}
3780	_ => panic!("expected once"),
3781	}
3782	}
3783
3784	#[test]
3785	fn parses_hourly_clock_anchor() {
3786	let parsed =
3787	AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30")
3788	.expect("parse anchored hourly schedule");
3789
3790	assert!(matches!(
3791	parsed,
3792	AutomationSchedule::Hourly {
3793	anchor_hour: Some(8),
3794	anchor_minute: Some(30),
3795	..
3796	}
3797	));
3798
3799	let minute_only = AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=1;BYMINUTE=15")
3800	.expect("parse minute-only anchor");
3801	assert!(matches!(
3802	minute_only,
3803	AutomationSchedule::Hourly {
3804	anchor_hour: None,
3805	anchor_minute: Some(15),
3806	..
3807	}
3808	));
3809	}
3810
3811	#[test]
3812	fn anchored_hourly_schedule_keeps_wall_time_across_spring_forward() {
3813	let schedule =
3814	AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30")
3815	.expect("parse");
3816	let created_at = eastern_datetime(2026, 3, 6, 7, 0);
3817	let after = eastern_datetime(2026, 3, 7, 9, 0);
3818
3819	let next = schedule
3820	.next_after_in_timezone(after, created_at, &Eastern2026)
3821	.expect("next run");
3822
3823	assert_eq!(next, eastern_datetime(2026, 3, 8, 8, 30));
3824	}
3825
3826	#[test]
3827	fn anchored_hourly_schedule_keeps_wall_time_across_fall_back() {
3828	let schedule =
3829	AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30")
3830	.expect("parse");
3831	let created_at = eastern_datetime(2026, 10, 30, 7, 0);
3832	let after = eastern_datetime(2026, 10, 31, 9, 0);
3833
3834	let next = schedule
3835	.next_after_in_timezone(after, created_at, &Eastern2026)
3836	.expect("next run");
3837
3838	assert_eq!(next, eastern_datetime(2026, 11, 1, 8, 30));
3839	}
3840
3841	#[test]
3842	fn anchored_hourly_schedule_skips_nonexistent_wall_time() {
3843	let schedule =
3844	AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=2;BYMINUTE=30")
3845	.expect("parse");
3846	let created_at = eastern_datetime(2026, 3, 7, 1, 0);
3847	let after = eastern_datetime(2026, 3, 7, 3, 0);
3848
3849	let next = schedule
3850	.next_after_in_timezone(after, created_at, &Eastern2026)
3851	.expect("next run after spring-forward gap");
3852
3853	assert_eq!(next, eastern_datetime(2026, 3, 9, 2, 30));
3854	}
3855
3856	#[test]
3857	fn anchored_hourly_schedule_uses_first_ambiguous_wall_time_once() {
3858	let schedule =
3859	AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=1;BYMINUTE=30")
3860	.expect("parse");
3861	let created_at = eastern_datetime(2026, 10, 31, 0, 0);
3862	let after = eastern_datetime(2026, 10, 31, 2, 0);
3863	let first_fold_occurrence = Eastern2026
3864	.with_ymd_and_hms(2026, 11, 1, 1, 30, 0)
3865	.earliest()
3866	.expect("first fold occurrence")
3867	.with_timezone(&Utc);
3868
3869	let next = schedule
3870	.next_after_in_timezone(after, created_at, &Eastern2026)
3871	.expect("next run at fall-back fold");
3872	assert_eq!(next, first_fold_occurrence);
3873
3874	let during_second_fold = Eastern2026
3875	.with_ymd_and_hms(2026, 11, 1, 1, 15, 0)
3876	.latest()
3877	.expect("second fold occurrence")
3878	.with_timezone(&Utc);
3879	let after_fold = schedule
3880	.next_after_in_timezone(during_second_fold, created_at, &Eastern2026)
3881	.expect("next run after fold");
3882	assert_eq!(after_fold, eastern_datetime(2026, 11, 2, 1, 30));
3883	}
3884
3885	#[test]
3886	fn anchored_hourly_schedule_reuses_persisted_anchor_after_restart_and_resume() {
3887	let rrule = "FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30";
3888	let created_at = eastern_datetime(2026, 3, 6, 7, 0);
3889	let schedule = AutomationSchedule::parse_rrule(rrule).expect("parse");
3890	let before_restart = schedule
3891	.next_after_in_timezone(
3892	eastern_datetime(2026, 3, 7, 12, 0),
3893	created_at,
3894	&Eastern2026,
3895	)
3896	.expect("next before restart");
3897	assert_eq!(before_restart, eastern_datetime(2026, 3, 8, 8, 30));
3898
3899	// Reparsing models a process restart; the persisted creation timestamp
3900	// remains the recurrence anchor when the record is loaded or resumed.
3901	let restarted = AutomationSchedule::parse_rrule(rrule).expect("reparse after restart");
3902	let after_restart = restarted
3903	.next_after_in_timezone(
3904	eastern_datetime(2026, 3, 8, 10, 0),
3905	created_at,
3906	&Eastern2026,
3907	)
3908	.expect("next after restart");
3909	assert_eq!(after_restart, eastern_datetime(2026, 3, 9, 8, 30));
3910
3911	let after_resume = restarted
3912	.next_after_in_timezone(
3913	eastern_datetime(2026, 3, 10, 12, 0),
3914	created_at,
3915	&Eastern2026,
3916	)
3917	.expect("next after resume");
3918	assert_eq!(after_resume, eastern_datetime(2026, 3, 11, 8, 30));
3919	}
3920
3921	#[test]
3922	fn scheduler_restart_uses_persisted_creation_anchor() {
3923	let tempdir = tempfile::tempdir().expect("tempdir");
3924	let now = Utc::now();
3925	let created_at = now - Duration::hours(51);
3926	let automation = anchored_automation(created_at, AutomationStatus::Active);
3927	let schedule = AutomationSchedule::parse_rrule(&automation.rrule).expect("parse");
3928	let expected = schedule
3929	.next_after_with_anchor(now, created_at)
3930	.expect("persisted-anchor schedule");
3931	let reset_anchor = schedule
3932	.next_after_with_anchor(now, now)
3933	.expect("reset-anchor schedule");
3934	assert_ne!(expected, reset_anchor, "fixture must detect anchor resets");
3935
3936	let manager =
3937	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
3938	manager.save_automation(&automation).expect("save");
3939	drop(manager);
3940
3941	let restarted =
3942	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("reopen");
3943	assert!(
3944	restarted
3945	.collect_due_runs(now)
3946	.expect("restart tick")
3947	.is_empty(),
3948	"an uninitialized future slot must not enqueue immediately"
3949	);
3950	let reloaded = restarted
3951	.get_automation(&automation.id)
3952	.expect("reloaded automation");
3953	assert_eq!(reloaded.next_run_at, Some(expected));
3954	}
3955
3956	#[test]
3957	fn resume_uses_persisted_creation_anchor() {
3958	let tempdir = tempfile::tempdir().expect("tempdir");
3959	let manager =
3960	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
3961	let before = Utc::now();
3962	let created_at = before - Duration::hours(51);
3963	let automation = anchored_automation(created_at, AutomationStatus::Paused);
3964	let schedule = AutomationSchedule::parse_rrule(&automation.rrule).expect("parse");
3965	manager.save_automation(&automation).expect("save");
3966
3967	let expected_before = schedule
3968	.next_after_with_anchor(before, created_at)
3969	.expect("next before resume");
3970	let reset_anchor = schedule
3971	.next_after_with_anchor(before, before)
3972	.expect("reset-anchor schedule");
3973	assert_ne!(
3974	expected_before, reset_anchor,
3975	"fixture must detect anchor resets"
3976	);
3977
3978	let resumed = manager
3979	.resume_automation(&automation.id)
3980	.expect("resume automation");
3981	let after = Utc::now();
3982	let expected_after = schedule
3983	.next_after_with_anchor(after, created_at)
3984	.expect("next after resume");
3985	let actual = resumed.next_run_at.expect("resumed next run");
3986	assert!(
3987	actual == expected_before \|\| actual == expected_after,
3988	"resume must keep the persisted creation anchor"
3989	);
3990	}
3991
3992	#[test]
3993	fn anchored_hourly_schedule_applies_byday_on_calendar_slots() {
3994	let schedule = AutomationSchedule::parse_rrule(
3995	"FREQ=HOURLY;INTERVAL=24;BYDAY=MO,TU,WE,TH,FR;BYHOUR=8;BYMINUTE=30",
3996	)
3997	.expect("parse");
3998	let created_at = eastern_datetime(2026, 3, 6, 7, 0);
3999
4000	let next = schedule
4001	.next_after_in_timezone(eastern_datetime(2026, 3, 6, 9, 0), created_at, &Eastern2026)
4002	.expect("next weekday run");
4003
4004	assert_eq!(next, eastern_datetime(2026, 3, 9, 8, 30));
4005	}
4006
4007	#[test]
4008	fn parses_weekly_rrule() {
4009	let parsed =
4010	AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYDAY=MO,WE;BYHOUR=9;BYMINUTE=30")
4011	.expect("parse");
4012	match parsed {
4013	AutomationSchedule::Weekly {
4014	byday,
4015	byhour,
4016	byminute,
4017	} => {
4018	assert_eq!(byday.len(), 2);
4019	assert_eq!(byhour, 9);
4020	assert_eq!(byminute, 30);
4021	}
4022	_ => panic!("expected weekly"),
4023	}
4024	}
4025
4026	#[test]
4027	fn parses_cron_rrule_and_computes_next_minute_slot() {
4028	let schedule =
4029	AutomationSchedule::parse_rrule("FREQ=CRON;EXPR=/17 * * *").expect("parse");
4030	let after = Utc
4031	.with_ymd_and_hms(2026, 8, 3, 9, 17, 1)
4032	.single()
4033	.expect("after");
4034	let next = schedule
4035	.next_after_in_timezone(after, after, &Utc)
4036	.expect("next cron run");
4037	assert_eq!(
4038	next,
4039	Utc.with_ymd_and_hms(2026, 8, 3, 9, 34, 0)
4040	.single()
4041	.expect("next")
4042	);
4043	}
4044
4045	#[test]
4046	fn cron_weekday_schedule_uses_standard_five_field_local_time() {
4047	let schedule =
4048	AutomationSchedule::parse_rrule("FREQ=CRON;EXPR=3 9 * * MON-FRI").expect("parse");
4049	let after = Utc
4050	.with_ymd_and_hms(2026, 8, 7, 9, 4, 0)
4051	.single()
4052	.expect("after");
4053	let next = schedule
4054	.next_after_in_timezone(after, after, &Utc)
4055	.expect("next weekday cron run");
4056	assert_eq!(
4057	next,
4058	Utc.with_ymd_and_hms(2026, 8, 10, 9, 3, 0)
4059	.single()
4060	.expect("next")
4061	);
4062	}
4063
4064	#[test]
4065	fn cron_rejects_impossible_date() {
4066	let err = AutomationSchedule::parse_rrule("FREQ=CRON;EXPR=0 9 31 2 *")
4067	.expect_err("impossible february date must fail");
4068	assert!(err.to_string().contains("can never occur"));
4069	}
4070
4071	#[test]
4072	fn rejects_invalid_rrule_fields() {
4073	let err =
4074	AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYSECOND=5").expect_err("should fail");
4075	assert!(err.to_string().contains("Unsupported RRULE field"));
4076	}
4077
4078	#[test]
4079	fn automation_model_round_trips_through_create_and_update() {
4080	let tempdir = tempfile::tempdir().expect("tempdir");
4081	let manager =
4082	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4083
4084	let created = manager
4085	.create_automation(CreateAutomationRequest {
4086	name: "Pinned model".to_string(),
4087	prompt: "prompt".to_string(),
4088	rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
4089	cwds: Vec::new(),
4090	model: Some(" scheduled-model ".to_string()),
4091	model_provider: None,
4092	model_provider_id: None,
4093	mode: None,
4094	allow_shell: None,
4095	trust_mode: None,
4096	auto_approve: None,
4097	delivery_mode: None,
4098	status: Some(AutomationStatus::Active),
4099	})
4100	.expect("create");
4101	assert_eq!(created.model.as_deref(), Some("scheduled-model"));
4102	assert_eq!(
4103	manager
4104	.get_automation(&created.id)
4105	.expect("reload")
4106	.model
4107	.as_deref(),
4108	Some("scheduled-model")
4109	);
4110
4111	let updated = manager
4112	.update_automation(
4113	&created.id,
4114	UpdateAutomationRequest {
4115	model: Some("replacement-model".to_string()),
4116	..UpdateAutomationRequest::default()
4117	},
4118	)
4119	.expect("update");
4120	assert_eq!(updated.model.as_deref(), Some("replacement-model"));
4121	}
4122
4123	#[test]
4124	fn deletes_definition_and_settled_runs_but_retains_unfinished_receipts() {
4125	let tempdir = tempfile::tempdir().expect("tempdir");
4126	let manager =
4127	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4128
4129	let created = manager
4130	.create_automation(CreateAutomationRequest {
4131	name: "Delete me".to_string(),
4132	prompt: "prompt".to_string(),
4133	rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
4134	cwds: Vec::new(),
4135	model: None,
4136	model_provider: None,
4137	model_provider_id: None,
4138	mode: None,
4139	allow_shell: None,
4140	trust_mode: None,
4141	auto_approve: None,
4142	delivery_mode: None,
4143	status: Some(AutomationStatus::Active),
4144	})
4145	.expect("create");
4146
4147	let run = AutomationRunRecord {
4148	schema_version: CURRENT_RUN_SCHEMA_VERSION,
4149	id: Uuid::new_v4().to_string(),
4150	automation_id: created.id.clone(),
4151	scheduled_for: Utc::now(),
4152	status: AutomationRunStatus::Queued,
4153	created_at: Utc::now(),
4154	started_at: None,
4155	ended_at: None,
4156	task_id: None,
4157	thread_id: None,
4158	turn_id: None,
4159	error: None,
4160	dispatch: None,
4161	};
4162	manager.save_run(&run).expect("save run");
4163	let settled = AutomationRunRecord {
4164	id: Uuid::new_v4().to_string(),
4165	status: AutomationRunStatus::Completed,
4166	ended_at: Some(Utc::now()),
4167	..run.clone()
4168	};
4169	manager.save_run(&settled).expect("save settled run");
4170	assert!(
4171	manager
4172	.runs_dir_for(&created.id)
4173	.expect("runs dir")
4174	.exists()
4175	);
4176
4177	manager
4178	.delete_automation(&created.id)
4179	.expect("delete automation");
4180
4181	assert!(manager.get_automation(&created.id).is_err());
4182	let remaining = manager
4183	.list_runs(&created.id, None)
4184	.expect("retained receipts");
4185	assert_eq!(remaining.len(), 1);
4186	assert_eq!(remaining[0].id, run.id);
4187	}
4188
4189	#[test]
4190	fn automation_storage_rejects_traversal_ids() {
4191	let tempdir = tempfile::tempdir().expect("tempdir");
4192	let manager =
4193	AutomationManager::open_for_test(tempdir.path().join("root")).expect("manager");
4194	let escaped_file = tempdir.path().join("escape.json");
4195	let escaped_runs = tempdir.path().join("escape-runs");
4196
4197	let err = manager
4198	.get_automation("../escape")
4199	.expect_err("traversal automation ids must be rejected");
4200	assert!(err.to_string().contains("single path component"));
4201	assert!(!escaped_file.exists());
4202
4203	let err = manager
4204	.list_runs("../escape-runs", None)
4205	.expect_err("traversal run dirs must be rejected");
4206	assert!(err.to_string().contains("single path component"));
4207	assert!(!escaped_runs.exists());
4208
4209	let run = AutomationRunRecord {
4210	schema_version: CURRENT_RUN_SCHEMA_VERSION,
4211	id: "../escape-run".to_string(),
4212	automation_id: Uuid::new_v4().to_string(),
4213	scheduled_for: Utc::now(),
4214	status: AutomationRunStatus::Queued,
4215	created_at: Utc::now(),
4216	started_at: None,
4217	ended_at: None,
4218	task_id: None,
4219	thread_id: None,
4220	turn_id: None,
4221	error: None,
4222	dispatch: None,
4223	};
4224	let err = manager
4225	.save_run(&run)
4226	.expect_err("traversal run ids must be rejected");
4227	assert!(err.to_string().contains("single path component"));
4228	assert!(!tempdir.path().join("escape-run.json").exists());
4229	}
4230
4231	#[test]
4232	fn automation_task_settings_default_for_legacy_records() {
4233	let now = Utc::now().to_rfc3339();
4234	let record: AutomationRecord = serde_json::from_value(serde_json::json!({
4235	"schema_version": CURRENT_AUTOMATION_SCHEMA_VERSION,
4236	"id": Uuid::new_v4().to_string(),
4237	"name": "Legacy automation",
4238	"prompt": "Run legacy automation",
4239	"rrule": "FREQ=HOURLY;INTERVAL=1",
4240	"cwds": [],
4241	"status": "active",
4242	"created_at": now,
4243	"updated_at": now
4244	}))
4245	.expect("legacy automation record should deserialize");
4246
4247	assert_eq!(record.mode, None);
4248	assert_eq!(record.task_mode(), "agent");
4249	assert!(!record.task_allow_shell());
4250	assert!(!record.task_trust_mode());
4251	assert!(!record.task_auto_approve());
4252	assert_eq!(record.delivery_mode(), AutomationDeliveryMode::Task);
4253	}
4254
4255	#[tokio::test]
4256	async fn automation_enqueue_uses_default_and_explicit_task_settings() -> Result<()> {
4257	let tempdir = tempfile::tempdir().expect("tempdir");
4258	let task_manager = TaskManager::start_with_executor(
4259	automation_task_config(tempdir.path().join("tasks")),
4260	std::sync::Arc::new(AutomationNoopExecutor),
4261	)
4262	.await?;
4263
4264	let default_automation = automation_record_with_settings(None, None, None, None);
4265	let mut default_run = queued_run_for(&default_automation);
4266	bind_run_dispatch(
4267	&mut default_run,
4268	&default_automation,
4269	&task_manager.data_dir(),
4270	false,
4271	)?;
4272	enqueue_run_task(&mut default_run, &task_manager).await;
4273	let default_task = task_manager
4274	.get_task(default_run.task_id.as_deref().expect("task id"))
4275	.await?;
4276	assert_eq!(default_task.model, "deepseek-v4-flash");
4277	assert_eq!(default_task.mode, "agent");
4278	assert!(!default_task.allow_shell);
4279	assert!(!default_task.trust_mode);
4280	assert!(!default_task.auto_approve);
4281
4282	let mut explicit_automation =
4283	automation_record_with_settings(Some("plan"), Some(true), Some(true), Some(true));
4284	explicit_automation.model = Some("scheduled-model".to_string());
4285	let mut explicit_run = queued_run_for(&explicit_automation);
4286	bind_run_dispatch(
4287	&mut explicit_run,
4288	&explicit_automation,
4289	&task_manager.data_dir(),
4290	false,
4291	)?;
4292	enqueue_run_task(&mut explicit_run, &task_manager).await;
4293	let explicit_task = task_manager
4294	.get_task(explicit_run.task_id.as_deref().expect("task id"))
4295	.await?;
4296	assert_eq!(explicit_task.model, "scheduled-model");
4297	assert_eq!(explicit_task.mode, "plan");
4298	assert!(explicit_task.allow_shell);
4299	assert!(explicit_task.trust_mode);
4300	assert!(explicit_task.auto_approve);
4301
4302	task_manager.shutdown();
4303	Ok(())
4304	}
4305
4306	#[tokio::test]
4307	async fn automation_provider_pin_survives_active_route_change_and_legacy_inherits() -> Result<()>
4308	{
4309	let _env = crate::test_support::lock_test_env();
4310	let root = tempfile::tempdir()?;
4311	let mut config: crate::config::Config = toml::from_str(
4312	r#"
4313	provider = "first"
4314	[providers.first]
4315	kind = "openai-compatible"
4316	base_url = "http://127.0.0.1:9/first/v1"
4317	api_key = "fixture-first"
4318	model = "private-model"
4319	[providers.second]
4320	kind = "openai-compatible"
4321	base_url = "http://127.0.0.1:9/second/v1"
4322	api_key = "fixture-second"
4323	model = "private-model"
4324	"#,
4325	)?;
4326	let automations = AutomationManager::open_for_test(root.path().join("schedules"))?;
4327	let mut record = automation_record_with_settings(None, None, None, None);
4328	record.schema_version = 1;
4329	record.model = Some("private-model".to_string());
4330	record.model_provider = Some("custom".to_string());
4331	record.model_provider_id = Some("first".to_string());
4332	automations.save_automation(&record)?;
4333	let record = automations.get_automation(&record.id)?;
4334	assert_eq!(
4335	record.schema_version, CURRENT_AUTOMATION_SCHEMA_VERSION,
4336	"a pin cannot be ignored by an older reader"
4337	);
4338
4339	let runtime = crate::runtime_threads::RuntimeThreadManager::open(
4340	config.clone(),
4341	root.path().to_path_buf(),
4342	crate::runtime_threads::RuntimeThreadManagerConfig::from_task_data_dir(
4343	root.path().join("runtime"),
4344	),
4345	)?;
4346	config.provider = Some("second".to_string());
4347	runtime.reload_config(config).await?;
4348	let tasks = TaskManager::start_with_executor(
4349	automation_task_config(root.path().join("tasks")),
4350	std::sync::Arc::new(AutomationNoopExecutor),
4351	)
4352	.await?;
4353	let mut run = queued_run_for(&record);
4354	bind_run_dispatch(&mut run, &record, &tasks.data_dir(), false)?;
4355	enqueue_run_task(&mut run, &tasks).await;
4356	let task = tasks
4357	.get_task(run.task_id.as_deref().expect("task id"))
4358	.await?;
4359	let task: crate::task_manager::TaskRecord =
4360	serde_json::from_slice(&serde_json::to_vec(&task)?)?;
4361	assert_eq!(task.schema_version, 4);
4362	let thread = runtime
4363	.create_thread(ExecutionTask::from(&task).thread_request())
4364	.await?;
4365	assert_eq!(thread.model, "private-model");
4366	assert_eq!(thread.model_provider.as_deref(), Some("custom"));
4367	assert_eq!(thread.model_provider_id.as_deref(), Some("first"));
4368
4369	let mut legacy = serde_json::to_value(&record)?;
4370	let object = legacy.as_object_mut().unwrap();
4371	object.remove("model_provider");
4372	object.remove("model_provider_id");
4373	object.insert("schema_version".to_string(), serde_json::json!(1));
4374	let legacy: AutomationRecord = serde_json::from_value(legacy)?;
4375	assert_eq!(legacy.model_provider, None);
4376	let mut run = queued_run_for(&legacy);
4377	bind_run_dispatch(&mut run, &legacy, &tasks.data_dir(), false)?;
4378	enqueue_run_task(&mut run, &tasks).await;
4379	let task = tasks
4380	.get_task(run.task_id.as_deref().expect("legacy task id"))
4381	.await?;
4382	let mut value = serde_json::to_value(&task)?;
4383	value.as_object_mut().unwrap().remove("model_provider");
4384	value.as_object_mut().unwrap().remove("model_provider_id");
4385	value["schema_version"] = serde_json::json!(2);
4386	let task: crate::task_manager::TaskRecord = serde_json::from_value(value)?;
4387	let thread = runtime
4388	.create_thread(ExecutionTask::from(&task).thread_request())
4389	.await?;
4390	assert_eq!(thread.model_provider_id.as_deref(), Some("second"));
4391	assert_eq!(thread.model, "private-model");
4392	tasks.shutdown();
4393	Ok(())
4394	}
4395
4396	#[tokio::test]
4397	async fn delayed_trigger_fires_task_with_same_owner_and_skips_legacy_ownerless() -> Result<()> {
4398	let tempdir = tempfile::tempdir().expect("tempdir");
4399	let task_manager = TaskManager::start_with_executor(
4400	automation_task_config(tempdir.path().join("tasks")),
4401	std::sync::Arc::new(AutomationNoopExecutor),
4402	)
4403	.await?;
4404	let manager = AutomationManager::open_for_test(tempdir.path().join("automations"))?;
4405
4406	let mut owned = manager.create_trigger(CreateDelayedTriggerRequest {
4407	fire_at: Utc::now() + Duration::hours(1),
4408	message: "owned delayed continuation".to_string(),
4409	workspace: None,
4410	owner_session_id: Some("session-a".to_string()),
4411	parent_trigger_id: None,
4412	})?;
4413	owned.fire_at = Utc::now() - Duration::minutes(1);
4414	manager.save_trigger(&owned)?;
4415
4416	let mut legacy = manager.create_trigger(CreateDelayedTriggerRequest {
4417	fire_at: Utc::now() + Duration::hours(1),
4418	message: "legacy delayed continuation".to_string(),
4419	workspace: None,
4420	owner_session_id: None,
4421	parent_trigger_id: None,
4422	})?;
4423	legacy.fire_at = Utc::now() - Duration::minutes(1);
4424	manager.save_trigger(&legacy)?;
4425
4426	let shared = Arc::new(Mutex::new(manager));
4427	fire_due_triggers_shared(&shared, &task_manager).await?;
4428
4429	let manager = shared.lock().await;
4430	let fired = manager.get_trigger(&owned.trigger_id)?;
4431	assert_eq!(fired.status, DelayedTriggerStatus::Fired);
4432	let task = task_manager
4433	.get_task(fired.task_id.as_deref().expect("fired task id"))
4434	.await?;
4435	assert_eq!(task.owner_session_id.as_deref(), Some("session-a"));
4436
4437	let legacy_after = manager.get_trigger(&legacy.trigger_id)?;
4438	assert_eq!(legacy_after.status, DelayedTriggerStatus::Pending);
4439	assert!(legacy_after.task_id.is_none());
4440	drop(manager);
4441	task_manager.shutdown();
4442	Ok(())
4443	}
4444
4445	#[test]
4446	fn legacy_delayed_trigger_deserializes_without_owner() -> Result<()> {
4447	let now = Utc::now();
4448	let record: DelayedTriggerRecord = serde_json::from_value(serde_json::json!({
4449	"schema_version": CURRENT_TRIGGER_SCHEMA_VERSION,
4450	"trigger_id": "trig_legacy",
4451	"fire_at": (now + Duration::hours(1)).to_rfc3339(),
4452	"message": "legacy trigger",
4453	"status": "pending",
4454	"created_at": now.to_rfc3339(),
4455	"fired_at": null,
4456	"task_id": null,
4457	"thread_id": null,
4458	"error": null,
4459	"parent_trigger_id": null
4460	}))?;
4461	assert_eq!(record.owner_session_id, None);
4462	Ok(())
4463	}
4464
4465	fn write_legacy_run_file(manager: &AutomationManager, run: &AutomationRunRecord) {
4466	let dir = manager.runs_dir_for(&run.automation_id).expect("runs dir");
4467	fs::create_dir_all(&dir).expect("create runs dir");
4468	fs::write(
4469	dir.join(format!("{}.json", run.id)),
4470	serde_json::to_string_pretty(run).expect("serialize run"),
4471	)
4472	.expect("write legacy run");
4473	}
4474
4475	fn run_created_at(
4476	automation: &AutomationRecord,
4477	created_at: DateTime<Utc>,
4478	) -> AutomationRunRecord {
4479	let mut run = queued_run_for(automation);
4480	run.created_at = created_at;
4481	run.scheduled_for = created_at;
4482	run
4483	}
4484
4485	#[test]
4486	fn interrupted_watcher_migration_deduplicates_before_visibility() -> Result<()> {
4487	for suppressed in [false, true] {
4488	let root = tempfile::tempdir()?;
4489	let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
4490	let automation = automation_record_with_settings(None, None, None, None);
4491	let mut run = queued_run_for(&automation);
4492	write_legacy_run_file(&manager, &run);
4493	bind_run_dispatch(&mut run, &automation, root.path(), true)?;
4494	run.status = AutomationRunStatus::Completed;
4495	let dispatch = run.dispatch.as_mut().unwrap();
4496	dispatch.accepted = true;
4497	dispatch.delivery_mode = AutomationDeliveryMode::Watcher;
4498	dispatch.suppress_report = suppressed;
4499	// Persist the post-write/pre-legacy-removal crash boundary.
4500	write_json_atomic(&manager.run_path(&run)?, &run)?;
4501	let ids = std::collections::BTreeSet::from([run.id.clone()]);
4502	for visible in [
4503	manager.list_runs(&automation.id, None)?,
4504	manager.list_runs(&automation.id, Some(1))?,
4505	manager.get_runs_by_ids(&automation.id, &ids)?,
4506	] {
4507	assert_eq!(visible.len(), usize::from(!suppressed));
4508	if let Some(receipt) = visible.first() {
4509	assert_eq!(receipt.status, AutomationRunStatus::Completed);
4510	assert_eq!(receipt.task_id, run.task_id);
4511	}
4512	}
4513	let durable = manager.list_runs_with_visibility(&automation.id, None, true)?;
4514	assert_eq!(durable.len(), 1);
4515	assert_eq!(durable[0].status, AutomationRunStatus::Completed);
4516	assert_eq!(
4517	durable[0].dispatch.as_ref().unwrap().suppress_report,
4518	suppressed
4519	);
4520	assert!(manager.collect_pending_runs()?.is_empty());
4521	}
4522	Ok(())
4523	}
4524
4525	#[test]
4526	fn save_run_uses_sortable_names_and_migrates_legacy_files() {
4527	let tempdir = tempfile::tempdir().expect("tempdir");
4528	let manager =
4529	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4530	let automation = automation_record_with_settings(None, None, None, None);
4531	let run = queued_run_for(&automation);
4532
4533	write_legacy_run_file(&manager, &run);
4534	manager.save_run(&run).expect("save run");
4535
4536	let dir = manager.runs_dir_for(&automation.id).expect("runs dir");
4537	let names: Vec<String> = fs::read_dir(&dir)
4538	.expect("read dir")
4539	.map(\|entry\| {
4540	entry
4541	.expect("entry")
4542	.file_name()
4543	.to_string_lossy()
4544	.into_owned()
4545	})
4546	.collect();
4547	let expected = format!("{}-{}.json", run_file_stamp(run.created_at), run.id);
4548	assert_eq!(names, vec![expected.clone()]);
4549	assert!(has_sortable_run_stem(expected.trim_end_matches(".json")));
4550	// Legacy uuid stems are not mistaken for sortable names.
4551	assert!(!has_sortable_run_stem(&run.id));
4552	}
4553
4554	#[test]
4555	fn finish_scheduled_run_persists_run_when_automation_deleted_mid_enqueue() {
4556	let tempdir = tempfile::tempdir().expect("tempdir");
4557	let manager =
4558	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4559	let automation = automation_record_with_settings(None, None, None, None);
4560	manager.save_automation(&automation).expect("save");
4561	let run = queued_run_for(&automation);
4562
4563	// Simulate the automation being deleted while the enqueue await ran
4564	// outside the lock. The task already exists in the task manager at
4565	// this point, so the run record must still be persisted — an early
4566	// return here orphans a real running task.
4567	manager.delete_automation(&automation.id).expect("delete");
4568	manager
4569	.finish_scheduled_run(&run, Utc::now())
4570	.expect("finish");
4571
4572	let runs = manager.list_runs(&automation.id, None).expect("list runs");
4573	assert_eq!(
4574	runs.iter().map(\|r\| r.id.as_str()).collect::<Vec<_>>(),
4575	vec![run.id.as_str()],
4576	"run must be persisted even though its automation was deleted"
4577	);
4578	assert!(
4579	manager.get_automation(&automation.id).is_err(),
4580	"the deleted automation must not be resurrected"
4581	);
4582	}
4583
4584	#[test]
4585	fn once_schedule_fires_once_and_auto_completes() {
4586	let tempdir = tempfile::tempdir().expect("tempdir");
4587	let manager =
4588	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4589	let due_at = Utc::now() - Duration::minutes(1);
4590	let automation = AutomationRecord {
4591	rrule: due_at
4592	.format("FREQ=ONCE;AT=%Y-%m-%dT%H:%M:%S+00:00")
4593	.to_string(),
4594	next_run_at: Some(due_at),
4595	created_at: due_at - Duration::minutes(5),
4596	updated_at: due_at - Duration::minutes(5),
4597	..automation_record_with_settings(None, None, None, None)
4598	};
4599	manager
4600	.save_automation(&automation)
4601	.expect("save automation");
4602
4603	let due = manager
4604	.collect_due_runs(Utc::now())
4605	.expect("collect due runs");
4606	assert_eq!(due.len(), 1);
4607	let (observed, proposed) = &due[0];
4608	let run = manager
4609	.claim_scheduled_run(observed, proposed.clone(), tempdir.path())
4610	.expect("claim one-shot occurrence")
4611	.expect("due occurrence admitted");
4612	assert_eq!(run.scheduled_for, due_at);
4613
4614	manager
4615	.finish_scheduled_run(&run, Utc::now())
4616	.expect("finish one-shot run");
4617	let updated = manager
4618	.get_automation(&automation.id)
4619	.expect("updated automation");
4620	assert_eq!(updated.status, AutomationStatus::Paused);
4621	assert_eq!(updated.next_run_at, None);
4622	assert!(
4623	manager
4624	.collect_due_runs(Utc::now() + Duration::hours(1))
4625	.expect("later tick")
4626	.is_empty()
4627	);
4628	}
4629
4630	#[test]
4631	fn get_runs_by_ids_finds_live_runs_past_the_newest_window() {
4632	let tempdir = tempfile::tempdir().expect("tempdir");
4633	let manager =
4634	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4635	let automation = automation_record_with_settings(None, None, None, None);
4636	let base = Utc::now();
4637
4638	// The OLDEST run is the long-running task; 25 newer runs stack on
4639	// top of it while it is still Running.
4640	let long_running = run_created_at(&automation, base - Duration::minutes(60));
4641	let mut long_running = long_running;
4642	long_running.status = AutomationRunStatus::Running;
4643	long_running.started_at = Some(base - Duration::minutes(60));
4644	manager.save_run(&long_running).expect("save live run");
4645	for i in 0..25 {
4646	let mut newer = run_created_at(&automation, base - Duration::minutes(30 - i as i64));
4647	newer.status = AutomationRunStatus::Completed;
4648	newer.ended_at = Some(base - Duration::minutes(29 - i as i64));
4649	manager.save_run(&newer).expect("save newer settled run");
4650	}
4651
4652	let window = manager
4653	.list_runs(&automation.id, Some(25))
4654	.expect("windowed list");
4655	assert!(
4656	window.iter().all(\|run\| run.id != long_running.id),
4657	"the live run sits past the newest-25 window"
4658	);
4659
4660	let wanted: std::collections::BTreeSet<String> =
4661	[long_running.id.clone()].into_iter().collect();
4662	let found = manager
4663	.get_runs_by_ids(&automation.id, &wanted)
4664	.expect("re-read live run");
4665	assert_eq!(found.len(), 1, "the run is found wherever it sits");
4666	assert_eq!(found[0].id, long_running.id);
4667	assert_eq!(found[0].status, AutomationRunStatus::Running);
4668	}
4669
4670	#[test]
4671	fn get_runs_by_ids_ignores_non_json_noise() {
4672	let tempdir = tempfile::tempdir().expect("tempdir");
4673	let manager =
4674	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4675	let automation = automation_record_with_settings(None, None, None, None);
4676	let run = queued_run_for(&automation);
4677	manager.save_run(&run).expect("save json run");
4678
4679	let dir = manager.runs_dir_for(&automation.id).expect("runs dir");
4680	let json = dir
4681	.read_dir()
4682	.expect("list")
4683	.filter_map(\|entry\| entry.ok())
4684	.map(\|entry\| entry.path())
4685	.find(\|path\| path.extension().and_then(\|ext\| ext.to_str()) == Some("json"))
4686	.expect("json run file");
4687	let stem = json
4688	.file_stem()
4689	.and_then(\|stem\| stem.to_str())
4690	.expect("stem");
4691	fs::write(dir.join(format!("{stem}.tmp")), "not-json").expect("write tmp noise");
4692	fs::write(dir.join(format!("{stem}.json.bak")), "not-json").expect("write bak noise");
4693
4694	let wanted: std::collections::BTreeSet<String> = [run.id.clone()].into_iter().collect();
4695	let found = manager
4696	.get_runs_by_ids(&automation.id, &wanted)
4697	.expect("noise must not poison the re-read");
4698	assert_eq!(found.len(), 1);
4699	assert_eq!(found[0].id, run.id);
4700	}
4701
4702	#[test]
4703	fn list_runs_merges_legacy_and_sortable_files_newest_first() {
4704	let tempdir = tempfile::tempdir().expect("tempdir");
4705	let manager =
4706	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4707	let automation = automation_record_with_settings(None, None, None, None);
4708	let base = Utc::now();
4709
4710	// Legacy files sit at both ends of the timeline to prove the merge is
4711	// by created_at, not by file-name era.
4712	let legacy_oldest = run_created_at(&automation, base - Duration::minutes(30));
4713	let legacy_newest = run_created_at(&automation, base + Duration::minutes(30));
4714	write_legacy_run_file(&manager, &legacy_oldest);
4715	write_legacy_run_file(&manager, &legacy_newest);
4716
4717	let sortable_old = run_created_at(&automation, base - Duration::minutes(20));
4718	let sortable_new = run_created_at(&automation, base + Duration::minutes(20));
4719	manager.save_run(&sortable_old).expect("save old");
4720	manager.save_run(&sortable_new).expect("save new");
4721
4722	let all = manager.list_runs(&automation.id, None).expect("list all");
4723	let ids: Vec<&str> = all.iter().map(\|run\| run.id.as_str()).collect();
4724	assert_eq!(
4725	ids,
4726	vec![
4727	legacy_newest.id.as_str(),
4728	sortable_new.id.as_str(),
4729	sortable_old.id.as_str(),
4730	legacy_oldest.id.as_str(),
4731	]
4732	);
4733
4734	let top_two = manager.list_runs(&automation.id, Some(2)).expect("list 2");
4735	let top_ids: Vec<&str> = top_two.iter().map(\|run\| run.id.as_str()).collect();
4736	assert_eq!(
4737	top_ids,
4738	vec![legacy_newest.id.as_str(), sortable_new.id.as_str()]
4739	);
4740	}
4741
4742	#[test]
4743	fn list_runs_with_limit_skips_older_sortable_files_entirely() {
4744	let tempdir = tempfile::tempdir().expect("tempdir");
4745	let manager =
4746	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4747	let automation = automation_record_with_settings(None, None, None, None);
4748	let base = Utc::now();
4749
4750	let newest = run_created_at(&automation, base);
4751	manager.save_run(&newest).expect("save newest");
4752
4753	// A corrupt sortable-named file older than the newest run: bounded
4754	// listing must never open it, while an unbounded listing fails.
4755	let dir = manager.runs_dir_for(&automation.id).expect("runs dir");
4756	let stale_stamp = run_file_stamp(base - Duration::minutes(5));
4757	fs::write(
4758	dir.join(format!("{stale_stamp}-{}.json", Uuid::new_v4())),
4759	"{ not json",
4760	)
4761	.expect("write corrupt run");
4762
4763	let bounded = manager
4764	.list_runs(&automation.id, Some(1))
4765	.expect("bounded list must not read files beyond the limit");
4766	assert_eq!(bounded.len(), 1);
4767	assert_eq!(bounded[0].id, newest.id);
4768
4769	assert!(manager.list_runs(&automation.id, None).is_err());
4770	}
4771
4772	#[tokio::test]
4773	async fn list_automations_completes_during_slow_enqueue() {
4774	let tempdir = tempfile::tempdir().expect("tempdir");
4775	let manager =
4776	AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
4777	let created = manager
4778	.create_automation(CreateAutomationRequest {
4779	name: "Slow enqueue".to_string(),
4780	prompt: "prompt".to_string(),
4781	rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
4782	cwds: Vec::new(),
4783	model: None,
4784	model_provider: None,
4785	model_provider_id: None,
4786	mode: None,
4787	allow_shell: None,
4788	trust_mode: None,
4789	auto_approve: None,
4790	delivery_mode: None,
4791	status: Some(AutomationStatus::Active),
4792	})
4793	.expect("create");
4794	let shared: SharedAutomationManager = Arc::new(Mutex::new(manager));
4795
4796	let (entered_tx, entered_rx) = tokio::sync::oneshot::channel::<()>();
4797	let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
4798
4799	let run_task = tokio::spawn({
4800	let shared = Arc::clone(&shared);
4801	let automation_id = created.id.clone();
4802	let task_data_dir = tempdir.path().to_path_buf();
4803	async move {
4804	run_now_with(
4805	&shared,
4806	&automation_id,
4807	&task_data_dir,
4808	move \|_, mut run\| async move {
4809	// Delayed task-manager stub: stall the enqueue await until
4810	// the test has proven the manager mutex is free.
4811	let _ = entered_tx.send(());
4812	let _ = release_rx.await;
4813	run.status = AutomationRunStatus::Failed;
4814	run.ended_at = Some(Utc::now());
4815	run.error = Some("stubbed enqueue".to_string());
4816	run
4817	},
4818	)
4819	.await
4820	}
4821	});
4822
4823	entered_rx.await.expect("enqueue phase entered");
4824
4825	let listed = tokio::time::timeout(std::time::Duration::from_secs(2), async {
4826	shared.lock().await.list_automations()
4827	})
4828	.await
4829	.expect("list_automations must not block behind a slow enqueue")
4830	.expect("list automations");
4831	assert_eq!(listed.len(), 1);
4832
4833	release_tx.send(()).expect("release stub");
4834	let run = run_task.await.expect("join").expect("run now");
4835	assert!(matches!(run.status, AutomationRunStatus::Failed));
4836
4837	// The final run state was persisted after the lock was reacquired.
4838	let manager = shared.lock().await;
4839	let runs = manager.list_runs(&created.id, None).expect("list runs");
4840	assert_eq!(runs.len(), 1);
4841	assert_eq!(runs[0].id, run.id);
4842	assert!(matches!(runs[0].status, AutomationRunStatus::Failed));
4843	let automation = manager.get_automation(&created.id).expect("automation");
4844	assert!(automation.last_run_at.is_some());
4845	}
4846
4847	#[tokio::test]
4848	async fn watcher_noop_completion_hides_but_retains_consumed_occurrence() -> Result<()> {
4849	let tempdir = tempfile::tempdir().expect("tempdir");
4850	let task_manager = TaskManager::start_with_executor(
4851	automation_task_config(tempdir.path().join("tasks")),
4852	std::sync::Arc::new(AutomationWatcherNoopExecutor),
4853	)
4854	.await?;
4855	let mut automation = automation_record_with_settings(None, None, None, None);
4856	automation.delivery_mode = Some(AutomationDeliveryMode::Watcher);
4857	automation.next_run_at = Some(Utc::now() - Duration::seconds(1));
4858	let manager =
4859	AutomationManager::open_for_test(tempdir.path().join("automations")).expect("manager");
4860	manager
4861	.save_automation(&automation)
4862	.expect("save automation");
4863	let shared: SharedAutomationManager = Arc::new(Mutex::new(manager));
4864
4865	scheduler_tick_shared(&shared, &task_manager).await?;
4866	let initial = shared
4867	.lock()
4868	.await
4869	.list_runs_with_visibility(&automation.id, None, true)?;
4870	assert_eq!(initial.len(), 1);
4871	let bound_id = initial[0]
4872	.task_id
4873	.as_deref()
4874	.context("watcher task binding")?;
4875	let completed = crate::task_manager::wait_for_terminal_state(
4876	&task_manager,
4877	bound_id,
4878	std::time::Duration::from_secs(5),
4879	)
4880	.await?;
4881	assert_eq!(completed.status, TaskStatus::Completed);
4882	reconcile_run_statuses_shared(&shared, &task_manager).await?;
4883
4884	let manager = shared.lock().await;
4885	assert!(
4886	manager.list_runs(&automation.id, None)?.is_empty(),
4887	"watcher no-op must not leave a phantom run row"
4888	);
4889	let durable = manager.list_runs_with_visibility(&automation.id, None, true)?;
4890	assert_eq!(durable.len(), 1);
4891	assert_eq!(durable[0].status, AutomationRunStatus::Completed);
4892	assert!(durable[0].dispatch.as_ref().unwrap().suppress_report);
4893	let mut updated = manager.get_automation(&automation.id)?;
4894	assert!(
4895	updated.next_run_at.is_some(),
4896	"watcher should keep scheduling"
4897	);
4898	assert_eq!(
4899	updated.last_run_at, None,
4900	"no-op checks are not reportable runs"
4901	);
4902	// Revisit the consumed slot after a torn schedule write. A hidden
4903	// receipt still owns its occurrence and must prevent another task.
4904	updated.next_run_at = Some(durable[0].scheduled_for);
4905	manager.save_automation(&updated)?;
4906	drop(manager);
4907	scheduler_tick_shared(&shared, &task_manager).await?;
4908	assert_eq!(task_manager.list_tasks(None).await?.len(), 1);
4909	assert!(
4910	shared
4911	.lock()
4912	.await
4913	.list_runs(&automation.id, None)?
4914	.is_empty()
4915	);
4916	task_manager.shutdown();
4917	Ok(())
4918	}
4919
4920	#[test]
4921	fn default_automations_dir_honors_codewhale_home_as_hard_override() {
4922	let _lock = crate::test_support::lock_test_env();
4923	let tmp = tempfile::TempDir::new().unwrap();
4924	// SAFETY: serialised by lock_test_env.
4925	unsafe {
4926	std::env::remove_var("DEEPSEEK_AUTOMATIONS_DIR");
4927	std::env::set_var("CODEWHALE_HOME", tmp.path());
4928	}
4929	// $CODEWHALE_HOME IS the home dir (no ".codewhale" appended); the
4930	// legacy ~/.deepseek fallback is bypassed entirely.
4931	assert_eq!(default_automations_dir(), tmp.path().join("automations"));
4932	// SAFETY: cleanup under the same lock.
4933	unsafe {
4934	std::env::remove_var("CODEWHALE_HOME");
4935	}
4936	}
4937
4938	#[test]
4939	fn default_automations_dir_prefers_deepseek_automations_dir_over_codewhale_home() {
4940	let _lock = crate::test_support::lock_test_env();
4941	let tmp = tempfile::TempDir::new().unwrap();
4942	// SAFETY: serialised by lock_test_env.
4943	unsafe {
4944	std::env::set_var("DEEPSEEK_AUTOMATIONS_DIR", tmp.path());
4945	std::env::set_var("CODEWHALE_HOME", "/should/not/be/used");
4946	}
4947	// The most-specific override wins over the base-data-dir override.
4948	assert_eq!(default_automations_dir(), tmp.path());
4949	// SAFETY: cleanup under the same lock.
4950	unsafe {
4951	std::env::remove_var("DEEPSEEK_AUTOMATIONS_DIR");
4952	std::env::remove_var("CODEWHALE_HOME");
4953	}
4954	}
4955	mod ownership;
4956	mod recovery;
4957
4958	/// #6162: the projection's Canceled branch must record why. A cooperative
4959	/// cancel arrives with no error of its own and takes the derived text; a
4960	/// task that already carries the task manager's receipt keeps it; a legacy
4961	/// record with neither degrades to no detail instead of inventing one.
4962	fn canceled_task_record(
4963	error: Option<&str>,
4964	terminal_reason: Option<&str>,
4965	) -> crate::task_manager::TaskRecord {
4966	serde_json::from_value(serde_json::json!({
4967	"id": "task_canceled",
4968	"prompt": "nightly report",
4969	"model": "deepseek-v4-pro",
4970	"workspace": "/tmp/automation-fixture",
4971	"mode": "agent",
4972	"allow_shell": false,
4973	"trust_mode": false,
4974	"status": "canceled",
4975	"created_at": "2026-09-14T10:00:00Z",
4976	"started_at": "2026-09-14T10:00:01Z",
4977	"ended_at": "2026-09-14T10:00:05Z",
4978	"duration_ms": 4000,
4979	"result_summary": null,
4980	"result_detail_path": null,
4981	"error": error,
4982	"terminal_reason": terminal_reason,
4983	"tool_calls": [],
4984	"timeline": [],
4985	}))
4986	.expect("a canceled task record fixture")
4987	}
4988
4989	#[test]
4990	fn a_cooperatively_canceled_task_names_the_cancellation_on_the_run() {
4991	let mut run = new_run_record("auto_1", Utc::now(), Utc::now());
4992	run.status = AutomationRunStatus::Running;
4993	let task = canceled_task_record(None, Some("canceled"));
4994	assert!(apply_task_status(&mut run, &task));
4995	assert_eq!(run.status, AutomationRunStatus::Canceled);
4996	assert_eq!(run.error.as_deref(), Some("canceled by request"));
4997	assert_eq!(run.started_at, task.started_at);
4998	assert_eq!(run.ended_at, task.ended_at);
4999	}
5000
5001	#[test]
5002	fn a_canceled_task_with_its_own_error_keeps_that_error_on_the_run() {
5003	let mut run = new_run_record("auto_1", Utc::now(), Utc::now());
5004	run.status = AutomationRunStatus::Running;
5005	let task = canceled_task_record(
5006	Some("Task canceled because the task manager shut down"),
5007	Some("shutdown"),
5008	);
5009	assert!(apply_task_status(&mut run, &task));
5010	assert_eq!(run.status, AutomationRunStatus::Canceled);
5011	assert_eq!(
5012	run.error.as_deref(),
5013	Some("Task canceled because the task manager shut down")
5014	);
5015	}
5016
5017	#[test]
5018	fn a_legacy_canceled_task_without_a_reason_records_no_detail() {
5019	let mut run = new_run_record("auto_1", Utc::now(), Utc::now());
5020	run.status = AutomationRunStatus::Running;
5021	let task = canceled_task_record(None, None);
5022	assert!(apply_task_status(&mut run, &task));
5023	assert_eq!(run.status, AutomationRunStatus::Canceled);
5024	assert_eq!(run.error, None);
5025	}
5026	}
5027

5027 lines RUST