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 signals (
158               id INTEGER PRIMARY KEY AUTOINCREMENT, node_id TEXT NOT NULL, workspace_id TEXT NOT NULL,
159               instance_id INTEGER NOT NULL, provider TEXT NOT NULL, kind TEXT NOT NULL, hint TEXT,
160               at_ms INTEGER NOT NULL);
161             CREATE INDEX IF NOT EXISTS signals_node ON signals(node_id, id);
162             CREATE TABLE IF NOT EXISTS ops (
163               op_id TEXT PRIMARY KEY, result TEXT NOT NULL, at_ms INTEGER NOT NULL);",
164        )
165        .map_err(db)?;
166        // Additive migration of the P6 table.
167        for (col, ddl) in [
168            ("revision", "ALTER TABLE tasks ADD COLUMN revision INTEGER NOT NULL DEFAULT 1"),
169            ("role", "ALTER TABLE tasks ADD COLUMN role TEXT"),
170            ("parent_id", "ALTER TABLE tasks ADD COLUMN parent_id INTEGER"),
171        ] {
172            let has: bool = conn
173                .prepare("SELECT 1 FROM pragma_table_info('tasks') WHERE name=?1")
174                .and_then(|mut s| s.exists(params![col]))
175                .map_err(db)?;
176            if !has {
177                conn.execute(ddl, []).map_err(db)?;
178            }
179        }
180        let has_nick: bool = conn
181            .prepare("SELECT 1 FROM pragma_table_info('sessions') WHERE name='nick'")
182            .and_then(|mut s| s.exists([]))
183            .map_err(db)?;
184        if !has_nick {
185            conn.execute("ALTER TABLE sessions ADD COLUMN nick TEXT", []).map_err(db)?;
186        }
187        Ok(Self { conn })
188    }
189
190    pub fn set_session_nick(&self, session_id: &str, nick: &str) -> Result<(), ServerError> {
191        self.conn.execute("UPDATE sessions SET nick=?2 WHERE session_id=?1", params![session_id, nick]).map_err(db)?;
192        Ok(())
193    }
194
195    pub fn session_nick(&self, session_id: &str) -> Result<Option<String>, ServerError> {
196        self.conn
197            .query_row("SELECT nick FROM sessions WHERE session_id=?1", params![session_id], |r| r.get::<_, Option<String>>(0))
198            .optional()
199            .map(Option::flatten)
200            .map_err(db)
201    }
202
203    /// (run, session nick) for every active run whose session has a nick.
204    pub fn active_runs_with_nick(&self) -> Result<Vec<(Run, String)>, ServerError> {
205        let mut stmt = self
206            .conn
207            .prepare(&format!(
208                "SELECT {}, s.nick FROM runs r JOIN sessions s ON s.session_id=r.session_id WHERE r.status IN ('running','review') AND s.nick IS NOT NULL",
209                RUN_COLS.split(',').map(|c| format!("r.{c}")).collect::<Vec<_>>().join(",")
210            ))
211            .map_err(db)?;
212        let rows = stmt.query_map([], |r| Ok((run_row(r)?, r.get::<_, String>(11)?))).map_err(db)?;
213        rows.collect::<Result<_, _>>().map_err(db)
214    }
215
216    /// Replay a recorded operation result, if `op_id` was seen.
217    pub fn replay(&self, op_id: Option<&str>) -> Result<Option<Value>, ServerError> {
218        let Some(op) = op_id else { return Ok(None) };
219        let raw: Option<String> = self
220            .conn
221            .query_row("SELECT result FROM ops WHERE op_id=?1", params![op], |r| r.get(0))
222            .optional()
223            .map_err(db)?;
224        Ok(raw.and_then(|r| serde_json::from_str(&r).ok()))
225    }
226
227    pub fn record(&self, op_id: Option<&str>, result: &Value) -> Result<(), ServerError> {
228        if let Some(op) = op_id {
229            self.conn
230                .execute("INSERT OR IGNORE INTO ops (op_id, result, at_ms) VALUES (?1,?2,?3)", params![op, result.to_string(), unix_ms()])
231                .map_err(db)?;
232        }
233        Ok(())
234    }
235
236    pub fn create(&self, title: &str, body: &str, assignee: Option<&str>) -> Result<Task, ServerError> {
237        self.create_full(title, body, assignee, None, None, None)
238    }
239
240    pub fn create_full(
241        &self,
242        title: &str,
243        body: &str,
244        assignee: Option<&str>,
245        role: Option<&str>,
246        parent_id: Option<i64>,
247        status: Option<&str>,
248    ) -> Result<Task, ServerError> {
249        let now = unix_ms();
250        let status = status.unwrap_or(if assignee.is_some() { "assigned" } else { "open" });
251        self.conn
252            .execute(
253                "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)",
254                params![title, body, assignee, role, parent_id, status, now],
255            )
256            .map_err(db)?;
257        self.get(self.conn.last_insert_rowid())?.ok_or_else(|| ServerError::sub("tasks", "lost row"))
258    }
259
260    pub fn get(&self, id: i64) -> Result<Option<Task>, ServerError> {
261        self.conn
262            .query_row(&format!("SELECT {TASK_COLS} FROM tasks WHERE id=?1"), params![id], task_row)
263            .optional()
264            .map_err(db)
265    }
266
267    pub fn list(&self, assignee: Option<&str>, open_only: bool) -> Result<Vec<Task>, ServerError> {
268        self.list_filtered(assignee, None, open_only)
269    }
270
271    /// `roles`: when set, only tasks without a role or with one of these roles.
272    pub fn list_filtered(&self, assignee: Option<&str>, roles: Option<&[String]>, open_only: bool) -> Result<Vec<Task>, ServerError> {
273        let mut stmt = self
274            .conn
275            .prepare(&format!(
276                "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"
277            ))
278            .map_err(db)?;
279        let rows = stmt.query_map(params![assignee, open_only as i64], task_row).map_err(db)?;
280        let all: Vec<Task> = rows.collect::<Result<_, _>>().map_err(db)?;
281        Ok(match roles {
282            None => all,
283            Some(roles) => all.into_iter().filter(|t| t.role.as_ref().is_none_or(|r| roles.iter().any(|x| x == r))).collect(),
284        })
285    }
286
287    fn bump(&self, id: i64, expected: Option<i64>, set: &str, extra: &[&dyn rusqlite::ToSql]) -> Result<Result<Task, Refused>, ServerError> {
288        let Some(task) = self.get(id)? else { return Ok(Err(Refused::NotFound)) };
289        if FINAL_TASK_STATES.contains(&task.status.as_str()) {
290            return Ok(Err(Refused::Conflict("task is final")));
291        }
292        if expected.is_some_and(|e| e != task.revision) {
293            return Ok(Err(Refused::Conflict("stale revision")));
294        }
295        let sql = format!("UPDATE tasks SET {set}, revision=revision+1, updated_at_ms=?2 WHERE id=?1 AND revision=?3");
296        let mut args: Vec<&dyn rusqlite::ToSql> = vec![&id];
297        let now = unix_ms();
298        args.push(&now);
299        args.push(&task.revision);
300        args.extend_from_slice(extra);
301        let n = self.conn.execute(&sql, args.as_slice()).map_err(db)?;
302        if n != 1 {
303            return Ok(Err(Refused::Conflict("stale revision")));
304        }
305        Ok(Ok(self.get(id)?.expect("task exists")))
306    }
307
308    pub fn assign(&self, id: i64, assignee: &str) -> Result<bool, ServerError> {
309        let Some(t) = self.get(id)? else { return Ok(false) };
310        if !matches!(t.status.as_str(), "backlog" | "open" | "assigned") {
311            return Ok(false);
312        }
313        Ok(self.bump(id, None, "assignee=?4, status='assigned'", &[&assignee])?.is_ok())
314    }
315
316    /// Move a task to `status` (operator), with optional revision check.
317    pub fn move_to(&self, id: i64, status: &str, expected: Option<i64>) -> Result<Result<Task, Refused>, ServerError> {
318        if !TASK_STATES.contains(&status) {
319            return Ok(Err(Refused::Invalid("unknown status")));
320        }
321        self.bump(id, expected, "status=?4", &[&status])
322    }
323
324    /// Operator cancel, or the assignee moving its own task along (`in_progress`, `done`).
325    pub fn set_status(&self, id: i64, status: &str, only_assignee: Option<&str>) -> Result<bool, ServerError> {
326        if let Some(who) = only_assignee {
327            match self.get(id)? {
328                Some(t) if t.assignee.as_deref() == Some(who) => {}
329                _ => return Ok(false),
330            }
331        }
332        Ok(self.move_to(id, status, None)?.is_ok())
333    }
334
335    // ---- sessions (only real provider sessions) ----
336
337    pub fn upsert_session(&self, s: &Session) -> Result<Session, ServerError> {
338        let now = unix_ms();
339        self.conn
340            .execute(
341                "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)
342                 ON CONFLICT(session_id) DO UPDATE SET workspace_id=COALESCE(?4,workspace_id), role=COALESCE(?5,role), state=?6, updated_at_ms=?7",
343                params![s.session_id, s.node_id, s.provider, s.workspace_id, s.role, s.state, now],
344            )
345            .map_err(db)?;
346        self.session(&s.session_id)?.ok_or_else(|| ServerError::sub("tasks", "lost session"))
347    }
348
349    pub fn session(&self, id: &str) -> Result<Option<Session>, ServerError> {
350        self.conn
351            .query_row(&format!("SELECT {SESSION_COLS} FROM sessions WHERE session_id=?1"), params![id], session_row)
352            .optional()
353            .map_err(db)
354    }
355
356    pub fn sessions(&self, node: Option<&str>) -> Result<Vec<Session>, ServerError> {
357        let mut stmt = self
358            .conn
359            .prepare(&format!("SELECT {SESSION_COLS} FROM sessions WHERE (?1 IS NULL OR node_id=?1) ORDER BY updated_at_ms DESC"))
360            .map_err(db)?;
361        let rows = stmt.query_map(params![node], session_row).map_err(db)?;
362        rows.collect::<Result<_, _>>().map_err(db)
363    }
364
365    // ---- runs ----
366
367    /// Bind a task to a recorded session. The task moves to `in_progress`.
368    pub fn start_run(&self, task_id: i64, session_id: &str) -> Result<Result<Run, Refused>, ServerError> {
369        let Some(session) = self.session(session_id)? else { return Ok(Err(Refused::Invalid("unknown session"))) };
370        let Some(task) = self.get(task_id)? else { return Ok(Err(Refused::NotFound)) };
371        if self.active_run(task_id)?.is_some() {
372            return Ok(Err(Refused::Conflict("task has an active run")));
373        }
374        if let Err(e) = self.move_to(task_id, "in_progress", Some(task.revision))? {
375            return Ok(Err(e));
376        }
377        let now = unix_ms();
378        self.conn
379            .execute(
380                "INSERT INTO runs (task_id,node_id,session_id,status,created_at_ms,updated_at_ms) VALUES (?1,?2,?3,'running',?4,?4)",
381                params![task_id, session.node_id, session_id, now],
382            )
383            .map_err(db)?;
384        Ok(Ok(self.run(self.conn.last_insert_rowid())?.expect("run")))
385    }
386
387    pub fn run(&self, id: i64) -> Result<Option<Run>, ServerError> {
388        self.conn.query_row(&format!("SELECT {RUN_COLS} FROM runs WHERE id=?1"), params![id], run_row).optional().map_err(db)
389    }
390
391    pub fn runs(&self, task_id: i64) -> Result<Vec<Run>, ServerError> {
392        let mut stmt = self.conn.prepare(&format!("SELECT {RUN_COLS} FROM runs WHERE task_id=?1 ORDER BY id")).map_err(db)?;
393        let rows = stmt.query_map(params![task_id], run_row).map_err(db)?;
394        rows.collect::<Result<_, _>>().map_err(db)
395    }
396
397    pub fn active_run(&self, task_id: i64) -> Result<Option<Run>, ServerError> {
398        self.conn
399            .query_row(&format!("SELECT {RUN_COLS} FROM runs WHERE task_id=?1 AND status IN ('running','review')"), params![task_id], run_row)
400            .optional()
401            .map_err(db)
402    }
403
404    /// Close (or move to review) a run. `by` is `session` (the run's own session, via m4a)
405    /// or `operator` (override). A session may not reopen or touch a closed run; the
406    /// operator may override any non-cancelled run. The task follows the run.
407    pub fn close_run(
408        &self,
409        run_id: i64,
410        status: &str,
411        by: &str,
412        caller_session: Option<&str>,
413        branch: Option<&str>,
414        commit_id: Option<&str>,
415        result_message_id: Option<&str>,
416    ) -> Result<Result<Run, Refused>, ServerError> {
417        if !matches!(status, "review" | "done" | "failed" | "cancelled") {
418            return Ok(Err(Refused::Invalid("bad run status")));
419        }
420        let Some(run) = self.run(run_id)? else { return Ok(Err(Refused::NotFound)) };
421        match by {
422            "session" => {
423                if caller_session != Some(run.session_id.as_str()) {
424                    return Ok(Err(Refused::Conflict("not this run's session")));
425                }
426                if !matches!(run.status.as_str(), "running" | "review") {
427                    return Ok(Err(Refused::Conflict("run is closed")));
428                }
429            }
430            "operator" => {
431                if run.status == "cancelled" && status != "cancelled" {
432                    return Ok(Err(Refused::Conflict("run was cancelled")));
433                }
434            }
435            _ => return Ok(Err(Refused::Invalid("bad closer"))),
436        }
437        self.conn
438            .execute(
439                "UPDATE runs SET status=?2, closed_by=?3, branch=COALESCE(?4,branch), commit_id=COALESCE(?5,commit_id),
440                 result_message_id=COALESCE(?6,result_message_id), updated_at_ms=?7 WHERE id=?1",
441                params![run_id, status, by, branch, commit_id, result_message_id, unix_ms()],
442            )
443            .map_err(db)?;
444        // Task follows: review -> review, done/failed/cancelled -> same (operator may reopen via move).
445        let task_state = status;
446        let task = self.get(run.task_id)?.expect("task");
447        if FINAL_TASK_STATES.contains(&task.status.as_str()) && by == "operator" {
448            self.conn
449                .execute("UPDATE tasks SET status=?2, revision=revision+1, updated_at_ms=?3 WHERE id=?1", params![task.id, task_state, unix_ms()])
450                .map_err(db)?;
451        } else {
452            let _ = self.move_to(task.id, task_state, None)?;
453        }
454        Ok(Ok(self.run(run_id)?.expect("run")))
455    }
456}
457
458#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
459pub struct SessionSignal {
460    pub id: i64,
461    pub node_id: String,
462    pub workspace_id: String,
463    pub instance_id: i64,
464    pub provider: String,
465    /// `limit` | `login_required` | `ready`
466    pub kind: String,
467    pub hint: Option<String>,
468    pub at_ms: u64,
469}
470
471pub const SIGNAL_KINDS: &[&str] = &["limit", "login_required", "ready"];
472const MAX_SIGNALS: i64 = 5000;
473
474impl TaskStore {
475    /// Record a short session signal from a node (state changes only; bounded table).
476    pub fn add_signal(&self, node_id: &str, workspace_id: &str, instance_id: i64, provider: &str, kind: &str, hint: Option<&str>) -> Result<Result<SessionSignal, Refused>, ServerError> {
477        if !SIGNAL_KINDS.contains(&kind) {
478            return Ok(Err(Refused::Invalid("unknown signal kind")));
479        }
480        let hint = hint.map(|h| h.chars().filter(|c| !c.is_control()).take(160).collect::<String>());
481        self.conn
482            .execute(
483                "INSERT INTO signals (node_id,workspace_id,instance_id,provider,kind,hint,at_ms) VALUES (?1,?2,?3,?4,?5,?6,?7)",
484                params![node_id, workspace_id, instance_id, provider, kind, hint, unix_ms()],
485            )
486            .map_err(db)?;
487        let id = self.conn.last_insert_rowid();
488        self.conn.execute("DELETE FROM signals WHERE id <= ?1", params![id - MAX_SIGNALS]).map_err(db)?;
489        Ok(Ok(self.signals(None, Some(id - 1), 1)?.remove(0)))
490    }
491
492    /// Signals after `after_id` (oldest first), optionally for one node.
493    pub fn signals(&self, node: Option<&str>, after_id: Option<i64>, limit: i64) -> Result<Vec<SessionSignal>, ServerError> {
494        let mut stmt = self
495            .conn
496            .prepare("SELECT id,node_id,workspace_id,instance_id,provider,kind,hint,at_ms FROM signals WHERE (?1 IS NULL OR node_id=?1) AND id>?2 ORDER BY id LIMIT ?3")
497            .map_err(db)?;
498        let rows = stmt
499            .query_map(params![node, after_id.unwrap_or(0), limit.clamp(1, 500)], |r| {
500                Ok(SessionSignal {
501                    id: r.get(0)?,
502                    node_id: r.get(1)?,
503                    workspace_id: r.get(2)?,
504                    instance_id: r.get(3)?,
505                    provider: r.get(4)?,
506                    kind: r.get(5)?,
507                    hint: r.get(6)?,
508                    at_ms: r.get::<_, i64>(7)? as u64,
509                })
510            })
511            .map_err(db)?;
512        rows.collect::<Result<_, _>>().map_err(db)
513    }
514}
515
516#[cfg(test)]
517mod tests {
518    use super::*;
519
520    fn sess(store: &TaskStore, id: &str) {
521        store
522            .upsert_session(&Session {
523                session_id: id.into(),
524                node_id: "alpha".into(),
525                provider: "claude".into(),
526                workspace_id: Some("primary".into()),
527                role: None,
528                state: "live".into(),
529                created_at_ms: 0,
530                updated_at_ms: 0,
531            })
532            .unwrap();
533    }
534
535    #[test]
536    fn lifecycle_and_ownership() {
537        let s = TaskStore::in_memory().unwrap();
538        let t = s.create("build", "run it", None).unwrap();
539        assert_eq!(t.status, "open");
540        assert!(s.assign(t.id, "alpha").unwrap());
541        assert_eq!(s.list(Some("alpha"), true).unwrap().len(), 1);
542        assert!(s.list(Some("beta"), true).unwrap().is_empty());
543        assert!(!s.set_status(t.id, "done", Some("beta")).unwrap());
544        assert!(s.set_status(t.id, "in_progress", Some("alpha")).unwrap());
545        assert!(s.set_status(t.id, "done", Some("alpha")).unwrap());
546        assert!(!s.set_status(t.id, "open", None).unwrap());
547        assert!(!s.assign(t.id, "beta").unwrap());
548        assert!(s.list(Some("alpha"), true).unwrap().is_empty());
549        assert_eq!(s.list(None, false).unwrap().len(), 1);
550    }
551
552    #[test]
553    fn create_move_cancel_with_revisions() {
554        let s = TaskStore::in_memory().unwrap();
555        let t = s.create_full("port", "kernel", None, Some("dev"), None, Some("backlog")).unwrap();
556        assert_eq!((t.status.as_str(), t.revision), ("backlog", 1));
557        let t2 = s.move_to(t.id, "open", Some(1)).unwrap().unwrap();
558        assert_eq!(t2.revision, 2);
559        assert_eq!(s.move_to(t.id, "assigned", Some(1)).unwrap().unwrap_err(), Refused::Conflict("stale revision"));
560        assert_eq!(s.move_to(t.id, "bogus", None).unwrap().unwrap_err(), Refused::Invalid("unknown status"));
561        let t3 = s.move_to(t.id, "cancelled", Some(2)).unwrap().unwrap();
562        assert_eq!(t3.status, "cancelled");
563        assert_eq!(s.move_to(t.id, "open", Some(3)).unwrap().unwrap_err(), Refused::Conflict("task is final"));
564    }
565
566    #[test]
567    fn role_filter() {
568        let s = TaskStore::in_memory().unwrap();
569        s.create_full("a", "", None, Some("dev"), None, None).unwrap();
570        s.create_full("b", "", None, Some("advisor"), None, None).unwrap();
571        s.create_full("c", "", None, None, None, None).unwrap();
572        let dev = s.list_filtered(None, Some(&["dev".into()]), false).unwrap();
573        assert_eq!(dev.iter().map(|t| t.title.as_str()).collect::<Vec<_>>(), ["a", "c"]);
574    }
575
576    #[test]
577    fn runs_close_by_session_and_operator_override() {
578        let s = TaskStore::in_memory().unwrap();
579        sess(&s, "conv-1");
580        let t = s.create("t", "", Some("alpha")).unwrap();
581        assert_eq!(s.start_run(t.id, "nope").unwrap().unwrap_err(), Refused::Invalid("unknown session"));
582        let r = s.start_run(t.id, "conv-1").unwrap().unwrap();
583        assert_eq!(s.get(t.id).unwrap().unwrap().status, "in_progress");
584        assert_eq!(s.start_run(t.id, "conv-1").unwrap().unwrap_err(), Refused::Conflict("task has an active run"));
585        // Another session cannot close it.
586        assert!(s.close_run(r.id, "done", "session", Some("conv-2"), None, None, None).unwrap().is_err());
587        let r = s.close_run(r.id, "review", "session", Some("conv-1"), Some("feat/x"), Some("abc123"), Some("m-1")).unwrap().unwrap();
588        assert_eq!((r.status.as_str(), r.branch.as_deref()), ("review", Some("feat/x")));
589        let r = s.close_run(r.id, "done", "session", Some("conv-1"), None, None, None).unwrap().unwrap();
590        assert_eq!(s.get(t.id).unwrap().unwrap().status, "done");
591        assert!(s.close_run(r.id, "failed", "session", Some("conv-1"), None, None, None).unwrap().is_err());
592        // Operator override of a closed run.
593        let r = s.close_run(r.id, "failed", "operator", None, None, None, None).unwrap().unwrap();
594        assert_eq!((r.status.as_str(), r.closed_by.as_deref()), ("failed", Some("operator")));
595        assert_eq!(s.get(t.id).unwrap().unwrap().status, "failed");
596        assert_eq!(s.runs(t.id).unwrap().len(), 1);
597    }
598
599    #[test]
600    fn signals_store_and_page() {
601        let s = TaskStore::in_memory().unwrap();
602        assert_eq!(s.add_signal("a", "w", 1, "claude", "bogus", None).unwrap().unwrap_err(), Refused::Invalid("unknown signal kind"));
603        let x = s.add_signal("a", "w", 1, "claude", "limit", Some("resets 3pm\x1b")).unwrap().unwrap();
604        assert_eq!(x.hint.as_deref(), Some("resets 3pm"));
605        s.add_signal("b", "w", 2, "codex", "ready", None).unwrap().unwrap();
606        assert_eq!(s.signals(Some("a"), None, 10).unwrap().len(), 1);
607        assert_eq!(s.signals(None, Some(x.id), 10).unwrap()[0].node_id, "b");
608    }
609
610    #[test]
611    fn op_ids_replay() {
612        let s = TaskStore::in_memory().unwrap();
613        assert!(s.replay(Some("op-1")).unwrap().is_none());
614        s.record(Some("op-1"), &serde_json::json!({"task": 1})).unwrap();
615        assert_eq!(s.replay(Some("op-1")).unwrap().unwrap()["task"], 1);
616    }
617
618    #[test]
619    fn migrates_p6_table_and_survives_reopen() {
620        let dir = tempfile::tempdir().unwrap();
621        std::fs::create_dir_all(dir.path().join("tasks")).unwrap();
622        let old = Connection::open(dir.path().join("tasks/tasks.db")).unwrap();
623        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);
624            INSERT INTO tasks (title,body,assignee,status,created_at_ms,updated_at_ms) VALUES ('old','b','alpha','assigned',1,1);").unwrap();
625        drop(old);
626        let s = TaskStore::open(dir.path()).unwrap();
627        let t = &s.list(Some("alpha"), true).unwrap()[0];
628        assert_eq!((t.title.as_str(), t.revision, t.role.clone()), ("old", 1, None));
629        drop(s);
630        assert_eq!(TaskStore::open(dir.path()).unwrap().list(None, false).unwrap().len(), 1);
631    }
632}
hatchery_server/tasks.rs

hatchery_server/
tasks.rs
