A new swarm service, swarm-notify-router, in its own container on the host
that sets deploy.notify-router.enable (defaulted on by allSwarmServices where
the host has the router's store leaf, like the claude login minter).
- swarm-controller registers instance-wide hooks for `issues` +
`issue_assign` and `pull_request` + `pull_request_assign`, updates the
event list of a hook already registered with fewer (leaving its active flag as it is), and keeps each
verified body unchanged in the SWARM_EVENTS stream on
$SWARM.events.forge.<X-Forgejo-Event-Type>, deduped by X-Forgejo-Delivery.
A delivery the stream did not keep answers 503.
- swarm-notify-router drains its durable consumer on that stream, runs its
flows (source -> match -> todo sink, stored as nodes plus edges and checked
to be a tree) and queues each op on $SWARM.router.todo.<agent> in the
work-queue stream ROUTER_TODOS. Built-in flows: assigned, unassigned (same
key, rewritten) and closed (prefix-clears every holder's router todos). A
per-issue snapshot in its notify-router-state bucket holds the last
assignees and the todo holders; only agents in the agent-status bucket get
ops, and an assignee that was not one is left out of the snapshot so the
next assign event retries it. After the first event that must be retried,
the rest of the batch is nakked so redelivery keeps stream order.
- hive-agent creates its own durable consumer todo-<agent> and applies each
op to its todo store as subsystem `router`, then acks; a refused op stops
the rest of its batch the same way. The store's (subsystem, key) dedup
makes a redelivery harmless. New Todos::clear_prefix.
- swarm-nats-auth: a --notify-router-client arm; the controller may publish
$SWARM.events.forge.*; each agent gets three {agent} templates (extended
consumer create whose filter the server checks, MSG.NEXT, ACK).
- Credentials: a notify-router store leaf minted by glue-bao-tls, a
swarm-notify-router bao policy (read on its own queue client secret), an
authelia machine client, the secret publisher entry, a journal forwarder
route. No manual step.
hive-forge-notify is unchanged, so agents also still get forge's own
notifications.
Refs #5571
693 lines
27 KiB
Rust
693 lines
27 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/dismissal has two different paths. The **producing subsystem**
|
|
//! `clear`s a todo it has itself resolved (keyed by subsystem + key) — a
|
|
//! one-shot event has nothing left to represent once resolved, so this
|
|
//! hard-deletes the row. The **agent** `mark_done`s one by id (via
|
|
//! `cancel_loose_end`) — that only means "stop showing it to me", so it sets
|
|
//! `acked` instead of deleting: a *reconciled* producer like `disk_watch`
|
|
//! (re-derives its summary on a timer, not a one-shot event) needs the row
|
|
//! to survive ack so [`Todos::upsert`] can still tell "unchanged" from
|
|
//! "genuinely different". Acked rows stay out of [`Todos::list`] but keep
|
|
//! anchoring that comparison; a materially different summary un-acks +
|
|
//! wakes again. Acked rows are reaped after a retention window
|
|
//! (`vacuum.rs`) so the table doesn't grow unbounded.
|
|
|
|
use std::path::Path;
|
|
use std::sync::Mutex;
|
|
|
|
use anyhow::{Context, Result};
|
|
use chrono::{DateTime, Utc};
|
|
use hive_sh4re::wire_time;
|
|
use rusqlite::{Connection, params};
|
|
|
|
/// SQL bootstrap. `CREATE TABLE IF NOT EXISTS` so first-boot agents and
|
|
/// existing ones converge on the same base shape; additive columns land via
|
|
/// [`MIGRATIONS`] as try-and-ignore `ALTER TABLE`s so existing dbs catch up
|
|
/// without a hard schema-version bump.
|
|
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);
|
|
";
|
|
|
|
/// Additive column migrations — see `turn_stats.rs`'s identical pattern.
|
|
/// Each runs unconditionally and ignores `duplicate column name` errors, so
|
|
/// re-running on an already-migrated db is a silent no-op.
|
|
const MIGRATIONS: &[&str] = &[
|
|
"ALTER TABLE todos ADD COLUMN acked INTEGER NOT NULL DEFAULT 0",
|
|
// No default: only meaningful once `acked = 1`; NULL on every
|
|
// pre-migration / never-acked row, which `reap_acked`'s `acked_at < ?`
|
|
// comparison naturally excludes (NULL compares false, not true).
|
|
"ALTER TABLE todos ADD COLUMN acked_at INTEGER",
|
|
];
|
|
|
|
/// One dynamic, subsystem-pushed todo. 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: DateTime<Utc>,
|
|
}
|
|
|
|
/// 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")?;
|
|
for stmt in MIGRATIONS {
|
|
// Ignore "duplicate column name" — the migration already ran.
|
|
// Any other error is logged but doesn't fail open(): the base
|
|
// schema still works and we'd rather keep the harness alive
|
|
// than crash on an upgrade hiccup.
|
|
if let Err(e) = conn.execute(stmt, []) {
|
|
let msg = e.to_string();
|
|
if !msg.contains("duplicate column name") {
|
|
tracing::warn!(error = %msg, stmt, "todos migration failed");
|
|
}
|
|
}
|
|
}
|
|
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). A **materially different**
|
|
/// summary always un-acks + reports `changed = true`, unconditionally —
|
|
/// the agent's earlier dismissal never suppresses a real
|
|
/// re-deterioration.
|
|
///
|
|
/// `reopen_if_acked` decides what an identical re-push against an
|
|
/// **acked** row means: `false` (a reconciler like `disk_watch`) stays
|
|
/// quiet — re-deriving the same still-true summary isn't new
|
|
/// information. `true` (an event-style producer like a scheduled
|
|
/// prompt) reopens anyway — each push is a distinct occurrence, not a
|
|
/// restatement, so acking the *previous* firing doesn't cover *this*
|
|
/// one.
|
|
///
|
|
/// # 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>,
|
|
reopen_if_acked: bool,
|
|
) -> Result<(i64, bool)> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let now = Utc::now().timestamp();
|
|
let existing: Option<(i64, String, Option<String>, bool)> = if key.is_some() {
|
|
conn.query_row(
|
|
"SELECT id, summary, source, acked FROM todos \
|
|
WHERE subsystem = ?1 AND subsystem_key IS ?2",
|
|
params![subsystem, key],
|
|
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
|
|
)
|
|
.ok()
|
|
} else {
|
|
None
|
|
};
|
|
if let Some((id, cur_summary, cur_source, cur_acked)) = existing {
|
|
let content_unchanged = cur_summary == summary && cur_source.as_deref() == source;
|
|
if content_unchanged && !(cur_acked && reopen_if_acked) {
|
|
return Ok((id, false));
|
|
}
|
|
// Either the summary genuinely differs, or it's identical but
|
|
// the caller asked to treat an acked row's re-push as a new
|
|
// occurrence — either way this un-acks + reports changed.
|
|
conn.execute(
|
|
"UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3, \
|
|
acked = 0, acked_at = NULL \
|
|
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 of `subsystem` whose key starts with `prefix`, acked
|
|
/// or not. Returns the number of rows deleted.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn clear_prefix(&self, subsystem: &str, prefix: &str) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
// `substr`, not `LIKE`: a key holds `_` and `%` literally.
|
|
let n = conn.execute(
|
|
"DELETE FROM todos WHERE subsystem = ?1 \
|
|
AND substr(subsystem_key, 1, length(?2)) = ?2",
|
|
params![subsystem, prefix],
|
|
)?;
|
|
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 — i.e. dismisses it from
|
|
/// its own list. Sets `acked` rather than deleting the row (see the
|
|
/// module doc comment for why): the row stays intact so a reconciled
|
|
/// producer's next `upsert` can still tell "unchanged" from "genuinely
|
|
/// different" instead of seeing an empty table and treating a re-poll of
|
|
/// the same condition as brand new. Returns the number of rows newly
|
|
/// acked (0 when the id was unknown, already gone, or already acked —
|
|
/// acking twice is not a new action).
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite update 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(
|
|
"UPDATE todos SET acked = 1, acked_at = ?1 WHERE id = ?2 AND acked = 0",
|
|
params![Utc::now().timestamp(), id],
|
|
)?;
|
|
Ok(n)
|
|
}
|
|
|
|
/// Bulk-ack a specific, explicit list of todo ids in one shot. Exists
|
|
/// for the case a per-id `mark_done` loop isn't worth the round-trips:
|
|
/// an agent that hasn't called `get_loose_ends` in a long stretch (or
|
|
/// one whose producers pile up faster than it triages) can end up with
|
|
/// a backlog large enough that clearing it one call at a time is
|
|
/// impractical. Deliberately **explicit ids, not a `<= threshold`
|
|
/// range** — a reviewer's call on the design: a range-based
|
|
/// bulk-ack risks silently acking something the agent never actually
|
|
/// looked at, since todos are heterogeneous unrelated items (bash /
|
|
/// matrix / forge) rather than a sequentially-read stream the way inbox
|
|
/// messages are. The caller is expected to have looked at each id
|
|
/// (typically the ids `get_loose_ends` just rendered) before passing
|
|
/// them here. Same `acked`-not-deleted semantics as [`Self::mark_done`]
|
|
/// (a reconciled producer's next `upsert` still sees the row to compare
|
|
/// against). Unknown/already-acked ids are silently skipped — same "not
|
|
/// a new action" idempotence as the single-id path. Returns the number
|
|
/// of rows newly acked.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite update failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn mark_done_many(&self, ids: &[i64]) -> Result<usize> {
|
|
if ids.is_empty() {
|
|
return Ok(0);
|
|
}
|
|
let conn = self.conn.lock().unwrap();
|
|
let now = Utc::now().timestamp();
|
|
let mut total = 0usize;
|
|
for id in ids {
|
|
total += conn.execute(
|
|
"UPDATE todos SET acked = 1, acked_at = ?1 WHERE acked = 0 AND id = ?2",
|
|
params![now, id],
|
|
)?;
|
|
}
|
|
Ok(total)
|
|
}
|
|
|
|
/// List todos, newest-updated first. `subsystem = Some(..)` filters to
|
|
/// one producer's set; `None` returns all. Excludes acked rows — once
|
|
/// the agent has dismissed a todo it stays out of its own list, even
|
|
/// though the row itself survives (see `mark_done`).
|
|
///
|
|
/// # 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 acked = 0 AND (?1 IS NULL OR subsystem = ?1) \
|
|
ORDER BY updated_at DESC, id DESC",
|
|
)?;
|
|
let rows = stmt
|
|
.query_map(params![subsystem], |row| {
|
|
let updated_at_secs: i64 = row.get(5)?;
|
|
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: wire_time::from_secs(updated_at_secs),
|
|
})
|
|
})?
|
|
.collect::<rusqlite::Result<Vec<_>>>()?;
|
|
Ok(rows)
|
|
}
|
|
|
|
/// Cheap existence check — `true` when at least one un-acked 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 WHERE acked = 0)",
|
|
[],
|
|
|row| row.get(0),
|
|
)?;
|
|
Ok(any)
|
|
}
|
|
|
|
/// Reap acked rows whose `acked_at` is older than `cutoff` (a unix
|
|
/// timestamp). Called by the harness's hourly vacuum sweep
|
|
/// (`vacuum.rs`), not on any hot path — bounds the table against acked
|
|
/// one-shot todos (bash/matrix/forge) that will never naturally be
|
|
/// upserted again and so would otherwise sit forever. Never touches
|
|
/// un-acked rows. Returns the number of rows deleted.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn reap_acked(&self, cutoff: i64) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute(
|
|
"DELETE FROM todos WHERE acked = 1 AND acked_at < ?1",
|
|
params![cutoff],
|
|
)?;
|
|
Ok(n)
|
|
}
|
|
}
|
|
|
|
#[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 clear_prefix_removes_matching_keys_of_one_subsystem_acked_or_not() {
|
|
let (_dir, s) = store();
|
|
let (acked, _) = s
|
|
.upsert("router", Some("forge:o/r#5:assignment"), "a", None, true)
|
|
.unwrap();
|
|
s.mark_done(acked).unwrap();
|
|
s.upsert("router", Some("forge:o/r#5:review"), "b", None, true)
|
|
.unwrap();
|
|
s.upsert("router", Some("forge:o/r#55:assignment"), "c", None, true) // lint:allow: test key, not a tracker reference
|
|
.unwrap();
|
|
s.upsert("forge", Some("forge:o/r#5:x"), "d", None, true)
|
|
.unwrap();
|
|
s.upsert("router", Some("forge:axb#5:x"), "e", None, true)
|
|
.unwrap();
|
|
assert_eq!(s.clear_prefix("router", "forge:o/r#5:").unwrap(), 2);
|
|
// `_` is no wildcard.
|
|
assert_eq!(s.clear_prefix("router", "forge:a_b#5:").unwrap(), 0);
|
|
let mut left: Vec<String> = s
|
|
.list(None)
|
|
.unwrap()
|
|
.into_iter()
|
|
.map(|t| t.summary)
|
|
.collect();
|
|
left.sort();
|
|
assert_eq!(left, ["c", "d", "e"]);
|
|
}
|
|
|
|
#[test]
|
|
fn keyed_upsert_dedups_and_reports_changed() {
|
|
let (_dir, s) = store();
|
|
let (id1, changed1) = s
|
|
.upsert("matrix", Some("!room:x"), "1 unread", None, false)
|
|
.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, false)
|
|
.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, false)
|
|
.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, false).unwrap();
|
|
let (b, _) = s.upsert("bash", None, "task done", None, false).unwrap();
|
|
assert_ne!(a, b, "keyless pushes are distinct one-offs");
|
|
assert_eq!(s.list(Some("bash")).unwrap().len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn clear_deletes_and_mark_done_hides_from_list() {
|
|
let (_dir, s) = store();
|
|
s.upsert("matrix", Some("!a:x"), "unread", None, false)
|
|
.unwrap();
|
|
let (id, _) = s
|
|
.upsert("forge", Some("pr-1"), "review", None, false)
|
|
.unwrap();
|
|
assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1, "clear deletes");
|
|
assert_eq!(
|
|
s.mark_done(id).unwrap(),
|
|
1,
|
|
"mark_done acks, doesn't delete"
|
|
);
|
|
assert!(
|
|
s.list(None).unwrap().is_empty(),
|
|
"both are gone from the agent-visible list either way"
|
|
);
|
|
}
|
|
|
|
#[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, false).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, false).unwrap();
|
|
s.upsert("matrix", Some("!b:x"), "u", None, false).unwrap();
|
|
s.upsert("forge", Some("pr-1"), "r", None, false).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");
|
|
}
|
|
|
|
/// The whole point of the fix: acking a reconciled todo must not defeat
|
|
/// the producer's own anti-nag comparison. A re-push with an identical
|
|
/// summary after ack stays quiet AND stays out of the list — the `disk_watch`
|
|
/// bug this closes.
|
|
#[test]
|
|
fn acked_row_survives_unchanged_reupsert_and_stays_hidden() {
|
|
let (_dir, s) = store();
|
|
let (id, changed1) = s
|
|
.upsert("disk", Some("usage"), "over 85%", None, false)
|
|
.unwrap();
|
|
assert!(changed1);
|
|
assert_eq!(s.mark_done(id).unwrap(), 1, "first ack succeeds");
|
|
assert!(s.list(None).unwrap().is_empty(), "acked row is hidden");
|
|
|
|
// Same producer, same reconcile tick, identical summary — must NOT
|
|
// look like a new todo just because the agent dismissed it.
|
|
let (id2, changed2) = s
|
|
.upsert("disk", Some("usage"), "over 85%", None, false)
|
|
.unwrap();
|
|
assert_eq!(id, id2, "same row, not a fresh insert");
|
|
assert!(!changed2, "identical summary stays quiet even though acked");
|
|
assert!(
|
|
s.list(None).unwrap().is_empty(),
|
|
"still hidden — an unchanged re-push does not un-ack"
|
|
);
|
|
}
|
|
|
|
/// The other half of the same mechanism: an event-style producer (a
|
|
/// recurring scheduled prompt is the real caller) sets
|
|
/// `reopen_if_acked = true` because each identical-content push is a
|
|
/// distinct occurrence, not a restatement — an acked row must reopen
|
|
/// and report `changed = true` even though the summary didn't move.
|
|
#[test]
|
|
fn reopen_if_acked_resurfaces_an_identical_acked_row() {
|
|
let (_dir, s) = store();
|
|
let (id, changed1) = s
|
|
.upsert(
|
|
"schedule",
|
|
Some("schedule:1"),
|
|
"check the thing",
|
|
None,
|
|
true,
|
|
)
|
|
.unwrap();
|
|
assert!(
|
|
changed1,
|
|
"first push is new → changed regardless of the flag"
|
|
);
|
|
assert_eq!(s.mark_done(id).unwrap(), 1, "agent reviewed fire #1");
|
|
assert!(s.list(None).unwrap().is_empty());
|
|
|
|
// Second fire, byte-identical body, but reopen_if_acked = true —
|
|
// this must NOT collapse into silence the way the reconciler test
|
|
// above expects it to.
|
|
let (id2, changed2) = s
|
|
.upsert(
|
|
"schedule",
|
|
Some("schedule:1"),
|
|
"check the thing",
|
|
None,
|
|
true,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(id, id2, "same keyed row, not a fresh insert");
|
|
assert!(changed2, "acked + identical + reopen_if_acked still wakes");
|
|
assert_eq!(
|
|
s.list(None).unwrap().len(),
|
|
1,
|
|
"reopened row is visible again"
|
|
);
|
|
}
|
|
|
|
/// A materially different summary on an acked row is real information —
|
|
/// it must un-ack, re-surface in `list`, and report `changed = true`.
|
|
#[test]
|
|
fn acked_row_resurfaces_on_genuine_change() {
|
|
let (_dir, s) = store();
|
|
let (id, _) = s
|
|
.upsert("disk", Some("usage"), "over 85%", None, false)
|
|
.unwrap();
|
|
s.mark_done(id).unwrap();
|
|
assert!(s.list(None).unwrap().is_empty());
|
|
|
|
let (id2, changed) = s
|
|
.upsert("disk", Some("usage"), "over 90%", None, false)
|
|
.unwrap();
|
|
assert_eq!(id, id2);
|
|
assert!(changed, "a genuine re-deterioration must not be silenced");
|
|
let visible = s.list(None).unwrap();
|
|
assert_eq!(visible.len(), 1, "un-acked row is visible again");
|
|
assert_eq!(visible[0].summary, "over 90%");
|
|
}
|
|
|
|
/// Acking an already-acked (or unknown) id is a no-op, not a new action.
|
|
#[test]
|
|
fn mark_done_twice_is_idempotent() {
|
|
let (_dir, s) = store();
|
|
let (id, _) = s.upsert("bash", None, "task done", None, false).unwrap();
|
|
assert_eq!(s.mark_done(id).unwrap(), 1);
|
|
assert_eq!(s.mark_done(id).unwrap(), 0, "already acked");
|
|
assert_eq!(s.mark_done(999_999).unwrap(), 0, "unknown id");
|
|
}
|
|
|
|
/// `has_any` must also treat an acked-only table as empty — it gates a
|
|
/// wake against a payload the agent has already drained.
|
|
#[test]
|
|
fn has_any_excludes_acked_rows() {
|
|
let (_dir, s) = store();
|
|
let (id, _) = s.upsert("bash", None, "task done", None, false).unwrap();
|
|
assert!(s.has_any().unwrap());
|
|
s.mark_done(id).unwrap();
|
|
assert!(!s.has_any().unwrap(), "acked-only table reads as empty");
|
|
}
|
|
|
|
/// `mark_done_many` acks exactly the ids passed, leaves the rest
|
|
/// untouched — no threshold/range semantics.
|
|
#[test]
|
|
fn mark_done_many_acks_only_the_listed_ids() {
|
|
let (_dir, s) = store();
|
|
let (id1, _) = s.upsert("bash", None, "task 1", None, false).unwrap();
|
|
let (id2, _) = s.upsert("bash", None, "task 2", None, false).unwrap();
|
|
let (id3, _) = s.upsert("bash", None, "task 3", None, false).unwrap();
|
|
assert_eq!(
|
|
s.mark_done_many(&[id1, id3]).unwrap(),
|
|
2,
|
|
"acks id1 and id3, not id2 — not a range"
|
|
);
|
|
let left = s.list(None).unwrap();
|
|
assert_eq!(left.len(), 1);
|
|
assert_eq!(left[0].id, id2);
|
|
// Idempotent: re-running over the same ids acks nothing new.
|
|
assert_eq!(s.mark_done_many(&[id1, id3]).unwrap(), 0);
|
|
// Empty input is a no-op, not an error.
|
|
assert_eq!(s.mark_done_many(&[]).unwrap(), 0);
|
|
}
|
|
|
|
/// `reap_acked` only removes acked rows past the cutoff — a recent ack
|
|
/// and any un-acked row both survive.
|
|
#[test]
|
|
fn reap_acked_only_removes_old_acked_rows() {
|
|
let (dir, s) = store();
|
|
let (old_id, _) = s.upsert("bash", None, "old task", None, false).unwrap();
|
|
let (recent_id, _) = s.upsert("bash", None, "recent task", None, false).unwrap();
|
|
let (never_acked_id, _) = s
|
|
.upsert("forge", Some("pr-1"), "review", None, false)
|
|
.unwrap();
|
|
s.mark_done(old_id).unwrap();
|
|
s.mark_done(recent_id).unwrap();
|
|
|
|
// Backdate the old row's ack past the cutoff directly — mark_done
|
|
// always acks "now", so simulate age via the raw connection.
|
|
{
|
|
let conn = s.conn.lock().unwrap();
|
|
conn.execute(
|
|
"UPDATE todos SET acked_at = ?1 WHERE id = ?2",
|
|
params![Utc::now().timestamp() - 1000, old_id],
|
|
)
|
|
.unwrap();
|
|
}
|
|
|
|
let removed = s.reap_acked(Utc::now().timestamp() - 500).unwrap();
|
|
assert_eq!(removed, 1, "only the backdated row is past the cutoff");
|
|
|
|
let conn = Connection::open(dir.path().join("todos.sqlite")).unwrap();
|
|
let remaining: i64 = conn
|
|
.query_row("SELECT COUNT(*) FROM todos", [], |r| r.get(0))
|
|
.unwrap();
|
|
assert_eq!(
|
|
remaining, 2,
|
|
"recent ack + never-acked row both survive the reap"
|
|
);
|
|
let _ = never_acked_id;
|
|
}
|
|
}
|