oxios_kernel/task/
store.rs

1// Task store — SQLite-backed CRUD for tasks (RFC-043)
2use anyhow::{Context, Result};
3use chrono::Utc;
4use rusqlite::{Connection, params};
5use std::collections::HashMap;
6use std::sync::Arc;
7use tokio::sync::Mutex;
8
9use super::model::*;
10
11/// SQLite-backed task store.
12pub struct TaskStore {
13    conn: Arc<Mutex<Connection>>,
14}
15
16impl TaskStore {
17    /// Create a TaskStore from a raw connection. Schema is initialized
18    /// on the connection *before* it is wrapped in the async mutex, so
19    /// this constructor is safe to call from inside a Tokio runtime —
20    /// no `blocking_lock` is involved.
21    pub fn new(conn: Connection) -> Result<Self> {
22        init_schema(&conn)?;
23        Ok(Self {
24            conn: Arc::new(Mutex::new(conn)),
25        })
26    }
27
28    /// Create a TaskStore from a database file path.
29    pub fn open(path: &str) -> Result<Self> {
30        let conn = Connection::open(path)
31            .with_context(|| format!("Failed to open task database: {path}"))?;
32        conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
33        Self::new(conn)
34    }
35
36    /// Create an in-memory TaskStore (for tests).
37    pub fn in_memory() -> Result<Self> {
38        let conn = Connection::open_in_memory()?;
39        Self::new(conn)
40    }
41
42    pub async fn create_task(&self, params: CreateTaskParams) -> Result<Task> {
43        let id = uuid::Uuid::new_v4().to_string();
44        {
45            let conn = self.conn.lock().await;
46            let now = Utc::now().to_rfc3339();
47            let identifier = params
48                .identifier
49                .unwrap_or_else(|| Task::slug_from_name(&params.name));
50
51            conn.execute(
52                r#"INSERT INTO tasks
53                   (id, identifier, name, description, instruction, status, priority,
54                    sort_order, parent_task_id, assignee_agent_id, created_at, updated_at,
55                    verify_enabled, execution_count, consecutive_failures)
56                   VALUES (?1, ?2, ?3, ?4, ?5, 'backlog', ?6, ?7, ?8, ?9, ?10, ?11, 0, 0, 0)"#,
57                params![
58                    id,
59                    identifier,
60                    params.name,
61                    params.description,
62                    params.instruction,
63                    params.priority.unwrap_or(0),
64                    params.sort_order,
65                    params.parent_task_id,
66                    params.assignee_agent_id,
67                    now,
68                    now,
69                ],
70            )
71            .context("insert task")?;
72        }
73        // Lock released — safe to call another `&self` method.
74        self.get_task_by_id(&id).await
75    }
76
77    pub async fn get_task_by_id(&self, id: &str) -> Result<Task> {
78        let conn = self.conn.lock().await;
79        let mut stmt = conn.prepare(
80            r#"SELECT id, identifier, name, description, instruction, status, priority,
81                      sort_order, parent_task_id, assignee_agent_id, created_by_agent_id,
82                      created_by_session_id, automation_mode, schedule_pattern,
83                      schedule_timezone, heartbeat_interval_secs, max_executions,
84                      execution_count, verify_enabled, verify_requirement,
85                      verify_max_iterations, verify_verifier_agent_id,
86                      created_at, updated_at, started_at, completed_at,
87                      last_run_at, next_run_at, last_error, consecutive_failures,
88                      context_json
89               FROM tasks WHERE id = ?1"#,
90        )?;
91
92        let task = stmt.query_row(params![id], map_task_row)?;
93        Ok(task)
94    }
95
96    pub async fn list_tasks(&self, list_params: ListTasksParams) -> Result<Vec<Task>> {
97        let conn = self.conn.lock().await;
98        let limit = list_params.limit.unwrap_or(100).min(500);
99        let offset = list_params.offset.unwrap_or(0);
100
101        let mut sql = String::from(
102            r#"SELECT id, identifier, name, description, instruction, status, priority,
103                      sort_order, parent_task_id, assignee_agent_id, created_by_agent_id,
104                      created_by_session_id, automation_mode, schedule_pattern,
105                      schedule_timezone, heartbeat_interval_secs, max_executions,
106                      execution_count, verify_enabled, verify_requirement,
107                      verify_max_iterations, verify_verifier_agent_id,
108                      created_at, updated_at, started_at, completed_at,
109                      last_run_at, next_run_at, last_error, consecutive_failures,
110                      context_json
111               FROM tasks WHERE 1=1"#,
112        );
113
114        let mut param_values: Vec<Box<dyn rusqlite::ToSql>> =
115            vec![Box::new(limit), Box::new(offset)];
116
117        if let Some(statuses) = &list_params.statuses {
118            let placeholders: Vec<String> = statuses
119                .iter()
120                .enumerate()
121                .map(|(i, _)| format!("?{}", param_values.len() + i + 1))
122                .collect();
123            sql.push_str(&format!(" AND status IN ({})", placeholders.join(",")));
124            for s in statuses {
125                param_values.push(Box::new(s.clone()));
126            }
127        }
128        if let Some(ref assignee) = list_params.assignee_agent_id {
129            sql.push_str(&format!(
130                " AND assignee_agent_id = ?{}",
131                param_values.len() + 1
132            ));
133            param_values.push(Box::new(assignee.clone()));
134        }
135        if let Some(ref parent) = list_params.parent_task_id {
136            sql.push_str(&format!(
137                " AND parent_task_id = ?{}",
138                param_values.len() + 1
139            ));
140            param_values.push(Box::new(parent.clone()));
141        }
142
143        sql.push_str(" ORDER BY sort_order, created_at DESC LIMIT ?1 OFFSET ?2");
144
145        let param_refs: Vec<&dyn rusqlite::ToSql> =
146            param_values.iter().map(|p| p.as_ref()).collect();
147        let mut stmt = conn.prepare(&sql)?;
148        let tasks = stmt
149            .query_map(param_refs.as_slice(), map_task_row)?
150            .filter_map(|r| r.ok())
151            .collect();
152
153        Ok(tasks)
154    }
155
156    pub async fn delete_task(&self, id: &str) -> Result<()> {
157        let conn = self.conn.lock().await;
158        conn.execute("DELETE FROM tasks WHERE id = ?1", params![id])
159            .context("delete task")?;
160        Ok(())
161    }
162
163    pub async fn update_status(&self, id: &str, status: &TaskStatus) -> Result<()> {
164        let conn = self.conn.lock().await;
165        let now = Utc::now().to_rfc3339();
166        let completed = if *status == TaskStatus::Completed {
167            Some(now.clone())
168        } else {
169            None
170        };
171        conn.execute(
172            r#"UPDATE tasks SET status = ?1, updated_at = ?2, completed_at = COALESCE(?3, completed_at)
173               WHERE id = ?4"#,
174            params![status.to_string(), now, completed, id],
175        )?;
176        Ok(())
177    }
178
179    pub async fn list_due_tasks(&self) -> Result<Vec<Task>> {
180        let conn = self.conn.lock().await;
181        let now = Utc::now().to_rfc3339();
182        let mut stmt = conn.prepare(
183            r#"SELECT id, identifier, name, description, instruction, status, priority,
184                      sort_order, parent_task_id, assignee_agent_id, created_by_agent_id,
185                      created_by_session_id, automation_mode, schedule_pattern,
186                      schedule_timezone, heartbeat_interval_secs, max_executions,
187                      execution_count, verify_enabled, verify_requirement,
188                      verify_max_iterations, verify_verifier_agent_id,
189                      created_at, updated_at, started_at, completed_at,
190                      last_run_at, next_run_at, last_error, consecutive_failures,
191                      context_json
192               FROM tasks
193               WHERE automation_mode IS NOT NULL
194                 AND status IN ('scheduled', 'running')
195                 AND next_run_at IS NOT NULL
196                 AND next_run_at <= ?1
197               ORDER BY next_run_at"#,
198        )?;
199        let tasks = stmt
200            .query_map(params![now], map_task_row)?
201            .filter_map(|r| r.ok())
202            .collect();
203        Ok(tasks)
204    }
205
206    pub async fn set_next_run(&self, id: &str, next_run: Option<&str>) -> Result<()> {
207        let conn = self.conn.lock().await;
208        let now = Utc::now().to_rfc3339();
209        conn.execute(
210            "UPDATE tasks SET next_run_at = ?1, updated_at = ?2 WHERE id = ?3",
211            params![next_run, now, id],
212        )?;
213        Ok(())
214    }
215}
216fn init_schema(conn: &Connection) -> Result<()> {
217    conn.execute_batch(
218        r#"
219        CREATE TABLE IF NOT EXISTS tasks (
220            id TEXT PRIMARY KEY,
221            identifier TEXT UNIQUE NOT NULL,
222            name TEXT NOT NULL,
223            description TEXT,
224            instruction TEXT NOT NULL,
225            status TEXT NOT NULL DEFAULT 'backlog',
226            priority INTEGER DEFAULT 0,
227            sort_order REAL,
228            parent_task_id TEXT REFERENCES tasks(id) ON DELETE CASCADE,
229            assignee_agent_id TEXT,
230            created_by_agent_id TEXT,
231            created_by_session_id TEXT,
232            automation_mode TEXT,
233            schedule_pattern TEXT,
234            schedule_timezone TEXT,
235            heartbeat_interval_secs INTEGER,
236            max_executions INTEGER,
237            execution_count INTEGER DEFAULT 0,
238            verify_enabled INTEGER DEFAULT 0,
239            verify_requirement TEXT,
240            verify_max_iterations INTEGER DEFAULT 3,
241            verify_verifier_agent_id TEXT,
242            created_at TEXT NOT NULL,
243            updated_at TEXT NOT NULL,
244            started_at TEXT,
245            completed_at TEXT,
246            last_run_at TEXT,
247            next_run_at TEXT,
248            last_error TEXT,
249            consecutive_failures INTEGER DEFAULT 0,
250            context_json TEXT
251        );
252
253        CREATE TABLE IF NOT EXISTS task_dependencies (
254            task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
255            depends_on TEXT NOT NULL,
256            PRIMARY KEY (task_id, depends_on)
257        );
258
259        CREATE TABLE IF NOT EXISTS task_comments (
260            id TEXT PRIMARY KEY,
261            task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
262            content TEXT NOT NULL,
263            author_agent_id TEXT,
264            created_at TEXT NOT NULL,
265            updated_at TEXT
266        );
267
268        CREATE TABLE IF NOT EXISTS task_runs (
269            id TEXT PRIMARY KEY,
270            task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
271            session_id TEXT,
272            trigger TEXT NOT NULL,
273            status TEXT NOT NULL DEFAULT 'running',
274            summary TEXT,
275            result_content TEXT,
276            started_at TEXT NOT NULL,
277            completed_at TEXT,
278            error TEXT,
279            cost_usd REAL,
280            tokens_used INTEGER
281        );
282
283        CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
284        CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_task_id);
285        CREATE INDEX IF NOT EXISTS idx_tasks_next_run ON tasks(next_run_at);
286        CREATE INDEX IF NOT EXISTS idx_runs_task ON task_runs(task_id);
287        "#,
288    )?;
289    Ok(())
290}
291
292// ── Row mapper ──
293
294fn map_task_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Task> {
295    let automation_mode_str: Option<String> = row.get(12)?;
296    let automation_mode = automation_mode_str.as_deref().and_then(|s| s.parse().ok());
297
298    let status_str: String = row.get(5)?;
299    let status = status_str.parse().unwrap_or(TaskStatus::Backlog);
300
301    let context_json: Option<String> = row.get(30)?;
302    let context: HashMap<String, serde_json::Value> = context_json
303        .and_then(|s| serde_json::from_str(&s).ok())
304        .unwrap_or_default();
305
306    Ok(Task {
307        id: row.get(0)?,
308        identifier: row.get(1)?,
309        name: row.get(2)?,
310        description: row.get(3)?,
311        instruction: row.get(4)?,
312        status,
313        priority: row.get(6)?,
314        sort_order: row.get(7)?,
315        parent_task_id: row.get(8)?,
316        assignee_agent_id: row.get(9)?,
317        created_by_agent_id: row.get(10)?,
318        created_by_session_id: row.get(11)?,
319        automation_mode,
320        schedule_pattern: row.get(13)?,
321        schedule_timezone: row.get(14)?,
322        heartbeat_interval_secs: row.get(15)?,
323        max_executions: row.get(16)?,
324        execution_count: row.get(17)?,
325        verify_enabled: row.get::<_, i64>(18)? != 0,
326        verify_requirement: row.get(19)?,
327        verify_max_iterations: row.get::<_, i64>(20)? as u32,
328        verify_verifier_agent_id: row.get(21)?,
329        created_at: row.get(22)?,
330        updated_at: row.get(23)?,
331        started_at: row.get(24)?,
332        completed_at: row.get(25)?,
333        last_run_at: row.get(26)?,
334        next_run_at: row.get(27)?,
335        last_error: row.get(28)?,
336        consecutive_failures: row.get::<_, i64>(29)? as u32,
337        context,
338        dependencies: Vec::new(),
339    })
340}
341
342#[cfg(test)]
343mod tests {
344    use super::*;
345
346    fn sample_params(name: &str) -> CreateTaskParams {
347        CreateTaskParams {
348            name: name.to_string(),
349            instruction: format!("do {name}"),
350            identifier: None,
351            description: None,
352            priority: None,
353            parent_task_id: None,
354            assignee_agent_id: None,
355            sort_order: None,
356        }
357    }
358
359    // Regression: `TaskStore::open` / `in_memory` must be safe to call from
360    // inside a Tokio runtime. The production web surface constructs the
361    // store on the runtime (`src/api/plugin.rs`); an earlier version used
362    // `blocking_lock()` during schema init and panicked at startup with
363    // "Cannot block the current thread from within a runtime".
364    #[tokio::test]
365    async fn in_memory_store_construction_does_not_panic_on_runtime() {
366        let store = TaskStore::in_memory().expect("in-memory store builds");
367        // Sanity: schema is usable.
368        let task = store
369            .create_task(sample_params("regression"))
370            .await
371            .expect("create works");
372        assert_eq!(task.name, "regression");
373    }
374
375    #[tokio::test]
376    async fn open_from_file_path_works_on_runtime() {
377        let dir = tempfile::tempdir().expect("tempdir");
378        let path = dir.path().join("tasks.db");
379        let path_str = path.to_str().expect("utf8 path");
380        let store = TaskStore::open(path_str).expect("open builds");
381        let created = store
382            .create_task(sample_params("from-disk"))
383            .await
384            .expect("create");
385        // Re-open the same file — schema init must be idempotent and the
386        // row must survive reopen.
387        drop(store);
388        let reopened = TaskStore::open(path_str).expect("reopen builds");
389        let fetched = reopened
390            .get_task_by_id(&created.id)
391            .await
392            .expect("get_by_id");
393        assert_eq!(fetched.name, "from-disk");
394    }
395
396    #[tokio::test]
397    async fn create_list_update_delete_roundtrip() {
398        let store = TaskStore::in_memory().expect("in-memory store builds");
399        let t1 = store
400            .create_task(sample_params("alpha"))
401            .await
402            .expect("create alpha");
403        let _t2 = store
404            .create_task(sample_params("beta"))
405            .await
406            .expect("create beta");
407
408        let listed = store
409            .list_tasks(ListTasksParams::default())
410            .await
411            .expect("list");
412        assert_eq!(listed.len(), 2);
413
414        store
415            .update_status(&t1.id, &TaskStatus::Completed)
416            .await
417            .expect("update");
418        let fetched = store.get_task_by_id(&t1.id).await.expect("get_by_id");
419        assert_eq!(fetched.status, TaskStatus::Completed);
420        assert!(fetched.completed_at.is_some());
421
422        store.delete_task(&t1.id).await.expect("delete");
423        let after = store
424            .list_tasks(ListTasksParams::default())
425            .await
426            .expect("list after delete");
427        assert_eq!(after.len(), 1);
428    }
429}
oxios_kernel/task/store.rs

oxios_kernel/task/
store.rs
