aidens_daemon_kit/
lib.rs

1//! Daemon facade for P11 queue, schedule, wake, leases, and safe mode.
2
3use aidens_contracts::{
4    ArtifactId, CanonicalToolSideEffectClass, DaemonNamespaceV1, JobV1, QueueHopReportV1,
5    QueueLeaseV1, SafeModeReportV1,
6};
7use aidens_queue_kit::{
8    DurableQueueLogV1, QueueEnqueueOutcomeV1, QueueError, QueueLeaseOutcomeV1, QueueSnapshotV1,
9};
10use aidens_schedule_kit::{one_shot_occurrence, ScheduleError};
11use aidens_wake_kit::{wake_signal, WakeError};
12use chrono::{DateTime, Utc};
13use std::path::{Path, PathBuf};
14use thiserror::Error;
15
16#[derive(Debug, Error)]
17pub enum DaemonError {
18    #[error(transparent)]
19    Queue(#[from] QueueError),
20    #[error(transparent)]
21    Schedule(#[from] ScheduleError),
22    #[error(transparent)]
23    Wake(#[from] WakeError),
24}
25
26#[derive(Debug, Clone)]
27pub struct DaemonControllerV1 {
28    queue: DurableQueueLogV1,
29}
30
31impl DaemonControllerV1 {
32    pub fn namespace(
33        root_path: impl AsRef<Path>,
34        name: impl Into<String>,
35        owner: impl Into<String>,
36    ) -> DaemonNamespaceV1 {
37        let root_path = root_path.as_ref().display().to_string();
38        DaemonNamespaceV1::new(name, root_path, owner)
39    }
40
41    pub fn open(
42        root_path: impl AsRef<Path>,
43        namespace: DaemonNamespaceV1,
44        owner: impl Into<String>,
45    ) -> Result<Self, DaemonError> {
46        Ok(Self {
47            queue: DurableQueueLogV1::open(root_path, namespace, owner)?,
48        })
49    }
50
51    pub fn open_read_only(
52        root_path: impl AsRef<Path>,
53        namespace: DaemonNamespaceV1,
54    ) -> Result<Self, DaemonError> {
55        Ok(Self {
56            queue: DurableQueueLogV1::open_read_only(root_path, namespace)?,
57        })
58    }
59
60    pub fn queue_root(&self) -> PathBuf {
61        self.queue.root_path().to_path_buf()
62    }
63
64    pub fn namespace_id(&self) -> ArtifactId {
65        self.queue.namespace().namespace_id.clone()
66    }
67
68    pub fn enqueue_schedule_occurrence(
69        &self,
70        schedule_id: impl Into<String>,
71        occurrence_key: impl Into<String>,
72        due_at: DateTime<Utc>,
73        payload: serde_json::Value,
74        risk: CanonicalToolSideEffectClass,
75    ) -> Result<QueueEnqueueOutcomeV1, DaemonError> {
76        let occurrence = one_shot_occurrence(
77            self.namespace_id(),
78            schedule_id,
79            occurrence_key,
80            due_at,
81            payload,
82            risk,
83        )?;
84        Ok(self.queue.enqueue_from_schedule(occurrence)?)
85    }
86
87    pub fn enqueue_wake_signal(
88        &self,
89        source: impl Into<String>,
90        signal_key: impl Into<String>,
91        payload: serde_json::Value,
92        risk: CanonicalToolSideEffectClass,
93    ) -> Result<QueueEnqueueOutcomeV1, DaemonError> {
94        let signal = wake_signal(self.namespace_id(), source, signal_key, payload, risk)?;
95        Ok(self.queue.enqueue_from_wake(signal)?)
96    }
97
98    pub fn enqueue_job(&self, job: JobV1) -> Result<QueueEnqueueOutcomeV1, DaemonError> {
99        Ok(self.queue.enqueue_job(job)?)
100    }
101
102    pub fn acquire_next(
103        &self,
104        owner: impl Into<String>,
105        ttl_seconds: i64,
106    ) -> Result<Option<QueueLeaseOutcomeV1>, DaemonError> {
107        Ok(self.queue.acquire_next_lease(owner, ttl_seconds)?)
108    }
109
110    pub fn cancel(
111        &self,
112        job_id: &ArtifactId,
113        reason: impl Into<String>,
114    ) -> Result<QueueHopReportV1, DaemonError> {
115        Ok(self.queue.cancel_job(job_id, reason)?)
116    }
117
118    pub fn complete(
119        &self,
120        job_id: &ArtifactId,
121        lease: &QueueLeaseV1,
122    ) -> Result<QueueHopReportV1, DaemonError> {
123        Ok(self
124            .queue
125            .complete_job(job_id, Some(lease.lease_id.clone()))?)
126    }
127
128    pub fn set_safe_mode(
129        &self,
130        enabled: bool,
131        reason: impl Into<String>,
132    ) -> Result<SafeModeReportV1, DaemonError> {
133        Ok(self.queue.set_safe_mode(enabled, reason)?)
134    }
135
136    pub fn drain(&self, reason: impl Into<String>) -> Result<Vec<QueueHopReportV1>, DaemonError> {
137        Ok(self.queue.drain_non_terminal(reason)?)
138    }
139
140    pub fn snapshot(&self) -> Result<QueueSnapshotV1, DaemonError> {
141        Ok(self.queue.snapshot()?)
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148    use aidens_contracts::JobStateV1;
149
150    fn temp_root(name: &str) -> PathBuf {
151        let root =
152            std::env::temp_dir().join(format!("aidens-daemon-p11-{name}-{}", std::process::id()));
153        let _ = std::fs::remove_dir_all(&root);
154        root
155    }
156
157    #[test]
158    fn daemon_restart_preserves_jobs_and_cancelled_state() {
159        let root = temp_root("restart");
160        let ns = DaemonControllerV1::namespace(&root, "daemon-restart", "daemon-a");
161        let daemon = DaemonControllerV1::open(&root, ns.clone(), "daemon-a").unwrap();
162        let outcome = daemon
163            .enqueue_schedule_occurrence(
164                "once",
165                "occurrence-1",
166                Utc::now(),
167                serde_json::json!({"task":"work"}),
168                CanonicalToolSideEffectClass::ReadOnly,
169            )
170            .unwrap();
171        let job = outcome.job.unwrap();
172        daemon.cancel(&job.job_id, "operator-cancelled").unwrap();
173
174        let restarted = DaemonControllerV1::open(&root, ns, "daemon-a").unwrap();
175        let snapshot = restarted.snapshot().unwrap();
176        assert_eq!(snapshot.jobs.len(), 1);
177        assert_eq!(snapshot.jobs[0].state, JobStateV1::Cancelled);
178        assert!(restarted.acquire_next("daemon-a", 30).unwrap().is_none());
179        let _ = std::fs::remove_dir_all(&root);
180    }
181
182    #[test]
183    fn repeated_schedule_occurrence_suppresses_duplicate_job() {
184        let root = temp_root("duplicate");
185        let ns = DaemonControllerV1::namespace(&root, "daemon-duplicate", "daemon-a");
186        let daemon = DaemonControllerV1::open(&root, ns, "daemon-a").unwrap();
187        let first = daemon
188            .enqueue_schedule_occurrence(
189                "once",
190                "same-logical-occurrence",
191                Utc::now(),
192                serde_json::json!({"task":"work"}),
193                CanonicalToolSideEffectClass::ReadOnly,
194            )
195            .unwrap();
196        let second = daemon
197            .enqueue_schedule_occurrence(
198                "once",
199                "same-logical-occurrence",
200                Utc::now(),
201                serde_json::json!({"task":"work"}),
202                CanonicalToolSideEffectClass::ReadOnly,
203            )
204            .unwrap();
205        assert!(first.enqueued);
206        assert!(!second.enqueued);
207        assert!(second.duplicate_suppression_receipt.is_some());
208        assert_eq!(daemon.snapshot().unwrap().jobs.len(), 1);
209        let _ = std::fs::remove_dir_all(&root);
210    }
211
212    #[test]
213    fn safe_mode_blocks_risky_wake_and_allows_drain() {
214        let root = temp_root("safe");
215        let ns = DaemonControllerV1::namespace(&root, "daemon-safe", "daemon-a");
216        let daemon = DaemonControllerV1::open(&root, ns, "daemon-a").unwrap();
217        daemon.set_safe_mode(true, "operator-safe-mode").unwrap();
218        let blocked = daemon
219            .enqueue_wake_signal(
220                "filesystem",
221                "dangerous-change",
222                serde_json::json!({"cmd":"cargo test"}),
223                CanonicalToolSideEffectClass::Admin,
224            )
225            .unwrap();
226        assert!(!blocked.enqueued);
227        assert!(blocked.safe_mode_receipt.is_some());
228
229        daemon
230            .enqueue_wake_signal(
231                "filesystem",
232                "inspect-change",
233                serde_json::json!({"path":"README.md"}),
234                CanonicalToolSideEffectClass::ReadOnly,
235            )
236            .unwrap();
237        assert_eq!(daemon.snapshot().unwrap().jobs.len(), 1);
238        assert_eq!(daemon.drain("safe-mode-drain").unwrap().len(), 1);
239        assert_eq!(
240            daemon.snapshot().unwrap().jobs[0].state,
241            JobStateV1::Cancelled
242        );
243        let _ = std::fs::remove_dir_all(&root);
244    }
245}
aidens_daemon_kit/lib.rs

aidens_daemon_kit/
lib.rs
