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    pub fn new(conn: Arc<Mutex<Connection>>) -> Result<Self> {
18        let store = Self { conn };
19        store.init_schema()?;
20        Ok(store)
21    }
22
23    /// Create a TaskStore from a database file path.
24    pub fn open(path: &str) -> Result<Self> {
25        let conn = Connection::open(path)
26            .with_context(|| format!("Failed to open task database: {path}"))?;
27        conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
28        Self::new(Arc::new(Mutex::new(conn)))
29    }
30
31    /// Create an in-memory TaskStore (for tests).
32    pub fn in_memory() -> Result<Self> {
33        let conn = Connection::open_in_memory()?;
34        Self::new(Arc::new(Mutex::new(conn)))
35    }
36
37    fn init_schema(&self) -> Result<()> {
38        let conn = self.conn.clone();
39        // Block on init — called during startup
40        let conn = conn.blocking_lock();
41        conn.execute_batch(
42            r#"
43            CREATE TABLE IF NOT EXISTS tasks (
44                id TEXT PRIMARY KEY,
45                identifier TEXT UNIQUE NOT NULL,
46                name TEXT NOT NULL,
47                description TEXT,
48                instruction TEXT NOT NULL,
49                status TEXT NOT NULL DEFAULT 'backlog',
50                priority INTEGER DEFAULT 0,
51                sort_order REAL,
52                parent_task_id TEXT REFERENCES tasks(id) ON DELETE CASCADE,
53                assignee_agent_id TEXT,
54                created_by_agent_id TEXT,
55                created_by_session_id TEXT,
56                automation_mode TEXT,
57                schedule_pattern TEXT,
58                schedule_timezone TEXT,
59                heartbeat_interval_secs INTEGER,
60                max_executions INTEGER,
61                execution_count INTEGER DEFAULT 0,
62                verify_enabled INTEGER DEFAULT 0,
63                verify_requirement TEXT,
64                verify_max_iterations INTEGER DEFAULT 3,
65                verify_verifier_agent_id TEXT,
66                created_at TEXT NOT NULL,
67                updated_at TEXT NOT NULL,
68                started_at TEXT,
69                completed_at TEXT,
70                last_run_at TEXT,
71                next_run_at TEXT,
72                last_error TEXT,
73                consecutive_failures INTEGER DEFAULT 0,
74                context_json TEXT
75            );
76
77            CREATE TABLE IF NOT EXISTS task_dependencies (
78                task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
79                depends_on TEXT NOT NULL,
80                PRIMARY KEY (task_id, depends_on)
81            );
82
83            CREATE TABLE IF NOT EXISTS task_comments (
84                id TEXT PRIMARY KEY,
85                task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
86                content TEXT NOT NULL,
87                author_agent_id TEXT,
88                created_at TEXT NOT NULL,
89                updated_at TEXT
90            );
91
92            CREATE TABLE IF NOT EXISTS task_runs (
93                id TEXT PRIMARY KEY,
94                task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
95                session_id TEXT,
96                trigger TEXT NOT NULL,
97                status TEXT NOT NULL DEFAULT 'running',
98                summary TEXT,
99                result_content TEXT,
100                started_at TEXT NOT NULL,
101                completed_at TEXT,
102                error TEXT,
103                cost_usd REAL,
104                tokens_used INTEGER
105            );
106
107            CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
108            CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_task_id);
109            CREATE INDEX IF NOT EXISTS idx_tasks_next_run ON tasks(next_run_at);
110            CREATE INDEX IF NOT EXISTS idx_runs_task ON task_runs(task_id);
111            "#,
112        )?;
113        Ok(())
114    }
115
116    pub async fn create_task(&self, params: CreateTaskParams) -> Result<Task> {
117        let conn = self.conn.lock().await;
118        let now = Utc::now().to_rfc3339();
119        let id = uuid::Uuid::new_v4().to_string();
120        let identifier = params
121            .identifier
122            .unwrap_or_else(|| Task::slug_from_name(&params.name));
123
124        conn.execute(
125            r#"INSERT INTO tasks
126               (id, identifier, name, description, instruction, status, priority,
127                sort_order, parent_task_id, assignee_agent_id, created_at, updated_at,
128                verify_enabled, execution_count, consecutive_failures)
129               VALUES (?1, ?2, ?3, ?4, ?5, 'backlog', ?6, ?7, ?8, ?9, ?10, ?11, 0, 0, 0)"#,
130            params![
131                id,
132                identifier,
133                params.name,
134                params.description,
135                params.instruction,
136                params.priority.unwrap_or(0),
137                params.sort_order,
138                params.parent_task_id,
139                params.assignee_agent_id,
140                now,
141                now,
142            ],
143        )
144        .context("insert task")?;
145
146        self.get_task_by_id(&id).await
147    }
148
149    pub async fn get_task_by_id(&self, id: &str) -> Result<Task> {
150        let conn = self.conn.lock().await;
151        let mut stmt = conn.prepare(
152            r#"SELECT id, identifier, name, description, instruction, status, priority,
153                      sort_order, parent_task_id, assignee_agent_id, created_by_agent_id,
154                      created_by_session_id, automation_mode, schedule_pattern,
155                      schedule_timezone, heartbeat_interval_secs, max_executions,
156                      execution_count, verify_enabled, verify_requirement,
157                      verify_max_iterations, verify_verifier_agent_id,
158                      created_at, updated_at, started_at, completed_at,
159                      last_run_at, next_run_at, last_error, consecutive_failures,
160                      context_json
161               FROM tasks WHERE id = ?1"#,
162        )?;
163
164        let task = stmt.query_row(params![id], map_task_row)?;
165        Ok(task)
166    }
167
168    pub async fn list_tasks(&self, list_params: ListTasksParams) -> Result<Vec<Task>> {
169        let conn = self.conn.lock().await;
170        let limit = list_params.limit.unwrap_or(100).min(500);
171        let offset = list_params.offset.unwrap_or(0);
172
173        let mut sql = String::from(
174            r#"SELECT id, identifier, name, description, instruction, status, priority,
175                      sort_order, parent_task_id, assignee_agent_id, created_by_agent_id,
176                      created_by_session_id, automation_mode, schedule_pattern,
177                      schedule_timezone, heartbeat_interval_secs, max_executions,
178                      execution_count, verify_enabled, verify_requirement,
179                      verify_max_iterations, verify_verifier_agent_id,
180                      created_at, updated_at, started_at, completed_at,
181                      last_run_at, next_run_at, last_error, consecutive_failures,
182                      context_json
183               FROM tasks WHERE 1=1"#,
184        );
185
186        let mut param_values: Vec<Box<dyn rusqlite::ToSql>> =
187            vec![Box::new(limit), Box::new(offset)];
188
189        if let Some(statuses) = &list_params.statuses {
190            let placeholders: Vec<String> = statuses
191                .iter()
192                .enumerate()
193                .map(|(i, _)| format!("?{}", param_values.len() + i + 1))
194                .collect();
195            sql.push_str(&format!(" AND status IN ({})", placeholders.join(",")));
196            for s in statuses {
197                param_values.push(Box::new(s.clone()));
198            }
199        }
200        if let Some(ref assignee) = list_params.assignee_agent_id {
201            sql.push_str(&format!(
202                " AND assignee_agent_id = ?{}",
203                param_values.len() + 1
204            ));
205            param_values.push(Box::new(assignee.clone()));
206        }
207        if let Some(ref parent) = list_params.parent_task_id {
208            sql.push_str(&format!(
209                " AND parent_task_id = ?{}",
210                param_values.len() + 1
211            ));
212            param_values.push(Box::new(parent.clone()));
213        }
214
215        sql.push_str(" ORDER BY sort_order, created_at DESC LIMIT ?1 OFFSET ?2");
216
217        let param_refs: Vec<&dyn rusqlite::ToSql> =
218            param_values.iter().map(|p| p.as_ref()).collect();
219        let mut stmt = conn.prepare(&sql)?;
220        let tasks = stmt
221            .query_map(param_refs.as_slice(), map_task_row)?
222            .filter_map(|r| r.ok())
223            .collect();
224
225        Ok(tasks)
226    }
227
228    pub async fn delete_task(&self, id: &str) -> Result<()> {
229        let conn = self.conn.lock().await;
230        conn.execute("DELETE FROM tasks WHERE id = ?1", params![id])
231            .context("delete task")?;
232        Ok(())
233    }
234
235    pub async fn update_status(&self, id: &str, status: &TaskStatus) -> Result<()> {
236        let conn = self.conn.lock().await;
237        let now = Utc::now().to_rfc3339();
238        let completed = if *status == TaskStatus::Completed {
239            Some(now.clone())
240        } else {
241            None
242        };
243        conn.execute(
244            r#"UPDATE tasks SET status = ?1, updated_at = ?2, completed_at = COALESCE(?3, completed_at)
245               WHERE id = ?4"#,
246            params![status.to_string(), now, completed, id],
247        )?;
248        Ok(())
249    }
250
251    pub async fn list_due_tasks(&self) -> Result<Vec<Task>> {
252        let conn = self.conn.lock().await;
253        let now = Utc::now().to_rfc3339();
254        let mut stmt = conn.prepare(
255            r#"SELECT id, identifier, name, description, instruction, status, priority,
256                      sort_order, parent_task_id, assignee_agent_id, created_by_agent_id,
257                      created_by_session_id, automation_mode, schedule_pattern,
258                      schedule_timezone, heartbeat_interval_secs, max_executions,
259                      execution_count, verify_enabled, verify_requirement,
260                      verify_max_iterations, verify_verifier_agent_id,
261                      created_at, updated_at, started_at, completed_at,
262                      last_run_at, next_run_at, last_error, consecutive_failures,
263                      context_json
264               FROM tasks
265               WHERE automation_mode IS NOT NULL
266                 AND status IN ('scheduled', 'running')
267                 AND next_run_at IS NOT NULL
268                 AND next_run_at <= ?1
269               ORDER BY next_run_at"#,
270        )?;
271        let tasks = stmt
272            .query_map(params![now], map_task_row)?
273            .filter_map(|r| r.ok())
274            .collect();
275        Ok(tasks)
276    }
277
278    pub async fn set_next_run(&self, id: &str, next_run: Option<&str>) -> Result<()> {
279        let conn = self.conn.lock().await;
280        let now = Utc::now().to_rfc3339();
281        conn.execute(
282            "UPDATE tasks SET next_run_at = ?1, updated_at = ?2 WHERE id = ?3",
283            params![next_run, now, id],
284        )?;
285        Ok(())
286    }
287}
288
289// ── Row mapper ──
290
291fn map_task_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Task> {
292    let automation_mode_str: Option<String> = row.get(12)?;
293    let automation_mode = automation_mode_str.as_deref().and_then(|s| s.parse().ok());
294
295    let status_str: String = row.get(5)?;
296    let status = status_str.parse().unwrap_or(TaskStatus::Backlog);
297
298    let context_json: Option<String> = row.get(30)?;
299    let context: HashMap<String, serde_json::Value> = context_json
300        .and_then(|s| serde_json::from_str(&s).ok())
301        .unwrap_or_default();
302
303    Ok(Task {
304        id: row.get(0)?,
305        identifier: row.get(1)?,
306        name: row.get(2)?,
307        description: row.get(3)?,
308        instruction: row.get(4)?,
309        status,
310        priority: row.get(6)?,
311        sort_order: row.get(7)?,
312        parent_task_id: row.get(8)?,
313        assignee_agent_id: row.get(9)?,
314        created_by_agent_id: row.get(10)?,
315        created_by_session_id: row.get(11)?,
316        automation_mode,
317        schedule_pattern: row.get(13)?,
318        schedule_timezone: row.get(14)?,
319        heartbeat_interval_secs: row.get(15)?,
320        max_executions: row.get(16)?,
321        execution_count: row.get(17)?,
322        verify_enabled: row.get::<_, i64>(18)? != 0,
323        verify_requirement: row.get(19)?,
324        verify_max_iterations: row.get::<_, i64>(20)? as u32,
325        verify_verifier_agent_id: row.get(21)?,
326        created_at: row.get(22)?,
327        updated_at: row.get(23)?,
328        started_at: row.get(24)?,
329        completed_at: row.get(25)?,
330        last_run_at: row.get(26)?,
331        next_run_at: row.get(27)?,
332        last_error: row.get(28)?,
333        consecutive_failures: row.get::<_, i64>(29)? as u32,
334        context,
335        dependencies: Vec::new(),
336    })
337}
oxios_kernel/task/store.rs

oxios_kernel/task/
store.rs
