hatchery_server/
tasks.rs

1//! hq-core work model: one task store (former P6 tasks + the harness kernel's task/run
2//! ideas). Tasks carry a revision for compare-and-set and can be scoped to a role; a run
3//! binds a task to one real provider session on a node; sessions are recorded only when
4//! they are real provider conversations (drafts stay in hq-edge). Operator calls are
5//! idempotent by `op_id`. SQLite under the server data dir.
6//!
7//! Task states: backlog | open | assigned | in_progress | review | done | failed | cancelled.
8//! Run states: running | review | done | failed | cancelled.
9
10use crate::error::ServerError;
11use rusqlite::{params, Connection, OptionalExtension};
12use serde::Serialize;
13use serde_json::Value;
14use std::path::Path;
15
16pub const TASK_STATES: &[&str] =
17    &["backlog", "open", "assigned", "in_progress", "review", "done", "failed", "cancelled"];
18pub const FINAL_TASK_STATES: &[&str] = &["done", "failed", "cancelled"];
19pub const RUN_STATES: &[&str] = &["running", "review", "done", "failed", "cancelled"];
20
21#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
22pub struct Task {
23    pub id: i64,
24    pub revision: i64,
25    pub title: String,
26    pub body: String,
27    pub assignee: Option<String>,
28    pub role: Option<String>,
29    pub parent_id: Option<i64>,
30    pub status: String,
31    pub created_at_ms: u64,
32    pub updated_at_ms: u64,
33}
34
35#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
36pub struct Run {
37    pub id: i64,
38    pub task_id: i64,
39    pub node_id: String,
40    pub session_id: String,
41    pub status: String,
42    pub branch: Option<String>,
43    pub commit_id: Option<String>,
44    pub result_message_id: Option<String>,
45    pub closed_by: Option<String>,
46    pub created_at_ms: u64,
47    pub updated_at_ms: u64,
48}
49
50#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
51pub struct Session {
52    pub session_id: String,
53    pub node_id: String,
54    pub provider: String,
55    pub workspace_id: Option<String>,
56    pub role: Option<String>,
57    pub state: String,
58    pub created_at_ms: u64,
59    pub updated_at_ms: u64,
60}
61
62/// Why a mutation did not apply.
63#[derive(Debug, PartialEq, Eq)]
64pub enum Refused {
65    NotFound,
66    Conflict(&'static str),
67    Invalid(&'static str),
68}
69
70pub struct TaskStore {
71    conn: Connection,
72}
73
74fn unix_ms() -> i64 {
75    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as i64
76}
77
78fn db(e: rusqlite::Error) -> ServerError {
79    ServerError::sub("tasks", e.to_string())
80}
81
82const TASK_COLS: &str = "id,revision,title,body,assignee,role,parent_id,status,created_at_ms,updated_at_ms";
83const RUN_COLS: &str = "id,task_id,node_id,session_id,status,branch,commit_id,result_message_id,closed_by,created_at_ms,updated_at_ms";
84const SESSION_COLS: &str = "session_id,node_id,provider,workspace_id,role,state,created_at_ms,updated_at_ms";
85
86fn task_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Task> {
87    Ok(Task {
88        id: r.get(0)?,
89        revision: r.get(1)?,
90        title: r.get(2)?,
91        body: r.get(3)?,
92        assignee: r.get(4)?,
93        role: r.get(5)?,
94        parent_id: r.get(6)?,
95        status: r.get(7)?,
96        created_at_ms: r.get::<_, i64>(8)? as u64,
97        updated_at_ms: r.get::<_, i64>(9)? as u64,
98    })
99}
100
101fn run_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Run> {
102    Ok(Run {
103        id: r.get(0)?,
104        task_id: r.get(1)?,
105        node_id: r.get(2)?,
106        session_id: r.get(3)?,
107        status: r.get(4)?,
108        branch: r.get(5)?,
109        commit_id: r.get(6)?,
110        result_message_id: r.get(7)?,
111        closed_by: r.get(8)?,
112        created_at_ms: r.get::<_, i64>(9)? as u64,
113        updated_at_ms: r.get::<_, i64>(10)? as u64,
114    })
115}
116
117fn session_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Session> {
118    Ok(Session {
119        session_id: r.get(0)?,
120        node_id: r.get(1)?,
121        provider: r.get(2)?,
122        workspace_id: r.get(3)?,
123        role: r.get(4)?,
124        state: r.get(5)?,
125        created_at_ms: r.get::<_, i64>(6)? as u64,
126        updated_at_ms: r.get::<_, i64>(7)? as u64,
127    })
128}
129
130impl TaskStore {
131    pub fn open(data_dir: &Path) -> Result<Self, ServerError> {
132        let dir = data_dir.join("tasks");
133        std::fs::create_dir_all(&dir)?;
134        Self::from_conn(Connection::open(dir.join("tasks.db")).map_err(db)?)
135    }
136
137    pub fn in_memory() -> Result<Self, ServerError> {
138        Self::from_conn(Connection::open_in_memory().map_err(db)?)
139    }
140
141    fn from_conn(conn: Connection) -> Result<Self, ServerError> {
142        conn.execute_batch(
143            "PRAGMA journal_mode=WAL;
144             CREATE TABLE IF NOT EXISTS tasks (
145               id INTEGER PRIMARY KEY AUTOINCREMENT,
146               title TEXT NOT NULL, body TEXT NOT NULL, assignee TEXT,
147               status TEXT NOT NULL, created_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL);
148             CREATE TABLE IF NOT EXISTS runs (
149               id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, node_id TEXT NOT NULL,
150               session_id TEXT NOT NULL, status TEXT NOT NULL, branch TEXT, commit_id TEXT,
151               result_message_id TEXT, closed_by TEXT,
152               created_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL);
153             CREATE TABLE IF NOT EXISTS sessions (
154               session_id TEXT PRIMARY KEY, node_id TEXT NOT NULL, provider TEXT NOT NULL,
155               workspace_id TEXT, role TEXT, state TEXT NOT NULL,
156               created_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL);
157             CREATE TABLE IF NOT EXISTS ops (
158               op_id TEXT PRIMARY KEY, result TEXT NOT NULL, at_ms INTEGER NOT NULL);",
159        )
160        .map_err(db)?;
161        // Additive migration of the P6 table.
162        for (col, ddl) in [
163            ("revision", "ALTER TABLE tasks ADD COLUMN revision INTEGER NOT NULL DEFAULT 1"),
164            ("role", "ALTER TABLE tasks ADD COLUMN role TEXT"),
165            ("parent_id", "ALTER TABLE tasks ADD COLUMN parent_id INTEGER"),
166        ] {
167            let has: bool = conn
168                .prepare("SELECT 1 FROM pragma_table_info('tasks') WHERE name=?1")
169                .and_then(|mut s| s.exists(params![col]))
170                .map_err(db)?;
171            if !has {
172                conn.execute(ddl, []).map_err(db)?;
173            }
174        }
175        Ok(Self { conn })
176    }
177
178    /// Replay a recorded operation result, if `op_id` was seen.
179    pub fn replay(&self, op_id: Option<&str>) -> Result<Option<Value>, ServerError> {
180        let Some(op) = op_id else { return Ok(None) };
181        let raw: Option<String> = self
182            .conn
183            .query_row("SELECT result FROM ops WHERE op_id=?1", params![op], |r| r.get(0))
184            .optional()
185            .map_err(db)?;
186        Ok(raw.and_then(|r| serde_json::from_str(&r).ok()))
187    }
188
189    pub fn record(&self, op_id: Option<&str>, result: &Value) -> Result<(), ServerError> {
190        if let Some(op) = op_id {
191            self.conn
192                .execute("INSERT OR IGNORE INTO ops (op_id, result, at_ms) VALUES (?1,?2,?3)", params![op, result.to_string(), unix_ms()])
193                .map_err(db)?;
194        }
195        Ok(())
196    }
197
198    pub fn create(&self, title: &str, body: &str, assignee: Option<&str>) -> Result<Task, ServerError> {
199        self.create_full(title, body, assignee, None, None, None)
200    }
201
202    pub fn create_full(
203        &self,
204        title: &str,
205        body: &str,
206        assignee: Option<&str>,
207        role: Option<&str>,
208        parent_id: Option<i64>,
209        status: Option<&str>,
210    ) -> Result<Task, ServerError> {
211        let now = unix_ms();
212        let status = status.unwrap_or(if assignee.is_some() { "assigned" } else { "open" });
213        self.conn
214            .execute(
215                "INSERT INTO tasks (title, body, assignee, role, parent_id, status, revision, created_at_ms, updated_at_ms) VALUES (?1,?2,?3,?4,?5,?6,1,?7,?7)",
216                params![title, body, assignee, role, parent_id, status, now],
217            )
218            .map_err(db)?;
219        self.get(self.conn.last_insert_rowid())?.ok_or_else(|| ServerError::sub("tasks", "lost row"))
220    }
221
222    pub fn get(&self, id: i64) -> Result<Option<Task>, ServerError> {
223        self.conn
224            .query_row(&format!("SELECT {TASK_COLS} FROM tasks WHERE id=?1"), params![id], task_row)
225            .optional()
226            .map_err(db)
227    }
228
229    pub fn list(&self, assignee: Option<&str>, open_only: bool) -> Result<Vec<Task>, ServerError> {
230        self.list_filtered(assignee, None, open_only)
231    }
232
233    /// `roles`: when set, only tasks without a role or with one of these roles.
234    pub fn list_filtered(&self, assignee: Option<&str>, roles: Option<&[String]>, open_only: bool) -> Result<Vec<Task>, ServerError> {
235        let mut stmt = self
236            .conn
237            .prepare(&format!(
238                "SELECT {TASK_COLS} FROM tasks WHERE (?1 IS NULL OR assignee=?1) AND (?2=0 OR status NOT IN ('done','failed','cancelled')) ORDER BY id"
239            ))
240            .map_err(db)?;
241        let rows = stmt.query_map(params![assignee, open_only as i64], task_row).map_err(db)?;
242        let all: Vec<Task> = rows.collect::<Result<_, _>>().map_err(db)?;
243        Ok(match roles {
244            None => all,
245            Some(roles) => all.into_iter().filter(|t| t.role.as_ref().is_none_or(|r| roles.iter().any(|x| x == r))).collect(),
246        })
247    }
248
249    fn bump(&self, id: i64, expected: Option<i64>, set: &str, extra: &[&dyn rusqlite::ToSql]) -> Result<Result<Task, Refused>, ServerError> {
250        let Some(task) = self.get(id)? else { return Ok(Err(Refused::NotFound)) };
251        if FINAL_TASK_STATES.contains(&task.status.as_str()) {
252            return Ok(Err(Refused::Conflict("task is final")));
253        }
254        if expected.is_some_and(|e| e != task.revision) {
255            return Ok(Err(Refused::Conflict("stale revision")));
256        }
257        let sql = format!("UPDATE tasks SET {set}, revision=revision+1, updated_at_ms=?2 WHERE id=?1 AND revision=?3");
258        let mut args: Vec<&dyn rusqlite::ToSql> = vec![&id];
259        let now = unix_ms();
260        args.push(&now);
261        args.push(&task.revision);
262        args.extend_from_slice(extra);
263        let n = self.conn.execute(&sql, args.as_slice()).map_err(db)?;
264        if n != 1 {
265            return Ok(Err(Refused::Conflict("stale revision")));
266        }
267        Ok(Ok(self.get(id)?.expect("task exists")))
268    }
269
270    pub fn assign(&self, id: i64, assignee: &str) -> Result<bool, ServerError> {
271        let Some(t) = self.get(id)? else { return Ok(false) };
272        if !matches!(t.status.as_str(), "backlog" | "open" | "assigned") {
273            return Ok(false);
274        }
275        Ok(self.bump(id, None, "assignee=?4, status='assigned'", &[&assignee])?.is_ok())
276    }
277
278    /// Move a task to `status` (operator), with optional revision check.
279    pub fn move_to(&self, id: i64, status: &str, expected: Option<i64>) -> Result<Result<Task, Refused>, ServerError> {
280        if !TASK_STATES.contains(&status) {
281            return Ok(Err(Refused::Invalid("unknown status")));
282        }
283        self.bump(id, expected, "status=?4", &[&status])
284    }
285
286    /// Operator cancel, or the assignee moving its own task along (`in_progress`, `done`).
287    pub fn set_status(&self, id: i64, status: &str, only_assignee: Option<&str>) -> Result<bool, ServerError> {
288        if let Some(who) = only_assignee {
289            match self.get(id)? {
290                Some(t) if t.assignee.as_deref() == Some(who) => {}
291                _ => return Ok(false),
292            }
293        }
294        Ok(self.move_to(id, status, None)?.is_ok())
295    }
296
297    // ---- sessions (only real provider sessions) ----
298
299    pub fn upsert_session(&self, s: &Session) -> Result<Session, ServerError> {
300        let now = unix_ms();
301        self.conn
302            .execute(
303                "INSERT INTO sessions (session_id,node_id,provider,workspace_id,role,state,created_at_ms,updated_at_ms) VALUES (?1,?2,?3,?4,?5,?6,?7,?7)
304                 ON CONFLICT(session_id) DO UPDATE SET workspace_id=COALESCE(?4,workspace_id), role=COALESCE(?5,role), state=?6, updated_at_ms=?7",
305                params![s.session_id, s.node_id, s.provider, s.workspace_id, s.role, s.state, now],
306            )
307            .map_err(db)?;
308        self.session(&s.session_id)?.ok_or_else(|| ServerError::sub("tasks", "lost session"))
309    }
310
311    pub fn session(&self, id: &str) -> Result<Option<Session>, ServerError> {
312        self.conn
313            .query_row(&format!("SELECT {SESSION_COLS} FROM sessions WHERE session_id=?1"), params![id], session_row)
314            .optional()
315            .map_err(db)
316    }
317
318    pub fn sessions(&self, node: Option<&str>) -> Result<Vec<Session>, ServerError> {
319        let mut stmt = self
320            .conn
321            .prepare(&format!("SELECT {SESSION_COLS} FROM sessions WHERE (?1 IS NULL OR node_id=?1) ORDER BY updated_at_ms DESC"))
322            .map_err(db)?;
323        let rows = stmt.query_map(params![node], session_row).map_err(db)?;
324        rows.collect::<Result<_, _>>().map_err(db)
325    }
326
327    // ---- runs ----
328
329    /// Bind a task to a recorded session. The task moves to `in_progress`.
330    pub fn start_run(&self, task_id: i64, session_id: &str) -> Result<Result<Run, Refused>, ServerError> {
331        let Some(session) = self.session(session_id)? else { return Ok(Err(Refused::Invalid("unknown session"))) };
332        let Some(task) = self.get(task_id)? else { return Ok(Err(Refused::NotFound)) };
333        if self.active_run(task_id)?.is_some() {
334            return Ok(Err(Refused::Conflict("task has an active run")));
335        }
336        if let Err(e) = self.move_to(task_id, "in_progress", Some(task.revision))? {
337            return Ok(Err(e));
338        }
339        let now = unix_ms();
340        self.conn
341            .execute(
342                "INSERT INTO runs (task_id,node_id,session_id,status,created_at_ms,updated_at_ms) VALUES (?1,?2,?3,'running',?4,?4)",
343                params![task_id, session.node_id, session_id, now],
344            )
345            .map_err(db)?;
346        Ok(Ok(self.run(self.conn.last_insert_rowid())?.expect("run")))
347    }
348
349    pub fn run(&self, id: i64) -> Result<Option<Run>, ServerError> {
350        self.conn.query_row(&format!("SELECT {RUN_COLS} FROM runs WHERE id=?1"), params![id], run_row).optional().map_err(db)
351    }
352
353    pub fn runs(&self, task_id: i64) -> Result<Vec<Run>, ServerError> {
354        let mut stmt = self.conn.prepare(&format!("SELECT {RUN_COLS} FROM runs WHERE task_id=?1 ORDER BY id")).map_err(db)?;
355        let rows = stmt.query_map(params![task_id], run_row).map_err(db)?;
356        rows.collect::<Result<_, _>>().map_err(db)
357    }
358
359    pub fn active_run(&self, task_id: i64) -> Result<Option<Run>, ServerError> {
360        self.conn
361            .query_row(&format!("SELECT {RUN_COLS} FROM runs WHERE task_id=?1 AND status IN ('running','review')"), params![task_id], run_row)
362            .optional()
363            .map_err(db)
364    }
365
366    /// Close (or move to review) a run. `by` is `session` (the run's own session, via m4a)
367    /// or `operator` (override). A session may not reopen or touch a closed run; the
368    /// operator may override any non-cancelled run. The task follows the run.
369    pub fn close_run(
370        &self,
371        run_id: i64,
372        status: &str,
373        by: &str,
374        caller_session: Option<&str>,
375        branch: Option<&str>,
376        commit_id: Option<&str>,
377        result_message_id: Option<&str>,
378    ) -> Result<Result<Run, Refused>, ServerError> {
379        if !matches!(status, "review" | "done" | "failed" | "cancelled") {
380            return Ok(Err(Refused::Invalid("bad run status")));
381        }
382        let Some(run) = self.run(run_id)? else { return Ok(Err(Refused::NotFound)) };
383        match by {
384            "session" => {
385                if caller_session != Some(run.session_id.as_str()) {
386                    return Ok(Err(Refused::Conflict("not this run's session")));
387                }
388                if !matches!(run.status.as_str(), "running" | "review") {
389                    return Ok(Err(Refused::Conflict("run is closed")));
390                }
391            }
392            "operator" => {
393                if run.status == "cancelled" && status != "cancelled" {
394                    return Ok(Err(Refused::Conflict("run was cancelled")));
395                }
396            }
397            _ => return Ok(Err(Refused::Invalid("bad closer"))),
398        }
399        self.conn
400            .execute(
401                "UPDATE runs SET status=?2, closed_by=?3, branch=COALESCE(?4,branch), commit_id=COALESCE(?5,commit_id),
402                 result_message_id=COALESCE(?6,result_message_id), updated_at_ms=?7 WHERE id=?1",
403                params![run_id, status, by, branch, commit_id, result_message_id, unix_ms()],
404            )
405            .map_err(db)?;
406        // Task follows: review -> review, done/failed/cancelled -> same (operator may reopen via move).
407        let task_state = status;
408        let task = self.get(run.task_id)?.expect("task");
409        if FINAL_TASK_STATES.contains(&task.status.as_str()) && by == "operator" {
410            self.conn
411                .execute("UPDATE tasks SET status=?2, revision=revision+1, updated_at_ms=?3 WHERE id=?1", params![task.id, task_state, unix_ms()])
412                .map_err(db)?;
413        } else {
414            let _ = self.move_to(task.id, task_state, None)?;
415        }
416        Ok(Ok(self.run(run_id)?.expect("run")))
417    }
418}
419
420#[cfg(test)]
421mod tests {
422    use super::*;
423
424    fn sess(store: &TaskStore, id: &str) {
425        store
426            .upsert_session(&Session {
427                session_id: id.into(),
428                node_id: "alpha".into(),
429                provider: "claude".into(),
430                workspace_id: Some("primary".into()),
431                role: None,
432                state: "live".into(),
433                created_at_ms: 0,
434                updated_at_ms: 0,
435            })
436            .unwrap();
437    }
438
439    #[test]
440    fn lifecycle_and_ownership() {
441        let s = TaskStore::in_memory().unwrap();
442        let t = s.create("build", "run it", None).unwrap();
443        assert_eq!(t.status, "open");
444        assert!(s.assign(t.id, "alpha").unwrap());
445        assert_eq!(s.list(Some("alpha"), true).unwrap().len(), 1);
446        assert!(s.list(Some("beta"), true).unwrap().is_empty());
447        assert!(!s.set_status(t.id, "done", Some("beta")).unwrap());
448        assert!(s.set_status(t.id, "in_progress", Some("alpha")).unwrap());
449        assert!(s.set_status(t.id, "done", Some("alpha")).unwrap());
450        assert!(!s.set_status(t.id, "open", None).unwrap());
451        assert!(!s.assign(t.id, "beta").unwrap());
452        assert!(s.list(Some("alpha"), true).unwrap().is_empty());
453        assert_eq!(s.list(None, false).unwrap().len(), 1);
454    }
455
456    #[test]
457    fn create_move_cancel_with_revisions() {
458        let s = TaskStore::in_memory().unwrap();
459        let t = s.create_full("port", "kernel", None, Some("dev"), None, Some("backlog")).unwrap();
460        assert_eq!((t.status.as_str(), t.revision), ("backlog", 1));
461        let t2 = s.move_to(t.id, "open", Some(1)).unwrap().unwrap();
462        assert_eq!(t2.revision, 2);
463        assert_eq!(s.move_to(t.id, "assigned", Some(1)).unwrap().unwrap_err(), Refused::Conflict("stale revision"));
464        assert_eq!(s.move_to(t.id, "bogus", None).unwrap().unwrap_err(), Refused::Invalid("unknown status"));
465        let t3 = s.move_to(t.id, "cancelled", Some(2)).unwrap().unwrap();
466        assert_eq!(t3.status, "cancelled");
467        assert_eq!(s.move_to(t.id, "open", Some(3)).unwrap().unwrap_err(), Refused::Conflict("task is final"));
468    }
469
470    #[test]
471    fn role_filter() {
472        let s = TaskStore::in_memory().unwrap();
473        s.create_full("a", "", None, Some("dev"), None, None).unwrap();
474        s.create_full("b", "", None, Some("advisor"), None, None).unwrap();
475        s.create_full("c", "", None, None, None, None).unwrap();
476        let dev = s.list_filtered(None, Some(&["dev".into()]), false).unwrap();
477        assert_eq!(dev.iter().map(|t| t.title.as_str()).collect::<Vec<_>>(), ["a", "c"]);
478    }
479
480    #[test]
481    fn runs_close_by_session_and_operator_override() {
482        let s = TaskStore::in_memory().unwrap();
483        sess(&s, "conv-1");
484        let t = s.create("t", "", Some("alpha")).unwrap();
485        assert_eq!(s.start_run(t.id, "nope").unwrap().unwrap_err(), Refused::Invalid("unknown session"));
486        let r = s.start_run(t.id, "conv-1").unwrap().unwrap();
487        assert_eq!(s.get(t.id).unwrap().unwrap().status, "in_progress");
488        assert_eq!(s.start_run(t.id, "conv-1").unwrap().unwrap_err(), Refused::Conflict("task has an active run"));
489        // Another session cannot close it.
490        assert!(s.close_run(r.id, "done", "session", Some("conv-2"), None, None, None).unwrap().is_err());
491        let r = s.close_run(r.id, "review", "session", Some("conv-1"), Some("feat/x"), Some("abc123"), Some("m-1")).unwrap().unwrap();
492        assert_eq!((r.status.as_str(), r.branch.as_deref()), ("review", Some("feat/x")));
493        let r = s.close_run(r.id, "done", "session", Some("conv-1"), None, None, None).unwrap().unwrap();
494        assert_eq!(s.get(t.id).unwrap().unwrap().status, "done");
495        assert!(s.close_run(r.id, "failed", "session", Some("conv-1"), None, None, None).unwrap().is_err());
496        // Operator override of a closed run.
497        let r = s.close_run(r.id, "failed", "operator", None, None, None, None).unwrap().unwrap();
498        assert_eq!((r.status.as_str(), r.closed_by.as_deref()), ("failed", Some("operator")));
499        assert_eq!(s.get(t.id).unwrap().unwrap().status, "failed");
500        assert_eq!(s.runs(t.id).unwrap().len(), 1);
501    }
502
503    #[test]
504    fn op_ids_replay() {
505        let s = TaskStore::in_memory().unwrap();
506        assert!(s.replay(Some("op-1")).unwrap().is_none());
507        s.record(Some("op-1"), &serde_json::json!({"task": 1})).unwrap();
508        assert_eq!(s.replay(Some("op-1")).unwrap().unwrap()["task"], 1);
509    }
510
511    #[test]
512    fn migrates_p6_table_and_survives_reopen() {
513        let dir = tempfile::tempdir().unwrap();
514        std::fs::create_dir_all(dir.path().join("tasks")).unwrap();
515        let old = Connection::open(dir.path().join("tasks/tasks.db")).unwrap();
516        old.execute_batch("CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, body TEXT NOT NULL, assignee TEXT, status TEXT NOT NULL, created_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL);
517            INSERT INTO tasks (title,body,assignee,status,created_at_ms,updated_at_ms) VALUES ('old','b','alpha','assigned',1,1);").unwrap();
518        drop(old);
519        let s = TaskStore::open(dir.path()).unwrap();
520        let t = &s.list(Some("alpha"), true).unwrap()[0];
521        assert_eq!((t.title.as_str(), t.revision, t.role.clone()), ("old", 1, None));
522        drop(s);
523        assert_eq!(TaskStore::open(dir.path()).unwrap().list(None, false).unwrap().len(), 1);
524    }
525}
hatchery_server/tasks.rs

hatchery_server/
tasks.rs
