327 lines
12 KiB
Rust
327 lines
12 KiB
Rust
//! Harness-local todo store — the persistent, DB-backed half of the
|
|
//! "todos" (loose-ends v2) system, owned by the in-container harness.
|
|
//!
|
|
//! In-container subsystems (matrix, forge-notify, bash) push *todos* to
|
|
//! the harness over the in-agent socket instead of firing wakes directly.
|
|
//! The harness owns this store locally (one sqlite db under the harness
|
|
//! dir) and signals its own turn loop on a new/changed row — hive-c0re is
|
|
//! not involved (no broker round-trip, no marker files).
|
|
//!
|
|
//! Because the store lives inside a single agent's container, todos are
|
|
//! **not** agent-scoped here (unlike the old c0re store): every row
|
|
//! belongs to this agent. A todo is tagged with a `subsystem` marker plus
|
|
//! an optional `subsystem_key` (a matrix room id, a bash task id, …), and
|
|
//! `(subsystem, subsystem_key)` is the upsert/dedup key — re-pushing the
|
|
//! same item is idempotent, and a producer can list / clear / rebuild only
|
|
//! its own set (e.g. matrix wipes + recreates its todos on daemon restart).
|
|
//!
|
|
//! Removal has two paths (mara's call): the producing subsystem `clear`s a
|
|
//! todo it has resolved (keyed by subsystem + key), or the agent itself
|
|
//! `mark_done`s one by id. Both delete the row.
|
|
|
|
use std::path::Path;
|
|
use std::sync::Mutex;
|
|
|
|
use anyhow::{Context, Result};
|
|
use hive_sh4re::wire_time::now_unix;
|
|
use rusqlite::{Connection, params};
|
|
|
|
const SCHEMA: &str = r"
|
|
CREATE TABLE IF NOT EXISTS todos (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
subsystem TEXT NOT NULL,
|
|
subsystem_key TEXT,
|
|
summary TEXT NOT NULL,
|
|
source TEXT,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
-- (subsystem, subsystem_key) is the upsert/dedup key. A NULL key never
|
|
-- conflicts (SQLite treats NULLs as distinct), so keyless todos always
|
|
-- insert as one-offs; keyed todos update in place.
|
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup
|
|
ON todos (subsystem, subsystem_key);
|
|
";
|
|
|
|
/// One dynamic, subsystem-pushed todo. Timestamps are unix seconds; the
|
|
/// consumer derives `age_seconds` from `updated_at`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct Todo {
|
|
pub id: i64,
|
|
/// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …).
|
|
pub subsystem: String,
|
|
/// Optional subsystem-specific dedup key (matrix room id, bash task
|
|
/// id, …). `None` = a keyless one-off todo.
|
|
pub subsystem_key: Option<String>,
|
|
/// Human-readable one-line summary shown to the agent.
|
|
pub summary: String,
|
|
/// Optional free-text provenance (e.g. the room name / task label).
|
|
pub source: Option<String>,
|
|
pub updated_at: i64,
|
|
}
|
|
|
|
/// The harness-local todo store. Cheap to share behind an `Arc`; the inner
|
|
/// connection is guarded by a `Mutex` (todo ops are short sqlite writes).
|
|
pub struct Todos {
|
|
conn: Mutex<Connection>,
|
|
}
|
|
|
|
impl Todos {
|
|
/// Open (creating if needed) the todo store at `path`.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates sqlite open / schema-apply failures.
|
|
pub fn open(path: &Path) -> Result<Self> {
|
|
let conn =
|
|
Connection::open(path).with_context(|| format!("open todos db {}", path.display()))?;
|
|
conn.execute_batch(SCHEMA).context("apply todos schema")?;
|
|
Ok(Self {
|
|
conn: Mutex::new(conn),
|
|
})
|
|
}
|
|
|
|
/// Insert a todo, or update the existing one for `(subsystem, key)`
|
|
/// when `key` is `Some` and already present. A `None` key never
|
|
/// conflicts, so it always inserts a fresh row.
|
|
///
|
|
/// Returns `(id, changed)` where `changed` is `true` when the row is
|
|
/// new OR its `summary`/`source` actually differed — the caller uses
|
|
/// this to decide whether to signal the turn loop (re-pushing an
|
|
/// identical keyed todo is a no-op and must not re-wake).
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates sqlite query / execute failures.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn upsert(
|
|
&self,
|
|
subsystem: &str,
|
|
key: Option<&str>,
|
|
summary: &str,
|
|
source: Option<&str>,
|
|
) -> Result<(i64, bool)> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let now = now_unix();
|
|
let existing: Option<(i64, String, Option<String>)> = if key.is_some() {
|
|
conn.query_row(
|
|
"SELECT id, summary, source FROM todos \
|
|
WHERE subsystem = ?1 AND subsystem_key IS ?2",
|
|
params![subsystem, key],
|
|
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
|
|
)
|
|
.ok()
|
|
} else {
|
|
None
|
|
};
|
|
if let Some((id, cur_summary, cur_source)) = existing {
|
|
let unchanged = cur_summary == summary && cur_source.as_deref() == source;
|
|
if unchanged {
|
|
return Ok((id, false));
|
|
}
|
|
conn.execute(
|
|
"UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4",
|
|
params![summary, source, now, id],
|
|
)?;
|
|
return Ok((id, true));
|
|
}
|
|
conn.execute(
|
|
"INSERT INTO todos \
|
|
(subsystem, subsystem_key, summary, source, updated_at) \
|
|
VALUES (?1, ?2, ?3, ?4, ?5)",
|
|
params![subsystem, key, summary, source, now],
|
|
)?;
|
|
Ok((conn.last_insert_rowid(), true))
|
|
}
|
|
|
|
/// Clear producer-resolved todo(s) by `(subsystem, key)`. `key =
|
|
/// Some(k)` targets the one keyed row; `key = None` matches
|
|
/// `subsystem_key IS NULL`, i.e. **all** keyless todos for that
|
|
/// subsystem (clear a specific keyless one via [`Todos::mark_done`] by
|
|
/// id instead). Returns the number of rows deleted.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn clear(&self, subsystem: &str, key: Option<&str>) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute(
|
|
"DELETE FROM todos WHERE subsystem = ?1 AND subsystem_key IS ?2",
|
|
params![subsystem, key],
|
|
)?;
|
|
Ok(n)
|
|
}
|
|
|
|
/// Clear every todo `subsystem` owns — used by a producer that rebuilds
|
|
/// its whole set on restart (cancel-and-recreate). Returns the number
|
|
/// of rows deleted.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn clear_subsystem(&self, subsystem: &str) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute("DELETE FROM todos WHERE subsystem = ?1", params![subsystem])?;
|
|
Ok(n)
|
|
}
|
|
|
|
/// The agent marks one of its todos done, by id. Returns the number of
|
|
/// rows deleted (0 when the id was unknown / already gone).
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn mark_done(&self, id: i64) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute("DELETE FROM todos WHERE id = ?1", params![id])?;
|
|
Ok(n)
|
|
}
|
|
|
|
/// List todos, newest-updated first. `subsystem = Some(..)` filters to
|
|
/// one producer's set; `None` returns all.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite prepare / query failures.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn list(&self, subsystem: Option<&str>) -> Result<Vec<Todo>> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let mut stmt = conn.prepare(
|
|
"SELECT id, subsystem, subsystem_key, summary, source, updated_at \
|
|
FROM todos \
|
|
WHERE (?1 IS NULL OR subsystem = ?1) \
|
|
ORDER BY updated_at DESC, id DESC",
|
|
)?;
|
|
let rows = stmt
|
|
.query_map(params![subsystem], |row| {
|
|
Ok(Todo {
|
|
id: row.get(0)?,
|
|
subsystem: row.get(1)?,
|
|
subsystem_key: row.get(2)?,
|
|
summary: row.get(3)?,
|
|
source: row.get(4)?,
|
|
updated_at: row.get(5)?,
|
|
})
|
|
})?
|
|
.collect::<rusqlite::Result<Vec<_>>>()?;
|
|
Ok(rows)
|
|
}
|
|
|
|
/// Cheap existence check — `true` when at least one todo row exists
|
|
/// (across every subsystem). Used to gate a `todo_wake` notification
|
|
/// against being turned into a turn when its payload has already been
|
|
/// drained by an earlier turn (see the serve loop's `LocalTodo` arm):
|
|
/// an `EXISTS` probe, not a full `list` + row materialization, since
|
|
/// this runs on every wake.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite query failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn has_any(&self) -> Result<bool> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let any: bool =
|
|
conn.query_row("SELECT EXISTS(SELECT 1 FROM todos)", [], |row| row.get(0))?;
|
|
Ok(any)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
// Return the `TempDir` alongside the store so it outlives the test —
|
|
// dropping it early deletes the dir and SQLite fails with
|
|
// `SQLITE_READONLY_DBMOVED`.
|
|
fn store() -> (tempfile::TempDir, Todos) {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap();
|
|
(dir, db)
|
|
}
|
|
|
|
#[test]
|
|
fn keyed_upsert_dedups_and_reports_changed() {
|
|
let (_dir, s) = store();
|
|
let (id1, changed1) = s
|
|
.upsert("matrix", Some("!room:x"), "1 unread", None)
|
|
.unwrap();
|
|
assert!(changed1, "first push is new → changed");
|
|
// Same key + same summary → no-op, not changed (must not re-wake).
|
|
let (id2, changed2) = s
|
|
.upsert("matrix", Some("!room:x"), "1 unread", None)
|
|
.unwrap();
|
|
assert_eq!(id1, id2, "keyed upsert updates in place, same row");
|
|
assert!(!changed2, "identical re-push is a no-op");
|
|
// Same key, new summary → updates, changed.
|
|
let (id3, changed3) = s
|
|
.upsert("matrix", Some("!room:x"), "3 unread", None)
|
|
.unwrap();
|
|
assert_eq!(id1, id3);
|
|
assert!(changed3);
|
|
assert_eq!(s.list(Some("matrix")).unwrap().len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn keyless_todos_always_insert() {
|
|
let (_dir, s) = store();
|
|
let (a, _) = s.upsert("bash", None, "task done", None).unwrap();
|
|
let (b, _) = s.upsert("bash", None, "task done", None).unwrap();
|
|
assert_ne!(a, b, "keyless pushes are distinct one-offs");
|
|
assert_eq!(s.list(Some("bash")).unwrap().len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn clear_and_mark_done_remove_rows() {
|
|
let (_dir, s) = store();
|
|
s.upsert("matrix", Some("!a:x"), "unread", None).unwrap();
|
|
let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap();
|
|
assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1);
|
|
assert_eq!(s.mark_done(id).unwrap(), 1);
|
|
assert!(s.list(None).unwrap().is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn has_any_reflects_emptiness() {
|
|
let (_dir, s) = store();
|
|
assert!(!s.has_any().unwrap(), "fresh store has no todos");
|
|
let (id, _) = s.upsert("bash", None, "task done", None).unwrap();
|
|
assert!(s.has_any().unwrap());
|
|
s.mark_done(id).unwrap();
|
|
assert!(
|
|
!s.has_any().unwrap(),
|
|
"empty again after draining the only row"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn clear_subsystem_wipes_only_its_own() {
|
|
let (_dir, s) = store();
|
|
s.upsert("matrix", Some("!a:x"), "u", None).unwrap();
|
|
s.upsert("matrix", Some("!b:x"), "u", None).unwrap();
|
|
s.upsert("forge", Some("pr-1"), "r", None).unwrap();
|
|
assert_eq!(s.clear_subsystem("matrix").unwrap(), 2);
|
|
let left = s.list(None).unwrap();
|
|
assert_eq!(left.len(), 1);
|
|
assert_eq!(left[0].subsystem, "forge");
|
|
}
|
|
}
|