dguiducci/Skald-Circle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: conversation review — a nightly report on a supervised person's conversations
Nightly Build / build (push) Successful in 7m40s
Details

Browse Source 
The first AgentScope::PerSubject system agent, and the reason that scope
exists. Once a night, for each person with a supervision edge, it reads every
message that person and the assistant exchanged since the previous review —
across all their conversations — and writes one report for the people who
supervise them.

Schema (all registry except reports):
- supervision(subject_user_id, supervisor_user_id): the generic §0.1 edge,
  answering both 'whom does a background agent look at' and 'who may read
  what it produced', with real FKs so deleting a user cascades both ways
- system_agent_coverage(agent_id, subject_user_id, covered_through): the
  per-subject watermark that makes 'everything since last time' a window —
  neither system_agent_runs (history for humans) nor system_agent_state
  (advances before the work), and advanced only on a completed pass so a
  crash re-covers instead of skipping
- reports (owner schema, the second two-homes table after memory_docs):
  instance rows land in system.db, deliberately cleartext to the box owner,
  who is the intended reader (§2); the subject cannot see them structurally

The pass reads the subject's database inside a supervisor's runtime, so the
ephemeral session and run row land in the watcher's file; iteration is over
subjects, so two parents watching one child get one review; and the subject
need not be logged in when their space is unencrypted — via the new
UserManager::open_unencrypted, which refuses an encrypted user outright (no
key to be had) and never registers the pool as unlocked.

The agent declares the new AgentMeta flag allow_tools: false, so its turn
gets an empty tool registry — nothing for a prompt injection in the
transcript to call — and produces its report as its final assistant message,
read back from chat_history and parsed (NOTHING_TO_REPORT sentinel, no row on
quiet days). chat_history::conversation_window is the transcript query; its
four filters (non-ephemeral, depth 0, non-synthetic, non-empty) each guard a
specific way the review would otherwise be wrong, and tool calls are absent
by construction.

Cadence is Run at (hour) rather than Interval — 4am local by default — with
due-ness answered inside has_work against the coverage watermark, so a
machine off for three days covers the whole stretch in one pass. Reports
announce ReportCreated on the system bus (no subscriber yet). run_ephemeral_turn
gains a per-pass system_substitutions map, which the review uses to hand the
model the subject's profile under __SUBJECT_PROFILE__ — the system-context
substitutions describe the session owner, the wrong person here.

docs/system-agents.md gains the conversation review section; CLAUDE.md
documents the scope, the tables and the tool-less design.
This commit is contained in:
dguiducci
2026-08-02 20:30:27 +01:00
parent e6818408cb
commit 4f10528368
19 changed files with 2394 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/skald-core/src/db/chat_history.rs
+271
View File
@@ -213,6 +213,136 @@ pub async fn for_stack_since(
rows.into_iter().map(row_to_message).collect()
}
/// One line of a cross-session transcript: a message with the conversation it
/// belongs to. See [`conversation_window`].
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct TranscriptLine {
pub session_id: i64,
pub session_title: Option<String>,
pub source: String,
pub agent_id: String,
pub role: String,
pub content: String,
pub created_at: String,
}
/// Every message this database's owner exchanged with an assistant between
/// `since` (inclusive) and `until` (exclusive), across **all** their
/// conversations, oldest first.
///
/// The window is half-open so consecutive calls tile without overlapping or
/// skipping: today's `until` is tomorrow's `since`. Both bounds are UTC
/// `'YYYY-MM-DD HH:MM:SS'`, the shape `datetime('now')` writes, so they compare
/// as plain strings against `created_at`.
///
/// **Four filters, and each one exists because of a specific way the result would
/// otherwise be wrong:**
///
/// - `is_ephemeral = 0` — a background agent's own throwaway sessions live in the
/// same table. Without this, a pass that reads conversations would read the
/// transcript its *previous* pass was given, and report on itself.
/// - `depth = 0` — only the root frame. Deeper frames are sub-agents talking to
/// each other: machine-to-machine chatter that nobody typed.
/// - `is_synthetic = 0` — turns the machinery injected as if they were the user
/// (notification briefings, job results). Attributing those to the person would
/// be a lie about who said what.
/// - `content <> ''` — an assistant row whose whole content was a tool call.
///
/// **Tool calls and their results are not here at all**, and that is by
/// construction rather than by filter: they live in `chat_llm_tools`, keyed to a
/// message id. So this returns what was *said*, never what was *done* — a web
/// search the assistant ran is invisible, including its query.
pub async fn conversation_window(
pool: &SqlitePool,
since: &str,
until: &str,
limit: i64,
) -> anyhow::Result<Vec<TranscriptLine>> {
// Newest-first with a LIMIT, then reversed: over budget, the window that
// matters is the recent end, not whatever happened to come first.
let mut rows = sqlx::query_as::<_, TranscriptLine>(
"SELECT s.id AS session_id,
s.title AS session_title,
s.source AS source,
s.agent_id AS agent_id,
h.role AS role,
h.content AS content,
h.created_at AS created_at
FROM chat_history h
JOIN chat_sessions_stack st ON st.id = h.session_stack_id
JOIN chat_sessions s ON s.id = st.session_id
WHERE h.created_at >= ? AND h.created_at < ?
AND h.status = 'ok'
AND h.is_synthetic = 0
AND h.content <> ''
AND h.role IN ('user', 'assistant')
AND st.depth = 0
AND s.is_ephemeral = 0
ORDER BY h.created_at DESC, h.id DESC
LIMIT ?",
)
.bind(since)
.bind(until)
.bind(limit)
.fetch_all(pool)
.await?;
rows.reverse();
Ok(rows)
}
/// How many messages [`conversation_window`] would return, without loading them.
/// The cheap look a scheduler takes before deciding a pass is worth opening.
pub async fn conversation_window_count(
pool: &SqlitePool,
since: &str,
until: &str,
) -> anyhow::Result<i64> {
let n = sqlx::query_scalar::<_, i64>(
"SELECT count(*)
FROM chat_history h
JOIN chat_sessions_stack st ON st.id = h.session_stack_id
JOIN chat_sessions s ON s.id = st.session_id
WHERE h.created_at >= ? AND h.created_at < ?
AND h.status = 'ok'
AND h.is_synthetic = 0
AND h.content <> ''
AND h.role IN ('user', 'assistant')
AND st.depth = 0
AND s.is_ephemeral = 0",
)
.bind(since)
.bind(until)
.fetch_one(pool)
.await?;
Ok(n)
}
/// The last thing the assistant said in a session's **root** frame.
///
/// For a caller whose agent produces a document rather than a side effect: the
/// turn's answer is the deliverable, and it has to be read back from the store
/// because `handle_message` returns nothing. Root frame only — the deepest
/// sub-agent's last words are not the session's answer.
pub async fn last_assistant_for_session(
pool: &SqlitePool,
session_id: i64,
) -> anyhow::Result<Option<String>> {
let content = sqlx::query_scalar::<_, String>(
"SELECT h.content
FROM chat_history h
JOIN chat_sessions_stack st ON st.id = h.session_stack_id
WHERE st.session_id = ? AND st.depth = 0
AND h.role = 'assistant' AND h.status = 'ok' AND h.content <> ''
ORDER BY h.id DESC
LIMIT 1",
)
.bind(session_id)
.fetch_optional(pool)
.await?;
Ok(content)
}
/// Returns the most recent ok message for a stack frame, or `None` if empty.
/// Used by Telegram's `/context` command to show last turn's token usage.
pub async fn last_message_for_stack(
@@ -274,3 +404,144 @@ pub async fn estimate_tokens_for_stack(
Ok((total_chars / 4).max(0) as u32)
}
#[cfg(test)]
mod tests {
use super::*;
/// A standalone owner-schema database with one ordinary conversation and one
/// of every thing the window must leave out.
async fn seeded() -> SqlitePool {
let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
crate::db::create_owner_tables(&pool).await.unwrap();
let q = |sql: &'static str| sqlx::query(sql).execute(&pool);
// A real conversation, and a second one the same day.
q("INSERT INTO chat_sessions (id, title, source, agent_id, is_ephemeral) VALUES (1, 'Homework', 'web', 'kid', 0)").await.unwrap();
q("INSERT INTO chat_sessions (id, title, source, agent_id, is_ephemeral) VALUES (2, NULL, 'telegram', 'kid', 0)").await.unwrap();
// A background agent's throwaway session — the one that would make a
// review read its own previous pass.
q("INSERT INTO chat_sessions (id, title, source, agent_id, is_ephemeral) VALUES (3, 'review', 'conversation-review', 'conversation-review', 1)").await.unwrap();
q("INSERT INTO chat_sessions_stack (id, session_id, depth) VALUES (1, 1, 0)").await.unwrap();
q("INSERT INTO chat_sessions_stack (id, session_id, depth) VALUES (2, 2, 0)").await.unwrap();
q("INSERT INTO chat_sessions_stack (id, session_id, depth) VALUES (3, 3, 0)").await.unwrap();
// A sub-agent frame of the real conversation.
q("INSERT INTO chat_sessions_stack (id, session_id, depth) VALUES (4, 1, 1)").await.unwrap();
let msg = |stack: i64, role: &'static str, content: &'static str, at: &'static str,
synthetic: i64, status: &'static str| {
sqlx::query(
"INSERT INTO chat_history (session_stack_id, role, content, created_at, is_synthetic, status)
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(stack).bind(role).bind(content).bind(at).bind(synthetic).bind(status)
.execute(&pool)
};
msg(1, "user", "kept: in window", "2026-07-28 21:00:00", 0, "ok").await.unwrap();
msg(1, "assistant", "kept: the reply", "2026-07-28 21:00:30", 0, "ok").await.unwrap();
msg(2, "user", "kept: other session", "2026-07-29 02:00:00", 0, "ok").await.unwrap();
msg(1, "user", "dropped: before", "2026-07-27 10:00:00", 0, "ok").await.unwrap();
msg(1, "user", "dropped: after", "2026-07-30 10:00:00", 0, "ok").await.unwrap();
msg(3, "user", "dropped: ephemeral", "2026-07-28 22:00:00", 0, "ok").await.unwrap();
msg(4, "assistant", "dropped: sub-agent", "2026-07-28 22:00:00", 0, "ok").await.unwrap();
msg(1, "user", "dropped: synthetic", "2026-07-28 22:00:00", 1, "ok").await.unwrap();
msg(1, "assistant", "dropped: failed", "2026-07-28 22:00:00", 0, "failed").await.unwrap();
msg(1, "assistant", "", "2026-07-28 22:00:00", 0, "ok").await.unwrap();
msg(1, "agent", "dropped: agent role", "2026-07-28 22:00:00", 0, "ok").await.unwrap();
pool
}
const SINCE: &str = "2026-07-28 04:00:00";
const UNTIL: &str = "2026-07-29 04:00:00";
/// Each exclusion is a way the review would otherwise be wrong; assert them
/// together, because it is the *set* that defines "what was said".
#[tokio::test]
async fn the_window_keeps_only_what_was_said_in_it() {
let pool = seeded().await;
let lines = conversation_window(&pool, SINCE, UNTIL, 100).await.unwrap();
let kept: Vec<&str> = lines.iter().map(|l| l.content.as_str()).collect();
assert_eq!(kept, vec![
"kept: in window",
"kept: the reply",
"kept: other session",
], "everything else is a way the transcript would lie");
// The count is the same question, asked cheaply.
assert_eq!(conversation_window_count(&pool, SINCE, UNTIL).await.unwrap(), 3);
// Oldest first, and each line carries the conversation it belongs to.
assert_eq!(lines[0].session_id, 1);
assert_eq!(lines[0].session_title.as_deref(), Some("Homework"));
assert_eq!(lines[2].session_id, 2);
assert_eq!(lines[2].source, "telegram");
assert!(lines[2].session_title.is_none());
}
/// Half-open, so consecutive windows tile: a message exactly on the boundary
/// belongs to the later window, never to both and never to neither.
#[tokio::test]
async fn the_window_is_half_open() {
let pool = seeded().await;
sqlx::query(
"INSERT INTO chat_history (session_stack_id, role, content, created_at)
VALUES (1, 'user', 'exactly on the edge', ?)",
)
.bind(UNTIL)
.execute(&pool).await.unwrap();
let before = conversation_window(&pool, SINCE, UNTIL, 100).await.unwrap();
assert!(!before.iter().any(|l| l.content == "exactly on the edge"),
"`until` is exclusive");
let after = conversation_window(&pool, UNTIL, "2026-07-30 04:00:00", 100).await.unwrap();
assert!(after.iter().any(|l| l.content == "exactly on the edge"),
"`since` is inclusive, so nothing falls between two windows");
}
/// Over budget, the recent end is what survives — a truncated review of last
/// night beats a complete review of last month.
#[tokio::test]
async fn a_capped_window_keeps_the_most_recent_messages() {
let pool = seeded().await;
let lines = conversation_window(&pool, SINCE, UNTIL, 2).await.unwrap();
let kept: Vec<&str> = lines.iter().map(|l| l.content.as_str()).collect();
assert_eq!(kept, vec!["kept: the reply", "kept: other session"]);
// The count ignores the cap, which is how the caller knows it truncated.
assert_eq!(conversation_window_count(&pool, SINCE, UNTIL).await.unwrap(), 3);
}
#[tokio::test]
async fn the_last_assistant_message_comes_from_the_root_frame() {
let pool = seeded().await;
// Session 1's newest root-frame assistant line, not the sub-agent's.
sqlx::query(
"INSERT INTO chat_history (session_stack_id, role, content, created_at)
VALUES (1, 'assistant', 'the answer', '2026-07-29 03:00:00')",
)
.execute(&pool).await.unwrap();
sqlx::query(
"INSERT INTO chat_history (session_stack_id, role, content, created_at)
VALUES (4, 'assistant', 'sub-agent chatter', '2026-07-29 03:30:00')",
)
.execute(&pool).await.unwrap();
assert_eq!(
last_assistant_for_session(&pool, 1).await.unwrap().as_deref(),
Some("the answer"),
);
// A session that never got an answer says so rather than inventing one.
assert!(last_assistant_for_session(&pool, 2).await.unwrap().is_none());
assert!(last_assistant_for_session(&pool, 99).await.unwrap().is_none());
}
}
crates/skald-core/src/db/mod.rs
+140
View File
@@ -24,12 +24,15 @@ pub mod oauth_providers;
pub mod plugins;
pub mod plugin_access;
pub mod plugin_user_configs;
pub mod reports;
pub mod role_capabilities;
pub mod roles;
pub mod scheduled_jobs;
pub mod scratchpad;
pub mod shared_folders;
pub mod sources;
pub mod supervision;
pub mod system_agent_coverage;
pub mod system_agent_runs;
pub mod system_agent_state;
pub mod tool_permission_groups;
@@ -679,6 +682,76 @@ async fn create_registry_tables(pool: &SqlitePool) -> Result<()> {
.execute(pool)
.await?;
// The supervision edge (§0.1): one person's activity may be read on another's
// behalf. **A generic edge between two users, and nothing more** — the domain
// reading of it ("a parent watches a child") lives in the seed data and the UI
// copy, never here, so a pivot to a mentor watching a trainee, or a care worker
// watching a resident, renames nothing.
//
// It answers two questions with one table, which is why it is an edge and not a
// per-agent list of subjects: *whom does a background agent look at* (the
// distinct subjects) and *who may read what it produced* (the supervisors of a
// given subject). The second is what the reports' `audience = 'supervisors'`
// resolves against.
//
// Both FKs are registry→registry (same file), so they are allowed and the
// cascade is real: deleting a user takes their edges with them, in both
// directions.
sqlx::query(
"CREATE TABLE IF NOT EXISTS supervision (
subject_user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
supervisor_user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
PRIMARY KEY (subject_user_id, supervisor_user_id)
)",
)
.execute(pool)
.await?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_supervision_supervisor
ON supervision(supervisor_user_id)",
)
.execute(pool)
.await?;
// How far a background agent has *processed* a subject — the watermark that
// makes "everything since last time" a well-defined window.
//
// **Not `system_agent_runs`, and not `system_agent_state`**, though it sits
// between them and the difference is the whole reason it exists:
//
// `system_agent_state` when an agent last *attempted* a pass. Advances on
// every tick, including idle ones, and is marked
// *before* the work — so it can never delimit the
// window the work is about.
// this table how far the work actually got. Advances **only on a
// completed pass**, so a crash mid-pass re-covers the
// same stretch next time. For a review, a duplicate
// report is a nuisance and a skipped window is a blind
// spot: at-least-once is the only acceptable direction.
//
// The obvious alternative — deriving the watermark from the last report's
// `period_end` — fails on a single ordinary action: a supervisor deleting an
// old report would move the scheduler's window back and regenerate the very
// report they discarded. A document is the user's to delete; scheduler state is
// not, so they cannot be the same row.
//
// Registry, not owner, for a reason specific to how these passes run: the pass
// executes inside *some* supervisor's runtime, and which one depends on who is
// logged in tonight. A watermark in the acting user's file would give one
// subject two unsynchronised clocks.
sqlx::query(
"CREATE TABLE IF NOT EXISTS system_agent_coverage (
agent_id TEXT NOT NULL,
subject_user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
covered_through TEXT NOT NULL, -- UTC 'YYYY-MM-DD HH:MM:SS'
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
PRIMARY KEY (agent_id, subject_user_id)
)",
)
.execute(pool)
.await?;
Ok(())
}
@@ -1128,6 +1201,68 @@ pub async fn create_owner_tables(pool: &SqlitePool) -> Result<()> {
sqlx::query(trigger).execute(pool).await?;
}
// Reports — the documents system agents write about a stretch of time
// (blueprint §13). Like `memory_docs` above, one owner schema backs **two
// homes**, and which file a row lands in *is* its audience:
//
// `{userid}.db` a report that belongs to that user, about that user —
// their weekly "what you struggled to get done" digest.
// Behind SQLCipher: nobody else can read it, admin included.
// `system.db` an instance report, written about someone *for* the
// people who supervise them. Cleartext to whoever owns the
// box, deliberately — they are the intended reader (§2).
//
// That split is why nothing here filters by reader: a report's subject can
// never see an instance report about them, because their tools only ever
// touch their own pool. The invisibility is structural, not a rule someone
// has to remember in each query.
//
// The producer's scope decides the file with no extra concept:
// `AgentScope::PerUser` writes into `ctx.pool`, `AgentScope::Instance` into
// the registry pool the agent already holds.
//
// `subject_user_id` / `producer_user_id` / `run_id` are **bare** columns, not
// foreign keys: `users` lives in the registry (an owner→registry FK would
// fail every INSERT), and for an instance row the `system_agent_runs` trace
// sits in the *acting* user's file. They are snapshots, and a deleted user
// leaves them dangling on purpose — the report outlives the account.
//
// `kind` is free-form producer-declared text, never an enum (§0.1). Rows are
// immutable once written: the only UPDATE is the read acknowledgement.
sqlx::query(
"CREATE TABLE IF NOT EXISTS reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
kind TEXT NOT NULL, -- producer-declared type, not an enum
title TEXT NOT NULL,
summary TEXT, -- one line: lists + notification text
body TEXT NOT NULL DEFAULT '', -- markdown
severity TEXT NOT NULL DEFAULT 'info', -- 'info' | 'notice' | 'alert'
subject_user_id TEXT, -- who it is about (bare snapshot)
audience TEXT NOT NULL DEFAULT 'owner', -- 'owner' | 'admins' | 'supervisors'
period_start TEXT, -- the window it covers
period_end TEXT,
produced_by TEXT NOT NULL, -- system agent id
producer_user_id TEXT, -- whose runtime ran the pass
run_id INTEGER, -- system_agent_runs.id (bare snapshot)
metadata TEXT, -- JSON counters; never contents
read_at TEXT, -- shared acknowledgement: first reader wins
read_by TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)",
)
.execute(pool)
.await?;
// Listing is always newest-first, optionally narrowed to one subject. `kind`
// is deliberately unindexed: a handful of rows a week means a scan is
// cheaper than the index it would need.
for index in [
"CREATE INDEX IF NOT EXISTS idx_reports_created ON reports(created_at DESC, id DESC)",
"CREATE INDEX IF NOT EXISTS idx_reports_subject ON reports(subject_user_id, created_at DESC)",
] {
sqlx::query(index).execute(pool).await?;
}
Ok(())
}
@@ -1186,6 +1321,11 @@ mod tests {
one("INSERT INTO llm_request_payloads (request_id, request_json) VALUES ('r1', '{}')").await.unwrap();
// Fires the AFTER INSERT trigger into the external-content FTS5 table.
one("INSERT INTO memory_docs (path, content) VALUES ('notes/x.md', 'hello world')").await.unwrap();
// Bare `subject_user_id` / `producer_user_id` (registry `users`) and a
// bare `run_id` that points at no row in this file — an FK on any of the
// three would die right here.
one("INSERT INTO reports (kind, title, body, produced_by, subject_user_id, producer_user_id, run_id)
VALUES ('conversation-review', 't', 'b', 'agent', 'u-absent', 'u-also-absent', 4242)").await.unwrap();
// ...and the FTS index actually answers a MATCH.
let (hits,): (i64,) = sqlx::query_as(
crates/skald-core/src/db/reports.rs
+435
View File
@@ -0,0 +1,435 @@
//! Accessor for `reports` — the documents system agents write about a stretch
//! of time (blueprint §13).
//!
//! **The pool is the audience.** Like [`super::memory_docs`], one owner schema
//! backs two homes and the file a row lands in decides who may read it: a user's
//! own encrypted database holds the reports that belong to them, `system.db`
//! holds the instance ones — written about someone, for the people who supervise
//! them. Nothing in here filters by reader, because there is nothing to filter:
//! a subject's tools only ever reach their own pool. The separation is
//! structural, not a predicate someone has to remember to add.
//!
//! Which file a producer writes into falls out of its own scope with no new
//! concept: `AgentScope::PerUser` passes `ctx.pool`, `AgentScope::Instance`
//! passes the registry pool it already holds.
//!
//! **A report is immutable.** It is a snapshot of a window that has closed, so
//! there is no `update`: the only write after [`create`] is [`mark_read`], and
//! even that is once — see its "first reader wins" note.
use anyhow::Result;
use sqlx::SqlitePool;
/// Severity, in ascending order of "someone should look at this". Free text in
/// the column; these are the vocabulary the UI knows how to render.
pub const SEVERITY_INFO: &str = "info";
pub const SEVERITY_NOTICE: &str = "notice";
pub const SEVERITY_ALERT: &str = "alert";
/// The report belongs to whoever owns the file it is in — the default, and the
/// only meaningful value inside a `{userid}.db`.
pub const AUDIENCE_OWNER: &str = "owner";
/// An instance report (`system.db`) for the admins.
pub const AUDIENCE_ADMINS: &str = "admins";
/// An instance report for whoever holds a [`super::supervision`] edge over its
/// `subject_user_id` — the audience that is *computed*, not enumerated, so adding
/// a second parent to the edge widens the readership of every past report at once.
pub const AUDIENCE_SUPERVISORS: &str = "supervisors";
/// A report with its body. Use [`ReportSummary`] for listings — the body is the
/// bulk of the row and a list never renders it.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Report {
pub id: i64,
pub kind: String,
pub title: String,
pub summary: Option<String>,
pub body: String,
pub severity: String,
pub subject_user_id: Option<String>,
pub audience: String,
pub period_start: Option<String>,
pub period_end: Option<String>,
pub produced_by: String,
pub producer_user_id: Option<String>,
pub run_id: Option<i64>,
pub metadata: Option<String>,
pub read_at: Option<String>,
pub read_by: Option<String>,
pub created_at: String,
}
/// A listing row: everything but `body`.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct ReportSummary {
pub id: i64,
pub kind: String,
pub title: String,
pub summary: Option<String>,
pub severity: String,
pub subject_user_id: Option<String>,
pub audience: String,
pub period_start: Option<String>,
pub period_end: Option<String>,
pub produced_by: String,
pub producer_user_id: Option<String>,
pub run_id: Option<i64>,
pub metadata: Option<String>,
pub read_at: Option<String>,
pub read_by: Option<String>,
pub created_at: String,
}
/// The fields a producer supplies. `kind`, `title`, `body` and `produced_by` are
/// the ones with no sensible default; everything else has one.
#[derive(Debug, Clone)]
pub struct NewReport<'a> {
/// Producer-declared type — data, never an enum (§0.1). Groups the UI.
pub kind: &'a str,
pub title: &'a str,
/// One line for lists and for the notification that announces it.
pub summary: Option<&'a str>,
/// Markdown.
pub body: &'a str,
pub severity: &'a str,
/// Who the report is about. `None` for a report about nobody in particular.
pub subject_user_id: Option<&'a str>,
pub audience: &'a str,
/// The window covered, ISO-8601. Both `None` for a point-in-time report.
pub period_start: Option<&'a str>,
pub period_end: Option<&'a str>,
/// The system agent's id.
pub produced_by: &'a str,
/// Whose runtime ran the pass — for an instance report, not the subject.
pub producer_user_id: Option<&'a str>,
/// `system_agent_runs.id`. A bare snapshot: for an instance report that row
/// lives in the acting user's file, not this one.
pub run_id: Option<i64>,
/// JSON counters. Never contents — the body is the only place text belongs.
pub metadata: Option<&'a str>,
}
/// Hand-written, not derived, for the same reason `RoleAttrs`'s is: a derived
/// `Default` would leave `severity` and `audience` empty strings, and both are
/// `NOT NULL` columns whose value the UI dispatches on. The defaults are the
/// quiet, narrow ones — informational, and readable only by the file's owner.
impl Default for NewReport<'_> {
fn default() -> Self {
Self {
kind: "",
title: "",
summary: None,
body: "",
severity: SEVERITY_INFO,
subject_user_id: None,
audience: AUDIENCE_OWNER,
period_start: None,
period_end: None,
produced_by: "",
producer_user_id: None,
run_id: None,
metadata: None,
}
}
}
/// How to narrow a [`list`]. All-`None` lists everything, newest first.
#[derive(Debug, Clone, Default)]
pub struct ListFilter<'a> {
pub kind: Option<&'a str>,
pub subject_user_id: Option<&'a str>,
/// Only reports nobody has acknowledged yet.
pub unread_only: bool,
/// Only reports created at or after this ISO timestamp.
pub since: Option<&'a str>,
pub limit: Option<i64>,
}
const SUMMARY_COLS: &str = "id, kind, title, summary, severity, subject_user_id, audience, \
period_start, period_end, produced_by, producer_user_id, run_id, metadata, \
read_at, read_by, created_at";
/// Write a report. Returns its id.
pub async fn create(pool: &SqlitePool, report: &NewReport<'_>) -> Result<i64> {
let id = sqlx::query_scalar::<_, i64>(
"INSERT INTO reports
(kind, title, summary, body, severity, subject_user_id, audience,
period_start, period_end, produced_by, producer_user_id, run_id, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
RETURNING id",
)
.bind(report.kind)
.bind(report.title)
.bind(report.summary)
.bind(report.body)
.bind(report.severity)
.bind(report.subject_user_id)
.bind(report.audience)
.bind(report.period_start)
.bind(report.period_end)
.bind(report.produced_by)
.bind(report.producer_user_id)
.bind(report.run_id)
.bind(report.metadata)
.fetch_one(pool)
.await?;
Ok(id)
}
/// Fetch one report, body included.
pub async fn get(pool: &SqlitePool, id: i64) -> Result<Option<Report>> {
let row = sqlx::query_as::<_, Report>(
"SELECT id, kind, title, summary, body, severity, subject_user_id, audience,
period_start, period_end, produced_by, producer_user_id, run_id, metadata,
read_at, read_by, created_at
FROM reports WHERE id = ?",
)
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row)
}
/// List reports newest first, without their bodies.
///
/// `id DESC` breaks ties: `created_at` has second resolution, and two reports of
/// the same pass land inside one tick often enough that the order would
/// otherwise be whatever SQLite felt like.
pub async fn list(pool: &SqlitePool, filter: &ListFilter<'_>) -> Result<Vec<ReportSummary>> {
// The SQL text is assembled only from these literals — every caller-supplied
// value goes through a bind, in the same order the predicates were pushed.
let mut predicates: Vec<&str> = Vec::new();
if filter.kind.is_some() { predicates.push("kind = ?"); }
if filter.subject_user_id.is_some() { predicates.push("subject_user_id = ?"); }
if filter.unread_only { predicates.push("read_at IS NULL"); }
if filter.since.is_some() { predicates.push("created_at >= ?"); }
let mut sql = format!("SELECT {SUMMARY_COLS} FROM reports");
if !predicates.is_empty() {
sql.push_str(" WHERE ");
sql.push_str(&predicates.join(" AND "));
}
sql.push_str(" ORDER BY created_at DESC, id DESC");
if filter.limit.is_some() {
sql.push_str(" LIMIT ?");
}
let mut query = sqlx::query_as::<_, ReportSummary>(sqlx::AssertSqlSafe(sql));
if let Some(kind) = filter.kind { query = query.bind(kind); }
if let Some(subject) = filter.subject_user_id { query = query.bind(subject); }
if let Some(since) = filter.since { query = query.bind(since); }
if let Some(limit) = filter.limit { query = query.bind(limit); }
Ok(query.fetch_all(pool).await?)
}
/// How many reports nobody has acknowledged — the badge count.
pub async fn unread_count(pool: &SqlitePool) -> Result<i64> {
let n = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM reports WHERE read_at IS NULL")
.fetch_one(pool)
.await?;
Ok(n)
}
/// Acknowledge a report on behalf of `user_id`. Returns whether this call is the
/// one that marked it.
///
/// **First reader wins, and that is the semantics, not an optimisation.** An
/// instance report can have several readers (two admins); an alert about the
/// same evening is one thing to deal with, dealt with once. The `read_at IS
/// NULL` guard makes the write idempotent and keeps `read_by` pointing at
/// whoever actually took it, instead of whoever opened it last.
pub async fn mark_read(pool: &SqlitePool, id: i64, user_id: &str) -> Result<bool> {
let n = sqlx::query(
"UPDATE reports SET read_at = datetime('now'), read_by = ?
WHERE id = ? AND read_at IS NULL",
)
.bind(user_id)
.bind(id)
.execute(pool)
.await?
.rows_affected();
Ok(n > 0)
}
/// Delete a report. Returns whether a row was removed.
pub async fn delete(pool: &SqlitePool, id: i64) -> Result<bool> {
let n = sqlx::query("DELETE FROM reports WHERE id = ?")
.bind(id)
.execute(pool)
.await?
.rows_affected();
Ok(n > 0)
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::PathBuf;
/// A standalone owner-schema database in a throwaway temp dir, as in
/// `memory_docs`: `tag` plus a counter keep parallel tests off one file.
async fn owner_pool(tag: &str) -> (SqlitePool, PathBuf) {
use std::sync::atomic::{AtomicU64, Ordering};
static SEQ: AtomicU64 = AtomicU64::new(0);
let n = SEQ.fetch_add(1, Ordering::Relaxed);
let dir = std::env::temp_dir()
.join(format!("skald-reports-{}-{tag}-{n}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let pool = crate::db::create_user_pool(&dir.join("owner.db"), None).await.unwrap();
(pool, dir)
}
#[tokio::test]
async fn create_stores_every_field_and_defaults_the_rest() {
let (pool, dir) = owner_pool("create").await;
// The minimum a producer must supply: the defaults fill the rest.
let bare = create(&pool, &NewReport {
kind: "usage-digest",
title: "Your week with the assistant",
body: "You asked for a calendar three times and got nowhere.",
produced_by: "usage-digest",
..Default::default()
}).await.unwrap();
let got = get(&pool, bare).await.unwrap().unwrap();
assert_eq!(got.severity, SEVERITY_INFO, "a bare report is informational");
assert_eq!(got.audience, AUDIENCE_OWNER, "...and readable only by its file's owner");
assert!(got.subject_user_id.is_none());
assert!(got.read_at.is_none(), "a fresh report is unread");
assert!(!got.created_at.is_empty());
// A full instance report: subject and run_id are bare snapshots, so
// neither has to exist anywhere in this file.
let full = create(&pool, &NewReport {
kind: "conversation-review",
title: "Something to look at",
summary: Some("one line for the notification"),
body: "# Detail\n\nnarrated, not quoted.",
severity: SEVERITY_ALERT,
subject_user_id: Some("u-nobody"),
audience: AUDIENCE_ADMINS,
period_start: Some("2026-07-28T00:00:00Z"),
period_end: Some("2026-07-29T00:00:00Z"),
produced_by: "conversation-review",
producer_user_id: Some("u-someone-else"),
run_id: Some(4242),
metadata: Some(r#"{"sessions_scanned":7}"#),
}).await.unwrap();
let got = get(&pool, full).await.unwrap().unwrap();
assert_eq!(got.severity, SEVERITY_ALERT);
assert_eq!(got.audience, AUDIENCE_ADMINS);
assert_eq!(got.subject_user_id.as_deref(), Some("u-nobody"));
assert_eq!(got.producer_user_id.as_deref(), Some("u-someone-else"));
assert_eq!(got.run_id, Some(4242));
assert_eq!(got.period_end.as_deref(), Some("2026-07-29T00:00:00Z"));
assert!(got.body.starts_with("# Detail"));
assert!(get(&pool, 9999).await.unwrap().is_none());
pool.close().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn list_narrows_and_orders_newest_first() {
let (pool, dir) = owner_pool("list").await;
let mk = |kind: &'static str, subject: Option<&'static str>| {
let pool = pool.clone();
async move {
create(&pool, &NewReport {
kind,
title: "t",
body: "b",
subject_user_id: subject,
produced_by: "agent",
..Default::default()
}).await.unwrap()
}
};
let first = mk("usage-digest", None).await;
let second = mk("conversation-review", Some("u-kid")).await;
let third = mk("conversation-review", Some("u-other")).await;
// Newest first, with `id` breaking the same-second tie.
let all = list(&pool, &ListFilter::default()).await.unwrap();
assert_eq!(all.iter().map(|r| r.id).collect::<Vec<_>>(), vec![third, second, first]);
let by_kind = list(&pool, &ListFilter {
kind: Some("conversation-review"), ..Default::default()
}).await.unwrap();
assert_eq!(by_kind.iter().map(|r| r.id).collect::<Vec<_>>(), vec![third, second]);
let by_subject = list(&pool, &ListFilter {
subject_user_id: Some("u-kid"), ..Default::default()
}).await.unwrap();
assert_eq!(by_subject.len(), 1);
assert_eq!(by_subject[0].id, second);
// Two filters compose, and `limit` applies after the ordering.
let both = list(&pool, &ListFilter {
kind: Some("conversation-review"),
subject_user_id: Some("u-other"),
..Default::default()
}).await.unwrap();
assert_eq!(both.len(), 1);
assert_eq!(both[0].id, third);
let capped = list(&pool, &ListFilter { limit: Some(2), ..Default::default() }).await.unwrap();
assert_eq!(capped.iter().map(|r| r.id).collect::<Vec<_>>(), vec![third, second]);
// `since` is inclusive, and a future timestamp excludes everything.
assert!(list(&pool, &ListFilter {
since: Some("2999-01-01T00:00:00Z"), ..Default::default()
}).await.unwrap().is_empty());
pool.close().await;
let _ = std::fs::remove_dir_all(&dir);
}
/// Several admins share one instance report; whoever gets there first is the
/// one who took it, and the count reflects the household, not each reader.
#[tokio::test]
async fn acknowledgement_is_shared_and_first_writer_wins() {
let (pool, dir) = owner_pool("read").await;
let id = create(&pool, &NewReport {
kind: "conversation-review", title: "t", body: "b",
audience: AUDIENCE_ADMINS, produced_by: "agent", ..Default::default()
}).await.unwrap();
let other = create(&pool, &NewReport {
kind: "usage-digest", title: "t2", body: "b", produced_by: "agent", ..Default::default()
}).await.unwrap();
assert_eq!(unread_count(&pool).await.unwrap(), 2);
assert_eq!(list(&pool, &ListFilter { unread_only: true, ..Default::default() })
.await.unwrap().len(), 2);
assert!(mark_read(&pool, id, "u-anna").await.unwrap(), "the first reader takes it");
assert!(!mark_read(&pool, id, "u-bruno").await.unwrap(), "the second changes nothing");
let got = get(&pool, id).await.unwrap().unwrap();
assert_eq!(got.read_by.as_deref(), Some("u-anna"), "read_by keeps whoever took it");
assert!(got.read_at.is_some());
assert_eq!(unread_count(&pool).await.unwrap(), 1);
let unread = list(&pool, &ListFilter { unread_only: true, ..Default::default() })
.await.unwrap();
assert_eq!(unread.len(), 1);
assert_eq!(unread[0].id, other);
assert!(!mark_read(&pool, 9999, "u-anna").await.unwrap(), "an absent report marks nothing");
assert!(delete(&pool, id).await.unwrap());
assert!(!delete(&pool, id).await.unwrap(), "a second delete is a no-op");
assert!(get(&pool, id).await.unwrap().is_none());
pool.close().await;
let _ = std::fs::remove_dir_all(&dir);
}
}
crates/skald-core/src/db/supervision.rs
+183
View File
@@ -0,0 +1,183 @@
//! Accessor for `supervision` — the edge that says one person's activity may be
//! read on another's behalf (§0.1).
//!
//! Deliberately anaemic: an edge, two directions, no attributes. It carries no
//! notion of *what* the supervisor may see, because that belongs to whatever
//! reads it — today one background agent, tomorrow a read gate on the reports it
//! writes. Putting "may read conversations" / "may read memory" on the row here
//! would be inventing a permission model before anything asks for one.
//!
//! Registry table, so both foreign keys are real and the cascade is too: deleting
//! either user removes the edge.
use anyhow::Result;
use sqlx::SqlitePool;
/// One edge.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct SupervisionEdge {
pub subject_user_id: String,
pub supervisor_user_id: String,
pub created_at: String,
}
/// Every user somebody supervises, in a stable order.
///
/// The order matters more than it looks: it is the order a background pass walks
/// its subjects in, and a stable one makes a partial pass (the process died
/// halfway) resume predictably instead of favouring whoever sorts first by
/// accident.
pub async fn subjects(pool: &SqlitePool) -> Result<Vec<String>> {
let rows = sqlx::query_scalar::<_, String>(
"SELECT DISTINCT subject_user_id FROM supervision ORDER BY subject_user_id",
)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Who supervises `subject`, in a stable order.
pub async fn supervisors_of(pool: &SqlitePool, subject: &str) -> Result<Vec<String>> {
let rows = sqlx::query_scalar::<_, String>(
"SELECT supervisor_user_id FROM supervision
WHERE subject_user_id = ?
ORDER BY supervisor_user_id",
)
.bind(subject)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Whom `supervisor` watches, in a stable order.
pub async fn subjects_of(pool: &SqlitePool, supervisor: &str) -> Result<Vec<String>> {
let rows = sqlx::query_scalar::<_, String>(
"SELECT subject_user_id FROM supervision
WHERE supervisor_user_id = ?
ORDER BY subject_user_id",
)
.bind(supervisor)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Is this edge present? The question a future read gate on a report asks.
pub async fn supervises(pool: &SqlitePool, supervisor: &str, subject: &str) -> Result<bool> {
let n = sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM supervision
WHERE supervisor_user_id = ? AND subject_user_id = ?",
)
.bind(supervisor)
.bind(subject)
.fetch_one(pool)
.await?;
Ok(n > 0)
}
/// Every edge, for an admin listing.
pub async fn list(pool: &SqlitePool) -> Result<Vec<SupervisionEdge>> {
let rows = sqlx::query_as::<_, SupervisionEdge>(
"SELECT subject_user_id, supervisor_user_id, created_at FROM supervision
ORDER BY subject_user_id, supervisor_user_id",
)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Add an edge. Idempotent on the primary key.
///
/// Refuses a self-edge: supervising yourself would make every subject their own
/// supervisor, which is not a special case anyone wants — it is the pass reading
/// its own runtime's data and reporting it to itself.
pub async fn add(pool: &SqlitePool, subject: &str, supervisor: &str) -> Result<()> {
if subject == supervisor {
anyhow::bail!("a user cannot supervise themselves");
}
sqlx::query(
"INSERT INTO supervision (subject_user_id, supervisor_user_id)
VALUES (?, ?)
ON CONFLICT(subject_user_id, supervisor_user_id) DO NOTHING",
)
.bind(subject)
.bind(supervisor)
.execute(pool)
.await?;
Ok(())
}
/// Remove an edge. Returns whether one was there.
pub async fn remove(pool: &SqlitePool, subject: &str, supervisor: &str) -> Result<bool> {
let n = sqlx::query(
"DELETE FROM supervision WHERE subject_user_id = ? AND supervisor_user_id = ?",
)
.bind(subject)
.bind(supervisor)
.execute(pool)
.await?
.rows_affected();
Ok(n > 0)
}
#[cfg(test)]
mod tests {
use super::*;
/// A registry pool with two users to hang edges off — the FKs are enforced,
/// so the rows have to exist.
async fn registry() -> SqlitePool {
let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
crate::db::create_registry_tables(&pool).await.unwrap();
sqlx::query("INSERT INTO roles (id, label, permission_group) VALUES ('member', 'Member', 'default')")
.execute(&pool).await.unwrap();
for (id, name) in [("u-anna", "anna"), ("u-bruno", "bruno"), ("u-kid", "kid")] {
sqlx::query(
"INSERT INTO users (id, username, display_name, role_id, encrypted)
VALUES (?, ?, ?, 'member', 0)",
)
.bind(id).bind(name).bind(name)
.execute(&pool).await.unwrap();
}
pool
}
#[tokio::test]
async fn an_edge_reads_from_both_ends() {
let pool = registry().await;
add(&pool, "u-kid", "u-anna").await.unwrap();
add(&pool, "u-kid", "u-bruno").await.unwrap();
add(&pool, "u-kid", "u-anna").await.unwrap(); // idempotent
assert_eq!(subjects(&pool).await.unwrap(), vec!["u-kid"]);
assert_eq!(supervisors_of(&pool, "u-kid").await.unwrap(), vec!["u-anna", "u-bruno"]);
assert_eq!(subjects_of(&pool, "u-anna").await.unwrap(), vec!["u-kid"]);
assert!(supervises(&pool, "u-anna", "u-kid").await.unwrap());
assert!(!supervises(&pool, "u-kid", "u-anna").await.unwrap(), "the edge is directed");
assert_eq!(list(&pool).await.unwrap().len(), 2);
assert!(remove(&pool, "u-kid", "u-anna").await.unwrap());
assert!(!remove(&pool, "u-kid", "u-anna").await.unwrap());
assert_eq!(supervisors_of(&pool, "u-kid").await.unwrap(), vec!["u-bruno"]);
// One supervisor left, so the subject is still watched.
assert_eq!(subjects(&pool).await.unwrap(), vec!["u-kid"]);
}
#[tokio::test]
async fn a_self_edge_is_refused() {
let pool = registry().await;
assert!(add(&pool, "u-anna", "u-anna").await.is_err());
}
#[tokio::test]
async fn deleting_a_user_takes_their_edges_from_both_directions() {
let pool = registry().await;
add(&pool, "u-kid", "u-anna").await.unwrap();
add(&pool, "u-bruno", "u-anna").await.unwrap();
// The supervisor goes: both edges they were on go with them.
sqlx::query("DELETE FROM users WHERE id = 'u-anna'").execute(&pool).await.unwrap();
assert!(list(&pool).await.unwrap().is_empty(), "cascade must clear both directions");
}
}
crates/skald-core/src/db/system_agent_coverage.rs
+167
View File
@@ -0,0 +1,167 @@
//! Accessor for `system_agent_coverage` — how far a background agent has
//! processed a given subject.
//!
//! This is the watermark that turns "everything since last time" into a
//! well-defined window `[covered_through, now)`. Two properties carry the whole
//! design, and both are the opposite of [`super::system_agent_state`]:
//!
//! - **It advances only on a completed pass.** A crash halfway leaves the mark
//! where it was, so the next pass re-covers that stretch. For a review, a
//! duplicate is a nuisance and a gap is a blind spot.
//! - **It is written after the work, not before.** `mark_attempt` is deliberately
//! the first thing `run_and_record` does, which is exactly why it can never
//! delimit the window the work is about.
//!
//! Timestamps are UTC `'YYYY-MM-DD HH:MM:SS'` — the format SQLite's
//! `datetime('now')` produces — so they compare as strings against the
//! `created_at` columns they are used to filter.
use anyhow::Result;
use sqlx::SqlitePool;
/// Format an instant the way SQLite's `datetime('now')` does, so the two are
/// string-comparable. The one place that knows the format.
pub fn stamp(at: chrono::DateTime<chrono::Utc>) -> String {
at.format("%Y-%m-%d %H:%M:%S").to_string()
}
/// Now, in that format.
pub fn now_stamp() -> String {
stamp(chrono::Utc::now())
}
/// How far `agent_id` has processed `subject`, or `None` if it never has.
pub async fn covered_through(
pool: &SqlitePool,
agent_id: &str,
subject: &str,
) -> Result<Option<String>> {
let at = sqlx::query_scalar::<_, String>(
"SELECT covered_through FROM system_agent_coverage
WHERE agent_id = ? AND subject_user_id = ?",
)
.bind(agent_id)
.bind(subject)
.fetch_optional(pool)
.await?;
Ok(at)
}
/// Move the watermark forward to `through`.
///
/// **Monotonic**: an older value than the one stored is ignored rather than
/// applied. Two passes for the same subject cannot run concurrently today (the
/// scheduler is sequential and single-instance), so this is not a race guard — it
/// is a guard against a caller computing a window start and writing *that* back
/// instead of the window end, which would silently make the agent re-read the
/// same stretch forever.
pub async fn advance(
pool: &SqlitePool,
agent_id: &str,
subject: &str,
through: &str,
) -> Result<()> {
sqlx::query(
"INSERT INTO system_agent_coverage (agent_id, subject_user_id, covered_through)
VALUES (?, ?, ?)
ON CONFLICT(agent_id, subject_user_id) DO UPDATE SET
covered_through = MAX(system_agent_coverage.covered_through, excluded.covered_through),
updated_at = datetime('now')",
)
.bind(agent_id)
.bind(subject)
.bind(through)
.execute(pool)
.await?;
Ok(())
}
/// Forget a subject's watermark, so the next pass starts from scratch. For an
/// admin-side "review this person again from the beginning".
pub async fn clear(pool: &SqlitePool, agent_id: &str, subject: &str) -> Result<bool> {
let n = sqlx::query(
"DELETE FROM system_agent_coverage WHERE agent_id = ? AND subject_user_id = ?",
)
.bind(agent_id)
.bind(subject)
.execute(pool)
.await?
.rows_affected();
Ok(n > 0)
}
#[cfg(test)]
mod tests {
use super::*;
async fn registry() -> SqlitePool {
let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
crate::db::create_registry_tables(&pool).await.unwrap();
sqlx::query("INSERT INTO roles (id, label, permission_group) VALUES ('member', 'Member', 'default')")
.execute(&pool).await.unwrap();
sqlx::query(
"INSERT INTO users (id, username, role_id, encrypted)
VALUES ('u-kid', 'kid', 'member', 0)",
)
.execute(&pool).await.unwrap();
pool
}
#[tokio::test]
async fn never_covered_reads_as_none_then_advances() {
let pool = registry().await;
assert!(covered_through(&pool, "conversation-review", "u-kid").await.unwrap().is_none());
advance(&pool, "conversation-review", "u-kid", "2026-07-28 04:00:00").await.unwrap();
assert_eq!(
covered_through(&pool, "conversation-review", "u-kid").await.unwrap().unwrap(),
"2026-07-28 04:00:00",
);
advance(&pool, "conversation-review", "u-kid", "2026-07-29 04:00:00").await.unwrap();
assert_eq!(
covered_through(&pool, "conversation-review", "u-kid").await.unwrap().unwrap(),
"2026-07-29 04:00:00",
);
}
/// The guard that stops a caller from writing the window *start* back.
#[tokio::test]
async fn the_watermark_never_moves_backwards() {
let pool = registry().await;
advance(&pool, "conversation-review", "u-kid", "2026-07-29 04:00:00").await.unwrap();
advance(&pool, "conversation-review", "u-kid", "2026-07-01 04:00:00").await.unwrap();
assert_eq!(
covered_through(&pool, "conversation-review", "u-kid").await.unwrap().unwrap(),
"2026-07-29 04:00:00",
"an older value must not rewind the watermark",
);
}
#[tokio::test]
async fn agents_and_subjects_do_not_share_a_row() {
let pool = registry().await;
sqlx::query(
"INSERT INTO users (id, username, role_id, encrypted)
VALUES ('u-two', 'two', 'member', 0)",
)
.execute(&pool).await.unwrap();
advance(&pool, "conversation-review", "u-kid", "2026-07-29 04:00:00").await.unwrap();
assert!(covered_through(&pool, "conversation-review", "u-two").await.unwrap().is_none());
assert!(covered_through(&pool, "weekly-digest", "u-kid").await.unwrap().is_none());
assert!(clear(&pool, "conversation-review", "u-kid").await.unwrap());
assert!(!clear(&pool, "conversation-review", "u-kid").await.unwrap());
assert!(covered_through(&pool, "conversation-review", "u-kid").await.unwrap().is_none());
}
#[test]
fn the_stamp_matches_sqlite_datetime_shape() {
let s = now_stamp();
assert_eq!(s.len(), 19, "'YYYY-MM-DD HH:MM:SS'");
assert_eq!(&s[4..5], "-");
assert_eq!(&s[10..11], " ");
assert_eq!(&s[13..14], ":");
}
}