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

hatchery_server/
tasks.rs
